feat(otel): Give each session's cycle its own reactive_update span

CHANGELOG.md

@@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Breaking changes
 
+* OpenTelemetry `reactive_update` spans now follow each session's cycle: one span per session, from when the session turns busy until all of its effects (including async ones) have finished, carrying the session's `session.id`. The session's calc, effect, and output spans are its children. Before, there was one span per reactive flush, which can serve several sessions, so with concurrent effects it couldn't carry a single `session.id` and ended before async effects finished. Effects without a session no longer have a `reactive_update` parent. (#2522)
+
 * Reactive effects now run concurrently. A reactive flush starts each invalidated effect and no longer waits for an `async` effect's awaited part, so a slow `async` effect, calc, or render function no longer delays other sessions. Within a session, input changes and `reactive.invalidate_later()` still wait until all of that session's effects have finished, and outputs are sent together once they have. What app authors may notice:
 
     * `async` effects interleave at each `await` instead of running one after another. `priority` orders when effects start, not when they finish. Runs of the same effect still don't overlap: a re-run waits for the previous run.

shiny/.agents/skills/shiny-for-python/references/otel.md

@@ -39,11 +39,18 @@ At the `all` level the span hierarchy looks like:
 
 ```text
 session_start
-  └─ reactive_update
-      ├─ reactive.calc filtered_data
+  └─ reactive_update              # the session's first cycle
       └─ output result
+          └─ reactive.calc filtered_data
+reactive_update                   # each later cycle: a trace of its own
+  └─ output result
+      └─ reactive.calc filtered_data
 ```
 
+A `reactive_update` span covers one session cycle: from when the session turns
+busy until all of its effects, including async ones, have finished. It carries
+`session.id`. Effects without a session have no `reactive_update` parent.
+
 ## Collection levels: `SHINY_OTEL_COLLECT`
 
 Controls how much Shiny telemetry is emitted (default `all`):
@@ -52,7 +59,7 @@ Controls how much Shiny telemetry is emitted (default `all`):
 |---|---|---|
 | `none` | nothing from Shiny | disable Shiny spans, keep your own |
 | `session` | session lifecycle only | low-overhead production |
-| `reactive_update` | + one span per flush cycle | balanced production |
+| `reactive_update` | + one span per session cycle | balanced production |
 | `reactivity` | + per calc/effect/output spans, value-update logs | development, debugging |
 | `all` | everything (currently = `reactivity`) | maximum detail |
 

shiny/otel/__init__.py

@@ -215,9 +215,13 @@ def result():
 **Use case**: Basic session tracking in production
 
 ### `reactive_update`
-Session spans + reactive update cycle spans (one span per flush cycle).
+Session spans + reactive update cycle spans: one `reactive_update` span per session
+cycle, from when the session turns busy until all of its effects (including async
+ones) have finished. It carries the session's `session.id`, and the session's
+calc, effect, and output spans are its children. The first cycle's span is a child
+of `session_start`; later cycles each start a trace of their own.
 
-**Overhead**: Low (1 span per reactive flush)
+**Overhead**: Low (1 span per session cycle)
 **Use case**: Understanding how many update cycles occur
 
 ### `reactivity`

shiny/otel/_span_wrappers.py

@@ -54,16 +54,30 @@
 
 from __future__ import annotations
 
-from contextlib import asynccontextmanager
-from typing import Any, AsyncGenerator, AsyncIterable, Callable, Dict, Mapping, Union
-
-from opentelemetry.trace import Span, Status, StatusCode
+from contextlib import asynccontextmanager, contextmanager
+from typing import (
+    Any,
+    AsyncGenerator,
+    AsyncIterable,
+    Callable,
+    Dict,
+    Generator,
+    Mapping,
+    Union,
+)
+
+from opentelemetry.trace import Span, Status, StatusCode, use_span
 
 from ._collect import OtelCollectLevel, get_level
 from ._constants import ATTR_SESSION_ID
 from ._core import get_otel_tracer, is_otel_tracing_enabled
 
-__all__ = ("shiny_otel_span", "shiny_otel_span_stream")
+__all__ = (
+    "shiny_otel_span",
+    "shiny_otel_span_stream",
+    "start_otel_span",
+    "use_otel_span",
+)
 
 # Type aliases for parameters
 AttributesValue = Mapping[str, Any] | None
@@ -224,6 +238,37 @@ async def my_async_function():
             raise
 
 
+def start_otel_span(
+    name: str,
+    *,
+    attributes: Mapping[str, Any],
+    required_level: OtelCollectLevel,
+    collection_level: OtelCollectLevel,
+) -> Span | None:
+    """
+    Start a span that the caller ends, for work that isn't one block of code.
+
+    Unlike `shiny_otel_span()`, the span doesn't become the current span; run code
+    in it with `use_otel_span()`, and call `span.end()` when the work is done.
+    Returns `None` when tracing is off or `collection_level < required_level`.
+    """
+    if not is_otel_tracing_enabled() or collection_level < required_level:
+        return None
+    return get_otel_tracer().start_span(name, attributes=attributes)
+
+
+@contextmanager
+def use_otel_span(span: Span | None) -> Generator[None, None, None]:
+    """Make `span` the current span for a block, without ending it. No-op on `None`."""
+    if not isinstance(span, Span):
+        yield
+        return
+    with use_span(
+        span, end_on_exit=False, record_exception=False, set_status_on_exception=False
+    ):
+        yield
+
+
 async def shiny_otel_span_stream(
     name: str,
     inner: AsyncIterable[bytes],

shiny/reactive/_core.py

@@ -34,8 +34,6 @@
 from .._datastructures import PriorityQueueFIFO
 from .._docstring import add_example, no_example
 from .._typing_extensions import TypeGuard
-from ..otel._collect import OtelCollectLevel, _get_env_level
-from ..otel._span_wrappers import shiny_otel_span
 from ..types import MISSING, MISSING_TYPE
 
 if TYPE_CHECKING:
@@ -243,21 +241,17 @@ async def start_round(self) -> None:
         self._next_round_waiters = []
         token = _enclosing_run.set(asyncio.current_task())
         try:
-            # Wrap the round in a reactive_update span (or no-op if not collecting)
-            async with shiny_otel_span(
-                "reactive_update",
-                infer_session_id=True,
-                required_level=OtelCollectLevel.REACTIVE_UPDATE,
-                collection_level=_get_env_level(),
-            ):
-                while not self._effect_queue.empty():
-                    ctx = self._effect_queue.get()
-                    self._spawn(self._run_context(ctx))
-                    # CPython runs ready callbacks FIFO, so the task's sync part runs
-                    # now, and anything it invalidates is queued before we take the
-                    # next ctx.
-                    await _yield_to_loop()
-                await self._round_finished_callbacks.invoke()
+            # No span here: a round can serve several sessions. Each session's
+            # `reactive_update` span follows its own cycle (see
+            # `AppSession._increment_busy_count()`).
+            while not self._effect_queue.empty():
+                ctx = self._effect_queue.get()
+                self._spawn(self._run_context(ctx))
+                # CPython runs ready callbacks FIFO, so the task's sync part runs
+                # now, and anything it invalidates is queued before we take the
+                # next ctx.
+                await _yield_to_loop()
+            await self._round_finished_callbacks.invoke()
         finally:
             _enclosing_run.reset(token)
             self._round_running = False

shiny/reactive/_reactives.py

@@ -46,7 +46,7 @@
     get_otel_label_modifier,
     set_otel_label_modifier,
 )
-from ..otel._span_wrappers import shiny_otel_span
+from ..otel._span_wrappers import shiny_otel_span, use_otel_span
 from ..types import (
     MISSING,
     MISSING_TYPE,
@@ -1185,7 +1185,12 @@ async def _run(self) -> None:
 
         from ..session import session_context
 
-        with session_context(self._session):
+        # Run in the session's cycle span, so this effect's span (and those of the
+        # calcs it reads) are its children.
+        cycle_span = (
+            self._session._otel_reactive_update_span() if self._session else None
+        )
+        with session_context(self._session), use_otel_span(cycle_span):
             async with shiny_otel_span(
                 self._otel_label,
                 attributes=self._otel_attrs,

shiny/session/_session.py

@@ -33,6 +33,7 @@
 
 import orjson
 from htmltools import TagChild, TagList
+from opentelemetry.trace import Span, Status, StatusCode
 from starlette.requests import HTTPConnection, Request
 from starlette.responses import (
     HTMLResponse,
@@ -66,7 +67,11 @@
 from ..otel._decorators import suppress as otel_suppress
 from ..otel._function_attrs import resolve_func_otel_level
 from ..otel._labels import create_otel_label, create_otel_span_name
-from ..otel._span_wrappers import shiny_otel_span, shiny_otel_span_stream
+from ..otel._span_wrappers import (
+    shiny_otel_span,
+    shiny_otel_span_stream,
+    start_otel_span,
+)
 from ..reactive import Effect_, Value, effect, isolate
 from ..reactive._core import _enclosing_run, _reactive_environment
 from ..render.renderer import Renderer, RendererT
@@ -827,6 +832,10 @@ def _increment_busy_count(self) -> None: ...
     @abstractmethod
     def _decrement_busy_count(self) -> None: ...
 
+    def _otel_reactive_update_span(self) -> Optional[Span]:
+        """The `reactive_update` span of this session's current cycle, if any."""
+        return None
+
     def run_once_when_idle(self, fn: Callable[[], object]) -> None:
         """
         Run a function once, at the start of this session's next cycle.
@@ -967,6 +976,9 @@ def __init__(
         self._conn: Connection = conn
         self._debug: bool = debug
         self._busy_count: int = 0
+        # Spans this session's busy period (its cycle), as R does; see
+        # `_increment_busy_count()`.
+        self._otel_cycle_span: Optional[Span] = None
         # Actions (input updates, timer invalidations) waiting for the session to be
         # idle.
         self._cycle_start_action_queue: list[Callable[[], None]] = []
@@ -1084,6 +1096,7 @@ async def _run_session_ended_tasks(self) -> None:
                     # on_ended handlers can still read reactive state.
                     await self.destroy()
                 finally:
+                    self._end_otel_cycle_span()
                     self.app._remove_session(self)
 
     def is_stub_session(self) -> Literal[False]:
@@ -1815,15 +1828,36 @@ def _increment_busy_count(self) -> None:
         self._busy_count += 1
         if self._busy_count == 1:
             self._send_message_sync({"busy": "busy"})
+            # A flush is global and can serve several sessions, so the
+            # `reactive_update` span follows this session's cycle instead: it starts
+            # now and ends when all of the session's effects have finished. Its
+            # parent is the current span, e.g. `session_start` for the first cycle.
+            if self._otel_cycle_span is None:
+                self._otel_cycle_span = start_otel_span(
+                    "reactive_update",
+                    attributes=get_session_id_attrs(self),
+                    required_level=OtelCollectLevel.REACTIVE_UPDATE,
+                    collection_level=_get_env_level(),
+                )
 
     def _decrement_busy_count(self) -> None:
         self._busy_count -= 1
         if self._busy_count == 0:
+            self._end_otel_cycle_span()
             self._send_message_sync({"busy": "idle"})
             # `_output_flush` sends this cycle's outputs, then starts the next action.
             self._cycle_ended = True
             self._request_output_flush()
 
+    def _otel_reactive_update_span(self) -> Optional[Span]:
+        return self._otel_cycle_span
+
+    def _end_otel_cycle_span(self) -> None:
+        span, self._otel_cycle_span = self._otel_cycle_span, None
+        if span is not None:
+            span.set_status(Status(StatusCode.OK))
+            span.end()
+
     def run_once_when_idle(self, fn: Callable[[], object]) -> None:
         # Actions are sync, so they can't be interrupted partway, and only
         # `_output_flush` runs them (never whichever task happened to queue one).
@@ -2145,6 +2179,9 @@ def _increment_busy_count(self) -> None:
     def _decrement_busy_count(self) -> None:
         self._root_session._decrement_busy_count()
 
+    def _otel_reactive_update_span(self) -> Optional[Span]:
+        return self._root_session._otel_reactive_update_span()
+
     def run_once_when_idle(self, fn: Callable[[], object]) -> None:
         _validate_idle_action(fn)
 

tests/pytest/test_otel_async_context.py

@@ -18,7 +18,6 @@
 from shiny.otel._collect import OtelCollectLevel
 from shiny.otel._span_wrappers import shiny_otel_span
 from shiny.reactive import Calc_
-from shiny.reactive._core import ReactiveEnvironment
 
 from .otel_helpers import get_exported_spans, patch_otel_tracing_state
 
@@ -178,43 +177,6 @@ async def slow_calc_2():
         # Verify calc2 finished before calc1 (because it sleeps less)
         assert execution_order.index("calc2_end") < execution_order.index("calc1_end")
 
-    @pytest.mark.asyncio
-    async def test_parallel_flush_cycles_maintain_context(
-        self, otel_tracer_provider: Tuple[TracerProvider, InMemorySpanExporter]
-    ):
-        """Test that parallel flush cycles each maintain their own span context"""
-        provider, memory_exporter = otel_tracer_provider
-
-        with patch_otel_tracing_state(tracing_enabled=True):
-            # Create two separate reactive environments
-            env1 = ReactiveEnvironment()
-            env2 = ReactiveEnvironment()
-
-            # Flush them in parallel
-            await asyncio.gather(
-                env1.start_round(),
-                env2.start_round(),
-            )
-
-        # Get exported spans
-        spans = get_exported_spans(provider, memory_exporter)
-
-        # Filter to reactive_update spans
-        update_spans = [s for s in spans if s.name == "reactive_update"]
-
-        # Should have two separate reactive_update spans
-        assert len(update_spans) == 2
-
-        # Verify they are separate spans (different span IDs)
-        assert (
-            update_spans[0].context is not None
-        ), "update_spans[0] should have a context"
-        assert (
-            update_spans[1].context is not None
-        ), "update_spans[1] should have a context"
-        assert update_spans[0].context.span_id != update_spans[1].context.span_id
-
-
 class TestAsyncContextIsolation:
     """Test that async operations maintain proper context isolation"""