feat(storage): implement production session snapshots

packages/runtime/package.json

@@ -30,6 +30,7 @@
     "./runtime-event-read-model": "./dist/runtime-event-read-model.js",
     "./runtime-resume": "./dist/runtime-resume.js",
     "./runtime-kernel": "./dist/runtime-kernel.js",
+    "./quiescent-session-snapshot": "./dist/quiescent-session-snapshot.js",
     "./session-event-runtime-mapper": "./dist/session-event-runtime-mapper.js",
     "./stream-graph-readiness": "./dist/stream-graph-readiness.js",
     "./stream-graph-admission": "./dist/stream-graph-admission.js",

packages/runtime/src/__tests__/quiescent-session-snapshot.test.ts

@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import assert from 'node:assert/strict';
+import { test } from 'node:test';
+import { SessionSnapshotError } from '@maka/storage/quiescent-session-snapshot';
+import { RuntimeKernel, SessionQuiescentMutationBusyError } from '../runtime-kernel.js';
+import { createRuntimeSessionSnapshotQuiescenceAuthority } from '../quiescent-session-snapshot.js';
+
+test('enters Runtime Kernel quiescence before preparing a Session snapshot', async () => {
+  const calls: string[][] = [];
+  const authority = createRuntimeSessionSnapshotQuiescenceAuthority(
+    {
+      async runSessionQuiescentMutation(sessionIds, operation) {
+        calls.push([...sessionIds]);
+        return await operation();
+      },
+    },
+    {
+      assertSnapshotEligible(makaSessionId) {
+        assert.equal(makaSessionId, 'session-1');
+      },
+    },
+  );
+  const value = await authority.runQuiescent(
+    { makaSessionId: 'session-1', cancellation: { signal: new AbortController().signal } },
+    async () => 'prepared',
+  );
+  assert.equal(value, 'prepared');
+  assert.deepEqual(calls, [['session-1']]);
+});
+
+test('maps an active Runtime execution claim to a stable snapshot_busy error', async () => {
+  const authority = createRuntimeSessionSnapshotQuiescenceAuthority(
+    {
+      async runSessionQuiescentMutation() {
+        throw new SessionQuiescentMutationBusyError(['session-1']);
+      },
+    },
+    {
+      assertSnapshotEligible() {},
+    },
+  );
+  await assert.rejects(
+    authority.runQuiescent(
+      { makaSessionId: 'session-1', cancellation: { signal: new AbortController().signal } },
+      async () => undefined,
+    ),
+    (error) => {
+      assert.ok(error instanceof SessionSnapshotError);
+      assert.equal(error.code, 'snapshot_busy');
+      assert.deepEqual(error.details, { phase: 'admission' });
+      return true;
+    },
+  );
+});
+
+test('does not invoke snapshot work when Host eligibility rejects the Session', async () => {
+  let invoked = false;
+  const authority = createRuntimeSessionSnapshotQuiescenceAuthority(
+    {
+      async runSessionQuiescentMutation(_sessionIds, operation) {
+        return await operation();
+      },
+    },
+    {
+      assertSnapshotEligible() {
+        throw new SessionSnapshotError('snapshot_busy', 'Session has a pending approval', {
+          details: { phase: 'admission' },
+        });
+      },
+    },
+  );
+  await assert.rejects(
+    authority.runQuiescent(
+      { makaSessionId: 'session-1', cancellation: { signal: new AbortController().signal } },
+      async () => {
+        invoked = true;
+      },
+    ),
+    (error) => error instanceof SessionSnapshotError && error.code === 'snapshot_busy',
+  );
+  assert.equal(invoked, false);
+});
+
+test('actual Runtime Kernel serializes an admitted mutation before snapshot work', async () => {
+  const kernel = new RuntimeKernel({} as never);
+  let releaseMutation!: () => void;
+  const mutationRelease = new Promise<void>((resolve) => {
+    releaseMutation = resolve;
+  });
+  let markMutationStarted!: () => void;
+  const mutationStarted = new Promise<void>((resolve) => {
+    markMutationStarted = resolve;
+  });
+  const mutation = kernel.runSessionAdmissionMutation(['session-1'], async () => {
+    markMutationStarted();
+    await mutationRelease;
+  });
+  await mutationStarted;
+
+  let snapshotInvoked = false;
+  const authority = createRuntimeSessionSnapshotQuiescenceAuthority(kernel, {
+    assertSnapshotEligible() {},
+  });
+  const snapshot = authority.runQuiescent(
+    { makaSessionId: 'session-1', cancellation: { signal: new AbortController().signal } },
+    async () => {
+      snapshotInvoked = true;
+    },
+  );
+  await Promise.resolve();
+  assert.equal(snapshotInvoked, false);
+  releaseMutation();
+  await Promise.all([mutation, snapshot]);
+  assert.equal(snapshotInvoked, true);
+});
+
+test('does not run a snapshot admitted after its cancellation', async () => {
+  const controller = new AbortController();
+  controller.abort();
+  let invoked = false;
+  const authority = createRuntimeSessionSnapshotQuiescenceAuthority(
+    {
+      async runSessionQuiescentMutation(_sessionIds, operation) {
+        return await operation();
+      },
+    },
+    {
+      assertSnapshotEligible() {},
+    },
+  );
+  await assert.rejects(
+    authority.runQuiescent(
+      { makaSessionId: 'session-1', cancellation: { signal: controller.signal } },
+      async () => {
+        invoked = true;
+      },
+    ),
+    (error) => error instanceof SessionSnapshotError && error.code === 'snapshot_cancelled',
+  );
+  assert.equal(invoked, false);
+});

