diff --git a/packages/kernel-test/src/crank-rollback.test.ts b/packages/kernel-test/src/crank-rollback.test.ts index 36131a8a1..5b11f2f0c 100644 --- a/packages/kernel-test/src/crank-rollback.test.ts +++ b/packages/kernel-test/src/crank-rollback.test.ts @@ -238,6 +238,33 @@ 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. diff --git a/packages/kernel-test/src/garbage-collection.test.ts b/packages/kernel-test/src/garbage-collection.test.ts index 44345bfcd..33a78638c 100644 --- a/packages/kernel-test/src/garbage-collection.test.ts +++ b/packages/kernel-test/src/garbage-collection.test.ts @@ -257,26 +257,39 @@ describe('Garbage Collection', () => { * Give an importer a chance to notice a dropped object and tell the kernel, * then keep cranking until the resulting GC actions have all been consumed. * + * 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 done - The outcome being waited for. */ - async function reapAndSettle(vatId: VatId, rootKRef: KRef): Promise { - 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. + async function reapAndSettle( + vatId: VatId, + rootKRef: KRef, + done: () => boolean, + ): Promise { 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) { + if ([...kernelStore.getGCActions()].length === 0 && done()) { return; } } throw Error( - `GC actions still pending after ${maxRounds} rounds: ${[ - ...kernelStore.getGCActions(), - ].join(', ')}`, + `GC did not settle after ${maxRounds} rounds; actions pending: ${ + [...kernelStore.getGCActions()].join(', ') || '(none)' + }`, ); } @@ -311,7 +324,11 @@ describe('Garbage Collection', () => { await kernel.queueMessage(importerKRef, 'makeWeak', [objectId]); await kernel.queueMessage(importerKRef, 'forgetImport', []); await waitUntilQuiescent(); - await reapAndSettle(importerVatId, importerKRef); + await reapAndSettle( + importerVatId, + importerKRef, + () => !kernelStore.getImporters(sharedKRef).includes(importerVatId), + ); // The exporter must not have been told to drop it: the second importer // legitimately still holds it @@ -344,7 +361,11 @@ describe('Garbage Collection', () => { await kernel.queueMessage(secondImporterKRef, 'makeWeak', [objectId]); await kernel.queueMessage(secondImporterKRef, 'forgetImport', []); await waitUntilQuiescent(); - await reapAndSettle(secondImporterVatId, secondImporterKRef); + await reapAndSettle( + secondImporterVatId, + secondImporterKRef, + () => kernelStore.getImporters(sharedKRef).length === 0, + ); expect(kernelStore.getImporters(sharedKRef)).toStrictEqual([]); // Only the createObject result's stored value still names it diff --git a/packages/ocap-kernel/CHANGELOG.md b/packages/ocap-kernel/CHANGELOG.md index 86794f910..dcbc46ba3 100644 --- a/packages/ocap-kernel/CHANGELOG.md +++ b/packages/ocap-kernel/CHANGELOG.md @@ -77,6 +77,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Forget a crank's savepoints when the store call that ends them fails, in both `rollbackCrank` and `endCrank` ([#1021](https://github.com/MetaMask/ocap-kernel/pull/1021)) - A failed rollback or release discards the whole transaction, savepoints included. Keeping them listed had `endCrank` throw `No such savepoint: t0` from the run loop's `finally`, over whatever really killed the kernel - `rollbackCrank` now also reverts the two pieces of state a database rollback cannot reach, whether or not the rollback itself succeeded: every cached stored value is re-read, and the crank's accumulated garbage-collection candidates are discarded ([#1021](https://github.com/MetaMask/ocap-kernel/pull/1021)) + - The candidate set is restored to its state at the savepoint rather than emptied, since candidates accrued while the run loop was idle are owed a collection an unrelated crank's rollback must not cancel ([#1023](https://github.com/MetaMask/ocap-kernel/pull/1023)) - A cached stored value answers reads from a closure and only writes through to the database, so reverting the database left it holding the abandoned crank's value and the next write persisted that. `processGCActionSet` takes an action out of the set before delivering it, so an aborted delivery lost the action outright instead of retrying it - `maybeFreeKrefs` lives in RAM, so nothing reverted it either. Its entries are collection candidates only because of decrements the rollback undid, and a later `collectGarbage` threw outright on a promise the rollback had deleted, killing the run loop - A failed rollback discards the whole transaction, moving the database back at least as far as a successful rollback would have — so reverting only on success left exactly the state that is least able to tolerate it @@ -95,6 +96,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Renames `krefsToExistingErefs` to `krefsToErefs`, which now throws on an unmapped kref instead of silently dropping it - 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/MetaMask/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/MetaMask/ocap-kernel/pull/1022)) @@ -105,6 +107,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - 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 the crank fails there instead of committing a release the returning incarnation would disagree with - 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/MetaMask/ocap-kernel/pull/1022)) - Tear down and mark for cleanup a vat whose worker launched but whose kernel-side registration then failed ([#1022](https://github.com/MetaMask/ocap-kernel/pull/1022)) +- 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/MetaMask/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/MetaMask/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 +- 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/MetaMask/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/MetaMask/ocap-kernel/pull/1023)) +- 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/MetaMask/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/MetaMask/ocap-kernel/pull/1023)) +- `getImporters` now counts remotes, so retiring an object queues a `retireImport` for a remote importer rather than deleting the object and leaving the remote's c-list entry naming nothing ([#1015](https://github.com/MetaMask/ocap-kernel/issues/1015)) +- 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/MetaMask/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 d85cb0223..2b3d38123 100644 --- a/packages/ocap-kernel/src/Kernel.test.ts +++ b/packages/ocap-kernel/src/Kernel.test.ts @@ -31,13 +31,23 @@ 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); + }); /** * Fail the run loop, in the order the real `KernelQueue.run` does: the diff --git a/packages/ocap-kernel/src/Kernel.ts b/packages/ocap-kernel/src/Kernel.ts index f615ea6cb..5cd4076ad 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 9e4de322e..786a25216 100644 --- a/packages/ocap-kernel/src/KernelQueue.test.ts +++ b/packages/ocap-kernel/src/KernelQueue.test.ts @@ -1224,6 +1224,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 f4e1eff9f..6ca2d153f 100644 --- a/packages/ocap-kernel/src/KernelQueue.ts +++ b/packages/ocap-kernel/src/KernelQueue.ts @@ -501,6 +501,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.test.ts b/packages/ocap-kernel/src/KernelRouter.test.ts index e88c54b52..0847f0da5 100644 --- a/packages/ocap-kernel/src/KernelRouter.test.ts +++ b/packages/ocap-kernel/src/KernelRouter.test.ts @@ -12,6 +12,7 @@ import type { RunQueueItemGCAction, RunQueueItemBringOutYourDead, EndpointId, + VatId, GCRunQueueType, CrankResult, EndpointHandle, @@ -21,8 +22,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(() => { @@ -68,6 +72,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; @@ -77,6 +82,7 @@ describe('KernelRouter', () => { } as unknown as KernelQueue; const mockInvokeKernelService = vi.fn(); + restartVat = vi.fn().mockResolvedValue(undefined); // Create the router to test kernelRouter = new KernelRouter( @@ -84,6 +90,7 @@ describe('KernelRouter', () => { kernelQueue, getEndpoint, mockInvokeKernelService, + restartVat, ); }); @@ -352,6 +359,48 @@ describe('KernelRouter', () => { ); }); + // 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 +492,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 +599,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'; @@ -789,6 +899,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'); @@ -925,6 +1067,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 = { @@ -948,6 +1133,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 5117905f3..a5b5a11c8 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}`; @@ -235,14 +248,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, [ @@ -389,6 +403,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 +438,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,9 +502,36 @@ 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), + const stillHeld = (): KRef[] => + krefs.filter((kref) => this.#kernelStore.hasCListEntry(endpointId, kref)); + if (stillHeld().length === 0) { + 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. 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 that is absent without + // being 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(); 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`, @@ -447,39 +540,6 @@ export class KernelRouter { if (live.length === 0) { 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. Fail the crank rather than commit that. - // Nothing here can make the restart safe: the action is already spent from - // the durable set, and a crank that neither delivers nor releases would - // simply be handed the same action again on the next one. - if ( - isVatId(endpointId) && - !this.#kernelStore.isVatTerminated(endpointId) - ) { - throw error; - } - // 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, - ); - } const erefs = this.#kernelStore.krefsToErefs(endpointId, live); // Telling an endpoint to let go is also the kernel letting go. Otherwise a // dropped export stays flagged reachable, so the same action gets derived @@ -560,8 +620,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/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.test.ts b/packages/ocap-kernel/src/store/index.test.ts index d2fb46efe..3dd0e677f 100644 --- a/packages/ocap-kernel/src/store/index.test.ts +++ b/packages/ocap-kernel/src/store/index.test.ts @@ -116,6 +116,7 @@ describe('kernel store', () => { 'getRelayEntries', 'getRemoteIdentityValue', 'getRemoteIdentityValueRequired', + 'getRemoteIds', 'getRemoteInfo', 'getRemoteSeqState', 'getRootObject', 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 5ce0dc12b..5f4bb3a29 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'; @@ -371,4 +372,42 @@ 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([]); + }); + }); }); diff --git a/packages/ocap-kernel/src/store/methods/crank.test.ts b/packages/ocap-kernel/src/store/methods/crank.test.ts index 426890d20..f19a6ee38 100644 --- a/packages/ocap-kernel/src/store/methods/crank.test.ts +++ b/packages/ocap-kernel/src/store/methods/crank.test.ts @@ -2,7 +2,18 @@ import type { KernelDatabase } from '@metamask/kernel-store'; import { expect, describe, it, vi, beforeEach } from 'vitest'; import { getCrankMethods } from './crank.ts'; -import type { StoreContext } from '../types.ts'; +import type { KRef } from '../../types.ts'; +import type { Savepoint, StoreContext } from '../types.ts'; + +/** + * Build savepoint records holding no collection candidates, for tests that only + * care which savepoints are listed. + * + * @param names - The savepoint names, in order. + * @returns The savepoint records. + */ +const savepoints = (...names: string[]): Savepoint[] => + names.map((name) => ({ name, maybeFreeKrefs: new Set() })); describe('crank methods', () => { let context: StoreContext; @@ -10,6 +21,12 @@ describe('crank methods', () => { let crankMethods: ReturnType; let mockCrankBuffer: unknown[]; + /** + * @returns The names of the currently listed savepoints, in order. + */ + const savepointNames = (): string[] => + context.savepoints.map(({ name }) => name); + beforeEach(() => { mockCrankBuffer = []; context = { @@ -53,7 +70,7 @@ describe('crank methods', () => { context.inCrank = true; crankMethods.createCrankSavepoint('test'); - expect(context.savepoints).toStrictEqual(['test']); + expect(savepointNames()).toStrictEqual(['test']); expect(kdb.createSavepoint).toHaveBeenCalledWith('t0'); }); @@ -62,7 +79,7 @@ describe('crank methods', () => { crankMethods.createCrankSavepoint('first'); crankMethods.createCrankSavepoint('second'); - expect(context.savepoints).toStrictEqual(['first', 'second']); + expect(savepointNames()).toStrictEqual(['first', 'second']); expect(kdb.createSavepoint).toHaveBeenCalledWith('t0'); expect(kdb.createSavepoint).toHaveBeenCalledWith('t1'); }); @@ -94,7 +111,7 @@ describe('crank methods', () => { describe('rollbackCrank', () => { it('forgets the savepoint even if the database rollback fails', () => { context.inCrank = true; - context.savepoints = ['start']; + context.savepoints = savepoints('start'); vi.mocked(kdb.rollbackSavepoint).mockImplementationOnce(() => { throw new Error('database is gone'); }); @@ -112,17 +129,17 @@ describe('crank methods', () => { it('should rollback to specified savepoint', () => { context.inCrank = true; - context.savepoints = ['first', 'second', 'third']; + context.savepoints = savepoints('first', 'second', 'third'); crankMethods.rollbackCrank('second'); expect(kdb.rollbackSavepoint).toHaveBeenCalledWith('t1'); - expect(context.savepoints).toStrictEqual(['first']); + expect(savepointNames()).toStrictEqual(['first']); }); it('should throw when savepoint does not exist', () => { context.inCrank = true; - context.savepoints = ['first', 'second']; + context.savepoints = savepoints('first', 'second'); expect(() => crankMethods.rollbackCrank('nonexistent')).toThrow( 'no such savepoint as ""nonexistent""', @@ -143,12 +160,12 @@ describe('crank methods', () => { crankMethods.rollbackCrank('b'); crankMethods.createCrankSavepoint('b2'); expect(kdb.createSavepoint).toHaveBeenLastCalledWith('t1'); - expect(context.savepoints).toStrictEqual(['a', 'b2']); + expect(savepointNames()).toStrictEqual(['a', 'b2']); }); it('clears the crank buffer', () => { context.inCrank = true; - context.savepoints = ['start']; + context.savepoints = savepoints('start'); mockCrankBuffer.push({ type: 'send' }, { type: 'notify' }); crankMethods.rollbackCrank('start'); @@ -184,9 +201,14 @@ describe('crank methods', () => { // at least as stale. it('reverts the caches the database cannot reach even when the rollback fails', () => { context.inCrank = true; + // Predates the savepoint, so it survives: only `collectGarbage` empties + // this set, and a candidate owed a collection before this crank began is + // still owed one after it is abandoned. context.maybeFreeKrefs.add('kp1'); crankMethods.createCrankSavepoint('crank'); crankMethods.createCrankSavepoint('delivery'); + // Added by the crank being rolled back, so it goes. + context.maybeFreeKrefs.add('kp2'); vi.mocked(kdb.rollbackSavepoint).mockImplementationOnce(() => { throw new Error('disk I/O error'); }); @@ -198,7 +220,7 @@ describe('crank methods', () => { expect(context.refreshCachedValues).toHaveBeenCalled(); expect(context.refreshRunQueue).toHaveBeenCalled(); expect(context.runQueueLengthCache).toBe(-1); - expect([...context.maybeFreeKrefs]).toStrictEqual([]); + expect([...context.maybeFreeKrefs]).toStrictEqual(['kp1']); }); // Reverting must not become a way to lose the database error either. @@ -233,7 +255,7 @@ describe('crank methods', () => { it('should release savepoints if they exist', () => { context.inCrank = true; - context.savepoints = ['test']; + context.savepoints = savepoints('test'); crankMethods.endCrank(); expect(kdb.releaseSavepoint).toHaveBeenCalledWith('t0'); expect(context.savepoints).toStrictEqual([]); @@ -261,7 +283,7 @@ describe('crank methods', () => { it('settles the crank even if releasing savepoints fails', async () => { crankMethods.startCrank(); - context.savepoints = ['test']; + context.savepoints = savepoints('test'); const waiter = crankMethods.waitForCrank(); vi.mocked(kdb.releaseSavepoint).mockImplementationOnce(() => { throw new Error('database is gone'); @@ -291,7 +313,7 @@ describe('crank methods', () => { describe('releaseAllSavepoints', () => { it('should release all savepoints', () => { context.inCrank = true; - context.savepoints = ['test']; + context.savepoints = savepoints('test'); crankMethods.releaseAllSavepoints(); expect(kdb.releaseSavepoint).toHaveBeenCalledWith('t0'); expect(context.savepoints).toStrictEqual([]); diff --git a/packages/ocap-kernel/src/store/methods/crank.ts b/packages/ocap-kernel/src/store/methods/crank.ts index a3bba1956..0be4cf306 100644 --- a/packages/ocap-kernel/src/store/methods/crank.ts +++ b/packages/ocap-kernel/src/store/methods/crank.ts @@ -2,7 +2,7 @@ import { Fail, q } from '@endo/errors'; import { makePromiseKit } from '@endo/promise-kit'; import type { KernelDatabase } from '@metamask/kernel-store'; -import type { CrankBufferItem, StoreContext } from '../types.ts'; +import type { CrankBufferItem, Savepoint, StoreContext } from '../types.ts'; /** * Get the crank methods. @@ -36,7 +36,12 @@ export function getCrankMethods(ctx: StoreContext, kdb: KernelDatabase) { // first would leave `endCrank` trying to release a savepoint that was never // created, and that error would replace whatever really went wrong. kdb.createSavepoint(`t${ordinal}`); - ctx.savepoints.push(name); + // Copied, not referenced: `maybeFreeKrefs` is mutated in place from here on, + // and this is the "before" a rollback restores. + ctx.savepoints.push({ + name, + maybeFreeKrefs: new Set(ctx.maybeFreeKrefs), + }); } /** @@ -48,7 +53,8 @@ export function getCrankMethods(ctx: StoreContext, kdb: KernelDatabase) { ctx.inCrank || Fail`rollbackCrank outside of crank`; ctx.crankBuffer.length = 0; // Discard buffered outputs for (const ordinal of ctx.savepoints.keys()) { - if (ctx.savepoints[ordinal] === savepoint) { + const restored = ctx.savepoints[ordinal]; + if (restored?.name === savepoint) { try { kdb.rollbackSavepoint(`t${ordinal}`); // Left listed, `endCrank`'s release would commit the crank we just @@ -66,10 +72,10 @@ export function getCrankMethods(ctx: StoreContext, kdb: KernelDatabase) { // and these caches are at least as stale. Rethrowing ahead of this // would leave the dying crank holding the GC action it consumed and // the freed krefs it was about to collect. - revertStateBeneathRollback(error); + revertStateBeneathRollback(restored, error); throw error; } - revertStateBeneathRollback(); + revertStateBeneathRollback(restored); return; } } @@ -80,11 +86,16 @@ export function getCrankMethods(ctx: StoreContext, kdb: KernelDatabase) { * Revert what a database rollback cannot reach: the in-memory caches built * over the abandoned crank's writes. * + * @param restored - The savepoint being rolled back to, whose snapshot of + * `maybeFreeKrefs` is the "before" this restores. * @param rollbackError - The error the rollback threw, if it threw. Kept as * the `cause` should reverting fail too, since it is the root cause an * operator needs. */ - function revertStateBeneathRollback(rollbackError?: unknown): void { + function revertStateBeneathRollback( + restored: Savepoint, + rollbackError?: unknown, + ): void { try { // Recreate the run queue so its cached head/tail are re-read from the // database, and invalidate the length cache, since the rollback may have @@ -97,13 +108,19 @@ export function getCrankMethods(ctx: StoreContext, kdb: KernelDatabase) { // action out of the set before delivering it, so an action not restored // here is lost rather than retried. ctx.refreshCachedValues(); - // Nothing rolls back RAM. These krefs are collection candidates only - // because this crank decremented them, and that is precisely what was just - // undone. Left in place, `collectGarbage` throws on a later crank for any - // promise this one created — killing the run loop over work that no longer - // exists. Correct only while every rollback discards the whole delivery, - // which is all any caller asks for. + // Nothing rolls back RAM. Krefs this crank added are collection + // candidates only because of decrements that were just undone; left in + // 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. ctx.maybeFreeKrefs.clear(); + for (const kref of restored.maybeFreeKrefs) { + ctx.maybeFreeKrefs.add(kref); + } } catch (revertError) { if (rollbackError === undefined) { throw revertError; diff --git a/packages/ocap-kernel/src/store/methods/remote.ts b/packages/ocap-kernel/src/store/methods/remote.ts index 23ae73aec..a32c3d078 100644 --- a/packages/ocap-kernel/src/store/methods/remote.ts +++ b/packages/ocap-kernel/src/store/methods/remote.ts @@ -47,6 +47,17 @@ export function getRemoteMethods(ctx: StoreContext) { } } + /** + * The IDs of every remote the kernel knows about, without reading their info. + * + * @returns The remote IDs. + */ + function getRemoteIds(): RemoteId[] { + return Array.from(getPrefixedKeys(REMOTE_INFO_BASE)).map( + (remoteKey) => remoteKey.slice(REMOTE_INFO_BASE_LEN) as RemoteId, + ); + } + /** * Fetch the stored info about a remote. * @@ -299,6 +310,7 @@ export function getRemoteMethods(ctx: StoreContext) { return { getAllRemoteRecords, + getRemoteIds, getRemoteInfo, setRemoteInfo, deleteRemoteInfo, diff --git a/packages/ocap-kernel/src/store/methods/vat.ts b/packages/ocap-kernel/src/store/methods/vat.ts index c1875d858..bd4497dc5 100644 --- a/packages/ocap-kernel/src/store/methods/vat.ts +++ b/packages/ocap-kernel/src/store/methods/vat.ts @@ -5,6 +5,7 @@ import { getCListMethods } from './clist.ts'; import { getObjectMethods } from './object.ts'; import { getPromiseMethods } from './promise.ts'; import { getReachableMethods } from './reachable.ts'; +import { getRemoteMethods } from './remote.ts'; import type { EndpointId, KRef, @@ -42,6 +43,7 @@ export function getVatMethods(ctx: StoreContext) { getPromiseMethods(ctx); const { initKernelObject } = getObjectMethods(ctx); const { addCListEntry } = getCListMethods(ctx); + const { getRemoteIds } = getRemoteMethods(ctx); /** * Delete all persistent state associated with an endpoint. @@ -123,14 +125,17 @@ export function getVatMethods(ctx: StoreContext) { } /** - * Checks if a vat imports the specified kernel slot. + * Checks if an endpoint imports the specified kernel slot. * - * @param vatID - The ID of the vat to check. + * @param endpointId - The ID of the vat or remote to check. * @param kernelSlot - The kernel slot reference. - * @returns True if the vat imports the kernel slot, false otherwise. + * @returns True if the endpoint imports the kernel slot, false otherwise. */ - function importsKernelSlot(vatID: VatId, kernelSlot: KRef): boolean { - const data = ctx.kv.get(getSlotKey(vatID, kernelSlot)); + function importsKernelSlot( + endpointId: EndpointId, + kernelSlot: KRef, + ): boolean { + const data = ctx.kv.get(getSlotKey(endpointId, kernelSlot)); if (data) { const { vatSlot } = parseReachableAndVatSlot(data); const { direction } = parseRef(vatSlot); @@ -142,15 +147,19 @@ export function getVatMethods(ctx: StoreContext) { } /** - * Gets all vats that import a specific kernel object. + * Gets all endpoints that import a specific kernel object. + * + * Remotes count. `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. * * @param koid - The kernel object ID. - * @returns An array of vat IDs that import the kernel object. + * @returns An array of endpoint IDs that import the kernel object. */ - function getImporters(koid: KRef): VatId[] { - const importers = []; - importers.push( - ...getVatIDs().filter((vatID) => importsKernelSlot(vatID, koid)), + function getImporters(koid: KRef): EndpointId[] { + const importers: EndpointId[] = [...getVatIDs(), ...getRemoteIds()].filter( + (endpointId) => importsKernelSlot(endpointId, koid), ); importers.sort(); return importers; diff --git a/packages/ocap-kernel/src/store/types.ts b/packages/ocap-kernel/src/store/types.ts index b9886c174..10bacd658 100644 --- a/packages/ocap-kernel/src/store/types.ts +++ b/packages/ocap-kernel/src/store/types.ts @@ -23,7 +23,7 @@ export type StoreContext = { inCrank: boolean; crankSettled?: Promise; resolveCrank?: (() => void) | undefined; - savepoints: string[]; + savepoints: Savepoint[]; crankBuffer: CrankBufferItem[]; // Buffer for sends and notifications during crank subclusters: StoredValue; // Holds Subcluster[] nextSubclusterId: StoredValue; // Holds string @@ -32,6 +32,17 @@ export type StoreContext = { logger?: Logger | undefined; }; +/** + * A database savepoint, paired with the RAM state a database rollback cannot + * reach. `maybeFreeKrefs` is the collection-candidate set as it stood when the + * savepoint was taken, so a rollback can put back exactly what the abandoned + * work added and no more. + */ +export type Savepoint = { + name: string; + maybeFreeKrefs: Set; +}; + export type StoredValue = { get(): string | undefined; set(newValue: string): void; 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 e437d042b..f8aaf6ad5 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; @@ -35,7 +46,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); @@ -70,6 +83,8 @@ describe('VatManager', () => { ), getVatSubcluster: vi.fn().mockReturnValue('s1'), markVatAsTerminated: vi.fn(), + deleteVat: vi.fn(), + getPromisesByDecider: vi.fn().mockReturnValue([]), getRootObject: vi.fn().mockReturnValue('ko1'), pinObject: vi.fn(), unpinObject: vi.fn(), @@ -80,6 +95,16 @@ describe('VatManager', () => { 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); + }), } as unknown as Mocked; mockLogger = new Logger('test'); @@ -394,6 +419,205 @@ describe('VatManager', () => { expect.objectContaining({ message: 'Vat termination: Custom reason' }), ); }); + + it('waits out the crank in flight before recording the vat as in flux', async () => { + await vatManager.runVat('v1', createMockVatConfig()); + let finishCrank!: () => void; + ( + mockKernelQueue.waitForCrank as unknown as MockInstance + ).mockReturnValueOnce( + new Promise((resolve) => { + finishCrank = resolve; + }), + ); + + const terminated = vatManager.terminateVat('v1'); + await drainMicrotasks(); + + // Recording first and waiting after would deadlock, and this is the + // assertion that catches it: a crank already running reaches its endpoint + // lookup here, and would find a record whose teardown is waiting for that + // same crank to end. Reverse the order in `#trackFlux` and this hangs. + expect(await vatManager.provideVat('v1')).toBe(vatHandles[0]); + + finishCrank(); + await terminated; + + expect(mockKernelStore.markVatAsTerminated).toHaveBeenCalledWith('v1'); + }); + }); + + 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'); + }); + + 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', () => { @@ -404,7 +628,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); @@ -418,6 +642,178 @@ 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('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]); + }); + }); + + 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 0870a9dac..d7381b7b6 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,36 @@ 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; + this.#retireVat(vatId, error); + this.#startFailedVatTeardown(vatId, failedVat, error); + }, logger: vatLogger, allowedGlobalNames: this.#allowedGlobalNames, }); + if (fatalError) { + throw fatalError; + } this.#vats.set(vatId, vat); } @@ -208,7 +271,14 @@ 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}`); @@ -216,14 +286,123 @@ export class VatManager { terminationError = new VatDeletedError(vatId); } if (terminating) { - // A restart keeps the pin: the same root comes back. - this.releaseVatRootPin(vatId); + // 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); + 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. + * + * @param vatId - The vat being retired. + * @param error - Why, for the rejections its subscribers are owed. + */ + #retireVat(vatId: VatId, error: Error): void { + 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]]); + } this.#vats.delete(vatId); + // 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. + * + * Never rejects: both steps are best-effort 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 { + 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, + ); + }), + ]); } /** @@ -233,24 +412,182 @@ 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. + * * @param vatId - The ID of the vat. * @returns A promise for the restarted vat. */ async restartVat(vatId: VatId): Promise { + // Rejects an unknown vat here rather than from inside a crank, where the + // caller could only be told by way of a dead run loop. + this.getVat(vatId); + const restarted = this.#awaitRestart(vatId); + try { + this.#kernelQueue.enqueueRestartVat(vatId); + } catch (error) { + // Nothing was queued, so nothing will ever settle the waiter just + // registered. Take it back out: left behind, the next request for this vat + // would reject it as superseded, and since this caller never got as far as + // awaiting it that rejection would go unhandled. + this.#restartWaiters.delete(vatId); + 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 }); + return await promise; + } + + /** + * 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. + * + * Both steps live here, in this order, because the order is the whole + * mechanism and reversing it deadlocks. See the comments inline; a caller + * cannot get it wrong because a caller does not sequence it. + * + * @param vatId - The vat being taken out of reach. + * @param start - Begins the operation. Called once, after the wait. + * @returns The operation's own result, failure included. + */ + async #trackFlux(vatId: VatId, start: () => Promise): Promise { + // First: wait out the crank in flight, so the operation does not pull a + // worker out from under a delivery. This has to happen *before* the record + // exists. A crank that is already running has not necessarily reached its + // endpoint lookup yet, so if the record were there it would find it and wait + // for this operation — which is waiting for that crank to end. await this.#kernelQueue.waitForCrank(); - const vat = this.getVat(vatId); - const { config } = vat; - await this.stopVat(vatId, false); - await this.runVat(vatId, config); + const flux = start(); + // Second: record, with neither `start()` nor this function having awaited + // since, so no crank can run between the operation's first step and the + // record. An await introduced between these two lines reopens the window the + // record exists to close. + // + // Waiters see a plain completion rather than a failure, because the vat ends + // up marked terminated either way — `#endVat` marks it in a `finally` — and + // "gone" is what they should act on. The caller still gets the failure, from + // `flux` itself. + this.#vatsInFlux.set( + vatId, + flux.catch(() => undefined), + ); + 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); }