Rename batch stuff to v3 so it's not confusing

Author: ericallam

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

@@ -100,6 +100,7 @@ export class BatchListPresenter extends BasePresenter {
         status: BatchTaskRunStatus;
         createdAt: Date;
         updatedAt: Date;
+        completedAt: Date | null;
         runCount: BigInt;
         batchVersion: string;
       }[]
@@ -111,6 +112,7 @@ export class BatchListPresenter extends BasePresenter {
     b.status,
     b."createdAt",
     b."updatedAt",
+    b."completedAt",
     b."runCount",
     b."batchVersion"
 FROM
@@ -196,7 +198,11 @@ WHERE
           createdAt: batch.createdAt.toISOString(),
           updatedAt: batch.updatedAt.toISOString(),
           hasFinished,
-          finishedAt: hasFinished ? batch.updatedAt.toISOString() : undefined,
+          finishedAt: batch.completedAt
+            ? batch.completedAt.toISOString()
+            : hasFinished
+            ? batch.updatedAt.toISOString()
+            : undefined,
           status: batch.status,
           environment: displayableEnvironment(environment, userId),
           runCount: Number(batch.runCount),

apps/webapp/app/routes/api.v1.tasks.batch.ts

@@ -12,8 +12,8 @@ import { resolveIdempotencyKeyTTL } from "~/utils/idempotencyKeys.server";
 import { ServiceValidationError } from "~/v3/services/baseService.server";
 import {
   BatchProcessingStrategy,
-  BatchTriggerV2Service,
-} from "~/v3/services/batchTriggerV2.server";
+  BatchTriggerV3Service,
+} from "~/v3/services/batchTriggerV3.server";
 import { OutOfEntitlementError } from "~/v3/services/triggerTask.server";
 import { HeadersSchema } from "./api.v1.tasks.$taskId.trigger";
 
