Mark workflow-boundary span exceptions with escaped=True

Set exception.escaped=True on SERVER (5xx response), PRODUCER (failed
enqueue) and CONSUMER (failed job) span exception events so downstream
tooling can distinguish workflow-level failures from internal child-span
exceptions. The OTel SDK's auto-record path uses escaped=False, which
made the attribute useless across all 17,692 historical events.

For the PRODUCER span, pass record_exception=False to suppress the SDK
auto-record and call record_span_error inside the except so the failed
send carries a single, correctly-marked event (also stamping error.type
on the span, matching SERVER/CONSUMER).

Author: davegaeddert

plain-jobs/plain/jobs/jobs.py

@@ -27,10 +27,14 @@
 from plain import postgres
 from plain.postgres import transaction
 from plain.utils import timezone
-from plain.utils.otel import format_exception_type
 
 from .locks import postgres_advisory_lock
-from .otel import operation_duration_histogram, sent_messages_counter, tracer
+from .otel import (
+    operation_duration_histogram,
+    record_span_error,
+    sent_messages_counter,
+    tracer,
+)
 from .registry import JobParameters, jobs_registry
 
 if TYPE_CHECKING:
@@ -89,12 +93,16 @@ def run_in_worker(
         }
         start_time = time.perf_counter()
         skipped = False
-        try:
-            with tracer.start_as_current_span(
-                f"send {queue}",
-                kind=SpanKind.PRODUCER,
-                attributes={**metric_attributes, MESSAGING_OPERATION_NAME: "send"},
-            ) as span:
+        with tracer.start_as_current_span(
+            f"send {queue}",
+            kind=SpanKind.PRODUCER,
+            attributes={**metric_attributes, MESSAGING_OPERATION_NAME: "send"},
+            # We record manually via record_span_error (escaped=True) at the
+            # workflow boundary; suppress the SDK's escaped=False auto-record
+            # so failed sends carry a single, correctly-marked event.
+            record_exception=False,
+        ) as span:
+            try:
                 try:
                     frame = sys._getframe(1)
                     filename = frame.f_code.co_filename
@@ -176,30 +184,33 @@ def run_in_worker(
                         )
 
                         return job_request