packages/runtime/src/quiescent-session-snapshot.ts

@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import {
+  createFileProductionSessionSnapshotService,
+  type FileProductionSessionSnapshotService,
+  type FileProductionSessionSnapshotServiceOptions,
+} from '@maka/storage/production-session-snapshot';
+import {
+  SessionSnapshotError,
+  type SessionSnapshotCancellation,
+  type SessionSnapshotQuiescenceAuthority,
+} from '@maka/storage/quiescent-session-snapshot';
+import { SessionQuiescentMutationBusyError, type RuntimeKernelLike } from './runtime-kernel.js';
+
+export interface RuntimeSessionSnapshotQuiescenceOptions {
+  /**
+   * Host/control-plane check for mutable states that RuntimeKernel does not
+   * own directly, such as pending approvals, background processes, or
+   * externally resumable actions. It must reject any state that could mutate
+   * Session state or the workspace during the snapshot operation.
+   */
+  readonly assertSnapshotEligible: (
+    makaSessionId: string,
+    cancellation: SessionSnapshotCancellation,
+  ) => Promise<void> | void;
+}
+
+export type RuntimeProductionSessionSnapshotServiceOptions = Omit<
+  FileProductionSessionSnapshotServiceOptions,
+  'quiescence'
+> & {
+  readonly kernel: Pick<RuntimeKernelLike, 'runSessionQuiescentMutation'>;
+  readonly quiescence: RuntimeSessionSnapshotQuiescenceOptions;
+};
+
+/**
+ * The Runtime-owned composition point for the #2369 production pipeline. A
+ * control plane supplies roots, limits, and lifetime ownership; this factory
+ * supplies the authoritative Runtime mutation boundary before any state or
+ * workspace bytes are copied.
+ */
+export function createRuntimeProductionSessionSnapshotService(
+  options: RuntimeProductionSessionSnapshotServiceOptions,
+): Promise<FileProductionSessionSnapshotService> {
+  const { kernel, quiescence, ...storageOptions } = options;
+
+  return createFileProductionSessionSnapshotService({
+    ...storageOptions,
+    quiescence: createRuntimeSessionSnapshotQuiescenceAuthority(kernel, quiescence),
+  });
+}
+
+/**
+ * Adapts RuntimeKernel's authoritative mutation lane to the storage snapshot
+ * coordinator. The callback rechecks cancellation after admission, so an
+ * aborted request which was queued behind an earlier mutation performs no copy.
+ */
+export function createRuntimeSessionSnapshotQuiescenceAuthority(
+  kernel: Pick<RuntimeKernelLike, 'runSessionQuiescentMutation'>,
+  options: RuntimeSessionSnapshotQuiescenceOptions,
+): SessionSnapshotQuiescenceAuthority {
+  if (!kernel.runSessionQuiescentMutation) {
+    throw new TypeError('Runtime Kernel does not expose Session quiescence authority');
+  }
+  return {
+    async runQuiescent(input, operation): Promise<unknown> {
+      assertActive(input.cancellation);
+      try {
+        return await kernel.runSessionQuiescentMutation!([input.makaSessionId], async () => {
+          assertActive(input.cancellation);
+          await options.assertSnapshotEligible(input.makaSessionId, input.cancellation);
+          assertActive(input.cancellation);
+          return operation();
+        });
+      } catch (error) {
+        if (error instanceof SessionSnapshotError) throw error;
+        if (error instanceof SessionQuiescentMutationBusyError) {
+          throw new SessionSnapshotError(
+            'snapshot_busy',
+            'Session is busy and cannot be snapshotted',
+            {
+              cause: error,
+              details: { phase: 'admission' },
+            },
+          );
+        }
+        throw new SessionSnapshotError(
+          'io_failure',
+          'Unable to enter the Session snapshot boundary',
+          {
+            cause: error,
+            details: { phase: 'admission' },
+          },
+        );
+      }
+    },
+  } as SessionSnapshotQuiescenceAuthority;
+}
+
+function assertActive(cancellation: SessionSnapshotCancellation): void {
+  if (!cancellation.signal.aborted) return;
+  throw new SessionSnapshotError(
+    'snapshot_cancelled',
+    'Session snapshot preparation was cancelled',
+    {
+      details: { phase: 'admission' },
+    },
+  );
+}

packages/storage/package.json

@@ -41,6 +41,8 @@
     "./runtime-event-persistence": "./dist/runtime-event-persistence.js",
     "./runtime-policy-stores": "./dist/runtime-policy-stores.js",
     "./scheduled-task-store": "./dist/scheduled-task-store.js",
+    "./quiescent-session-snapshot": "./dist/quiescent-session-snapshot.js",
+    "./production-session-snapshot": "./dist/production-session-snapshot.js",
     "./session-bundle-policy": "./dist/session-bundle-policy.js",
     "./session-copy-cleanup": "./dist/session-copy-cleanup.js",
     "./session-store": "./dist/session-store.js",