diff --git a/packages/core/package.json b/packages/core/package.json index f89c5c5..542912a 100644 --- a/packages/core/package.json +++ b/packages/core/package.json @@ -16,6 +16,7 @@ "@notation/reconciler": "workspace:*", "@notation/resource": "workspace:*", "@notation/state": "workspace:*", + "@notation/state-sqlite": "workspace:*", "deep-object-diff": "^1.1.9", "js-base64": "^3.7.7", "lodash-es": "^4.17.21", diff --git a/packages/core/src/provisioner/operations/operation.create.ts b/packages/core/src/provisioner/operations/operation.create.ts index 13a8022..a2d0fdd 100644 --- a/packages/core/src/provisioner/operations/operation.create.ts +++ b/packages/core/src/provisioner/operations/operation.create.ts @@ -25,7 +25,7 @@ async function create( resource.setOutput({ ...resource.output, ...readResult }); - await state.update(resource.id, { + await state.update(resource.id, 0, { id: resource.id, groupId: resource.groupId, groupType: resource.groupType, diff --git a/packages/core/src/provisioner/operations/operation.delete.ts b/packages/core/src/provisioner/operations/operation.delete.ts index f1c54ec..7e09040 100644 --- a/packages/core/src/provisioner/operations/operation.delete.ts +++ b/packages/core/src/provisioner/operations/operation.delete.ts @@ -6,6 +6,10 @@ export const deleteResource = operation("Destroying", delete_); async function delete_(opts: { resource: BaseResource; state: StateBackend }) { const { resource, state } = opts; + const stateNode = await state.get(resource.id); + if (!stateNode) { + throw new Error(`Missing state for ${resource.type} ${resource.id}`); + } try { await resource.delete(resource.key, resource.toState(resource.output)); @@ -27,5 +31,5 @@ async function delete_(opts: { resource: BaseResource; state: StateBackend }) { } } - await state.delete(resource.id); + await state.delete(resource.id, stateNode.rev); } diff --git a/packages/core/src/provisioner/operations/operation.update.ts b/packages/core/src/provisioner/operations/operation.update.ts index d061832..172f382 100644 --- a/packages/core/src/provisioner/operations/operation.update.ts +++ b/packages/core/src/provisioner/operations/operation.update.ts @@ -11,6 +11,10 @@ async function update(opts: { patch: any; }): Promise { const { resource, state, patch } = opts; + const stateNode = await state.get(resource.id); + if (!stateNode) { + throw new Error(`Missing state for ${resource.type} ${resource.id}`); + } if (!resource.update) { throw new Error( @@ -31,7 +35,7 @@ async function update(opts: { const result = await readResource({ resource, state, quiet: true }); resource.setOutput({ ...resource.output, ...result }); - await state.update(resource.id, { + await state.update(resource.id, stateNode.rev, { lastOperation: "update", lastOperationAt: new Date().toISOString(), params: resource.toState(params), diff --git a/packages/core/src/provisioner/state-backend.ts b/packages/core/src/provisioner/state-backend.ts index b912b0f..437e56c 100644 --- a/packages/core/src/provisioner/state-backend.ts +++ b/packages/core/src/provisioner/state-backend.ts @@ -1,4 +1,5 @@ import { FileStateBackend, type StateBackend } from "@notation/state"; +import { SqliteStateBackend } from "@notation/state-sqlite"; export const DEFAULT_STATE_PATH = "./.notation/state.json"; @@ -7,5 +8,9 @@ export function resolveStatePath(): string { } export function createDefaultStateBackend(): StateBackend { - return new FileStateBackend(resolveStatePath()); + const statePath = resolveStatePath(); + if (statePath.endsWith(".db") || statePath.endsWith(".sqlite")) { + return new SqliteStateBackend(statePath); + } + return new FileStateBackend(statePath); } diff --git a/packages/core/test/provisioner/operation.create.test.ts b/packages/core/test/provisioner/operation.create.test.ts index 562c806..2de482d 100644 --- a/packages/core/test/provisioner/operation.create.test.ts +++ b/packages/core/test/provisioner/operation.create.test.ts @@ -31,6 +31,7 @@ describe("resource creation", () => { createResourceOperation(step, { resource: testResource, state: stateBackend, + expectedRev: 0, }), ); diff --git a/packages/core/test/provisioner/state-backend.test.ts b/packages/core/test/provisioner/state-backend.test.ts new file mode 100644 index 0000000..5ac04a4 --- /dev/null +++ b/packages/core/test/provisioner/state-backend.test.ts @@ -0,0 +1,59 @@ +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { FileStateBackend } from "@notation/state"; +import { SqliteStateBackend } from "@notation/state-sqlite"; +import { createDefaultStateBackend } from "src/provisioner/state-backend"; + +describe("createDefaultStateBackend", () => { + let directory: string; + const originalStatePath = process.env.NOTATION_STATE_PATH; + + beforeEach(() => { + directory = mkdtempSync(path.join(tmpdir(), "notation-state-")); + }); + + afterEach(() => { + if (originalStatePath === undefined) { + delete process.env.NOTATION_STATE_PATH; + } else { + process.env.NOTATION_STATE_PATH = originalStatePath; + } + rmSync(directory, { recursive: true, force: true }); + }); + + it("uses the file backend for the default JSON path", () => { + delete process.env.NOTATION_STATE_PATH; + + expect(createDefaultStateBackend()).toBeInstanceOf(FileStateBackend); + }); + + it("uses the sqlite backend for .db paths", () => { + process.env.NOTATION_STATE_PATH = path.join(directory, "state.db"); + + const backend = createDefaultStateBackend(); + expect(backend).toBeInstanceOf(SqliteStateBackend); + (backend as SqliteStateBackend).close(); + }); + + it("uses the sqlite backend for .sqlite paths", () => { + process.env.NOTATION_STATE_PATH = path.join(directory, "state.sqlite"); + + const backend = createDefaultStateBackend(); + expect(backend).toBeInstanceOf(SqliteStateBackend); + (backend as SqliteStateBackend).close(); + }); + + it("creates missing parent directories for sqlite paths", () => { + process.env.NOTATION_STATE_PATH = path.join( + directory, + ".notation", + "state.db", + ); + + const backend = createDefaultStateBackend(); + expect(backend).toBeInstanceOf(SqliteStateBackend); + (backend as SqliteStateBackend).close(); + }); +}); diff --git a/packages/reconciler/src/operations/operation.create.ts b/packages/reconciler/src/operations/operation.create.ts index 49496d0..98eac89 100644 --- a/packages/reconciler/src/operations/operation.create.ts +++ b/packages/reconciler/src/operations/operation.create.ts @@ -60,7 +60,7 @@ export async function* createResourceOperation( }); yield* step.run("create:persist-state", async () => { - await params.state.update(params.resource.id, { + await params.state.update(params.resource.id, params.expectedRev, { id: params.resource.id, groupId: params.resource.groupId, groupType: params.resource.groupType, diff --git a/packages/reconciler/src/operations/operation.delete.ts b/packages/reconciler/src/operations/operation.delete.ts index b8da37b..ffa5de4 100644 --- a/packages/reconciler/src/operations/operation.delete.ts +++ b/packages/reconciler/src/operations/operation.delete.ts @@ -49,7 +49,7 @@ export async function* deleteResourceOperation( } yield* step.run("delete:persist-state", () => - params.state.delete(params.resource.id), + params.state.delete(params.resource.id, params.expectedRev), ); await emitLifecycleEvent(params, "delete", "success"); diff --git a/packages/reconciler/src/operations/operation.types.ts b/packages/reconciler/src/operations/operation.types.ts index c29f282..3def119 100644 --- a/packages/reconciler/src/operations/operation.types.ts +++ b/packages/reconciler/src/operations/operation.types.ts @@ -1,14 +1,14 @@ -import type { BaseResource, ErrorMatcher, ResourceType } from "@notation/resource"; +import type { + BaseResource, + ErrorMatcher, + ResourceType, +} from "@notation/resource"; import type { State } from "@notation/state"; export type OperationName = "create" | "read" | "update" | "delete"; export type OperationLifecycleStatus = - | "start" - | "success" - | "error" - | "skip" - | "dry-run"; + "start" | "success" | "error" | "skip" | "dry-run"; export type OperationLifecycleEvent = { level: "info" | "error"; @@ -59,15 +59,20 @@ export type ResourceOperationBaseParams = { readPollOptions?: PollOptions; }; -export type CreateResourceParams = ResourceOperationBaseParams; +export type CreateResourceParams = ResourceOperationBaseParams & { + expectedRev: number; +}; export type ReadResourceParams = ResourceOperationBaseParams; export type UpdateResourceParams = ResourceOperationBaseParams & { patch: Record; + expectedRev: number; }; -export type DeleteResourceParams = ResourceOperationBaseParams; +export type DeleteResourceParams = ResourceOperationBaseParams & { + expectedRev: number; +}; export const DEFAULT_RETRY_OPTIONS: PollOptions = { maxAttempts: 10, diff --git a/packages/reconciler/src/operations/operation.update.ts b/packages/reconciler/src/operations/operation.update.ts index a345bdc..55dcba1 100644 --- a/packages/reconciler/src/operations/operation.update.ts +++ b/packages/reconciler/src/operations/operation.update.ts @@ -70,9 +70,14 @@ export async function* updateResourceOperation( }); yield* step.run("update:persist-state", async () => { - await params.state.update(params.resource.id, { + 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), }); diff --git a/packages/reconciler/src/plan.ts b/packages/reconciler/src/plan.ts index 0944c09..7f0aff0 100644 --- a/packages/reconciler/src/plan.ts +++ b/packages/reconciler/src/plan.ts @@ -43,7 +43,11 @@ export type ResourceAction = | { decision: "noop" } | { decision: "drift-recreate" } | { decision: "update"; patch: Record; diff: PlanDiff } - | { decision: "drift-update"; patch: Record; diff: PlanDiff }; + | { + decision: "drift-update"; + patch: Record; + diff: PlanDiff; + }; export function decideAction(opts: { resource: BaseResource; @@ -52,52 +56,60 @@ export function decideAction(opts: { driftRead?: DriftRead; }): ResourceAction { const { resource, stateNode, params, driftRead } = opts; - - if (!stateNode) { - return { decision: "create" }; - } - - const previousComparable = resource.toComparable(stateNode.params); const desiredComparable = resource.toComparable(params ?? {}); + const previousComparable = resource.toComparable(stateNode?.params ?? {}); const localPatch = diff(previousComparable, desiredComparable) as Record< string, unknown >; - if (Object.keys(localPatch).length > 0) { + if (driftRead) { + if (driftRead.status === "not-found") { + return { decision: stateNode ? "drift-recreate" : "create" }; + } + + const remoteComparable = resource.toComparable(driftRead.output); + const remotePatch = diff(remoteComparable, desiredComparable) as Record< + string, + unknown + >; + + if (Object.keys(remotePatch).length === 0) { + return { decision: "noop" }; + } + + const remoteDetailedDiff = detailedDiff( + remoteComparable, + desiredComparable, + ); + if (!stateNode || Object.keys(localPatch).length > 0) { + return { + decision: "update", + patch: remotePatch, + diff: toPlanDiff(remoteDetailedDiff), + }; + } + return { - decision: "update", - patch: localPatch, - diff: toPlanDiff(detailedDiff(previousComparable, desiredComparable)), + decision: "drift-update", + patch: remotePatch, + diff: toPlanDiff(remoteDetailedDiff), }; } - if (!driftRead) { - return { decision: "noop" }; - } - - if (driftRead.status === "not-found") { - return { decision: "drift-recreate" }; + if (!stateNode) { + return { decision: "create" }; } - const remoteDetailedDiff = detailedDiff( - resource.toComparable(driftRead.output), - resource.toComparable(stateNode.output), - ); - const remotePatch = { - ...remoteDetailedDiff.updated, - ...remoteDetailedDiff.added, - } as Record; - - if (Object.keys(remotePatch).length === 0) { - return { decision: "noop" }; + if (Object.keys(localPatch).length > 0) { + return { + decision: "update", + patch: localPatch, + diff: toPlanDiff(detailedDiff(previousComparable, desiredComparable)), + }; } - return { - decision: "drift-update", - patch: remotePatch, - diff: toPlanDiff(remoteDetailedDiff), - }; + return { decision: "noop" }; } export async function resolvePlanParams( diff --git a/packages/reconciler/src/reconciler.ts b/packages/reconciler/src/reconciler.ts index 874c5ea..2db4628 100644 --- a/packages/reconciler/src/reconciler.ts +++ b/packages/reconciler/src/reconciler.ts @@ -1,6 +1,7 @@ import type { BaseResource, ResourceType } from "@notation/resource"; -import type { State, StateNode } from "@notation/state"; +import { RevConflict, type State, type StateNode } from "@notation/state"; import { RetryableError } from "yieldstar"; +import { setTimeout as sleep } from "node:timers/promises"; import { buildResourceDepthLevels } from "./dependency-graph"; import { decideAction, @@ -55,7 +56,10 @@ export type ReconcilerEventEmitter = ( event: ReconcilerEvent, ) => void | Promise; -export type ReconcilerState = Pick; +export type ReconcilerState = Pick< + State, + "get" | "update" | "delete" | "values" | "lease" +>; export type ReconcilerOptions = { state: ReconcilerState; @@ -65,6 +69,7 @@ export type ReconcilerOptions = { emit?: ReconcilerEventEmitter; retryOptions?: PollOptions; readPollOptions?: PollOptions; + mutationLeaseTtl?: number; }; export type DeployOptions = { @@ -92,6 +97,7 @@ export class Reconciler { readonly #emit?: ReconcilerEventEmitter; readonly #retryOptions?: PollOptions; readonly #readPollOptions?: PollOptions; + readonly #mutationLeaseTtl: number; readonly #stepRunner: StepRunner; constructor(opts: ReconcilerOptions) { @@ -102,18 +108,26 @@ export class Reconciler { this.#emit = opts.emit; this.#retryOptions = opts.retryOptions; this.#readPollOptions = opts.readPollOptions; + this.#mutationLeaseTtl = opts.mutationLeaseTtl ?? 30_000; this.#stepRunner = createStepRunner(); } - async deploy(resources: BaseResource[], opts: DeployOptions = {}): Promise { + async deploy( + resources: BaseResource[], + opts: DeployOptions = {}, + ): Promise { const dryRun = opts.dryRun ?? this.#defaultDryRun; const driftDetection = opts.driftDetection ?? this.#defaultDriftDetection; - const resourceById = new Map(resources.map((resource) => [resource.id, resource])); + const resourceById = new Map( + resources.map((resource) => [resource.id, resource]), + ); const dependencyLevels = buildResourceDepthLevels(resources); for (const level of dependencyLevels) { await Promise.all( - level.map((resource) => this.#deployResource(resource, dryRun, driftDetection)), + level.map((resource) => + this.#deployResource(resource, dryRun, driftDetection), + ), ); } @@ -122,7 +136,9 @@ 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 resourceById = new Map( + resources.map((resource) => [resource.id, resource]), + ); const nodes: PlanNode[] = []; const dependencyLevels = buildResourceDepthLevels(resources); @@ -151,19 +167,33 @@ export class Reconciler { }; } - async destroy(resources: BaseResource[], opts: DestroyOptions = {}): Promise { + async destroy( + resources: BaseResource[], + opts: DestroyOptions = {}, + ): Promise { const dryRun = opts.dryRun ?? this.#defaultDryRun; const dependencyLevels = buildResourceDepthLevels(resources); - for (let levelIndex = dependencyLevels.length - 1; levelIndex >= 0; levelIndex -= 1) { + for ( + let levelIndex = dependencyLevels.length - 1; + levelIndex >= 0; + levelIndex -= 1 + ) { const level = dependencyLevels[levelIndex]!; - await Promise.all(level.map((resource) => this.#destroyResource(resource, dryRun))); + await Promise.all( + level.map((resource) => this.#destroyResource(resource, dryRun)), + ); } } - async refresh(resources: BaseResource[], opts: RefreshOptions = {}): Promise { + async refresh( + resources: BaseResource[], + opts: RefreshOptions = {}, + ): Promise { const dryRun = opts.dryRun ?? this.#defaultDryRun; - const resourceById = new Map(resources.map((resource) => [resource.id, resource])); + const resourceById = new Map( + resources.map((resource) => [resource.id, resource]), + ); await this.#deleteOrphans(resources, resourceById, dryRun, "refresh"); } @@ -173,6 +203,76 @@ export class Reconciler { dryRun: boolean, driftDetection: boolean, ) { + await this.#withMutationLease(resource.id, () => + this.#retryOnRevConflict((conflict) => + this.#deployResourceOnce(resource, dryRun, driftDetection, conflict), + ), + ); + } + + async #withMutationLease(resourceId: string, fn: () => Promise) { + return this.#withLease(`reconciler:resource:${resourceId}`, fn); + } + + async #withLease(scope: string, fn: () => Promise): Promise { + const lease = await this.#state.lease(scope, this.#mutationLeaseTtl); + const controller = new AbortController(); + let renewalError: unknown; + const heartbeat = (async () => { + try { + while (!controller.signal.aborted) { + await sleep( + Math.max(1, Math.floor(this.#mutationLeaseTtl / 3)), + undefined, + { + signal: controller.signal, + }, + ); + await lease.renew(this.#mutationLeaseTtl); + } + } catch (error) { + if (!controller.signal.aborted) renewalError = error; + } + })(); + + try { + const result = await fn(); + if (renewalError) throw renewalError; + return result; + } finally { + controller.abort(); + await heartbeat; + await lease.release(); + } + } + + async #retryOnRevConflict(fn: (conflict?: RevConflict) => Promise) { + let conflict: RevConflict | undefined; + for (let attempt = 0; attempt < 3; attempt += 1) { + try { + await fn(conflict); + return; + } catch (error) { + if (!(error instanceof RevConflict) || attempt === 2) throw error; + // Re-throwing the conflict supplied for recovery means the resource + // cannot be recovered safely (for example, it has no read operation). + if (error === conflict) throw error; + conflict = error; + } + } + } + + 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; @@ -218,6 +318,71 @@ export class Reconciler { emit: this.#emit, retryOptions: this.#retryOptions, readPollOptions: this.#readPollOptions, + 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, + retryOptions: this.#retryOptions, + readPollOptions: this.#readPollOptions, + 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.status === "found") 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, + retryOptions: this.#retryOptions, + readPollOptions: this.#readPollOptions, + expectedRev: stateNode?.rev ?? 0, }), ); return; @@ -232,10 +397,23 @@ export class Reconciler { emit: this.#emit, retryOptions: this.#retryOptions, readPollOptions: this.#readPollOptions, + 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; } } @@ -291,47 +469,80 @@ export class Reconciler { dryRun: boolean, workflow: "deploy" | "refresh", ) { - const stateNodes = await this.#state.values(); - const registry = this.#registry ?? createResourceRegistryFromResources(resources); - - for (const stateNode of stateNodes) { - if (resourceById.has(stateNode.id)) continue; - - const stateNodeResourceType = stateNode.type as ResourceType; + await this.#withLease("reconciler:orphan-deletion", async () => { + const stateNodes = await this.#state.values(); + const registry = + this.#registry ?? createResourceRegistryFromResources(resources); + + for (const stateNode of stateNodes) { + if (resourceById.has(stateNode.id)) continue; + + const stateNodeResourceType = stateNode.type as ResourceType; + + const Resource = resolveResourceClass(registry, stateNodeResourceType); + if (!Resource) { + await this.#emit?.( + createMissingResourceRegistryMatchWarningEvent({ + workflow, + resourceId: stateNode.id, + resourceType: stateNodeResourceType, + }), + ); + continue; + } - const Resource = resolveResourceClass(registry, stateNodeResourceType); - if (!Resource) { - await this.#emit?.( - createMissingResourceRegistryMatchWarningEvent({ - workflow, - resourceId: stateNode.id, - resourceType: stateNodeResourceType, + 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, + ); }), ); - continue; } + }); + } - const orphanResource = hydrateResourceFromState(Resource, stateNode); + 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; + } - await runOperation( - deleteResourceOperation(this.#stepRunner, { - resource: orphanResource, - state: this.#state, - dryRun, - emit: this.#emit, - retryOptions: this.#retryOptions, - }), - ); - } + resource.setOutput(stateNode.output); + await this.#deleteResourceOnce(resource, stateNode, dryRun, conflict); + }), + ); } - async #destroyResource(resource: BaseResource, dryRun: boolean) { - const stateNode = await this.#state.get(resource.id); - if (!stateNode) { - return; - } + async #deleteResourceOnce( + resource: BaseResource, + stateNode: StateNode, + dryRun: boolean, + conflict?: RevConflict, + ) { + if (conflict) { + if (!resource.read) throw conflict; - resource.setOutput(stateNode.output); + const remote = await this.#readForDrift(resource); + if (remote.status === "not-found") { + if (!dryRun) await this.#state.delete(resource.id, stateNode.rev); + return; + } + resource.setOutput(remote.output); + } await runOperation( deleteResourceOperation(this.#stepRunner, { @@ -340,12 +551,15 @@ export class Reconciler { dryRun, emit: this.#emit, retryOptions: this.#retryOptions, + expectedRev: stateNode.rev, }), ); } } -export async function runOperation(operation: AsyncGenerator) { +export async function runOperation( + operation: AsyncGenerator, +) { let next = await operation.next(); while (!next.done) { next = await operation.next(); @@ -354,7 +568,10 @@ export async function runOperation(operation: AsyncGenerator }) => BaseResource, + Resource: new (opts: { + id: string; + config: Record; + }) => BaseResource, stateNode: StateNode, ): BaseResource { const resource = new Resource({ @@ -372,8 +589,7 @@ export function createStepRunner(): StepRunner { arg2?: () => T | Promise, ): AsyncGenerator { const fn = (typeof arg1 === "string" ? arg2 : arg1) as - | (() => T | Promise) - | undefined; + (() => T | Promise) | undefined; if (!fn) { throw new Error("Missing run function"); @@ -396,8 +612,7 @@ export function createStepRunner(): StepRunner { ): AsyncGenerator { const opts = (typeof arg1 === "string" ? arg2 : arg1) as PollOptions; const predicate = (typeof arg1 === "string" ? arg3 : arg2) as - | (() => boolean | Promise) - | undefined; + (() => boolean | Promise) | undefined; if (!predicate) { throw new Error("Missing poll predicate"); diff --git a/packages/reconciler/test/operation.workflows.test.ts b/packages/reconciler/test/operation.workflows.test.ts index cdf347b..8709086 100644 --- a/packages/reconciler/test/operation.workflows.test.ts +++ b/packages/reconciler/test/operation.workflows.test.ts @@ -112,6 +112,7 @@ describe("operation workflows", () => { createResourceOperation(step, { resource: testResource, state, + expectedRev: 0, emit: async (event) => { events.push(event); }, @@ -211,13 +212,14 @@ describe("operation workflows", () => { deleteResourceOperation(step, { resource: testResource, state, + expectedRev: 1, emit: async (event) => { events.push(event); }, }), ); - expect(state.delete).toHaveBeenCalledWith("test-delete"); + expect(state.delete).toHaveBeenCalledWith("test-delete", 1); expect(events.map((event) => event.status)).toEqual([ "start", "skip", @@ -257,6 +259,7 @@ describe("operation workflows", () => { deleteResourceOperation(step, { resource: testResource, state, + expectedRev: 1, }), ), ).rejects.toMatchObject({ name: "DifferentError", message: "still exists" }); @@ -291,6 +294,7 @@ describe("operation workflows", () => { createResourceOperation(step, { resource: testResource, state, + expectedRev: 0, emit: async (event) => { events.push(event); }, diff --git a/packages/reconciler/test/reconciler.deploy.test.ts b/packages/reconciler/test/reconciler.deploy.test.ts index 8e8bc5f..7220a65 100644 --- a/packages/reconciler/test/reconciler.deploy.test.ts +++ b/packages/reconciler/test/reconciler.deploy.test.ts @@ -1,6 +1,11 @@ import { describe, expect, it, vi } from "vitest"; -import { resource } from "@notation/resource"; -import type { StateNode } from "@notation/state"; +import { resource, type ErrorMatcher } from "@notation/resource"; +import { + LeaseConflict, + MemoryStateBackend, + RevConflict, + type StateNode, +} from "@notation/state"; import { Reconciler, createResourceRegistry } from "../src"; function createMemoryState(initial: Record = {}) { @@ -9,22 +14,51 @@ function createMemoryState(initial: Record = {}) { return { store, get: vi.fn(async (id: string) => store[id]), - update: vi.fn(async (id: string, patch: Partial) => { - store[id] = { - ...(store[id] ?? {}), - ...patch, - } as StateNode; - }), - delete: vi.fn(async (id: string) => { + update: vi.fn( + async (id: string, expectedRev: number, patch: Partial) => { + const actualRev = store[id]?.rev ?? 0; + if (actualRev !== expectedRev) { + throw new RevConflict(id, expectedRev, store[id]?.rev); + } + const rev = actualRev + 1; + store[id] = { + ...(store[id] ?? {}), + ...patch, + rev, + } as StateNode; + return { rev }; + }, + ), + delete: vi.fn(async (id: string, expectedRev: number) => { + const actualRev = store[id]?.rev ?? 0; + if (actualRev !== expectedRev) { + throw new RevConflict(id, expectedRev, store[id]?.rev); + } delete store[id]; }), values: vi.fn(async () => Object.values(store)), + lease: vi.fn(async (scope: string, ttl: number) => { + let expiresAt = new Date(Date.now() + ttl).toISOString(); + return { + scope, + get expiresAt() { + return expiresAt; + }, + renew: vi.fn(async (nextTtl: number) => { + expiresAt = new Date(Date.now() + nextTtl).toISOString(); + return expiresAt; + }), + release: vi.fn(async () => undefined), + }; + }), }; } function createTestResourceClass(opts: { type: `${string}/${string}/${string}`; - create?: (params: Record) => Promise | void>; + create?: ( + params: Record, + ) => Promise | void>; read?: (key: Record) => Promise>; update?: ( key: Record, @@ -32,7 +66,11 @@ function createTestResourceClass(opts: { params: Record, state: Record, ) => Promise; - delete?: (key: Record, state: Record) => Promise; + delete?: ( + key: Record, + state: Record, + ) => Promise; + notFoundOnError?: ErrorMatcher[]; }) { return resource({ type: opts.type }) .defineSchema({ @@ -47,6 +85,7 @@ function createTestResourceClass(opts: { read: opts.read, update: opts.update, delete: opts.delete ?? (async () => undefined), + notFoundOnError: opts.notFoundOnError, }); } @@ -70,6 +109,7 @@ describe("reconciler deploy", () => { const state = createMemoryState({ existing: { + rev: 1, id: "existing", groupId: -1, groupType: "", @@ -105,6 +145,210 @@ describe("reconciler deploy", () => { expect(events).toContain("update:success:existing"); }); + it("persists first-time creates with an expect-absent revision", async () => { + const CreateResource = createTestResourceClass({ + type: "test/service/first-create", + create: async () => ({ name: "new" }), + read: async () => ({ name: "new" }), + }); + const state = createMemoryState(); + const reconciler = new Reconciler({ state, driftDetection: false }); + + await reconciler.deploy([ + new CreateResource({ id: "new", config: { name: "new" } }), + ]); + + expect(state.update).toHaveBeenCalledWith("new", 0, expect.any(Object)); + }); + + it("leases a resource before remote create so concurrent deploys cannot duplicate it", async () => { + let signalCreateStarted!: () => void; + const createStarted = new Promise((resolve) => { + signalCreateStarted = resolve; + }); + let allowCreateToFinish!: () => void; + const createCanFinish = new Promise((resolve) => { + allowCreateToFinish = resolve; + }); + const createSpy = vi.fn(async () => { + signalCreateStarted(); + await createCanFinish; + return { name: "new" }; + }); + const CreateResource = createTestResourceClass({ + type: "test/service/concurrent-create", + create: createSpy, + read: async () => ({ name: "new" }), + }); + const state = new MemoryStateBackend(); + const first = new Reconciler({ state, driftDetection: false }); + const second = new Reconciler({ state, driftDetection: false }); + + const firstDeploy = first.deploy([ + new CreateResource({ id: "new", config: { name: "new" } }), + ]); + await createStarted; + + await expect( + second.deploy([ + new CreateResource({ id: "new", config: { name: "new" } }), + ]), + ).rejects.toBeInstanceOf(LeaseConflict); + + allowCreateToFinish(); + await firstDeploy; + expect(createSpy).toHaveBeenCalledOnce(); + }); + + it("reads remote state after an update conflict instead of repeating the update", async () => { + let remoteName = "old"; + const readSpy = vi.fn(async () => ({ name: remoteName })); + const updateSpy = vi.fn(async (_key, _patch, params) => { + remoteName = params.name as string; + }); + const UpdateResource = createTestResourceClass({ + type: "test/service/update-conflict", + 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: "concurrent" }, + params: { name: "concurrent" }, + output: { name: "concurrent" }, + }; + throw new RevConflict("existing", 1, 2); + }) + .mockImplementation(updateState); + + const reconciler = new Reconciler({ state, driftDetection: false }); + await reconciler.deploy([ + new UpdateResource({ id: "existing", config: { name: "new" } }), + ]); + + expect(updateSpy).toHaveBeenCalledOnce(); + expect(readSpy).toHaveBeenCalledTimes(2); + expect(state.update).toHaveBeenLastCalledWith( + "existing", + 2, + expect.objectContaining({ + params: { name: "new" }, + output: { name: "new" }, + lastOperation: "drift", + }), + ); + expect(state.store.existing).toMatchObject({ + rev: 3, + params: { name: "new" }, + output: { name: "new" }, + }); + }); + + it("reads remote state after a create conflict instead of creating twice", async () => { + let remoteName: string | undefined; + const createSpy = vi.fn(async (params) => { + remoteName = params.name as string; + return { name: remoteName }; + }); + const readSpy = vi.fn(async () => ({ name: remoteName! })); + const CreateResource = createTestResourceClass({ + type: "test/service/create-conflict", + create: createSpy, + read: readSpy, + }); + const state = createMemoryState(); + const updateState = state.update.getMockImplementation()!; + state.update + .mockImplementationOnce(async () => { + state.store.new = { + rev: 1, + id: "new", + groupId: -1, + groupType: "", + type: CreateResource.type, + config: { name: "concurrent" }, + params: { name: "concurrent" }, + output: { name: "concurrent" }, + lastOperation: "create", + lastOperationAt: new Date().toISOString(), + }; + throw new RevConflict("new", 0, 1); + }) + .mockImplementation(updateState); + + const reconciler = new Reconciler({ state, driftDetection: false }); + await reconciler.deploy([ + new CreateResource({ id: "new", config: { name: "new" } }), + ]); + + expect(createSpy).toHaveBeenCalledOnce(); + expect(readSpy).toHaveBeenCalledTimes(2); + expect(state.store.new).toMatchObject({ + rev: 2, + params: { name: "new" }, + output: { name: "new" }, + lastOperation: "drift", + }); + }); + + it("does not blindly retry a conflicted mutation without a read operation", async () => { + const updateSpy = vi.fn(async () => undefined); + const UpdateResource = createTestResourceClass({ + type: "test/service/unreadable-conflict", + 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(), + }, + }); + state.update.mockImplementationOnce(async () => { + state.store.existing = { ...state.store.existing!, rev: 2 }; + throw new RevConflict("existing", 1, 2); + }); + + const reconciler = new Reconciler({ state, driftDetection: false }); + await expect( + reconciler.deploy([ + new UpdateResource({ id: "existing", config: { name: "new" } }), + ]), + ).rejects.toMatchObject({ + id: "existing", + expectedRev: 1, + actualRev: 2, + }); + + expect(updateSpy).toHaveBeenCalledOnce(); + expect(state.update).toHaveBeenCalledOnce(); + }); + it("runs independent resources concurrently per dependency depth", async () => { const marks: Record = {}; @@ -164,6 +408,7 @@ describe("reconciler deploy", () => { const state = createMemoryState({ resource: { + rev: 1, id: "resource", groupId: -1, groupType: "", @@ -207,6 +452,7 @@ describe("reconciler deploy", () => { const state = createMemoryState({ orphan: { + rev: 1, id: "orphan", groupId: -1, groupType: "", @@ -228,7 +474,7 @@ describe("reconciler deploy", () => { await reconciler.deploy([]); expect(deleteSpy).toHaveBeenCalledOnce(); - expect(state.delete).toHaveBeenCalledWith("orphan"); + expect(state.delete).toHaveBeenCalledWith("orphan", 1); }); it("dryRun emits operation intent without applying side effects", async () => { @@ -247,6 +493,7 @@ describe("reconciler deploy", () => { const state = createMemoryState({ orphan: { + rev: 1, id: "orphan", groupId: -1, groupType: "", @@ -267,7 +514,9 @@ describe("reconciler deploy", () => { driftDetection: false, emit: async (event) => { if ("operation" in event) { - operationEvents.push(`${event.operation}:${event.status}:${event.resourceId}`); + operationEvents.push( + `${event.operation}:${event.status}:${event.resourceId}`, + ); } }, }); @@ -286,6 +535,79 @@ describe("reconciler deploy", () => { }); describe("reconciler destroy + refresh", () => { + it("reads remote state after a delete conflict instead of deleting twice", async () => { + let remoteExists = true; + const deleteSpy = vi.fn(async () => { + remoteExists = false; + }); + const readSpy = vi.fn(async () => { + if (!remoteExists) { + const error = new Error("gone"); + error.name = "RemoteMissing"; + throw error; + } + return { name: "doomed" }; + }); + const DestroyResource = createTestResourceClass({ + type: "test/service/destroy-retry", + read: readSpy, + delete: deleteSpy, + notFoundOnError: [{ name: "RemoteMissing", reason: "deleted" }], + }); + const state = createMemoryState({ + doomed: { + rev: 1, + id: "doomed", + groupId: -1, + groupType: "", + type: DestroyResource.type, + config: { name: "doomed" }, + params: { name: "doomed" }, + output: { name: "doomed" }, + lastOperation: "create", + lastOperationAt: new Date().toISOString(), + }, + }); + const deleteState = state.delete.getMockImplementation()!; + state.delete + .mockImplementationOnce(async () => { + state.store.doomed = { ...state.store.doomed!, rev: 2 }; + throw new RevConflict("doomed", 1, 2); + }) + .mockImplementation(deleteState); + + const reconciler = new Reconciler({ state }); + await reconciler.destroy([ + new DestroyResource({ id: "doomed", config: { name: "doomed" } }), + ]); + + expect(deleteSpy).toHaveBeenCalledOnce(); + expect(readSpy).toHaveBeenCalledOnce(); + expect(state.delete).toHaveBeenCalledTimes(2); + expect(state.store.doomed).toBeUndefined(); + }); + + it("holds a backend lease for the orphan snapshot", async () => { + const state = createMemoryState(); + const release = vi.fn(async () => undefined); + const lease = vi.fn(async () => ({ + scope: "reconciler:orphan-deletion", + expiresAt: new Date(Date.now() + 10_000).toISOString(), + renew: vi.fn(async () => new Date(Date.now() + 10_000).toISOString()), + release, + })); + const reconciler = new Reconciler({ + state: { ...state, lease }, + mutationLeaseTtl: 10_000, + }); + + await reconciler.refresh([]); + + expect(lease).toHaveBeenCalledWith("reconciler:orphan-deletion", 10_000); + expect(state.values).toHaveBeenCalledOnce(); + expect(release).toHaveBeenCalledOnce(); + }); + it("destroys resources in reverse dependency order", async () => { const destroyOrder: string[] = []; const deleteA = vi.fn(async () => { @@ -325,6 +647,7 @@ describe("reconciler destroy + refresh", () => { const state = createMemoryState({ a: { + rev: 1, id: "a", groupId: -1, groupType: "", @@ -336,6 +659,7 @@ describe("reconciler destroy + refresh", () => { lastOperationAt: new Date().toISOString(), }, b: { + rev: 1, id: "b", groupId: -1, groupType: "", @@ -347,6 +671,7 @@ describe("reconciler destroy + refresh", () => { lastOperationAt: new Date().toISOString(), }, c: { + rev: 1, id: "c", groupId: -1, groupType: "", @@ -363,9 +688,9 @@ describe("reconciler destroy + refresh", () => { await reconciler.destroy([resourceA, resourceB, resourceC]); expect(destroyOrder).toEqual(["c", "b", "a"]); - expect(state.delete).toHaveBeenCalledWith("a"); - expect(state.delete).toHaveBeenCalledWith("b"); - expect(state.delete).toHaveBeenCalledWith("c"); + expect(state.delete).toHaveBeenCalledWith("a", 1); + expect(state.delete).toHaveBeenCalledWith("b", 1); + expect(state.delete).toHaveBeenCalledWith("c", 1); }); it("refresh removes orphan state entries", async () => { @@ -381,6 +706,7 @@ describe("reconciler destroy + refresh", () => { const keep = new KeepResource({ id: "keep", config: { name: "keep" } }); const state = createMemoryState({ keep: { + rev: 1, id: "keep", groupId: -1, groupType: "", @@ -392,6 +718,7 @@ describe("reconciler destroy + refresh", () => { lastOperationAt: new Date().toISOString(), }, orphan: { + rev: 1, id: "orphan", groupId: -1, groupType: "", @@ -412,8 +739,8 @@ describe("reconciler destroy + refresh", () => { await reconciler.refresh([keep]); expect(deleteSpy).toHaveBeenCalledOnce(); - expect(state.delete).toHaveBeenCalledWith("orphan"); - expect(state.delete).not.toHaveBeenCalledWith("keep"); + expect(state.delete).toHaveBeenCalledWith("orphan", 1); + expect(state.delete).not.toHaveBeenCalledWith("keep", expect.anything()); }); it("destroy and refresh dryRun emit operation events without side effects", async () => { @@ -434,6 +761,7 @@ describe("reconciler destroy + refresh", () => { const state = createMemoryState({ "destroy-me": { + rev: 1, id: "destroy-me", groupId: -1, groupType: "", @@ -445,6 +773,7 @@ describe("reconciler destroy + refresh", () => { lastOperationAt: new Date().toISOString(), }, orphan: { + rev: 1, id: "orphan", groupId: -1, groupType: "", @@ -464,7 +793,9 @@ describe("reconciler destroy + refresh", () => { registry: createResourceRegistry([OrphanResource]), emit: async (event) => { if ("operation" in event) { - operationEvents.push(`${event.operation}:${event.status}:${event.resourceId}`); + operationEvents.push( + `${event.operation}:${event.status}:${event.resourceId}`, + ); } }, }); diff --git a/packages/reconciler/test/reconciler.plan.test.ts b/packages/reconciler/test/reconciler.plan.test.ts index 989fda0..1d5ce49 100644 --- a/packages/reconciler/test/reconciler.plan.test.ts +++ b/packages/reconciler/test/reconciler.plan.test.ts @@ -9,12 +9,14 @@ function createMemoryState(initial: Record = {}) { return { store, get: vi.fn(async (id: string) => store[id]), - update: vi.fn(async (id: string, patch: Partial) => { - store[id] = { - ...(store[id] ?? {}), - ...patch, - } as StateNode; - }), + update: vi.fn( + async (id: string, expectedRev: number, patch: Partial) => { + store[id] = { + ...(store[id] ?? {}), + ...patch, + } as StateNode; + }, + ), delete: vi.fn(async (id: string) => { delete store[id]; }), diff --git a/packages/state-sqlite/package.json b/packages/state-sqlite/package.json new file mode 100644 index 0000000..1ad01b7 --- /dev/null +++ b/packages/state-sqlite/package.json @@ -0,0 +1,20 @@ +{ + "type": "module", + "name": "@notation/state-sqlite", + "version": "0.1.0", + "main": "./dist/index.js", + "types": "./dist/index.d.ts", + "files": [ + "dist" + ], + "scripts": { + "build": "tsup --clean", + "dev": "tsup --watch" + }, + "dependencies": { + "@notation/state": "workspace:*" + }, + "devDependencies": { + "@types/node": "^22.13.4" + } +} diff --git a/packages/state-sqlite/src/index.ts b/packages/state-sqlite/src/index.ts new file mode 100644 index 0000000..057628f --- /dev/null +++ b/packages/state-sqlite/src/index.ts @@ -0,0 +1,195 @@ +import { randomUUID } from "node:crypto"; +import { mkdirSync } from "node:fs"; +import { dirname } from "node:path"; +import { DatabaseSync } from "node:sqlite"; +import { + LeaseConflict, + RevConflict, + type Lease, + type StateBackend, + type StateNode, +} from "@notation/state"; + +export class SqliteStateBackend implements StateBackend { + readonly #database: DatabaseSync; + + constructor(path: string) { + mkdirSync(dirname(path), { recursive: true }); + this.#database = new DatabaseSync(path); + this.#database.exec("PRAGMA busy_timeout = 5000"); + this.#database.exec(` + CREATE TABLE IF NOT EXISTS resources ( + id TEXT PRIMARY KEY, + rev INTEGER NOT NULL, + value TEXT NOT NULL + ) + `); + this.#database.exec(` + CREATE TABLE IF NOT EXISTS resource_leases ( + scope TEXT PRIMARY KEY, + owner TEXT NOT NULL, + expires_at INTEGER NOT NULL + ) + `); + } + + close(): void { + this.#database.close(); + } + + async get(id: string): Promise { + const row = this.#database + .prepare("SELECT value FROM resources WHERE id = ?") + .get(id) as { value: string } | undefined; + return row ? (JSON.parse(row.value) as StateNode) : undefined; + } + + async has(id: string): Promise { + return Boolean( + this.#database + .prepare("SELECT 1 FROM resources WHERE id = ?") + .get(id), + ); + } + + async update( + id: string, + expectedRev: number, + patch: Partial, + ): Promise<{ rev: number }> { + this.#database.exec("BEGIN IMMEDIATE"); + try { + const current = await this.get(id); + // A missing record counts as rev 0, so expectedRev: 0 = "must not exist". + if ((current?.rev ?? 0) !== expectedRev) { + throw new RevConflict(id, expectedRev, current?.rev); + } + + const rev = (current?.rev ?? 0) + 1; + const node = { ...current, ...patch, rev } as StateNode; + if (current) { + const result = this.#database + .prepare( + "UPDATE resources SET rev = ?, value = ? WHERE id = ? AND rev = ?", + ) + .run(rev, JSON.stringify(node), id, current.rev); + if (result.changes !== 1) { + const actual = await this.get(id); + throw new RevConflict(id, current.rev, actual?.rev); + } + } else { + this.#database + .prepare( + "INSERT INTO resources (id, rev, value) VALUES (?, ?, ?)", + ) + .run(id, rev, JSON.stringify(node)); + } + this.#database.exec("COMMIT"); + return { rev }; + } catch (error) { + this.#database.exec("ROLLBACK"); + throw error; + } + } + + async delete(id: string, expectedRev: number): Promise { + const current = await this.get(id); + if ((current?.rev ?? 0) !== expectedRev) { + throw new RevConflict(id, expectedRev, current?.rev); + } + if (!current) return; + + const result = this.#database + .prepare("DELETE FROM resources WHERE id = ? AND rev = ?") + .run(id, current.rev); + if (result.changes !== 1) { + const actual = await this.get(id); + throw new RevConflict(id, current.rev, actual?.rev); + } + } + + async values(): Promise { + const rows = this.#database + .prepare("SELECT value FROM resources ORDER BY id") + .all() as { value: string }[]; + return rows.map(({ value }) => JSON.parse(value) as StateNode); + } + + async lease(scope: string, ttl: number): Promise { + if (!Number.isFinite(ttl) || ttl <= 0) { + throw new RangeError( + "Lease TTL must be a positive number of milliseconds", + ); + } + + const owner = randomUUID(); + const expiresAtMs = Date.now() + ttl; + this.#database.exec("BEGIN IMMEDIATE"); + try { + this.#database + .prepare( + "DELETE FROM resource_leases WHERE scope = ? AND expires_at <= ?", + ) + .run(scope, Date.now()); + const current = this.#database + .prepare("SELECT expires_at FROM resource_leases WHERE scope = ?") + .get(scope) as { expires_at: number } | undefined; + if (current) { + throw new LeaseConflict( + scope, + new Date(current.expires_at).toISOString(), + ); + } + this.#database + .prepare( + "INSERT INTO resource_leases (scope, owner, expires_at) VALUES (?, ?, ?)", + ) + .run(scope, owner, expiresAtMs); + this.#database.exec("COMMIT"); + } catch (error) { + this.#database.exec("ROLLBACK"); + throw error; + } + + let released = false; + let currentExpiresAtMs = expiresAtMs; + return { + scope, + get expiresAt() { + return new Date(currentExpiresAtMs).toISOString(); + }, + renew: async (nextTtl) => { + if (!Number.isFinite(nextTtl) || nextTtl <= 0) { + throw new RangeError( + "Lease TTL must be a positive number of milliseconds", + ); + } + const now = Date.now(); + const nextExpiresAtMs = now + nextTtl; + const result = this.#database + .prepare( + "UPDATE resource_leases SET expires_at = ? WHERE scope = ? AND owner = ? AND expires_at > ?", + ) + .run(nextExpiresAtMs, scope, owner, now); + if (result.changes !== 1) { + const current = this.#database + .prepare("SELECT expires_at FROM resource_leases WHERE scope = ?") + .get(scope) as { expires_at: number } | undefined; + throw new LeaseConflict( + scope, + new Date(current?.expires_at ?? 0).toISOString(), + ); + } + currentExpiresAtMs = nextExpiresAtMs; + return new Date(nextExpiresAtMs).toISOString(); + }, + release: async () => { + if (released) return; + this.#database + .prepare("DELETE FROM resource_leases WHERE scope = ? AND owner = ?") + .run(scope, owner); + released = true; + }, + }; + } +} diff --git a/packages/state-sqlite/src/node-sqlite.d.ts b/packages/state-sqlite/src/node-sqlite.d.ts new file mode 100644 index 0000000..04691f7 --- /dev/null +++ b/packages/state-sqlite/src/node-sqlite.d.ts @@ -0,0 +1,20 @@ +declare module "node:sqlite" { + export type StatementResult = { changes: number | bigint }; + export class StatementSync { + get(...values: unknown[]): unknown; + all(...values: unknown[]): unknown[]; + run(...values: unknown[]): StatementResult; + } + export class DatabaseSync { + constructor(path: string); + exec(sql: string): void; + prepare(sql: string): StatementSync; + close(): void; + } +} + +// The shared tsconfig does not load @types/node, so declare the one export +// this package uses. +declare module "node:crypto" { + export function randomUUID(): string; +} diff --git a/packages/state-sqlite/test/state-sqlite.test.ts b/packages/state-sqlite/test/state-sqlite.test.ts new file mode 100644 index 0000000..31eb35d --- /dev/null +++ b/packages/state-sqlite/test/state-sqlite.test.ts @@ -0,0 +1,122 @@ +import { mkdtemp, rm } from "node:fs/promises"; +import { spawn } from "node:child_process"; +import { once } from "node:events"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; +import { SqliteStateBackend } from "../src"; + +const cleanups: (() => Promise)[] = []; +afterEach(async () => + Promise.all(cleanups.splice(0).map((cleanup) => cleanup())), +); + +async function createBackend() { + const directory = await mkdtemp(path.join(tmpdir(), "notation-sqlite-")); + const backend = new SqliteStateBackend(path.join(directory, "state.db")); + cleanups.push(async () => { + backend.close(); + await rm(directory, { recursive: true, force: true }); + }); + return backend; +} + +describe("SqliteStateBackend", () => { + it("persists revisions and enforces compare-and-swap", async () => { + const backend = await createBackend(); + await expect( + backend.update("service", 0, { + id: "service", + type: "test/service/main", + config: {}, + params: {}, + output: {}, + lastOperation: "create", + lastOperationAt: "2026-07-15T00:00:00.000Z", + }), + ).resolves.toEqual({ rev: 1 }); + await expect( + backend.update("service", 1, { output: { ready: true } }), + ).resolves.toEqual({ + rev: 2, + }); + await expect(backend.delete("service", 1)).rejects.toMatchObject({ + name: "RevConflict", + actualRev: 2, + }); + }); + + it("coordinates leases across backend instances and releases by owner", async () => { + const directory = await mkdtemp( + path.join(tmpdir(), "notation-sqlite-lease-"), + ); + const databasePath = path.join(directory, "state.db"); + const first = new SqliteStateBackend(databasePath); + const second = new SqliteStateBackend(databasePath); + cleanups.push(async () => { + first.close(); + second.close(); + await rm(directory, { recursive: true, force: true }); + }); + + const lease = await first.lease("orphans", 10_000); + await expect(second.lease("orphans", 10_000)).rejects.toMatchObject({ + name: "LeaseConflict", + scope: "orphans", + }); + const firstExpiry = lease.expiresAt; + await lease.renew(20_000); + expect(lease.expiresAt).not.toBe(firstExpiry); + await lease.release(); + const nextLease = await second.lease("orphans", 10_000); + expect(nextLease).toMatchObject({ scope: "orphans" }); + await nextLease.release(); + }); + + it("waits for a concurrent writer instead of raising database locked", async () => { + const directory = await mkdtemp( + path.join(tmpdir(), "notation-sqlite-busy-"), + ); + const databasePath = path.join(directory, "state.db"); + const backend = new SqliteStateBackend(databasePath); + cleanups.push(async () => { + backend.close(); + await rm(directory, { recursive: true, force: true }); + }); + + const blocker = spawn( + process.execPath, + [ + "--input-type=module", + "--eval", + ` + import { DatabaseSync } from "node:sqlite"; + const database = new DatabaseSync(process.argv[1]); + database.exec("BEGIN IMMEDIATE"); + process.stdout.write("locked\\n"); + setTimeout(() => { + database.exec("ROLLBACK"); + database.close(); + }, 100); + `, + databasePath, + ], + { stdio: ["ignore", "pipe", "inherit"] }, + ); + const blockerExited = once(blocker, "exit"); + await once(blocker.stdout!, "data"); + + await expect( + backend.update("service", 0, { + id: "service", + type: "test/service/main", + config: {}, + params: {}, + output: {}, + lastOperation: "create", + lastOperationAt: "2026-07-15T00:00:00.000Z", + }), + ).resolves.toEqual({ rev: 1 }); + await blockerExited; + }); +}); diff --git a/packages/state-sqlite/tsconfig.json b/packages/state-sqlite/tsconfig.json new file mode 100644 index 0000000..13487e3 --- /dev/null +++ b/packages/state-sqlite/tsconfig.json @@ -0,0 +1,7 @@ +{ + "extends": "tsconfig/base.json", + "compilerOptions": { + "types": ["node"] + }, + "include": ["src", "test"] +} diff --git a/packages/state-sqlite/tsup.config.ts b/packages/state-sqlite/tsup.config.ts new file mode 100644 index 0000000..a61d2e2 --- /dev/null +++ b/packages/state-sqlite/tsup.config.ts @@ -0,0 +1,10 @@ +import { defineConfig } from "tsup"; + +export default defineConfig({ + entry: ["src/index.ts"], + format: ["esm"], + dts: true, + sourcemap: true, + // node:sqlite only resolves with the node: prefix; don't let tsup strip it. + removeNodeProtocol: false, +}); diff --git a/packages/state/src/conflicts.ts b/packages/state/src/conflicts.ts new file mode 100644 index 0000000..81559dc --- /dev/null +++ b/packages/state/src/conflicts.ts @@ -0,0 +1,24 @@ +export class RevConflict extends Error { + readonly name = "RevConflict"; + + constructor( + readonly id: string, + readonly expectedRev: number, + readonly actualRev: number | undefined, + ) { + super( + `State revision conflict for ${id}: expected ${expectedRev}, got ${actualRev ?? "missing"}`, + ); + } +} + +export class LeaseConflict extends Error { + readonly name = "LeaseConflict"; + + constructor( + readonly scope: string, + readonly expiresAt: string, + ) { + super(`State lease conflict for ${scope}: held until ${expiresAt}`); + } +} diff --git a/packages/state/src/index.ts b/packages/state/src/index.ts index be6f715..fd99403 100644 --- a/packages/state/src/index.ts +++ b/packages/state/src/index.ts @@ -1 +1,2 @@ export * from "./state"; +export * from "./conflicts"; diff --git a/packages/state/src/state.ts b/packages/state/src/state.ts index 71106dc..003faa1 100644 --- a/packages/state/src/state.ts +++ b/packages/state/src/state.ts @@ -1,7 +1,18 @@ -import { mkdir, readFile, rename, unlink, writeFile } from "node:fs/promises"; +import { randomUUID } from "node:crypto"; +import { + mkdir, + readFile, + rename, + stat, + unlink, + writeFile, +} from "node:fs/promises"; import path from "node:path"; +import { setTimeout as sleep } from "node:timers/promises"; +import { LeaseConflict, RevConflict } from "./conflicts"; export type StateNode = { + rev: number; id: string; type: string; config: Record; @@ -15,15 +26,32 @@ export type StateNode = { export interface StateBackend { get(id: string): Promise; has(id: string): Promise; - update(id: string, patch: Partial): Promise; - delete(id: string): Promise; + /** + * The stored revision must match expectedRev. A missing record counts as + * revision 0, so expectedRev: 0 asserts that the record does not exist yet. + */ + update( + id: string, + expectedRev: number, + patch: Partial, + ): Promise<{ rev: number }>; + delete(id: string, expectedRev: number): Promise; values(): Promise; -}; + lease(scope: string, ttl: number): Promise; +} + +export interface Lease { + readonly scope: string; + readonly expiresAt: string; + renew(ttl: number): Promise; + release(): Promise; +} export type State = StateBackend; export class MemoryStateBackend implements StateBackend { #state: Record; + #leases = new Map(); constructor(initialState: Record = {}) { this.#state = cloneAsPersistedState(initialState); @@ -39,17 +67,26 @@ export class MemoryStateBackend implements StateBackend { return id in state; } - async update(id: string, patch: Partial): Promise { + async update( + id: string, + expectedRev: number, + patch: Partial, + ): Promise<{ rev: number }> { const state = await this.readState(); + assertExpectedRev(id, state[id], expectedRev); + const rev = (state[id]?.rev ?? 0) + 1; state[id] = { ...state[id], ...patch, + rev, } as StateNode; await this.writeState(state); + return { rev }; } - async delete(id: string): Promise { + async delete(id: string, expectedRev: number): Promise { const state = await this.readState(); + assertExpectedRev(id, state[id], expectedRev); delete state[id]; await this.writeState(state); } @@ -71,6 +108,46 @@ export class MemoryStateBackend implements StateBackend { .map(([, value]) => value); } + async lease(scope: string, ttl: number): Promise { + assertLeaseTtl(ttl); + const now = Date.now(); + const current = this.#leases.get(scope); + if (current && current.expiresAtMs > now) { + throw new LeaseConflict( + scope, + new Date(current.expiresAtMs).toISOString(), + ); + } + + const owner = randomUUID(); + let expiresAtMs = now + ttl; + this.#leases.set(scope, { owner, expiresAtMs }); + + return { + scope, + get expiresAt() { + return new Date(expiresAtMs).toISOString(); + }, + renew: async (nextTtl) => { + assertLeaseTtl(nextTtl); + const held = this.#leases.get(scope); + if (!held || held.owner !== owner || held.expiresAtMs <= Date.now()) { + throw new LeaseConflict( + scope, + new Date(held?.expiresAtMs ?? 0).toISOString(), + ); + } + expiresAtMs = Date.now() + nextTtl; + held.expiresAtMs = expiresAtMs; + return new Date(expiresAtMs).toISOString(); + }, + release: async () => { + if (this.#leases.get(scope)?.owner === owner) + this.#leases.delete(scope); + }, + }; + } + private async readState(): Promise> { return cloneAsPersistedState(this.#state); } @@ -80,6 +157,10 @@ export class MemoryStateBackend implements StateBackend { } } +const FILE_LOCK_STALE_MS = 10_000; +const FILE_LOCK_TIMEOUT_MS = 5_000; +const FILE_LOCK_RETRY_MS = 25; + export class FileStateBackend implements StateBackend { constructor(private readonly stateFilePath: string) {} @@ -93,19 +174,32 @@ export class FileStateBackend implements StateBackend { return id in state; } - async update(id: string, patch: Partial): Promise { - const state = await this.readState(); - state[id] = { - ...state[id], - ...patch, - } as StateNode; - await this.writeState(state); + async update( + id: string, + expectedRev: number, + patch: Partial, + ): Promise<{ rev: number }> { + return this.withLock(async () => { + const state = await this.readState(); + assertExpectedRev(id, state[id], expectedRev); + const rev = (state[id]?.rev ?? 0) + 1; + state[id] = { + ...state[id], + ...patch, + rev, + } as StateNode; + await this.writeState(state); + return { rev }; + }); } - async delete(id: string): Promise { - const state = await this.readState(); - delete state[id]; - await this.writeState(state); + async delete(id: string, expectedRev: number): Promise { + await this.withLock(async () => { + const state = await this.readState(); + assertExpectedRev(id, state[id], expectedRev); + delete state[id]; + await this.writeState(state); + }); } async values(): Promise { @@ -113,6 +207,65 @@ export class FileStateBackend implements StateBackend { return Object.values(state); } + async lease(scope: string, ttl: number): Promise { + assertLeaseTtl(ttl); + const leaseFilePath = `${this.stateFilePath}.${encodeURIComponent(scope)}.lease`; + const owner = randomUUID(); + let expiresAtMs: number; + await mkdir(path.dirname(this.stateFilePath), { recursive: true }); + + for (;;) { + expiresAtMs = Date.now() + ttl; + try { + await writeFile(leaseFilePath, JSON.stringify({ owner, expiresAtMs }), { + flag: "wx", + }); + break; + } catch (error) { + if (!isFileExistsError(error)) throw error; + const current = await readFileLease(leaseFilePath); + if (!current || current.expiresAtMs <= Date.now()) { + await unlink(leaseFilePath).catch(() => undefined); + continue; + } + throw new LeaseConflict( + scope, + new Date(current.expiresAtMs).toISOString(), + ); + } + } + + return { + scope, + get expiresAt() { + return new Date(expiresAtMs).toISOString(); + }, + renew: async (nextTtl) => { + assertLeaseTtl(nextTtl); + const current = await readFileLease(leaseFilePath); + if ( + !current || + current.owner !== owner || + current.expiresAtMs <= Date.now() + ) { + throw new LeaseConflict( + scope, + new Date(current?.expiresAtMs ?? 0).toISOString(), + ); + } + expiresAtMs = Date.now() + nextTtl; + await writeFile(leaseFilePath, JSON.stringify({ owner, expiresAtMs })); + return new Date(expiresAtMs).toISOString(); + }, + release: async () => { + const current = await readFileLease(leaseFilePath); + if (current?.owner === owner) { + await unlink(leaseFilePath).catch(() => undefined); + } + }, + }; + } + private async readState(): Promise> { try { const file = await readFile(this.stateFilePath, "utf8"); @@ -126,13 +279,54 @@ export class FileStateBackend implements StateBackend { } } + /** + * The read-check-write in update/delete is only safe if no other process + * interleaves, so writers hold an exclusive lock file. A lock older than + * FILE_LOCK_STALE_MS is treated as abandoned by a crashed process. + */ + private async withLock(fn: () => Promise): Promise { + const lockFilePath = `${this.stateFilePath}.lock`; + await mkdir(path.dirname(this.stateFilePath), { recursive: true }); + + const deadline = Date.now() + FILE_LOCK_TIMEOUT_MS; + for (;;) { + try { + await writeFile( + lockFilePath, + JSON.stringify({ pid: process.pid, acquiredAt: Date.now() }), + { flag: "wx" }, + ); + break; + } catch (error) { + if (!isFileExistsError(error)) throw error; + const lockStat = await stat(lockFilePath).catch(() => undefined); + if (lockStat && Date.now() - lockStat.mtimeMs > FILE_LOCK_STALE_MS) { + await unlink(lockFilePath).catch(() => undefined); + continue; + } + if (Date.now() > deadline) { + throw new Error( + `Timed out acquiring state lock at ${lockFilePath}; delete it if no other deploy is running`, + ); + } + await sleep(FILE_LOCK_RETRY_MS); + } + } + + try { + return await fn(); + } finally { + await unlink(lockFilePath).catch(() => undefined); + } + } + private async writeState(state: Record): Promise { const directory = path.dirname(this.stateFilePath); await mkdir(directory, { recursive: true }); const tempFilePath = path.join( directory, - `${path.basename(this.stateFilePath)}.${process.pid}.${Date.now()}.tmp`, + `${path.basename(this.stateFilePath)}.${randomUUID()}.tmp`, ); const serialized = `${JSON.stringify(state, null, 2)}\n`; @@ -148,12 +342,51 @@ export class FileStateBackend implements StateBackend { } } +// A missing record counts as rev 0, so expectedRev: 0 means "must not exist". +function assertExpectedRev( + id: string, + node: StateNode | undefined, + expectedRev: number, +): void { + if ((node?.rev ?? 0) !== expectedRev) { + throw new RevConflict(id, expectedRev, node?.rev); + } +} + +function assertLeaseTtl(ttl: number): void { + if (!Number.isFinite(ttl) || ttl <= 0) { + throw new RangeError("Lease TTL must be a positive number of milliseconds"); + } +} + +type FileLeaseRecord = { owner: string; expiresAtMs: number }; + +async function readFileLease( + filePath: string, +): Promise { + try { + return JSON.parse(await readFile(filePath, "utf8")) as FileLeaseRecord; + } catch (error) { + if (isFileMissingError(error) || error instanceof SyntaxError) + return undefined; + throw error; + } +} + function isFileMissingError(error: unknown): boolean { + return isErrorWithCode(error, "ENOENT"); +} + +function isFileExistsError(error: unknown): boolean { + return isErrorWithCode(error, "EEXIST"); +} + +function isErrorWithCode(error: unknown, code: string): boolean { return ( typeof error === "object" && error !== null && "code" in error && - error.code === "ENOENT" + error.code === code ); } diff --git a/packages/state/test/state-backend.test.ts b/packages/state/test/state-backend.test.ts index e7bbb69..fde5e5b 100644 --- a/packages/state/test/state-backend.test.ts +++ b/packages/state/test/state-backend.test.ts @@ -19,6 +19,7 @@ function createStateNode( overrides: Partial = {}, ): StateNode { return { + rev: 0, id, groupId: 1, groupType: "stack", @@ -54,14 +55,15 @@ function runStateBackendContractTests( const initialNode = createStateNode("resource-a"); try { - await fixture.backend.update(initialNode.id, initialNode); - await fixture.backend.update(initialNode.id, { + await fixture.backend.update(initialNode.id, 0, initialNode); + await fixture.backend.update(initialNode.id, 1, { output: { status: "ready" }, lastOperation: "update", }); await expect(fixture.backend.get(initialNode.id)).resolves.toEqual({ ...initialNode, + rev: 2, output: { status: "ready" }, lastOperation: "update", }); @@ -70,13 +72,58 @@ function runStateBackendContractTests( } }); + it("rejects stale updates and deletes", async () => { + const fixture = await createBackend(); + const initialNode = createStateNode("resource-a"); + + try { + await expect( + fixture.backend.update(initialNode.id, 0, initialNode), + ).resolves.toEqual({ rev: 1 }); + await expect( + fixture.backend.update(initialNode.id, 0, { output: {} }), + ).rejects.toMatchObject({ + name: "RevConflict", + expectedRev: 0, + actualRev: 1, + }); + await expect( + fixture.backend.delete(initialNode.id, 0), + ).rejects.toMatchObject({ + name: "RevConflict", + }); + } finally { + await fixture.cleanup(); + } + }); + + it("treats expectedRev 0 as an expect-absent assertion", async () => { + const fixture = await createBackend(); + const initialNode = createStateNode("resource-a"); + + try { + await expect( + fixture.backend.update(initialNode.id, 0, initialNode), + ).resolves.toEqual({ rev: 1 }); + await expect( + fixture.backend.update(initialNode.id, 0, initialNode), + ).rejects.toMatchObject({ + name: "RevConflict", + expectedRev: 0, + actualRev: 1, + }); + } finally { + await fixture.cleanup(); + } + }); + it("deletes values", async () => { const fixture = await createBackend(); const initialNode = createStateNode("resource-a"); try { - await fixture.backend.update(initialNode.id, initialNode); - await fixture.backend.delete(initialNode.id); + await fixture.backend.update(initialNode.id, 0, initialNode); + await fixture.backend.delete(initialNode.id, 1); await expect( fixture.backend.get(initialNode.id), @@ -94,13 +141,39 @@ function runStateBackendContractTests( const secondNode = createStateNode("resource-b"); try { - await fixture.backend.update(firstNode.id, firstNode); - await fixture.backend.update(secondNode.id, secondNode); + await fixture.backend.update(firstNode.id, 0, firstNode); + await fixture.backend.update(secondNode.id, 0, secondNode); const values = await fixture.backend.values(); expect(values).toHaveLength(2); - expect(values).toEqual(expect.arrayContaining([firstNode, secondNode])); + expect(values).toEqual( + expect.arrayContaining([ + { ...firstNode, rev: 1 }, + { ...secondNode, rev: 1 }, + ]), + ); + } finally { + await fixture.cleanup(); + } + }); + + it("holds and renews an exclusive lease", async () => { + const fixture = await createBackend(); + + try { + const lease = await fixture.backend.lease("resource:a", 1_000); + const firstExpiry = lease.expiresAt; + await expect( + fixture.backend.lease("resource:a", 1_000), + ).rejects.toMatchObject({ name: "LeaseConflict" }); + + await lease.renew(2_000); + expect(lease.expiresAt).not.toBe(firstExpiry); + await lease.release(); + + const next = await fixture.backend.lease("resource:a", 1_000); + await next.release(); } finally { await fixture.cleanup(); } @@ -121,15 +194,50 @@ runStateBackendContractTests("MemoryStateBackend", async () => ({ cleanup: async () => undefined, })); +describe("FileStateBackend", () => { + it("serialises concurrent CAS writers so only one wins", async () => { + const tempDirectory = await mkdtemp(path.join(tmpdir(), "notation-state-")); + const statePath = path.join(tempDirectory, "state.json"); + const first = new FileStateBackend(statePath); + const second = new FileStateBackend(statePath); + const initialNode = createStateNode("resource-a"); + + try { + await first.update(initialNode.id, 0, initialNode); + + const results = await Promise.allSettled([ + first.update(initialNode.id, 1, { output: { writer: "first" } }), + second.update(initialNode.id, 1, { output: { writer: "second" } }), + ]); + + const fulfilled = results.filter((r) => r.status === "fulfilled"); + const rejected = results.filter((r) => r.status === "rejected"); + expect(fulfilled).toHaveLength(1); + expect(rejected).toHaveLength(1); + expect((rejected[0] as PromiseRejectedResult).reason).toMatchObject({ + name: "RevConflict", + }); + await expect(first.get(initialNode.id)).resolves.toMatchObject({ + rev: 2, + }); + } finally { + await rm(tempDirectory, { recursive: true, force: true }); + } + }); +}); + describe("MemoryStateBackend", () => { it("returns values in deterministic id order", async () => { const backend = new MemoryStateBackend(); const laterNode = createStateNode("resource-z"); const earlierNode = createStateNode("resource-a"); - await backend.update(laterNode.id, laterNode); - await backend.update(earlierNode.id, earlierNode); + await backend.update(laterNode.id, 0, laterNode); + await backend.update(earlierNode.id, 0, earlierNode); - await expect(backend.values()).resolves.toEqual([earlierNode, laterNode]); + await expect(backend.values()).resolves.toEqual([ + { ...earlierNode, rev: 1 }, + { ...laterNode, rev: 1 }, + ]); }); }); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index a76bfc0..8a98b39 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -205,6 +205,9 @@ importers: '@notation/state': specifier: workspace:* version: link:../state + '@notation/state-sqlite': + specifier: workspace:* + version: link:../state-sqlite deep-object-diff: specifier: ^1.1.9 version: 1.1.9 @@ -335,6 +338,16 @@ importers: specifier: ^22.13.4 version: 22.13.4 + packages/state-sqlite: + dependencies: + '@notation/state': + specifier: workspace:* + version: link:../state + devDependencies: + '@types/node': + specifier: ^22.13.4 + version: 22.13.4 + packages/std.iac: dependencies: '@notation/core':