Re-arm the flush timer after a broker reconnect

[ShuaiShao93]

Summary

Fixes the long-standing "RMQ reconnection" wedge (#99): after the broker connection is lost and re-established, a Batches worker can stop flushing batches entirely and sit holding its prefetched messages unacked until it is restarted. I dug into the root cause and have a one-spot fix in Batches.Strategy(). Full analysis (with a minimal repro) is in #107.

Root cause

Strategy() runs every time the consumer (re)starts, including after a reconnect. The flush timer is armed lazily inside the message handler:

if self._tref is None:  # first request starts flush timer.
    self._tref = timer.call_repeatedly(self.flush_interval, flush_buffer)

self._tref lives on the task instance, which survives reconnects, but the timer it points at is bound to the consumer's old event-loop hub. After a reconnect Celery rebuilds the consumer (new hub, new channel) and calls Strategy() again, but self._tref is still set, so the if self._tref is None guard never re-arms the timer on the new hub.

The interval-based flush is then dead. The only remaining flush path is the count-based one (flush_every). With the common broker setup worker_prefetch_multiplier=1 (so the broker delivers fewer than flush_every messages and waits for acks), that count threshold is never reached either, so the worker buffers its prefetched messages forever without acking and looks "connected but idle". This matches the symptoms reported in #99.

Fix

Reset the per-consumer state (timer, buffers, count) at the start of Strategy(), before the new message handler closes over self._buffer. The next message then re-arms the flush timer on the new hub via the existing guard. Buffered requests belong to the old connection and are redelivered by the broker, so they are dropped.

Because Strategy() only runs at consumer (re)start — not per message — there is no steady-state overhead, and on a normal first start the reset is a no-op (the buffers are already empty and _tref is None).

Testing

• Added t/unit/test_strategy.py covering both the reconnect reset (stale timer cancelled, _tref cleared so it re-arms, buffers/count reset) and the first-start no-op path.
• pytest, mypy pass locally.
• Verified end-to-end on RabbitMQ: a Batches worker with worker_prefetch_multiplier=1 wedges on main after a broker restart and recovers on its own with this change, including under sustained memory/disk alarms.

Closes #99

[clokep]

Looks like GitHub Actions is having a day again. 😞

[clokep]

Thank you!