diff --git a/architecture.md b/architecture.md index 9bd37de4..f8d18266 100644 --- a/architecture.md +++ b/architecture.md @@ -380,6 +380,26 @@ historical execution that reached it. Replay never asks current state to prove a past effect: a file written and later deleted is absent at the frontier, while both completed effects still restore in order. +Structured durable operations accept an explicit provider-neutral live +coordinator. The default coordinator executes once, converts execution success +or failure to the existing durable protocol `Result`, calls the existing Yield +publication continuation once, and returns that same result only after +publication completes. The continuation publishes through the durable stream's +ordered append fence. A backing append failure activates fail-stop state and +raises `DurablePersistenceError` with the adapter error as its cause; it writes +no compensating `Close`, and catching it cannot admit later durable work. A +marked pre-persistence policy rejection remains an ordinary policy failure. +There is no generic durable validation hook: validation owned by a caller or +provider occurs before it constructs the durable effect. Replay, including the +replayed prefix of a partial run, bypasses the coordinator, execution and live +publication. Callback-based durable effects keep their existing path. + +The shared Workspace durable-operation wrapper selects a contextual Workspace +coordinator explicitly. Its default fails before execution or publication, so +installing no provider cannot leak a mutation outside its transaction. Selecting +Workspace coordination for one operation does not enlist unrelated durable +operations in the same scope. + Every Workspace-local expansion publishes one effect through one effect transaction: @@ -472,8 +492,22 @@ 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. Provider-neutral durable coordination, filtered -journal routing, and atomic Workspace effect publication are also absent. +operation at this layer. Provider-neutral durable coordination and explicit +filtered-journal routing are present, while the Deno Workspace coordinator that +combines mutation, root publication and journal publication atomically is +absent. + +The Deno journal adapter routes an append ordinarily when no destination is +bound. A publication may instead bind one exact transaction destination for its +own lexical scope after the existing secret gate. The route validates the +database lease, connection generation, transaction identity, token and open +state before delegating to the existing `transaction.journal`; it contains no +insertion SQL of its own. A nested route for another run delegates past itself, +and `readAll()` always follows ordinary replay without routing or secret +filtering. The provider-owned ordinary destination and each publication-local +routed destination are terminal `{ at: "min" }` handlers, so an enclosing +loaded copy with the same stable contextual name cannot suppress an append or +run ahead of exact-token validation. The initial topology requires neither writable FUSE nor native subprocess access and does not bundle `workerd`. A Cloudflare-hosted or workerd-backed @@ -793,6 +827,9 @@ Status is measured against 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, retains its identity, state, document executions and filtered journal, and validates immutable Workspace roots through one provider-owned connection entry | built on the #365 stack; Workspace effect publication is unbuilt | | caller-owned storage transaction | publishes several changes, including journal events, in one transaction nothing else enlists in | built on main | +| live durable-operation coordinator | explicitly coordinates structured live execution with existing Yield publication while leaving replay and callback effects unchanged | built on the #365 stack | +| Workspace coordination API | fails closed by default and lets a Workspace operation explicitly select provider coordination | built on the #365 stack; the atomic Deno Workspace handler is unbuilt | +| explicit WorkflowRun journal route | binds one already-filtered publication to one exact active transaction and otherwise uses ordinary serialized journal storage | built on the #365 stack | | `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 | | `ephemeral eval` | reconstructs live middleware and bindings without a journal entry | built on main | diff --git a/packages/durable-streams/README.md b/packages/durable-streams/README.md index 61d7dc33..0c028ffa 100644 --- a/packages/durable-streams/README.md +++ b/packages/durable-streams/README.md @@ -286,6 +286,24 @@ gate's ordinary failure and may produce a separately admitted `Close(err)`. Violating this invariant (advancing the generator before the write) creates an unrecoverable gap: the journal would be missing an entry, and replay would feed the wrong result to a subsequent effect. +### Live operation coordinators + +`createDurableOperation` accepts an optional +`LiveDurableOperationCoordinator`. The default coordinator executes the live +operation once, converts its success or failure to the existing protocol +`Result`, invokes the Yield publication continuation once, and returns that +same result after publication completes. Replay bypasses the coordinator, +executor, continuation, and live append; a partially replayed run coordinates +only its live suffix. + +The publication continuation uses the ordered append fence described above. A +backing append failure therefore activates the same fail-stop state and raises +`DurablePersistenceError` with the adapter error as its cause. A marked +pre-persistence policy rejection remains an ordinary policy failure. There is no +generic validation option on durable operations: a caller or provider validates +before constructing the durable effect. Callback-based durable effects retain +their existing execution path. + --- ## Divergence detection diff --git a/packages/durable-streams/effect.ts b/packages/durable-streams/effect.ts index 5cffa6f6..000d3529 100644 --- a/packages/durable-streams/effect.ts +++ b/packages/durable-streams/effect.ts @@ -29,6 +29,10 @@ import { rememberDurabilityFailure, } from "./durability.ts"; import { StaleInputError } from "./errors.ts"; +import { + defaultLiveDurableOperationCoordinator, + type LiveDurableOperationCoordinator, +} from "./live-coordinator.ts"; import { ReplayGuard } from "./replay-guard.ts"; import { protocolToEffection, serializeError } from "./serialize.ts"; import type { @@ -306,10 +310,15 @@ export function createDurableEffect( * * @param desc Structured description for the journal and divergence detection * @param execute Returns an Operation to run during live execution + * @param options.coordinator Selects the live execution/publication boundary; + * replay never invokes it */ export function createDurableOperation( desc: EffectDescription, execute: () => Operation, + options: { + coordinator?: LiveDurableOperationCoordinator; + } = {}, ): DurableEffect { return { description: `${desc.type}(${desc.name})`, @@ -340,24 +349,17 @@ export function createDurableOperation( return; } - let result: Result; try { - const value = yield* execute(); - result = { status: "ok", value: value as Json }; - } catch (e) { - const error = e instanceof Error ? e : new Error(String(e)); - result = { status: "err", error: serializeError(error) }; - } - - const event: Yield = { - type: "yield", - coroutineId: ctx.coroutineId, - description: desc, - result, - }; - - try { - yield* appendDurableEvent(ctx, event); + const coordinator = options.coordinator ?? defaultLiveDurableOperationCoordinator; + const result = yield* coordinator.run(execute, function* (published) { + const event: Yield = { + type: "yield", + coroutineId: ctx.coroutineId, + description: desc, + result: published, + }; + yield* appendDurableEvent(ctx, event); + }); resolve(protocolToEffection(result)); } catch (err) { resolve({ diff --git a/packages/durable-streams/live-coordinator.ts b/packages/durable-streams/live-coordinator.ts new file mode 100644 index 00000000..3a84e113 --- /dev/null +++ b/packages/durable-streams/live-coordinator.ts @@ -0,0 +1,30 @@ +import type { Operation } from "effection"; +import { serializeError } from "./serialize.ts"; +import type { Json, Result } from "./types.ts"; + +/** Coordinates one live structured durable operation with its publication. */ +export interface LiveDurableOperationCoordinator { + run( + execute: () => Operation, + publish: (result: Result) => Operation, + ): Operation; +} + +/** The ordinary live path: execute once, publish once, then return the same result. */ +export const defaultLiveDurableOperationCoordinator: LiveDurableOperationCoordinator = { + *run( + execute: () => Operation, + publish: (result: Result) => Operation, + ): Operation { + let result: Result; + try { + result = { status: "ok", value: yield* execute() }; + } catch (error) { + const failure = error instanceof Error ? error : new Error(String(error)); + result = { status: "err", error: serializeError(failure) }; + } + + yield* publish(result); + return result; + }, +}; diff --git a/packages/durable-streams/mod.ts b/packages/durable-streams/mod.ts index 5cbf605e..111631f8 100644 --- a/packages/durable-streams/mod.ts +++ b/packages/durable-streams/mod.ts @@ -78,6 +78,10 @@ export { parseDurableEvent } from "./parse.ts"; export { createDurableEffect, createDurableOperation } from "./effect.ts"; export type { Executor } from "./effect.ts"; +// Structured live-operation coordination +export { defaultLiveDurableOperationCoordinator } from "./live-coordinator.ts"; +export type { LiveDurableOperationCoordinator } from "./live-coordinator.ts"; + // Workflow-enabled effects export { durableAction, durableCall, durableSleep, versionCheck } from "./operations.ts"; diff --git a/packages/durable-streams/tests/live-coordinator.test.ts b/packages/durable-streams/tests/live-coordinator.test.ts new file mode 100644 index 00000000..c19dd43b --- /dev/null +++ b/packages/durable-streams/tests/live-coordinator.test.ts @@ -0,0 +1,385 @@ +import { describe, it } from "@executablemd/test-support/bdd"; +import { expect } from "@executablemd/test-support/expect"; +import { type Operation, spawn, suspend, withResolvers } from "effection"; +import { + createDurableOperation, + durableAction, + durableCall, + durableRun, + DurablePersistenceError, + InMemoryStream, + type DurableEvent, + type DurableStream, + type Json, + type LiveDurableOperationCoordinator, + type Result, + type Workflow, +} from "../mod.ts"; + +function yields(events: DurableEvent[]): DurableEvent[] { + return events.filter((event) => event.type === "yield"); +} + +function* raised(operation: Operation): Operation { + try { + yield* operation; + return undefined; + } catch (error) { + return error; + } +} + +function* coordinatedStep( + name: string, + execute: () => Operation, + coordinator?: LiveDurableOperationCoordinator, +): Workflow { + yield createDurableOperation({ type: "coordinated", name }, execute, { coordinator }); +} + +describe("Tier DLC — live durable-operation coordination", () => { + it("DLC1: default live success executes and publishes once before resumption", function* () { + const timeline: string[] = []; + const stream = new InMemoryStream(); + stream.onAppend = (event) => { + if (event.type === "yield") { + timeline.push("publish"); + } + }; + + function* workflow(): Workflow { + yield* coordinatedStep("success", function* () { + timeline.push("execute"); + return "value"; + }); + timeline.push("resume"); + return "done"; + } + + expect(yield* durableRun(workflow, { stream })).toBe("done"); + expect(timeline).toEqual(["execute", "publish", "resume"]); + expect(yields(stream.snapshot())).toHaveLength(1); + }); + + it("DLC2: execution failure publishes the existing failed Result once", function* () { + const stream = new InMemoryStream(); + let executions = 0; + + function* workflow(): Workflow { + yield* coordinatedStep("failure", function* () { + executions += 1; + throw new TypeError("operation failed"); + }); + } + + const failure = yield* raised(durableRun(workflow, { stream })); + expect(failure).toBeInstanceOf(Error); + if (!(failure instanceof Error)) { + throw new Error("the durable failure was not restored as an Error"); + } + expect(failure.name).toBe("TypeError"); + expect(executions).toBe(1); + const recorded = yields(stream.snapshot()); + expect(recorded).toHaveLength(1); + expect(recorded[0]).toEqual({ + type: "yield", + coroutineId: "root", + description: { type: "coordinated", name: "failure" }, + result: { + status: "err", + error: expect.objectContaining({ name: "TypeError", message: "operation failed" }), + }, + }); + }); + + it("DLC3: publication failure reaches the caller and is never republished", function* () { + const publicationFailure = new Error("publication failed"); + const accepted: DurableEvent[] = []; + let appendAttempts = 0; + let resumed = false; + const stream: DurableStream = { + // deno-lint-ignore require-yield + *readAll(): Operation { + return [...accepted]; + }, + // deno-lint-ignore require-yield + *append(event): Operation { + appendAttempts += 1; + throw publicationFailure; + }, + }; + + function* workflow(): Workflow { + yield* coordinatedStep("publish-failure", function* () { + return "executed"; + }); + resumed = true; + } + + const failure = yield* raised(durableRun(workflow, { stream })); + expect(failure).toBeInstanceOf(DurablePersistenceError); + if (!(failure instanceof DurablePersistenceError)) { + throw new Error("expected coordinated publication to fail durably"); + } + expect(failure.cause).toBe(publicationFailure); + expect(appendAttempts).toBe(1); + expect(accepted).toEqual([]); + expect(resumed).toBe(false); + }); + + it("DLC4: active fail-stop state prevents later coordination and execution", function* () { + const publicationFailure = new Error("first publication failed"); + let appendAttempts = 0; + const stream: DurableStream = { + // deno-lint-ignore require-yield + *readAll(): Operation { + return []; + }, + // deno-lint-ignore require-yield + *append(): Operation { + appendAttempts += 1; + throw publicationFailure; + }, + }; + let firstExecutions = 0; + let laterExecutions = 0; + let coordinations = 0; + let caught: unknown; + const coordinator: LiveDurableOperationCoordinator = { + *run( + execute: () => Operation, + publish: (result: Result) => Operation, + ): Operation { + coordinations += 1; + const value = yield* execute(); + const result: Result = { status: "ok", value }; + yield* publish(result); + return result; + }, + }; + + function* workflow(): Workflow { + try { + yield* coordinatedStep("poison", function* () { + firstExecutions += 1; + return "first"; + }); + } catch (error) { + caught = error; + } + yield* coordinatedStep( + "blocked", + function* () { + laterExecutions += 1; + return "not reached"; + }, + coordinator, + ); + } + + const failure = yield* raised(durableRun(workflow, { stream })); + expect(failure).toBe(caught); + expect(failure).toBeInstanceOf(DurablePersistenceError); + if (!(failure instanceof DurablePersistenceError)) { + throw new Error("expected the first durability failure to remain active"); + } + expect(failure.cause).toBe(publicationFailure); + expect(firstExecutions).toBe(1); + expect(laterExecutions).toBe(0); + expect(coordinations).toBe(0); + expect(appendAttempts).toBe(1); + }); + + it("DLC5: complete replay bypasses coordinator, execution, publication and append", function* () { + const stream = new InMemoryStream([ + { + type: "yield", + coroutineId: "root", + description: { type: "coordinated", name: "replayed" }, + result: { status: "ok", value: "stored" }, + }, + { type: "close", coroutineId: "root", result: { status: "ok", value: "done" } }, + ]); + let executions = 0; + let coordinations = 0; + const coordinator: LiveDurableOperationCoordinator = { + *run( + execute: () => Operation, + publish: (result: Result) => Operation, + ): Operation { + coordinations += 1; + const result: Result = { status: "ok", value: yield* execute() }; + yield* publish(result); + return result; + }, + }; + + function* workflow(): Workflow { + yield* coordinatedStep( + "replayed", + function* () { + executions += 1; + return "live"; + }, + coordinator, + ); + return "done"; + } + + expect(yield* durableRun(workflow, { stream })).toBe("done"); + expect({ executions, coordinations, appends: stream.appendCount }).toEqual({ + executions: 0, + coordinations: 0, + appends: 0, + }); + }); + + it("DLC6: partial replay coordinates only the live suffix", function* () { + const stream = new InMemoryStream([ + { + type: "yield", + coroutineId: "root", + description: { type: "coordinated", name: "first" }, + result: { status: "ok", value: "stored" }, + }, + ]); + const executed: string[] = []; + let coordinations = 0; + const coordinator: LiveDurableOperationCoordinator = { + *run( + execute: () => Operation, + publish: (result: Result) => Operation, + ): Operation { + coordinations += 1; + const result: Result = { status: "ok", value: yield* execute() }; + yield* publish(result); + return result; + }, + }; + + function* workflow(): Workflow { + yield* coordinatedStep( + "first", + function* () { + executed.push("first"); + return "live-first"; + }, + coordinator, + ); + yield* coordinatedStep( + "second", + function* () { + executed.push("second"); + return "live-second"; + }, + coordinator, + ); + return "done"; + } + + expect(yield* durableRun(workflow, { stream })).toBe("done"); + expect(executed).toEqual(["second"]); + expect(coordinations).toBe(1); + expect(yields(stream.snapshot())).toHaveLength(2); + }); + + it("DLC7: cancellation during execution or publication produces no late Yield", function* () { + const executionStream = new InMemoryStream(); + const executionStarted = withResolvers(); + function* executionWorkflow(): Workflow { + yield* coordinatedStep("cancel-execute", function* () { + executionStarted.resolve(); + yield* suspend(); + return null; + }); + } + const executionTask = yield* spawn(() => + durableRun(executionWorkflow, { + stream: executionStream, + }), + ); + yield* executionStarted.operation; + yield* executionTask.halt(); + expect(yields(executionStream.snapshot())).toEqual([]); + + const publicationStarted = withResolvers(); + let publicationAttempts = 0; + const publicationStream: DurableStream = { + // deno-lint-ignore require-yield + *readAll(): Operation { + return []; + }, + *append(event): Operation { + if (event.type === "yield") { + publicationAttempts += 1; + publicationStarted.resolve(); + yield* suspend(); + } + }, + }; + function* publicationWorkflow(): Workflow { + yield* coordinatedStep("cancel-publish", function* () { + return "ready"; + }); + } + const publicationTask = yield* spawn(() => + durableRun(publicationWorkflow, { stream: publicationStream }), + ); + yield* publicationStarted.operation; + yield* publicationTask.halt(); + expect(publicationAttempts).toBe(1); + expect(yield* publicationStream.readAll()).toEqual([]); + }); + + it("DLC8: an explicit coordinator affects only its selected operation", function* () { + const stream = new InMemoryStream(); + let coordinated = 0; + let ordinary = 0; + const coordinator: LiveDurableOperationCoordinator = { + *run( + execute: () => Operation, + publish: (result: Result) => Operation, + ): Operation { + coordinated += 1; + const result: Result = { status: "ok", value: yield* execute() }; + yield* publish(result); + return result; + }, + }; + + function* workflow(): Workflow { + yield* coordinatedStep( + "selected", + function* () { + return "selected"; + }, + coordinator, + ); + yield* durableCall("ordinary", function* () { + ordinary += 1; + return "ordinary"; + }); + return "done"; + } + + expect(yield* durableRun(workflow, { stream })).toBe("done"); + expect({ coordinated, ordinary }).toEqual({ coordinated: 1, ordinary: 1 }); + }); + + it("DLC9: callback-based durable effects retain their existing path", function* () { + const stream = new InMemoryStream(); + let executions = 0; + function* workflow(): Workflow { + yield* durableAction("callback", (resolve) => { + executions += 1; + resolve("callback-value"); + return () => {}; + }); + return "done"; + } + + expect(yield* durableRun(workflow, { stream })).toBe("done"); + expect(executions).toBe(1); + expect(yields(stream.snapshot())).toHaveLength(1); + }); +}); diff --git a/packages/workflow/mod.ts b/packages/workflow/mod.ts index a5f7b872..f505148a 100644 --- a/packages/workflow/mod.ts +++ b/packages/workflow/mod.ts @@ -30,6 +30,10 @@ export { getWorkflowRun, useWorkflow } from "./src/run.ts"; export type { WorkflowRun } from "./src/run.ts"; export { useWorkflowServiceDenial, WorkflowServiceDeniedError } from "./src/service-denial.ts"; +export { WorkspaceCoordination, WorkspaceCoordinationProviderError } from "./src/workspace/api.ts"; +export type { WorkspaceCoordinationApi } from "./src/workspace/api.ts"; +export { createDurableWorkspaceOperation } from "./src/workspace/effect.ts"; + export { WorkflowRunStorage, WorkflowStorageProviderError } from "./src/storage/api.ts"; export type { CreateWorkflowRunRequest, diff --git a/packages/workflow/src/deno/database.ts b/packages/workflow/src/deno/database.ts index 813b4fa1..94cc3147 100644 --- a/packages/workflow/src/deno/database.ts +++ b/packages/workflow/src/deno/database.ts @@ -54,6 +54,7 @@ import { type WorkflowRunRecord, } from "../storage/record.ts"; import { insertJournalEvent, readJournalEntries } from "./journal.ts"; +import { routeWorkflowRunJournal } from "./journal-route.ts"; import type { RunConnection, RunConnectionLease, @@ -265,7 +266,7 @@ function createHandle(connection: OpenConnection): Handle { }); } - const journal: DurableStream = { + const ordinaryJournal: DurableStream = { *readAll(): Operation { const entries = yield* mustSucceed(read(() => readJournalEntries(database))); return entries.map((entry) => entry.event); @@ -276,6 +277,7 @@ function createHandle(connection: OpenConnection): Handle { }, }; + let journal: DurableStream | undefined; const handle: WorkflowRunDatabase = { get record() { return record; @@ -285,7 +287,12 @@ function createHandle(connection: OpenConnection): Handle { return retrieval; }, - journal, + get journal() { + if (journal === undefined) { + throw new WorkflowTransactionError("the WorkflowRun journal route is not installed."); + } + return journal; + }, transact, @@ -405,6 +412,7 @@ function createHandle(connection: OpenConnection): Handle { }; lease = connection.connections.registerLease(handle, runConnection); + journal = routeWorkflowRunJournal(handle, ordinaryJournal); return { database: handle, diff --git a/packages/workflow/src/deno/journal-route.ts b/packages/workflow/src/deno/journal-route.ts new file mode 100644 index 00000000..93e2e969 --- /dev/null +++ b/packages/workflow/src/deno/journal-route.ts @@ -0,0 +1,132 @@ +import { type Api, createApi } from "@effectionx/context-api"; +import type { DurableEvent, DurableStream } from "@executablemd/durable-streams"; +import { type Operation, scoped } from "effection"; +import type { WorkflowRunDatabase, WorkflowRunTransaction } from "../storage/api.ts"; +import { WorkflowTransactionError } from "../storage/errors.ts"; +import type { WorkflowRunConnections, WorkflowRunTransactionToken } from "./connections.ts"; + +interface JournalDestinationApi { + append(database: WorkflowRunDatabase, event: DurableEvent): Operation; +} + +const ordinaryJournalDestination: JournalDestinationApi = { + // deno-lint-ignore require-yield + *append(_database: WorkflowRunDatabase, _event: DurableEvent): Operation { + return false; + }, +}; + +const JournalDestination: Api = createApi( + "executablemd.workflow.deno.journal.destination", + ordinaryJournalDestination, +); + +interface JournalRouteApi { + bind( + database: WorkflowRunDatabase, + transaction: WorkflowRunTransaction, + token: WorkflowRunTransactionToken, + publication: Operation, + ): Operation; +} + +function unavailable(): never { + throw new WorkflowTransactionError( + "the journal route is not owned by the active Deno workflow storage provider.", + ); +} + +const JournalRoute: Api = createApi( + "executablemd.workflow.deno.journal.route", + { + // deno-lint-ignore require-yield + *bind( + _database: WorkflowRunDatabase, + _transaction: WorkflowRunTransaction, + _token: WorkflowRunTransactionToken, + _publication: Operation, + ): Operation { + return unavailable(); + }, + }, +); + +function validateRoute( + connections: WorkflowRunConnections, + database: WorkflowRunDatabase, + transaction: WorkflowRunTransaction, + token: WorkflowRunTransactionToken, +): void { + const authorized = connections.authorizeTransaction(database, transaction); + const tokenTransaction = connections.validateToken(database, token); + if (authorized !== tokenTransaction) { + throw new WorkflowTransactionError( + "the journal destination token does not name this exact active transaction.", + ); + } +} + +export function* useJournalRouting(connections: WorkflowRunConnections): Operation { + yield* JournalDestination.around( + { + // deno-lint-ignore require-yield + *append(): Operation { + return false; + }, + }, + { at: "min" }, + ); + yield* JournalRoute.around( + { + *bind([database, transaction, token, publication]: [ + WorkflowRunDatabase, + WorkflowRunTransaction, + WorkflowRunTransactionToken, + Operation, + ]): Operation { + validateRoute(connections, database, transaction, token); + return yield* scoped(function* () { + yield* JournalDestination.around( + { + *append([candidate, event], next): Operation { + if (candidate !== database) { + return yield* next(candidate, event); + } + validateRoute(connections, database, transaction, token); + yield* transaction.journal.append(event); + return true; + }, + }, + { at: "min" }, + ); + return yield* publication; + }); + }, + }, + { at: "min" }, + ); +} + +export function routeWorkflowRunJournal( + database: WorkflowRunDatabase, + ordinary: DurableStream, +): DurableStream { + return { + readAll: () => ordinary.readAll(), + + *append(event: DurableEvent): Operation { + if (!(yield* JournalDestination.operations.append(database, event))) { + yield* ordinary.append(event); + } + }, + }; +} + +export function withEnlistedJournalRoute( + database: WorkflowRunDatabase, + transaction: WorkflowRunTransaction, + token: WorkflowRunTransactionToken, + publication: Operation, +): Operation { + return JournalRoute.operations.bind(database, transaction, token, publication); +} diff --git a/packages/workflow/src/deno/provider.ts b/packages/workflow/src/deno/provider.ts index 9bab35cb..377819c6 100644 --- a/packages/workflow/src/deno/provider.ts +++ b/packages/workflow/src/deno/provider.ts @@ -64,6 +64,7 @@ import { type WorkflowRunConnections, } from "./connections.ts"; import { workflowRunPath } from "./path.ts"; +import { useJournalRouting } from "./journal-route.ts"; import { readTransaction } from "./reading.ts"; import { initializeSchema, isUninitialized, translateSqliteError, verifySchema } from "./schema.ts"; import { SavepointObservation } from "./savepoints.ts"; @@ -111,6 +112,7 @@ export function* useWorkflowRunStorage(options: WorkflowRunStorageOptions): Oper yield* ensure(() => { connections.close(); }); + yield* useJournalRouting(connections); yield* usePrivateWorkspace(connections); yield* WorkflowRunStorage.around( diff --git a/packages/workflow/src/workspace/api.ts b/packages/workflow/src/workspace/api.ts new file mode 100644 index 00000000..b3fe0a3f --- /dev/null +++ b/packages/workflow/src/workspace/api.ts @@ -0,0 +1,27 @@ +import { type Api, createApi } from "@effectionx/context-api"; +import type { Json, LiveDurableOperationCoordinator, Result } from "@executablemd/durable-streams"; +import type { Operation } from "effection"; + +export type WorkspaceCoordinationApi = LiveDurableOperationCoordinator; + +/** A Workspace operation has no safe live fallback without its owning provider. */ +export class WorkspaceCoordinationProviderError extends Error { + override name = "WorkspaceCoordinationProviderError"; + + constructor() { + super( + "no Workspace coordinator is installed, so a live Workspace operation cannot execute or publish", + ); + } +} + +export const WorkspaceCoordination: Api = + createApi("executablemd.workflow.workspace.coordination", { + // deno-lint-ignore require-yield + *run( + _execute: () => Operation, + _publish: (result: Result) => Operation, + ): Operation { + throw new WorkspaceCoordinationProviderError(); + }, + }); diff --git a/packages/workflow/src/workspace/effect.ts b/packages/workflow/src/workspace/effect.ts new file mode 100644 index 00000000..d524e799 --- /dev/null +++ b/packages/workflow/src/workspace/effect.ts @@ -0,0 +1,29 @@ +import { + createDurableOperation, + type DurableEffect, + type EffectDescription, + type Json, + type LiveDurableOperationCoordinator, + type Result, +} from "@executablemd/durable-streams"; +import type { Operation } from "effection"; +import { WorkspaceCoordination } from "./api.ts"; + +const workspaceCoordinator: LiveDurableOperationCoordinator = { + *run( + execute: () => Operation, + publish: (result: Result) => Operation, + ): Operation { + return yield* WorkspaceCoordination.operations.run(execute, publish); + }, +}; + +/** Create a structured durable operation whose live path requires Workspace coordination. */ +export function createDurableWorkspaceOperation( + description: EffectDescription, + execute: () => Operation, +): DurableEffect { + return createDurableOperation(description, execute, { + coordinator: workspaceCoordinator, + }); +} diff --git a/packages/workflow/tests/workflow-run-journal.test.ts b/packages/workflow/tests/workflow-run-journal.test.ts index d01bfeee..366e9107 100644 --- a/packages/workflow/tests/workflow-run-journal.test.ts +++ b/packages/workflow/tests/workflow-run-journal.test.ts @@ -18,6 +18,7 @@ import { readFileSync, writeFileSync } from "node:fs"; import { join } from "node:path"; import process from "node:process"; import { fileURLToPath } from "node:url"; +import { createApi } from "@effectionx/context-api"; import { describe, it } from "@executablemd/test-support/bdd"; import { expect } from "@executablemd/test-support/expect"; import { exec } from "@executablemd/runtime"; @@ -33,13 +34,17 @@ import { all, ensure, type Operation, race, sleep, spawn, suspend, withResolvers import { WorkflowRecordMalformedError, WorkflowRequestError, + type WorkflowRunDatabase, WorkflowRunConflictError, WorkflowRunStorage, type WorkflowRunTransaction, WorkflowTransactionError, } from "../mod.ts"; +import { WorkflowRunTransactionToken } from "../src/deno/connections.ts"; +import { withEnlistedJournalRoute } from "../src/deno/journal-route.ts"; import { NoOpenTransactionError, savepoint } from "../src/deno/transaction.ts"; import { EMPTY_WORKSPACE_ROOT_ID } from "../src/deno/workspace/manifest.ts"; +import { workflowRunTransactionToken } from "../src/deno/workspace/private.ts"; import { allowJournalInserts, committedEventCount, @@ -54,6 +59,20 @@ import { const { create } = WorkflowRunStorage.operations; +interface CollidingJournalDestinationApi { + append(database: WorkflowRunDatabase, event: DurableEvent): Operation; +} + +const CollidingJournalDestination = createApi( + "executablemd.workflow.deno.journal.destination", + { + // deno-lint-ignore require-yield + *append(_database: WorkflowRunDatabase, _event: DurableEvent): Operation { + return false; + }, + }, +); + const REPOSITORY = fileURLToPath(new URL("../../../", import.meta.url)); const CHILD = fileURLToPath(new URL("./support/restart-child.ts", import.meta.url)); @@ -112,6 +131,15 @@ function* attempt(body: () => Operation): Operation { return undefined; } +function tokenValue( + value: unknown, + seed: WorkflowRunTransactionToken, +): WorkflowRunTransactionToken { + const container = { token: seed }; + Object.defineProperty(container, "token", { value, enumerable: true }); + return container.token; +} + /** One whole run in a process of its own, through the production adapter. */ function* runChild( root: string, @@ -1066,6 +1094,437 @@ describe("Tier WJ — two callers creating at once", () => { }); }); +describe("Tier WJ — explicit transaction journal routing", () => { + it("WJ26: an exact active token routes an already-filtered event after the gate", function* () { + const root = yield* useStorageRoot(); + + const events = yield* withStorage(root, function* () { + const database = yield* createRun({ runId: "routed" }); + const result = yield* database.transact(function* (transaction) { + const token = yield* workflowRunTransactionToken(database, transaction); + const guarded = guardDurableStream(database.journal, function* () { + expect(names(yield* transaction.journal.readAll())).toEqual([]); + }); + yield* withEnlistedJournalRoute( + database, + transaction, + token, + guarded.append(yielded("routed", "filtered")), + ); + expect(names(yield* transaction.journal.readAll())).toEqual(["routed"]); + }); + if (!result.ok) { + throw result.error; + } + return yield* database.journal.readAll(); + }); + + expect(names(events)).toEqual(["routed"]); + }); + + it("WJ27: rejected and cancelled gates reach no routed insertion", function* () { + const root = yield* useStorageRoot(); + + const events = yield* withStorage(root, function* () { + const database = yield* createRun({ runId: "route-gates" }); + const result = yield* database.transact(function* (transaction) { + const token = yield* workflowRunTransactionToken(database, transaction); + const rejected = guardDurableStream(database.journal, function* () { + throw new Error("secret gate refused the event"); + }); + expect( + yield* attempt(() => + withEnlistedJournalRoute( + database, + transaction, + token, + rejected.append(yielded("rejected", "rejected")), + ), + ), + ).toBeInstanceOf(Error); + + const gateStarted = withResolvers(); + const cancelled = guardDurableStream(database.journal, function* () { + gateStarted.resolve(); + yield* suspend(); + }); + const task = yield* spawn(() => + withEnlistedJournalRoute( + database, + transaction, + token, + cancelled.append(yielded("cancelled", "cancelled")), + ), + ); + yield* gateStarted.operation; + yield* task.halt(); + expect(names(yield* transaction.journal.readAll())).toEqual([]); + yield* transaction.journal.append(yielded("companion", "companion")); + }); + if (!result.ok) { + throw result.error; + } + return yield* database.journal.readAll(); + }); + + expect(names(events)).toEqual(["companion"]); + }); + + it("WJ28: missing, fabricated, foreign and cross-run routes fail before insertion", function* () { + const root = yield* useStorageRoot(); + + const result = yield* withStorage(root, function* () { + const first = yield* createRun({ runId: "route-first" }); + const second = yield* createRun({ runId: "route-second" }); + const transacted = yield* first.transact(function* (firstTransaction) { + const valid = yield* workflowRunTransactionToken(first, firstTransaction); + const missing = tokenValue(undefined, valid); + const fabricated = new WorkflowRunTransactionToken(); + const candidates = [missing, fabricated]; + for (const token of candidates) { + expect( + yield* attempt(() => + withEnlistedJournalRoute( + first, + firstTransaction, + token, + first.journal.append(yielded("unauthorized", "unauthorized")), + ), + ), + ).toBeInstanceOf(WorkflowTransactionError); + } + + const nested = yield* second.transact(function* (secondTransaction) { + expect( + yield* attempt(() => + withEnlistedJournalRoute( + second, + secondTransaction, + valid, + second.journal.append(yielded("cross-run", "cross-run")), + ), + ), + ).toBeInstanceOf(WorkflowTransactionError); + }); + if (!nested.ok) { + throw nested.error; + } + }); + if (!transacted.ok) { + throw transacted.error; + } + return { + first: yield* first.journal.readAll(), + second: yield* second.journal.readAll(), + }; + }); + + expect(result).toEqual({ first: [], second: [] }); + }); + + it("WJ29: completed, closed and stale-generation authority cannot bind", function* () { + const root = yield* useStorageRoot(); + let closedDatabase: WorkflowRunDatabase | undefined; + let completedTransaction: WorkflowRunTransaction | undefined; + let staleToken: WorkflowRunTransactionToken | undefined; + + yield* withStorage(root, function* () { + const database = yield* createRun({ runId: "route-stale" }); + closedDatabase = database; + const result = yield* database.transact(function* (transaction) { + completedTransaction = transaction; + staleToken = yield* workflowRunTransactionToken(database, transaction); + }); + if (!result.ok) { + throw result.error; + } + const transaction = completedTransaction; + const token = staleToken; + if (transaction === undefined || token === undefined) { + throw new Error("the transaction did not leave route authority for the refusal proof"); + } + expect( + yield* attempt(() => + withEnlistedJournalRoute( + database, + transaction, + token, + database.journal.append(yielded("completed", "completed")), + ), + ), + ).toBeInstanceOf(WorkflowTransactionError); + }); + + const closed = closedDatabase; + const transaction = completedTransaction; + const token = staleToken; + if (closed === undefined || transaction === undefined || token === undefined) { + throw new Error("the prior provider did not leave route authority"); + } + const events = yield* withStorage(root, function* () { + const found = yield* WorkflowRunStorage.operations.lookup("route-stale"); + if (!found.ok) { + throw found.error; + } + const database = found.value; + expect( + yield* attempt(() => + withEnlistedJournalRoute( + closed, + transaction, + token, + closed.journal.append(yielded("closed", "closed")), + ), + ), + ).toBeInstanceOf(WorkflowTransactionError); + + const current = yield* database.transact(function* (currentTransaction) { + expect( + yield* attempt(() => + withEnlistedJournalRoute( + database, + currentTransaction, + token, + database.journal.append(yielded("stale", "stale")), + ), + ), + ).toBeInstanceOf(WorkflowTransactionError); + }); + if (!current.ok) { + throw current.error; + } + return yield* database.journal.readAll(); + }); + expect(events).toEqual([]); + }); + + it("WJ30: escaped route authority appends nothing after transaction completion", function* () { + const root = yield* useStorageRoot(); + + const events = yield* withStorage(root, function* () { + const database = yield* createRun({ runId: "route-escaped" }); + let escapedTransaction: WorkflowRunTransaction | undefined; + let escapedToken: WorkflowRunTransactionToken | undefined; + const result = yield* database.transact(function* (transaction) { + escapedTransaction = transaction; + escapedToken = yield* workflowRunTransactionToken(database, transaction); + }); + if (!result.ok) { + throw result.error; + } + const transaction = escapedTransaction; + const token = escapedToken; + if (transaction === undefined || token === undefined) { + throw new Error("route authority did not escape for the refusal proof"); + } + expect( + yield* attempt(() => + withEnlistedJournalRoute( + database, + transaction, + token, + database.journal.append(yielded("late", "late")), + ), + ), + ).toBeInstanceOf(WorkflowTransactionError); + return yield* database.journal.readAll(); + }); + + expect(events).toEqual([]); + }); + + it("WJ31: an unrelated concurrent append never inherits an enlisted route", function* () { + const root = yield* useStorageRoot(); + + const events = yield* withStorage(root, function* () { + const database = yield* createRun({ runId: "route-unrelated" }); + const beginUnrelated = withResolvers(); + const unrelated = yield* spawn(function* () { + yield* beginUnrelated.operation; + yield* database.journal.append(yielded("unrelated", "unrelated")); + }); + + const result = yield* database.transact(function* (transaction) { + const token = yield* workflowRunTransactionToken(database, transaction); + yield* withEnlistedJournalRoute( + database, + transaction, + token, + database.journal.append(yielded("rolled-back", "rolled-back")), + ); + beginUnrelated.resolve(); + throw new Error("roll back the routed transaction"); + }); + expect(result.ok).toBe(false); + yield* unrelated; + return yield* database.journal.readAll(); + }); + + expect(names(events)).toEqual(["unrelated"]); + }); + + it("WJ32: a nested route for another run delegates to the enclosing destination", function* () { + const root = yield* useStorageRoot(); + let collisions = 0; + yield* CollidingJournalDestination.around( + { + // deno-lint-ignore require-yield + *append(): Operation { + collisions += 1; + return true; + }, + }, + { at: "min" }, + ); + + const events = yield* withStorage(root, function* () { + const first = yield* createRun({ runId: "route-outer" }); + const second = yield* createRun({ runId: "route-inner" }); + const outer = yield* first.transact(function* (firstTransaction) { + const firstToken = yield* workflowRunTransactionToken(first, firstTransaction); + const nested = yield* withEnlistedJournalRoute( + first, + firstTransaction, + firstToken, + (function* () { + return yield* second.transact(function* (secondTransaction) { + const secondToken = yield* workflowRunTransactionToken(second, secondTransaction); + yield* withEnlistedJournalRoute( + second, + secondTransaction, + secondToken, + (function* () { + yield* first.journal.append(yielded("outer", "outer")); + yield* second.journal.append(yielded("inner", "inner")); + })(), + ); + }); + })(), + ); + if (!nested.ok) { + throw nested.error; + } + }); + if (!outer.ok) { + throw outer.error; + } + return { + first: yield* first.journal.readAll(), + second: yield* second.journal.readAll(), + }; + }); + + expect(names(events.first)).toEqual(["outer"]); + expect(names(events.second)).toEqual(["inner"]); + expect(collisions).toBe(0); + }); + + it("WJ33: readAll remains ordinary replay and never invokes a gate or route", function* () { + const root = yield* useStorageRoot(); + + const result = yield* withStorage(root, function* () { + const database = yield* createRun({ runId: "route-read" }); + yield* database.journal.append(yielded("existing", "existing")); + let gates = 0; + const guarded = guardDurableStream(database.journal, function* () { + gates += 1; + }); + expect(names(yield* guarded.readAll())).toEqual(["existing"]); + + const transacted = yield* database.transact(function* (transaction) { + const token = yield* workflowRunTransactionToken(database, transaction); + yield* withEnlistedJournalRoute( + database, + transaction, + token, + (function* () { + expect(yield* attempt(() => guarded.readAll())).toBeInstanceOf( + WorkflowTransactionError, + ); + expect(names(yield* transaction.journal.readAll())).toEqual(["existing"]); + })(), + ); + }); + if (!transacted.ok) { + throw transacted.error; + } + return { gates, events: yield* guarded.readAll() }; + }); + + expect(result.gates).toBe(0); + expect(names(result.events)).toEqual(["existing"]); + }); + + it("WJ34: a colliding destination cannot suppress an ordinary append", function* () { + const root = yield* useStorageRoot(); + let collisions = 0; + yield* CollidingJournalDestination.around( + { + // deno-lint-ignore require-yield + *append(): Operation { + collisions += 1; + return true; + }, + }, + { at: "min" }, + ); + + const events = yield* withStorage(root, function* () { + const database = yield* createRun({ runId: "route-collision-ordinary" }); + yield* database.journal.append(yielded("ordinary", "ordinary")); + return yield* database.journal.readAll(); + }); + + expect(names(events)).toEqual(["ordinary"]); + expect(collisions).toBe(0); + }); + + it("WJ35: a colliding destination cannot suppress or authorize a routed append", function* () { + const root = yield* useStorageRoot(); + let collisions = 0; + yield* CollidingJournalDestination.around( + { + // deno-lint-ignore require-yield + *append(): Operation { + collisions += 1; + return true; + }, + }, + { at: "min" }, + ); + + const events = yield* withStorage(root, function* () { + const database = yield* createRun({ runId: "route-collision-routed" }); + const result = yield* database.transact(function* (transaction) { + const token = yield* workflowRunTransactionToken(database, transaction); + yield* withEnlistedJournalRoute( + database, + transaction, + token, + database.journal.append(yielded("routed", "routed")), + ); + expect( + yield* attempt(() => + withEnlistedJournalRoute( + database, + transaction, + tokenValue({}, token), + database.journal.append(yielded("fabricated", "fabricated")), + ), + ), + ).toBeInstanceOf(WorkflowTransactionError); + }); + if (!result.ok) { + throw result.error; + } + return yield* database.journal.readAll(); + }); + + expect(names(events)).toEqual(["routed"]); + expect(collisions).toBe(0); + }); +}); + describe("Tier WJ — surviving a process", () => { it("WJ24: two processes racing to create one run leave one winner", function* () { const root = yield* useStorageRoot(); diff --git a/packages/workflow/tests/workspace-effect.test.ts b/packages/workflow/tests/workspace-effect.test.ts new file mode 100644 index 00000000..789af599 --- /dev/null +++ b/packages/workflow/tests/workspace-effect.test.ts @@ -0,0 +1,131 @@ +import { describe, it } from "@executablemd/test-support/bdd"; +import { expect } from "@executablemd/test-support/expect"; +import { readTextFile } from "@effectionx/fs"; +import { type Operation } from "effection"; +import { + durableCall, + durableRun, + InMemoryStream, + type DurableEvent, + type Json, + type Result, + type Workflow, +} from "@executablemd/durable-streams"; +import { + createDurableWorkspaceOperation, + WorkspaceCoordination, + WorkspaceCoordinationProviderError, +} from "../mod.ts"; + +function* raised(operation: Operation): Operation { + try { + yield* operation; + return undefined; + } catch (error) { + return error; + } +} + +function yieldEvents(events: DurableEvent[]): DurableEvent[] { + return events.filter((event) => event.type === "yield"); +} + +function* workspaceStep(name: string, execute: () => Operation): Workflow { + yield createDurableWorkspaceOperation({ type: "workspace", name }, execute); +} + +describe("Tier DLC — Workspace coordination selection", () => { + it("DLC10: a missing Workspace provider fails before execution or publication", function* () { + const stream = new InMemoryStream(); + let executions = 0; + function* workflow(): Workflow { + yield* workspaceStep("missing", function* () { + executions += 1; + return "not reached"; + }); + } + + const failure = yield* raised(durableRun(workflow, { stream })); + expect(failure).toBeInstanceOf(WorkspaceCoordinationProviderError); + expect(executions).toBe(0); + expect(yieldEvents(stream.snapshot())).toEqual([]); + }); + + it("DLC11: explicit Workspace selection leaves unrelated durable operations ordinary", function* () { + const stream = new InMemoryStream(); + const coordinated: string[] = []; + const ordinary: string[] = []; + yield* WorkspaceCoordination.around({ + *run([execute, publish]: [ + () => Operation, + (result: Result) => Operation, + ]): Operation { + coordinated.push("workspace"); + const result: Result = { status: "ok", value: yield* execute() }; + yield* publish(result); + return result; + }, + }); + + function* workflow(): Workflow { + yield* workspaceStep("selected", function* () { + return "workspace"; + }); + yield* durableCall("ordinary", function* () { + ordinary.push("ordinary"); + return "ordinary"; + }); + return "done"; + } + + expect(yield* durableRun(workflow, { stream })).toBe("done"); + expect(coordinated).toEqual(["workspace"]); + expect(ordinary).toEqual(["ordinary"]); + expect(yieldEvents(stream.snapshot())).toHaveLength(2); + }); + + it("DLC12: replayed Workspace operations never require a live provider", function* () { + const stream = new InMemoryStream([ + { + type: "yield", + coroutineId: "root", + description: { type: "workspace", name: "replayed" }, + result: { status: "ok", value: "stored" }, + }, + ]); + let executions = 0; + function* workflow(): Workflow { + yield* workspaceStep("replayed", function* () { + executions += 1; + return "live"; + }); + return "done"; + } + + expect(yield* durableRun(workflow, { stream })).toBe("done"); + expect(executions).toBe(0); + expect(yieldEvents(stream.snapshot())).toHaveLength(1); + }); + + it("DLC13: shared Workspace coordination source stays runtime-neutral", function* () { + const sources = [ + yield* readTextFile(new URL("../src/workspace/api.ts", import.meta.url)), + yield* readTextFile(new URL("../src/workspace/effect.ts", import.meta.url)), + ]; + const forbidden = [ + "node:sqlite", + "DatabaseSync", + "SQLite", + "Cloudflare", + "DOFS", + "savepoint", + "ConnectionGeneration", + "TransactionIdentity", + ]; + for (const source of sources) { + for (const name of forbidden) { + expect(source.includes(name)).toBe(false); + } + } + }); +}); diff --git a/specs/executable-mdx-spec.md b/specs/executable-mdx-spec.md index 997624f9..700e7c4e 100644 --- a/specs/executable-mdx-spec.md +++ b/specs/executable-mdx-spec.md @@ -6638,6 +6638,35 @@ Defined in [Workflow runs](./workflow-spec.md) §9.5–§9.6. | WJ22/WJ23 | Concurrent creation | Compatible callers converge on one run; conflicting ones produce one winner and one conflict | | 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 | +| WJ26 | Exact routed publication | The existing secret gate completes before one exact active token delegates the already-filtered event to `transaction.journal` | +| WJ27 | Routed gate refusal | Gate rejection and cancellation reach no routed insertion | +| WJ28 | Invalid route authority | Missing, fabricated, foreign and cross-run authority is refused before insertion | +| WJ29/WJ30 | Expired route authority | Completed, closed, stale-generation and escaped authority cannot bind or append | +| WJ31 | No ambient enlistment | An unrelated concurrent append does not inherit a publication-local route and survives the routed transaction's rollback | +| WJ32 | Nested run routes | A route for another WorkflowRun delegates to the enclosing run's destination instead of hiding it, even under a colliding loaded-copy handler | +| WJ33 | Replay stays ordinary | `readAll()` never enlists and never invokes a secret gate | +| WJ34 | Terminal ordinary destination | A same-named enclosing handler cannot suppress an unbound ordinary append ahead of the provider-owned terminal destination | +| WJ35 | Terminal routed destination | A same-named enclosing handler cannot suppress a routed append or bypass provider-owned exact-token validation | + +### Tier DLC — Live durable-operation coordination + +Defined in [Workflow runs](./workflow-spec.md) §9.5–§9.6. + +| # | Test | Verify | +|---|------|--------| +| DLC1 | Default live success | Execution and publication each occur once, and publication completes before the caller resumes | +| DLC2 | Live execution failure | The existing serialized failed protocol `Result` is published exactly once | +| DLC3 | Publication failure | One failed backing append raises `DurablePersistenceError` with the adapter error as cause, persists no Yield or Close, and never resumes the workflow past publication | +| DLC4 | Active fail-stop | Catching a failed coordinated publication cannot invoke a later coordinator or executor, cannot attempt another append, and the first durability failure escapes at termination | +| DLC5 | Complete replay | Coordinator, execution, publication continuation and live append are all bypassed | +| DLC6 | Partial replay | Only the live suffix enters coordination | +| DLC7 | Cancellation | Cancellation during execution or publication produces no late or duplicate Yield | +| DLC8 | Explicit selection | A selected coordinator affects only the operation that names it | +| DLC9 | Callback compatibility | Callback-based durable effects retain their existing behavior | +| DLC10 | Fail-closed Workspace | A missing Workspace provider fails before execution or publication | +| DLC11 | Workspace isolation | Explicit Workspace selection leaves unrelated durable operations on the default coordinator | +| DLC12 | Workspace replay | Replayed Workspace operations require no live provider | +| DLC13 | Runtime-neutral boundary | Shared Workspace coordination source exposes no runtime or storage implementation type | ### Tier WTX — WorkflowRun savepoints and transaction authority diff --git a/specs/workflow-spec.md b/specs/workflow-spec.md index 969d15e5..e5a7a80d 100644 --- a/specs/workflow-spec.md +++ b/specs/workflow-spec.md @@ -319,13 +319,28 @@ root and otherwise retain their established behavior. Events arrive already filtered: ```text -DurableEvent → secret gate → journal append +DurableEvent → secret gate → Deno journal router → journal append ``` Storage performs no filtering of its own — a second policy in a second place is a second thing to keep in agreement with the first — and a gate that rejects or is cancelled leaves no row at all. +An unbound router append follows the ordinary standalone journal path and takes +its own serialized connection turn. Only a publication continuation may bind a +transaction destination, and that binding is lexical to the publication rather +than the operation's execution. The route validates the exact database lease, +connection generation, transaction identity, private token and open state +before delegating to the existing `transaction.journal`; it does not duplicate +insertion SQL. Missing, foreign, fabricated, completed, closed, cross-run and +stale authority reaches no SQL. A route for another WorkflowRun delegates to an +enclosing route instead of hiding it. The provider's ordinary destination and +each publication-local routed destination are terminal `{ at: "min" }` +handlers. An enclosing loaded copy with the same stable contextual name cannot +acknowledge an append before the provider either selects the ordinary path or +performs exact route validation. `readAll()` remains ordinary replay and neither +routes nor invokes the secret gate. + ### 9.6 One authoritative connection, one operation The Deno provider maps each canonical workflow-run database path to one @@ -421,9 +436,27 @@ last-seen metadata. The supplied private Workspace body runs in an inner scope, and final live/current validation waits for that scope's children and resources to finish teardown. -Successful effect coordination finishes its mutation scope before capturing -the root. The provider-level coordinator that orders mutation teardown, root -capture and filtered journal publication is not part of this storage layer. +Structured durable operations accept an explicit provider-neutral live +coordinator. The default executes once, serializes execution success or failure +into the existing protocol `Result`, invokes the existing Yield publication +continuation exactly once, and returns that same result only after publication. +The continuation uses the durable stream's ordered append fence. A backing +append failure activates fail-stop state and raises `DurablePersistenceError` +with the adapter error as its cause; it writes no compensating `Close`, and +later durable work cannot execute or append even when workflow code catches the +failure. A marked pre-persistence policy rejection remains the policy's ordinary +failure. There is no generic durable validation hook: a caller or provider +validates before constructing the durable effect. Replay bypasses the +coordinator, execution, publication and live append; partial replay coordinates +only its live suffix. Cancellation cannot append or resolve late. The +callback-based durable-effect factory remains unchanged. + +The shared Workspace operation wrapper explicitly selects a contextual +Workspace coordinator. Its default fails before execution or publication, and +installing a provider does not enlist unrelated durable operations. Successful +Workspace effect coordination finishes its mutation scope before capturing the +root. The Deno coordinator that orders mutation teardown, root capture and +filtered journal publication atomically is not part of this storage layer. The private restoration materializer loads a fully validated retained root and rebuilds directories, files, chunks, modes, mtimes, symbolic links and hardlink @@ -503,8 +536,7 @@ also left unchanged. Public `xmd workflow` lifecycle commands; lifecycle transition policy, executor leases and stale-owner recovery; public Workspace mutation and filesystem -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; +effects; provider-level atomic Workspace effect/journal publication; public root +selection, history checkpoints and forks; `` integration; workflow-owned worktrees; and deterministic Git and GitHub effects. Retained roots and private restoration do not expose any of those behaviors.