diff --git a/AGENTS.md b/AGENTS.md index 189ab7d..ce79b8a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -23,11 +23,11 @@ Runner must not import server, HTTP, WebSocket, gRPC, broker, scheduler, or opti - `TaskGroup` is an immutable nested declaration, not a lifetime owner. Submitted leaves belong directly to the Supervisor's Job. - `execute()` and `fork()` create real Jobs. There is no separate Task handle or compatibility lifecycle layer. - Context frames are independent of ownership. `withContext()` changes the call-chain environment without creating a Job. -- Compose support state without duplicating ownership: `FailureSet` stores errors, `CancellationBindings` releases registrations, and `GroupRunner` applies declaration policy. None owns a Job lifetime, creates an AbortController, or stores independent completion state. +- Compose support state without duplicating ownership: `FailureSet` stores errors, `Cancellation` holds one Job's incoming cancellation (its signal, followed sources, and deadline, each created on demand), and `GroupRunner` applies declaration policy. None owns a Job lifetime or stores independent completion state, and every delivered cancellation goes through the owning Job's `cancel()` so cascades stay the Job's. - Do not restore DI, EventBus, tracing, or legacy APIs to this core. - Context is immutable. Derive it with `provide(...)`; never introduce a mutable request-style bag. - Cancellation and deadlines are live state. Recheck them after awaits and before commitment points. -- An external cancellation source is linked lazily: reading `job.signal`, `signal()`, or `context.signal`, and starting a child, are the observation points that subscribe. Internal bookkeeping reads the controller's signal and never subscribes. Already-aborted sources are honored by synchronous rechecks before the body, after it, and before a handoff offer. +- An external cancellation source is linked lazily: reading `job.signal`, `signal()`, or `context.signal`, and starting a child, are the observation points that subscribe. Job bookkeeping asks its `Cancellation` and never subscribes. Already-aborted sources are honored by synchronous rechecks before the body, after it, and before a handoff offer. - Preserve native error identity and `cause`; use `AggregateError` when independent operation and cleanup failures both matter. - Cancellation classification accepts the original signal reason or a Node-style `AbortError` with `code: "ABORT_ERR"` and matching `cause`. An ordinary application error remains a failure even when its cause is the cancellation reason. diff --git a/src/execution/context/frame.ts b/src/execution/context/frame.ts index 83658b9..cb5fc53 100644 --- a/src/execution/context/frame.ts +++ b/src/execution/context/frame.ts @@ -1,11 +1,41 @@ import type { ContextEntry } from "./key.js"; -function entryMap(entries: readonly ContextEntry[]): Map { +/** What a frame needs from its own bindings; `Map` satisfies it for many, one object for one. */ +interface Bindings { + has(key: PropertyKey): boolean; + get(key: PropertyKey): unknown; + keys(): Iterable; +} + +/** The common case — one binding per frame — without a hash table. */ +class SingleBinding implements Bindings { + constructor( + private readonly key: PropertyKey, + private readonly value: unknown, + ) {} + + has(key: PropertyKey): boolean { + return key === this.key; + } + + get(key: PropertyKey): unknown { + return key === this.key ? this.value : undefined; + } + + *keys(): Iterable { + yield this.key; + } +} + +function bindingsOf(entries: readonly ContextEntry[]): Bindings { + if (entries.length === 1) { + const [key, value] = entries[0]!; + return new SingleBinding(key.id, value); + } const values = new Map(); for (const [key, value] of entries) { values.set(key.id, value); } - return values; } @@ -19,9 +49,9 @@ function entryMap(entries: readonly ContextEntry[]): Map { */ export class ContextFrame { readonly #parent: ContextFrame | null; - readonly #own: ReadonlyMap; + readonly #own: Bindings; - private constructor(parent: ContextFrame | null, own: ReadonlyMap) { + private constructor(parent: ContextFrame | null, own: Bindings) { this.#parent = parent; this.#own = own; } @@ -31,7 +61,7 @@ export class ContextFrame { /** Create a root frame containing the supplied context bindings. */ static from(entries: readonly ContextEntry[]): ContextFrame { - return entries.length === 0 ? ContextFrame.empty : new ContextFrame(null, entryMap(entries)); + return entries.length === 0 ? ContextFrame.empty : new ContextFrame(null, bindingsOf(entries)); } get(key: PropertyKey): unknown { @@ -62,18 +92,20 @@ export class ContextFrame { if (keys.length === 0) { return this; } - + if (keys.length === 1) { + const key = keys[0]!; + return new ContextFrame(this, new SingleBinding(key, values[key])); + } const own = new Map(); for (const key of keys) { - own.set(key, (values as Record)[key]); + own.set(key, values[key]); } - return new ContextFrame(this, own); } /** Return a child frame containing the supplied context bindings. */ withEntries(entries: readonly ContextEntry[]): ContextFrame { - return entries.length === 0 ? this : new ContextFrame(this, entryMap(entries)); + return entries.length === 0 ? this : new ContextFrame(this, bindingsOf(entries)); } keys(): IterableIterator { diff --git a/src/job/cancellation-bindings.ts b/src/job/cancellation-bindings.ts deleted file mode 100644 index b32f87b..0000000 --- a/src/job/cancellation-bindings.ts +++ /dev/null @@ -1,47 +0,0 @@ -import { addAbortListener } from "node:events"; -import { scheduleDeadline } from "./deadline.js"; - -/** Incoming cancellation resources, released only when their owner has drained. */ -export class CancellationBindings implements Disposable { - private links: Map | undefined; - private timer: Disposable | undefined; - - link(source: AbortSignal, onAbort: (reason: unknown) => void): void { - const links = (this.links ??= new Map()); - if (links.has(source)) { - return; - } - if (source.aborted) { - links.set(source, undefined); - onAbort(source.reason); - return; - } - - let delivered = false; - const deliver = (): void => { - if (!delivered) { - delivered = true; - onAbort(source.reason); - } - }; - const subscription = addAbortListener(source, deliver); - links.set(source, subscription); - // Listener installation can itself reenter source cancellation. - if (source.aborted) { - deliver(); - } - } - - deadline(at: number, onElapsed: () => void): void { - this.timer = scheduleDeadline(at, onElapsed); - } - - [Symbol.dispose](): void { - this.timer?.[Symbol.dispose](); - this.timer = undefined; - for (const subscription of this.links?.values() ?? []) { - subscription?.[Symbol.dispose](); - } - this.links = undefined; - } -} diff --git a/src/job/cancellation.ts b/src/job/cancellation.ts new file mode 100644 index 0000000..85cc1ee --- /dev/null +++ b/src/job/cancellation.ts @@ -0,0 +1,159 @@ +import { addAbortListener } from "node:events"; +import { isCancellation } from "./abort.js"; +import { scheduleDeadline } from "./deadline.js"; + +/** The Job whose subtree a delivered cancellation must reach. */ +export interface CancellationTarget { + cancel(reason: unknown): void; +} + +const DEADLINE_EXCEEDED = "Job deadline exceeded"; + +/** + * One Job's incoming cancellation: its own signal, the external sources it + * follows, and the deadline it owns. Everything here is created on demand. + * + * - The `AbortController` exists only once something aborts or reads the signal. + * - External sources are recorded by `follow()` and subscribed to by `observe()`, + * the first time the Job's cancellation becomes observable. `recheck()` honors a + * source that aborted while nothing observed, without subscribing. + * - The target's `cancel()` receives every delivery, so cascades stay the Job's. + * + * It owns no lifetime and no completion state. + */ +export class Cancellation { + private controller: AbortController | undefined; + private pendingInherited: AbortSignal | undefined; + private pendingExternal: AbortSignal | undefined; + private links: Map | undefined; + private timer: Disposable | undefined; + + constructor(private readonly target: CancellationTarget) {} + + get aborted(): boolean { + return this.controller?.signal.aborted === true; + } + + get reason(): unknown { + return this.controller?.signal.reason; + } + + /** The signal only once it has been created; `undefined` means "not aborted". */ + get current(): AbortSignal | undefined { + return this.controller?.signal; + } + + /** The Job's signal; reading it subscribes to the followed sources. */ + get signal(): AbortSignal { + if (this.pendingInherited !== undefined || this.pendingExternal !== undefined) { + this.observe(); + } + return (this.controller ??= new AbortController()).signal; + } + + throwIfAborted(): void { + this.controller?.signal.throwIfAborted(); + } + + isCancellation(error: unknown): boolean { + return this.controller !== undefined && isCancellation(error, this.controller.signal); + } + + abort(reason: unknown): void { + (this.controller ??= new AbortController()).abort(reason); + } + + /** + * Record the sources this Job follows. An already-aborted source cancels now; + * a live one is subscribed to only when the Job's cancellation is observed. + */ + follow(inherited: AbortSignal | undefined, external: AbortSignal | undefined): void { + const own = this.controller?.signal; + if (inherited !== undefined && inherited !== own) { + this.pendingInherited = inherited; + } + if (external !== undefined && external !== own) { + this.pendingExternal = external; + } + this.recheck(); + } + + /** Honor a followed source that aborted while nothing observed, without subscribing. */ + recheck(): void { + if (this.aborted) { + return; + } + const inherited = this.pendingInherited; + if (inherited?.aborted) { + this.target.cancel(inherited.reason); + return; + } + const external = this.pendingExternal; + if (external?.aborted) { + this.target.cancel(external.reason); + } + } + + /** Subscribe to the followed sources now that the Job's cancellation is observable. */ + observe(): void { + const inherited = this.pendingInherited; + const external = this.pendingExternal; + this.pendingInherited = undefined; + this.pendingExternal = undefined; + if (this.aborted) { + return; + } + if (inherited !== undefined) { + this.link(inherited); + } + // Linking an already-aborted inherited source cancels synchronously. + if (external !== undefined && !this.aborted) { + this.link(external); + } + } + + /** Own a deadline; the target is cancelled with a `TimeoutError` when it elapses. */ + deadline(at: number): void { + this.timer = scheduleDeadline(at, () => { + this.target.cancel(new DOMException(DEADLINE_EXCEEDED, "TimeoutError")); + }); + } + + /** Release subscriptions and the timer. The signal, if any, keeps its state. */ + dispose(): void { + this.timer?.[Symbol.dispose](); + this.timer = undefined; + for (const subscription of this.links?.values() ?? []) { + subscription?.[Symbol.dispose](); + } + this.links = undefined; + this.pendingInherited = undefined; + this.pendingExternal = undefined; + } + + private link(source: AbortSignal): void { + const links = (this.links ??= new Map()); + if (links.has(source)) { + return; + } + if (source.aborted) { + links.set(source, undefined); + this.target.cancel(source.reason); + return; + } + + let delivered = false; + const deliver = (): void => { + if (!delivered) { + delivered = true; + this.target.cancel(source.reason); + } + }; + const subscription = addAbortListener(source, deliver); + links.set(source, subscription); + // Listener installation can itself reenter source cancellation. + if (source.aborted) { + deliver(); + } + } +} diff --git a/src/job/job.ts b/src/job/job.ts index d324bcd..8fad4d7 100644 --- a/src/job/job.ts +++ b/src/job/job.ts @@ -2,8 +2,8 @@ import { ContextFrame } from "../execution/context/frame.js"; import type { ExecutionContext, ExecutionSeed } from "../execution/context/execution-context.js"; import { combinedError, LifecycleDependencyError, LifecycleStateError } from "../errors.js"; import type { Supervisor } from "../supervisor/supervisor.js"; -import { isCancellation } from "./abort.js"; -import { CancellationBindings } from "./cancellation-bindings.js"; +import { Cancellation } from "./cancellation.js"; +import type { CancellationTarget } from "./cancellation.js"; import { FailureSet } from "./failure-set.js"; import { peekState, runWith, withoutExecution } from "../execution/state.js"; import type { CancellationOwner, OwnedHandoff } from "./handoff.js"; @@ -22,8 +22,9 @@ export type JobResult = const CHILD_FAILED = new DOMException("A child job failed", "AbortError"); /** A cold, single-use execution that settles once its body and descendants finish. */ -export class Job implements PromiseLike, AsyncDisposable, CancellationOwner { - private readonly controller = new AbortController(); +export class Job + implements PromiseLike, AsyncDisposable, CancellationTarget, CancellationOwner +{ private readonly settled = Promise.withResolvers>(); private children: Set> | undefined; private owner: Job | undefined; @@ -33,10 +34,8 @@ export class Job implements PromiseLike, AsyncDisposable, CancellationOwne private supervisor: Supervisor | undefined; private propagateFailureToParent = false; private failures: FailureSet | undefined; - private cancellation: CancellationBindings | undefined; - // External sources are linked only once this Job's cancellation becomes observable. - private pendingInherited: AbortSignal | undefined; - private pendingExternal: AbortSignal | undefined; + /** Incoming cancellation; absent until something can abort or observe this Job. */ + private cancellation: Cancellation | undefined; private closing: Promise | undefined; private handoff: OwnedHandoff | undefined; @@ -65,10 +64,7 @@ export class Job implements PromiseLike, AsyncDisposable, CancellationOwne * the Job to its external cancellation sources. */ get signal(): AbortSignal { - if (this.pendingInherited !== undefined || this.pendingExternal !== undefined) { - this.observeCancellation(); - } - return this.controller.signal; + return (this.cancellation ??= new Cancellation(this)).signal; } get state(): JobState { @@ -92,7 +88,7 @@ export class Job implements PromiseLike, AsyncDisposable, CancellationOwne return error; } const failure = failures.value; - return Object.is(error, failure) || Object.is(error, this.controller.signal.reason) + return Object.is(error, failure) || Object.is(error, this.cancellation?.reason) ? failure : combinedError([error, failure], "Job rejection and execution failed."); } @@ -126,6 +122,11 @@ export class Job implements PromiseLike, AsyncDisposable, CancellationOwne this.handoff = handoff; } + /** @internal Honor a followed source that aborted while nothing observed this Job. */ + recheckCancellation(): void { + this.cancellation?.recheck(); + } + start(options: JobStartOptions = {}): this { if (this.phase !== "created") { throw new LifecycleStateError("Job", "start", this.phase); @@ -133,7 +134,8 @@ export class Job implements PromiseLike, AsyncDisposable, CancellationOwne const current = peekState(); const explicitParent = Object.hasOwn(options, "parent"); const parent = explicitParent ? options.parent : current?.job; - const inherited = options.context ?? (explicitParent ? parent?.context : current?.context); + const foreign = options.context; + const inherited = foreign ?? (explicitParent ? parent?.context : current?.context); const propagation = options.propagation ?? "propagate"; // Option accessors may have started or closed this same declaration. if (this.phase !== "created") { @@ -143,7 +145,9 @@ export class Job implements PromiseLike, AsyncDisposable, CancellationOwne if (parent.phase !== "running") { throw new LifecycleStateError("Job", "start", parent.phase); } - parent.signal.throwIfAborted(); + // A child is an observer of its parent's cancellation. + parent.cancellation?.observe(); + parent.cancellation?.throwIfAborted(); } this.owner = parent; this.propagation = propagation; @@ -152,20 +156,12 @@ export class Job implements PromiseLike, AsyncDisposable, CancellationOwne (parent.children ??= new Set()).add(this); } withoutExecution(() => { - void this.perform(inherited); + void this.perform(inherited, foreign !== undefined); }); return this; } - private isExternalCancellationSource(source: AbortSignal | undefined): source is AbortSignal { - return ( - source !== undefined && - source !== this.controller.signal && - source !== this.owner?.controller.signal - ); - } - - private prepare(inherited: ExecutionContext | undefined): void { + private prepare(inherited: ExecutionContext | undefined, foreign: boolean): void { // The Job is already running: a seed accessor may start a child under it, and a // running Job always has a context. this.executionContext = new JobContext( @@ -197,76 +193,36 @@ export class Job implements PromiseLike, AsyncDisposable, CancellationOwne deadline, attachment, ); - const inheritedSignal = inherited?.signal; - if (this.isExternalCancellationSource(inheritedSignal)) { - this.pendingInherited = inheritedSignal; - } - if (this.isExternalCancellationSource(externalSignal)) { - this.pendingExternal = externalSignal; + + // The owner's own signal reaches this Job through the cascade, never a subscription. + // A foreign context may hand back the owner's signal, so read it before comparing. + const foreignSignal = foreign ? inherited?.signal : undefined; + const ownerSignal = this.owner?.cancellation?.current; + const followedInherited = foreignSignal === ownerSignal ? undefined : foreignSignal; + const followedExternal = externalSignal === ownerSignal ? undefined : externalSignal; + if (followedInherited !== undefined || followedExternal !== undefined) { + (this.cancellation ??= new Cancellation(this)).follow(followedInherited, followedExternal); } - this.recheckCancellation(); const ownerDeadline = this.owner?.context.deadline; if ( - !this.controller.signal.aborted && deadline !== undefined && - (ownerDeadline === undefined || deadline < ownerDeadline) + (ownerDeadline === undefined || deadline < ownerDeadline) && + !this.cancellation?.aborted ) { - (this.cancellation ??= new CancellationBindings()).deadline(deadline, () => { - this.cancel(new DOMException("Job deadline exceeded", "TimeoutError")); - }); + (this.cancellation ??= new Cancellation(this)).deadline(deadline); } } - /** Subscribe to the external sources now that this Job's cancellation is observable. */ - private observeCancellation(): void { - const inherited = this.pendingInherited; - const external = this.pendingExternal; - this.pendingInherited = undefined; - this.pendingExternal = undefined; - const signal = this.controller.signal; - if (this.phase === "closed" || signal.aborted) { - return; - } - const cancellation = (this.cancellation ??= new CancellationBindings()); - const cancel = (reason: unknown): void => this.cancel(reason); - if (inherited !== undefined) { - cancellation.link(inherited, cancel); - } - // Linking an already-aborted inherited source cancels synchronously. - if (external !== undefined && !signal.aborted) { - cancellation.link(external, cancel); - } - } - - /** - * @internal Honor an external source that aborted while nothing observed this - * Job's cancellation, without subscribing to it. - */ - recheckCancellation(): void { - if (this.controller.signal.aborted) { - return; - } - const inherited = this.pendingInherited; - if (inherited?.aborted) { - this.cancel(inherited.reason); - return; - } - const external = this.pendingExternal; - if (external?.aborted) { - this.cancel(external.reason); - } - } - - private async perform(inherited: ExecutionContext | undefined): Promise { + private async perform(inherited: ExecutionContext | undefined, foreign: boolean): Promise { let value: T | undefined; let rejected = false; let rejection: unknown; let preserveCancellation = false; try { - this.prepare(inherited); + this.prepare(inherited, foreign); value = await runWith({ job: this, context: this.context }, async () => { try { - this.controller.signal.throwIfAborted(); + this.cancellation?.throwIfAborted(); const value = await this.body(); if (this.handoff && !this.handoff.offered) { throw new TypeError("HandoffJob body completed without offering a value."); @@ -279,20 +235,22 @@ export class Job implements PromiseLike, AsyncDisposable, CancellationOwne } }); // An external abort during the body is honored even if nothing observed it. - this.recheckCancellation(); - this.controller.signal.throwIfAborted(); + this.cancellation?.recheck(); + this.cancellation?.throwIfAborted(); } catch (error) { rejected = true; rejection = error; this.recordFailure(error); this.cancel(error); preserveCancellation = - isCancellation(error, this.controller.signal) && - this.controller.signal.reason !== CHILD_FAILED && + this.cancellation!.isCancellation(error) && + this.cancellation!.reason !== CHILD_FAILED && !this.failures?.hasRecorded(error); } this.phase = "closing"; - await this.drain(); + if (this.children?.size) { + await this.drain(); + } let result: JobResult; if (this.failed) { const failure = preserveCancellation @@ -302,8 +260,8 @@ export class Job implements PromiseLike, AsyncDisposable, CancellationOwne this.rememberFailure(failure); } result = { ok: false, error: failure }; - } else if (rejected || this.controller.signal.aborted) { - result = { ok: false, error: rejected ? rejection : this.controller.signal.reason }; + } else if (rejected || this.cancellation?.aborted) { + result = { ok: false, error: rejected ? rejection : this.cancellation!.reason }; } else { result = { ok: true, value }; } @@ -311,7 +269,7 @@ export class Job implements PromiseLike, AsyncDisposable, CancellationOwne } private recordFailure(error: unknown): void { - if (this.failures?.recognizes(error) || isCancellation(error, this.controller.signal)) { + if (this.failures?.recognizes(error) || this.cancellation?.isCancellation(error)) { return; } const failures = (this.failures ??= new FailureSet("Job execution failed.")); @@ -345,11 +303,12 @@ export class Job implements PromiseLike, AsyncDisposable, CancellationOwne if (job.phase === "closed") { continue; } - job.controller.abort(received); - job.handoff?.release(job.controller.signal.reason); + const cancellation = (job.cancellation ??= new Cancellation(job)); + cancellation.abort(received); + job.handoff?.release(cancellation.reason); for (const child of job.children ?? []) { jobs.push(child); - reasons.push(job.controller.signal.reason); + reasons.push(cancellation.reason); } } } @@ -448,7 +407,7 @@ export class Job implements PromiseLike, AsyncDisposable, CancellationOwne this.phase = "closing"; this.cancel(reason); if (cold) { - this.complete({ ok: false, error: this.signal.reason }); + this.complete({ ok: false, error: this.cancellation!.reason }); } } void this.settled.promise.then((result) => { @@ -463,10 +422,7 @@ export class Job implements PromiseLike, AsyncDisposable, CancellationOwne private complete(result: JobResult): void { this.phase = "closed"; - this.cancellation?.[Symbol.dispose](); - this.cancellation = undefined; - this.pendingInherited = undefined; - this.pendingExternal = undefined; + this.cancellation?.dispose(); this.owner?.children?.delete(this); if (this.handoff && !this.handoff.offered && !result.ok) { this.handoff.settle(result.error);