From 5d5a1a9e0bdc358cc2aad82d6079690c4e9b9f44 Mon Sep 17 00:00:00 2001 From: miinhho Date: Sat, 12 Sep 2026 17:10:31 +0900 Subject: [PATCH] perf: link an external cancellation source lazily, on first observation Closes #23. A Job seeded with an external signal subscribed to it eagerly in prepare(), so every Job paid an AbortSignal listener registration and removal whether or not anything could observe its cancellation. The subscription now happens the first time the Job's cancellation becomes observable: a job.signal, signal(), or context.signal read, or a child start. Internal bookkeeping reads the controller's signal directly and never subscribes. Already-aborted sources are honored without a listener by synchronous rechecks before the body, after it, and before a HandoffJob offer. An external abort during an offer suspension reaches the body only if it observed its signal first; otherwise the consumer's resume() ends the suspension and the result still reports the cancellation. Per-Job (Node 24, 400k iterations): external signal unobserved 1.99 -> 0.98 us, HandoffJob offer/resume with external signal 2.41 -> 1.44 us, plain Job unchanged. Server /echo against the packed artifact: Node 31.6k -> 41.5k rps, Bun 78k -> 101.7k rps. --- AGENTS.md | 9 ++ README.md | 3 + src/execution/context/access.ts | 12 ++- src/job/handoff-job.ts | 1 + src/job/handoff.ts | 10 +++ src/job/job.ts | 149 +++++++++++++++++++++++--------- tests/handoff-job.test.ts | 68 +++++++++++++++ tests/job-cancellation.test.ts | 74 ++++++++++++++-- 8 files changed, 277 insertions(+), 49 deletions(-) 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();