fix(ocap-kernel): plug second-round leaks in peer-incarnation restart path

Follow-up review of the issue #944 fix surfaced four real defects, each
fixed here:

1. RemoteHandle.#sendRemoteCommand swallowed PeerRestartedError and
   NetworkStoppedError. Only `intentional close` triggered the cleanup
   path; the other two terminal verdicts fell through to logger.error
   with the message persisted to remotePending and #nextSendSeq
   advanced, leaving it pending until ACK timeout × MAX_RETRIES. The
   catch now routes all terminal errors through the shared
   isTerminalSendError predicate.

2. RemoteManager.#handleIncarnationChange enqueued kernelQueue
   notifications and ran in-memory mutations inside the savepoint
   window. A kv rollback would not undo either, leaving the kernel
   with run-queue entries against promises the persisted store still
   considers unresolved. Split RemoteHandle.handlePeerRestart into
   `persistPeerRestart` (kv-only, savepoint-safe) and
   `finalizePeerRestart` (in-memory + redemption rejections); the
   manager now snapshots `getPromisesByDecider` before any kv mutation,
   runs `persistPeerRestart` inside the savepoint, and defers
   `finalizePeerRestart` and `resolvePromises` to after release.

3. doInboundHandshake silently ignored kernelDetectedRestart. The
   outbound path closes the freshly-dialed channel and throws
   PeerRestartedError on detected restart; the inbound path only
   logged. A concurrent in-flight outbound send could then write a
   pre-restart payload over a fresh-incarnation channel. Inbound now
   returns false when the kernel detects a restart, and the caller
   closes the channel without registering it — symmetric with the
   outbound guard.

4. deleteRemoteInfo left peerIncarnation.{peerId} stranded. A
   re-established remote with the same peerId would mis-classify its
   first handshake as a restart against the leftover incarnation. Now
   reads the info before deletion and clears the peerIncarnation row;
   corrupt JSON is logged and swallowed so the rest of the cleanup
   still runs.

Other tightening:
- #rejectAllPending is a no-op when nothing is pending, so the catch
  path is safe even when the kernel-side restart cleanup ran upstream.
- #handleIncarnationChange logs a warning when the restart verdict
  fires but no live RemoteHandle exists (boot race / peer teardown
  scoping issue), instead of silently advancing the persisted
  incarnation.
