fix(celery): Propagate user-set headers

Author: sentrivana

sentry_sdk/integrations/celery/__init__.py

@@ -231,6 +231,12 @@ def _update_celery_task_headers(
                 if key.startswith("sentry-"):
                     updated_headers["headers"][key] = value
 
+            # Preserve user-provided custom headers in the inner "headers" dict
+            # so they survive to task.request.headers on the worker (celery#4875).
+            for key, value in original_headers.items():
+                if key != "headers" and key not in updated_headers["headers"]:
+                    updated_headers["headers"][key] = value
+
     return updated_headers
 
 

tests/integrations/celery/test_celery.py

@@ -847,6 +847,62 @@ def test_send_task_wrapped(
     assert span["trace_id"] == kwargs["headers"]["sentry-trace"].split("-")[0]
 
 
+def test_user_custom_headers_accessible_in_task(init_celery):
+    """
+    Regression test for https://github.com/getsentry/sentry-python/issues/5566
+
+    User-provided custom headers passed to apply_async() must be accessible
+    via task.request.headers on the worker side.
+    """
+    celery = init_celery(traces_sample_rate=1.0)
+
+    @celery.task(name="custom_headers_task", bind=True)
+    def custom_headers_task(self):
+        return dict(self.request.headers or {})
+
+    custom_headers = {
+        "my_custom_key": "my_value",
+        "correlation_id": "abc-123",
+        "tenant_id": "tenant-42",
+    }
+
+    with start_transaction(name="test"):
+        result = custom_headers_task.apply_async(headers=custom_headers)
+
+    received_headers = result.get()
+    for key, value in custom_headers.items():
+        assert received_headers.get(key) == value, (
+            f"Custom header {key!r} not found in task.request.headers"
+        )
+
+
+def test_user_custom_headers_do_not_overwrite_sentry_headers(init_celery):
+    """
+    If a user passes a header with a key that collides with a Sentry header,
+    the Sentry-generated value must take precedence.
+    """
+    celery = init_celery(traces_sample_rate=1.0)
+
+    @celery.task(name="headers_precedence_task", bind=True)
+    def headers_precedence_task(self):
+        return dict(self.request.headers or {})
+
+    with start_transaction(name="test"):
+        result = headers_precedence_task.apply_async(
+            headers={
+                "sentry-trace": "user-supplied-value",
+                "my_custom_key": "my_value",
+            },
+        )
+
+    received_headers = result.get()
+    # Sentry's own sentry-trace must NOT be the user-supplied value
+    assert received_headers.get("sentry-trace") != "user-supplied-value"
+    assert "sentry-trace" in received_headers
+    # User custom header is still preserved
+    assert received_headers.get("my_custom_key") == "my_value"
+
+
 @pytest.mark.skip(reason="placeholder so that forked test does not come last")
 def test_placeholder():
     """Forked tests must not come last in the module.