Use priority offsets in MarQS and schedule future messages using redis worker (#1720)

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

* Add a test to make sure priority offsets don't unfairly favor environments

* requeuing should clear concurrency sets

* Heartbeats should only reschedule existing heartbeat jobs

* Fix type error

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);
+      },
     },
   });
 

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

@@ -0,0 +1,4 @@
+export const MARQS_RESUME_PRIORITY_TIMESTAMP_OFFSET = 31_556_952 * 1000; // 1 year
+export const MARQS_RETRY_PRIORITY_TIMESTAMP_OFFSET = 15_778_476 * 1000; // 6 months
+export const MARQS_DELAYED_REQUEUE_THRESHOLD_IN_MS = 500;
+export const MARQS_SCHEDULED_REQUEUE_AVAILABLE_AT_THRESHOLD_IN_MS = 500;