fix: release concurrency token exhaustion no longer happens on batchTriggerAndWait

Author: ericallam

internal-packages/run-engine/src/engine/systems/waitpointSystem.ts

@@ -430,6 +430,11 @@ export class WaitpointSystem {
 
         // Let the worker know immediately, so it can suspend the run
         await sendNotificationToWorker({ runId, snapshot, eventBus: this.$.eventBus });
+
+        if (isRunBlocked) {
+          //release concurrency
+          await this.releaseConcurrencySystem.releaseConcurrencyForSnapshot(snapshot);
+        }
       }
 
       if (timeout) {
@@ -448,10 +453,7 @@ export class WaitpointSystem {
 
       //no pending waitpoint, schedule unblocking the run
       //debounce if we're rapidly adding waitpoints
-      if (isRunBlocked) {
-        //release concurrency
-        await this.releaseConcurrencySystem.releaseConcurrencyForSnapshot(snapshot);
-      } else {
+      if (!isRunBlocked) {
         await this.$.worker.enqueue({
           //this will debounce the call
           id: `continueRunIfUnblocked:${runId}`,

references/hello-world/src/trigger/release-concurrency.ts

@@ -147,3 +147,24 @@ export const waitReleaseConcurrencyTestTask = task({
     };
   },
 });
+
+export const batchTriggerAndWaitReleaseConcurrency = task({
+  id: "batch-trigger-and-wait-release-concurrency",
+  retry: {
+    maxAttempts: 1,
+  },
+  run: async (payload, { ctx }) => {
+    return await batch.triggerAndWait([
+      { id: batchTriggerAndWaitChildTask.id, payload: { waitSeconds: 1 } },
+      { id: batchTriggerAndWaitChildTask.id, payload: { waitSeconds: 1 } },
+      { id: batchTriggerAndWaitChildTask.id, payload: { waitSeconds: 1 } },
+    ]);
+  },
+});
+
+const batchTriggerAndWaitChildTask = task({
+  id: "batch-trigger-and-wait-child-task",
+  run: async (payload: { waitSeconds: number }, { ctx }) => {
+    await setTimeout(payload.waitSeconds * 1000);
+  },
+});