centralize queue timestamp logic in EnqueueSystem, adding queueTimestamp support and propagation

Author: ericallam

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

@@ -371,6 +371,10 @@ export class RunEngineTriggerTaskService extends WithRunEngine {
                     machine: body.options?.machine,
                     priorityMs: body.options?.priority ? body.options.priority * 1_000 : undefined,
                     releaseConcurrency: body.options?.releaseConcurrency,
+                    queueTimestamp:
+                      parentRun && body.options?.resumeParentOnCompletion
+                        ? parentRun.queueTimestamp ?? undefined
+                        : undefined,
                   },
                   this._prisma
                 );

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

@@ -334,6 +334,7 @@ export class RunEngine {
       maxAttempts,
       taskEventStore,
       priorityMs,
+      queueTimestamp,
       ttl,
       tags,
       parentTaskRunId,
@@ -414,6 +415,7 @@ export class RunEngine {
               maxAttempts,
               taskEventStore,
               priorityMs,
+              queueTimestamp: queueTimestamp ?? delayUntil ?? new Date(),
               ttl,
               tags:
                 tags.length === 0
@@ -520,7 +522,6 @@ export class RunEngine {
             await this.enqueueSystem.enqueueRun({
               run: taskRun,
               env: environment,
-              timestamp: Date.now() - taskRun.priorityMs,
               workerId,
               runnerId,
               tx: prisma,

internal-packages/run-engine/src/engine/systems/checkpointSystem.ts

@@ -161,7 +161,6 @@ export class CheckpointSystem {
         const newSnapshot = await this.enqueueSystem.enqueueRun({
           run,
           env: run.runtimeEnvironment,
-          timestamp: run.createdAt.getTime() - run.priorityMs,
           snapshot: {
             status: "QUEUED",
             description:

internal-packages/run-engine/src/engine/systems/delayedRunSystem.ts

@@ -98,7 +98,6 @@ export class DelayedRunSystem {
     await this.enqueueSystem.enqueueRun({
       run,
       env: run.runtimeEnvironment,
-      timestamp: run.createdAt.getTime() - run.priorityMs,
       batchId: run.batchId ?? undefined,
     });
 

internal-packages/run-engine/src/engine/systems/enqueueSystem.ts

@@ -25,7 +25,6 @@ export class EnqueueSystem {
   public async enqueueRun({
     run,
     env,
-    timestamp,
     tx,
     snapshot,
     previousSnapshotId,
@@ -37,7 +36,6 @@ export class EnqueueSystem {
   }: {
     run: TaskRun;
     env: MinimalAuthenticatedEnvironment;
-    timestamp: number;
     tx?: PrismaClientOrTransaction;
     snapshot?: {
       status?: Extract<TaskRunExecutionStatus, "QUEUED" | "QUEUED_EXECUTING">;
@@ -81,6 +79,8 @@ export class EnqueueSystem {
         masterQueues.push(run.secondaryMasterQueue);
       }
 
+      const timestamp = (run.queueTimestamp ?? run.createdAt).getTime() - run.priorityMs;
+
       await this.$.runQueue.enqueueMessage({
         env,
         masterQueues,

internal-packages/run-engine/src/engine/systems/pendingVersionSystem.ts

@@ -97,9 +97,6 @@ export class PendingVersionSystem {
         await this.enqueueSystem.enqueueRun({
           run: updatedRun,
           env: backgroundWorker.runtimeEnvironment,
-          //add to the queue using the original run created time
-          //this should ensure they're in the correct order in the queue
-          timestamp: updatedRun.createdAt.getTime() - updatedRun.priorityMs,
           tx,
         });
       });

internal-packages/run-engine/src/engine/systems/waitpointSystem.ts

@@ -540,7 +540,6 @@ export class WaitpointSystem {
           await this.enqueueSystem.enqueueRun({
             run,
             env: run.runtimeEnvironment,
-            timestamp: run.createdAt.getTime() - run.priorityMs,
             snapshot: {
               status: "QUEUED_EXECUTING",
               description: "Run can continue, but is waiting for concurrency",
@@ -564,7 +563,6 @@ export class WaitpointSystem {
         await this.enqueueSystem.enqueueRun({
           run,
           env: run.runtimeEnvironment,
-          timestamp: run.createdAt.getTime() - run.priorityMs,
           snapshot: {
             description: "Run was QUEUED, because all waitpoints are completed",
           },