v3: fix dependency trigger and wait (#1030)

* fix dev dependecy resumes

* add catalog entry with multiple trigger waits

* update docker provider lifecycle command

* switch to x for clearing run filters

* Revert "fix dev dependecy resumes"

This reverts commit b4061f2ae71f21889adcf061928e2e77c436f0ed.

* fix dependency resumes, properly this time

* add catalog entry for dependency waits in loops

* advice in docs re parallel dependency waits

* fix link from v3 to v2 docs

* move lifecycle command logging to debug only

* Removed batchOptions from the trigger options and the docs

---------

Co-authored-by: Matt Aitken <matt@mattaitken.com>

.changeset/smart-needles-move.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+Remove unimplemented batchOptions

apps/docker-provider/src/index.ts

@@ -257,6 +257,7 @@ class DockerTaskOperations implements TaskOperations {
       return await execa("docker", [
         "exec",
         containerName,
+        "busybox",
         "wget",
         "-q",
         "-O-",

apps/kubernetes-provider/src/index.ts

@@ -409,7 +409,7 @@ class KubernetesTaskOperations implements TaskOperations {
       `for i in $(seq ${retries}); do sleep 1; busybox wget -q -O- 127.0.0.1:8000/${type}?cause=${cause} && break; done`,
     ];
 
-    logger.log("getLifecycleCommand()", { exec });
+    logger.debug("getLifecycleCommand()", { exec });
 
     return exec;
   }

apps/webapp/app/components/runs/v3/RunFilters.tsx

@@ -1,4 +1,4 @@
-import { TrashIcon } from "@heroicons/react/20/solid";
+import { XMarkIcon } from "@heroicons/react/20/solid";
 import { useNavigate } from "@remix-run/react";
 import type { TaskRunStatus as TaskRunStatusType } from "@trigger.dev/database";
 import { RuntimeEnvironment, TaskRunAttemptStatus, TaskRunStatus } from "@trigger.dev/database";
@@ -247,7 +247,7 @@ export function RunsFilters({ possibleEnvironments, possibleTasks }: RunFiltersP
 
       <TimeFrameFilter from={from} to={to} onRangeChanged={handleTimeFrameChange} />
 
-      <Button variant="minimal/small" onClick={() => clearFilters()} LeadingIcon={TrashIcon} />
+      <Button variant="minimal/small" onClick={() => clearFilters()} LeadingIcon={XMarkIcon} />
     </div>
   );
 }

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

@@ -4,12 +4,13 @@ import {
   TaskRunExecution,
   TaskRunExecutionResult,
 } from "@trigger.dev/core/v3";
-import { $transaction } from "~/db.server";
+import { $transaction, PrismaClientOrTransaction } from "~/db.server";
 import { logger } from "~/services/logger.server";
 import { marqs } from "~/v3/marqs/index.server";
 import { socketIo } from "../handleSocketIo.server";
 import { sharedQueueTasks } from "../marqs/sharedQueueConsumer.server";
 import { BaseService } from "./baseService.server";
+import { TaskRunAttempt } from "@trigger.dev/database";
 
 export class ResumeAttemptService extends BaseService {
   public async call(
@@ -24,7 +25,7 @@ export class ResumeAttemptService extends BaseService {
         },
         include: {
           taskRun: true,
-          taskRunDependency: {
+          dependencies: {
             include: {
               taskRun: {
                 include: {
@@ -40,8 +41,12 @@ export class ResumeAttemptService extends BaseService {
                 },
               },
             },
+            orderBy: {
+              createdAt: "desc",
+            },
+            take: 1,
           },
-          batchTaskRunDependency: {
+          batchDependencies: {
             include: {
               items: {
                 include: {
@@ -61,6 +66,10 @@ export class ResumeAttemptService extends BaseService {
                 },
               },
             },
+            orderBy: {
+              createdAt: "desc",
+            },
+            take: 1,
           },
         },
       });
@@ -78,6 +87,8 @@ export class ResumeAttemptService extends BaseService {
         return;
       }
 
+      let completedAttemptIds: string[] = [];
+
       switch (params.type) {
         case "WAIT_FOR_DURATION": {
           logger.error(
@@ -93,12 +104,10 @@ export class ResumeAttemptService extends BaseService {
           });
           break;
         }
-        case "WAIT_FOR_TASK":
+        case "WAIT_FOR_TASK": {
-        case "WAIT_FOR_BATCH": {
+          if (attempt.dependencies.length) {
-          let completedAttemptIds: string[] = [];
+            // We only care about the latest dependency
-
+            const dependentAttempt = attempt.dependencies[0].taskRun.attempts[0];
-          if (attempt.taskRunDependency) {
-            const dependentAttempt = attempt.taskRunDependency.taskRun.attempts[0];
 
             if (!dependentAttempt) {
               logger.error("No dependent attempt", { attemptId: attempt.id });
@@ -106,19 +115,16 @@ export class ResumeAttemptService extends BaseService {
             }
 
             completedAttemptIds = [dependentAttempt.id];
-
+          } else {
-            await tx.taskRunAttempt.update({
+            logger.error("No task dependency", { attemptId: attempt.id });
-              where: {
+            return;
-                id: attempt.id,
+          }
-              },
+          break;
-              data: {
+        }
-                taskRunDependency: {
+        case "WAIT_FOR_BATCH": {
-                  disconnect: true,
+          if (attempt.batchDependencies) {
-                },
+            // We only care about the latest batch dependency
-              },
+            const dependentBatchItems = attempt.batchDependencies[0].items;
-            });
-          } else if (attempt.batchTaskRunDependency) {
-            const dependentBatchItems = attempt.batchTaskRunDependency.items;
 
             if (!dependentBatchItems) {
               logger.error("No dependent batch items", { attemptId: attempt.id });
@@ -126,115 +132,112 @@ export class ResumeAttemptService extends BaseService {
             }
 
             completedAttemptIds = dependentBatchItems.map((item) => item.taskRun.attempts[0]?.id);
-
-            await tx.taskRunAttempt.update({
-              where: {
-                id: attempt.id,
-              },
-              data: {
-                batchTaskRunDependency: {
-                  disconnect: true,
-                },
-              },
-            });
           } else {
-            logger.error("No dependencies", { attemptId: attempt.id });
+            logger.error("No batch dependency", { attemptId: attempt.id });
             return;
           }
-
-          if (completedAttemptIds.length === 0) {
-            logger.error("No completed attempt IDs", { attemptId: attempt.id });
-            return;
-          }
-
-          const completions: TaskRunExecutionResult[] = [];
-          const executions: TaskRunExecution[] = [];
-
-          for (const completedAttemptId of completedAttemptIds) {
-            const completedAttempt = await tx.taskRunAttempt.findUnique({
-              where: {
-                id: completedAttemptId,
-                taskRun: {
-                  lockedAt: {
-                    not: null,
-                  },
-                  lockedById: {
-                    not: null,
-                  },
-                },
-              },
-            });
-
-            if (!completedAttempt) {
-              logger.error("Completed attempt not found", {
-                attemptId: attempt.id,
-                completedAttemptId,
-              });
-              await marqs?.acknowledgeMessage(attempt.taskRunId);
-              return;
-            }
-
-            const completion = await sharedQueueTasks.getCompletionPayloadFromAttempt(
-              completedAttempt.id
-            );
-
-            if (!completion) {
-              logger.error("Failed to get completion payload", {
-                attemptId: attempt.id,
-                completedAttemptId,
-              });
-              await marqs?.acknowledgeMessage(attempt.taskRunId);
-              return;
-            }
-
-            completions.push(completion);
-
-            const executionPayload = await sharedQueueTasks.getExecutionPayloadFromAttempt(
-              completedAttempt.id
-            );
-
-            if (!executionPayload) {
-              logger.error("Failed to get execution payload", {
-                attemptId: attempt.id,
-                completedAttemptId,
-              });
-              await marqs?.acknowledgeMessage(attempt.taskRunId);
-              return;
-            }
-
-            executions.push(executionPayload.execution);
-          }
-
-          const updated = await tx.taskRunAttempt.update({
-            where: {
-              id: attempt.id,
-            },
-            data: {
-              status: "EXECUTING",
-              taskRun: {
-                update: {
-                  data: {
-                    status: attempt.number > 1 ? "RETRYING_AFTER_FAILURE" : "EXECUTING",
-                  },
-                },
-              },
-            },
-          });
-
-          socketIo.coordinatorNamespace.emit("RESUME_AFTER_DEPENDENCY", {
-            version: "v1",
-            runId: attempt.taskRunId,
-            attemptId: attempt.id,
-            attemptFriendlyId: attempt.friendlyId,
-            completions,
-            executions,
-          });
           break;
         }
         default: {
           break;
         }
       }
+
+      await this.#handleDependencyResume(attempt, completedAttemptIds, tx);
+    });
+  }
+
+  async #handleDependencyResume(
+    attempt: TaskRunAttempt,
+    completedAttemptIds: string[],
+    tx: PrismaClientOrTransaction
+  ) {
+    if (completedAttemptIds.length === 0) {
+      logger.error("No completed attempt IDs", { attemptId: attempt.id });
+      return;
+    }
+
+    const completions: TaskRunExecutionResult[] = [];
+    const executions: TaskRunExecution[] = [];
+
+    for (const completedAttemptId of completedAttemptIds) {
+      const completedAttempt = await tx.taskRunAttempt.findUnique({
+        where: {
+          id: completedAttemptId,
+          taskRun: {
+            lockedAt: {
+              not: null,
+            },
+            lockedById: {
+              not: null,
+            },
+          },
+        },
+      });
+
+      if (!completedAttempt) {
+        logger.error("Completed attempt not found", {
+          attemptId: attempt.id,
+          completedAttemptId,
+        });
+        await marqs?.acknowledgeMessage(attempt.taskRunId);
+        return;
+      }
+
+      const completion = await sharedQueueTasks.getCompletionPayloadFromAttempt(
+        completedAttempt.id
+      );
+
+      if (!completion) {
+        logger.error("Failed to get completion payload", {
+          attemptId: attempt.id,
+          completedAttemptId,
+        });
+        await marqs?.acknowledgeMessage(attempt.taskRunId);
+        return;
+      }
+
+      completions.push(completion);
+
+      const executionPayload = await sharedQueueTasks.getExecutionPayloadFromAttempt(
+        completedAttempt.id
+      );
+
+      if (!executionPayload) {
+        logger.error("Failed to get execution payload", {
+          attemptId: attempt.id,
+          completedAttemptId,
+        });
+        await marqs?.acknowledgeMessage(attempt.taskRunId);
+        return;
+      }
+
+      executions.push(executionPayload.execution);
+    }
+
+    const updated = await tx.taskRunAttempt.update({
+      where: {
+        id: attempt.id,
+      },
+      data: {
+        status: "EXECUTING",
+        taskRun: {
+          update: {
+            data: {
+              status: attempt.number > 1 ? "RETRYING_AFTER_FAILURE" : "EXECUTING",
+            },
+          },
+        },
+      },
+    });
+
+    socketIo.coordinatorNamespace.emit("RESUME_AFTER_DEPENDENCY", {
+      version: "v1",
+      runId: attempt.taskRunId,
+      attemptId: attempt.id,
+      attemptFriendlyId: attempt.friendlyId,
+      completions,
+      executions,
     });
   }
 }

docs/mint.json

@@ -49,7 +49,7 @@
     },
     {
       "name": "v2",
-      "url": "https://trigger.dev/docs",
+      "url": "https://trigger.dev/docs/documentation",
       "version": "v3 (Developer Preview)"
     },
     {

docs/v3/triggering.mdx

@@ -165,6 +165,45 @@ export const myTask = task({
 
 This is where it gets interesting. You can trigger a task and then wait for the result. This is useful when you need to call a different task and then use the result to continue with your task.
 
+<Accordion title="Don't use this in parallel, e.g. with `Promise.all()`">
+  Instead, use `batchTriggerAndWait()` if you can, or a for loop if you can't.
+
+  To control concurrency using batch triggers, you can set `queue.concurrencyLimit` on the child task.
+
+<CodeGroup>
+```ts /trigger/batch.ts
+export const batchTask = task({
+  id: "batch-task",
+  run: async (payload: string) => {
+    const results = await childTask.batchTriggerAndWait({
+      items: [{ payload: "item1" }, { payload: "item2" }],
+    });
+    console.log("Results", results);
+
+    //...do stuff with the results
+  },
+});
+```
+
+```ts /trigger/loop.ts
+export const loopTask = task({
+  id: "loop-task",
+  run: async (payload: string) => {
+    //this will be slower than the batch version
+    //as we have to resume the parent after each iteration
+    for (let i = 0; i < 2; i++) {
+      const result = await childTask.triggerAndWait({ payload: `item${i}` });
+      console.log("Result", result);
+
+      //...do stuff with the result
+    }
+  },
+});
+```
+</CodeGroup>
+
+</Accordion>
+
 ```ts /trigger/parent.ts
 export const parentTask = task({
   id: "parent-task",
@@ -181,6 +220,47 @@ export const parentTask = task({
 
 You can batch trigger a task and wait for all the results. This is useful for the fan-out pattern, where you need to call a task multiple times and then wait for all the results to continue with your task.
 
+<Accordion title="Don't use this in parallel, e.g. with `Promise.all()`">
+  Instead, pass in all items at once and set an appropriate `maxConcurrency`. Alternatively, use sequentially with a for loop.
+
+  To control concurrency, you can set `queue.concurrencyLimit` on the child task.
+
+<CodeGroup>
+```ts /trigger/batch.ts
+export const batchTask = task({
+  id: "batch-task",
+  run: async (payload: string) => {
+    const results = await childTask.batchTriggerAndWait({
+      items: [{ payload: "item1" }, { payload: "item2" }],
+    });
+    console.log("Results", results);
+
+    //...do stuff with the results
+  },
+});
+```
+
+```ts /trigger/loop.ts
+export const loopTask = task({
+  id: "loop-task",
+  run: async (payload: string) => {
+    //this will be slower than a single batchTriggerAndWait()
+    //as we have to resume the parent after each iteration
+    for (let i = 0; i < 2; i++) {
+      const result = await childTask.batchTriggerAndWait({
+        items: [{ payload: `itemA${i}` }, { payload: `itemB${i}` }],
+      });
+      console.log("Result", result);
+
+      //...do stuff with the result
+    }
+  },
+});
+```
+</CodeGroup>
+
+</Accordion>
+
 ```ts /trigger/nested.ts
 export const batchParentTask = task({
   id: "parent-task",

packages/database/prisma/migrations/20240415135132_make_attempt_dependencies_an_array/migration.sql

@@ -0,0 +1,2 @@
+-- DropIndex
+DROP INDEX "TaskRunDependency_dependentAttemptId_key";

packages/database/prisma/migrations/20240415143325_remove_attempt_unique_constraint_from_batch_runs/migration.sql

@@ -0,0 +1,2 @@
+-- DropIndex
+DROP INDEX "BatchTaskRun_dependentTaskAttemptId_key";

packages/database/prisma/schema.prisma

@@ -1686,8 +1686,8 @@ model TaskRunDependency {
   checkpointEventId String?                 @unique
 
   /// An attempt that is dependent on this task run. 
-  dependentAttempt   TaskRunAttempt? @relation("dependentAttempt", fields: [dependentAttemptId], references: [id])
+  dependentAttempt   TaskRunAttempt? @relation(fields: [dependentAttemptId], references: [id])
-  dependentAttemptId String?         @unique
+  dependentAttemptId String?
 
   /// A batch run that is dependent on this task run
   dependentBatchRun   BatchTaskRun? @relation("dependentBatchRun", fields: [dependentBatchRunId], references: [id])
@@ -1751,8 +1751,8 @@ model TaskRunAttempt {
   output     String?
   outputType String  @default("application/json")
 
-  taskRunDependency      TaskRunDependency? @relation("dependentAttempt")
+  dependencies      TaskRunDependency[]
-  batchTaskRunDependency BatchTaskRun?
+  batchDependencies BatchTaskRun[]
 
   checkpoints            Checkpoint[]
   batchTaskRunItems      BatchTaskRunItem[]
@@ -1923,7 +1923,7 @@ model BatchTaskRun {
   runtimeEnvironmentId String
 
   dependentTaskAttempt   TaskRunAttempt? @relation(fields: [dependentTaskAttemptId], references: [id], onDelete: Cascade, onUpdate: Cascade)
-  dependentTaskAttemptId String?         @unique
+  dependentTaskAttemptId String?
 
   items           BatchTaskRunItem[]
   runDependencies TaskRunDependency[] @relation("dependentBatchRun")

packages/trigger-sdk/src/v3/shared.ts

@@ -179,12 +179,12 @@ export interface Task<TInput, TOutput = any> {
   trigger: (params: { payload: TInput; options?: TaskRunOptions }) => Promise<InvokeHandle>;
   batchTrigger: (params: {
     items: { payload: TInput; options?: TaskRunOptions }[];
-    batchOptions?: BatchRunOptions;
+    // batchOptions?: BatchRunOptions;
   }) => Promise<InvokeBatchHandle>;
   triggerAndWait: (params: { payload: TInput; options?: TaskRunOptions }) => Promise<TOutput>;
   batchTriggerAndWait: (params: {
     items: { payload: TInput; options?: TaskRunOptions }[];
-    batchOptions?: BatchRunOptions;
+    // batchOptions?: BatchRunOptions;
   }) => Promise<BatchResult<TOutput>>;
 }
 

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

@@ -114,3 +114,54 @@ export const subtasksWithRetries = task({
     };
   },
 });
+
+export const multipleTriggerWaits = task({
+  id: "multiple-trigger-waits",
+  run: async ({ message = "test" }: { message?: string }) => {
+    await simpleChildTask.triggerAndWait({ payload: { message: `${message} - 1.a` } });
+    await simpleChildTask.triggerAndWait({ payload: { message: `${message} - 2.a` } });
+
+    await simpleChildTask.batchTriggerAndWait({
+      items: [
+        { payload: { message: `${message} - 3.a` } },
+        { payload: { message: `${message} - 3.b` } },
+      ],
+    });
+    await simpleChildTask.batchTriggerAndWait({
+      items: [
+        { payload: { message: `${message} - 4.a` } },
+        { payload: { message: `${message} - 4.b` } },
+      ],
+    });
+
+    return {
+      hello: "world",
+    };
+  },
+});
+
+export const triggerAndWaitLoops = task({
+  id: "trigger-wait-loops",
+  run: async ({ message = "test" }: { message?: string }) => {
+    for (let i = 0; i < 2; i++) {
+      await simpleChildTask.triggerAndWait({ payload: { message: `${message} - ${i}` } });
+    }
+
+    for (let i = 0; i < 2; i++) {
+      await simpleChildTask.batchTriggerAndWait({
+        items: [
+          { payload: { message: `${message} - ${i}.a` } },
+          { payload: { message: `${message} - ${i}.b` } },
+        ],
+        // batchOptions: { maxConcurrency: 1 },
+      });
+    }
+
+    // Don't do this!
+    // await Promise.all(
+    //   [{ message: `${message} - 1` }, { message: `${message} - 2` }].map((payload) =>
+    //     simpleChildTask.triggerAndWait({ payload })
+    //   )
+    // );
+  },
+});