Skip to content

Harden durable Yieldstar reconciliation#37

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
djgrant wants to merge 1 commit into from
Closed

Harden durable Yieldstar reconciliation #37

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions docs/manual/reconciler.md
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -29,6 +29,8 @@ The outer workflow supplies durable step execution, timers, shared stores, waiti

Each live resource is one YieldStar store. Absence is represented by no store, not a tombstone. YieldStar's UUIDv7 store `instanceId` and version are authoritative for conditional update and delete; Notation exposes the version as the resource state's `rev` for its existing state contract.

Deployments against the same `deploymentId` are serialized through a coordination store keyed by `executionId`. If a deployment crashes while holding the coordination store, resume it by running the same execution ID again: replay reclaims the acquisition through YieldStar's applied-step ledger and releases it on completion. A different execution ID waits durably until the holder releases.

Pass the complete desired set on every invocation. Persisted resources absent from that set are deleted through the supplied resource registry.

The runnable Node SQLite version is in `examples/reconciler`.
94 changes: 57 additions & 37 deletions packages/reconciler/src/yieldstar.ts
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -177,13 +177,21 @@ async function* reconcileResource(
resource,
stateNode: stateNode ?? undefined,
params,
driftRead:
remote.status === "not-found"
? remote
: { status: "found", output: remote.output },
driftRead: remote,
});
}

if (action.decision === "drift-update") {
const diff = action.patch;
yield* emitDurably(step, `${prefix}:drift-detected`, opts.emit, () => ({
level: "info",
event: "reconciler.drift.detected",
resourceId: resource.id,
resourceType: resource.type,
diff,
}));
}

yield* emitDurably(step, `${prefix}:decision`, opts.emit, () => ({
level: "info",
event: "reconciler.deploy.decision",
Expand All@@ -206,7 +214,18 @@ async function* reconcileResource(
resource.setOutput(params);
if (primaryKey) resource.setOutput({ ...primaryKey, ...resource.output });
} else {
if (!resource.update) return;
if (!resource.update) {
yield* emitDurably(step, `${prefix}:update-skip`, opts.emit, () => ({
level: "info",
event: "reconciler.operation.lifecycle",
operation: "update",
status: "skip",
resourceId: resource.id,
resourceType: resource.type,
reason: "update-not-implemented",
}));
return;
}
yield* runProviderCall(
step,
`${prefix}:update`,
Expand All@@ -231,7 +250,6 @@ async function* reconcileResource(
);
if (read.status === "found")
resource.setOutput({ ...resource.output, ...read.output });
if (opts.dryRun) return;

const operation =
action.decision === "create" || action.decision === "drift-recreate"
Expand DownExpand Up@@ -406,15 +424,31 @@ export class YieldStarStateBackend {
}

async get(id: string): Promise<StateNode | undefined> {
const storeId = this.storeId(id);
const ids = await this.#client.listStores(yieldStarResourceStateStore);
if (!ids.includes(storeId)) return undefined;
return toStateNode(
await this.#client.getStore({
const snapshot = await this.#tryGetSnapshot(this.storeId(id));
return snapshot ? toStateNode(snapshot) : undefined;
}

/**
* Reads a store snapshot in one round trip. A missing store is resource
* absence, so a read failure is re-checked against the store listing before
* it is allowed to propagate.
*/
async #tryGetSnapshot(
storeId: string,
): Promise<
| { state: StoredResourceState; instanceId: string; version: number }
| undefined
> {
try {
return await this.#client.getStore({
definition: yieldStarResourceStateStore,
id: storeId,
}),
);
});
} catch (error) {
const ids = await this.#client.listStores(yieldStarResourceStateStore);
if (!ids.includes(storeId)) return undefined;
throw error;
}
}

async has(id: string): Promise<boolean> {
Expand All@@ -427,8 +461,8 @@ export class YieldStarStateBackend {
patch: Partial<StateNode>,
): Promise<{ rev: number }> {
const storeId = this.storeId(id);
const ids = await this.#client.listStores(yieldStarResourceStateStore);
if (!ids.includes(storeId)) {
const snapshot = await this.#tryGetSnapshot(storeId);
if (!snapshot) {
if (expectedRev !== 0) throw new RevConflict(id, expectedRev, undefined);
const initial = { ...patch, id } as StoredResourceState;
const created = await this.#client.getOrCreateStore({
Expand All@@ -439,14 +473,9 @@ export class YieldStarStateBackend {
return { rev: created.version + 1 };
}

const snapshot = await this.#client.getStore({
definition: yieldStarResourceStateStore,
id: storeId,
});
const actualRev = snapshot.version + 1;
if (actualRev !== expectedRev)
throw new RevConflict(id, expectedRev, actualRev);
const rev = actualRev + 1;
const result = await this.#client.updateStoreFrom({
definition: yieldStarResourceStateStore,
id: storeId,
Expand All@@ -456,20 +485,16 @@ export class YieldStarStateBackend {
},
});
if (!result.updated) throw new RevConflict(id, expectedRev, undefined);
return { rev };
return { rev: result.version + 1 };
}

