From 1d4d764c84f6720c3824863f13f9a7ab746c2e57 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 13 Aug 2026 07:29:07 +0000 Subject: [PATCH 1/3] Refresh the realtime JWT before it expires (ROB-4017) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit RealtimeWorker._maybe_refresh_auth() only re-pushed the JWT when the token string had changed. It never looked at `exp`, and nothing else refreshes the token on a realtime-only path: the DAL refreshes reactively, from patch_postgrest_execute on a PGRST301/expired error, and an idle worker issues no postgrest queries. So the same 1-hour token was re-sent every tick until Supabase closed the socket with "InvalidJWTToken: Token has expired N seconds ago". In 48h of production logs that is 48 "Realtime channel unhealthy (ChannelStates.CLOSED), reconnecting" lines — one per hour, on the hour — plus the matching re-sign-ins and a few bare 1006 closes. The reconnect path does force a real sign_in(), so it self-heals, but every hour there is a connection drop and a multi-minute window running an already-dead token during which inbound broadcasts are lost; only the much slower claim poll catches that work. Decode the token's exp (signature deliberately unverified — it is our own token and the expiry is the only claim needed) and proactively re-sign-in when it falls within CONVERSATION_WORKER_AUTH_REFRESH_LEEWAY_SECONDS (default 300, comfortably above the 60s refresh interval so a tick always lands inside the window). The re-sign-in reuses _full_reconnect's bounded-sign_in pattern so a half-open connection cannot stall the loop and stop health checks with it. An undecodable or exp-less token, or a sign_in that hands back the same expiring token, falls through to the existing unhealthy -> _full_reconnect safety net, and _proactive_refresh_attempted_for bounds us to one attempt per token so that case cannot spin. Also log the first reconnect of a run at INFO: a single self-healing reconnect is routine, and only one that did not take last time round says something is actually wrong. Signed-off-by: Claude --- holmes/common/env_vars.py | 7 + .../conversations_worker/realtime_manager.py | 86 +++++++- .../test_realtime_manager.py | 185 +++++++++++++++++- 3 files changed, 275 insertions(+), 3 deletions(-) diff --git a/holmes/common/env_vars.py b/holmes/common/env_vars.py index 7f6c4a4bc..7fd2b9d59 100644 --- a/holmes/common/env_vars.py +++ b/holmes/common/env_vars.py @@ -265,6 +265,13 @@ def _load_temperature() -> Optional[float]: CONVERSATION_WORKER_AUTH_REFRESH_INTERVAL_SECONDS = float( os.environ.get("CONVERSATION_WORKER_AUTH_REFRESH_INTERVAL_SECONDS", 60) ) +# How close to its `exp` the realtime JWT may get before we proactively +# re-sign-in. Must stay comfortably above the refresh interval above so at +# least one refresh tick lands inside the window; otherwise the token expires +# unrefreshed and Supabase closes the socket with InvalidJWTToken. +CONVERSATION_WORKER_AUTH_REFRESH_LEEWAY_SECONDS = float( + os.environ.get("CONVERSATION_WORKER_AUTH_REFRESH_LEEWAY_SECONDS", 300) +) # Upper bound on how long a silently-dead realtime WebSocket can go undetected. # The realtime library can leave a stale connection in place when the server # closes the socket cleanly (ConnectionClosedOK) — _listen_task exits, no diff --git a/holmes/core/conversations_worker/realtime_manager.py b/holmes/core/conversations_worker/realtime_manager.py index 7d9feeade..e0cc2cbae 100644 --- a/holmes/core/conversations_worker/realtime_manager.py +++ b/holmes/core/conversations_worker/realtime_manager.py @@ -23,15 +23,18 @@ import os import ssl import threading +import time import urllib.parse from typing import Any, Callable, Dict, Optional, TYPE_CHECKING +import jwt import realtime._async.client as rt_client from realtime._async.channel import ChannelStates from realtime._async.client import AsyncRealtimeClient from holmes.common.env_vars import ( CONVERSATION_WORKER_AUTH_REFRESH_INTERVAL_SECONDS, + CONVERSATION_WORKER_AUTH_REFRESH_LEEWAY_SECONDS, CONVERSATION_WORKER_REALTIME_HEALTH_TICK_SECONDS, CONVERSATION_WORKER_REALTIME_RECONNECT_MAX_SECONDS, CONVERSATION_WORKER_USE_REALTIME_BROADCAST, @@ -54,6 +57,29 @@ _RECONNECT_SIGN_IN_TIMEOUT_SECONDS = 90 +def _jwt_expires_within(token: str, leeway_seconds: float) -> bool: + """True if ``token``'s ``exp`` claim is within ``leeway_seconds`` from now. + + The signature is deliberately not verified: this is our own token and the + only claim we need is the expiry, which we use to decide *when* to refresh. + An unreadable token or one with no usable ``exp`` returns False — we leave + it alone and let the unhealthy→reconnect path handle it, rather than + re-signing-in on every tick against a token we can't reason about. + """ + try: + claims = jwt.decode( + token, + options={"verify_signature": False, "verify_exp": False}, + ) + except Exception: + logging.debug("Could not decode realtime JWT to read exp", exc_info=True) + return False + exp = claims.get("exp") + if not isinstance(exp, (int, float)) or isinstance(exp, bool): + return False + return exp - time.time() <= leeway_seconds + + # ---- channel topic helpers ---- @@ -234,6 +260,10 @@ def __init__( self._connected = False # Last JWT we pushed to the realtime client via set_auth. self._last_auth_jwt: Optional[str] = None + # Token we last attempted a proactive (near-expiry) re-sign-in for, so + # a sign_in that keeps handing back an expiring token is retried once + # rather than on every refresh tick. + self._proactive_refresh_attempted_for: Optional[str] = None # Set from the async loop to wake the sleep in _run() on stop(). self._async_stop: Optional[asyncio.Event] = None @@ -262,6 +292,7 @@ def start(self) -> None: self._channel = None self._connected = False self._last_auth_jwt = None + self._proactive_refresh_attempted_for = None self._async_stop = None self._thread = threading.Thread( target=self._thread_entry, @@ -328,7 +359,12 @@ async def _run(self) -> None: # own full teardown/reconnect on any failure signal. unhealthy_reason = self._channel_unhealthy() if unhealthy_reason is not None: - logging.warning( + # A single reconnect is routine (network blip, edge + # recycling the socket) and self-healing, so it is not worth + # a WARNING — only a reconnect that did not take last time + # round (attempts > 0) says something is actually wrong. + logging.log( + logging.INFO if reconnect_attempts == 0 else logging.WARNING, "Realtime channel unhealthy (%s), reconnecting", unhealthy_reason, ) @@ -475,6 +511,7 @@ async def _full_reconnect(self) -> None: self._client = None self._channel = None self._last_auth_jwt = None + self._proactive_refresh_attempted_for = None # Bound the re-sign-in (ROB-759): sign_in is a sync HTTP call whose # timeout depends on the DAL's httpx client config; a half-open # connection there would otherwise stall this reconnect loop for the @@ -488,7 +525,31 @@ async def _full_reconnect(self) -> None: await self._connect_and_subscribe() async def _maybe_refresh_auth(self) -> None: - """Re-push the Supabase JWT to the realtime client if it rotated.""" + """Keep the realtime client's JWT fresh. + + Two paths: + + * the token rotated elsewhere (a postgrest query hit PGRST301 and the + DAL re-signed in) → re-push the new one to the client; + * the token is within ``CONVERSATION_WORKER_AUTH_REFRESH_LEEWAY_SECONDS`` + of its ``exp`` → proactively re-sign-in *before* it dies. + + The second path exists because nothing else refreshes the JWT on a + realtime-only path. The DAL refreshes reactively, from + ``patch_postgrest_execute`` on a PGRST301/expired error, and an idle + worker issues no postgrest queries — so this method used to re-read the + same unexpired-when-cached token every tick, return early on + ``new_jwt == self._last_auth_jwt``, and let it lapse. Supabase then + closed the socket (``InvalidJWTToken: Token has expired N seconds ago``), + costing an hourly reconnect plus a window in which the token was already + dead and inbound broadcasts were dropped — only the claim poll's much + slower cadence caught the work. + + A pathological token that comes back still-expiring is not retried in a + hot loop: ``_proactive_refresh_attempted_for`` bounds us to one attempt + per distinct token, leaving the unhealthy→``_full_reconnect`` path as the + safety net it already was. + """ if not self._client: return try: @@ -496,6 +557,27 @@ async def _maybe_refresh_auth(self) -> None: if session is None: return new_jwt = session.access_token + if ( + new_jwt + and new_jwt != self._proactive_refresh_attempted_for + and _jwt_expires_within( + new_jwt, CONVERSATION_WORKER_AUTH_REFRESH_LEEWAY_SECONDS + ) + ): + logging.info( + "Realtime JWT expires within %ss, re-signing in to Supabase", + CONVERSATION_WORKER_AUTH_REFRESH_LEEWAY_SECONDS, + ) + self._proactive_refresh_attempted_for = new_jwt + # Same bounded-sign-in pattern as _full_reconnect (ROB-759): + # sign_in is a sync HTTP call, so a half-open connection there + # would otherwise stall this loop and stop health checks too. + await asyncio.wait_for( + asyncio.to_thread(self.dal.sign_in), + timeout=_RECONNECT_SIGN_IN_TIMEOUT_SECONDS, + ) + session = self.dal.client.auth.get_session() # type: ignore[attr-defined] + new_jwt = session.access_token if session is not None else None if not new_jwt or new_jwt == self._last_auth_jwt: return await self._client.set_auth(new_jwt) diff --git a/tests/core/conversations_worker/test_realtime_manager.py b/tests/core/conversations_worker/test_realtime_manager.py index 302070e17..202ea92bf 100644 --- a/tests/core/conversations_worker/test_realtime_manager.py +++ b/tests/core/conversations_worker/test_realtime_manager.py @@ -3,9 +3,11 @@ import logging import os import ssl as _ssl -from unittest.mock import MagicMock +import time +from unittest.mock import AsyncMock, MagicMock import certifi +import jwt import pytest import realtime._async.client as rt_client from realtime._async.channel import ChannelStates @@ -15,6 +17,7 @@ _build_ssl_context, _install_realtime_log_filter_if_needed, _install_ssl_patch_if_needed, + _jwt_expires_within, _RealtimeConnectivityWarningFilter, broadcast_submit_topic, pg_changes_topic, @@ -470,3 +473,183 @@ async def boom_connect(): m._connect_and_subscribe = boom_connect # type: ignore[method-assign] with pytest.raises(type(exc)): asyncio.run(m._full_reconnect()) + + +# ---- _jwt_expires_within / proactive near-expiry refresh (ROB-4017) ---- + + +# Only the exp claim matters — _jwt_expires_within never verifies the signature. +# Long enough to keep PyJWT's InsecureKeyLengthWarning out of the test output. +_TEST_JWT_KEY = "x" * 32 + + +def _jwt_expiring_in(seconds: float) -> str: + """Mint a JWT whose exp is `seconds` from now.""" + return jwt.encode( + {"exp": int(time.time() + seconds)}, _TEST_JWT_KEY, algorithm="HS256" + ) + + +def test_jwt_expires_within_detects_near_expiry(): + assert _jwt_expires_within(_jwt_expiring_in(60), 300) is True + assert _jwt_expires_within(_jwt_expiring_in(-10), 300) is True # already expired + + +def test_jwt_expires_within_ignores_fresh_token(): + assert _jwt_expires_within(_jwt_expiring_in(3600), 300) is False + + +@pytest.mark.parametrize( + "token", + [ + "not-a-jwt", + "", + jwt.encode({"sub": "no-exp-claim"}, _TEST_JWT_KEY, algorithm="HS256"), + jwt.encode({"exp": "not-a-number"}, _TEST_JWT_KEY, algorithm="HS256"), + ], +) +def test_jwt_expires_within_is_false_for_unreadable_tokens(token): + """An undecodable or exp-less token must not trigger a re-sign-in on every + tick — the unhealthy/reconnect path stays the safety net.""" + assert _jwt_expires_within(token, 300) is False + + +def _manager_with_session(token): + m = _make_manager() + m._client = MagicMock() + m._client.set_auth = AsyncMock() + session = MagicMock() + session.access_token = token + m.dal.client.auth.get_session = MagicMock(return_value=session) + return m + + +def test_refresh_auth_re_signs_in_when_token_near_expiry(): + """The bug: nothing refreshed the JWT on a realtime-only path, so the token + lapsed and Supabase closed the socket with InvalidJWTToken.""" + expiring = _jwt_expiring_in(30) + fresh = _jwt_expiring_in(3600) + m = _manager_with_session(expiring) + + def sign_in(): + # Post-sign-in, get_session returns the rotated token. + rotated = MagicMock() + rotated.access_token = fresh + m.dal.client.auth.get_session = MagicMock(return_value=rotated) + + m.dal.sign_in = MagicMock(side_effect=sign_in) + + asyncio.run(m._maybe_refresh_auth()) + + m.dal.sign_in.assert_called_once() + m._client.set_auth.assert_awaited_once_with(fresh) + assert m._last_auth_jwt == fresh + + +def test_refresh_auth_leaves_fresh_token_alone(): + fresh = _jwt_expiring_in(3600) + m = _manager_with_session(fresh) + m.dal.sign_in = MagicMock() + m._last_auth_jwt = fresh + + asyncio.run(m._maybe_refresh_auth()) + + m.dal.sign_in.assert_not_called() + m._client.set_auth.assert_not_awaited() + + +def test_refresh_auth_still_pushes_externally_rotated_token(): + """Pre-existing behaviour: a token rotated by the DAL's PGRST301 path is + re-pushed even though it is nowhere near expiry.""" + fresh = _jwt_expiring_in(3600) + m = _manager_with_session(fresh) + m.dal.sign_in = MagicMock() + m._last_auth_jwt = "some-older-token" + + asyncio.run(m._maybe_refresh_auth()) + + m.dal.sign_in.assert_not_called() + m._client.set_auth.assert_awaited_once_with(fresh) + + +def test_refresh_auth_attempts_proactive_sign_in_once_per_token(): + """A sign_in that keeps returning the same expiring token must not be + retried on every 60s tick.""" + expiring = _jwt_expiring_in(30) + m = _manager_with_session(expiring) + m.dal.sign_in = MagicMock() # no rotation — same token comes back + + asyncio.run(m._maybe_refresh_auth()) + asyncio.run(m._maybe_refresh_auth()) + asyncio.run(m._maybe_refresh_auth()) + + m.dal.sign_in.assert_called_once() + + +def test_refresh_auth_survives_sign_in_failure(): + """A failing re-sign-in must not escape into the _run loop and kill the + thread; the reconnect path remains the fallback.""" + m = _manager_with_session(_jwt_expiring_in(30)) + m.dal.sign_in = MagicMock(side_effect=ConnectionError("network unreachable")) + + asyncio.run(m._maybe_refresh_auth()) # must not raise + + m._client.set_auth.assert_not_awaited() + + +def test_refresh_auth_bounds_hanging_proactive_sign_in(monkeypatch): + """Same bound as _full_reconnect: a hanging sign_in must not stall the loop + (which would also stop health checks).""" + import holmes.core.conversations_worker.realtime_manager as rm + + monkeypatch.setattr(rm, "_RECONNECT_SIGN_IN_TIMEOUT_SECONDS", 0.2) + m = _manager_with_session(_jwt_expiring_in(30)) + + def hang_forever(): + time.sleep(5) + + m.dal.sign_in = hang_forever + + asyncio.run(m._maybe_refresh_auth()) # swallowed, not raised + + m._client.set_auth.assert_not_awaited() + + +# ---- reconnect log level (noise) ---- + + +def test_first_reconnect_logs_info_and_repeat_logs_warning(caplog): + """A single self-healing reconnect is routine; only a reconnect that did not + take last time round warrants a WARNING.""" + async def _scenario(): + m = _make_manager() # channel is None → unhealthy on first check + m._async_stop = asyncio.Event() + + attempts = [] + + async def fake_reconnect(): + attempts.append(1) + if len(attempts) == 1: + raise ConnectionError("first connect fails") + m._async_stop.set() + m._stop_event.set() + + m._full_reconnect = fake_reconnect # type: ignore[method-assign] + + import holmes.core.conversations_worker.realtime_manager as _rm + original = _rm.CONVERSATION_WORKER_REALTIME_RECONNECT_MAX_SECONDS + _rm.CONVERSATION_WORKER_REALTIME_RECONNECT_MAX_SECONDS = 0 + try: + await asyncio.wait_for(m._run(), timeout=5.0) + finally: + _rm.CONVERSATION_WORKER_REALTIME_RECONNECT_MAX_SECONDS = original + + with caplog.at_level(logging.INFO): + asyncio.run(_scenario()) + + unhealthy = [ + r for r in caplog.records if "Realtime channel unhealthy" in r.getMessage() + ] + assert len(unhealthy) >= 2 + assert unhealthy[0].levelno == logging.INFO + assert unhealthy[1].levelno == logging.WARNING From 99dc3be780a179a4a421d1694b331575f24c3305 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 13 Aug 2026 07:29:23 +0000 Subject: [PATCH 2/3] Drop an unverified ticket reference from the new test section header I do not have the real ticket number for this change; the ROB-4017 cited in supabase_dal.py is the RemoteProtocolError hardening, which is a different fault. Signed-off-by: Claude --- tests/core/conversations_worker/test_realtime_manager.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/core/conversations_worker/test_realtime_manager.py b/tests/core/conversations_worker/test_realtime_manager.py index 202ea92bf..7ab0a8a78 100644 --- a/tests/core/conversations_worker/test_realtime_manager.py +++ b/tests/core/conversations_worker/test_realtime_manager.py @@ -475,7 +475,7 @@ async def boom_connect(): asyncio.run(m._full_reconnect()) -# ---- _jwt_expires_within / proactive near-expiry refresh (ROB-4017) ---- +# ---- _jwt_expires_within / proactive near-expiry refresh ---- # Only the exp claim matters — _jwt_expires_within never verifies the signature. From 26edbd83036f327c96be1d8a919850edbb6f243b Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 13 Aug 2026 07:56:46 +0000 Subject: [PATCH 3/3] Cut the JWT refresh down to its minimal form Drop the asyncio.wait_for around asyncio.to_thread(dal.sign_in). wait_for does not cancel the thread, so a timeout left the sign-in running and free to mutate the DAL session after the method returned, and asyncio.run waits for the default executor at shutdown anyway (per review). The DAL's httpx client already bounds the call at 60s, so the wrapper bought nothing. Also drop the per-token attempt tracker and the new env var, shrink the helper and the docstrings, and cut the tests to the cases that matter. Net source change is now ~11 lines. Signed-off-by: Claude --- holmes/common/env_vars.py | 7 -- .../conversations_worker/realtime_manager.py | 86 ++----------- .../test_realtime_manager.py | 115 ++++-------------- 3 files changed, 35 insertions(+), 173 deletions(-) diff --git a/holmes/common/env_vars.py b/holmes/common/env_vars.py index 7fd2b9d59..7f6c4a4bc 100644 --- a/holmes/common/env_vars.py +++ b/holmes/common/env_vars.py @@ -265,13 +265,6 @@ def _load_temperature() -> Optional[float]: CONVERSATION_WORKER_AUTH_REFRESH_INTERVAL_SECONDS = float( os.environ.get("CONVERSATION_WORKER_AUTH_REFRESH_INTERVAL_SECONDS", 60) ) -# How close to its `exp` the realtime JWT may get before we proactively -# re-sign-in. Must stay comfortably above the refresh interval above so at -# least one refresh tick lands inside the window; otherwise the token expires -# unrefreshed and Supabase closes the socket with InvalidJWTToken. -CONVERSATION_WORKER_AUTH_REFRESH_LEEWAY_SECONDS = float( - os.environ.get("CONVERSATION_WORKER_AUTH_REFRESH_LEEWAY_SECONDS", 300) -) # Upper bound on how long a silently-dead realtime WebSocket can go undetected. # The realtime library can leave a stale connection in place when the server # closes the socket cleanly (ConnectionClosedOK) — _listen_task exits, no diff --git a/holmes/core/conversations_worker/realtime_manager.py b/holmes/core/conversations_worker/realtime_manager.py index e0cc2cbae..17d66432f 100644 --- a/holmes/core/conversations_worker/realtime_manager.py +++ b/holmes/core/conversations_worker/realtime_manager.py @@ -34,7 +34,6 @@ from holmes.common.env_vars import ( CONVERSATION_WORKER_AUTH_REFRESH_INTERVAL_SECONDS, - CONVERSATION_WORKER_AUTH_REFRESH_LEEWAY_SECONDS, CONVERSATION_WORKER_REALTIME_HEALTH_TICK_SECONDS, CONVERSATION_WORKER_REALTIME_RECONNECT_MAX_SECONDS, CONVERSATION_WORKER_USE_REALTIME_BROADCAST, @@ -56,28 +55,14 @@ # reconnect loop can never be stalled indefinitely by a hung auth call. _RECONNECT_SIGN_IN_TIMEOUT_SECONDS = 90 +# Must exceed the auth refresh interval so a tick lands inside it. +_AUTH_REFRESH_LEEWAY_SECONDS = 300 -def _jwt_expires_within(token: str, leeway_seconds: float) -> bool: - """True if ``token``'s ``exp`` claim is within ``leeway_seconds`` from now. - The signature is deliberately not verified: this is our own token and the - only claim we need is the expiry, which we use to decide *when* to refresh. - An unreadable token or one with no usable ``exp`` returns False — we leave - it alone and let the unhealthy→reconnect path handle it, rather than - re-signing-in on every tick against a token we can't reason about. - """ - try: - claims = jwt.decode( - token, - options={"verify_signature": False, "verify_exp": False}, - ) - except Exception: - logging.debug("Could not decode realtime JWT to read exp", exc_info=True) - return False - exp = claims.get("exp") - if not isinstance(exp, (int, float)) or isinstance(exp, bool): - return False - return exp - time.time() <= leeway_seconds +def _expires_within(token: str, seconds: float) -> bool: + # Signature is irrelevant; exp is our own claim. + exp = jwt.decode(token, options={"verify_signature": False})["exp"] + return exp - time.time() <= seconds # ---- channel topic helpers ---- @@ -260,10 +245,6 @@ def __init__( self._connected = False # Last JWT we pushed to the realtime client via set_auth. self._last_auth_jwt: Optional[str] = None - # Token we last attempted a proactive (near-expiry) re-sign-in for, so - # a sign_in that keeps handing back an expiring token is retried once - # rather than on every refresh tick. - self._proactive_refresh_attempted_for: Optional[str] = None # Set from the async loop to wake the sleep in _run() on stop(). self._async_stop: Optional[asyncio.Event] = None @@ -292,7 +273,6 @@ def start(self) -> None: self._channel = None self._connected = False self._last_auth_jwt = None - self._proactive_refresh_attempted_for = None self._async_stop = None self._thread = threading.Thread( target=self._thread_entry, @@ -359,10 +339,7 @@ async def _run(self) -> None: # own full teardown/reconnect on any failure signal. unhealthy_reason = self._channel_unhealthy() if unhealthy_reason is not None: - # A single reconnect is routine (network blip, edge - # recycling the socket) and self-healing, so it is not worth - # a WARNING — only a reconnect that did not take last time - # round (attempts > 0) says something is actually wrong. + # A first reconnect is routine and self-healing. logging.log( logging.INFO if reconnect_attempts == 0 else logging.WARNING, "Realtime channel unhealthy (%s), reconnecting", @@ -511,7 +488,6 @@ async def _full_reconnect(self) -> None: self._client = None self._channel = None self._last_auth_jwt = None - self._proactive_refresh_attempted_for = None # Bound the re-sign-in (ROB-759): sign_in is a sync HTTP call whose # timeout depends on the DAL's httpx client config; a half-open # connection there would otherwise stall this reconnect loop for the @@ -525,31 +501,7 @@ async def _full_reconnect(self) -> None: await self._connect_and_subscribe() async def _maybe_refresh_auth(self) -> None: - """Keep the realtime client's JWT fresh. - - Two paths: - - * the token rotated elsewhere (a postgrest query hit PGRST301 and the - DAL re-signed in) → re-push the new one to the client; - * the token is within ``CONVERSATION_WORKER_AUTH_REFRESH_LEEWAY_SECONDS`` - of its ``exp`` → proactively re-sign-in *before* it dies. - - The second path exists because nothing else refreshes the JWT on a - realtime-only path. The DAL refreshes reactively, from - ``patch_postgrest_execute`` on a PGRST301/expired error, and an idle - worker issues no postgrest queries — so this method used to re-read the - same unexpired-when-cached token every tick, return early on - ``new_jwt == self._last_auth_jwt``, and let it lapse. Supabase then - closed the socket (``InvalidJWTToken: Token has expired N seconds ago``), - costing an hourly reconnect plus a window in which the token was already - dead and inbound broadcasts were dropped — only the claim poll's much - slower cadence caught the work. - - A pathological token that comes back still-expiring is not retried in a - hot loop: ``_proactive_refresh_attempted_for`` bounds us to one attempt - per distinct token, leaving the unhealthy→``_full_reconnect`` path as the - safety net it already was. - """ + """Re-push the Supabase JWT, re-signing in first if it is near expiry.""" if not self._client: return try: @@ -557,25 +509,9 @@ async def _maybe_refresh_auth(self) -> None: if session is None: return new_jwt = session.access_token - if ( - new_jwt - and new_jwt != self._proactive_refresh_attempted_for - and _jwt_expires_within( - new_jwt, CONVERSATION_WORKER_AUTH_REFRESH_LEEWAY_SECONDS - ) - ): - logging.info( - "Realtime JWT expires within %ss, re-signing in to Supabase", - CONVERSATION_WORKER_AUTH_REFRESH_LEEWAY_SECONDS, - ) - self._proactive_refresh_attempted_for = new_jwt - # Same bounded-sign-in pattern as _full_reconnect (ROB-759): - # sign_in is a sync HTTP call, so a half-open connection there - # would otherwise stall this loop and stop health checks too. - await asyncio.wait_for( - asyncio.to_thread(self.dal.sign_in), - timeout=_RECONNECT_SIGN_IN_TIMEOUT_SECONDS, - ) + if new_jwt and _expires_within(new_jwt, _AUTH_REFRESH_LEEWAY_SECONDS): + # Nothing else refreshes the JWT on a realtime-only path. + await asyncio.to_thread(self.dal.sign_in) session = self.dal.client.auth.get_session() # type: ignore[attr-defined] new_jwt = session.access_token if session is not None else None if not new_jwt or new_jwt == self._last_auth_jwt: diff --git a/tests/core/conversations_worker/test_realtime_manager.py b/tests/core/conversations_worker/test_realtime_manager.py index 7ab0a8a78..c2fe8fd75 100644 --- a/tests/core/conversations_worker/test_realtime_manager.py +++ b/tests/core/conversations_worker/test_realtime_manager.py @@ -15,9 +15,9 @@ from holmes.core.conversations_worker.realtime_manager import ( RealtimeWorker, _build_ssl_context, + _expires_within, _install_realtime_log_filter_if_needed, _install_ssl_patch_if_needed, - _jwt_expires_within, _RealtimeConnectivityWarningFilter, broadcast_submit_topic, pg_changes_topic, @@ -475,43 +475,11 @@ async def boom_connect(): asyncio.run(m._full_reconnect()) -# ---- _jwt_expires_within / proactive near-expiry refresh ---- +# ---- proactive near-expiry auth refresh ---- -# Only the exp claim matters — _jwt_expires_within never verifies the signature. -# Long enough to keep PyJWT's InsecureKeyLengthWarning out of the test output. -_TEST_JWT_KEY = "x" * 32 - - -def _jwt_expiring_in(seconds: float) -> str: - """Mint a JWT whose exp is `seconds` from now.""" - return jwt.encode( - {"exp": int(time.time() + seconds)}, _TEST_JWT_KEY, algorithm="HS256" - ) - - -def test_jwt_expires_within_detects_near_expiry(): - assert _jwt_expires_within(_jwt_expiring_in(60), 300) is True - assert _jwt_expires_within(_jwt_expiring_in(-10), 300) is True # already expired - - -def test_jwt_expires_within_ignores_fresh_token(): - assert _jwt_expires_within(_jwt_expiring_in(3600), 300) is False - - -@pytest.mark.parametrize( - "token", - [ - "not-a-jwt", - "", - jwt.encode({"sub": "no-exp-claim"}, _TEST_JWT_KEY, algorithm="HS256"), - jwt.encode({"exp": "not-a-number"}, _TEST_JWT_KEY, algorithm="HS256"), - ], -) -def test_jwt_expires_within_is_false_for_unreadable_tokens(token): - """An undecodable or exp-less token must not trigger a re-sign-in on every - tick — the unhealthy/reconnect path stays the safety net.""" - assert _jwt_expires_within(token, 300) is False +def _token(expires_in: float) -> str: + return jwt.encode({"exp": int(time.time() + expires_in)}, "k" * 32, algorithm="HS256") def _manager_with_session(token): @@ -524,15 +492,19 @@ def _manager_with_session(token): return m +def test_expires_within(): + assert _expires_within(_token(60), 300) is True + assert _expires_within(_token(-10), 300) is True + assert _expires_within(_token(3600), 300) is False + + def test_refresh_auth_re_signs_in_when_token_near_expiry(): - """The bug: nothing refreshed the JWT on a realtime-only path, so the token - lapsed and Supabase closed the socket with InvalidJWTToken.""" - expiring = _jwt_expiring_in(30) - fresh = _jwt_expiring_in(3600) - m = _manager_with_session(expiring) + """The bug: nothing refreshed the JWT on a realtime-only path, so it lapsed + and Supabase closed the socket with InvalidJWTToken.""" + fresh = _token(3600) + m = _manager_with_session(_token(30)) def sign_in(): - # Post-sign-in, get_session returns the rotated token. rotated = MagicMock() rotated.access_token = fresh m.dal.client.auth.get_session = MagicMock(return_value=rotated) @@ -547,7 +519,7 @@ def sign_in(): def test_refresh_auth_leaves_fresh_token_alone(): - fresh = _jwt_expiring_in(3600) + fresh = _token(3600) m = _manager_with_session(fresh) m.dal.sign_in = MagicMock() m._last_auth_jwt = fresh @@ -559,12 +531,10 @@ def test_refresh_auth_leaves_fresh_token_alone(): def test_refresh_auth_still_pushes_externally_rotated_token(): - """Pre-existing behaviour: a token rotated by the DAL's PGRST301 path is - re-pushed even though it is nowhere near expiry.""" - fresh = _jwt_expiring_in(3600) + fresh = _token(3600) m = _manager_with_session(fresh) m.dal.sign_in = MagicMock() - m._last_auth_jwt = "some-older-token" + m._last_auth_jwt = "older-token" asyncio.run(m._maybe_refresh_auth()) @@ -572,59 +542,24 @@ def test_refresh_auth_still_pushes_externally_rotated_token(): m._client.set_auth.assert_awaited_once_with(fresh) -def test_refresh_auth_attempts_proactive_sign_in_once_per_token(): - """A sign_in that keeps returning the same expiring token must not be - retried on every 60s tick.""" - expiring = _jwt_expiring_in(30) - m = _manager_with_session(expiring) - m.dal.sign_in = MagicMock() # no rotation — same token comes back - - asyncio.run(m._maybe_refresh_auth()) - asyncio.run(m._maybe_refresh_auth()) - asyncio.run(m._maybe_refresh_auth()) - - m.dal.sign_in.assert_called_once() - - def test_refresh_auth_survives_sign_in_failure(): - """A failing re-sign-in must not escape into the _run loop and kill the - thread; the reconnect path remains the fallback.""" - m = _manager_with_session(_jwt_expiring_in(30)) + """A failure must not escape into _run and kill the thread; the reconnect + path stays the fallback.""" + m = _manager_with_session(_token(30)) m.dal.sign_in = MagicMock(side_effect=ConnectionError("network unreachable")) - asyncio.run(m._maybe_refresh_auth()) # must not raise - - m._client.set_auth.assert_not_awaited() - - -def test_refresh_auth_bounds_hanging_proactive_sign_in(monkeypatch): - """Same bound as _full_reconnect: a hanging sign_in must not stall the loop - (which would also stop health checks).""" - import holmes.core.conversations_worker.realtime_manager as rm - - monkeypatch.setattr(rm, "_RECONNECT_SIGN_IN_TIMEOUT_SECONDS", 0.2) - m = _manager_with_session(_jwt_expiring_in(30)) - - def hang_forever(): - time.sleep(5) - - m.dal.sign_in = hang_forever - - asyncio.run(m._maybe_refresh_auth()) # swallowed, not raised + asyncio.run(m._maybe_refresh_auth()) m._client.set_auth.assert_not_awaited() -# ---- reconnect log level (noise) ---- +# ---- reconnect log level ---- def test_first_reconnect_logs_info_and_repeat_logs_warning(caplog): - """A single self-healing reconnect is routine; only a reconnect that did not - take last time round warrants a WARNING.""" async def _scenario(): - m = _make_manager() # channel is None → unhealthy on first check + m = _make_manager() # channel is None -> unhealthy on first check m._async_stop = asyncio.Event() - attempts = [] async def fake_reconnect(): @@ -647,9 +582,7 @@ async def fake_reconnect(): with caplog.at_level(logging.INFO): asyncio.run(_scenario()) - unhealthy = [ - r for r in caplog.records if "Realtime channel unhealthy" in r.getMessage() - ] + unhealthy = [r for r in caplog.records if "Realtime channel unhealthy" in r.getMessage()] assert len(unhealthy) >= 2 assert unhealthy[0].levelno == logging.INFO assert unhealthy[1].levelno == logging.WARNING