celery-celery-10179-10189

"""Async I/O backend support utilities."""

import logging
import socket
import threading
import time
from collections import deque
from queue import Empty
from time import sleep
from weakref import WeakKeyDictionary

from kombu.utils.compat import detect_environment

from celery import states
from celery.exceptions import TimeoutError
from celery.utils.threads import THREAD_TIMEOUT_MAX

E_CELERY_RESTART_REQUIRED = "Celery must be restarted because a shutdown signal was detected."

__all__ = (
    'AsyncBackendMixin', 'BaseResultConsumer', 'Drainer',
    'register_drainer',
)


class EventletAdaptedEvent:
    """
    An adapted eventlet event, designed to match the API of `threading.Event` and
    `gevent.event.Event`.
    """

    def __init__(self):
        import eventlet
        self.evt = eventlet.Event()

    def is_set(self):
        return self.evt.ready()

    def set(self):

…[6101 chars omitted]…
 weak=weak, start_drainer=False)
                for result in results]

    def remove_pending_result(self, result):
        self._remove_pending_result(result.id)
        self.on_result_fulfilled(result)
        return result

    def _remove_pending_result(self, task_id):
        for mapping in self._pending_results:
            mapping.pop(task_id, None)

    def on_result_fulfilled(self, result):

"""Redis result store backend."""
import time
from contextlib import contextmanager
from functools import partial
from ssl import CERT_NONE, CERT_OPTIONAL, CERT_REQUIRED
from urllib.parse import unquote

from kombu.utils import symbol_by_name
from kombu.utils.functional import retry_over_time
from kombu.utils.objects import cached_property
from kombu.utils.url import _parse_url, maybe_sanitize_url
from redis import CredentialProvider

from celery import states
from celery._state import task_join_will_block
from celery.backends.base import _create_chord_error_with_cause
from celery.canvas import maybe_signature
from celery.exceptions import BackendStoreError, ChordError, ImproperlyConfigured
from celery.result import GroupResult, allow_join_result
from celery.utils.functional import _regen, dictfilter
from celery.utils.log import get_logger
from celery.utils.time import humanize_second
…[8630 chars omitted]…
ce = self._get_sentinel_instance(**params)

        master_name = self._transport_options.get("master_name", None)

        credentials = {
            k: params[k] for k in ("username", "password") if k in params
        }

        return sentinel_instance.master_for(
            service_name=master_name,
            redis_class=self._get_client(),
            **credentials,
        ).connection_pool

"""The ``RPC`` result backend for AMQP brokers.

RPC-style result backend, using reply-to and one queue per client.
"""
import time

import kombu
from kombu.common import maybe_declare
from kombu.utils.compat import register_after_fork
from kombu.utils.objects import cached_property

from celery import states

from celery._state import current_task, task_join_will_block

from . import base
from .asynchronous import AsyncBackendMixin, BaseResultConsumer

__all__ = ('BacklogLimitExceeded', 'RPCBackend')

E_NO_CHORD_SUPPORT = """
The "rpc" result backend does not support chords!

Note that a group chained with a task is also upgraded to be a chord,
as this pattern requires synchronization.

Result backends that supports chords: Redis, Database, Memcached, and more.
"""


class BacklogLimitExceeded(Exception):
    """Too much state history to fast-forward."""


def _on_after_fork_cleanup_b
…[8629 chars omitted]…
 expires=self.expires,
        ))

    @property
    def binding(self):
        return self.Queue(
            self.oid, self.exchange, self.oid,
            durable=False,
            auto_delete=True,
            expires=self.expires,
        )

    @cached_property
    def oid(self):
        # cached here is the app thread OID: name of queue we receive results on.
        return self.app.thread_oid

W_REDIS_SSL_CERT_OPTIONAL = """
Setting ssl_cert_reqs=CERT_OPTIONAL when connecting to redis means that \
celery might not validate the identity of the redis broker when connecting. \
This leaves you vulnerable to man in the middle attacks.
"""

W_REDIS_SSL_CERT_NONE = """
Setting ssl_cert_reqs=CERT_NONE when connecting to redis means that celery \
will not validate the identity of the redis broker when connecting. This \
leaves you vulnerable to man in the middle attacks.
"""

E_REDIS_SSL_PARAMS_AND_SCHEME_MISMATCH = """
SSL connection parameters have been provided but the specified URL scheme \
is redis://. A Redis SSL connection URL should use the scheme rediss://.
"""

E_REDIS_SSL_CERT_REQS_MISSING_INVALID = """
A rediss:// URL must have parameter ssl_cert_reqs and this must be set to \
CERT_REQUIRED, CERT_OPTIONAL, or CERT_NONE
"""

E_LOST = 'Connection to Redis lost: Retry (%s/
…[6462 chars omitted]…
            pipe.setex(key, self.expires, value)
            else:
                pipe.set(key, value)
            pipe.publish(key, value)
            pipe.execute()

    def forget(self, task_id):
        super().forget(task_id)
        self.result_consumer.cancel_for(task_id)

    def delete(self, key):
        self.client.delete(key)

    def incr(self, key):
        return self.client.incr(key)

__init__.py
test_arangodb.py
test_asynchronous.py
test_azureblockblob.py
test_base.py
test_cache.py
test_cassandra.py
test_consul.py
test_cosmosdbsql.py
test_couchbase.py
test_couchdb.py
test_database.py

