diff --git a/integration-tests/cli/CHANGELOG.md b/integration-tests/cli/CHANGELOG.md index 938e2c38e..29228bc52 100644 --- a/integration-tests/cli/CHANGELOG.md +++ b/integration-tests/cli/CHANGELOG.md @@ -1,5 +1,12 @@ # @openfn/integration-tests-cli +## 1.0.5 + +### Patch Changes + +- Updated dependencies [c47b65f] + - @openfn/lightning-mock@2.2.3 + ## 1.0.4 ### Patch Changes diff --git a/integration-tests/cli/package.json b/integration-tests/cli/package.json index 614914c23..4c1ece191 100644 --- a/integration-tests/cli/package.json +++ b/integration-tests/cli/package.json @@ -1,7 +1,7 @@ { "name": "@openfn/integration-tests-cli", "private": true, - "version": "1.0.4", + "version": "1.0.5", "description": "CLI integration tests", "author": "Open Function Group ", "license": "ISC", diff --git a/integration-tests/worker/CHANGELOG.md b/integration-tests/worker/CHANGELOG.md index 55282c724..fbba28699 100644 --- a/integration-tests/worker/CHANGELOG.md +++ b/integration-tests/worker/CHANGELOG.md @@ -1,5 +1,15 @@ # @openfn/integration-tests-worker +## 1.0.93 + +### Patch Changes + +- Updated dependencies [5a613e7] +- Updated dependencies [3a33557] +- Updated dependencies [c47b65f] + - @openfn/ws-worker@1.14.4 + - @openfn/lightning-mock@2.2.3 + ## 1.0.92 ### Patch Changes diff --git a/integration-tests/worker/package.json b/integration-tests/worker/package.json index 1cd591dcf..915d2d240 100644 --- a/integration-tests/worker/package.json +++ b/integration-tests/worker/package.json @@ -1,7 +1,7 @@ { "name": "@openfn/integration-tests-worker", "private": true, - "version": "1.0.92", + "version": "1.0.93", "description": "Lightning WOrker integration tests", "author": "Open Function Group ", "license": "ISC", diff --git a/packages/lightning-mock/CHANGELOG.md b/packages/lightning-mock/CHANGELOG.md index efb2dbbe8..5f740eb6a 100644 --- a/packages/lightning-mock/CHANGELOG.md +++ b/packages/lightning-mock/CHANGELOG.md @@ -1,5 +1,11 @@ # @openfn/lightning-mock +## 2.2.3 + +### Patch Changes + +- c47b65f: Ensure json payloads respect sockey delay + ## 2.2.2 ### Patch Changes diff --git a/packages/lightning-mock/package.json b/packages/lightning-mock/package.json index fb5ff4fdb..4fcba93d2 100644 --- a/packages/lightning-mock/package.json +++ b/packages/lightning-mock/package.json @@ -1,6 +1,6 @@ { "name": "@openfn/lightning-mock", - "version": "2.2.2", + "version": "2.2.3", "private": true, "description": "A mock Lightning server", "main": "dist/index.js", diff --git a/packages/lightning-mock/src/socket-server.ts b/packages/lightning-mock/src/socket-server.ts index cd30c4549..4a0746ba4 100644 --- a/packages/lightning-mock/src/socket-server.ts +++ b/packages/lightning-mock/src/socket-server.ts @@ -190,8 +190,10 @@ function createServer({ topic, payload: stringify(payload), // TODO do we stringify this? All of it? }); - // @ts-ignore - ws.send(evt); + setTimeout(() => { + // @ts-ignore + ws.send(evt); + }, socketDelay); }; ws.on('message', async function (data: string) { diff --git a/packages/lightning-mock/test.js b/packages/lightning-mock/test.js new file mode 100644 index 000000000..99a54d361 --- /dev/null +++ b/packages/lightning-mock/test.js @@ -0,0 +1,27 @@ +let id = 0; + +const wf = { + id: 'my-attempt', + triggers: [], + edges: [], + jobs: [ + { + id: 'job1', + adaptor: '@openfn/language-common@1.7.7', + body: 'fn(() => new Promise((resolve) => setTimeout(resolve, 5000)) )', + }, + ], +}; + +// setInterval(() => { +fetch('http://localhost:8888/run', { + method: 'POST', + body: JSON.stringify({ + ...wf, + id: `${++id}`, + }), + headers: { + 'content-type': 'application/json', + }, +}); +// }, 500); diff --git a/packages/runtime/test/runtime.test.ts b/packages/runtime/test/runtime.test.ts index 08f5ef29e..209fdf865 100644 --- a/packages/runtime/test/runtime.test.ts +++ b/packages/runtime/test/runtime.test.ts @@ -643,7 +643,7 @@ test('data can be an array (workflow)', async (t) => { t.deepEqual(result.data, [1, 2, 3]); }); -test.only('import from a module', async (t) => { +test('import from a module', async (t) => { const expression = ` import { x } from 'x'; export default [(s) => ({ data: x })]; diff --git a/packages/ws-worker/CHANGELOG.md b/packages/ws-worker/CHANGELOG.md index a60af26a1..66bcfdd1b 100644 --- a/packages/ws-worker/CHANGELOG.md +++ b/packages/ws-worker/CHANGELOG.md @@ -1,5 +1,12 @@ # ws-worker +## 1.14.4 + +### Patch Changes + +- 5a613e7: Adjust logging +- 3a33557: Count outstanding claim requests as capacity. This fixes an issue where `work-available` messages can cause a worker to over-claim, particularly during periods of high load on the Lightning database. + ## 1.14.3 ### Patch Changes diff --git a/packages/ws-worker/package.json b/packages/ws-worker/package.json index 47dbb03c6..e2aa052d7 100644 --- a/packages/ws-worker/package.json +++ b/packages/ws-worker/package.json @@ -1,6 +1,6 @@ { "name": "@openfn/ws-worker", - "version": "1.14.3", + "version": "1.14.4", "description": "A Websocket Worker to connect Lightning to a Runtime Engine", "main": "dist/index.js", "type": "module", diff --git a/packages/ws-worker/src/api/claim.ts b/packages/ws-worker/src/api/claim.ts index 8cd782d24..7a64e0d84 100644 --- a/packages/ws-worker/src/api/claim.ts +++ b/packages/ws-worker/src/api/claim.ts @@ -24,6 +24,7 @@ export const verifyToken = async (token: string, publicKey: string) => { type ClaimOptions = { maxWorkers?: number; + demand?: number; }; // used to report the pod name in logging, for tracking @@ -38,20 +39,41 @@ class ClaimError extends Error { } } +let claimIdGen = 0; + +export const resetClaimIdGen = () => { + claimIdGen = 0; +}; + const claim = ( app: ServerApp, logger: Logger = mockLogger, options: ClaimOptions = {} ) => { return new Promise((resolve, reject) => { - const { maxWorkers = 5 } = options; + app.openClaims ??= {}; + + const { maxWorkers = 5, demand = 1 } = options; const podName = NAME ? `[${NAME}] ` : ''; const activeWorkers = Object.keys(app.workflows).length; + + const pendingClaims = Object.values(app.openClaims).reduce( + (a, b) => a + b, + 0 + ); + if (activeWorkers >= maxWorkers) { // Important: stop the workloop so that we don't try and claim any more app.workloop?.stop(`server at capacity (${activeWorkers}/${maxWorkers})`); return reject(new ClaimError('Server at capacity')); + } else if (activeWorkers + pendingClaims >= maxWorkers) { + // There are active claims which haven't yet been fulfilled + // This can happen in response to the work-available event + app.workloop?.stop( + `server at capacity (${activeWorkers}/${maxWorkers}, ${pendingClaims} pending)` + ); + return reject(new ClaimError('Server at capacity')); } if (!app.queueChannel) { @@ -67,15 +89,20 @@ const claim = ( return reject(e); } + const claimId = ++claimIdGen; + + app.openClaims[claimId] = demand; + logger.debug(`requesting run (capacity ${activeWorkers}/${maxWorkers})`); const start = Date.now(); app.queueChannel .push(CLAIM, { - demand: 1, + demand, worker_name: NAME || null, }) .receive('ok', ({ runs }: ClaimReply) => { + delete app.openClaims[claimId]; const duration = Date.now() - start; logger.debug( `${podName}claimed ${runs.length} runs in ${duration}ms (${ @@ -114,9 +141,12 @@ const claim = ( // TODO need implementations for both of these really // What do we do if we fail to join the worker channel? .receive('error', (e) => { + delete app.openClaims[claimId]; logger.error('Error on claim', e); + reject(new Error('claim error')); }) .receive('timeout', () => { + delete app.openClaims[claimId]; logger.error('TIMEOUT on claim. Runs may be lost.'); reject(new Error('timeout')); }); diff --git a/packages/ws-worker/src/channels/run.ts b/packages/ws-worker/src/channels/run.ts index 315d1adef..811ff40f1 100644 --- a/packages/ws-worker/src/channels/run.ts +++ b/packages/ws-worker/src/channels/run.ts @@ -60,11 +60,10 @@ const joinRunChannel = ( // channel was explicitly closed by the client or server logger.debug(`Leaving ${channelName}`); }); - channel.onError((e: any) => { + channel.onError((...args: any) => { // Error occurred on the channel // (the socket will try to reconnect with backoff) - logger.debug(`Error in ${channelName}`); - logger.debug(e); + logger.debug(`Critical error in channel ${channelName}`, args); }); }); }; diff --git a/packages/ws-worker/src/server.ts b/packages/ws-worker/src/server.ts index d7fa9db14..ab23d7e23 100644 --- a/packages/ws-worker/src/server.ts +++ b/packages/ws-worker/src/server.ts @@ -60,6 +60,7 @@ export interface ServerApp extends Koa { socket?: any; queueChannel?: Channel; workflows: Record; + openClaims: Record; destroyed: boolean; events: EventEmitter; server: Server; @@ -221,6 +222,7 @@ function createServer(engine: RuntimeEngine, options: ServerOptions = {}) { }) ); + app.openClaims = {}; app.workflows = {}; app.destroyed = false; @@ -273,7 +275,7 @@ function createServer(engine: RuntimeEngine, options: ServerOptions = {}) { collectionsVersion: app.options.collectionsVersion, monorepoPath: app.options.monorepoDir, }); - logger.debug('converted run body into execution plan:', plan); + //logger.debug('converted run body into execution plan:', plan); // Setup collections if (plan.workflow.credentials?.collections_token) { diff --git a/packages/ws-worker/test/api/claim.test.ts b/packages/ws-worker/test/api/claim.test.ts index fb9b36d2a..4bd197993 100644 --- a/packages/ws-worker/test/api/claim.test.ts +++ b/packages/ws-worker/test/api/claim.test.ts @@ -2,8 +2,12 @@ import test from 'ava'; import * as jose from 'jose'; import crypto from 'node:crypto'; -import { verifyToken } from '../../src/api/claim'; +import claim, { resetClaimIdGen, verifyToken } from '../../src/api/claim'; import { generateKeys } from '@openfn/lightning-mock/src/util'; +import { createMockLogger } from '@openfn/logger'; +import { ServerApp } from '../../src/server'; +import { mockChannel } from '../../src/mock/sockets'; +import { CLAIM } from '../../src'; let keys = { public: '.', private: '.' }; @@ -11,6 +15,10 @@ test.before(async () => { keys = await generateKeys(); }); +test.beforeEach(() => { + resetClaimIdGen(); +}); + // Helper function to generate a token with custom iat and nbf offsets const generateToken = async ( runId: string, @@ -117,3 +125,246 @@ test('verifyToken should accept a token with NBF exactly 2 seconds in future (us '✅ Token that is 2 seconds off is accepted within 5-second clock tolerance' ); }); + +const createMockApp = (opts: any) => { + const { + onClaim = () => ({ runs: [] }), + onExecute = () => {}, + workflows = {}, + } = opts; + + const channel = mockChannel({ + [CLAIM]: (args) => { + return onClaim(args); + }, + }); + + return { + workloop: undefined, // should be safe + openClaims: {}, + workflows, + queueChannel: channel, + execute: (...args) => { + onExecute(...args); + }, + } as ServerApp; +}; +const logger = createMockLogger(); + +test.todo('claim: should do nothing if no runs returned'); + +test('claim: should call execute for a single run', async (t) => { + let executeArgs: any[] = []; + const onClaim = () => ({ runs: [{ id: 'abc' }] }); + const onExecute = (...args: any[]) => { + executeArgs = args; + }; + + const options = { maxWorkers: 1 }; + const app = createMockApp({ + onClaim, + onExecute, + }); + + await claim(app, logger, options); + t.deepEqual(executeArgs[0], { id: 'abc' }); +}); + +test('should not claim if worker is at capacity', async (t) => { + const options = { maxWorkers: 1 }; + + const app = createMockApp({ + workflows: { + a: true, + }, + }); + + await t.throwsAsync(() => claim(app, logger, options), { + message: 'Server at capacity', + }); +}); + +test('should mark a claim when in flight', async (t) => { + const options = {}; + + const app = createMockApp({ + workflows: {}, + }); + + let claimPromise = claim(app, logger, options); + + t.is(app.openClaims['1'], 1); + t.is(Object.keys(app.openClaims).length, 1); + + await t.throwsAsync(claimPromise, { + message: 'No runs returned', + }); +}); + +test('should remove an open claim when completed', async (t) => { + const options = {}; + + const app = createMockApp({ + workflows: {}, + }); + + await t.throwsAsync(() => claim(app, logger, options), { + message: 'No runs returned', + }); + + t.falsy(app.openClaims['1']); + t.is(Object.keys(app.openClaims).length, 0); +}); + +test('should remove an open claim on error', async (t) => { + const options = {}; + + const app = createMockApp({ + workflows: {}, + onClaim: () => { + throw {}; + }, + }); + + await t.throwsAsync(() => claim(app, logger, options), { + message: 'claim error', + }); + + t.falsy(app.openClaims['1']); + t.is(Object.keys(app.openClaims).length, 0); +}); + +// TODO not really sure how to check this +test.skip('should remove an open claim on timeout', async (t) => { + const options = {}; + + const app = createMockApp({ + workflows: {}, + onClaim: async () => { + await new Promise((resolve) => setTimeout(resolve, 1000 * 60 * 60)); + }, + }); + + await t.throwsAsync(() => claim(app, logger, options), { + message: 'timeout', + }); + + t.falsy(app.openClaims['1']); + t.is(Object.keys(app.openClaims).length, 0); +}); + +test('should mark a claim when in flight with demand: 2', async (t) => { + const options = { + demand: 2, + }; + + const app = createMockApp({ + workflows: { + a: true, + }, + }); + + let claimPromise = claim(app, logger, options); + + t.is(app.openClaims['1'], 2); + t.is(Object.keys(app.openClaims).length, 1); + + await t.throwsAsync(claimPromise, { + message: 'No runs returned', + }); +}); + +test('should not claim if open claims exceeds capacity', async (t) => { + let didStopWorkloop = false; + + const options = { + maxWorkers: 1, + demand: 1, + }; + + const app = createMockApp({ + workflows: {}, + // Slow claim event + onClaim: () => + new Promise((resolve) => { + // @ts-ignore + setTimeout(resolve({ runs: [] }), 100); + }), + }); + + app.workloop = { + stop: () => { + didStopWorkloop = true; + }, + } as any; + + // @ts-ignore + app.execute = ({ id }) => { + app.workflows[id] = true; + }; + + // first claim should be fine + let claimPromise = claim(app, logger, options); + + // second claim should error and stop the loop actually + await t.throwsAsync(() => claim(app, logger, options), { + message: 'Server at capacity', + }); + t.true(didStopWorkloop); + + // The prior claim should not have counted for anything + t.is(Object.keys(app.workflows).length, 0); + + // wait for the original return to finish + await t.throwsAsync(claimPromise, { + message: 'No runs returned', + }); +}); + +test('should not claim if open claims + active runs exceeds capacity', async (t) => { + const options = { + maxWorkers: 2, + demand: 1, + }; + + const app = createMockApp({ + workflows: { + // pretend a run is in progress + a: true, + }, + // Slow claim event + onClaim: () => + new Promise((resolve) => { + // @ts-ignore + setTimeout(resolve({ runs: [] }), 100); + }), + }); + + // @ts-ignore + app.execute = ({ id }) => { + app.workflows[id] = true; + }; + + // first claim should be fine + let claimPromise = claim(app, logger, options); + + // second claim should error + await t.throwsAsync(() => claim(app, logger, options), { + message: 'Server at capacity', + }); + + // The prior claim should not have counted for anything + t.is(Object.keys(app.workflows).length, 1); + + // wait for the original return to finish + await t.throwsAsync(claimPromise, { + message: 'No runs returned', + }); +}); + +test.todo('should handle multiple runs'); +test.todo('claim payload should have a demand'); +test.todo('claim payload should include a worker name'); +test.todo('should stop the workloop if at capacity'); +// TODO I'd rather return true/false really and let the backoff itself decide whether to throw or not +test.todo('should throw if there are no runs available (to trigger backoff)'); diff --git a/packages/ws-worker/test/api/workloop.test.ts b/packages/ws-worker/test/api/workloop.test.ts index f67e5afb4..6a920e9af 100644 --- a/packages/ws-worker/test/api/workloop.test.ts +++ b/packages/ws-worker/test/api/workloop.test.ts @@ -16,7 +16,7 @@ test.afterEach(() => { test('workloop can be cancelled', async (t) => { let count = 0; - + const app = { workflows: {}, queueChannel: mockChannel({ @@ -30,12 +30,12 @@ test('workloop can be cancelled', async (t) => { }; workloop = startWorkloop(app as any, logger, 1, 1); - t.false(workloop.isStopped()) + t.false(workloop.isStopped()); await sleep(100); // A quirk of how cancel works is that the loop will be called a few times t.true(count <= 5); - t.true(workloop.isStopped()) + t.true(workloop.isStopped()); }); test('workloop sends the runs:claim event', (t) => { diff --git a/packages/ws-worker/test/wake-up.test.ts b/packages/ws-worker/test/wake-up.test.ts index 44932a148..d148da9e2 100644 --- a/packages/ws-worker/test/wake-up.test.ts +++ b/packages/ws-worker/test/wake-up.test.ts @@ -23,12 +23,13 @@ const urls = { const ONE_HOUR = 1000 * 60 * 60; let rollingRunId = 1; -const getRun = (): LightningPlan => + +const getRun = (body = 'fn(s => s)'): LightningPlan => ({ id: `a${++rollingRunId}`, jobs: [ { - body: 'fn(s => s)', + body, }, ], } as LightningPlan); @@ -37,6 +38,7 @@ test.before(async () => { engine = await createMockRTE(); lng = createLightningServer({ port: 7654, + socketDelay: 10, }); createWorkerServer(engine, { @@ -56,10 +58,14 @@ test.before(async () => { }); }); +test.afterEach(() => { + lng.reset(); +}); + const test_timeout = 150; // control test! -test('should not initiate a quick claim', async (t) => { +test.serial('should not initiate a quick claim', async (t) => { t.plan(1); t.timeout(test_timeout); return new Promise(async (done) => { @@ -72,6 +78,9 @@ test('should not initiate a quick claim', async (t) => { const run = getRun(); lng.onSocketEvent(e.CLAIM, run.id, () => { t.fail('unexpected claim received'); + }); + + lng.onSocketEvent(e.RUN_COMPLETE, run.id, () => { done(); }); @@ -87,12 +96,15 @@ test('should not initiate a quick claim', async (t) => { }); // main test -test('should initiate a claim when /run?wakeup=true', async (t) => { +test.serial.skip('should initiate a claim when /run?wakeup=true', async (t) => { t.plan(1); return new Promise(async (done) => { const run = getRun(); lng.onSocketEvent(e.CLAIM, run.id, () => { t.pass('claim happened on lightning'); + }); + + lng.onSocketEvent(e.RUN_COMPLETE, run.id, () => { done(); }); @@ -106,3 +118,68 @@ test('should initiate a claim when /run?wakeup=true', async (t) => { }); }); }); + +// This tests creates two runs, call wake_up on them both, +// and ensures they are executed sequentially +// This doesn't feel like a great test, but if the pending claims validation +// is removed from claim.ts, it fails. So it is kinda working +test.serial('should not claim beyond capacity', async (t) => { + t.plan(2); + let claimCount = 0; + return new Promise(async (done) => { + const run_1 = getRun( + 's => new Promise(resolve => setTimeout(resolve, 200))' + ); + const run_2 = getRun( + 's => new Promise(resolve => setTimeout(resolve, 200))' + ); + let run_1_finished = false; + + lng.onSocketEvent(e.RUN_COMPLETE, run_1.id, (evt) => { + if (evt.runId === run_1.id) { + t.log(' RUN 1 complete'); + run_1_finished = true; + } + }); + + lng.onSocketEvent(e.RUN_START, run_2.id, (evt) => { + if (evt.runId === run_2.id) { + t.log(' RUN 2 start'); + // Ensure that run 1 is finished when run 2 starts + t.true(run_1_finished); + } + }); + + // TODO there's a terrible bug in the mock and these events don't bind right + lng.onSocketEvent(e.RUN_COMPLETE, run_2.id, (evt) => { + if (evt.runId === run_2.id) { + t.log(' RUN 2 complete'); + // Only 2 claims should be made + t.is(claimCount, 2); + done(); + } + }); + + lng.onSocketEvent(e.CLAIM, run_1.id, () => { + claimCount++; + }); + + // Send two runs up at the same time + fetch(`${urls.lngServer}/run?wakeup=true`, { + method: 'POST', + body: JSON.stringify(run_1), + headers: { + Accept: 'application/json', + 'Content-Type': 'application/json', + }, + }); + fetch(`${urls.lngServer}/run?wakeup=true`, { + method: 'POST', + body: JSON.stringify(run_2), + headers: { + Accept: 'application/json', + 'Content-Type': 'application/json', + }, + }); + }); +});