Add durable Yieldstar reconciliation foundation

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,
  });
 

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);
+ },
  }),
  );
 

packages/reconciler/package.json

@@ -2,19 +2,36 @@
  "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": [
  "dist"
  ],
  "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"
  }
 }

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, WorkflowStore<CoordinationState>, 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<CoordinationState>,
+ 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<any, void, any>,
+): AsyncGenerator<any, void, any> {
+ 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<DeploymentHoldTakeover> {
+ 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 };
+}

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<any, void, any> {
+ 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");
+ });
+}

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<any, void, any> {
+ 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",
+ );
+ });
+}

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:<id>:* per-resource reconciliation steps
+ * notation:destroy:<id>:* per-resource deletion steps
+ * notation:orphans:<id>:* orphan sweep on deploy, per record
+ * notation:destroy:orphans:<id>:* orphan sweep on destroy, per record
+ * *:remote:attempt:<n> one provider call attempt
+ * *:remote:retry-delay:<n> the wait between two attempts
+ * emit:<event>[:<operation>:<status>] event delivery checkpoint
+ * notation:coordination:* deployment hold: inspect/acquire/release
+ * state:persist:<id> conditional write of a resource record
+ * state:delete:<id> 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";