async delete(id: string, expectedRev: number): Promise<void> {
const storeId = this.storeId(id);
const ids = await this.#client.listStores(yieldStarResourceStateStore);
if (!ids.includes(storeId)) {
const snapshot = await this.#tryGetSnapshot(storeId);
if (!snapshot) {
if (expectedRev !== 0) throw new RevConflict(id, expectedRev, undefined);
return;
}
const snapshot = await this.#client.getStore({
definition: yieldStarResourceStateStore,
id: storeId,
});
const actualRev = snapshot.version + 1;
if (actualRev !== expectedRev)
throw new RevConflict(id, expectedRev, actualRev);
Expand All@@ -484,19 +509,14 @@ export class YieldStarStateBackend {
async values(): Promise<StateNode[]> {
const prefix = `${this.#deploymentId}:`;
const ids = await this.#client.listStores(yieldStarResourceStateStore);
const nodes = await Promise.all(
const snapshots = await Promise.all(
ids
.filter((id) => id.startsWith(prefix))
.map(async (id) =>
toStateNode(
await this.#client.getStore({
definition: yieldStarResourceStateStore,
id,
}),
),
),
.map((id) => this.#tryGetSnapshot(id)),
);
return nodes;
return snapshots
.filter((snapshot) => snapshot !== undefined)
.map(toStateNode);
}

snapshot(id: string) {
Expand Down
95 changes: 86 additions & 9 deletions packages/reconciler/test/yieldstar.integration.test.ts
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -17,6 +17,11 @@ import {
reconcileWithYieldStar,
yieldStarResourceStateStore,
} from "../src/yieldstar";
import type { ReconcilerEvent } from "../src/reconciler";
import {
createResourceRegistry,
type ResourceRegistry,
} from "../src/resource-registry";

const logger = pino({ level: "silent" });

Expand All@@ -42,7 +47,7 @@ describe("YieldStar reconciliation", () => {
const runtime = createRuntime(
[new PendingResource({ id: "pending" })],
"durable-wait",
{ maxAttempts: 3, retryInterval: 1 },
{ retryOptions: { maxAttempts: 3, retryInterval: 1 } },
);

await runtime.run("wait-execution");
Expand All@@ -67,8 +72,7 @@ describe("YieldStar reconciliation", () => {
const runtime = createRuntime(
[new TestResource({ id: "resume" })],
"crash-resume",
undefined,
"notation:resource:resume:create",
{ crashAfterStep: "notation:resource:resume:create" },
);

await expect(runtime.run("resume-execution")).rejects.toThrow(
Expand DownExpand Up@@ -160,19 +164,90 @@ describe("YieldStar reconciliation", () => {
expect(await runtime.state.values()).toHaveLength(1);
runtime.close();
});

it("deletes orphaned resources through the registry on a later deployment", async () => {
const deleteSpy = vi.fn(async () => undefined);
const OrphanResource = resource({ type: "test/yieldstar/orphan" })
.defineSchema({})
.defineOperations({ create: async () => undefined, delete: deleteSpy });
const resources: BaseResource[] = [new OrphanResource({ id: "orphan" })];
const runtime = createRuntime(resources, "orphan-deletion", {
registry: createResourceRegistry([OrphanResource]),
});

await runtime.run("deploy-1");
expect(await runtime.state.values()).toHaveLength(1);

resources.length = 0;
await runtime.run("deploy-2");

expect(deleteSpy).toHaveBeenCalledOnce();
expect(await runtime.state.values()).toHaveLength(0);
expect(await runtime.state.get("orphan")).toBeUndefined();
runtime.close();
});

it("emits drift detection and repairs remote drift with update", async () => {
let remote = { name: "expected" };
const updateSpy = vi.fn(async () => {
remote = { name: "expected" };
});
const DriftResource = resource({ type: "test/yieldstar/drift" })
.defineSchema({
name: {
presence: "required",
propertyType: "param",
valueType: "string" as any,
},
})
.defineOperations({
create: async () => remote,
read: async () => remote,
update: updateSpy,
delete: async () => undefined,
});
const events: ReconcilerEvent[] = [];
const runtime = createRuntime(
[new DriftResource({ id: "drifted", config: { name: "expected" } })],
"drift-repair",
{ driftDetection: true, emit: (event) => void events.push(event) },
);

await runtime.run("deploy-1");
remote = { name: "drifted" };
await runtime.run("deploy-2");

expect(updateSpy).toHaveBeenCalledOnce();
expect(
events.find((event) => event.event === "reconciler.drift.detected"),
).toMatchObject({ resourceId: "drifted", diff: { name: "expected" } });
expect(
events.filter(
(event) =>
event.event === "reconciler.deploy.decision" &&
event.decision === "drift-update",
),
).toHaveLength(1);
runtime.close();
});
});

function createRuntime(
resources: BaseResource[],
deploymentId: string,
retryOptions?: { maxAttempts: number; retryInterval: number },
crashAfterStep?: string,
options: {
retryOptions?: { maxAttempts: number; retryInterval: number };
crashAfterStep?: string;
registry?: ResourceRegistry;
driftDetection?: boolean;
emit?: (event: ReconcilerEvent) => void;
} = {},
) {
const database = createSqliteDb({ path: ":memory:" });
const scheduler = new TestScheduler();
const sqliteHeap = new SqliteHeapClient(database);
const heap = crashAfterStep
? new CrashAfterWriteHeap(sqliteHeap, crashAfterStep)
const heap = options.crashAfterStep
? new CrashAfterWriteHeap(sqliteHeap, options.crashAfterStep)
: sqliteHeap;
const storeClient = new SqliteStoreClient({
db: database,
Expand All@@ -185,8 +260,10 @@ function createRuntime(
executionId: event.executionId,
resources,
state,
driftDetection: false,
retryOptions,
registry: options.registry,
driftDetection: options.driftDetection ?? false,
emit: options.emit,
retryOptions: options.retryOptions,
});
});
const router = createWorkflowRouter({ deploy });
Expand Down
Loading