- Sentinel error matching uses the centralized `isTerminalSendError`
  imported from @metamask/kernel-errors (this commit's prerequisite),
  removing the brittle string-match in the predicate.
- Outbound PeerRestartedError now logs at info severity, not error —
  it is an expected, recoverable event.

Tests:
- New: PeerRestartedError/IntentionalCloseError/NetworkStoppedError
  from initial #sendRemoteCommand reject pending and fire onGiveUp.
- New: handlePeerRestart actually clears c-list state (seeds an
  exportFromEndpoint pair, asserts both halves gone after restart).
- New: PeerRestartedError does not trigger reconnection from the
  outbound catch path.
- New: inbound handshake closes the channel without registration when
  the kernel detects a restart.
- New: deleteRemoteInfo clears the peer-incarnation row.
- Updated: RemoteManager savepoint-rollback test now spies on
  persistPeerRestart and asserts finalizePeerRestart is NOT called
  when the persisted phase throws.
- Updated: #handleIncarnationChange verdict tests assert false on
  first observation and stable incarnation (not just true on restart).

E2E: extension/test/e2e/remote-comms.test.ts had its toBeVisible
budget raised to 50s (suite to 90s) to fit inside the 40s URL
redemption budget that the new awaited handshake RPC pushes the
round-trip into under CI load.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: sirtimid

packages/extension/test/e2e/remote-comms.test.ts

@@ -5,7 +5,7 @@ import { rm } from 'node:fs/promises';
 
 import { loadExtension, sessionPath } from '../helpers.ts';
 
-test.describe.configure({ mode: 'serial', timeout: 60_000 });
+test.describe.configure({ mode: 'serial', timeout: 90_000 });
 
 /**
  * End-to-end tests for remote communications functionality.
@@ -140,7 +140,11 @@ test.describe('Remote Communications', () => {
     const messageResponse = popupPage1.locator(
       '[data-testid="message-response"]',
     );
-    await expect(messageResponse).toBeVisible({ timeout: 30_000 });
+    // Budget: redemption timeout is `ackTimeoutMs * (MAX_RETRIES + 1)` =
+    // 40s with prod defaults; the response (success or rejection) is
+    // guaranteed to render before then. 50s gives 10s headroom for the
+    // post-redemption render path under CI load.
+    await expect(messageResponse).toBeVisible({ timeout: 50_000 });
     await expect(messageResponse).toContainText(
       // eslint-disable-next-line no-useless-escape
       `Response:{\"body\":\"#\\\"vat Bob got \\\\\\\"hello\\\\\\\" from remote Alice\\\"\",\"slots\":[]}`,

packages/ocap-kernel/src/remotes/kernel/RemoteHandle.test.ts

@@ -1674,4 +1674,79 @@ describe('RemoteHandle', () => {
       expect(onGiveUp).not.toHaveBeenCalled();
     });
   });
+
+  describe('first-send terminal errors', () => {
+    it.each([
+      [
+        'PeerRestartedError',
+        Object.assign(new Error('peer restarted'), {
+          name: 'PeerRestartedError',
+        }),
+      ],
+      [
+        'IntentionalCloseError',
+        Object.assign(new Error('intentional close'), {
+          name: 'IntentionalCloseError',
+        }),
+      ],
+      [
+        'NetworkStoppedError',
+        Object.assign(new Error('Network stopped'), {
+          name: 'NetworkStoppedError',
+        }),
+      ],
+    ])(
+      'rejects pending and fires onGiveUp when initial send rejects with %s',
+      async (_name, terminalError) => {
+        const onGiveUp = vi.fn();
+        const remote = RemoteHandle.make({
+          remoteId: mockRemoteId,
+          peerId: mockRemotePeerId,
+          kernelStore: mockKernelStore,
+          kernelQueue: mockKernelQueue,
+          remoteComms: mockRemoteComms,
+          ackTimeoutMs: 100,
+          onGiveUp,
+        });
+
+        vi.mocked(mockRemoteComms.sendRemoteMessage).mockRejectedValueOnce(
+          terminalError,
+        );
+
+        const redeem = remote.redeemOcapURL('ocap:something@peer,relay');
+        // Drain microtasks so the catch handler runs.
+        for (let i = 0; i < 5; i += 1) {
+          await Promise.resolve();
+        }
+
+        // The redemption rejects with the giveUp/rejectAllPending reason,
+        // which is the terminal error's message string.
+        await expect(redeem).rejects.toThrow(terminalError.message);
+        expect(onGiveUp).toHaveBeenCalledWith(mockRemotePeerId);
+      },
+    );
+  });
+
+  describe('handlePeerRestart c-list teardown', () => {
+    it('clears the peer’s "+"-direction c-list entries via forgetEndpointImports', async () => {
+      const remote = makeRemote();
+
+      // Seed an object export from the peer (peer-allocated eref ro+5
+      // mapped to a fresh kernel object).
+      const eref = 'ro+5';
+      const kref = mockKernelStore.exportFromEndpoint(mockRemoteId, eref);
+
+      // Sanity: c-list is populated in both directions before restart.
+      // Use the raw `krefToEref`/`erefToKref` lookups (not the translating
+      // wrappers, which flip RRef polarity for receiver-frame interpretation).
+      expect(mockKernelStore.erefToKref(mockRemoteId, eref)).toBe(kref);
+      expect(mockKernelStore.krefToEref(mockRemoteId, kref)).toBe(eref);
+
+      remote.handlePeerRestart();
+
+      // Both halves of the c-list pair are gone after restart.
+      expect(mockKernelStore.erefToKref(mockRemoteId, eref)).toBeUndefined();
+      expect(mockKernelStore.krefToEref(mockRemoteId, kref)).toBeUndefined();
+    });
+  });
 });

packages/ocap-kernel/src/remotes/kernel/RemoteHandle.ts

@@ -1,6 +1,7 @@
 import type { VatOneResolution } from '@agoric/swingset-liveslots';
 import type { CapData } from '@endo/marshal';
 import { makePromiseKit } from '@endo/promise-kit';
+import { isTerminalSendError } from '@metamask/kernel-errors';
 import { Logger } from '@metamask/logger';
 
 import {
@@ -34,33 +35,6 @@ const MAX_RETRIES = 3;
 /** Maximum number of pending messages awaiting ACK. */
 const MAX_PENDING_MESSAGES = 200;
 
-/**
- * Send-side errors that should abort retransmit instead of being treated as
- * a transient failure to retry. These come from definitive transport-level
- * verdicts: the user/peer closed the connection, the network is shutting
- * down, or the peer's incarnation changed (so any pending message was
- * generated against a now-dead session). Continuing to iterate just produces
- * identical failures on every queued seq until MAX_RETRIES exhausts.
- *
- * Identified by name (`PeerRestartedError`) for the incarnation case so the
- * check survives across the platform-services RPC boundary that might
- * unwrap the class identity, and by message for the others (which originate
- * as plain `Error`s in transport.ts).
- *
- * @param error - The error thrown by sendRemoteMessage.
- * @returns Whether the error indicates retransmit should stop.
- */
-function isTerminalSendError(error: unknown): boolean {
-  if (!(error instanceof Error)) {
-    return false;
-  }
-  return (
-    error.name === 'PeerRestartedError' ||
-    error.message.includes('intentional close') ||
-    error.message === 'Network stopped'
-  );
-}
-
 type RemoteHandleConstructorProps = {
   remoteId: RemoteId;
   peerId: string;
@@ -448,7 +422,7 @@ export class RemoteHandle implements EndpointHandle {
    * Retransmit all pending messages.
    *
    * Sends sequentially so a peer-restart detection during the first send can
-   * short-circuit the rest: clearRemoteSeqState (called by handlePeerRestart)
+   * short-circuit the rest: clearRemoteSeqState (called by persistPeerRestart)
    * deletes both the seq counters and the `remotePending.*` payloads, so
    * `getPendingMessage` returns undefined for subsequent iterations and the
    * loop bound (`seq <= this.#nextSendSeq`, now 0) terminates immediately.
@@ -495,12 +469,17 @@ export class RemoteHandle implements EndpointHandle {
   }
 
   /**
-   * Discard all pending messages due to delivery failure.
+   * Discard all pending messages due to delivery failure. Safe no-op when
+   * the queue is already empty — guards against clobbering kv state that a
+   * prior cleanup (e.g. {@link persistPeerRestart}) already cleared.
    *
    * @param reason - The reason for failure.
    */
   #rejectAllPending(reason: string): void {
     const pendingCount = this.#getPendingCount();
+    if (pendingCount === 0) {
+      return;
+    }
     for (let i = 0; i < pendingCount; i += 1) {
       this.#logger.warn(
         `Message ${this.#startSeq + i} delivery failed: ${reason}`,
@@ -635,25 +614,34 @@ export class RemoteHandle implements EndpointHandle {
       this.#startAckTimeout();
     }
 
-    // Send the message (non-blocking - don't wait for ACK)
+    // Send the message (non-blocking - don't wait for ACK).
+    //
+    // Terminal verdicts from the transport (peer restarted, intentional
+    // close, network stopped) mean the message we just persisted will
+    // never be delivered as-is: reject all pending now and signal give-up
+    // rather than letting the message linger until ACK timeout × MAX_RETRIES.
+    //
+    // For PeerRestartedError specifically, the kernel-side
+    // `onIncarnationChange` callback ran `persistPeerRestart` and
+    // `finalizePeerRestart` synchronously *before* the transport threw, so
+    // most of the cleanup below is already a no-op. The guards inside
+    // `#rejectAllPending` and `rejectPendingRedemptions` keep this safe;
+    // calling `#onGiveUp` again exercises an idempotent path.
     this.#remoteComms
       .sendRemoteMessage(this.#peerId, messageString)
       .catch((error) => {
-        // Handle intentional close errors specially - reject pending redemptions
-        if (
-          error instanceof Error &&
-          error.message.includes('intentional close')
-        ) {
+        if (isTerminalSendError(error)) {
+          const reason = (error as Error).message;
           this.#clearAckTimeout();
-          this.#rejectAllPending('intentional close');
-          this.rejectPendingRedemptions(
-            'Message delivery failed after intentional close',
-          );
-          // Notify RemoteManager to reject kernel promises for this remote
+          this.#rejectAllPending(reason);
+          this.rejectPendingRedemptions(reason);
           this.#onGiveUp?.(this.#peerId);
           return;
         }
-        this.#logger.error('Error sending remote message:', error);
+        this.#logger.error(
+          `${this.#peerId.slice(0, 8)}:: error sending remote message seq=${seq}:`,
+          error,
+        );
       });
   }
 
@@ -1164,61 +1152,55 @@ export class RemoteHandle implements EndpointHandle {
   }
 
   /**
-   * Handle a peer restart (incarnation change).
-   *
-   * Persisted writes happen first so that if `forgetEndpointImports` throws
-   * (e.g. corrupt c-list entry, refcount underflow), the caller's savepoint
-   * can roll back kv state and our in-memory fields stay consistent with
-   * the persisted view. Only after the kv writes succeed do we mutate
-   * in-memory counters, cancel timers, and reject pending URL redemption
-   * promises — those mutations are not reversible by a savepoint, so we
-   * defer them until we know the persisted side committed.
+   * Persist the peer-restart side effects (kv-only). Reversible by the
+   * caller's savepoint: if `forgetEndpointImports` throws (e.g. corrupt
+   * c-list entry, refcount underflow) and the caller rolls back, no
+   * in-memory state has been disturbed.
    *
-   * Called when the handshake detects that the remote peer has restarted.
+   * Pairs with {@link finalizePeerRestart}, which the caller MUST invoke
+   * exactly once after `releaseSavepoint` succeeds. Splitting the work
+   * keeps non-reversible mutations (timers, rejected redemption promises,
+   * in-memory seq counters) out of the savepoint window — those would
+   * leave the in-memory view inconsistent with the persisted view if the
+   * kv layer rolled back.
    */
-  handlePeerRestart(): void {
+  persistPeerRestart(): void {
     this.#logger.log(
       `${this.#peerId.slice(0, 8)}:: handling peer restart, resetting state`,
     );
-
-    // Snapshot pending state before any mutation, so we can log accurately
-    // even though the KV side will be cleared shortly.
-    const pendingCount = this.#getPendingCount();
-    const hadPending = this.#hasPendingMessages();
-
-    // --- Persisted writes (transactional with the caller's savepoint) ---
-
-    // Clear persisted sequence state (counters + remotePending.* payloads).
     this.#kernelStore.clearRemoteSeqState(this.remoteId);
-
-    // Drop the peer's contributions to our c-list (erefs they allocated)
-    // along with the owner / refcount / decider bookkeeping the kernel was
-    // holding on their behalf. Without this, fresh incarnations' reused eref
-    // labels would route into stale kernel state — e.g. an already-resolved
-    // promise from before the restart — and dead kernel objects would never
-    // be GC'd. May throw on corrupt state; the caller's savepoint will roll
-    // back the clearRemoteSeqState above and we will not have mutated any
-    // in-memory fields yet.
     this.#kernelStore.forgetEndpointImports(this.remoteId);
+  }
 
-    // --- In-memory state (only after persisted writes commit) ---
-
-    // Cancel timers.
-    this.#clearAckTimeout();
-    this.#clearDelayedAck();
+  /**
+   * Convenience wrapper that runs the persisted and in-memory phases
+   * back-to-back. Use only outside a savepoint window — transactional
+   * callers (e.g. {@link RemoteManager.handleIncarnationChange}) must call
+   * {@link persistPeerRestart} inside the savepoint and
+   * {@link finalizePeerRestart} after release, so a kv rollback can't
+   * leave the in-memory view inconsistent with the persisted view.
+   */
+  handlePeerRestart(): void {
+    this.persistPeerRestart();
+    this.finalizePeerRestart();
+  }
 
-    // Reject in-memory URL redemption promises (cannot be undone, so they
-    // run after persistence). Pending messages are tracked entirely in KV
-    // and have no JS-level promises to reject — clearRemoteSeqState above
-    // already removed their persisted records.
-    if (hadPending) {
+  /**
+   * Apply the in-memory side of a peer restart: cancel timers, reject
+   * in-flight URL redemption promises, and reset sequence counters. Must
+   * be called after {@link persistPeerRestart} and after the caller's
+   * savepoint has been released.
+   */
+  finalizePeerRestart(): void {
+    const pendingCount = this.#getPendingCount();
+    if (this.#hasPendingMessages()) {
       this.#logger.log(
         `${this.#peerId.slice(0, 8)}:: discarding ${pendingCount} pending messages due to peer restart`,
       );
     }
+    this.#clearAckTimeout();
+    this.#clearDelayedAck();
     this.rejectPendingRedemptions('Remote peer restarted');
-
-    // Reset in-memory sequence counters and flags for fresh start.
     this.#nextSendSeq = 0;
     this.#highestReceivedSeq = 0;
     this.#startSeq = 0;