fix(webapp): idempotent DeploymentBackgroundWorker creation (#3772)

myftija · web-flow · commit 9cb6fd17c376 · 2026-05-29T10:03:55.000+02:00
Make `POST /api/v1/deployments/:deploymentId/background-workers`
idempotent so client-side retries no longer collide on the
`BackgroundWorker` `(project, env, version)` unique index. Helps make
deployments more resilient against the class of indexing failures that
surfaces in the dashboard as "Indexing timed out", e.g. during transient
database issues.
diff --git a/.server-changes/idempotent-background-worker-registration.md b/.server-changes/idempotent-background-worker-registration.md
@@ -0,0 +1,6 @@
+---
+area: webapp
+type: fix
+---
+
+Make `POST /api/v1/deployments/:deploymentId/background-workers` idempotent (for sequential requests only) so client-side retries no longer collide on the `BackgroundWorker` `(project, env, version)` unique index. Helps make deployments more resilient against the class of indexing failures that surfaces in the dashboard as "Indexing timed out", e.g. during transient database issues.
diff --git a/apps/webapp/app/v3/services/createBackgroundWorker.server.ts b/apps/webapp/app/v3/services/createBackgroundWorker.server.ts
@@ -34,31 +34,8 @@ import { tryCatch } from "@trigger.dev/core/v3";
 import { engine } from "../runEngine.server";
 import { scheduleEngine } from "../scheduleEngine.server";
 
-/**
- * Strip BackgroundWorkerMetadata down to the slice that's actually read after
- * storage. Everything else is duplicated to dedicated columns/tables
- * (BackgroundWorker.{contentHash,cliVersion,sdkVersion,runtime,runtimeVersion},
- * BackgroundWorkerTask, BackgroundWorkerFile, TaskQueue, Prompt). Today the
- * only post-write reader is changeCurrentDeployment.server.ts, which feeds
- * tasks[].schedule into syncDeclarativeSchedules. packageVersion, contentHash,
- * and tasks[].filePath are kept solely to satisfy BackgroundWorkerMetadata's
- * required fields when the column is parsed back.
- */
-export function stripBackgroundWorkerMetadataForStorage(
-  metadata: BackgroundWorkerMetadata
-): Prisma.InputJsonValue {
-  return {
-    packageVersion: metadata.packageVersion,
-    contentHash: metadata.contentHash,
-    tasks: metadata.tasks
-      .filter((t) => t.schedule)
-      .map((t) => ({
-        id: t.id,
-        filePath: t.filePath,
-        schedule: t.schedule,
-      })),
-  };
-}
+import { stripBackgroundWorkerMetadataForStorage } from "./stripBackgroundWorkerMetadataForStorage.server";
+export { stripBackgroundWorkerMetadataForStorage };
 
 export class CreateBackgroundWorkerService extends BaseService {
   private readonly _taskMetaCache: TaskMetadataCache;
@@ -356,8 +333,13 @@ 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" | "STANDARD" | undefined;
+  let resolvedTtl: string | null | undefined;
+
   try {
-    let queue = queues.find((queue) => queue.name === task.queue?.name);
+    queue = queues.find((queue) => queue.name === task.queue?.name);
 
     if (!queue) {
       // Create a TaskQueue
@@ -374,14 +356,14 @@ async function createWorkerTask(
       );
     }
 
-    const resolvedTriggerSource =
+    resolvedTriggerSource =
       task.triggerSource === "schedule"
         ? ("SCHEDULED" as const)
         : task.triggerSource === "agent"
           ? ("AGENT" as const)
           : ("STANDARD" as const);
 
-    const resolvedTtl =
+    resolvedTtl =
       typeof task.ttl === "number" ? stringifyDuration(task.ttl) ?? null : task.ttl ?? null;
 
     await prisma.backgroundWorkerTask.create({
@@ -418,10 +400,26 @@ async function createWorkerTask(
     if (error instanceof Prisma.PrismaClientKnownRequestError) {
       // The error code for unique constraint violation in Prisma is P2002
       if (error.code === "P2002") {
-        logger.warn("Task already exists", {
+        // 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: {
diff --git a/apps/webapp/app/v3/services/createDeploymentBackgroundWorkerV4.server.ts b/apps/webapp/app/v3/services/createDeploymentBackgroundWorkerV4.server.ts
@@ -1,16 +1,19 @@
 import { CreateBackgroundWorkerRequestBody, logger, tryCatch } from "@trigger.dev/core/v3";
-import { BackgroundWorkerId } from "@trigger.dev/core/v3/isomorphic";
-import type { BackgroundWorker, PrismaClientOrTransaction, WorkerDeployment } from "@trigger.dev/database";
+import type {
+  BackgroundWorker,
+  PrismaClientOrTransaction,
+  WorkerDeployment,
+} from "@trigger.dev/database";
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { type TaskMetadataCache } from "~/services/taskMetadataCache.server";
 import { taskMetadataCacheInstance } from "~/services/taskMetadataCacheInstance.server";
 import { BaseService, ServiceValidationError } from "./baseService.server";
 import {
   createBackgroundFiles,
   createWorkerResources,
-  stripBackgroundWorkerMetadataForStorage,
   syncDeclarativeSchedules,
 } from "./createBackgroundWorker.server";
+import { findOrCreateBackgroundWorker } from "./createDeploymentBackgroundWorkerV4/findOrCreateBackgroundWorker.server";
 import { TimeoutDeploymentService } from "./timeoutDeployment.server";
 import { env } from "~/env.server";
 
@@ -50,6 +53,11 @@ export class CreateDeploymentBackgroundWorkerServiceV4 extends BaseService {
       });
 
       if (!deployment) {
+        logger.warn("createDeploymentBackgroundWorker: deployment not found", {
+          deploymentId,
+          environmentId: environment.id,
+          projectId: environment.projectId,
+        });
         return;
       }
 
@@ -69,26 +77,44 @@ export class CreateDeploymentBackgroundWorkerServiceV4 extends BaseService {
         }
       }
 
+      // Late-retry idempotency: if a worker was registered by a prior fully-
+      // successful attempt and the deployment already moved past BUILDING, return
+      // that worker so the CLI can finalize instead of seeing a 5xx.
+      if (deployment.workerId) {
+        const linkedWorker = await this._prisma.backgroundWorker.findFirst({
+          where: { id: deployment.workerId },
+        });
+        if (linkedWorker) {
+          return linkedWorker;
+        }
+      }
+
       if (deployment.status !== "BUILDING") {
+        logger.warn("createDeploymentBackgroundWorker: deployment not in BUILDING state", {
+          deploymentId,
+          deploymentStatus: deployment.status,
+          environmentId: environment.id,
+          projectId: environment.projectId,
+        });
         return;
       }
 
-      const backgroundWorker = await this._prisma.backgroundWorker.create({
-        data: {
-          ...BackgroundWorkerId.generate(),
-          version: deployment.version,
-          runtimeEnvironmentId: environment.id,
-          projectId: environment.projectId,
-          metadata: stripBackgroundWorkerMetadataForStorage(body.metadata),
-          contentHash: body.metadata.contentHash,
-          cliVersion: body.metadata.cliPackageVersion,
-          sdkVersion: body.metadata.packageVersion,
-          supportsLazyAttempts: body.supportsLazyAttempts,
-          engine: body.engine,
-          runtime: body.metadata.runtime,
-          runtimeVersion: body.metadata.runtimeVersion,
-        },
-      });
+      const [findOrCreateError, backgroundWorker] = await tryCatch(
+        findOrCreateBackgroundWorker(environment, deployment, body, this._prisma)
+      );
+
+      if (findOrCreateError) {
+        // Definitive failures (e.g. contentHash drift) surface as
+        // `ServiceValidationError` — fail the deployment so the operator sees it
+        // immediately instead of waiting 8 minutes for the timeout. Transient
+        // races throw a plain `Error` and propagate as 5xx without failing.
+        if (findOrCreateError instanceof ServiceValidationError) {
+          // `#failBackgroundWorkerDeployment` already throws its argument; the
+          // outer `throw` covers the non-SVE branch.
+          await this.#failBackgroundWorkerDeployment(deployment, findOrCreateError);
+        }
+        throw findOrCreateError;
+      }
 
       //upgrade the project to engine "V2" if it's not already
       if (environment.project.engine === "V1" && body.engine === "V2") {
@@ -188,10 +214,11 @@ export class CreateDeploymentBackgroundWorkerServiceV4 extends BaseService {
         throw serviceError;
       }
 
-      // Link the deployment with the background worker
-      await this._prisma.workerDeployment.update({
+      // Guarded BUILDING → DEPLOYING transition. `updateMany` for optimistic concurrency control
+      const { count: updatedCount } = await this._prisma.workerDeployment.updateMany({
         where: {
           id: deployment.id,
+          status: "BUILDING",
         },
         data: {
           status: "DEPLOYING",
@@ -203,6 +230,18 @@ export class CreateDeploymentBackgroundWorkerServiceV4 extends BaseService {
         },
       });
 
+      if (updatedCount === 0) {
+        logger.warn(
+          "createDeploymentBackgroundWorker: deployment no longer in BUILDING state, skipping DEPLOYING transition",
+          {
+            deploymentId,
+            environmentId: environment.id,
+            projectId: environment.projectId,
+          }
+        );
+        return backgroundWorker;
+      }
+
       await TimeoutDeploymentService.enqueue(
         deployment.id,
         "DEPLOYING",
@@ -215,9 +254,14 @@ export class CreateDeploymentBackgroundWorkerServiceV4 extends BaseService {
   }
 
   async #failBackgroundWorkerDeployment(deployment: WorkerDeployment, error: Error) {
-    await this._prisma.workerDeployment.update({
+    // Guarded BUILDING → FAILED transition, symmetric with the BUILDING → DEPLOYING
+    // transition in `call()`. With idempotent retries, two attempts can run side-by-side;
+    // without the predicate, one attempt's failure could downgrade the deployment after
+    // the other already flipped it to DEPLOYING, leaving it stuck in FAILED with a worker.
+    const { count: updatedCount } = await this._prisma.workerDeployment.updateMany({
       where: {
         id: deployment.id,
+        status: "BUILDING",
       },
       data: {
         status: "FAILED",
@@ -229,7 +273,20 @@ export class CreateDeploymentBackgroundWorkerServiceV4 extends BaseService {
       },
     });
 
-    await TimeoutDeploymentService.dequeue(deployment.id, this._prisma);
+    if (updatedCount === 0) {
+      logger.warn(
+        "failBackgroundWorkerDeployment: deployment moved out of BUILDING during call, skipping FAILED transition",
+        {
+          deploymentId: deployment.id,
+          originalError: error.message,
+        }
+      );
+    } else {
+      // Only dequeue the timeout if we actually flipped to FAILED — otherwise a
+      // sibling attempt may have just enqueued it as part of a successful
+      // BUILDING → DEPLOYING transition.
+      await TimeoutDeploymentService.dequeue(deployment.id, this._prisma);
+    }
 
     throw error;
   }
diff --git a/apps/webapp/app/v3/services/createDeploymentBackgroundWorkerV4/findOrCreateBackgroundWorker.server.ts b/apps/webapp/app/v3/services/createDeploymentBackgroundWorkerV4/findOrCreateBackgroundWorker.server.ts
@@ -0,0 +1,81 @@
+import type { CreateBackgroundWorkerRequestBody } from "@trigger.dev/core/v3";
+import { BackgroundWorkerId } from "@trigger.dev/core/v3/isomorphic";
+import {
+  isUniqueConstraintError,
+  type BackgroundWorker,
+  type PrismaClientOrTransaction,
+  type WorkerDeployment,
+} from "@trigger.dev/database";
+import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
+import { ServiceValidationError } from "../common.server";
+import { stripBackgroundWorkerMetadataForStorage } from "../stripBackgroundWorkerMetadataForStorage.server";
+
+/**
+ * Idempotent on `(project, environment, version)` for sequential calls, not concurrent calls.
+ *
+ * Failure shapes the caller distinguishes:
+ *   - `ServiceValidationError` (409): definitive — contentHash drift means a different build
+ *     is being pushed under the same deployment version, so the caller should fail the
+ *     deployment instead of waiting for a timeout.
+ *   - Plain `Error`: transient — two attempts raced the `create()` call and the loser caught
+ *     the unique-index violation. The caller should propagate this as 5xx so the CLI's
+ *     retry/backoff hits findFirst on the next attempt and returns the winner's row.
+ */
+export async function findOrCreateBackgroundWorker(
+  environment: AuthenticatedEnvironment,
+  deployment: WorkerDeployment,
+  body: CreateBackgroundWorkerRequestBody,
+  prisma: PrismaClientOrTransaction
+): Promise<BackgroundWorker> {
+  const existing = await prisma.backgroundWorker.findFirst({
+    where: {
+      projectId: environment.projectId,
+      runtimeEnvironmentId: environment.id,
+      version: deployment.version,
+    },
+  });
+
+  if (existing && existing.contentHash === body.metadata.contentHash) {
+    return existing;
+  }
+
+  if (existing) {
+    throw new ServiceValidationError(
+      "A background worker for this deployment version already exists with a different content hash",
+      409
+    );
+  }
+
+  try {
+    return await prisma.backgroundWorker.create({
+      data: {
+        ...BackgroundWorkerId.generate(),
+        version: deployment.version,
+        runtimeEnvironmentId: environment.id,
+        projectId: environment.projectId,
+        metadata: stripBackgroundWorkerMetadataForStorage(body.metadata),
+        contentHash: body.metadata.contentHash,
+        cliVersion: body.metadata.cliPackageVersion,
+        sdkVersion: body.metadata.packageVersion,
+        supportsLazyAttempts: body.supportsLazyAttempts,
+        engine: body.engine,
+        runtime: body.metadata.runtime,
+        runtimeVersion: body.metadata.runtimeVersion,
+      },
+    });
+  } catch (error) {
+    // Concurrent attempts raced past `findFirst` and both reached `create`. Surface
+    // a clear, non-Prisma error so the 5xx the caller returns isn't an opaque
+    // P2002 — the CLI's retry will then hit `findFirst` and find the winner's row.
+    // Intentionally NOT a ServiceValidationError so the caller doesn't fail-deploy
+    // on a transient race.
+    if (
+      isUniqueConstraintError(error, ["projectId", "runtimeEnvironmentId", "version"])
+    ) {
+      throw new Error(
+        "Concurrent background worker registration detected for this deployment version; please retry"
+      );
+    }
+    throw error;
+  }
+}
diff --git a/apps/webapp/app/v3/services/stripBackgroundWorkerMetadataForStorage.server.ts b/apps/webapp/app/v3/services/stripBackgroundWorkerMetadataForStorage.server.ts
@@ -0,0 +1,28 @@
+import { BackgroundWorkerMetadata } from "@trigger.dev/core/v3";
+import { Prisma } from "@trigger.dev/database";
+
+/**
+ * Strip BackgroundWorkerMetadata down to the slice that's actually read after
+ * storage. Everything else is duplicated to dedicated columns/tables
+ * (BackgroundWorker.{contentHash,cliVersion,sdkVersion,runtime,runtimeVersion},
+ * BackgroundWorkerTask, BackgroundWorkerFile, TaskQueue, Prompt). Today the
+ * only post-write reader is changeCurrentDeployment.server.ts, which feeds
+ * tasks[].schedule into syncDeclarativeSchedules. packageVersion, contentHash,
+ * and tasks[].filePath are kept solely to satisfy BackgroundWorkerMetadata's
+ * required fields when the column is parsed back.
+ */
+export function stripBackgroundWorkerMetadataForStorage(
+  metadata: BackgroundWorkerMetadata
+): Prisma.InputJsonValue {
+  return {
+    packageVersion: metadata.packageVersion,
+    contentHash: metadata.contentHash,
+    tasks: metadata.tasks
+      .filter((t) => t.schedule)
+      .map((t) => ({
+        id: t.id,
+        filePath: t.filePath,
+        schedule: t.schedule,
+      })),
+  };
+}
diff --git a/apps/webapp/test/findOrCreateBackgroundWorker.test.ts b/apps/webapp/test/findOrCreateBackgroundWorker.test.ts
