✅ Prove Workspace effects survive a real host crash (#365 slice 7)

architecture.md

@@ -550,6 +550,23 @@ supplies a provider-neutral failure activator so a provider can retain that
 first infrastructure failure by identity; the default coordinator does not use
 it and keeps its existing behavior.
 
+Losing the host is not one of those outcomes, because nothing runs to handle
+it. No cleanup, commit or rollback happens after the process dies: the
+operating system closes its connection and releases the locks it held, and the
+interrupted transaction is left for the next connection to recover. The
+mutation, the immutable root, the current-root pointer and the routed event
+have all been written by then, and recovery exposes the last committed state
+and none of them.
+
+That is the same boundary every other reader already observes rather than a
+second rule for crashes. A second connection sees the last committed state for
+as long as the writer's transaction is uncommitted, so a crash publishes
+nothing that was not visible before it. What a later process finds is that
+committed state — the filesystem, the current root, the retained roots and
+references, and the ordered journal with its event identities and root
+associations — from which it performs no recorded effect again and can
+materialize any retained root.
+
 The Deno journal adapter routes an append ordinarily when no destination is
 bound. A publication may instead bind one exact transaction destination for its
 own lexical scope after the existing secret gate. The route validates the

deno.json

@@ -42,6 +42,7 @@
     "@std/testing/bdd": "jsr:@std/testing@^1/bdd",
     "oxlint": "npm:oxlint@1.74.0",
     "oxlint-tsgolint": "npm:oxlint-tsgolint@0.25.0",
+    "typescript": "npm:typescript@^5.0.0",
     "unist-util-select": "npm:unist-util-select@^5",
     "zod": "npm:zod@^4.3.6",
     "mdast-util-to-string": "npm:mdast-util-to-string@^4"

packages/workflow/src/run.ts

@@ -28,7 +28,6 @@
 
 import { createContext } from "effection";
 import type { Context, Operation } from "effection";
-import { randomUUID } from "node:crypto";
 import { createDurableOperation } from "@executablemd/durable-streams";
 import { ReplayGuard } from "@executablemd/durable-streams";
 import type { EffectDescription, Json, Workflow, Yield } from "@executablemd/durable-streams";
@@ -77,7 +76,9 @@ function* record(description: EffectDescription, base: string): Workflow<unknown
     // value back without running this at all, so neither the identifier nor Git
     // is reached a second time.
     const pinnedCommit = yield* revParse(`${base}^{commit}`);
-    return { runId: randomUUID(), base, pinnedCommit };
+    // Web Crypto rather than `node:crypto`: a run id is allocated in shared
+    // code, which names no host.
+    return { runId: crypto.randomUUID(), base, pinnedCommit };
   });
 }
 

packages/workflow/tests/support/workspace-crash-child.ts

