Fix: Worker fails to reconnect after Redis failover (regression from #9986 and #7273)

Fixes #10096

Description

After upgrading to Celery 5.6.x, workers using Redis Sentinel as the broker fail to reconnect after a Redis failover. Running tasks appear permanently stuck in active state, no new tasks are consumed, and the worker becomes invisible to celery inspect ping / active_queues.

The root cause is a combination of two regressions introduced by #9986 and #7273 in AsynPool.flush() and asynloop(). This PR fixes all three bugs while preserving the intended behavior of both original PRs.

Bug 1 — hub.reset() removed from asynloop() error path (regression from #9986)

PR #9986 removed the hub.reset() call from asynloop() to keep timers alive during graceful shutdown. However, this also removed the reset on the error path (e.g., broker connection loss), which means stale file descriptors and callbacks from the old connection remain registered in the hub. When the consumer restarts and re-registers with the event loop, these stale descriptors cause EBADF errors or 100% CPU poll loops, preventing the worker from reconnecting.

Fix: Wrap the event loop in try/except Exception and call hub.reset() only on error. This preserves the #9986 fix because:

• On graceful shutdown, the loop exits normally (no exception), so hub.reset() is not called — timers keep firing while the pool drains, exactly as Fix: Broker heartbeats not sent during graceful shutdown #9986 intended.
• On error (connection loss, etc.), hub.reset() is called — stale fds and callbacks are cleaned up before the consumer restarts.
• WorkerShutdown / WorkerTerminate extend SystemExit (not Exception), so they pass through without triggering the reset.

Note: The issue reporter's patch used finally: (unconditional reset), which would re-break #9986 / #5998. Using except Exception: is the correct approach — it targets only error-driven reconnection while leaving graceful shutdown intact.

Bug 2 — Unaccepted jobs stuck in cache when synack=False (regression from #7273)

PR #7273 added logic in AsynPool.flush() to clear the _cache when there are no active writers. However, the cleanup of individual unaccepted jobs was only done when synack=True:

# Before (only handles synack=True)
if self.synack:
    for job in self._cache.values():
        if not job._accepted:
            job._cancel()

When synack=False (the default), unaccepted jobs were never removed from _cache. After a broker reconnection, these orphaned entries make the worker think it still has active tasks, blocking new task consumption (especially with worker_concurrency=1).

Fix: Iterate over all unaccepted jobs regardless of synack. When synack=True, call job._cancel() (sends NACK). When synack=False, call job.discard() (removes from cache; broker will redeliver). Uses tuple() to safely copy the dict during iteration.

Bug 3 — Infinite loop in flush() when worker process is dead (latent bug in #7273)

In the while self._active_writers loop of flush(), when a writer generator had already started but its target process was dead:

1. _flush_writer() was skipped (process not alive)
2. job.discard() was called
3. But the generator was never removed from _active_writers

This caused the while loop to spin forever, hanging the worker on connection loss.

Additionally, the original code had a shortcut if not self._active_writers: self._cache.clear() that aggressively cleared the entire cache, including jobs that were already accepted and actively executing in worker processes.

Fix:

• Always call self._active_writers.discard(gen) in the else branch, regardless of whether the process is alive or dead.
• Remove the self._cache.clear() shortcut — jobs should only be discarded individually based on their actual state.
• When a process is dead, explicitly discard the job from cache since it will never complete.

How these bugs interact (the #10096 scenario)

The issue reporter's environment: Redis Sentinel broker, worker_concurrency=1, task_acks_late=False.

1. A task is running when Redis fails over → connection to old master is lost.
2. asynloop() gets an exception from the broken connection.
3. Without Bug 1 fix: hub.reset() is never called → stale fds remain → consumer cannot cleanly re-register with the event loop → worker becomes invisible (inspect ping fails).
4. Without Bug 2 fix: The in-flight task completes, but its job entry remains in _cache → worker thinks it still has 1 active task → with concurrency=1, no new tasks are accepted.
5. Without Bug 3 fix: If flush() is called while a writer targets a dead process, the worker hangs in an infinite loop.

All three fixes are needed for full recovery. The reporter confirmed that versions prior to 5.6.1 only needed the asynpool.py fix (Bugs 2+3), and from 5.6.1 onwards the loops.py fix (Bug 1) became necessary as well — matching the timeline of #9986 being merged.

Changes

Test coverage

asynloop hub reset tests (t/unit/worker/test_loops.py)

AsynPool.flush() tests (t/unit/concurrency/test_prefork.py)

Backwards compatibility

• No user-facing API changes.
• No configuration changes.
• Graceful shutdown behavior from Fix: Broker heartbeats not sent during graceful shutdown #9986 is preserved (timers keep firing during pool drain).
• Memory leak fix from Only clear the cache if there are no active writers. #7273 is preserved (jobs are still discarded from cache on flush).
• The only behavioral change is that hub.reset() is now called on connection errors (restoring pre-Fix: Broker heartbeats not sent during graceful shutdown #9986 behavior for the error path only), and unaccepted jobs are properly cleaned up regardless of synack setting.