diff --git a/packages/durable/CHANGELOG.md b/packages/durable/CHANGELOG.md index 96d53480d..009e0367f 100644 --- a/packages/durable/CHANGELOG.md +++ b/packages/durable/CHANGELOG.md @@ -163,6 +163,7 @@ - A tool slot's `running` status and a nested call's final arguments, and a generation's attempt, are stored with the next commit, like streamed progress: a run with one tool call stores 12 commits instead of 14. After a crash a slot may show `pending` while its task's checkpoint is at execution; the rerun sets it again. - A wait whose commit is still being stored when the Harness closes ends with the closed error; the commit is still stored and published before close completes. - Cancellation is checked when a job starts, which can be earlier than before: an invocation-bound submit queued behind an abort mark that is not stored yet lands, and the abort's cascade then aborts its run. +- The scheduler's per-commit work no longer grows with the number of conversations: after a commit it re-checks idleness only for the conversations above the tasks that became live or ended, expires idle context ranges from an expiry-ordered index, and `wake()` takes the earliest sleep from the sleeping tasks and scans live tasks only while nothing runs. Before, every commit re-walked every conversation read in the last `contextRetentionMs` (default 10 minutes). One Session with 2048 conversations went from 100 to 805 agent runs per second (4096: 46 to 772; local SQLite at 2048: 88 to 459), with CPU per run flat at about 1.6 ms instead of growing linearly. - `subscribeCommits()` listeners are read when a commit's dispatch runs: a listener added between publication and the dispatch gets the commit only when one was registered at adoption; one added after the dispatch ran misses it. - The abort signal of a newly marked running task fires after `subscribeCommits()` listeners instead of before them. - A throwing Durable commit observer fails the Session after `Storage.commit()` returned, after the other observers and the host listeners got the commit; the commit still stands. When Storage also rejects that commit after adopting it, the storage error fails the Session and the observer's error is reported as `listener_error`. diff --git a/packages/durable/src/harness/harness.ts b/packages/durable/src/harness/harness.ts index 5ef69ef23..8fb91bcc0 100644 --- a/packages/durable/src/harness/harness.ts +++ b/packages/durable/src/harness/harness.ts @@ -106,6 +106,9 @@ type ConversationHost = { /** Key of the Harness's internal index-size getter, for tests. */ export const SCHEDULER_INDEX_SIZES = Symbol.for("pi-durable.schedulerIndexSizes"); +/** Key of the Harness's internal idle-state getter (kept contexts against a full recompute), for tests. */ +export const SCHEDULER_IDLE_STATE = Symbol.for("pi-durable.schedulerIdleState"); + class ConversationImpl implements Conversation { readonly id: ConversationId; readonly #host: ConversationHost; @@ -318,6 +321,11 @@ class HarnessImpl extends SessionImpl implements return this.#tasks.indexSizes; } + /** Internal, for tests: the kept contexts' idle state and its disagreements with a full recompute. */ + get [SCHEDULER_IDLE_STATE](): TaskScheduler["idleState"] { + return this.#tasks.idleState; + } + inspect(context: Context): Promise { return this.readOnLine(async (storage) => { const { scheduling, tasks } = await this.#tasks.inspect(this.#options.registry.snapshot()); diff --git a/packages/durable/src/harness/scheduler.ts b/packages/durable/src/harness/scheduler.ts index c0bd82ef4..63cbe89ac 100644 --- a/packages/durable/src/harness/scheduler.ts +++ b/packages/durable/src/harness/scheduler.ts @@ -314,6 +314,20 @@ export class TaskScheduler { ConversationId, { range: ContextRange; idleSince: number | undefined; expiresAt?: number } >(); + /** + * The idle kept contexts by `idleSince`, oldest first: the expiry order, so neither an expiry nor finding the next one + * scans `#contexts`. Kept sorted also when the Harness clock goes back (see `#markIdle`). + */ + readonly #idleOrder = new Map(); + /** At least the newest `idleSince` in `#idleOrder` (drops do not lower it): an older one added means the clock went back. */ + #idleNewest = Number.NEGATIVE_INFINITY; + /** + * A chain load resolved what `#idle()` treats as unknown for a live task whose chain is not loaded, which can make a + * scope idle without a task change below it; the next idle check then covers every scope. + */ + #unknownsChanged = false; + /** Idle checks that covered every scope (`idleState`, a test diagnostic). */ + #fullIdleChecks = 0; /** Background timer for the earliest expiry of an idle context, unless the host declined it (see `#scheduleExpiry`). */ #expiry: { readonly at: number; readonly disarm: () => void } | undefined; readonly #invocations = new Map(); @@ -503,8 +517,22 @@ export class TaskScheduler { const now = this.#now(); // A reconcile pass with work to do is work too: it finalizes held outcomes and marks cascades, at open as // recovery. A process that dies before it lands must be woken again. A pass with nothing to do is not. - let active = this.#invocations.size > 0 || this.#reconcilingWork > 0 || this.#reconcileWork(); + const active = + this.#invocations.size > 0 || this.#reconcilingWork > 0 || this.#reconcileWork() || this.#runsNow(now); + // The earliest sleep not yet due: only `#sleeping` holds sleeping tasks, so no pass visits every live task. let at: number | undefined; + for (const id of this.#sleeping) { + const record = this.#live.get(id); + const state = record?.state; + if (state?.status === "sleeping" && !record!.abortRequested && state.until > now) { + at = at === undefined ? state.until : Math.min(at, state.until); + } + } + return at === undefined ? { active } : { active, at }; + } + + /** Whether a live task can run now: the scan `wake()` needs only while no invocation runs. */ + #runsNow(now: number): boolean { let snapshot: RegistrySnapshot | undefined; let snapshotFailed = false; for (const record of this.#live.values()) { @@ -512,11 +540,7 @@ export class TaskScheduler { if (state.status === "completing") continue; if (state.status === "waiting" && !record.abortRequested && state.on.some((id) => this.#live.has(id))) continue; - if (state.status === "sleeping" && !record.abortRequested && state.until > now) { - at = at === undefined ? state.until : Math.min(at, state.until); - continue; - } - if (active) continue; + if (state.status === "sleeping" && !record.abortRequested && state.until > now) continue; // An abort invocation waits for the live work below it, which counts itself. if (record.abortRequested && this.#waitingOn(record, true).length > 0) continue; // A task no definition can take waits for a registry change, except an abort-marked one, which is orphaned. @@ -531,9 +555,9 @@ export class TaskScheduler { } if (snapshot !== undefined && "reason" in this.#fit(record, snapshot.task(record.kind))) continue; } - active = true; + return true; } - return at === undefined ? { active } : { active, at }; + return false; } /** @@ -744,6 +768,8 @@ export class TaskScheduler { const updated: AnyTaskRecord[] = []; const failed: TaskId[] = []; let changed = false; + // The conversations whose idleness the commit's task changes can change; undefined when a chain is not known. + let dirty: Set | undefined = new Set(); // 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)) { @@ -757,6 +783,10 @@ export class TaskScheduler { const previous = this.#live.get(record.id); if (failedOutcome(record) && (previous === undefined || !failedOutcome(previous))) failed.push(record.id); this.#track(previous, record); + // Idleness counts live tasks, so only a task that became live or ended can change it. Collected before `#sweep()`, + // which may forget the chain of a task that ended. + const liveness = previous === undefined || record.state.status === "terminal"; + if (liveness && dirty !== undefined && !this.#addScopesAbove(record, dirty)) dirty = undefined; if (record.state.status === "terminal") { this.#failedMigrations.delete(record.id); this.#failFastChecks.delete(record.id); @@ -817,10 +847,24 @@ export class TaskScheduler { if (this.#cascadePending) this.#scheduleReconcile(); this.#sweep(); if (!changed) return; - this.#settleIdle(); + this.#settleIdle(dirty); this.#kick(); } + /** + * Add the conversations above `record` to `scopes`: those whose idleness a change of it can change. False when a node + * on the way up is not known. A conversation already in `scopes` has everything above it there too. + */ + #addScopesAbove(record: AnyTaskRecord, scopes: Set): boolean { + for (const step of this.#above(parentOf(record))) { + if ("unknown" in step) return false; + if (!("conversation" in step)) continue; + if (scopes.has(step.conversation)) return true; + scopes.add(step.conversation); + } + return true; + } + /** * Extend each kept context range with the entries a stored commit appended to its conversation, so a later read * needs no Storage scan. Entries are published in ID order. @@ -841,11 +885,11 @@ export class TaskScheduler { // An idle range past its retention is dropped here too: entry-only commits do not run `#settleIdle()`, and a // host may decline the expiry timer. `expiresAt` was fixed when it went idle, so no settings are read here. if (kept.idleSince !== undefined && kept.expiresAt !== undefined && kept.expiresAt <= this.#now()) { - this.#contexts.delete(conversationId); + this.#dropContext(conversationId); continue; } const range = advanceRange(kept.range, entries); - if (range === undefined) this.#contexts.delete(conversationId); + if (range === undefined) this.#dropContext(conversationId); else kept.range = range; } } @@ -863,50 +907,98 @@ export class TaskScheduler { } } - /** Resolve idle waiters, and drop each kept context whose conversation has been idle for the retention period. */ - #settleIdle(): void { - for (const conversationId of this.#idleWaiters.keys()) { - if (this.#idle(conversationId)) this.#idleWaiters.resolve(conversationId); - } + /** + * After task changes below the conversations in `scopes` (every conversation when undefined): resolve their idle + * waiters and the Harness-wide ones, drop the kept contexts whose idle period passed the retention, then start or end + * the idle periods of the kept contexts in `scopes`. A task change can change the idleness only of the conversations + * above it, so no check visits every conversation. + */ + #settleIdle(scopes: ReadonlySet | undefined): void { + const all = scopes === undefined || this.#unknownsChanged; + this.#unknownsChanged = false; + if (all) this.#fullIdleChecks++; + this.#resolveIdleWaiters(all ? undefined : scopes); const now = this.#now(); const retention = this.#contextRetentionMs(); - for (const [conversationId, kept] of this.#contexts) { - if (kept.idleSince !== undefined && now - kept.idleSince >= retention) { - this.#contexts.delete(conversationId); - } else if (!this.#idle(conversationId)) { + // Expired first: a context past its retention is dropped even when the same commit makes it busy again. + this.#expireContexts(now, retention); + for (const conversationId of all ? [...this.#contexts.keys()] : scopes) { + const kept = this.#contexts.get(conversationId); + if (kept === undefined) continue; + if (!this.#idle(conversationId)) { + if (kept.idleSince === undefined) continue; kept.idleSince = undefined; - } else if (retention > 0) { - if (kept.idleSince === undefined) { - kept.idleSince = now; - kept.expiresAt = now + retention; - } - } else { - this.#contexts.delete(conversationId); + this.#idleOrder.delete(conversationId); + } else if (retention <= 0) { + this.#dropContext(conversationId); + } else if (kept.idleSince === undefined) { + kept.idleSince = now; + kept.expiresAt = now + retention; + this.#markIdle(conversationId, now); } } this.#scheduleExpiry(); } + /** Resolve the idle waiters of the conversations in `scopes` (all when undefined) and the Harness-wide ones. */ + #resolveIdleWaiters(scopes: ReadonlySet | undefined): void { + for (const conversationId of this.#idleWaiters.keys()) { + if (conversationId !== undefined && scopes !== undefined && !scopes.has(conversationId)) continue; + if (this.#idle(conversationId)) this.#idleWaiters.resolve(conversationId); + } + } + /** - * Run `#settleIdle()` when the earliest idle context expires, on a background timer, which never keeps the process - * alive. Where the host declines it, as the default does in Cloudflare Workers, none is kept: a pending timer could - * keep a Durable Object from being evicted, and eviction frees the contexts. There, task changes alone check expiry. + * Drop the kept contexts idle for at least `retention`, oldest first, with the retention setting of now; with no + * retention, every idle one, also those whose idle period a clock that went back makes start after now. + */ + #expireContexts(now: number, retention: number): void { + for (const [conversationId, idleSince] of this.#idleOrder) { + if (retention > 0 && now - idleSince < retention) return; + this.#dropContext(conversationId); + } + } + + /** + * Start the idle period of a kept context in the expiry order. An idle period older than the newest one kept, which + * a clock that went back starts, sorts the order again, and the newest is taken from what is kept then. + */ + #markIdle(conversationId: ConversationId, idleSince: number): void { + this.#idleOrder.delete(conversationId); + this.#idleOrder.set(conversationId, idleSince); + if (this.#idleOrder.size === 1 || idleSince >= this.#idleNewest) { + this.#idleNewest = idleSince; + return; + } + const sorted = [...this.#idleOrder].sort((a, b) => a[1] - b[1]); + this.#idleOrder.clear(); + for (const [id, since] of sorted) this.#idleOrder.set(id, since); + this.#idleNewest = sorted.at(-1)![1]; + } + + /** Forget a kept context, idle or not, and its place in the expiry order. */ + #dropContext(conversationId: ConversationId): void { + this.#contexts.delete(conversationId); + this.#idleOrder.delete(conversationId); + } + + /** + * Drop the expired contexts when the earliest idle context expires, on a background timer, which never keeps the + * process alive. Where the host declines it, as the default does in Cloudflare Workers, none is kept: a pending timer + * could keep a Durable Object from being evicted, and eviction frees the contexts. There, task changes alone check + * expiry. */ #scheduleExpiry(): void { - const retention = this.#contextRetentionMs(); - let at: number | undefined; - for (const kept of this.#contexts.values()) { - if (kept.idleSince !== undefined && (at === undefined || kept.idleSince + retention < at)) { - at = kept.idleSince + retention; - } - } + const oldest = this.#idleOrder.values().next(); + const at = oldest.done === true ? undefined : oldest.value + this.#contextRetentionMs(); if (this.#expiry !== undefined && this.#expiry.at === at) return; this.#expiry?.disarm(); this.#expiry = undefined; if (at === undefined || this.#closing) return; const disarm = this.#timers.background(() => { this.#expiry = undefined; - this.#settleIdle(); + this.#expireContexts(this.#now(), this.#contextRetentionMs()); + this.#scheduleExpiry(); }, at - this.#now()); if (disarm !== undefined) this.#expiry = { at, disarm }; } @@ -954,7 +1046,16 @@ export class TaskScheduler { } if (work) this.#reconcilingWork--; this.#scheduleWake(); - this.#settleIdle(); + // Its commit's task changes were settled at their publication; a waiter that asked for this pass checks again, and + // what its chain loads resolved can make any scope idle. + if (this.#unknownsChanged) { + this.#settleIdle(undefined); + return; + } + this.#resolveIdleWaiters(undefined); + // A pass checks expiry too, for hosts that decline the expiry timer. + this.#expireContexts(this.#now(), this.#contextRetentionMs()); + this.#scheduleExpiry(); } /** @@ -1138,46 +1239,65 @@ export class TaskScheduler { */ async #loadChain(start: Up, overlay?: Overlay): Promise { const passed: TaskId[] = []; + let installed = false; let at: Up | undefined = start; - while (at !== undefined) { - if ("task" in at) { - 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) { - // Through the line's Storage: a task whose creation or end is not published yet waits for it. - const record = await this.#session.lineStorage.task(id, this.#context); - if (record === undefined) return; - node = nodeOf(record); - if (record.state.status === "terminal") { - this.#settle(id, node); - this.#pin(id); + try { + while (at !== undefined) { + if ("task" in at) { + 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) { + // Through the line's Storage: a task whose creation or end is not published yet waits for it. + const record = await this.#session.lineStorage.task(id, this.#context); + if (record === undefined) return; + node = nodeOf(record); + if (record.state.status === "terminal") { + this.#settle(id, node); + this.#pin(id); + installed = true; + } + // A loaded node can reveal a cancelling owner: the next reconcile derives marks again. + this.#scheduleReconcile(); } - // 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); + if (edge === undefined) { + // Through the line's Storage: a conversation whose creation is not published yet waits for it, so a missing + // record is never mistaken for an ownerless one. + const record = await this.#session.lineStorage.conversation(at.conversation, this.#context); + edge = record?.owner?.taskId ?? null; + this.#setEdge(at.conversation, edge); + this.#pin(conversationNode(at.conversation)); + installed = true; + // A loaded node can reveal a cancelling owner: the next reconcile derives marks again. + this.#scheduleReconcile(); + } + at = edge === null ? undefined : { task: edge }; } - passed.push(id); - at = parentOf(node); - } else { - let edge = this.#edge(at.conversation, overlay); - if (edge === undefined) { - // Through the line's Storage: a conversation whose creation is not published yet waits for it, so a missing - // record is never mistaken for an ownerless one. - const record = await this.#session.lineStorage.conversation(at.conversation, this.#context); - edge = record?.owner?.taskId ?? null; - this.#setEdge(at.conversation, edge); - this.#pin(conversationNode(at.conversation)); - // A loaded node can reveal a cancelling owner: the next reconcile derives marks again. - this.#scheduleReconcile(); - } - at = edge === null ? undefined : { task: edge }; + } + } finally { + // `#idle()` counts a live task whose chain is not loaded as inside every scope it may be in: what this load + // resolved, also when it stopped early, can make such a scope idle without a task change below it. + if (passed.some((id) => this.#live.has(id)) || (installed && this.#liveUnloaded())) { + this.#unknownsChanged = true; + // A waiter such a task held is checked again at the pass's end, also when this load was not a reconcile's. + this.#scheduleReconcile(); } } for (const id of passed) this.#unloaded.delete(id); } + /** Whether a live task's chain is not loaded yet: one `#idle()` counts as inside every scope it may be in. */ + #liveUnloaded(): boolean { + for (const id of this.#unloaded) if (this.#live.has(id)) return true; + return false; + } + #setEdge(conversationId: ConversationId, owner: TaskId | null): void { this.#edges.set(conversationId, owner); // The conversation now hangs from its owner, or is a root; with nothing below it, it waits for the sweep. @@ -1629,6 +1749,7 @@ export class TaskScheduler { this.#taskWaiters.rejectAll(error); this.#idleWaiters.rejectAll(error); this.#contexts.clear(); + this.#idleOrder.clear(); this.#expiry?.disarm(); this.#expiry = undefined; this.#timer?.disarm(); @@ -1969,6 +2090,32 @@ export class TaskScheduler { }; } + /** + * The kept contexts' idle state, for tests (a diagnostic, computed on demand): how many are kept and how many are + * idle, how many disagree with a full recompute (an idle period without `#idle()`, or `#idle()` without one; an + * expiry order that differs from the idle periods, or is not sorted), and how many idle checks covered every scope. + */ + get idleState(): Readonly> { + let mismatches = 0; + let idle = 0; + for (const [conversationId, kept] of this.#contexts) { + if (kept.idleSince !== undefined) idle++; + if ((kept.idleSince !== undefined) !== this.#idle(conversationId)) mismatches++; + if (this.#idleOrder.get(conversationId) !== kept.idleSince) mismatches++; + } + let previous = Number.NEGATIVE_INFINITY; + for (const [conversationId, idleSince] of this.#idleOrder) { + if (idleSince < previous || this.#contexts.get(conversationId)?.idleSince !== idleSince) mismatches++; + previous = idleSince; + } + return { + contexts: this.#contexts.size, + idleContexts: idle, + idleMismatches: mismatches, + fullIdleChecks: this.#fullIdleChecks, + }; + } + /** * Scheduling state and every live task with its derived state, read on the Session line. Runs no task code: a * pending migration shows as `ready` with `migrates`, and only a migration the scheduler already tried, or one that @@ -2713,14 +2860,16 @@ export class TaskScheduler { // A read of another, idle conversation starts or continues its retention period. if (!this.#idle(conversationId)) { this.#contexts.set(conversationId, { range, idleSince: undefined }); + this.#idleOrder.delete(conversationId); } else if (this.#contextRetentionMs() > 0) { - // An idle period that already started keeps its expiry; a new one starts now. + // An idle period that already started keeps its expiry and its place; a new one starts now. const idleSince = kept?.idleSince ?? this.#now(); const expiresAt = kept?.idleSince !== undefined && kept.expiresAt !== undefined ? kept.expiresAt : idleSince + this.#contextRetentionMs(); this.#contexts.set(conversationId, { range, idleSince, expiresAt }); + if (kept?.idleSince === undefined) this.#markIdle(conversationId, idleSince); this.#scheduleExpiry(); } } diff --git a/packages/durable/test/harness-idle-scopes.test.ts b/packages/durable/test/harness-idle-scopes.test.ts new file mode 100644 index 000000000..b034150b0 --- /dev/null +++ b/packages/durable/test/harness-idle-scopes.test.ts @@ -0,0 +1,591 @@ +// Idle checks without per-commit scans: a task change settles only the conversations above it (idle waiters and kept +// context ranges), expiry runs oldest first from an ordered index, and `wake()` reads only sleeping tasks while work +// is active. Every case checks the incremental idle state against a full recompute (`SCHEDULER_IDLE_STATE`). +import { createModels } from "@earendil-works/pi-ai"; +import { + type Conversation, + type ConversationId, + createRegistry, + defineExtension, + defineTask, + Harness, + MemoryStorage, + type SetTimer, + type Storage, + type TaskId, +} from "@earendil-works/pi-durable"; +import { afterEach, describe, expect, it } from "vitest"; +import { SCHEDULER_IDLE_STATE } from "../src/harness/harness.ts"; +import { commitWrites } from "../src/testing/commit-writes.ts"; +import { ROOT_CONVERSATION_ID, type StorageWrite } from "../src/types.ts"; +import { ControlledStorage, context, flush } from "./session-support.ts"; +import { aborted, type Deferred, deferred, eventually, openTasks, settled } from "./task-support.ts"; + +const gates = new Map>(); +/** Resolves once the named task's context read returned. */ +const reads = new Map>(); + +function named(map: Map>, name: string): Deferred { + let found = map.get(name); + if (found === undefined) { + found = deferred(); + map.set(name, found); + } + return found; +} + +afterEach(() => { + gates.clear(); + reads.clear(); +}); + +/** + * Reads the context of conversation `read` (when given), then holds until its gate opens and completes; its abort + * handler ends it `aborted`. + */ +const Work = defineTask<{ name: string; read?: number }, { phase: "run" }, null>({ + name: "test.idle.work", + version: 1, + initial: () => ({ phase: "run" }), + phases: { + run: async (task, runtime, ctx) => { + if (task.input.read !== undefined) { + await runtime.context(task.input.read as ConversationId, ctx); + named(reads, task.input.name).resolve(); + } + const opened = await Promise.race([ + named(gates, 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), +}); + +type IdleState = Record<"contexts" | "idleContexts" | "idleMismatches" | "fullIdleChecks", number>; + +function idleState(harness: Harness): IdleState { + return (harness as unknown as Record)[SCHEDULER_IDLE_STATE]!; +} + +/** Background timers declined, as Cloudflare Workers do; foreground ones on `setTimeout`. */ +const declineBackground: SetTimer = (callback, delayMs, { background }) => { + if (background) return undefined; + const timer = setTimeout(callback, delayMs); + return () => clearTimeout(timer); +}; + +/** + * A resumed Harness running `Work`, over MemoryStorage counting the entry rows each scan returns, with `clock` as its + * time, the retention `retention` returns, and background timers declined unless `timers`. + */ +async function open(options: { clock?: () => number; retention?: () => number; timers?: boolean } = {}) { + const scanned = { rows: 0 }; + const memory = new MemoryStorage(); + const storage = new Proxy(memory, { + get(target, key) { + const value: unknown = Reflect.get(target, key, target); + if (key === "scanEntries") { + return async (...args: Parameters) => { + const page = await target.scanEntries(...args); + scanned.rows += page.items.length; + return page; + }; + } + return typeof value === "function" ? value.bind(target) : value; + }, + }); + const registry = createRegistry(); + registry.install(defineExtension({ name: "tasks", tasks: [Work] })); + const retention = options.retention ?? (() => 600_000); + const harness = await Harness.open( + storage, + { + models: createModels(), + registry, + settings: () => ({ contextRetentionMs: retention() }), + ...(options.clock === undefined ? {} : { now: options.clock }), + ...(options.timers === true ? {} : { setTimer: declineBackground }), + }, + context, + ); + harness.resume(); + const root = await harness.root(context); + /** A new ownerless conversation with `entries` entries. */ + const conversation = async (entries = 2): Promise => { + const created = await harness.createConversation({ ownership: { kind: "ownerless" } }, context); + for (let index = 0; index < entries; index++) { + await created.commit((tx) => tx.appendEntry(created.id, { kind: "note", data: { index } }), context); + } + return created; + }; + /** Start `Work` named `name` in `into`, reading conversation `read`, and wait for its read. */ + const work = async (name: string, into: Conversation, read?: ConversationId): Promise => { + const input = read === undefined ? { name } : { name, read: read as number }; + const id = await into.commit( + (tx) => tx.createTask(Work, input, { ownership: { kind: "conversation" }, conversationId: into.id }), + context, + ); + if (read !== undefined) await named(reads, name).promise; + return id; + }; + /** Open the gate of `name` and wait for its task, `id`, to end. */ + const end = async (name: string, id: TaskId): Promise => { + named(gates, name).resolve(); + await harness.waitForTask(id, context); + }; + /** The entry rows a read of `read` from a new task in `into` scans. */ + const rowsOf = async (name: string, into: Conversation, read: ConversationId): Promise => { + const before = scanned.rows; + const id = await work(name, into, read); + const rows = scanned.rows - before; + await end(name, id); + return rows; + }; + return { harness, root, conversation, work, end, rowsOf }; +} + +/** Flush until scheduler passes started by the last changes settled. */ +async function quiet(): Promise { + for (let round = 0; round < 4; round++) await flush(); +} + +describe("idle waiters by scope", () => { + it("resolve through owner chains, owned conversations and ended tasks between, ignoring background work", async () => { + const { harness, root } = await open(); + const tree = await root.commit(async (tx) => { + const p = await tx.createTask(Work, { name: "p" }, { ownership: { kind: "conversation" } }); + const b = await tx.createTask(Work, { name: "b" }, { ownership: { kind: "task", taskId: p } }); + const a = await tx.createTask(Work, { name: "a" }, { ownership: { kind: "task", taskId: b } }); + const c = await tx.createConversation({ ownership: { kind: "task", taskId: a } }); + const d = await tx.createTask( + Work, + { name: "d" }, + { ownership: { kind: "conversation" }, conversationId: c.id }, + ); + const g = await tx.createTask( + Work, + { name: "g" }, + { ownership: { kind: "conversation" }, conversationId: c.id, background: true }, + ); + return { p, b, a, c: c.id, d, g }; + }, context); + const c = (await harness.conversation(tree.c, context))!; + const idleC = c.waitForIdle(context); + const idleRoot = root.waitForIdle(context); + const idleAll = harness.waitForIdle(context); + // The owners A and B hold for D: live, though their handlers returned. + named(gates, "a").resolve(); + named(gates, "b").resolve(); + await eventually(async () => (await harness.getTask(tree.b, context))!.state.status === "completing"); + expect([await settled(idleC), await settled(idleRoot), await settled(idleAll)]).toEqual([false, false, false]); + // D ends: C is idle (G is background); A and B finalize; P still holds the root. + named(gates, "d").resolve(); + await idleC; + await harness.waitForTask(tree.b, context); + expect([await settled(idleRoot), await settled(idleAll)]).toEqual([false, false]); + named(gates, "p").resolve(); + await idleRoot; + await idleAll; + expect((await harness.getTask(tree.g, context))!.state.status).not.toBe("terminal"); + await harness.close(context); + }); + + it("count work whose chain is not loaded yet as inside, until the open reconcile loads it", async () => { + const storage = new ControlledStorage(); + const first = await openTasks(storage, [Work]); + const firstRoot = await first.harness.root(context); + const tree = await firstRoot.commit(async (tx) => { + const p = await tx.createTask(Work, { name: "p" }, { ownership: { kind: "conversation" } }); + const c = await tx.createConversation({ ownership: { kind: "task", taskId: p } }); + const d = await tx.createTask( + Work, + { name: "d" }, + { ownership: { kind: "conversation" }, conversationId: c.id }, + ); + const other = await tx.createConversation({ ownership: { kind: "ownerless" } }); + return { p, c: c.id, d, other: other.id }; + }, context); + await first.harness.close(context); + // Reopened, no chain is loaded: P and D count as inside every scope until the reconcile pass loads theirs. + const { harness } = await openTasks(storage.reopen(), [Work]); + const idleOther = (await harness.conversation(tree.other, context))!.waitForIdle(context); + const idleRoot = (await harness.root(context)).waitForIdle(context); + await idleOther; + expect(await settled(idleRoot)).toBe(false); + harness.resume(); + named(gates, "d").resolve(); + await harness.waitForTask(tree.d, context); + expect(await settled(idleRoot)).toBe(false); + named(gates, "p").resolve(); + await idleRoot; + await harness.close(context); + }); + + it("resolve a waiter on an unrelated conversation without settling the busy one", async () => { + const { harness, conversation, work, end } = await open(); + const [busy, other] = [await conversation(0), await conversation(0)]; + const held = await work("held", busy); + const quick = await work("quick", other); + const idleBusy = busy.waitForIdle(context); + const idleOther = other.waitForIdle(context); + await end("quick", quick); + await idleOther; + expect(await settled(idleBusy)).toBe(false); + await end("held", held); + await idleBusy; + await harness.close(context); + }); +}); + +describe("kept context ranges", () => { + it("settles a run end whose sweep drops the chain without a check of every scope", async () => { + const { harness, root } = await open(); + const tree = await root.commit(async (tx) => { + const p = await tx.createTask(Work, { name: "p" }, { ownership: { kind: "conversation" } }); + const t = await tx.createTask(Work, { name: "t" }, { ownership: { kind: "task", taskId: p } }); + const c = await tx.createConversation({ ownership: { kind: "task", taskId: t } }); + // A context range keeps entries: an empty conversation has none to keep. + await tx.appendEntry(c.id, { kind: "note" }); + return { p, t, c: c.id }; + }, context); + const c = (await harness.conversation(tree.c, context))!; + const d = await c.commit( + (tx) => + tx.createTask( + Work, + { name: "d", read: tree.c as number }, + { ownership: { kind: "conversation" }, conversationId: tree.c }, + ), + context, + ); + await named(reads, "d").promise; + named(gates, "t").resolve(); + await eventually(async () => (await harness.getTask(tree.t, context))!.state.status === "completing"); + await quiet(); + const before = idleState(harness); + expect(before).toMatchObject({ contexts: 1, idleContexts: 0, idleMismatches: 0 }); + // D's end makes C idle, T finalizes, and the sweep forgets C's chain: none of it checks every scope. + named(gates, "d").resolve(); + await harness.waitForTask(d, context); + await harness.waitForTask(tree.t, context); + await quiet(); + expect(idleState(harness)).toEqual({ ...before, idleContexts: 1 }); + named(gates, "p").resolve(); + await harness.waitForTask(tree.p, context); + await harness.close(context); + }); + + it("drops a range past its retention when a task change makes it busy again, with background timers declined", async () => { + let clock = 1_000; + const { harness, conversation, work, end } = await open({ clock: () => clock, retention: () => 1_000 }); + const [x, z] = [await conversation(), await conversation(0)]; + await end("reader", await work("reader", z, x.id)); + expect(idleState(harness)).toMatchObject({ contexts: 1, idleContexts: 1, idleMismatches: 0 }); + clock = 2_500; + // No timer expired it; the task change that makes X busy checks expiry first. + const busy = await work("busy", x); + await quiet(); + expect(idleState(harness)).toMatchObject({ contexts: 0, idleContexts: 0, idleMismatches: 0 }); + await end("busy", busy); + await harness.close(context); + }); + + it("drops a range past its retention at a reconcile pass with nothing to do, with background timers declined (review)", async () => { + let clock = 1_000; + const { harness, conversation, work, end } = await open({ clock: () => clock, retention: () => 1_000 }); + const [x, z] = [await conversation(), await conversation(0)]; + await end("reader", await work("reader", z, x.id)); + const held = await work("held", z); + // Held runs: no task changes from here. + await quiet(); + expect(idleState(harness)).toMatchObject({ contexts: 1, idleContexts: 1 }); + clock = 2_500; + // A wait on busy Z asks for a reconcile pass, which has no work and changes no task: it checks expiry. + const idle = z.waitForIdle(context); + await quiet(); + expect(idleState(harness)).toMatchObject({ contexts: 0, idleMismatches: 0 }); + await end("held", held); + await idle; + await harness.close(context); + }); + + it("expires in idle order when the clock goes back", async () => { + let clock = 1_000; + const { harness, conversation, work, end, rowsOf } = await open({ + clock: () => clock, + retention: () => 1_000, + }); + const [a, b, z] = [await conversation(3), await conversation(5), await conversation(0)]; + await end("read a", await work("read a", z, a.id)); + clock = 500; + await end("read b", await work("read b", z, b.id)); + expect(idleState(harness)).toMatchObject({ contexts: 2, idleContexts: 2, idleMismatches: 0 }); + // B went idle at 500 and expires first, though it went idle last. + clock = 1_600; + await end("unrelated", await work("unrelated", z)); + expect(idleState(harness)).toMatchObject({ contexts: 1, idleMismatches: 0 }); + expect(await rowsOf("again b", z, b.id)).toBe(5); + expect(await rowsOf("again a", z, a.id)).toBe(0); + await harness.close(context); + }); + + it("drops every idle range when retention becomes zero, also one that went idle after the clock's now (review)", async () => { + let clock = 2_000; + let retention = 1_000; + const { harness, conversation, work, end } = await open({ clock: () => clock, retention: () => retention }); + const [a, z] = [await conversation(), await conversation(0)]; + await end("read", await work("read", z, a.id)); + expect(idleState(harness)).toMatchObject({ contexts: 1, idleContexts: 1 }); + // The clock goes back below A's idle start; then retention 0 and an unrelated task change. + clock = 1_000; + retention = 0; + await end("unrelated", await work("unrelated", z)); + expect(idleState(harness)).toMatchObject({ contexts: 0, idleMismatches: 0 }); + await harness.close(context); + }); + + it("keeps an idle range's expiry when another task reads it again", async () => { + let clock = 1_000; + const { harness, conversation, work, end, rowsOf } = await open({ + clock: () => clock, + retention: () => 1_000, + }); + const [a, b, z] = [await conversation(3), await conversation(5), await conversation(0)]; + await end("read a", await work("read a", z, a.id)); + clock = 1_500; + await end("read b", await work("read b", z, b.id)); + clock = 1_800; + expect(await rowsOf("again a", z, a.id)).toBe(0); + clock = 2_100; + await end("unrelated", await work("unrelated", z)); + expect(idleState(harness)).toMatchObject({ contexts: 1, idleMismatches: 0 }); + expect(await rowsOf("late b", z, b.id)).toBe(0); + await harness.close(context); + }); + + it("drops a range from the expiry order when an entry commit finds it expired", async () => { + let clock = 1_000; + const { harness, conversation, work, end } = await open({ clock: () => clock, retention: () => 1_000 }); + const [a, z] = [await conversation(), await conversation(0)]; + await end("read", await work("read", z, a.id)); + clock = 2_500; + // An entry-only commit settles no idleness: the range is dropped where it would be extended. + await a.commit((tx) => tx.appendEntry(a.id, { kind: "note", data: { late: true } }), context); + expect(idleState(harness)).toMatchObject({ contexts: 0, idleContexts: 0, idleMismatches: 0 }); + await harness.close(context); + }); + + it("drops a range from the expiry order when an entry commit cannot extend it", async () => { + const { harness, conversation, work, end } = await open(); + const [a, z] = [await conversation(0), await conversation(0)]; + const first = await a.commit((tx) => tx.appendEntry(a.id, { kind: "note" }), context); + const second = await a.commit((tx) => tx.appendEntry(a.id, { kind: "note" }), context); + await a.commit((tx) => tx.appendEntry(a.id, { kind: "summary", head: second.id }), context); + await end("read", await work("read", z, a.id)); + expect(idleState(harness)).toMatchObject({ contexts: 1, idleContexts: 1, idleMismatches: 0 }); + // A marker whose head is older than the range's: the range cannot be extended and is dropped. + await a.commit((tx) => tx.appendEntry(a.id, { kind: "summary", head: first.id }), context); + expect(idleState(harness)).toMatchObject({ contexts: 0, idleContexts: 0, idleMismatches: 0 }); + await harness.close(context); + }); + + it.each([ + ["shorter", 400, 0], + ["longer", 10_000, 1], + ["zero", 0, 0], + ] as const)("applies a %s retention to idle ranges at an unrelated commit", async (_name, changed, kept) => { + let clock = 1_000; + let retention = 1_000; + const { harness, conversation, work, end } = await open({ clock: () => clock, retention: () => retention }); + const [a, z] = [await conversation(), await conversation(0)]; + await end("read", await work("read", z, a.id)); + clock = 1_500; + retention = changed; + await end("unrelated", await work("unrelated", z)); + expect(idleState(harness)).toMatchObject({ contexts: kept, idleMismatches: 0 }); + await harness.close(context); + }); + + it("drops a range that goes idle with zero retention", async () => { + const { harness, conversation, work, end } = await open({ retention: () => 0 }); + const a = await conversation(); + const own = await work("own", a, a.id); + await quiet(); + expect(idleState(harness)).toMatchObject({ contexts: 1, idleContexts: 0, idleMismatches: 0 }); + await end("own", own); + await quiet(); + expect(idleState(harness)).toMatchObject({ contexts: 0, idleMismatches: 0 }); + await harness.close(context); + }); +}); + +/** A random source with a fixed seed. */ +function seeded(seed: number): () => number { + let state = seed; + return () => { + state = (state * 1103515245 + 12345) % 2147483648; + return state / 2147483648; + }; +} + +describe("incremental idle state", () => { + it("equals a full recompute after every step of random nested work", async () => { + const random = seeded(42); + const pick = (items: readonly T[]): T => items[Math.floor(random() * items.length)]!; + const { harness, root } = await open(); + // A context range keeps entries: every conversation gets one. + await root.commit((tx) => tx.appendEntry(root.id, { kind: "note" }), context); + const conversations: ConversationId[] = [root.id]; + const live = new Map(); + let count = 0; + let fullChecks = idleState(harness).fullIdleChecks; + const handle = async (id: ConversationId) => (await harness.conversation(id, context))!; + for (let step = 0; step < 200; step++) { + const action = random(); + const name = `t${count++}`; + if (action < 0.1) { + const created = await harness.createConversation({ ownership: { kind: "ownerless" } }, context); + await created.commit((tx) => tx.appendEntry(created.id, { kind: "note" }), context); + conversations.push(created.id); + } else if (action < 0.45 || live.size === 0) { + // A task in any conversation, also one whose owner chain ended and was swept; it may read a context. + const into = pick(conversations); + const read = random() < 0.6 ? pick(conversations) : undefined; + const background = random() < 0.15; + const input = read === undefined ? { name } : { name, read: read as number }; + const target = await handle(into); + const id = await target.commit( + (tx) => + tx.createTask(Work, input, { ownership: { kind: "conversation" }, conversationId: into, background }), + context, + ); + live.set(name, { id, conversation: into }); + if (read !== undefined) await named(reads, name).promise; + } else if (action < 0.6) { + // Work a live task owns: a child task, or an owned conversation with a task that reads it. + const [ownerName, owner] = pick([...live]); + const state = (await harness.getTask(owner.id, context))?.state.status; + if (state === undefined || state === "terminal" || state === "completing") { + live.delete(ownerName); + continue; + } + const conversation = random() < 0.5; + const created = await (await handle(owner.conversation)).commit(async (tx) => { + if (!conversation) { + const id = await tx.createTask(Work, { name }, { ownership: { kind: "task", taskId: owner.id } }); + return { id, into: owner.conversation, read: false }; + } + const child = await tx.createConversation({ ownership: { kind: "task", taskId: owner.id } }); + await tx.appendEntry(child.id, { kind: "note" }); + const id = await tx.createTask( + Work, + { name, read: child.id as number }, + { ownership: { kind: "conversation" }, conversationId: child.id }, + ); + return { id, into: child.id, read: true }; + }, context); + if (created.read) conversations.push(created.into); + live.set(name, { id: created.id, conversation: created.into }); + if (created.read) await named(reads, name).promise; + } else { + const [ended] = pick([...live]); + live.delete(ended); + named(gates, ended).resolve(); + } + await quiet(); + const state = idleState(harness); + expect(state.idleMismatches, `step ${step}`).toBe(0); + fullChecks = state.fullIdleChecks; + } + const state = idleState(harness); + expect(state.contexts).toBeGreaterThan(5); + expect(state.idleContexts).toBeGreaterThan(0); + expect(state.idleContexts).toBeLessThan(state.contexts); + expect(fullChecks).toBe(0); + for (const gate of gates.values()) gate.resolve(); + await harness.close(context); + }); +}); + +describe("wake()", () => { + const kind = Work.definition.name; + const record = (id: number, state: unknown, extra: Record = {}): StorageWrite => ({ + type: "task", + value: { + id, + conversationId: ROOT_CONVERSATION_ID, + kind, + version: 1, + input: { name: `seeded${id}` }, + state, + background: false, + abortRequested: false, + ...extra, + } as never, + }); + const sleeping = (until: number) => ({ status: "sleeping", checkpoint: { phase: "run" }, until }); + const SLEEPS = record(10, sleeping(5_000)); + + it.each([ + ["a sleeping task", [SLEEPS], { active: false, at: 5_000 }], + [ + "an abort-marked sleeping task, which runs its handler", + [SLEEPS, record(11, sleeping(3_000), { abortRequested: true })], + { active: true, at: 5_000 }, + ], + ["a sleeping task that is due", [record(11, sleeping(500))], { active: true }], + [ + "a task waiting on a live one", + [SLEEPS, record(11, { status: "waiting", checkpoint: { phase: "run" }, on: [10], policy: "allSettled" })], + { active: false, at: 5_000 }, + ], + [ + "a completing task holding for a sleeping child", + [ + record(11, { status: "completing", outcome: { status: "completed", result: null } }), + record(10, sleeping(5_000), { owner: 11 }), + ], + { active: false, at: 5_000 }, + ], + [ + "a task no definition can take", + [SLEEPS, record(11, { status: "pending", checkpoint: { phase: "run" } }, { kind: "test.idle.missing" })], + { active: false, at: 5_000 }, + ], + [ + "a pending task", + [SLEEPS, record(11, { status: "pending", checkpoint: { phase: "run" } })], + { active: true, at: 5_000 }, + ], + ] as const)("with %s", async (_name, writes, expected) => { + const storage = new MemoryStorage(); + await commitWrites(storage, [{ type: "conversation", value: { id: ROOT_CONVERSATION_ID } }, ...writes], context); + const { harness } = await openTasks(storage, [Work], { now: () => 1_000 }); + await quiet(); + expect(harness.wake()).toEqual(expected); + await harness.close(context); + }); + + it("with an invocation running: active, and the earliest sleep still found", async () => { + const storage = new MemoryStorage(); + await commitWrites(storage, [{ type: "conversation", value: { id: ROOT_CONVERSATION_ID } }, SLEEPS], context); + const { harness } = await openTasks(storage, [Work], { now: () => 1_000 }); + harness.resume(); + const root = await harness.root(context); + const id = await root.commit( + (tx) => tx.createTask(Work, { name: "running" }, { ownership: { kind: "conversation" } }), + context, + ); + await eventually(async () => (await harness.getTask(id, context))!.state.status === "running"); + expect(harness.wake()).toEqual({ active: true, at: 5_000 }); + named(gates, "running").resolve(); + await harness.waitForTask(id, context); + await quiet(); + expect(harness.wake()).toEqual({ active: false, at: 5_000 }); + await harness.close(context); + }); +});