Added priority support when triggering

Author: matt-aitken

internal-packages/run-engine/src/engine/index.ts

@@ -1305,6 +1305,7 @@ export class RunEngine {
             /** @deprecated */
             baseCostInCents: run.baseCostInCents,
             traceContext: run.traceContext as Record<string, string | undefined>,
+            priority: run.priorityMs === 0 ? undefined : run.priorityMs / 1_000,
           },
           queue: {
             id: queue.friendlyId,

packages/core/src/v3/schemas/api.ts

@@ -118,6 +118,7 @@ export const TriggerTaskRequestBody = z.object({
       tags: RunTags.optional(),
       test: z.boolean().optional(),
       ttl: z.string().or(z.number().nonnegative().int()).optional(),
+      priority: z.number().optional(),
     })
     .optional(),
 });
@@ -160,6 +161,7 @@ export const BatchTriggerTaskItem = z.object({
       tags: RunTags.optional(),
       test: z.boolean().optional(),
       ttl: z.string().or(z.number().nonnegative().int()).optional(),
+      priority: z.number().optional(),
     })
     .optional(),
 });

packages/core/src/v3/schemas/common.ts

@@ -229,6 +229,8 @@ export const TaskRun = z.object({
    * @link https://trigger.dev/docs/run-usage
    */
   baseCostInCents: z.number().default(0),
+  /** The priority of the run. Wih a value of 10 it will be dequeued before runs that were triggered 9 seconds before it (assuming they had no priority set).  */
+  priority: z.number().optional(),
 });
 
 export type TaskRun = z.infer<typeof TaskRun>;

packages/core/src/v3/types/tasks.ts

@@ -707,9 +707,26 @@ export type TriggerOptions = {
    * Specify a duration string like "1h", "10s", "30m", etc.
    */
   idempotencyKeyTTL?: string;
+
+  /**
+   * The maximum number of retry attempts for the task if it fails.
+   * If not specified, it will use the task or the default retry policy from your trigger.config file.
+   */
   maxAttempts?: number;
+
+  /**
+   * You can override the queue for the task. If a queue doesn't exist for the given name, it will be created.
+   * Setting the `concurrencyLimit` here will modify the limit for this queue everywhere it's used.
+   */
   queue?: TaskRunConcurrencyOptions;
+
+  /**
+   * The `concurrencyKey` creates a copy of the queue for every unique value of the key.
+   * For example, if the queue (set when triggering or on the task) has a concurrency limit of 10,
+   * and you set the concurrency key to `userId`, then each user will have their own queue with a concurrency limit of 10.
+   */
   concurrencyKey?: string;
+
   /**
    * The delay before the task is executed. This can be a string like "1h" or a Date object.
    *
@@ -739,6 +756,26 @@ export type TriggerOptions = {
    */
   ttl?: string | number;
 
+  /**
+   * If triggered at the same time, a higher priority run will be executed first.
+   *
+   The value is a time offset in seconds that determines the order of dequeuing.
+   * If you trigger two runs 9 seconds apart but the second one has `priority: 10`, it will be executed before the first one.
+   *
+   * @example
+   * ```ts
+   // no priority = 0
+   await myTask.trigger({ foo: "bar" });
+
+   //... imagine 9s pass by
+
+   // this run will start before the run above that was triggered 9s ago (with no priority)
+   await myTask.trigger({ foo: "bar" }, { priority: 10 });
+   ```
+   *
+   */
+  priority?: number;
+
   /**
    * Tags to attach to the run. Tags can be used to filter runs in the dashboard and using the SDK.
    *
@@ -777,6 +814,10 @@ export type TriggerOptions = {
 export type TriggerAndWaitOptions = TriggerOptions;
 
 export type BatchTriggerOptions = {
+  /**
+   * If no idempotencyKey is set on an individual item in the batch, it will use this key on each item + the array index.
+   * This is useful to prevent work being done again if the task has to retry.
+   */
   idempotencyKey?: IdempotencyKey | string | string[];
   idempotencyKeyTTL?: string;
 

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

@@ -619,6 +619,7 @@ export async function batchTriggerById<TTask extends AnyTask>(
                 (await makeIdempotencyKey(item.options?.idempotencyKey)) ?? batchItemIdempotencyKey,
               idempotencyKeyTTL: item.options?.idempotencyKeyTTL ?? options?.idempotencyKeyTTL,
               machine: item.options?.machine,
+              priority: item.options?.priority,
             },
           } satisfies BatchTriggerTaskV2RequestBody["items"][0];
         })
