From e616a1e0bebe8081865690739c74e5ad7c2f2914 Mon Sep 17 00:00:00 2001 From: Taras Mankovski <74687+taras@users.noreply.github.com> Date: Sat, 8 Aug 2026 05:15:15 -0400 Subject: [PATCH 1/3] =?UTF-8?q?=F0=9F=8C=B3=20Retain=20and=20restore=20imm?= =?UTF-8?q?utable=20Workspace=20roots?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- architecture.md | 35 +- packages/workflow/src/deno/connections.ts | 7 +- packages/workflow/src/deno/database.ts | 25 +- packages/workflow/src/deno/provider.ts | 8 +- packages/workflow/src/deno/schema.ts | 6 +- packages/workflow/src/deno/workspace/empty.ts | 144 ---- .../workflow/src/deno/workspace/filesystem.ts | 114 +++ .../workflow/src/deno/workspace/manifest.ts | 290 +++++++ .../workflow/src/deno/workspace/private.ts | 53 ++ .../workflow/src/deno/workspace/restore.ts | 193 +++++ packages/workflow/src/deno/workspace/root.ts | 793 ++++++++++++++++++ .../tests/workflow-run-journal.test.ts | 2 +- .../tests/workflow-run-storage.test.ts | 2 +- .../tests/workspace-root-restoration.test.ts | 386 +++++++++ .../workflow/tests/workspace-root.test.ts | 317 +++++++ scripts/runtime-test-exclusions.ts | 12 + specs/workflow-spec.md | 55 +- 17 files changed, 2274 insertions(+), 168 deletions(-) delete mode 100644 packages/workflow/src/deno/workspace/empty.ts create mode 100644 packages/workflow/src/deno/workspace/filesystem.ts create mode 100644 packages/workflow/src/deno/workspace/manifest.ts create mode 100644 packages/workflow/src/deno/workspace/private.ts create mode 100644 packages/workflow/src/deno/workspace/restore.ts create mode 100644 packages/workflow/src/deno/workspace/root.ts create mode 100644 packages/workflow/tests/workspace-root-restoration.test.ts create mode 100644 packages/workflow/tests/workspace-root.test.ts diff --git a/architecture.md b/architecture.md index b0e1810e..ac1cafbd 100644 --- a/architecture.md +++ b/architecture.md @@ -206,8 +206,23 @@ Workspace-root tables, current-root state, and a non-null Workspace-root association on every journal event. A fresh run starts with one content-addressed root manifest describing only the root directory, empty retained manifest and blob reference sets, and the corresponding root-only DOFS -frontier. This storage layer recognizes that canonical empty frontier; it does -not publish nonempty roots or Workspace mutations. +frontier. + +A Workspace root is a complete, immutable filesystem checkpoint. Its canonical +format-1 JSON includes `/` and every reachable absolute POSIX path in UTF-8 byte +order, with kind, mode, observable mtime, file size and DOFS manifest identity, +symbolic-link target, and deterministic file-hardlink groups. The root ID is +the lowercase SHA-256 of `xmd-workspace-root\0v1\0` followed by those exact +canonical bytes. Mutable inode numbers, revisions, tombstones and cache state +do not participate in the identity. + +The root tables are authoritative checkpoints; the DOFS node, dirent and chunk +tables are the live materialization of the current root. Files remain solely in +DOFS content-addressed blobs. Normalized root-to-manifest and root-to-blob rows +equal the root's transitive content exactly and prevent retained content from +being deleted. Opening a run validates every root and referenced manifest and +blob, then snapshots the live frontier read-only and requires it to equal +`current_root`. Which status transitions are legal, and what a caller may do to a run in each of them, is lifecycle policy applied above storage. @@ -393,8 +408,16 @@ second long-lived DOFS connection is not a coherent reader because provider caches may retain negative entries across another connection's commit. Complete schema version 1 retains the canonical empty Workspace root and its -root-only live frontier. The provider exposes no Workspace mutation operation -and publishes no nonempty retained root at this layer. +root-only live frontier initially. It retains arbitrary canonical roots and can +materialize one privately through the authoritative connection. Capture runs +inside the caller-owned transaction. Restoration runs in a nested savepoint, +clears the authoritative resolution and blob caches, and resnapshots to the +selected identity before release. + +Retained roots, manifests and blobs remain indefinitely. Cloudflare garbage +collection is not in the production closure and is never invoked. The provider +exposes no public Workspace mutation effect, history selection or fork +operation at this layer. The initial topology requires neither writable FUSE nor native subprocess access and does not bundle `workerd`. A Cloudflare-hosted or workerd-backed @@ -683,7 +706,7 @@ Status is measured against main. | `Expansion` / `getExpansion()` | describes the current logical element expansion | built on main | | `useWorkflow()` / `getWorkflowRun()` | associates one document execution with a workflow run | built on main | | `Git.revParse()` | verifies and resolves one Git revision expression contextually | built on main | -| workflow run storage | creates or compatibly finds one run by public run ID, and retains its identity, state, document executions, filtered journal and canonical empty Workspace root through one provider-owned connection entry | built on the #365 stack; Workspace mutation publication is unbuilt | +| workflow run storage | creates or compatibly finds one run by public run ID, retains its identity, state, document executions and filtered journal, and validates immutable Workspace roots through one provider-owned connection entry | built on the #365 stack; Workspace effect publication is unbuilt | | caller-owned storage transaction | publishes several changes, including journal events, in one transaction nothing else enlists in | built on main | | `API.Service` / `startService()` | creates an authenticated, supervised loopback service attachment through a provider-neutral operation | built on main | | `service=` | publishes the attachment's endpoint into the live binding overlay for its invocation | built on main | @@ -694,7 +717,7 @@ Status is measured against main. | Repository / Worktree / transactional Git effects | compose named checkouts and publish local mutations with their journal result | defined in `specs/workflow-workspace-spec.md`, unbuilt | | workflow inspection and history fork | reads status/history without advancing a run and creates a new run from a checkpoint | defined in `specs/workflow-workspace-spec.md`, unbuilt | | read-only workflow Agent / generated XMD | lets an Agent inspect a derived view and propose constrained executable changes | defined in `specs/workflow-workspace-spec.md`, unbuilt | -| Deno-local DOFS provider | owns one authoritative SQLite/DOFS connection per run path and recognizes the complete-v1 canonical empty frontier | built on the #365 stack; nonempty roots and effect-transaction integration are unbuilt | +| Deno-local DOFS provider | owns one authoritative SQLite/DOFS connection per run path, captures arbitrary canonical retained roots, and privately restores them with cache-coherent savepoints | built on the #365 stack; public mutation and effect-transaction integration are unbuilt | | scoped Worker Shell | executes `just-bash` through the Workspace adapter inside a Deno Worker | containment and effect-transaction POCs complete (#351, #357); production integration unbuilt | | `` | retry a region until it completes | defined, unbuilt | | suspension effect | suspend durably | defined, unbuilt | diff --git a/packages/workflow/src/deno/connections.ts b/packages/workflow/src/deno/connections.ts index c265e9d6..e107751b 100644 --- a/packages/workflow/src/deno/connections.ts +++ b/packages/workflow/src/deno/connections.ts @@ -18,6 +18,7 @@ export interface RunConnection { readonly lock: ConnectionLock; readonly savepoints: SavepointManager; transactionOpen: boolean; + setClock(now: () => number): void; close(): void; } @@ -80,12 +81,13 @@ function createConnection(path: string): RunConnection { const dofs = new CloudflareDatabase(durableStorage); const savepoints = createSavepointManager(database, () => connection.transactionOpen); connection.savepoints = savepoints; + let clock = Date.now; return { path, database, dofs, - filesystem: new WorkspaceFilesystem(dofs), + filesystem: new WorkspaceFilesystem(dofs, { now: () => clock() }), lock: createConnectionLock(), savepoints, get transactionOpen() { @@ -94,6 +96,9 @@ function createConnection(path: string): RunConnection { set transactionOpen(value: boolean) { connection.transactionOpen = value; }, + setClock(now: () => number): void { + clock = now; + }, close() { if (open) { open = false; diff --git a/packages/workflow/src/deno/database.ts b/packages/workflow/src/deno/database.ts index afdaf7bc..0b45ea9f 100644 --- a/packages/workflow/src/deno/database.ts +++ b/packages/workflow/src/deno/database.ts @@ -110,6 +110,27 @@ interface Handle { close(): void; } +const DENO_CONNECTION = Symbol("executablemd.workflow.deno.connection"); + +interface DenoWorkflowRunDatabase extends WorkflowRunDatabase { + readonly [DENO_CONNECTION]: RunConnection; +} + +export function workflowRunConnection(database: WorkflowRunDatabase): RunConnection { + if (!isDenoWorkflowRunDatabase(database)) { + throw new WorkflowTransactionError( + "the WorkflowRun database is not owned by this Deno storage provider.", + ); + } + return database[DENO_CONNECTION]; +} + +function isDenoWorkflowRunDatabase( + database: WorkflowRunDatabase, +): database is DenoWorkflowRunDatabase { + return DENO_CONNECTION in database; +} + function createHandle(connection: OpenConnection): Handle { const { database, path, lock } = connection.connection; @@ -245,7 +266,9 @@ function createHandle(connection: OpenConnection): Handle { }, }; - const handle: WorkflowRunDatabase = { + const handle: DenoWorkflowRunDatabase = { + [DENO_CONNECTION]: connection.connection, + get record() { return record; }, diff --git a/packages/workflow/src/deno/provider.ts b/packages/workflow/src/deno/provider.ts index 785eac37..5fd74c98 100644 --- a/packages/workflow/src/deno/provider.ts +++ b/packages/workflow/src/deno/provider.ts @@ -210,7 +210,7 @@ function* lookupWorkflowRun( const record = yield* scoped(function* (): Operation> { yield* lock.hold(); try { - verifySchema(database, path); + verifySchema(database, path, connection.dofs); return Ok(readRunRow(database, path)); } catch (error) { return refusal(error, path); @@ -246,7 +246,7 @@ function establish( const { database } = connection; try { if (!isUninitialized(database, path)) { - verifySchema(database, path); + verifySchema(database, path, connection.dofs); } database.exec("BEGIN IMMEDIATE"); @@ -267,10 +267,10 @@ function establish( ); }); } else { - verifySchema(database, path); + verifySchema(database, path, connection.dofs); } - verifySchema(database, path); + verifySchema(database, path, connection.dofs); const record = readRunRow(database, path); connection.transactionOpen = false; database.exec("COMMIT"); diff --git a/packages/workflow/src/deno/schema.ts b/packages/workflow/src/deno/schema.ts index 47aa422f..d99ea155 100644 --- a/packages/workflow/src/deno/schema.ts +++ b/packages/workflow/src/deno/schema.ts @@ -32,7 +32,7 @@ import { WorkflowIncompleteVersionOneError, WorkflowSchemaVersionError, } from "../storage/errors.ts"; -import { initializeEmptyWorkspace, verifyEmptyWorkspace } from "./workspace/empty.ts"; +import { initializeEmptyWorkspace, verifyWorkspace } from "./workspace/root.ts"; /** * The bytes `XMD1` as a 32-bit integer, written into the SQLite header. @@ -433,7 +433,7 @@ export function isUninitialized(database: DatabaseSync, path: string): boolean { * Structure only. Whether the rows describe the run that was asked for is a * separate question, asked after this one succeeds. */ -export function verifySchema(database: DatabaseSync, path: string): void { +export function verifySchema(database: DatabaseSync, path: string, dofs: CloudflareDatabase): void { checkIntegrity(database, path); const applicationId = readPragmaNumber(database, "application_id", path); @@ -463,7 +463,7 @@ export function verifySchema(database: DatabaseSync, path: string): void { verifyStructure(database, path); checkForeignKeys(database, path); - verifyEmptyWorkspace(database, path); + verifyWorkspace(database, dofs, path); } /** diff --git a/packages/workflow/src/deno/workspace/empty.ts b/packages/workflow/src/deno/workspace/empty.ts deleted file mode 100644 index e23c4c15..00000000 --- a/packages/workflow/src/deno/workspace/empty.ts +++ /dev/null @@ -1,144 +0,0 @@ -import { createHash } from "node:crypto"; -import type { DatabaseSync } from "node:sqlite"; -import { WorkflowDatabaseCorruptError } from "../../storage/errors.ts"; - -export const WORKSPACE_ROOT_FORMAT = 1; -export const EMPTY_WORKSPACE_MANIFEST = - '{"format":1,"entries":[{"path":"/","kind":"directory","mode":493,"mtime":0}]}'; - -const ROOT_DOMAIN = "xmd-workspace-root\0v1\0"; - -export const EMPTY_WORKSPACE_ROOT_ID = workspaceRootId(EMPTY_WORKSPACE_MANIFEST); - -export function workspaceRootId(manifest: string): string { - const hash = createHash("sha256"); - hash.update(ROOT_DOMAIN, "utf8"); - hash.update(manifest, "utf8"); - return hash.digest("hex"); -} - -export function initializeEmptyWorkspace(database: DatabaseSync): void { - database - .prepare("INSERT INTO workspace_roots (root_id, format_version, manifest) VALUES (?, ?, ?)") - .run(EMPTY_WORKSPACE_ROOT_ID, WORKSPACE_ROOT_FORMAT, EMPTY_WORKSPACE_MANIFEST); - database - .prepare("INSERT INTO workspace_state (singleton_id, current_root_id) VALUES (1, ?)") - .run(EMPTY_WORKSPACE_ROOT_ID); -} - -export function verifyEmptyWorkspace(database: DatabaseSync, path: string): void { - const state = database.prepare("SELECT singleton_id, current_root_id FROM workspace_state").all(); - if ( - state.length !== 1 || - number(state[0]?.["singleton_id"]) !== 1 || - state[0]?.["current_root_id"] !== EMPTY_WORKSPACE_ROOT_ID - ) { - corrupt(path, "it does not hold the canonical empty Workspace current-root pointer"); - } - - const roots = database - .prepare("SELECT root_id, format_version, manifest FROM workspace_roots") - .all(); - if ( - roots.length !== 1 || - roots[0]?.["root_id"] !== EMPTY_WORKSPACE_ROOT_ID || - number(roots[0]?.["format_version"]) !== WORKSPACE_ROOT_FORMAT || - roots[0]?.["manifest"] !== EMPTY_WORKSPACE_MANIFEST || - workspaceRootId(String(roots[0]?.["manifest"])) !== EMPTY_WORKSPACE_ROOT_ID - ) { - corrupt(path, "its retained Workspace root is not the canonical empty root"); - } - - requireCount(database, "workspace_root_manifest_refs", 0, path); - requireCount(database, "workspace_root_blob_refs", 0, path); - requireCount(database, "vfs_dirents", 0, path); - requireCount(database, "vfs_chunks", 0, path); - requireCount(database, "vfs_blobs", 0, path); - requireCount(database, "vfs_blob_bytes", 0, path); - requireCount(database, "vfs_manifests", 0, path); - requireCount(database, "vfs_changes", 0, path); - requireCount(database, "_vfs_mounts", 0, path); - - const nodes = database - .prepare( - `SELECT inode, type, mode, mtime, rev, mount_root, stub_size, - manifest_hash, link_target, size - FROM vfs_nodes`, - ) - .all(); - const root = nodes[0]; - if ( - nodes.length !== 1 || - number(root?.["inode"]) !== 1 || - root?.["type"] !== "dir" || - number(root?.["mode"]) !== 0o755 || - number(root?.["mtime"]) !== 0 || - number(root?.["rev"]) !== 0 || - root?.["mount_root"] !== null || - root?.["stub_size"] !== null || - root?.["manifest_hash"] !== null || - root?.["link_target"] !== null || - number(root?.["size"]) !== 0 - ) { - corrupt(path, "its live Workspace frontier is not the canonical empty root filesystem"); - } - - const metadata = database.prepare("SELECT k, v FROM vfs_meta ORDER BY k").all(); - if ( - metadata.length !== 2 || - metadata[0]?.["k"] !== "rev" || - number(metadata[0]?.["v"]) !== 1 || - metadata[1]?.["k"] !== "schema_version" || - number(metadata[1]?.["v"]) !== 5 - ) { - corrupt(path, "its Workspace filesystem metadata is not the pinned empty version-5 state"); - } - - const watermarks = database.prepare("SELECT k, backend, v FROM _vfs_watermark ORDER BY k").all(); - if ( - watermarks.length !== 2 || - watermarks[0]?.["k"] !== "fetchRev" || - watermarks[0]?.["backend"] !== "default" || - number(watermarks[0]?.["v"]) !== 0 || - watermarks[1]?.["k"] !== "pushRev" || - watermarks[1]?.["backend"] !== "default" || - number(watermarks[1]?.["v"]) !== 0 - ) { - corrupt(path, "its Workspace synchronization watermarks are malformed"); - } - - const cursors = database.prepare("SELECT k, backend, path FROM _vfs_fetch_cursor").all(); - if ( - cursors.length !== 1 || - cursors[0]?.["k"] !== "fetch" || - cursors[0]?.["backend"] !== "default" || - cursors[0]?.["path"] !== null - ) { - corrupt(path, "its Workspace synchronization cursor is malformed"); - } - - const foreignJournalRoots = database - .prepare("SELECT COUNT(*) AS count FROM journal_events WHERE workspace_root_id <> ?") - .get(EMPTY_WORKSPACE_ROOT_ID); - if (number(foreignJournalRoots?.["count"]) !== 0) { - corrupt(path, "a journal event does not reference the current retained Workspace root"); - } -} - -function requireCount(database: DatabaseSync, table: string, expected: number, path: string): void { - const row = database.prepare(`SELECT COUNT(*) AS count FROM ${table}`).get(); - if (number(row?.["count"]) !== expected) { - corrupt(path, `its ${table} rows do not describe the canonical empty Workspace`); - } -} - -function number(value: unknown): number | undefined { - if (typeof value === "bigint") { - return Number(value); - } - return typeof value === "number" ? value : undefined; -} - -function corrupt(path: string, reason: string): never { - throw new WorkflowDatabaseCorruptError(path, reason); -} diff --git a/packages/workflow/src/deno/workspace/filesystem.ts b/packages/workflow/src/deno/workspace/filesystem.ts new file mode 100644 index 00000000..b8c1dd2d --- /dev/null +++ b/packages/workflow/src/deno/workspace/filesystem.ts @@ -0,0 +1,114 @@ +import { type Operation, until } from "effection"; +import { link as linkFile } from "../../../vendor/cloudflare-computer-dofs/generated/fs/link.js"; +import type { WorkspaceDirentResult } from "../../../vendor/cloudflare-computer-dofs/generated/fs/readdir.d.ts"; +import { rename as renamePath } from "../../../vendor/cloudflare-computer-dofs/generated/fs/rename.js"; +import type { RunConnection } from "../connections.ts"; + +export interface DenoWorkspaceEntry { + readonly name: string; + readonly kind: "file" | "directory" | "symlink"; +} + +export interface DenoWorkspaceStat { + readonly kind: "file" | "directory" | "symlink"; + readonly mode: number; + readonly mtime: number; + readonly size: number; +} + +export interface DenoWorkspaceFilesystem { + readFile(path: string): Operation; + readTextFile(path: string): Operation; + stat(path: string): Operation; + lstat(path: string): Operation; + readlink(path: string): Operation; + readdir(path: string): Operation; + writeFile(path: string, content: string | Uint8Array, mode?: number): Operation; + mkdir(path: string, options?: { recursive?: boolean; mode?: number }): Operation; + remove(path: string, options?: { recursive?: boolean; force?: boolean }): Operation; + rename(from: string, to: string): Operation; + chmod(path: string, mode: number): Operation; + symlink(target: string, path: string): Operation; + link(existingPath: string, newPath: string): Operation; +} + +export function createDenoWorkspaceFilesystem(connection: RunConnection): DenoWorkspaceFilesystem { + const { dofs, filesystem } = connection; + + function stat(value: { + mode: number; + mtime: number; + size: number; + isFile: boolean; + isDirectory: boolean; + isSymbolicLink: boolean; + }): DenoWorkspaceStat { + const kind = value.isFile ? "file" : value.isDirectory ? "directory" : "symlink"; + return { kind, mode: value.mode, mtime: value.mtime, size: value.size }; + } + + return { + *readFile(path): Operation { + const stream = yield* until(filesystem.readFile(path)); + return new Uint8Array(yield* until(new Response(stream).arrayBuffer())); + }, + + *readTextFile(path): Operation { + const value = yield* until(filesystem.readFile(path, "utf8")); + if (typeof value !== "string") { + throw new Error("the Workspace text read returned a byte stream"); + } + return value; + }, + + *stat(path): Operation { + return stat(yield* until(filesystem.stat(path))); + }, + + *lstat(path): Operation { + return stat(yield* until(filesystem.lstat(path))); + }, + + *readlink(path): Operation { + return yield* until(filesystem.readlink(path)); + }, + + *readdir(path): Operation { + const entries = yield* until(filesystem.readdir(path)); + return entries.map((entry: WorkspaceDirentResult) => ({ + name: entry.name, + kind: entry.isFile ? "file" : entry.isDirectory ? "directory" : "symlink", + })); + }, + + *writeFile(path, content, mode): Operation { + yield* until(filesystem.writeFile(path, content, mode === undefined ? {} : { mode })); + }, + + *mkdir(path, options = {}): Operation { + yield* until(filesystem.mkdir(path, options)); + }, + + *remove(path, options = {}): Operation { + yield* until(filesystem.rm(path, options)); + }, + + // deno-lint-ignore require-yield + *rename(from, to): Operation { + renamePath(dofs, from, to); + }, + + *chmod(path, mode): Operation { + yield* until(filesystem.chmod(path, mode)); + }, + + *symlink(target, path): Operation { + yield* until(filesystem.symlink(target, path)); + }, + + // deno-lint-ignore require-yield + *link(existingPath, newPath): Operation { + linkFile(dofs, existingPath, newPath); + }, + }; +} diff --git a/packages/workflow/src/deno/workspace/manifest.ts b/packages/workflow/src/deno/workspace/manifest.ts new file mode 100644 index 00000000..025802f4 --- /dev/null +++ b/packages/workflow/src/deno/workspace/manifest.ts @@ -0,0 +1,290 @@ +import { createHash } from "node:crypto"; +import { z } from "zod"; +import { WorkflowDatabaseCorruptError } from "../../storage/errors.ts"; + +export const WORKSPACE_ROOT_FORMAT = 1; +export const WORKSPACE_ROOT_DOMAIN = "xmd-workspace-root\0v1\0"; + +const SHA256 = /^[0-9a-f]{64}$/; +const encoder = new TextEncoder(); + +const directoryEntrySchema = z + .object({ + path: z.string(), + kind: z.literal("directory"), + mode: z.number().int().min(0).max(0o7777), + mtime: z.number().int().safe(), + }) + .strict(); + +const fileEntrySchema = z + .object({ + path: z.string(), + kind: z.literal("file"), + mode: z.number().int().min(0).max(0o7777), + mtime: z.number().int().safe(), + size: z.number().int().safe().nonnegative(), + manifest: z.string().regex(SHA256), + hardlink: z + .string() + .regex(/^h[0-9]+$/) + .nullable(), + }) + .strict(); + +const symlinkEntrySchema = z + .object({ + path: z.string(), + kind: z.literal("symlink"), + mode: z.number().int().min(0).max(0o7777), + mtime: z.number().int().safe(), + target: z.string(), + }) + .strict(); + +const rootManifestSchema = z + .object({ + format: z.literal(WORKSPACE_ROOT_FORMAT), + entries: z.array( + z.discriminatedUnion("kind", [directoryEntrySchema, fileEntrySchema, symlinkEntrySchema]), + ), + }) + .strict(); + +export type WorkspaceRootEntry = z.infer["entries"][number]; +export type WorkspaceRootManifest = z.infer; + +export interface StoredWorkspaceRoot { + readonly rootId: string; + readonly manifest: string; + readonly manifestHashes: readonly string[]; + readonly blobHashes: readonly string[]; +} + +export const EMPTY_WORKSPACE_MANIFEST = + '{"format":1,"entries":[{"path":"/","kind":"directory","mode":493,"mtime":0}]}'; + +export const EMPTY_WORKSPACE_ROOT = workspaceRoot(EMPTY_WORKSPACE_MANIFEST, [], []); +export const EMPTY_WORKSPACE_ROOT_ID = EMPTY_WORKSPACE_ROOT.rootId; + +export function workspaceRoot( + manifest: string, + manifestHashes: readonly string[], + blobHashes: readonly string[], +): StoredWorkspaceRoot { + const hash = createHash("sha256"); + hash.update(WORKSPACE_ROOT_DOMAIN, "utf8"); + hash.update(manifest, "utf8"); + return Object.freeze({ + rootId: hash.digest("hex"), + manifest, + manifestHashes: Object.freeze([...manifestHashes]), + blobHashes: Object.freeze([...blobHashes]), + }); +} + +export function workspaceRootId(manifest: string): string { + return workspaceRoot(manifest, [], []).rootId; +} + +export function encodeWorkspaceManifest( + entries: readonly WorkspaceRootEntry[], + databasePath: string, +): string { + const manifest = { format: WORKSPACE_ROOT_FORMAT, entries: [...entries] }; + validateWorkspaceEntries(manifest.entries, databasePath); + return JSON.stringify(manifest); +} + +export function parseWorkspaceManifest( + manifest: string, + databasePath: string, +): WorkspaceRootManifest { + let offered: unknown; + try { + offered = JSON.parse(manifest); + } catch { + corrupt(databasePath, "one of its retained Workspace roots is not JSON"); + } + const parsed = rootManifestSchema.safeParse(offered); + if (!parsed.success) { + corrupt(databasePath, "one of its retained Workspace roots has an invalid manifest"); + } + validateWorkspaceEntries(parsed.data.entries, databasePath); + if (JSON.stringify(parsed.data) !== manifest) { + corrupt(databasePath, "one of its retained Workspace roots is not canonically encoded"); + } + return parsed.data; +} + +export function validateWorkspaceEntries( + entries: readonly WorkspaceRootEntry[], + databasePath: string, +): void { + if (entries.length === 0 || entries[0]?.path !== "/" || entries[0]?.kind !== "directory") { + corrupt(databasePath, "a Workspace root does not begin with its root directory"); + } + + let previous: string | undefined; + let nextHardlink = 0; + const directories = new Set(); + const hardlinkMembers = new Map(); + const hardlinkFirst = new Map(); + + for (const entry of entries) { + validateCanonicalPath(entry.path, databasePath); + if (previous !== undefined && compareUtf8(previous, entry.path) >= 0) { + corrupt(databasePath, "a Workspace root's paths are duplicated or out of canonical order"); + } + previous = entry.path; + + if (entry.path !== "/" && !directories.has(parentPath(entry.path))) { + corrupt(databasePath, "a Workspace root contains an entry without a parent directory"); + } + if (entry.kind === "directory") { + directories.add(entry.path); + } + if ( + entry.kind === "symlink" && + (entry.target.includes("\0") || hasUnpairedSurrogate(entry.target)) + ) { + corrupt(databasePath, "a Workspace root contains an invalid symbolic-link target"); + } + if (entry.kind === "file" && entry.hardlink !== null) { + const first = hardlinkFirst.get(entry.hardlink); + if (first === undefined) { + if (entry.hardlink !== `h${nextHardlink}`) { + corrupt(databasePath, "a Workspace root's hardlinks are not canonically numbered"); + } + nextHardlink += 1; + hardlinkFirst.set(entry.hardlink, entry); + } else if ( + first.mode !== entry.mode || + first.mtime !== entry.mtime || + first.size !== entry.size || + first.manifest !== entry.manifest + ) { + corrupt(databasePath, "a Workspace root's hardlink group has inconsistent metadata"); + } + hardlinkMembers.set(entry.hardlink, (hardlinkMembers.get(entry.hardlink) ?? 0) + 1); + } + } + + for (const count of hardlinkMembers.values()) { + if (count < 2) { + corrupt(databasePath, "a Workspace root contains a one-member hardlink group"); + } + } +} + +export function validateCanonicalPath(value: string, databasePath: string): void { + if (value === "/") { + return; + } + if ( + !value.startsWith("/") || + value.endsWith("/") || + value.includes("\0") || + hasUnpairedSurrogate(value) + ) { + corrupt(databasePath, "a Workspace root contains a noncanonical path"); + } + for (const part of value.slice(1).split("/")) { + if (part === "" || part === "." || part === "..") { + corrupt(databasePath, "a Workspace root contains a noncanonical path component"); + } + } +} + +export function validatePathName(name: string, databasePath: string): void { + if ( + name === "" || + name === "." || + name === ".." || + name.includes("/") || + name.includes("\0") || + hasUnpairedSurrogate(name) + ) { + corrupt(databasePath, "its live Workspace contains a noncanonical name"); + } +} + +export function compareUtf8(left: string, right: string): number { + return Buffer.compare(encoder.encode(left), encoder.encode(right)); +} + +export function parentFirst(left: WorkspaceRootEntry, right: WorkspaceRootEntry): number { + const depth = left.path.split("/").length - right.path.split("/").length; + return depth === 0 ? compareUtf8(left.path, right.path) : depth; +} + +export function parentPath(path: string): string { + const boundary = path.lastIndexOf("/"); + return boundary === 0 ? "/" : path.slice(0, boundary); +} + +export function sha256(value: Uint8Array): Uint8Array { + return new Uint8Array(createHash("sha256").update(value).digest()); +} + +export function toHex(value: Uint8Array): string { + return Buffer.from(value).toString("hex"); +} + +export function fromHex(value: string, databasePath: string, label: string): Uint8Array { + if (!SHA256.test(value)) { + corrupt(databasePath, `${label} is not a lowercase SHA-256 identity`); + } + return new Uint8Array(Buffer.from(value, "hex")); +} + +export function bytes(value: unknown, databasePath: string, label: string): Uint8Array { + if (!(value instanceof Uint8Array)) { + corrupt(databasePath, `${label} is not bytes`); + } + return value; +} + +export function integer(value: unknown, databasePath: string, label: string): number { + const parsed = typeof value === "bigint" ? Number(value) : value; + if (typeof parsed !== "number" || !Number.isSafeInteger(parsed)) { + corrupt(databasePath, `${label} is not a safe integer`); + } + return parsed; +} + +export function nonnegative(value: unknown, databasePath: string, label: string): number { + const parsed = integer(value, databasePath, label); + if (parsed < 0) { + corrupt(databasePath, `${label} is negative`); + } + return parsed; +} + +export function mode(value: unknown, databasePath: string): number { + const parsed = integer(value, databasePath, "Workspace mode"); + if (parsed < 0 || parsed > 0o7777) { + corrupt(databasePath, "a Workspace node has an invalid mode"); + } + return parsed; +} + +export function corrupt(databasePath: string, reason: string): never { + throw new WorkflowDatabaseCorruptError(databasePath, reason); +} + +function hasUnpairedSurrogate(value: string): boolean { + for (let index = 0; index < value.length; index += 1) { + const code = value.charCodeAt(index); + if (code >= 0xd800 && code <= 0xdbff) { + const next = value.charCodeAt(index + 1); + if (next < 0xdc00 || next > 0xdfff) { + return true; + } + index += 1; + } else if (code >= 0xdc00 && code <= 0xdfff) { + return true; + } + } + return false; +} diff --git a/packages/workflow/src/deno/workspace/private.ts b/packages/workflow/src/deno/workspace/private.ts new file mode 100644 index 00000000..fd224439 --- /dev/null +++ b/packages/workflow/src/deno/workspace/private.ts @@ -0,0 +1,53 @@ +import { type Operation, type Result } from "effection"; +import type { WorkflowRunDatabase } from "../../storage/api.ts"; +import { workflowRunConnection } from "../database.ts"; +import { createDenoWorkspaceFilesystem, type DenoWorkspaceFilesystem } from "./filesystem.ts"; +import { type StoredWorkspaceRoot } from "./manifest.ts"; +import { + captureWorkspaceRoot, + type CaptureWorkspaceRootOptions, + currentWorkspaceRoot, + verifyWorkspace, +} from "./root.ts"; +import { restoreWorkspaceRoot, type RestoreWorkspaceRootOptions } from "./restore.ts"; + +export interface PrivateWorkspaceTransaction { + readonly filesystem: DenoWorkspaceFilesystem; + currentRoot(): Operation; + capture(options?: CaptureWorkspaceRootOptions): Operation; + restore(rootId: string, options?: RestoreWorkspaceRootOptions): Operation; +} + +export function* transactWorkspaceRoots( + database: WorkflowRunDatabase, + body: (workspace: PrivateWorkspaceTransaction) => Operation, +): Operation> { + const connection = workflowRunConnection(database); + return yield* database.transact(function* () { + const workspace: PrivateWorkspaceTransaction = { + filesystem: createDenoWorkspaceFilesystem(connection), + + // deno-lint-ignore require-yield + *currentRoot(): Operation { + return currentWorkspaceRoot(connection.database, connection.path); + }, + + // deno-lint-ignore require-yield + *capture(options = {}): Operation { + return captureWorkspaceRoot(connection, options); + }, + + // deno-lint-ignore require-yield + *restore(rootId, options = {}): Operation { + return restoreWorkspaceRoot(connection, rootId, options); + }, + }; + const value = yield* body(workspace); + verifyWorkspace(connection.database, connection.dofs, connection.path); + return value; + }); +} + +export function setPrivateWorkspaceClock(database: WorkflowRunDatabase, now: () => number): void { + workflowRunConnection(database).setClock(now); +} diff --git a/packages/workflow/src/deno/workspace/restore.ts b/packages/workflow/src/deno/workspace/restore.ts new file mode 100644 index 00000000..b725ed6f --- /dev/null +++ b/packages/workflow/src/deno/workspace/restore.ts @@ -0,0 +1,193 @@ +import type { DatabaseSync } from "node:sqlite"; +import { clearBlobCache } from "../../../vendor/cloudflare-computer-dofs/generated/fs/blobCache.js"; +import { clearResolveCache } from "../../../vendor/cloudflare-computer-dofs/generated/fs/resolveCache.js"; +import { WorkflowTransactionError } from "../../storage/errors.ts"; +import type { RunConnection } from "../connections.ts"; +import { + corrupt, + fromHex, + integer, + parentFirst, + parentPath, + parseWorkspaceManifest, + type StoredWorkspaceRoot, + type WorkspaceRootEntry, +} from "./manifest.ts"; +import { + loadWorkspaceRoot, + readDofsManifest, + setCurrentWorkspaceRoot, + snapshotWorkspace, + verifyWorkspace, +} from "./root.ts"; + +export interface RestoreWorkspaceRootOptions { + readonly publish?: boolean; +} + +export function restoreWorkspaceRoot( + connection: RunConnection, + rootId: string, + options: RestoreWorkspaceRootOptions = {}, +): StoredWorkspaceRoot { + if (!connection.transactionOpen) { + throw new WorkflowTransactionError( + "restoring a Workspace root requires the caller-owned workflow transaction to be open.", + ); + } + + const { database, dofs, path, savepoints } = connection; + verifyWorkspace(database, dofs, path); + const selected = loadWorkspaceRoot(database, rootId, path); + clearCaches(connection); + try { + return savepoints.synchronous(() => { + rebuild(database, selected, path); + clearCaches(connection); + const restored = snapshotWorkspace(database, dofs, path, false); + if ( + restored.rootId !== selected.rootId || + restored.manifest !== selected.manifest || + !equalStrings(restored.manifestHashes, selected.manifestHashes) || + !equalStrings(restored.blobHashes, selected.blobHashes) + ) { + corrupt(path, "a retained Workspace root did not materialize to its own identity"); + } + if (options.publish === true) { + setCurrentWorkspaceRoot(database, selected.rootId, path); + } + return selected; + }); + } finally { + clearCaches(connection); + } +} + +export function clearWorkspaceCaches(connection: RunConnection): void { + clearCaches(connection); +} + +function rebuild(database: DatabaseSync, root: StoredWorkspaceRoot, databasePath: string): void { + const parsed = parseWorkspaceManifest(root.manifest, databasePath); + const rootEntry = parsed.entries[0]; + if (rootEntry === undefined || rootEntry.kind !== "directory" || rootEntry.path !== "/") { + corrupt(databasePath, "a retained Workspace root has no root directory"); + } + + database.exec("DELETE FROM vfs_dirents"); + database.exec("DELETE FROM vfs_chunks"); + database.exec("DELETE FROM vfs_changes"); + database.exec("DELETE FROM vfs_nodes"); + + const revision = nextRevision(database, databasePath); + database + .prepare( + `INSERT INTO vfs_nodes + (inode, type, mode, mtime, rev, mount_root, stub_size, manifest_hash, link_target, size) + VALUES (1, 'dir', ?, ?, ?, NULL, NULL, NULL, NULL, 0)`, + ) + .run(rootEntry.mode, rootEntry.mtime, revision); + + const inodes = new Map([["/", 1]]); + const hardlinks = new Map(); + const entries = parsed.entries.slice(1).sort(parentFirst); + for (const entry of entries) { + const parent = parentPath(entry.path); + const parentInode = inodes.get(parent); + if (parentInode === undefined) { + corrupt(databasePath, "a retained Workspace root names a child without a parent"); + } + const name = entry.path.slice(parent === "/" ? 1 : parent.length + 1); + const inode = materializeNode(database, entry, revision, hardlinks, databasePath); + database + .prepare("INSERT INTO vfs_dirents (parent_inode, name, child_inode) VALUES (?, ?, ?)") + .run(parentInode, name, inode); + inodes.set(entry.path, inode); + } +} + +function materializeNode( + database: DatabaseSync, + entry: WorkspaceRootEntry, + revision: number, + hardlinks: Map, + databasePath: string, +): number { + if (entry.kind === "file" && entry.hardlink !== null) { + const existing = hardlinks.get(entry.hardlink); + if (existing !== undefined) { + return existing; + } + } + + let inode: number; + if (entry.kind === "directory") { + const result = database + .prepare( + `INSERT INTO vfs_nodes + (type, mode, mtime, rev, mount_root, stub_size, manifest_hash, link_target, size) + VALUES ('dir', ?, ?, ?, NULL, NULL, NULL, NULL, 0)`, + ) + .run(entry.mode, entry.mtime, revision); + inode = Number(result.lastInsertRowid); + } else if (entry.kind === "symlink") { + const result = database + .prepare( + `INSERT INTO vfs_nodes + (type, mode, mtime, rev, mount_root, stub_size, manifest_hash, link_target, size) + VALUES ('symlink', ?, ?, ?, NULL, NULL, NULL, ?, 0)`, + ) + .run(entry.mode, entry.mtime, revision, entry.target); + inode = Number(result.lastInsertRowid); + } else { + const manifest = readDofsManifest(database, entry.manifest, databasePath); + if (manifest.size !== entry.size) { + corrupt(databasePath, "a retained file size differs from its DOFS manifest"); + } + const result = database + .prepare( + `INSERT INTO vfs_nodes + (type, mode, mtime, rev, mount_root, stub_size, manifest_hash, link_target, size) + VALUES ('file', ?, ?, ?, NULL, NULL, ?, NULL, ?)`, + ) + .run( + entry.mode, + entry.mtime, + revision, + fromHex(entry.manifest, databasePath, "DOFS manifest identity"), + entry.size, + ); + inode = Number(result.lastInsertRowid); + for (const [index, chunk] of manifest.chunks.entries()) { + database + .prepare("INSERT INTO vfs_chunks (inode, idx, hash, size) VALUES (?, ?, ?, ?)") + .run(inode, index, fromHex(chunk.hash, databasePath, "DOFS blob identity"), chunk.size); + } + } + + if (!Number.isSafeInteger(inode) || inode < 1) { + corrupt(databasePath, "restoration did not allocate a valid Workspace inode"); + } + if (entry.kind === "file" && entry.hardlink !== null) { + hardlinks.set(entry.hardlink, inode); + } + return inode; +} + +function nextRevision(database: DatabaseSync, databasePath: string): number { + const row = database.prepare("UPDATE vfs_meta SET v = v + 1 WHERE k = 'rev' RETURNING v").get(); + const revision = integer(row?.["v"], databasePath, "Workspace revision"); + if (revision < 1) { + corrupt(databasePath, "restoration did not establish a valid Workspace revision"); + } + return revision; +} + +function clearCaches(connection: RunConnection): void { + clearResolveCache(connection.dofs); + clearBlobCache(connection.dofs); +} + +function equalStrings(left: readonly string[], right: readonly string[]): boolean { + return JSON.stringify(left) === JSON.stringify(right); +} diff --git a/packages/workflow/src/deno/workspace/root.ts b/packages/workflow/src/deno/workspace/root.ts new file mode 100644 index 00000000..682a04ae --- /dev/null +++ b/packages/workflow/src/deno/workspace/root.ts @@ -0,0 +1,793 @@ +import type { DatabaseSync } from "node:sqlite"; +import { z } from "zod"; +import type { Database as CloudflareDatabase } from "../../../vendor/cloudflare-computer-dofs/generated/storage.js"; +import { buildManifest } from "../../../vendor/cloudflare-computer-dofs/generated/sync/manifests.js"; +import { WorkflowTransactionError } from "../../storage/errors.ts"; +import type { RunConnection } from "../connections.ts"; +import { + bytes, + compareUtf8, + corrupt, + EMPTY_WORKSPACE_ROOT, + encodeWorkspaceManifest, + fromHex, + integer, + mode, + nonnegative, + parseWorkspaceManifest, + sha256, + type StoredWorkspaceRoot, + toHex, + validateCanonicalPath, + validatePathName, + type WorkspaceRootEntry, + workspaceRoot, + WORKSPACE_ROOT_FORMAT, +} from "./manifest.ts"; + +const decoder = new TextDecoder("utf-8", { fatal: true }); +const SHA256 = /^[0-9a-f]{64}$/; + +const dofsManifestSchema = z + .object({ + version: z.literal(1), + chunks: z.array( + z + .object({ + hash: z.string().regex(SHA256), + size: z.number().int().safe().positive(), + }) + .strict(), + ), + }) + .strict(); + +export interface DofsChunk { + readonly hash: Uint8Array; + readonly size: number; +} + +export interface DofsManifest { + readonly size: number; + readonly chunks: readonly { readonly hash: string; readonly size: number }[]; +} + +interface NodeRow { + readonly inode: number; + readonly type: "file" | "dir" | "symlink"; + readonly mode: number; + readonly mtime: number; + readonly rev: number; + readonly manifestHash: Uint8Array | null; + readonly linkTarget: string | null; + readonly size: number; +} + +interface FileContent { + readonly manifest: string; + readonly blobs: readonly string[]; + readonly chunks: readonly DofsChunk[]; +} + +export interface CaptureWorkspaceRootOptions { + readonly publish?: boolean; +} + +export function initializeEmptyWorkspace(database: DatabaseSync): void { + database + .prepare("INSERT INTO workspace_roots (root_id, format_version, manifest) VALUES (?, ?, ?)") + .run(EMPTY_WORKSPACE_ROOT.rootId, WORKSPACE_ROOT_FORMAT, EMPTY_WORKSPACE_ROOT.manifest); + database + .prepare("INSERT INTO workspace_state (singleton_id, current_root_id) VALUES (1, ?)") + .run(EMPTY_WORKSPACE_ROOT.rootId); +} + +export function captureWorkspaceRoot( + connection: RunConnection, + options: CaptureWorkspaceRootOptions = {}, +): StoredWorkspaceRoot { + if (!connection.transactionOpen) { + throw new WorkflowTransactionError( + "capturing a Workspace root requires the caller-owned workflow transaction to be open.", + ); + } + const root = snapshotWorkspace(connection.database, connection.dofs, connection.path, true); + retainWorkspaceRoot(connection.database, root, connection.path); + if (options.publish === true) { + setCurrentWorkspaceRoot(connection.database, root.rootId, connection.path); + } + return root; +} + +export function snapshotWorkspace( + database: DatabaseSync, + dofs: CloudflareDatabase, + databasePath: string, + materializeMissingManifests: boolean, +): StoredWorkspaceRoot { + const currentRev = validateDofsBookkeeping(database, databasePath); + const entries: Array<{ entry: WorkspaceRootEntry; inode: number }> = []; + const visitingDirectories = new Set(); + const reachableNodes = new Set(); + const nonFileNodes = new Set(); + const filePaths = new Map(); + const fileContents = new Map(); + const manifestHashes = new Set(); + const blobHashes = new Set(); + let reachableDirents = 0; + let reachableChunks = 0; + + function visit(inode: number, canonicalPath: string): void { + const node = readNode(database, inode, currentRev, databasePath); + if (node.type === "dir") { + if (visitingDirectories.has(inode) || reachableNodes.has(inode)) { + corrupt( + databasePath, + "its live Workspace contains a directory cycle or directory hardlink", + ); + } + visitingDirectories.add(inode); + reachableNodes.add(inode); + entries.push({ + inode, + entry: { + path: canonicalPath, + kind: "directory", + mode: node.mode, + mtime: node.mtime, + }, + }); + for (const child of readDirents(database, inode, databasePath)) { + reachableDirents += 1; + validatePathName(child.name, databasePath); + const childPath = + canonicalPath === "/" ? `/${child.name}` : `${canonicalPath}/${child.name}`; + visit(child.inode, childPath); + } + visitingDirectories.delete(inode); + return; + } + + if (node.type === "symlink") { + if (nonFileNodes.has(inode) || reachableNodes.has(inode)) { + corrupt(databasePath, "its live Workspace contains a non-file hardlink"); + } + nonFileNodes.add(inode); + reachableNodes.add(inode); + const target = node.linkTarget; + if (target === null || target.includes("\0")) { + corrupt(databasePath, "its live Workspace contains an invalid symbolic-link target"); + } + entries.push({ + inode, + entry: { + path: canonicalPath, + kind: "symlink", + mode: node.mode, + mtime: node.mtime, + target, + }, + }); + return; + } + + reachableNodes.add(inode); + const paths = filePaths.get(inode) ?? []; + paths.push(canonicalPath); + filePaths.set(inode, paths); + let content = fileContents.get(inode); + if (content === undefined) { + content = validateFile(database, dofs, node, databasePath, materializeMissingManifests); + fileContents.set(inode, content); + reachableChunks += content.chunks.length; + manifestHashes.add(content.manifest); + for (const hash of content.blobs) { + blobHashes.add(hash); + } + } + entries.push({ + inode, + entry: { + path: canonicalPath, + kind: "file", + mode: node.mode, + mtime: node.mtime, + size: node.size, + manifest: content.manifest, + hardlink: null, + }, + }); + } + + visit(1, "/"); + if (count(database, "vfs_nodes", databasePath) !== reachableNodes.size) { + corrupt(databasePath, "its live Workspace contains unreachable filesystem nodes"); + } + if (count(database, "vfs_dirents", databasePath) !== reachableDirents) { + corrupt(databasePath, "its live Workspace contains unreachable directory entries"); + } + if (count(database, "vfs_chunks", databasePath) !== reachableChunks) { + corrupt(databasePath, "its live Workspace contains chunks outside reachable files"); + } + + entries.sort((left, right) => compareUtf8(left.entry.path, right.entry.path)); + const groups = [...filePaths.values()] + .filter((paths) => paths.length > 1) + .map((paths) => [...paths].sort(compareUtf8)) + .sort((left, right) => compareUtf8(left[0] ?? "", right[0] ?? "")); + for (const [index, paths] of groups.entries()) { + const group = `h${index}`; + const members = new Set(paths); + for (const item of entries) { + if (item.entry.kind === "file" && members.has(item.entry.path)) { + item.entry.hardlink = group; + } + } + } + + validateDofsContentStore(database, databasePath); + const manifest = encodeWorkspaceManifest( + entries.map((item) => item.entry), + databasePath, + ); + return workspaceRoot( + manifest, + [...manifestHashes].sort(compareUtf8), + [...blobHashes].sort(compareUtf8), + ); +} + +export function retainWorkspaceRoot( + database: DatabaseSync, + root: StoredWorkspaceRoot, + databasePath: string, +): void { + const parsed = parseWorkspaceManifest(root.manifest, databasePath); + const derived = rootFromManifest(database, root.manifest, parsed, databasePath); + if ( + derived.rootId !== root.rootId || + !equalStrings(derived.manifestHashes, root.manifestHashes) || + !equalStrings(derived.blobHashes, root.blobHashes) + ) { + corrupt(databasePath, "a Workspace root does not match its canonical content references"); + } + + const existing = database + .prepare("SELECT format_version, manifest FROM workspace_roots WHERE root_id = ?") + .get(root.rootId); + if (existing === undefined) { + database + .prepare("INSERT INTO workspace_roots (root_id, format_version, manifest) VALUES (?, ?, ?)") + .run(root.rootId, WORKSPACE_ROOT_FORMAT, root.manifest); + for (const hash of root.manifestHashes) { + database + .prepare("INSERT INTO workspace_root_manifest_refs (root_id, manifest_hash) VALUES (?, ?)") + .run(root.rootId, fromHex(hash, databasePath, "Workspace manifest reference")); + } + for (const hash of root.blobHashes) { + database + .prepare("INSERT INTO workspace_root_blob_refs (root_id, blob_hash) VALUES (?, ?)") + .run(root.rootId, fromHex(hash, databasePath, "Workspace blob reference")); + } + } else if ( + integer(existing["format_version"], databasePath, "Workspace root format") !== + WORKSPACE_ROOT_FORMAT || + existing["manifest"] !== root.manifest + ) { + corrupt(databasePath, "a retained Workspace root identity has different stored bytes"); + } + + requireReferenceSet( + database, + root.rootId, + "workspace_root_manifest_refs", + "manifest_hash", + root.manifestHashes, + databasePath, + ); + requireReferenceSet( + database, + root.rootId, + "workspace_root_blob_refs", + "blob_hash", + root.blobHashes, + databasePath, + ); +} + +export function loadWorkspaceRoot( + database: DatabaseSync, + rootId: string, + databasePath: string, +): StoredWorkspaceRoot { + if (!SHA256.test(rootId)) { + corrupt(databasePath, "the selected Workspace root identity is malformed"); + } + const row = database + .prepare("SELECT root_id, format_version, manifest FROM workspace_roots WHERE root_id = ?") + .get(rootId); + if (row === undefined) { + corrupt(databasePath, "the selected Workspace root is not retained"); + } + return parseStoredRoot(database, row, databasePath); +} + +export function setCurrentWorkspaceRoot( + database: DatabaseSync, + rootId: string, + databasePath: string, +): void { + const changed = database + .prepare("UPDATE workspace_state SET current_root_id = ? WHERE singleton_id = 1") + .run(rootId); + if (changed.changes !== 1) { + corrupt(databasePath, "its Workspace current-root pointer is missing"); + } +} + +export function currentWorkspaceRoot(database: DatabaseSync, databasePath: string): string { + const rows = database.prepare("SELECT singleton_id, current_root_id FROM workspace_state").all(); + const row = rows[0]; + if ( + rows.length !== 1 || + integer(row?.["singleton_id"], databasePath, "Workspace singleton") !== 1 || + typeof row?.["current_root_id"] !== "string" || + !SHA256.test(row["current_root_id"]) + ) { + corrupt(databasePath, "it does not hold exactly one valid Workspace current-root pointer"); + } + return row["current_root_id"]; +} + +export function verifyWorkspace( + database: DatabaseSync, + dofs: CloudflareDatabase, + databasePath: string, +): void { + validateDofsContentStore(database, databasePath); + const retained = new Map(); + for (const row of database + .prepare("SELECT root_id, format_version, manifest FROM workspace_roots ORDER BY root_id") + .all()) { + const root = parseStoredRoot(database, row, databasePath); + if (retained.has(root.rootId)) { + corrupt(databasePath, "it contains a duplicate retained Workspace root"); + } + retained.set(root.rootId, root); + requireReferenceSet( + database, + root.rootId, + "workspace_root_manifest_refs", + "manifest_hash", + root.manifestHashes, + databasePath, + ); + requireReferenceSet( + database, + root.rootId, + "workspace_root_blob_refs", + "blob_hash", + root.blobHashes, + databasePath, + ); + } + if (retained.size === 0) { + corrupt(databasePath, "it contains no retained Workspace root"); + } + + const unretainedJournalRoots = database + .prepare( + `SELECT COUNT(*) AS count + FROM journal_events AS event + LEFT JOIN workspace_roots AS root ON root.root_id = event.workspace_root_id + WHERE root.root_id IS NULL`, + ) + .get(); + if (integer(unretainedJournalRoots?.["count"], databasePath, "journal root count") !== 0) { + corrupt(databasePath, "a journal event names a Workspace root that is not retained"); + } + + const current = currentWorkspaceRoot(database, databasePath); + if (!retained.has(current)) { + corrupt(databasePath, "its current Workspace root is not retained"); + } + const live = snapshotWorkspace(database, dofs, databasePath, false); + if (live.rootId !== current) { + corrupt(databasePath, "its live Workspace frontier does not equal its current root"); + } +} + +export function readDofsManifest( + database: DatabaseSync, + hash: string, + databasePath: string, +): DofsManifest { + const hashBytes = fromHex(hash, databasePath, "DOFS manifest identity"); + const row = database + .prepare("SELECT hash, size, encoded, last_seen FROM vfs_manifests WHERE hash = ?") + .get(hashBytes); + if (row === undefined) { + corrupt(databasePath, "a retained Workspace root names a missing DOFS manifest"); + } + if (toHex(bytes(row["hash"], databasePath, "DOFS manifest hash")) !== hash) { + corrupt(databasePath, "a DOFS manifest row carries the wrong identity"); + } + const size = nonnegative(row["size"], databasePath, "DOFS manifest size"); + nonnegative(row["last_seen"], databasePath, "DOFS manifest last-seen value"); + const encoded = bytes(row["encoded"], databasePath, "DOFS manifest encoding"); + if (toHex(sha256(encoded)) !== hash) { + corrupt(databasePath, "a DOFS manifest hash does not match its bytes"); + } + let text: string; + let offered: unknown; + try { + text = decoder.decode(encoded); + offered = JSON.parse(text); + } catch { + corrupt(databasePath, "a DOFS manifest is not canonical UTF-8 JSON"); + } + const parsed = dofsManifestSchema.safeParse(offered); + if (!parsed.success || JSON.stringify(parsed.data) !== text) { + corrupt(databasePath, "a DOFS manifest is not canonically encoded"); + } + const total = parsed.data.chunks.reduce((sum, chunk) => sum + chunk.size, 0); + if (!Number.isSafeInteger(total) || total !== size) { + corrupt(databasePath, "a DOFS manifest size does not equal its chunks"); + } + for (const chunk of parsed.data.chunks) { + validateBlob(database, chunk.hash, chunk.size, databasePath); + } + return Object.freeze({ size, chunks: Object.freeze(parsed.data.chunks) }); +} + +function parseStoredRoot( + database: DatabaseSync, + row: Record, + databasePath: string, +): StoredWorkspaceRoot { + const rootId = row["root_id"]; + const format = integer(row["format_version"], databasePath, "Workspace root format"); + const manifest = row["manifest"]; + if ( + typeof rootId !== "string" || + !SHA256.test(rootId) || + format !== WORKSPACE_ROOT_FORMAT || + typeof manifest !== "string" + ) { + corrupt(databasePath, "one of its retained Workspace roots is malformed"); + } + const parsed = parseWorkspaceManifest(manifest, databasePath); + const root = rootFromManifest(database, manifest, parsed, databasePath); + if (root.rootId !== rootId) { + corrupt(databasePath, "one of its retained Workspace root identities does not match its bytes"); + } + return root; +} + +function rootFromManifest( + database: DatabaseSync, + manifest: string, + parsed: ReturnType, + databasePath: string, +): StoredWorkspaceRoot { + const manifests = new Set(); + for (const entry of parsed.entries) { + if (entry.kind === "file") { + manifests.add(entry.manifest); + } + } + const blobs = new Set(); + for (const hash of manifests) { + for (const chunk of readDofsManifest(database, hash, databasePath).chunks) { + blobs.add(chunk.hash); + } + } + return workspaceRoot(manifest, [...manifests].sort(compareUtf8), [...blobs].sort(compareUtf8)); +} + +function validateFile( + database: DatabaseSync, + dofs: CloudflareDatabase, + node: NodeRow, + databasePath: string, + materializeMissingManifest: boolean, +): FileContent { + const chunks = readChunks(database, node.inode, databasePath); + const total = chunks.reduce((sum, chunk) => sum + chunk.size, 0); + if (!Number.isSafeInteger(total) || total !== node.size) { + corrupt(databasePath, "a Workspace file size does not equal its ordered chunks"); + } + const blobs = chunks.map((chunk) => + validateBlob(database, toHex(chunk.hash), chunk.size, databasePath), + ); + let manifestHash = node.manifestHash; + if (manifestHash === null) { + if (!materializeMissingManifest) { + corrupt(databasePath, "a Workspace file has no retained DOFS manifest"); + } + manifestHash = buildManifest( + dofs, + chunks, + nonnegative(node.mtime, databasePath, "Workspace mtime"), + ); + database + .prepare("UPDATE vfs_nodes SET manifest_hash = ? WHERE inode = ?") + .run(manifestHash, node.inode); + } + if (manifestHash.byteLength !== 32) { + corrupt(databasePath, "a Workspace file has an invalid DOFS manifest identity"); + } + const manifest = toHex(manifestHash); + const encoded = readDofsManifest(database, manifest, databasePath); + if ( + encoded.size !== node.size || + !equalChunks( + encoded.chunks, + chunks.map((chunk) => ({ hash: toHex(chunk.hash), size: chunk.size })), + ) + ) { + corrupt(databasePath, "a Workspace file's DOFS manifest does not equal its chunks"); + } + return Object.freeze({ manifest, blobs: Object.freeze(blobs), chunks: Object.freeze(chunks) }); +} + +function validateDofsContentStore(database: DatabaseSync, databasePath: string): void { + const blobs = database.prepare("SELECT hash, size, last_seen FROM vfs_blobs ORDER BY hash").all(); + for (const row of blobs) { + const hash = bytes(row["hash"], databasePath, "DOFS blob hash"); + if (hash.byteLength !== 32) { + corrupt(databasePath, "a DOFS blob has an invalid hash length"); + } + nonnegative(row["last_seen"], databasePath, "DOFS blob last-seen value"); + validateBlob( + database, + toHex(hash), + nonnegative(row["size"], databasePath, "DOFS blob size"), + databasePath, + ); + } + if (count(database, "vfs_blob_bytes", databasePath) !== blobs.length) { + corrupt(databasePath, "the DOFS blob index and retained bytes are incomplete"); + } + + for (const row of database.prepare("SELECT hash FROM vfs_manifests ORDER BY hash").all()) { + const hash = bytes(row["hash"], databasePath, "DOFS manifest hash"); + if (hash.byteLength !== 32) { + corrupt(databasePath, "a DOFS manifest has an invalid hash length"); + } + readDofsManifest(database, toHex(hash), databasePath); + } +} + +function validateBlob( + database: DatabaseSync, + hash: string, + expectedSize: number, + databasePath: string, +): string { + const hashBytes = fromHex(hash, databasePath, "DOFS blob identity"); + const row = database + .prepare( + `SELECT blob.hash, blob.size, blob.last_seen, content.bytes + FROM vfs_blobs AS blob + JOIN vfs_blob_bytes AS content ON content.hash = blob.hash + WHERE blob.hash = ?`, + ) + .get(hashBytes); + if (row === undefined) { + corrupt(databasePath, "a Workspace file names missing DOFS blob bytes"); + } + if (toHex(bytes(row["hash"], databasePath, "DOFS blob hash")) !== hash) { + corrupt(databasePath, "a DOFS blob row carries the wrong identity"); + } + const size = nonnegative(row["size"], databasePath, "DOFS blob size"); + nonnegative(row["last_seen"], databasePath, "DOFS blob last-seen value"); + const content = bytes(row["bytes"], databasePath, "DOFS blob bytes"); + if ( + size !== expectedSize || + content.byteLength !== expectedSize || + toHex(sha256(content)) !== hash + ) { + corrupt(databasePath, "a DOFS blob's hash or size does not match its bytes"); + } + return hash; +} + +function readNode( + database: DatabaseSync, + inode: number, + currentRev: number, + databasePath: string, +): NodeRow { + const row = database + .prepare( + `SELECT inode, type, mode, mtime, rev, mount_root, stub_size, + manifest_hash, link_target, size + FROM vfs_nodes WHERE inode = ?`, + ) + .get(inode); + if (row === undefined) { + corrupt(databasePath, "its live Workspace contains a dangling directory entry"); + } + const type = row["type"]; + if (type !== "file" && type !== "dir" && type !== "symlink") { + corrupt(databasePath, "its live Workspace contains an unknown node type"); + } + const manifest = row["manifest_hash"]; + if (manifest !== null && !(manifest instanceof Uint8Array)) { + corrupt(databasePath, "its live Workspace contains an invalid manifest hash"); + } + const target = row["link_target"]; + if (target !== null && typeof target !== "string") { + corrupt(databasePath, "its live Workspace contains an invalid link target"); + } + const rev = nonnegative(row["rev"], databasePath, "Workspace node revision"); + if (rev > currentRev) { + corrupt(databasePath, "a Workspace node revision is ahead of the filesystem revision"); + } + if (row["mount_root"] !== null || row["stub_size"] !== null) { + corrupt(databasePath, "its retained Workspace contains unsupported mount bookkeeping"); + } + const result: NodeRow = { + inode: nonnegative(row["inode"], databasePath, "Workspace inode"), + type, + mode: mode(row["mode"], databasePath), + mtime: integer(row["mtime"], databasePath, "Workspace mtime"), + rev, + manifestHash: manifest, + linkTarget: target, + size: nonnegative(row["size"], databasePath, "Workspace size"), + }; + if (result.inode < 1) { + corrupt(databasePath, "a Workspace inode is not positive"); + } + if ( + result.type === "dir" && + (result.manifestHash !== null || result.linkTarget !== null || result.size !== 0) + ) { + corrupt(databasePath, "a Workspace directory carries file or symbolic-link metadata"); + } + if ( + result.type === "symlink" && + (result.manifestHash !== null || result.linkTarget === null || result.size !== 0) + ) { + corrupt(databasePath, "a Workspace symbolic link carries inconsistent metadata"); + } + if (result.type === "file" && result.linkTarget !== null) { + corrupt(databasePath, "a Workspace file carries a symbolic-link target"); + } + return result; +} + +function readDirents( + database: DatabaseSync, + inode: number, + databasePath: string, +): Array<{ name: string; inode: number }> { + const entries: Array<{ name: string; inode: number }> = []; + for (const row of database + .prepare("SELECT name, child_inode FROM vfs_dirents WHERE parent_inode = ?") + .all(inode)) { + const name = row["name"]; + if (typeof name !== "string") { + corrupt(databasePath, "its live Workspace contains an invalid directory-entry name"); + } + const child = nonnegative(row["child_inode"], databasePath, "Workspace child inode"); + if (child < 1) { + corrupt(databasePath, "a Workspace directory entry names an invalid inode"); + } + entries.push({ name, inode: child }); + } + return entries.sort((left, right) => compareUtf8(left.name, right.name)); +} + +function readChunks(database: DatabaseSync, inode: number, databasePath: string): DofsChunk[] { + const chunks: DofsChunk[] = []; + for (const [expected, row] of database + .prepare("SELECT idx, hash, size FROM vfs_chunks WHERE inode = ? ORDER BY idx") + .all(inode) + .entries()) { + const index = nonnegative(row["idx"], databasePath, "Workspace chunk index"); + const hash = bytes(row["hash"], databasePath, "Workspace chunk hash"); + const size = nonnegative(row["size"], databasePath, "Workspace chunk size"); + if (index !== expected || hash.byteLength !== 32 || size === 0) { + corrupt(databasePath, "a Workspace file has malformed or unordered chunks"); + } + validateBlob(database, toHex(hash), size, databasePath); + chunks.push({ hash, size }); + } + return chunks; +} + +function validateDofsBookkeeping(database: DatabaseSync, databasePath: string): number { + const metadata = database.prepare("SELECT k, v FROM vfs_meta ORDER BY k").all(); + if ( + metadata.length !== 2 || + metadata[0]?.["k"] !== "rev" || + metadata[1]?.["k"] !== "schema_version" || + integer(metadata[1]?.["v"], databasePath, "DOFS schema version") !== 5 + ) { + corrupt(databasePath, "its Workspace filesystem metadata is malformed"); + } + const rev = nonnegative(metadata[0]?.["v"], databasePath, "Workspace revision"); + if (rev < 1) { + corrupt(databasePath, "its Workspace revision is not initialized"); + } + + const watermarks = database.prepare("SELECT k, backend, v FROM _vfs_watermark ORDER BY k").all(); + if ( + watermarks.length !== 2 || + watermarks[0]?.["k"] !== "fetchRev" || + watermarks[0]?.["backend"] !== "default" || + integer(watermarks[0]?.["v"], databasePath, "DOFS fetch watermark") !== 0 || + watermarks[1]?.["k"] !== "pushRev" || + watermarks[1]?.["backend"] !== "default" || + integer(watermarks[1]?.["v"], databasePath, "DOFS push watermark") !== 0 + ) { + corrupt(databasePath, "its Workspace synchronization watermarks are malformed"); + } + const cursors = database.prepare("SELECT k, backend, path FROM _vfs_fetch_cursor").all(); + if ( + cursors.length !== 1 || + cursors[0]?.["k"] !== "fetch" || + cursors[0]?.["backend"] !== "default" || + cursors[0]?.["path"] !== null + ) { + corrupt(databasePath, "its Workspace synchronization cursor is malformed"); + } + if (count(database, "_vfs_mounts", databasePath) !== 0) { + corrupt(databasePath, "its retained Workspace contains an unsupported mount"); + } + + for (const row of database + .prepare("SELECT id, rev, path, op FROM vfs_changes ORDER BY id") + .all()) { + const id = nonnegative(row["id"], databasePath, "Workspace change identity"); + const changeRev = nonnegative(row["rev"], databasePath, "Workspace change revision"); + if ( + id < 1 || + changeRev < 1 || + changeRev > rev || + typeof row["path"] !== "string" || + row["op"] !== "delete" + ) { + corrupt(databasePath, "its Workspace change bookkeeping is malformed"); + } + validateCanonicalPath(row["path"], databasePath); + } + return rev; +} + +function requireReferenceSet( + database: DatabaseSync, + rootId: string, + table: string, + column: string, + expected: readonly string[], + databasePath: string, +): void { + const actual = database + .prepare(`SELECT ${column} FROM ${table} WHERE root_id = ?`) + .all(rootId) + .map((row) => toHex(bytes(row[column], databasePath, `${table}.${column}`))) + .sort(compareUtf8); + if (!equalStrings(actual, expected)) { + corrupt(databasePath, `a retained Workspace root has an inexact ${table} reference set`); + } +} + +function count(database: DatabaseSync, table: string, databasePath: string): number { + const row = database.prepare(`SELECT COUNT(*) AS count FROM ${table}`).get(); + return nonnegative(row?.["count"], databasePath, `${table} row count`); +} + +function equalStrings(left: readonly string[], right: readonly string[]): boolean { + return JSON.stringify(left) === JSON.stringify(right); +} + +function equalChunks( + left: readonly { readonly hash: string; readonly size: number }[], + right: readonly { readonly hash: string; readonly size: number }[], +): boolean { + return JSON.stringify(left) === JSON.stringify(right); +} diff --git a/packages/workflow/tests/workflow-run-journal.test.ts b/packages/workflow/tests/workflow-run-journal.test.ts index 35273a69..a2b082bb 100644 --- a/packages/workflow/tests/workflow-run-journal.test.ts +++ b/packages/workflow/tests/workflow-run-journal.test.ts @@ -39,7 +39,7 @@ import { WorkflowTransactionError, } from "../mod.ts"; import { NoOpenTransactionError, savepoint } from "../src/deno/transaction.ts"; -import { EMPTY_WORKSPACE_ROOT_ID } from "../src/deno/workspace/empty.ts"; +import { EMPTY_WORKSPACE_ROOT_ID } from "../src/deno/workspace/manifest.ts"; import { allowJournalInserts, committedEventCount, diff --git a/packages/workflow/tests/workflow-run-storage.test.ts b/packages/workflow/tests/workflow-run-storage.test.ts index 0d4a7d92..b528df75 100644 --- a/packages/workflow/tests/workflow-run-storage.test.ts +++ b/packages/workflow/tests/workflow-run-storage.test.ts @@ -48,7 +48,7 @@ import { EMPTY_WORKSPACE_MANIFEST, EMPTY_WORKSPACE_ROOT_ID, WORKSPACE_ROOT_FORMAT, -} from "../src/deno/workspace/empty.ts"; +} from "../src/deno/workspace/manifest.ts"; import { createRun, definition, diff --git a/packages/workflow/tests/workspace-root-restoration.test.ts b/packages/workflow/tests/workspace-root-restoration.test.ts new file mode 100644 index 00000000..6b4bdc2e --- /dev/null +++ b/packages/workflow/tests/workspace-root-restoration.test.ts @@ -0,0 +1,386 @@ +import { readFileSync } from "node:fs"; +import { DatabaseSync } from "node:sqlite"; +import { describe, it } from "@executablemd/test-support/bdd"; +import { expect } from "@executablemd/test-support/expect"; +import { type Operation } from "effection"; +import { + WorkflowDatabaseCorruptError, + type WorkflowRunDatabase, + WorkflowRunStorage, +} from "../mod.ts"; +import { workflowRunConnection } from "../src/deno/database.ts"; +import { type StoredWorkspaceRoot } from "../src/deno/workspace/manifest.ts"; +import { + type PrivateWorkspaceTransaction, + setPrivateWorkspaceClock, + transactWorkspaceRoots, +} from "../src/deno/workspace/private.ts"; +import { createRun, runPath, tamper, useStorageRoot, withStorage } from "./support/storage.ts"; + +function* transact( + database: WorkflowRunDatabase, + body: (workspace: PrivateWorkspaceTransaction) => Operation, +): Operation { + const result = yield* transactWorkspaceRoots(database, body); + if (!result.ok) { + throw result.error; + } + return result.value; +} + +function* capture( + database: WorkflowRunDatabase, + body: (workspace: PrivateWorkspaceTransaction) => Operation, +): Operation { + return yield* transact(database, function* (workspace) { + yield* body(workspace); + return yield* workspace.capture({ publish: true }); + }); +} + +function count(database: DatabaseSync, table: string): number { + const value = database.prepare(`SELECT COUNT(*) AS count FROM ${table}`).get()?.["count"]; + return typeof value === "bigint" ? Number(value) : Number(value); +} + +function* createCorruptionFixture(storage: string, runId: string): Operation { + yield* withStorage(storage, function* () { + const database = yield* createRun({ runId }); + setPrivateWorkspaceClock(database, () => 1_000); + yield* capture(database, function* (workspace) { + yield* workspace.filesystem.mkdir("/dir"); + yield* workspace.filesystem.writeFile("/dir/file.txt", "first retained bytes", 0o640); + }); + setPrivateWorkspaceClock(database, () => 2_000); + yield* capture(database, function* (workspace) { + yield* workspace.filesystem.writeFile("/dir/file.txt", "second retained bytes", 0o600); + }); + }); +} + +describe("Tier WRR — private Workspace root restoration", () => { + it("WRR4: an older root restores exact state and clears authoritative negative caches", function* () { + const storage = yield* useStorageRoot(); + const path = runPath(storage, "restore-history"); + let historical: StoredWorkspaceRoot | undefined; + let later: StoredWorkspaceRoot | undefined; + + yield* withStorage(storage, function* () { + const database = yield* createRun({ runId: "restore-history" }); + setPrivateWorkspaceClock(database, () => 10_000); + historical = yield* capture(database, function* (workspace) { + yield* workspace.filesystem.mkdir("/tree", { mode: 0o750 }); + yield* workspace.filesystem.writeFile("/tree/file.txt", "historical", 0o640); + yield* workspace.filesystem.link("/tree/file.txt", "/tree/hardlink.txt"); + yield* workspace.filesystem.symlink("file.txt", "/tree/current.txt"); + }); + + setPrivateWorkspaceClock(database, () => 20_000); + later = yield* capture(database, function* (workspace) { + yield* workspace.filesystem.writeFile("/tree/file.txt", "later", 0o600); + yield* workspace.filesystem.rename("/tree/file.txt", "/renamed.txt"); + yield* workspace.filesystem.remove("/tree/current.txt"); + yield* workspace.filesystem.remove("/tree/hardlink.txt"); + yield* workspace.filesystem.mkdir("/later", { mode: 0o700 }); + }); + + const historicalRoot = historical; + const laterRoot = later; + if (historicalRoot === undefined || laterRoot === undefined) { + throw new Error("the historical roots were not captured"); + } + expect(historicalRoot.rootId).not.toBe(laterRoot.rootId); + + const restored = yield* transact(database, function* (workspace) { + let absent: unknown; + try { + yield* workspace.filesystem.readTextFile("/tree/file.txt"); + } catch (error) { + absent = error; + } + expect(absent).toBeInstanceOf(Error); + + const selected = yield* workspace.restore(historicalRoot.rootId, { publish: true }); + expect(yield* workspace.filesystem.readTextFile("/tree/file.txt")).toBe("historical"); + expect(yield* workspace.filesystem.readTextFile("/tree/hardlink.txt")).toBe("historical"); + expect(yield* workspace.filesystem.readlink("/tree/current.txt")).toBe("file.txt"); + expect(yield* workspace.filesystem.lstat("/tree/file.txt")).toEqual({ + kind: "file", + mode: 0o640, + mtime: 10_000, + size: 10, + }); + expect( + (yield* workspace.filesystem.readdir("/tree")).map((entry) => entry.name).toSorted(), + ).toEqual(["current.txt", "file.txt", "hardlink.txt"]); + const resnapshot = yield* workspace.capture({ publish: true }); + expect(resnapshot).toEqual(selected); + return selected; + }); + expect(restored.rootId).toBe(historicalRoot.rootId); + }); + + if (historical === undefined || later === undefined) { + throw new Error("the retained roots are unavailable"); + } + const sqlite = new DatabaseSync(path); + try { + sqlite.exec("PRAGMA foreign_keys = ON"); + const historicalManifest = sqlite + .prepare("SELECT manifest_hash FROM workspace_root_manifest_refs WHERE root_id = ?") + .get(historical.rootId)?.["manifest_hash"]; + const historicalBlob = sqlite + .prepare("SELECT blob_hash FROM workspace_root_blob_refs WHERE root_id = ?") + .get(historical.rootId)?.["blob_hash"]; + if (!(historicalManifest instanceof Uint8Array) || !(historicalBlob instanceof Uint8Array)) { + throw new Error("the historical root has no retained content references"); + } + expect( + sqlite + .prepare( + "SELECT COUNT(*) AS count FROM workspace_root_manifest_refs WHERE root_id = ? AND manifest_hash = ?", + ) + .get(later.rootId, historicalManifest)?.["count"], + ).toBe(0); + + const before = { + manifests: count(sqlite, "vfs_manifests"), + blobs: count(sqlite, "vfs_blobs"), + bytes: count(sqlite, "vfs_blob_bytes"), + }; + expect(() => + sqlite.prepare("DELETE FROM vfs_manifests WHERE hash = ?").run(historicalManifest), + ).toThrow(); + expect(() => + sqlite.prepare("DELETE FROM vfs_blobs WHERE hash = ?").run(historicalBlob), + ).toThrow(); + expect(() => + sqlite.prepare("DELETE FROM vfs_blob_bytes WHERE hash = ?").run(historicalBlob), + ).toThrow(); + expect({ + manifests: count(sqlite, "vfs_manifests"), + blobs: count(sqlite, "vfs_blobs"), + bytes: count(sqlite, "vfs_blob_bytes"), + }).toEqual(before); + } finally { + sqlite.close(); + } + + yield* withStorage(storage, function* () { + const result = yield* WorkflowRunStorage.operations.lookup("restore-history"); + if (!result.ok) { + throw result.error; + } + expect( + yield* transact(result.value, function* (workspace) { + expect(yield* workspace.filesystem.readTextFile("/tree/file.txt")).toBe("historical"); + return yield* workspace.currentRoot(); + }), + ).toBe(historical?.rootId); + }); + }); + + it("WRR5: an in-savepoint restoration failure preserves the prior frontier and pointer", function* () { + const storage = yield* useStorageRoot(); + let historical = ""; + let current = ""; + + yield* withStorage(storage, function* () { + const database = yield* createRun({ runId: "restore-rollback" }); + setPrivateWorkspaceClock(database, () => 100); + historical = (yield* capture(database, function* (workspace) { + yield* workspace.filesystem.writeFile("/historical.txt", "historical"); + })).rootId; + setPrivateWorkspaceClock(database, () => 200); + current = (yield* capture(database, function* (workspace) { + yield* workspace.filesystem.remove("/historical.txt"); + yield* workspace.filesystem.writeFile("/current.txt", "current"); + })).rootId; + + const connection = workflowRunConnection(database); + yield* transact(database, function* (workspace) { + connection.database.exec(` + CREATE TEMP TRIGGER fail_workspace_restore + BEFORE INSERT ON vfs_nodes + WHEN NEW.inode <> 1 + BEGIN + SELECT raise(ABORT, 'restoration insertion refused'); + END + `); + let failure: unknown; + try { + yield* workspace.restore(historical, { publish: true }); + } catch (error) { + failure = error; + } finally { + connection.database.exec("DROP TRIGGER fail_workspace_restore"); + } + expect(failure).toBeInstanceOf(Error); + expect(yield* workspace.currentRoot()).toBe(current); + expect(yield* workspace.filesystem.readTextFile("/current.txt")).toBe("current"); + let historicalFile: unknown; + try { + yield* workspace.filesystem.readTextFile("/historical.txt"); + } catch (error) { + historicalFile = error; + } + expect(historicalFile).toBeInstanceOf(Error); + }); + }); + + const sqlite = new DatabaseSync(runPath(storage, "restore-rollback")); + try { + expect( + sqlite + .prepare("SELECT current_root_id FROM workspace_state WHERE singleton_id = 1") + .get()?.["current_root_id"], + ).toBe(current); + } finally { + sqlite.close(); + } + }); + + it("WRR6: retained-root, content, topology, and live mismatches are read-only corruption", function* () { + const storage = yield* useStorageRoot(); + const cases: Array<{ runId: string; damage(database: DatabaseSync): void }> = [ + { + runId: "missing-manifest-ref", + damage(database) { + const current = currentRoot(database); + database + .prepare("DELETE FROM workspace_root_manifest_refs WHERE root_id = ?") + .run(current); + }, + }, + { + runId: "extra-blob-ref", + damage(database) { + const current = currentRoot(database); + const blob = database + .prepare("SELECT blob_hash FROM workspace_root_blob_refs WHERE root_id <> ? LIMIT 1") + .get(current)?.["blob_hash"]; + if (!(blob instanceof Uint8Array)) { + throw new Error("the historical root has no blob reference"); + } + database + .prepare("INSERT INTO workspace_root_blob_refs (root_id, blob_hash) VALUES (?, ?)") + .run(current, blob); + }, + }, + { + runId: "altered-root-bytes", + damage(database) { + database + .prepare("UPDATE workspace_roots SET manifest = ? WHERE root_id = ?") + .run('{"format":1,"entries":[]}', currentRoot(database)); + }, + }, + { + runId: "wrong-root-id", + damage(database) { + const current = currentRoot(database); + const wrong = "0".repeat(64); + database.exec("PRAGMA foreign_keys = OFF"); + database + .prepare("UPDATE workspace_roots SET root_id = ? WHERE root_id = ?") + .run(wrong, current); + database + .prepare("UPDATE workspace_root_manifest_refs SET root_id = ? WHERE root_id = ?") + .run(wrong, current); + database + .prepare("UPDATE workspace_root_blob_refs SET root_id = ? WHERE root_id = ?") + .run(wrong, current); + database.prepare("UPDATE workspace_state SET current_root_id = ?").run(wrong); + }, + }, + { + runId: "malformed-root-path", + damage(database) { + const current = currentRoot(database); + const manifest = JSON.parse( + String( + database + .prepare("SELECT manifest FROM workspace_roots WHERE root_id = ?") + .get(current)?.["manifest"], + ), + ); + manifest.entries[1].path = "relative"; + database + .prepare("UPDATE workspace_roots SET manifest = ? WHERE root_id = ?") + .run(JSON.stringify(manifest), current); + }, + }, + { + runId: "corrupt-manifest-bytes", + damage(database) { + database.prepare("UPDATE vfs_manifests SET encoded = X'7b7d'").run(); + }, + }, + { + runId: "corrupt-blob-bytes", + damage(database) { + database.prepare("UPDATE vfs_blob_bytes SET bytes = X'00'").run(); + }, + }, + { + runId: "corrupt-blob-size", + damage(database) { + database.prepare("UPDATE vfs_blobs SET size = size + 1").run(); + }, + }, + { + runId: "unordered-chunks", + damage(database) { + database.prepare("UPDATE vfs_chunks SET idx = idx + 1").run(); + }, + }, + { + runId: "missing-live-manifest", + damage(database) { + database.prepare("UPDATE vfs_nodes SET manifest_hash = NULL WHERE type = 'file'").run(); + }, + }, + { + runId: "dangling-dirent", + damage(database) { + database + .prepare("UPDATE vfs_dirents SET child_inode = 999 WHERE name = 'file.txt'") + .run(); + }, + }, + { + runId: "directory-cycle", + damage(database) { + database.prepare("UPDATE vfs_dirents SET child_inode = 1 WHERE name = 'dir'").run(); + }, + }, + { + runId: "live-root-mismatch", + damage(database) { + database.prepare("UPDATE vfs_nodes SET mode = 448 WHERE inode = 1").run(); + }, + }, + ]; + + for (const one of cases) { + yield* createCorruptionFixture(storage, one.runId); + const path = runPath(storage, one.runId); + tamper(path, one.damage); + const before = readFileSync(path); + const result = yield* withStorage(storage, function* () { + return yield* WorkflowRunStorage.operations.lookup(one.runId); + }); + expect(result.ok).toBe(false); + expect(!result.ok && result.error).toBeInstanceOf(WorkflowDatabaseCorruptError); + expect(readFileSync(path)).toEqual(before); + } + }); +}); + +function currentRoot(database: DatabaseSync): string { + return String( + database.prepare("SELECT current_root_id FROM workspace_state WHERE singleton_id = 1").get()?.[ + "current_root_id" + ], + ); +} diff --git a/packages/workflow/tests/workspace-root.test.ts b/packages/workflow/tests/workspace-root.test.ts new file mode 100644 index 00000000..ea44a3f4 --- /dev/null +++ b/packages/workflow/tests/workspace-root.test.ts @@ -0,0 +1,317 @@ +import { readFileSync, readdirSync } from "node:fs"; +import { DatabaseSync } from "node:sqlite"; +import { join } from "node:path"; +import { fileURLToPath } from "node:url"; +import { describe, it } from "@executablemd/test-support/bdd"; +import { expect } from "@executablemd/test-support/expect"; +import { type Operation } from "effection"; +import type { WorkflowRunDatabase } from "../mod.ts"; +import { workflowRunConnection } from "../src/deno/database.ts"; +import { + EMPTY_WORKSPACE_MANIFEST, + EMPTY_WORKSPACE_ROOT_ID, + type StoredWorkspaceRoot, + WORKSPACE_ROOT_DOMAIN, +} from "../src/deno/workspace/manifest.ts"; +import { + type PrivateWorkspaceTransaction, + setPrivateWorkspaceClock, + transactWorkspaceRoots, +} from "../src/deno/workspace/private.ts"; +import { createRun, runPath, useStorageRoot, withStorage } from "./support/storage.ts"; + +function* transact( + database: WorkflowRunDatabase, + body: (workspace: PrivateWorkspaceTransaction) => Operation, +): Operation { + const result = yield* transactWorkspaceRoots(database, body); + if (!result.ok) { + throw result.error; + } + return result.value; +} + +function* capture( + database: WorkflowRunDatabase, + body: (workspace: PrivateWorkspaceTransaction) => Operation, +): Operation { + return yield* transact(database, function* (workspace) { + yield* body(workspace); + return yield* workspace.capture({ publish: true }); + }); +} + +function parseRoot(root: StoredWorkspaceRoot): { + format: number; + entries: Array>; +} { + return JSON.parse(root.manifest); +} + +describe("Tier WRR — immutable retained Workspace roots", () => { + it("WRR1: the canonical empty root remains byte-for-byte compatible with schema v1", function* () { + const storage = yield* useStorageRoot(); + + yield* withStorage(storage, function* () { + const database = yield* createRun(); + const observed = yield* transact(database, function* (workspace) { + expect(yield* workspace.currentRoot()).toBe(EMPTY_WORKSPACE_ROOT_ID); + return yield* workspace.capture({ publish: true }); + }); + + expect(observed.rootId).toBe(EMPTY_WORKSPACE_ROOT_ID); + expect(observed.manifest).toBe(EMPTY_WORKSPACE_MANIFEST); + expect(observed.manifestHashes).toEqual([]); + expect(observed.blobHashes).toEqual([]); + }); + + const sqlite = new DatabaseSync(runPath(storage, "release-1.4")); + try { + expect(sqlite.prepare("SELECT COUNT(*) AS count FROM workspace_roots").get()?.["count"]).toBe( + 1, + ); + expect(WORKSPACE_ROOT_DOMAIN).toBe("xmd-workspace-root\0v1\0"); + } finally { + sqlite.close(); + } + }); + + it("WRR2: complete topology is canonical, content-addressed, and idempotently retained", function* () { + const storage = yield* useStorageRoot(); + let first: StoredWorkspaceRoot | undefined; + let independentlyBuilt: StoredWorkspaceRoot | undefined; + + yield* withStorage(storage, function* () { + const database = yield* createRun({ runId: "canonical-a" }); + setPrivateWorkspaceClock(database, () => 1_700_000_000_000); + first = yield* capture(database, function* (workspace) { + yield* workspace.filesystem.mkdir("/tree", { mode: 0o750 }); + yield* workspace.filesystem.mkdir("/tree/nested", { mode: 0o700 }); + yield* workspace.filesystem.writeFile( + "/tree/nested/file.txt", + "retained-only-in-dofs-blobs", + 0o640, + ); + workflowRunConnection(database) + .database.prepare("UPDATE vfs_nodes SET manifest_hash = NULL WHERE type = 'file'") + .run(); + yield* workspace.filesystem.link("/tree/nested/file.txt", "/tree/hardlink.txt"); + yield* workspace.filesystem.symlink("nested/file.txt", "/tree/current.txt"); + }); + + const repeated = yield* transact(database, function* (workspace) { + return yield* workspace.capture({ publish: true }); + }); + expect(repeated).toEqual(first); + + const other = yield* createRun({ runId: "canonical-b" }); + setPrivateWorkspaceClock(other, () => 1_700_000_000_000); + independentlyBuilt = yield* capture(other, function* (workspace) { + yield* workspace.filesystem.mkdir("/tree", { mode: 0o750 }); + yield* workspace.filesystem.mkdir("/tree/nested", { mode: 0o700 }); + yield* workspace.filesystem.writeFile( + "/tree/nested/file.txt", + "retained-only-in-dofs-blobs", + 0o640, + ); + yield* workspace.filesystem.link("/tree/nested/file.txt", "/tree/hardlink.txt"); + yield* workspace.filesystem.symlink("nested/file.txt", "/tree/current.txt"); + }); + }); + + if (first === undefined || independentlyBuilt === undefined) { + throw new Error("the canonical roots were not captured"); + } + expect(independentlyBuilt.rootId).toBe(first.rootId); + expect(independentlyBuilt.manifest).toBe(first.manifest); + const parsed = parseRoot(first); + expect(parsed).toEqual({ + format: 1, + entries: [ + { path: "/", kind: "directory", mode: 0o755, mtime: 0 }, + { path: "/tree", kind: "directory", mode: 0o750, mtime: 1_700_000_000_000 }, + { + path: "/tree/current.txt", + kind: "symlink", + mode: 0o777, + mtime: 1_700_000_000_000, + target: "nested/file.txt", + }, + { + path: "/tree/hardlink.txt", + kind: "file", + mode: 0o640, + mtime: 1_700_000_000_000, + size: 27, + manifest: first.manifestHashes[0], + hardlink: "h0", + }, + { + path: "/tree/nested", + kind: "directory", + mode: 0o700, + mtime: 1_700_000_000_000, + }, + { + path: "/tree/nested/file.txt", + kind: "file", + mode: 0o640, + mtime: 1_700_000_000_000, + size: 27, + manifest: first.manifestHashes[0], + hardlink: "h0", + }, + ], + }); + expect(first.manifest).not.toContain("retained-only-in-dofs-blobs"); + + const sqlite = new DatabaseSync(runPath(storage, "canonical-a")); + try { + expect(sqlite.prepare("SELECT COUNT(*) AS count FROM workspace_roots").get()?.["count"]).toBe( + 2, + ); + expect( + sqlite.prepare("SELECT COUNT(*) AS count FROM workspace_root_manifest_refs").get()?.[ + "count" + ], + ).toBe(1); + expect( + sqlite.prepare("SELECT COUNT(*) AS count FROM workspace_root_blob_refs").get()?.["count"], + ).toBe(1); + expect(sqlite.prepare("SELECT COUNT(*) AS count FROM vfs_blob_bytes").get()?.["count"]).toBe( + 1, + ); + } finally { + sqlite.close(); + } + }); + + it("WRR3: every observable mutation produces the corresponding immutable root", function* () { + const storage = yield* useStorageRoot(); + const roots: StoredWorkspaceRoot[] = []; + + yield* withStorage(storage, function* () { + const database = yield* createRun({ runId: "root-sequence" }); + let time = 100; + setPrivateWorkspaceClock(database, () => time); + + roots.push( + yield* capture(database, function* (workspace) { + yield* workspace.filesystem.mkdir("/dir", { mode: 0o750 }); + yield* workspace.filesystem.writeFile("/dir/file.txt", "first", 0o640); + yield* workspace.filesystem.writeFile("/delete.txt", "delete-me", 0o600); + yield* workspace.filesystem.link("/dir/file.txt", "/hardlink.txt"); + yield* workspace.filesystem.symlink("/dir/file.txt", "/current"); + }), + ); + + time += 1; + roots.push( + yield* capture(database, function* (workspace) { + yield* workspace.filesystem.writeFile("/dir/file.txt", "second", 0o640); + }), + ); + roots.push( + yield* capture(database, function* (workspace) { + yield* workspace.filesystem.remove("/delete.txt"); + }), + ); + roots.push( + yield* capture(database, function* (workspace) { + yield* workspace.filesystem.rename("/dir/file.txt", "/renamed.txt"); + }), + ); + time += 1; + roots.push( + yield* capture(database, function* (workspace) { + yield* workspace.filesystem.mkdir("/later", { mode: 0o700 }); + }), + ); + time += 1; + roots.push( + yield* capture(database, function* (workspace) { + yield* workspace.filesystem.chmod("/renamed.txt", 0o600); + }), + ); + time += 1; + roots.push( + yield* capture(database, function* (workspace) { + yield* workspace.filesystem.remove("/current"); + yield* workspace.filesystem.symlink("/renamed.txt", "/current"); + }), + ); + roots.push( + yield* capture(database, function* (workspace) { + yield* workspace.filesystem.remove("/hardlink.txt"); + }), + ); + + const repeated = yield* transact(database, function* (workspace) { + return yield* workspace.capture({ publish: true }); + }); + expect(repeated.rootId).toBe(roots.at(-1)?.rootId); + }); + + expect(new Set(roots.map((root) => root.rootId)).size).toBe(roots.length); + const final = roots.at(-1); + if (final === undefined) { + throw new Error("the final root was not captured"); + } + const entries = parseRoot(final).entries; + expect(entries.map((entry) => entry.path)).toEqual([ + "/", + "/current", + "/dir", + "/later", + "/renamed.txt", + ]); + expect(entries.find((entry) => entry.path === "/current")).toEqual({ + path: "/current", + kind: "symlink", + mode: 0o777, + mtime: 104, + target: "/renamed.txt", + }); + expect(entries.find((entry) => entry.path === "/renamed.txt")?.["mode"]).toBe(0o600); + expect(entries.find((entry) => entry.path === "/renamed.txt")?.["hardlink"]).toBe(null); + + const sqlite = new DatabaseSync(runPath(storage, "root-sequence")); + try { + expect(sqlite.prepare("SELECT COUNT(*) AS count FROM workspace_roots").get()?.["count"]).toBe( + roots.length + 1, + ); + } finally { + sqlite.close(); + } + }); + + it("WRR7: the production closure contains no DOFS garbage-collection path", function* () { + // deno-lint-ignore require-yield + const denoAdapter = fileURLToPath(new URL("../src/deno", import.meta.url)); + const vendorManifest = fileURLToPath( + new URL("../vendor/cloudflare-computer-dofs/MANIFEST.json", import.meta.url), + ); + const sources = sourceFiles(denoAdapter); + expect(sources.some((source) => /from\s+["'][^"']*\/gc(?:\.[^"']*)?["']/.test(source))).toBe( + false, + ); + expect(sources.some((source) => /\.gc\s*\(/.test(source))).toBe(false); + const manifest = JSON.parse(readFileSync(vendorManifest, "utf8")); + expect( + manifest.files.some((file: { path: string }) => /(^|\/)gc(?:\.[^/]*)?$/.test(file.path)), + ).toBe(false); + }); +}); + +function sourceFiles(directory: string): string[] { + const sources: string[] = []; + for (const entry of readdirSync(directory, { withFileTypes: true })) { + const path = join(directory, entry.name); + if (entry.isDirectory()) { + sources.push(...sourceFiles(path)); + } else if (entry.isFile() && entry.name.endsWith(".ts")) { + sources.push(readFileSync(path, "utf8")); + } + } + return sources; +} diff --git a/scripts/runtime-test-exclusions.ts b/scripts/runtime-test-exclusions.ts index bc974231..b906b322 100644 --- a/scripts/runtime-test-exclusions.ts +++ b/scripts/runtime-test-exclusions.ts @@ -96,6 +96,18 @@ const DENO_ONLY_TOOLING: RuntimeExclusion[] = [ "the same Deno storage adapter, plus a restart proof that relaunches the run under the Deno executable; `node:sqlite` is behind --experimental-sqlite on Node 22", issue: DERIVED_SCOPE, }, + { + path: "packages/workflow/tests/workspace-root.test.ts", + reason: + "exercises the Deno-private authoritative DOFS/SQLite adapter through node:sqlite, which remains behind --experimental-sqlite on Node 22", + issue: "https://github.com/taras/executable.md/issues/365", + }, + { + path: "packages/workflow/tests/workspace-root-restoration.test.ts", + reason: + "restores and corrupts real Deno-owned node:sqlite WorkflowRun databases; the provider mechanics are intentionally runtime-specific", + issue: "https://github.com/taras/executable.md/issues/365", + }, ]; /** diff --git a/specs/workflow-spec.md b/specs/workflow-spec.md index 0a47a094..13693fef 100644 --- a/specs/workflow-spec.md +++ b/specs/workflow-spec.md @@ -275,9 +275,23 @@ version domains. A fresh database contains one content-addressed Workspace root whose canonical manifest describes only `/` as a directory. Its retained manifest and blob reference sets are empty, its current-root pointer names that root, and its DOFS -frontier contains only the corresponding root directory. This implementation -recognizes that canonical empty frontier and rejects any other live frontier as -corruption. It exposes no Workspace mutation through the storage handle. +frontier contains only the corresponding root directory. + +Every later retained root is a complete canonical filesystem checkpoint. Root +format 1 uses fixed-key-order UTF-8 JSON containing `/` and every reachable +absolute POSIX path sorted by UTF-8 bytes. Each entry records its kind, mode and +observable mtime; files also record size, their DOFS manifest identity and a +deterministic hardlink group when shared; symbolic links record their verbatim +target. Paths are not Unicode-normalized. Mutable DOFS inode numbers, revisions, +tombstones, caches and synchronization bookkeeping are not part of root +identity. + +The lowercase root ID is SHA-256 over +`xmd-workspace-root\0v1\0 || canonical_manifest_bytes`. Reusing an ID requires +the stored format and bytes to be identical. File bytes remain only in DOFS +blobs. The normalized root-to-manifest and root-to-blob rows equal the exact +transitive content of each root and prevent that content from being deleted +while the root is retained. ### 9.5 The journal @@ -347,6 +361,25 @@ Cloudflare's synchronous transactions use uniquely named SQLite savepoints on that same connection and only while XMD's caller-owned transaction is open. DOFS does not begin, commit or roll back a top-level transaction. +Adapter-private root operations also run only inside this caller-owned +transaction. Capture traverses and validates the complete live DOFS frontier, +builds or reuses a canonical DOFS file manifest when ordered chunks do not yet +have one, retains the immutable root and exact reference sets, and optionally +sets it current. Read-only recognition never builds a manifest or changes +last-seen metadata. + +The private restoration materializer loads a fully validated retained root and +rebuilds directories, files, chunks, modes, mtimes, symbolic links and hardlink +relationships inside a nested savepoint. It establishes valid mutable revision +state, clears the authoritative DOFS resolution and blob caches, and requires a +read-only resnapshot to reproduce the selected root ID before the savepoint is +released. A failure restores the prior live frontier and current-root pointer. + +Immutable roots are authoritative checkpoints and DOFS tables are the current +live materialization. Every retained root remains indefinitely. The production +closure neither exposes nor invokes Cloudflare garbage collection; root-aware +deletion and collection are separate lifecycle behavior. + A transaction opened inside another on the same database is refused rather than nested, and so is an ordinary operation called from inside a body — that call would otherwise wait for a transaction its own scope is holding open. @@ -388,6 +421,13 @@ genuinely unsupported nonzero version remains a schema-version refusal. Rows are held to what they mean and not only to their column types: a timestamp is an instant, an identity is not the empty string, and props are an object. +Semantic recognition parses every retained root canonically, recomputes its ID +and exact manifest/blob reachability, validates every referenced manifest, +blob, byte payload and live chunk, and requires the read-only live snapshot to +equal the singleton current root. Malformed paths or topology, dangling or +cyclic dirents, invalid hardlinks, corrupt hashes or sizes, inexact references, +and a live/current mismatch are damage. Recognition performs no repair. + No message repeats a stored value — or a stored *name*. Props and journal payloads are retained history, and a member name can carry a credential as readily as a member value, so an unexpected member is refused without being @@ -404,7 +444,8 @@ also left unchanged. ## 10. Intentionally excluded Public `xmd workflow` lifecycle commands; lifecycle transition policy, executor -leases and stale-owner recovery; Workspace mutations, nonempty retained roots -and restoration; provider-level Workspace effect publication; history -checkpoints and forks; workflow-owned worktrees; and deterministic Git and -GitHub effects. +leases and stale-owner recovery; public Workspace mutation and filesystem +effects; provider-level atomic Workspace effect/journal publication; public +root selection, history checkpoints and forks; `` integration; +workflow-owned worktrees; and deterministic Git and GitHub effects. Retained +roots and private restoration do not expose any of those behaviors. From cbe78a13459094efd5fc8368a24eb9986fd2ee4b Mon Sep 17 00:00:00 2001 From: Taras Mankovski <74687+taras@users.noreply.github.com> Date: Sat, 8 Aug 2026 16:06:09 -0400 Subject: [PATCH 2/3] =?UTF-8?q?=F0=9F=90=9B=20Harden=20retained=20Workspac?= =?UTF-8?q?e=20root=20recognition?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- architecture.md | 18 +- packages/workflow/src/deno/database.ts | 17 +- packages/workflow/src/deno/provider.ts | 41 +++- packages/workflow/src/deno/reading.ts | 32 +++ packages/workflow/src/deno/schema.ts | 3 +- .../workflow/src/deno/workspace/private.ts | 6 +- .../workflow/src/deno/workspace/restore.ts | 3 +- packages/workflow/src/deno/workspace/root.ts | 96 ++++----- .../tests/workflow-run-storage.test.ts | 68 +++++++ .../tests/workspace-root-restoration.test.ts | 186 +++++++++++++++++- .../workflow/tests/workspace-root.test.ts | 2 +- specs/workflow-spec.md | 14 +- 12 files changed, 408 insertions(+), 78 deletions(-) create mode 100644 packages/workflow/src/deno/reading.ts diff --git a/architecture.md b/architecture.md index ac1cafbd..5a10aac4 100644 --- a/architecture.md +++ b/architecture.md @@ -222,7 +222,10 @@ DOFS content-addressed blobs. Normalized root-to-manifest and root-to-blob rows equal the root's transitive content exactly and prevent retained content from being deleted. Opening a run validates every root and referenced manifest and blob, then snapshots the live frontier read-only and requires it to equal -`current_root`. +`current_root`. The schema structure, retained content, live/current comparison +and run row are read through one explicit SQLite snapshot. This recognition +transaction is not a caller-owned Workspace transaction and enables no DOFS +savepoints. Which status transitions are legal, and what a caller may do to a run in each of them, is lifecycle policy applied above storage. @@ -259,7 +262,9 @@ A transaction commits only once the work inside it has finished, including work that is still unwinding when the body returns. Cleanup belonging to that work appends through the same transaction, and a commit that happened first would leave those appends to publish themselves outside it. Failure and cancellation -roll back everything the transaction did. +roll back everything the transaction did. Adapter-private Workspace work also +waits for its supplied scope to finish teardown before it performs the final +live/current validation. A transaction opened inside another on the same storage is refused rather than nested, as is an ordinary operation called from inside a transaction body. @@ -360,6 +365,11 @@ before their parent's effect begins. Declarative Git operations, including staging, switching and committing, operate on the same transactional Workspace rather than invoking an untracked native Git side effect. +Successful effect coordination finishes the mutation scope, including child +cleanup, before capturing the resulting root. The provider-level coordinator +that performs this ordering and journal publication is not installed at this +layer. + An external provider cannot join that transaction. Prompt, Git push and pull request effects derive a stable identity from the run and expansion, ask the provider to perform or reconcile that identity, then append one local result @@ -412,7 +422,9 @@ root-only live frontier initially. It retains arbitrary canonical roots and can materialize one privately through the authoritative connection. Capture runs inside the caller-owned transaction. Restoration runs in a nested savepoint, clears the authoritative resolution and blob caches, and resnapshots to the -selected identity before release. +selected identity before release. Private Workspace transaction bodies finish +their child teardown before final live/current validation; a later effect +coordinator finishes its mutation scope before it invokes capture. Retained roots, manifests and blobs remain indefinitely. Cloudflare garbage collection is not in the production closure and is never invoked. The provider diff --git a/packages/workflow/src/deno/database.ts b/packages/workflow/src/deno/database.ts index 0b45ea9f..d0d26cb9 100644 --- a/packages/workflow/src/deno/database.ts +++ b/packages/workflow/src/deno/database.ts @@ -31,7 +31,7 @@ */ import { randomUUID } from "node:crypto"; -import type { DatabaseSync, StatementSync } from "node:sqlite"; +import type { DatabaseSync } from "node:sqlite"; import { ensure, Err, Ok, type Operation, resource, type Result, scoped } from "effection"; import type { DurableEvent, DurableStream, Json } from "@executablemd/durable-streams"; import type { JournalEntry, WorkflowRunDatabase, WorkflowRunTransaction } from "../storage/api.ts"; @@ -62,6 +62,7 @@ import { useTransactionSavepoints, } from "./transaction.ts"; import { readDocumentExecution, readRetrieval, readRunRecord, stopReasonColumns } from "./rows.ts"; +import { reading } from "./reading.ts"; import { isSqliteForeignKeyConstraint, translateSqliteError } from "./schema.ts"; const SELECT_RUN = "SELECT * FROM workflow_run WHERE id = 1"; @@ -527,20 +528,6 @@ function readRetrievalRow(database: DatabaseSync): DefinitionRetrieval | undefin return row === undefined ? undefined : readRetrieval(row); } -/** - * A statement that answers with `bigint` rather than refusing to answer. - * - * `node:sqlite` throws a `RangeError` when a column holds a 64-bit value — - * and quotes the value in the message. Reading integers as `bigint` puts the - * decision back where every other stored value is decided, in a parser that - * refuses without repeating what it refused. - */ -function reading(database: DatabaseSync, sql: string): StatementSync { - const statement = database.prepare(sql); - statement.setReadBigInts(true); - return statement; -} - function readExecution(database: DatabaseSync, executionId: string): DocumentExecutionRecord { const row = reading(database, SELECT_EXECUTION).get(executionId); if (row === undefined) { diff --git a/packages/workflow/src/deno/provider.ts b/packages/workflow/src/deno/provider.ts index 5fd74c98..6072c88e 100644 --- a/packages/workflow/src/deno/provider.ts +++ b/packages/workflow/src/deno/provider.ts @@ -26,8 +26,9 @@ */ import { dirname, isAbsolute } from "node:path"; +import type { DatabaseSync } from "node:sqlite"; import { ensureDir, exists } from "@effectionx/fs"; -import { ensure, Err, Ok, type Operation, type Result, scoped } from "effection"; +import { createContext, ensure, Err, Ok, type Operation, type Result, scoped } from "effection"; import { type CreateWorkflowRunRequest, type WorkflowRunDatabase, @@ -62,6 +63,7 @@ import { type WorkflowRunConnections, } from "./connections.ts"; import { workflowRunPath } from "./path.ts"; +import { readTransaction } from "./reading.ts"; import { initializeSchema, isUninitialized, translateSqliteError, verifySchema } from "./schema.ts"; const INSERT_RUN = `INSERT INTO workflow_run @@ -80,6 +82,14 @@ export interface WorkflowRunStorageOptions { readonly root: string; } +export type WorkflowRunRecognitionProbe = (database: DatabaseSync) => void; + +/** Adapter-private observation seam for deterministic snapshot tests. */ +export const WorkflowRunRecognition = createContext( + "executablemd.workflow.deno.recognition", + () => {}, +); + /** * Install this host's run storage for the current scope and its descendants. * @@ -156,6 +166,7 @@ function* createWorkflowRun( } try { + const inspectRecognition = yield* WorkflowRunRecognition.expect(); const connection = connections.at(path); const { lock } = connection; // Held across initialization, so a second caller creating the same run @@ -163,7 +174,7 @@ function* createWorkflowRun( // would stop the host while the first one is still committing. const stored = yield* scoped(function* () { yield* lock.hold(); - return establish(connection, path, wanted); + return establish(connection, path, wanted, inspectRecognition); }); if (!stored.ok) { return stored; @@ -205,13 +216,13 @@ function* lookupWorkflowRun( } try { + const inspectRecognition = yield* WorkflowRunRecognition.expect(); const connection = connections.at(path); const { database, lock } = connection; const record = yield* scoped(function* (): Operation> { yield* lock.hold(); try { - verifySchema(database, path, connection.dofs); - return Ok(readRunRow(database, path)); + return Ok(recognizeExisting(connection, path, inspectRecognition)); } catch (error) { return refusal(error, path); } @@ -242,12 +253,16 @@ function establish( connection: RunConnection, path: string, request: CheckedRequest, + inspectRecognition: WorkflowRunRecognitionProbe, ): Result { const { database } = connection; try { - if (!isUninitialized(database, path)) { - verifySchema(database, path, connection.dofs); - } + readTransaction(database, () => { + if (!isUninitialized(database, path)) { + verifySchema(database, path, connection.dofs); + inspectRecognition(database); + } + }); database.exec("BEGIN IMMEDIATE"); connection.transactionOpen = true; @@ -285,6 +300,18 @@ function establish( } } +function recognizeExisting( + connection: RunConnection, + path: string, + inspectRecognition: WorkflowRunRecognitionProbe, +): WorkflowRunRecord { + return readTransaction(connection.database, () => { + verifySchema(connection.database, path, connection.dofs); + inspectRecognition(connection.database); + return readRunRow(connection.database, path); + }); +} + /** * Report a storage refusal as itself, and let anything else propagate. * diff --git a/packages/workflow/src/deno/reading.ts b/packages/workflow/src/deno/reading.ts new file mode 100644 index 00000000..3b148a2c --- /dev/null +++ b/packages/workflow/src/deno/reading.ts @@ -0,0 +1,32 @@ +import type { DatabaseSync, StatementSync } from "node:sqlite"; + +/** + * A statement that returns SQLite integers as `bigint` rather than throwing. + * + * A plain node:sqlite read raises a RangeError, including the stored value in + * its message, when an INTEGER exceeds JavaScript's safe range. Adapter + * parsers need to receive that value so they can refuse it without disclosing + * it. + */ +export function reading(database: DatabaseSync, sql: string): StatementSync { + const statement = database.prepare(sql); + statement.setReadBigInts(true); + return statement; +} + +/** One consistent SQLite snapshot, without advertising a caller-owned write transaction. */ +export function readTransaction(database: DatabaseSync, body: () => T): T { + database.exec("BEGIN"); + try { + const value = body(); + database.exec("COMMIT"); + return value; + } catch (error) { + try { + database.exec("ROLLBACK"); + } catch (rollbackError) { + throw rollbackError; + } + throw error; + } +} diff --git a/packages/workflow/src/deno/schema.ts b/packages/workflow/src/deno/schema.ts index d99ea155..c67f9a97 100644 --- a/packages/workflow/src/deno/schema.ts +++ b/packages/workflow/src/deno/schema.ts @@ -32,6 +32,7 @@ import { WorkflowIncompleteVersionOneError, WorkflowSchemaVersionError, } from "../storage/errors.ts"; +import { reading } from "./reading.ts"; import { initializeEmptyWorkspace, verifyWorkspace } from "./workspace/root.ts"; /** @@ -598,7 +599,7 @@ function readPragmaNumber(database: DatabaseSync, pragma: string, path: string): function query(database: DatabaseSync, sql: string, path: string): Record[] { try { - return database.prepare(sql).all(); + return reading(database, sql).all(); } catch (error) { throw translateSqliteError(error, path); } diff --git a/packages/workflow/src/deno/workspace/private.ts b/packages/workflow/src/deno/workspace/private.ts index fd224439..4588723e 100644 --- a/packages/workflow/src/deno/workspace/private.ts +++ b/packages/workflow/src/deno/workspace/private.ts @@ -1,4 +1,4 @@ -import { type Operation, type Result } from "effection"; +import { type Operation, type Result, scoped } from "effection"; import type { WorkflowRunDatabase } from "../../storage/api.ts"; import { workflowRunConnection } from "../database.ts"; import { createDenoWorkspaceFilesystem, type DenoWorkspaceFilesystem } from "./filesystem.ts"; @@ -42,7 +42,9 @@ export function* transactWorkspaceRoots( return restoreWorkspaceRoot(connection, rootId, options); }, }; - const value = yield* body(workspace); + const value = yield* scoped(function* () { + return yield* body(workspace); + }); verifyWorkspace(connection.database, connection.dofs, connection.path); return value; }); diff --git a/packages/workflow/src/deno/workspace/restore.ts b/packages/workflow/src/deno/workspace/restore.ts index b725ed6f..8a4414e7 100644 --- a/packages/workflow/src/deno/workspace/restore.ts +++ b/packages/workflow/src/deno/workspace/restore.ts @@ -3,6 +3,7 @@ import { clearBlobCache } from "../../../vendor/cloudflare-computer-dofs/generat import { clearResolveCache } from "../../../vendor/cloudflare-computer-dofs/generated/fs/resolveCache.js"; import { WorkflowTransactionError } from "../../storage/errors.ts"; import type { RunConnection } from "../connections.ts"; +import { reading } from "../reading.ts"; import { corrupt, fromHex, @@ -175,7 +176,7 @@ function materializeNode( } function nextRevision(database: DatabaseSync, databasePath: string): number { - const row = database.prepare("UPDATE vfs_meta SET v = v + 1 WHERE k = 'rev' RETURNING v").get(); + const row = reading(database, "UPDATE vfs_meta SET v = v + 1 WHERE k = 'rev' RETURNING v").get(); const revision = integer(row?.["v"], databasePath, "Workspace revision"); if (revision < 1) { corrupt(databasePath, "restoration did not establish a valid Workspace revision"); diff --git a/packages/workflow/src/deno/workspace/root.ts b/packages/workflow/src/deno/workspace/root.ts index 682a04ae..abf22b9b 100644 --- a/packages/workflow/src/deno/workspace/root.ts +++ b/packages/workflow/src/deno/workspace/root.ts @@ -4,6 +4,7 @@ import type { Database as CloudflareDatabase } from "../../../vendor/cloudflare- import { buildManifest } from "../../../vendor/cloudflare-computer-dofs/generated/sync/manifests.js"; import { WorkflowTransactionError } from "../../storage/errors.ts"; import type { RunConnection } from "../connections.ts"; +import { reading } from "../reading.ts"; import { bytes, compareUtf8, @@ -252,9 +253,10 @@ export function retainWorkspaceRoot( corrupt(databasePath, "a Workspace root does not match its canonical content references"); } - const existing = database - .prepare("SELECT format_version, manifest FROM workspace_roots WHERE root_id = ?") - .get(root.rootId); + const existing = reading( + database, + "SELECT format_version, manifest FROM workspace_roots WHERE root_id = ?", + ).get(root.rootId); if (existing === undefined) { database .prepare("INSERT INTO workspace_roots (root_id, format_version, manifest) VALUES (?, ?, ?)") @@ -303,9 +305,10 @@ export function loadWorkspaceRoot( if (!SHA256.test(rootId)) { corrupt(databasePath, "the selected Workspace root identity is malformed"); } - const row = database - .prepare("SELECT root_id, format_version, manifest FROM workspace_roots WHERE root_id = ?") - .get(rootId); + const row = reading( + database, + "SELECT root_id, format_version, manifest FROM workspace_roots WHERE root_id = ?", + ).get(rootId); if (row === undefined) { corrupt(databasePath, "the selected Workspace root is not retained"); } @@ -326,7 +329,7 @@ export function setCurrentWorkspaceRoot( } export function currentWorkspaceRoot(database: DatabaseSync, databasePath: string): string { - const rows = database.prepare("SELECT singleton_id, current_root_id FROM workspace_state").all(); + const rows = reading(database, "SELECT singleton_id, current_root_id FROM workspace_state").all(); const row = rows[0]; if ( rows.length !== 1 || @@ -346,9 +349,10 @@ export function verifyWorkspace( ): void { validateDofsContentStore(database, databasePath); const retained = new Map(); - for (const row of database - .prepare("SELECT root_id, format_version, manifest FROM workspace_roots ORDER BY root_id") - .all()) { + for (const row of reading( + database, + "SELECT root_id, format_version, manifest FROM workspace_roots ORDER BY root_id", + ).all()) { const root = parseStoredRoot(database, row, databasePath); if (retained.has(root.rootId)) { corrupt(databasePath, "it contains a duplicate retained Workspace root"); @@ -375,14 +379,13 @@ export function verifyWorkspace( corrupt(databasePath, "it contains no retained Workspace root"); } - const unretainedJournalRoots = database - .prepare( - `SELECT COUNT(*) AS count + const unretainedJournalRoots = reading( + database, + `SELECT COUNT(*) AS count FROM journal_events AS event LEFT JOIN workspace_roots AS root ON root.root_id = event.workspace_root_id WHERE root.root_id IS NULL`, - ) - .get(); + ).get(); if (integer(unretainedJournalRoots?.["count"], databasePath, "journal root count") !== 0) { corrupt(databasePath, "a journal event names a Workspace root that is not retained"); } @@ -403,9 +406,10 @@ export function readDofsManifest( databasePath: string, ): DofsManifest { const hashBytes = fromHex(hash, databasePath, "DOFS manifest identity"); - const row = database - .prepare("SELECT hash, size, encoded, last_seen FROM vfs_manifests WHERE hash = ?") - .get(hashBytes); + const row = reading( + database, + "SELECT hash, size, encoded, last_seen FROM vfs_manifests WHERE hash = ?", + ).get(hashBytes); if (row === undefined) { corrupt(databasePath, "a retained Workspace root names a missing DOFS manifest"); } @@ -532,7 +536,10 @@ function validateFile( } function validateDofsContentStore(database: DatabaseSync, databasePath: string): void { - const blobs = database.prepare("SELECT hash, size, last_seen FROM vfs_blobs ORDER BY hash").all(); + const blobs = reading( + database, + "SELECT hash, size, last_seen FROM vfs_blobs ORDER BY hash", + ).all(); for (const row of blobs) { const hash = bytes(row["hash"], databasePath, "DOFS blob hash"); if (hash.byteLength !== 32) { @@ -550,7 +557,7 @@ function validateDofsContentStore(database: DatabaseSync, databasePath: string): corrupt(databasePath, "the DOFS blob index and retained bytes are incomplete"); } - for (const row of database.prepare("SELECT hash FROM vfs_manifests ORDER BY hash").all()) { + for (const row of reading(database, "SELECT hash FROM vfs_manifests ORDER BY hash").all()) { const hash = bytes(row["hash"], databasePath, "DOFS manifest hash"); if (hash.byteLength !== 32) { corrupt(databasePath, "a DOFS manifest has an invalid hash length"); @@ -566,14 +573,13 @@ function validateBlob( databasePath: string, ): string { const hashBytes = fromHex(hash, databasePath, "DOFS blob identity"); - const row = database - .prepare( - `SELECT blob.hash, blob.size, blob.last_seen, content.bytes + const row = reading( + database, + `SELECT blob.hash, blob.size, blob.last_seen, content.bytes FROM vfs_blobs AS blob JOIN vfs_blob_bytes AS content ON content.hash = blob.hash WHERE blob.hash = ?`, - ) - .get(hashBytes); + ).get(hashBytes); if (row === undefined) { corrupt(databasePath, "a Workspace file names missing DOFS blob bytes"); } @@ -599,13 +605,12 @@ function readNode( currentRev: number, databasePath: string, ): NodeRow { - const row = database - .prepare( - `SELECT inode, type, mode, mtime, rev, mount_root, stub_size, + const row = reading( + database, + `SELECT inode, type, mode, mtime, rev, mount_root, stub_size, manifest_hash, link_target, size FROM vfs_nodes WHERE inode = ?`, - ) - .get(inode); + ).get(inode); if (row === undefined) { corrupt(databasePath, "its live Workspace contains a dangling directory entry"); } @@ -665,9 +670,10 @@ function readDirents( databasePath: string, ): Array<{ name: string; inode: number }> { const entries: Array<{ name: string; inode: number }> = []; - for (const row of database - .prepare("SELECT name, child_inode FROM vfs_dirents WHERE parent_inode = ?") - .all(inode)) { + for (const row of reading( + database, + "SELECT name, child_inode FROM vfs_dirents WHERE parent_inode = ?", + ).all(inode)) { const name = row["name"]; if (typeof name !== "string") { corrupt(databasePath, "its live Workspace contains an invalid directory-entry name"); @@ -683,8 +689,10 @@ function readDirents( function readChunks(database: DatabaseSync, inode: number, databasePath: string): DofsChunk[] { const chunks: DofsChunk[] = []; - for (const [expected, row] of database - .prepare("SELECT idx, hash, size FROM vfs_chunks WHERE inode = ? ORDER BY idx") + for (const [expected, row] of reading( + database, + "SELECT idx, hash, size FROM vfs_chunks WHERE inode = ? ORDER BY idx", + ) .all(inode) .entries()) { const index = nonnegative(row["idx"], databasePath, "Workspace chunk index"); @@ -700,7 +708,7 @@ function readChunks(database: DatabaseSync, inode: number, databasePath: string) } function validateDofsBookkeeping(database: DatabaseSync, databasePath: string): number { - const metadata = database.prepare("SELECT k, v FROM vfs_meta ORDER BY k").all(); + const metadata = reading(database, "SELECT k, v FROM vfs_meta ORDER BY k").all(); if ( metadata.length !== 2 || metadata[0]?.["k"] !== "rev" || @@ -714,7 +722,7 @@ function validateDofsBookkeeping(database: DatabaseSync, databasePath: string): corrupt(databasePath, "its Workspace revision is not initialized"); } - const watermarks = database.prepare("SELECT k, backend, v FROM _vfs_watermark ORDER BY k").all(); + const watermarks = reading(database, "SELECT k, backend, v FROM _vfs_watermark ORDER BY k").all(); if ( watermarks.length !== 2 || watermarks[0]?.["k"] !== "fetchRev" || @@ -726,7 +734,7 @@ function validateDofsBookkeeping(database: DatabaseSync, databasePath: string): ) { corrupt(databasePath, "its Workspace synchronization watermarks are malformed"); } - const cursors = database.prepare("SELECT k, backend, path FROM _vfs_fetch_cursor").all(); + const cursors = reading(database, "SELECT k, backend, path FROM _vfs_fetch_cursor").all(); if ( cursors.length !== 1 || cursors[0]?.["k"] !== "fetch" || @@ -739,9 +747,10 @@ function validateDofsBookkeeping(database: DatabaseSync, databasePath: string): corrupt(databasePath, "its retained Workspace contains an unsupported mount"); } - for (const row of database - .prepare("SELECT id, rev, path, op FROM vfs_changes ORDER BY id") - .all()) { + for (const row of reading( + database, + "SELECT id, rev, path, op FROM vfs_changes ORDER BY id", + ).all()) { const id = nonnegative(row["id"], databasePath, "Workspace change identity"); const changeRev = nonnegative(row["rev"], databasePath, "Workspace change revision"); if ( @@ -766,8 +775,7 @@ function requireReferenceSet( expected: readonly string[], databasePath: string, ): void { - const actual = database - .prepare(`SELECT ${column} FROM ${table} WHERE root_id = ?`) + const actual = reading(database, `SELECT ${column} FROM ${table} WHERE root_id = ?`) .all(rootId) .map((row) => toHex(bytes(row[column], databasePath, `${table}.${column}`))) .sort(compareUtf8); @@ -777,7 +785,7 @@ function requireReferenceSet( } function count(database: DatabaseSync, table: string, databasePath: string): number { - const row = database.prepare(`SELECT COUNT(*) AS count FROM ${table}`).get(); + const row = reading(database, `SELECT COUNT(*) AS count FROM ${table}`).get(); return nonnegative(row?.["count"], databasePath, `${table} row count`); } diff --git a/packages/workflow/tests/workflow-run-storage.test.ts b/packages/workflow/tests/workflow-run-storage.test.ts index b528df75..f71a2e0e 100644 --- a/packages/workflow/tests/workflow-run-storage.test.ts +++ b/packages/workflow/tests/workflow-run-storage.test.ts @@ -43,6 +43,7 @@ import { import type { JsonObject } from "../src/storage/members.ts"; import { APPLICATION_ID, hashRunId, useWorkflowRunStorage } from "../deno.ts"; import { createWorkflowRunConnections } from "../src/deno/connections.ts"; +import { WorkflowRunRecognition } from "../src/deno/provider.ts"; import { EXPECTED_SCHEMA, initializeSchema } from "../src/deno/schema.ts"; import { EMPTY_WORKSPACE_MANIFEST, @@ -261,6 +262,73 @@ describe("Tier WS — authoritative connection and complete schema", () => { database.close(); } }); + + it("WS0e: existing-run recognition and its run row share one SQLite snapshot", function* () { + const root = yield* useStorageRoot(); + const path = runPath(root, "recognition-snapshot"); + yield* withStorage(root, function* () { + yield* createRun({ runId: "recognition-snapshot" }); + }); + + const writer = new DatabaseSync(path); + writer.exec("PRAGMA journal_mode = WAL"); + let observedTransaction = false; + let probes = 0; + try { + const found = yield* scoped(function* () { + yield* WorkflowRunRecognition.set((reader) => { + probes += 1; + observedTransaction = reader.isTransaction; + writer + .prepare("UPDATE workflow_run SET run_id = ? WHERE id = 1") + .run("committed-after-validation"); + }); + return yield* withStorage(root, function* () { + return yield* lookup("recognition-snapshot"); + }); + }); + + expect(found.ok).toBe(true); + expect(found.ok && found.value.record.runId).toBe("recognition-snapshot"); + expect(probes).toBe(1); + expect(observedTransaction).toBe(true); + expect(writer.prepare("SELECT run_id FROM workflow_run WHERE id = 1").get()?.["run_id"]).toBe( + "committed-after-validation", + ); + } finally { + writer.close(); + } + }); + + it("WS0f: a recognition failure closes its SQLite read transaction", function* () { + const root = yield* useStorageRoot(); + const path = runPath(root, "recognition-rollback"); + yield* withStorage(root, function* () { + yield* createRun({ runId: "recognition-rollback" }); + }); + const before = readFileSync(path); + let observed: DatabaseSync | undefined; + + const raised = yield* scoped(function* () { + yield* WorkflowRunRecognition.set((reader) => { + observed = reader; + throw new Error("recognition probe failed"); + }); + return yield* withStorage(root, function* () { + let failure: unknown; + try { + yield* lookup("recognition-rollback"); + } catch (error) { + failure = error; + } + expect(observed?.isTransaction).toBe(false); + return failure; + }); + }); + + expect(raised).toBeInstanceOf(Error); + expect(readFileSync(path)).toEqual(before); + }); }); describe("Tier WS — creating and finding a run", () => { diff --git a/packages/workflow/tests/workspace-root-restoration.test.ts b/packages/workflow/tests/workspace-root-restoration.test.ts index 6b4bdc2e..e72f4bb0 100644 --- a/packages/workflow/tests/workspace-root-restoration.test.ts +++ b/packages/workflow/tests/workspace-root-restoration.test.ts @@ -1,8 +1,9 @@ import { readFileSync } from "node:fs"; import { DatabaseSync } from "node:sqlite"; +import { inspect } from "node:util"; import { describe, it } from "@executablemd/test-support/bdd"; import { expect } from "@executablemd/test-support/expect"; -import { type Operation } from "effection"; +import { ensure, type Operation, spawn, suspend, withResolvers } from "effection"; import { WorkflowDatabaseCorruptError, type WorkflowRunDatabase, @@ -17,6 +18,8 @@ import { } from "../src/deno/workspace/private.ts"; import { createRun, runPath, tamper, useStorageRoot, withStorage } from "./support/storage.ts"; +const SQLITE_MAX_INTEGER = "9223372036854775807"; + function* transact( database: WorkflowRunDatabase, body: (workspace: PrivateWorkspaceTransaction) => Operation, @@ -111,8 +114,8 @@ describe("Tier WRR — private Workspace root restoration", () => { size: 10, }); expect( - (yield* workspace.filesystem.readdir("/tree")).map((entry) => entry.name).toSorted(), - ).toEqual(["current.txt", "file.txt", "hardlink.txt"]); + new Set((yield* workspace.filesystem.readdir("/tree")).map((entry) => entry.name)), + ).toEqual(new Set(["current.txt", "file.txt", "hardlink.txt"])); const resnapshot = yield* workspace.capture({ publish: true }); expect(resnapshot).toEqual(selected); return selected; @@ -375,8 +378,185 @@ describe("Tier WRR — private Workspace root restoration", () => { expect(readFileSync(path)).toEqual(before); } }); + + it("WRR7: retained-root integer corruption is redacted, typed, and read-only", function* () { + const storage = yield* useStorageRoot(); + const cases: Array<{ runId: string; damage(database: DatabaseSync): void }> = [ + { + runId: "huge-root-format", + damage(database) { + database.exec("PRAGMA ignore_check_constraints = ON"); + database.exec(`UPDATE workspace_roots SET format_version = ${SQLITE_MAX_INTEGER}`); + }, + }, + { + runId: "huge-root-singleton", + damage(database) { + database.exec("PRAGMA ignore_check_constraints = ON"); + database.exec(`UPDATE workspace_state SET singleton_id = ${SQLITE_MAX_INTEGER}`); + }, + }, + { + runId: "huge-node-mode", + damage(database) { + database.exec(`UPDATE vfs_nodes SET mode = ${SQLITE_MAX_INTEGER} WHERE inode = 1`); + }, + }, + { + runId: "huge-chunk-index", + damage(database) { + database.exec(`UPDATE vfs_chunks SET idx = ${SQLITE_MAX_INTEGER}`); + }, + }, + { + runId: "huge-manifest-size", + damage(database) { + database.exec(`UPDATE vfs_manifests SET size = ${SQLITE_MAX_INTEGER}`); + }, + }, + { + runId: "huge-blob-last-seen", + damage(database) { + database.exec(`UPDATE vfs_blobs SET last_seen = ${SQLITE_MAX_INTEGER}`); + }, + }, + { + runId: "huge-revision", + damage(database) { + database.exec(`UPDATE vfs_meta SET v = ${SQLITE_MAX_INTEGER} WHERE k = 'rev'`); + }, + }, + { + runId: "huge-watermark", + damage(database) { + database.exec(`UPDATE _vfs_watermark SET v = ${SQLITE_MAX_INTEGER} WHERE k = 'fetchRev'`); + }, + }, + { + runId: "huge-change-id", + damage(database) { + database.exec( + `INSERT INTO vfs_changes (id, rev, path, op) + VALUES (${SQLITE_MAX_INTEGER}, 1, '/gone', 'delete')`, + ); + }, + }, + ]; + + for (const one of cases) { + yield* createCorruptionFixture(storage, one.runId); + const path = runPath(storage, one.runId); + tamper(path, one.damage); + const before = readFileSync(path); + const result = yield* withStorage(storage, function* () { + return yield* WorkflowRunStorage.operations.lookup(one.runId); + }); + + expect(result.ok).toBe(false); + if (result.ok) { + continue; + } + expect(result.error).toBeInstanceOf(WorkflowDatabaseCorruptError); + expect(result.error).not.toBeInstanceOf(RangeError); + expect(errorSurface(result.error)).not.toContain(SQLITE_MAX_INTEGER); + expect(readFileSync(path)).toEqual(before); + } + }); + + it("WRR8: cleanup mutations finish before final root validation and cannot commit", function* () { + const storage = yield* useStorageRoot(); + const path = runPath(storage, "cleanup-mutation"); + let baselineRoot = ""; + let baselineRoots = 0; + let baselineJournal = 0; + + yield* withStorage(storage, function* () { + const database = yield* createRun({ runId: "cleanup-mutation" }); + baselineRoot = (yield* capture(database, function* (workspace) { + yield* workspace.filesystem.writeFile("/baseline.txt", "baseline"); + })).rootId; + yield* database.journal.append({ + type: "close", + coroutineId: "root", + result: { status: "ok", value: "baseline" }, + }); + + const connection = workflowRunConnection(database); + baselineRoots = count(connection.database, "workspace_roots"); + baselineJournal = count(connection.database, "journal_events"); + const started = withResolvers(); + const result = yield* transactWorkspaceRoots(database, function* (workspace) { + yield* workspace.filesystem.writeFile("/captured.txt", "captured"); + yield* spawn(function* () { + yield* ensure(function* () { + yield* workspace.filesystem.writeFile("/cleanup.txt", "cleanup"); + }); + started.resolve(); + yield* suspend(); + }); + yield* started.operation; + yield* workspace.capture({ publish: true }); + }); + + expect(result.ok).toBe(false); + expect(!result.ok && result.error).toBeInstanceOf(WorkflowDatabaseCorruptError); + expect(count(connection.database, "workspace_roots")).toBe(baselineRoots); + expect(count(connection.database, "journal_events")).toBe(baselineJournal); + + const observed = yield* transact(database, function* (workspace) { + const missing: string[] = []; + for (const file of ["/captured.txt", "/cleanup.txt"]) { + try { + yield* workspace.filesystem.readTextFile(file); + } catch { + missing.push(file); + } + } + return { root: yield* workspace.currentRoot(), missing }; + }); + expect(observed).toEqual({ + root: baselineRoot, + missing: ["/captured.txt", "/cleanup.txt"], + }); + expect(yield* database.journal.readAll()).toHaveLength(baselineJournal); + }); + + yield* withStorage(storage, function* () { + const found = yield* WorkflowRunStorage.operations.lookup("cleanup-mutation"); + if (!found.ok) { + throw found.error; + } + expect( + yield* transact(found.value, function* (workspace) { + expect(yield* workspace.filesystem.readTextFile("/baseline.txt")).toBe("baseline"); + return yield* workspace.currentRoot(); + }), + ).toBe(baselineRoot); + expect(yield* found.value.journal.readAll()).toHaveLength(baselineJournal); + }); + + const reopened = new DatabaseSync(path); + try { + expect(count(reopened, "workspace_roots")).toBe(baselineRoots); + expect(count(reopened, "journal_events")).toBe(baselineJournal); + } finally { + reopened.close(); + } + }); }); +function errorSurface(error: unknown): string { + const parts: string[] = [String(error), inspect(error)]; + const seen = new Set(); + let current = error; + while (current instanceof Error && !seen.has(current)) { + seen.add(current); + parts.push(current.name, current.message, current.stack ?? ""); + current = current.cause; + } + return parts.join("\n"); +} + function currentRoot(database: DatabaseSync): string { return String( database.prepare("SELECT current_root_id FROM workspace_state WHERE singleton_id = 1").get()?.[ diff --git a/packages/workflow/tests/workspace-root.test.ts b/packages/workflow/tests/workspace-root.test.ts index ea44a3f4..883fea2b 100644 --- a/packages/workflow/tests/workspace-root.test.ts +++ b/packages/workflow/tests/workspace-root.test.ts @@ -285,7 +285,7 @@ describe("Tier WRR — immutable retained Workspace roots", () => { } }); - it("WRR7: the production closure contains no DOFS garbage-collection path", function* () { + it("WRR9: the production closure contains no DOFS garbage-collection path", function* () { // deno-lint-ignore require-yield const denoAdapter = fileURLToPath(new URL("../src/deno", import.meta.url)); const vendorManifest = fileURLToPath( diff --git a/specs/workflow-spec.md b/specs/workflow-spec.md index 13693fef..ad17e9c6 100644 --- a/specs/workflow-spec.md +++ b/specs/workflow-spec.md @@ -328,6 +328,12 @@ connection queue and one synchronous savepoint allocator. It remains alive until provider-scope teardown, after the provider's child scopes finish. Different database paths have independent entries. +Opening existing storage performs structural recognition, retained-root and +content validation, the live/current comparison and the singleton run-row read +inside one explicit SQLite read transaction. Those dependent reads therefore +describe one committed version. Recognition does not mark the connection as a +caller-owned Workspace transaction and does not permit DOFS savepoints. + Operations through every lease on one entry are serialized, and each runs inside a transaction. A caller that needs several statements published together holds the transaction itself: @@ -366,7 +372,13 @@ transaction. Capture traverses and validates the complete live DOFS frontier, builds or reuses a canonical DOFS file manifest when ordered chunks do not yet have one, retains the immutable root and exact reference sets, and optionally sets it current. Read-only recognition never builds a manifest or changes -last-seen metadata. +last-seen metadata. The supplied private Workspace body runs in an inner scope, +and final live/current validation waits for that scope's children and resources +to finish teardown. + +Successful effect coordination finishes its mutation scope before capturing +the root. The provider-level coordinator that orders mutation teardown, root +capture and filtered journal publication is not part of this storage layer. The private restoration materializer loads a fully validated retained root and rebuilds directories, files, chunks, modes, mtimes, symbolic links and hardlink From e62ecc7c0d22144e464484f249ce52c61c7e838d Mon Sep 17 00:00:00 2001 From: Taras Mankovski <74687+taras@users.noreply.github.com> Date: Sat, 8 Aug 2026 17:35:07 -0400 Subject: [PATCH 3/3] =?UTF-8?q?=F0=9F=9B=A1=EF=B8=8F=20Keep=20retained=20W?= =?UTF-8?q?orkspace=20rollback=20cache-coherent?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- architecture.md | 12 ++ packages/workflow/src/deno/connections.ts | 7 + packages/workflow/src/deno/database.ts | 1 + .../workflow/src/deno/workspace/restore.ts | 17 +- packages/workflow/src/deno/workspace/root.ts | 21 ++- .../tests/workspace-root-restoration.test.ts | 159 +++++++++++++++++- specs/executable-mdx-spec.md | 18 ++ specs/workflow-spec.md | 15 +- 8 files changed, 229 insertions(+), 21 deletions(-) diff --git a/architecture.md b/architecture.md index 5a10aac4..87eb1a68 100644 --- a/architecture.md +++ b/architecture.md @@ -227,6 +227,11 @@ and run row are read through one explicit SQLite snapshot. This recognition transaction is not a caller-owned Workspace transaction and enables no DOFS savepoints. +Recognition also requires every file entry in every retained root, including a +historical root, to declare the size held by its referenced DOFS manifest. A +root whose XMD identity and reference rows are internally consistent but whose +file size disagrees with DOFS is corrupt and not restorable. + Which status transitions are legal, and what a caller may do to a run in each of them, is lifecycle policy applied above storage. @@ -426,6 +431,13 @@ selected identity before release. Private Workspace transaction bodies finish their child teardown before final live/current validation; a later effect coordinator finishes its mutation scope before it invokes capture. +Every unsuccessful caller-owned transaction attempts SQLite rollback and then +invalidates both caches on the provider-owned DOFS wrapper while it still holds +the serialized connection turn. This includes body failure, cancellation, +teardown or final-validation failure, and commit failure. The surviving wrapper +therefore cannot answer from uncommitted positive or negative cache entries +after SQLite has restored the prior frontier. + Retained roots, manifests and blobs remain indefinitely. Cloudflare garbage collection is not in the production closure and is never invoked. The provider exposes no public Workspace mutation effect, history selection or fork diff --git a/packages/workflow/src/deno/connections.ts b/packages/workflow/src/deno/connections.ts index e107751b..01516f66 100644 --- a/packages/workflow/src/deno/connections.ts +++ b/packages/workflow/src/deno/connections.ts @@ -2,6 +2,8 @@ import { DatabaseSync } from "node:sqlite"; import { resolve } from "node:path"; import { Database as CloudflareDatabase } from "../../vendor/cloudflare-computer-dofs/generated/storage.js"; import { WorkspaceFilesystem } from "../../vendor/cloudflare-computer-dofs/generated/fs/filesystem.js"; +import { clearBlobCache } from "../../vendor/cloudflare-computer-dofs/generated/fs/blobCache.js"; +import { clearResolveCache } from "../../vendor/cloudflare-computer-dofs/generated/fs/resolveCache.js"; import type { DurableObjectStorageLike, SQLCursorLike, @@ -18,6 +20,7 @@ export interface RunConnection { readonly lock: ConnectionLock; readonly savepoints: SavepointManager; transactionOpen: boolean; + invalidateDofsCaches(): void; setClock(now: () => number): void; close(): void; } @@ -96,6 +99,10 @@ function createConnection(path: string): RunConnection { set transactionOpen(value: boolean) { connection.transactionOpen = value; }, + invalidateDofsCaches(): void { + clearResolveCache(dofs); + clearBlobCache(dofs); + }, setClock(now: () => number): void { clock = now; }, diff --git a/packages/workflow/src/deno/database.ts b/packages/workflow/src/deno/database.ts index d0d26cb9..21ac65a9 100644 --- a/packages/workflow/src/deno/database.ts +++ b/packages/workflow/src/deno/database.ts @@ -220,6 +220,7 @@ function createHandle(connection: OpenConnection): Handle { connection.connection.transactionOpen = false; if (!committed) { rollback(database); + connection.connection.invalidateDofsCaches(); } }); diff --git a/packages/workflow/src/deno/workspace/restore.ts b/packages/workflow/src/deno/workspace/restore.ts index 8a4414e7..04f00d3e 100644 --- a/packages/workflow/src/deno/workspace/restore.ts +++ b/packages/workflow/src/deno/workspace/restore.ts @@ -1,6 +1,4 @@ import type { DatabaseSync } from "node:sqlite"; -import { clearBlobCache } from "../../../vendor/cloudflare-computer-dofs/generated/fs/blobCache.js"; -import { clearResolveCache } from "../../../vendor/cloudflare-computer-dofs/generated/fs/resolveCache.js"; import { WorkflowTransactionError } from "../../storage/errors.ts"; import type { RunConnection } from "../connections.ts"; import { reading } from "../reading.ts"; @@ -40,11 +38,11 @@ export function restoreWorkspaceRoot( const { database, dofs, path, savepoints } = connection; verifyWorkspace(database, dofs, path); const selected = loadWorkspaceRoot(database, rootId, path); - clearCaches(connection); + connection.invalidateDofsCaches(); try { return savepoints.synchronous(() => { rebuild(database, selected, path); - clearCaches(connection); + connection.invalidateDofsCaches(); const restored = snapshotWorkspace(database, dofs, path, false); if ( restored.rootId !== selected.rootId || @@ -60,14 +58,10 @@ export function restoreWorkspaceRoot( return selected; }); } finally { - clearCaches(connection); + connection.invalidateDofsCaches(); } } -export function clearWorkspaceCaches(connection: RunConnection): void { - clearCaches(connection); -} - function rebuild(database: DatabaseSync, root: StoredWorkspaceRoot, databasePath: string): void { const parsed = parseWorkspaceManifest(root.manifest, databasePath); const rootEntry = parsed.entries[0]; @@ -184,11 +178,6 @@ function nextRevision(database: DatabaseSync, databasePath: string): number { return revision; } -function clearCaches(connection: RunConnection): void { - clearResolveCache(connection.dofs); - clearBlobCache(connection.dofs); -} - function equalStrings(left: readonly string[], right: readonly string[]): boolean { return JSON.stringify(left) === JSON.stringify(right); } diff --git a/packages/workflow/src/deno/workspace/root.ts b/packages/workflow/src/deno/workspace/root.ts index abf22b9b..8a298bd9 100644 --- a/packages/workflow/src/deno/workspace/root.ts +++ b/packages/workflow/src/deno/workspace/root.ts @@ -474,19 +474,30 @@ function rootFromManifest( parsed: ReturnType, databasePath: string, ): StoredWorkspaceRoot { - const manifests = new Set(); + const manifests = new Map(); for (const entry of parsed.entries) { if (entry.kind === "file") { - manifests.add(entry.manifest); + let manifest = manifests.get(entry.manifest); + if (manifest === undefined) { + manifest = readDofsManifest(database, entry.manifest, databasePath); + manifests.set(entry.manifest, manifest); + } + if (entry.size !== manifest.size) { + corrupt(databasePath, "a retained file size differs from its DOFS manifest"); + } } } const blobs = new Set(); - for (const hash of manifests) { - for (const chunk of readDofsManifest(database, hash, databasePath).chunks) { + for (const manifest of manifests.values()) { + for (const chunk of manifest.chunks) { blobs.add(chunk.hash); } } - return workspaceRoot(manifest, [...manifests].sort(compareUtf8), [...blobs].sort(compareUtf8)); + return workspaceRoot( + manifest, + [...manifests.keys()].sort(compareUtf8), + [...blobs].sort(compareUtf8), + ); } function validateFile( diff --git a/packages/workflow/tests/workspace-root-restoration.test.ts b/packages/workflow/tests/workspace-root-restoration.test.ts index e72f4bb0..734d2205 100644 --- a/packages/workflow/tests/workspace-root-restoration.test.ts +++ b/packages/workflow/tests/workspace-root-restoration.test.ts @@ -10,7 +10,12 @@ import { WorkflowRunStorage, } from "../mod.ts"; import { workflowRunConnection } from "../src/deno/database.ts"; -import { type StoredWorkspaceRoot } from "../src/deno/workspace/manifest.ts"; +import { + encodeWorkspaceManifest, + parseWorkspaceManifest, + type StoredWorkspaceRoot, + workspaceRootId, +} from "../src/deno/workspace/manifest.ts"; import { type PrivateWorkspaceTransaction, setPrivateWorkspaceClock, @@ -543,6 +548,158 @@ describe("Tier WRR — private Workspace root restoration", () => { reopened.close(); } }); + + it("WRR10: outer failure rollback invalidates authoritative DOFS caches", function* () { + const storage = yield* useStorageRoot(); + + yield* withStorage(storage, function* () { + const database = yield* createRun({ runId: "rollback-cache" }); + const baselineRoot = (yield* capture(database, function* (workspace) { + yield* workspace.filesystem.writeFile("/kept.txt", "known retained bytes"); + })).rootId; + yield* database.journal.append({ + type: "close", + coroutineId: "root", + result: { status: "ok", value: "baseline" }, + }); + const baselineJournal = yield* database.journal.readAll(); + + const failed = yield* transactWorkspaceRoots(database, function* (workspace) { + yield* workspace.filesystem.remove("/kept.txt"); + let missing: unknown; + try { + yield* workspace.filesystem.readTextFile("/kept.txt"); + } catch (error) { + missing = error; + } + expect(missing).toBeInstanceOf(Error); + throw new Error("force the caller-owned transaction to roll back"); + }); + expect(failed.ok).toBe(false); + + const observed = yield* transact(database, function* (workspace) { + return { + content: yield* workspace.filesystem.readTextFile("/kept.txt"), + root: yield* workspace.currentRoot(), + }; + }); + expect(observed).toEqual({ content: "known retained bytes", root: baselineRoot }); + expect(yield* database.journal.readAll()).toEqual(baselineJournal); + }); + }); + + it("WRR10b: outer cancellation rollback invalidates authoritative DOFS caches", function* () { + const storage = yield* useStorageRoot(); + + yield* withStorage(storage, function* () { + const database = yield* createRun({ runId: "cancel-cache" }); + const baselineRoot = (yield* capture(database, function* (workspace) { + yield* workspace.filesystem.writeFile("/kept.txt", "known retained bytes"); + })).rootId; + yield* database.journal.append({ + type: "close", + coroutineId: "root", + result: { status: "ok", value: "baseline" }, + }); + const baselineJournal = yield* database.journal.readAll(); + const reached = withResolvers(); + + const transacting = yield* spawn(function* () { + yield* transactWorkspaceRoots(database, function* (workspace) { + yield* workspace.filesystem.remove("/kept.txt"); + let missing: unknown; + try { + yield* workspace.filesystem.readTextFile("/kept.txt"); + } catch (error) { + missing = error; + } + expect(missing).toBeInstanceOf(Error); + reached.resolve(); + yield* suspend(); + }); + }); + yield* reached.operation; + yield* transacting.halt(); + + const observed = yield* transact(database, function* (workspace) { + return { + content: yield* workspace.filesystem.readTextFile("/kept.txt"), + root: yield* workspace.currentRoot(), + }; + }); + expect(observed).toEqual({ content: "known retained bytes", root: baselineRoot }); + expect(yield* database.journal.readAll()).toEqual(baselineJournal); + }); + }); + + it("WRR11: historical file sizes must agree with retained DOFS manifests", function* () { + const storage = yield* useStorageRoot(); + const path = runPath(storage, "historical-size"); + let historicalRoot = ""; + let currentRootId = ""; + + yield* withStorage(storage, function* () { + const database = yield* createRun({ runId: "historical-size" }); + historicalRoot = (yield* capture(database, function* (workspace) { + yield* workspace.filesystem.writeFile("/historical.txt", "historical bytes"); + })).rootId; + yield* database.journal.append({ + type: "close", + coroutineId: "root", + result: { status: "ok", value: "historical" }, + }); + currentRootId = (yield* capture(database, function* (workspace) { + yield* workspace.filesystem.remove("/historical.txt"); + yield* workspace.filesystem.writeFile("/current.txt", "current bytes"); + })).rootId; + }); + + tamper(path, (database) => { + const stored = database + .prepare("SELECT manifest FROM workspace_roots WHERE root_id = ?") + .get(historicalRoot)?.["manifest"]; + if (typeof stored !== "string") { + throw new Error("the historical Workspace root is missing"); + } + const parsed = parseWorkspaceManifest(stored, path); + let changed = false; + const entries = parsed.entries.map((entry) => { + if (entry.kind !== "file") { + return entry; + } + changed = true; + return { ...entry, size: entry.size + 1 }; + }); + if (!changed) { + throw new Error("the historical Workspace root contains no file"); + } + const manifest = encodeWorkspaceManifest(entries, path); + const rootId = workspaceRootId(manifest); + + database.exec("PRAGMA foreign_keys = OFF"); + database + .prepare("UPDATE workspace_roots SET root_id = ?, manifest = ? WHERE root_id = ?") + .run(rootId, manifest, historicalRoot); + database + .prepare("UPDATE workspace_root_manifest_refs SET root_id = ? WHERE root_id = ?") + .run(rootId, historicalRoot); + database + .prepare("UPDATE workspace_root_blob_refs SET root_id = ? WHERE root_id = ?") + .run(rootId, historicalRoot); + database + .prepare("UPDATE journal_events SET workspace_root_id = ? WHERE workspace_root_id = ?") + .run(rootId, historicalRoot); + expect(currentRoot(database)).toBe(currentRootId); + }); + + const before = readFileSync(path); + const result = yield* withStorage(storage, function* () { + return yield* WorkflowRunStorage.operations.lookup("historical-size"); + }); + expect(result.ok).toBe(false); + expect(!result.ok && result.error).toBeInstanceOf(WorkflowDatabaseCorruptError); + expect(readFileSync(path)).toEqual(before); + }); }); function errorSurface(error: unknown): string { diff --git a/specs/executable-mdx-spec.md b/specs/executable-mdx-spec.md index b1e2894a..6e3c3d7b 100644 --- a/specs/executable-mdx-spec.md +++ b/specs/executable-mdx-spec.md @@ -6565,6 +6565,24 @@ Defined in [Workflow runs](./workflow-spec.md) §9.5–§9.6. | WJ24 | Two processes | Two real processes racing to create one run leave one winner and one conflict | | WJ25 | A second process | Restores the run, preserves journal order and identity, and performs no recorded operation again | +### Tier WRR — Immutable retained Workspace roots + +Defined in [Workflow runs](./workflow-spec.md) §9.4 and §9.6–§9.7. + +| # | Test | Verify | +|---|------|--------| +| WRR1 | Canonical empty root | Fresh complete-v1 storage retains the exact root-only canonical manifest and its content-addressed identity | +| WRR2 | Complete canonical topology | Paths, metadata, symlinks and deterministic hardlinks form canonical bytes; DOFS manifests and blobs are retained exactly without copied file bytes | +| WRR3 | Mutation-derived roots | Create, overwrite, delete, rename, directory, mode, symlink and hardlink changes produce the corresponding immutable roots | +| WRR4 | Historical restoration | An older root restores exact topology and content, resnapshots to its identity and clears authoritative negative caches | +| WRR5 | Restoration rollback | A restoration failure rolls back its savepoint and preserves the prior live frontier and current-root pointer | +| WRR6 | Read-only corruption | Root, reference, content, chunk, topology and live/current corruption is refused without changing the database | +| WRR7 | Bigint-safe corruption | Maximum-width SQLite integers produce a redacted `WorkflowDatabaseCorruptError`, never a raw or value-leaking conversion failure | +| WRR8 | Teardown before validation | Child cleanup finishes before final live/current validation, and a stale capture cannot commit | +| WRR9 | No unsafe garbage collection | The production closure neither exposes nor invokes Cloudflare DOFS garbage collection | +| WRR10/WRR10b | Outer rollback cache coherence | Failure and cancellation after an uncommitted removal and negative lookup roll back and invalidate both authoritative DOFS caches | +| WRR11 | Historical file size | Every historical file entry's declared size agrees with its retained DOFS manifest during read-only recognition | + ### Tier SL — Own-scope context updates | # | Test | Verify | diff --git a/specs/workflow-spec.md b/specs/workflow-spec.md index ad17e9c6..e3449b95 100644 --- a/specs/workflow-spec.md +++ b/specs/workflow-spec.md @@ -293,6 +293,11 @@ blobs. The normalized root-to-manifest and root-to-blob rows equal the exact transitive content of each root and prevent that content from being deleted while the root is retained. +Each file entry's declared size equals the size in its referenced DOFS +manifest. Recognition checks that agreement for every file in every retained +root, including roots that are not current, while it validates the manifest's +chunks and blobs transitively. + ### 9.5 The journal `WorkflowRunDatabase.journal` is an ordinary `DurableStream`, so `durableRun` @@ -367,6 +372,13 @@ Cloudflare's synchronous transactions use uniquely named SQLite savepoints on that same connection and only while XMD's caller-owned transaction is open. DOFS does not begin, commit or roll back a top-level transaction. +If a caller-owned transaction does not commit, its finalizer attempts SQLite +rollback and then invalidates both the resolution and blob caches on the +authoritative DOFS wrapper before releasing the serialized connection turn. +The same cleanup covers body failure, cancellation during the body or child +teardown, final Workspace validation failure, and commit failure. Rolled-back +topology therefore cannot survive as a positive or negative cache entry. + Adapter-private root operations also run only inside this caller-owned transaction. Capture traverses and validates the complete live DOFS frontier, builds or reuses a canonical DOFS file manifest when ordered chunks do not yet @@ -438,7 +450,8 @@ and exact manifest/blob reachability, validates every referenced manifest, blob, byte payload and live chunk, and requires the read-only live snapshot to equal the singleton current root. Malformed paths or topology, dangling or cyclic dirents, invalid hardlinks, corrupt hashes or sizes, inexact references, -and a live/current mismatch are damage. Recognition performs no repair. +a file entry whose declared size differs from its referenced DOFS manifest, and +a live/current mismatch are damage. Recognition performs no repair. No message repeats a stored value — or a stored *name*. Props and journal payloads are retained history, and a member name can carry a credential as