Coordinate concurrent reconciler state

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",

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,

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

packages/core/src/provisioner/operations/operation.update.ts

@@ -11,6 +11,10 @@ async function update(opts: {
  patch: any;
 }): Promise<void> {
  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),

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

packages/core/test/provisioner/operation.create.test.ts

@@ -31,6 +31,7 @@ describe("resource creation", () => {
  createResourceOperation(step, {
  resource: testResource,
  state: stateBackend,
+ expectedRev: 0,
  }),
  );
 

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

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,

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");

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<string, unknown>;
+ expectedRev: number;
 };
 
-export type DeleteResourceParams = ResourceOperationBaseParams;
+export type DeleteResourceParams = ResourceOperationBaseParams & {
+ expectedRev: number;
+};
 
 export const DEFAULT_RETRY_OPTIONS: PollOptions = {
  maxAttempts: 10,

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

packages/reconciler/src/plan.ts

@@ -43,7 +43,11 @@ export type ResourceAction =
  | { decision: "noop" }
  | { decision: "drift-recreate" }
  | { decision: "update"; patch: Record<string, unknown>; diff: PlanDiff }
- | { decision: "drift-update"; patch: Record<string, unknown>; diff: PlanDiff };
+ | {
+ decision: "drift-update";
+ patch: Record<string, unknown>;
+ 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<string, unknown>;
-
- 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(