feat: P2 operator, preprocessing state/service, and runtime config lease

2026-08-11 18:40:11 +02:00
parent ca391ba59c
commit f7c2b69837
10 changed files with 2501 additions and 31 deletions
@@ -0,0 +1,119 @@
+import { existsSync, mkdtempSync, readFileSync, rmSync, statSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { afterEach, expect, test } from "vitest";
+import { PreprocessingStateStore } from "../src/workspaces/preprocessing-state.js";
+
+const roots: string[] = [];
+
+afterEach(() => {
+  roots.splice(0).forEach((root) => rmSync(root, { recursive: true, force: true }));
+});
+
+function fixture() {
+  const dataRoot = mkdtempSync(join(tmpdir(), "tht-preprocessing-state-"));
+  roots.push(dataRoot);
+  return {
+    dataRoot,
+    store: new PreprocessingStateStore({ dataRoot, workspaceId: "psd-clinical" }),
+  };
+}
+
+test("job state creates durable 0600 JSON and enforces same-revision resume", async () => {
+  const { store } = fixture();
+  const job = await store.beginJob({
+    operation: "preprocess dwh",
+    workspaceRevision: "a".repeat(40),
+    descriptorBlob: "b".repeat(40),
+    catalogBlob: "c".repeat(40),
+    configDigest: "sha256:config",
+    bindingDigest: "sha256:bindings",
+  });
+
+  const path = store.jobPath(job.runId);
+  expect(existsSync(path)).toBe(true);
+  expect(statSync(path).mode & 0o777).toBe(0o600);
+  expect(JSON.parse(readFileSync(path, "utf8"))).toMatchObject({
+    schemaVersion: 1,
+    operation: "preprocess dwh",
+    workspaceId: "psd-clinical",
+    workspaceRevision: "a".repeat(40),
+    descriptorBlob: "b".repeat(40),
+    catalogBlob: "c".repeat(40),
+    configDigest: "sha256:config",
+    bindingDigest: "sha256:bindings",
+  });
+
+  await expect(store.beginJob({
+    operation: "preprocess dwh",
+    runId: job.runId,
+    workspaceRevision: "d".repeat(40),
+    descriptorBlob: "b".repeat(40),
+    catalogBlob: "c".repeat(40),
+    configDigest: "sha256:config",
+    bindingDigest: "sha256:bindings",
+  })).rejects.toMatchObject({ code: "preprocessing_resume_mismatch" });
+
+  const resumed = await store.beginJob({
+    operation: "preprocess dwh",
+    runId: job.runId,
+    workspaceRevision: "a".repeat(40),
+    descriptorBlob: "b".repeat(40),
+    catalogBlob: "c".repeat(40),
+    configDigest: "sha256:config",
+    bindingDigest: "sha256:bindings",
+  });
+  expect(resumed.runId).toBe(job.runId);
+});
+
+test("writer lock rejects a concurrent contender and the kernel releases it after holder death", async () => {
+  const f = fixture();
+  const other = new PreprocessingStateStore({ dataRoot: f.dataRoot, workspaceId: "psd-clinical" });
+  const first = await f.store.acquireWriterLock();
+  await expect(other.acquireWriterLock()).rejects.toMatchObject({ code: "preprocessing_conflict" });
+
+  process.kill(first.holderPid, "SIGKILL");
+  const deadline = Date.now() + 5_000;
+  while (Date.now() < deadline) {
+    try {
+      const recovered = await other.acquireWriterLock();
+      await recovered.release();
+      return;
+    } catch (error) {
+      if ((error as { code?: string }).code !== "preprocessing_conflict") throw error;
+      await new Promise((resolve) => setTimeout(resolve, 50));
+    }
+  }
+  throw new Error("writer lock was not released after holder death");
+});
+
+test("session inventory guard blocks resumable sessions pinned to a different revision", async () => {
+  const { store } = fixture();
+
+  await expect(store.assertSessionInventoryCompatible("a".repeat(40), [
+    { id: "open-other", status: "closed", archived: false, workspaceRevision: "b".repeat(40) },
+  ])).rejects.toMatchObject({ code: "preprocessing_conflict" });
+
+  await expect(store.assertSessionInventoryCompatible("a".repeat(40), [
+    { id: "current", status: "open", archived: false, workspaceRevision: "a".repeat(40) },
+    { id: "finalized", status: "finalized", archived: false, workspaceRevision: "b".repeat(40) },
+    { id: "archived", status: "closed", archived: true, workspaceRevision: "c".repeat(40) },
+  ])).resolves.toBeUndefined();
+});
+
+test("candidate and review artifacts are digest-bound durable files", async () => {
+  const { store } = fixture();
+  const runId = "1".repeat(32);
+  const candidate = await store.writeFkCandidates(runId, `tables: []
+`);
+  const review = await store.writeFkReview(runId, {
+    reviewedCandidatesDigest: candidate.digest,
+    annotationsDigest: "sha256:annotations",
+    workspaceRevision: "a".repeat(40),
+  });
+
+  expect(candidate.digest).toMatch(/^sha256:[0-9a-f]{64}$/);
+  expect(review.digest).toMatch(/^sha256:[0-9a-f]{64}$/);
+  expect(store.readFkCandidates(runId)?.digest).toBe(candidate.digest);
+  expect(store.readFkReview(runId)?.reviewedCandidatesDigest).toBe(candidate.digest);
+});