child tasks inherit the queue timestamp from their parent tasks to prioritize completing child tasks based on when their parent started

Author: ericallam

apps/webapp/app/v3/marqs/index.server.ts

@@ -174,7 +174,7 @@ export class MarQS {
     messageId: string,
     messageData: Record<string, unknown>,
     concurrencyKey?: string,
-    timestamp?: number
+    timestamp?: number | Date
   ) {
     return await this.#trace(
       "enqueueMessage",
@@ -190,7 +190,12 @@ export class MarQS {
           data: messageData,
           queue: messageQueue,
           concurrencyKey,
-          timestamp: timestamp ?? Date.now(),
+          timestamp:
+            typeof timestamp === "undefined"
+              ? Date.now()
+              : typeof timestamp === "number"
+              ? timestamp
+              : timestamp.getTime(),
           messageId,
           parentQueue,
         };

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

@@ -94,7 +94,8 @@ export class EnqueueDelayedRunService extends BaseService {
         environmentId: run.runtimeEnvironment.id,
         environmentType: run.runtimeEnvironment.type,
       },
-      run.concurrencyKey ?? undefined
+      run.concurrencyKey ?? undefined,
+      run.queueTimestamp ?? undefined
     );
   }
 }

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

@@ -186,6 +186,7 @@ export class ResumeBatchRunService extends BaseService {
           dependentTaskAttemptId: dependentTaskAttempt.id,
         });
 
+        // TODO: use the new priority queue thingie
         await marqs?.enqueueMessage(
           environment,
           dependentRun.queue,

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

@@ -49,6 +49,8 @@ export class ResumeTaskDependencyService extends BaseService {
           runId: dependentRun.id,
         }
       );
+
+      // TODO: use the new priority queue thingie
       await marqs?.enqueueMessage(
         dependency.taskRun.runtimeEnvironment,
         dependentRun.queue,

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

@@ -186,6 +186,7 @@ export class TriggerTaskService extends BaseService {
                   taskIdentifier: true,
                   rootTaskRunId: true,
                   depth: true,
+                  queueTimestamp: true,
                 },
               },
             },
@@ -242,6 +243,7 @@ export class TriggerTaskService extends BaseService {
                       taskIdentifier: true,
                       rootTaskRunId: true,
                       depth: true,
+                      queueTimestamp: true,
                     },
                   },
                 },
@@ -367,6 +369,12 @@ export class TriggerTaskService extends BaseService {
                   ? dependentBatchRun.dependentTaskAttempt.taskRun.depth + 1
                   : 0;
 
+                const queueTimestamp =
+                  dependentAttempt?.taskRun.queueTimestamp ??
+                  dependentBatchRun?.dependentTaskAttempt?.taskRun.queueTimestamp ??
+                  delayUntil ??
+                  new Date();
+
                 const taskRun = await tx.taskRun.create({
                   data: {
                     status: delayUntil ? "DELAYED" : "PENDING",
@@ -394,6 +402,7 @@ export class TriggerTaskService extends BaseService {
                     isTest: body.options?.test ?? false,
                     delayUntil,
                     queuedAt: delayUntil ? undefined : new Date(),
+                    queueTimestamp,
                     maxAttempts: body.options?.maxAttempts,
                     taskEventStore: getTaskEventStore(),
                     ttl,
@@ -573,7 +582,8 @@ export class TriggerTaskService extends BaseService {
                   environmentId: environment.id,
                   environmentType: environment.type,
                 },
-                body.options?.concurrencyKey
+                body.options?.concurrencyKey,
+                run.queueTimestamp ?? undefined
               );
             }
 

internal-packages/database/prisma/migrations/20250210164232_add_queue_timestamp_to_run/migration.sql

@@ -0,0 +1,8 @@
+-- DropIndex
+DROP INDEX "SecretStore_key_idx";
+
+-- AlterTable
+ALTER TABLE "TaskRun" ADD COLUMN     "queueTimestamp" TIMESTAMP(3);
+
+-- CreateIndex
+CREATE INDEX "SecretStore_key_idx" ON "SecretStore"("key" text_pattern_ops);

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

@@ -129,3 +129,110 @@ export const testReserveConcurrencyRecursiveWaits = task({
     };
   },
 });
+
+export const testChildTaskPriorityController = task({
+  id: "test-child-task-priority-controller",
+  run: async ({ delay = 5000 }: { delay: number }) => {
+    logger.info(`Delaying for ${delay}ms`);
+
+    await new Promise((resolve) => setTimeout(resolve, delay));
+
+    return {
+      completedAt: new Date(),
+    };
+  },
+});
+
+export const testChildTaskPriorityParent = task({
+  id: "test-child-task-priority-parent",
+  run: async ({ delay = 5000 }: { delay: number }) => {
+    logger.info(`Delaying for ${delay}ms`);
+
+    await new Promise((resolve) => setTimeout(resolve, delay));
+
+    await testChildTaskPriorityChild.triggerAndWait({
+      delay,
+    });
+
+    logger.info(`Delay of ${delay}ms completed`);
+
+    return {
+      completedAt: new Date(),
+    };
+  },
+});
+
+export const testChildTaskPriorityChildCreator = task({
+  id: "test-child-task-priority-child-creator",
+  run: async ({ delay = 5000 }: { delay: number }) => {
+    await testChildTaskPriorityChild.batchTrigger([
+      { payload: { delay, propagate: false } },
+      { payload: { delay, propagate: false } },
+      { payload: { delay, propagate: false } },
+      { payload: { delay, propagate: false } },
+    ]);
+
+    return {
+      completedAt: new Date(),
+    };
+  },
+});
+
+export const testChildTaskPriorityChild = task({
+  id: "test-child-task-priority-child",
+  queue: {
+    concurrencyLimit: 1,
+  },
+  run: async ({ delay = 5000, propagate }: { delay: number; propagate?: boolean }) => {
+    logger.info(`Delaying for ${delay}ms`);
+
+    await new Promise((resolve) => setTimeout(resolve, delay));
+
+    if (typeof propagate === "undefined" || propagate) {
+      await testChildTaskPriorityGrandChild.triggerAndWait({
+        delay,
+      });
+    }
+
+    logger.info(`Delay of ${delay}ms completed`);
+
+    return {
+      completedAt: new Date(),
+    };
+  },
+});
+
+export const testChildTaskPriorityGrandChildCreator = task({
+  id: "test-child-task-priority-grand-child-creator",
+  run: async ({ delay = 5000 }: { delay: number }) => {
+    await testChildTaskPriorityGrandChild.batchTrigger([
+      { payload: { delay } },
+      { payload: { delay } },
+      { payload: { delay } },
+    ]);
+
+    logger.info(`Delay of ${delay}ms completed`);
+
+    return {
+      completedAt: new Date(),
+    };
+  },
+});
+
+export const testChildTaskPriorityGrandChild = task({
+  id: "test-child-task-priority-grandchild",
+  queue: {
+    concurrencyLimit: 1,
+  },
+  run: async ({ delay = 5000 }: { delay: number }) => {
+    logger.info(`Delaying for ${delay}ms`);
+
+    await new Promise((resolve) => setTimeout(resolve, delay));
+
+    logger.info(`Delay of ${delay}ms completed`);
+
+    return {
+      completedAt: new Date(),
+    };
+  },
+});