test_dynamodb.py
test_elasticsearch.py
test_filesystem.py
test_gcs.py
test_mongodb.py
test_redis.py
test_rpc.py
test_s3.py
25:class DrainerTests:
40:            consumer.drain_events = Mock(side_effect=self.result_consumer_drain_events)
48:    def sleep(self):
66:    def result_consumer_drain_events(self, timeout=None):
69:        drainer.result_consumer.drain_events.
77:            self.sleep(self.interval * 2)
83:        for _ in self.drainer.drain_events_until(p,
96:        This test makes sure that other greenlets can still operate while drain_events_until is
103:            self.sleep(self.interval * 2)
110:                self.sleep(self.interval / 10)
117:        for _ in self.
…[1296 chars omitted]…
vents(self, timeout=None):
234:        time.sleep(timeout)
245:class test_GeventDrainer(GreenletDrainerTests):
251:    def sleep(self):
252:        from gevent import sleep
253:        return sleep
255:    def result_consumer_drain_events(self, timeout=None):
258:        # `drain_events` of asynchronous backends with pubsub have to sleep
261:        gevent.sleep(timeout/10)
266:        gevent.sleep(0)

import os
import socket
import sys
import threading
import time
from unittest.mock import Mock, patch

import pytest
from vine import promise

from celery.backends.asynchronous import E_CELERY_RESTART_REQUIRED, BaseResultConsumer
from celery.backends.base import Backend

from celery.utils import cached_property

pytest.importorskip('gevent')
pytest.importorskip('eventlet')


@pytest.fixture(autouse=True)
def setup_eventlet():
    # By default eventlet will patch the DNS resolver when imported.
    os.environ.update(EVENTLET_NO_GREENDNS='yes')


class DrainerTests:
    """
    Base test class for the Default / Gevent / Eventlet drainers.
    """

    interval = 0.1  # Check every tenth of a second
    MAX_TIMEOUT = 10  # Specify a max timeout so it doesn't run forever

    def get_drainer(self, environment):
        with patch('celery.backends.asynchronous.detect_environment') as d:
   
…[7549 chars omitted]…
 backends with pubsub have to sleep
        # while waiting events for not more then `interval` timeout,
        # but events may coming sooner
        gevent.sleep(timeout/10)

    def schedule_thread(self, thread):
        import gevent
        g = gevent.spawn(thread)
        gevent.sleep(0)
        return g

    def teardown_thread(self, thread):
        import gevent
        gevent.wait([thread])

import uuid
from unittest.mock import Mock, patch

import pytest

from celery import chord, group
from celery._state import _task_stack
from celery.backends.rpc import RPCBackend


class test_RPCResultConsumer:
    def get_backend(self):

        return RPCBackend(app=self.app)

    def get_consumer(self):
        return self.get_backend().result_consumer

    def test_drain_events_before_start(self):
        consumer = self.get_consumer()
        # drain_events shouldn't crash when called before start
        consumer.drain_events(0.001)


class test_RPCBackend:

    def setup_method(self):
        self.b = RPCBackend(app=self.app)

    def test_oid(self):
        oid = self.b.oid
        oid2 = self.b.oid
        assert uuid.UUID(oid)
        assert oid == oid2
        assert oid == self.app.thread_oid

    def test_oid_threads(self):
        # Verify that two RPC backends executed i
…[2390 chars omitted]…
  with self.app.amqp.producer_pool.acquire() as prod:
                self.b.on_task_call(prod, 'task_id'),
                md.assert_called_with(
                    self.b.binding(prod.channel),
                    retry=True,
                )

    def test_create_exchange(self):
        ex = self.b._create_exchange('name')
        assert isinstance(ex, self.b.Exchange)
        assert ex.name == ''

import itertools
import json
import random
import ssl
from contextlib import contextmanager
from datetime import timedelta
from pickle import dumps, loads
from unittest.mock import ANY, Mock, call, patch

import pytest

try:

    from redis import CredentialProvider, exceptions
except ImportError:
    exceptions = None
    CredentialProvider = None

from celery import signature, states, uuid
from celery.canvas import Signature
from celery.contrib.testing.mocks import ContextMock
from celery.exceptions import BackendStoreError, ChordError, ImproperlyConfigured
from celery.result import AsyncResult, GroupResult
from celery.utils.collections import AttributeDict
from t.unit import conftest


def raise_on_second_call(mock, exc, *retval):
    def on_first_call(*args, **kwargs):
        mock.side_effect = exc
        return mock.return_value

    mock.side_effect = on_first_call
    if retva
…[6613 chars omitted]…
state_change(meta, message)
        parent_method.assert_called_once_with(meta, message)
        cancel_for.assert_not_called()

    def test_drain_events_before_start(self):
        consumer = self.get_consumer()
        # drain_events shouldn't crash when called before start
        consumer.drain_events(0.001)

    def test_consume_from_connection_error(self):
        consumer = self.get_consumer()

5.6.2 /repo/celery/__init__.py
[]
class BaseResultConsumer:
    """Manager responsible for consuming result messages."""

    def __init__(self, backend, app, accept,
                 pending_results, pending_messages):
        self.backend = backend
        self.app = app
        self.accept = accept
        self._pending_results = pending_results
        self._pending_messages = pending_messages

        self.on_message = None
        self.buckets = WeakKeyDictionary()
        self.drainer = drainers[detect_environment()](self)

    def start(self, initial_task_id, **kwargs):
        raise NotImplementedError()

    def stop(self):
        pass

    def drain_events(self, timeout=None):
        raise NotImplementedError()

    def consume_from(self, task_id):
        raise NotImplementedError()

    def cancel_for(self, task_id):
        raise NotImplementedError()

    def _after_fo
…[1926 chars omitted]…
    result._maybe_set_cache(meta)
                buckets = self.buckets
                try:
                    # remove bucket for this result, since it's fulfilled
                    bucket = buckets.pop(result)
                except KeyError:
                    pass
                else:
                    # send to waiter via bucket
                    bucket.append(result)
        sleep(0)

WARNING: Retrying (Retry(total=4, connect=None, read=None, redirect=None, status=None)) after connection broken by 'NameResolutionError("HTTPSConnection(host='pypi.org', port=443): Failed to resolve 'pypi.org' ([Errno -3] Temporary failure in name resolution)")': /simple/celery/
WARNING: Retrying (Retry(total=3, connect=None, read=None, redirect=None, status=None)) after connection broken by 'NameResolutionError("HTTPSConnection(host='pypi.org', port=443): Failed to resolve 'pypi.org' ([Errno -3] Temporary failure in name resolution)")': /simple/celery/
WARNING: Retrying (Retry(total=2, connect=None, read=None, redirect=None, status=None)) after connection broken by 'NameResolutionError("HTTPSConnection(host='pypi.org', port=443): Failed to resolve 'pypi.org' ([Errno -3] Temporary failure in name resolution)")': /simple/celery/
WARNING: Retrying (Retry(total=1, connect=None, read=None
…[253 chars omitted]…
=0, connect=None, read=None, redirect=None, status=None)) after connection broken by 'NameResolutionError("HTTPSConnection(host='pypi.org', port=443): Failed to resolve 'pypi.org' ([Errno -3] Temporary failure in name resolution)")': /simple/celery/
ERROR: Could not find a version that satisfies the requirement celery==5.6.3 (from versions: none)
ERROR: No matching distribution found for celery==5.6.3

* master
celery/backends/redis.py:91:        self._connection_errors = self.backend.connection_errors
celery/backends/redis.py:129:        except self._connection_errors:
celery/backends/redis.py:132:            except self._connection_errors as e:
celery/backends/redis.py:322:        self.connection_errors, self.channel_errors = (
celery/backends/redis.py:428:        if isinstance(exc, self.connection_errors):
celery/backends/redis.py:455:            fun, self.connection_errors, args, {},

<class 'kombu.connection.Connection'> True True
(<class 'amqp.exceptions.ConnectionError'>,)
(<class 'amqp.exceptions.ChannelError'>,)
(<class 'amqp.exceptions.ConnectionError'>, <class 'amqp.exceptions.ChannelError'>)

docs/history/changelog-1.0.rst:1786:* The daemon now tries to reconnect if the connection is lost.
docs/history/whatsnew-5.3.rst:207:Celery will automatically retry reconnecting to the broker after 
docs/history/whatsnew-5.3.rst:209:whether to automatically retry reconnecting to the broker for subsequent
docs/history/whatsnew-5.3.rst:210:reconnects.
docs/userguide/configuration.rst:3419:Upon a connection loss, Celery will attempt to reconnect to the broker automatically,
docs/userguide/configuration.rst:3428:during periods of reconnection.
docs/userguide/configuration.rst:3436:reconnection, set ``worker_enable_prefetch_count_reduction`` to False. Disabling this setting might
docs/userguide/workers.rst:294:Celery will automatically retry reconnecting to the broker after the first
docs/userguide/workers.rst:296:retry reconnecting to the broker for subsequent reconnects.

/root/.cache/pip/http-v2/e/5/7/4/b/e574b82e6a6e3336681eba9677daefe03ce0a7dc2e7a0bf51860a9e1.body
/root/.cache/pip/http-v2/f/4/8/d/5/f48d54f8259d6db30ecd31d2a1f0bc992d6ada4df11686317391422d.body
/root/.cache/pip/http-v2/f/7/c/b/d/f7cbd6a7e518ae15aedfad204421d1b50bc8ad39b01dcdc3147f5232.body
/root/.cache/pip/http-v2/6/1/6/3/f/6163fb662f135ae03eac4717570481bfd6f4367ba953ba94b8da84e9.body
/root/.cache/pip/http-v2/6/1/6/3/f/6163fb662f135ae03eac4717570481bfd6f4367ba953ba94b8da84e9
/root/.cache/pip/http-v2/6/8/6/0/4/686043b70df9a0e6b8d36a3b359ce15f3c77ba4d31ca0ab315b6ef65.body
/root/.cache/pip/http-v2/9/6/e/1/6/96e162394a0c844b4fd44e361464c13a82b06a5c690ebc7e4a7f9cf2
/root/.cache/pip/http-v2/9/6/e/1/6/96e162394a0c844b4fd44e361464c13a82b06a5c690ebc7e4a7f9cf2.body

==== /root/.cache/pip/http-v2/e/5/7/4/b/e574b82e6a6e3336681eba9677daefe03ce0a7dc2e7a0bf51860a9e1.body
bash: line 71: file: command not found
Metadata-Version: 2.4
Name: kombu
Version: 5.7.0
Summary: Messaging library for Python.
Home-page: https://kombu.readthedocs.io
==== /root/.cache/pip/http-v2/f/4/8/d/5/f48d54f8259d6db30ecd31d2a1f0bc992d6ada4df11686317391422d.body
bash: line 71: file: command not found
Metadata-Version: 2.4
Name: amqp
Version: 5.4.2

Summary: Low-level AMQP client for Python (fork of amqplib).
Home-page: http://github.com/celery/py-amqp
==== /root/.cache/pip/http-v2/f/7/c/b/d/f7cbd6a7e518ae15aedfad204421d1b50bc8ad39b01dcdc3147f5232.body
bash: line 71: file: command not found
Metadata-Version: 2.4
Name: billiard
Version: 4.3.2
Summary: Python multiprocessing fork with improvements and bugfixes
Home-page: https://github.com/celery/billiard
==== /root/.cache/pip/http-
…[612 chars omitted]…
tp-v2/9/6/e/1/6/96e162394a0c844b4fd44e361464c13a82b06a5c690ebc7e4a7f9cf2
bash: line 71: file: command not found
cc=4,
response
body
headers
Connection
==== /root/.cache/pip/http-v2/9/6/e/1/6/96e162394a0c844b4fd44e361464c13a82b06a5c690ebc7e4a7f9cf2.body
bash: line 71: file: command not found
Metadata-Version: 2.4
Name: pytest-celery
Version: 1.3.0
Summary: Pytest plugin for Celery
License: BSD-3-Clause

count: 887
size: 3380
in-pack: 0
packs: 0
size-pack: 0
prune-packable: 0
garbage: 0
size-garbage: 0

t/unit/backends/test_rpc.py:11:class test_RPCResultConsumer:
t/unit/backends/test_asynchronous.py:11:from celery.backends.asynchronous import E_CELERY_RESTART_REQUIRED, BaseResultConsumer
t/unit/backends/test_asynchronous.py:37:            consumer = BaseResultConsumer(backend, self.app, backend.accept,
t/unit/backends/test_redis.py:197:class test_RedisResultConsumer:
t/unit/backends/test_redis.py:211:    @patch('celery.backends.asynchronous.BaseResultConsumer.on_after_fork')
t/unit/backends/test_redis.py:236:    @patch('celery.backends.redis.ResultConsumer.cancel_for')
t/unit/backends/test_redis.py:237:    @patch('celery.backends.asynchronous.BaseResultConsumer.on_state_change')
t/unit/backends/test_redis.py:274:    @patch('celery.backends.redis.ResultConsumer.cancel_for')
t/unit/backends/test_redis.py:275:    @patch('celery.backends.asynchronous.BaseResultConsumer.on_state_change')
t/unit/backends/test_rpc.py:11:class test_RPCResultConsumer:

import uuid
from unittest.mock import Mock, patch

import pytest

from celery import chord, group
from celery._state import _task_stack
from celery.backends.rpc import RPCBackend


class test_RPCResultConsumer:
    def get_backend(self):

        return RPCBackend(app=self.app)

    def get_consumer(self):
        return self.get_backend().result_consumer

    def test_drain_events_before_start(self):
        consumer = self.get_consumer()
        # drain_events shouldn't crash when called before start
        consumer.drain_events(0.001)


class test_RPCBackend:

    def setup_method(self):
        self.b = RPCBackend(app=self.app)

    def test_oid(self):
        oid = self.b.oid
        oid2 = self.b.oid
        assert uuid.UUID(oid)
        assert oid == oid2
        assert oid == self.app.thread_oid

    def test_oid_threads(self):
        # Verify that two RPC backends executed i
…[2390 chars omitted]…
  with self.app.amqp.producer_pool.acquire() as prod:
                self.b.on_task_call(prod, 'task_id'),
                md.assert_called_with(
                    self.b.binding(prod.channel),
                    retry=True,
                )

    def test_create_exchange(self):
        ex = self.b._create_exchange('name')
        assert isinstance(ex, self.b.Exchange)
        assert ex.name == ''

0
1
2
3
4
5
6
7
8
9
44M	/root/.cache/pip

class Consumer:
    """Message consumer.

    Arguments:
    ---------
        channel (kombu.Connection, ChannelT): see :attr:`channel`.
        queues (Sequence[kombu.Queue]): see :attr:`queues`.
        no_ack (bool): see :attr:`no_ack`.
        auto_declare (bool): see :attr:`auto_declare`
        callbacks (Sequence[Callable]): see :attr:`callbacks`.
        on_message (Callable): See :attr:`on_message`
        on_decode_error (Callable): see :attr:`on_decode_error`.

        prefetch_count (int): see :attr:`prefetch_count`.
    """

    ContentDisallowed = ContentDisallowed

    #: The connection/channel to use for this consumer.
    channel = None

    #: A single :class:`~kombu.Queue`, or a list of queues to
    #: consume from.
    queues = None

    #: Flag for automatic message acknowledgment.
    #: If enabled the messages are automatically acknowledged by the
    #: broker
…[3699 chars omitted]…
l(channel)
        # modify dict size while iterating over it is not allowed
        for qname, queue in list(self._queues.items()):
            # name may have changed after declare
            self._queues.pop(qname, None)
            queue = self._queues[queue.name] = queue(self.channel)
            queue.revive(channel)

        if self.auto_declare:
            self.declare()

        if self.pre

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
ModuleNotFoundError: No module named 'pkg_resources'

docs/history/changelog-1.0.rst:1786:* The daemon now tries to reconnect if the connection is lost.
docs/history/whatsnew-5.3.rst:207:Celery will automatically retry reconnecting to the broker after 
docs/history/whatsnew-5.3.rst:209:whether to automatically retry reconnecting to the broker for subsequent
docs/history/whatsnew-5.3.rst:210:reconnects.
docs/userguide/configuration.rst:3419:Upon a connection loss, Celery will attempt to reconnect to the broker automatically,
docs/userguide/configuration.rst:3428:during periods of reconnection.
docs/userguide/configuration.rst:3436:reconnection, set ``worker_enable_prefetch_count_reduction`` to False. Disabling this setting might
docs/userguide/workers.rst:294:Celery will automatically retry reconnecting to the broker after the first
docs/userguide/workers.rst:296:retry reconnecting to the broker for subsequent reconnects.
celery/backends/
…[1500 chars omitted]…
9:            amqp_context.reconnect()

celery/bin/amqp.py:301:        amqp_context.reconnect()
celery/bin/amqp.py:307:            amqp_context.reconnect()
celery/events/dumper.py:99:            dumper.say('-> Connection lost, attempting reconnect')
celery/worker/consumer/consumer.py:47:consumer: Connection to broker lost. \
celery/worker/consumer/consumer.py:381:              'Trying to reconnect...')

task_reserved)

__all__ = ('Consumer', 'Evloop', 'dump_body')

CLOSE = bootsteps.CLOSE
TERMINATE = bootsteps.TERMINATE
STOP_CONDITIONS = {CLOSE, TERMINATE}
logger = get_logger(__name__)
debug, info, warn, error, crit = (logger.debug, logger.info, logger.warning,
                                  logger.error, logger.critical)

CONNECTION_RETRY = """\

consumer: Connection to broker lost. \
Trying to re-establish the connection...\
"""

CONNECTION_RETRY_STEP = """\
Trying again {when}... ({retries}/{max_retries})\
"""

CONNECTION_ERROR = """\
celery/backends/redis.py:73:E_LOST = 'Connection to Redis lost: Retry (%s/%s) %s.'
celery/backends/redis.py:462:            E_LOST.strip(),
t/unit/backends/test_database.py:142:                raise DatabaseError("connection lost", "SELECT 1", [], Exception())
t/unit/backends/test_database.py:268:                rai
…[750 chars omitted]…
.00 seconds')
t/unit/backends/test_redis.py:780:        logger.error.assert_called_with(self.E_LOST, 3, 10, 'in 30.00 seconds')
t/unit/backends/test_redis.py:1400:    def get_E_LOST(self):
t/unit/backends/test_redis.py:1401:        from celery.backends.redis import E_LOST
t/unit/backends/test_redis.py:1402:        return E_LOST
t/unit/backends/test_redis.py:1406:        self.E_LOST = self.get_E_LOST()

celery/backends/asynchronous.py | 44 ++++++++++++++++++++++++++++
 celery/backends/redis.py        | 18 ++----------
 celery/backends/rpc.py          | 63 ++++++++++++++++++++++++++++++++++++++++-
 3 files changed, 108 insertions(+), 17 deletions(-)
diff --git a/celery/backends/redis.py b/celery/backends/redis.py
index 4e8c2d9..73462d4 100644
--- a/celery/backends/redis.py
+++ b/celery/backends/redis.py
@@ -1,6 +1,5 @@
 """Redis result store backend."""
 import time
-from contextlib import contextmanager

 from functools import partial
 from ssl import CERT_NONE, CERT_OPTIONAL, CERT_REQUIRED
 from urllib.parse import unquote
@@ -72,11 +71,6 @@ CERT_REQUIRED, CERT_OPTIONAL, or CERT_NONE
 
 E_LOST = 'Connection to Redis lost: Retry (%s/%s) %s.'
 
-E_RETRY_LIMIT_EXCEEDED = """
-Retry limit exceeded while trying to reconnect to the Celery redis result \
-store backend. The Celery applicat
…[472 chars omitted]…
             self._ensure(self._reconnect_pubsub, ())
-            except self._connection_errors as e:
-                logger.critical(E_RETRY_LIMIT_EXCEEDED)
-                raise RuntimeError(E_RETRY_LIMIT_EXCEEDED) from e
+    def _reconnect(self):
+        self._ensure(self._reconnect_pubsub, ())
 
     def _maybe_cancel_ready_task(self, meta):
         if meta['status'] in states.READY_STATES:

diff --git a/celery/backends/asynchronous.py b/celery/backends/asynchronous.py
index a5e0e5d..d3f92fe 100644
--- a/celery/backends/asynchronous.py
+++ b/celery/backends/asynchronous.py
@@ -5,6 +5,7 @@ import socket
 import threading
 import time
 from collections import deque
+from contextlib import contextmanager
 from queue import Empty
 from time import sleep
 from weakref import WeakKeyDictionary

@@ -17,6 +18,11 @@ from celery.utils.threads import THREAD_TIMEOUT_MAX
 
 E_CELERY_RESTART_REQUIRED = "Celery must be restarted because a shutdown signal was detected."
 
+E_RETRY_LIMIT_EXCEEDED = """
+Retry limit exceeded while trying to reconnect to the Celery result backend. \
+The Celery application must be restarted.
+"""
+
 __all__ = (
     'AsyncBackendMixin', 'BaseResultConsumer', 'Drainer',
     'register_drainer',
@@ -79,6 +85,13 @@ class Drainer:
                 yield self.wai
…[5868 chars omitted]…
fo=True)
+
+        self._connection = self.app.connection()
+        self._set_connection_errors()
+        self._consumer = self.Consumer(
+            self._connection.default_channel, queues,
+            callbacks=[self.on_state_change], no_ack=self._no_ack,
+            accept=self.accept)
+        self._consumer.consume()
+
     def stop(self):
         try:
             self._consumer.cancel()

........................................................................ [ 65%]
......................................                                   [100%]
=============================== warnings summary ===============================
t/unit/backends/test_redis.py::test_RedisBackend_chords_simple::test_on_chord_part_return
  /usr/local/lib/python3.12/site-packages/_pytest/fixtures.py:1392: PytestRemovedIn10Warning: Class-scoped fixtures defined as instance methods are deprecated.
  Instance attributes set in the 'simple_header_result' fixture will NOT be visible to test methods,
  as each test gets a new instance while the fixture runs only once per class.
  Use a @classmethod decorator below @pytest.fixture and set attributes on cls instead.
  See https://docs.pytest.org/en/stable/deprecations.html#class-scoped-fixture-as-instance-method
    fixturefunc = resolve_fixture_function(fixturedef, request)

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html

110 passed, 1 warning in 1.48s

Traceback (most recent call last):
  File "<stdin>", line 14, in <module>
TypeError: BaseResultConsumer.__init__() missing 5 required positional arguments: 'backend', 'app', 'accept', 'pending_results', and 'pending_messages'

Traceback (most recent call last):
  File "<stdin>", line 13, in <module>
TypeError: BaseResultConsumer.__init__() missing 5 required positional arguments: 'backend', 'app', 'accept', 'pending_results', and 'pending_messages'

CRITICAL:root:
Retry limit exceeded while trying to reconnect to the Celery result backend. The Celery application must be restarted.


(command exceeded 600s)

green calls 4 warnings 3 sleeps 3 exc None

initial errors ()
after start errors (<class 'amqp.exceptions.ConnectionError'>, <class 'amqp.exceptions.ChannelError'>) noack True
queues ['1d4d5d33-16c4-3b0e-9d72-86aebf7968c7']
drain result None new conn? True consume? True
warning calls 1
['Connection to the broker lost: Reconnecting...']
new queues ['1d4d5d33-16c4-3b0e-9d72-86aebf7968c7']
new errors (<class 'amqp.exceptions.ConnectionError'>, <class 'amqp.exceptions.ChannelError'>)
unexpected 
CRITICAL:root:
Retry limit exceeded while trying to reconnect to the Celery result backend. The Celery application must be restarted.

runtime! True <class 'amqp.exceptions.ConnectionError'>
WARNING:celery.backends.rpc:Connection to the broker lost: Reconnecting...
NoneType: None
WARNING:celery.backends.rpc:Error cancelling the old result consumer
Traceback (most recent call last):
  File "/repo/celery/backends/rpc.py", line 103, in _reconn
…[1013 chars omitted]…
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/unittest/mock.py", line 1198, in _execute_mock_call
    raise effect
OSError: close
after errors cancel/close new conn True
new consumer <Consumer: [<Queue 1d4d5d33-16c4-3b0e-9d72-86aebf7968c7 -> <Exchange ''(direct) bound to chan:1> -> 1d4d5d33-16c4-3b0e-9d72-86aebf7968c7 bound to chan:1>]> queues ['1d4d5d33-16c4-3b0e-9d72-86aebf7968c7']

<class 'urllib.error.URLError'> <urlopen error [Errno -3] Temporary failure in name resolution>

__editable__.celery-5.6.2.pth
__editable___celery_5_6_2_finder.py
celery-5.6.2.dist-info
pytest_celery
pytest_celery-1.3.0.dist-info
celery 5.6.2 .
celery 5.6.2 /usr/local/lib/python3.12/site-packages
pytest-celery 1.3.0 /usr/local/lib/python3.12/site-packages
kombu 5.7.0 /usr/local/lib/python3.12/site-packages

timeout: failed to run command ‘curl’: No such file or directory

........................................................................ [ 56%]
.......................................................                  [100%]
=============================== warnings summary ===============================
t/unit/backends/test_redis.py::test_RedisBackend_chords_simple::test_on_chord_part_return
  /usr/local/lib/python3.12/site-packages/_pytest/fixtures.py:1392: PytestRemovedIn10Warning: Class-scoped fixtures defined as instance methods are deprecated.
  Instance attributes set in the 'simple_header_result' fixture will NOT be visible to test methods,
  as each test gets a new instance while the fixture runs only once per class.
  Use a @classmethod decorator below @pytest.fixture and set attributes on cls instead.
  See https://docs.pytest.org/en/stable/deprecations.html#class-scoped-fixture-as-instance-method
    fixturefunc = resolve_fixture_function(fixturedef, request)

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html

127 passed, 1 warning in 4.66s

total 68
drwxr-xr-x   1 root root 4096 Nov  5 11:47 .
drwxr-xr-x   1 root root 4096 Nov  5 11:47 ..
-rwxr-xr-x   1 root root    0 Nov  5 11:47 .dockerenv
lrwxrwxrwx   1 root root    7 Jul  4 09:05 bin -> usr/bin
drwxr-xr-x   2 root root 4096 Jul  4 09:05 boot
drwxr-xr-x   5 root root  340 Nov  5 11:47 dev
drwxr-xr-x   1 root root 4096 Nov  5 11:49 etc
drwxr-xr-x   2 root root 4096 Jul  4 09:05 home
lrwxrwxrwx   1 root root    7 Jul  4 09:05 lib -> usr/lib
lrwxrwxrwx   1 root root    9 Jul  4 09:05 lib64 -> usr/lib64
drwxr-xr-x   2 root root 4096 Aug  3 00:00 media

drwxr-xr-x   2 root root 4096 Aug  3 00:00 mnt
drwxr-xr-x   2 root root 4096 Aug  3 00:00 opt
dr-xr-xr-x 316 root root    0 Nov  5 11:47 proc
drwxr-xr-x   1 root root 4096 Nov  5 11:57 repo
drwx------   1 root root 4096 Nov  5 11:49 root
drwxr-xr-x   1 root root 4096 Nov  5 11:48 run
lrwxrwxrwx   1 root root    8 Jul  4 09:0
…[143 chars omitted]…
Nov  5 12:10 tmp
drwxr-xr-x   1 root root 4096 Aug  3 00:00 usr
drwxr-xr-x   1 root root 4096 Aug  3 00:00 var
total 12
drwxrwxrwt 1 root root 4096 Nov  5 12:10 .
drwxr-xr-x 1 root root 4096 Nov  5 11:47 ..
drwxr-xr-x 2 root root 4096 Nov  5 11:51 celerydl
-rw-r--r-- 1 root root    0 Nov  5 11:50 install.log
total 8
drwxr-xr-x 2 root root 4096 Aug  3 00:00 .
drwxr-xr-x 1 root root 4096 Nov  5 11:47 ..

/usr/local/bin/python: No module named flake8
rc=0

conn set True errors (<class 'ConnectionError'>, <class 'OSError'>)
consumer call call(<Mock name='mock.default_channel' id='138463129937648'>, [], callbacks=[<bound method BaseResultConsumer.on_state_change of <celery.backends.rpc.ResultConsumer object at 0x7dee75a29220>>], no_ack=True, accept={'application/json'})
warning count 3
['Connection to the broker lost: Reconnecting...', 'Error cancelling the old result consumer', 'Error closing the old result consumer connection']

Hint: make sure your test modules/packages have valid Python names.
Traceback:
/usr/local/lib/python3.12/importlib/__init__.py:90: in import_module
    return _bootstrap._gcd_import(name[level:], package, level)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
t/unit/backends/test_gcs.py:5: in <module>
    from google.cloud.exceptions import NotFound
E   ModuleNotFoundError: No module named 'google'
_______________ ERROR collecting t/unit/backends/test_mongodb.py _______________
ImportError while importing test module '/repo/t/unit/backends/test_mongodb.py'.
Hint: make sure your test modules/packages have valid Python names.
Traceback:

/usr/local/lib/python3.12/importlib/__init__.py:90: in import_module
    return _bootstrap._gcd_import(name[level:], package, level)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
t/unit/backends/test_mongodb.py:6: in <mo
…[766 chars omitted]…
_s3 as mock_aws
E   ModuleNotFoundError: No module named 'moto'
=========================== short test summary info ============================
ERROR t/unit/backends/test_azureblockblob.py
ERROR t/unit/backends/test_gcs.py
ERROR t/unit/backends/test_mongodb.py
ERROR t/unit/backends/test_s3.py
!!!!!!!!!!!!!!!!!!! Interrupted: 4 errors during collection !!!!!!!!!!!!!!!!!!!!
7 skipped, 4 errors in 2.05s

>       raise SerializerNotInstalled(
            'No decoder installed for msgpack. '
            'Please install the msgpack-python library')
E       kombu.exceptions.SerializerNotInstalled: No decoder installed for msgpack. Please install the msgpack-python library

/usr/local/lib/python3.12/site-packages/kombu/serialization.py:396: SerializerNotInstalled
=============================== warnings summary ===============================
t/unit/backends/test_redis.py::test_RedisBackend_chords_simple::test_on_chord_part_return
  /usr/local/lib/python3.12/site-packages/_pytest/fixtures.py:1392: PytestRemovedIn10Warning: Class-scoped fixtures defined as instance methods are deprecated.
  Instance attributes set in the 'simple_header_result' fixture will NOT be visible to test methods,
  as each test gets a new instance while the fixture runs only once per class.
  Use a @classmethod deco

…[214 chars omitted]…
, request)

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
=========================== short test summary info ============================
FAILED t/unit/backends/test_base.py::test_KeyValueStoreBackend::test_store_result_parent_id[yaml]
FAILED t/unit/backends/test_base.py::test_KeyValueStoreBackend::test_store_result_parent_id[msgpack]
2 failed, 248 passed, 1 warning in 6.25s

