Merge branch 'main' into patch-1

Author: iFlyinq

.github/workflows/release.yml

@@ -69,7 +69,7 @@ jobs:
           commit: "chore: Update version for release"
           title: "chore: Update version for release"
           publish: pnpm run changeset:release
-          createGithubReleases: false
+          createGithubReleases: true
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
           NPM_TOKEN: ${{ secrets.NPM_TOKEN }}

apps/coordinator/src/index.ts

@@ -162,6 +162,49 @@ class TaskCoordinator {
 
           taskSocket.emit("RESUME_AFTER_DEPENDENCY", message);
         },
+        RESUME_AFTER_DEPENDENCY_WITH_ACK: async (message) => {
+          const taskSocket = await this.#getAttemptSocket(message.attemptFriendlyId);
+
+          if (!taskSocket) {
+            logger.log("Socket for attempt not found", {
+              attemptFriendlyId: message.attemptFriendlyId,
+            });
+            return {
+              success: false,
+              error: {
+                name: "SocketNotFoundError",
+                message: "Socket for attempt not found",
+              },
+            };
+          }
+
+          //if this is set, we want to kill the process because it will be resumed with the checkpoint from the queue
+          if (taskSocket.data.requiresCheckpointResumeWithMessage) {
+            logger.log("RESUME_AFTER_DEPENDENCY_WITH_ACK: Checkpoint is set so going to nack", {
+              socketData: taskSocket.data,
+            });
+
+            return {
+              success: false,
+              error: {
+                name: "CheckpointMessagePresentError",
+                message:
+                  "Checkpoint message is present, so we need to kill the process and resume from the queue.",
+              },
+            };
+          }
+
+          await chaosMonkey.call();
+
+          // In case the task resumed faster than we could checkpoint
+          this.#cancelCheckpoint(message.runId);
+
+          taskSocket.emit("RESUME_AFTER_DEPENDENCY", message);
+
+          return {
+            success: true,
+          };
+        },
         RESUME_AFTER_DURATION: async (message) => {
           const taskSocket = await this.#getAttemptSocket(message.attemptFriendlyId);
 
@@ -792,6 +835,18 @@ class TaskCoordinator {
             return;
           }
 
+          logger.log("WAIT_FOR_TASK checkpoint created", {
+            checkpoint,
+            socketData: socket.data,
+          });
+
+          //setting this means we can only resume from a checkpoint
+          socket.data.requiresCheckpointResumeWithMessage = `location:${checkpoint.location}-docker:${checkpoint.docker}`;
+          logger.log("WAIT_FOR_TASK set requiresCheckpointResumeWithMessage", {
+            checkpoint,
+            socketData: socket.data,
+          });
+
           const ack = await this.#platformSocket?.sendWithAck("CHECKPOINT_CREATED", {
             version: "v1",
             attemptFriendlyId: message.attemptFriendlyId,
@@ -804,6 +859,7 @@ class TaskCoordinator {
           });
 
           if (ack?.keepRunAlive) {
+            socket.data.requiresCheckpointResumeWithMessage = undefined;
             logger.log("keeping run alive after task checkpoint", { runId: socket.data.runId });
             return;
           }
@@ -862,6 +918,18 @@ class TaskCoordinator {
             return;
           }
 
+          logger.log("WAIT_FOR_BATCH checkpoint created", {
+            checkpoint,
+            socketData: socket.data,
+          });
+
+          //setting this means we can only resume from a checkpoint
+          socket.data.requiresCheckpointResumeWithMessage = `location:${checkpoint.location}-docker:${checkpoint.docker}`;
+          logger.log("WAIT_FOR_BATCH set checkpoint", {
+            checkpoint,
+            socketData: socket.data,
+          });
+
           const ack = await this.#platformSocket?.sendWithAck("CHECKPOINT_CREATED", {
             version: "v1",
             attemptFriendlyId: message.attemptFriendlyId,
@@ -875,6 +943,7 @@ class TaskCoordinator {
           });
 
           if (ack?.keepRunAlive) {
+            socket.data.requiresCheckpointResumeWithMessage = undefined;
             logger.log("keeping run alive after batch checkpoint", { runId: socket.data.runId });
             return;
           }

apps/webapp/app/env.server.ts