@@ -85,7 +85,7 @@ const { action, loader } = createActionApiRoute(
       resolveIdempotencyKeyTTL(idempotencyKeyTTL) ??
       new Date(Date.now() + 24 * 60 * 60 * 1000 * 30);
 
-    const service = new BatchTriggerV2Service(batchProcessingStrategy ?? undefined);
+    const service = new BatchTriggerV3Service(batchProcessingStrategy ?? undefined);
 
     try {
       const batch = await service.call(authentication.environment, body, {

apps/webapp/app/services/worker.server.ts

@@ -55,7 +55,7 @@ import {
   CancelDevSessionRunsServiceOptions,
 } from "~/v3/services/cancelDevSessionRuns.server";
 import { logger } from "./logger.server";
-import { BatchProcessingOptions, BatchTriggerV2Service } from "~/v3/services/batchTriggerV2.server";
+import { BatchProcessingOptions, BatchTriggerV3Service } from "~/v3/services/batchTriggerV3.server";
 
 const workerCatalog = {
   indexEndpoint: z.object({
@@ -733,7 +733,7 @@ function getWorkerQueue() {
         priority: 0,
         maxAttempts: 5,
         handler: async (payload, job) => {
-          const service = new BatchTriggerV2Service(payload.strategy);
+          const service = new BatchTriggerV3Service(payload.strategy);
 
           await service.processBatchTaskRun(payload);
         },

apps/webapp/app/v3/services/batchTriggerV2.server.ts renamed to apps/webapp/app/v3/services/batchTriggerV3.server.ts

@@ -57,7 +57,34 @@ type RunItemData = {
   taskIdentifier: string;
 };
 
-export class BatchTriggerV2Service extends BaseService {
+/**
+ * ### V3
+ *
+ * BatchTrigger v3 doesn't have any changes from v2, other than a different system for tracking if the
+ * batch is completed.
+ *
+ * v3 BatchTaskRun's now must be "sealed" before they could be considered completed. Being "sealed" means
+ * that all the items in the batch have been processed and the batch is ready to be considered completed.
+ *
+ * We also now track the expected count of items in the batch, and then as each BatchTaskRunItem is set to COMPLETED,
+ * we increment the BatchTaskRun's completed count. Once the completed count is equal to the expected count, and the
+ * batch is sealed, we can consider the batch completed.
+ *
+ * So now when the v3 batch is considered completed, we will enqueue the ResumeBatchRunService to resume the dependent
+ * task attempt if there is one. This is in contrast to v2 batches where every time a task was completed, we would schedule
+ * the ResumeBatchRunService to check if the batch was completed and set it to completed if it was.
+ *
+ * We've also introduced a new column "resumedAt" that will be set when the batch is resumed. Previously in v2 batches, the status == "COMPLETED" was overloaded
+ * to mean that the batch was completed and resumed. Now we have a separate column to track when the batch was resumed (and to make sure it's only resumed once).
+ *
+ * ### V2
+ *
+ * Batch v2 added the ability to trigger more than 100 tasks in a single batch. This was done by offloading the payload to the object store and
+ * then processing the batch in chunks of 50 tasks at a time in the background.
+ *
+ * The other main difference from v1 is that a single batch in v2 could trigger multiple different tasks, whereas in v1 a batch could only trigger a single task.
+ */
+export class BatchTriggerV3Service extends BaseService {
   private _batchProcessingStrategy: BatchProcessingStrategy;
 
   constructor(

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

@@ -1,8 +1,12 @@
 import { FlushedRunMetadata, sanitizeError, TaskRunError } from "@trigger.dev/core/v3";
 import { type Prisma, type TaskRun } from "@trigger.dev/database";
+import { findQueueInEnvironment } from "~/models/taskQueue.server";
+import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
+import { updateMetadataService } from "~/services/metadata/updateMetadata.server";
 import { marqs } from "~/v3/marqs/index.server";
 import { generateFriendlyId } from "../friendlyIdentifiers";
+import { socketIo } from "../handleSocketIo.server";
 import {
   FINAL_ATTEMPT_STATUSES,
   isFailedRunStatus,
@@ -11,15 +15,10 @@ import {
 } from "../taskStatus";
 import { PerformTaskRunAlertsService } from "./alerts/performTaskRunAlerts.server";
 import { BaseService } from "./baseService.server";
-import { ResumeDependentParentsService } from "./resumeDependentParents.server";
+import { completeBatchTaskRunItemV3 } from "./batchTriggerV3.server";
 import { ExpireEnqueuedRunService } from "./expireEnqueuedRun.server";
-import { socketIo } from "../handleSocketIo.server";
 import { ResumeBatchRunService } from "./resumeBatchRun.server";
-import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
-import { updateMetadataService } from "~/services/metadata/updateMetadata.server";
-import { findQueueInEnvironment, sanitizeQueueName } from "~/models/taskQueue.server";
-import { $transaction } from "~/db.server";
-import { completeBatchTaskRunItemV3 } from "./batchTriggerV2.server";
+import { ResumeDependentParentsService } from "./resumeDependentParents.server";
 
 type BaseInput = {
   id: string;

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

@@ -5,7 +5,7 @@ import { BaseService } from "./baseService.server";
 import { ResumeBatchRunService } from "./resumeBatchRun.server";
 import { ResumeTaskDependencyService } from "./resumeTaskDependency.server";
 import { $transaction } from "~/db.server";
-import { completeBatchTaskRunItemV3 } from "./batchTriggerV2.server";
+import { completeBatchTaskRunItemV3 } from "./batchTriggerV3.server";
 
 type Output =
   | {

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

@@ -9,7 +9,7 @@ import { workerQueue } from "~/services/worker.server";
 import { BaseService } from "./baseService.server";
 import { ResumeBatchRunService } from "./resumeBatchRun.server";
 import { ResumeTaskDependencyService } from "./resumeTaskDependency.server";
-import { completeBatchTaskRunItemV3 } from "./batchTriggerV2.server";
+import { completeBatchTaskRunItemV3 } from "./batchTriggerV3.server";
 
 export class ResumeTaskRunDependenciesService extends BaseService {
   public async call(attemptId: string) {

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

@@ -792,3 +792,41 @@ export const batchAutoIdempotencyKeyChild = task({
     return payload;
   },
 });
+
+export const batchTriggerIdempotencyKeyTest = task({
+  id: "batch-trigger-idempotency-key-test",
+  retry: {
+    maxAttempts: 1,
+  },
+  run: async () => {
+    // Trigger a batch of 2 items with the same idempotency key
+    await batchV2TestChild.batchTrigger(
+      [
+        { payload: { foo: "bar" }, options: { idempotencyKey: "test-idempotency-key" } },
+        { payload: { foo: "baz" }, options: { idempotencyKey: "test-idempotency-key" } },
+      ],
+      {
+        triggerSequentially: true,
+      }
+    );
+
+    // Now trigger a batch of 21 items, with just the last one having an idempotency key
+    await batchV2TestChild.batchTrigger(
+      Array.from({ length: 20 }, (_, i) => ({
+        payload: { foo: `bar${i}` },
+        options: {},
+      })).concat([
+        {
+          payload: { foo: "baz" },
+          options: { idempotencyKey: "test-idempotency-key-2" },
+        },
+      ]),
+      {
+        triggerSequentially: true,
+      }
+    );
+
+    // Now while that batch is being processed in the background, lets trigger the same task with that idempotency key
+    await batchV2TestChild.trigger({ foo: "baz" }, { idempotencyKey: "test-idempotency-key-2" });
+  },
+});