From a94e062ad155c961ca4620a2c7c1740e4aecb1e2 Mon Sep 17 00:00:00 2001
From: Stuart Corbishley <corbish@gmail.com>
Date: Wed, 2 Sep 2026 11:25:18 +0200
Subject: [PATCH] Worker: don't hang when fetch:plan replies with an error

joinRunChannel awaited sendEvent(GET_PLAN) inside an unguarded async
callback, so a rejection (now possible since Lightning can reply an
error to fetch:plan for adaptor resolution failures) became an
unhandled rejection instead of settling the outer promise. The
worker's local capacity slot for that run leaked until restart.

Wrap the await in try/catch and reject, mirroring the existing
error/timeout branches in the same file.
---
 .changeset/khaki-jars-relax.md               |  5 ++++
 packages/ws-worker/src/channels/run.ts       | 15 ++++++++----
 packages/ws-worker/test/channels/run.test.ts | 24 ++++++++++++++++++++
 3 files changed, 39 insertions(+), 5 deletions(-)
 create mode 100644 .changeset/khaki-jars-relax.md

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