fix(ocap-kernel): record a vat's death in one synchronous step

packages/kernel-test/src/vat-lifecycle.test.ts

@@ -1,6 +1,8 @@
+import { NodejsPlatformServices } from '@metamask/kernel-node-runtime';
 import { makeSQLKernelDatabase } from '@metamask/kernel-store/sqlite/nodejs';
 import { waitUntilQuiescent } from '@metamask/kernel-utils';
 import { makeKernelStore } from '@metamask/ocap-kernel';
+import type { VatId } from '@metamask/ocap-kernel';
 import { describe, expect, it, beforeEach } from 'vitest';
 
 import {
@@ -144,4 +146,75 @@ describe('Vat Lifecycle', { timeout: 30_000 }, () => {
     // The target root object should be cleaned up after GC
     expect(kernelStore.getRootObject(deadVatId)).toBeUndefined();
   });
+
+  it('leaves no record of a terminated vat for the next boot to restore', async () => {
+    const kernelDatabase = await makeSQLKernelDatabase({
+      dbFilename: ':memory:',
+    });
+    const kernel = await makeKernel(
+      kernelDatabase,
+      true,
+      logger.logger.subLogger({ tags: ['test'] }),
+    );
+    const kernelStore = makeKernelStore(kernelDatabase);
+
+    await runTestVats(kernel, {
+      bootstrap: 'main',
+      vats: {
+        main: {
+          bundleSpec: getBundleSpec('logger-vat'),
+          parameters: { name: 'DoomedVat' },
+        },
+      },
+    });
+    await waitUntilQuiescent();
+    const vatId = kernel.getVats()[0]?.id as string;
+
+    await kernel.terminateVat(vatId);
+    await waitUntilQuiescent();
+    kernel.collectGarbage();
+    await waitUntilQuiescent();
+
+    expect(kernelStore.isVatActive(vatId)).toBe(false);
+    expect([...kernelStore.getAllVatRecords()]).toStrictEqual([]);
+  });
+
+  it('retires a vat whose channel to its worker fails', async () => {
+    const kernelDatabase = await makeSQLKernelDatabase({
+      dbFilename: ':memory:',
+    });
+    const platformServices = new NodejsPlatformServices({
+      logger: logger.logger.subLogger({ tags: ['vat-worker-manager'] }),
+    });
+    const kernel = await makeKernel(
+      kernelDatabase,
+      true,
+      logger.logger.subLogger({ tags: ['test'] }),
+      undefined,
+      platformServices,
+    );
+    const kernelStore = makeKernelStore(kernelDatabase);
+
+    await runTestVats(kernel, {
+      bootstrap: 'main',
+      vats: {
+        main: {
+          bundleSpec: getBundleSpec('logger-vat'),
+          parameters: { name: 'DoomedVat' },
+        },
+      },
+    });
+    await waitUntilQuiescent();
+    const vatId = kernel.getVats()[0]?.id as VatId;
+
+    // The kernel cannot see a killed worker, only a bad frame from a live one.
+    platformServices.workers.get(vatId)?.worker.emit('message', NaN);
+    await waitUntilQuiescent();
+    kernel.collectGarbage();
+    await waitUntilQuiescent();
+
+    expect(kernel.getVats()).toStrictEqual([]);
+    expect(kernelStore.isVatActive(vatId)).toBe(false);
+    expect([...kernelStore.getAllVatRecords()]).toStrictEqual([]);
+  });
 });

packages/ocap-kernel/CHANGELOG.md

