Remove the runlock from the trigger path

Author: ericallam

internal-packages/run-engine/src/engine/index.ts

@@ -486,56 +486,52 @@ export class RunEngine {
 
         span.setAttribute("runId", taskRun.id);
 
-        await this.runLock.lock("trigger", [taskRun.id], 5000, async (signal) => {
-          //create associated waitpoint (this completes when the run completes)
-          const associatedWaitpoint = await this.waitpointSystem.createRunAssociatedWaitpoint(
-            prisma,
-            {
-              projectId: environment.project.id,
-              environmentId: environment.id,
-              completedByTaskRunId: taskRun.id,
-            }
-          );
-
-          //triggerAndWait or batchTriggerAndWait
-          if (resumeParentOnCompletion && parentTaskRunId) {
-            //this will block the parent run from continuing until this waitpoint is completed (and removed)
-            await this.waitpointSystem.blockRunWithWaitpoint({
-              runId: parentTaskRunId,
-              waitpoints: associatedWaitpoint.id,
-              projectId: associatedWaitpoint.projectId,
-              organizationId: environment.organization.id,
-              batch,
-              workerId,
-              runnerId,
-              tx: prisma,
-              releaseConcurrency,
-            });
+        //create associated waitpoint (this completes when the run completes)
+        const associatedWaitpoint = await this.waitpointSystem.createRunAssociatedWaitpoint(
+          prisma,
+          {
+            projectId: environment.project.id,
+            environmentId: environment.id,
+            completedByTaskRunId: taskRun.id,
           }
+        );
 
-          //Make sure lock extension succeeded
-          signal.throwIfAborted();
-
-          if (taskRun.delayUntil) {
-            // Schedule the run to be enqueued at the delayUntil time
-            await this.delayedRunSystem.scheduleDelayedRunEnqueuing({
-              runId: taskRun.id,
-              delayUntil: taskRun.delayUntil,
-            });
-          } else {
-            await this.enqueueSystem.enqueueRun({
-              run: taskRun,
-              env: environment,
-              workerId,
-              runnerId,
-              tx: prisma,
-            });
+        //triggerAndWait or batchTriggerAndWait
+        if (resumeParentOnCompletion && parentTaskRunId) {
+          //this will block the parent run from continuing until this waitpoint is completed (and removed)
+          await this.waitpointSystem.blockRunWithWaitpoint({
+            runId: parentTaskRunId,
+            waitpoints: associatedWaitpoint.id,
+            projectId: associatedWaitpoint.projectId,
+            organizationId: environment.organization.id,
+            batch,
+            workerId,
+            runnerId,
+            tx: prisma,
+            releaseConcurrency,
+          });
+        }
 
-            if (taskRun.ttl) {
-              await this.ttlSystem.scheduleExpireRun({ runId: taskRun.id, ttl: taskRun.ttl });
-            }
+        if (taskRun.delayUntil) {
+          // Schedule the run to be enqueued at the delayUntil time
+          await this.delayedRunSystem.scheduleDelayedRunEnqueuing({
+            runId: taskRun.id,
+            delayUntil: taskRun.delayUntil,
+          });
+        } else {
+          if (taskRun.ttl) {
+            await this.ttlSystem.scheduleExpireRun({ runId: taskRun.id, ttl: taskRun.ttl });
           }
-        });
+
+          await this.enqueueSystem.enqueueRun({
+            run: taskRun,
+            env: environment,
+            workerId,
+            runnerId,
+            tx: prisma,
+            skipRunLock: true,
+          });
+        }
 
         this.eventBus.emit("runCreated", {
           time: new Date(),

internal-packages/run-engine/src/engine/locking.ts

@@ -165,6 +165,20 @@ export class RunLocker {
     );
   }
 
+  async lockIf<T>(
+    condition: boolean,
+    name: string,
+    resources: string[],
+    duration: number,
+    routine: () => Promise<T>
+  ): Promise<T> {
+    if (condition) {
+      return this.lock(name, resources, duration, routine);
+    } else {
+      return routine();
+    }
+  }
+
   isInsideLock(): boolean {
     return !!this.asyncLocalStorage.getStore();
   }

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

@@ -33,6 +33,7 @@ export class EnqueueSystem {
     completedWaitpoints,
     workerId,
     runnerId,
+    skipRunLock,
   }: {
     run: TaskRun;
     env: MinimalAuthenticatedEnvironment;
@@ -51,10 +52,11 @@ export class EnqueueSystem {
     }[];
     workerId?: string;
     runnerId?: string;
+    skipRunLock?: boolean;
   }) {
     const prisma = tx ?? this.$.prisma;
 
-    return await this.$.runLock.lock("enqueueRun", [run.id], 5000, async () => {
+    return await this.$.runLock.lockIf(!skipRunLock, "enqueueRun", [run.id], 5000, async () => {
       const newSnapshot = await this.executionSnapshotSystem.createExecutionSnapshot(prisma, {
         run: run,
         snapshot: {