asyncio and native threads timeout handling fix

.github/workflows/run-tests.yml

@@ -26,7 +26,7 @@ jobs:
       max-parallel: 1
       fail-fast: true
       matrix:
-        python: [3.9.21, 3.10.16, 3.11.11, 3.12.9, 3.13.2]
+        python: ["3.9", "3.10", "3.11", "3.12", "3.13"]
     timeout-minutes: 5
     steps:
       - name: Checkout repository

pubnub/event_engine/effects.py

@@ -6,7 +6,7 @@
 from pubnub.endpoints.presence.leave import Leave
 from pubnub.endpoints.pubsub.subscribe import Subscribe
 from pubnub.enums import PNReconnectionPolicy
-from pubnub.exceptions import PubNubException
+from pubnub.exceptions import PubNubAsyncioException, PubNubException
 from pubnub.features import feature_enabled
 from pubnub.models.server.subscribe import SubscribeMessage
 from pubnub.pubnub import PubNub
@@ -80,9 +80,10 @@ async def handshake_async(self, channels, groups, stop_event, timetoken: int = 0
         request.timetoken(0)
         response = await request.future()
 
-        if isinstance(response, Exception):
+        if isinstance(response, PubNubAsyncioException):
             self.logger.warning(f'Handshake failed: {str(response)}')
-            handshake_failure = events.HandshakeFailureEvent(response, 1, timetoken=timetoken)
+            reason = response.status.error_data if response.status and response.status.error_data else str(response)
+            handshake_failure = events.HandshakeFailureEvent(reason, 1, timetoken=timetoken)
             self.event_engine.trigger(handshake_failure)
         elif response.status.error:
             self.logger.warning(f'Handshake failed: {response.status.error_data.__dict__}')
@@ -184,8 +185,15 @@ def calculate_reconnection_delay(self, attempts):
 
         return delay
 
+    def _should_give_up(self, attempts):
+        if self.reconnection_policy is PNReconnectionPolicy.NONE:
+            return True
+        if self.max_retry_attempts == -1:
+            return False
+        return attempts > self.max_retry_attempts
+
     def run(self):
-        if self.reconnection_policy is PNReconnectionPolicy.NONE or self.invocation.attempts > self.max_retry_attempts:
+        if self._should_give_up(self.invocation.attempts):
             self.give_up(reason=self.invocation.reason, attempt=self.invocation.attempts)
         else:
             attempts = self.invocation.attempts
@@ -214,9 +222,10 @@ async def delayed_reconnect_async(self, delay, attempt):
 
         response = await request.future()
 
-        if isinstance(response, PubNubException):
+        if isinstance(response, PubNubAsyncioException):
             self.logger.warning(f'Reconnect failed: {str(response)}')
-            self.failure(str(response), attempt, self.get_timetoken())
+            reason = response.status.error_data if response.status and response.status.error_data else str(response)
+            self.failure(reason, attempt, self.get_timetoken())
 
         elif response.status.error:
             self.logger.warning(f'Reconnect failed: {response.status.error_data.__dict__}')
@@ -302,10 +311,11 @@ async def heartbeat(self, channels, groups, stop_event):
 
         response = await request.future()
 
-        if isinstance(response, PubNubException):
+        if isinstance(response, PubNubAsyncioException):
             self.logger.warning(f'Heartbeat failed: {str(response)}')
+            reason = response.status.error_data if response.status and response.status.error_data else str(response)
             self.event_engine.trigger(events.HeartbeatFailureEvent(channels=channels, groups=groups,
-                                                                   reason=response.status.error_data, attempt=1))
+                                                                   reason=reason, attempt=1))
         elif response.status and response.status.error:
             self.logger.warning(f'Heartbeat failed: {response.status.error_data.__dict__}')
             self.event_engine.trigger(events.HeartbeatFailureEvent(channels=channels, groups=groups,
@@ -345,18 +355,36 @@ async def leave(self, channels, groups, stop_event):
         leave_request = Leave(self.pubnub).channels(channels).channel_groups(groups).cancellation_event(stop_event)
         leave = await leave_request.future()
 
-        if leave.status.error:
-            self.logger.warning(f'Heartbeat failed: {leave.status.error_data.__dict__}')
+        if isinstance(leave, PubNubAsyncioException):
+            self.logger.warning(f'Leave failed: {str(leave)}')
+        elif leave.status and leave.status.error:
+            self.logger.warning(f'Leave failed: {leave.status.error_data.__dict__}')
 
 
 class HeartbeatDelayedEffect(Effect):
     def __init__(self, pubnub_instance, event_engine_instance,
                  invocation: Union[invocations.PNManageableInvocation, invocations.PNCancelInvocation]) -> None:
         super().__init__(pubnub_instance, event_engine_instance, invocation)
         self.reconnection_policy = pubnub_instance.config.reconnect_policy
-        self.max_retry_attempts = pubnub_instance.config.maximum_reconnection_retries
         self.interval = pubnub_instance.config.reconnection_interval
 
+        if self.reconnection_policy is PNReconnectionPolicy.EXPONENTIAL:
+            self.max_retry_attempts = ExponentialDelay.MAX_RETRIES
+        elif self.reconnection_policy is PNReconnectionPolicy.LINEAR:
+            self.max_retry_attempts = LinearDelay.MAX_RETRIES
+        else:
+            self.max_retry_attempts = 0
+
+        if pubnub_instance.config.maximum_reconnection_retries is not None:
+            self.max_retry_attempts = pubnub_instance.config.maximum_reconnection_retries
+
+    def _should_give_up(self, attempts):
+        if self.reconnection_policy is PNReconnectionPolicy.NONE:
+            return True
+        if self.max_retry_attempts == -1:
+            return False
+        return attempts > self.max_retry_attempts
+
     def calculate_reconnection_delay(self, attempts):
         if self.reconnection_policy is PNReconnectionPolicy.EXPONENTIAL:
             delay = ExponentialDelay.calculate(attempts)
@@ -368,23 +396,19 @@ def calculate_reconnection_delay(self, attempts):
         return delay
 
     def run(self):
-        if self.reconnection_policy is PNReconnectionPolicy.NONE or self.invocation.attempts > self.max_retry_attempts:
+        if self._should_give_up(self.invocation.attempts):
             self.event_engine.trigger(events.HeartbeatGiveUpEvent(channels=self.invocation.channels,
                                                                   groups=self.invocation.groups,
                                                                   reason=self.invocation.reason,
                                                                   attempt=self.invocation.attempts))
+            return
 
         if hasattr(self.pubnub, 'event_loop'):
             self.stop_event = self.get_new_stop_event()
             self.run_async(self.heartbeat(channels=self.invocation.channels, groups=self.invocation.groups,
                                           attempt=self.invocation.attempts, stop_event=self.stop_event))
 
     async def heartbeat(self, channels, groups, attempt, stop_event):
-        if self.reconnection_policy is PNReconnectionPolicy.NONE or self.invocation.attempts > self.max_retry_attempts:
-            self.event_engine.trigger(events.HeartbeatGiveUpEvent(channels=self.invocation.channels,
-                                                                  groups=self.invocation.groups,
-                                                                  reason=self.invocation.reason,
-                                                                  attempt=self.invocation.attempts))
 
         channels = list(filter(lambda ch: not ch.endswith('-pnpres'), self.invocation.channels))
         groups = list(filter(lambda gr: not gr.endswith('-pnpres'), self.invocation.groups))
@@ -395,12 +419,13 @@ async def heartbeat(self, channels, groups, attempt, stop_event):
         await asyncio.sleep(delay)
 
         response = await request.future()
-        if isinstance(response, PubNubException):
+        if isinstance(response, PubNubAsyncioException):
             self.logger.warning(f'Heartbeat failed: {str(response)}')
+            reason = response.status.error_data if response.status and response.status.error_data else str(response)
             self.event_engine.trigger(events.HeartbeatFailureEvent(channels=channels, groups=groups,
-                                                                   reason=response.status.error_data,
+                                                                   reason=reason,
                                                                    attempt=attempt))
-        elif response.status.error:
+        elif response.status and response.status.error:
             self.logger.warning(f'Heartbeat failed: {response.status.error_data.__dict__}')
             self.event_engine.trigger(events.HeartbeatFailureEvent(channels=channels, groups=groups,
                                                                    reason=response.status.error_data,

pubnub/exceptions.py

@@ -43,7 +43,9 @@ def __init__(self, result, status):
         self.status = status
 
     def __str__(self):
-        return str(self.status.error_data.exception)
+        if self.status and hasattr(self.status, 'error_data') and self.status.error_data:
+            return str(self.status.error_data.exception)
+        return f"PubNubAsyncioException(result={self.result}, status={self.status})"
 
     @staticmethod
     def is_error():

pubnub/pubnub.py

@@ -218,10 +218,14 @@ def stop(self):
         Raises:
             Exception: If subscription manager is not enabled
         """
-        if self._subscription_manager is not None:
-            self._subscription_manager.stop()
-        else:
-            raise Exception("Subscription manager is not enabled for this instance")
+        try:
+            if self._subscription_manager is not None:
+                self._subscription_manager.stop()
+            else:
+                raise Exception("Subscription manager is not enabled for this instance")
+        finally:
+            if hasattr(self._request_handler, 'close'):
+                self._request_handler.close()
 
     def request_deferred(self, options_func):
         raise NotImplementedError

pubnub/pubnub_asyncio.py

@@ -62,20 +62,21 @@ async def main():
 from pubnub.event_engine.models import events, states
 
 from pubnub.models.consumer.common import PNStatus
+from pubnub.models.consumer.pn_error_data import PNErrorData
 from pubnub.dtos import SubscribeOperation, UnsubscribeOperation
 from pubnub.event_engine.statemachine import StateMachine
 from pubnub.endpoints.presence.heartbeat import Heartbeat
 from pubnub.endpoints.presence.leave import Leave
 from pubnub.endpoints.pubsub.subscribe import Subscribe
 from pubnub.pubnub_core import PubNubCore
 from pubnub.request_handlers.base import BaseRequestHandler
-from pubnub.request_handlers.async_httpx import AsyncHttpxRequestHandler
+from pubnub.request_handlers.async_httpx import AsyncHttpxRequestHandler, WallClockTimeoutError
 from pubnub.workers import SubscribeMessageWorker
 from pubnub.managers import SubscriptionManager, PublishSequenceManager, ReconnectionManager
 from pubnub import utils
 from pubnub.enums import PNStatusCategory, PNHeartbeatNotificationOptions, PNOperationType, PNReconnectionPolicy
 from pubnub.callbacks import SubscribeCallback, ReconnectionCallback
-from pubnub.errors import PNERR_REQUEST_CANCELLED, PNERR_CLIENT_TIMEOUT
+from pubnub.errors import PNERR_REQUEST_CANCELLED, PNERR_CLIENT_TIMEOUT, PNERR_CONNECTION_ERROR
 from pubnub.exceptions import PubNubAsyncioException, PubNubException
 
 # flake8: noqa
@@ -234,9 +235,30 @@ async def request_future(self, options_func, cancellation_event):
             res = await self._request_handler.async_request(options_func, cancellation_event)
             return res
         except PubNubException as e:
+            if e.status is not None:
+                status = e.status
+            else:
+                status = PNStatus()
+                status.category = PNStatusCategory.PNBadRequestCategory
+                status.error = True
+                status.error_data = PNErrorData(str(e), e)
+                status.status_code = e._status_code if e._status_code != 0 else None
             return PubNubAsyncioException(
                 result=None,
-                status=e.status
+                status=status
+            )
+        except WallClockTimeoutError:
+            return PubNubAsyncioException(
+                result=None,
+                status=options_func().create_status(
+                    PNStatusCategory.PNUnexpectedDisconnectCategory,
+                    None,
+                    None,
+                    exception=PubNubException(
+                        pn_error=PNERR_CONNECTION_ERROR,
+                        errormsg="Wall-clock deadline exceeded (system sleep detected)"
+                    )
+                )
             )
         except asyncio.TimeoutError:
             return PubNubAsyncioException(

pubnub/request_handlers/async_httpx.py

@@ -1,6 +1,7 @@
 from asyncio import Event
 import asyncio
 import logging
+import time
 import httpx
 import json  # noqa # pylint: disable=W0611
 import urllib
@@ -16,6 +17,11 @@
 logger = logging.getLogger("pubnub")
 
 
+class WallClockTimeoutError(asyncio.TimeoutError):
+    """Raised when a wall-clock deadline is exceeded, typically due to system sleep."""
+    pass
+
+
 class PubNubAsyncHTTPTransport(httpx.AsyncHTTPTransport):
     is_closed = False
 
@@ -56,6 +62,39 @@ def sync_request(self, **_):
     def threaded_request(self, **_):
         raise NotImplementedError("threaded_request is not implemented for asyncio handler")
 
+    WALL_CLOCK_CHECK_INTERVAL = 5.0
+
+    async def _request_with_wall_clock_deadline(self, request_arguments, timeout):
+        """Execute an HTTP request with wall-clock deadline enforcement.
+
+        On macOS and Linux, time.monotonic() (and thus asyncio timeouts, socket timeouts)
+        does not advance during system sleep. A 310-second subscribe timeout can take hours
+        of wall-clock time if the machine sleeps. This method uses time.time() (wall clock)
+        to enforce the deadline regardless of sleep, while yielding to the event loop between checks.
+        """
+        if timeout is None:
+            return await self._session.request(**request_arguments)
+
+        wall_deadline = time.time() + timeout
+        request_task = asyncio.ensure_future(self._session.request(**request_arguments))
+
+        try:
+            while True:
+                remaining = wall_deadline - time.time()
+                if remaining <= 0:
+                    request_task.cancel()
+                    raise WallClockTimeoutError("Wall-clock deadline exceeded (system sleep detected)")
+
+                done, _ = await asyncio.wait(
+                    {request_task},
+                    timeout=min(self.WALL_CLOCK_CHECK_INTERVAL, remaining)
+                )
+                if done:
+                    return request_task.result()
+        except BaseException:
+            request_task.cancel()
+            raise
+
     async def async_request(self, options_func, cancellation_event):
         """
         Query string should be provided as a manually serialized and encoded string.
@@ -103,7 +142,11 @@ async def async_request(self, options_func, cancellation_event):
             'headers': request_headers,
             'url': full_url,
             'follow_redirects': options.allow_redirects,
-            'timeout': (options.connect_timeout, options.request_timeout),
+            'timeout': httpx.Timeout(
+                connect=options.connect_timeout,
+                read=options.request_timeout,
+                write=options.connect_timeout,
+                pool=options.connect_timeout),
         }
         if options.is_post() or options.is_patch():
             request_arguments['content'] = options.data
@@ -112,9 +155,8 @@ async def async_request(self, options_func, cancellation_event):
         try:
             if not self._session:
                 await self.create_session()
-            response = await asyncio.wait_for(
-                self._session.request(**request_arguments),
-                options.request_timeout
+            response = await self._request_with_wall_clock_deadline(
+                request_arguments, options.request_timeout
             )
         except (asyncio.TimeoutError, asyncio.CancelledError):
             raise