From 7865e2291ee93587d181fae9c5e99d13dfc2cda4 Mon Sep 17 00:00:00 2001 From: miinhho Date: Sat, 12 Sep 2026 16:16:09 +0900 Subject: [PATCH 1/4] refactor: return Job to a single-result lifetime Remove the Published type parameter, JobPublisher, the publisher body argument, Job.value(), and publication state. A Job answers one question: the final result once its body and descendants settle. One-shot handoff belongs to the consumer that needs it, composed with an ordinary Job. Closes #21 --- README.md | 23 +--------- src/execution/state.ts | 2 +- src/index.ts | 2 +- src/job/job.ts | 82 +++++----------------------------- src/supervisor/group-plan.ts | 4 +- src/supervisor/group-runner.ts | 6 +-- src/supervisor/supervisor.ts | 20 ++++----- src/supervisor/task-group.ts | 4 +- tests/job.test.ts | 78 -------------------------------- 9 files changed, 30 insertions(+), 191 deletions(-) diff --git a/README.md b/README.md index 64de170..7e0bf5a 100644 --- a/README.md +++ b/README.md @@ -30,7 +30,6 @@ const result = await execute(async () => { - `fork()` starts a child of the current Job. `new Job(body).start()` does the same inside an execution, or starts a root outside one. - `start({ parent })` selects an explicit owner and inherits that owner's context. `start({ parent: undefined })` starts an independent root. - `cancel(reason)` requests cooperative cancellation. It does not mean the work has finished. -- `value()` observes a value published by the body without waiting for descendants. It rejects if the Job settles without invoking its publisher. - `join()` and `await job` observe the complete result. Awaiting a cold Job rejects; awaiting never starts it. - `finish()` stops admission of direct children and joins without cancellation. `close(reason)` stops admission, cancels, and joins. Repeated `close()` calls share their result. - `close()` ignores expected cancellation but rejects genuine execution or finalizer failures. Jobs support `await using`. @@ -55,27 +54,7 @@ if (result.ok) { } ``` -Use `ok` to distinguish success from failure: both a successful value and a thrown error may be `undefined`. `result()` and `value()` reject self/ancestor observation synchronously. - -The body receives a publisher that may be called once while the body is running. -`Job` keeps the published value's type separate from the -body's eventual result. The publisher accepts a value or `PromiseLike` value. -Cancellation does not disable publication; `result()` and `await job` still report the -Job's final success or failure independently. - -```ts -const release = Promise.withResolvers(); -const job = new Job(async (publish) => { - publish("ready"); - await release.promise; - return 42; -}); - -job.start(); -console.log(await job.value()); // "ready"; the Job is still running. -release.resolve(); -console.log(await job); // 42; the body and descendants have settled. -``` +Use `ok` to distinguish success from failure: both a successful value and a thrown error may be `undefined`. `result()` rejects self/ancestor observation synchronously. `cancelChildren(reason)` cancels direct children, lets cancellation cascade through their descendants, and resolves after those child lifetimes settle. It does not cancel diff --git a/src/execution/state.ts b/src/execution/state.ts index 87ddffc..3aabe9f 100644 --- a/src/execution/state.ts +++ b/src/execution/state.ts @@ -7,7 +7,7 @@ import type { Job } from "../job/job.js"; */ export interface RuntimeState { readonly context: ExecutionContext; - readonly job: Job; + readonly job: Job; } const storage = new AsyncLocalStorage(); diff --git a/src/index.ts b/src/index.ts index 5562fde..3c60473 100644 --- a/src/index.ts +++ b/src/index.ts @@ -24,6 +24,6 @@ export { } from "./execution/context/access.js"; export { Job } from "./job/job.js"; -export type { JobPublisher, JobStartOptions, JobState, JobResult } from "./job/job.js"; +export type { JobStartOptions, JobState, JobResult } from "./job/job.js"; export { TaskGroup } from "./supervisor/task-group.js"; export type { TaskGroupOptions, GroupMember, GroupResults } from "./supervisor/task-group.js"; diff --git a/src/job/job.ts b/src/job/job.ts index 23fec58..757244a 100644 --- a/src/job/job.ts +++ b/src/job/job.ts @@ -8,7 +8,7 @@ import { FailureSet } from "./failure-set.js"; import { peekState, runWith, withoutExecution } from "../execution/state.js"; export interface JobStartOptions { - readonly parent?: Job; + readonly parent?: Job; readonly context?: ExecutionContext; readonly propagation?: "propagate" | "isolate"; } @@ -18,30 +18,25 @@ export type JobResult = | { readonly ok: true; readonly value: T } | { readonly ok: false; readonly error: unknown }; -export type JobPublisher = (value: T | PromiseLike) => void; - const CHILD_FAILED = new DOMException("A child job failed", "AbortError"); -/** A cold, single-use execution that may publish before its full lifetime settles. */ -export class Job implements PromiseLike, AsyncDisposable { +/** A cold, single-use execution that settles once its body and descendants finish. */ +export class Job implements PromiseLike, AsyncDisposable { private readonly controller = new AbortController(); private readonly settled = Promise.withResolvers>(); - private publication?: PromiseWithResolvers>; - private publicationResult?: JobResult; - private publishing = false; - private children: Set> | undefined; - private owner: Job | undefined; + private children: Set> | undefined; + private owner: Job | undefined; private executionContext: ExecutionContext | undefined; private phase: JobState = "created"; private propagation: "propagate" | "isolate" = "propagate"; - private supervisor: Supervisor | undefined; + private supervisor: Supervisor | undefined; private propagateFailureToParent = false; private failures: FailureSet | undefined; private cancellation: CancellationBindings | undefined; private closing: Promise | undefined; constructor( - private readonly body: (publish: JobPublisher) => T | PromiseLike, + private readonly body: () => T | PromiseLike, private readonly seed?: ExecutionSeed, ) { if (typeof body !== "function") { @@ -49,24 +44,7 @@ export class Job implements PromiseLike, AsyncDisposable { } } - /** - * Publish one value while the body is running, independently of Job cancellation. - */ - private readonly publish = (value: unknown | PromiseLike): void => { - if (this.phase !== "running") { - throw new LifecycleStateError("Job", "publish", this.phase); - } - if (this.publishing) { - throw new TypeError("Job.publish() may only be called once."); - } - this.publishing = true; - void Promise.resolve(value).then( - (published) => this.resolvePublication({ ok: true, value: published }), - (error: unknown) => this.resolvePublication({ ok: false, error }), - ); - }; - - get parent(): Job | undefined { + get parent(): Job | undefined { return this.owner; } @@ -111,7 +89,7 @@ export class Job implements PromiseLike, AsyncDisposable { return this.children?.size ?? 0; } - owns(other: Job | undefined): boolean { + owns(other: Job | undefined): boolean { for (let node = other; node; node = node.owner) { if (node === this) { return true; @@ -121,7 +99,7 @@ export class Job implements PromiseLike, AsyncDisposable { } /** Bind an optional manager before activation; ownership remains on this Job. */ - manage(supervisor: Supervisor): void { + manage(supervisor: Supervisor): void { if (this.phase !== "created" || this.supervisor) { throw new LifecycleStateError("Job", "manage", this.phase); } @@ -230,7 +208,7 @@ export class Job implements PromiseLike, AsyncDisposable { value = await runWith({ job: this, context: this.context }, async () => { try { this.signal.throwIfAborted(); - return await this.body(this.publish); + return await this.body(); } catch (error) { // Capture before a caller can cancel with this same value. this.recordFailure(error); @@ -294,7 +272,7 @@ export class Job implements PromiseLike, AsyncDisposable { } cancel(reason?: unknown): void { - const jobs: Job[] = [this]; + const jobs: Job[] = [this]; const reasons: unknown[] = [reason]; while (jobs.length > 0) { const job = jobs.pop()!; @@ -348,29 +326,6 @@ export class Job implements PromiseLike, AsyncDisposable { return this.settled.promise as Promise>; } - /** - * Observe the body-published value without joining the Job's descendants. - * - * A Job that closes without publishing rejects with its failure, or with a - * lifecycle error when the Job itself succeeded. - */ - value(): Promise { - const dependency = this.dependency("value"); - if (dependency) { - throw dependency; - } - const publication = (this.publication ??= Promise.withResolvers>()); - if (this.publicationResult) { - publication.resolve(this.publicationResult); - } - return publication.promise.then((result) => { - if (!result.ok) { - throw result.error; - } - return result.value as Published; - }); - } - joinChildren(): Promise { const current = peekState()?.job; if (current !== this && this.owns(current)) { @@ -440,21 +395,8 @@ export class Job implements PromiseLike, AsyncDisposable { return promise; } - private resolvePublication(result: JobResult): void { - this.publicationResult = result; - this.publication?.resolve(result); - } - private complete(result: JobResult): void { this.phase = "closed"; - if (!this.publishing) { - this.publishing = true; - this.resolvePublication( - result.ok - ? { ok: false, error: new LifecycleStateError("Job", "value", this.phase) } - : result, - ); - } this.cancellation?.[Symbol.dispose](); this.cancellation = undefined; this.owner?.children?.delete(this); diff --git a/src/supervisor/group-plan.ts b/src/supervisor/group-plan.ts index e96d343..5a6756b 100644 --- a/src/supervisor/group-plan.ts +++ b/src/supervisor/group-plan.ts @@ -12,7 +12,7 @@ export interface GroupBoundary { } export interface GroupLeaf { - readonly job: Job; + readonly job: Job; readonly boundary: GroupBoundary; readonly results: unknown[]; readonly index: number; @@ -24,7 +24,7 @@ export function planGroup(group: TaskGroup): { results: unknown[]; } { const leaves: GroupLeaf[] = []; - const seen = new Set>(); + const seen = new Set>(); const visit = ( declaration: TaskGroup, diff --git a/src/supervisor/group-runner.ts b/src/supervisor/group-runner.ts index 2290f94..43c3a57 100644 --- a/src/supervisor/group-runner.ts +++ b/src/supervisor/group-runner.ts @@ -9,12 +9,12 @@ const GROUP_FAILED = new DOMException("A TaskGroup member failed", "AbortError") /** Submission metadata and policy only; every leaf belongs to the supplied owner. */ export class GroupRunner { - readonly #members = new Map, GroupBoundary>(); + readonly #members = new Map, GroupBoundary>(); - constructor(private readonly owner: Job) {} + constructor(private readonly owner: Job) {} /** Apply declaration policy before the Supervisor decides root propagation. */ - childFailed(child: Job): boolean { + childFailed(child: Job): boolean { let boundary = this.#members.get(child); while (boundary) { if (boundary.failure === "isolate") { diff --git a/src/supervisor/supervisor.ts b/src/supervisor/supervisor.ts index 530d614..0640332 100644 --- a/src/supervisor/supervisor.ts +++ b/src/supervisor/supervisor.ts @@ -12,12 +12,12 @@ export interface SupervisorOptions { type Handler = () => T | PromiseLike; /** Optional admission and failure policy for an ordinary, explicitly supplied Job. */ -export class Supervisor implements AsyncDisposable { - readonly job: Job; +export class Supervisor implements AsyncDisposable { + readonly job: Job; readonly failure: "fail-fast" | "isolate"; #groupRunner: GroupRunner | undefined; - constructor(job: Job, options: SupervisorOptions = {}) { + constructor(job: Job, options: SupervisorOptions = {}) { const failure = options.failure ?? "isolate"; if (failure !== "fail-fast" && failure !== "isolate") { throw new TypeError("Supervisor failure must be fail-fast or isolate."); @@ -31,7 +31,7 @@ export class Supervisor implements AsyncDisposable { return this.job.state; } - start(options?: JobStartOptions): Job { + start(options?: JobStartOptions): Job { if (this.job.state !== "created") { return this.job; } @@ -40,18 +40,14 @@ export class Supervisor implements AsyncDisposable { run(handler: Handler): Job; run(seed: ExecutionSeed, handler: Handler): Job; - run(job: Job): Job; + run(job: Job): Job; run( group: TaskGroup, ): Promise>; run( - input: - | ExecutionSeed - | Handler - | Job - | TaskGroup, + input: ExecutionSeed | Handler | Job | TaskGroup, handler?: Handler, - ): Job | Promise { + ): Job | Promise { if (this.job.state !== "running" || this.job.signal.aborted) { throw new LifecycleStateError("Supervisor", "run", this.state); } @@ -84,7 +80,7 @@ export class Supervisor implements AsyncDisposable { } /** @internal Decide propagation from an actual child's genuine failure. */ - childFailed(child: Job): boolean { + childFailed(child: Job): boolean { const propagate = this.#groupRunner?.childFailed(child) ?? true; return propagate && this.failure === "fail-fast"; } diff --git a/src/supervisor/task-group.ts b/src/supervisor/task-group.ts index 446bb32..64916a7 100644 --- a/src/supervisor/task-group.ts +++ b/src/supervisor/task-group.ts @@ -1,9 +1,9 @@ import type { Job } from "../job/job.js"; -export type GroupMember = Job | TaskGroup; +export type GroupMember = Job | TaskGroup; export type GroupResults = { - -readonly [Index in keyof Members]: Members[Index] extends Job + -readonly [Index in keyof Members]: Members[Index] extends Job ? Result : Members[Index] extends TaskGroup ? GroupResults diff --git a/tests/job.test.ts b/tests/job.test.ts index d1b0b7b..a2b7dd4 100644 --- a/tests/job.test.ts +++ b/tests/job.test.ts @@ -44,82 +44,6 @@ test("a fork cannot settle before its nested child", async () => { }); }); -test("a Job can publish its value before its body and descendants settle", async () => { - const releaseBody = Promise.withResolvers(); - const releaseChild = Promise.withResolvers(); - const childStarted = Promise.withResolvers(); - let settled = false; - const job = new Job(async (publish) => { - fork(async () => { - childStarted.resolve(); - await releaseChild.promise; - }); - publish(Promise.resolve("ready")); - expect(() => publish("again")).toThrowError(TypeError); - await releaseBody.promise; - return 7; - }).start(); - const completion = job.then((value) => { - settled = true; - return value; - }); - - await childStarted.promise; - expect(await job.value()).toBe("ready"); - expect(settled).toBe(false); - releaseBody.resolve(); - await nextTurn(); - expect(settled).toBe(false); - releaseChild.resolve(); - expect(await completion).toBe(7); -}); - -test("value rejects when a Job settles before publishing", async () => { - const unreported = new Job(() => 42); - const missing = unreported.value(); - unreported.start(); - await expect(missing).rejects.toBeInstanceOf(LifecycleStateError); - expect(await unreported).toBe(42); - - const failure = new Error("before publication"); - const failed = new Job(() => { - throw failure; - }); - const value = failed.value(); - failed.start(); - await expect(value).rejects.toBe(failure); -}); - -test("a cancelled Job still publishes the answer its body owes", async () => { - const childFailure = new Error("child"); - const released = Promise.withResolvers(); - const job = new Job(async (publish) => { - fork(() => { - throw childFailure; - }); - await nextTurn(); - // The child's failure has cancelled this Job; the answer is still owed. - expect(signal().aborted).toBe(true); - publish("mapped"); - await released.promise; - }).start(); - - expect(await job.value()).toBe("mapped"); - released.resolve(); - expect(await job.result()).toEqual({ ok: false, error: childFailure }); -}); - -test("a rejected publication does not replace the Job's lifetime result", async () => { - const publicationFailure = new Error("publication"); - const job = new Job((publish) => { - publish(Promise.reject(publicationFailure)); - return 42; - }).start(); - - await expect(job.value()).rejects.toBe(publicationFailure); - expect(await job).toBe(42); -}); - test("successful completion joins naturally without cancelling body or child signals", async () => { let body!: AbortSignal; let child!: AbortSignal; @@ -276,10 +200,8 @@ test("self and descendant result observations reject without replacing the owner const job = new Job(async () => { const owner = currentState().job; expect(() => owner.result()).toThrowError(LifecycleDependencyError); - expect(() => owner.value()).toThrowError(LifecycleDependencyError); return await fork(() => { expect(() => owner.result()).toThrowError(LifecycleDependencyError); - expect(() => owner.value()).toThrowError(LifecycleDependencyError); return 42; }); }); From 91ee971dbbacf2f8fe1979297aaf924e94e797a6 Mon Sep 17 00:00:00 2001 From: miinhho Date: Sat, 12 Sep 2026 16:40:44 +0900 Subject: [PATCH 2/4] feat: add HandoffJob for one-shot two-way handoff A HandoffJob's body offers one value mid-execution and suspends until the consumer resumes it. The rendezvous is state released by the Job's own cancellation and closure: no listener, settlement observer, gate Job, or queue. A body that returns without offering fails; a Job that closes without offering rejects its receiver with its failure. Supervisor.run(job) now preserves the submitted Job subtype. --- AGENTS.md | 1 + README.md | 27 +++++ src/index.ts | 2 + src/job/handoff-job.ts | 43 ++++++++ src/job/handoff.ts | 86 +++++++++++++++ src/job/job.ts | 20 +++- src/supervisor/supervisor.ts | 2 +- tests/handoff-job.test.ts | 201 +++++++++++++++++++++++++++++++++++ 8 files changed, 379 insertions(+), 3 deletions(-) create mode 100644 src/job/handoff-job.ts create mode 100644 src/job/handoff.ts create mode 100644 tests/handoff-job.test.ts diff --git a/AGENTS.md b/AGENTS.md index 037429d..d035277 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -16,6 +16,7 @@ Runner must not import server, HTTP, WebSocket, gRPC, broker, scheduler, or opti ## Public contracts - `Job` is a cold, single-use execution and awaitable lifetime node. It settles only after its body and actual descendants finish. +- `HandoffJob` is a Job whose body offers one value and suspends until the consumer resumes it. The handoff is rendezvous state released by the Job's own cancellation and closure; it owns no lifetime, signal, or queue. - `Supervisor` manages a supplied ordinary Job. It does not create hidden startup, background, or shutdown owners. - `Supervisor.start(options)` follows `Job.start(options)` ownership and context rules. Use `{ parent: undefined }` to explicitly select an independent root; Supervisor state is its Job state. - Applications own initialization order and error reporting. Do not add readiness handshakes, automatic logging, or reporting callbacks to the execution core. diff --git a/README.md b/README.md index 7e0bf5a..d6686c6 100644 --- a/README.md +++ b/README.md @@ -66,6 +66,33 @@ are preserved together in an `AggregateError`. Use native `try/finally`, `using`, or `await using` for resources. Body-local cleanup has its ordinary lexical lifetime. If a resource must outlive an entire subtree, acquire it outside the awaited Job or `execute()` call. +### Handing a value over mid-execution + +A `HandoffJob` is a Job whose body hands one value to a consumer before its work is +done and suspends until that consumer answers. The Job's lifetime is unchanged: it still +settles only after its body and descendants finish. A handoff has no queue, buffering, +repeated delivery, or hidden work; it is one value and one answer. + +```ts +import { HandoffJob } from "@tiberjs/runner"; + +const exchange = new HandoffJob(async (handoff) => { + const response = await prepare(); + const outcome = await handoff.offer(response); // suspended until resume() + await release(outcome); +}); + +exchange.start(); +const response = await exchange.receive(); // the Job is still running +exchange.resume(await deliver(response)); +await exchange; // body and descendants have settled +``` + +- `offer()` and `resume()` are each accepted once; a second call throws `TypeError`. +- Cancelling the Job while its body is suspended rejects the pending `offer()` with the cancellation reason, so an abandoned consumer cannot keep a closing Job suspended. An offer made after cancellation still reaches the consumer and rejects immediately for the body. A `resume()` after that release is discarded. +- A Job that closes without offering rejects `receive()` with its failure. A body that returns without offering fails with `TypeError`. +- `receive()` rejects self/ancestor observation synchronously, like `result()`. + ## Supervision A `Supervisor` manages an ordinary Job supplied by the caller. It directs submissions to that Job and applies failure policy. The application decides when initialization has finished and when to submit subsequent work. diff --git a/src/index.ts b/src/index.ts index 3c60473..2e4f9d6 100644 --- a/src/index.ts +++ b/src/index.ts @@ -25,5 +25,7 @@ export { export { Job } from "./job/job.js"; export type { JobStartOptions, JobState, JobResult } from "./job/job.js"; +export { HandoffJob } from "./job/handoff-job.js"; +export type { Handoff } from "./job/handoff.js"; export { TaskGroup } from "./supervisor/task-group.js"; export type { TaskGroupOptions, GroupMember, GroupResults } from "./supervisor/task-group.js"; diff --git a/src/job/handoff-job.ts b/src/job/handoff-job.ts new file mode 100644 index 0000000..edb3431 --- /dev/null +++ b/src/job/handoff-job.ts @@ -0,0 +1,43 @@ +import type { ExecutionSeed } from "../execution/context/execution-context.js"; +import { HandoffState, type Handoff } from "./handoff.js"; +import { Job } from "./job.js"; + +/** + * A Job whose body hands one value to a consumer mid-execution and suspends + * until that consumer answers, while the Job's lifetime still settles only + * after its body and descendants finish. + */ +export class HandoffJob extends Job { + private readonly rendezvous: HandoffState; + + constructor( + body: (handoff: Handoff) => Result | PromiseLike, + seed?: ExecutionSeed, + ) { + if (typeof body !== "function") { + throw new TypeError("HandoffJob requires a body."); + } + const rendezvous = new HandoffState(); + super(() => body(rendezvous), seed); + this.rendezvous = rendezvous; + this.attach(rendezvous); + } + + /** + * Observe the offered value without joining the Job's descendants. + * + * Rejects with the Job's failure when it closes without offering. + */ + receive(): Promise { + const dependency = this.dependency("receive"); + if (dependency) { + throw dependency; + } + return this.rendezvous.receive(); + } + + /** Answer the offer once. An answer after cancellation released the body is discarded. */ + resume(value: Resumed): void { + this.rendezvous.resume(value); + } +} diff --git a/src/job/handoff.ts b/src/job/handoff.ts new file mode 100644 index 0000000..4aa371a --- /dev/null +++ b/src/job/handoff.ts @@ -0,0 +1,86 @@ +import type { JobResult } from "./job.js"; + +/** The body's side of a one-shot, two-way rendezvous. */ +export interface Handoff { + /** + * Hand over the single value and suspend until the consumer resumes it. + * + * Rejects with the Job's cancellation reason when the Job is cancelled + * while suspended, or is already cancelled when the offer is made. + */ + offer(value: Offered): Promise; +} + +/** What a Job needs from its rendezvous: offer bookkeeping plus release on cancellation and closure. */ +export interface OwnedHandoff { + readonly hasOffered: boolean; + release(reason: unknown): void; + settle(result: JobResult): void; +} + +/** Rendezvous state owned by one HandoffJob: no Job, signal, queue, or hidden work. */ +export class HandoffState implements Handoff { + private readonly offered = Promise.withResolvers(); + private readonly resumed = Promise.withResolvers(); + private offering = false; + private resuming = false; + private observed = false; + private released = false; + private releaseReason: unknown; + + get hasOffered(): boolean { + return this.offering; + } + + offer(value: Offered): Promise { + if (this.offering) { + throw new TypeError("Handoff.offer() may only be called once."); + } + this.offering = true; + this.offered.resolve(value); + if (this.released) { + // The consumer still receives the answer; the body is not kept waiting for it. + this.resumed.reject(this.releaseReason); + } + return this.resumed.promise; + } + + receive(): Promise { + this.observed = true; + return this.offered.promise; + } + + resume(value: Resumed): void { + if (this.resuming) { + throw new TypeError("Handoff.resume() may only be called once."); + } + this.resuming = true; + this.resumed.resolve(value); + } + + /** Cancellation: release a body suspended in `offer()`, now or when it offers. */ + release(reason: unknown): void { + if (this.released) { + return; + } + this.released = true; + this.releaseReason = reason; + if (this.offering) { + this.resumed.reject(reason); + } + } + + /** Closure: a Job that closed without offering answers its consumer with its failure. */ + settle(result: JobResult): void { + if (this.offering) { + return; + } + this.offering = true; + this.offered.reject( + result.ok ? new TypeError("HandoffJob closed without offering a value.") : result.error, + ); + if (!this.observed) { + void this.offered.promise.catch(() => {}); + } + } +} diff --git a/src/job/job.ts b/src/job/job.ts index 757244a..a003b4e 100644 --- a/src/job/job.ts +++ b/src/job/job.ts @@ -6,6 +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"; export interface JobStartOptions { readonly parent?: Job; @@ -34,6 +35,7 @@ export class Job implements PromiseLike, AsyncDisposable { private failures: FailureSet | undefined; private cancellation: CancellationBindings | undefined; private closing: Promise | undefined; + private handoff: OwnedHandoff | undefined; constructor( private readonly body: () => T | PromiseLike, @@ -106,6 +108,14 @@ export class Job implements PromiseLike, AsyncDisposable { this.supervisor = supervisor; } + /** @internal Bind a HandoffJob's rendezvous so cancellation and closure release it. */ + attach(handoff: OwnedHandoff): void { + if (this.phase !== "created" || this.handoff) { + throw new LifecycleStateError("Job", "attach", this.phase); + } + this.handoff = handoff; + } + start(options: JobStartOptions = {}): this { if (this.phase !== "created") { throw new LifecycleStateError("Job", "start", this.phase); @@ -208,7 +218,11 @@ export class Job implements PromiseLike, AsyncDisposable { value = await runWith({ job: this, context: this.context }, async () => { try { this.signal.throwIfAborted(); - return await this.body(); + const value = await this.body(); + if (this.handoff && !this.handoff.hasOffered) { + throw new TypeError("HandoffJob body completed without offering a value."); + } + return value; } catch (error) { // Capture before a caller can cancel with this same value. this.recordFailure(error); @@ -281,6 +295,7 @@ export class Job implements PromiseLike, AsyncDisposable { continue; } job.controller.abort(received); + job.handoff?.release(job.signal.reason); for (const child of job.children ?? []) { jobs.push(child); reasons.push(job.signal.reason); @@ -288,7 +303,7 @@ export class Job implements PromiseLike, AsyncDisposable { } } - private dependency(operation: string): LifecycleDependencyError | undefined { + protected dependency(operation: string): LifecycleDependencyError | undefined { return this.owns(peekState()?.job) ? new LifecycleDependencyError("Job", operation, "Job") : undefined; @@ -400,6 +415,7 @@ export class Job implements PromiseLike, AsyncDisposable { this.cancellation?.[Symbol.dispose](); this.cancellation = undefined; this.owner?.children?.delete(this); + this.handoff?.settle(result); this.settled.resolve(result); } diff --git a/src/supervisor/supervisor.ts b/src/supervisor/supervisor.ts index 0640332..ced3225 100644 --- a/src/supervisor/supervisor.ts +++ b/src/supervisor/supervisor.ts @@ -40,7 +40,7 @@ export class Supervisor implements AsyncDisposable { run(handler: Handler): Job; run(seed: ExecutionSeed, handler: Handler): Job; - run(job: Job): Job; + run>(job: J): J; run( group: TaskGroup, ): Promise>; diff --git a/tests/handoff-job.test.ts b/tests/handoff-job.test.ts new file mode 100644 index 0000000..e41904f --- /dev/null +++ b/tests/handoff-job.test.ts @@ -0,0 +1,201 @@ +import { setImmediate as nextTurn } from "node:timers/promises"; +import { expect, test } from "vitest"; +import { + currentState, + fork, + signal, + HandoffJob, + Job, + LifecycleDependencyError, + Supervisor, +} from "../src/index.js"; + +test("the body suspends at its offer until the consumer resumes, and the Job still settles after descendants", async () => { + const releaseChild = Promise.withResolvers(); + const childStarted = Promise.withResolvers(); + const order: string[] = []; + const job = new HandoffJob(async (handoff) => { + fork(async () => { + childStarted.resolve(); + await releaseChild.promise; + order.push("child"); + }); + const answer = await handoff.offer(1); + order.push(`resumed:${answer}`); + return "done"; + }).start(); + let settled = false; + const completion = job.then((value) => { + settled = true; + return value; + }); + + expect(await job.receive()).toBe(1); + await childStarted.promise; + await nextTurn(); + expect(order).toEqual([]); + + job.resume("ack"); + await nextTurn(); + expect(order).toEqual(["resumed:ack"]); + expect(settled).toBe(false); + + releaseChild.resolve(); + expect(await completion).toBe("done"); + expect(order).toEqual(["resumed:ack", "child"]); +}); + +test("offer and resume are each accepted at most once", async () => { + const job = new HandoffJob(async (handoff) => { + const first = handoff.offer("first"); + expect(() => handoff.offer("second")).toThrowError(TypeError); + await first; + }).start(); + + expect(await job.receive()).toBe("first"); + job.resume("ack"); + expect(() => job.resume("again")).toThrowError(TypeError); + await job; +}); + +test("failure before an offer rejects the receiver with the composed Job failure", async () => { + const bodyFailure = new Error("body"); + const childFailure = new Error("child"); + const job = new HandoffJob(async () => { + fork(() => { + throw childFailure; + }); + await nextTurn(); + throw bodyFailure; + }); + const received = job.receive(); + job.start(); + + const result = await job.result(); + if (result.ok) { + throw new Error("expected the Job to fail"); + } + await expect(received).rejects.toBe(result.error); + await expect(received).rejects.toMatchObject({ errors: [childFailure, bodyFailure] }); +}); + +test("a body that completes without offering fails instead of succeeding", async () => { + const job = new HandoffJob(() => 42).start(); + + await expect(job.receive()).rejects.toBeInstanceOf(TypeError); + await expect(job).rejects.toBeInstanceOf(TypeError); +}); + +test("cancellation releases a body suspended in its offer with the original reason", async () => { + const reason = new Error("stop"); + let released: unknown; + const job = new HandoffJob(async (handoff) => { + try { + await handoff.offer("value"); + } catch (error) { + released = error; + } + return "cleaned"; + }).start(); + + expect(await job.receive()).toBe("value"); + job.cancel(reason); + await job.result(); + expect(released).toBe(reason); + // A late acknowledgement is discarded rather than rejected. + job.resume("late"); +}); + +test("an offer made after cancellation still reaches the consumer without suspending the body", async () => { + const childFailure = new Error("child"); + const job = new HandoffJob(async (handoff) => { + fork(() => { + throw childFailure; + }); + await nextTurn(); + expect(signal().aborted).toBe(true); + await expect(handoff.offer("mapped")).rejects.toBe(signal().reason); + }).start(); + + expect(await job.receive()).toBe("mapped"); + expect(await job.result()).toEqual({ ok: false, error: childFailure }); +}); + +test("an abandoned consumer cannot keep a closing Job suspended", async () => { + const job = new HandoffJob(async (handoff) => { + await handoff.offer("value").catch(() => {}); + }).start(); + expect(await job.receive()).toBe("value"); + await nextTurn(); + expect(job.state).toBe("running"); + + await job.close(); + expect(job.state).toBe("closed"); +}); + +test("owner cancellation cascades into a supervised HandoffJob's suspended offer", async () => { + const reason = new Error("shutdown"); + let released: unknown; + const owner = new Supervisor(new Job(() => untilCancelled())); + owner.start({ parent: undefined }); + const job = owner.run( + new HandoffJob(async (handoff) => { + try { + await handoff.offer("value"); + } catch (error) { + released = error; + } + }), + ); + expect(await job.receive()).toBe("value"); + + await owner.close(reason); + expect(released).toBe(reason); + expect(job.state).toBe("closed"); +}); + +test("independent delivery and cleanup failures keep their identities", async () => { + const delivery = new Error("delivery"); + const cleanup = new Error("cleanup"); + const job = new HandoffJob(async (handoff) => { + const outcome = await handoff.offer("value"); + try { + throw cleanup; + } catch (error) { + throw new AggregateError([outcome.error, error], "delivery and cleanup failed"); + } + }).start(); + + expect(await job.receive()).toBe("value"); + job.resume({ error: delivery }); + await expect(job.result()).resolves.toMatchObject({ + ok: false, + error: { errors: [delivery, cleanup] }, + }); +}); + +test("receive rejects self and descendant observation synchronously", async () => { + const job = new HandoffJob(async (handoff) => { + const owner = currentState().job as HandoffJob; + expect(() => owner.receive()).toThrowError(LifecycleDependencyError); + const value = await fork(() => { + expect(() => owner.receive()).toThrowError(LifecycleDependencyError); + return 42; + }); + await handoff.offer("ready"); + return value; + }); + const received = job.receive(); + job.start(); + expect(await received).toBe("ready"); + job.resume(); + expect(await job).toBe(42); +}); + +function untilCancelled(): Promise { + const current = signal(); + current.throwIfAborted(); + const { promise, resolve } = Promise.withResolvers(); + current.addEventListener("abort", () => resolve(), { once: true }); + return promise; +} From a5404d5a543bcf1a0422d4531bbeb1088bfe5006 Mon Sep 17 00:00:00 2001 From: miinhho Date: Sat, 12 Sep 2026 16:47:17 +0900 Subject: [PATCH 3/4] refactor: simplify handoff state and harden its tests Expose the offered flag directly, name the rendezvous sides by role, and settle a receiver only with a Job failure: a successful close without an offer is already impossible because the body fails first. Replace a tautological failure-identity test with the actual contract (a resumed answer is the body's to interpret), assert receiver and Job share one failure, and cover the unobserved-receiver path. --- src/job/handoff.ts | 63 ++++++++++++++++----------------------- src/job/job.ts | 6 ++-- tests/handoff-job.test.ts | 53 +++++++++++++++++++++----------- 3 files changed, 66 insertions(+), 56 deletions(-) diff --git a/src/job/handoff.ts b/src/job/handoff.ts index 4aa371a..daeb292 100644 --- a/src/job/handoff.ts +++ b/src/job/handoff.ts @@ -1,5 +1,3 @@ -import type { JobResult } from "./job.js"; - /** The body's side of a one-shot, two-way rendezvous. */ export interface Handoff { /** @@ -13,49 +11,45 @@ export interface Handoff { /** What a Job needs from its rendezvous: offer bookkeeping plus release on cancellation and closure. */ export interface OwnedHandoff { - readonly hasOffered: boolean; + readonly offered: boolean; release(reason: unknown): void; - settle(result: JobResult): void; + settle(failure: unknown): void; } /** Rendezvous state owned by one HandoffJob: no Job, signal, queue, or hidden work. */ -export class HandoffState implements Handoff { - private readonly offered = Promise.withResolvers(); - private readonly resumed = Promise.withResolvers(); - private offering = false; - private resuming = false; - private observed = false; +export class HandoffState implements Handoff, OwnedHandoff { + private readonly value = Promise.withResolvers(); + private readonly answer = Promise.withResolvers(); + offered = false; + private resumed = false; + private received = false; private released = false; private releaseReason: unknown; - get hasOffered(): boolean { - return this.offering; - } - offer(value: Offered): Promise { - if (this.offering) { + if (this.offered) { throw new TypeError("Handoff.offer() may only be called once."); } - this.offering = true; - this.offered.resolve(value); + this.offered = true; + this.value.resolve(value); if (this.released) { // The consumer still receives the answer; the body is not kept waiting for it. - this.resumed.reject(this.releaseReason); + this.answer.reject(this.releaseReason); } - return this.resumed.promise; + return this.answer.promise; } receive(): Promise { - this.observed = true; - return this.offered.promise; + this.received = true; + return this.value.promise; } resume(value: Resumed): void { - if (this.resuming) { + if (this.resumed) { throw new TypeError("Handoff.resume() may only be called once."); } - this.resuming = true; - this.resumed.resolve(value); + this.resumed = true; + this.answer.resolve(value); } /** Cancellation: release a body suspended in `offer()`, now or when it offers. */ @@ -65,22 +59,17 @@ export class HandoffState implements Handoff } this.released = true; this.releaseReason = reason; - if (this.offering) { - this.resumed.reject(reason); + if (this.offered) { + this.answer.reject(reason); } } - /** Closure: a Job that closed without offering answers its consumer with its failure. */ - settle(result: JobResult): void { - if (this.offering) { - return; - } - this.offering = true; - this.offered.reject( - result.ok ? new TypeError("HandoffJob closed without offering a value.") : result.error, - ); - if (!this.observed) { - void this.offered.promise.catch(() => {}); + /** Closure without an offer: the consumer receives the Job's failure instead. */ + settle(failure: unknown): void { + this.offered = true; + this.value.reject(failure); + if (!this.received) { + void this.value.promise.catch(() => {}); } } } diff --git a/src/job/job.ts b/src/job/job.ts index a003b4e..176bcaa 100644 --- a/src/job/job.ts +++ b/src/job/job.ts @@ -219,7 +219,7 @@ export class Job implements PromiseLike, AsyncDisposable { try { this.signal.throwIfAborted(); const value = await this.body(); - if (this.handoff && !this.handoff.hasOffered) { + if (this.handoff && !this.handoff.offered) { throw new TypeError("HandoffJob body completed without offering a value."); } return value; @@ -415,7 +415,9 @@ export class Job implements PromiseLike, AsyncDisposable { this.cancellation?.[Symbol.dispose](); this.cancellation = undefined; this.owner?.children?.delete(this); - this.handoff?.settle(result); + if (this.handoff && !this.handoff.offered && !result.ok) { + this.handoff.settle(result.error); + } this.settled.resolve(result); } diff --git a/tests/handoff-job.test.ts b/tests/handoff-job.test.ts index e41904f..214a8f5 100644 --- a/tests/handoff-job.test.ts +++ b/tests/handoff-job.test.ts @@ -79,11 +79,27 @@ test("failure before an offer rejects the receiver with the composed Job failure await expect(received).rejects.toMatchObject({ errors: [childFailure, bodyFailure] }); }); -test("a body that completes without offering fails instead of succeeding", async () => { +test("a body that completes without offering fails, and its receiver fails with the same error", async () => { const job = new HandoffJob(() => 42).start(); - await expect(job.receive()).rejects.toBeInstanceOf(TypeError); - await expect(job).rejects.toBeInstanceOf(TypeError); + const result = await job.result(); + if (result.ok) { + throw new Error("expected the Job to fail"); + } + expect(result.error).toBeInstanceOf(TypeError); + await expect(job.receive()).rejects.toBe(result.error); +}); + +test("a failed HandoffJob that nobody receives from does not raise an unhandled rejection", async () => { + const failure = new Error("unobserved"); + const job = new HandoffJob(() => { + throw failure; + }).start(); + + expect(await job.result()).toEqual({ ok: false, error: failure }); + await nextTurn(); + // A late receiver still learns why no value arrived. + await expect(job.receive()).rejects.toBe(failure); }); test("cancellation releases a body suspended in its offer with the original reason", async () => { @@ -100,10 +116,11 @@ test("cancellation releases a body suspended in its offer with the original reas expect(await job.receive()).toBe("value"); job.cancel(reason); - await job.result(); + expect(await job.result()).toEqual({ ok: false, error: reason }); expect(released).toBe(reason); // A late acknowledgement is discarded rather than rejected. job.resume("late"); + expect(() => job.resume("twice")).toThrowError(TypeError); }); test("an offer made after cancellation still reaches the consumer without suspending the body", async () => { @@ -154,24 +171,26 @@ test("owner cancellation cascades into a supervised HandoffJob's suspended offer expect(job.state).toBe("closed"); }); -test("independent delivery and cleanup failures keep their identities", async () => { +test("a resumed answer is the body's to interpret; only what the body raises fails the Job", async () => { const delivery = new Error("delivery"); const cleanup = new Error("cleanup"); - const job = new HandoffJob(async (handoff) => { + type Outcome = { readonly error?: unknown }; + + const tolerant = new HandoffJob(async (handoff) => { const outcome = await handoff.offer("value"); - try { - throw cleanup; - } catch (error) { - throw new AggregateError([outcome.error, error], "delivery and cleanup failed"); - } + return outcome.error === delivery ? "tolerated" : "unexpected"; }).start(); + expect(await tolerant.receive()).toBe("value"); + tolerant.resume({ error: delivery }); + expect(await tolerant.result()).toEqual({ ok: true, value: "tolerated" }); - expect(await job.receive()).toBe("value"); - job.resume({ error: delivery }); - await expect(job.result()).resolves.toMatchObject({ - ok: false, - error: { errors: [delivery, cleanup] }, - }); + const strict = new HandoffJob(async (handoff) => { + await handoff.offer("value"); + throw cleanup; + }).start(); + expect(await strict.receive()).toBe("value"); + strict.resume({ error: delivery }); + expect(await strict.result()).toEqual({ ok: false, error: cleanup }); }); test("receive rejects self and descendant observation synchronously", async () => { From 9fe2f8415a7e09e7474741aa1aeefb8335183112 Mon Sep 17 00:00:00 2001 From: miinhho Date: Sat, 12 Sep 2026 16:50:10 +0900 Subject: [PATCH 4/4] docs: trim handoff comments and README to the contract --- AGENTS.md | 2 +- README.md | 15 ++++++--------- src/job/handoff-job.ts | 13 ++----------- src/job/handoff.ts | 13 +------------ src/job/job.ts | 2 +- 5 files changed, 11 insertions(+), 34 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index d035277..9502840 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -16,7 +16,7 @@ Runner must not import server, HTTP, WebSocket, gRPC, broker, scheduler, or opti ## Public contracts - `Job` is a cold, single-use execution and awaitable lifetime node. It settles only after its body and actual descendants finish. -- `HandoffJob` is a Job whose body offers one value and suspends until the consumer resumes it. The handoff is rendezvous state released by the Job's own cancellation and closure; it owns no lifetime, signal, or queue. +- `HandoffJob` is a Job whose body offers one value and suspends until resumed. The handoff is rendezvous state released by the Job's cancellation and closure; it owns no lifetime, signal, or queue. - `Supervisor` manages a supplied ordinary Job. It does not create hidden startup, background, or shutdown owners. - `Supervisor.start(options)` follows `Job.start(options)` ownership and context rules. Use `{ parent: undefined }` to explicitly select an independent root; Supervisor state is its Job state. - Applications own initialization order and error reporting. Do not add readiness handshakes, automatic logging, or reporting callbacks to the execution core. diff --git a/README.md b/README.md index d6686c6..02817fa 100644 --- a/README.md +++ b/README.md @@ -66,12 +66,9 @@ are preserved together in an `AggregateError`. Use native `try/finally`, `using`, or `await using` for resources. Body-local cleanup has its ordinary lexical lifetime. If a resource must outlive an entire subtree, acquire it outside the awaited Job or `execute()` call. -### Handing a value over mid-execution +### HandoffJob -A `HandoffJob` is a Job whose body hands one value to a consumer before its work is -done and suspends until that consumer answers. The Job's lifetime is unchanged: it still -settles only after its body and descendants finish. A handoff has no queue, buffering, -repeated delivery, or hidden work; it is one value and one answer. +A `HandoffJob` body offers one value and suspends until the consumer resumes it. The Job still settles only after its body and descendants finish. ```ts import { HandoffJob } from "@tiberjs/runner"; @@ -88,10 +85,10 @@ exchange.resume(await deliver(response)); await exchange; // body and descendants have settled ``` -- `offer()` and `resume()` are each accepted once; a second call throws `TypeError`. -- Cancelling the Job while its body is suspended rejects the pending `offer()` with the cancellation reason, so an abandoned consumer cannot keep a closing Job suspended. An offer made after cancellation still reaches the consumer and rejects immediately for the body. A `resume()` after that release is discarded. -- A Job that closes without offering rejects `receive()` with its failure. A body that returns without offering fails with `TypeError`. -- `receive()` rejects self/ancestor observation synchronously, like `result()`. +- `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. +- A body that returns without offering fails. A Job that closes without offering rejects `receive()` with its failure. +- `receive()` rejects self/ancestor observation synchronously. ## Supervision diff --git a/src/job/handoff-job.ts b/src/job/handoff-job.ts index edb3431..af3b2d8 100644 --- a/src/job/handoff-job.ts +++ b/src/job/handoff-job.ts @@ -2,11 +2,7 @@ import type { ExecutionSeed } from "../execution/context/execution-context.js"; import { HandoffState, type Handoff } from "./handoff.js"; import { Job } from "./job.js"; -/** - * A Job whose body hands one value to a consumer mid-execution and suspends - * until that consumer answers, while the Job's lifetime still settles only - * after its body and descendants finish. - */ +/** A Job whose body offers one value mid-execution and suspends until the consumer resumes it. */ export class HandoffJob extends Job { private readonly rendezvous: HandoffState; @@ -23,11 +19,7 @@ export class HandoffJob extends Job { this.attach(rendezvous); } - /** - * Observe the offered value without joining the Job's descendants. - * - * Rejects with the Job's failure when it closes without offering. - */ + /** The offered value, or the Job's failure if it closes without offering. */ receive(): Promise { const dependency = this.dependency("receive"); if (dependency) { @@ -36,7 +28,6 @@ export class HandoffJob extends Job { return this.rendezvous.receive(); } - /** Answer the offer once. An answer after cancellation released the body is discarded. */ resume(value: Resumed): void { this.rendezvous.resume(value); } diff --git a/src/job/handoff.ts b/src/job/handoff.ts index daeb292..eeb15d4 100644 --- a/src/job/handoff.ts +++ b/src/job/handoff.ts @@ -1,22 +1,14 @@ -/** The body's side of a one-shot, two-way rendezvous. */ export interface Handoff { - /** - * Hand over the single value and suspend until the consumer resumes it. - * - * Rejects with the Job's cancellation reason when the Job is cancelled - * while suspended, or is already cancelled when the offer is made. - */ + /** Hand over one value and suspend until resumed. Rejects with the Job's cancellation reason. */ offer(value: Offered): Promise; } -/** What a Job needs from its rendezvous: offer bookkeeping plus release on cancellation and closure. */ export interface OwnedHandoff { readonly offered: boolean; release(reason: unknown): void; settle(failure: unknown): void; } -/** Rendezvous state owned by one HandoffJob: no Job, signal, queue, or hidden work. */ export class HandoffState implements Handoff, OwnedHandoff { private readonly value = Promise.withResolvers(); private readonly answer = Promise.withResolvers(); @@ -33,7 +25,6 @@ export class HandoffState implements Handoff this.offered = true; this.value.resolve(value); if (this.released) { - // The consumer still receives the answer; the body is not kept waiting for it. this.answer.reject(this.releaseReason); } return this.answer.promise; @@ -52,7 +43,6 @@ export class HandoffState implements Handoff this.answer.resolve(value); } - /** Cancellation: release a body suspended in `offer()`, now or when it offers. */ release(reason: unknown): void { if (this.released) { return; @@ -64,7 +54,6 @@ export class HandoffState implements Handoff } } - /** Closure without an offer: the consumer receives the Job's failure instead. */ settle(failure: unknown): void { this.offered = true; this.value.reject(failure); diff --git a/src/job/job.ts b/src/job/job.ts index 176bcaa..641a4a9 100644 --- a/src/job/job.ts +++ b/src/job/job.ts @@ -108,7 +108,7 @@ export class Job implements PromiseLike, AsyncDisposable { this.supervisor = supervisor; } - /** @internal Bind a HandoffJob's rendezvous so cancellation and closure release it. */ + /** @internal Bind a HandoffJob's rendezvous before activation. */ attach(handoff: OwnedHandoff): void { if (this.phase !== "created" || this.handoff) { throw new LifecycleStateError("Job", "attach", this.phase);