mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-05 23:37:25 +00:00
fix(durable): scheduler no longer visits every live task on each pass
Every reconcile pass walked up from every live task to the root to find cancelling owners, and owned live work, idle scopes, chain loads, reservation, finalization, and failFast checks each scanned all live tasks. A chain of n held tasks cost about n^3 steps: 500 took 12 s, 20,000 never finished. The ownership semantics are unchanged; in-memory indexes, rebuilt at open, hold each pass's candidates: - #below: the ownership tree downward around live work; owned live work, idle scopes, and conversation aborts walk down from one node. Ended tasks stay while walks pass through them (conversation owners, chain loads, owners of known tasks), so loaded chains stay loaded. - #intent: cascades walk down from tasks with cancellation intent, not entering another such task, so the nearest cancelling owner still decides the reason; upward checks return at once when it is empty. - #unloaded: only tasks whose chain is not loaded are loaded, and count as inside every scope until then. - #finalizeChecks: finalize is a worklist in rounds from the nearest live holder of ended work. - #runnable and #waiters: reservation and failFast candidates. No walk recurses. Chain of 500 held tasks: 12.3 s -> 50 ms; 20,000: 1.5 s. 8000 parallel nested calls: 18.8 s -> 9.2 s (bench:scheduler-fanout); the rest is document copying of the nested-call index.
This commit is contained in:
Mario Zechner
parent
c5f5b3282d
commit
cb43754931
6 changed files with 1078 additions and 104 deletions
|
|
@ -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
|
||||
|
||||
|
|
|
|||
|
|
@ -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": [
|
||||
|
|
|
|||
|
|
@ -125,8 +125,26 @@ type Step =
|
|||
type Overlay = {
|
||||
readonly tasks: ReadonlyMap<TaskId, AnyTaskRecord>;
|
||||
readonly edges: ReadonlyMap<ConversationId, TaskId | null>;
|
||||
/** Links to the staged tasks `#below` does not list, by parent node, for walks down; built at first use. */
|
||||
links?: ReadonlyMap<Node, ReadonlySet<Node>>;
|
||||
};
|
||||
|
||||
/** A node of the downward index: a task, or a conversation as `c<id>`. */
|
||||
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<TaskId, { readonly task: AnyTask; readonly error: unknown }>();
|
||||
/** Owner task of each loaded conversation, `null` when ownerless. */
|
||||
readonly #edges = new Map<ConversationId, TaskId | null>();
|
||||
/** 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<TaskId>();
|
||||
/** 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<TaskId, TaskNode>();
|
||||
/**
|
||||
* 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<Node, Set<Node>>();
|
||||
readonly #roots = new Set<ConversationId>();
|
||||
/** Known tasks whose chain may not be loaded; every other known task's chain is, and stays, loaded. */
|
||||
readonly #unloaded = new Set<TaskId>();
|
||||
/** Live tasks with cancellation intent, the sources of abort cascades (spec §5.4). */
|
||||
readonly #intent = new Set<TaskId>();
|
||||
/** Live tasks that are not `completing`, the candidates of a reservation pass; ones found waiting leave it. */
|
||||
readonly #runnable = new Set<TaskId>();
|
||||
/** `completing` tasks the next finalize pass checks: those that started holding, and holders of ended work. */
|
||||
readonly #finalizeChecks = new Set<TaskId>();
|
||||
/** Live waiting tasks by each task in their `on`. */
|
||||
readonly #waiters = new Map<TaskId, Set<TaskId>>();
|
||||
/** Terminal tasks a chain load read; kept in `#settled`, so loaded chains stay loaded. */
|
||||
readonly #pinned = new Set<TaskId>();
|
||||
/**
|
||||
* 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<TaskId, number>();
|
||||
/**
|
||||
* `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<JsonValue>);
|
||||
// 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<void> {
|
||||
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<JsonValue> }).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<TaskId>();
|
||||
const overlay = overlayOf(tx);
|
||||
while (round.length > 0) {
|
||||
const next = new Set<TaskId>();
|
||||
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<TaskId, AnyTaskRecord>).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<ConversationId[]> {
|
||||
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<void> {
|
||||
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<AnyTaskRecord> {
|
||||
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<TaskId, TaskId[]> {
|
||||
const owned = new Map<TaskId, TaskId[]>();
|
||||
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<Node, ReadonlySet<Node>> {
|
||||
if (overlay.links !== undefined) return overlay.links;
|
||||
const links = new Map<Node, Set<Node>>();
|
||||
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> } = {},
|
||||
): 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<TaskId>): 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<TaskId, readonly TaskId[]>): 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<TaskId, readonly TaskId[]>,
|
||||
): 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<void> {
|
||||
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<K, V>(map: Map<K, Set<V>>, 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<K, V>(map: Map<K, Set<V>>, 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;
|
||||
|
|
|
|||
|
|
@ -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<unknown>) {
|
||||
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<unknown>) {
|
||||
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 });
|
||||
|
|
|
|||
173
packages/durable/test/harness-scheduler-indexes.test.ts
Normal file
173
packages/durable/test/harness-scheduler-indexes.test.ts
Normal file
|
|
@ -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<string, Deferred<void>>();
|
||||
function gate(name: string): Deferred<void> {
|
||||
let found = gates.get(name);
|
||||
if (found === undefined) {
|
||||
found = deferred<void>();
|
||||
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<string> {
|
||||
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<TaskId, number>();
|
||||
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<void>();
|
||||
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);
|
||||
});
|
||||
});
|
||||
74
packages/durable/test/scheduler-fanout-bench.ts
Normal file
74
packages/durable/test/scheduler-fanout-bench.ts
Normal file
|
|
@ -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)}`,
|
||||
);
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue