From 333b97e76f673a4490fe60d509503495d9a0ed11 Mon Sep 17 00:00:00 2001 From: Samuel Mbabhazi Date: Fri, 28 Aug 2026 21:03:53 +0200 Subject: [PATCH] fix(reactor-cli): keep serving the other reactors when one reactor's poll rejects The host awaited every reactor's poll, gateway poll and shutdown with Promise.all, so a single reactor's rejection aborted the whole cycle and the serve loop exited, taking the healthy reactors down with it. Every per-reactor task now settles before failures are handled: with the new onReactorError handler the host keeps serving and reports the failing reactor, and without one the first failure is rethrown only once every reactor has settled. The serve command logs each failed cycle and emits the categorized error event once per reactor. --- CHANGELOG.md | 10 ++ .../src/__tests__/serve-host.test.ts | 131 ++++++++++++++++++ packages/reactor-cli/src/commands/serve.ts | 30 ++++ packages/reactor-cli/src/run/host.ts | 66 ++++++++- 4 files changed, 231 insertions(+), 6 deletions(-) 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 {