From f8b9b75b8071d664734e07a8c15806374bacb571 Mon Sep 17 00:00:00 2001
From: rbehal <rahulbehal01@hotmail.com>
Date: Mon, 14 Sep 2026 19:42:46 +0000
Subject: [PATCH 1/7] Keep busy pool workers and in-flight reservations across
 a broker reconnect

Backport of upstream celery/celery#10346 and celery/celery#10345 onto the
5.6.3+gumloop_0.2.0 base.

On a broker connection loss Consumer.start() calls on_close(), which wipes
reserved_requests and calls AsynPool.flush(). flush() ended by clearing
_busy_workers unconditionally, even for pool children still executing an
accepted job. _busy_workers is the only state the fair scheduler consults
(on_poll_start / schedule_writes), so after a reconnect the next received
task could be written into the pipe of a child that was still mid-task and
sit there until that task finished, while other children stayed idle.
Clearing reserved_requests likewise let the worker_disable_prefetch gate
under-count the slots in use.

- AsynPool.flush(): trim _busy_workers to the inqueue fds of children that
  still hold an accepted job and are alive, instead of clearing the set.
- Consumer.on_close(): keep active requests in reserved_requests and only
  drop the bookkeeping of requests that were never accepted.

Regression tests from the upstream commits are included; the upstream
change to the RabbitMQ quorum-queue smoke test is not (not used here).
---
 celery/concurrency/asynpool.py     |  33 +++++-
 celery/worker/consumer/consumer.py |   7 +-
 t/unit/concurrency/test_prefork.py | 172 +++++++++++++++++++++++++++++
 t/unit/worker/test_consumer.py     | 124 +++++++++++++++++++++
 4 files changed, 332 insertions(+), 4 deletions(-)

diff --git a/celery/concurrency/asynpool.py b/celery/concurrency/asynpool.py
index 10783847a96..276ff91eb6a 100644
--- a/celery/concurrency/asynpool.py
+++ b/celery/concurrency/asynpool.py
@@ -1080,7 +1080,38 @@ def flush(self):
             self.outbound_buffer.clear()
             self._active_writers.clear()
             self._active_writes.clear()
-            self._busy_workers.clear()
+
+            # Trim _busy_workers down to the workers still executing an
+            # accepted (running) job instead of clearing it outright, so the
+            # fair scheduler keeps avoiding mid-task workers across a broker
+            # reconnect. A worker is marked busy keyed on
+            # job._scheduled_for.inqW_fd at write time; once the body has been
+            # written job._write_to points at the same process, so prefer it
+            # (the process actually executing the job) and fall back to
+            # _scheduled_for.
+            #
+            # intersection_update only ever removes fds from _busy_workers,
+            # never adds one (it filters the existing set rather than
+            # repopulating it), so this trim can't fabricate a phantom-busy
+            # entry that would permanently sideline a healthy worker. Avoiding
+            # double-booking (writing a second task onto a still-running
+            # worker) instead depends on still_busy including every accepted
+            # job's executing fd, which is why we source it from
+            # _write_to/_scheduled_for of the running job above.
+            #
+            # The proc._is_alive() guard keeps the set limited to workers that
+            # are actually still executing: a genuinely running accepted job
+            # always has a live worker, while a worker that died (whose fd may
+            # since have been reused by a fresh replacement) must not be
+            # re-counted as busy.
+            still_busy = set()
+            for job in self._cache.values():
+                if not job._accepted:
+                    continue
+                proc = job._write_to or job._scheduled_for
+                if proc is not None and proc._is_alive():
+                    still_busy.add(proc.inqW_fd)
+            self._busy_workers.intersection_update(still_busy)
 
     def _flush_writer(self, proc, writer):
         fds = {proc.inq._writer}
diff --git a/celery/worker/consumer/consumer.py b/celery/worker/consumer/consumer.py
index 2a5d955fef3..cc41e44e101 100644
--- a/celery/worker/consumer/consumer.py
+++ b/celery/worker/consumer/consumer.py
@@ -459,10 +459,11 @@ def on_close(self):
         for bucket in self.task_buckets.values():
             if bucket:
                 bucket.clear_pending()
-        for request_id in reserved_requests:
-            if request_id in requests:
-                del requests[request_id]
+        for r in tuple(reserved_requests):
+            if r not in active_requests:
+                requests.pop(r.id, None)
         reserved_requests.clear()
+        reserved_requests.update(tuple(active_requests))
         if self.pool and self.pool.flush:
             self.pool.flush()
 
diff --git a/t/unit/concurrency/test_prefork.py b/t/unit/concurrency/test_prefork.py
index 8f266f78302..afd0c2dff57 100644
--- a/t/unit/concurrency/test_prefork.py
+++ b/t/unit/concurrency/test_prefork.py
@@ -735,6 +735,178 @@ def test_flush_not_started_write_job_is_discarded(self, _create_worker_process):
         job.discard.assert_called_once()
         assert gen not in pool._active_writers
 
