celery-kombu-2498-2492
Results
The prompt the agent sees
When a Celery worker uses the Redis transport and the Redis connection is interrupted, it can behave incorrectly in several ways:
- A connection error raised while the worker’s periodic Redis maintenance callbacks run escapes into the event loop. The worker’s event loop then stops and is repeatedly restarted instead of continuing to operate normally. - Reconnecting repeatedly registers additional maintenance timers without retiring the timers from earlier connections. The same maintenance work is consequently triggered multiple times for each tick, causing timer and callback activity to grow after every reconnect. - The disconnect handler removes the polling callback that the transport registered on the event loop’s per-tick hook, so a stale connection’s disconnect can unregister the callback a newer connection relies on. The worker remains alive and may respond to health checks, but nothing triggers polling of Redis for tasks any more; newly queued tasks remain unconsumed until the worker is restarted. A connection loss must never unregister the polling callback from the tick hook, whether or not any file descriptors are active. - Disconnect cleanup is unreliable for sockets that have already lost their file descriptor, have no socket object, or belong to a subclient. Stale polling entries and disconnected channel state can remain registered, leading to invalid resources being retained or processed after the connection is gone.
The worker should tolerate transient Redis connection errors during periodic maintenance, cleanly discard disconnected polling resources, and continue consuming tasks after reconnection without accumulating duplicate timer activity or losing its polling trigger.
Contract for cleaning up after a dropped connection, as observed through the poller’s disconnect handling: if the dropped connection has a socket object or a raw integer descriptor, that same object is removed from the event loop exactly once. Registrations for the connection are removed only when a valid descriptor is known: `fileno()` returning 0 or more, or the raw integer itself. If `fileno()` returns `-1` or raises, no descriptor registration is removed, because that descriptor may already belong to another socket. If the connection has no socket, registrations associated with channels whose `client` or `subclient` is backed by the dropped connection are removed, while registrations belonging to other connections remain. The channels expose `client` and `subclient` as ordinary attributes and may be lightweight objects without a populated instance dictionary. Cleanup must not raise, must work whether or not the loop currently tracks descriptors, and must leave the per-tick polling callback registered on the loop. The poller object the transport is registered with may support nothing beyond its descriptor set, its descriptor-to-channel mapping and its callbacks: the transport's own disconnect handling must perform the loop removal and the pruning itself, must not read poller attributes that nothing has set, and must not rely on a disconnect method of the poller doing the work (it may be inert); any per-registration state such as timer handles has to be kept by the transport's own code.
Contract for the periodic maintenance callbacks: `maybe_restore_messages()` processes each channel with active queues by calling `channel.qos.restore_visible(num=channel.unacked_restore_limit)` exactly once. Channels without active queues are ignored. `maybe_check_subclient_health()` calls `check_health()` exactly once for each channel with a cached `subclient`; channels without one are silently skipped. If either operation fails, the failure is swallowed for that tick only when it is an instance of one of the classes listed in that channel’s `connection_errors` attribute. That attribute may include classes that are not redis-py errors, so a fixed exception class is not sufficient.
Re-registering with the event loop after a reconnect cancels all timer entries created by the previous `register_with_event_loop()` call before registering new entries. The new registration’s timer entries remain live, and repeated reconnects must not accumulate duplicate timer activity.
How the tests reach the cleanup: they replace `transport.cycle` with a `Mock` (spec'd to `fds`, `_fd_to_chan`, `on_poll_init`, `on_poll_start`, `maybe_restore_messages`, `maybe_check_subclient_health` and `_on_connection_disconnect`) BEFORE calling `redis.Transport.register_with_event_loop(transport, conn, loop)`, then call `transport.cycle._on_connection_disconnect(connection)` and assert on `loop.remove`, `cycle._fd_to_chan` and `loop.on_tick`. So registering with the event loop must assign the disconnect cleanup callable to the poller's `_on_connection_disconnect` attribute, and that callable must act on the `loop` object that was passed to `register_with_event_loop`; cleanup logic that lives only as a method of the real poller class is never executed.
Hidden tests · 9 fail-to-pass, 141 pass-to-passrun after the agent submits, in a clean verifier
Test patch · 420 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')
diff --git a/t/unit/transport/test_redis.py b/t/unit/transport/test_redis.py
index 11cfa4b9a4..16c3a183ea 100644
--- a/t/unit/transport/test_redis.py
+++ b/t/unit/transport/test_redis.py
@@ -1203,11 +1203,19 @@ def test_register_with_event_loop(self):
@pytest.mark.parametrize('fds', [{12: 'LISTEN', 13: 'BRPOP'}, {}])
def test_register_with_event_loop__on_disconnect__loop_cleanup(self, fds):
- """Ensure event loop polling stops on disconnect (if started)."""
+ """Ensure on_poll_start stays in on_tick after disconnect.
+
+ on_poll_start is idempotent (no-op when there are no fds), so it
+ must NOT be removed on disconnect. Removing it caused a race
+ condition where a late-firing _on_disconnect from a stale channel
+ would remove the callback just registered by a new channel,
+ leaving the worker unable to consume tasks after reconnection.
+ """
transport = self.connection.transport
self.connection._sock = None
transport.cycle = Mock(name='cycle')
transport.cycle.fds = fds
+ transport.cycle._fd_to_chan = {}
conn = Mock(name='conn')
conn.client = Mock(name='client', transport_options={})
loop = Mock(name='loop')
@@ -1215,11 +1223,229 @@ def test_register_with_event_loop__on_disconnect__loop_cleanup(self, fds):
redis.Transport.register_with_event_loop(transport, conn, loop)
assert len(loop.on_tick) == 1
transport.cycle._on_connection_disconnect(self.connection)
- if fds:
- assert len(loop.on_tick) == 0
- else:
- # on_tick shouldn't
… [12050 more characters]Reference fix · 1 file, +87 −12the 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
)
diff --git a/kombu/transport/redis.py b/kombu/transport/redis.py
index 68153ef827..5ebd4d3085 100644
--- a/kombu/transport/redis.py
+++ b/kombu/transport/redis.py
@@ -1447,14 +1447,61 @@ def register_with_event_loop(self, connection, loop):
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
+ # Prune the disconnected file descriptor from cycle._fd_to_chan
+ # so that the next on_poll_start tick does not re-register a
+ # stale/disconnected socket. fileno() returns -1 on a socket
+ # that has been closed (but not yet garbage-collected), so we
+ # only prune when we get a valid (>= 0) file descriptor.
+ sock = connection._sock
+ fd = None
try:
- loop.on_tick.remove(on_poll_start)
- except KeyError:
+ if hasattr(sock, "fileno"):
+ raw_fd = sock.fileno()
+ # fileno() returns -1 for a closed-but-not-GC'd socket;
+ # in that case there is no valid fd to prune.
+ if raw_fd >= 0:
+ fd = raw_fd
+ else:
+ # Plain integer file descriptor (no fileno() method).
+ fd = sock
+ except OSError:
+ # Socket already closed at OS level; nothing to prune.
pass
+ if fd is not None:
+ try:
+ del cycle._fd_to_chan[fd]
+ except KeyError:
+ # fd was never tracked or already pruned — safe to ignore.
+ pass
+ else:
+ # In async Redis mode, Connection.disconnect() may have already
+ # cleared connection._sock (set to None) before invoking this
+ # callback. In that case we can no longer derive the fd from the
+ # socket itself, so we conservatively scan cycle._fd_to_chan for
+ # channels that are backed by this connection and prune them.
+ stale_fds = []
+ for fd, (chan, _type) in list(cycle._fd_to_chan.items()):
+ client = getattr(chan, "client", None)
+ subclient = getattr(chan, "subclient", None)
+ client_conn = getattr(client, "connection", None)
+ subclient_conn = getattr(subclient, "connection", None)
+ if client_conn is connection or subclient_conn is connection:
+ stale_fds.append(fd)
+ for fd in stale_fds:
+ try:
+ del cycle._fd_to_chan[fd]
+ except KeyError:
+ # fd was never tracked or already pruned — safe to ignore.
+ pass
+ # Note: we intentionally do NOT remove on_poll_start from
+ # loop.on_tick here. on_poll_start is idempotent — when there
+ # are no active file descriptors it simply does nothing.
+ # Removing it caused a race condition where a late-firing
+ # _on_disconnect from a stale channel would remove the
+ # on_poll_start callback that a newly-reconnected channel had
+ # just registered, leaving the worker alive but unable to
+ # consume any tasks ("catatonic worker" after broker restart).
+ # See: https://github.com/celery/celery/issues/8030
cycle._on_connection_disconnect = _on_disconnect
def on_poll_start():
GLM-5.3 Flash · attempt 2 · not solved · 28 steps · $0.012
Submitted diff · 211 lines
diff --git a/kombu/transport/redis.py b/kombu/transport/redis.py
index 322bab0..a72b419 100644
--- a/kombu/transport/redis.py
+++ b/kombu/transport/redis.py
@@ -568,21 +568,41 @@ class MultiChannelPoller:
num=channel.unacked_restore_limit,
)
+ def _swallow_connection_error(self, channel, exc):
+ """Swallow *exc* if it's a connection error for *channel*.
+
+ Transient redis connection errors are tolerated during periodic
+ maintenance, any other error is propagated to the caller.
+ """
+ errors = getattr(channel, 'connection_errors', None)
+ if errors is not None and isinstance(exc, errors):
+ return
+ raise exc
+
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 Exception as exc:
+ self._swallow_connection_error(channel, exc)
def maybe_check_subclient_health(self):
for channel in self._channels:
# only if subclient property is cached
- client = channel.__dict__.get('subclient')
+ try:
+ client = channel.__dict__.get('subclient')
+ except AttributeError:
+ # channel may not have an instance dictionary at all.
+ continue
if client is not None \
and callable(getattr(client, 'check_health', None)):
- client.check_health()
+ try:
+ client.check_health()
+ except Exception as exc:
+ self._swallow_connection_error(channel, exc)
def on_readable(self, fileno):
chan, type = self._fd_to_chan[fileno]
@@ -1414,6 +1434,9 @@ class Transport(virtual.Transport):
# All channels share the same poller.
self.cycle = MultiChannelPoller()
+ # Timer entries created by register_with_event_loop(), so that
+ # they can be cancelled when registering with a (new) event loop.
+ self._event_loop_timer_entries = []
# 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
@@ -1421,7 +1444,75 @@ class Transport(virtual.Transport):
def driver_version(self):
return redis.__version__
+ def _cancel_event_loop_timers(self):
+ """Cancel timer entries registered by a previous registration."""
+ for entry in self._event_loop_timer_entries:
+ try:
+ entry.cancel()
+ except Exception: # pragma: no cover
+ pass
+ self._event_loop_timer_entries = []
+
+ def _on_disconnect_cleanup(self, loop, connection):
+ """Clean up after a dropped connection.
+
+ Removes the resources belonging to *connection* from the event
+ loop *loop*, without ever unregistering the per-tick polling
+ callback (a reconnecting worker must keep polling redis).
+ """
+ cycle = self.cycle
+ sock = getattr(connection, '_sock', None)
+
+ # The socket (or raw file descriptor) itself is removed from the
+ # event loop exactly once, whether or not the loop currently
+ # tracks it.
+ if sock is not None:
+ try:
+ loop.remove(sock)
+ except Exception:
+ pass
+
+ # Figure out the descriptor of the connection, if it's still
+ # valid. A descriptor of -1 (or a failing fileno()) may already
+ # have been reused by another socket, so in that case no
+ # registration is removed at all.
+ fd = None
+ if isinstance(sock, int):
+ fd = sock
+ elif sock is not None:
+ try:
+ fileno = sock.fileno()
+ except Exception:
+ fileno = None
+ if isinstance(fileno, int) and fileno >= 0:
+ fd = fileno
+
+ try:
+ fd_to_chan = cycle._fd_to_chan
+ if fd is not None:
+ # remove the registration belonging to this connection.
+ fd_to_chan.pop(fd, None)
+ else:
+ # Without a valid descriptor we identify the
+ # registrations by the channels whose client or
+ # subclient is backed by the dropped connection.
+ for chan_fd, (channel, type) in list(fd_to_chan.items()):
+ if type == 'BRPOP':
+ client = getattr(channel, 'client', None)
+ else:
+ client = getattr(channel, 'subclient', None)
+ if client is not None and getattr(
+ client, 'connection', None) is connection:
+ fd_to_chan.pop(chan_fd, None)
+ except Exception: # pragma: no cover
+ # cleanup must never raise.
+ pass
+
def register_with_event_loop(self, connection, loop):
+ # cancel timers registered by a previous registration, so that
+ # reconnecting does not accumulate duplicate timer activity.
+ self._cancel_event_loop_timers()
+
cycle = self.cycle
cycle.on_poll_init(loop.poller)
cycle_poll_start = cycle.on_poll_start
@@ -1429,31 +1520,26 @@ class Transport(virtual.Transport):
on_readable = self.on_readable
def _on_disconnect(connection):
- if connection._sock:
- loop.remove(connection._sock)
-
… [3004 more characters]step 1
$ ls /repo && ls /repo/kombu/transport | head -30
output · 44 lines
AUTHORS Changelog.rst FAQ INSTALL LICENSE MANIFEST.in Makefile README.rst SECURITY.md THANKS TODO conftest.py
32 more lines
docs examples kombu kombu.egg-info pyproject.toml requirements setup.cfg setup.py t tox.ini SLMQ.py SQS __init__.py azureservicebus.py azurestoragequeues.py base.py confluentkafka.py consul.py etcd.py filesystem.py gcpubsub.py librabbitmq.py memory.py mongodb.py native_delayed_delivery.py pyamqp.py pyro.py qpid.py redis.py sqlalchemy virtual zookeeper.py
step 2
$ grep -n "register_with_event_loop\|maybe_restore_messages\|maybe_check_subclient_health\|_on_connection_disconnect\|on_tick" /repo/kombu/transport/redis.py
output · 13 lines
511: def _on_connection_disconnect(self, connection): 571: def maybe_restore_messages(self): 579: def maybe_check_subclient_health(self): 617: self.maybe_restore_messages() 800: def _on_connection_disconnect(self, connection): 806: self.connection.cycle._on_connection_disconnect(connection) 1314: channel._on_connection_disconnect(self) 1424: def register_with_event_loop(self, connection, loop): 1439: loop.on_tick.remove(on_poll_start) 1442: cycle._on_connection_disconnect = _on_disconnect 1447: loop.on_tick.add(on_poll_start) 1448: loop.call_repeatedly(10, cycle.maybe_restore_messages)
1 more lines
1455: cycle.maybe_check_subclient_health
step 3
$ sed -n 480,640p /repo/kombu/transport/redis.py; echo ----; sed -n 780,830p /repo/kombu/transport/redis.py; echo ----; sed -n 1400,1470p /repo/kombu/transport/redis.py
output · 43 lines
#: Set of one-shot callbacks to call after reading from socket.
after_read = None
def __init__(self):
# active channels
self._channels = set()
# file descriptor -> channel map.
self._fd_to_chan = {}
# channel -> socket map
self._chan_to_sock = {}
# poll implementation (epoll/kqueue/select)
self.poller = poll()31 more lines
# one-shot callbacks called after reading from socket.
self.after_read = set()
def close(self):
for fd in self._chan_to_sock.values():
try:
self.poller.unregister(fd)
except (KeyError, ValueError):
pass
self._channels.clear()
self._fd_to_chan.clear()
self._chan_to_sock.clear()
def add(self, channel):
self._channels.add(channel)
def discard(self, channel):
self._channels.
…[8628 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 4
$ grep -n "register_with_event_loop\|_on_connection_disconnect\|on_tick" /repo/t/unit/transport/test_redis.py | head -40
output · 17 lines
427: def _on_connection_disconnect(self, connection): 486: def _on_connection_disconnect(self, connection): 1181: def test_register_with_event_loop(self): 1188: redis.Transport.register_with_event_loop(transport, conn, loop) 1194: loop.on_tick.add.assert_called() 1195: on_poll_start = loop.on_tick.add.call_args[0][0] 1205: def test_register_with_event_loop__on_disconnect__loop_cleanup(self, fds): 1214: loop.on_tick = set() 1215: redis.Transport.register_with_event_loop(transport, conn, loop) 1216: assert len(loop.on_tick) == 1 1217: transport.cycle._on_connection_disconnect(self.connection) 1219: assert len(loop.on_tick) == 0
5 more lines
1221: # on_tick shouldn't be cleared when polling hasn't started 1222: assert len(loop.on_tick) == 1 1233: redis.Transport.register_with_event_loop(transport, conn, loop) 1239: loop.on_tick.add.assert_called() 1240: on_poll_start = loop.on_tick.add.call_args[0][0]
step 5
$ sed -n 1170,1260p /repo/t/unit/transport/test_redis.py
output · 33 lines
def test_get_response_error(self):
from redis.exceptions import ResponseError
assert redis.Channel._get_response_error(self.channel) is ResponseError
def test_avail_client(self):
self.channel._pool = Mock()
cc = self.channel._create_client = Mock()
with self.channel.conn_or_acquire():
pass
cc.assert_called_with()
def test_register_with_event_loop(self):21 more lines
transport = self.connection.transport
transport.cycle = Mock(name='cycle')
transport.cycle.fds = {12: 'LISTEN', 13: 'BRPOP'}
conn = Mock(name='conn')
conn.client = Mock(name='client', transport_options={})
loop = Mock(name='loop')
redis.Transport.register_with_event_loop(transport, conn, loop)
transport.cycle.on_poll_init.assert_called_with(loop.poller)
loop.call_repeatedly.assert_has_calls([
…[2460 chars omitted]…
self):
transport = self.connection.transport
cycle = transport.cycle = Mock(name='cyle')
cycle.on_readable.return_value = None
redis.Transport.on_readable(transport, 13)
cycle.on_readable.assert_called_with(13)
def test_transport_connection_errors(self):
"""Ensure connection_errors are populated."""
assert redis.Transport.connection_errorsstep 6
$ grep -n "def call_repeatedly\|def call_at\|def call_later" -A15 /repo/kombu/asynchronous/hub.py; grep -n "def remove\b" -A20 /repo/kombu/asynchronous/hub.py
output · 42 lines
211: def call_later(self, delay, callback, *args): 212- return self.timer.call_after(delay, callback, args) 213- 214: def call_at(self, when, callback, *args): 215- return self.timer.call_at(when, callback, args) 216- 217: def call_repeatedly(self, delay, callback, *args): 218- return self.timer.call_repeatedly(delay, callback, args) 219- 220- def add_reader(self, fds, callback, *args): 221- return self.add(fds, callback, READ | ERR, args) 222-
30 more lines
223- def add_writer(self, fds, callback, *args): 224- return self.add(fds, callback, WRITE, args) 225- 226- def remove_reader(self, fd): 227- writable = fd in self.writers 228- on_write = self.writers.get(fd) 229- try: 230- self._remove_from_loop(fd) 231- finally: 232- if writable: 183: def remove(self, fd): 184- fd = fileno(fd) 185- …[83 chars omitted]… self._running = True 189- try: 190- while 1: 191- try: 192- self.run_once() 193- except Stop: 194- break 195- finally: 196- self._running = False 197- 198- def run_once(self): 199- try: 200- next(self.loop) 201- except StopIteration: 202- self._loop = None 203-
step 7
$ grep -n "subclient" /repo/kombu/transport/redis.py | head -30; grep -rn "maybe_restore_messages\|maybe_check_subclient_health\|_on_connection_disconnect" /repo/t/unit/transport/test_redis.py | head
output · 20 lines
550: if not self._client_registered(channel, channel.subclient, 'LISTEN'):
552: self._register(channel, channel.subclient, 'LISTEN')
579: def maybe_check_subclient_health(self):
581: # only if subclient property is cached
582: client = channel.__dict__.get('subclient')
640: _subclient = None
911: c = self.subclient
919: c = self.subclient
940: c = self.subclient
1015: self.subclient.parse_response()
1190: for attr in 'client', 'subclient':
1370: def subclient(self):8 more lines
1455: cycle.maybe_check_subclient_health 427: def _on_connection_disconnect(self, connection): 486: def _on_connection_disconnect(self, connection): 1191: call(10, transport.cycle.maybe_restore_messages), 1192: call(25, transport.cycle.maybe_check_subclient_health), 1217: transport.cycle._on_connection_disconnect(self.connection) 1236: call(10, transport.cycle.maybe_restore_messages), 1237: call(15, transport.cycle.maybe_check_subclient_health),
step 8
$ sed -n 1360,1380p /repo/kombu/transport/redis.py; sed -n 1180,1200p /repo/kombu/transport/redis.py
output · 38 lines
if self._async_pool is None:
self._async_pool = self._get_pool(asynchronous=True)
return self._async_pool
@cached_property
def client(self):
"""Client used to publish messages, BRPOP etc."""
return self._create_client(asynchronous=True)
@cached_property
def subclient(self):
"""Pub/Sub connection used to consume fanout queues."""26 more lines
client = self._create_client(asynchronous=True)
return client.pubsub()
def _update_queue_cycle(self):
self._queue_cycle.update(self.active_queues)
def _get_response_error(self):
from redis import exceptions
return exceptions.ResponseError
if client is not None:
for queue in self._fanout_queues:
if queue in self.auto_delete_queues:
self.queue_delete(queue, client=client)
…[192 chars omitted]…
t':
try:
client = self.__dict__[attr]
connection, client.connection = client.connection, None
connection.disconnect()
except (KeyError, AttributeError, self.ResponseError):
pass
def _prepare_virtual_host(self, vhost):
if not isinstance(vhost, numbers.Integral):
if not vhost or vhost == '/':step 9
Now let me implement the fix:
$ cd /repo && python - <<'PY'
output · 37 lines
self.poller = poller
for channel in self._channels:
return channel.qos.restore_visible(
num=channel.unacked_restore_limit,
)
def _swallow_connection_error(self, channel, exc):
"""Swallow *exc* if it's a connection error for *channel*.
Transient redis connection errors are tolerated during periodic
maintenance, any other error is propagated to the caller.
"""25 more lines
errors = getattr(channel, 'connection_errors', None)
if errors is not None and isinstance(exc, errors):
return
raise exc
def maybe_restore_messages(self):
for channel in self._channels:
if channel.active_queues:
try:
channel.qos.restore_visible(
num=channel.unacked_restore_limit,
)
ex
…[453 chars omitted]…
not None \
and callable(getattr(client, 'check_health', None)):
try:
client.check_health()
except Exception as exc:
self._swallow_connection_error(channel, exc)
def on_readable(self, fileno):
chan, type = self._fd_to_chan[fileno]
if chan.qos.can_consume():
chan.handlers[type]()step 10
Now the transport's `register_with_event_loop` and disconnect cleanup:
$ cd /repo && python - <<'PY'
step 11
Now write a reproducer based on the issue's contract:
$ cd /repo && cat > /tmp/repro.py <<'PY'
output · 5 lines
Traceback (most recent call last):
File "/tmp/repro.py", line 69, in <module>
assert set(t.cycle._fd_to_chan) == {3}
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
AssertionErrorstep 12
$ cd /repo && sed -i 's/assert set(t.cycle._fd_to_chan) == {3}/assert set(t.cycle._fd_to_chan) == {2, 3}/' /tmp/repro.py && python /tmp/repro.pyoutput · 5 lines
Traceback (most recent call last):
File "/tmp/repro.py", line 105, in <module>
redis.Transport.maybe_restore_messages(t)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
AttributeError: type object 'Transport' has no attribute 'maybe_restore_messages'step 13
$ cd /repo && python - <<'PY'
output · 13 lines
Traceback (most recent call last):
File "/tmp/repro.py", line 123, in <module>
ch.connection_errors = ()
^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.12/unittest/mock.py", line 773, in __setattr__
elif (self._spec_set and self._mock_methods is not None and
^^^^^^^^^^^^^^
File "/usr/local/lib/python3.12/unittest/mock.py", line 658, in __getattr__
elif self._mock_methods is not None:
^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.12/unittest/mock.py", line 657, in __getattr__
raise AttributeError(name)1 more lines
AttributeError: _mock_methods
step 14
$ cd /repo && python - <<'PY'
output · 13 lines
Traceback (most recent call last):
File "/tmp/repro.py", line 123, in <module>
ch.connection_errors = ()
^^^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.12/unittest/mock.py", line 773, in __setattr__
elif (self._spec_set and self._mock_methods is not None and
^^^^^^^^^^^^^^
File "/usr/local/lib/python3.12/unittest/mock.py", line 658, in __getattr__
elif self._mock_methods is not None:
^^^^^^^^^^^^^^^^^^
File "/usr/local/lib/python3.12/unittest/mock.py", line 657, in __getattr__
raise AttributeError(name)1 more lines
AttributeError: _mock_methods
step 15
$ cd /repo && python - <<'PY'
output · 1 lines
ALL OK
step 16
$ cd /repo && python -m pytest t/unit/transport/test_redis.py -x -q 2>&1 | tail -20
output · 20 lines
fileno = None
if isinstance(fileno, int) and fileno >= 0:
fd = fileno
if fd is not None:
# remove registrations belonging to this connection.
cycle._fd_to_chan.pop(fd, None)
else:
# Without a valid descriptor we identify the registrations
# by the channels whose client or subclient is backed by the
# dropped connection.
> for chan_fd, (channel, type) in list(cycle._fd_to_chan.items()):8 more lines
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ E TypeError: 'Mock' object is not iterable kombu/transport/redis.py:1497: TypeError =========================== short test summary info ============================ FAILED t/unit/transport/test_redis.py::test_Channel::test_register_with_event_loop__on_disconnect__loop_cleanup[fds0] !!!!!!!!!!!!!!!!!!!!!!!!!! stopping after 1 failures !!!!!!!!!!!!!!!!!!!!!!!!!!! 1 failed, 62 passed in 1.37s
step 17
$ cd /repo && python - <<'PY'
output · 6 lines
ALL OK t/unit/transport/test_redis.py:1219: AssertionError =========================== short test summary info ============================ FAILED t/unit/transport/test_redis.py::test_Channel::test_register_with_event_loop__on_disconnect__loop_cleanup[fds0] 1 failed, 134 passed in 5.41s
step 18
$ cd /repo && python - <<'PY'
output · 3 lines
........................................................................ [ 53%] ............................................................... [100%] 135 passed in 5.35s
step 19
$ cd /repo && python -m pytest t/unit -q 2>&1 | tail -3; git diff --stat
output · 6 lines
ERROR t/unit/utils/test_json.py !!!!!!!!!!!!!!!!!!! Interrupted: 7 errors during collection !!!!!!!!!!!!!!!!!!!! 9 skipped, 7 errors in 6.52s kombu/transport/redis.py | 124 ++++++++++++++++++++++++++++++++++------- t/unit/transport/test_redis.py | 11 ++-- 2 files changed, 110 insertions(+), 25 deletions(-)
step 20
$ cd /repo && python -m pytest t/unit -q 2>&1 | grep -m3 Error; python -m pytest t/unit/transport t/unit/asynchronous -q 2>&1 | tail -3
output · 6 lines
ImportError while importing test module '/repo/t/unit/asynchronous/aws/sqs/test_connection.py'. E ModuleNotFoundError: No module named 'botocore' ImportError while importing test module '/repo/t/unit/asynchronous/aws/test_aws.py'. ERROR t/unit/asynchronous/aws/test_connection.py !!!!!!!!!!!!!!!!!!! Interrupted: 6 errors during collection !!!!!!!!!!!!!!!!!!!! 9 skipped, 6 errors in 3.06s
step 21
$ cd /repo && python -m pytest t/unit/transport t/unit/asynchronous/hub.py t/unit/asynchronous/timer.py -q 2>&1 | tail -3
output · 1 lines
no tests ran in 0.01s
step 22
$ cd /repo && python -m pytest t/unit/transport -q 2>&1 | tail -3 && python -m pytest t/unit/asynchronous --ignore=t/unit/asynchronous/aws -q 2>&1 | tail -3
output · 6 lines
ERROR t/unit/transport/test_gcpubsub.py !!!!!!!!!!!!!!!!!!! Interrupted: 3 errors during collection !!!!!!!!!!!!!!!!!!!! 7 skipped, 3 errors in 1.81s ............s........................................................... [ 87%] .......... [100%] 81 passed, 2 skipped in 1.38s
step 23
$ cd /repo && python -m pytest t/unit/transport --ignore=t/unit/transport/test_gcpubsub.py -q 2>&1 | tail -3
output · 3 lines
ERROR t/unit/transport/test_azurestoragequeues.py !!!!!!!!!!!!!!!!!!! Interrupted: 2 errors during collection !!!!!!!!!!!!!!!!!!!! 7 skipped, 2 errors in 1.64s
step 24
$ cd /repo && python -m pytest t/unit/transport --ignore=t/unit/transport/test_gcpubsub.py --ignore=t/unit/transport/test_azurestoragequeues.py -q 2>&1 | tail -3
output · 3 lines
ERROR t/unit/transport/SQS - ModuleNotFoundError: No module named 'botocore' !!!!!!!!!!!!!!!!!!!! Interrupted: 1 error during collection !!!!!!!!!!!!!!!!!!!! 7 skipped, 1 error in 1.77s
step 25
$ cd /repo && python -m pytest t/unit/transport --ignore=t/unit/transport/test_gcpubsub.py --ignore=t/unit/transport/test_azurestoragequeues.py --ignore=t/unit/transport/SQS -q 2>&1 | tail -3
output · 3 lines
FAILED t/unit/transport/test_pyro.py::test_PyroTransport::test_driver_version 1 failed, 314 passed, 174 skipped in 8.58s Restoring 1 unacknowledged message(s)
step 26
$ 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 1.06s
step 27
$ cd /repo && git diff kombu/transport/redis.py
output · 37 lines
diff --git a/kombu/transport/redis.py b/kombu/transport/redis.py
index 322bab0..a72b419 100644
--- a/kombu/transport/redis.py
+++ b/kombu/transport/redis.py
@@ -568,21 +568,41 @@ class MultiChannelPoller:
num=channel.unacked_restore_limit,
)
+ def _swallow_connection_error(self, channel, exc):
+ """Swallow *exc* if it's a connection error for *channel*.
+
+ Transient redis connection errors are tolerated during periodic25 more lines
+ maintenance, any other error is propagated to the caller.
+ """
+ errors = getattr(channel, 'connection_errors', None)
+ if errors is not None and isinstance(exc, errors):
+ return
+ raise exc
+
def maybe_restore_messages(self):
for channel in self._channels:
if channel.active_queues:
- # only need to do this once, as they are not l
…[6130 chars omitted]…
t(
'health_check_interval',
DEFAULT_HEALTH_CHECK_INTERVAL
)
- loop.call_repeatedly(
+ self._event_loop_timer_entries.append(loop.call_repeatedly(
health_check_interval,
cycle.maybe_check_subclient_health
- )
+ ))
def on_readable(self, fileno):
"""Handle AIO event for one of our file descriptors."""step 28
All checks pass. Summary of the fix in `kombu/transport/redis.py`:
$ done: {"notes": "Fixed redis transport disconnect handling: transport-owned disconnect cleanup (loop removal + _fd_to_chan pruning by fd or by channel client/subclient ownership, never removing the on_tick polling callback), connection-error-tolerant periodic maintenance callbacks (restore_visible for all channels with active queues, check_health for all cached subclients, swallowing only channel.connection_errors), and cancellation of previous event-loop timer entries on re-registration."}output · 1 lines
submitted.