[GMLP-10778] Keep busy pool workers and in-flight reservations across a broker reconnect (5.6.3+gumloop_0.2.1)

.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" .

.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
@@ -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:
@@ -48,11 +48,11 @@ 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 }}
-        uses: useblacksmith/setup-python@v6
+        uses: actions/setup-python@v6
         with:
           python-version: ${{ matrix.python-version }}
           allow-prereleases: true

.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

.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']
-        os: ["blacksmith-4vcpu-ubuntu-2404", "windows-latest"]
+        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'
@@ -53,12 +51,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
+        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: useblacksmith/setup-python@v6
+      uses: actions/setup-python@v6
       with:
         python-version: ${{ matrix.python-version }}
         allow-prereleases: true
@@ -77,7 +75,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 +85,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 +113,7 @@ jobs:
 
   Smoke-tests:
     needs: [Unit]
-    if: needs.Unit.result == 'success'
+    if: needs.Unit.result == 'success' && github.event_name == 'workflow_dispatch'
     strategy:
       matrix:
         module: [

.github/workflows/smoke-tests.yml

@@ -11,14 +11,14 @@ 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
 
 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

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'

celery/concurrency/asynpool.py

@@ -1080,7 +1080,17 @@ def flush(self):
             self.outbound_buffer.clear()
             self._active_writers.clear()
             self._active_writes.clear()
-            self._busy_workers.clear()
+
+            # _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:
+                    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}

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()
 

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

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(