+    @t.skip.if_pypy
+    @patch('billiard.pool.Pool._create_worker_process')
+    def test_flush_preserves_busy_worker_for_accepted_job(self, _create_worker_process):
+        """flush() must keep a worker marked busy while it runs an accepted job.
+
+        On a broker reconnect Consumer.on_close calls pool.flush(). A worker
+        still executing an accepted (running) task is genuinely busy, so its
+        inqueue write-fd must remain in _busy_workers. Clearing it would
+        desynchronize the fair scheduler from reality and let a new task be
+        written onto the busy worker, blocking it behind the long-running task
+        even while another worker is idle.
+        """
+        pool = asynpool.AsynPool(processes=2, synack=False, threads=False)
+        pool._state = asynpool.RUN
+        pool.maintain_pool = Mock(name='maintain_pool')
+
+        # Worker still running an accepted job, dispatched to fd 7. Once the
+        # body is written job._write_to points at the executing process, so it
+        # is the preferred source of truth for the busy fd.
+        proc = Mock(name='proc')
+        proc.inqW_fd = 7
+        proc._is_alive.return_value = True
+        job = Mock(name='job')
+        job._accepted = True
+        job._write_to = proc
+        job._scheduled_for = proc
+        job._writer.return_value = None
+
+        pool._cache = {1: job}
+        pool._busy_workers = {7}
+        pool._active_writers.clear()
+        pool.outbound_buffer.clear()
+
+        pool.flush()
+
+        # The busy worker must still be marked busy after the flush.
+        assert 7 in pool._busy_workers
+
+    @t.skip.if_pypy
+    @patch('billiard.pool.Pool._create_worker_process')
+    def test_flush_preserves_busy_worker_via_scheduled_for_fallback(self, _create_worker_process):
+        """flush() falls back to _scheduled_for when _write_to is unset.
+
+        A job can be accepted before its body finished writing, leaving
+        _write_to unset while _scheduled_for already identifies the worker the
+        task was dispatched to. The busy fd must still be preserved.
+        """
+        pool = asynpool.AsynPool(processes=2, synack=False, threads=False)
+        pool._state = asynpool.RUN
+        pool.maintain_pool = Mock(name='maintain_pool')
+
+        proc = Mock(name='proc')
+        proc.inqW_fd = 5
+        proc._is_alive.return_value = True
+        job = Mock(name='job')
+        job._accepted = True
+        job._write_to = None
+        job._scheduled_for = proc
+        job._writer.return_value = None
+
+        pool._cache = {1: job}
+        pool._busy_workers = {5}
+        pool._active_writers.clear()
+        pool.outbound_buffer.clear()
+
+        pool.flush()
+
+        assert 5 in pool._busy_workers
+
+    @t.skip.if_pypy
+    @patch('billiard.pool.Pool._create_worker_process')
+    def test_flush_releases_busy_worker_for_unaccepted_job(self, _create_worker_process):
+        """flush() must free a worker whose job was not accepted yet.
+
+        Unaccepted jobs are discarded/redelivered by the broker, so the worker
+        they were tentatively scheduled to is no longer busy and its fd must be
+        dropped from _busy_workers.
+        """
+        pool = asynpool.AsynPool(processes=2, synack=False, threads=False)
+        pool._state = asynpool.RUN
+        pool.maintain_pool = Mock(name='maintain_pool')
+
+        proc = Mock(name='proc')
+        proc.inqW_fd = 9
+        job = Mock(name='job')
+        job._accepted = False
+        job._write_to = proc
+        job._scheduled_for = proc
+        job._writer.return_value = None
+
+        pool._cache = {1: job}
+        pool._busy_workers = {9}
+        pool._active_writers.clear()
+        pool.outbound_buffer.clear()
+
+        pool.flush()
+
+        assert 9 not in pool._busy_workers
+
+    @t.skip.if_pypy
+    @patch('billiard.pool.Pool._create_worker_process')
+    def test_flush_preserves_accepted_and_releases_unaccepted_together(self, _create_worker_process):
+        """flush() must resolve a mixed _busy_workers set in a single call.
+
+        When one worker runs an accepted job and another only had an
+        unaccepted job tentatively scheduled to it, the same flush() must keep
+        the accepted worker busy while releasing the unaccepted one, rather
+        than treating the set all-or-nothing.
+        """
+        pool = asynpool.AsynPool(processes=2, synack=False, threads=False)
+        pool._state = asynpool.RUN
+        pool.maintain_pool = Mock(name='maintain_pool')
+
+        accepted_proc = Mock(name='accepted_proc')
+        accepted_proc.inqW_fd = 7
+        accepted_proc._is_alive.return_value = True
+        accepted_job = Mock(name='accepted_job')
+        accepted_job._accepted = True
+        accepted_job._write_to = accepted_proc
+        accepted_job._scheduled_for = accepted_proc
+        accepted_job._writer.return_value = None
+
+        unaccepted_proc = Mock(name='unaccepted_proc')
+        unaccepted_proc.inqW_fd = 9
+        unaccepted_job = Mock(name='unaccepted_job')
+        unaccepted_job._accepted = False
+        unaccepted_job._write_to = unaccepted_proc
+        unaccepted_job._scheduled_for = unaccepted_proc
+        unaccepted_job._writer.return_value = None
+
+        pool._cache = {1: accepted_job, 2: unaccepted_job}
+        pool._busy_workers = {7, 9}
+        pool._active_writers.clear()
+        pool.outbound_buffer.clear()
+
+        pool.flush()
+
+        assert 7 in pool._busy_workers
+        assert 9 not in pool._busy_workers
+
+    @t.skip.if_pypy
+    @patch('billiard.pool.Pool._create_worker_process')
+    def test_flush_releases_busy_worker_for_dead_process(self, _create_worker_process):
+        """flush() must not keep an accepted job's fd if its worker died.
+
+        If the worker executing an accepted job has exited, its inqueue
+        write-fd may be reused by a replacement process. Keeping the fd marked
+        busy would sideline that healthy replacement, so a dead worker's fd is
+        dropped from _busy_workers.
+        """
+        pool = asynpool.AsynPool(processes=2, synack=False, threads=False)
+        pool._state = asynpool.RUN
+        pool.maintain_pool = Mock(name='maintain_pool')
+
+        proc = Mock(name='proc')
+        proc.inqW_fd = 3
+        proc._is_alive.return_value = False
+        job = Mock(name='job')
+        job._accepted = True
+        job._write_to = proc
+        job._scheduled_for = proc
+        job._writer.return_value = None
+
+        pool._cache = {1: job}
+        pool._busy_workers = {3}
+        pool._active_writers.clear()
+        pool.outbound_buffer.clear()
+
+        pool.flush()
+
+        assert 3 not in pool._busy_workers
+
     def test_process_result(self):
         x = asynpool.ResultHandler(
             Mock(), Mock(), {}, Mock(),
diff --git a/t/unit/worker/test_consumer.py b/t/unit/worker/test_consumer.py
index 8cdb908795f..1181f0efbfe 100644
--- a/t/unit/worker/test_consumer.py
+++ b/t/unit/worker/test_consumer.py
@@ -389,6 +389,49 @@ def test_on_close_clears_semaphore_and_reqs(self):
             c.pool = None
             c.on_close()
 
+    def test_on_close_purges_orphan_reservations_from_requests_dict(self):
+        """Regression: ``on_close()`` must remove ``state.requests[id]``
+        entries for Requests that were reserved but never accepted (e.g.
+        ETA tasks queued in ``reserved_requests`` at the moment of a
+        connection loss). PR #7771 attempted this but iterated
+        ``reserved_requests`` (Request objects) and tested membership in
+        ``requests`` (a ``dict[str, Request]``), so ``Request in requests``
+        was always False and nothing was ever deleted.
+        """
+        from celery.worker import state
+        from celery.worker.consumer.consumer import Consumer
+
+        class FakeRequest:
+            def __init__(self, id):
+                self.id = id
+
+        consumer = Mock()
+        consumer.controller = Mock()
+        consumer.controller.semaphore = Mock()
+        consumer.task_buckets = {}
+        consumer.pool = Mock()
+        consumer.pool.flush = Mock()
+
+        state.reset_state()
+        try:
+            # Orphan reservation: present in ``requests`` and
+            # ``reserved_requests`` but never moved to ``active_requests``.
+            orphan = FakeRequest('orphan-1')
+            state.requests[orphan.id] = orphan
+            state.reserved_requests.add(orphan)
+
+            Consumer.on_close(consumer)
+
+            assert orphan.id not in state.requests, (
+                "on_close() did not purge an orphan reserved-but-not-"
+                "accepted Request from state.requests; this is the leak "
+                "PR #7771 tried to fix but its loop variable ('request_id') "
+                "actually held Request objects, so the membership check "
+                "never matched."
+            )
+        finally:
+            state.reset_state()
+
     def test_connect_error_handler(self):
         self.app._connection = _amqp_connection()
         conn = self.app._connection.return_value
@@ -685,6 +728,87 @@ def test_disable_prefetch_respects_autoscale_max_concurrency(self):
             # Should not be able to consume when at autoscale limit
             assert consumer.task_consumer.channel.qos.can_consume() is False
 
+    def test_disable_prefetch_after_connection_loss_keeps_gate_closed(self):
+        """Regression: ``can_consume`` must still refuse new messages while a
+        task from before a broker reconnect is still running in the pool.
+
+        With ``worker_disable_prefetch`` enabled and concurrency=1, after a
+        channel/connection interruption ``Consumer.on_close()`` clears
+        ``state.reserved_requests``. The disable-prefetch gate then sees an
+        empty set and lets a new message through, even though the pool is
+        still busy with the in-flight task. The gate must remain closed.
+        """
+        from celery.worker import state
+        from celery.worker.consumer.consumer import Consumer
+        from celery.worker.consumer.tasks import Tasks
+
+        self.app.conf.worker_disable_prefetch = True
+
+        # Plain class so it supports weakref (Mock() does not, and WeakSet
+        # would silently reject it).
+        class FakeRequest:
+            def __init__(self, id):
+                self.id = id
+
+        consumer = Mock()
+        consumer.app = self.app
+        consumer.pool = Mock()
+        consumer.pool.num_processes = 1
+        consumer.pool.flush = Mock()
+        consumer.controller = Mock()
+        consumer.controller.max_concurrency = None
+        consumer.controller.semaphore = Mock()
+        consumer.task_buckets = {}
+        consumer.initial_prefetch_count = 1
+        consumer.connection = Mock()
+        consumer.connection.connection_errors = ()
+        consumer.connection.channel_errors = ()
+        consumer.connection.default_channel = Mock()
+        consumer.connection.transport = Mock()
+        consumer.connection.transport.driver_type = 'redis'
+        consumer.update_strategies = Mock()
+        consumer.on_decode_error = Mock()
+
+        consumer.task_consumer = Mock()
+        consumer.task_consumer.channel = Mock()
+        consumer.task_consumer.channel.qos = Mock()
+        consumer.task_consumer.channel.qos.can_consume = Mock(return_value=True)
+        consumer.task_consumer.qos = Mock()
+        consumer.app.amqp = Mock()
+        consumer.app.amqp.TaskConsumer = Mock(return_value=consumer.task_consumer)
+
+        # Install the disable-prefetch gate on the (mocked) channel.
+        Tasks(consumer).start(consumer)
+        can_consume = consumer.task_consumer.channel.qos.can_consume
+
+        state.reset_state()
+        try:
+            # One task running in the pool: by the normal lifecycle
+            # (task_reserved + task_accepted) it lives in BOTH sets.
+            in_flight = FakeRequest('task-1')
+            state.reserved_requests.add(in_flight)
+            state.active_requests.add(in_flight)
+
+            # Pre-condition: gate refuses; concurrency is 1 and slot is taken.
+            assert can_consume() is False
+
+            # Simulate a broker channel/connection interruption.
+            Consumer.on_close(consumer)
+
+            # Pool still has the task running — active_requests is
+            # intentionally not cleared by on_close(). The gate must still
+            # refuse so we don't over-fetch a second message into a
+            # concurrency=1 pool.
+            assert in_flight in state.active_requests
+            assert can_consume() is False, (
+                "After a connection loss, can_consume() returned True while "
+                "a task is still running in the pool. Consumer.on_close() "
+                "cleared reserved_requests, hiding the in-flight task from "
+                "the worker_disable_prefetch gate."
+            )
+        finally:
+            state.reset_state()
+
     def test_disable_prefetch_ignored_for_non_redis_brokers(self):
         """Test that disable_prefetch is ignored for non-Redis brokers."""
         self.app.conf.worker_disable_prefetch = True

From eec3d0bf1b48a68c640fd1a4940d15f307699c87 Mon Sep 17 00:00:00 2001
From: rbehal <rahulbehal01@hotmail.com>
Date: Mon, 14 Sep 2026 19:42:46 +0000
Subject: [PATCH 2/7] Bump version to 5.6.3+gumloop_0.2.1

---
 celery/__init__.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/celery/__init__.py b/celery/__init__.py
index 3467851685f..07281e693f2 100644
--- a/celery/__init__.py
+++ b/celery/__init__.py
@@ -24,7 +24,7 @@
 
 SERIES = 'recovery'
 
-__version__ = '5.6.3+gumloop_0.2.0'
+__version__ = '5.6.3+gumloop_0.2.1'
 __author__ = 'Rahul Behal'
 __contact__ = 'rahul@gumloop.com'
 __homepage__ = 'https://github.com/gumloop/gumloop-celery'

From cce152df36de6ac61cba2d2caf18fbbc0c1bb77c Mon Sep 17 00:00:00 2001
From: rbehal <rahulbehal01@hotmail.com>
Date: Mon, 14 Sep 2026 19:57:28 +0000
Subject: [PATCH 3/7] Trim backported comments and test docstrings

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 celery/concurrency/asynpool.py     | 25 ++-----------------
 t/unit/concurrency/test_prefork.py | 39 ------------------------------
 t/unit/worker/test_consumer.py     | 31 +-----------------------
 3 files changed, 3 insertions(+), 92 deletions(-)

diff --git a/celery/concurrency/asynpool.py b/celery/concurrency/asynpool.py
index 276ff91eb6a..deac56fa0fa 100644
--- a/celery/concurrency/asynpool.py
+++ b/celery/concurrency/asynpool.py
@@ -1081,29 +1081,8 @@ def flush(self):
             self._active_writers.clear()
             self._active_writes.clear()
 
-            # Trim _busy_workers down to the workers still executing an
-            # accepted (running) job instead of clearing it outright, so the
-            # fair scheduler keeps avoiding mid-task workers across a broker
-            # reconnect. A worker is marked busy keyed on
-            # job._scheduled_for.inqW_fd at write time; once the body has been
-            # written job._write_to points at the same process, so prefer it
-            # (the process actually executing the job) and fall back to
-            # _scheduled_for.
-            #
-            # intersection_update only ever removes fds from _busy_workers,
-            # never adds one (it filters the existing set rather than
-            # repopulating it), so this trim can't fabricate a phantom-busy
-            # entry that would permanently sideline a healthy worker. Avoiding
-            # double-booking (writing a second task onto a still-running
-            # worker) instead depends on still_busy including every accepted
-            # job's executing fd, which is why we source it from
-            # _write_to/_scheduled_for of the running job above.
-            #
-            # The proc._is_alive() guard keeps the set limited to workers that
-            # are actually still executing: a genuinely running accepted job
-            # always has a live worker, while a worker that died (whose fd may
-            # since have been reused by a fresh replacement) must not be
-            # re-counted as busy.
+            # _write_to is only set once the body has been written; a dead
+            # worker's inqW_fd may already have been reused by its replacement.
             still_busy = set()
             for job in self._cache.values():
                 if not job._accepted:
diff --git a/t/unit/concurrency/test_prefork.py b/t/unit/concurrency/test_prefork.py
index afd0c2dff57..7e362d5fcf9 100644
--- a/t/unit/concurrency/test_prefork.py
+++ b/t/unit/concurrency/test_prefork.py
@@ -738,22 +738,10 @@ def test_flush_not_started_write_job_is_discarded(self, _create_worker_process):
     @t.skip.if_pypy
     @patch('billiard.pool.Pool._create_worker_process')
     def test_flush_preserves_busy_worker_for_accepted_job(self, _create_worker_process):
-        """flush() must keep a worker marked busy while it runs an accepted job.
-
-        On a broker reconnect Consumer.on_close calls pool.flush(). A worker
-        still executing an accepted (running) task is genuinely busy, so its
-        inqueue write-fd must remain in _busy_workers. Clearing it would
-        desynchronize the fair scheduler from reality and let a new task be
-        written onto the busy worker, blocking it behind the long-running task
-        even while another worker is idle.
-        """
         pool = asynpool.AsynPool(processes=2, synack=False, threads=False)
         pool._state = asynpool.RUN
         pool.maintain_pool = Mock(name='maintain_pool')
 
-        # Worker still running an accepted job, dispatched to fd 7. Once the
-        # body is written job._write_to points at the executing process, so it
-        # is the preferred source of truth for the busy fd.
         proc = Mock(name='proc')
         proc.inqW_fd = 7
         proc._is_alive.return_value = True
@@ -770,18 +758,11 @@ def test_flush_preserves_busy_worker_for_accepted_job(self, _create_worker_proce
 
         pool.flush()
 
-        # The busy worker must still be marked busy after the flush.
         assert 7 in pool._busy_workers
 
     @t.skip.if_pypy
     @patch('billiard.pool.Pool._create_worker_process')
     def test_flush_preserves_busy_worker_via_scheduled_for_fallback(self, _create_worker_process):
-        """flush() falls back to _scheduled_for when _write_to is unset.
-
-        A job can be accepted before its body finished writing, leaving
-        _write_to unset while _scheduled_for already identifies the worker the
-        task was dispatched to. The busy fd must still be preserved.
-        """
         pool = asynpool.AsynPool(processes=2, synack=False, threads=False)
         pool._state = asynpool.RUN
         pool.maintain_pool = Mock(name='maintain_pool')
@@ -807,12 +788,6 @@ def test_flush_preserves_busy_worker_via_scheduled_for_fallback(self, _create_wo
     @t.skip.if_pypy
     @patch('billiard.pool.Pool._create_worker_process')
     def test_flush_releases_busy_worker_for_unaccepted_job(self, _create_worker_process):
-        """flush() must free a worker whose job was not accepted yet.
-
-        Unaccepted jobs are discarded/redelivered by the broker, so the worker
-        they were tentatively scheduled to is no longer busy and its fd must be
-        dropped from _busy_workers.
-        """
         pool = asynpool.AsynPool(processes=2, synack=False, threads=False)
         pool._state = asynpool.RUN
         pool.maintain_pool = Mock(name='maintain_pool')
@@ -837,13 +812,6 @@ def test_flush_releases_busy_worker_for_unaccepted_job(self, _create_worker_proc
     @t.skip.if_pypy
     @patch('billiard.pool.Pool._create_worker_process')
     def test_flush_preserves_accepted_and_releases_unaccepted_together(self, _create_worker_process):
-        """flush() must resolve a mixed _busy_workers set in a single call.
-
-        When one worker runs an accepted job and another only had an
-        unaccepted job tentatively scheduled to it, the same flush() must keep
-        the accepted worker busy while releasing the unaccepted one, rather
-        than treating the set all-or-nothing.
-        """
         pool = asynpool.AsynPool(processes=2, synack=False, threads=False)
         pool._state = asynpool.RUN
         pool.maintain_pool = Mock(name='maintain_pool')
@@ -878,13 +846,6 @@ def test_flush_preserves_accepted_and_releases_unaccepted_together(self, _create
     @t.skip.if_pypy
     @patch('billiard.pool.Pool._create_worker_process')
     def test_flush_releases_busy_worker_for_dead_process(self, _create_worker_process):
-        """flush() must not keep an accepted job's fd if its worker died.
-
-        If the worker executing an accepted job has exited, its inqueue
-        write-fd may be reused by a replacement process. Keeping the fd marked
-        busy would sideline that healthy replacement, so a dead worker's fd is
-        dropped from _busy_workers.
-        """
         pool = asynpool.AsynPool(processes=2, synack=False, threads=False)
         pool._state = asynpool.RUN
         pool.maintain_pool = Mock(name='maintain_pool')
diff --git a/t/unit/worker/test_consumer.py b/t/unit/worker/test_consumer.py
index 1181f0efbfe..e28e2d9b610 100644
--- a/t/unit/worker/test_consumer.py
+++ b/t/unit/worker/test_consumer.py
@@ -390,14 +390,6 @@ def test_on_close_clears_semaphore_and_reqs(self):
             c.on_close()
 
     def test_on_close_purges_orphan_reservations_from_requests_dict(self):
-        """Regression: ``on_close()`` must remove ``state.requests[id]``
-        entries for Requests that were reserved but never accepted (e.g.
-        ETA tasks queued in ``reserved_requests`` at the moment of a
-        connection loss). PR #7771 attempted this but iterated
-        ``reserved_requests`` (Request objects) and tested membership in
-        ``requests`` (a ``dict[str, Request]``), so ``Request in requests``
-        was always False and nothing was ever deleted.
-        """
         from celery.worker import state
         from celery.worker.consumer.consumer import Consumer
 
@@ -414,8 +406,6 @@ def __init__(self, id):
 
         state.reset_state()
         try:
-            # Orphan reservation: present in ``requests`` and
-            # ``reserved_requests`` but never moved to ``active_requests``.
             orphan = FakeRequest('orphan-1')
             state.requests[orphan.id] = orphan
             state.reserved_requests.add(orphan)
@@ -729,23 +719,13 @@ def test_disable_prefetch_respects_autoscale_max_concurrency(self):
             assert consumer.task_consumer.channel.qos.can_consume() is False
 
     def test_disable_prefetch_after_connection_loss_keeps_gate_closed(self):
-        """Regression: ``can_consume`` must still refuse new messages while a
-        task from before a broker reconnect is still running in the pool.
-
-        With ``worker_disable_prefetch`` enabled and concurrency=1, after a
-        channel/connection interruption ``Consumer.on_close()`` clears
-        ``state.reserved_requests``. The disable-prefetch gate then sees an
-        empty set and lets a new message through, even though the pool is
-        still busy with the in-flight task. The gate must remain closed.
-        """
         from celery.worker import state
         from celery.worker.consumer.consumer import Consumer
         from celery.worker.consumer.tasks import Tasks
 
         self.app.conf.worker_disable_prefetch = True
 
-        # Plain class so it supports weakref (Mock() does not, and WeakSet
-        # would silently reject it).
+        # Mock() isn't weakref-able; state's WeakSets would silently drop it.
         class FakeRequest:
             def __init__(self, id):
                 self.id = id
@@ -777,28 +757,19 @@ def __init__(self, id):
         consumer.app.amqp = Mock()
         consumer.app.amqp.TaskConsumer = Mock(return_value=consumer.task_consumer)
 
-        # Install the disable-prefetch gate on the (mocked) channel.
         Tasks(consumer).start(consumer)
         can_consume = consumer.task_consumer.channel.qos.can_consume
 
         state.reset_state()
         try:
-            # One task running in the pool: by the normal lifecycle
-            # (task_reserved + task_accepted) it lives in BOTH sets.
             in_flight = FakeRequest('task-1')
             state.reserved_requests.add(in_flight)
             state.active_requests.add(in_flight)
 
-            # Pre-condition: gate refuses; concurrency is 1 and slot is taken.
             assert can_consume() is False
 
-            # Simulate a broker channel/connection interruption.
             Consumer.on_close(consumer)
 
-            # Pool still has the task running — active_requests is
-            # intentionally not cleared by on_close(). The gate must still
-            # refuse so we don't over-fetch a second message into a
-            # concurrency=1 pool.
             assert in_flight in state.active_requests
             assert can_consume() is False, (
                 "After a connection loss, can_consume() returned True while "

From ef5ba7ea8bf89ab5794320f8957fcf03b3e5fd79 Mon Sep 17 00:00:00 2001
From: rbehal <rahulbehal01@hotmail.com>
Date: Mon, 14 Sep 2026 20:11:39 +0000
Subject: [PATCH 4/7] Match click 8.5's 'No such option' message in
 test_preload_options

Backport of the upstream assertion; click 8.5 changed the message from
"No such option: --ini" to "No such option '--ini'".

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .github/workflows/codeql-analysis.yml | 71 ---------------------------
 t/unit/app/test_preload_cli.py        |  2 +-
 2 files changed, 1 insertion(+), 72 deletions(-)
 delete mode 100644 .github/workflows/codeql-analysis.yml

diff --git a/.github/workflows/codeql-analysis.yml b/.github/workflows/codeql-analysis.yml
deleted file mode 100644
index 3f014e19107..00000000000
--- a/.github/workflows/codeql-analysis.yml
+++ /dev/null
@@ -1,71 +0,0 @@
-# For most projects, this workflow file will not need changing; you simply need
-# to commit it to your repository.
-#
-# You may wish to alter this file to override the set of languages analyzed,
-# or to provide custom queries or build logic.
-#
-# ******** NOTE ********
-# We have attempted to detect the languages in your repository. Please check
-# the `language` matrix defined below to confirm you have the correct set of
-# supported CodeQL languages.
-#
-name: "CodeQL"
-
-on:
-  push:
-    branches: [ main ]
-  pull_request:
-    # The branches below must be a subset of the branches above
-    branches: [ main ]
-  workflow_dispatch:
-
-
-
-jobs:
-  analyze:
-    name: Analyze
-    runs-on: blacksmith-4vcpu-ubuntu-2204
-    permissions:
-      actions: read
-      contents: read
-      security-events: write
-
-    strategy:
-      fail-fast: false
-      matrix:
-        language: [ 'python' ]
-        # CodeQL supports [ 'cpp', 'csharp', 'go', 'java', 'javascript', 'python', 'ruby' ]
-        # Learn more about CodeQL language support at https://git.io/codeql-language-support
-
-    steps:
-    - name: Checkout repository
-      uses: actions/checkout@v6
-
-    # Initializes the CodeQL tools for scanning.
-    - name: Initialize CodeQL
-      uses: github/codeql-action/init@v4
-      with:
-        languages: ${{ matrix.language }}
-        # If you wish to specify custom queries, you can do so here or in a config file.
-        # By default, queries listed here will override any specified in a config file.
-        # Prefix the list here with "+" to use these queries and those in the config file.
-        # queries: ./path/to/local/query, your-org/your-repo/queries@main
-
-    # Autobuild attempts to build any compiled languages  (C/C++, C#, or Java).
-    # If this step fails, then you should remove it and run the build manually (see below)
-    - name: Autobuild
-      uses: github/codeql-action/autobuild@v4
-
-    # ℹ️ Command-line programs to run using the OS shell.
-    # 📚 https://git.io/JvXDl
-
-    # ✏️ If the Autobuild fails above, remove it and uncomment the following three lines
-    #    and modify them (or add more) to build your code if your project
-    #    uses a compiled language
-
-    #- run: |
-    #   make bootstrap
-    #   make release
-
-    - name: Perform CodeQL Analysis
-      uses: github/codeql-action/analyze@v4
diff --git a/t/unit/app/test_preload_cli.py b/t/unit/app/test_preload_cli.py
index cb07b7866cb..889e43e6822 100644
--- a/t/unit/app/test_preload_cli.py
+++ b/t/unit/app/test_preload_cli.py
@@ -38,7 +38,7 @@ def test_preload_options(subcommand_with_params: Tuple[str, ...], isolated_cli_r
         catch_exceptions=False,
     )
 
-    assert "No such option: --ini" in res_without_preload.output
+    assert "No such option" in res_without_preload.output
     assert res_without_preload.exit_code == 2
 
     res_with_preload = isolated_cli_runner.invoke(

From aa8f6bdb080fede1b5cb0205cb2d74b4bf24e49c Mon Sep 17 00:00:00 2001
From: rbehal <rahulbehal01@hotmail.com>
Date: Mon, 14 Sep 2026 20:11:39 +0000
Subject: [PATCH 5/7] Run CI on GitHub-hosted runners

The fork inherited upstream's Blacksmith runner labels, which this org
has no runners for, so every Ubuntu job sat queued until the 24h timeout.

- blacksmith-4vcpu-ubuntu-* -> ubuntu-24.04 / ubuntu-22.04
- useblacksmith/setup-python -> actions/setup-python
- useblacksmith/setup-docker-builder -> docker/setup-buildx-action
- Codecov upload no longer fails the job (no CODECOV_TOKEN on the fork)
- Integration/Smoke fan-out only on workflow_dispatch
- Drop codeql-analysis.yml: the repo uses default CodeQL setup, which
  rejects uploads from an advanced workflow

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .github/workflows/docker.yml            | 20 ++++++++++----------
 .github/workflows/integration-tests.yml |  4 ++--
 .github/workflows/linter.yml            |  2 +-
 .github/workflows/python-package.yml    | 14 ++++++++------
 .github/workflows/smoke-tests.yml       |  6 +++---
 5 files changed, 24 insertions(+), 22 deletions(-)

diff --git a/.github/workflows/docker.yml b/.github/workflows/docker.yml
index 43fde7bb735..9ecc90d008e 100644
--- a/.github/workflows/docker.yml
+++ b/.github/workflows/docker.yml
@@ -26,51 +26,51 @@ on:
 
 jobs:
   docker-build:
-    runs-on: blacksmith-4vcpu-ubuntu-2404
+    runs-on: ubuntu-24.04
     timeout-minutes: 60
     steps:
     - uses: actions/checkout@v6
     - name: Setup Docker Builder
-      uses: useblacksmith/setup-docker-builder@v1
+      uses: docker/setup-buildx-action@v3
     - name: Build Docker container
       run: make docker-build
 
   smoke-tests_dev:
-    runs-on: blacksmith-4vcpu-ubuntu-2404
+    runs-on: ubuntu-24.04
     timeout-minutes: 10
     steps:
     - uses: actions/checkout@v6
     - name: Setup Docker Builder
-      uses: useblacksmith/setup-docker-builder@v1
+      uses: docker/setup-buildx-action@v3
     - name: "Build smoke tests container: dev"
       run: docker build -f t/smoke/workers/docker/dev .
 
   smoke-tests_latest:
-    runs-on: blacksmith-4vcpu-ubuntu-2404
+    runs-on: ubuntu-24.04
     timeout-minutes: 10
     steps:
     - uses: actions/checkout@v6
     - name: Setup Docker Builder
-      uses: useblacksmith/setup-docker-builder@v1
+      uses: docker/setup-buildx-action@v3
     - name: "Build smoke tests container: latest"
       run: docker build -f t/smoke/workers/docker/pypi .
 
   smoke-tests_pypi:
-    runs-on: blacksmith-4vcpu-ubuntu-2404
+    runs-on: ubuntu-24.04
     timeout-minutes: 10
     steps:
     - uses: actions/checkout@v6
     - name: Setup Docker Builder
-      uses: useblacksmith/setup-docker-builder@v1
+      uses: docker/setup-buildx-action@v3
     - name: "Build smoke tests container: pypi"
       run: docker build -f t/smoke/workers/docker/pypi --build-arg CELERY_VERSION="5" .
 
   smoke-tests_legacy:
-    runs-on: blacksmith-4vcpu-ubuntu-2204
+    runs-on: ubuntu-22.04
     timeout-minutes: 10
     steps:
     - uses: actions/checkout@v6
     - name: Setup Docker Builder
-      uses: useblacksmith/setup-docker-builder@v1
+      uses: docker/setup-buildx-action@v3
     - name: "Build smoke tests container: legacy"
       run: docker build -f t/smoke/workers/docker/pypi --build-arg CELERY_VERSION="4" .
diff --git a/.github/workflows/integration-tests.yml b/.github/workflows/integration-tests.yml
index e3f411bc7ea..fe34ebe6b25 100644
--- a/.github/workflows/integration-tests.yml
+++ b/.github/workflows/integration-tests.yml
@@ -21,7 +21,7 @@ on:
 jobs:
   testing-with:
     timeout-minutes: 240
-    runs-on: blacksmith-4vcpu-ubuntu-2404
+    runs-on: ubuntu-24.04
     strategy:
       fail-fast: false
       matrix:
@@ -52,7 +52,7 @@ jobs:
 
       - uses: actions/checkout@v6
       - name: Set up Python ${{ matrix.python-version }}
-        uses: useblacksmith/setup-python@v6
+        uses: actions/setup-python@v6
         with:
           python-version: ${{ matrix.python-version }}
           allow-prereleases: true
diff --git a/.github/workflows/linter.yml b/.github/workflows/linter.yml
index 249592d4fb3..dd222114dc1 100644
--- a/.github/workflows/linter.yml
+++ b/.github/workflows/linter.yml
@@ -7,7 +7,7 @@ permissions:
 
 jobs:
   linter:
-    runs-on: blacksmith-4vcpu-ubuntu-2204
+    runs-on: ubuntu-22.04
     steps:
 
       - name: Checkout branch
diff --git a/.github/workflows/python-package.yml b/.github/workflows/python-package.yml
index 997585a0b9a..e39718153ba 100644
--- a/.github/workflows/python-package.yml
+++ b/.github/workflows/python-package.yml
@@ -34,7 +34,7 @@ jobs:
       fail-fast: false
       matrix:
         python-version: ['3.9', '3.10', '3.11', '3.12', '3.13', '3.14', 'pypy3.11']
-        os: ["blacksmith-4vcpu-ubuntu-2404", "windows-latest"]
+        os: ["ubuntu-24.04", "windows-latest"]
         exclude:
             - python-version: '3.9'
               os: "windows-latest"
@@ -53,12 +53,12 @@ jobs:
 
     steps:
     - name: Install apt packages
-      if: startsWith(matrix.os, 'blacksmith-4vcpu-ubuntu')
+      if: startsWith(matrix.os, 'ubuntu')
       run: |
         sudo apt-get update && sudo apt-get install -f libcurl4-openssl-dev libssl-dev libgnutls28-dev httping expect libmemcached-dev
     - uses: actions/checkout@v6
     - name: Set up Python ${{ matrix.python-version }}
-      uses: useblacksmith/setup-python@v6
+      uses: actions/setup-python@v6
       with:
         python-version: ${{ matrix.python-version }}
         allow-prereleases: true
@@ -77,7 +77,7 @@ jobs:
     - uses: codecov/codecov-action@v5
       with:
         flags: unittests # optional
-        fail_ci_if_error: true # optional (default = false)
+        fail_ci_if_error: false # fork has no CODECOV_TOKEN
         token: ${{ secrets.CODECOV_TOKEN }}
         verbose: true # optional (default = false)
 
@@ -87,9 +87,11 @@ jobs:
       with:
         token: ${{ secrets.CODECOV_TOKEN }}
 
+  # Integration/smoke fan out to ~370 jobs and need pytest-celery's PyPI celery
+  # image, so on the fork they only run on manual dispatch.
   Integration-tests:
     needs: [Unit]
-    if: needs.Unit.result == 'success'
+    if: needs.Unit.result == 'success' && github.event_name == 'workflow_dispatch'
     strategy:
       matrix:
         module: [
@@ -113,7 +115,7 @@ jobs:
 
   Smoke-tests:
     needs: [Unit]
-    if: needs.Unit.result == 'success'
+    if: needs.Unit.result == 'success' && github.event_name == 'workflow_dispatch'
     strategy:
       matrix:
         module: [
diff --git a/.github/workflows/smoke-tests.yml b/.github/workflows/smoke-tests.yml
index d40307c578d..493d3fbe787 100644
--- a/.github/workflows/smoke-tests.yml
+++ b/.github/workflows/smoke-tests.yml
@@ -18,7 +18,7 @@ permissions:
 
 jobs:
   testing-with:
-    runs-on: blacksmith-4vcpu-ubuntu-2404
+    runs-on: ubuntu-24.04
     strategy:
       fail-fast: false
       matrix:
@@ -38,9 +38,9 @@ jobs:
 
       - uses: actions/checkout@v6
       - name: Setup Docker Builder
-        uses: useblacksmith/setup-docker-builder@v1
+        uses: docker/setup-buildx-action@v3
       - name: Set up Python ${{ matrix.python-version }}
-        uses: useblacksmith/setup-python@v6
+        uses: actions/setup-python@v6
         with:
           python-version: ${{ matrix.python-version }}
           allow-prereleases: true

From a5e0a0aaa9d42536f10b97d8d28148ff8c633f1e Mon Sep 17 00:00:00 2001
From: rbehal <rahulbehal01@hotmail.com>
Date: Mon, 14 Sep 2026 20:15:30 +0000
Subject: [PATCH 6/7] Drop Python 3.9 from CI matrices

kombu's current release line requires Python >=3.10, so the 3.9 unit env
can no longer resolve its dependencies. Upstream celery has dropped 3.9
as well; the backend runs 3.10.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .github/workflows/integration-tests.yml | 2 +-
 .github/workflows/python-package.yml    | 4 +---
 .github/workflows/smoke-tests.yml       | 2 +-
 3 files changed, 3 insertions(+), 5 deletions(-)

diff --git a/.github/workflows/integration-tests.yml b/.github/workflows/integration-tests.yml
index fe34ebe6b25..00721b363bb 100644
--- a/.github/workflows/integration-tests.yml
+++ b/.github/workflows/integration-tests.yml
@@ -11,7 +11,7 @@ on:
         description: 'JSON array of Python versions to test'
         required: false
         type: string
-        default: '["3.9", "3.10", "3.11", "3.12", "3.13", "3.14", "3.14t"]'
+        default: '["3.10", "3.11", "3.12", "3.13", "3.14", "3.14t"]'
       tox_environments:
         description: 'JSON array of tox environments to test'
         required: false
diff --git a/.github/workflows/python-package.yml b/.github/workflows/python-package.yml
index e39718153ba..3a064cc887d 100644
--- a/.github/workflows/python-package.yml
+++ b/.github/workflows/python-package.yml
@@ -33,11 +33,9 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        python-version: ['3.9', '3.10', '3.11', '3.12', '3.13', '3.14', 'pypy3.11']
+        python-version: ['3.10', '3.11', '3.12', '3.13', '3.14', 'pypy3.11']
         os: ["ubuntu-24.04", "windows-latest"]
         exclude:
-            - python-version: '3.9'
-              os: "windows-latest"
             - python-version: '3.10'
               os: "windows-latest"
             - python-version: '3.11'
diff --git a/.github/workflows/smoke-tests.yml b/.github/workflows/smoke-tests.yml
index 493d3fbe787..6a516c98cb3 100644
--- a/.github/workflows/smoke-tests.yml
+++ b/.github/workflows/smoke-tests.yml
@@ -11,7 +11,7 @@ on:
         description: 'JSON array of Python versions to test'
         required: false
         type: string
-        default: '["3.9", "3.10", "3.11", "3.12", "3.13", "3.14", "3.14t"]'
+        default: '["3.10", "3.11", "3.12", "3.13", "3.14", "3.14t"]'
 
 permissions:
   contents: read

From 5ea73f1a8300c23bbaf476a14f82658c9ab94245 Mon Sep 17 00:00:00 2001
From: rbehal <rahulbehal01@hotmail.com>
Date: Mon, 14 Sep 2026 20:40:57 +0000
Subject: [PATCH 7/7] Install libev-dev in CI and drop Python 3.9 from the dev
 image

cassandra-driver has no PyPy wheel and needs ev.h to build from source
(upstream's apt step already lists libev-dev). The dev Dockerfile still
installed the dev requirements into a Python 3.9 interpreter, which can
no longer resolve isort==7.0.0.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .github/workflows/integration-tests.yml |  2 +-
 .github/workflows/python-package.yml    |  2 +-
 docker/Dockerfile                       | 19 +++----------------
 3 files changed, 5 insertions(+), 18 deletions(-)

diff --git a/.github/workflows/integration-tests.yml b/.github/workflows/integration-tests.yml
index 00721b363bb..2875162dc5c 100644
--- a/.github/workflows/integration-tests.yml
+++ b/.github/workflows/integration-tests.yml
@@ -48,7 +48,7 @@ jobs:
     steps:
       - name: Install apt packages
         run: |
-          sudo apt-get update && sudo apt-get install -f libcurl4-openssl-dev libssl-dev libgnutls28-dev httping expect libmemcached-dev
+          sudo apt-get update && sudo apt-get install -f libcurl4-openssl-dev libev-dev libssl-dev libgnutls28-dev httping expect libmemcached-dev
 
       - uses: actions/checkout@v6
       - name: Set up Python ${{ matrix.python-version }}
diff --git a/.github/workflows/python-package.yml b/.github/workflows/python-package.yml
index 3a064cc887d..dbc923e6d80 100644
--- a/.github/workflows/python-package.yml
+++ b/.github/workflows/python-package.yml
@@ -53,7 +53,7 @@ jobs:
     - name: Install apt packages
       if: startsWith(matrix.os, 'ubuntu')
       run: |
-        sudo apt-get update && sudo apt-get install -f libcurl4-openssl-dev libssl-dev libgnutls28-dev httping expect libmemcached-dev
+        sudo apt-get update && sudo apt-get install -f libcurl4-openssl-dev libev-dev libssl-dev libgnutls28-dev httping expect libmemcached-dev
     - uses: actions/checkout@v6
     - name: Set up Python ${{ matrix.python-version }}
       uses: actions/setup-python@v6
diff --git a/docker/Dockerfile b/docker/Dockerfile
index cebba9314ac..42cb5042c19 100644
--- a/docker/Dockerfile
+++ b/docker/Dockerfile
@@ -18,6 +18,7 @@ RUN apt-get update && apt-get install -y build-essential \
     lsb-release \
     git \
     libmemcached-dev \
+    libev-dev \
     make \
     liblzma-dev \
     libreadline-dev \
@@ -70,12 +71,11 @@ RUN pyenv install 3.13 && \
     pyenv install 3.12 && \
     pyenv install 3.11 && \
     pyenv install 3.10 && \
-    pyenv install 3.9 && \
     pyenv install pypy3.11
 
 
 # Set global Python versions
-RUN pyenv global 3.13 3.12 3.11 3.10 3.9 pypy3.11
+RUN pyenv global 3.13 3.12 3.11 3.10 pypy3.11
 
 # Install celery
 WORKDIR $HOME
@@ -84,14 +84,13 @@ COPY --chown=1000:1000 docker/entrypoint /entrypoint
 RUN chmod gu+x /entrypoint
 
 # Define the local pyenvs
-RUN pyenv local 3.13 3.12 3.11 3.10 3.9 pypy3.11
+RUN pyenv local 3.13 3.12 3.11 3.10 pypy3.11
 
 RUN --mount=type=cache,target=/home/$CELERY_USER/.cache/pip \
     pyenv exec python3.13 -m pip install --upgrade pip setuptools wheel && \
     pyenv exec python3.12 -m pip install --upgrade pip setuptools wheel && \
     pyenv exec python3.11 -m pip install --upgrade pip setuptools wheel && \
     pyenv exec python3.10 -m pip install --upgrade pip setuptools wheel && \
-    pyenv exec python3.9 -m pip install --upgrade pip setuptools wheel && \
     pyenv exec pypy3.11 -m pip install --upgrade pip setuptools wheel
 
 # Install requirements first to leverage Docker layer caching
@@ -140,17 +139,6 @@ RUN --mount=type=cache,target=/home/$CELERY_USER/.cache/pip \
     -r requirements/test.txt \
     --build-constraint requirements/constraints.txt
 
-RUN --mount=type=cache,target=/home/$CELERY_USER/.cache/pip \
-    pyenv exec python3.9 -m pip install -r requirements/default.txt \
-    -r requirements/dev.txt \
-    -r requirements/docs.txt \
-    -r requirements/pkgutils.txt \
-    -r requirements/test-ci-base.txt \
-    -r requirements/test-ci-default.txt \
-    -r requirements/test-integration.txt \
-    -r requirements/test.txt \
-    --build-constraint requirements/constraints.txt
-
 RUN --mount=type=cache,target=/home/$CELERY_USER/.cache/pip \
     pyenv exec pypy3.11 -m pip install -r requirements/default.txt \
     -r requirements/dev.txt \
@@ -170,7 +158,6 @@ RUN --mount=type=cache,target=/home/$CELERY_USER/.cache/pip \
     pyenv exec python3.12 -m pip install --no-deps -e $HOME/celery && \
     pyenv exec python3.11 -m pip install --no-deps -e $HOME/celery && \
     pyenv exec python3.10 -m pip install --no-deps -e $HOME/celery && \
-    pyenv exec python3.9 -m pip install --no-deps -e $HOME/celery && \
     pyenv exec pypy3.11 -m pip install --no-deps -e $HOME/celery
 
 WORKDIR $HOME/celery