fix(tracing): Skip child span creation in streaming path when no current span (task queues) (#6814)

## Summary
- When span streaming is enabled and there is no current span, task
queue integrations should not create new root segments for child-span
operations
- Adds a `sentry_sdk.traces.get_current_span() is None` guard before
creating spans in the streaming path
- Affected integrations: celery, huey, arq, ray

## Test plan
- [ ] Existing tests pass
- [ ] Verify that in streaming mode, no orphan root segments are created
for task queue operations when there's no active span

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
Co-authored-by: Erica Pisani <hey@ericapisani.dev>

Author: 3 people

sentry_sdk/integrations/arq.py

@@ -79,6 +79,9 @@ async def _sentry_enqueue_job(
             return await old_enqueue_job(self, function, *args, **kwargs)
 
         if has_span_streaming_enabled(client.options):
+            if sentry_sdk.traces.get_current_span() is None:
+                return await old_enqueue_job(self, function, *args, **kwargs)
+
             with sentry_sdk.traces.start_span(
                 name=function,
                 attributes={

sentry_sdk/integrations/celery/__init__.py

@@ -419,6 +419,9 @@ def _inner(*args: "Any", **kwargs: "Any") -> "Any":
         span_streaming = has_span_streaming_enabled(client.options)
 
         try:
+            if span_streaming and get_current_span() is None:
+                return f(*args, **kwargs)
+
             span: "Union[Span, StreamedSpan]"
             if span_streaming:
                 span = sentry_sdk.traces.start_span(
@@ -462,7 +465,8 @@ def _inner(*args: "Any", **kwargs: "Any") -> "Any":
 
                 with capture_internal_exceptions():
                     set_on_span(
-                        SPANDATA.MESSAGING_MESSAGE_RETRY_COUNT, task.request.retries
+                        SPANDATA.MESSAGING_MESSAGE_RETRY_COUNT,
+                        task.request.retries,
                     )
 
                 with capture_internal_exceptions():

sentry_sdk/integrations/huey.py

@@ -79,7 +79,18 @@ def _sentry_enqueue(
             sentry_sdk.get_client().options
         )
 
-        span_ctx = None
+        no_headers_types = (PeriodicTask,) + tuple(
+            t for t in [HueyGroup, HueyChord] if t is not None
+        )
+
+        if is_span_streaming_enabled and sentry_sdk.traces.get_current_span() is None:
+            if not isinstance(item, no_headers_types):
+                item.kwargs["sentry_headers"] = {
+                    BAGGAGE_HEADER_NAME: get_baggage(),
+                    SENTRY_TRACE_HEADER_NAME: get_traceparent(),
+                }
+            return old_enqueue(self, item)
+
         if is_span_streaming_enabled:
             span_ctx = sentry_sdk.traces.start_span(
                 name=span_name,
@@ -97,9 +108,6 @@ def _sentry_enqueue(
             )
             span_ctx.set_data(SPANDATA.MESSAGING_DESTINATION_NAME, self.name)
 
-        no_headers_types = (PeriodicTask,) + tuple(
-            t for t in [HueyGroup, HueyChord] if t is not None
-        )
         with span_ctx:
             if not isinstance(item, no_headers_types):
                 # Attach trace propagation data to task kwargs. We do

sentry_sdk/integrations/ray.py

@@ -156,6 +156,23 @@ def _remote_method_with_header_propagation(
                 )
                 if span_streaming:
                     function_name = qualname_from_function(user_f)
+
+                    if sentry_sdk.traces.get_current_span() is None:
+                        tracing = {
+                            k: v
+                            for k, v in sentry_sdk.get_current_scope().iter_trace_propagation_headers()
+                        }
+                        try:
+                            result = old_remote_method(
+                                *args, **kwargs, _sentry_tracing=tracing
+                            )
+                        except Exception:
+                            exc_info = sys.exc_info()
+                            _capture_exception(exc_info)
+                            reraise(*exc_info)
+
+                        return result
+
                     with sentry_sdk.traces.start_span(
                         name="unknown Ray task"
                         if function_name is None

tests/integrations/arq/test_arq.py

@@ -245,18 +245,22 @@ async def retry_job(ctx):
         sentry_sdk.flush()
         spans = [item.payload for item in items]
 
-        assert spans[4]["attributes"]["sentry.op"] == "queue.task.arq"
-        assert spans[4]["status"] == "ok"
-        assert spans[4]["name"] == "retry_job"
+        # The retry re-enqueue happens without an active span, so no producer
+        # (queue.submit.arq) span is created for it; only the consumer segment
+        # is emitted. The consumer segment is preceded by the redis spans for
+        # the re-enqueue, so it lands at index 3.
+        assert spans[3]["attributes"]["sentry.op"] == "queue.task.arq"
+        assert spans[3]["status"] == "ok"
+        assert spans[3]["name"] == "retry_job"
 
         await worker.run_job(job.job_id, timestamp_ms())
 
         sentry_sdk.flush()
         spans = [item.payload for item in items]
 
-        assert spans[7]["attributes"]["sentry.op"] == "queue.task.arq"
-        assert spans[7]["status"] == "ok"
-        assert spans[7]["name"] == "retry_job"
+        assert spans[6]["attributes"]["sentry.op"] == "queue.task.arq"
+        assert spans[6]["status"] == "ok"
+        assert spans[6]["name"] == "retry_job"
     else:
         events = capture_events()
 
@@ -594,10 +598,13 @@ async def job(ctx):
         sentry_sdk.flush()
         spans = [item.payload for item in items]
 
-        assert spans[4]["attributes"]["sentry.op"] == "queue.task.arq"
-        assert spans[4]["attributes"]["sentry.origin"] == "auto.queue.arq"
-        assert spans[3]["attributes"]["sentry.origin"] == "auto.db.redis"
+        # No producer (queue.submit.arq) span is created for the re-enqueue
+        # triggered by the retry, since it happens without an active span, so
+        # the consumer segment lands at index 3.
+        assert spans[3]["attributes"]["sentry.op"] == "queue.task.arq"
+        assert spans[3]["attributes"]["sentry.origin"] == "auto.queue.arq"
         assert spans[2]["attributes"]["sentry.origin"] == "auto.db.redis"
+        assert spans[1]["attributes"]["sentry.origin"] == "auto.db.redis"
     else:
         job = await pool.enqueue_job("retry_job")
 

tests/integrations/huey/test_huey.py

@@ -93,13 +93,12 @@ def division(a, b):
         sentry_sdk.get_client().flush()
 
         payloads = [i.payload for i in items]
-        assert len(payloads) == 2
-        enqueue_span, execute_span = payloads
+        # The task is enqueued without a wrapping span, so in streaming mode no
+        # producer (queue.submit.huey) span is created (see the early return in
+        # patch_enqueue). Only the consumer segment is emitted.
+        assert len(payloads) == 1
+        (execute_span,) = payloads
 
-        assert enqueue_span["attributes"]["sentry.op"] == OP.QUEUE_SUBMIT_HUEY
-        assert (
-            enqueue_span["attributes"][SPANDATA.MESSAGING_DESTINATION_NAME] == huey.name
-        )
         assert execute_span["is_segment"]
         assert execute_span["attributes"]["sentry.op"] == OP.QUEUE_TASK_HUEY
         assert (
@@ -157,12 +156,12 @@ def retry_task(context):
         sentry_sdk.get_client().flush()
 
         payloads = [i.payload for i in items]
-        assert len(payloads) == 3
-
-        enqueue_span, re_enqueue_span, execute_span = payloads
+        # The initial enqueue happens without a wrapping span, so no producer
+        # span is created for it. The re-enqueue triggered by RetryTask happens
+        # inside the running consumer segment, so it does get a child span.
+        assert len(payloads) == 2
 
-        assert enqueue_span["attributes"]["sentry.op"] == OP.QUEUE_SUBMIT_HUEY
-        assert enqueue_span["is_segment"]
+        re_enqueue_span, execute_span = payloads
 
         assert re_enqueue_span["attributes"]["sentry.op"] == OP.QUEUE_SUBMIT_HUEY
         assert not re_enqueue_span["is_segment"]
@@ -181,8 +180,8 @@ def retry_task(context):
 
         all_payloads = [i.payload for i in items]
 
-        assert len(all_payloads) == 4
-        retry_span = all_payloads[3]
+        assert len(all_payloads) == 3
+        retry_span = all_payloads[2]
 
         assert retry_span["is_segment"]
         assert retry_span["name"] == "retry_task"
@@ -224,10 +223,10 @@ def cancel_task():
         sentry_sdk.get_client().flush()
 
         payloads = [i.payload for i in items]
-        assert len(payloads) == 2
-        enqueue_span, execute_span = payloads
+        # Enqueued without a wrapping span -> no producer span in streaming mode.
+        assert len(payloads) == 1
+        (execute_span,) = payloads
 
-        assert enqueue_span["attributes"]["sentry.op"] == OP.QUEUE_SUBMIT_HUEY
         assert execute_span["attributes"]["sentry.op"] == OP.QUEUE_TASK_HUEY
         assert execute_span["is_segment"]
         assert execute_span["name"] == "cancel_task"
@@ -270,10 +269,10 @@ def maybe_locked_task():
         sentry_sdk.get_client().flush()
 
         payloads = [i.payload for i in items]
-        assert len(payloads) == 2
-        enqueue_span, execute_span = payloads
+        # Enqueued without a wrapping span -> no producer span in streaming mode.
+        assert len(payloads) == 1
+        (execute_span,) = payloads
 
-        assert enqueue_span["attributes"]["sentry.op"] == OP.QUEUE_SUBMIT_HUEY
         assert execute_span["attributes"]["sentry.op"] == OP.QUEUE_TASK_HUEY
 
         assert execute_span["is_segment"]
@@ -391,29 +390,36 @@ def task2():
     if has_span_streaming:
         items = capture_items("span")
 
-        huey.enqueue(group([task1.s(), task2.s()]))
+        with sentry_sdk.traces.start_span(name="submission"):
+            huey.enqueue(group([task1.s(), task2.s()]))
 
         for _ in range(2):
             task = huey.dequeue()
             huey.execute(task)
 
         sentry_sdk.get_client().flush()
-        assert len(items) == 5
+        assert len(items) == 6
 
         (
             task1_enqueue_span,
             task2_enqueue_span,
             group_span,
+            submission_span,
             task1_execute_span,
             task2_execute_span,
         ) = [i.payload for i in items]
 
-        assert group_span["is_segment"]
+        # The enqueue happens inside a wrapping span, so the group producer
+        # tree is created and parented under that segment.
+        assert submission_span["is_segment"]
+        assert submission_span["name"] == "submission"
+        assert not group_span["is_segment"]
         assert not task1_enqueue_span["is_segment"]
         assert not task2_enqueue_span["is_segment"]
         assert task1_execute_span["is_segment"]
         assert task2_execute_span["is_segment"]
 
+        assert group_span["parent_span_id"] == submission_span["span_id"]
         assert group_span["name"] == "Huey Task Group"
         assert group_span["status"] == "ok"
         assert group_span["attributes"]["sentry.op"] == "queue.submit.huey"
@@ -422,18 +428,14 @@ def task2():
         assert task1_enqueue_span["name"] == "task1"
         assert task1_enqueue_span["status"] == "ok"
         assert task1_enqueue_span["parent_span_id"] == group_span["span_id"]
-        assert (
-            task1_enqueue_span["attributes"]["sentry.segment.name"] == "Huey Task Group"
-        )
+        assert task1_enqueue_span["attributes"]["sentry.segment.name"] == "submission"
         assert task1_enqueue_span["attributes"]["sentry.op"] == "queue.submit.huey"
         assert task1_enqueue_span["attributes"]["sentry.origin"] == "auto.queue.huey"
 
         assert task2_enqueue_span["name"] == "task2"
         assert task2_enqueue_span["status"] == "ok"
         assert task2_enqueue_span["parent_span_id"] == group_span["span_id"]
-        assert (
-            task2_enqueue_span["attributes"]["sentry.segment.name"] == "Huey Task Group"
-        )
+        assert task2_enqueue_span["attributes"]["sentry.segment.name"] == "submission"
         assert task2_enqueue_span["attributes"]["sentry.op"] == "queue.submit.huey"
         assert task2_enqueue_span["attributes"]["sentry.origin"] == "auto.queue.huey"
 
@@ -518,29 +520,36 @@ def task2(results):
 
     if has_span_streaming:
         items = capture_items("span")
-        huey.enqueue(chord([task1.s()], task2.s()))
+        with sentry_sdk.traces.start_span(name="submission"):
+            huey.enqueue(chord([task1.s()], task2.s()))
 
         for _ in range(2):
             task = huey.dequeue()
             huey.execute(task)
 
         sentry_sdk.get_client().flush()
-        assert len(items) == 5
+        assert len(items) == 6
 
         (
             task1_enqueue_span,
             chord_span,
+            submission_span,
             task2_enqueue_span,
             task1_execute_span,
             task2_execute_span,
         ) = [i.payload for i in items]
 
-        assert chord_span["is_segment"]
+        # The enqueue happens inside a wrapping span, so the chord producer
+        # tree is created and parented under that segment.
+        assert submission_span["is_segment"]
+        assert submission_span["name"] == "submission"
+        assert not chord_span["is_segment"]
         assert not task1_enqueue_span["is_segment"]
         assert not task2_enqueue_span["is_segment"]
         assert task1_execute_span["is_segment"]
         assert task2_execute_span["is_segment"]
 
+        assert chord_span["parent_span_id"] == submission_span["span_id"]
         assert chord_span["name"] == "Huey Chord"
         assert chord_span["status"] == "ok"
         assert chord_span["attributes"]["sentry.op"] == "queue.submit.huey"
@@ -549,7 +558,7 @@ def task2(results):
         assert task1_enqueue_span["name"] == "task1"
         assert task1_enqueue_span["status"] == "ok"
         assert task1_enqueue_span["parent_span_id"] == chord_span["span_id"]
-        assert task1_enqueue_span["attributes"]["sentry.segment.name"] == "Huey Chord"
+        assert task1_enqueue_span["attributes"]["sentry.segment.name"] == "submission"
         assert task1_enqueue_span["attributes"]["sentry.op"] == "queue.submit.huey"
         assert task1_enqueue_span["attributes"]["sentry.origin"] == "auto.queue.huey"