Fix/open stream probe timeout crash

CHANGELOG.md

@@ -1,5 +1,11 @@
 # @contextvm/sdk
 
+## 0.14.1
+
+### Patch Changes
+
+- c66cbb5: Fix a fatal `unhandledRejection` in the CEP-41 server-side `OpenStreamWriter` keepalive path (process exit 1, every ~9h under load in the reported deployment). When a dead client had multiple open streams, the first writer's probe-timeout teardown evicted the client session; every sibling writer's probe-timeout abort then rejected with `No active session found for client` from `sendNotification`, and the timer callback discarded that rejection. Both keepalive timer callbacks now attach `.catch(() => undefined)`, mirroring `OpenStreamSession` — teardown state is complete before the abort publish, so dropping the rejection is safe. Also fixes the adjacent livelock: an undeliverable keepalive ping (evicted session, relays rejecting the event, transport closing — the relay pool already retries transient failures internally) used to re-arm the idle timer forever, hanging the stream and its deferred response; the writer now aborts with `Failed to send keepalive ping`, matching the client session's policy, while a rejection for a probe the peer already acked still recovers on the next idle cycle. Hardens `ApplesauceRelayPool.rebuild` the same way: a synchronous throw from relay/subscription reconstruction is logged instead of escaping the fire-and-forget rebuild as an unhandled rejection.
+
 ## 0.14.0
 
 ### Minor Changes

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@contextvm/sdk",
-  "version": "0.14.0",
+  "version": "0.14.1",
   "description": "ContextVM SDK for JavaScript/TypeScript",
   "license": "LGPL-3.0-1",
   "author": "ContextVM",

src/relay/applesauce-relay-pool.test.ts

@@ -1248,3 +1248,37 @@ describe('ApplesauceRelayPool.probe (integration)', () => {
     15_000,
   );
 });
+
+describe('ApplesauceRelayPool rebuild failure containment', () => {
+  test('a failing rebuild is logged, not surfaced as an unhandled rejection', async () => {
+    const unhandled: unknown[] = [];
+    const onUnhandled = (reason: unknown): void => {
+      unhandled.push(reason);
+    };
+    process.on('unhandledRejection', onUnhandled);
+
+    const pool = new ApplesauceRelayPool(['wss://relay.invalid.example']);
+    const testPool = pool as unknown as TestableApplesauceRelayPool;
+    const originalCreateRelay = testPool.createRelay.bind(testPool);
+    testPool.createRelay = (): Relay => {
+      throw new Error('relay construction failed');
+    };
+
+    try {
+      testPool.rebuild('test-rebuild-failure');
+
+      const deadline = Date.now() + 5_000;
+      while (testPool.rebuildInFlight && Date.now() < deadline) {
+        await sleep(10);
+      }
+      expect(testPool.rebuildInFlight).toBeUndefined();
+      // Give any would-be unhandled rejection a macrotask to surface.
+      await sleep(50);
+      expect(unhandled).toEqual([]);
+    } finally {
+      testPool.createRelay = originalCreateRelay;
+      process.off('unhandledRejection', onUnhandled);
+      await pool.disconnect();
+    }
+  }, 10_000);
+});

src/relay/applesauce-relay-pool.ts

@@ -846,8 +846,15 @@ export class ApplesauceRelayPool implements RelayHandler {
 
       // Resume ping monitor
       this.startPingMonitor();
-    })().finally(() => {
-      this.rebuildInFlight = undefined;
-    });
+    })()
+      .catch((error: unknown) => {
+        // Rebuild is fire-and-forget at every call site; a throw from relay
+        // or subscription reconstruction must not escape as an unhandled
+        // rejection and kill the host process.
+        logger.error('Relay pool rebuild failed', { reason, error });
+      })
+      .finally(() => {
+        this.rebuildInFlight = undefined;
+      });
   }
 }

src/transport/open-stream/writer.test.ts

@@ -512,6 +512,133 @@ describe('OpenStreamWriter keepalive', () => {
     await new Promise((resolve) => setTimeout(resolve, 60));
     expect(frames.map((frame) => frame.cvm.frameType)).not.toContain('abort');
   });
