celery-kombu-2498
Results
The prompt the agent sees
When a Redis connection is lost while a Celery worker is running, periodic Redis maintenance callbacks can raise the connection exception into the event loop. This causes the worker’s event loop to stop and restart instead of continuing to run and allowing the normal connection-recovery logic to proceed. A failed maintenance attempt should be ignored for that tick so the worker remains running and can retry later.
The same maintenance callbacks are registered again whenever the Redis connection reconnects through `redis.Transport.register_with_event_loop`. Previously registered timer callbacks remain active, so repeated reconnects leave duplicate callbacks in the timer queue. Each callback should have only one active scheduled entry after reconnecting: entries from the previous registration must be cancelled before the new entries are registered.
The affected maintenance operations are `maybe_restore_messages`, which restores visible messages for channels with active queues, and `maybe_check_subclient_health`, which checks the health of a cached Redis subclient. Connection failures from either operation must not escape the callback, while successful operations should continue to run normally. `maybe_restore_messages` must call `channel.qos.restore_visible(num=channel.unacked_restore_limit)` once for each channel with active queues, and must skip channels without active queues. `maybe_check_subclient_health` must call the cached subclient’s `check_health()` once when a subclient is present, and must skip channels without a cached subclient.
The failures each maintenance callback must tolerate are exactly the exception classes the channel itself lists in its `connection_errors` attribute. A failure from either operation must be ignored for that tick, and the operation must still be attempted exactly once before the callback finishes. Do not hard-code redis-py's `ConnectionError` or any other fixed class: the channel's tuple is the source of truth and may contain classes that are not Redis errors.
Hidden tests · 4 fail-to-pass, 139 pass-to-passrun after the agent submits, in a clean verifier
Test patch · 160 lines
diff --git a/t/unit/transport/test_redis.py b/t/unit/transport/test_redis.py
index 5d9fafec61..11cfa4b9a4 100644
--- a/t/unit/transport/test_redis.py
+++ b/t/unit/transport/test_redis.py
@@ -1246,6 +1246,54 @@ def test_configurable_health_check(self):
call(13, transport.on_readable, 13),
])
+ def test_register_with_event_loop_stores_trefs_on_cycle(self):
+ """Re-registering must retire the timer entries of the previous
+ registration and leave the new ones live, whatever the transport
+ stores them as: the third registration cancels the second set."""
+ transport = self.connection.transport
+ transport.cycle = Mock(name='cycle')
+ transport.cycle.fds = {}
+ conn = Mock(name='conn')
+ conn.client = Mock(name='client', transport_options={})
+ loop = Mock(name='loop')
+ sets = [[Mock(name=f'tref_{i}_{j}') for j in range(2)] for i in range(3)]
+ loop.call_repeatedly.side_effect = [t for s in sets for t in s]
+
+ for _ in range(3):
+ redis.Transport.register_with_event_loop(transport, conn, loop)
+
+ for t in sets[0] + sets[1]:
+ t.cancel.assert_called_once()
+ for t in sets[2]:
+ t.cancel.assert_not_called()
+
+ def test_register_with_event_loop_cancels_stale_trefs_on_reconnect(self):
+ """Stale timer entries from a previous connection must be cancelled.
+
+ Each call to register_with_event_loop (i.e. each reconnect) must
+ cancel the timer entries the previous call registered before
+ registering new ones, so hub.timer._queue never accumulates
+ duplicate entries."""
+ transport = self.connection.transport
+ transport.cycle = Mock(name='cycle')
+ transport.cycle.fds = {}
+ conn = Mock(name='conn')
+ conn.client = Mock(name='client', transport_options={})
+ loop = Mock(name='loop')
+ first = [Mock(name='tref_restore_1'), Mock(name='tref_health_1')]
+ second = [Mock(name='tref_restore_2'), Mock(name='tref_health_2')]
+ loop.call_repeatedly.side_effect = first + second
+
+ redis.Transport.register_with_event_loop(transport, conn, loop)
+ for t in first:
+ t.cancel.assert_not_called()
+
+ redis.Transport.register_with_event_loop(transport, conn, loop)
+ for t in first:
+ t.cancel.assert_called_once()
+ for t in second:
+ t.cancel.assert_not_called()
+
def test_transport_on_readable(self):
transport = self.connection.transport
cycle = transport.cycle = Mock(name='cyle')
@@ -1712,6 +1757,100 @@ def test_on_poll_init(self):
num=chan1.unacked_restore_limit,
)
+ def test_maybe_restore_messages_calls_restore_visible(self):
+ """Happy path: restore_visible is called for a channel with active queues."""
+ p = self.Poller()
+ channel = Mock(name='channel')
+ channel.active_queues = ['a_queue']
+ p._channels = [channel]
+
+ p.maybe_restore_messages()
+
+ channel.qos.restore_visible.assert_called_once_with(
+ num=channel.unacked_restore_limit,
+ )
+
+ def test_maybe_restore_messages_skips_channel_without_active_queues(self):
+ """Channels with no active queues must be ignored."""
+ p = self.Poller()
+ channel = Mock(name='channel')
+ channel.active_queues = []
+ p._channels = [channel]
+
+ p.maybe_restore_messages()
+
+ channel.qos.restore_visible.assert_not_called()
+
+ def test_maybe_restore_messages_swallows_connection_error(self):
+ """Connection errors from timer callbacks must not propagate.
+
+ maybe_restore_messages is scheduled via call_repeatedly and runs
+ inside fire_timers. If a ConnectionError escapes, it matches
+ hub.propagate_errors and tears down the entire event loop.
+ The fix catches channel.connection_errors and returns early.
+ """
+ p = self.Poller()
+
+ class ConnError(Exception):
+ pass
+
+ channel = Mock(name='channel')
+ channel.active_queues = ['a_queue']
+ channel.connection_errors = (ConnError,)
+ channel.qos.restore_visible.side_effect = ConnError('connection lost')
+ p._channels = [channel]
+
+ # Must not raise
+ p.maybe_restore_messages()
+
+ channel.qos.restore_visible.assert_called_once()
+
+ def test_maybe_check_subclient_health_calls_check_health(self):
+ """Happy path: check_health is called when subclient is cached."""
+ p = self.Poller()
+ channel = Mock(name='channel')
+ client = Mock(name='subclient')
+ channel.__dict__['subclient'] = client
+ p._channels = [channel]
+
+ p.maybe_check_subclient_health()
+
+ client.check_health.assert_called_once()
+
+ def test_maybe_check_subclient_health_skips_when_no_subclient(self):
+ """Channels with no cached subclient must be silently skipped."""
+ p = self.Poller()
+ channel = Mock(name='channel')
+ # Ensure 'subclient' is not in __dict__ (not yet accessed/cached)
+ channel.__dict__.pop('subclient', None)
+ p._channels = [channel]
+
+ p.maybe_check_subclient_health() # must not raise
+
+ def test_maybe_check_subclient_health_swallows_connection_error(self):
+ """Connection errors from timer callbacks must not propagate.
+
+ Same reasoning as test_maybe_restore_messages_swallows_connection_error:
+ the fix catches channel.connection_errors and returns early instead of
+ letting the exception bubble up through fire_timers.
+ """
+ p = self.Poller()
+
+ class ConnError(Exception):
+ pass
+
+ channel = Mock(name='channel')
+ channel.connection_errors = (ConnError,)
+ client = Mock(name='subclient')
+ client.check_health.side_effect = ConnError('connection lost')
+ channel.__dict__['subclient'] = client
+ p._channels = [channel]
+
+ # Must not raise
+ p.maybe_check_subclient_health()
+
+ client.check_health.assert_called_once()
+
def test_handle_event(self):
p = self.Poller()
chan = Mock(name='chan')
Reference fix · 1 file, +34 −6the upstream merge, used only for grading calibration
The agent could not see this: the repository holds one commit and the sandbox has no network. Leak audit.
kombu/transport/redis.py
diff --git a/kombu/transport/redis.py b/kombu/transport/redis.py
index 322bab0bc8..68153ef827 100644
--- a/kombu/transport/redis.py
+++ b/kombu/transport/redis.py
@@ -572,9 +572,18 @@ def maybe_restore_messages(self):
for channel in self._channels:
if channel.active_queues:
# only need to do this once, as they are not local to channel.
- return channel.qos.restore_visible(
- num=channel.unacked_restore_limit,
- )
+ try:
+ return channel.qos.restore_visible(
+ num=channel.unacked_restore_limit,
+ )
+ except channel.connection_errors:
+ # Connection is broken; skip this cycle and retry next tick.
+ # The main polling loop handles reconnection independently.
+ logger.debug(
+ 'maybe_restore_messages: connection error, '
+ 'will retry on next cycle', exc_info=True
+ )
+ return
def maybe_check_subclient_health(self):
for channel in self._channels:
@@ -582,7 +591,14 @@ def maybe_check_subclient_health(self):
client = channel.__dict__.get('subclient')
if client is not None \
and callable(getattr(client, 'check_health', None)):
- client.check_health()
+ try:
+ client.check_health()
+ except channel.connection_errors:
+ logger.debug(
+ 'maybe_check_subclient_health: connection error, '
+ 'will retry on next cycle', exc_info=True
+ )
+ return
def on_readable(self, fileno):
chan, type = self._fd_to_chan[fileno]
@@ -1445,12 +1461,24 @@ def on_poll_start():
cycle_poll_start()
[add_reader(fd, on_readable, fd) for fd in cycle.fds]
loop.on_tick.add(on_poll_start)
- loop.call_repeatedly(10, cycle.maybe_restore_messages)
+
+ # Cancel stale timer entries from a previous connection before
+ # registering new ones. Without this, each reconnect accumulates
+ # an extra entry in hub.timer._queue; they all fire against the
+ # same cycle and can crash the event loop during reconnect.
+ for attr in ('_restore_messages_tref', '_subclient_health_tref'):
+ old_tref = getattr(cycle, attr, None)
+ if old_tref is not None:
+ old_tref.cancel()
+
+ cycle._restore_messages_tref = loop.call_repeatedly(
+ 10, cycle.maybe_restore_messages
+ )
health_check_interval = connection.client.transport_options.get(
'health_check_interval',
DEFAULT_HEALTH_CHECK_INTERVAL
)
- loop.call_repeatedly(
+ cycle._subclient_health_tref = loop.call_repeatedly(
health_check_interval,
cycle.maybe_check_subclient_health
)
GLM-5.3 Flash · attempt 2 · solved · 29 steps · $0.006
Submitted diff · 75 lines
diff --git a/kombu/transport/redis.py b/kombu/transport/redis.py
index 322bab0..56a3a9d 100644
--- a/kombu/transport/redis.py
+++ b/kombu/transport/redis.py
@@ -571,10 +571,13 @@ class MultiChannelPoller:
def maybe_restore_messages(self):
for channel in self._channels:
if channel.active_queues:
- # only need to do this once, as they are not local to channel.
- return channel.qos.restore_visible(
- num=channel.unacked_restore_limit,
- )
+ try:
+ channel.qos.restore_visible(
+ num=channel.unacked_restore_limit,
+ )
+ except channel.connection_errors:
+ # connection lost, ignore for this tick
+ pass
def maybe_check_subclient_health(self):
for channel in self._channels:
@@ -582,7 +585,11 @@ class MultiChannelPoller:
client = channel.__dict__.get('subclient')
if client is not None \
and callable(getattr(client, 'check_health', None)):
- client.check_health()
+ try:
+ client.check_health()
+ except channel.connection_errors:
+ # connection lost, ignore for this tick
+ pass
def on_readable(self, fileno):
chan, type = self._fd_to_chan[fileno]
@@ -1445,15 +1452,30 @@ class Transport(virtual.Transport):
cycle_poll_start()
[add_reader(fd, on_readable, fd) for fd in cycle.fds]
loop.on_tick.add(on_poll_start)
- loop.call_repeatedly(10, cycle.maybe_restore_messages)
- health_check_interval = connection.client.transport_options.get(
- 'health_check_interval',
- DEFAULT_HEALTH_CHECK_INTERVAL
- )
- loop.call_repeatedly(
- health_check_interval,
- cycle.maybe_check_subclient_health
- )
+ # cancel previously registered maintenance timers, if any, so that
+ # re-registering after a reconnect does not leave duplicates behind.
+ previous = getattr(cycle, '_maintenance_timers', None)
+ for tref in previous if isinstance(previous, (list, tuple)) else ():
+ tref.cancel()
+ try:
+ # drop the cancelled entry from the timer queue as well,
+ # so it does not linger until its next scheduled run.
+ loop.timer._queue[:] = [
+ event for event in loop.timer._queue
+ if event[2] is not tref
+ ]
+ except (AttributeError, IndexError, TypeError):
+ pass
+ cycle._maintenance_timers = [
+ loop.call_repeatedly(10, cycle.maybe_restore_messages),
+ loop.call_repeatedly(
+ connection.client.transport_options.get(
+ 'health_check_interval',
+ DEFAULT_HEALTH_CHECK_INTERVAL
+ ),
+ cycle.maybe_check_subclient_health,
+ ),
+ ]
def on_readable(self, fileno):
"""Handle AIO event for one of our file descriptors."""step 1
$ cd /repo && ls && grep -rn "maybe_restore_messages\|maybe_check_subclient_health\|register_with_event_loop" --include=*.py | head -30
output · 37 lines
AUTHORS Changelog.rst FAQ INSTALL LICENSE MANIFEST.in Makefile README.rst SECURITY.md THANKS TODO conftest.py
25 more lines
docs examples kombu kombu.egg-info pyproject.toml requirements setup.cfg setup.py t tox.ini grep: .git/objects/d3: Cannot allocate memory examples/experimental/async_consume.py:27: conn.register_with_event_loop(hub) kombu/connection.py:302: def register_with_event_loop(self, loop): kombu/connection.py:303: self.transport.register_with_event_loop(self.connection, loop) kombu/transport/base.py:213: def register_with_event_loop(self, connection, loop): kombu/transport/librabbitmq.py:163: def register_with_event_loop(self, connection, loop): kombu/transport/pyamqp.py:222: def register_with_event_loop(self, connection, loop): kombu/transport/qpid.py:1542: def register_with_event_loop(self, connection, loop): kombu/transport/redis.py:571: def maybe_restor …[1530 chars omitted]… ith_event_loop(Mock(), Mock()) t/unit/transport/test_qpid.py:1716: self.transport.register_with_event_loop(Mock(), Mock()) t/unit/transport/test_qpid.py:1812: my_transport.register_with_event_loop(Mock(), Mock()) t/unit/transport/test_qpid.py:1820: my_transport.register_with_event_loop(Mock(), Mock()) t/unit/transport/test_redis.py:1181: def test_register_with_event_loop(self):
step 2
$ cd /repo && sed -n 540,640p kombu/transport/redis.py && sed -n 1410,1470p kombu/transport/redis.py
output · 34 lines
"""Enable BRPOP mode for channel."""
ident = channel, channel.client, 'BRPOP'
if not self._client_registered(channel, channel.client, 'BRPOP'):
channel._in_poll = False
self._register(*ident)
if not channel._in_poll: # send BRPOP
channel._brpop_start()
def _register_LISTEN(self, channel):
"""Enable LISTEN mode for channel."""
if not self._client_registered(channel, channel.subclient, 'LISTEN'):
channel._in_listen = False22 more lines
self._register(channel, channel.subclient, 'LISTEN')
if not channel._in_listen:
channel._subscribe() # send SUBSCRIBE
def on_poll_start(self):
for channel in self._channels:
if channel.active_queues: # BRPOP mode?
if channel.qos.can_consume():
self._register_BRPOP(c
…[4503 chars omitted]…
ur file descriptors."""
self.cycle.on_readable(fileno)
if sentinel:
class SentinelManagedSSLConnection(
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 anstep 3
$ cd /repo && grep -rn "def call_repeatedly\|def call_at\|def call_later" kombu/asynchronous/*.py kombu/asynchronous/**/*.py | head; grep -rn "connection_errors" kombu/transport/redis.py | head
output · 11 lines
kombu/asynchronous/hub.py:211: def call_later(self, delay, callback, *args): kombu/asynchronous/hub.py:214: def call_at(self, when, callback, *args): kombu/asynchronous/hub.py:217: def call_repeatedly(self, delay, callback, *args): kombu/asynchronous/timer.py:118: def call_at(self, eta, fun, args=(), kwargs=None, priority=0): kombu/asynchronous/timer.py:126: def call_repeatedly(self, secs, fun, args=(), kwargs=None, priority=0): 119: 'connection_errors', 'channel_errors', 148: (virtual.Transport.connection_errors + ( 780: self.connection_errors = self.connection.connection_errors 954: except self.connection_errors: 997: except self.connection_errors: 1408: connection_errors, channel_errors = get_redis_error_classes()
step 4
$ cd /repo && sed -n 200,230p kombu/asynchronous/hub.py && sed -n 110,140p kombu/asynchronous/timer.py
output · 39 lines
next(self.loop)
except StopIteration:
self._loop = None
def call_soon(self, callback, *args):
if not isinstance(callback, Thenable):
callback = promise(callback, args)
with self._ready_lock:
self._ready.add(callback)
return callback
def call_later(self, delay, callback, *args):27 more lines
return self.timer.call_after(delay, callback, args)
def call_at(self, when, callback, *args):
return self.timer.call_at(when, callback, args)
def call_repeatedly(self, delay, callback, *args):
return self.timer.call_repeatedly(delay, callback, args)
def add_reader(self, fds, callback, *args):
return self.add(fds, callback, READ | ERR, args)
def add_writer(self, fds, callback, *args):
return self.add(fds, callback, WRITE, args)
def remove_reader(self, fd):
…[896 chars omitted]…
def _reschedules(*args, **kwargs):
last, now = tref._last_run, monotonic()
lsince = (now - tref._last_run) if last else secs
try:
if lsince and lsince >= secs:
tref._last_run = now
return fun(*args, **kwargs)
finally:
if not tref.canceled:
last = tref._last_runstep 5
$ cd /repo && python - <<'PY'
output · 2 lines
kombu/transport/redis.py | 40 ++++++++++++++++++++++++++-------------- 1 file changed, 26 insertions(+), 14 deletions(-)
step 6
$ cd /repo && python - <<'PY'
output · 6 lines
ok
Traceback (most recent call last):
File "<stdin>", line 27, in <module>
File "/repo/kombu/transport/redis.py", line 751, in __init__
super().__init__(*args, **kwargs)
TypeError: Channel.__init__() takes 2 positional arguments but 3 were givenstep 7
$ cd /repo && python - <<'PY'
output · 11 lines
Traceback (most recent call last):
File "<stdin>", line 6, in <module>
File "/repo/kombu/transport/redis.py", line 751, in __init__
super().__init__(*args, **kwargs)
File "/repo/kombu/transport/virtual/base.py", line 482, in __init__
self.channel_id = self._get_free_channel_id()
^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/repo/kombu/transport/virtual/base.py", line 870, in _get_free_channel_id
used_channel_ids = set(self.connection._used_channel_ids)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
TypeError: 'Mock' object is not iterablestep 8
$ cd /repo && python - <<'PY'
output · 10 lines
Traceback (most recent call last):
File "<stdin>", line 7, in <module>
File "/repo/kombu/transport/redis.py", line 751, in __init__
super().__init__(*args, **kwargs)
File "/repo/kombu/transport/virtual/base.py", line 482, in __init__
self.channel_id = self._get_free_channel_id()
^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/repo/kombu/transport/virtual/base.py", line 877, in _get_free_channel_id
raise ResourceError(
amqp.exceptions.ResourceError: Channel.open: (506) No free channel ids, current=0, channel_max=<MagicMock name='mock.channel_max' id='126211054059488'>step 9
$ cd /repo && python - <<'PY'
output · 26 lines
Traceback (most recent call last):
File "<stdin>", line 8, in <module>
File "/repo/kombu/transport/redis.py", line 776, in __init__
self.client.ping()
^^^^^^^^^^^
File "/repo/kombu/utils/objects.py", line 40, in __get__
return super().__get__(instance, owner)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.12/functools.py", line 998, in __get__
val = self.func(instance)
^^^^^^^^^^^^^^^^^^^
File "/repo/kombu/transport/redis.py", line 1374, in client14 more lines
return self._create_client(asynchronous=True)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/repo/kombu/transport/redis.py", line 1330, in _create_client
return self.Client(connection_pool=self.async_pool)
^^^^^^^^^^^^^^^
File "/repo/kombu/transport/redis.py", line 1368, in async_pool
self._async_pool = self._get_poo
…[200 chars omitted]…
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/repo/kombu/transport/redis.py", line 1258, in _connparams
self._process_credential_provider(conninfo.credential_provider, connparams)
File "/repo/kombu/transport/redis.py", line 1231, in _process_credential_provider
raise ValueError(
ValueError: Credential provider is not an instance of a redis.CredentialProvider or a subclassstep 10
$ cd /repo && python - <<'PY'
output · 3 lines
Traceback (most recent call last): File "<stdin>", line 12, in <module> AttributeError: type object 'Channel' has no attribute 'register_with_event_loop'
step 11
$ cd /repo && grep -n "def register_with_event_loop" kombu/transport/redis.py
output · 1 lines
1431: def register_with_event_loop(self, connection, loop):
step 12
$ cd /repo && sed -n 1420,1432p kombu/transport/redis.py
output · 13 lines
super().__init__(*args, **kwargs)
# All channels share the same poller.
self.cycle = MultiChannelPoller()
# Use polling_interval to set brpop_timeout if provided, but do not modify polling_interval itself.
if self.polling_interval is not None:
self.brpop_timeout = self.polling_interval
def driver_version(self):
return redis.__version__
def register_with_event_loop(self, connection, loop):1 more lines
cycle = self.cycle
step 13
$ cd /repo && grep -n "^class " kombu/transport/redis.py
output · 11 lines
170:class MutexHeld(Exception): 201:class GlobalKeyPrefixMixin: 285:class PrefixedStrictRedis(GlobalKeyPrefixMixin, redis.Redis): 300:class PrefixedRedisPipeline(GlobalKeyPrefixMixin, redis.client.Pipeline): 313:class PrefixedRedisPubSub(redis.client.PubSub): 365:class QoS(virtual.QoS): 472:class MultiChannelPoller: 641:class Channel(virtual.Channel): 1398:class Transport(virtual.Transport): 1489:class SentinelChannel(Channel): 1575:class SentinelTransport(Transport):
step 14
$ cd /repo && python - <<'PY'
output · 4 lines
2 4 Traceback (most recent call last): File "<stdin>", line 17, in <module> AssertionError: (2, 4)
step 15
$ cd /repo && sed -n 1,110p kombu/asynchronous/timer.py
output · 59 lines
"""Timer scheduling Python callbacks.""" from __future__ import annotations import heapq import sys from collections import namedtuple from datetime import datetime from functools import total_ordering from time import monotonic from time import time as _time from typing import TYPE_CHECKING
47 more lines
from weakref import proxy as weakrefproxy
from vine.utils import wraps
from kombu.log import get_logger
if sys.version_info >= (3, 9):
from zoneinfo import ZoneInfo
else:
from backports.zoneinfo import ZoneInfo
if TYPE_CHECKING:
from types import TracebackType
__all__ = ('Entry', 'Timer', 'to_timestamp')
logger = get_logger(__name__)
DEFAULT_MAX_INTERVAL = 2
EPOCH = datetime.fromtimestamp(0, ZoneInfo("UTC"))
IS_PYPY = hasattr(sys, 'pypy_version_info')
scheduled = namedtuple('scheduled', ('eta', 'priority', 'entry'))
def to_timestamp(d, default_timezone=ZoneInfo("UTC"), time
…[1388 chars omitted]…
self.canceled = value
class Timer:
"""Async timer implementation."""
Entry = Entry
on_error = None
def __init__(self, max_interval=None, on_error=None, **kwargs):
self.max_interval = float(max_interval or DEFAULT_MAX_INTERVAL)
self.on_error = on_error or self.on_error
self._queue = []
def __enter__(self):
return self
def __exit__(step 16
$ cd /repo && pip download kombu==5.5.0 --no-deps -d /tmp/k 2>/dev/null | tail -1; ls /tmp/k 2>/dev/null; python -c "import kombu; print(kombu.__version__)"
output · 1 lines
5.6.2
step 17
$ cd /tmp && pip download kombu==5.5.8 --no-deps -d /tmp/k -q; ls /tmp/k; cd /tmp/k && tar xf *.gz 2>/dev/null || unzip -q *.whl; grep -rn "maintenance_timers\|call_repeatedly" kombu/transport/redis.py 2>/dev/null | head
output · 8 lines
WARNING: Retrying (Retry(total=4, connect=None, read=None, redirect=None, status=None)) after connection broken by 'NewConnectionError('<pip._vendor.urllib3.connection.HTTPSConnection object at 0x7280c4c44260>: Failed to establish a new connection: [Errno -3] Temporary failure in name resolution')': /simple/kombu/
WARNING: Retrying (Retry(total=3, connect=None, read=None, redirect=None, status=None)) after connection broken by 'NewConnectionError('<pip._vendor.urllib3.connection.HTTPSConnection object at 0x7280c4bacb90>: Failed to establish a new connection: [Errno -3] Temporary failure in name resolution')': /simple/kombu/
WARNING: Retrying (Retry(total=2, connect=None, read=None, redirect=None, status=None)) after connection broken by 'NewConnectionError('<pip._vendor.urllib3.connection.HTTPSConnection object at 0x7280c4c2df70>: Failed to establish a new connection: [Errno -3] Tempo
…[472 chars omitted]…
broken by 'NewConnectionError('<pip._vendor.urllib3.connection.HTTPSConnection object at 0x7280c4c2e330>: Failed to establish a new connection: [Errno -3] Temporary failure in name resolution')': /simple/kombu/
ERROR: Could not find a version that satisfies the requirement kombu==5.5.8 (from versions: none)
ERROR: No matching distribution found for kombu==5.5.8
bash: line 248: unzip: command not foundstep 18
$ cd /repo && sed -n 140,200p kombu/asynchronous/timer.py
output · 38 lines
last = tref._last_run
next = secs - (now - last) if last else secs
self.enter_after(next, tref, priority)
tref.fun = _reschedules
tref._last_run = None
return self.enter_after(secs, tref, priority)
def enter_at(self, entry, eta=None, priority=0, time=monotonic):
"""Enter function into the scheduler.
Arguments:26 more lines
---------
entry (~kombu.asynchronous.timer.Entry): Item to enter.
eta (datetime.datetime): Scheduled time.
priority (int): Unused.
"""
if eta is None:
eta = time()
if isinstance(eta, datetime):
try:
eta = to_timestamp(eta)
except Exception as exc:
if not self.handle_error(exc):
raise
return
return
…[707 chars omitted]…
__(self, min=min, nowfun=monotonic,
pop=heapq.heappop, push=heapq.heappush):
"""Iterate over schedule.
This iterator yields a tuple of ``(wait_seconds, entry)``,
where if entry is :const:`None` the caller should wait
for ``wait_seconds`` until it polls the schedule again.
"""
max_interval = self.max_interval
queue = self._queuestep 19
$ cd /repo && sed -n 200,240p kombu/asynchronous/timer.py
output · 40 lines
while 1:
if queue:
eventA = queue[0]
now, eta = nowfun(), eventA[0]
if now < eta:
yield min(eta - now, max_interval), None
else:
eventB = pop(queue)
if eventB is eventA:
entry = eventA[2]28 more lines
if not entry.canceled:
yield None, entry
continue
else:
push(queue, eventB)
else:
yield None, None
def clear(self):
self._queue[:] = [] # atomic, without creating a new list.
def cancel(self, tref):
tref.cancel()
def __len__(self):
return len(self._queue)
def __nonzero__(self):
return True
@property
def queue(self, _pop=heapq.heappop):
"""Snapshot of underlying datastructure."""
events = list(self._queue)
return [_pop(v) for v in [events] * len(events)]
@property
def schedule(self):step 20
$ cd /repo && python - <<'PY'
output · 2 lines
2 2 ok
step 21
$ cd /repo && python -m pytest t/unit/transport/test_redis.py -q 2>&1 | tail -5
output · 5 lines
FAILED t/unit/transport/test_redis.py::test_Channel::test_register_with_event_loop FAILED t/unit/transport/test_redis.py::test_Channel::test_register_with_event_loop__on_disconnect__loop_cleanup[fds0] FAILED t/unit/transport/test_redis.py::test_Channel::test_register_with_event_loop__on_disconnect__loop_cleanup[fds1] FAILED t/unit/transport/test_redis.py::test_Channel::test_configurable_health_check 4 failed, 131 passed in 5.69s
step 22
$ cd /repo && python -m pytest t/unit/transport/test_redis.py::test_Channel::test_register_with_event_loop -q 2>&1 | tail -30
output · 30 lines
add_reader = loop.add_reader
on_readable = self.on_readable
def _on_disconnect(connection):
if connection._sock:
loop.remove(connection._sock)
# must have started polling or this will break reconnection
if cycle.fds:
# stop polling in the event loop
try:
loop.on_tick.remove(on_poll_start)18 more lines
except KeyError:
pass
cycle._on_connection_disconnect = _on_disconnect
def on_poll_start():
cycle_poll_start()
[add_reader(fd, on_readable, fd) for fd in cycle.fds]
loop.on_tick.add(on_poll_start)
# cancel previously registered maintenance timers, if any, so that
# re-registering after a reconnect does not leave duplicates behind.
> for tref in getattr(cycle, '_maintenance_timers', None) or []:
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
E TypeError: 'Mock' object is not iterable
kombu/transport/redis.py:1457: TypeError
=========================== short test summary info ============================
FAILED t/unit/transport/test_redis.py::test_Channel::test_register_with_event_loop
1 failed in 0.87sstep 23
$ cd /repo && python - <<'PY'
output · 3 lines
........................................................................ [ 53%] ............................................................... [100%] 135 passed in 5.08s
step 24
$ cd /repo && python -m pytest t/unit -q 2>&1 | tail -3 && python - <<'PY'
output · 4 lines
ERROR t/unit/utils/test_json.py !!!!!!!!!!!!!!!!!!! Interrupted: 7 errors during collection !!!!!!!!!!!!!!!!!!!! 9 skipped, 7 errors in 5.13s all ok
step 25
$ cd /repo && python -m pytest t/unit -q 2>&1 | grep -E "^ERROR|error" | head; python -m pytest t/unit -q --ignore t/unit/utils/test_json.py 2>&1 | tail -3
output · 12 lines
ERROR t/unit/asynchronous/aws/sqs/test_connection.py ERROR t/unit/asynchronous/aws/test_aws.py ERROR t/unit/asynchronous/aws/test_connection.py ERROR t/unit/transport/SQS - ModuleNotFoundError: No module named 'botocore' ERROR t/unit/transport/test_azurestoragequeues.py ERROR t/unit/transport/test_gcpubsub.py ERROR t/unit/utils/test_json.py !!!!!!!!!!!!!!!!!!! Interrupted: 7 errors during collection !!!!!!!!!!!!!!!!!!!! 9 skipped, 7 errors in 3.29s ERROR t/unit/transport/test_gcpubsub.py !!!!!!!!!!!!!!!!!!! Interrupted: 6 errors during collection !!!!!!!!!!!!!!!!!!!! 9 skipped, 6 errors in 3.09s
step 26
$ cd /repo && python -m pytest t/unit -q --ignore t/unit/asynchronous/aws --ignore t/unit/transport/SQS --ignore t/unit/transport/test_azurestoragequeues.py --ignore t/unit/transport/test_gcpubsub.py --ignore t/unit/utils/test_json.py 2>&1 | tail -3
output · 3 lines
FAILED t/unit/transport/test_pyro.py::test_PyroTransport::test_driver_version 1 failed, 1014 passed, 189 skipped in 25.64s Restoring 1 unacknowledged message(s)
step 27
$ cd /repo && git stash -q && python -m pytest t/unit/transport/test_pyro.py -q 2>&1 | tail -2; git stash pop -q
output · 2 lines
FAILED t/unit/transport/test_pyro.py::test_PyroTransport::test_driver_version 1 failed, 1 passed, 3 skipped in 0.93s
step 28
Pre-existing failure, unrelated. Final diff review:
$ cd /repo && git diff
output · 35 lines
diff --git a/kombu/transport/redis.py b/kombu/transport/redis.py
index 322bab0..56a3a9d 100644
--- a/kombu/transport/redis.py
+++ b/kombu/transport/redis.py
@@ -571,10 +571,13 @@ class MultiChannelPoller:
def maybe_restore_messages(self):
for channel in self._channels:
if channel.active_queues:
- # only need to do this once, as they are not local to channel.
- return channel.qos.restore_visible(
- num=channel.unacked_restore_limit,
- )23 more lines
+ try:
+ channel.qos.restore_visible(
+ num=channel.unacked_restore_limit,
+ )
+ except channel.connection_errors:
+ # connection lost, ignore for this tick
+ pass
def maybe_check_subclient_health(self):
for channel in se
…[1940 chars omitted]…
_messages),
+ loop.call_repeatedly(
+ connection.client.transport_options.get(
+ 'health_check_interval',
+ DEFAULT_HEALTH_CHECK_INTERVAL
+ ),
+ cycle.maybe_check_subclient_health,
+ ),
+ ]
def on_readable(self, fileno):
"""Handle AIO event for one of our file descriptors."""step 29
$ done: {"notes": "Fixed kombu/transport/redis.py: (1) MultiChannelPoller.maybe_restore_messages now calls channel.qos.restore_visible once per channel with active queues (no early return) and swallows exceptions listed in channel.connection_errors; (2) maybe_check_subclient_health wraps the cached subclient's check_health() in the same cooutput · 3 lines
…[164 chars omitted]… ore registering new ones, preventing duplicate callbacks after reconnects. Reproducers and full unit suite pass (only pre-existing unrelated pyro/botocore failures)."} submitted.