From 8296c3cccc5cdba74ac282d92a49c790f7f12e36 Mon Sep 17 00:00:00 2001 From: "Elias W. BA" Date: Fri, 14 Aug 2026 15:22:16 +0000 Subject: [PATCH 1/5] Stop the worker integration tests guessing port numbers The tests picked a port at random from a thousand wide range and started a server on it, without checking the number was free. There are around a dozen of those picks in a run, so two of them landing on the same number happened often enough to fail the job every few runs: the second server couldn't bind, its setup hook never finished, and the whole file sat pending until the two minute timeout. The files run one at a time and each is its own process, so handing out ports from a counter is enough. This also removes the three files that all hardcoded 4321, which was harmless while the files run serially but would bite anyone who turned on concurrency later. Closes #1501 --- integration-tests/worker/src/init.ts | 14 ++++++++++++-- integration-tests/worker/test/autoinstall.test.ts | 4 ++-- integration-tests/worker/test/benchmark.test.ts | 4 ++-- integration-tests/worker/test/exit-reasons.test.ts | 4 ++-- integration-tests/worker/test/integration.test.ts | 4 ++-- integration-tests/worker/test/runs.test.ts | 4 ++-- 6 files changed, 22 insertions(+), 12 deletions(-) diff --git a/integration-tests/worker/src/init.ts b/integration-tests/worker/src/init.ts index 36918dbc7..0383bacb1 100644 --- a/integration-tests/worker/src/init.ts +++ b/integration-tests/worker/src/init.ts @@ -10,7 +10,17 @@ import createLogger from '@openfn/logger'; const debugWorker = process.env.OPENFN_DEBUG_WORKER; const debugLightning = process.env.OPENFN_DEBUG_LIGHTNING; -export const randomPort = () => Math.round(2000 + Math.random() * 1000); +// Hand out a fresh port for every server we start. This used to pick a random +// number in a 1000 wide range without checking it was free, which collided +// often enough to fail the integration job every few runs: the second server +// couldn't bind, its setup hook never finished, and the file timed out. +// +// ava runs these files one at a time and each one is its own process, so a +// counter is enough - ports are released when the previous file exits. The base +// sits clear of the 3000s that the server tests walk through. +let nextPortNumber = 4400; + +export const nextPort = () => nextPortNumber++; export const initLightning = (port = 4000, privateKey?: string) => { // TODO the lightning mock right now doesn't use the secret @@ -30,7 +40,7 @@ export const initWorker = async ( engineArgs = {}, workerArgs = {} ) => { - const workerPort = randomPort(); + const workerPort = nextPort(); const engineLogger = createMockLogger('engine', { level: 'debug', diff --git a/integration-tests/worker/test/autoinstall.test.ts b/integration-tests/worker/test/autoinstall.test.ts index 2b612e786..d7832a96c 100644 --- a/integration-tests/worker/test/autoinstall.test.ts +++ b/integration-tests/worker/test/autoinstall.test.ts @@ -3,7 +3,7 @@ import path from 'node:path'; import { rm } from 'node:fs/promises'; import { generateKeys } from '@openfn/lightning-mock'; -import { initLightning, initWorker } from '../src/init'; +import { initLightning, initWorker, nextPort } from '../src/init'; import { createRun, createJob } from '../src/factories'; const generate = (adaptor, version) => { @@ -39,7 +39,7 @@ test.before(async () => { } catch (e) {} const keys = await generateKeys(); - const lightningPort = 4321; + const lightningPort = nextPort(); lightning = initLightning(lightningPort, keys.private); diff --git a/integration-tests/worker/test/benchmark.test.ts b/integration-tests/worker/test/benchmark.test.ts index 83990ccef..8d60c2657 100644 --- a/integration-tests/worker/test/benchmark.test.ts +++ b/integration-tests/worker/test/benchmark.test.ts @@ -2,7 +2,7 @@ import test from 'ava'; import path from 'node:path'; import { createRun } from '../src/factories'; -import { initLightning, initWorker } from '../src/init'; +import { initLightning, initWorker, nextPort } from '../src/init'; import { run, humanMb } from '../src/util'; let lightning; @@ -11,7 +11,7 @@ let worker; const maxConcurrency = 20; test.before(async () => { - const lightningPort = 4322; + const lightningPort = nextPort(); lightning = initLightning(lightningPort); diff --git a/integration-tests/worker/test/exit-reasons.test.ts b/integration-tests/worker/test/exit-reasons.test.ts index d2bc91fa6..94cd477f3 100644 --- a/integration-tests/worker/test/exit-reasons.test.ts +++ b/integration-tests/worker/test/exit-reasons.test.ts @@ -2,13 +2,13 @@ import test from 'ava'; import crypto from 'node:crypto'; import path from 'node:path'; -import { initLightning, initWorker } from '../src/init'; +import { initLightning, initWorker, nextPort } from '../src/init'; let lightning; let worker; test.before(async () => { - const lightningPort = 4321; + const lightningPort = nextPort(); lightning = initLightning(lightningPort); diff --git a/integration-tests/worker/test/integration.test.ts b/integration-tests/worker/test/integration.test.ts index dd7492ded..8ea364753 100644 --- a/integration-tests/worker/test/integration.test.ts +++ b/integration-tests/worker/test/integration.test.ts @@ -4,7 +4,7 @@ import crypto from 'node:crypto'; import Koa from 'koa'; import { generateKeys } from '@openfn/lightning-mock'; -import { initLightning, initWorker, randomPort } from '../src/init'; +import { initLightning, initWorker, nextPort } from '../src/init'; let lightning; let worker; @@ -14,7 +14,7 @@ let lightningPort; test.before(async () => { const keys = await generateKeys(); - lightningPort = randomPort(); + lightningPort = nextPort(); lightning = initLightning(lightningPort, keys.private); const engineArgs = { diff --git a/integration-tests/worker/test/runs.test.ts b/integration-tests/worker/test/runs.test.ts index 2a09eb88c..bcaa1f922 100644 --- a/integration-tests/worker/test/runs.test.ts +++ b/integration-tests/worker/test/runs.test.ts @@ -8,14 +8,14 @@ import { createJob, createTrigger, } from '../src/factories'; -import { initLightning, initWorker } from '../src/init'; +import { initLightning, initWorker, nextPort } from '../src/init'; let lightning; let worker; test.before(async () => { const keys = await generateKeys(); - const lightningPort = 4321; + const lightningPort = nextPort(); lightning = initLightning(lightningPort, keys.private); From 782aa0f70646ea4cbd10cdd10b207031ab67868f Mon Sep 17 00:00:00 2001 From: "Elias W. BA" Date: Fri, 14 Aug 2026 16:01:55 +0000 Subject: [PATCH 2/5] Wait for the worker to connect, and stop the sigterm test racing Two more causes of the intermittent failures, both in the tests. The worker server is handed back before it has connected to Lightning - it fetches the collections version over http first and only then opens the socket. Several tests swap the worker out and queue a run straight afterwards, so the run could end up sitting in the queue with nothing listening, and the file would hang until the timeout. That's the four pending tests we kept seeing in the integration file. Waiting for the socket to come up removes the race for every test at once. The sigterm test failed whenever a claim already in flight arrived after we sent the signal, which the worker can't do anything about. It now keys off the worker telling us it received the signal instead. --- integration-tests/worker/src/init.ts | 20 +++++++++++++++++++- integration-tests/worker/test/server.test.ts | 13 +++++++++---- 2 files changed, 28 insertions(+), 5 deletions(-) diff --git a/integration-tests/worker/src/init.ts b/integration-tests/worker/src/init.ts index 0383bacb1..46addd6bf 100644 --- a/integration-tests/worker/src/init.ts +++ b/integration-tests/worker/src/init.ts @@ -3,7 +3,7 @@ import crypto from 'node:crypto'; import createLightningServer, { toBase64 } from '@openfn/lightning-mock'; import createEngine from '@openfn/engine-multi'; -import createWorkerServer from '@openfn/ws-worker'; +import createWorkerServer, { INTERNAL_SOCKET_READY } from '@openfn/ws-worker'; import { createMockLogger } from '@openfn/logger'; import createLogger from '@openfn/logger'; @@ -66,5 +66,23 @@ export const initWorker = async ( ...workerArgs, }); + // The server is returned before it has connected to Lightning: it fetches the + // collections version over http first and only then opens the socket. Tests + // which swap the worker out and queue a run straight afterwards could leave + // that run sitting in the queue with nothing listening for it, and the whole + // file would hang until ava's timeout. Waiting for the socket closes that. + if (!worker.socket) { + await new Promise((resolve, reject) => { + const timeout = setTimeout(() => { + reject(new Error('worker did not connect to lightning within 10s')); + }, 10_000); + + worker.events.once(INTERNAL_SOCKET_READY, () => { + clearTimeout(timeout); + resolve(); + }); + }); + } + return { engine, engineLogger, worker }; }; diff --git a/integration-tests/worker/test/server.test.ts b/integration-tests/worker/test/server.test.ts index 15182dcb9..3c1788f1e 100644 --- a/integration-tests/worker/test/server.test.ts +++ b/integration-tests/worker/test/server.test.ts @@ -180,7 +180,6 @@ test.serial('allow a job to complete after receiving a sigterm', (t) => { test.serial("don't restore the claim loop after a sigterm", (t) => { return new Promise(async (done) => { let abort = false; - let didSendSigterm = false; const port = getPort(); const job = createJob({ @@ -208,7 +207,6 @@ test.serial("don't restore the claim loop after a sigterm", (t) => { // After the second run starts, there should be no more claims lightning.on('run:start', (evt) => { if (evt.runId === b.id) { - didSendSigterm = true; // Kill the worker once the second job has started // This will force an overlap of two pending runs at full capacity workerProcess.kill('SIGTERM'); @@ -220,8 +218,15 @@ test.serial("don't restore the claim loop after a sigterm", (t) => { lightning.enqueueRun(a); lightning.enqueueRun(b); - lightning.on('claim', (e) => { - if (didSendSigterm && !abort) { + lightning.on('claim', () => { + // Sending the signal and the worker acting on it are not the same moment, + // so a claim already in flight can land here through no fault of the + // worker. Only count claims made after it told us it had the signal. + const didReceiveSigterm = workerLogs.some((l) => + l.match(/SIGTERM RECEIVED/) + ); + + if (didReceiveSigterm && !abort) { abort = true; t.fail('Claim triggered after sigterm'); done(); From f518e3f950df15ef73ccb3a8128267098cbb6e5a Mon Sep 17 00:00:00 2001 From: "Elias W. BA" Date: Fri, 14 Aug 2026 16:16:06 +0000 Subject: [PATCH 3/5] Don't lose the run:complete listener in the exit reason tests The helper every test in that file goes through listened with a plain once and then checked the run id inside the callback. Any completion for another run - a late one from the previous test, or a duplicate - used up the listener without resolving, so the promise hung for good and the rest of the file never ran. The mock already has onSocketEvent for exactly this: it only releases the listener once the id matches. --- integration-tests/worker/test/exit-reasons.test.ts | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/integration-tests/worker/test/exit-reasons.test.ts b/integration-tests/worker/test/exit-reasons.test.ts index 94cd477f3..79b38515f 100644 --- a/integration-tests/worker/test/exit-reasons.test.ts +++ b/integration-tests/worker/test/exit-reasons.test.ts @@ -23,8 +23,13 @@ test.after(async () => { }); const run = async (attempt) => { - return new Promise(async (done) => { - lightning.once('run:complete', (evt) => { + return new Promise((done) => { + // This used to be a plain `once` with the id checked inside it, which meant + // any completion for another run - a late one from the test before, or a + // duplicate - used up the listener and left this promise hanging for good, + // stalling the rest of the file until ava gave up. onSocketEvent only + // releases the listener once the id actually matches. + lightning.onSocketEvent('run:complete', attempt.id, (evt) => { if (attempt.id === evt.runId) { done(evt.payload); } From 7fbb9d6e654a85856638c8a90fb46bd84e07a17c Mon Sep 17 00:00:00 2001 From: "Elias W. BA" Date: Fri, 14 Aug 2026 16:27:30 +0000 Subject: [PATCH 4/5] Rerun CI to check the fixes hold From f35362362fe9e113079ba1307162cd0eca03dff6 Mon Sep 17 00:00:00 2001 From: "Elias W. BA" Date: Fri, 14 Aug 2026 16:28:17 +0000 Subject: [PATCH 5/5] Drop the explanatory comments per review --- integration-tests/worker/src/init.ts | 13 ------------- integration-tests/worker/test/exit-reasons.test.ts | 5 ----- integration-tests/worker/test/server.test.ts | 3 --- 3 files changed, 21 deletions(-) diff --git a/integration-tests/worker/src/init.ts b/integration-tests/worker/src/init.ts index 46addd6bf..978886c3a 100644 --- a/integration-tests/worker/src/init.ts +++ b/integration-tests/worker/src/init.ts @@ -10,14 +10,6 @@ import createLogger from '@openfn/logger'; const debugWorker = process.env.OPENFN_DEBUG_WORKER; const debugLightning = process.env.OPENFN_DEBUG_LIGHTNING; -// Hand out a fresh port for every server we start. This used to pick a random -// number in a 1000 wide range without checking it was free, which collided -// often enough to fail the integration job every few runs: the second server -// couldn't bind, its setup hook never finished, and the file timed out. -// -// ava runs these files one at a time and each one is its own process, so a -// counter is enough - ports are released when the previous file exits. The base -// sits clear of the 3000s that the server tests walk through. let nextPortNumber = 4400; export const nextPort = () => nextPortNumber++; @@ -66,11 +58,6 @@ export const initWorker = async ( ...workerArgs, }); - // The server is returned before it has connected to Lightning: it fetches the - // collections version over http first and only then opens the socket. Tests - // which swap the worker out and queue a run straight afterwards could leave - // that run sitting in the queue with nothing listening for it, and the whole - // file would hang until ava's timeout. Waiting for the socket closes that. if (!worker.socket) { await new Promise((resolve, reject) => { const timeout = setTimeout(() => { diff --git a/integration-tests/worker/test/exit-reasons.test.ts b/integration-tests/worker/test/exit-reasons.test.ts index 79b38515f..8fd3fa67b 100644 --- a/integration-tests/worker/test/exit-reasons.test.ts +++ b/integration-tests/worker/test/exit-reasons.test.ts @@ -24,11 +24,6 @@ test.after(async () => { const run = async (attempt) => { return new Promise((done) => { - // This used to be a plain `once` with the id checked inside it, which meant - // any completion for another run - a late one from the test before, or a - // duplicate - used up the listener and left this promise hanging for good, - // stalling the rest of the file until ava gave up. onSocketEvent only - // releases the listener once the id actually matches. lightning.onSocketEvent('run:complete', attempt.id, (evt) => { if (attempt.id === evt.runId) { done(evt.payload); diff --git a/integration-tests/worker/test/server.test.ts b/integration-tests/worker/test/server.test.ts index 3c1788f1e..3208764f9 100644 --- a/integration-tests/worker/test/server.test.ts +++ b/integration-tests/worker/test/server.test.ts @@ -219,9 +219,6 @@ test.serial("don't restore the claim loop after a sigterm", (t) => { lightning.enqueueRun(b); lightning.on('claim', () => { - // Sending the signal and the worker acting on it are not the same moment, - // so a claim already in flight can land here through no fault of the - // worker. Only count claims made after it told us it had the signal. const didReceiveSigterm = workerLogs.some((l) => l.match(/SIGTERM RECEIVED/) );