Worker: Work around lost run case (#1417)

* add workaround

* versions

* update test

Author: josephjclark

packages/ws-worker/CHANGELOG.md

@@ -1,5 +1,11 @@
 # ws-worker
 
+## 1.25.1
+
+### Patch Changes
+
+- Add workaround for rare event duplication
+
 ## 1.25.0
 
 ### Minor Changes

packages/ws-worker/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@openfn/ws-worker",
-  "version": "1.25.0",
+  "version": "1.25.1",
   "description": "A Websocket Worker to connect Lightning to a Runtime Engine",
   "main": "dist/index.js",
   "type": "module",

packages/ws-worker/src/events/step-complete.ts

@@ -22,6 +22,20 @@ export default async function onStepComplete(
   const step_id = state.activeStep as string;
   const job_id = state.activeJob as string;
 
+  if (!step_id) {
+    // Runs are lost in production now because sometimes the step-complete
+    // event gets triggered twice. In this case, the second one does not
+    // have an activeStep on the state object, so will send with
+    // step_id null. And lightning will complain (rightly!) and refuse
+    // to listen to subsequent events on the run (wrongly!)
+    // This is hard to diagnose in the wild so as a temporary measure,
+    // we're going to abort with a strong warning
+    context.logger?.warn(
+      `DUPLICATE_EVENT_ERROR: Run ${context.id} received two step:complete events for the same step. The second event has been suppressed to prevent a lost run.`
+    );
+    return;
+  }
+
   if (!state.dataclips) {
     state.dataclips = {};
   }

packages/ws-worker/test/api/execute.test.ts

@@ -206,6 +206,7 @@ test('jobError should trigger step:complete with a reason and default state', as
   let stepCompleteEvent: any;
 
   const state = createRunState({ id: 'run-23' } as ExecutionPlan);
+  state.activeStep = 'a';
 
   const channel = mockChannel({
     [STEP_COMPLETE]: (evt) => {