diff --git a/CHANGELOG.md b/CHANGELOG.md index 2a530d1..367b171 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,38 @@ # @contextvm/sdk +## 0.13.10 + +### Patch Changes + +- fix(open-stream): arm keepalive probe before the ping publish and expose session staleness + + The CEP-41 open-stream keepalive armed its probe timer only *after* the ping + publish resolved, on both `OpenStreamSession` and `OpenStreamWriter`. A slow + or unhealthy Nostr relay whose publish retries indefinitely parked the probe + window forever: the probe timer never armed, the idle timer (one-shot) never + rescheduled, the async iterator blocked on its waiter with no rejection, and + the stream died silently — most acutely for long-lived browser clients + consuming server→client `subscribe` streams. The writer gained sender-side + keepalive in 0.13.8 (`2aaa38a`), which exposed the pre-existing symmetric + receiver weakness. + + - `OpenStreamSession.handleIdleTimeout` and `OpenStreamWriter.handleIdleTimeout` + now arm the probe timer **before** awaiting the ping publish. A stuck publish + can no longer suppress liveness detection: if the relay cannot deliver a + control frame within `probeTimeoutMs`, that is itself treated as a probe + failure. Late pong/resolve is reconciled by existing guards (the nonce match + in `handlePong`/`ackProbe`, `clearTimers`/`clearKeepalive` on finalize). The + writer's "publish-failed ⇒ re-arm idle" policy is preserved. + - `OpenStreamSession` now exposes `lastActivityAt` (wall-clock ms of the last + received frame) and `isStale(marginMs = 0)`, which returns true when + `idleTimeoutMs + probeTimeoutMs` (+ margin) has elapsed without any inbound + frame. The threshold uses the session's own client-driven cadence, so it does + not depend on the peer's keepalive policy. Consumers in timer-throttled + environments (e.g. browser background tabs, where the session's own + `setTimeout` may not fire) can read `isStale()` from a reliable trigger they + own (e.g. Page Visibility) to build an app-level watchdog. Read-only and + additive; no transport, handshake, or browser coupling added. + ## 0.13.9 ### Patch Changes diff --git a/docs/contextvm-docs b/docs/contextvm-docs index d8d6890..6e98b6e 160000 --- a/docs/contextvm-docs +++ b/docs/contextvm-docs @@ -1 +1 @@ -Subproject commit d8d68906c3a2b600a8c0b0a6f4b86ffc49e29fd8 +Subproject commit 6e98b6ef088ba75dcf537ea9aac64f068480ecb5 diff --git a/package.json b/package.json index c9b28ae..8c64461 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@contextvm/sdk", - "version": "0.13.9", + "version": "0.13.10", "description": "ContextVM SDK for JavaScript/TypeScript", "license": "LGPL-3.0-1", "author": "ContextVM", diff --git a/src/transport/open-stream/session.test.ts b/src/transport/open-stream/session.test.ts index 4e1e674..ed32acb 100644 --- a/src/transport/open-stream/session.test.ts +++ b/src/transport/open-stream/session.test.ts @@ -796,3 +796,133 @@ describe('OpenStreamSession', () => { await expect(session.closed).rejects.toBeInstanceOf(OpenStreamAbortError); }); }); + +describe('OpenStreamSession keepalive deadlock', () => { + test('aborts with "Probe timeout" when sendPing hangs (never resolves)', async () => { + // Regression: pre-0.13.10 armed the probe timer only after the ping + // publish resolved, so a stuck relay parked the session forever. + const pings: string[] = []; + const aborts: Array = []; + let resolvePing: () => void = () => undefined; + const session = new OpenStreamSession({ + progressToken: 'token-stuck-ping', + maxBufferedChunks: 8, + maxBufferedBytes: 1024, + idleTimeoutMs: 10, + probeTimeoutMs: 10, + closeGracePeriodMs: 100, + sendPing: (nonce: string): Promise => { + pings.push(nonce); + return new Promise((resolve) => { + resolvePing = resolve; + }); + }, + sendAbort: async (reason?: string): Promise => { + aborts.push(reason); + }, + }); + const closed = session.closed.catch((error: unknown) => error); + + await session.processFrame(1, { type: 'open-stream', frameType: 'start' }); + + await new Promise((resolve) => setTimeout(resolve, 35)); + + expect(pings).toHaveLength(1); + expect(aborts).toEqual(['Probe timeout']); + expect(await closed).toBeInstanceOf(OpenStreamAbortError); + + // Release the hung publish so the test does not leak a pending promise. + resolvePing(); + }); + + test('aborts with "Failed to send keepalive ping" when sendPing rejects', async () => { + const reasons: Array = []; + const session = new OpenStreamSession({ + progressToken: 'token-rejected-ping', + maxBufferedChunks: 8, + maxBufferedBytes: 1024, + idleTimeoutMs: 10, + probeTimeoutMs: 1000, // long probe so the rejection wins the race + closeGracePeriodMs: 100, + sendPing: async (): Promise => { + throw new Error('relay down'); + }, + onAbort: async (reason?: string): Promise => { + reasons.push(reason); + }, + }); + const closed = session.closed.catch((error: unknown) => error); + + await session.processFrame(1, { type: 'open-stream', frameType: 'start' }); + + await new Promise((resolve) => setTimeout(resolve, 30)); + + // publishAbort=false, so the reason surfaces via onAbort, not sendAbort; + // `closed` rejects with the underlying ping error. + expect(reasons).toEqual(['Failed to send keepalive ping']); + const error = (await closed) as Error; + expect(error).toBeInstanceOf(Error); + expect(error.message).toBe('relay down'); + }); +}); + +describe('OpenStreamSession staleness', () => { + test('a fresh session is not stale and exposes lastActivityAt', () => { + const session = new OpenStreamSession({ + progressToken: 'token-fresh', + maxBufferedChunks: 8, + maxBufferedBytes: 1024, + idleTimeoutMs: 1000, + probeTimeoutMs: 1000, + }); + expect(session.isStale()).toBe(false); + expect(typeof session.lastActivityAt).toBe('number'); + session.dispose(); + }); + + test('becomes stale after idle + probe window; margin widens the threshold', async () => { + const session = new OpenStreamSession({ + progressToken: 'token-stale', + maxBufferedChunks: 8, + maxBufferedBytes: 1024, + idleTimeoutMs: 10, + probeTimeoutMs: 10, + closeGracePeriodMs: 100, + sendPing: async (): Promise => undefined, + sendAbort: async (): Promise => undefined, + }); + await session.processFrame(1, { type: 'open-stream', frameType: 'start' }); + + await new Promise((resolve) => setTimeout(resolve, 30)); + + expect(session.isStale()).toBe(true); + // A 100ms margin pushes the threshold past the elapsed window. + expect(session.isStale(100)).toBe(false); + await session.closed.catch(() => undefined); + }); + + test('receiving a frame advances lastActivityAt and clears staleness', async () => { + const session = new OpenStreamSession({ + progressToken: 'token-reset', + maxBufferedChunks: 8, + maxBufferedBytes: 1024, + idleTimeoutMs: 1000, + probeTimeoutMs: 1000, + }); + await session.processFrame(1, { type: 'open-stream', frameType: 'start' }); + const first = session.lastActivityAt; + + await new Promise((resolve) => setTimeout(resolve, 20)); + // A peer ping counts as activity and refreshes the timestamp. + await session.processFrame(2, { + type: 'open-stream', + frameType: 'ping', + nonce: 'n1', + }); + + expect(session.lastActivityAt).toBeGreaterThanOrEqual(first + 20); + expect(session.isStale()).toBe(false); + session.dispose(); + await session.closed.catch(() => undefined); + }); +}); diff --git a/src/transport/open-stream/session.ts b/src/transport/open-stream/session.ts index e3d4656..792d2ab 100644 --- a/src/transport/open-stream/session.ts +++ b/src/transport/open-stream/session.ts @@ -90,6 +90,7 @@ export class OpenStreamSession implements OpenStreamSessionLike { private idleTimer: ReturnType | undefined; private probeTimer: ReturnType | undefined; private closeGraceTimer: ReturnType | undefined; + private lastActivityTimestamp = Date.now(); constructor(options: OpenStreamSessionOptions) { this.progressToken = options.progressToken; @@ -119,6 +120,28 @@ export class OpenStreamSession implements OpenStreamSessionLike { return this.active; } + /** + * Wall-clock timestamp (ms) of the last received frame (data, ping, or + * pong). Updated on every inbound frame regardless of active/closed state. + */ + public get lastActivityAt(): number { + return this.lastActivityTimestamp; + } + + /** + * True when the keepalive should have confirmed liveness by now but hasn't. + * Pure wall-clock over the session's own idle + probe window (+ margin), so + * consumers in timer-throttled environments (browser background tabs) can + * read it from a reliable trigger they own (e.g. Page Visibility). A cleanly + * closed stream eventually reads stale — gate on {@link isActive} first. + */ + public isStale(marginMs = 0): boolean { + return ( + Date.now() - this.lastActivityTimestamp > + this.idleTimeoutMs + this.probeTimeoutMs + marginMs + ); + } + public async abort(reason?: string): Promise { if (!this.active) { return; @@ -379,6 +402,8 @@ export class OpenStreamSession implements OpenStreamSessionLike { } private refreshIdleTimer(): void { + // Refresh activity even past close so staleness holds during close-grace. + this.lastActivityTimestamp = Date.now(); if (!this.active || this.closedRemotely) { return; } @@ -396,22 +421,24 @@ export class OpenStreamSession implements OpenStreamSessionLike { const nonce = this.nextControlNonce(); this.pendingProbeNonce = nonce; + // Arm BEFORE publishing: a stuck sendPing must not suppress detection; the + // probe window intentionally covers publish latency. A late pong/resolve is + // reconciled by handlePong's nonce match and clearTimers in finalize. + this.clearProbeTimer(); + this.probeTimer = setTimeout(() => { + this.handleProbeTimeout(nonce).catch(() => undefined); + }, this.probeTimeoutMs); try { await this.sendPing?.(nonce); } catch (error) { + if (!this.active) return; // probe timeout may have already finalized await this.finishAborted( error instanceof Error ? error : new Error(String(error)), 'Failed to send keepalive ping', false, ); - return; } - - this.clearProbeTimer(); - this.probeTimer = setTimeout(() => { - this.handleProbeTimeout(nonce).catch(() => undefined); - }, this.probeTimeoutMs); } private async handleProbeTimeout(nonce: string): Promise { diff --git a/src/transport/open-stream/writer.test.ts b/src/transport/open-stream/writer.test.ts index 61b020c..495d463 100644 --- a/src/transport/open-stream/writer.test.ts +++ b/src/transport/open-stream/writer.test.ts @@ -397,6 +397,47 @@ describe('OpenStreamWriter keepalive', () => { expect(types.filter((type) => type === 'abort')).toHaveLength(1); }); + test('aborts with "Probe timeout" when the keepalive publish hangs (never resolves)', async () => { + // Regression: pre-0.13.10 armed the probe timer only after the ping + // publish resolved, so a stuck relay parked the writer forever. + const frames: OpenStreamProgress[] = []; + const aborts: Array = []; + let resolvePing: () => void = () => undefined; + const writer = new OpenStreamWriter({ + progressToken: 'token-stuck-publish', + publishFrame: (frame): Promise => { + frames.push(frame); + // Let start/abort resolve so streaming begins and teardown completes; + // only the ping publish hangs to simulate a stuck relay. + if (frame.cvm.frameType === 'ping') { + return new Promise((resolve) => { + resolvePing = () => resolve(undefined); + }); + } + return Promise.resolve(undefined); + }, + onAbort: async (reason?: string): Promise => { + aborts.push(reason); + }, + idleTimeoutMs: 10, + probeTimeoutMs: 10, + }); + + await writer.start(); + await waitFor(() => !writer.isActive); + + expect(writer.isActive).toBe(false); + expect(aborts).toEqual(['Probe timeout']); + expect( + frames + .map((frame) => frame.cvm.frameType) + .filter((type) => type === 'ping'), + ).toHaveLength(1); + + // Release the hung publish so the test does not leak a pending promise. + resolvePing(); + }); + test('stays alive while the peer acks each keepalive probe', async () => { const frames: OpenStreamProgress[] = []; const writer = new OpenStreamWriter({ diff --git a/src/transport/open-stream/writer.ts b/src/transport/open-stream/writer.ts index bf73f94..b4e590a 100644 --- a/src/transport/open-stream/writer.ts +++ b/src/transport/open-stream/writer.ts @@ -279,6 +279,12 @@ export class OpenStreamWriter { const nonce = this.nextControlNonce(); this.pendingProbeNonce = nonce; + // Arm BEFORE publishing: a stuck publishFrame must not suppress detection. + // A racing ackProbe/abort reconciles via the nonce check and clearKeepalive. + this.clearProbeTimer(); + this.probeTimer = setTimeout(() => { + void this.handleProbeTimeout(nonce); + }, this.probeTimeoutMs ?? DEFAULT_OPEN_STREAM_PROBE_TIMEOUT_MS); try { // Bypass the operation queue so a stuck app write cannot block @@ -293,20 +299,12 @@ export class OpenStreamWriter { ); } catch { if (!this.active) { - return; + return; // probe timeout may have already finalized } this.pendingProbeNonce = undefined; + this.clearProbeTimer(); this.armIdle(); - return; - } - - if (!this.active || this.pendingProbeNonce !== nonce) { - return; } - - this.probeTimer = setTimeout(() => { - void this.handleProbeTimeout(nonce); - }, this.probeTimeoutMs ?? DEFAULT_OPEN_STREAM_PROBE_TIMEOUT_MS); } private async handleProbeTimeout(nonce: string): Promise {