diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index ed150b865..49068ddf9 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -24,6 +24,7 @@ - A throwing `onReport` no longer replaces the error being reported or becomes an unhandled rejection, a throwing `now()` is reported once and `Date.now` used, and a watch listener's error is reported as well as ending the watch with `listener_error`. - A task whose fault write failed no longer runs again in the same process. +- The scheduler no longer visits every live task on each pass: abort cascades start only from tasks with cancellation intent, owned live work and idle scopes are walks down an in-memory ownership index, and reservation, finalization, and failFast checks keep their own candidate sets. A chain of 500 owned tasks that took 12 s to settle takes 50 ms, 20,000 take 1.5 s; a tool making 8000 parallel nested calls finishes in half the time (`npm run bench:scheduler-fanout`). ## [1.1.0] - 2026-10-07 diff --git a/packages/durable/package.json b/packages/durable/package.json index 14903974d..ff91a4b11 100644 --- a/packages/durable/package.json +++ b/packages/durable/package.json @@ -107,6 +107,7 @@ "bench:tool-output": "node --conditions=source --experimental-strip-types --expose-gc test/tool-output-bench.ts", "bench:nested-tools": "node --conditions=source --experimental-strip-types --expose-gc test/nested-tools-bench.ts", "bench:nested-images": "node --conditions=source --experimental-strip-types --expose-gc test/nested-images-bench.ts", + "bench:scheduler-fanout": "node --conditions=source --experimental-strip-types test/scheduler-fanout-bench.ts", "prepublishOnly": "npm run clean && npm run build" }, "keywords": [ diff --git a/packages/durable/src/harness/scheduler.ts b/packages/durable/src/harness/scheduler.ts index bd0bfb481..13ce79b75 100644 --- a/packages/durable/src/harness/scheduler.ts +++ b/packages/durable/src/harness/scheduler.ts @@ -125,8 +125,26 @@ type Step = type Overlay = { readonly tasks: ReadonlyMap; readonly edges: ReadonlyMap; + /** Links to the staged tasks `#below` does not list, by parent node, for walks down; built at first use. */ + links?: ReadonlyMap>; }; +/** A node of the downward index: a task, or a conversation as `c`. */ +type Node = TaskId | `c${number}`; + +function conversationNode(id: ConversationId): Node { + return `c${id}`; +} + +function conversationOf(node: `c${number}`): ConversationId { + return Number(node.slice(1)) as ConversationId; +} + +/** The node a task hangs from: its owner task, or its conversation. */ +function parentNode(node: TaskNode): Node { + return node.owner ?? conversationNode(node.conversationId); +} + /** Where ordinary ownership traversal starts: one conversation, or every ownerless conversation. */ type Scope = { readonly conversation: ConversationId } | { readonly roots: true }; @@ -164,8 +182,10 @@ export type TaskSchedulerOptions = { * commit listener updates it on the Session line, so code running on the line reads exactly the committed state from it. * * Tasks and conversations form one ownership tree (spec §5.5): a task's parent is its owner task, or its conversation; - * a conversation's parent is its owner task, if any. Walks up that tree decide cascades, idle scopes, and whether a - * task's ordinary owned work is live, which holds its outcome as `completing` and delays its abort handler. + * a conversation's parent is its owner task, if any. Walks up that tree decide what reaches one task: a cancelling + * owner, a scope. `#below` lists the tree downward around live work, so what lies below one node is a walk down from + * it: a task's live ordinary owned work, which holds its outcome as `completing` and delays its abort handler, a + * scope's tasks, a cascade's reach. No pass visits every live task; the sets below hold each pass's candidates. * * Invariant: every task transition is decided and written by one callback serialized on the Session line. That covers * reservation, marks, runtime commits, finalization, and the synchronous step before each phase, which applies the @@ -203,10 +223,39 @@ export class TaskScheduler { readonly #failedMigrations = new Map(); /** Owner task of each loaded conversation, `null` when ownerless. */ readonly #edges = new Map(); - /** Tasks that own a loaded conversation; their nodes stay in `#settled` once terminal. */ + /** + * Tasks that own a loaded conversation; their nodes stay in `#settled` once terminal, and with them, through + * `#knownChildren`, every task above them. Never cleared: about one entry per task on a subagent's chain. + */ readonly #conversationOwners = new Set(); - /** Ownership fields of terminal tasks that walks pass through. */ + /** + * Ownership fields of terminal tasks that walks pass through: conversation owners, chain loads, owners of known + * tasks, and those with nodes listed below them. `#forget` drops the rest. + */ readonly #settled = new Map(); + /** + * The tree downward around live work: each node's child nodes that are live tasks or have nodes listed below them. + * A conversation's parent is its owner task; ownerless ones with work are in `#roots`. + */ + readonly #below = new Map>(); + readonly #roots = new Set(); + /** Known tasks whose chain may not be loaded; every other known task's chain is, and stays, loaded. */ + readonly #unloaded = new Set(); + /** Live tasks with cancellation intent, the sources of abort cascades (spec §5.4). */ + readonly #intent = new Set(); + /** Live tasks that are not `completing`, the candidates of a reservation pass; ones found waiting leave it. */ + readonly #runnable = new Set(); + /** `completing` tasks the next finalize pass checks: those that started holding, and holders of ended work. */ + readonly #finalizeChecks = new Set(); + /** Live waiting tasks by each task in their `on`. */ + readonly #waiters = new Map>(); + /** Terminal tasks a chain load read; kept in `#settled`, so loaded chains stay loaded. */ + readonly #pinned = new Set(); + /** + * Per task, how many known tasks (live, or in `#settled`) it owns directly. An ended task stays known while any does, + * so the chain above every known task stays loaded. + */ + readonly #knownChildren = new Map(); /** * `failFast` waiters the next reconcile checks for a failed task in `on`: at open, when they start waiting, and when * one of their tasks fails. @@ -254,7 +303,7 @@ export class TaskScheduler { } for (const records of scans) { for (const record of records) { - this.#live.set(record.id, record); + this.#track(undefined, record); if (record.abandonOnRestart === true) this.#abandoned.add(record.id); if (record.state.status === "running") { tx.setTask(withState(record, { status: "pending", checkpoint: record.state.checkpoint })); @@ -302,7 +351,7 @@ export class TaskScheduler { if (keepRestart && current.abortReason === "restart") return { result: "marked" as const }; if (invocation === undefined && current.state.status !== "completing") { await this.#loadScopes(false); - if (!this.#ownedLive().has(id)) { + if (!this.#hasOwnedLive(id)) { const resolution = this.#resolve(current as RunnableTaskRecord, this.#registry.snapshot()); if (resolution.kind === "blocked") { await this.#terminate(tx, current, { status: "orphaned", reason: resolution.reason }); @@ -354,11 +403,11 @@ export class TaskScheduler { const queued = await this.#loadScopes(true); const scope = { conversation: conversationId }; const reached: TaskId[] = []; - for (const record of this.#live.values()) { - if (record.background && !background) continue; - if (this.#inScope(parentOf(record), scope, background) !== true) continue; - reached.push(record.id); - if (!record.abortRequested || record.abortReason !== undefined) tx.setTask(withAbortMark(record)); + const walk = this.#walkDown(conversationNode(conversationId), { crossBackground: background }); + for (const { node, live } of walk) { + if (live === undefined || (node.background && !background)) continue; + reached.push(live.id); + if (!live.abortRequested || live.abortReason !== undefined) tx.setTask(withAbortMark(live)); } for (const id of queued) { if (this.#inScope({ conversation: id }, scope, background) === true) await this.#withdrawInputs(tx, id); @@ -375,17 +424,22 @@ export class TaskScheduler { const updated: AnyTaskRecord[] = []; const failed: TaskId[] = []; let changed = false; + // Edges first, so the tasks a commit creates in a new conversation hang from it. + for (const change of publication.changes) { + if (change.type === "conversation" && !this.#edges.has(change.value.id)) { + this.#setEdge(change.value.id, change.value.owner?.taskId ?? null); + } + } for (const change of publication.changes) { if (change.type !== "task") continue; changed = true; const record = change.value; const previous = this.#live.get(record.id); if (failedOutcome(record) && (previous === undefined || !failedOutcome(previous))) failed.push(record.id); + this.#track(previous, record); if (record.state.status === "terminal") { - this.#live.delete(record.id); this.#failedMigrations.delete(record.id); this.#failFastChecks.delete(record.id); - if (this.#conversationOwners.has(record.id)) this.#settled.set(record.id, nodeOf(record)); this.#taskWaiters.resolve(record.id, record as SettledTask); // Its owner may finalize now. this.#scheduleReconcile(); @@ -408,33 +462,30 @@ export class TaskScheduler { this.#failFastChecks.add(record.id); this.#scheduleReconcile(); } - this.#live.set(record.id, record); - updated.push(record); + // Its owners and place never change: what is above it is checked once, when it becomes live. + if (previous === undefined) updated.push(record); } for (const id of failed) { - for (const record of this.#live.values()) { - const state = record.state; - if (state.status === "waiting" && state.policy === "failFast" && state.on.includes(id)) { - this.#failFastChecks.add(record.id); + for (const waiter of this.#waiters.get(id) ?? []) { + const state = this.#live.get(waiter)?.state; + if (state?.status === "waiting" && state.policy === "failFast") { + this.#failFastChecks.add(waiter); this.#scheduleReconcile(); } } } - for (const change of publication.changes) { - if (change.type === "conversation" && !this.#edges.has(change.value.id)) { - this.#setEdge(change.value.id, change.value.owner?.taskId ?? null); - } - } for (const change of publication.changes) { // A queued input below a cancelled owner is withdrawn, even after its cascade. if (change.type !== "submission" || change.value.status !== "queued" || change.value.type !== "input") continue; - const up = { conversation: change.value.conversationId }; - if (!this.#chainKnown(up) || this.#belowCancelled(up)) this.#cascadePending = true; + const conversation = change.value.conversationId; + if (!this.#known(conversationNode(conversation)) || this.#belowCancelled({ conversation })) { + this.#cascadePending = true; + } } for (const record of updated) { // Work created below a cancelled owner, even after its cascade, is aborted too. - if (!this.#chainKnown(parentOf(record))) this.#scheduleReconcile(); + if (this.#unloaded.has(record.id)) this.#scheduleReconcile(); else if (!record.background && !record.abortRequested && this.#belowCancelled(parentOf(record))) { this.#cascadePending = true; } @@ -546,14 +597,16 @@ export class TaskScheduler { if (record.abortRequested && (record.abortReason === undefined || reason !== undefined)) return; marks.set(record.id, { record, reason }); }; - // Loading edges can reveal a cancelled owner, so marks are derived on every pass. - for (const record of this.#live.values()) { - if (record.background) continue; - const owner = this.#cancellingOwner(parentOf(record)); - if (owner === undefined) continue; + // Every live foreground task below a live owner with cancellation intent, which is its nearest one: a walk down + // from each owner with intent does not enter another, whose own walk covers what is below it. Derived again on + // every pass, as loading edges can reveal a cancelled owner. + for (const id of this.#intent) { + const owner = this.#live.get(id)!; // Only an owner abandoned after a restart, and nothing else, passes its reason on. const restart = owner.abortReason === "restart" && !failedOutcome(owner); - mark(record, restart ? "restart" : undefined); + for (const { node, live } of this.#walkDown(id, { enter: (below) => !this.#intent.has(below) })) { + if (live !== undefined && !node.background) mark(live, restart ? "restart" : undefined); + } } for (const id of checks) { const waiter = this.#live.get(id); @@ -591,31 +644,45 @@ export class TaskScheduler { * Harness cleanup here. */ async #finalize(tx: Transaction): Promise { - for (;;) { - const overlay = overlayOf(tx); - const owned = this.#ownedLive(overlay); - const done = [...this.#liveRecords(overlay)].filter( - (record) => record.state.status === "completing" && !owned.has(record.id), - ); - if (done.length === 0) return; - for (const record of done) { - const outcome = (record.state as { readonly outcome: TaskOutcome }).outcome; - tx.setTask(withState(record, { status: "terminal", outcome })); + // A worklist: the tasks that started holding, and the holders of work that ended. Finalizing one can free the + // nearest live task above it, which goes on the list. Tasks finalized here have no live work below them, so walks + // down skip them. + // In rounds: the holders the tasks of one round free are checked once, in the next, so a holder of many tasks that + // end together is walked once, not once per task. + let round = [...this.#finalizeChecks]; + this.#finalizeChecks.clear(); + const done = new Set(); + const overlay = overlayOf(tx); + while (round.length > 0) { + const next = new Set(); + for (const id of round) { + const record = overlay.tasks.get(id) ?? this.#live.get(id); + if (record?.state.status !== "completing" || done.has(id)) continue; + if (this.#hasOwnedLive(id, overlay, done)) continue; + const outcome = record.state.outcome; + const terminal = withState(record, { status: "terminal", outcome }); + tx.setTask(terminal); + (overlay.tasks as Map).set(id, terminal); + done.add(id); // REMINDER: only the scheduler writes `faulted` and `orphaned` (spec §5.4); their cleanup waits for this commit. if (outcome.status === "faulted" || outcome.status === "orphaned") { await this.#settleOutcome(tx, record, outcome); } + const above = this.#nearestLiveAbove(record, overlay); + if (above?.state.status === "completing") next.add(above.id); } + round = [...next]; } } /** - * Load the owner chains of every live task and, with `queued`, of every conversation with queued submissions, on the - * Session line; returns the latter. Reads committed Storage directly, so it may run inside a commit callback. + * Load the owner chains of the tasks in `#unloaded` and, with `queued`, of every conversation with queued submissions, + * on the Session line; returns the latter. Reads committed Storage directly, so it may run inside a commit callback. */ async #loadScopes(queued: boolean): Promise { - for (const record of [...this.#live.values()]) { - if (!this.#chainKnown(parentOf(record))) await this.#loadChain(parentOf(record)); + for (const id of [...this.#unloaded]) { + // An earlier load in this loop may have passed it. + if (this.#unloaded.has(id)) await this.#loadChain({ task: id }); } if (!queued) return []; const submissions = await scanAll((cursor) => @@ -626,18 +693,30 @@ export class TaskScheduler { return conversations; } - /** Load the owner edges and task nodes from `start` up to its ownerless root. */ + /** + * Load the owner edges and task nodes from `start` up to its ownerless root. A known task outside `#unloaded` ends the + * walk, as its chain is loaded; the `#unloaded` tasks passed have a loaded chain afterwards. Loading a node can reveal + * a cancelled owner, so it makes the next reconcile derive cascade marks. + */ async #loadChain(start: Up, overlay?: Overlay): Promise { + const passed: TaskId[] = []; let at: Up | undefined = start; while (at !== undefined) { if ("task" in at) { - let node = this.#node(at.task, overlay); + const id: TaskId = at.task; + let node: TaskNode | undefined = this.#live.get(id) ?? this.#settled.get(id); + // Created in this commit: the load from its own record covers what is above it. + if (node === undefined && overlay?.tasks.has(id) === true) break; + if (node !== undefined && !this.#unloaded.has(id)) break; if (node === undefined) { - const record = await this.#storage.task(at.task, this.#context); + const record = await this.#storage.task(id, this.#context); if (record === undefined) return; node = nodeOf(record); - if (record.state.status === "terminal") this.#settled.set(record.id, node); + if (record.state.status === "terminal") this.#settle(id, node); + // A loaded node can reveal a cancelling owner: the next reconcile derives marks again. + this.#scheduleReconcile(); } + passed.push(id); at = parentOf(node); } else { let edge = this.#edge(at.conversation, overlay); @@ -645,15 +724,167 @@ export class TaskScheduler { const record = await this.#storage.conversation(at.conversation, this.#context); edge = record?.owner?.taskId ?? null; this.#setEdge(at.conversation, edge); + // A loaded node can reveal a cancelling owner: the next reconcile derives marks again. + this.#scheduleReconcile(); } at = edge === null ? undefined : { task: edge }; } } + for (const id of passed) this.#unloaded.delete(id); } #setEdge(conversationId: ConversationId, owner: TaskId | null): void { this.#edges.set(conversationId, owner); if (owner !== null) this.#conversationOwners.add(owner); + // The conversation now hangs from its owner, or is a root. + this.#relist(conversationNode(conversationId)); + } + + /** Keep a terminal task a chain load read, so walks pass through it; its own chain is not loaded yet. */ + #settle(id: TaskId, node: TaskNode): void { + this.#settled.set(id, node); + this.#pinned.add(id); + this.#unloaded.add(id); + this.#countChild(node, 1); + this.#relist(id); + } + + /** Count a task that became known in its owner task's `#knownChildren`, or one that is forgotten out of it. */ + #countChild(node: TaskNode, delta: 1 | -1): void { + if (node.owner === undefined) return; + const count = (this.#knownChildren.get(node.owner) ?? 0) + delta; + if (count === 0) this.#knownChildren.delete(node.owner); + else this.#knownChildren.set(node.owner, count); + } + + /** + * Forget ended task `id` unless walks up still need it: it owns a loaded conversation, a chain load read it, or it + * owns a known task; then its owner, as far up as that changes. A loop: chains can be deeper than the call stack. + */ + #forget(id: TaskId): void { + for (let at: TaskId | undefined = id; at !== undefined; ) { + const node = this.#settled.get(at); + if (node === undefined || this.#live.has(at)) return; + if (this.#conversationOwners.has(at) || this.#pinned.has(at) || this.#knownChildren.has(at)) return; + if (this.#below.has(at)) return; + this.#settled.delete(at); + this.#unloaded.delete(at); + this.#countChild(node, -1); + at = node.owner; + } + } + + /** + * Bring the scheduler's sets in line with a task's committed record. A task that ended may free the nearest live task + * above it: one holding its outcome finalizes, an abort-marked one runs its abort handler; its waiters may run. + */ + #track(previous: AnyTaskRecord | undefined, record: AnyTaskRecord): void { + const id = record.id; + if (previous?.state.status === "waiting") for (const on of previous.state.on) deleteFrom(this.#waiters, on, id); + if (record.state.status === "terminal") { + const node = nodeOf(record); + this.#live.delete(id); + this.#intent.delete(id); + this.#runnable.delete(id); + this.#finalizeChecks.delete(id); + this.#settled.set(id, node); + // Before anything is forgotten: the walk up to the task it held passes the ended tasks between them. Its chain is + // loaded, as every commit that ends a task loads the chains of `#unloaded` first. + this.#release(node); + for (const waiter of this.#waiters.get(id) ?? []) this.#runnable.add(waiter); + this.#relist(id); + this.#forget(id); + return; + } + this.#live.set(id, record); + if (previous === undefined) { + if (!this.#known(parentNode(record))) this.#unloaded.add(id); + this.#countChild(record, 1); + this.#relist(id); + } + if (cancellationIntent(record)) this.#intent.add(id); + else this.#intent.delete(id); + if (record.state.status === "completing") { + this.#runnable.delete(id); + this.#finalizeChecks.add(id); + } else { + this.#runnable.add(id); + } + if (record.state.status === "waiting") for (const on of record.state.on) addTo(this.#waiters, on, id); + } + + /** The nearest live task above ended ordinary work may be free now: one holding its outcome, or an aborting one. */ + #release(node: TaskNode, overlay?: Overlay): AnyTaskRecord | undefined { + const above = this.#nearestLiveAbove(node, overlay); + if (above?.state.status === "completing") this.#finalizeChecks.add(above.id); + else if (above?.abortRequested === true) this.#runnable.add(above.id); + return above; + } + + /** + * List `start` under its parent node in `#below` while it is a live task or has nodes listed below it, and unlist it + * otherwise; then its parent, as far up as that changes. So `#below` holds exactly the tree around live work; an + * ended parent that loses its last listed node may be forgotten (`#forget`). A loop, as ownership chains can be + * deeper than the call stack. + */ + #relist(start: Node): void { + let node: Node = start; + for (;;) { + const needed = (typeof node !== "string" && this.#live.has(node)) || this.#below.has(node); + let parent: Node | null | undefined; + if (typeof node === "string") { + parent = this.#edges.get(conversationOf(node)); + } else { + const known = this.#live.get(node) ?? this.#settled.get(node); + if (known === undefined) return; + parent = parentNode(known); + } + if (parent === undefined) return; + if (parent === null) { + const root = conversationOf(node as `c${number}`); + if (needed) this.#roots.add(root); + else this.#roots.delete(root); + return; + } + if ((this.#below.get(parent)?.has(node) === true) === needed) return; + if (needed) { + const first = !this.#below.has(parent); + addTo(this.#below, parent, node); + if (!first) return; + } else { + deleteFrom(this.#below, parent, node); + if (this.#below.has(parent)) return; + // An ended task with nothing listed below it any more may be forgotten. + if (typeof parent !== "string") this.#forget(parent); + } + node = parent; + } + } + + /** + * The nearest live task above ended ordinary work `node`, in the overlay's view: the one that work held, if any. Work + * at or below a background task holds nothing above it. + */ + #nearestLiveAbove(node: TaskNode, overlay?: Overlay): AnyTaskRecord | undefined { + for (let current = node; !current.background; ) { + const owner = current.owner ?? this.#edge(current.conversationId, overlay); + if (owner === undefined || owner === null) return undefined; + const record = overlay?.tasks.get(owner) ?? this.#live.get(owner); + if (record !== undefined && record.state.status !== "terminal") return record; + const above = record ?? this.#settled.get(owner); + if (above === undefined) return undefined; + current = above; + } + return undefined; + } + + /** Whether the chain above `node` is loaded, without walking: known tasks outside `#unloaded` have a loaded chain. */ + #known(node: Node): boolean { + if (typeof node !== "string") { + return (this.#live.has(node) || this.#settled.has(node)) && !this.#unloaded.has(node); + } + const owner = this.#edges.get(conversationOf(node)); + return owner === null || (owner !== undefined && this.#known(owner)); } /** Owner task of a conversation, `null` when ownerless, `undefined` while not loaded. */ @@ -689,42 +920,100 @@ export class TaskScheduler { } } - /** Whether every owner above `start` is loaded. */ - #chainKnown(start: Up, overlay?: Overlay): boolean { - for (const step of this.#above(start, overlay)) if ("unknown" in step) return false; - return true; - } - - /** Live records, with the overlay's candidates replacing committed ones; terminal candidates are gone. */ - *#liveRecords(overlay?: Overlay): Generator { - for (const record of this.#live.values()) { - const candidate = overlay?.tasks.get(record.id) ?? record; - if (candidate.state.status !== "terminal") yield candidate; - } - if (overlay === undefined) return; - for (const record of overlay.tasks.values()) { - if (!this.#live.has(record.id) && record.state.status !== "terminal") yield record; + /** + * The tasks below `start` that a walk down reaches: those `#below` lists and, under an overlay, the commit's new tasks, + * with its candidates in place of committed records. Background tasks are reached but not entered, unless + * `crossBackground`, and neither are the tasks `enter` rejects. A loop, as chains can be deeper than the call stack. + */ + *#walkDown( + start: Node, + options: { + readonly overlay?: Overlay; + readonly crossBackground?: boolean; + readonly enter?: (id: TaskId) => boolean; + } = {}, + ): Generator<{ readonly id: TaskId; readonly node: TaskNode; readonly live: AnyTaskRecord | undefined }> { + const { overlay, crossBackground = false, enter } = options; + const links = overlay === undefined ? undefined : this.#links(overlay); + const pending: Node[] = [start]; + for (let at = pending.pop(); at !== undefined; at = pending.pop()) { + for (const children of [this.#below.get(at), links?.get(at)]) { + for (const child of children ?? []) { + if (typeof child === "string") { + pending.push(child); + continue; + } + const record = overlay?.tasks.get(child) ?? this.#live.get(child); + const node = record ?? this.#settled.get(child); + if (node === undefined) continue; + const live = record !== undefined && record.state.status !== "terminal" ? record : undefined; + yield { id: child, node, live }; + if ((crossBackground || !node.background) && (enter === undefined || enter(child))) pending.push(child); + } + } } } /** - * Every task with live ordinary owned work (spec §5.5), mapped to that work: each live non-background task counts for - * every owner task above it up to and including the first background one. Owner chains must be loaded. + * The links walks down need for the commit's new tasks: each one under its parent node, then each node above it that + * `#below` does not list yet, until one it lists, a live task (always listed), or the chain's end. */ - #ownedLive(overlay?: Overlay): Map { - const owned = new Map(); - for (const record of this.#liveRecords(overlay)) { - if (record.background) continue; - for (const step of this.#above(parentOf(record), overlay)) { - if ("unknown" in step) break; - if (!("task" in step)) continue; - const below = owned.get(step.task); - if (below === undefined) owned.set(step.task, [record.id]); - else below.push(record.id); - if (step.node.background) break; + #links(overlay: Overlay): ReadonlyMap> { + if (overlay.links !== undefined) return overlay.links; + const links = new Map>(); + for (const record of overlay.tasks.values()) { + if (this.#live.has(record.id) || this.#settled.has(record.id)) continue; + let child: Node = record.id; + let parent: Node | null | undefined = parentNode(record); + while (parent !== undefined && parent !== null && links.get(parent)?.has(child) !== true) { + addTo(links, parent, child); + if (typeof parent !== "string" && (this.#live.has(parent) || overlay.tasks.has(parent))) break; + let above: Node | null | undefined; + if (typeof parent === "string") { + above = this.#edge(conversationOf(parent), overlay); + } else { + const fields = this.#settled.get(parent); + above = fields === undefined ? undefined : parentNode(fields); + } + if (above !== undefined && above !== null && this.#below.get(above)?.has(parent) === true) break; + child = parent; + parent = above; } } - return owned; + overlay.links = links; + return links; + } + + /** + * The live ordinary owned work of `owner` (spec §5.5): the live non-background tasks below it that a walk down reaches + * without entering a background task; with `first`, only the first. Candidates in `overlay` replace committed records; + * the walk does not enter tasks in `done`, which have none. Owner chains must be loaded. + */ + #ownedLive( + owner: TaskId, + options: { readonly overlay?: Overlay; readonly first?: boolean; readonly done?: ReadonlySet } = {}, + ): TaskId[] { + const { overlay, first = false, done } = options; + const found: TaskId[] = []; + const walk = this.#walkDown(owner, { + ...(overlay === undefined ? {} : { overlay }), + ...(done === undefined ? {} : { enter: (id: TaskId) => !done.has(id) }), + }); + for (const { id, node, live } of walk) { + if (live === undefined || node.background) continue; + found.push(id); + if (first) break; + } + return found; + } + + #hasOwnedLive(owner: TaskId, overlay?: Overlay, done?: ReadonlySet): boolean { + const options = { + first: true, + ...(overlay === undefined ? {} : { overlay }), + ...(done === undefined ? {} : { done }), + }; + return this.#ownedLive(owner, options).length > 0; } /** @@ -754,6 +1043,8 @@ export class TaskScheduler { /** The nearest live owner above `start` whose cancellation intent reaches it, if any; see `#belowCancelled`. */ #cancellingOwner(start: Up): AnyTaskRecord | undefined { + // Nothing cancels, the common case: no walk. + if (this.#intent.size === 0) return undefined; for (const step of this.#above(start)) { if ("unknown" in step) return undefined; if (!("task" in step)) continue; @@ -828,12 +1119,19 @@ export class TaskScheduler { return; } await this.#loadScopes(false); - const owned = this.#ownedLive(); // Taken once per pass, and only when some task is a candidate. let snapshot: RegistrySnapshot | undefined; - for (const record of [...this.#live.values()]) { - if (this.#invocations.has(record.id) || this.#waitingOn(record, owned).length > 0) continue; - if (record.state.status === "completing") continue; + for (const id of [...this.#runnable]) { + const record = this.#live.get(id); + if (record === undefined || record.state.status === "completing" || this.#invocations.has(id)) { + this.#runnable.delete(id); + continue; + } + // It returns once what it waits for ends, its record changes, or the work below it ends. + if (this.#waitingOn(record, true).length > 0) { + this.#runnable.delete(id); + continue; + } const runnable = record as RunnableTaskRecord; const mode = record.abortRequested ? "abort" : "run"; // Work below an owner with cancellation intent waits for its cascade mark instead of running a phase; schedule @@ -860,7 +1158,9 @@ export class TaskScheduler { }), ); } - // Registered on the line, so marks and later reservations see it and close joins it. + // Registered on the line, so marks and later reservations see it and close joins it; `#end` makes it a + // candidate again. + this.#runnable.delete(id); const invocation = this.#createInvocation(record, mode); reservations.push({ invocation, task: resolution.task, snapshot }); } @@ -868,6 +1168,7 @@ export class TaskScheduler { } catch (error) { for (const { invocation } of reservations) { this.#invocations.delete(invocation.taskId); + this.#runnable.add(invocation.taskId); invocation.finish(); } throw error; @@ -882,10 +1183,12 @@ export class TaskScheduler { * Live tasks a task waits for before its next invocation: its live ordinary owned work when abort-marked, since * abort runs bottom-up, otherwise the live part of the `on` of a wait. */ - #waitingOn(record: AnyTaskRecord, owned: ReadonlyMap): readonly TaskId[] { - if (record.abortRequested) return owned.get(record.id) ?? []; + #waitingOn(record: AnyTaskRecord, first = false): readonly TaskId[] { + if (record.abortRequested) return this.#ownedLive(record.id, { first }); if (record.state.status !== "waiting") return []; - return record.state.on.filter((id) => this.#live.has(id)); + if (!first) return record.state.on.filter((id) => this.#live.has(id)); + const live = record.state.on.find((id) => this.#live.has(id)); + return live === undefined ? [] : [live]; } /** Resolve the record's definition by kind, migrating an older stored version. */ @@ -926,23 +1229,16 @@ export class TaskScheduler { snapshot: RegistrySnapshot, ): Promise<{ scheduling: HarnessInspection["scheduling"]; tasks: TaskInspection[] }> { await this.#loadScopes(false); - const owned = this.#ownedLive(); const tasks: TaskInspection[] = []; - for (const record of this.#live.values()) { - tasks.push({ record, state: this.#inspectTask(record, snapshot, owned) }); - } + for (const record of this.#live.values()) tasks.push({ record, state: this.#inspectTask(record, snapshot) }); const scheduling = this.#closing ? "closing" : this.#enabled ? "running" : "paused"; return { scheduling, tasks }; } - #inspectTask( - record: AnyTaskRecord, - snapshot: RegistrySnapshot, - owned: ReadonlyMap, - ): TaskInspection["state"] { + #inspectTask(record: AnyTaskRecord, snapshot: RegistrySnapshot): TaskInspection["state"] { if (this.#invocations.has(record.id)) return { kind: "running" }; if (record.state.status === "completing") return { kind: "completing" }; - const on = this.#waitingOn(record, owned); + const on = this.#waitingOn(record); if (on.length > 0) return { kind: "waiting", on }; const fit = this.#fit(record, snapshot.task(record.kind)); if ("reason" in fit) return { kind: "blocked", ...fit }; @@ -1109,7 +1405,7 @@ export class TaskScheduler { */ async #terminate(tx: Transaction, record: AnyTaskRecord, outcome: SchedulerOutcome): Promise { await this.#loadScopes(false); - if (this.#ownedLive(overlayOf(tx)).has(record.id)) { + if (this.#hasOwnedLive(record.id, overlayOf(tx))) { tx.setTask(withState(record, { status: "completing", outcome })); return; } @@ -1133,7 +1429,7 @@ export class TaskScheduler { const overlay = overlayOf(tx); await this.#loadScopes(false); for (const record of overlay.tasks.values()) await this.#loadChain(parentOf(record), overlay); - if (this.#ownedLive(overlay).has(current.id)) { + if (this.#hasOwnedLive(current.id, overlay)) { tx.setTask(withState(current, { status: "completing", outcome: next.outcome })); return; } @@ -1173,7 +1469,10 @@ export class TaskScheduler { #end(invocation: Invocation): void { if (invocation.ended) return; invocation.ended = true; - if (this.#invocations.get(invocation.taskId) === invocation) this.#invocations.delete(invocation.taskId); + if (this.#invocations.get(invocation.taskId) === invocation) { + this.#invocations.delete(invocation.taskId); + if (this.#live.has(invocation.taskId)) this.#runnable.add(invocation.taskId); + } for (const watch of invocation.watches) void watch.stop(); // Pending waits bound to the invocation, such as a tool's waitForTask(), reject with it. invocation.controller.abort(endedError(invocation)); @@ -1182,8 +1481,18 @@ export class TaskScheduler { /** No live non-background task in the scope; a task whose owner edges are not loaded yet counts as inside. */ #idle(conversationId: ConversationId | undefined): boolean { const scope: Scope = conversationId === undefined ? { roots: true } : { conversation: conversationId }; - for (const record of this.#live.values()) { - if (!record.background && this.#inScope(parentOf(record), scope) !== false) return false; + // Tasks whose chain is not loaded are not listed below their scope yet; one counts as inside unless its walk up + // decides otherwise. + for (const id of this.#unloaded) { + const record = this.#live.get(id); + if (record !== undefined && !record.background && this.#inScope(parentOf(record), scope) !== false) + return false; + } + const starts = conversationId === undefined ? this.#roots : [conversationId]; + for (const start of starts) { + for (const { node, live } of this.#walkDown(conversationNode(start))) { + if (live !== undefined && !node.background) return false; + } } return true; } @@ -1456,6 +1765,19 @@ function overlayOf(tx: Transaction): Overlay { return { tasks, edges }; } +function addTo(map: Map>, key: K, value: V): void { + const values = map.get(key); + if (values === undefined) map.set(key, new Set([value])); + else values.add(value); +} + +function deleteFrom(map: Map>, key: K, value: V): void { + const values = map.get(key); + if (values === undefined) return; + values.delete(value); + if (values.size === 0) map.delete(key); +} + /** Own memo entry only; memo names such as `toString` must not resolve to inherited properties. */ function memoOf(record: AnyTaskRecord | undefined, name: string): JsonValue | undefined { const memos = record?.memos; diff --git a/packages/durable/test/harness-ownership.test.ts b/packages/durable/test/harness-ownership.test.ts index 389dfbe20..c502cfa5f 100644 --- a/packages/durable/test/harness-ownership.test.ts +++ b/packages/durable/test/harness-ownership.test.ts @@ -86,6 +86,27 @@ const Waiter = defineTask<{ on: TaskId[] }, { phase: "wait" } | { phase: "done" runtime.commit(() => ({ status: "terminal", outcome: { status: "aborted" } }), ctx), }); +/** Tries to create, in the commit that ends it, a conversation it owns with a held task in it, named by its input. */ +const Spawner = defineTask<{ name: string }, { phase: "spawn" }, null>({ + name: "test.spawner", + version: 1, + initial: () => ({ phase: "spawn" }), + phases: { + spawn: async (task, runtime, ctx) => + runtime.commit(async (tx) => { + const child = await tx.createConversation({ ownership: { kind: "task", taskId: task.id } }); + await tx.createTask( + Hold, + { name: task.input.name }, + { ownership: { kind: "conversation" }, conversationId: child.id }, + ); + return { status: "terminal", outcome: { status: "completed", result: null } }; + }, ctx), + }, + abort: async (_task, runtime, ctx) => + runtime.commit(() => ({ status: "terminal", outcome: { status: "aborted" } }), ctx), +}); + /** Never registered: aborting it can only orphan it. */ const Unregistered = defineTask<{ name: string }, { phase: "hold" }, null>({ name: "test.unregistered", @@ -162,7 +183,7 @@ async function status(harness: Harness, id: TaskId) { } async function openHarness(storage: Storage = new MemoryStorage()) { - const opened = await openTasks(storage, [Hold, Waiter]); + const opened = await openTasks(storage, [Hold, Waiter, Spawner]); const root = await opened.harness.root(context); opened.harness.resume(); return { ...opened, root }; @@ -184,6 +205,388 @@ describe("ownership", () => { await harness.close(context); }); + it("holds an owner for work in a conversation its ended child task owns", async () => { + const { harness, root } = await openHarness(); + const { owner, child, conversation } = await root.commit(async (tx) => { + const owner = await tx.createTask(Hold, { name: "owner" }, { ownership: { kind: "conversation" } }); + const child = await tx.createTask(Hold, { name: "child" }, { ownership: { kind: "task", taskId: owner } }); + const conversation = await tx.createConversation({ ownership: { kind: "task", taskId: child } }); + return { owner, child, conversation: conversation.id }; + }, context); + open("child", "completed"); + expect((await harness.waitForTask(child, context)).state.outcome.status).toBe("completed"); + // The ended child's conversation still belongs to the owner above it. + const handle = (await harness.conversation(conversation, context))!; + await handle.commit( + (tx) => tx.createTask(Hold, { name: "late" }, { ownership: { kind: "conversation" } }), + context, + ); + open("owner", "completed"); + await waitUntil(async () => (await status(harness, owner)).status === "completing"); + open("late", "completed"); + expect((await harness.waitForTask(owner, context)).state.outcome.status).toBe("completed"); + await harness.close(context); + }); + + it("cascades through ended tasks into an owned background subtree when aborting across background", async () => { + const { harness, root } = await openHarness(); + // root conversation -> A -> B -> conversation -> background C -> D + const tree = await root.commit(async (tx) => { + const a = await tx.createTask(Hold, { name: "a" }, { ownership: { kind: "conversation" } }); + const b = await tx.createTask(Hold, { name: "b" }, { ownership: { kind: "task", taskId: a } }); + const conversation = await tx.createConversation({ ownership: { kind: "task", taskId: b } }); + const c = await tx.createTask( + Hold, + { name: "c" }, + { ownership: { kind: "conversation" }, conversationId: conversation.id, background: true }, + ); + const d = await tx.createTask(Hold, { name: "d" }, { ownership: { kind: "task", taskId: c } }); + return { a, b, c, d }; + }, context); + open("b", "completed"); + open("a", "completed"); + expect((await harness.waitForTask(tree.a, context)).state.outcome.status).toBe("completed"); + await root.abort(context, { background: true }); + for (const id of [tree.c, tree.d]) { + expect((await harness.waitForTask(id, context)).state.outcome.status).toBe("aborted"); + } + await harness.close(context); + }); + + it("keeps ownership queries off finished subtrees", async () => { + const { harness, root } = await openHarness(); + // Ended chains under a live owner, each ending in an empty owned conversation. + const owner = await root.commit( + (tx) => tx.createTask(Hold, { name: "owner" }, { ownership: { kind: "conversation" } }), + context, + ); + const chains = 50; + const tasks: TaskId[] = []; + for (let n = 0; n < chains; n++) { + const ids = await root.commit(async (tx) => { + const first = await tx.createTask( + Hold, + { name: `c${n}.0` }, + { ownership: { kind: "task", taskId: owner } }, + ); + const second = await tx.createTask( + Hold, + { name: `c${n}.1` }, + { ownership: { kind: "task", taskId: first } }, + ); + await tx.createConversation({ ownership: { kind: "task", taskId: second } }); + return [first, second]; + }, context); + tasks.push(...ids); + } + for (let n = 0; n < chains; n++) { + open(`c${n}.1`, "completed"); + open(`c${n}.0`, "completed"); + } + for (const id of tasks) await harness.waitForTask(id, context); + // Set iterations while the owner settles, a measure of the walks down: one through every ended task iterates + // hundreds of entries; one over live work alone, none below the owner. + const values = Set.prototype.values; + const iterator = Set.prototype[Symbol.iterator]; + let entries = 0; + function* counted(this: Set) { + for (const value of values.call(this)) { + entries++; + yield value; + } + } + Set.prototype[Symbol.iterator] = counted as unknown as typeof iterator; + try { + open("owner", "completed"); + expect((await harness.waitForTask(owner, context)).state.outcome.status).toBe("completed"); + } finally { + Set.prototype[Symbol.iterator] = iterator; + } + expect(entries).toBeLessThan(chains); + await harness.close(context); + }); + + it("tracks ownership chains deeper than the call stack", async () => { + const { harness, root } = await openHarness(); + // 20,000 tasks, each owning the next, created in one commit above a live leaf; they end top-down and hold. + const depth = 20_000; + const { top, leaf } = await root.commit(async (tx) => { + const top = await tx.createTask(Hold, { name: "top" }, { ownership: { kind: "conversation" } }); + let owner = top; + for (let n = 0; n < depth; n++) { + owner = await tx.createTask(Hold, { name: `link${n}` }, { ownership: { kind: "task", taskId: owner } }); + } + const leaf = await tx.createTask(Hold, { name: "leaf" }, { ownership: { kind: "task", taskId: owner } }); + return { top, leaf }; + }, context); + for (let n = 0; n < depth; n++) open(`link${n}`, "completed"); + open("top", "completed"); + await waitUntil(async () => (await status(harness, top)).status === "completing"); + open("leaf", "completed"); + expect((await harness.waitForTask(leaf, context)).state.outcome.status).toBe("completed"); + expect((await harness.waitForTask(top, context)).state.outcome.status).toBe("completed"); + await harness.close(context); + }, 60_000); + + it("holds an owner for work a commit creates below ended tasks loaded from storage", async () => { + const directory = await mkdtemp(join(tmpdir(), "pi-durable-ownership-")); + directories.add(directory); + const path = join(directory, "session.sqlite"); + let opened = await openHarness(await openNodeSqliteStorage(path)); + // Live P owns B, which owns A, which owns the empty conversation C; B and A end. + const tree = await opened.root.commit(async (tx) => { + const p = await tx.createTask(Hold, { name: "p" }, { ownership: { kind: "conversation" } }); + const b = await tx.createTask(Hold, { name: "b" }, { ownership: { kind: "task", taskId: p } }); + const a = await tx.createTask(Hold, { name: "a" }, { ownership: { kind: "task", taskId: b } }); + const c = await tx.createConversation({ ownership: { kind: "task", taskId: a } }); + return { p, a, b, c: c.id }; + }, context); + open("a", "completed"); + open("b", "completed"); + await opened.harness.waitForTask(tree.a, context); + await opened.harness.waitForTask(tree.b, context); + await opened.harness.close(context); + // After reopen, B and A are known only once a chain load reads them. + opened = await openHarness(await openNodeSqliteStorage(path)); + const c = (await opened.harness.conversation(tree.c, context))!; + const d = await c.commit( + (tx) => tx.createTask(Hold, { name: "d" }, { ownership: { kind: "conversation" } }), + context, + ); + open("p", "completed"); + await waitUntil(async () => (await status(opened.harness, tree.p)).status === "completing"); + open("d", "completed"); + expect((await opened.harness.waitForTask(d, context)).state.outcome.status).toBe("completed"); + expect((await opened.harness.waitForTask(tree.p, context)).state.outcome.status).toBe("completed"); + await opened.harness.close(context); + }); + + it("holds an owner for work created later below ended tasks, without a reopen", async () => { + const { harness, root } = await openHarness(); + // Live P owns B, which owns A, which owns the empty conversation C; B and A end, C stays. + const tree = await root.commit(async (tx) => { + const p = await tx.createTask(Hold, { name: "p" }, { ownership: { kind: "conversation" } }); + const b = await tx.createTask(Hold, { name: "b" }, { ownership: { kind: "task", taskId: p } }); + const a = await tx.createTask(Hold, { name: "a" }, { ownership: { kind: "task", taskId: b } }); + const c = await tx.createConversation({ ownership: { kind: "task", taskId: a } }); + return { p, a, b, c: c.id }; + }, context); + open("a", "completed"); + open("b", "completed"); + await harness.waitForTask(tree.b, context); + const c = (await harness.conversation(tree.c, context))!; + const d = await c.commit( + (tx) => tx.createTask(Hold, { name: "d" }, { ownership: { kind: "conversation" } }), + context, + ); + open("p", "completed"); + await waitUntil(async () => (await status(harness, tree.p)).status === "completing"); + open("d", "completed"); + expect((await harness.waitForTask(d, context)).state.outcome.status).toBe("completed"); + expect((await harness.waitForTask(tree.p, context)).state.outcome.status).toBe("completed"); + await harness.close(context); + }); + + // Live P owns `depth` ended tasks in a chain, the last owning conversation C, empty when they ended. Work started in C + // later is still P's: every case below must reach it through the ended tasks. + async function endedBridge(depth: number) { + const { harness, root } = await openHarness(); + const tree = await root.commit(async (tx) => { + const p = await tx.createTask(Hold, { name: "p", slowAbort: true }, { ownership: { kind: "conversation" } }); + const links: TaskId[] = []; + let owner = p; + for (let n = 0; n < depth; n++) { + owner = await tx.createTask(Hold, { name: `link${n}` }, { ownership: { kind: "task", taskId: owner } }); + links.push(owner); + } + const c = await tx.createConversation({ ownership: { kind: "task", taskId: owner } }); + return { p, links, c: c.id }; + }, context); + for (let n = depth - 1; n >= 0; n--) open(`link${n}`, "completed"); + for (const id of tree.links) await harness.waitForTask(id, context); + const c = (await harness.conversation(tree.c, context))!; + const create = (name: string) => + c.commit((tx) => tx.createTask(Hold, { name }, { ownership: { kind: "conversation" } }), context); + return { harness, root, p: tree.p, create }; + } + + for (const depth of [2, 5]) { + it(`aborts work below ${depth} ended tasks with their live owner, which waits for it`, async () => { + const { harness, p, create } = await endedBridge(depth); + const d = await create("d"); + await harness.abortTask(p, context); + expect((await harness.waitForTask(d, context)).state.outcome.status).toBe("aborted"); + // P's abort handler ran only after D ended. + open("abort.p", "completed"); + expect((await harness.waitForTask(p, context)).state.outcome.status).toBe("aborted"); + await harness.close(context); + }); + + it(`keeps the root busy and aborts with it work below ${depth} ended tasks`, async () => { + const { harness, root, p, create } = await endedBridge(depth); + open("p", "completed"); + await harness.waitForTask(p, context); + // P has ended: D and D2 hang below ended tasks only, and are still the root conversation's work. + const d = await create("d"); + const d2 = await create("d2"); + expect(await settled(root.waitForIdle(context))).toBe(false); + expect(await settled(harness.waitForIdle(context))).toBe(false); + await root.abort(context); + for (const id of [d, d2]) + expect((await harness.waitForTask(id, context)).state.outcome.status).toBe("aborted"); + await harness.close(context); + }); + } + + it("reaches work below ended tasks across two owned conversations", async () => { + const { harness, root } = await openHarness(); + // Live P -> ended B -> conversation C -> ended X -> ended A -> conversation C2; work starts in C2 later. X owns no + // conversation, only the task A, so only owning a known task keeps it. + const tree = await root.commit(async (tx) => { + const p = await tx.createTask(Hold, { name: "p" }, { ownership: { kind: "conversation" } }); + const b = await tx.createTask(Hold, { name: "b" }, { ownership: { kind: "task", taskId: p } }); + const c = await tx.createConversation({ ownership: { kind: "task", taskId: b } }); + const x = await tx.createTask( + Hold, + { name: "x" }, + { ownership: { kind: "conversation" }, conversationId: c.id }, + ); + const a = await tx.createTask(Hold, { name: "a" }, { ownership: { kind: "task", taskId: x } }); + const c2 = await tx.createConversation({ ownership: { kind: "task", taskId: a } }); + return { p, b, x, c2: c2.id }; + }, context); + open("a", "completed"); + open("x", "completed"); + open("b", "completed"); + await harness.waitForTask(tree.b, context); + const c2 = (await harness.conversation(tree.c2, context))!; + const d = await c2.commit( + (tx) => tx.createTask(Hold, { name: "d" }, { ownership: { kind: "conversation" } }), + context, + ); + open("p", "completed"); + await waitUntil(async () => (await status(harness, tree.p)).status === "completing"); + expect(await settled(root.waitForIdle(context))).toBe(false); + await root.abort(context); + expect((await harness.waitForTask(d, context)).state.outcome.status).toBe("aborted"); + expect((await harness.waitForTask(tree.p, context)).state.outcome.status).toBe("completed"); + await harness.close(context); + }); + + for (const ending of ["completes", "is aborted"] as const) { + it(`frees an owner that ${ending} once the last work below ended tasks ends after a background task there`, async () => { + const { harness, root } = await openHarness(); + // Live P owns B, which owns A, which owns conversation C with background Z and, later, foreground D. + const tree = await root.commit(async (tx) => { + const p = await tx.createTask(Hold, { name: "p" }, { ownership: { kind: "conversation" } }); + const b = await tx.createTask(Hold, { name: "b" }, { ownership: { kind: "task", taskId: p } }); + const a = await tx.createTask(Hold, { name: "a" }, { ownership: { kind: "task", taskId: b } }); + const c = await tx.createConversation({ ownership: { kind: "task", taskId: a } }); + const z = await tx.createTask( + Hold, + { name: "z" }, + { ownership: { kind: "conversation" }, conversationId: c.id, background: true }, + ); + return { p, a, b, c: c.id, z }; + }, context); + open("a", "completed"); + open("b", "completed"); + await harness.waitForTask(tree.b, context); + const c = (await harness.conversation(tree.c, context))!; + const d = await c.commit( + (tx) => tx.createTask(Hold, { name: "d", slowAbort: true }, { ownership: { kind: "conversation" } }), + context, + ); + if (ending === "completes") { + open("p", "completed"); + await waitUntil(async () => (await status(harness, tree.p)).status === "completing"); + } else { + // P waits for D's abort before its own abort handler runs. + await harness.abortTask(tree.p, context); + await waitUntil(async () => (await harness.getTask(d, context))!.abortRequested); + } + // Z ends first, D last. + open("z", "completed"); + await harness.waitForTask(tree.z, context); + if (ending === "completes") open("d", "completed"); + else open("abort.d", "completed"); + await harness.waitForTask(d, context); + const expected = ending === "completes" ? "completed" : "aborted"; + expect((await harness.waitForTask(tree.p, context)).state.outcome.status).toBe(expected); + await harness.close(context); + }); + } + + it("finalizes many held siblings that end together in one pass", async () => { + const { harness, root } = await openHarness(); + // Top holds 1000 held children, each holding one live leaf; the leaves end in one commit. + const count = 1000; + const { top, leaves } = await root.commit(async (tx) => { + const top = await tx.createTask(Hold, { name: "top" }, { ownership: { kind: "conversation" } }); + const siblings: TaskId[] = []; + for (let n = 0; n < count; n++) { + siblings.push(await tx.createTask(Hold, { name: `s${n}` }, { ownership: { kind: "task", taskId: top } })); + } + // Leaves in reverse sibling order: the last sibling is freed first, so a holder checked after each freed sibling + // would scan the siblings already finalized each time. + const leaves: TaskId[] = []; + for (let n = count - 1; n >= 0; n--) { + const owner = siblings[n]!; + leaves.push(await tx.createTask(Hold, { name: `l${n}` }, { ownership: { kind: "task", taskId: owner } })); + } + return { top, leaves }; + }, context); + for (let n = 0; n < count; n++) open(`s${n}`, "completed"); + open("top", "completed"); + await waitUntil(async () => (await status(harness, top)).status === "completing"); + await root.commit(async (tx) => { + // Reads before the first write. Ended last to first, so a holder checked after each sibling would scan the ones + // already finalized every time. + const records = []; + for (const id of leaves) records.push((await tx.task(id))!); + for (const record of records) { + (tx as unknown as { setTask(value: unknown): void }).setTask({ + ...record, + state: { status: "terminal", outcome: { status: "completed", result: null } }, + }); + } + }, context); + // Set entries iterated while it all finalizes: one pass is linear, a recheck per sibling quadratic (~500,000). + const iterator = Set.prototype[Symbol.iterator]; + const values = Set.prototype.values; + let entries = 0; + function* counted(this: Set) { + for (const value of values.call(this)) { + entries++; + yield value; + } + } + Set.prototype[Symbol.iterator] = counted as unknown as typeof iterator; + try { + expect((await harness.waitForTask(top, context)).state.outcome.status).toBe("completed"); + } finally { + Set.prototype[Symbol.iterator] = iterator; + } + expect(entries).toBeLessThan(count * 20); + await harness.close(context); + }); + + // Pins the Transaction rule that a conversation cannot be created in the commit that ends its owner task. + it("rejects a conversation created in the commit that ends its owner task", async () => { + const { harness, root } = await openHarness(); + const spawner = await root.commit( + (tx) => tx.createTask(Spawner, { name: "spawned" }, { ownership: { kind: "conversation" } }), + context, + ); + expect((await harness.waitForTask(spawner, context)).state.outcome).toEqual({ + status: "faulted", + error: { + message: expect.stringMatching(new RegExp(`^Conversation owner task ${spawner} is (completing|terminal)$`)), + }, + }); + await harness.close(context); + }); + it("stops idle traversal at a background owner", async () => { const { harness, root } = await openHarness(); const tree = await ownedChild(root, "background", { background: true }); diff --git a/packages/durable/test/harness-scheduler-indexes.test.ts b/packages/durable/test/harness-scheduler-indexes.test.ts new file mode 100644 index 000000000..6db765f9d --- /dev/null +++ b/packages/durable/test/harness-scheduler-indexes.test.ts @@ -0,0 +1,173 @@ +// The scheduler's in-memory indexes: cases where a wrong index changes behaviour, each against what the committed +// records imply (spec §5.4, §5.5). +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { createSession, defineTask, type Harness, type TaskId } from "@earendil-works/pi-durable"; +import { afterEach, describe, expect, it } from "vitest"; +import { openNodeSqliteStorage } from "../src/storage/sqlite/node.ts"; +import { context } from "./session-support.ts"; +import { aborted, type Deferred, deferred, eventually, openTasks, settled } from "./task-support.ts"; + +const gates = new Map>(); +function gate(name: string): Deferred { + let found = gates.get(name); + if (found === undefined) { + found = deferred(); + gates.set(name, found); + } + return found; +} + +/** Holds until its gate opens, then completes; its abort handler ends it `aborted`. */ +const Hold = defineTask<{ name: string }, { phase: "hold" }, null>({ + name: "test.index.hold", + version: 1, + initial: () => ({ phase: "hold" }), + phases: { + hold: async (task, runtime, ctx) => { + const opened = await Promise.race([gate(task.input.name).promise.then(() => true), aborted(runtime.signal)]); + if (opened !== true) return; + await runtime.commit(() => ({ status: "terminal", outcome: { status: "completed", result: null } }), ctx); + }, + }, + abort: async (_task, runtime, ctx) => + runtime.commit(() => ({ status: "terminal", outcome: { status: "aborted" } }), ctx), +}); + +const directories: string[] = []; +afterEach(async () => { + gates.clear(); + for (const directory of directories.splice(0)) await rm(directory, { recursive: true, force: true }); +}); + +async function sqlitePath(): Promise { + const directory = await mkdtemp(join(tmpdir(), "pi-durable-indexes-")); + directories.push(directory); + return join(directory, "session.sqlite"); +} + +async function state(harness: Harness, id: TaskId) { + return (await harness.getTask(id, context))!; +} + +describe("scheduler indexes", () => { + it("takes a cascade's reason from the nearest cancelling owner", async () => { + // top (abort requested) -> middle (restart-marked) -> leaf: the leaf gets middle's `restart` first, as the nearest + // cancelling owner, and the request only once middle carries it. + const path = await sqlitePath(); + const session = createSession(await openNodeSqliteStorage(path)); + const ids = await session.commit(async (tx) => { + const conversation = await tx.createConversation({ ownership: { kind: "ownerless" } }); + const top = await tx.createTask( + Hold, + { name: "top" }, + { conversationId: conversation.id, ownership: { kind: "conversation" } }, + ); + const middle = await tx.createTask(Hold, { name: "middle" }, { ownership: { kind: "task", taskId: top } }); + const leaf = await tx.createTask(Hold, { name: "leaf" }, { ownership: { kind: "task", taskId: middle } }); + return { top, middle, leaf }; + }, context); + await session.commit(async (tx) => { + const top = (await tx.task(ids.top))!; + const middle = (await tx.task(ids.middle))!; + const set = (tx as unknown as { setTask(value: unknown): void }).setTask.bind(tx); + set({ ...top, abortRequested: true }); + set({ ...middle, abortRequested: true, abortReason: "restart" }); + }, context); + await session.close(context); + // Record every mark of the leaf, in commit order. + const storage = await openNodeSqliteStorage(path); + const reasons: (string | undefined)[] = []; + const commit = storage.commit.bind(storage); + storage.commit = (writes, commitContext) => { + for (const write of writes) { + if (write.type === "task" && write.value.id === ids.leaf && write.value.abortRequested) { + reasons.push(write.value.abortReason); + } + } + return commit(writes, commitContext); + }; + const { harness } = await openTasks(storage, [Hold]); + harness.resume(); + for (const id of [ids.leaf, ids.middle, ids.top]) await harness.waitForTask(id, context); + expect(reasons[0]).toBe("restart"); + await harness.close(context); + }); + + it("finalizes a chain of held owners in the commit that frees the last one", async () => { + const storage = await openNodeSqliteStorage(await sqlitePath()); + // The commit, by count, that wrote each terminal record. + const endedIn = new Map(); + let commits = 0; + const commit = storage.commit.bind(storage); + storage.commit = (writes, commitContext) => { + commits++; + for (const write of writes) { + if (write.type === "task" && write.value.state.status === "terminal") endedIn.set(write.value.id, commits); + } + return commit(writes, commitContext); + }; + const opened = await openTasks(storage, [Hold]); + const harness = opened.harness; + const root = await harness.root(context); + harness.resume(); + const { top, middle, leaf } = await root.commit(async (tx) => { + const top = await tx.createTask(Hold, { name: "top" }, { ownership: { kind: "conversation" } }); + const middle = await tx.createTask(Hold, { name: "middle" }, { ownership: { kind: "task", taskId: top } }); + const leaf = await tx.createTask(Hold, { name: "leaf" }, { ownership: { kind: "task", taskId: middle } }); + return { top, middle, leaf }; + }, context); + gate("top").resolve(); + gate("middle").resolve(); + await eventually(async () => (await state(harness, top)).state.status === "completing"); + await eventually(async () => (await state(harness, middle)).state.status === "completing"); + gate("leaf").resolve(); + expect((await harness.waitForTask(top, context)).state.outcome.status).toBe("completed"); + // Both owners end in one finalize commit, after the leaf's. + expect(endedIn.get(middle)).toBe(endedIn.get(top)); + expect(endedIn.get(middle)).toBeGreaterThan(endedIn.get(leaf)!); + await harness.close(context); + }); + + it("keeps a scope busy after reopen for work below an ended task whose chain is not loaded yet", async () => { + const path = await sqlitePath(); + let opened = await openTasks(await openNodeSqliteStorage(path), [Hold]); + let root = await opened.harness.root(context); + opened.harness.resume(); + // root conversation -> owner (ends) -> conversation C -> inner (created after owner ended), as a subagent call's. + const { owner, child } = await root.commit(async (tx) => { + const owner = await tx.createTask(Hold, { name: "owner" }, { ownership: { kind: "conversation" } }); + const child = await tx.createConversation({ ownership: { kind: "task", taskId: owner } }); + return { owner, child: child.id }; + }, context); + gate("owner").resolve(); + await opened.harness.waitForTask(owner, context); + const conversation = (await opened.harness.conversation(child, context))!; + const inner = await conversation.commit( + (tx) => tx.createTask(Hold, { name: "inner" }, { ownership: { kind: "conversation" } }), + context, + ); + await opened.harness.close(context); + gates.clear(); + // Hold the read of C's record, so the chain above `inner` stays unloaded while idle is decided. + const storage = await openNodeSqliteStorage(path); + const read = storage.conversation.bind(storage); + const held = deferred(); + storage.conversation = async (id, readContext) => { + if (id === child) await held.promise; + return read(id, readContext); + }; + opened = await openTasks(storage, [Hold]); + // A task whose chain is not loaded counts as inside every scope. (The held read blocks the Session line, so the + // root handle comes after.) + expect(await settled(opened.harness.waitForIdle(context))).toBe(false); + held.resolve(); + root = await opened.harness.root(context); + opened.harness.resume(); + gate("inner").resolve(); + await opened.harness.waitForTask(inner, context); + await root.waitForIdle(context); + await opened.harness.close(context); + }); +}); diff --git a/packages/durable/test/scheduler-fanout-bench.ts b/packages/durable/test/scheduler-fanout-bench.ts new file mode 100644 index 000000000..6976cb8d6 --- /dev/null +++ b/packages/durable/test/scheduler-fanout-bench.ts @@ -0,0 +1,74 @@ +/** + * Scheduler fan-out benchmark. One model-issued tool makes `calls` parallel nested calls of a tool that returns at + * once, on in-memory storage, so the time is the Harness's and the scheduler's own. Quadratic work shows as four times + * the time per doubling of `calls`. + * + * node --conditions=source --experimental-strip-types test/scheduler-fanout-bench.ts [calls...] + */ +import { BACKGROUND_CONTEXT as context } from "@earendil-works/chord/context"; +import { createModels } from "@earendil-works/pi-ai/models"; +import { fauxAssistantMessage, fauxProvider, fauxText, fauxToolCall } from "@earendil-works/pi-ai/providers/faux"; +import { Type } from "typebox"; +import { createRegistry, defineExtension, defineTool, Harness, ToolResultEntry } from "../src/index.ts"; +import { MemoryStorage } from "../src/storage/memory.ts"; + +async function run(calls: number): Promise<{ ms: number; commits: number }> { + const faux = fauxProvider(); + const models = createModels(); + models.setProvider(faux.provider); + const registry = createRegistry(); + const leaf = defineTool({ + name: "leaf", + description: "Return at once", + parameters: Type.Object({ n: Type.Number() }), + execute: async () => ({ output: [{ type: "text", text: "ok" }] }), + }); + const fan = defineTool({ + name: "fan", + description: "Call leaf in parallel", + parameters: Type.Object({ count: Type.Number() }), + execute: async (args, api, callContext) => { + const results = await Promise.all( + Array.from({ length: args.count }, (_, n) => + api.executeTool("leaf", { n }, callContext, { progress: false }), + ), + ); + const failed = results.filter((result) => result.isError).length; + return { output: [{ type: "text", text: `${failed} failed` }], isError: failed > 0 }; + }, + }); + registry.install(defineExtension({ name: "bench", tools: [leaf, fan] })); + const storage = new MemoryStorage(); + let commits = 0; + const commit = storage.commit.bind(storage); + storage.commit = (writes, commitContext) => { + commits++; + return commit(writes, commitContext); + }; + const harness = await Harness.open(storage, { models, registry }, context); + const root = await harness.root(context, { agent: { model: { provider: "faux", modelId: "faux-1" } } }); + faux.setResponses([ + fauxAssistantMessage([fauxToolCall("fan", { count: calls }, { id: "fan" })], { stopReason: "toolUse" }), + fauxAssistantMessage([fauxText("done")]), + ]); + const started = performance.now(); + const settled = await (await root.submit({ type: "input", content: "go" }, context)).wait(context); + if (settled.status !== "done") throw new Error(`run: ${settled.status}`); + await harness.waitForIdle(context); + const ms = Math.round(performance.now() - started); + // A run whose nested calls failed measures nothing. + const page = await root.entries({}, 10, undefined, context); + const result = page.items.find((entry) => ToolResultEntry.is(entry))?.model?.[0]; + if (result?.role !== "toolResult" || result.isError) throw new Error(`run: nested calls failed`); + await harness.close(context); + return { ms, commits }; +} + +const sizes = process.argv.slice(2).map(Number); +console.log("calls | ms | commits | ms per call"); +for (const calls of sizes.length > 0 ? sizes : [250, 500, 1000, 2000, 4000]) { + const { ms, commits } = await run(calls); + console.log( + `${String(calls).padEnd(5)} | ${String(ms).padEnd(6)} | ${String(commits).padEnd(7)} | ${(ms / calls).toFixed(2)}`, + ); +}