@@ -71,6 +71,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - A service or IO channel named `__proto__` in a cluster config reaches the bootstrap vat as an ordinary key of `services` instead of replacing that object's prototype ([#1121](https://github.com/Consensys-Incorporated/ocap-kernel/pull/1121))
 - A delivery to a remote peer is written down inside the crank making it and sent once that crank commits, instead of going out mid-crank. A crank that then failed took the kernel's record of the message back while the peer already had it, leaving the kernel's next send sequence behind what the peer had seen. The send sequence counters move with that record rather than in memory beside it, so a crank that rolls back takes them too, and outgoing messages reach the peer in sequence order however long a crank takes to commit — and across a restart, which sends the queue it inherits before anything numbered after it. Giving up on a peer covers a delivery whose crank has yet to commit, so it is not sent to a peer whose promise for it has just been rejected. A peer's messages go to the transport one at a time, since a send that has to dial and shake hands first can otherwise be overtaken by one issued after it, and one whose turn comes after a peer restart or a give-up has retired the queue is dropped rather than sent to a peer it is no longer owed to ([#1101](https://github.com/Consensys-Incorporated/ocap-kernel/pull/1101))
 - With the reference count audit crediting the crank buffer, a delivery that queued a message for later no longer reads as a leak, and the audit runs before the flush hands those messages on — so a violation stops the kernel while the crank can still be undone, rather than after the answers have gone out ([#1095](https://github.com/Consensys-Incorporated/ocap-kernel/pull/1095))
+- A vat's death is recorded in one synchronous step, so a crank can no longer observe a half-terminated vat ([#1093](https://github.com/Consensys-Incorporated/ocap-kernel/pull/1093))
+- `terminateVat` retires a vat that is persisted but not running, such as one whose relaunch failed, instead of throwing `VatNotFoundError` ([#1093](https://github.com/Consensys-Incorporated/ocap-kernel/pull/1093))
+- A worker whose vat fails to finish launching is stopped rather than left running with no record ([#1093](https://github.com/Consensys-Incorporated/ocap-kernel/pull/1093))
+- A vat whose channel to its worker fails is retired, rather than kept as running over a dead channel ([#1093](https://github.com/Consensys-Incorporated/ocap-kernel/pull/1093))
+- `cleanupTerminatedVat` discards the vat's `vatConfig` row, so a partly retired vat is not relaunched on the next boot ([#1093](https://github.com/Consensys-Incorporated/ocap-kernel/pull/1093))
 - A kernel object whose owner retires or abandons it is now orphaned, so its owner and reference-count records no longer outlive the c-list entry they were reachable through; before, the next collection to visit such a kref read an entry that was gone and killed the run loop ([#1091](https://github.com/Consensys-Incorporated/ocap-kernel/pull/1091))
 - A crank that aborts keeps the writes that follow the abort — the vat's death, the collection after it — instead of autocommitting them one statement at a time outside any transaction: the run loop takes a savepoint for the crank and one for the delivery, and only the delivery is rolled back ([#1090](https://github.com/Consensys-Incorporated/ocap-kernel/pull/1090))
 - A vat that exits gracefully stays dead when the rest of its crank then throws; before, the run loop's catch rolled the death record back and would have relaunched a vat whose worker was gone and whose callers had been answered ([#1090](https://github.com/Consensys-Incorporated/ocap-kernel/pull/1090))

packages/ocap-kernel/src/Kernel.ts

@@ -151,10 +151,8 @@ 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);
-      },
+      async (vatId, reason) =>
+        await this.#vatManager.stopVat(vatId, true, reason),
     );
 
     this.#vatManager = new VatManager({

packages/ocap-kernel/src/store/index.test.ts

@@ -1,5 +1,5 @@
 import type { KernelDatabase } from '@metamask/kernel-store';
-import { describe, it, expect, beforeEach } from 'vitest';
+import { describe, it, expect, beforeEach, vi } from 'vitest';
 
 import { makeKernelStore } from './index.ts';
 import { makeMapKernelDatabase } from '../../test/storage.ts';
@@ -616,6 +616,37 @@ describe('kernel store', () => {
     });
   });
 
+  describe('deleteVat', () => {
+    it('keeps the config until the rest of the vat is gone', () => {
+      const ks = makeKernelStore(mockKernelDatabase);
+      ks.setVatConfig('v1', { sourceSpec: 'vat.js' });
+      vi.spyOn(mockKernelDatabase, 'deleteVatStore').mockImplementationOnce(
+        () => {
+          throw new Error('deleteVatStore failed');
+        },
+      );
+
+      expect(() => ks.deleteVat('v1')).toThrow('deleteVatStore failed');
+      expect(ks.isVatActive('v1')).toBe(true);
+
+      ks.deleteVat('v1');
+      expect(ks.isVatActive('v1')).toBe(false);
+    });
+  });
+
+  describe('cleanupTerminatedVat', () => {
+    it('discards the config of a vat whose retirement did not', () => {
+      const ks = makeKernelStore(mockKernelDatabase);
+      ks.setVatConfig('v1', { sourceSpec: 'vat.js' });
+      ks.markVatAsTerminated('v1');
+
+      ks.cleanupTerminatedVat('v1');
+
+      expect(ks.isVatActive('v1')).toBe(false);
+      expect(ks.isVatTerminated('v1')).toBe(false);
+    });
+  });
+
   describe('incarnation ID', () => {
     it('generates a new incarnation ID on first call', () => {
       const ks = makeKernelStore(mockKernelDatabase);

packages/ocap-kernel/src/store/index.ts

@@ -251,9 +251,10 @@ export function makeKernelStore(kdb: KernelDatabase, logger?: Logger) {
    * @param vatId - The vat whose state is to be deleted.
    */
   function deleteVat(vatId: VatId): void {
-    vat.deleteVatConfig(vatId);
     kdb.deleteVatStore(vatId);
     subclusters.removeVatFromSubcluster(vatId);
+    // Last: a retry finds the vat by this row.
+    vat.deleteVatConfig(vatId);
   }
 
   /**

packages/ocap-kernel/src/store/methods/subclusters.test.ts

@@ -495,11 +495,16 @@ describe('getSubclusterMethods', () => {
       expect(map[vatId2]).toBe(scId);
     });
 
-    it('should throw an error if the vat is not in any subcluster', () => {
+    it('leaves a vat that is in no subcluster alone', () => {
       const nonMappedVat = 'vNonMapped' as VatId;
+
       expect(() =>
         subclusterMethods.removeVatFromSubcluster(nonMappedVat),
-      ).toThrow('Vat "vNonMapped" has no subcluster');
+      ).not.toThrow();
+      expect(subclusterMethods.getSubcluster(scId)?.vats).toStrictEqual({
+        vat1: vatId1,
+        vat2: vatId2,
+      });
     });
 
     it('should handle removing the last vat from a subcluster', () => {

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

@@ -239,11 +239,16 @@ export function getSubclusterMethods(ctx: StoreContext) {
   /**
    * Removes a vat from its subcluster.
    *
+   * A no-op for a vat in no subcluster, since `deleteVat` calls this for every
+   * vat it discards.
+   *
    * @param vatId - The ID of the vat to remove.
    */
   function removeVatFromSubcluster(vatId: VatId): void {
-    const subclusterId = getVatSubcluster(vatId);
-    deleteSubclusterVat(subclusterId, vatId);
+    const subclusterId = getVatToSubclusterMap()[vatId];
+    if (subclusterId) {
+      deleteSubclusterVat(subclusterId, vatId);
+    }
   }
 
   // System subcluster mapping methods

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

@@ -312,6 +312,11 @@ export function getVatMethods(ctx: StoreContext) {
     // Clean up any remaining c-list entries and vat-specific counters
     deleteEndpoint(vatID);
 
+    // `vatConfig.${vatID}` is outside the `${vatID}.` sweep. Left behind, the
+    // vat reads as active once the mark is dropped, and the next boot
+    // relaunches it.
+    deleteVatConfig(vatID);
+
     // Remove the vat from the terminated vats list
     forgetTerminatedVat(vatID);
 

packages/ocap-kernel/src/vats/VatHandle.test.ts

@@ -33,9 +33,11 @@ let mockKernelStore: KernelStore;
 const makeVat = async ({
   logger,
   dispatch,
+  onStreamFailure = () => undefined,
 }: {
   logger?: Logger;
   dispatch?: (input: unknown) => void | Promise<void>;
+  onStreamFailure?: (error: Error) => void;
 } = {}): Promise<{
   vat: VatHandle;
   stream: TestDuplexStream<JsonRpcMessage, JsonRpcMessage>;
@@ -54,6 +56,7 @@ const makeVat = async ({
       vatConfig: { sourceSpec: 'not-really-there.js' },
       vatStream,
       logger,
+      onStreamFailure,
     }),
     stream: vatStream,
   };
@@ -121,6 +124,35 @@ describe('VatHandle', () => {
         }),
       );
     });
+
+    it('reports a dead channel to its owner', async () => {
+      const onStreamFailure = vi.fn();
+      const { stream } = await makeVat({ onStreamFailure });
+
+      await stream.receiveInput(NaN);
+      await delay(10);
+
+      expect(onStreamFailure).toHaveBeenCalledWith(
+        expect.objectContaining({
+          message: 'Unexpected stream read error.',
+        }),
+      );
+    });
+
+    it('rejects pending commands when the channel dies', async () => {
+      const { vat, stream } = await makeVat();
+      sendVatCommandMock.mockRestore();
+      const messagePromise = vat.sendVatCommand({
+        method: 'ping' as const,
+        params: [],
+      });
+
+      await stream.receiveInput(NaN);
+
+      await expect(messagePromise).rejects.toThrow(
+        'Unexpected stream read error.',
+      );
+    });
   });
 
   describe('ping', () => {
@@ -295,5 +327,55 @@ describe('VatHandle', () => {
         value: undefined,
       });
     });
+
+    it('rejects pending commands before closing the channel', async () => {
+      const inner = await TestDuplexStream.make<JsonRpcMessage, JsonRpcMessage>(
+        () => undefined,
+        { validateInput: isJsonRpcMessage },
+      );
+      let releaseEnd = (): void => undefined;
+      const stalling = new Promise<void>((resolve) => {
+        releaseEnd = resolve;
+      });
+      const vat = await VatHandle.make({
+        kernelQueue: null as unknown as KernelQueue,
+        kernelStore: mockKernelStore,
+        vatId: 'v0',
+        vatConfig: { sourceSpec: 'not-really-there.js' },
+        // `end` never settles until released, so a rejection that arrives
+        // before then is one that did not wait for the channel to close.
+        vatStream: {
+          drain: inner.drain.bind(inner),
+          write: inner.write.bind(inner),
+          end: async () => stalling,
+        } as unknown as typeof inner,
+        onStreamFailure: () => undefined,
+      });
+      sendVatCommandMock.mockRestore();
+      const messagePromise = vat.sendVatCommand({
+        method: 'ping' as const,
+        params: [],
+      });
+
+      const terminating = vat.terminate(true);
+
+      await expect(messagePromise).rejects.toThrow('Vat was deleted.');
+      releaseEnd();
+      await terminating;
+    });
+
+    it('leaves pending commands alone across a restart', async () => {
+      const { vat } = await makeVat();
+      const settled = vi.fn();
+      // eslint-disable-next-line promise/catch-or-return
+      vat
+        .sendVatCommand({ method: 'ping' as const, params: [] })
+        .then(settled, settled);
+
+      await vat.terminate(false);
+      await delay(10);
+
+      expect(settled).not.toHaveBeenCalled();
+    });
   });
 });