fix(subscribe): lazy session init and silent restart on watchdog timeout

Defer `httpx.Client()` creation to first use and map watchdog timeout to `PNTimeoutCategory` for
silent subscribe restart instead of announcing unexpected disconnect.

Author: parfeon

pubnub/pubnub_asyncio.py

@@ -76,7 +76,7 @@ async def main():
 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, PNERR_CONNECTION_ERROR
+from pubnub.errors import PNERR_REQUEST_CANCELLED, PNERR_CLIENT_TIMEOUT
 from pubnub.exceptions import PubNubAsyncioException, PubNubException
 
 # flake8: noqa
@@ -251,11 +251,11 @@ async def request_future(self, options_func, cancellation_event):
             return PubNubAsyncioException(
                 result=None,
                 status=options_func().create_status(
-                    PNStatusCategory.PNUnexpectedDisconnectCategory,
+                    PNStatusCategory.PNTimeoutCategory,
                     None,
                     None,
                     exception=PubNubException(
-                        pn_error=PNERR_CONNECTION_ERROR,
+                        pn_error=PNERR_CLIENT_TIMEOUT,
                         errormsg="Wall-clock deadline exceeded (system sleep detected)"
                     )
                 )

pubnub/request_handlers/httpx.py

@@ -163,15 +163,26 @@ class HttpxRequestHandler(BaseRequestHandler):
     ENDPOINT_THREAD_COUNTER: int = 0
 
     def __init__(self, pubnub):
-        self.session = httpx.Client()
+        self._session = None
+        self._session_lock = threading.Lock()
         self._watchdog = WallClockDeadlineWatchdog()
 
         self.pubnub = pubnub
 
+    def _ensure_session(self):
+        """Return the current httpx.Client, creating one if needed. Thread-safe."""
+        with self._session_lock:
+            if self._session is None or self._session.is_closed:
+                logger.debug("Creating new HTTP session")
+                self._session = httpx.Client()
+            return self._session
+
     def close(self):
         """Clean up resources: stop the watchdog thread and close the HTTP session."""
         self._watchdog.stop()
-        self.session.close()
+        with self._session_lock:
+            if self._session is not None:
+                self._session.close()
 
     async def async_request(self, options_func, cancellation_event):
         raise NotImplementedError("async_request is not implemented for synchronous handler")
@@ -369,9 +380,7 @@ def _invoke_request(self, p_options, e_options, base_origin):
         assert isinstance(p_options, PlatformOptions)
         assert isinstance(e_options, RequestOptions)
 
-        if self.session.is_closed:
-            logger.debug("HTTP session was closed (e.g. by wall-clock watchdog), recreating")
-            self.session = httpx.Client()
+        session = self._ensure_session()
 
         if base_origin:
             url = p_options.pn_config.scheme() + "://" + base_origin + e_options.path
@@ -420,10 +429,10 @@ def _invoke_request(self, p_options, e_options, base_origin):
         use_watchdog = e_options.request_timeout is not None and e_options.request_timeout > 30
 
         if use_watchdog:
-            self._watchdog.set_deadline(self.session, time.time() + e_options.request_timeout)
+            self._watchdog.set_deadline(session, time.time() + e_options.request_timeout)
 
         try:
-            res = self.session.request(**args)
+            res = session.request(**args)
             # Safely access response text - read content first for streaming responses
             try:
                 logger.debug("GOT %s" % res.text)
@@ -436,9 +445,8 @@ def _invoke_request(self, p_options, e_options, base_origin):
 
         except httpx.ConnectError as e:
             if use_watchdog and self._watchdog.triggered:
-                self.session = httpx.Client()
                 raise PubNubException(
-                    pn_error=PNERR_CONNECTION_ERROR,
+                    pn_error=PNERR_CLIENT_TIMEOUT,
                     errormsg="Wall-clock deadline exceeded (system sleep detected)"
                 )
             raise PubNubException(
@@ -447,9 +455,8 @@ def _invoke_request(self, p_options, e_options, base_origin):
             )
         except httpx.TimeoutException as e:
             if use_watchdog and self._watchdog.triggered:
-                self.session = httpx.Client()
                 raise PubNubException(
-                    pn_error=PNERR_CONNECTION_ERROR,
+                    pn_error=PNERR_CLIENT_TIMEOUT,
                     errormsg="Wall-clock deadline exceeded (system sleep detected)"
                 )
             raise PubNubException(
@@ -469,9 +476,8 @@ def _invoke_request(self, p_options, e_options, base_origin):
             )
         except Exception as e:
             if use_watchdog and self._watchdog.triggered:
-                self.session = httpx.Client()
                 raise PubNubException(
-                    pn_error=PNERR_CONNECTION_ERROR,
+                    pn_error=PNERR_CLIENT_TIMEOUT,
                     errormsg="Wall-clock deadline exceeded (system sleep detected)"
                 )
             raise PubNubException(

tests/unit/test_wall_clock_deadline.py

@@ -362,8 +362,9 @@ def test_close_stops_watchdog(self):
         from pubnub.request_handlers.httpx import HttpxRequestHandler
 
         handler = HttpxRequestHandler(MagicMock())
+        session = handler._ensure_session()
         # Start the watchdog by setting a deadline
-        handler._watchdog.set_deadline(handler.session, time.time() + 300)
+        handler._watchdog.set_deadline(session, time.time() + 300)
         self.assertIsNotNone(handler._watchdog._thread)
 
         handler.close()
@@ -391,11 +392,11 @@ def _make_handler(self):
     def test_closed_session_is_recreated_on_next_request(self):
         """After session.close(), the next _invoke_request should recreate the session."""
         handler = self._make_handler()
-        original_session = handler.session
+        original_session = handler._ensure_session()
 
         # Simulate what the watchdog does
-        handler.session.close()
-        self.assertTrue(handler.session.is_closed)
+        original_session.close()
+        self.assertTrue(handler._session.is_closed)
 
         # Build minimal request options
         from pubnub.structures import RequestOptions, PlatformOptions
@@ -422,17 +423,17 @@ def test_closed_session_is_recreated_on_next_request(self):
         except Exception:
             pass
 
-        self.assertFalse(handler.session.is_closed)
-        self.assertIsNot(handler.session, original_session)
+        self.assertFalse(handler._session.is_closed)
+        self.assertIsNot(handler._session, original_session)
 
         handler.close()
 
     def test_open_session_is_not_recreated(self):
         """An open session should not be replaced."""
         handler = self._make_handler()
-        original_session = handler.session
+        original_session = handler._ensure_session()
 
-        self.assertFalse(handler.session.is_closed)
+        self.assertFalse(handler._session.is_closed)
 
         from pubnub.structures import RequestOptions, PlatformOptions
         p_options = MagicMock(spec=PlatformOptions)
@@ -457,18 +458,18 @@ def test_open_session_is_not_recreated(self):
         except Exception:
             pass
 
-        self.assertIs(handler.session, original_session)
+        self.assertIs(handler._session, original_session)
         handler.close()
 
     def test_watchdog_trigger_with_timeout_exception_recreates_session(self):
         """When watchdog triggers and the request gets TimeoutException, session should be recreated."""
         handler = self._make_handler()
-        original_session = handler.session
+        original_session = handler._ensure_session()
 
         # Set up watchdog as triggered
-        handler._watchdog.set_deadline(handler.session, time.time() - 1)
+        handler._watchdog.set_deadline(original_session, time.time() - 1)
         time.sleep(0.3)
-        self.assertTrue(handler.session.is_closed)
+        self.assertTrue(original_session.is_closed)
 
         from pubnub.structures import RequestOptions, PlatformOptions
         p_options = MagicMock(spec=PlatformOptions)
@@ -493,9 +494,9 @@ def test_watchdog_trigger_with_timeout_exception_recreates_session(self):
         except Exception:
             pass
 
-        # Session should have been recreated (either by is_closed check or watchdog trigger handler)
-        self.assertFalse(handler.session.is_closed)
-        self.assertIsNot(handler.session, original_session)
+        # Session should have been recreated by _ensure_session() since original was closed
+        self.assertFalse(handler._session.is_closed)
+        self.assertIsNot(handler._session, original_session)
 
         handler._watchdog.stop()
         handler.close()
@@ -509,10 +510,11 @@ def test_reconnection_time_request_works_after_session_close(self):
         3. /time/0 calls should detect closed session, recreate it, and eventually succeed
         """
         handler = self._make_handler()
+        original_session = handler._ensure_session()
 
         # Step 1: Watchdog closes the session
-        handler.session.close()
-        self.assertTrue(handler.session.is_closed)
+        original_session.close()
+        self.assertTrue(handler._session.is_closed)
 
         from pubnub.structures import RequestOptions, PlatformOptions
         p_options = MagicMock(spec=PlatformOptions)
@@ -540,21 +542,22 @@ def test_reconnection_time_request_works_after_session_close(self):
             # Should be a connection error, NOT "Cannot send a request, as the client has been closed"
             self.assertNotIn("client has been closed", str(e).lower())
 
-        self.assertFalse(handler.session.is_closed)
+        self.assertFalse(handler._session.is_closed)
         handler.close()
 
 
 class TestWallClockErrorCategory(unittest.TestCase):
-    """Tests that wall-clock sleep timeouts produce PNUnexpectedDisconnectCategory,
-    not PNTimeoutCategory — so they route through the reconnection manager with
-    configured retry delays instead of an immediate silent restart.
+    """Tests that wall-clock sleep timeouts produce PNTimeoutCategory,
+    so they silently restart the subscribe loop. If the network is actually
+    down, the next request will fail with a real connection error that routes
+    through the reconnection manager with configured retry delays.
     """
 
-    def test_watchdog_triggered_produces_connection_error(self):
-        """When watchdog triggers during a request, the exception should use PNERR_CONNECTION_ERROR
-        which maps to PNUnexpectedDisconnectCategory in _build_envelope."""
+    def test_watchdog_triggered_produces_timeout_error(self):
+        """When watchdog triggers during a request, the exception should use PNERR_CLIENT_TIMEOUT
+        which maps to PNTimeoutCategory in _build_envelope."""
         from pubnub.request_handlers.httpx import HttpxRequestHandler
-        from pubnub.errors import PNERR_CONNECTION_ERROR
+        from pubnub.errors import PNERR_CLIENT_TIMEOUT
         from pubnub.exceptions import PubNubException
         from pubnub.structures import RequestOptions, PlatformOptions
 
@@ -565,7 +568,7 @@ def test_watchdog_triggered_produces_connection_error(self):
         mock_session = MagicMock(spec=httpx.Client)
         mock_session.is_closed = False
         mock_session.request.side_effect = httpx.ReadTimeout("timed out")
-        handler.session = mock_session
+        handler._session = mock_session
 
         # Mock watchdog.triggered to return True (simulates watchdog firing during request)
         type(handler._watchdog).triggered = property(lambda self: True)
@@ -590,10 +593,8 @@ def test_watchdog_triggered_produces_connection_error(self):
         with self.assertRaises(PubNubException) as ctx:
             handler._invoke_request(p_options, e_options, 'ps.pndsn.com')
 
-        self.assertEqual(ctx.exception._pn_error, PNERR_CONNECTION_ERROR)
+        self.assertEqual(ctx.exception._pn_error, PNERR_CLIENT_TIMEOUT)
         self.assertIn("Wall-clock deadline exceeded", ctx.exception._errormsg)
-        # Session should have been recreated
-        self.assertIsNot(handler.session, mock_session)
 
         handler._watchdog.stop()
         handler.close()
@@ -636,8 +637,8 @@ def mock_time():
 
         asyncio.run(run_test())
 
-    def test_async_wall_clock_timeout_maps_to_unexpected_disconnect(self):
-        """WallClockTimeoutError should produce PNUnexpectedDisconnectCategory in request_future."""
+    def test_async_wall_clock_timeout_maps_to_timeout_category(self):
+        """WallClockTimeoutError should produce PNTimeoutCategory in request_future."""
         from pubnub.request_handlers.async_httpx import WallClockTimeoutError
         from pubnub.pubnub_asyncio import PubNubAsyncio
         from pubnub.pnconfiguration import PNConfiguration
@@ -668,7 +669,7 @@ def options_func():
                 self.assertTrue(result.is_error())
                 self.assertEqual(
                     result.status.category,
-                    PNStatusCategory.PNUnexpectedDisconnectCategory
+                    PNStatusCategory.PNTimeoutCategory
                 )
             finally:
                 await pubnub.stop()