From 25381f70a363663af94360a6403bf2096f4cba0f Mon Sep 17 00:00:00 2001 From: Ilia Denisov Date: Tue, 14 Jul 2026 08:27:38 +0200 Subject: [PATCH] fix(net): detect a silently-dead live stream via a heartbeat watchdog MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An idle live stream whose connection dies without an error (airplane mode cutting the radio) left the net-state machine stuck "online": the streaming fetch neither errored nor delivered, no unary call was made, and the OS `navigator` offline hint is unreliable in the Telegram/VK WebViews — so the "Connecting…"/offline state only surfaced on the next foreground resync, not while idle. The gateway already pushes a keep-alive `heartbeat` every 10s, so add a client-side watchdog (streamwatchdog.ts) that resets on every delivered event and, after ~25s of silence, treats the stream as dropped (reportOffline + reconnect). It is background-aware (paused while suspended) so a throttled foreground return does not false-trip. The mock client mirrors the heartbeat so an idle e2e session stays online. --- docs/ARCHITECTURE.md | 6 ++- ui/src/lib/app.svelte.ts | 29 +++++++++++ ui/src/lib/mock/client.ts | 9 +++- ui/src/lib/streamwatchdog.test.ts | 82 +++++++++++++++++++++++++++++++ ui/src/lib/streamwatchdog.ts | 63 ++++++++++++++++++++++++ 5 files changed, 186 insertions(+), 3 deletions(-) create mode 100644 ui/src/lib/streamwatchdog.test.ts create mode 100644 ui/src/lib/streamwatchdog.ts diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 3e4d037..23c8f9c 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -107,8 +107,10 @@ dropped). Horizontal scaling is explicit future work. the header **"Connecting…"** spinner, chrome stays online); `offlineNoNetwork` (sustained loss — blue chrome, a local-only lobby, the transport **kill switch**); and `offlineVersionLocked` (the gateway is reachable but the build is below the hard minimum — the *update* notice, the client-version gate below). **Offline is - implicit** — detected from failed calls, a reachability probe and the OS hint (`navigator` / - `@capacitor/network`), **never a user toggle** — with **hysteresis** so a brief blip lives entirely in + implicit** — detected from failed calls, a reachability probe, a **live-stream heartbeat watchdog** + (a silence past ~25 s of the gateway's 10 s keep-alive `heartbeat` means a silently-dead stream — e.g. + the radio was cut with no stream error; `ui/src/lib/streamwatchdog.ts`, background-aware) and the OS hint + (`navigator` / `@capacitor/network`), **never a user toggle** — with **hysteresis** so a brief blip lives entirely in `connecting` (K consecutive probe failures *or* a debounce window trips `offlineNoNetwork`; the first probe/call success heals back, with a toast). The transport **auto-retries with capped exponential backoff** — every op on a rate-limit (the gateway rejected it before processing, so it is safe), but diff --git a/ui/src/lib/app.svelte.ts b/ui/src/lib/app.svelte.ts index bf5c7ab..d57885c 100644 --- a/ui/src/lib/app.svelte.ts +++ b/ui/src/lib/app.svelte.ts @@ -10,6 +10,7 @@ import { navigate, router } from './router.svelte'; import { errorKey, localeFrom, setLocale, t, type Locale } from './i18n/index.svelte'; import { languageNeedsServerSync } from './language'; import { startLocalEvalMetrics } from './localeval-metrics'; +import { createStreamWatchdog } from './streamwatchdog'; import { applyReduceMotion, applyTelegramTheme, applyTheme, type ThemePref, type TelegramThemeParams } from './theme'; import { insideTelegram, @@ -215,6 +216,7 @@ function bannerSuppressed(): boolean { function goBackground(): void { backgrounded = true; + streamWatchdog.pause(); // do not count silence against a suspended (throttled) stream } function goForeground(): void { @@ -227,6 +229,7 @@ function goForeground(): void { // The stream stayed alive across the suspend, so the reconnect refetch will not fire — but an // event may have been shed from a full hub buffer while away; nudge an open game to refetch. app.resync++; + streamWatchdog.resume(); // resume liveness watching paused on the way to the background } void refreshNotifications(); } @@ -353,12 +356,35 @@ function viewingGame(gameId: string): boolean { return router.route.name === 'game' && router.route.params.id === gameId; } +// The live stream carries a heartbeat every GATEWAY_PUSH_HEARTBEAT_INTERVAL (10 s by default), so a +// healthy stream delivers an event well inside this window even when the game is idle. A silence +// past it means the stream died without erroring (e.g. airplane mode cut the radio): tear the dead +// subscription down and drop, so the machine leaves "online" instead of waiting for the next call. +// 25 s tolerates a couple of missed heartbeats; it must stay above the server interval + slack. +const STREAM_IDLE_TIMEOUT_MS = 25000; +const streamWatchdog = createStreamWatchdog({ + timeoutMs: STREAM_IDLE_TIMEOUT_MS, + onDead: () => { + // Do not trip while suspended: timers are throttled and a backgrounded stream is legitimately + // silent (the reopen on resume covers a genuine drop). + if (backgrounded || documentHidden()) return; + // Abort the dead subscription explicitly — an aborted stream skips onError (transport gates on + // signal.aborted) — then follow the same drop path as an error close. + unsubscribeStream?.(); + unsubscribeStream = null; + app.streamAlive = false; + if (!bannerSuppressed()) reportOffline(); + scheduleReconnect(); + }, +}); + function openStream(): void { closeStream(); app.streamAlive = true; unsubscribeStream = gateway.subscribe( (e) => { reportOnline(); // a delivered event proves the gateway is reachable + streamWatchdog.reset(); // any event (incl. the heartbeat) proves the stream is still live app.lastEvent = e; if (e.kind === 'chat_message' && e.message.senderId !== app.session?.userId) { // While the player is in that game's comms hub (chat or dictionary tab), neither @@ -457,6 +483,7 @@ function openStream(): void { } }, () => { + streamWatchdog.clear(); app.streamAlive = false; // A background suspend drops the single-shot stream. Keep the indicator hidden while // backgrounded or just-resumed (bannerSuppressed); otherwise show "Connecting…" (the @@ -465,6 +492,7 @@ function openStream(): void { scheduleReconnect(); }, ); + streamWatchdog.reset(); // start the liveness watch for this stream (armed by the first heartbeat) } /** scheduleReconnect reopens a dropped stream once, after a short delay, while the @@ -529,6 +557,7 @@ export async function refreshFeedbackBadge(): Promise { } function closeStream(): void { + streamWatchdog.clear(); if (reconnectTimer) { clearTimeout(reconnectTimer); reconnectTimer = null; diff --git a/ui/src/lib/mock/client.ts b/ui/src/lib/mock/client.ts index 84b0e4e..4658a3c 100644 --- a/ui/src/lib/mock/client.ts +++ b/ui/src/lib/mock/client.ts @@ -829,7 +829,14 @@ export class MockGateway implements GatewayClient { // --- live stream --- subscribe(onEvent: (e: PushEvent) => void): Unsubscribe { this.subs.add(onEvent); - return () => this.subs.delete(onEvent); + // Mirror the gateway's idle keep-alive (an immediate heartbeat, then one every 10 s) so the + // client's stream watchdog sees a live stream through an idle mock session, as in production. + onEvent({ kind: 'heartbeat' }); + const hb = setInterval(() => onEvent({ kind: 'heartbeat' }), 10000); + return () => { + clearInterval(hb); + this.subs.delete(onEvent); + }; } // Fabricate an opponent reply shortly after the human moves, then hand the turn back. diff --git a/ui/src/lib/streamwatchdog.test.ts b/ui/src/lib/streamwatchdog.test.ts new file mode 100644 index 0000000..470781e --- /dev/null +++ b/ui/src/lib/streamwatchdog.test.ts @@ -0,0 +1,82 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { createStreamWatchdog } from './streamwatchdog'; + +describe('createStreamWatchdog', () => { + beforeEach(() => vi.useFakeTimers()); + afterEach(() => vi.useRealTimers()); + + it('fires onDead after the timeout with no activity', () => { + const onDead = vi.fn(); + const w = createStreamWatchdog({ timeoutMs: 25000, onDead }); + w.reset(); + vi.advanceTimersByTime(24999); + expect(onDead).not.toHaveBeenCalled(); + vi.advanceTimersByTime(1); + expect(onDead).toHaveBeenCalledTimes(1); + }); + + it('reset before the timeout re-arms and prevents the fire', () => { + const onDead = vi.fn(); + const w = createStreamWatchdog({ timeoutMs: 25000, onDead }); + w.reset(); + vi.advanceTimersByTime(20000); + w.reset(); // a fresh event arrived; the deadline moves out + vi.advanceTimersByTime(20000); + expect(onDead).not.toHaveBeenCalled(); + vi.advanceTimersByTime(5000); + expect(onDead).toHaveBeenCalledTimes(1); + }); + + it('a steady heartbeat keeps it from ever firing', () => { + const onDead = vi.fn(); + const w = createStreamWatchdog({ timeoutMs: 25000, onDead }); + w.reset(); + // A 10 s server heartbeat resets it well inside the 25 s window, indefinitely. + for (let i = 0; i < 20; i++) { + vi.advanceTimersByTime(10000); + w.reset(); + } + expect(onDead).not.toHaveBeenCalled(); + }); + + it('pause clears the pending timer so it cannot fire', () => { + const onDead = vi.fn(); + const w = createStreamWatchdog({ timeoutMs: 25000, onDead }); + w.reset(); + w.pause(); + vi.advanceTimersByTime(60000); + expect(onDead).not.toHaveBeenCalled(); + }); + + it('resume re-arms after a pause and then fires on continued silence', () => { + const onDead = vi.fn(); + const w = createStreamWatchdog({ timeoutMs: 25000, onDead }); + w.reset(); + w.pause(); + vi.advanceTimersByTime(60000); // stayed silent while paused: no fire + expect(onDead).not.toHaveBeenCalled(); + w.resume(); + vi.advanceTimersByTime(24999); + expect(onDead).not.toHaveBeenCalled(); + vi.advanceTimersByTime(1); + expect(onDead).toHaveBeenCalledTimes(1); + }); + + it('clear stops a pending fire and does not re-arm', () => { + const onDead = vi.fn(); + const w = createStreamWatchdog({ timeoutMs: 25000, onDead }); + w.reset(); + w.clear(); + vi.advanceTimersByTime(60000); + expect(onDead).not.toHaveBeenCalled(); + }); + + it('reset after a pause resumes watching (unpauses)', () => { + const onDead = vi.fn(); + const w = createStreamWatchdog({ timeoutMs: 25000, onDead }); + w.pause(); + w.reset(); // a reopen after a suspend must start watching again + vi.advanceTimersByTime(25000); + expect(onDead).toHaveBeenCalledTimes(1); + }); +}); diff --git a/ui/src/lib/streamwatchdog.ts b/ui/src/lib/streamwatchdog.ts new file mode 100644 index 0000000..e610da7 --- /dev/null +++ b/ui/src/lib/streamwatchdog.ts @@ -0,0 +1,63 @@ +// A liveness watchdog for the live event stream. The gateway pushes a `heartbeat` event on a +// fixed cadence (GATEWAY_PUSH_HEARTBEAT_INTERVAL, default 10 s), so a healthy stream delivers at +// least one event well inside the timeout even when the game itself is idle. A network that dies +// silently — e.g. airplane mode cutting the radio — leaves the streaming fetch neither erroring +// nor delivering; without this watchdog the machine would stay "online" until the next active +// call fails (often only on a foreground resync). The watchdog resets on every delivered event +// and fires onDead once the stream has been silent for the whole timeout, so the app can report +// the drop and reconnect. + +/** StreamWatchdog is the control surface returned by {@link createStreamWatchdog}. */ +export interface StreamWatchdog { + /** reset (re)starts the countdown; call on stream open and on every delivered event. It also + * clears a prior pause, so a reopen after a suspend resumes watching. */ + reset(): void; + /** pause stops the countdown while the app is backgrounded (timers are throttled while hidden, + * and a suspended stream legitimately delivers nothing — neither should trip a false drop). */ + pause(): void; + /** resume restarts the countdown after a pause (on returning to the foreground). */ + resume(): void; + /** clear stops the countdown without re-arming; call when the stream is closed or has errored. */ + clear(): void; +} + +/** + * createStreamWatchdog builds a watchdog that calls onDead once no reset has happened for + * timeoutMs. timeoutMs must exceed the server heartbeat interval plus proxy/jitter slack, or a + * healthy but idle stream would trip it. + */ +export function createStreamWatchdog(opts: { timeoutMs: number; onDead: () => void }): StreamWatchdog { + let timer: ReturnType | null = null; + let paused = false; + + function stop(): void { + if (timer !== null) { + clearTimeout(timer); + timer = null; + } + } + function arm(): void { + stop(); + timer = setTimeout(opts.onDead, opts.timeoutMs); + } + + return { + reset(): void { + paused = false; + arm(); + }, + pause(): void { + paused = true; + stop(); + }, + resume(): void { + if (paused) { + paused = false; + arm(); + } + }, + clear(): void { + stop(); + }, + }; +}