"""The ``RPC`` result backend for AMQP brokers.

RPC-style result backend, using reply-to and one queue per client.
"""
import time
from contextlib import contextmanager

import kombu
from kombu.common import maybe_declare
from kombu.utils.compat import register_after_fork
from kombu.utils.objects import cached_property

from celery import states
from celery._state import current_task, task_join_will_block
from celery.utils.log import get_logger

from . import base
from .asynchronous import AsyncBackendMixin, BaseResultConsumer

__all__ = ('BacklogLimitExceeded', 'RPCBackend')

logger = get_logger(__name__)

E_NO_CHORD_SUPPORT = """
The "rpc" result backend does not support chords!

Note that a group chained with a task is also upgraded to be a chord,
as this pattern requires synchronization.

Result backends that supports chords: Redis, Database, Memcached, and more.
"""


class Back
…[3270 chars omitted]…
f):
        self._consumer = None
        if self._connection is not None:
            self._connection.collect()
            self._connection = None

    def consume_from(self, task_id):
        if self._consumer is None:
            return self.start(task_id)
        queue = self._create_binding(task_id)
        if not self._consumer.consuming_from(queue):
            self._consumer.add_queue(queue)

handle called reconnect 1
other propagated other
not reconnected 0
CRITICAL:root:
Retry limit exceeded while trying to reconnect to the Celery result backend. The Celery application must be restarted.

