fix: clear the timer while catch the exception (#10218)

* fix: clear the timer while catch the exception

* fix: move the timer clear to another try catch block

* Update t/unit/worker/test_loops.py

---------

Co-authored-by: Asif Saif Uddin {"Auvi":"অভি"} <auvipy@gmail.com>

Author: 2 people

celery/worker/loops.py

@@ -100,6 +100,12 @@ def asynloop(obj, connection, consumer, blueprint, hub, qos,
     except Exception:
         # Reset the hub on error (e.g. connection loss) to clean up
         # stale file descriptors and callbacks from the old connection.
+        # Also clear the timer queue so that stale periodic entries added by
+        # register_with_event_loop (e.g. maybe_restore_messages) do not fire
+        # against the broken connection after reconnect and trigger another
+        # crash before the new connection is fully established.
+        # All hub timers are re-registered during blueprint.start() once this
+        # exception propagates and the consumer reconnects.
         # We intentionally do NOT reset on normal exit (graceful shutdown)
         # so that timers (e.g. heartbeat) keep firing while the pool drains.
         # WorkerShutdown/WorkerTerminate extend SystemExit (not Exception)
@@ -109,6 +115,18 @@ def asynloop(obj, connection, consumer, blueprint, hub, qos,
         except Exception as exc:  # pylint: disable=broad-except
             logger.exception(
                 'Error cleaning up after event loop: %r', exc)
+        # Clear stale timer entries accumulated across reconnects (e.g.
+        # maybe_restore_messages registered via call_repeatedly). Without
+        # this, each reconnect appends a new entry; all of them fire during
+        # the reconnect window, raise again, and trigger another restart.
+        # Use a separate try/except so this always runs even if hub.reset()
+        # raised above. Timers are re-registered by register_with_event_loop
+        # when blueprint.start() is called after reconnect.
+        try:
+            hub.timer.clear()
+        except Exception as exc:  # pylint: disable=broad-except
+            logger.exception(
+                'Error clearing hub timer after event loop: %r', exc)
         raise
 
 

t/unit/worker/test_loops.py

@@ -466,6 +466,87 @@ def test_hub_reset_on_connection_error(self):
             asynloop(*x.args)
         x.hub.reset.assert_called_once()
 
+    def test_hub_timer_cleared_on_connection_error(self):
+        # Stale timer entries (e.g. maybe_restore_messages) must be cleared
+        # when the event loop exits due to a connection error.  Without this,
+        # entries accumulated across reconnects can fire against the broken
+        # connection and crash the loop again before the new connection is
+        # fully established, causing a rapid restart loop.
+        x = X(self.app)
+        x.hub.readers = {6: Mock()}
+        x.hub.timer._queue = [1]
+        x.hub.reset = Mock(name='hub.reset()')
+        x.close_then_error(x.hub.poller.poll)
+        x.hub.fire_timers.return_value = 33.37
+        x.hub.poller.poll.return_value = []
+        with pytest.raises(socket.error):
+            asynloop(*x.args)
+        x.hub.timer.clear.assert_called_once()
+
+    def test_hub_timer_not_cleared_on_graceful_shutdown(self):
+        # On graceful shutdown the timer queue must be left intact so that
+        # periodic timers (e.g. heartbeat) keep firing while the pool drains.
+        x = X(self.app)
+        x.hub.reset = Mock(name='hub.reset()')
+        x.hub.on_tick.add(x.closer(mod=2))
+        asynloop(*x.args)
+        x.hub.timer.clear.assert_not_called()
+
+    def test_hub_timer_not_cleared_on_worker_shutdown(self):
+        x = X(self.app)
+        x.hub.reset = Mock(name='hub.reset()')
+        state.should_stop = 303
+        try:
+            with pytest.raises(WorkerShutdown):
+                asynloop(*x.args)
+        finally:
+            state.should_stop = None
+        x.hub.timer.clear.assert_not_called()
+
+    def test_hub_timer_not_cleared_on_worker_terminate(self):
+        x = X(self.app)
+        x.hub.reset = Mock(name='hub.reset()')
+        state.should_terminate = True
+        try:
+            with pytest.raises(WorkerTerminate):
+                asynloop(*x.args)
+        finally:
+            state.should_terminate = None
+        x.hub.timer.clear.assert_not_called()
+
+    def test_hub_timer_clear_error_still_reraises_original(self):
+        # If hub.timer.clear() itself raises, the original connection error
+        # must still be propagated, not the cleanup error.
+        x = X(self.app)
+        x.hub.readers = {6: Mock()}
+        x.hub.timer._queue = [1]
+        x.hub.reset = Mock(name='hub.reset()')
+        x.hub.timer.clear = Mock(
+            name='hub.timer.clear()', side_effect=RuntimeError('clear failed')
+        )
+        x.close_then_error(x.hub.poller.poll)
+        x.hub.fire_timers.return_value = 33.37
+        x.hub.poller.poll.return_value = []
+        with pytest.raises(socket.error):
+            asynloop(*x.args)
+        x.hub.timer.clear.assert_called_once()
+
+    def test_hub_timer_cleared_even_when_reset_raises(self):
+        # hub.timer.clear() must still be called even if hub.reset() raises.
+        # The two cleanup calls are in separate try/except blocks so that a
+        # failure in hub.reset() does not prevent stale timer entries from
+        # being discarded, avoiding stale timers persisting after a reset error.
+        x = X(self.app)
+        x.hub.readers = {6: Mock()}
+        x.hub.timer._queue = [1]
+        x.hub.reset = Mock(name='hub.reset()', side_effect=RuntimeError('reset failed'))
+        x.close_then_error(x.hub.poller.poll)
+        x.hub.fire_timers.return_value = 33.37
+        x.hub.poller.poll.return_value = []
+        with pytest.raises(socket.error):
+            asynloop(*x.args)
+        x.hub.timer.clear.assert_called_once()
+
     def test_hub_not_reset_on_graceful_shutdown(self):
         x = X(self.app)
         x.hub.reset = Mock(name='hub.reset()')