triggerdotdev · ericallam · Feb 20, 2025 · Feb 19, 2025 · Feb 20, 2025 · Feb 20, 2025
diff --git a/apps/webapp/app/v3/legacyRunEngineWorker.server.ts b/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);
+      },
    },
  });


diff --git a/apps/webapp/app/v3/marqs/constants.server.ts b/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;
diff --git a/apps/webapp/app/v3/marqs/envPriorityDequeuingStrategy.server.ts b/apps/webapp/app/v3/marqs/envPriorityDequeuingStrategy.server.ts