Fix for permanently frozen tasks when at max concurrency

Heartbeats now free up concurrency for paused/frozen runs, then they don’t get crashed but get acked by the sharedQueueConsumer

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

@@ -306,7 +306,7 @@ export class SharedQueueConsumer {
           (!retryingFromCheckpoint &&
             !EXECUTABLE_RUN_STATUSES.withoutCheckpoint.includes(existingTaskRun.status))
         ) {
-          logger.error("Task run has invalid status for execution", {
+          logger.error("Task run has invalid status for execution. Going to ack", {
             queueMessage: message.data,
             messageId: message.messageId,
             taskRun: existingTaskRun.id,
@@ -314,12 +314,6 @@ export class SharedQueueConsumer {
             retryingFromCheckpoint,
           });
 
-          const service = new CrashTaskRunService();
-          await service.call(existingTaskRun.id, {
-            crashAttempts: true,
-            reason: `Invalid run status for execution: ${existingTaskRun.status}`,
-          });
-
           await this.#ackAndDoMoreWork(message.messageId);
           return;
         }

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

@@ -59,6 +59,9 @@ export class RequeueTaskRunService extends BaseService {
       case "WAITING_TO_RESUME":
       case "PAUSED": {
         logger.debug("[RequeueTaskRunService] Requeueing task run", { taskRun });
+
+        await marqs?.nackMessage(taskRun.id);
+
         break;
       }
       case "SYSTEM_FAILURE":

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

@@ -111,3 +111,40 @@ export const noop = task({
   id: "noop",
   run: async () => {},
 });
+
+export const fixedLengthTask = task({
+  id: "fixedLengthTask",
+  run: async ({ waitSeconds }: { waitSeconds: number }) => {
+    await new Promise((resolve) => setTimeout(resolve, waitSeconds * 1000));
+  },
+});
+
+export const permanentlyFrozen = task({
+  id: "permanently-frozen",
+  run: async ({ waitSeconds = 160, count = 1 }: { waitSeconds?: number; count?: number }) => {
+    for (let i = 0; i < count; i++) {
+      await fixedLengthTask.triggerAndWait({ waitSeconds });
+      logger.log(`Successfully complted task ${i}`);
+    }
+  },
+});
+
+//sending max concurrency of these at once will test the freeze + max concurrency logic
+export const bulkPermanentlyFrozen = task({
+  id: "bulk-permanently-frozen",
+  run: async ({
+    count = 1,
+    waitSeconds = 160,
+    grandChildCount = 1,
+  }: {
+    count?: number;
+    waitSeconds?: number;
+    grandChildCount?: number;
+  }) => {
+    await permanentlyFrozen.batchTrigger(
+      Array.from({ length: count }, (_, i) => ({
+        payload: { waitSeconds, count: grandChildCount },
+      }))
+    );
+  },
+});