From aeea3c1e1339b69047e96576ef3b07a06ecbb243 Mon Sep 17 00:00:00 2001 From: Taras Mankovski <74687+taras@users.noreply.github.com> Date: Fri, 7 Aug 2026 18:45:01 -0400 Subject: [PATCH 1/2] =?UTF-8?q?=F0=9F=97=84=EF=B8=8F=20Own=20WorkflowRun?= =?UTF-8?q?=20connections=20and=20complete=20schema=20v1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- packages/workflow/mod.ts | 1 + packages/workflow/src/deno/connections.ts | 140 +++++++ packages/workflow/src/deno/database.ts | 32 +- packages/workflow/src/deno/journal.ts | 22 +- packages/workflow/src/deno/lock.ts | 42 +-- packages/workflow/src/deno/provider.ts | 152 +++----- packages/workflow/src/deno/savepoints.ts | 48 +++ packages/workflow/src/deno/schema.ts | 349 ++++++++++++++++-- packages/workflow/src/deno/transaction.ts | 21 +- packages/workflow/src/deno/workspace/empty.ts | 144 ++++++++ packages/workflow/src/storage/errors.ts | 13 + .../tests/workflow-run-journal.test.ts | 20 + .../tests/workflow-run-storage.test.ts | 331 ++++++++++++++++- 13 files changed, 1095 insertions(+), 220 deletions(-) create mode 100644 packages/workflow/src/deno/connections.ts create mode 100644 packages/workflow/src/deno/savepoints.ts create mode 100644 packages/workflow/src/deno/workspace/empty.ts diff --git a/packages/workflow/mod.ts b/packages/workflow/mod.ts index c36be847..a5f7b872 100644 --- a/packages/workflow/mod.ts +++ b/packages/workflow/mod.ts @@ -67,6 +67,7 @@ export { WorkflowDatabaseFormatError, WorkflowDefinitionError, WorkflowDocumentExecutionError, + WorkflowIncompleteVersionOneError, WorkflowRecordMalformedError, WorkflowRequestError, WorkflowRunConflictError, diff --git a/packages/workflow/src/deno/connections.ts b/packages/workflow/src/deno/connections.ts new file mode 100644 index 00000000..c265e9d6 --- /dev/null +++ b/packages/workflow/src/deno/connections.ts @@ -0,0 +1,140 @@ +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 type { + DurableObjectStorageLike, + SQLCursorLike, + SQLStorageLike, +} from "../../vendor/cloudflare-computer-dofs/generated/types.d.ts"; +import { type ConnectionLock, createConnectionLock } from "./lock.ts"; +import { createSavepointManager, type SavepointManager } from "./savepoints.ts"; + +export interface RunConnection { + readonly path: string; + readonly database: DatabaseSync; + readonly dofs: CloudflareDatabase; + readonly filesystem: WorkspaceFilesystem; + readonly lock: ConnectionLock; + readonly savepoints: SavepointManager; + transactionOpen: boolean; + close(): void; +} + +export interface WorkflowRunConnections { + at(path: string): RunConnection; + close(): void; +} + +class SqliteStorage implements SQLStorageLike { + readonly database: DatabaseSync; + readonly savepoints: () => SavepointManager; + + constructor(database: DatabaseSync, savepoints: () => SavepointManager) { + this.database = database; + this.savepoints = savepoints; + } + + exec>( + query: string, + ...bindings: unknown[] + ): SQLCursorLike { + const statement = this.database.prepare(query); + const rows = Reflect.apply(statement.all, statement, bindings); + return { + toArray(): Row[] { + return rows; + }, + }; + } +} + +function createConnection(path: string): RunConnection { + const database = new DatabaseSync(path); + try { + database.exec("PRAGMA foreign_keys = ON"); + database.exec("PRAGMA busy_timeout = 5000"); + } catch (error) { + database.close(); + throw error; + } + + let open = true; + const connection: { + savepoints: SavepointManager | undefined; + transactionOpen: boolean; + } = { savepoints: undefined, transactionOpen: false }; + const storage = new SqliteStorage(database, () => { + const savepoints = connection.savepoints; + if (savepoints === undefined) { + throw new WorkflowConnectionStateError("the savepoint manager is not installed"); + } + return savepoints; + }); + const durableStorage: DurableObjectStorageLike = { + sql: storage, + transactionSync(closure: () => T): T { + return storage.savepoints().synchronous(closure); + }, + }; + const dofs = new CloudflareDatabase(durableStorage); + const savepoints = createSavepointManager(database, () => connection.transactionOpen); + connection.savepoints = savepoints; + + return { + path, + database, + dofs, + filesystem: new WorkspaceFilesystem(dofs), + lock: createConnectionLock(), + savepoints, + get transactionOpen() { + return connection.transactionOpen; + }, + set transactionOpen(value: boolean) { + connection.transactionOpen = value; + }, + close() { + if (open) { + open = false; + database.close(); + } + }, + }; +} + +export class WorkflowConnectionStateError extends Error { + override name = "WorkflowConnectionStateError"; +} + +export function createWorkflowRunConnections(): WorkflowRunConnections { + const entries = new Map(); + let open = true; + + return { + at(path: string): RunConnection { + if (!open) { + throw new WorkflowConnectionStateError("the workflow storage provider has closed"); + } + const canonical = resolve(path); + const existing = entries.get(canonical); + if (existing !== undefined) { + return existing; + } + const created = createConnection(canonical); + entries.set(canonical, created); + return created; + }, + + close(): void { + if (!open) { + return; + } + open = false; + for (const entry of entries.values()) { + entry.close(); + } + entries.clear(); + }, + }; +} diff --git a/packages/workflow/src/deno/database.ts b/packages/workflow/src/deno/database.ts index 95bdc06c..02e22e6c 100644 --- a/packages/workflow/src/deno/database.ts +++ b/packages/workflow/src/deno/database.ts @@ -25,9 +25,9 @@ * * ## Lifetime * - * The handle belongs to the scope that opened it. When that scope ends the - * connection closes, and every later call answers with a closed-handle failure - * rather than reopening the file behind the caller's back. + * The handle belongs to the scope that opened it. When that scope ends its + * lease closes, and every later call answers with a closed-handle failure. The + * provider owns the authoritative physical connection for its own scope. */ import { randomUUID } from "node:crypto"; @@ -54,7 +54,7 @@ import { type WorkflowRunRecord, } from "../storage/record.ts"; import { insertJournalEvent, readJournalEntries } from "./journal.ts"; -import type { ConnectionLock } from "./lock.ts"; +import type { RunConnection } from "./connections.ts"; import { ActiveTransaction, enclosing, @@ -86,19 +86,12 @@ const SELECT_EXECUTIONS = "SELECT * FROM document_executions ORDER BY sequence A /** What opening needs from whoever found the file and checked its schema. */ export interface OpenConnection { - readonly database: DatabaseSync; - readonly path: string; + readonly connection: RunConnection; readonly record: WorkflowRunRecord; - /** Shared by every handle on this file, so turns are taken per database. */ - readonly lock: ConnectionLock; } /** - * Open a run's database for the life of the calling scope. - * - * The connection closes through ordinary teardown rather than a caller - * remembering to close it, so an interrupted host leaves no connection open on - * a file another process is about to take a write lock on. + * Open a scope-owned lease on a run's provider-owned database connection. */ export function openWorkflowRunDatabase( connection: OpenConnection, @@ -118,7 +111,7 @@ interface Handle { } function createHandle(connection: OpenConnection): Handle { - const { database, path, lock } = connection; + const { database, path, lock } = connection.connection; let closed = false; let record = connection.record; @@ -195,12 +188,14 @@ function createHandle(connection: OpenConnection): Handle { } const transaction = { open: true }; + connection.connection.transactionOpen = true; let committed = false; // Registered after the lock, so teardown rolls back while the connection // is still ours and releases it only once that is done. yield* ensure(() => { transaction.open = false; + connection.connection.transactionOpen = false; if (!committed) { rollback(database); } @@ -209,7 +204,7 @@ function createHandle(connection: OpenConnection): Handle { // The chain, not just this path: a transaction on another run nested // inside this one must not hide that this one is held. yield* ActiveTransaction.set(yield* enclosing(path)); - yield* useTransactionSavepoints(database, () => transaction.open); + yield* useTransactionSavepoints(connection.connection.savepoints, () => transaction.open); try { // The body runs in a scope of its own, so everything it started — @@ -219,17 +214,21 @@ function createHandle(connection: OpenConnection): Handle { // would let that append autocommit on its own, published whatever the // transaction went on to decide. const value = yield* scoped(function* () { - return yield* body({ journal: enlistedJournal(database, transaction, path) }); + return yield* body({ + journal: enlistedJournal(database, transaction, path), + }); }); // Closed before the commit, not after: nothing may append to a // transaction whose contents are already decided. transaction.open = false; + connection.connection.transactionOpen = false; database.exec("COMMIT"); committed = true; return Ok(value); } catch (error) { transaction.open = false; + connection.connection.transactionOpen = false; return Err(translateSqliteError(error, path)); } }); @@ -370,7 +369,6 @@ function createHandle(connection: OpenConnection): Handle { database: handle, close() { closed = true; - database.close(); }, }; } diff --git a/packages/workflow/src/deno/journal.ts b/packages/workflow/src/deno/journal.ts index afddb156..707477bf 100644 --- a/packages/workflow/src/deno/journal.ts +++ b/packages/workflow/src/deno/journal.ts @@ -30,7 +30,8 @@ export interface JournalEntry { readonly event: DurableEvent; } -const INSERT = "INSERT INTO journal_events (event_id, record) VALUES (?, ?)"; +const INSERT = `INSERT INTO journal_events (event_id, record, workspace_root_id) + VALUES (?, ?, ?)`; const SELECT = "SELECT event_id, record FROM journal_events ORDER BY sequence ASC"; /** @@ -40,12 +41,27 @@ const SELECT = "SELECT event_id, record FROM journal_events ORDER BY sequence AS * one a caller opened is decided above this function, which is what lets a * standalone append and an enlisted append share one statement. */ -export function insertJournalEvent(database: DatabaseSync, event: DurableEvent): string { +export function insertJournalEvent( + database: DatabaseSync, + event: DurableEvent, + workspaceRootId = currentWorkspaceRoot(database), +): string { const eventId = randomUUID(); - database.prepare(INSERT).run(eventId, serializeDurableEvent(event)); + database.prepare(INSERT).run(eventId, serializeDurableEvent(event), workspaceRootId); return eventId; } +function currentWorkspaceRoot(database: DatabaseSync): string { + const row = database + .prepare("SELECT current_root_id FROM workspace_state WHERE singleton_id = 1") + .get(); + const rootId = row?.["current_root_id"]; + if (typeof rootId !== "string") { + throw new Error("the workflow Workspace has no current root"); + } + return rootId; +} + /** Every retained event, in the order it was appended. */ export function readJournalEntries(database: DatabaseSync): JournalEntry[] { const entries: JournalEntry[] = []; diff --git a/packages/workflow/src/deno/lock.ts b/packages/workflow/src/deno/lock.ts index 8fb9375a..1d7c4e3c 100644 --- a/packages/workflow/src/deno/lock.ts +++ b/packages/workflow/src/deno/lock.ts @@ -8,14 +8,11 @@ * serializes everything that touches one database rather than relying on * callers to take turns. * - * Turns are taken per database file, not per connection. Two handles opened for - * the same run have two connections, and the second one entering SQLite while - * the first holds a write lock does not wait politely: `node:sqlite` is - * synchronous, so it stops the host's event loop for the whole busy timeout — - * during which the first transaction cannot resume to commit, and the second - * ends up reporting the database busy. Waiting here instead leaves the host - * running and lets the first transaction finish. SQLite's own locking remains - * responsible for contention between processes. + * Turns are taken by the provider-owned entry for one database file. Every + * handle for that run leases the same physical connection, so a second + * operation must wait here before it can issue statements inside the first + * operation's transaction. Waiting cooperatively leaves the host running; + * SQLite's own locking remains responsible for contention between processes. * * Waiting is cancellable and hand-off is synchronous. A caller torn down while * queued leaves the queue without ever running its statements, and a caller @@ -23,40 +20,13 @@ * passes it on — the two states that would otherwise strand it. */ -import { ensure, type Operation, resource, withResolvers, type WithResolvers } from "effection"; +import { ensure, type Operation, resource, type WithResolvers, withResolvers } from "effection"; /** A turn at one database, held for as long as the acquiring scope lives. */ export interface ConnectionLock { hold(): Operation; } -/** - * The turns for every database one provider has opened. - * - * Owned by the provider installation rather than the module, so the - * coordination lasts exactly as long as the scope that installed the provider - * and nothing accumulates across runs. - */ -export interface ConnectionLocks { - at(path: string): ConnectionLock; -} - -export function createConnectionLocks(): ConnectionLocks { - const locks = new Map(); - - return { - at(path: string): ConnectionLock { - const existing = locks.get(path); - if (existing !== undefined) { - return existing; - } - const created = createConnectionLock(); - locks.set(path, created); - return created; - }, - }; -} - interface Turn { readonly gate: WithResolvers; granted: boolean; diff --git a/packages/workflow/src/deno/provider.ts b/packages/workflow/src/deno/provider.ts index 0cbcb850..785eac37 100644 --- a/packages/workflow/src/deno/provider.ts +++ b/packages/workflow/src/deno/provider.ts @@ -26,7 +26,6 @@ */ import { dirname, isAbsolute } from "node:path"; -import { DatabaseSync } from "node:sqlite"; import { ensureDir, exists } from "@effectionx/fs"; import { ensure, Err, Ok, type Operation, type Result, scoped } from "effection"; import { @@ -57,19 +56,14 @@ import { } from "../storage/members.ts"; import { canonicalJson, type WorkflowRunRecord } from "../storage/record.ts"; import { openWorkflowRunDatabase, readRunRow } from "./database.ts"; -import { type ConnectionLocks, createConnectionLocks } from "./lock.ts"; +import { + createWorkflowRunConnections, + type RunConnection, + type WorkflowRunConnections, +} from "./connections.ts"; import { workflowRunPath } from "./path.ts"; import { initializeSchema, isUninitialized, translateSqliteError, verifySchema } from "./schema.ts"; -/** - * How long a connection waits for another host's write lock. - * - * SQLite is reached synchronously, so this is also how long the thread can - * stop. Long enough for a transaction that is committing, short enough that a - * host holding a lock it will never release is reported rather than waited on. - */ -const BUSY_TIMEOUT_MS = 5_000; - const INSERT_RUN = `INSERT INTO workflow_run (id, run_id, definition, base, props, status, created_at, updated_at) VALUES (1, ?, ?, ?, ?, 'running', ?, ?)`; @@ -100,15 +94,18 @@ export interface WorkflowRunStorageOptions { */ export function* useWorkflowRunStorage(options: WorkflowRunStorageOptions): Operation { const root = authorizedRoot(options.root); - const locks = createConnectionLocks(); + const connections = createWorkflowRunConnections(); + yield* ensure(() => { + connections.close(); + }); yield* WorkflowRunStorage.around( { *create([request]) { - return yield* createWorkflowRun(root, locks, request); + return yield* createWorkflowRun(root, connections, request); }, *lookup([runId]) { - return yield* lookupWorkflowRun(root, locks, runId); + return yield* lookupWorkflowRun(root, connections, runId); }, }, { at: "min" }, @@ -144,7 +141,7 @@ interface CheckedRequest { function* createWorkflowRun( root: string, - locks: ConnectionLocks, + connections: WorkflowRunConnections, request: CreateWorkflowRunRequest, ): Operation> { const checked = checkRequest(request); @@ -158,15 +155,15 @@ function* createWorkflowRun( yield* ensureDir(dirname(path)); } - const lock = locks.at(path); - - return yield* withConnection(path, function* (database): Operation> { + try { + const connection = connections.at(path); + const { lock } = connection; // Held across initialization, so a second caller creating the same run // waits here rather than inside a synchronous `BEGIN IMMEDIATE` that // would stop the host while the first one is still committing. const stored = yield* scoped(function* () { yield* lock.hold(); - return establish(database, path, wanted); + return establish(connection, path, wanted); }); if (!stored.ok) { return stored; @@ -182,13 +179,15 @@ function* createWorkflowRun( return Err(new WorkflowRunConflictError(wanted.runId, differing)); } - return Ok(yield* openWorkflowRunDatabase({ database, path, record, lock })); - }); + return Ok(yield* openWorkflowRunDatabase({ connection, record })); + } catch (error) { + return refusal(error, path); + } } function* lookupWorkflowRun( root: string, - locks: ConnectionLocks, + connections: WorkflowRunConnections, runId: string, ): Operation> { const checked = checkRunId(runId); @@ -205,9 +204,9 @@ function* lookupWorkflowRun( return Err(new WorkflowRunNotFoundError(wanted)); } - const lock = locks.at(path); - - return yield* withConnection(path, function* (database): Operation> { + try { + const connection = connections.at(path); + const { database, lock } = connection; const record = yield* scoped(function* (): Operation> { yield* lock.hold(); try { @@ -225,81 +224,10 @@ function* lookupWorkflowRun( return Err(new WorkflowRunIdMismatchError(runId, path)); } - return Ok(yield* openWorkflowRunDatabase({ database, path, record: record.value, lock })); - }); -} - -/** - * Open the file, and close it again unless a handle takes ownership. - * - * A refused database must not leave a connection open on a file the caller is - * about to be told is unusable — that connection would hold a lock nothing was - * going to release until the process ended. That includes a refusal raised on - * the way to producing the handle: reading a row while the handle is being - * built is as capable of finding an unreadable record as reading one later. - * - * Between opening the file and handing it to a handle there is checking to do, - * and a caller may be cancelled during it. The connection is therefore given - * up through ordinary teardown as well, so an interrupted open closes what it - * opened rather than leaving the file locked by a connection nobody holds. - */ -function* withConnection( - path: string, - body: (database: DatabaseSync) => Operation>, -): Operation> { - let database: DatabaseSync; - try { - database = new DatabaseSync(path); - } catch (error) { - return refusal(error, path); - } - - let adopted = false; - let released = false; - - function release(): void { - if (adopted || released) { - return; - } - released = true; - database.close(); - } - - // Registered immediately, before the connection is even configured: from - // here on there is an open file handle, and every way out of this function — - // a failing pragma, a refusal, cancellation part-way through the checking — - // has to close it. Once a handle owns the connection this is a no-op and the - // handle's own teardown closes it. - yield* ensure(release); - - try { - // Connection settings, not changes to the file. Without a busy timeout - // SQLite refuses a contended write lock immediately, so a second host - // reaching the same run would be told the database is busy rather than - // waiting the moment it takes the first one to commit. Foreign keys are - // off by default and per connection, and without them a stop reason could - // name a journal event that is not there. - database.exec(`PRAGMA busy_timeout = ${BUSY_TIMEOUT_MS}`); - database.exec("PRAGMA foreign_keys = ON"); - } catch (error) { - release(); - return refusal(error, path); - } - - let result: Result; - try { - result = yield* body(database); + return Ok(yield* openWorkflowRunDatabase({ connection, record: record.value })); } catch (error) { - release(); return refusal(error, path); } - - if (result.ok) { - adopted = true; - } else { - release(); - } - return result; } /** @@ -311,38 +239,44 @@ function* withConnection( * run in between. */ function establish( - database: DatabaseSync, + connection: RunConnection, path: string, request: CheckedRequest, ): Result { + const { database } = connection; try { if (!isUninitialized(database, path)) { verifySchema(database, path); } database.exec("BEGIN IMMEDIATE"); + connection.transactionOpen = true; try { if (isUninitialized(database, path)) { const stamp = new Date().toISOString(); - initializeSchema(database); - database - .prepare(INSERT_RUN) - .run( - request.runId, - canonicalJson(definitionToJson(request.definition)), - request.base, - canonicalJson(request.props), - stamp, - stamp, - ); + initializeSchema(database, connection.dofs, () => { + database + .prepare(INSERT_RUN) + .run( + request.runId, + canonicalJson(definitionToJson(request.definition)), + request.base, + canonicalJson(request.props), + stamp, + stamp, + ); + }); } else { verifySchema(database, path); } + verifySchema(database, path); const record = readRunRow(database, path); + connection.transactionOpen = false; database.exec("COMMIT"); return Ok(record); } catch (error) { + connection.transactionOpen = false; database.exec("ROLLBACK"); throw error; } diff --git a/packages/workflow/src/deno/savepoints.ts b/packages/workflow/src/deno/savepoints.ts new file mode 100644 index 00000000..c457458e --- /dev/null +++ b/packages/workflow/src/deno/savepoints.ts @@ -0,0 +1,48 @@ +import type { DatabaseSync } from "node:sqlite"; +import { WorkflowTransactionError } from "../storage/errors.ts"; + +export interface SavepointManager { + synchronous(body: () => T): T; +} + +export function createSavepointManager( + database: DatabaseSync, + isTransactionOpen: () => boolean, +): SavepointManager { + let next = 0; + + function allocate(): string { + const name = `xmd_savepoint_${next}`; + next += 1; + return name; + } + + function assertOpen(): void { + if (!isTransactionOpen()) { + throw new WorkflowTransactionError( + "a savepoint needs the caller-owned workflow transaction to remain open.", + ); + } + } + + function rollback(name: string): void { + database.exec(`ROLLBACK TO ${name}`); + database.exec(`RELEASE ${name}`); + } + + return { + synchronous(body: () => T): T { + assertOpen(); + const name = allocate(); + database.exec(`SAVEPOINT ${name}`); + try { + const value = body(); + database.exec(`RELEASE ${name}`); + return value; + } catch (error) { + rollback(name); + throw error; + } + }, + }; +} diff --git a/packages/workflow/src/deno/schema.ts b/packages/workflow/src/deno/schema.ts index d0c8c707..c063409d 100644 --- a/packages/workflow/src/deno/schema.ts +++ b/packages/workflow/src/deno/schema.ts @@ -24,12 +24,16 @@ */ import type { DatabaseSync } from "node:sqlite"; +import type { Database as CloudflareDatabase } from "../../vendor/cloudflare-computer-dofs/generated/storage.js"; +import { initializeSchema as initializeCloudflareSchema } from "../../vendor/cloudflare-computer-dofs/generated/schema/index.js"; import { WorkflowDatabaseCorruptError, WorkflowDatabaseFormatError, + WorkflowIncompleteVersionOneError, WorkflowRequestError, WorkflowSchemaVersionError, } from "../storage/errors.ts"; +import { initializeEmptyWorkspace, verifyEmptyWorkspace } from "./workspace/empty.ts"; /** * The bytes `XMD1` as a 32-bit integer, written into the SQLite header. @@ -66,23 +70,259 @@ function coherentStopReason(): string { * Kept as separate definitions so verification can compare what a file holds * with what this build writes, rather than settling for the table's name. * - * The journal is here from the first version even though metadata landed - * first: a schema that grew a table between two commits of the same release - * would owe a migration to databases that never existed. It is also created - * first, because the stop-reason references point at it. + * The complete version-1 shape includes the pinned DOFS objects, retained + * Workspace roots, journal and metadata. Dependency order is explicit: DOFS + * content precedes root references, and roots precede the journal rows that + * name them. */ -const TABLES: ReadonlyMap = new Map([ +interface DeclaredObject { + readonly type: "table" | "index"; + readonly sql: string; +} + +const OBJECTS: ReadonlyMap = new Map([ + [ + "vfs_meta", + { + type: "table", + sql: `CREATE TABLE vfs_meta ( + k TEXT PRIMARY KEY, + v INTEGER NOT NULL + )`, + }, + ], + [ + "vfs_nodes", + { + type: "table", + sql: `CREATE TABLE vfs_nodes ( + inode INTEGER PRIMARY KEY AUTOINCREMENT, + type TEXT NOT NULL CHECK(type IN ('file','dir','symlink')), + mode INTEGER NOT NULL DEFAULT 493, + mtime INTEGER NOT NULL, + rev INTEGER NOT NULL DEFAULT 0, + mount_root TEXT, + stub_size INTEGER, + manifest_hash BLOB, + link_target TEXT, + size INTEGER NOT NULL DEFAULT 0 + )`, + }, + ], + [ + "vfs_dirents", + { + type: "table", + sql: `CREATE TABLE vfs_dirents ( + parent_inode INTEGER NOT NULL, + name TEXT NOT NULL, + child_inode INTEGER NOT NULL, + PRIMARY KEY (parent_inode, name) + ) WITHOUT ROWID`, + }, + ], + [ + "vfs_dirents_by_child", + { + type: "index", + sql: "CREATE INDEX vfs_dirents_by_child ON vfs_dirents(child_inode)", + }, + ], + [ + "vfs_nodes_by_rev", + { + type: "index", + sql: "CREATE INDEX vfs_nodes_by_rev ON vfs_nodes(rev)", + }, + ], + [ + "vfs_nodes_by_manifest_hash", + { + type: "index", + sql: `CREATE INDEX vfs_nodes_by_manifest_hash + ON vfs_nodes(manifest_hash) WHERE manifest_hash IS NOT NULL`, + }, + ], + [ + "vfs_blobs", + { + type: "table", + sql: `CREATE TABLE vfs_blobs ( + hash BLOB PRIMARY KEY, + size INTEGER NOT NULL, + last_seen INTEGER NOT NULL + )`, + }, + ], + [ + "vfs_blob_bytes", + { + type: "table", + sql: `CREATE TABLE vfs_blob_bytes ( + hash BLOB PRIMARY KEY REFERENCES vfs_blobs(hash) ON DELETE CASCADE, + bytes BLOB NOT NULL + )`, + }, + ], + [ + "vfs_chunks", + { + type: "table", + sql: `CREATE TABLE vfs_chunks ( + inode INTEGER NOT NULL, + idx INTEGER NOT NULL, + hash BLOB NOT NULL, + size INTEGER NOT NULL, + PRIMARY KEY (inode, idx) + ) WITHOUT ROWID`, + }, + ], + [ + "vfs_chunks_by_hash", + { + type: "index", + sql: "CREATE INDEX vfs_chunks_by_hash ON vfs_chunks(hash)", + }, + ], + [ + "vfs_manifests", + { + type: "table", + sql: `CREATE TABLE vfs_manifests ( + hash BLOB PRIMARY KEY, + size INTEGER NOT NULL, + encoded BLOB NOT NULL, + last_seen INTEGER NOT NULL DEFAULT 0 + )`, + }, + ], + [ + "vfs_changes", + { + type: "table", + sql: `CREATE TABLE vfs_changes ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + rev INTEGER NOT NULL, + path TEXT NOT NULL, + op TEXT NOT NULL CHECK(op IN ('delete')) + )`, + }, + ], + [ + "vfs_changes_by_rev", + { + type: "index", + sql: "CREATE INDEX vfs_changes_by_rev ON vfs_changes(rev)", + }, + ], + [ + "vfs_changes_by_path", + { + type: "index", + sql: "CREATE INDEX vfs_changes_by_path ON vfs_changes(path, id DESC)", + }, + ], + [ + "_vfs_watermark", + { + type: "table", + sql: `CREATE TABLE _vfs_watermark ( + k TEXT NOT NULL, + backend TEXT NOT NULL DEFAULT 'default', + v INTEGER NOT NULL, + PRIMARY KEY (k, backend) + )`, + }, + ], + [ + "_vfs_fetch_cursor", + { + type: "table", + sql: `CREATE TABLE _vfs_fetch_cursor ( + k TEXT NOT NULL CHECK(k = 'fetch'), + backend TEXT NOT NULL DEFAULT 'default', + path TEXT, + PRIMARY KEY (k, backend) + )`, + }, + ], + [ + "_vfs_mounts", + { + type: "table", + sql: `CREATE TABLE _vfs_mounts ( + root TEXT PRIMARY KEY, + kind TEXT NOT NULL, + indexed INTEGER NOT NULL DEFAULT 0, + mode TEXT NOT NULL DEFAULT 'read-only' + CHECK(mode IN ('read-only', 'read-write')) + )`, + }, + ], + [ + "workspace_roots", + { + type: "table", + sql: `CREATE TABLE workspace_roots ( + root_id TEXT PRIMARY KEY CHECK ( + length(root_id) = 64 AND root_id NOT GLOB '*[^0-9a-f]*' + ), + format_version INTEGER NOT NULL CHECK (format_version = 1), + manifest TEXT NOT NULL CHECK (json_valid(manifest)) +) STRICT`, + }, + ], + [ + "workspace_root_manifest_refs", + { + type: "table", + sql: `CREATE TABLE workspace_root_manifest_refs ( + root_id TEXT NOT NULL REFERENCES workspace_roots(root_id) ON DELETE CASCADE, + manifest_hash BLOB NOT NULL REFERENCES vfs_manifests(hash) ON DELETE RESTRICT, + PRIMARY KEY (root_id, manifest_hash) +) STRICT, WITHOUT ROWID`, + }, + ], + [ + "workspace_root_blob_refs", + { + type: "table", + sql: `CREATE TABLE workspace_root_blob_refs ( + root_id TEXT NOT NULL REFERENCES workspace_roots(root_id) ON DELETE CASCADE, + blob_hash BLOB NOT NULL, + PRIMARY KEY (root_id, blob_hash), + FOREIGN KEY (blob_hash) REFERENCES vfs_blobs(hash) ON DELETE RESTRICT, + FOREIGN KEY (blob_hash) REFERENCES vfs_blob_bytes(hash) ON DELETE RESTRICT +) STRICT, WITHOUT ROWID`, + }, + ], + [ + "workspace_state", + { + type: "table", + sql: `CREATE TABLE workspace_state ( + singleton_id INTEGER PRIMARY KEY CHECK (singleton_id = 1), + current_root_id TEXT NOT NULL REFERENCES workspace_roots(root_id) ON DELETE RESTRICT +) STRICT`, + }, + ], [ "journal_events", - `CREATE TABLE journal_events ( + { + type: "table", + sql: `CREATE TABLE journal_events ( sequence INTEGER PRIMARY KEY AUTOINCREMENT, event_id TEXT NOT NULL UNIQUE, - record TEXT NOT NULL CHECK (json_valid(record)) + record TEXT NOT NULL CHECK (json_valid(record)), + workspace_root_id TEXT NOT NULL REFERENCES workspace_roots(root_id) ON DELETE RESTRICT ) STRICT`, + }, ], [ "workflow_run", - `CREATE TABLE workflow_run ( + { + type: "table", + sql: `CREATE TABLE workflow_run ( id INTEGER PRIMARY KEY CHECK (id = 1), run_id TEXT NOT NULL, definition TEXT NOT NULL CHECK (json_valid(definition)), @@ -96,19 +336,25 @@ const TABLES: ReadonlyMap = new Map([ updated_at TEXT NOT NULL, ${coherentStopReason()} ) STRICT`, + }, ], [ "definition_retrieval", - `CREATE TABLE definition_retrieval ( + { + type: "table", + sql: `CREATE TABLE definition_retrieval ( id INTEGER PRIMARY KEY CHECK (id = 1), metadata TEXT NOT NULL CHECK (json_valid(metadata)), revision INTEGER NOT NULL CHECK (revision >= 1 AND revision <= 9007199254740991), updated_at TEXT NOT NULL ) STRICT`, + }, ], [ "document_executions", - `CREATE TABLE document_executions ( + { + type: "table", + sql: `CREATE TABLE document_executions ( sequence INTEGER PRIMARY KEY AUTOINCREMENT, execution_id TEXT NOT NULL UNIQUE, started_at TEXT NOT NULL, @@ -121,14 +367,30 @@ const TABLES: ReadonlyMap = new Map([ CHECK (stop_status IS NOT NULL OR stop_reason_kind IS NULL), ${coherentStopReason()} ) STRICT`, + }, ], ]); +export const EXPECTED_SCHEMA = Object.freeze( + [...OBJECTS.entries()].map(([name, object]) => + Object.freeze({ name, type: object.type, sql: normalize(object.sql) }), + ), +); + +/** Objects version 1 declares, including the pinned Cloudflare structure. */ +export const REQUIRED_OBJECTS: readonly string[] = Object.freeze([...OBJECTS.keys()]); + /** Tables version 1 declares. */ -export const REQUIRED_TABLES: readonly string[] = Object.freeze([...TABLES.keys()]); +export const REQUIRED_TABLES: readonly string[] = Object.freeze( + [...OBJECTS.entries()].filter(([, object]) => object.type === "table").map(([name]) => name), +); /** Version 1 in full. */ -export const SCHEMA_SQL = [...TABLES.values()].map((sql) => `${sql};`).join("\n\n"); +export const SCHEMA_SQL = [...OBJECTS.values()] + .filter((object) => object.type === "table" && !object.sql.startsWith("CREATE TABLE vfs_")) + .filter((object) => !object.sql.startsWith("CREATE TABLE _vfs_")) + .map((object) => `${object.sql};`) + .join("\n\n"); /** * Write the version-1 schema into a database that holds nothing. @@ -137,10 +399,17 @@ export const SCHEMA_SQL = [...TABLES.values()].map((sql) => `${sql};`).join("\n\ * and the tables appear together or not at all — a half-initialized file would * be indistinguishable from one this build must refuse. */ -export function initializeSchema(database: DatabaseSync): void { +export function initializeSchema( + database: DatabaseSync, + dofs: CloudflareDatabase, + initializeRun: () => void, +): void { database.exec(`PRAGMA application_id = ${APPLICATION_ID};`); - database.exec(`PRAGMA user_version = ${SCHEMA_VERSION};`); database.exec(SCHEMA_SQL); + initializeCloudflareSchema(dofs, () => 0); + initializeEmptyWorkspace(database); + initializeRun(); + database.exec(`PRAGMA user_version = ${SCHEMA_VERSION};`); } /** @@ -170,6 +439,12 @@ export function verifySchema(database: DatabaseSync, path: string): void { const applicationId = readPragmaNumber(database, "application_id", path); if (applicationId !== APPLICATION_ID) { + if (applicationId === 0 && hasDeclaredVersionOneObjects(database, path)) { + throw new WorkflowDatabaseCorruptError( + path, + "it contains a partial version-1 initialization without the XMD application identity", + ); + } throw new WorkflowDatabaseFormatError( path, `it carries application id ${applicationId} rather than ${APPLICATION_ID}`, @@ -183,6 +458,7 @@ export function verifySchema(database: DatabaseSync, path: string): void { verifyStructure(database, path); checkForeignKeys(database, path); + verifyEmptyWorkspace(database, path); } /** @@ -194,36 +470,50 @@ export function verifySchema(database: DatabaseSync, path: string): void { */ function verifyStructure(database: DatabaseSync, path: string): void { const objects = schemaObjects(database, path); + const intermediate = [ + "definition_retrieval", + "document_executions", + "journal_events", + "workflow_run", + ]; + if ( + objects.length === intermediate.length && + objects.every((object) => object.type === "table") && + objects + .map((object) => object.name) + .sort() + .join("\0") === intermediate.join("\0") + ) { + throw new WorkflowIncompleteVersionOneError(path); + } for (const object of objects) { - if (object.type !== "table") { - throw new WorkflowDatabaseCorruptError( - path, - `it declares a ${object.type} that version ${SCHEMA_VERSION} does not`, - ); - } - const expected = TABLES.get(object.name); + const expected = OBJECTS.get(object.name); if (expected === undefined) { throw new WorkflowDatabaseCorruptError( path, - `it declares a table that version ${SCHEMA_VERSION} does not`, + `it declares an object that version ${SCHEMA_VERSION} does not`, ); } - if (normalize(object.sql) !== normalize(expected)) { + if (object.type !== expected.type || normalize(object.sql) !== normalize(expected.sql)) { throw new WorkflowDatabaseCorruptError( path, - `its ${object.name} table is not shaped the way version ${SCHEMA_VERSION} declares it`, + `its ${object.name} object is not shaped the way version ${SCHEMA_VERSION} declares it`, ); } } const present = new Set(objects.map((object) => object.name)); - const missing = REQUIRED_TABLES.filter((table) => !present.has(table)); + const missing = REQUIRED_OBJECTS.filter((name) => !present.has(name)); if (missing.length > 0) { throw new WorkflowDatabaseCorruptError(path, `it is missing the table ${missing.join(", ")}`); } } +function hasDeclaredVersionOneObjects(database: DatabaseSync, path: string): boolean { + return schemaObjects(database, path).some((object) => OBJECTS.has(object.name)); +} + /** * Ask SQLite whether it can still read its own file. * @@ -243,8 +533,8 @@ export function checkIntegrity(database: DatabaseSync, path: string): void { /** * Ask SQLite whether its references still point at anything. * - * A stop reason naming a journal event is only a reason while that event - * exists; a row pointing at one that does not is damage, not a reason. + * A retained reference names an object only while that object exists; a row + * pointing at nothing is damage rather than a partial retained state. */ function checkForeignKeys(database: DatabaseSync, path: string): void { if (query(database, "PRAGMA foreign_key_check", path).length > 0) { @@ -315,7 +605,7 @@ const SQLITE_CORRUPT = 11; /** `SQLITE_NOTADB`: the bytes are not a SQLite database at all. */ const SQLITE_NOTADB = 26; -/** `SQLITE_CONSTRAINT_FOREIGNKEY`: a reference points at a row that is not there. */ +/** `SQLITE_CONSTRAINT_FOREIGNKEY`: a requested journal reference does not exist. */ const SQLITE_CONSTRAINT_FOREIGNKEY = 787; /** @@ -332,9 +622,6 @@ export function translateSqliteError(error: unknown, path: string): unknown { case SQLITE_CORRUPT: return new WorkflowDatabaseCorruptError(path, "SQLite reported a damaged image"); case SQLITE_CONSTRAINT_FOREIGNKEY: - // The only reference version 1 declares. A stop reason may name a - // journal event, and naming one this run does not hold is a reason that - // refers to nothing. return new WorkflowRequestError( "the stop reason names a journal event this run does not hold. A journal reason " + "points at an event that has already been appended and filtered.", diff --git a/packages/workflow/src/deno/transaction.ts b/packages/workflow/src/deno/transaction.ts index 19d84e05..5c6e8bfb 100644 --- a/packages/workflow/src/deno/transaction.ts +++ b/packages/workflow/src/deno/transaction.ts @@ -17,8 +17,8 @@ import { type Api, createApi } from "@effectionx/context-api"; import { type Context, createContext, type Operation } from "effection"; -import type { DatabaseSync } from "node:sqlite"; import { WorkflowTransactionError } from "../storage/errors.ts"; +import type { SavepointManager } from "./savepoints.ts"; /** * Every database the current scope holds a transaction on. @@ -100,11 +100,9 @@ export const savepoint: TransactionApi["savepoint"] = Transaction.operations.sav /** What the open transaction installs so `savepoint()` can answer. */ export function useTransactionSavepoints( - database: DatabaseSync, + savepoints: SavepointManager, isOpen: () => boolean, ): Operation { - let depth = 0; - return Transaction.around( { // deno-lint-ignore require-yield @@ -115,20 +113,7 @@ export function useTransactionSavepoints( ); } - const name = `xmd_savepoint_${depth}`; - depth += 1; - database.exec(`SAVEPOINT ${name}`); - try { - const value = body(); - database.exec(`RELEASE ${name}`); - return value; - } catch (error) { - database.exec(`ROLLBACK TO ${name}`); - database.exec(`RELEASE ${name}`); - throw error; - } finally { - depth -= 1; - } + return savepoints.synchronous(body); }, }, { at: "min" }, diff --git a/packages/workflow/src/deno/workspace/empty.ts b/packages/workflow/src/deno/workspace/empty.ts new file mode 100644 index 00000000..e23c4c15 --- /dev/null +++ b/packages/workflow/src/deno/workspace/empty.ts @@ -0,0 +1,144 @@ +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/storage/errors.ts b/packages/workflow/src/storage/errors.ts index 7f919183..ceaabdc3 100644 --- a/packages/workflow/src/storage/errors.ts +++ b/packages/workflow/src/storage/errors.ts @@ -124,6 +124,19 @@ export class WorkflowDatabaseCorruptError extends WorkflowStorageError { } } +/** The unsupported pre-release schema that claimed the now-complete version 1. */ +export class WorkflowIncompleteVersionOneError extends WorkflowDatabaseCorruptError { + override name = "WorkflowIncompleteVersionOneError"; + + constructor(path: string) { + super(path, "it contains the incomplete pre-release version-1 structure"); + this.message = + `The workflow-run database at ${path} contains the unsupported incomplete ` + + "pre-release version-1 structure. It is not migrated or changed. Delete and recreate " + + "this pre-release database with the complete version-1 provider."; + } +} + /** A stored row does not describe what its column claims to hold. */ export class WorkflowRecordMalformedError extends WorkflowStorageError { override name = "WorkflowRecordMalformedError"; diff --git a/packages/workflow/tests/workflow-run-journal.test.ts b/packages/workflow/tests/workflow-run-journal.test.ts index 774969e5..bf4397ee 100644 --- a/packages/workflow/tests/workflow-run-journal.test.ts +++ b/packages/workflow/tests/workflow-run-journal.test.ts @@ -38,6 +38,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 { allowJournalInserts, committedEventCount, @@ -156,6 +157,25 @@ describe("Tier WJ — appending and replaying the journal", () => { expect(serializeDurableEvent(stored[0])).toBe(serializeDurableEvent(event)); }); + it("WJ2b: every ordinary journal append records the current retained root", function* () { + const root = yield* useStorageRoot(); + const path = runPath(root, "release-1.4"); + + yield* withStorage(root, function* () { + const database = yield* createRun(); + yield* database.journal.append(yielded("first", "one")); + yield* database.transact(function* (transaction) { + yield* transaction.journal.append(yielded("second", "two")); + }); + }); + + tamper(path, (database) => { + expect( + database.prepare("SELECT DISTINCT workspace_root_id FROM journal_events").all(), + ).toEqual([{ workspace_root_id: EMPTY_WORKSPACE_ROOT_ID }]); + }); + }); + it("WJ3: an event keeps its opaque id, across reads and across processes", function* () { const root = yield* useStorageRoot(); diff --git a/packages/workflow/tests/workflow-run-storage.test.ts b/packages/workflow/tests/workflow-run-storage.test.ts index f96e13d6..4f519439 100644 --- a/packages/workflow/tests/workflow-run-storage.test.ts +++ b/packages/workflow/tests/workflow-run-storage.test.ts @@ -13,12 +13,13 @@ */ import { readFileSync, readdirSync, writeFileSync } from "node:fs"; +import { join } from "node:path"; import { describe, it } from "@executablemd/test-support/bdd"; import { expect } from "@executablemd/test-support/expect"; import { exists } from "@effectionx/fs"; import { type Result, scoped } from "effection"; import type { Json } from "@executablemd/durable-streams"; -import type { DatabaseSync } from "node:sqlite"; +import { DatabaseSync } from "node:sqlite"; import { type CreateWorkflowRunRequest, type GitWorkflowDefinitionV1, @@ -27,6 +28,7 @@ import { WorkflowDatabaseCorruptError, WorkflowDatabaseFormatError, WorkflowDefinitionError, + WorkflowIncompleteVersionOneError, WorkflowRecordMalformedError, WorkflowRequestError, WorkflowRunConflictError, @@ -40,6 +42,13 @@ import { } from "../mod.ts"; import type { JsonObject } from "../src/storage/members.ts"; import { APPLICATION_ID, hashRunId, useWorkflowRunStorage } from "../deno.ts"; +import { createWorkflowRunConnections } from "../src/deno/connections.ts"; +import { EXPECTED_SCHEMA, initializeSchema } from "../src/deno/schema.ts"; +import { + EMPTY_WORKSPACE_MANIFEST, + EMPTY_WORKSPACE_ROOT_ID, + WORKSPACE_ROOT_FORMAT, +} from "../src/deno/workspace/empty.ts"; import { createRun, definition, @@ -59,6 +68,73 @@ function entries(root: string): string[] { return readdirSync(root); } +function normalizedSchema(database: DatabaseSync): Array<{ + name: string; + type: string; + sql: string; +}> { + return database + .prepare("SELECT name, type, sql FROM sqlite_schema WHERE name NOT LIKE 'sqlite_%'") + .all() + .map((row) => ({ + name: String(row["name"]), + type: String(row["type"]), + sql: String(row["sql"]).replace(/\s+/g, " ").trim(), + })) + .sort((left, right) => left.name.localeCompare(right.name)); +} + +function initializeIntermediateVersionOne(database: DatabaseSync): void { + const statuses = "'running', 'suspended', 'interrupted', 'completed', 'failed', 'cancelled'"; + const reason = `CHECK ( + (stop_reason_kind IS NULL AND stop_reason_code IS NULL AND stop_reason_event_id IS NULL) + OR (stop_reason_kind = 'host' AND stop_reason_code IS NOT NULL AND stop_reason_event_id IS NULL) + OR (stop_reason_kind = 'journal' AND stop_reason_code IS NULL AND stop_reason_event_id IS NOT NULL) + )`; + database.exec(` + PRAGMA application_id = ${APPLICATION_ID}; + PRAGMA user_version = 1; + CREATE TABLE journal_events ( + sequence INTEGER PRIMARY KEY AUTOINCREMENT, + event_id TEXT NOT NULL UNIQUE, + record TEXT NOT NULL CHECK (json_valid(record)) + ) STRICT; + CREATE TABLE workflow_run ( + id INTEGER PRIMARY KEY CHECK (id = 1), + run_id TEXT NOT NULL, + definition TEXT NOT NULL CHECK (json_valid(definition)), + base TEXT NOT NULL, + props TEXT NOT NULL CHECK (json_valid(props) AND json_type(props) = 'object'), + status TEXT NOT NULL CHECK (status IN (${statuses})), + stop_reason_kind TEXT CHECK (stop_reason_kind IS NULL OR stop_reason_kind IN ('host', 'journal')), + stop_reason_code TEXT, + stop_reason_event_id TEXT REFERENCES journal_events (event_id), + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + ${reason} + ) STRICT; + CREATE TABLE definition_retrieval ( + id INTEGER PRIMARY KEY CHECK (id = 1), + metadata TEXT NOT NULL CHECK (json_valid(metadata)), + revision INTEGER NOT NULL CHECK (revision >= 1 AND revision <= 9007199254740991), + updated_at TEXT NOT NULL + ) STRICT; + CREATE TABLE document_executions ( + sequence INTEGER PRIMARY KEY AUTOINCREMENT, + execution_id TEXT NOT NULL UNIQUE, + started_at TEXT NOT NULL, + stopped_at TEXT, + stop_status TEXT CHECK (stop_status IS NULL OR stop_status IN (${statuses})), + stop_reason_kind TEXT CHECK (stop_reason_kind IS NULL OR stop_reason_kind IN ('host', 'journal')), + stop_reason_code TEXT, + stop_reason_event_id TEXT REFERENCES journal_events (event_id), + CHECK ((stopped_at IS NULL) = (stop_status IS NULL)), + CHECK (stop_status IS NOT NULL OR stop_reason_kind IS NULL), + ${reason} + ) STRICT; + `); +} + /** * A props value the interface forbids, for testing that storage checks anyway. * @@ -84,6 +160,109 @@ function fabricatedId(value: unknown): string { return container.id; } +describe("Tier WS — authoritative connection and complete schema", () => { + it("WS0: one provider entry owns each run connection and its DOFS wrappers", function* () { + const root = yield* useStorageRoot(); + const connections = createWorkflowRunConnections(); + const first = connections.at(join(root, "first.sqlite")); + const again = connections.at(join(root, ".", "first.sqlite")); + const other = connections.at(join(root, "other.sqlite")); + + expect(again).toBe(first); + expect(again.database).toBe(first.database); + expect(again.dofs).toBe(first.dofs); + expect(again.filesystem).toBe(first.filesystem); + expect(other).not.toBe(first); + + connections.close(); + expect(() => first.database.prepare("SELECT 1")).toThrow(); + expect(() => connections.at(join(root, "later.sqlite"))).toThrow(); + }); + + it("WS0b: DOFS transactionSync uses a savepoint in the caller-owned transaction", function* () { + const root = yield* useStorageRoot(); + const connections = createWorkflowRunConnections(); + const connection = connections.at(join(root, "savepoint.sqlite")); + + connection.database.exec("BEGIN IMMEDIATE"); + connection.transactionOpen = true; + expect(() => + connection.dofs.transactionSync(() => { + connection.dofs.run("CREATE TABLE rolled_back (value TEXT)"); + throw new Error("roll back the nested work"); + }), + ).toThrow(Error); + connection.database.exec("CREATE TABLE outer_survives (value TEXT)"); + connection.transactionOpen = false; + connection.database.exec("COMMIT"); + + const names = connection.database + .prepare("SELECT name FROM sqlite_schema WHERE type = 'table'") + .all() + .map((row) => row["name"]); + expect(names).toContain("outer_survives"); + expect(names).not.toContain("rolled_back"); + connections.close(); + }); + + it("WS0c: rolling back fresh initialization leaves no partial schema", function* () { + const root = yield* useStorageRoot(); + const connections = createWorkflowRunConnections(); + const connection = connections.at(join(root, "atomic.sqlite")); + + connection.database.exec("BEGIN IMMEDIATE"); + connection.transactionOpen = true; + expect(() => + initializeSchema(connection.database, connection.dofs, () => { + throw new Error("fail after the filesystem and root are initialized"); + }), + ).toThrow(Error); + connection.transactionOpen = false; + connection.database.exec("ROLLBACK"); + + expect(normalizedSchema(connection.database)).toEqual([]); + expect(connection.database.prepare("PRAGMA application_id").get()?.["application_id"]).toBe(0); + expect(connection.database.prepare("PRAGMA user_version").get()?.["user_version"]).toBe(0); + connections.close(); + }); + + it("WS0d: a fresh run has the frozen complete-v1 schema and canonical empty root", function* () { + const root = yield* useStorageRoot(); + yield* withStorage(root, function* () { + yield* createRun(); + }); + + const database = new DatabaseSync(runPath(root, "release-1.4")); + try { + const expected = [...EXPECTED_SCHEMA].sort((left, right) => + left.name.localeCompare(right.name), + ); + expect(normalizedSchema(database)).toEqual(expected); + expect(database.prepare("PRAGMA application_id").get()?.["application_id"]).toBe( + APPLICATION_ID, + ); + expect(database.prepare("PRAGMA user_version").get()?.["user_version"]).toBe(1); + expect( + database.prepare("SELECT v FROM vfs_meta WHERE k = 'schema_version'").get()?.["v"], + ).toBe(5); + expect(database.prepare("SELECT * FROM workspace_roots").all()).toEqual([ + { + root_id: EMPTY_WORKSPACE_ROOT_ID, + format_version: WORKSPACE_ROOT_FORMAT, + manifest: EMPTY_WORKSPACE_MANIFEST, + }, + ]); + expect(database.prepare("SELECT * FROM workspace_state").all()).toEqual([ + { singleton_id: 1, current_root_id: EMPTY_WORKSPACE_ROOT_ID }, + ]); + expect(database.prepare("SELECT * FROM workspace_root_manifest_refs").all()).toEqual([]); + expect(database.prepare("SELECT * FROM workspace_root_blob_refs").all()).toEqual([]); + } finally { + database.close(); + } + }); +}); + describe("Tier WS — creating and finding a run", () => { it("WS1: a run is one file named for its id, and there is no registry", function* () { const root = yield* useStorageRoot(); @@ -645,6 +824,26 @@ describe("Tier WS — surviving the process", () => { expect(result.ok).toBe(false); expect(!result.ok && result.error).toBeInstanceOf(WorkflowDatabaseClosedError); }); + + it("WS18b: closing one lease leaves another lease on the authoritative entry usable", function* () { + const root = yield* useStorageRoot(); + + yield* withStorage(root, function* () { + const escaped: WorkflowRunDatabase[] = []; + yield* scoped(function* () { + escaped.push(yield* createRun()); + }); + + const current = yield* createRun(); + const closed = yield* escaped[0].updateRunState({ status: "completed" }); + const updated = yield* current.updateRunState({ status: "completed" }); + + expect(closed.ok).toBe(false); + expect(!closed.ok && closed.error).toBeInstanceOf(WorkflowDatabaseClosedError); + expect(updated.ok).toBe(true); + expect(updated.ok && updated.value.status).toBe("completed"); + }); + }); }); describe("Tier WS — refusing what is not this run's database", () => { @@ -752,6 +951,10 @@ describe("Tier WS — refusing what is not this run's database", () => { ["no-row", (database) => database.exec("DELETE FROM workflow_run")], ["relaxed", relaxRunConstraints], ["extra-table", (database) => database.exec("CREATE TABLE souvenirs (a TEXT)")], + [ + "extra-index", + (database) => database.exec("CREATE INDEX souvenirs ON workflow_run(updated_at)"), + ], [ "extra-view", (database) => database.exec("CREATE VIEW shortcut AS SELECT run_id FROM workflow_run"), @@ -759,17 +962,29 @@ describe("Tier WS — refusing what is not this run's database", () => { ]; const results = yield* withStorage(root, function* () { - const seen: { runId: string; result: Result }[] = []; + const seen: { + runId: string; + result: Result; + unchanged: boolean; + }[] = []; for (const [runId, damage] of damaged) { yield* createRun({ runId }); - tamper(runPath(root, runId), damage); - seen.push({ runId, result: yield* lookup(runId) }); + const path = runPath(root, runId); + tamper(path, damage); + const before = readFileSync(path); + const result = yield* lookup(runId); + seen.push({ + runId, + result, + unchanged: readFileSync(path).equals(before), + }); } return seen; }); - for (const { runId, result } of results) { + for (const { runId, result, unchanged } of results) { expect(result.ok).toBe(false); + expect(unchanged).toBe(true); if (result.ok) { continue; } @@ -780,6 +995,107 @@ describe("Tier WS — refusing what is not this run's database", () => { } }); + it("WS22b: the exact intermediate metadata-only version 1 is refused unchanged", function* () { + const root = yield* useStorageRoot(); + const path = runPath(root, "release-1.4"); + tamper(path, initializeIntermediateVersionOne); + const before = readFileSync(path); + + const result = yield* withStorage(root, function* () { + return yield* lookup("release-1.4"); + }); + + expect(result.ok).toBe(false); + expect(!result.ok && result.error).toBeInstanceOf(WorkflowIncompleteVersionOneError); + expect(!result.ok && result.error.message).toContain("Delete and recreate"); + expect(readFileSync(path)).toEqual(before); + }); + + it("WS22c: partial DOFS and retained-root initialization are corruption and stay unchanged", function* () { + const root = yield* useStorageRoot(); + const partials: Array<[string, (database: DatabaseSync) => void]> = [ + [ + "partial-dofs", + (database) => { + database.exec("CREATE TABLE vfs_meta (k TEXT PRIMARY KEY, v INTEGER NOT NULL)"); + }, + ], + [ + "partial-root", + (database) => { + database.exec(` + PRAGMA application_id = ${APPLICATION_ID}; + PRAGMA user_version = 1; + CREATE TABLE workspace_roots ( + root_id TEXT PRIMARY KEY, + format_version INTEGER NOT NULL, + manifest TEXT NOT NULL + ) STRICT; + `); + }, + ], + ]; + + for (const [runId, initialize] of partials) { + const path = runPath(root, runId); + tamper(path, initialize); + const before = readFileSync(path); + const result = yield* withStorage(root, function* () { + return yield* lookup(runId); + }); + + expect(result.ok).toBe(false); + expect(!result.ok && result.error).toBeInstanceOf(WorkflowDatabaseCorruptError); + expect(readFileSync(path)).toEqual(before); + } + }); + + it("WS22d: malformed empty-root and live-frontier state is refused unchanged", function* () { + const root = yield* useStorageRoot(); + const corruptions: Array<[string, (database: DatabaseSync) => void]> = [ + [ + "malformed-root", + (database) => { + database.prepare("UPDATE workspace_roots SET manifest = '{}'").run(); + }, + ], + [ + "changed-frontier", + (database) => { + database.prepare("UPDATE vfs_nodes SET mtime = 1 WHERE inode = 1").run(); + }, + ], + [ + "unexpected-blob", + (database) => { + const bytes = new Uint8Array([1]); + database + .prepare("INSERT INTO vfs_blobs (hash, size, last_seen) VALUES (?, 1, 0)") + .run(bytes); + database + .prepare("INSERT INTO vfs_blob_bytes (hash, bytes) VALUES (?, ?)") + .run(bytes, bytes); + }, + ], + ]; + + for (const [runId, corrupt] of corruptions) { + yield* withStorage(root, function* () { + yield* createRun({ runId }); + }); + const path = runPath(root, runId); + tamper(path, corrupt); + const before = readFileSync(path); + + const result = yield* withStorage(root, function* () { + return yield* lookup(runId); + }); + expect(result.ok).toBe(false); + expect(!result.ok && result.error).toBeInstanceOf(WorkflowDatabaseCorruptError); + expect(readFileSync(path)).toEqual(before); + } + }); + it("WS23: a stored descriptor that describes no definition is refused", function* () { const root = yield* useStorageRoot(); const path = runPath(root, "release-1.4"); @@ -992,7 +1308,10 @@ describe("Tier WS — refusing what is not this run's database", () => { tamper(path, (database) => { for (let index = 0; index < 400; index++) { database - .prepare("INSERT INTO journal_events (event_id, record) VALUES (?, ?)") + .prepare( + `INSERT INTO journal_events (event_id, record, workspace_root_id) + SELECT ?, ?, current_root_id FROM workspace_state WHERE singleton_id = 1`, + ) .run(`e${index}`, JSON.stringify({ padding: "x".repeat(200), index })); } }); From 02a736331b96220830a746b1da5cd9166b6f8ad1 Mon Sep 17 00:00:00 2001 From: Taras Mankovski <74687+taras@users.noreply.github.com> Date: Sat, 8 Aug 2026 04:22:24 -0400 Subject: [PATCH 2/2] =?UTF-8?q?=F0=9F=90=9B=20Narrow=20WorkflowRun=20stora?= =?UTF-8?q?ge=20failure=20classification?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- architecture.md | 53 ++++++++--- packages/workflow/src/deno/database.ts | 50 +++++++--- packages/workflow/src/deno/schema.ts | 18 ++-- .../tests/workflow-run-journal.test.ts | 36 +++++++ .../tests/workflow-run-storage.test.ts | 94 +++++++++++++++---- specs/workflow-spec.md | 68 ++++++++++---- 6 files changed, 251 insertions(+), 68 deletions(-) diff --git a/architecture.md b/architecture.md index 3ca34a7e..b0e1810e 100644 --- a/architecture.md +++ b/architecture.md @@ -152,9 +152,11 @@ imports them or detects the active runtime. The absence of a provider is reported rather than answered with an empty store: a run that appears to start and retains nothing has not started. -A storage handle is owned by the scope that opened it. Its connection closes -through ordinary teardown, and a call after that scope closes fails rather than -reopening anything. +A storage handle is a lease owned by the scope that opened it. Lease teardown +makes that handle unusable without closing the run's physical connection or +invalidating another handle. The Deno provider owns the authoritative +SQLite/DOFS connection for each canonical workflow-run database path and closes +it at provider-scope teardown after its child scopes finish. ### Identity is separate from retrieval @@ -198,6 +200,15 @@ A stop reason is a categorical host code or a reference to an already-filtered journal event. Arbitrary exception text is never duplicated outside the journal that filtered it. +WorkflowRun schema version 1 is complete in place. It contains the run records, +filtered journal, pinned Cloudflare DOFS version-5 structure, immutable +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. + Which status transitions are legal, and what a caller may do to a run in each of them, is lifecycle policy applied above storage. @@ -215,10 +226,13 @@ public identifier. ### One database at a time, and a transaction a caller can hold Operations on one run's storage are serialized, and each runs inside a -transaction. Turns belong to the storage a run lives in rather than to a handle -on it, so two handles for one run take turns instead of contending — a host -whose storage is reached synchronously would otherwise stop while a second -handle waited for a transaction the first one cannot resume to finish. +transaction. The Deno provider's per-path entry owns the one physical SQLite +connection, the one Cloudflare DOFS wrapper and Workspace filesystem, the +cooperative connection queue, and the savepoint allocator. Turns belong to that +entry rather than to a handle, so two leases for one run take turns instead of +contending — a host whose storage is reached synchronously would otherwise stop +while a second handle waited for a transaction the first one cannot resume to +finish. Different workflow-run paths have independent entries. A caller that must publish several changes together holds the transaction itself and receives a handle for taking part in it. Enlistment travels with @@ -253,6 +267,10 @@ it is not shaped like is damage rather than a version this build has not learned. Storage is initialized only when it is pristine: something that merely looks unused is not. +Complete version 1 is the first XMD schema. A database carrying the XMD +application identity with schema version zero is a partial initialization and +therefore damage, not a supported historical version. + Records are held to what they mean and not only to the types they are stored in. A retained timestamp is an instant, a retained identity is not empty, and normalized props are an object. @@ -366,12 +384,17 @@ again. ## Local Workspace topology The local workflow host owns SQLite directly in Deno and reuses Cloudflare's -DOFS filesystem layer behind the provider-neutral Workspace boundary. The -journal and DOFS adapter share the operation-scoped transaction. One -authoritative host-owned DOFS connection serves each workflow database, and the -host serializes its Workspace-local effect transactions. A second long-lived -DOFS connection is not a coherent reader because provider caches may retain -negative entries across another connection's commit. +DOFS filesystem layer behind the provider-neutral Workspace boundary. One +authoritative provider-owned connection entry serves each canonical workflow +database path until provider teardown. The journal and DOFS adapter use that +same SQLite connection; Cloudflare's synchronous initialization transactions +become uniquely named savepoints inside XMD's caller-owned transaction. A +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. The initial topology requires neither writable FUSE nor native subprocess access and does not bundle `workerd`. A Cloudflare-hosted or workerd-backed @@ -660,7 +683,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 and filtered journal | 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 | | 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 | @@ -671,7 +694,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 | stores the authoritative local Workspace in SQLite | persistence POC complete; effect-transaction integration 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 | | 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/database.ts b/packages/workflow/src/deno/database.ts index 02e22e6c..afdaf7bc 100644 --- a/packages/workflow/src/deno/database.ts +++ b/packages/workflow/src/deno/database.ts @@ -62,7 +62,7 @@ import { useTransactionSavepoints, } from "./transaction.ts"; import { readDocumentExecution, readRetrieval, readRunRecord, stopReasonColumns } from "./rows.ts"; -import { translateSqliteError } from "./schema.ts"; +import { isSqliteForeignKeyConstraint, translateSqliteError } from "./schema.ts"; const SELECT_RUN = "SELECT * FROM workflow_run WHERE id = 1"; const UPDATE_RUN_STATE = `UPDATE workflow_run @@ -319,16 +319,20 @@ function createHandle(connection: OpenConnection): Handle { const stoppedAt = now(); return yield* write(() => { - const changed = database - .prepare(FINISH_EXECUTION) - .run( - stoppedAt, - completion.status, - columns.kind, - columns.code, - columns.eventId, - completion.executionId, - ); + const changed = runStopReasonStatement( + () => + database + .prepare(FINISH_EXECUTION) + .run( + stoppedAt, + completion.status, + columns.kind, + columns.code, + columns.eventId, + completion.executionId, + ), + path, + ); if (changed.changes === 0) { throw new WorkflowDocumentExecutionError(completion.executionId); } @@ -352,9 +356,13 @@ function createHandle(connection: OpenConnection): Handle { const updatedAt = now(); const written = yield* write(() => { - database - .prepare(UPDATE_RUN_STATE) - .run(state.status, columns.kind, columns.code, columns.eventId, updatedAt); + runStopReasonStatement( + () => + database + .prepare(UPDATE_RUN_STATE) + .run(state.status, columns.kind, columns.code, columns.eventId, updatedAt), + path, + ); return readRunRow(database, path); }); if (!written.ok) { @@ -441,6 +449,20 @@ function inTransaction(database: DatabaseSync, path: string, body: () => T): } } +function runStopReasonStatement(body: () => T, path: string): T { + try { + return body(); + } catch (error) { + if (isSqliteForeignKeyConstraint(error)) { + throw new WorkflowRequestError( + "the stop reason names a journal event this run does not hold. A journal reason " + + "points at an event that has already been appended and filtered.", + ); + } + throw translateSqliteError(error, path); + } +} + /** * Roll back without reporting a failure of its own. * diff --git a/packages/workflow/src/deno/schema.ts b/packages/workflow/src/deno/schema.ts index c063409d..47aa422f 100644 --- a/packages/workflow/src/deno/schema.ts +++ b/packages/workflow/src/deno/schema.ts @@ -30,7 +30,6 @@ import { WorkflowDatabaseCorruptError, WorkflowDatabaseFormatError, WorkflowIncompleteVersionOneError, - WorkflowRequestError, WorkflowSchemaVersionError, } from "../storage/errors.ts"; import { initializeEmptyWorkspace, verifyEmptyWorkspace } from "./workspace/empty.ts"; @@ -452,6 +451,12 @@ export function verifySchema(database: DatabaseSync, path: string): void { } const version = readPragmaNumber(database, "user_version", path); + if (version === 0) { + throw new WorkflowDatabaseCorruptError( + path, + "it carries the XMD application identity without a complete version-1 schema", + ); + } if (version !== SCHEMA_VERSION) { throw new WorkflowSchemaVersionError(path, version, SCHEMA_VERSION); } @@ -605,7 +610,7 @@ const SQLITE_CORRUPT = 11; /** `SQLITE_NOTADB`: the bytes are not a SQLite database at all. */ const SQLITE_NOTADB = 26; -/** `SQLITE_CONSTRAINT_FOREIGNKEY`: a requested journal reference does not exist. */ +/** `SQLITE_CONSTRAINT_FOREIGNKEY`: one statement violated a foreign key. */ const SQLITE_CONSTRAINT_FOREIGNKEY = 787; /** @@ -621,16 +626,15 @@ export function translateSqliteError(error: unknown, path: string): unknown { return new WorkflowDatabaseFormatError(path, "SQLite does not recognize it as a database"); case SQLITE_CORRUPT: return new WorkflowDatabaseCorruptError(path, "SQLite reported a damaged image"); - case SQLITE_CONSTRAINT_FOREIGNKEY: - return new WorkflowRequestError( - "the stop reason names a journal event this run does not hold. A journal reason " + - "points at an event that has already been appended and filtered.", - ); default: return error; } } +export function isSqliteForeignKeyConstraint(error: unknown): boolean { + return sqliteErrorCode(error) === SQLITE_CONSTRAINT_FOREIGNKEY; +} + /** The SQLite result code behind a failure, when SQLite is what raised it. */ function sqliteErrorCode(error: unknown): number | undefined { if (!(error instanceof Error) || !("code" in error) || error.code !== "ERR_SQLITE_ERROR") { diff --git a/packages/workflow/tests/workflow-run-journal.test.ts b/packages/workflow/tests/workflow-run-journal.test.ts index bf4397ee..35273a69 100644 --- a/packages/workflow/tests/workflow-run-journal.test.ts +++ b/packages/workflow/tests/workflow-run-journal.test.ts @@ -32,6 +32,7 @@ import { createSecretScanner, SecretDetectedError } from "@executablemd/core"; import { all, ensure, type Operation, race, sleep, spawn, suspend, withResolvers } from "effection"; import { WorkflowRecordMalformedError, + WorkflowRequestError, WorkflowRunConflictError, WorkflowRunStorage, type WorkflowRunTransaction, @@ -363,6 +364,41 @@ describe("Tier WJ — what reaches SQLite", () => { expect(seen.events).toEqual([]); }); + it("WJ5e: a retained-manifest FK failure stays a storage failure", function* () { + const root = yield* useStorageRoot(); + const path = runPath(root, "release-1.4"); + + const seen = yield* withStorage(root, function* () { + const database = yield* createRun(); + tamper(path, (raw) => { + raw.exec(` + CREATE TRIGGER fail_second_journal_insert + BEFORE INSERT ON journal_events + WHEN (SELECT COUNT(*) FROM journal_events) = 1 + BEGIN + INSERT INTO workspace_root_manifest_refs (root_id, manifest_hash) + VALUES ('${EMPTY_WORKSPACE_ROOT_ID}', X'00'); + END; + `); + }); + + const result = yield* database.transact(function* (transaction) { + yield* transaction.journal.append(yielded("companion", "companion")); + yield* transaction.journal.append(yielded("fails-on-manifest-fk", "doomed")); + }); + + return { result, events: yield* database.journal.readAll() }; + }); + + expect(seen.result.ok).toBe(false); + expect(!seen.result.ok && seen.result.error).toBeInstanceOf(Error); + expect(!seen.result.ok && seen.result.error).not.toBeInstanceOf(WorkflowRequestError); + expect(seen.events).toEqual([]); + tamper(path, (database) => { + expect(database.prepare("SELECT * FROM workspace_root_manifest_refs").all()).toEqual([]); + }); + }); + it("WJ6: a gate cancelled mid-scan produces no row either", function* () { const root = yield* useStorageRoot(); diff --git a/packages/workflow/tests/workflow-run-storage.test.ts b/packages/workflow/tests/workflow-run-storage.test.ts index 4f519439..0d4a7d92 100644 --- a/packages/workflow/tests/workflow-run-storage.test.ts +++ b/packages/workflow/tests/workflow-run-storage.test.ts @@ -556,6 +556,34 @@ describe("Tier WS — what a run retains", () => { expect(!result.ok && result.error).toBeInstanceOf(WorkflowRequestError); }); + it("WS11c: an execution stop reason naming a missing event is refused", function* () { + const root = yield* useStorageRoot(); + + const seen = yield* withStorage(root, function* () { + const database = yield* createRun(); + const started = yield* database.beginDocumentExecution(); + if (!started.ok) { + throw started.error; + } + + const result = yield* database.finishDocumentExecution({ + executionId: started.value.executionId, + status: "failed", + reason: { kind: "journal", eventId: "an-event-that-was-never-appended" }, + }); + const executions = yield* database.readDocumentExecutions(); + if (!executions.ok) { + throw executions.error; + } + return { result, execution: executions.value[0] }; + }); + + expect(seen.result.ok).toBe(false); + expect(!seen.result.ok && seen.result.error).toBeInstanceOf(WorkflowRequestError); + expect(seen.execution.stoppedAt).toBeUndefined(); + expect(seen.execution.stopStatus).toBeUndefined(); + }); + it("WS12: a stop reason is parsed on the way in, not only type-checked", function* () { const root = yield* useStorageRoot(); @@ -882,29 +910,63 @@ describe("Tier WS — refusing what is not this run's database", () => { expect(!result.ok && result.error).toBeInstanceOf(WorkflowDatabaseFormatError); }); - it("WS21: neither an older nor a newer schema version is read or migrated", function* () { + it("WS21: an unsupported schema version is not read or migrated", function* () { + const root = yield* useStorageRoot(); + const path = runPath(root, "run-2"); + + const result = yield* withStorage(root, function* () { + yield* createRun({ runId: "run-2" }); + tamper(path, (database) => { + database.exec("PRAGMA user_version = 2"); + }); + return yield* lookup("run-2"); + }); + + expect(result.ok).toBe(false); + expect(!result.ok && result.error).toBeInstanceOf(WorkflowSchemaVersionError); + + tamper(path, (database) => { + expect(database.prepare("PRAGMA user_version").get()?.["user_version"]).toBe(2); + expect(database.prepare("PRAGMA application_id").get()?.["application_id"]).toBe( + APPLICATION_ID, + ); + }); + }); + + it("WS21a: XMD-identified version zero is partial initialization", function* () { const root = yield* useStorageRoot(); + const partials: Array<[string, (database: DatabaseSync) => void]> = [ + [ + "identified-v0", + (database) => { + database.exec(`PRAGMA application_id = ${APPLICATION_ID}; PRAGMA user_version = 0`); + }, + ], + [ + "identified-v0-dofs", + (database) => { + database.exec(` + PRAGMA application_id = ${APPLICATION_ID}; + PRAGMA user_version = 0; + CREATE TABLE vfs_meta (k TEXT PRIMARY KEY, v INTEGER NOT NULL); + `); + }, + ], + ]; - for (const version of [0, 2]) { - const path = runPath(root, `run-${version}`); + for (const [runId, initialize] of partials) { + const path = runPath(root, runId); + tamper(path, initialize); + const before = readFileSync(path); const result = yield* withStorage(root, function* () { - yield* createRun({ runId: `run-${version}` }); - tamper(path, (database) => { - database.exec(`PRAGMA user_version = ${version}`); - }); - return yield* lookup(`run-${version}`); + return yield* lookup(runId); }); expect(result.ok).toBe(false); - expect(!result.ok && result.error).toBeInstanceOf(WorkflowSchemaVersionError); - - tamper(path, (database) => { - expect(database.prepare("PRAGMA user_version").get()?.["user_version"]).toBe(version); - expect(database.prepare("PRAGMA application_id").get()?.["application_id"]).toBe( - APPLICATION_ID, - ); - }); + expect(!result.ok && result.error).toBeInstanceOf(WorkflowDatabaseCorruptError); + expect(!result.ok && result.error).not.toBeInstanceOf(WorkflowSchemaVersionError); + expect(readFileSync(path)).toEqual(before); } }); diff --git a/specs/workflow-spec.md b/specs/workflow-spec.md index db66eb35..0a47a094 100644 --- a/specs/workflow-spec.md +++ b/specs/workflow-spec.md @@ -178,9 +178,10 @@ after an interruption. The Deno host installs its own with entrypoint is the only place SQLite, run-id hashing, filesystem paths and host behavior appear. Shared modules import none of them and detect no runtime. -A handle belongs to the scope that asked for it. Its connection closes through -ordinary teardown, and every later call answers with a closed-handle failure -rather than reopening the file. +A handle is a lease belonging to the scope that asked for it. Lease teardown +makes that handle unusable, and every later call answers with a closed-handle +failure rather than reopening the file. It does not close the run's physical +connection or invalidate another lease. ### 9.1 What identifies a run @@ -264,6 +265,20 @@ collision or tampering, reported as its own failure and left unchanged. it was last cleared. - The filtered journal. +Complete WorkflowRun schema version 1 also contains the pinned Cloudflare DOFS +version-5 tables and indexes, immutable Workspace-root tables, exact root-to- +manifest and root-to-blob reference tables, singleton current-root state, and a +non-null Workspace-root association on every journal event. XMD schema version +1, DOFS schema version 5 and Workspace-root format version 1 are independent +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. + ### 9.5 The journal `WorkflowRunDatabase.journal` is an ordinary `DurableStream`, so `durableRun` @@ -276,6 +291,10 @@ An event's opaque id is stored separately from its physical position. The id is stable once written and is what a journal stop reason points at; the position is what ordering uses and is not a public identifier. +Every journal row also names the retained Workspace root current when the row +is inserted. Existing non-Workspace appends use the canonical empty current +root and otherwise retain their established behavior. + Events arrive already filtered: ```text @@ -286,11 +305,18 @@ Storage performs no filtering of its own — a second policy in a second place i a second thing to keep in agreement with the first — and a gate that rejects or is cancelled leaves no row at all. -### 9.6 One connection, one operation +### 9.6 One authoritative connection, one operation -Operations on one handle are serialized, and each runs inside a transaction. A -caller that needs several statements published together holds the transaction -itself: +The Deno provider maps each canonical workflow-run database path to one +authoritative entry. The entry owns one physical SQLite connection, one +Cloudflare DOFS database wrapper, one Workspace filesystem, one cooperative +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. + +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: ```ts yield* database.transact(function* (transaction) { @@ -312,10 +338,14 @@ committing first would leave those appends to publish themselves, outside the transaction that was meant to decide about them. The transaction is closed to further appends before the commit rather than after it. -Turns are taken per database rather than per handle. Two handles on one run -share them, so a second handle waits while the first holds the database instead -of entering SQLite and stopping the host. Contention between processes remains -SQLite's own. +Turns are taken through the authoritative entry rather than per handle. Two +leases on one run share them, so a second lease waits cooperatively while the +first holds the connection instead of entering synchronous SQLite and stopping +the host. Contention between processes remains SQLite's own. + +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. 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 @@ -351,6 +381,10 @@ A database is initialized only when it is pristine — no application id, no schema version and not one object anybody created. A file carrying a version but no tables, or tables belonging to something else, is not empty. +Complete version 1 is the first XMD schema. An XMD-identified database carrying +schema version zero is a partial initialization and is reported as corrupt. A +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. @@ -363,12 +397,14 @@ An incompatible or damaged database is described and left exactly as it was found. Nothing initializes, migrates, truncates, deletes or replaces one, and a lookup that finds nothing creates no file. -Version 1 reads and writes version 1. An older version with no implemented -migration, and every newer version, are refused without the file being touched. +Version 1 reads and writes version 1. Unsupported versions are refused without +the file being touched; partial version-1 initialization is corruption and is +also left unchanged. ## 10. Intentionally excluded Public `xmd workflow` lifecycle commands; lifecycle transition policy, executor -leases and stale-owner recovery; Workspace filesystem storage and its -transactions; history checkpoints and forks; workflow-owned worktrees; and -deterministic Git and GitHub effects. +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.