From ef145afd3fa79a7ce10c55f661d8a8d9fe9a1b94 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Dias?= Date: Fri, 4 Sep 2026 11:29:28 +0100 Subject: [PATCH 1/3] fix: reset the pending ping after a timeout so later ping() calls succeed Once a ping timed out, ConnectionManager.ping() left the cancelled future in place, so every subsequent call awaited it and failed immediately with "Ping request cancelled due to request timeout" for the life of the client, even though the connection was healthy. A ping rejected for being in an invalid state left an unresolved future behind in the same way, making the next ping hang. Track each ping's pending heartbeat by its own id (RTN13e) and remove it in a finally block, so success, timeout, send failure and cancellation all clean up and concurrent pings are independent. Measure the round trip with a monotonic clock and fix the swapped code/status on the invalid-state error. --- ably/realtime/connectionmanager.py | 46 ++++++++----------- test/ably/realtime/realtimeconnection_test.py | 39 +++++++++++++--- 2 files changed, 51 insertions(+), 34 deletions(-) diff --git a/ably/realtime/connectionmanager.py b/ably/realtime/connectionmanager.py index 8b51fb0f..8a479936 100644 --- a/ably/realtime/connectionmanager.py +++ b/ably/realtime/connectionmanager.py @@ -2,8 +2,8 @@ import asyncio import logging +import time from collections import deque -from datetime import datetime from itertools import zip_longest from typing import TYPE_CHECKING @@ -125,7 +125,7 @@ def __init__(self, realtime: AblyRealtime, initial_state): self.options = realtime.options self.__ably = realtime self.__state: ConnectionState = initial_state - self.__ping_future: asyncio.Future | None = None + self.__pending_pings: dict[str, asyncio.Future[None]] = {} self.__timeout_in_secs: float = self.options.realtime_request_timeout / 1000 self.transport: WebSocketTransport | None = None self.__connection_details: ConnectionDetails | None = None @@ -332,29 +332,21 @@ def fail_queued_messages(self, err) -> None: self.pending_message_queue.complete_all_messages(error) async def ping(self) -> float: - if self.__ping_future: - try: - response = await self.__ping_future - except asyncio.CancelledError: - raise AblyException("Ping request cancelled due to request timeout", 504, 50003) from None - return response - - self.__ping_future = asyncio.Future() - if self.__state in [ConnectionState.CONNECTED, ConnectionState.CONNECTING]: - self.__ping_id = get_random_id() - ping_start_time = datetime.now().timestamp() - await self.send_protocol_message({"action": ProtocolMessageAction.HEARTBEAT, - "id": self.__ping_id}) - else: - raise AblyException("Cannot send ping request. Calling ping in invalid state", 40000, 400) + if self.__state not in (ConnectionState.CONNECTED, ConnectionState.CONNECTING): + raise AblyException("Cannot send ping request. Calling ping in invalid state", 400, 40000) + + # RTN13e: the id tells this ping's echo apart from server heartbeats and other pings + ping_id = get_random_id() + echo = self.__pending_pings[ping_id] = asyncio.get_running_loop().create_future() + start_time = time.monotonic() try: - await asyncio.wait_for(self.__ping_future, self.__timeout_in_secs) + await self.send_protocol_message({"action": ProtocolMessageAction.HEARTBEAT, "id": ping_id}) + await asyncio.wait_for(echo, self.__timeout_in_secs) except asyncio.TimeoutError: raise AblyException("Timeout waiting for ping response", 504, 50003) from None - - ping_end_time = datetime.now().timestamp() - response_time_ms = (ping_end_time - ping_start_time) * 1000 - return round(response_time_ms, 2) + finally: + self.__pending_pings.pop(ping_id, None) + return round((time.monotonic() - start_time) * 1000, 2) def on_connected(self, connection_details: ConnectionDetails, connection_id: str, reason: AblyException | None = None) -> None: @@ -458,12 +450,10 @@ def on_channel_message(self, msg: dict) -> None: self.__ably.channels._on_channel_message(msg) def on_heartbeat(self, id: str | None) -> None: - if self.__ping_future: - # Resolve on heartbeat from ping request. - if self.__ping_id == id: - if not self.__ping_future.cancelled(): - self.__ping_future.set_result(None) - self.__ping_future = None + echo = self.__pending_pings.pop(id, None) + # the echo can arrive while wait_for is still cancelling a timed-out ping + if echo is not None and not echo.done(): + echo.set_result(None) def on_ack( self, serial: int, count: int, res: list[PublishResult] | None diff --git a/test/ably/realtime/realtimeconnection_test.py b/test/ably/realtime/realtimeconnection_test.py index 2593eb3e..133212cc 100644 --- a/test/ably/realtime/realtimeconnection_test.py +++ b/test/ably/realtime/realtimeconnection_test.py @@ -127,8 +127,8 @@ async def test_connection_ping_initialized(self): assert ably.connection.state == ConnectionState.INITIALIZED with pytest.raises(AblyException) as exception: await ably.connection.ping() - assert exception.value.code == 400 - assert exception.value.status_code == 40000 + assert exception.value.code == 40000 + assert exception.value.status_code == 400 async def test_connection_ping_failed(self): ably = await TestApp.get_ably_realtime(key=self.valid_key_format) @@ -136,8 +136,8 @@ async def test_connection_ping_failed(self): assert ably.connection.state == ConnectionState.FAILED with pytest.raises(AblyException) as exception: await ably.connection.ping() - assert exception.value.code == 400 - assert exception.value.status_code == 40000 + assert exception.value.code == 40000 + assert exception.value.status_code == 400 await ably.close() async def test_connection_ping_closed(self): @@ -147,8 +147,8 @@ async def test_connection_ping_closed(self): await ably.close() with pytest.raises(AblyException) as exception: await ably.connection.ping() - assert exception.value.code == 400 - assert exception.value.status_code == 40000 + assert exception.value.code == 40000 + assert exception.value.status_code == 400 async def test_auto_connect(self): ably = await TestApp.get_ably_realtime() @@ -212,6 +212,33 @@ async def new_send_protocol_message(protocol_message): assert exception.value.code == 50003 assert exception.value.status_code == 504 + + # one timed-out ping must not break later pings + ably.connection.connection_manager.send_protocol_message = original_send_protocol_message + response_time_ms = await asyncio.wait_for(ably.connection.ping(), timeout=5) + assert type(response_time_ms) is float + await ably.close() + + async def test_ping_after_invalid_state_ping(self): + # a ping rejected for bad state must not break later pings + ably = await TestApp.get_ably_realtime(auto_connect=False) + with pytest.raises(AblyException) as exception: + await ably.connection.ping() + assert exception.value.code == 40000 + + ably.connect() + await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5) + response_time_ms = await asyncio.wait_for(ably.connection.ping(), timeout=5) + assert type(response_time_ms) is float + await ably.close() + + async def test_concurrent_pings(self): + ably = await TestApp.get_ably_realtime() + await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5) + results = await asyncio.wait_for( + asyncio.gather(ably.connection.ping(), ably.connection.ping(), ably.connection.ping()), timeout=5 + ) + assert all(type(response_time_ms) is float for response_time_ms in results) await ably.close() async def test_disconnected_retry_timeout(self): From 083895b5247dc852d855fd46d19328c63a258bd8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Dias?= Date: Mon, 7 Sep 2026 10:54:24 +0100 Subject: [PATCH 2/3] fix: fail pending pings immediately when the connection drops A heartbeat echo cannot arrive once the connection has left the connected state, so a ping whose heartbeat was lost to a dropped connection used to sit there until realtime_request_timeout expired. Fail pending pings as soon as the state changes to DISCONNECTED, SUSPENDED, CLOSING, CLOSED or FAILED, with the state change reason or the matching ConnectionErrors entry. --- ably/realtime/connectionmanager.py | 17 ++++++++ test/ably/realtime/realtimeconnection_test.py | 42 +++++++++++++++++++ 2 files changed, 59 insertions(+) diff --git a/ably/realtime/connectionmanager.py b/ably/realtime/connectionmanager.py index 8a479936..b7d0cb3c 100644 --- a/ably/realtime/connectionmanager.py +++ b/ably/realtime/connectionmanager.py @@ -150,6 +150,17 @@ def enact_state_change(self, state: ConnectionState, reason: AblyException | Non if reason: self.__error_reason = reason + # A heartbeat echo cannot arrive once the connection is no longer connected, + # so fail pending pings now rather than letting them run to the request timeout + if state in ( + ConnectionState.DISCONNECTED, + ConnectionState.SUSPENDED, + ConnectionState.CLOSING, + ConnectionState.CLOSED, + ConnectionState.FAILED, + ): + self.__fail_pending_pings(reason or ConnectionErrors[state]) + # RTN16d: Clear connection state when entering SUSPENDED or terminal states if state == ConnectionState.SUSPENDED or state in ( ConnectionState.CLOSED, @@ -455,6 +466,12 @@ def on_heartbeat(self, id: str | None) -> None: if echo is not None and not echo.done(): echo.set_result(None) + def __fail_pending_pings(self, error: AblyException) -> None: + pending, self.__pending_pings = self.__pending_pings, {} + for echo in pending.values(): + if not echo.done(): + echo.set_exception(error) + def on_ack( self, serial: int, count: int, res: list[PublishResult] | None ) -> None: diff --git a/test/ably/realtime/realtimeconnection_test.py b/test/ably/realtime/realtimeconnection_test.py index 133212cc..6e15c236 100644 --- a/test/ably/realtime/realtimeconnection_test.py +++ b/test/ably/realtime/realtimeconnection_test.py @@ -232,6 +232,48 @@ async def test_ping_after_invalid_state_ping(self): assert type(response_time_ms) is float await ably.close() + async def test_ping_fails_immediately_when_connection_drops(self): + # a ping whose heartbeat is lost to a dropped connection must fail as soon as the + # connection is known to be down, not after realtime_request_timeout + async with WsProxy(self.test_vars["host"]) as proxy: + ably = await TestApp.get_ably_realtime( + realtime_request_timeout=20000, + tls=False, + endpoint=proxy.endpoint, + ) + try: + await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=10) + + connection_manager = ably.connection.connection_manager + original_send_protocol_message = connection_manager.send_protocol_message + + async def drop_heartbeats(protocol_message): + if protocol_message.get('action') == ProtocolMessageAction.HEARTBEAT: + return + await original_send_protocol_message(protocol_message) + + connection_manager.send_protocol_message = drop_heartbeats + ping = asyncio.ensure_future(ably.connection.ping()) + await asyncio.sleep(0.5) + assert not ping.done() + + # Simulate server sending a normal WS close frame + await proxy.close_active_connection() + + with pytest.raises(AblyException) as exception: + await asyncio.wait_for(ping, timeout=5) + assert exception.value.code == 80003 + assert exception.value.status_code == 400 + + # once reconnected, pings work again + connection_manager.send_protocol_message = original_send_protocol_message + if ably.connection.state != ConnectionState.CONNECTED: + await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=10) + response_time_ms = await asyncio.wait_for(ably.connection.ping(), timeout=5) + assert type(response_time_ms) is float + finally: + await ably.close() + async def test_concurrent_pings(self): ably = await TestApp.get_ably_realtime() await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5) From 44829cc70e26d5c623d87ab9c3e32c97190e863e Mon Sep 17 00:00:00 2001 From: owenpearson Date: Mon, 7 Sep 2026 15:13:43 +0100 Subject: [PATCH 3/3] fix: re-send a pending ping's heartbeat once the connection is back Failing pending pings on DISCONNECTED made a recoverable blip a user-visible error: DISCONNECTED is absent from RTN13b's list, and RTN13d puts it alongside CONNECTING as a state where the ping should be done once the connection is CONNECTED. The common drop sets retry_immediately (RTN15a) and reconnects well inside realtimeRequestTimeout, so those pings can still succeed. Track each pending ping's start time alongside its future and send its heartbeat on the transition to CONNECTED, covering both a ping registered while connecting and one whose heartbeat was lost with a dropped transport. The round trip is measured from that send, and the single realtime_request_timeout still bounds the total wait. Pings are still failed on transition to SUSPENDED, CLOSING, CLOSED or FAILED per RTN13b. --- ably/realtime/connectionmanager.py | 75 ++++++++++++++----- test/ably/realtime/realtimeconnection_test.py | 61 +++++++++++---- 2 files changed, 105 insertions(+), 31 deletions(-) diff --git a/ably/realtime/connectionmanager.py b/ably/realtime/connectionmanager.py index b7d0cb3c..f910dfe6 100644 --- a/ably/realtime/connectionmanager.py +++ b/ably/realtime/connectionmanager.py @@ -120,12 +120,29 @@ def clear(self) -> None: self.messages.clear() +class PendingPing: + """Represents a ping awaiting its heartbeat echo from the server""" + + def __init__(self, id: str): + self.id = id + self.future: asyncio.Future[None] = asyncio.get_running_loop().create_future() + self.start_time: float = time.monotonic() + + @property + def message(self) -> dict: + return {"action": ProtocolMessageAction.HEARTBEAT, "id": self.id} + + @property + def response_time_ms(self) -> float: + return round((time.monotonic() - self.start_time) * 1000, 2) + + class ConnectionManager(EventEmitter): def __init__(self, realtime: AblyRealtime, initial_state): self.options = realtime.options self.__ably = realtime self.__state: ConnectionState = initial_state - self.__pending_pings: dict[str, asyncio.Future[None]] = {} + self.__pending_pings: dict[str, PendingPing] = {} self.__timeout_in_secs: float = self.options.realtime_request_timeout / 1000 self.transport: WebSocketTransport | None = None self.__connection_details: ConnectionDetails | None = None @@ -150,10 +167,9 @@ def enact_state_change(self, state: ConnectionState, reason: AblyException | Non if reason: self.__error_reason = reason - # A heartbeat echo cannot arrive once the connection is no longer connected, - # so fail pending pings now rather than letting them run to the request timeout + # RTN13b: a ping cannot complete from these states, and the echo of a heartbeat already + # sent will never arrive, so fail pending pings rather than leaving them to time out if state in ( - ConnectionState.DISCONNECTED, ConnectionState.SUSPENDED, ConnectionState.CLOSING, ConnectionState.CLOSED, @@ -343,21 +359,43 @@ def fail_queued_messages(self, err) -> None: self.pending_message_queue.complete_all_messages(error) async def ping(self) -> float: + # RTN13b if self.__state not in (ConnectionState.CONNECTED, ConnectionState.CONNECTING): raise AblyException("Cannot send ping request. Calling ping in invalid state", 400, 40000) # RTN13e: the id tells this ping's echo apart from server heartbeats and other pings - ping_id = get_random_id() - echo = self.__pending_pings[ping_id] = asyncio.get_running_loop().create_future() - start_time = time.monotonic() + pending_ping = PendingPing(get_random_id()) + self.__pending_pings[pending_ping.id] = pending_ping try: - await self.send_protocol_message({"action": ProtocolMessageAction.HEARTBEAT, "id": ping_id}) - await asyncio.wait_for(echo, self.__timeout_in_secs) + # RTN13d: while connecting, the heartbeat goes out once the connection is established + if self.__state == ConnectionState.CONNECTED: + await self.__send_heartbeat(pending_ping) + # RTN13c + await asyncio.wait_for(pending_ping.future, self.__timeout_in_secs) except asyncio.TimeoutError: raise AblyException("Timeout waiting for ping response", 504, 50003) from None finally: - self.__pending_pings.pop(ping_id, None) - return round((time.monotonic() - start_time) * 1000, 2) + self.__pending_pings.pop(pending_ping.id, None) + return pending_ping.response_time_ms + + async def __send_heartbeat(self, pending_ping: PendingPing) -> None: + pending_ping.start_time = time.monotonic() + try: + await self.send_protocol_message(pending_ping.message) + except Exception as error: + # the caller is waiting on the future, so the send failure is surfaced there + if not pending_ping.future.done(): + pending_ping.future.set_exception(error) + + def __send_pending_pings(self) -> None: + """RTN13d: send the heartbeat for each ping waiting on the connection + + A ping registered while the connection was establishing has not sent its heartbeat yet, + and one whose heartbeat went out before the transport went away will never see that echo, + so both are (re)sent here and their round trip is measured from this point. + """ + for pending_ping in list(self.__pending_pings.values()): + asyncio.create_task(self.__send_heartbeat(pending_ping)) def on_connected(self, connection_details: ConnectionDetails, connection_id: str, reason: AblyException | None = None) -> None: @@ -461,16 +499,18 @@ def on_channel_message(self, msg: dict) -> None: self.__ably.channels._on_channel_message(msg) def on_heartbeat(self, id: str | None) -> None: - echo = self.__pending_pings.pop(id, None) + if id is None: + return + pending_ping = self.__pending_pings.pop(id, None) # the echo can arrive while wait_for is still cancelling a timed-out ping - if echo is not None and not echo.done(): - echo.set_result(None) + if pending_ping is not None and not pending_ping.future.done(): + pending_ping.future.set_result(None) def __fail_pending_pings(self, error: AblyException) -> None: pending, self.__pending_pings = self.__pending_pings, {} - for echo in pending.values(): - if not echo.done(): - echo.set_exception(error) + for pending_ping in pending.values(): + if not pending_ping.future.done(): + pending_ping.future.set_exception(error) def on_ack( self, serial: int, count: int, res: list[PublishResult] | None @@ -637,6 +677,7 @@ def notify_state(self, state: ConnectionState, reason: AblyException | None = No if state == ConnectionState.CONNECTED: self.send_queued_messages() + self.__send_pending_pings() elif state in ( ConnectionState.CLOSING, ConnectionState.CLOSED, diff --git a/test/ably/realtime/realtimeconnection_test.py b/test/ably/realtime/realtimeconnection_test.py index 6e15c236..dbd20698 100644 --- a/test/ably/realtime/realtimeconnection_test.py +++ b/test/ably/realtime/realtimeconnection_test.py @@ -1,4 +1,5 @@ import asyncio +import time import pytest from websockets import connect as _ws_connect @@ -232,9 +233,10 @@ async def test_ping_after_invalid_state_ping(self): assert type(response_time_ms) is float await ably.close() - async def test_ping_fails_immediately_when_connection_drops(self): - # a ping whose heartbeat is lost to a dropped connection must fail as soon as the - # connection is known to be down, not after realtime_request_timeout + async def test_ping_survives_connection_drop(self): + # RTN13d: a heartbeat lost with a dropped connection is sent again once the connection is + # back, rather than the ping failing or waiting out realtime_request_timeout + outage_secs = 1 async with WsProxy(self.test_vars["host"]) as proxy: ably = await TestApp.get_ably_realtime( realtime_request_timeout=20000, @@ -253,27 +255,58 @@ async def drop_heartbeats(protocol_message): await original_send_protocol_message(protocol_message) connection_manager.send_protocol_message = drop_heartbeats + ping_start_time = time.monotonic() ping = asyncio.ensure_future(ably.connection.ping()) - await asyncio.sleep(0.5) + await asyncio.sleep(outage_secs) assert not ping.done() # Simulate server sending a normal WS close frame + connection_manager.send_protocol_message = original_send_protocol_message await proxy.close_active_connection() - with pytest.raises(AblyException) as exception: - await asyncio.wait_for(ping, timeout=5) - assert exception.value.code == 80003 - assert exception.value.status_code == 400 - - # once reconnected, pings work again - connection_manager.send_protocol_message = original_send_protocol_message - if ably.connection.state != ConnectionState.CONNECTED: - await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=10) - response_time_ms = await asyncio.wait_for(ably.connection.ping(), timeout=5) + response_time_ms = await asyncio.wait_for(ping, timeout=15) + elapsed_ms = (time.monotonic() - ping_start_time) * 1000 assert type(response_time_ms) is float + # the round trip is measured from the heartbeat that was sent on reconnection, + # so it excludes the time spent waiting for the connection + assert response_time_ms < elapsed_ms - outage_secs * 1000 finally: await ably.close() + async def test_ping_fails_when_connection_closes(self): + # RTN13b: a ping cannot complete once the connection has transitioned to CLOSING + ably = await TestApp.get_ably_realtime(realtime_request_timeout=20000) + await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5) + + connection_manager = ably.connection.connection_manager + original_send_protocol_message = connection_manager.send_protocol_message + + async def drop_heartbeats(protocol_message): + if protocol_message.get('action') == ProtocolMessageAction.HEARTBEAT: + return + await original_send_protocol_message(protocol_message) + + connection_manager.send_protocol_message = drop_heartbeats + ping = asyncio.ensure_future(ably.connection.ping()) + await asyncio.sleep(0.5) + assert not ping.done() + + await ably.close() + + with pytest.raises(AblyException) as exception: + await asyncio.wait_for(ping, timeout=5) + assert exception.value.code == 80017 + assert exception.value.status_code == 400 + + async def test_ping_while_connecting(self): + # RTN13d: a ping requested while connecting is done once the connection is established + ably = await TestApp.get_ably_realtime(auto_connect=False) + ably.connect() + assert ably.connection.state == ConnectionState.CONNECTING + response_time_ms = await asyncio.wait_for(ably.connection.ping(), timeout=15) + assert type(response_time_ms) is float + await ably.close() + async def test_concurrent_pings(self): ably = await TestApp.get_ably_realtime() await asyncio.wait_for(ably.connection.once_async(ConnectionState.CONNECTED), timeout=5)