@@ -177,6 +177,11 @@ const EnvironmentSchema = z.object({
 
   LOOPS_API_KEY: z.string().optional(),
   MARQS_DISABLE_REBALANCING: z.coerce.boolean().default(false),
+  MARQS_VISIBILITY_TIMEOUT_MS: z.coerce
+    .number()
+    .int()
+    .default(60 * 1000 * 15),
+  PROD_TASK_HEARTBEAT_INTERVAL_MS: z.coerce.number().int().optional(),
 
   VERBOSE_GRAPHILE_LOGGING: z.string().default("false"),
   V2_MARQS_ENABLED: z.string().default("0"),

apps/webapp/app/hooks/useSearchParam.ts

@@ -18,13 +18,13 @@ export function useSearchParams() {
         }
 
         if (typeof value === "string") {
-          search.set(param, encodeURIComponent(value));
+          search.set(param, value);
           continue;
         }
 
         search.delete(param);
         for (const v of value) {
-          search.append(param, encodeURIComponent(v));
+          search.append(param, v);
         }
       }
     },

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.v3.$projectParam.runs.electric.$runParam/route.tsx

@@ -1155,7 +1155,7 @@ function ConnectedDevWarning() {
           <Paragraph variant="small">
             Runs usually start within 2 seconds in{" "}
             <EnvironmentLabel environment={{ type: "DEVELOPMENT" }} />. Check you're running the
-            CLI: <InlineCode className="whitespace-nowrap">npx trigger.dev@beta dev</InlineCode>
+            CLI: <InlineCode className="whitespace-nowrap">npx trigger.dev@latest dev</InlineCode>
           </Paragraph>
         </div>
       </Callout>

apps/webapp/app/v3/environmentVariables/environmentVariablesRepository.server.ts

@@ -798,6 +798,15 @@ async function resolveBuiltInProdVariables(runtimeEnvironment: RuntimeEnvironmen
     ]);
   }
 
+  if (env.PROD_TASK_HEARTBEAT_INTERVAL_MS) {
+    result = result.concat([
+      {
+        key: "HEARTBEAT_INTERVAL_MS",
+        value: String(env.PROD_TASK_HEARTBEAT_INTERVAL_MS),
+      },
+    ]);
+  }
+
   const commonVariables = await resolveCommonBuiltInVariables(runtimeEnvironment);
 
   return [...result, ...commonVariables];

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

