symfonic.services.observability.otel_spans

otel_spans

What one run holds in OTEL, and when it is given back.

Two resources are opened when a run starts and must be unwound when it ends: the root span (a context manager entered by hand, because a run's start and its terminal are two separate awaits on the event stream) and the OTEL carrier token that publishes the service's run identity. They are opened together, released together, and released in a fixed order — so they are one object here rather than two dicts and four helpers on the observer, which drives the event lifecycle and should not also own the resource lifecycle.

Keyed by run, and only correct because of it. A single slot made an observer's two possible lifetimes mutually exclusive: built per run it leaked exporters, shared across runs it silently lost run A's root span the moment run B started. Keying by run_id removes that, but keying alone is not enough, and the two remaining holes are both silent:

• A reused key is the same single-slot leak narrowed to one key. Overwriting the entry abandons a root span that was never exited and a token that is never reset, and the span reaches the exporter only if the collector happens to throw GeneratorExit into it — with the wrong end time. :meth:open therefore displaces the old entry explicitly and logs it.
• Wholesale release is not run cleanup. :meth:release exists so a per-run owner (ObservabilityBridge.aclose) can give back its own run without ending the spans of every other run in flight; :meth:release_all is the process-shutdown call.

Nothing here raises. Telemetry that fails on the way out must not take down the run or the caller's cleanup path.

RunSpanTable

RunSpanTable(handles: Any)

The open root span and carrier token of every run in flight.

Source code in src/symfonic/services/observability/otel_spans.py

def __init__(self, handles: Any) -> None:
    self._handles = handles
    #: run_id -> (open root-span context manager | None, carrier token | None)
    self._runs: dict[str, tuple[Any, object | None]] = {}

open

open(observation: Any) -> None

Start one run's root span and publish its identity.

A run id that is already open is a caller bug — a retried invocation reusing its id, or two sinks built over one scope. It is survivable, so the displaced entry is closed rather than dropped, but it is not hidden.

Source code in src/symfonic/services/observability/otel_spans.py

def open(self, observation: Any) -> None:
    """Start one run's root span and publish its identity.

    A run id that is already open is a caller bug — a retried invocation
    reusing its id, or two sinks built over one scope. It is survivable, so
    the displaced entry is closed rather than dropped, but it is not hidden.
    """
    run_id = observation.scope.run_id
    if run_id in self._runs:
        logger.warning(
            "OTel run %s started while its previous span was still open; "
            "closing the displaced span",
            run_id,
        )
        self.release(run_id)
    token = self._publish_run_context(observation.scope)
    self._runs[run_id] = (self._open_run_span(observation), token)

release

release(run_id: str) -> None

Unwind one run. Unknown ids are a no-op — a terminal may follow a cancellation that already released the run.

Root span first, carrier second: the span's own exit resets the carrier entry it pushed, so unwinding in the other order would reset tokens out of the order they were taken.

Source code in src/symfonic/services/observability/otel_spans.py

def release(self, run_id: str) -> None:
    """Unwind one run. Unknown ids are a no-op — a terminal may follow a
    cancellation that already released the run.

    Root span first, carrier second: the span's own exit resets the carrier
    entry it pushed, so unwinding in the other order would reset tokens out
    of the order they were taken.
    """
    manager, token = self._runs.pop(run_id, (None, None))
    self._close_run_span(manager)
    self._release_run_context(token)

release_all

release_all() -> None

Unwind every run still held. Process shutdown, not run cleanup.

Source code in src/symfonic/services/observability/otel_spans.py

def release_all(self) -> None:
    """Unwind every run still held. Process shutdown, not run cleanup."""
    runs, self._runs = self._runs, {}
    for manager, token in runs.values():
        self._close_run_span(manager)
        self._release_run_context(token)