diff --git a/AGENTS.md b/AGENTS.md index 9502840..189ab7d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -27,6 +27,7 @@ Runner must not import server, HTTP, WebSocket, gRPC, broker, scheduler, or opti - 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. - 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. @@ -69,6 +70,14 @@ must fail. - Use logical paragraph breaks. Extract helpers for real responsibilities, not line-count targets. - Remove obsolete exports and call paths instead of retaining compatibility shims. +## Review before handing off + +Every change ends with these three passes; they are not optional cleanup. + +- Responsibility: each module, class, and function owns one thing and its name says which. A helper that only forwards, a field that duplicates state held elsewhere, or a method exposed only so another internal caller can reach it is a smell to remove, not to document. +- Comments: a comment states an invariant, an ownership rule, or a non-obvious reason. Delete comments that narrate the code, repeat a name, or describe behavior the change removed. A stale comment is a bug. +- Documentation: `README.md` describes the current contract. When behavior, ownership, options, or a lifecycle rule changes, update the prose and examples in the same change, and remove text that describes the old model. + ## Tests and verification - Test public behavior, ownership, transitions, cancellation, cleanup ordering, and simultaneous failures. diff --git a/README.md b/README.md index 02817fa..16934f7 100644 --- a/README.md +++ b/README.md @@ -87,6 +87,7 @@ await exchange; // body and descendants have settled - `offer()` and `resume()` each accept one call; a second throws `TypeError`. - Cancellation rejects a pending `offer()` with its reason. An offer after cancellation still reaches `receive()` but rejects for the body. A later `resume()` is discarded. +- An offer consults the Job's external sources before suspending: one that already aborted rejects the offer. A source that aborts during the suspension releases it only if the body observed `signal()` first; otherwise the consumer's `resume()` ends the suspension and the Job's result still reports the cancellation. - A body that returns without offering fails. A Job that closes without offering rejects `receive()` with its failure. - `receive()` rejects self/ancestor observation synchronously. @@ -202,6 +203,8 @@ await execute({ values: [provide(Tenant, "outer")] }, async () => { `signal()` returns the current Job's cancellation signal. Pass it to cancellable native APIs, and check it after awaits and before irreversible work. Runner cannot force uncooperative code to stop. +A Job seeded with an external `signal` subscribes to it lazily, the first time its own cancellation is observed: a `signal()` or `job.signal` read, or a child starting. A Job that never observes it registers no listener. An external source that aborted before the body starts, or while it runs, still yields a cancelled result; only the wake-up of code already suspended requires the subscription. + ```ts import { setTimeout } from "node:timers/promises"; import { execute, signal, timeout } from "@tiberjs/runner"; diff --git a/src/execution/context/access.ts b/src/execution/context/access.ts index 3fa1c59..644a1c4 100644 --- a/src/execution/context/access.ts +++ b/src/execution/context/access.ts @@ -35,12 +35,18 @@ export function withContext(entries: readonly ContextEntry[], handler: () => if (entries.length === 0) { return handler(); } + const { context } = state; return runWith( { - ...state, + job: state.job, context: { - ...state.context, - values: state.context.values.withEntries(entries), + values: context.values.withEntries(entries), + // Forwarded, not copied: reading the signal stays the Job's observation point. + get signal() { + return context.signal; + }, + deadline: context.deadline, + attachment: context.attachment, }, }, handler, diff --git a/src/job/handoff-job.ts b/src/job/handoff-job.ts index af3b2d8..b7121a5 100644 --- a/src/job/handoff-job.ts +++ b/src/job/handoff-job.ts @@ -15,6 +15,7 @@ export class HandoffJob extends Job { } const rendezvous = new HandoffState(); super(() => body(rendezvous), seed); + rendezvous.owner = this; this.rendezvous = rendezvous; this.attach(rendezvous); } diff --git a/src/job/handoff.ts b/src/job/handoff.ts index eeb15d4..da9488c 100644 --- a/src/job/handoff.ts +++ b/src/job/handoff.ts @@ -9,9 +9,16 @@ export interface OwnedHandoff { settle(failure: unknown): void; } +/** The Job whose cancellation state an offer consults before suspending. */ +export interface CancellationOwner { + recheckCancellation(): void; +} + export class HandoffState implements Handoff, OwnedHandoff { private readonly value = Promise.withResolvers(); private readonly answer = Promise.withResolvers(); + /** @internal Set by HandoffJob before activation. */ + owner: CancellationOwner | undefined; offered = false; private resumed = false; private received = false; @@ -22,6 +29,9 @@ export class HandoffState implements Handoff if (this.offered) { throw new TypeError("Handoff.offer() may only be called once."); } + // A source that already aborted releases this handoff before the body suspends; + // a later abort reaches it only through the Job's own cancellation. + this.owner?.recheckCancellation(); this.offered = true; this.value.resolve(value); if (this.released) { diff --git a/src/job/job.ts b/src/job/job.ts index 641a4a9..d324bcd 100644 --- a/src/job/job.ts +++ b/src/job/job.ts @@ -6,7 +6,7 @@ import { isCancellation } from "./abort.js"; import { CancellationBindings } from "./cancellation-bindings.js"; import { FailureSet } from "./failure-set.js"; import { peekState, runWith, withoutExecution } from "../execution/state.js"; -import type { OwnedHandoff } from "./handoff.js"; +import type { CancellationOwner, OwnedHandoff } from "./handoff.js"; export interface JobStartOptions { readonly parent?: Job; @@ -22,7 +22,7 @@ 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 { +export class Job implements PromiseLike, AsyncDisposable, CancellationOwner { private readonly controller = new AbortController(); private readonly settled = Promise.withResolvers>(); private children: Set> | undefined; @@ -34,6 +34,9 @@ export class Job implements PromiseLike, AsyncDisposable { 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; private closing: Promise | undefined; private handoff: OwnedHandoff | undefined; @@ -57,7 +60,14 @@ export class Job implements PromiseLike, AsyncDisposable { return this.executionContext; } + /** + * This Job's cancellation signal. Reading it is the observation that links + * the Job to its external cancellation sources. + */ get signal(): AbortSignal { + if (this.pendingInherited !== undefined || this.pendingExternal !== undefined) { + this.observeCancellation(); + } return this.controller.signal; } @@ -82,7 +92,7 @@ export class Job implements PromiseLike, AsyncDisposable { return error; } const failure = failures.value; - return Object.is(error, failure) || Object.is(error, this.signal.reason) + return Object.is(error, failure) || Object.is(error, this.controller.signal.reason) ? failure : combinedError([error, failure], "Job rejection and execution failed."); } @@ -148,17 +158,22 @@ export class Job implements PromiseLike, AsyncDisposable { } private isExternalCancellationSource(source: AbortSignal | undefined): source is AbortSignal { - return source !== undefined && source !== this.signal && source !== this.owner?.signal; + return ( + source !== undefined && + source !== this.controller.signal && + source !== this.owner?.controller.signal + ); } private prepare(inherited: ExecutionContext | undefined): void { - // Ownership is committed; seed access and abort delivery see a base context. - this.executionContext = { - values: inherited?.values ?? ContextFrame.empty, - signal: this.signal, - deadline: inherited?.deadline, - attachment: inherited?.attachment, - }; + // 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( + this, + inherited?.values ?? ContextFrame.empty, + inherited?.deadline, + inherited?.attachment, + ); const entries = this.seed?.values; const externalSignal = this.seed?.signal; const requestedDeadline = this.seed?.deadline; @@ -172,33 +187,27 @@ export class Job implements PromiseLike, AsyncDisposable { ? inherited?.deadline : Math.min(requestedDeadline, inherited?.deadline ?? Infinity); const base = inherited?.values ?? ContextFrame.empty; - this.executionContext = { - values: - entries instanceof ContextFrame - ? entries - : entries === undefined - ? base - : base.withEntries(entries), - signal: this.signal, + this.executionContext = new JobContext( + this, + entries instanceof ContextFrame + ? entries + : entries === undefined + ? base + : base.withEntries(entries), deadline, attachment, - }; + ); const inheritedSignal = inherited?.signal; - const receivesInherited = this.isExternalCancellationSource(inheritedSignal); - const receivesExternal = this.isExternalCancellationSource(externalSignal); - if (!this.signal.aborted && (receivesInherited || receivesExternal)) { - const cancellation = (this.cancellation ??= new CancellationBindings()); - const cancel = (reason: unknown): void => this.cancel(reason); - if (receivesInherited) { - cancellation.link(inheritedSignal, cancel); - } - if (receivesExternal && !this.signal.aborted) { - cancellation.link(externalSignal, cancel); - } + if (this.isExternalCancellationSource(inheritedSignal)) { + this.pendingInherited = inheritedSignal; + } + if (this.isExternalCancellationSource(externalSignal)) { + this.pendingExternal = externalSignal; } + this.recheckCancellation(); const ownerDeadline = this.owner?.context.deadline; if ( - !this.signal.aborted && + !this.controller.signal.aborted && deadline !== undefined && (ownerDeadline === undefined || deadline < ownerDeadline) ) { @@ -208,6 +217,46 @@ export class Job implements PromiseLike, AsyncDisposable { } } + /** 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 { let value: T | undefined; let rejected = false; @@ -217,7 +266,7 @@ export class Job implements PromiseLike, AsyncDisposable { this.prepare(inherited); value = await runWith({ job: this, context: this.context }, async () => { try { - this.signal.throwIfAborted(); + this.controller.signal.throwIfAborted(); const value = await this.body(); if (this.handoff && !this.handoff.offered) { throw new TypeError("HandoffJob body completed without offering a value."); @@ -229,15 +278,17 @@ export class Job implements PromiseLike, AsyncDisposable { throw error; } }); - this.signal.throwIfAborted(); + // An external abort during the body is honored even if nothing observed it. + this.recheckCancellation(); + this.controller.signal.throwIfAborted(); } catch (error) { rejected = true; rejection = error; this.recordFailure(error); this.cancel(error); preserveCancellation = - isCancellation(error, this.signal) && - this.signal.reason !== CHILD_FAILED && + isCancellation(error, this.controller.signal) && + this.controller.signal.reason !== CHILD_FAILED && !this.failures?.hasRecorded(error); } this.phase = "closing"; @@ -251,8 +302,8 @@ export class Job implements PromiseLike, AsyncDisposable { this.rememberFailure(failure); } result = { ok: false, error: failure }; - } else if (rejected || this.signal.aborted) { - result = { ok: false, error: rejected ? rejection : this.signal.reason }; + } else if (rejected || this.controller.signal.aborted) { + result = { ok: false, error: rejected ? rejection : this.controller.signal.reason }; } else { result = { ok: true, value }; } @@ -260,7 +311,7 @@ export class Job implements PromiseLike, AsyncDisposable { } private recordFailure(error: unknown): void { - if (this.failures?.recognizes(error) || isCancellation(error, this.signal)) { + if (this.failures?.recognizes(error) || isCancellation(error, this.controller.signal)) { return; } const failures = (this.failures ??= new FailureSet("Job execution failed.")); @@ -295,10 +346,10 @@ export class Job implements PromiseLike, AsyncDisposable { continue; } job.controller.abort(received); - job.handoff?.release(job.signal.reason); + job.handoff?.release(job.controller.signal.reason); for (const child of job.children ?? []) { jobs.push(child); - reasons.push(job.signal.reason); + reasons.push(job.controller.signal.reason); } } } @@ -414,6 +465,8 @@ export class Job implements PromiseLike, AsyncDisposable { this.phase = "closed"; this.cancellation?.[Symbol.dispose](); this.cancellation = undefined; + this.pendingInherited = undefined; + this.pendingExternal = undefined; this.owner?.children?.delete(this); if (this.handoff && !this.handoff.offered && !result.ok) { this.handoff.settle(result.error); @@ -425,3 +478,17 @@ export class Job implements PromiseLike, AsyncDisposable { return this.close(); } } + +/** A Job's context: reading its signal is the Job's observation point. */ +class JobContext implements ExecutionContext { + constructor( + private readonly job: Job, + readonly values: ContextFrame, + readonly deadline: number | undefined, + readonly attachment: unknown, + ) {} + + get signal(): AbortSignal { + return this.job.signal; + } +} diff --git a/tests/handoff-job.test.ts b/tests/handoff-job.test.ts index 214a8f5..30649e5 100644 --- a/tests/handoff-job.test.ts +++ b/tests/handoff-job.test.ts @@ -1,3 +1,4 @@ +import { getEventListeners } from "node:events"; import { setImmediate as nextTurn } from "node:timers/promises"; import { expect, test } from "vitest"; import { @@ -150,6 +151,73 @@ test("an abandoned consumer cannot keep a closing Job suspended", async () => { expect(job.state).toBe("closed"); }); +test("an offer consults an external source that already aborted and rejects without subscribing", async () => { + const source = new AbortController(); + const reason = new Error("gone before the offer"); + let released: unknown; + const job = new HandoffJob( + async (handoff) => { + source.abort(reason); + try { + await handoff.offer("value"); + } catch (error) { + released = error; + } + }, + { signal: source.signal }, + ).start(); + + expect(await job.receive()).toBe("value"); + expect(await job.result()).toEqual({ ok: false, error: reason }); + expect(released).toBe(reason); + expect(getEventListeners(source.signal, "abort")).toEqual([]); +}); + +test("an unobserved external abort during suspension reaches the body only through the consumer's resume", async () => { + const source = new AbortController(); + const reason = new Error("aborted while suspended"); + let resumed: string | undefined; + const job = new HandoffJob( + async (handoff) => { + resumed = await handoff.offer("value"); + }, + { signal: source.signal }, + ).start(); + + expect(await job.receive()).toBe("value"); + source.abort(reason); + await nextTurn(); + expect(job.state).toBe("running"); + expect(getEventListeners(source.signal, "abort")).toEqual([]); + + job.resume("delivered"); + expect(await job.result()).toEqual({ ok: false, error: reason }); + expect(resumed).toBe("delivered"); +}); + +test("a body that observed its signal before offering is released by a later external abort", async () => { + const source = new AbortController(); + const reason = new Error("aborted while suspended"); + let released: unknown; + const job = new HandoffJob( + async (handoff) => { + signal(); + try { + await handoff.offer("value"); + } catch (error) { + released = error; + } + }, + { signal: source.signal }, + ).start(); + + expect(await job.receive()).toBe("value"); + source.abort(reason); + expect(await job.result()).toEqual({ ok: false, error: reason }); + expect(released).toBe(reason); + expect(getEventListeners(source.signal, "abort")).toEqual([]); +}); + test("owner cancellation cascades into a supervised HandoffJob's suspended offer", async () => { const reason = new Error("shutdown"); let released: unknown; diff --git a/tests/job-cancellation.test.ts b/tests/job-cancellation.test.ts index e9bb34c..b24a947 100644 --- a/tests/job-cancellation.test.ts +++ b/tests/job-cancellation.test.ts @@ -76,11 +76,11 @@ test("cancelling a deep ownership chain does not depend on the JavaScript call s expect(leaf.signal.reason).toBe(reason); }); -test("reentrant external cancellation never starts or leaks an admitted Job", async () => { +test("reentrant external cancellation during the first observation is visible before the observer continues", async () => { const source = new AbortController(); const reason = new Error("cancel while linking"); const add = source.signal.addEventListener.bind(source.signal); - let ran = false; + let observed: boolean | undefined; const registration = vi.spyOn(source.signal, "addEventListener").mockImplementation((...args) => { source.abort(reason); add(...args); @@ -88,12 +88,12 @@ test("reentrant external cancellation never starts or leaks an admitted Job", as try { const job = new Job( () => { - ran = true; + observed = signal().aborted; }, { signal: source.signal }, ).start(); await expect(job.join()).rejects.toBe(reason); - expect(ran).toBe(false); + expect(observed).toBe(true); expect(getEventListeners(source.signal, "abort")).toEqual([]); } finally { registration.mockRestore(); @@ -116,7 +116,13 @@ test("one external source shared by context and seed has one subscription", asyn const source = new AbortController(); const release = Promise.withResolvers(); const reason = new Error("shared source"); - const job = new Job(() => release.promise, { signal: source.signal }).start({ + const job = new Job( + () => { + signal(); + return release.promise; + }, + { signal: source.signal }, + ).start({ context: { values: ContextFrame.empty, signal: source.signal, @@ -131,6 +137,64 @@ test("one external source shared by context and seed has one subscription", asyn expect(getEventListeners(source.signal, "abort")).toEqual([]); }); +test("an unobserved external source is never subscribed to, yet its abort still ends the Job", async () => { + const source = new AbortController(); + const release = Promise.withResolvers(); + const reason = new Error("nobody looked"); + const job = new Job(() => release.promise, { signal: source.signal }).start(); + await nextTurn(); + expect(getEventListeners(source.signal, "abort")).toEqual([]); + + source.abort(reason); + release.resolve(); + expect(await job.result()).toEqual({ ok: false, error: reason }); + expect(getEventListeners(source.signal, "abort")).toEqual([]); +}); + +test("the first signal read subscribes once and a late reader sees an abort that preceded it", async () => { + const source = new AbortController(); + const release = Promise.withResolvers(); + const reason = new Error("aborted before observation"); + let seen: AbortSignal | undefined; + const job = new Job( + async () => { + await release.promise; + seen = signal(); + seen.throwIfAborted(); + }, + { signal: source.signal }, + ).start(); + source.abort(reason); + expect(getEventListeners(source.signal, "abort")).toEqual([]); + + release.resolve(); + expect(await job.result()).toEqual({ ok: false, error: reason }); + expect(seen?.aborted).toBe(true); + expect(seen?.reason).toBe(reason); +}); + +test("starting a child subscribes the parent so the cascade can reach it", async () => { + const source = new AbortController(); + const reason = new Error("cancel the tree"); + let childStopped = false; + const job = new Job( + () => { + expect(getEventListeners(source.signal, "abort")).toEqual([]); + fork(async () => { + await untilAbort(); + childStopped = true; + }); + expect(getEventListeners(source.signal, "abort")).toHaveLength(1); + }, + { signal: source.signal }, + ).start(); + + source.abort(reason); + expect(await job.result()).toEqual({ ok: false, error: reason }); + expect(childStopped).toBe(true); + expect(getEventListeners(source.signal, "abort")).toEqual([]); +}); + test("the first of two external sources wins and the other subscription is released", async () => { const inherited = new AbortController(); const seeded = new AbortController();