Fix pnpm lock file

Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>

.changeset/breezy-gorillas-mate.md

@@ -0,0 +1,6 @@
+---
+"trigger.dev": patch
+"@trigger.dev/core": patch
+---
+
+better handle task metadata parse errors, and display nicely formatted errors

.changeset/cool-glasses-bake.md

@@ -0,0 +1,7 @@
+---
+"@trigger.dev/sdk": patch
+"trigger.dev": patch
+"@trigger.dev/core": patch
+---
+
+Configurable log levels in the config file and via env var

.changeset/khaki-apricots-design.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/core": patch
+---
+
+Retry 429, 500, and connection error API requests to the trigger.dev server

.changeset/mighty-camels-joke.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+Improve error messages during dev/deploy and handle deploy image build issues

.changeset/odd-poets-own.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+Added a Node.js runtime check for the CLI

.changeset/pre.json

@@ -44,12 +44,18 @@
     "@trigger.dev/yalt": "2.3.18"
   },
   "changesets": [
+    "breezy-gorillas-mate",
     "chilled-hornets-move",
     "clean-pianos-listen",
+    "cool-glasses-bake",
+    "khaki-apricots-design",
     "lemon-jobs-repair",
     "light-bulldogs-press",
     "many-ligers-pump",
+    "mighty-camels-joke",
+    "odd-poets-own",
     "real-planets-stare",
+    "smart-olives-eat",
     "strange-ghosts-matter",
     "stupid-bulldogs-applaud",
     "sweet-lizards-press",

.changeset/smart-olives-eat.md

@@ -0,0 +1,5 @@
+---
+"trigger.dev": patch
+---
+
+Fix CLI logout and add list-profiles command

apps/webapp/app/env.server.ts

@@ -63,6 +63,8 @@ const EnvironmentSchema = z.object({
   REDIS_PASSWORD: z.string().optional(),
   REDIS_TLS_DISABLED: z.string().optional(),
 
+  DEFAULT_QUEUE_EXECUTION_CONCURRENCY_LIMIT: z.coerce.number().int().default(5),
+  DEFAULT_ENV_EXECUTION_CONCURRENCY_LIMIT: z.coerce.number().int().default(10),
   DEFAULT_ORG_EXECUTION_CONCURRENCY_LIMIT: z.coerce.number().int().default(10),
   DEFAULT_DEV_ENV_EXECUTION_ATTEMPTS: z.coerce.number().int().positive().default(1),
 

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

@@ -1,8 +1,15 @@
+import {
+  DeploymentErrorData,
+  TaskMetadataFailedToParseData,
+  groupTaskMetadataIssuesByTask,
+} from "@trigger.dev/core/v3";
 import { WorkerDeployment, WorkerDeploymentStatus } from "@trigger.dev/database";
+import { z } from "zod";
 import { PrismaClient, prisma } from "~/db.server";
 import { Organization } from "~/models/organization.server";
 import { Project } from "~/models/project.server";
 import { User } from "~/models/user.server";
+import { safeJsonParse } from "~/utils/json";
 import { getUsername } from "~/utils/username";
 
 export class DeploymentPresenter {
@@ -51,6 +58,7 @@ export class DeploymentPresenter {
         id: true,
         shortCode: true,
         version: true,
+        errorData: true,
         environment: {
           select: {
             id: true,
@@ -120,7 +128,81 @@ export class DeploymentPresenter {
           userName: getUsername(deployment.environment.orgMember?.user),
         },
         deployedBy: deployment.triggeredBy,
+        errorData: this.#prepareErrorData(deployment.errorData),
       },
     };
   }
+
+  #prepareErrorData(errorData: WorkerDeployment["errorData"]) {
+    if (!errorData) {
+      return;
+    }
+
+    const parsedErrorData = DeploymentErrorData.safeParse(errorData);
+
+    if (!parsedErrorData.success) {
+      return;
+    }
+
+    if (parsedErrorData.data.name === "TaskMetadataParseError") {
+      const errorJson = safeJsonParse(parsedErrorData.data.stack);
+
+      if (errorJson) {
+        const parsedError = TaskMetadataFailedToParseData.safeParse(errorJson);
+
+        if (parsedError.success) {
+          return {
+            name: parsedErrorData.data.name,
+            message: parsedErrorData.data.message,
+            stack: createTaskMetadataFailedErrorStack(parsedError.data),
+          };
+        } else {
+          return {
+            name: parsedErrorData.data.name,
+            message: parsedErrorData.data.message,
+          };
+        }
+      } else {
+        return {
+          name: parsedErrorData.data.name,
+          message: parsedErrorData.data.message,
+        };
+      }
+    }
+
+    return {
+      name: parsedErrorData.data.name,
+      message: parsedErrorData.data.message,
+      stack: parsedErrorData.data.stack,
+    };
+  }
+}
+
+function createTaskMetadataFailedErrorStack(
+  data: z.infer<typeof TaskMetadataFailedToParseData>
+): string {
+  const stack = [];
+
+  const groupedIssues = groupTaskMetadataIssuesByTask(data.tasks, data.zodIssues);
+
+  for (const key in groupedIssues) {
+    const taskWithIssues = groupedIssues[key];
+
+    if (!taskWithIssues) {
+      continue;
+    }
+
+    stack.push("\n");
+    stack.push(`  ❯ ${taskWithIssues.exportName} in ${taskWithIssues.filePath}`);
+
+    for (const issue of taskWithIssues.issues) {
+      if (issue.path) {
+        stack.push(`    x ${issue.path} ${issue.message}`);
+      } else {
+        stack.push(`    x ${issue.message}`);
+      }
+    }
+  }
+
+  return stack.join("\n");
 }

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

@@ -3,6 +3,7 @@ import { LoaderFunctionArgs } from "@remix-run/server-runtime";
 import { typedjson, useTypedLoaderData } from "remix-typedjson";
 import { ExitIcon } from "~/assets/icons/ExitIcon";
 import { UserAvatar } from "~/components/UserProfilePhoto";
+import { CodeBlock } from "~/components/code/CodeBlock";
 import { EnvironmentLabel } from "~/components/environments/EnvironmentLabel";
 import { Badge } from "~/components/primitives/Badge";
 import { LinkButton } from "~/components/primitives/Buttons";
@@ -156,6 +157,26 @@ export default function Page() {
                 </TableBody>
               </Table>
             </div>
+          ) : deployment.errorData ? (
+            <div className="flex flex-col">
+              {deployment.errorData.stack ? (
+                <CodeBlock
+                  language="markdown"
+                  rowTitle={deployment.errorData.message}
+                  code={deployment.errorData.stack}
+                  maxLines={20}
+                />
+              ) : (
+                <div className="flex flex-col">
+                  <Paragraph
+                    variant="base/bright"
+                    className="w-full border-b border-grid-dimmed py-2.5"
+                  >
+                    {deployment.errorData.message}
+                  </Paragraph>
+                </div>
+              )}
+            </div>
           ) : null}
         </div>
       </div>

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.v3.$projectParam.environment-variables.new/route.tsx

@@ -185,7 +185,11 @@ export default function Page() {
                       >
                         <EnvironmentLabel environment={environment} className="h-5 px-2" />
                       </label>
-                      <Input name={`values[${index}].value`} placeholder="Not set" />
+                      <Input
+                        type="password"
+                        name={`values[${index}].value`}
+                        placeholder="Not set"
+                      />
                     </Fragment>
                   );
                 })}

apps/webapp/app/routes/admin.api.v1.environments.$environmentId.ts

@@ -0,0 +1,66 @@
+import { ActionFunctionArgs, json } from "@remix-run/server-runtime";
+import { z } from "zod";
+import { prisma } from "~/db.server";
+import { authenticateApiRequestWithPersonalAccessToken } from "~/services/personalAccessToken.server";
+import { marqs } from "~/v3/marqs/index.server";
+
+const ParamsSchema = z.object({
+  environmentId: z.string(),
+});
+
+const RequestBodySchema = z.object({
+  envMaximumConcurrencyLimit: z.number(),
+  orgMaximumConcurrencyLimit: z.number(),
+});
+
+export async function action({ request, params }: ActionFunctionArgs) {
+  // Next authenticate the request
+  const authenticationResult = await authenticateApiRequestWithPersonalAccessToken(request);
+
+  if (!authenticationResult) {
+    return json({ error: "Invalid or Missing API key" }, { status: 401 });
+  }
+
+  const user = await prisma.user.findUnique({
+    where: {
+      id: authenticationResult.userId,
+    },
+  });
+
+  if (!user) {
+    return json({ error: "Invalid or Missing API key" }, { status: 401 });
+  }
+
+  if (!user.admin) {
+    return json({ error: "You must be an admin to perform this action" }, { status: 403 });
+  }
+
+  const parsedParams = ParamsSchema.parse(params);
+
+  const rawBody = await request.json();
+  const body = RequestBodySchema.parse(rawBody);
+
+  const environment = await prisma.runtimeEnvironment.update({
+    where: {
+      id: parsedParams.environmentId,
+    },
+    data: {
+      maximumConcurrencyLimit: body.envMaximumConcurrencyLimit,
+      organization: {
+        update: {
+          data: {
+            maximumConcurrencyLimit: body.orgMaximumConcurrencyLimit,
+          },
+        },
+      },
+    },
+    include: {
+      organization: true,
+      project: true,
+    },
+  });
+
+  await marqs?.updateEnvConcurrencyLimits(environment);
+
+  return json({ success: true });
+}

apps/webapp/app/utils/json.ts

@@ -1,6 +1,10 @@
 import { z } from "zod";
 
