WIP task run status, revamped resuming task dependencies

Author: ericallam

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.v3.$projectParam.test/route.tsx

@@ -52,7 +52,7 @@ export default function Page() {
         <div className={cn("grid h-full max-h-full grid-cols-1")}>
           <ResizablePanelGroup direction="horizontal" className="h-full max-h-full">
             <ResizablePanel order={1} minSize={20} defaultSize={30}>
-              <div className="flex flex-col px-3">
+              <div className="flex h-full max-h-full flex-col overflow-hidden px-3">
                 {tasks.length === 0 ? (
                   <NoTaskInstructions />
                 ) : (
@@ -81,7 +81,7 @@ function TaskSelector({ tasks }: { tasks: TaskListItem[] }) {
   const project = useProject();
 
   return (
-    <div className="flex flex-col divide-y divide-charcoal-800">
+    <div className="flex flex-col divide-y divide-charcoal-800 overflow-y-auto">
       {tasks.map((t) => (
         <NavLink key={t.id} to={v3TestTaskPath(organization, project, t)}>
           {({ isActive, isPending }) => (

apps/webapp/app/services/worker.server.ts

@@ -30,6 +30,9 @@ import { DeliverWebhookRequestService } from "./sources/deliverWebhookRequest.se
 import { PerformTaskOperationService } from "./tasks/performTaskOperation.server";
 import { ProcessCallbackTimeoutService } from "./tasks/processCallbackTimeout.server";
 import { ResumeTaskService } from "./tasks/resumeTask.server";
+import { ResumeTaskRunDependenciesService } from "~/v3/services/resumeTaskRunDependencies.server";
+import { ResumeBatchRunService } from "~/v3/services/resumeBatchRun.server";
+import { ResumeTaskDependencyService } from "~/v3/services/resumeTaskDependency.server";
 
 const workerCatalog = {
   indexEndpoint: z.object({
@@ -107,6 +110,17 @@ const workerCatalog = {
   "v3.indexDeployment": z.object({
     id: z.string(),
   }),
+  "v3.resumeTaskRunDependencies": z.object({
+    attemptId: z.string(),
+  }),
+  "v3.resumeBatchRun": z.object({
+    batchRunId: z.string(),
+    sourceTaskAttemptId: z.string(),
+  }),
+  "v3.resumeTaskDependency": z.object({
+    dependencyId: z.string(),
+    sourceTaskAttemptId: z.string(),
+  }),
 };
 
 const executionWorkerCatalog = {
@@ -443,6 +457,33 @@ function getWorkerQueue() {
           return await service.call(payload.id);
         },
       },
+      "v3.resumeTaskRunDependencies": {
+        priority: 0,
+        maxAttempts: 5,
+        handler: async (payload, job) => {
+          const service = new ResumeTaskRunDependenciesService();
+
+          return await service.call(payload.attemptId);
+        },
+      },
+      "v3.resumeBatchRun": {
+        priority: 0,
+        maxAttempts: 5,
+        handler: async (payload, job) => {
+          const service = new ResumeBatchRunService();
+
+          return await service.call(payload.batchRunId, payload.sourceTaskAttemptId);
+        },
+      },
+      "v3.resumeTaskDependency": {
+        priority: 0,
+        maxAttempts: 5,
+        handler: async (payload, job) => {
+          const service = new ResumeTaskDependencyService();
+
+          return await service.call(payload.dependencyId, payload.sourceTaskAttemptId);
+        },
+      },
     },
   });
 }

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

@@ -309,6 +309,7 @@ export class DevQueueConsumer {
       data: {
         lockedAt: new Date(),
         lockedById: backgroundTask.id,
+        status: "EXECUTING",
       },
       include: {
         attempts: {

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

@@ -31,9 +31,11 @@ const MessageBody = z.discriminatedUnion("type", [
   z.object({
     type: z.literal("RESUME"),
     completedAttemptIds: z.string().array(),
+    resumableAttemptId: z.string(),
   }),
   z.object({
     type: z.literal("RESUME_AFTER_DURATION"),
+    resumableAttemptId: z.string(),
   }),
 ]);
 
@@ -478,30 +480,36 @@ export class SharedQueueConsumer {
           return;
         }
 
-        const resumableRun = await prisma.taskRun.findFirst({
+        const resumableRun = await prisma.taskRun.findUnique({
           where: {
             id: message.messageId,
           },
+        });
+
+        if (!resumableRun) {
+          logger.error("Resumable run not found", {
+            queueMessage: message.data,
+            messageId: message.messageId,
+          });
+          await marqs?.acknowledgeMessage(message.messageId);
+          setTimeout(() => this.#doWork(), 100);
+          return;
+        }
+
+        const resumableAttempt = await prisma.taskRunAttempt.findUnique({
+          where: {
+            id: messageBody.data.resumableAttemptId,
+          },
           include: {
-            attempts: {
+            checkpoints: {
+              take: 1,
               orderBy: {
                 createdAt: "desc",
               },
-              take: 1,
-              include: {
-                checkpoints: {
-                  take: 1,
-                  orderBy: {
-                    createdAt: "desc",
-                  },
-                },
-              },
             },
           },
         });
 
-        const resumableAttempt = resumableRun?.attempts[0];
-
         if (!resumableAttempt) {
           logger.error("Resumable attempt not found", {
             queueMessage: message.data,
@@ -553,6 +561,13 @@ export class SharedQueueConsumer {
             },
             data: {
               status: "EXECUTING",
+              taskRun: {
+                update: {
+                  data: {
+                    status: "EXECUTING",
+                  },
+                },
+              },
             },
           });
 
@@ -647,30 +662,21 @@ export class SharedQueueConsumer {
       }
       // Resume after duration-based wait
       case "RESUME_AFTER_DURATION": {
-        const resumableRun = await prisma.taskRun.findFirst({
+        const resumableAttempt = await prisma.taskRunAttempt.findUnique({
           where: {
-            id: message.messageId,
+            id: messageBody.data.resumableAttemptId,
           },
           include: {
-            attempts: {
+            checkpoints: {
+              take: 1,
               orderBy: {
                 createdAt: "desc",
               },
-              take: 1,
-              include: {
-                checkpoints: {
-                  take: 1,
-                  orderBy: {
-                    createdAt: "desc",
-                  },
-                },
-              },
             },
+            taskRun: true,
           },
         });
 
-        const resumableAttempt = resumableRun?.attempts[0];
-
         if (!resumableAttempt) {
           logger.error("Resumable attempt not found", {
             queueMessage: message.data,