diff --git a/nats-core/src/nats/client/__init__.py b/nats-core/src/nats/client/__init__.py index 73671199..a0c25dad 100644 --- a/nats-core/src/nats/client/__init__.py +++ b/nats-core/src/nats/client/__init__.py @@ -454,7 +454,7 @@ def __init__( self._ping_interval = ping_interval self._max_outstanding_pings = max_outstanding_pings self._pings_outstanding = 0 - self._last_pong_received = asyncio.get_event_loop().time() + self._last_pong_received = asyncio.get_running_loop().time() self._last_ping_sent = self._last_pong_received self._pong_waker = asyncio.Event() self._disconnected_callbacks = [] @@ -982,6 +982,11 @@ async def _force_disconnect(self) -> None: self._status = ClientStatus.CONNECTED self._last_server = server + # Reset keepalive state; a ping-exhaustion reconnect must not inherit a stale counter. + self._pings_outstanding = 0 + self._last_pong_received = asyncio.get_running_loop().time() + self._last_ping_sent = self._last_pong_received + if new_server_info.connect_urls: for url in new_server_info.connect_urls: if url not in self._server_pool: diff --git a/nats-core/tests/test_client.py b/nats-core/tests/test_client.py index da41c2b9..97bfd810 100644 --- a/nats-core/tests/test_client.py +++ b/nats-core/tests/test_client.py @@ -197,6 +197,61 @@ def on_reconnect(): pass +async def test_reconnect_resets_ping_state(server): + """A ping-exhaustion reconnect must reset outstanding-ping state (C2). + + The reconnect success path restored the connection and subscriptions but + never reset ``_pings_outstanding``. When the disconnect was itself caused by + reaching ``max_outstanding_pings``, the stale counter survived into the new + connection and the write loop re-tripped it one ``ping_interval`` later, + reconnecting forever. + """ + reconnects = 0 + + def on_reconnect(): + nonlocal reconnects + reconnects += 1 + + client = await connect( + server.client_url, + timeout=1.0, + allow_reconnect=True, + reconnect_time_wait=0.1, + ping_interval=0.2, + max_outstanding_pings=2, + ) + client.add_reconnected_callback(on_reconnect) + + try: + # Put the client in the exact state a ping-timeout disconnect leaves + # behind: outstanding pings already at the max. Set directly so the + # repro is deterministic without needing an unresponsive server. + client._pings_outstanding = client._max_outstanding_pings + + await client.force_reconnect() + assert reconnects == 1 + + # The reconnect must clear the stale counter; otherwise the write loop + # force-disconnects again on its first idle timeout. + assert client._pings_outstanding == 0 + + # Let several ping intervals pass: with the stale counter the client + # spirals into repeated reconnects; reset, it stays put. + await asyncio.sleep(0.6) + assert reconnects == 1, f"connection spiraled into {reconnects} reconnects" + assert client._status == ClientStatus.CONNECTED + + # Still usable after the reconnect. + subject = f"test.ping.spiral.{uuid.uuid4()}" + sub = await client.subscribe(subject) + await client.publish(subject, b"alive") + await client.flush() + msg = await sub.next(timeout=1.0) + assert msg.data == b"alive" + finally: + await client.close() + + @pytest.mark.asyncio async def test_reconnect_with_invalid_auth_does_not_silently_succeed(): """Server-side CONNECT rejection on reconnect must not flip status to CONNECTED.