@@ -0,0 +1,207 @@
+/**
+ * The two processes a crash proof needs, and neither of them is the test.
+ *
+ * A cancelled task, a thrown error and a closed scope all unwind. A killed
+ * process does not: its transaction is open when the signal arrives, no
+ * application cleanup runs, and the operating system closes the connection and
+ * releases its locks. The next connection to open the database recovers that
+ * interrupted transaction to the last committed state. Proving that therefore
+ * takes a process the test can kill without warning, and a second one that has
+ * never seen the first — which is what these two modes are.
+ *
+ * ```sh
+ * deno run -A workspace-crash-child.ts crash <root> <run-id>
+ * deno run -A workspace-crash-child.ts inspect <root> <run-id>
+ * ```
+ *
+ * `crash` opens the run, performs one real Workspace effect, and stops inside
+ * the still-open transaction with the mutation, the immutable root, the
+ * current-root pointer and the routed journal row all written and none of them
+ * committed. It reports what that connection can see and then waits to be
+ * killed.
+ *
+ * That stopping point is the connection registry's construction-time routed
+ * append hook, which only a caller that builds the registry can install. So
+ * `crash` assembles the same adapter modules the Deno provider installs, at the
+ * path the provider derives, rather than calling `useWorkflowRunStorage`.
+ * `inspect` has no such need and uses the provider itself.
+ */
+
+import process from "node:process";
+import { durableRun, guardDurableStream, type Workflow } from "@executablemd/durable-streams";
+import { ensure, main, type Operation, suspend } from "effection";
+import { WorkflowRunStorage } from "../../mod.ts";
+import { useWorkflowRunStorage, workflowRunPath } from "../../deno.ts";
+import { createWorkflowRunConnections } from "../../src/deno/connections.ts";
+import { openWorkflowRunDatabase, readRunRow } from "../../src/deno/database.ts";
+import { useJournalRouting } from "../../src/deno/journal-route.ts";
+import { readTransaction } from "../../src/deno/reading.ts";
+import { verifySchema } from "../../src/deno/schema.ts";
+import {
+  createWorkspaceProofEffect,
+  useWorkspaceEffects,
+  withWorkspaceEffects,
+} from "../../src/deno/workspace/effect.ts";
+import type { DenoWorkspaceFilesystem } from "../../src/deno/workspace/filesystem.ts";
+import { currentWorkspaceRoot } from "../../src/deno/workspace/root.ts";
+import {
+  setPrivateWorkspaceClock,
+  transactWorkspaceRoots,
+  usePrivateWorkspace,
+} from "../../src/deno/workspace/private.ts";
+import {
+  BASELINE_EFFECT,
+  count,
+  CRASH_CONTENT,
+  CRASH_EFFECT,
+  CRASH_PATH,
+  readTree,
+  report,
+} from "./workspace-process.ts";
+
+const CLOCK = 1_750_000_100_000;
+
+function* crash(root: string, runId: string): Operation<void> {
+  const path = workflowRunPath(root, runId);
+  let filesystem: DenoWorkspaceFilesystem | undefined;
+  let gateCalls = 0;
+  let baselineExecutions = 0;
+
+  const connections = createWorkflowRunConnections(() => {}, {
+    *afterRoutedJournalAppend(_database, event): Operation<void> {
+      if (event.type !== "yield" || filesystem === undefined) {
+        return;
+      }
+      // Every read below is on the connection that opened the transaction, so
+      // it sees that transaction's own uncommitted writes. Nothing else can.
+      const sqlite = connection.database;
+      const currentRoot = currentWorkspaceRoot(sqlite, path);
+      const journalRow = sqlite
+        .prepare(
+          `SELECT event_id, workspace_root_id FROM journal_events
+             WHERE record LIKE ? ORDER BY sequence DESC LIMIT 1`,
+        )
+        .get(`%"name":"${CRASH_EFFECT}"%`);
+      report({
+        ready: true,
+        content: yield* filesystem.readTextFile(CRASH_PATH),
+        currentRoot,
+        retainedRoots: count(
+          sqlite.prepare("SELECT COUNT(*) AS count FROM workspace_roots").get()?.["count"],
+        ),
+        currentRootRetained: count(
+          sqlite
+            .prepare("SELECT COUNT(*) AS count FROM workspace_roots WHERE root_id = ?")
+            .get(currentRoot)?.["count"],
+        ),
+        journalEventId: journalRow?.["event_id"],
+        journalRootId: journalRow?.["workspace_root_id"],
+        journalRows: count(
+          sqlite.prepare("SELECT COUNT(*) AS count FROM journal_events").get()?.["count"],
+        ),
+        gateCalls,
+        baselineExecutions,
+      });
+      // Deno leaves when its event loop is empty, and a suspended Effection
+      // task is not on it. A timer nothing clears is what keeps this process
+      // and its open transaction alive until the signal arrives.
+      setInterval(() => {}, 1_000);
+      // The transaction stays open from here until the operating system takes
+      // this process away, which is what the process is for.
+      yield* suspend();
+    },
+  });
+  yield* ensure(() => connections.close());
+
+  const connection = connections.at(path);
+  readTransaction(connection.database, () => {
+    verifySchema(connection.database, path, connection.dofs);
+  });
+  const record = readRunRow(connection.database, path);
+
+  yield* useJournalRouting(connections);
+  yield* usePrivateWorkspace(connections);
+  yield* useWorkspaceEffects(connections);
+  const database = yield* openWorkflowRunDatabase({ connection, connections, record });
+  yield* setPrivateWorkspaceClock(database, () => CLOCK);
+
+  const guarded = guardDurableStream(database.journal, function* (event) {
+    if (event.type === "yield") {
+      gateCalls += 1;
+    }
+  });
+
+  function* workflow(): Workflow<void> {
+    // The run this process resumes already holds this effect's result, so it
+    // replays. Executing it would mean the crash effect below is not the
+    // first live work of the process, and the count says which happened.
+    yield createWorkspaceProofEffect(
+      database,
+      { type: "workspace-proof", name: BASELINE_EFFECT },
+      // deno-lint-ignore require-yield
+      function* () {
+        baselineExecutions += 1;
+        return null;
+      },
+    );
+    yield createWorkspaceProofEffect(
+      database,
+      { type: "workspace-proof", name: CRASH_EFFECT },
+      function* (selected) {
+        filesystem = selected;
+        yield* selected.writeFile(CRASH_PATH, CRASH_CONTENT, 0o640);
+        return null;
+      },
+    );
+  }
+
+  yield* withWorkspaceEffects(database, durableRun(workflow, { stream: guarded }));
+  report({ ready: false, reason: "the crash effect committed" });
+}
+
+function* inspect(root: string, runId: string): Operation<void> {
+  yield* useWorkflowRunStorage({ root });
+  const opened = yield* WorkflowRunStorage.operations.lookup(runId);
+  if (!opened.ok) {
+    throw opened.error;
+  }
+  const database = opened.value;
+
+  const entries = yield* database.readJournalEntries();
+  if (!entries.ok) {
+    throw entries.error;
+  }
+
+  const observed = yield* transactWorkspaceRoots(database, function* (workspace) {
+    return {
+      currentRoot: yield* workspace.currentRoot(),
+      tree: yield* readTree(workspace.filesystem, "/"),
+    };
+  });
+  if (!observed.ok) {
+    throw observed.error;
+  }
+
+  report({
+    ...observed.value,
+    events: entries.value.map((entry) => ({
+      eventId: entry.eventId,
+      name: entry.event.type === "yield" ? entry.event.description.name : undefined,
+    })),
+  });
+}
+
+main(function* () {
+  // `process.argv` rather than `Deno.args`: this file is Deno-only to run, and
+  // still has to typecheck under the Node project like every other source.
+  const [mode, root, runId] = process.argv.slice(2);
+  if (mode === "crash") {
+    yield* crash(root, runId);
+    return;
+  }
+  if (mode === "inspect") {
+    yield* inspect(root, runId);
+    return;
+  }
+  throw new Error(`the Workspace crash helper has no ${mode} mode`);
+});

