Wrap job worker maintenance in an INTERNAL span

Currently DB transients during maintenance (heartbeat, rescue, schedule)
get caught by the outer except-Exception and swallowed via logger.exception
with no surrounding span — so they're invisible to OTel-based exception
tooling even though they're real failures. Wrapping the maintenance work
in a `worker loop` INTERNAL span stamps the canonical status=ERROR plus
error.type signal on each failed iteration, keeping the existing log-and-
continue behavior intact.

Author: davegaeddert

plain-jobs/plain/jobs/workers.py

@@ -14,15 +14,19 @@
 from functools import partial
 from typing import TYPE_CHECKING, Any
 
+from opentelemetry import trace
+from opentelemetry.semconv.attributes.error_attributes import ERROR_TYPE
+
 from plain.logs import get_framework_logger
 from plain.postgres import transaction
 from plain.postgres.db import return_database_connection
 from plain.runtime import settings
 from plain.utils import timezone
 from plain.utils.module_loading import import_string
 from plain.utils.os import get_cpu_count
+from plain.utils.otel import format_exception_type
 
-from .otel import WorkerMetrics
+from .otel import WorkerMetrics, tracer
 from .registry import jobs_registry
 
 if TYPE_CHECKING:
@@ -184,15 +188,23 @@ def _run_loop(self) -> None:
         from .models import JobRequest
 
         while not self._is_shutting_down:
-            try:
-                self.maybe_heartbeat()
-                self.maybe_log_stats()
-                self.maybe_check_job_results()
-                self.maybe_schedule_jobs()
-            except Exception as e:
-                # Log the issue, but don't stop the worker
-                # (these tasks are kind of ancilarry to the main job processing)
-                logger.exception(e)
+            with tracer.start_as_current_span(
+                "worker loop", kind=trace.SpanKind.INTERNAL
+            ) as span:
+                try:
+                    self.maybe_heartbeat()
+                    self.maybe_log_stats()
+                    self.maybe_check_job_results()
+                    self.maybe_schedule_jobs()
+                except Exception as e:
+                    # The catch is inside the span, so the SDK's auto-record
+                    # on context exit won't fire — stamp the canonical
+                    # failure signal explicitly. Log and continue: these
+                    # tasks are ancillary to the main job processing.
+                    span.record_exception(e)
+                    span.set_status(trace.StatusCode.ERROR)
+                    span.set_attribute(ERROR_TYPE, format_exception_type(e))
+                    logger.exception(e)
 
             # Re-check shutdown after maintenance — a signal may have arrived
             # between the loop condition and now. Don't pick up new work.

plain-jobs/tests/internal/test_otel.py

@@ -8,6 +8,7 @@
 
 from __future__ import annotations
 
+import threading
 import uuid
 
 import pytest
@@ -164,6 +165,81 @@ def _boom(*args, **kwargs):
     )
 
 
+# --- Worker run-loop span -----------------------------------------------
+
+
+def _build_worker_for_loop_test() -> Worker:
+    """Construct a Worker bypassing __init__ so `_run_loop` can run a single
+    iteration without a real ProcessPoolExecutor. Tests override the maintenance
+    methods to control when the loop exits."""
+    worker = Worker.__new__(Worker)
+    worker.queues = ["default"]
+    worker._is_shutting_down = False
+    worker._heartbeat_registered = True
+    worker._inflight_futures = {}
+    worker._inflight_lock = threading.Lock()
+    worker.max_processes = 1
+    worker.max_pending_per_process = 1
+    worker.maybe_heartbeat = lambda: None
+    worker.maybe_log_stats = lambda: None
+    worker.maybe_check_job_results = lambda: None
+    worker.maybe_schedule_jobs = lambda: None
+    return worker
+
+
+@pytest.mark.usefixtures("db")
+def test_worker_loop_emits_internal_span_per_iteration(
+    otel_spans: InMemorySpanExporter,
+) -> None:
+    """Each worker loop iteration wraps the maintenance work in a `worker loop`
+    INTERNAL span so DB transients during maintenance land on a span."""
+    from opentelemetry.trace import StatusCode
+
+    worker = _build_worker_for_loop_test()
+
+    def shutdown_during_heartbeat() -> None:
+        worker._is_shutting_down = True
+
+    worker.maybe_heartbeat = shutdown_during_heartbeat  # ty: ignore[invalid-assignment]
+    worker._run_loop()
+
+    loop_spans = [s for s in otel_spans.get_finished_spans() if s.name == "worker loop"]
+    assert len(loop_spans) == 1
+    span = loop_spans[0]
+    assert span.kind == SpanKind.INTERNAL
+    assert span.status.status_code == StatusCode.UNSET
+
+
+@pytest.mark.usefixtures("db")
+def test_worker_loop_records_error_when_maintenance_fails(
+    otel_spans: InMemorySpanExporter,
+) -> None:
+    """A maintenance exception leaves the loop running and stamps the canonical
+    failure signal (status=ERROR + error.type) on the `worker loop` span. This
+    is the path that previously swallowed DB transients like the production
+    `psycopg.OperationalError` we saw escaping `rescue_job_results`."""
+    from opentelemetry.trace import StatusCode
+
+    worker = _build_worker_for_loop_test()
+
+    def boom_then_shutdown() -> None:
+        worker._is_shutting_down = True
+        raise RuntimeError("db transient")
+
+    worker.maybe_heartbeat = boom_then_shutdown  # ty: ignore[invalid-assignment]
+    # Must NOT raise — the loop catches and continues, just like in production.
+    worker._run_loop()
+
+    loop_spans = [s for s in otel_spans.get_finished_spans() if s.name == "worker loop"]
+    assert len(loop_spans) == 1
+    span = loop_spans[0]
+    assert span.status.status_code == StatusCode.ERROR
+    assert span.attributes is not None
+    assert span.attributes["error.type"] == "RuntimeError"
+    exception_events = [e for e in span.events if e.name == "exception"]
+    assert exception_events
+
+
 # --- Worker-state observable gauges -------------------------------------
 #
 # Each Worker owns a WorkerMetrics; instantiating one swaps it in as the