+
+  test('probe-timeout abort tolerates an already-evicted session', async () => {
+    // Regression (production crash): a sibling writer's probe-timeout
+    // teardown evicted the client session between our ping and our abort;
+    // the abort publish rejected and the discarded timer promise killed
+    // the process with an unhandled rejection.
+    const unhandled: unknown[] = [];
+    const onUnhandled = (reason: unknown): void => {
+      unhandled.push(reason);
+    };
+    process.on('unhandledRejection', onUnhandled);
+    try {
+      const frames: OpenStreamProgress[] = [];
+      const aborts: Array<string | undefined> = [];
+      const writer = new OpenStreamWriter({
+        progressToken: 'token-probe-evicted-session',
+        publishFrame: (frame): Promise<string | undefined> => {
+          frames.push(frame);
+          if (frame.cvm.frameType === 'abort') {
+            return Promise.reject(
+              new Error('No active session found for client: deadpubkey'),
+            );
+          }
+          return Promise.resolve(undefined);
+        },
+        onAbort: async (reason?: string): Promise<void> => {
+          aborts.push(reason);
+        },
+        idleTimeoutMs: 10,
+        probeTimeoutMs: 10,
+      });
+
+      await writer.start();
+      await waitFor(() => !writer.isActive);
+      // Let any would-be unhandled rejection surface before asserting.
+      await new Promise((resolve) => setTimeout(resolve, 30));
+
+      expect(aborts).toEqual(['Probe timeout']);
+      expect(frames.some((frame) => frame.cvm.frameType === 'abort')).toBe(
+        true,
+      );
+      expect(writer.signal.aborted).toBe(true);
+      expect(unhandled).toEqual([]);
+    } finally {
+      process.off('unhandledRejection', onUnhandled);
+    }
+  });
+
+  test('tears down the stream when keepalive pings are undeliverable', async () => {
+    // Regression (livelock): once the session is evicted every ping publish
+    // rejects; the old catch re-armed idle forever, so neither the stream
+    // nor its deferred response ever completed.
+    const unhandled: unknown[] = [];
+    const onUnhandled = (reason: unknown): void => {
+      unhandled.push(reason);
+    };
+    process.on('unhandledRejection', onUnhandled);
+    try {
+      const aborts: Array<string | undefined> = [];
+      const writer = new OpenStreamWriter({
+        progressToken: 'token-undeliverable-ping',
+        publishFrame: (frame): Promise<string | undefined> => {
+          if (
+            frame.cvm.frameType === 'ping' ||
+            frame.cvm.frameType === 'abort'
+          ) {
+            return Promise.reject(
+              new Error('No active session found for client: deadpubkey'),
+            );
+          }
+          return Promise.resolve(undefined);
+        },
+        onAbort: async (reason?: string): Promise<void> => {
+          aborts.push(reason);
+        },
+        idleTimeoutMs: 10,
+        probeTimeoutMs: 10,
+      });
+
+      await writer.start();
+      await waitFor(() => !writer.isActive);
+      await new Promise((resolve) => setTimeout(resolve, 30));
+
+      expect(aborts).toEqual(['Failed to send keepalive ping']);
+      expect(writer.signal.aborted).toBe(true);
+      expect(unhandled).toEqual([]);
+    } finally {
+      process.off('unhandledRejection', onUnhandled);
+    }
+  });
+
+  test('keeps the stream alive when a ping publish rejects after the peer acked', async () => {
+    // Mirrors OpenStreamSession: a late rejection for a probe the peer
+    // already acked must not abort a stream with proven liveness.
+    const aborts: Array<string | undefined> = [];
+    let pingCount = 0;
+    const writer = new OpenStreamWriter({
+      progressToken: 'token-late-reject-after-ack',
+      publishFrame: (frame): Promise<string | undefined> => {
+        if (frame.cvm.frameType === 'ping') {
+          pingCount += 1;
+          // Peer acks inside the publish window; the first publish then
+          // fails late — keepalive must recover on the next idle cycle.
+          writer.ackProbe(frame.cvm.nonce);
+          if (pingCount === 1) {
+            return Promise.reject(
+              new Error('No active session found for client: deadpubkey'),
+            );
+          }
+        }
+        return Promise.resolve(undefined);
+      },
+      onAbort: async (reason?: string): Promise<void> => {
+        aborts.push(reason);
+      },
+      idleTimeoutMs: 10,
+      probeTimeoutMs: 10,
+    });
+
+    await writer.start();
+    await waitFor(() => pingCount >= 2);
+    await new Promise((resolve) => setTimeout(resolve, 30));
+
+    expect(writer.isActive).toBe(true);
+    expect(aborts).toEqual([]);
+    writer.dispose();
+  });
 });
 
 describe('OpenStreamWriter signal', () => {

src/transport/open-stream/writer.ts

@@ -255,7 +255,7 @@ export class OpenStreamWriter {
 
     this.clearIdleTimer();
     this.idleTimer = setTimeout(() => {
-      void this.handleIdleTimeout();
+      this.handleIdleTimeout().catch(() => undefined);
     }, this.idleTimeoutMs);
   }
 
@@ -301,7 +301,7 @@ export class OpenStreamWriter {
     // A racing ackProbe/abort reconciles via the nonce check and clearKeepalive.
     this.clearProbeTimer();
     this.probeTimer = setTimeout(() => {
-      void this.handleProbeTimeout(nonce);
+      this.handleProbeTimeout(nonce).catch(() => undefined);
     }, this.probeTimeoutMs ?? DEFAULT_OPEN_STREAM_PROBE_TIMEOUT_MS);
 
     try {
@@ -316,12 +316,16 @@ export class OpenStreamWriter {
         }),
       );
     } catch {
-      if (!this.active) {
-        return; // probe timeout may have already finalized
+      // A matching ackProbe already reconciled this probe (or a newer probe
+      // superseded it): a late publish rejection must not abort a stream
+      // with proven liveness. Mirrors OpenStreamSession.handleIdleTimeout.
+      if (!this.active || this.pendingProbeNonce !== nonce) {
+        return;
       }
-      this.pendingProbeNonce = undefined;
-      this.clearProbeTimer();
-      this.armIdle();
+      // Undeliverable ping (evicted session, relays rejecting the event,
+      // transport closing): the relay pool already retries transient
+      // failures internally, so retrying here would loop forever. Tear down.
+      await this.abort('Failed to send keepalive ping');
     }
   }