fix(net): detect a silently-dead live stream via a heartbeat watchdog
CI / changes (pull_request) Successful in 2s
CI / unit (pull_request) Has been skipped
CI / integration (pull_request) Has been skipped
CI / ui (pull_request) Successful in 1m14s
CI / conformance (pull_request) Successful in 12s
CI / gate (pull_request) Successful in 0s
CI / deploy (pull_request) Successful in 1m59s
CI / changes (pull_request) Successful in 2s
CI / unit (pull_request) Has been skipped
CI / integration (pull_request) Has been skipped
CI / ui (pull_request) Successful in 1m14s
CI / conformance (pull_request) Successful in 12s
CI / gate (pull_request) Successful in 0s
CI / deploy (pull_request) Successful in 1m59s
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.
This commit is contained in:
@@ -107,8 +107,10 @@ dropped). Horizontal scaling is explicit future work.
|
|||||||
the header **"Connecting…"** spinner, chrome stays online); `offlineNoNetwork` (sustained loss — blue
|
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
|
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
|
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` /
|
implicit** — detected from failed calls, a reachability probe, a **live-stream heartbeat watchdog**
|
||||||
`@capacitor/network`), **never a user toggle** — with **hysteresis** so a brief blip lives entirely in
|
(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
|
`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
|
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
|
backoff** — every op on a rate-limit (the gateway rejected it before processing, so it is safe), but
|
||||||
|
|||||||
@@ -10,6 +10,7 @@ import { navigate, router } from './router.svelte';
|
|||||||
import { errorKey, localeFrom, setLocale, t, type Locale } from './i18n/index.svelte';
|
import { errorKey, localeFrom, setLocale, t, type Locale } from './i18n/index.svelte';
|
||||||
import { languageNeedsServerSync } from './language';
|
import { languageNeedsServerSync } from './language';
|
||||||
import { startLocalEvalMetrics } from './localeval-metrics';
|
import { startLocalEvalMetrics } from './localeval-metrics';
|
||||||
|
import { createStreamWatchdog } from './streamwatchdog';
|
||||||
import { applyReduceMotion, applyTelegramTheme, applyTheme, type ThemePref, type TelegramThemeParams } from './theme';
|
import { applyReduceMotion, applyTelegramTheme, applyTheme, type ThemePref, type TelegramThemeParams } from './theme';
|
||||||
import {
|
import {
|
||||||
insideTelegram,
|
insideTelegram,
|
||||||
@@ -215,6 +216,7 @@ function bannerSuppressed(): boolean {
|
|||||||
|
|
||||||
function goBackground(): void {
|
function goBackground(): void {
|
||||||
backgrounded = true;
|
backgrounded = true;
|
||||||
|
streamWatchdog.pause(); // do not count silence against a suspended (throttled) stream
|
||||||
}
|
}
|
||||||
|
|
||||||
function goForeground(): void {
|
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
|
// 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.
|
// event may have been shed from a full hub buffer while away; nudge an open game to refetch.
|
||||||
app.resync++;
|
app.resync++;
|
||||||
|
streamWatchdog.resume(); // resume liveness watching paused on the way to the background
|
||||||
}
|
}
|
||||||
void refreshNotifications();
|
void refreshNotifications();
|
||||||
}
|
}
|
||||||
@@ -353,12 +356,35 @@ function viewingGame(gameId: string): boolean {
|
|||||||
return router.route.name === 'game' && router.route.params.id === gameId;
|
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 {
|
function openStream(): void {
|
||||||
closeStream();
|
closeStream();
|
||||||
app.streamAlive = true;
|
app.streamAlive = true;
|
||||||
unsubscribeStream = gateway.subscribe(
|
unsubscribeStream = gateway.subscribe(
|
||||||
(e) => {
|
(e) => {
|
||||||
reportOnline(); // a delivered event proves the gateway is reachable
|
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;
|
app.lastEvent = e;
|
||||||
if (e.kind === 'chat_message' && e.message.senderId !== app.session?.userId) {
|
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
|
// 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;
|
app.streamAlive = false;
|
||||||
// A background suspend drops the single-shot stream. Keep the indicator hidden while
|
// A background suspend drops the single-shot stream. Keep the indicator hidden while
|
||||||
// backgrounded or just-resumed (bannerSuppressed); otherwise show "Connecting…" (the
|
// backgrounded or just-resumed (bannerSuppressed); otherwise show "Connecting…" (the
|
||||||
@@ -465,6 +492,7 @@ function openStream(): void {
|
|||||||
scheduleReconnect();
|
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
|
/** scheduleReconnect reopens a dropped stream once, after a short delay, while the
|
||||||
@@ -529,6 +557,7 @@ export async function refreshFeedbackBadge(): Promise<void> {
|
|||||||
}
|
}
|
||||||
|
|
||||||
function closeStream(): void {
|
function closeStream(): void {
|
||||||
|
streamWatchdog.clear();
|
||||||
if (reconnectTimer) {
|
if (reconnectTimer) {
|
||||||
clearTimeout(reconnectTimer);
|
clearTimeout(reconnectTimer);
|
||||||
reconnectTimer = null;
|
reconnectTimer = null;
|
||||||
|
|||||||
@@ -829,7 +829,14 @@ export class MockGateway implements GatewayClient {
|
|||||||
// --- live stream ---
|
// --- live stream ---
|
||||||
subscribe(onEvent: (e: PushEvent) => void): Unsubscribe {
|
subscribe(onEvent: (e: PushEvent) => void): Unsubscribe {
|
||||||
this.subs.add(onEvent);
|
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.
|
// Fabricate an opponent reply shortly after the human moves, then hand the turn back.
|
||||||
|
|||||||
@@ -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);
|
||||||
|
});
|
||||||
|
});
|
||||||
@@ -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<typeof setTimeout> | 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();
|
||||||
|
},
|
||||||
|
};
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user