diff --git a/plugins/platforms/photon/adapter.py b/plugins/platforms/photon/adapter.py index 219c2f2c1dca..0f263dabaa16 100644 --- a/plugins/platforms/photon/adapter.py +++ b/plugins/platforms/photon/adapter.py @@ -311,6 +311,33 @@ def sidecar_deps_installed() -> bool: return (_SIDECAR_DIR / "node_modules" / "spectrum-ts").exists() +def _coerce_float(value: Any, default: float) -> float: + try: + return float(value) + except (TypeError, ValueError): + return default + + +def _coerce_int(value: Any, default: int) -> int: + try: + return int(value) + except (TypeError, ValueError): + return default + + +def _first_set(*values: Any) -> Any: + """Return the first value that is not None. + + Unlike ``a or b``, this preserves an explicit falsy value (e.g. ``0`` or + ``""``) so config can intentionally set a zero/empty without it silently + falling through to a later source or a default. + """ + for v in values: + if v is not None: + return v + return None + + def check_requirements() -> bool: """Return True when both Python deps and the Node sidecar are available.""" if not HTTPX_AVAILABLE: @@ -638,6 +665,42 @@ class PhotonAdapter(BasePlatformAdapter): ).lower() not in ("0", "false", "no") self._node_bin = os.getenv("PHOTON_NODE_BIN") or shutil.which("node") or "node" + # Presence watchdog. spectrum-ts only reconnects when its inbound + # iterator throws or ends; a half-open ("zombie") gRPC socket makes the + # iterator hang forever (no error, no end), so inbound silently dies + # until the sidecar is restarted. We periodically drive a cheap upstream + # unary read via the sidecar's /probe endpoint; repeated failures mean + # the stream is dead -> respawn the sidecar (fresh Spectrum() = fresh + # gRPC stream). A successful probe is also a real round-trip that keeps + # the channel warm, helping prevent the zombie forming at all. + # Behavioural settings -> config.yaml (extra), bridged to env. + # Use _first_set (not ``or``) so an explicit 0 is honored — ``0 or X`` + # would silently fall through to the default and you could never + # disable the watchdog with probe_interval_seconds: 0. + self._probe_interval = _coerce_float( + _first_set( + extra.get("probe_interval_seconds"), + os.getenv("PHOTON_PROBE_INTERVAL_SECONDS"), + ), + 60.0, + ) + self._probe_timeout = _coerce_float( + _first_set( + extra.get("probe_timeout_seconds"), + os.getenv("PHOTON_PROBE_TIMEOUT_SECONDS"), + ), + 10.0, + ) + self._probe_max_failures = _coerce_int( + _first_set( + extra.get("probe_max_failures"), + os.getenv("PHOTON_PROBE_MAX_FAILURES"), + ), + 3, + ) + # A non-positive interval disables the watchdog entirely (escape hatch). + self._probe_enabled = self._probe_interval > 0 + # With markdown on, format_message preserves fences and the sidecar's # markdown() builder renders them (or degrades them readably). self.supports_code_blocks = _markdown_enabled() @@ -650,6 +713,15 @@ class PhotonAdapter(BasePlatformAdapter): self._inbound_running = False self._http_client: Optional["httpx.AsyncClient"] = None self._sidecar_health_interval = 15.0 + # Presence-watchdog runtime state. + self._watchdog_task: Optional[asyncio.Task] = None + self._watchdog_running = False + self._probe_failures = 0 + # Monotonic timestamp of the last real upstream activity (inbound + # message or a successful probe). The watchdog skips its own probe when + # natural traffic already proved the channel live within the interval. + self._last_upstream_activity = 0.0 + self._respawn_lock: Optional[asyncio.Lock] = None # Lightweight in-memory dedup. The gRPC stream is at-least-once, so we # may see the same messageId more than once (e.g. after a reconnect). self._seen_messages: Dict[str, float] = {} @@ -796,6 +868,16 @@ class PhotonAdapter(BasePlatformAdapter): self._monitor_sidecar_health() ) + # Start the presence watchdog (detects a half-open/zombie gRPC stream + # the SDK can't see and respawns the sidecar to recover inbound). + self._last_upstream_activity = time.monotonic() + if self._probe_enabled and self._autostart_sidecar: + self._respawn_lock = asyncio.Lock() + self._watchdog_running = True + self._watchdog_task = asyncio.get_event_loop().create_task( + self._presence_watchdog() + ) + self._mark_connected() logger.info( "[photon] connected — sidecar on %s:%d, streaming inbound over gRPC", @@ -805,6 +887,9 @@ class PhotonAdapter(BasePlatformAdapter): async def disconnect(self) -> None: self._inbound_running = False + # Stop the watchdog first so it can't trigger a respawn while we tear + # the sidecar down. + await self._stop_watchdog() if self._sidecar_health_task is not None: task = self._sidecar_health_task self._sidecar_health_task = None @@ -965,6 +1050,12 @@ class PhotonAdapter(BasePlatformAdapter): break async def _on_inbound_line(self, line: str) -> None: + # Any line on the inbound stream — message OR heartbeat — proves the + # upstream gRPC channel is live, so reset the watchdog's failure count + # and activity clock. (Heartbeats arrive as blank lines, already + # filtered before this method, but a real message is the strongest + # liveness signal we have.) + self._note_upstream_activity() try: event = json.loads(line) except json.JSONDecodeError: @@ -1611,6 +1702,126 @@ class PhotonAdapter(BasePlatformAdapter): self._sidecar_supervisor_task.cancel() self._sidecar_supervisor_task = None + # -- Presence watchdog ------------------------------------------------- + + def _note_upstream_activity(self) -> None: + """Record proof the upstream gRPC channel is live, and clear failures. + + Called on every inbound line and after every successful probe. + """ + self._last_upstream_activity = time.monotonic() + self._probe_failures = 0 + + async def _probe_once(self) -> bool: + """Drive one upstream liveness probe via the sidecar's ``/probe``. + + Returns True if the sidecar confirmed a live gRPC round-trip within + the timeout, False on timeout/error (a likely zombie stream). + """ + client = self._http_client + if client is None: + return False + url = f"http://{self._sidecar_bind}:{self._sidecar_port}/probe" + try: + resp = await client.post( + url, + headers={"X-Hermes-Sidecar-Token": self._sidecar_token}, + timeout=self._probe_timeout, + ) + except Exception as e: + logger.debug("[photon] probe transport error: %s", e) + return False + return resp.status_code == 200 + + async def _respawn_sidecar(self, reason: str) -> None: + """Tear down and restart the sidecar to recover a dead gRPC stream. + + A fresh ``Spectrum()`` re-subscribes the inbound stream and + re-registers presence with Photon cloud. The inbound loop's existing + reconnect logic re-opens the loopback NDJSON stream automatically once + the new sidecar's ``/inbound`` is up. Guarded by a lock so overlapping + triggers (watchdog + a manual call) can't double-spawn. + """ + lock = self._respawn_lock + if lock is None: + lock = self._respawn_lock = asyncio.Lock() + if lock.locked(): + logger.info("[photon] respawn already in progress; skipping") + return + async with lock: + logger.warning( + "[photon] presence watchdog: %s — respawning sidecar", reason + ) + try: + await self._stop_sidecar() + except Exception: + logger.exception("[photon] error stopping sidecar during respawn") + try: + await self._start_sidecar() + except Exception: + logger.exception( + "[photon] failed to respawn sidecar; watchdog will retry" + ) + return + # Fresh sidecar -> fresh stream. Reset the activity clock so we give + # it a full interval before probing again, and clear failures. + self._note_upstream_activity() + logger.info("[photon] presence watchdog: sidecar respawned, gRPC stream renewed") + + async def _presence_watchdog(self) -> None: + """Periodically confirm the upstream gRPC stream is alive. + + spectrum-ts only recovers when its inbound iterator throws or ends; a + half-open ("zombie") socket makes it hang forever with no error, so + inbound silently dies. We probe the channel on an interval (skipping + the probe when natural inbound traffic already proved liveness within + the window). After ``_probe_max_failures`` consecutive failed probes we + respawn the sidecar to force a fresh stream. + """ + # Stagger the first probe so a fleet of restarts doesn't synchronize, + # and so a freshly-started sidecar isn't probed before it's warm. + await asyncio.sleep(self._probe_interval) + while self._watchdog_running: + try: + # Skip our own probe if inbound traffic already proved the + # channel live within the last interval (cheaper, and avoids + # piling synthetic reads on a busy line). + idle = time.monotonic() - self._last_upstream_activity + if idle < self._probe_interval: + await asyncio.sleep(self._probe_interval - idle) + continue + + alive = await self._probe_once() + if alive: + self._note_upstream_activity() + else: + self._probe_failures += 1 + logger.warning( + "[photon] presence probe failed (%d/%d)", + self._probe_failures, self._probe_max_failures, + ) + if self._probe_failures >= self._probe_max_failures: + await self._respawn_sidecar( + f"{self._probe_failures} consecutive probe failures" + ) + except asyncio.CancelledError: + raise + except Exception: + logger.exception("[photon] presence watchdog iteration failed") + await asyncio.sleep(self._probe_interval) + + async def _stop_watchdog(self) -> None: + self._watchdog_running = False + if self._watchdog_task is not None: + self._watchdog_task.cancel() + try: + await self._watchdog_task + except asyncio.CancelledError: + pass + except Exception: + pass + self._watchdog_task = None + # -- Outbound ---------------------------------------------------------- async def send( diff --git a/plugins/platforms/photon/sidecar/index.mjs b/plugins/platforms/photon/sidecar/index.mjs index 1bc9ac96e9fd..b63cbba7b471 100644 --- a/plugins/platforms/photon/sidecar/index.mjs +++ b/plugins/platforms/photon/sidecar/index.mjs @@ -211,6 +211,14 @@ console.log = (...args) => { originalConsoleLog(...args); }; +// Upstream liveness probe (see `/probe` handler). A synthetic DM space id + +// a unique-per-probe bogus message id drive a cheap unary read over the same +// gRPC channel the inbound stream uses, so we can tell a live channel from a +// half-open ("zombie") one. `space.get` is purely local in shared/dedicated +// mode (no chat is created or messaged); only the message read hits the wire. +const PROBE_SPACE_ID = process.env.PHOTON_PROBE_SPACE_ID || "any;-;+10000000000"; +const PROBE_MSG_PREFIX = "hermes-liveness-probe-"; + if (!projectId || !projectSecret || !sharedToken) { console.error( "photon-sidecar: PHOTON_PROJECT_ID, PHOTON_PROJECT_SECRET and " + @@ -831,6 +839,40 @@ const server = http.createServer(async (req, res) => { if (req.url === "/healthz") { return ok(res, { stream: streamHealthSnapshot() }); } + if (req.url === "/probe") { + // Upstream liveness probe. Drives a cheap unary read over the SAME gRPC + // channel the inbound stream uses. On a live channel this round-trips + // fast (the server returns "not found" for the synthetic id, which the + // SDK surfaces as a thrown error — that's a SUCCESS for our purposes: + // the wire is alive). On a half-open zombie socket the call hangs and + // the caller's timeout fires, so the adapter can respawn us. It sends + // nothing to any user and creates no chat (space.get is local; only the + // message read touches the network). + if (typeof app?.stop !== "function") { + // app failed to construct — definitely not live. + return serverError(res); + } + try { + const im = imessage(app); + const space = await im.space.get(PROBE_SPACE_ID); + const probeId = PROBE_MSG_PREFIX + Date.now() + "-" + Math.random().toString(36).slice(2); + try { + await space.getMessage(probeId); + } catch { + // Expected: the synthetic id doesn't exist. Reaching here means the + // unary call completed a round-trip — the channel is ALIVE. + } + return ok(res, { alive: true }); + } catch (e) { + // space.get is local; an error here is unexpected but means we can't + // even build a probe — report not-alive so the adapter recycles us. + console.error( + "photon-sidecar: probe failed: " + + (e && e.message ? e.message : String(e)) + ); + return serverError(res); + } + } if (req.url === "/shutdown") { ok(res, {}); setTimeout(() => process.kill(process.pid, "SIGTERM"), 50); diff --git a/tests/plugins/platforms/photon/test_presence_watchdog.py b/tests/plugins/platforms/photon/test_presence_watchdog.py new file mode 100644 index 000000000000..bbe04e287a26 --- /dev/null +++ b/tests/plugins/platforms/photon/test_presence_watchdog.py @@ -0,0 +1,231 @@ +"""Presence-watchdog tests. + +spectrum-ts only reconnects when its inbound iterator throws or ends; a +half-open ("zombie") gRPC socket makes the iterator hang forever (no error, no +end), so inbound silently dies until the sidecar is restarted. The adapter's +presence watchdog probes the upstream channel via the sidecar's ``/probe`` +endpoint and respawns the sidecar after repeated probe failures. + +These tests exercise the watchdog's decision logic (probe -> count failures -> +respawn; success resets; recent inbound traffic skips the probe) without +spawning Node, binding ports, or hitting the network. +""" +from __future__ import annotations + +import time +from typing import Any, List + +import pytest + +from gateway.config import PlatformConfig +from plugins.platforms.photon.adapter import PhotonAdapter + + +def _make_adapter(monkeypatch: pytest.MonkeyPatch, **extra: Any) -> PhotonAdapter: + monkeypatch.setenv("PHOTON_PROJECT_ID", "test-project-id") + monkeypatch.setenv("PHOTON_PROJECT_SECRET", "test-project-secret") + cfg = PlatformConfig(enabled=True, token="", extra=dict(extra)) + return PhotonAdapter(cfg) + + +def test_probe_config_defaults(monkeypatch: pytest.MonkeyPatch) -> None: + a = _make_adapter(monkeypatch) + assert a._probe_interval == 60.0 + assert a._probe_timeout == 10.0 + assert a._probe_max_failures == 3 + assert a._probe_enabled is True + + +def test_probe_config_from_extra(monkeypatch: pytest.MonkeyPatch) -> None: + a = _make_adapter( + monkeypatch, + probe_interval_seconds=30, + probe_timeout_seconds=5, + probe_max_failures=2, + ) + assert a._probe_interval == 30.0 + assert a._probe_timeout == 5.0 + assert a._probe_max_failures == 2 + assert a._probe_enabled is True + + +def test_probe_disabled_when_interval_nonpositive( + monkeypatch: pytest.MonkeyPatch, +) -> None: + a = _make_adapter(monkeypatch, probe_interval_seconds=0) + assert a._probe_enabled is False + + +def test_note_activity_resets_failures(monkeypatch: pytest.MonkeyPatch) -> None: + a = _make_adapter(monkeypatch) + a._probe_failures = 2 + before = a._last_upstream_activity + time.sleep(0.001) + a._note_upstream_activity() + assert a._probe_failures == 0 + assert a._last_upstream_activity > before + + +@pytest.mark.asyncio +async def test_probe_once_alive_on_200(monkeypatch: pytest.MonkeyPatch) -> None: + a = _make_adapter(monkeypatch) + + class _Resp: + status_code = 200 + + class _Client: + async def post(self, *args: Any, **kwargs: Any) -> Any: + return _Resp() + + a._http_client = _Client() # type: ignore[assignment] + assert await a._probe_once() is True + + +@pytest.mark.asyncio +async def test_probe_once_dead_on_500(monkeypatch: pytest.MonkeyPatch) -> None: + a = _make_adapter(monkeypatch) + + class _Resp: + status_code = 500 + + class _Client: + async def post(self, *args: Any, **kwargs: Any) -> Any: + return _Resp() + + a._http_client = _Client() # type: ignore[assignment] + assert await a._probe_once() is False + + +@pytest.mark.asyncio +async def test_probe_once_dead_on_timeout(monkeypatch: pytest.MonkeyPatch) -> None: + a = _make_adapter(monkeypatch) + + class _Client: + async def post(self, *args: Any, **kwargs: Any) -> Any: + raise TimeoutError("zombie stream — probe hung") + + a._http_client = _Client() # type: ignore[assignment] + # A hung/half-open probe must read as NOT alive (this is the zombie case). + assert await a._probe_once() is False + + +@pytest.mark.asyncio +async def test_probe_once_false_without_client( + monkeypatch: pytest.MonkeyPatch, +) -> None: + a = _make_adapter(monkeypatch) + a._http_client = None + assert await a._probe_once() is False + + +@pytest.mark.asyncio +async def test_respawn_after_max_failures(monkeypatch: pytest.MonkeyPatch) -> None: + """The core fix: N consecutive dead probes -> exactly one respawn.""" + a = _make_adapter(monkeypatch, probe_max_failures=3) + + respawns: List[str] = [] + + async def _fake_respawn(reason: str) -> None: + respawns.append(reason) + a._note_upstream_activity() # mirror real respawn (clears failures) + + async def _dead_probe() -> bool: + return False + + monkeypatch.setattr(a, "_respawn_sidecar", _fake_respawn) + monkeypatch.setattr(a, "_probe_once", _dead_probe) + + # Simulate the watchdog's per-iteration decision logic directly (no sleeps). + a._last_upstream_activity = time.monotonic() - 999 # force a probe each time + for _ in range(3): + alive = await a._probe_once() + assert alive is False + a._probe_failures += 1 + if a._probe_failures >= a._probe_max_failures: + await a._respawn_sidecar("test") + + assert respawns == ["test"] + assert a._probe_failures == 0 # reset by the (faked) respawn + + +@pytest.mark.asyncio +async def test_success_resets_failure_count( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A live probe between dead ones prevents a respawn (failures reset).""" + a = _make_adapter(monkeypatch, probe_max_failures=3) + + respawns: List[str] = [] + + async def _fake_respawn(reason: str) -> None: + respawns.append(reason) + + monkeypatch.setattr(a, "_respawn_sidecar", _fake_respawn) + + # Two failures, then a success, then two more failures: never hits 3 in a row. + sequence = [False, False, True, False, False] + for alive in sequence: + if alive: + a._note_upstream_activity() + else: + a._probe_failures += 1 + if a._probe_failures >= a._probe_max_failures: + await a._respawn_sidecar("should-not-fire") + + assert respawns == [] + assert a._probe_failures == 2 + + +@pytest.mark.asyncio +async def test_respawn_calls_stop_then_start( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Respawn tears the sidecar down and brings a fresh one up, in order.""" + a = _make_adapter(monkeypatch) + calls: List[str] = [] + + async def _fake_stop() -> None: + calls.append("stop") + + async def _fake_start() -> None: + calls.append("start") + + monkeypatch.setattr(a, "_stop_sidecar", _fake_stop) + monkeypatch.setattr(a, "_start_sidecar", _fake_start) + + await a._respawn_sidecar("test reason") + assert calls == ["stop", "start"] + # A fresh stream means failures are cleared. + assert a._probe_failures == 0 + + +@pytest.mark.asyncio +async def test_respawn_is_lock_guarded(monkeypatch: pytest.MonkeyPatch) -> None: + """A second respawn while one is in flight is skipped, not double-spawned.""" + import asyncio + + a = _make_adapter(monkeypatch) + starts: List[str] = [] + + started = asyncio.Event() + release = asyncio.Event() + + async def _slow_stop() -> None: + started.set() + await release.wait() + + async def _fake_start() -> None: + starts.append("start") + + monkeypatch.setattr(a, "_stop_sidecar", _slow_stop) + monkeypatch.setattr(a, "_start_sidecar", _fake_start) + a._respawn_lock = asyncio.Lock() + + first = asyncio.create_task(a._respawn_sidecar("first")) + await started.wait() # first respawn is now holding the lock inside _stop + # Second call should see the lock held and return immediately. + await a._respawn_sidecar("second") + assert starts == [] # second did not proceed to start + release.set() + await first + assert starts == ["start"] # only the first respawn started a new sidecar