batchTriggerAndWait checkpoint race condition when at max concurrency (#1296)

* Ignore /packages/cli-v3/src/package.json

* Added more logs when resuming a dependency, added the runId

* A task for reproducing a race condition with checkpoints

* Fix for doing remote image build when not self-hosting

* Set team members, alerts and schedule limits to 100m for self-hosting

* Import fix

* Set the checkpointEventId in marqs when the checkpoint is created for batchTriggerAndWait

This should fix a horrible race condition when at max concurrency

Author: matt-aitken

.gitignore

@@ -56,4 +56,5 @@ apps/**/public/build
 .trigger
 .tshy*
 .yarn
-*.tsbuildinfo
+*.tsbuildinfo
+/packages/cli-v3/src/package.json

apps/webapp/app/presenters/TeamPresenter.server.ts

@@ -1,6 +1,6 @@
 import { getTeamMembersAndInvites } from "~/models/member.server";
-import { BasePresenter } from "./v3/basePresenter.server";
 import { getLimit } from "~/services/platform.v3.server";
+import { BasePresenter } from "./v3/basePresenter.server";
 
 export class TeamPresenter extends BasePresenter {
   public async call({ userId, organizationId }: { userId: string; organizationId: string }) {
@@ -13,7 +13,7 @@ export class TeamPresenter extends BasePresenter {
       return;
     }
 
-    const limit = await getLimit(organizationId, "teamMembers", 25);
+    const limit = await getLimit(organizationId, "teamMembers", 100_000_000);
 
     return {
       ...result,

apps/webapp/app/presenters/v3/AlertChannelListPresenter.server.ts

@@ -43,7 +43,7 @@ export class AlertChannelListPresenter extends BasePresenter {
       throw new Error(`Project not found: ${projectId}`);
     }
 
-    const limit = await getLimit(organization.organizationId, "alerts", 25);
+    const limit = await getLimit(organization.organizationId, "alerts", 100_000_000);
 
     return {
       alertChannels: await Promise.all(

apps/webapp/app/presenters/v3/ScheduleListPresenter.server.ts

@@ -256,7 +256,7 @@ export class ScheduleListPresenter extends BasePresenter {
       };
     });
 
-    const limit = await getLimit(project.organizationId, "schedules", 500);
+    const limit = await getLimit(project.organizationId, "schedules", 100_000_000);
 
     return {
       currentPage: page,

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

@@ -76,7 +76,7 @@ export class CheckScheduleService extends BaseService {
         throw new ServiceValidationError("Project not found");
       }
 
-      const limit = await getLimit(project.organizationId, "schedules", 500);
+      const limit = await getLimit(project.organizationId, "schedules", 100_000_000);
       const schedulesCount = await this._prisma.taskSchedule.count({
         where: {
           projectId,

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

@@ -258,6 +258,16 @@ export class CreateCheckpointService extends BaseService {
             };
           }
 
+          //if there's a message in the queue, we make sure the checkpoint event is on it
+          await marqs?.replaceMessage(
+            attempt.taskRun.id,
+            {
+              checkpointEventId: checkpointEvent.id,
+            },
+            undefined,
+            true
+          );
+
           await ResumeBatchRunService.enqueue(batchRun.id, this._prisma);
 
           return {

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

@@ -29,9 +29,9 @@ export class InitializeDeploymentService extends BaseService {
       const nextVersion = calculateNextBuildVersion(latestDeployment?.version);
 
       // Try and create a depot build and get back the external build data
-      const externalBuildData = !!payload.selfHosted
-        ? await createRemoteImageBuild(environment.project)
-        : undefined;
+      const externalBuildData = payload.selfHosted
+        ? undefined
+        : await createRemoteImageBuild(environment.project);
 
       const triggeredBy = payload.userId
         ? await this._prisma.user.findUnique({

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

@@ -39,6 +39,16 @@ export class ResumeTaskDependencyService extends BaseService {
     const dependentRun = dependency.dependentAttempt.taskRun;
 
     if (dependency.dependentAttempt.status === "PAUSED" && dependency.checkpointEventId) {
+      logger.debug(
+        "Task dependency resume: Attempt is paused and there's a checkpoint. Enqueuing resume with checkpoint.",
+        {
+          attemptId: dependency.id,
+          dependentAttempt: dependency.dependentAttempt,
+          checkpointEventId: dependency.checkpointEventId,
+          hasCheckpointEvent: !!dependency.checkpointEventId,
+          runId: dependentRun.id,
+        }
+      );
       await marqs?.enqueueMessage(
         dependency.taskRun.runtimeEnvironment,
         dependentRun.queue,
@@ -61,6 +71,7 @@ export class ResumeTaskDependencyService extends BaseService {
         dependentAttempt: dependency.dependentAttempt,
         checkpointEventId: dependency.checkpointEventId,
         hasCheckpointEvent: !!dependency.checkpointEventId,
+        runId: dependentRun.id,
       });
 
       if (dependency.dependentAttempt.status === "PAUSED" && !dependency.checkpointEventId) {

references/v3-catalog/src/trigger/checkpoints.ts

@@ -1,4 +1,4 @@
-import { logger, task, wait } from "@trigger.dev/sdk/v3";
+import { logger, queue, task, wait } from "@trigger.dev/sdk/v3";
 
 type Payload = {
   count?: number;
@@ -70,6 +70,7 @@ export const nestedDependencies = task({
           maxDepth,
           waitSeconds,
           failAttemptChance,
+          batchSize,
         });
         logger.log(`Triggered complete ${i + 1}/${batchSize}`);
 
@@ -153,3 +154,36 @@ export const bulkPermanentlyFrozen = task({
     );
   },
 });
+
+const oneAtATime = queue({
+  name: "race-condition",
+  concurrencyLimit: 1,
+});
+
+export const raceConditionCheckpointDequeue = task({
+  id: "race-condition-checkpoint-dequeue",
+  queue: oneAtATime,
+  run: async ({ isBatch = true }: { isBatch?: boolean }) => {
+    await holdConcurrency.trigger({ waitSeconds: 45 });
+
+    if (isBatch) {
+      await fixedLengthTask.batchTriggerAndWait(
+        Array.from({ length: 1 }, (_, i) => ({
+          payload: { waitSeconds: 5 },
+        }))
+      );
+    } else {
+      await fixedLengthTask.triggerAndWait({ waitSeconds: 5 });
+    }
+
+    logger.log(`Successfully completed task`);
+  },
+});
+
+export const holdConcurrency = task({
+  id: "hold-concurrency",
+  queue: oneAtATime,
+  run: async ({ waitSeconds = 60 }: { waitSeconds?: number }) => {
+    await new Promise((resolve) => setTimeout(resolve, waitSeconds * 1000));
+  },
+});