Merge branch 'main' into docs/python-doc-to-markdown

Author: D-K-P

.changeset/pre.json

@@ -33,7 +33,10 @@
     "shiny-kiwis-beam",
     "smart-coins-hammer",
     "sour-mirrors-accept",
+    "spotty-ducks-punch",
     "spotty-pants-wink",
+    "tender-jobs-collect",
+    "tidy-books-smell",
     "tiny-buckets-teach",
     "tricky-houses-invite",
     "two-tigers-dream",

.changeset/spotty-ducks-punch.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+fix: default machine config indexing now works

.changeset/tender-jobs-collect.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+TriggerApiError 4xx errors will no longer cause tasks to be retried

.changeset/tidy-books-smell.md

@@ -0,0 +1,8 @@
+---
+"trigger.dev": patch
+"@trigger.dev/core": patch
+---
+
+- Fix polling interval reset bug that could create duplicate intervals
+- Protect against unexpected attempt number changes
+- Prevent run execution zombies after warm starts

.vscode/launch.json

@@ -25,8 +25,8 @@
     {
       "type": "node-terminal",
       "request": "launch",
-      "name": "Debug fairDequeuingStrategy.test.ts",
-      "command": "pnpm run test -t FairDequeuingStrategy",
+      "name": "Debug triggerTask.test.ts",
+      "command": "pnpm run test --run ./test/engine/triggerTask.test.ts",
       "envFile": "${workspaceFolder}/.env",
       "cwd": "${workspaceFolder}/apps/webapp",
       "sourceMaps": true

apps/webapp/app/components/runs/v3/RunFilters.tsx

@@ -841,8 +841,8 @@ function BatchIdDropdown({
   if (batchId) {
     if (!batchId.startsWith("batch_")) {
       error = "Batch IDs start with 'batch_'";
-    } else if (batchId.length !== 27) {
-      error = "Batch IDs are 27 characters long";
+    } else if (batchId.length !== 27 && batchId.length !== 31) {
+      error = "Batch IDs are 27 or 31 characters long";
     }
   }
 

apps/webapp/app/components/runs/v3/SpanEvents.tsx

@@ -83,7 +83,13 @@ export function SpanEventError({
         time={spanEvent.time}
         titleClassName="text-rose-500"
       />
-      {enhancedException.message && <Callout variant="error">{enhancedException.message}</Callout>}
+      {enhancedException.message && (
+        <Callout variant="error">
+          <pre className="text-wrap font-sans text-sm font-normal text-rose-200">
+            {enhancedException.message}
+          </pre>
+        </Callout>
+      )}
       {enhancedException.link &&
         (enhancedException.link.magic === "CONTACT_FORM" ? (
           <Feedback

apps/webapp/app/env.server.ts

@@ -384,6 +384,7 @@ const EnvironmentSchema = z.object({
     .int()
     .default(60 * 1000 * 15),
   MARQS_SHARED_QUEUE_LIMIT: z.coerce.number().int().default(1000),
+  MARQS_MAXIMUM_QUEUE_PER_ENV_COUNT: z.coerce.number().int().default(50),
   MARQS_DEV_QUEUE_LIMIT: z.coerce.number().int().default(1000),
   MARQS_MAXIMUM_NACK_COUNT: z.coerce.number().int().default(64),
   MARQS_CONCURRENCY_LIMIT_BIAS: z.coerce.number().default(0.75),
@@ -459,7 +460,6 @@ const EnvironmentSchema = z.object({
   RUN_ENGINE_REUSE_SNAPSHOT_COUNT: z.coerce.number().int().default(0),
   RUN_ENGINE_MAXIMUM_ENV_COUNT: z.coerce.number().int().optional(),
   RUN_ENGINE_WORKER_SHUTDOWN_TIMEOUT_MS: z.coerce.number().int().default(60_000),
-  RUN_ENGINE_MAX_DEQUEUE_LOOP_ATTEMPTS: z.coerce.number().int().default(10),
 
   RUN_ENGINE_WORKER_REDIS_HOST: z
     .string()

apps/webapp/app/presenters/v3/NewAlertChannelPresenter.server.ts

@@ -100,28 +100,47 @@ async function getSlackChannelsForToken(integration: AuthenticatableIntegration)
 
 type Channels = Awaited<ReturnType<WebClient["conversations"]["list"]>>["channels"];
 
-async function getSlackConversationsPage(client: WebClient, nextCursor?: string) {
-  return client.conversations.list({
-    types: "public_channel,private_channel",
-    exclude_archived: true,
-    cursor: nextCursor,
-  });
-}
-
 async function getAllSlackConversations(client: WebClient) {
   let nextCursor: string | undefined = undefined;
   let channels: Channels = [];
 
-  do {
-    const response = await getSlackConversationsPage(client, nextCursor);
-
-    if (!response.ok) {
-      throw new Error(`Failed to get channels: ${response.error}`);
+  try {
+    do {
+      // The `tryCatch` util runs into a type error due to self referencing.
+      // So we fall back to a good old try/catch block here.
+      const response = await client.conversations.list({
+        types: "public_channel,private_channel",
+        exclude_archived: true,
+        cursor: nextCursor,
+        limit: 999,
+      });
+
+      channels = channels.concat(response.channels ?? []);
+      nextCursor = response.response_metadata?.next_cursor;
+    } while (nextCursor);
+  } catch (error) {
+    if (error && isSlackError(error) && error.data.error === "ratelimited") {
+      logger.warn("Rate limiting issue occurred while fetching Slack channels", {
+        error,
+      });
+
+      // This is a workaround to the current way we handle Slack channels for creating alerts.
+      // For workspaces with a lot of channels (>10000) we might hit the rate limit for this slack endpoint,
+      // as multiple requests are needed to fetch all channels.
+      // We use the largest allowed page size of 999 to reduce the chance of hitting the rate limit.
+
+      // This is mainly due to workspaces with a large number of archived channels,
+      // which this slack endpoint unfortunately filters out only after fetching the page of channels without applying any filters first.
+      // https://api.slack.com/methods/conversations.list#markdown
+
+      // We expect most workspaces not to run into this issue, but if they do, we at least return some channels.
+      // In the future, we might revisit the way we handle Slack channels and cache them on our side to support
+      // proper searching. Until then, we track occurrences of this issue using a metric.
+      return channels;
     }
 
-    channels = channels.concat(response.channels ?? []);
-    nextCursor = response.response_metadata?.next_cursor;
-  } while (nextCursor);
+    throw error;
+  }
 
   return channels;
 }

apps/webapp/app/routes/api.v1.tasks.$taskId.trigger.ts

@@ -7,6 +7,7 @@ import {
 import { TaskRun } from "@trigger.dev/database";
 import { z } from "zod";
 import { env } from "~/env.server";
+import { EngineServiceValidationError } from "~/runEngine/concerns/errors";
 import { AuthenticatedEnvironment, getOneTimeUseToken } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
 import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
@@ -116,7 +117,9 @@ const { action, loader } = createActionApiRoute(
       );
     } catch (error) {
       if (error instanceof ServiceValidationError) {
-        return json({ error: error.message }, { status: 422 });
+        return json({ error: error.message }, { status: error.status ?? 422 });
+      } else if (error instanceof EngineServiceValidationError) {
+        return json({ error: error.message }, { status: error.status ?? 422 });
       } else if (error instanceof OutOfEntitlementError) {
         return json({ error: error.message }, { status: 422 });
       } else if (error instanceof Error) {

apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.spans.$spanParam/route.tsx

@@ -808,7 +808,13 @@ function RunError({ error }: { error: TaskRunError }) {
       return (
         <div className="flex flex-col gap-2 rounded-sm border border-rose-500/50 px-3 pb-3 pt-2">
           <Header3 className="text-rose-500">{name}</Header3>
-          {enhancedError.message && <Callout variant="error">{enhancedError.message}</Callout>}
+          {enhancedError.message && (
+            <Callout variant="error">
+              <pre className="text-wrap font-sans text-sm font-normal text-rose-200">
+                {enhancedError.message}
+              </pre>
+            </Callout>
+          )}
           {enhancedError.link &&
             (enhancedError.link.magic === "CONTACT_FORM" ? (
               <Feedback

apps/webapp/app/runEngine/concerns/errors.ts

@@ -0,0 +1,6 @@
+export class EngineServiceValidationError extends Error {
+  constructor(message: string, public status?: number) {
+    super(message);
+    this.name = "EngineServiceValidationError";
+  }
+}

apps/webapp/app/runEngine/concerns/idempotencyKeys.server.ts

@@ -0,0 +1,96 @@
+import { RunId } from "@trigger.dev/core/v3/isomorphic";
+import type { PrismaClientOrTransaction, TaskRun } from "@trigger.dev/database";
+import { logger } from "~/services/logger.server";
+import { resolveIdempotencyKeyTTL } from "~/utils/idempotencyKeys.server";
+import type { RunEngine } from "~/v3/runEngine.server";
+import type { TraceEventConcern, TriggerTaskRequest } from "../types";
+
+export type IdempotencyKeyConcernResult =
+  | { isCached: true; run: TaskRun }
+  | { isCached: false; idempotencyKey?: string; idempotencyKeyExpiresAt?: Date };
+
+export class IdempotencyKeyConcern {
+  constructor(
+    private readonly prisma: PrismaClientOrTransaction,
+    private readonly engine: RunEngine,
+    private readonly traceEventConcern: TraceEventConcern
+  ) {}
+
+  async handleTriggerRequest(request: TriggerTaskRequest): Promise<IdempotencyKeyConcernResult> {
+    const idempotencyKey = request.options?.idempotencyKey ?? request.body.options?.idempotencyKey;
+    const idempotencyKeyExpiresAt =
+      request.options?.idempotencyKeyExpiresAt ??
+      resolveIdempotencyKeyTTL(request.body.options?.idempotencyKeyTTL) ??
+      new Date(Date.now() + 24 * 60 * 60 * 1000 * 30); // 30 days
+
+    if (!idempotencyKey) {
+      return { isCached: false, idempotencyKey, idempotencyKeyExpiresAt };
+    }
+
+    const existingRun = idempotencyKey
+      ? await this.prisma.taskRun.findFirst({
+          where: {
+            runtimeEnvironmentId: request.environment.id,
+            idempotencyKey,
+            taskIdentifier: request.taskId,
+          },
+          include: {
+            associatedWaitpoint: true,
+          },
+        })
+      : undefined;
+
+    if (existingRun) {
+      if (existingRun.idempotencyKeyExpiresAt && existingRun.idempotencyKeyExpiresAt < new Date()) {
+        logger.debug("[TriggerTaskService][call] Idempotency key has expired", {
+          idempotencyKey: request.options?.idempotencyKey,
+          run: existingRun,
+        });
+
+        // Update the existing run to remove the idempotency key
+        await this.prisma.taskRun.updateMany({
+          where: { id: existingRun.id, idempotencyKey },
+          data: { idempotencyKey: null, idempotencyKeyExpiresAt: null },
+        });
+      } else {
+        const associatedWaitpoint = existingRun.associatedWaitpoint;
+        const parentRunId = request.body.options?.parentRunId;
+        const resumeParentOnCompletion = request.body.options?.resumeParentOnCompletion;
+        //We're using `andWait` so we need to block the parent run with a waitpoint
+        if (associatedWaitpoint && resumeParentOnCompletion && parentRunId) {
+          await this.traceEventConcern.traceIdempotentRun(
+            request,
+            {
+              existingRun,
+              idempotencyKey,
+              incomplete: associatedWaitpoint.status === "PENDING",
+              isError: associatedWaitpoint.outputIsError,
+            },
+            async (event) => {
+              //block run with waitpoint
+              await this.engine.blockRunWithWaitpoint({
+                runId: RunId.fromFriendlyId(parentRunId),
+                waitpoints: associatedWaitpoint.id,
+                spanIdToComplete: event.spanId,
+                batch: request.options?.batchId
+                  ? {
+                      id: request.options.batchId,
+                      index: request.options.batchIndex ?? 0,
+                    }
+                  : undefined,
+                projectId: request.environment.projectId,
+                organizationId: request.environment.organizationId,
+                tx: this.prisma,
+                releaseConcurrency: request.body.options?.releaseConcurrency,
+              });
+            }
+          );
+        }
+
+        return { isCached: true, run: existingRun };
+      }
+    }
+
+    return { isCached: false, idempotencyKey, idempotencyKeyExpiresAt };
+  }
+}

apps/webapp/app/runEngine/concerns/payloads.server.ts

@@ -0,0 +1,63 @@
+import { IOPacket, packetRequiresOffloading, tryCatch } from "@trigger.dev/core/v3";
+import { PayloadProcessor, TriggerTaskRequest } from "../types";
+import { env } from "~/env.server";
+import { startActiveSpan } from "~/v3/tracer.server";
+import { uploadPacketToObjectStore } from "~/v3/r2.server";
+import { EngineServiceValidationError } from "./errors";
+
+export class DefaultPayloadProcessor implements PayloadProcessor {
+  async process(request: TriggerTaskRequest): Promise<IOPacket> {
+    return await startActiveSpan("handlePayloadPacket()", async (span) => {
+      const payload = request.body.payload;
+      const payloadType = request.body.options?.payloadType ?? "application/json";
+
+      const packet = this.#createPayloadPacket(payload, payloadType);
+
+      if (!packet.data) {
+        return packet;
+      }
+
+      const { needsOffloading, size } = packetRequiresOffloading(
+        packet,
+        env.TASK_PAYLOAD_OFFLOAD_THRESHOLD
+      );
+
+      span.setAttribute("needsOffloading", needsOffloading);
+      span.setAttribute("size", size);
+
+      if (!needsOffloading) {
+        return packet;
+      }
+
+      const filename = `${request.friendlyId}/payload.json`;
+
+      const [uploadError] = await tryCatch(
+        uploadPacketToObjectStore(filename, packet.data, packet.dataType, request.environment)
+      );
+
+      if (uploadError) {
+        throw new EngineServiceValidationError(
+          "Failed to upload large payload to object store",
+          500
+        ); // This is retryable
+      }
+
+      return {
+        data: filename,
+        dataType: "application/store",
+      };
+    });
+  }
+
+  #createPayloadPacket(payload: any, payloadType: string): IOPacket {
+    if (payloadType === "application/json") {
+      return { data: JSON.stringify(payload), dataType: "application/json" };
+    }
+
+    if (typeof payload === "string") {
+      return { data: payload, dataType: payloadType };
+    }
+
+    return { dataType: payloadType };
+  }
+}