send task_received signal before receive log

CONTRIBUTORS.txt

@@ -295,5 +295,6 @@ JoonHwan Kim, 2022/08/01
 Kaustav Banerjee, 2022/11/10
 Austin Snoeyink 2022/12/06
 Jeremy Z. Othieno 2023/07/27
+Nikolai Vidov 2023/12/12
 Tomer Nosrati, 2022/17/07
-Andy Zickler, 2024/01/18
+Andy Zickler, 2024/01/18

celery/worker/strategy.py

@@ -148,17 +148,7 @@ def task_message_handler(message, body, ack, reject, callbacks,
             eventer=eventer, task=task, connection_errors=connection_errors,
             body=body, headers=headers, decoded=decoded, utc=utc,
         )
-        if _does_info:
-            # Similar to `app.trace.info()`, we pass the formatting args as the
-            # `extra` kwarg for custom log handlers
-            context = {
-                'id': req.id,
-                'name': req.name,
-                'args': req.argsrepr,
-                'kwargs': req.kwargsrepr,
-                'eta': req.eta,
-            }
-            info(_app_trace.LOG_RECEIVED, context, extra={'data': context})
+
         if (req.expires or req.id in revoked_tasks) and req.revoked():
             return
 
@@ -175,6 +165,18 @@ def task_message_handler(message, body, ack, reject, callbacks,
                 expires=req.expires and req.expires.isoformat(),
             )
 
+        if _does_info:
+            # Similar to `app.trace.info()`, we pass the formatting args as the
+            # `extra` kwarg for custom log handlers
+            context = {
+                'id': req.id,
+                'name': req.name,
+                'args': req.argsrepr,
+                'kwargs': req.kwargsrepr,
+                'eta': req.eta,
+            }
+            info(_app_trace.LOG_RECEIVED, context, extra={'data': context})
+
         bucket = None
         eta = None
         if req.eta:

t/unit/worker/test_strategy.py

@@ -234,6 +234,35 @@ def test_signal_task_received(self):
                                              request=ANY,
                                              signal=signals.task_received)
 
+    def test_log_task_received_meta(self, caplog):
+        class Formatter:
+            def __init__(self, base_formatter):
+                self.base_formatter = base_formatter
+
+            def format(self, record):
+                record.foo = "bar"
+                return self.base_formatter.format(record)
+
+        old_formatter = caplog.handler.formatter
+
+        def callback(*args, **kwargs):
+            caplog.handler.setFormatter(Formatter(old_formatter))
+
+        caplog.set_level(logging.INFO, logger="celery.worker.strategy")
+
+        with self._context(
+            self.add.s(2, 2)
+        ) as C:
+            signals.task_received.connect(callback)
+            C()
+        for record in caplog.records:
+            if record.msg == LOG_RECEIVED:
+                assert record.foo == "bar"
+                caplog.handler.setFormatter(old_formatter)
+                break
+        else:
+            raise ValueError("Expected message not in captured log records")
+
     def test_when_events_disabled(self):
         with self._context(self.add.s(2, 2), events=False) as C:
             C()

tox.ini

@@ -33,7 +33,7 @@ deps=
 
     3.8,3.9,3.10,3.11,3.12: -r{toxinidir}/requirements/test-ci-default.txt
     3.8,3.9,3.10,3.11,3.12: -r{toxinidir}/requirements/docs.txt
-    pypy3: -r{toxinidir}/requirements/test-ci-default.txt
+    pypy-3,pypy3: -r{toxinidir}/requirements/test-ci-default.txt
 
     integration: -r{toxinidir}/requirements/test-integration.txt
     smoke: pytest-xdist>=3.5