Improve the MarQS priority system by moving future messages into the LRE worker and using a priority timestamp offset to define priority in messages

Author: ericallam

apps/webapp/app/v3/legacyRunEngineWorker.server.ts

@@ -7,6 +7,7 @@ import { singleton } from "~/utils/singleton";
 import { TaskRunHeartbeatFailedService } from "./taskRunHeartbeatFailed.server";
 import { completeBatchTaskRunItemV3 } from "./services/batchTriggerV3.server";
 import { prisma } from "~/db.server";
+import { marqs } from "./marqs/index.server";
 
 function initializeWorker() {
   const redisOptions = {
@@ -49,6 +50,15 @@ function initializeWorker() {
           maxAttempts: 10,
         },
       },
+      scheduleRequeueMessage: {
+        schema: z.object({
+          messageId: z.string(),
+        }),
+        visibilityTimeoutMs: 60_000,
+        retry: {
+          maxAttempts: 5,
+        },
+      },
     },
     concurrency: {
       workers: env.LEGACY_RUN_ENGINE_WORKER_CONCURRENCY_WORKERS,
@@ -74,6 +84,9 @@ function initializeWorker() {
           attempt
         );
       },
+      scheduleRequeueMessage: async ({ payload }) => {
+        await marqs.requeueMessageById(payload.messageId);
+      },
     },
   });