UN-2123 [FEAT] Propagate request_id across services and workers

backend/backend/celery_service.py

@@ -27,4 +27,8 @@
 app.config_from_object("backend.celery_config.CeleryConfig")
 app.autodiscover_tasks()
 
+# Register signal handlers (e.g. request_id propagation onto published tasks).
+# Importing the module connects the @before_task_publish handler.
+import backend.celery_signals  # noqa: E402, F401
+
 logger.debug(f"Celery Configuration:\n {pformat(app.conf.table(with_defaults=True))}")

backend/backend/celery_signals.py

@@ -0,0 +1,82 @@
+"""Celery signal handlers carrying the HTTP ``request_id`` onto published tasks.
+
+The id travels in the task message headers under ``Common.REQUEST_ID``. Hooking
+``before_task_publish`` rather than each producer is deliberate: it covers every
+``send_task`` / ``.delay`` / ``.apply_async`` with no per-call-site change.
+"""
+
+import logging
+
+from account_v2.constants import Common
+from celery.signals import before_task_publish, task_postrun, task_prerun
+from log_request_id import local as log_request_id_local
+from utils.local_context import StateStore
+
+logger = logging.getLogger(__name__)
+
+
+@before_task_publish.connect
+def propagate_request_id(headers=None, **kwargs):
+    """Inject the current request_id into the outgoing task's message headers.
+
+    Relies on firing in the *producer* thread, where ``StateStore`` still holds
+    the id set by ``CustomRequestIDMiddleware``. No-ops without one (e.g. beat
+    publishes), leaving the worker to derive its own correlation id.
+    """
+    if headers is None:
+        return
+    try:
+        request_id = StateStore.get(Common.REQUEST_ID)
+    except Exception:
+        # StateStore can raise if CONCURRENCY_MODE is misconfigured; never let
+        # correlation plumbing break task publishing.
+        logger.debug("Unable to read request_id from StateStore", exc_info=True)
+        return
+    if request_id and not headers.get(Common.REQUEST_ID):
+        headers[Common.REQUEST_ID] = request_id
+
+
+def _request_id_from_task(task) -> str | None:
+    request = getattr(task, "request", None)
+    if request is None:
+        return None
+    request_id = getattr(request, Common.REQUEST_ID, None)
+    if not request_id:
+        task_headers = getattr(request, "headers", None)
+        if isinstance(task_headers, dict):
+            request_id = task_headers.get(Common.REQUEST_ID)
+    return request_id or None
+
+
+@task_prerun.connect
+def bind_request_id(task=None, **kwargs):
+    """Bind the propagated request_id for tasks run by the backend's own workers.
+
+    Two writes, both required: ``log_request_id``'s thread-local is the only
+    thing ``log_request_id.filters.RequestIDFilter`` reads (the HTTP middleware
+    is otherwise its sole writer, so a task would log ``request_id:-``), and
+    ``StateStore`` is what ``propagate_request_id`` reads, so tasks this worker
+    itself publishes carry the id onward.
+    """
+    request_id = _request_id_from_task(task)
+    if not request_id:
+        return
+    log_request_id_local.request_id = request_id
+    try:
+        StateStore.set(Common.REQUEST_ID, request_id)
+    except Exception:
+        logger.debug("Unable to set request_id on StateStore", exc_info=True)
+
+
+@task_postrun.connect
+def clear_request_id(**kwargs):
+    """Clear the task-scoped request_id -- worker threads are pooled and reused."""
+    if hasattr(log_request_id_local, "request_id"):
+        try:
+            del log_request_id_local.request_id
+        except AttributeError:
+            pass
+    try:
+        StateStore.clear(Common.REQUEST_ID)
+    except Exception:
+        logger.debug("Unable to clear request_id from StateStore", exc_info=True)

backend/backend/settings/base.py

@@ -276,10 +276,18 @@ def get_required_setting(setting_key: str, default: str | None = None) -> str |
 CORS_ALLOW_ALL_ORIGINS = False
 
 # Request ID middleware settings
