perf(webapp): avoid task registration conflict errors

Mono-RevId: fc08d78116292c16f3f9fc2145a2f6ce74bb11d0

Author: carderne

apps/webapp/app/v3/services/createBackgroundWorker.server.test.ts

@@ -1,11 +1,13 @@
 import { ScheduleEngine } from "@internal/schedule-engine";
 import { containerTest } from "@internal/testcontainers";
+import type { BackgroundWorkerMetadata } from "@trigger.dev/core/v3";
 import type { PrismaClient } from "@trigger.dev/database";
 import { describe, expect, vi } from "vitest";
 import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { FEATURE_FLAG } from "~/v3/featureFlags";
 import {
   CreateBackgroundWorkerService,
+  createWorkerResources,
   syncDeclarativeSchedules,
 } from "~/v3/services/createBackgroundWorker.server";
 
@@ -123,6 +125,22 @@ function declarativeTasks(schedule: { cron: string; timezone: string; window?: s
   return [{ id: "my-task", schedule }] as TasksArg;
 }
 
+function workerMetadata(description: string) {
+  return {
+    contentHash: "duplicate-task-content",
+    tasks: [
+      {
+        id: "duplicate-task",
+        description,
+        filePath: "src/trigger/duplicate-task.ts",
+        exportName: "duplicateTask",
+        queue: { name: "duplicate-task-queue" },
+      },
+    ],
+    queues: [{ name: "duplicate-task-queue" }],
+  } as unknown as BackgroundWorkerMetadata;
+}
+
 async function seedScheduledTask(
   prisma: PrismaClient,
   projectId: string,
@@ -152,6 +170,82 @@ async function seedScheduledTask(
   });
 }
 