@@ -786,6 +787,7 @@ export async function batchTriggerByIdAndWait<TTask extends AnyTask>(
                     batchItemIdempotencyKey,
                   idempotencyKeyTTL: item.options?.idempotencyKeyTTL ?? options?.idempotencyKeyTTL,
                   machine: item.options?.machine,
+                  priority: item.options?.priority,
                 },
               } satisfies BatchTriggerTaskV2RequestBody["items"][0];
             })
@@ -944,6 +946,7 @@ export async function batchTriggerTasks<TTasks extends readonly AnyTask[]>(
                 (await makeIdempotencyKey(item.options?.idempotencyKey)) ?? batchItemIdempotencyKey,
               idempotencyKeyTTL: item.options?.idempotencyKeyTTL ?? options?.idempotencyKeyTTL,
               machine: item.options?.machine,
+              priority: item.options?.priority,
             },
           } satisfies BatchTriggerTaskV2RequestBody["items"][0];
         })
@@ -1113,6 +1116,7 @@ export async function batchTriggerAndWaitTasks<TTasks extends readonly AnyTask[]
                     batchItemIdempotencyKey,
                   idempotencyKeyTTL: item.options?.idempotencyKeyTTL ?? options?.idempotencyKeyTTL,
                   machine: item.options?.machine,
+                  priority: item.options?.priority,
                 },
               } satisfies BatchTriggerTaskV2RequestBody["items"][0];
             })
@@ -1184,6 +1188,7 @@ async function trigger_internal<TRunTypes extends AnyRunTypes>(
         maxDuration: options?.maxDuration,
         parentRunId: taskContext.ctx?.run.id,
         machine: options?.machine,
+        priority: options?.priority,
       },
     },
     {
@@ -1250,6 +1255,7 @@ async function batchTrigger_internal<TRunTypes extends AnyRunTypes>(
                 (await makeIdempotencyKey(item.options?.idempotencyKey)) ?? batchItemIdempotencyKey,
               idempotencyKeyTTL: item.options?.idempotencyKeyTTL ?? options?.idempotencyKeyTTL,
               machine: item.options?.machine,
+              priority: item.options?.priority,
             },
           } satisfies BatchTriggerTaskV2RequestBody["items"][0];
         })
@@ -1333,6 +1339,7 @@ async function triggerAndWait_internal<TIdentifier extends string, TPayload, TOu
             idempotencyKey: await makeIdempotencyKey(options?.idempotencyKey),
             idempotencyKeyTTL: options?.idempotencyKeyTTL,
             machine: options?.machine,
+            priority: options?.priority,
           },
         },
         {},
@@ -1418,6 +1425,7 @@ async function batchTriggerAndWait_internal<TIdentifier extends string, TPayload
                     batchItemIdempotencyKey,
                   idempotencyKeyTTL: item.options?.idempotencyKeyTTL ?? options?.idempotencyKeyTTL,
                   machine: item.options?.machine,
+                  priority: item.options?.priority,
                 },
               } satisfies BatchTriggerTaskV2RequestBody["items"][0];
             })

references/hello-world/src/trigger/priority.ts

@@ -0,0 +1,32 @@
+import { logger, task } from "@trigger.dev/sdk/v3";
+
+export const priorityParent = task({
+  id: "priority-parent",
+  run: async (payload: any, { ctx }) => {
+    logger.log("Hello, world from the parent", { payload });
+
+    const batch1 = await priorityChild.batchTriggerAndWait([
+      {
+        payload: { order: 1 },
+      },
+      {
+        payload: { order: 2 },
+        options: { priority: 1 },
+      },
+      {
+        payload: { order: 3 },
+        options: { priority: 2 },
+      },
+    ]);
+  },
+});
+
+export const priorityChild = task({
+  id: "priority-child",
+  queue: {
+    concurrencyLimit: 1,
+  },
+  run: async ({ order }: { order: number }, { ctx }) => {
+    logger.log(`Priority ${ctx.run.priority}`);
+  },
+});