-LOG_REQUEST_ID_HEADER = "X-Request-ID"
+# django-log-request-id resolves this via request.META.get(...), where WSGI exposes
+# the incoming "X-Request-ID" header as the HTTP_-prefixed key HTTP_X_REQUEST_ID.
+# It MUST be the META key, not the raw header name, or the incoming id is never read
+# and a fresh one is minted on every request (breaking client/worker correlation).
+LOG_REQUEST_ID_HEADER = "HTTP_X_REQUEST_ID"
 REQUEST_ID_RESPONSE_HEADER = "X-Request-ID"
 GENERATE_REQUEST_ID_IF_NOT_IN_HEADER = True
 NO_REQUEST_ID = "-"
+# The frontend is a separate origin, so the browser hides every response header
+# not named here -- without this the id the backend logged is unreadable in JS
+# and the error toast falls back to showing an id that appears in no log.
+CORS_EXPOSE_HEADERS = [REQUEST_ID_RESPONSE_HEADER]
 
 
 class OTelFieldFilter(logging.Filter):

backend/backend/test_celery_signals.py

@@ -0,0 +1,161 @@
+"""Unit checks for the backend's Celery request_id signal handlers.
+
+These are the producer half of the correlation chain: the HTTP request binds an
+id (see ``middleware/test_request_id.py``), and these handlers carry it onto
+every task the backend publishes and back off the message for tasks the
+backend's *own* Celery workers run.
+
+The handlers are written to fail open -- correlation plumbing must never break
+task publishing -- which means a regression here is silent by construction: the
+header simply stops being set and every worker log line reverts to
+``request_id:-``. Nothing raises, so only assertions catch it.
+
+Pure-logic: ``StateStore`` is a thread-local and the signals are called
+directly, so no broker, worker or DB is involved.
+"""
+
+import pytest
+from account_v2.constants import Common
+from log_request_id import local as log_request_id_local
+from utils.local_context import StateStore
+
+from backend.celery_signals import (
+    bind_request_id,
+    clear_request_id,
+    propagate_request_id,
+)
+
+REQUEST_ID = "6f1c2d3e-4a5b-4c6d-8e9f-0a1b2c3d4e5f"
+
+
+class _Request:
+    """Stands in for a Celery task ``Context``."""
+
+    def __init__(self, request_id=None, headers=None):
+        if request_id is not None:
+            self.request_id = request_id
+        self.headers = headers
+
+
+class _Task:
+    def __init__(self, request=None):
+        self.request = request
+
+
+@pytest.fixture(autouse=True)
+def _clean_context():
+    """Both stores are thread-local and shared across tests in a worker, so a
+    leaked id would make a later test pass for the wrong reason.
+    """
+    yield
+    for store_key in (Common.REQUEST_ID,):
+        try:
+            StateStore.clear(store_key)
+        except AttributeError:
+            pass
+    if hasattr(log_request_id_local, "request_id"):
+        del log_request_id_local.request_id
+
+
+# Producer side: before_task_publish
+
+
+def test_publish_injects_request_id_from_state_store():
+    """The id bound by the HTTP middleware rides out on the task message."""
+    StateStore.set(Common.REQUEST_ID, REQUEST_ID)
+    headers = {}
+
+    propagate_request_id(headers=headers)
+
+    assert headers[Common.REQUEST_ID] == REQUEST_ID
+
+
+def test_publish_is_a_noop_without_a_request_id():
+    """Beat-scheduled publishes have no request in scope; the worker falls back
+    to its own execution_id/task_id rather than receiving an empty header.
+    """
+    headers = {}
+
+    propagate_request_id(headers=headers)
+
+    assert headers == {}
+
+
+def test_publish_does_not_clobber_an_explicit_header():
+    """A caller that set the header deliberately outranks the ambient id."""
+    StateStore.set(Common.REQUEST_ID, REQUEST_ID)
+    headers = {Common.REQUEST_ID: "caller-supplied"}
+
+    propagate_request_id(headers=headers)
+
+    assert headers[Common.REQUEST_ID] == "caller-supplied"
+
+
+def test_publish_tolerates_missing_headers():
+    """``before_task_publish`` must never raise -- it would break the publish
+    itself, taking the actual task down with the correlation plumbing.
+    """
+    propagate_request_id(headers=None)  # does not raise
+
+
+# Consumer side: task_prerun / task_postrun on the backend's own workers
+
+
+def test_prerun_binds_id_from_task_attribute():
+    """Celery surfaces the custom header as an attribute on the Context."""
+    bind_request_id(task=_Task(_Request(request_id=REQUEST_ID)))
+
+    assert log_request_id_local.request_id == REQUEST_ID
+    assert StateStore.get(Common.REQUEST_ID) == REQUEST_ID
+
+
+def test_prerun_falls_back_to_raw_headers_mapping():
+    """Fallback for when Celery leaves the header only in the raw mapping."""
+    bind_request_id(task=_Task(_Request(headers={Common.REQUEST_ID: REQUEST_ID})))
+
+    assert log_request_id_local.request_id == REQUEST_ID
+
+
+def test_prerun_binding_makes_the_id_re_propagate():
+    """Second-order effect that motivated the handler: a task the backend
+    worker itself publishes must carry the inherited id onward, or the chain
+    dies at the first backend-worker hop.
+    """
+    bind_request_id(task=_Task(_Request(request_id=REQUEST_ID)))
+    headers = {}
+
+    propagate_request_id(headers=headers)
+
+    assert headers[Common.REQUEST_ID] == REQUEST_ID
+
+
+def test_prerun_without_an_id_leaves_stores_untouched():
+    """No header means no correlation to inherit -- and, importantly, no empty
+    string bound over whatever the logger would otherwise show.
+    """
+    bind_request_id(task=_Task(_Request()))
+
+    assert not hasattr(log_request_id_local, "request_id")
+    assert StateStore.get(Common.REQUEST_ID) is None
+
+
+def test_prerun_tolerates_a_task_without_a_request():
+    bind_request_id(task=_Task(request=None))  # does not raise
+
+    assert not hasattr(log_request_id_local, "request_id")
+
+
+def test_postrun_clears_the_bound_id():
+    """Celery reuses worker threads; a surviving id would mislabel the *next*
+    task's logs with the previous task's correlation id.
+    """
+    bind_request_id(task=_Task(_Request(request_id=REQUEST_ID)))
+
+    clear_request_id()
+
+    assert not hasattr(log_request_id_local, "request_id")
+    assert StateStore.get(Common.REQUEST_ID) is None
+
+
+def test_postrun_is_safe_when_nothing_was_bound():
+    clear_request_id()  # does not raise

