Release 3.3.14

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

.github/workflows/pr_checks.yml

@@ -28,34 +28,3 @@ jobs:
     with:
       package: cli-v3
     secrets: inherit
-
-  preview-release:
-    name: Preview Release
-    needs: [typecheck, units, e2e]
-    if: github.repository == 'triggerdotdev/trigger.dev'
-    runs-on: ubuntu-latest
-    steps:
-      - name: ⬇️ Checkout repo
-        uses: actions/checkout@v4
-        with:
-          fetch-depth: 0
-
-      - name: ⎔ Setup pnpm
-        uses: pnpm/action-setup@v4
-        with:
-          version: 8.15.5
-
-      - name: ⎔ Setup node
-        uses: buildjet/setup-node@v4
-        with:
-          node-version: 20.11.1
-          cache: "pnpm"
-
-      - name: 📥 Download deps
-        run: pnpm install --frozen-lockfile
-
-      - name: 🏗️ Build
-        run: pnpm run build --filter "@trigger.dev/*" --filter "trigger.dev"
-
-      - name: ⚡ Publish preview release
-        run: npx pkg-pr-new publish --no-template $(ls -d ./packages/*)

.vscode/launch.json

@@ -46,6 +46,14 @@
       "cwd": "${workspaceFolder}/references/init-shell",
       "sourceMaps": true
     },
+    {
+      "type": "node-terminal",
+      "request": "launch",
+      "name": "Debug V3 init dev CLI",
+      "command": "pnpm exec trigger dev",
+      "cwd": "${workspaceFolder}/references/init-shell",
+      "sourceMaps": true
+    },
     {
       "type": "node-terminal",
       "request": "launch",

apps/kubernetes-provider/src/index.ts

@@ -37,6 +37,8 @@ const UPTIME_MAX_PENDING_ERRORS = Number(process.env.UPTIME_MAX_PENDING_ERRORS |
 const POD_EPHEMERAL_STORAGE_SIZE_LIMIT = process.env.POD_EPHEMERAL_STORAGE_SIZE_LIMIT || "10Gi";
 const POD_EPHEMERAL_STORAGE_SIZE_REQUEST = process.env.POD_EPHEMERAL_STORAGE_SIZE_REQUEST || "2Gi";
 
+const PRE_PULL_DISABLED = process.env.PRE_PULL_DISABLED === "true";
+
 const logger = new SimpleLogger(`[${NODE_NAME}]`);
 logger.log(`running in ${RUNTIME_ENV} mode`);
 
@@ -301,6 +303,11 @@ class KubernetesTaskOperations implements TaskOperations {
   }
 
   async prePullDeployment(opts: TaskOperationsPrePullDeploymentOptions) {
+    if (PRE_PULL_DISABLED) {
+      logger.debug("Pre-pull is disabled, skipping.", { opts });
+      return;
+    }
+
     const metaName = this.#getPrePullContainerName(opts.shortCode);
 
     const metaLabels = {
@@ -332,6 +339,22 @@ class KubernetesTaskOperations implements TaskOperations {
             spec: {
               ...this.#defaultPodSpec,
               restartPolicy: "Always",
+              affinity: {
+                nodeAffinity: {
+                  requiredDuringSchedulingIgnoredDuringExecution: {
+                    nodeSelectorTerms: [
+                      {
+                        matchExpressions: [
+                          {
+                            key: "trigger.dev/pre-pull-disabled",
+                            operator: "DoesNotExist",
+                          },
+                        ],
+                      },
+                    ],
+                  },
+                },
+              },
               initContainers: [
                 {
                   name: "prepull",

apps/webapp/app/components/navigation/SideMenu.tsx

@@ -6,7 +6,6 @@ import {
   ChartBarIcon,
   ClockIcon,
   CreditCardIcon,
-  CursorArrowRaysIcon,
   IdentificationIcon,
   KeyIcon,
   RectangleStackIcon,
@@ -38,7 +37,6 @@ import {
   organizationTeamPath,
   personalAccessTokensPath,
   projectEnvironmentsPath,
-  projectEventsPath,
   projectHttpEndpointsPath,
   projectPath,
   projectRunsPath,
@@ -451,12 +449,6 @@ function V2ProjectSideMenu({
         to={projectTriggersPath(organization, project)}
         data-action="triggers"
       />
-      <SideMenuItem
-        name="Events"
-        icon={CursorArrowRaysIcon}
-        activeIconColor="text-sky-500"
-        to={projectEventsPath(organization, project)}
-      />
       <SideMenuItem
         name="HTTP endpoints"
         icon="http-endpoint"

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

@@ -444,7 +444,7 @@ function TasksDropdown({
         <SelectList>
           {filtered.map((item, index) => (
             <SelectItem
-              key={item.slug}
+              key={`${item.triggerSource}-${item.slug}`}
               value={item.slug}
               icon={
                 <TaskTriggerSourceIcon source={item.triggerSource} className="size-4 flex-none" />

apps/webapp/app/db.server.ts

@@ -111,6 +111,7 @@ function getClient() {
   const databaseUrl = extendQueryParams(DATABASE_URL, {
     connection_limit: env.DATABASE_CONNECTION_LIMIT.toString(),
     pool_timeout: env.DATABASE_POOL_TIMEOUT.toString(),
+    connection_timeout: env.DATABASE_CONNECTION_TIMEOUT.toString(),
   });
 
   console.log(`🔌 setting up prisma client to ${redactUrlSecrets(databaseUrl)}`);
@@ -162,6 +163,7 @@ function getReplicaClient() {
   const replicaUrl = extendQueryParams(env.DATABASE_READ_REPLICA_URL, {
     connection_limit: env.DATABASE_CONNECTION_LIMIT.toString(),
     pool_timeout: env.DATABASE_POOL_TIMEOUT.toString(),
+    connection_timeout: env.DATABASE_CONNECTION_TIMEOUT.toString(),
   });
 
   console.log(`🔌 setting up read replica connection to ${redactUrlSecrets(replicaUrl)}`);

apps/webapp/app/env.server.ts

@@ -13,6 +13,7 @@ const EnvironmentSchema = z.object({
     ),
   DATABASE_CONNECTION_LIMIT: z.coerce.number().int().default(10),
   DATABASE_POOL_TIMEOUT: z.coerce.number().int().default(60),
+  DATABASE_CONNECTION_TIMEOUT: z.coerce.number().int().default(20),
   DIRECT_URL: z
     .string()
     .refine(
@@ -307,6 +308,39 @@ const EnvironmentSchema = z.object({
   ALERT_SMTP_SECURE: z.coerce.boolean().optional(),
   ALERT_SMTP_USER: z.string().optional(),
   ALERT_SMTP_PASSWORD: z.string().optional(),
+  ALERT_RATE_LIMITER_EMISSION_INTERVAL: z.coerce.number().int().default(2_500),
+  ALERT_RATE_LIMITER_BURST_TOLERANCE: z.coerce.number().int().default(10_000),
+  ALERT_RATE_LIMITER_REDIS_HOST: z
+    .string()
+    .optional()
+    .transform((v) => v ?? process.env.REDIS_HOST),
+  ALERT_RATE_LIMITER_REDIS_READER_HOST: z
+    .string()
+    .optional()
+    .transform((v) => v ?? process.env.REDIS_READER_HOST),
+  ALERT_RATE_LIMITER_REDIS_READER_PORT: z.coerce
+    .number()
+    .optional()
+    .transform(
+      (v) =>
+        v ?? (process.env.REDIS_READER_PORT ? parseInt(process.env.REDIS_READER_PORT) : undefined)
+    ),
+  ALERT_RATE_LIMITER_REDIS_PORT: z.coerce
+    .number()
+    .optional()
+    .transform((v) => v ?? (process.env.REDIS_PORT ? parseInt(process.env.REDIS_PORT) : undefined)),
+  ALERT_RATE_LIMITER_REDIS_USERNAME: z
+    .string()
+    .optional()
+    .transform((v) => v ?? process.env.REDIS_USERNAME),
+  ALERT_RATE_LIMITER_REDIS_PASSWORD: z
+    .string()
+    .optional()
+    .transform((v) => v ?? process.env.REDIS_PASSWORD),
+  ALERT_RATE_LIMITER_REDIS_TLS_DISABLED: z
+    .string()
+    .default(process.env.REDIS_TLS_DISABLED ?? "false"),
+  ALERT_RATE_LIMITER_REDIS_CLUSTER_MODE_ENABLED: z.string().default("0"),
 
   MAX_SEQUENTIAL_INDEX_FAILURE_COUNT: z.coerce.number().default(96),
 
@@ -368,6 +402,82 @@ const EnvironmentSchema = z.object({
   BATCH_METADATA_OPERATIONS_FLUSH_INTERVAL_MS: z.coerce.number().int().default(1000),
   BATCH_METADATA_OPERATIONS_FLUSH_ENABLED: z.string().default("1"),
   BATCH_METADATA_OPERATIONS_FLUSH_LOGGING_ENABLED: z.string().default("1"),
+
+  LEGACY_RUN_ENGINE_WORKER_ENABLED: z.string().default(process.env.WORKER_ENABLED ?? "true"),
+  LEGACY_RUN_ENGINE_WORKER_CONCURRENCY_WORKERS: z.coerce.number().int().default(2),
+  LEGACY_RUN_ENGINE_WORKER_CONCURRENCY_TASKS_PER_WORKER: z.coerce.number().int().default(1),
+  LEGACY_RUN_ENGINE_WORKER_POLL_INTERVAL: z.coerce.number().int().default(1000),
+  LEGACY_RUN_ENGINE_WORKER_IMMEDIATE_POLL_INTERVAL: z.coerce.number().int().default(50),
+  LEGACY_RUN_ENGINE_WORKER_CONCURRENCY_LIMIT: z.coerce.number().int().default(100),
+
+  LEGACY_RUN_ENGINE_WORKER_REDIS_HOST: z
+    .string()
+    .optional()
+    .transform((v) => v ?? process.env.REDIS_HOST),
+  LEGACY_RUN_ENGINE_WORKER_REDIS_READER_HOST: z
+    .string()
+    .optional()
+    .transform((v) => v ?? process.env.REDIS_READER_HOST),
+  LEGACY_RUN_ENGINE_WORKER_REDIS_READER_PORT: z.coerce
+    .number()
+    .optional()
+    .transform(
+      (v) =>
+        v ?? (process.env.REDIS_READER_PORT ? parseInt(process.env.REDIS_READER_PORT) : undefined)
+    ),
+  LEGACY_RUN_ENGINE_WORKER_REDIS_PORT: z.coerce
+    .number()
+    .optional()
+    .transform((v) => v ?? (process.env.REDIS_PORT ? parseInt(process.env.REDIS_PORT) : undefined)),
+  LEGACY_RUN_ENGINE_WORKER_REDIS_USERNAME: z
+    .string()
+    .optional()
+    .transform((v) => v ?? process.env.REDIS_USERNAME),
+  LEGACY_RUN_ENGINE_WORKER_REDIS_PASSWORD: z
+    .string()
+    .optional()
+    .transform((v) => v ?? process.env.REDIS_PASSWORD),
+  LEGACY_RUN_ENGINE_WORKER_REDIS_TLS_DISABLED: z
+    .string()
+    .default(process.env.REDIS_TLS_DISABLED ?? "false"),
+  LEGACY_RUN_ENGINE_WORKER_REDIS_CLUSTER_MODE_ENABLED: z.string().default("0"),
+
+  COMMON_WORKER_ENABLED: z.string().default(process.env.WORKER_ENABLED ?? "true"),
+  COMMON_WORKER_CONCURRENCY_WORKERS: z.coerce.number().int().default(2),
+  COMMON_WORKER_CONCURRENCY_TASKS_PER_WORKER: z.coerce.number().int().default(10),
+  COMMON_WORKER_POLL_INTERVAL: z.coerce.number().int().default(1000),
+  COMMON_WORKER_IMMEDIATE_POLL_INTERVAL: z.coerce.number().int().default(50),
+  COMMON_WORKER_CONCURRENCY_LIMIT: z.coerce.number().int().default(100),
+
+  COMMON_WORKER_REDIS_HOST: z
+    .string()
+    .optional()
+    .transform((v) => v ?? process.env.REDIS_HOST),
+  COMMON_WORKER_REDIS_READER_HOST: z
+    .string()
+    .optional()
+    .transform((v) => v ?? process.env.REDIS_READER_HOST),
+  COMMON_WORKER_REDIS_READER_PORT: z.coerce
+    .number()
+    .optional()
+    .transform(
+      (v) =>
+        v ?? (process.env.REDIS_READER_PORT ? parseInt(process.env.REDIS_READER_PORT) : undefined)
+    ),
+  COMMON_WORKER_REDIS_PORT: z.coerce
+    .number()
+    .optional()
+    .transform((v) => v ?? (process.env.REDIS_PORT ? parseInt(process.env.REDIS_PORT) : undefined)),
+  COMMON_WORKER_REDIS_USERNAME: z
+    .string()
+    .optional()
+    .transform((v) => v ?? process.env.REDIS_USERNAME),
+  COMMON_WORKER_REDIS_PASSWORD: z
+    .string()
+    .optional()
+    .transform((v) => v ?? process.env.REDIS_PASSWORD),
+  COMMON_WORKER_REDIS_TLS_DISABLED: z.string().default(process.env.REDIS_TLS_DISABLED ?? "false"),
+  COMMON_WORKER_REDIS_CLUSTER_MODE_ENABLED: z.string().default("0"),
 });
 
 export type Environment = z.infer<typeof EnvironmentSchema>;

apps/webapp/app/presenters/EventListPresenter.server.ts

@@ -1,169 +0,0 @@
-import { PrismaClient, prisma } from "~/db.server";
-import { Direction, FilterableEnvironment } from "~/components/runs/RunStatuses";
-import { getUsername } from "~/utils/username";
-
-type EventListOptions = {
-  userId: string;
-  organizationSlug: string;
-  projectSlug: string;
-  direction?: Direction;
-  filterEnvironment?: FilterableEnvironment;
-  cursor?: string;
-  pageSize?: number;
-  from?: number;
-  to?: number;
-};
-
-const DEFAULT_PAGE_SIZE = 20;
-
-export type EventList = Awaited<ReturnType<EventListPresenter["call"]>>;
-
-export class EventListPresenter {
-  #prismaClient: PrismaClient;
-
-  constructor(prismaClient: PrismaClient = prisma) {
-    this.#prismaClient = prismaClient;
-  }
-
-  public async call({
-    userId,
-    organizationSlug,
-    projectSlug,
-    filterEnvironment,
-    direction = "forward",
-    cursor,
-    pageSize = DEFAULT_PAGE_SIZE,
-    from,
-    to,
-  }: EventListOptions) {
-    const directionMultiplier = direction === "forward" ? 1 : -1;
-
-    // Find the organization that the user is a member of
-    const organization = await this.#prismaClient.organization.findFirstOrThrow({
-      select: {
-        id: true,
-      },
-      where: {
-        slug: organizationSlug,
-        members: { some: { userId } },
-      },
-    });
-
-    const project = await this.#prismaClient.project.findFirstOrThrow({
-      select: {
-        id: true,
-      },
-      where: {
-        slug: projectSlug,
-        organizationId: organization.id,
-      },
-    });
-
-    const events = await this.#prismaClient.eventRecord.findMany({
-      select: {
-        id: true,
-        name: true,
-        deliverAt: true,
-        deliveredAt: true,
-        isTest: true,
-        createdAt: true,
-        cancelledAt: true,
-        environment: {
-          select: {
-            type: true,
-            slug: true,
-            orgMember: {
-              select: {
-                user: {
-                  select: {
-                    id: true,
-                    name: true,
-                    displayName: true,
-                  },
-                },
-              },
-            },
-          },
-        },
-        runs: {
-          select: {
-            id: true,
-          },
-        },
-      },
-      where: {
-        internal: false,
-        name: {
-          notIn: ["trigger.scheduled", "dev.trigger.scheduled"],
-        },
-        projectId: project.id,
-        organizationId: organization.id,
-        environment: filterEnvironment ? { type: filterEnvironment } : undefined,
-        createdAt: {
-          gte: from ? new Date(from).toISOString() : undefined,
-          lte: to ? new Date(to).toISOString() : undefined,
-        },
-      },
-      orderBy: [{ id: "desc" }],
-      //take an extra record to tell if there are more
-      take: directionMultiplier * (pageSize + 1),
-      //skip the cursor if there is one
-      skip: cursor ? 1 : 0,
-      cursor: cursor
-        ? {
-            id: cursor,
-          }
-        : undefined,
-    });
-
-    const hasMore = events.length > pageSize;
-
-    //get cursors for next and previous pages
-    let next: string | undefined;
-    let previous: string | undefined;
-    switch (direction) {
-      case "forward":
-        previous = cursor ? events.at(0)?.id : undefined;
-        if (hasMore) {
-          next = events[pageSize - 1]?.id;
-        }
-        break;
-      case "backward":
-        if (hasMore) {
-          previous = events[1]?.id;
-          next = events[pageSize]?.id;
-        } else {
-          next = events[pageSize - 1]?.id;
-        }
-        break;
-    }
-
-    const eventsToReturn =
-      direction === "backward" && hasMore
-        ? events.slice(1, pageSize + 1)
-        : events.slice(0, pageSize);
-
-    return {
-      events: eventsToReturn.map((event) => ({
-        id: event.id,
-        name: event.name,
-        deliverAt: event.deliverAt,
-        deliveredAt: event.deliveredAt,
-        createdAt: event.createdAt,
-        cancelledAt: event.cancelledAt,
-        isTest: event.isTest,
-        environment: {
-          type: event.environment.type,
-          slug: event.environment.slug,
-          userId: event.environment.orgMember?.user.id,
-          userName: getUsername(event.environment.orgMember?.user),
-        },
-        runs: event.runs.length,
-      })),
-      pagination: {
-        next,
-        previous,
-      },
-    };
-  }
-}

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

@@ -39,6 +39,7 @@ const commonRunSelect = {
   idempotencyKey: true,
   isTest: true,
   depth: true,
+  scheduleId: true,
   lockedToVersion: {
     select: {
       version: true,
@@ -71,7 +72,6 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
       include: {
         attempts: true,
         lockedToVersion: true,
-        schedule: true,
         tags: true,
         batch: {
           select: {
@@ -157,20 +157,7 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
         output: $output,
         outputPresignedUrl: $outputPresignedUrl,
         error: ApiRetrieveRunPresenter.apiErrorFromError(taskRun.error),
-        schedule: taskRun.schedule
-          ? {
-              id: taskRun.schedule.friendlyId,
-              externalId: taskRun.schedule.externalId ?? undefined,
-              deduplicationKey: taskRun.schedule.userProvidedDeduplicationKey
-                ? taskRun.schedule.deduplicationKey
-                : undefined,
-              generator: {
-                type: "CRON" as const,
-                expression: taskRun.schedule.generatorExpression,
-                description: taskRun.schedule.generatorDescription,
-              },
-            }
-          : undefined,
+        schedule: await resolveSchedule(taskRun),
         // We're removing attempts from the API
         attemptCount: taskRun.attempts.length,
         attempts: [],
@@ -320,6 +307,33 @@ export class ApiRetrieveRunPresenter extends BasePresenter {
   }
 }
 
+async function resolveSchedule(run: CommonRelatedRun) {
+  if (!run.scheduleId) {
+    return undefined;
+  }
+
+  const schedule = await prisma.taskSchedule.findFirst({
+    where: {
+      id: run.scheduleId,
+    },
+  });
+
+  if (!schedule) {
+    return undefined;
+  }
+
+  return {
+    id: schedule.friendlyId,
+    externalId: schedule.externalId ?? undefined,
+    deduplicationKey: schedule.userProvidedDeduplicationKey ? schedule.deduplicationKey : undefined,
+    generator: {
+      type: "CRON" as const,
+      expression: schedule.generatorExpression,
+      description: schedule.generatorDescription,
+    },
+  };
+}
+
 async function createCommonRunStructure(run: CommonRelatedRun) {
   const metadata = await parsePacket({
     data: run.metadata ?? undefined,

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

@@ -102,14 +102,7 @@ export class SpanPresenter extends BasePresenter {
         queue: true,
         concurrencyKey: true,
         //schedule
-        schedule: {
-          select: {
-            friendlyId: true,
-            generatorExpression: true,
-            timezone: true,
-            generatorDescription: true,
-          },
-        },
+        scheduleId: true,
         //usage
         baseCostInCents: true,
         costInCents: true,
@@ -281,14 +274,7 @@ export class SpanPresenter extends BasePresenter {
       sdkVersion: run.lockedToVersion?.sdkVersion,
       isTest: run.isTest,
       environmentId: run.runtimeEnvironment.id,
-      schedule: run.schedule
-        ? {
-            friendlyId: run.schedule.friendlyId,
-            generatorExpression: run.schedule.generatorExpression,
-            description: run.schedule.generatorDescription,
-            timezone: run.schedule.timezone,
-          }
-        : undefined,
+      schedule: await this.resolveSchedule(run.scheduleId ?? undefined),
       queue: {
         name: run.queue,
         isCustomQueue: !run.queue.startsWith("task/"),
@@ -323,6 +309,35 @@ export class SpanPresenter extends BasePresenter {
     };
   }
 
+  async resolveSchedule(scheduleId?: string) {
+    if (!scheduleId) {
+      return;
+    }
+
+    const schedule = await this._replica.taskSchedule.findFirst({
+      where: {
+        id: scheduleId,
+      },
+      select: {
+        friendlyId: true,
+        generatorExpression: true,
+        timezone: true,
+        generatorDescription: true,
+      },
+    });
+
+    if (!schedule) {
+      return;
+    }
+
+    return {
+      friendlyId: schedule.friendlyId,
+      generatorExpression: schedule.generatorExpression,
+      description: schedule.generatorDescription,
+      timezone: schedule.timezone,
+    };
+  }
+
   async getSpan(runFriendlyId: string, spanId: string) {
     const run = await this._prisma.taskRun.findFirst({
       select: {

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.events.$eventParam/route.tsx

@@ -3,7 +3,6 @@ import { LoaderFunctionArgs } from "@remix-run/server-runtime";
 import { typedjson, useTypedLoaderData } from "remix-typedjson";
 import { EventDetail } from "~/components/event/EventDetail";
 import { PageBody, PageContainer } from "~/components/layout/AppLayout";
-import { NavBar, PageTitle } from "~/components/primitives/PageHeader";
 import { RunsFilters } from "~/components/runs/RunFilters";
 import { RunListSearchSchema } from "~/components/runs/RunStatuses";
 import { RunsTable } from "~/components/runs/RunsTable";
@@ -13,7 +12,7 @@ import { useUser } from "~/hooks/useUser";
 import { EventPresenter } from "~/presenters/EventPresenter.server";
 import { RunListPresenter } from "~/presenters/RunListPresenter.server";
 import { requireUserId } from "~/services/session.server";
-import { EventParamSchema, projectEventsPath, projectPath } from "~/utils/pathBuilder";
+import { EventParamSchema, projectPath } from "~/utils/pathBuilder";
 import { ListPagination } from "../../components/ListPagination";
 
 export const loader = async ({ request, params }: LoaderFunctionArgs) => {
@@ -69,16 +68,6 @@ export default function Page() {
 
   return (
     <PageContainer>
-      <NavBar>
-        <PageTitle
-          title={event.name}
-          backButton={{
-            to: projectEventsPath(organization, project),
-            text: "Events",
-          }}
-        />
-      </NavBar>
-
       <PageBody scrollable={false}>
         <div className="grid h-full grid-cols-2">
           <div className="overflow-y-auto p-4 scrollbar-thin scrollbar-track-transparent scrollbar-thumb-charcoal-600">

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.events._index/route.tsx

@@ -1,87 +0,0 @@
-import { useNavigation } from "@remix-run/react";
-import { LoaderFunctionArgs } from "@remix-run/server-runtime";
-import { typedjson, useTypedLoaderData } from "remix-typedjson";
-import { EventListSearchSchema } from "~/components/events/EventStatuses";
-import { EventsFilters } from "~/components/events/EventsFilters";
-import { EventsTable } from "~/components/events/EventsTable";
-import { PageBody, PageContainer } from "~/components/layout/AppLayout";
-import { LinkButton } from "~/components/primitives/Buttons";
-import { NavBar, PageAccessories, PageTitle } from "~/components/primitives/PageHeader";
-import { useOrganization } from "~/hooks/useOrganizations";
-import { useProject } from "~/hooks/useProject";
-import { useUser } from "~/hooks/useUser";
-import { EventListPresenter } from "~/presenters/EventListPresenter.server";
-import { requireUserId } from "~/services/session.server";
-import { ProjectParamSchema, docsPath, projectPath } from "~/utils/pathBuilder";
-import { ListPagination } from "../../components/ListPagination";
-import { BookOpenIcon } from "@heroicons/react/20/solid";
-
-export const loader = async ({ request, params }: LoaderFunctionArgs) => {
-  const userId = await requireUserId(request);
-  const { projectParam, organizationSlug } = ProjectParamSchema.parse(params);
-
-  const url = new URL(request.url);
-  const s = Object.fromEntries(url.searchParams.entries());
-  const searchParams = EventListSearchSchema.parse(s);
-
-  const presenter = new EventListPresenter();
-  const list = await presenter.call({
-    userId,
-    filterEnvironment: searchParams.environment,
-    projectSlug: projectParam,
-    organizationSlug,
-    direction: searchParams.direction,
-    cursor: searchParams.cursor,
-    from: searchParams.from,
-    to: searchParams.to,
-    pageSize: 25,
-  });
-
-  return typedjson({
-    list,
-  });
-};
-
-export default function Page() {
-  const { list } = useTypedLoaderData<typeof loader>();
-  const navigation = useNavigation();
-  const isLoading = navigation.state !== "idle";
-  const organization = useOrganization();
-  const project = useProject();
-  const user = useUser();
-
-  return (
-    <PageContainer>
-      <NavBar>
-        <PageTitle title={`${project.name} events`} />
-        <PageAccessories>
-          <LinkButton
-            variant={"minimal/small"}
-            LeadingIcon={BookOpenIcon}
-            to={docsPath("documentation/concepts/triggers/events")}
-          >
-            Event documentation
-          </LinkButton>
-        </PageAccessories>
-      </NavBar>
-
-      <PageBody scrollable={false}>
-        <div className="h-full overflow-y-auto p-4 scrollbar-thin scrollbar-track-transparent scrollbar-thumb-charcoal-600">
-          <div className="mb-2 flex items-center justify-between gap-x-2">
-            <EventsFilters />
-            <ListPagination list={list} />
-          </div>
-          <EventsTable
-            total={list.events.length}
-            hasFilters={false}
-            events={list.events}
-            isLoading={isLoading}
-            eventsParentPath={projectPath(organization, project)}
-            currentUser={user}
-          />
-          <ListPagination list={list} className="mt-2 justify-end" />
-        </div>
-      </PageBody>
-    </PageContainer>
-  );
-}

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

@@ -1,14 +1,14 @@
 import type { ActionFunctionArgs } from "@remix-run/server-runtime";
 import { json } from "@remix-run/server-runtime";
-import { BatchTriggerTaskRequestBody } from "@trigger.dev/core/v3";
+import { BatchTriggerTaskRequestBody, BatchTriggerTaskV2RequestBody } from "@trigger.dev/core/v3";
 import { z } from "zod";
+import { fromZodError } from "zod-validation-error";
 import { MAX_BATCH_TRIGGER_ITEMS } from "~/consts";
+import { env } from "~/env.server";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
-import { BatchTriggerTaskService } from "~/v3/services/batchTriggerTask.server";
+import { BatchTriggerV3Service } from "~/v3/services/batchTriggerV3.server";
 import { HeadersSchema } from "./api.v1.tasks.$taskId.trigger";
-import { env } from "~/env.server";
-import { fromZodError } from "zod-validation-error";
 
 const ParamsSchema = z.object({
   taskId: z.string(),
@@ -85,15 +85,17 @@ export async function action({ request, params }: ActionFunctionArgs) {
     );
   }
 
-  const service = new BatchTriggerTaskService();
+  const service = new BatchTriggerV3Service();
 
   const traceContext =
     traceparent && isFromWorker // If the request is from a worker, we should pass the trace context
       ? { traceparent, tracestate }
       : undefined;
 
+  const v3Body = convertV1BodyToV2Body(body.data, taskId);
+
   try {
-    const result = await service.call(taskId, authenticationResult.environment, body.data, {
+    const result = await service.call(authenticationResult.environment, v3Body, {
       idempotencyKey: idempotencyKey ?? undefined,
       triggerVersion: triggerVersion ?? undefined,
       traceContext,
@@ -106,8 +108,8 @@ export async function action({ request, params }: ActionFunctionArgs) {
 
     return json(
       {
-        batchId: result.batch.friendlyId,
-        runs: result.runs,
+        batchId: result.id,
+        runs: result.runs.map((run) => run.id),
       },
       {
         headers: {
@@ -126,3 +128,29 @@ export async function action({ request, params }: ActionFunctionArgs) {
     return json({ error: "Something went wrong" }, { status: 500 });
   }
 }
+
+// Strip from options:
+// - dependentBatch
+// - dependentAttempt
+// - parentBatch
+function convertV1BodyToV2Body(
+  body: BatchTriggerTaskRequestBody,
+  taskIdentifier: string
+): BatchTriggerTaskV2RequestBody {
+  return {
+    items: body.items.map((item) => ({
+      task: taskIdentifier,
+      payload: item.payload,
+      context: item.context,
+      options: item.options
+        ? {
+            ...item.options,
+            dependentBatch: undefined,
+            parentBatch: undefined,
+            dependentAttempt: undefined,
+          }
+        : undefined,
+    })),
+    dependentAttempt: body.dependentAttempt,
+  };
+}

apps/webapp/app/routes/resources.runs.$runParam.ts

@@ -54,14 +54,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
       queue: true,
       concurrencyKey: true,
       //schedule
-      schedule: {
-        select: {
-          friendlyId: true,
-          generatorExpression: true,
-          timezone: true,
-          generatorDescription: true,
-        },
-      },
+      scheduleId: true,
       //usage
       baseCostInCents: true,
       costInCents: true,
@@ -212,14 +205,7 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
     sdkVersion: run.lockedToVersion?.sdkVersion,
     isTest: run.isTest,
     environmentId: run.runtimeEnvironment.id,
-    schedule: run.schedule
-      ? {
-          friendlyId: run.schedule.friendlyId,
-          generatorExpression: run.schedule.generatorExpression,
-          description: run.schedule.generatorDescription,
-          timezone: run.schedule.timezone,
-        }
-      : undefined,
+    schedule: await resolveSchedule(run.scheduleId ?? undefined),
     queue: {
       name: run.queue,
       isCustomQueue: !run.queue.startsWith("task/"),
@@ -240,3 +226,32 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
     context: JSON.stringify(context, null, 2),
   });
 };
+
+async function resolveSchedule(scheduleId?: string) {
+  if (!scheduleId) {
+    return;
+  }
+
+  const schedule = await $replica.taskSchedule.findFirst({
+    where: {
+      id: scheduleId,
+    },
+    select: {
+      friendlyId: true,
+      generatorExpression: true,
+      timezone: true,
+      generatorDescription: true,
+    },
+  });
+
+  if (!schedule) {
+    return;
+  }
+
+  return {
+    friendlyId: schedule.friendlyId,
+    generatorExpression: schedule.generatorExpression,
+    description: schedule.generatorDescription,
+    timezone: schedule.timezone,
+  };
+}

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

@@ -6,10 +6,8 @@ import { $replica, prisma } from "~/db.server";
 import { env } from "~/env.server";
 import { MarqsConcurrencyMonitor } from "~/v3/marqs/concurrencyMonitor.server";
 import { RequeueV2Message } from "~/v3/marqs/requeueV2Message.server";
-import { RequeueTaskRunService } from "~/v3/requeueTaskRun.server";
 import { DeliverAlertService } from "~/v3/services/alerts/deliverAlert.server";
 import { PerformDeploymentAlertsService } from "~/v3/services/alerts/performDeploymentAlerts.server";
-import { PerformTaskAttemptAlertsService } from "~/v3/services/alerts/performTaskAttemptAlerts.server";
 import { PerformBulkActionService } from "~/v3/services/bulk/performBulkAction.server";
 import { CancelTaskAttemptDependenciesService } from "~/v3/services/cancelTaskAttemptDependencies.server";
 import { EnqueueDelayedRunService } from "~/v3/services/enqueueDelayedRun.server";
@@ -157,9 +155,6 @@ const workerCatalog = {
   "v3.performTaskRunAlerts": z.object({
     runId: z.string(),
   }),
-  "v3.performTaskAttemptAlerts": z.object({
-    attemptId: z.string(),
-  }),
   "v3.deliverAlert": z.object({
     alertId: z.string(),
   }),
@@ -610,15 +605,6 @@ function getWorkerQueue() {
           return await service.call(payload.runId);
         },
       },
-      "v3.performTaskAttemptAlerts": {
-        priority: 0,
-        maxAttempts: 3,
-        handler: async (payload, job) => {
-          const service = new PerformTaskAttemptAlertsService();
-
-          return await service.call(payload.attemptId);
-        },
-      },
       "v3.deliverAlert": {
         priority: 0,
         maxAttempts: 8,
@@ -658,11 +644,7 @@ function getWorkerQueue() {
       "v3.requeueTaskRun": {
         priority: 0,
         maxAttempts: 3,
-        handler: async (payload, job) => {
-          const service = new RequeueTaskRunService();
-
-          await service.call(payload.runId);
-        },
+        handler: async (payload, job) => {}, // This is now handled by redisWorker
       },
       "v3.retryAttempt": {
         priority: 0,

apps/webapp/app/utils/pathBuilder.ts

@@ -252,7 +252,7 @@ export function projectTriggersPath(organization: OrgForPath, project: ProjectFo
   return `${projectPath(organization, project)}/triggers`;
 }
 
-export function projectEventsPath(organization: OrgForPath, project: ProjectForPath) {
+function projectEventsPath(organization: OrgForPath, project: ProjectForPath) {
   return `${projectPath(organization, project)}/events`;
 }
 

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

@@ -0,0 +1,171 @@
+import Redis, { Cluster } from "ioredis";
+
+/**
+ * Options for configuring the RateLimiter.
+ */
+export interface GCRARateLimiterOptions {
+  /** An instance of ioredis. */
+  redis: Redis | Cluster;
+  /**
+   * A string prefix to namespace keys in Redis.
+   * Defaults to "ratelimit:".
+   */
+  keyPrefix?: string;
+  /**
+   * The minimum interval between requests (the emission interval) in milliseconds.
+   * For example, 1000 ms for one request per second.
+   */
+  emissionInterval: number;
+  /**
+   * The burst tolerance in milliseconds. This represents how much “credit” can be
+   * accumulated to allow short bursts beyond the average rate.
+   * For example, if you want to allow 3 requests in a burst with an emission interval of 1000 ms,
+   * you might set this to 3000.
+   */
+  burstTolerance: number;
+  /**
+   * Expiration for the Redis key in milliseconds.
+   * Defaults to the larger of 60 seconds or (emissionInterval + burstTolerance).
+   */
+  keyExpiration?: number;
+}
+
+/**
+ * The result of a rate limit check.
+ */
+export interface RateLimitResult {
+  /** Whether the request is allowed. */
+  allowed: boolean;
+  /**
+   * If not allowed, this is the number of milliseconds the caller should wait
+   * before retrying.
+   */
+  retryAfter?: number;
+}
+
+/**
+ * A rate limiter using Redis and the Generic Cell Rate Algorithm (GCRA).
+ *
+ * The GCRA is implemented using a Lua script that runs atomically in Redis.
+ *
+ * When a request comes in, the algorithm:
+ *  - Retrieves the current "Theoretical Arrival Time" (TAT) from Redis (or initializes it if missing).
+ *  - If the current time is greater than or equal to the TAT, the request is allowed and the TAT is updated to now + emissionInterval.
+ *  - Otherwise, if the current time plus the burst tolerance is at least the TAT, the request is allowed and the TAT is incremented.
+ *  - If neither condition is met, the request is rejected and a Retry-After value is returned.
+ */
+export class GCRARateLimiter {
+  private redis: Redis | Cluster;
+  private keyPrefix: string;
+  private emissionInterval: number;
+  private burstTolerance: number;
+  private keyExpiration: number;
+
+  constructor(options: GCRARateLimiterOptions) {
+    this.redis = options.redis;
+    this.keyPrefix = options.keyPrefix || "gcra:ratelimit:";
+    this.emissionInterval = options.emissionInterval;
+    this.burstTolerance = options.burstTolerance;
+    // Default expiration: at least 60 seconds or the sum of emissionInterval and burstTolerance
+    this.keyExpiration =
+      options.keyExpiration || Math.max(60_000, this.emissionInterval + this.burstTolerance);
+
+    // Define a custom Redis command 'gcra' that implements the GCRA algorithm.
+    // Using defineCommand ensures the Lua script is loaded once and run atomically.
+    this.redis.defineCommand("gcra", {
+      numberOfKeys: 1,
+      lua: `
+--[[
+  GCRA Lua script
+  KEYS[1]         - The rate limit key (e.g. "ratelimit:<identifier>")
+  ARGV[1]         - Current time in ms (number)
+  ARGV[2]         - Emission interval in ms (number)
+  ARGV[3]         - Burst tolerance in ms (number)
+  ARGV[4]         - Key expiration in ms (number)
+  
+  Returns: { allowedFlag, value }
+    allowedFlag: 1 if allowed, 0 if rate-limited.
+    value: 0 when allowed; if not allowed, the number of ms to wait.
+]]--
+
+local key = KEYS[1]
+local now = tonumber(ARGV[1])
+local emission_interval = tonumber(ARGV[2])
+local burst_tolerance = tonumber(ARGV[3])
+local expire = tonumber(ARGV[4])
+
+-- Get the stored Theoretical Arrival Time (TAT) or default to 0.
+local tat = tonumber(redis.call("GET", key) or 0)
+if tat == 0 then
+  tat = now
+end
+
+local allowed, new_tat, retry_after
+
+if now >= tat then
+  -- No delay: request is on schedule.
+  new_tat = now + emission_interval
+  allowed = true
+elseif (now + burst_tolerance) >= tat then
+  -- Within burst capacity: allow request.
+  new_tat = tat + emission_interval
+  allowed = true
+else
+  -- Request exceeds the allowed burst; calculate wait time.
+  allowed = false
+  retry_after = tat - (now + burst_tolerance)
+end
+
+if allowed then
+  redis.call("SET", key, new_tat, "PX", expire)
+  return {1, 0}
+else
+  return {0, retry_after}
+end
+`,
+    });
+  }
+
+  /**
+   * Checks whether a request associated with the given identifier is allowed.
+   *
+   * @param identifier A unique string identifying the subject of rate limiting (e.g. user ID, IP address, or domain).
+   * @returns A promise that resolves to a RateLimitResult.
+   *
+   * @example
+   * const result = await rateLimiter.check('user:12345');
+   * if (!result.allowed) {
+   *   // Tell the client to retry after result.retryAfter milliseconds.
+   * }
+   */
+  async check(identifier: string): Promise<RateLimitResult> {
+    const key = `${this.keyPrefix}${identifier}`;
+    const now = Date.now();
+
+    try {
+      // Call the custom 'gcra' command.
+      // The script returns an array: [allowedFlag, value]
+      //   - allowedFlag: 1 if allowed; 0 if rejected.
+      //   - value: 0 when allowed; if rejected, the number of ms to wait before retrying.
+      // @ts-expect-error: The custom command is defined via defineCommand.
+      const result: [number, number] = await this.redis.gcra(
+        key,
+        now,
+        this.emissionInterval,
+        this.burstTolerance,
+        this.keyExpiration
+      );
+      const allowed = result[0] === 1;
+      if (allowed) {
+        return { allowed: true };
+      } else {
+        return { allowed: false, retryAfter: result[1] };
+      }
+    } catch (error) {
+      // In a production system you might log the error and either
+      // allow the request (fail open) or deny it (fail closed).
+      // Here we choose to propagate the error.
+      throw error;
+    }
+  }
+}

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

@@ -0,0 +1,30 @@
+import { env } from "~/env.server";
+import { createRedisClient } from "~/redis.server";
+import { GCRARateLimiter } from "./GCRARateLimiter.server";
+import { singleton } from "~/utils/singleton";
+import { logger } from "~/services/logger.server";
+
+export const alertsRateLimiter = singleton("alertsRateLimiter", initializeAlertsRateLimiter);
+
+function initializeAlertsRateLimiter() {
+  const redis = createRedisClient("alerts:ratelimiter", {
+    keyPrefix: "alerts:ratelimiter:",
+    host: env.ALERT_RATE_LIMITER_REDIS_HOST,
+    port: env.ALERT_RATE_LIMITER_REDIS_PORT,
+    username: env.ALERT_RATE_LIMITER_REDIS_USERNAME,
+    password: env.ALERT_RATE_LIMITER_REDIS_PASSWORD,
+    tlsDisabled: env.ALERT_RATE_LIMITER_REDIS_TLS_DISABLED === "true",
+    clusterMode: env.ALERT_RATE_LIMITER_REDIS_CLUSTER_MODE_ENABLED === "1",
+  });
+
+  logger.debug(`🚦 Initializing alerts rate limiter at host ${env.ALERT_RATE_LIMITER_REDIS_HOST}`, {
+    emissionInterval: env.ALERT_RATE_LIMITER_EMISSION_INTERVAL,
+    burstTolerance: env.ALERT_RATE_LIMITER_BURST_TOLERANCE,
+  });
+
+  return new GCRARateLimiter({
+    redis,
+    emissionInterval: env.ALERT_RATE_LIMITER_EMISSION_INTERVAL,
+    burstTolerance: env.ALERT_RATE_LIMITER_BURST_TOLERANCE,
+  });
+}

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

@@ -0,0 +1,123 @@
+import { Worker as RedisWorker } from "@internal/redis-worker";
+import { Logger } from "@trigger.dev/core/logger";
+import { z } from "zod";
+import { env } from "~/env.server";
+import { logger } from "~/services/logger.server";
+import { singleton } from "~/utils/singleton";
+import { DeliverAlertService } from "./services/alerts/deliverAlert.server";
+import { PerformDeploymentAlertsService } from "./services/alerts/performDeploymentAlerts.server";
+import { PerformTaskRunAlertsService } from "./services/alerts/performTaskRunAlerts.server";
+import { ExpireEnqueuedRunService } from "./services/expireEnqueuedRun.server";
+import { EnqueueDelayedRunService } from "./services/enqueueDelayedRun.server";
+
+function initializeWorker() {
+  const redisOptions = {
+    keyPrefix: "common:worker:",
+    host: env.COMMON_WORKER_REDIS_HOST,
+    port: env.COMMON_WORKER_REDIS_PORT,
+    username: env.COMMON_WORKER_REDIS_USERNAME,
+    password: env.COMMON_WORKER_REDIS_PASSWORD,
+    enableAutoPipelining: true,
+    ...(env.COMMON_WORKER_REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }),
+  };
+
+  logger.debug(`👨‍🏭 Initializing common worker at host ${env.COMMON_WORKER_REDIS_HOST}`);
+
+  const worker = new RedisWorker({
+    name: "common-worker",
+    redisOptions,
+    catalog: {
+      "v3.performTaskRunAlerts": {
+        schema: z.object({
+          runId: z.string(),
+        }),
+        visibilityTimeoutMs: 60_000,
+        retry: {
+          maxAttempts: 3,
+        },
+      },
+      "v3.performDeploymentAlerts": {
+        schema: z.object({
+          deploymentId: z.string(),
+        }),
+        visibilityTimeoutMs: 60_000,
+        retry: {
+          maxAttempts: 3,
+        },
+      },
+      "v3.deliverAlert": {
+        schema: z.object({
+          alertId: z.string(),
+        }),
+        visibilityTimeoutMs: 60_000,
+        retry: {
+          maxAttempts: 3,
+        },
+      },
+      "v3.expireRun": {
+        schema: z.object({
+          runId: z.string(),
+        }),
+        visibilityTimeoutMs: 60_000,
+        retry: {
+          maxAttempts: 6,
+        },
+      },
+      "v3.enqueueDelayedRun": {
+        schema: z.object({
+          runId: z.string(),
+        }),
+        visibilityTimeoutMs: 60_000,
+        retry: {
+          maxAttempts: 6,
+        },
+      },
+    },
+    concurrency: {
+      workers: env.COMMON_WORKER_CONCURRENCY_WORKERS,
+      tasksPerWorker: env.COMMON_WORKER_CONCURRENCY_TASKS_PER_WORKER,
+      limit: env.COMMON_WORKER_CONCURRENCY_LIMIT,
+    },
+    pollIntervalMs: env.COMMON_WORKER_POLL_INTERVAL,
+    immediatePollIntervalMs: env.COMMON_WORKER_IMMEDIATE_POLL_INTERVAL,
+    logger: new Logger("CommonWorker", "debug"),
+    jobs: {
+      "v3.deliverAlert": async ({ payload }) => {
+        const service = new DeliverAlertService();
+
+        await service.call(payload.alertId);
+      },
+      "v3.performDeploymentAlerts": async ({ payload }) => {
+        const service = new PerformDeploymentAlertsService();
+
+        await service.call(payload.deploymentId);
+      },
+      "v3.performTaskRunAlerts": async ({ payload }) => {
+        const service = new PerformTaskRunAlertsService();
+        await service.call(payload.runId);
+      },
+      "v3.expireRun": async ({ payload }) => {
+        const service = new ExpireEnqueuedRunService();
+
+        await service.call(payload.runId);
+      },
+      "v3.enqueueDelayedRun": async ({ payload }) => {
+        const service = new EnqueueDelayedRunService();
+
+        await service.call(payload.runId);
+      },
+    },
+  });
+
+  if (env.COMMON_WORKER_ENABLED === "true") {
+    logger.debug(
+      `👨‍🏭 Starting common worker at host ${env.COMMON_WORKER_REDIS_HOST}, pollInterval = ${env.COMMON_WORKER_POLL_INTERVAL}, immediatePollInterval = ${env.COMMON_WORKER_IMMEDIATE_POLL_INTERVAL}, workers = ${env.COMMON_WORKER_CONCURRENCY_WORKERS}, tasksPerWorker = ${env.COMMON_WORKER_CONCURRENCY_TASKS_PER_WORKER}, concurrencyLimit = ${env.COMMON_WORKER_CONCURRENCY_LIMIT}`
+    );
+
+    worker.start();
+  }
+
+  return worker;
+}
+
+export const commonWorker = singleton("commonWorker", initializeWorker);

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