-export function safeJsonParse(json: string): unknown {
+export function safeJsonParse(json?: string): unknown {
+  if (!json) {
+    return;
+  }
+
   try {
     return JSON.parse(json);
   } catch (e) {

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

@@ -13,7 +13,7 @@ import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
 import { EnvironmentVariablesRepository } from "../environmentVariables/environmentVariablesRepository.server";
 import { generateFriendlyId } from "../friendlyIdentifiers";
-import { marqs } from "../marqs.server";
+import { marqs } from "~/v3/marqs/index.server";
 import { CancelAttemptService } from "../services/cancelAttempt.server";
 import { CompleteAttemptService } from "../services/completeAttempt.server";
 import { attributesFromAuthenticatedEnv } from "../tracer.server";

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

@@ -1,48 +1,30 @@
+import { Span, SpanKind, SpanOptions, trace } from "@opentelemetry/api";
+import { flattenAttributes } from "@trigger.dev/core/v3";
 import Redis, { type Callback, type RedisOptions, type Result } from "ioredis";
-import { z } from "zod";
 import { env } from "~/env.server";
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
-import { singleton } from "~/utils/singleton";
-import { AsyncWorker } from "./marqs/asyncWorker.server";
 import { logger } from "~/services/logger.server";
-import { attributesFromAuthenticatedEnv } from "./tracer.server";
-import { Span, SpanKind, SpanOptions, trace } from "@opentelemetry/api";
+import { singleton } from "~/utils/singleton";
+import { attributesFromAuthenticatedEnv } from "../tracer.server";
+import { AsyncWorker } from "./asyncWorker.server";
+import { MarQSShortKeyProducer } from "./marqsKeyProducer.server";
+import { SimpleWeightedChoiceStrategy } from "./priorityStrategy.server";
+import {
+  MarQSKeyProducer,
+  MarQSQueuePriorityStrategy,
+  MessagePayload,
+  QueueCapacities,
+} from "./types";
 
 const tracer = trace.getTracer("marqs");
 
 const KEY_PREFIX = "marqs:";
 
-type MarQSOptions = {
-  redis: RedisOptions;
-  defaultConcurrency?: number;
-  windowSize?: number;
-  visibilityTimeoutInMs?: number;
-  workers: number;
-};
-
 const constants = {
   SHARED_QUEUE: "sharedQueue",
   MESSAGE_VISIBILITY_TIMEOUT_QUEUE: "msgVisibilityTimeout",
-  CURRENT_CONCURRENCY_PART: "currentConcurrency",
-  CONCURRENCY_LIMIT_PART: "concurrency",
-  ENV_PART: "env",
-  QUEUE_PART: "queue",
-  CONCURRENCY_KEY_PART: "ck",
-  MESSAGE_PART: "message",
 } as const;
 
-const MessagePayload = z.object({
-  version: z.literal("1"),
-  data: z.record(z.unknown()),
-  queue: z.string(),
-  messageId: z.string(),
-  timestamp: z.number(),
-  parentQueue: z.string(),
-  concurrencyKey: z.string().optional(),
-});
-
-type MessagePayload = z.infer<typeof MessagePayload>;
-
 const SemanticAttributes = {
   QUEUE: "marqs.queue",
   PARENT_QUEUE: "marqs.parentQueue",
@@ -50,11 +32,26 @@ const SemanticAttributes = {
   CONCURRENCY_KEY: "marqs.concurrencyKey",
 };
 
+export type MarQSOptions = {
+  redis: RedisOptions;
+  defaultQueueConcurrency: number;
+  defaultEnvConcurrency: number;
+  defaultOrgConcurrency: number;
+  windowSize?: number;
+  visibilityTimeoutInMs?: number;
+  workers: number;
+  keysProducer: MarQSKeyProducer;
+  queuePriorityStrategy: MarQSQueuePriorityStrategy;
+  envQueuePriorityStrategy: MarQSQueuePriorityStrategy;
+};
+
 /**
  * MarQS - Multitenant Asynchronous Reliable Queueing System (pronounced "markus")
  */
 export class MarQS {
   private redis: Redis;
+  private keys: MarQSKeyProducer;
+  private queuePriorityStrategy: MarQSQueuePriorityStrategy;
   #requeueingWorkers: Array<AsyncWorker> = [];
 
   constructor(private readonly options: MarQSOptions) {
@@ -62,19 +59,27 @@ export class MarQS {
 
     // Spawn options.workers workers to requeue visible messages
     this.#startRequeuingWorkers();
-
     this.#registerCommands();
+
+    this.keys = options.keysProducer;
+    this.queuePriorityStrategy = options.queuePriorityStrategy;
   }
 
-  public async updateQueueConcurrency(
+  public async updateQueueConcurrencyLimits(
     env: AuthenticatedEnvironment,
     queue: string,
     concurrency: number
   ) {
-    return this.redis.set(
-      `${constants.ENV_PART}:${env.id}:${constants.QUEUE_PART}:${queue}:${constants.CONCURRENCY_LIMIT_PART}`,
-      concurrency
-    );
+    return this.redis.set(this.keys.queueConcurrencyLimitKey(env, queue), concurrency);
+  }
+
+  public async updateEnvConcurrencyLimits(env: AuthenticatedEnvironment) {
+    await this.#callUpdateGlobalConcurrencyLimits({
+      envConcurrencyLimitKey: this.keys.envConcurrencyLimitKey(env),
+      orgConcurrencyLimitKey: this.keys.orgConcurrencyLimitKey(env),
+      envConcurrencyLimit: env.maximumConcurrencyLimit,
+      orgConcurrencyLimit: env.organization.maximumConcurrencyLimit,
+    });
   }
 
   public async enqueueMessage(
@@ -87,16 +92,11 @@ export class MarQS {
     return await this.#trace(
       "enqueueMessage",
       async (span) => {
-        const messageQueue = `${constants.ENV_PART}:${env.id}:${constants.QUEUE_PART}:${queue}${
-          concurrencyKey ? `:${constants.CONCURRENCY_KEY_PART}:${concurrencyKey}` : ""
-        }`;
+        const messageQueue = this.keys.queueKey(env, queue, concurrencyKey);
 
         const timestamp = Date.now();
 
-        const parentQueue =
-          env.type === "DEVELOPMENT"
-            ? `${constants.ENV_PART}:${env.id}:${constants.SHARED_QUEUE}`
-            : constants.SHARED_QUEUE;
+        const parentQueue = this.keys.envSharedQueueKey(env);
 
         const messagePayload: MessagePayload = {
           version: "1",
@@ -125,15 +125,13 @@ export class MarQS {
     return this.#trace(
       "dequeueMessageInEnv",
       async (span, abort) => {
-        const parentQueue =
-          env.type === "DEVELOPMENT"
-            ? `${constants.ENV_PART}:${env.id}:${constants.SHARED_QUEUE}`
-            : constants.SHARED_QUEUE;
+        const parentQueue = this.keys.envSharedQueueKey(env);
 
         // Read the parent queue for matching queues
         const messageQueue = await this.#getRandomQueueFromParentQueue(
           parentQueue,
-          (queue, score) => this.#calculateMessageQueueWeight(queue, score)
+          this.options.envQueuePriorityStrategy,
+          (queue) => this.#calculateMessageQueueCapacities(queue)
         );
 
         if (!messageQueue) {
@@ -141,15 +139,16 @@ export class MarQS {
           return;
         }
 
-        // If the queue includes a concurrency key, we need to remove the ck:concurrencyKey from the queue name
-        const concurrencyQueueName = messageQueue.replace(/:ck:.+$/, "");
-
         const messageData = await this.#callDequeueMessage({
           messageQueue,
           parentQueue,
           visibilityQueue: constants.MESSAGE_VISIBILITY_TIMEOUT_QUEUE,
-          concurrencyLimitKey: `${concurrencyQueueName}:${constants.CONCURRENCY_LIMIT_PART}`,
-          currentConcurrencyKey: `${messageQueue}:${constants.CURRENT_CONCURRENCY_PART}`,
+          concurrencyLimitKey: this.keys.concurrencyLimitKeyFromQueue(messageQueue),
+          currentConcurrencyKey: this.keys.currentConcurrencyKeyFromQueue(messageQueue),
+          envConcurrencyLimitKey: this.keys.envConcurrencyLimitKeyFromQueue(messageQueue),
+          envCurrentConcurrencyKey: this.keys.envCurrentConcurrencyKeyFromQueue(messageQueue),
+          orgConcurrencyLimitKey: this.keys.orgConcurrencyLimitKeyFromQueue(messageQueue),
+          orgCurrentConcurrencyKey: this.keys.orgCurrentConcurrencyKeyFromQueue(messageQueue),
         });
 
         if (!messageData) {
@@ -188,7 +187,8 @@ export class MarQS {
         // Read the parent queue for matching queues
         const messageQueue = await this.#getRandomQueueFromParentQueue(
           parentQueue,
-          (queue, score) => this.#calculateMessageQueueWeight(queue, score)
+          this.options.queuePriorityStrategy,
+          (queue) => this.#calculateMessageQueueCapacities(queue)
         );
 
         if (!messageQueue) {
@@ -197,14 +197,16 @@ export class MarQS {
         }
 
         // If the queue includes a concurrency key, we need to remove the ck:concurrencyKey from the queue name
-        const concurrencyQueueName = messageQueue.replace(/:ck:.+$/, "");
-
         const messageData = await this.#callDequeueMessage({
           messageQueue,
           parentQueue,
           visibilityQueue: constants.MESSAGE_VISIBILITY_TIMEOUT_QUEUE,
-          concurrencyLimitKey: `${concurrencyQueueName}:${constants.CONCURRENCY_LIMIT_PART}`,
-          currentConcurrencyKey: `${messageQueue}:${constants.CURRENT_CONCURRENCY_PART}`,
+          concurrencyLimitKey: this.keys.concurrencyLimitKeyFromQueue(messageQueue),
+          currentConcurrencyKey: this.keys.currentConcurrencyKeyFromQueue(messageQueue),
+          envConcurrencyLimitKey: this.keys.envConcurrencyLimitKeyFromQueue(messageQueue),
+          envCurrentConcurrencyKey: this.keys.envCurrentConcurrencyKeyFromQueue(messageQueue),
+          orgConcurrencyLimitKey: this.keys.orgConcurrencyLimitKeyFromQueue(messageQueue),
+          orgCurrentConcurrencyKey: this.keys.orgCurrentConcurrencyKeyFromQueue(messageQueue),
         });
 
         if (!messageData) {
@@ -249,9 +251,11 @@ export class MarQS {
         });
 
         await this.#callAcknowledgeMessage({
-          messageKey: `${constants.MESSAGE_PART}:${messageId}`,
+          messageKey: this.keys.messageKey(messageId),
           visibilityQueue: constants.MESSAGE_VISIBILITY_TIMEOUT_QUEUE,
-          concurrencyKey: `${message.queue}:${constants.CURRENT_CONCURRENCY_PART}`,
+          concurrencyKey: this.keys.currentConcurrencyKeyFromQueue(message.queue),
+          envConcurrencyKey: this.keys.envCurrentConcurrencyKeyFromQueue(message.queue),
+          orgConcurrencyKey: this.keys.orgCurrentConcurrencyKeyFromQueue(message.queue),
           messageId,
         });
       },
@@ -281,9 +285,11 @@ export class MarQS {
         });
 
         await this.#callAcknowledgeMessage({
-          messageKey: `${constants.MESSAGE_PART}:${messageId}`,
+          messageKey: this.keys.messageKey(messageId),
           visibilityQueue: constants.MESSAGE_VISIBILITY_TIMEOUT_QUEUE,
-          concurrencyKey: `${oldMessage.queue}:${constants.CURRENT_CONCURRENCY_PART}`,
+          concurrencyKey: this.keys.currentConcurrencyKeyFromQueue(oldMessage.queue),
+          envConcurrencyKey: this.keys.envCurrentConcurrencyKeyFromQueue(oldMessage.queue),
+          orgConcurrencyKey: this.keys.orgCurrentConcurrencyKeyFromQueue(oldMessage.queue),
           messageId,
         });
 
@@ -353,10 +359,12 @@ export class MarQS {
         });
 
         await this.#callNackMessage({
-          messageKey: `${constants.MESSAGE_PART}:${messageId}`,
+          messageKey: this.keys.messageKey(messageId),
           messageQueue: message.queue,
           parentQueue: message.parentQueue,
-          concurrencyKey: `${message.queue}:${constants.CURRENT_CONCURRENCY_PART}`,
+          concurrencyKey: this.keys.currentConcurrencyKeyFromQueue(message.queue),
+          envConcurrencyKey: this.keys.envCurrentConcurrencyKeyFromQueue(message.queue),
+          orgConcurrencyKey: this.keys.orgCurrentConcurrencyKeyFromQueue(message.queue),
           visibilityQueue: constants.MESSAGE_VISIBILITY_TIMEOUT_QUEUE,
           messageId,
           messageScore: retryAt,
@@ -384,7 +392,7 @@ export class MarQS {
     return this.#trace(
       "readMessage",
       async (span) => {
-        const rawMessage = await this.redis.get(`${constants.MESSAGE_PART}:${messageId}`);
+        const rawMessage = await this.redis.get(this.keys.messageKey(messageId));
 
         if (!rawMessage) {
           return;
@@ -409,81 +417,77 @@ export class MarQS {
 
   async #getRandomQueueFromParentQueue(
     parentQueue: string,
-    calculateWeight: (queue: string, score: number) => Promise<number>
+    queuePriorityStrategy: MarQSQueuePriorityStrategy,
+    calculateCapacities: (queue: string) => Promise<QueueCapacities>
   ) {
     return this.#trace(
       "getRandomQueueFromParentQueue",
       async (span, abort) => {
-        const queues = await this.#zrangeWithScores(parentQueue, 0, -1);
+        const { range, selectionId } = await queuePriorityStrategy.nextCandidateSelection(
+          parentQueue
+        );
 
-        if (queues.length === 0) {
+        const queues = await this.#zrangeWithScores(parentQueue, range[0], range[1]);
+
+        const queuesWithScores = await this.#calculateQueueScores(queues, calculateCapacities);
+
+        // We need to priority shuffle here to ensure all workers aren't just working on the highest priority queue
+        const choice = this.queuePriorityStrategy.chooseQueue(
+          queuesWithScores,
+          parentQueue,
+          selectionId
+        );
+
+        if (typeof choice !== "string") {
           abort();
           return;
         }
 
+        span.setAttributes({
+          ...flattenAttributes(queues, "marqs.queues"),
+        });
+        span.setAttributes({
+          ...flattenAttributes(queuesWithScores, "marqs.queuesWithScores"),
+        });
+        span.setAttribute("marqs.nextRange", range);
         span.setAttribute("marqs.queueCount", queues.length);
+        span.setAttribute("marqs.queueChoice", choice);
 
-        const queuesWithWeights = await this.#calculateQueueWeights(queues, calculateWeight);
-
-        // We need to priority shuffle here to ensure all workers aren't just working on the highest priority queue
-        return await this.#weightedRandomChoice(queuesWithWeights);
+        return choice;
       },
       { kind: SpanKind.CONSUMER, attributes: { [SemanticAttributes.PARENT_QUEUE]: parentQueue } }
     );
   }
 
   // Calculate the weights of the queues based on the age and the capacity
-  async #calculateQueueWeights(
+  async #calculateQueueScores(
     queues: Array<{ value: string; score: number }>,
-    calculateWeight: (queue: string, score: number) => Promise<number>
+    calculateCapacities: (queue: string) => Promise<QueueCapacities>
   ) {
-    const queueWeights = await Promise.all(
+    const now = Date.now();
+
+    const queueScores = await Promise.all(
       queues.map(async (queue) => {
         return {
           queue: queue.value,
-          weight: await calculateWeight(queue.value, queue.score),
+          capacities: await calculateCapacities(queue.value),
+          age: now - queue.score,
         };
       })
     );
 
-    return queueWeights;
+    return queueScores;
   }
 
-  async #calculateMessageQueueWeight(queue: string, score: number) {
-    const concurrencyQueueName = queue.replace(/:ck:.+$/, "");
-
-    const concurrencyLimit =
-      (await this.redis.get(`${concurrencyQueueName}:${constants.CONCURRENCY_LIMIT_PART}`)) ?? 100;
-
-    const guardedConcurrencyLimit = Math.max(Number(concurrencyLimit), 1); // Ensure we don't divide by 0
-
-    const currentConcurrency = await this.redis.scard(
-      `${queue}:${constants.CURRENT_CONCURRENCY_PART}`
-    );
-
-    const guardedCurrentConcurrency = Math.max(Number(currentConcurrency), 0);
-
-    const capacity = Math.max(guardedConcurrencyLimit - guardedCurrentConcurrency, 0); // Ensure we don't have negative capacity
-
-    const capacityWeight = capacity / guardedConcurrencyLimit;
-    const ageWeight = Date.now() - score;
-
-    return ageWeight * 0.8 + capacityWeight * 0.2;
-  }
-
-  async #weightedRandomChoice(queues: Array<{ queue: string; weight: number }>) {
-    const totalWeight = queues.reduce((acc, queue) => acc + queue.weight, 0);
-    const randomNum = Math.random() * totalWeight;
-    let weightSum = 0;
-
-    for (const queue of queues) {
-      weightSum += queue.weight;
-      if (randomNum <= weightSum) {
-        return queue.queue;
-      }
-    }
-
-    return queues[queues.length - 1].queue;
+  async #calculateMessageQueueCapacities(queue: string) {
+    return await this.#callCalculateMessageCapacities({
+      currentConcurrencyKey: this.keys.currentConcurrencyKeyFromQueue(queue),
+      currentEnvConcurrencyKey: this.keys.envCurrentConcurrencyKeyFromQueue(queue),
+      currentOrgConcurrencyKey: this.keys.orgCurrentConcurrencyKeyFromQueue(queue),
+      concurrencyLimitKey: this.keys.concurrencyLimitKeyFromQueue(queue),
+      envConcurrencyLimitKey: this.keys.envConcurrencyLimitKeyFromQueue(queue),
+      orgConcurrencyLimitKey: this.keys.orgConcurrencyLimitKeyFromQueue(queue),
+    });
   }
 
   async #zrangeWithScores(
@@ -533,7 +537,7 @@ export class MarQS {
     for (let i = 0; i < messages.length; i++) {
       const message = messages[i];
 
-      const messageData = await this.redis.get(`${constants.MESSAGE_PART}:${message}`);
+      const messageData = await this.redis.get(this.keys.messageKey(message));
 
       if (!messageData) {
         // The message has been removed for some reason (TTL, etc.), so we should remove it from the timeout queue
@@ -551,10 +555,12 @@ export class MarQS {
       }
 
       await this.#callNackMessage({
-        messageKey: `${constants.MESSAGE_PART}:${message}`,
+        messageKey: this.keys.messageKey(message),
         messageQueue: parsedMessage.data.queue,
         parentQueue: parsedMessage.data.parentQueue,
-        concurrencyKey: `${parsedMessage.data.queue}:${constants.CURRENT_CONCURRENCY_PART}`,
+        concurrencyKey: this.keys.currentConcurrencyKeyFromQueue(parsedMessage.data.queue),
+        envConcurrencyKey: this.keys.envCurrentConcurrencyKeyFromQueue(parsedMessage.data.queue),
+        orgConcurrencyKey: this.keys.orgCurrentConcurrencyKeyFromQueue(parsedMessage.data.queue),
         visibilityQueue: constants.MESSAGE_VISIBILITY_TIMEOUT_QUEUE,
         messageId: parsedMessage.data.messageId,
         messageScore: parsedMessage.data.timestamp,
@@ -570,7 +576,7 @@ export class MarQS {
     return this.redis.enqueueMessage(
       message.queue,
       message.parentQueue,
-      `${constants.MESSAGE_PART}:${message.messageId}`,
+      this.keys.messageKey(message.messageId),
       message.queue,
       message.messageId,
       JSON.stringify(message),
@@ -583,24 +589,38 @@ export class MarQS {
     parentQueue,
     visibilityQueue,
     concurrencyLimitKey,
+    envConcurrencyLimitKey,
+    orgConcurrencyLimitKey,
     currentConcurrencyKey,
+    envCurrentConcurrencyKey,
+    orgCurrentConcurrencyKey,
   }: {
     messageQueue: string;
     parentQueue: string;
     visibilityQueue: string;
     concurrencyLimitKey: string;
+    envConcurrencyLimitKey: string;
+    orgConcurrencyLimitKey: string;
     currentConcurrencyKey: string;
+    envCurrentConcurrencyKey: string;
+    orgCurrentConcurrencyKey: string;
   }) {
     const result = await this.redis.dequeueMessage(
       messageQueue,
       parentQueue,
       visibilityQueue,
       concurrencyLimitKey,
+      envConcurrencyLimitKey,
+      orgConcurrencyLimitKey,
       currentConcurrencyKey,
+      envCurrentConcurrencyKey,
+      orgCurrentConcurrencyKey,
       messageQueue,
       String(this.options.visibilityTimeoutInMs ?? 300000), // 5 minutes
       String(Date.now()),
-      String(this.options.defaultConcurrency ?? 10)
+      String(this.options.defaultQueueConcurrency),
+      String(this.options.defaultEnvConcurrency),
+      String(this.options.defaultOrgConcurrency)
     );
 
     if (!result) {
@@ -625,21 +645,34 @@ export class MarQS {
     messageKey,
     visibilityQueue,
     concurrencyKey,
+    envConcurrencyKey,
+    orgConcurrencyKey,
     messageId,
   }: {
     messageKey: string;
     visibilityQueue: string;
     concurrencyKey: string;
+    envConcurrencyKey: string;
+    orgConcurrencyKey: string;
     messageId: string;
   }) {
     logger.debug("Calling acknowledgeMessage", {
       messageKey,
       visibilityQueue,
       concurrencyKey,
+      envConcurrencyKey,
+      orgConcurrencyKey,
       messageId,
     });
 
-    return this.redis.acknowledgeMessage(messageKey, visibilityQueue, concurrencyKey, messageId);
+    return this.redis.acknowledgeMessage(
+      messageKey,
+      visibilityQueue,
+      concurrencyKey,
+      envConcurrencyKey,
+      orgConcurrencyKey,
+      messageId
+    );
   }
 
   async #callNackMessage({
@@ -647,6 +680,8 @@ export class MarQS {
     messageQueue,
     parentQueue,
     concurrencyKey,
+    envConcurrencyKey,
+    orgConcurrencyKey,
     visibilityQueue,
     messageId,
     messageScore,
@@ -655,6 +690,8 @@ export class MarQS {
     messageQueue: string;
     parentQueue: string;
     concurrencyKey: string;
+    envConcurrencyKey: string;
+    orgConcurrencyKey: string;
     visibilityQueue: string;
     messageId: string;
     messageScore: number;
@@ -664,6 +701,8 @@ export class MarQS {
       messageQueue,
       parentQueue,
       concurrencyKey,
+      envConcurrencyKey,
+      orgConcurrencyKey,
       visibilityQueue,
       messageId,
       messageScore,
@@ -674,6 +713,8 @@ export class MarQS {
       messageQueue,
       parentQueue,
       concurrencyKey,
+      envConcurrencyKey,
+      orgConcurrencyKey,
       visibilityQueue,
       messageQueue,
       messageId,
@@ -701,6 +742,60 @@ export class MarQS {
     );
   }
 
+  async #callCalculateMessageCapacities({
+    currentConcurrencyKey,
+    currentEnvConcurrencyKey,
+    currentOrgConcurrencyKey,
+    concurrencyLimitKey,
+    envConcurrencyLimitKey,
+    orgConcurrencyLimitKey,
+  }: {
+    currentConcurrencyKey: string;
+    currentEnvConcurrencyKey: string;
+    currentOrgConcurrencyKey: string;
+    concurrencyLimitKey: string;
+    envConcurrencyLimitKey: string;
+    orgConcurrencyLimitKey: string;
+  }): Promise<QueueCapacities> {
+    const capacities = await this.redis.calculateMessageQueueCapacities(
+      currentConcurrencyKey,
+      currentEnvConcurrencyKey,
+      currentOrgConcurrencyKey,
+      concurrencyLimitKey,
+      envConcurrencyLimitKey,
+      orgConcurrencyLimitKey,
+      String(this.options.defaultQueueConcurrency),
+      String(this.options.defaultEnvConcurrency),
+      String(this.options.defaultOrgConcurrency)
+    );
+
+    // [queue current, queue limit, env current, env limit, org current, org limit]
+    return {
+      queue: { current: Number(capacities[0]), limit: Number(capacities[1]) },
+      env: { current: Number(capacities[2]), limit: Number(capacities[3]) },
+      org: { current: Number(capacities[4]), limit: Number(capacities[5]) },
+    };
+  }
+
+  #callUpdateGlobalConcurrencyLimits({
+    envConcurrencyLimitKey,
+    orgConcurrencyLimitKey,
+    envConcurrencyLimit,
+    orgConcurrencyLimit,
+  }: {
+    envConcurrencyLimitKey: string;
+    orgConcurrencyLimitKey: string;
+    envConcurrencyLimit: number;
+    orgConcurrencyLimit: number;
+  }) {
+    return this.redis.updateGlobalConcurrencyLimits(
+      envConcurrencyLimitKey,
+      orgConcurrencyLimitKey,
+      String(envConcurrencyLimit),
+      String(orgConcurrencyLimit)
+    );
+  }
+
   #registerCommands() {
     this.redis.defineCommand("enqueueMessage", {
       numberOfKeys: 3,
@@ -731,62 +826,91 @@ end
     });
 
     this.redis.defineCommand("dequeueMessage", {
-      numberOfKeys: 5,
+      numberOfKeys: 9,
       lua: `
--- Keys: childQueue, parentQueue, visibilityQueue, concurrencyLimitKey, currentConcurrencyKey
--- Args: visibilityTimeout, currentTime
+-- Keys: childQueue, parentQueue, visibilityQueue, concurrencyLimitKey, envConcurrencyLimitKey, orgConcurrencyLimitKey, currentConcurrencyKey, envCurrentConcurrencyKey, orgCurrentConcurrencyKey
 local childQueue = KEYS[1]
 local parentQueue = KEYS[2]
 local visibilityQueue = KEYS[3]
 local concurrencyLimitKey = KEYS[4]
-local currentConcurrencyKey = KEYS[5]
+local envConcurrencyLimitKey = KEYS[5]
+local orgConcurrencyLimitKey = KEYS[6]
+local currentConcurrencyKey = KEYS[7]
+local envCurrentConcurrencyKey = KEYS[8]
+local orgCurrentConcurrencyKey = KEYS[9]
+
+-- Args: childQueueName, visibilityQueue, currentTime, defaultConcurrencyLimit, defaultEnvConcurrencyLimit, defaultOrgConcurrencyLimit
 local childQueueName = ARGV[1]
 local visibilityTimeout = tonumber(ARGV[2])
 local currentTime = tonumber(ARGV[3])
 local defaultConcurrencyLimit = ARGV[4]
+local defaultEnvConcurrencyLimit = ARGV[5]
+local defaultOrgConcurrencyLimit = ARGV[6]
 
--- Check current concurrency against the limit
+-- Check current org concurrency against the limit
+local orgCurrentConcurrency = tonumber(redis.call('SCARD', orgCurrentConcurrencyKey) or '0')
+local orgConcurrencyLimit = tonumber(redis.call('GET', orgConcurrencyLimitKey) or defaultOrgConcurrencyLimit)
+
+if orgCurrentConcurrency >= orgConcurrencyLimit then
+    return nil
+end
+
+-- Check current env concurrency against the limit
+local envCurrentConcurrency = tonumber(redis.call('SCARD', envCurrentConcurrencyKey) or '0')
+local envConcurrencyLimit = tonumber(redis.call('GET', envConcurrencyLimitKey) or defaultEnvConcurrencyLimit)
+
+if envCurrentConcurrency >= envConcurrencyLimit then
+    return nil
+end
+
+-- Check current queue concurrency against the limit
 local currentConcurrency = tonumber(redis.call('SCARD', currentConcurrencyKey) or '0')
 local concurrencyLimit = tonumber(redis.call('GET', concurrencyLimitKey) or defaultConcurrencyLimit)
 
-if currentConcurrency < concurrencyLimit then
-    -- Attempt to dequeue the next message
-    local messages = redis.call('ZRANGEBYSCORE', childQueue, '-inf', currentTime, 'WITHSCORES', 'LIMIT', 0, 1)
-    if #messages == 0 then
-        return nil
-    end
-    local messageId = messages[1]
-    local messageScore = tonumber(messages[2])
-    local timeoutScore = currentTime + visibilityTimeout
-
-    -- Move message to timeout queue and update concurrency
-    redis.call('ZREM', childQueue, messageId)
-    redis.call('ZADD', visibilityQueue, timeoutScore, messageId)
-    redis.call('SADD', currentConcurrencyKey, messageId)
-
-    -- Rebalance the parent queue
-    local earliestMessage = redis.call('ZRANGE', childQueue, 0, 0, 'WITHSCORES')
-    if #earliestMessage == 0 then
-        redis.call('ZREM', parentQueue, childQueueName)
-    else
-        redis.call('ZADD', parentQueue, earliestMessage[2], childQueueName)
-    end
-    
-    return {messageId, messageScore} -- Return message details
+if currentConcurrency >= concurrencyLimit then
+    return nil
 end
 
-return nil
+-- Attempt to dequeue the next message
+local messages = redis.call('ZRANGEBYSCORE', childQueue, '-inf', currentTime, 'WITHSCORES', 'LIMIT', 0, 1)
 
+if #messages == 0 then
+    return nil
+end
+
+local messageId = messages[1]
+local messageScore = tonumber(messages[2])
+local timeoutScore = currentTime + visibilityTimeout
+
+-- Move message to timeout queue and update concurrency
+redis.call('ZREM', childQueue, messageId)
+redis.call('ZADD', visibilityQueue, timeoutScore, messageId)
+redis.call('SADD', currentConcurrencyKey, messageId)
+redis.call('SADD', envCurrentConcurrencyKey, messageId)
+redis.call('SADD', orgCurrentConcurrencyKey, messageId)
+
+-- Rebalance the parent queue
+local earliestMessage = redis.call('ZRANGE', childQueue, 0, 0, 'WITHSCORES')
+if #earliestMessage == 0 then
+    redis.call('ZREM', parentQueue, childQueueName)
+else
+    redis.call('ZADD', parentQueue, earliestMessage[2], childQueueName)
+end
+
+return {messageId, messageScore} -- Return message details
       `,
     });
 
     this.redis.defineCommand("acknowledgeMessage", {
-      numberOfKeys: 3,
+      numberOfKeys: 5,
       lua: `
--- Keys: messageKey, visibilityQueue, concurrencyKey
+-- Keys: messageKey, visibilityQueue, concurrencyKey, envCurrentConcurrencyKey, orgCurrentConcurrencyKey
 local messageKey = KEYS[1]
 local visibilityQueue = KEYS[2]
 local concurrencyKey = KEYS[3]
+local envCurrentConcurrencyKey = KEYS[4]
+local orgCurrentConcurrencyKey = KEYS[5]
+local globalCurrentConcurrencyKey = KEYS[6]
 
 -- Args: messageId
 local messageId = ARGV[1]
@@ -797,20 +921,24 @@ redis.call('DEL', messageKey)
 -- Remove the message from the timeout queue
 redis.call('ZREM', visibilityQueue, messageId)
 
--- Update the concurrency key
+-- Update the concurrency keys
 redis.call('SREM', concurrencyKey, messageId)
+redis.call('SREM', envCurrentConcurrencyKey, messageId)
+redis.call('SREM', orgCurrentConcurrencyKey, messageId)
 `,
     });
 
     this.redis.defineCommand("nackMessage", {
-      numberOfKeys: 5,
+      numberOfKeys: 7,
       lua: `
--- Keys: childQueueKey, parentQueueKey, visibilityQueue, concurrencyKey, messageId
+-- Keys: childQueueKey, parentQueueKey, visibilityQueue, concurrencyKey, envConcurrencyKey, orgConcurrencyKey, messageId
 local messageKey = KEYS[1]
 local childQueueKey = KEYS[2]
 local parentQueueKey = KEYS[3]
 local concurrencyKey = KEYS[4]
-local visibilityQueue = KEYS[5]
+local envConcurrencyKey = KEYS[5]
+local orgConcurrencyKey = KEYS[6]
+local visibilityQueue = KEYS[7]
 
 -- Args: childQueueName, messageId, currentTime, messageScore
 local childQueueName = ARGV[1]
@@ -825,8 +953,10 @@ if messageVisibility == 0 then
     return
 end
 
--- Update the concurrency key
+-- Update the concurrency keys
 redis.call('SREM', concurrencyKey, messageId)
+redis.call('SREM', envConcurrencyKey, messageId)
+redis.call('SREM', orgConcurrencyKey, messageId)
 
 -- Remove the message from the timeout queue
 redis.call('ZREM', visibilityQueue, messageId)
@@ -869,6 +999,52 @@ local newVisibilityTimeout = math.min(currentVisibilityTimeout + milliseconds *
 redis.call('ZADD', visibilityQueue, newVisibilityTimeout, messageId)
       `,
     });
+
+    this.redis.defineCommand("calculateMessageQueueCapacities", {
+      numberOfKeys: 6,
+      lua: `
+-- Keys: currentConcurrencyKey, currentEnvConcurrencyKey, currentOrgConcurrencyKey, concurrencyLimitKey, envConcurrencyLimitKey, orgConcurrencyLimitKey
+local currentConcurrencyKey = KEYS[1]
+local currentEnvConcurrencyKey = KEYS[2]
+local currentOrgConcurrencyKey = KEYS[3]
+local concurrencyLimitKey = KEYS[4]
+local envConcurrencyLimitKey = KEYS[5]
+local orgConcurrencyLimitKey = KEYS[6]
+
+-- Args defaultConcurrencyLimit, defaultEnvConcurrencyLimit, defaultOrgConcurrencyLimit
+local defaultConcurrencyLimit = tonumber(ARGV[1])
+local defaultEnvConcurrencyLimit = tonumber(ARGV[2])
+local defaultOrgConcurrencyLimit = tonumber(ARGV[3])
+
+local currentOrgConcurrency = tonumber(redis.call('SCARD', currentOrgConcurrencyKey) or '0')
+local orgConcurrencyLimit = tonumber(redis.call('GET', orgConcurrencyLimitKey) or defaultOrgConcurrencyLimit)
+
+local currentEnvConcurrency = tonumber(redis.call('SCARD', currentEnvConcurrencyKey) or '0')
+local envConcurrencyLimit = tonumber(redis.call('GET', envConcurrencyLimitKey) or defaultEnvConcurrencyLimit)
+
+local currentConcurrency = tonumber(redis.call('SCARD', currentConcurrencyKey) or '0')
+local concurrencyLimit = tonumber(redis.call('GET', concurrencyLimitKey) or defaultConcurrencyLimit)
+
+-- Return current capacity and concurrency limits for the queue, env, org
+return { currentConcurrency, concurrencyLimit, currentEnvConcurrency, envConcurrencyLimit, currentOrgConcurrency, orgConcurrencyLimit } 
+      `,
+    });
+
+    this.redis.defineCommand("updateGlobalConcurrencyLimits", {
+      numberOfKeys: 2,
+      lua: `
+-- Keys: envConcurrencyLimitKey, orgConcurrencyLimitKey
+local envConcurrencyLimitKey = KEYS[1]
+local orgConcurrencyLimitKey = KEYS[2]
+
+-- Args: envConcurrencyLimit, orgConcurrencyLimit
+local envConcurrencyLimit = ARGV[1]
+local orgConcurrencyLimit = ARGV[2]
+
+redis.call('SET', envConcurrencyLimitKey, envConcurrencyLimit)
+redis.call('SET', orgConcurrencyLimitKey, orgConcurrencyLimit)
+      `,
+    });
   }
 }
 
@@ -890,11 +1066,17 @@ declare module "ioredis" {
       parentQueue: string,
       visibilityQueue: string,
       concurrencyLimitKey: string,
+      envConcurrencyLimitKey: string,
+      orgConcurrencyLimitKey: string,
       currentConcurrencyKey: string,
+      envCurrentConcurrencyKey: string,
+      orgCurrentConcurrencyKey: string,
       childQueueName: string,
       visibilityTimeout: string,
       currentTime: string,
       defaultConcurrencyLimit: string,
+      defaultEnvConcurrencyLimit: string,
+      defaultOrgConcurrencyLimit: string,
       callback?: Callback<[string, string]>
     ): Result<[string, string] | null, Context>;
 
@@ -902,6 +1084,8 @@ declare module "ioredis" {
       messageKey: string,
       visibilityQueue: string,
       concurrencyKey: string,
+      envConcurrencyKey: string,
+      orgConcurrencyKey: string,
       messageId: string,
       callback?: Callback<void>
     ): Result<void, Context>;
@@ -911,6 +1095,8 @@ declare module "ioredis" {
       childQueueKey: string,
       parentQueueKey: string,
       concurrencyKey: string,
+      envConcurrencyKey: string,
+      orgConcurrencyKey: string,
       visibilityQueue: string,
       childQueueName: string,
       messageId: string,
@@ -926,6 +1112,27 @@ declare module "ioredis" {
       maxVisibilityTimeout: string,
       callback?: Callback<void>
     ): Result<void, Context>;
+
+    calculateMessageQueueCapacities(
+      currentConcurrencyKey: string,
+      currentEnvConcurrencyKey: string,
+      currentOrgConcurrencyKey: string,
+      concurrencyLimitKey: string,
+      envConcurrencyLimitKey: string,
+      orgConcurrencyLimitKey: string,
+      defaultConcurrencyLimit: string,
+      defaultEnvConcurrencyLimit: string,
+      defaultOrgConcurrencyLimit: string,
+      callback?: Callback<number[]>
+    ): Result<number[], Context>;
+
+    updateGlobalConcurrencyLimits(
+      envConcurrencyLimitKey: string,
+      orgConcurrencyLimitKey: string,
+      envConcurrencyLimit: string,
+      orgConcurrencyLimit: string,
+      callback?: Callback<void>
+    ): Result<void, Context>;
   }
 }
 
@@ -934,19 +1141,26 @@ export const marqs = singleton("marqs", getMarQSClient);
 function getMarQSClient() {
   if (env.V3_ENABLED) {
     if (env.REDIS_HOST && env.REDIS_PORT) {
+      const redisOptions = {
+        keyPrefix: KEY_PREFIX,
+        port: env.REDIS_PORT,
+        host: env.REDIS_HOST,
+        username: env.REDIS_USERNAME,
+        password: env.REDIS_PASSWORD,
+        enableAutoPipelining: true,
+        ...(env.REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }),
+      };
+
       return new MarQS({
+        keysProducer: new MarQSShortKeyProducer(KEY_PREFIX),
+        queuePriorityStrategy: new SimpleWeightedChoiceStrategy({ queueSelectionCount: 12 }),
+        envQueuePriorityStrategy: new SimpleWeightedChoiceStrategy({ queueSelectionCount: 12 }),
         workers: 1,
-        redis: {
-          keyPrefix: KEY_PREFIX,
-          port: env.REDIS_PORT,
-          host: env.REDIS_HOST,
-          username: env.REDIS_USERNAME,
-          password: env.REDIS_PASSWORD,
-          enableAutoPipelining: true,
-          ...(env.REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }),
-        },
-        defaultConcurrency: env.DEFAULT_ORG_EXECUTION_CONCURRENCY_LIMIT,
-        visibilityTimeoutInMs: 120 * 1000, // 2 minutes
+        redis: redisOptions,
+        defaultQueueConcurrency: env.DEFAULT_QUEUE_EXECUTION_CONCURRENCY_LIMIT,
+        defaultEnvConcurrency: env.DEFAULT_ENV_EXECUTION_CONCURRENCY_LIMIT,
+        defaultOrgConcurrency: env.DEFAULT_ORG_EXECUTION_CONCURRENCY_LIMIT,
+        visibilityTimeoutInMs: 120 * 1000, // 2 minutes,
       });
     } else {
       console.warn(
@@ -955,3 +1169,8 @@ function getMarQSClient() {
     }
   }
 }
+
+// Only allow alphanumeric characters, underscores, hyphens, and slashes (and only the first 128 characters)
+export function sanitizeQueueName(queueName: string) {
+  return queueName.replace(/[^a-zA-Z0-9_\-\/]/g, "").substring(0, 128);
+}

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

@@ -0,0 +1,119 @@
+import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
+import { MarQSKeyProducer } from "./types";
+
+const constants = {
+  SHARED_QUEUE: "sharedQueue",
+  CURRENT_CONCURRENCY_PART: "currentConcurrency",
+  CONCURRENCY_LIMIT_PART: "concurrency",
+  ENV_PART: "env",
+  ORG_PART: "org",
+  QUEUE_PART: "queue",
+  CONCURRENCY_KEY_PART: "ck",
+  MESSAGE_PART: "message",
+} as const;
+
+export class MarQSShortKeyProducer implements MarQSKeyProducer {
+  constructor(private _prefix: string) {}
+
+  queueConcurrencyLimitKey(env: AuthenticatedEnvironment, queue: string) {
+    return [this.queueKey(env, queue), constants.CONCURRENCY_LIMIT_PART].join(":");
+  }
+
+  envConcurrencyLimitKey(env: AuthenticatedEnvironment) {
+    return [this.envKeySection(env.id), constants.CONCURRENCY_LIMIT_PART].join(":");
+  }
+
+  orgConcurrencyLimitKey(env: AuthenticatedEnvironment) {
+    return [this.orgKeySection(env.organizationId), constants.CONCURRENCY_LIMIT_PART].join(":");
+  }
+
+  queueKey(env: AuthenticatedEnvironment, queue: string, concurrencyKey?: string) {
+    return [
+      this.orgKeySection(env.organizationId),
+      this.envKeySection(env.id),
+      this.queueSection(queue),
+    ]
+      .concat(concurrencyKey ? this.concurrencyKeySection(concurrencyKey) : [])
+      .join(":");
+  }
+
+  envSharedQueueKey(env: AuthenticatedEnvironment) {
+    if (env.type === "DEVELOPMENT") {
+      return [
+        this.orgKeySection(env.organizationId),
+        this.envKeySection(env.id),
+        constants.SHARED_QUEUE,
+      ].join(":");
+    }
+
+    return constants.SHARED_QUEUE;
+  }
+
+  concurrencyLimitKeyFromQueue(queue: string) {
+    const concurrencyQueueName = queue.replace(/:ck:.+$/, "");
+
+    return `${concurrencyQueueName}:${constants.CONCURRENCY_LIMIT_PART}`;
+  }
+
+  currentConcurrencyKeyFromQueue(queue: string) {
+    return `${queue}:${constants.CURRENT_CONCURRENCY_PART}`;
+  }
+
+  orgConcurrencyLimitKeyFromQueue(queue: string) {
+    const orgId = this.normalizeQueue(queue).split(":")[1];
+
+    return `${constants.ORG_PART}:${orgId}:${constants.CONCURRENCY_LIMIT_PART}`;
+  }
+
+  orgCurrentConcurrencyKeyFromQueue(queue: string) {
+    const orgId = this.normalizeQueue(queue).split(":")[1];
+
+    return `${constants.ORG_PART}:${orgId}:${constants.CURRENT_CONCURRENCY_PART}`;
+  }
+
+  envConcurrencyLimitKeyFromQueue(queue: string) {
+    const envId = this.normalizeQueue(queue).split(":")[3];
+
+    return `${constants.ENV_PART}:${envId}:${constants.CONCURRENCY_LIMIT_PART}`;
+  }
+
+  envCurrentConcurrencyKeyFromQueue(queue: string) {
+    const envId = this.normalizeQueue(queue).split(":")[3];
+
+    return `${constants.ENV_PART}:${envId}:${constants.CURRENT_CONCURRENCY_PART}`;
+  }
+
+  messageKey(messageId: string) {
+    return `${constants.MESSAGE_PART}:${messageId}`;
+  }
+
+  private shortId(id: string) {
+    // Return the last 12 characters of the id
+    return id.slice(-12);
+  }
+
+  private envKeySection(envId: string) {
+    return `${constants.ENV_PART}:${this.shortId(envId)}`;
+  }
+
+  private orgKeySection(orgId: string) {
+    return `${constants.ORG_PART}:${this.shortId(orgId)}`;
+  }
+
+  private queueSection(queue: string) {
+    return `${constants.QUEUE_PART}:${queue}`;
+  }
+
+  private concurrencyKeySection(concurrencyKey: string) {
+    return `${constants.CONCURRENCY_KEY_PART}:${concurrencyKey}`;
+  }
+
+  // This removes the leading prefix from the queue name if it exists
+  private normalizeQueue(queue: string) {
+    if (queue.startsWith(this._prefix)) {
+      return queue.slice(this._prefix.length);
+    }
+
+    return queue;
+  }
+}

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

@@ -0,0 +1,150 @@
+import { RedisOptions } from "ioredis";
+import { MarQSQueuePriorityStrategy, PriorityStrategyChoice, QueueWithScores } from "./types";
+import { nanoid } from "nanoid";
+import seedrandom from "seedrandom";
+
+export type DynamicWeightedChoiceStrategyOptions = {
+  initialQueueSelectionSize: number;
+  redis: RedisOptions;
+};
+
+// This implementation of the priority strategy will "react" over time, giving more weight to queues that have been selected less frequently.
+// It will also change the next candidate selection range based on if previous choices only had queues that were at capacity.
+// Some other ideas:
+// - Implement a "cooldown" period for queues that have been selected recently
+// - Implement a "decay" for queues that have been selected recently
+//
+// The "memory" of this strategy is stored in Redis, to coordinate between multiple instances of the webapp (coming soon?)
+export class DynamicWeightedChoiceStrategy implements MarQSQueuePriorityStrategy {
+  constructor(private options: DynamicWeightedChoiceStrategyOptions) {}
+
+  chooseQueue(
+    queues: QueueWithScores[],
+    parentQueue: string,
+    selectionId: string
+  ): PriorityStrategyChoice {
+    throw new Error("Method not implemented.");
+  }
+
+  nextCandidateSelection(
+    parentQueue: string
+  ): Promise<{ range: [number, number]; selectionId: string }> {
+    throw new Error("Method not implemented.");
+  }
+}
+
+export type SimpleWeightedChoiceStrategyOptions = {
+  queueSelectionCount: number;
+  randomSeed?: string;
+};
+
+export class SimpleWeightedChoiceStrategy implements MarQSQueuePriorityStrategy {
+  private _nextRangesByParentQueue: Map<string, [number, number]> = new Map();
+  private _randomGenerator = seedrandom(this.options.randomSeed);
+
+  constructor(private options: SimpleWeightedChoiceStrategyOptions) {}
+
+  private nextRangeForParentQueue(parentQueue: string) {
+    return this._nextRangesByParentQueue.get(parentQueue) ?? [0, this.options.queueSelectionCount];
+  }
+
+  chooseQueue(
+    queues: QueueWithScores[],
+    parentQueue: string,
+    selectionId: string
+  ): PriorityStrategyChoice {
+    const filteredQueues = filterQueuesAtCapacity(queues);
+
+    if (filteredQueues.length === 0) {
+      if (queues.length === this.options.queueSelectionCount) {
+        const nextRangeForParentQueue = this.nextRangeForParentQueue(parentQueue);
+        const nextRange: [number, number] = nextRangeForParentQueue
+          ? [
+              nextRangeForParentQueue[1],
+              nextRangeForParentQueue[1] + this.options.queueSelectionCount,
+            ]
+          : [this.options.queueSelectionCount, this.options.queueSelectionCount * 2];
+        // If all queues are at capacity, and we were passed the max number of queues, then we will slide the window "to the right"
+        this._nextRangesByParentQueue.set(parentQueue, nextRange);
+      } else {
+        this._nextRangesByParentQueue.delete(parentQueue);
+      }
+
+      return { abort: true };
+    }
+
+    this._nextRangesByParentQueue.delete(parentQueue);
+
+    const queueWeights = this.#calculateQueueWeights(filteredQueues);
+
+    return weightedRandomChoice(queueWeights, this._randomGenerator());
+  }
+
+  async nextCandidateSelection(
+    parentQueue: string
+  ): Promise<{ range: [number, number]; selectionId: string }> {
+    return { range: this.nextRangeForParentQueue(parentQueue), selectionId: nanoid(24) };
+  }
+
+  // This function calculates the weight of each queue based on the age of the queue and the capacity of the queue, env, and org
+  // First, it normalizes the age, queue capacity, env capacity, and org capacity to a value between 0 and 1 based on the maximum value of each
+  // Then, it calculates the weight of each queue based on the following factors:
+  // - Age is 50% of the weight
+  // - Queue capacity is 30% of the weight
+  // - Env capacity is 10% of the weight
+  // - Org capacity is 10% of the weight
+  #calculateQueueWeights(queues: QueueWithScores[]) {
+    const maximumAge = Math.max(...queues.map((queue) => queue.age));
+    const maximumQueueCapacity = Math.max(
+      ...queues.map((queue) => queue.capacities.queue.limit - queue.capacities.queue.current)
+    );
+    const maximumEnvCapacity = Math.max(
+      ...queues.map((queue) => queue.capacities.env.limit - queue.capacities.env.current)
+    );
+    const maximumOrgCapacity = Math.max(
+      ...queues.map((queue) => queue.capacities.org.limit - queue.capacities.org.current)
+    );
+
+    return queues.map(({ capacities, age, queue }) => {
+      const ageWeight = 0.5 * (age / maximumAge);
+      const queueWeight =
+        0.3 * (1 - (capacities.queue.limit - capacities.queue.current) / maximumQueueCapacity);
+      const envWeight =
+        0.1 * (1 - (capacities.env.limit - capacities.env.current) / maximumEnvCapacity);
+      const orgWeight =
+        0.1 * (1 - (capacities.org.limit - capacities.org.current) / maximumOrgCapacity);
+
+      return {
+        queue,
+        weight: ageWeight + queueWeight + envWeight + orgWeight,
+      };
+    });
+  }
+}
+
+function filterQueuesAtCapacity(queues: QueueWithScores[]) {
+  return queues.filter(
+    (queue) =>
+      queue.capacities.queue.current < queue.capacities.queue.limit &&
+      queue.capacities.env.current < queue.capacities.env.limit &&
+      queue.capacities.org.current < queue.capacities.org.limit
+  );
+}
+
+function weightedRandomChoice(
+  queues: Array<{ queue: string; weight: number }>,
+  randomNumber: number
+) {
+  const totalWeight = queues.reduce((acc, queue) => acc + queue.weight, 0);
+  const randomNum = randomNumber * totalWeight;
+  let weightSum = 0;
+
+  for (const queue of queues) {
+    weightSum += queue.weight;
+    if (randomNum <= weightSum) {
+      return queue.queue;
+    }
+  }
+
+  return queues[queues.length - 1].queue;
+}

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

@@ -21,7 +21,7 @@ import { z } from "zod";
 import { prisma } from "~/db.server";
 import { logger } from "~/services/logger.server";
 import { generateFriendlyId } from "../friendlyIdentifiers";
-import { marqs } from "../marqs.server";
+import { marqs } from "~/v3/marqs/index.server";
 import { EnvironmentVariablesRepository } from "../environmentVariables/environmentVariablesRepository.server";
 import { CancelAttemptService } from "../services/cancelAttempt.server";
 import { socketIo } from "../handleSocketIo.server";
@@ -251,35 +251,12 @@ export class SharedQueueConsumer {
 
     logger.log("dequeueMessageInSharedQueue()", { queueMessage: message });
 
-    const envId = this.#envIdFromQueue(message.queue);
-
-    const environment = await prisma.runtimeEnvironment.findUnique({
-      include: {
-        organization: true,
-        project: true,
-      },
-      where: {
-        id: envId,
-      },
-    });
-
-    if (!environment) {
-      logger.error("Environment not found", {
-        queueMessage: message.data,
-        envId,
-      });
-
-      this.#ackAndDoMoreWork(message.messageId);
-      return;
-    }
-
     const messageBody = MessageBody.safeParse(message.data);
 
     if (!messageBody.success) {
       logger.error("Failed to parse message", {
         queueMessage: message.data,
         error: messageBody.error,
-        env: environment,
       });
 
       this.#ackAndDoMoreWork(message.messageId);
@@ -381,6 +358,7 @@ export class SharedQueueConsumer {
             lockedById: backgroundTask.id,
           },
           include: {
+            runtimeEnvironment: true,
             attempts: {
               take: 1,
               orderBy: { number: "desc" },
@@ -411,7 +389,7 @@ export class SharedQueueConsumer {
         const queue = await prisma.taskQueue.findUnique({
           where: {
             runtimeEnvironmentId_name: {
-              runtimeEnvironmentId: environment.id,
+              runtimeEnvironmentId: lockedTaskRun.runtimeEnvironmentId,
               name: lockedTaskRun.queue,
             },
           },
@@ -437,7 +415,7 @@ export class SharedQueueConsumer {
             backgroundWorkerTaskId: backgroundTask.id,
             status: "PENDING" as const,
             queueId: queue.id,
-            runtimeEnvironmentId: environment.id,
+            runtimeEnvironmentId: lockedTaskRun.runtimeEnvironmentId,
           },
           include: {
             backgroundWorkerTask: true,
@@ -493,10 +471,10 @@ export class SharedQueueConsumer {
                 machine: machine.data,
                 // identifiers
                 id: taskRunAttempt.id,
-                envId: environment.id,
-                envType: environment.type,
-                orgId: environment.organizationId,
-                projectId: environment.projectId,
+                envId: lockedTaskRun.runtimeEnvironment.id,
+                envType: lockedTaskRun.runtimeEnvironment.type,
+                orgId: lockedTaskRun.runtimeEnvironment.organizationId,
+                projectId: lockedTaskRun.runtimeEnvironment.projectId,
                 runId: taskRunAttempt.taskRunId,
               },
             });
@@ -625,7 +603,7 @@ export class SharedQueueConsumer {
         const queue = await prisma.taskQueue.findUnique({
           where: {
             runtimeEnvironmentId_name: {
-              runtimeEnvironmentId: environment.id,
+              runtimeEnvironmentId: resumableAttempt.runtimeEnvironmentId,
               name: resumableRun.queue,
             },
           },
@@ -754,10 +732,6 @@ export class SharedQueueConsumer {
     return;
   }
 
-  #envIdFromQueue(queueName: string) {
-    return queueName.split(":")[1];
-  }
-
   #doMoreWork(intervalInMs = this._options.interval) {
     setTimeout(() => this.#doWork(), intervalInMs);
   }

apps/webapp/app/v3/marqs/types.ts

@@ -0,0 +1,79 @@
+import { RedisOptions } from "ioredis";
+import { z } from "zod";
+import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
+
+export type QueueCapacity = {
+  current: number;
+  limit: number;
+};
+
+export type QueueCapacities = {
+  queue: QueueCapacity;
+  env: QueueCapacity;
+  org: QueueCapacity;
+};
+
+export type QueueWithScores = {
+  queue: string;
+  capacities: QueueCapacities;
+  age: number;
+};
+
+export interface MarQSKeyProducer {
+  queueConcurrencyLimitKey(env: AuthenticatedEnvironment, queue: string): string;
+  envConcurrencyLimitKey(env: AuthenticatedEnvironment): string;
+  orgConcurrencyLimitKey(env: AuthenticatedEnvironment): string;
+  queueKey(env: AuthenticatedEnvironment, queue: string, concurrencyKey?: string): string;
+  envSharedQueueKey(env: AuthenticatedEnvironment): string;
+  concurrencyLimitKeyFromQueue(queue: string): string;
+  currentConcurrencyKeyFromQueue(queue: string): string;
+  orgConcurrencyLimitKeyFromQueue(queue: string): string;
+  orgCurrentConcurrencyKeyFromQueue(queue: string): string;
+  envConcurrencyLimitKeyFromQueue(queue: string): string;
+  envCurrentConcurrencyKeyFromQueue(queue: string): string;
+  messageKey(messageId: string): string;
+}
+
+export type PriorityStrategyChoice = string | { abort: true };
+
+export interface MarQSQueuePriorityStrategy {
+  /**
+   * chooseQueue is called to select the next queue to process a message from
+   *
+   * @param queues
+   * @param parentQueue
+   * @param selectionId
+   *
+   * @returns The queue to process the message from, or an object with `abort: true` if no queue is available
+   */
+  chooseQueue(
+    queues: Array<QueueWithScores>,
+    parentQueue: string,
+    selectionId: string
+  ): PriorityStrategyChoice;
+
+  /**
+   * This function is called to get the next candidate selection for the queue
+   * The `range` is used to select the set of queues that will be considered for the next selection (passed to chooseQueue)
+   * The `selectionId` is used to identify the selection and should be passed to chooseQueue
+   *
+   * @param parentQueue The parent queue that holds the candidate queues
+   *
+   * @returns The scores and the selectionId for the next candidate selection
+   */
+  nextCandidateSelection(
+    parentQueue: string
+  ): Promise<{ range: [number, number]; selectionId: string }>;
+}
+
+export const MessagePayload = z.object({
+  version: z.literal("1"),
+  data: z.record(z.unknown()),
+  queue: z.string(),
+  messageId: z.string(),
+  timestamp: z.number(),
+  parentQueue: z.string(),
+  concurrencyKey: z.string().optional(),
+});
+
+export type MessagePayload = z.infer<typeof MessagePayload>;

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

@@ -1,6 +1,6 @@
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { eventRepository } from "../eventRepository.server";
-import { marqs } from "../marqs.server";
+import { marqs } from "~/v3/marqs/index.server";
 import { BaseService } from "./baseService.server";
 import { logger } from "~/services/logger.server";
 

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

@@ -1,6 +1,6 @@
 import { TaskRun, TaskRunAttemptStatus, TaskRunStatus } from "@trigger.dev/database";
 import { eventRepository } from "../eventRepository.server";
-import { marqs } from "../marqs.server";
+import { marqs } from "~/v3/marqs/index.server";
 import { devPubSub } from "../marqs/devPubSub.server";
 import { BaseService } from "./baseService.server";
 import { socketIo } from "../handleSocketIo.server";

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

@@ -12,7 +12,7 @@ import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
 import { safeJsonParse } from "~/utils/json";
 import { eventRepository } from "../eventRepository.server";
-import { marqs } from "../marqs.server";
+import { marqs } from "~/v3/marqs/index.server";
 import { BaseService } from "./baseService.server";
 import { CancelAttemptService } from "./cancelAttempt.server";
 import { ResumeTaskRunDependenciesService } from "./resumeTaskRunDependencies.server";

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

@@ -4,10 +4,11 @@ import { Prisma, PrismaClientOrTransaction } from "~/db.server";
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
 import { generateFriendlyId } from "../friendlyIdentifiers";
-import { marqs } from "../marqs.server";
+import { marqs, sanitizeQueueName } from "~/v3/marqs/index.server";
 import { calculateNextBuildVersion } from "../utils/calculateNextBuildVersion";
 import { BaseService } from "./baseService.server";
 import { projectPubSub } from "./projectPubSub.server";
+import { env } from "~/env.server";
 
 export class CreateBackgroundWorkerService extends BaseService {
   public async call(
@@ -68,14 +69,39 @@ export class CreateBackgroundWorkerService extends BaseService {
 
       await createBackgroundTasks(body.metadata.tasks, backgroundWorker, environment, this._prisma);
 
-      //send a notification that a new worker has been created
-      await projectPubSub.publish(`project:${project.id}:env:${environment.id}`, "WORKER_CREATED", {
-        environmentId: environment.id,
-        environmentType: environment.type,
-        createdAt: backgroundWorker.createdAt,
-        taskCount: body.metadata.tasks.length,
-        type: "local",
-      });
+      try {
+        //send a notification that a new worker has been created
+        await projectPubSub.publish(
+          `project:${project.id}:env:${environment.id}`,
+          "WORKER_CREATED",
+          {
+            environmentId: environment.id,
+            environmentType: environment.type,
+            createdAt: backgroundWorker.createdAt,
+            taskCount: body.metadata.tasks.length,
+            type: "local",
+          }
+        );
+
+        await marqs?.updateEnvConcurrencyLimits(environment);
+      } catch (err) {
+        logger.error(
+          "Error publishing WORKER_CREATED event or updating global concurrency limits",
+          {
+            error:
+              err instanceof Error
+                ? {
+                    name: err.name,
+                    message: err.message,
+                    stack: err.stack,
+                  }
+                : err,
+            project,
+            environment,
+            backgroundWorker,
+          }
+        );
+      }
 
       return backgroundWorker;
     });
@@ -85,7 +111,7 @@ export class CreateBackgroundWorkerService extends BaseService {
 export async function createBackgroundTasks(
   tasks: TaskResource[],
   worker: BackgroundWorker,
-  env: AuthenticatedEnvironment,
+  environment: AuthenticatedEnvironment,
   prisma: PrismaClientOrTransaction
 ) {
   for (const task of tasks) {
@@ -105,7 +131,24 @@ export async function createBackgroundTasks(
         },
       });
 
-      const queueName = task.queue?.name ?? `task/${task.id}`;
+      let queueName = sanitizeQueueName(task.queue?.name ?? `task/${task.id}`);
+
+      // Check that the queuename is not an empty string
+      if (!queueName) {
+        queueName = sanitizeQueueName(`task/${task.id}`);
+      }
+
+      const concurrencyLimit =
+        typeof task.queue?.concurrencyLimit === "number"
+          ? Math.max(
+              Math.min(
+                task.queue.concurrencyLimit,
+                environment.maximumConcurrencyLimit,
+                environment.organization.maximumConcurrencyLimit
+              ),
+              0
+            )
+          : null;
 
       const taskQueue = await prisma.taskQueue.upsert({
         where: {
@@ -115,13 +158,13 @@ export async function createBackgroundTasks(
           },
         },
         update: {
-          concurrencyLimit: task.queue?.concurrencyLimit,
+          concurrencyLimit,
           rateLimit: task.queue?.rateLimit,
         },
         create: {
           friendlyId: generateFriendlyId("queue"),
           name: queueName,
-          concurrencyLimit: task.queue?.concurrencyLimit,
+          concurrencyLimit,
           runtimeEnvironmentId: worker.runtimeEnvironmentId,
           projectId: worker.projectId,
           rateLimit: task.queue?.rateLimit,
@@ -130,7 +173,11 @@ export async function createBackgroundTasks(
       });
 
       if (taskQueue.concurrencyLimit) {
-        await marqs?.updateQueueConcurrency(env, taskQueue.name, taskQueue.concurrencyLimit);
+        await marqs?.updateQueueConcurrencyLimits(
+          environment,
+          taskQueue.name,
+          taskQueue.concurrencyLimit
+        );
       }
     } catch (error) {
       if (error instanceof Prisma.PrismaClientKnownRequestError) {

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

@@ -6,7 +6,7 @@ import type {
 } from "@trigger.dev/database";
 import { logger } from "~/services/logger.server";
 import { generateFriendlyId } from "../friendlyIdentifiers";
-import { marqs } from "../marqs.server";
+import { marqs } from "~/v3/marqs/index.server";
 import { CreateCheckpointRestoreEventService } from "./createCheckpointRestoreEvent.server";
 import { BaseService } from "./baseService.server";
 

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

@@ -6,6 +6,8 @@ import { BaseService } from "./baseService.server";
 import { createBackgroundTasks } from "./createBackgroundWorker.server";
 import { CURRENT_DEPLOYMENT_LABEL } from "~/consts";
 import { projectPubSub } from "./projectPubSub.server";
+import { marqs } from "~/v3/marqs/index.server";
+import { logger } from "~/services/logger.server";
 
 export class CreateDeployedBackgroundWorkerService extends BaseService {
   public async call(
@@ -76,18 +78,23 @@ export class CreateDeployedBackgroundWorkerService extends BaseService {
         },
       });
 
-      //send a notification that a new worker has been created
-      await projectPubSub.publish(
-        `project:${environment.projectId}:env:${environment.id}`,
-        "WORKER_CREATED",
-        {
-          environmentId: environment.id,
-          environmentType: environment.type,
-          createdAt: backgroundWorker.createdAt,
-          taskCount: body.metadata.tasks.length,
-          type: "deployed",
-        }
-      );
+      try {
+        //send a notification that a new worker has been created
+        await projectPubSub.publish(
+          `project:${environment.projectId}:env:${environment.id}`,
+          "WORKER_CREATED",
+          {
+            environmentId: environment.id,
+            environmentType: environment.type,
+            createdAt: backgroundWorker.createdAt,
+            taskCount: body.metadata.tasks.length,
+            type: "deployed",
+          }
+        );
+        await marqs?.updateEnvConcurrencyLimits(environment);
+      } catch (err) {
+        logger.error("Failed to publish WORKER_CREATED event", { err });
+      }
 
       return backgroundWorker;
     });

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

@@ -6,7 +6,7 @@ import {
 } from "@trigger.dev/core/v3";
 import { $transaction } from "~/db.server";
 import { logger } from "~/services/logger.server";
-import { marqs } from "../marqs.server";
+import { marqs } from "~/v3/marqs/index.server";
 import { socketIo } from "../handleSocketIo.server";
 import { sharedQueueTasks } from "../marqs/sharedQueueConsumer.server";
 import { BaseService } from "./baseService.server";

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

@@ -1,6 +1,6 @@
 import { PrismaClientOrTransaction } from "~/db.server";
 import { workerQueue } from "~/services/worker.server";
-import { marqs } from "../marqs.server";
+import { marqs } from "~/v3/marqs/index.server";
 import { BaseService } from "./baseService.server";
 import { logger } from "~/services/logger.server";
 

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

@@ -1,6 +1,6 @@
 import { PrismaClientOrTransaction } from "~/db.server";
 import { workerQueue } from "~/services/worker.server";
-import { marqs } from "../marqs.server";
+import { marqs } from "~/v3/marqs/index.server";
 import { BaseService } from "./baseService.server";
 import { logger } from "~/services/logger.server";
 

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

@@ -1,5 +1,4 @@
 import {
-  PRIMARY_VARIANT,
   SemanticInternalAttributes,
   TriggerTaskRequestBody,
   packetRequiresOffloading,
@@ -10,10 +9,9 @@ import { $transaction } from "~/db.server";
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { eventRepository } from "../eventRepository.server";
 import { generateFriendlyId } from "../friendlyIdentifiers";
-import { marqs } from "../marqs.server";
-import { BaseService } from "./baseService.server";
+import { marqs } from "~/v3/marqs/index.server";
 import { uploadToObjectStore } from "../r2.server";
-import { logger } from "~/services/logger.server";
+import { BaseService } from "./baseService.server";
 
 export type TriggerTaskServiceOptions = {
   idempotencyKey?: string;

apps/webapp/package.json

@@ -19,7 +19,8 @@
     "generate:sourcemaps": "remix build --sourcemap",
     "clean:sourcemaps": "run-s clean:sourcemaps:*",
     "clean:sourcemaps:public": "rimraf ./build/**/*.map",
-    "clean:sourcemaps:build": "rimraf ./public/build/**/*.map"
+    "clean:sourcemaps:build": "rimraf ./public/build/**/*.map",
+    "test": "vitest"
   },
   "eslintIgnore": [
     "/node_modules",
@@ -140,6 +141,7 @@
     "remix-auth-github": "^1.6.0",
     "remix-typedjson": "0.3.1",
     "remix-utils": "^7.1.0",
+    "seedrandom": "^3.0.5",
     "semver": "^7.5.0",
     "simple-oauth2": "^5.0.0",
     "simplur": "^3.0.1",
@@ -186,6 +188,7 @@
     "@types/react": "18.2.69",
     "@types/react-collapse": "^5.0.4",
     "@types/react-dom": "18.2.7",
+    "@types/seedrandom": "^3.0.8",
     "@types/semver": "^7.3.13",
     "@types/simple-oauth2": "^5.0.4",
     "@types/slug": "^5.0.3",
@@ -214,7 +217,8 @@
     "tailwindcss": "3.4.1",
     "ts-node": "^10.7.0",
     "tsconfig-paths": "^3.14.1",
-    "typescript": "^5.1.6"
+    "typescript": "^5.1.6",
+    "vitest": "^1.4.0"
   },
   "engines": {
     "node": ">=16.0.0"

apps/webapp/test/marqs.test.ts

@@ -0,0 +1,263 @@
+import { SimpleWeightedChoiceStrategy } from "../app/v3/marqs/priorityStrategy.server";
+
+describe("SimpleWeightedChoiceStrategy", () => {
+  it("should use a weighted random choice algorithm to choose a queue", async () => {
+    const stategy = new SimpleWeightedChoiceStrategy({
+      queueSelectionCount: 3,
+      randomSeed: "test",
+    });
+
+    const chosenQueue = stategy.chooseQueue(
+      [
+        {
+          queue: "queue1",
+          age: 4497,
+          capacities: {
+            queue: { current: 0, limit: 10 },
+            env: { current: 0, limit: 10 },
+            org: { current: 0, limit: 10 },
+          },
+        },
+        {
+          queue: "queue2",
+          age: 19670,
+          capacities: {
+            queue: { current: 0, limit: 10 },
+            env: { current: 0, limit: 10 },
+            org: { current: 0, limit: 10 },
+          },
+        },
+        {
+          queue: "queue3",
+          age: 12828,
+          capacities: {
+            queue: { current: 0, limit: 10 },
+            env: { current: 0, limit: 10 },
+            org: { current: 0, limit: 10 },
+          },
+        },
+      ],
+      "parentQueue",
+      "selectionId"
+    );
+
+    expect(chosenQueue).toEqual("queue3");
+  });
+
+  it("should filter out queues if any capacity is full", async () => {
+    const stategy = new SimpleWeightedChoiceStrategy({
+      queueSelectionCount: 3,
+      randomSeed: "test",
+    });
+
+    const chosenQueue = stategy.chooseQueue(
+      [
+        {
+          queue: "queue1",
+          age: 4497,
+          capacities: {
+            queue: { current: 10, limit: 10 },
+            env: { current: 0, limit: 10 },
+            org: { current: 0, limit: 10 },
+          },
+        },
+        {
+          queue: "queue2",
+          age: 19670,
+          capacities: {
+            queue: { current: 0, limit: 10 },
+            env: { current: 10, limit: 10 },
+            org: { current: 0, limit: 10 },
+          },
+        },
+        {
+          queue: "queue3",
+          age: 12828,
+          capacities: {
+            queue: { current: 0, limit: 10 },
+            env: { current: 0, limit: 10 },
+            org: { current: 10, limit: 10 },
+          },
+        },
+      ],
+      "parentQueue",
+      "selectionId"
+    );
+
+    expect(chosenQueue).toEqual({ abort: true });
+
+    const nextSelection = await stategy.nextCandidateSelection("parentQueue");
+
+    expect(nextSelection).toEqual({ range: [3, 6], selectionId: expect.any(String) });
+
+    // Now pass some queues that have some capacity
+    const chosenQueue2 = stategy.chooseQueue(
+      [
+        {
+          queue: "queue1",
+          age: 4497,
+          capacities: {
+            queue: { current: 0, limit: 10 },
+            env: { current: 0, limit: 10 },
+            org: { current: 0, limit: 10 },
+          },
+        },
+        {
+          queue: "queue2",
+          age: 19670,
+          capacities: {
+            queue: { current: 0, limit: 10 },
+            env: { current: 0, limit: 10 },
+            org: { current: 0, limit: 10 },
+          },
+        },
+        {
+          queue: "queue3",
+          age: 12828,
+          capacities: {
+            queue: { current: 0, limit: 10 },
+            env: { current: 0, limit: 10 },
+            org: { current: 0, limit: 10 },
+          },
+        },
+      ],
+      "parentQueue",
+      "selectionId"
+    );
+
+    expect(chosenQueue2).toEqual("queue3");
+
+    const nextSelection2 = await stategy.nextCandidateSelection("parentQueue");
+
+    expect(nextSelection2).toEqual({ range: [0, 3], selectionId: expect.any(String) });
+  });
+
+  it("should adjust the next filter range only if passed the maximum number of queues", async () => {
+    const stategy = new SimpleWeightedChoiceStrategy({
+      queueSelectionCount: 3,
+      randomSeed: "test",
+    });
+
+    const chosenQueue = stategy.chooseQueue(
+      [
+        {
+          queue: "queue1",
+          age: 4497,
+          capacities: {
+            queue: { current: 10, limit: 10 },
+            env: { current: 0, limit: 10 },
+            org: { current: 0, limit: 10 },
+          },
+        },
+        {
+          queue: "queue2",
+          age: 19670,
+          capacities: {
+            queue: { current: 0, limit: 10 },
+            env: { current: 10, limit: 10 },
+            org: { current: 0, limit: 10 },
+          },
+        },
+      ],
+      "parentQueue",
+      "selectionId"
+    );
+
+    expect(chosenQueue).toEqual({ abort: true });
+
+    const nextSelection = await stategy.nextCandidateSelection("parentQueue");
+
+    expect(nextSelection).toEqual({ range: [0, 3], selectionId: expect.any(String) });
+  });
+
+  it("should adjust the next candidate range ONLY for the matching parent queue", async () => {
+    const stategy = new SimpleWeightedChoiceStrategy({
+      queueSelectionCount: 3,
+      randomSeed: "test",
+    });
+
+    const chosenQueue = stategy.chooseQueue(
+      [
+        {
+          queue: "queue1",
+          age: 4497,
+          capacities: {
+            queue: { current: 10, limit: 10 },
+            env: { current: 0, limit: 10 },
+            org: { current: 0, limit: 10 },
+          },
+        },
+        {
+          queue: "queue2",
+          age: 19670,
+          capacities: {
+            queue: { current: 10, limit: 10 },
+            env: { current: 0, limit: 10 },
+            org: { current: 0, limit: 10 },
+          },
+        },
+        {
+          queue: "queue3",
+          age: 12828,
+          capacities: {
+            queue: { current: 10, limit: 10 },
+            env: { current: 0, limit: 10 },
+            org: { current: 0, limit: 10 },
+          },
+        },
+      ],
+      "parentQueue",
+      "selectionId"
+    );
+
+    expect(chosenQueue).toEqual({ abort: true });
+
+    const nextSelection = await stategy.nextCandidateSelection("parentQueue2");
+
+    expect(nextSelection).toEqual({ range: [0, 3], selectionId: expect.any(String) });
+
+    const nextSelection2 = await stategy.nextCandidateSelection("parentQueue");
+
+    expect(nextSelection2).toEqual({ range: [3, 6], selectionId: expect.any(String) });
+
+    const chosenQueue2 = stategy.chooseQueue(
+      [
+        {
+          queue: "queue1",
+          age: 4497,
+          capacities: {
+            queue: { current: 0, limit: 10 },
+            env: { current: 0, limit: 10 },
+            org: { current: 0, limit: 10 },
+          },
+        },
+        {
+          queue: "queue2",
+          age: 19670,
+          capacities: {
+            queue: { current: 0, limit: 10 },
+            env: { current: 0, limit: 10 },
+            org: { current: 0, limit: 10 },
+          },
+        },
+        {
+          queue: "queue3",
+          age: 12828,
+          capacities: {
+            queue: { current: 0, limit: 10 },
+            env: { current: 0, limit: 10 },
+            org: { current: 0, limit: 10 },
+          },
+        },
+      ],
+      "parentQueue2",
+      "selectionId"
+    );
+
+    expect(chosenQueue2).toEqual("queue3");
+
+    const nextSelection3 = await stategy.nextCandidateSelection("parentQueue2");
+
+    expect(nextSelection3).toEqual({ range: [0, 3], selectionId: expect.any(String) });
+  });
+});

apps/webapp/tsconfig.json

@@ -2,6 +2,7 @@
   "exclude": ["./cypress", "./cypress.config.ts"],
   "include": ["remix.env.d.ts", "**/*.ts", "**/*.tsx"],
   "compilerOptions": {
+    "types": ["vitest/globals"],
     "lib": ["DOM", "DOM.Iterable", "ES2019"],
     "isolatedModules": true,
     "esModuleInterop": true,

apps/webapp/vitest.config.ts

@@ -0,0 +1,8 @@
+import { defineConfig } from "vitest/config";
+
+export default defineConfig({
+  test: {
+    include: ["test/**/*.test.ts"],
+    globals: true,
+  },
+});

integrations/airtable/CHANGELOG.md

@@ -1,5 +1,13 @@
 # @trigger.dev/airtable
 
+## 3.0.0-beta.3
+
+### Patch Changes
+
+- Updated dependencies [b271742dc]
+  - @trigger.dev/sdk@3.0.0-beta.3
+  - @trigger.dev/integration-kit@3.0.0-beta.3
+
 ## 3.0.0-beta.2
 
 ### Patch Changes

integrations/airtable/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/airtable",
-  "version": "3.0.0-beta.2",
+  "version": "3.0.0-beta.3",
   "description": "Trigger.dev integration for airtable",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",
@@ -25,8 +25,8 @@
     "typecheck": "tsc --noEmit"
   },
   "dependencies": {
-    "@trigger.dev/integration-kit": "workspace:^3.0.0-beta.2",
-    "@trigger.dev/sdk": "workspace:^3.0.0-beta.2",
+    "@trigger.dev/integration-kit": "workspace:^3.0.0-beta.3",
+    "@trigger.dev/sdk": "workspace:^3.0.0-beta.3",
     "airtable": "^0.12.1",
     "zod": "3.22.3"
   },

integrations/github/CHANGELOG.md

@@ -1,5 +1,13 @@
 # @trigger.dev/github
 
+## 3.0.0-beta.3
+
+### Patch Changes
+
+- Updated dependencies [b271742dc]
+  - @trigger.dev/sdk@3.0.0-beta.3
+  - @trigger.dev/integration-kit@3.0.0-beta.3
+
 ## 3.0.0-beta.2
 
 ### Patch Changes

integrations/github/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/github",
-  "version": "3.0.0-beta.2",
+  "version": "3.0.0-beta.3",
   "description": "The official GitHub integration for Trigger.dev",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",
@@ -30,8 +30,8 @@
     "@octokit/request-error": "^5.0.1",
     "@octokit/webhooks": "^12.0.10",
     "octokit": "^3.1.2",
-    "@trigger.dev/integration-kit": "workspace:^3.0.0-beta.2",
-    "@trigger.dev/sdk": "workspace:^3.0.0-beta.2",
+    "@trigger.dev/integration-kit": "workspace:^3.0.0-beta.3",
+    "@trigger.dev/sdk": "workspace:^3.0.0-beta.3",
     "zod": "3.22.3"
   },
   "engines": {

integrations/linear/CHANGELOG.md

@@ -1,5 +1,13 @@
 # @trigger.dev/linear
 
+## 3.0.0-beta.3
+
+### Patch Changes
+
+- Updated dependencies [b271742dc]
+  - @trigger.dev/sdk@3.0.0-beta.3
+  - @trigger.dev/integration-kit@3.0.0-beta.3
+
 ## 3.0.0-beta.2
 
 ### Patch Changes

integrations/linear/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/linear",
-  "version": "3.0.0-beta.2",
+  "version": "3.0.0-beta.3",
   "description": "Trigger.dev integration for @linear/sdk",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",
@@ -26,8 +26,8 @@
   },
   "dependencies": {
     "@linear/sdk": "^8.0.0",
-    "@trigger.dev/integration-kit": "workspace:^3.0.0-beta.2",
-    "@trigger.dev/sdk": "workspace:^3.0.0-beta.2",
+    "@trigger.dev/integration-kit": "workspace:^3.0.0-beta.3",
+    "@trigger.dev/sdk": "workspace:^3.0.0-beta.3",
     "zod": "3.22.3"
   },
   "engines": {

integrations/openai/CHANGELOG.md

@@ -1,5 +1,13 @@
 # @trigger.dev/slack
 
+## 3.0.0-beta.3
+
+### Patch Changes
+
+- Updated dependencies [b271742dc]
+  - @trigger.dev/sdk@3.0.0-beta.3
+  - @trigger.dev/integration-kit@3.0.0-beta.3
+
 ## 3.0.0-beta.2
 
 ### Patch Changes

integrations/openai/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/openai",
-  "version": "3.0.0-beta.2",
+  "version": "3.0.0-beta.3",
   "description": "The official OpenAI integration for Trigger.dev",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",
@@ -42,8 +42,8 @@
   },
   "dependencies": {
     "openai": "^4.16.1",
-    "@trigger.dev/sdk": "workspace:^3.0.0-beta.2",
-    "@trigger.dev/integration-kit": "workspace:^3.0.0-beta.2"
+    "@trigger.dev/sdk": "workspace:^3.0.0-beta.3",
+    "@trigger.dev/integration-kit": "workspace:^3.0.0-beta.3"
   },
   "engines": {
     "node": ">=18.0.0"

integrations/plain/CHANGELOG.md

@@ -1,5 +1,13 @@
 # @trigger.dev/plain
 
+## 3.0.0-beta.3
+
+### Patch Changes
+
+- Updated dependencies [b271742dc]
+  - @trigger.dev/sdk@3.0.0-beta.3
+  - @trigger.dev/integration-kit@3.0.0-beta.3
+
 ## 3.0.0-beta.2
 
 ### Patch Changes

integrations/plain/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/plain",
-  "version": "3.0.0-beta.2",
+  "version": "3.0.0-beta.3",
   "description": "The official Plain.com integration for Trigger.dev",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",
@@ -24,8 +24,8 @@
     "build:tsup": "tsup"
   },
   "dependencies": {
-    "@trigger.dev/integration-kit": "workspace:^3.0.0-beta.2",
-    "@trigger.dev/sdk": "workspace:^3.0.0-beta.2",
+    "@trigger.dev/integration-kit": "workspace:^3.0.0-beta.3",
+    "@trigger.dev/sdk": "workspace:^3.0.0-beta.3",
     "@team-plain/typescript-sdk": "^2.7.0"
   },
   "engines": {

integrations/replicate/CHANGELOG.md

@@ -1,5 +1,13 @@
 # @trigger.dev/replicate
 
+## 3.0.0-beta.3
+
+### Patch Changes
+
+- Updated dependencies [b271742dc]
+  - @trigger.dev/sdk@3.0.0-beta.3
+  - @trigger.dev/integration-kit@3.0.0-beta.3
+
 ## 3.0.0-beta.2
 
 ### Patch Changes

integrations/replicate/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/replicate",
-  "version": "3.0.0-beta.2",
+  "version": "3.0.0-beta.3",
   "description": "Trigger.dev integration for replicate",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",
@@ -25,8 +25,8 @@
     "typecheck": "tsc --noEmit"
   },
   "dependencies": {
-    "@trigger.dev/integration-kit": "workspace:^3.0.0-beta.2",
-    "@trigger.dev/sdk": "workspace:^3.0.0-beta.2",
+    "@trigger.dev/integration-kit": "workspace:^3.0.0-beta.3",
+    "@trigger.dev/sdk": "workspace:^3.0.0-beta.3",
     "replicate": "^0.18.1",
     "zod": "3.22.3"
   },

integrations/resend/CHANGELOG.md

@@ -1,5 +1,13 @@
 # @trigger.dev/resend
 
+## 3.0.0-beta.3
+
+### Patch Changes
+
+- Updated dependencies [b271742dc]
+  - @trigger.dev/sdk@3.0.0-beta.3
+  - @trigger.dev/integration-kit@3.0.0-beta.3
+
 ## 3.0.0-beta.2
 
 ### Patch Changes

integrations/resend/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/resend",
-  "version": "3.0.0-beta.2",
+  "version": "3.0.0-beta.3",
   "description": "The official Resend.com integration for Trigger.dev",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",
@@ -24,8 +24,8 @@
     "build:tsup": "tsup"
   },
   "dependencies": {
-    "@trigger.dev/integration-kit": "workspace:^3.0.0-beta.2",
-    "@trigger.dev/sdk": "workspace:^3.0.0-beta.2",
+    "@trigger.dev/integration-kit": "workspace:^3.0.0-beta.3",
+    "@trigger.dev/sdk": "workspace:^3.0.0-beta.3",
     "resend": "^2.1.0"
   },
   "engines": {

integrations/sendgrid/CHANGELOG.md

@@ -1,5 +1,13 @@
 # @trigger.dev/sendgrid
 
+## 3.0.0-beta.3
+
+### Patch Changes
+
+- Updated dependencies [b271742dc]
+  - @trigger.dev/sdk@3.0.0-beta.3
+  - @trigger.dev/integration-kit@3.0.0-beta.3
+
 ## 3.0.0-beta.2
 
 ### Patch Changes

integrations/sendgrid/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/sendgrid",
-  "version": "3.0.0-beta.2",
+  "version": "3.0.0-beta.3",
   "description": "Trigger.dev integration for @sendgrid/mail",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",
@@ -26,8 +26,8 @@
   },
   "dependencies": {
     "@sendgrid/mail": "^7.7.0",
-    "@trigger.dev/sdk": "workspace:^3.0.0-beta.2",
-    "@trigger.dev/integration-kit": "workspace:^3.0.0-beta.2"
+    "@trigger.dev/sdk": "workspace:^3.0.0-beta.3",
+    "@trigger.dev/integration-kit": "workspace:^3.0.0-beta.3"
   },
   "engines": {
     "node": ">=16.8.0"

integrations/shopify/CHANGELOG.md

@@ -1,5 +1,13 @@
 # @trigger.dev/shopify
 
+## 3.0.0-beta.3
+
+### Patch Changes
+
+- Updated dependencies [b271742dc]
+  - @trigger.dev/sdk@3.0.0-beta.3
+  - @trigger.dev/integration-kit@3.0.0-beta.3
+
 ## 3.0.0-beta.2
 
 ### Patch Changes

integrations/shopify/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/shopify",
-  "version": "3.0.0-beta.2",
+  "version": "3.0.0-beta.3",
   "description": "Trigger.dev integration for @shopify/shopify-api",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",
@@ -26,8 +26,8 @@
   },
   "dependencies": {
     "@shopify/shopify-api": "^8.0.2",
-    "@trigger.dev/sdk": "workspace:^3.0.0-beta.2",
-    "@trigger.dev/integration-kit": "workspace:^3.0.0-beta.2",
+    "@trigger.dev/sdk": "workspace:^3.0.0-beta.3",
+    "@trigger.dev/integration-kit": "workspace:^3.0.0-beta.3",
     "zod": "3.22.3"
   },
   "engines": {

integrations/slack/CHANGELOG.md

@@ -1,5 +1,12 @@
 # @trigger.dev/slack
 
+## 3.0.0-beta.3
+
+### Patch Changes
+
+- Updated dependencies [b271742dc]
+  - @trigger.dev/sdk@3.0.0-beta.3
+
 ## 3.0.0-beta.2
 
 ### Patch Changes

integrations/slack/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/slack",
-  "version": "3.0.0-beta.2",
+  "version": "3.0.0-beta.3",
   "description": "The official Slack integration for Trigger.dev",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",
@@ -25,7 +25,7 @@
   },
   "dependencies": {
     "@slack/web-api": "^6.8.1",
-    "@trigger.dev/sdk": "workspace:^3.0.0-beta.2",
+    "@trigger.dev/sdk": "workspace:^3.0.0-beta.3",
     "zod": "3.22.3"
   },
   "engines": {

integrations/stripe/CHANGELOG.md

@@ -1,5 +1,13 @@
 # @trigger.dev/stripe
 
+## 3.0.0-beta.3
+
+### Patch Changes
+
+- Updated dependencies [b271742dc]
+  - @trigger.dev/sdk@3.0.0-beta.3
+  - @trigger.dev/integration-kit@3.0.0-beta.3
+
 ## 3.0.0-beta.2
 
 ### Patch Changes

integrations/stripe/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/stripe",
-  "version": "3.0.0-beta.2",
+  "version": "3.0.0-beta.3",
   "description": "Trigger.dev integration for stripe",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",
@@ -25,8 +25,8 @@
     "typecheck": "tsc --noEmit"
   },
   "dependencies": {
-    "@trigger.dev/integration-kit": "workspace:^3.0.0-beta.2",
-    "@trigger.dev/sdk": "workspace:^3.0.0-beta.2",
+    "@trigger.dev/integration-kit": "workspace:^3.0.0-beta.3",
+    "@trigger.dev/sdk": "workspace:^3.0.0-beta.3",
     "stripe": "^12.14.0",
     "zod": "3.22.3"
   },

integrations/supabase/CHANGELOG.md

@@ -1,5 +1,13 @@
 # @trigger.dev/supabase
 
+## 3.0.0-beta.3
+
+### Patch Changes
+
+- Updated dependencies [b271742dc]
+  - @trigger.dev/sdk@3.0.0-beta.3
+  - @trigger.dev/integration-kit@3.0.0-beta.3
+
 ## 3.0.0-beta.2
 
 ### Patch Changes

integrations/supabase/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/supabase",
-  "version": "3.0.0-beta.2",
+  "version": "3.0.0-beta.3",
   "description": "Trigger.dev integration for @supabase/supabase-js",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",
@@ -26,8 +26,8 @@
   },
   "dependencies": {
     "@supabase/supabase-js": "^2.26.0",
-    "@trigger.dev/integration-kit": "workspace:^3.0.0-beta.2",
-    "@trigger.dev/sdk": "workspace:^3.0.0-beta.2",
+    "@trigger.dev/integration-kit": "workspace:^3.0.0-beta.3",
+    "@trigger.dev/sdk": "workspace:^3.0.0-beta.3",
     "supabase-management-js": "^1.0.0",
     "zod": "3.22.3"
   },

integrations/typeform/CHANGELOG.md

@@ -1,5 +1,13 @@
 # @trigger.dev/typeform
 
+## 3.0.0-beta.3
+
+### Patch Changes
+
+- Updated dependencies [b271742dc]
+  - @trigger.dev/sdk@3.0.0-beta.3
+  - @trigger.dev/integration-kit@3.0.0-beta.3
+
 ## 3.0.0-beta.2
 
 ### Patch Changes

integrations/typeform/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/typeform",
-  "version": "3.0.0-beta.2",
+  "version": "3.0.0-beta.3",
   "description": "The official Typeform integration for Trigger.dev",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",
@@ -24,8 +24,8 @@
     "typecheck": "tsc --noEmit"
   },
   "dependencies": {
-    "@trigger.dev/integration-kit": "workspace:^3.0.0-beta.2",
-    "@trigger.dev/sdk": "workspace:^3.0.0-beta.2",
+    "@trigger.dev/integration-kit": "workspace:^3.0.0-beta.3",
+    "@trigger.dev/sdk": "workspace:^3.0.0-beta.3",
     "@typeform/api-client": "^1.8.0",
     "zod": "3.22.3"
   },

packages/astro/CHANGELOG.md

@@ -1,5 +1,12 @@
 # @trigger.dev/astro
 
+## 3.0.0-beta.3
+
+### Patch Changes
+
+- Updated dependencies [b271742dc]
+  - @trigger.dev/sdk@3.0.0-beta.3
+
 ## 3.0.0-beta.2
 
 ### Patch Changes

packages/astro/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@trigger.dev/astro",
   "description": "An Astro-native integration for Trigger.dev background jobs platform",
-  "version": "3.0.0-beta.2",
+  "version": "3.0.0-beta.3",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",
   "files": [
@@ -20,7 +20,7 @@
     "build:tsup": "tsup"
   },
   "peerDependencies": {
-    "@trigger.dev/sdk": "workspace:^3.0.0-beta.2"
+    "@trigger.dev/sdk": "workspace:^3.0.0-beta.3"
   },
   "devDependencies": {
     "astro": "^3.0.12",

packages/cli-v3/CHANGELOG.md

@@ -1,5 +1,19 @@
 # trigger.dev
 
+## 3.0.0-beta.4
+
+### Patch Changes
+
+- c702d6a9c: better handle task metadata parse errors, and display nicely formatted errors
+- b271742dc: Configurable log levels in the config file and via env var
+- 8c4df326c: Improve error messages during dev/deploy and handle deploy image build issues
+- b271742dc: Added a Node.js runtime check for the CLI
+- 8694e573f: Fix CLI logout and add list-profiles command
+- Updated dependencies [c702d6a9c]
+- Updated dependencies [b271742dc]
+- Updated dependencies [9af2570da]
+  - @trigger.dev/core@3.0.0-beta.3
+
 ## 3.0.0-beta.3
 
 ### Patch Changes

packages/cli-v3/package.json

@@ -1,6 +1,6 @@
 {
   "name": "trigger.dev",
-  "version": "3.0.0-beta.3",
+  "version": "3.0.0-beta.4",
   "description": "A Command-Line Interface for Trigger.dev (v3) projects",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",
@@ -89,7 +89,7 @@
     "@opentelemetry/sdk-trace-base": "^1.22.0",
     "@opentelemetry/sdk-trace-node": "^1.22.0",
     "@opentelemetry/semantic-conventions": "^1.22.0",
-    "@trigger.dev/core": "workspace:^3.0.0-beta.2",
+    "@trigger.dev/core": "workspace:^3.0.0-beta.3",
     "@types/degit": "^2.8.3",
     "chalk": "^5.2.0",
     "chokidar": "^3.5.3",

packages/cli-v3/src/cli/common.ts

@@ -5,6 +5,7 @@ import { getTracer, provider } from "../telemetry/tracing";
 import { fromZodError } from "zod-validation-error";
 import { logger } from "../utilities/logger";
 import { outro } from "@clack/prompts";
+import { chalkError } from "../utilities/cliOutput";
 
 export const CommonCommandOptions = z.object({
   apiUrl: z.string().optional(),
@@ -21,15 +22,15 @@ export function commonOptions(command: Command) {
     .option("-a, --api-url <value>", "Override the API URL", "https://api.trigger.dev")
     .option(
       "-l, --log-level <level>",
-      "The log level to use (debug, info, log, warn, error, none)",
+      "The CLI log level to use (debug, info, log, warn, error, none). This does not effect the log level of your trigger.dev tasks.",
       "log"
     )
     .option("--skip-telemetry", "Opt-out of sending telemetry");
 }
 
-export class SkipLoggingError extends Error { }
-export class SkipCommandError extends Error { }
-export class OutroCommandError extends SkipCommandError { }
+export class SkipLoggingError extends Error {}
+export class SkipCommandError extends Error {}
+export class OutroCommandError extends SkipCommandError {}
 
 export async function handleTelemetry(action: () => Promise<void>) {
   try {
@@ -84,7 +85,8 @@ export async function wrapCommandAction<T extends z.AnyZodObject, TResult>(
         // do nothing
       } else {
         recordSpanException(span, e);
-        logger.error(e instanceof Error ? e.message : String(e));
+
+        logger.log(`${chalkError("X Error:")} ${e instanceof Error ? e.message : String(e)}`);
       }
 
       span.end();

packages/cli-v3/src/cli/index.ts

@@ -7,6 +7,7 @@ import { configureLogoutCommand } from "../commands/logout.js";
 import { configureWhoamiCommand } from "../commands/whoami.js";
 import { COMMAND_NAME } from "../consts.js";
 import { getVersion } from "../utilities/getVersion.js";
+import { configureListProfilesCommand } from "../commands/list-profiles.js";
 
 export const program = new Command();
 
@@ -21,3 +22,4 @@ configureDevCommand(program);
 configureDeployCommand(program);
 configureWhoamiCommand(program);
 configureLogoutCommand(program);
+configureListProfilesCommand(program);

packages/cli-v3/src/commands/deploy.ts

@@ -3,6 +3,7 @@ import { depot } from "@depot/cli";
 import { context, trace } from "@opentelemetry/api";
 import {
   ResolvedConfig,
+  TaskMetadataFailedToParseData,
   detectDependencyVersion,
   flattenAttributes,
   recordSpanException,
@@ -43,9 +44,17 @@ import { logger } from "../utilities/logger.js";
 import { createTaskFileImports, gatherTaskFiles } from "../utilities/taskFiles";
 import { login } from "./login";
 
+import { Glob } from "glob";
 import type { SetOptional } from "type-fest";
 import { bundleDependenciesPlugin, workerSetupImportConfigPlugin } from "../utilities/build";
-import { Glob } from "glob";
+import { chalkError, chalkPurple, chalkWarning } from "../utilities/cliOutput";
+import {
+  logESMRequireError,
+  logTaskMetadataParseError,
+  parseBuildErrorStack,
+  parseNpmInstallError,
+} from "../utilities/deployErrors";
+import { safeJsonParse } from "../utilities/safeJsonParse";
 
 const DeployCommandOptions = CommonCommandOptions.extend({
   skipTypecheck: z.boolean().default(false),
@@ -272,28 +281,44 @@ async function _deployCommand(dir: string, options: DeployCommandOptions) {
       );
     }
 
-    return buildAndPushImage({
-      registryHost,
-      auth: authorization.auth.accessToken,
-      imageTag: deploymentResponse.data.imageTag,
-      buildId: deploymentResponse.data.externalBuildData.buildId,
-      buildToken: deploymentResponse.data.externalBuildData.buildToken,
-      buildProjectId: deploymentResponse.data.externalBuildData.projectId,
-      cwd: compilation.path,
-      projectId: resolvedConfig.config.project,
-      deploymentId: deploymentResponse.data.id,
-      deploymentVersion: deploymentResponse.data.version,
-      contentHash: deploymentResponse.data.contentHash,
-      projectRef: resolvedConfig.config.project,
-      loadImage: options.loadImage,
-      buildPlatform: options.buildPlatform,
-    });
+    return buildAndPushImage(
+      {
+        registryHost,
+        auth: authorization.auth.accessToken,
+        imageTag: deploymentResponse.data.imageTag,
+        buildId: deploymentResponse.data.externalBuildData.buildId,
+        buildToken: deploymentResponse.data.externalBuildData.buildToken,
+        buildProjectId: deploymentResponse.data.externalBuildData.projectId,
+        cwd: compilation.path,
+        projectId: resolvedConfig.config.project,
+        deploymentId: deploymentResponse.data.id,
+        deploymentVersion: deploymentResponse.data.version,
+        contentHash: deploymentResponse.data.contentHash,
+        projectRef: resolvedConfig.config.project,
+        loadImage: options.loadImage,
+        buildPlatform: options.buildPlatform,
+      },
+      deploymentSpinner
+    );
   };
 
   const image = await buildImage();
 
   if (!image.ok) {
-    deploymentSpinner.stop(`Failed to build project image: ${image.error}`);
+    deploymentSpinner.stop(`Failed to build project.`);
+
+    // If there are logs, let's write it out to a temporary file and include the path in the error message
+    if (image.logs.trim() !== "") {
+      const logPath = join(await createTempDir(), `build-${deploymentResponse.data.shortCode}.log`);
+
+      await writeFile(logPath, image.logs);
+
+      logger.log(
+        `${chalkError("X Error:")} ${image.error}. Full build logs have been saved to ${logPath})`
+      );
+    } else {
+      logger.log(`${chalkError("X Error:")} ${image.error}.`);
+    }
 
     throw new SkipLoggingError(`Failed to build project image: ${image.error}`);
   }
@@ -379,10 +404,37 @@ async function _deployCommand(dir: string, options: DeployCommandOptions) {
     }
     case "FAILED": {
       if (finishedDeployment.errorData) {
-        deploymentSpinner.stop(
-          `Deployment encountered an error: ${finishedDeployment.errorData.name}. ${deploymentLink}`
-        );
-        logger.error(finishedDeployment.errorData.stack);
+        if (finishedDeployment.errorData.name === "TaskMetadataParseError") {
+          const errorJson = safeJsonParse(finishedDeployment.errorData.stack);
+
+          if (errorJson) {
+            const parsedError = TaskMetadataFailedToParseData.safeParse(errorJson);
+
+            if (parsedError.success) {
+              deploymentSpinner.stop(`Deployment encountered an error. ${deploymentLink}`);
+
+              logTaskMetadataParseError(parsedError.data.zodIssues, parsedError.data.tasks);
+
+              throw new SkipLoggingError(
+                `Deployment encountered an error: ${finishedDeployment.errorData.name}`
+              );
+            }
+          }
+        }
+
+        const parsedError = finishedDeployment.errorData.stack
+          ? parseBuildErrorStack(finishedDeployment.errorData)
+          : finishedDeployment.errorData.message;
+
+        if (typeof parsedError === "string") {
+          deploymentSpinner.stop(`Deployment encountered an error. ${deploymentLink}`);
+
+          logger.log(`${chalkError("X Error:")} ${parsedError}`);
+        } else {
+          deploymentSpinner.stop(`Deployment encountered an error. ${deploymentLink}`);
+
+          logESMRequireError(parsedError, resolvedConfig);
+        }
 
         throw new SkipLoggingError(
           `Deployment encountered an error: ${finishedDeployment.errorData.name}`
@@ -551,15 +603,18 @@ type BuildAndPushImageResults =
   | {
       ok: true;
       image: string;
+      logs: string;
       digest?: string;
     }
   | {
       ok: false;
       error: string;
+      logs: string;
     };
 
 async function buildAndPushImage(
-  options: BuildAndPushImageOptions
+  options: BuildAndPushImageOptions,
+  updater: ReturnType<typeof spinner>
 ): Promise<BuildAndPushImageResults> {
   return tracer.startActiveSpan("buildAndPushImage", async (span) => {
     span.setAttributes({
@@ -626,7 +681,7 @@ async function buildAndPushImage(
     const errors: string[] = [];
 
     try {
-      await new Promise<void>((res, rej) => {
+      const processCode = await new Promise<number | null>((res, rej) => {
         // For some reason everything is output on stderr, not stdout
         childProcess.stderr?.on("data", (data: Buffer) => {
           const text = data.toString();
@@ -636,9 +691,19 @@ async function buildAndPushImage(
         });
 
         childProcess.on("error", (e) => rej(e));
-        childProcess.on("close", () => res());
+        childProcess.on("close", (code) => res(code));
       });
 
+      const logs = extractLogs(errors);
+
+      if (processCode !== 0) {
+        return {
+          ok: false as const,
+          error: `Error building image`,
+          logs,
+        };
+      }
+
       const digest = extractImageDigest(errors);
 
       span.setAttributes({
@@ -650,6 +715,7 @@ async function buildAndPushImage(
       return {
         ok: true as const,
         image: options.imageTag,
+        logs,
         digest,
       };
     } catch (e) {
@@ -659,6 +725,7 @@ async function buildAndPushImage(
       return {
         ok: false as const,
         error: e instanceof Error ? e.message : JSON.stringify(e),
+        logs: extractLogs(errors),
       };
     }
   });
@@ -751,6 +818,7 @@ async function buildAndPushSelfHostedImage(
       return {
         ok: false as const,
         error: e instanceof Error ? e.message : JSON.stringify(e),
+        logs: extractLogs(errors),
       };
     }
 
@@ -793,6 +861,7 @@ async function buildAndPushSelfHostedImage(
         return {
           ok: false as const,
           error: e instanceof Error ? e.message : JSON.stringify(e),
+          logs: extractLogs(errors),
         };
       }
     }
@@ -803,6 +872,7 @@ async function buildAndPushSelfHostedImage(
       ok: true as const,
       image: options.imageTag,
       digest,
+      logs: extractLogs(errors),
     };
   });
 }
@@ -820,6 +890,13 @@ function extractImageDigest(outputs: string[]) {
   }
 }
 
+function extractLogs(outputs: string[]) {
+  // Remove empty lines
+  const cleanedOutputs = outputs.map((line) => line.trim()).filter((line) => line !== "");
+
+  return cleanedOutputs.map((line) => line.trim()).join("\n");
+}
+
 async function compileProject(
   config: ResolvedConfig,
   options: DeployCommandOptions,
@@ -1057,7 +1134,7 @@ async function compileProject(
       );
 
       if (!resolvingDependenciesResult) {
-        throw new Error("Failed to resolve dependencies");
+        throw new SkipLoggingError("Failed to resolve dependencies");
       }
 
       // Write the Containerfile to /tmp/dir/Containerfile
@@ -1188,15 +1265,39 @@ async function resolveDependencies(
 
         return true;
       } catch (installError) {
-        logger.debug(`Failed to resolve dependencies: ${JSON.stringify(installError)}`);
-
         recordSpanException(span, installError);
-
         span.end();
 
-        resolvingDepsSpinner.stop(
-          "Failed to resolve dependencies. Rerun with --log-level=debug for more information"
-        );
+        const parsedError = parseNpmInstallError(installError);
+
+        if (typeof parsedError === "string") {
+          resolvingDepsSpinner.stop(`Failed to resolve dependencies: ${parsedError}`);
+        } else {
+          switch (parsedError.type) {
+            case "package-not-found-error": {
+              resolvingDepsSpinner.stop(`Failed to resolve dependencies`);
+
+              logger.log(
+                `\n${chalkError("X Error:")} The package ${chalkPurple(
+                  parsedError.packageName
+                )} could not be found in the npm registry.`
+              );
+
+              break;
+            }
+            case "no-matching-version-error": {
+              resolvingDepsSpinner.stop(`Failed to resolve dependencies`);
+
+              logger.log(
+                `\n${chalkError("X Error:")} The package ${chalkPurple(
+                  parsedError.packageName
+                )} could not resolve because the version doesn't exist`
+              );
+
+              break;
+            }
+          }
+        }
 
         return false;
       }
@@ -1312,8 +1413,12 @@ async function gatherRequiredDependencies(
           dependencies[packageParts.name] = externalDependencyVersion;
           continue;
         } else {
-          logger.warn(
-            `Could not find version for package ${packageName}, add a version specifier to the package name (e.g. ${packageParts.name}@latest) or add it to your project's package.json`
+          logger.log(
+            `${chalkWarning("X Warning:")} Could not find version for package ${chalkPurple(
+              packageName
+            )}, add a version specifier to the package name (e.g. ${
+              packageParts.name
+            }@latest) or add it to your project's package.json`
           );
         }
       }

packages/cli-v3/src/commands/dev.tsx

@@ -38,8 +38,15 @@ import {
 import { logger } from "../utilities/logger.js";
 import { isLoggedIn } from "../utilities/session.js";
 import { createTaskFileImports, gatherTaskFiles } from "../utilities/taskFiles";
-import { UncaughtExceptionError } from "../workers/common/errors";
+import { TaskMetadataParseError, UncaughtExceptionError } from "../workers/common/errors";
 import { BackgroundWorker, BackgroundWorkerCoordinator } from "../workers/dev/backgroundWorker.js";
+import { runtimeCheck } from "../utilities/runtimeCheck";
+import {
+  logESMRequireError,
+  logTaskMetadataParseError,
+  parseBuildErrorStack,
+  parseNpmInstallError,
+} from "../utilities/deployErrors";
 
 let apiClient: CliApiClient | undefined;
 
@@ -72,7 +79,18 @@ export function configureDevCommand(program: Command) {
   });
 }
 
+const MINIMUM_NODE_MAJOR = 18;
+const MINIMUM_NODE_MINOR = 16;
+
 export async function devCommand(dir: string, options: DevCommandOptions) {
+  try {
+    runtimeCheck(MINIMUM_NODE_MAJOR, MINIMUM_NODE_MINOR);
+  } catch (e) {
+    logger.log(`${chalkError("X Error:")} ${e}`);
+    process.exitCode = 1;
+    return;
+  }
+
   const authorization = await isLoggedIn(options.profile);
 
   if (!authorization.ok) {
@@ -531,21 +549,59 @@ function useDev({
                     backgroundWorker
                   );
                 } catch (e) {
-                  if (e instanceof UncaughtExceptionError) {
+                  if (e instanceof TaskMetadataParseError) {
+                    logTaskMetadataParseError(e.zodIssues, e.tasks);
+                    return;
+                  } else if (e instanceof UncaughtExceptionError) {
+                    const parsedBuildError = parseBuildErrorStack(e.originalError);
+
+                    if (typeof parsedBuildError !== "string") {
+                      logESMRequireError(
+                        parsedBuildError,
+                        configPath
+                          ? { status: "file", path: configPath, config }
+                          : { status: "in-memory", config }
+                      );
+                      return;
+                    } else {
+                    }
+
                     if (e.originalError.stack) {
-                      logger.error("Background worker failed to start", e.originalError.stack);
+                      logger.log(
+                        `${chalkError("X Error:")} Worker failed to start`,
+                        e.originalError.stack
+                      );
                     }
 
                     return;
                   }
 
-                  if (e instanceof Error) {
-                    logger.error(`Background worker failed to start`, e.stack);
+                  const parsedError = parseNpmInstallError(e);
 
-                    return;
+                  if (typeof parsedError === "string") {
+                    logger.log(`${chalkError("X Error:")} ${parsedError}`);
+                  } else {
+                    switch (parsedError.type) {
+                      case "package-not-found-error": {
+                        logger.log(
+                          `\n${chalkError("X Error:")} The package ${chalkPurple(
+                            parsedError.packageName
+                          )} could not be found in the npm registry.`
+                        );
+
+                        break;
+                      }
+                      case "no-matching-version-error": {
+                        logger.log(
+                          `\n${chalkError("X Error:")} The package ${chalkPurple(
+                            parsedError.packageName
+                          )} could not resolve because the version doesn't exist`
+                        );
+
+                        break;
+                      }
+                    }
                   }
-
-                  logger.error(`Background worker failed to start: ${e}`);
                 }
               });
             },

packages/cli-v3/src/commands/list-profiles.ts

@@ -0,0 +1,67 @@
+import { Command } from "commander";
+import {
+  deleteAuthConfigProfile,
+  readAuthConfigFile,
+  readAuthConfigProfile,
+  writeAuthConfigProfile,
+} from "../utilities/configFiles.js";
+import { logger } from "../utilities/logger.js";
+import {
+  CommonCommandOptions,
+  commonOptions,
+  handleTelemetry,
+  wrapCommandAction,
+} from "../cli/common.js";
+import { printInitialBanner } from "../utilities/initialBanner.js";
+import { z } from "zod";
+import { chalkGrey } from "../utilities/cliOutput.js";
+import { log, outro, text } from "@clack/prompts";
+
+const ListProfilesOptions = CommonCommandOptions;
+
+type ListProfilesOptions = z.infer<typeof ListProfilesOptions>;
+
+export function configureListProfilesCommand(program: Command) {
+  return program
+    .command("list-profiles")
+    .description("List all of your CLI profiles")
+    .option(
+      "-l, --log-level <level>",
+      "The CLI log level to use (debug, info, log, warn, error, none). This does not effect the log level of your trigger.dev tasks.",
+      "log"
+    )
+    .option("--skip-telemetry", "Opt-out of sending telemetry")
+    .action(async (options) => {
+      await handleTelemetry(async () => {
+        await printInitialBanner(true);
+        await listProfilesCommand(options);
+      });
+    });
+}
+
+export async function listProfilesCommand(options: unknown) {
+  return await wrapCommandAction("listProfiles", ListProfilesOptions, options, async (opts) => {
+    return await listProfiles(opts);
+  });
+}
+
+export async function listProfiles(options: ListProfilesOptions) {
+  const authConfig = readAuthConfigFile();
+
+  if (!authConfig) {
+    logger.info("No profiles found");
+    return;
+  }
+
+  const profiles = Object.keys(authConfig);
+
+  log.message("Profiles:");
+
+  for (const profile of profiles) {
+    const profileConfig = authConfig[profile];
+
+    log.info(`${profile}${profileConfig?.apiUrl ? ` - ${chalkGrey(profileConfig.apiUrl)}` : ""}`);
+  }
+
+  outro("Retrieve account info by running whoami --profile <profile>");
+}

packages/cli-v3/src/commands/logout.ts

@@ -1,24 +1,28 @@
 import { Command } from "commander";
-import { readAuthConfigProfile, writeAuthConfigProfile } from "../utilities/configFiles.js";
-import { logger } from "../utilities/logger.js";
-import { CommonCommandOptions, commonOptions, handleTelemetry, wrapCommandAction } from "../cli/common.js";
-import { printInitialBanner } from "../utilities/initialBanner.js";
 import { z } from "zod";
+import {
+  CommonCommandOptions,
+  commonOptions,
+  handleTelemetry,
+  wrapCommandAction,
+} from "../cli/common.js";
+import { deleteAuthConfigProfile, readAuthConfigProfile } from "../utilities/configFiles.js";
+import { printInitialBanner } from "../utilities/initialBanner.js";
+import { logger } from "../utilities/logger.js";
 
 const LogoutCommandOptions = CommonCommandOptions;
 
 type LogoutCommandOptions = z.infer<typeof LogoutCommandOptions>;
 
 export function configureLogoutCommand(program: Command) {
-  return commonOptions(program
-    .command("logout")
-    .description("Logout of Trigger.dev"))
-    .action(async (options) => {
+  return commonOptions(program.command("logout").description("Logout of Trigger.dev")).action(
+    async (options) => {
       await handleTelemetry(async () => {
         await printInitialBanner(false);
         await logoutCommand(options);
       });
-    });
+    }
+  );
 }
 
 export async function logoutCommand(options: unknown) {
@@ -31,11 +35,11 @@ export async function logout(options: LogoutCommandOptions) {
   const config = readAuthConfigProfile(options.profile);
 
   if (!config?.accessToken) {
-    logger.info(`You are already logged out [${options.profile ?? "default"}]`);
+    logger.info(`You are already logged out [${options.profile}]`);
     return;
   }
 
-  writeAuthConfigProfile({ ...config, accessToken: undefined, apiUrl: undefined }, options.profile);
+  deleteAuthConfigProfile(options.profile);
 
-  logger.info(`Logged out of Trigger.dev [${options.profile ?? "default"}]`);
+  logger.info(`Logged out of Trigger.dev [${options.profile}]`);
 }

packages/cli-v3/src/templates/trigger.config.ts.template

@@ -2,6 +2,7 @@ import type { TriggerConfig } from "@trigger.dev/sdk/v3";
 
 export const config: TriggerConfig = {
   project: "${projectRef}",
+  logLevel: "log",
   retries: {
     enabledInDev: false,
     default: {

packages/cli-v3/src/utilities/configFiles.ts

@@ -56,7 +56,15 @@ export function readAuthConfigProfile(profile: string = "default"): UserAuthConf
   }
 }
 
-function readAuthConfigFile(): UserAuthConfigFile | undefined {
+export function deleteAuthConfigProfile(profile: string = "default") {
+  const existingConfig = readAuthConfigFile() || {};
+
+  delete existingConfig[profile];
+
+  writeAuthConfigFile(existingConfig);
+}
+
+export function readAuthConfigFile(): UserAuthConfigFile | undefined {
   try {
     const authConfigFilePath = getAuthConfigFilePath();
 
@@ -71,7 +79,7 @@ function readAuthConfigFile(): UserAuthConfigFile | undefined {
   }
 }
 
-function writeAuthConfigFile(config: UserAuthConfigFile) {
+export function writeAuthConfigFile(config: UserAuthConfigFile) {
   const authConfigFilePath = getAuthConfigFilePath();
   mkdirSync(path.dirname(authConfigFilePath), {
     recursive: true,

packages/cli-v3/src/utilities/deployErrors.ts

@@ -0,0 +1,181 @@
+import chalk from "chalk";
+import { relative } from "node:path";
+import { chalkError, chalkPurple, chalkGrey, chalkGreen, chalkWarning } from "./cliOutput";
+import { logger } from "./logger";
+import { ReadConfigResult } from "./configFiles";
+import { TaskMetadataParseError } from "../workers/common/errors";
+import { z } from "zod";
+import { groupTaskMetadataIssuesByTask } from "@trigger.dev/core/v3";
+
+export type ESMRequireError = {
+  type: "esm-require-error";
+  moduleName: string;
+};
+
+export type BuildError = ESMRequireError | string;
+
+function errorIsErrorLike(error: unknown): error is Error {
+  return (
+    error instanceof Error || (typeof error === "object" && error !== null && "message" in error)
+  );
+}
+
+export function parseBuildErrorStack(error: unknown): BuildError {
+  if (typeof error === "string") {
+    return error;
+  }
+
+  if (errorIsErrorLike(error)) {
+    if (typeof error.stack === "string") {
+      const isErrRequireEsm = error.stack.includes("ERR_REQUIRE_ESM");
+
+      let moduleName = null;
+
+      if (isErrRequireEsm) {
+        // Regular expression to match the module path
+        const moduleRegex = /node_modules\/(@[^\/]+\/[^\/]+|[^\/]+)\/[^\/]+\s/;
+        const match = moduleRegex.exec(error.stack);
+        if (match) {
+          moduleName = match[1] as string; // Capture the module name
+
+          return {
+            type: "esm-require-error",
+            moduleName,
+          };
+        }
+      }
+    } else {
+      return error.message;
+    }
+  }
+
+  return "Unknown error";
+}
+
+export function logESMRequireError(parsedError: ESMRequireError, resolvedConfig: ReadConfigResult) {
+  logger.log(
+    `\n${chalkError("X Error:")} The ${chalkPurple(
+      parsedError.moduleName
+    )} module is being required even though it's ESM only, and builds only support CommonJS. There are two ${chalk.underline(
+      "possible"
+    )} ways to fix this:`
+  );
+  logger.log(
+    `\n${chalkGrey("○")} Dynamically import the module in your code: ${chalkGrey(
+      `const myModule = await import("${parsedError.moduleName}");`
+    )}`
+  );
+
+  if (resolvedConfig.status === "file") {
+    const relativePath = relative(resolvedConfig.config.projectDir, resolvedConfig.path).replace(
+      /\\/g,
+      "/"
+    );
+
+    logger.log(
+      `${chalkGrey("○")} ${chalk.underline("Or")} add ${chalkPurple(
+        parsedError.moduleName
+      )} to the ${chalkGreen("dependenciesToBundle")} array in your config file ${chalkGrey(
+        `(${relativePath})`
+      )}. This will bundle the module with your code.\n`
+    );
+  } else {
+    logger.log(
+      `${chalkGrey("○")} ${chalk.underline("Or")} add ${chalkPurple(
+        parsedError.moduleName
+      )} to the ${chalkGreen("dependenciesToBundle")} array in your config file ${chalkGrey(
+        "(you'll need to create one)"
+      )}. This will bundle the module with your code.\n`
+    );
+  }
+}
+
+export type PackageNotFoundError = {
+  type: "package-not-found-error";
+  packageName: string;
+};
+
+export type NoMatchingVersionError = {
+  type: "no-matching-version-error";
+  packageName: string;
+};
+
+export type NpmInstallError = PackageNotFoundError | NoMatchingVersionError | string;
+
+export function parseNpmInstallError(error: unknown): NpmInstallError {
+  if (typeof error === "string") {
+    return error;
+  }
+
+  if (error instanceof Error) {
+    if (typeof error.stack === "string") {
+      const isPackageNotFoundError =
+        error.stack.includes("ERR! 404 Not Found") &&
+        error.stack.includes("is not in this registry");
+      let packageName = null;
+
+      if (isPackageNotFoundError) {
+        // Regular expression to match the package name
+        const packageNameRegex = /'([^']+)' is not in this registry/;
+        const match = packageNameRegex.exec(error.stack);
+        if (match) {
+          packageName = match[1] as string; // Capture the package name
+        }
+      }
+
+      if (packageName) {
+        return {
+          type: "package-not-found-error",
+          packageName,
+        };
+      }
+
+      const noMatchingVersionRegex = /No matching version found for ([^\s]+)\s/;
+      const noMatchingVersionMatch = noMatchingVersionRegex.exec(error.stack);
+      if (noMatchingVersionMatch) {
+        return {
+          type: "no-matching-version-error",
+          packageName: (noMatchingVersionMatch[1] as string).replace(/.$/, ""),
+        };
+      }
+
+      return error.message;
+    } else {
+      return error.message;
+    }
+  }
+
+  return "Unknown error";
+}
+
+export function logTaskMetadataParseError(zodIssues: z.ZodIssue[], tasks: any) {
+  logger.log(
+    `\n${chalkError("X Error:")} Failed to start. The following ${
+      zodIssues.length === 1 ? "task issue was" : "task issues were"
+    } found:`
+  );
+
+  const groupedIssues = groupTaskMetadataIssuesByTask(tasks, zodIssues);
+
+  for (const key in groupedIssues) {
+    const taskWithIssues = groupedIssues[key];
+
+    if (!taskWithIssues) {
+      continue;
+    }
+
+    logger.log(
+      `\n  ${chalkWarning("❯")} ${taskWithIssues.exportName} ${chalkGrey("in")} ${
+        taskWithIssues.filePath
+      }`
+    );
+
+    for (const issue of taskWithIssues.issues) {
+      if (issue.path) {
+        logger.log(`    ${chalkError("x")} ${issue.path} ${chalkGrey(issue.message)}`);
+      } else {
+        logger.log(`    ${chalkError("x")} ${chalkGrey(issue.message)}`);
+      }
+    }
+  }
+}

packages/cli-v3/src/utilities/installPackages.ts

@@ -15,23 +15,14 @@ export async function installPackages(
 
   await setPackageJsonDeps(join(cwd, "package.json"), packages);
 
-  const childProcess = execa(
+  await execa(
     "npm",
     ["install", "--install-strategy", "nested", "--ignore-scripts", "--no-audit", "--no-fund"],
     {
       cwd,
-      stderr: "inherit",
+      stderr: "pipe",
     }
   );
-
-  await new Promise<void>((res, rej) => {
-    childProcess.on("error", (e) => rej(e));
-    childProcess.on("close", () => res());
-  });
-
-  await childProcess;
-
-  return;
 }
 
 async function getPackageVersion(path: string) {

packages/cli-v3/src/utilities/runtimeCheck.ts

@@ -0,0 +1,28 @@
+import { logger } from "./logger";
+
+/**
+ * This function is used by the dev CLI to make sure that the runtime is compatible
+ */
+export function runtimeCheck(minimumMajor: number, minimumMinor: number) {
+  // Check if the runtime is Node.js
+  if (typeof process === "undefined") {
+    throw "The dev CLI can only be run in a Node.js compatible environment";
+  }
+
+  // Check if the runtime version is compatible
+  const [major = 0, minor = 0] = process.versions.node.split(".").map(Number);
+
+  const isBun = typeof process.versions.bun === "string";
+
+  if (major < minimumMajor || (major === minimumMajor && minor < minimumMinor)) {
+    if (isBun) {
+      throw `The dev CLI requires at least Node.js ${minimumMajor}.${minimumMinor}. You are running Bun ${process.versions.bun}, which is compatible with Node.js ${process.versions.node}`;
+    } else {
+      throw `The dev CLI requires at least Node.js ${minimumMajor}.${minimumMinor}. You are running Node.js ${process.versions.node}`;
+    }
+  }
+
+  logger.debug(
+    `Node.js version: ${process.versions.node}${isBun ? ` (Bun ${process.versions.bun})` : ""}`
+  );
+}

packages/cli-v3/src/utilities/safeJsonParse.ts

@@ -0,0 +1,11 @@
+export function safeJsonParse(json?: string): unknown {
+  if (!json) {
+    return undefined;
+  }
+
+  try {
+    return JSON.parse(json);
+  } catch {
+    return undefined;
+  }
+}

packages/cli-v3/src/workers/common/errors.ts

@@ -1,3 +1,5 @@
+import { z } from "zod";
+
 export class UncaughtExceptionError extends Error {
   constructor(
     public readonly originalError: { name: string; message: string; stack?: string },
@@ -8,3 +10,14 @@ export class UncaughtExceptionError extends Error {
     this.name = "UncaughtExceptionError";
   }
 }
+
+export class TaskMetadataParseError extends Error {
+  constructor(
+    public readonly zodIssues: z.ZodIssue[],
+    public readonly tasks: any
+  ) {
+    super(`Failed to parse task metadata`);
+
+    this.name = "TaskMetadataParseError";
+  }
+}

packages/cli-v3/src/workers/dev/backgroundWorker.ts

@@ -37,7 +37,7 @@ import {
 import { safeDeleteFileSync } from "../../utilities/fileSystem.js";
 import { installPackages } from "../../utilities/installPackages.js";
 import { logger } from "../../utilities/logger.js";
-import { UncaughtExceptionError } from "../common/errors.js";
+import { TaskMetadataParseError, UncaughtExceptionError } from "../common/errors.js";
 
 export type CurrentWorkers = BackgroundWorkerCoordinator["currentWorkers"];
 export class BackgroundWorkerCoordinator {
@@ -352,6 +352,11 @@ export class BackgroundWorker {
           resolved = true;
           reject(new UncaughtExceptionError(message.payload.error, message.payload.origin));
           child.kill();
+        } else if (message.type === "TASKS_FAILED_TO_PARSE") {
+          clearTimeout(timeout);
+          resolved = true;
+          reject(new TaskMetadataParseError(message.payload.zodIssues, message.payload.tasks));
+          child.kill();
         }
       });
 
@@ -385,7 +390,7 @@ export class BackgroundWorker {
 
     if (!this._taskRunProcesses.has(payload.execution.run.id)) {
       const taskRunProcess = new TaskRunProcess(
-        payload.execution.run.id,
+        payload.execution,
         this.path,
         {
           ...this.params.env,
@@ -542,7 +547,7 @@ class TaskRunProcess {
   public onExit: Evt<number> = new Evt();
 
   constructor(
-    private runId: string,
+    private execution: TaskRunExecution,
     private path: string,
     private env: NodeJS.ProcessEnv,
     private metadata: BackgroundWorkerProperties,
@@ -565,22 +570,25 @@ class TaskRunProcess {
   }
 
   async initialize() {
-    logger.debug(`[${this.runId}] initializing task run process`, {
-      env: this.env,
+    const fullEnv = {
+      ...(this.execution.run.isTest ? { TRIGGER_LOG_LEVEL: "debug" } : {}),
+      ...this.env,
+      OTEL_RESOURCE_ATTRIBUTES: JSON.stringify({
+        [SemanticInternalAttributes.PROJECT_DIR]: this.worker.projectConfig.projectDir,
+      }),
+      OTEL_EXPORTER_OTLP_COMPRESSION: "none",
+      ...(this.worker.debugOtel ? { OTEL_LOG_LEVEL: "debug" } : {}),
+    };
+
+    logger.debug(`[${this.execution.run.id}] initializing task run process`, {
+      env: fullEnv,
       path: this.path,
     });
 
     this._child = fork(this.path, {
       stdio: [/*stdin*/ "ignore", /*stdout*/ "pipe", /*stderr*/ "pipe", "ipc"],
       cwd: dirname(this.path),
-      env: {
-        ...this.env,
-        OTEL_RESOURCE_ATTRIBUTES: JSON.stringify({
-          [SemanticInternalAttributes.PROJECT_DIR]: this.worker.projectConfig.projectDir,
-        }),
-        OTEL_EXPORTER_OTLP_COMPRESSION: "none",
-        ...(this.worker.debugOtel ? { OTEL_LOG_LEVEL: "debug" } : {}),
-      },
+      env: fullEnv,
       execArgv: this.worker.debuggerOn
         ? ["--inspect-brk", "--trace-uncaught", "--no-warnings=ExperimentalWarning"]
         : ["--trace-uncaught", "--no-warnings=ExperimentalWarning"],
@@ -597,7 +605,7 @@ class TaskRunProcess {
       return;
     }
 
-    logger.debug(`[${this.runId}] cleaning up task run process`, { kill });
+    logger.debug(`[${this.execution.run.id}] cleaning up task run process`, { kill });
 
     await this._sender.send("CLEANUP", {
       flush: true,
@@ -643,12 +651,15 @@ class TaskRunProcess {
       return;
     }
 
-    if (execution.run.id === this.runId) {
+    if (execution.run.id === this.execution.run.id) {
       // We don't need to notify the task run process if it's the same as the one we're running
       return;
     }
 
-    logger.debug(`[${this.runId}] task run completed notification`, { completion, execution });
+    logger.debug(`[${this.execution.run.id}] task run completed notification`, {
+      completion,
+      execution,
+    });
 
     this._sender.send("TASK_RUN_COMPLETED_NOTIFICATION", {
       completion,
@@ -700,7 +711,7 @@ class TaskRunProcess {
   }
 
   async #handleExit(code: number) {
-    logger.debug(`[${this.runId}] task run process exiting`, { code });
+    logger.debug(`[${this.execution.run.id}] task run process exiting`, { code });
 
     // Go through all the attempts currently pending and reject them
     for (const [id, status] of this._attemptStatuses.entries()) {

packages/cli-v3/src/workers/dev/worker-facade.ts

@@ -6,6 +6,10 @@ import {
   type HandleErrorFunction,
   DurableClock,
   clock,
+  logLevels,
+  LogLevel,
+  getEnvVar,
+  ZodSchemaParsedError,
 } from "@trigger.dev/core/v3";
 
 __WORKER_SETUP__;
@@ -53,10 +57,18 @@ const devRuntimeManager = new DevRuntimeManager();
 
 runtime.setGlobalRuntimeManager(devRuntimeManager);
 
+const triggerLogLevel = getEnvVar("TRIGGER_LOG_LEVEL");
+
+const configLogLevel = triggerLogLevel
+  ? triggerLogLevel
+  : importedConfig
+  ? importedConfig.logLevel
+  : __PROJECT_CONFIG__.logLevel;
+
 const otelTaskLogger = new OtelTaskLogger({
   logger: otelLogger,
   tracer: tracer,
-  level: "info",
+  level: logLevels.includes(configLogLevel as any) ? (configLogLevel as LogLevel) : "log",
 });
 
 logger.setGlobalTaskLogger(otelTaskLogger);
@@ -208,8 +220,14 @@ process.on("message", async (msg: any) => {
   await handler.handleMessage(msg);
 });
 
-sender.send("TASKS_READY", { tasks: getTaskMetadata() }).catch((err) => {
-  console.error("Failed to send TASKS_READY message", err);
+const TASK_METADATA = getTaskMetadata();
+
+sender.send("TASKS_READY", { tasks: TASK_METADATA }).catch((err) => {
+  if (err instanceof ZodSchemaParsedError) {
+    sender.send("TASKS_FAILED_TO_PARSE", { zodIssues: err.error.issues, tasks: TASK_METADATA });
+  } else {
+    console.error("Failed to send TASKS_READY message", err);
+  }
 });
 
 process.title = "trigger-dev-worker";

packages/cli-v3/src/workers/prod/backgroundWorker.ts

@@ -4,6 +4,7 @@ import {
   CreateBackgroundWorkerResponse,
   InferSocketMessageSchema,
   ProdChildToWorkerMessages,
+  ProdTaskRunExecution,
   ProdTaskRunExecutionPayload,
   ProdWorkerToChildMessages,
   SemanticInternalAttributes,
@@ -18,7 +19,7 @@ import {
 } from "@trigger.dev/core/v3";
 import { Evt } from "evt";
 import { ChildProcess, fork } from "node:child_process";
-import { UncaughtExceptionError } from "../common/errors";
+import { TaskMetadataParseError, UncaughtExceptionError } from "../common/errors";
 
 class UnexpectedExitError extends Error {
   constructor(public code: number) {
@@ -148,6 +149,14 @@ export class ProdBackgroundWorker {
               child.kill();
             }
           },
+          TASKS_FAILED_TO_PARSE: async (message) => {
+            if (!resolved) {
+              clearTimeout(timeout);
+              resolved = true;
+              reject(new TaskMetadataParseError(message.zodIssues, message.tasks));
+              child.kill();
+            }
+          },
         },
       });
 
@@ -200,6 +209,7 @@ export class ProdBackgroundWorker {
 
     if (!this._taskRunProcess) {
       const taskRunProcess = new TaskRunProcess(
+        payload.execution,
         this.path,
         {
           ...this.params.env,
@@ -374,6 +384,7 @@ class TaskRunProcess {
   public onCancelCheckpoint = Evt.create<{ version?: "v1" }>();
 
   constructor(
+    private execution: ProdTaskRunExecution,
     private path: string,
     private env: NodeJS.ProcessEnv,
     private metadata: BackgroundWorkerProperties,
@@ -384,6 +395,7 @@ class TaskRunProcess {
     this._child = fork(this.path, {
       stdio: [/*stdin*/ "ignore", /*stdout*/ "pipe", /*stderr*/ "pipe", "ipc"],
       env: {
+        ...(this.execution.run.isTest ? { TRIGGER_LOG_LEVEL: "debug" } : {}),
         ...this.env,
         OTEL_RESOURCE_ATTRIBUTES: JSON.stringify({
           [SemanticInternalAttributes.PROJECT_DIR]: this.worker.projectConfig.projectDir,

packages/cli-v3/src/workers/prod/entry-point.ts

@@ -13,7 +13,7 @@ import { readFile } from "node:fs/promises";
 import { createServer } from "node:http";
 import { z } from "zod";
 import { ProdBackgroundWorker } from "./backgroundWorker";
-import { UncaughtExceptionError } from "../common/errors";
+import { TaskMetadataParseError, UncaughtExceptionError } from "../common/errors";
 import { setTimeout } from "node:timers/promises";
 
 declare const __PROJECT_CONFIG__: Config;
@@ -416,7 +416,19 @@ class ProdWorker {
               process.exit(1);
             }
           } catch (e) {
-            if (e instanceof UncaughtExceptionError) {
+            if (e instanceof TaskMetadataParseError) {
+              logger.error("tasks metadata parse error", { message: e.zodIssues, tasks: e.tasks });
+
+              socket.emit("INDEXING_FAILED", {
+                version: "v1",
+                deploymentId: this.deploymentId,
+                error: {
+                  name: "TaskMetadataParseError",
+                  message: "There was an error parsing the task metadata",
+                  stack: JSON.stringify({ zodIssues: e.zodIssues, tasks: e.tasks }),
+                },
+              });
+            } else if (e instanceof UncaughtExceptionError) {
               logger.error("uncaught exception", { message: e.originalError.message });
 
               socket.emit("INDEXING_FAILED", {

packages/cli-v3/src/workers/prod/worker-facade.ts

@@ -9,6 +9,10 @@ import {
   HandleErrorFunction,
   DurableClock,
   clock,
+  getEnvVar,
+  logLevels,
+  LogLevel,
+  ZodSchemaParsedError,
 } from "@trigger.dev/core/v3";
 import "source-map-support/register.js";
 
@@ -47,10 +51,18 @@ clock.setGlobalClock(durableClock);
 const tracer = new TriggerTracer({ tracer: otelTracer, logger: otelLogger });
 const consoleInterceptor = new ConsoleInterceptor(otelLogger);
 
+const triggerLogLevel = getEnvVar("TRIGGER_LOG_LEVEL");
+
+const configLogLevel = triggerLogLevel
+  ? triggerLogLevel
+  : importedConfig
+  ? importedConfig.logLevel
+  : __PROJECT_CONFIG__.logLevel;
+
 const otelTaskLogger = new OtelTaskLogger({
   logger: otelLogger,
   tracer: tracer,
-  level: "info",
+  level: logLevels.includes(configLogLevel as any) ? (configLogLevel as LogLevel) : "log",
 });
 
 logger.setGlobalTaskLogger(otelTaskLogger);
@@ -208,8 +220,14 @@ const prodRuntimeManager = new ProdRuntimeManager(zodIpc, {
 
 runtime.setGlobalRuntimeManager(prodRuntimeManager);
 
-zodIpc.send("TASKS_READY", { tasks: getTaskMetadata() }).catch((err) => {
-  console.error("Failed to send TASKS_READY message", err);
+const TASK_METADATA = getTaskMetadata();
+
+zodIpc.send("TASKS_READY", { tasks: TASK_METADATA }).catch((err) => {
+  if (err instanceof ZodSchemaParsedError) {
+    zodIpc.send("TASKS_FAILED_TO_PARSE", { zodIssues: err.error.issues, tasks: TASK_METADATA });
+  } else {
+    console.error("Failed to send TASKS_READY message", err);
+  }
 });
 
 process.title = "trigger-prod-worker";

packages/cli/CHANGELOG.md

@@ -1,5 +1,15 @@
 # create-trigger
 
+## 3.0.0-beta.3
+
+### Patch Changes
+
+- Updated dependencies [c702d6a9c]
+- Updated dependencies [b271742dc]
+- Updated dependencies [9af2570da]
+  - @trigger.dev/core@3.0.0-beta.3
+  - @trigger.dev/yalt@3.0.0-beta.3
+
 ## 3.0.0-beta.2
 
 ### Patch Changes

packages/cli/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/cli",
-  "version": "3.0.0-beta.2",
+  "version": "3.0.0-beta.3",
   "description": "The Trigger.dev CLI",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",

packages/core-apps/CHANGELOG.md

@@ -1,5 +1,7 @@
 # @trigger.dev/core-apps
 
+## 3.0.0-beta.3
+
 ## 3.0.0-beta.2
 
 ## 3.0.0-beta.1

packages/core-apps/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@trigger.dev/core-apps",
   "description": "Backend core code used across apps",
-  "version": "3.0.0-beta.2",
+  "version": "3.0.0-beta.3",
   "private": true,
   "license": "MIT",
   "main": "./dist/index.js",

packages/core-backend/CHANGELOG.md

@@ -1,5 +1,7 @@
 # @trigger.dev/core-backend
 
+## 3.0.0-beta.3
+
 ## 3.0.0-beta.2
 
 ## 3.0.0-beta.1

packages/core-backend/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/core-backend",
-  "version": "3.0.0-beta.2",
+  "version": "3.0.0-beta.3",
   "description": "Core code used across `@trigger.dev/sdk` and Trigger.dev server",
   "license": "MIT",
   "main": "./dist/index.js",

packages/core/CHANGELOG.md

@@ -1,5 +1,13 @@
 # internal-platform
 
+## 3.0.0-beta.3
+
+### Patch Changes
+
+- c702d6a9c: better handle task metadata parse errors, and display nicely formatted errors
+- b271742dc: Configurable log levels in the config file and via env var
+- 9af2570da: Retry 429, 500, and connection error API requests to the trigger.dev server
+
 ## 3.0.0-beta.2
 
 ### Patch Changes

packages/core/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/core",
-  "version": "3.0.0-beta.2",
+  "version": "3.0.0-beta.3",
   "description": "Core code used across the Trigger.dev SDK and platform",
   "license": "MIT",
   "main": "./dist/index.js",

packages/core/src/v3/apiClient/index.ts

@@ -1,5 +1,5 @@
 import { context, propagation } from "@opentelemetry/api";
-import { zodfetch } from "../../zodfetch";
+import { ZodFetchOptions, zodfetch } from "../../zodfetch";
 import { taskContextManager } from "../tasks/taskContextManager";
 import { SafeAsyncLocalStorage } from "../utils/safeAsyncLocalStorage";
 import { getEnvVar } from "../utils/getEnv";
@@ -15,6 +15,16 @@ export type TriggerOptions = {
   spanParentAsLink?: boolean;
 };
 
+const zodFetchOptions: ZodFetchOptions = {
+  retry: {
+    maxAttempts: 5,
+    minTimeoutInMs: 1000,
+    maxTimeoutInMs: 30_000,
+    factor: 2,
+    randomize: false,
+  },
+};
+
 /**
  * Trigger.dev v3 API client
  */
@@ -29,19 +39,29 @@ export class ApiClient {
   }
 
   triggerTask(taskId: string, body: TriggerTaskRequestBody, options?: TriggerOptions) {
-    return zodfetch(TriggerTaskResponse, `${this.baseUrl}/api/v1/tasks/${taskId}/trigger`, {
-      method: "POST",
-      headers: this.#getHeaders(options?.spanParentAsLink ?? false),
-      body: JSON.stringify(body),
-    });
+    return zodfetch(
+      TriggerTaskResponse,
+      `${this.baseUrl}/api/v1/tasks/${taskId}/trigger`,
+      {
+        method: "POST",
+        headers: this.#getHeaders(options?.spanParentAsLink ?? false),
+        body: JSON.stringify(body),
+      },
+      zodFetchOptions
+    );
   }
 
   batchTriggerTask(taskId: string, body: BatchTriggerTaskRequestBody, options?: TriggerOptions) {
-    return zodfetch(BatchTriggerTaskResponse, `${this.baseUrl}/api/v1/tasks/${taskId}/batch`, {
-      method: "POST",
-      headers: this.#getHeaders(options?.spanParentAsLink ?? false),
-      body: JSON.stringify(body),
-    });
+    return zodfetch(
+      BatchTriggerTaskResponse,
+      `${this.baseUrl}/api/v1/tasks/${taskId}/batch`,
+      {
+        method: "POST",
+        headers: this.#getHeaders(options?.spanParentAsLink ?? false),
+        body: JSON.stringify(body),
+      },
+      zodFetchOptions
+    );
   }
 
   createUploadPayloadUrl(filename: string) {
@@ -51,7 +71,8 @@ export class ApiClient {
       {
         method: "PUT",
         headers: this.#getHeaders(false),
-      }
+      },
+      zodFetchOptions
     );
   }
 
@@ -62,7 +83,8 @@ export class ApiClient {
       {
         method: "GET",
         headers: this.#getHeaders(false),
-      }
+      },
+      zodFetchOptions
     );
   }
 

packages/core/src/v3/errors.ts

@@ -1,3 +1,4 @@
+import { z } from "zod";
 import { TaskRunError } from "./schemas/common";
 import nodePath from "node:path";
 
@@ -95,3 +96,59 @@ function correctStackTraceLine(line: string, projectDir?: string) {
 
   return line;
 }
+
+export function groupTaskMetadataIssuesByTask(tasks: any, issues: z.ZodIssue[]) {
+  return issues.reduce(
+    (acc, issue) => {
+      if (issue.path.length === 0) {
+        return acc;
+      }
+
+      const taskIndex = issue.path[1];
+
+      if (typeof taskIndex !== "number") {
+        return acc;
+      }
+
+      const task = tasks[taskIndex];
+
+      if (!task) {
+        return acc;
+      }
+
+      const restOfPath = issue.path.slice(2);
+
+      const taskId = task.id;
+      const taskName = task.exportName;
+      const filePath = task.filePath;
+
+      const key = taskIndex;
+
+      const existing = acc[key] ?? {
+        id: taskId,
+        exportName: taskName,
+        filePath,
+        issues: [] as Array<{ message: string; path?: string }>,
+      };
+
+      existing.issues.push({
+        message: issue.message,
+        path: restOfPath.length === 0 ? undefined : restOfPath.join("."),
+      });
+
+      return {
+        ...acc,
+        [key]: existing,
+      };
+    },
+    {} as Record<
+      number,
+      {
+        id: any;
+        exportName: string;
+        filePath: string;
+        issues: Array<{ message: string; path?: string }>;
+      }
+    >
+  );
+}

packages/core/src/v3/index.ts

@@ -39,8 +39,8 @@ export { ProdRuntimeManager } from "./runtime/prodRuntimeManager";
 export { PreciseWallClock as DurableClock } from "./clock/preciseWallClock";
 export { TriggerTracer } from "./tracer";
 
-export type { TaskLogger } from "./logger/taskLogger";
-export { OtelTaskLogger } from "./logger/taskLogger";
+export type { TaskLogger, LogLevel } from "./logger/taskLogger";
+export { OtelTaskLogger, logLevels } from "./logger/taskLogger";
 export { ConsoleInterceptor } from "./consoleInterceptor";
 export {
   flattenAttributes,

packages/core/src/v3/logger/taskLogger.ts

@@ -7,9 +7,9 @@ import { flattenAttributes } from "../utils/flattenAttributes";
 import { ClockTime } from "../clock/clock";
 import { clock } from "../clock-api";
 
-export type LogLevel = "log" | "error" | "warn" | "info" | "debug";
+export type LogLevel = "none" | "log" | "error" | "warn" | "info" | "debug";
 
-const logLevels: Array<LogLevel> = ["error", "warn", "log", "info", "debug"];
+export const logLevels: Array<LogLevel> = ["none", "error", "warn", "log", "info", "debug"];
 
 export type TaskLoggerConfig = {
   logger: Logger;
@@ -34,31 +34,31 @@ export class OtelTaskLogger implements TaskLogger {
   }
 
   debug(message: string, properties?: Record<string, unknown>) {
-    if (this._level < 4) return;
+    if (this._level < 5) return;
 
     this.#emitLog(message, this.#getTimestampInHrTime(), "debug", SeverityNumber.DEBUG, properties);
   }
 
   log(message: string, properties?: Record<string, unknown>) {
-    if (this._level < 2) return;
+    if (this._level < 3) return;
 
     this.#emitLog(message, this.#getTimestampInHrTime(), "log", SeverityNumber.INFO, properties);
   }
 
   info(message: string, properties?: Record<string, unknown>) {
-    if (this._level < 3) return;
+    if (this._level < 4) return;
 
     this.#emitLog(message, this.#getTimestampInHrTime(), "info", SeverityNumber.INFO, properties);
   }
 
   warn(message: string, properties?: Record<string, unknown>) {
-    if (this._level < 1) return;
+    if (this._level < 2) return;
 
     this.#emitLog(message, this.#getTimestampInHrTime(), "warn", SeverityNumber.WARN, properties);
   }
 
   error(message: string, properties?: Record<string, unknown>) {
-    if (this._level < 0) return;
+    if (this._level < 1) return;
 
     this.#emitLog(message, this.#getTimestampInHrTime(), "error", SeverityNumber.ERROR, properties);
   }

packages/core/src/v3/schemas/api.ts

@@ -153,6 +153,12 @@ export const InitializeDeploymentRequestBody = z.object({
 
 export type InitializeDeploymentRequestBody = z.infer<typeof InitializeDeploymentRequestBody>;
 
+export const DeploymentErrorData = z.object({
+  name: z.string(),
+  message: z.string(),
+  stack: z.string().optional(),
+});
+
 export const GetDeploymentResponseBody = z.object({
   id: z.string(),
   status: z.enum([
@@ -168,14 +174,7 @@ export const GetDeploymentResponseBody = z.object({
   shortCode: z.string(),
   version: z.string(),
   imageReference: z.string().optional(),
-  errorData: z
-    .object({
-      name: z.string(),
-      message: z.string(),
-      stack: z.string().optional(),
-    })
-    .optional()
-    .nullable(),
+  errorData: DeploymentErrorData.optional().nullable(),
   worker: z
     .object({
       id: z.string(),

packages/core/src/v3/schemas/messages.ts

@@ -1,7 +1,7 @@
 import { z } from "zod";
 import { TaskRunExecution, TaskRunExecutionResult } from "./common";
 
-export const EnvironmentType = z.enum(["PRODUCTION", "STAGING", "DEVELOPMENT", "PREVIEW"])
+export const EnvironmentType = z.enum(["PRODUCTION", "STAGING", "DEVELOPMENT", "PREVIEW"]);
 export type EnvironmentType = z.infer<typeof EnvironmentType>;
 
 export const MachineCpu = z
@@ -244,7 +244,7 @@ export const QueueOptions = z.object({
   /** An optional property that specifies the maximum number of concurrent run executions.
    *
    * If this property is omitted, the task can potentially use up the full concurrency of an environment. */
-  concurrencyLimit: z.number().int().min(1).max(1000).optional(),
+  concurrencyLimit: z.number().int().min(0).max(1000).optional(),
   /** @deprecated This feature is coming soon */
   rateLimit: RateLimitOptions.optional(),
 });
@@ -278,6 +278,14 @@ export const UncaughtExceptionMessage = z.object({
   origin: z.enum(["uncaughtException", "unhandledRejection"]),
 });
 
+export const TaskMetadataFailedToParseData = z.object({
+  version: z.literal("v1").default("v1"),
+  tasks: z.unknown(),
+  zodIssues: z.custom<z.ZodIssue[]>((v) => {
+    return Array.isArray(v) && v.every((issue) => typeof issue === "object" && "message" in issue);
+  }),
+});
+
 export const childToWorkerMessages = {
   TASK_RUN_COMPLETED: z.object({
     version: z.literal("v1").default("v1"),
@@ -288,6 +296,7 @@ export const childToWorkerMessages = {
     version: z.literal("v1").default("v1"),
     tasks: TaskMetadataWithFilePath.array(),
   }),
+  TASKS_FAILED_TO_PARSE: TaskMetadataFailedToParseData,
   TASK_HEARTBEAT: z.object({
     version: z.literal("v1").default("v1"),
     id: z.string(),
@@ -323,6 +332,9 @@ export const ProdChildToWorkerMessages = {
       tasks: TaskMetadataWithFilePath.array(),
     }),
   },
+  TASKS_FAILED_TO_PARSE: {
+    message: TaskMetadataFailedToParseData,
+  },
   TASK_HEARTBEAT: {
     message: z.object({
       version: z.literal("v1").default("v1"),

packages/core/src/v3/schemas/schemas.ts

@@ -42,6 +42,7 @@ export const Config = z.object({
   additionalPackages: z.string().array().optional(),
   additionalFiles: z.string().array().optional(),
   dependenciesToBundle: z.array(z.union([z.string(), RegexSchema])).optional(),
+  logLevel: z.string().optional(),
 });
 
 export type Config = z.infer<typeof Config>;

packages/core/src/v3/types/config.ts

@@ -1,3 +1,4 @@
+import { LogLevel } from "../logger/taskLogger";
 import { RetryOptions } from "../schemas";
 import type { InstrumentationOption } from "@opentelemetry/instrumentation";
 
@@ -31,4 +32,13 @@ export interface ProjectConfig {
    * The OpenTelemetry instrumentations to enable
    */
   instrumentations?: InstrumentationOption[];
+
+  /**
+   * Set the log level for the logger. Defaults to "log", so you will see "log", "warn", and "error" messages, but not "info", or "debug" messages.
+   *
+   * We automatically set the logLevel to "debug" during test runs
+   *
+   * @default "log"
+   */
+  logLevel?: LogLevel;
 }

packages/core/src/v3/zodIpc.ts

@@ -9,6 +9,7 @@ import {
   ZodSocketMessageCatalogSchema,
 } from "./zodSocket";
 import { z } from "zod";
+import { ZodSchemaParsedError } from "./zodMessageHandler";
 
 interface ZodIpcMessageSender<TEmitCatalog extends ZodSocketMessageCatalogSchema> {
   send<K extends GetSocketMessagesWithoutCallback<TEmitCatalog>>(
@@ -272,7 +273,7 @@ export class ZodIpcConnection<
     const parsedPayload = schema.safeParse(payload);
 
     if (!parsedPayload.success) {
-      throw new Error(`Failed to parse message payload: ${JSON.stringify(parsedPayload.error)}`);
+      throw new ZodSchemaParsedError(parsedPayload.error, payload);
     }
 
     await this.#sendPacket({

packages/core/src/v3/zodMessageHandler.ts

@@ -1,6 +1,15 @@
 import { z } from "zod";
 import { StructuredLogger } from "./zodNamespace";
 
+export class ZodSchemaParsedError extends Error {
+  constructor(
+    public error: z.ZodError,
+    public payload: unknown
+  ) {
+    super(error.message);
+  }
+}
+
 export type ZodMessageValueSchema<TDiscriminatedUnion extends z.ZodDiscriminatedUnion<any, any>> =
   | z.ZodFirstPartySchemaTypes
   | TDiscriminatedUnion;
@@ -160,7 +169,7 @@ export class ZodMessageSender<TMessageCatalog extends ZodMessageCatalogSchema> {
     const parsedPayload = schema.safeParse(payload);
 
     if (!parsedPayload.success) {
-      throw new Error(`Failed to parse message payload: ${JSON.stringify(parsedPayload.error)}`);
+      throw new ZodSchemaParsedError(parsedPayload.error, payload);
     }
 
     await this.#sender({ type, payload, version: "v1" });

packages/core/src/zodfetch.ts

@@ -1,17 +1,32 @@
 import { z } from "zod";
-import { context, propagation } from "@opentelemetry/api";
+import { RetryOptions, calculateNextRetryDelay, defaultRetryOptions } from "./v3";
 
-type ApiResult<TSuccessResult> =
+export type ApiResult<TSuccessResult> =
   | { ok: true; data: TSuccessResult }
   | {
       ok: false;
       error: string;
     };
 
+export type ZodFetchOptions = {
+  retry?: RetryOptions;
+};
+
 export async function zodfetch<TResponseBody extends any>(
   schema: z.Schema<TResponseBody>,
   url: string,
-  requestInit?: RequestInit
+  requestInit?: RequestInit,
+  options?: ZodFetchOptions
+): Promise<ApiResult<TResponseBody>> {
+  return await _doZodFetch(schema, url, requestInit, options);
+}
+
+async function _doZodFetch<TResponseBody extends any>(
+  schema: z.Schema<TResponseBody>,
+  url: string,
+  requestInit?: RequestInit,
+  options?: ZodFetchOptions,
+  attempt = 1
 ): Promise<ApiResult<TResponseBody>> {
   try {
     const response = await fetch(url, requestInit);
@@ -23,7 +38,7 @@ export async function zodfetch<TResponseBody extends any>(
       };
     }
 
-    if (response.status >= 400 && response.status < 500) {
+    if (response.status >= 400 && response.status < 500 && response.status !== 429) {
       const body = await response.json();
       if (!body.error) {
         return { ok: false, error: "Something went wrong" };
@@ -32,6 +47,31 @@ export async function zodfetch<TResponseBody extends any>(
       return { ok: false, error: body.error };
     }
 
+    // Retryable errors
+    if (response.status === 429 || response.status >= 500) {
+      if (!options?.retry) {
+        return {
+          ok: false,
+          error: `Failed to fetch ${url}, got status code ${response.status}`,
+        };
+      }
+
+      const retry = { ...defaultRetryOptions, ...options.retry };
+
+      if (attempt > retry.maxAttempts) {
+        return {
+          ok: false,
+          error: `Failed to fetch ${url}, got status code ${response.status}`,
+        };
+      }
+
+      const delay = calculateNextRetryDelay(retry, attempt);
+
+      await new Promise((resolve) => setTimeout(resolve, delay));
+
+      return await _doZodFetch(schema, url, requestInit, options, attempt + 1);
+    }
+
     if (response.status !== 200) {
       return {
         ok: false,
@@ -55,6 +95,23 @@ export async function zodfetch<TResponseBody extends any>(
 
     return { ok: false, error: parsedResult.error.message };
   } catch (error) {
+    if (options?.retry) {
+      const retry = { ...defaultRetryOptions, ...options.retry };
+
+      if (attempt > retry.maxAttempts) {
+        return {
+          ok: false,
+          error: error instanceof Error ? error.message : JSON.stringify(error),
+        };
+      }
+
+      const delay = calculateNextRetryDelay(retry, attempt);
+
+      await new Promise((resolve) => setTimeout(resolve, delay));
+
+      return await _doZodFetch(schema, url, requestInit, options, attempt + 1);
+    }
+
     return {
       ok: false,
       error: error instanceof Error ? error.message : JSON.stringify(error),