retry runtime True <class 'ConnectionError'> again

celery/backends/asynchronous.py | 44 +++++++++++++++++++++++++++
 celery/backends/redis.py        | 18 ++---------
 celery/backends/rpc.py          | 67 ++++++++++++++++++++++++++++++++++++++++-
 3 files changed, 112 insertions(+), 17 deletions(-)
 M celery/backends/asynchronous.py
 M celery/backends/redis.py
 M celery/backends/rpc.py

start conn True consumer True
drain returned None
old cancel 1 old close 1
new conn True new consumer True consume 1
new consumer args call(<Mock name='mock.default_channel' id='129741325240128'>, ['q1', 'q2'], callbacks=[<bound method BaseResultConsumer.on_state_change of <celery.backends.rpc.ResultConsumer object at 0x75ffc04a1af0>>], no_ack=True, accept={'application/json'})
errors (<class 'ConnectionError'>, <class 'OSError'>)
warnings 1

total 24
drwx------ 1 root root 4096 Nov  5 11:49 .
drwxr-xr-x 1 root root 4096 Nov  5 11:47 ..
-rw-r--r-- 1 root root  607 Jul  4 09:05 .bashrc
drwxr-xr-x 3 root root 4096 Nov  5 11:49 .cache
-rw-r--r-- 1 root root  132 Jul  4 09:05 .profile
-rw-r--r-- 1 root root  169 Aug  5 01:04 .wget-hsts
/root/.bashrc
/root/.profile
/root/.wget-hsts

timeout: failed to run command ‘wget’: No such file or directory
rc=0

/usr/bin/openssl
/usr/local/bin/python3
connect fail [Errno 101] Network is unreachable

