diff --git a/contributors/emails/nickkarhan@users.noreply.github.com b/contributors/emails/nickkarhan@users.noreply.github.com new file mode 100644 index 000000000000..6b5e3a846be7 --- /dev/null +++ b/contributors/emails/nickkarhan@users.noreply.github.com @@ -0,0 +1,2 @@ +nickkarhan +# PR #53283 salvage diff --git a/plugins/platforms/photon/adapter.py b/plugins/platforms/photon/adapter.py index 0f263dabaa16..597327f1e385 100644 --- a/plugins/platforms/photon/adapter.py +++ b/plugins/platforms/photon/adapter.py @@ -338,6 +338,17 @@ def _first_set(*values: Any) -> Any: return None +def _is_timeout_error(exc: BaseException) -> bool: + """True when *exc* indicates the request timed out (call hung).""" + if isinstance(exc, (asyncio.TimeoutError, TimeoutError)): + return True + if HTTPX_AVAILABLE: + timeout_exc = getattr(httpx, "TimeoutException", None) + if timeout_exc is not None and isinstance(exc, timeout_exc): + return True + return "timeout" in type(exc).__name__.lower() + + def check_requirements() -> bool: """Return True when both Python deps and the Node sidecar are available.""" if not HTTPX_AVAILABLE: @@ -668,11 +679,17 @@ class PhotonAdapter(BasePlatformAdapter): # 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. + # until the sidecar is restarted. The sidecar owns primary zombie + # detection (stream-staleness + upstream probe -> degraded -> exit 75; + # surfaced here via /healthz in _monitor_sidecar_health). This adapter- + # side watchdog is a conservative second layer that only respawns the + # sidecar when the sidecar's own HTTP loop stops responding (probe + # HTTP call hangs) — an "inconclusive" probe (sidecar answered but + # could not prove upstream liveness) NEVER counts toward a respawn: + # the network may simply be down, and restarting cannot fix that. + # Thresholds are deliberately conservative (10 min interval) — shared + # lines can be legitimately quiet for hours, so we never restart on + # silence alone. # 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 @@ -682,7 +699,7 @@ class PhotonAdapter(BasePlatformAdapter): extra.get("probe_interval_seconds"), os.getenv("PHOTON_PROBE_INTERVAL_SECONDS"), ), - 60.0, + 600.0, ) self._probe_timeout = _coerce_float( _first_set( @@ -1029,7 +1046,26 @@ class PhotonAdapter(BasePlatformAdapter): continue stream = data.get("stream") if isinstance(data, dict) else None - if not isinstance(stream, dict) or stream.get("ok") is not False: + if not isinstance(stream, dict): + continue + + # Surface the sidecar's zombie-stream staleness state early: a + # suspected half-open stream (silence past threshold + probe-proven + # connectivity) is worth a loud log line even before the sidecar's + # degraded->exit-75 path fires. + staleness = stream.get("staleness") + if ( + isinstance(staleness, dict) + and staleness.get("zombieSuspected") is True + ): + logger.warning( + "[photon] sidecar reports suspected zombie stream" + " (silentForMs=%s, lastProbeOutcome=%s)", + staleness.get("silentForMs"), + staleness.get("lastProbeOutcome"), + ) + + if stream.get("ok") is not False: continue state = str(stream.get("state") or "unknown") @@ -1712,15 +1748,24 @@ class PhotonAdapter(BasePlatformAdapter): 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``. + async def _probe_once(self) -> str: + """Drive one 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). + Returns a strict tri-state verdict: + + - ``"alive"`` — the sidecar completed a real upstream gRPC + round-trip (HTTP 200). Only this counts as proof of liveness. + - ``"hung"`` — the probe HTTP call itself timed out: the + sidecar's event loop is unresponsive. Counts toward respawn. + - ``"inconclusive"`` — anything else (503 from the sidecar's strict + probe, connection refused, transport error). NEVER counts as alive + and NEVER counts toward a respawn: a rejected upstream probe may + just mean the network is down, and a dead sidecar process is the + supervisor's job, not ours. """ client = self._http_client if client is None: - return False + return "inconclusive" url = f"http://{self._sidecar_bind}:{self._sidecar_port}/probe" try: resp = await client.post( @@ -1728,10 +1773,15 @@ class PhotonAdapter(BasePlatformAdapter): headers={"X-Hermes-Sidecar-Token": self._sidecar_token}, timeout=self._probe_timeout, ) + except asyncio.CancelledError: + raise except Exception as e: - logger.debug("[photon] probe transport error: %s", e) - return False - return resp.status_code == 200 + if _is_timeout_error(e): + logger.debug("[photon] probe HTTP call hung: %s", e) + return "hung" + logger.debug("[photon] probe transport error (inconclusive): %s", e) + return "inconclusive" + return "alive" if resp.status_code == 200 else "inconclusive" async def _respawn_sidecar(self, reason: str) -> None: """Tear down and restart the sidecar to recover a dead gRPC stream. @@ -1769,14 +1819,18 @@ class PhotonAdapter(BasePlatformAdapter): logger.info("[photon] presence watchdog: sidecar respawned, gRPC stream renewed") async def _presence_watchdog(self) -> None: - """Periodically confirm the upstream gRPC stream is alive. + """Periodically confirm the sidecar (and its stream) is responsive. - 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. + The sidecar owns primary zombie-stream detection (silence tracking + + upstream probe -> degraded /healthz -> exit 75). This loop is the + adapter's conservative second layer: it probes on a long interval, + skips the probe when natural inbound traffic already proved liveness, + and only counts a *hung* probe (the sidecar HTTP call itself timing + out) toward a respawn. Inconclusive probes — the sidecar answered but + couldn't prove upstream liveness — reset nothing and trigger nothing: + the network may just be down, and restarting can't fix that. After + ``_probe_max_failures`` consecutive hung 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. @@ -1791,19 +1845,26 @@ class PhotonAdapter(BasePlatformAdapter): await asyncio.sleep(self._probe_interval - idle) continue - alive = await self._probe_once() - if alive: + verdict = await self._probe_once() + if verdict == "alive": self._note_upstream_activity() - else: + elif verdict == "hung": + # Only a hung sidecar HTTP loop counts toward respawn — + # strict success-only-on-probe-OK semantics mean an + # inconclusive probe is never evidence in either direction. self._probe_failures += 1 logger.warning( - "[photon] presence probe failed (%d/%d)", + "[photon] presence probe hung (%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" + f"{self._probe_failures} consecutive hung probes" ) + else: + logger.debug( + "[photon] presence probe inconclusive; taking no action" + ) except asyncio.CancelledError: raise except Exception: diff --git a/plugins/platforms/photon/sidecar/index.mjs b/plugins/platforms/photon/sidecar/index.mjs index 5ea99ce0a5fb..5c8611f30b00 100644 --- a/plugins/platforms/photon/sidecar/index.mjs +++ b/plugins/platforms/photon/sidecar/index.mjs @@ -67,6 +67,11 @@ import crypto from "node:crypto"; import { once } from "node:events"; import { patchSpectrumTs } from "./patch-spectrum-mixed-attachments.mjs"; import { chooseSendFormat } from "./send-format.mjs"; +import { + classifyProbeRejection, + shouldProbe, + isZombieSuspect, +} from "./stream-staleness.mjs"; const projectId = process.env.PHOTON_PROJECT_ID; const projectSecret = process.env.PHOTON_PROJECT_SECRET; @@ -93,6 +98,22 @@ const STREAM_DEGRADED_RESTART_MS = Number(process.env.PHOTON_STREAM_DEGRADED_RESTART_MS) || 90 * 1000; const STREAM_INTERRUPTED_DEGRADE_COUNT = Number(process.env.PHOTON_STREAM_INTERRUPTED_DEGRADE_COUNT) || 3; +// Zombie-stream (half-open gRPC) watchdog. The inbound iterator can hang +// forever without erroring; we track when it last yielded and, once it has +// been silent past this threshold, drive a cheap authenticated probe. Silence +// alone NEVER degrades the stream (shared lines can be quiet for hours) and a +// rejected probe is INCONCLUSIVE, never proof either way — degradation +// requires silence past the threshold plus a probe-proven live channel. +// A non-positive threshold disables the watchdog. +const STREAM_SILENCE_PROBE_MS = (() => { + const raw = Number(process.env.PHOTON_STREAM_SILENCE_PROBE_MS); + return Number.isFinite(raw) ? raw : 10 * 60 * 1000; +})(); +const STREAM_PROBE_COOLDOWN_MS = + Number(process.env.PHOTON_STREAM_PROBE_COOLDOWN_MS) || 2 * 60 * 1000; +const STREAM_PROBE_TIMEOUT_MS = + Number(process.env.PHOTON_STREAM_PROBE_TIMEOUT_MS) || 10 * 1000; +const STREAM_WATCHDOG_TICK_MS = 30 * 1000; const streamHealth = { state: "starting", @@ -104,6 +125,33 @@ const streamHealth = { }; let streamRestartTimer = null; +// Zombie-watchdog runtime state (see stream-staleness.mjs for the rules). +const staleness = { + lastInboundAt: Date.now(), + lastProbeAt: 0, + lastProbeOutcome: null, // "alive" | "inconclusive" | null + zombieSuspected: false, +}; + +function noteInboundYield() { + staleness.lastInboundAt = Date.now(); + staleness.zombieSuspected = false; +} + +function stalenessSnapshot(now) { + return { + lastInboundAt: new Date(staleness.lastInboundAt).toISOString(), + silentForMs: now - staleness.lastInboundAt, + silenceThresholdMs: STREAM_SILENCE_PROBE_MS, + lastProbeAt: + staleness.lastProbeAt > 0 + ? new Date(staleness.lastProbeAt).toISOString() + : null, + lastProbeOutcome: staleness.lastProbeOutcome, + zombieSuspected: staleness.zombieSuspected, + }; +} + function streamHealthSnapshot() { const now = Date.now(); const degradedForMs = @@ -117,6 +165,7 @@ function streamHealthSnapshot() { lastIssueAt: streamHealth.lastIssueAt, lastIssue: streamHealth.lastIssue, issueCount: streamHealth.issueCount, + staleness: stalenessSnapshot(now), }; } @@ -599,6 +648,8 @@ function inboundStreamErrorMessage(e) { for await (const [space, message] of app.messages) { backoff = 1000; // healthy traffic — reset markStreamHealthy(); + // Every yield — any direction — proves the inbound stream is live. + noteInboundYield(); // Only forward inbound messages (ignore our own outbound echoes). if (message && message.direction && message.direction !== "inbound") { continue; @@ -623,6 +674,99 @@ function inboundStreamErrorMessage(e) { } })(); +// --------------------------------------------------------------------------- +// Zombie-stream watchdog: detect a half-open gRPC stream the SDK can't see. +// +// The inbound iterator above can hang forever on a half-open socket — no +// error, no end — so the re-subscribe loop never fires and classifyStreamLog +// never sees a "[spectrum.stream]" line. We track the iterator's last yield +// (noteInboundYield) and, once it has been silent past +// STREAM_SILENCE_PROBE_MS, drive a cheap unary read (space.getMessage on a +// synthetic id) over the same authenticated channel. Decision rules (see +// stream-staleness.mjs): +// probe proves connectivity while the stream is silent -> the stream itself +// is the dead part -> markStreamDegraded -> existing exit-75 restart path. +// probe inconclusive (rejected/hung) -> do NOTHING: the network may just be +// down, and in that case the iterator eventually throws and the +// re-subscribe loop recovers on its own. Never restart on silence alone — +// shared lines can be legitimately quiet for hours. + +async function probeUpstream() { + if (typeof app?.stop !== "function") { + return { alive: false, hung: false, reason: "spectrum app not constructed" }; + } + const probeId = + PROBE_MSG_PREFIX + Date.now() + "-" + Math.random().toString(36).slice(2); + let timer = null; + const timeout = new Promise((resolve) => { + timer = setTimeout( + () => resolve({ alive: false, hung: true, reason: "probe timed out" }), + STREAM_PROBE_TIMEOUT_MS + ); + timer.unref(); + }); + const attempt = (async () => { + try { + const im = imessage(app); + const space = await im.space.get(PROBE_SPACE_ID); + await space.getMessage(probeId); + // Resolving for a synthetic id is unexpected but is still a completed + // round-trip: the channel is alive. + return { alive: true, hung: false, reason: "round-trip completed" }; + } catch (e) { + const verdict = classifyProbeRejection(e); + return { alive: verdict.alive, hung: false, reason: verdict.reason }; + } + })(); + const outcome = await Promise.race([attempt, timeout]); + if (timer) clearTimeout(timer); + staleness.lastProbeAt = Date.now(); + staleness.lastProbeOutcome = outcome.alive ? "alive" : "inconclusive"; + return outcome; +} + +let watchdogProbeInFlight = false; +async function zombieWatchdogTick() { + if (watchdogProbeInFlight) return; + const now = Date.now(); + const silentForMs = now - staleness.lastInboundAt; + if ( + !shouldProbe( + silentForMs, + STREAM_SILENCE_PROBE_MS, + now - staleness.lastProbeAt, + STREAM_PROBE_COOLDOWN_MS + ) + ) { + return; + } + watchdogProbeInFlight = true; + try { + const outcome = await probeUpstream(); + if (isZombieSuspect(silentForMs, STREAM_SILENCE_PROBE_MS, outcome)) { + staleness.zombieSuspected = true; + const reason = + `inbound stream silent for ${silentForMs}ms while an upstream probe ` + + "succeeded — half-open (zombie) gRPC stream suspected"; + console.error("photon-sidecar: " + reason); + markStreamDegraded(reason); + } + // Inconclusive: deliberately no action (see block comment above). + } catch (e) { + console.error( + "photon-sidecar: zombie watchdog tick failed: " + + (e && e.message ? e.message : String(e)) + ); + } finally { + watchdogProbeInFlight = false; + } +} + +if (STREAM_SILENCE_PROBE_MS > 0) { + const watchdogTimer = setInterval(zombieWatchdogTick, STREAM_WATCHDOG_TICK_MS); + watchdogTimer.unref(); +} + // --------------------------------------------------------------------------- // HTTP control + inbound server (loopback only). @@ -842,37 +986,24 @@ const server = http.createServer(async (req, res) => { } 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); + // channel the inbound stream uses. STRICT semantics: only a completed + // round-trip (resolution, or a not-found rejection for our synthetic + // id) counts as alive; any other rejection or a timeout is + // INCONCLUSIVE — never reported as alive. + const outcome = await probeUpstream(); + if (outcome.alive) { + return ok(res, { alive: true, outcome: "alive" }); } + res.statusCode = 503; + res.setHeader("Content-Type", "application/json"); + return res.end( + JSON.stringify({ + ok: false, + alive: false, + outcome: outcome.hung ? "hung" : "inconclusive", + reason: outcome.reason, + }) + ); } if (req.url === "/shutdown") { ok(res, {}); diff --git a/plugins/platforms/photon/sidecar/stream-staleness.mjs b/plugins/platforms/photon/sidecar/stream-staleness.mjs new file mode 100644 index 000000000000..a05472c205d9 --- /dev/null +++ b/plugins/platforms/photon/sidecar/stream-staleness.mjs @@ -0,0 +1,80 @@ +// Pure decision helpers for the zombie-stream (half-open gRPC) watchdog. +// +// spectrum-ts only reconnects when its inbound async iterator throws or ends. +// A half-open ("zombie") socket makes the iterator hang forever — no error, +// no end — so inbound silently dies while /healthz still looks fine. The +// watchdog in index.mjs tracks the last time the inbound iterator yielded and, +// once the stream has been silent past a conservative threshold, drives a +// cheap authenticated unary read over the same channel. STRICT semantics: +// +// - probe resolves, or rejects with a not-found-shaped error for our +// synthetic id -> ALIVE (the wire round-tripped) +// - probe rejects any other way (UNAVAILABLE, DEADLINE_EXCEEDED, network +// down, ...) -> INCONCLUSIVE — never treated as alive, and +// never treated as zombie-proof either +// +// A zombie is only declared when the stream is silent past the threshold AND +// a probe proves connectivity (the wire works but the stream is deaf). Silence +// alone NEVER degrades the stream: shared lines can be legitimately quiet for +// hours. Inconclusive probes NEVER degrade it either: the network may simply +// be down, and in that case the iterator will eventually throw and the +// existing re-subscribe loop recovers on its own. +// +// These helpers are pure (no SDK, no timers) so tests can execute them under +// node — see tests/plugins/platforms/photon/test_zombie_stream_watchdog.py. + +// gRPC NOT_FOUND is code 5; SDKs also surface it as "not found" / "NotFound" +// message text. Anything not clearly not-found is inconclusive. +const NOT_FOUND_RE = /not[\s_-]?found/i; + +/** + * Classify the rejection of the synthetic-id probe read. + * + * @param {unknown} err error thrown by `space.getMessage()` + * @returns {{alive: boolean, inconclusive: boolean, reason: string}} + */ +export function classifyProbeRejection(err) { + const code = err && typeof err === "object" ? err.code : undefined; + const message = + err && typeof err === "object" && err.message + ? String(err.message) + : String(err); + if (code === 5 || code === "notFound" || NOT_FOUND_RE.test(message)) { + // Expected: the synthetic id doesn't exist. The unary call completed a + // round-trip, so the channel is provably alive. + return { alive: true, inconclusive: false, reason: "not-found round-trip" }; + } + // Anything else (UNAVAILABLE, DEADLINE_EXCEEDED, TLS, auth, ...) does NOT + // prove liveness — and doesn't prove a zombie either. + return { alive: false, inconclusive: true, reason: message }; +} + +/** + * Should the watchdog probe at all this tick? + * + * @param {number} silentForMs ms since the inbound iterator last yielded + * @param {number} thresholdMs silence threshold (<= 0 disables the watchdog) + * @param {number} sinceLastProbeMs ms since the previous probe attempt + * @param {number} probeCooldownMs min spacing between probe attempts + * @returns {boolean} + */ +export function shouldProbe(silentForMs, thresholdMs, sinceLastProbeMs, probeCooldownMs) { + if (!(thresholdMs > 0)) return false; + if (silentForMs < thresholdMs) return false; + return sinceLastProbeMs >= probeCooldownMs; +} + +/** + * Final classification: zombie only on silence past threshold + probe-proven + * connectivity. Never on silence alone, never on an inconclusive probe. + * + * @param {number} silentForMs ms since the inbound iterator last yielded + * @param {number} thresholdMs silence threshold (<= 0 disables the watchdog) + * @param {{alive: boolean}} probeOutcome + * @returns {boolean} + */ +export function isZombieSuspect(silentForMs, thresholdMs, probeOutcome) { + if (!(thresholdMs > 0)) return false; + if (silentForMs < thresholdMs) return false; + return probeOutcome != null && probeOutcome.alive === true; +} diff --git a/tests/plugins/platforms/photon/test_presence_watchdog.py b/tests/plugins/platforms/photon/test_presence_watchdog.py index bbe04e287a26..3034f23b3999 100644 --- a/tests/plugins/platforms/photon/test_presence_watchdog.py +++ b/tests/plugins/platforms/photon/test_presence_watchdog.py @@ -30,7 +30,9 @@ def _make_adapter(monkeypatch: pytest.MonkeyPatch, **extra: Any) -> PhotonAdapte def test_probe_config_defaults(monkeypatch: pytest.MonkeyPatch) -> None: a = _make_adapter(monkeypatch) - assert a._probe_interval == 60.0 + # Conservative by default: probe only after 10+ minutes of stream silence + # so quiet shared lines never trigger restart storms. + assert a._probe_interval == 600.0 assert a._probe_timeout == 10.0 assert a._probe_max_failures == 3 assert a._probe_enabled is True @@ -78,26 +80,31 @@ async def test_probe_once_alive_on_200(monkeypatch: pytest.MonkeyPatch) -> None: return _Resp() a._http_client = _Client() # type: ignore[assignment] - assert await a._probe_once() is True + assert await a._probe_once() == "alive" @pytest.mark.asyncio -async def test_probe_once_dead_on_500(monkeypatch: pytest.MonkeyPatch) -> None: +async def test_probe_once_inconclusive_on_error_status( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A non-200 from /probe is INCONCLUSIVE, never alive: the sidecar's + strict probe returns 503 when it cannot prove upstream liveness, and + that must not count as proof in either direction.""" a = _make_adapter(monkeypatch) class _Resp: - status_code = 500 + status_code = 503 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 + assert await a._probe_once() == "inconclusive" @pytest.mark.asyncio -async def test_probe_once_dead_on_timeout(monkeypatch: pytest.MonkeyPatch) -> None: +async def test_probe_once_hung_on_timeout(monkeypatch: pytest.MonkeyPatch) -> None: a = _make_adapter(monkeypatch) class _Client: @@ -105,17 +112,17 @@ async def test_probe_once_dead_on_timeout(monkeypatch: pytest.MonkeyPatch) -> No 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 + # A hung probe HTTP call is the only verdict that counts toward respawn. + assert await a._probe_once() == "hung" @pytest.mark.asyncio -async def test_probe_once_false_without_client( +async def test_probe_once_inconclusive_without_client( monkeypatch: pytest.MonkeyPatch, ) -> None: a = _make_adapter(monkeypatch) a._http_client = None - assert await a._probe_once() is False + assert await a._probe_once() == "inconclusive" @pytest.mark.asyncio @@ -129,17 +136,17 @@ async def test_respawn_after_max_failures(monkeypatch: pytest.MonkeyPatch) -> No respawns.append(reason) a._note_upstream_activity() # mirror real respawn (clears failures) - async def _dead_probe() -> bool: - return False + async def _hung_probe() -> str: + return "hung" monkeypatch.setattr(a, "_respawn_sidecar", _fake_respawn) - monkeypatch.setattr(a, "_probe_once", _dead_probe) + monkeypatch.setattr(a, "_probe_once", _hung_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 + verdict = await a._probe_once() + assert verdict == "hung" a._probe_failures += 1 if a._probe_failures >= a._probe_max_failures: await a._respawn_sidecar("test") diff --git a/tests/plugins/platforms/photon/test_zombie_stream_watchdog.py b/tests/plugins/platforms/photon/test_zombie_stream_watchdog.py new file mode 100644 index 000000000000..e3f9dd6749fb --- /dev/null +++ b/tests/plugins/platforms/photon/test_zombie_stream_watchdog.py @@ -0,0 +1,303 @@ +"""Zombie-stream watchdog tests (half-open gRPC stream, issue #54036). + +spectrum-ts only reconnects when its inbound iterator throws or ends; a +half-open ("zombie") socket makes the iterator hang forever — no error, no +end — so inbound silently dies while the sidecar process looks healthy. + +The salvaged design has two layers: + +1. Sidecar (node): ``stream-staleness.mjs`` decision rules + a watchdog in + ``index.mjs`` that tracks the iterator's last yield, probes only after a + conservative silence threshold, and classifies degraded ONLY when a probe + proves connectivity while the stream is silent (never on silence alone, + never on an inconclusive probe). Degraded feeds the existing exit-75 + restart path and the ``staleness`` block on ``/healthz``. +2. Adapter (python): ``_monitor_sidecar_health`` surfaces the new staleness + fields; ``_probe_once`` has strict tri-state semantics (alive / hung / + inconclusive). + +These tests execute the real node decision module and drive the adapter +against mocked ``/healthz`` responses — style follows +test_overflow_recovery.py / test_spectrum_patch.py. No ports are bound and no +gRPC traffic occurs. +""" +from __future__ import annotations + +import json +import subprocess +from pathlib import Path +from typing import Any, Dict + +import pytest + +from gateway.config import PlatformConfig +from plugins.platforms.photon.adapter import PhotonAdapter + +_MODULE = Path("plugins/platforms/photon/sidecar/stream-staleness.mjs").resolve() + + +def _make_adapter(monkeypatch: pytest.MonkeyPatch) -> PhotonAdapter: + monkeypatch.setenv("PHOTON_PROJECT_ID", "test-project-id") + monkeypatch.setenv("PHOTON_PROJECT_SECRET", "test-project-secret") + cfg = PlatformConfig(enabled=True, token="", extra={}) + return PhotonAdapter(cfg) + + +# -- Sidecar decision rules (execute the real node module) ------------------- + +def _run_staleness_harness(script: str) -> Dict[str, Any]: + harness = ( + "import { classifyProbeRejection, shouldProbe, isZombieSuspect } " + f"from {json.dumps(_MODULE.as_uri())};\n" + + script + ) + run = subprocess.run( + ["node", "--input-type=module", "-e", harness], + cwd=Path.cwd(), + text=True, + capture_output=True, + check=False, + ) + assert run.returncode == 0, run.stderr + return json.loads(run.stdout) + + +def test_probe_rejection_classification_is_strict() -> None: + """Only not-found-shaped rejections prove liveness; everything else is + inconclusive — a rejected probe is NEVER treated as alive (#45580's + original /probe treated any rejection as alive, which was too loose).""" + out = _run_staleness_harness( + """ + const results = { + notFoundCode: classifyProbeRejection({ code: 5, message: "5 NOT_FOUND: nope" }), + notFoundText: classifyProbeRejection(new Error("message not found")), + sdkNotFound: classifyProbeRejection({ code: "notFound", message: "missing" }), + unavailable: classifyProbeRejection({ code: 14, message: "14 UNAVAILABLE: connect failed" }), + deadline: classifyProbeRejection({ code: 4, message: "4 DEADLINE_EXCEEDED" }), + generic: classifyProbeRejection(new Error("socket hang up")), + weird: classifyProbeRejection("string error"), + }; + process.stdout.write(JSON.stringify(results)); + """ + ) + # Completed round-trips (server said not-found for our synthetic id). + for name in ("notFoundCode", "notFoundText", "sdkNotFound"): + assert out[name]["alive"] is True, name + assert out[name]["inconclusive"] is False, name + # Everything else: not alive AND explicitly inconclusive. + for name in ("unavailable", "deadline", "generic", "weird"): + assert out[name]["alive"] is False, name + assert out[name]["inconclusive"] is True, name + + +def test_should_probe_requires_silence_past_threshold_and_cooldown() -> None: + out = _run_staleness_harness( + """ + const MIN10 = 10 * 60 * 1000; + const results = { + quietButUnderThreshold: shouldProbe(MIN10 - 1, MIN10, MIN10, 120000), + pastThreshold: shouldProbe(MIN10 + 1, MIN10, MIN10, 120000), + pastThresholdButCoolingDown: shouldProbe(MIN10 + 1, MIN10, 1000, 120000), + watchdogDisabled: shouldProbe(MIN10 * 100, 0, MIN10, 120000), + watchdogDisabledNegative: shouldProbe(MIN10 * 100, -1, MIN10, 120000), + }; + process.stdout.write(JSON.stringify(results)); + """ + ) + assert out["quietButUnderThreshold"] is False + assert out["pastThreshold"] is True + assert out["pastThresholdButCoolingDown"] is False + assert out["watchdogDisabled"] is False + assert out["watchdogDisabledNegative"] is False + + +def test_zombie_requires_probe_proven_connectivity_never_silence_alone() -> None: + """The core conservatism rule: shared lines can be quiet for hours, so a + zombie is declared only when the stream is silent past threshold AND a + probe PROVED the wire works (stream dead, channel alive).""" + out = _run_staleness_harness( + """ + const MIN10 = 10 * 60 * 1000; + const alive = { alive: true }; + const inconclusive = { alive: false }; + const results = { + silentAndProbeAlive: isZombieSuspect(MIN10 * 2, MIN10, alive), + silentButProbeInconclusive: isZombieSuspect(MIN10 * 2, MIN10, inconclusive), + silentNoProbe: isZombieSuspect(MIN10 * 2, MIN10, null), + hoursOfSilenceInconclusive: isZombieSuspect(MIN10 * 36, MIN10, inconclusive), + notSilentEnough: isZombieSuspect(MIN10 - 1, MIN10, alive), + disabled: isZombieSuspect(MIN10 * 2, 0, alive), + }; + process.stdout.write(JSON.stringify(results)); + """ + ) + assert out["silentAndProbeAlive"] is True + # Silence alone — even 6 hours of it — is NEVER a zombie verdict. + assert out["silentButProbeInconclusive"] is False + assert out["silentNoProbe"] is False + assert out["hoursOfSilenceInconclusive"] is False + assert out["notSilentEnough"] is False + assert out["disabled"] is False + + +# -- Adapter surfacing of the new /healthz staleness fields ------------------ + +def _healthz_payload(**staleness: Any) -> Dict[str, Any]: + return { + "ok": True, + "stream": { + "ok": True, + "state": "healthy", + "degradedForMs": 0, + "staleness": { + "lastInboundAt": "2026-07-28T00:00:00.000Z", + "silentForMs": 0, + "silenceThresholdMs": 600000, + "lastProbeAt": None, + "lastProbeOutcome": None, + "zombieSuspected": False, + **staleness, + }, + }, + } + + +@pytest.mark.asyncio +async def test_monitor_surfaces_zombie_suspected_without_fatal( + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, +) -> None: + """zombieSuspected on /healthz is surfaced as a warning while the stream + is still 'ok' — the fatal path stays owned by the degraded state (the + sidecar escalates to degraded -> exit 75 itself).""" + adapter = _make_adapter(monkeypatch) + adapter._inbound_running = True + adapter._sidecar_health_interval = 0.0 + + polls = 0 + + async def _fake_call(path: str, payload: Dict[str, Any]) -> Any: + nonlocal polls + assert path == "/healthz" + polls += 1 + if polls >= 2: + adapter._inbound_running = False + return _healthz_payload( + silentForMs=1_200_000, + lastProbeOutcome="alive", + zombieSuspected=True, + ) + + monkeypatch.setattr(adapter, "_sidecar_call", _fake_call) + + with caplog.at_level("WARNING"): + await adapter._monitor_sidecar_health() + + assert adapter.has_fatal_error is False + assert any( + "suspected zombie stream" in rec.message for rec in caplog.records + ) + + +@pytest.mark.asyncio +async def test_monitor_still_raises_fatal_when_zombie_degrades_stream( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Once the sidecar's watchdog escalates the zombie to degraded, the + existing UPSTREAM_STREAM_DEGRADED reconnect path fires unchanged.""" + adapter = _make_adapter(monkeypatch) + adapter._inbound_running = True + adapter._sidecar_health_interval = 0.0 + + async def _fake_call(path: str, payload: Dict[str, Any]) -> Any: + assert path == "/healthz" + return { + "ok": True, + "stream": { + "ok": False, + "state": "degraded", + "degradedForMs": 95000, + "lastIssue": ( + "inbound stream silent for 1200000ms while an upstream " + "probe succeeded — half-open (zombie) gRPC stream suspected" + ), + "staleness": { + "silentForMs": 1_200_000, + "lastProbeOutcome": "alive", + "zombieSuspected": True, + }, + }, + } + + notified: list[bool] = [] + + async def _fake_notify() -> None: + notified.append(True) + adapter._inbound_running = False + + monkeypatch.setattr(adapter, "_sidecar_call", _fake_call) + monkeypatch.setattr(adapter, "_notify_fatal_error", _fake_notify) + + await adapter._monitor_sidecar_health() + + assert adapter.has_fatal_error is True + assert adapter.fatal_error_code == "UPSTREAM_STREAM_DEGRADED" + assert adapter.fatal_error_retryable is True + assert notified == [True] + + +@pytest.mark.asyncio +async def test_monitor_ignores_healthz_without_staleness_block( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Old sidecars (no staleness field) keep working: the monitor must not + crash or raise fatals on the legacy /healthz shape.""" + adapter = _make_adapter(monkeypatch) + adapter._inbound_running = True + adapter._sidecar_health_interval = 0.0 + + polls = 0 + + async def _fake_call(path: str, payload: Dict[str, Any]) -> Any: + nonlocal polls + polls += 1 + if polls >= 2: + adapter._inbound_running = False + return {"ok": True, "stream": {"ok": True, "state": "healthy"}} + + monkeypatch.setattr(adapter, "_sidecar_call", _fake_call) + + await adapter._monitor_sidecar_health() + + assert adapter.has_fatal_error is False + + +# -- Adapter watchdog: inconclusive never counts toward respawn -------------- + +@pytest.mark.asyncio +async def test_inconclusive_probes_never_accumulate_toward_respawn( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Strict semantics end-to-end at the adapter: a 503/transport-error probe + (inconclusive) must not increment the failure counter the way the original + #45580 booleans did — only hung probes do.""" + adapter = _make_adapter(monkeypatch) + + class _Resp503: + status_code = 503 + + class _Client: + async def post(self, *args: Any, **kwargs: Any) -> Any: + return _Resp503() + + adapter._http_client = _Client() # type: ignore[assignment] + + # Many inconclusive probes in a row: mirror the watchdog's per-iteration + # bookkeeping (only "hung" increments) and assert no failures accrue. + for _ in range(10): + verdict = await adapter._probe_once() + assert verdict == "inconclusive" + if verdict == "hung": + adapter._probe_failures += 1 + + assert adapter._probe_failures == 0