diff --git a/architecture.md b/architecture.md index 8d921091..5cc28fbd 100644 --- a/architecture.md +++ b/architecture.md @@ -398,11 +398,43 @@ 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. +The shared Workspace durable-operation wrapper reads a contextual provider +selection explicitly. That replaceable API carries only a non-operational +selection identity: execution, publication, failure activation and durable +publication identity never pass through that middleware. Provider selection +creates a one-use route and a separate opaque credential. The live call keeps +that credential out of its contextual invocation request and places the exact +executor, publisher and failure activator behind an execution-owned capability +that accepts only the credential. A provider installed by any loaded package +copy terminally consumes the route and invokes that capability directly. It +does not send inspection, execution, publication, failure or completion through +the contextual continuation. Enclosing middleware at either priority can +therefore route or refuse the initial request but cannot observe or acknowledge +an operational phase. No module registry or replaceable structural value +resolves provider or invocation authority. + +The execution-owned capability accepts execution and publication once and +records the exact `Result` only after the provider returns it from the completed +publication. The live operation ignores the contextual call's response as +completion evidence and resumes from that recorded result alone. A short +circuit or altered response therefore activates fail-stop before resume, while +an exception or substituted response after authoritative completion cannot +replace the published result. Foreign or substituted selectors and reused, +completed or stale calls fail before transaction work. Installing no provider +cannot leak execution or publication, and selecting Workspace coordination for +one operation does not enlist unrelated durable operations in the same scope. + +The Deno adapter retains the canonical module's non-operational identity for +each WorkflowRun journal. The existing secret guard privately associates only +its own wrapper, including nested official wrappers, with that same identity. +Before it opens a transaction, the adapter requires both the exact proof +executor and the publication identity from the consumed invocation to belong to +the selected WorkflowRun. An in-memory stream, another run's journal, a copied +property, or an unproven wrapper is refused before mutation or publication. +Inside the transaction, the provider's publication operation calls the +execution-owned publisher directly and returns only after that publisher records +the exact Result. The transaction body cannot return or commit on a contextual +publication acknowledgement. Every Workspace-local expansion publishes one effect through one effect transaction: @@ -421,9 +453,14 @@ staging, switching and committing, operate on the same transactional Workspace rather than invoking an untracked native Git side effect. Successful effect coordination finishes the mutation scope, including child -cleanup, before capturing the resulting root. The provider-level coordinator -that performs this ordering and journal publication is not installed at this -layer. +cleanup, before capturing the resulting root. The Deno provider installs that +ordering for its adapter-private Workspace proof operation: the mutation +savepoint, root publication, filtered routed Yield, and caller-owned transaction +commit form one boundary. The proof filesystem uses the pinned synchronous DOFS +entry points for its string and byte-array surface, so cancellation leaves no +eager promise or stream pull able to reach the connection after transaction +authority ends. Public filesystem components and workflow lifecycle commands do +not yet select that operation. An external provider cannot join that transaction. Prompt, Git push and pull request effects derive a stable identity from the run and expansion, ask the @@ -482,8 +519,9 @@ materialize one privately through the authoritative connection. Capture runs inside the caller-owned transaction. Restoration runs in a nested savepoint, clears the authoritative resolution and blob caches, and resnapshots to the selected identity before release. 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. +their child teardown before final live/current validation. The Workspace effect +coordinator also finishes its mutation savepoint scope before it invokes +capture. Separately, every unsuccessful caller-owned transaction attempts top-level SQLite rollback and then @@ -496,10 +534,21 @@ 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 and explicit -filtered-journal routing are present, while the Deno Workspace coordinator that -combines mutation, root publication and journal publication atomically is -absent. +operation at this layer. Its adapter-private coordinator combines one mutation, +immutable-root publication and one filtered journal result atomically for the +provider-level proof; declarative `` and workflow start/resume do not +reach it yet. + +The coordinator treats only errors produced through its private filesystem +adapter's documented path and mutation refusals as journalable operation +failures. It rolls their savepoint back, keeps the prior current root, and +commits one failed protocol Yield. Connection, savepoint, content, capture, +routing, filtering, insertion, teardown and commit failures roll back the outer +transaction and activate the durable run's first-failure fence. Cancellation is +control flow: it publishes no failed result. The durable coordinator contract +supplies a provider-neutral failure activator so a provider can retain that +first infrastructure failure by identity; the default coordinator does not use +it and keeps its existing behavior. The Deno journal adapter routes an append ordinarily when no destination is bound. A publication may instead bind one exact transaction destination for its @@ -892,10 +941,10 @@ Status is measured against main. | `Expansion` / `getExpansion()` | describes the current logical element expansion | built on main | | `useWorkflow()` / `getWorkflowRun()` | associates one document execution with a workflow run | built on main | | `Git.revParse()` | verifies and resolves one Git revision expression contextually | built on main | -| workflow run storage | creates or compatibly finds one run by public run ID, 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 | +| 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; public workflow execution 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 | +| Workspace coordination API | fails closed by default; replaceable context routes only a one-use provider selection, while the selected provider directly invokes an execution-owned credentialed capability for execution, publication and failure activation | built on the #365 stack; the Deno provider installs an adapter-private atomic handler | | 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 | | `API.Files` | routes every document filesystem operation to the installed provider, with no host default and structural failure data | built on the #227 stack | @@ -909,7 +958,7 @@ Status is measured against main. | Repository / Worktree / transactional Git effects | compose named checkouts and publish local mutations with their journal result | defined in `specs/workflow-workspace-spec.md`, unbuilt | | workflow inspection and history fork | reads status/history without advancing a run and creates a new run from a checkpoint | defined in `specs/workflow-workspace-spec.md`, unbuilt | | read-only workflow Agent / generated XMD | lets an Agent inspect a derived view and propose constrained executable changes | defined in `specs/workflow-workspace-spec.md`, unbuilt | -| Deno-local DOFS provider | owns one authoritative SQLite/DOFS connection per run path, captures arbitrary canonical retained roots, and privately restores them with cache-coherent savepoints | built on the #365 stack; public mutation and effect-transaction integration are unbuilt | +| Deno-local DOFS provider | owns one authoritative SQLite/DOFS connection per run path, captures arbitrary canonical retained roots, privately restores them, and atomically coordinates an adapter-private mutation proof with its filtered Yield | built on the #365 stack; public mutation and workflow lifecycle reachability are unbuilt | | scoped Worker Shell | executes `just-bash` through the Workspace adapter inside a Deno Worker | containment and effect-transaction POCs complete (#351, #357); production integration unbuilt | | `` | retry a region until it completes | defined, unbuilt | | suspension effect | suspend durably | defined, unbuilt | diff --git a/packages/durable-streams/README.md b/packages/durable-streams/README.md index 0c028ffa..7531ba9f 100644 --- a/packages/durable-streams/README.md +++ b/packages/durable-streams/README.md @@ -296,6 +296,21 @@ same result after publication completes. Replay bypasses the coordinator, executor, continuation, and live append; a partially replayed run coordinates only its live suffix. +The coordinator receives a non-operational publication identity and an +infrastructure-failure activator. The canonical durable-stream module keeps the +association between a claimed backend and that identity in private weak state. +`guardDurableStream` associates only the wrapper it constructs with its source's +identity, including through nested guards. Streams expose no identity property +or inheritance callback; copied properties, custom wrappers and wrappers from +another loaded copy acquire no authority. + +A provider +whose execution and publication share a larger durability boundary calls it +when that boundary cannot commit. It returns the first active failure by +identity, so the provider throws that exact failure and later durable work stays +fenced. The default coordinator does not call this continuation: an ordinary +execution failure remains the existing failed protocol `Result`. + 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 diff --git a/packages/durable-streams/effect.ts b/packages/durable-streams/effect.ts index 46ae2a61..8b8c1c6e 100644 --- a/packages/durable-streams/effect.ts +++ b/packages/durable-streams/effect.ts @@ -29,7 +29,9 @@ import { rememberDurabilityFailure, } from "./durability.ts"; import { StaleInputError } from "./errors.ts"; +import { durablePublicationIdentity } from "./guard.ts"; import { + type ActivateDurabilityFailure, defaultLiveDurableOperationCoordinator, type LiveDurableOperationCoordinator, } from "./live-coordinator.ts"; @@ -351,15 +353,28 @@ export function createDurableOperation( try { 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); - }); + const activateFailure: ActivateDurabilityFailure = (failure) => { + const existing = activeDurabilityFailure(ctx); + if (existing) { + return existing; + } + const error = failure instanceof Error ? failure : new Error(String(failure)); + return rememberDurabilityFailure(ctx, error); + }; + const result = yield* coordinator.run( + execute, + function* (published) { + const event: Yield = { + type: "yield", + coroutineId: ctx.coroutineId, + description: desc, + result: published, + }; + yield* appendDurableEvent(ctx, event); + }, + activateFailure, + durablePublicationIdentity(ctx.stream), + ); resolve(protocolToEffection(result)); } catch (err) { resolve({ diff --git a/packages/durable-streams/guard.ts b/packages/durable-streams/guard.ts index e5bef107..60a3a62f 100644 --- a/packages/durable-streams/guard.ts +++ b/packages/durable-streams/guard.ts @@ -18,6 +18,53 @@ import { createContext, type Operation } from "effection"; import type { DurableStream } from "./stream.ts"; import type { DurableEvent } from "./types.ts"; +class PublicationIdentity { + #opaque = undefined; +} + +/** Non-operational identity for one durable publication backend. */ +export type DurablePublicationIdentity = PublicationIdentity; + +const publicationIdentities = (() => { + // This security identity is deliberately canonical-module-local. A loaded + // copy cannot enroll its wrappers into this copy's authority. + const identities = new WeakMap(); + + return { + claim(stream: DurableStream): DurablePublicationIdentity { + if (identities.has(stream)) { + throw new Error("this durable stream already has a publication identity"); + } + const identity = new PublicationIdentity(); + identities.set(stream, identity); + return identity; + }, + + inherit(source: DurableStream, target: DurableStream): void { + const identity = identities.get(source); + if (identity !== undefined) { + identities.set(target, identity); + } + }, + + get(stream: DurableStream): DurablePublicationIdentity | undefined { + return identities.get(stream); + }, + }; +})(); + +/** Claim the exact backend identity retained by a provider. */ +export function claimDurablePublicationIdentity(stream: DurableStream): DurablePublicationIdentity { + return publicationIdentities.claim(stream); +} + +/** @internal The live durable path reads identity without receiving stream authority. */ +export function durablePublicationIdentity( + stream: DurableStream, +): DurablePublicationIdentity | undefined { + return publicationIdentities.get(stream); +} + export interface DurableEventRejectionOccurrence { rejected: boolean; error?: unknown; @@ -60,7 +107,7 @@ export type DurableEventGate = (event: DurableEvent) => Operation; * event with an `err` result, and that close crosses the gate on its own. */ export function guardDurableStream(stream: DurableStream, gate: DurableEventGate): DurableStream { - return { + const guarded: DurableStream = { readAll: () => stream.readAll(), *append(event: DurableEvent): Operation { @@ -80,4 +127,6 @@ export function guardDurableStream(stream: DurableStream, gate: DurableEventGate yield* stream.append(event); }, }; + publicationIdentities.inherit(stream, guarded); + return guarded; } diff --git a/packages/durable-streams/live-coordinator.ts b/packages/durable-streams/live-coordinator.ts index 3a84e113..0dbdd431 100644 --- a/packages/durable-streams/live-coordinator.ts +++ b/packages/durable-streams/live-coordinator.ts @@ -1,12 +1,18 @@ import type { Operation } from "effection"; +import type { DurablePublicationIdentity } from "./guard.ts"; import { serializeError } from "./serialize.ts"; import type { Json, Result } from "./types.ts"; +/** Activates the first infrastructure failure for the enclosing durable run. */ +export type ActivateDurabilityFailure = (failure: unknown) => Error; + /** Coordinates one live structured durable operation with its publication. */ export interface LiveDurableOperationCoordinator { run( execute: () => Operation, publish: (result: Result) => Operation, + activateFailure: ActivateDurabilityFailure, + publicationIdentity: DurablePublicationIdentity | undefined, ): Operation; } @@ -15,6 +21,8 @@ export const defaultLiveDurableOperationCoordinator: LiveDurableOperationCoordin *run( execute: () => Operation, publish: (result: Result) => Operation, + _activateFailure: ActivateDurabilityFailure, + _publicationIdentity: DurablePublicationIdentity | undefined, ): Operation { let result: Result; try { diff --git a/packages/durable-streams/mod.ts b/packages/durable-streams/mod.ts index ef2fde44..00ca7d03 100644 --- a/packages/durable-streams/mod.ts +++ b/packages/durable-streams/mod.ts @@ -33,7 +33,8 @@ export { InMemoryStream } from "./stream.ts"; // Pre-persistence gate — runs before an event reaches its backend export { guardDurableStream } from "./guard.ts"; -export type { DurableEventGate } from "./guard.ts"; +export { claimDurablePublicationIdentity } from "./guard.ts"; +export type { DurableEventGate, DurablePublicationIdentity } from "./guard.ts"; // HTTP-backed stream adapter export { useHttpDurableStream } from "./http-stream.ts"; @@ -79,7 +80,10 @@ export type { Executor } from "./effect.ts"; // Structured live-operation coordination export { defaultLiveDurableOperationCoordinator } from "./live-coordinator.ts"; -export type { LiveDurableOperationCoordinator } from "./live-coordinator.ts"; +export type { + ActivateDurabilityFailure, + LiveDurableOperationCoordinator, +} from "./live-coordinator.ts"; // Workflow-enabled effects export { durableAction, durableCall, durableSleep, versionCheck } from "./operations.ts"; diff --git a/packages/durable-streams/tests/guard-stream.test.ts b/packages/durable-streams/tests/guard-stream.test.ts index 4a8c5c1a..6114568b 100644 --- a/packages/durable-streams/tests/guard-stream.test.ts +++ b/packages/durable-streams/tests/guard-stream.test.ts @@ -16,6 +16,7 @@ import * as fs from "node:fs"; import * as os from "node:os"; import * as path from "node:path"; import { + claimDurablePublicationIdentity, durableAll, durableCall, durableRun, @@ -23,6 +24,7 @@ import { InMemoryStream, serializeDurableEvent, } from "../mod.ts"; +import { durablePublicationIdentity } from "../guard.ts"; import type { DurableEvent, DurableStream, Workflow } from "../mod.ts"; /** A short, readable identity for an event in a timeline assertion. */ @@ -96,6 +98,37 @@ const EVENT: DurableEvent = { }; describe("guardDurableStream", () => { + it("retains publication identity only through canonical guarded wrappers", function* () { + const backend = new InMemoryStream(); + const identity = claimDurablePublicationIdentity(backend); + const guarded = guardDurableStream(backend, function* () {}); + const nested = guardDurableStream(guarded, function* () {}); + const loadedCopySpecifier = "../guard.ts" + "?loaded-copy=publication-identity"; + const loadGuardCopy: () => Promise = () => + import(loadedCopySpecifier); + const loadedCopy = yield* until(loadGuardCopy()); + const unverifiable = loadedCopy.guardDurableStream(backend, function* () {}); + const other = new InMemoryStream(); + const otherIdentity = claimDurablePublicationIdentity(other); + const unknown: DurableStream = { + readAll: () => nested.readAll(), + append: (event) => nested.append(event), + }; + const oldInheritance = Symbol.for("executablemd.durable-stream.inherit-provenance"); + Object.defineProperty(unknown, oldInheritance, { + value: () => undefined, + }); + + expect(durablePublicationIdentity(backend)).toBe(identity); + expect(durablePublicationIdentity(guarded)).toBe(identity); + expect(durablePublicationIdentity(nested)).toBe(identity); + expect(durablePublicationIdentity(other)).toBe(otherIdentity); + expect(otherIdentity).not.toBe(identity); + expect(durablePublicationIdentity(unknown)).toBe(undefined); + expect(durablePublicationIdentity(unverifiable)).toBe(undefined); + expect(Reflect.ownKeys(nested).includes(oldInheritance)).toBe(false); + }); + it("delegates readAll without invoking the gate", function* () { const timeline: string[] = []; const guarded = guardDurableStream(recordingStream(timeline, [EVENT]), function* () { diff --git a/packages/durable-streams/tests/live-coordinator.test.ts b/packages/durable-streams/tests/live-coordinator.test.ts index c19dd43b..aa270b5d 100644 --- a/packages/durable-streams/tests/live-coordinator.test.ts +++ b/packages/durable-streams/tests/live-coordinator.test.ts @@ -3,12 +3,15 @@ import { expect } from "@executablemd/test-support/expect"; import { type Operation, spawn, suspend, withResolvers } from "effection"; import { createDurableOperation, + type ActivateDurabilityFailure, + claimDurablePublicationIdentity, durableAction, durableCall, durableRun, DurablePersistenceError, InMemoryStream, type DurableEvent, + type DurablePublicationIdentity, type DurableStream, type Json, type LiveDurableOperationCoordinator, @@ -333,13 +336,19 @@ describe("Tier DLC — live durable-operation coordination", () => { it("DLC8: an explicit coordinator affects only its selected operation", function* () { const stream = new InMemoryStream(); + const identity = claimDurablePublicationIdentity(stream); let coordinated = 0; let ordinary = 0; const coordinator: LiveDurableOperationCoordinator = { *run( execute: () => Operation, publish: (result: Result) => Operation, + _activateFailure: ActivateDurabilityFailure, + publicationIdentity: DurablePublicationIdentity | undefined, ): Operation { + expect(publicationIdentity).toBe(identity); + expect(Reflect.get(publicationIdentity ?? {}, "append")).toBe(undefined); + expect(Reflect.get(publicationIdentity ?? {}, "readAll")).toBe(undefined); coordinated += 1; const result: Result = { status: "ok", value: yield* execute() }; yield* publish(result); @@ -382,4 +391,49 @@ describe("Tier DLC — live durable-operation coordination", () => { expect(executions).toBe(1); expect(yields(stream.snapshot())).toHaveLength(1); }); + + it("DLC14: a coordinator activates one infrastructure failure by identity", function* () { + const infrastructureFailure = new Error("coordinated infrastructure failed"); + const stream = new InMemoryStream(); + let caught: unknown; + let laterExecutions = 0; + const coordinator: LiveDurableOperationCoordinator = { + *run( + execute: () => Operation, + _publish: (result: Result) => Operation, + activateFailure: ActivateDurabilityFailure, + ): Operation { + try { + yield* execute(); + } catch (error) { + throw activateFailure(error); + } + throw new Error("the infrastructure proof unexpectedly completed"); + }, + }; + + function* workflow(): Workflow { + try { + yield* coordinatedStep( + "infrastructure", + function* () { + throw infrastructureFailure; + }, + coordinator, + ); + } catch (error) { + caught = error; + } + yield* durableCall("fenced", function* () { + laterExecutions += 1; + return null; + }); + } + + const escaped = yield* raised(durableRun(workflow, { stream })); + expect(escaped).toBe(infrastructureFailure); + expect(caught).toBe(infrastructureFailure); + expect(laterExecutions).toBe(0); + expect(stream.appendCount).toBe(0); + }); }); diff --git a/packages/workflow/src/deno/connections.ts b/packages/workflow/src/deno/connections.ts index 33686f4a..31d4785c 100644 --- a/packages/workflow/src/deno/connections.ts +++ b/packages/workflow/src/deno/connections.ts @@ -1,6 +1,13 @@ import { DatabaseSync } from "node:sqlite"; import { resolve } from "node:path"; import type { WorkflowRunDatabase, WorkflowRunTransaction } from "../storage/api.ts"; +import { + claimDurablePublicationIdentity, + type DurableEvent, + type DurablePublicationIdentity, + type DurableStream, +} from "@executablemd/durable-streams"; +import type { Operation } from "effection"; 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"; @@ -36,6 +43,7 @@ export interface RunConnectionLease { readonly generation: ConnectionGeneration; readonly path: string; readonly database: WorkflowRunDatabase; + journalIdentity: DurablePublicationIdentity | undefined; open: boolean; } @@ -70,8 +78,15 @@ export interface RunConnection { export interface WorkflowRunConnections { at(path: string): RunConnection; registerLease(database: WorkflowRunDatabase, connection: RunConnection): RunConnectionLease; + registerJournal(database: WorkflowRunDatabase, journal: DurableStream): void; closeLease(lease: RunConnectionLease): void; validateLease(database: WorkflowRunDatabase): RunConnectionLease; + validateJournal( + database: WorkflowRunDatabase, + identity: DurablePublicationIdentity | undefined, + ): void; + afterRoutedJournalAppend(database: WorkflowRunDatabase, event: DurableEvent): Operation; + beforeCommit(database: WorkflowRunDatabase): Operation; authorizeTransaction( database: WorkflowRunDatabase, transaction: WorkflowRunTransaction, @@ -273,8 +288,19 @@ export class WorkflowConnectionStateError extends Error { override name = "WorkflowConnectionStateError"; } +export interface WorkflowRunConnectionHooks { + afterRoutedJournalAppend?(database: WorkflowRunDatabase, event: DurableEvent): Operation; + beforeCommit?(database: WorkflowRunDatabase): Operation; +} + +// deno-lint-ignore require-yield +function* noop(): Operation { + return undefined; +} + export function createWorkflowRunConnections( observeSavepoint: SavepointObserver = () => {}, + hooks: WorkflowRunConnectionHooks = {}, ): WorkflowRunConnections { const entries = new Map(); const leases = new WeakMap(); @@ -337,17 +363,50 @@ export function createWorkflowRunConnections( generation: connection.generation, path: connection.path, database, + journalIdentity: undefined, open: true, }; leases.set(database, lease); return lease; }, + registerJournal(database: WorkflowRunDatabase, journal: DurableStream): void { + const lease = validateLease(database); + if (lease.journalIdentity !== undefined) { + throw new WorkflowTransactionError( + "the WorkflowRun database journal identity is already installed.", + ); + } + lease.journalIdentity = claimDurablePublicationIdentity(journal); + }, + closeLease(lease: RunConnectionLease): void { lease.open = false; }, validateLease, + + validateJournal( + database: WorkflowRunDatabase, + identity: DurablePublicationIdentity | undefined, + ): void { + const selected = validateLease(database).journalIdentity; + if (selected === undefined || selected !== identity) { + throw new WorkflowTransactionError( + "the live Workspace journal is foreign to the selected WorkflowRun database.", + ); + } + }, + + afterRoutedJournalAppend(database: WorkflowRunDatabase, event: DurableEvent): Operation { + validateLease(database); + return hooks.afterRoutedJournalAppend?.(database, event) ?? noop(); + }, + + beforeCommit(database: WorkflowRunDatabase): Operation { + validateLease(database); + return hooks.beforeCommit?.(database) ?? noop(); + }, authorizeTransaction, issueToken( diff --git a/packages/workflow/src/deno/database.ts b/packages/workflow/src/deno/database.ts index 94cc3147..aa0ce9ce 100644 --- a/packages/workflow/src/deno/database.ts +++ b/packages/workflow/src/deno/database.ts @@ -250,6 +250,8 @@ function createHandle(connection: OpenConnection): Handle { return yield* body(transaction); }); + yield* connection.connections.beforeCommit(handle); + // Closed before the commit, not after: nothing may append to a // transaction whose contents are already decided. runConnection.validateTransaction(active); @@ -413,6 +415,7 @@ function createHandle(connection: OpenConnection): Handle { lease = connection.connections.registerLease(handle, runConnection); journal = routeWorkflowRunJournal(handle, ordinaryJournal); + connection.connections.registerJournal(handle, journal); return { database: handle, diff --git a/packages/workflow/src/deno/journal-route.ts b/packages/workflow/src/deno/journal-route.ts index 93e2e969..a5bddc1b 100644 --- a/packages/workflow/src/deno/journal-route.ts +++ b/packages/workflow/src/deno/journal-route.ts @@ -94,6 +94,7 @@ export function* useJournalRouting(connections: WorkflowRunConnections): Operati } validateRoute(connections, database, transaction, token); yield* transaction.journal.append(event); + yield* connections.afterRoutedJournalAppend(database, event); return true; }, }, diff --git a/packages/workflow/src/deno/provider.ts b/packages/workflow/src/deno/provider.ts index 377819c6..5d5379ae 100644 --- a/packages/workflow/src/deno/provider.ts +++ b/packages/workflow/src/deno/provider.ts @@ -69,6 +69,7 @@ import { readTransaction } from "./reading.ts"; import { initializeSchema, isUninitialized, translateSqliteError, verifySchema } from "./schema.ts"; import { SavepointObservation } from "./savepoints.ts"; import { usePrivateWorkspace } from "./workspace/private.ts"; +import { useWorkspaceEffects } from "./workspace/effect.ts"; const INSERT_RUN = `INSERT INTO workflow_run (id, run_id, definition, base, props, status, created_at, updated_at) @@ -114,6 +115,7 @@ export function* useWorkflowRunStorage(options: WorkflowRunStorageOptions): Oper }); yield* useJournalRouting(connections); yield* usePrivateWorkspace(connections); + yield* useWorkspaceEffects(connections); yield* WorkflowRunStorage.around( { diff --git a/packages/workflow/src/deno/workspace/effect.ts b/packages/workflow/src/deno/workspace/effect.ts new file mode 100644 index 00000000..39409d4b --- /dev/null +++ b/packages/workflow/src/deno/workspace/effect.ts @@ -0,0 +1,230 @@ +import { type Api, createApi } from "@effectionx/context-api"; +import { + type DurableEffect, + type EffectDescription, + type Json, + type Result as DurableResult, + serializeError, +} from "@executablemd/durable-streams"; +import { ensure, type Operation, scoped } from "effection"; +import type { WorkflowRunDatabase, WorkflowRunTransaction } from "../../storage/api.ts"; +import { WorkflowTransactionError } from "../../storage/errors.ts"; +import { + createOwnedDurableWorkspaceOperation, + type WorkspaceCoordinationAuthority, + type WorkspaceCoordinationProvider, + withWorkspaceCoordinationProvider, +} from "../../workspace/effect.ts"; +import type { WorkflowRunConnections } from "../connections.ts"; +import { withEnlistedJournalRoute } from "../journal-route.ts"; +import { savepoint } from "../transaction.ts"; +import { isJournalableWorkspaceFailure } from "./errors.ts"; +import type { DenoWorkspaceFilesystem } from "./filesystem.ts"; +import { + type PrivateWorkspaceTransaction, + withPrivateWorkspaceTransaction, + workflowRunTransactionToken, +} from "./private.ts"; + +export type DenoWorkspaceMutation = ( + filesystem: DenoWorkspaceFilesystem, +) => Operation; + +interface WorkspaceMutationApi { + run( + database: WorkflowRunDatabase, + mutate: DenoWorkspaceMutation, + ): Operation; +} + +function unavailable(): never { + throw new WorkflowTransactionError( + "the Workspace effect is not bound to this active Deno WorkflowRun transaction.", + ); +} + +const WorkspaceMutation: Api = createApi( + "executablemd.workflow.deno.workspace.effect.mutation", + { + // deno-lint-ignore require-yield + *run( + _database: WorkflowRunDatabase, + _mutate: DenoWorkspaceMutation, + ): Operation { + return unavailable(); + }, + }, +); + +const workspaceEffectOwners = (() => { + const owners = new WeakMap(); + return { + claim(execute: object, database: WorkflowRunDatabase): void { + owners.set(execute, database); + }, + + get(execute: object): WorkflowRunDatabase | undefined { + return owners.get(execute); + }, + }; +})(); + +interface WorkspaceEffectProviderApi { + readonly provider: object | undefined; +} + +interface WorkspaceEffectProviderRegistration { + open: boolean; + readonly connections: WorkflowRunConnections; +} + +const WorkspaceEffectProvider: Api = + createApi("executablemd.workflow.deno.workspace.effect.provider", { + provider: undefined, + }); + +const workspaceEffectProviders = (() => { + const providers = new WeakMap(); + + return { + register(connections: WorkflowRunConnections): { + selection: object; + close: () => void; + } { + const selection = Object.freeze({}); + const registration: WorkspaceEffectProviderRegistration = { open: true, connections }; + providers.set(selection, registration); + return { + selection, + close(): void { + registration.open = false; + providers.delete(selection); + }, + }; + }, + + get(selection: object): WorkflowRunConnections | undefined { + const registration = providers.get(selection); + return registration?.open ? registration.connections : undefined; + }, + }; +})(); + +function* runMutation( + database: WorkflowRunDatabase, + workspace: PrivateWorkspaceTransaction, + execute: () => Operation, +): Operation { + return yield* scoped(function* () { + yield* WorkspaceMutation.around( + { + *run([candidate, mutate]: [ + WorkflowRunDatabase, + DenoWorkspaceMutation, + ]): Operation { + if (candidate !== database) { + return unavailable(); + } + return yield* mutate(workspace.filesystem); + }, + }, + { at: "min" }, + ); + return yield* execute(); + }); +} + +function* coordinateTransaction( + database: WorkflowRunDatabase, + transaction: WorkflowRunTransaction, + workspace: PrivateWorkspaceTransaction, + execute: () => Operation, + publish: (result: DurableResult) => Operation, +): Operation { + const token = yield* workflowRunTransactionToken(database, transaction); + + let result: DurableResult; + try { + const value = yield* savepoint(runMutation(database, workspace, execute)); + result = { status: "ok", value }; + const root = yield* workspace.capture(); + yield* workspace.publish(root.rootId); + } catch (error) { + if (!isJournalableWorkspaceFailure(error)) { + throw error; + } + result = { status: "err", error: serializeError(error) }; + } + + yield* withEnlistedJournalRoute(database, transaction, token, publish(result)); + return result; +} + +function coordinator( + connections: WorkflowRunConnections, + database: WorkflowRunDatabase, +): WorkspaceCoordinationProvider { + return { + *run(authority: WorkspaceCoordinationAuthority): Operation { + let transacted; + try { + if (workspaceEffectOwners.get(authority.executionIdentity) !== database) { + throw new WorkflowTransactionError( + "the live Workspace effect is missing, foreign, completed, or stale for this WorkflowRun database.", + ); + } + connections.validateJournal(database, authority.publicationIdentity); + transacted = yield* database.transact(function* (transaction) { + return yield* withPrivateWorkspaceTransaction(database, transaction, (workspace) => + coordinateTransaction( + database, + transaction, + workspace, + authority.execute, + authority.publish, + ), + ); + }); + } catch (error) { + throw yield* authority.activateFailure(error); + } + if (!transacted.ok) { + throw yield* authority.activateFailure(transacted.error); + } + return transacted.value; + }, + }; +} + +export function* useWorkspaceEffects(connections: WorkflowRunConnections): Operation { + const registration = workspaceEffectProviders.register(connections); + yield* ensure(registration.close); + yield* WorkspaceEffectProvider.around({ provider: () => registration.selection }, { at: "min" }); +} + +export function withWorkspaceEffects( + database: WorkflowRunDatabase, + operation: Operation, +): Operation { + return scoped(function* () { + const selection = yield* WorkspaceEffectProvider.operations.provider; + const connections = + selection === undefined ? undefined : workspaceEffectProviders.get(selection); + if (connections === undefined) { + return unavailable(); + } + connections.validateLease(database); + return yield* withWorkspaceCoordinationProvider(coordinator(connections, database), operation); + }); +} + +export function createWorkspaceProofEffect( + database: WorkflowRunDatabase, + description: EffectDescription, + mutate: DenoWorkspaceMutation, +): DurableEffect { + const execute = () => WorkspaceMutation.operations.run(database, mutate); + const executionIdentity = Object.freeze({}); + workspaceEffectOwners.claim(executionIdentity, database); + return createOwnedDurableWorkspaceOperation(description, execute, executionIdentity); +} diff --git a/packages/workflow/src/deno/workspace/errors.ts b/packages/workflow/src/deno/workspace/errors.ts new file mode 100644 index 00000000..ebd6c156 --- /dev/null +++ b/packages/workflow/src/deno/workspace/errors.ts @@ -0,0 +1,49 @@ +const JOURNALABLE_CODES = new Set([ + "ENOENT", + "ENOTEMPTY", + "ENOTDIR", + "EISDIR", + "EEXIST", + "EINVAL", + "EACCES", + "EPERM", + "EROFS", + "ENOSYS", + "EBADF", + "ELOOP", +]); + +class JournalableWorkspaceFailure extends Error { + override name = "WorkspaceFsError"; + readonly code: string; + + constructor(source: Error, code: string) { + super(source.message, { cause: source }); + this.code = code; + } +} + +function journalableCode(error: unknown): string | undefined { + const code = error instanceof Error ? Reflect.get(error, "code") : undefined; + if ( + error instanceof Error && + error.name === "WorkspaceFsError" && + typeof code === "string" && + JOURNALABLE_CODES.has(code) + ) { + return code; + } + return undefined; +} + +export function throwWorkspaceFilesystemFailure(error: unknown): never { + const code = journalableCode(error); + if (error instanceof Error && code !== undefined) { + throw new JournalableWorkspaceFailure(error, code); + } + throw error; +} + +export function isJournalableWorkspaceFailure(error: unknown): error is Error { + return error instanceof JournalableWorkspaceFailure; +} diff --git a/packages/workflow/src/deno/workspace/filesystem.ts b/packages/workflow/src/deno/workspace/filesystem.ts index 9f30ac0b..6fd30412 100644 --- a/packages/workflow/src/deno/workspace/filesystem.ts +++ b/packages/workflow/src/deno/workspace/filesystem.ts @@ -1,8 +1,21 @@ -import { type Operation, until } from "effection"; +import { type Operation } from "effection"; +import { chmod as chmodPath } from "../../../vendor/cloudflare-computer-dofs/generated/fs/chmod.js"; import { link as linkFile } from "../../../vendor/cloudflare-computer-dofs/generated/fs/link.js"; +import { mkdir as mkdirPath } from "../../../vendor/cloudflare-computer-dofs/generated/fs/mkdir.js"; +import { readdir as readDirectory } from "../../../vendor/cloudflare-computer-dofs/generated/fs/readdir.js"; import type { WorkspaceDirentResult } from "../../../vendor/cloudflare-computer-dofs/generated/fs/readdir.d.ts"; +import { readRangeSync } from "../../../vendor/cloudflare-computer-dofs/generated/fs/readFile.js"; +import { readlink as readLink } from "../../../vendor/cloudflare-computer-dofs/generated/fs/readlink.js"; import { rename as renamePath } from "../../../vendor/cloudflare-computer-dofs/generated/fs/rename.js"; +import { rm as removePath } from "../../../vendor/cloudflare-computer-dofs/generated/fs/rm.js"; +import { + lstat as lstatPath, + stat as statPath, +} from "../../../vendor/cloudflare-computer-dofs/generated/fs/stat.js"; +import { symlink as createSymlink } from "../../../vendor/cloudflare-computer-dofs/generated/fs/symlink.js"; +import { writeFileSync } from "../../../vendor/cloudflare-computer-dofs/generated/fs/writeFile.js"; import type { RunConnection } from "../connections.ts"; +import { throwWorkspaceFilesystemFailure } from "./errors.ts"; export interface DenoWorkspaceEntry { readonly name: string; @@ -50,40 +63,48 @@ export function createDenoWorkspaceFilesystem( return { kind, mode: value.mode, mtime: value.mtime, size: value.size }; } + function filesystemOperation(operation: () => T): T { + try { + return operation(); + } catch (error) { + return throwWorkspaceFilesystemFailure(error); + } + } + + function execute(operation: () => T): T { + authorize(); + return filesystemOperation(operation); + } + + function readBytes(path: string): Uint8Array { + const metadata = statPath(dofs, path); + return new Uint8Array(readRangeSync(dofs, path, 0, metadata.size)); + } + return { *readFile(path): Operation { - authorize(); - const stream = yield* until(filesystem.readFile(path)); - return new Uint8Array(yield* until(new Response(stream).arrayBuffer())); + return execute(() => readBytes(path)); }, *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"); - } - return value; + const bytes = execute(() => readBytes(path)); + return new TextDecoder().decode(bytes); }, *stat(path): Operation { - authorize(); - return stat(yield* until(filesystem.stat(path))); + return stat(execute(() => statPath(dofs, path))); }, *lstat(path): Operation { - authorize(); - return stat(yield* until(filesystem.lstat(path))); + return stat(execute(() => lstatPath(dofs, path))); }, *readlink(path): Operation { - authorize(); - return yield* until(filesystem.readlink(path)); + return execute(() => readLink(dofs, path)); }, *readdir(path): Operation { - authorize(); - const entries = yield* until(filesystem.readdir(path)); + const entries = execute(() => readDirectory(dofs, path)); return entries.map((entry: WorkspaceDirentResult) => ({ name: entry.name, kind: entry.isFile ? "file" : entry.isDirectory ? "directory" : "symlink", @@ -91,40 +112,34 @@ export function createDenoWorkspaceFilesystem( }, *writeFile(path, content, mode): Operation { - authorize(); - yield* until(filesystem.writeFile(path, content, mode === undefined ? {} : { mode })); + const bytes = typeof content === "string" ? new TextEncoder().encode(content) : content; + execute(() => + writeFileSync(dofs, path, bytes, mode === undefined ? {} : { mode }, filesystem.now), + ); }, *mkdir(path, options = {}): Operation { - authorize(); - yield* until(filesystem.mkdir(path, options)); + execute(() => mkdirPath(dofs, path, options, filesystem.now)); }, *remove(path, options = {}): Operation { - authorize(); - yield* until(filesystem.rm(path, options)); + execute(() => removePath(dofs, path, options)); }, - // deno-lint-ignore require-yield *rename(from, to): Operation { - authorize(); - renamePath(dofs, from, to); + execute(() => renamePath(dofs, from, to)); }, *chmod(path, mode): Operation { - authorize(); - yield* until(filesystem.chmod(path, mode)); + execute(() => chmodPath(dofs, path, mode, filesystem.now)); }, *symlink(target, path): Operation { - authorize(); - yield* until(filesystem.symlink(target, path)); + execute(() => createSymlink(dofs, target, path, filesystem.now)); }, - // deno-lint-ignore require-yield *link(existingPath, newPath): Operation { - authorize(); - linkFile(dofs, existingPath, newPath); + execute(() => linkFile(dofs, existingPath, newPath)); }, }; } diff --git a/packages/workflow/src/deno/workspace/private.ts b/packages/workflow/src/deno/workspace/private.ts index a478a334..713087ef 100644 --- a/packages/workflow/src/deno/workspace/private.ts +++ b/packages/workflow/src/deno/workspace/private.ts @@ -9,6 +9,7 @@ import { captureWorkspaceRoot, type CaptureWorkspaceRootOptions, currentWorkspaceRoot, + setCurrentWorkspaceRoot, verifyWorkspace, } from "./root.ts"; import { restoreWorkspaceRoot, type RestoreWorkspaceRootOptions } from "./restore.ts"; @@ -17,6 +18,7 @@ export interface PrivateWorkspaceTransaction { readonly filesystem: DenoWorkspaceFilesystem; currentRoot(): Operation; capture(options?: CaptureWorkspaceRootOptions): Operation; + publish(rootId: string): Operation; restore(rootId: string, options?: RestoreWorkspaceRootOptions): Operation; } @@ -97,6 +99,12 @@ export function usePrivateWorkspace(connections: WorkflowRunConnections): Operat return captureWorkspaceRoot(connection, active, options); }, + // deno-lint-ignore require-yield + *publish(rootId: string): Operation { + authorize(); + setCurrentWorkspaceRoot(connection.database, rootId, connection.path); + }, + // deno-lint-ignore require-yield *restore(rootId, options = {}): Operation { authorize(); diff --git a/packages/workflow/src/workspace/api.ts b/packages/workflow/src/workspace/api.ts index b3fe0a3f..0a870e5f 100644 --- a/packages/workflow/src/workspace/api.ts +++ b/packages/workflow/src/workspace/api.ts @@ -1,27 +1,22 @@ 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; +export interface WorkspaceCoordinationApi { + readonly provider: object | undefined; +} /** 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", - ); + constructor( + message = "no Workspace coordinator is installed, so a live Workspace operation cannot execute or publish", + ) { + super(message); } } +/** Selects a Workspace provider without carrying live-operation authority. */ 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(); - }, + provider: undefined, }); diff --git a/packages/workflow/src/workspace/effect.ts b/packages/workflow/src/workspace/effect.ts index d524e799..1ea2534a 100644 --- a/packages/workflow/src/workspace/effect.ts +++ b/packages/workflow/src/workspace/effect.ts @@ -1,29 +1,312 @@ +import { type Api, createApi } from "@effectionx/context-api"; import { createDurableOperation, + type ActivateDurabilityFailure, type DurableEffect, + type DurablePublicationIdentity, 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); +import { ensure, type Operation, scoped } from "effection"; +import { WorkspaceCoordination, WorkspaceCoordinationProviderError } from "./api.ts"; + +export interface WorkspaceCoordinationAuthority { + readonly executionIdentity: object; + readonly publicationIdentity: DurablePublicationIdentity | undefined; + execute(): Operation; + publish(result: Result): Operation; + activateFailure(failure: unknown): Operation; +} + +export interface WorkspaceCoordinationProvider { + run(authority: WorkspaceCoordinationAuthority): Operation; +} + +interface StartRequest { + readonly type: "start"; + readonly route: object; + readonly invocation: WorkspaceInvocationCapability; +} + +interface WorkspaceInvocationDetails { + readonly executionIdentity: object; + readonly publicationIdentity: DurablePublicationIdentity | undefined; +} + +interface WorkspaceInvocationCapability { + inspect(credential: object): Operation; + execute(credential: object): Operation; + publish(credential: object, result: Result): Operation; + activateFailure(credential: object, failure: unknown): Operation; + complete(credential: object, result: Result): Operation; +} + +interface WorkspaceInvocationApi { + coordinate(request: StartRequest): Operation; +} + +interface ProviderSelection { + readonly route: object; + readonly credential: object; +} + +const WORKSPACE_INVOCATION_API = "executablemd.workflow.workspace.coordination.invocation"; + +function unavailable(message: string): WorkspaceCoordinationProviderError { + return new WorkspaceCoordinationProviderError(message); +} + +const WorkspaceInvocation: Api = createApi( + WORKSPACE_INVOCATION_API, + { + // deno-lint-ignore require-yield + *coordinate(): Operation { + throw unavailable("no Workspace coordinator accepted this live invocation"); + }, }, -}; +); -/** Create a structured durable operation whose live path requires Workspace coordination. */ -export function createDurableWorkspaceOperation( +function providerSelection(value: object | undefined): ProviderSelection { + const route = value === undefined ? undefined : Reflect.get(value, "route"); + const credential = value === undefined ? undefined : Reflect.get(value, "credential"); + if ( + typeof route !== "object" || + route === null || + typeof credential !== "object" || + credential === null + ) { + throw unavailable("the selected Workspace coordinator is missing, foreign, or substituted"); + } + return { route, credential }; +} + +export function withWorkspaceCoordinationProvider( + provider: WorkspaceCoordinationProvider, + operation: Operation, +): Operation { + return scoped(function* () { + const registrations = new WeakMap(); + let registrationOpen = true; + yield* ensure(() => { + registrationOpen = false; + }); + yield* WorkspaceCoordination.around( + { + provider(): object { + const route = Object.freeze({}); + const credential = Object.freeze({}); + registrations.set(route, credential); + return Object.freeze({ route, credential }); + }, + }, + { at: "min" }, + ); + yield* WorkspaceInvocation.around( + { + *coordinate([request]): Operation { + const credential = registrations.get(request.route); + registrations.delete(request.route); + if (!registrationOpen || request.type !== "start" || credential === undefined) { + throw unavailable( + "the selected Workspace coordinator is missing, foreign, completed, or stale", + ); + } + + const details = yield* request.invocation.inspect(credential); + let authorityOpen = true; + const authority: WorkspaceCoordinationAuthority = Object.freeze({ + executionIdentity: details.executionIdentity, + publicationIdentity: details.publicationIdentity, + *execute(): Operation { + if (!authorityOpen) { + throw unavailable( + "the live Workspace coordination authority is completed or stale", + ); + } + return yield* request.invocation.execute(credential); + }, + *publish(result: Result): Operation { + if (!authorityOpen) { + throw unavailable( + "the live Workspace coordination authority is completed or stale", + ); + } + yield* request.invocation.publish(credential, result); + }, + *activateFailure(failure: unknown): Operation { + if (!authorityOpen) { + throw unavailable( + "the live Workspace coordination authority is completed or stale", + ); + } + return yield* request.invocation.activateFailure(credential, failure); + }, + }); + + try { + const result = yield* provider.run(authority); + yield* request.invocation.complete(credential, result); + } finally { + authorityOpen = false; + } + }, + }, + { at: "min" }, + ); + return yield* operation; + }); +} + +function invocationCapability( + credential: object, + executionIdentity: object, + execute: () => Operation, + publish: (result: Result) => Operation, + activateFailure: ActivateDurabilityFailure, + publicationIdentity: DurablePublicationIdentity | undefined, +): { + capability: WorkspaceInvocationCapability; + authoritativeResult: () => Result | undefined; + close: () => void; +} { + let state: "available" | "active" | "complete" = "available"; + let executionAttempted = false; + let publishedResult: Result | undefined; + let completedResult: Result | undefined; + + function requireActive(): void { + if (state !== "active") { + throw unavailable("the live Workspace coordination invocation is completed or stale"); + } + } + + function requireCredential(candidate: object): void { + if (candidate !== credential) { + throw unavailable("the live Workspace coordination invocation has foreign authority"); + } + } + + return { + capability: Object.freeze({ + *inspect(candidate: object): Operation { + requireCredential(candidate); + if (state !== "available") { + throw unavailable( + "the live Workspace coordination invocation is missing, reused, completed, or stale", + ); + } + state = "active"; + return { executionIdentity, publicationIdentity }; + }, + *execute(candidate: object): Operation { + requireCredential(candidate); + requireActive(); + if (executionAttempted) { + throw unavailable("the live Workspace execution is already consumed"); + } + executionAttempted = true; + return yield* execute(); + }, + *publish(candidate: object, result: Result): Operation { + requireCredential(candidate); + requireActive(); + if (!executionAttempted || publishedResult !== undefined) { + throw unavailable("the live Workspace publication is missing or already consumed"); + } + yield* publish(result); + publishedResult = result; + }, + *activateFailure(candidate: object, failure: unknown): Operation { + requireCredential(candidate); + requireActive(); + return activateFailure(failure); + }, + *complete(candidate: object, result: Result): Operation { + requireCredential(candidate); + requireActive(); + if (publishedResult === undefined || result !== publishedResult) { + throw unavailable("the selected Workspace provider omitted its live publication"); + } + completedResult = publishedResult; + state = "complete"; + }, + }), + authoritativeResult(): Result | undefined { + return completedResult; + }, + close(): void { + state = "complete"; + }, + }; +} + +function workspaceCoordinator(executionIdentity: object): LiveDurableOperationCoordinator { + return { + *run( + execute: () => Operation, + publish: (result: Result) => Operation, + activateFailure: ActivateDurabilityFailure, + publicationIdentity: DurablePublicationIdentity | undefined, + ): Operation { + let invocation: ReturnType | undefined; + + try { + const selection = providerSelection(yield* WorkspaceCoordination.operations.provider); + invocation = invocationCapability( + selection.credential, + executionIdentity, + execute, + publish, + activateFailure, + publicationIdentity, + ); + yield* WorkspaceInvocation.operations.coordinate({ + type: "start", + route: selection.route, + invocation: invocation.capability, + }); + const result = invocation.authoritativeResult(); + if (result === undefined) { + throw unavailable("the selected Workspace provider did not complete its live invocation"); + } + return result; + } catch (error) { + const result = invocation?.authoritativeResult(); + if (result !== undefined) { + return result; + } + throw activateFailure(error); + } finally { + invocation?.close(); + } + }, + }; +} + +function createDurableWorkspaceOperationWithIdentity( description: EffectDescription, execute: () => Operation, + executionIdentity: object, ): DurableEffect { return createDurableOperation(description, execute, { - coordinator: workspaceCoordinator, + coordinator: workspaceCoordinator(executionIdentity), }); } + +/** Create a structured durable operation whose live path requires Workspace coordination. */ +export function createDurableWorkspaceOperation( + description: EffectDescription, + execute: () => Operation, +): DurableEffect { + return createDurableWorkspaceOperationWithIdentity(description, execute, Object.freeze({})); +} + +export function createOwnedDurableWorkspaceOperation( + description: EffectDescription, + execute: () => Operation, + executionIdentity: object, +): DurableEffect { + return createDurableWorkspaceOperationWithIdentity(description, execute, executionIdentity); +} diff --git a/packages/workflow/tests/workspace-effect-loaded-copy.test.ts b/packages/workflow/tests/workspace-effect-loaded-copy.test.ts new file mode 100644 index 00000000..1756c3f9 --- /dev/null +++ b/packages/workflow/tests/workspace-effect-loaded-copy.test.ts @@ -0,0 +1,148 @@ +import { describe, it } from "@executablemd/test-support/bdd"; +import { expect } from "@executablemd/test-support/expect"; +import { readTextFile, rm, writeTextFile } from "@effectionx/fs"; +import { ensure, type Operation, scoped, until } from "effection"; +import { pathToFileURL } from "node:url"; +import { + claimDurablePublicationIdentity, + durableRun, + InMemoryStream, + type Json, + type Result, + type Workflow, +} from "@executablemd/durable-streams"; +import { WorkspaceCoordination, WorkspaceCoordinationProviderError } from "../src/workspace/api.ts"; +import { + type WorkspaceCoordinationAuthority, + type WorkspaceCoordinationProvider, + withWorkspaceCoordinationProvider, +} from "../src/workspace/effect.ts"; +import { createDurableWorkspaceOperation } from "../mod.ts"; + +interface LoadedWorkspaceCopy { + createDurableWorkspaceOperation: typeof createDurableWorkspaceOperation; +} + +function loadedWorkspaceCopy(value: unknown): value is LoadedWorkspaceCopy { + return ( + typeof value === "object" && + value !== null && + typeof Reflect.get(value, "createDurableWorkspaceOperation") === "function" + ); +} + +function* physicalWorkspaceCopy(): Operation { + const directory = yield* until(Deno.makeTempDir({ prefix: "xmd-workflow-copy-" })); + yield* ensure(() => rm(directory, { recursive: true, force: true })); + const source = new URL("../src/workspace/", import.meta.url); + const destination = pathToFileURL(`${directory}/`); + yield* writeTextFile( + new URL("api.ts", destination), + yield* readTextFile(new URL("api.ts", source)), + ); + yield* writeTextFile( + new URL("effect.ts", destination), + yield* readTextFile(new URL("effect.ts", source)), + ); + const copy = yield* until(import(new URL("effect.ts", destination).href)); + if (!loadedWorkspaceCopy(copy)) { + throw new Error("the physical Workspace package copy did not export its durable operation"); + } + return copy; +} + +function* raised(operation: Operation): Operation { + try { + yield* operation; + return undefined; + } catch (error) { + return error; + } +} + +describe("Tier DLC — physical Workspace package composition", () => { + it("DLC16: provider selection composes without sharing invocation authority", function* () { + const copy = yield* physicalWorkspaceCopy(); + const stream = new InMemoryStream(); + claimDurablePublicationIdentity(stream); + let providers = 0; + let transactions = 0; + let executions = 0; + let publications = 0; + let capturedAuthority: WorkspaceCoordinationAuthority | undefined; + const provider: WorkspaceCoordinationProvider = { + *run(authority: WorkspaceCoordinationAuthority): Operation { + providers += 1; + transactions += 1; + capturedAuthority = authority; + const result: Result = { status: "ok", value: yield* authority.execute() }; + publications += 1; + yield* authority.publish(result); + return result; + }, + }; + function* workflow(): Workflow { + yield copy.createDurableWorkspaceOperation( + { type: "workspace", name: "physical-copy" }, + function* (): Operation { + executions += 1; + return "copied"; + }, + ); + } + + yield* withWorkspaceCoordinationProvider(provider, durableRun(workflow, { stream })); + + expect({ providers, transactions, executions, publications }).toEqual({ + providers: 1, + transactions: 1, + executions: 1, + publications: 1, + }); + expect(stream.snapshot().filter((event) => event.type === "yield")).toHaveLength(1); + if (capturedAuthority === undefined) { + throw new Error("the provider did not receive the physical copy invocation"); + } + expect(yield* raised(capturedAuthority.execute())).toBeInstanceOf( + WorkspaceCoordinationProviderError, + ); + expect( + yield* raised(capturedAuthority.publish({ status: "ok", value: "reused" })), + ).toBeInstanceOf(WorkspaceCoordinationProviderError); + expect(yield* raised(capturedAuthority.activateFailure(new Error("reused")))).toBeInstanceOf( + WorkspaceCoordinationProviderError, + ); + + const refusedStream = new InMemoryStream(); + claimDurablePublicationIdentity(refusedStream); + let refusedExecutions = 0; + function* refusedWorkflow(): Workflow { + yield copy.createDurableWorkspaceOperation( + { type: "workspace", name: "substituted-copy" }, + function* (): Operation { + refusedExecutions += 1; + return "not reached"; + }, + ); + } + + const failure = yield* scoped(function* () { + yield* WorkspaceCoordination.around({ provider: () => Object.freeze({}) }); + return yield* raised( + withWorkspaceCoordinationProvider( + provider, + durableRun(refusedWorkflow, { stream: refusedStream }), + ), + ); + }); + + expect(Reflect.get(failure ?? {}, "name")).toBe("WorkspaceCoordinationProviderError"); + expect({ providers, transactions, refusedExecutions, publications }).toEqual({ + providers: 1, + transactions: 1, + refusedExecutions: 0, + publications: 1, + }); + expect(refusedStream.snapshot()).toEqual([]); + }); +}); diff --git a/packages/workflow/tests/workspace-effect-transaction.test.ts b/packages/workflow/tests/workspace-effect-transaction.test.ts new file mode 100644 index 00000000..4cc9743e --- /dev/null +++ b/packages/workflow/tests/workspace-effect-transaction.test.ts @@ -0,0 +1,1483 @@ +import { join } from "node:path"; +import { constants, DatabaseSync } from "node:sqlite"; +import { describe, it } from "@executablemd/test-support/bdd"; +import { expect } from "@executablemd/test-support/expect"; +import { createApi } from "@effectionx/context-api"; +import { readTextFile } from "@effectionx/fs"; +import { + type ActivateDurabilityFailure, + durableCall, + durableRun, + guardDurableStream, + InMemoryStream, + type DurableEvent, + type DurablePublicationIdentity, + type DurableStream, + DurablePersistenceError, + type Json, + type Result, + type Workflow, + type Yield, +} from "@executablemd/durable-streams"; +import { ensure, type Operation, scoped, spawn, suspend, withResolvers } from "effection"; +import { + createDurableWorkspaceOperation, + WorkspaceCoordination, + type WorkflowRunDatabase, +} from "../mod.ts"; +import { + createWorkflowRunConnections, + type RunConnection, + type WorkflowRunConnectionHooks, +} from "../src/deno/connections.ts"; +import { openWorkflowRunDatabase, readRunRow } from "../src/deno/database.ts"; +import { useJournalRouting } from "../src/deno/journal-route.ts"; +import { SavepointObservation, type SavepointObserver } from "../src/deno/savepoints.ts"; +import { initializeSchema } from "../src/deno/schema.ts"; +import { + createWorkspaceProofEffect, + useWorkspaceEffects, + withWorkspaceEffects, +} from "../src/deno/workspace/effect.ts"; +import { definitionToJson } from "../src/storage/definition.ts"; +import { canonicalJson } from "../src/storage/record.ts"; +import { type DenoWorkspaceFilesystem } from "../src/deno/workspace/filesystem.ts"; +import { + setPrivateWorkspaceClock, + usePrivateWorkspace, + withPrivateWorkspaceTransaction, +} from "../src/deno/workspace/private.ts"; +import { + allowJournalInserts, + committedEventCount, + createRun, + refuseJournalInsertNamed, + request, + runPath, + tamper, + useStorageRoot, + withStorage, +} from "./support/storage.ts"; + +function sqliteConstant(name: string): number { + const value = Reflect.get(constants, name); + if (typeof value !== "number") { + throw new Error(`this Deno node:sqlite adapter does not expose ${name}`); + } + return value; +} + +const SQLITE_SAVEPOINT = sqliteConstant("SQLITE_SAVEPOINT"); +const SQLITE_OK = sqliteConstant("SQLITE_OK"); +const SQLITE_DENY = sqliteConstant("SQLITE_DENY"); + +interface InvocationCollisionApi { + coordinate(request: unknown): Operation; +} + +const WorkspaceInvocationCollision = createApi( + "executablemd.workflow.workspace.coordination.invocation", + { + // deno-lint-ignore require-yield + *coordinate(): Operation { + throw new Error("the collision handler did not delegate"); + }, + }, +); + +function* raised(operation: Operation): Operation { + try { + yield* operation; + return undefined; + } catch (error) { + return error; + } +} + +function workspaceYields(events: DurableEvent[]): Yield[] { + return events.filter( + (event): event is Yield => + event.type === "yield" && event.description.type === "workspace-proof", + ); +} + +function* workspaceStep( + database: WorkflowRunDatabase, + name: string, + mutate: (filesystem: DenoWorkspaceFilesystem) => Operation, +): Workflow { + yield createWorkspaceProofEffect(database, { type: "workspace-proof", name }, mutate); +} + +function* inspectWorkspace( + database: WorkflowRunDatabase, + path: string, +): Operation<{ root: string; content: string | undefined }> { + const inspected = yield* database.transact(function* (transaction) { + return yield* withPrivateWorkspaceTransaction(database, transaction, function* (workspace) { + let content: string | undefined; + try { + content = yield* workspace.filesystem.readTextFile(path); + } catch { + content = undefined; + } + return { + root: yield* workspace.currentRoot(), + content, + }; + }); + }); + if (!inspected.ok) { + throw inspected.error; + } + return inspected.value; +} + +function journalRoot(path: string, name: string): string | undefined { + const sqlite = new DatabaseSync(path); + try { + const row = sqlite + .prepare( + `SELECT workspace_root_id FROM journal_events + WHERE record LIKE ? ORDER BY sequence LIMIT 1`, + ) + .get(`%"name":"${name}"%`); + const root = row?.["workspace_root_id"]; + return typeof root === "string" ? root : undefined; + } finally { + sqlite.close(); + } +} + +function retainedRootCount(path: string): number { + const sqlite = new DatabaseSync(path); + try { + const value = sqlite.prepare("SELECT COUNT(*) AS count FROM workspace_roots").get()?.["count"]; + return typeof value === "number" ? value : Number(value); + } finally { + sqlite.close(); + } +} + +function withDirectWorkspaceStorage( + root: string, + name: string, + observe: SavepointObserver, + body: (database: WorkflowRunDatabase, connection: RunConnection) => Operation, + hooks: WorkflowRunConnectionHooks = {}, +): Operation { + return scoped(function* () { + const path = join(root, `${name}.sqlite`); + const connections = createWorkflowRunConnections(observe, hooks); + yield* ensure(() => connections.close()); + const connection = connections.at(path); + const wanted = request({ runId: name }); + const stamp = new Date(1_750_000_000_000).toISOString(); + connection.database.exec("BEGIN IMMEDIATE"); + const initializing = connection.beginTransaction(); + try { + initializeSchema(connection.database, connection.dofs, () => { + connection.database + .prepare( + `INSERT INTO workflow_run + (id, run_id, definition, base, props, status, created_at, updated_at) + VALUES (1, ?, ?, ?, ?, 'running', ?, ?)`, + ) + .run( + wanted.runId, + canonicalJson(definitionToJson(wanted.definition)), + wanted.base, + canonicalJson(wanted.props), + stamp, + stamp, + ); + }); + connection.validateTransaction(initializing); + connection.finishTransaction(initializing); + connection.database.exec("COMMIT"); + } catch (error) { + if (initializing.open) { + connection.finishTransaction(initializing); + } + connection.database.exec("ROLLBACK"); + throw error; + } + + yield* useJournalRouting(connections); + yield* usePrivateWorkspace(connections); + yield* useWorkspaceEffects(connections); + const database = yield* openWorkflowRunDatabase({ + connection, + connections, + record: readRunRow(connection.database, path), + }); + return yield* body(database, connection); + }); +} + +function setSqliteAuthorizer( + database: DatabaseSync, + authorize: ((action: number, operation: string | null, name: string | null) => number) | null, +): void { + const install = Reflect.get(database, "setAuthorizer"); + if (typeof install !== "function") { + throw new Error("this Deno node:sqlite adapter does not expose setAuthorizer"); + } + Reflect.apply(install, database, [authorize]); +} + +describe("Tier WAC — atomic provider-level Workspace effects", () => { + it("WAC1: mutation, root, filtered Yield and commit become visible together", function* () { + const root = yield* useStorageRoot(); + const runId = "atomic-success"; + const path = join(root, `${runId}.sqlite`); + let selected: WorkflowRunDatabase | undefined; + let baselineRoot: string | undefined; + let observedRoutedAppend = false; + + yield* withDirectWorkspaceStorage( + root, + runId, + () => {}, + function* (database) { + selected = database; + yield* setPrivateWorkspaceClock(database, () => 1_750_000_000_000); + const baseline = yield* inspectWorkspace(database, "/kept.txt"); + baselineRoot = baseline.root; + let gateCalls = 0; + const guardedOnce = guardDurableStream(database.journal, function* (event) { + if (event.type === "yield") { + gateCalls += 1; + } + }); + const guarded = guardDurableStream(guardedOnce, function* () {}); + + function* workflow(): Workflow { + yield* workspaceStep(database, "write", function* (filesystem) { + yield* filesystem.writeFile("/kept.txt", "atomic bytes", 0o640); + return "written"; + }); + return "done"; + } + + expect( + yield* withWorkspaceEffects(database, durableRun(workflow, { stream: guarded })), + ).toBe("done"); + const committed = yield* inspectWorkspace(database, "/kept.txt"); + expect(committed.content).toBe("atomic bytes"); + expect(committed.root).not.toBe(baseline.root); + expect(journalRoot(path, "write")).toBe(committed.root); + expect(gateCalls).toBe(1); + + 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(), + events: yield* transaction.journal.readAll(), + }; + }, + ); + }); + if (!next.ok) { + throw next.error; + } + expect(next.value.content).toBe("atomic bytes"); + expect(next.value.root).toBe(committed.root); + expect(workspaceYields(next.value.events)).toHaveLength(1); + expect(observedRoutedAppend).toBe(true); + }, + { + // deno-lint-ignore require-yield + *afterRoutedJournalAppend(candidate, event): Operation { + if (candidate !== selected || event.type !== "yield") { + return; + } + observedRoutedAppend = true; + const observer = new DatabaseSync(path); + try { + expect(committedEventCount(path)).toBe(0); + expect( + observer.prepare("SELECT current_root_id FROM workspace_state").get()?.[ + "current_root_id" + ], + ).toBe(baselineRoot); + expect( + observer + .prepare("SELECT COUNT(*) AS count FROM vfs_dirents WHERE name = 'kept.txt'") + .get()?.["count"], + ).toBe(0); + } finally { + observer.close(); + } + }, + }, + ); + }); + + it("WAC2: supported topology mutations pass through the provider proof operation", function* () { + const root = yield* useStorageRoot(); + + yield* withStorage(root, function* () { + const database = yield* createRun({ runId: "atomic-topology" }); + yield* setPrivateWorkspaceClock(database, () => 1_750_000_001_000); + + function* workflow(): Workflow { + yield* workspaceStep(database, "topology", function* (filesystem) { + yield* filesystem.mkdir("/nested", { mode: 0o750 }); + yield* filesystem.writeFile("/nested/file.txt", "first"); + yield* filesystem.writeFile("/nested/file.txt", "overwritten"); + yield* filesystem.writeFile("/discard.txt", "discard"); + yield* filesystem.remove("/discard.txt"); + yield* filesystem.rename("/nested/file.txt", "/nested/renamed.txt"); + yield* filesystem.chmod("/nested/renamed.txt", 0o600); + yield* filesystem.symlink("/nested/renamed.txt", "/current.txt"); + yield* filesystem.link("/nested/renamed.txt", "/hardlink.txt"); + return null; + }); + } + + yield* withWorkspaceEffects(database, durableRun(workflow, { stream: database.journal })); + const inspected = yield* database.transact(function* (transaction) { + return yield* withPrivateWorkspaceTransaction(database, transaction, function* (workspace) { + return { + renamed: yield* workspace.filesystem.readTextFile("/nested/renamed.txt"), + hardlink: yield* workspace.filesystem.readTextFile("/hardlink.txt"), + target: yield* workspace.filesystem.readlink("/current.txt"), + mode: (yield* workspace.filesystem.stat("/nested/renamed.txt")).mode, + discarded: yield* raised(workspace.filesystem.stat("/discard.txt")), + }; + }); + }); + if (!inspected.ok) { + throw inspected.error; + } + expect(inspected.value).toEqual({ + renamed: "overwritten", + hardlink: "overwritten", + target: "/nested/renamed.txt", + mode: 0o600, + discarded: expect.any(Error), + }); + }); + }); + + it("WAC3: a known filesystem failure commits one failed Yield on the prior root", function* () { + const root = yield* useStorageRoot(); + const runId = "atomic-known-failure"; + + yield* withStorage(root, function* () { + const database = yield* createRun({ runId }); + const baseline = yield* inspectWorkspace(database, "/temporary.txt"); + let caught: unknown; + function* workflow(): Workflow { + try { + yield* workspaceStep(database, "known", function* (filesystem) { + yield* filesystem.writeFile("/temporary.txt", "rolled back"); + yield* filesystem.readTextFile("/missing.txt"); + return null; + }); + } catch (error) { + caught = error; + } + } + + yield* withWorkspaceEffects(database, durableRun(workflow, { stream: database.journal })); + expect(caught).toBeInstanceOf(Error); + if (!(caught instanceof Error)) { + throw new Error("the failed Workspace result did not restore an Error"); + } + expect(caught.name).toBe("WorkspaceFsError"); + const after = yield* inspectWorkspace(database, "/temporary.txt"); + expect(after).toEqual(baseline); + const events = workspaceYields(yield* database.journal.readAll()); + expect(events).toHaveLength(1); + expect(events[0]?.result.status).toBe("err"); + expect(journalRoot(runPath(root, runId), "known")).toBe(baseline.root); + }); + }); + + it("WAC4: journal insertion failure rolls back mutation, root and pointer", function* () { + const root = yield* useStorageRoot(); + const runId = "atomic-insert-failure"; + + yield* withStorage(root, function* () { + const database = yield* createRun({ runId }); + const baseline = yield* inspectWorkspace(database, "/uncommitted.txt"); + const path = runPath(root, runId); + refuseJournalInsertNamed(path, "refused"); + let caught: unknown; + let laterExecutions = 0; + function* workflow(): Workflow { + try { + yield* workspaceStep(database, "refused", function* (filesystem) { + yield* filesystem.writeFile("/uncommitted.txt", "must roll back"); + return null; + }); + } catch (error) { + caught = error; + } + yield* durableCall("fenced-after-insert", function* () { + laterExecutions += 1; + return null; + }); + } + + const failure = yield* raised( + withWorkspaceEffects(database, durableRun(workflow, { stream: database.journal })), + ); + expect(failure).toBeInstanceOf(DurablePersistenceError); + expect(failure).toBe(caught); + expect(laterExecutions).toBe(0); + if (!(failure instanceof DurablePersistenceError)) { + throw new Error("the journal refusal did not activate durable persistence failure"); + } + expect(failure.cause).toBeInstanceOf(Error); + allowJournalInserts(path); + expect(yield* inspectWorkspace(database, "/uncommitted.txt")).toEqual(baseline); + expect(yield* database.journal.readAll()).toEqual([]); + }); + }); + + it("WAC5: secret-gate rejection precedes insertion and rolls back everything", function* () { + const root = yield* useStorageRoot(); + const gateFailure = new Error("secret gate rejected Workspace output"); + + yield* withStorage(root, function* () { + const database = yield* createRun({ runId: "atomic-secret" }); + const baseline = yield* inspectWorkspace(database, "/secret.txt"); + let gateCalls = 0; + const guarded = guardDurableStream(database.journal, function* (event) { + if (event.type === "yield") { + gateCalls += 1; + throw gateFailure; + } + }); + function* workflow(): Workflow { + yield* workspaceStep(database, "secret", function* (filesystem) { + yield* filesystem.writeFile("/secret.txt", "filtered bytes"); + return null; + }); + } + + const failure = yield* raised( + withWorkspaceEffects(database, durableRun(workflow, { stream: guarded })), + ); + expect(failure).toBe(gateFailure); + expect(gateCalls).toBe(1); + expect(yield* inspectWorkspace(database, "/secret.txt")).toEqual(baseline); + expect(yield* database.journal.readAll()).toEqual([]); + }); + }); + + it("WAC6: cancellation before, during and during teardown publishes nothing", function* () { + const root = yield* useStorageRoot(); + + yield* withStorage(root, function* () { + for (const point of ["before", "during", "teardown"] as const) { + yield* scoped(function* () { + const database = yield* createRun({ runId: `atomic-cancel-${point}` }); + const path = `/${point}.txt`; + const baseline = yield* inspectWorkspace(database, path); + const reached = withResolvers(); + const release = withResolvers(); + + function* workflow(): Workflow { + yield* workspaceStep(database, `cancel-${point}`, function* (filesystem) { + if (point === "before") { + reached.resolve(); + yield* suspend(); + } + yield* filesystem.writeFile(path, "cancelled"); + if (point === "during") { + reached.resolve(); + yield* suspend(); + } + if (point === "teardown") { + const childReady = withResolvers(); + yield* spawn(function* () { + yield* ensure(function* () { + reached.resolve(); + yield* release.operation; + }); + childReady.resolve(); + yield* suspend(); + }); + yield* childReady.operation; + } + return null; + }); + } + + const task = yield* spawn(() => + withWorkspaceEffects(database, durableRun(workflow, { stream: database.journal })), + ); + yield* reached.operation; + const halting = yield* spawn(() => task.halt()); + if (point === "teardown") { + release.resolve(); + } + yield* halting; + expect(yield* inspectWorkspace(database, path)).toEqual(baseline); + expect(yield* database.journal.readAll()).toEqual([]); + }); + } + }); + }); + + it("WAC7: infrastructure failure poisons later Workspace and ordinary effects", function* () { + const root = yield* useStorageRoot(); + + yield* withStorage(root, function* () { + const runId = "atomic-poison"; + const database = yield* createRun({ runId }); + const baseline = yield* inspectWorkspace(database, "/poisoned.txt"); + const path = runPath(root, runId); + let caught: unknown; + let laterWorkspace = 0; + let laterOrdinary = 0; + tamper(path, (sqlite) => { + sqlite.exec(` + CREATE TRIGGER refuse_root_capture BEFORE INSERT ON workspace_roots + BEGIN + SELECT raise(ABORT, 'root capture refused'); + END + `); + }); + + function* workflow(): Workflow { + try { + yield* workspaceStep(database, "poison", function* (filesystem) { + yield* filesystem.writeFile("/poisoned.txt", "must roll back"); + return null; + }); + } catch (error) { + caught = error; + } + yield* workspaceStep(database, "later-workspace", function* () { + laterWorkspace += 1; + return null; + }); + yield* durableCall("later-ordinary", function* () { + laterOrdinary += 1; + return null; + }); + } + + const escaped = yield* raised( + withWorkspaceEffects(database, durableRun(workflow, { stream: database.journal })), + ); + expect(escaped).toBe(caught); + expect(escaped).toBeInstanceOf(Error); + expect({ laterWorkspace, laterOrdinary }).toEqual({ laterWorkspace: 0, laterOrdinary: 0 }); + tamper(path, (sqlite) => sqlite.exec("DROP TRIGGER refuse_root_capture")); + expect(yield* inspectWorkspace(database, "/poisoned.txt")).toEqual(baseline); + expect(yield* database.journal.readAll()).toEqual([]); + }); + }); + + it("WAC8: unrelated same-run work waits, while another run remains usable", function* () { + const root = yield* useStorageRoot(); + + yield* withStorage(root, function* () { + const first = yield* createRun({ runId: "atomic-concurrent-first" }); + const sameRun = yield* createRun({ runId: "atomic-concurrent-first" }); + const second = yield* createRun({ runId: "atomic-concurrent-second" }); + const mutationStarted = withResolvers(); + const releaseMutation = withResolvers(); + let unrelatedFinished = false; + const unrelated: Yield = { + type: "yield", + coroutineId: "unrelated", + description: { type: "call", name: "unrelated" }, + result: { status: "ok", value: null }, + }; + + function* workflow(): Workflow { + yield* workspaceStep(first, "concurrent", function* (filesystem) { + yield* filesystem.writeFile("/first.txt", "first"); + yield* second.journal.append({ + ...unrelated, + coroutineId: "second-run", + }); + mutationStarted.resolve(); + yield* releaseMutation.operation; + return null; + }); + } + + const effect = yield* spawn(() => + withWorkspaceEffects(first, durableRun(workflow, { stream: first.journal })), + ); + yield* mutationStarted.operation; + const append = yield* spawn(function* () { + yield* sameRun.journal.append(unrelated); + unrelatedFinished = true; + }); + expect(unrelatedFinished).toBe(false); + expect(yield* second.journal.readAll()).toHaveLength(1); + releaseMutation.resolve(); + yield* effect; + yield* append; + expect(unrelatedFinished).toBe(true); + expect(workspaceYields(yield* first.journal.readAll())).toHaveLength(1); + }); + }); + + it("WAC9: replay bypasses the Deno coordinator and mutation", function* () { + const root = yield* useStorageRoot(); + + yield* withStorage(root, function* () { + const database = yield* createRun({ runId: "atomic-replay" }); + yield* database.journal.append({ + type: "yield", + coroutineId: "root", + description: { type: "workspace-proof", name: "replayed" }, + result: { status: "ok", value: "retained" }, + }); + let mutations = 0; + function* workflow(): Workflow { + yield* workspaceStep(database, "replayed", function* () { + mutations += 1; + return "live"; + }); + return "done"; + } + + expect( + yield* withWorkspaceEffects(database, durableRun(workflow, { stream: database.journal })), + ).toBe("done"); + expect(mutations).toBe(0); + }); + }); + + it("WAC10: closed and foreign database authority reaches no mutation", function* () { + const root = yield* useStorageRoot(); + let closed: WorkflowRunDatabase | undefined; + const savepoints: string[] = []; + yield* SavepointObservation.set((event) => { + if (event.kind === "create") { + savepoints.push(event.name); + } + }); + + yield* withStorage(root, function* () { + const first = yield* createRun({ runId: "atomic-authority-first" }); + const second = yield* createRun({ runId: "atomic-authority-second" }); + closed = first; + let foreignMutations = 0; + function* foreignWorkflow(): Workflow { + yield* workspaceStep(second, "foreign", function* () { + foreignMutations += 1; + return null; + }); + } + savepoints.length = 0; + expect( + yield* raised( + withWorkspaceEffects(first, durableRun(foreignWorkflow, { stream: first.journal })), + ), + ).toBeInstanceOf(Error); + expect(foreignMutations).toBe(0); + expect(savepoints).toEqual([]); + + let missingMutations = 0; + function* missingWorkflow(): Workflow { + yield createDurableWorkspaceOperation( + { type: "workspace-proof", name: "missing-authority" }, + function* () { + missingMutations += 1; + return null; + }, + ); + } + expect( + yield* raised( + withWorkspaceEffects(first, durableRun(missingWorkflow, { stream: first.journal })), + ), + ).toBeInstanceOf(Error); + expect(missingMutations).toBe(0); + expect(savepoints).toEqual([]); + + let forgedMutations = 0; + function* forgedExecute(): Operation { + forgedMutations += 1; + return null; + } + Object.defineProperty( + forgedExecute, + Symbol.for("executablemd.workflow.deno.workspace.effect.owner"), + { value: first }, + ); + function* forgedWorkflow(): Workflow { + yield createDurableWorkspaceOperation( + { type: "workspace-proof", name: "forged-authority" }, + forgedExecute, + ); + } + expect( + yield* raised( + withWorkspaceEffects(first, durableRun(forgedWorkflow, { stream: first.journal })), + ), + ).toBeInstanceOf(Error); + expect(forgedMutations).toBe(0); + expect(savepoints).toEqual([]); + expect(yield* first.journal.readAll()).toEqual([]); + expect(yield* second.journal.readAll()).toEqual([]); + }); + if (closed === undefined) { + throw new Error("the closed database proof did not retain its handle"); + } + const stale = closed; + yield* withStorage(root, function* () { + const current = yield* createRun({ runId: "atomic-authority-first" }); + let mutations = 0; + function* workflow(): Workflow { + yield* workspaceStep(stale, "closed", function* () { + mutations += 1; + return null; + }); + } + savepoints.length = 0; + expect( + yield* raised(withWorkspaceEffects(stale, durableRun(workflow, { stream: stale.journal }))), + ).toBeInstanceOf(Error); + expect(mutations).toBe(0); + expect(savepoints).toEqual([]); + expect(yield* current.journal.readAll()).toEqual([]); + }); + }); + + it("WAC11: foreign publication destinations are refused before mutation", function* () { + const root = yield* useStorageRoot(); + + yield* withStorage(root, function* () { + const selected = yield* createRun({ runId: "atomic-destination-selected" }); + const other = yield* createRun({ runId: "atomic-destination-other" }); + const baseline = yield* inspectWorkspace(selected, "/wrong-destination.txt"); + const copied: DurableStream = { + readAll: () => selected.journal.readAll(), + append: (event) => selected.journal.append(event), + }; + for (const key of Reflect.ownKeys(selected.journal)) { + const descriptor = Object.getOwnPropertyDescriptor(selected.journal, key); + if (descriptor !== undefined) { + Object.defineProperty(copied, key, descriptor); + } + } + Object.defineProperty(copied, Symbol.for("executablemd.durable-stream.inherit-provenance"), { + value: () => undefined, + }); + + for (const stream of [new InMemoryStream(), other.journal, copied]) { + let mutations = 0; + let caught: unknown; + let laterExecutions = 0; + function* workflow(): Workflow { + try { + yield* workspaceStep(selected, "wrong-destination", function* (filesystem) { + mutations += 1; + yield* filesystem.writeFile("/wrong-destination.txt", "must not run"); + return null; + }); + } catch (error) { + caught = error; + } + yield* durableCall("fenced-after-wrong-destination", function* () { + laterExecutions += 1; + return null; + }); + } + + const failure = yield* raised( + withWorkspaceEffects(selected, durableRun(workflow, { stream })), + ); + expect(failure).toBe(caught); + expect(failure).toBeInstanceOf(Error); + expect(mutations).toBe(0); + expect(laterExecutions).toBe(0); + expect(yield* inspectWorkspace(selected, "/wrong-destination.txt")).toEqual(baseline); + expect(yield* selected.journal.readAll()).toEqual([]); + expect(yield* other.journal.readAll()).toEqual([]); + if (stream instanceof InMemoryStream) { + expect(stream.snapshot()).toEqual([]); + } + } + }); + }); + + it("WAC12: failure after routed publication rolls the whole transaction back", function* () { + const root = yield* useStorageRoot(); + const commitFailure = new Error("transaction owner refused commit"); + const runId = "atomic-post-publication"; + const path = join(root, `${runId}.sqlite`); + let selected: WorkflowRunDatabase | undefined; + let armed = false; + + yield* withDirectWorkspaceStorage( + root, + runId, + () => {}, + function* (database) { + selected = database; + const baseline = yield* inspectWorkspace(database, "/post-publication.txt"); + const roots = retainedRootCount(path); + let gateCalls = 0; + let caught: unknown; + let laterExecutions = 0; + const guarded = guardDurableStream(database.journal, function* (event) { + if (event.type === "yield") { + gateCalls += 1; + } + }); + function* workflow(): Workflow { + try { + yield* workspaceStep(database, "post-publication", function* (filesystem) { + yield* filesystem.writeFile("/post-publication.txt", "rolled back"); + return null; + }); + } catch (error) { + caught = error; + } + yield* durableCall("fenced-after-commit", function* () { + laterExecutions += 1; + return null; + }); + } + + armed = true; + const failure = yield* raised( + withWorkspaceEffects(database, durableRun(workflow, { stream: guarded })), + ); + armed = false; + expect(failure).toBe(commitFailure); + expect(caught).toBe(commitFailure); + expect(gateCalls).toBe(1); + expect(laterExecutions).toBe(0); + expect(yield* inspectWorkspace(database, "/post-publication.txt")).toEqual(baseline); + expect(retainedRootCount(path)).toBe(roots); + expect(yield* database.journal.readAll()).toEqual([]); + }, + { + // deno-lint-ignore require-yield + *beforeCommit(candidate): Operation { + if (armed && candidate === selected) { + throw commitFailure; + } + }, + }, + ); + }); + + it("WAC13: cancellation before commit publishes no protocol event", function* () { + const root = yield* useStorageRoot(); + const runId = "atomic-cancel-before-commit"; + const path = join(root, `${runId}.sqlite`); + const reachedPublication = withResolvers(); + let selected: WorkflowRunDatabase | undefined; + + yield* withDirectWorkspaceStorage( + root, + runId, + () => {}, + function* (database) { + selected = database; + const baseline = yield* inspectWorkspace(database, "/cancel-before-commit.txt"); + const roots = retainedRootCount(path); + let gateCalls = 0; + const guarded = guardDurableStream(database.journal, function* (event) { + if (event.type === "yield") { + gateCalls += 1; + } + }); + function* workflow(): Workflow { + yield* workspaceStep(database, "cancel-before-commit", function* (filesystem) { + yield* filesystem.writeFile("/cancel-before-commit.txt", "rolled back"); + return null; + }); + } + + const task = yield* spawn(() => + withWorkspaceEffects(database, durableRun(workflow, { stream: guarded })), + ); + yield* reachedPublication.operation; + yield* task.halt(); + expect(gateCalls).toBe(1); + expect(yield* inspectWorkspace(database, "/cancel-before-commit.txt")).toEqual(baseline); + expect(retainedRootCount(path)).toBe(roots); + expect(yield* database.journal.readAll()).toEqual([]); + }, + { + *afterRoutedJournalAppend(candidate, event): Operation { + if (candidate === selected && event.type === "yield") { + reachedPublication.resolve(); + yield* suspend(); + } + }, + }, + ); + }); + + it("WAC14: cancellation leaves no DOFS continuation beyond savepoint teardown", function* () { + const source = yield* readTextFile( + new URL("../src/deno/workspace/filesystem.ts", import.meta.url), + ); + expect(source.includes("writeFileSync")).toBe(true); + expect(source.includes("readRangeSync")).toBe(true); + for (const forbidden of [ + "until(", + "Response", + "ReadableStream", + "WorkspaceFilesystemOperations", + ]) { + expect(source.includes(forbidden)).toBe(false); + } + + const root = yield* useStorageRoot(); + let activeOrder: string[] | undefined; + yield* SavepointObservation.set((event) => { + if (event.kind === "rollback") { + activeOrder?.push("savepoint-rollback"); + } + }); + + yield* withStorage(root, function* () { + for (const phase of ["before", "after"] as const) { + yield* scoped(function* () { + const runId = `atomic-synchronous-${phase}`; + const database = yield* createRun({ runId }); + const baseline = yield* inspectWorkspace(database, "/pending.txt"); + const reached = withResolvers(); + const order: string[] = []; + activeOrder = order; + + function* workflow(): Workflow { + yield* workspaceStep(database, `synchronous-${phase}`, function* (filesystem) { + yield* ensure(function* () { + order.push("mutation-teardown"); + }); + if (phase === "before") { + order.push("before-call"); + reached.resolve(); + yield* suspend(); + } + yield* filesystem.writeFile("/pending.txt", "cancelled bytes"); + order.push("after-call"); + reached.resolve(); + yield* suspend(); + return null; + }); + } + const task = yield* spawn(() => + withWorkspaceEffects(database, durableRun(workflow, { stream: database.journal })), + ); + yield* reached.operation; + yield* task.halt(); + activeOrder = undefined; + expect(order[0]).toBe(`${phase}-call`); + expect(order.indexOf("mutation-teardown")).toBeGreaterThan(0); + expect(order.indexOf("savepoint-rollback")).toBeGreaterThan( + order.indexOf("mutation-teardown"), + ); + expect(yield* inspectWorkspace(database, "/pending.txt")).toEqual(baseline); + expect(yield* database.journal.readAll()).toEqual([]); + }); + } + }); + }); + + it("WAC15: mutation-child teardown failure is infrastructure failure", function* () { + const root = yield* useStorageRoot(); + const teardownFailure = new Error("mutation child teardown failed"); + + yield* withStorage(root, function* () { + const database = yield* createRun({ runId: "atomic-teardown-failure" }); + const baseline = yield* inspectWorkspace(database, "/teardown.txt"); + let caught: unknown; + let laterExecutions = 0; + function* workflow(): Workflow { + try { + yield* workspaceStep(database, "teardown-failure", function* (filesystem) { + const ready = withResolvers(); + yield* spawn(function* () { + yield* ensure(function* () { + throw teardownFailure; + }); + ready.resolve(); + yield* suspend(); + }); + yield* ready.operation; + yield* filesystem.writeFile("/teardown.txt", "rolled back"); + return null; + }); + } catch (error) { + caught = error; + } + yield* durableCall("fenced-after-teardown", function* () { + laterExecutions += 1; + return null; + }); + } + + const failure = yield* raised( + withWorkspaceEffects(database, durableRun(workflow, { stream: database.journal })), + ); + expect(failure).toBe(caught); + expect(failure).toBe(teardownFailure); + expect(laterExecutions).toBe(0); + expect(yield* inspectWorkspace(database, "/teardown.txt")).toEqual(baseline); + expect(yield* database.journal.readAll()).toEqual([]); + }); + }); + + it("WAC16: current-root publication failure rolls retained state back", function* () { + const root = yield* useStorageRoot(); + + yield* withStorage(root, function* () { + const runId = "atomic-current-root-failure"; + const database = yield* createRun({ runId }); + const path = runPath(root, runId); + const baseline = yield* inspectWorkspace(database, "/current-root.txt"); + const roots = retainedRootCount(path); + let caught: unknown; + let laterExecutions = 0; + tamper(path, (sqlite) => { + sqlite.exec(` + CREATE TRIGGER refuse_current_root BEFORE UPDATE OF current_root_id ON workspace_state + BEGIN + SELECT raise(ABORT, 'current root publication refused'); + END + `); + }); + function* workflow(): Workflow { + try { + yield* workspaceStep(database, "current-root-failure", function* (filesystem) { + yield* filesystem.writeFile("/current-root.txt", "rolled back"); + return null; + }); + } catch (error) { + caught = error; + } + yield* durableCall("fenced-after-current-root", function* () { + laterExecutions += 1; + return null; + }); + } + + const failure = yield* raised( + withWorkspaceEffects(database, durableRun(workflow, { stream: database.journal })), + ); + expect(failure).toBe(caught); + expect(failure).toBeInstanceOf(Error); + expect(laterExecutions).toBe(0); + tamper(path, (sqlite) => sqlite.exec("DROP TRIGGER refuse_current_root")); + expect(yield* inspectWorkspace(database, "/current-root.txt")).toEqual(baseline); + expect(retainedRootCount(path)).toBe(roots); + expect(yield* database.journal.readAll()).toEqual([]); + }); + }); + + it("WAC17: an existing durability failure wins before Workspace coordination", function* () { + const root = yield* useStorageRoot(); + + yield* withStorage(root, function* () { + const runId = "atomic-existing-failure"; + const database = yield* createRun({ runId }); + const path = runPath(root, runId); + const baseline = yield* inspectWorkspace(database, "/never-runs.txt"); + refuseJournalInsertNamed(path, "first-failure"); + let caught: unknown; + let workspaceExecutions = 0; + function* workflow(): Workflow { + try { + yield* durableCall("first-failure", function* () { + return null; + }); + } catch (error) { + caught = error; + } + yield* workspaceStep(database, "blocked-workspace", function* (filesystem) { + workspaceExecutions += 1; + yield* filesystem.writeFile("/never-runs.txt", "blocked"); + return null; + }); + } + + const failure = yield* raised( + withWorkspaceEffects(database, durableRun(workflow, { stream: database.journal })), + ); + allowJournalInserts(path); + expect(failure).toBe(caught); + expect(failure).toBeInstanceOf(DurablePersistenceError); + expect(workspaceExecutions).toBe(0); + expect(yield* inspectWorkspace(database, "/never-runs.txt")).toEqual(baseline); + expect(yield* database.journal.readAll()).toEqual([]); + }); + }); + + it("WAC18: savepoint SQL failure poisons the coordinated outer transaction", function* () { + const root = yield* useStorageRoot(); + + for (const phase of ["create", "release", "rollback"] as const) { + let armed = false; + let operationName: string | undefined; + const observed: Array<{ kind: string; name: string }> = []; + const observe: SavepointObserver = (event) => { + observed.push(event); + if (armed && operationName === undefined && event.kind === "create") { + operationName = event.name; + } + }; + + yield* withDirectWorkspaceStorage( + root, + `atomic-savepoint-${phase}`, + observe, + function* (database, connection) { + const baseline = yield* inspectWorkspace(database, "/savepoint.txt"); + const roots = retainedRootCount(connection.path); + let caught: unknown; + let laterExecutions = 0; + observed.length = 0; + armed = true; + setSqliteAuthorizer(connection.database, (action, operation, name) => { + if (!armed || action !== SQLITE_SAVEPOINT) { + return SQLITE_OK; + } + if (phase === "create" && operation === "BEGIN") { + return SQLITE_DENY; + } + if (phase === "release" && operation === "RELEASE" && name === operationName) { + return SQLITE_DENY; + } + if (phase === "rollback" && operation === "ROLLBACK" && name === operationName) { + return SQLITE_DENY; + } + return SQLITE_OK; + }); + + function* workflow(): Workflow { + try { + yield* workspaceStep(database, `savepoint-${phase}`, function* (filesystem) { + yield* filesystem.writeFile("/savepoint.txt", "rolled back"); + if (phase === "rollback") { + throw new Error("force operation savepoint rollback"); + } + return null; + }); + } catch (error) { + caught = error; + } + yield* durableCall(`fenced-after-savepoint-${phase}`, function* () { + laterExecutions += 1; + return null; + }); + } + + const failure = yield* raised( + withWorkspaceEffects(database, durableRun(workflow, { stream: database.journal })), + ); + armed = false; + setSqliteAuthorizer(connection.database, null); + expect(failure).toBe(caught); + expect(failure).toBeInstanceOf(Error); + expect(laterExecutions).toBe(0); + expect(yield* inspectWorkspace(database, "/savepoint.txt")).toEqual(baseline); + expect(retainedRootCount(connection.path)).toBe(roots); + expect(yield* database.journal.readAll()).toEqual([]); + if (phase === "create") { + expect(observed.some((event) => event.kind === "create")).toBe(false); + } else { + expect(operationName).toBeDefined(); + expect( + observed.some((event) => event.kind === phase && event.name === operationName), + ).toBe(false); + } + }, + ); + } + }); + + it("WAC19: enclosing coordination middleware cannot suppress publication", function* () { + const root = yield* useStorageRoot(); + + yield* withStorage(root, function* () { + const database = yield* createRun({ runId: "atomic-middleware-publication" }); + let selections = 0; + let replacementRuns = 0; + const replacingMiddleware = { + provider(_args: [], next: () => object | undefined): object | undefined { + selections += 1; + return next(); + }, + *run( + [execute, _publish, activateFailure, identity]: [ + () => Operation, + (result: Result) => Operation, + ActivateDurabilityFailure, + DurablePublicationIdentity | undefined, + ], + next: ( + execute: () => Operation, + publish: (result: Result) => Operation, + activateFailure: ActivateDurabilityFailure, + identity: DurablePublicationIdentity | undefined, + ) => Operation, + ): Operation { + replacementRuns += 1; + return yield* next( + execute, + // deno-lint-ignore require-yield + function* () {}, + activateFailure, + identity, + ); + }, + }; + function* workflow(): Workflow { + yield* workspaceStep(database, "middleware-publication", function* (filesystem) { + yield* filesystem.writeFile("/middleware.txt", "committed together"); + return null; + }); + } + + yield* scoped(function* () { + yield* WorkspaceCoordination.around(replacingMiddleware); + yield* withWorkspaceEffects(database, durableRun(workflow, { stream: database.journal })); + }); + + expect(selections).toBe(1); + expect(replacementRuns).toBe(0); + expect((yield* inspectWorkspace(database, "/middleware.txt")).content).toBe( + "committed together", + ); + expect(workspaceYields(yield* database.journal.readAll())).toHaveLength(1); + }); + }); + + it("WAC20: enclosing middleware cannot replace infrastructure failure activation", function* () { + const root = yield* useStorageRoot(); + + yield* withStorage(root, function* () { + const runId = "atomic-middleware-failure"; + const database = yield* createRun({ runId }); + const path = runPath(root, runId); + const baseline = yield* inspectWorkspace(database, "/middleware-failure.txt"); + const roots = retainedRootCount(path); + let replacementRuns = 0; + let caught: unknown; + let laterExecutions = 0; + const replacingMiddleware = { + provider(_args: [], next: () => object | undefined): object | undefined { + return next(); + }, + *run( + [execute, publish, _activateFailure, identity]: [ + () => Operation, + (result: Result) => Operation, + ActivateDurabilityFailure, + DurablePublicationIdentity | undefined, + ], + next: ( + execute: () => Operation, + publish: (result: Result) => Operation, + activateFailure: ActivateDurabilityFailure, + identity: DurablePublicationIdentity | undefined, + ) => Operation, + ): Operation { + replacementRuns += 1; + return yield* next(execute, publish, () => new Error("replacement failure"), identity); + }, + }; + tamper(path, (sqlite) => { + sqlite.exec(` + CREATE TRIGGER refuse_middleware_root BEFORE INSERT ON workspace_roots + BEGIN + SELECT raise(ABORT, 'middleware root capture refused'); + END + `); + }); + function* workflow(): Workflow { + try { + yield* workspaceStep(database, "middleware-failure", function* (filesystem) { + yield* filesystem.writeFile("/middleware-failure.txt", "must roll back"); + return null; + }); + } catch (error) { + caught = error; + } + yield* durableCall("fenced-after-middleware-failure", function* () { + laterExecutions += 1; + return null; + }); + } + + let failure: unknown; + yield* scoped(function* () { + yield* WorkspaceCoordination.around(replacingMiddleware); + failure = yield* raised( + withWorkspaceEffects(database, durableRun(workflow, { stream: database.journal })), + ); + }); + tamper(path, (sqlite) => sqlite.exec("DROP TRIGGER refuse_middleware_root")); + + expect(replacementRuns).toBe(0); + expect(failure).toBe(caught); + expect(failure).toBeInstanceOf(Error); + expect(laterExecutions).toBe(0); + expect(yield* inspectWorkspace(database, "/middleware-failure.txt")).toEqual(baseline); + expect(retainedRootCount(path)).toBe(roots); + expect(yield* database.journal.readAll()).toEqual([]); + }); + }); + + it("WAC21: substituted provider selection is refused before transaction work", function* () { + const root = yield* useStorageRoot(); + const savepoints: string[] = []; + yield* SavepointObservation.set((event) => { + if (event.kind === "create") { + savepoints.push(event.name); + } + }); + + yield* withStorage(root, function* () { + const runId = "atomic-substituted-provider"; + const database = yield* createRun({ runId }); + const path = runPath(root, runId); + const baseline = yield* inspectWorkspace(database, "/substituted.txt"); + const roots = retainedRootCount(path); + let mutations = 0; + function* workflow(): Workflow { + yield* workspaceStep(database, "substituted-provider", function* (filesystem) { + mutations += 1; + yield* filesystem.writeFile("/substituted.txt", "must not run"); + return null; + }); + } + + savepoints.length = 0; + let failure: unknown; + yield* scoped(function* () { + yield* WorkspaceCoordination.around({ provider: () => Object.freeze({}) }); + failure = yield* raised( + withWorkspaceEffects(database, durableRun(workflow, { stream: database.journal })), + ); + }); + + expect(failure).toBeInstanceOf(Error); + expect(mutations).toBe(0); + expect(savepoints).toEqual([]); + expect(yield* inspectWorkspace(database, "/substituted.txt")).toEqual(baseline); + expect(retainedRootCount(path)).toBe(roots); + expect(yield* database.journal.readAll()).toEqual([]); + }); + }); + + it("WAC22: minimum-priority middleware cannot split publication from commit", function* () { + const root = yield* useStorageRoot(); + + yield* withStorage(root, function* () { + const runId = "atomic-minimum-publication"; + const database = yield* createRun({ runId }); + const path = runPath(root, runId); + const baseline = yield* inspectWorkspace(database, "/minimum.txt"); + const roots = retainedRootCount(path); + const observed: unknown[] = []; + function* workflow(): Workflow { + yield* workspaceStep(database, "minimum-publication", function* (filesystem) { + yield* filesystem.writeFile("/minimum.txt", "committed with Yield"); + return null; + }); + } + + yield* scoped(function* () { + yield* WorkspaceInvocationCollision.around( + { + // deno-lint-ignore require-yield + *coordinate(args): Operation { + observed.push(args[0]); + return { type: "published" }; + }, + }, + { at: "min" }, + ); + yield* withWorkspaceEffects(database, durableRun(workflow, { stream: database.journal })); + }); + + const committed = yield* inspectWorkspace(database, "/minimum.txt"); + const events = yield* database.journal.readAll(); + expect(observed).toEqual([]); + expect(committed.content).toBe("committed with Yield"); + expect(committed.root).not.toBe(baseline.root); + expect(retainedRootCount(path)).toBe(roots + 1); + expect(workspaceYields(events)).toHaveLength(1); + expect(journalRoot(path, "minimum-publication")).toBe(committed.root); + }); + }); + + it("WAC23: minimum-priority middleware cannot replace failure activation", function* () { + const root = yield* useStorageRoot(); + + yield* withStorage(root, function* () { + const runId = "atomic-minimum-failure"; + const database = yield* createRun({ runId }); + const path = runPath(root, runId); + const baseline = yield* inspectWorkspace(database, "/minimum-failure.txt"); + const roots = retainedRootCount(path); + let collisions = 0; + let caught: unknown; + let laterExecutions = 0; + tamper(path, (sqlite) => { + sqlite.exec(` + CREATE TRIGGER refuse_minimum_root BEFORE INSERT ON workspace_roots + BEGIN + SELECT raise(ABORT, 'minimum root capture refused'); + END + `); + }); + function* workflow(): Workflow { + try { + yield* workspaceStep(database, "minimum-failure", function* (filesystem) { + yield* filesystem.writeFile("/minimum-failure.txt", "must roll back"); + return null; + }); + } catch (error) { + caught = error; + } + yield* durableCall("fenced-after-minimum-failure", function* () { + laterExecutions += 1; + return null; + }); + } + + let failure: unknown; + yield* scoped(function* () { + yield* WorkspaceInvocationCollision.around( + { + // deno-lint-ignore require-yield + *coordinate(): Operation { + collisions += 1; + return { type: "failure", failure: new Error("replacement failure") }; + }, + }, + { at: "min" }, + ); + failure = yield* raised( + withWorkspaceEffects(database, durableRun(workflow, { stream: database.journal })), + ); + }); + tamper(path, (sqlite) => sqlite.exec("DROP TRIGGER refuse_minimum_root")); + + expect(collisions).toBe(0); + expect(failure).toBe(caught); + expect(failure).toBeInstanceOf(Error); + expect(laterExecutions).toBe(0); + expect(yield* inspectWorkspace(database, "/minimum-failure.txt")).toEqual(baseline); + expect(retainedRootCount(path)).toBe(roots); + expect(yield* database.journal.readAll()).toEqual([]); + }); + }); +}); diff --git a/packages/workflow/tests/workspace-effect.test.ts b/packages/workflow/tests/workspace-effect.test.ts index 789af599..b2572266 100644 --- a/packages/workflow/tests/workspace-effect.test.ts +++ b/packages/workflow/tests/workspace-effect.test.ts @@ -1,8 +1,10 @@ import { describe, it } from "@executablemd/test-support/bdd"; import { expect } from "@executablemd/test-support/expect"; +import { createApi } from "@effectionx/context-api"; import { readTextFile } from "@effectionx/fs"; -import { type Operation } from "effection"; +import { type Operation, scoped } from "effection"; import { + claimDurablePublicationIdentity, durableCall, durableRun, InMemoryStream, @@ -11,11 +13,12 @@ import { type Result, type Workflow, } from "@executablemd/durable-streams"; +import { createDurableWorkspaceOperation, WorkspaceCoordinationProviderError } from "../mod.ts"; import { - createDurableWorkspaceOperation, - WorkspaceCoordination, - WorkspaceCoordinationProviderError, -} from "../mod.ts"; + type WorkspaceCoordinationAuthority, + type WorkspaceCoordinationProvider, + withWorkspaceCoordinationProvider, +} from "../src/workspace/effect.ts"; function* raised(operation: Operation): Operation { try { @@ -34,6 +37,42 @@ function* workspaceStep(name: string, execute: () => Operation): Workflow< yield createDurableWorkspaceOperation({ type: "workspace", name }, execute); } +interface InvocationCollisionApi { + coordinate(request: unknown): Operation; +} + +const WorkspaceInvocationCollision = createApi( + "executablemd.workflow.workspace.coordination.invocation", + { + // deno-lint-ignore require-yield + *coordinate(): Operation { + throw new Error("the collision handler did not delegate"); + }, + }, +); + +function successfulProvider(observe?: (authority: WorkspaceCoordinationAuthority) => void): { + provider: WorkspaceCoordinationProvider; + counts: { providers: number; executions: number; publications: number }; +} { + const counts = { providers: 0, executions: 0, publications: 0 }; + return { + counts, + provider: { + *run(authority: WorkspaceCoordinationAuthority): Operation { + counts.providers += 1; + observe?.(authority); + const value = yield* authority.execute(); + counts.executions += 1; + const result: Result = { status: "ok", value }; + yield* authority.publish(result); + counts.publications += 1; + return result; + }, + }, + }; +} + describe("Tier DLC — Workspace coordination selection", () => { it("DLC10: a missing Workspace provider fails before execution or publication", function* () { const stream = new InMemoryStream(); @@ -48,24 +87,25 @@ describe("Tier DLC — Workspace coordination selection", () => { const failure = yield* raised(durableRun(workflow, { stream })); expect(failure).toBeInstanceOf(WorkspaceCoordinationProviderError); expect(executions).toBe(0); - expect(yieldEvents(stream.snapshot())).toEqual([]); + expect(stream.snapshot()).toEqual([]); }); it("DLC11: explicit Workspace selection leaves unrelated durable operations ordinary", function* () { const stream = new InMemoryStream(); + const publicationIdentity = claimDurablePublicationIdentity(stream); const coordinated: string[] = []; const ordinary: string[] = []; - yield* WorkspaceCoordination.around({ - *run([execute, publish]: [ - () => Operation, - (result: Result) => Operation, - ]): Operation { + const provider: WorkspaceCoordinationProvider = { + *run(authority: WorkspaceCoordinationAuthority): Operation { + expect(authority.publicationIdentity).toBe(publicationIdentity); + expect(Reflect.get(authority.publicationIdentity ?? {}, "append")).toBe(undefined); + expect(Reflect.get(authority.publicationIdentity ?? {}, "readAll")).toBe(undefined); coordinated.push("workspace"); - const result: Result = { status: "ok", value: yield* execute() }; - yield* publish(result); + const result: Result = { status: "ok", value: yield* authority.execute() }; + yield* authority.publish(result); return result; }, - }); + }; function* workflow(): Workflow { yield* workspaceStep("selected", function* () { @@ -78,7 +118,9 @@ describe("Tier DLC — Workspace coordination selection", () => { return "done"; } - expect(yield* durableRun(workflow, { stream })).toBe("done"); + expect( + yield* withWorkspaceCoordinationProvider(provider, durableRun(workflow, { stream })), + ).toBe("done"); expect(coordinated).toEqual(["workspace"]); expect(ordinary).toEqual(["ordinary"]); expect(yieldEvents(stream.snapshot())).toHaveLength(2); @@ -128,4 +170,284 @@ describe("Tier DLC — Workspace coordination selection", () => { } } }); + + it("DLC15: live Workspace invocation authority is one-shot", function* () { + const stream = new InMemoryStream(); + claimDurablePublicationIdentity(stream); + let capturedAuthority: WorkspaceCoordinationAuthority | undefined; + let executions = 0; + const provider: WorkspaceCoordinationProvider = { + *run(authority: WorkspaceCoordinationAuthority): Operation { + capturedAuthority = authority; + const result: Result = { status: "ok", value: yield* authority.execute() }; + yield* authority.publish(result); + return result; + }, + }; + function* workflow(): Workflow { + yield* workspaceStep("one-shot", function* () { + executions += 1; + return null; + }); + } + + yield* withWorkspaceCoordinationProvider(provider, durableRun(workflow, { stream })); + if (capturedAuthority === undefined) { + throw new Error("the provider did not receive its live invocation authority"); + } + expect(yield* raised(capturedAuthority.execute())).toBeInstanceOf( + WorkspaceCoordinationProviderError, + ); + expect( + yield* raised(capturedAuthority.publish({ status: "ok", value: "late" })), + ).toBeInstanceOf(WorkspaceCoordinationProviderError); + const activationFailure = yield* raised( + capturedAuthority.activateFailure(new Error("late activation")), + ); + expect(activationFailure).toBeInstanceOf(WorkspaceCoordinationProviderError); + expect(executions).toBe(1); + expect(yieldEvents(stream.snapshot())).toHaveLength(1); + }); + + it("DLC17: a forged contextual result cannot complete or resume a live invocation", function* () { + const stream = new InMemoryStream(); + claimDurablePublicationIdentity(stream); + const { provider, counts } = successfulProvider(); + let laterExecutions = 0; + function* workflow(): Workflow { + try { + yield* workspaceStep("forged-result", function* () { + return "not reached"; + }); + } catch { + // The durable fail-stop boundary, rather than workflow recovery, decides termination. + } + yield* durableCall("after-forgery", function* () { + laterExecutions += 1; + return null; + }); + } + + const failure = yield* scoped(function* () { + yield* WorkspaceInvocationCollision.around({ + // deno-lint-ignore require-yield + *coordinate(): Operation { + return { type: "result", result: { status: "ok", value: "forged" } }; + }, + }); + return yield* raised( + withWorkspaceCoordinationProvider(provider, durableRun(workflow, { stream })), + ); + }); + + expect(failure).toBeInstanceOf(WorkspaceCoordinationProviderError); + expect(counts).toEqual({ providers: 0, executions: 0, publications: 0 }); + expect(laterExecutions).toBe(0); + expect(stream.snapshot()).toEqual([]); + }); + + it("DLC18: invocation phases are unreachable without a selected provider", function* () { + const stream = new InMemoryStream(); + claimDurablePublicationIdentity(stream); + let collisionCalls = 0; + let executions = 0; + function* workflow(): Workflow { + yield* workspaceStep("direct-phases", function* () { + executions += 1; + return "not reached"; + }); + } + + const failure = yield* scoped(function* () { + yield* WorkspaceInvocationCollision.around({ + *coordinate(args, next): Operation { + collisionCalls += 1; + return yield* next(...args); + }, + }); + return yield* raised(durableRun(workflow, { stream })); + }); + + expect(failure).toBeInstanceOf(WorkspaceCoordinationProviderError); + expect(collisionCalls).toBe(0); + expect(executions).toBe(0); + expect(stream.snapshot()).toEqual([]); + }); + + it("DLC19: contextual middleware cannot replace the authoritative published Result", function* () { + const stream = new InMemoryStream(); + claimDurablePublicationIdentity(stream); + const { provider, counts } = successfulProvider(); + function* workflow(): Workflow { + const result = yield createDurableWorkspaceOperation( + { type: "workspace", name: "replace-result" }, + function* () { + return "authoritative"; + }, + ); + if (typeof result !== "string") { + throw new Error("the Workspace operation did not return its string result"); + } + return result; + } + + const value = yield* scoped(function* () { + yield* WorkspaceInvocationCollision.around({ + *coordinate(args, next): Operation { + yield* next(...args); + return { type: "result", result: { status: "ok", value: "forged" } }; + }, + }); + return yield* withWorkspaceCoordinationProvider(provider, durableRun(workflow, { stream })); + }); + + expect(value).toBe("authoritative"); + expect(counts).toEqual({ providers: 1, executions: 1, publications: 1 }); + expect(yieldEvents(stream.snapshot())).toEqual([ + expect.objectContaining({ result: { status: "ok", value: "authoritative" } }), + ]); + }); + + it("DLC20: post-completion middleware cannot suppress, throw, or duplicate work", function* () { + for (const behavior of ["suppress", "throw", "duplicate"]) { + const stream = new InMemoryStream(); + claimDurablePublicationIdentity(stream); + const { provider, counts } = successfulProvider(); + function* workflow(): Workflow { + const result = yield createDurableWorkspaceOperation( + { type: "workspace", name: behavior }, + function* () { + return behavior; + }, + ); + if (typeof result !== "string") { + throw new Error("the Workspace operation did not return its string result"); + } + return result; + } + + const value = yield* scoped(function* () { + yield* WorkspaceInvocationCollision.around({ + *coordinate(args, next): Operation { + const response = yield* next(...args); + if (behavior === "throw") { + throw new Error("post-completion middleware failure"); + } + if (behavior === "duplicate") { + yield* next(...args); + } + return behavior === "suppress" ? { type: "published" } : response; + }, + }); + return yield* withWorkspaceCoordinationProvider(provider, durableRun(workflow, { stream })); + }); + + expect(value).toBe(behavior); + expect(counts).toEqual({ providers: 1, executions: 1, publications: 1 }); + expect(yieldEvents(stream.snapshot())).toHaveLength(1); + } + }); + + it("DLC21: retained contextual continuation cannot reuse a completed invocation", function* () { + const stream = new InMemoryStream(); + claimDurablePublicationIdentity(stream); + const { provider, counts } = successfulProvider(); + let retained: ((request: unknown) => Operation) | undefined; + let retainedRequest: unknown; + function* workflow(): Workflow { + yield* workspaceStep("retained-continuation", function* () { + return null; + }); + } + + yield* scoped(function* () { + yield* WorkspaceInvocationCollision.around({ + *coordinate(args, next): Operation { + retained = (request) => next(request); + retainedRequest = args[0]; + return yield* next(...args); + }, + }); + yield* withWorkspaceCoordinationProvider(provider, durableRun(workflow, { stream })); + }); + if (retained === undefined) { + throw new Error("the collision middleware did not retain its continuation"); + } + + expect(yield* raised(retained(retainedRequest))).toBeInstanceOf( + WorkspaceCoordinationProviderError, + ); + expect(counts).toEqual({ providers: 1, executions: 1, publications: 1 }); + expect(yieldEvents(stream.snapshot())).toHaveLength(1); + }); + + it("DLC22: minimum-priority collision middleware receives no invocation capability", function* () { + const stream = new InMemoryStream(); + claimDurablePublicationIdentity(stream); + const { provider, counts } = successfulProvider(); + const observed: unknown[] = []; + function* workflow(): Workflow { + yield* workspaceStep("minimum-priority", function* () { + return "published"; + }); + } + + yield* scoped(function* () { + yield* WorkspaceInvocationCollision.around( + { + // deno-lint-ignore require-yield + *coordinate(args): Operation { + observed.push(args[0]); + return { type: "published" }; + }, + }, + { at: "min" }, + ); + yield* withWorkspaceCoordinationProvider(provider, durableRun(workflow, { stream })); + }); + + expect(observed).toEqual([]); + expect(counts).toEqual({ providers: 1, executions: 1, publications: 1 }); + expect(yieldEvents(stream.snapshot())).toHaveLength(1); + }); + + it("DLC23: minimum-priority middleware cannot replace first-failure activation", function* () { + const stream = new InMemoryStream(); + claimDurablePublicationIdentity(stream); + const first = new Error("authoritative infrastructure failure"); + let activated: Error | undefined; + let collisions = 0; + const provider: WorkspaceCoordinationProvider = { + *run(authority: WorkspaceCoordinationAuthority): Operation { + activated = yield* authority.activateFailure(first); + throw activated; + }, + }; + function* workflow(): Workflow { + yield* workspaceStep("minimum-failure", function* () { + return "not reached"; + }); + } + + const failure = yield* scoped(function* () { + yield* WorkspaceInvocationCollision.around( + { + // deno-lint-ignore require-yield + *coordinate(): Operation { + collisions += 1; + return { type: "failure", failure: new Error("replacement") }; + }, + }, + { at: "min" }, + ); + return yield* raised( + withWorkspaceCoordinationProvider(provider, durableRun(workflow, { stream })), + ); + }); + + expect(collisions).toBe(0); + expect(activated).toBe(first); + expect(failure).toBe(first); + expect(stream.snapshot()).toEqual([]); + }); }); diff --git a/scripts/runtime-test-exclusions.ts b/scripts/runtime-test-exclusions.ts index 9a63b7fd..8f17b6a2 100644 --- a/scripts/runtime-test-exclusions.ts +++ b/scripts/runtime-test-exclusions.ts @@ -120,6 +120,18 @@ const DENO_ONLY_TOOLING: RuntimeExclusion[] = [ "builds the second copy of the runtime's Files module with `deno bundle`, which is Deno's; the structural recognition it proves is runtime-neutral and is also covered by fatal-cause.test.ts under all three", issue: DERIVED_SCOPE, }, + { + path: "packages/workflow/tests/workspace-effect-transaction.test.ts", + reason: + "proves Deno-private atomic Workspace coordination against the authoritative node:sqlite and DOFS adapter; node:sqlite remains behind --experimental-sqlite on Node 22", + issue: "https://github.com/taras/executable.md/issues/365", + }, + { + path: "packages/workflow/tests/workspace-effect-loaded-copy.test.ts", + reason: + "creates and imports a physical temporary copy of the Workspace modules through Deno.makeTempDir and the Deno module loader", + issue: "https://github.com/taras/executable.md/issues/365", + }, ]; /** diff --git a/specs/executable-mdx-spec.md b/specs/executable-mdx-spec.md index 5beb9e4b..81c4616d 100644 --- a/specs/executable-mdx-spec.md +++ b/specs/executable-mdx-spec.md @@ -7013,12 +7013,53 @@ Defined in [Workflow runs](./workflow-spec.md) §9.5–§9.6. | 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 | +| DLC8 | Explicit selection | A selected coordinator affects only the operation that names it and receives a non-operational publication identity rather than stream capabilities | | 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 | +| DLC10 | Fail-closed Workspace | A missing Workspace provider activates fail-stop before execution or publication and persists no Yield or Close | +| DLC11 | Workspace isolation | Replaceable context carries only provider routing; the selected provider directly invokes the credentialed execution-owned capability, while unrelated durable operations stay 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 | +| DLC14 | Provider infrastructure failure | A selected coordinator activates one first failure by identity and fences later execution and publication | +| DLC15 | One-shot Workspace invocation | A provider can use the execution-owned invocation authority only during its original call; retained execution, publication and failure operations are refused after completion | +| DLC16 | Loaded-copy Workspace selection | A provider installed by one physical package copy coordinates one operation created by another copy exactly once without sharing authority through context or a module registry; substituted selection and retained authority remain fail-closed | +| DLC17 | Forged contextual completion | A same-named invocation middleware that returns a forged successful response reaches no provider, executor or publication; fail-stop prevents Yield and Close persistence and fences later durable work | +| DLC18 | Missing-provider phase refusal | No same-named invocation middleware or execution capability is reached when provider selection is missing | +| DLC19 | Authoritative published result | Middleware that delegates and replaces the returned response cannot replace the exact Result recorded by the execution-owned publication | +| DLC20 | Post-completion isolation | Throwing, suppressing or delegating twice after authoritative completion cannot alter the result or repeat provider, execution or publication work | +| DLC21 | Retained continuation refusal | A contextual continuation retained beyond its original live invocation is stale and cannot repeat execution or publication | +| DLC22 | Minimum-priority invocation isolation | An enclosing same-named minimum-priority handler receives no invocation capability or operational phase and cannot acknowledge publication without delegation | +| DLC23 | Minimum-priority failure isolation | A minimum-priority collision cannot replace first-failure activation; the exact execution-owned failure remains authoritative and no event persists | + +### Tier WAC — Atomic provider-level Workspace coordination + +Defined in [Workflow runs](./workflow-spec.md) §9.5–§9.6 and [Workflow +workspaces](./workflow-workspace-spec.md) §13. + +| # | Test | Verify | +|---|------|--------| +| WAC1 | Atomic success | A real DOFS mutation, immutable root, current-root pointer and filtered Yield remain invisible until their one caller-owned transaction commits; the next serialized turn sees both state and event | +| WAC2 | Supported topology | Write, overwrite, delete, rename, directory, mode, symlink and hardlink operations pass through the adapter-private proof operation | +| WAC3 | Known operation failure | A documented filesystem refusal rolls back its mutation savepoint and commits one failed Yield against the previous root | +| WAC4 | Backing insertion failure | A real journal refusal rolls back mutation, root, pointer and event, activates the exact `DurablePersistenceError`, and fences later work | +| WAC5 | Secret-filter refusal | The existing gate runs before routed insertion; rejection rolls back the outer transaction and activates no compensating event | +| WAC6 | Cancellation | Cancellation before mutation, during mutation and during child teardown publishes no state or event | +| WAC7 | Infrastructure fail-stop | A caught provider infrastructure failure retains identity, rolls back everything and prevents later Workspace and ordinary durable execution | +| WAC8 | Concurrency isolation | A second same-run handle waits cooperatively without enlisting while a different run remains usable inside the outer transaction scope | +| WAC9 | Replay | A retained Workspace Yield bypasses the Deno coordinator and mutation completely | +| WAC10 | Effect authority | Exact proof executors work; missing, symbol-forged, foreign, closed and stale authority is refused before savepoint SQL or mutation | +| WAC11 | Publication identity | The selected journal and nested canonical guards work; an in-memory stream, another run's journal, copied properties, the former symbol name and custom wrappers are refused before mutation or publication | +| WAC12 | Post-publication rollback | A transaction-owner failure after the routed append rolls mutation, retained root, pointer and event back and fences later work | +| WAC13 | Pre-commit cancellation | Cancellation after routed publication but before commit retains no mutation, root, pointer, Yield or Close | +| WAC14 | DOFS continuation lifetime | The proof adapter uses only pinned synchronous byte operations; cancellation before a call performs no mutation, cancellation after it rolls back, mutation teardown precedes savepoint rollback, and the connection remains usable | +| WAC15 | Mutation teardown failure | Child-teardown failure rolls the mutation and outer transaction back, activates that infrastructure failure and fences later work | +| WAC16 | Current-root failure | A real current-root update refusal rolls the captured root, live mutation, pointer and event back | +| WAC17 | Existing fail-stop precedence | An already-active `DurablePersistenceError` retains exact identity and prevents Workspace coordination and mutation | +| WAC18 | Savepoint SQL failure | Actual savepoint create, rollback and release refusals poison the coordinated outer transaction and publish nothing | +| WAC19 | Middleware publication isolation | Enclosing default-order Workspace middleware receives only provider selection and cannot replace publication with a no-op; mutation, root, pointer and one filtered Yield still commit together | +| WAC20 | Middleware failure isolation | Enclosing Workspace middleware cannot replace failure activation; provider infrastructure failure rolls back all state, retains exact identity and fences later effects | +| WAC21 | Selection refusal | A substituted or foreign provider selection is rejected before transaction or savepoint work and leaves no mutation, retained root, pointer change, Yield or Close | +| WAC22 | Minimum-priority publication isolation | A minimum-priority same-named handler cannot observe or acknowledge publication; the real mutation, retained root, current pointer and filtered Yield commit together | +| WAC23 | Minimum-priority failure isolation | A minimum-priority same-named handler cannot replace infrastructure-failure activation; the exact first failure rolls back mutation, roots and journal and fences later work | ### Tier WTX — WorkflowRun savepoints and transaction authority diff --git a/specs/workflow-spec.md b/specs/workflow-spec.md index e5a7a80d..93ed93c6 100644 --- a/specs/workflow-spec.md +++ b/specs/workflow-spec.md @@ -451,12 +451,70 @@ 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 shared Workspace operation wrapper explicitly reads a contextual provider +selection. The replaceable selection API carries no live-operation authority: +it receives no executor, publisher, failure activator or durable publication +identity. The live call retains those exact values behind an execution-owned +capability associated with a same-named contextual invocation operation. +Provider selection creates a one-use route and a separate opaque credential. +The contextual invocation carries the route and an execution-owned capability +but not its credential. A provider installed by another loaded package copy +terminally consumes the route and calls that capability directly. It never sends inspection, execution, +publication, failure activation or completion through `next`, so enclosing +middleware at either priority receives no operational phase. Provider selection +and invocation authority require no module registry and trust no replaceable +structural value. + +The execution-owned capability permits one inspection, execution and +publication during its original live call. It records the exact result only after the +selected provider completes publication, and the durable operation resumes from +that record rather than from the contextual call's structural response. A +short-circuit, premature response or missing publication activates fail-stop; +middleware cannot transform an authoritative published result or turn a later +middleware exception into a different outcome. A foreign or substituted +selector and a reused, completed or stale invocation fail before the provider +opens a transaction. The default missing-provider path also 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 provider binds an adapter-private proof operation +to one exact WorkflowRun handle through module-private executor identity. It +retains the canonical durable-stream module's non-operational publication +identity for that run's journal; the existing secret-filter wrapper privately +preserves the identity only for wrappers it constructs. The provider receives +the invocation's exact executor and publication identities from its +execution-owned capability and validates them before it opens the caller-owned +transaction. It refuses a foreign executor, +publication identity, custom wrapper or copied property before transaction +work. It runs the mutation in one operation savepoint, waits for mutation child +teardown, captures and publishes the immutable root, and routes the +already-filtered Yield through that transaction's journal before commit. Its +publication operation calls the execution-owned publisher directly, so the +transaction body cannot return until the exact Result has been appended and +recorded as authoritative. + +The adapter-private proof filesystem invokes the pinned synchronous DOFS +functions for its supported string and byte-array contract. It does not leave a +Promise, response body or stream pull capable of reaching the authoritative +connection after a cancelled mutation scope has torn down. + +A documented filesystem refusal is an operation result only after its mutation +savepoint has rolled back successfully. It keeps the previous current root and +commits exactly one failed Yield against that root. Connection or authority, +savepoint, DOFS, schema, corruption, capture, current-root, routing, filtering, +serialization, insertion, teardown and commit failures instead roll back the +outer transaction and activate the durable run's first infrastructure failure. +That identity fences later coordinators, executors and appends. An already +active durability failure takes precedence. Cancellation at any phase publishes +nothing. + +The provider-neutral coordinator receives the failure-activation continuation +needed for this boundary. The default live coordinator ignores it and preserves +ordinary success/failure publication. Replay bypasses coordination, and only an +explicit Workspace operation selects the Workspace coordinator. The Deno proof +operation is adapter-private: public filesystem effects and workflow +start/resume do not reach it. The private restoration materializer loads a fully validated retained root and rebuilds directories, files, chunks, modes, mtimes, symbolic links and hardlink @@ -536,7 +594,6 @@ 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 root -selection, history checkpoints and forks; `` integration; +effects; 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. diff --git a/specs/workflow-workspace-spec.md b/specs/workflow-workspace-spec.md index dc39507b..e046613f 100644 --- a/specs/workflow-workspace-spec.md +++ b/specs/workflow-workspace-spec.md @@ -815,6 +815,32 @@ is outside the initial local capability set; Worker Shell follows §10.3. A late Cloudflare-hosted or workerd-backed provider may install the same Workspace and lifecycle contracts; documents do not choose that topology. +The Deno provider's adapter-private proof operation already coordinates one +real DOFS mutation savepoint, immutable root publication and filtered durable +Yield in the caller-owned transaction. Replaceable Workspace context selects a +provider through a non-operational identity and never receives the +live executor, publisher, failure activator or publication witness. The live +call retains those values in an execution-owned capability associated with a +same-named contextual invocation operation. This operation composes with a +provider installed by another loaded package copy without a module registry; +provider selection creates a one-use route and a separate credential, and the +terminal provider handler consumes that route before calling the credentialed +capability directly. Inspection, execution, publication, failure activation and +completion never traverse the contextual continuation, including at minimum +priority. The capability records completion only after the exact published +result returns from the provider. The durable operation resumes from that +record, never from a middleware-supplied response. Before opening the +transaction, the provider requires that capability's proof executor and guarded +durable stream identity to carry the selected WorkflowRun's exact provider-owned +authority. The transaction body calls the execution-owned publisher directly +and cannot return until the exact Result has been appended and recorded. Its +supported filesystem calls use synchronous pinned DOFS primitives, leaving no +asynchronous continuation after mutation teardown. It distinguishes a documented filesystem +refusal from infrastructure failure and cancellation, and activates the durable +fail-stop fence for infrastructure failures. This foundation is not a public +filesystem effect: ``, `API.Files`, workflow start/resume and history +commands do not route to it in this slice. + SQLite is a host implementation detail. The CLI deliberately exposes no remote host-selection option yet, while retaining a control surface that can be delegated without changing the document language. diff --git a/tsconfig.node.json b/tsconfig.node.json index 7fbaf1af..b6afc876 100644 --- a/tsconfig.node.json +++ b/tsconfig.node.json @@ -93,6 +93,10 @@ // suite exercises it. "exclude": [ "packages/durable-streams/demo/**", + // This proof creates and imports a physical package copy through Deno's + // temporary-directory and module-loader APIs; the runtime exclusions keep + // the same file out of the Node and Bun execution suites. + "packages/workflow/tests/workspace-effect-loaded-copy.test.ts", "packages/web/client/main.tsx", "packages/web/client/request.ts" ]