diff --git a/docs/manual/reconciler.md b/docs/manual/reconciler.md index 2dd96fb..cf10677 100644 --- a/docs/manual/reconciler.md +++ b/docs/manual/reconciler.md @@ -29,6 +29,8 @@ The outer workflow supplies durable step execution, timers, shared stores, waiti Each live resource is one YieldStar store. Absence is represented by no store, not a tombstone. YieldStar's UUIDv7 store `instanceId` and version are authoritative for conditional update and delete; Notation exposes the version as the resource state's `rev` for its existing state contract. +Deployments against the same `deploymentId` are serialized through a coordination store keyed by `executionId`. If a deployment crashes while holding the coordination store, resume it by running the same execution ID again: replay reclaims the acquisition through YieldStar's applied-step ledger and releases it on completion. A different execution ID waits durably until the holder releases. + Pass the complete desired set on every invocation. Persisted resources absent from that set are deleted through the supplied resource registry. The runnable Node SQLite version is in `examples/reconciler`. diff --git a/packages/reconciler/src/yieldstar.ts b/packages/reconciler/src/yieldstar.ts index 24375ba..bf59769 100644 --- a/packages/reconciler/src/yieldstar.ts +++ b/packages/reconciler/src/yieldstar.ts @@ -177,13 +177,21 @@ async function* reconcileResource( resource, stateNode: stateNode ?? undefined, params, - driftRead: - remote.status === "not-found" - ? remote - : { status: "found", output: remote.output }, + driftRead: remote, }); } + if (action.decision === "drift-update") { + const diff = action.patch; + yield* emitDurably(step, `${prefix}:drift-detected`, opts.emit, () => ({ + level: "info", + event: "reconciler.drift.detected", + resourceId: resource.id, + resourceType: resource.type, + diff, + })); + } + yield* emitDurably(step, `${prefix}:decision`, opts.emit, () => ({ level: "info", event: "reconciler.deploy.decision", @@ -206,7 +214,18 @@ async function* reconcileResource( resource.setOutput(params); if (primaryKey) resource.setOutput({ ...primaryKey, ...resource.output }); } else { - if (!resource.update) return; + if (!resource.update) { + yield* emitDurably(step, `${prefix}:update-skip`, opts.emit, () => ({ + level: "info", + event: "reconciler.operation.lifecycle", + operation: "update", + status: "skip", + resourceId: resource.id, + resourceType: resource.type, + reason: "update-not-implemented", + })); + return; + } yield* runProviderCall( step, `${prefix}:update`, @@ -231,7 +250,6 @@ async function* reconcileResource( ); if (read.status === "found") resource.setOutput({ ...resource.output, ...read.output }); - if (opts.dryRun) return; const operation = action.decision === "create" || action.decision === "drift-recreate" @@ -406,15 +424,31 @@ export class YieldStarStateBackend { } async get(id: string): Promise { - const storeId = this.storeId(id); - const ids = await this.#client.listStores(yieldStarResourceStateStore); - if (!ids.includes(storeId)) return undefined; - return toStateNode( - await this.#client.getStore({ + const snapshot = await this.#tryGetSnapshot(this.storeId(id)); + return snapshot ? toStateNode(snapshot) : undefined; + } + + /** + * Reads a store snapshot in one round trip. A missing store is resource + * absence, so a read failure is re-checked against the store listing before + * it is allowed to propagate. + */ + async #tryGetSnapshot( + storeId: string, + ): Promise< + | { state: StoredResourceState; instanceId: string; version: number } + | undefined + > { + try { + return await this.#client.getStore({ definition: yieldStarResourceStateStore, id: storeId, - }), - ); + }); + } catch (error) { + const ids = await this.#client.listStores(yieldStarResourceStateStore); + if (!ids.includes(storeId)) return undefined; + throw error; + } } async has(id: string): Promise { @@ -427,8 +461,8 @@ export class YieldStarStateBackend { patch: Partial, ): Promise<{ rev: number }> { const storeId = this.storeId(id); - const ids = await this.#client.listStores(yieldStarResourceStateStore); - if (!ids.includes(storeId)) { + const snapshot = await this.#tryGetSnapshot(storeId); + if (!snapshot) { if (expectedRev !== 0) throw new RevConflict(id, expectedRev, undefined); const initial = { ...patch, id } as StoredResourceState; const created = await this.#client.getOrCreateStore({ @@ -439,14 +473,9 @@ export class YieldStarStateBackend { return { rev: created.version + 1 }; } - const snapshot = await this.#client.getStore({ - definition: yieldStarResourceStateStore, - id: storeId, - }); const actualRev = snapshot.version + 1; if (actualRev !== expectedRev) throw new RevConflict(id, expectedRev, actualRev); - const rev = actualRev + 1; const result = await this.#client.updateStoreFrom({ definition: yieldStarResourceStateStore, id: storeId, @@ -456,20 +485,16 @@ export class YieldStarStateBackend { }, }); if (!result.updated) throw new RevConflict(id, expectedRev, undefined); - return { rev }; + return { rev: result.version + 1 }; } async delete(id: string, expectedRev: number): Promise { const storeId = this.storeId(id); - const ids = await this.#client.listStores(yieldStarResourceStateStore); - if (!ids.includes(storeId)) { + const snapshot = await this.#tryGetSnapshot(storeId); + if (!snapshot) { if (expectedRev !== 0) throw new RevConflict(id, expectedRev, undefined); return; } - const snapshot = await this.#client.getStore({ - definition: yieldStarResourceStateStore, - id: storeId, - }); const actualRev = snapshot.version + 1; if (actualRev !== expectedRev) throw new RevConflict(id, expectedRev, actualRev); @@ -484,19 +509,14 @@ export class YieldStarStateBackend { async values(): Promise { const prefix = `${this.#deploymentId}:`; const ids = await this.#client.listStores(yieldStarResourceStateStore); - const nodes = await Promise.all( + const snapshots = await Promise.all( ids .filter((id) => id.startsWith(prefix)) - .map(async (id) => - toStateNode( - await this.#client.getStore({ - definition: yieldStarResourceStateStore, - id, - }), - ), - ), + .map((id) => this.#tryGetSnapshot(id)), ); - return nodes; + return snapshots + .filter((snapshot) => snapshot !== undefined) + .map(toStateNode); } snapshot(id: string) { diff --git a/packages/reconciler/test/yieldstar.integration.test.ts b/packages/reconciler/test/yieldstar.integration.test.ts index 0ddbc5d..5e1b570 100644 --- a/packages/reconciler/test/yieldstar.integration.test.ts +++ b/packages/reconciler/test/yieldstar.integration.test.ts @@ -17,6 +17,11 @@ import { reconcileWithYieldStar, yieldStarResourceStateStore, } from "../src/yieldstar"; +import type { ReconcilerEvent } from "../src/reconciler"; +import { + createResourceRegistry, + type ResourceRegistry, +} from "../src/resource-registry"; const logger = pino({ level: "silent" }); @@ -42,7 +47,7 @@ describe("YieldStar reconciliation", () => { const runtime = createRuntime( [new PendingResource({ id: "pending" })], "durable-wait", - { maxAttempts: 3, retryInterval: 1 }, + { retryOptions: { maxAttempts: 3, retryInterval: 1 } }, ); await runtime.run("wait-execution"); @@ -67,8 +72,7 @@ describe("YieldStar reconciliation", () => { const runtime = createRuntime( [new TestResource({ id: "resume" })], "crash-resume", - undefined, - "notation:resource:resume:create", + { crashAfterStep: "notation:resource:resume:create" }, ); await expect(runtime.run("resume-execution")).rejects.toThrow( @@ -160,19 +164,90 @@ describe("YieldStar reconciliation", () => { expect(await runtime.state.values()).toHaveLength(1); runtime.close(); }); + + it("deletes orphaned resources through the registry on a later deployment", async () => { + const deleteSpy = vi.fn(async () => undefined); + const OrphanResource = resource({ type: "test/yieldstar/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(); + }); + + 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/yieldstar/drift" }) + .defineSchema({ + name: { + presence: "required", + propertyType: "param", + valueType: "string" as any, + }, + }) + .defineOperations({ + create: async () => remote, + 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(); + }); }); function createRuntime( resources: BaseResource[], deploymentId: string, - retryOptions?: { maxAttempts: number; retryInterval: number }, - crashAfterStep?: string, + options: { + retryOptions?: { maxAttempts: number; retryInterval: number }; + crashAfterStep?: string; + registry?: ResourceRegistry; + driftDetection?: boolean; + emit?: (event: ReconcilerEvent) => void; + } = {}, ) { const database = createSqliteDb({ path: ":memory:" }); const scheduler = new TestScheduler(); const sqliteHeap = new SqliteHeapClient(database); - const heap = crashAfterStep - ? new CrashAfterWriteHeap(sqliteHeap, crashAfterStep) + const heap = options.crashAfterStep + ? new CrashAfterWriteHeap(sqliteHeap, options.crashAfterStep) : sqliteHeap; const storeClient = new SqliteStoreClient({ db: database, @@ -185,8 +260,10 @@ function createRuntime( executionId: event.executionId, resources, state, - driftDetection: false, - retryOptions, + registry: options.registry, + driftDetection: options.driftDetection ?? false, + emit: options.emit, + retryOptions: options.retryOptions, }); }); const router = createWorkflowRouter({ deploy });