feat(ocap-kernel): leverage libp2p v3 features in remote communications (#915)

Closes #903

## Summary

Adopts several libp2p v3 features to improve error handling,
observability, and robustness of the remote communications layer:

- **Async protocol handlers with auto-abort** — inbound connection
handler is now `async`, so handler rejections automatically abort the
stream instead of silently dropping errors
- **Fine-grained stream close events** — v3 streams emit close events
with `{ error, local }` properties for better diagnostics distinguishing
local vs remote and clean vs error closes
- **Stream inactivity timeout** — configurable
`streamInactivityTimeoutMs` (default 120s) detects dead streams via
bidirectional silence, with a 5s minimum floor to prevent self-DoS
- **`peer:disconnect` safety net** — listens for peer-level disconnect
events as a fallback when `readChannel` cleanup misses a disconnection,
with `signal.aborted` guard to avoid spurious reconnection during
shutdown
- **`StreamResetError` detection** — recognizes remote stream resets
without permanently marking the peer as intentionally closed (prevents
malicious peers from suppressing reconnection)
- **Graceful `closeChannel` rewrite** — uses `channel.stream.close()`
directly instead of the fragile `unwrap()` pattern, with abort fallback
- **Graceful shutdown** — `stop()` sends FIN to remote peers so they see
clean stream ends rather than `StreamResetError`
- **`Channel` type gains `stream`** — direct access to stream lifecycle
APIs, avoiding `unwrap()`
- **`TooManyOutboundProtocolStreamsError` handling** — re-throws
immediately since trying other addresses won't help
- **`writeWithTimeout` stream status guard** — fast-fails on
already-closed streams before attempting a write
- **`RemoteHandle.giveUp()` method** — transport-level give-up now
properly cleans up (stops ACK retransmission + rejects pending messages
+ rejects pending URL redemptions)
- **Flaky multi-peer reconnection E2E test** — fixed by adding
`maxConnectionAttemptsPerMinute: 500` (rate limiter was the root cause),
parallel restarts, and `ackTimeoutMs: 5000`

## Test plan

New unit tests cover stream status fast-fail (`it.each` over 4
statuses), `peer:disconnect` handler (registration, forwarding, relay
filtering, shutdown guard), stream close event diagnostics (4 scenarios
via `it.each`), `TooManyOutboundProtocolStreamsError` rethrow,
`StreamResetError` reconnection behavior, `closeChannel`
graceful/abort/double-fail paths, `stop()` stream closing,
`RemoteHandle.giveUp()`, and `streamInactivityTimeoutMs` query string
round-trip. New E2E test file (`libp2p-v3-features.test.ts`) covers
stream inactivity timeout recovery end-to-end.


## Suggested review order

1. `packages/ocap-kernel/src/remotes/types.ts` — `Channel` type gains
`stream`, new `PeerDisconnectHandler` and `streamInactivityTimeoutMs`
option. Read this first to understand the type changes everything else
builds on.
2. `packages/ocap-kernel/src/remotes/platform/constants.ts` — new
`STREAM_INACTIVITY_TIMEOUT_MS` and `MIN_STREAM_INACTIVITY_TIMEOUT_MS`
constants.
3. `packages/ocap-kernel/src/remotes/platform/channel-utils.ts` + test —
`writeWithTimeout` stream status guard (small, self-contained).
4. `packages/ocap-kernel/src/remotes/platform/connection-factory.ts` +
test — `peer:disconnect` listener, async inbound handler, `closeChannel`
rewrite, `TooManyOutboundProtocolStreamsError`. Core plumbing.
5. `packages/ocap-kernel/src/remotes/platform/transport.ts` + test —
biggest file: `StreamResetError` handling, `registerChannel` stream
events/timeout, `peer:disconnect` safety net, graceful `stop()`. Builds
on 3 and 4.
6. `packages/ocap-kernel/src/remotes/kernel/RemoteHandle.ts` + test —
new `giveUp()` method and `#handleAckTimeout` cleanup path.
7. `packages/ocap-kernel/src/remotes/kernel/RemoteManager.ts` + test —
`#handleRemoteGiveUp` wiring to `giveUp()`.
8. `packages/kernel-node-runtime/test/e2e/remote-comms.test.ts` — flaky
multi-peer reconnection fix (rate limit + parallel restarts).
9. `packages/kernel-node-runtime/test/e2e/libp2p-v3-features.test.ts` —
new E2E tests for the v3 features.
10. `packages/kernel-browser-runtime/src/utils/comms-query-string.ts` +
test — `streamInactivityTimeoutMs` query string support (trivial).

🤖 Generated with [Claude Code](https://claude.com/claude-code)

<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> **Medium Risk**
> Touches core remote-communications transport/retry logic (stream
lifecycle, reconnection triggers, and give-up cleanup), which can affect
message delivery and connection stability; changes are well-covered by
new/updated unit and E2E tests.
> 
> **Overview**
> Improves remote comms robustness by adopting libp2p v3
stream/connection signals: `Channel` now carries the underlying
`stream`, inbound protocol handlers can be async, and the transport
configures per-stream inactivity timeouts and logs detailed stream close
diagnostics.
> 
> Reworks failure handling and cleanup paths: fast-fail writes on
non-`open` streams, treat `StreamResetError`/`peer:disconnect` as
reconnection signals (with relay filtering and shutdown guards), handle
`TooManyOutboundProtocolStreamsError` as a non-retryable dial error, and
make shutdown/`closeChannel` prefer graceful `stream.close()` with abort
fallback.
> 
> Fixes give-up semantics so pending messages and URL redemptions are
always rejected via new `RemoteHandle.giveUp()` and updated
`RemoteManager` wiring, adds `streamInactivityTimeoutMs` support to
comms query strings, and expands/adjusts unit + E2E coverage (including
a new `libp2p-v3-features` E2E suite and less-flaky multi-peer
reconnection timing/rate-limit settings).
> 
> <sup>Reviewed by [Cursor Bugbot](https://cursor.com/bugbot) for commit
52fb402. Bugbot is set up for automated
code reviews on this repo. Configure
[here](https://www.cursor.com/dashboard/bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: sirtimid

packages/kernel-browser-runtime/src/utils/comms-query-string.test.ts

@@ -68,6 +68,7 @@ describe('comms-query-string', () => {
         allowedWsHosts: ['relay.example.com'],
         maxRetryAttempts: 5,
         maxQueue: 200,
+        streamInactivityTimeoutMs: 60000,
       };
       const params = createCommsQueryString(options);
       expect(parseCommsQueryString(`?${params.toString()}`)).toStrictEqual(

packages/kernel-browser-runtime/src/utils/comms-query-string.ts

@@ -57,6 +57,7 @@ const NUMBER_PARAM_NAMES = [
   'handshakeTimeoutMs',
   'writeTimeoutMs',
   'ackTimeoutMs',
+  'streamInactivityTimeoutMs',
 ] as const satisfies readonly NumberParamKey[];
 
 const NonNegativeInteger = min(integer(), 0);

packages/kernel-cli/src/commands/relay.test.ts

@@ -27,7 +27,10 @@ vi.mock('../utils.ts', () => ({
   waitFor: vi.fn(),
 }));
 
-const mockLogger = { info: vi.fn(), error: vi.fn() } as unknown as Logger;
+const mockLogger: Logger = {
+  info: vi.fn(),
+  error: vi.fn(),
+} as unknown as Logger;
 
 const makeLibp2pMock = (
   addrs: string[] = ['/ip4/127.0.0.1/tcp/9001/ws/p2p/QmFoo'],

packages/kernel-node-runtime/test/e2e/libp2p-v3-features.test.ts

@@ -0,0 +1,291 @@
+import type { Libp2p } from '@libp2p/interface';
+import { makeSQLKernelDatabase } from '@metamask/kernel-store/sqlite/nodejs';
+import { startRelay } from '@metamask/kernel-utils/libp2p';
+import { Kernel, kunser, makeKernelStore } from '@metamask/ocap-kernel';
+import { delay } from '@ocap/repo-tools/test-utils';
+import { mkdtemp, rm } from 'node:fs/promises';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import {
+  describe,
+  it,
+  expect,
+  beforeAll,
+  afterAll,
+  beforeEach,
+  afterEach,
+} from 'vitest';
+
+import { makeTestKernel } from '../helpers/kernel.ts';
+import {
+  makeRemoteVatConfig,
+  restartKernelAndReloadVat,
+  sendRemoteMessage,
+  setupAliceAndBob,
+} from '../helpers/remote-comms.ts';
+import { stopWithTimeout } from '../helpers/stop-with-timeout.ts';
+
+const NETWORK_TIMEOUT = 30_000;
+const relayPeerId = '12D3KooWJBDqsyHQF2MWiCdU4kdqx4zTsSTLRdShg7Ui6CRWB4uc';
+const testRelays = [`/ip4/127.0.0.1/tcp/9001/ws/p2p/${relayPeerId}`];
+
+const testBackoffOptions = {
+  reconnectionBaseDelayMs: 10,
+  reconnectionMaxDelayMs: 50,
+  handshakeTimeoutMs: 3_000,
+  writeTimeoutMs: 3_000,
+  ackTimeoutMs: 2_000,
+};
+
+describe.sequential('libp2p v3 Features E2E', () => {
+  let relay: Libp2p;
+  let kernel1: Kernel;
+  let kernel2: Kernel;
+  let dbFilename1: string;
+  let dbFilename2: string;
+  let tempDir: string;
+  let kernelStore1: ReturnType<typeof makeKernelStore>;
+  let kernelStore2: ReturnType<typeof makeKernelStore>;
+
+  beforeAll(async () => {
+    relay = await startRelay(console);
+  });
+
+  afterAll(async () => {
+    if (relay) {
+      await relay.stop();
+    }
+  });
+
+  beforeEach(async () => {
+    tempDir = await mkdtemp(join(tmpdir(), 'ocap-v3-'));
+    dbFilename1 = join(tempDir, 'kernel1.db');
+    dbFilename2 = join(tempDir, 'kernel2.db');
+
+    const kernelDatabase1 = await makeSQLKernelDatabase({
+      dbFilename: dbFilename1,
+    });
+    kernelStore1 = makeKernelStore(kernelDatabase1);
+
+    const kernelDatabase2 = await makeSQLKernelDatabase({
+      dbFilename: dbFilename2,
+    });
+    kernelStore2 = makeKernelStore(kernelDatabase2);
+
+    kernel1 = await makeTestKernel(kernelDatabase1);
+    kernel2 = await makeTestKernel(kernelDatabase2);
+  });
+
+  afterEach(async () => {
+    const STOP_TIMEOUT = 3000;
+    await Promise.all([
+      kernel1 &&
+        stopWithTimeout(
+          async () => kernel1.stop(),
+          STOP_TIMEOUT,
+          'kernel1.stop',
+        ),
+      kernel2 &&
+        stopWithTimeout(
+          async () => kernel2.stop(),
+          STOP_TIMEOUT,
+          'kernel2.stop',
+        ),
+    ]);
+    if (tempDir) {
+      await rm(tempDir, { recursive: true, force: true });
+    }
+  });
+
+  describe('peer:disconnect Reconnection', () => {
+    it(
+      'recovers queued message after peer:disconnect triggers reconnection',
+      async () => {
+        const { aliceRef, bobURL } = await setupAliceAndBob(
+          kernel1,
+          kernel2,
+          kernelStore1,
+          kernelStore2,
+          testRelays,
+          testBackoffOptions,
+        );
+
+        // Establish initial communication
+        const initial = await sendRemoteMessage(
+          kernel1,
+          aliceRef,
+          bobURL,
+          'hello',
+          ['Alice'],
+        );
+        expect(initial).toContain('vat Bob got "hello" from Alice');
+
+        // Stop kernel2 — triggers both readChannel error and peer:disconnect.
+        // The peer:disconnect event acts as a safety net ensuring reconnection
+        // is attempted even after readChannel clears the channel.
+        await kernel2.stop();
+
+        // Queue a message while kernel2 is down — this triggers reconnection
+        const recoveryPromise = kernel1.queueMessage(
+          aliceRef,
+          'sendRemoteMessage',
+          [bobURL, 'hello', ['Alice']],
+        );
+
+        // Restart kernel2 — reconnection loop delivers the queued message
+        const bobConfig = makeRemoteVatConfig('Bob');
+        const restartResult = await restartKernelAndReloadVat(
+          dbFilename2,
+          false,
+          testRelays,
+          bobConfig,
+          testBackoffOptions,
+        );
+        // eslint-disable-next-line require-atomic-updates
+        kernel2 = restartResult.kernel;
+
+        const result = kunser(await recoveryPromise) as string;
+        expect(result).toContain('vat Bob got "hello" from Alice');
+
+        // Verify ongoing connectivity after peer:disconnect recovery
+        const followUp = await sendRemoteMessage(
+          kernel1,
+          aliceRef,
+          bobURL,
+          'hello',
+          ['Alice'],
+        );
+        expect(followUp).toContain('vat Bob got "hello" from Alice');
+      },
+      NETWORK_TIMEOUT * 2,
+    );
+  });
+
+  describe('Stream Inactivity Timeout', () => {
+    it(
+      'recovers communication after idle period exceeds inactivity timeout',
+      async () => {
+        // Use a short inactivity timeout to test the auto-abort behavior.
+        // Must be >= MIN_STREAM_INACTIVITY_TIMEOUT_MS (5 s) since the
+        // transport clamps lower values.
+        const shortTimeoutOptions = {
+          ...testBackoffOptions,
+          streamInactivityTimeoutMs: 6_000,
+        };
+
+        const { aliceRef, bobURL } = await setupAliceAndBob(
+          kernel1,
+          kernel2,
+          kernelStore1,
+          kernelStore2,
+          testRelays,
+          shortTimeoutOptions,
+        );
+
+        // Establish initial communication
+        const initial = await sendRemoteMessage(
+          kernel1,
+          aliceRef,
+          bobURL,
+          'hello',
+          ['Alice'],
+        );
+        expect(initial).toContain('vat Bob got "hello" from Alice');
+
+        // Wait longer than the inactivity timeout (6s + buffer).
+        // The stream should auto-abort due to inactivityTimeout,
+        // triggering connection loss handling and reconnection.
+        await delay(8_000);
+
+        // Send another message — the transport layer should
+        // reconnect since the previous stream was aborted by inactivity.
+        const afterIdle = await sendRemoteMessage(
+          kernel1,
+          aliceRef,
+          bobURL,
+          'hello',
+          ['Alice'],
+        );
+        expect(afterIdle).toContain('vat Bob got "hello" from Alice');
+      },
+      NETWORK_TIMEOUT * 2,
+    );
+  });
+
+  describe('Fast Failure on Closed Streams', () => {
+    it(
+      'handles rapid sends during disconnect without hanging',
+      async () => {
+        const { aliceRef, bobURL, peerId2 } = await setupAliceAndBob(
+          kernel1,
+          kernel2,
+          kernelStore1,
+          kernelStore2,
+          testRelays,
+          testBackoffOptions,
+        );
+
+        // Establish initial connectivity
+        await sendRemoteMessage(kernel1, aliceRef, bobURL, 'hello', ['Alice']);
+
+        // Intentionally close the connection — writes should fail fast
+        // via stream.status check rather than waiting for timeout
+        await kernel1.closeConnection(peerId2);
+
+        const start = Date.now();
+        // This should fail quickly (stream.status !== 'open') rather than
+        // waiting for the full write timeout (3000ms)
+        await expect(
+          kernel1.queueMessage(aliceRef, 'sendRemoteMessage', [
+            bobURL,
+            'hello',
+            ['Alice'],
+          ]),
+        ).rejects.toMatchObject({
+          body: expect.stringContaining(
+            'Message delivery failed after intentional close',
+          ),
+        });
+        const elapsed = Date.now() - start;
+
+        // Should fail well under the write timeout
+        expect(elapsed).toBeLessThan(testBackoffOptions.writeTimeoutMs);
+      },
+      NETWORK_TIMEOUT,
+    );
+  });
+
+  describe('Connection Type Awareness', () => {
+    it(
+      'establishes relayed connections and communicates successfully',
+      async () => {
+        // Both kernels connect via relay — the connection.direct property
+        // should be false (relayed), and the log should include "relayed".
+        // This validates that the connection type detection works with real
+        // libp2p connections.
+        const { aliceRef, bobURL, peerId1, peerId2 } = await setupAliceAndBob(
+          kernel1,
+          kernel2,
+          kernelStore1,
+          kernelStore2,
+          testRelays,
+          testBackoffOptions,
+        );
+
+        // Verify distinct peer IDs (confirms real libp2p nodes)
+        expect(peerId1).not.toBe(peerId2);
+
+        // Bidirectional communication through relay
+        const response = await sendRemoteMessage(
+          kernel1,
+          aliceRef,
+          bobURL,
+          'hello',
+          ['Alice'],
+        );
+        expect(response).toContain('vat Bob got "hello" from Alice');
+      },
+      NETWORK_TIMEOUT,
+    );
+  });
+});