From 6ef0e7dd7f523d1dadd0211488b838a6282103c2 Mon Sep 17 00:00:00 2001 From: Taras Mankovski <74687+taras@users.noreply.github.com> Date: Sat, 8 Aug 2026 17:02:13 -0400 Subject: [PATCH 1/2] =?UTF-8?q?=F0=9F=94=92=20Fence=20WorkflowRun=20transa?= =?UTF-8?q?ctions=20and=20unify=20savepoints?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- architecture.md | 34 +- packages/workflow/src/deno/connections.ts | 280 ++++++++++- packages/workflow/src/deno/database.ts | 124 +++-- packages/workflow/src/deno/provider.ts | 25 +- packages/workflow/src/deno/savepoints.ts | 130 ++++- packages/workflow/src/deno/transaction.ts | 18 +- .../workflow/src/deno/workspace/filesystem.ts | 18 +- .../workflow/src/deno/workspace/private.ts | 171 +++++-- .../workflow/src/deno/workspace/restore.ts | 12 +- packages/workflow/src/deno/workspace/root.ts | 10 +- .../tests/workflow-run-journal.test.ts | 14 +- .../tests/workflow-run-storage.test.ts | 8 +- .../tests/workspace-root-restoration.test.ts | 48 +- .../workflow/tests/workspace-root.test.ts | 19 +- .../tests/workspace-transaction.test.ts | 472 ++++++++++++++++++ scripts/runtime-test-exclusions.ts | 6 + specs/workflow-spec.md | 33 +- 17 files changed, 1219 insertions(+), 203 deletions(-) create mode 100644 packages/workflow/tests/workspace-transaction.test.ts diff --git a/architecture.md b/architecture.md index 87eb1a68..b1bd4936 100644 --- a/architecture.md +++ b/architecture.md @@ -156,7 +156,10 @@ 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. +it at provider-scope teardown after its child scopes finish. The public handle +contains no raw connection association; the provider owns an exact-object lease +registration and refuses foreign, fabricated, closed, or stale handles before +they can reach SQLite. ### Identity is separate from retrieval @@ -257,6 +260,15 @@ contending — a host whose storage is reached synchronously would otherwise sto while a second handle waited for a transaction the first one cannot resume to finish. Different workflow-run paths have independent entries. +Each entry has an opaque generation identity created with the physical +connection. Reopening the same path after provider teardown creates a different +generation. Each top-level transaction has its own opaque identity and exact +active record, including its path, generation, authorized lease and public +transaction handle. Finishing invalidates that record before commit or rollback +can expose the connection to later work. Retained transaction handles and +tokens stay invalid across a later transaction on the same connection and +across a later provider generation. + A caller that must publish several changes together holds the transaction itself and receives a handle for taking part in it. Enlistment travels with that handle rather than with the storage, so work that never received one @@ -273,6 +285,19 @@ 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. +The active-path context chain only detects and refuses that accidental nesting; +it never authorizes transaction or savepoint use. Authorization comes from the +provider-owned identities behind an adapter-private contextual operation, which +validates exact handle possession each time. + +One monotonically unique allocator serves both synchronous Cloudflare DOFS +savepoints and Effection operation savepoints. The operation form receives an +`Operation`, runs it in a child scope, waits for every child and resource to +finish teardown, and only then releases. Ordinary failure rolls back to and +releases that savepoint while leaving the outer transaction usable. Synchronous +cleanup rolls an open savepoint back on cancellation or halt. A failure to +create, roll back, or release poisons the active transaction, so its top-level +owner cannot commit it. Serialization is not a single-executor policy. Which executor may advance a run is decided above storage. @@ -418,7 +443,9 @@ 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 +become uniquely named savepoints inside XMD's caller-owned transaction. Those +savepoints and operation-spanning savepoints share one allocator and cannot +collide or release one another. A second long-lived DOFS connection is not a coherent reader because provider caches may retain negative entries across another connection's commit. @@ -441,7 +468,8 @@ 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 -operation at this layer. +operation at this layer. Provider-neutral durable coordination, filtered +journal routing, and atomic Workspace effect publication are also absent. The initial topology requires neither writable FUSE nor native subprocess access and does not bundle `workerd`. A Cloudflare-hosted or workerd-backed diff --git a/packages/workflow/src/deno/connections.ts b/packages/workflow/src/deno/connections.ts index 01516f66..3a07a88c 100644 --- a/packages/workflow/src/deno/connections.ts +++ b/packages/workflow/src/deno/connections.ts @@ -1,5 +1,7 @@ import { DatabaseSync } from "node:sqlite"; import { resolve } from "node:path"; +import type { WorkflowRunDatabase, WorkflowRunTransaction } from "../storage/api.ts"; +import { WorkflowTransactionError } from "../storage/errors.ts"; 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"; @@ -10,33 +12,85 @@ import type { SQLStorageLike, } from "../../vendor/cloudflare-computer-dofs/generated/types.d.ts"; import { type ConnectionLock, createConnectionLock } from "./lock.ts"; -import { createSavepointManager, type SavepointManager } from "./savepoints.ts"; +import { + createSavepointManager, + type SavepointManager, + type SavepointObserver, + type SavepointTransaction, +} from "./savepoints.ts"; + +export class ConnectionGeneration { + #opaque = undefined; +} + +export class TransactionIdentity { + #opaque = undefined; +} + +export class WorkflowRunTransactionToken { + #opaque = undefined; +} + +export interface RunConnectionLease { + readonly connection: RunConnection; + readonly generation: ConnectionGeneration; + readonly path: string; + readonly database: WorkflowRunDatabase; + open: boolean; +} + +export interface RunTransaction extends SavepointTransaction { + readonly path: string; + readonly generation: ConnectionGeneration; + readonly identity: TransactionIdentity; + readonly lease: RunConnectionLease | undefined; + handle: WorkflowRunTransaction | undefined; + open: boolean; + failure: unknown | undefined; +} export interface RunConnection { readonly path: string; + readonly generation: ConnectionGeneration; readonly database: DatabaseSync; readonly dofs: CloudflareDatabase; readonly filesystem: WorkspaceFilesystem; readonly lock: ConnectionLock; readonly savepoints: SavepointManager; - transactionOpen: boolean; invalidateDofsCaches(): void; + beginTransaction(lease?: RunConnectionLease): RunTransaction; + bindTransaction(transaction: RunTransaction, handle: WorkflowRunTransaction): void; + validateTransaction(transaction: RunTransaction): void; + finishTransaction(transaction: RunTransaction): void; + currentTransaction(): RunTransaction; setClock(now: () => number): void; close(): void; } export interface WorkflowRunConnections { at(path: string): RunConnection; + registerLease(database: WorkflowRunDatabase, connection: RunConnection): RunConnectionLease; + closeLease(lease: RunConnectionLease): void; + validateLease(database: WorkflowRunDatabase): RunConnectionLease; + authorizeTransaction( + database: WorkflowRunDatabase, + transaction: WorkflowRunTransaction, + ): RunTransaction; + issueToken( + database: WorkflowRunDatabase, + transaction: WorkflowRunTransaction, + ): WorkflowRunTransactionToken; + validateToken(database: WorkflowRunDatabase, token: WorkflowRunTransactionToken): RunTransaction; close(): void; } class SqliteStorage implements SQLStorageLike { readonly database: DatabaseSync; - readonly savepoints: () => SavepointManager; + readonly connection: () => RunConnection; - constructor(database: DatabaseSync, savepoints: () => SavepointManager) { + constructor(database: DatabaseSync, connection: () => RunConnection) { this.database = database; - this.savepoints = savepoints; + this.connection = connection; } exec>( @@ -53,7 +107,7 @@ class SqliteStorage implements SQLStorageLike { } } -function createConnection(path: string): RunConnection { +function createConnection(path: string, observeSavepoint: SavepointObserver): RunConnection { const database = new DatabaseSync(path); try { database.exec("PRAGMA foreign_keys = ON"); @@ -63,66 +117,193 @@ function createConnection(path: string): RunConnection { throw error; } + const generation = new ConnectionGeneration(); let open = true; - const connection: { - savepoints: SavepointManager | undefined; - transactionOpen: boolean; - } = { savepoints: undefined, transactionOpen: false }; + let active: RunTransaction | undefined; + let installed: RunConnection | undefined; + + function validate(transaction: SavepointTransaction): void { + if (!open || active !== transaction || !transaction.open) { + throw new WorkflowTransactionError( + "the caller-owned workflow transaction is missing, foreign, stale, or already finished.", + ); + } + if (active.path !== path || active.generation !== generation) { + throw new WorkflowTransactionError( + "the caller-owned workflow transaction does not belong to this connection generation.", + ); + } + if (active.failure !== undefined) { + throw new WorkflowTransactionError( + "the caller-owned workflow transaction cannot continue after a savepoint failure.", + ); + } + } + + const savepoints = createSavepointManager( + database, + { + validate, + poison(transaction, failure): void { + if (active === transaction && transaction.open) { + active.failure = failure; + } + }, + }, + observeSavepoint, + ); const storage = new SqliteStorage(database, () => { - const savepoints = connection.savepoints; - if (savepoints === undefined) { - throw new WorkflowConnectionStateError("the savepoint manager is not installed"); + if (installed === undefined) { + throw new WorkflowConnectionStateError("the workflow connection is not installed"); } - return savepoints; + return installed; }); const durableStorage: DurableObjectStorageLike = { sql: storage, transactionSync(closure: () => T): T { - return storage.savepoints().synchronous(closure); + const connection = storage.connection(); + return connection.savepoints.synchronous(connection.currentTransaction(), closure); }, }; const dofs = new CloudflareDatabase(durableStorage); - const savepoints = createSavepointManager(database, () => connection.transactionOpen); - connection.savepoints = savepoints; let clock = Date.now; - return { + const connection: RunConnection = { path, + generation, database, dofs, filesystem: new WorkspaceFilesystem(dofs, { now: () => clock() }), lock: createConnectionLock(), savepoints, - get transactionOpen() { - return connection.transactionOpen; + + beginTransaction(lease?: RunConnectionLease): RunTransaction { + if (!open || active !== undefined) { + throw new WorkflowTransactionError( + "the authoritative workflow connection cannot open another transaction.", + ); + } + if ( + lease !== undefined && + (!lease.open || lease.connection !== connection || lease.generation !== generation) + ) { + throw new WorkflowTransactionError( + "the workflow database lease is foreign, stale, or already closed.", + ); + } + const transaction: RunTransaction = { + path, + generation, + identity: new TransactionIdentity(), + lease, + handle: undefined, + open: true, + failure: undefined, + }; + active = transaction; + return transaction; }, - set transactionOpen(value: boolean) { - connection.transactionOpen = value; + + bindTransaction(transaction: RunTransaction, handle: WorkflowRunTransaction): void { + validate(transaction); + if (transaction.lease === undefined || transaction.handle !== undefined) { + throw new WorkflowTransactionError( + "the caller-owned workflow transaction cannot be associated with this handle.", + ); + } + transaction.handle = handle; }, + invalidateDofsCaches(): void { clearResolveCache(dofs); clearBlobCache(dofs); }, + + validateTransaction(transaction: RunTransaction): void { + validate(transaction); + }, + + finishTransaction(transaction: RunTransaction): void { + if (!open || active !== transaction || !transaction.open) { + throw new WorkflowTransactionError( + "the caller-owned workflow transaction is missing, foreign, stale, or already finished.", + ); + } + transaction.open = false; + active = undefined; + }, + + currentTransaction(): RunTransaction { + if (active === undefined) { + throw new WorkflowTransactionError( + "a DOFS savepoint needs an active caller-owned workflow transaction.", + ); + } + validate(active); + return active; + }, + setClock(now: () => number): void { clock = now; }, - close() { + + close(): void { if (open) { open = false; + if (active !== undefined) { + active.open = false; + active = undefined; + } database.close(); } }, }; + installed = connection; + return connection; } export class WorkflowConnectionStateError extends Error { override name = "WorkflowConnectionStateError"; } -export function createWorkflowRunConnections(): WorkflowRunConnections { +export function createWorkflowRunConnections( + observeSavepoint: SavepointObserver = () => {}, +): WorkflowRunConnections { const entries = new Map(); + const leases = new WeakMap(); + const tokens = new WeakMap(); let open = true; + function validateLease(database: WorkflowRunDatabase): RunConnectionLease { + const lease = leases.get(database); + if ( + !open || + lease === undefined || + !lease.open || + lease.connection.generation !== lease.generation || + lease.connection.path !== lease.path + ) { + throw new WorkflowTransactionError( + "the WorkflowRun database handle is foreign, fabricated, stale, or already closed.", + ); + } + return lease; + } + + function authorizeTransaction( + database: WorkflowRunDatabase, + transaction: WorkflowRunTransaction, + ): RunTransaction { + const lease = validateLease(database); + const active = lease.connection.currentTransaction(); + if (active.lease !== lease || active.handle !== transaction) { + throw new WorkflowTransactionError( + "the WorkflowRun transaction handle is missing, foreign, stale, or already finished.", + ); + } + return active; + } + return { at(path: string): RunConnection { if (!open) { @@ -133,11 +314,60 @@ export function createWorkflowRunConnections(): WorkflowRunConnections { if (existing !== undefined) { return existing; } - const created = createConnection(canonical); + const created = createConnection(canonical, observeSavepoint); entries.set(canonical, created); return created; }, + registerLease(database: WorkflowRunDatabase, connection: RunConnection): RunConnectionLease { + if (!open || entries.get(connection.path) !== connection) { + throw new WorkflowTransactionError( + "the WorkflowRun database cannot lease a foreign or stale connection.", + ); + } + const lease: RunConnectionLease = { + connection, + generation: connection.generation, + path: connection.path, + database, + open: true, + }; + leases.set(database, lease); + return lease; + }, + + closeLease(lease: RunConnectionLease): void { + lease.open = false; + }, + + validateLease, + authorizeTransaction, + + issueToken( + database: WorkflowRunDatabase, + transaction: WorkflowRunTransaction, + ): WorkflowRunTransactionToken { + const active = authorizeTransaction(database, transaction); + const token = new WorkflowRunTransactionToken(); + tokens.set(token, active); + return token; + }, + + validateToken( + database: WorkflowRunDatabase, + token: WorkflowRunTransactionToken, + ): RunTransaction { + const lease = validateLease(database); + const transaction = tokens.get(token); + if (transaction === undefined || transaction.lease !== lease) { + throw new WorkflowTransactionError( + "the WorkflowRun transaction token is foreign, fabricated, or stale.", + ); + } + lease.connection.validateTransaction(transaction); + return transaction; + }, + close(): void { if (!open) { return; diff --git a/packages/workflow/src/deno/database.ts b/packages/workflow/src/deno/database.ts index 21ac65a9..813b4fa1 100644 --- a/packages/workflow/src/deno/database.ts +++ b/packages/workflow/src/deno/database.ts @@ -54,7 +54,12 @@ import { type WorkflowRunRecord, } from "../storage/record.ts"; import { insertJournalEvent, readJournalEntries } from "./journal.ts"; -import type { RunConnection } from "./connections.ts"; +import type { + RunConnection, + RunConnectionLease, + RunTransaction, + WorkflowRunConnections, +} from "./connections.ts"; import { ActiveTransaction, enclosing, @@ -88,6 +93,7 @@ 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 connection: RunConnection; + readonly connections: WorkflowRunConnections; readonly record: WorkflowRunRecord; } @@ -111,34 +117,22 @@ 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; + const runConnection = connection.connection; + const { database, path, lock } = runConnection; let closed = false; + let lease: RunConnectionLease | undefined; let record = connection.record; let retrieval = readRetrievalRow(database); + function activeLease(): RunConnectionLease { + if (lease === undefined) { + throw new WorkflowTransactionError("the WorkflowRun database lease is not installed."); + } + return lease; + } + /** Whether this scope may reach the database at all, and why not. */ function* admit(): Operation> { if (closed) { @@ -164,7 +158,7 @@ function createHandle(connection: OpenConnection): Handle { } return yield* scoped(function* (): Operation> { yield* lock.hold(); - return inTransaction(database, path, body); + return inTransaction(runConnection, activeLease(), body); }); } @@ -209,25 +203,40 @@ function createHandle(connection: OpenConnection): Handle { return Err(translateSqliteError(error, path)); } - const transaction = { open: true }; - connection.connection.transactionOpen = true; + let active: RunTransaction; + try { + active = runConnection.beginTransaction(activeLease()); + } catch (error) { + rollback(database); + return Err(translateSqliteError(error, path)); + } 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 (active.open) { + runConnection.finishTransaction(active); + } if (!committed) { rollback(database); connection.connection.invalidateDofsCaches(); } }); + const transaction: WorkflowRunTransaction = { + journal: enlistedJournal(runConnection, active, path), + }; + try { + runConnection.bindTransaction(active, transaction); + } catch (error) { + return Err(translateSqliteError(error, path)); + } + // 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(connection.connection.savepoints, () => transaction.open); + yield* useTransactionSavepoints(runConnection.savepoints, active); try { // The body runs in a scope of its own, so everything it started — @@ -237,21 +246,20 @@ 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(transaction); }); // Closed before the commit, not after: nothing may append to a // transaction whose contents are already decided. - transaction.open = false; - connection.connection.transactionOpen = false; + runConnection.validateTransaction(active); + runConnection.finishTransaction(active); database.exec("COMMIT"); committed = true; return Ok(value); } catch (error) { - transaction.open = false; - connection.connection.transactionOpen = false; + if (active.open) { + runConnection.finishTransaction(active); + } return Err(translateSqliteError(error, path)); } }); @@ -268,9 +276,7 @@ function createHandle(connection: OpenConnection): Handle { }, }; - const handle: DenoWorkflowRunDatabase = { - [DENO_CONNECTION]: connection.connection, - + const handle: WorkflowRunDatabase = { get record() { return record; }, @@ -398,10 +404,13 @@ function createHandle(connection: OpenConnection): Handle { }, }; + lease = connection.connections.registerLease(handle, runConnection); + return { database: handle, close() { closed = true; + connection.connections.closeLease(activeLease()); }, }; } @@ -413,14 +422,15 @@ function createHandle(connection: OpenConnection): Handle { * decides whether these rows survive, and nothing here commits. */ function enlistedJournal( - database: DatabaseSync, - transaction: { open: boolean }, + connection: RunConnection, + transaction: RunTransaction, path: string, ): DurableStream { + const { database } = connection; return { // deno-lint-ignore require-yield *readAll(): Operation { - assertOpen(transaction); + connection.validateTransaction(transaction); try { return readJournalEntries(database).map((entry) => entry.event); } catch (error) { @@ -430,7 +440,7 @@ function enlistedJournal( // deno-lint-ignore require-yield *append(event: DurableEvent): Operation { - assertOpen(transaction); + connection.validateTransaction(transaction); try { insertJournalEvent(database, event); } catch (error) { @@ -440,15 +450,6 @@ function enlistedJournal( }; } -function assertOpen(transaction: { open: boolean }): void { - if (!transaction.open) { - throw new WorkflowTransactionError( - "this transaction has already finished, so nothing more can be appended through it. " + - "A handle kept past the end of the body it was given commits nothing.", - ); - } -} - /** A `DurableStream` reports a failure by raising it, so unwrap and throw. */ function* mustSucceed(operation: Operation>): Operation { const result = yield* operation; @@ -458,17 +459,34 @@ function* mustSucceed(operation: Operation>): Operation { return result.value; } -function inTransaction(database: DatabaseSync, path: string, body: () => T): Result { +function inTransaction( + connection: RunConnection, + lease: RunConnectionLease, + body: () => T, +): Result { + const { database, path } = connection; try { database.exec("BEGIN IMMEDIATE"); } catch (error) { return Err(translateSqliteError(error, path)); } + let transaction: RunTransaction; + try { + transaction = connection.beginTransaction(lease); + } catch (error) { + rollback(database); + return Err(translateSqliteError(error, path)); + } try { const value = body(); + connection.validateTransaction(transaction); + connection.finishTransaction(transaction); database.exec("COMMIT"); return Ok(value); } catch (error) { + if (transaction.open) { + connection.finishTransaction(transaction); + } rollback(database); return Err(translateSqliteError(error, path)); } diff --git a/packages/workflow/src/deno/provider.ts b/packages/workflow/src/deno/provider.ts index 6072c88e..9bab35cb 100644 --- a/packages/workflow/src/deno/provider.ts +++ b/packages/workflow/src/deno/provider.ts @@ -60,11 +60,14 @@ import { openWorkflowRunDatabase, readRunRow } from "./database.ts"; import { createWorkflowRunConnections, type RunConnection, + type RunTransaction, type WorkflowRunConnections, } from "./connections.ts"; import { workflowRunPath } from "./path.ts"; import { readTransaction } from "./reading.ts"; import { initializeSchema, isUninitialized, translateSqliteError, verifySchema } from "./schema.ts"; +import { SavepointObservation } from "./savepoints.ts"; +import { usePrivateWorkspace } from "./workspace/private.ts"; const INSERT_RUN = `INSERT INTO workflow_run (id, run_id, definition, base, props, status, created_at, updated_at) @@ -104,10 +107,11 @@ export const WorkflowRunRecognition = createContext */ export function* useWorkflowRunStorage(options: WorkflowRunStorageOptions): Operation { const root = authorizedRoot(options.root); - const connections = createWorkflowRunConnections(); + const connections = createWorkflowRunConnections(yield* SavepointObservation.get()); yield* ensure(() => { connections.close(); }); + yield* usePrivateWorkspace(connections); yield* WorkflowRunStorage.around( { @@ -190,7 +194,7 @@ function* createWorkflowRun( return Err(new WorkflowRunConflictError(wanted.runId, differing)); } - return Ok(yield* openWorkflowRunDatabase({ connection, record })); + return Ok(yield* openWorkflowRunDatabase({ connection, connections, record })); } catch (error) { return refusal(error, path); } @@ -235,7 +239,7 @@ function* lookupWorkflowRun( return Err(new WorkflowRunIdMismatchError(runId, path)); } - return Ok(yield* openWorkflowRunDatabase({ connection, record: record.value })); + return Ok(yield* openWorkflowRunDatabase({ connection, connections, record: record.value })); } catch (error) { return refusal(error, path); } @@ -265,7 +269,13 @@ function establish( }); database.exec("BEGIN IMMEDIATE"); - connection.transactionOpen = true; + let transaction: RunTransaction; + try { + transaction = connection.beginTransaction(); + } catch (error) { + database.exec("ROLLBACK"); + throw error; + } try { if (isUninitialized(database, path)) { const stamp = new Date().toISOString(); @@ -287,11 +297,14 @@ function establish( verifySchema(database, path, connection.dofs); const record = readRunRow(database, path); - connection.transactionOpen = false; + connection.validateTransaction(transaction); + connection.finishTransaction(transaction); database.exec("COMMIT"); return Ok(record); } catch (error) { - connection.transactionOpen = false; + if (transaction.open) { + connection.finishTransaction(transaction); + } database.exec("ROLLBACK"); throw error; } diff --git a/packages/workflow/src/deno/savepoints.ts b/packages/workflow/src/deno/savepoints.ts index c457458e..4cda5f5d 100644 --- a/packages/workflow/src/deno/savepoints.ts +++ b/packages/workflow/src/deno/savepoints.ts @@ -1,48 +1,140 @@ import type { DatabaseSync } from "node:sqlite"; +import { createContext, ensure, type Operation, scoped } from "effection"; import { WorkflowTransactionError } from "../storage/errors.ts"; +export interface SavepointTransaction { + readonly open: boolean; +} + +export interface SavepointTransactionController { + validate(transaction: SavepointTransaction): void; + poison(transaction: SavepointTransaction, failure: unknown): void; +} + export interface SavepointManager { - synchronous(body: () => T): T; + synchronous(transaction: SavepointTransaction, body: () => T): T; + operation(transaction: SavepointTransaction, body: Operation): Operation; +} + +export interface SavepointObservationEvent { + readonly kind: "create" | "release" | "rollback"; + readonly name: string; +} + +export type SavepointObserver = (event: SavepointObservationEvent) => void; + +export const SavepointObservation = createContext( + "executablemd.workflow.deno.savepoint.observation", + () => {}, +); + +interface OpenSavepoint { + readonly name: string; + open: boolean; } export function createSavepointManager( database: DatabaseSync, - isTransactionOpen: () => boolean, + transactions: SavepointTransactionController, + observe: SavepointObserver = () => {}, ): SavepointManager { let next = 0; - function allocate(): string { - const name = `xmd_savepoint_${next}`; + function report(kind: SavepointObservationEvent["kind"], name: string): void { + try { + observe(Object.freeze({ kind, name })); + } catch { + // Observation cannot change the storage decision it reports. + } + } + + function open(transaction: SavepointTransaction): OpenSavepoint { + transactions.validate(transaction); + const savepoint = { name: `xmd_savepoint_${next}`, open: true }; next += 1; - return name; + try { + database.exec(`SAVEPOINT ${savepoint.name}`); + report("create", savepoint.name); + return savepoint; + } catch (error) { + savepoint.open = false; + transactions.poison(transaction, error); + throw error; + } } - function assertOpen(): void { - if (!isTransactionOpen()) { - throw new WorkflowTransactionError( - "a savepoint needs the caller-owned workflow transaction to remain open.", - ); + function release(transaction: SavepointTransaction, savepoint: OpenSavepoint): void { + transactions.validate(transaction); + if (!savepoint.open) { + throw new WorkflowTransactionError("this savepoint has already finished."); + } + savepoint.open = false; + try { + database.exec(`RELEASE ${savepoint.name}`); + report("release", savepoint.name); + } catch (error) { + transactions.poison(transaction, error); + throw error; } } - function rollback(name: string): void { - database.exec(`ROLLBACK TO ${name}`); - database.exec(`RELEASE ${name}`); + function rollback(transaction: SavepointTransaction, savepoint: OpenSavepoint): void { + if (!savepoint.open) { + return; + } + transactions.validate(transaction); + savepoint.open = false; + let failure: unknown; + try { + database.exec(`ROLLBACK TO ${savepoint.name}`); + } catch (error) { + failure = error; + } + try { + database.exec(`RELEASE ${savepoint.name}`); + } catch (error) { + if (failure === undefined) { + failure = error; + } + } + if (failure !== undefined) { + transactions.poison(transaction, failure); + throw failure; + } + report("rollback", savepoint.name); } return { - synchronous(body: () => T): T { - assertOpen(); - const name = allocate(); - database.exec(`SAVEPOINT ${name}`); + synchronous(transaction: SavepointTransaction, body: () => T): T { + const savepoint = open(transaction); try { const value = body(); - database.exec(`RELEASE ${name}`); + release(transaction, savepoint); return value; } catch (error) { - rollback(name); + rollback(transaction, savepoint); throw error; } }, + + *operation(transaction: SavepointTransaction, body: Operation): Operation { + return yield* scoped(function* () { + const savepoint = open(transaction); + yield* ensure(() => { + rollback(transaction, savepoint); + }); + + try { + const value = yield* scoped(function* () { + return yield* body; + }); + release(transaction, savepoint); + return value; + } catch (error) { + rollback(transaction, savepoint); + throw error; + } + }); + }, }; } diff --git a/packages/workflow/src/deno/transaction.ts b/packages/workflow/src/deno/transaction.ts index 5c6e8bfb..1d979f8a 100644 --- a/packages/workflow/src/deno/transaction.ts +++ b/packages/workflow/src/deno/transaction.ts @@ -18,6 +18,7 @@ import { type Api, createApi } from "@effectionx/context-api"; import { type Context, createContext, type Operation } from "effection"; import { WorkflowTransactionError } from "../storage/errors.ts"; +import type { RunTransaction } from "./connections.ts"; import type { SavepointManager } from "./savepoints.ts"; /** @@ -70,7 +71,7 @@ export interface TransactionApi { * and propagates, leaving the surrounding transaction open and free to * continue or to fail on its own terms. */ - savepoint(body: () => T): Operation; + savepoint(body: Operation): Operation; } /** No transaction is open in this scope, so there is nothing to nest inside. */ @@ -89,7 +90,7 @@ export const Transaction: Api = createApi( "executablemd.workflow.deno.savepoint", { // deno-lint-ignore require-yield - *savepoint(_body: () => T): Operation { + *savepoint(_body: Operation): Operation { throw new NoOpenTransactionError(); }, }, @@ -101,19 +102,12 @@ export const savepoint: TransactionApi["savepoint"] = Transaction.operations.sav /** What the open transaction installs so `savepoint()` can answer. */ export function useTransactionSavepoints( savepoints: SavepointManager, - isOpen: () => boolean, + transaction: RunTransaction, ): Operation { return Transaction.around( { - // deno-lint-ignore require-yield - *savepoint([body]: [() => T]): Operation { - if (!isOpen()) { - throw new WorkflowTransactionError( - "this transaction has already finished, so nothing more can happen inside it.", - ); - } - - return savepoints.synchronous(body); + *savepoint([body]: [Operation]): Operation { + return yield* savepoints.operation(transaction, body); }, }, { at: "min" }, diff --git a/packages/workflow/src/deno/workspace/filesystem.ts b/packages/workflow/src/deno/workspace/filesystem.ts index b8c1dd2d..9f30ac0b 100644 --- a/packages/workflow/src/deno/workspace/filesystem.ts +++ b/packages/workflow/src/deno/workspace/filesystem.ts @@ -32,7 +32,10 @@ export interface DenoWorkspaceFilesystem { link(existingPath: string, newPath: string): Operation; } -export function createDenoWorkspaceFilesystem(connection: RunConnection): DenoWorkspaceFilesystem { +export function createDenoWorkspaceFilesystem( + connection: RunConnection, + authorize: () => void, +): DenoWorkspaceFilesystem { const { dofs, filesystem } = connection; function stat(value: { @@ -49,11 +52,13 @@ export function createDenoWorkspaceFilesystem(connection: RunConnection): DenoWo return { *readFile(path): Operation { + authorize(); const stream = yield* until(filesystem.readFile(path)); return new Uint8Array(yield* until(new Response(stream).arrayBuffer())); }, *readTextFile(path): Operation { + authorize(); const value = yield* until(filesystem.readFile(path, "utf8")); if (typeof value !== "string") { throw new Error("the Workspace text read returned a byte stream"); @@ -62,18 +67,22 @@ export function createDenoWorkspaceFilesystem(connection: RunConnection): DenoWo }, *stat(path): Operation { + authorize(); return stat(yield* until(filesystem.stat(path))); }, *lstat(path): Operation { + authorize(); return stat(yield* until(filesystem.lstat(path))); }, *readlink(path): Operation { + authorize(); return yield* until(filesystem.readlink(path)); }, *readdir(path): Operation { + authorize(); const entries = yield* until(filesystem.readdir(path)); return entries.map((entry: WorkspaceDirentResult) => ({ name: entry.name, @@ -82,32 +91,39 @@ export function createDenoWorkspaceFilesystem(connection: RunConnection): DenoWo }, *writeFile(path, content, mode): Operation { + authorize(); yield* until(filesystem.writeFile(path, content, mode === undefined ? {} : { mode })); }, *mkdir(path, options = {}): Operation { + authorize(); yield* until(filesystem.mkdir(path, options)); }, *remove(path, options = {}): Operation { + authorize(); yield* until(filesystem.rm(path, options)); }, // deno-lint-ignore require-yield *rename(from, to): Operation { + authorize(); renamePath(dofs, from, to); }, *chmod(path, mode): Operation { + authorize(); yield* until(filesystem.chmod(path, mode)); }, *symlink(target, path): Operation { + authorize(); yield* until(filesystem.symlink(target, path)); }, // deno-lint-ignore require-yield *link(existingPath, newPath): Operation { + authorize(); linkFile(dofs, existingPath, newPath); }, }; diff --git a/packages/workflow/src/deno/workspace/private.ts b/packages/workflow/src/deno/workspace/private.ts index 4588723e..a478a334 100644 --- a/packages/workflow/src/deno/workspace/private.ts +++ b/packages/workflow/src/deno/workspace/private.ts @@ -1,6 +1,8 @@ +import { type Api, createApi } from "@effectionx/context-api"; import { type Operation, type Result, scoped } from "effection"; -import type { WorkflowRunDatabase } from "../../storage/api.ts"; -import { workflowRunConnection } from "../database.ts"; +import type { WorkflowRunDatabase, WorkflowRunTransaction } from "../../storage/api.ts"; +import { WorkflowTransactionError } from "../../storage/errors.ts"; +import type { WorkflowRunConnections, WorkflowRunTransactionToken } from "../connections.ts"; import { createDenoWorkspaceFilesystem, type DenoWorkspaceFilesystem } from "./filesystem.ts"; import { type StoredWorkspaceRoot } from "./manifest.ts"; import { @@ -18,38 +20,151 @@ export interface PrivateWorkspaceTransaction { 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); +interface PrivateWorkspaceApi { + transact( + database: WorkflowRunDatabase, + transaction: WorkflowRunTransaction, + body: (workspace: PrivateWorkspaceTransaction) => Operation, + ): Operation; + setClock(database: WorkflowRunDatabase, now: () => number): void; + issueToken( + database: WorkflowRunDatabase, + transaction: WorkflowRunTransaction, + ): WorkflowRunTransactionToken; + validateToken(database: WorkflowRunDatabase, token: WorkflowRunTransactionToken): void; +} + +function unavailable(): never { + throw new WorkflowTransactionError( + "the WorkflowRun handle is not owned by the active Deno storage provider.", + ); +} + +const PrivateWorkspace: Api = createApi( + "executablemd.workflow.deno.workspace.private", + { + // deno-lint-ignore require-yield + *transact( + _database: WorkflowRunDatabase, + _transaction: WorkflowRunTransaction, + _body: (workspace: PrivateWorkspaceTransaction) => Operation, + ): Operation { + return unavailable(); + }, + setClock(_database: WorkflowRunDatabase, _now: () => number): void { + unavailable(); + }, + issueToken( + _database: WorkflowRunDatabase, + _transaction: WorkflowRunTransaction, + ): WorkflowRunTransactionToken { + return unavailable(); + }, + validateToken(_database: WorkflowRunDatabase, _token: WorkflowRunTransactionToken): void { + unavailable(); + }, + }, +); + +export function usePrivateWorkspace(connections: WorkflowRunConnections): Operation { + return PrivateWorkspace.around( + { + *transact([database, transaction, body]: [ + WorkflowRunDatabase, + WorkflowRunTransaction, + (workspace: PrivateWorkspaceTransaction) => Operation, + ]): Operation { + const active = connections.authorizeTransaction(database, transaction); + const connection = active.lease?.connection; + if (connection === undefined) { + return unavailable(); + } + const authorize = () => { + connections.authorizeTransaction(database, transaction); + }; + const workspace: PrivateWorkspaceTransaction = { + filesystem: createDenoWorkspaceFilesystem(connection, authorize), + + // deno-lint-ignore require-yield + *currentRoot(): Operation { + authorize(); + return currentWorkspaceRoot(connection.database, connection.path); + }, + + // deno-lint-ignore require-yield + *capture(options = {}): Operation { + authorize(); + return captureWorkspaceRoot(connection, active, options); + }, + + // deno-lint-ignore require-yield + *restore(rootId, options = {}): Operation { + authorize(); + return restoreWorkspaceRoot(connection, active, rootId, options); + }, + }; + + const value = yield* scoped(function* () { + return yield* body(workspace); + }); + authorize(); + verifyWorkspace(connection.database, connection.dofs, connection.path); + return value; + }, + + setClock([database, now]: [WorkflowRunDatabase, () => number]): void { + connections.validateLease(database).connection.setClock(now); }, - // deno-lint-ignore require-yield - *capture(options = {}): Operation { - return captureWorkspaceRoot(connection, options); + issueToken([database, transaction]: [ + WorkflowRunDatabase, + WorkflowRunTransaction, + ]): WorkflowRunTransactionToken { + return connections.issueToken(database, transaction); }, - // deno-lint-ignore require-yield - *restore(rootId, options = {}): Operation { - return restoreWorkspaceRoot(connection, rootId, options); + validateToken([database, token]: [WorkflowRunDatabase, WorkflowRunTransactionToken]): void { + connections.validateToken(database, token); }, - }; - const value = yield* scoped(function* () { - return yield* body(workspace); - }); - verifyWorkspace(connection.database, connection.dofs, connection.path); - return value; + }, + { at: "min" }, + ); +} + +export function* transactWorkspaceRoots( + database: WorkflowRunDatabase, + body: (workspace: PrivateWorkspaceTransaction) => Operation, +): Operation> { + return yield* database.transact(function* (transaction) { + return yield* withPrivateWorkspaceTransaction(database, transaction, body); }); } -export function setPrivateWorkspaceClock(database: WorkflowRunDatabase, now: () => number): void { - workflowRunConnection(database).setClock(now); +export function withPrivateWorkspaceTransaction( + database: WorkflowRunDatabase, + transaction: WorkflowRunTransaction, + body: (workspace: PrivateWorkspaceTransaction) => Operation, +): Operation { + return PrivateWorkspace.operations.transact(database, transaction, body); +} + +export function setPrivateWorkspaceClock( + database: WorkflowRunDatabase, + now: () => number, +): Operation { + return PrivateWorkspace.operations.setClock(database, now); +} + +export function workflowRunTransactionToken( + database: WorkflowRunDatabase, + transaction: WorkflowRunTransaction, +): Operation { + return PrivateWorkspace.operations.issueToken(database, transaction); +} + +export function validateWorkflowRunTransactionToken( + database: WorkflowRunDatabase, + token: WorkflowRunTransactionToken, +): Operation { + return PrivateWorkspace.operations.validateToken(database, token); } diff --git a/packages/workflow/src/deno/workspace/restore.ts b/packages/workflow/src/deno/workspace/restore.ts index 04f00d3e..375994ef 100644 --- a/packages/workflow/src/deno/workspace/restore.ts +++ b/packages/workflow/src/deno/workspace/restore.ts @@ -1,6 +1,5 @@ import type { DatabaseSync } from "node:sqlite"; -import { WorkflowTransactionError } from "../../storage/errors.ts"; -import type { RunConnection } from "../connections.ts"; +import type { RunConnection, RunTransaction } from "../connections.ts"; import { reading } from "../reading.ts"; import { corrupt, @@ -26,21 +25,18 @@ export interface RestoreWorkspaceRootOptions { export function restoreWorkspaceRoot( connection: RunConnection, + transaction: RunTransaction, rootId: string, options: RestoreWorkspaceRootOptions = {}, ): StoredWorkspaceRoot { - if (!connection.transactionOpen) { - throw new WorkflowTransactionError( - "restoring a Workspace root requires the caller-owned workflow transaction to be open.", - ); - } + connection.validateTransaction(transaction); const { database, dofs, path, savepoints } = connection; verifyWorkspace(database, dofs, path); const selected = loadWorkspaceRoot(database, rootId, path); connection.invalidateDofsCaches(); try { - return savepoints.synchronous(() => { + return savepoints.synchronous(transaction, () => { rebuild(database, selected, path); connection.invalidateDofsCaches(); const restored = snapshotWorkspace(database, dofs, path, false); diff --git a/packages/workflow/src/deno/workspace/root.ts b/packages/workflow/src/deno/workspace/root.ts index 8a298bd9..12b3d826 100644 --- a/packages/workflow/src/deno/workspace/root.ts +++ b/packages/workflow/src/deno/workspace/root.ts @@ -2,8 +2,7 @@ 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 type { RunConnection, RunTransaction } from "../connections.ts"; import { reading } from "../reading.ts"; import { bytes, @@ -85,13 +84,10 @@ export function initializeEmptyWorkspace(database: DatabaseSync): void { export function captureWorkspaceRoot( connection: RunConnection, + transaction: RunTransaction, options: CaptureWorkspaceRootOptions = {}, ): StoredWorkspaceRoot { - if (!connection.transactionOpen) { - throw new WorkflowTransactionError( - "capturing a Workspace root requires the caller-owned workflow transaction to be open.", - ); - } + connection.validateTransaction(transaction); const root = snapshotWorkspace(connection.database, connection.dofs, connection.path, true); retainWorkspaceRoot(connection.database, root, connection.path); if (options.publish === true) { diff --git a/packages/workflow/tests/workflow-run-journal.test.ts b/packages/workflow/tests/workflow-run-journal.test.ts index a2b082bb..d01bfeee 100644 --- a/packages/workflow/tests/workflow-run-journal.test.ts +++ b/packages/workflow/tests/workflow-run-journal.test.ts @@ -679,9 +679,11 @@ describe("Tier WJ — a transaction a caller holds", () => { yield* transaction.journal.append(yielded("kept", "kept")); try { - yield* savepoint(() => { - throw new Error("the nested mutation failed"); - }); + yield* savepoint( + (function* () { + throw new Error("the nested mutation failed"); + })(), + ); } catch { // The savepoint rolled back; the transaction around it continues. } @@ -755,7 +757,11 @@ describe("Tier WJ — one connection, one operation at a time", () => { it("WJ15b: a savepoint outside any transaction is refused, not improvised", function* () { let raised: unknown; try { - yield* savepoint(() => "nothing to be inside"); + yield* savepoint( + (function* () { + return "nothing to be inside"; + })(), + ); } catch (error) { raised = error; } diff --git a/packages/workflow/tests/workflow-run-storage.test.ts b/packages/workflow/tests/workflow-run-storage.test.ts index f71a2e0e..ac68c8af 100644 --- a/packages/workflow/tests/workflow-run-storage.test.ts +++ b/packages/workflow/tests/workflow-run-storage.test.ts @@ -186,7 +186,7 @@ describe("Tier WS — authoritative connection and complete schema", () => { const connection = connections.at(join(root, "savepoint.sqlite")); connection.database.exec("BEGIN IMMEDIATE"); - connection.transactionOpen = true; + const transaction = connection.beginTransaction(); expect(() => connection.dofs.transactionSync(() => { connection.dofs.run("CREATE TABLE rolled_back (value TEXT)"); @@ -194,7 +194,7 @@ describe("Tier WS — authoritative connection and complete schema", () => { }), ).toThrow(Error); connection.database.exec("CREATE TABLE outer_survives (value TEXT)"); - connection.transactionOpen = false; + connection.finishTransaction(transaction); connection.database.exec("COMMIT"); const names = connection.database @@ -212,13 +212,13 @@ describe("Tier WS — authoritative connection and complete schema", () => { const connection = connections.at(join(root, "atomic.sqlite")); connection.database.exec("BEGIN IMMEDIATE"); - connection.transactionOpen = true; + const transaction = connection.beginTransaction(); expect(() => initializeSchema(connection.database, connection.dofs, () => { throw new Error("fail after the filesystem and root are initialized"); }), ).toThrow(Error); - connection.transactionOpen = false; + connection.finishTransaction(transaction); connection.database.exec("ROLLBACK"); expect(normalizedSchema(connection.database)).toEqual([]); diff --git a/packages/workflow/tests/workspace-root-restoration.test.ts b/packages/workflow/tests/workspace-root-restoration.test.ts index 734d2205..342175fe 100644 --- a/packages/workflow/tests/workspace-root-restoration.test.ts +++ b/packages/workflow/tests/workspace-root-restoration.test.ts @@ -9,7 +9,6 @@ import { type WorkflowRunDatabase, WorkflowRunStorage, } from "../mod.ts"; -import { workflowRunConnection } from "../src/deno/database.ts"; import { encodeWorkspaceManifest, parseWorkspaceManifest, @@ -54,12 +53,12 @@ function count(database: DatabaseSync, table: string): number { function* createCorruptionFixture(storage: string, runId: string): Operation { yield* withStorage(storage, function* () { const database = yield* createRun({ runId }); - setPrivateWorkspaceClock(database, () => 1_000); + yield* 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* setPrivateWorkspaceClock(database, () => 2_000); yield* capture(database, function* (workspace) { yield* workspace.filesystem.writeFile("/dir/file.txt", "second retained bytes", 0o600); }); @@ -75,7 +74,7 @@ describe("Tier WRR — private Workspace root restoration", () => { yield* withStorage(storage, function* () { const database = yield* createRun({ runId: "restore-history" }); - setPrivateWorkspaceClock(database, () => 10_000); + yield* 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); @@ -83,7 +82,7 @@ describe("Tier WRR — private Workspace root restoration", () => { yield* workspace.filesystem.symlink("file.txt", "/tree/current.txt"); }); - setPrivateWorkspaceClock(database, () => 20_000); + yield* 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"); @@ -195,33 +194,31 @@ describe("Tier WRR — private Workspace root restoration", () => { yield* withStorage(storage, function* () { const database = yield* createRun({ runId: "restore-rollback" }); - setPrivateWorkspaceClock(database, () => 100); + yield* setPrivateWorkspaceClock(database, () => 100); historical = (yield* capture(database, function* (workspace) { yield* workspace.filesystem.writeFile("/historical.txt", "historical"); })).rootId; - setPrivateWorkspaceClock(database, () => 200); + yield* 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); + const fault = new DatabaseSync(runPath(storage, "restore-rollback")); + fault.exec(` + CREATE TRIGGER fail_workspace_restore + BEFORE INSERT ON vfs_nodes + WHEN NEW.inode <> 1 + BEGIN + SELECT raise(ABORT, 'restoration insertion refused'); + END + `); 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); @@ -234,6 +231,8 @@ describe("Tier WRR — private Workspace root restoration", () => { } expect(historicalFile).toBeInstanceOf(Error); }); + fault.exec("DROP TRIGGER fail_workspace_restore"); + fault.close(); }); const sqlite = new DatabaseSync(runPath(storage, "restore-rollback")); @@ -486,9 +485,10 @@ describe("Tier WRR — private Workspace root restoration", () => { result: { status: "ok", value: "baseline" }, }); - const connection = workflowRunConnection(database); - baselineRoots = count(connection.database, "workspace_roots"); - baselineJournal = count(connection.database, "journal_events"); + const observer = new DatabaseSync(path); + baselineRoots = count(observer, "workspace_roots"); + baselineJournal = count(observer, "journal_events"); + observer.close(); const started = withResolvers(); const result = yield* transactWorkspaceRoots(database, function* (workspace) { yield* workspace.filesystem.writeFile("/captured.txt", "captured"); @@ -505,8 +505,10 @@ describe("Tier WRR — private Workspace root restoration", () => { 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 after = new DatabaseSync(path); + expect(count(after, "workspace_roots")).toBe(baselineRoots); + expect(count(after, "journal_events")).toBe(baselineJournal); + after.close(); const observed = yield* transact(database, function* (workspace) { const missing: string[] = []; diff --git a/packages/workflow/tests/workspace-root.test.ts b/packages/workflow/tests/workspace-root.test.ts index 883fea2b..bfde0b7c 100644 --- a/packages/workflow/tests/workspace-root.test.ts +++ b/packages/workflow/tests/workspace-root.test.ts @@ -6,7 +6,6 @@ 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, @@ -83,7 +82,7 @@ describe("Tier WRR — immutable retained Workspace roots", () => { yield* withStorage(storage, function* () { const database = yield* createRun({ runId: "canonical-a" }); - setPrivateWorkspaceClock(database, () => 1_700_000_000_000); + yield* 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 }); @@ -92,20 +91,26 @@ describe("Tier WRR — immutable retained Workspace roots", () => { "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 materialization = new DatabaseSync(runPath(storage, "canonical-a")); + try { + materialization + .prepare("UPDATE vfs_nodes SET manifest_hash = NULL WHERE type = 'file'") + .run(); + } finally { + materialization.close(); + } + 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); + yield* 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 }); @@ -193,7 +198,7 @@ describe("Tier WRR — immutable retained Workspace roots", () => { yield* withStorage(storage, function* () { const database = yield* createRun({ runId: "root-sequence" }); let time = 100; - setPrivateWorkspaceClock(database, () => time); + yield* setPrivateWorkspaceClock(database, () => time); roots.push( yield* capture(database, function* (workspace) { diff --git a/packages/workflow/tests/workspace-transaction.test.ts b/packages/workflow/tests/workspace-transaction.test.ts new file mode 100644 index 00000000..32deac5a --- /dev/null +++ b/packages/workflow/tests/workspace-transaction.test.ts @@ -0,0 +1,472 @@ +import { join } from "node:path"; +import { describe, it } from "@executablemd/test-support/bdd"; +import { expect } from "@executablemd/test-support/expect"; +import type { DurableEvent, Yield } from "@executablemd/durable-streams"; +import { ensure, type Operation, spawn, suspend, withResolvers } from "effection"; +import { + type WorkflowRunDatabase, + type WorkflowRunTransaction, + WorkflowRunStorage, + WorkflowTransactionError, +} from "../mod.ts"; +import { + createWorkflowRunConnections, + WorkflowRunTransactionToken, +} from "../src/deno/connections.ts"; +import { SavepointObservation, type SavepointObservationEvent } from "../src/deno/savepoints.ts"; +import { savepoint } from "../src/deno/transaction.ts"; +import { + type PrivateWorkspaceTransaction, + setPrivateWorkspaceClock, + validateWorkflowRunTransactionToken, + withPrivateWorkspaceTransaction, + workflowRunTransactionToken, +} from "../src/deno/workspace/private.ts"; +import { createRun, useStorageRoot, withStorage } from "./support/storage.ts"; + +function yielded(name: string): Yield { + return { + type: "yield", + coroutineId: "root", + description: { type: "call", name }, + result: { status: "ok", value: name }, + }; +} + +function names(events: DurableEvent[]): string[] { + return events.flatMap((event) => (event.type === "yield" ? [event.description.name] : [])); +} + +function* raised(operation: Operation): Operation { + try { + yield* operation; + return undefined; + } catch (error) { + return error; + } +} + +function fabricatedDatabase(value: unknown, seed: WorkflowRunDatabase): WorkflowRunDatabase { + const container: { database: WorkflowRunDatabase } = { + database: seed, + }; + Object.defineProperty(container, "database", { value, enumerable: true }); + return container.database; +} + +describe("Tier WTX — unified WorkflowRun savepoints", () => { + it("WTX1: operation savepoints release after child teardown and retain successful work", function* () { + const root = yield* useStorageRoot(); + const order: string[] = []; + yield* SavepointObservation.set((event) => { + order.push(`${event.kind}:${event.name}`); + }); + + const events = yield* withStorage(root, function* () { + const database = yield* createRun(); + order.length = 0; + const result = yield* database.transact(function* (transaction) { + yield* transaction.journal.append(yielded("before")); + yield* savepoint( + (function* () { + yield* transaction.journal.append(yielded("inside")); + const ready = withResolvers(); + yield* spawn(function* () { + yield* ensure(function* () { + yield* transaction.journal.append(yielded("cleanup")); + order.push("cleanup"); + }); + ready.resolve(); + yield* suspend(); + }); + yield* ready.operation; + })(), + ); + yield* transaction.journal.append(yielded("after")); + }); + if (!result.ok) { + throw result.error; + } + return yield* database.journal.readAll(); + }); + + expect(names(events)).toEqual(["before", "inside", "cleanup", "after"]); + const created = order.findIndex((entry) => entry.startsWith("create:")); + const cleaned = order.indexOf("cleanup"); + const released = order.findIndex((entry) => entry.startsWith("release:")); + expect(created).toBeGreaterThanOrEqual(0); + expect(cleaned).toBeGreaterThan(created); + expect(released).toBeGreaterThan(cleaned); + }); + + it("WTX2: failures and nested rollback discard only their own operation savepoint", function* () { + const root = yield* useStorageRoot(); + const observed: SavepointObservationEvent[] = []; + yield* SavepointObservation.set((event) => observed.push(event)); + + const events = yield* withStorage(root, function* () { + const database = yield* createRun(); + observed.length = 0; + const result = yield* database.transact(function* (transaction) { + yield* savepoint( + (function* () { + yield* transaction.journal.append(yielded("outer-savepoint")); + const innerFailure = yield* raised( + savepoint( + (function* () { + yield* transaction.journal.append(yielded("inner-rolled-back")); + throw new Error("inner failure"); + })(), + ), + ); + expect(innerFailure).toBeInstanceOf(Error); + yield* transaction.journal.append(yielded("outer-after-inner")); + })(), + ); + + const failed = yield* raised( + savepoint( + (function* () { + yield* transaction.journal.append(yielded("ordinary-rolled-back")); + throw new Error("ordinary failure"); + })(), + ), + ); + expect(failed).toBeInstanceOf(Error); + yield* transaction.journal.append(yielded("transaction-continues")); + }); + if (!result.ok) { + throw result.error; + } + return yield* database.journal.readAll(); + }); + + expect(names(events)).toEqual([ + "outer-savepoint", + "outer-after-inner", + "transaction-continues", + ]); + expect(observed.filter((event) => event.kind === "create")).toHaveLength(3); + expect(observed.filter((event) => event.kind === "rollback")).toHaveLength(2); + expect(observed.filter((event) => event.kind === "release")).toHaveLength(1); + expect(new Set(observed.map((event) => event.name)).size).toBe(3); + }); + + it("WTX3: a cleanup failure rolls the operation savepoint back", function* () { + const root = yield* useStorageRoot(); + + const events = yield* withStorage(root, function* () { + const database = yield* createRun(); + const result = yield* database.transact(function* (transaction) { + const failed = yield* raised( + savepoint( + (function* () { + yield* transaction.journal.append(yielded("cleanup-failure")); + const ready = withResolvers(); + yield* spawn(function* () { + yield* ensure(() => { + throw new Error("cleanup failed"); + }); + ready.resolve(); + yield* suspend(); + }); + yield* ready.operation; + })(), + ), + ); + expect(failed).toBeInstanceOf(Error); + yield* transaction.journal.append(yielded("survives")); + }); + if (!result.ok) { + throw result.error; + } + return yield* database.journal.readAll(); + }); + + expect(names(events)).toEqual(["survives"]); + }); + + it("WTX4: cancellation before, during, and during teardown strands no savepoint", function* () { + const root = yield* useStorageRoot(); + const observed: SavepointObservationEvent[] = []; + yield* SavepointObservation.set((event) => observed.push(event)); + + yield* withStorage(root, function* () { + const database = yield* createRun(); + observed.length = 0; + + const before = withResolvers(); + const beforeTask = yield* spawn(function* () { + yield* database.transact(function* () { + before.resolve(); + yield* suspend(); + yield* savepoint((function* () {})()); + }); + }); + yield* before.operation; + yield* beforeTask.halt(); + expect(observed).toEqual([]); + + const during = withResolvers(); + const duringTask = yield* spawn(function* () { + yield* database.transact(function* (transaction) { + yield* savepoint( + (function* () { + yield* transaction.journal.append(yielded("cancelled-during")); + during.resolve(); + yield* suspend(); + })(), + ); + }); + }); + yield* during.operation; + yield* duringTask.halt(); + expect(observed.at(-1)?.kind).toBe("rollback"); + + const childReady = withResolvers(); + const tearingDown = withResolvers(); + const releaseCleanup = withResolvers(); + const teardownTask = yield* spawn(function* () { + yield* database.transact(function* (transaction) { + yield* savepoint( + (function* () { + yield* transaction.journal.append(yielded("cancelled-teardown")); + yield* spawn(function* () { + yield* ensure(function* () { + tearingDown.resolve(); + yield* releaseCleanup.operation; + }); + childReady.resolve(); + yield* suspend(); + }); + yield* childReady.operation; + })(), + ); + }); + }); + yield* tearingDown.operation; + const halting = yield* spawn(function* () { + yield* teardownTask.halt(); + }); + releaseCleanup.resolve(); + yield* halting; + expect(observed.at(-1)?.kind).toBe("rollback"); + + yield* database.journal.append(yielded("after-cancellation")); + expect(names(yield* database.journal.readAll())).toEqual(["after-cancellation"]); + }); + }); + + it("WTX5: synchronous DOFS nesting shares the operation-savepoint allocator", function* () { + const root = yield* useStorageRoot(); + const observed: SavepointObservationEvent[] = []; + yield* SavepointObservation.set((event) => observed.push(event)); + + yield* withStorage(root, function* () { + const database = yield* createRun(); + observed.length = 0; + const result = yield* database.transact(function* (transaction) { + return yield* withPrivateWorkspaceTransaction(database, transaction, function* (workspace) { + const failure = yield* raised( + savepoint( + (function* () { + yield* workspace.filesystem.writeFile("/rolled-back.txt", "temporary"); + throw new Error("discard the DOFS mutation"); + })(), + ), + ); + expect(failure).toBeInstanceOf(Error); + }); + }); + if (!result.ok) { + throw result.error; + } + }); + + const creates = observed.filter((event) => event.kind === "create"); + expect(creates.length).toBeGreaterThanOrEqual(2); + expect(new Set(creates.map((event) => event.name)).size).toBe(creates.length); + const outer = creates[0]; + if (outer === undefined) { + throw new Error("the operation savepoint was not observed"); + } + expect(observed.at(-1)).toEqual({ kind: "rollback", name: outer.name }); + }); + + it("WTX6: savepoint SQL failures poison the outer transaction identity", function* () { + const root = yield* useStorageRoot(); + + const phases: Array<"release" | "rollback"> = ["release", "rollback"]; + for (const phase of phases) { + let activeName = ""; + const connections = createWorkflowRunConnections((event) => { + if (event.kind === "create") { + activeName = event.name; + } + }); + const connection = connections.at(join(root, `${phase}.sqlite`)); + connection.database.exec("BEGIN IMMEDIATE"); + const transaction = connection.beginTransaction(); + const failure = yield* raised( + connection.savepoints.operation( + transaction, + (function* () { + connection.database.exec(`RELEASE ${activeName}`); + if (phase === "rollback") { + throw new Error("force rollback after removing the savepoint"); + } + })(), + ), + ); + expect(failure).toBeInstanceOf(Error); + expect(() => connection.validateTransaction(transaction)).toThrow(WorkflowTransactionError); + connection.finishTransaction(transaction); + connection.database.exec("ROLLBACK"); + connections.close(); + } + + const connections = createWorkflowRunConnections(); + const connection = connections.at(join(root, "creation.sqlite")); + connection.database.exec("BEGIN IMMEDIATE"); + const transaction = connection.beginTransaction(); + connection.database.close(); + const creationFailure = yield* raised( + connection.savepoints.operation(transaction, (function* () {})()), + ); + expect(creationFailure).toBeInstanceOf(Error); + expect(() => connection.validateTransaction(transaction)).toThrow(WorkflowTransactionError); + connection.finishTransaction(transaction); + }); +}); + +describe("Tier WTX — WorkflowRun identity fences", () => { + it("WTX7: exact handles and tokens work only during their active transaction", function* () { + const root = yield* useStorageRoot(); + let escapedTransaction: WorkflowRunTransaction | undefined; + let escapedToken: WorkflowRunTransactionToken | undefined; + let escapedWorkspace: PrivateWorkspaceTransaction | undefined; + + yield* withStorage(root, function* () { + const database = yield* createRun(); + const first = yield* database.transact(function* (transaction) { + escapedTransaction = transaction; + escapedToken = yield* workflowRunTransactionToken(database, transaction); + yield* validateWorkflowRunTransactionToken(database, escapedToken); + yield* withPrivateWorkspaceTransaction(database, transaction, function* (workspace) { + escapedWorkspace = workspace; + }); + }); + expect(first.ok).toBe(true); + + const token = escapedToken; + if (token === undefined) { + throw new Error("the active transaction did not issue its token"); + } + expect(yield* raised(validateWorkflowRunTransactionToken(database, token))).toBeInstanceOf( + WorkflowTransactionError, + ); + const workspace = escapedWorkspace; + if (workspace === undefined) { + throw new Error("the private Workspace handle did not escape for the refusal proof"); + } + expect(yield* raised(workspace.currentRoot())).toBeInstanceOf(WorkflowTransactionError); + + const second = yield* database.transact(function* () { + expect(yield* raised(validateWorkflowRunTransactionToken(database, token))).toBeInstanceOf( + WorkflowTransactionError, + ); + }); + expect(second.ok).toBe(true); + + const transaction = escapedTransaction; + if (transaction === undefined) { + throw new Error("the transaction handle did not escape for the refusal proof"); + } + expect(yield* raised(workflowRunTransactionToken(database, transaction))).toBeInstanceOf( + WorkflowTransactionError, + ); + }); + }); + + it("WTX8: foreign, fabricated, and cross-run identities are refused", function* () { + const root = yield* useStorageRoot(); + + yield* withStorage(root, function* () { + const first = yield* createRun({ runId: "first" }); + const firstAgain = yield* createRun({ runId: "first" }); + const second = yield* createRun({ runId: "second" }); + const result = yield* first.transact(function* (firstTransaction) { + const firstToken = yield* workflowRunTransactionToken(first, firstTransaction); + expect( + yield* raised(validateWorkflowRunTransactionToken(firstAgain, firstToken)), + ).toBeInstanceOf(WorkflowTransactionError); + const nested = yield* second.transact(function* (secondTransaction) { + expect( + yield* raised(workflowRunTransactionToken(first, secondTransaction)), + ).toBeInstanceOf(WorkflowTransactionError); + expect( + yield* raised(validateWorkflowRunTransactionToken(second, firstToken)), + ).toBeInstanceOf(WorkflowTransactionError); + const fabricatedTransaction: WorkflowRunTransaction = { journal: first.journal }; + expect( + yield* raised(workflowRunTransactionToken(first, fabricatedTransaction)), + ).toBeInstanceOf(WorkflowTransactionError); + expect( + yield* raised( + validateWorkflowRunTransactionToken(first, new WorkflowRunTransactionToken()), + ), + ).toBeInstanceOf(WorkflowTransactionError); + }); + if (!nested.ok) { + throw nested.error; + } + }); + expect(result.ok).toBe(true); + + expect( + yield* raised(setPrivateWorkspaceClock(fabricatedDatabase({}, first), () => 0)), + ).toBeInstanceOf(WorkflowTransactionError); + }); + }); + + it("WTX9: leases and provider generations fence private authority", function* () { + const root = yield* useStorageRoot(); + let closed: WorkflowRunDatabase | undefined; + let priorToken: WorkflowRunTransactionToken | undefined; + + yield* withStorage(root, function* () { + const database = yield* createRun({ runId: "generation" }); + closed = database; + const result = yield* database.transact(function* (transaction) { + priorToken = yield* workflowRunTransactionToken(database, transaction); + }); + expect(result.ok).toBe(true); + expect(Reflect.ownKeys(database).filter((key) => typeof key === "symbol")).toEqual([]); + }); + + const closedHandle = closed; + if (closedHandle === undefined || priorToken === undefined) { + throw new Error("the prior provider did not leave identity evidence"); + } + const oldToken = priorToken; + yield* withStorage(root, function* () { + const found = yield* WorkflowRunStorage.operations.lookup("generation"); + if (!found.ok) { + throw found.error; + } + const database = found.value; + const result = yield* database.transact(function* () { + expect( + yield* raised(validateWorkflowRunTransactionToken(database, oldToken)), + ).toBeInstanceOf(WorkflowTransactionError); + expect( + yield* raised(validateWorkflowRunTransactionToken(closedHandle, oldToken)), + ).toBeInstanceOf(WorkflowTransactionError); + }); + expect(result.ok).toBe(true); + expect(yield* raised(setPrivateWorkspaceClock(closedHandle, () => 0))).toBeInstanceOf( + WorkflowTransactionError, + ); + }); + }); +}); diff --git a/scripts/runtime-test-exclusions.ts b/scripts/runtime-test-exclusions.ts index b906b322..6f2d2333 100644 --- a/scripts/runtime-test-exclusions.ts +++ b/scripts/runtime-test-exclusions.ts @@ -108,6 +108,12 @@ const DENO_ONLY_TOOLING: RuntimeExclusion[] = [ "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", }, + { + path: "packages/workflow/tests/workspace-transaction.test.ts", + reason: + "exercises Deno-private node:sqlite transaction identities and real SQLite savepoint failure behavior; node:sqlite remains behind --experimental-sqlite on Node 22", + issue: "https://github.com/taras/executable.md/issues/365", + }, ]; /** diff --git a/specs/workflow-spec.md b/specs/workflow-spec.md index e3449b95..2bbddb92 100644 --- a/specs/workflow-spec.md +++ b/specs/workflow-spec.md @@ -181,7 +181,9 @@ behavior appear. Shared modules import none of them and detect no runtime. 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. +connection or invalidate another lease. The provider keeps the association in +exact-object adapter-private state; the public handle contains no discoverable +SQLite or DOFS connection. ### 9.1 What identifies a run @@ -329,10 +331,18 @@ is cancelled leaves no row at all. 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 +connection queue and one unified savepoint allocator. It remains alive until provider-scope teardown, after the provider's child scopes finish. Different database paths have independent entries. +The entry receives an opaque generation identity when it is created. Reopening +the same canonical path after provider teardown produces a different identity. +Every top-level transaction receives a separate opaque identity and an exact +active record containing its path, connection generation, open state, +authorized lease and transaction handle. The provider invalidates that record +before commit or rollback, so a retained handle or token never becomes valid +again in a later transaction. + 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 @@ -379,6 +389,22 @@ 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. +The same monotonically unique allocator also owns operation-spanning +savepoints. An operation savepoint validates the exact active transaction +identity before SQL, runs its operation in a child scope, and waits for all +children and resources to tear down before release. An ordinary failure rolls +back to and releases only that savepoint, leaving the outer transaction free to +continue. Cancellation or halt invokes synchronous cleanup so no savepoint is +stranded. A savepoint creation, rollback, or release failure makes the outer +transaction uncommittable. + +The active-path context chain remains structural refusal data: it detects that +an enclosing scope already holds a path, but never authorizes work. Adapter- +private contextual operations validate provider-owned exact identities instead. +Missing, foreign, fabricated, completed, closed and stale handles or tokens are +refused before SQLite is touched. A transaction on a different run neither +hides nor replaces the outer run's active record. + 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 @@ -470,7 +496,8 @@ also left unchanged. Public `xmd workflow` lifecycle commands; lifecycle transition policy, executor leases and stale-owner recovery; public Workspace mutation and filesystem -effects; provider-level atomic Workspace effect/journal publication; public +effects; provider-neutral live effect coordination and filtered journal +routing; 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 454bad7bca1daf33cdb887475e14c719de113fa6 Mon Sep 17 00:00:00 2001 From: Taras Mankovski <74687+taras@users.noreply.github.com> Date: Sat, 8 Aug 2026 20:13:22 -0400 Subject: [PATCH 2/2] =?UTF-8?q?=F0=9F=A7=B9=20Restore=20DOFS=20caches=20af?= =?UTF-8?q?ter=20savepoint=20rollback?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- architecture.md | 9 ++- packages/workflow/src/deno/connections.ts | 7 ++ packages/workflow/src/deno/savepoints.ts | 7 ++ .../tests/workspace-transaction.test.ts | 78 ++++++++++++++++++- specs/executable-mdx-spec.md | 17 ++++ specs/workflow-spec.md | 11 ++- 6 files changed, 123 insertions(+), 6 deletions(-) diff --git a/architecture.md b/architecture.md index b1bd4936..85952781 100644 --- a/architecture.md +++ b/architecture.md @@ -445,8 +445,10 @@ 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. Those savepoints and operation-spanning savepoints share one allocator and cannot -collide or release one another. A -second long-lived DOFS connection is not a coherent reader because provider +collide or release one another. After a savepoint successfully rolls back and +releases, the shared rollback path invalidates both caches on the authoritative +DOFS wrapper before outer transaction work resumes. 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 @@ -458,7 +460,8 @@ 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 +Separately, every unsuccessful caller-owned transaction attempts top-level +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 diff --git a/packages/workflow/src/deno/connections.ts b/packages/workflow/src/deno/connections.ts index 3a07a88c..33686f4a 100644 --- a/packages/workflow/src/deno/connections.ts +++ b/packages/workflow/src/deno/connections.ts @@ -149,6 +149,13 @@ function createConnection(path: string, observeSavepoint: SavepointObserver): Ru active.failure = failure; } }, + afterRollback(transaction): void { + validate(transaction); + if (installed === undefined) { + throw new WorkflowConnectionStateError("the workflow connection is not installed"); + } + installed.invalidateDofsCaches(); + }, }, observeSavepoint, ); diff --git a/packages/workflow/src/deno/savepoints.ts b/packages/workflow/src/deno/savepoints.ts index 4cda5f5d..0d5102fa 100644 --- a/packages/workflow/src/deno/savepoints.ts +++ b/packages/workflow/src/deno/savepoints.ts @@ -9,6 +9,7 @@ export interface SavepointTransaction { export interface SavepointTransactionController { validate(transaction: SavepointTransaction): void; poison(transaction: SavepointTransaction, failure: unknown): void; + afterRollback(transaction: SavepointTransaction): void; } export interface SavepointManager { @@ -101,6 +102,12 @@ export function createSavepointManager( transactions.poison(transaction, failure); throw failure; } + try { + transactions.afterRollback(transaction); + } catch (error) { + transactions.poison(transaction, error); + throw error; + } report("rollback", savepoint.name); } diff --git a/packages/workflow/tests/workspace-transaction.test.ts b/packages/workflow/tests/workspace-transaction.test.ts index 32deac5a..bc5d4f4d 100644 --- a/packages/workflow/tests/workspace-transaction.test.ts +++ b/packages/workflow/tests/workspace-transaction.test.ts @@ -1,4 +1,5 @@ import { join } from "node:path"; +import { DatabaseSync } from "node:sqlite"; import { describe, it } from "@executablemd/test-support/bdd"; import { expect } from "@executablemd/test-support/expect"; import type { DurableEvent, Yield } from "@executablemd/durable-streams"; @@ -22,7 +23,7 @@ import { withPrivateWorkspaceTransaction, workflowRunTransactionToken, } from "../src/deno/workspace/private.ts"; -import { createRun, useStorageRoot, withStorage } from "./support/storage.ts"; +import { createRun, runPath, useStorageRoot, withStorage } from "./support/storage.ts"; function yielded(name: string): Yield { return { @@ -337,6 +338,81 @@ describe("Tier WTX — unified WorkflowRun savepoints", () => { expect(() => connection.validateTransaction(transaction)).toThrow(WorkflowTransactionError); connection.finishTransaction(transaction); }); + + it("WTX10: savepoint rollback restores cache coherence before outer commit", function* () { + const root = yield* useStorageRoot(); + const runId = "savepoint-cache"; + let retainedRoot = ""; + + yield* withStorage(root, function* () { + const database = yield* createRun({ runId }); + const initialized = yield* database.transact(function* (transaction) { + return yield* withPrivateWorkspaceTransaction(database, transaction, function* (workspace) { + yield* workspace.filesystem.writeFile("/kept.txt", "known retained bytes"); + const root = yield* workspace.capture({ publish: true }); + expect(yield* workspace.filesystem.readTextFile("/kept.txt")).toBe( + "known retained bytes", + ); + return root.rootId; + }); + }); + if (!initialized.ok) { + throw initialized.error; + } + retainedRoot = initialized.value; + + const committed = yield* database.transact(function* (transaction) { + yield* withPrivateWorkspaceTransaction(database, transaction, function* (workspace) { + expect(yield* workspace.currentRoot()).toBe(retainedRoot); + const failure = yield* raised( + savepoint( + (function* () { + yield* workspace.filesystem.remove("/kept.txt"); + expect( + yield* raised(workspace.filesystem.readTextFile("/kept.txt")), + ).toBeInstanceOf(Error); + throw new Error("known operation failure"); + })(), + ), + ); + expect(failure).toBeInstanceOf(Error); + expect(yield* workspace.filesystem.readTextFile("/kept.txt")).toBe( + "known retained bytes", + ); + expect(yield* workspace.currentRoot()).toBe(retainedRoot); + }); + yield* transaction.journal.append(yielded("savepoint-cache-rollback")); + }); + if (!committed.ok) { + throw committed.error; + } + + const next = yield* database.transact(function* (transaction) { + return yield* withPrivateWorkspaceTransaction(database, transaction, function* (workspace) { + return { + content: yield* workspace.filesystem.readTextFile("/kept.txt"), + root: yield* workspace.currentRoot(), + }; + }); + }); + if (!next.ok) { + throw next.error; + } + expect(next.value).toEqual({ content: "known retained bytes", root: retainedRoot }); + expect(names(yield* database.journal.readAll())).toEqual(["savepoint-cache-rollback"]); + }); + + const database = new DatabaseSync(runPath(root, runId)); + try { + const rows = database + .prepare("SELECT workspace_root_id FROM journal_events ORDER BY sequence") + .all(); + expect(rows).toHaveLength(1); + expect(rows[0]?.["workspace_root_id"]).toBe(retainedRoot); + } finally { + database.close(); + } + }); }); describe("Tier WTX — WorkflowRun identity fences", () => { diff --git a/specs/executable-mdx-spec.md b/specs/executable-mdx-spec.md index 2ab311b4..9023343c 100644 --- a/specs/executable-mdx-spec.md +++ b/specs/executable-mdx-spec.md @@ -6615,6 +6615,23 @@ 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 WTX — WorkflowRun savepoints and transaction authority + +Defined in [Workflow runs](./workflow-spec.md) §9.6. + +| # | Test | Verify | +|---|------|--------| +| WTX1 | Successful operation savepoint | Release follows child teardown and retains the savepoint's work inside the outer transaction | +| WTX2 | Nested and ordinary failure | Each failed savepoint rolls back only its own work, and the outer transaction may continue | +| WTX3 | Cleanup failure | A failure during child teardown rolls back the operation savepoint | +| WTX4 | Cancellation | Cancellation before entry, during mutation and during teardown strands no savepoint | +| WTX5 | Shared allocator | Synchronous DOFS and operation savepoints draw collision-free names from one connection-owned allocator | +| WTX6 | Savepoint SQL failure | Creation, rollback or release failure poisons the outer transaction so it cannot commit | +| WTX7 | Exact active authority | Handles and tokens authorize work only during their exact active transaction | +| WTX8 | Foreign authority | Foreign, fabricated and cross-run identities are refused before SQL | +| WTX9 | Lease and generation fences | Closed leases and stale connection generations cannot recover private authority | +| WTX10 | Savepoint rollback cache coherence | A failed mutation restores the file and both authoritative caches before the caller continues and commits the outer transaction | + ### Tier WRR — Immutable retained Workspace roots Defined in [Workflow runs](./workflow-spec.md) §9.4 and §9.6–§9.7. diff --git a/specs/workflow-spec.md b/specs/workflow-spec.md index 2bbddb92..969d15e5 100644 --- a/specs/workflow-spec.md +++ b/specs/workflow-spec.md @@ -382,8 +382,8 @@ 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 +If a caller-owned transaction does not commit, its finalizer attempts top-level +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 @@ -398,6 +398,13 @@ continue. Cancellation or halt invokes synchronous cleanup so no savepoint is stranded. A savepoint creation, rollback, or release failure makes the outer transaction uncommittable. +After `ROLLBACK TO` and `RELEASE` both succeed, the shared savepoint rollback +path invalidates the authoritative resolution and blob caches before the caller +may resume outer transaction work. This applies equally to operation savepoints +and synchronous DOFS savepoints. If rollback, release or cache invalidation +fails, the active transaction is poisoned and cannot commit; its top-level +rollback finalizer performs the separate outer-transaction invalidation above. + The active-path context chain remains structural refusal data: it detects that an enclosing scope already holds a path, but never authorizes work. Adapter- private contextual operations validate provider-owned exact identities instead.