feat(ocap-kernel): launch subcluster vats in parallel (#983)

## Explanation

Subcluster startup was O(sum of vat init times) because
`#launchVatsForSubcluster` used a serial `for...of` loop — each vat's
`initVat` RPC handshake had to complete before the next one began.

This PR changes startup to O(max vat init time) by launching all vats
concurrently via `Promise.all`. The mechanism:

1. Before any `await`, one kernel promise (`kp<N>`) is pre-allocated per
vat and marked as kernel-decided.
2. The bootstrap message is queued immediately, targeting the bootstrap
vat's unresolved `kp<N>`. `KernelRouter` parks the send on the promise
via `enqueuePromiseMessage`.
3. All vats launch in parallel. As each vat's `initVat` handshake
completes, its root kernel promise is resolved via
`resolvePromises('kernel', ...)`, and the run loop forwards any queued
messages.
4. The bootstrap vat receives kernel promise KRefs for all peer vats in
its `bootstrap(roots, services)` call — it can pipeline calls to peers
while they are still initializing.

## Changes

- **`SubclusterManager.ts`** — rewrote `#launchVatsForSubcluster` to
pre-allocate kernel promises, queue bootstrap immediately, and launch
all vats with `Promise.all`.
- **`SubclusterManager.test.ts`** — updated mocks (`initKernelPromise`,
`setPromiseDecider`, `resolvePromises`) and assertions to match the new
flow.
- **`Kernel.test.ts`** — added `resolvePromises = vi.fn()` to the
`KernelQueue` mock class.

## Checklist

- [x] Tests pass (`yarn workspace @MetaMask/ocap-kernel test:dev:quiet`)
- [x] Build passes (`yarn workspace @MetaMask/ocap-kernel build`)
- [ ] Changelog updated

<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> **Medium Risk**
> Changes core subcluster launch ordering, failure propagation, and
cleanup in the kernel; mistakes could leak workers or break bootstrap
peer wiring, though coverage is expanded in unit and integration tests.
> 
> **Overview**
> **Parallel subcluster startup** replaces serial vat launch in
`SubclusterManager` with `Promise.allSettled`, so subcluster bring-up
time tracks the slowest vat instead of the sum of init times. Kernel
service resolution still runs before any vat starts.
> 
> After launches settle, bootstrap is invoked with real root `ko` refs
for successful vats. **Failed peer vats** get an immediately rejected
kernel promise (`VAT_TERMINATED`) in the `roots` map so bootstrap can
observe the failure via pipelined `E(roots.peer)` calls;
`launchSubcluster` still rejects once bootstrap has run.
**`SubclusterLaunchResult`** now includes **`vatRootKrefs`** (name →
root kref for vats that launched successfully).
> 
> Failed launches **tear down** any vats that did start
(`#terminateVatQuietly` in reverse order) before IO/subcluster rollback.
Vat cleanup skips a baseline refcount decrement when GC already zeroed
reachability (`vat.ts`).
> 
> Integration tests drop hardcoded `ko4`/`ko5`/`ko6` and use
`vatRootKrefs` / `rootKref` from `launchSubcluster`; a new **peer
rejection** integration test and bootstrap vat bundle were added.
Changelog documents concurrent launch and peer rejection behavior.
> 
> <sup>Reviewed by [Cursor Bugbot](https://cursor.com/bugbot) for commit
6a2492a. 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 Sonnet 4.6 <noreply@anthropic.com>

Author: grypez

packages/kernel-test/src/cluster-launch.test.ts

@@ -1,4 +1,5 @@
 import { makeSQLKernelDatabase } from '@metamask/kernel-store/sqlite/nodejs';
+import { waitUntilQuiescent } from '@metamask/kernel-utils';
 import { Logger } from '@metamask/logger';
 import type { LogEntry } from '@metamask/logger';
 import type { Kernel } from '@metamask/ocap-kernel';
@@ -11,7 +12,10 @@ import {
   makeTestLogger,
 } from './utils.ts';
 
-describe('cluster initialization', { timeout: 4_000 }, () => {
+// No per-suite timeout override: the package's 30s default covers the kernel
+// construction each test does in `beforeEach`, which is slow enough on a
+// loaded CI runner to blow a tighter budget.
+describe('cluster initialization', () => {
   let logger: Logger;
   let entries: LogEntry[];
   let kernel: Kernel;
@@ -123,3 +127,50 @@ describe('cluster initialization', { timeout: 4_000 }, () => {
     ]);
   });
 });
+
+describe('peer rejection propagation', () => {
+  let logger: Logger;
+  let entries: LogEntry[];
+  let kernel: Kernel;
+
+  beforeEach(async () => {
+    const testLogger = makeTestLogger();
+    logger = testLogger.logger;
+    entries = testLogger.entries;
+    const database = await makeSQLKernelDatabase({});
+    kernel = await makeKernel(
+      database,
+      true,
+      logger.subLogger({ tags: ['test'] }),
+    );
+  });
+
+  it('bootstrap observes peer rejection when a peer vat fails to launch', async () => {
+    await expect(
+      kernel.launchSubcluster({
+        bootstrap: 'main',
+        vats: {
+          main: {
+            bundleSpec: getBundleSpec('peer-rejection-bootstrap'),
+            parameters: {},
+          },
+          peer: {
+            bundleSpec: getBundleSpec('error-build-throw'),
+            parameters: {},
+          },
+        },
+      }),
+    ).rejects.toMatchObject({
+      message: expect.stringMatching(/^Failed to launch vat \S+ \(peer\)$/u),
+    });
+
+    // Let the kernel run loop deliver the parked bootstrap message to the
+    // main vat, which will observe the peer's rejected root promise.
+    await waitUntilQuiescent(200);
+
+    const vatLogs = extractTestLogs(entries, 'console');
+    expect(vatLogs).toContainEqual(
+      expect.stringMatching(/^peer rejected:.*VAT_TERMINATED/u),
+    );
+  });
+});

packages/kernel-test/src/persistence.test.ts

@@ -96,10 +96,15 @@ describe('persistent storage', { timeout: 20_000 }, () => {
       false,
       logger.logger.subLogger({ tags: ['test'] }),
     );
-    const result1 = await runTestVats(kernel1, multiVatCluster);
-    expect(result1).toBe('Coordinator initialized with 2 workers');
+    // Capture rootKref directly: concurrent vat launch means the coordinator
+    // may not be assigned ko4, so we cannot use a hardcoded ref here.
+    const { bootstrapResult: launch1Result, rootKref: coordinatorRoot } =
+      await kernel1.launchSubcluster(multiVatCluster);
     await waitUntilQuiescent();
-    const workResult1 = await runResume(kernel1, v1Root);
+    expect(kunser(launch1Result as CapData<string>)).toBe(
+      'Coordinator initialized with 2 workers',
+    );
+    const workResult1 = await runResume(kernel1, coordinatorRoot);
     expect(workResult1).toBe('Work completed: Worker1(1), Worker2(1)');
     await waitUntilQuiescent();
     await kernel1.stop();
@@ -110,7 +115,8 @@ describe('persistent storage', { timeout: 20_000 }, () => {
       logger.logger.subLogger({ tags: ['test'] }),
     );
     await new Promise((resolve) => setTimeout(resolve, 1000));
-    const workResult2 = await runResume(kernel2, v1Root);
+    // coordinatorRoot (ko<N>) is stable across kernel restarts.
+    const workResult2 = await runResume(kernel2, coordinatorRoot);
     expect(workResult2).toBe('Work completed: Worker1(2), Worker2(2)');
     await kernel2.stop();
   });

packages/kernel-test/src/rejection.test.ts

@@ -31,7 +31,7 @@ describe('rejection', () => {
     });
     expect(vat).toBeDefined();
     const vats = kernel.getVatIds();
-    expect(vats).toStrictEqual(vatIds);
+    expect([...vats].sort()).toStrictEqual([...vatIds].sort());
 
     await waitUntilQuiescent();
     const vatLogs = vatIds.map((vatId) => extractTestLogs(entries, vatId));

packages/kernel-test/src/resume.test.ts

@@ -1,14 +1,14 @@
+import type { CapData } from '@endo/marshal';
 import { makeSQLKernelDatabase } from '@metamask/kernel-store/sqlite/nodejs';
 import { waitUntilQuiescent } from '@metamask/kernel-utils';
-import type { KRef } from '@metamask/ocap-kernel';
+import { kunser } from '@metamask/ocap-kernel';
 import { describe, expect, it } from 'vitest';
 
 import {
   getBundleSpec,
   makeKernel,
   makeTestLogger,
   runResume,
-  runTestVats,
   sortLogs,
   extractTestLogs,
 } from './utils.ts';
@@ -100,21 +100,22 @@ const reference = sortLogs([
   ...carolResumeReference,
 ]);
 
-// Vat root objects start with ko4 due to the kernel facet and other kernel service objects being created before any vats.
-const v1Root: KRef = 'ko4';
-const v2Root: KRef = 'ko5';
-const v3Root: KRef = 'ko6';
-
 describe('restarting vats', async () => {
   it('exercise restart vats individually', async () => {
     const kernelDatabase = await makeSQLKernelDatabase({
       dbFilename: ':memory:',
     });
     const { logger, entries } = makeTestLogger();
     const kernel = await makeKernel(kernelDatabase, true, logger);
-    const bootstrapResult = await runTestVats(kernel, testSubcluster);
-    expect(bootstrapResult).toBe('bootstrap Alice');
+    // Use launchSubcluster directly to get vatRootKrefs: concurrent vat launch
+    // means ko<N> assignment order depends on worker startup speed.
+    const { bootstrapResult, vatRootKrefs } =
+      await kernel.launchSubcluster(testSubcluster);
     await waitUntilQuiescent();
+    expect(kunser(bootstrapResult as CapData<string>)).toBe('bootstrap Alice');
+    const v1Root = vatRootKrefs.alice;
+    const v2Root = vatRootKrefs.bob;
+    const v3Root = vatRootKrefs.carol;
     await kernel.restartVat('v1');
     await kernel.restartVat('v2');
     await kernel.restartVat('v3');
@@ -136,9 +137,15 @@ describe('restarting vats', async () => {
     });
     const { logger: logger1, entries: entries1 } = makeTestLogger();
     const kernel1 = await makeKernel(kernelDatabase, true, logger1);
-    const bootstrapResult = await runTestVats(kernel1, testSubcluster);
-    expect(bootstrapResult).toBe('bootstrap Alice');
+    // Capture vatRootKrefs from first kernel: ko<N> refs are stable across
+    // kernel restarts because they are persisted in the kernel store.
+    const { bootstrapResult, vatRootKrefs } =
+      await kernel1.launchSubcluster(testSubcluster);
     await waitUntilQuiescent();
+    expect(kunser(bootstrapResult as CapData<string>)).toBe('bootstrap Alice');
+    const v1Root = vatRootKrefs.alice;
+    const v2Root = vatRootKrefs.bob;
+    const v3Root = vatRootKrefs.carol;
     const { logger: logger2, entries: entries2 } = makeTestLogger();
     const kernel2 = await makeKernel(kernelDatabase, false, logger2);
     await new Promise((resolve) => setTimeout(resolve, 1000));

packages/kernel-test/src/vats/peer-rejection-bootstrap.ts

@@ -0,0 +1,28 @@
+import { E } from '@endo/eventual-send';
+import { makeDefaultExo } from '@metamask/kernel-utils/exo';
+
+/**
+ * Bootstrap vat for testing peer-rejection propagation.
+ * Receives a `peer` root reference that may be a rejected kernel promise
+ * (e.g. if the peer vat failed to launch), and logs whether calls resolve
+ * or reject so integration tests can inspect the outcome.
+ *
+ * @returns The root object for this vat.
+ */
+// eslint-disable-next-line @typescript-eslint/explicit-function-return-type
+export function buildRootObject() {
+  return makeDefaultExo('root', {
+    async bootstrap({ peer }: { peer: unknown }) {
+      await E(peer as object)
+        .ping()
+        // eslint-disable-next-line no-console
+        .then(() => console.log('peer resolved'))
+        .catch((error: unknown) => {
+          const message =
+            error instanceof Error ? error.message : String(error);
+          // eslint-disable-next-line no-console
+          console.log(`peer rejected: ${message}`);
+        });
+    },
+  });
+}

packages/ocap-kernel/CHANGELOG.md

@@ -17,6 +17,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   - `error` is the failure's message and `detail` its whole cause chain, because only strings cross the wire: when a crank dies and its rollback then fails, the message names the rollback and only the chain names what killed the kernel
   - **BREAKING:** `runLoop` is required, so `KernelStatus` gains a mandatory property and a `getStatus` reply from a kernel built before this field fails result validation outright. It cannot be made optional: `exactOptional` would leave the type and the validator disagreeing inside a `type()`, and `optional` widens the property to `| undefined`, which an RPC result may not be
 - Add `onRunLoopFailure` to `Kernel.make` options, called with the error that killed the run loop so an embedder that outlives the kernel can exit or restart ([#1005](https://github.com/MetaMask/ocap-kernel/pull/1005))
+- Launch all vats in a subcluster concurrently during `launchSubcluster`, reducing startup latency from serial to parallel; failed peer vats receive a rejected kernel promise observable via `E(roots.peer).method()` pipelining ([#983](https://github.com/MetaMask/ocap-kernel/pull/983))
 - Add `fetch`, `Request`, `Headers`, and `Response` to available vat endowments ([#942](https://github.com/MetaMask/ocap-kernel/pull/942))
   - Add `VatConfig.network: { allowedHosts: string[] }`; requesting `'fetch'` without it rejects `initVat`
 - Integrate Snaps attenuated endowment factories into vat globals ([#937](https://github.com/MetaMask/ocap-kernel/pull/937))

packages/ocap-kernel/src/Kernel.test.ts

@@ -76,6 +76,8 @@ const mocks = vi.hoisted(() => {
     }
 
     waitForCrank = vi.fn().mockResolvedValue(undefined);
+
+    resolvePromises = vi.fn();
   }
 
   class RemoteManager {
@@ -341,6 +343,7 @@ describe('Kernel', () => {
         subclusterId: 's1',
         bootstrapResult: { body: '{"result":"ok"}', slots: [] },
         rootKref: expect.stringMatching(/^ko\d+$/u),
+        vatRootKrefs: { alice: expect.stringMatching(/^ko\d+$/u) },
       });
     });
   });

packages/ocap-kernel/src/store/methods/vat.ts

@@ -44,7 +44,7 @@ export function getVatMethods(ctx: StoreContext) {
     getKernelPromise,
     addPromiseSubscriber,
   } = getPromiseMethods(ctx);
-  const { initKernelObject } = getObjectMethods(ctx);
+  const { initKernelObject, getObjectRefCount } = getObjectMethods(ctx);
   const { addCListEntry } = getCListMethods(ctx);
   const { incrementRefCount, decrementRefCount } = getRefCountMethods(ctx);
 
@@ -261,8 +261,11 @@ export function getVatMethods(ctx: StoreContext) {
       const { vatSlot } = getReachableAndVatSlot(vatID, kref);
       ctx.kv.delete(getSlotKey(vatID, kref));
       ctx.kv.delete(getSlotKey(vatID, vatSlot));
-      // Decrease refcounts that belonged to the terminating vat
-      decrementRefCount(kref, 'cleanup|export|baseline');
+      // Skip baseline decrement if GC already zeroed reachable via dropImports.
+      const { reachable } = getObjectRefCount(kref);
+      if (reachable > 0) {
+        decrementRefCount(kref, 'cleanup|export|baseline');
+      }
       ctx.maybeFreeKrefs.add(kref);
       work.exports += 1;
     }

packages/ocap-kernel/src/types.ts

@@ -746,6 +746,8 @@ export type SubclusterLaunchResult = {
   rootKref: KRef;
   /** The CapData result of calling bootstrap() on the root object, if any. */
   bootstrapResult: CapData<KRef> | undefined;
+  /** Map from vat name to root kref for all successfully launched vats. */
+  vatRootKrefs: Record<string, KRef>;
 };
 
 const RemoteCommsDisconnectedStruct = object({