Add unique index to batch task run errors on the index field and more efficiently create errors in the completion callback

Author: ericallam

apps/webapp/app/v3/runEngineHandlers.server.ts

@@ -746,35 +746,38 @@ export function setupBatchQueueCallbacks() {
     }
 
     try {
-      // Update BatchTaskRun
-      await prisma.batchTaskRun.update({
-        where: { id: batchId },
-        data: {
-          status,
-          runIds,
-          successfulRunCount,
-          failedRunCount,
-          completedAt: status === "ABORTED" ? new Date() : undefined,
-          processingCompletedAt: new Date(),
-        },
-      });
+      // Use a transaction to ensure atomicity of batch update and error record creation
+      // skipDuplicates handles idempotency when callback is retried (relies on unique constraint)
+      await prisma.$transaction(async (tx) => {
+        // Update BatchTaskRun
+        await tx.batchTaskRun.update({
+          where: { id: batchId },
+          data: {
+            status,
+            runIds,
+            successfulRunCount,
+            failedRunCount,
+            completedAt: status === "ABORTED" ? new Date() : undefined,
+            processingCompletedAt: new Date(),
+          },
+        });
 
-      // Create error records if there were failures
-      if (failures.length > 0) {
-        for (const failure of failures) {
-          await prisma.batchTaskRunError.create({
-            data: {
+        // Create error records if there were failures
+        if (failures.length > 0) {
+          await tx.batchTaskRunError.createMany({
+            data: failures.map((failure) => ({
               batchTaskRunId: batchId,
               index: failure.index,
               taskIdentifier: failure.taskIdentifier,
               payload: failure.payload,
               options: failure.options as Prisma.InputJsonValue | undefined,
               error: failure.error,
               errorCode: failure.errorCode,
-            },
+            })),
+            skipDuplicates: true,
           });
         }
-      }
+      });
 
       // Try to complete the batch (handles waitpoint completion if all runs are done)
       if (status !== "ABORTED") {

internal-packages/database/prisma/migrations/20251213093011_add_batch_task_run_error_unique_index/migration.sql

@@ -0,0 +1,8 @@
+/*
+  Warnings:
+
+  - A unique constraint covering the columns `[batchTaskRunId,index]` on the table `BatchTaskRunError` will be added. If there are existing duplicate values, this will fail.
+
+*/
+-- CreateIndex
+CREATE UNIQUE INDEX "BatchTaskRunError_batchTaskRunId_index_key" ON "public"."BatchTaskRunError"("batchTaskRunId", "index");

internal-packages/database/prisma/schema.prisma

@@ -1661,6 +1661,7 @@ model BatchTaskRunError {
 
   createdAt DateTime @default(now())
 
+  @@unique([batchTaskRunId, index])
   @@index([batchTaskRunId])
 }