+describe("worker task creation", () => {
+  containerTest(
+    "preserves one task when concurrent registration transactions target the same worker and slug",
+    async ({ prisma }) => {
+      const { project, devEnv } = await seedProjectWithEnvs(prisma);
+      const worker = await prisma.backgroundWorker.create({
+        data: {
+          friendlyId: `worker_${devEnv.id}`,
+          contentHash: "duplicate-task-content",
+          version: "20260811.1",
+          metadata: {},
+          projectId: project.id,
+          runtimeEnvironmentId: devEnv.id,
+        },
+      });
+      const queue = await prisma.taskQueue.create({
+        data: {
+          friendlyId: `queue_${devEnv.id}`,
+          name: "duplicate-task-queue",
+          type: "NAMED",
+          version: "V2",
+          paused: true,
+          projectId: project.id,
+          runtimeEnvironmentId: devEnv.id,
+        },
+      });
+      const environment = { ...asEnv(devEnv), project } as AuthenticatedEnvironment;
+
+      const entries = await Promise.all(
+        ["first registration", "second registration"].map((description) =>
+          prisma.$transaction((tx) =>
+            createWorkerResources(workerMetadata(description), worker, environment, tx)
+          )
+        )
+      );
+
+      expect(entries).toEqual([
+        [
+          {
+            slug: "duplicate-task",
+            ttl: null,
+            triggerSource: "STANDARD",
+            queueId: queue.id,
+            queueName: queue.name,
+          },
+        ],
+        [
+          {
+            slug: "duplicate-task",
+            ttl: null,
+            triggerSource: "STANDARD",
+            queueId: queue.id,
+            queueName: queue.name,
+          },
+        ],
+      ]);
+
+      const created = await prisma.backgroundWorkerTask.findUniqueOrThrow({
+        where: { workerId_slug: { workerId: worker.id, slug: "duplicate-task" } },
+      });
+      expect(["first registration", "second registration"]).toContain(created.description);
+      expect(await prisma.backgroundWorkerTask.count({ where: { workerId: worker.id } })).toBe(1);
+
+      await prisma.$transaction((tx) =>
+        createWorkerResources(workerMetadata("replacement"), worker, environment, tx)
+      );
+
+      const afterRetry = await prisma.backgroundWorkerTask.findUniqueOrThrow({
+        where: { workerId_slug: { workerId: worker.id, slug: "duplicate-task" } },
+      });
+      expect(afterRetry.id).toBe(created.id);
+      expect(afterRetry.description).toBe(created.description);
+    }
+  );
+});
+
 describe("declarative schedule preflight", () => {
   containerTest("rejects identical retries before persisting a worker", async ({ prisma }) => {
     const { project, devEnv } = await seedProjectWithEnvs(prisma);

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

@@ -410,7 +410,6 @@ async function createWorkerTask(
   prisma: PrismaClientOrTransaction,
   tasksToBackgroundFiles?: Map<string, string>
 ): Promise<TaskMetadataEntry | null> {
-  // Hoisted so the P2002 catch branch can return the same entry shape.
   let queue: TaskQueue | undefined;
   let resolvedTriggerSource: "SCHEDULED" | "AGENT" | "WEBHOOK" | "STANDARD" | undefined;
   let resolvedTtl: string | null | undefined;
@@ -445,28 +444,51 @@ async function createWorkerTask(
     resolvedTtl =
       typeof task.ttl === "number" ? (stringifyDuration(task.ttl) ?? null) : (task.ttl ?? null);
 
-    await prisma.backgroundWorkerTask.create({
-      data: {
-        friendlyId: generateFriendlyId("task"),
-        projectId: worker.projectId,
-        runtimeEnvironmentId: worker.runtimeEnvironmentId,
-        workerId: worker.id,
-        slug: task.id,
-        description: task.description,
-        filePath: task.filePath,
-        exportName: task.exportName,
-        retryConfig: task.retry,
-        queueConfig: task.queue,
-        machineConfig: task.machine,
-        triggerSource: resolvedTriggerSource,
-        config: task.agentConfig ? (task.agentConfig as any) : undefined,
-        fileId: tasksToBackgroundFiles?.get(task.id) ?? null,
-        maxDurationInSeconds: task.maxDuration ? clampMaxDuration(task.maxDuration) : null,
-        ttl: resolvedTtl,
-        queueId: queue.id,
-        payloadSchema: task.payloadSchema as any,
-      },
-    });
+    let taskPersisted = false;
+    for (let attempt = 0; attempt < 3; attempt++) {
+      const result = await prisma.backgroundWorkerTask.createMany({
+        data: {
+          friendlyId: generateFriendlyId("task"),
+          projectId: worker.projectId,
+          runtimeEnvironmentId: worker.runtimeEnvironmentId,
+          workerId: worker.id,
+          slug: task.id,
+          description: task.description,
+          filePath: task.filePath,
+          exportName: task.exportName,
+          retryConfig: task.retry,
+          queueConfig: task.queue,
+          machineConfig: task.machine,
+          triggerSource: resolvedTriggerSource,
+          config: task.agentConfig ? (task.agentConfig as any) : undefined,
+          fileId: tasksToBackgroundFiles?.get(task.id) ?? null,
+          maxDurationInSeconds: task.maxDuration ? clampMaxDuration(task.maxDuration) : null,
+          ttl: resolvedTtl,
+          queueId: queue.id,
+          payloadSchema: task.payloadSchema as any,
+        },
+        skipDuplicates: true,
+      });
+
+      if (result.count === 1) {
+        taskPersisted = true;
+        break;
+      }
+
+      const existing = await prisma.backgroundWorkerTask.findFirst({
+        where: { workerId: worker.id, slug: task.id },
+        select: { id: true },
+      });
+
+      if (existing) {
+        taskPersisted = true;
+        break;
+      }
+    }
+
+    if (!taskPersisted) {
+      throw new Error("Failed to create background worker task after unique constraint conflicts");
+    }
 
     return {
       slug: task.id,
@@ -477,53 +499,26 @@ async function createWorkerTask(
     };
   } catch (error) {
     if (error instanceof Prisma.PrismaClientKnownRequestError) {
-      // The error code for unique constraint violation in Prisma is P2002
-      if (error.code === "P2002") {
-        // Retry landing after the first attempt's row was already written.
-        const existing = await prisma.backgroundWorkerTask.findFirst({
-          where: { workerId: worker.id, slug: task.id },
-          select: { id: true },
-        });
-
-        logger.warn("Attempted to recreate background worker task", {
-          task,
-          worker,
-        });
-
-        if (existing && queue && resolvedTriggerSource && resolvedTtl !== undefined) {
-          return {
-            slug: task.id,
-            ttl: resolvedTtl,
-            triggerSource: resolvedTriggerSource,
-            queueId: queue.id,
-            queueName: queue.name,
-          };
-        }
-      } else {
-        logger.error("Prisma Error creating background worker task", {
-          error: {
-            code: error.code,
-            message: error.message,
-          },
-          task,
-          worker,
-        });
-      }
+      logger.error("Prisma Error creating background worker task", {
+        error: {
+          code: error.code,
+          message: error.message,
+        },
+        workerId: worker.id,
+      });
     } else if (error instanceof Error) {
       logger.error("Error creating background worker task", {
         error: {
           name: error.name,
           message: error.message,
           stack: error.stack,
         },
-        task,
-        worker,
+        workerId: worker.id,
       });
     } else {
       logger.error("Unknown error creating background worker task", {
         error,
-        task,
-        worker,
+        workerId: worker.id,
       });
     }
     return null;