backend/middleware/request_id.py

@@ -2,7 +2,40 @@
 
 from log_request_id.middleware import RequestIDMiddleware
 
+# The internal service boundary: our own workers call back here while executing
+# a workflow, forwarding the request_id of the HTTP call that started it.
+INTERNAL_PATH_PREFIX = "/internal/"
+
+
+def _canonical_uuid(value: str | None) -> str | None:
+    """Return ``value`` only if it is a canonical hyphenated UUID."""
+    try:
+        return value if str(uuid.UUID(value)) == value else None
+    except (AttributeError, TypeError, ValueError):
+        return None
+
 
 class CustomRequestIDMiddleware(RequestIDMiddleware):
+    """Provisions the request id here rather than trusting the caller's.
+
+    Adopting a caller-supplied id lets any client repeat one value across
+    unrelated requests. That does not merely lose correlation -- a log query for
+    that id returns other tenants' requests and reads as though it worked, which
+    is worse than returning nothing. The id also reaches every log line, Celery
+    message header and outbound internal-API call, so an unvalidated one can
+    forge a log record outright.
+
+    The internal boundary is the exception: there the caller is our own worker
+    forwarding the id of the request that started the execution, which is the
+    hop that makes worker logs correlate back to the originating call.
+    """
+
+    def _get_request_id(self, request):
+        if request.path.startswith(INTERNAL_PATH_PREFIX):
+            forwarded = _canonical_uuid(request.META.get(self.request_id_header))
+            if forwarded:
+                return forwarded
+        return self._generate_id()
+
     def _generate_id(self):
         return str(uuid.uuid4())