diff --git a/packages/core/src/provisioner/workflows/workflow.destroy.ts b/packages/core/src/provisioner/workflows/workflow.destroy.ts index 813239a..d5ee58e 100644 --- a/packages/core/src/provisioner/workflows/workflow.destroy.ts +++ b/packages/core/src/provisioner/workflows/workflow.destroy.ts @@ -7,7 +7,6 @@ import { import type { StateBackend } from "@notation/state"; import { getResourceGraph } from "src/orchestrator/graph"; import { createDefaultStateBackend } from "../state-backend"; -import { refreshState } from "./workflow.refresh"; export type DestroyAppOptions = { entryPoint: string; @@ -23,11 +22,16 @@ export async function destroyApp({ emit = createLoggerReconcilerSubscriber(), }: DestroyAppOptions) { const state = stateBackend ?? createDefaultStateBackend(); - await refreshState({ entryPoint, registry, state, emit }); - const graph = await getResourceGraph(entryPoint); + + // The registry has to be threaded through: destroy sweeps orphans itself + // now, and without one the sweep falls back to the types of the resources + // still declared — so an orphan whose type the app no longer declares would + // be skipped with a warning instead of deleted. The parameter is optional, + // so nothing but this would catch it. const reconciler = new Reconciler({ state, + registry, emit, }); diff --git a/packages/core/test/provisioner/operation.create.test.ts b/packages/core/test/provisioner/operation.create.test.ts index 2e4f506..84ed0e3 100644 --- a/packages/core/test/provisioner/operation.create.test.ts +++ b/packages/core/test/provisioner/operation.create.test.ts @@ -34,8 +34,10 @@ describe("resource creation", () => { await runOperation( createResourceOperation(step, { resource: testResource, - state: stateBackend, - expectedRev: 0, + resourceParams: await testResource.getParams(), + persist: async function* (next) { + await stateBackend.update(testResource.id, 0, next); + }, }), ); diff --git a/packages/reconciler/package.json b/packages/reconciler/package.json index 39ed2f0..c36012c 100644 --- a/packages/reconciler/package.json +++ b/packages/reconciler/package.json @@ -2,6 +2,16 @@ "type": "module", "name": "@notation/reconciler", "version": "0.12.0", + "exports": { + ".": { + "types": "./dist/index.d.ts", + "default": "./dist/index.js" + }, + "./durable": { + "types": "./dist/durable/index.d.ts", + "default": "./dist/durable/index.js" + } + }, "main": "./dist/index.js", "types": "./dist/index.d.ts", "files": [ @@ -9,12 +19,19 @@ ], "scripts": { "build": "tsup --clean", + "typecheck": "tsc --noEmit", "dev": "tsup --watch" }, "dependencies": { "@notation/resource": "workspace:*", "@notation/state": "workspace:*", + "@yieldstar/core": "0.5.0", "deep-object-diff": "^1.1.9", - "yieldstar": "^0.4.6" + "valibot": "^1.4.2", + "yieldstar": "0.5.0" + }, + "devDependencies": { + "@yieldstar/sqlite-runtime": "0.5.0", + "pino": "^9.9.0" } } diff --git a/packages/reconciler/src/durable/coordination.ts b/packages/reconciler/src/durable/coordination.ts new file mode 100644 index 0000000..094e6f3 --- /dev/null +++ b/packages/reconciler/src/durable/coordination.ts @@ -0,0 +1,130 @@ +import type { ReconcilerEventEmitter } from "../events"; +import { durableEmitter, scopeStep } from "./step"; +import { deploymentCoordinationStore, type CoordinationState } from "./stores"; +import type { DurableStep, StoreClient, WorkflowStore } from "./yieldstar"; + +type CoordinationOptions = { + deploymentId: string; + executionId: string; + emit?: ReconcilerEventEmitter; +}; + +/** + * Prevents concurrent executions from mutating the same deployment. Names + * the holder so an operator can resume it after a crash. + */ +async function* acquireDeploymentCoordination( + step: DurableStep, + opts: CoordinationOptions, +): AsyncGenerator, any> { + const coordination = yield* step.store(deploymentCoordinationStore, { + id: opts.deploymentId, + initial: { holder: null }, + }); + + const snapshot = yield* coordination.get("notation:coordination:inspect"); + const holder = snapshot.state.holder; + if (holder !== null && holder !== opts.executionId) { + yield* durableEmitter(scopeStep(step, "notation:coordination"), opts.emit)({ + level: "warn", + event: "reconciler.coordination.waiting", + deploymentId: opts.deploymentId, + executionId: opts.executionId, + holderExecutionId: holder, + }); + } + + yield* coordination.take( + "notation:coordination:acquire", + (state) => state.holder === null || state.holder === opts.executionId, + (draft) => { + draft.holder = opts.executionId; + }, + ); + + return coordination; +} + +function releaseDeploymentCoordination( + coordination: WorkflowStore, + executionId: string, +) { + return coordination.update("notation:coordination:release", (draft) => { + if (draft.holder === executionId) draft.holder = null; + }); +} + +/** + * Runs `body` while holding the deployment, releasing the hold only once + * `body` has completed. A failed or suspended execution keeps the hold, which + * is what makes it safe to resume: the resumed execution replays `take` from + * the step cache and so never re-acquires anything, so a hold released on the + * way out would leave the resumption mutating a deployment it does not hold. + * + * The cost is that an execution which will never be resumed holds its + * deployment indefinitely. That is deliberate — nothing here can tell "will + * retry" from "abandoned" — and it is resolved by an operator calling + * `takeOverDeploymentHold`. + */ +export async function* withDeploymentHold( + step: DurableStep, + opts: CoordinationOptions, + body: () => AsyncGenerator, +): AsyncGenerator { + const coordination = yield* acquireDeploymentCoordination(step, opts); + yield* body(); + yield* releaseDeploymentCoordination(coordination, opts.executionId); +} + +export type DeploymentHoldTakeover = + | { taken: true; previousHolder: string } + | { taken: false; holder: string | null }; + +/** + * Clears a deployment hold left by an execution that will not be resumed, so + * that later deployments are not blocked behind it. Named separately from the + * workflow path because it is the only supported way out of that state: the + * hold is otherwise released solely by an execution completing. + * + * The write is conditional on `fromExecutionId` still being the named holder, + * so it cannot clear a hold that has since been released and re-taken by + * another execution. Confirm the holder is genuinely dead first: it may still + * be mid-flight, and taking its hold away permits a concurrent mutation of + * the same deployment. + * + * Throws if the deployment has no coordination store, i.e. if it has never + * been deployed. + */ +export async function takeOverDeploymentHold(params: { + storeClient: StoreClient; + deploymentId: string; + fromExecutionId: string; + toExecutionId?: string | null; +}): Promise { + const { storeClient, deploymentId, fromExecutionId } = params; + const read = () => + storeClient.getStore({ + definition: deploymentCoordinationStore, + id: deploymentId, + }); + + const snapshot = await read(); + if (snapshot.state.holder !== fromExecutionId) { + return { taken: false, holder: snapshot.state.holder }; + } + + const result = await storeClient.updateStoreFrom({ + definition: deploymentCoordinationStore, + id: deploymentId, + snapshot, + updater: (draft) => { + draft.holder = params.toExecutionId ?? null; + }, + }); + + if (!result.updated) { + return { taken: false, holder: (await read()).state.holder }; + } + + return { taken: true, previousHolder: fromExecutionId }; +} diff --git a/packages/reconciler/src/durable/deploy.ts b/packages/reconciler/src/durable/deploy.ts new file mode 100644 index 0000000..b28c273 --- /dev/null +++ b/packages/reconciler/src/durable/deploy.ts @@ -0,0 +1,24 @@ +import { buildResourceDepthLevels } from "../dependency-graph"; +import { withDeploymentHold } from "./coordination"; +import { reconcileResource, sweepOrphans } from "./operations"; +import { scopeStep } from "./step"; +import type { DurableDeployOptions } from "./types"; +import type { DurableStep } from "./yieldstar"; + +export async function* deploy( + step: DurableStep, + opts: DurableDeployOptions, +): AsyncGenerator { + yield* withDeploymentHold(step, opts, async function* () { + // Reconcile in dependency order, so a resource only runs once its + // dependencies have converged. + for (const level of buildResourceDepthLevels(opts.resources)) { + for (const resource of level) { + yield* reconcileResource(step, resource, opts); + } + } + + // Then delete resources that are in state but no longer declared. + yield* sweepOrphans(scopeStep(step, "notation:orphans"), opts, "deploy"); + }); +} diff --git a/packages/reconciler/src/durable/destroy.ts b/packages/reconciler/src/durable/destroy.ts new file mode 100644 index 0000000..fa6158a --- /dev/null +++ b/packages/reconciler/src/durable/destroy.ts @@ -0,0 +1,35 @@ +import { buildResourceDepthLevels } from "../dependency-graph"; +import { withDeploymentHold } from "./coordination"; +import { deleteResource, sweepOrphans } from "./operations"; +import { scopeStep } from "./step"; +import type { DurableDestroyOptions } from "./types"; +import type { DurableStep } from "./yieldstar"; + +/** Durably destroys persisted resources in reverse dependency order. */ +export async function* destroy( + step: DurableStep, + opts: DurableDestroyOptions, +): AsyncGenerator { + yield* withDeploymentHold(step, opts, async function* () { + // Delete in reverse dependency order, so dependents are gone before the + // resources they depend on. Resources with no persisted state were never + // created (or are already deleted) and are skipped by deleteResource. + const levels = buildResourceDepthLevels(opts.resources); + for (let index = levels.length - 1; index >= 0; index -= 1) { + for (const resource of levels[index]!) { + yield* deleteResource( + scopeStep(step, `notation:destroy:${resource.id}`), + resource, + opts, + ); + } + } + + // Then delete resources that are in state but no longer declared. + yield* sweepOrphans( + scopeStep(step, "notation:destroy:orphans"), + opts, + "destroy", + ); + }); +} diff --git a/packages/reconciler/src/durable/index.ts b/packages/reconciler/src/durable/index.ts new file mode 100644 index 0000000..72646b2 --- /dev/null +++ b/packages/reconciler/src/durable/index.ts @@ -0,0 +1,44 @@ +/** + * Step keys are a public contract: they are what a resumed execution matches + * its cached work against, so changing one re-executes the work behind it. + * The shapes in use are: + * + * notation:resource::* per-resource reconciliation steps + * notation:destroy::* per-resource deletion steps + * notation:orphans::* orphan sweep on deploy, per record + * notation:destroy:orphans::* orphan sweep on destroy, per record + * *:remote:attempt: one provider call attempt + * *:remote:retry-delay: the wait between two attempts + * emit:[::] event delivery checkpoint + * notation:coordination:* deployment hold: inspect/acquire/release + * state:persist: conditional write of a resource record + * state:delete: conditional removal of one + * + * The state: keys are store-handle keys and so are not scope-prefixed: a + * store outlives the scope that opened it, which is why they carry the + * resource id themselves. + * + * Resource ids are spliced in unescaped, so the delimiter is ambiguous: a + * resource named "orphans" sits in the same key space as the sweep's own + * segment. That predates the key map and is recorded here rather than fixed, + * since changing the composition invalidates in-flight executions. + */ +export { deploy } from "./deploy"; +export { destroy } from "./destroy"; +export { + takeOverDeploymentHold, + type DeploymentHoldTakeover, +} from "./coordination"; +export { DurableStateBackend } from "./state-backend"; +export { + deploymentCoordinationStore, + resourceStateStore, + type CoordinationState, + type StoredResourceState, +} from "./stores"; +export { + type DurableDeployOptions, + type DurableDestroyOptions, + type DurableOperationOptions, +} from "./types"; +export type { DurableStep } from "./yieldstar"; diff --git a/packages/reconciler/src/durable/operations.ts b/packages/reconciler/src/durable/operations.ts new file mode 100644 index 0000000..5ccc5c0 --- /dev/null +++ b/packages/reconciler/src/durable/operations.ts @@ -0,0 +1,197 @@ +import type { BaseResource, ResourceType } from "@notation/resource"; +import { RevConflict } from "@notation/state"; +import { + createMissingResourceRegistryMatchWarningEvent, + createResourceRegistryFromResources, + resolveResourceClass, +} from "../resource-registry"; +import type { PersistState, RemoveState, StepRunner } from "../operations"; +import { + destroyResource, + reconcileResource as reconcile, + type EmitFromStep, + type OpenStateSession, +} from "../reconcile"; +import { durableEmitter, scopeStep, type DurableStepRunner } from "./step"; +import { + resourceStateStore, + toStateNode, + type ResourceSnapshot, +} from "./stores"; +import type { DurableDeployOptions, DurableOperationOptions } from "./types"; +import type { DurableStep } from "./yieldstar"; + +export async function* reconcileResource( + step: DurableStep, + resource: BaseResource, + opts: DurableDeployOptions, +): AsyncGenerator { + const scope = scopeStep(step, `notation:resource:${resource.id}`); + + // Resolved once and then carried: deriveParams is user code and need not be + // deterministic, so an operation resolving them again could persist params + // other than the ones the decision was taken against. The step also pins + // the answer across a replay. + const params = yield* scope.run("params", () => resource.getParams()); + + yield* reconcile(scope, { + resource, + resourceParams: params, + openSession: durableSession(scope, opts), + emit: durableEmit(opts), + dryRun: opts.dryRun, + driftDetection: opts.driftDetection, + maxOperationAttempts: opts.maxOperationAttempts, + }); +} + +export async function* deleteResource( + step: DurableStepRunner, + resource: BaseResource, + opts: DurableOperationOptions, +): AsyncGenerator { + // No recovery pass: a workflow's conditional writes are stamped with the + // step that made them, so a replay is served the recorded result rather + // than losing a race with itself. + yield* destroyResource(step, { + resource, + openSession: durableSession(step, opts), + emit: durableEmit(opts), + dryRun: opts.dryRun, + maxOperationAttempts: opts.maxOperationAttempts, + }); +} + +/** + * Deletes persisted resources that are no longer in the desired set. A state + * node whose type has no registry entry is left in place and surfaced as a + * warning, because deleting it would need a resource class we cannot resolve. + */ +export async function* sweepOrphans( + step: DurableStepRunner, + opts: DurableOperationOptions, + workflow: "deploy" | "destroy", +): AsyncGenerator { + const resourceById = new Map( + opts.resources.map((resource) => [resource.id, resource]), + ); + const persisted = yield* step.run("list", () => opts.state.values()); + const registry = + opts.registry ?? createResourceRegistryFromResources(opts.resources); + + for (const node of persisted) { + if (resourceById.has(node.id)) continue; + const nodeScope = step.scope(node.id); + + const Resource = resolveResourceClass(registry, node.type as ResourceType); + if (!Resource) { + const emit = durableEmitter(nodeScope, opts.emit); + yield* emit( + createMissingResourceRegistryMatchWarningEvent({ + workflow, + resourceId: node.id, + resourceType: node.type as ResourceType, + }), + ); + continue; + } + + const resource = new Resource({ id: node.id, config: node.config }); + resource.setOutput(node.output); + yield* deleteResource(nodeScope, resource, opts); + } +} + +/** Delivery is checkpointed per scope, so the scope decides the key. */ +function durableEmit(opts: DurableOperationOptions): EmitFromStep { + return (step: StepRunner) => durableEmitter(step, opts.emit); +} + +/** + * Reads the persisted record once and binds the writes conditional on it. + * + * The snapshot is the precondition: it names the exact store instance and + * version the record was read at, so a write made against it cannot land on a + * record another writer has moved on. It is re-served to the operations so + * they need no second read. + */ +function durableSession( + step: DurableStepRunner, + opts: DurableOperationOptions, +): OpenStateSession { + return async function* (resource: BaseResource) { + const snapshot = yield* step.run("state:snapshot", () => + opts.state.snapshot(resource.id), + ); + const persist = persistResourceState(step, opts, resource, snapshot); + if (!snapshot) return { node: undefined, persist }; + + return { + node: toStateNode(snapshot), + persist, + remove: removeResourceState(step, opts, resource, snapshot), + }; + }; +} + +/** + * State writes go through the workflow store, never through the state backend: + * the store stamps the write with the step that made it, so the applied-step + * ledger and the state change commit together. Replaying then returns the + * recorded result instead of retrying a compare-and-set that would now fail. + */ +function persistResourceState( + step: DurableStepRunner, + opts: DurableOperationOptions, + resource: BaseResource, + snapshot: ResourceSnapshot | undefined, +): PersistState { + return async function* (next) { + if (!snapshot) { + // Create-if-absent. A racing writer would win here and this record would + // be silently adopted rather than written, which is safe only because a + // deployment is held exclusively for the length of the workflow. + yield* step.store(resourceStateStore, { + id: opts.state.storeId(resource.id), + initial: next, + }); + return; + } + + const store = yield* step.store(resourceStateStore, { + id: opts.state.storeId(resource.id), + }); + const result = yield* store.updateFrom( + `state:persist:${resource.id}`, + snapshot, + () => next, + ); + if (!result.updated) { + throw new RevConflict( + resource.id, + snapshot.version + 1, + result.actualVersion + 1, + ); + } + }; +} + +function removeResourceState( + step: DurableStepRunner, + opts: DurableOperationOptions, + resource: BaseResource, + snapshot: ResourceSnapshot, +): RemoveState { + return async function* () { + const store = yield* step.store(resourceStateStore, { + id: opts.state.storeId(resource.id), + }); + const result = yield* store.deleteFrom( + `state:delete:${resource.id}`, + snapshot, + ); + if (!result.deleted) { + throw new RevConflict(resource.id, snapshot.version + 1, undefined); + } + }; +} diff --git a/packages/reconciler/src/durable/state-backend.ts b/packages/reconciler/src/durable/state-backend.ts new file mode 100644 index 0000000..6e1130d --- /dev/null +++ b/packages/reconciler/src/durable/state-backend.ts @@ -0,0 +1,73 @@ +import type { StateNode } from "@notation/state"; +import { + resourceStateStore, + toStateNode, + type ResourceSnapshot, +} from "./stores"; +import type { StoreClient } from "./yieldstar"; + +/** + * Reads deployment state from outside a workflow, for the planner and for + * anything reporting on a deployment. + * + * Deliberately read-only. Writes belong to the workflow, which makes them + * through the store handle so they are stamped with the step that made them. + * There is nowhere on this interface to carry that idempotency key, so a + * write made here would be repeated on replay rather than recognised. + */ +export class DurableStateBackend { + readonly #client: StoreClient; + readonly #prefix: string; + + constructor(client: StoreClient, deploymentId: string) { + this.#client = client; + // Keep deployment prefixes disjoint so orphan cleanup cannot delete + // another deployment's stores. + this.#prefix = `${encodeURIComponent(deploymentId)}:`; + } + + storeId(resourceId: string) { + return `${this.#prefix}${resourceId}`; + } + + async get(id: string): Promise { + const snapshot = await this.snapshot(id); + return snapshot ? toStateNode(snapshot) : undefined; + } + + async has(id: string): Promise { + return (await this.get(id)) !== undefined; + } + + snapshot(id: string): Promise { + return this.#read(this.storeId(id)); + } + + async values(): Promise { + const ids = await this.#client.listStores(resourceStateStore); + const snapshots = await Promise.all( + ids + .filter((id) => id.startsWith(this.#prefix)) + .map((id) => this.#read(id)), + ); + return snapshots + .filter((snapshot) => snapshot !== undefined) + .map(toStateNode); + } + + // getStore throws for a store that does not exist rather than returning + // undefined, and the error is not distinguishable from a real failure, so + // absence is confirmed by listing. Kept here so no caller has to know that. + async #read(storeId: string): Promise { + try { + return await this.#client.getStore({ + definition: resourceStateStore, + id: storeId, + }); + } catch (error) { + const ids = await this.#client.listStores(resourceStateStore); + if (!ids.includes(storeId)) return undefined; + throw error; + } + } +} diff --git a/packages/reconciler/src/durable/step.ts b/packages/reconciler/src/durable/step.ts new file mode 100644 index 0000000..3a37e6f --- /dev/null +++ b/packages/reconciler/src/durable/step.ts @@ -0,0 +1,75 @@ +import type { + EmitStep, + ReconcilerEvent, + ReconcilerEventEmitter, +} from "../events"; +import type { StepRunner } from "../operations"; +import type { DurableStep } from "./yieldstar"; + +/** + * A durable step runner: the operation seam plus the store handle, which only + * the durable driver uses. + */ +export type DurableStepRunner = { + run( + key: string, + fn: () => T | Promise, + ): AsyncGenerator; + delay(key: string, ms: number): AsyncGenerator; + /** Narrower than StepRunner's, so a scope keeps its store handle. */ + scope(prefix: string): DurableStepRunner; + store: DurableStep["store"]; +}; + +// A durable runner is one of the step runners the operations accept. +type AssertStepRunner = DurableStepRunner extends StepRunner ? true : never; +export type DurableStepRunnerIsStepRunner = AssertStepRunner; + +/** + * Namespaces the step keys of `step` so an operation can be written once and + * replayed at several call sites without its keys colliding. + * + * Opening a store is not prefixed: yieldstar derives that key from the store + * name and store id, which is already unique. The keys a store *handle* takes + * are caller-supplied and are left alone too — a store outlives the scope + * that opened it, so its call sites qualify their own keys. + */ +export function scopeStep( + step: DurableStep, + prefix: string, +): DurableStepRunner { + const scoped = (key: string) => `${prefix}:${key}`; + + return { + run: ((key: string, fn: any) => + step.run(scoped(key), fn)) as DurableStepRunner["run"], + delay: (key: string, ms: number) => step.delay(scoped(key), ms), + store: step.store, + scope: (nested: string) => scopeStep(step, scoped(nested)), + }; +} + +/** + * Checkpoints delivery so that replaying a workflow does not re-emit. The key + * is derived from the event itself, which keeps it deterministic across a + * replay; the enclosing scope is what keeps it unique, since an operation + * emits each (operation, status) pair at most once. + * + * Emitters must still tolerate a duplicate: the process can crash after the + * event is delivered but before the checkpoint is written. + */ +export function durableEmitter( + step: Pick, + emit: ReconcilerEventEmitter | undefined, +): EmitStep { + return async function* (event) { + if (!emit) return; + yield* step.run(emitKey(event), () => emit(event)); + }; +} + +function emitKey(event: ReconcilerEvent): string { + return event.event === "reconciler.operation.lifecycle" + ? `emit:${event.event}:${event.operation}:${event.status}` + : `emit:${event.event}`; +} diff --git a/packages/reconciler/src/durable/stores.ts b/packages/reconciler/src/durable/stores.ts new file mode 100644 index 0000000..c5f03b5 --- /dev/null +++ b/packages/reconciler/src/durable/stores.ts @@ -0,0 +1,47 @@ +import type { StateNode } from "@notation/state"; +import * as v from "valibot"; +import { defineStore, type StoreSnapshot } from "./yieldstar"; + +/** + * Loose on purpose: PersistedResourceState carries an index signature so a + * driver can persist fields this schema does not name yet, and v.object would + * strip those at the store boundary rather than carry them forward. Every + * field an operation actually writes is still declared, so the schema is a + * floor on the record rather than a description of one. + */ +export const resourceStateStore = defineStore( + "resource-state", + v.looseObject({ + id: v.string(), + type: v.string(), + // Written by create and update. -1 and "" are BaseResource's defaults for + // a resource that belongs to no group. + groupId: v.number(), + groupType: v.string(), + config: v.record(v.string(), v.unknown()), + params: v.record(v.string(), v.unknown()), + output: v.record(v.string(), v.unknown()), + lastOperation: v.picklist(["drift", "create", "update", "delete"]), + lastOperationAt: v.string(), + }), +); + +export const deploymentCoordinationStore = defineStore( + "deployment-coordination", + v.object({ holder: v.nullable(v.string()) }), +); + +export type StoredResourceState = v.InferOutput< + typeof resourceStateStore.schema +>; +export type CoordinationState = v.InferOutput< + typeof deploymentCoordinationStore.schema +>; + +/** A read of a resource record, carrying the identity a write is made against. */ +export type ResourceSnapshot = StoreSnapshot; + +/** Store versions count from zero, state revisions from one. */ +export function toStateNode(snapshot: ResourceSnapshot): StateNode { + return { ...snapshot.state, rev: snapshot.version + 1 }; +} diff --git a/packages/reconciler/src/durable/types.ts b/packages/reconciler/src/durable/types.ts new file mode 100644 index 0000000..76f7000 --- /dev/null +++ b/packages/reconciler/src/durable/types.ts @@ -0,0 +1,21 @@ +import type { BaseResource } from "@notation/resource"; +import type { ReconcilerEventEmitter } from "../events"; +import type { ResourceRegistry } from "../resource-registry"; +import type { DurableStateBackend } from "./state-backend"; + +export type DurableOperationOptions = { + deploymentId: string; + executionId: string; + resources: BaseResource[]; + state: DurableStateBackend; + registry?: ResourceRegistry; + dryRun?: boolean; + emit?: ReconcilerEventEmitter; + maxOperationAttempts?: number; +}; + +export type DurableDeployOptions = DurableOperationOptions & { + driftDetection?: boolean; +}; + +export type DurableDestroyOptions = DurableOperationOptions; diff --git a/packages/reconciler/src/durable/yieldstar.ts b/packages/reconciler/src/durable/yieldstar.ts new file mode 100644 index 0000000..afa9871 --- /dev/null +++ b/packages/reconciler/src/durable/yieldstar.ts @@ -0,0 +1,8 @@ +import type { WorkflowFn } from "yieldstar"; + +export { defineStore } from "yieldstar"; +export type { WorkflowStore } from "yieldstar"; +export type { StoreClient, StoreSnapshot } from "@yieldstar/core"; + +/** The durable step primitive the runtime hands to workflow functions. */ +export type DurableStep = Parameters>[0]; diff --git a/packages/reconciler/src/events.ts b/packages/reconciler/src/events.ts new file mode 100644 index 0000000..46aacaf --- /dev/null +++ b/packages/reconciler/src/events.ts @@ -0,0 +1,73 @@ +import type { ResourceType } from "@notation/resource"; +import type { MissingResourceRegistryMatchWarningEvent } from "./resource-registry"; + +export type OperationName = "create" | "read" | "update" | "delete"; + +export type OperationLifecycleStatus = + "start" | "success" | "error" | "skip" | "dry-run"; + +export type OperationLifecycleEvent = { + level: "info" | "error"; + event: "reconciler.operation.lifecycle"; + operation: OperationName; + status: OperationLifecycleStatus; + resourceId: string; + resourceType: ResourceType; + reason?: string; + errorName?: string; + errorMessage?: string; +}; + +export type ReconcilerDeployEvent = { + level: "info"; + event: "reconciler.deploy.decision"; + resourceId: string; + resourceType: string; + decision: "create" | "update" | "drift-update" | "drift-recreate" | "noop"; +}; + +export type ReconcilerDriftDetectedEvent = { + level: "info"; + event: "reconciler.drift.detected"; + resourceId: string; + resourceType: string; + diff: Record; +}; + +export type CoordinationWaitingEvent = { + level: "warn"; + event: "reconciler.coordination.waiting"; + deploymentId: string; + executionId: string; + holderExecutionId: string; +}; + +export type ReconcilerEvent = + | OperationLifecycleEvent + | CoordinationWaitingEvent + | ReconcilerDeployEvent + | ReconcilerDriftDetectedEvent + | MissingResourceRegistryMatchWarningEvent; + +export type ReconcilerEventEmitter = ( + event: ReconcilerEvent, +) => void | Promise; + +/** + * The seam a driver fills in to deliver an event. Emission is a step so that + * each driver decides how it is recorded: the in-process driver simply awaits + * the emitter, while the durable driver checkpoints it so that replaying a + * workflow does not re-emit events it has already delivered. + */ +export type EmitStep = ( + event: TEvent, +) => AsyncGenerator; + +/** Adapts a plain emitter to the driver seam, for drivers that just await. */ +export function toEmitStep( + emit: ((event: TEvent) => void | Promise) | undefined, +): EmitStep { + return async function* (event) { + await emit?.(event); + }; +} diff --git a/packages/reconciler/src/index.ts b/packages/reconciler/src/index.ts index 44fc402..cdc37c8 100644 --- a/packages/reconciler/src/index.ts +++ b/packages/reconciler/src/index.ts @@ -1,12 +1,13 @@ export type ResourceApi = typeof import("@notation/resource"); export type StateApi = typeof import("@notation/state"); export type DeepObjectDiffApi = typeof import("deep-object-diff"); -export type YieldstarApi = typeof import("yieldstar"); export * from "./resource-registry"; export * from "./operations"; export * from "./dependency-graph"; +export * from "./events"; export * from "./plan"; +export * from "./planner"; export * from "./reconciler"; export * from "./logger-subscriber"; export * from "./protocol"; diff --git a/packages/reconciler/src/operations/operation.create.ts b/packages/reconciler/src/operations/operation.create.ts index e0aee4f..d76d06f 100644 --- a/packages/reconciler/src/operations/operation.create.ts +++ b/packages/reconciler/src/operations/operation.create.ts @@ -1,4 +1,3 @@ -import { createWorkflow } from "yieldstar"; import { type CreateResourceParams, type StepRunner, @@ -12,17 +11,15 @@ export async function* createResourceOperation( step: StepRunner, params: CreateResourceParams, ): AsyncGenerator { - await emitLifecycleEvent(params, "create", "start"); + yield* emitLifecycleEvent(params, "create", "start"); if (params.dryRun) { - await emitLifecycleEvent(params, "create", "dry-run"); + yield* emitLifecycleEvent(params, "create", "dry-run"); return; } try { - const resourceParams = yield* step.run("create:get-params", () => - params.resource.getParams(), - ); + const resourceParams = params.resourceParams; const computedPrimaryKey = yield* runPendingOperation( step, @@ -41,7 +38,8 @@ export async function* createResourceOperation( const readResult = yield* readResourceOperation(step, { resource: params.resource, - state: params.state, + resourceParams, + persistedOutput: params.persistedOutput, emit: params.emit, maxOperationAttempts: params.maxOperationAttempts, }); @@ -51,32 +49,21 @@ export async function* createResourceOperation( ...readResult, }); - yield* step.run("create:persist-state", async () => { - await params.state.update(params.resource.id, params.expectedRev, { - id: params.resource.id, - groupId: params.resource.groupId, - groupType: params.resource.groupType, - type: params.resource.type, - lastOperation: "create", - lastOperationAt: new Date().toISOString(), - config: params.resource.config, - params: params.resource.toState(resourceParams), - output: params.resource.toState(params.resource.output), - }); + yield* params.persist({ + id: params.resource.id, + groupId: params.resource.groupId, + groupType: params.resource.groupType, + type: params.resource.type, + lastOperation: "create", + lastOperationAt: new Date().toISOString(), + config: params.resource.config, + params: params.resource.toState(resourceParams), + output: params.resource.toState(params.resource.output), }); - await emitLifecycleEvent(params, "create", "success"); + yield* emitLifecycleEvent(params, "create", "success"); } catch (err) { - await emitLifecycleEvent(params, "create", "error", getErrorDetails(err)); + yield* emitLifecycleEvent(params, "create", "error", getErrorDetails(err)); throw err; } } - -export const createResourceWorkflow: unknown = createWorkflow( - async function* (step, event) { - return yield* createResourceOperation( - step as StepRunner, - event.params as CreateResourceParams, - ); - }, -); diff --git a/packages/reconciler/src/operations/operation.delete.ts b/packages/reconciler/src/operations/operation.delete.ts index 51975b6..3bb4e8b 100644 --- a/packages/reconciler/src/operations/operation.delete.ts +++ b/packages/reconciler/src/operations/operation.delete.ts @@ -1,4 +1,3 @@ -import { createWorkflow } from "yieldstar"; import { type DeleteResourceParams, type StepRunner, @@ -11,10 +10,10 @@ export async function* deleteResourceOperation( step: StepRunner, params: DeleteResourceParams, ): AsyncGenerator { - await emitLifecycleEvent(params, "delete", "start"); + yield* emitLifecycleEvent(params, "delete", "start"); if (params.dryRun) { - await emitLifecycleEvent(params, "delete", "dry-run"); + yield* emitLifecycleEvent(params, "delete", "dry-run"); return; } @@ -31,22 +30,11 @@ export async function* deleteResourceOperation( params.maxOperationAttempts, ); - yield* step.run("delete:persist-state", () => - params.state.delete(params.resource.id, params.expectedRev), - ); + yield* params.remove(); - await emitLifecycleEvent(params, "delete", "success"); + yield* emitLifecycleEvent(params, "delete", "success"); } catch (err) { - await emitLifecycleEvent(params, "delete", "error", getErrorDetails(err)); + yield* emitLifecycleEvent(params, "delete", "error", getErrorDetails(err)); throw err; } } - -export const deleteResourceWorkflow: unknown = createWorkflow( - async function* (step, event) { - return yield* deleteResourceOperation( - step as StepRunner, - event.params as DeleteResourceParams, - ); - }, -); diff --git a/packages/reconciler/src/operations/operation.read.ts b/packages/reconciler/src/operations/operation.read.ts index e74196a..76e02da 100644 --- a/packages/reconciler/src/operations/operation.read.ts +++ b/packages/reconciler/src/operations/operation.read.ts @@ -1,4 +1,5 @@ -import { createWorkflow } from "yieldstar"; +import { ResourceNotFoundError } from "@notation/resource"; +import type { DriftRead } from "../plan"; import { type ReadResourceParams, type StepRunner, @@ -11,30 +12,25 @@ export async function* readResourceOperation( step: StepRunner, params: ReadResourceParams, ): AsyncGenerator, unknown> { - await emitLifecycleEvent(params, "read", "start"); + yield* emitLifecycleEvent(params, "read", "start"); if (params.dryRun) { - await emitLifecycleEvent(params, "read", "dry-run"); + yield* emitLifecycleEvent(params, "read", "dry-run"); return {}; } try { - const resourceParams = yield* step.run("read:get-params", () => - params.resource.getParams(), - ); + const resourceParams = params.resourceParams; if (!params.resource.read) { - const stateNode = yield* step.run("read:get-state-node", () => - params.state.get(params.resource.id), - ); - const merged = stateNode - ? { ...stateNode.output, ...resourceParams } + const merged = params.persistedOutput + ? { ...params.persistedOutput, ...resourceParams } : resourceParams; - await emitLifecycleEvent(params, "read", "skip", { + yield* emitLifecycleEvent(params, "read", "skip", { reason: "read-not-implemented", }); - await emitLifecycleEvent(params, "read", "success"); + yield* emitLifecycleEvent(params, "read", "success"); return merged as Record; } @@ -50,19 +46,28 @@ export async function* readResourceOperation( ...remote, }; - await emitLifecycleEvent(params, "read", "success"); + yield* emitLifecycleEvent(params, "read", "success"); return mergedOutput; } catch (err) { - await emitLifecycleEvent(params, "read", "error", getErrorDetails(err)); + yield* emitLifecycleEvent(params, "read", "error", getErrorDetails(err)); throw err; } } -export const readResourceWorkflow: unknown = createWorkflow( - async function* (step, event) { - return yield* readResourceOperation( - step as StepRunner, - event.params as ReadResourceParams, - ); - }, -); +/** + * Reads the remote to compare it against persisted state. An absent resource + * is a fact about the world rather than a failure, so it is reported as such; + * every other error still propagates. + */ +export async function* readDriftOperation( + step: StepRunner, + params: ReadResourceParams, +): AsyncGenerator { + try { + const output = yield* readResourceOperation(step, params); + return { kind: "present", output }; + } catch (error) { + if (ResourceNotFoundError.is(error)) return { kind: "absent" }; + throw error; + } +} diff --git a/packages/reconciler/src/operations/operation.types.ts b/packages/reconciler/src/operations/operation.types.ts index 6cbcd1c..492f4c9 100644 --- a/packages/reconciler/src/operations/operation.types.ts +++ b/packages/reconciler/src/operations/operation.types.ts @@ -1,58 +1,103 @@ -import type { BaseResource, ResourceType } from "@notation/resource"; -import type { State } from "@notation/state"; +import type { BaseResource } from "@notation/resource"; +import type { StateNode } from "@notation/state"; +import type { + EmitStep, + OperationLifecycleEvent, + OperationLifecycleStatus, + OperationName, +} from "../events"; -export type OperationName = "create" | "read" | "update" | "delete"; +export type { + OperationLifecycleEvent, + OperationLifecycleStatus, + OperationName, +} from "../events"; -export type OperationLifecycleStatus = - "start" | "success" | "error" | "skip" | "dry-run"; - -export type OperationLifecycleEvent = { - level: "info" | "error"; - event: "reconciler.operation.lifecycle"; - operation: OperationName; - status: OperationLifecycleStatus; - resourceId: string; - resourceType: ResourceType; - reason?: string; - errorName?: string; - errorMessage?: string; -}; - -export type OperationEventEmitter = ( - event: OperationLifecycleEvent, -) => void | Promise; +export type OperationEventEmitter = EmitStep; +/** + * How an operation runs a step, and how it namespaces the steps it runs. + * + * Keys are mandatory. A driver may ignore them — the in-process one does — + * but a keyless step would have to derive a key from its call site, which for + * a step reached through `scope` is the scoping wrapper's call site rather + * than the caller's, so any two keyless steps under one scope would collide. + * + * `scope` is the seam that lets one operation run at several call sites in a + * single execution: in process it is the identity, and in a workflow it + * prefixes the keys the runtime caches against. + */ export type StepRunner = { - run(fn: () => T | Promise): AsyncGenerator; run( key: string, fn: () => T | Promise, ): AsyncGenerator; - delay(ms: number): AsyncGenerator; delay(key: string, ms: number): AsyncGenerator; + scope(prefix: string): StepRunner; +}; + +/** + * The record an operation wants persisted; the driver owns the revision. + * Spelled out rather than derived with Omit, which would collapse against + * StateNode's index signature and widen every field to unknown. + */ +export type PersistedResourceState = Pick< + StateNode, + | "id" + | "type" + | "config" + | "params" + | "output" + | "lastOperation" + | "lastOperationAt" +> & { + // Not on StateNode itself, where they arrive through its index signature. + groupId: number; + groupType: string; + [key: string]: unknown; }; +/** + * How a driver writes state. Both are steps so that each driver can carry its + * own concurrency control: in process that is a compare-and-set against the + * revision read before the operation, and in a workflow it is a store write + * stamped with the step that made it, so a replay does not repeat it. + */ +export type PersistState = ( + next: PersistedResourceState, +) => AsyncGenerator; + +export type RemoveState = () => AsyncGenerator; + export type ResourceOperationBaseParams = { resource: BaseResource; - state: Pick; dryRun?: boolean; emit?: OperationEventEmitter; maxOperationAttempts?: number; }; -export type CreateResourceParams = ResourceOperationBaseParams & { - expectedRev: number; +/** + * Everything a read needs is resolved before the operation starts: the + * desired params, and — for a resource with no read operation — the output + * the last write persisted. An operation that resolved either itself could + * see a different answer from the one the decision was taken against. + */ +export type ReadResourceParams = ResourceOperationBaseParams & { + resourceParams: Record; + persistedOutput?: Record; }; -export type ReadResourceParams = ResourceOperationBaseParams; +export type CreateResourceParams = ReadResourceParams & { + persist: PersistState; +}; -export type UpdateResourceParams = ResourceOperationBaseParams & { +export type UpdateResourceParams = ReadResourceParams & { patch: Record; - expectedRev: number; + persist: PersistState; }; export type DeleteResourceParams = ResourceOperationBaseParams & { - expectedRev: number; + remove: RemoveState; }; export function getErrorDetails(err: unknown): { @@ -72,15 +117,15 @@ export function getErrorDetails(err: unknown): { }; } -export async function emitLifecycleEvent( +export async function* emitLifecycleEvent( params: ResourceOperationBaseParams, operation: OperationName, status: OperationLifecycleStatus, extra: Partial = {}, -) { +): AsyncGenerator { if (!params.emit) return; - await params.emit({ + yield* params.emit({ level: status === "error" ? "error" : "info", event: "reconciler.operation.lifecycle", operation, diff --git a/packages/reconciler/src/operations/operation.update.ts b/packages/reconciler/src/operations/operation.update.ts index fc62a8c..838161e 100644 --- a/packages/reconciler/src/operations/operation.update.ts +++ b/packages/reconciler/src/operations/operation.update.ts @@ -1,4 +1,3 @@ -import { createWorkflow } from "yieldstar"; import { type StepRunner, type UpdateResourceParams, @@ -12,25 +11,23 @@ export async function* updateResourceOperation( step: StepRunner, params: UpdateResourceParams, ): AsyncGenerator { - await emitLifecycleEvent(params, "update", "start"); + yield* emitLifecycleEvent(params, "update", "start"); if (params.dryRun) { - await emitLifecycleEvent(params, "update", "dry-run"); + yield* emitLifecycleEvent(params, "update", "dry-run"); return; } if (!params.resource.update) { - await emitLifecycleEvent(params, "update", "skip", { + yield* emitLifecycleEvent(params, "update", "skip", { reason: "update-not-implemented", }); - await emitLifecycleEvent(params, "update", "success"); + yield* emitLifecycleEvent(params, "update", "success"); return; } try { - const resourceParams = yield* step.run("update:get-params", () => - params.resource.getParams(), - ); + const resourceParams = params.resourceParams; yield* runPendingOperation( step, @@ -53,7 +50,8 @@ export async function* updateResourceOperation( const readResult = yield* readResourceOperation(step, { resource: params.resource, - state: params.state, + resourceParams, + persistedOutput: params.persistedOutput, emit: params.emit, maxOperationAttempts: params.maxOperationAttempts, }); @@ -63,32 +61,21 @@ export async function* updateResourceOperation( ...readResult, }); - yield* step.run("update:persist-state", async () => { - await params.state.update(params.resource.id, params.expectedRev, { - id: params.resource.id, - groupId: params.resource.groupId, - groupType: params.resource.groupType, - type: params.resource.type, - lastOperation: "update", - lastOperationAt: new Date().toISOString(), - config: params.resource.config, - params: params.resource.toState(resourceParams), - output: params.resource.toState(params.resource.output), - }); + yield* params.persist({ + id: params.resource.id, + groupId: params.resource.groupId, + groupType: params.resource.groupType, + type: params.resource.type, + lastOperation: "update", + lastOperationAt: new Date().toISOString(), + config: params.resource.config, + params: params.resource.toState(resourceParams), + output: params.resource.toState(params.resource.output), }); - await emitLifecycleEvent(params, "update", "success"); + yield* emitLifecycleEvent(params, "update", "success"); } catch (err) { - await emitLifecycleEvent(params, "update", "error", getErrorDetails(err)); + yield* emitLifecycleEvent(params, "update", "error", getErrorDetails(err)); throw err; } } - -export const updateResourceWorkflow: unknown = createWorkflow( - async function* (step, event) { - return yield* updateResourceOperation( - step as StepRunner, - event.params as UpdateResourceParams, - ); - }, -); diff --git a/packages/reconciler/src/planner.ts b/packages/reconciler/src/planner.ts new file mode 100644 index 0000000..8ac8b27 --- /dev/null +++ b/packages/reconciler/src/planner.ts @@ -0,0 +1,88 @@ +import type { BaseResource } from "@notation/resource"; +import type { State } from "@notation/state"; +import { buildResourceDepthLevels } from "./dependency-graph"; +import { toEmitStep, type ReconcilerEventEmitter } from "./events"; +import { readDriftOperation } from "./operations"; +import { + decideAction, + getDependencyIds, + resolvePlanParams, + type Plan, + type PlanNode, +} from "./plan"; +import { createStepRunner, runOperation } from "./step-runner"; + +/** Planning reads state and never mutates it, so no lease is required. */ +export type PlannerState = Pick; + +export type CreatePlanOptions = { + resources: BaseResource[]; + state: PlannerState; + driftDetection?: boolean; + emit?: ReconcilerEventEmitter; + maxOperationAttempts?: number; +}; + +export async function createPlan({ + resources, + state, + driftDetection = true, + emit, + maxOperationAttempts, +}: CreatePlanOptions): Promise { + const resourceById = new Map( + resources.map((resource) => [resource.id, resource]), + ); + const emitStep = emit ? toEmitStep(emit) : undefined; + const nodes: PlanNode[] = []; + + for (const level of buildResourceDepthLevels(resources)) { + for (const resource of level) { + const stateNode = await state.get(resource.id); + if (stateNode) resource.setOutput(stateNode.output); + const params = await resolvePlanParams(resource); + let action = decideAction({ resource, stateNode, params }); + + if (action.decision === "noop" && driftDetection && resource.read) { + const driftRead = await runOperation( + readDriftOperation(createStepRunner(), { + resource, + resourceParams: params, + persistedOutput: stateNode?.output, + emit: emitStep, + maxOperationAttempts, + }), + ); + + action = decideAction({ + resource, + stateNode, + params, + driftRead, + }); + } + + nodes.push({ + id: resource.id, + type: resource.type, + decision: action.decision, + ...("diff" in action ? { diff: action.diff } : {}), + params, + dependsOn: getDependencyIds(resource), + }); + } + } + + for (const stateNode of await state.values()) { + if (resourceById.has(stateNode.id)) continue; + nodes.push({ + id: stateNode.id, + type: stateNode.type, + decision: "delete-orphan", + params: stateNode.params, + dependsOn: [], + }); + } + + return { createdAt: new Date().toISOString(), nodes }; +} diff --git a/packages/reconciler/src/protocol.ts b/packages/reconciler/src/protocol.ts index bf3706e..a8ab407 100644 --- a/packages/reconciler/src/protocol.ts +++ b/packages/reconciler/src/protocol.ts @@ -1,4 +1,4 @@ -import type { ReconcilerEvent, ReconcilerEventEmitter } from "./reconciler"; +import type { ReconcilerEvent, ReconcilerEventEmitter } from "./events"; export const EVENT_STREAM_VERSION = 1 as const; diff --git a/packages/reconciler/src/reconcile.ts b/packages/reconciler/src/reconcile.ts new file mode 100644 index 0000000..2b125e9 --- /dev/null +++ b/packages/reconciler/src/reconcile.ts @@ -0,0 +1,251 @@ +import type { BaseResource } from "@notation/resource"; +import type { RevConflict, StateNode } from "@notation/state"; +import type { EmitStep, ReconcilerEvent } from "./events"; +import { + createResourceOperation, + deleteResourceOperation, + readDriftOperation, + updateResourceOperation, + type PersistState, + type RemoveState, + type StepRunner, +} from "./operations"; +import { decideAction } from "./plan"; + +/** + * A read of a resource's persisted record, together with the writes that are + * conditional on that exact read. + * + * The union is the point: `remove` exists only alongside a `node`, because + * removing a record that was never read is not a thing either driver can do + * safely. Keeping the writes bound to the read that produced them is what + * stops a node being combined with a precondition from a different read — a + * revision in process, a store snapshot in a workflow. + */ +export type ResourceStateSession = + | { node: undefined; persist: PersistState; remove?: never } + | { node: StateNode; persist: PersistState; remove: RemoveState }; + +/** + * Opens a session. A factory rather than an open session because recovery + * re-reads: the first pass opens once, and a recovering pass opens again + * against whatever the winning writer left behind. + */ +export type OpenStateSession = ( + resource: BaseResource, +) => AsyncGenerator; + +/** + * How a driver delivers an event from a given scope. Emission is scoped + * because a workflow checkpoints it, and two reads of the same resource in + * one execution must not share a checkpoint key. + */ +export type EmitFromStep = (step: StepRunner) => EmitStep; + +export type ReconcileResourceOptions = { + resource: BaseResource; + /** Resolved once per scheduled reconciliation, recovery included. */ + resourceParams: Record; + openSession: OpenStateSession; + emit?: EmitFromStep; + dryRun?: boolean; + driftDetection?: boolean; + maxOperationAttempts?: number; + /** + * Set when a previous attempt lost a conditional write. The remote is then + * read unconditionally and the decision retaken against it, so the attempt + * adopts what the winning writer did rather than repeating its own work. + */ + recoverFrom?: RevConflict; +}; + +/** + * Reconciles one resource: hydrate, decide, read the remote when the decision + * needs it, announce the decision, then act. + * + * Both drivers run this same generator. What stays outside it is scheduling + * (dependency levels, concurrency), the conflict retry policy, opening state + * sessions, and how a step is run — in process a plain await under a mutation + * lease, in a workflow a checkpointed step under a deployment hold. + */ +export async function* reconcileResource( + step: StepRunner, + opts: ReconcileResourceOptions, +): AsyncGenerator { + const { resource, resourceParams: params } = opts; + const recovering = opts.recoverFrom !== undefined; + + // Recovery is built on re-reading the remote, so a resource that cannot be + // read cannot be recovered: the conflict is the caller's answer. + if (recovering && !resource.read) throw opts.recoverFrom; + + const emit = opts.emit?.(step); + const session = yield* opts.openSession(resource); + if (session.node) resource.setOutput(session.node.output); + + let action = decideAction({ resource, stateNode: session.node, params }); + + // A noop is only trusted once the remote has been read back: the provider + // may have drifted from persisted state, which upgrades the decision. + // Recovery always reads, because the point of it is to see what the writer + // that won the race actually left behind. + if ( + recovering || + (action.decision === "noop" && (opts.driftDetection ?? true)) + ) { + // Its own scope: the operation that follows reads the remote again, and + // the two reads must not share step keys. + const driftStep = step.scope("drift-read"); + const driftRead = yield* readDriftOperation(driftStep, { + resource, + resourceParams: params, + persistedOutput: session.node?.output, + // Deliberately no dryRun: a dry run suppresses mutations, not reads. + // Reading is how a dry run reports drift at all. + emit: opts.emit?.(driftStep), + maxOperationAttempts: opts.maxOperationAttempts, + }); + action = decideAction({ + resource, + stateNode: session.node, + params, + driftRead, + }); + if (recovering && driftRead.kind === "present") { + resource.setOutput(driftRead.output); + } + } + + if (action.decision === "drift-update") { + yield* emitEvent(emit, { + level: "info", + event: "reconciler.drift.detected", + resourceId: resource.id, + resourceType: resource.type, + diff: action.patch, + }); + } + + yield* emitEvent(emit, { + level: "info", + event: "reconciler.deploy.decision", + resourceId: resource.id, + resourceType: resource.type, + decision: action.decision, + }); + + const shared = { + resource, + resourceParams: params, + persistedOutput: session.node?.output, + dryRun: opts.dryRun, + emit, + maxOperationAttempts: opts.maxOperationAttempts, + }; + + switch (action.decision) { + case "create": + case "drift-recreate": + yield* createResourceOperation(step, { + ...shared, + persist: session.persist, + }); + return; + case "update": + case "drift-update": + yield* updateResourceOperation(step, { + ...shared, + patch: action.patch, + persist: session.persist, + }); + return; + case "noop": + // A first-pass noop writes nothing. A recovering noop has to: it read + // the remote, found it already converged, and owes a record of that + // adoption at the revision it re-read, or the next deployment would + // reconcile against the losing writer's view. + if (!recovering || opts.dryRun) return; + yield* session.persist({ + id: resource.id, + groupId: resource.groupId, + groupType: resource.groupType, + type: resource.type, + lastOperation: "drift", + lastOperationAt: new Date().toISOString(), + config: resource.config, + params: resource.toState(params), + output: resource.toState(resource.output), + }); + } +} + +export type DestroyResourceOptions = { + resource: BaseResource; + openSession: OpenStateSession; + emit?: EmitFromStep; + dryRun?: boolean; + maxOperationAttempts?: number; + /** + * Set when a previous attempt lost its conditional removal. Carries the + * params because the recovery read needs them, and deletion is the one path + * that never resolves them otherwise. + */ + recoverFrom?: { + conflict: RevConflict; + resourceParams: Record; + }; +}; + +/** + * Deletes one resource. A resource with no persisted record was never created + * — or has already been deleted — and is skipped, which is also what makes + * the sweep of a partly-deleted deployment idempotent. + */ +export async function* destroyResource( + step: StepRunner, + opts: DestroyResourceOptions, +): AsyncGenerator { + const { resource } = opts; + const session = yield* opts.openSession(resource); + if (!session.node) return; + resource.setOutput(session.node.output); + + if (opts.recoverFrom) { + const { conflict, resourceParams } = opts.recoverFrom; + if (!resource.read) throw conflict; + + const driftStep = step.scope("drift-read"); + const remote = yield* readDriftOperation(driftStep, { + resource, + resourceParams, + persistedOutput: session.node.output, + emit: opts.emit?.(driftStep), + maxOperationAttempts: opts.maxOperationAttempts, + }); + + // Already gone remotely: the delete succeeded and only the record is + // left, so drop the record rather than calling the provider again. + if (remote.kind !== "present") { + if (!opts.dryRun) yield* session.remove(); + return; + } + resource.setOutput(remote.output); + } + + yield* deleteResourceOperation(step, { + resource, + dryRun: opts.dryRun, + emit: opts.emit?.(step), + maxOperationAttempts: opts.maxOperationAttempts, + remove: session.remove, + }); +} + +function emitEvent( + emit: EmitStep | undefined, + event: ReconcilerEvent, +): AsyncGenerator { + return emit ? emit(event) : noSteps(); +} + +async function* noSteps(): AsyncGenerator {} diff --git a/packages/reconciler/src/reconciler.ts b/packages/reconciler/src/reconciler.ts index fab6c0b..8c119cf 100644 --- a/packages/reconciler/src/reconciler.ts +++ b/packages/reconciler/src/reconciler.ts @@ -1,58 +1,32 @@ -import { ResourceNotFoundError } from "@notation/resource"; import type { BaseResource, ResourceType } from "@notation/resource"; import { RevConflict, type State, type StateNode } from "@notation/state"; import { setTimeout as sleep } from "node:timers/promises"; import { buildResourceDepthLevels } from "./dependency-graph"; +import type { Plan } from "./plan"; +import type { PersistState, RemoveState, StepRunner } from "./operations"; import { - decideAction, - getDependencyIds, - resolvePlanParams, - type DriftRead, - type Plan, - type PlanNode, - type ResourceAction, -} from "./plan"; -import { - createResourceOperation, - deleteResourceOperation, - readResourceOperation, - type OperationLifecycleEvent, - type StepRunner, - updateResourceOperation, -} from "./operations"; + destroyResource, + reconcileResource, + type EmitFromStep, + type OpenStateSession, +} from "./reconcile"; import { createMissingResourceRegistryMatchWarningEvent, createResourceRegistryFromResources, resolveResourceClass, - type MissingResourceRegistryMatchWarningEvent, type ResourceRegistry, } from "./resource-registry"; - -export type ReconcilerDeployEvent = { - level: "info"; - event: "reconciler.deploy.decision"; - resourceId: string; - resourceType: string; - decision: "create" | "update" | "drift-update" | "drift-recreate" | "noop"; -}; - -export type ReconcilerDriftDetectedEvent = { - level: "info"; - event: "reconciler.drift.detected"; - resourceId: string; - resourceType: string; - diff: Record; -}; - -export type ReconcilerEvent = - | OperationLifecycleEvent - | ReconcilerDeployEvent - | ReconcilerDriftDetectedEvent - | MissingResourceRegistryMatchWarningEvent; - -export type ReconcilerEventEmitter = ( - event: ReconcilerEvent, -) => void | Promise; +import { toEmitStep, type ReconcilerEventEmitter } from "./events"; +import { createPlan } from "./planner"; +import { createStepRunner, runOperation } from "./step-runner"; + +export type { + ReconcilerDeployEvent, + ReconcilerDriftDetectedEvent, + ReconcilerEvent, + ReconcilerEventEmitter, +} from "./events"; +export { createStepRunner, runOperation } from "./step-runner"; export type ReconcilerState = Pick< State, @@ -92,6 +66,7 @@ export class Reconciler { readonly #defaultDryRun: boolean; readonly #defaultDriftDetection: boolean; readonly #emit?: ReconcilerEventEmitter; + readonly #emitFromStep?: EmitFromStep; readonly #maxOperationAttempts?: number; readonly #mutationLeaseTtl: number; readonly #stepRunner: StepRunner; @@ -102,6 +77,10 @@ export class Reconciler { this.#defaultDryRun = opts.dryRun ?? false; this.#defaultDriftDetection = opts.driftDetection ?? true; this.#emit = opts.emit; + // Everything the shared generator emits — decisions, drift and operation + // lifecycle alike — goes through one seam; in process it is a plain + // await, and the scope it is handed is ignored because nothing is keyed. + this.#emitFromStep = opts.emit ? emitDirectly(opts.emit) : undefined; this.#maxOperationAttempts = opts.maxOperationAttempts; this.#mutationLeaseTtl = opts.mutationLeaseTtl ?? 30_000; this.#stepRunner = createStepRunner(); @@ -130,36 +109,13 @@ export class Reconciler { } async plan(resources: BaseResource[], opts: PlanOptions = {}): Promise { - const driftDetection = opts.driftDetection ?? this.#defaultDriftDetection; - const resourceById = new Map( - resources.map((resource) => [resource.id, resource]), - ); - const nodes: PlanNode[] = []; - - const dependencyLevels = buildResourceDepthLevels(resources); - for (const level of dependencyLevels) { - for (const resource of level) { - nodes.push(await this.#planResource(resource, driftDetection)); - } - } - - const stateNodes = await this.#state.values(); - for (const stateNode of stateNodes) { - if (resourceById.has(stateNode.id)) continue; - - nodes.push({ - id: stateNode.id, - type: stateNode.type, - decision: "delete-orphan", - params: stateNode.params, - dependsOn: [], - }); - } - - return { - createdAt: new Date().toISOString(), - nodes, - }; + return createPlan({ + resources, + state: this.#state, + driftDetection: opts.driftDetection ?? this.#defaultDriftDetection, + emit: this.#emit, + maxOperationAttempts: this.#maxOperationAttempts, + }); } async destroy( @@ -167,6 +123,9 @@ export class Reconciler { opts: DestroyOptions = {}, ): Promise { const dryRun = opts.dryRun ?? this.#defaultDryRun; + const resourceById = new Map( + resources.map((resource) => [resource.id, resource]), + ); const dependencyLevels = buildResourceDepthLevels(resources); for ( @@ -179,6 +138,10 @@ export class Reconciler { level.map((resource) => this.#destroyResource(resource, dryRun)), ); } + + // Then the records that were never declared, so a destroy leaves the + // deployment empty rather than leaving orphans for a later refresh. + await this.#deleteOrphans(resources, resourceById, dryRun, "destroy"); } async refresh( @@ -198,11 +161,28 @@ export class Reconciler { dryRun: boolean, driftDetection: boolean, ) { - await this.#withMutationLease(resource.id, () => - this.#retryOnRevConflict((conflict) => - this.#deployResourceOnce(resource, dryRun, driftDetection, conflict), - ), - ); + await this.#withMutationLease(resource.id, async () => { + // Resolved once for the whole scheduled reconciliation, recovery + // included: deriveParams is user code and need not be deterministic, so + // a second resolution could decide against one set of params and + // persist another, with nothing to reconcile the two afterwards. + const params = (await resource.getParams()) as Record; + + await this.#retryOnRevConflict((conflict) => + runOperation( + reconcileResource(this.#stepRunner, { + resource, + resourceParams: params, + openSession: this.#openSession(), + emit: this.#emitFromStep, + dryRun, + driftDetection, + maxOperationAttempts: this.#maxOperationAttempts, + recoverFrom: conflict, + }), + ), + ); + }); } async #withMutationLease(resourceId: string, fn: () => Promise) { @@ -257,207 +237,34 @@ export class Reconciler { } } - async #deployResourceOnce( - resource: BaseResource, - dryRun: boolean, - driftDetection: boolean, - conflict?: RevConflict, - ) { - if (conflict) { - await this.#recoverDeployResource(resource, dryRun, conflict); - return; - } - - const stateNode = await this.#state.get(resource.id); - - let action: ResourceAction; - if (!stateNode) { - action = decideAction({ resource }); - } else { - resource.setOutput(stateNode.output); - const params = await resource.getParams(); - action = decideAction({ resource, stateNode, params }); - - if (action.decision === "noop" && driftDetection) { - const driftRead = await this.#readForDrift(resource); - action = decideAction({ resource, stateNode, params, driftRead }); - } - } - - if (action.decision === "drift-update") { - await this.#emit?.({ - level: "info", - event: "reconciler.drift.detected", - resourceId: resource.id, - resourceType: resource.type, - diff: action.patch, - }); - } - - await this.#emit?.({ - level: "info", - event: "reconciler.deploy.decision", - resourceId: resource.id, - resourceType: resource.type, - decision: action.decision, - }); - - switch (action.decision) { - case "create": - case "drift-recreate": - await runOperation( - createResourceOperation(this.#stepRunner, { - resource, - state: this.#state, - dryRun, - emit: this.#emit, - maxOperationAttempts: this.#maxOperationAttempts, - expectedRev: stateNode?.rev ?? 0, - }), - ); - return; - case "update": - case "drift-update": - // decideAction only returns update decisions for an existing stateNode - await runOperation( - updateResourceOperation(this.#stepRunner, { - resource, - state: this.#state, - patch: action.patch, - dryRun, - emit: this.#emit, - maxOperationAttempts: this.#maxOperationAttempts, - expectedRev: stateNode!.rev, - }), - ); - return; - case "noop": - return; - } - } - - async #recoverDeployResource( - resource: BaseResource, - dryRun: boolean, - conflict: RevConflict, - ) { - if (!resource.read) throw conflict; - - const stateNode = await this.#state.get(resource.id); - if (stateNode) resource.setOutput(stateNode.output); - - const params = await resource.getParams(); - const remote = await this.#readForDrift(resource); - const action = decideAction({ - resource, - stateNode, - params, - driftRead: remote, - }); - if (remote.kind === "present") resource.setOutput(remote.output); - - await this.#emit?.({ - level: "info", - event: "reconciler.deploy.decision", - resourceId: resource.id, - resourceType: resource.type, - decision: action.decision, - }); - - switch (action.decision) { - case "create": - case "drift-recreate": - await runOperation( - createResourceOperation(this.#stepRunner, { - resource, - state: this.#state, - dryRun, - emit: this.#emit, - maxOperationAttempts: this.#maxOperationAttempts, - expectedRev: stateNode?.rev ?? 0, - }), - ); - return; - case "update": - case "drift-update": - await runOperation( - updateResourceOperation(this.#stepRunner, { - resource, - state: this.#state, - patch: action.patch, - dryRun, - emit: this.#emit, - maxOperationAttempts: this.#maxOperationAttempts, - expectedRev: stateNode?.rev ?? 0, - }), - ); - return; - case "noop": - if (dryRun) return; - await this.#state.update(resource.id, stateNode?.rev ?? 0, { - id: resource.id, - groupId: resource.groupId, - groupType: resource.groupType, - type: resource.type, - lastOperation: "drift", - lastOperationAt: new Date().toISOString(), - config: resource.config, - params: resource.toState(params), - output: resource.toState(resource.output), - }); - return; - } - } - - async #planResource( - resource: BaseResource, - driftDetection: boolean, - ): Promise { - const stateNode = await this.#state.get(resource.id); - if (stateNode) { - resource.setOutput(stateNode.output); - } - - const params = await resolvePlanParams(resource); - let action = decideAction({ resource, stateNode, params }); - - if (action.decision === "noop" && driftDetection) { - const driftRead = await this.#readForDrift(resource); - action = decideAction({ resource, stateNode, params, driftRead }); - } - - return { - id: resource.id, - type: resource.type, - decision: action.decision, - ...("diff" in action ? { diff: action.diff } : {}), - params, - dependsOn: getDependencyIds(resource), + /** + * Reads a resource's record and binds the writes that are conditional on + * that read. In process the precondition is the revision read here, which + * the mutation lease keeps other writers away from. + */ + #openSession(): OpenStateSession { + const state = this.#state; + return async function* (resource: BaseResource) { + const node = await state.get(resource.id); + const expectedRev = node?.rev ?? 0; + const persist: PersistState = async function* (next) { + await state.update(resource.id, expectedRev, next); + }; + + if (!node) return { node: undefined, persist }; + + const remove: RemoveState = async function* () { + await state.delete(resource.id, node.rev); + }; + return { node, persist, remove }; }; } - async #readForDrift(resource: BaseResource): Promise { - try { - const output = await runOperation( - readResourceOperation(this.#stepRunner, { - resource, - state: this.#state, - emit: this.#emit, - maxOperationAttempts: this.#maxOperationAttempts, - }), - ); - return { kind: "present", output }; - } catch (error) { - if (ResourceNotFoundError.is(error)) return { kind: "absent" }; - throw error; - } - } - async #deleteOrphans( resources: BaseResource[], resourceById: Map, dryRun: boolean, - workflow: "deploy" | "refresh", + workflow: "deploy" | "refresh" | "destroy", ) { await this.#withLease("reconciler:orphan-deletion", async () => { const stateNodes = await this.#state.values(); @@ -481,23 +288,17 @@ export class Reconciler { continue; } + // Built from the listing, so its config is as of that read rather + // than of the lease taken below — the session re-reads and refreshes + // output, but not this. Config reaches nothing but deriveParams on + // the delete-recovery read, so a racing write can only make that read + // one revision stale; the removal itself is still conditional on the + // session's own read. + const orphanResource = hydrateResourceFromState(Resource, stateNode); await this.#withMutationLease(stateNode.id, () => - this.#retryOnRevConflict(async (conflict) => { - const currentNode = await this.#state.get(stateNode.id); - if (!currentNode) return; - - const orphanResource = hydrateResourceFromState( - Resource, - currentNode, - ); - - await this.#deleteResourceOnce( - orphanResource, - currentNode, - dryRun, - conflict, - ); - }), + this.#retryOnRevConflict((conflict) => + this.#deleteResource(orphanResource, dryRun, conflict), + ), ); } }); @@ -505,58 +306,42 @@ export class Reconciler { async #destroyResource(resource: BaseResource, dryRun: boolean) { await this.#withMutationLease(resource.id, () => - this.#retryOnRevConflict(async (conflict) => { - const stateNode = await this.#state.get(resource.id); - if (!stateNode) { - return; - } - - resource.setOutput(stateNode.output); - await this.#deleteResourceOnce(resource, stateNode, dryRun, conflict); - }), + this.#retryOnRevConflict((conflict) => + this.#deleteResource(resource, dryRun, conflict), + ), ); } - async #deleteResourceOnce( + async #deleteResource( resource: BaseResource, - stateNode: StateNode, dryRun: boolean, conflict?: RevConflict, ) { - if (conflict) { - if (!resource.read) throw conflict; - - const remote = await this.#readForDrift(resource); - if (remote.kind !== "present") { - if (!dryRun) await this.#state.delete(resource.id, stateNode.rev); - return; - } - resource.setOutput(remote.output); - } + // Deletion never needs the desired params, so they are resolved only for + // the recovery read, which is their sole consumer on this path. + const recoverFrom = conflict + ? { + conflict, + resourceParams: (await resource.getParams()) as Record< + string, + unknown + >, + } + : undefined; await runOperation( - deleteResourceOperation(this.#stepRunner, { + destroyResource(this.#stepRunner, { resource, - state: this.#state, + openSession: this.#openSession(), + emit: this.#emitFromStep, dryRun, - emit: this.#emit, maxOperationAttempts: this.#maxOperationAttempts, - expectedRev: stateNode.rev, + recoverFrom, }), ); } } -export async function runOperation( - operation: AsyncGenerator, -) { - let next = await operation.next(); - while (!next.done) { - next = await operation.next(); - } - return next.value; -} - function hydrateResourceFromState( Resource: new (opts: { id: string; @@ -572,31 +357,11 @@ function hydrateResourceFromState( return resource; } -export function createStepRunner(): StepRunner { - return { - async *run( - arg1: string | (() => T | Promise), - arg2?: () => T | Promise, - ): AsyncGenerator { - const fn = (typeof arg1 === "string" ? arg2 : arg1) as - (() => T | Promise) | undefined; - - if (!fn) { - throw new Error("Missing run function"); - } - - return await fn(); - }, - async *delay( - arg1: string | number, - arg2?: number, - ): AsyncGenerator { - const ms = typeof arg1 === "number" ? arg1 : arg2; - if (ms === undefined) { - throw new Error("Missing delay duration"); - } - - await new Promise((resolve) => setTimeout(resolve, ms)); - }, - }; +/** + * The in-process emit seam: a scope carries no meaning here, since nothing + * keys anything, so every scope delivers through the same emitter. + */ +function emitDirectly(emit: ReconcilerEventEmitter): EmitFromStep { + const step = toEmitStep(emit); + return () => step; } diff --git a/packages/reconciler/src/resource-registry.ts b/packages/reconciler/src/resource-registry.ts index 916c0ab..6916caa 100644 --- a/packages/reconciler/src/resource-registry.ts +++ b/packages/reconciler/src/resource-registry.ts @@ -6,7 +6,7 @@ export type MissingResourceRegistryMatchWarningEvent = { level: "warn"; event: "reconciler.orphan-deletion.skipped"; reason: "resource-type-not-registered"; - workflow: "deploy" | "refresh"; + workflow: "deploy" | "refresh" | "destroy"; resourceId: string; resourceType: ResourceType; }; @@ -46,7 +46,7 @@ export function resolveResourceClass( } export function createMissingResourceRegistryMatchWarningEvent(opts: { - workflow: "deploy" | "refresh"; + workflow: "deploy" | "refresh" | "destroy"; resourceId: string; resourceType: ResourceType; }): MissingResourceRegistryMatchWarningEvent { diff --git a/packages/reconciler/src/step-runner.ts b/packages/reconciler/src/step-runner.ts new file mode 100644 index 0000000..ff39fdc --- /dev/null +++ b/packages/reconciler/src/step-runner.ts @@ -0,0 +1,27 @@ +import type { StepRunner } from "./operations"; + +export async function runOperation( + operation: AsyncGenerator, +) { + let next = await operation.next(); + while (!next.done) { + next = await operation.next(); + } + return next.value; +} + +export function createStepRunner(): StepRunner { + const runner: StepRunner = { + async *run(_key: string, fn: () => T | Promise) { + return await fn(); + }, + async *delay(_key: string, ms: number) { + await new Promise((resolve) => setTimeout(resolve, ms)); + }, + // Nothing is replayed in process, so no step key is ever read and a scope + // has nothing to namespace: one runner serves every scope. + scope: () => runner, + }; + + return runner; +} diff --git a/packages/reconciler/test/durable-reconciliation.test.ts b/packages/reconciler/test/durable-reconciliation.test.ts new file mode 100644 index 0000000..516663b --- /dev/null +++ b/packages/reconciler/test/durable-reconciliation.test.ts @@ -0,0 +1,798 @@ +import { + WorkflowRunner, + type HeapClient, + type WorkflowEvent, +} from "@yieldstar/core"; +import { + SqliteHeapClient, + SqliteStoreClient, + createSqliteDb, +} from "@yieldstar/sqlite-runtime/node"; +import { + resource, + ResourceNotFoundError, + ResourceOperationPendingError, + type BaseResource, +} from "@notation/resource"; +import { setTimeout as sleep } from "node:timers/promises"; +import pino from "pino"; +import { createWorkflowRouter, workflow } from "yieldstar"; +import { describe, expect, it, vi } from "vitest"; +import * as durable from "../src/durable"; +import type { ReconcilerEvent } from "../src/events"; +import { + createResourceRegistry, + type ResourceRegistry, +} from "../src/resource-registry"; + +const logger = pino({ level: "silent" }); + +/** + * A retry delay has to outlive the heap write that follows it. The workflow + * loop continues inline for a delay that has already elapsed by the time it + * is reached, so a delay shorter than a SQLite write runs the retry in the + * same execution — which is the opposite of what these tests assert. + */ +const RETRY_AFTER_MS = 50; +const PAST_RETRY_MS = RETRY_AFTER_MS + 25; + +describe("durable execution and replay", () => { + it("waits durably for a retryable provider and persists after success", async () => { + let attempts = 0; + const PendingResource = resource({ type: "test/durable/pending" }) + .defineSchema({}) + .defineOperations({ + create: async (_params, context) => { + attempts += 1; + if (attempts === 1) { + expect(context).toBeUndefined(); + throw new ResourceOperationPendingError("provider is not ready", { + retryAfterMs: RETRY_AFTER_MS, + callbackContext: { requestId: "request-123" }, + }); + } + expect(context).toEqual({ requestId: "request-123" }); + }, + delete: async () => undefined, + }); + const runtime = createRuntime( + [new PendingResource({ id: "pending" })], + "durable-wait", + { maxOperationAttempts: 3 }, + ); + + await runtime.run("wait-execution"); + expect(attempts).toBe(1); + expect(runtime.scheduler.events).toHaveLength(1); + + await sleep(PAST_RETRY_MS); + await runtime.run("wait-execution"); + expect(attempts).toBe(2); + expect(await runtime.state.get("pending")).toMatchObject({ + id: "pending", + lastOperation: "create", + rev: 1, + }); + runtime.close(); + }); + + it("resumes after a crash following the create checkpoint", async () => { + const create = vi.fn(async () => undefined); + const TestResource = resource({ type: "test/durable/resume" }) + .defineSchema({}) + .defineOperations({ create, delete: async () => undefined }); + const runtime = createRuntime( + [new TestResource({ id: "resume" })], + "crash-resume", + { crashAfterStep: "notation:resource:resume:create:remote:attempt:0" }, + ); + + await expect(runtime.run("resume-execution")).rejects.toThrow( + "simulated process crash", + ); + expect(create).toHaveBeenCalledOnce(); + expect(await runtime.state.get("resume")).toBeUndefined(); + + await runtime.run("resume-execution"); + expect(create).toHaveBeenCalledOnce(); + expect(await runtime.state.get("resume")).toMatchObject({ rev: 1 }); + runtime.close(); + }); + + it("resumes after a crash following the delete checkpoint", async () => { + const remove = vi.fn(async () => undefined); + const TestResource = resource({ type: "test/durable/destroy-resume" }) + .defineSchema({}) + .defineOperations({ create: async () => undefined, delete: remove }); + const runtime = createRuntime( + [new TestResource({ id: "destroyed" })], + "destroy-crash-resume", + { crashAfterStep: "notation:destroy:destroyed:delete:remote:attempt:0" }, + ); + + await runtime.run("deploy-before-destroy"); + await expect(runtime.destroy("destroy-execution")).rejects.toThrow( + "simulated process crash", + ); + expect(remove).toHaveBeenCalledOnce(); + expect(await runtime.state.get("destroyed")).toBeDefined(); + + await runtime.destroy("destroy-execution"); + expect(remove).toHaveBeenCalledOnce(); + expect(await runtime.state.get("destroyed")).toBeUndefined(); + runtime.close(); + }); + + it("waits durably for a retryable delete before removing state", async () => { + let attempts = 0; + const PendingDelete = resource({ type: "test/durable/pending-delete" }) + .defineSchema({}) + .defineOperations({ + create: async () => undefined, + delete: async () => { + attempts += 1; + if (attempts === 1) { + throw new ResourceOperationPendingError("delete is not ready", { + retryAfterMs: RETRY_AFTER_MS, + }); + } + }, + }); + const runtime = createRuntime( + [new PendingDelete({ id: "pending-delete" })], + "durable-destroy-wait", + { maxOperationAttempts: 3 }, + ); + + await runtime.run("deploy-before-wait"); + await runtime.destroy("destroy-wait"); + expect(attempts).toBe(1); + expect(await runtime.state.get("pending-delete")).toBeDefined(); + + await sleep(PAST_RETRY_MS); + await runtime.destroy("destroy-wait"); + expect(attempts).toBe(2); + expect(await runtime.state.get("pending-delete")).toBeUndefined(); + runtime.close(); + }); + + it("waits when a resource reports that its post-write read is pending", async () => { + let reads = 0; + const EventuallyReadable = resource({ + type: "test/durable/eventually-readable", + }) + .defineSchema({}) + .defineOperations({ + create: async () => undefined, + read: async () => { + reads += 1; + if (reads === 1) { + throw new ResourceOperationPendingError( + "resource is not visible yet", + { retryAfterMs: RETRY_AFTER_MS }, + ); + } + return {} as const; + }, + delete: async () => undefined, + }); + const runtime = createRuntime( + [new EventuallyReadable({ id: "eventually-readable" })], + "post-write-read", + { maxOperationAttempts: 3 }, + ); + + await runtime.run("post-write-read-execution"); + expect(reads).toBe(1); + expect(await runtime.state.get("eventually-readable")).toBeUndefined(); + + await sleep(PAST_RETRY_MS); + await runtime.run("post-write-read-execution"); + expect(reads).toBe(2); + expect(await runtime.state.get("eventually-readable")).toMatchObject({ + rev: 1, + }); + runtime.close(); + }); + + it("does not infer that not-found after a write is pending", async () => { + const MissingAfterCreate = resource({ + type: "test/durable/missing-after-create", + }) + .defineSchema({}) + .defineOperations({ + create: async () => undefined, + read: async () => { + throw new ResourceNotFoundError("resource is absent"); + }, + delete: async () => undefined, + }); + const runtime = createRuntime( + [new MissingAfterCreate({ id: "missing-after-create" })], + "missing-after-create", + ); + + await expect(runtime.run("missing-after-create-execution")).rejects.toThrow( + "resource is absent", + ); + expect(await runtime.state.get("missing-after-create")).toBeUndefined(); + runtime.close(); + }); +}); + +describe("dependency ordering", () => { + it("destroys dependents before their dependencies", async () => { + const order: string[] = []; + const Dependency = resource({ type: "test/durable/dependency" }) + .defineSchema({}) + .defineOperations({ + create: async () => undefined, + delete: async () => void order.push("dependency"), + }); + const Dependent = resource({ type: "test/durable/dependent" }) + .defineSchema({}) + .defineOperations({ + create: async () => undefined, + delete: async () => void order.push("dependent"), + }); + const dependency = new Dependency({ id: "dependency" }); + const dependent = new Dependent({ + id: "dependent", + dependencies: { dependency }, + }); + const runtime = createRuntime([dependency, dependent], "destroy-order"); + + await runtime.run("deploy-before-ordered-destroy"); + await runtime.destroy("ordered-destroy"); + + expect(order).toEqual(["dependent", "dependency"]); + runtime.close(); + }); +}); + +describe("conditional state persistence", () => { + it("rejects a state write whose snapshot another writer has moved past", async () => { + const RaceResource = resource({ type: "test/durable/write-race" }) + // Cast: a resource declared without API types constrains every schema + // key to be a key of an `any` API schema, which no named key satisfies. + .defineSchema({ + name: { presence: "required", propertyType: "param" }, + } as any) + .defineOperations({ + create: async () => undefined, + // Moves the store on between the workflow reading its snapshot and + // persisting against it, which is what the conditional write guards. + update: async () => { + await runtime.storeClient.updateStore({ + definition: durable.resourceStateStore, + id: runtime.state.storeId("raced"), + updater: (draft: any) => { + draft.lastOperationAt = "1999-01-01T00:00:00.000Z"; + }, + }); + }, + delete: async () => undefined, + }); + const resources = [ + new RaceResource({ id: "raced", config: { name: "before" } }), + ]; + const runtime = createRuntime(resources, "write-race"); + + await runtime.run("deploy-1"); + resources[0] = new RaceResource({ id: "raced", config: { name: "after" } }); + + await expect(runtime.run("deploy-2")).rejects.toMatchObject({ + name: "RevConflict", + }); + // The losing write left the other writer's record intact. + expect(await runtime.state.get("raced")).toMatchObject({ + rev: 2, + lastOperationAt: "1999-01-01T00:00:00.000Z", + }); + runtime.close(); + }); + + it("rejects a state removal whose snapshot another writer has moved past", async () => { + const RaceResource = resource({ type: "test/durable/delete-race" }) + .defineSchema({}) + .defineOperations({ + create: async () => undefined, + delete: async () => { + await runtime.storeClient.updateStore({ + definition: durable.resourceStateStore, + id: runtime.state.storeId("delete-raced"), + updater: (draft: any) => { + draft.lastOperationAt = "1999-01-01T00:00:00.000Z"; + }, + }); + }, + }); + const runtime = createRuntime( + [new RaceResource({ id: "delete-raced" })], + "delete-race", + ); + + await runtime.run("deploy-1"); + await expect(runtime.destroy("destroy-1")).rejects.toMatchObject({ + name: "RevConflict", + }); + // State survives a removal that could not be proven safe. + expect(await runtime.state.get("delete-raced")).toBeDefined(); + runtime.close(); + }); +}); + +describe("deployment coordination", () => { + it("serializes concurrent deployments through durable store waiting", async () => { + let unblockCreate!: () => void; + const blocked = new Promise((resolve) => { + unblockCreate = resolve; + }); + let started!: () => void; + const createStarted = new Promise((resolve) => { + started = resolve; + }); + const create = vi.fn(async () => { + started(); + await blocked; + }); + const TestResource = resource({ type: "test/durable/concurrent" }) + .defineSchema({}) + .defineOperations({ create, delete: async () => undefined }); + const runtime = createRuntime( + [new TestResource({ id: "shared" })], + "concurrent", + ); + + const first = runtime.run("deployment-a"); + await createStarted; + await runtime.run("deployment-b"); + expect(create).toHaveBeenCalledOnce(); + + unblockCreate(); + await first; + const wake = runtime.scheduler.events.find( + (event) => event.executionId === "deployment-b", + ); + expect(wake).toBeDefined(); + await runtime.runner.run(wake!, logger); + + expect(create).toHaveBeenCalledOnce(); + expect(await runtime.state.values()).toHaveLength(1); + runtime.close(); + }); + + it("still holds the deployment when a failed execution is resumed", async () => { + // The failure has to live in plain generator code. A step that fails + // caches a StepError, and a cached StepError is rethrown on replay before + // the step's function is reached, so no later work would ever run + // uncached. decideAction calls toComparable outside any step, after the + // resource's reads have been checkpointed. + let failBeforeSecond = true; + const holders: Array = []; + const Resource = resource({ type: "test/durable/hold-replay" }) + .defineSchema({}) + .defineOperations({ + create: async () => { + const snapshot = await runtime.storeClient.getStore({ + definition: durable.deploymentCoordinationStore, + id: "hold-replay", + }); + holders.push(snapshot.state.holder); + }, + delete: async () => undefined, + }); + + const first = new Resource({ id: "first" }); + const second = new Resource({ id: "second" }); + const toComparable = second.toComparable.bind(second); + second.toComparable = (output) => { + if (failBeforeSecond) throw new Error("simulated mid-deployment failure"); + return toComparable(output); + }; + const runtime = createRuntime([first, second], "hold-replay"); + + await expect(runtime.run("replayed-execution")).rejects.toThrow( + "simulated mid-deployment failure", + ); + expect(holders).toEqual(["replayed-execution"]); + + failBeforeSecond = false; + await runtime.run("replayed-execution"); + + // The second resource's create is the first uncached work after the + // failure, so it observes whichever hold the resumed execution is running + // under. Releasing on the way out of a failure would leave it null here. + expect(holders).toEqual(["replayed-execution", "replayed-execution"]); + runtime.close(); + }); + + it("emits a coordination waiting event when another execution holds the deployment", async () => { + let unblockCreate!: () => void; + const blocked = new Promise((resolve) => { + unblockCreate = resolve; + }); + let started!: () => void; + const createStarted = new Promise((resolve) => { + started = resolve; + }); + const TestResource = resource({ type: "test/durable/coordination" }) + .defineSchema({}) + .defineOperations({ + create: async () => { + started(); + await blocked; + }, + delete: async () => undefined, + }); + const events: ReconcilerEvent[] = []; + const runtime = createRuntime( + [new TestResource({ id: "held" })], + "coordination-waiting", + { emit: (event) => void events.push(event) }, + ); + + const first = runtime.run("holder-execution"); + await createStarted; + await runtime.run("waiter-execution"); + + expect( + events.find((event) => event.event === "reconciler.coordination.waiting"), + ).toMatchObject({ + level: "warn", + deploymentId: "coordination-waiting", + executionId: "waiter-execution", + holderExecutionId: "holder-execution", + }); + + unblockCreate(); + await first; + runtime.close(); + }); +}); + +describe("deployment hold takeover", () => { + it("clears a hold its named holder still has, and unblocks the deployment", async () => { + const create = vi.fn(async () => undefined); + const Resource = resource({ type: "test/durable/takeover" }) + .defineSchema({}) + .defineOperations({ create, delete: async () => undefined }); + const runtime = createRuntime([new Resource({ id: "held" })], "takeover"); + await runtime.storeClient.getOrCreateStore({ + definition: durable.deploymentCoordinationStore, + id: "takeover", + initial: { holder: "abandoned-execution" }, + }); + + const result = await durable.takeOverDeploymentHold({ + storeClient: runtime.storeClient, + deploymentId: "takeover", + fromExecutionId: "abandoned-execution", + }); + + expect(result).toEqual({ + taken: true, + previousHolder: "abandoned-execution", + }); + await runtime.run("later-execution"); + expect(create).toHaveBeenCalledOnce(); + runtime.close(); + }); + + it("refuses to clear a hold that has moved to another execution", async () => { + const runtime = createRuntime([], "takeover-race"); + await runtime.storeClient.getOrCreateStore({ + definition: durable.deploymentCoordinationStore, + id: "takeover-race", + initial: { holder: "current-execution" }, + }); + + const result = await durable.takeOverDeploymentHold({ + storeClient: runtime.storeClient, + deploymentId: "takeover-race", + fromExecutionId: "abandoned-execution", + }); + + expect(result).toEqual({ taken: false, holder: "current-execution" }); + const snapshot = await runtime.storeClient.getStore({ + definition: durable.deploymentCoordinationStore, + id: "takeover-race", + }); + expect(snapshot.state.holder).toBe("current-execution"); + runtime.close(); + }); +}); + +describe("deployment scoping", () => { + it("scopes store listing to the exact deployment despite prefix-like IDs", async () => { + const database = createSqliteDb({ path: ":memory:" }); + const storeClient = new SqliteStoreClient({ + db: database, + schedulerClient: new TestScheduler(), + }); + const app = new durable.DurableStateBackend(storeClient, "app"); + const appBlue = new durable.DurableStateBackend(storeClient, "app:blue"); + + await seedResourceState(storeClient, app.storeId("site"), "site"); + await seedResourceState(storeClient, appBlue.storeId("site"), "blue-site"); + + // A deployment named "app:blue" falls inside a naive "app:" prefix scan; + // encoding the deployment id is what keeps the two listings disjoint. + expect((await app.values()).map((node) => node.id)).toEqual(["site"]); + expect((await appBlue.values()).map((node) => node.id)).toEqual([ + "blue-site", + ]); + expect(await app.get("site")).toMatchObject({ id: "site" }); + expect(await appBlue.get("site")).toMatchObject({ id: "blue-site" }); + database.close(); + }); +}); + +describe("orphan deletion", () => { + it("deletes orphaned resources through the registry on a later deployment", async () => { + const deleteSpy = vi.fn(async () => undefined); + const OrphanResource = resource({ type: "test/durable/orphan" }) + .defineSchema({}) + .defineOperations({ create: async () => undefined, delete: deleteSpy }); + const resources: BaseResource[] = [new OrphanResource({ id: "orphan" })]; + const runtime = createRuntime(resources, "orphan-deletion", { + registry: createResourceRegistry([OrphanResource]), + }); + + await runtime.run("deploy-1"); + expect(await runtime.state.values()).toHaveLength(1); + + resources.length = 0; + await runtime.run("deploy-2"); + + expect(deleteSpy).toHaveBeenCalledOnce(); + expect(await runtime.state.values()).toHaveLength(0); + expect(await runtime.state.get("orphan")).toBeUndefined(); + runtime.close(); + }); +}); + +describe("drift detection and repair", () => { + it("emits drift detection and repairs remote drift with update", async () => { + let remote = { name: "expected" }; + const updateSpy = vi.fn(async () => { + remote = { name: "expected" }; + }); + const DriftResource = resource({ type: "test/durable/drift" }) + // Cast: a resource declared without API types constrains every schema + // key to be a key of an `any` API schema, which no named key satisfies. + .defineSchema({ + name: { presence: "required", propertyType: "param" }, + } as any) + .defineOperations({ + create: (async () => remote) as any, + read: async () => remote, + update: updateSpy, + delete: async () => undefined, + }); + const events: ReconcilerEvent[] = []; + const runtime = createRuntime( + [new DriftResource({ id: "drifted", config: { name: "expected" } })], + "drift-repair", + { driftDetection: true, emit: (event) => void events.push(event) }, + ); + + await runtime.run("deploy-1"); + remote = { name: "drifted" }; + await runtime.run("deploy-2"); + + expect(updateSpy).toHaveBeenCalledOnce(); + expect( + events.find((event) => event.event === "reconciler.drift.detected"), + ).toMatchObject({ resourceId: "drifted", diff: { name: "expected" } }); + expect( + events.filter( + (event) => + event.event === "reconciler.deploy.decision" && + event.decision === "drift-update", + ), + ).toHaveLength(1); + runtime.close(); + }); + + it("reads the remote during a dry run rather than reporting false drift", async () => { + const remote = { name: "expected" }; + const read = vi.fn(async () => remote); + const DryRunDriftResource = resource({ type: "test/durable/dry-run-drift" }) + .defineSchema({ + name: { presence: "required", propertyType: "param" }, + } as any) + .defineOperations({ + create: (async () => remote) as any, + read, + update: async () => undefined, + delete: async () => undefined, + }); + const events: ReconcilerEvent[] = []; + const options = { + driftDetection: true, + dryRun: false, + emit: (event: ReconcilerEvent) => void events.push(event), + }; + const runtime = createRuntime( + [new DryRunDriftResource({ id: "steady", config: { name: "expected" } })], + "dry-run-drift", + options, + ); + + await runtime.run("deploy-1"); + const readsAfterDeploy = read.mock.calls.length; + + options.dryRun = true; + events.length = 0; + await runtime.run("dry-run"); + + // Suppressing the drift read under dryRun would leave decideAction + // diffing an empty read against the desired params, which reports every + // param as drift; skipping it entirely would make a dry run unable to + // report the drift it exists to report. + expect(read.mock.calls.length).toBeGreaterThan(readsAfterDeploy); + expect( + events.filter((event) => event.event === "reconciler.drift.detected"), + ).toEqual([]); + expect( + events.find((event) => event.event === "reconciler.deploy.decision"), + ).toMatchObject({ resourceId: "steady", decision: "noop" }); + runtime.close(); + }); +}); + +function createRuntime( + resources: BaseResource[], + deploymentId: string, + options: { + maxOperationAttempts?: number; + crashAfterStep?: string; + registry?: ResourceRegistry; + driftDetection?: boolean; + dryRun?: boolean; + emit?: (event: ReconcilerEvent) => void; + } = {}, +) { + const database = createSqliteDb({ path: ":memory:" }); + const scheduler = new TestScheduler(); + const sqliteHeap = new SqliteHeapClient(database); + const heap = options.crashAfterStep + ? new CrashAfterWriteHeap(sqliteHeap, options.crashAfterStep) + : sqliteHeap; + const storeClient = new SqliteStoreClient({ + db: database, + schedulerClient: scheduler, + }); + const state = new durable.DurableStateBackend(storeClient, deploymentId); + const deploy = workflow(async function* (step, event) { + yield* durable.deploy(step, { + deploymentId, + executionId: event.executionId, + resources, + state, + registry: options.registry, + driftDetection: options.driftDetection ?? false, + // Read at execution time, so a test can switch it between runs. + dryRun: options.dryRun, + emit: options.emit, + maxOperationAttempts: options.maxOperationAttempts, + }); + }); + const destroy = workflow(async function* (step, event) { + yield* durable.destroy(step, { + deploymentId, + executionId: event.executionId, + resources, + state, + registry: options.registry, + emit: options.emit, + maxOperationAttempts: options.maxOperationAttempts, + }); + }); + const router = createWorkflowRouter({ deploy, destroy }); + const runner = new WorkflowRunner({ + router, + heapClient: heap, + storeClient, + schedulerClient: scheduler, + logger, + }); + + return { + runner, + scheduler, + state, + storeClient, + run(executionId: string) { + return runner.run( + { + workflowId: "deploy", + executionId, + params: {}, + context: new Map(), + }, + logger, + ); + }, + destroy(executionId: string) { + return runner.run( + { + workflowId: "destroy", + executionId, + params: {}, + context: new Map(), + }, + logger, + ); + }, + close() { + database.close(); + }, + }; +} + +class TestScheduler { + readonly events: WorkflowEvent[] = []; + + async requestWakeUp(event: WorkflowEvent) { + this.events.push(event); + } +} + +class CrashAfterWriteHeap implements HeapClient { + #crashed = false; + + constructor( + private readonly inner: HeapClient, + private readonly crashAfterStep: string, + ) {} + + readStep(params: { executionId: string; stepKey: string }) { + return this.inner.readStep(params); + } + + async writeStep(params: { + executionId: string; + stepKey: string; + stepAttempt: number; + stepDone: boolean; + stepResponseJson: string; + }) { + await this.inner.writeStep(params); + if ( + !this.#crashed && + params.stepKey === this.crashAfterStep && + params.stepDone + ) { + this.#crashed = true; + throw new Error("simulated process crash"); + } + } +} + +function seedResourceState( + storeClient: SqliteStoreClient, + storeId: string, + resourceId: string, +) { + return storeClient.getOrCreateStore({ + definition: durable.resourceStateStore, + id: storeId, + initial: statePatch(resourceId), + }); +} + +function statePatch(id: string) { + return { + id, + type: "test/durable/state", + groupId: -1, + groupType: "", + config: {}, + params: {}, + output: {}, + lastOperation: "create" as const, + lastOperationAt: new Date().toISOString(), + }; +} diff --git a/packages/reconciler/test/operation.workflows.test.ts b/packages/reconciler/test/operation.workflows.test.ts index 2afc6df..1d4c410 100644 --- a/packages/reconciler/test/operation.workflows.test.ts +++ b/packages/reconciler/test/operation.workflows.test.ts @@ -11,32 +11,31 @@ import { type OperationLifecycleEvent, type StepRunner, } from "../src/operations"; +import { toEmitStep } from "../src/events"; -function createStepRunnerDouble(): StepRunner { +function createStepRunnerDouble() { const run = vi.fn(async function* ( - arg1: string | (() => T | Promise), - arg2?: () => T | Promise, + _key: string, + fn: () => T | Promise, ): AsyncGenerator { - const fn = (typeof arg1 === "string" ? arg2 : arg1) as () => T | Promise; - if (!fn) { - throw new Error("Missing run function"); - } - return await fn(); }); - const delay = vi.fn(async function* (): AsyncGenerator< - unknown, - void, - unknown - > { + const delay = vi.fn(async function* ( + _key: string, + _ms: number, + ): AsyncGenerator { return; }); - return { - run, - delay, + // vi.fn erases the generic, so the seam's signature is restored here. + const runner: StepRunner = { + run: run as unknown as StepRunner["run"], + delay: delay as unknown as StepRunner["delay"], + scope: () => runner, }; + + return runner; } async function runOperation(operation: AsyncGenerator) { @@ -51,14 +50,10 @@ describe("operation workflows", () => { it("create performs create + read-after-create + state persistence", async () => { const step = createStepRunnerDouble(); const events: OperationLifecycleEvent[] = []; - const state = { - get: vi.fn(async () => undefined), - update: vi.fn(async () => undefined), - delete: vi.fn(async () => undefined), - }; + const persist = vi.fn(async function* () {}); let createAttempts = 0; - const createMock = vi.fn(async (_params, context) => { + const createMock = vi.fn(async (_params: unknown, context: unknown) => { createAttempts += 1; if (createAttempts === 1) { expect(context).toBeUndefined(); @@ -75,7 +70,8 @@ describe("operation workflows", () => { const TestResource = resource({ type: "test/service/create" }) .defineSchema({}) .defineOperations({ - create: createMock, + // Cast: an empty schema declares no primary key to return. + create: createMock as any, read: async () => ({ remoteId: "abc", status: "ready" }), delete: async () => undefined, }); @@ -85,16 +81,26 @@ describe("operation workflows", () => { await runOperation( createResourceOperation(step, { resource: testResource, - state, - expectedRev: 0, - emit: async (event) => { - events.push(event); - }, + resourceParams: await testResource.getParams(), + persist, + emit: toEmitStep((event) => void events.push(event)), }), ); expect(createAttempts).toBe(2); - expect(state.update).toHaveBeenCalledOnce(); + expect(persist).toHaveBeenCalledOnce(); + expect(persist).toHaveBeenCalledWith({ + id: "test-create", + groupId: testResource.groupId, + groupType: testResource.groupType, + type: TestResource.type, + lastOperation: "create", + lastOperationAt: expect.any(String), + config: testResource.config, + params: {}, + // The resource declares no schema, so nothing survives toState. + output: {}, + }); expect(createMock).toHaveBeenNthCalledWith( 1, await testResource.getParams(), @@ -118,17 +124,12 @@ describe("operation workflows", () => { it("read follows pending retry instructions", async () => { const step = createStepRunnerDouble(); - const state = { - get: vi.fn(async () => undefined), - update: vi.fn(async () => undefined), - delete: vi.fn(async () => undefined), - }; let readAttempts = 0; const TestResource = resource({ type: "test/service/read" }) .defineSchema({}) .defineOperations({ - create: async () => ({}), + create: (async () => ({})) as any, read: async (_key, context) => { readAttempts += 1; if (readAttempts < 3) { @@ -148,7 +149,7 @@ describe("operation workflows", () => { const result = await runOperation( readResourceOperation(step, { resource: testResource, - state, + resourceParams: await testResource.getParams(), }), ); @@ -168,11 +169,6 @@ describe("operation workflows", () => { it("fails when an operation remains pending past the safety limit", async () => { const step = createStepRunnerDouble(); - const state = { - get: vi.fn(async () => undefined), - update: vi.fn(async () => undefined), - delete: vi.fn(async () => undefined), - }; const read = vi.fn(async () => { throw new ResourceOperationPendingError("still pending", { retryAfterMs: 10, @@ -181,7 +177,7 @@ describe("operation workflows", () => { const TestResource = resource({ type: "test/service/pending-limit" }) .defineSchema({}) .defineOperations({ - create: async () => ({}), + create: (async () => ({})) as any, read, delete: async () => undefined, }); @@ -190,7 +186,7 @@ describe("operation workflows", () => { runOperation( readResourceOperation(step, { resource: new TestResource({ id: "pending-limit" }), - state, + resourceParams: {}, maxOperationAttempts: 2, }), ), @@ -201,15 +197,11 @@ describe("operation workflows", () => { it("does not infer that not-found after creation is retryable", async () => { const step = createStepRunnerDouble(); - const state = { - get: vi.fn(async () => undefined), - update: vi.fn(async () => undefined), - delete: vi.fn(async () => undefined), - }; + const persist = vi.fn(async function* () {}); const TestResource = resource({ type: "test/service/eventually-visible" }) .defineSchema({}) .defineOperations({ - create: async () => ({}), + create: (async () => ({})) as any, read: async () => { throw new ResourceNotFoundError("resource is absent"); }, @@ -220,27 +212,23 @@ describe("operation workflows", () => { runOperation( createResourceOperation(step, { resource: new TestResource({ id: "eventually-visible" }), - state, - expectedRev: 0, + resourceParams: {}, + persist, }), ), ).rejects.toThrowError("resource is absent"); - expect(state.update).not.toHaveBeenCalled(); + expect(persist).not.toHaveBeenCalled(); }); it("delete treats an already-absent remote as success through its idempotent resource contract", async () => { const step = createStepRunnerDouble(); const events: OperationLifecycleEvent[] = []; - const state = { - get: vi.fn(async () => undefined), - update: vi.fn(async () => undefined), - delete: vi.fn(async () => undefined), - }; + const remove = vi.fn(async function* () {}); const TestResource = resource({ type: "test/service/delete" }) .defineSchema({}) .defineOperations({ - create: async () => ({}), + create: (async () => ({})) as any, delete: async () => undefined, }); @@ -249,30 +237,25 @@ describe("operation workflows", () => { await runOperation( deleteResourceOperation(step, { resource: testResource, - state, - expectedRev: 1, - emit: async (event) => { - events.push(event); - }, + remove, + emit: toEmitStep((event) => void events.push(event)), }), ); - expect(state.delete).toHaveBeenCalledWith("test-delete", 1); + // State is removed only after the provider delete resolves; which record + // and revision that targets is the driver's concern, not the operation's. + expect(remove).toHaveBeenCalledOnce(); expect(events.map((event) => event.status)).toEqual(["start", "success"]); }); it("delete rethrows an unclassified resource error", async () => { const step = createStepRunnerDouble(); - const state = { - get: vi.fn(async () => undefined), - update: vi.fn(async () => undefined), - delete: vi.fn(async () => undefined), - }; + const remove = vi.fn(async function* () {}); const TestResource = resource({ type: "test/service/delete-miss" }) .defineSchema({}) .defineOperations({ - create: async () => ({}), + create: (async () => ({})) as any, delete: async () => { const err = new Error("still exists"); err.name = "DifferentError"; @@ -286,8 +269,7 @@ describe("operation workflows", () => { runOperation( deleteResourceOperation(step, { resource: testResource, - state, - expectedRev: 1, + remove, }), ), ).rejects.toMatchObject({ @@ -295,17 +277,13 @@ describe("operation workflows", () => { message: "still exists", }); - expect(state.delete).not.toHaveBeenCalled(); + expect(remove).not.toHaveBeenCalled(); }); it("emits structured error details on operation failure", async () => { const step = createStepRunnerDouble(); const events: OperationLifecycleEvent[] = []; - const state = { - get: vi.fn(async () => undefined), - update: vi.fn(async () => undefined), - delete: vi.fn(async () => undefined), - }; + const persist = vi.fn(async function* () {}); const TestResource = resource({ type: "test/service/create-error" }) .defineSchema({}) @@ -324,11 +302,9 @@ describe("operation workflows", () => { runOperation( createResourceOperation(step, { resource: testResource, - state, - expectedRev: 0, - emit: async (event) => { - events.push(event); - }, + resourceParams: {}, + persist, + emit: toEmitStep((event) => void events.push(event)), }), ), ).rejects.toMatchObject({ name: "CreateFailed", message: "boom" }); diff --git a/packages/reconciler/test/planner.test.ts b/packages/reconciler/test/planner.test.ts new file mode 100644 index 0000000..8066578 --- /dev/null +++ b/packages/reconciler/test/planner.test.ts @@ -0,0 +1,142 @@ +import { + ResourceNotFoundError, + ResourceOperationPendingError, + resource, +} from "@notation/resource"; +import { MemoryStateBackend } from "@notation/state"; +import { describe, expect, it } from "vitest"; +import { createPlan } from "../src/planner"; + +describe("createPlan", () => { + it("plans desired creates and persisted orphans without mutation execution", async () => { + const TestResource = resource({ type: "test/planner/resource" }) + .defineSchema({}) + .defineOperations({ + create: async () => undefined, + delete: async () => undefined, + }); + const state = new MemoryStateBackend(); + await state.update("orphan", 0, { + id: "orphan", + type: TestResource.type, + config: {}, + params: {}, + output: {}, + lastOperation: "create", + lastOperationAt: "2026-07-22T00:00:00.000Z", + }); + + const plan = await createPlan({ + resources: [new TestResource({ id: "desired" })], + state, + driftDetection: false, + }); + + expect(plan.nodes).toEqual([ + expect.objectContaining({ id: "desired", decision: "create" }), + expect.objectContaining({ id: "orphan", decision: "delete-orphan" }), + ]); + }); + + it("propagates unexpected read failures", async () => { + const TestResource = resource({ type: "test/planner/read-failure" }) + .defineSchema({}) + .defineOperations({ + create: async () => undefined, + read: async () => { + throw new Error("access denied"); + }, + delete: async () => undefined, + }); + const state = new MemoryStateBackend(); + await state.update("existing", 0, { + id: "existing", + type: TestResource.type, + config: {}, + params: {}, + output: {}, + lastOperation: "create", + lastOperationAt: "2026-07-22T00:00:00.000Z", + }); + + await expect( + createPlan({ + resources: [new TestResource({ id: "existing" })], + state, + }), + ).rejects.toThrow("access denied"); + }); + + it("plans recreation when the resource reports absence", async () => { + const TestResource = resource({ type: "test/planner/absent" }) + .defineSchema({}) + .defineOperations({ + create: async () => undefined, + read: async () => { + throw new ResourceNotFoundError("resource is absent"); + }, + delete: async () => undefined, + }); + const state = new MemoryStateBackend(); + await state.update("existing", 0, { + id: "existing", + type: TestResource.type, + config: {}, + params: {}, + output: {}, + lastOperation: "create", + lastOperationAt: "2026-07-22T00:00:00.000Z", + }); + + const plan = await createPlan({ + resources: [new TestResource({ id: "existing" })], + state, + }); + + expect(plan.nodes[0]).toMatchObject({ + id: "existing", + decision: "drift-recreate", + }); + }); + + it("waits for a pending read before planning", async () => { + let attempts = 0; + const TestResource = resource({ type: "test/planner/pending" }) + .defineSchema({}) + .defineOperations({ + create: async () => undefined, + read: async () => { + attempts += 1; + if (attempts === 1) { + throw new ResourceOperationPendingError( + "Waiting for the provider", + { retryAfterMs: 0 }, + ); + } + return {}; + }, + delete: async () => undefined, + }); + const state = new MemoryStateBackend(); + await state.update("existing", 0, { + id: "existing", + type: TestResource.type, + config: {}, + params: {}, + output: {}, + lastOperation: "create", + lastOperationAt: "2026-07-22T00:00:00.000Z", + }); + + const plan = await createPlan({ + resources: [new TestResource({ id: "existing" })], + state, + }); + + expect(plan.nodes[0]).toMatchObject({ + id: "existing", + decision: "noop", + }); + expect(attempts).toBe(2); + }); +}); diff --git a/packages/reconciler/test/reconciler.deploy.test.ts b/packages/reconciler/test/reconciler.deploy.test.ts index eaa5ce5..61c328b 100644 --- a/packages/reconciler/test/reconciler.deploy.test.ts +++ b/packages/reconciler/test/reconciler.deploy.test.ts @@ -72,15 +72,15 @@ function createTestResourceClass(opts: { ) => Promise; }) { return resource({ type: opts.type }) + // Cast: a resource declared without API types constrains every schema key + // to be a key of an `any` API schema, which no named key can satisfy. .defineSchema({ - name: { - presence: "required", - propertyType: "param", - valueType: "string" as any, - }, - }) + name: { presence: "required", propertyType: "param" }, + } as any) .defineOperations({ - create: opts.create ?? (async () => ({})), + // Cast: with no API types the schema resolves to no primary key, so the + // inferred create signature returns void. + create: (opts.create ?? (async () => ({}))) as any, read: opts.read, update: opts.update, delete: opts.delete ?? (async () => undefined), @@ -93,7 +93,9 @@ const found = (output: Record) => output; describe("reconciler deploy", () => { it("chooses create vs update from desired params vs state", async () => { const createSpy = vi.fn(async () => ({ name: "new" })); - const updateSpy = vi.fn(async () => undefined); + const updateSpy = vi.fn( + async (_key: unknown, _patch: Record) => undefined, + ); const CreateResource = createTestResourceClass({ type: "test/service/create-choice", @@ -261,6 +263,79 @@ describe("reconciler deploy", () => { }); }); + it("reports the drift it adopts when recovering from a conflict", async () => { + // The provider ignored the update, so the remote still holds the old + // value while another writer has moved persisted state to the desired + // one: recovery re-decides as drift rather than as a repeat update. + const readSpy = vi.fn(async () => found({ name: "stale" })); + const updateSpy = vi.fn( + async (_key: unknown, _patch: Record) => undefined, + ); + const UpdateResource = createTestResourceClass({ + type: "test/service/recovery-drift", + read: readSpy, + update: updateSpy, + }); + const state = createMemoryState({ + existing: { + rev: 1, + id: "existing", + groupId: -1, + groupType: "", + type: UpdateResource.type, + config: { name: "old" }, + params: { name: "old" }, + output: { name: "old" }, + lastOperation: "create", + lastOperationAt: new Date().toISOString(), + }, + }); + const updateState = state.update.getMockImplementation()!; + state.update + .mockImplementationOnce(async () => { + state.store.existing = { + ...state.store.existing!, + rev: 2, + config: { name: "new" }, + params: { name: "new" }, + output: { name: "new" }, + }; + throw new RevConflict("existing", 1, 2); + }) + .mockImplementation(updateState); + + const events: Array> = []; + const reconciler = new Reconciler({ + state, + driftDetection: false, + emit: async (event) => { + events.push(event as unknown as Record); + }, + }); + await reconciler.deploy([ + new UpdateResource({ id: "existing", config: { name: "new" } }), + ]); + + expect( + events.filter((event) => event.event === "reconciler.drift.detected"), + ).toEqual([ + { + level: "info", + event: "reconciler.drift.detected", + resourceId: "existing", + resourceType: UpdateResource.type, + diff: { name: "new" }, + }, + ]); + expect( + events.filter( + (event) => + event.event === "reconciler.deploy.decision" && + event.decision === "drift-update", + ), + ).toHaveLength(1); + }); + it("reads remote state after a create conflict instead of creating twice", async () => { let remoteName: string | undefined; const createSpy = vi.fn(async (params) => { @@ -397,7 +472,9 @@ describe("reconciler deploy", () => { }); it("detects drift using live read output and converges with update", async () => { - const updateSpy = vi.fn(async () => undefined); + const updateSpy = vi.fn( + async (_key: unknown, _patch: Record) => undefined, + ); const events: Array> = []; const TestResource = createTestResourceClass({ type: "test/service/drift", @@ -689,6 +766,59 @@ describe("reconciler destroy + refresh", () => { expect(state.delete).toHaveBeenCalledWith("c", 1); }); + it("destroy sweeps orphans after the resources it was given", async () => { + const order: string[] = []; + const DeclaredResource = createTestResourceClass({ + type: "test/service/destroy-sweep-declared", + delete: async () => void order.push("declared"), + }); + // Its type is only in the registry: an app that stopped declaring it is + // exactly the case where falling back to the declared types would skip it. + const OrphanResource = createTestResourceClass({ + type: "test/service/destroy-sweep-orphan", + delete: async () => void order.push("orphan"), + }); + + const state = createMemoryState({ + declared: { + rev: 1, + id: "declared", + groupId: -1, + groupType: "", + type: DeclaredResource.type, + config: { name: "declared" }, + params: { name: "declared" }, + output: { name: "declared" }, + lastOperation: "create", + lastOperationAt: new Date().toISOString(), + }, + orphan: { + rev: 1, + id: "orphan", + groupId: -1, + groupType: "", + type: OrphanResource.type, + config: { name: "orphan" }, + params: { name: "orphan" }, + output: { name: "orphan" }, + lastOperation: "create", + lastOperationAt: new Date().toISOString(), + }, + }); + + const reconciler = new Reconciler({ + state, + registry: createResourceRegistry([DeclaredResource, OrphanResource]), + }); + + await reconciler.destroy([ + new DeclaredResource({ id: "declared", config: { name: "declared" } }), + ]); + + expect(order).toEqual(["declared", "orphan"]); + expect(state.store).toEqual({}); + }); + it("refresh removes orphan state entries", async () => { const deleteSpy = vi.fn(async () => undefined); const OrphanResource = createTestResourceClass({ diff --git a/packages/reconciler/test/reconciler.plan.test.ts b/packages/reconciler/test/reconciler.plan.test.ts index d98e55a..3ea8752 100644 --- a/packages/reconciler/test/reconciler.plan.test.ts +++ b/packages/reconciler/test/reconciler.plan.test.ts @@ -13,16 +13,19 @@ function createMemoryState(initial: Record = {}) { return { store, - get: vi.fn(async (id: string) => store[id]), + get: vi.fn(async (id: string): Promise => store[id]), update: vi.fn( - async (id: string, expectedRev: number, patch: Partial) => { + async (id: string, _expectedRev: number, patch: Partial) => { + const rev = (store[id]?.rev ?? 0) + 1; store[id] = { ...(store[id] ?? {}), ...patch, + rev, } as StateNode; + return { rev }; }, ), - delete: vi.fn(async (id: string) => { + delete: vi.fn(async (id: string, _expectedRev: number) => { delete store[id]; }), values: vi.fn(async () => Object.values(store)), @@ -55,20 +58,16 @@ function createTestResourceClass(opts: { ) => Promise; }) { return resource({ type: opts.type }) + // Cast: a resource declared without API types constrains every schema key + // to be a key of an `any` API schema, which no named key can satisfy. .defineSchema({ - name: { - presence: "required", - propertyType: "param", - valueType: "string" as any, - }, - tag: { - presence: "optional", - propertyType: "param", - valueType: "string" as any, - }, - }) + name: { presence: "required", propertyType: "param" }, + tag: { presence: "optional", propertyType: "param" }, + } as any) .defineOperations({ - create: opts.create ?? (async () => ({})), + // Cast: with no API types the schema resolves to no primary key, so the + // inferred create signature returns void. + create: (opts.create ?? (async () => ({}))) as any, read: opts.read, update: opts.update, delete: opts.delete ?? (async () => undefined), @@ -82,6 +81,7 @@ function createStateNode( output: Record = params, ): StateNode { return { + rev: 1, id, groupId: -1, groupType: "", @@ -369,14 +369,10 @@ describe("reconciler plan", () => { type: "test/service/plan-derive-failure", }) .defineSchema({ - name: { - presence: "required", - propertyType: "param", - valueType: "string" as any, - }, - }) + name: { presence: "required", propertyType: "param" }, + } as any) .defineOperations({ - create: async () => ({}), + create: (async () => ({})) as any, delete: async () => undefined, deriveParams: () => { throw new Error("invalid derived configuration"); diff --git a/packages/reconciler/test/resource-registry.test.ts b/packages/reconciler/test/resource-registry.test.ts index dd63f48..de3fdf8 100644 --- a/packages/reconciler/test/resource-registry.test.ts +++ b/packages/reconciler/test/resource-registry.test.ts @@ -9,14 +9,14 @@ import { const TestResourceA = resource({ type: "test/service/a" }) .defineSchema({}) .defineOperations({ - create: async () => ({}), + create: (async () => ({})) as any, delete: async () => undefined, }); const TestResourceB = resource({ type: "test/service/b" }) .defineSchema({}) .defineOperations({ - create: async () => ({}), + create: (async () => ({})) as any, delete: async () => undefined, }); diff --git a/packages/reconciler/tsup.config.ts b/packages/reconciler/tsup.config.ts index f0ac238..2834ee4 100644 --- a/packages/reconciler/tsup.config.ts +++ b/packages/reconciler/tsup.config.ts @@ -1,7 +1,7 @@ import { defineConfig } from "tsup"; export default defineConfig({ - entry: ["src/index.ts"], + entry: ["src/index.ts", "src/durable/index.ts"], dts: true, format: ["esm"], }); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 82d8d14..fd80fe6 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -345,12 +345,25 @@ importers: '@notation/state': specifier: workspace:* version: link:../state + '@yieldstar/core': + specifier: 0.5.0 + version: 0.5.0 deep-object-diff: specifier: ^1.1.9 version: 1.1.9 + valibot: + specifier: ^1.4.2 + version: 1.4.2(typescript@6.0.3) yieldstar: - specifier: ^0.4.6 - version: 0.4.6 + specifier: 0.5.0 + version: 0.5.0 + devDependencies: + '@yieldstar/sqlite-runtime': + specifier: 0.5.0 + version: 0.5.0 + pino: + specifier: ^9.9.0 + version: 9.14.0 packages/resource: {} @@ -1540,8 +1553,12 @@ packages: '@vitest/utils@4.1.10': resolution: {integrity: sha512-fy9am/HWxbaGt/Sawrp90vt6Y6jQwf1RX77cz3uwoJwJVMli/e1IEwRPnMNJ7vKfPTwo0diXifkpPvwH9v7nGA==} - '@yieldstar/core@0.4.6': - resolution: {integrity: sha512-6aJQ2NwA07YKdQpiyxdg+SCKwnHQsuxAGGW6uKZYydTS10ljL3uTE6mKdrt/22ehOtbXzjRa/xnECAWh3GM7pw==} + '@yieldstar/core@0.5.0': + resolution: {integrity: sha512-KaN1+AVg54W9G4VXHNmCRixV0MY123714YZgaA9EZx7mDgB0vfFA6T71rUnldGH86QEkocCwI4GCZzZ+0G4fyQ==} + + '@yieldstar/sqlite-runtime@0.5.0': + resolution: {integrity: sha512-zdG4kvOzEJFrTW52SY+WtdUuSvKhXs438H2/+TQotwi3fltLqva9EMgRJ/l9mcvC7nqAq1i0g2v+TZWjO/py4A==} + engines: {node: '>=22.6'} abstract-logging@2.0.1: resolution: {integrity: sha512-2BjRTZxTPvheOvGbBslFSYOUkr+SjPtOnrLP33f+VIWLzezQpZcqVg7ja3L4dBXmzzgwT+a029jRx5PCi3JuiA==} @@ -2228,9 +2245,6 @@ packages: pino-abstract-transport@3.0.0: resolution: {integrity: sha512-wlfUczU+n7Hy/Ha5j9a/gZNy7We5+cXp8YL+X+PG8S0KXxw7n/JXA3c46Y0zQznIJ83URJiwy7Lh56WLokNuxg==} - pino-std-serializers@7.0.0: - resolution: {integrity: sha512-e906FRY0+tV27iq4juKzSYPbUj2do2X2JX4EzSca1631EB2QJQUqGbDuERal7LCtOpxl6x3+nvo9NPZcmjkiFA==} - pino-std-serializers@7.1.0: resolution: {integrity: sha512-BndPH67/JxGExRgiX1dX0w1FvZck5Wa4aal9198SrRhZjH3GxKQUKIBnYJTdj2HDN3UQAS06HlfcSbQj2OHmaw==} @@ -2421,9 +2435,6 @@ packages: resolution: {integrity: sha512-g9Q1haeby36OSStwb4ntCGGGaKsaVSjQ68fBxoQcutl5fS1vuY18H3wSt3jFyFtrkx+Kz0V1G85A4MyAdDMi2Q==} engines: {node: '>=8'} - sonic-boom@4.2.0: - resolution: {integrity: sha512-INb7TM37/mAcsGmc9hyyI6+QR3rR1zVRu36B0NeGXKnOOLiZOfER5SA+N7X7k3yUYRzLWafduTDvJAfDswwEww==} - sonic-boom@4.2.1: resolution: {integrity: sha512-w6AxtubXa2wTXAUsZMMWERrsIRAdrK0Sc+FUytWvYAhBJLyuI4llrMIC1DtlNSdI99EI86KZum2MMq3EAZlF9Q==} @@ -2600,6 +2611,18 @@ packages: util-deprecate@1.0.2: resolution: {integrity: sha512-EPD5q1uXyFxJpCrLnCc1nHnq3gOa6DZBocAIiI2TaSCA7VCJ1UJDMagCzIkXNsUYfD1daK//LTEQ8xiIbrHtcw==} + uuid@11.1.1: + resolution: {integrity: sha512-vIYxrBCC/N/K+Js3qSN88go7kIfNPssr/hHCesKCQNAjmgvYS2oqr69kIufEG+O4+PfezOH4EbIeHCfFov8ZgQ==} + hasBin: true + + valibot@1.4.2: + resolution: {integrity: sha512-gjdCvJ6d3RyHAneqxMYMW9QMCwYMb3jpOO0IyHZV1bnRHFBHrX3VkIILt5XYR0WhwHiH7Mty8ovuPZ/O3gamrg==} + peerDependencies: + typescript: '>=5' + peerDependenciesMeta: + typescript: + optional: true + vite@8.1.3: resolution: {integrity: sha512-Ds+gBRbj0lwRO2Y5hwnUBdxSwlAve9LeRyU4sNnAr0ewW0gWF0n5bgXgUzbgZ49MV9BVUAQUFYVcDUcilUExMA==} engines: {node: ^20.19.0 || >=22.12.0} @@ -2701,8 +2724,8 @@ packages: yallist@3.1.1: resolution: {integrity: sha512-a4UGQaWPH59mOXUYnAG2ewncQS4i4F43Tv3JoAM+s2VDAmS9NsK8GpDMLrCHPksFT7h3K6TOoUNn2pb7RoXx4g==} - yieldstar@0.4.6: - resolution: {integrity: sha512-toVxX0hHx+AlD5S3MT7voOW/Jw7AX5Mqejp28BJErSHGdZVcuKyVIfzYknbfnRnmBxAeZd3wfVnk7JQJQzZUZQ==} + yieldstar@0.5.0: + resolution: {integrity: sha512-MKuo2uaHYdy+1u0O0Q3Po3eVLGEmiNACsbaQ9aqOD4miZyRT63fXc/mhk0V4n8fNtoMN1BORD9QiXq5ZuOBtkQ==} yoctocolors@2.1.1: resolution: {integrity: sha512-GQHQqAopRhwU8Kt1DDM8NjibDXHC8eoh1erhGAJPEyveY9qqVeXvVikNKrDz69sHowPMorbPUrH/mx8c50eiBQ==} @@ -3835,7 +3858,15 @@ snapshots: convert-source-map: 2.0.0 tinyrainbow: 3.1.0 - '@yieldstar/core@0.4.6': {} + '@yieldstar/core@0.5.0': + dependencies: + '@standard-schema/spec': 1.1.0 + + '@yieldstar/sqlite-runtime@0.5.0': + dependencies: + '@yieldstar/core': 0.5.0 + pino: 9.14.0 + uuid: 11.1.1 abstract-logging@2.0.1: {} @@ -4461,8 +4492,6 @@ snapshots: dependencies: split2: 4.2.0 - pino-std-serializers@7.0.0: {} - pino-std-serializers@7.1.0: {} pino@10.3.1: @@ -4485,12 +4514,12 @@ snapshots: atomic-sleep: 1.0.0 on-exit-leak-free: 2.1.2 pino-abstract-transport: 2.0.0 - pino-std-serializers: 7.0.0 + pino-std-serializers: 7.1.0 process-warning: 5.0.0 quick-format-unescaped: 4.0.4 real-require: 0.2.0 safe-stable-stringify: 2.5.0 - sonic-boom: 4.2.0 + sonic-boom: 4.2.1 thread-stream: 3.1.0 pirates@4.0.7: {} @@ -4661,10 +4690,6 @@ snapshots: slash@3.0.0: {} - sonic-boom@4.2.0: - dependencies: - atomic-sleep: 1.0.0 - sonic-boom@4.2.1: dependencies: atomic-sleep: 1.0.0 @@ -4816,6 +4841,12 @@ snapshots: util-deprecate@1.0.2: {} + uuid@11.1.1: {} + + valibot@1.4.2(typescript@6.0.3): + optionalDependencies: + typescript: 6.0.3 + vite@8.1.3(@types/node@22.13.4)(esbuild@0.28.1)(jiti@2.7.0): dependencies: lightningcss: 1.32.0 @@ -4869,9 +4900,9 @@ snapshots: yallist@3.1.1: {} - yieldstar@0.4.6: + yieldstar@0.5.0: dependencies: - '@yieldstar/core': 0.4.6 + '@yieldstar/core': 0.5.0 nanoid: 5.1.16 pino: 9.14.0 serialize-error: 11.0.3 diff --git a/pnpm-workspace.yaml b/pnpm-workspace.yaml index 8d0e4c7..d1d8a41 100644 --- a/pnpm-workspace.yaml +++ b/pnpm-workspace.yaml @@ -3,6 +3,10 @@ packages: - "packages/*" allowBuilds: esbuild: true +minimumReleaseAgeExclude: + - "@yieldstar/core@0.5.0" + - "@yieldstar/sqlite-runtime@0.5.0" + - yieldstar@0.5.0 overrides: lodash-es@<4.18.1: ^4.18.1 js-yaml@3: 3.15.0