fix(durable): reject waits that close a wait cycle

A wait is rejected when it closes a cycle in the wait-for graph: waits of
tasks that are not abort-marked, and holds of live ordinary owned work. The
rejection names the cycle. Replaces the owner-chain rule, so a background task
may wait on its conversation's owner. Approach and tests from #10533 by
@odysseus0.

fixes #10411

packages/durable/CHANGELOG.md

@@ -28,6 +28,7 @@
 
 ### Fixed
 
+- A wait that closes a cycle through waits and held owners is rejected by `runtime.commit()`, naming the cycle, instead of leaving every task on it waiting forever; waiting on itself or on an owner chain were the only cycles caught. A background task may now wait on the owner of its conversation, which is no cycle ([#10411](https://github.com/earendil-works/pi/issues/10411); tests and approach from [#10533](https://github.com/earendil-works/pi/pull/10533) by [@odysseus0](https://github.com/odysseus0))
 - Appending to an array in a document no longer costs a walk over the whole array per commit: Chord's delta tracker plans inserts and removals that keep the remaining elements in order from the changed ranges alone, and builds an appended array in one copy instead of two. 32,000 one-item appends, one commit each, take 1.4 s instead of 11.3 s; what remains is the copy each immutable value needs (`npm run bench:document-append`).
 - A generation whose blocking threshold compaction failed, faulted, was orphaned, or was aborted directly now fails with `model_error` (`Compaction failed: <error>`) instead of sending its request anyway. The context was still above the blocking threshold, so the request at best left no room for the answer: a session at 996K of a 1M window got an empty `length` answer on every turn, with no visible error.
 

packages/durable/docs/spec.md

@@ -1827,8 +1827,8 @@ commit, so the checkpoint or outcome is atomic with the callback's entries,
 documents, and child tasks and is type-checked against the task's checkpoint and
 result types. Returning nothing leaves the state unchanged. A terminal or
 completing state drops the memos. A `waiting` state ends the invocation even
-when its checkpoint is unchanged; it rejects when `on` names a missing task, the
-task itself, or a task on its owner chain, which could never finish first, and
+when its checkpoint is unchanged; it rejects when `on` names a missing task or
+closes a wait cycle (section 5.5), which no task on it could finish first, and
 an empty `on` resumes at the next scheduling pass. `pending` is never returned;
 only reconciliation and handover write it.
 
@@ -2224,8 +2224,17 @@ finish.
 The task stops without an invocation and resumes at `checkpoint` once every task
 in `on` is terminal. `on` may name any tasks, including already terminal ones;
 tasks the waiting task does not own, whose `owner` is not the waiting task,
-require `policy: "allSettled"`. `on` may not name a missing task, the task
-itself, or a task on its owner chain (section 5.1). With `failFast`, the first
+require `policy: "allSettled"`. `on` may not name a missing task or close a
+cycle in the wait-for graph (section 5.1), whose edges run from each task to the
+tasks it waits on and to its live ordinary owned work, which holds it as
+`completing`. A task waiting on itself, or on an owner that holds it, closes the
+shortest such cycles; a background task may wait on the owner of its
+conversation, which it does not hold. The check follows both kinds of edge from
+the tasks in `on`, visiting each task once, and the rejection names the cycle.
+Only durable waits are edges: a wait held only in memory, such as
+`runtime.waitForTask()` or a nested call's, is not checked. An abort-marked
+task waits only for its owned work, so its `on` is no edge.
+With `failFast`, the first
 task in `on` that holds or ends with a non-`completed` outcome gives every other
 live task in `on` an abort mark, in the scheduler's next reconcile commit. The waiting task itself is not marked:
 it resumes once all of `on` is terminal and reads their outcomes with

packages/durable/src/harness/scheduler.ts

@@ -1470,8 +1470,12 @@ export class TaskScheduler {
 	}
 
 	/**
-	 * A wait names existing tasks other than the waiter and its owners, which could never finish first; `failFast` only
-	 * tasks the waiter owns. An abort handler cannot wait.
+	 * A wait names existing tasks, and `failFast` only tasks the waiter owns. It may not close a cycle in the wait-for
+	 * graph, whose edges run from each task to what it waits on and to its live ordinary owned work (spec §5.5); waiting
+	 * on itself or on an owner holding it are the shortest such cycles. An abort handler cannot wait.
+	 *
+	 * Invariant: the wait-for graph is acyclic. Only a wait adds an edge between existing tasks, as a new task has only
+	 * incoming edges, so checking each wait maintains it.
 	 */
 	async #validateWait(
 		tx: Transaction,
@@ -1482,19 +1486,71 @@ export class TaskScheduler {
 	): Promise<void> {
 		if (invocation.mode === "abort") throw new Error(`Abort handler of task ${current.id} cannot wait`);
 		const overlay = overlayOf(tx);
-		await this.#loadChain(parentOf(current));
-		const owners = new Set<TaskId>();
-		for (const step of this.#above(parentOf(current))) if ("task" in step) owners.add(step.task);
+		const targets: TaskId[] = [];
 		for (const id of on) {
-			if (id === current.id || owners.has(id)) {
-				throw new Error(`Task ${current.id} cannot wait on itself or its owner ${id}`);
-			}
 			const member = overlay.tasks.get(id) ?? this.#live.get(id) ?? (await this.#storage.task(id, this.#context));
 			if (member === undefined) throw new Error(`Task ${id} does not exist`);
 			if (policy === "failFast" && member.owner !== current.id) {
 				throw new Error(`Task ${current.id} can wait failFast only on tasks it owns; ${id} is not one`);
 			}
+			// A live child already holds the waiter (task-owned tasks are never background): waiting on it adds no edge. A
+			// terminal task has no edges.
+			if (member.state.status !== "terminal" && member.owner !== current.id) targets.push(id);
 		}
+		if (targets.length === 0) return;
+		const cycle = await this.#waitCycle(current.id, targets, overlay);
+		if (cycle !== undefined) {
+			throw new Error(
+				`Task ${current.id} cannot wait on ${cycle[0]}: ${[current.id, ...cycle].join(" -> ")} is a wait cycle`,
+			);
+		}
+	}
+
+	/**
+	 * The path from a member of `on` back to `task` in the wait-for graph, if the wait would close a cycle (spec §5.1).
+	 * An edge X -> Y means X cannot settle before Y: Y is in the `on` of waiting X that is not abort-marked, or Y is live
+	 * ordinary owned work of X,
+	 * which holds X as `completing`. Each task is visited once and each subtree walked once, so a check is bounded by the
+	 * live work it reaches.
+	 */
+	async #waitCycle(task: TaskId, on: readonly TaskId[], overlay: Overlay): Promise<TaskId[] | undefined> {
+		// Owned work is found by walks down, which need the chains of every live task loaded.
+		await this.#loadScopes(false);
+		const from = new Map<TaskId, TaskId | undefined>();
+		const walked = new Set<TaskId>();
+		const pending: TaskId[] = [];
+		const reach = (id: TaskId, via: TaskId | undefined) => {
+			if (from.has(id)) return;
+			from.set(id, via);
+			pending.push(id);
+		};
+		const path = (end: TaskId): TaskId[] => {
+			const steps: TaskId[] = [];
+			for (let at: TaskId | undefined = end; at !== undefined; at = from.get(at)) steps.unshift(at);
+			return steps;
+		};
+		for (const id of on) reach(id, undefined);
+		for (let id = pending.pop(); id !== undefined; id = pending.pop()) {
+			if (id === task) return path(id);
+			const record = overlay.tasks.get(id) ?? this.#live.get(id);
+			if (record === undefined || record.state.status === "terminal") continue;
+			// An abort-marked task waits only for its owned work, as abort runs bottom-up (`#waitingOn`).
+			if (record.state.status === "waiting" && !record.abortRequested) {
+				for (const member of record.state.on) reach(member, id);
+			}
+			if (walked.has(id)) continue;
+			walked.add(id);
+			// Below a walked task the walk does not go again: its own walk covered what lies below.
+			const found: TaskId[] = [];
+			const walk = this.#walkDown(id, { overlay, enter: (below) => !walked.has(below) });
+			for (const { id: below, node, live } of walk) {
+				if (live === undefined || node.background) continue;
+				found.push(below);
+				reach(below, id);
+			}
+			for (const below of found) walked.add(below);
+		}
+		return undefined;
 	}
 
 	/** End an invocation: its runtime operations reject from now on, its signal aborts, its watches stop, and its task is free. */

packages/durable/test/harness-scheduler-retention.test.ts

@@ -540,7 +540,9 @@ describe("scheduler retention", () => {
 		await harness.close(context);
 	});
 
-	it("still rejects waiting on a forgotten ended owner", async () => {
+	// A wait on an ended task closes no cycle (#10411), also when the ended task owns the waiter's conversation and the
+	// scheduler forgot it: the wait resumes at once.
+	it("lets a task wait on a forgotten ended owner", async () => {
 		const opened = await openTasks(await openNodeSqliteStorage(await sqlitePath()), [Hold, Waiter]);
 		const root = await opened.harness.root(context);
 		opened.harness.resume();
@@ -557,8 +559,7 @@ describe("scheduler retention", () => {
 			context,
 		);
 		const outcome = (await opened.harness.waitForTask(waiter, context)).state.outcome;
-		expect(outcome).toMatchObject({ status: "faulted" });
-		expect(JSON.stringify(outcome)).toMatch(/cannot wait on itself or its owner/);
+		expect(outcome).toMatchObject({ status: "completed" });
 		await opened.harness.close(context);
 	});
 });

packages/durable/test/harness-structured.test.ts

@@ -462,7 +462,7 @@ describe("waiting", () => {
 		const { harness, root } = await openNodes();
 		const other = await start(root, "other");
 		const cases: [string, (runtime: NodeRuntime) => readonly TaskId[], JoinPolicy, string][] = [
-			["self", (runtime) => [runtime.taskId], "allSettled", "cannot wait on itself or its owner"],
+			["self", (runtime) => [runtime.taskId], "allSettled", "is a wait cycle"],
 			["missing", () => [999_999 as TaskId], "allSettled", "does not exist"],
 			["foreign", () => [other], "failFast", "only on tasks it owns"],
 		];
@@ -489,12 +489,210 @@ describe("waiting", () => {
 		expect((await harness.waitForTask(parent, context)).state.outcome.status).toBe("completed");
 		expect((await harness.waitForTask(child!, context)).state.outcome).toMatchObject({
 			status: "faulted",
-			error: { message: expect.stringContaining("cannot wait on itself or its owner") },
+			error: { message: expect.stringContaining("is a wait cycle") },
 		});
 		open("other");
 		await harness.close(context);
 	});
 
+	// #10411; this test and the next two are from #10533 by @odysseus0.
+	it("rejects the wait that closes a cycle", async () => {
+		const { harness, root } = await openNodes();
+		let a: TaskId | undefined;
+		let b: TaskId | undefined;
+		script("a", { run: (runtime, ctx) => runtime.commit(() => waitOn([b!], "allSettled"), ctx) });
+		script("b", { run: (runtime, ctx) => runtime.commit(() => waitOn([a!], "allSettled"), ctx) });
+		await root.commit(async (tx) => {
+			a = await tx.createTask(Node, { name: "a" }, OWN_CONVERSATION);
+			b = await tx.createTask(Node, { name: "b" }, OWN_CONVERSATION);
+		}, context);
+		// a -> b and b -> a are waits, and whichever comes second closes the cycle.
+		const outcomes = await Promise.all([outcomeOf(harness, a!), outcomeOf(harness, b!)]);
+		expect(outcomes.sort()).toEqual(["completed", "faulted"]);
+		await harness.close(context);
+	});
+
+	// #10411
+	it("rejects the wait that closes a cycle through a held owner", async () => {
+		const { harness, root } = await openNodes();
+		let held: TaskId | undefined;
+		let other: TaskId | undefined;
+		const child = spawnAndFinish("held", "child");
+		script("child", { run: (runtime, ctx) => runtime.commit(() => waitOn([other!], "allSettled"), ctx) });
+		script("other", { run: (runtime, ctx) => runtime.commit(() => waitOn([held!], "allSettled"), ctx) });
+		await root.commit(async (tx) => {
+			held = await tx.createTask(Node, { name: "held" }, OWN_CONVERSATION);
+			other = await tx.createTask(Node, { name: "other" }, OWN_CONVERSATION);
+		}, context);
+		// held -> child is a hold; child -> other -> held are waits, and whichever wait comes second closes the cycle.
+		expect(await outcomeOf(harness, held!)).toBe("completed");
+		const outcomes = await Promise.all([outcomeOf(harness, child.id!), outcomeOf(harness, other!)]);
+		expect(outcomes.sort()).toEqual(["completed", "faulted"]);
+		await harness.close(context);
+	});
+
+	// #10411
+	it("rejects a cycle through holds two levels deep", async () => {
+		const { harness, root } = await openNodes();
+		let held: TaskId | undefined;
+		let other: TaskId | undefined;
+		const child = spawnAndFinish("held", "child");
+		const grandchild = spawnAndFinish("child", "grandchild");
+		script("grandchild", { run: (runtime, ctx) => runtime.commit(() => waitOn([other!], "allSettled"), ctx) });
+		script("other", { run: (runtime, ctx) => runtime.commit(() => waitOn([held!], "allSettled"), ctx) });
+		await root.commit(async (tx) => {
+			held = await tx.createTask(Node, { name: "held" }, OWN_CONVERSATION);
+			other = await tx.createTask(Node, { name: "other" }, OWN_CONVERSATION);
+		}, context);
+		// held -> child -> grandchild are holds; grandchild -> other -> held are waits.
+		await until(() => grandchild.id !== undefined);
+		const outcomes = await Promise.all([outcomeOf(harness, grandchild.id!), outcomeOf(harness, other!)]);
+		expect(outcomes.sort()).toEqual(["completed", "faulted"]);
+		expect(await outcomeOf(harness, child.id!)).toBe("completed");
+		await harness.close(context);
+	});
+
+	// #10411
+	it("lets a background task wait on the owner of its conversation", async () => {
+		const { harness, root } = await openNodes();
+		let owner: TaskId | undefined;
+		script("background", { run: (runtime, ctx) => runtime.commit(() => waitOn([owner!], "allSettled"), ctx) });
+		const background = await root.commit(async (tx) => {
+			owner = await tx.createTask(Node, { name: "owner" }, OWN_CONVERSATION);
+			const conversation = await tx.createConversation({ ownership: { kind: "task", taskId: owner } });
+			const options = { ...OWN_CONVERSATION, conversationId: conversation.id, background: true };
+			return tx.createTask(Node, { name: "background" }, options);
+		}, context);
+		// background -> owner is a wait but owner -> background is no hold, so there is no cycle.
+		await until(async () => (await state(harness, background)).status === "waiting");
+		open("owner");
+		expect(await outcomeOf(harness, background)).toBe("completed");
+		await harness.close(context);
+	});
+
+	// #10411: the issue's orchestrator, worker, and reviewer.
+	it("rejects a cycle of a hold and two waits that each look reasonable", async () => {
+		const { harness, root } = await openNodes();
+		const ids: Record<string, TaskId> = {};
+		const reviewerWaits = deferred();
+		// The orchestrator spawns a worker and finishes; the running worker holds it as completing.
+		script("orchestrator", {
+			run: (runtime, ctx) =>
+				runtime.commit(async (tx) => {
+					ids.worker = await spawn(tx, runtime.taskId, "worker");
+					return { status: "running", checkpoint: { phase: "resume", round: 1 } };
+				}, ctx),
+		});
+		// The reviewer waits for the orchestrator's result.
+		script("reviewer", {
+			run: async (runtime, ctx) => {
+				await runtime.commit(() => waitOn([ids.orchestrator!], "allSettled"), ctx);
+				reviewerWaits.resolve();
+			},
+		});
+		// The worker waits for the reviewer, closing the cycle.
+		script("worker", {
+			run: async (runtime, ctx) => {
+				await reviewerWaits.promise;
+				await runtime.commit(() => waitOn([ids.reviewer!], "allSettled"), ctx);
+			},
+		});
+		ids.orchestrator = await start(root, "orchestrator");
+		await until(async () => (await state(harness, ids.orchestrator!)).status === "completing");
+		ids.reviewer = await start(root, "reviewer");
+		const outcomes = await Promise.all(
+			["orchestrator", "worker", "reviewer"].map((name) => outcomeOf(harness, ids[name]!)),
+		);
+		expect(outcomes).toEqual(["completed", "faulted", "completed"]);
+		const worker = (await harness.waitForTask(ids.worker!, context)).state.outcome;
+		expect(JSON.stringify(worker)).toContain(
+			`${ids.worker} -> ${ids.reviewer} -> ${ids.orchestrator} -> ${ids.worker} is a wait cycle`,
+		);
+		await harness.close(context);
+	});
+
+	// #10411
+	it("allows a diamond of waits", async () => {
+		const { harness, root } = await openNodes();
+		const ids: Record<string, TaskId> = {};
+		// A diamond: top waits on left and right, both wait on bottom. No cycle.
+		script("top", {
+			run: (runtime, ctx) => runtime.commit(() => waitOn([ids.left!, ids.right!], "allSettled"), ctx),
+		});
+		script("left", { run: (runtime, ctx) => runtime.commit(() => waitOn([ids.bottom!], "allSettled"), ctx) });
+		script("right", { run: (runtime, ctx) => runtime.commit(() => waitOn([ids.bottom!], "allSettled"), ctx) });
+		await root.commit(async (tx) => {
+			ids.bottom = await tx.createTask(Node, { name: "bottom" }, OWN_CONVERSATION);
+			ids.left = await tx.createTask(Node, { name: "left" }, OWN_CONVERSATION);
+			ids.right = await tx.createTask(Node, { name: "right" }, OWN_CONVERSATION);
+			ids.top = await tx.createTask(Node, { name: "top" }, OWN_CONVERSATION);
+		}, context);
+		open("bottom");
+		expect(await outcomeOf(harness, ids.top)).toBe("completed");
+		await harness.close(context);
+	});
+
+	// #10411
+	it("follows no wait of an abort-marked task, which waits only for its owned work", async () => {
+		const { harness, root } = await openNodes();
+		const ids: Record<string, TaskId> = {};
+		const cleanup = deferred();
+		// b owns c and waits on a; aborting b leaves c's cleanup pending, so b waits only for c.
+		script("b", {
+			run: (runtime, ctx) =>
+				runtime.commit(async (tx) => {
+					ids.c = await spawn(tx, runtime.taskId, "c");
+					return waitOn([ids.a!], "allSettled");
+				}, ctx),
+		});
+		script("c", {
+			abort: async (runtime, ctx) => {
+				await cleanup.promise;
+				await runtime.commit(() => ({ status: "terminal", outcome: { status: "aborted", result: "c" } }), ctx);
+			},
+		});
+		const aWaits = deferred();
+		script("a", {
+			run: async (runtime, ctx) => {
+				await aWaits.promise;
+				await runtime.commit(() => waitOn([ids.b!], "allSettled"), ctx);
+			},
+			resume: (runtime, ctx) => runtime.commit(() => end("completed", "a"), ctx),
+		});
+		await root.commit(async (tx) => {
+			ids.a = await tx.createTask(Node, { name: "a" }, OWN_CONVERSATION);
+			ids.b = await tx.createTask(Node, { name: "b" }, OWN_CONVERSATION);
+		}, context);
+		await until(async () => (await state(harness, ids.b!)).status === "waiting");
+		expect(await harness.abortTask(ids.b!, context)).toBe("marked");
+		await until(() => log.includes("abort:c"));
+		// a -> b would close a -> b -> a through b's stale wait; b now waits only for c.
+		aWaits.resolve();
+		await until(async () => (await state(harness, ids.a!)).status === "waiting");
+		cleanup.resolve();
+		expect(await outcomeOf(harness, ids.b!)).toBe("aborted");
+		expect(await outcomeOf(harness, ids.a!)).toBe("completed");
+		await harness.close(context);
+	});
+
+	// #10411
+	it("rejects a longer cycle of waits", async () => {
+		const { harness, root } = await openNodes();
+		const ids: Record<string, TaskId> = {};
+		const names = ["x", "y", "z"];
+		for (const [index, name] of names.entries()) {
+			const next = names[(index + 1) % names.length]!;
+			script(name, { run: (runtime, ctx) => runtime.commit(() => waitOn([ids[next]!], "allSettled"), ctx) });
+		}
+		await root.commit(async (tx) => {
+			for (const name of names) ids[name] = await tx.createTask(Node, { name }, OWN_CONVERSATION);
+		}, context);
+		const outcomes = await Promise.all(names.map((name) => outcomeOf(harness, ids[name]!)));
+		expect(outcomes.filter((outcome) => outcome === "faulted")).toHaveLength(1);
+		expect(outcomes.filter((outcome) => outcome === "completed")).toHaveLength(2);
+		await harness.close(context);
+	});
+
 	it("resumes at the next pass when it waits on nothing", async () => {
 		const { harness, root } = await openNodes();
 		script("parent", { run: (runtime, ctx) => runtime.commit(() => waitOn([], "failFast"), ctx) });