celery-celery-10189-10158

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
92d274d base
On branch master
nothing to commit, working tree clean

celery/backends/redis.py:76:Retry limit exceeded while trying to reconnect to the Celery redis result \
celery/backends/redis.py:91:        self._connection_errors = self.backend.connection_errors
celery/backends/redis.py:103:    def _reconnect_pubsub(self):
celery/backends/redis.py:126:    def reconnect_on_error(self):
celery/backends/redis.py:129:        except self._connection_errors:
celery/backends/redis.py:131:                self._ensure(self._reconnect_pubsub, ())
celery/backends/redis.py:132:            except self._connection_errors as e:
celery/backends/redis.py:161:            with self.reconnect_on_error():
celery/backends/redis.py:177:            with self.reconnect_on_error():
celery/backends/redis.py:184:            with self.reconnect_on_error():
celery/backends/redis.py:322:        self.connection_errors, self.channel_errors = (
celery/backends/redis.py:428:        i

…[705 chars omitted]…
ring reconnect',
celery/backends/rpc.py:107:                    'connection during reconnect',
celery/backends/rpc.py:122:        logger.info('RPC result consumer: reconnected successfully.')
celery/backends/asynchronous.py:84:                # drain_events handles reconnection internally; if an
celery/backends/asynchronous.py:136:                    # are handled inside drain_events via reconnection.

"""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]…
f.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)

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

RPC-style result backend, using reply-to and one queue per client.
"""
import logging
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 . import base
from .asynchronous import AsyncBackendMixin, BaseResultConsumer

__all__ = ('BacklogLimitExceeded', 'RPCBackend')

logger = logging.getLogger(__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 BacklogLimitExceeded(E
…[3455 chars omitted]…
f._create_binding(task_id)
        if not self._consumer.consuming_from(queue):
            self._consumer.add_queue(queue)
            self._consumer.consume()

    def cancel_for(self, task_id):
        if self._consumer:
            self._consumer.cancel_by_queue(self._create_binding(task_id).name)


class RPCBackend(base.Backend, AsyncBackendMixin):
    """Base class for the RPC result backend."""

./celery/backends/redis.py:75:E_RETRY_LIMIT_EXCEEDED = """
./celery/backends/redis.py:91:        self._connection_errors = self.backend.connection_errors
./celery/backends/redis.py:103:    def _reconnect_pubsub(self):
./celery/backends/redis.py:126:    def reconnect_on_error(self):
./celery/backends/redis.py:129:        except self._connection_errors:
./celery/backends/redis.py:131:                self._ensure(self._reconnect_pubsub, ())
./celery/backends/redis.py:132:            except self._connection_errors as e:
./celery/backends/redis.py:133:                logger.critical(E_RETRY_LIMIT_EXCEEDED)
./celery/backends/redis.py:134:                raise RuntimeError(E_RETRY_LIMIT_EXCEEDED) from e
./celery/backends/redis.py:161:            with self.reconnect_on_error():
./celery/backends/redis.py:177:            with self.reconnect_on_error():
./celery/backends/redis.py:184:          

…[1912 chars omitted]…
in_events_no_reconnect_on_other_errors(self):
./t/unit/backends/test_rpc.py:93:    def test_reconnect_handles_close_failures_gracefully(self):
./t/unit/backends/test_rpc.py:110:        # _reconnect should NOT raise even if cancel/close fail
./t/unit/backends/test_rpc.py:111:        consumer._reconnect()
./t/unit/backends/test_rpc.py:116:    def test_drain_events_channel_error_triggers_reconnect(self):

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)

    def test_drain_events_reconnects_on_connection_error(self):
        consumer = self.get_consumer()
        # Simulate a started consumer with a live connection.
        mock_conn = Mock(name='connection')
        mock_conn.connection_errors = (OSError,)
        mock_conn.channel_errors = ()
        mock_conn.drain_events.side_effect = OSError(
  
…[5332 chars omitted]…
ith pytest.raises(NotImplementedError):
            self.b.apply_chord(self.app.GroupResult(), None)

    @pytest.mark.celery(result_backend='rpc')
    def test_chord_raises_error(self):
        with pytest.raises(NotImplementedError):
            chord(self.add.s(i, i) for i in range(10))(self.add.s([2]))

    @pytest.mark.celery(result_backend='rpc')
    def test_chain_with_chord_raises_error(self):

208:        consumer._connection_errors = (ConnectionError,)
784:        self.app.conf.result_backend_transport_options = dict(
801:            fn, b.connection_errors, (), {}, ANY,
1039:        self.app.conf.result_backend_transport_options = dict(
1059:        self.app.conf.result_backend_transport_options = dict(
1114:        self.app.conf.result_backend_transport_options = dict(
1135:        self.app.conf.result_backend_transport_options = dict(
1163:        self.app.conf.result_backend_transport_options = dict(
1174:        self.app.conf.result_backend_transport_options = dict(
1194:        self.app.conf.result_backend_transport_options = dict(
1207:        self.app.conf.result_backend_transport_options = dict(
1232:        self.app.conf.result_backend_transport_options = dict(

1248:        self.app.conf.result_backend_transport_options = dict(
1276:        self.app.conf.result_b
…[5980 chars omitted]…
sumer._reconnect_pubsub()
        consumer.backend.client.mget.assert_called_once()
        consumer._pubsub.subscribe.assert_called_once()
        consumer._pubsub.connection.register_connect_callback.assert_not_called()


class basetest_RedisBackend:
    def get_backend(self):
        from celery.backends.redis import RedisBackend

        class _RedisBackend(RedisBackend):
            redis = redis

45:    _connection = None
55:        self._connection = self.app.connection()
58:            self._connection.default_channel, [initial_queue],
64:    def _handle_connection_errors(self):
65:        """Context manager that catches broker connection/channel errors and reconnects."""
68:        except (self._connection.connection_errors
69:                + self._connection.channel_errors) as exc:
71:                'RPC result consumer: connection lost (%s), '
77:        if self._connection:
78:            with self._handle_connection_errors():
79:                return self._connection.drain_events(timeout=timeout)
84:        """Close the stale connection and rebuild the consumer.

101:        if self._connection is not None:
103:                self._connection.close()
107:                    'connection during reconnect',
110:            self._connection = None
112:        # Establis
…[8628 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

* master
5.6.2
27:__version__ = '5.6.2'
43:VERSION_BANNER = f'{__version__} ({SERIES})'
45:version_info_t = namedtuple('version_info_t', (
49:# bumpversion can only search for {current_version}
50:# so we have to parse the version here.
52:    r'(\d+)\.(\d+)\.(\d+)(.+)?', __version__).groups()
53:VERSION = version_info = version_info_t(
171:    __path__=__path__, __doc__=__doc__, __version__=__version__,
175:    version_info_t=version_info_t,
176:    version_info=version_info,

import logging
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, greenletDrainer

from celery.backends.base import Backend
from celery.utils import cached_property

# ---- helpers ---------------------------------------------------------------


def _make_consumer(app, environment='default'):
    """Create a BaseResultConsumer with a mocked drainer environment."""
    with patch('celery.backends.asynchronous.detect_environment') as det:
        det.return_value = environment
        backend = Backend(app)
        consumer = BaseResultConsumer(
            backend, app, backend.accept,
            pending_results={}, pending_messages={},
        )
    return consumer


# --------------------------------------
…[8630 chars omitted]…
store") as p:
            yield p

    @pytest.mark.parametrize(['results', 'assert_save_called'], [
        # No results in the header at all - won't call `save()`
        (tuple(), False),
        # Simple results in the header - won't call `save()`
        ((AsyncResult("foo"), ), False),
        # Many simple results in the header - won't call `save()`
        ((AsyncResult("foo"), ) * 42, False),

12:from celery.backends.asynchronous import E_CELERY_RESTART_REQUIRED, BaseResultConsumer, greenletDrainer
20:    """Create a BaseResultConsumer with a mocked drainer environment."""
24:        consumer = BaseResultConsumer(
346:            consumer = BaseResultConsumer(backend, self.app, backend.accept,
        assert x.connparams['db'] == 1
        assert x.connparams['port'] == 123
        assert x.connparams['ssl_cert_reqs'] == ssl.CERT_OPTIONAL

        from redis.connection import SSLConnection
        assert x.connparams['connection_class'] is SSLConnection

    @pytest.mark.parametrize("uri", [

        'rediss://:bosco@vandelay.com:123//1?ssl_cert_reqs=CERT_KITTY_CATS',
        'rediss://:bosco@vandelay.com:123//1'
    ])
    def test_backend_ssl_url_invalid(self, uri):
        pytest.importorskip('redis')

        with pytest.raises(ValueError):
            self.Backend(
    
…[2066 chars omitted]…
eption_safe_to_retry(exceptions.ConnectionError("service unavailable"))
        assert b.exception_safe_to_retry(exceptions.TimeoutError("timeout"))

    def test_incr(self):
        self.b.client = Mock(name='client')
        self.b.incr('foo')
        self.b.client.incr.assert_called_with('foo')

    def test_expire(self):
        self.b.client = Mock(name='client')
        self.b.expire('foo', 300)

)
        drainer.start()

        assert drainer._started.is_set()
        assert drainer._shutdown.is_set()

    def test_start_raises_if_already_shut_down(self, app):
        """start() raises if drainer already completed."""
        drainer = self._make_greenlet_drainer(app)
        drainer._shutdown.set()

        with pytest.raises(Exception, match=E_CELERY_RESTART_REQUIRED):

            drainer.start()

    def test_stop_signals_and_waits(self, app):
        """stop() sets _stopped and waits for _shutdown."""
        drainer = self._make_greenlet_drainer(app)
        # Pre-set _shutdown so wait returns immediately.
        drainer._shutdown.set()
        drainer.stop()

        assert drainer._stopped.is_set()


# ---------------------------------------------------------------------------
# 3. Integration tests with real greenlet runtimes (gevent + eventlet)
# ---------
…[6506 chars omitted]…
thread(self.drainer.run)

            with pytest.raises(Exception, match="Test Exception"):
                self.drainer.start()

            self.teardown_thread(thread)

    def test_start_raises_if_drainer_already_stopped(self):
        with patch.object(self.drainer.result_consumer, 'drain_events', side_effect=lambda **_: self.sleep(0)):
            thread = self.schedule_thread(self.drainer.run)

fatal: unable to access 'https://github.com/celery/celery.git/': Could not resolve host: github.com

Refused: `find /` walks the whole container FS. Scope under /repo.

/repo/celery
/root/.cache/pip

/root/.cache/pip/selfcheck/fe300af6f7d708c14827daac3afc81fbb8306b73de8dd6e3f1f8ea3b
/root/.cache/pip/http-v2/1/1/1/c/d/111cd9e42fa70410a5e10ceb5aba5ab2aad2b6f47ec6ce37a2775dc4.body
/root/.cache/pip/http-v2/1/1/1/c/d/111cd9e42fa70410a5e10ceb5aba5ab2aad2b6f47ec6ce37a2775dc4
/root/.cache/pip/http-v2/1/d/f/b/6/1dfb68cc191d605377cf41cbc6bc1ece9e01530936ab5fac88314df4
/root/.cache/pip/http-v2/1/d/f/b/6/1dfb68cc191d605377cf41cbc6bc1ece9e01530936ab5fac88314df4.body
/root/.cache/pip/http-v2/1/f/a/3/0/1fa3067eeb33a97b3994f1b9c14cfd5136e622dda78a1a82da0b1f5a.body
/root/.cache/pip/http-v2/1/f/a/3/0/1fa3067eeb33a97b3994f1b9c14cfd5136e622dda78a1a82da0b1f5a
/root/.cache/pip/http-v2/1/6/2/f/0/162f02b59107f7f8725a208a8de4172a22557991b6f4dc5cc182b640
/root/.cache/pip/http-v2/1/6/2/f/0/162f02b59107f7f8725a208a8de4172a22557991b6f4dc5cc182b640.body
/root/.cache/pip/http-v2/1/7/9/0/3/179035e06c5c0599483ee5
…[580 chars omitted]…
b16e88374207437d18b9d

/root/.cache/pip/http-v2/b/0/c/5/d/b0c5df80997fa8f009c6e76b4fe9ba5d3b7b16e88374207437d18b9d.body
/root/.cache/pip/http-v2/b/0/2/1/8/b02187b079c9cdd7c6026b274bfbf83fd3d25bd63fc7ab60b94a568f
/root/.cache/pip/http-v2/b/0/2/1/8/b02187b079c9cdd7c6026b274bfbf83fd3d25bd63fc7ab60b94a568f.body
/root/.cache/pip/http-v2/b/4/c/3/7/b4c37e6dc671767ebc965d6574972726925d5507071596100b6905b5.body

bash: line 53: file: command not found
bash: line 53: file: command not found
bash: line 53: file: command not found
bash: line 53: file: command not found
bash: line 53: file: command not found
bash: line 53: file: command not found
bash: line 53: file: command not found
bash: line 53: file: command not found
bash: line 53: file: command not found
bash: line 53: file: command not found
bash: line 53: file: command not found
bash: line 53: file: command not found

bash: line 53: file: command not found
bash: line 53: file: command not found
bash: line 53: file: command not found
bash: line 53: file: command not found
bash: line 53: file: command not found
bash: line 53: file: command not found
bash: line 53: file: command not found
bash: line 53: file: command not found
bash: line 53: file: command not found
bash: line 53: file: command not found
bash: line 53: file: command not found

…[2091 chars omitted]…
mand not found
bash: line 53: file: command not found
bash: line 53: file: command not found
bash: line 53: file: command not found
bash: line 53: file: command not found
bash: line 53: file: command not found
bash: line 53: file: command not found
bash: line 53: file: command not found
bash: line 53: file: command not found
bash: line 53: file: command not found
bash: line 53: file: command not found

celery/backends/base.py:949:        This method checks if a global keyprefix is configured in `result_backend_transport_options` using the
celery/backends/base.py:952:        global_keyprefix = self.app.conf.get('result_backend_transport_options', {}).get("global_keyprefix", None)
celery/app/defaults.py:222:        backend_transport_options=Option({}, type='dict'),
949:        This method checks if a global keyprefix is configured in `result_backend_transport_options` using the
952:        global_keyprefix = self.app.conf.get('result_backend_transport_options', {}).get("global_keyprefix", None)

949:        This method checks if a global keyprefix is configured in `result_backend_transport_options` using the
952:        global_keyprefix = self.app.conf.get('result_backend_transport_options', {}).get("global_keyprefix", None)
                frozen_group.revoke()

                # Handle each task in the group individually
                for result in frozen_group.results:
                    try:
                        # Create fake request for error callbacks
                        fake_request = _create_fake_task_request(
                            task_id=result.id,
                            errbacks=group_callback.options.get("link_error", []),
                            task_name=getattr(result, 'task', 'unknown')

                        )

                        # Call error callbacks for this task with original exception
                        try:
          
…[5356 chars omitted]…
   def _encode_prefixes(self):
        self.task_keyprefix = self.key_t(self.task_keyprefix)
        self.group_keyprefix = self.key_t(self.group_keyprefix)
        self.chord_keyprefix = self.key_t(self.chord_keyprefix)

    def get(self, key):
        raise NotImplementedError('Must implement the get method.')

    def mget(self, keys):
        raise NotImplementedError('Does not support get_many')

celery/app/base.py:1063:                conf.broker_transport_options, **transport_options or {}
celery/app/defaults.py:222:        backend_transport_options=Option({}, type='dict'),
celery/backends/base.py:949:        This method checks if a global keyprefix is configured in `result_backend_transport_options` using the
celery/backends/base.py:952:        global_keyprefix = self.app.conf.get('result_backend_transport_options', {}).get("global_keyprefix", None)
celery/backends/redis.py:435:        if "retry_policy" in self._transport_options:
celery/backends/redis.py:437:            retry_policy.update(self._transport_options['retry_policy'])
celery/backends/redis.py:527:        return self._transport_options.get('result_chord_ordered', True)
celery/backends/redis.py:530:    def _transport_options(self):
celery/backends/redis.py:531:        return self.app.conf.get('result_backend_tran
…[5943 chars omitted]…
 __header__ property, but Celery task created from
                        # Task class do not have this property.

                        # That's why we have to check if this property exists
                        # before checking is it partial function.
                        hasattr(errback.type, '__header__') and

                        # workaround to support tasks with bind=True executed as

_, tid, state, retval = decode(tup)
        if state in EXCEPTION_STATES:
            retval = self.exception_to_python(retval)
        if state in PROPAGATE_STATES:
            chord_error = _create_chord_error_with_cause(
                message=f'Dependency {tid} raised {retval!r}', original_exc=retval
            )
            raise chord_error
        return retval

    def set_chord_size(self, group_id, chord_size):
        self.set(self.get_key_for_group(group_id, '.s'), chord_size)

    def apply_chord(self, header_result_args, body, **kwargs):
        # If any of the child results of this chord are complex (ie. group
        # results themselves), we need to save `header_result` to ensure that
        # the expected structure is retained when we finish the chord and pass
        # the results onward to the body in `on_chord_part_return()`. We don't
        # do this i
…[5482 chars omitted]…
re = False

    #: Set to true if the backend is persistent by default.
    persistent = True

    retry_policy = {
        'max_retries': 20,
        'interval_start': 0,
        'interval_step': 1,
        'interval_max': 1,
    }

    def __init__(self, app,
                 serializer=None, max_cached_results=None, accept=None,
                 expires=None, expires_type=None, url=None, **kwargs):

def get_redis_error_classes():
    """Return tuple of redis error classes."""
    from redis import exceptions

    # This exception suddenly changed name between redis-py versions
    if hasattr(exceptions, 'InvalidData'):
        DataError = exceptions.InvalidData
    else:
        DataError = exceptions.DataError
    return error_classes_t(
        (virtual.Transport.connection_errors + (
            InconsistencyError,

            socket.error,
            IOError,
            OSError,
            exceptions.ConnectionError,
            exceptions.BusyLoadingError,
            exceptions.AuthenticationError,
            exceptions.TimeoutError,
            # A stale fd can still fire on_readable for a channel that was
            # already dropped after a broker restart (#2582): the poller
            # guard from #2561 makes the common path a debug log, but any
            # KeyE
…[458 chars omitted]…
or'>, <class 'OSError'>, <class 'redis.exceptions.ConnectionError'>, <class 'redis.exceptions.BusyLoadingError'>, <class 'redis.exceptions.AuthenticationError'>, <class 'redis.exceptions.TimeoutError'>, <class 'KeyError'>), channel_errors=(<class 'amqp.exceptions.ChannelError'>, <class 'redis.exceptions.DataError'>, <class 'redis.exceptions.InvalidResponse'>, <class 'redis.exceptions.ResponseError'>))

==================================== ERRORS ====================================
_________________ ERROR collecting t/unit/backends/test_rpc.py _________________
'celery' not found in `markers` configuration option
=========================== short test summary info ============================
ERROR t/unit/backends/test_rpc.py - Failed: 'celery' not found in `markers` c...
!!!!!!!!!!!!!!!!!!!!!!!!!! stopping after 1 failures !!!!!!!!!!!!!!!!!!!!!!!!!!!
1 error in 0.50s

pyproject.toml:1:[tool.pytest.ini_options]
pyproject.toml:2:addopts = "--strict-markers"
pyproject.toml:6:markers = ["sleepdeprived_patched_module", "masked_modules", "patched_environ", "patched_module", "flaky", "timeout", "amqp"]
pyproject.toml:54:    "*celery/contrib/pytest.py"
tox.ini:39:    smoke: pytest-xdist>=3.5
tox.ini:48:    unit: coverage run --source=celery -m pytest -vv --maxfail=10 --capture=no -v --junitxml=junit.xml -o junit_family=legacy {posargs}
tox.ini:51:    integration: pytest -xsvv t/integration {posargs}
tox.ini:52:    smoke: pytest -xsvv t/smoke --reruns 5 --reruns-delay 10 {posargs}
tox.ini:131:    bash -c 'containers=$(docker ps -aq --filter label=creator=pytest-docker-tools); if [ -n "$containers" ]; then echo "Removed Docker container(s):"; docker rm -f $containers; fi'
tox.ini:132:    bash -c 'networks=$(docker network ls --filter name=pytest- -q); if [ -n "$networks" ]; then echo "Removed Docker network(s):"; docker network rm $networks; fi'
tox.ini:133:    bash -c 'volumes=$(docker volume ls --filter name=pytest- -q); if [ -n "$volumes" ]; then echo "Removed Docker volume(s):"; docker volume rm $volumes; fi'
__init__.py

__pycache__
app
apps
backends
bin
concurrency
conftest.py
contrib
events

[tool.pytest.ini_options]
addopts = "--strict-markers"
testpaths = "t/unit/"
python_classes = "test_*"
xfail_strict=true
markers = ["sleepdeprived_patched_module", "masked_modules", "patched_environ", "patched_module", "flaky", "timeout", "amqp"]

[tool.mypy]
warn_unused_configs = true
strict = false
follow_imports = "skip"
show_error_codes = true

disallow_untyped_defs = true
ignore_missing_imports = true
files = [
    "celery/__main__.py",
    "celery/states.py",
    "celery/signals.py",
    "celery/fixups",
    "celery/concurrency/thread.py",
    "celery/security/certificate.py",
    "celery/utils/text.py",
    "celery/schedules.py",
    "celery/apps/beat.py",
]

[tool.codespell]
ignore-words-list = "assertin"
skip = "./.*,docs/AUTHORS.txt,docs/history/*,docs/spelling_wordlist.txt,Changelog.rst,CONTRIBUTORS.txt,*.key"

[tool.coverage.run]
branch = true
cover_pylib = false
include = 
…[2625 chars omitted]…
),
        ),
        'accept_content': ('json', 'pickle'),

        # Mongo results tests (only executed if installed and running)
        'mongodb_backend_settings': {
            'host': os.environ.get('MONGO_HOST') or 'localhost',
            'port': os.environ.get('MONGO_PORT') or 27017,
            'database': os.environ.get('MONGO_DB') or 'celery_unittests',
            'taskmeta_collection': (

Name: celery
Version: 5.6.2
Summary: Distributed Task Queue.
Home-page: https://docs.celeryq.dev/
Author: Ask Solem
Author-email: auvipy@gmail.com
License: BSD-3-Clause
Location: /usr/local/lib/python3.12/site-packages
Editable project location: /repo
Requires: billiard, click, click-didyoumean, click-plugins, click-repl, kombu, python-dateutil, tzlocal, vine
Required-by: 
2:24:    # add the pytest.mark.celery() marker registration to the pytest.ini [markers] section

3:28:        "markers", "celery(**overrides): override celery configuration for a test case"
10:72:    mark = request.node.get_closest_marker('celery')
43:179:    mark = request.node.get_closest_marker('celery')

'celery' not found in `markers` configuration option
    from celery.backends import azureblockblob
celery/backends/azureblockblob.py:2: in <module>
'celery' not found in `markers` configuration option
'celery' not found in `markers` configuration option
    from celery.security.certificate import Certificate, CertStore, FSCertStore
celery/security/__init__.py:7: in <module>
celery/security/serialization.py:8: in <module>
celery/security/certificate.py:9: in <module>
    from celery.security.key import PrivateKey

..................                                                       [100%]
18 passed in 0.56s

92d274d refs/heads/master@{0}: commit (initial): base
92d274d HEAD@{0}: commit (initial): base

Mock <class 'TypeError'> unsupported operand type(s) for +: 'Mock' and 'Mock'
MagicMock <MagicMock name='mock.connection_errors.__add__()' id='134755458324752'>

grep: celery/backends/__pycache__/rpc.cpython-312.pyc: binary file matches
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:3429:Upon a connection loss, Celery will attempt to reconnect to the broker automatically,
docs/userguide/configuration.rst:3438:during periods of reconnection.
docs/userguide/configuration.rst:3446: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:r
…[954 chars omitted]…
            'attempting to reconnect...', exc,

celery/backends/rpc.py:74:            self._reconnect()
celery/backends/rpc.py:83:    def _reconnect(self):
celery/backends/rpc.py:97:                    'consumer during reconnect',
celery/backends/rpc.py:107:                    'connection during reconnect',
celery/backends/rpc.py:122:        logger.info('RPC result consumer: reconnected successfully.')

celery/backends/redis.py:3:from contextlib import contextmanager
celery/backends/redis.py:125:    @contextmanager
celery/backends/rpc.py:7:from contextlib import contextmanager
celery/backends/rpc.py:63:    @contextmanager
3:import logging
87:                logging.warning(
139:                    logging.warning(
153:                logging.error(f"Failed to set shutdown event: {e}")

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

import logging
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

from kombu.utils.compat import detect_environment

from celery import states
from celery.exceptions import TimeoutError
from celery.utils.log import get_logger
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 \
store backend. The Celery application must be restarted.
"""

logger = get_logger(__name__)

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


class EventletAdaptedEvent:
    """
    An
…[2120 chars omitted]…
                        **kwargs):
        self.on_wait_for_pending(result, timeout=timeout, **kwargs)
        prev_on_m, self.on_message = self.on_message, on_message
        try:
            for _ in self.drain_events_until(
                    result.on_ready, timeout=timeout,
                    on_interval=on_interval):
                yield
                sleep(0)
        except socket.timeout:

"""Redis result store backend."""
import time
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_seconds

from .asynchronous import AsyncBack
…[6527 chars omitted]…
t', None)
            connparams.pop('socket_connect_timeout')
        else:
            connparams['db'] = path

        ssl_param_keys = ['ssl_ca_certs', 'ssl_certfile', 'ssl_keyfile',
                          'ssl_cert_reqs']

        if scheme == 'redis':
            # If connparams or query string contain ssl params, raise error
            if (any(key in connparams for key in ssl_param_keys) or

raise ValueError(E_REDIS_SSL_CERT_REQS_MISSING_INVALID)

            if ssl_cert_reqs == CERT_OPTIONAL:
                logger.warning(W_REDIS_SSL_CERT_OPTIONAL)
            elif ssl_cert_reqs == CERT_NONE:
                logger.warning(W_REDIS_SSL_CERT_NONE)
            self.connparams['ssl_cert_reqs'] = ssl_cert_reqs

        self.url = url

        # Add driver identification for redis-py
        self._add_driver_info()

        self.connection_errors, self.channel_errors = (
            get_redis_error_classes() if get_redis_error_classes
            else ((), ()))
        self.connection_errors = (
            tuple(self.connection_errors)
            + self._get_additional_connection_errors()
        )
        self.result_consumer = self.ResultConsumer(
            self, self.app, self.accept,
            self._pending_results, self._pending_messages,
        )

…[162 chars omitted]…
not additional_errors:
            return ()

        if isinstance(additional_errors, (str, type)):
            additional_errors = (additional_errors,)
        else:
            try:
                additional_errors = tuple(additional_errors)
            except TypeError:
                additional_errors = (additional_errors,)

        connection_errors = []
        for error in additional_errors:

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

RPC-style result backend, using reply-to and one queue per client.
"""
import logging
import sys
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')

logger = logging.getLogger(__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 BacklogLimitExceeded(Exception):
    """Too much 
…[3678 chars omitted]…
k_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)
            self._consumer.consume()

    def cancel_for(self, task_id):
        if self._consumer:
            self._consumer.cancel_by_queue(self._create_binding(task_id).name)

........................................................................ [ 46%]
.......F
=================================== FAILURES ===================================
_____ test_RedisBackend_chords_simple.test_on_chord_part_return__unordered _____

self = <t.unit.backends.test_redis.test_RedisBackend_chords_simple object at 0x7b3c99fdaf30>

    def test_on_chord_part_return__unordered(self):
        self.app.conf.result_backend_transport_options = dict(
            result_chord_ordered=False,
        )
    

        tasks = [self.create_task(i) for i in range(10)]
        random.shuffle(tasks)
    
        for i in range(10):
            self.b.on_chord_part_return(tasks[i].request, states.SUCCESS, i)
>           assert self.b.client.rpush.call_count
E           AssertionError: assert 0
E            +  where 0 = <Mock name='Redis.rpush' id='135500199446336'>.call_count
E            
…[480 chars omitted]…
_chords_simple object at 0x7b3c99fdaf30>.b

t/unit/backends/test_redis.py:1048: AssertionError
=========================== short test summary info ============================
FAILED t/unit/backends/test_redis.py::test_RedisBackend_chords_simple::test_on_chord_part_return__unordered
!!!!!!!!!!!!!!!!!!!!!!!!!! stopping after 1 failures !!!!!!!!!!!!!!!!!!!!!!!!!!!
1 failed, 79 passed, 1 warning in 1.72s

F                                                                        [100%]
=================================== FAILURES ===================================
_____ test_RedisBackend_chords_simple.test_on_chord_part_return__unordered _____

self = <t.unit.backends.test_redis.test_RedisBackend_chords_simple object at 0x75d82f80a4e0>

    def test_on_chord_part_return__unordered(self):
        self.app.conf.result_backend_transport_options = dict(
            result_chord_ordered=False,
        )
    
        tasks = [self.create_task(i) for i in range(10)]

        random.shuffle(tasks)
    
        for i in range(10):
            self.b.on_chord_part_return(tasks[i].request, states.SUCCESS, i)
>           assert self.b.client.rpush.call_count
E           AssertionError: assert 0
E            +  where 0 = <Mock name='Redis.rpush' id='129571369327056'>.call_count
E            +    wher
…[989 chars omitted]…
 staged for commit:
  (use "git add <file>..." to update what will be committed)
  (use "git restore <file>..." to discard changes in working directory)
	modified:   celery/backends/asynchronous.py
	modified:   celery/backends/redis.py
	modified:   celery/backends/rpc.py

no changes added to commit (use "git add" and/or "git commit -a")
Dropped refs/stash@{0} (f56e35e398ab84b0f8833241fc68e7940a32a407)

.                                                                        [100%]
1 passed, 1 warning in 0.55s

........................................................................ [ 46%]
............................................................sssssssss... [ 93%]
.sssssssss                                                               [100%]
136 passed, 18 skipped, 1 warning in 3.06s

base ok
Traceback (most recent call last):
  File "<stdin>", line 96, in <module>
AssertionError: no.such.module.Error

no.such.module.Error warnings 1
<class 'int'> warnings 1
42 warnings 1
(<class 't.unit.backends.test_redis.ConnectionError'>, <class 'int'>) warnings 1
[42, 'no.such.module.Error'] warnings 2
redis invalid ok
redis consumer ok
Traceback (most recent call last):
  File "<stdin>", line 51, in <module>
AssertionError

warning called call('RPC result consumer: connection lost (%s), attempting to reconnect...', None)
conn is new True
errors (<class 'ValueError'>, <class 'KeyError'>)
new attrs (<class 'ValueError'>,) (<class 'KeyError'>,)

........................................................................ [ 74%]
.........................                                                [100%]
97 passed, 1 warning in 1.46s

all redis backend forms ok
base details ok

diff --git a/celery/backends/asynchronous.py b/celery/backends/asynchronous.py
index 0413afe..eb7ed86 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

@@ -13,10 +14,18 @@ from kombu.utils.compat import detect_environment
 
 from celery import states
 from celery.exceptions import TimeoutError
+from celery.utils.log import get_logger
 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 \
+store backend. The Celery application must be restarte
…[7877 chars omitted]…
  self._consumer = self.Consumer(
             self._connection.default_channel,
             old_queues,
@@ -132,6 +143,7 @@ class ResultConsumer(BaseResultConsumer):
         if self._connection is not None:
             self._connection.collect()
             self._connection = None
+        self._update_connection_errors()
 
     def consume_from(self, task_id):
         if self._consumer is None:

For variant 'global', will try loading '/etc/xdg/pip/pip.conf'
For variant 'global', will try loading '/etc/pip.conf'
For variant 'user', will try loading '/root/.pip/pip.conf'
For variant 'user', will try loading '/root/.config/pip/pip.conf'
For variant 'site', will try loading '/usr/local/pip.conf'
# Generated by Docker Engine.
# This file can be edited; Docker Engine will not make further changes once it
# has been modified.

nameserver 192.168.65.7

# Based on host file: '/etc/resolv.conf' (legacy)

# Overrides: []

97:def get_logger(name):
                safe_data = safe_str(data).rstrip('\n')
                if safe_data:
                    self.logger.log(self.loglevel, safe_data)
                    return len(safe_data)
            finally:
                self._thread.recurse_protection = False
        return 0

    def writelines(self, sequence):
        # type: (Sequence[str]) -> None
        """Write list of strings to file.

        The sequence can be any iterable object producing strings.
        This is equivalent to calling :meth:`write` for each string.
        """
        for part in sequence:
            self.write(part)

    def flush(self):
        # This object is not buffered so any :meth:`flush`
        # requests are ignored.
        pass

    def close(self):
        # when the object is closed, no write requests are
        # forwarded to the logging object anymore.
        self.closed = True

    def isatty(self):
        """Here for file support."""
        return False

def _using_logger_parent(parent_logger, logger_):
    if not logger_isa(logger_, parent_logger):
        logger_.parent = parent_logger
    return logger_


def get_logger(name):
    """Get logger by name."""
    l = _get_logger(name)
    if logging.root not in (l, l.parent) and l is not base_logger:
        l = _using_logger_parent(base_logger, l)
    return l

task_logger = get_logger('celery.task')
worker_logger = get_logger('celery.worker')


def get_task_logger(name):
    """Get logger for task module by name."""
    if name in RESERVED_LOGGER_NAMES:
        raise RuntimeError(f'Logger name {name!r} is reserved!')
    return _using_logger_parent(task_logger, get_logger(name))

==================================== 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.44s

==================================== ERRORS ====================================
_________________ ERROR collecting t/unit/backends/test_gcs.py _________________
ImportError while importing test module '/repo/t/unit/backends/test_gcs.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_gcs.py:5: in <module>
    from google.cloud.exceptions import NotFound
E   ModuleNotFoundError: No module named 'google'
=========================== short test summary info ============================

ERROR t/unit/backends/test_gcs.py
!!!!!!!!!!!!!!!!!!!!!!!!!! stopping after 1 failures !!!!!!!!!!!!!!!!!!!!!!!!!!!
8 skipped, 1 error in 0.84s

........................................................................ [ 46%]
............................................................sssssssss... [ 93%]
.sssssssss                                                               [100%]
136 passed, 18 skipped, 1 warning in 2.82s

def test_chain_with_chord_raises_error(self):
        with pytest.raises(NotImplementedError):
            (self.add.s(2, 2) |
             group(self.add.s(2, 2),
                   self.add.s(5, 6)) | self.add.s()).delay()

    def test_destination_for(self):
        req = Mock(name='request')
        req.reply_to = 'reply_to'
        req.correlation_id = 'corid'
        assert self.b.destination_for('task_id', req) == ('reply_to', 'corid')
        task = Mock()

        _task_stack.push(task)
        try:
            task.request.reply_to = 'reply_to'
            task.request.correlation_id = 'corid'
            assert self.b.destination_for('task_id', None) == (
                'reply_to', 'corid',
            )
        finally:
            _task_stack.pop()

        with pytest.raises(RuntimeError):
            self.b.destination_for('task_id', None)

    def test_binding(self
…[441 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 == ''

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

WARNING RPC result consumer: connection lost (original lost), attempting to reconnect...

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

==================================== ERRORS ====================================
_______________ 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:5: in <module>
    import dns.version
E   ModuleNotFoundError: No module named 'dns'
=========================== short test summary info ============================

ERROR t/unit/backends/test_mongodb.py
!!!!!!!!!!!!!!!!!!!!!!!!!! stopping after 1 failures !!!!!!!!!!!!!!!!!!!!!!!!!!!
8 skipped, 1 error in 0.90s

args = ({'status': 'SUCCESS', 'result': 10, 'traceback': None, 'children': [], ...},)
kwargs = {}

    def not_available(*args, **kwargs):
>       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
=========================== 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, 257 passed, 18 skipped, 1 warning in 4.10s

<function symbol_by_name at 0x726b54ea85e0> <function symbol_by_name at 0x726b54ea85e0> True

tuple(self.connection_errors)
            + self._get_additional_connection_errors()
        )
        self.result_consumer = self.ResultConsumer(
            self, self.app, self.accept,
            self._pending_results, self._pending_messages,
        )

    def _get_additional_connection_errors(self):
        transport_options = self.app.conf.get(
            'result_backend_transport_options', {}) or {}
        additional_errors = transport_options.get(

            'additional_connection_errors')
        if additional_errors is None:
            return ()

        if isinstance(additional_errors, (str, type)):
            additional_errors = (additional_errors,)
        else:
            try:
                additional_errors = tuple(additional_errors)
            except Exception:
                additional_errors = (additional_errors,)

        connection_errors = [
…[291 chars omitted]…
ror(
                        'additional connection error must be an Exception '
                        'subclass or an import path to one')
            except Exception as exc:
                logger.warning(
                    'Ignoring invalid additional connection error %r: %s',
                    error, exc,
                )
                continue
            connection_errors.append(error)

Retry limit exceeded while trying to reconnect to the Celery result store backend. The Celery application must be restarted. <class 't.unit.backends.test_redis.ConnectionError'>
redis additional reconnection catch ok

........................................................................ [ 46%]
............................................................sssssssss... [ 93%]
.sssssssss                                                               [100%]
136 passed, 18 skipped, 1 warning in 2.81s

........................................................................ [ 46%]
............................................................sssssssss... [ 93%]
.sssssssss                                                               [100%]
136 passed, 18 skipped, 1 warning in 2.90s
auto update on assignment ok

/usr/local/bin/python: No module named flake8

119:102:    _drain_complete_event = None    # event, sended (and recreated) after every drain_events iteration
115:102:            # even if there is nothing to subscribe, we should not lose the callback after connecting.
116:100:            # The on_connect callback will re-subscribe to any channels we previously subscribed to.
253:104:                    "Credential provider is not an instance of a redis.CredentialProvider or a subclass"
409:104:                    "Credential provider is not an instance of a redis.CredentialProvider or a subclass"

celery/backends/asynchronous.py | 25 ++++++++++++++++++
 celery/backends/redis.py        | 57 +++++++++++++++++++++++++++++------------
 celery/backends/rpc.py          | 48 +++++++++++++++++++++++-----------
 3 files changed, 99 insertions(+), 31 deletions(-)
diff --git a/celery/backends/redis.py b/celery/backends/redis.py
index 4e8c2d9..b8fcbed 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 Cele
…[6389 chars omitted]…
  self._consumer = self.Consumer(
             self._connection.default_channel,
             old_queues,
@@ -132,6 +149,7 @@ class ResultConsumer(BaseResultConsumer):
         if self._connection is not None:
             self._connection.collect()
             self._connection = None
+        self._update_connection_errors()
 
     def consume_from(self, task_id):
         if self._consumer is None:

grep: celery/backends/__pycache__/redis.cpython-312.pyc: binary file matches
celery/backends/redis.py:313:            + self._get_additional_connection_errors()
celery/backends/redis.py:320:    def _get_additional_connection_errors(self):
celery/backends/redis.py:323:        additional_errors = transport_options.get(
celery/backends/redis.py:324:            'additional_connection_errors')
celery/backends/redis.py:325:        if additional_errors is None:
celery/backends/redis.py:328:        if isinstance(additional_errors, (str, type)):
celery/backends/redis.py:329:            additional_errors = (additional_errors,)
celery/backends/redis.py:332:                additional_errors = tuple(additional_errors)
celery/backends/redis.py:334:                additional_errors = (additional_errors,)
celery/backends/redis.py:337:        for error in additional_errors:
celery/backends/redis.py:344:                        'additional connection error must be an Exception '

celery/backends/redis.py:348:                    'Ignoring invalid additional connection error %r: %s',

./celery/backends/redis.py:313:            + self._get_additional_connection_errors()
./celery/backends/redis.py:320:    def _get_additional_connection_errors(self):
./celery/backends/redis.py:324:            'additional_connection_errors')

all custom checks passed
........................................................................ [ 46%]
............................................................sssssssss... [ 93%]
.sssssssss                                                               [100%]
136 passed, 18 skipped, 1 warning in 2.87s

........................................................................ [ 74%]
.........................                                                [100%]
97 passed, 1 warning in 1.36s
tuple
tuple True

self.url = url

        # Add driver identification for redis-py
        self._add_driver_info()

        self.connection_errors, self.channel_errors = (
            get_redis_error_classes() if get_redis_error_classes
            else ((), ()))
        additional_connection_errors = (
            self._get_additional_connection_errors())
        if additional_connection_errors:
            self.connection_errors = (

                tuple(self.connection_errors)
                + additional_connection_errors
            )
        self.result_consumer = self.ResultConsumer(
            self, self.app, self.accept,
            self._pending_results, self._pending_messages,
        )

    def _get_additional_connection_errors(self):
        transport_options = self.app.conf.get(
            'result_backend_transport_options', {}) or {}
<class 'tuple'> tuple
<class 'tuple'> tuple
()

error_classes_t(connection_errors=(<class 'amqp.exceptions.ConnectionError'>, <class 'kombu.exceptions.InconsistencyError'>, <class 'OSError'>, <class 'OSError'>, <class 'OSError'>, <class 'redis.exceptions.ConnectionError'>, <class 'redis.exceptions.BusyLoadingError'>, <class 'redis.exceptions.AuthenticationError'>, <class 'redis.exceptions.TimeoutError'>, <class 'KeyError'>), channel_errors=(<class 'amqp.exceptions.ChannelError'>, <class 'redis.exceptions.DataError'>, <class 'redis.exceptions.InvalidResponse'>, <class 'redis.exceptions.ResponseError'>))
<class 'kombu.transport.redis.error_classes_t'> error_classes_t
(<class 'amqp.exceptions.ConnectionError'>, <class 'kombu.exceptions.InconsistencyError'>, <class 'OSError'>, <class 'OSError'>, <class 'OSError'>, <class 'redis.exceptions.ConnectionError'>, <class 'redis.exceptions.BusyLoadingError'>, <class 'redis.exceptions.AuthenticationError'>, <class 'redis.exceptions.TimeoutError'>, <class 'KeyError'>) <class 'tuple'> tuple

M celery/backends/asynchronous.py
 M celery/backends/redis.py
 M celery/backends/rpc.py
 celery/backends/asynchronous.py | 25 +++++++++++++++++
 celery/backends/redis.py        | 60 ++++++++++++++++++++++++++++++-----------
 celery/backends/rpc.py          | 48 ++++++++++++++++++++++-----------
 3 files changed, 102 insertions(+), 31 deletions(-)

........................................................................ [ 46%]
............................................................sssssssss... [ 93%]
.sssssssss                                                               [100%]
136 passed, 18 skipped, 1 warning in 3.78s

submitted.

/repo
total 332
drwxr-xr-x 1 root root   4096 Nov  2 16:23 .
drwxr-xr-x 1 root root   4096 Nov  2 16:22 ..
-rw-r--r-- 1 root root    343 Nov  2 16:21 .bumpversion.cfg
-rw-r--r-- 1 root root    306 Nov  2 16:21 .dockerignore
-rw-r--r-- 1 root root    227 Nov  2 16:21 .editorconfig
drwxr-xr-x 1 root root   4096 Nov  2 16:22 .git
drwxr-xr-x 1 root root   4096 Nov  2 16:21 .github
-rw-r--r-- 1 root root    437 Nov  2 16:21 .gitignore
-rw-r--r-- 1 root root   1072 Nov  2 16:21 .pre-commit-config.yaml
-rw-r--r-- 1 root root    616 Nov  2 16:21 .readthedocs.yaml

-rw-r--r-- 1 root root  44762 Nov  2 16:21 CONTRIBUTING.rst
-rw-r--r-- 1 root root   8574 Nov  2 16:21 CONTRIBUTORS.txt
-rw-r--r-- 1 root root 128887 Nov  2 16:21 Changelog.rst
-rw-r--r-- 1 root root   2717 Nov  2 16:21 LICENSE
-rw-r--r-- 1 root root    709 Nov  2 16:21 MANIFEST.in
-rw-r--r-- 1 root root   5866 Nov  2 16:21 Makefile

…[1116 chars omitted]…
elery/backends/cosmosdbsql.py
celery/backends/couchbase.py
celery/backends/couchdb.py
celery/backends/database/__init__.py
celery/backends/database/models.py
celery/backends/database/session.py
celery/backends/dynamodb.py
celery/backends/elasticsearch.py
celery/backends/filesystem.py
celery/backends/gcs.py
celery/backends/mongodb.py
celery/backends/redis.py
celery/backends/rpc.py
celery/backends/s3.py

celery/backends/redis.py:76:Retry limit exceeded while trying to reconnect to the Celery redis result \
celery/backends/redis.py:126:    def reconnect_on_error(self):
celery/backends/redis.py:161:            with self.reconnect_on_error():
celery/backends/redis.py:177:            with self.reconnect_on_error():
celery/backends/redis.py:184:            with self.reconnect_on_error():
celery/backends/rpc.py:64:    def _handle_connection_errors(self):
celery/backends/rpc.py:78:            with self._handle_connection_errors():

t/integration/test_backend.py
t/integration/test_database_backend.py
t/smoke/redis.conf
t/unit/app/test_backends.py
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

11:class test_RPCResultConsumer:
23:    def test_drain_events_reconnects_on_connection_error(self):
25:        # Simulate a started consumer with a live connection.
26:        mock_conn = Mock(name='connection')
27:        mock_conn.connection_errors = (OSError,)
30:            'Server unexpectedly closed connection'
32:        consumer._connection = mock_conn
38:        # Patch app.connection() to return a fresh mock connection
40:        new_conn = Mock(name='new_connection')
43:        consumer.app.connection.return_value = new_conn
46:        # drain_events should NOT raise; it should reconnect instead.
49:        # Old connection should be closed.

51:        # New connection should be established.
52:        consumer.app.connection.assert_called_once()
53:        assert consumer._connection is new_conn
58:    def test_drain_events_reconnect_preserves_queues(self):
60:        mock
…[842 chars omitted]…
 consumer._connection is new_conn
116:    def test_drain_events_channel_error_triggers_reconnect(self):
118:        mock_conn = Mock(name='connection')
119:        mock_conn.connection_errors = ()
122:        consumer._connection = mock_conn
128:        new_conn = Mock(name='new_connection')
130:        consumer.app.connection.return_value = new_conn
135:        assert consumer._connection is new_conn

t/unit/backends/test_redis.py:208:        consumer._connection_errors = (ConnectionError,)
t/unit/backends/test_rpc.py:27:        mock_conn.connection_errors = (OSError,)
t/unit/backends/test_rpc.py:61:        mock_conn.connection_errors = (ConnectionError,)
t/unit/backends/test_rpc.py:85:        mock_conn.connection_errors = (OSError,)
t/unit/backends/test_rpc.py:119:        mock_conn.connection_errors = ()
t/unit/worker/test_bootsteps.py:125:        self.connection.connection_errors = ()
t/unit/worker/test_bootsteps.py:146:        self.connection.connection_errors = ()
t/unit/worker/test_consumer.py:41:        consumer.connection_errors = (socket.error, OSError,)
t/unit/worker/test_consumer.py:537:        consumer.connection.connection_errors = ()
t/unit/worker/test_consumer.py:576:        consumer.connection.connection_errors = ()
t/unit/worker/test_consumer.py:618:        consumer
…[670 chars omitted]…

n_errors = (socket.error,)
t/unit/worker/test_worker.py:565:        c.connection_errors = (socket.error,)
t/unit/worker/test_worker.py:645:        Transport.connection_errors = (ChannelError,)
t/unit/worker/test_worker.py:1157:        w._conninfo.connection_errors = w._conninfo.channel_errors = ()
t/unit/worker/test_worker.py:1172:        w._conninfo.connection_errors = w._conninfo.channel_errors = ()

celery/contrib/testing/manager.py:36:        self.connerrors = self.app.connection().recoverable_connection_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)

celery/backends/redis.py:262:                credential_provider_cls = symbol_by_name(credential_provider)
celery/backends/redis.py:379:                credential_provider_cls = symbol_by_name(credential_provider)

celery/backends/redis.py:3:from contextlib import contextmanager
celery/backends/redis.py:125:    @contextmanager
celery/backends/rpc.py:7:from contextlib import contextmanager
celery/backends/rpc.py:63:    @contextmanager
./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
celery/backends/database/__init__.py:21:logger = logging.getLogger(__name__)
celery/backends/rpc.py:22:logger = logging.getLogger(__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
…[1849 chars omitted]…
res
        return expires

    def delete(self, key):
        return bool(self.keyspace.pop(key, None))

    def pipeline(self):
        return self.Pipeline(self)

    def _get_unsorted_list(self, key):
        # We simply store the values in append (rpush) order
        return self.keyspace.setdefault(key, list())

    def rpush(self, key, value):
        self._get_unsorted_list(key).append(value)

def symbol_by_name(name, aliases=None, imp=None, package=None,
                   sep='.', default=None, **kwargs):
    """Get symbol by qualified name.

    The name should be the full dot-separated path to the class::

        modulename.ClassName

    Example::

        celery.concurrency.processes.TaskPool
                                    ^- class name

    or using ':' to separate module and symbol::

        celery.concurrency.processes:TaskPool

    If `aliases` is provided, a dict containing short name/long name
    mappings, the name is looked up in the aliases first.

    Examples
    --------
        >>> symbol_by_name('celery.concurrency.processes.TaskPool')
        <class 'celery.concurrency.processes.TaskPool'>

        >>> symbol_by_name('default', {
        ...     'default': 'celery.concurrency.processes.TaskPool'})
        <class 'celery.concurrency.processes.Task
…[651 chars omitted]…
      module = imp(module_name, package=package, **kwargs)
        except ValueError as exc:
            reraise(ValueError,
                    ValueError(f"Couldn't import {name!r}: {exc}"),
                    sys.exc_info()[2])
        return getattr(module, cls_name) if cls_name else module
    except (ImportError, AttributeError):
        if default is None:
            raise
    return default

949:        This method checks if a global keyprefix is configured in `result_backend_transport_options` using the
952:        global_keyprefix = self.app.conf.get('result_backend_transport_options', {}).get("global_keyprefix", None)

amqp.send_task_message(P, name, message, **options)
        result = (result_cls or self.AsyncResult)(task_id)
        # We avoid using the constructor since a custom result class
        # can be used, in which case the constructor may still use
        # the old signature.
        result.ignored = ignore_result

        if add_to_parent:
            if not have_parent:
                parent, have_parent = self.current_worker_task, True
            if parent:
                parent.add_trail(result)

        return result

    def connection_for_read(self, url=None, **kwargs):
        """Establish connection used for consuming.

        See Also:
            :meth:`connection` for supported arguments.
        """
        return self._connection(url or self.conf.broker_read_url, **kwargs)

    def connection_for_write(self, url=None, **kwargs):
        """Establish con
…[1606 chars omitted]…
n method to use (AMQP only).
            failover_strategy (str, Callable): Custom failover strategy.
            **kwargs: Additional arguments to :class:`kombu.Connection`.

        Returns:
            kombu.Connection: the lazy connection instance.
        """
        return self.connection_for_write(
            hostname or self.conf.broker_write_url,
            userid=userid, password=password,

base has reconnect_on_error: False
base connection errors: <missing>

grep: t/conftest.py: No such file or directory

<Mock name='mock.connection_errors' id='129611610086224'> <Mock name='mock.channel_errors' id='129611609093952'>
<class 'TypeError'> unsupported operand type(s) for +: 'Mock' and 'Mock'

.................sssssssss....sssssssss................................. [ 46%]
........................................................................ [ 93%]
..........                                                               [100%]
=============================== warnings summary ===============================
t/unit/backends/test_rpc.py:174
  /repo/t/unit/backends/test_rpc.py:174: PytestUnknownMarkWarning: Unknown pytest.mark.celery - is this a typo?  You can register custom marks to avoid this warning - for details, see https://docs.pytest.org/en/stable/how-to/mark.html
    @pytest.mark.celery(result_backend='rpc')

t/unit/backends/test_rpc.py:179
  /repo/t/unit/backends/test_rpc.py:179: PytestUnknownMarkWarning: Unknown pytest.mark.celery - is this a typo?  You can register custom marks to avoid this warning - for details, see https://docs.pytest.org/en/stable/how-to/mark.
…[444 chars omitted]…
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
136 passed, 18 skipped, 3 warnings in 2.74s

except (<class 'ValueError'>, ValueError('x'), <traceback object at 0x7dca1e3b8f80>)
inside (<class 'ValueError'>, ValueError('x'), <traceback object at 0x7dca1e3b8f80>)

grep: celery/app/__pycache__/backends.cpython-312.pyc: binary file matches
grep: celery/app/__pycache__/base.cpython-312.pyc: binary file matches
grep: celery/app/__pycache__/routes.cpython-312.pyc: binary file matches
grep: celery/app/__pycache__/utils.cpython-312.pyc: binary file matches
grep: celery/backends/__pycache__/redis.cpython-312.pyc: binary file matches
grep: celery/concurrency/__pycache__/__init__.cpython-312.pyc: binary file matches
grep: celery/contrib/testing/__pycache__/app.cpython-312.pyc: binary file matches
grep: celery/fixups/__pycache__/django.cpython-312.pyc: binary file matches
grep: celery/loaders/__pycache__/base.cpython-312.pyc: binary file matches
grep: celery/loaders/__pycache__/__init__.cpython-312.pyc: binary file matches
grep: celery/utils/__pycache__/imports.cpython-312.pyc: binary file matches
grep: celery/utils/__pycache__/__init__.cpython-312.pyc: b

…[3406 chars omitted]…
y:44:    return symbol_by_name(name)(*args, **kwargs)
celery/utils/imports.py:159:            cls = symbol_by_name(class_name)
celery/utils/__init__.py:12:from .imports import symbol_by_name as get_cls_by_name
celery/worker/strategy.py:9:from celery.utils.imports import symbol_by_name
celery/worker/strategy.py:126:    Request = symbol_by_name(task.Request)
has symbol True
/repo/celery/utils/imports.py

16:        return self.get_backend().result_consumer
23:    def test_drain_events_reconnects_on_connection_error(self):
32:        consumer._connection = mock_conn
38:        # Patch app.connection() to return a fresh mock connection
40:        new_conn = Mock(name='new_connection')
43:        consumer.app.connection.return_value = new_conn
52:        consumer.app.connection.assert_called_once()
53:        assert consumer._connection is new_conn
64:        consumer._connection = mock_conn
71:        new_conn = Mock(name='new_connection')
73:        consumer.app.connection.return_value = new_conn
88:        consumer._connection = mock_conn

97:        consumer._connection = mock_conn
104:        new_conn = Mock(name='new_connection')
107:        consumer.app.connection.return_value = new_conn
113:        assert consumer._connection is new_conn
122:        consumer._connection = mock_conn
128:        new_conn = Mock(name='new_connection')
130:        consumer.app.connection.return_value = new_conn
135:        assert consumer._connection is new_conn

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)

    def expire(self, key, value):
        return self.client.expire(key, value)

    def add_to_chord(self, group_id, result):
        self.client.incr(self.get_key_for_group(group_id, '.t'), 1)

    def _unpack_chord_result(self, tup, decode,
                             EXCEPTION_STATES=states.EXCEPTION_STATES,
                             PROPAGATE_STATES=states.PROPAGATE_STATES):
        _, tid, state, retval = decode(tup)
        if state in EXCEPTION_STATES:
            retval = self.exception_to_python(retval)
        if state in PROPAGATE_STATES:
            chord_error = _create_chord_error_with_cause(
                message=f'Dependency {tid
…[2842 chars omitted]…
urn q


def instantiate(name, *args, **kwargs):
    """Instantiate class by name.

    See Also:
        :func:`symbol_by_name`.
    """
    return symbol_by_name(name)(*args, **kwargs)


@contextmanager
def cwd_in_path():
    """Context adding the current working directory to sys.path."""
    try:
        cwd = os.getcwd()
    except FileNotFoundError:
        cwd = None
    if not cwd:
        yield

base reconnect API exists: False

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

<kombu.utils.objects.cached_property object at 0x72f9cfa76e40>
Traceback (most recent call last):
  File "<stdin>", line 4, in <module>
AttributeError: 'cached_property' object has no attribute 'fget'

class cached_property(_cached_property):
    """Implementation of Cached property."""

    def __init__(self, fget=None, fset=None, fdel=None):
        super().__init__(fget)
        self.__set = fset
        self.__del = fdel

        if not hasattr(self, 'attrname'):
            # This is a backport so we set this ourselves.
            self.attrname = self.func.__name__

        if not hasattr(self, 'lock'):
            # Prior to Python 3.12, functools.cached_property has an
            # undocumented lock which is required for thread-safe __set__
            # and __delete__. Create one if it isn't already present.
            self.lock = RLock()

    def __get__(self, instance, owner=None):
        # TODO: Remove this after we drop support for Python<3.8
        #  or fix the signature in the cached_property package
        with self.lock:
            return super().__get__(inst
…[610 chars omitted]…
set):
        return self.__class__(self.func, fset, self.__del)

    def deleter(self, fdel):
        return self.__class__(self.func, self.__set, fdel)

    def __get__(self, instance, owner=None):
        # TODO: Remove this after we drop support for Python<3.8
        #  or fix the signature in the cached_property package
        with self.lock:
            return super().__get__(instance, owner)

celery/backends/redis.py:76:Retry limit exceeded while trying to reconnect to the Celery redis result \
grep: celery/backends/__pycache__/redis.cpython-312.pyc: binary file matches

transport options {}

diff --git a/celery/backends/asynchronous.py b/celery/backends/asynchronous.py
index 0413afe..bc075fb 100644
--- a/celery/backends/asynchronous.py
+++ b/celery/backends/asynchronous.py
@@ -4,6 +4,7 @@ import logging
 import socket
 import threading
 import time
+from contextlib import contextmanager
 from collections import deque
 from queue import Empty
 from time import sleep

@@ -16,6 +17,11 @@ 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."
+E_RETRY_LIMIT_EXCEEDED = """
+Retry limit exceeded while trying to reconnect to the Celery result backend. The Celery application must be restarted.
+"""
+
+logger = logging.getLogger(__name__)
 
 __all__ = (
     'AsyncBackendMixin', 'BaseResultConsumer', 'Drainer',
@@ -307,6 +313,8 @@ class AsyncBac
…[8614 chars omitted]…
if self._consumer is not None:
             old_queues = list(self._consumer.queues)
@@ -111,6 +113,7 @@ class ResultConsumer(BaseResultConsumer):
 
         # Establish a fresh connection and consumer.
         self._connection = self.app.connection()
+        self._set_connection_errors()
         self._consumer = self.Consumer(
             self._connection.default_channel,
             old_queues,

def __reduce__(self, args=(), kwargs=None):
        kwargs = {} if not kwargs else kwargs
        return super().__reduce__(
            args, dict(kwargs, expires=self.expires, url=self.url))


if getattr(redis, "sentinel", None):
    class SentinelManagedSSLConnection(
            redis.sentinel.SentinelManagedConnection,
            redis.SSLConnection):
        """Connect to a Redis server using Sentinel + TLS.

        Use Sentinel to identify which Redis server is the current master
        to connect to and when connecting to the Master server, use an
        SSL Connection.
        """


class SentinelBackend(RedisBackend):
    """Redis sentinel task result store."""

    # URL looks like `sentinel://0.0.0.0:26347/3;sentinel://0.0.0.0:26348/3`
    _SERVER_URI_SEPARATOR = ";"

    sentinel = getattr(redis, "sentinel", None)
    connection_class_ssl = SentinelManagedSSLConne
…[3905 chars omitted]…
ication must be restarted.
"""

logger = logging.getLogger(__name__)

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

311:        self.connection_errors += self._additional_connection_errors()
317:    def _additional_connection_errors(self):
346:    def _params_from_url(self, url, defaults):
718:    def _params_from_url(self, url, defaults):

<class 'tuple'> (<class 'amqp.exceptions.ConnectionError'>, <class 'kombu.exceptions.InconsistencyError'>, <class 'OSError'>, <class 'OSError'>, <class 'OSError'>, <class 'redis.exceptions.ConnectionError'>, <class 'redis.exceptions.BusyLoadingError'>, <class 'redis.exceptions.AuthenticationError'>, <class 'redis.exceptions.TimeoutError'>)
<class 'tuple'>

85:        self._connection_errors = self.backend.connection_errors
311:        self.connection_errors += self._additional_connection_errors()
317:    def _additional_connection_errors(self):
319:        errors = self._transport_options.get('additional_connection_errors')
339:                    'Ignoring invalid additional_connection_errors entry %r: %s',
"""The ``RPC`` result backend for AMQP brokers.

RPC-style result backend, using reply-to and one queue per client.
"""
import logging
import sys
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')

logger = logging.getLogger(__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 BacklogLimitExceeded(Exception):
    """Too much state history to fast-forward."""

.................sssssssss....sssssssss................................. [ 46%]
............................F...F........F..F....................FF.FF.. [ 93%]
..........                                                               [100%]
=================================== FAILURES ===================================
_____ test_RedisBackend_chords_simple.test_on_chord_part_return__unordered _____

self = <t.unit.backends.test_redis.test_RedisBackend_chords_simple object at 0x75a5111f1100>

    def test_on_chord_part_return__unordered(self):
        self.app.conf.result_backend_transport_options = dict(
            result_chord_ordered=False,
        )

    
        tasks = [self.create_task(i) for i in range(10)]
        random.shuffle(tasks)
    
        for i in range(10):
            self.b.on_chord_part_return(tasks[i].request, states.SUCCESS, i)
>           assert self.b.client.
…[8630 chars omitted]…
nects_on_connection_error
FAILED t/unit/backends/test_rpc.py::test_RPCResultConsumer::test_drain_events_reconnect_preserves_queues
FAILED t/unit/backends/test_rpc.py::test_RPCResultConsumer::test_reconnect_handles_close_failures_gracefully
FAILED t/unit/backends/test_rpc.py::test_RPCResultConsumer::test_drain_events_channel_error_triggers_reconnect
8 failed, 128 passed, 18 skipped, 3 warnings in 3.18s

.................sssssssss....sssssssss................................. [ 46%]
........................................................................ [ 93%]
..........                                                               [100%]
=============================== warnings summary ===============================
t/unit/backends/test_rpc.py:174
  /repo/t/unit/backends/test_rpc.py:174: PytestUnknownMarkWarning: Unknown pytest.mark.celery - is this a typo?  You can register custom marks to avoid this warning - for details, see https://docs.pytest.org/en/stable/how-to/mark.html
    @pytest.mark.celery(result_backend='rpc')

t/unit/backends/test_rpc.py:179
  /repo/t/unit/backends/test_rpc.py:179: PytestUnknownMarkWarning: Unknown pytest.mark.celery - is this a typo?  You can register custom marks to avoid this warning - for details, see https://docs.pytest.org/en/stable/how-to/mark.
…[444 chars omitted]…
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
136 passed, 18 skipped, 3 warnings in 2.61s

reconnection and Redis extra-error reproduction passed
Retry limit exceeded while trying to reconnect to the Celery result backend. The Celery application must be restarted.
celery/backends/asynchronous.py 21 118 Retry limit exceeded while trying to reconnect to the Celery result backend. The Celery application must be restarted.

celery/backends/asynchronous.py | 23 +++++++++++++++++++
 celery/backends/redis.py        | 51 +++++++++++++++++++++++++++--------------
 celery/backends/rpc.py          | 31 ++++++++++++++-----------
 3 files changed, 74 insertions(+), 31 deletions(-)
diff --git a/celery/backends/asynchronous.py b/celery/backends/asynchronous.py
index 0413afe..d6c7004 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
@@ -16,6 +17,11 @@ 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."
+E_RETRY_LIMIT_EXCEEDED = 
…[7828 chars omitted]…
111,6 +113,7 @@ class ResultConsumer(BaseResultConsumer):
 
         # Establish a fresh connection and consumer.
         self._connection = self.app.connection()
+        self._set_connection_errors()
         self._consumer = self.Consumer(
             self._connection.default_channel,
             old_queues,
 M celery/backends/asynchronous.py
 M celery/backends/redis.py
 M celery/backends/rpc.py

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."
E_RETRY_LIMIT_EXCEEDED = """
Retry limit exceeded while trying to reconnect to the Celery result store \
backend. The Celery application must be restarted.
"""

logger = logging.getLogger(__name__)

__all__ = (

949:        This method checks if a global keyprefix is configured in `result_backend_transport_options` using the
952:        global_keyprefix = self.app.conf.get('result_backend_transport_options', {}).get("global_keyprefix", None)
() [] True
None [] True
t.unit.backends.test_redis.ConnectionError [] True
(<class 'ConnectionError'>,) [<class 'ConnectionError'>] True
(<class 'ConnectionError'>, <class 'int'>) [<class 'ConnectionError'>] True
42 [] True
[None, 'no.such.module.Error'] [] True

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

RPC-style result backend, using reply-to and one queue per client.
"""
import logging
import sys
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')

logger = logging.getLogger(__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 BacklogLimitExceeded(Exception):
    """Too much 
…[8238 chars omitted]…
 number should be given in 'query'
            connparams.update({
                'connection_class': self.redis.UnixDomainSocketConnection,
                'path': '/' + path,
            })
            # host+port are invalid options when using this connection type.
            connparams.pop('host', None)
            connparams.pop('port', None)
            connparams.pop('socket_connect_timeout')

==================================== 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 imp
…[3120 chars omitted]…
s.pytest.org/en/stable/how-to/capture-warnings.html
=========================== 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 !!!!!!!!!!!!!!!!!!!!
8 skipped, 4 warnings, 4 errors in 1.45s

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

celery/backends/asynchronous.py:20:E_RETRY_LIMIT_EXCEEDED = """
celery/backends/asynchronous.py:317:    _connection_errors = ()
celery/backends/asynchronous.py:337:    def reconnect_on_error(self):
celery/backends/asynchronous.py:341:        except self._connection_errors:
celery/backends/asynchronous.py:343:                self._reconnect()
celery/backends/asynchronous.py:344:            except self._connection_errors as exc:
celery/backends/asynchronous.py:345:                logger.critical(E_RETRY_LIMIT_EXCEEDED)
celery/backends/asynchronous.py:346:                raise RuntimeError(E_RETRY_LIMIT_EXCEEDED) from exc
celery/backends/asynchronous.py:348:    def _reconnect(self):
celery/backends/redis.py:85:        self._connection_errors = self.backend.connection_errors
celery/backends/redis.py:97:    def _reconnect_pubsub(self):
celery/backends/redis.py:119:    def _reconnect(self):

…[827 chars omitted]…
rpc.py:68:            self._connection_errors = connection_errors + channel_errors
celery/backends/rpc.py:70:            self._connection_errors = ()
celery/backends/rpc.py:74:            self._set_connection_errors()
celery/backends/rpc.py:75:            with self.reconnect_on_error():
celery/backends/rpc.py:80:    def _reconnect(self):
celery/backends/rpc.py:123:        self._set_connection_errors()

.................sssssssss....sssssssss................................. [ 46%]
........................................................................ [ 93%]
..........                                                               [100%]
=============================== warnings summary ===============================
t/unit/backends/test_rpc.py:174
  /repo/t/unit/backends/test_rpc.py:174: PytestUnknownMarkWarning: Unknown pytest.mark.celery - is this a typo?  You can register custom marks to avoid this warning - for details, see https://docs.pytest.org/en/stable/how-to/mark.html
    @pytest.mark.celery(result_backend='rpc')

t/unit/backends/test_rpc.py:179
  /repo/t/unit/backends/test_rpc.py:179: PytestUnknownMarkWarning: Unknown pytest.mark.celery - is this a typo?  You can register custom marks to avoid this warning - for details, see https://docs.pytest.org/en/stable/how-to/mark.
…[444 chars omitted]…
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
136 passed, 18 skipped, 3 warnings in 2.58s

celery/backends/asynchronous.py | 24 +++++++++++++++++++
 celery/backends/redis.py        | 51 +++++++++++++++++++++++++++--------------
 celery/backends/rpc.py          | 36 ++++++++++++++++++-----------
 3 files changed, 81 insertions(+), 30 deletions(-)
 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 0413afe..e603eb4 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

@@ -16,6 +17,12 @@ 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."
+E_RETRY_LIMIT_EXCEEDED = """
+Retry limit exceeded while trying to reconnect to the Celery result store \
+backend. The Celery application must be restarted.
+"""
+
+logger = logging.getLogger(__name__)
 
 __all__ = (
     'AsyncBackendMixin', 'BaseResultConsumer', 'Drainer',
@@
…[7642 chars omitted]…
if self._consumer is not None:
             old_queues = list(self._consumer.queues)
@@ -111,6 +120,7 @@ class ResultConsumer(BaseResultConsumer):
 
         # Establish a fresh connection and consumer.
         self._connection = self.app.connection()
+        self._set_connection_errors()
         self._consumer = self.Consumer(
             self._connection.default_channel,
             old_queues,

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

submitted.