@@ -1,7 +1,9 @@
+import { nanoid } from "nanoid";
+
 export type DynamicFlushSchedulerConfig<T> = {
   batchSize: number;
   flushInterval: number;
-  callback: (batch: T[]) => Promise<void>;
+  callback: (flushId: string, batch: T[]) => Promise<void>;
 };
 
 export class DynamicFlushScheduler<T> {
@@ -10,7 +12,7 @@ export class DynamicFlushScheduler<T> {
   private readonly BATCH_SIZE: number;
   private readonly FLUSH_INTERVAL: number;
   private flushTimer: NodeJS.Timeout | null;
-  private readonly callback: (batch: T[]) => Promise<void>;
+  private readonly callback: (flushId: string, batch: T[]) => Promise<void>;
 
   constructor(config: DynamicFlushSchedulerConfig<T>) {
     this.batchQueue = [];
@@ -57,7 +59,7 @@ export class DynamicFlushScheduler<T> {
 
     const batchToFlush = this.batchQueue.shift();
     try {
-      await this.callback(batchToFlush!);
+      await this.callback(nanoid(), batchToFlush!);
       if (this.batchQueue.length > 0) {
         this.flushNextBatch();
       }

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

@@ -1,4 +1,4 @@
-import { Attributes, Link, TraceFlags } from "@opentelemetry/api";
+import { Attributes, Link, trace, TraceFlags, Tracer } from "@opentelemetry/api";
 import { RandomIdGenerator } from "@opentelemetry/sdk-trace-base";
 import { SemanticResourceAttributes } from "@opentelemetry/semantic-conventions";
 import {
@@ -32,6 +32,8 @@ import { singleton } from "~/utils/singleton";
 import { DynamicFlushScheduler } from "./dynamicFlushScheduler.server";
 import { startActiveSpan } from "./tracer.server";
 import { createRedisClient, RedisClient, RedisWithClusterOptions } from "~/redis.server";
+import { startSpan } from "./tracing.server";
+import { nanoid } from "nanoid";
 
 const MAX_FLUSH_DEPTH = 5;
 
@@ -99,6 +101,7 @@ export type EventRepoConfig = {
   batchInterval: number;
   redis: RedisWithClusterOptions;
   retentionInDays: number;
+  tracer?: Tracer;
 };
 
 export type QueryOptions = Prisma.TaskEventWhereInput;
@@ -202,6 +205,8 @@ export class EventRepository {
   private _randomIdGenerator = new RandomIdGenerator();
   private _redisPublishClient: RedisClient;
   private _subscriberCount = 0;
+  private _tracer: Tracer;
+  private _lastFlushedAt: Date | undefined;
 
   get subscriberCount() {
     return this._subscriberCount;
@@ -219,6 +224,7 @@ export class EventRepository {
     });
 
     this._redisPublishClient = createRedisClient("trigger:eventRepoPublisher", this._config.redis);
+    this._tracer = _config.tracer ?? trace.getTracer("eventRepo", "0.0.1");
   }
 
   async insert(event: CreatableEvent) {
@@ -226,7 +232,7 @@ export class EventRepository {
   }
 
   async insertImmediate(event: CreatableEvent) {
-    await this.#flushBatch([event]);
+    await this.#flushBatch(nanoid(), [event]);
   }
 
   async insertMany(events: CreatableEvent[]) {
@@ -234,7 +240,7 @@ export class EventRepository {
   }
 
   async insertManyImmediate(events: CreatableEvent[]) {
-    return await this.#flushBatch(events);
+    return await this.#flushBatch(nanoid(), events);
   }
 
   async completeEvent(spanId: string, options?: UpdateEventOptions) {
@@ -1019,82 +1025,120 @@ export class EventRepository {
     };
   }
 
-  async #flushBatch(batch: CreatableEvent[]) {
-    const events = excludePartialEventsWithCorrespondingFullEvent(batch);
+  async #flushBatch(flushId: string, batch: CreatableEvent[]) {
+    return await startSpan(this._tracer, "flushBatch", async (span) => {
+      const events = excludePartialEventsWithCorrespondingFullEvent(batch);
 
-    const flushedEvents = await this.#doFlushBatch(events);
+      span.setAttribute("flush_id", flushId);
+      span.setAttribute("event_count", events.length);
+      span.setAttribute("partial_event_count", batch.length - events.length);
+      span.setAttribute(
+        "last_flush_in_ms",
+        this._lastFlushedAt ? new Date().getTime() - this._lastFlushedAt.getTime() : 0
+      );
 
-    if (flushedEvents.length !== events.length) {
-      logger.debug("[EventRepository][flushBatch] Failed to insert all events", {
-        attemptCount: events.length,
-        successCount: flushedEvents.length,
-      });
-    }
+      const flushedEvents = await this.#doFlushBatch(flushId, events);
 
-    this.#publishToRedis(flushedEvents);
-  }
+      this._lastFlushedAt = new Date();
 
-  async #doFlushBatch(events: CreatableEvent[], depth: number = 1): Promise<CreatableEvent[]> {
-    try {
-      await this.db.taskEvent.createMany({
-        data: events as Prisma.TaskEventCreateManyInput[],
-      });
-
-      return events;
-    } catch (error) {
-      if (error instanceof Prisma.PrismaClientUnknownRequestError) {
-        logger.error("Failed to insert events, most likely because of null characters", {
-          error: {
-            name: error.name,
-            message: error.message,
-            stack: error.stack,
-            clientVersion: error.clientVersion,
-          },
+      if (flushedEvents.length !== events.length) {
+        logger.debug("[EventRepository][flushBatch] Failed to insert all events", {
+          attemptCount: events.length,
+          successCount: flushedEvents.length,
         });
 
-        if (events.length === 1) {
-          logger.debug("Attempting to insert event individually and it failed", {
-            event: events[0],
-            error: {
-              name: error.name,
-              message: error.message,
-              stack: error.stack,
-              clientVersion: error.clientVersion,
-            },
-          });
-
-          return [];
-        }
-
-        if (depth > MAX_FLUSH_DEPTH) {
-          logger.error("Failed to insert events, reached maximum depth", {
-            error: {
-              name: error.name,
-              message: error.message,
-              stack: error.stack,
-              clientVersion: error.clientVersion,
-            },
-            depth,
-            eventsCount: events.length,
-          });
-
-          return [];
-        }
-
-        // Split the events into two batches, and recursively try to insert them.
-        const middle = Math.floor(events.length / 2);
-        const [firstHalf, secondHalf] = [events.slice(0, middle), events.slice(middle)];
-
-        const [firstHalfEvents, secondHalfEvents] = await Promise.all([
-          this.#doFlushBatch(firstHalf, depth + 1),
-          this.#doFlushBatch(secondHalf, depth + 1),
-        ]);
-
-        return firstHalfEvents.concat(secondHalfEvents);
+        span.setAttribute("failed_event_count", events.length - flushedEvents.length);
       }
 
-      throw error;
-    }
+      this.#publishToRedis(flushedEvents);
+    });
+  }
+
+  async #doFlushBatch(
+    flushId: string,
+    events: CreatableEvent[],
+    depth: number = 1
+  ): Promise<CreatableEvent[]> {
+    return await startSpan(this._tracer, "doFlushBatch", async (span) => {
+      try {
+        span.setAttribute("event_count", events.length);
+        span.setAttribute("depth", depth);
+        span.setAttribute("flush_id", flushId);
+
+        await this.db.taskEvent.createMany({
+          data: events as Prisma.TaskEventCreateManyInput[],
+        });
+
+        span.setAttribute("inserted_event_count", events.length);
+
+        return events;
+      } catch (error) {
+        if (error instanceof Prisma.PrismaClientUnknownRequestError) {
+          logger.error("Failed to insert events, most likely because of null characters", {
+            error: {
+              name: error.name,
+              message: error.message,
+              stack: error.stack,
+              clientVersion: error.clientVersion,
+            },
+          });
+
+          if (events.length === 1) {
+            logger.debug("Attempting to insert event individually and it failed", {
+              event: events[0],
+              error: {
+                name: error.name,
+                message: error.message,
+                stack: error.stack,
+                clientVersion: error.clientVersion,
+              },
+            });
+
+            span.setAttribute("failed_event_count", 1);
+
+            return [];
+          }
+
+          if (depth > MAX_FLUSH_DEPTH) {
+            logger.error("Failed to insert events, reached maximum depth", {
+              error: {
+                name: error.name,
+                message: error.message,
+                stack: error.stack,
+                clientVersion: error.clientVersion,
+              },
+              depth,
+              eventsCount: events.length,
+            });
+
+            span.setAttribute("reached_max_flush_depth", true);
+            span.setAttribute("failed_event_count", events.length);
+
+            return [];
+          }
+
+          // Split the events into two batches, and recursively try to insert them.
+          const middle = Math.floor(events.length / 2);
+          const [firstHalf, secondHalf] = [events.slice(0, middle), events.slice(middle)];
+
+          return await startSpan(this._tracer, "bisectBatch", async (span) => {
+            span.setAttribute("first_half_count", firstHalf.length);
+            span.setAttribute("second_half_count", secondHalf.length);
+            span.setAttribute("depth", depth);
+            span.setAttribute("flush_id", flushId);
+
+            const [firstHalfEvents, secondHalfEvents] = await Promise.all([
+              this.#doFlushBatch(flushId, firstHalf, depth + 1),
+              this.#doFlushBatch(flushId, secondHalf, depth + 1),
+            ]);
+
+            return firstHalfEvents.concat(secondHalfEvents);
+          });
+        }
+
+        throw error;
+      }
+    });
   }
 
   async #publishToRedis(events: CreatableEvent[]) {

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

@@ -5,14 +5,14 @@ import {
   TaskRunExecutionRetry,
   TaskRunFailedExecutionResult,
 } from "@trigger.dev/core/v3";
-import { logger } from "~/services/logger.server";
-import { BaseService } from "./services/baseService.server";
-import { isFailableRunStatus, isFinalAttemptStatus } from "./taskStatus";
 import type { Prisma, TaskRun } from "@trigger.dev/database";
+import * as semver from "semver";
+import { logger } from "~/services/logger.server";
+import { sharedQueueTasks } from "./marqs/sharedQueueConsumer.server";
+import { BaseService } from "./services/baseService.server";
 import { CompleteAttemptService } from "./services/completeAttempt.server";
 import { CreateTaskRunAttemptService } from "./services/createTaskRunAttempt.server";
-import { sharedQueueTasks } from "./marqs/sharedQueueConsumer.server";
-import * as semver from "semver";
+import { isFailableRunStatus, isFinalAttemptStatus } from "./taskStatus";
 
 const FailedTaskRunRetryGetPayload = {
   select: {
@@ -180,13 +180,52 @@ export class FailedTaskRunRetryHelper extends BaseService {
     }
   }
 
-  static async getExecutionRetry({
+  static getExecutionRetry({
     run,
     execution,
   }: {
     run: TaskRunWithWorker;
     execution: TaskRunExecution;
-  }): Promise<TaskRunExecutionRetry | undefined> {
+  }): TaskRunExecutionRetry | undefined {
+    try {
+      const retryConfig = FailedTaskRunRetryHelper.getRetryConfig({ run, execution });
+      if (!retryConfig) {
+        return;
+      }
+
+      const delay = calculateNextRetryDelay(retryConfig, execution.attempt.number);
+
+      if (!delay) {
+        logger.debug("[FailedTaskRunRetryHelper] No more retries", {
+          run,
+          execution,
+        });
+
+        return;
+      }
+
+      return {
+        timestamp: Date.now() + delay,
+        delay,
+      };
+    } catch (error) {
+      logger.error("[FailedTaskRunRetryHelper] Failed to get execution retry", {
+        run,
+        execution,
+        error,
+      });
+
+      return;
+    }
+  }
+
+  static getRetryConfig({
+    run,
+    execution,
+  }: {
+    run: TaskRunWithWorker;
+    execution: TaskRunExecution;
+  }): RetryOptions | undefined {
     try {
       const retryConfig = run.lockedBy?.retryConfig;
 
@@ -247,21 +286,7 @@ export class FailedTaskRunRetryHelper extends BaseService {
         return;
       }
 
-      const delay = calculateNextRetryDelay(parsedRetryConfig.data, execution.attempt.number);
-
-      if (!delay) {
-        logger.debug("[FailedTaskRunRetryHelper] No more retries", {
-          run,
-          execution,
-        });
-
-        return;
-      }
-
-      return {
-        timestamp: Date.now() + delay,
-        delay,
-      };
+      return parsedRetryConfig.data;
     } catch (error) {
       logger.error("[FailedTaskRunRetryHelper] Failed to get execution retry", {
         run,

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

@@ -0,0 +1,91 @@
+import { Worker as RedisWorker } from "@internal/redis-worker";
+import { Logger } from "@trigger.dev/core/logger";
+import { z } from "zod";
+import { env } from "~/env.server";
+import { logger } from "~/services/logger.server";
+import { singleton } from "~/utils/singleton";
+import { TaskRunHeartbeatFailedService } from "./taskRunHeartbeatFailed.server";
+import { completeBatchTaskRunItemV3 } from "./services/batchTriggerV3.server";
+import { prisma } from "~/db.server";
+
+function initializeWorker() {
+  const redisOptions = {
+    keyPrefix: "legacy-run-engine:worker:",
+    host: env.LEGACY_RUN_ENGINE_WORKER_REDIS_HOST,
+    port: env.LEGACY_RUN_ENGINE_WORKER_REDIS_PORT,
+    username: env.LEGACY_RUN_ENGINE_WORKER_REDIS_USERNAME,
+    password: env.LEGACY_RUN_ENGINE_WORKER_REDIS_PASSWORD,
+    enableAutoPipelining: true,
+    ...(env.LEGACY_RUN_ENGINE_WORKER_REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }),
+  };
+
+  logger.debug(
+    `👨‍🏭 Initializing legacy run engine worker at host ${env.LEGACY_RUN_ENGINE_WORKER_REDIS_HOST}`
+  );
+
+  const worker = new RedisWorker({
+    name: "legacy-run-engine-worker",
+    redisOptions,
+    catalog: {
+      runHeartbeat: {
+        schema: z.object({
+          runId: z.string(),
+        }),
+        visibilityTimeoutMs: 60_000,
+        retry: {
+          maxAttempts: 3,
+        },
+      },
+      completeBatchTaskRunItem: {
+        schema: z.object({
+          itemId: z.string(),
+          batchTaskRunId: z.string(),
+          scheduleResumeOnComplete: z.boolean(),
+          taskRunAttemptId: z.string().optional(),
+          attempt: z.number().optional(),
+        }),
+        visibilityTimeoutMs: 60_000,
+        retry: {
+          maxAttempts: 10,
+        },
+      },
+    },
+    concurrency: {
+      workers: env.LEGACY_RUN_ENGINE_WORKER_CONCURRENCY_WORKERS,
+      tasksPerWorker: env.LEGACY_RUN_ENGINE_WORKER_CONCURRENCY_TASKS_PER_WORKER,
+      limit: env.LEGACY_RUN_ENGINE_WORKER_CONCURRENCY_LIMIT,
+    },
+    pollIntervalMs: env.LEGACY_RUN_ENGINE_WORKER_POLL_INTERVAL,
+    immediatePollIntervalMs: env.LEGACY_RUN_ENGINE_WORKER_IMMEDIATE_POLL_INTERVAL,
+    logger: new Logger("LegacyRunEngineWorker", "debug"),
+    jobs: {
+      runHeartbeat: async ({ payload }) => {
+        const service = new TaskRunHeartbeatFailedService();
+
+        await service.call(payload.runId);
+      },
+      completeBatchTaskRunItem: async ({ payload, attempt }) => {
+        await completeBatchTaskRunItemV3(
+          payload.itemId,
+          payload.batchTaskRunId,
+          prisma,
+          payload.scheduleResumeOnComplete,
+          payload.taskRunAttemptId,
+          attempt
+        );
+      },
+    },
+  });
+
+  if (env.LEGACY_RUN_ENGINE_WORKER_ENABLED === "true") {
+    logger.debug(
+      `👨‍🏭 Starting legacy run engine worker at host ${env.LEGACY_RUN_ENGINE_WORKER_REDIS_HOST}, pollInterval = ${env.LEGACY_RUN_ENGINE_WORKER_POLL_INTERVAL}, immediatePollInterval = ${env.LEGACY_RUN_ENGINE_WORKER_IMMEDIATE_POLL_INTERVAL}, workers = ${env.LEGACY_RUN_ENGINE_WORKER_CONCURRENCY_WORKERS}, tasksPerWorker = ${env.LEGACY_RUN_ENGINE_WORKER_CONCURRENCY_TASKS_PER_WORKER}, concurrencyLimit = ${env.LEGACY_RUN_ENGINE_WORKER_CONCURRENCY_LIMIT}`
+    );
+
+    worker.start();
+  }
+
+  return worker;
+}
+
+export const legacyRunEngineWorker = singleton("legacyRunEngineWorker", initializeWorker);

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

@@ -30,7 +30,7 @@ import {
   MessageQueueSubscriber,
   VisibilityTimeoutStrategy,
 } from "./types";
-import { V3VisibilityTimeout } from "./v3VisibilityTimeout.server";
+import { V3LegacyRunEngineWorkerVisibilityTimeout } from "./v3VisibilityTimeout.server";
 
 const KEY_PREFIX = "marqs:";
 
@@ -638,7 +638,8 @@ export class MarQS {
   }
 
   /**
-   * Negative acknowledge a message, which will requeue the message
+   * Negative acknowledge a message, which will requeue the message.
+   * Returns whether it went back into the queue or not.
    */
   public async nackMessage(
     messageId: string,
@@ -657,7 +658,7 @@ export class MarQS {
             updates,
             service: this.name,
           });
-          return;
+          return false;
         }
 
         const nackCount = await this.#getNackCount(messageId);
@@ -676,7 +677,7 @@ export class MarQS {
 
           // If we have reached the maximum nack count, we will ack the message
           await this.acknowledgeMessage(messageId, "maximum nack count reached");
-          return;
+          return false;
         }
 
         span.setAttributes({
@@ -705,6 +706,8 @@ export class MarQS {
         });
 
         await this.options.subscriber?.messageNacked(message);
+
+        return true;
       },
       {
         kind: SpanKind.CONSUMER,
@@ -1264,7 +1267,7 @@ 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 '1000000')
+local concurrencyLimit = tonumber(redis.call('GET', concurrencyLimitKey) or envConcurrencyLimit)
 
 -- Check condition only if concurrencyLimit exists
 if currentConcurrency >= concurrencyLimit then
@@ -1434,7 +1437,7 @@ local currentEnvConcurrency = tonumber(redis.call('SCARD', currentEnvConcurrency
 
 local currentConcurrency = tonumber(redis.call('SCARD', currentConcurrencyKey) or '0')
 
-return { currentOrgConcurrency, currentEnvConcurrency, currentConcurrency } 
+return { currentOrgConcurrency, currentEnvConcurrency, currentConcurrency }
       `,
     });
 
@@ -1611,7 +1614,7 @@ function getMarQSClient() {
         name: "marqs",
         tracer: trace.getTracer("marqs"),
         keysProducer,
-        visibilityTimeoutStrategy: new V3VisibilityTimeout(),
+        visibilityTimeoutStrategy: new V3LegacyRunEngineWorkerVisibilityTimeout(),
         queuePriorityStrategy: new FairDequeuingStrategy({
           tracer: tracer,
           redis,

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

@@ -1,12 +1,28 @@
-import { RequeueTaskRunService } from "../requeueTaskRun.server";
+import { legacyRunEngineWorker } from "../legacyRunEngineWorker.server";
+import { TaskRunHeartbeatFailedService } from "../taskRunHeartbeatFailed.server";
 import { VisibilityTimeoutStrategy } from "./types";
 
-export class V3VisibilityTimeout implements VisibilityTimeoutStrategy {
+export class V3GraphileVisibilityTimeout implements VisibilityTimeoutStrategy {
   async heartbeat(messageId: string, timeoutInMs: number): Promise<void> {
-    await RequeueTaskRunService.enqueue(messageId, new Date(Date.now() + timeoutInMs));
+    await TaskRunHeartbeatFailedService.enqueue(messageId, new Date(Date.now() + timeoutInMs));
   }
 
   async cancelHeartbeat(messageId: string): Promise<void> {
-    await RequeueTaskRunService.dequeue(messageId);
+    await TaskRunHeartbeatFailedService.dequeue(messageId);
+  }
+}
+
+export class V3LegacyRunEngineWorkerVisibilityTimeout implements VisibilityTimeoutStrategy {
+  async heartbeat(messageId: string, timeoutInMs: number): Promise<void> {
+    await legacyRunEngineWorker.enqueue({
+      id: `heartbeat:${messageId}`,
+      job: "runHeartbeat",
+      payload: { runId: messageId },
+      availableAt: new Date(Date.now() + timeoutInMs),
+    });
+  }
+
+  async cancelHeartbeat(messageId: string): Promise<void> {
+    await legacyRunEngineWorker.ack(`heartbeat:${messageId}`);
   }
 }

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

@@ -25,6 +25,8 @@ import {
   CreatableEventEnvironmentType,
 } from "./eventRepository.server";
 import { logger } from "~/services/logger.server";
+import { trace, Tracer } from "@opentelemetry/api";
+import { startSpan } from "./tracing.server";
 
 export type OTLPExporterConfig = {
   batchSize: number;
@@ -32,51 +34,63 @@ export type OTLPExporterConfig = {
 };
 
 class OTLPExporter {
+  private _tracer: Tracer;
+
   constructor(
     private readonly _eventRepository: EventRepository,
     private readonly _verbose: boolean
-  ) {}
+  ) {
+    this._tracer = trace.getTracer("otlp-exporter");
+  }
 
   async exportTraces(
     request: ExportTraceServiceRequest,
     immediate: boolean = false
   ): Promise<ExportTraceServiceResponse> {
-    this.#logExportTracesVerbose(request);
+    return await startSpan(this._tracer, "exportTraces", async (span) => {
+      this.#logExportTracesVerbose(request);
 
-    const events = this.#filterResourceSpans(request.resourceSpans).flatMap((resourceSpan) => {
-      return convertSpansToCreateableEvents(resourceSpan);
+      const events = this.#filterResourceSpans(request.resourceSpans).flatMap((resourceSpan) => {
+        return convertSpansToCreateableEvents(resourceSpan);
+      });
+
+      this.#logEventsVerbose(events);
+
+      span.setAttribute("event_count", events.length);
+
+      if (immediate) {
+        await this._eventRepository.insertManyImmediate(events);
+      } else {
+        await this._eventRepository.insertMany(events);
+      }
+
+      return ExportTraceServiceResponse.create();
     });
-
-    this.#logEventsVerbose(events);
-
-    if (immediate) {
-      await this._eventRepository.insertManyImmediate(events);
-    } else {
-      await this._eventRepository.insertMany(events);
-    }
-
-    return ExportTraceServiceResponse.create();
   }
 
   async exportLogs(
     request: ExportLogsServiceRequest,
     immediate: boolean = false
   ): Promise<ExportLogsServiceResponse> {
-    this.#logExportLogsVerbose(request);
+    return await startSpan(this._tracer, "exportLogs", async (span) => {
+      this.#logExportLogsVerbose(request);
 
-    const events = this.#filterResourceLogs(request.resourceLogs).flatMap((resourceLog) => {
-      return convertLogsToCreateableEvents(resourceLog);
+      const events = this.#filterResourceLogs(request.resourceLogs).flatMap((resourceLog) => {
+        return convertLogsToCreateableEvents(resourceLog);
+      });
+
+      this.#logEventsVerbose(events);
+
+      span.setAttribute("event_count", events.length);
+
+      if (immediate) {
+        await this._eventRepository.insertManyImmediate(events);
+      } else {
+        await this._eventRepository.insertMany(events);
+      }
+
+      return ExportLogsServiceResponse.create();
     });
-
-    this.#logEventsVerbose(events);
-
-    if (immediate) {
-      await this._eventRepository.insertManyImmediate(events);
-    } else {
-      await this._eventRepository.insertMany(events);
-    }
-
-    return ExportLogsServiceResponse.create();
   }
 
   #logEventsVerbose(events: CreatableEvent[]) {

apps/webapp/app/v3/services/alerts/deliverAlert.server.ts

@@ -9,7 +9,7 @@ import {
 import { TaskRunError, createJsonErrorObject } from "@trigger.dev/core/v3";
 import assertNever from "assert-never";
 import { subtle } from "crypto";
-import { Prisma, PrismaClientOrTransaction, prisma } from "~/db.server";
+import { Prisma, prisma, PrismaClientOrTransaction } from "~/db.server";
 import { env } from "~/env.server";
 import {
   OrgIntegrationRepository,
@@ -25,9 +25,12 @@ import { DeploymentPresenter } from "~/presenters/v3/DeploymentPresenter.server"
 import { sendAlertEmail } from "~/services/email.server";
 import { logger } from "~/services/logger.server";
 import { decryptSecret } from "~/services/secrets/secretStore.server";
-import { workerQueue } from "~/services/worker.server";
-import { BaseService } from "../baseService.server";
+import { commonWorker } from "~/v3/commonWorker.server";
 import { FINAL_ATTEMPT_STATUSES } from "~/v3/taskStatus";
+import { BaseService } from "../baseService.server";
+import { generateFriendlyId } from "~/v3/friendlyIdentifiers";
+import { ProjectAlertType } from "@trigger.dev/database";
+import { alertsRateLimiter } from "~/v3/alertsRateLimiter.server";
 
 type FoundAlert = Prisma.Result<
   typeof prisma.projectAlert,
@@ -1092,22 +1095,73 @@ export class DeliverAlertService extends BaseService {
     return text;
   }
 
-  static async enqueue(
-    alertId: string,
-    tx: PrismaClientOrTransaction,
-    options?: { runAt?: Date; queueName?: string }
+  static async enqueue(alertId: string, runAt?: Date) {
+    return await commonWorker.enqueue({
+      id: `alert:${alertId}`,
+      job: "v3.deliverAlert",
+      payload: { alertId },
+      availableAt: runAt,
+    });
+  }
+
+  static async createAndSendAlert(
+    {
+      channelId,
+      projectId,
+      environmentId,
+      alertType,
+      deploymentId,
+      taskRunId,
+    }: {
+      channelId: string;
+      projectId: string;
+      environmentId: string;
+      alertType: ProjectAlertType;
+      deploymentId?: string;
+      taskRunId?: string;
+    },
+    db: PrismaClientOrTransaction
   ) {
-    return await workerQueue.enqueue(
-      "v3.deliverAlert",
-      {
-        alertId,
-      },
-      {
-        tx,
-        runAt: options?.runAt,
-        jobKey: `deliverAlert:${alertId}`,
+    if (taskRunId) {
+      try {
+        const result = await alertsRateLimiter.check(channelId);
+
+        if (!result.allowed) {
+          logger.warn("[DeliverAlert] Rate limited", {
+            taskRunId,
+            environmentId,
+            alertType,
+            channelId,
+            result,
+          });
+
+          return;
+        }
+      } catch (error) {
+        logger.error("[DeliverAlert] Rate limiter error", {
+          taskRunId,
+          environmentId,
+          alertType,
+          channelId,
+          error,
+        });
       }
-    );
+    }
+
+    const alert = await db.projectAlert.create({
+      data: {
+        friendlyId: generateFriendlyId("alert"),
+        channelId,
+        projectId,
+        environmentId,
+        status: "PENDING",
+        type: alertType,
+        workerDeploymentId: deploymentId,
+        taskRunId,
+      },
+    });
+
+    await DeliverAlertService.enqueue(alert.id);
   }
 }
 

apps/webapp/app/v3/services/alerts/performDeploymentAlerts.server.ts

@@ -4,6 +4,7 @@ import { workerQueue } from "~/services/worker.server";
 import { generateFriendlyId } from "~/v3/friendlyIdentifiers";
 import { BaseService } from "../baseService.server";
 import { DeliverAlertService } from "./deliverAlert.server";
+import { commonWorker } from "~/v3/commonWorker.server";
 
 export class PerformDeploymentAlertsService extends BaseService {
   public async call(deploymentId: string) {
@@ -45,34 +46,24 @@ export class PerformDeploymentAlertsService extends BaseService {
     deployment: WorkerDeployment,
     alertType: ProjectAlertType
   ) {
-    await $transaction(this._prisma, "create and send deploy alert", async (tx) => {
-      const alert = await this._prisma.projectAlert.create({
-        data: {
-          friendlyId: generateFriendlyId("alert"),
-          channelId: alertChannel.id,
-          projectId: deployment.projectId,
-          environmentId: deployment.environmentId,
-          status: "PENDING",
-          type: alertType,
-          workerDeploymentId: deployment.id,
-        },
-      });
-
-      await DeliverAlertService.enqueue(alert.id, tx);
-    });
-  }
-
-  static async enqueue(deploymentId: string, tx: PrismaClientOrTransaction, runAt?: Date) {
-    return await workerQueue.enqueue(
-      "v3.performDeploymentAlerts",
+    await DeliverAlertService.createAndSendAlert(
       {
-        deploymentId,
+        channelId: alertChannel.id,
+        projectId: deployment.projectId,
+        environmentId: deployment.environmentId,
+        alertType,
+        deploymentId: deployment.id,
       },
-      {
-        tx,
-        runAt,
-        jobKey: `performDeploymentAlerts:${deploymentId}`,
-      }
+      this._prisma
     );
   }
+
+  static async enqueue(deploymentId: string, runAt?: Date) {
+    return await commonWorker.enqueue({
+      id: `performDeploymentAlerts:${deploymentId}`,
+      job: "v3.performDeploymentAlerts",
+      payload: { deploymentId },
+      availableAt: runAt,
+    });
+  }
 }

apps/webapp/app/v3/services/alerts/performTaskAttemptAlerts.server.ts

@@ -1,79 +0,0 @@
-import { Prisma, ProjectAlertChannel } from "@trigger.dev/database";
-import { $transaction, PrismaClientOrTransaction, prisma } from "~/db.server";
-import { workerQueue } from "~/services/worker.server";
-import { generateFriendlyId } from "~/v3/friendlyIdentifiers";
-import { BaseService } from "../baseService.server";
-import { DeliverAlertService } from "./deliverAlert.server";
-
-type FoundTaskAttempt = Prisma.Result<
-  typeof prisma.taskRunAttempt,
-  { include: { taskRun: true; backgroundWorkerTask: true; runtimeEnvironment: true } },
-  "findUniqueOrThrow"
->;
-
-export class PerformTaskAttemptAlertsService extends BaseService {
-  public async call(attemptId: string) {
-    const taskAttempt = await this._prisma.taskRunAttempt.findFirst({
-      where: { id: attemptId },
-      include: {
-        taskRun: true,
-        backgroundWorkerTask: true,
-        runtimeEnvironment: true,
-      },
-    });
-
-    if (!taskAttempt) {
-      return;
-    }
-
-    // Find all the alert channels
-    const alertChannels = await this._prisma.projectAlertChannel.findMany({
-      where: {
-        projectId: taskAttempt.taskRun.projectId,
-        alertTypes: {
-          has: "TASK_RUN_ATTEMPT",
-        },
-        environmentTypes: {
-          has: taskAttempt.runtimeEnvironment.type,
-        },
-        enabled: true,
-      },
-    });
-
-    for (const alertChannel of alertChannels) {
-      await this.#createAndSendAlert(alertChannel, taskAttempt);
-    }
-  }
-
-  async #createAndSendAlert(alertChannel: ProjectAlertChannel, taskAttempt: FoundTaskAttempt) {
-    await $transaction(this._prisma, "create and send attempt alert", async (tx) => {
-      const alert = await this._prisma.projectAlert.create({
-        data: {
-          friendlyId: generateFriendlyId("alert"),
-          channelId: alertChannel.id,
-          projectId: taskAttempt.taskRun.projectId,
-          environmentId: taskAttempt.runtimeEnvironmentId,
-          status: "PENDING",
-          type: "TASK_RUN_ATTEMPT",
-          taskRunAttemptId: taskAttempt.id,
-        },
-      });
-
-      await DeliverAlertService.enqueue(alert.id, tx);
-    });
-  }
-
-  static async enqueue(attemptId: string, tx: PrismaClientOrTransaction, runAt?: Date) {
-    return await workerQueue.enqueue(
-      "v3.performTaskAttemptAlerts",
-      {
-        attemptId,
-      },
-      {
-        tx,
-        runAt,
-        jobKey: `performTaskAttemptAlerts:${attemptId}`,
-      }
-    );
-  }
-}

apps/webapp/app/v3/services/alerts/performTaskRunAlerts.server.ts

@@ -4,6 +4,7 @@ import { workerQueue } from "~/services/worker.server";
 import { generateFriendlyId } from "~/v3/friendlyIdentifiers";
 import { BaseService } from "../baseService.server";
 import { DeliverAlertService } from "./deliverAlert.server";
+import { commonWorker } from "~/v3/commonWorker.server";
 
 type FoundRun = Prisma.Result<
   typeof prisma.taskRun,
@@ -45,34 +46,24 @@ export class PerformTaskRunAlertsService extends BaseService {
   }
 
   async #createAndSendAlert(alertChannel: ProjectAlertChannel, run: FoundRun) {
-    await $transaction(this._prisma, "create and send run alert", async (tx) => {
-      const alert = await this._prisma.projectAlert.create({
-        data: {
-          friendlyId: generateFriendlyId("alert"),
-          channelId: alertChannel.id,
-          projectId: run.projectId,
-          environmentId: run.runtimeEnvironmentId,
-          status: "PENDING",
-          type: "TASK_RUN",
-          taskRunId: run.id,
-        },
-      });
-
-      await DeliverAlertService.enqueue(alert.id, tx);
-    });
-  }
-
-  static async enqueue(runId: string, tx: PrismaClientOrTransaction, runAt?: Date) {
-    return await workerQueue.enqueue(
-      "v3.performTaskRunAlerts",
+    await DeliverAlertService.createAndSendAlert(
       {
-        runId,
+        channelId: alertChannel.id,
+        projectId: run.projectId,
+        environmentId: run.runtimeEnvironmentId,
+        alertType: "TASK_RUN",
+        taskRunId: run.id,
       },
-      {
-        tx,
-        runAt,
-        jobKey: `performTaskRunAlerts:${runId}`,
-      }
+      this._prisma
     );
   }
+
+  static async enqueue(runId: string, runAt?: Date) {
+    return await commonWorker.enqueue({
+      id: `performTaskRunAlerts:${runId}`,
+      job: "v3.performTaskRunAlerts",
+      payload: { runId },
+      availableAt: runAt,
+    });
+  }
 }

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

@@ -7,6 +7,8 @@ import {
 } from "@trigger.dev/core/v3";
 import {
   BatchTaskRun,
+  isPrismaRaceConditionError,
+  isPrismaRetriableError,
   isUniqueConstraintError,
   Prisma,
   TaskRunAttempt,
@@ -20,6 +22,7 @@ import { logger } from "~/services/logger.server";
 import { getEntitlement } from "~/services/platform.v3.server";
 import { workerQueue } from "~/services/worker.server";
 import { generateFriendlyId } from "../friendlyIdentifiers";
+import { legacyRunEngineWorker } from "../legacyRunEngineWorker.server";
 import { marqs } from "../marqs/index.server";
 import { guardQueueSizeLimitsForEnv } from "../queueSizeLimits.server";
 import { downloadPacketFromObjectStore, uploadPacketToObjectStore } from "../r2.server";
@@ -91,14 +94,17 @@ type RunItemData = {
  */
 export class BatchTriggerV3Service extends BaseService {
   private _batchProcessingStrategy: BatchProcessingStrategy;
+  private _asyncBatchProcessSizeThreshold: number;
 
   constructor(
     batchProcessingStrategy?: BatchProcessingStrategy,
+    asyncBatchProcessSizeThreshold: number = ASYNC_BATCH_PROCESS_SIZE_THRESHOLD,
     protected readonly _prisma: PrismaClientOrTransaction = prisma
   ) {
     super(_prisma);
 
     this._batchProcessingStrategy = batchProcessingStrategy ?? "parallel";
+    this._asyncBatchProcessSizeThreshold = asyncBatchProcessSizeThreshold;
   }
 
   public async call(
@@ -323,24 +329,40 @@ export class BatchTriggerV3Service extends BaseService {
       }));
     }
 
-    const idempotencyKeys = body.items.map((i) => i.options?.idempotencyKey).filter(Boolean);
+    // Group items by taskIdentifier
+    const itemsByTask = body.items.reduce((acc, item) => {
+      if (!item.options?.idempotencyKey) return acc;
 
-    const cachedRuns =
-      idempotencyKeys.length > 0
-        ? await this._prisma.taskRun.findMany({
-            where: {
-              runtimeEnvironmentId: environment.id,
-              idempotencyKey: {
-                in: body.items.map((i) => i.options?.idempotencyKey).filter(Boolean),
-              },
+      if (!acc[item.task]) {
+        acc[item.task] = [];
+      }
+      acc[item.task].push(item);
+      return acc;
+    }, {} as Record<string, typeof body.items>);
+
+    logger.debug("[BatchTriggerV2][call] Grouped items by task identifier", {
+      itemsByTask,
+    });
+
+    // Fetch cached runs for each task identifier separately to make use of the index
+    const cachedRuns = await Promise.all(
+      Object.entries(itemsByTask).map(([taskIdentifier, items]) =>
+        this._prisma.taskRun.findMany({
+          where: {
+            runtimeEnvironmentId: environment.id,
+            taskIdentifier,
+            idempotencyKey: {
+              in: items.map((i) => i.options?.idempotencyKey).filter(Boolean),
             },
-            select: {
-              friendlyId: true,
-              idempotencyKey: true,
-              idempotencyKeyExpiresAt: true,
-            },
-          })
-        : [];
+          },
+          select: {
+            friendlyId: true,
+            idempotencyKey: true,
+            idempotencyKeyExpiresAt: true,
+          },
+        })
+      )
+    ).then((results) => results.flat());
 
     // Now we need to create an array of all the run IDs, in order
     // If we have a cached run, that isn't expired, we should use that run ID
@@ -400,7 +422,7 @@ export class BatchTriggerV3Service extends BaseService {
     options: BatchTriggerTaskServiceOptions = {},
     dependentAttempt?: TaskRunAttempt
   ) {
-    if (runs.length <= ASYNC_BATCH_PROCESS_SIZE_THRESHOLD) {
+    if (runs.length <= this._asyncBatchProcessSizeThreshold) {
       const batch = await this._prisma.batchTaskRun.create({
         data: {
           friendlyId: batchId,
@@ -923,71 +945,123 @@ export async function completeBatchTaskRunItemV3(
   batchTaskRunId: string,
   tx: PrismaClientOrTransaction,
   scheduleResumeOnComplete = false,
-  taskRunAttemptId?: string
+  taskRunAttemptId?: string,
+  retryAttempt?: number
 ) {
-  await $transaction(
-    tx,
-    "completeBatchTaskRunItemV3",
-    async (tx, span) => {
-      span?.setAttribute("batch_id", batchTaskRunId);
+  const isRetry = retryAttempt !== undefined;
 
-      // Update the item to complete
-      const updated = await tx.batchTaskRunItem.updateMany({
-        where: {
-          id: itemId,
-          status: "PENDING",
-        },
-        data: {
-          status: "COMPLETED",
-          taskRunAttemptId,
-        },
-      });
+  if (isRetry) {
+    logger.debug("completeBatchTaskRunItemV3 retrying", {
+      itemId,
+      batchTaskRunId,
+      scheduleResumeOnComplete,
+      taskRunAttemptId,
+      retryAttempt,
+    });
+  }
 
-      if (updated.count === 0) {
-        return;
-      }
+  try {
+    await $transaction(
+      tx,
+      "completeBatchTaskRunItemV3",
+      async (tx, span) => {
+        span?.setAttribute("batch_id", batchTaskRunId);
 
-      const updatedBatchRun = await tx.batchTaskRun.update({
-        where: {
-          id: batchTaskRunId,
-        },
-        data: {
-          completedCount: {
-            increment: 1,
+        // Update the item to complete
+        const updated = await tx.batchTaskRunItem.updateMany({
+          where: {
+            id: itemId,
+            status: "PENDING",
           },
-        },
-        select: {
-          sealed: true,
-          status: true,
-          completedCount: true,
-          expectedCount: true,
-          dependentTaskAttemptId: true,
-        },
-      });
+          data: {
+            status: "COMPLETED",
+            taskRunAttemptId,
+          },
+        });
 
-      if (
-        updatedBatchRun.status === "PENDING" &&
-        updatedBatchRun.completedCount === updatedBatchRun.expectedCount &&
-        updatedBatchRun.sealed
-      ) {
-        await tx.batchTaskRun.update({
+        if (updated.count === 0) {
+          return;
+        }
+
+        const updatedBatchRun = await tx.batchTaskRun.update({
           where: {
             id: batchTaskRunId,
           },
           data: {
-            status: "COMPLETED",
-            completedAt: new Date(),
+            completedCount: {
+              increment: 1,
+            },
+          },
+          select: {
+            sealed: true,
+            status: true,
+            completedCount: true,
+            expectedCount: true,
+            dependentTaskAttemptId: true,
           },
         });
 
-        // We only need to resume the batch if it has a dependent task attempt ID
-        if (scheduleResumeOnComplete && updatedBatchRun.dependentTaskAttemptId) {
-          await ResumeBatchRunService.enqueue(batchTaskRunId, true, tx);
+        if (
+          updatedBatchRun.status === "PENDING" &&
+          updatedBatchRun.completedCount === updatedBatchRun.expectedCount &&
+          updatedBatchRun.sealed
+        ) {
+          await tx.batchTaskRun.update({
+            where: {
+              id: batchTaskRunId,
+            },
+            data: {
+              status: "COMPLETED",
+              completedAt: new Date(),
+            },
+          });
+
+          // We only need to resume the batch if it has a dependent task attempt ID
+          if (scheduleResumeOnComplete && updatedBatchRun.dependentTaskAttemptId) {
+            await ResumeBatchRunService.enqueue(batchTaskRunId, true, tx);
+          }
         }
+      },
+      {
+        timeout: 10_000,
+        maxWait: 4_000,
       }
-    },
-    {
-      timeout: 10000,
+    );
+  } catch (error) {
+    if (isPrismaRetriableError(error) || isPrismaRaceConditionError(error)) {
+      logger.error("completeBatchTaskRunItemV3 failed with a Prisma Error, scheduling a retry", {
+        itemId,
+        batchTaskRunId,
+        error,
+        retryAttempt,
+        isRetry,
+      });
+
+      if (isRetry) {
+        //throwing this error will cause the Redis worker to retry the job
+        throw error;
+      } else {
+        //schedule a retry
+        await legacyRunEngineWorker.enqueue({
+          id: `completeBatchTaskRunItem:${itemId}`,
+          job: "completeBatchTaskRunItem",
+          payload: {
+            itemId,
+            batchTaskRunId,
+            scheduleResumeOnComplete,
+            taskRunAttemptId,
+          },
+          availableAt: new Date(Date.now() + 2_000),
+        });
+      }
+    } else {
+      logger.error("completeBatchTaskRunItemV3 failed with a non-retriable error", {
+        itemId,
+        batchTaskRunId,
+        error,
+        retryAttempt,
+        isRetry,
+      });
     }
-  );
+  }
 }

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

@@ -1,13 +1,16 @@
 import { Attributes } from "@opentelemetry/api";
 import {
   TaskRunContext,
+  TaskRunError,
   TaskRunErrorCodes,
   TaskRunExecution,
   TaskRunExecutionResult,
   TaskRunExecutionRetry,
   TaskRunFailedExecutionResult,
   TaskRunSuccessfulExecutionResult,
+  exceptionEventEnhancer,
   flattenAttributes,
+  internalErrorFromUnexpectedExit,
   sanitizeError,
   shouldRetryError,
   taskRunErrorEnhancer,
@@ -233,7 +236,7 @@ export class CompleteAttemptService extends BaseService {
 
     if (!executionRetry && shouldInfer) {
       executionRetryInferred = true;
-      executionRetry = await FailedTaskRunRetryHelper.getExecutionRetry({
+      executionRetry = FailedTaskRunRetryHelper.getExecutionRetry({
         run: {
           ...taskRunAttempt.taskRun,
           lockedBy: taskRunAttempt.backgroundWorkerTask,
@@ -243,7 +246,47 @@ export class CompleteAttemptService extends BaseService {
       });
     }
 
-    const retriableError = shouldRetryError(taskRunErrorEnhancer(completion.error));
+    let retriableError = shouldRetryError(taskRunErrorEnhancer(completion.error));
+    let isOOMRetry = false;
+
+    //OOM errors should retry (if an OOM machine is specified)
+    if (isOOMError(completion.error)) {
+      const retryConfig = FailedTaskRunRetryHelper.getRetryConfig({
+        run: {
+          ...taskRunAttempt.taskRun,
+          lockedBy: taskRunAttempt.backgroundWorkerTask,
+          lockedToVersion: taskRunAttempt.backgroundWorker,
+        },
+        execution,
+      });
+
+      if (
+        retryConfig?.outOfMemory?.machine &&
+        retryConfig.outOfMemory.machine !== taskRunAttempt.taskRun.machinePreset
+      ) {
+        //we will retry
+        isOOMRetry = true;
+        retriableError = true;
+        executionRetry = FailedTaskRunRetryHelper.getExecutionRetry({
+          run: {
+            ...taskRunAttempt.taskRun,
+            lockedBy: taskRunAttempt.backgroundWorkerTask,
+            lockedToVersion: taskRunAttempt.backgroundWorker,
+          },
+          execution,
+        });
+
+        //update the machine on the run
+        await this._prisma.taskRun.update({
+          where: {
+            id: taskRunAttempt.taskRunId,
+          },
+          data: {
+            machinePreset: retryConfig.outOfMemory.machine,
+          },
+        });
+      }
+    }
 
     if (
       retriableError &&
@@ -257,6 +300,7 @@ export class CompleteAttemptService extends BaseService {
         taskRunAttempt,
         environment,
         checkpoint,
+        forceRequeue: isOOMRetry,
       });
     }
 
@@ -378,12 +422,14 @@ export class CompleteAttemptService extends BaseService {
     executionRetryInferred,
     checkpointEventId,
     supportsLazyAttempts,
+    forceRequeue = false,
   }: {
     run: TaskRun;
     executionRetry: TaskRunExecutionRetry;
     executionRetryInferred: boolean;
     checkpointEventId?: string;
     supportsLazyAttempts: boolean;
+    forceRequeue?: boolean;
   }) {
     const retryViaQueue = () => {
       logger.debug("[CompleteAttemptService] Enqueuing retry attempt", { runId: run.id });
@@ -434,6 +480,12 @@ export class CompleteAttemptService extends BaseService {
       return;
     }
 
+    if (forceRequeue) {
+      logger.debug("[CompleteAttemptService] Forcing retry via queue", { runId: run.id });
+      await retryViaQueue();
+      return;
+    }
+
     // Workers that never checkpoint between attempts will exit after completing their current attempt if the retry delay exceeds the threshold
     if (
       !this.opts.supportsRetryCheckpoints &&
@@ -466,6 +518,7 @@ export class CompleteAttemptService extends BaseService {
     taskRunAttempt,
     environment,
     checkpoint,
+    forceRequeue = false,
   }: {
     execution: TaskRunExecution;
     executionRetry: TaskRunExecutionRetry;
@@ -473,6 +526,7 @@ export class CompleteAttemptService extends BaseService {
     taskRunAttempt: NonNullable<FoundAttempt>;
     environment: AuthenticatedEnvironment;
     checkpoint?: CheckpointData;
+    forceRequeue?: boolean;
   }) {
     const retryAt = new Date(executionRetry.timestamp);
 
@@ -533,6 +587,7 @@ export class CompleteAttemptService extends BaseService {
       executionRetry,
       supportsLazyAttempts: taskRunAttempt.backgroundWorker.supportsLazyAttempts,
       executionRetryInferred,
+      forceRequeue,
     });
 
     return "RETRIED";
@@ -634,3 +689,24 @@ async function findAttempt(prismaClient: PrismaClientOrTransaction, friendlyId:
     },
   });
 }
+
+function isOOMError(error: TaskRunError) {
+  if (error.type !== "INTERNAL_ERROR") return false;
+  if (error.code === "TASK_PROCESS_OOM_KILLED" || error.code === "TASK_PROCESS_MAYBE_OOM_KILLED") {
+    return true;
+  }
+
+  // For the purposes of retrying on a larger machine, we're going to treat this is an OOM error.
+  // This is what they look like if we're executing using k8s. They then get corrected later, but it's too late.
+  // {"code": "TASK_PROCESS_EXITED_WITH_NON_ZERO_CODE", "type": "INTERNAL_ERROR", "message": "Process exited with code -1 after signal SIGKILL."}
+  if (
+    error.code === "TASK_PROCESS_EXITED_WITH_NON_ZERO_CODE" &&
+    error.message &&
+    error.message.includes("SIGKILL") &&
+    error.message.includes("-1")
+  ) {
+    return true;
+  }
+
+  return false;
+}

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

@@ -148,7 +148,7 @@ export class CreateDeployedBackgroundWorkerService extends BaseService {
       }
 
       await ExecuteTasksWaitingForDeployService.enqueue(backgroundWorker.id, this._prisma);
-      await PerformDeploymentAlertsService.enqueue(deployment.id, this._prisma);
+      await PerformDeploymentAlertsService.enqueue(deployment.id);
       await TimeoutDeploymentService.dequeue(deployment.id, this._prisma);
 
       return backgroundWorker;

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

@@ -156,7 +156,7 @@ export class CreateTaskRunAttemptService extends BaseService {
           });
 
           if (taskRun.ttl) {
-            await ExpireEnqueuedRunService.dequeue(taskRun.id, tx);
+            await ExpireEnqueuedRunService.ack(taskRun.id, tx);
           }
         }
 

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

@@ -43,7 +43,7 @@ export class DeleteTaskScheduleService extends BaseService {
 
       await this._prisma.taskSchedule.delete({
         where: {
-          friendlyId,
+          id: schedule.id,
         },
       });
     } catch (e) {

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

@@ -66,7 +66,7 @@ export class DeploymentIndexFailed extends BaseService {
       },
     });
 
-    await PerformDeploymentAlertsService.enqueue(failedDeployment.id, this._prisma);
+    await PerformDeploymentAlertsService.enqueue(failedDeployment.id);
 
     return failedDeployment;
   }

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

@@ -4,8 +4,34 @@ import { logger } from "~/services/logger.server";
 import { marqs } from "~/v3/marqs/index.server";
 import { BaseService } from "./baseService.server";
 import { ExpireEnqueuedRunService } from "./expireEnqueuedRun.server";
+import { commonWorker } from "../commonWorker.server";
+import { workerQueue } from "~/services/worker.server";
 
 export class EnqueueDelayedRunService extends BaseService {
+  public static async enqueue(runId: string, runAt?: Date) {
+    await commonWorker.enqueue({
+      job: "v3.enqueueDelayedRun",
+      payload: { runId },
+      availableAt: runAt,
+      id: `v3.enqueueDelayed:${runId}`,
+    });
+  }
+
+  public static async reschedule(runId: string, runAt?: Date) {
+    // We have to do this for now because it's possible that the workerQueue
+    // was used when the run was first delayed, and EnqueueDelayedRunService.reschedule
+    // is called from RescheduleTaskRunService, which allows the runAt to be changed
+    // so if we don't dequeue the old job, we might end up with multiple jobs
+    await workerQueue.dequeue(`v3.enqueueDelayedRun.${runId}`);
+
+    await commonWorker.enqueue({
+      job: "v3.enqueueDelayedRun",
+      payload: { runId },
+      availableAt: runAt,
+      id: `v3.enqueueDelayed:${runId}`,
+    });
+  }
+
   public async call(runId: string) {
     const run = await this._prisma.taskRun.findFirst({
       where: {
@@ -52,7 +78,7 @@ export class EnqueueDelayedRunService extends BaseService {
         const expireAt = parseNaturalLanguageDuration(run.ttl);
 
         if (expireAt) {
-          await ExpireEnqueuedRunService.enqueue(run.id, expireAt, tx);
+          await ExpireEnqueuedRunService.enqueue(run.id, expireAt);
         }
       }
     });

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

@@ -1,21 +1,24 @@
-import { logger } from "~/services/logger.server";
-import { BaseService } from "./baseService.server";
-import { eventRepository } from "../eventRepository.server";
-import { FinalizeTaskRunService } from "./finalizeTaskRun.server";
-import { workerQueue } from "~/services/worker.server";
 import { PrismaClientOrTransaction } from "~/db.server";
+import { logger } from "~/services/logger.server";
+import { commonWorker } from "../commonWorker.server";
+import { eventRepository } from "../eventRepository.server";
+import { BaseService } from "./baseService.server";
+import { FinalizeTaskRunService } from "./finalizeTaskRun.server";
 
 export class ExpireEnqueuedRunService extends BaseService {
-  public static async dequeue(runId: string, tx?: PrismaClientOrTransaction) {
-    return await workerQueue.dequeue(`v3.expireRun:${runId}`, { tx });
+  public static async ack(runId: string, tx?: PrismaClientOrTransaction) {
+    // We don't "dequeue" from the workerQueue here because it would be redundant and if this service
+    // is called for a run that has already started, nothing happens
+    await commonWorker.ack(`v3.expireRun:${runId}`);
   }
 
-  public static async enqueue(runId: string, runAt?: Date, tx?: PrismaClientOrTransaction) {
-    return await workerQueue.enqueue(
-      "v3.expireRun",
-      { runId },
-      { runAt, jobKey: `v3.expireRun:${runId}`, tx }
-    );
+  public static async enqueue(runId: string, runAt?: Date) {
+    return await commonWorker.enqueue({
+      job: "v3.expireRun",
+      payload: { runId },
+      availableAt: runAt,
+      id: `v3.expireRun:${runId}`,
+    });
   }
 
   public async call(runId: string) {

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

@@ -49,7 +49,7 @@ export class FailDeploymentService extends BaseService {
       },
     });
 
-    await PerformDeploymentAlertsService.enqueue(failedDeployment.id, this._prisma);
+    await PerformDeploymentAlertsService.enqueue(failedDeployment.id);
 
     return failedDeployment;
   }

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

@@ -124,7 +124,7 @@ export class FinalizeDeploymentService extends BaseService {
     }
 
     await ExecuteTasksWaitingForDeployService.enqueue(deployment.worker.id, this._prisma);
-    await PerformDeploymentAlertsService.enqueue(deployment.id, this._prisma);
+    await PerformDeploymentAlertsService.enqueue(deployment.id);
 
     return finalizedDeployment;
   }

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

@@ -28,11 +28,14 @@ export class FinalizeDeploymentV2Service extends BaseService {
         friendlyId: id,
         environmentId: authenticatedEnv.id,
       },
-      include: {
+      select: {
+        status: true,
+        id: true,
+        version: true,
+        externalBuildData: true,
         environment: true,
         worker: {
-          include: {
-            tasks: true,
+          select: {
             project: true,
           },
         },
@@ -84,6 +87,14 @@ export class FinalizeDeploymentV2Service extends BaseService {
       throw new ServiceValidationError("Missing depot token");
     }
 
+    const digest = extractImageDigest(body.imageReference);
+
+    logger.debug("Pushing image to registry", {
+      id,
+      deployment,
+      digest,
+    });
+
     const pushResult = await executePushToRegistry(
       {
         depot: {
@@ -110,10 +121,19 @@ export class FinalizeDeploymentV2Service extends BaseService {
       throw new ServiceValidationError(pushResult.error);
     }
 
+    const fullImage = digest ? `${pushResult.image}@${digest}` : pushResult.image;
+
+    logger.debug("Image pushed to registry", {
+      id,
+      deployment,
+      body,
+      fullImage,
+    });
+
     const finalizeService = new FinalizeDeploymentService();
 
     const finalizedDeployment = await finalizeService.call(authenticatedEnv, id, {
-      imageReference: pushResult.image,
+      imageReference: fullImage,
       skipRegistryProxy: true,
     });
 
@@ -121,6 +141,19 @@ export class FinalizeDeploymentV2Service extends BaseService {
   }
 }
 
+// Extracts the sha256 digest from an image reference
+// For example the image ref "registry.depot.dev/gn57tl6chn:8qfjm8w83w@sha256:aa6fd2bdcbbd611556747e72d0b57797f03aa9b39dc910befc83eea2b08a5b85"
+// would return "sha256:aa6fd2bdcbbd611556747e72d0b57797f03aa9b39dc910befc83eea2b08a5b85"
+function extractImageDigest(image: string) {
+  const digestIndex = image.lastIndexOf("@");
+
+  if (digestIndex === -1) {
+    return;
+  }
+
+  return image.substring(digestIndex + 1);
+}
+
 type ExecutePushToRegistryOptions = {
   depot: {
     buildId: string;

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

@@ -101,7 +101,7 @@ export class FinalizeTaskRunService extends BaseService {
     });
 
     if (run.ttl) {
-      await ExpireEnqueuedRunService.dequeue(run.id);
+      await ExpireEnqueuedRunService.ack(run.id);
     }
 
     if (attemptStatus || error) {
@@ -122,14 +122,17 @@ export class FinalizeTaskRunService extends BaseService {
     const result = await resumeService.call({ id: run.id });
 
     if (result.success) {
-      logger.log("FinalizeTaskRunService: Resumed dependent parents", { result, run });
+      logger.log("FinalizeTaskRunService: Resumed dependent parents", { result, run: run.id });
     } else {
-      logger.error("FinalizeTaskRunService: Failed to resume dependent parents", { result, run });
+      logger.error("FinalizeTaskRunService: Failed to resume dependent parents", {
+        result,
+        run: run.id,
+      });
     }
 
     //enqueue alert
     if (isFailedRunStatus(run.status)) {
-      await PerformTaskRunAlertsService.enqueue(run.id, this._prisma);
+      await PerformTaskRunAlertsService.enqueue(run.id);
     }
 
     if (isFatalRunStatus(run.status)) {

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

@@ -1,9 +1,8 @@
+import { RescheduleRunRequestBody } from "@trigger.dev/core/v3";
 import { TaskRun } from "@trigger.dev/database";
 import { BaseService, ServiceValidationError } from "./baseService.server";
-import { RescheduleRunRequestBody } from "@trigger.dev/core/v3";
+import { EnqueueDelayedRunService } from "./enqueueDelayedRun.server";
 import { parseDelay } from "./triggerTask.server";
-import { $transaction } from "~/db.server";
-import { workerQueue } from "~/services/worker.server";
 
 export class RescheduleTaskRunService extends BaseService {
   public async call(taskRun: TaskRun, body: RescheduleRunRequestBody) {
@@ -17,23 +16,17 @@ export class RescheduleTaskRunService extends BaseService {
       throw new ServiceValidationError(`Invalid delay: ${body.delay}`);
     }
 
-    return await $transaction(this._prisma, "reschedule run", async (tx) => {
-      const updatedRun = await tx.taskRun.update({
-        where: {
-          id: taskRun.id,
-        },
-        data: {
-          delayUntil: delay,
-        },
-      });
-
-      await workerQueue.enqueue(
-        "v3.enqueueDelayedRun",
-        { runId: taskRun.id },
-        { tx, runAt: delay, jobKey: `v3.enqueueDelayedRun.${taskRun.id}` }
-      );
-
-      return updatedRun;
+    const updatedRun = await this._prisma.taskRun.update({
+      where: {
+        id: taskRun.id,
+      },
+      data: {
+        delayUntil: delay,
+      },
     });
+
+    await EnqueueDelayedRunService.reschedule(taskRun.id, delay);
+
+    return updatedRun;
   }
 }

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

@@ -39,7 +39,7 @@ export class TimeoutDeploymentService extends BaseService {
       },
     });
 
-    await PerformDeploymentAlertsService.enqueue(deployment.id, this._prisma);
+    await PerformDeploymentAlertsService.enqueue(deployment.id);
   }
 
   static async enqueue(

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

@@ -32,6 +32,16 @@ export class TriggerScheduledTaskService extends BaseService {
       return;
     }
 
+    if (instance.environment.organization.deletedAt) {
+      logger.debug("Organization is deleted, disabling schedule", {
+        instanceId,
+        scheduleId: instance.taskSchedule.friendlyId,
+        organizationId: instance.environment.organization.id,
+      });
+
+      return;
+    }
+
     try {
       let shouldTrigger = true;
 
@@ -40,23 +50,6 @@ export class TriggerScheduledTaskService extends BaseService {
       }
 
       if (!instance.taskSchedule.active) {
-        shouldTrigger = false;
-      } else if (instance.environment.organization.deletedAt) {
-        logger.debug("Organization is deleted, disabling schedule", {
-          instanceId,
-          scheduleId: instance.taskSchedule.friendlyId,
-          organizationId: instance.environment.organization.id,
-        });
-
-        await this._prisma.taskSchedule.update({
-          where: {
-            id: instance.taskSchedule.id,
-          },
-          data: {
-            active: false,
-          },
-        });
-
         shouldTrigger = false;
       }
 

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

@@ -28,6 +28,7 @@ import { clampMaxDuration } from "../utils/maxDuration";
 import { resolveIdempotencyKeyTTL } from "~/utils/idempotencyKeys.server";
 import { Prisma, TaskRun } from "@trigger.dev/database";
 import { sanitizeQueueName } from "~/models/taskQueue.server";
+import { EnqueueDelayedRunService } from "./enqueueDelayedRun.server";
 
 export type TriggerTaskServiceOptions = {
   idempotencyKey?: string;
@@ -515,18 +516,14 @@ export class TriggerTaskService extends BaseService {
                 }
 
                 if (taskRun.delayUntil) {
-                  await workerQueue.enqueue(
-                    "v3.enqueueDelayedRun",
-                    { runId: taskRun.id },
-                    { tx, runAt: delayUntil, jobKey: `v3.enqueueDelayedRun.${taskRun.id}` }
-                  );
+                  await EnqueueDelayedRunService.enqueue(taskRun.id, taskRun.delayUntil);
                 }
 
                 if (!taskRun.delayUntil && taskRun.ttl) {
                   const expireAt = parseNaturalLanguageDuration(taskRun.ttl);
 
                   if (expireAt) {
-                    await ExpireEnqueuedRunService.enqueue(taskRun.id, expireAt, tx);
+                    await ExpireEnqueuedRunService.enqueue(taskRun.id, expireAt);
                   }
                 }
 

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

@@ -9,7 +9,7 @@ import { workerQueue } from "~/services/worker.server";
 import { socketIo } from "./handleSocketIo.server";
 import { TaskRunErrorCodes } from "@trigger.dev/core/v3";
 
-export class RequeueTaskRunService extends BaseService {
+export class TaskRunHeartbeatFailedService extends BaseService {
   public async call(runId: string) {
     const taskRun = await this._prisma.taskRun.findFirst({
       where: {
@@ -30,27 +30,42 @@ export class RequeueTaskRunService extends BaseService {
             supportsLazyAttempts: true,
           },
         },
+        _count: {
+          select: {
+            attempts: true,
+          },
+        },
       },
     });
 
     if (!taskRun) {
-      logger.error("[RequeueTaskRunService] Task run not found", {
+      logger.error("[TaskRunHeartbeatFailedService] Task run not found", {
         runId,
       });
 
       return;
     }
 
+    const service = new FailedTaskRunService();
+
     switch (taskRun.status) {
-      case "PENDING": {
-        if (taskRun.lockedAt) {
+      case "PENDING":
+      case "WAITING_TO_RESUME":
+      case "PAUSED": {
+        const backInQueue = await marqs?.nackMessage(taskRun.id);
+
+        if (backInQueue) {
           logger.debug(
-            "[RequeueTaskRunService] Failing task run because the heartbeat failed and it's PENDING but locked",
+            `[TaskRunHeartbeatFailedService] ${taskRun.status} run is back in the queue run`,
+            {
+              taskRun,
+            }
+          );
+        } else {
+          logger.debug(
+            `[TaskRunHeartbeatFailedService] ${taskRun.status} run not back in the queue, failing`,
             { taskRun }
           );
-
-          const service = new FailedTaskRunService();
-
           await service.call(taskRun.friendlyId, {
             ok: false,
             id: taskRun.friendlyId,
@@ -61,19 +76,13 @@ export class RequeueTaskRunService extends BaseService {
               message: "Did not receive a heartbeat from the worker in time",
             },
           });
-        } else {
-          logger.debug("[RequeueTaskRunService] Nacking task run", { taskRun });
-
-          await marqs?.nackMessage(taskRun.id);
         }
 
         break;
       }
       case "EXECUTING":
       case "RETRYING_AFTER_FAILURE": {
-        logger.debug("[RequeueTaskRunService] Failing task run", { taskRun });
-
-        const service = new FailedTaskRunService();
+        logger.debug(`[RequeueTaskRunService] ${taskRun.status} failing task run`, { taskRun });
 
         await service.call(taskRun.friendlyId, {
           ok: false,
@@ -90,23 +99,18 @@ export class RequeueTaskRunService extends BaseService {
       }
       case "DELAYED":
       case "WAITING_FOR_DEPLOY": {
-        logger.debug("[RequeueTaskRunService] Removing task run from queue", { taskRun });
+        logger.debug(
+          `[TaskRunHeartbeatFailedService] ${taskRun.status} Removing task run from queue`,
+          { taskRun }
+        );
 
         await marqs?.acknowledgeMessage(
           taskRun.id,
-          "Run is either DELAYED or WAITING_FOR_DEPLOY so we cannot requeue it in RequeueTaskRunService"
+          "Run is either DELAYED or WAITING_FOR_DEPLOY so we cannot requeue it in TaskRunHeartbeatFailedService"
         );
 
         break;
       }
-      case "WAITING_TO_RESUME":
-      case "PAUSED": {
-        logger.debug("[RequeueTaskRunService] Requeueing task run", { taskRun });
-
-        await marqs?.nackMessage(taskRun.id);
-
-        break;
-      }
       case "SYSTEM_FAILURE":
       case "INTERRUPTED":
       case "CRASHED":
@@ -115,11 +119,11 @@ export class RequeueTaskRunService extends BaseService {
       case "EXPIRED":
       case "TIMED_OUT":
       case "CANCELED": {
-        logger.debug("[RequeueTaskRunService] Task run is completed", { taskRun });
+        logger.debug("[TaskRunHeartbeatFailedService] Task run is completed", { taskRun });
 
         await marqs?.acknowledgeMessage(
           taskRun.id,
-          "Task run is already completed in RequeueTaskRunService"
+          "Task run is already completed in TaskRunHeartbeatFailedService"
         );
 
         try {
@@ -135,7 +139,7 @@ export class RequeueTaskRunService extends BaseService {
             delayInMs: taskRun.lockedToVersion?.supportsLazyAttempts ? 5_000 : undefined,
           });
         } catch (error) {
-          logger.error("[RequeueTaskRunService] Error signaling run cancellation", {
+          logger.error("[TaskRunHeartbeatFailedService] Error signaling run cancellation", {
             runId: taskRun.id,
             error: error instanceof Error ? error.message : error,
           });

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

@@ -8,6 +8,7 @@ import {
   SpanKind,
   SpanOptions,
   SpanStatusCode,
+  Tracer,
   diag,
   trace,
 } from "@opentelemetry/api";

apps/webapp/package.json

@@ -51,6 +51,7 @@
     "@headlessui/react": "^1.7.8",
     "@heroicons/react": "^2.0.12",
     "@internal/zod-worker": "workspace:*",
+    "@internal/redis-worker": "workspace:*",
     "@internationalized/date": "^3.5.1",
     "@lezer/highlight": "^1.1.6",
     "@opentelemetry/api": "1.9.0",

apps/webapp/remix.config.js

@@ -11,6 +11,8 @@ module.exports = {
     /^remix-utils.*/,
     "marked",
     "axios",
+    "p-limit",
+    "yocto-queue",
     "@trigger.dev/core",
     "@trigger.dev/sdk",
     "@trigger.dev/platform",

apps/webapp/test/GCRARateLimiter.test.ts

@@ -0,0 +1,217 @@
+// GCRARateLimiter.test.ts
+import { redisTest } from "@internal/testcontainers";
+import { describe, expect, vi } from "vitest";
+import { GCRARateLimiter } from "../app/v3/GCRARateLimiter.server.js"; // adjust the import as needed
+
+// Extend the timeout to 30 seconds (as in your redis tests)
+vi.setConfig({ testTimeout: 30_000 });
+
+describe("GCRARateLimiter", () => {
+  redisTest("should allow a single request when under the rate limit", async ({ redis }) => {
+    const limiter = new GCRARateLimiter({
+      redis,
+      emissionInterval: 1000, // 1 request per second on average
+      burstTolerance: 3000, // Allows a burst of 4 requests (3 * 1000 + 1)
+      keyPrefix: "test:ratelimit:",
+    });
+
+    const result = await limiter.check("user:1");
+    expect(result.allowed).toBe(true);
+  });
+
+  redisTest(
+    "should allow bursts up to the configured limit and then reject further requests",
+    async ({ redis }) => {
+      const limiter = new GCRARateLimiter({
+        redis,
+        emissionInterval: 1000,
+        burstTolerance: 3000, // With an emission interval of 1000ms, burstTolerance of 3000ms allows 4 rapid requests.
+        keyPrefix: "test:ratelimit:",
+      });
+
+      // Call 4 times in rapid succession (all should be allowed)
+      const results = await Promise.all([
+        limiter.check("user:burst"),
+        limiter.check("user:burst"),
+        limiter.check("user:burst"),
+        limiter.check("user:burst"),
+      ]);
+      results.forEach((result) => expect(result.allowed).toBe(true));
+
+      // The 5th call should be rejected.
+      const fifthResult = await limiter.check("user:burst");
+      expect(fifthResult.allowed).toBe(false);
+      expect(fifthResult.retryAfter).toBeGreaterThan(0);
+    }
+  );
+
+  redisTest("should allow a request after the required waiting period", async ({ redis }) => {
+    const limiter = new GCRARateLimiter({
+      redis,
+      emissionInterval: 1000,
+      burstTolerance: 3000,
+      keyPrefix: "test:ratelimit:",
+    });
+
+    // Exhaust burst capacity with 4 rapid calls.
+    await limiter.check("user:wait");
+    await limiter.check("user:wait");
+    await limiter.check("user:wait");
+    await limiter.check("user:wait");
+
+    // The 5th call should be rejected.
+    const rejection = await limiter.check("user:wait");
+    expect(rejection.allowed).toBe(false);
+    expect(rejection.retryAfter).toBeGreaterThan(0);
+
+    // Wait for the period specified in retryAfter (plus a small buffer)
+    await new Promise((resolve) => setTimeout(resolve, rejection.retryAfter! + 50));
+
+    // Now the next call should be allowed.
+    const allowedAfterWait = await limiter.check("user:wait");
+    expect(allowedAfterWait.allowed).toBe(true);
+  });
+
+  redisTest("should rate limit independently for different identifiers", async ({ redis }) => {
+    const limiter = new GCRARateLimiter({
+      redis,
+      emissionInterval: 1000,
+      burstTolerance: 3000,
+      keyPrefix: "test:ratelimit:",
+    });
+
+    // For "user:independent", exhaust burst capacity.
+    await limiter.check("user:independent");
+    await limiter.check("user:independent");
+    await limiter.check("user:independent");
+    await limiter.check("user:independent");
+    const rejected = await limiter.check("user:independent");
+    expect(rejected.allowed).toBe(false);
+
+    // A different identifier should start fresh.
+    const fresh = await limiter.check("user:different");
+    expect(fresh.allowed).toBe(true);
+  });
+
+  redisTest("should gradually reduce retryAfter with time", async ({ redis }) => {
+    const limiter = new GCRARateLimiter({
+      redis,
+      emissionInterval: 1000,
+      burstTolerance: 3000,
+      keyPrefix: "test:ratelimit:",
+    });
+
+    // Exhaust the burst capacity.
+    await limiter.check("user:gradual");
+    await limiter.check("user:gradual");
+    await limiter.check("user:gradual");
+    await limiter.check("user:gradual");
+
+    const firstRejection = await limiter.check("user:gradual");
+    expect(firstRejection.allowed).toBe(false);
+    const firstRetry = firstRejection.retryAfter!;
+
+    // Wait 500ms, then perform another check.
+    await new Promise((resolve) => setTimeout(resolve, 500));
+    const secondRejection = await limiter.check("user:gradual");
+    // It should still be rejected but with a smaller wait time.
+    expect(secondRejection.allowed).toBe(false);
+    const secondRetry = secondRejection.retryAfter!;
+    expect(secondRetry).toBeLessThan(firstRetry);
+  });
+
+  redisTest("should expire the key after the TTL", async ({ redis }) => {
+    // For this test, override keyExpiration to a short value.
+    const keyExpiration = 1500; // 1.5 seconds TTL
+    const limiter = new GCRARateLimiter({
+      redis,
+      emissionInterval: 100,
+      burstTolerance: 300, // These values are arbitrary for this test.
+      keyPrefix: "test:expire:",
+      keyExpiration,
+    });
+    const identifier = "user:expire";
+
+    // Make a call to set the key.
+    const result = await limiter.check(identifier);
+    expect(result.allowed).toBe(true);
+
+    // Immediately verify the key exists.
+    const key = `test:expire:${identifier}`;
+    let stored = await redis.get(key);
+    expect(stored).not.toBeNull();
+
+    // Wait for longer than keyExpiration.
+    await new Promise((resolve) => setTimeout(resolve, keyExpiration + 200));
+    stored = await redis.get(key);
+    expect(stored).toBeNull();
+  });
+
+  redisTest("should not share state across different key prefixes", async ({ redis }) => {
+    const limiter1 = new GCRARateLimiter({
+      redis,
+      emissionInterval: 1000,
+      burstTolerance: 3000,
+      keyPrefix: "test:ratelimit1:",
+    });
+    const limiter2 = new GCRARateLimiter({
+      redis,
+      emissionInterval: 1000,
+      burstTolerance: 3000,
+      keyPrefix: "test:ratelimit2:",
+    });
+
+    // Exhaust the burst capacity for a given identifier in limiter1.
+    await limiter1.check("user:shared");
+    await limiter1.check("user:shared");
+    await limiter1.check("user:shared");
+    await limiter1.check("user:shared");
+    const rejection1 = await limiter1.check("user:shared");
+    expect(rejection1.allowed).toBe(false);
+
+    // With a different key prefix, the same identifier should be fresh.
+    const result2 = await limiter2.check("user:shared");
+    expect(result2.allowed).toBe(true);
+  });
+
+  redisTest("should increment TAT correctly on sequential allowed requests", async ({ redis }) => {
+    const limiter = new GCRARateLimiter({
+      redis,
+      emissionInterval: 1000,
+      burstTolerance: 3000,
+      keyPrefix: "test:ratelimit:",
+    });
+
+    // The first request should be allowed.
+    const r1 = await limiter.check("user:sequential");
+    expect(r1.allowed).toBe(true);
+
+    // Wait a bit longer than the emission interval.
+    await new Promise((resolve) => setTimeout(resolve, 1100));
+    const r2 = await limiter.check("user:sequential");
+    expect(r2.allowed).toBe(true);
+  });
+
+  redisTest("should throw an error if redis command fails", async ({ redis }) => {
+    const limiter = new GCRARateLimiter({
+      redis,
+      emissionInterval: 1000,
+      burstTolerance: 3000,
+      keyPrefix: "test:ratelimit:",
+    });
+
+    // Stub redis.gcra to simulate a failure.
+    // @ts-expect-error
+    const originalGcra = redis.gcra;
+    // @ts-ignore
+    redis.gcra = vi.fn(() => {
+      throw new Error("Simulated Redis error");
+    });
+
+    await expect(limiter.check("user:error")).rejects.toThrow("Simulated Redis error");
+
+    // Restore the original command.
+    // @ts-expect-error
+    redis.gcra = originalGcra;
+  });
+});

apps/webapp/tsconfig.json

@@ -34,7 +34,9 @@
       "emails": ["../../internal-packages/emails/src/index"],
       "emails/*": ["../../internal-packages/emails/src/*"],
       "@internal/zod-worker": ["../../internal-packages/zod-worker/src/index"],
-      "@internal/zod-worker/*": ["../../internal-packages/zod-worker/src/*"]
+      "@internal/zod-worker/*": ["../../internal-packages/zod-worker/src/*"],
+      "@internal/redis-worker": ["../../internal-packages/redis-worker/src/index"],
+      "@internal/redis-worker/*": ["../../internal-packages/redis-worker/src/*"]
     },
     "noEmit": true
   }

docs/machines.mdx

@@ -8,9 +8,7 @@ The `machine` configuration is optional. Using higher spec machines will increas
 ```ts /trigger/heavy-task.ts
 export const heavyTask = task({
   id: "heavy-task",
-  machine: {
-    preset: "large-1x",
-  },
+  machine: "large-1x",
   run: async ({ payload, ctx }) => {
     //...
   },
@@ -28,6 +26,37 @@ export const config: TriggerConfig = {
 };
 ```
 
+## Out Of Memory errors
+
+Sometimes you might see one of your runs fail with an "Out Of Memory" error.
+
+> TASK_PROCESS_OOM_KILLED. Your task ran out of memory. Try increasing the machine specs. If this doesn't fix it there might be a memory leak.
+
+If this happens regularly you need to either optimize the memory-efficiency of your code, or increase the machine.
+
+### Retrying with a larger machine
+
+If you are seeing rare OOM errors, you can add a setting to your task to retry with a large machine if you get an OOM error:
+
+```ts /trigger/heavy-task.ts
+export const yourTask = task({
+  id: "your-task",
+  machine: "medium-1x",
+  retry: {
+    outOfMemory: {
+      machine: "large-1x",
+    },
+  },
+  run: async (payload: any, { ctx }) => {
+    //...
+  },
+});
+```
+
+<Note>
+  This will only retry the task if you get an OOM error. It won't permanently change the machine that a new run starts on, so if you consistently see OOM errors you should change the machine in the `machine` property.
+</Note>
+
 ## Machine configurations
 
 | Preset              | vCPU | Memory | Disk space |

internal-packages/database/prisma/migrations/20250206170153_drop_task_schedule_foreign_key_constraints_on_task_run/migration.sql

@@ -0,0 +1,7 @@
+-- DropForeignKey
+ALTER TABLE
+  "TaskRun" DROP CONSTRAINT IF EXISTS "TaskRun_scheduleId_fkey";
+
+-- DropForeignKey
+ALTER TABLE
+  "TaskRun" DROP CONSTRAINT IF EXISTS "TaskRun_scheduleInstanceId_fkey";

internal-packages/database/prisma/schema.prisma

@@ -1735,11 +1735,8 @@ model TaskRun {
 
   alerts ProjectAlert[]
 
-  scheduleInstance   TaskScheduleInstance? @relation(fields: [scheduleInstanceId], references: [id], onDelete: SetNull)
   scheduleInstanceId String?
-
-  schedule   TaskSchedule? @relation(fields: [scheduleId], references: [id], onDelete: SetNull)
-  scheduleId String?
+  scheduleId         String?
 
   sourceBulkActionItems      BulkActionItem[] @relation("SourceActionItemRun")
   destinationBulkActionItems BulkActionItem[] @relation("DestinationActionItemRun")
@@ -2452,8 +2449,6 @@ model TaskSchedule {
 
   active Boolean @default(true)
 
-  runs TaskRun[]
-
   @@unique([projectId, deduplicationKey])
 }
 
@@ -2486,8 +2481,6 @@ model TaskScheduleInstance {
   lastScheduledTimestamp DateTime?
   nextScheduledTimestamp DateTime?
 
-  runs TaskRun[]
-
   //you can only have a schedule attached to each environment once
   @@unique([taskScheduleId, environmentId])
 }

internal-packages/database/src/transaction.ts

@@ -13,12 +13,38 @@ function isTransactionClient(prisma: PrismaClientOrTransaction): prisma is Prism
   return !("$transaction" in prisma);
 }
 
-function isPrismaKnownError(error: unknown): error is Prisma.PrismaClientKnownRequestError {
+export function isPrismaKnownError(error: unknown): error is Prisma.PrismaClientKnownRequestError {
   return (
     typeof error === "object" && error !== null && "code" in error && typeof error.code === "string"
   );
 }
 
+/*
+•	P2024: Connection timeout errors
+•	P2028: Transaction timeout errors
+•	P2034: Transaction deadlock/conflict errors
+*/
+const retryCodes = ["P2024", "P2028", "P2034"];
+
+export function isPrismaRetriableError(error: unknown): boolean {
+  if (!isPrismaKnownError(error)) {
+    return false;
+  }
+
+  return retryCodes.includes(error.code);
+}
+
+/*
+•	P2025: Record not found errors (in race conditions) [not included for now]
+*/
+export function isPrismaRaceConditionError(error: unknown): boolean {
+  if (!isPrismaKnownError(error)) {
+    return false;
+  }
+
+  return error.code === "P2025";
+}
+
 export type PrismaTransactionOptions = {
   /** The maximum amount of time (in ms) Prisma Client will wait to acquire a transaction from the database. The default value is 2000ms. */
   maxWait?: number;
@@ -55,7 +81,7 @@ export async function $transaction<R>(
   } catch (error) {
     if (isPrismaKnownError(error)) {
       if (
-        error.code === "P2034" &&
+        retryCodes.includes(error.code) &&
         typeof options?.maxRetries === "number" &&
         attempt < options.maxRetries
       ) {

internal-packages/redis-worker/package.json

@@ -11,7 +11,7 @@
     "ioredis": "^5.3.2",
     "lodash.omit": "^4.5.0",
     "nanoid": "^5.0.7",
-    "typescript": "^5.5.4",
+    "p-limit": "^6.2.0",
     "zod": "3.23.8"
   },
   "devDependencies": {
@@ -21,6 +21,6 @@
   },
   "scripts": {
     "typecheck": "tsc --noEmit",
-    "test": "vitest"
+    "test": "vitest --no-file-parallelism"
   }
 }

internal-packages/redis-worker/src/index.ts

@@ -0,0 +1,2 @@
+export * from "./queue";
+export * from "./worker";

internal-packages/redis-worker/src/queue.test.ts

@@ -30,13 +30,16 @@ describe("SimpleQueue", () => {
       expect(await queue.size()).toBe(2);
 
       const [first] = await queue.dequeue(1);
-      expect(first).toEqual({
-        id: "1",
-        job: "test",
-        item: { value: 1 },
-        visibilityTimeoutMs: 2000,
-        attempt: 0,
-      });
+      expect(first).toEqual(
+        expect.objectContaining({
+          id: "1",
+          job: "test",
+          item: { value: 1 },
+          visibilityTimeoutMs: 2000,
+          attempt: 0,
+          timestamp: expect.any(Date),
+        })
+      );
       expect(await queue.size()).toBe(1);
       expect(await queue.size({ includeFuture: true })).toBe(2);
 
@@ -44,13 +47,16 @@ describe("SimpleQueue", () => {
       expect(await queue.size({ includeFuture: true })).toBe(1);
 
       const [second] = await queue.dequeue(1);
-      expect(second).toEqual({
-        id: "2",
-        job: "test",
-        item: { value: 2 },
-        visibilityTimeoutMs: 2000,
-        attempt: 0,
-      });
+      expect(second).toEqual(
+        expect.objectContaining({
+          id: "2",
+          job: "test",
+          item: { value: 2 },
+          visibilityTimeoutMs: 2000,
+          attempt: 0,
+          timestamp: expect.any(Date),
+        })
+      );
 
       await queue.ack(second.id);
       expect(await queue.size({ includeFuture: true })).toBe(0);
@@ -81,13 +87,16 @@ describe("SimpleQueue", () => {
 
       await queue.enqueue({ id: "1", job: "test", item: { value: 1 }, visibilityTimeoutMs: 2000 });
       const [hitOne] = await queue.dequeue(1);
-      expect(hitOne).toEqual({
-        id: "1",
-        job: "test",
-        item: { value: 1 },
-        visibilityTimeoutMs: 2000,
-        attempt: 0,
-      });
+      expect(hitOne).toEqual(
+        expect.objectContaining({
+          id: "1",
+          job: "test",
+          item: { value: 1 },
+          visibilityTimeoutMs: 2000,
+          attempt: 0,
+          timestamp: expect.any(Date),
+        })
+      );
 
       const missTwo = await queue.dequeue(1);
       expect(missTwo).toEqual([]);
@@ -128,13 +137,16 @@ describe("SimpleQueue", () => {
       await new Promise((resolve) => setTimeout(resolve, 50));
 
       const [first] = await queue.dequeue();
-      expect(first).toEqual({
-        id: "1",
-        job: "test",
-        item: { value: 1 },
-        visibilityTimeoutMs: 2000,
-        attempt: 0,
-      });
+      expect(first).toEqual(
+        expect.objectContaining({
+          id: "1",
+          job: "test",
+          item: { value: 1 },
+          visibilityTimeoutMs: 2000,
+          attempt: 0,
+          timestamp: expect.any(Date),
+        })
+      );
     } finally {
       await queue.close();
     }
@@ -160,13 +172,16 @@ describe("SimpleQueue", () => {
       await queue.enqueue({ id: "1", job: "test", item: { value: 1 }, visibilityTimeoutMs: 1_000 });
 
       const [first] = await queue.dequeue();
-      expect(first).toEqual({
-        id: "1",
-        job: "test",
-        item: { value: 1 },
-        visibilityTimeoutMs: 1_000,
-        attempt: 0,
-      });
+      expect(first).toEqual(
+        expect.objectContaining({
+          id: "1",
+          job: "test",
+          item: { value: 1 },
+          visibilityTimeoutMs: 1_000,
+          attempt: 0,
+          timestamp: expect.any(Date),
+        })
+      );
 
       const missImmediate = await queue.dequeue(1);
       expect(missImmediate).toEqual([]);
@@ -174,13 +189,16 @@ describe("SimpleQueue", () => {
       await new Promise((resolve) => setTimeout(resolve, 1_000));
 
       const [second] = await queue.dequeue();
-      expect(second).toEqual({
-        id: "1",
-        job: "test",
-        item: { value: 1 },
-        visibilityTimeoutMs: 1_000,
-        attempt: 0,
-      });
+      expect(second).toEqual(
+        expect.objectContaining({
+          id: "1",
+          job: "test",
+          item: { value: 1 },
+          visibilityTimeoutMs: 1_000,
+          attempt: 0,
+          timestamp: expect.any(Date),
+        })
+      );
     } finally {
       await queue.close();
     }
@@ -211,20 +229,26 @@ describe("SimpleQueue", () => {
 
       const dequeued = await queue.dequeue(2);
       expect(dequeued).toHaveLength(2);
-      expect(dequeued[0]).toEqual({
-        id: "1",
-        job: "test",
-        item: { value: 1 },
-        visibilityTimeoutMs: 2000,
-        attempt: 0,
-      });
-      expect(dequeued[1]).toEqual({
-        id: "2",
-        job: "test",
-        item: { value: 2 },
-        visibilityTimeoutMs: 2000,
-        attempt: 0,
-      });
+      expect(dequeued[0]).toEqual(
+        expect.objectContaining({
+          id: "1",
+          job: "test",
+          item: { value: 1 },
+          visibilityTimeoutMs: 2000,
+          attempt: 0,
+          timestamp: expect.any(Date),
+        })
+      );
+      expect(dequeued[1]).toEqual(
+        expect.objectContaining({
+          id: "2",
+          job: "test",
+          item: { value: 2 },
+          visibilityTimeoutMs: 2000,
+          attempt: 0,
+          timestamp: expect.any(Date),
+        })
+      );
 
       expect(await queue.size()).toBe(1);
       expect(await queue.size({ includeFuture: true })).toBe(3);
@@ -235,13 +259,16 @@ describe("SimpleQueue", () => {
       expect(await queue.size({ includeFuture: true })).toBe(1);
 
       const [last] = await queue.dequeue(1);
-      expect(last).toEqual({
-        id: "3",
-        job: "test",
-        item: { value: 3 },
-        visibilityTimeoutMs: 2000,
-        attempt: 0,
-      });
+      expect(last).toEqual(
+        expect.objectContaining({
+          id: "3",
+          job: "test",
+          item: { value: 3 },
+          visibilityTimeoutMs: 2000,
+          attempt: 0,
+          timestamp: expect.any(Date),
+        })
+      );
 
       await queue.ack(last.id);
       expect(await queue.size({ includeFuture: true })).toBe(0);
@@ -288,13 +315,16 @@ describe("SimpleQueue", () => {
 
       // Dequeue the redriven item
       const [redrivenItem] = await queue.dequeue(1);
-      expect(redrivenItem).toEqual({
-        id: "1",
-        job: "test",
-        item: { value: 1 },
-        visibilityTimeoutMs: 2000,
-        attempt: 0,
-      });
+      expect(redrivenItem).toEqual(
+        expect.objectContaining({
+          id: "1",
+          job: "test",
+          item: { value: 1 },
+          visibilityTimeoutMs: 2000,
+          attempt: 0,
+          timestamp: expect.any(Date),
+        })
+      );
 
       // Acknowledge the item
       await queue.ack(redrivenItem.id);

internal-packages/redis-worker/src/queue.ts

@@ -13,6 +13,25 @@ export type MessageCatalogValue<
   TKey extends MessageCatalogKey<TMessageCatalog>,
 > = z.infer<TMessageCatalog[TKey]>;
 
+export type AnyMessageCatalog = MessageCatalogSchema;
+export type QueueItem<TMessageCatalog extends MessageCatalogSchema> = {
+  id: string;
+  job: MessageCatalogKey<TMessageCatalog>;
+  item: MessageCatalogValue<TMessageCatalog, MessageCatalogKey<TMessageCatalog>>;
+  visibilityTimeoutMs: number;
+  attempt: number;
+  timestamp: Date;
+};
+
+export type AnyQueueItem = {
+  id: string;
+  job: string;
+  item: any;
+  visibilityTimeoutMs: number;
+  attempt: number;
+  timestamp: Date;
+};
+
 export class SimpleQueue<TMessageCatalog extends MessageCatalogSchema> {
   name: string;
   private redis: Redis;
@@ -33,7 +52,7 @@ export class SimpleQueue<TMessageCatalog extends MessageCatalogSchema> {
     this.name = name;
     this.redis = new Redis({
       ...redisOptions,
-      keyPrefix: `{queue:${name}:}`,
+      keyPrefix: `${redisOptions.keyPrefix ?? ""}{queue:${name}:}`,
       retryStrategy(times) {
         const delay = Math.min(times * 50, 1000);
         return delay;
@@ -107,15 +126,7 @@ export class SimpleQueue<TMessageCatalog extends MessageCatalogSchema> {
       throw e;
     }
   }
-  async dequeue(count: number = 1): Promise<
-    Array<{
-      id: string;
-      job: MessageCatalogKey<TMessageCatalog>;
-      item: MessageCatalogValue<TMessageCatalog, MessageCatalogKey<TMessageCatalog>>;
-      visibilityTimeoutMs: number;
-      attempt: number;
-    }>
-  > {
+  async dequeue(count: number = 1): Promise<Array<QueueItem<TMessageCatalog>>> {
     const now = Date.now();
 
     try {
@@ -127,13 +138,15 @@ export class SimpleQueue<TMessageCatalog extends MessageCatalogSchema> {
 
       const dequeuedItems = [];
 
-      for (const [id, serializedItem] of results) {
-        const parsedItem = JSON.parse(serializedItem);
+      for (const [id, serializedItem, score] of results) {
+        const parsedItem = JSON.parse(serializedItem) as any;
         if (typeof parsedItem.job !== "string") {
           this.logger.error(`Invalid item in queue`, { queue: this.name, id, item: parsedItem });
           continue;
         }
 
+        const timestamp = new Date(Number(score));
+
         const schema = this.schema[parsedItem.job];
 
         if (!schema) {
@@ -142,6 +155,7 @@ export class SimpleQueue<TMessageCatalog extends MessageCatalogSchema> {
             id,
             item: parsedItem,
             job: parsedItem.job,
+            timestamp,
           });
           continue;
         }
@@ -155,6 +169,7 @@ export class SimpleQueue<TMessageCatalog extends MessageCatalogSchema> {
             item: parsedItem,
             errors: validatedItem.error,
             attempt: parsedItem.attempt,
+            timestamp,
           });
           continue;
         }
@@ -170,6 +185,7 @@ export class SimpleQueue<TMessageCatalog extends MessageCatalogSchema> {
           item: validatedItem.data,
           visibilityTimeoutMs,
           attempt: parsedItem.attempt ?? 0,
+          timestamp,
         });
       }
 
@@ -336,7 +352,7 @@ export class SimpleQueue<TMessageCatalog extends MessageCatalogSchema> {
             local invisibleUntil = now + visibilityTimeoutMs
 
             redis.call('ZADD', queue, invisibleUntil, id)
-            table.insert(dequeued, {id, serializedItem})
+            table.insert(dequeued, {id, serializedItem, score})
           end
         end
 
@@ -376,10 +392,13 @@ export class SimpleQueue<TMessageCatalog extends MessageCatalogSchema> {
         local parsedItem = cjson.decode(item)
         parsedItem.errorMessage = errorMessage
 
+        local time = redis.call('TIME')
+        local now = tonumber(time[1]) * 1000 + math.floor(tonumber(time[2]) / 1000)
+
         redis.call('ZREM', queue, id)
         redis.call('HDEL', items, id)
 
-        redis.call('ZADD', dlq, redis.call('TIME')[1], id)
+        redis.call('ZADD', dlq, now, id)
         redis.call('HSET', dlqItems, id, cjson.encode(parsedItem))
 
         return 1
@@ -403,10 +422,13 @@ export class SimpleQueue<TMessageCatalog extends MessageCatalogSchema> {
         local parsedItem = cjson.decode(item)
         parsedItem.errorMessage = nil
 
+        local time = redis.call('TIME')
+        local now = tonumber(time[1]) * 1000 + math.floor(tonumber(time[2]) / 1000)
+
         redis.call('ZREM', dlq, id)
         redis.call('HDEL', dlqItems, id)
 
-        redis.call('ZADD', queue, redis.call('TIME')[1], id)
+        redis.call('ZADD', queue, now, id)
         redis.call('HSET', items, id, cjson.encode(parsedItem))
 
         return 1
@@ -435,8 +457,8 @@ declare module "ioredis" {
       //args
       now: number,
       count: number,
-      callback?: Callback<Array<[string, string]>>
-    ): Result<Array<[string, string]>, Context>;
+      callback?: Callback<Array<[string, string, string]>>
+    ): Result<Array<[string, string, string]>, Context>;
 
     ackItem(
       queue: string,

internal-packages/redis-worker/src/telemetry.ts

@@ -0,0 +1,31 @@
+import { SpanOptions, SpanStatusCode, Span, Tracer } from "@opentelemetry/api";
+
+export async function startSpan<T>(
+  tracer: Tracer,
+  name: string,
+  fn: (span: Span) => Promise<T>,
+  options?: SpanOptions
+): Promise<T> {
+  return tracer.startActiveSpan(name, options ?? {}, async (span) => {
+    try {
+      return await fn(span);
+    } catch (error) {
+      if (error instanceof Error) {
+        span.recordException(error);
+      } else if (typeof error === "string") {
+        span.recordException(new Error(error));
+      } else {
+        span.recordException(new Error(String(error)));
+      }
+
+      span.setStatus({
+        code: SpanStatusCode.ERROR,
+        message: error instanceof Error ? error.message : String(error),
+      });
+
+      throw error;
+    } finally {
+      span.end();
+    }
+  });
+}

internal-packages/redis-worker/src/worker.test.ts

@@ -34,7 +34,7 @@ describe("Worker", () => {
         tasksPerWorker: 3,
       },
       logger: new Logger("test", "log"),
-    });
+    }).start();
 
     try {
       // Enqueue 10 items
@@ -47,10 +47,8 @@ describe("Worker", () => {
         });
       }
 
-      worker.start();
-
       // Wait for items to be processed
-      await new Promise((resolve) => setTimeout(resolve, 600));
+      await new Promise((resolve) => setTimeout(resolve, 2000));
 
       expect(processedItems.length).toBe(10);
       expect(new Set(processedItems).size).toBe(10); // Ensure all items were processed uniquely
@@ -97,7 +95,7 @@ describe("Worker", () => {
         },
         pollIntervalMs: 50,
         logger: new Logger("test", "error"),
-      });
+      }).start();
 
       try {
         // Enqueue 10 items
@@ -110,8 +108,6 @@ describe("Worker", () => {
           });
         }
 
-        worker.start();
-
         // Wait for items to be processed
         await new Promise((resolve) => setTimeout(resolve, 500));
 
@@ -158,7 +154,7 @@ describe("Worker", () => {
         },
         pollIntervalMs: 50,
         logger: new Logger("test", "error"),
-      });
+      }).start();
 
       try {
         // Enqueue the item that will permanently fail
@@ -175,8 +171,6 @@ describe("Worker", () => {
           payload: { value: 1 },
         });
 
-        worker.start();
-
         // Wait for items to be processed and retried
         await new Promise((resolve) => setTimeout(resolve, 1000));
 
@@ -229,7 +223,7 @@ describe("Worker", () => {
         },
         pollIntervalMs: 50,
         logger: new Logger("test", "error"),
-      });
+      }).start();
 
       try {
         // Enqueue the item that will fail 3 times
@@ -239,8 +233,6 @@ describe("Worker", () => {
           payload: { value: 999 },
         });
 
-        worker.start();
-
         // Wait for the item to be processed and moved to DLQ
         await new Promise((resolve) => setTimeout(resolve, 1000));
 
@@ -274,11 +266,4 @@ describe("Worker", () => {
       }
     }
   );
-
-  //todo test that throwing an error doesn't screw up the other items
-  //todo process more items when finished
-
-  //todo add a Dead Letter Queue when items are failed, with the error
-  //todo add a function on the worker to redrive them
-  //todo add an API endpoint to redrive with an ID
 });

internal-packages/redis-worker/src/worker.ts

@@ -1,19 +1,20 @@
+import { SpanKind, trace, Tracer } from "@opentelemetry/api";
 import { Logger } from "@trigger.dev/core/logger";
-import { type RetryOptions } from "@trigger.dev/core/v3/schemas";
 import { calculateNextRetryDelay } from "@trigger.dev/core/v3";
+import { type RetryOptions } from "@trigger.dev/core/v3/schemas";
 import { type RedisOptions } from "ioredis";
-import os from "os";
-import { Worker as NodeWorker } from "worker_threads";
 import { z } from "zod";
-import { SimpleQueue } from "./queue.js";
-
+import { AnyQueueItem, SimpleQueue } from "./queue.js";
 import Redis from "ioredis";
+import { nanoid } from "nanoid";
+import { startSpan } from "./telemetry.js";
+import pLimit from "p-limit";
 
-type WorkerCatalog = {
+export type WorkerCatalog = {
   [key: string]: {
     schema: z.ZodFirstPartySchemaTypes | z.ZodDiscriminatedUnion<any, any>;
     visibilityTimeoutMs: number;
-    retry: RetryOptions;
+    retry?: RetryOptions;
   };
 };
 
@@ -28,6 +29,12 @@ type JobHandler<Catalog extends WorkerCatalog, K extends keyof Catalog> = (param
   attempt: number;
 }) => Promise<void>;
 
+export type WorkerConcurrencyOptions = {
+  workers?: number;
+  tasksPerWorker?: number;
+  limit?: number;
+};
+
 type WorkerOptions<TCatalog extends WorkerCatalog> = {
   name: string;
   redisOptions: RedisOptions;
@@ -35,31 +42,46 @@ type WorkerOptions<TCatalog extends WorkerCatalog> = {
   jobs: {
     [K in keyof TCatalog]: JobHandler<TCatalog, K>;
   };
-  concurrency?: {
-    workers?: number;
-    tasksPerWorker?: number;
-  };
+  concurrency?: WorkerConcurrencyOptions;
   pollIntervalMs?: number;
+  immediatePollIntervalMs?: number;
   logger?: Logger;
+  tracer?: Tracer;
+};
+
+// This results in attempt 12 being a delay of 1 hour
+const defaultRetrySettings = {
+  maxAttempts: 12,
+  factor: 2,
+  //one second
+  minTimeoutInMs: 1_000,
+  //one hour
+  maxTimeoutInMs: 3_600_000,
+  randomize: true,
 };
 
 class Worker<TCatalog extends WorkerCatalog> {
-  private subscriber: Redis;
+  private subscriber: Redis | undefined;
+  private tracer: Tracer;
 
   queue: SimpleQueue<QueueCatalogFromWorkerCatalog<TCatalog>>;
   private jobs: WorkerOptions<TCatalog>["jobs"];
   private logger: Logger;
-  private workers: NodeWorker[] = [];
+  private workerLoops: Promise<void>[] = [];
   private isShuttingDown = false;
   private concurrency: Required<NonNullable<WorkerOptions<TCatalog>["concurrency"]>>;
 
+  // The p-limit limiter to control overall concurrency.
+  private limiter: ReturnType<typeof pLimit>;
+
   constructor(private options: WorkerOptions<TCatalog>) {
     this.logger = options.logger ?? new Logger("Worker", "debug");
+    this.tracer = options.tracer ?? trace.getTracer(options.name);
 
     const schema: QueueCatalogFromWorkerCatalog<TCatalog> = Object.fromEntries(
       Object.entries(this.options.catalog).map(([key, value]) => [key, value.schema])
     ) as QueueCatalogFromWorkerCatalog<TCatalog>;
-    //
+
     this.queue = new SimpleQueue({
       name: options.name,
       redisOptions: options.redisOptions,
@@ -69,187 +91,251 @@ class Worker<TCatalog extends WorkerCatalog> {
 
     this.jobs = options.jobs;
 
-    const { workers = os.cpus().length, tasksPerWorker = 1 } = options.concurrency ?? {};
-    this.concurrency = { workers, tasksPerWorker };
+    const { workers = 1, tasksPerWorker = 1, limit = 10 } = options.concurrency ?? {};
+    this.concurrency = { workers, tasksPerWorker, limit };
 
-    // Initialize worker threads
+    // Create a p-limit instance using this limit.
+    this.limiter = pLimit(this.concurrency.limit);
+  }
+
+  public start() {
+    const { workers, tasksPerWorker } = this.concurrency;
+
+    // Launch a number of "worker loops" on the main thread.
     for (let i = 0; i < workers; i++) {
-      this.createWorker(tasksPerWorker);
+      this.workerLoops.push(this.runWorkerLoop(`worker-${nanoid(12)}`, tasksPerWorker));
     }
 
     this.setupShutdownHandlers();
 
-    this.subscriber = new Redis(options.redisOptions);
+    this.subscriber = new Redis(this.options.redisOptions);
     this.setupSubscriber();
+
+    return this;
   }
 
+  /**
+   * Enqueues a job for processing.
+   * @param options - The enqueue options.
+   * @param options.id - Optional unique identifier for the job. If not provided, one will be generated. It prevents duplication.
+   * @param options.job - The job type from the worker catalog.
+   * @param options.payload - The job payload that matches the schema defined in the catalog.
+   * @param options.visibilityTimeoutMs - Optional visibility timeout in milliseconds. Defaults to value from catalog.
+   * @param options.availableAt - Optional date when the job should become available for processing. Defaults to now.
+   * @returns A promise that resolves when the job is enqueued.
+   */
   enqueue<K extends keyof TCatalog>({
     id,
     job,
     payload,
     visibilityTimeoutMs,
+    availableAt,
   }: {
     id?: string;
     job: K;
     payload: z.infer<TCatalog[K]["schema"]>;
     visibilityTimeoutMs?: number;
+    availableAt?: Date;
   }) {
-    const timeout = visibilityTimeoutMs ?? this.options.catalog[job].visibilityTimeoutMs;
-    return this.queue.enqueue({
-      id,
-      job,
-      item: payload,
-      visibilityTimeoutMs: timeout,
-    });
-  }
+    return startSpan(
+      this.tracer,
+      "enqueue",
+      async (span) => {
+        const timeout = visibilityTimeoutMs ?? this.options.catalog[job].visibilityTimeoutMs;
 
-  private createWorker(tasksPerWorker: number) {
-    const worker = new NodeWorker(
-      `
-      const { parentPort } = require('worker_threads');
+        span.setAttribute("job_visibility_timeout_ms", timeout);
 
-      parentPort.on('message', async (message) => {
-        if (message.type === 'process') {
-          // Process items here
-          parentPort.postMessage({ type: 'done' });
-        }
-      });
-    `,
-      { eval: true }
+        return this.queue.enqueue({
+          id,
+          job,
+          item: payload,
+          visibilityTimeoutMs: timeout,
+          availableAt,
+        });
+      },
+      {
+        kind: SpanKind.PRODUCER,
+        attributes: {
+          job_type: job as string,
+          job_id: id,
+        },
+      }
     );
-
-    worker.on("message", (message) => {
-      if (message.type === "done") {
-        this.processItems(worker, tasksPerWorker);
-      }
-    });
-
-    worker.on("error", (error) => {
-      this.logger.error("Worker error:", { error });
-    });
-
-    worker.on("exit", (code) => {
-      if (code !== 0) {
-        this.logger.warn(`Worker stopped with exit code ${code}`);
-      }
-      if (!this.isShuttingDown) {
-        this.createWorker(tasksPerWorker);
-      }
-    });
-
-    this.workers.push(worker);
-    this.processItems(worker, tasksPerWorker);
   }
 
-  private async processItems(worker: NodeWorker, count: number) {
-    if (this.isShuttingDown) return;
+  ack(id: string) {
+    return startSpan(
+      this.tracer,
+      "ack",
+      () => {
+        return this.queue.ack(id);
+      },
+      {
+        attributes: {
+          job_id: id,
+        },
+      }
+    );
+  }
 
+  /**
+   * The main loop that each worker runs. It repeatedly polls for items,
+   * processes them, and then waits before the next iteration.
+   */
+  private async runWorkerLoop(workerId: string, taskCount: number): Promise<void> {
     const pollIntervalMs = this.options.pollIntervalMs ?? 1000;
+    const immediatePollIntervalMs = this.options.immediatePollIntervalMs ?? 100;
 
-    try {
-      const items = await this.queue.dequeue(count);
-      if (items.length === 0) {
-        setTimeout(() => this.processItems(worker, count), pollIntervalMs);
-        return;
+    while (!this.isShuttingDown) {
+      // Check overall load. If at capacity, wait a bit before trying to dequeue more.
+      if (this.limiter.activeCount + this.limiter.pendingCount >= this.concurrency.limit) {
+        await Worker.delay(pollIntervalMs);
+        continue;
       }
 
-      await Promise.all(
-        items.map(async ({ id, job, item, visibilityTimeoutMs, attempt }) => {
-          const catalogItem = this.options.catalog[job as any];
-          const handler = this.jobs[job as any];
-          if (!handler) {
-            this.logger.error(`No handler found for job type: ${job as string}`);
-            return;
-          }
+      try {
+        const items = await this.queue.dequeue(taskCount);
 
-          try {
-            await handler({ id, payload: item, visibilityTimeoutMs, attempt });
+        if (items.length === 0) {
+          await Worker.delay(pollIntervalMs);
+          continue;
+        }
 
-            //succeeded, acking the item
-            await this.queue.ack(id);
-          } catch (error) {
-            const errorMessage = error instanceof Error ? error.message : String(error);
-            this.logger.error(`Error processing item, it threw an error:`, {
-              name: this.options.name,
-              id,
-              job,
-              item,
-              visibilityTimeoutMs,
-              error,
-              errorMessage,
-            });
-            // Requeue the failed item with a delay
-            try {
-              attempt = attempt + 1;
-
-              const retryDelay = calculateNextRetryDelay(catalogItem.retry, attempt);
-
-              if (!retryDelay) {
-                this.logger.error(
-                  `Failed item ${id} has reached max attempts, moving to the DLQ.`,
-                  {
-                    name: this.options.name,
-                    id,
-                    job,
-                    item,
-                    visibilityTimeoutMs,
-                    attempt,
-                    errorMessage,
-                  }
-                );
-
-                await this.queue.moveToDeadLetterQueue(id, errorMessage);
-                return;
-              }
-
-              const retryDate = new Date(Date.now() + retryDelay);
-              this.logger.info(`Requeued failed item ${id} with delay`, {
-                name: this.options.name,
-                id,
-                job,
-                item,
-                retryDate,
-                retryDelay,
-                visibilityTimeoutMs,
-                attempt,
-              });
-              await this.queue.enqueue({
-                id,
-                job,
-                item,
-                availableAt: retryDate,
-                attempt,
-                visibilityTimeoutMs,
-              });
-            } catch (requeueError) {
-              this.logger.error(
-                `Failed to requeue item, threw error. Will automatically get rescheduled after the visilibity timeout.`,
-                {
-                  name: this.options.name,
-                  id,
-                  job,
-                  item,
-                  visibilityTimeoutMs,
-                  error: requeueError,
-                }
-              );
+        // Schedule each item using the limiter.
+        for (const item of items) {
+          this.limiter(() => this.processItem(item as AnyQueueItem, items.length, workerId)).catch(
+            (err) => {
+              this.logger.error("Unhandled error in processItem:", { error: err, workerId, item });
             }
-          }
-        })
-      );
-    } catch (error) {
-      this.logger.error("Error dequeuing items:", { name: this.options.name, error });
-      setTimeout(() => this.processItems(worker, count), pollIntervalMs);
+          );
+        }
+      } catch (error) {
+        this.logger.error("Error dequeuing items:", { name: this.options.name, error });
+        await Worker.delay(pollIntervalMs);
+        continue;
+      }
+
+      // Wait briefly before immediately polling again since we processed items
+      await Worker.delay(immediatePollIntervalMs);
+    }
+  }
+
+  /**
+   * Processes a single item.
+   */
+  private async processItem(
+    { id, job, item, visibilityTimeoutMs, attempt, timestamp }: AnyQueueItem,
+    batchSize: number,
+    workerId: string
+  ): Promise<void> {
+    const catalogItem = this.options.catalog[job as any];
+    const handler = this.jobs[job as any];
+    if (!handler) {
+      this.logger.error(`No handler found for job type: ${job}`);
       return;
     }
 
-    // Immediately process next batch because there were items in the queue
-    this.processItems(worker, count);
+    await startSpan(
+      this.tracer,
+      "processItem",
+      async () => {
+        await handler({ id, payload: item, visibilityTimeoutMs, attempt });
+        // On success, acknowledge the item.
+        await this.queue.ack(id);
+      },
+      {
+        kind: SpanKind.CONSUMER,
+        attributes: {
+          job_id: id,
+          job_type: job,
+          attempt,
+          job_timestamp: timestamp.getTime(),
+          job_age_in_ms: Date.now() - timestamp.getTime(),
+          worker_id: workerId,
+          worker_limit_concurrency: this.limiter.concurrency,
+          worker_limit_active: this.limiter.activeCount,
+          worker_limit_pending: this.limiter.pendingCount,
+          worker_name: this.options.name,
+          batch_size: batchSize,
+        },
+      }
+    ).catch(async (error) => {
+      const errorMessage = error instanceof Error ? error.message : String(error);
+      this.logger.error(`Error processing item:`, {
+        name: this.options.name,
+        id,
+        job,
+        item,
+        visibilityTimeoutMs,
+        error,
+        errorMessage,
+      });
+      // Attempt requeue logic.
+      try {
+        const newAttempt = attempt + 1;
+        const retrySettings = {
+          ...defaultRetrySettings,
+          ...catalogItem.retry,
+        };
+        const retryDelay = calculateNextRetryDelay(retrySettings, newAttempt);
+
+        if (!retryDelay) {
+          this.logger.error(`Item ${id} reached max attempts. Moving to DLQ.`, {
+            name: this.options.name,
+            id,
+            job,
+            item,
+            visibilityTimeoutMs,
+            attempt: newAttempt,
+            errorMessage,
+          });
+          await this.queue.moveToDeadLetterQueue(id, errorMessage);
+          return;
+        }
+
+        const retryDate = new Date(Date.now() + retryDelay);
+        this.logger.info(`Requeuing failed item ${id} with delay`, {
+          name: this.options.name,
+          id,
+          job,
+          item,
+          retryDate,
+          retryDelay,
+          visibilityTimeoutMs,
+          attempt: newAttempt,
+        });
+        await this.queue.enqueue({
+          id,
+          job,
+          item,
+          availableAt: retryDate,
+          attempt: newAttempt,
+          visibilityTimeoutMs,
+        });
+      } catch (requeueError) {
+        this.logger.error(
+          `Failed to requeue item ${id}. It will be retried after the visibility timeout.`,
+          {
+            name: this.options.name,
+            id,
+            job,
+            item,
+            visibilityTimeoutMs,
+            error: requeueError,
+          }
+        );
+      }
+    });
+  }
+
+  // A simple helper to delay for a given number of milliseconds.
+  private static delay(ms: number): Promise<void> {
+    return new Promise((resolve) => setTimeout(resolve, ms));
   }
 
   private setupSubscriber() {
     const channel = `${this.options.name}:redrive`;
-    this.subscriber.subscribe(channel, (err) => {
+    this.subscriber?.subscribe(channel, (err) => {
       if (err) {
         this.logger.error(`Failed to subscribe to ${channel}`, { error: err });
       } else {
@@ -257,12 +343,12 @@ class Worker<TCatalog extends WorkerCatalog> {
       }
     });
 
-    this.subscriber.on("message", this.handleRedriveMessage.bind(this));
+    this.subscriber?.on("message", this.handleRedriveMessage.bind(this));
   }
 
   private async handleRedriveMessage(channel: string, message: string) {
     try {
-      const { id } = JSON.parse(message);
+      const { id } = JSON.parse(message) as any;
       if (typeof id !== "string") {
         throw new Error("Invalid message format: id must be a string");
       }
@@ -281,28 +367,19 @@ class Worker<TCatalog extends WorkerCatalog> {
   private async shutdown() {
     if (this.isShuttingDown) return;
     this.isShuttingDown = true;
-    this.logger.log("Shutting down workers...");
+    this.logger.log("Shutting down worker loops...");
 
-    for (const worker of this.workers) {
-      worker.terminate();
-    }
+    // Wait for all worker loops to finish.
+    await Promise.all(this.workerLoops);
 
-    await this.subscriber.unsubscribe();
-    await this.subscriber.quit();
+    await this.subscriber?.unsubscribe();
+    await this.subscriber?.quit();
     await this.queue.close();
     this.logger.log("All workers and subscribers shut down.");
   }
 
-  public start() {
-    this.logger.log("Starting workers...");
-    this.isShuttingDown = false;
-    for (const worker of this.workers) {
-      this.processItems(worker, this.concurrency.tasksPerWorker);
-    }
-  }
-
-  public stop() {
-    this.shutdown();
+  public async stop() {
+    await this.shutdown();
   }
 }
 

internal-packages/redis-worker/tsconfig.json

@@ -1,7 +1,7 @@
 {
   "compilerOptions": {
     "target": "ES2019",
-    "lib": ["ES2019", "DOM", "DOM.Iterable"],
+    "lib": ["ES2019", "DOM", "DOM.Iterable", "DOM.AsyncIterable"],
     "module": "CommonJS",
     "moduleResolution": "Node",
     "moduleDetection": "force",

internal-packages/redis-worker/vitest.config.ts

@@ -4,5 +4,11 @@ export default defineConfig({
   test: {
     include: ["**/*.test.ts"],
     globals: true,
+    fileParallelism: false,
+    poolOptions: {
+      threads: {
+        singleThread: true,
+      },
+    },
   },
 });

packages/build/CHANGELOG.md

@@ -1,5 +1,13 @@
 # @trigger.dev/build
 
+## 3.3.14
+
+### Patch Changes
+
+- Missing construction option in `AudioWaveformExtension` ([#1684](https://github.com/triggerdotdev/trigger.dev/pull/1684))
+- Updated dependencies:
+  - `@trigger.dev/core@3.3.14`
+
 ## 3.3.13
 
 ### Patch Changes

packages/build/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/build",
-  "version": "3.3.13",
+  "version": "3.3.14",
   "description": "trigger.dev build extensions",
   "license": "MIT",
   "publishConfig": {
@@ -65,7 +65,7 @@
     "check-exports": "attw --pack ."
   },
   "dependencies": {
-    "@trigger.dev/core": "workspace:3.3.13",
+    "@trigger.dev/core": "workspace:3.3.14",
     "pkg-types": "^1.1.3",
     "tinyglobby": "^0.2.2",
     "tsconfck": "3.1.3"

packages/build/src/extensions/audioWaveform.ts

@@ -11,7 +11,7 @@ const AUDIOWAVEFORM_CHECKSUM =
   "sha256:00b41ea4d6e7a5b4affcfe4ac99951ec89da81a8cba40af19e9b98c3a8f9b4b8";
 
 export function audioWaveform(options: AudioWaveformOptions = {}): BuildExtension {
-  return new AudioWaveformExtension();
+  return new AudioWaveformExtension(options);
 }
 
 class AudioWaveformExtension implements BuildExtension {

packages/cli-v3/CHANGELOG.md

@@ -1,5 +1,14 @@
 # trigger.dev
 
+## 3.3.14
+
+### Patch Changes
+
+- Auto-fix /trigger or /src/trigger config.dirs to relative paths to prevent misconfiguration from preventing dev CLI from working ([#1665](https://github.com/triggerdotdev/trigger.dev/pull/1665))
+- Updated dependencies:
+  - `@trigger.dev/build@3.3.14`
+  - `@trigger.dev/core@3.3.14`
+
 ## 3.3.13
 
 ### Patch Changes

packages/cli-v3/package.json

@@ -1,6 +1,6 @@
 {
   "name": "trigger.dev",
-  "version": "3.3.13",
+  "version": "3.3.14",
   "description": "A Command-Line Interface for Trigger.dev (v3) projects",
   "type": "module",
   "license": "MIT",
@@ -87,8 +87,8 @@
     "@opentelemetry/sdk-trace-base": "1.25.1",
     "@opentelemetry/sdk-trace-node": "1.25.1",
     "@opentelemetry/semantic-conventions": "1.25.1",
-    "@trigger.dev/build": "workspace:3.3.13",
-    "@trigger.dev/core": "workspace:3.3.13",
+    "@trigger.dev/build": "workspace:3.3.14",
+    "@trigger.dev/core": "workspace:3.3.14",
     "c12": "^1.11.1",
     "chalk": "^5.2.0",
     "chokidar": "^3.6.0",

packages/cli-v3/src/config.ts

@@ -155,11 +155,11 @@ async function resolveConfig(
 
   let dirs = config.dirs ? config.dirs : await autoDetectDirs(workingDir);
 
-  dirs = dirs.map((dir) => (isAbsolute(dir) ? relative(workingDir, dir) : dir));
+  dirs = dirs.map((dir) => resolveTriggerDir(dir, workingDir));
 
   const mergedConfig = defu(
     {
-      workingDir: packageJsonPath ? dirname(packageJsonPath) : cwd,
+      workingDir,
       configFile: result.configFile,
       packageJsonPath,
       tsconfigPath,
@@ -187,11 +187,24 @@ async function resolveConfig(
 
   return {
     ...mergedConfig,
-    dirs: Array.from(new Set(mergedConfig.dirs)),
+    dirs: Array.from(new Set(dirs)),
     instrumentedPackageNames: getInstrumentedPackageNames(mergedConfig),
   };
 }
 
+function resolveTriggerDir(dir: string, workingDir: string): string {
+  if (isAbsolute(dir)) {
+    // If dir is `/trigger` or `/src/trigger`, we should add a `.` to make it relative to the working directory
+    if (dir === "/trigger" || dir === "/src/trigger") {
+      return `.${dir}`;
+    } else {
+      return relative(workingDir, dir);
+    }
+  }
+
+  return dir;
+}
+
 async function safeResolveTsConfig(cwd: string) {
   try {
     return await resolveTSConfig(cwd);

packages/core/CHANGELOG.md

@@ -1,5 +1,7 @@
 # internal-platform
 
+## 3.3.14
+
 ## 3.3.13
 
 ### Patch Changes

packages/core/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/core",
-  "version": "3.3.13",
+  "version": "3.3.14",
   "description": "Core code used across the Trigger.dev SDK and platform",
   "license": "MIT",
   "publishConfig": {

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

@@ -509,7 +509,7 @@ const isSafari = () => {
  */
 
 if (isSafari()) {
-  // @ts-expect-error
+  // @ts-ignore-error
   ReadableStream.prototype.values ??= function ({ preventCancel = false } = {}) {
     const reader = this.getReader();
     return {
@@ -541,6 +541,6 @@ if (isSafari()) {
     };
   };
 
-  // @ts-expect-error
+  // @ts-ignore-error
   ReadableStream.prototype[Symbol.asyncIterator] ??= ReadableStream.prototype.values;
 }

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

@@ -74,25 +74,25 @@ export const TriggerTaskRequestBody = z.object({
   context: z.any(),
   options: z
     .object({
-      dependentAttempt: z.string().optional(),
-      parentAttempt: z.string().optional(),
-      dependentBatch: z.string().optional(),
-      parentBatch: z.string().optional(),
-      lockToVersion: z.string().optional(),
-      queue: QueueOptions.optional(),
       concurrencyKey: z.string().optional(),
+      delay: z.string().or(z.coerce.date()).optional(),
+      dependentAttempt: z.string().optional(),
+      dependentBatch: z.string().optional(),
       idempotencyKey: z.string().optional(),
       idempotencyKeyTTL: z.string().optional(),
-      test: z.boolean().optional(),
-      payloadType: z.string().optional(),
-      delay: z.string().or(z.coerce.date()).optional(),
-      ttl: z.string().or(z.number().nonnegative().int()).optional(),
-      tags: RunTags.optional(),
+      lockToVersion: z.string().optional(),
+      machine: MachinePresetName.optional(),
       maxAttempts: z.number().int().optional(),
+      maxDuration: z.number().optional(),
       metadata: z.any(),
       metadataType: z.string().optional(),
-      maxDuration: z.number().optional(),
-      machine: MachinePresetName.optional(),
+      parentAttempt: z.string().optional(),
+      parentBatch: z.string().optional(),
+      payloadType: z.string().optional(),
+      queue: QueueOptions.optional(),
+      tags: RunTags.optional(),
+      test: z.boolean().optional(),
+      ttl: z.string().or(z.number().nonnegative().int()).optional(),
     })
     .optional(),
 });
@@ -118,22 +118,22 @@ export const BatchTriggerTaskItem = z.object({
   context: z.any(),
   options: z
     .object({
-      lockToVersion: z.string().optional(),
-      queue: QueueOptions.optional(),
       concurrencyKey: z.string().optional(),
+      delay: z.string().or(z.coerce.date()).optional(),
       idempotencyKey: z.string().optional(),
       idempotencyKeyTTL: z.string().optional(),
-      test: z.boolean().optional(),
-      payloadType: z.string().optional(),
-      delay: z.string().or(z.coerce.date()).optional(),
-      ttl: z.string().or(z.number().nonnegative().int()).optional(),
-      tags: RunTags.optional(),
+      lockToVersion: z.string().optional(),
+      machine: MachinePresetName.optional(),
       maxAttempts: z.number().int().optional(),
+      maxDuration: z.number().optional(),
       metadata: z.any(),
       metadataType: z.string().optional(),
-      maxDuration: z.number().optional(),
       parentAttempt: z.string().optional(),
-      machine: MachinePresetName.optional(),
+      payloadType: z.string().optional(),
+      queue: QueueOptions.optional(),
+      tags: RunTags.optional(),
+      test: z.boolean().optional(),
+      ttl: z.string().or(z.number().nonnegative().int()).optional(),
     })
     .optional(),
 });

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

@@ -1,6 +1,6 @@
 import { z } from "zod";
 import { RequireKeys } from "../types/index.js";
-import { MachineConfig, MachinePreset, TaskRunExecution } from "./common.js";
+import { MachineConfig, MachinePreset, MachinePresetName, TaskRunExecution } from "./common.js";
 
 /*
     WARNING: Never import anything from ./messages here. If it's needed in both, put it here instead.
@@ -95,15 +95,25 @@ export const RetryOptions = z.object({
    * This can be useful to prevent the thundering herd problem where all retries happen at the same time.
    */
   randomize: z.boolean().optional(),
+
+  /** If a run fails with an Out Of Memory (OOM) error and you have this set, it will retry with the machine you specify.
+   * Note: it will not default to this [machine](https://trigger.dev/docs/machines) for new runs, only for failures caused by OOM errors.
+   * So if you frequently have attempts failing with OOM errors, you should set the [default machine](https://trigger.dev/docs/machines) to be higher.
+   */
+  outOfMemory: z
+    .object({
+      machine: MachinePresetName.optional(),
+    })
+    .optional(),
 });
 
 export type RetryOptions = z.infer<typeof RetryOptions>;
 
 export const QueueOptions = z.object({
   /** You can define a shared queue and then pass the name in to your task.
-   * 
+   *
    * @example
-   * 
+   *
    * ```ts
    * const myQueue = queue({
       name: "my-queue",

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

@@ -202,17 +202,14 @@ type CommonTaskOptions<
    * ```
    */
   queue?: QueueOptions;
-  /** Configure the spec of the machine you want your task to run on.
+  /** Configure the spec of the [machine](https://trigger.dev/docs/machines) you want your task to run on.
    *
    * @example
    *
    * ```ts
    * export const heavyTask = task({
       id: "heavy-task",
-      machine: {
-        cpu: 2,
-        memory: 4,
-      },
+      machine: "medium-1x",
       run: async ({ payload, ctx }) => {
         //...
       },

packages/react-hooks/CHANGELOG.md

@@ -1,5 +1,12 @@
 # @trigger.dev/react-hooks
 
+## 3.3.14
+
+### Patch Changes
+
+- Updated dependencies:
+  - `@trigger.dev/core@3.3.14`
+
 ## 3.3.13
 
 ### Patch Changes

packages/react-hooks/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/react-hooks",
-  "version": "3.3.13",
+  "version": "3.3.14",
   "description": "trigger.dev react hooks",
   "license": "MIT",
   "publishConfig": {
@@ -37,7 +37,7 @@
     "check-exports": "attw --pack ."
   },
   "dependencies": {
-    "@trigger.dev/core": "workspace:^3.3.13",
+    "@trigger.dev/core": "workspace:^3.3.14",
     "swr": "^2.2.5"
   },
   "devDependencies": {

packages/rsc/CHANGELOG.md

@@ -1,5 +1,12 @@
 # @trigger.dev/rsc
 
+## 3.3.14
+
+### Patch Changes
+
+- Updated dependencies:
+  - `@trigger.dev/core@3.3.14`
+
 ## 3.3.13
 
 ### Patch Changes

packages/rsc/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/rsc",
-  "version": "3.3.13",
+  "version": "3.3.14",
   "description": "trigger.dev rsc",
   "license": "MIT",
   "publishConfig": {
@@ -37,14 +37,14 @@
     "check-exports": "attw --pack ."
   },
   "dependencies": {
-    "@trigger.dev/core": "workspace:^3.3.13",
+    "@trigger.dev/core": "workspace:^3.3.14",
     "mlly": "^1.7.1",
     "react": "19.0.0-rc.1",
     "react-dom": "19.0.0-rc.1"
   },
   "devDependencies": {
     "@arethetypeswrong/cli": "^0.15.4",
-    "@trigger.dev/build": "workspace:^3.3.13",
+    "@trigger.dev/build": "workspace:^3.3.14",
     "@types/node": "^20.14.14",
     "@types/react": "*",
     "@types/react-dom": "*",

packages/trigger-sdk/CHANGELOG.md

@@ -1,5 +1,13 @@
 # @trigger.dev/sdk
 
+## 3.3.14
+
+### Patch Changes
+
+- Added the ability to retry runs that fail with an Out Of Memory (OOM) error on a larger machine. ([#1691](https://github.com/triggerdotdev/trigger.dev/pull/1691))
+- Updated dependencies:
+  - `@trigger.dev/core@3.3.14`
+
 ## 3.3.13
 
 ### Patch Changes

packages/trigger-sdk/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@trigger.dev/sdk",
-  "version": "3.3.13",
+  "version": "3.3.14",
   "description": "trigger.dev Node.JS SDK",
   "license": "MIT",
   "publishConfig": {
@@ -48,7 +48,7 @@
     "@opentelemetry/api": "1.9.0",
     "@opentelemetry/api-logs": "0.52.1",
     "@opentelemetry/semantic-conventions": "1.25.1",
-    "@trigger.dev/core": "workspace:3.3.13",
+    "@trigger.dev/core": "workspace:3.3.14",
     "chalk": "^5.2.0",
     "cronstrue": "^2.21.0",
     "debug": "^4.3.4",

pnpm-lock.yaml

@@ -243,6 +243,9 @@ importers:
       '@heroicons/react':
         specifier: ^2.0.12
         version: 2.0.13(react@18.2.0)
+      '@internal/redis-worker':
+        specifier: workspace:*
+        version: link:../../internal-packages/redis-worker
       '@internal/zod-worker':
         specifier: workspace:*
         version: link:../../internal-packages/zod-worker
@@ -948,9 +951,9 @@ importers:
       nanoid:
         specifier: ^5.0.7
         version: 5.0.7
-      typescript:
-        specifier: ^5.5.4
-        version: 5.5.4
+      p-limit:
+        specifier: ^6.2.0
+        version: 6.2.0
       zod:
         specifier: 3.23.8
         version: 3.23.8
@@ -1033,7 +1036,7 @@ importers:
   packages/build:
     dependencies:
       '@trigger.dev/core':
-        specifier: workspace:3.3.13
+        specifier: workspace:3.3.14
         version: link:../core
       pkg-types:
         specifier: ^1.1.3
@@ -1112,10 +1115,10 @@ importers:
         specifier: 1.25.1
         version: 1.25.1
       '@trigger.dev/build':
-        specifier: workspace:3.3.13
+        specifier: workspace:3.3.14
         version: link:../build
       '@trigger.dev/core':
-        specifier: workspace:3.3.13
+        specifier: workspace:3.3.14
         version: link:../core
       c12:
         specifier: ^1.11.1
@@ -1417,7 +1420,7 @@ importers:
   packages/react-hooks:
     dependencies:
       '@trigger.dev/core':
-        specifier: workspace:^3.3.13
+        specifier: workspace:^3.3.14
         version: link:../core
       react:
         specifier: ^18.0 || ^19.0 || ^19.0.0-rc
@@ -1457,7 +1460,7 @@ importers:
   packages/rsc:
     dependencies:
       '@trigger.dev/core':
-        specifier: workspace:^3.3.13
+        specifier: workspace:^3.3.14
         version: link:../core
       mlly:
         specifier: ^1.7.1
@@ -1473,7 +1476,7 @@ importers:
         specifier: ^0.15.4
         version: 0.15.4
       '@trigger.dev/build':
-        specifier: workspace:^3.3.13
+        specifier: workspace:^3.3.14
         version: link:../build
       '@types/node':
         specifier: ^20.14.14
@@ -1509,7 +1512,7 @@ importers:
         specifier: 1.25.1
         version: 1.25.1
       '@trigger.dev/core':
-        specifier: workspace:3.3.13
+        specifier: workspace:3.3.14
         version: link:../core
       chalk:
         specifier: ^5.2.0
@@ -26272,6 +26275,13 @@ packages:
       yocto-queue: 1.0.0
     dev: true
 
+  /p-limit@6.2.0:
+    resolution: {integrity: sha512-kuUqqHNUqoIWp/c467RI4X6mmyuojY5jGutNU0wVTmEOOfcuwLqyMVoAi9MKi2Ak+5i9+nhmrK4ufZE8069kHA==}
+    engines: {node: '>=18'}
+    dependencies:
+      yocto-queue: 1.1.1
+    dev: false
+
   /p-locate@4.1.0:
     resolution: {integrity: sha512-R79ZZ/0wAxKGu3oYMlz8jy/kbhsNrS7SKZ7PxEHBgJ5+F2mtFW2fK2cOtBh1cHYkQsbzFV7I+EoRKe6Yt0oK7A==}
     engines: {node: '>=8'}
@@ -32852,6 +32862,11 @@ packages:
     engines: {node: '>=12.20'}
     dev: true
 
+  /yocto-queue@1.1.1:
+    resolution: {integrity: sha512-b4JR1PFR10y1mKjhHY9LaGo6tmrgjit7hxVIeAmyMw3jegXR4dhYqLaQF5zMXZxY7tLpMyJeLjr1C4rLmkVe8g==}
+    engines: {node: '>=12.20'}
+    dev: false
+
   /youch@3.3.3:
     resolution: {integrity: sha512-qSFXUk3UZBLfggAW3dJKg0BMblG5biqSF8M34E06o5CSsZtH92u9Hqmj2RzGiHDi64fhe83+4tENFP2DB6t6ZA==}
     dependencies:

references/hello-world/src/trigger/example.ts

@@ -68,3 +68,21 @@ export const maxDurationParentTask = task({
     return result;
   },
 });
+
+export const batchTask = task({
+  id: "batch",
+  run: async (payload: { count: number }, { ctx }) => {
+    logger.info("Starting batch task", { count: payload.count });
+
+    const items = Array.from({ length: payload.count }, (_, i) => ({
+      payload: { message: `Batch item ${i + 1}` },
+    }));
+
+    const results = await childTask.batchTriggerAndWait(items);
+
+    return {
+      batchCount: payload.count,
+      results,
+    };
+  },
+});

references/hello-world/src/trigger/oom.ts

@@ -0,0 +1,41 @@
+import { logger, task } from "@trigger.dev/sdk/v3";
+import { setTimeout } from "timers/promises";
+
+export const oomTask = task({
+  id: "oom-task",
+  machine: "micro",
+  retry: {
+    outOfMemory: {
+      machine: "small-1x",
+    },
+  },
+  run: async ({ succeedOnLargerMachine }: { succeedOnLargerMachine: boolean }, { ctx }) => {
+    logger.info("running out of memory below this line");
+
+    logger.info(`Running on ${ctx.machine?.name}`);
+
+    await setTimeout(2000);
+
+    if (ctx.machine?.name !== "micro" && succeedOnLargerMachine) {
+      logger.info("Going to succeed now");
+      return {
+        success: true,
+      };
+    }
+
+    let a = "a";
+
+    try {
+      while (true) {
+        a += a;
+      }
+    } catch (error) {
+      logger.error(error instanceof Error ? error.message : "Unknown error", { error });
+
+      let b = [];
+      while (true) {
+        b.push(a.replace(/a/g, "b"));
+      }
+    }
+  },
+});

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

@@ -126,12 +126,24 @@ export const allV2TestTask = task({
   retry: {
     maxAttempts: 1,
   },
-  run: async ({ triggerSequentially }: { triggerSequentially?: boolean }) => {
+  run: async ({ triggerSequentially }: { triggerSequentially?: boolean }, { ctx }) => {
     const response1 = await batch.trigger<typeof allV2ChildTask1 | typeof allV2ChildTask2>(
       [
-        { id: "all-v2-test-child-1", payload: { child1: "foo" } },
-        { id: "all-v2-test-child-2", payload: { child2: "bar" } },
-        { id: "all-v2-test-child-1", payload: { child1: "baz" } },
+        {
+          id: "all-v2-test-child-1",
+          payload: { child1: "foo" },
+          options: { idempotencyKey: randomUUID() },
+        },
+        {
+          id: "all-v2-test-child-2",
+          payload: { child2: "bar" },
+          options: { idempotencyKey: randomUUID() },
+        },
+        {
+          id: "all-v2-test-child-1",
+          payload: { child1: "baz" },
+          options: { idempotencyKey: randomUUID() },
+        },
       ],
       {
         triggerSequentially,

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

@@ -106,15 +106,33 @@ export const immediateReturn = task({
     console.info("some");
     console.warn("random");
     console.error("logs");
+
+    await new Promise((resolve) => setTimeout(resolve, 20000));
+  },
+});
+
+export const simulateErrorTester = task({
+  id: "simulateErrorTester",
+  run: async (payload: { message: string }) => {
+    await simulateError.batchTrigger([
+      { payload: { message: payload.message }, options: { maxAttempts: 1 } },
+      { payload: { message: payload.message }, options: { maxAttempts: 1 } },
+      { payload: { message: payload.message }, options: { maxAttempts: 1 } },
+      { payload: { message: payload.message }, options: { maxAttempts: 1 } },
+      { payload: { message: payload.message }, options: { maxAttempts: 1 } },
+      { payload: { message: payload.message }, options: { maxAttempts: 1 } },
+      { payload: { message: payload.message }, options: { maxAttempts: 1 } },
+      { payload: { message: payload.message }, options: { maxAttempts: 1 } },
+    ]);
   },
 });
 
 export const simulateError = task({
   id: "simulateError",
+  retry: {
+    maxAttempts: 1,
+  },
   run: async (payload: { message: string }) => {
-    // Sleep for 1 second
-    await new Promise((resolve) => setTimeout(resolve, 1000));
-
     thisFunctionWillThrow();
   },
 });