From 362125c738cf3034c7a71a8aafbfd1e350c9c479 Mon Sep 17 00:00:00 2001 From: Joe Clark Date: Fri, 16 Oct 2026 11:55:06 +0100 Subject: [PATCH 1/2] run worker integratio n tests wuth clean new instances each time --- integration-tests/worker/test/integration.test.ts | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/integration-tests/worker/test/integration.test.ts b/integration-tests/worker/test/integration.test.ts index 8ea364753..d282ea37b 100644 --- a/integration-tests/worker/test/integration.test.ts +++ b/integration-tests/worker/test/integration.test.ts @@ -12,7 +12,7 @@ let engine; let engineLogger; let lightningPort; -test.before(async () => { +test.beforeEach(async () => { const keys = await generateKeys(); lightningPort = nextPort(); lightning = initLightning(lightningPort, keys.private); @@ -33,11 +33,8 @@ test.before(async () => { )); }); -test.afterEach(() => { +test.afterEach(async () => { engineLogger._reset(); -}); - -test.after(async () => { lightning.destroy(); await worker.destroy(); }); From c3544246ad432e512cb82171ef122c89ebd4af29 Mon Sep 17 00:00:00 2001 From: Joe Clark Date: Fri, 16 Oct 2026 12:19:46 +0100 Subject: [PATCH 2/2] worker: add immediate destroy for unit tests --- .changeset/quick-pens-close.md | 5 ++ integration-tests/worker/src/init.ts | 2 +- .../worker/test/integration.test.ts | 22 +++--- packages/ws-worker/src/api/destroy.ts | 29 ++++++-- packages/ws-worker/src/server.ts | 5 +- packages/ws-worker/test/api/destroy.test.ts | 69 +++++++++++++++++++ 6 files changed, 113 insertions(+), 19 deletions(-) create mode 100644 .changeset/quick-pens-close.md diff --git a/.changeset/quick-pens-close.md b/.changeset/quick-pens-close.md new file mode 100644 index 000000000..3141a96c4 --- /dev/null +++ b/.changeset/quick-pens-close.md @@ -0,0 +1,5 @@ +--- +'@openfn/ws-worker': minor +--- + +Add an `immediate` option to worker destroy to tear down without waiting for runs diff --git a/integration-tests/worker/src/init.ts b/integration-tests/worker/src/init.ts index 978886c3a..24cf77456 100644 --- a/integration-tests/worker/src/init.ts +++ b/integration-tests/worker/src/init.ts @@ -53,7 +53,7 @@ export const initWorker = async ( lightning: `ws://localhost:${lightningPort}/worker`, secret: crypto.randomUUID(), collectionsVersion: '1.0.0', - messageTimeoutSeconds: 0.01, + messageTimeoutSeconds: 0.1, batchLogs: true, ...workerArgs, }); diff --git a/integration-tests/worker/test/integration.test.ts b/integration-tests/worker/test/integration.test.ts index d282ea37b..bde90d0c3 100644 --- a/integration-tests/worker/test/integration.test.ts +++ b/integration-tests/worker/test/integration.test.ts @@ -36,7 +36,7 @@ test.beforeEach(async () => { test.afterEach(async () => { engineLogger._reset(); lightning.destroy(); - await worker.destroy(); + await worker.destroy({ immediate: true }); }); const createDummyWorker = () => { @@ -650,7 +650,7 @@ test.serial('use local adaptor versions from monorepo', (t) => { test.serial("Don't send adaptor logs to stdout", (t) => { return new Promise(async (done) => { // We have to create a new worker with a different repo for this one - await worker.destroy(); + await worker.destroy({ immediate: true }); ({ worker, engineLogger } = await createDummyWorker()); const message = 've have been expecting you meester bond'; @@ -689,7 +689,7 @@ test.serial( (t) => { return new Promise(async (done) => { // We want to create our own special worker here - await worker.destroy(); + await worker.destroy({ immediate: true }); ({ worker, engineLogger } = await createDummyWorker()); const attempt1 = { @@ -736,7 +736,7 @@ test.serial( test.serial('worker should exit if it has an invalid key', (t) => { return new Promise(async (done) => { if (!worker.destroyed) { - await worker.destroy(); + await worker.destroy({ immediate: true }); } // generate a new, invalid, public key @@ -789,7 +789,7 @@ test.serial('worker should exit if it has an invalid key', (t) => { test.serial('set a default timeout on the worker', (t) => { return new Promise(async (done) => { if (!worker.destroyed) { - await worker.destroy(); + await worker.destroy({ immediate: true }); } ({ worker } = await initWorker(lightningPort, { @@ -839,7 +839,7 @@ test.serial('set a default timeout on the worker', (t) => { test.serial('set a timeout on a run', (t) => { return new Promise(async (done) => { if (!worker.destroyed) { - await worker.destroy(); + await worker.destroy({ immediate: true }); } ({ worker } = await initWorker(lightningPort, { @@ -879,7 +879,7 @@ test.serial('set a timeout on a run', (t) => { test.serial('set a default payload limit on the worker', (t) => { return new Promise(async (done) => { if (!worker.destroyed) { - await worker.destroy(); + await worker.destroy({ immediate: true }); } ({ worker } = await initWorker( @@ -920,7 +920,7 @@ test.serial('set a default payload limit on the worker', (t) => { test.serial('override the worker payload through run options', (t) => { return new Promise(async (done) => { if (!worker.destroyed) { - await worker.destroy(); + await worker.destroy({ immediate: true }); } ({ worker } = await initWorker( @@ -962,7 +962,7 @@ test.serial('override the worker payload through run options', (t) => { test.serial('Redact logs which exceed the payload limit', (t) => { return new Promise(async (done) => { if (!worker.destroyed) { - await worker.destroy(); + await worker.destroy({ immediate: true }); } ({ worker } = await initWorker(lightningPort, { @@ -1003,7 +1003,7 @@ test.serial('Redact logs which exceed the payload limit', (t) => { test.serial("Don't return dataclips which exceed the payload limit", (t) => { return new Promise(async (done) => { if (!worker.destroyed) { - await worker.destroy(); + await worker.destroy({ immediate: true }); } ({ worker } = await initWorker(lightningPort, { @@ -1043,7 +1043,7 @@ test.serial( "Don't send job logs to stdout when job_log_level is set to none", (t) => { return new Promise(async (done) => { - await worker.destroy(); + await worker.destroy({ immediate: true }); ({ worker, engineLogger } = await createDummyWorker()); const message = 'log that will never exist'; diff --git a/packages/ws-worker/src/api/destroy.ts b/packages/ws-worker/src/api/destroy.ts index a4a521c0e..bb9e0f867 100644 --- a/packages/ws-worker/src/api/destroy.ts +++ b/packages/ws-worker/src/api/destroy.ts @@ -3,8 +3,20 @@ import { INTERNAL_CLAIM_COMPLETE, INTERNAL_RUN_COMPLETE } from '../events'; import type { Logger } from '@openfn/logger'; -const destroy = async (app: ServerApp, logger: Logger) => { - logger.info('Closing server...'); +export type DestroyOptions = { + // Tear everything down now: don't wait for active runs or claims to complete + immediate?: boolean; +}; + +const destroy = async ( + app: ServerApp, + logger: Logger, + options: DestroyOptions = {} +) => { + const { immediate = false } = options; + logger.info( + immediate ? 'Closing server immediately...' : 'Closing server...' + ); // Close the server AND wait for runs to complete in parallel // The http server can take a moment to close and the two @@ -22,14 +34,21 @@ const destroy = async (app: ServerApp, logger: Logger) => { app.server.close(async () => { resolve(); }); + + if (immediate) { + // Don't wait for open connections to drain + app.server.closeAllConnections?.(); + } }), (async () => { - // Let any active runs complete - await waitForRunsAndClaims(app, logger); + if (!immediate) { + // Let any active runs complete + await waitForRunsAndClaims(app, logger); + } app.queueChannel?.leave(); // Kill the engine and socket - await app.engine.destroy(); + await app.engine.destroy(immediate); app.socket?.disconnect(); })(), ]); diff --git a/packages/ws-worker/src/server.ts b/packages/ws-worker/src/server.ts index 3c95d2f0b..47fa04dba 100644 --- a/packages/ws-worker/src/server.ts +++ b/packages/ws-worker/src/server.ts @@ -17,6 +17,7 @@ import { WORK_AVAILABLE, } from './events'; import destroy from './api/destroy'; +import type { DestroyOptions } from './api/destroy'; import { Workloop } from './api/workloop'; import claim from './api/claim'; import { Context, execute } from './api/execute'; @@ -87,7 +88,7 @@ export interface ServerApp extends Koa { runWorkloopMap: Record; execute: ({ id, token }: ClaimRun) => Promise; - destroy: () => void; + destroy: (options?: DestroyOptions) => Promise; resumeWorkloop: (workloop?: Workloop) => void; pendingClaims: () => number; @@ -469,7 +470,7 @@ function createServer(engine: RuntimeEngine, options: ServerOptions = {}) { app.pendingClaims = () => app.workloops.reduce((sum, w) => sum + Object.keys(w.openClaims).length, 0); - app.destroy = () => destroy(app, logger); + app.destroy = (options?: DestroyOptions) => destroy(app, logger, options); app.use(router.routes()); diff --git a/packages/ws-worker/test/api/destroy.test.ts b/packages/ws-worker/test/api/destroy.test.ts index 748f1e8fc..9d7c81c6b 100644 --- a/packages/ws-worker/test/api/destroy.test.ts +++ b/packages/ws-worker/test/api/destroy.test.ts @@ -337,3 +337,72 @@ test("don't claim after destroy", async (t) => { lightning.enqueueRun(createRun()); }); }); + +test.serial( + 'immediate destroy does not wait for an outstanding claim', + async (t) => { + // Delay the claim response so it is still outstanding when destroy returns + initLightning({ socketDelay: 200 }); + await initWorker({ noLoop: true }); + + return new Promise((done) => { + const doDestroy = async () => { + t.is(worker.pendingClaims(), 1); + + await destroy(worker, logger, { immediate: true }); + + // A graceful destroy would block until the claim completed + t.is(worker.pendingClaims(), 1); + t.true(worker.destroyed); + t.false(await ping()); + + done(); + }; + + worker.events.once(INTERNAL_CLAIM_START, () => { + doDestroy(); + }); + + lightning.enqueueRun(createRun()); + worker.claim().catch(() => {}); + }); + } +); + +test.serial( + 'immediate destroy tells the engine to destroy instantly', + async (t) => { + initLightning(); + await initWorker(); + + let instant: boolean | undefined; + const originalDestroy = worker.engine.destroy; + worker.engine.destroy = async (arg?: boolean) => { + instant = arg; + return originalDestroy(arg); + }; + + await worker.destroy({ immediate: true }); + + t.true(instant); + } +); + +test.serial( + 'a normal destroy does not ask the engine to destroy instantly', + async (t) => { + initLightning(); + await initWorker(); + + let instant: boolean | undefined = true; + const originalDestroy = worker.engine.destroy; + worker.engine.destroy = async (arg?: boolean) => { + instant = arg; + return originalDestroy(arg); + }; + + await worker.destroy(); + + t.falsy(instant); + } +);