ensure we only execute after the run creation tx has completed

Author: nicktrn

apps/webapp/app/v3/marqs/sharedQueueConsumer.server.ts

@@ -295,6 +295,9 @@ export class SharedQueueConsumer {
             messageId: message.messageId,
           });
 
+          // INFO: There used to be a race condition where tasks could be triggered, but execute messages could be dequeued before the run finished being created in the DB
+          //       This should not be happening anymore. In case it does, consider reqeueuing here with a brief delay while limiting total retries.
+
           await this.#ackAndDoMoreWork(message.messageId);
           return;
         }

apps/webapp/app/v3/services/triggerTask.server.ts

@@ -71,7 +71,7 @@ export class TriggerTaskService extends BaseService {
         async (event, traceContext) => {
           const lockId = taskIdentifierToLockId(taskId);
 
-          return await $transaction(this._prisma, async (tx) => {
+          const run = await $transaction(this._prisma, async (tx) => {
             await tx.$executeRaw`SELECT pg_advisory_xact_lock(${lockId})`;
 
             const lockedToBackgroundWorker = body.options?.lockToVersion
@@ -169,17 +169,23 @@ export class TriggerTaskService extends BaseService {
               }
             }
 
-            // We need to enqueue the task run into the appropriate queue
-            await marqs?.enqueueMessage(
-              environment,
-              queueName,
-              taskRun.id,
-              { type: "EXECUTE", taskIdentifier: taskId },
-              body.options?.concurrencyKey
-            );
-
             return taskRun;
           });
+
+          if (!run) {
+            return;
+          }
+
+          // We need to enqueue the task run into the appropriate queue. This is done after the tx completes to prevent a race condition where the task run hasn't been created yet by the time we dequeue.
+          await marqs?.enqueueMessage(
+            environment,
+            run.queue,
+            run.id,
+            { type: "EXECUTE", taskIdentifier: taskId },
+            body.options?.concurrencyKey
+          );
+
+          return run;
         }
       );
     });