Worker: don't hang when fetch:plan replies with an error

.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

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<GetPlanReply>(
-            { channel, logger, id: runId, options: {} },
-            GET_PLAN
-          );
-          resolve({ channel, run });
+          try {
+            const run = await sendEvent<GetPlanReply>(
+              { channel, logger, id: runId, options: {} },
+              GET_PLAN
+            );
+            resolve({ channel, run });
+          } catch (err) {
+            channel?.leave();
+            reject(err);
+          }
         }
       })
       .receive('error', (err: any) => {

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({