celery/backends/base.py:622:                        logger.warning(
celery/backends/base.py:1171:            logger.warning('Chord counter incremented too many times for %r',
celery/backends/dynamodb.py:104:                logger.warning(
celery/backends/dynamodb.py:352:            logger.warning((
celery/backends/gcs.py:270:            logger.warning(
celery/backends/redis.py:94:            logger.warning(str(e))
celery/backends/redis.py:298:                logger.warning(W_REDIS_SSL_CERT_OPTIONAL)
celery/backends/redis.py:300:                logger.warning(W_REDIS_SSL_CERT_NONE)
celery/backends/rpc.py:87:        logger.warning(
celery/backends/rpc.py:109:                logger.warning(
celery/backends/rpc.py:116:                logger.warning(
celery/worker/consumer/delayed_delivery.py:99:                logger.warning(

celery/worker/consumer/delayed_delivery.py:136:                
…[185 chars omitted]…
 to bind for delayed delivery")
celery/worker/consumer/delayed_delivery.py:207:        logger.warning(
celery/worker/consumer/tasks.py:58:                logger.warning(
celery/worker/control.py:224:            logger.warning('Failed to mark task %s as revoked in backend: %s', task_id, exc)
celery/worker/control.py:590:    logger.warning(msg)
celery/worker/worker.py:434:            logger.warning(log)

time.sleep(amount)

    def _ensure_retryable(self, func, *args, fallback_exc=None, fallback_msg=None, **kwargs):
        """Helper to execute a function with the backend's retry policy."""
        retries = 0
        while True:
            try:
                return func(*args, **kwargs)
            except Exception as exc:
                if self.always_retry and self.exception_safe_to_retry(exc):
                    if retries < self.max_retries:
                        retries += 1

                        logger.warning(
                            'Failed operation %s. Retrying %s more times.',
                            getattr(func, '__name__', repr(func)), self.max_retries - retries,
                            exc_info=True)
                        try:
                            self.on_backend_retryable_error(exc)
                        except Exception:
      
…[1767 chars omitted]…
ed to setup delayed delivery for %r: %s",
                    maybe_sanitize_url(broker_url), str(e)
                )
                setup_errors.append((broker_url, e))

        if len(setup_errors) == len(broker_urls):
            logger.critical(
                "Failed to setup delayed delivery for all broker URLs. "
                "Native delayed delivery will not be available."
            )

.. _changelog:

================
 Change history
================

This document contains change notes for bugfix & new features
in the main branch & 5.6.x series, please see :ref:`whatsnew-5.6` for
an overview of what's new in Celery 5.6.

.. _version-5.6.2:

5.6.2
=====

:release-date: 2026-01-04
:release-by: Tomer Nosrati

What's Changed
~~~~~~~~~~~~~~

- Fix recursive WorkController instantiation in DjangoWorkerFixup + AttributeError when pool_cls is a string (#10045)
- Bugfix: Revoked tasks now immediately update backend status to REVOKED (#9869)
- Prepare for release: v5.6.2 (#10049)

.. _version-5.6.1:

5.6.1
=====

:release-date: 2025-12-29
:release-by: Tomer Nosrati

What's Changed
~~~~~~~~~~~~~~

- Fix Redis Sentinel ACL authentication support (#10013)
- Fix: Broker heartbeats not sent during graceful shutdown (#9986)
- docs #5410 -- Document confirm_publish broker transport
…[2868 chars omitted]…
in
`#9806 <https://github.com/celery/celery/pull/9806>`_.

ETA Task Memory Limit
---------------------
230:- Fix arithmetic overflow for MSSQL result backend (#9904)
771:- Update grpcio to 1.67.0 (#9365)
1453:- Update grpcio to 1.67.0 (#9365)
2815:    by setting the task's result to a failure in the result backend with the os,
2830:    The attacker would also have to gain access to the result backend.

==== /root/.cache/pip/http-v2/e/5/7/4/b/e574b82e6a6e3336681eba9677daefe03ce0a7dc2e7a0bf51860a9e1.body
Metadata-Version: 2.4
Name: kombu
Version: 5.7.0
Summary: Messaging library for Python.
Home-page: https://kombu.readthedocs.io
Author: Ask Solem
Author-email: auvipy@gmail.com
License: BSD-3-Clause
Project-URL: Source, https://github.com/celery/kombu
Keywords: messaging message amqp rabbitmq redis actor producer consumer
Platform: any

Classifier: Development Status :: 5 - Production/Stable
==== /root/.cache/pip/http-v2/f/4/8/d/5/f48d54f8259d6db30ecd31d2a1f0bc992d6ada4df11686317391422d.body
Metadata-Version: 2.4
Name: amqp
Version: 5.4.2
Summary: Low-level AMQP client for Python (fork of amqplib).
Home-page: http://github.com/celery/py-amqp
Author: Barry Pederson
Author-email: auvipy@gmail.com
Maintainer: Asif Saif Uddin
License: BSD
Keywords: amqp rabbitmq cloudamqp messaging
Platfor
…[1685 chars omitted]…
6/e/1/6/96e162394a0c844b4fd44e361464c13a82b06a5c690ebc7e4a7f9cf2.body
Metadata-Version: 2.4
Name: pytest-celery
Version: 1.3.0
Summary: Pytest plugin for Celery
License: BSD-3-Clause
License-File: LICENSE
Keywords: pytest,celery
Author: Tomer Nosrati
Author-email: tomer.nosrati@gmail.com
Requires-Python: >=3.9,<4.0
Classifier: Development Status :: 5 - Production/Stable
Classifier: Framework :: Celery

celery/backends/asynchronous.py | 44 +++++++++++++++++++++++++++
 celery/backends/redis.py        | 18 ++---------
 celery/backends/rpc.py          | 67 ++++++++++++++++++++++++++++++++++++++++-
 3 files changed, 112 insertions(+), 17 deletions(-)
.......................................................                  [100%]
=============================== warnings summary ===============================
t/unit/backends/test_redis.py::test_RedisBackend_chords_simple::test_on_chord_part_return
  /usr/local/lib/python3.12/site-packages/_pytest/fixtures.py:1392: PytestRemovedIn10Warning: Class-scoped fixtures defined as instance methods are deprecated.
  Instance attributes set in the 'simple_header_result' fixture will NOT be visible to test methods,
  as each test gets a new instance while the fixture runs only once per class.
  Use a @classmethod decorator below @pytest.fixture and set attributes on cls instead.
  See https://docs.pytest.org/en/stable/deprecations.html#class-scoped-fixture-as-instance-method

    fixturefunc = resolve_fixture_function(fixturedef, request)

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
127 passed, 1 warning in 4.72s

no_ack passed []
new args no_ack <Mock name='mock.no_ack' id='133127109257280'>
warnings ['Connection to the broker lost: Reconnecting...', 'Error cancelling the old result consumer', 'Error closing the old result consumer connection']
CRITICAL:root:
Retry limit exceeded while trying to reconnect to the Celery result backend. The Celery application must be restarted.

runtime as expected True <class 'ConnectionError'>

[<Queue caf2ed88-5fe1-30a8-9725-373ed913c4a3 -> <Exchange ''(direct) bound to chan:1> -> caf2ed88-5fe1-30a8-9725-373ed913c4a3 bound to chan:1>] [<Queue caf2ed88-5fe1-30a8-9725-373ed913c4a3 -> <Exchange ''(direct) bound to chan:1> -> caf2ed88-5fe1-30a8-9725-373ed913c4a3 bound to chan:1>]
True
[True]
[True]

770:    @patch('celery.backends.redis.logger')
771:    def test_on_connection_error(self, logger):
775:        logger.error.assert_called_with(
778:        logger.error.assert_called_with(self.E_LOST, 2, 10, 'in 20.00 seconds')
780:        logger.error.assert_called_with(self.E_LOST, 3, 10, 'in 30.00 seconds')
t/unit/backends/test_redis.py:770:    @patch('celery.backends.redis.logger')

diff --git a/celery/backends/rpc.py b/celery/backends/rpc.py
index 927c7f5..b70a2c2 100644
--- a/celery/backends/rpc.py
+++ b/celery/backends/rpc.py
@@ -3,6 +3,7 @@
 RPC-style result backend, using reply-to and one queue per client.
 """
 import time
+from contextlib import contextmanager
 
 import kombu
 from kombu.common import maybe_declare

@@ -11,12 +12,15 @@ from kombu.utils.objects import cached_property
 
 from celery import states
 from celery._state import current_task, task_join_will_block
+from celery.utils.log import get_logger
 
 from . import base
 from .asynchronous import AsyncBackendMixin, BaseResultConsumer
 
 __all__ = ('BacklogLimitExceeded', 'RPCBackend')
 
+logger = get_logger(__name__)
+
 E_NO_CHORD_SUPPORT = """
 The "rpc" result backend does not support chords!
 
@@ -40,13 +44,26 @@ class ResultConsumer(BaseResultConsumer):
 
     _connection = None
     _cons
…[3284 chars omitted]…
fo=True)
+
+        self._connection = self.app.connection()
+        self._set_connection_errors()
+        self._consumer = self.Consumer(
+            self._connection.default_channel, queues,
+            callbacks=[self.on_state_change], no_ack=self._no_ack,
+            accept=self.accept)
+        self._consumer.consume()
+
     def stop(self):
         try:
             self._consumer.cancel()

handle reconnects 1 warnings 1
other propagated other
other reconnect 0 warnings 0
CRITICAL:root:
Retry limit exceeded while trying to reconnect to the Celery result backend. The Celery application must be restarted.

retry runtime True <class 'ConnectionError'>
retry warnings 1
direct conn new True consume 1 warnings 3

drain warning count 1 ['Connection to the broker lost: Reconnecting...']
  Instance attributes set in the 'simple_header_result' fixture will NOT be visible to test methods,
  as each test gets a new instance while the fixture runs only once per class.
  Use a @classmethod decorator below @pytest.fixture and set attributes on cls instead.
  See https://docs.pytest.org/en/stable/deprecations.html#class-scoped-fixture-as-instance-method
    fixturefunc = resolve_fixture_function(fixturedef, request)

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
127 passed, 1 warning in 4.47s

OK

E           environment variable (but please think about this before you do).
E           
E           User information: uid=0 euid=0 gid=0 egid=0

celery/platforms.py:837: SecurityError
=============================== warnings summary ===============================
t/unit/backends/test_redis.py::test_RedisBackend_chords_simple::test_on_chord_part_return
  /usr/local/lib/python3.12/site-packages/_pytest/fixtures.py:1392: PytestRemovedIn10Warning: Class-scoped fixtures defined as instance methods are deprecated.
  Instance attributes set in the 'simple_header_result' fixture will NOT be visible to test methods,
  as each test gets a new instance while the fixture runs only once per class.
  Use a @classmethod decorator below @pytest.fixture and set attributes on cls instead.
  See https://docs.pytest.org/en/stable/deprecations.html#class-scoped-fixture-as-instance-method

    fixturefu
…[145 chars omitted]…
====== short test summary info ============================
FAILED t/unit/backends/test_base.py::test_KeyValueStoreBackend::test_store_result_parent_id[yaml]
FAILED t/unit/backends/test_base.py::test_KeyValueStoreBackend::test_store_result_parent_id[msgpack]
FAILED t/unit/backends/test_cache.py::test_CacheBackend::test_regression_worker_startup_info
3 failed, 335 passed, 7 skipped, 1 warning in 10.35s

celery/backends/asynchronous.py | 44 ++++++++++++++++++++++
 celery/backends/redis.py        | 18 +--------
 celery/backends/rpc.py          | 81 ++++++++++++++++++++++++++++++++++++++++-
 3 files changed, 126 insertions(+), 17 deletions(-)
 M celery/backends/asynchronous.py
 M celery/backends/redis.py
 M celery/backends/rpc.py

True
cause exact True
critical called 1 call('\nRetry limit exceeded while trying to reconnect to the Celery result backend. The Celery application must be restarted.\n')

grep: celery/backends/__pycache__/asynchronous.cpython-312.pyc: binary file matches
celery/backends/asynchronous.py:21:E_RETRY_LIMIT_EXCEEDED = """
celery/backends/asynchronous.py:336:                logging.critical(E_RETRY_LIMIT_EXCEEDED)
celery/backends/asynchronous.py:337:                raise RuntimeError(E_RETRY_LIMIT_EXCEEDED) from exc

M celery/backends/asynchronous.py
 M celery/backends/redis.py
 M celery/backends/rpc.py

diff --git a/celery/backends/asynchronous.py b/celery/backends/asynchronous.py
index a5e0e5d..d3f92fe 100644
--- a/celery/backends/asynchronous.py
+++ b/celery/backends/asynchronous.py
@@ -5,6 +5,7 @@ import socket
 import threading
 import time
 from collections import deque
+from contextlib import contextmanager
 from queue import Empty
 from time import sleep
 from weakref import WeakKeyDictionary

@@ -17,6 +18,11 @@ from celery.utils.threads import THREAD_TIMEOUT_MAX
 
 E_CELERY_RESTART_REQUIRED = "Celery must be restarted because a shutdown signal was detected."
 
+E_RETRY_LIMIT_EXCEEDED = """
+Retry limit exceeded while trying to reconnect to the Celery result backend. \
+The Celery application must be restarted.
+"""
+
 __all__ = (
     'AsyncBackendMixin', 'BaseResultConsumer', 'Drainer',
     'register_drainer',
@@ -79,6 +85,13 @@ class Drainer:
                 yield self.wai
…[8160 chars omitted]…
fo=True)
+
+        self._connection = self.app.connection()
+        self._set_connection_errors()
+        self._consumer = self.Consumer(
+            self._connection.default_channel, queues,
+            callbacks=[self.on_state_change], no_ack=self._no_ack,
+            accept=self.accept)
+        self._consumer.consume()
+
     def stop(self):
         try:
             self._consumer.cancel()

3 2 2 [call(0.2), call(0.2)] True

Use a @classmethod decorator below @pytest.fixture and set attributes on cls instead.
  See https://docs.pytest.org/en/stable/deprecations.html#class-scoped-fixture-as-instance-method
    fixturefunc = resolve_fixture_function(fixturedef, request)

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
127 passed, 1 warning in 4.70s
CLEAN

CRITICAL:root:
Retry limit exceeded while trying to reconnect to the Celery result backend. The Celery application must be restarted.

cause is rebuild True
context original? <class 'ConnectionError'> rebuild

…[1 chars omitted]…
 and no_ack tracking, and made both drainer loops retry on OSError with logging.warning(..., exc_info=True) and time.sleep. Affected backend tests pass (127 passed)."}
submitted.

/repo
CONTRIBUTING.rst
CONTRIBUTORS.txt
Changelog.rst
LICENSE
MANIFEST.in
Makefile
README.rst
SECURITY.md
TODO
bandit.json
celery

celery.egg-info
docker
docs
examples
extra
helm-chart
pyproject.toml
requirements
setup.cfg
setup.py
t
tox.ini

"""Async I/O backend support utilities."""

import logging
import socket
import threading
import time
from collections import deque
from queue import Empty
from time import sleep
from weakref import WeakKeyDictionary

from kombu.utils.compat import detect_environment

from celery import states
from celery.exceptions import TimeoutError
from celery.utils.threads import THREAD_TIMEOUT_MAX

E_CELERY_RESTART_REQUIRED = "Celery must be restarted because a shutdown signal was detected."

__all__ = (
    'AsyncBackendMixin', 'BaseResultConsumer', 'Drainer',
    'register_drainer',
)


class EventletAdaptedEvent:
    """
    An adapted eventlet event, designed to match the API of `threading.Event` and
    `gevent.event.Event`.
    """

    def __init__(self):
        import eventlet
        self.evt = eventlet.Event()

    def is_set(self):
        return self.evt.ready()

    def set(self):

…[8630 chars omitted]…
t_on_error():
celery/backends/rpc.py:38:class ResultConsumer(BaseResultConsumer):
t/unit/backends/test_asynchronous.py:25:class DrainerTests:
t/unit/backends/test_asynchronous.py:145:class GreenletDrainerTests(DrainerTests):
t/unit/backends/test_asynchronous.py:190:class test_EventletDrainer(GreenletDrainerTests):
t/unit/backends/test_asynchronous.py:245:class test_GeventDrainer(GreenletDrainerTests):

def _add_pending_result(self, task_id, result, weak=False):
        concrete, weak_ = self._pending_results
        if task_id not in weak_ and result.id not in concrete:
            (weak_ if weak else concrete)[task_id] = result
            self.result_consumer.consume_from(task_id)

    def add_pending_results(self, results, weak=False):
        self.result_consumer.drainer.start()
        return [self.add_pending_result(result, weak=weak, start_drainer=False)
                for result in results]

    def remove_pending_result(self, result):

        self._remove_pending_result(result.id)
        self.on_result_fulfilled(result)
        return result

    def _remove_pending_result(self, task_id):
        for mapping in self._pending_results:
            mapping.pop(task_id, None)

    def on_result_fulfilled(self, result):
        self.result_consumer.cancel_for(result.id)

 
…[8629 chars omitted]…
     """
        # Backends didn't always receive the `request`, so we must still
        # support old code that relies on current_task.
        try:
            request = request or current_task.request
        except AttributeError:
            raise RuntimeError(
                f'RPC backend missing task request for {task_id!r}')
        return request.reply_to, request.correlation_id or task_id

"""Redis result store backend."""
import time
from contextlib import contextmanager
from functools import partial
from ssl import CERT_NONE, CERT_OPTIONAL, CERT_REQUIRED
from urllib.parse import unquote

from kombu.utils import symbol_by_name
from kombu.utils.functional import retry_over_time
from kombu.utils.objects import cached_property
from kombu.utils.url import _parse_url, maybe_sanitize_url
from redis import CredentialProvider

from celery import states
from celery._state import task_join_will_block
from celery.backends.base import _create_chord_error_with_cause
from celery.canvas import maybe_signature
from celery.exceptions import BackendStoreError, ChordError, ImproperlyConfigured
from celery.result import GroupResult, allow_join_result
from celery.utils.functional import _regen, dictfilter
from celery.utils.log import get_logger
from celery.utils.time import humanize_second
…[8629 chars omitted]…
umer.buckets[result] = bucket

    def iter_native(self, result, no_ack=True, **kwargs):
        self._ensure_not_eager()

        results = result.results
        if not results:
            raise StopIteration()

        # we tell the result consumer to put consumed results
        # into these buckets.
        bucket = deque()
        for node in results:
            if not hasattr(node, '_cache'):

CERT_REQUIRED, CERT_OPTIONAL, or CERT_NONE
"""

E_LOST = 'Connection to Redis lost: Retry (%s/%s) %s.'

E_RETRY_LIMIT_EXCEEDED = """
Retry limit exceeded while trying to reconnect to the Celery redis result \
store backend. The Celery application must be restarted.
"""

logger = get_logger(__name__)

class ResultConsumer(BaseResultConsumer):
    _pubsub = None

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._get_key_for_task = self.backend.get_key_for_task
        self._decode_result = self.backend.decode_result
        self._ensure = self.backend.ensure
        self._connection_errors = self.backend.connection_errors
        self.subscribed_to = set()

    def on_after_fork(self):
        try:
            self.backend.client.connection_pool.reset()
            if self._pubsub is not None:
                self._pubsub.close()
        e
…[8404 chars omitted]…
.delivery_mode,
        )
        self.serializer = serializer or conf.result_serializer
        self.auto_delete = auto_delete
        self.result_consumer = self.ResultConsumer(
            self, self.app, self.accept,
            self._pending_results, self._pending_messages,
        )
        if register_after_fork is not None:
            register_after_fork(self, _on_after_fork_cleanup_backend)

t/unit/backends/__init__.py
t/unit/backends/test_arangodb.py
t/unit/backends/test_asynchronous.py
t/unit/backends/test_azureblockblob.py
t/unit/backends/test_base.py
t/unit/backends/test_cache.py
t/unit/backends/test_cassandra.py
t/unit/backends/test_consul.py
t/unit/backends/test_cosmosdbsql.py
t/unit/backends/test_couchbase.py
t/unit/backends/test_couchdb.py
t/unit/backends/test_database.py

t/unit/backends/test_dynamodb.py
t/unit/backends/test_elasticsearch.py
t/unit/backends/test_filesystem.py
t/unit/backends/test_gcs.py
t/unit/backends/test_mongodb.py
t/unit/backends/test_redis.py
t/unit/backends/test_rpc.py
t/unit/backends/test_s3.py
t/unit/backends/test_asynchronous.py:11:from celery.backends.asynchronous import E_CELERY_RESTART_REQUIRED, BaseResultConsumer
t/unit/backends/test_asynchronous.py:37:            consumer = BaseResultConsumer(backend, self.app, backend.accept,
t/unit
…[1569 chars omitted]…
it/backends/test_redis.py:304:    def test__reconnect_pubsub_with_state_change(self):
t/unit/backends/test_redis.py:309:        consumer._reconnect_pubsub()
t/unit/backends/test_redis.py:314:    def test__reconnect_pubsub_without_state_change(self):
t/unit/backends/test_redis.py:319:        consumer._reconnect_pubsub()
t/unit/backends/test_redis.py:801:            fn, b.connection_errors, (), {}, ANY,

import os
import socket
import sys
import threading
import time
from unittest.mock import Mock, patch

import pytest
from vine import promise

from celery.backends.asynchronous import E_CELERY_RESTART_REQUIRED, BaseResultConsumer
from celery.backends.base import Backend

from celery.utils import cached_property

pytest.importorskip('gevent')
pytest.importorskip('eventlet')


@pytest.fixture(autouse=True)
def setup_eventlet():
    # By default eventlet will patch the DNS resolver when imported.
    os.environ.update(EVENTLET_NO_GREENDNS='yes')


class DrainerTests:
    """
    Base test class for the Default / Gevent / Eventlet drainers.
    """

    interval = 0.1  # Check every tenth of a second
    MAX_TIMEOUT = 10  # Specify a max timeout so it doesn't run forever

    def get_drainer(self, environment):
        with patch('celery.backends.asynchronous.detect_environment') as d:
   
…[8629 chars omitted]…
sk(self, i, group_id="group_id"):
        tid = uuid()
        task = Mock(name=f'task-{tid}')
        task.name = 'foobarbaz'
        self.app.tasks['foobarbaz'] = task
        task.request.chord = signature(task)
        task.request.id = tid
        self.b.set_chord_size(group_id, 10)
        task.request.group = group_id
        task.request.group_index = i
        return task

    @contextmanager

import uuid
from unittest.mock import Mock, patch

import pytest

from celery import chord, group
from celery._state import _task_stack
from celery.backends.rpc import RPCBackend


class test_RPCResultConsumer:
    def get_backend(self):

        return RPCBackend(app=self.app)

    def get_consumer(self):
        return self.get_backend().result_consumer

    def test_drain_events_before_start(self):
        consumer = self.get_consumer()
        # drain_events shouldn't crash when called before start
        consumer.drain_events(0.001)


class test_RPCBackend:

    def setup_method(self):
        self.b = RPCBackend(app=self.app)

    def test_oid(self):
        oid = self.b.oid
        oid2 = self.b.oid
        assert uuid.UUID(oid)
        assert oid == oid2
        assert oid == self.app.thread_oid

    def test_oid_threads(self):
        # Verify that two RPC backends executed i
…[8629 chars omitted]…
od.assert_called_once()
        consumer.backend.client.connection_pool.reset.assert_called_once()

        # Continues on KeyError
        consumer._pubsub = Mock()
        consumer._pubsub.close = Mock(side_effect=KeyError)
        parent_method.reset_mock()
        consumer.backend.client.connection_pool.reset.reset_mock()
        consumer.on_after_fork()
        parent_method.assert_called_once()

3:from contextlib import contextmanager
21:from celery.utils.log import get_logger
75:E_RETRY_LIMIT_EXCEEDED = """
80:logger = get_logger(__name__)
100:            logger.warning(str(e))
125:    @contextmanager
133:                logger.critical(E_RETRY_LIMIT_EXCEEDED)
134:                raise RuntimeError(E_RETRY_LIMIT_EXCEEDED) from e
312:                logger.warning(W_REDIS_SSL_CERT_OPTIONAL)
314:                logger.warning(W_REDIS_SSL_CERT_NONE)
461:        logger.error(
605:                        logger.exception(

619:                logger.exception('Chord %r raised: %r', request.group, exc)
622:                logger.exception('Chord %r raised: %r', request.group, exc)
celery/backends/redis.py:91:        self._connection_errors = self.backend.connection_errors
celery/backends/redis.py:129:        except self._connection_errors:
celery/backends/redis.py:132:            e
…[410 chars omitted]…
tion_errors
celery/backends/redis.py:129:        except self._connection_errors:
celery/backends/redis.py:132:            except self._connection_errors as e:
celery/backends/redis.py:322:        self.connection_errors, self.channel_errors = (
celery/backends/redis.py:428:        if isinstance(exc, self.connection_errors):
celery/backends/redis.py:455:            fun, self.connection_errors, args, {},

if socket_keepalive:
            self.connparams['socket_keepalive'] = socket_keepalive

        # "redis_backend_use_ssl" must be a dict with the keys:
        # 'ssl_cert_reqs', 'ssl_ca_certs', 'ssl_certfile', 'ssl_keyfile'
        # (the same as "broker_use_ssl")
        ssl = _get('redis_backend_use_ssl')
        if ssl:
            self.connparams.update(ssl)
            self.connparams['connection_class'] = self.connection_class_ssl

        if url:

            self.connparams = self._params_from_url(url, self.connparams)

        # If we've received SSL parameters via query string or the
        # redis_backend_use_ssl dict, check ssl_cert_reqs is valid. If set
        # via query string ssl_cert_reqs will be a string so convert it here
        if ('connection_class' in self.connparams and
                issubclass(self.connparams['connection_class'], redis.SSLConnecti
…[8629 chars omitted]…
alse)

    yield

    result.task_join_will_block = prev_res_join_block
    _state.task_join_will_block = prev_state_join_block
    _state._set_task_join_will_block(False)


@pytest.fixture(scope='session', autouse=True)
def record_threads_at_startup(request):
    try:
        request.session._threads_at_startup
    except AttributeError:
        request.session._threads_at_startup = alive_threads()

11:class test_RPCResultConsumer:
16:        return self.get_backend().result_consumer
21:        consumer.drain_events(0.001)

kombu 5.6.2
    def __init__(self, channel, queues=None, no_ack=None, auto_declare=None,
                 callbacks=None, on_decode_error=None, on_message=None,
                 accept=None, prefetch_count=None, tag_prefix=None):
        self.channel = channel
        self.queues = maybe_list(queues or [])
        self.no_ack = self.no_ack if no_ack is None else no_ack
        self.callbacks = (self.callbacks or [] if callbacks is None
                          else callbacks)
        self.on_message = on_message
        self.tag_prefix = tag_prefix
        self._active_tags = {}

        if auto_declare is not None:
            self.auto_declare = auto_declare
        if on_decode_error is not None:
            self.on_decode_error = on_decode_error
        self.accept = prepare_accept_content(accept)
        self.prefetch_count = prefetch_count

        if self.channel:
            s
…[1408 chars omitted]…
umer.
        """
        cancel = self.channel.basic_cancel
        for tag in self._active_tags.values():
            cancel(tag)
        self._active_tags.clear()

    def consuming_from(self, queue):
        """Return :const:`True` if currently consuming from queue'."""
        name = queue
        if isinstance(queue, Queue):
            name = queue.name
        return name in self._active_tags

celery/backends/redis.py:322:        self.connection_errors, self.channel_errors = (
celery/events/cursesmon.py:508:            except conn.connection_errors + conn.channel_errors as exc:
celery/events/dumper.py:98:        except conn.connection_errors + conn.channel_errors:
celery/worker/consumer/consumer.py:342:                recoverable_errors = (self.connection_errors + self.channel_errors)
t/unit/worker/test_worker.py:1153:        w._conninfo.connection_errors = w._conninfo.channel_errors = ()
t/unit/worker/test_worker.py:1168:        w._conninfo.connection_errors = w._conninfo.channel_errors = ()
<class 'kombu.connection.Connection'> (<class 'amqp.exceptions.ConnectionError'>,) (<class 'amqp.exceptions.ChannelError'>,)
{'_initial_params': {'hostname': None, 'userid': None, 'password': None, 'virtual_host': None, 'port': None, 'insist': False, 'ssl': False, 'transport': 'memory'
…[204 chars omitted]…
'insist': False, 'connect_timeout': 4, 'ssl': False, 'transport_cls': 'memory', 'heartbeat': None, 'credential_provider': None, 'alt': [], '_failover_strategy': 'round-robin', 'failover_strategy': <class 'itertools.cycle'>, 'transport_options': {}, 'declared_entities': set(), 'connection_errors': (<class 'amqp.exceptions.ConnectionError'>,), 'channel_errors': (<class 'amqp.exceptions.ChannelError'>,)}

def revive(self, channel):
        """Revive consumer after connection loss."""
        self._active_tags.clear()
        channel = self.channel = maybe_channel(channel)
        # modify dict size while iterating over it is not allowed
        for qname, queue in list(self._queues.items()):
            # name may have changed after declare
            self._queues.pop(qname, None)
            queue = self._queues[queue.name] = queue(self.channel)
            queue.revive(channel)

        if self.auto_declare:

            self.declare()

        if self.prefetch_count is not None:
            self.qos(prefetch_count=self.prefetch_count)

Traceback (most recent call last):
  File "<stdin>", line 4, in <module>
AttributeError: type object 'Consumer' has no attribute '_restore'

3:from contextlib import contextmanager
125:    @contextmanager

class Consumer:
    """Message consumer.

    Arguments:
    ---------
        channel (kombu.Connection, ChannelT): see :attr:`channel`.
        queues (Sequence[kombu.Queue]): see :attr:`queues`.
        no_ack (bool): see :attr:`no_ack`.
        auto_declare (bool): see :attr:`auto_declare`
        callbacks (Sequence[Callable]): see :attr:`callbacks`.
        on_message (Callable): See :attr:`on_message`
        on_decode_error (Callable): see :attr:`on_decode_error`.

        prefetch_count (int): see :attr:`prefetch_count`.
    """

    ContentDisallowed = ContentDisallowed

    #: The connection/channel to use for this consumer.
    channel = None

    #: A single :class:`~kombu.Queue`, or a list of queues to
    #: consume from.
    queues = None

    #: Flag for automatic message acknowledgment.
    #: If enabled the messages are automatically acknowledged by the
    #: broker
…[8629 chars omitted]…
de_error:
                raise
            self.on_decode_error(message, exc)
        else:
            return on_m(message) if on_m else self.receive(decoded, message)

    def __repr__(self):
        return f'<{type(self).__name__}: {self.queues}>'

    @property
    def connection(self):
        try:
            return self.channel.connection.client
        except AttributeError:
            pass

celery/backends/base.py:40:logger = get_logger(__name__)
celery/backends/cassandra.py:21:logger = get_logger(__name__)
celery/backends/consul.py:18:logger = get_logger(__name__)
celery/backends/dynamodb.py:26:logger = get_logger(__name__)
celery/backends/gcs.py:40:logger = get_logger(__name__)
celery/backends/redis.py:80:logger = get_logger(__name__)
celery/backends/redis.py:73:E_LOST = 'Connection to Redis lost: Retry (%s/%s) %s.'
celery/contrib/testing/manager.py:131:                self.remark(f'join: connection lost: {exc!r}', '!')
celery/events/cursesmon.py:509:                print(f'Connection lost: {exc!r}', file=sys.stderr)
celery/events/dispatcher.py:42:            as soon as the connection is re-established.
celery/events/dumper.py:99:            dumper.say('-> Connection lost, attempting reconnect')
celery/worker/consumer/consumer.py:47:consumer: Connection to broker lost. \

celery/worker/consumer/consumer.py:48:Trying to re-establish the connection...\
celery/worker/consumer/consumer.py:356:                        f"Retrying to {'establish' if is_connection_loss_on_startup else 're-establish'} "

from celery.utils.objects import Bunch
from celery.utils.text import truncate
from celery.utils.time import humanize_seconds, rate
from celery.worker import loops
from celery.worker.state import (active_requests, maybe_shutdown, requests, reserved_requests, successful_requests,
                                 task_reserved)

__all__ = ('Consumer', 'Evloop', 'dump_body')

CLOSE = bootsteps.CLOSE
TERMINATE = bootsteps.TERMINATE
STOP_CONDITIONS = {CLOSE, TERMINATE}

logger = get_logger(__name__)
debug, info, warn, error, crit = (logger.debug, logger.info, logger.warning,
                                  logger.error, logger.critical)

CONNECTION_RETRY = """\
consumer: Connection to broker lost. \
Trying to re-establish the connection...\
"""

CONNECTION_RETRY_STEP = """\
Trying again {when}... ({retries}/{max_retries})\
"""

CONNECTION_ERROR = """\
consumer: Cannot connect to %s: %s.
%s
"""

CONNECTION_FAILOVER = """\
Will retry using next failover.\
"""

UNKNOWN_FORMAT = """\
Received and deleted unknown message.  Wrong destination?!?
celery/worker/consumer/consumer.py:47:consumer: Connection to broker lost. \

def get(self, key):
        return self.client.get(key)

    def mget(self, keys):
        return self.client.mget(keys)

    def ensure(self, fun, args, **policy):
        retry_policy = dict(self.retry_policy, **policy)
        max_retries = retry_policy.get('max_retries')
        return retry_over_time(
            fun, self.connection_errors, args, {},
            partial(self.on_connection_error, max_retries),

            **retry_policy)

    def on_connection_error(self, max_retries, exc, intervals, retries):
        tts = next(intervals)
        logger.error(
            E_LOST.strip(),
            retries, max_retries or 'Inf', humanize_seconds(tts, 'in '))
        return tts

./celery/backends/redis.py:75:E_RETRY_LIMIT_EXCEEDED = """
./celery/backends/redis.py:133:                logger.critical(E_RETRY_LIMIT_EXCEEDED)
./celery/backends/redis.py:134:                raise RuntimeError(E_RETRY_LIMIT_EXCEEDED) from e

<kombu.utils.objects.cached_property object at 0x7324aaa16990> <kombu.utils.objects.cached_property object at 0x7324aaa169c0>

def cancel_by_queue(self, queue):
        """Cancel consumer by queue name."""
        qname = queue.name if isinstance(queue, Queue) else queue
        try:
            tag = self._active_tags.pop(qname)
        except KeyError:
            pass
        else:
            self.channel.basic_cancel(tag)
        finally:
            self._queues.pop(qname, None)

    def cancel(self):
        """End all active queue consumers.

        Note:
        ----
            This does not affect already delivered messages, but it does
            mean the server will not send any more messages for this consumer.
        """
        cancel = self.channel.basic_cancel
        for tag in self._active_tags.values():
            cancel(tag)
        self._active_tags.clear()

    @property
    def queues(self):  # noqa
        return list(self._queues.values())

    def _basic_consume(self, queue, consumer_tag=None,
                       no_ack=no_ack, nowait=True):
        tag = self._active_tags.get(queue.name)
        if tag is None:
            tag = self._add_tag(queue, consumer_tag)
            queue.consume(tag, self._receive_callback,
                          no_ack=no_ack, nowait=nowait)
        return tag

base has reconnect_on_error: False
drainer propagated: OSError broker disconnected calls: 1 promise ready: False

celery/backends/asynchronous.py:20:E_RETRY_LIMIT_EXCEEDED = (
celery/backends/asynchronous.py:89:                    "Drainer: connection error during drain_events, "
celery/backends/asynchronous.py:134:                        "Drainer: connection error during drain_events, "
celery/backends/asynchronous.py:321:    @contextmanager
celery/backends/asynchronous.py:322:    def reconnect_on_error(self):
celery/backends/asynchronous.py:329:                logging.critical(E_RETRY_LIMIT_EXCEEDED)
celery/backends/asynchronous.py:330:                raise RuntimeError(E_RETRY_LIMIT_EXCEEDED) from exc
celery/backends/asynchronous.py:332:    def _reconnect(self):
celery/backends/redis.py:97:    def _reconnect_pubsub(self):
celery/backends/redis.py:119:    def _reconnect(self):
celery/backends/redis.py:147:            with self.reconnect_on_error():
celery/backends/redis.py:163:            with self.reconnect_on_error():

celery/backends/redis.py:170:            with self.reconnect_on_error():

"""The ``RPC`` result backend for AMQP brokers.

RPC-style result backend, using reply-to and one queue per client.
"""
import time
from contextlib import contextmanager

import kombu
from kombu.common import maybe_declare
from kombu.utils.compat import register_after_fork
from kombu.utils.objects import cached_property

from celery import states
from celery._state import current_task, task_join_will_block
from celery.utils.log import get_logger

from . import base
from .asynchronous import AsyncBackendMixin, BaseResultConsumer

__all__ = ('BacklogLimitExceeded', 'RPCBackend')

logger = get_logger(__name__)


E_NO_CHORD_SUPPORT = """
The "rpc" result backend does not support chords!

Note that a group chained with a task is also upgraded to be a chord,
as this pattern requires synchronization.

Result backends that supports chords: Redis, Database, Memcached, and more.
"""


class Bac
…[6833 chars omitted]…
   def _after_fork(self):
        self.buckets.clear()
        self.buckets = WeakKeyDictionary()
        self.on_message = None
        self.on_after_fork()
 celery/backends/asynchronous.py | 33 ++++++++++++++++++++++++
 celery/backends/redis.py        | 18 ++-----------
 celery/backends/rpc.py          | 57 ++++++++++++++++++++++++++++++++++++++---
 3 files changed, 89 insertions(+), 19 deletions(-)

........................................................................ [ 56%]
.......................................................                  [100%]
127 passed, 1 warning in 4.55s

def __call__(
        self: _MaybeChannelBoundType, channel: (Channel | Connection)
    ) -> _MaybeChannelBoundType:
        """`self(channel) -> self.bind(channel)`."""
        return self.bind(channel)

    def revive(self, channel: Channel) -> None:
        """Revive channel after the connection has been re-established.

        Used by :meth:`~kombu.Connection.ensure`.

        """

        if self.is_bound:
            self._channel = channel
            self.when_bound()

base reconnection ok; drainer retried 3 times, warnings: 2 sleeps: 2

RPC reconnect recovered queues/no_ack and swallowed cancel+close errors

diff --git a/celery/backends/asynchronous.py b/celery/backends/asynchronous.py
index a5e0e5d..6b36cf5 100644
--- a/celery/backends/asynchronous.py
+++ b/celery/backends/asynchronous.py
@@ -5,6 +5,7 @@ import socket
 import threading
 import time
 from collections import deque
+from contextlib import contextmanager
 from queue import Empty
 from time import sleep
 from weakref import WeakKeyDictionary

@@ -17,6 +18,11 @@ from celery.utils.threads import THREAD_TIMEOUT_MAX
 
 E_CELERY_RESTART_REQUIRED = "Celery must be restarted because a shutdown signal was detected."
 
+E_RETRY_LIMIT_EXCEEDED = (
+    "Retry limit exceeded while trying to reconnect to the Celery result "
+    "backend. The Celery application must be restarted."
+)
+
 __all__ = (
     'AsyncBackendMixin', 'BaseResultConsumer', 'Drainer',
     'register_drainer',
@@ -79,6 +85,11 @@ class Drainer:
                 yield s
…[6642 chars omitted]…
fo=True)
+
+        self._connection = self.app.connection()
+        self._set_connection_errors()
+        self._consumer = self.Consumer(
+            self._connection.default_channel, queues,
+            callbacks=[self.on_state_change], no_ack=self._no_ack,
+            accept=self.accept)
+        self._consumer.consume()
+
     def stop(self):
         try:
             self._consumer.cancel()

setup.cfg:6:[flake8]
setup.cfg:9:max-line-length = 117
tox.ini:9:    flake8
 M celery/backends/asynchronous.py
 M celery/backends/redis.py
 M celery/backends/rpc.py

bash: line 432: flake8: command not found

Traceback (most recent call last):
  File "/usr/local/lib/python3.12/site-packages/kombu/transport/virtual/base.py", line 997, in drain_events
    get(self._deliver, timeout=timeout)
  File "/usr/local/lib/python3.12/site-packages/kombu/utils/scheduling.py", line 55, in get
    return self.fun(resource, callback, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/kombu/transport/virtual/base.py", line 1035, in _drain_channel
    return channel.drain_events(callback=callback, timeout=timeout)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/kombu/transport/virtual/base.py", line 754, in drain_events
    return self._poll(self.cycle, callback, timeout=timeout)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

  File "/usr/local/lib/python3.12/site-packag
…[1066 chars omitted]…
^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/kombu/connection.py", line 343, in drain_events
    return self.transport.drain_events(self.connection, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/kombu/transport/virtual/base.py", line 1000, in drain_events
    raise socket.timeout()
TimeoutError

CRITICAL:root:Retry limit exceeded while trying to reconnect to the Celery result backend. The Celery application must be restarted.
Retry limit exceeded while trying to reconnect to the Celery result backend. The Celery application must be restarted.

greenlet drainer retried 4 times; warnings/sleeps: 3 3

==================================== ERRORS ====================================
___________ ERROR collecting t/unit/backends/test_azureblockblob.py ____________
ImportError while importing test module '/repo/t/unit/backends/test_azureblockblob.py'.
Hint: make sure your test modules/packages have valid Python names.
Traceback:
/usr/local/lib/python3.12/importlib/__init__.py:90: in import_module
    return _bootstrap._gcd_import(name[level:], package, level)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
t/unit/backends/test_azureblockblob.py:6: in <module>
    from celery.backends import azureblockblob
celery/backends/azureblockblob.py:2: in <module>
    from kombu.transport.azurestoragequeues import Transport as AzureStorageQueuesTransport

/usr/local/lib/python3.12/site-packages/kombu/transport/azurestoragequeues.py:59: in <module>
    from azure.core.exceptions import ResourceExistsError
E   ModuleNotFoundError: No module named 'azure'
=========================== short test summary info ============================
ERROR t/unit/backends/test_azureblockblob.py
!!!!!!!!!!!!!!!!!!!!!!!!!! stopping after 1 failures !!!!!!!!!!!!!!!!!!!!!!!!!!!
1 skipped, 1 error in 0.69s

M celery/backends/asynchronous.py
 M celery/backends/redis.py
 M celery/backends/rpc.py
........................................................................ [ 56%]
.......................................................                  [100%]
127 passed, 1 warning in 4.54s

submitted.