fix(asyncio): fix error propagation in async request path

Ensure `PubNubAsyncioException` always carries a valid `PNStatus` with error data instead of `None`.

fix(asyncio): fix `PubNubAsyncioException.__str__` crash

Handle cases where status or `error_data` is `None` instead of raising `AttributeError`.

fix(event-engine): fix error type checks in effects

Match `PubNubAsyncioException` which is what `request_future` actually returns on failure.

fix(event-engine): fix give-up logic for unlimited retries

Handle `-1 (unlimited)` correctly since `attempts > -1` was always `true`, causing immediate give-up.

fix(event-engine): initialize heartbeat max retry attempts

Use delay class defaults instead of config value which could be `None` causing `TypeError` on comparison.

fix(event-engine): add missing return after `heartbeat` give-up

Prevent falling through to start a heartbeat after deciding to give up.

fix(request-handlers): use explicit `httpx.Timeout` object

Set all four timeout fields explicitly instead of a 2-tuple that left write and pool unset.

fix(request-handlers): enforce wall-clock deadline to survive system sleep

On macOS and Linux, `time.monotonic()` does not advance during system sleep, causing socket and
`asyncio` timeouts (310s subscribe) to stall for hours of wall-clock time. Add `time.time()`-based
deadline checks that detect sleep and cancel stale requests within ~5s of wake.

fix(asyncio): replace `asyncio.wait_for` with wall-clock-aware loop

Use `asyncio.wait()` with periodic `time.time()` checks instead of a single monotonic-based
`wait_for()`, yielding to the event loop between checks.

fix(native-threads): add `WallClockDeadlineWatchdog`

Persistent single daemon thread monitors `time.time()` every 5s and closes the `httpx` session
when the wall-clock deadline passes, interrupting the blocking socket read. Tracks deadlines
per calling thread so concurrent requests (e.g., subscribe + publish) don't interfere. Only armed
for long-timeout requests (>30s). Session is recreated for subsequent requests

test(wall-clock-deadline): add unit tests for sleep detection

Cover both `asyncio` and threads paths: simulated clock jumps, normal passthrough, clean watchdog
shutdown, per-thread deadline isolation, concurrent request independence, cleanup, and exception
propagation.

test(native-threads): add try/finally cleanup to subscribe tests

Ensure `pubnub.stop()` always runs to prevent non-daemon threads from blocking process exit.

test(native-threads): fix flaky where_now and here_now tests

Enable presence heartbeat and use unique channel names so presence registers on the server.

test(file-upload): fix shared state leak in file upload tests

Restore `cipher_key` after use in `send_file` and pass it explicitly to `download_file`.

test(message-actions): use unique channel names

Avoid collisions with stale data from prior test runs.

Author: parfeon

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,6 +62,7 @@ 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
@@ -234,9 +235,17 @@ 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 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
@@ -56,6 +57,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 asyncio.TimeoutError("Wall-clock deadline exceeded")
+
+                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 +137,10 @@ 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 +149,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