From 908a0477ebdce8d16862e06e1f5d73f59cb0caf0 Mon Sep 17 00:00:00 2001 From: Ivana Kellyer Date: Mon, 13 Jul 2026 15:55:37 +0200 Subject: [PATCH 1/3] fix(tracing): Skip child span creation in streaming path when no current span (task queues) When span streaming is enabled and there is no current span, task queue integrations should not create new root segments for child-span operations. This adds a guard check using `sentry_sdk.traces.get_current_span()` before creating spans in the streaming path. Affected integrations: celery, huey, arq, ray. Co-Authored-By: Claude Opus 4.6 --- sentry_sdk/integrations/arq.py | 3 +++ sentry_sdk/integrations/celery/__init__.py | 6 +++++- sentry_sdk/integrations/huey.py | 16 ++++++++++++---- sentry_sdk/integrations/ray.py | 17 +++++++++++++++++ 4 files changed, 37 insertions(+), 5 deletions(-) diff --git a/sentry_sdk/integrations/arq.py b/sentry_sdk/integrations/arq.py index 3488457e20..011c5e51bf 100644 --- a/sentry_sdk/integrations/arq.py +++ b/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={ diff --git a/sentry_sdk/integrations/celery/__init__.py b/sentry_sdk/integrations/celery/__init__.py index 532b13539b..a4abac4618 100644 --- a/sentry_sdk/integrations/celery/__init__.py +++ b/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(): diff --git a/sentry_sdk/integrations/huey.py b/sentry_sdk/integrations/huey.py index a1c9d6a357..70814d4b7f 100644 --- a/sentry_sdk/integrations/huey.py +++ b/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 diff --git a/sentry_sdk/integrations/ray.py b/sentry_sdk/integrations/ray.py index f723a96f3c..f48bd59791 100644 --- a/sentry_sdk/integrations/ray.py +++ b/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 From e09daea02ec2f8957c36006f2a88a4e8f496eea0 Mon Sep 17 00:00:00 2001 From: Erica Pisani Date: Tue, 14 Jul 2026 11:38:14 -0400 Subject: [PATCH 2/3] test(huey): Update streaming tests for enqueue guard When span streaming is enabled and enqueue is called without a wrapping span, the producer (queue.submit.huey) span is intentionally not created (early return in patch_enqueue). Update the streaming test expectations: - Single-task tests (transaction/segment, cancel, lock) now assert only the consumer segment is emitted. - Retry test accounts for the initial enqueue producing no span while the re-enqueue (inside the running consumer segment) still gets a child span. - Group and chord tests wrap enqueue in an active span, reflecting real usage, so the full producer tree is created and parented under it. --- tests/integrations/huey/test_huey.py | 73 ++++++++++++++++------------ 1 file changed, 41 insertions(+), 32 deletions(-) diff --git a/tests/integrations/huey/test_huey.py b/tests/integrations/huey/test_huey.py index bfd04791dd..93e24a96d6 100644 --- a/tests/integrations/huey/test_huey.py +++ b/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" From e4213a3242eaeeac82fa897de27d97a2a2a12b54 Mon Sep 17 00:00:00 2001 From: Erica Pisani Date: Tue, 14 Jul 2026 11:38:18 -0400 Subject: [PATCH 3/3] test(arq): Update streaming tests for enqueue guard The retry re-enqueue happens without an active span, so no producer (queue.submit.arq) span is created for it. Adjust the positional span assertions in test_job_retry and test_span_origin_consumer to account for the suppressed producer span. --- tests/integrations/arq/test_arq.py | 25 ++++++++++++++++--------- 1 file changed, 16 insertions(+), 9 deletions(-) diff --git a/tests/integrations/arq/test_arq.py b/tests/integrations/arq/test_arq.py index fb40018d2a..013ea83167 100644 --- a/tests/integrations/arq/test_arq.py +++ b/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")