@@ -162,8 +162,8 @@ export class DevQueueConsumer {
   /**
    * @deprecated Use `taskRunHeartbeat` instead
    */
-  public async taskHeartbeat(workerId: string, id: string, seconds: number = 60) {
-    logger.debug("[DevQueueConsumer] taskHeartbeat()", { id, seconds });
+  public async taskHeartbeat(workerId: string, id: string) {
+    logger.debug("[DevQueueConsumer] taskHeartbeat()", { id });
 
     const taskRunAttempt = await prisma.taskRunAttempt.findUnique({
       where: { friendlyId: id },
@@ -173,13 +173,13 @@ export class DevQueueConsumer {
       return;
     }
 
-    await marqs?.heartbeatMessage(taskRunAttempt.taskRunId, seconds);
+    await marqs?.heartbeatMessage(taskRunAttempt.taskRunId);
   }
 
-  public async taskRunHeartbeat(workerId: string, id: string, seconds: number = 60) {
-    logger.debug("[DevQueueConsumer] taskRunHeartbeat()", { id, seconds });
+  public async taskRunHeartbeat(workerId: string, id: string) {
+    logger.debug("[DevQueueConsumer] taskRunHeartbeat()", { id });
 
-    await marqs?.heartbeatMessage(id, seconds);
+    await marqs?.heartbeatMessage(id);
   }
 
   public async stop(reason: string = "CLI disconnected") {

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

@@ -698,8 +698,8 @@ export class MarQS {
   }
 
   // This should increment by the number of seconds, but with a max value of Date.now() + visibilityTimeoutInMs
-  public async heartbeatMessage(messageId: string, seconds: number = 30) {
-    await this.options.visibilityTimeoutStrategy.heartbeat(messageId, seconds * 1000);
+  public async heartbeatMessage(messageId: string) {
+    await this.options.visibilityTimeoutStrategy.heartbeat(messageId, this.visibilityTimeoutInMs);
   }
 
   get visibilityTimeoutInMs() {
@@ -1871,7 +1871,7 @@ function getMarQSClient() {
         redis: redisOptions,
         defaultEnvConcurrency: env.DEFAULT_ENV_EXECUTION_CONCURRENCY_LIMIT,
         defaultOrgConcurrency: env.DEFAULT_ORG_EXECUTION_CONCURRENCY_LIMIT,
-        visibilityTimeoutInMs: 120 * 1000, // 2 minutes,
+        visibilityTimeoutInMs: env.MARQS_VISIBILITY_TIMEOUT_MS,
         enableRebalancing: !env.MARQS_DISABLE_REBALANCING,
         subscriber: concurrencyTracker,
       });

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

@@ -725,20 +725,47 @@ export class SharedQueueConsumer {
         }
 
         try {
-          logger.debug("Broadcasting RESUME_AFTER_DEPENDENCY", {
-            runId: resumableAttempt.taskRunId,
-            attemptId: resumableAttempt.id,
-          });
-
-          // The attempt should still be running so we can broadcast to all coordinators to resume immediately
-          socketIo.coordinatorNamespace.emit("RESUME_AFTER_DEPENDENCY", {
-            version: "v1",
+          const resumeMessage = {
+            version: "v1" as const,
             runId: resumableAttempt.taskRunId,
             attemptId: resumableAttempt.id,
             attemptFriendlyId: resumableAttempt.friendlyId,
             completions,
             executions,
+          };
+
+          logger.debug("Broadcasting RESUME_AFTER_DEPENDENCY_WITH_ACK", { resumeMessage, message });
+
+          // The attempt should still be running so we can broadcast to all coordinators to resume immediately
+          const responses = await socketIo.coordinatorNamespace
+            .timeout(10_000)
+            .emitWithAck("RESUME_AFTER_DEPENDENCY_WITH_ACK", resumeMessage);
+
+          logger.debug("RESUME_AFTER_DEPENDENCY_WITH_ACK received", {
+            resumeMessage,
+            responses,
+            message,
           });
+
+          if (responses.length === 0) {
+            logger.error("RESUME_AFTER_DEPENDENCY_WITH_ACK no response", {
+              resumeMessage,
+              message,
+            });
+            await this.#nackAndDoMoreWork(message.messageId, this._options.nextTickInterval, 5_000);
+            return;
+          }
+
+          const hasSuccess = responses.some((response) => response.success);
+          if (!hasSuccess) {
+            logger.warn("RESUME_AFTER_DEPENDENCY_WITH_ACK failed", {
+              resumeMessage,
+              responses,
+              message,
+            });
+            await this.#nackAndDoMoreWork(message.messageId, this._options.nextTickInterval, 5_000);
+            return;
+          }
         } catch (e) {
           if (e instanceof Error) {
             this._currentSpan?.recordException(e);
@@ -748,7 +775,12 @@ export class SharedQueueConsumer {
 
           this._endSpanInNextIteration = true;
 
-          await this.#nackAndDoMoreWork(message.messageId);
+          logger.error("RESUME_AFTER_DEPENDENCY_WITH_ACK threw, nacking with delay", {
+            message,
+            error: e,
+          });
+
+          await this.#nackAndDoMoreWork(message.messageId, this._options.nextTickInterval, 5_000);
           return;
         }
 
@@ -1169,8 +1201,8 @@ class SharedQueueTasks {
     } satisfies TaskRunExecutionLazyAttemptPayload;
   }
 
-  async taskHeartbeat(attemptFriendlyId: string, seconds: number = 60) {
-    logger.debug("[SharedQueueConsumer] taskHeartbeat()", { id: attemptFriendlyId, seconds });
+  async taskHeartbeat(attemptFriendlyId: string) {
+    logger.debug("[SharedQueueConsumer] taskHeartbeat()", { id: attemptFriendlyId });
 
     const taskRunAttempt = await prisma.taskRunAttempt.findUnique({
       where: { friendlyId: attemptFriendlyId },
@@ -1180,13 +1212,13 @@ class SharedQueueTasks {
       return;
     }
 
-    await marqs?.heartbeatMessage(taskRunAttempt.taskRunId, seconds);
+    await marqs?.heartbeatMessage(taskRunAttempt.taskRunId);
   }
 
-  async taskRunHeartbeat(runId: string, seconds: number = 60) {
-    logger.debug("[SharedQueueConsumer] taskRunHeartbeat()", { runId, seconds });
+  async taskRunHeartbeat(runId: string) {
+    logger.debug("[SharedQueueConsumer] taskRunHeartbeat()", { runId });
 
-    await marqs?.heartbeatMessage(runId, seconds);
+    await marqs?.heartbeatMessage(runId);
   }
 
   public async taskRunFailed(completion: TaskRunFailedExecutionResult) {

apps/webapp/app/v3/services/resumeBatchRun.server.ts

@@ -132,7 +132,9 @@ export class ResumeBatchRunService extends BaseService {
       if (wasUpdated) {
         logger.debug("ResumeBatchRunService: Resuming dependent run without checkpoint", {
           batchRunId: batchRun.id,
-          dependentTaskAttemptId: batchRun.dependentTaskAttempt.id,
+          dependentTaskAttempt: batchRun.dependentTaskAttempt,
+          checkpointEventId: batchRun.checkpointEventId,
+          hasCheckpointEvent: !!batchRun.checkpointEventId,
         });
         await marqs?.replaceMessage(dependentRun.id, {
           type: "RESUME",