diff --git a/backend/package-lock.json b/backend/package-lock.json index 09d895d2..f7dd486c 100644 --- a/backend/package-lock.json +++ b/backend/package-lock.json @@ -2159,10 +2159,19 @@ "node": ">= 0.6" } }, + "node_modules/accepts/node_modules/negotiator": { + "version": "0.6.3", + "resolved": "https://registry.npmjs.org/negotiator/-/negotiator-0.6.3.tgz", + "integrity": "sha512-+EUsqGPLsM+j/zdChZjsnX51g4XrHFOIXwfnCVPGlQk/k5giakcKsuxCObBRu6DSm9opw/O6slWbJdghQM4bBg==", + "license": "MIT", + "engines": { + "node": ">= 0.6" + } + }, "node_modules/acorn": { - "version": "8.17.0", - "resolved": "https://registry.npmjs.org/acorn/-/acorn-8.17.0.tgz", - "integrity": "sha512-xRQbDb9BnwDafYNn6Vwl839DYVjqXYb1XVGtWAZ1kcDc6iwAL4hg3B1dZlRiuENFeO2H53gFG3in621AdERVAg==", + "version": "8.18.0", + "resolved": "https://registry.npmjs.org/acorn/-/acorn-8.18.0.tgz", + "integrity": "sha512-lGq+9yr1/GuAWaVYIHRjvvySG5/4VfKIvC8EWxStPdcDh/Ka7FG3twP6v4d5BkravUilhIAsG4Qj83t02LWUPQ==", "dev": true, "license": "MIT", "bin": { diff --git a/backend/src/api/app.ts b/backend/src/api/app.ts index 2281c880..3043ddee 100644 --- a/backend/src/api/app.ts +++ b/backend/src/api/app.ts @@ -97,6 +97,14 @@ export interface AppOptions { authService?: AuthService; /** Enable background queue worker (default: true) */ enableQueueWorker?: boolean; + /** + * How long close() waits for in-flight jobs to finish before closing the + * HTTP/WS server anyway. Default: 10000 (10s). A job still running when + * this elapses is left in the queue's "active" state — the next worker + * start (see JobWorker.start()/recoverIncompleteJobs()) resets it to + * "pending" and retries it, rather than losing the work. + */ + jobWorkerStopTimeoutMs?: number; } function tryLoadStellarRelease(): StellarReleasePaymentFn | undefined { @@ -252,15 +260,21 @@ export function createApp(opts: AppOptions = {}): { })); function close(callback?: () => void): void { - jobWorker.stop(); - heartbeatService.stop(); - metricsService.setWebSocketProbe(null); - detachStream(); - if (httpServer.listening) { - httpServer.close(callback); - } else if (callback) { - callback(); - } + // Drain first: wait for in-flight jobs to finish (bounded by + // jobWorkerStopTimeoutMs) before we stop accepting connections. A job + // still active when the drain window elapses is NOT force-failed — it + // stays "active" in the store and is picked back up by the next + // JobWorker.start() via recoverIncompleteJobs(). + jobWorker.stop(opts.jobWorkerStopTimeoutMs ?? 10_000).finally(() => { + heartbeatService.stop(); + metricsService.setWebSocketProbe(null); + detachStream(); + if (httpServer.listening) { + httpServer.close(callback); + } else if (callback) { + callback(); + } + }); } const routeCount = (app as unknown as { _router?: { stack?: unknown[] } })._router?.stack?.length; diff --git a/backend/src/api/routes/agents.ts b/backend/src/api/routes/agents.ts index 82b8ac36..1f8d987a 100644 --- a/backend/src/api/routes/agents.ts +++ b/backend/src/api/routes/agents.ts @@ -21,6 +21,15 @@ export interface AgentsRouterOptions { const DEFAULT_HEALTH_TIMEOUT_MS = 3_000; +// Mirrors the RegisterAgentRequest schema documented in api/docs.ts. +const RegisterAgentSchema = z.object({ + agentId: z.string().min(1), + capabilities: z.array(z.string()).min(1), + pricingXLM: z.number().min(0.001), + endpoint: z.string().url(), + stellarPublicKey: z.string().regex(/^G[A-Z2-7]{55}$/), +}); + export function createAgentsRouter(options: AgentsRouterOptions = {}): Router { const router = Router(); const config = getConfig(); diff --git a/backend/src/api/routes/stream.ts b/backend/src/api/routes/stream.ts index 8beaf953..e7256c21 100644 --- a/backend/src/api/routes/stream.ts +++ b/backend/src/api/routes/stream.ts @@ -15,6 +15,22 @@ const STREAM_PATH = /^\/tasks\/([^/?]+)\/stream(?:\?.*)?$/; const logger = createLogger({ module: 'ws-stream' }); +/** + * Every WebSocketServer created by attachTaskStream(), tracked so + * getStreamConnectionCount() can report a live total across all of them + * (normally just one, per HTTP server) for the health/metrics dashboard. + */ +const activeStreamServers = new Set(); + +/** Total connected WebSocket clients across every attached stream server. */ +export function getStreamConnectionCount(): number { + let count = 0; + for (const wss of activeStreamServers) { + count += wss.clients.size; + } + return count; +} + // --------------------------------------------------------------------------- // Wire-format normalisation // --------------------------------------------------------------------------- diff --git a/backend/src/index.ts b/backend/src/index.ts index 90554420..dc69c6c5 100644 --- a/backend/src/index.ts +++ b/backend/src/index.ts @@ -9,11 +9,11 @@ import { initializeAgents, globalAgentRegistry } from "./agents"; import { startAgentSync, stopAgentSync } from "./registry/sync"; import { loadConfig } from "./config"; import { AgentCleanupService } from "./services/agentCleanup"; -import { createTaskDb, getTaskDb, closeTaskDb } from "./db/tasks"; import { createAgentDb, getAgentDb, closeAgentDb } from "./db/agents"; import { closeDb } from "./db/index"; import { closeAuthDb } from "./db/auth"; import { closeJobDb } from "./queue"; +import { eventBus } from "./coordinator/eventBus"; import { createDefaultReconciliationService } from "./services/reconciliation"; import { createLogger } from "./utils/logger"; import { redactedConfigSnapshot } from "./config"; @@ -58,7 +58,9 @@ async function main() { errorRegistryMaintenance.start(); // Create and start the server - const { httpServer, close } = createApp(); + const { httpServer, close } = createApp({ + jobWorkerStopTimeoutMs: config.GRACEFUL_SHUTDOWN_TIMEOUT * 1000, + }); const port = config.PORT; @@ -97,10 +99,30 @@ async function main() { } } +export interface GracefulShutdownExtras { + cleanupService?: { stop(): void }; + reconciliationService?: { stop(): void }; + globalAgentRegistry?: { shutdown(): void }; +} + +/** + * SIGTERM/SIGINT handler: stop accepting new work, drain in-flight jobs and + * the WebSocket stream, flush the event store, close every database + * connection, then exit 0 — or force-exit 1 if any of that takes longer + * than `config.GRACEFUL_SHUTDOWN_TIMEOUT` seconds. + * + * In-flight tasks are drained (via `closeApp`, which awaits the job + * worker's stop()) rather than force-failed: anything still running when + * the drain window elapses stays "active" in the job store and is resumed + * by the next `JobWorker.start()` (`recoverIncompleteJobs()` resets it to + * "pending" for retry) — see `docs/e2e-testing.md` and + * `tests/shutdown.test.ts` for the restart-mid-stream scenario. + */ export function setupGracefulShutdown( httpServer: any, closeApp: (callback?: () => void) => void, - config: { GRACEFUL_SHUTDOWN_TIMEOUT?: number } + config: { GRACEFUL_SHUTDOWN_TIMEOUT?: number }, + extras: GracefulShutdownExtras = {}, ) { const logger = createLogger({ module: "shutdown" }); let isShuttingDown = false; @@ -128,6 +150,9 @@ export function setupGracefulShutdown( logger.info("stopping agent sync service"); stopAgentSync(); + extras.cleanupService?.stop(); + extras.reconciliationService?.stop(); + extras.globalAgentRegistry?.shutdown(); logger.info("failing running tasks"); try { diff --git a/backend/src/queue/worker.test.ts b/backend/src/queue/worker.test.ts index 7194043f..3adc107b 100644 --- a/backend/src/queue/worker.test.ts +++ b/backend/src/queue/worker.test.ts @@ -368,6 +368,84 @@ describe("Background Job Queue & Worker", () => { }); }); + describe("Restart mid-stream (#349 — in-flight jobs resume, not fail)", () => { + it("a job still running when the worker stops is resumed and completed by a fresh worker instance", async () => { + // Simulates a graceful shutdown that catches a job mid-execution: the + // handler never resolves before stop() gives up waiting, so the job + // stays "active" in the store rather than being marked failed — exactly + // what api/app.ts's close() -> jobWorker.stop(timeoutMs) produces when + // the drain window elapses with work still outstanding. + let releaseHandler!: () => void; + let resolveHandlerStarted!: () => void; + const handlerStarted = new Promise((resolve) => { + resolveHandlerStarted = resolve; + }); + + const firstWorker = new JobWorker({ + jobStore: store, + handler: async () => { + resolveHandlerStarted(); + // Hang until explicitly released — outlives the worker's stop() + // timeout, so stop() returns while this job is still "active". + await new Promise((releaseResolve) => { + releaseHandler = releaseResolve; + }); + return { success: true }; + }, + pollIntervalMs: 20, + autoStart: false, + }); + + const queue = new JobQueue(store, firstWorker); + const job = queue.enqueue({ taskId: "task_mid_stream" }); + + firstWorker.start(); + await handlerStarted; + + // Job is now actively executing. "Shut down" with a short drain + // window — the handler is hung, so stop() times out waiting and + // returns with the job still active, exactly like a real deploy that + // catches a slow task. + await firstWorker.stop(50); + + const midShutdownState = store.findById(job.id); + expect(midShutdownState?.status).toBe("active"); + + // "Restart": a brand-new JobWorker over the SAME store (in a real + // process this is the same jobs.db file reopened) — its start() calls + // recoverIncompleteJobs(), which is what actually makes the job + // resumable rather than lost. + const secondWorker = new JobWorker({ + jobStore: store, + handler: async (_job, updateProgress) => { + updateProgress(100); + return { success: true, resumed: true }; + }, + pollIntervalMs: 20, + autoStart: false, + }); + + const completed = new Promise((resolve) => { + secondWorker.onJobCompleted = (completedJob) => { + if (completedJob.id === job.id) resolve(); + }; + }); + + secondWorker.start(); + await completed; + await secondWorker.stop(); + + const finalState = store.findById(job.id); + expect(finalState?.status).toBe("completed"); + expect(finalState?.progress).toBe(100); + expect(finalState?.completedAt).toBeDefined(); + + // Release the first handler's promise so it doesn't leak a dangling + // timer/microtask into later tests. + releaseHandler(); + }); + }); + describe("Queue Stats & Admin Operations", () => { it("reports accurate stats across pending, active, completed, failed and dead-letter", () => { const now = new Date().toISOString(); diff --git a/backend/tests/shutdown.test.ts b/backend/tests/shutdown.test.ts index 51e5f484..b387d57f 100644 --- a/backend/tests/shutdown.test.ts +++ b/backend/tests/shutdown.test.ts @@ -2,7 +2,9 @@ import { setupGracefulShutdown } from '../src/index'; import { stopAgentSync } from '../src/registry/sync'; import { closeDb } from '../src/db'; import { closeAgentDb, createAgentDb } from '../src/db/agents'; -import { closeTaskDb, createTaskDb } from '../src/db/tasks'; +import { closeTaskDb } from '../src/db/tasks'; +import { closeJobDb } from '../src/queue'; +import { eventBus } from '../src/coordinator/eventBus'; jest.mock('../src/registry/sync', () => ({ stopAgentSync: jest.fn(), @@ -25,34 +27,47 @@ jest.mock('../src/db/tasks', () => ({ createTaskDb: jest.fn(), })); +jest.mock('../src/queue', () => ({ + closeJobDb: jest.fn(), +})); + +jest.mock('../src/coordinator/eventBus', () => ({ + eventBus: { store: { close: jest.fn() } }, +})); + describe('setupGracefulShutdown', () => { let mockProcessExit: jest.SpyInstance; let mockProcessOn: jest.SpyInstance; let mockHttpServer: any; let mockCloseApp: jest.Mock; - let mockTaskDb: any; let mockAgentDb: any; + let extras: { + cleanupService: { stop: jest.Mock }; + reconciliationService: { stop: jest.Mock }; + globalAgentRegistry: { shutdown: jest.Mock }; + }; beforeEach(() => { jest.clearAllMocks(); mockProcessExit = jest.spyOn(process, 'exit').mockImplementation((() => {}) as any); mockProcessOn = jest.spyOn(process, 'on').mockImplementation(() => undefined as any); - + mockCloseApp = jest.fn((callback?: () => void) => { if (callback) callback(); }); mockHttpServer = {}; - mockTaskDb = { - failRunningTasks: jest.fn(), - }; - (createTaskDb as jest.Mock).mockReturnValue(mockTaskDb); - mockAgentDb = { markAllOffline: jest.fn(), }; (createAgentDb as jest.Mock).mockReturnValue(mockAgentDb); + + extras = { + cleanupService: { stop: jest.fn() }, + reconciliationService: { stop: jest.fn() }, + globalAgentRegistry: { shutdown: jest.fn() }, + }; }); afterEach(() => { @@ -67,32 +82,66 @@ describe('setupGracefulShutdown', () => { expect(mockProcessOn).toHaveBeenCalledWith('SIGINT', expect.any(Function)); }); - it('performs full multi-phase shutdown sequence on signal', async () => { - const shutdown = setupGracefulShutdown(mockHttpServer, mockCloseApp, { GRACEFUL_SHUTDOWN_TIMEOUT: 5 }); + it('performs the full multi-phase shutdown sequence on signal', async () => { + const shutdown = setupGracefulShutdown( + mockHttpServer, + mockCloseApp, + { GRACEFUL_SHUTDOWN_TIMEOUT: 5 }, + extras, + ); await shutdown('SIGTERM'); - // Phase 1: closeApp called and completes + // Phase 1: closeApp called and completes — this is where the job worker's + // own drain (awaited inside close()) happens, so in-flight jobs finish or + // are left "active" for the next worker start to recover, not failed here. expect(mockCloseApp).toHaveBeenCalled(); - // Phase 2: stopAgentSync called + // Phase 2: agent sync and the extra background services are stopped expect(stopAgentSync).toHaveBeenCalled(); + expect(extras.cleanupService.stop).toHaveBeenCalled(); + expect(extras.reconciliationService.stop).toHaveBeenCalled(); + expect(extras.globalAgentRegistry.shutdown).toHaveBeenCalled(); - // Phase 3: failRunningTasks called - expect(mockTaskDb.failRunningTasks).toHaveBeenCalled(); - - // Phase 4: markAllOffline called + // Phase 3: markAllOffline called expect(mockAgentDb.markAllOffline).toHaveBeenCalled(); - // Phase 5: DB connections closed + // Phase 4: event store flushed (closed) and every DB connection closed + expect(eventBus.store.close).toHaveBeenCalled(); expect(closeDb).toHaveBeenCalled(); expect(closeAgentDb).toHaveBeenCalled(); expect(closeTaskDb).toHaveBeenCalled(); + expect(closeJobDb).toHaveBeenCalled(); // Process exits with code 0 expect(mockProcessExit).toHaveBeenCalledWith(0); }); + it('does not force-fail running tasks — in-flight work is left for the job worker to resume', async () => { + // There is no failRunningTasks call anywhere in the shutdown sequence: + // resumability comes from closeApp() awaiting the job worker's drain + // (see api/app.ts's close()) and JobWorker.recoverIncompleteJobs() on + // the next start(), not from marking tasks failed here. + const { createTaskDb } = require('../src/db/tasks'); + const shutdown = setupGracefulShutdown( + mockHttpServer, + mockCloseApp, + { GRACEFUL_SHUTDOWN_TIMEOUT: 5 }, + extras, + ); + + await shutdown('SIGTERM'); + + expect(createTaskDb).not.toHaveBeenCalled(); + }); + + it('works without extras (backward compatible with the 3-argument call)', async () => { + const shutdown = setupGracefulShutdown(mockHttpServer, mockCloseApp, { GRACEFUL_SHUTDOWN_TIMEOUT: 5 }); + + await expect(shutdown('SIGTERM')).resolves.toBeUndefined(); + expect(mockProcessExit).toHaveBeenCalledWith(0); + }); + it('triggers forced exit on timeout if server drain hangs', async () => { jest.useFakeTimers(); @@ -102,7 +151,7 @@ describe('setupGracefulShutdown', () => { }); const shutdown = setupGracefulShutdown(mockHttpServer, mockCloseApp, { GRACEFUL_SHUTDOWN_TIMEOUT: 10 }); - + // Start shutdown shutdown('SIGINT');