packages/workflow/tests/support/workspace-process.ts

@@ -0,0 +1,67 @@
+/**
+ * What the crash and restart processes and the test that drives them agree on.
+ *
+ * The helpers below are imported by a child process and by the suite that
+ * launches it, so this module starts nothing: a child's `main()` lives in the
+ * child's own file, and importing a constant from it would run the child
+ * inside the test.
+ */
+
+import type { Operation } from "effection";
+import type { DenoWorkspaceFilesystem } from "../../src/deno/workspace/filesystem.ts";
+
+/** The mutation the killed process performs, and never publishes. */
+export const CRASH_PATH = "/crash.txt";
+export const CRASH_CONTENT = "bytes that must never be published";
+export const CRASH_EFFECT = "crash-before-commit";
+
+/** The baseline the crash runs against, committed before the child starts. */
+export const BASELINE_PATH = "/baseline.txt";
+export const BASELINE_CONTENT = "committed baseline bytes";
+export const NESTED_PATH = "/kept/nested.txt";
+export const NESTED_CONTENT = "nested baseline bytes";
+export const BASELINE_EFFECT = "baseline";
+export const BASELINE_CLOCK = 1_750_000_000_000;
+
+/** The two committed effects of the restart proof, and what the first retains. */
+export const SEED_CLOCK = 10_000;
+export const REVISE_CLOCK = 20_000;
+export const HISTORICAL_PATH = "/tree/file.txt";
+export const HISTORICAL_CONTENT = "historical bytes";
+
+/** A SQLite count, which arrives as a `bigint` from a read that asked for one. */
+export function count(value: unknown): number {
+  return typeof value === "bigint" ? Number(value) : Number(value);
+}
+
+/** Everything the Workspace holds, as one comparable value. */
+export function* readTree(
+  filesystem: DenoWorkspaceFilesystem,
+  directory: string,
+): Operation<Record<string, unknown>> {
+  const tree: Record<string, unknown> = {};
+  for (const entry of yield* filesystem.readdir(directory)) {
+    const path = directory === "/" ? `/${entry.name}` : `${directory}/${entry.name}`;
+    const stat = yield* filesystem.lstat(path);
+    if (entry.kind === "directory") {
+      tree[path] = { kind: "directory", mode: stat.mode, mtime: stat.mtime };
+      Object.assign(tree, yield* readTree(filesystem, path));
+    } else if (entry.kind === "symlink") {
+      tree[path] = { kind: "symlink", target: yield* filesystem.readlink(path) };
+    } else {
+      tree[path] = {
+        kind: "file",
+        mode: stat.mode,
+        mtime: stat.mtime,
+        size: stat.size,
+        content: yield* filesystem.readTextFile(path),
+      };
+    }
+  }
+  return tree;
+}
+
+/** One line of JSON on standard output is the whole protocol with the parent. */
+export function report(value: unknown): void {
+  console.log(JSON.stringify(value));
+}