From 3f1f26c0cca3c1434d4b12a05c10f79ef5d9c431 Mon Sep 17 00:00:00 2001 From: dKosarevsky <36589322+dKosarevsky@users.noreply.github.com> Date: Fri, 9 Oct 2026 00:55:54 -0700 Subject: [PATCH] Add AbortSignal cancellation for checks and fixture replay --- CHANGELOG.md | 6 + docs/API.md | 18 +++ docs/SAFETY.md | 2 + scripts/test-package.mjs | 15 ++- src/errors.ts | 5 + src/index.ts | 67 ++++++++--- src/types.ts | 2 + test/cancellation.test.mjs | 233 +++++++++++++++++++++++++++++++++++++ test/types.ts | 11 +- 9 files changed, 341 insertions(+), 18 deletions(-) create mode 100644 test/cancellation.test.mjs diff --git a/CHANGELOG.md b/CHANGELOG.md index bd30d9c..2427bf7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,12 @@ Version headings describe repository changes. Check [npm](https://www.npmjs.com/package/@agent-axiom/streamsplit) for published versions. +## Unreleased + +- Add optional `AbortSignal` cancellation to checking, assertions, and exact fixture replay, with a payload-redacted `CheckAbortedError`. +- Stop pending harness work without claiming to terminate the parser; clean up listeners and handle late emissions/rejections. +- Keep the version-1 replay format and zero-runtime-dependency contract unchanged. + ## 0.1.1 Documentation, examples, and release-tooling update. The public API, parser harness implementation, Node >=20 requirement, and zero-runtime-dependency contract are unchanged. diff --git a/docs/API.md b/docs/API.md index 33dd7f0..7474afd 100644 --- a/docs/API.md +++ b/docs/API.md @@ -19,6 +19,7 @@ All offsets and sizes are bytes. `createParser(emit)` must return a new object w - `ConfigurationError`: invalid options, factory contract, schedules, or replay schema. - `LimitExceededError`: input/output/execution budget exceeded. - `ParserTimeoutError`: elapsed per-run deadline exceeded. +- `CheckAbortedError`: the supplied `AbortSignal` was aborted; cancellation never returns a partial pass or failure. - `InvalidEventError`: unsupported output or normalization failure. These extend `StreamSplitError`. Keep harness failures separate from parser failures. All public types ship in the package declarations. @@ -62,6 +63,7 @@ For deliberately invalid fixtures, opt into `errorPolicy: 'compare'`. Outcomes m | `maxEvents` | `10000` | Emitted events per run | | `maxOutputCharacters` | `1048576` | Canonical JSON characters per run, including normalized errors | | `timeoutMs` | `2000` | Elapsed deadline for each parser run | +| `signal` | none | Optional `AbortSignal` to cancel checking or exact fixture replay | A schedule's nonnegative integer sizes must sum to the exact byte length. Sizes of zero call `write(new Uint8Array())`. Empty input gets one empty baseline write. Custom and generated schedules are deduplicated, and the baseline is not retested as a candidate. @@ -69,3 +71,19 @@ Coverage reports the seed, bytes, schedules checked, parser runs/calls, number o `generateSchedules(inputLength, options)` exposes the deterministic schedule generator for inspection. Generated partitions vary small-read and broader-read scales; they are not a uniform sample of all possible partitions. Empty-chunk testing is a documented stress case; set `emptyChunks: false` if your adapter deliberately excludes it. +## Cancellation + +Pass an `AbortSignal` to any asynchronous entry point to stop a check, assertion, or replay when its caller cancels: + +```ts +const controller = new AbortController(); +const pending = checkChunkInvariant({ input, createParser, signal: controller.signal }); +// For example, call this from your test runner's cancellation handler: +controller.abort(); +await pending; // rejects with CheckAbortedError +``` + +A pre-aborted signal rejects before the parser factory runs. A live signal can stop pending factory/write/end operations, schedule checking, stability checks, and reduction. Cancellation also wins if a parser or normalization hook aborts the signal before returning or throwing. It is a harness error even under `errorPolicy: 'compare'`; no partial result is returned. The error does not copy the signal's reason or attach it as a cause, because it may contain private data. + +Cancellation stops the harness from starting further parser calls. It does not forcibly terminate pending parser work, interrupt blocked synchronous JavaScript, or defeat microtask starvation. Give the same signal to your adapter's own cancellable I/O when needed. Later emissions are ignored, and later promise rejections remain handled. StreamSplit removes its own abort listeners after each lifecycle call, without removing caller listeners. Signals are not saved in replay fixtures; pass a new signal when replaying. + diff --git a/docs/SAFETY.md b/docs/SAFETY.md index 6787629..36673bb 100644 --- a/docs/SAFETY.md +++ b/docs/SAFETY.md @@ -6,6 +6,8 @@ A timeout can reject a pending asynchronous operation and stops the check. It cannot forcibly cancel that operation, interrupt synchronously blocked JavaScript, or defeat microtask starvation. A synchronous operation that eventually returns or throws after the deadline is reported as a timeout. Use worker/process isolation for untrusted or potentially nonterminating parsers. +An optional `AbortSignal` stops checking or replay with `CheckAbortedError`, including while awaiting a parser operation. It has the same isolation limits: pending parser work is not forcibly cancelled, and synchronous work or microtask starvation can delay an external cancellation handler. The abort reason is never included in the error or fixture. See [cancellation](API.md#cancellation) for the complete contract. + The factory, parser methods, and normalization hooks are caller-provided trusted code. StreamSplit is not a sandbox. Bounds constrain the harness; they do not bound arbitrary allocations or work inside the parser. A fixed seed reproduces schedules, not external I/O, random application state, or time-dependent behavior. There is no telemetry, network access, file access, dynamic module loading, or logging in the runtime library. It runs only the parser and hooks supplied by the caller. **Fixtures contain exact input bytes, and failure objects contain emitted values/errors.** They can hold secrets. Keep sensitive fixtures out of public issue reports and CI logs. The default assertion message omits payload values; inspect failure objects intentionally. diff --git a/scripts/test-package.mjs b/scripts/test-package.mjs index 29c21ed..06b37bc 100644 --- a/scripts/test-package.mjs +++ b/scripts/test-package.mjs @@ -18,7 +18,8 @@ try { assert.equal(files.some(file => /^(test|examples|scripts|node_modules|\.github)\//.test(file.path)), false); writeFileSync(join(directory, 'package.json'), JSON.stringify({ private: true, type: 'module' })); execFileSync('npm', ['install', '--offline', '--ignore-scripts', '--no-audit', '--no-fund', join(directory, filename)], { cwd: directory, env, stdio: 'pipe' }); - const consumer = `import { assertChunkInvariant, checkChunkInvariant, replayChunkInvariant } from '@agent-axiom/streamsplit'; + const consumer = `import { assertChunkInvariant, checkChunkInvariant, replayChunkInvariant, CheckAbortedError } from '@agent-axiom/streamsplit'; +import assert from 'node:assert/strict'; const input = new TextEncoder().encode('package consumer'); const coverage = await assertChunkInvariant({ input, createParser: emit => { const parts = []; return { write(chunk) { parts.push(...chunk); }, end() { emit(parts); } }; @@ -29,13 +30,19 @@ const result = await checkChunkInvariant({ input, createParser: broken }); if (result.ok) throw Error('Broken parser passed'); const replay = await replayChunkInvariant({ fixture: result.failure.fixture, createParser: broken }); if (replay.ok) throw Error('Replay did not reproduce'); -console.log('Packed ESM consumer and replay passed.');\n`; +await assert.rejects(checkChunkInvariant({ input, createParser: broken, signal: AbortSignal.abort('private') }), CheckAbortedError); +await assert.rejects(replayChunkInvariant({ fixture: result.failure.fixture, createParser: broken, signal: AbortSignal.abort() }), CheckAbortedError); +console.log('Packed ESM consumer, replay, and cancellation passed.');\n`; writeFileSync(join(directory, 'consumer.mjs'), consumer); execFileSync(process.execPath, ['consumer.mjs'], { cwd: directory, stdio: 'inherit' }); - writeFileSync(join(directory, 'consumer.mts'), `import { assertChunkInvariant, type ParserFactory } from '@agent-axiom/streamsplit'; + writeFileSync(join(directory, 'consumer.mts'), `import { assertChunkInvariant, replayChunkInvariant, CheckAbortedError, StreamSplitError, type ParserFactory, type ReplayOptions } from '@agent-axiom/streamsplit'; const createParser: ParserFactory = emit => ({ write(chunk) { emit(chunk.length); } }); -const coverage = await assertChunkInvariant({ input: new Uint8Array(), createParser }); +const signal = new AbortController().signal; +const coverage = await assertChunkInvariant({ input: new Uint8Array(), createParser, signal }); coverage.parserRuns satisfies number; +const replay: ReplayOptions = { fixture: { version: 1, inputHex: '', chunkSizes: [0], seed: 0, errorPolicy: 'reject' }, createParser, signal }; +await replayChunkInvariant(replay); +new CheckAbortedError() satisfies StreamSplitError; // @ts-expect-error Text input is not a byte fixture. await assertChunkInvariant({ input: 'no', createParser });\n`); execFileSync(process.execPath, [join(root, 'node_modules/typescript/bin/tsc'), '--noEmit', '--strict', '--target', 'ES2022', '--module', 'NodeNext', '--moduleResolution', 'NodeNext', 'consumer.mts'], { cwd: directory, stdio: 'inherit' }); diff --git a/src/errors.ts b/src/errors.ts index 8555f31..3e86217 100644 --- a/src/errors.ts +++ b/src/errors.ts @@ -9,6 +9,11 @@ export class StreamSplitError extends Error { export class ConfigurationError extends StreamSplitError {} export class LimitExceededError extends StreamSplitError {} export class ParserTimeoutError extends StreamSplitError {} +export class CheckAbortedError extends StreamSplitError { + constructor() { + super('StreamSplit check was aborted. Pending parser work is not forcibly cancelled.'); + } +} export class InvalidEventError extends StreamSplitError {} export class NonDeterministicParserError extends StreamSplitError { constructor(readonly chunkSizes: readonly number[]) { diff --git a/src/index.ts b/src/index.ts index 41a0a34..5280406 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,4 +1,4 @@ -import { BaselineParserError, ChunkInvariantError, ConfigurationError, InvalidEventError, LimitExceededError, NonDeterministicParserError, ParserTimeoutError, StreamSplitError } from './errors.js'; +import { BaselineParserError, CheckAbortedError, ChunkInvariantError, ConfigurationError, InvalidEventError, LimitExceededError, NonDeterministicParserError, ParserTimeoutError, StreamSplitError } from './errors.js'; import { snapshot } from './json.js'; import { cutsToSizes, generateSchedules, integer, settings, sizesToCuts, validateSizes } from './schedules.js'; import type { CheckOptions, CheckResult, Coverage, JsonValue, Outcome, ReproFixture, Schedule } from './types.js'; @@ -17,6 +17,10 @@ const same = (a: Recorded, b: Recorded): boolean => a.key === b.key; export async function checkChunkInvariant(options: CheckOptions): Promise { if (!options || !(options.input instanceof Uint8Array)) throw new ConfigurationError('input must be a Uint8Array.'); if (typeof options.createParser !== 'function') throw new ConfigurationError('createParser must be a function.'); + const signal = options.signal; + if (signal !== undefined && !(signal instanceof AbortSignal)) throw new ConfigurationError('signal must be an AbortSignal.'); + const throwIfAborted = (): void => { if (signal?.aborted) throw new CheckAbortedError(); }; + throwIfAborted(); const config = settings(options); const maxInputBytes = integer('maxInputBytes', options.maxInputBytes ?? 1048576, 0, 0x7fffffff); const maxRuns = integer('maxRuns', options.maxRuns ?? 512, 2, 1000000); @@ -40,6 +44,7 @@ export async function checkChunkInvariant(options: CheckOptions): Promise< allSingleCutsChecked: input.length <= 1, bytewiseChecked: input.length <= 1, }; const run = async (sizes: readonly number[]): Promise => { + throwIfAborted(); if (coverage.parserRuns >= maxRuns) throw new LimitExceededError('maxRuns exhausted before the check completed. Increase maxRuns or reduce the requested schedules.'); coverage.parserRuns++; const events: JsonValue[] = []; @@ -49,43 +54,59 @@ export async function checkChunkInvariant(options: CheckOptions): Promise< let emitFailure: StreamSplitError | undefined; const deadline = performance.now() + timeoutMs; const emit = (event: T): void => { - if (!active) return; // No unhandled late-emission error after a rejected asynchronous operation. + if (!active || signal?.aborted) return; // Ignore late emissions, including before the abort race settles. try { if (events.length >= maxEvents) throw new LimitExceededError('Parser emitted more than maxEvents.'); let normalized: unknown = event; if (options.normalizeEvent) { try { normalized = options.normalizeEvent(event); } - catch (cause) { throw new InvalidEventError('normalizeEvent threw an error.', { cause }); } + catch (cause) { if (signal?.aborted) return; throw new InvalidEventError('normalizeEvent threw an error.', { cause }); } } + if (signal?.aborted) return; const copy = snapshot(normalized, maxOutputCharacters - characters); characters += copy.key.length; events.push(copy.value); keys.push(copy.key); } catch (error) { + if (signal?.aborted) return; // The call-level race reports cancellation without throwing from a late callback. emitFailure = error instanceof StreamSplitError ? error : new InvalidEventError('Could not snapshot the emitted event.', { cause: error }); throw emitFailure; } }; const call = async (operation: () => R | PromiseLike): Promise => { + throwIfAborted(); if (coverage.parserCalls >= maxCalls) throw new LimitExceededError('maxCalls exhausted before the check completed.'); coverage.parserCalls++; const left = deadline - performance.now(); if (left <= 0) throw new ParserTimeoutError('Parser run exceeded timeoutMs.'); let timer: ReturnType | undefined; + let onAbort: (() => void) | undefined; try { - const result = await Promise.race([ + const pending: Promise[] = [ new Promise((_resolve, reject) => { timer = setTimeout(() => reject(new ParserTimeoutError('Parser run exceeded timeoutMs. Pending parser work is not forcibly cancelled.')), Math.ceil(left)); }), - Promise.resolve().then(operation), - ]); + Promise.resolve().then(() => { throwIfAborted(); return operation(); }), + ]; + if (signal) pending.push(new Promise((_resolve, reject) => { + onAbort = () => reject(new CheckAbortedError()); + signal.addEventListener('abort', onAbort, { once: true }); + // Also guard against a signal aborted before listener registration. + if (signal.aborted) onAbort(); + })); + const result = await Promise.race(pending); + throwIfAborted(); if (performance.now() > deadline) throw new ParserTimeoutError('Parser run exceeded timeoutMs. Synchronous JavaScript cannot be interrupted.'); if (emitFailure) throw emitFailure; return result; } catch (error) { + throwIfAborted(); if (emitFailure) throw emitFailure; if (performance.now() > deadline && !(error instanceof ParserTimeoutError)) throw new ParserTimeoutError('Parser run exceeded timeoutMs. Synchronous JavaScript cannot be interrupted.'); throw error; - } finally { if (timer !== undefined) clearTimeout(timer); } + } finally { + if (timer !== undefined) clearTimeout(timer); + if (onAbort) signal!.removeEventListener('abort', onAbort); + } }; let failure: JsonValue | null = null; let failed = false; @@ -100,27 +121,34 @@ export async function checkChunkInvariant(options: CheckOptions): Promise< } if (parser.end) await call(() => parser.end!()); } catch (error) { + throwIfAborted(); if (emitFailure) throw emitFailure; if (error instanceof StreamSplitError) throw error; failed = true; let normalized: unknown; try { normalized = (options.normalizeError ?? defaultError)(error); } - catch (cause) { throw new InvalidEventError('normalizeError threw an error.', { cause }); } + catch (cause) { throwIfAborted(); throw new InvalidEventError('normalizeError threw an error.', { cause }); } + throwIfAborted(); failure = snapshot(normalized, maxOutputCharacters - characters).value; + throwIfAborted(); if (performance.now() > deadline) throw new ParserTimeoutError('Parser run exceeded timeoutMs during error normalization.'); - } finally { active = false; } + } finally { active = false; throwIfAborted(); } const outcome: Outcome = { events, error: failure, failed }; return { outcome, key: JSON.stringify([keys, failed, failure]) }; }; const stable = async (sizes: readonly number[], initial?: Recorded): Promise => { const first = initial ?? await run(sizes); + throwIfAborted(); for (let repeat = 1; repeat < stabilityRuns; repeat++) { - if (!same(first, await run(sizes))) throw new NonDeterministicParserError(sizes); + const repeated = await run(sizes); + throwIfAborted(); + if (!same(first, repeated)) throw new NonDeterministicParserError(sizes); } return first; }; const baselineSizes = [input.length]; const baseline = await stable(baselineSizes); + throwIfAborted(); if (baseline.outcome.failed && errorPolicy === 'reject') throw new BaselineParserError(baseline.outcome); const seen = new Set([String(input.length)]); function* schedules(): Generator { @@ -132,13 +160,17 @@ export async function checkChunkInvariant(options: CheckOptions): Promise< if (seen.has(scheduleKey)) continue; seen.add(scheduleKey); const actual = await run(schedule.chunkSizes); + throwIfAborted(); coverage.schedulesChecked++; if (schedule.chunkSizes.length === 2 && schedule.chunkSizes.every(size => size > 0)) coverage.singleCutsChecked++; if (schedule.chunkSizes.length === input.length && schedule.chunkSizes.every(size => size === 1)) coverage.bytewiseChecked = true; coverage.allSingleCutsChecked = coverage.singleCutsChecked === Math.max(0, input.length - 1); if (same(baseline, actual)) continue; await stable(schedule.chunkSizes, actual); - if (!same(baseline, await run(baselineSizes))) throw new NonDeterministicParserError(baselineSizes); + throwIfAborted(); + const rechecked = await run(baselineSizes); + throwIfAborted(); + if (!same(baseline, rechecked)) throw new NonDeterministicParserError(baselineSizes); let best = actual; let cuts = sizesToCuts(schedule.chunkSizes); let granularity = 2; @@ -154,10 +186,12 @@ export async function checkChunkInvariant(options: CheckOptions): Promise< // Reserve enough lifecycle calls for this attempt and its stability check. if (coverage.parserCalls + stabilityRuns * (sizes.length + 2) > maxCalls) break reduction; const candidate = await run(sizes); shrinkRuns++; + throwIfAborted(); if (!same(baseline, candidate)) { if (candidateCuts.length === 0) throw new NonDeterministicParserError(baselineSizes); for (let repeat = 1; repeat < stabilityRuns; repeat++) { const repeated = await run(sizes); shrinkRuns++; + throwIfAborted(); if (!same(candidate, repeated)) throw new NonDeterministicParserError(sizes); } cuts = candidateCuts; best = candidate; granularity = Math.max(2, granularity - 1); @@ -178,27 +212,32 @@ export async function checkChunkInvariant(options: CheckOptions): Promise< inputHex: Array.from(input, byte => byte.toString(16).padStart(2, '0')).join(''), chunkSizes: cutsToSizes(cuts, input.length), seed: config.seed, errorPolicy, }; + throwIfAborted(); return { ok: false, coverage, failure: { baseline: baseline.outcome, actual: best.outcome, firstDifferentEvent, originalChunkSizes: schedule.chunkSizes, fixture, shrink: { runs: shrinkRuns, complete }, } }; } + throwIfAborted(); return { ok: true, coverage }; } /** Throw a typed, payload-redacted assertion error on a confirmed chunk-dependent outcome. */ export async function assertChunkInvariant(options: CheckOptions): Promise { + const signal = options?.signal; const result = await checkChunkInvariant(options); + if (signal?.aborted) throw new CheckAbortedError(); if (!result.ok) throw new ChunkInvariantError(result.failure, result.coverage); return result.coverage; } -export interface ReplayOptions extends Pick, 'createParser' | 'normalizeEvent' | 'normalizeError' | 'timeoutMs' | 'stabilityRuns' | 'maxInputBytes' | 'maxEvents' | 'maxOutputCharacters' | 'maxCalls' | 'maxRuns'> { +export interface ReplayOptions extends Pick, 'createParser' | 'normalizeEvent' | 'normalizeError' | 'signal' | 'timeoutMs' | 'stabilityRuns' | 'maxInputBytes' | 'maxEvents' | 'maxOutputCharacters' | 'maxCalls' | 'maxRuns'> { fixture: ReproFixture; } /** Replay the exact bytes/schedule; a failure result confirms reproduction. Reuse your normalization hooks. */ export async function replayChunkInvariant(options: ReplayOptions): Promise { + const signal = options.signal; const fixture = options.fixture; if (!fixture || fixture.version !== 1 || typeof fixture.inputHex !== 'string' || !/^(?:[0-9a-f]{2})*$/i.test(fixture.inputHex)) throw new ConfigurationError('Invalid version-1 StreamSplit fixture.'); integer('fixture.seed', fixture.seed, 0, 0xffffffff); @@ -209,8 +248,10 @@ export async function replayChunkInvariant(options: ReplayOptions): Promis const sizes = validateSizes(fixture.chunkSizes, byteLength, 65536); const input = new Uint8Array(byteLength); for (let index = 0; index < byteLength; index++) input[index] = Number.parseInt(fixture.inputHex.slice(index * 2, index * 2 + 2), 16); - return checkChunkInvariant({ ...options, input, seed: fixture.seed, errorPolicy: fixture.errorPolicy, + const result = await checkChunkInvariant({ ...options, input, seed: fixture.seed, errorPolicy: fixture.errorPolicy, schedules: [sizes], maxSingleCuts: 0, randomCases: 0, bytewise: false, emptyChunks: false, maxChunkCount: Math.max(1, sizes.length), maxShrinkRuns: 0, }); + if (signal?.aborted) throw new CheckAbortedError(); + return result; } diff --git a/src/types.ts b/src/types.ts index 962e44e..ce8645a 100644 --- a/src/types.ts +++ b/src/types.ts @@ -26,6 +26,8 @@ export interface CheckOptions extends ScheduleOptions { normalizeEvent?: (event: T) => JsonValue; /** Used with errorPolicy: 'compare'; defaults to name/message, excluding stack. */ normalizeError?: (error: unknown) => JsonValue; + /** Stop checking/replay on cancellation. Does not forcibly cancel pending parser work. */ + signal?: AbortSignal; maxInputBytes?: number; maxRuns?: number; maxCalls?: number; diff --git a/test/cancellation.test.mjs b/test/cancellation.test.mjs new file mode 100644 index 0000000..20d6d57 --- /dev/null +++ b/test/cancellation.test.mjs @@ -0,0 +1,233 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { getEventListeners } from 'node:events'; +import { + assertChunkInvariant, checkChunkInvariant, replayChunkInvariant, + CheckAbortedError, ConfigurationError, ParserTimeoutError, StreamSplitError, +} from '../dist/index.js'; + +const input = new Uint8Array([1, 2, 3, 4]); +const quiet = () => ({ write() {} }); +const fixture = { version: 1, inputHex: '01020304', chunkSizes: [2, 2], seed: 0, errorPolicy: 'compare' }; +const compact = { randomCases: 0, maxSingleCuts: 0, bytewise: false, emptyChunks: false }; +const deferred = () => { + let resolve, reject; + const promise = new Promise((yes, no) => { resolve = yes; reject = no; }); + return { promise, resolve, reject }; +}; + +test('pre-aborted check, assertion and replay do not invoke the parser or expose the abort reason', async () => { + const reason = { secret: 'private-cancellation-reason', toString() { throw Error('must not inspect reason'); } }; + const signal = AbortSignal.abort(reason); + let created = 0; + const createParser = () => { created++; return quiet(); }; + for (const operation of [ + () => checkChunkInvariant({ input, createParser, signal }), + () => assertChunkInvariant({ input, createParser, signal }), + () => replayChunkInvariant({ fixture, createParser, signal }), + ]) { + await assert.rejects(operation(), error => { + assert.ok(error instanceof CheckAbortedError); + assert.ok(error instanceof StreamSplitError); + assert.equal(error.name, 'CheckAbortedError'); + assert.equal('cause' in error, false); + assert.equal('reason' in error, false); + assert.doesNotMatch(error.message, /private-cancellation-reason/); + return true; + }); + } + assert.equal(created, 0); +}); + +test('invalid cancellation signals reject before parser execution', async () => { + for (const signal of [null, false, {}, { aborted: false }, new AbortController()]) { + let created = false; + await assert.rejects(checkChunkInvariant({ input, signal, createParser() { created = true; return quiet(); } }), ConfigurationError); + assert.equal(created, false); + } +}); + +test('cancels pending factory, write and end without starting another lifecycle call', async () => { + for (const stage of ['factory', 'write', 'end']) { + const controller = new AbortController(); + const started = deferred(); + const pending = deferred(); + const calls = []; + const operation = name => { calls.push(name); if (name === stage) { started.resolve(); return pending.promise; } }; + const result = checkChunkInvariant({ input, signal: controller.signal, errorPolicy: 'compare', createParser() { + if (stage === 'factory') return operation('factory'); + operation('factory'); + return { write() { return operation('write'); }, end() { return operation('end'); } }; + } }); + await started.promise; + assert.equal(getEventListeners(controller.signal, 'abort').length, 1); + controller.abort(); + await assert.rejects(result, CheckAbortedError); + assert.deepEqual(calls, ['factory', 'write', 'end'].slice(0, ['factory', 'write', 'end'].indexOf(stage) + 1)); + assert.equal(getEventListeners(controller.signal, 'abort').length, 0); + // Settle abandoned work after rejection to verify the harness already detached. + pending.resolve(quiet()); + } +}); + +test('abort before the queued factory microtask prevents parser execution', async () => { + const controller = new AbortController(); + let created = 0; + const result = checkChunkInvariant({ input, signal: controller.signal, createParser() { created++; return quiet(); } }); + controller.abort(); + await assert.rejects(result, CheckAbortedError); + assert.equal(created, 0); + assert.equal(getEventListeners(controller.signal, 'abort').length, 0); +}); + +test('synchronous lifecycle aborts cannot return success or become comparable parser failures', async () => { + for (const stage of ['factory', 'write', 'end']) { + for (const throws of [false, true]) { + const controller = new AbortController(); + let created = 0; + const operation = name => { if (stage === name) { controller.abort(); if (throws) throw Error('parser error'); } }; + await assert.rejects(checkChunkInvariant({ input, signal: controller.signal, errorPolicy: 'compare', createParser() { + created++; + operation('factory'); + return { write() { operation('write'); }, end() { operation('end'); } }; + } }), CheckAbortedError); + assert.equal(created, 1); + assert.equal(getEventListeners(controller.signal, 'abort').length, 0); + } + } +}); + +test('cancellation from event or error normalization stays a harness cancellation', async () => { + for (const hook of ['normalizeEvent', 'normalizeError']) { + for (const throws of [false, true]) { + const controller = new AbortController(); + const options = { input, signal: controller.signal, errorPolicy: 'compare', createParser: emit => ({ write() { + if (hook === 'normalizeError') throw Error('parser error'); + try { emit('event'); } catch {} // A parser swallowing emit errors must not hide cancellation. + } }), [hook]() { + controller.abort(); + if (throws) throw Error('normalizer error'); + return null; + } }; + await assert.rejects(checkChunkInvariant(options), CheckAbortedError); + assert.equal(getEventListeners(controller.signal, 'abort').length, 0); + } + } +}); + +test('cancellation stops boundary reduction instead of returning a partial failure', async () => { + const controller = new AbortController(); + let created = 0; + await assert.rejects(checkChunkInvariant({ input, ...compact, schedules: [[1, 1, 1, 1]], signal: controller.signal, createParser: emit => { + created++; + // Two baselines, two failing-schedule checks, baseline recheck, then reduction. + if (created === 6) controller.abort(); + return { write(chunk) { emit(chunk.length); } }; + } }), CheckAbortedError); + assert.equal(created, 6); + assert.equal(getEventListeners(controller.signal, 'abort').length, 0); +}); + +test('replay forwards a live signal and cancels its exact candidate schedule', async () => { + const controller = new AbortController(); + let created = 0; + await assert.rejects(replayChunkInvariant({ fixture, signal: controller.signal, createParser() { + created++; + return { write(chunk) { if (chunk.length === 2) controller.abort(); } }; + } }), CheckAbortedError); + assert.equal(created, 3); + assert.equal(getEventListeners(controller.signal, 'abort').length, 0); +}); + +test('late parser emissions and rejection after cancellation are safely ignored or consumed', async () => { + const controller = new AbortController(); + const started = deferred(); + const pending = deferred(); + let lateEmit; + const result = checkChunkInvariant({ input, signal: controller.signal, createParser(emit) { + lateEmit = emit; + return { write() { started.resolve(); return pending.promise; } }; + } }); + await started.promise; + controller.abort(); + assert.doesNotThrow(() => lateEmit(undefined)); // The abort rejection has not unwound the run yet. + await assert.rejects(result, CheckAbortedError); + assert.doesNotThrow(() => lateEmit(undefined)); + pending.reject(Error('late parser rejection')); + await new Promise(resolve => setImmediate(resolve)); + assert.equal(getEventListeners(controller.signal, 'abort').length, 0); +}); + +test('live signals preserve success and fixtures, with listeners cleaned on every exit', async () => { + const controller = new AbortController(); + const listener = () => {}; + controller.signal.addEventListener('abort', listener); + const options = { input, createParser: quiet }; + assert.deepEqual(await checkChunkInvariant({ ...options, signal: controller.signal }), await checkChunkInvariant(options)); + const broken = { input, createParser: emit => ({ write(chunk) { emit(chunk.length); } }) }; + assert.deepEqual(await checkChunkInvariant({ ...broken, signal: controller.signal }), await checkChunkInvariant(broken)); + await assert.rejects(checkChunkInvariant({ input, signal: controller.signal, createParser: () => null }), ConfigurationError); + await assert.rejects(checkChunkInvariant({ input, signal: controller.signal, timeoutMs: 10, createParser: () => new Promise(() => {}) }), ParserTimeoutError); + assert.deepEqual(getEventListeners(controller.signal, 'abort'), [listener]); + controller.signal.removeEventListener('abort', listener); +}); + +test('normalizer cancellation does not throw from an asynchronous emitter callback', async () => { + for (const throws of [false, true]) { + const controller = new AbortController(); + const started = deferred(); + const pending = deferred(); + let emitEvent; + const result = checkChunkInvariant({ input, signal: controller.signal, createParser(emit) { + emitEvent = emit; + return { write() { started.resolve(); return pending.promise; } }; + }, normalizeEvent() { + controller.abort(); + if (throws) throw Error('private normalizer details'); + return null; + } }); + await started.promise; + assert.doesNotThrow(() => emitEvent('event')); + await assert.rejects(result, error => { + assert.ok(error instanceof CheckAbortedError); + assert.equal('cause' in error, false); + assert.doesNotMatch(JSON.stringify(error), /private normalizer details/); + return true; + }); + pending.resolve(); + assert.equal(getEventListeners(controller.signal, 'abort').length, 0); + } +}); + +test('cancellation during final baseline microtasks cannot return a false pass', async () => { + for (const depth of [3, 4]) { + const controller = new AbortController(); + let runs = 0; + await assert.rejects(checkChunkInvariant({ input, ...compact, signal: controller.signal, createParser() { + runs++; + return { write() {}, end() { + if (runs === 2) { + let remaining = depth; + const tick = () => { if (remaining-- === 0) controller.abort(); else queueMicrotask(tick); }; + queueMicrotask(tick); + } + } }; + } }), CheckAbortedError); + assert.equal(runs, 2); + assert.equal(getEventListeners(controller.signal, 'abort').length, 0); + } +}); + +test('error snapshot cancellation takes priority without leaking a thrown normalization error', async () => { + const controller = new AbortController(); + await assert.rejects(checkChunkInvariant({ input, signal: controller.signal, errorPolicy: 'compare', + createParser: () => ({ write() { throw Error('parser error'); } }), + normalizeError: () => new Proxy({}, { ownKeys() { controller.abort(); throw Error('private snapshot details'); } }), + }), error => { + assert.ok(error instanceof CheckAbortedError); + assert.equal('cause' in error, false); + assert.doesNotMatch(error.message, /private snapshot details/); + return true; + }); + assert.equal(getEventListeners(controller.signal, 'abort').length, 0); +}); diff --git a/test/types.ts b/test/types.ts index 486efb6..83c7929 100644 --- a/test/types.ts +++ b/test/types.ts @@ -1,6 +1,6 @@ import { assertChunkInvariant, checkChunkInvariant, replayChunkInvariant, generateSchedules, - ChunkInvariantError, type ByteParser, type ParserFactory, type ReproFixture, + CheckAbortedError, StreamSplitError, ChunkInvariantError, type ByteParser, type ParserFactory, type ReproFixture, } from '../dist/index.js'; const parser: ParserFactory<{ text: string }> = emit => ({ @@ -11,9 +11,14 @@ const input = new Uint8Array([1]); const coverage = await assertChunkInvariant({ input, createParser: parser }); coverage.parserRuns satisfies number; const result = await checkChunkInvariant({ input, createParser: parser }); +const signal = new AbortController().signal; +await assertChunkInvariant({ input, createParser: parser, signal }); +await checkChunkInvariant({ input, createParser: parser, signal }); +new CheckAbortedError() satisfies StreamSplitError; if (!result.ok) { const fixture: ReproFixture = result.failure.fixture; await replayChunkInvariant({ fixture, createParser: parser }); + await replayChunkInvariant({ fixture, createParser: parser, signal }); new ChunkInvariantError(result.failure, result.coverage); } await assertChunkInvariant({ input, createParser: (emit: (value: Date) => void): ByteParser => ({ @@ -27,3 +32,7 @@ const invalid: ByteParser = { end() {} }; void invalid; // @ts-expect-error Error comparison policy is explicit. await checkChunkInvariant({ input, createParser: parser, errorPolicy: 'ignore' }); +// @ts-expect-error Supply the controller's signal, not the controller itself. +await checkChunkInvariant({ input, createParser: parser, signal: new AbortController() }); +// @ts-expect-error A boolean is not an AbortSignal. +await assertChunkInvariant({ input, createParser: parser, signal: true });