diff --git a/.changeset/khaki-jars-relax.md b/.changeset/khaki-jars-relax.md new file mode 100644 index 000000000..3975f7e6c --- /dev/null +++ b/.changeset/khaki-jars-relax.md @@ -0,0 +1,5 @@ +--- +'@openfn/ws-worker': patch +--- + +Fail a run cleanly instead of hanging when Lightning rejects the plan fetch after a claim diff --git a/packages/ws-worker/src/channels/run.ts b/packages/ws-worker/src/channels/run.ts index 2c2a95ec7..3cc150aef 100644 --- a/packages/ws-worker/src/channels/run.ts +++ b/packages/ws-worker/src/channels/run.ts @@ -37,11 +37,16 @@ const joinRunChannel = ( if (!didReceiveOk) { didReceiveOk = true; logger.success(`connected to ${channelName}`, e); - const run = await sendEvent( - { channel, logger, id: runId, options: {} }, - GET_PLAN - ); - resolve({ channel, run }); + try { + const run = await sendEvent( + { channel, logger, id: runId, options: {} }, + GET_PLAN + ); + resolve({ channel, run }); + } catch (err) { + channel?.leave(); + reject(err); + } } }) .receive('error', (err: any) => { diff --git a/packages/ws-worker/test/channels/run.test.ts b/packages/ws-worker/test/channels/run.test.ts index 0f631ed1a..fc872d962 100644 --- a/packages/ws-worker/test/channels/run.test.ts +++ b/packages/ws-worker/test/channels/run.test.ts @@ -44,6 +44,30 @@ test('should fail to join an run channel with an invalid token', async (t) => { } }); +test('should reject (not hang) when fetch:plan replies an error', async (t) => { + const logger = createMockLogger(); + const socket = new MockSocket('www', { + 'run:a': mockChannel({ + join: () => ({ status: 'ok' }), + [GET_PLAN]: () => { + throw { reason: 'adaptor_not_ready' }; + }, + }), + }); + + // race against a short timeout so a hang fails fast instead of hanging CI + const result = await Promise.race([ + joinRunChannel(socket, 'x.y.z', 'a', logger).then( + () => 'resolved', + () => 'rejected' + ), + new Promise((resolve) => setTimeout(() => resolve('timed-out'), 200)), + ]); + + // 'timed-out' would mean joinRunChannel's promise never settled + t.is(result, 'rejected'); +}); + test('should log an error including channel state when the channel errors', async (t) => { const logger = createMockLogger(); const channel = mockChannel({