diff --git a/apps/electron-backend-e2e/src/performance/m3u-refresh-main-capture.ts b/apps/electron-backend-e2e/src/performance/m3u-refresh-main-capture.ts index b05d01be3..e38defbd9 100644 --- a/apps/electron-backend-e2e/src/performance/m3u-refresh-main-capture.ts +++ b/apps/electron-backend-e2e/src/performance/m3u-refresh-main-capture.ts @@ -55,6 +55,11 @@ import { createWorkerTerminationGenerationApi, type WorkerTerminationGenerationApi, } from './worker-termination-generation'; +import { + assertWorkerSampleCaptureValid, + createWorkerSampleDeadlineApi, + type WorkerSampleDeadlineApi, +} from './worker-sample-deadline'; import { createRendererProcessRssCaptureApi, type RendererProcessRssCaptureApi, @@ -230,6 +235,9 @@ export async function installMainCapture( performanceTimelineMergeApiFactorySource: createPerformanceTimelineMergeApi.toString(), stateKey: MAIN_CAPTURE_STATE_KEY, + workerSampleDeadlineApiFactorySource: + createWorkerSampleDeadlineApi.toString(), + workerSampleDeadlineMs: 5_000, workerTerminationGenerationApiFactorySource: createWorkerTerminationGenerationApi.toString(), xtreamIpcMarkerCaptureFactorySource: @@ -325,7 +333,9 @@ export async function installMainCapture( resolvedProfileHandle: CpuProfileHandle | null; requestPerformance: WorkerRequestPerformance[]; responseEpochMs: number | null; + sampleKey: object; samplePromise: Promise | null; + sampleTimedOut: boolean; samplingStarted: boolean; snapshotPath: string | null; terminatedEpochMs: number | null; @@ -399,6 +409,9 @@ export async function installMainCapture( restoreFactory( input.workerTerminationGenerationApiFactorySource ); + const workerSampleDeadlineApi = restoreFactory( + input.workerSampleDeadlineApiFactorySource + ); const diagnosticsPerformancePhaseEventParserFactory = restoreFunction< typeof createDiagnosticsPerformancePhaseEventParser >(input.diagnosticsPerformancePhaseEventParserFactorySource); @@ -467,6 +480,7 @@ export async function installMainCapture( active: false, captureGeneration: 0, captureInvalidReasons: [] as string[], + capturePoisonedReason: null as string | null, captureOptions: null as MainCaptureStartOptions | null, captureStartedEpochMs: null as number | null, cpuStart: null as NodeJS.CpuUsage | null, @@ -735,7 +749,9 @@ export async function installMainCapture( resolvedProfileHandle: null, requestPerformance: [], responseEpochMs: null, + sampleKey: {}, samplePromise: null, + sampleTimedOut: false, samplingStarted: false, snapshotPath: null, terminatedEpochMs: null, @@ -864,59 +880,91 @@ export async function installMainCapture( return record; }; const sampleWorker = (record: WorkerRecord): Promise => { - if (record.finalized) { + if (record.finalized || record.sampleTimedOut) { return Promise.resolve(); } if (record.samplePromise) { return record.samplePromise; } - record.samplePromise = (async () => { - try { - const stats = await record.worker.getHeapStatistics?.(); - if (stats) { - const heapUsedSample = stats.used_heap_size; - if ( - typeof heapUsedSample === 'number' && - Number.isFinite(heapUsedSample) && - heapUsedSample >= 0 - ) { - record.heapUsedSampleCount += 1; - record.heapPeak = Math.max( - record.heapPeak, - heapUsedSample - ); + const captureGeneration = record.captureGeneration; + const sampleKey = {}; + record.sampleKey = sampleKey; + const boundedSample = workerSampleDeadlineApi + .run({ + apply: ({ cpu, elu, stats }) => { + if (stats) { + const heapUsedSample = stats.used_heap_size; + if ( + typeof heapUsedSample === 'number' && + Number.isFinite(heapUsedSample) && + heapUsedSample >= 0 + ) { + record.heapUsedSampleCount += 1; + record.heapPeak = Math.max( + record.heapPeak, + heapUsedSample + ); + } + const externalMemorySample = stats.external_memory; + if ( + typeof externalMemorySample === 'number' && + Number.isFinite(externalMemorySample) && + externalMemorySample >= 0 + ) { + record.externalMemorySampleCount += 1; + record.externalPeak = Math.max( + record.externalPeak, + externalMemorySample + ); + } } - const externalMemorySample = stats.external_memory; - if ( - typeof externalMemorySample === 'number' && - Number.isFinite(externalMemorySample) && - externalMemorySample >= 0 - ) { - record.externalMemorySampleCount += 1; - record.externalPeak = Math.max( - record.externalPeak, - externalMemorySample - ); + if (cpu) { + record.cpuFirst ??= cpu; + record.cpuLast = cpu; } + if (elu) { + record.elu = elu.utilization; + } + }, + capturedGeneration: captureGeneration, + currentIdentity: () => ({ + captureGeneration: state.captureGeneration, + recordGeneration: record.captureGeneration, + sampleKey: record.sampleKey, + }), + onTimeout: () => { + record.sampleTimedOut = true; + state.capturePoisonedReason = 'worker-sample-timeout'; + invalidateCapture('worker-sample-timeout'); + recordTimeline({ + operationId: record.operationId ?? undefined, + playlistId: record.playlistId ?? undefined, + type: `${record.kind}-sample-timeout`, + }); + }, + operation: async () => { + const stats = await record.worker.getHeapStatistics?.(); + const cpu = await record.worker.cpuUsage?.(); + const elu = + record.worker.performance?.eventLoopUtilization( + record.eluStart ?? undefined + ) ?? null; + return { cpu, elu, stats }; + }, + sampleKey, + timeoutMs: input.workerSampleDeadlineMs, + }) + .then(() => undefined) + .finally(() => { + if ( + record.sampleKey === sampleKey && + record.samplePromise === boundedSample + ) { + record.samplePromise = null; } - const cpu = await record.worker.cpuUsage?.(); - if (cpu) { - record.cpuFirst ??= cpu; - record.cpuLast = cpu; - } - const elu = record.worker.performance?.eventLoopUtilization( - record.eluStart ?? undefined - ); - if (elu) { - record.elu = elu.utilization; - } - } catch { - // A one-shot worker may terminate between sampling calls. - } - })().finally(() => { - record.samplePromise = null; - }); - return record.samplePromise; + }); + record.samplePromise = boundedSample; + return boundedSample; }; const resetWorkerForCapture = (record: WorkerRecord): void => { record.cancelPostedEpochMs = null; @@ -945,7 +993,9 @@ export async function installMainCapture( record.resolvedProfileHandle = null; record.requestPerformance = []; record.responseEpochMs = null; + record.sampleKey = {}; record.samplePromise = null; + record.sampleTimedOut = false; record.samplingStarted = false; record.snapshotPath = null; record.terminatedEpochMs = null; @@ -1563,6 +1613,11 @@ export async function installMainCapture( const startCapture = async ( options: MainCaptureStartOptions ): Promise => { + if (state.capturePoisonedReason !== null) { + throw new Error( + `xtream-main-capture-poisoned:${state.capturePoisonedReason}` + ); + } const captureStartedEpochMs = nowEpochMs(); state.rendererWindowSession?.detach(); state.rendererWindowSession = null; @@ -1641,6 +1696,11 @@ export async function installMainCapture( databaseRequestIdentityCapture.successMarkerCount(), }), start: async (options: MainCaptureStartOptions): Promise => { + if (state.capturePoisonedReason !== null) { + throw new Error( + `xtream-main-capture-poisoned:${state.capturePoisonedReason}` + ); + } databaseWorkerPostGcCutoffApi.beginCapture(); await startCapture(options); }, @@ -1754,7 +1814,7 @@ export async function installMainCapture( ); } await Promise.all( - currentDatabaseRecords.map((record) => + currentWorkerRecords.map((record) => joinFinalWorkerSample(record) ) ); @@ -1962,7 +2022,8 @@ export async function installMainCapture( ? currentDatabaseRecords[0] : null; const nextCaptureUnavailableReason = - cutoff.lateRequestCount > 0 + state.capturePoisonedReason ?? + (cutoff.lateRequestCount > 0 ? 'database-worker-activity-after-cutoff' : dbRequests.size > 0 ? 'database-worker-not-idle' @@ -1980,7 +2041,7 @@ export async function installMainCapture( null ? (databaseRecord?.postGcHeapUnavailableReason ?? 'post-gc-capture-invalid') - : null; + : null); if (nextCaptureUnavailableReason === null) { databaseWorkerPostGcCutoffApi.rolloverCapture(); await startCapture(nextOptions); @@ -2488,6 +2549,7 @@ export async function rolloverMainCapture( }, { options, stateKey: MAIN_CAPTURE_STATE_KEY } ); + assertWorkerSampleCaptureValid(transport.xtream.invalidReasons); if (transport.rollover === null) { throw new Error('main-capture-rollover-status-missing'); } @@ -2512,6 +2574,7 @@ export async function stopMainCapture( }, MAIN_CAPTURE_STATE_KEY ); + assertWorkerSampleCaptureValid(transport.xtream.invalidReasons); return selectMainCaptureGeneration(transport); } diff --git a/apps/electron-backend-e2e/src/performance/worker-sample-capture-validity.spec.ts b/apps/electron-backend-e2e/src/performance/worker-sample-capture-validity.spec.ts new file mode 100644 index 000000000..95e502d7d --- /dev/null +++ b/apps/electron-backend-e2e/src/performance/worker-sample-capture-validity.spec.ts @@ -0,0 +1,73 @@ +/* eslint-disable playwright/expect-expect -- This is a Node assertion-based performance contract test. */ +import assert from 'node:assert/strict'; +import { readFileSync } from 'node:fs'; +import test from 'node:test'; + +interface WorkerSampleDeadlineModule { + assertWorkerSampleCaptureValid?: ( + invalidReasons: readonly string[] + ) => void; +} + +const modulePromise = import( + new URL('./worker-sample-deadline.ts', import.meta.url).href +).then((module) => module as WorkerSampleDeadlineModule); + +test('rejects a timed-out worker sample before returning plain capture metrics', async () => { + const module = await modulePromise; + const assertCaptureValid = module.assertWorkerSampleCaptureValid; + assert.equal(typeof assertCaptureValid, 'function'); + + assert.doesNotThrow(() => assertCaptureValid?.([])); + assert.throws( + () => assertCaptureValid?.(['worker-sample-timeout']), + /main-capture-worker-sample-timeout/ + ); +}); + +test('plain capture stop and rollover both enforce worker sample validity', () => { + const source = readFileSync( + new URL('./m3u-refresh-main-capture.ts', import.meta.url), + 'utf8' + ); + const rolloverStart = source.indexOf( + 'export async function rolloverMainCapture' + ); + const stopStart = source.indexOf('export async function stopMainCapture'); + const xtreamRolloverStart = source.indexOf( + 'export async function rolloverXtreamMainCapture' + ); + const rollover = source.slice(rolloverStart, stopStart); + const stop = source.slice(stopStart, xtreamRolloverStart); + + assert.match( + rollover, + /assertWorkerSampleCaptureValid\(transport\.xtream\.invalidReasons\)/ + ); + assert.match( + stop, + /assertWorkerSampleCaptureValid\(transport\.xtream\.invalidReasons\)/ + ); +}); + +test('capture stop settles final samples for every current worker', () => { + const source = readFileSync( + new URL('./m3u-refresh-main-capture.ts', import.meta.url), + 'utf8' + ); + const stopStart = source.indexOf('stop: async ('); + const stopEnd = source.indexOf( + 'xtreamStatus: (): XtreamMainCaptureStatus', + stopStart + ); + const stop = source.slice(stopStart, stopEnd); + + assert.match( + stop, + /await Promise\.all\(\s*currentWorkerRecords\.map\(\(record\) =>\s*joinFinalWorkerSample\(record\)/ + ); + assert.doesNotMatch( + stop, + /currentDatabaseRecords\.map\(\(record\) =>\s*joinFinalWorkerSample\(record\)/ + ); +}); diff --git a/apps/electron-backend-e2e/src/performance/worker-sample-deadline.spec.ts b/apps/electron-backend-e2e/src/performance/worker-sample-deadline.spec.ts new file mode 100644 index 000000000..4b826fa2b --- /dev/null +++ b/apps/electron-backend-e2e/src/performance/worker-sample-deadline.spec.ts @@ -0,0 +1,295 @@ +/* eslint-disable playwright/expect-expect -- This is a Node assertion-based performance contract test. */ +import assert from 'node:assert/strict'; +import { readFileSync } from 'node:fs'; +import test from 'node:test'; + +interface WorkerSampleDeadlineIdentity { + readonly captureGeneration: number; + readonly recordGeneration: number | null; + readonly sampleKey: object; +} + +type WorkerSampleDeadlineOutcome = + | { readonly status: 'completed' } + | { readonly error: unknown; readonly status: 'failed' } + | { readonly status: 'stale' } + | { readonly status: 'timed-out' }; + +interface WorkerSampleDeadlineApi { + run(input: { + readonly apply: (value: T) => void; + readonly capturedGeneration: number | null; + readonly currentIdentity: () => WorkerSampleDeadlineIdentity; + readonly onTimeout: () => void; + readonly operation: () => Promise; + readonly sampleKey: object; + readonly timers?: WorkerSampleDeadlineTimers; + readonly timeoutMs: number; + }): Promise; +} + +interface WorkerSampleDeadlineTimers { + clearTimeout(handle: unknown): void; + setTimeout(callback: () => void, timeoutMs: number): unknown; +} + +interface WorkerSampleDeadlineModule { + createWorkerSampleDeadlineApi?: () => WorkerSampleDeadlineApi; +} + +const modulePromise = import( + new URL('./worker-sample-deadline.ts', import.meta.url).href +) + .then((module) => module as WorkerSampleDeadlineModule) + .catch(() => null); + +function deferred(): { + readonly promise: Promise; + readonly resolve: (value: T) => void; +} { + let resolve!: (value: T) => void; + const promise = new Promise((resolvePromise) => { + resolve = resolvePromise; + }); + return { promise, resolve }; +} + +function createScheduler(): WorkerSampleDeadlineTimers & { + readonly activeCount: () => number; + readonly fireNext: () => void; +} { + const tasks = new Map void>(); + + return { + activeCount: () => tasks.size, + clearTimeout(handle: unknown): void { + tasks.delete(handle); + }, + fireNext(): void { + const next = tasks.entries().next().value as + [unknown, () => void] | undefined; + assert.ok(next, 'expected one pending deadline'); + tasks.delete(next[0]); + next[1](); + }, + setTimeout(callback): object { + const handle = {}; + tasks.set(handle, callback); + return handle; + }, + }; +} + +async function restoreSerializableApi(): Promise { + const module = await modulePromise; + assert.ok(module, 'worker sample deadline module must exist'); + const factory = module.createWorkerSampleDeadlineApi; + assert.equal(typeof factory, 'function'); + + const source = factory.toString(); + assert.doesNotMatch(source, /__name/); + const restoredFactory = Function( + `"use strict"; return (${source});` + )() as () => WorkerSampleDeadlineApi; + return restoredFactory(); +} + +test('applies a worker sample only while its generation and identity are current', async () => { + const scheduler = createScheduler(); + const api = await restoreSerializableApi(); + const operation = deferred(); + const sampleKey = {}; + const applied: number[] = []; + + const outcomePromise = api.run({ + apply: (value) => applied.push(value), + capturedGeneration: 7, + currentIdentity: () => ({ + captureGeneration: 7, + recordGeneration: 7, + sampleKey, + }), + onTimeout: () => assert.fail('the operation completed in time'), + operation: () => operation.promise, + sampleKey, + timers: scheduler, + timeoutMs: 5_000, + }); + operation.resolve(42); + + assert.deepEqual(await outcomePromise, { status: 'completed' }); + assert.deepEqual(applied, [42]); + assert.equal(scheduler.activeCount(), 0); +}); + +test('ignores an already queued deadline callback after successful completion', async () => { + const api = await restoreSerializableApi(); + const operation = deferred(); + const sampleKey = {}; + const applied: number[] = []; + let queuedTimeout: (() => void) | null = null; + let timeoutCount = 0; + const timers: WorkerSampleDeadlineTimers = { + clearTimeout(): void { + // The callback may already be queued when cancellation runs. + }, + setTimeout(callback): object { + queuedTimeout = callback; + return {}; + }, + }; + + const outcomePromise = api.run({ + apply: (value) => applied.push(value), + capturedGeneration: 8, + currentIdentity: () => ({ + captureGeneration: 8, + recordGeneration: 8, + sampleKey, + }), + onTimeout: () => { + timeoutCount += 1; + }, + operation: () => operation.promise, + sampleKey, + timers, + timeoutMs: 5_000, + }); + operation.resolve(43); + + assert.deepEqual(await outcomePromise, { status: 'completed' }); + assert.deepEqual(applied, [43]); + assert.ok(queuedTimeout); + queuedTimeout(); + assert.equal(timeoutCount, 0); + assert.deepEqual(applied, [43]); +}); + +test('times out a stalled sample and ignores its late result', async () => { + const scheduler = createScheduler(); + const api = await restoreSerializableApi(); + const operation = deferred(); + const sampleKey = {}; + const applied: number[] = []; + let timeoutCount = 0; + + const outcomePromise = api.run({ + apply: (value) => applied.push(value), + capturedGeneration: 11, + currentIdentity: () => ({ + captureGeneration: 11, + recordGeneration: 11, + sampleKey, + }), + onTimeout: () => { + timeoutCount += 1; + }, + operation: () => operation.promise, + sampleKey, + timers: scheduler, + timeoutMs: 5_000, + }); + scheduler.fireNext(); + + assert.deepEqual(await outcomePromise, { status: 'timed-out' }); + assert.equal(timeoutCount, 1); + assert.deepEqual(applied, []); + + operation.resolve(99); + await Promise.resolve(); + await Promise.resolve(); + assert.deepEqual( + applied, + [], + 'a non-cancellable Worker promise must not mutate capture state after its deadline' + ); +}); + +test('fails closed when capture rollover makes a pending sample stale', async () => { + const scheduler = createScheduler(); + const api = await restoreSerializableApi(); + const operation = deferred(); + const sampleKey = {}; + const applied: number[] = []; + let currentGeneration = 4; + + const outcomePromise = api.run({ + apply: (value) => applied.push(value), + capturedGeneration: 4, + currentIdentity: () => ({ + captureGeneration: currentGeneration, + recordGeneration: currentGeneration, + sampleKey, + }), + onTimeout: () => assert.fail('a stale sample is not a timeout'), + operation: () => operation.promise, + sampleKey, + timers: scheduler, + timeoutMs: 5_000, + }); + currentGeneration = 5; + operation.resolve(7); + + assert.deepEqual(await outcomePromise, { status: 'stale' }); + assert.deepEqual(applied, []); + assert.equal(scheduler.activeCount(), 0); +}); + +test('the main capture bounds worker sampling and invalidates timed-out evidence', () => { + const source = readFileSync( + new URL('./m3u-refresh-main-capture.ts', import.meta.url), + 'utf8' + ); + const sampleWorkerStart = source.indexOf('const sampleWorker ='); + const sampleWorkerEnd = source.indexOf( + 'const resetWorkerForCapture', + sampleWorkerStart + ); + const sampleWorker = source.slice(sampleWorkerStart, sampleWorkerEnd); + const startCaptureStart = source.indexOf('const startCapture ='); + const startCaptureEnd = source.indexOf('const api =', startCaptureStart); + const startCapture = source.slice(startCaptureStart, startCaptureEnd); + const apiStartStart = source.indexOf( + 'start: async (options: MainCaptureStartOptions)' + ); + const apiStartEnd = source.indexOf( + 'beginMeasurement: async', + apiStartStart + ); + const apiStart = source.slice(apiStartStart, apiStartEnd); + const stopStart = source.indexOf('stop: async ('); + const stopEnd = source.indexOf( + 'xtreamStatus: (): XtreamMainCaptureStatus', + stopStart + ); + const stopCapture = source.slice(stopStart, stopEnd); + + assert.match( + source, + /workerSampleDeadlineApiFactorySource:\s+createWorkerSampleDeadlineApi\.toString\(\)/ + ); + assert.match(source, /workerSampleDeadlineMs:\s+5_000/); + assert.match(source, /capturePoisonedReason:\s+null as string \| null/); + assert.match(source, /sampleKey: object/); + assert.match(source, /sampleTimedOut: boolean/); + assert.match(sampleWorker, /workerSampleDeadlineApi\s*\.run\(/); + assert.match(sampleWorker, /invalidateCapture\('worker-sample-timeout'\)/); + assert.match( + sampleWorker, + /state\.capturePoisonedReason\s*=\s*'worker-sample-timeout'/ + ); + assert.match(sampleWorker, /record\.sampleKey === sampleKey/); + assert.match(sampleWorker, /record\.samplePromise === boundedSample/); + assert.match( + startCapture, + /if \(state\.capturePoisonedReason !== null\)[\s\S]*xtream-main-capture-poisoned/ + ); + assert.match( + apiStart, + /if \(state\.capturePoisonedReason !== null\)[\s\S]*xtream-main-capture-poisoned[\s\S]*databaseWorkerPostGcCutoffApi\.beginCapture\(\)/ + ); + assert.match( + stopCapture, + /const nextCaptureUnavailableReason =\s+state\.capturePoisonedReason \?\?/ + ); +}); diff --git a/apps/electron-backend-e2e/src/performance/worker-sample-deadline.ts b/apps/electron-backend-e2e/src/performance/worker-sample-deadline.ts new file mode 100644 index 000000000..2b0310132 --- /dev/null +++ b/apps/electron-backend-e2e/src/performance/worker-sample-deadline.ts @@ -0,0 +1,175 @@ +export interface WorkerSampleDeadlineIdentity { + readonly captureGeneration: number; + readonly recordGeneration: number | null; + readonly sampleKey: object; +} + +export type WorkerSampleDeadlineOutcome = + | { readonly status: 'completed' } + | { readonly error: unknown; readonly status: 'failed' } + | { readonly status: 'stale' } + | { readonly status: 'timed-out' }; + +export interface WorkerSampleDeadlineInput { + readonly apply: (value: T) => void; + readonly capturedGeneration: number | null; + readonly currentIdentity: () => WorkerSampleDeadlineIdentity; + readonly onTimeout: () => void; + readonly operation: () => Promise; + readonly sampleKey: object; + readonly timers?: WorkerSampleDeadlineTimers; + readonly timeoutMs: number; +} + +export interface WorkerSampleDeadlineTimers { + clearTimeout(handle: unknown): void; + setTimeout(callback: () => void, timeoutMs: number): unknown; +} + +export interface WorkerSampleDeadlineApi { + run( + input: WorkerSampleDeadlineInput + ): Promise; +} + +export function assertWorkerSampleCaptureValid( + invalidReasons: readonly string[] +): void { + if (invalidReasons.includes('worker-sample-timeout')) { + throw new Error('main-capture-worker-sample-timeout'); + } +} + +export function createWorkerSampleDeadlineApi(): WorkerSampleDeadlineApi { + const defaultTimers: WorkerSampleDeadlineTimers = { + clearTimeout(handle: unknown): void { + globalThis.clearTimeout( + handle as ReturnType + ); + }, + setTimeout(callback: () => void, timeoutMs: number): unknown { + return globalThis.setTimeout(callback, timeoutMs); + }, + }; + + const helpers = { + isCurrent(input: WorkerSampleDeadlineInput): boolean { + try { + const current = input.currentIdentity(); + return ( + Number.isSafeInteger(input.capturedGeneration) && + Number(input.capturedGeneration) > 0 && + current.captureGeneration === input.capturedGeneration && + current.recordGeneration === input.capturedGeneration && + current.sampleKey === input.sampleKey + ); + } catch { + return false; + } + }, + + run( + input: WorkerSampleDeadlineInput + ): Promise { + if (!Number.isFinite(input.timeoutMs) || input.timeoutMs <= 0) { + return Promise.resolve({ + error: new RangeError( + 'worker sample deadline must be a positive finite number' + ), + status: 'failed', + }); + } + + const timers = input.timers ?? defaultTimers; + return new Promise( + (resolvePromise) => { + let settled = false; + let timeoutHandle: unknown; + let timeoutScheduled = false; + const callbacks = { + complete(outcome: WorkerSampleDeadlineOutcome): void { + if (settled) { + return; + } + settled = true; + if (timeoutScheduled) { + try { + timers.clearTimeout(timeoutHandle); + } catch { + // Best-effort cleanup must not hide evidence. + } + } + resolvePromise(outcome); + }, + + onError(error: unknown): void { + if (settled) { + return; + } + callbacks.complete( + helpers.isCurrent(input) + ? { error, status: 'failed' } + : { status: 'stale' } + ); + }, + + onTimeout(): void { + if (settled) { + return; + } + if (!helpers.isCurrent(input)) { + callbacks.complete({ status: 'stale' }); + return; + } + try { + input.onTimeout(); + callbacks.complete({ status: 'timed-out' }); + } catch (error: unknown) { + callbacks.complete({ + error, + status: 'failed', + }); + } + }, + + onValue(value: T): void { + if (settled) { + return; + } + if (!helpers.isCurrent(input)) { + callbacks.complete({ status: 'stale' }); + return; + } + try { + input.apply(value); + callbacks.complete({ status: 'completed' }); + } catch (error: unknown) { + callbacks.complete({ + error, + status: 'failed', + }); + } + }, + }; + + try { + timeoutHandle = timers.setTimeout( + callbacks.onTimeout, + input.timeoutMs + ); + timeoutScheduled = true; + } catch (error: unknown) { + callbacks.complete({ error, status: 'failed' }); + return; + } + + void Promise.resolve() + .then(input.operation) + .then(callbacks.onValue, callbacks.onError); + } + ); + }, + }; + + return Object.freeze({ run: helpers.run }); +}