diff --git a/packages/kernel-store/src/sqlite/nodejs.transaction-survival.test.ts b/packages/kernel-store/src/sqlite/nodejs.transaction-survival.test.ts index 27cdd35a0..306a014e0 100644 --- a/packages/kernel-store/src/sqlite/nodejs.transaction-survival.test.ts +++ b/packages/kernel-store/src/sqlite/nodejs.transaction-survival.test.ts @@ -13,8 +13,8 @@ import type { KernelDatabase } from '../types.ts'; * * Both hold only while the abort doing the discarding succeeds. When it does * not, the driver logs it and refuses every later write rather than let one - * join a transaction nothing will commit; `ctx.savepoints` is truncated to - * zero either way, so it no longer matches the database. + * join a transaction nothing will commit; `_spStack` is emptied either way, so + * it no longer matches the database. */ /** Every statement and exec call, in order. */ diff --git a/packages/kernel-store/src/sqlite/wasm.transaction-survival.test.ts b/packages/kernel-store/src/sqlite/wasm.transaction-survival.test.ts index 6aadfcd2b..5d6e93fc0 100644 --- a/packages/kernel-store/src/sqlite/wasm.transaction-survival.test.ts +++ b/packages/kernel-store/src/sqlite/wasm.transaction-survival.test.ts @@ -280,6 +280,10 @@ describe('the wasm driver after a failure it tolerates', () => { message: 'SQLITE_IOERR: ROLLBACK TRANSACTION', }), ); + // A stack left populated makes `commitIfNeeded`'s "savepoints remain" + // early return permanent: the driver goes on accepting writes, begins a + // transaction for them, and never commits. + expect(mockDb._spStack).toStrictEqual([]); }, ); }); diff --git a/packages/kernel-test/src/crank-rollback.test.ts b/packages/kernel-test/src/crank-rollback.test.ts index cc71c5cbc..fd97a2355 100644 --- a/packages/kernel-test/src/crank-rollback.test.ts +++ b/packages/kernel-test/src/crank-rollback.test.ts @@ -205,6 +205,36 @@ describe('crank rollback against a real database', () => { kernelStore.endCrank(); }); + // The set is not per-crank: only `collectGarbage` empties it, and that runs at + // the end of a crank that had an item. So a candidate created while the run + // loop was idle — `terminateVat` unpinning a root is the real path — is still + // owed a collection, and an unrelated crank's rollback must not cancel it. + it('keeps GC candidates that predate the crank it rolled back', async () => { + const { kernelStore } = await makeStore(); + const idle = kernelStore.initKernelPromise()[0]; + kernelStore.decrementRefCount(idle, 'test'); + + kernelStore.startCrank(); + kernelStore.createCrankSavepoint('start'); + const abandoned = kernelStore.initKernelPromise()[0]; + kernelStore.decrementRefCount(abandoned, 'test'); + kernelStore.rollbackCrank('start'); + kernelStore.endCrank(); + + kernelStore.startCrank(); + kernelStore.createCrankSavepoint('start'); + kernelStore.collectGarbage(); + kernelStore.endCrank(); + + // Collected, because it was owed before the abandoned crank began. + expect(() => kernelStore.getKernelPromise(idle)).toThrow( + 'unknown kernel promise', + ); + }); + + // `createCrankSavepoint` records the name only once the database has the + // savepoint. Asking to roll back one that was never created must therefore say + // so, rather than releasing someone else's savepoint. it('refuses to roll back a savepoint that was never created', async () => { const { kernelStore } = await makeStore(); diff --git a/packages/kernel-test/src/garbage-collection.test.ts b/packages/kernel-test/src/garbage-collection.test.ts index fb20462b6..c1bba7b32 100644 --- a/packages/kernel-test/src/garbage-collection.test.ts +++ b/packages/kernel-test/src/garbage-collection.test.ts @@ -256,23 +256,40 @@ describe('Garbage Collection', () => { /** * Give an importer a chance to notice a dropped object and tell the kernel. * + * Waits for `done` as well as for an empty action set, because an empty set + * is also what "the vat has not told us anything yet" looks like. A vat + * reports a dropped import only once the engine has actually collected it, + * and `gcAndFinalize` can only provoke that, not guarantee it on the first + * try — so a round that reports nothing has to be retried rather than read + * as the end of the story. Reaped afresh each round for the same reason: + * the report rides on a `bringOutYourDead`. + * * @param vatId - The vat to reap. * @param rootKRef - That vat's root, to poke with cranks afterwards. - * @param settled - Whether the state under test has arrived yet. + * @param done - The outcome being waited for. */ async function reapAndSettle( vatId: VatId, rootKRef: KRef, - settled: () => boolean, + done: () => boolean, ): Promise { - // Reap until the vat's GC is visible rather than a fixed number of times: - // three was enough on an idle machine and not under a loaded one, which - // made this the last flake in the file. - for (let attempt = 0; attempt < 5 && !settled(); attempt += 1) { + const maxRounds = 10; + for (let round = 0; round < maxRounds; round++) { kernel.reapVats((id) => id === vatId); + // BOYD has to reach the vat, the vat has to answer, and the kernel has + // to act on the answer — but a round can queue more work, so loop until + // the queue is actually empty rather than guessing at a crank count. await kernel.queueMessage(rootKRef, 'noop', []); await waitUntilQuiescent(500); + if ([...kernelStore.getGCActions()].length === 0 && done()) { + return; + } } + throw Error( + `GC did not settle after ${maxRounds} rounds; actions pending: ${ + [...kernelStore.getGCActions()].join(', ') || '(none)' + }`, + ); } it('survives until both importers let go', async () => { @@ -306,10 +323,10 @@ describe('Garbage Collection', () => { await kernel.queueMessage(importerKRef, 'makeWeak', [objectId]); await kernel.queueMessage(importerKRef, 'forgetImport', []); await waitUntilQuiescent(); - await reapAndSettle(importerVatId, importerKRef, () => - kernelStore - .getImporters(sharedKRef) - .every((vatId) => vatId !== importerVatId), + await reapAndSettle( + importerVatId, + importerKRef, + () => !kernelStore.getImporters(sharedKRef).includes(importerVatId), ); // The exporter must not have been told to drop it: the second importer diff --git a/packages/ocap-kernel/CHANGELOG.md b/packages/ocap-kernel/CHANGELOG.md index 83cb0bbcd..a2a0c583f 100644 --- a/packages/ocap-kernel/CHANGELOG.md +++ b/packages/ocap-kernel/CHANGELOG.md @@ -86,7 +86,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Waiting for the store spans a whole crank, and a promise created in that crank was left with no decider and nothing to settle it, so the vat that sent the message waited forever - Keep a crank and a savepoint taken through `KernelStore.createSavepoint` from overlapping ([#1021](https://github.com/Consensys-Incorporated/ocap-kernel/pull/1021)) - Those savepoints bypass `ctx.savepoints` and so are invisible to `createCrankSavepoint`'s ordinal naming. One open across a crank left `releaseAllSavepoints` releasing `t0` into it rather than to a commit, so the caller's later rollback discarded the whole crank; one opened inside a crank was cancelled by a delivery rollback it had nothing to do with, or made durable by a release beneath it, after the peer had already been told the message was committed - - Both directions are now refused, and the two production callers — `RemoteHandle.handleRemoteMessage` and `RemoteManager`'s incarnation change — take their turn through the new `beginOutOfCrank`/`endOutOfCrank`, which the run loop consults via `outOfCrankWorkPending` before starting each crank. Work between the two must be synchronous + - Both directions are now refused, and the production callers take their turn through the new `withStoreOutOfCrank`, which the run loop consults via `outOfCrankWorkPending` before starting each crank - `handleRemoteMessage` now decodes an incoming `redeemURL` before opening its savepoint instead of awaiting inside it; the decode writes nothing, so the message stays atomic - Consequence: an inbound remote message or a peer's handshake arriving mid-crank waits for that crank to end, so their latency is now bounded by the slowest crank rather than being independent of it. A crank that sends to a remote can be slow. The alternative is to route inbound messages through the run queue as SwingSet's comms vat does, which is a larger change than this one - The reference count audit runs after the crank is committed rather than inside it, so a violation kills the run loop instead of rolling back a delivery whose caller the crank buffer flush had already answered ([#1021](https://github.com/Consensys-Incorporated/ocap-kernel/pull/1021)) @@ -106,6 +106,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **A store written by an earlier version must be reset.** There is no migration: every object in it is still at `(1, 1)` and no vat root is pinned, so the second importer's `dropImports` underflows mid-crank and the last importer's drop can retire a live vat's root. Pins also moved from a single `pinnedObjects` row to a count per object at `pinned.${kref}`, and the old row is no longer read by anything — so every pin in such a store is silently lost on open while the refcount unit each one took remains. `recomputeRefCounts` can rebuild the counts, but not the pins, so it is a diagnostic rather than an upgrade path - Pin vat root objects for the lifetime of their vat, and release the pin on termination ([#1020](https://github.com/MetaMask/ocap-kernel/pull/1020)) - A root is addressable while its vat lives whether or not anyone imports it; the old `(1, 1)` birth baseline was standing in for this + - The pin is released when a relaunch fails too, which vat cleanup does not do ([#1023](https://github.com/Consensys-Incorporated/ocap-kernel/pull/1023)) - Garbage-collection action delivery now moves the kernel's own c-list: `dropExports` clears the owner's reachable flag and `retireExports`/`retireImports` tear the entry down ([#1020](https://github.com/MetaMask/ocap-kernel/pull/1020)) - Previously the owner's flag never cleared, so the same action could be re-derived, and retired entries outlived the objects they named - Orphan a kernel object when its owner stops naming it (a delivered `retireExport`, or a `retireExports`/`abandonExports` syscall), so its `owner` and `refCount` records no longer outlive the c-list entry they were reachable through ([#1022](https://github.com/Consensys-Incorporated/ocap-kernel/pull/1022)) @@ -113,16 +114,39 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Reject a `retireExports`/`abandonExports` syscall for an object the calling endpoint does not own, instead of letting it erase another endpoint's claim to an object it is still exporting ([#1022](https://github.com/Consensys-Incorporated/ocap-kernel/pull/1022)) - Nothing upstream of `performExportCleanup` checked that the vref it was handed is even an export, and the audit could not see the damage, because an export entry carries no count - Garbage-collection action delivery releases the kernel's side even when the endpoint has vanished, and rolls back rather than committing a release the endpoint was never told about ([#1022](https://github.com/Consensys-Incorporated/ocap-kernel/pull/1022)) - - It releases only where the endpoint is genuinely gone: a terminated vat, whose cleanup tears the whole c-list down anyway, or a remote, which reconciles on its next incarnation. A vat that is absent yet not terminated is one `restartVat` has taken out of the kernel's reach while keeping its c-list, so that crank aborts instead of committing a release the returning incarnation would disagree with, and the returning incarnation is handed the action. Until it returns the action is reselected every crank; see [#1061](https://github.com/Consensys-Incorporated/ocap-kernel/issues/1061) + - It releases only where the endpoint is genuinely gone: a vat the store no longer calls active, whose cleanup tears the whole c-list down anyway, or a remote, which reconciles on its next incarnation. A vat the store still calls active but the kernel has no handle for is a disagreement rather than a departure, and the crank fails there instead of committing a release the vat's own tables would contradict - A failed garbage-collection delivery to a remote is logged and survived rather than escaping the crank and stopping the run loop ([#1022](https://github.com/Consensys-Incorporated/ocap-kernel/pull/1022)) - The next incarnation change reconciles a dropped or retired export but not a `retireImports`, whose entries `forgetEndpointImports` leaves in place - Two garbage-collection actions of one type for one endpoint no longer stop the run loop; one vat dropping two exports in a crank was enough ([#1022](https://github.com/Consensys-Incorporated/ocap-kernel/pull/1022)) - The krefs were hardened before being sorted, and `sort` writes back into the array even when it is already in order. Reproduces on `main`, so it predates this stack -- `getImporters` reports remote importers as well as vat ones, so retiring an object tells every endpoint holding it ([#1022](https://github.com/Consensys-Incorporated/ocap-kernel/pull/1022)) +- `getImporters` reports remote importers as well as vat ones, so retiring an object tells every endpoint holding it ([#1015](https://github.com/Consensys-Incorporated/ocap-kernel/issues/1015)) - The object is deleted once the importers have been told, so a remote left out kept a c-list entry naming a kref that no longer existed — which the reference count audit reports as dangling, killing the run loop - Tear down and mark for cleanup a vat whose worker launched but whose kernel-side registration then failed ([#1022](https://github.com/Consensys-Incorporated/ocap-kernel/pull/1022)) - `stopVat` forgets the vat whether or not its teardown succeeds ([#1022](https://github.com/Consensys-Incorporated/ocap-kernel/pull/1022)) - A throw from unpinning the root or from the handle's own `terminate` left the handle on the books while the store went on to wipe the vat's state, so `hasVat` and `getVatIds` reported a vat with no c-list and the next `bringOutYourDead` for it killed the run loop +- A vat's death is recorded in one synchronous step, so a worker that refuses to go cannot leave the record half-written ([#1023](https://github.com/Consensys-Incorporated/ocap-kernel/pull/1023)) + - Only `deleteVat` removes a vat's config, and terminated-vat cleanup does not call it, so the previous interleaving could leave a vat marked terminated whose config survived — which reads as _active_ again as soon as cleanup drops the mark, killing the run loop over the disagreement and resurrecting the vat on the next process start +- The run loop carries out a vat restart itself, as a queued request, so a vat is never out of the kernel's reach while cranks run; a crank that landed in that window read a live vat as a dead one ([#1023](https://github.com/Consensys-Incorporated/ocap-kernel/pull/1023)) + - Adds the `restartVat` run-queue item and `KernelQueue.enqueueRestartVat()`. `Kernel.restartVat` settles when the crank has done it, and rejects outright if the run loop is dead + - One request per vat is queued at a time. Two left the leftover one restarting a vat whose caller had already been handed a live handle, and terminating it if that relaunch failed +- A delivery whose endpoint has vanished is dropped only where the endpoint is gone for good — a terminated vat or a remote. `notify` and `bringOutYourDead` no longer take the run loop down when it has, and a `send` no longer reports a live endpoint as unreachable and discards a deliverable message ([#1023](https://github.com/Consensys-Incorporated/ocap-kernel/pull/1023)) + - Adds `VatManager.provideVat()`, which waits out a vat being torn down before answering, and makes the kernel's endpoint lookup asynchronous +- A restart that cannot relaunch its vat now terminates it and reports the failure to the caller, instead of killing the run loop — which rolled the crank back, undoing the termination records and returning the request to the queue, so every subsequent process start replayed the same failing restart ([#1023](https://github.com/Consensys-Incorporated/ocap-kernel/pull/1023)) + - Including a relaunch whose stream dies before its handle exists, which took the run loop down even so +- Terminating a vat with a restart still queued for it no longer kills the run loop when the crank reaches that request ([#1023](https://github.com/Consensys-Incorporated/ocap-kernel/pull/1023)) +- A vat's death is recorded while the store is held outside any crank, rather than after merely waiting for the crank in flight to end ([#1023](https://github.com/Consensys-Incorporated/ocap-kernel/pull/1023)) + - The run loop starts its next crank in the same turn it ends the last, so the wait resumed with that crank's savepoints already open: the death was written inside its delivery savepoint, for a rollback with nothing to do with the vat to undo — leaving the store calling a vat alive that the kernel had no handle for — and that crank could look the vat up before the record of its death existed and be handed a handle to a worker about to be killed +- A delivery that fails no longer resolves a result promise something has already settled ([#1023](https://github.com/Consensys-Incorporated/ocap-kernel/pull/1023)) + - The decider is set before the delivery, so a vat that loses its stream mid-delivery has the result rejected by its retirement and then again by the delivery's own catch. The second is a `Fail`, and it killed the run loop naming the promise rather than the dead worker +- A caller awaiting a queued vat restart is told when the run loop dies before carrying it out, instead of waiting forever ([#1023](https://github.com/Consensys-Incorporated/ocap-kernel/pull/1023)) + - The loop is checked alive when the request is queued and never again, and a restart has no kernel promise behind it the way a message result does. Adds `KernelQueue.onRunLoopDeath` +- A vat whose stream breaks is torn down even when recording its death fails, and a garbage-collection delivery reports krefs that cleanup reached first on every path rather than only some ([#1023](https://github.com/Consensys-Incorporated/ocap-kernel/pull/1023)) +- A vat whose stream breaks mid-crank stays dead even if that crank is then abandoned. The death was written inside whichever crank was open, so an abort rolled it back while the kernel had already dropped the handle; the store then called the vat live with nothing to deliver through, which killed the run loop at the next message addressed to it ([#1023](https://github.com/Consensys-Incorporated/ocap-kernel/pull/1023)) +- `restartVat` no longer reports a vat as missing while an earlier restart has it between workers. A second request now queues behind the first instead of rejecting with `VatNotFoundError` ([#1023](https://github.com/Consensys-Incorporated/ocap-kernel/pull/1023)) +- Work outliving a vat that has already been cleaned up — a `bringOutYourDead` scheduled before it died, say — is dropped rather than taken as a live vat the kernel has lost track of, which killed the run loop. Cleanup unmarks the vat it finishes, so "terminated" alone could not identify one ([#1023](https://github.com/Consensys-Incorporated/ocap-kernel/pull/1023)) +- `getImporters` also counts a terminated vat that cleanup has not reached, so retiring an object queues a `retireImport` for it rather than leaving its c-list entry naming nothing ([#1023](https://github.com/Consensys-Incorporated/ocap-kernel/pull/1023)) + - `deleteVat` drops the `vatConfig` row that enumerates a vat while its c-list survives until cleanup reaches it, one vat per crank. Terminating an object's owner and its importer close together, owner marked first, collects the orphan inside that window +- A vat reports its dropped imports on the `bringOutYourDead` that provoked the collection, rather than on some later one. The queues are now drained before the sweep, since a pending continuation still holds its closure's objects and a sweep run with work outstanding finds them reachable ([#1023](https://github.com/Consensys-Incorporated/ocap-kernel/pull/1023)) - Charge a delivered message's target reference against the run-queue item's own target rather than the routed target, so a message routed through a resolved promise no longer decrements an object nobody charged while leaking the promise ([#1020](https://github.com/MetaMask/ocap-kernel/pull/1020)) - Release a queued notification's reference before the paths that decide there is nothing to deliver, and stop decrementing references on promises retired alongside it that nobody had taken ([#1020](https://github.com/MetaMask/ocap-kernel/pull/1020)) - Transfer, rather than duplicate, the references a message carries when it is queued on an unresolved promise and later re-enqueued on resolution ([#1020](https://github.com/MetaMask/ocap-kernel/pull/1020)) diff --git a/packages/ocap-kernel/src/Kernel.test.ts b/packages/ocap-kernel/src/Kernel.test.ts index 6cf889206..47c591aef 100644 --- a/packages/ocap-kernel/src/Kernel.test.ts +++ b/packages/ocap-kernel/src/Kernel.test.ts @@ -31,13 +31,38 @@ const mocks = vi.hoisted(() => { #rejectRunLoop: ((error: Error) => void) | undefined; + #deliver: ((item: unknown) => Promise) | undefined; + // Like the real run loop, this settles only if the kernel dies. - run = vi.fn( - async () => - new Promise((_resolve, reject) => { - this.#rejectRunLoop = reject; - }), - ); + run = vi.fn(async (deliver: (item: unknown) => Promise) => { + this.#deliver = deliver; + return new Promise((_resolve, reject) => { + this.#rejectRunLoop = reject; + }); + }); + + // A restart is the run loop's work, so stand in for it reaching the request + // on its next crank. The failure is absorbed here rather than dropped: the + // real run loop would die of it, and the caller hears about it from the + // waiter `restartVat` registered, not from this call. + enqueueRestartVat = vi.fn((vatId: string) => { + this.#deliver?.({ type: 'restartVat', vatId }).catch(() => undefined); + }); + + readonly #pendingWorkWaiters = new Set<(error: Error) => void>(); + + onRunLoopDeath = vi.fn((reject: (error: Error) => void) => { + if (this.#runLoopFailure) { + reject( + new Error('Kernel run loop died', { + cause: this.#runLoopFailure, + }), + ); + return () => undefined; + } + this.#pendingWorkWaiters.add(reject); + return () => this.#pendingWorkWaiters.delete(reject); + }); /** * Fail the run loop, in the order the real `KernelQueue.run` does: the @@ -49,6 +74,11 @@ const mocks = vi.hoisted(() => { killRunLoop(error: Error): void { this.#runLoopFailure = error; this.#rejectRunLoop?.(error); + const abandoned = [...this.#pendingWorkWaiters]; + this.#pendingWorkWaiters.clear(); + for (const reject of abandoned) { + reject(new Error('Kernel run loop died', { cause: error })); + } } getRunLoopStatus = vi.fn(() => diff --git a/packages/ocap-kernel/src/Kernel.ts b/packages/ocap-kernel/src/Kernel.ts index a366eed95..312da6c38 100644 --- a/packages/ocap-kernel/src/Kernel.ts +++ b/packages/ocap-kernel/src/Kernel.ts @@ -111,12 +111,12 @@ export class Kernel { * @param options.onRunLoopFailure - Optional handler called if the run loop dies. * @param options.auditRefCounts - If true, verify every kref's reference * counts against the references the kernel actually holds at the end of each - * crank, and throw on any mismatch. This is the check standing in for the - * accounting invariant `collectGarbage` still cannot assert (see the comment - * on its `retireExport` branch), so it is not optional - * instrumentation: it is off by default only because it walks the whole store - * every crank. Any kernel whose accounting is under test wants it on, and - * every kernel `kernel-test` builds enables it. + * crank, and throw on any mismatch. Not optional instrumentation: it is what + * establishes that the accounting is right, and is off by default only because + * it walks the whole store every crank. Any kernel whose accounting is under + * test wants it on, and every kernel `kernel-test` builds enables it. Note + * that it checks counts against their holders, which is a different invariant + * from the one `collectGarbage`'s `retireExport` branch still cannot assert. */ // eslint-disable-next-line no-restricted-syntax private constructor( @@ -155,10 +155,13 @@ export class Kernel { // which would deadlock — this callback is invoked from within a crank. this.#kernelQueue = new KernelQueue( this.#kernelStore, - async (vatId, reason) => { - await this.#vatManager.stopVat(vatId, true, reason); - this.#kernelStore.markVatAsTerminated(vatId); - }, + // `stopVat` rather than `terminateVat`: this runs inside the crank that + // decided the vat has to go, and `terminateVat` would wait for that same + // crank to end. It needs no such wait — the run loop is right here — and + // `stopVat` puts the whole death on record before its first await, so a + // worker that refuses to die cannot leave the store half-told. + async (vatId, reason) => + await this.#vatManager.stopVat(vatId, true, reason), ); this.#vatManager = new VatManager({ @@ -230,6 +233,7 @@ export class Kernel { this.#kernelServiceManager.invokeKernelService.bind( this.#kernelServiceManager, ), + this.#vatManager.performVatRestart.bind(this.#vatManager), this.#logger, ); @@ -651,13 +655,19 @@ export class Kernel { /** * Gets an endpoint by its ID. * + * Asynchronous because a vat may be mid-teardown: `provideVat` waits that out + * rather than answering from a vat table the store has not caught up with, so + * by the time a caller is told the vat is gone the store says so too — which + * is what lets `#resolveEndpoint` tell a terminated vat from a missing one. A + * restart needs no such window, being carried out by the run loop itself. + * * @param endpointId - The ID of the endpoint to retrieve. - * @returns The endpoint handle for the given ID. + * @returns A promise for the endpoint handle for the given ID. * @throws If the endpoint ID is invalid (neither a vat ID nor a remote ID). */ - #getEndpoint(endpointId: EndpointId): EndpointHandle { + async #getEndpoint(endpointId: EndpointId): Promise { if (isVatId(endpointId)) { - return this.#vatManager.getVat(endpointId); + return await this.#vatManager.provideVat(endpointId); } if (isRemoteId(endpointId)) { return this.#remoteManager.getRemote(endpointId); diff --git a/packages/ocap-kernel/src/KernelQueue.test.ts b/packages/ocap-kernel/src/KernelQueue.test.ts index f66e79b05..e562d83b6 100644 --- a/packages/ocap-kernel/src/KernelQueue.test.ts +++ b/packages/ocap-kernel/src/KernelQueue.test.ts @@ -469,6 +469,41 @@ describe('KernelQueue', () => { expect(kernelQueue.subscriptions.size).toBe(0); }); + // A vat restart is the run loop's work, and has no kernel promise behind it + // the way a message result does, so this is the only thing that would ever + // tell its caller the request will not be carried out. + it('tells a caller awaiting queued work that the run loop died', async () => { + const told: Error[] = []; + kernelQueue.onRunLoopDeath((error) => told.push(error)); + const failure = new Error('crank exploded'); + + await killRunLoop(failure); + + expect(told).toHaveLength(1); + expect(told[0]?.message).toBe( + 'Kernel run loop died; this work will never be carried out', + ); + expect(told[0]?.cause).toBe(failure); + }); + + it('tells one registering after the fact, immediately', async () => { + await killRunLoop(new Error('crank exploded')); + const told: Error[] = []; + + kernelQueue.onRunLoopDeath((error) => told.push(error)); + + expect(told).toHaveLength(1); + }); + + it('does not tell one that unregistered first', async () => { + const told: Error[] = []; + kernelQueue.onRunLoopDeath((error) => told.push(error))(); + + await killRunLoop(new Error('crank exploded')); + + expect(told).toStrictEqual([]); + }); + it('rejects messages queued after the run loop dies', async () => { const failure = new Error('crank exploded'); await killRunLoop(failure); @@ -1310,6 +1345,27 @@ describe('KernelQueue', () => { }); }); + describe('enqueueRestartVat', () => { + it('enqueues the request for the run loop to carry out', () => { + kernelQueue.enqueueRestartVat('v1'); + + expect(kernelStore.enqueueRun).toHaveBeenCalledWith({ + type: 'restartVat', + vatId: 'v1', + }); + }); + + it('refuses once the run loop has died', async () => { + await killRunLoop(new Error('boom')); + + // The restart is the loop's work, so a dead loop will never do it and the + // caller would wait forever. + expect(() => kernelQueue.enqueueRestartVat('v1')).toThrow( + 'Kernel run loop died; cannot restart a vat', + ); + }); + }); + describe('waitForCrank', () => { it('handles when waitForCrank returns a delayed promise', async () => { let resolvePromise: ((value: void) => void) | undefined; diff --git a/packages/ocap-kernel/src/KernelQueue.ts b/packages/ocap-kernel/src/KernelQueue.ts index 8154ae015..074a1bdbe 100644 --- a/packages/ocap-kernel/src/KernelQueue.ts +++ b/packages/ocap-kernel/src/KernelQueue.ts @@ -19,6 +19,10 @@ import type { } from './types.ts'; import { Fail } from './utils/assert.ts'; +/** What a caller awaiting queued work is told when the run loop dies. */ +const DEAD_RUN_LOOP_WORK = + 'Kernel run loop died; this work will never be carried out'; + type RunLoopState = | Exclude | { state: 'failed'; error: Error }; @@ -51,6 +55,9 @@ export class KernelQueue { /** Promises resolved during this crank that have kernel subscriptions */ #resolvedWithKernelSubscription: KRef[] = []; + /** Callers awaiting work the run loop has been asked for but not yet done. */ + readonly #pendingWorkWaiters: Set<(error: Error) => void> = new Set(); + /** Thunk to signal run queue transition from empty to non-empty */ #wakeUpTheRunQueue: (() => void) | null; @@ -235,11 +242,30 @@ export class KernelQueue { } } + /** + * Tell a caller waiting on queued work if the run loop dies before carrying + * it out. `subscriptions` covers a message's result; a request with no kernel + * promise behind it — a vat restart — has nothing else that would settle it. + * + * @param reject - How to tell the caller. + * @returns A function that unregisters it, for the caller's own `finally`. + */ + onRunLoopDeath(reject: (error: Error) => void): () => void { + if (this.#runLoopState.state === 'failed') { + reject(this.#makeDeadRunLoopError(DEAD_RUN_LOOP_WORK)); + return () => undefined; + } + this.#pendingWorkWaiters.add(reject); + return () => { + this.#pendingWorkWaiters.delete(reject); + }; + } + /** * Record the death of the run loop and fail the kernel's own message-result - * subscriptions, which would otherwise hang forever. Kernel promises in the - * store stay unresolved, so vats awaiting a notify the dead loop owed them - * are not rescued by this. + * subscriptions and anyone waiting on queued work, which would otherwise hang + * forever. Kernel promises in the store stay unresolved, so vats awaiting a + * notify the dead loop owed them are not rescued by this. * * @param error - The error that killed the run loop. * @returns The failure, as an `Error` whatever was thrown. @@ -261,6 +287,12 @@ export class KernelQueue { ), ); } + + const abandoned = [...this.#pendingWorkWaiters]; + this.#pendingWorkWaiters.clear(); + for (const reject of abandoned) { + reject(this.#makeDeadRunLoopError(DEAD_RUN_LOOP_WORK)); + } return failure; } @@ -518,6 +550,23 @@ export class KernelQueue { } } + /** + * Enqueue a request to replace a vat's worker. + * + * The work itself belongs to the run loop, which is the point: a restart done + * where it is asked for takes the vat out of the kernel's reach while cranks + * continue, and a crank that lands in that window reads a live vat as a dead + * one. Queued, the restart happens in a crank of its own. + * + * @param vatId - The vat whose worker is to be replaced. + */ + enqueueRestartVat(vatId: VatId): void { + // The restart is the run loop's work now, so a dead loop will never do it, + // and a caller awaiting it would wait forever. + this.assertRunLoopAlive('restart a vat'); + this.#enqueueRun({ type: 'restartVat', vatId }); + } + /** * Enqueue a notification of promise resolution to an endpoint. * diff --git a/packages/ocap-kernel/src/KernelRouter.result-promise.test.ts b/packages/ocap-kernel/src/KernelRouter.result-promise.test.ts new file mode 100644 index 000000000..774b4e7eb --- /dev/null +++ b/packages/ocap-kernel/src/KernelRouter.result-promise.test.ts @@ -0,0 +1,132 @@ +import { makeSQLKernelDatabase } from '@metamask/kernel-store/sqlite/nodejs'; +import { describe, it, expect, vi } from 'vitest'; + +import { KernelQueue } from './KernelQueue.ts'; +import { KernelRouter } from './KernelRouter.ts'; +import { kser } from './liveslots/kernel-marshal.ts'; +import { makeKernelStore } from './store/index.ts'; +import type { EndpointHandle, KRef, RunQueueItem } from './types.ts'; + +/** + * What happens to a message's result promise when the delivery carrying it + * fails, against a real store and a real queue. + * + * `KernelRouter`'s own tests mock `resolvePromises`, so nothing there can see a + * second resolution of one promise — which is a `Fail`, and reaches the run + * loop from inside the very catch meant to contain the delivery's failure. + */ +describe('a result promise whose delivery fails', () => { + /** + * A router over a real store, delivering to one endpoint the test supplies. + * + * @param deliverMessage - What the endpoint does with a message. + * @returns The store, queue and a `runCrank` that delivers one queued item. + */ + async function makeFixture(deliverMessage: () => Promise): Promise<{ + kernelStore: ReturnType; + kernelQueue: KernelQueue; + runCrank: () => Promise; + }> { + const kdb = await makeSQLKernelDatabase({ dbFilename: ':memory:' }); + const kernelStore = makeKernelStore(kdb); + const kernelQueue = new KernelQueue(kernelStore, async () => undefined); + const endpoint = { + deliverMessage: vi.fn(deliverMessage), + deliverNotify: vi.fn(), + deliverDropExports: vi.fn(), + deliverRetireExports: vi.fn(), + deliverRetireImports: vi.fn(), + deliverBringOutYourDead: vi.fn(), + } as unknown as EndpointHandle; + const kernelRouter = new KernelRouter( + kernelStore, + kernelQueue, + async () => endpoint, + () => undefined, + async () => undefined, + ); + + const runCrank = async (): Promise => { + kernelStore.startCrank(); + kernelStore.createCrankSavepoint('crank'); + kernelStore.createCrankSavepoint('delivery'); + try { + const item = kernelStore.dequeueRun() as RunQueueItem; + await kernelRouter.deliver(item); + kernelStore.collectGarbage(); + } finally { + kernelStore.endCrank(); + } + }; + + return { kernelStore, kernelQueue, runCrank }; + } + + /** + * Queue a message to v1's root with a result promise. + * + * @param kernelStore - The store to set up. + * @param kernelQueue - The queue to send through. + * @returns The result promise's kref. + */ + function queueSendWithResult( + kernelStore: ReturnType, + kernelQueue: KernelQueue, + ): KRef { + kernelStore.setVatConfig('v1', { bundleName: 'vat1' }); + kernelStore.initEndpoint('v1'); + const target = kernelStore.exportFromEndpoint('v1', 'o+1'); + const [result] = kernelStore.initKernelPromise(); + kernelQueue.enqueueSend(target, { + methargs: kser(['ping', []]), + result, + }); + return result; + } + + it('rejects it once the delivery reports failure', async () => { + const { kernelStore, kernelQueue, runCrank } = await makeFixture( + async () => { + throw new Error('stream closed'); + }, + ); + const result = queueSendWithResult(kernelStore, kernelQueue); + + await runCrank(); + + expect(kernelStore.getKernelPromise(result).state).toBe('rejected'); + }); + + // A vat that resolves the result and then loses its stream is one way in; a + // stream that dies mid-delivery is the other, since retiring the vat rejects + // every promise it was deciding and this one's decider was just set. + it.each([ + { how: 'fulfilled', rejected: false, state: 'fulfilled' }, + { how: 'rejected', rejected: true, state: 'rejected' }, + ])( + 'leaves a result the endpoint already $how alone', + async ({ rejected, state }) => { + // Held in an object so the endpoint can reach it before the queue that + // settles the promise exists. + const endpointDoes = { settleTheResult: (): void => undefined }; + const { kernelStore, kernelQueue, runCrank } = await makeFixture( + async () => { + endpointDoes.settleTheResult(); + throw new Error('stream closed'); + }, + ); + const result = queueSendWithResult(kernelStore, kernelQueue); + endpointDoes.settleTheResult = () => + kernelQueue.resolvePromises('v1', [ + [result, rejected, kser('settled by the endpoint')], + ]); + + expect(await runCrank()).toBeUndefined(); + + const settled = kernelStore.getKernelPromise(result); + expect(settled.state).toBe(state); + // The endpoint's own settlement, not the delivery's `DELIVERY_FAILED`. + expect(settled.value?.body).toContain('settled by the endpoint'); + }, + ); +}); diff --git a/packages/ocap-kernel/src/KernelRouter.test.ts b/packages/ocap-kernel/src/KernelRouter.test.ts index d7dfd935f..8c61e4cbf 100644 --- a/packages/ocap-kernel/src/KernelRouter.test.ts +++ b/packages/ocap-kernel/src/KernelRouter.test.ts @@ -13,6 +13,7 @@ import type { RunQueueItemGCAction, RunQueueItemBringOutYourDead, EndpointId, + VatId, GCRunQueueType, CrankResult, EndpointHandle, @@ -22,8 +23,11 @@ describe('KernelRouter', () => { // Mock dependencies let kernelStore: KernelStore; let kernelQueue: KernelQueue; - let getEndpoint: (endpointId: EndpointId) => EndpointHandle; + let getEndpoint: ( + endpointId: EndpointId, + ) => EndpointHandle | Promise; let endpointHandle: EndpointHandle; + let restartVat: MockInstance<(vatId: VatId) => Promise>; let kernelRouter: KernelRouter; beforeEach(() => { @@ -46,7 +50,9 @@ describe('KernelRouter', () => { kernelStore = { getOwner: vi.fn(), isRevoked: vi.fn(), - getKernelPromise: vi.fn(), + // Unresolved by default: the delivery paths ask before settling a + // message's result, and a bare `vi.fn()` answers `undefined`. + getKernelPromise: vi.fn().mockReturnValue({ state: 'unresolved' }), decrementRefCount: vi.fn(), setPromiseDecider: vi.fn(), translateRefKtoE: vi.fn( @@ -69,6 +75,7 @@ describe('KernelRouter', () => { orphanKernelObject: vi.fn(), hasCListEntry: vi.fn().mockReturnValue(true), isVatTerminated: vi.fn().mockReturnValue(false), + isVatActive: vi.fn().mockReturnValue(true), createCrankSavepoint: vi.fn(), } as unknown as KernelStore; @@ -78,6 +85,7 @@ describe('KernelRouter', () => { } as unknown as KernelQueue; const mockInvokeKernelService = vi.fn(); + restartVat = vi.fn().mockResolvedValue(undefined); // Create the router to test kernelRouter = new KernelRouter( @@ -85,6 +93,7 @@ describe('KernelRouter', () => { kernelQueue, getEndpoint, mockInvokeKernelService, + restartVat, ); }); @@ -352,6 +361,48 @@ describe('KernelRouter', () => { ).toStrictEqual([[target, 'deliver|send|target']]); }); + // The same distinction, on the path that discovers the endpoint is gone + // only after routing has already succeeded. Every other test of this + // branch aims at a plain object, where the item's target and the routed + // target are the same kref and the two spellings are indistinguishable. + it('charges the promise, not the object it resolved to, when the endpoint is gone', async () => { + const promiseId = 'kp123'; + const resolvedObject = 'ko456'; + ( + kernelStore.getKernelPromise as unknown as MockInstance + ).mockReturnValueOnce({ + state: 'fulfilled', + value: { body: '#"$0"', slots: [resolvedObject] }, + }); + (kernelStore.getOwner as unknown as MockInstance).mockReturnValue('v1'); + ( + kernelStore.isVatTerminated as unknown as MockInstance + ).mockReturnValue(true); + (getEndpoint as unknown as MockInstance).mockRejectedValueOnce( + new Error('vat v1 not found'), + ); + + await kernelRouter.deliver({ + type: 'send', + target: promiseId, + message: { + methargs: { body: 'method args', slots: [] }, + result: null, + }, + }); + + expect(kernelStore.decrementRefCount).toHaveBeenCalledWith( + promiseId, + 'deliver|splat|target', + ); + // Charging this instead leaks the promise and collects an object that + // nobody released. + expect(kernelStore.decrementRefCount).not.toHaveBeenCalledWith( + resolvedObject, + 'deliver|splat|target', + ); + }); + it('splats message when promise resolves to a non-object', async () => { // Setup a fulfilled promise that doesn't resolve to an object const promiseId = 'kp123'; @@ -443,13 +494,41 @@ describe('KernelRouter', () => { ); }); + it('propagates a lookup failure for a vat that is absent but not terminated', async () => { + // Not a splat: reporting a live endpoint as unreachable would discard a + // deliverable message and reject its result for no reason. + (kernelStore.getOwner as unknown as MockInstance).mockReturnValueOnce( + 'v1', + ); + (getEndpoint as unknown as MockInstance).mockImplementationOnce(() => { + throw new Error('vat v1 not found'); + }); + + await expect( + kernelRouter.deliver({ + type: 'send', + target: 'ko123', + message: { + methargs: { body: 'method args', slots: [] }, + result: 'kp1', + } as unknown as SwingsetMessage, + }), + ).rejects.toThrow('vat v1 not found'); + + expect(kernelQueue.resolvePromises).not.toHaveBeenCalled(); + }); + it('splats message with ENDPOINT_UNREACHABLE when endpoint vanishes', async () => { const endpointId = 'v1'; const target = 'ko123'; (kernelStore.getOwner as unknown as MockInstance).mockReturnValueOnce( endpointId, ); - // getEndpoint throws (endpoint gone) + // The endpoint is gone for good, which is what makes it a splat rather + // than an error worth propagating. + ( + kernelStore.isVatTerminated as unknown as MockInstance + ).mockReturnValue(true); (getEndpoint as unknown as MockInstance).mockImplementationOnce(() => { throw new Error('vat not found'); }); @@ -522,6 +601,39 @@ describe('KernelRouter', () => { }); describe('notify', () => { + it('drops a notify whose endpoint is gone for good', async () => { + // Reachable while a vat is being torn down: `provideVat` waits for the + // teardown, then reports the vat gone. Without this the rejection escapes + // the crank and kills the run loop. + ( + kernelStore.getKernelPromise as unknown as MockInstance + ).mockReturnValueOnce({ + state: 'fulfilled', + value: { body: JSON.stringify({ value: 'v' }), slots: [] }, + }); + (kernelStore.krefToEref as unknown as MockInstance).mockReturnValueOnce( + 'p+123', + ); + ( + kernelStore.isVatTerminated as unknown as MockInstance + ).mockReturnValue(true); + (getEndpoint as unknown as MockInstance).mockRejectedValueOnce( + new Error('vat v1 not found'), + ); + + const result = await kernelRouter.deliver({ + type: 'notify', + endpointId: 'v1', + kpid: 'kp123', + }); + + expect(result).toStrictEqual({ didDelivery: 'v1' }); + expect(endpointHandle.deliverNotify).not.toHaveBeenCalled(); + // Resolved before the translation, which would otherwise mint c-list + // entries for an endpoint that cannot be told about them. + expect(kernelStore.translateRefKtoE).not.toHaveBeenCalled(); + }); + it('delivers a notify to a vat and returns crank results', async () => { const endpointId = 'v1'; const kpid = 'kp123'; @@ -814,6 +926,38 @@ describe('KernelRouter', () => { expect(kernelStore.orphanKernelObject).not.toHaveBeenCalled(); }); + it('waits for a vat that is coming back, then delivers to it', async () => { + // The restart window: `provideVat` answers once the new incarnation is + // up, so the crank waits instead of resolving a live vat as a dead one. + let finishRestart!: (handle: EndpointHandle) => void; + (getEndpoint as unknown as MockInstance).mockReturnValueOnce( + new Promise((resolve) => { + finishRestart = resolve; + }), + ); + + const delivered = kernelRouter.deliver({ + type: 'retireImports', + endpointId: 'v1', + krefs: ['ko1'], + }); + + // Nothing is released ahead of knowing where the action is going. + expect(kernelStore.deleteCListEntry).not.toHaveBeenCalled(); + + finishRestart(endpointHandle); + await delivered; + + expect(endpointHandle.deliverRetireImports).toHaveBeenCalledWith([ + 'translated-ko1', + ]); + expect(kernelStore.deleteCListEntry).toHaveBeenCalledWith( + 'v1', + 'ko1', + 'translated-ko1', + ); + }); + it('still releases the kernel side when a terminated vat has vanished', async () => { getEndpoint.mockImplementationOnce(() => { throw Error('vat v1 not found'); @@ -866,13 +1010,14 @@ describe('KernelRouter', () => { throw Error('vat v1 not found'); }); - const result = await kernelRouter.deliver({ - type: actionType, - endpointId: 'v1', - krefs: ['ko1'], - }); + await expect( + kernelRouter.deliver({ + type: actionType, + endpointId: 'v1', + krefs: ['ko1'], + }), + ).rejects.toThrow('vat v1 not found'); - expect(result).toStrictEqual({ abort: true }); expect(kernelStore.clearReachableFlag).not.toHaveBeenCalled(); expect(kernelStore.deleteCListEntry).not.toHaveBeenCalled(); expect(kernelStore.orphanKernelObject).not.toHaveBeenCalled(); @@ -949,6 +1094,49 @@ describe('KernelRouter', () => { }); describe('bringOutYourDead', () => { + it('skips a reap whose endpoint is gone for good', async () => { + ( + kernelStore.isVatTerminated as unknown as MockInstance + ).mockReturnValue(true); + (getEndpoint as unknown as MockInstance).mockRejectedValueOnce( + new Error('vat v1 not found'), + ); + + const result = await kernelRouter.deliver({ + type: 'bringOutYourDead', + endpointId: 'v1', + }); + + // A reap only asks an endpoint to tidy up, so one that is gone has + // nothing left to ask — and nothing was delivered. + expect(result).toBeUndefined(); + expect(endpointHandle.deliverBringOutYourDead).not.toHaveBeenCalled(); + }); + + // Nothing purges the reap queue when a vat dies, and cleanup ends by + // *unmarking* the vat it finished — so a reap scheduled before the vat + // died arrives at an endpoint that is neither present nor terminated. + // Read as a disagreement, that throw kills the run loop. + it('skips a reap for a vat that has already been cleaned up', async () => { + ( + kernelStore.isVatTerminated as unknown as MockInstance + ).mockReturnValue(false); + (kernelStore.isVatActive as unknown as MockInstance).mockReturnValue( + false, + ); + (getEndpoint as unknown as MockInstance).mockRejectedValueOnce( + new Error('vat v1 not found'), + ); + + const result = await kernelRouter.deliver({ + type: 'bringOutYourDead', + endpointId: 'v1', + }); + + expect(result).toBeUndefined(); + expect(endpointHandle.deliverBringOutYourDead).not.toHaveBeenCalled(); + }); + it('delivers bringOutYourDead to a vat and returns crank results', async () => { const endpointId = 'v1'; const bringOutYourDeadItem: RunQueueItemBringOutYourDead = { @@ -972,6 +1160,30 @@ describe('KernelRouter', () => { }); }); + describe('restartVat', () => { + it('carries out a queued restart and reports no delivery', async () => { + // Not a delivery: nothing was handed to the vat, and the incarnation that + // comes back has taken none yet. + const result = await kernelRouter.deliver({ + type: 'restartVat', + vatId: 'v1', + }); + + expect(restartVat).toHaveBeenCalledWith('v1'); + expect(result).toBeUndefined(); + }); + + it('lets a failed restart take the crank down', async () => { + // Aborting would undo the terminated mark that makes the half-restarted + // vat's c-list reclaimable. + restartVat.mockRejectedValueOnce(new Error('worker died')); + + await expect( + kernelRouter.deliver({ type: 'restartVat', vatId: 'v1' }), + ).rejects.toThrow('worker died'); + }); + }); + it('throws on unknown run queue item type', async () => { // @ts-expect-error - deliberately using an invalid type const invalidItem: RunQueueItem = { type: 'invalid' }; diff --git a/packages/ocap-kernel/src/KernelRouter.ts b/packages/ocap-kernel/src/KernelRouter.ts index 9d4c393e2..da51bbb25 100644 --- a/packages/ocap-kernel/src/KernelRouter.ts +++ b/packages/ocap-kernel/src/KernelRouter.ts @@ -12,6 +12,7 @@ import { extractSingleRef } from './store/utils/extract-ref.ts'; import { parseRef } from './store/utils/parse-ref.ts'; import { isPromiseRef } from './store/utils/promise-ref.ts'; import type { + VatId, EndpointId, EndpointHandle, ERef, @@ -22,6 +23,7 @@ import type { RunQueueItemBringOutYourDead, RunQueueItemNotify, RunQueueItemGCAction, + RunQueueItemRestartVat, CrankResult, } from './types.ts'; import { isVatId } from './types.ts'; @@ -46,11 +48,17 @@ export class KernelRouter { readonly #kernelQueue: KernelQueue; /** A function that returns an endpoint handle for a given endpoint id. */ - readonly #getEndpoint: (endpointId: EndpointId) => EndpointHandle; + readonly #getEndpoint: (endpointId: EndpointId) => Promise; /** A function that invokes a method on a kernel service. */ readonly #invokeKernelService: (target: KRef, message: KernelMessage) => void; + /** + * A function that replaces a vat's worker, for the crank that carries out a + * queued restart request. + */ + readonly #restartVat: (vatId: VatId) => Promise; + /** The logger, if any. */ readonly #logger: Logger | undefined; @@ -61,19 +69,22 @@ export class KernelRouter { * @param kernelQueue - The kernel's queue. * @param getEndpoint - A function that returns an endpoint handle for a given endpoint id. * @param invokeKernelService - A function that calls a method on a kernel service object. + * @param restartVat - A function that replaces a vat's worker. * @param logger - The logger. If not provided, no logging will be done. */ constructor( kernelStore: KernelStore, kernelQueue: KernelQueue, - getEndpoint: (endpointId: EndpointId) => EndpointHandle, + getEndpoint: (endpointId: EndpointId) => Promise, invokeKernelService: (target: KRef, message: KernelMessage) => void, + restartVat: (vatId: VatId) => Promise, logger?: Logger, ) { this.#kernelStore = kernelStore; this.#kernelQueue = kernelQueue; this.#getEndpoint = getEndpoint; this.#invokeKernelService = invokeKernelService; + this.#restartVat = restartVat; this.#logger = logger; } @@ -107,6 +118,8 @@ export class KernelRouter { return await this.#deliverGCAction(item); case 'bringOutYourDead': return await this.#deliverBringOutYourDead(item); + case 'restartVat': + return await this.#restartVatWorker(item); default: // @ts-expect-error Runtime does not respect "never". Fail`unsupported or unknown run queue item type ${item.type}`; @@ -198,6 +211,33 @@ export class KernelRouter { } } + /** + * Reject a message's result promise, unless something has already settled it. + * + * The delivery that failed may have settled it on its way down: a vat that + * resolves the result and then loses its stream, or one whose stream dies + * mid-delivery and is retired, which rejects every promise it was deciding — + * this one included, its decider having been set just before the delivery. + * Resolving a settled promise is a `Fail`, thrown from inside the very catch + * that is handling the delivery's failure, so it killed the run loop naming + * the promise rather than the dead worker. + * + * @param endpointId - The endpoint that was to have decided it. + * @param kpid - The result promise. + * @param failure - Why the message could not be delivered. + */ + #rejectResultIfPending( + endpointId: EndpointId, + kpid: KRef, + failure: CapData, + ): void { + const { state } = this.#kernelStore.getKernelPromise(kpid); + if (state !== 'unresolved') { + return; + } + this.#kernelQueue.resolvePromises(endpointId, [[kpid, true, failure]]); + } + /** * Deliver a 'send' run queue item. * @@ -235,14 +275,15 @@ export class KernelRouter { const isKernelServiceMessage = endpointId === 'kernel'; let endpoint: EndpointHandle | null = null; if (!isKernelServiceMessage) { - try { - endpoint = this.#getEndpoint(endpointId); - } catch { - // TODO: Narrow this catch to the expected error type (e.g., - // VatNotFoundError) so that unexpected errors are not silently - // swallowed and deliverable messages are not incorrectly discarded. - // Endpoint vanished (e.g., vat terminated but ownership entries not - // yet cleaned up). Treat the same as a splat. + // An endpoint that is gone for good — a terminated vat whose ownership + // entries are not cleaned up yet, or a disconnected remote — has nothing + // to deliver to, so the message goes splat. Anything else `resolveEndpoint` + // propagates, rather than reporting a live endpoint as unreachable and + // discarding a deliverable message. + endpoint = + (await this.#resolveEndpoint(endpointId, `send of ${target}`)) ?? + null; + if (!endpoint) { if (message.result) { const promise = this.#kernelStore.getKernelPromise(message.result); this.#kernelQueue.resolvePromises(promise.decider, [ @@ -307,13 +348,11 @@ export class KernelRouter { if (message.result) { const detail = error instanceof Error ? error.message : String(error); - this.#kernelQueue.resolvePromises(endpointId, [ - [ - message.result, - true, - makeKernelError('DELIVERY_FAILED', detail), - ], - ]); + this.#rejectResultIfPending( + eid, + message.result, + makeKernelError('DELIVERY_FAILED', detail), + ); } // Continue processing other messages - don't let one failure crash the queue } @@ -389,6 +428,15 @@ export class KernelRouter { // no c-list entry, already done return { didDelivery: endpointId }; } + // Ahead of the translation below, which would otherwise mint c-list entries + // for an endpoint with no way to hear about them. + const endpoint = await this.#resolveEndpoint( + endpointId, + `notify of ${kpid}`, + ); + if (!endpoint) { + return { didDelivery: endpointId }; + } const targets = this.#kernelStore.getKpidsToRetire(kpid, value); if (targets.length === 0) { // no kpids to retire, already done @@ -415,10 +463,53 @@ export class KernelRouter { // exported ocap URLs by scanning these entries. The cost of keeping them is // that a settled promise reached this way holds a count forever, so it is // never collected and its resolution slots are never released. - const endpoint = this.#getEndpoint(endpointId); return await endpoint.deliverNotify(resolutions); } + /** + * The handle for an endpoint, or `undefined` if the endpoint is gone for good + * and the work addressed to it can be dropped. + * + * Gone for good means a vat the store has no live record of — marked + * terminated, and so awaiting a cleanup that takes its whole c-list with it, + * or already cleaned up — or a remote, which reconciles on its next + * incarnation. Both halves are needed: cleanup ends with `forgetTerminatedVat`, + * so a vat that is long gone is no longer *marked* terminated either, and work + * outliving it (a `bringOutYourDead` scheduled before it died, say) would + * otherwise be read as a disagreement. + * + * A vat the store still calls active but the kernel has no handle for is that + * disagreement: `restartVat` is carried out by the run loop and `terminateVat` + * records the vat as in flux, so neither leaves a vat in that state, and the + * caller is better served by the error than by an answer that says "gone" + * about a vat that isn't. + * + * @param endpointId - The endpoint to resolve. + * @param what - What was being delivered, for the log. + * @returns The endpoint handle, or undefined if it will not be back. + */ + async #resolveEndpoint( + endpointId: EndpointId, + what: string, + ): Promise { + try { + return await this.#getEndpoint(endpointId); + } catch (error) { + if ( + isVatId(endpointId) && + this.#kernelStore.isVatActive(endpointId) && + !this.#kernelStore.isVatTerminated(endpointId) + ) { + throw error; + } + this.#logger?.error( + `Endpoint ${endpointId} vanished before ${what}:`, + error, + ); + return undefined; + } + } + /** * Deliver a Garbage Collection action run queue item. * @@ -436,57 +527,50 @@ export class KernelRouter { // survives still has to be released on the kernel's side: the action has // already been consumed from the durable set, so skipping the teardown // would lose it and leave the entry behind for good. - const live = krefs.filter((kref) => - this.#kernelStore.hasCListEntry(endpointId, kref), - ); - if (live.length < krefs.length) { - this.#logger?.error( - `${type} for ${endpointId}: ${krefs.length - live.length} of ${krefs.length} kref(s) were cleaned up before delivery`, - ); - } - if (live.length === 0) { + const stillHeld = (): KRef[] => + krefs.filter((kref) => this.#kernelStore.hasCListEntry(endpointId, kref)); + const reportCleanedUp = (held: KRef[]): void => { + if (held.length < krefs.length) { + this.#logger?.error( + `${type} for ${endpointId}: ${krefs.length - held.length} of ${krefs.length} kref(s) were cleaned up before delivery`, + ); + } + }; + const heldOnArrival = stillHeld(); + if (heldOnArrival.length === 0) { + // Reported here as well as below: this is the whole of the delivery when + // cleanup reached every kref first. + reportCleanedUp(heldOnArrival); return { didDelivery: endpointId }; } // Resolved before anything is torn down, so a lookup that fails has nothing // to undo, and so the two outcomes below are decided rather than discovered - // halfway through. - let endpoint: EndpointHandle | undefined; - try { - endpoint = this.#getEndpoint(endpointId); - } catch (error) { - // A vat absent from the kernel's vat table but not marked terminated is a - // vat between incarnations, and its c-list is whole: every kref here is one - // the returning incarnation still has in its own tables. `restartVat` - // takes a vat out of that table for as long as launching a worker and - // negotiating with it takes, so this is reachable, and releasing the - // kernel's side would commit exactly the disagreement the failed delivery - // below rolls back to avoid — the vat would mint fresh krefs for objects - // the kernel thinks it let go of. - // - // Abort rather than throw. A throw here leaves `deliver` by the run loop's - // catch and kills the run loop for good over a vat that is about to come - // back. The rollback restores the c-list and the action, so the returning - // incarnation is handed it instead. Until it returns the action is - // reselected every crank, since GC actions are chosen ahead of all other - // work — a spin, where this used to be a death. See #1061. - if ( - isVatId(endpointId) && - !this.#kernelStore.isVatTerminated(endpointId) - ) { - this.#logger?.error( - `Endpoint ${endpointId} is between incarnations; deferring ${type} of ${JSON.stringify(live)}:`, - error, - ); - return { abort: true }; - } - // A terminated vat's cleanup tears its c-list down wholesale, and a remote - // reconciles on its next incarnation, so for those the release below is - // safe to commit — and has to be, since the action is already spent from - // the durable set. - this.#logger?.error( - `Endpoint ${endpointId} vanished before ${type} of ${JSON.stringify(live)}; releasing the kernel's side anyway:`, - error, - ); + // halfway through. An endpoint that is gone for good still gets the release: + // the action is already spent from the durable set, and for a terminated vat + // cleanup would take the entries anyway. + // + // The throw `#resolveEndpoint` reserves for a vat the store still calls + // active and has not marked terminated is, here, the least bad of three. Committing the release + // corrupts silently — the vat's own tables still name every one of these + // krefs, which is the disagreement the failed delivery below rolls back to + // avoid. Aborting spins: it does keep the action, since `rollbackCrank` + // restores the cached GC set, but nothing about the vat changes between + // cranks, so the same action is re-selected and re-aborted with no delivery + // to wait on — a run loop that is dead without saying so. + const endpoint = await this.#resolveEndpoint( + endpointId, + `${type}; releasing the kernel's side anyway`, + ); + // Re-read after the await, not before it: resolving an endpoint yields to + // other work, and a remote's incarnation change tears its c-list down + // without waiting for the crank. Reusing the earlier answer would hand + // `krefsToErefs` a kref whose entry has since gone, and it throws rather + // than returning short — killing the run loop over an entry that is + // already, correctly, released. + const live = stillHeld(); + reportCleanedUp(live); + if (live.length === 0) { + return { didDelivery: endpointId }; } const erefs = this.#kernelStore.krefsToErefs(endpointId, live); // Telling an endpoint to let go is also the kernel letting go. Otherwise a @@ -573,8 +657,37 @@ export class KernelRouter { ): Promise { const { endpointId } = item; this.#logger?.log(`@@@@ deliver ${endpointId} bringOutYourDead`); - const endpoint = this.#getEndpoint(endpointId); - const crankResult = await endpoint.deliverBringOutYourDead(); - return crankResult; + const endpoint = await this.#resolveEndpoint( + endpointId, + 'bringOutYourDead', + ); + if (!endpoint) { + // A reap only asks an endpoint to tidy up, so one that is gone has nothing + // left to ask. No `didDelivery`, since nothing was delivered. + return undefined; + } + return await endpoint.deliverBringOutYourDead(); + } + + /** + * Carry out a queued request to replace a vat's worker. + * + * Not a delivery, so no `didDelivery`: nothing was handed to the vat, and the + * incarnation that comes back has taken no deliveries yet. + * + * `performVatRestart` reports a failed restart by terminating the vat rather + * than by throwing, so this commits either way. Neither ending a crank is open + * to it: aborting and throwing both roll the crank back, which would undo the + * termination records *and* put this request back on the run queue, leaving + * the same failing restart to be replayed for the life of the store. + * + * @param item - The restart request. + * @returns Nothing; the crank has no outcome to report. + */ + async #restartVatWorker( + item: RunQueueItemRestartVat, + ): Promise { + await this.#restartVat(item.vatId); + return undefined; } } diff --git a/packages/ocap-kernel/src/garbage-collection/garbage-collection.ts b/packages/ocap-kernel/src/garbage-collection/garbage-collection.ts index 393c5cf08..518decaf7 100644 --- a/packages/ocap-kernel/src/garbage-collection/garbage-collection.ts +++ b/packages/ocap-kernel/src/garbage-collection/garbage-collection.ts @@ -106,10 +106,10 @@ function filterActionsForProcessing( actionSetUpdated = true; } - // Sorted before hardening, not by the caller afterwards: `harden` freezes the - // array, and `sort` writes back into it even when it is already in order, so - // any endpoint with two actions of one type threw out of here and killed the - // run loop. One vat dropping two exports in a crank is enough. + // Sorted before hardening rather than by `processGCActionSet` afterwards: + // `harden` freezes the array, and `sort` writes back into it even when it is + // already in order, so any endpoint with two actions of one type threw there + // and killed the run loop. One vat dropping two exports in a crank is enough. krefs.sort(); return harden({ krefs, actionSetUpdated }); } diff --git a/packages/ocap-kernel/src/garbage-collection/gc-delivery.test.ts b/packages/ocap-kernel/src/garbage-collection/gc-delivery.test.ts index dbb9eceb0..d23df6d2b 100644 --- a/packages/ocap-kernel/src/garbage-collection/gc-delivery.test.ts +++ b/packages/ocap-kernel/src/garbage-collection/gc-delivery.test.ts @@ -1,4 +1,5 @@ import { makeSQLKernelDatabase } from '@metamask/kernel-store/sqlite/nodejs'; +import type { Logger } from '@metamask/logger'; import { describe, it, expect, vi } from 'vitest'; import { processGCActionSet } from './garbage-collection.ts'; @@ -19,12 +20,13 @@ import type { * Selection has its own real-store coverage next door and delivery has unit * coverage over a mocked store, but the two halves have never met: the mock * answers `hasCListEntry` and `krefsToErefs` from `vi.fn()`s, so nothing there - * pins what the kernel actually releases, and nothing pins that a crank the - * delivery aborts gives the action back. + * pins what the kernel actually releases. * * The crank is driven here rather than by `KernelQueue.run`, which never - * resolves and, for the abort below, would reselect the same action every crank - * (see #1061). `runCrank` is the shape `#runLoop` gives a delivery. + * resolves. `runCrank` mirrors the shape `#runLoop` gives a delivery, rollback + * on a throw included — without that the action `processGCActionSet` has + * already spent is committed away rather than restored, and the last test + * below would be asserting against a store the run loop would never produce. */ type Delivered = { method: string; erefs: ERef[] }; @@ -43,16 +45,28 @@ async function makeFixture(): Promise<{ ) => Promise; delivered: Delivered[]; endpoints: Map; + duringEndpointLookup: { run: () => void }; + logged: unknown[][]; }> { const kdb = await makeSQLKernelDatabase({ dbFilename: ':memory:' }); const kernelStore = makeKernelStore(kdb); const kernelQueue = new KernelQueue(kernelStore, async () => undefined); const endpoints = new Map(); const delivered: Delivered[] = []; + // Stands in for whatever else runs while the lookup is awaited. + const duringEndpointLookup = { run: (): void => undefined }; + const logged: unknown[][] = []; + const logger = { + log: vi.fn(), + error: vi.fn((...args: unknown[]) => { + logged.push(args); + }), + } as unknown as Logger; const kernelRouter = new KernelRouter( kernelStore, kernelQueue, - (endpointId) => { + async (endpointId) => { + duringEndpointLookup.run(); const endpoint = endpoints.get(endpointId); if (!endpoint) { throw new Error(`vat ${endpointId} not found`); @@ -60,6 +74,8 @@ async function makeFixture(): Promise<{ return endpoint; }, () => undefined, + async () => undefined, + logger, ); const runCrank = async ( @@ -74,9 +90,12 @@ async function makeFixture(): Promise<{ return undefined; } beforeDeliver?.(item); - const result = await kernelRouter.deliver(item); - if (result?.abort) { + let result: CrankResult | undefined; + try { + result = await kernelRouter.deliver(item); + } catch (error) { kernelStore.rollbackCrank('delivery'); + throw error; } kernelStore.collectGarbage(); return result; @@ -85,7 +104,14 @@ async function makeFixture(): Promise<{ } }; - return { kernelStore, runCrank, delivered, endpoints }; + return { + kernelStore, + runCrank, + delivered, + endpoints, + duringEndpointLookup, + logged, + }; } /** @@ -206,22 +232,87 @@ describe('a GC action the kernel issues', () => { ]); }); - // A vat between incarnations still holds these krefs. Releasing the kernel's - // side would leave the two disagreeing; throwing killed the run loop. - it('gives the action back when the vat is between incarnations', async () => { + // A vat absent from the kernel's tables but not marked terminated still holds + // these krefs as far as the c-list is concerned, so the kernel must not + // release its side. `provideVat` waits out a vat that is coming back, so one + // that reaches here is gone with nothing to wait on. + it('releases nothing for a vat that is absent but not terminated', async () => { const { kernelStore, runCrank } = await makeFixture(); + // The config row is what makes the store call the vat active, which is the + // half of "absent but not terminated" that distinguishes it from a vat + // cleanup has already finished with. + kernelStore.setVatConfig('v1', { bundleName: 'vat1' }); const kref = kernelStore.initKernelObject('v1'); kernelStore.addCListEntry('v1', kref, 'o+1'); kernelStore.setObjectRefCount(kref, { reachable: 0, recognizable: 1 }); kernelStore.addGCActions([`v1 dropExport ${kref}`]); - const result = await runCrank(); + await expect(runCrank()).rejects.toThrow('vat v1 not found'); - expect(result).toStrictEqual({ abort: true }); expect(kernelStore.getReachableFlag('v1', kref)).toBe(true); expect(kernelStore.hasCListEntry('v1', kref)).toBe(true); + // Selection spends the action from the durable set before delivery, so the + // rollback is the only thing that gives it back. expect([...kernelStore.getGCActions()]).toStrictEqual([ `v1 dropExport ${kref}`, ]); }); + + // `nextTerminatedVatCleanup` and a remote's incarnation change both tear + // c-list entries down without waiting for the crank, and resolving the + // endpoint yields to them. Reusing the answer from before that yield hands + // `krefsToErefs` a kref whose entry has gone, and it throws rather than + // returning short. + it('re-reads the c-list after resolving the endpoint', async () => { + const { + kernelStore, + runCrank, + delivered, + endpoints, + duringEndpointLookup, + } = await makeFixture(); + registerEndpoint(endpoints, delivered, 'v1'); + const gone = kernelStore.initKernelObject('v1'); + const kept = kernelStore.initKernelObject('v1'); + kernelStore.addCListEntry('v1', gone, 'o+1'); + kernelStore.addCListEntry('v1', kept, 'o+2'); + kernelStore.setObjectRefCount(gone, { reachable: 0, recognizable: 1 }); + kernelStore.setObjectRefCount(kept, { reachable: 0, recognizable: 1 }); + kernelStore.addGCActions([ + `v1 dropExport ${gone}`, + `v1 dropExport ${kept}`, + ]); + duringEndpointLookup.run = () => { + kernelStore.deleteCListEntry('v1', gone, 'o+1'); + }; + + await runCrank(); + + expect(delivered).toStrictEqual([ + { method: 'dropExports', erefs: ['o+2'] }, + ]); + }); + + // The kref going before delivery is ordinary, but it is the one thing an + // operator has to go on when a GC action produces nothing, and this is the + // whole of the delivery when cleanup reached every kref. + it('reports krefs that cleanup reached first, even when none survive', async () => { + const { kernelStore, runCrank, delivered, endpoints, logged } = + await makeFixture(); + registerEndpoint(endpoints, delivered, 'v1'); + const kref = kernelStore.initKernelObject('v1'); + kernelStore.addCListEntry('v1', kref, 'o+1'); + kernelStore.setObjectRefCount(kref, { reachable: 0, recognizable: 1 }); + kernelStore.addGCActions([`v1 dropExport ${kref}`]); + + // After selection, as `nextTerminatedVatCleanup` does. + await runCrank(() => { + kernelStore.deleteCListEntry('v1', kref, 'o+1'); + }); + + expect(delivered).toStrictEqual([]); + expect(logged.flat()).toContainEqual( + expect.stringContaining('1 of 1 kref(s) were cleaned up before delivery'), + ); + }); }); diff --git a/packages/ocap-kernel/src/garbage-collection/gc-finalize.ts b/packages/ocap-kernel/src/garbage-collection/gc-finalize.ts index fc78051ee..c574662a0 100644 --- a/packages/ocap-kernel/src/garbage-collection/gc-finalize.ts +++ b/packages/ocap-kernel/src/garbage-collection/gc-finalize.ts @@ -50,6 +50,14 @@ export function makeGCAndFinalize(logger?: Logger): () => Promise { const gcFunction = await gcFunctionPromise; if (gcFunction) { + // Drain the queues *before* collecting. A pending continuation still + // holds its closure's objects, so a sweep run with work outstanding + // finds them reachable and drops nothing — which is the difference + // between a vat reporting its dead imports on this `bringOutYourDead` + // and reporting them on some later one. Twice, because a drained turn + // can itself schedule the next. + await delay(0); + await delay(0); // First GC pass gcFunction(); // Allow finalization callbacks to run diff --git a/packages/ocap-kernel/src/store/index.ts b/packages/ocap-kernel/src/store/index.ts index 7186fca95..e735ed4de 100644 --- a/packages/ocap-kernel/src/store/index.ts +++ b/packages/ocap-kernel/src/store/index.ts @@ -303,22 +303,31 @@ export function makeKernelStore(kdb: KernelDatabase, logger?: Logger) { * These are invisible to `createCrankSavepoint`'s ordinal naming, so one * opened inside a crank would be rolled back by a delivery that has nothing * to do with it — after its owner had already reported success to a peer. - * Callers take their turn through `beginOutOfCrank`. + * Callers take their turn through `withStoreOutOfCrank`. * * @param name - The savepoint name. */ function createSavepoint(name: string): void { !context.inCrank || - Fail`createSavepoint ${q(name)} inside a crank; use beginOutOfCrank`; + Fail`createSavepoint ${q(name)} inside a crank; use withStoreOutOfCrank`; kdb.createSavepoint(name); } /** * Release (commit) a savepoint. * + * Refused inside a crank for the reason `createSavepoint` gives, and because + * a caller that got here anyway would find its savepoint below the crank's + * on the stack: releasing it takes the crank's two with it and commits a + * delivery still in flight. This is the half `createSavepoint`'s guard + * cannot cover, since a caller can be inside a crank by the time it releases + * without having been inside one when it opened. + * * @param name - The savepoint name. */ function releaseSavepoint(name: string): void { + !context.inCrank || + Fail`releaseSavepoint ${q(name)} inside a crank; use withStoreOutOfCrank`; kdb.releaseSavepoint(name); } diff --git a/packages/ocap-kernel/src/store/methods/clist-accounting.test.ts b/packages/ocap-kernel/src/store/methods/clist-accounting.test.ts index 34494f305..8a5175325 100644 --- a/packages/ocap-kernel/src/store/methods/clist-accounting.test.ts +++ b/packages/ocap-kernel/src/store/methods/clist-accounting.test.ts @@ -1,6 +1,7 @@ import { describe, it, expect, beforeEach } from 'vitest'; import { makeMapKernelDatabase } from '../../../test/storage.ts'; +import type { RemoteInfo } from '../../remotes/types.ts'; import type { VatConfig, VatId } from '../../types.ts'; import { makeKernelStore } from '../index.ts'; @@ -389,4 +390,72 @@ describe('c-list reference accounting', () => { expect(kernelStore.auditRefCounts()).toStrictEqual([]); }); }); + + describe('a remote importer', () => { + beforeEach(() => { + kernelStore.setRemoteInfo('r1', { peerId: 'peer-1' } as RemoteInfo); + kernelStore.initEndpoint('r1'); + }); + + it('counts towards an object the same as a vat does', () => { + const kref = kernelStore.exportFromEndpoint('v1', 'o+1'); + kernelStore.translateRefKtoE('r1', kref, true); + + expect(kernelStore.getObjectRefCount(kref)).toStrictEqual({ + reachable: 1, + recognizable: 1, + }); + expect(kernelStore.getImporters(kref)).toStrictEqual(['r1']); + expect(kernelStore.auditRefCounts()).toStrictEqual([]); + }); + + // `retireKernelObjects` deletes the object once it has told every importer, + // so an importer it never enumerated is left holding a c-list entry naming + // nothing — which nothing tears down, and which the audit reports as + // dangling, taking the run loop with it. + it('is told to retire an object the owner has abandoned', () => { + const kref = kernelStore.exportFromEndpoint('v1', 'o+1'); + kernelStore.translateRefKtoE('r1', kref, true); + // Dropped but still recognized, so collection retires rather than drops. + kernelStore.clearReachableFlag('r1', kref); + kernelStore.orphanKernelObject(kref, 'v1'); + + kernelStore.collectGarbage(); + + expect([...kernelStore.getGCActions()]).toStrictEqual([ + `r1 retireImport ${kref}`, + ]); + expect(kernelStore.auditRefCounts()).toStrictEqual([]); + }); + }); + + describe('a terminated importer cleanup has not reached', () => { + beforeEach(() => { + // `deleteVat` reaches `removeVatFromSubcluster`, which fails for a vat + // belonging to no subcluster. + const subclusterId = kernelStore.addSubcluster({ + bootstrap: 'alice', + vats: {}, + } as unknown as Parameters[0]); + kernelStore.addSubclusterVat(subclusterId, 'alice', 'v1'); + kernelStore.addSubclusterVat(subclusterId, 'bob', 'v2'); + }); + + it('is told to retire an object the owner has abandoned', () => { + const kref = kernelStore.exportFromEndpoint('v1', 'o+1'); + kernelStore.translateRefKtoE('v2', kref, true); + kernelStore.clearReachableFlag('v2', kref); + kernelStore.deleteVat('v2'); + kernelStore.markVatAsTerminated('v2'); + kernelStore.orphanKernelObject(kref, 'v1'); + + kernelStore.collectGarbage(); + + expect(kernelStore.getImporters(kref)).toStrictEqual(['v2']); + expect([...kernelStore.getGCActions()]).toStrictEqual([ + `v2 retireImport ${kref}`, + ]); + expect(kernelStore.auditRefCounts()).toStrictEqual([]); + }); + }); }); diff --git a/packages/ocap-kernel/src/store/methods/crank.cross-crank-gc.test.ts b/packages/ocap-kernel/src/store/methods/crank.cross-crank-gc.test.ts index e503d7981..5cd8d90b3 100644 --- a/packages/ocap-kernel/src/store/methods/crank.cross-crank-gc.test.ts +++ b/packages/ocap-kernel/src/store/methods/crank.cross-crank-gc.test.ts @@ -9,7 +9,7 @@ import { makeKernelStore } from '../index.ts'; * candidate added while no crank was open is still owed a collection and has to * survive an unrelated crank's rollback. * - * `RemoteManager.#handlePeerIncarnation` is one such producer. It runs from a + * `RemoteManager.#handleIncarnationChange` is one such producer. It runs from a * network callback with no crank open, under its own `peerIncarnation_` * savepoint, and `persistPeerRestart` -> `forgetEndpointImports` adds every * export the restarting peer abandoned. It calls no `collectGarbage` of its own, @@ -30,7 +30,7 @@ describe('a GC candidate produced outside a crank', () => { function orphanARemoteExport(): string { const kref = kernelStore.initKernelObject('r1'); kernelStore.addCListEntry('r1', kref, 'o+1'); - // RemoteManager.#handlePeerIncarnation, inside its own savepoint, no crank. + // RemoteManager.#handleIncarnationChange, in its own savepoint, no crank. kernelStore.forgetEndpointImports('r1'); return kref; } diff --git a/packages/ocap-kernel/src/store/methods/crank.out-of-crank.test.ts b/packages/ocap-kernel/src/store/methods/crank.out-of-crank.test.ts index 5130b57dd..e1d373a6d 100644 --- a/packages/ocap-kernel/src/store/methods/crank.out-of-crank.test.ts +++ b/packages/ocap-kernel/src/store/methods/crank.out-of-crank.test.ts @@ -31,6 +31,18 @@ describe('store work outside a crank', () => { ); }); + // The other half: a caller can be outside a crank when it opens its + // savepoint and inside one by the time it releases, and that release takes + // the crank's savepoints with it. + it('refuses a savepoint released inside a crank', () => { + kernelStore.createSavepoint('receive_r1_7'); + kernelStore.startCrank(); + + expect(() => kernelStore.releaseSavepoint('receive_r1_7')).toThrow( + 'releaseSavepoint "receive_r1_7" inside a crank', + ); + }); + it('refuses a crank started while a caller holds the store', async () => { await kernelStore.withStoreOutOfCrank(() => { expect(() => kernelStore.startCrank()).toThrow( diff --git a/packages/ocap-kernel/src/store/methods/crank.ts b/packages/ocap-kernel/src/store/methods/crank.ts index db38a24ef..fe31d98b3 100644 --- a/packages/ocap-kernel/src/store/methods/crank.ts +++ b/packages/ocap-kernel/src/store/methods/crank.ts @@ -105,7 +105,7 @@ export function getCrankMethods(ctx: StoreContext, kdb: KernelDatabase) { * both callers take a savepoint inside it, which would then nest inside that * crank rather than being the commit point it has to be. The type refuses the * plain cases and the check below catches the rest; a caller that needs to - * await does it with what `work` hands back. + * await does it with what `work` hands back, as `VatManager.#trackFlux` does. * * @param work - The synchronous work to do while holding the store. * @returns What `work` returned. @@ -208,10 +208,10 @@ export function getCrankMethods(ctx: StoreContext, kdb: KernelDatabase) { // place, `collectGarbage` throws on a later crank for any promise this one // created, killing the run loop over work that no longer exists. // Restored to the savepoint's snapshot rather than cleared, because the - // set is not per-crank: only `collectGarbage` empties it, so a candidate - // added while the run loop was idle — `terminateVat` unpinning a root is - // the real path — is still owed a collection and must survive an - // unrelated crank's rollback. + // set is not per-crank: nothing empties it between cranks, so a candidate + // added while no crank was open — `terminateVat` unpinning a root is the + // real path — is still owed a collection and must survive an unrelated + // crank's rollback. ctx.maybeFreeKrefs.clear(); for (const kref of restored.maybeFreeKrefs) { ctx.maybeFreeKrefs.add(kref); diff --git a/packages/ocap-kernel/src/store/methods/gc.ts b/packages/ocap-kernel/src/store/methods/gc.ts index 92a093c96..f35462eee 100644 --- a/packages/ocap-kernel/src/store/methods/gc.ts +++ b/packages/ocap-kernel/src/store/methods/gc.ts @@ -140,8 +140,8 @@ export function getGCMethods(ctx: StoreContext) { const newActions: GCAction[] = []; for (const koid of koids) { const importers = getImporters(koid); - for (const vatID of importers) { - newActions.push(makeGCAction(vatID, 'retireImport', koid)); + for (const endpointId of importers) { + newActions.push(makeGCAction(endpointId, 'retireImport', koid)); } deleteKernelObject(koid); } diff --git a/packages/ocap-kernel/src/store/methods/remote.ts b/packages/ocap-kernel/src/store/methods/remote.ts index 16a99cc99..5e0aa4b6e 100644 --- a/packages/ocap-kernel/src/store/methods/remote.ts +++ b/packages/ocap-kernel/src/store/methods/remote.ts @@ -48,7 +48,7 @@ export function getRemoteMethods(ctx: StoreContext) { } /** - * Get the IDs of all active remotes. + * The IDs of every remote the kernel knows about, without reading their info. * * @returns The remote IDs. */ diff --git a/packages/ocap-kernel/src/store/methods/vat.ts b/packages/ocap-kernel/src/store/methods/vat.ts index 083ff9bfa..c69fa0953 100644 --- a/packages/ocap-kernel/src/store/methods/vat.ts +++ b/packages/ocap-kernel/src/store/methods/vat.ts @@ -147,7 +147,7 @@ export function getVatMethods(ctx: StoreContext) { /** * Checks if an endpoint imports the specified kernel slot. * - * @param endpointId - The ID of the endpoint to check. + * @param endpointId - The ID of the vat or remote to check. * @param kernelSlot - The kernel slot reference. * @returns True if the endpoint imports the kernel slot, false otherwise. */ @@ -169,17 +169,33 @@ export function getVatMethods(ctx: StoreContext) { /** * Gets all endpoints that import a specific kernel object. * - * Remotes as well as vats: `retireKernelObjects` deletes the object once it - * has told the importers, so one missed here keeps a c-list entry naming a - * kref that no longer exists — which the reference count audit reports as - * dangling, killing the run loop. + * Remotes count, and so do terminated vats cleanup has not reached yet. + * `retireKernelObjects` deletes the object once it has queued a + * `retireImport` for each importer, so an importer missing from this list + * keeps a c-list entry naming an object that no longer exists — which nothing + * ever tears down, and which the refcount audit reports as dangling. + * + * A terminated vat is the case `getVatIDs` alone cannot see: `deleteVat` + * drops the `vatConfig` row it enumerates, while the vat's c-list survives + * until `nextTerminatedVatCleanup` reaches it, one vat per crank. Terminate an + * object's owner and its importer close together, owner marked first, and the + * importer is deregistered but still holding the import when the orphaned + * object is collected. * * @param koid - The kernel object ID. * @returns An array of endpoint IDs that import the kernel object. */ function getImporters(koid: KRef): EndpointId[] { - const importers: EndpointId[] = [...getVatIDs(), ...getRemoteIDs()].filter( - (endpointId) => importsKernelSlot(endpointId, koid), + // Deduplicated: a vat marked terminated whose config survives — a launch + // whose cleanup could not record the death — appears in both lists, and a + // repeated importer would be a second `retireImport` for one entry. + const endpointIds = new Set([ + ...getVatIDs(), + ...getTerminatedVats(), + ...getRemoteIDs(), + ]); + const importers: EndpointId[] = [...endpointIds].filter((endpointId) => + importsKernelSlot(endpointId, koid), ); importers.sort(); return importers; diff --git a/packages/ocap-kernel/src/types.ts b/packages/ocap-kernel/src/types.ts index 9a9f1e536..f82f7e250 100644 --- a/packages/ocap-kernel/src/types.ts +++ b/packages/ocap-kernel/src/types.ts @@ -376,11 +376,27 @@ export type RunQueueItemBringOutYourDead = Infer< typeof RunQueueItemBringOutYourDeadStruct >; +/** + * A request to replace a vat's worker, queued so the run loop performs it. + * + * Queued rather than done where it is asked for, because the run loop is then the + * only thing that takes a vat out of the kernel's reach: no crank can observe the + * vat mid-replacement, and the vat is idle when it happens, since the crank doing + * the work is the one that would otherwise be delivering to it. + */ +const RunQueueItemRestartVatStruct = object({ + type: literal('restartVat'), + vatId: VatIdStruct, +}); + +export type RunQueueItemRestartVat = Infer; + export const RunQueueItemStruct = union([ RunQueueItemSendStruct, RunQueueItemNotifyStruct, RunQueueItemGCActionStruct, RunQueueItemBringOutYourDeadStruct, + RunQueueItemRestartVatStruct, ]); export type RunQueueItem = Infer; diff --git a/packages/ocap-kernel/src/vats/VatHandle.ts b/packages/ocap-kernel/src/vats/VatHandle.ts index a1c18a72c..771f7d4dd 100644 --- a/packages/ocap-kernel/src/vats/VatHandle.ts +++ b/packages/ocap-kernel/src/vats/VatHandle.ts @@ -16,10 +16,7 @@ import { isJsonRpcNotification, isJsonRpcResponse } from '@metamask/utils'; import type { JsonRpcNotification, JsonRpcResponse } from '@metamask/utils'; import type { KernelQueue } from '../KernelQueue.ts'; -import { - makeKernelError, - makeFatalKernelError, -} from '../liveslots/kernel-marshal.ts'; +import { makeFatalKernelError } from '../liveslots/kernel-marshal.ts'; import { vatMethodSpecs, vatSyscallHandlers } from '../rpc/index.ts'; import type { PingVatResult, VatMethod } from '../rpc/index.ts'; import type { KernelStore } from '../store/index.ts'; @@ -45,6 +42,14 @@ type VatConstructorProps = { vatStream: VatStream; kernelStore: KernelStore; kernelQueue: KernelQueue; + /** + * Called when this vat has failed in a way it cannot come back from, so the + * manager can end it. See the drain handler in {@link VatHandle.make}. + * + * Handed the handle, because the failure can come before `make` has returned + * it. + */ + onCriticalFailure: (error: Error, vat: VatHandle) => void; logger?: Logger | undefined; allowedGlobalNames?: AllowedGlobalName[] | undefined; }; @@ -68,17 +73,14 @@ export class VatHandle implements EndpointHandle { /** Optional list of allowed global names for vat endowments */ readonly #allowedGlobalNames: AllowedGlobalName[] | undefined; - /** Storage holding the kernel's persistent state */ - readonly #kernelStore: KernelStore; - /** Storage holding this vat's persistent state */ readonly #vatStore: VatStore; /** The vat's syscall */ readonly #vatSyscall: VatSyscall; - /** The kernel's queue */ - readonly #kernelQueue: KernelQueue; + /** Tells the manager this vat cannot be delivered to again */ + readonly #onCriticalFailure: (error: Error, vat: VatHandle) => void; readonly #rpcClient: RpcClient; @@ -93,6 +95,7 @@ export class VatHandle implements EndpointHandle { * @param params.vatStream - Communications channel connected to the vat worker. * @param params.kernelStore - The kernel's persistent state store. * @param params.kernelQueue - The kernel's queue. + * @param params.onCriticalFailure - Called when the vat has failed unrecoverably. * @param params.logger - Optional logger for error and diagnostic output. * @param params.allowedGlobalNames - Optional list of allowed global names for vat endowments. */ @@ -103,6 +106,7 @@ export class VatHandle implements EndpointHandle { vatStream, kernelStore, kernelQueue, + onCriticalFailure, logger, allowedGlobalNames, }: VatConstructorProps) { @@ -111,9 +115,8 @@ export class VatHandle implements EndpointHandle { this.#logger = logger; this.#allowedGlobalNames = allowedGlobalNames; this.#vatStream = vatStream; - this.#kernelStore = kernelStore; this.#vatStore = kernelStore.makeVatStore(vatId); - this.#kernelQueue = kernelQueue; + this.#onCriticalFailure = onCriticalFailure; this.#vatSyscall = new VatSyscall({ vatId, kernelQueue, @@ -144,6 +147,7 @@ export class VatHandle implements EndpointHandle { * @param params.vatStream - Communications channel connected to the vat worker. * @param params.kernelStore - The kernel's persistent state store. * @param params.kernelQueue - The kernel's queue. + * @param params.onCriticalFailure - Called when the vat has failed unrecoverably. * @param params.logger - Optional logger for error and diagnostic output. * @returns A promise for the new VatHandle instance. */ @@ -165,11 +169,16 @@ export class VatHandle implements EndpointHandle { */ async #init(): Promise { Promise.all([this.#vatStream.drain(this.#handleMessage.bind(this))]).catch( - async (error) => { + (error) => { this.#logger?.error(`Unexpected read error`, error); - await this.terminate( - true, + // Handed to the manager rather than torn down here. A handle that + // retires itself leaves the manager still holding it and the store + // still calling the vat live, so the next delivery is handed to a + // worker that cannot answer and the crank never completes. Only the + // manager can put the vat's death on record. + this.#onCriticalFailure( new StreamReadError({ vatId: this.vatId }, error), + this, ); }, ); @@ -306,27 +315,25 @@ export class VatHandle implements EndpointHandle { } /** - * Terminates the vat. + * Closes this handle's channel to the vat worker. + * + * Only the handle's own business: the store side of a vat's death belongs to + * `VatManager.#retireVat`, which writes it in one synchronous step. Split that + * way because the two have opposite failure requirements — ending a stream can + * fail and it does not matter, since the worker is already being killed, while + * a store left half-told about a vat is a state nothing recovers from. * - * @param terminating - If true, the vat is being killed permanently, so clean - * up its state and reject any promises that would be left dangling. + * @param terminating - If true, the vat is being killed permanently, so + * callers waiting on a command it will never answer are told now. * @param error - The error to terminate the vat with. */ async terminate(terminating: boolean, error?: Error): Promise { - await this.#vatStream.end(error); - const terminationError = error ?? new VatDeletedError(this.vatId); if (terminating) { - // Reject promises exported to other vats for which this vat is the decider - const failure = makeKernelError( - 'VAT_TERMINATED', - terminationError.message, - ); - for (const kpid of this.#kernelStore.getPromisesByDecider(this.vatId)) { - this.#kernelQueue.resolvePromises(this.vatId, [[kpid, true, failure]]); - } - this.#rpcClient.rejectAll(terminationError); - this.#kernelStore.deleteVat(this.vatId); + // Ahead of the stream, so a stream that refuses to close does not leave + // these callers waiting on a worker that is already dead. + this.#rpcClient.rejectAll(error ?? new VatDeletedError(this.vatId)); } + await this.#vatStream.end(error); } /** diff --git a/packages/ocap-kernel/src/vats/VatManager.test.ts b/packages/ocap-kernel/src/vats/VatManager.test.ts index 90cec52c3..abc5653ff 100644 --- a/packages/ocap-kernel/src/vats/VatManager.test.ts +++ b/packages/ocap-kernel/src/vats/VatManager.test.ts @@ -15,6 +15,17 @@ import type { VatId, VatConfig, PlatformServices } from '../types.ts'; import { VatHandle } from './VatHandle.ts'; import { VatManager } from './VatManager.ts'; +/** + * Let the pending microtasks run, so an operation under test gets as far as its + * first real await. + * + * @returns A promise that resolves once the microtask queue has drained. + */ +const drainMicrotasks = async (): Promise => + new Promise((resolve) => { + setTimeout(resolve, 0); + }); + describe('VatManager', () => { let mockPlatformServices: Mocked; let mockKernelStore: Mocked; @@ -23,6 +34,7 @@ describe('VatManager', () => { let vatManager: VatManager; let makeVatHandleMock: MockInstance; let vatHandles: Mocked[]; + let runLoopDeathWaiters: Set<(error: Error) => void>; const createMockVatConfig = (name = 'test'): VatConfig => ({ sourceSpec: `${name}.js`, @@ -35,7 +47,9 @@ describe('VatManager', () => { const handle = { vatId, config, - terminate: vi.fn(), + // Resolved rather than bare, so callers that chain off it — rather than + // awaiting — behave as they would against the real async method. + terminate: vi.fn().mockResolvedValue(undefined), ping: vi.fn().mockResolvedValue({ pong: true }), } as unknown as Mocked; vatHandles.push(handle); @@ -44,6 +58,13 @@ describe('VatManager', () => { beforeEach(() => { vatHandles = []; + runLoopDeathWaiters = new Set(); + + // Stateful rather than bare mocks, because the real `deleteVat` refuses a + // repeat: against one that shrugs, a caller retiring a vat twice passes + // here and throws in production. + const terminatedVats = new Set(); + const deletedVats = new Set(); mockPlatformServices = { launch: vi.fn().mockResolvedValue({ @@ -69,17 +90,49 @@ describe('VatManager', () => { })(), ), getVatSubcluster: vi.fn().mockReturnValue('s1'), - markVatAsTerminated: vi.fn(), + markVatAsTerminated: vi.fn((vatId: VatId) => { + terminatedVats.add(vatId); + }), + isVatTerminated: vi.fn((vatId: VatId) => terminatedVats.has(vatId)), + // These tests reach the store through `runVat`, which writes no config + // row, so the store knows of no vat unless a test says otherwise. + isVatActive: vi.fn().mockReturnValue(false), + deleteVat: vi.fn((vatId: VatId) => { + if (deletedVats.has(vatId)) { + throw new Error(`Vat "${vatId}" has no subcluster`); + } + deletedVats.add(vatId); + }), + getPromisesByDecider: vi.fn().mockReturnValue([]), getRootObject: vi.fn().mockReturnValue('ko1'), pinObject: vi.fn(), unpinObject: vi.fn(), scheduleReap: vi.fn(), nextTerminatedVatCleanup: vi.fn().mockReturnValue(false), collectGarbage: vi.fn(), + // The real one holds the store out of crank for the duration of `work`; + // what matters to these tests is that `work` runs synchronously within it. + withStoreOutOfCrank: vi.fn(async (work: () => unknown) => work()), } as unknown as Mocked; mockKernelQueue = { waitForCrank: vi.fn().mockResolvedValue(undefined), + resolvePromises: vi.fn(), + // A restart is the run loop's work, so stand in for it reaching the + // request on its next crank. Nothing is expected to come back out: + // `performVatRestart` reports a failure through the waiter `restartVat` + // registered, precisely so that it never takes the crank down. The catch + // is here so that a regression on that shows up as a failing assertion + // rather than an unhandled rejection. + enqueueRestartVat: vi.fn((vatId: VatId) => { + vatManager.performVatRestart(vatId).catch(() => undefined); + }), + // The real one hands back an unregister and calls every rejecter it + // holds when the loop dies; the tests below reach into the set directly. + onRunLoopDeath: vi.fn((reject: (error: Error) => void) => { + runLoopDeathWaiters.add(reject); + return () => runLoopDeathWaiters.delete(reject); + }), } as unknown as Mocked; mockLogger = new Logger('test'); @@ -402,7 +455,7 @@ describe('VatManager', () => { await vatManager.terminateVat('v1'); - expect(mockKernelQueue.waitForCrank).toHaveBeenCalled(); + expect(mockKernelStore.withStoreOutOfCrank).toHaveBeenCalled(); expect(mockPlatformServices.terminate).toHaveBeenCalled(); expect(vatHandles[0]?.terminate).toHaveBeenCalled(); expect(mockKernelStore.markVatAsTerminated).toHaveBeenCalledWith('v1'); @@ -421,9 +474,323 @@ describe('VatManager', () => { expect.objectContaining({ message: 'Vat termination: Custom reason' }), ); }); + + // The run loop starts its next crank in the same turn it ends the last, so + // waiting for the crank in flight left the death to be written inside the + // next one's delivery savepoint, and left that crank free to look the vat + // up before the record existed. Holding the store is what closes both. + it('writes nothing until it holds the store out of crank', async () => { + await vatManager.runVat('v1', createMockVatConfig()); + let openTheGate!: () => void; + ( + mockKernelStore.withStoreOutOfCrank as unknown as MockInstance + ).mockImplementationOnce(async (work: () => unknown) => { + await new Promise((resolve) => { + openTheGate = resolve; + }); + return work(); + }); + + const terminated = vatManager.terminateVat('v1'); + await drainMicrotasks(); + + expect(mockKernelStore.markVatAsTerminated).not.toHaveBeenCalled(); + expect(vatManager.hasVat('v1')).toBe(true); + + openTheGate(); + await terminated; + + expect(mockKernelStore.markVatAsTerminated).toHaveBeenCalledWith('v1'); + expect(vatManager.hasVat('v1')).toBe(false); + }); + }); + + describe('recording a vat as dead', () => { + /** + * The four writes that make up a vat's death, as the store saw them. + * + * @returns How many times each was made. + */ + const recorded = (): { + rejectedItsPromises: number; + unpinnedItsRoot: number; + deletedItsRecords: number; + marked: number; + } => { + const callsTo = (mock: unknown): number => + (mock as MockInstance).mock.calls.length; + return { + rejectedItsPromises: callsTo(mockKernelQueue.resolvePromises), + unpinnedItsRoot: callsTo(mockKernelStore.unpinObject), + deletedItsRecords: callsTo(mockKernelStore.deleteVat), + marked: callsTo(mockKernelStore.markVatAsTerminated), + }; + }; + + it('records all of it even when the worker refuses to go', async () => { + await vatManager.runVat('v1', createMockVatConfig()); + ( + mockKernelStore.getPromisesByDecider as unknown as MockInstance + ).mockReturnValueOnce(['kp1']); + ( + vatHandles[0]?.terminate as unknown as MockInstance + ).mockRejectedValueOnce(new Error('stream would not close')); + + await expect(vatManager.terminateVat('v1')).rejects.toThrow( + 'stream would not close', + ); + + // A partial record is the state nothing recovers from: marked terminated + // while `vatConfig` survives reads as *active* again as soon as cleanup + // drops the mark, and the router kills the run loop over the + // disagreement. All four land, or the failure above is the lesser bug. + expect(recorded()).toStrictEqual({ + rejectedItsPromises: 1, + unpinnedItsRoot: 1, + deletedItsRecords: 1, + marked: 1, + }); + expect(vatManager.hasVat('v1')).toBe(false); + }); + + it('records it for a vat the store still lists but the kernel has lost', async () => { + // What `terminateSubcluster` hands us: it iterates the store's own vat + // list, which can name a vat whose handle is already gone. + (mockKernelStore.isVatActive as unknown as MockInstance) = vi + .fn() + .mockReturnValue(true); + + await vatManager.terminateVat('v1'); + + expect(mockKernelStore.deleteVat).toHaveBeenCalledWith('v1'); + expect(mockKernelStore.markVatAsTerminated).toHaveBeenCalledWith('v1'); + }); + + it('refuses a vat neither the kernel nor the store knows about', async () => { + (mockKernelStore.isVatActive as unknown as MockInstance) = vi + .fn() + .mockReturnValue(false); + + await expect(vatManager.terminateVat('v9')).rejects.toThrow( + VatNotFoundError, + ); + expect(mockKernelStore.markVatAsTerminated).not.toHaveBeenCalled(); + }); + + /** + * Report a fatal stream failure for a vat, as its handle's drain catch does. + * + * @param vat - The handle reporting it. + */ + const failStream = (vat: VatHandle): void => { + const { onCriticalFailure } = makeVatHandleMock.mock + .calls[0]?.[0] as unknown as { + onCriticalFailure: (error: Error, failed: VatHandle) => void; + }; + onCriticalFailure(new Error('read error'), vat); + }; + + it('records it when a vat`s stream fails under it', async () => { + await vatManager.runVat('v1', createMockVatConfig()); + + failStream(vatHandles[0] as VatHandle); + + // Left on the books, the handle stays resolvable, so the next delivery + // goes to a worker that cannot answer and the crank never completes — + // the RPC client has no timeout. + expect(vatManager.hasVat('v1')).toBe(false); + expect(mockKernelStore.markVatAsTerminated).toHaveBeenCalledWith('v1'); + }); + + // A write failing partway leaves the handle already dropped. The mark is + // what stops the store from going on calling the vat active, which is the + // disagreement the endpoint lookup kills the run loop over. + it('marks the vat terminated even when the rest of the record fails', async () => { + await vatManager.runVat('v1', createMockVatConfig()); + mockKernelStore.deleteVat.mockImplementationOnce(() => { + throw new Error('no subcluster'); + }); + + failStream(vatHandles[0] as VatHandle); + + expect(vatManager.hasVat('v1')).toBe(false); + expect(mockKernelStore.markVatAsTerminated).toHaveBeenCalledWith('v1'); + // And the worker is still torn down, rather than the teardown being lost + // with the throw. + expect(mockPlatformServices.terminate).toHaveBeenCalledWith( + 'v1', + expect.any(Error), + ); + }); + + it('records it once for a vat retired twice', async () => { + await vatManager.runVat('v1', createMockVatConfig()); + ( + mockKernelStore.getPromisesByDecider as unknown as MockInstance + ).mockReturnValueOnce(['kp1']); + + failStream(vatHandles[0] as VatHandle); + failStream(vatHandles[0] as VatHandle); + + expect(recorded()).toStrictEqual({ + rejectedItsPromises: 1, + unpinnedItsRoot: 1, + deletedItsRecords: 1, + marked: 1, + }); + }); + + it('rejects the delivery in flight when a vat`s stream fails under it', async () => { + await vatManager.runVat('v1', createMockVatConfig()); + + failStream(vatHandles[0] as VatHandle); + + // Recording the death only helps the *next* delivery. The one that was in + // flight when the worker died is still parked on an RPC client with no + // timeout, so its crank never completes — the same hang, one delivery + // earlier. `terminate` is what rejects it, and the worker has to go too. + await vi.waitFor(() => { + expect(vatHandles[0]?.terminate).toHaveBeenCalled(); + expect(mockPlatformServices.terminate).toHaveBeenCalledWith( + 'v1', + expect.any(Error), + ); + }); + }); + + it('rejects it without waiting for the worker to die', async () => { + await vatManager.runVat('v1', createMockVatConfig()); + ( + mockPlatformServices.terminate as unknown as MockInstance + ).mockReturnValueOnce(new Promise(() => undefined)); + + failStream(vatHandles[0] as VatHandle); + + // A worker that will not go must not be what keeps the delivery parked. + await vi.waitFor(() => { + expect(vatHandles[0]?.terminate).toHaveBeenCalledWith( + true, + expect.any(Error), + ); + }); + }); + + it('records it when the stream fails before the handle is returned', async () => { + // The failure can land while `VatHandle.make` is still initializing the + // vat, when the manager has no handle of its own to tear down with — and + // the pending `initVat` that nothing else will settle is exactly what is + // owed a rejection. + makeVatHandleMock.mockImplementationOnce( + async ({ vatId, vatConfig, onCriticalFailure }) => { + const handle = createMockVatHandle(vatId, vatConfig); + onCriticalFailure(new Error('read error'), handle); + return handle; + }, + ); + + await expect( + vatManager.runVat('v1', createMockVatConfig()), + ).rejects.toThrow('read error'); + + expect(vatHandles[0]?.terminate).toHaveBeenCalledWith( + true, + expect.any(Error), + ); + // On the books, this handle would be one the store already calls dead. + expect(vatManager.hasVat('v1')).toBe(false); + expect(mockKernelStore.markVatAsTerminated).toHaveBeenCalledWith('v1'); + }); + + it('records none of it for a restart', async () => { + await vatManager.runVat('v1', createMockVatConfig()); + + await vatManager.stopVat('v1', false); + + // The same vat, and the same root, are coming back. + expect(recorded()).toStrictEqual({ + rejectedItsPromises: 0, + unpinnedItsRoot: 0, + deletedItsRecords: 0, + marked: 0, + }); + }); }); describe('restartVat', () => { + // The run loop is what carries the request out, and it is checked alive + // only at enqueue time. A loop that dies afterwards leaves this caller with + // nothing to settle it: no kernel promise stands behind a restart the way + // one stands behind a message result. + it('reports the run loop dying before it got to the request', async () => { + await vatManager.runVat('v1', createMockVatConfig()); + ( + mockKernelQueue.enqueueRestartVat as unknown as MockInstance + ).mockImplementationOnce(() => undefined); + + const restarted = vatManager.restartVat('v1'); + expect(runLoopDeathWaiters.size).toBe(1); + for (const reject of runLoopDeathWaiters) { + reject(new Error('Kernel run loop died')); + } + + await expect(restarted).rejects.toThrow('Kernel run loop died'); + }); + + // Both outcomes, because only the resolving one unregisters on its own + // path: a rejecter left behind is called later against a promise nobody is + // awaiting. + it.each([ + { outcome: 'succeeds', arrange: () => undefined }, + { + outcome: 'fails', + arrange: () => { + ( + mockPlatformServices.launch as unknown as MockInstance + ).mockRejectedValueOnce(new Error('no worker')); + }, + }, + ])( + 'stops watching the run loop once the restart $outcome', + async ({ arrange }) => { + await vatManager.runVat('v1', createMockVatConfig()); + arrange(); + + await vatManager.restartVat('v1').catch(() => undefined); + + expect(runLoopDeathWaiters.size).toBe(0); + }, + ); + + // `enqueueRestartVat` refuses because the run loop is dead, which is the + // same thing the waiter registered for — so it has already been rejected, + // and this throw is what the caller sees instead. Nothing else is awaiting + // it. + it('leaves no unhandled rejection when the request cannot be queued', async () => { + await vatManager.runVat('v1', createMockVatConfig()); + const unhandled: unknown[] = []; + const onUnhandled = (reason: unknown): void => { + unhandled.push(reason); + }; + process.on('unhandledRejection', onUnhandled); + ( + mockKernelQueue.enqueueRestartVat as unknown as MockInstance + ).mockImplementationOnce((vatId: VatId) => { + for (const reject of runLoopDeathWaiters) { + reject(new Error('Kernel run loop died')); + } + throw new Error(`Kernel run loop died; cannot restart a vat ${vatId}`); + }); + + await expect(vatManager.restartVat('v1')).rejects.toThrow( + 'cannot restart a vat', + ); + await drainMicrotasks(); + process.off('unhandledRejection', onUnhandled); + + expect(unhandled).toStrictEqual([]); + expect(runLoopDeathWaiters.size).toBe(0); + }); + it('restarts a vat successfully', async () => { const config = createMockVatConfig(); await vatManager.runVat('v1', config); @@ -431,7 +798,7 @@ describe('VatManager', () => { const result = await vatManager.restartVat('v1'); - expect(mockKernelQueue.waitForCrank).toHaveBeenCalled(); + expect(mockKernelQueue.enqueueRestartVat).toHaveBeenCalledWith('v1'); expect(originalHandle?.terminate).toHaveBeenCalledWith(false, undefined); expect(mockPlatformServices.launch).toHaveBeenCalledTimes(2); expect(makeVatHandleMock).toHaveBeenCalledTimes(2); @@ -445,6 +812,269 @@ describe('VatManager', () => { VatNotFoundError, ); }); + + it('marks a vat terminated when its relaunch fails', async () => { + await vatManager.runVat('v1', createMockVatConfig()); + makeVatHandleMock.mockRejectedValueOnce(new Error('worker died')); + + await expect(vatManager.restartVat('v1')).rejects.toThrow('worker died'); + + // Nothing else reclaims a vat with no worker that the store still counts + // among the living. + expect(mockKernelStore.markVatAsTerminated).toHaveBeenCalledWith('v1'); + await expect(vatManager.provideVat('v1')).rejects.toThrow( + VatNotFoundError, + ); + }); + + it('releases the root pin when its relaunch fails', async () => { + await vatManager.runVat('v1', createMockVatConfig()); + makeVatHandleMock.mockRejectedValueOnce(new Error('worker died')); + + await expect(vatManager.restartVat('v1')).rejects.toThrow('worker died'); + + // The restart's `stopVat` was told the vat was coming back, so it kept the + // pin, and vat cleanup does not release pins. Without this the root's + // refcount is held for the life of the kernel. + expect(mockKernelStore.unpinObject).toHaveBeenCalledWith('ko1'); + }); + + // The crank has to commit for those records to survive. Thrown instead, the + // run loop's catch rolls the crank back — unmarking the vat, re-pinning its + // root, and returning this very request to the run queue, so the next + // process start dequeues it and fails the same way, forever. + it('reports a failed relaunch without taking the crank down', async () => { + await vatManager.runVat('v1', createMockVatConfig()); + makeVatHandleMock.mockRejectedValueOnce(new Error('worker died')); + const restarted = vatManager.restartVat('v1'); + + await expect(restarted).rejects.toThrow('worker died'); + // The caller heard about it; the crank did not. + expect(await vatManager.performVatRestart('v1')).toBeUndefined(); + }); + + it('records a relaunch failure once when the stream dies before the handle exists', async () => { + await vatManager.runVat('v1', createMockVatConfig()); + ( + mockKernelQueue.enqueueRestartVat as unknown as MockInstance + ).mockImplementation(() => undefined); + const restarted = vatManager.restartVat('v1'); + makeVatHandleMock.mockImplementationOnce( + async ({ vatId, vatConfig, onCriticalFailure }) => { + const handle = createMockVatHandle(vatId, vatConfig); + onCriticalFailure(new Error('read error'), handle); + return handle; + }, + ); + + expect(await vatManager.performVatRestart('v1')).toBeUndefined(); + + expect(mockKernelStore.deleteVat).toHaveBeenCalledTimes(1); + await expect(restarted).rejects.toThrow('read error'); + }); + + it('rejects the promises a vat was deciding when its relaunch fails', async () => { + await vatManager.runVat('v1', createMockVatConfig()); + ( + mockKernelStore.getPromisesByDecider as unknown as MockInstance + ).mockReturnValue(['kp1']); + makeVatHandleMock.mockRejectedValueOnce(new Error('worker died')); + + await expect(vatManager.restartVat('v1')).rejects.toThrow('worker died'); + + // Nothing else will ever decide them: the incarnation that owed them is + // gone and cleanup only tears the c-list down. + expect(mockKernelQueue.resolvePromises).toHaveBeenCalledWith('v1', [ + ['kp1', true, expect.objectContaining({ body: expect.any(String) })], + ]); + }); + + // Both are exposed as RPCs, and `terminateVat` does not go through the run + // queue, so it lands in the window between the request and the crank. + it('drops a queued restart for a vat that was terminated first', async () => { + await vatManager.runVat('v1', createMockVatConfig()); + ( + mockKernelQueue.enqueueRestartVat as unknown as MockInstance + ).mockImplementation(() => undefined); + const restarted = vatManager.restartVat('v1'); + + await vatManager.terminateVat('v1'); + + // The caller is told, rather than left waiting on a request nothing will + // carry out. + await expect(restarted).rejects.toThrow(VatDeletedError); + // And the request itself goes quietly when the run loop reaches it. A + // throw here is a dead kernel: `#restartVatWorker` does not catch. + expect(await vatManager.performVatRestart('v1')).toBeUndefined(); + }); + + it('does not strand a waiter when the request cannot be queued', async () => { + await vatManager.runVat('v1', createMockVatConfig()); + ( + mockKernelQueue.enqueueRestartVat as unknown as MockInstance + ).mockImplementationOnce(() => { + throw new Error('run loop died'); + }); + + await expect(vatManager.restartVat('v1')).rejects.toThrow( + 'run loop died', + ); + + // Left registered, the next request would reject it as superseded — and + // nobody ever awaited it, so that rejection goes unhandled. + ( + mockKernelQueue.enqueueRestartVat as unknown as MockInstance + ).mockImplementation(() => undefined); + const second = vatManager.restartVat('v1'); + await vatManager.performVatRestart('v1'); + expect(await second).toBe(vatHandles[1]); + }); + + it('leaves the vat in place until the run loop takes the request', async () => { + await vatManager.runVat('v1', createMockVatConfig()); + const originalHandle = vatHandles[0]; + // Queue the request without standing in for the run loop. + ( + mockKernelQueue.enqueueRestartVat as unknown as MockInstance + ).mockImplementation(() => undefined); + + const restarted = vatManager.restartVat('v1'); + await drainMicrotasks(); + + // The vat is only ever out of reach inside the crank that carries the + // request out, where no other crank can see it. + expect(vatManager.getVat('v1')).toBe(originalHandle); + expect(originalHandle?.terminate).not.toHaveBeenCalled(); + + await vatManager.performVatRestart('v1'); + + expect(await restarted).toBe(vatHandles[1]); + }); + + it('supersedes a caller waiting on an earlier request for the same vat', async () => { + await vatManager.runVat('v1', createMockVatConfig()); + ( + mockKernelQueue.enqueueRestartVat as unknown as MockInstance + ).mockImplementation(() => undefined); + + const first = vatManager.restartVat('v1'); + const second = vatManager.restartVat('v1'); + + // One waiter per vat, so the earlier caller is told rather than left + // waiting on a restart the later one will consume. + await expect(first).rejects.toThrow('superseded'); + await vatManager.performVatRestart('v1'); + expect(await second).toBe(vatHandles[1]); + }); + + it('queues one request when a second arrives before the crank', async () => { + await vatManager.runVat('v1', createMockVatConfig()); + ( + mockKernelQueue.enqueueRestartVat as unknown as MockInstance + ).mockImplementation(() => undefined); + + const first = vatManager.restartVat('v1'); + const second = vatManager.restartVat('v1'); + await expect(first).rejects.toThrow('superseded'); + + expect(mockKernelQueue.enqueueRestartVat).toHaveBeenCalledOnce(); + await vatManager.performVatRestart('v1'); + expect(await second).toBe(vatHandles[1]); + // The initial launch and exactly one relaunch. + expect(mockPlatformServices.launch).toHaveBeenCalledTimes(2); + }); + + it('queues a fresh request once the crank has taken the last one', async () => { + await vatManager.runVat('v1', createMockVatConfig()); + ( + mockKernelQueue.enqueueRestartVat as unknown as MockInstance + ).mockImplementation(() => undefined); + + const first = vatManager.restartVat('v1'); + await vatManager.performVatRestart('v1'); + await first; + const second = vatManager.restartVat('v1'); + await vatManager.performVatRestart('v1'); + + // The waiter is taken when the crank starts, so a request arriving after + // that has no item to join and needs one of its own. + expect(mockKernelQueue.enqueueRestartVat).toHaveBeenCalledTimes(2); + expect(await second).toBe(vatHandles[2]); + }); + + it('queues a request that arrives while the vat is between workers', async () => { + await vatManager.runVat('v1', createMockVatConfig()); + ( + mockKernelQueue.enqueueRestartVat as unknown as MockInstance + ).mockImplementation(() => undefined); + // Between workers is not gone: the store still lists the vat. + (mockKernelStore.isVatActive as unknown as MockInstance) = vi + .fn() + .mockReturnValue(true); + // Parks the worker kill, which holds `performVatRestart` after it has + // dropped the handle and before `runVat` puts the next one on the books. + let releaseWorkerKill = (): void => undefined; + mockPlatformServices.terminate.mockImplementationOnce( + async () => + new Promise((resolve) => { + releaseWorkerKill = resolve; + }), + ); + + const first = vatManager.restartVat('v1'); + const carriedOut = vatManager.performVatRestart('v1'); + await drainMicrotasks(); + expect(vatManager.hasVat('v1')).toBe(false); + + const second = vatManager.restartVat('v1'); + releaseWorkerKill(); + await carriedOut; + expect(await first).toBe(vatHandles[1]); + + // The second request could not join an item already taken, so it has one + // of its own, which the run loop reaches after this crank. + expect(mockKernelQueue.enqueueRestartVat).toHaveBeenCalledTimes(2); + await vatManager.performVatRestart('v1'); + expect(await second).toBe(vatHandles[2]); + }); + }); + + describe('provideVat', () => { + it('returns the running handle when the vat is not in flux', async () => { + await vatManager.runVat('v1', createMockVatConfig()); + + expect(await vatManager.provideVat('v1')).toBe(vatHandles[0]); + }); + + it('throws if vat not found', async () => { + await expect(vatManager.provideVat('v1')).rejects.toThrow( + VatNotFoundError, + ); + }); + + it('reports a vat gone only once its termination has been recorded', async () => { + await vatManager.runVat('v1', createMockVatConfig()); + let finishStop!: () => void; + (vatHandles[0]?.terminate as unknown as MockInstance).mockImplementation( + async () => + new Promise((resolve) => { + finishStop = resolve; + }), + ); + + const terminated = vatManager.terminateVat('v1'); + await drainMicrotasks(); + const provided = vatManager.provideVat('v1'); + + finishStop(); + + await expect(provided).rejects.toThrow(VatNotFoundError); + // The store agrees by the time a waiter is told, so a caller acting on + // "gone" — releasing the kernel's side of a GC action, say — is acting on + // a vat the store also calls terminated. + expect(mockKernelStore.markVatAsTerminated).toHaveBeenCalledWith('v1'); + await terminated; + }); }); describe('pingVat', () => { diff --git a/packages/ocap-kernel/src/vats/VatManager.ts b/packages/ocap-kernel/src/vats/VatManager.ts index 533940943..d13e568b0 100644 --- a/packages/ocap-kernel/src/vats/VatManager.ts +++ b/packages/ocap-kernel/src/vats/VatManager.ts @@ -1,4 +1,5 @@ import type { CapData } from '@endo/marshal'; +import { makePromiseKit } from '@endo/promise-kit'; import { VatAlreadyExistsError, VatDeletedError, @@ -8,6 +9,7 @@ import { stringify } from '@metamask/kernel-utils'; import { Logger, splitLoggerStream } from '@metamask/logger'; import type { KernelQueue } from '../KernelQueue.ts'; +import { makeKernelError } from '../liveslots/kernel-marshal.ts'; import type { KernelStore } from '../store/index.ts'; import type { VatId, @@ -36,6 +38,36 @@ export class VatManager { /** Currently running vats, by ID */ readonly #vats: Map; + /** + * Vats being torn down, by ID, each mapped to a promise for the teardown. + * {@link provideVat} waits on these, which is what keeps the kernel's answer + * about a dying vat in step with the store's: by the time a waiter is told the + * vat is gone, it is marked terminated, and callers that must tell "terminated" + * from "missing" — {@link KernelRouter}'s endpoint lookup above all — get the + * former rather than a disagreement to raise. + * + * Recorded rather than guarded against: the run loop is free to run cranks + * throughout, and a delivery that arrives mid-flux waits for the vat instead + * of the flux waiting for the run loop. Inverted the other way — a lock the + * operation holds while the loop stands still — the holder must never await + * anything the run loop has to deliver, which is a much sharper edge. + * + * Only termination goes through here. A restart is queued for the run loop + * (see {@link restartVat}), which leaves no window at all; termination cannot + * be, because it has to work on a kernel whose run loop has died. + */ + readonly #vatsInFlux: Map>; + + /** + * Callers waiting for the run loop to carry out a queued restart, by vat ID. + * In RAM only: a request that outlives the kernel that queued it is still in + * the run queue, and is carried out with nobody left to tell. + */ + readonly #restartWaiters: Map< + VatId, + { resolve: () => void; reject: (error: unknown) => void } + >; + /** Service to spawn workers (in iframes) for vats to run in */ readonly #platformServices: PlatformServices; @@ -69,6 +101,8 @@ export class VatManager { allowedGlobalNames, }: VatManagerOptions) { this.#vats = new Map(); + this.#vatsInFlux = new Map(); + this.#restartWaiters = new Map(); this.#platformServices = platformServices; this.#kernelStore = kernelStore; this.#kernelQueue = kernelQueue; @@ -139,6 +173,10 @@ export class VatManager { } catch (error) { // The worker is already running, so leaving it would strand a vat the // kernel has no record of. Tear it down before reporting the failure. + // `stopVat` records the vat as dead before it touches the worker, so + // whatever store records the partial launch did write — the endpoint + // counters, the root's c-list pair, its owner entry — are reclaimed by the + // terminated-vat cleanup even if the worker refuses to go. let stopFailure: unknown; try { await this.stopVat(vatId, true); @@ -149,10 +187,14 @@ export class VatManager { caught, ); } - // `stopVat` only tears down the worker. Whatever store records the - // partial launch did write — the endpoint counters, the root's c-list - // pair, its owner entry — are reclaimed by the terminated-vat cleanup, - // which never runs unless the vat is marked. + // `stopVat` normally records the death itself, via `#retireVat`, before it + // touches the worker. But it can refuse before it gets that far — a vat + // the kernel has no handle for and the store does not call active is one + // it declines outright — and a partial launch is exactly the shape that + // reaches. The mark is what makes the terminated-vat cleanup reclaim the + // endpoint counters, the root's c-list pair and its owner entry, so it is + // asserted here rather than assumed. Marking an already-marked vat is a + // no-op. this.#kernelStore.markVatAsTerminated(vatId); throw new Error( `Failed to launch vat ${vatId} (${vatName})${stopFailure ? ' (cleanup also failed)' : ''}`, @@ -178,15 +220,59 @@ export class VatManager { loggerStream as unknown as Parameters[0], (error) => this.#logger.error(`Vat ${vatId} error: ${stringify(error)}`), ); + // A handle put on the books after its vat was retired is the + // store-says-dead, kernel-says-live disagreement all of this exists to + // prevent, and the stream can break at any point below. + let fatalError: Error | undefined; const vat = await VatHandle.make({ vatId, vatConfig, vatStream, kernelStore: this.#kernelStore, kernelQueue: this.#kernelQueue, + // Takes the handle rather than closing over `vat`, which does not exist + // yet while `make` is initializing — the very window in which the pending + // `initVat` needs rejecting, since nothing else would ever settle it. + onCriticalFailure: (error, failedVat) => { + // The vat's channel has broken, so nothing can be delivered to it again + // and no worker teardown is going to change that. Retire it rather than + // leaving a handle the router will keep resolving successfully, which is + // a crank that never completes: the write goes nowhere and the RPC + // client has no timeout. + this.#logger.error(`Retiring vat ${vatId} after a fatal error:`, error); + fatalError = error; + try { + this.#retireVat(vatId, error); + } catch (retireError) { + // Logged rather than thrown: this is a callback off a stream's drain, + // with nobody to catch it, and the teardown below still has RPCs to + // reject. `fatalError` carries the real diagnosis to `runVat`. + this.#logger.error( + `Failed to record the death of vat ${vatId}:`, + retireError, + ); + // The handle is gone either way — `#retireVat` drops it first — so + // the mark is the one write that cannot be skipped. Without it the + // store goes on calling the vat active while the kernel has no + // handle for it, and `#resolveEndpoint` kills the run loop over the + // disagreement at the next delivery addressed to it. + try { + this.#kernelStore.markVatAsTerminated(vatId); + } catch (markError) { + this.#logger.error( + `Vat ${vatId} could not be marked terminated; the store still calls it active and the kernel has no handle for it:`, + markError, + ); + } + } + this.#startFailedVatTeardown(vatId, failedVat, error); + }, logger: vatLogger, allowedGlobalNames: this.#allowedGlobalNames, }); + if (fatalError) { + throw fatalError; + } this.#vats.set(vatId, vat); } @@ -208,28 +294,204 @@ export class VatManager { terminating: boolean, reason?: CapData, ): Promise { - const vat = this.getVat(vatId); + // A restart needs a live handle to read its config from and to come back + // into; an ending vat does not, and must not, since the vat may be one the + // store still lists while the kernel has already lost its handle. Retiring + // it is exactly what puts that right. + const vat = terminating ? this.#vats.get(vatId) : this.getVat(vatId); + if (terminating && !vat && !this.#kernelStore.isVatActive(vatId)) { + throw new VatNotFoundError(vatId); + } let terminationError: Error | undefined; if (reason) { terminationError = new Error(`Vat termination: ${reason.body}`); } else if (terminating) { terminationError = new VatDeletedError(vatId); } + if (terminating) { + // Everything the kernel has to record about this vat's death, before the + // first await below. See {@link #retireVat}. + this.#retireVat(vatId, terminationError as Error); + } else { + // A restart keeps the pin and the records: the same vat, and the same + // root, are coming back. Only the handle goes. + this.#vats.delete(vatId); + } + // Best-effort from here on, and deliberately after the records: the worker + // is being killed either way, and a teardown that fails must not leave the + // kernel's account of the vat half-written. + await this.#platformServices + .terminate(vatId, terminationError) + .catch(this.#logger.error); + await vat?.terminate(terminating, terminationError); + } + + /** + * Record a vat's death: everything the kernel has to remember about it, in one + * synchronous step. + * + * Synchronous is the whole point. A vat's death is four writes — the promises + * it was deciding rejected, its root unpinned, its config and store dropped, + * the terminated mark set — and none of them means much without the others. + * Interleaved with awaits, as they used to be, a failure part-way leaves states + * nothing recovers from. The sharpest: marked terminated while `vatConfig` + * survives (only `deleteVat` removes it; `cleanupTerminatedVat` sweeps + * `${vatId}.` keys, which never match `vatConfig.${vatId}`) reads as *active* + * again the moment cleanup drops the mark, and `KernelRouter`'s endpoint lookup + * kills the run loop over the disagreement. With no await between them, that + * state cannot arise. + * + * Killing the worker is deliberately not part of this. It can fail, and + * nothing here needs it to have succeeded — a vat being retired has a worker + * that is gone or going, and a store that says so is worth more than a store + * still waiting to find out. + * + * Calling it twice records nothing the second time. Two of the writes cannot + * be repeated: `deleteVat` fails on a vat whose subcluster mapping the first + * call removed, and a second `releaseVatRootPin` would unpin a root this vat + * no longer holds. + * + * @param vatId - The vat being retired. + * @param error - Why, for the rejections its subscribers are owed. + */ + #retireVat(vatId: VatId, error: Error): void { + // Ahead of the guard, and safe there because nothing below reads it: a vat + // the store already calls dead must not keep a handle the router would go + // on resolving. + this.#vats.delete(vatId); + // Reached from `performVatRestart` when the relaunch breaks the stream: + // `onCriticalFailure` retires the vat and `runVat` rethrows into the catch + // that retires it again. + if (this.#kernelStore.isVatTerminated(vatId)) { + return; + } + const failure = makeKernelError('VAT_TERMINATED', error.message); + // First, while the c-list this reads through is still there: subscribers are + // told rather than left waiting on a decider that no longer exists. + for (const kpid of this.#kernelStore.getPromisesByDecider(vatId)) { + this.#kernelQueue.resolvePromises(vatId, [[kpid, true, failure]]); + } + // Before `deleteVat`, which is fine either way, but the root is found + // through the c-list and this keeps the reads ahead of the deletes. + this.releaseVatRootPin(vatId); + this.#kernelStore.deleteVat(vatId); + // Last: the mark is what makes the vat eligible for + // `nextTerminatedVatCleanup`, which reclaims the c-list everything above + // needed, and which must not run against a vat still being written. + this.#kernelStore.markVatAsTerminated(vatId); + } + + /** + * Begin closing down a vat whose channel has broken. Detached deliberately: + * the stream's drain catch has nobody to await it, and the teardown settles + * its own failures rather than rejecting. + * + * @param vatId - The vat that failed. + * @param vat - Its handle, whose pending RPCs are owed a rejection. + * @param error - What broke, for those rejections. + */ + #startFailedVatTeardown(vatId: VatId, vat: VatHandle, error: Error): void { + this.#tearDownFailedVat(vatId, vat, error).catch((unexpected: unknown) => + this.#logger.error( + `Unexpected failure tearing down vat ${vatId}:`, + unexpected, + ), + ); + } + + /** + * Close down a vat whose channel has broken, after {@link #retireVat} has put + * its death on record. + * + * Recording the death only saves the deliveries that come after it. Any + * already in flight are parked on an RPC client with no timeout, so without + * this their cranks never finish either — the same hang, one delivery + * earlier. `terminate` rejects them, and the worker is stopped because + * nothing else will now that the handle is off the books. + * + * It also re-asserts the death, because the stream can break at any point in + * a crank and {@link #retireVat} writes into whichever one is open. A crank + * that aborts rolls those writes back — and the run loop carries on, since an + * abort is an outcome rather than a failure — while the handle this already + * deleted stays gone. That is the store-says-live, kernel-says-dead + * disagreement, reached by a route `#trackFlux` does not cover. + * + * Best-effort throughout: both steps work against a vat that is already gone, + * and there is nobody left to report to. + * + * @param vatId - The vat that failed. + * @param vat - Its handle, whose pending RPCs are owed a rejection. + * @param error - What broke, for those rejections. + */ + async #tearDownFailedVat( + vatId: VatId, + vat: VatHandle, + error: Error, + ): Promise { + // Started in this turn rather than after the kill below, because + // `beginOutOfCrank` takes the gate synchronously: from here the run loop + // cannot open a crank once the one in flight ends, so there is no window + // for a delivery to find the disagreement. It cannot be awaited before the + // rejections either — the crank in flight may be parked on one of them, and + // the gate waits for that crank. + const reasserted = this.#reassertDeathOutOfCrank(vatId, error); + await Promise.all([ + // `terminate` rejects the vat's pending RPCs before it awaits anything, + // so starting it first frees the parked delivery in this turn rather than + // behind a worker kill that may be slow to settle, or never settle. + vat.terminate(true, error).catch((terminateError: unknown) => { + this.#logger.error( + `Failed to close the channel of vat ${vatId} after a fatal error:`, + terminateError, + ); + }), + this.#platformServices + .terminate(vatId, error) + .catch((terminateError: unknown) => { + this.#logger.error( + `Failed to stop the worker of vat ${vatId} after a fatal error:`, + terminateError, + ); + }), + ]); + await reasserted; + } + + /** + * Write a failed vat's death again if the crank it was first written in + * rolled it back. Held out of crank, so the store's answer is final rather + * than one an abort can still undo. + * + * `isVatActive` reads the config row {@link #retireVat} deletes, so a death + * that committed is not rewritten, and neither is a vat whose cleanup has + * since run. + * + * @param vatId - The vat that failed. + * @param error - What broke, for the rejections a redo owes its subscribers. + */ + async #reassertDeathOutOfCrank(vatId: VatId, error: Error): Promise { try { - if (terminating) { - // A restart keeps the pin: the same root comes back. - this.releaseVatRootPin(vatId); + await this.#kernelStore.withStoreOutOfCrank(() => { + if (this.#kernelStore.isVatActive(vatId)) { + this.#retireVat(vatId, error); + } + }); + } catch (reassertError) { + this.#logger.error( + `Failed to record the death of vat ${vatId} again after its crank:`, + reassertError, + ); + // As in `onCriticalFailure`: the mark is the one write that cannot be + // skipped, since without it the store goes on calling the vat active + // while the kernel has no handle for it. + try { + this.#kernelStore.markVatAsTerminated(vatId); + } catch (markError) { + this.#logger.error( + `Vat ${vatId} could not be marked terminated; the store still calls it active and the kernel has no handle for it:`, + markError, + ); } - await this.#platformServices - .terminate(vatId, terminationError) - .catch(this.#logger.error); - await vat.terminate(terminating, terminationError); - } finally { - // A handle left behind outlives the store state the terminated-vat - // cleanup wipes, so `hasVat`, `getVatIds` and `#getEndpoint` go on - // reporting a vat with no c-list, and the next `bringOutYourDead` - // selected for it kills the run loop. - this.#vats.delete(vatId); } } @@ -240,24 +502,215 @@ export class VatManager { * @param reason - If the vat is being terminated, the reason for the termination. */ async terminateVat(vatId: VatId, reason?: CapData): Promise { - await this.#kernelQueue.waitForCrank(); - await this.stopVat(vatId, true, reason); - // Mark for deletion (which will happen later, in vat-cleanup events) - this.#kernelStore.markVatAsTerminated(vatId); + // A restart still queued for this vat is overtaken by the termination, and + // will be dropped when the run loop reaches it. Tell whoever asked for it + // now, rather than leaving them waiting on a request that can no longer be + // carried out. + const superseded = this.#restartWaiters.get(vatId); + this.#restartWaiters.delete(vatId); + superseded?.reject(new VatDeletedError(vatId)); + // Not queued for the run loop the way `restartVat` is: teardown has to work + // on a kernel whose run loop has died, which `reset` depends on. So this one + // closes its window with a flux record instead. + await this.#trackFlux(vatId, async () => this.stopVat(vatId, true, reason)); } /** * Restarts a vat. * + * Asks the run loop to do it, rather than doing it here. A restart keeps the + * vat's c-list while taking the vat itself out of the kernel's reach for as + * long as launching a worker and negotiating with it takes, and doing that + * alongside a running run loop means a crank can land in the window and read a + * live vat as a dead one. In a crank of its own there is no window: the run + * loop is the only thing that delivers, and it is here instead. + * + * A request that arrives while one is still *queued* takes over its item + * rather than adding one of its own: the item carries only the vat's ID, so + * two of them are two restarts, and the crank that ran the first would + * already have handed this caller a live handle. The leftover would then stop + * that worker and — if the relaunch failed — terminate the vat its caller was + * told about. A request that arrives while one is being *carried out* does + * queue an item of its own, which runs after the crank in flight: it has its + * own caller to answer, so it is not the leftover that hazard is about. + * * @param vatId - The ID of the vat. * @returns A promise for the restarted vat. */ async restartVat(vatId: VatId): Promise { - await this.#kernelQueue.waitForCrank(); - const vat = this.getVat(vatId); - const { config } = vat; - await this.stopVat(vatId, false); - await this.runVat(vatId, config); + // The store, not the handle: `performVatRestart` holds the vat between + // workers with no handle on the books for as long as launching one takes, + // and a request landing in that window is for a vat that is coming back. + // Rejected here rather than from inside a crank where it can be, so that + // the caller is not told by way of a dead run loop. + if (!this.#vats.has(vatId) && !this.#kernelStore.isVatActive(vatId)) { + throw new VatNotFoundError(vatId); + } + // Read before `#awaitRestart` replaces the waiter: an unconsumed waiter is + // how an item still queued for this vat makes itself known, since + // `performVatRestart` takes the waiter the moment it starts. + const alreadyQueued = this.#restartWaiters.has(vatId); + const restarted = this.#awaitRestart(vatId); + if (!alreadyQueued) { + try { + this.#kernelQueue.enqueueRestartVat(vatId); + } catch (error) { + // Nothing was queued, so nothing will carry the request out. Settling + // the waiter lets `#awaitRestart` unwind and stop watching the run + // loop; awaiting it keeps the rejection from going unhandled, since + // `enqueueRestartVat` refuses for the very reason that waiter watches + // for and this throw is what the caller sees. + this.#restartWaiters.get(vatId)?.reject(error); + this.#restartWaiters.delete(vatId); + await restarted.catch(() => undefined); + throw error; + } + } + await restarted; + return this.getVat(vatId); + } + + /** + * Replace a vat's worker. Called by the run loop, for a queued restart request. + * + * @param vatId - The ID of the vat. + */ + async performVatRestart(vatId: VatId): Promise { + const settle = this.#restartWaiters.get(vatId); + this.#restartWaiters.delete(vatId); + if (!this.#vats.has(vatId)) { + // The vat went away between the request and this crank. `terminateVat` + // does not go through the run queue, so it can land in that window, and a + // request for a vat that no longer exists has nothing to carry out and + // nothing to put right. Dropped rather than thrown: the alternative is a + // dead run loop over work that is merely obsolete. + const error = new VatNotFoundError(vatId); + this.#logger.error( + `Restart of vat ${vatId} dropped; the vat is gone:`, + error, + ); + settle?.reject(error); + return; + } + try { + // Read before the handle goes away, and from the handle rather than the + // store, so the incarnation that comes back is configured like the one + // that left. + const { config } = this.getVat(vatId); + await this.stopVat(vatId, false); + await this.runVat(vatId, config); + } catch (error) { + // The vat has no worker and is not coming back, so it is terminated in + // fact; record that so the rest of the kernel agrees. This must not throw + // out of the crank, and not only to keep the run loop alive: the run + // loop's catch rolls the crank back, which would undo the very records + // written here *and* restore this request to the run queue, so the next + // process start would replay the same failing restart forever. + this.#retireVat( + vatId, + error instanceof Error ? error : new Error(String(error)), + ); + this.#logger.error( + `Restart of vat ${vatId} failed; terminating it:`, + error, + ); + settle?.reject(error); + return; + } + settle?.resolve(); + } + + /** + * Wait for the run loop to carry out this vat's queued restart. + * + * Registered before the request is enqueued, so a crank cannot complete the + * restart before there is anything to tell. A request that outlives the kernel + * that queued it has no waiter when the new one gets to it, which is why + * settling is optional. + * + * @param vatId - The vat being restarted. + * @returns A promise that settles when the restart does. + */ + async #awaitRestart(vatId: VatId): Promise { + const { promise, resolve, reject } = makePromiseKit(); + // One waiter per vat: a second request for a vat already awaiting one would + // otherwise strand the first caller forever. + this.#restartWaiters + .get(vatId) + ?.reject(new Error(`Restart of vat ${vatId} superseded by a later one`)); + this.#restartWaiters.set(vatId, { resolve, reject }); + // The run loop is what carries the request out, and a loop that dies has no + // kernel promise for this the way a message result does, so nothing else + // would ever settle this caller. + const stopWatchingTheRunLoop = this.#kernelQueue.onRunLoopDeath(reject); + try { + return await promise; + } finally { + stopWatchingTheRunLoop(); + } + } + + /** + * Run an operation that takes a vat out of the kernel's reach, recording the + * vat as mid-flux for its duration so a delivery arriving meanwhile waits for + * the outcome instead of reading the vat as gone. + * + * @param vatId - The vat being taken out of reach. + * @param start - Begins the operation. Called once, while the store is held. + * @returns The operation's own result, failure included. + */ + async #trackFlux(vatId: VatId, start: () => Promise): Promise { + // Held out of crank rather than merely waiting for the crank in flight to + // end. The run loop starts its next crank in the same turn it ends the + // last, so a caller that only awaited `waitForCrank` resumed with that + // crank's savepoints already open: the vat's death written inside its + // delivery savepoint, for an unrelated rollback to undo while the handle + // stayed deleted, and that crank free to reach the vat before the record + // existed and be handed a handle to a worker about to be killed. + // + // Wrapped in an object because a bare promise is what `withStoreOutOfCrank` + // refuses: an async function adopts a returned promise, so the result would + // collapse to the teardown's own `undefined` and the destructure below + // would throw. + const { flux } = await this.#kernelStore.withStoreOutOfCrank(() => { + const started = start(); + // Recorded with nothing awaited since `start()`, so no crank can run + // between the vat's first step towards death and the record of it. + // + // Waiters see a plain completion rather than a failure, because "gone" + // is what they should act on and the handle is dropped before anything + // that can fail. The caller still gets the failure, from `flux` itself. + this.#vatsInFlux.set( + vatId, + started.catch(() => undefined), + ); + return { flux: started }; + }); + try { + return await flux; + } finally { + this.#vatsInFlux.delete(vatId); + } + } + + /** + * The handle for a vat, waiting first for any teardown in flight. The + * counterpart to {@link getVat} for callers that can afford to wait — a crank, + * above all, which would otherwise be told a vat is missing before the store + * records why. + * + * @param vatId - The ID of the vat. + * @returns A promise for the vat's handle. + * @throws If the vat does not exist, or stopped existing while being awaited. + */ + async provideVat(vatId: VatId): Promise { + const flux = this.#vatsInFlux.get(vatId); + if (flux) { + // Only a teardown is ever recorded, so waiting it out settles the vat's + // fate: it is gone, and the store now says so. + await flux; + throw new VatNotFoundError(vatId); + } return this.getVat(vatId); } diff --git a/packages/ocap-kernel/src/vats/vat-death-and-cranks.test.ts b/packages/ocap-kernel/src/vats/vat-death-and-cranks.test.ts new file mode 100644 index 000000000..b5ca5f211 --- /dev/null +++ b/packages/ocap-kernel/src/vats/vat-death-and-cranks.test.ts @@ -0,0 +1,234 @@ +import { makeSQLKernelDatabase } from '@metamask/kernel-store/sqlite/nodejs'; +import type { JsonRpcMessage } from '@metamask/kernel-utils'; +import { Logger } from '@metamask/logger'; +import type { DuplexStream } from '@metamask/streams'; +import { describe, it, expect, vi } from 'vitest'; + +import { VatHandle } from './VatHandle.ts'; +import { VatManager } from './VatManager.ts'; +import { KernelQueue } from '../KernelQueue.ts'; +import { makeKernelStore } from '../store/index.ts'; +import type { + CrankResult, + PlatformServices, + RunQueueItem, + VatConfig, + VatId, +} from '../types.ts'; + +/** + * Where a vat's death meets the run loop, over a real store. + * + * `VatManager`'s own tests mock the store, so the turn `terminateVat` resumes + * in is whatever the mock chooses — and that turn is the whole question here. + */ + +const config: VatConfig = { sourceSpec: 'test.js' }; + +/** + * A kernel store, run loop and vat manager over one in-memory database. + * + * @returns The pieces, plus a `deliver` the test drives the run loop with. + */ +async function makeFixture(): Promise<{ + kernelStore: ReturnType; + kernelQueue: KernelQueue; + vatManager: VatManager; + deliveries: ((result: CrankResult) => void)[]; + deliver: (item: RunQueueItem) => Promise; + streamDeaths: ((error: Error, vat: VatHandle) => void)[]; +}> { + const kdb = await makeSQLKernelDatabase({ dbFilename: ':memory:' }); + const kernelStore = makeKernelStore(kdb); + const platformServices = { + launch: vi.fn().mockResolvedValue({ + end: vi.fn(), + } as unknown as DuplexStream), + terminate: vi.fn().mockResolvedValue(undefined), + terminateAll: vi.fn().mockResolvedValue(undefined), + } as unknown as PlatformServices; + + // Captured per fixture rather than read off the shared spy's `mock.calls`, + // which accumulate across tests. + const streamDeaths: ((error: Error, vat: VatHandle) => void)[] = []; + vi.spyOn(VatHandle, 'make').mockImplementation( + async ({ vatId, vatConfig, onCriticalFailure }) => { + streamDeaths.push(onCriticalFailure); + return { + vatId, + config: vatConfig, + terminate: vi.fn().mockResolvedValue(undefined), + ping: vi.fn(), + } as unknown as VatHandle; + }, + ); + + // eslint-disable-next-line prefer-const + let vatManager: VatManager; + const kernelQueue = new KernelQueue(kernelStore, async (vatId, reason) => + vatManager.stopVat(vatId, true, reason), + ); + vatManager = new VatManager({ + platformServices, + kernelStore, + kernelQueue, + logger: new Logger('test'), + }); + + // Each delivery parks until the test settles it, which is how the test gets + // to act while a crank is open. + const deliveries: ((result: CrankResult) => void)[] = []; + const deliver = async (): Promise => + new Promise((resolve) => { + deliveries.push(resolve); + }); + + return { + kernelStore, + kernelQueue, + vatManager, + deliveries, + deliver, + streamDeaths, + }; +} + +/** + * Let every pending microtask and timer callback run. + * + * @returns A promise that resolves once they have. + */ +const settle = async (): Promise => + new Promise((resolve) => { + setTimeout(resolve, 0); + }); + +/** + * Wait until the run loop has parked on a delivery. + * + * @param deliveries - The resolvers collected so far. + * @param count - How many deliveries to wait for. + */ +async function deliveriesReach( + deliveries: unknown[], + count: number, +): Promise { + for (let tries = 0; tries < 50 && deliveries.length < count; tries += 1) { + await settle(); + } + expect(deliveries).toHaveLength(count); +} + +describe("a vat's death while the run loop is running", () => { + it('survives a later crank being rolled back', async () => { + const { kernelStore, kernelQueue, vatManager, deliveries, deliver } = + await makeFixture(); + // Through `launchVat` so the store holds everything a real termination + // reads: the config row, the subcluster mapping, the pinned root. + const subclusterId = kernelStore.addSubcluster({ + bootstrap: 'bob', + vats: { bob: { sourceSpec: 'test.js' } }, + }); + await vatManager.launchVat(config, 'bob', subclusterId); + + kernelStore.enqueueRun({ + type: 'send', + target: 'ko1', + message: { methargs: { body: '#[]', slots: [] } }, + } as unknown as RunQueueItem); + kernelStore.enqueueRun({ + type: 'send', + target: 'ko2', + message: { methargs: { body: '#[]', slots: [] } }, + } as unknown as RunQueueItem); + + const loop = kernelQueue.run(deliver); + loop.catch(() => undefined); + await deliveriesReach(deliveries, 1); + + // Asked for mid-crank, so it has to wait out the crank in flight. + const terminated = vatManager.terminateVat('v1' as VatId); + await settle(); + deliveries[0]?.({ didDelivery: 'v1' }); + await terminated; + + // `isVatTerminated` is not the signal: cleanup unmarks the vat it finishes + // with. The config row `deleteVat` removed is what stays removed. + expect({ + where: 'after terminate', + active: kernelStore.isVatActive('v1' as VatId), + hasVat: vatManager.hasVat('v1' as VatId), + }).toStrictEqual({ + where: 'after terminate', + active: false, + hasVat: false, + }); + + // The next crank aborts, and has nothing to do with this vat. + await deliveriesReach(deliveries, 2); + deliveries[1]?.({ abort: true }); + await settle(); + + expect({ + where: 'after the unrelated crank aborted', + active: kernelStore.isVatActive('v1' as VatId), + hasVat: vatManager.hasVat('v1' as VatId), + }).toStrictEqual({ + where: 'after the unrelated crank aborted', + active: false, + hasVat: false, + }); + }); + + // `terminateVat` above records the death out of crank. A broken stream does + // not: `onCriticalFailure` writes into whichever crank is open, and an abort + // rolls those writes back while the handle it deleted stays deleted. + it('survives the crank it died in being rolled back', async () => { + const { + kernelStore, + kernelQueue, + vatManager, + deliveries, + deliver, + streamDeaths, + } = await makeFixture(); + const subclusterId = kernelStore.addSubcluster({ + bootstrap: 'bob', + vats: { bob: { sourceSpec: 'test.js' } }, + }); + await vatManager.launchVat(config, 'bob', subclusterId); + const vat = vatManager.getVat('v1' as VatId); + const streamDied = streamDeaths[0] as ( + error: Error, + vat: VatHandle, + ) => void; + + kernelStore.enqueueRun({ + type: 'send', + target: 'ko1', + message: { methargs: { body: '#[]', slots: [] } }, + } as unknown as RunQueueItem); + + const loop = kernelQueue.run(deliver); + loop.catch(() => undefined); + await deliveriesReach(deliveries, 1); + + // The drain catch fires in whatever turn the read error lands in, which is + // routinely one with a crank open. + streamDied(new Error('the worker went away'), vat); + // Aborting for a reason of its own, so nothing re-records the death the way + // a `terminate` result would. + deliveries[0]?.({ abort: true }); + await settle(); + + expect({ + where: 'after the crank it died in aborted', + active: kernelStore.isVatActive('v1' as VatId), + hasVat: vatManager.hasVat('v1' as VatId), + }).toStrictEqual({ + where: 'after the crank it died in aborted', + active: false, + hasVat: false, + }); + }); +});