fix(open-stream): arm keepalive probe before ping, expose staleness

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

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",

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<string | undefined> = [];
+    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<void> => {
+        pings.push(nonce);
+        return new Promise<void>((resolve) => {
+          resolvePing = resolve;
+        });
+      },
+      sendAbort: async (reason?: string): Promise<void> => {
+        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<string | undefined> = [];
+    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<void> => {
+        throw new Error('relay down');
+      },
+      onAbort: async (reason?: string): Promise<void> => {
+        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<void> => undefined,
+      sendAbort: async (): Promise<void> => 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);
+  });
+});

src/transport/open-stream/session.ts

@@ -90,6 +90,7 @@ export class OpenStreamSession implements OpenStreamSessionLike<string> {
   private idleTimer: ReturnType<typeof setTimeout> | undefined;
   private probeTimer: ReturnType<typeof setTimeout> | undefined;
   private closeGraceTimer: ReturnType<typeof setTimeout> | undefined;
+  private lastActivityTimestamp = Date.now();
 
   constructor(options: OpenStreamSessionOptions) {
     this.progressToken = options.progressToken;
@@ -119,6 +120,28 @@ export class OpenStreamSession implements OpenStreamSessionLike<string> {
     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<void> {
     if (!this.active) {
       return;
@@ -379,6 +402,8 @@ export class OpenStreamSession implements OpenStreamSessionLike<string> {
   }
 
   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<string> {
 
     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<void> {

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<string | undefined> = [];
+    let resolvePing: () => void = () => undefined;
+    const writer = new OpenStreamWriter({
+      progressToken: 'token-stuck-publish',
+      publishFrame: (frame): Promise<string | undefined> => {
+        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<string | undefined>((resolve) => {
+            resolvePing = () => resolve(undefined);
+          });
+        }
+        return Promise.resolve(undefined);
+      },
+      onAbort: async (reason?: string): Promise<void> => {
+        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({

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<void> {