Apply dispatch pub/sub no-op to thread-local backends (v2.6.57)

Celery stores the result backend in a threading.local unless
result_backend_thread_safe is set, so each thread gets its own backend
instance. The 2.6.55/2.6.56 no-op only patched the import thread's
instance; the cleanup/integrity producer threads created fresh
unpatched backends and kept subscribing to result channels (verified
live and reproduced locally with a stack trace through send_task).
disable_dispatch_result_subscription now wraps Celery._get_backend so
every backend instance any thread creates carries the no-op. Validated
end-to-end: the API-triggered cleanup repro went from 2,633 standing
result-channel subscriptions to zero.

Author: ttlequals0

CHANGELOG.MD

@@ -5,6 +5,12 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0).
 
+## [2.6.57] - 2026-06-11
+
+### Fixed
+
+- **The dispatch pub/sub no-op now actually applies to the threads that dispatch - the 2.6.55/2.6.56 fix never took effect in production.** Celery's `Celery._backend` is stored in a `threading.local` unless `result_backend_thread_safe` is set, so every thread gets its OWN backend instance. The no-op was patched onto the backend created on the import thread, while the cleanup/integrity producer threads (spawned per run by the maintenance routes) lazily created fresh, unpatched backends and kept subscribing - verified live: the first 2.6.56 cleanup run still showed 3,000-5,000 standing subscriptions and ~70-second Redis forced disconnects. Root-caused with a local reproduction (the run is clean when driven from the main thread, subscribes when driven from a `threading.Thread`, confirmed via a stack trace through `send_task -> backend.on_task_call`). `disable_dispatch_result_subscription()` now wraps `Celery._get_backend` so every backend instance created by any thread carries the no-op; validated end-to-end in a local stack where the API-triggered cleanup previously accumulated 2,633 result-channel subscriptions and now accumulates zero. The regression test now asserts the no-op holds in a freshly spawned thread. This retroactively explains all three production runs today: 2.6.53 stalled (get() subscriptions plus dispatch subscriptions), 2.6.54 was clean (its per-dispatch `cancel_for` ran on the same thread-local objects that subscribed), and 2.6.56 regressed (dispatch subscription alive in the producer thread with the unsubscribe removed).
+
 ## [2.6.56] - 2026-06-11
 
 ### Fixed

docker-compose.yml

@@ -63,7 +63,7 @@ services:
 
   # PixelProbe application
   pixelprobe:
-    image: ttlequals0/pixelprobe:${PIXELPROBE_VERSION:-2.6.56}
+    image: ttlequals0/pixelprobe:${PIXELPROBE_VERSION:-2.6.57}
     container_name: pixelprobe-app
     environment:
       # Security
@@ -131,7 +131,7 @@ services:
 
   # P1 Celery Worker for distributed task processing
   celery-worker:
-    image: ttlequals0/pixelprobe:${PIXELPROBE_VERSION:-2.6.56}
+    image: ttlequals0/pixelprobe:${PIXELPROBE_VERSION:-2.6.57}
     container_name: pixelprobe-celery-worker
     command: python celery_worker.py
     environment:

pixelprobe/utils/celery_utils.py

@@ -25,20 +25,39 @@ def check_celery_available():
     return celery_enabled
 
 
+def _noop_on_task_call(producer, task_id):
+    return None
+
+
 def disable_dispatch_result_subscription(celery_instance):
     """Stop apply_async subscribing the producer to result pub/sub channels.
 
     send_task calls backend.on_task_call -> SUBSCRIBE per dispatched task.
     Producers read stored result meta (safe_task_get) and never drain that
-    socket, so its replies pile up until Redis force-closes the connection
-    (client-output-buffer-limit, observed every ~3min during cleanup runs).
+    socket, so unread data piles up until Redis force-closes the connection
+    (client-output-buffer-limit, observed every ~70s during cleanup runs).
     AsyncResult.get() still works: it subscribes itself (add_pending_result)
     and reconciles from stored meta. Producer-side GroupResult.join_native()
     would NOT resolve promptly without the dispatch-time subscription - the
     only group joins live worker-side (discovery), where this hook was
     already skipped via task_join_will_block().
+
+    Celery creates a SEPARATE backend instance per thread (Celery._backend
+    routes through threading.local unless result_backend_thread_safe), so
+    patching one instance misses every other thread - the cleanup producer
+    thread kept subscribing after the first version of this fix. Wrapping
+    _get_backend patches every instance any thread creates.
     """
-    celery_instance.backend.on_task_call = lambda producer, task_id: None
+    orig_get_backend = celery_instance._get_backend
+
+    def _get_backend_without_dispatch_subscription():
+        backend = orig_get_backend()
+        backend.on_task_call = _noop_on_task_call
+        return backend
+
+    celery_instance._get_backend = _get_backend_without_dispatch_subscription
+    # Patch the calling thread's already-created instance too
+    celery_instance.backend.on_task_call = _noop_on_task_call
 
 
 def is_db_connection_corruption(exc) -> bool:

pixelprobe/version.py

@@ -4,7 +4,7 @@
 # Default version - this is the single source of truth
 
 
-_DEFAULT_VERSION = '2.6.56'
+_DEFAULT_VERSION = '2.6.57'
 
 
 # Allow override via environment variable for CI/CD, but default to the hardcoded version

tests/unit/test_maintenance_service.py

@@ -305,17 +305,41 @@ class TestDispatchResultSubscriptionDisabled:
     nothing in the producer reads that socket, so Redis kills the connection
     on output-buffer overrun (observed every ~3min during cleanup runs)."""
 
-    def test_on_task_call_is_noop_instance_override(self):
+    @staticmethod
+    def _patched_celery():
         from celery import Celery
         from pixelprobe.utils.celery_utils import disable_dispatch_result_subscription
 
         celery = Celery('t', broker='redis://localhost:6379/0',
                         backend='redis://localhost:6379/0')
         disable_dispatch_result_subscription(celery)
+        return celery
 
+    def test_on_task_call_is_noop_instance_override(self):
+        celery = self._patched_celery()
         assert 'on_task_call' in vars(celery.backend)
         assert celery.backend.on_task_call(MagicMock(), 'task-1') is None
 
+    def test_on_task_call_is_noop_in_new_threads(self):
+        # Celery._backend is thread-local (one backend instance per thread
+        # unless result_backend_thread_safe); a single-instance patch missed
+        # the cleanup producer thread in production (2026-06-11, v2.6.56).
+        import threading
+
+        celery = self._patched_celery()
+        seen = {}
+
+        def probe():
+            backend = celery.backend
+            seen['fresh_instance'] = backend is not celery._backend_cache
+            seen['noop'] = 'on_task_call' in vars(backend)
+
+        t = threading.Thread(target=probe)
+        t.start()
+        t.join()
+        assert seen['fresh_instance'] is True
+        assert seen['noop'] is True
+
 
 class TestEnvInt: