diff --git a/CHANGELOG.md b/CHANGELOG.md index b1f9f56..cc71427 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed + +- `reactor serve` no longer takes every hosted reactor down when one reactor's + continuity poll, gateway poll or shutdown rejects: the host settles every + reactor before handling failures, the failing reactor is reported (`reactor + : poll failed: `, with the categorized error event sampled once + per reactor), and the other reactors keep being served. `bootHost` gains an + `onReactorError` handler; without one the first failure is still rethrown, but + only once every reactor has settled. + ## [reactor 0.3.3 / reactor-cli 0.2.4 / reactor-devtools 0.3.1] - 2026-08-24 Metadata release from the standalone repository. No runtime changes. diff --git a/packages/reactor-cli/src/__tests__/serve-host.test.ts b/packages/reactor-cli/src/__tests__/serve-host.test.ts index fd92c1a..7f0eb71 100644 --- a/packages/reactor-cli/src/__tests__/serve-host.test.ts +++ b/packages/reactor-cli/src/__tests__/serve-host.test.ts @@ -298,6 +298,137 @@ describe('reactor serve host — multi-reactor isolation (offline gate)', () => } }); + it('a reactor whose poll rejects is reported while the other reactors keep being served', async () => { + const stateRoot = freshState(); + const projectDir = writeTwoReactorProject(stateRoot); + // alpha's freshness reader arms the monitor at boot (a lapsed `valid_until`), + // then rejects on the next read: the one the continuity poll performs. + let alphaReads = 0; + const alphaFreshness = (node: string): unknown => { + if (node !== MONITOR) { + return null; + } + alphaReads += 1; + if (alphaReads > 1) { + throw new Error('alpha freshness store unavailable'); + } + return { + node: MONITOR, + contract_fingerprint: 'c:monitor@1', + input_fingerprints: {}, + facets: [{ facet: 'freshness', fingerprint: 'f:stale', valid_until: '2020-01-01T00:00:00.000Z' }], + prev: null, + }; + }; + const failures: { name: string; phase: string; message: string }[] = []; + try { + const host = await bootHost({ + projectDir, + stateDir: stateRoot, + offline: true, + concurrency: 2, + testSeams: { + alpha: { ...seamFor(stateRoot, 'alpha'), testReadFreshness: alphaFreshness }, + beta: seamFor(stateRoot, 'beta'), + }, + onReactorError: (failure) => { + failures.push({ + name: failure.name, + phase: failure.phase, + message: failure.error instanceof Error ? failure.error.message : String(failure.error), + }); + }, + }); + try { + const beta = host.byName('beta'); + assert.ok(beta, 'beta resolvable'); + const betaReceiptsBefore = beta.reactor.ledger.all().length; + + // The host resolves: alpha's failure is handed over, beta is untouched. + await host.pollAll(host.reactors[0]!.reactor.clock.now()); + assert.deepEqual(failures, [ + { name: 'alpha', phase: 'poll', message: 'alpha freshness store unavailable' }, + ]); + assert.equal(alphaReads, 2, 'alpha was armed at boot and read once by the poll'); + assert.equal(beta.reactor.ledger.all().length, betaReceiptsBefore, 'beta ledger untouched'); + + // beta keeps being served on the next cycle; alpha keeps being reported. + await host.pollAll(host.reactors[0]!.reactor.clock.now()); + assert.equal(failures.length, 2); + assert.ok(failures.every((f) => f.name === 'alpha')); + await beta.trigger(MONITOR); + assert.ok(beta.reactor.ledger.all().length > betaReceiptsBefore, 'beta still triggers'); + } finally { + await host.shutdown(); + } + } finally { + rmSync(stateRoot, { recursive: true, force: true }); + } + }); + + it('without an error handler a rejected poll surfaces only after every reactor settled', async () => { + const stateRoot = freshState(); + const projectDir = writeTwoReactorProject(stateRoot); + // alpha rejects on the poll's read; beta's reader is armed the same way but + // keeps answering, and records the read the poll performs. Serial pool: + // alpha polls first, so beta's poll can only run after alpha rejected. + const stale = { + node: MONITOR, + contract_fingerprint: 'c:monitor@1', + input_fingerprints: {}, + facets: [{ facet: 'freshness', fingerprint: 'f:stale', valid_until: '2020-01-01T00:00:00.000Z' }], + prev: null, + }; + let alphaReads = 0; + const alphaFreshness = (node: string): unknown => { + if (node !== MONITOR) { + return null; + } + alphaReads += 1; + if (alphaReads > 1) { + throw new Error('alpha freshness store unavailable'); + } + return stale; + }; + let betaPollReads = 0; + let betaArmed = false; + const betaFreshness = (node: string): unknown => { + if (node !== MONITOR) { + return null; + } + if (betaArmed) { + betaPollReads += 1; + return null; // freshness moved: re-arm, fire nothing + } + betaArmed = true; + return stale; + }; + try { + const host = await bootHost({ + projectDir, + stateDir: stateRoot, + offline: true, + concurrency: 1, + testSeams: { + alpha: { ...seamFor(stateRoot, 'alpha'), testReadFreshness: alphaFreshness }, + beta: { ...seamFor(stateRoot, 'beta'), testReadFreshness: betaFreshness }, + }, + }); + try { + await assert.rejects( + host.pollAll(host.reactors[0]!.reactor.clock.now()), + /alpha freshness store unavailable/, + ); + // By the time the rejection is observed, beta's poll has already run. + assert.equal(betaPollReads, 1, 'beta polled despite alpha rejecting first'); + } finally { + await host.shutdown(); + } + } finally { + rmSync(stateRoot, { recursive: true, force: true }); + } + }); + it('committed fingerprints match between a concurrency=2 host and a serial host', async () => { const fingerprintsFor = async (concurrency: number) => { const stateRoot = freshState(); diff --git a/packages/reactor-cli/src/commands/serve.ts b/packages/reactor-cli/src/commands/serve.ts index d1d54b4..e2f3b94 100644 --- a/packages/reactor-cli/src/commands/serve.ts +++ b/packages/reactor-cli/src/commands/serve.ts @@ -581,6 +581,7 @@ export async function runServeCommand( }; let host: Awaited>; + const reactorErrorsReported = new Set(); try { host = await bootHost({ ...(options.stateDir !== undefined ? { stateDir: options.stateDir } : {}), @@ -589,6 +590,35 @@ export async function runServeCommand( ...(options.offline !== undefined ? { offline: options.offline } : {}), ...(options.concurrency !== undefined ? { concurrency: options.concurrency } : {}), ...(options.testSeams !== undefined ? { testSeams: options.testSeams } : {}), + onReactorError: (failure) => { + // One reactor's failed poll is an operator-visible line every cycle, but + // the categorized error event is sampled once per reactor (a persistent + // fault polled on a 60s cadence would otherwise flood the backend). + const msg = + failure.error instanceof Error ? failure.error.message : String(failure.error); + if (options.json === true) { + write( + JSON.stringify({ + status: 'reactor-error', + reactor: failure.name, + phase: failure.phase, + message: msg, + }), + ); + } else { + write(`reactor ${failure.name}: ${failure.phase} failed: ${msg}`); + } + if (!reactorErrorsReported.has(failure.name)) { + reactorErrorsReported.add(failure.name); + telemetry.event( + TelemetryEvent.ERROR, + buildEventProperties( + { command: 'serve', outcome: 'failure', durationMs: Date.now() - startedAt }, + { errorCategory: errorCategory(failure.error) }, + ), + ); + } + }, }); } catch (err) { // Boot failure (stale IR / compile failed / no contracts) gets a clean diff --git a/packages/reactor-cli/src/run/host.ts b/packages/reactor-cli/src/run/host.ts index 99f66d7..9a669dc 100644 --- a/packages/reactor-cli/src/run/host.ts +++ b/packages/reactor-cli/src/run/host.ts @@ -68,6 +68,19 @@ export interface BootHostOptions extends ConfigOverrides { * A single-reactor host uses the name `"default"`. */ readonly testSeams?: Readonly>; + /** + * Receives each reactor whose poll, gateway poll or shutdown rejected. With a + * handler the host keeps serving the other reactors and resolves; without one + * the first failure is rethrown, but only once every reactor has settled. + */ + readonly onReactorError?: (failure: ReactorFailure) => void; +} + +/** One hosted reactor's rejected poll, gateway poll or shutdown. */ +export interface ReactorFailure { + readonly name: string; + readonly phase: 'poll' | 'gateways' | 'shutdown'; + readonly error: unknown; } /** A running multi-reactor host: the isolated handles + the across-reactor pool. */ @@ -181,20 +194,51 @@ export async function bootHost(options: BootHostOptions = {}): Promise [h.name, h] as const)); + // Reactors are isolated: one reactor's rejection must not cut the others' + // work short, so every per-reactor task settles before failures are handled. + // With `onReactorError` each failure is handed over and the host carries on; + // without it the first failure is rethrown once everything has settled. + const settleAll = async ( + phase: ReactorFailure['phase'], + run: (handle: ServeHandle) => Promise, + ): Promise => { + const settled = await Promise.allSettled(handles.map((h) => run(h))); + const fulfilled: T[] = []; + let firstError: { error: unknown } | undefined; + settled.forEach((result, i) => { + if (result.status === 'fulfilled') { + fulfilled.push(result.value); + return; + } + const failure: ReactorFailure = { + name: handles[i]!.name, + phase, + error: result.reason, + }; + if (options.onReactorError !== undefined) { + options.onReactorError(failure); + } else { + firstError ??= { error: result.reason }; + } + }); + if (firstError !== undefined) { + throw firstError.error; + } + return fulfilled; + }; + const pollAll = async (now: string): Promise => { // Submit each reactor's poll to the across-reactor pool (bounded parallel); // each poll is itself serialized behind that reactor's own queue. - await Promise.all( - handles.map((h) => pool.submit(() => h.pollOnce(now))), - ); + await settleAll('poll', (h) => pool.submit(() => h.pollOnce(now))); }; const pollGatewaysAll = async (now: string): Promise => { // Each reactor's gateway poll is submitted to the across-reactor pool; the // poll itself enqueues onto that reactor's serialization queue (correction #4), // so within a reactor a gateway poll never overlaps a continuity poll/trigger. - const perReactor = await Promise.all( - handles.map((h) => pool.submit(() => h.pollGatewaysOnce(now))), + const perReactor = await settleAll('gateways', (h) => + pool.submit(() => h.pollGatewaysOnce(now)), ); return perReactor.flat(); }; @@ -209,8 +253,18 @@ export async function bootHost(options: BootHostOptions = {}): Promise => { - await Promise.all(handles.map((h) => h.shutdown())); + // Drain every reactor before surfacing a failed shutdown, so one reactor's + // rejection never leaves another's in-flight work undrained. + let failure: unknown; + try { + await settleAll('shutdown', (h) => h.shutdown()); + } catch (err) { + failure = err; + } await pool.onIdle(); + if (failure !== undefined) { + throw failure; + } }; return {