-        except Exception as e:
-            metric_attributes[ERROR_TYPE] = format_exception_type(e)
-            raise
-        finally:
-            # Skipped enqueues are visible on the span (`job.enqueue.skipped`)
-            # but do not fire the messaging counter — no message was sent, so
-            # there's nothing for `messaging.client.sent.messages` to count.
-            if not skipped:
-                duration = time.perf_counter() - start_time
-                if ERROR_TYPE in metric_attributes:
-                    # No commit is coming — record now so failed sends are visible.
-                    sent_messages_counter.add(1, metric_attributes)
-                    operation_duration_histogram.record(duration, metric_attributes)
-                else:
-                    # Defer to the outer commit so a caller-level rollback
-                    # doesn't leave a phantom send. Runs immediately if not
-                    # inside a transaction.
-                    attrs = metric_attributes
-
-                    def _emit() -> None:
-                        sent_messages_counter.add(1, attrs)
-                        operation_duration_histogram.record(duration, attrs)
-
-                    transaction.on_commit(_emit)
+            except Exception as e:
+                # Stamps escaped=True on the span event, ERROR_TYPE on both
+                # the span and `metric_attributes` (so the finally below
+                # picks up the failed-send branch).
+                record_span_error(span, e, metric_attributes)
+                raise
+            finally:
+                # Skipped enqueues are visible on the span (`job.enqueue.skipped`)
+                # but do not fire the messaging counter — no message was sent, so
+                # there's nothing for `messaging.client.sent.messages` to count.
+                if not skipped:
+                    duration = time.perf_counter() - start_time
+                    if ERROR_TYPE in metric_attributes:
+                        # No commit is coming — record now so failed sends are visible.
+                        sent_messages_counter.add(1, metric_attributes)
+                        operation_duration_histogram.record(duration, metric_attributes)
+                    else:
+                        # Defer to the outer commit so a caller-level rollback
+                        # doesn't leave a phantom send. Runs immediately if not
+                        # inside a transaction.
+                        attrs = metric_attributes
+
+                        def _emit() -> None:
+                            sent_messages_counter.add(1, attrs)
+                            operation_duration_histogram.record(duration, attrs)
+
+                        transaction.on_commit(_emit)
 
     def get_requested_jobs(
         self, *, concurrency_key: str | None = None, include_retries: bool = False

plain-jobs/plain/jobs/otel.py

@@ -65,7 +65,7 @@ def record_span_error(
     metric attribute dict, and return the error.type string so the caller
     can forward it to other instruments."""
     error_type = format_exception_type(exc)
-    span.record_exception(exc)
+    span.record_exception(exc, escaped=True)
     span.set_status(trace.StatusCode.ERROR)
     span.set_attribute(ERROR_TYPE, error_type)
     metric_attributes[ERROR_TYPE] = error_type

plain-jobs/tests/internal/test_otel.py

@@ -76,6 +76,64 @@ def test_enqueue_skipped_marks_span(otel_spans: InMemorySpanExporter) -> None:
     assert span.attributes["job.enqueue.skipped"] is True
 
 
+@pytest.mark.usefixtures("db")
+def test_failed_enqueue_marks_producer_span_exception_as_escaped(
+    monkeypatch: pytest.MonkeyPatch,
+    otel_spans: InMemorySpanExporter,
+) -> None:
+    """A failing enqueue's PRODUCER span must record the exception event with
+    `exception.escaped=True` — caller-side workflow boundary, same signal as
+    SERVER/CONSUMER."""
+
+    def _boom(*args, **kwargs):
+        raise RuntimeError("save failed")
+
+    from plain.jobs.models import JobRequest
+
+    monkeypatch.setattr(JobRequest, "save", _boom)
+
+    with pytest.raises(RuntimeError):
+        _NoopJob().run_in_worker()
+
+    producer_spans = [
+        s for s in otel_spans.get_finished_spans() if s.kind == SpanKind.PRODUCER
+    ]
+    assert producer_spans, "expected PRODUCER span from run_in_worker()"
+    span = producer_spans[-1]
+    exception_events = [e for e in span.events if e.name == "exception"]
+    # Exactly one event — `record_exception=False` on start_as_current_span
+    # suppresses the SDK's escaped=False auto-record so the manual call is
+    # the sole event.
+    assert len(exception_events) == 1
+    attrs = exception_events[0].attributes
+    assert attrs is not None
+    assert attrs["exception.escaped"] == "True"
+
+
+@pytest.mark.usefixtures("db")
+def test_failing_job_marks_consumer_span_exception_as_escaped(
+    otel_spans: InMemorySpanExporter,
+) -> None:
+    """A failing job's CONSUMER span must record the exception event with
+    `exception.escaped=True` — the workflow-level failure signal that
+    downstream tools filter on."""
+    request = _BoomJob().run_in_worker()
+    assert request is not None
+    process = request.convert_to_job_process(worker_id=uuid.uuid4())
+    process.run()
+
+    consumer_spans = [
+        s for s in otel_spans.get_finished_spans() if s.kind == SpanKind.CONSUMER
+    ]
+    assert consumer_spans, "expected CONSUMER span from JobProcess.run()"
+    span = consumer_spans[-1]
+    exception_events = [e for e in span.events if e.name == "exception"]
+    assert exception_events
+    attrs = exception_events[0].attributes
+    assert attrs is not None
+    assert attrs["exception.escaped"] == "True"
+
+
 @pytest.mark.usefixtures("db")
 def test_enqueue_failure_records_error_type_on_metric(
     monkeypatch: pytest.MonkeyPatch,

plain/plain/internal/handlers/base.py

@@ -178,7 +178,7 @@ def _finalize_span(self, span: trace.Span, response: Response) -> None:
         if response.status_code >= 500:
             span.set_status(trace.StatusCode.ERROR)
             if response.exception:
-                span.record_exception(response.exception)
+                span.record_exception(response.exception, escaped=True)
                 span.set_attribute(
                     error_attributes.ERROR_TYPE,
                     format_exception_type(response.exception),

plain/tests/internal/test_otel_spans.py

@@ -66,6 +66,9 @@ def test_500_records_exception_and_error_status(error_client) -> None:
     assert span.attributes["error.type"] == "RuntimeError"
     exception_events = [e for e in span.events if e.name == "exception"]
     assert exception_events
+    attrs = exception_events[0].attributes
+    assert attrs is not None
+    assert attrs["exception.escaped"] == "True"
 
 
 def _invoke_handler(router_path: str) -> None: