Review 2: harden Yieldstar deployment scoping and coordination visibility

docs/internals/reconciler.md

@@ -33,8 +33,8 @@ Every provider call, event emission, state read, state write, and coordination t
 
 Each resource is stored under `notation/resource-state` with a deployment-scoped ID. Conditional updates and deletes compare the snapshot's UUIDv7 `instanceId` and version, so a stale execution cannot modify a deleted and recreated store.
 
-Deploy and destroy share one `notation/deployment-coordination` store per deployment. `store.take` suspends a competing execution as a durable waiter and wakes it after the holder releases.
+Deploy and destroy share one `notation/deployment-coordination` store per deployment. `store.take` suspends a competing execution as a durable waiter and wakes it after the holder releases. Before suspending, the waiter emits `reconciler.coordination.waiting` naming the holding execution ID, so a wait behind a crashed execution is visible instead of silent.
 
 ## Events
 
-The durable workflows emit `reconciler.deploy.decision`, `reconciler.drift.detected`, `reconciler.operation.lifecycle`, and `reconciler.orphan-deletion.skipped`. Lifecycle events cover create, read, update, and delete with `start`, `success`, `error`, `skip`, or `dry-run` status.
+The durable workflows emit `reconciler.deploy.decision`, `reconciler.drift.detected`, `reconciler.operation.lifecycle`, `reconciler.coordination.waiting`, and `reconciler.orphan-deletion.skipped`. Lifecycle events cover create, read, update, and delete with `start`, `success`, `error`, `skip`, or `dry-run` status.

docs/manual/reconciler.md

@@ -38,7 +38,7 @@ 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`.
 
-Operations against the same `deploymentId` are serialized through a coordination store keyed by `executionId`. Resume a crashed operation with the same execution ID; use a new globally unique execution ID for every new deploy or destroy.
+Operations against the same `deploymentId` are serialized through a coordination store keyed by `executionId`. Resume a crashed operation with the same execution ID; use a new globally unique execution ID for every new deploy or destroy. An execution that must wait emits a `reconciler.coordination.waiting` event naming the holder before it suspends, which also identifies a crashed holder that should be resumed instead.
 
 Pass the complete desired set on every deployment. Persisted resources absent from that set are deleted through the supplied resource registry. Destroy removes current resources in reverse dependency order and then removes any persisted orphans that the registry can hydrate.
 

packages/cli/src/destroy.ts

@@ -27,5 +27,19 @@ export async function destroy(
  logger.info(`Destroying ${entryPoint}\n`);
  const executionId = opts.executionId ?? randomUUID();
  logger.info(`YieldStar execution ${executionId}`);
- await destroyApp({ entryPoint, emit, executionId });
+
+ try {
+ await destroyApp({ entryPoint, emit, executionId });
+ } catch (err: any) {
+ if (err.name === "CredentialsProviderError") {
+ logger.error(
+ "\nAWS credentials not found.",
+ "\n\nEnsure you have a default profile set up in ~/.aws/credentials.",
+ "\n\nIf using another profile run AWS_PROFILE=otherProfile notation destroy.\n",
+ );
+ process.exit(1);
+ }
+ logger.error(err);
+ process.exit(1);
+ }
 }

packages/reconciler/src/events.ts

@@ -33,8 +33,17 @@ export type ReconcilerDriftDetectedEvent = {
  diff: Record<string, unknown>;
 };
 
+export type CoordinationWaitingEvent = {
+ level: "warn";
+ event: "reconciler.coordination.waiting";
+ deploymentId: string;
+ executionId: string;
+ holderExecutionId: string;
+};
+
 export type ReconcilerEvent =
  | OperationLifecycleEvent
+ | CoordinationWaitingEvent
  | ReconcilerDeployEvent
  | ReconcilerDriftDetectedEvent
  | import("./resource-registry").MissingResourceRegistryMatchWarningEvent;

packages/reconciler/src/yieldstar.ts

@@ -81,18 +81,7 @@ export async function* deployWithYieldStar(
  step: YieldStarStep,
  opts: YieldStarDeployOptions,
 ): AsyncGenerator<any, void, any> {
- const coordination = yield* step.store(yieldStarDeploymentCoordinationStore, {
- id: opts.deploymentId,
- initial: { holder: null },
- });
-
- yield* coordination.take(
- "notation:coordination:acquire",
- (state) => state.holder === null || state.holder === opts.executionId,
- (draft) => {
- draft.holder = opts.executionId;
- },
- );
+ const coordination = yield* acquireDeploymentCoordination(step, opts);
 
  try {
  const resourceById = new Map(
@@ -149,18 +138,7 @@ export async function* destroyWithYieldStar(
  step: YieldStarStep,
  opts: YieldStarDestroyOptions,
 ): AsyncGenerator<any, void, any> {
- const coordination = yield* step.store(yieldStarDeploymentCoordinationStore, {
- id: opts.deploymentId,
- initial: { holder: null },
- });
-
- yield* coordination.take(
- "notation:coordination:acquire",
- (state) => state.holder === null || state.holder === opts.executionId,
- (draft) => {
- draft.holder = opts.executionId;
- },
- );
+ const coordination = yield* acquireDeploymentCoordination(step, opts);
 
  try {
  const resourceById = new Map(
@@ -218,6 +196,43 @@ export async function* destroyWithYieldStar(
  }
 }
 
+/**
+ * Serializes deploy and destroy per deployment. A stale holder (a crashed
+ * execution that was never resumed) parks this execution as a durable waiter,
+ * so the wait is surfaced as a warning event before suspending.
+ */
+async function* acquireDeploymentCoordination(
+ step: YieldStarStep,
+ opts: YieldStarOperationOptions,
+): AsyncGenerator<any, WorkflowStore<CoordinationState>, any> {
+ const coordination = yield* step.store(yieldStarDeploymentCoordinationStore, {
+ id: opts.deploymentId,
+ initial: { holder: null },
+ });
+
+ const snapshot = yield* coordination.get("notation:coordination:inspect");
+ const holder = snapshot.state.holder;
+ if (holder !== null && holder !== opts.executionId) {
+ yield* emitDurably(step, "notation:coordination:waiting", opts.emit, () => ({
+ level: "warn",
+ event: "reconciler.coordination.waiting",
+ deploymentId: opts.deploymentId,
+ executionId: opts.executionId,
+ holderExecutionId: holder,
+ }));
+ }
+
+ yield* coordination.take(
+ "notation:coordination:acquire",
+ (state) => state.holder === null || state.holder === opts.executionId,
+ (draft) => {
+ draft.holder = opts.executionId;
+ },
+ );
+
+ return coordination;
+}
+
 async function* reconcileResource(
  step: YieldStarStep,
  resource: BaseResource,
@@ -646,14 +661,19 @@ function emitOperationLifecycle(
 export class YieldStarStateBackend {
  readonly #client: StoreClient;
  readonly #deploymentId: string;
+ // The deployment segment is URI-encoded so the ":" delimiter cannot appear
+ // inside it; otherwise deployment "app" would match stores of "app:blue"
+ // during prefix listing and delete them as orphans.
+ readonly #prefix: string;
 
  constructor(client: StoreClient, deploymentId: string) {
  this.#client = client;
  this.#deploymentId = deploymentId;
+ this.#prefix = `${encodeURIComponent(deploymentId)}:`;
  }
 
  storeId(resourceId: string) {
- return `${this.#deploymentId}:${resourceId}`;
+ return `${this.#prefix}${resourceId}`;
  }
 
  async get(id: string): Promise<StateNode | undefined> {
@@ -740,11 +760,10 @@ export class YieldStarStateBackend {
  }
 
  async values(): Promise<StateNode[]> {
- const prefix = `${this.#deploymentId}:`;
  const ids = await this.#client.listStores(yieldStarResourceStateStore);
  const snapshots = await Promise.all(
  ids
- .filter((id) => id.startsWith(prefix))
+ .filter((id) => id.startsWith(this.#prefix))
  .map((id) => this.#tryGetSnapshot(id)),
  );
  return snapshots
@@ -760,11 +779,10 @@ export class YieldStarStateBackend {
  }
 
  async clear(): Promise<void> {
- const prefix = `${this.#deploymentId}:`;
  const ids = await this.#client.listStores(yieldStarResourceStateStore);
  await Promise.all(
  ids
- .filter((id) => id.startsWith(prefix))
+ .filter((id) => id.startsWith(this.#prefix))
  .map((id) =>
  this.#client.deleteStore({
  definition: yieldStarResourceStateStore,

packages/reconciler/test/yieldstar.integration.test.ts

@@ -253,6 +253,73 @@ describe("YieldStar reconciliation", () => {
  runtime.close();
  });
 
+ it("emits a coordination waiting event when another execution holds the deployment", async () => {
+ let unblockCreate!: () => void;
+ const blocked = new Promise<void>((resolve) => {
+ unblockCreate = resolve;
+ });
+ let started!: () => void;
+ const createStarted = new Promise<void>((resolve) => {
+ started = resolve;
+ });
+ const TestResource = resource({ type: "test/yieldstar/coordination" })
+ .defineSchema({})
+ .defineOperations({
+ create: async () => {
+ started();
+ await blocked;
+ },
+ delete: async () => undefined,
+ });
+ const events: ReconcilerEvent[] = [];
+ const runtime = createRuntime(
+ [new TestResource({ id: "held" })],
+ "coordination-waiting",
+ { emit: (event) => void events.push(event) },
+ );
+
+ const first = runtime.run("holder-execution");
+ await createStarted;
+ await runtime.run("waiter-execution");
+
+ expect(
+ events.find(
+ (event) => event.event === "reconciler.coordination.waiting",
+ ),
+ ).toMatchObject({
+ level: "warn",
+ deploymentId: "coordination-waiting",
+ executionId: "waiter-execution",
+ holderExecutionId: "holder-execution",
+ });
+
+ unblockCreate();
+ await first;
+ runtime.close();
+ });
+
+ it("scopes store listing to the exact deployment despite prefix-like IDs", async () => {
+ const database = createSqliteDb({ path: ":memory:" });
+ const storeClient = new SqliteStoreClient({
+ db: database,
+ schedulerClient: new TestScheduler(),
+ });
+ const app = new YieldStarStateBackend(storeClient, "app");
+ const appBlue = new YieldStarStateBackend(storeClient, "app:blue");
+
+ await app.update("site", 0, statePatch("site"));
+ await appBlue.update("site", 0, statePatch("site"));
+
+ expect(await app.values()).toHaveLength(1);
+ expect(await appBlue.values()).toHaveLength(1);
+
+ await app.clear();
+ expect(await app.values()).toHaveLength(0);
+ expect(await appBlue.values()).toHaveLength(1);
+ expect(await appBlue.get("site")).toBeDefined();
+ database.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" })