From 96d21d0514ccddccacbbf8619df01fb050f6e472 Mon Sep 17 00:00:00 2001 From: Josh Wright Date: Fri, 9 Oct 2026 12:33:15 -0400 Subject: [PATCH] Handle status fanout rejections (Refs #66) --- src/hooks.server.ts | 3 ++ src/lib/server/pipeline/worker.ts | 6 ++-- tests/unit/hooks-server.test.ts | 51 ++++++++++++++++++++++++++++ tests/unit/worker-scheduling.test.ts | 32 +++++++++++++++++ 4 files changed, 90 insertions(+), 2 deletions(-) create mode 100644 tests/unit/hooks-server.test.ts diff --git a/src/hooks.server.ts b/src/hooks.server.ts index b7b9e8c..cdefd6e 100644 --- a/src/hooks.server.ts +++ b/src/hooks.server.ts @@ -12,6 +12,9 @@ let booted = false; function boot(): void { if (booted) return; booted = true; + process.on('unhandledRejection', reason => { + console.error('[process] unhandled rejection:', reason); + }); migrate(); interruptDeployments(); startWorker(); diff --git a/src/lib/server/pipeline/worker.ts b/src/lib/server/pipeline/worker.ts index 7e27cf3..82c003a 100644 --- a/src/lib/server/pipeline/worker.ts +++ b/src/lib/server/pipeline/worker.ts @@ -276,9 +276,11 @@ export function startWorker(): void { void tick(); reconcileRequesters(); - void fanoutStatus(); + void fanoutStatus().catch(err => { console.error('[worker] status fanout error:', err); }); tickTimer = setInterval(() => { void tick(); reconcileRequesters(); }, TICK_MS); - statusTimer = setInterval(fanoutStatus, STATUS_REFRESH_MS); + statusTimer = setInterval(() => { + void fanoutStatus().catch(err => { console.error('[worker] status fanout error:', err); }); + }, STATUS_REFRESH_MS); console.log('[worker] started — sweep every', TICK_MS, 'ms, status every', STATUS_REFRESH_MS, 'ms'); } diff --git a/tests/unit/hooks-server.test.ts b/tests/unit/hooks-server.test.ts new file mode 100644 index 0000000..854dd19 --- /dev/null +++ b/tests/unit/hooks-server.test.ts @@ -0,0 +1,51 @@ +import type { Handle } from '@sveltejs/kit'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +vi.mock('../../src/lib/server/db/migrate.js', () => ({ migrate: vi.fn() })); +vi.mock('../../src/lib/server/pipeline/worker.js', () => ({ startWorker: vi.fn() })); +vi.mock('../../src/lib/server/provisioning/history.js', () => ({ interruptDeployments: vi.fn() })); +vi.mock('../../src/lib/server/auth/guard.js', () => ({ authenticationHandle: vi.fn(async () => new Response('OK')) })); +vi.mock('../../src/lib/server/pipeline/bus.js', () => ({ invalidateFleet: vi.fn() })); +vi.mock('../../src/lib/server/performance.js', () => ({ startPerformanceMonitor: vi.fn() })); + +let originalListeners = process.listeners('unhandledRejection'); +beforeEach(() => { + vi.resetModules(); vi.clearAllMocks(); + originalListeners = process.listeners('unhandledRejection'); +}); +afterEach(() => { + for (const listener of process.listeners('unhandledRejection')) { + if (!originalListeners.includes(listener)) process.off('unhandledRejection', listener); + } + vi.restoreAllMocks(); +}); + +describe('controller error handling', () => { + it('registers one process rejection logger before starting workers, even across requests', async () => { + const on = vi.spyOn(process, 'on'); + const log = vi.spyOn(console, 'error').mockImplementation(() => {}); + const { handle } = await import('../../src/hooks.server.js'); + const { startWorker } = await import('../../src/lib/server/pipeline/worker.js'); + const input = { + event: { request: new Request('http://localhost/'), route: { id: '/' }, locals: {} }, + resolve: vi.fn() + } as unknown as Parameters[0]; + + await handle(input); + await handle(input); + + const listeners = process.listeners('unhandledRejection').filter(listener => !originalListeners.includes(listener)); + expect(listeners).toHaveLength(1); + expect(startWorker).toHaveBeenCalledTimes(1); + const registration = on.mock.calls.findIndex(([event]) => event === 'unhandledRejection'); + expect(on.mock.invocationCallOrder[registration]).toBeLessThan(vi.mocked(startWorker).mock.invocationCallOrder[0]); + + const failure = new Error('unexpected background failure'); + listeners[0](failure, Promise.resolve()); + listeners[0]('non-Error rejection', Promise.resolve()); + expect(log.mock.calls).toEqual([ + ['[process] unhandled rejection:', failure], + ['[process] unhandled rejection:', 'non-Error rejection'] + ]); + }); +}); diff --git a/tests/unit/worker-scheduling.test.ts b/tests/unit/worker-scheduling.test.ts index c996502..371e334 100644 --- a/tests/unit/worker-scheduling.test.ts +++ b/tests/unit/worker-scheduling.test.ts @@ -1,5 +1,6 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { newFixture, type Fixture } from '../helpers/fixture.js'; +import { STATUS_REFRESH_MS } from '../../src/lib/server-freshness.js'; vi.mock('../../src/lib/server/pipeline/sweep.js', () => ({ sweep: vi.fn() })); vi.mock('../../src/lib/server/pipeline/beta-runner.js', () => ({ betaTick: vi.fn() })); @@ -33,6 +34,37 @@ async function setup() { } describe('independent provisioning scheduling', () => { + it.each(['startup', 'scheduled'] as const)('logs a failed %s status snapshot and retries on the next interval', async phase => { + const s = await setup(); + const { MonitoringSnapshot } = await import('../../src/lib/server/fleet/monitoring-snapshot.js'); + const refresh = vi.spyOn(MonitoringSnapshot.prototype, 'refresh'); + const log = vi.spyOn(console, 'error').mockImplementation(() => {}); + const failure = new Error('database or disk is full'); + vi.useFakeTimers(); + try { + if (phase === 'scheduled') { + s.startWorker(); + await vi.advanceTimersByTimeAsync(0); + refresh.mockClear(); + } + refresh.mockImplementationOnce(() => { throw failure; }); + if (phase === 'startup') s.startWorker(); + await vi.advanceTimersByTimeAsync(phase === 'startup' ? 0 : STATUS_REFRESH_MS); + expect(refresh).toHaveBeenCalledTimes(1); + expect(log).toHaveBeenCalledExactlyOnceWith('[worker] status fanout error:', failure); + + await vi.advanceTimersByTimeAsync(STATUS_REFRESH_MS); + expect(refresh).toHaveBeenCalledTimes(2); + expect(log).toHaveBeenCalledTimes(1); + } finally { + s.stopWorker(); s.background.resolve(); + await vi.advanceTimersByTimeAsync(0); + vi.useRealTimers(); + refresh.mockRestore(); log.mockRestore(); + await vi.waitFor(() => expect(s.workerActivity()).toEqual([])); + } + }); + it('collects match chat on worker ticks without overlapping a slow collection', async () => { const s = await setup(); const { collectFleetChat } = await import('../../src/lib/server/fleet/chat.js');