Job run performance improvements and adding "worker only" mode (#360)

* WIP job run performance improvements

- Added a `perf` tool to better measure job run performance under heavy load
- Removed `runFinished` job (not really needed)
- startQueuedRuns now uses a jobKey with replace
- Fixed an issue with ZodWorker when using jobKey

* Publish improvement docker images

* fixed the improvement docker publishing

* Downgrade back to prisma 4.16.0 because 5.1.x broke docker builds

* Changes to how queued runs work

- Split the worker into two different workers, one dedicated to performRunExecution
- Schedule performRunExecution in a single place, with a queue and using a round robin manually controlled concurrency
- Remove startQueuedRuns
- All runs are queued before they are started
- Setting the worker maxPoolSize to the same as the worker concurrency
- Starting to be able to split the docker image

* Remove queue name from startRun graphile job

* Make the prisma connection pool stuff configurable through env vars

* Hardcode (for now) the max concurrent runs limit

* Rewrite performRunExecution to be more performant

PerformRunExecutionV2:

- Does not create and manage jobRunExecution records
- Does not reimplement retrying, uses graphile worker retrying instead

I’ve kept around PerformRunExecutionV1 so this works when deploying. Definitely needs LOTS of testing

* Fix issues with cached tasks

- Limit the size of the cached tasks sent when executing a run, using the knapsack problem dynamic programming approach
- Actually USE the cached tasks in IO by using the idempotencyKey instead of the task ID
- Remove output from all logs
- Added a stress test job catalog

* Forgot to commit the logger updates

* Never log connectionString

* Login to docker hub to get around rate limits

* Add additional logging to the graphile workers

* Fix the *_ENABLED  env vars

* Allow adding and removing jobs to be done from the webapp

* Don’t set the job to failed if it’s being retried

* Deprecated queue options in the job and removed startPosition. Now using the job/env combo as the job queue name

* Dequeung jobs doesn’t check if the runner is initialized

* Fixed issues with retrying a run getting stuck on a cancelled task, and errors from parsing the results of dequeing a job

* Remove queued round robin thing that isn’t used anymore

* Added slack to job catalog

* Better forwards compat

* Added long delay

* Fixed lock file

.changeset/fuzzy-trees-lay.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+Fixed IO not setting the cached task key correctly, resulting in unnecessary API calls to trigger.dev

.changeset/happy-foxes-play.md

@@ -0,0 +1,6 @@
+---
+"@trigger.dev/sdk": patch
+"@trigger.dev/core": patch
+---
+
+Deprecated queue options in the job and removed startPosition

.github/workflows/publish.yml

@@ -4,6 +4,7 @@ on:
   push:
     branches:
       - main
+      - improvements/*
     tags:
       - "v.docker.*"
     paths:
@@ -95,6 +96,12 @@ jobs:
     name: e2e Tests
     runs-on: buildjet-4vcpu-ubuntu-2204
     steps:
+      - name: 🐳 Login to Docker Hub
+        uses: docker/login-action@v2
+        with:
+          username: ${{ secrets.DOCKERHUB_USERNAME }}
+          password: ${{ secrets.DOCKERHUB_TOKEN }}
+
       - name: ⬇️ Checkout repo
         uses: actions/checkout@v3
         with:
@@ -154,6 +161,11 @@ jobs:
       version: ${{ steps.get_version.outputs.version }}
       short_sha: ${{ steps.get_commit.outputs.sha_short }}
     steps:
+      - name: 🐳 Login to Docker Hub
+        uses: docker/login-action@v2
+        with:
+          username: ${{ secrets.DOCKERHUB_USERNAME }}
+          password: ${{ secrets.DOCKERHUB_TOKEN }}
       - name: ⬇️ Checkout repo
         uses: actions/checkout@v3
 
@@ -167,6 +179,10 @@ jobs:
               IMAGE_TAG="v${ORIGINAL_VERSION}"
             fi
             echo "IMAGE_TAG=${IMAGE_TAG}"
+          elif [[ $GITHUB_REF == refs/heads/improvements/* ]]; then
+            ORIGINAL_VERSION="${GITHUB_REF#refs/heads/improvements/}"
+            IMAGE_TAG="${ORIGINAL_VERSION}.rc"
+            echo "IMAGE_TAG=${IMAGE_TAG}"
           elif [[ $GITHUB_REF == refs/heads/* ]]; then
             IMAGE_TAG="${GITHUB_REF#refs/heads/}"
             echo "IMAGE_TAG=${IMAGE_TAG}"

apps/webapp/app/consts.ts

@@ -1,6 +1,7 @@
 export const LIVE_ENVIRONMENT = "live";
 export const DEV_ENVIRONMENT = "development";
 export const MAX_LIVE_PROJECTS = 1;
-export const DEFAULT_MAX_CONCURRENT_RUNS = 10000;
+export const DEFAULT_MAX_CONCURRENT_RUNS = 10;
+export const MAX_CONCURRENT_RUNS_LIMIT = 20;
 export const PREPROCESS_RETRY_LIMIT = 2;
 export const EXECUTE_JOB_RETRY_LIMIT = 10;

apps/webapp/app/db.server.ts

@@ -2,6 +2,7 @@ import { PrismaClient, Prisma } from "@trigger.dev/database";
 import invariant from "tiny-invariant";
 import { z } from "zod";
 import { logger } from "./services/logger.server";
+import { env } from "./env.server";
 
 export type PrismaTransactionClient = Omit<
   PrismaClient,
@@ -84,8 +85,16 @@ function getClient() {
   const { DATABASE_URL } = process.env;
   invariant(typeof DATABASE_URL === "string", "DATABASE_URL env var not set");
 
+  const databaseUrl = new URL(DATABASE_URL);
+
+  // We need to add the connection_limit and pool_timeout query params to the url, in a way that works if the DATABASE_URL already has query params
+  const query = databaseUrl.searchParams;
+  query.set("connection_limit", env.DATABASE_CONNECTION_LIMIT.toString());
+  query.set("pool_timeout", env.DATABASE_POOL_TIMEOUT.toString());
+  databaseUrl.search = query.toString();
+
   // Remove the username:password in the url and print that to the console
-  const urlWithoutCredentials = new URL(DATABASE_URL);
+  const urlWithoutCredentials = new URL(databaseUrl.href);
   urlWithoutCredentials.password = "";
 
   console.log(`🔌 setting up prisma client to ${urlWithoutCredentials.toString()}`);
@@ -93,8 +102,7 @@ function getClient() {
   const client = new PrismaClient({
     datasources: {
       db: {
-        url: DATABASE_URL,
-        // We can't set directUrl here, and we don't have to
+        url: databaseUrl.href,
       },
     },
     log: [

apps/webapp/app/env.server.ts

@@ -4,6 +4,8 @@ import { SecretStoreOptionsSchema } from "./services/secrets/secretStore.server"
 const EnvironmentSchema = z.object({
   NODE_ENV: z.union([z.literal("development"), z.literal("production"), z.literal("test")]),
   DATABASE_URL: z.string(),
+  DATABASE_CONNECTION_LIMIT: z.coerce.number().int().default(10),
+  DATABASE_POOL_TIMEOUT: z.coerce.number().int().default(60),
   DIRECT_URL: z.string(),
   SESSION_SECRET: z.string(),
   MAGIC_LINK_SECRET: z.string(),
@@ -31,6 +33,13 @@ const EnvironmentSchema = z.object({
   RESEND_API_KEY: z.string().optional(),
   PLAIN_API_KEY: z.string().optional(),
   RUNTIME_PLATFORM: z.enum(["docker-compose", "ecs", "local"]).default("local"),
+  WORKER_SCHEMA: z.string().default("graphile_worker"),
+  WORKER_CONCURRENCY: z.coerce.number().int().default(10),
+  WORKER_POLL_INTERVAL: z.coerce.number().int().default(1000),
+  EXECUTION_WORKER_CONCURRENCY: z.coerce.number().int().default(10),
+  EXECUTION_WORKER_POLL_INTERVAL: z.coerce.number().int().default(1000),
+  WORKER_ENABLED: z.string().default("true"),
+  EXECUTION_WORKER_ENABLED: z.string().default("true"),
 });
 
 export type Environment = z.infer<typeof EnvironmentSchema>;

apps/webapp/app/models/jobRunExecution.server.ts

@@ -0,0 +1,58 @@
+import { JobRun, JobRunExecution } from "@trigger.dev/database";
+import { PrismaClientOrTransaction } from "~/db.server";
+import { executionWorker } from "~/services/worker.server";
+
+export async function enqueueRunExecutionV1(
+  execution: JobRunExecution,
+  queueId: string,
+  concurrency: number,
+  tx: PrismaClientOrTransaction,
+  runAt?: Date
+) {
+  const job = await executionWorker.enqueue(
+    "performRunExecution",
+    {
+      id: execution.id,
+    },
+    {
+      queueName: `job:queue:${queueId}`,
+      tx,
+      runAt,
+      jobKey: `execution:${execution.runId}`,
+    }
+  );
+}
+
+export type EnqueueRunExecutionV2Options = {
+  runAt?: Date;
+  resumeTaskId?: string;
+  isRetry?: boolean;
+};
+
+export async function enqueueRunExecutionV2(
+  run: JobRun,
+  tx: PrismaClientOrTransaction,
+  options: EnqueueRunExecutionV2Options = {}
+) {
+  const job = await executionWorker.enqueue(
+    "performRunExecutionV2",
+    {
+      id: run.id,
+      reason: run.status === "PREPROCESSING" ? "PREPROCESS" : "EXECUTE_JOB",
+      resumeTaskId: options.resumeTaskId,
+      isRetry: typeof options.isRetry === "boolean" ? options.isRetry : false,
+    },
+    {
+      queueName: `job:${run.jobId}:env:${run.environmentId}`,
+      tx,
+      runAt: options.runAt,
+      jobKey: `job_run:${run.id}`,
+    }
+  );
+}
+
+export async function dequeueRunExecutionV2(run: JobRun, tx: PrismaClientOrTransaction) {
+  return await executionWorker.dequeue(`job_run:${run.id}`, {
+    tx,
+  });
+}

apps/webapp/app/platform/zodWorker.server.ts

@@ -51,6 +51,7 @@ const AddJobResultsSchema = z.array(GraphileJobSchema);
 export type ZodTasks<TConsumerSchema extends MessageCatalogSchema> = {
   [K in keyof TConsumerSchema]: {
     queueName?: string | ((payload: z.infer<TConsumerSchema[K]>) => string);
+    jobKey?: string | ((payload: z.infer<TConsumerSchema[K]>) => string | undefined);
     priority?: number;
     maxAttempts?: number;
     jobKeyMode?: "replace" | "preserve_run_at" | "unsafe_dedupe";
@@ -76,7 +77,12 @@ export type ZodWorkerEnqueueOptions = TaskSpec & {
   tx?: PrismaClientOrTransaction;
 };
 
+export type ZodWorkerDequeueOptions = {
+  tx?: PrismaClientOrTransaction;
+};
+
 export type ZodWorkerOptions<TMessageCatalog extends MessageCatalogSchema> = {
+  name: string;
   runnerOptions: RunnerOptions;
   prisma: PrismaClient;
   schema: TMessageCatalog;
@@ -85,6 +91,7 @@ export type ZodWorkerOptions<TMessageCatalog extends MessageCatalogSchema> = {
 };
 
 export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
+  #name: string;
   #schema: TMessageCatalog;
   #prisma: PrismaClient;
   #runnerOptions: RunnerOptions;
@@ -93,6 +100,7 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
   #runner?: GraphileRunner;
 
   constructor(options: ZodWorkerOptions<TMessageCatalog>) {
+    this.#name = options.name;
     this.#schema = options.schema;
     this.#prisma = options.prisma;
     this.#runnerOptions = options.runnerOptions;
@@ -105,7 +113,7 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
       return true;
     }
 
-    logger.debug("Initializing worker queue with options", {
+    this.#logDebug("Initializing worker queue with options", {
       runnerOptions: this.#runnerOptions,
     });
 
@@ -121,6 +129,54 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
       throw new Error("Failed to initialize worker queue");
     }
 
+    this.#runner?.events.on("pool:create", ({ workerPool }) => {
+      this.#logDebug("pool:create");
+    });
+
+    this.#runner?.events.on("pool:listen:connecting", ({ workerPool, attempts }) => {
+      this.#logDebug("pool:create", { attempts });
+    });
+
+    this.#runner?.events.on("pool:listen:success", ({ workerPool, client }) => {
+      this.#logDebug("pool:listen:success");
+    });
+
+    this.#runner?.events.on("pool:listen:error", ({ error }) => {
+      this.#logDebug("pool:listen:error", { error });
+    });
+
+    this.#runner?.events.on("pool:gracefulShutdown", ({ message }) => {
+      this.#logDebug("pool:gracefulShutdown", { workerMessage: message });
+    });
+
+    this.#runner?.events.on("pool:gracefulShutdown:error", ({ error }) => {
+      this.#logDebug("pool:gracefulShutdown:error", { error });
+    });
+
+    this.#runner?.events.on("worker:create", ({ worker }) => {
+      this.#logDebug("worker:create", { workerId: worker.workerId });
+    });
+
+    this.#runner?.events.on("worker:release", ({ worker }) => {
+      this.#logDebug("worker:release", { workerId: worker.workerId });
+    });
+
+    this.#runner?.events.on("worker:stop", ({ worker, error }) => {
+      this.#logDebug("worker:stop", { workerId: worker.workerId, error });
+    });
+
+    this.#runner?.events.on("worker:fatalError", ({ worker, error, jobError }) => {
+      this.#logDebug("worker:fatalError", { workerId: worker.workerId, error, jobError });
+    });
+
+    this.#runner?.events.on("gracefulShutdown", ({ signal }) => {
+      this.#logDebug("gracefulShutdown", { signal });
+    });
+
+    this.#runner?.events.on("stop", () => {
+      this.#logDebug("stop");
+    });
+
     return true;
   }
 
@@ -133,23 +189,34 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
     payload: z.infer<TMessageCatalog[K]>,
     options?: ZodWorkerEnqueueOptions
   ): Promise<GraphileJob> {
-    if (!this.#runner) {
-      throw new Error("Worker not initialized");
-    }
-
     const task = this.#tasks[identifier];
 
     const optionsWithoutTx = omit(options ?? {}, ["tx"]);
+    const taskWithoutJobKey = omit(task, ["jobKey"]);
 
     const spec = {
       ...optionsWithoutTx,
-      ...task,
+      ...taskWithoutJobKey,
     };
 
     if (typeof task.queueName === "function") {
       spec.queueName = task.queueName(payload);
     }
 
+    if (typeof task.jobKey === "function") {
+      const jobKey = task.jobKey(payload);
+
+      if (jobKey) {
+        spec.jobKey = jobKey;
+      }
+    }
+
+    logger.debug("Enqueuing worker task", {
+      identifier,
+      payload,
+      spec,
+    });
+
     const job = await this.#addJob(
       identifier as string,
       payload,
@@ -167,6 +234,17 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
     return job;
   }
 
+  public async dequeue(
+    jobKey: string,
+    option?: ZodWorkerDequeueOptions
+  ): Promise<GraphileJob | undefined> {
+    const results = await this.#removeJob(jobKey, option?.tx ?? this.#prisma);
+
+    logger.debug("dequeued worker task", { results, jobKey });
+
+    return results;
+  }
+
   async #addJob(
     identifier: string,
     payload: unknown,
@@ -192,8 +270,8 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
       spec.maxAttempts || null,
       spec.jobKey || null,
       spec.priority || null,
-      spec.jobKeyMode || null,
-      spec.flags || null
+      spec.flags || null,
+      spec.jobKeyMode || null
     );
 
     const rows = AddJobResultsSchema.safeParse(results);
@@ -209,6 +287,32 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
     return job as GraphileJob;
   }
 
+  async #removeJob(jobKey: string, tx: PrismaClientOrTransaction) {
+    try {
+      const result = await tx.$queryRawUnsafe(
+        `SELECT * FROM graphile_worker.remove_job(
+          job_key => $1::text
+        )`,
+        jobKey
+      );
+      const job = AddJobResultsSchema.safeParse(result);
+
+      if (!job.success) {
+        logger.debug("results returned from remove_job could not be parsed", {
+          error: job.error.flatten(),
+          result,
+          jobKey,
+        });
+
+        return;
+      }
+
+      return job.data[0] as GraphileJob;
+    } catch (e) {
+      throw new Error(`Failed to remove job from queue, ${e}}`);
+    }
+  }
+
   #createTaskListFromTasks() {
     const taskList: TaskList = {};
 
@@ -324,4 +428,8 @@ export class ZodWorker<TMessageCatalog extends MessageCatalogSchema> {
       throw error;
     }
   }
+
+  #logDebug(message: string, args?: any) {
+    logger.debug(`[worker][${this.#name}] ${message}`, args);
+  }
 }

apps/webapp/app/routes/api.v1.runs.$runId.tasks.ts

@@ -107,6 +107,32 @@ export class RunTaskService {
       });
 
       if (existingTask) {
+        if (existingTask.status === "CANCELED") {
+          const existingTaskStatus =
+            (taskBody.delayUntil && taskBody.delayUntil.getTime() > Date.now()) || taskBody.trigger
+              ? "WAITING"
+              : taskBody.noop
+              ? "COMPLETED"
+              : "RUNNING";
+
+          const resumedExistingTask = await tx.task.update({
+            where: {
+              id: existingTask.id,
+            },
+            data: {
+              status: existingTaskStatus,
+              startedAt: new Date(),
+              completedAt: existingTaskStatus === "COMPLETED" ? new Date() : undefined,
+            },
+            include: {
+              run: true,
+              attempts: true,
+            },
+          });
+
+          return resumedExistingTask;
+        }
+
         return existingTask;
       }
 

apps/webapp/app/routes/resources.runs.$runId.cancel.ts

@@ -2,6 +2,7 @@ import { parse } from "@conform-to/zod";
 import { ActionFunction, json } from "@remix-run/node";
 import { z } from "zod";
 import { redirectWithSuccessMessage } from "~/models/message.server";
+import { logger } from "~/services/logger.server";
 import { CancelRunService } from "~/services/runs/cancelRun.server";
 
 export const cancelSchema = z.object({
@@ -31,7 +32,19 @@ export const action: ActionFunction = async ({ request, params }) => {
       request,
       `Canceled run. Any pending tasks will be canceled.`
     );
-  } catch (error: any) {
-    return json({ errors: { body: error.message } }, { status: 400 });
+  } catch (error) {
+    if (error instanceof Error) {
+      logger.error("Failed to cancel run", {
+        error: {
+          name: error.name,
+          message: error.message,
+          stack: error.stack,
+        },
+      });
+      return json({ errors: { body: error.message } }, { status: 400 });
+    } else {
+      logger.error("Failed to cancel run", { error });
+      return json({ errors: { body: "Unknown error" } }, { status: 400 });
+    }
   }
 };

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

@@ -162,6 +162,10 @@ export class EndpointApi {
   }
 
   async executeJobRequest(options: RunJobBody) {
+    logger.debug("executeJobRequest()", {
+      options,
+    });
+
     const response = await safeFetch(this.url, {
       method: "POST",
       headers: {

apps/webapp/app/services/externalApis/integrationConnectionCreated.server.ts

@@ -109,15 +109,9 @@ export class IntegrationConnectionCreatedService {
       });
 
       // We need to start the run again
-      await workerQueue.enqueue(
-        "startRun",
-        {
-          id: run.id,
-        },
-        {
-          queueName: `job-queue:${run.queue.id}`,
-        }
-      );
+      await workerQueue.enqueue("startRun", {
+        id: run.id,
+      });
     }
   }
 }

apps/webapp/app/services/jobs/registerJob.server.ts

@@ -1,18 +1,18 @@
-import type { Endpoint, Integration, Job, JobIntegration, JobVersion } from "@trigger.dev/database";
 import {
   IntegrationConfig,
   JobMetadata,
   SCHEDULED_EVENT,
   TriggerMetadata,
 } from "@trigger.dev/core";
+import type { Endpoint, Integration, Job, JobIntegration, JobVersion } from "@trigger.dev/database";
 import { DEFAULT_MAX_CONCURRENT_RUNS } from "~/consts";
 import type { PrismaClient } from "~/db.server";
 import { prisma } from "~/db.server";
+import { ExtendedEndpoint, findEndpoint } from "~/models/endpoint.server";
+import type { RuntimeEnvironment } from "~/models/runtimeEnvironment.server";
 import type { AuthenticatedEnvironment } from "../apiAuth.server";
 import { logger } from "../logger.server";
-import type { RuntimeEnvironment } from "~/models/runtimeEnvironment.server";
 import { RegisterScheduleSourceService } from "../schedules/registerScheduleSource.server";
-import { ExtendedEndpoint, findEndpoint } from "~/models/endpoint.server";
 
 export class RegisterJobService {
   #prismaClient: PrismaClient;
@@ -166,13 +166,9 @@ export class RegisterJobService {
     });
 
     // Upsert the JobQueue
-    const queueName =
-      typeof metadata.queue === "string"
-        ? metadata.queue
-        : typeof metadata.queue === "object"
-        ? metadata.queue.name
-        : "default";
+    const queueName = "default";
 
+    // Job Queues are going to be deprecated or used for something else, we're just doing this for now
     const jobQueue = await this.#prismaClient.jobQueue.upsert({
       where: {
         environmentId_name: {
@@ -187,16 +183,10 @@ export class RegisterJobService {
           },
         },
         name: queueName,
-        maxJobs:
-          typeof metadata.queue === "object"
-            ? metadata.queue.maxConcurrent || DEFAULT_MAX_CONCURRENT_RUNS
-            : DEFAULT_MAX_CONCURRENT_RUNS,
+        maxJobs: DEFAULT_MAX_CONCURRENT_RUNS,
       },
       update: {
-        maxJobs:
-          typeof metadata.queue === "object"
-            ? metadata.queue.maxConcurrent || DEFAULT_MAX_CONCURRENT_RUNS
-            : DEFAULT_MAX_CONCURRENT_RUNS,
+        maxJobs: DEFAULT_MAX_CONCURRENT_RUNS,
       },
     });
 
@@ -245,10 +235,10 @@ export class RegisterJobService {
         version: metadata.version,
         eventSpecification,
         preprocessRuns: metadata.preprocessRuns,
-        startPosition: metadata.startPosition === "initial" ? "INITIAL" : "LATEST",
+        startPosition: "LATEST",
       },
       update: {
-        startPosition: metadata.startPosition === "initial" ? "INITIAL" : "LATEST",
+        startPosition: "LATEST",
         eventSpecification,
         preprocessRuns: metadata.preprocessRuns,
         queue: {

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

@@ -5,6 +5,6 @@ import { sensitiveDataReplacer } from "./sensitiveDataReplacer";
 export const logger = new Logger(
   "webapp",
   (process.env.APP_LOG_LEVEL ?? "debug") as LogLevel,
-  ["examples"],
+  ["examples", "output", "connectionString", "payload"],
   sensitiveDataReplacer
 );

apps/webapp/app/services/runs/cancelRun.server.ts

@@ -1,5 +1,6 @@
 import { PrismaClient, prisma } from "~/db.server";
-import { workerQueue } from "../worker.server";
+import { executionWorker } from "../worker.server";
+import { dequeueRunExecutionV2 } from "~/models/jobRunExecution.server";
 
 export class CancelRunService {
   #prismaClient: PrismaClient;
@@ -17,21 +18,11 @@ export class CancelRunService {
           },
         });
 
-        const shouldDecrementQueue = run.status === "STARTED" || run.status === "PREPROCESSING";
         await tx.jobRun.update({
           where: { id: runId },
           data: {
             status: "CANCELED",
             completedAt: new Date(),
-            queue: shouldDecrementQueue
-              ? {
-                  update: {
-                    jobCount: {
-                      decrement: 1,
-                    },
-                  },
-                }
-              : undefined,
           },
         });
 
@@ -48,13 +39,7 @@ export class CancelRunService {
           },
         });
 
-        await workerQueue.enqueue(
-          "startQueuedRuns",
-          {
-            id: run.queueId,
-          },
-          { tx }
-        );
+        await dequeueRunExecutionV2(run, tx);
       });
     } catch (error) {
       throw error;

apps/webapp/app/services/runs/continueRun.server.ts

@@ -1,6 +1,5 @@
-import { EXECUTE_JOB_RETRY_LIMIT } from "~/consts";
 import { $transaction, Prisma, PrismaClient, prisma } from "~/db.server";
-import { workerQueue } from "../worker.server";
+import { enqueueRunExecutionV2 } from "~/models/jobRunExecution.server";
 
 const RESUMABLE_STATUSES = ["FAILURE", "TIMED_OUT", "ABORTED", "CANCELED"];
 
@@ -17,86 +16,26 @@ export class ContinueRunService {
       async (tx) => {
         const run = await tx.jobRun.findUniqueOrThrow({
           where: { id: runId },
-          include: {
-            queue: true,
-          },
         });
 
         if (!RESUMABLE_STATUSES.includes(run.status)) {
           throw new Error("Run is not resumable");
         }
 
-        if (run.queue.jobCount >= run.queue.maxJobs) {
-          await tx.jobRun.update({
-            where: { id: runId },
-            data: {
-              status: "QUEUED",
-              queuedAt: new Date(),
-              startedAt: null,
-              completedAt: null,
-              output: Prisma.DbNull,
-              timedOutAt: null,
-              timedOutReason: null,
-            },
-          });
-        } else {
-          await tx.jobRun.update({
-            where: { id: runId },
-            data: {
-              status: "STARTED",
-              queuedAt: null,
-              startedAt: new Date(),
-              completedAt: null,
-              output: Prisma.DbNull,
-              timedOutAt: null,
-              timedOutReason: null,
-              queue: {
-                update: {
-                  jobCount: {
-                    increment: 1,
-                  },
-                },
-              },
-            },
-          });
+        await tx.jobRun.update({
+          where: { id: runId },
+          data: {
+            status: "QUEUED",
+            queuedAt: new Date(),
+            startedAt: null,
+            completedAt: null,
+            output: Prisma.DbNull,
+            timedOutAt: null,
+            timedOutReason: null,
+          },
+        });
 
-          const execution = await tx.jobRunExecution.create({
-            data: {
-              run: {
-                connect: {
-                  id: runId,
-                },
-              },
-              status: "PENDING",
-              reason: "EXECUTE_JOB",
-              retryLimit: EXECUTE_JOB_RETRY_LIMIT,
-              isRetry: true,
-            },
-          });
-
-          const job = await workerQueue.enqueue(
-            "performRunExecution",
-            {
-              id: execution.id,
-            },
-            { tx }
-          );
-
-          await tx.jobRunExecution.update({
-            where: { id: execution.id },
-            data: {
-              graphileJobId: job.id,
-            },
-          });
-
-          await workerQueue.enqueue(
-            "startQueuedRuns",
-            {
-              id: run.queueId,
-            },
-            { tx }
-          );
-        }
+        await enqueueRunExecutionV2(run, tx);
       },
       { timeout: 10000 }
     );

apps/webapp/app/services/runs/createRun.server.ts

@@ -75,7 +75,7 @@ export class CreateRunService {
         {
           id: run.id,
         },
-        { tx, queueName: `job-queue:${jobQueue.id}` }
+        { tx }
       );
 
       return run;

apps/webapp/app/services/runs/performRunExecutionV1.server.ts

@@ -1,27 +1,26 @@
-import type { Task } from "@trigger.dev/database";
 import {
   ApiEventLogSchema,
   CachedTaskSchema,
-  RunJobCanceledWithTask,
   RunJobError,
   RunJobResumeWithTask,
   RunJobRetryWithTask,
   RunJobSuccess,
   RunSourceContextSchema,
 } from "@trigger.dev/core";
+import type { Task } from "@trigger.dev/database";
 import { generateErrorMessage } from "zod-error";
 import { EXECUTE_JOB_RETRY_LIMIT } from "~/consts";
 import { $transaction, PrismaClient, PrismaClientOrTransaction, prisma } from "~/db.server";
+import { enqueueRunExecutionV1 } from "~/models/jobRunExecution.server";
 import { resolveRunConnections } from "~/models/runConnection.server";
+import { formatError } from "~/utils/formatErrors.server";
 import { safeJsonZodParse } from "~/utils/json";
 import { EndpointApi } from "../endpointApi.server";
-import { workerQueue } from "../worker.server";
-import { formatError } from "~/utils/formatErrors.server";
 import { logger } from "../logger.server";
 
 type FoundRunExecution = NonNullable<Awaited<ReturnType<typeof findRunExecution>>>;
 
-export class PerformRunExecutionService {
+export class PerformRunExecutionV1Service {
   #prismaClient: PrismaClient;
 
   constructor(prismaClient: PrismaClient = prisma) {
@@ -162,22 +161,7 @@ export class PerformRunExecutionService {
           },
         });
 
-        const job = await workerQueue.enqueue(
-          "performRunExecution",
-          {
-            id: runExecution.id,
-          },
-          { tx }
-        );
-
-        await tx.jobRunExecution.update({
-          where: {
-            id: runExecution.id,
-          },
-          data: {
-            graphileJobId: job.id,
-          },
-        });
+        await enqueueRunExecutionV1(runExecution, run.queue.id, run.queue.maxJobs, tx);
       });
     }
   }
@@ -201,6 +185,12 @@ export class PerformRunExecutionService {
       data: {
         status: "STARTED",
         startedAt,
+        run: {
+          update: {
+            status: run.status === "QUEUED" ? "STARTED" : run.status,
+            startedAt: run.startedAt ?? new Date(),
+          },
+        },
       },
     });
 
@@ -388,14 +378,6 @@ export class PerformRunExecutionService {
           completedAt: new Date(),
         },
       });
-
-      await workerQueue.enqueue(
-        "runFinished",
-        {
-          id: run.id,
-        },
-        { tx }
-      );
     });
   }
 
@@ -426,22 +408,13 @@ export class PerformRunExecutionService {
           },
         });
 
-        const graphileJob = await workerQueue.enqueue(
-          "performRunExecution",
-          {
-            id: newJobExecution.id,
-          },
-          { tx, runAt: data.task.delayUntil ?? undefined }
+        await enqueueRunExecutionV1(
+          newJobExecution,
+          run.queue.id,
+          run.queue.maxJobs,
+          tx,
+          data.task.delayUntil ?? undefined
         );
-
-        await tx.jobRunExecution.update({
-          where: {
-            id: newJobExecution.id,
-          },
-          data: {
-            graphileJobId: graphileJob.id,
-          },
-        });
       }
     });
   }
@@ -522,22 +495,13 @@ export class PerformRunExecutionService {
         },
       });
 
-      const graphileJob = await workerQueue.enqueue(
-        "performRunExecution",
-        {
-          id: newJobExecution.id,
-        },
-        { tx, runAt: data.retryAt }
+      await enqueueRunExecutionV1(
+        newJobExecution,
+        run.queue.id,
+        run.queue.maxJobs,
+        tx,
+        data.retryAt
       );
-
-      await tx.jobRunExecution.update({
-        where: {
-          id: newJobExecution.id,
-        },
-        data: {
-          graphileJobId: graphileJob.id,
-        },
-      });
     });
   }
 
@@ -557,6 +521,13 @@ export class PerformRunExecutionService {
       // So when retryCount is 1, retryDelayInMs is 500ms
       // When retryCount is 2, retryDelayInMs is 750ms
       // When retryCount is 3, retryDelayInMs is 1125ms
+      // When retryCount is 4, retryDelayInMs is 1687ms
+      // When retryCount is 5, retryDelayInMs is 2531ms
+      // When retryCount is 6, retryDelayInMs is 3796ms
+      // When retryCount is 7, retryDelayInMs is 5694ms
+      // When retryCount is 8, retryDelayInMs is 8541ms
+      // When retryCount is 9, retryDelayInMs is 12812ms
+      // When retryCount is 10, retryDelayInMs is 19218ms
       const retryDelayInMs = Math.round(500 * Math.pow(1.5, retryCount - 1));
 
       await tx.jobRunExecution.update({
@@ -572,20 +543,13 @@ export class PerformRunExecutionService {
 
       const runAt = new Date(Date.now() + retryDelayInMs);
 
-      const job = await workerQueue.enqueue(
-        "performRunExecution",
-        { id: execution.id },
-        { runAt, tx }
+      await enqueueRunExecutionV1(
+        execution,
+        execution.run.queue.id,
+        execution.run.queue.maxJobs,
+        tx,
+        runAt
       );
-
-      await tx.jobRunExecution.update({
-        where: {
-          id: execution.id,
-        },
-        data: {
-          graphileJobId: job.id,
-        },
-      });
     });
   }
 
@@ -617,13 +581,6 @@ export class PerformRunExecutionService {
             },
           });
 
-          await workerQueue.enqueue(
-            "runFinished",
-            {
-              id: run.id,
-            },
-            { tx }
-          );
           break;
         }
         case "PREPROCESS": {
@@ -645,14 +602,6 @@ export class PerformRunExecutionService {
               },
             });
 
-            await workerQueue.enqueue(
-              "runFinished",
-              {
-                id: run.id,
-              },
-              { tx }
-            );
-
             break;
           }
 
@@ -675,22 +624,7 @@ export class PerformRunExecutionService {
             },
           });
 
-          const job = await workerQueue.enqueue(
-            "performRunExecution",
-            {
-              id: runExecution.id,
-            },
-            { tx }
-          );
-
-          await tx.jobRunExecution.update({
-            where: {
-              id: runExecution.id,
-            },
-            data: {
-              graphileJobId: job.id,
-            },
-          });
+          await enqueueRunExecutionV1(runExecution, run.queue.id, run.queue.maxJobs, tx);
 
           break;
         }
@@ -733,6 +667,7 @@ async function findRunExecution(prisma: PrismaClientOrTransaction, id: string) {
           endpoint: true,
           organization: true,
           externalAccount: true,
+          queue: true,
           runConnections: {
             include: {
               integration: true,

apps/webapp/app/services/runs/performRunExecutionV2.server.ts

@@ -0,0 +1,593 @@
+import {
+  ApiEventLogSchema,
+  CachedTask,
+  RunJobError,
+  RunJobResumeWithTask,
+  RunJobRetryWithTask,
+  RunJobSuccess,
+  RunSourceContextSchema,
+} from "@trigger.dev/core";
+import type { Task } from "@trigger.dev/database";
+import { generateErrorMessage } from "zod-error";
+import { $transaction, PrismaClient, PrismaClientOrTransaction, prisma } from "~/db.server";
+import { enqueueRunExecutionV2 } from "~/models/jobRunExecution.server";
+import { resolveRunConnections } from "~/models/runConnection.server";
+import { formatError } from "~/utils/formatErrors.server";
+import { safeJsonZodParse } from "~/utils/json";
+import { EndpointApi } from "../endpointApi.server";
+import { logger } from "../logger.server";
+
+type FoundRun = NonNullable<Awaited<ReturnType<typeof findRun>>>;
+type FoundTask = FoundRun["tasks"][number];
+
+export class PerformRunExecutionV2Service {
+  #prismaClient: PrismaClient;
+
+  constructor(prismaClient: PrismaClient = prisma) {
+    this.#prismaClient = prismaClient;
+  }
+
+  public async call(
+    id: string,
+    reason: "PREPROCESS" | "EXECUTE_JOB",
+    isRetry: boolean = false,
+    resumeTaskId?: string
+  ) {
+    const run = await findRun(this.#prismaClient, id);
+
+    if (!run) {
+      return;
+    }
+
+    switch (reason) {
+      case "PREPROCESS": {
+        await this.#executePreprocessing(run);
+        break;
+      }
+      case "EXECUTE_JOB": {
+        await this.#executeJob(run, isRetry, resumeTaskId);
+        break;
+      }
+    }
+  }
+
+  // Execute the preprocessing step of a run, which will send the payload to the endpoint and give the job
+  // an opportunity to generate run properties based on the payload.
+  // If the endpoint is not available, or the response is not ok,
+  // the run execution will be marked as failed and the run will start
+  async #executePreprocessing(run: FoundRun) {
+    const client = new EndpointApi(run.environment.apiKey, run.endpoint.url);
+    const event = ApiEventLogSchema.parse({ ...run.event, id: run.eventId });
+
+    const { response, parser } = await client.preprocessRunRequest({
+      event,
+      job: {
+        id: run.version.job.slug,
+        version: run.version.version,
+      },
+      run: {
+        id: run.id,
+        isTest: run.isTest,
+      },
+      environment: {
+        id: run.environment.id,
+        slug: run.environment.slug,
+        type: run.environment.type,
+      },
+      organization: {
+        id: run.organization.id,
+        slug: run.organization.slug,
+        title: run.organization.title,
+      },
+      account: run.externalAccount
+        ? {
+            id: run.externalAccount.identifier,
+            metadata: run.externalAccount.metadata,
+          }
+        : undefined,
+    });
+
+    if (!response) {
+      return await this.#failRunExecution(this.#prismaClient, "PREPROCESS", run, {
+        message: "Could not connect to the endpoint",
+      });
+    }
+
+    if (!response.ok) {
+      return await this.#failRunExecution(this.#prismaClient, "PREPROCESS", run, {
+        message: `Endpoint responded with ${response.status} status code`,
+      });
+    }
+
+    const rawBody = await response.text();
+    const safeBody = safeJsonZodParse(parser, rawBody);
+
+    if (!safeBody) {
+      return await this.#failRunExecution(this.#prismaClient, "PREPROCESS", run, {
+        message: "Endpoint responded with invalid JSON",
+      });
+    }
+
+    if (!safeBody.success) {
+      return await this.#failRunExecution(this.#prismaClient, "PREPROCESS", run, {
+        message: generateErrorMessage(safeBody.error.issues),
+      });
+    }
+
+    if (safeBody.data.abort) {
+      return this.#failRunExecution(
+        this.#prismaClient,
+        "PREPROCESS",
+        run,
+        { message: "Endpoint aborted the run" },
+        "ABORTED"
+      );
+    } else {
+      await $transaction(this.#prismaClient, async (tx) => {
+        await tx.jobRun.update({
+          where: {
+            id: run.id,
+          },
+          data: {
+            status: "STARTED",
+            startedAt: new Date(),
+            properties: safeBody.data.properties,
+          },
+        });
+
+        await enqueueRunExecutionV2(run, tx);
+      });
+    }
+  }
+  async #executeJob(run: FoundRun, isRetry: boolean, resumeTaskId?: string) {
+    if (run.status === "CANCELED") {
+      await this.#cancelExecution(run);
+      return;
+    }
+
+    const client = new EndpointApi(run.environment.apiKey, run.endpoint.url);
+    const event = ApiEventLogSchema.parse({ ...run.event, id: run.eventId });
+
+    const startedAt = new Date();
+
+    await this.#prismaClient.jobRun.update({
+      where: {
+        id: run.id,
+      },
+      data: {
+        status: run.status === "QUEUED" ? "STARTED" : run.status,
+        startedAt: run.startedAt ?? new Date(),
+      },
+    });
+
+    const connections = await resolveRunConnections(run.runConnections);
+
+    if (!connections.success) {
+      return this.#failRunExecutionWithRetry({
+        message: `Could not resolve all connections for run ${run.id}, attempting to retry`,
+      });
+    }
+
+    let resumedTask: Task | undefined;
+
+    if (resumeTaskId) {
+      resumedTask =
+        (await this.#prismaClient.task.findUnique({
+          where: {
+            id: resumeTaskId,
+          },
+        })) ?? undefined;
+
+      if (resumedTask) {
+        resumedTask = await this.#prismaClient.task.update({
+          where: {
+            id: resumeTaskId,
+          },
+          data: {
+            status: resumedTask.noop ? "COMPLETED" : "RUNNING",
+            completedAt: resumedTask.noop ? new Date() : undefined,
+          },
+        });
+      }
+    }
+
+    const sourceContext = RunSourceContextSchema.safeParse(run.event.sourceContext);
+
+    const { response, parser, errorParser } = await client.executeJobRequest({
+      event,
+      job: {
+        id: run.version.job.slug,
+        version: run.version.version,
+      },
+      run: {
+        id: run.id,
+        isTest: run.isTest,
+        startedAt,
+        isRetry,
+      },
+      environment: {
+        id: run.environment.id,
+        slug: run.environment.slug,
+        type: run.environment.type,
+      },
+      organization: {
+        id: run.organization.id,
+        slug: run.organization.slug,
+        title: run.organization.title,
+      },
+      account: run.externalAccount
+        ? {
+            id: run.externalAccount.identifier,
+            metadata: run.externalAccount.metadata,
+          }
+        : undefined,
+      connections: connections.auth,
+      source: sourceContext.success ? sourceContext.data : undefined,
+      tasks: prepareTasksForRun([run.tasks, resumedTask].flat().filter(Boolean)),
+    });
+
+    if (!response) {
+      return await this.#failRunExecutionWithRetry({
+        message: `Connection could not be established to the endpoint (${run.endpoint.url})`,
+      });
+    }
+
+    const rawBody = await response.text();
+
+    if (!response.ok) {
+      logger.debug("Endpoint responded with non-200 status code", {
+        status: response.status,
+        runId: run.id,
+        endpoint: run.endpoint.url,
+      });
+
+      const errorBody = safeJsonZodParse(errorParser, rawBody);
+
+      if (errorBody && errorBody.success) {
+        // Only retry if the error isn't a 4xx
+        if (response.status >= 400 && response.status <= 499) {
+          return await this.#failRunExecution(
+            this.#prismaClient,
+            "EXECUTE_JOB",
+            run,
+            errorBody.data
+          );
+        } else {
+          return await this.#failRunExecutionWithRetry(errorBody.data);
+        }
+      }
+
+      // Only retry if the error isn't a 4xx
+      if (response.status >= 400 && response.status <= 499 && response.status !== 408) {
+        return await this.#failRunExecution(this.#prismaClient, "EXECUTE_JOB", run, {
+          message: `Endpoint responded with ${response.status} status code`,
+        });
+      } else {
+        return await this.#failRunExecutionWithRetry({
+          message: `Endpoint responded with ${response.status} status code`,
+        });
+      }
+    }
+
+    const safeBody = safeJsonZodParse(parser, rawBody);
+
+    if (!safeBody) {
+      return await this.#failRunExecution(this.#prismaClient, "EXECUTE_JOB", run, {
+        message: "Endpoint responded with invalid JSON",
+      });
+    }
+
+    if (!safeBody.success) {
+      return await this.#failRunExecution(this.#prismaClient, "EXECUTE_JOB", run, {
+        message: generateErrorMessage(safeBody.error.issues),
+      });
+    }
+
+    const status = safeBody.data.status;
+
+    switch (status) {
+      case "SUCCESS": {
+        await this.#completeRunWithSuccess(run, safeBody.data);
+
+        break;
+      }
+      case "RESUME_WITH_TASK": {
+        await this.#resumeRunWithTask(run, safeBody.data, isRetry);
+
+        break;
+      }
+      case "ERROR": {
+        await this.#failRunWithError(run, safeBody.data);
+
+        break;
+      }
+      case "RETRY_WITH_TASK": {
+        await this.#retryRunWithTask(run, safeBody.data, isRetry);
+
+        break;
+      }
+      case "CANCELED": {
+        await this.#cancelExecution(run);
+        break;
+      }
+      default: {
+        const _exhaustiveCheck: never = status;
+        throw new Error(`Non-exhaustive match for value: ${status}`);
+      }
+    }
+  }
+
+  async #completeRunWithSuccess(run: FoundRun, data: RunJobSuccess) {
+    await this.#prismaClient.jobRun.update({
+      where: { id: run.id },
+      data: {
+        completedAt: new Date(),
+        status: "SUCCESS",
+        output: data.output ?? undefined,
+      },
+    });
+  }
+
+  async #resumeRunWithTask(run: FoundRun, data: RunJobResumeWithTask, isRetry: boolean) {
+    return await $transaction(this.#prismaClient, async (tx) => {
+      // If the task has an operation, then the next performRunExecution will occur
+      // when that operation has finished
+      if (!data.task.operation) {
+        await enqueueRunExecutionV2(run, tx, {
+          runAt: data.task.delayUntil ?? undefined,
+          resumeTaskId: data.task.id,
+          isRetry,
+        });
+      }
+    });
+  }
+
+  async #failRunWithError(execution: FoundRun, data: RunJobError) {
+    return await $transaction(this.#prismaClient, async (tx) => {
+      if (data.task) {
+        await tx.task.update({
+          where: {
+            id: data.task.id,
+          },
+          data: {
+            status: "ERRORED",
+            completedAt: new Date(),
+            output: data.error ?? undefined,
+          },
+        });
+      }
+
+      await this.#failRunExecution(tx, "EXECUTE_JOB", execution, data.error ?? undefined);
+    });
+  }
+
+  async #retryRunWithTask(run: FoundRun, data: RunJobRetryWithTask, isRetry: boolean) {
+    return await $transaction(this.#prismaClient, async (tx) => {
+      // We need to check for an existing task attempt
+      const existingAttempt = await tx.taskAttempt.findFirst({
+        where: {
+          taskId: data.task.id,
+          status: "PENDING",
+        },
+        orderBy: {
+          number: "desc",
+        },
+      });
+
+      if (existingAttempt) {
+        await tx.taskAttempt.update({
+          where: {
+            id: existingAttempt.id,
+          },
+          data: {
+            status: "ERRORED",
+            error: formatError(data.error),
+          },
+        });
+      }
+
+      // We need to create a new task attempt
+      await tx.taskAttempt.create({
+        data: {
+          taskId: data.task.id,
+          number: existingAttempt ? existingAttempt.number + 1 : 1,
+          status: "PENDING",
+          runAt: data.retryAt,
+        },
+      });
+
+      await tx.task.update({
+        where: {
+          id: data.task.id,
+        },
+        data: {
+          status: "WAITING",
+        },
+      });
+
+      await enqueueRunExecutionV2(run, tx, {
+        runAt: data.retryAt,
+        resumeTaskId: data.task.id,
+        isRetry,
+      });
+    });
+  }
+
+  async #failRunExecutionWithRetry(output: Record<string, any>): Promise<void> {
+    throw new Error(JSON.stringify(output));
+  }
+
+  async #failRunExecution(
+    prisma: PrismaClientOrTransaction,
+    reason: "EXECUTE_JOB" | "PREPROCESS",
+    run: FoundRun,
+    output: Record<string, any>,
+    status: "FAILURE" | "ABORTED" = "FAILURE"
+  ): Promise<void> {
+    await $transaction(prisma, async (tx) => {
+      switch (reason) {
+        case "EXECUTE_JOB": {
+          // If the execution is an EXECUTE_JOB reason, we need to fail the run
+          await tx.jobRun.update({
+            where: { id: run.id },
+            data: {
+              completedAt: new Date(),
+              status,
+              output,
+            },
+          });
+
+          break;
+        }
+        case "PREPROCESS": {
+          // If the status is ABORTED, we need to fail the run
+          if (status === "ABORTED") {
+            await tx.jobRun.update({
+              where: { id: run.id },
+              data: {
+                completedAt: new Date(),
+                status,
+                output,
+              },
+            });
+
+            break;
+          }
+
+          await tx.jobRun.update({
+            where: {
+              id: run.id,
+            },
+            data: {
+              status: "STARTED",
+              startedAt: new Date(),
+            },
+          });
+
+          await enqueueRunExecutionV2(run, tx);
+
+          break;
+        }
+      }
+    });
+  }
+
+  async #cancelExecution(run: FoundRun) {
+    return;
+  }
+}
+
+function prepareTasksForRun(tasks: FoundTask[]): CachedTask[] {
+  // We need to limit the cached tasks to not be too large >3.5MB when serialized
+  const TOTAL_CACHED_TASK_BYTE_LIMIT = 3500000;
+
+  const cachedTasks = new Map<string, CachedTask>(); // Cache for prepared tasks
+  const cachedTaskSizes = new Map<string, number>(); // Cache for calculated task sizes
+
+  // Helper function to get the cached prepared task, or prepare and cache if not already cached
+  function getCachedTask(task: FoundTask): CachedTask {
+    const taskId = task.id;
+    if (!cachedTasks.has(taskId)) {
+      cachedTasks.set(taskId, prepareTaskForRun(task));
+    }
+    return cachedTasks.get(taskId)!;
+  }
+
+  // Helper function to get the cached task size, or calculate and cache if not already cached
+  function getCachedTaskSize(task: CachedTask): number {
+    const taskId = task.id;
+    if (!cachedTaskSizes.has(taskId)) {
+      cachedTaskSizes.set(taskId, calculateCachedTaskSize(task));
+    }
+    return cachedTaskSizes.get(taskId)!;
+  }
+
+  // Create a dynamic programming array to store intermediate results
+  const dp: number[][] = [];
+  for (let i = 0; i <= tasks.length; i++) {
+    dp[i] = [];
+    for (let j = 0; j <= TOTAL_CACHED_TASK_BYTE_LIMIT; j++) {
+      dp[i][j] = 0;
+    }
+  }
+
+  // Fill the dynamic programming array
+  for (let i = 1; i <= tasks.length; i++) {
+    const task = tasks[i - 1];
+    const cachedTask = getCachedTask(task);
+    const taskSize = getCachedTaskSize(cachedTask);
+    for (let j = 0; j <= TOTAL_CACHED_TASK_BYTE_LIMIT; j++) {
+      if (taskSize <= j) {
+        dp[i][j] = Math.max(dp[i - 1][j], dp[i - 1][j - taskSize] + taskSize);
+      } else {
+        dp[i][j] = dp[i - 1][j];
+      }
+    }
+  }
+
+  // Traverse the dynamic programming array to find the included tasks
+  const tasksToRun: CachedTask[] = [];
+  let j = TOTAL_CACHED_TASK_BYTE_LIMIT;
+  for (let i = tasks.length; i > 0 && j > 0; i--) {
+    if (dp[i][j] !== dp[i - 1][j]) {
+      const task = tasks[i - 1];
+      const cachedTask = getCachedTask(task);
+      tasksToRun.unshift(cachedTask);
+      j -= getCachedTaskSize(cachedTask);
+    }
+  }
+
+  return tasksToRun;
+}
+
+function prepareTaskForRun(task: FoundTask): CachedTask {
+  return {
+    id: task.id,
+    status: task.status,
+    idempotencyKey: task.idempotencyKey,
+    noop: task.noop,
+    output: task.output as any,
+    parentId: task.parentId,
+  };
+}
+
+function calculateCachedTaskSize(task: CachedTask): number {
+  return JSON.stringify(task).length;
+}
+
+async function findRun(prisma: PrismaClientOrTransaction, id: string) {
+  return await prisma.jobRun.findUnique({
+    where: { id },
+    include: {
+      environment: true,
+      endpoint: true,
+      organization: true,
+      externalAccount: true,
+      queue: true,
+      runConnections: {
+        include: {
+          integration: true,
+          connection: {
+            include: {
+              dataReference: true,
+            },
+          },
+        },
+      },
+      tasks: {
+        where: {
+          status: {
+            in: ["COMPLETED"],
+          },
+        },
+      },
+      event: true,
+      version: {
+        include: {
+          job: true,
+          organization: true,
+        },
+      },
+    },
+  });
+}

apps/webapp/app/services/runs/runFinished.server.ts

@@ -1,56 +0,0 @@
-import type { PrismaClient } from "~/db.server";
-import { prisma } from "~/db.server";
-import { workerQueue } from "../worker.server";
-import { z } from "zod";
-import { RawEventSchema, SendEventOptionsSchema } from "@trigger.dev/core";
-import { IngestSendEvent } from "../events/ingestSendEvent.server";
-
-const SendEventOutputSchema = z.object({
-  events: z.array(RawEventSchema),
-  options: SendEventOptionsSchema.optional(),
-});
-
-export class RunFinishedService {
-  #prismaClient: PrismaClient;
-  #ingestEventService = new IngestSendEvent();
-
-  constructor(prismaClient: PrismaClient = prisma) {
-    this.#prismaClient = prismaClient;
-  }
-
-  public async call(id: string) {
-    const run = await this.#prismaClient.jobRun.findUniqueOrThrow({
-      where: { id },
-      include: {
-        environment: {
-          include: {
-            project: true,
-            organization: true,
-          },
-        },
-      },
-    });
-
-    // Make sure to start any queued runs once this run is finished
-    await workerQueue.enqueue("startQueuedRuns", {
-      id: run.queueId,
-    });
-
-    if (
-      run.status === "SUCCESS" &&
-      run.output &&
-      typeof run.output === "object" &&
-      "events" in run.output
-    ) {
-      // If the run successfully completes, we will parse the output and
-      // if it's in the form of { events: Array<RawEvent> } then we will send the events
-      const parsedOutput = SendEventOutputSchema.safeParse(run.output);
-
-      if (parsedOutput.success) {
-        for (const newEvent of parsedOutput.data.events) {
-          await this.#ingestEventService.call(run.environment, newEvent, parsedOutput.data.options);
-        }
-      }
-    }
-  }
-}

apps/webapp/app/services/runs/startQueuedRuns.server.ts

@@ -1,56 +0,0 @@
-import type { PrismaClient } from "~/db.server";
-import { prisma } from "~/db.server";
-import { workerQueue } from "../worker.server";
-
-export class StartQueuedRunsService {
-  #prismaClient: PrismaClient;
-
-  constructor(prismaClient: PrismaClient = prisma) {
-    this.#prismaClient = prismaClient;
-  }
-
-  public async call(id: string) {
-    const queue = await this.#prismaClient.jobQueue.findUnique({
-      where: { id },
-      include: {
-        runs: {
-          where: {
-            status: "QUEUED",
-          },
-          orderBy: {
-            queuedAt: "asc",
-          },
-          take: 1,
-        },
-      },
-    });
-
-    if (!queue) {
-      return;
-    }
-
-    if (queue.runs.length === 0) {
-      return;
-    }
-
-    if (queue.jobCount >= queue.maxJobs) {
-      return;
-    }
-
-    const run = queue.runs[0];
-
-    if (!run) {
-      return;
-    }
-
-    await workerQueue.enqueue(
-      "startRun",
-      {
-        id: run.id,
-      },
-      {
-        queueName: `job-queue:${queue.id}`,
-      }
-    );
-  }
-}

apps/webapp/app/services/runs/startRun.server.ts

@@ -1,7 +1,7 @@
 import type { ConnectionType, Integration, IntegrationConnection } from "@trigger.dev/database";
-import { EXECUTE_JOB_RETRY_LIMIT, PREPROCESS_RETRY_LIMIT } from "~/consts";
 import type { PrismaClient, PrismaClientOrTransaction } from "~/db.server";
 import { prisma } from "~/db.server";
+import { enqueueRunExecutionV2 } from "~/models/jobRunExecution.server";
 import { workerQueue } from "../worker.server";
 
 type FoundRun = NonNullable<Awaited<ReturnType<typeof findRun>>>;
@@ -21,34 +21,20 @@ export class StartRunService {
       return;
     }
 
-    if (run.queue.jobCount >= run.queue.maxJobs) {
-      await this.#queueRun(id);
-    } else {
-      const runConnectionsByKey = await createRunConnections(this.#prismaClient, run);
+    const runConnectionsByKey = await createRunConnections(this.#prismaClient, run);
 
-      if (hasMissingConnections(runConnectionsByKey)) {
-        await this.#handleMissingConnections(id, runConnectionsByKey);
-      } else {
-        await this.#startRun(id, run, runConnectionsByKey);
-      }
+    if (hasMissingConnections(runConnectionsByKey)) {
+      await this.#handleMissingConnections(id, runConnectionsByKey);
+    } else {
+      await this.#startRun(id, run, runConnectionsByKey);
     }
   }
 
   #runIsStartable(run: FoundRun) {
-    const startableStatuses = ["PENDING", "QUEUED", "WAITING_ON_CONNECTIONS"] as const;
+    const startableStatuses = ["PENDING", "WAITING_ON_CONNECTIONS"] as const;
     return startableStatuses.includes(run.status);
   }
 
-  async #queueRun(id: string) {
-    await this.#prismaClient.jobRun.update({
-      where: { id },
-      data: {
-        status: "QUEUED",
-        queuedAt: new Date(),
-      },
-    });
-  }
-
   async #startRun(id: string, run: FoundRun, runConnectionsByKey: RunConnectionsByKey) {
     const createRunConnections = Object.entries(runConnectionsByKey)
       .map(([key, runConnection]) =>
@@ -69,89 +55,35 @@ export class StartRunService {
       )
       .filter(Boolean);
 
-    const updateRunAndCreateExecution = async () => {
+    const updateRun = async () => {
       if (run.preprocess) {
         // Start the jobRun and increment the jobCount
-        await this.#prismaClient.jobRun.update({
+        return await this.#prismaClient.jobRun.update({
           where: { id },
           data: {
             status: "PREPROCESSING",
-            queue: {
-              update: {
-                jobCount: {
-                  increment: 1,
-                },
-              },
-            },
             runConnections: {
               create: createRunConnections,
             },
           },
         });
-
-        return await this.#prismaClient.jobRunExecution.create({
-          data: {
-            run: {
-              connect: {
-                id,
-              },
-            },
-            status: "PENDING",
-            reason: "PREPROCESS",
-            retryLimit: PREPROCESS_RETRY_LIMIT,
-          },
-        });
       } else {
-        // Start the jobRun and increment the jobCount
-        await this.#prismaClient.jobRun.update({
+        return await this.#prismaClient.jobRun.update({
           where: { id },
           data: {
-            status: "STARTED",
-            startedAt: new Date(),
-            queue: {
-              update: {
-                jobCount: {
-                  increment: 1,
-                },
-              },
-            },
+            status: "QUEUED",
+            queuedAt: new Date(),
             runConnections: {
               create: createRunConnections,
             },
           },
         });
-
-        return await this.#prismaClient.jobRunExecution.create({
-          data: {
-            run: {
-              connect: {
-                id,
-              },
-            },
-            status: "PENDING",
-            reason: "EXECUTE_JOB",
-            retryLimit: EXECUTE_JOB_RETRY_LIMIT,
-          },
-        });
       }
     };
 
-    const execution = await updateRunAndCreateExecution();
+    const updatedRun = await updateRun();
 
-    const job = await workerQueue.enqueue("performRunExecution", {
-      id: execution.id,
-    });
-
-    await this.#prismaClient.jobRunExecution.update({
-      where: { id: execution.id },
-      data: {
-        graphileJobId: job.id,
-      },
-    });
-
-    await workerQueue.enqueue("startQueuedRuns", {
-      id: run.queueId,
-    });
+    await enqueueRunExecutionV2(updatedRun, this.#prismaClient);
   }
 
   async #handleMissingConnections(id: string, runConnectionsByKey: RunConnectionsByKey) {

apps/webapp/app/services/tasks/performTaskOperation.server.ts

@@ -1,7 +1,3 @@
-import type { Task } from "@trigger.dev/database";
-import { EXECUTE_JOB_RETRY_LIMIT } from "~/consts";
-import { $transaction, PrismaClient, PrismaClientOrTransaction, prisma } from "~/db.server";
-import { workerQueue } from "../worker.server";
 import {
   FetchOperationSchema,
   FetchRequestInit,
@@ -10,9 +6,13 @@ import {
   RedactString,
   calculateRetryAt,
 } from "@trigger.dev/core";
+import type { Task } from "@trigger.dev/database";
+import { $transaction, PrismaClient, PrismaClientOrTransaction, prisma } from "~/db.server";
+import { enqueueRunExecutionV2 } from "~/models/jobRunExecution.server";
+import { formatUnknownError } from "~/utils/formatErrors.server";
 import { safeJsonFromResponse } from "~/utils/json";
 import { logger } from "../logger.server";
-import { formatUnknownError } from "~/utils/formatErrors.server";
+import { workerQueue } from "../worker.server";
 
 type FoundTask = Awaited<ReturnType<typeof findTask>>;
 
@@ -192,7 +192,7 @@ export class PerformTaskOperationService {
     });
   }
 
-  async #resumeTaskWithError(task: Task, output: any) {
+  async #resumeTaskWithError(task: NonNullable<FoundTask>, output: any) {
     await $transaction(this.#prismaClient, async (tx) => {
       await tx.task.update({
         where: { id: task.id },
@@ -243,34 +243,8 @@ export class PerformTaskOperationService {
     });
   }
 
-  async #resumeRunExecution(task: Task, prisma: PrismaClientOrTransaction) {
-    await $transaction(prisma, async (tx) => {
-      const newJobExecution = await tx.jobRunExecution.create({
-        data: {
-          runId: task.runId,
-          reason: "EXECUTE_JOB",
-          status: "PENDING",
-          retryLimit: EXECUTE_JOB_RETRY_LIMIT,
-        },
-      });
-
-      const graphileJob = await workerQueue.enqueue(
-        "performRunExecution",
-        {
-          id: newJobExecution.id,
-        },
-        { tx }
-      );
-
-      await tx.jobRunExecution.update({
-        where: {
-          id: newJobExecution.id,
-        },
-        data: {
-          graphileJobId: graphileJob.id,
-        },
-      });
-    });
+  async #resumeRunExecution(task: NonNullable<FoundTask>, prisma: PrismaClientOrTransaction) {
+    await enqueueRunExecutionV2(task.run, prisma);
   }
 }
 
@@ -305,6 +279,11 @@ async function findTask(prisma: PrismaClient, id: string) {
     where: { id },
     include: {
       attempts: true,
+      run: {
+        include: {
+          queue: true,
+        },
+      },
     },
   });
 }

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

@@ -12,9 +12,8 @@ import { InvokeDispatcherService } from "./events/invokeDispatcher.server";
 import { integrationAuthRepository } from "./externalApis/integrationAuthRepository.server";
 import { IntegrationConnectionCreatedService } from "./externalApis/integrationConnectionCreated.server";
 import { MissingConnectionCreatedService } from "./runs/missingConnectionCreated.server";
-import { PerformRunExecutionService } from "./runs/performRunExecution.server";
-import { RunFinishedService } from "./runs/runFinished.server";
-import { StartQueuedRunsService } from "./runs/startQueuedRuns.server";
+import { PerformRunExecutionV1Service } from "./runs/performRunExecutionV1.server";
+import { PerformRunExecutionV2Service } from "./runs/performRunExecutionV2.server";
 import { StartRunService } from "./runs/startRun.server";
 import { DeliverScheduledEventService } from "./schedules/deliverScheduledEvent.server";
 import { ActivateSourceService } from "./sources/activateSource.server";
@@ -30,13 +29,9 @@ const workerCatalog = {
   }),
   scheduleEmail: DeliverEmailSchema,
   startRun: z.object({ id: z.string() }),
-  performRunExecution: z.object({
-    id: z.string(),
-  }),
   performTaskOperation: z.object({
     id: z.string(),
   }),
-  runFinished: z.object({ id: z.string() }),
   deliverHttpSourceRequest: z.object({ id: z.string() }),
   refreshOAuthToken: z.object({
     organizationId: z.string(),
@@ -46,7 +41,7 @@ const workerCatalog = {
     id: z.string(),
     orphanedEvents: z.array(z.string()).optional(),
   }),
-  startQueuedRuns: z.object({ id: z.string() }),
+
   deliverEvent: z.object({ id: z.string() }),
   "events.invokeDispatcher": z.object({
     id: z.string(),
@@ -64,10 +59,24 @@ const workerCatalog = {
   }),
 };
 
+const executionWorkerCatalog = {
+  performRunExecution: z.object({
+    id: z.string(),
+  }),
+  performRunExecutionV2: z.object({
+    id: z.string(),
+    reason: z.enum(["EXECUTE_JOB", "PREPROCESS"]),
+    resumeTaskId: z.string().optional(),
+    isRetry: z.boolean(),
+  }),
+};
+
 let workerQueue: ZodWorker<typeof workerCatalog>;
+let executionWorker: ZodWorker<typeof executionWorkerCatalog>;
 
 declare global {
   var __worker__: ZodWorker<typeof workerCatalog>;
+  var __executionWorker__: ZodWorker<typeof executionWorkerCatalog>;
 }
 
 // this is needed because in development we don't want to restart
@@ -76,25 +85,41 @@ declare global {
 // in production we'll have a single connection to the DB.
 if (env.NODE_ENV === "production") {
   workerQueue = getWorkerQueue();
+  executionWorker = getExecutionWorkerQueue();
 } else {
   if (!global.__worker__) {
     global.__worker__ = getWorkerQueue();
   }
   workerQueue = global.__worker__;
+
+  if (!global.__executionWorker__) {
+    global.__executionWorker__ = getExecutionWorkerQueue();
+  }
+
+  executionWorker = global.__executionWorker__;
 }
 
 export async function init() {
-  await workerQueue.initialize();
+  if (env.WORKER_ENABLED === "true") {
+    await workerQueue.initialize();
+  }
+
+  if (env.EXECUTION_WORKER_ENABLED === "true") {
+    await executionWorker.initialize();
+  }
 }
 
 function getWorkerQueue() {
   return new ZodWorker({
+    name: "workerQueue",
     prisma,
     runnerOptions: {
       connectionString: env.DATABASE_URL,
-      concurrency: 5,
-      pollInterval: 1000,
+      concurrency: env.WORKER_CONCURRENCY,
+      pollInterval: env.WORKER_POLL_INTERVAL,
       noPreparedStatements: env.DATABASE_URL !== env.DIRECT_URL,
+      schema: env.WORKER_SCHEMA,
+      maxPoolSize: env.WORKER_CONCURRENCY,
     },
     schema: workerCatalog,
     recurringTasks: {
@@ -124,6 +149,7 @@ function getWorkerQueue() {
     },
     tasks: {
       "events.invokeDispatcher": {
+        priority: 0, // smaller number = higher priority
         maxAttempts: 3,
         handler: async (payload, job) => {
           const service = new InvokeDispatcherService();
@@ -132,6 +158,7 @@ function getWorkerQueue() {
         },
       },
       "events.deliverScheduled": {
+        priority: 0, // smaller number = higher priority
         maxAttempts: 5,
         handler: async ({ id, payload }, job) => {
           const service = new DeliverScheduledEventService();
@@ -140,6 +167,7 @@ function getWorkerQueue() {
         },
       },
       connectionCreated: {
+        priority: 10, // smaller number = higher priority
         maxAttempts: 3,
         handler: async (payload, job) => {
           const service = new IntegrationConnectionCreatedService();
@@ -148,6 +176,7 @@ function getWorkerQueue() {
         },
       },
       missingConnectionCreated: {
+        priority: 10, // smaller number = higher priority
         maxAttempts: 3,
         handler: async (payload, job) => {
           const service = new MissingConnectionCreatedService();
@@ -155,24 +184,8 @@ function getWorkerQueue() {
           await service.call(payload.id);
         },
       },
-      runFinished: {
-        maxAttempts: 3,
-        handler: async (payload, job) => {
-          const service = new RunFinishedService();
-
-          await service.call(payload.id);
-        },
-      },
-      startQueuedRuns: {
-        maxAttempts: 3,
-        queueName: (payload) => `queue:${payload.id}`,
-        handler: async (payload, job) => {
-          const service = new StartQueuedRunsService();
-
-          await service.call(payload.id);
-        },
-      },
       activateSource: {
+        priority: 10, // smaller number = higher priority
         maxAttempts: 3,
         handler: async (payload, job) => {
           const service = new ActivateSourceService();
@@ -181,6 +194,7 @@ function getWorkerQueue() {
         },
       },
       deliverHttpSourceRequest: {
+        priority: 1, // smaller number = higher priority
         maxAttempts: 25,
         handler: async (payload, job) => {
           const service = new DeliverHttpSourceRequestService();
@@ -189,6 +203,7 @@ function getWorkerQueue() {
         },
       },
       startRun: {
+        priority: 0, // smaller number = higher priority
         maxAttempts: 8,
         handler: async (payload, job) => {
           const service = new StartRunService();
@@ -196,16 +211,8 @@ function getWorkerQueue() {
           await service.call(payload.id);
         },
       },
-      performRunExecution: {
-        queueName: (payload) => `runs:${payload.id}`,
-        maxAttempts: 1,
-        handler: async (payload, job) => {
-          const service = new PerformRunExecutionService();
-
-          await service.call(payload.id);
-        },
-      },
       performTaskOperation: {
+        priority: 0, // smaller number = higher priority
         queueName: (payload) => `tasks:${payload.id}`,
         maxAttempts: 3,
         handler: async (payload, job) => {
@@ -223,6 +230,7 @@ function getWorkerQueue() {
         },
       },
       indexEndpoint: {
+        priority: 1, // smaller number = higher priority
         handler: async (payload, job) => {
           const service = new IndexEndpointService();
 
@@ -230,6 +238,7 @@ function getWorkerQueue() {
         },
       },
       deliverEvent: {
+        priority: 0, // smaller number = higher priority
         handler: async (payload, job) => {
           const service = new DeliverEventService();
 
@@ -237,6 +246,7 @@ function getWorkerQueue() {
         },
       },
       refreshOAuthToken: {
+        priority: 8, // smaller number = higher priority
         queueName: "internal-queue",
         maxAttempts: 10,
         handler: async (payload, job) => {
@@ -249,4 +259,42 @@ function getWorkerQueue() {
   });
 }
 
-export { workerQueue };
+function getExecutionWorkerQueue() {
+  return new ZodWorker({
+    name: "executionWorker",
+    prisma,
+    runnerOptions: {
+      connectionString: env.DATABASE_URL,
+      concurrency: env.EXECUTION_WORKER_CONCURRENCY,
+      pollInterval: env.EXECUTION_WORKER_POLL_INTERVAL,
+      noPreparedStatements: env.DATABASE_URL !== env.DIRECT_URL,
+      schema: env.WORKER_SCHEMA,
+      maxPoolSize: env.EXECUTION_WORKER_CONCURRENCY,
+    },
+    schema: executionWorkerCatalog,
+    tasks: {
+      performRunExecution: {
+        priority: 0, // smaller number = higher priority
+        maxAttempts: 1,
+        handler: async (payload, job) => {
+          // This is a legacy task that we don't use anymore, but needs to be here for backwards compatibility
+          // TODO: remove this once all performRunExecution tasks have been processed
+          const service = new PerformRunExecutionV1Service();
+
+          await service.call(payload.id);
+        },
+      },
+      performRunExecutionV2: {
+        priority: 0, // smaller number = higher priority
+        maxAttempts: 18,
+        handler: async (payload, job) => {
+          const service = new PerformRunExecutionV2Service();
+
+          await service.call(payload.id, payload.reason, payload.isRetry, payload.resumeTaskId);
+        },
+      },
+    },
+  });
+}
+
+export { executionWorker, workerQueue };

apps/webapp/server.ts

@@ -54,14 +54,18 @@ app.all(
 
 const port = process.env.REMIX_APP_PORT || 3000;
 
-const server = app.listen(port, () => {
-  // require the built app so we're ready when the first request comes in
-  require(BUILD_DIR);
-  console.log(`✅ app ready: http://localhost:${port}`);
-});
+if (process.env.HTTP_SERVER_DISABLED !== "true") {
+  const server = app.listen(port, () => {
+    // require the built app so we're ready when the first request comes in
+    require(BUILD_DIR);
+    console.log(`✅ app ready: http://localhost:${port}`);
+  });
 
-// Handle shutdowns gracefully
-createTerminus(server, { signals: ["SIGINT", "SIGTERM"], timeout: 5000 });
+  // Handle shutdowns gracefully
+  createTerminus(server, { signals: ["SIGINT", "SIGTERM"], timeout: 5000 });
+} else {
+  console.log(`✅ app ready (skipping http server)`);
+}
 
 function purgeRequireCache() {
   // purge require cache on requests for "server side HMR" this won't let

docker/Dockerfile

@@ -31,7 +31,8 @@ RUN corepack enable
 ENV NODE_ENV production
 RUN pnpm install --prod --no-frozen-lockfile
 COPY --from=pruner --chown=node:node /triggerdotdev/packages/database/prisma/schema.prisma /triggerdotdev/packages/database/prisma/schema.prisma
-RUN pnpx prisma@5.1.1 generate --schema /triggerdotdev/packages/database/prisma/schema.prisma
+# RUN pnpm add @prisma/client@5.1.1 -w
+RUN pnpx prisma@4.16.0 generate --schema /triggerdotdev/packages/database/prisma/schema.prisma
 
 ## Builder (builds the webapp)
 FROM base AS builder

docs/sdk/job.mdx

@@ -114,12 +114,6 @@ client.defineJob({
       Imports the specified integrations into the Job. The integrations will be available on the `io` object in the `run()` function with the same name as the key. For example:
       <Snippet file="how-to-pass-integrations.mdx" />
     </ParamField>
-    <ParamField body="enabled" type="boolean">
-      The `enabled` property is used to enable or disable the Job. If you disable a Job, it will not run.
-    </ParamField>
-    <ParamField body="queue" type="string | QueueOptions">
-      The `queue` property is used to specify a custom queue. If you use an Object and specify the `maxConcurrent` option, you can control how many simulataneous runs can happen.
-    </ParamField>
     <ParamField body="logLevel" type="log | error | warn | info | debug">
       The `logLevel` property is an optional property that specifies the level of
       logging for the Job. The level is inherited from the client if you omit this property.

examples/job-catalog/package.json

@@ -4,9 +4,14 @@
   "private": true,
   "scripts": {
     "stripe": "nodemon --watch src/stripe.ts -r tsconfig-paths/register -r dotenv/config src/stripe.ts",
+    "slack": "nodemon --watch src/slack.ts -r tsconfig-paths/register -r dotenv/config src/slack.ts",
+    "openai": "nodemon --watch src/openai.ts -r tsconfig-paths/register -r dotenv/config src/openai.ts",
     "sendgrid": "nodemon --watch src/sendgrid.ts -r tsconfig-paths/register -r dotenv/config src/sendgrid.ts",
     "supabase": "nodemon --watch src/supabase.ts -r tsconfig-paths/register -r dotenv/config src/supabase.ts",
     "supabase:types": "npx supabase gen types typescript --project-id $SUPABASE_PROJECT_ID --schema public --schema auth --schema storage > src/supabase-types.ts",
+    "events": "nodemon --watch src/events.ts -r tsconfig-paths/register -r dotenv/config src/events.ts",
+    "stressTest": "nodemon --watch src/stressTest.ts -r tsconfig-paths/register -r dotenv/config src/stressTest.ts",
+    "delays": "nodemon --watch src/delays.ts -r tsconfig-paths/register -r dotenv/config src/delays.ts",
     "dev:trigger": "trigger-cli dev --port 8080"
   },
   "dependencies": {
@@ -37,4 +42,4 @@
     "ts-node": "^10.9.1",
     "tsconfig-paths": "^3.14.1"
   }
-}
+}

examples/job-catalog/src/delays.ts

@@ -0,0 +1,36 @@
+import { createExpressServer } from "@trigger.dev/express";
+import { TriggerClient, eventTrigger } from "@trigger.dev/sdk";
+
+export const client = new TriggerClient({
+  id: "job-catalog",
+  apiKey: process.env["TRIGGER_API_KEY"],
+  apiUrl: process.env["TRIGGER_API_URL"],
+  verbose: false,
+  ioLogLocalEnabled: true,
+});
+
+client.defineJob({
+  id: "delays-example-1",
+  name: "Delays Example 1",
+  version: "1.0.0",
+  trigger: eventTrigger({
+    name: "delays.example",
+  }),
+  run: async (payload, io, ctx) => {
+    await io.wait("wait-1", 60);
+  },
+});
+
+client.defineJob({
+  id: "delays-example-2",
+  name: "Delays Example 2 - Long Delay",
+  version: "1.0.0",
+  trigger: eventTrigger({
+    name: "delays.example.long",
+  }),
+  run: async (payload, io, ctx) => {
+    await io.wait("wait-1", 60 * 30);
+  },
+});
+
+createExpressServer(client);

examples/job-catalog/src/events.ts

@@ -0,0 +1,32 @@
+import { createExpressServer } from "@trigger.dev/express";
+import { TriggerClient, eventTrigger } from "@trigger.dev/sdk";
+
+export const client = new TriggerClient({
+  id: "job-catalog",
+  apiKey: process.env["TRIGGER_API_KEY"],
+  apiUrl: process.env["TRIGGER_API_URL"],
+  verbose: false,
+  ioLogLocalEnabled: true,
+});
+
+client.defineJob({
+  id: "event-example-1",
+  name: "Event Example 1",
+  version: "1.0.0",
+  trigger: eventTrigger({
+    name: "event.example",
+  }),
+  run: async (payload, io, ctx) => {
+    await io.runTask("task-example-1", { name: "Task 1" }, async () => {
+      return {
+        message: "Hello World",
+      };
+    });
+
+    await io.wait("wait-1", 1);
+
+    await io.logger.info("Hello World", { ctx });
+  },
+});
+
+createExpressServer(client);

examples/job-catalog/src/openai.ts

@@ -0,0 +1,80 @@
+import { createExpressServer } from "@trigger.dev/express";
+import { TriggerClient, eventTrigger } from "@trigger.dev/sdk";
+import { OpenAI } from "@trigger.dev/openai";
+
+export const client = new TriggerClient({
+  id: "job-catalog",
+  apiKey: process.env["TRIGGER_API_KEY"],
+  apiUrl: process.env["TRIGGER_API_URL"],
+  verbose: false,
+  ioLogLocalEnabled: true,
+});
+
+const openai = new OpenAI({
+  id: "openai",
+  apiKey: process.env["OPENAI_API_KEY"]!,
+});
+
+client.defineJob({
+  id: "openai-tasks",
+  name: "OpenAI Tasks",
+  version: "0.0.1",
+  trigger: eventTrigger({
+    name: "openai.tasks",
+  }),
+  integrations: {
+    openai,
+  },
+  run: async (payload, io, ctx) => {
+    const models = await io.openai.listModels("list-models");
+
+    if (models.data.length > 0) {
+      await io.openai.retrieveModel("get-model", {
+        model: models.data[0].id,
+      });
+    }
+
+    await io.openai.backgroundCreateChatCompletion("background-chat-completion", {
+      model: "gpt-3.5-turbo",
+      messages: [
+        {
+          role: "user",
+          content: "Create a good programming joke about background jobs",
+        },
+      ],
+    });
+
+    await io.openai.createChatCompletion("chat-completion", {
+      model: "gpt-3.5-turbo",
+      messages: [
+        {
+          role: "user",
+          content: "Create a good programming joke about background jobs",
+        },
+      ],
+    });
+
+    await io.openai.backgroundCreateCompletion("background-completion", {
+      model: "text-davinci-003",
+      prompt: "Create a good programming joke about Tasks",
+    });
+
+    await io.openai.createCompletion("completion", {
+      model: "text-davinci-003",
+      prompt: "Create a good programming joke about Tasks",
+    });
+
+    await io.openai.createEdit("edit", {
+      model: "text-davinci-edit-001",
+      input: "Thsi is ridddled with erors",
+      instruction: "Fix the spelling errors",
+    });
+
+    await io.openai.createEmbedding("embedding", {
+      model: "text-embedding-ada-002",
+      input: "The food was delicious and the waiter...",
+    });
+  },
+});
+
+createExpressServer(client);

examples/job-catalog/src/slack.ts

@@ -0,0 +1,33 @@
+import { createExpressServer } from "@trigger.dev/express";
+import { TriggerClient, eventTrigger } from "@trigger.dev/sdk";
+import { Slack } from "@trigger.dev/slack";
+
+export const slack = new Slack({ id: "slack" });
+
+export const client = new TriggerClient({
+  id: "job-catalog",
+  apiKey: process.env["TRIGGER_API_KEY"],
+  apiUrl: process.env["TRIGGER_API_URL"],
+  verbose: false,
+  ioLogLocalEnabled: true,
+});
+
+client.defineJob({
+  id: "slack-example-1",
+  name: "Slack Example 1",
+  version: "1.0.0",
+  trigger: eventTrigger({
+    name: "slack.example",
+  }),
+  integrations: {
+    slack,
+  },
+  run: async (payload, io, ctx) => {
+    await io.slack.postMessage("Slack 📝", {
+      channel: "C04GWUTDC3W",
+      text: "Welcome to the team, Eric!",
+    });
+  },
+});
+
+createExpressServer(client);

examples/job-catalog/src/stressTest.ts

@@ -0,0 +1,41 @@
+import { createExpressServer } from "@trigger.dev/express";
+import { TriggerClient, eventTrigger } from "@trigger.dev/sdk";
+
+export const client = new TriggerClient({
+  id: "job-catalog",
+  apiKey: process.env["TRIGGER_API_KEY"],
+  apiUrl: process.env["TRIGGER_API_URL"],
+  verbose: false,
+  ioLogLocalEnabled: true,
+});
+
+client.defineJob({
+  id: "stress-test-1",
+  name: "Stress Test 1",
+  version: "1.0.0",
+  trigger: eventTrigger({
+    name: "stress.test.1",
+  }),
+  run: async (payload, io, ctx) => {
+    // Run 10 tasks, each with a 300KB output
+    for (let i = 0; i < 10; i++) {
+      await io.runTask(`task-${i}`, { name: `Task ${i}` }, async (task) => {
+        return {
+          output: "a".repeat(300 * 1024),
+        };
+      });
+    }
+
+    // Now run a single task with 5MB output
+    await io.runTask(`task-5mb`, { name: `Task 5MB` }, async (task) => {
+      return {
+        output: "a".repeat(5 * 1024 * 1024),
+      };
+    });
+
+    // Now do a wait for 5 seconds
+    await io.wait("wait", 5);
+  },
+});
+
+createExpressServer(client);

package.json

@@ -66,4 +66,4 @@
     "@changesets/cli": "^2.26.0",
     "node-fetch": "2.6.x"
   }
-}
+}

packages/core/src/logger.ts

@@ -166,6 +166,11 @@ function filterKeys(obj: unknown, keys: string[]): any {
 
   for (const [key, value] of Object.entries(obj)) {
     if (keys.includes(key)) {
+      if (value) {
+        filteredObj[key] = `[filtered ${prettyPrintBytes(value)}]`;
+      } else {
+        filteredObj[key] = value;
+      }
       continue;
     }
 
@@ -174,3 +179,25 @@ function filterKeys(obj: unknown, keys: string[]): any {
 
   return filteredObj;
 }
+
+function prettyPrintBytes(value: unknown): string {
+  if (process.env.NODE_ENV === "production") {
+    return "skipped size";
+  }
+
+  const sizeInBytes = Buffer.byteLength(JSON.stringify(value), "utf8");
+
+  if (sizeInBytes < 1024) {
+    return `${sizeInBytes} bytes`;
+  }
+
+  if (sizeInBytes < 1024 * 1024) {
+    return `${(sizeInBytes / 1024).toFixed(2)} KB`;
+  }
+
+  if (sizeInBytes < 1024 * 1024 * 1024) {
+    return `${(sizeInBytes / (1024 * 1024)).toFixed(2)} MB`;
+  }
+
+  return `${(sizeInBytes / (1024 * 1024 * 1024)).toFixed(2)} GB`;
+}

packages/core/src/schemas/api.ts

@@ -160,9 +160,8 @@ export const JobMetadataSchema = z.object({
   trigger: TriggerMetadataSchema,
   integrations: z.record(IntegrationConfigSchema),
   internal: z.boolean().default(false),
-  queue: z.union([QueueOptionsSchema, z.string()]).optional(),
-  startPosition: z.enum(["initial", "latest"]),
   enabled: z.boolean(),
+  startPosition: z.enum(["initial", "latest"]),
   preprocessRuns: z.boolean(),
 });
 

packages/database/package.json

@@ -5,11 +5,11 @@
   "main": "./src/index.ts",
   "types": "./src/index.ts",
   "dependencies": {
-    "@prisma/client": "5.1.1",
+    "@prisma/client": "4.16.0",
     "typescript": "^4.8.4"
   },
   "devDependencies": {
-    "prisma": "^5.1.0"
+    "prisma": "4.16.0"
   },
   "scripts": {
     "generate": "prisma generate",
@@ -18,4 +18,4 @@
     "db:studio": "prisma studio",
     "typecheck": "tsc --noEmit"
   }
-}
+}

packages/trigger-sdk/src/io.ts

@@ -69,7 +69,7 @@ export class IO {
 
     if (options.cachedTasks) {
       options.cachedTasks.forEach((task) => {
-        this._cachedTasks.set(task.id, task);
+        this._cachedTasks.set(task.idempotencyKey, task);
       });
     }
 

packages/trigger-sdk/src/job.ts

@@ -42,9 +42,9 @@ export type JobOptions<
       });
       ``` */
   integrations?: TIntegrations;
-  /** The `queue` property is used to specify a custom queue. If you use an Object and specify the `maxConcurrent` option, you can control how many simulataneous runs can happen. */
+  /** @deprecated This property is deprecated and no longer effects the execution of the Job
+   * */
   queue?: QueueOptions | string;
-  startPosition?: "initial" | "latest";
   /** The `enabled` property is used to enable or disable the Job. If you disable a Job, it will not run. */
   enabled?: boolean;
   /** This function gets called automatically when a Run is Triggered.
@@ -142,8 +142,7 @@ export class Job<
       event: this.trigger.event,
       trigger: this.trigger.toJSON(),
       integrations: this.integrations,
-      queue: this.options.queue,
-      startPosition: this.options.startPosition ?? "latest",
+      startPosition: "latest", // this is deprecated, leaving this for now to make sure newer clients work with older servers
       enabled: typeof this.options.enabled === "boolean" ? this.options.enabled : true,
       preprocessRuns: this.trigger.preprocessRuns,
       internal,

packages/trigger-sdk/src/triggerClient.ts

@@ -508,11 +508,6 @@ export class TriggerClient {
       integrations: {
         integration: options.source.integration,
       },
-      queue: {
-        name: options.key,
-        maxConcurrent: 1,
-      },
-      startPosition: "initial",
       run: async (event, io, ctx) => {
         const updates = await options.source.register(options.params, event, io, ctx);
 

perf/package.json

@@ -0,0 +1,41 @@
+{
+  "name": "perf",
+  "version": "1.0.0",
+  "description": "Performance testing for Trigger.dev",
+  "main": "src/index.ts",
+  "scripts": {
+    "start": "ts-node -r tsconfig-paths/register -r dotenv/config src/index.ts",
+    "server": "ts-node -r tsconfig-paths/register -r dotenv/config src/server.ts",
+    "dev:trigger": "trigger-cli dev --port ${PORT:-3000}"
+  },
+  "author": "",
+  "license": "ISC",
+  "dependencies": {
+    "@trigger.dev/express": "workspace:*",
+    "@trigger.dev/github": "workspace:*",
+    "@trigger.dev/openai": "workspace:*",
+    "@trigger.dev/plain": "workspace:*",
+    "@trigger.dev/resend": "workspace:*",
+    "@trigger.dev/sdk": "workspace:*",
+    "@trigger.dev/sendgrid": "workspace:*",
+    "@trigger.dev/slack": "workspace:*",
+    "@trigger.dev/stripe": "workspace:*",
+    "@trigger.dev/supabase": "workspace:*",
+    "@trigger.dev/typeform": "workspace:*",
+    "zod": "3.21.4"
+  },
+  "devDependencies": {
+    "@trigger.dev/cli": "workspace:*",
+    "@trigger.dev/tsconfig": "workspace:*",
+    "@types/node": "^20.5.0",
+    "concurrently": "^8.2.0",
+    "dotenv": "^16.3.1",
+    "nodemon": "^3.0.1",
+    "ts-node": "^10.9.1",
+    "tsconfig-paths": "^3.14.1",
+    "typescript": "^5.1.6"
+  },
+  "trigger.dev": {
+    "endpointId": "perf"
+  }
+}

perf/src/index.ts

@@ -0,0 +1,83 @@
+import { triggerClient } from "./trigger";
+
+async function sendEvent() {
+  try {
+    return await triggerClient.sendEvent({
+      name: "perf.test",
+      payload: {
+        string: "Hello, World!",
+        number: 42,
+        boolean: true,
+        nullValue: null,
+        array: [1, 2, 3],
+        object: {
+          nestedString: "Nested value",
+          nestedNumber: 3.14,
+          nestedArray: ["apple", "banana", "cherry"],
+          nestedObject: {
+            nestedBoolean: false,
+            nestedNull: null,
+          },
+        },
+      },
+    });
+  } catch (err) {
+    console.error(err);
+  }
+}
+
+async function main() {
+  console.log("Preparing perf tests...");
+
+  // wait for 10 seconds
+  await new Promise((resolve) => setTimeout(resolve, 10000));
+
+  console.log("Starting perf tests in 1 second...");
+
+  // wait for 1 seconds
+  await new Promise((resolve) => setTimeout(resolve, 1000));
+
+  // Send 5 events per second for 30 seconds (1 event == 10 runs)
+  for (let i = 0; i < 30; i++) {
+    console.log("Sending 5 event...");
+
+    await sendEvent();
+    await sendEvent();
+    await sendEvent();
+    await sendEvent();
+    await sendEvent();
+
+    await new Promise((resolve) => setTimeout(resolve, 950));
+  }
+
+  // console.log("Sending 30 events...");
+  // for (let i = 0; i < 30; i++) {
+  //   await sendEvent();
+  // }
+}
+
+async function mainLong() {
+  console.log("Preparing long perf tests...");
+
+  // wait for 10 seconds
+  await new Promise((resolve) => setTimeout(resolve, 10000));
+
+  console.log("Starting long perf tests in 1 second...");
+
+  // wait for 1 seconds
+  await new Promise((resolve) => setTimeout(resolve, 1000));
+
+  // Send 1 events every 5 seconds for 30 minutes
+  for (let i = 0; i < 360; i++) {
+    console.log("Sending 1 events...");
+
+    await sendEvent();
+
+    await new Promise((resolve) => setTimeout(resolve, 5000));
+  }
+}
+
+mainLong().catch((err) => {
+  console.error(err);
+  process.exit(1);
+});

perf/src/server.ts

@@ -0,0 +1,7 @@
+import { triggerClient } from "./trigger";
+import { createExpressServer } from "@trigger.dev/express";
+
+const app = createExpressServer(
+  triggerClient,
+  process.env.PORT ? parseInt(process.env.PORT) : 3000
+);

perf/src/trigger.ts

@@ -0,0 +1,37 @@
+import { TriggerClient, eventTrigger } from "@trigger.dev/sdk";
+import { z } from "zod";
+
+export const triggerClient = new TriggerClient({
+  id: "perf",
+  apiKey: process.env.TRIGGER_API_KEY!,
+  apiUrl: process.env.TRIGGER_API_URL!,
+});
+
+// Define 10 jobs in a for loop
+for (let i = 0; i < 10; i++) {
+  triggerClient.defineJob({
+    id: `perf-test-${i + 1}`,
+    name: `Perf Test ${i + 1}`,
+    version: "1.0.0",
+    trigger: eventTrigger({
+      name: "perf.test",
+    }),
+    queue: {
+      name: "perf-test",
+      maxConcurrent: 50,
+    },
+    run: async (payload, io, ctx) => {
+      await io.runTask("task-1", { name: "task 1" }, async (task) => {
+        return {
+          value: Math.random(),
+        };
+      });
+
+      await io.runTask("task-2", { name: "task 2" }, async (task) => {
+        return {
+          value: Math.random(),
+        };
+      });
+    },
+  });
+}

perf/tsconfig.json

@@ -0,0 +1,37 @@
+{
+  "extends": "@trigger.dev/tsconfig/node18.json",
+  "include": ["./src/**/*.ts"],
+  "compilerOptions": {
+    "baseUrl": ".",
+    "lib": ["DOM", "DOM.Iterable"],
+    "paths": {
+      "@/*": ["./src/*"],
+      "@trigger.dev/sdk": ["../../packages/trigger-sdk/src/index"],
+      "@trigger.dev/sdk/*": ["../../packages/trigger-sdk/src/*"],
+      "@trigger.dev/express": ["../../packages/express/src/index"],
+      "@trigger.dev/express/*": ["../../packages/express/src/*"],
+      "@trigger.dev/core": ["../../packages/core/src/index"],
+      "@trigger.dev/core/*": ["../../packages/core/src/*"],
+      "@trigger.dev/integration-kit": ["../../packages/integration-kit/src/index"],
+      "@trigger.dev/integration-kit/*": ["../../packages/integration-kit/src/*"],
+      "@trigger.dev/github": ["../../integrations/github/src/index"],
+      "@trigger.dev/github/*": ["../../integrations/github/src/*"],
+      "@trigger.dev/slack": ["../../integrations/slack/src/index"],
+      "@trigger.dev/slack/*": ["../../integrations/slack/src/*"],
+      "@trigger.dev/openai": ["../../integrations/openai/src/index"],
+      "@trigger.dev/openai/*": ["../../integrations/openai/src/*"],
+      "@trigger.dev/resend": ["../../integrations/resend/src/index"],
+      "@trigger.dev/resend/*": ["../../integrations/resend/src/*"],
+      "@trigger.dev/typeform": ["../../integrations/typeform/src/index"],
+      "@trigger.dev/typeform/*": ["../../integrations/typeform/src/*"],
+      "@trigger.dev/plain": ["../../integrations/plain/src/index"],
+      "@trigger.dev/plain/*": ["../../integrations/plain/src/*"],
+      "@trigger.dev/supabase": ["../../integrations/supabase/src/index"],
+      "@trigger.dev/supabase/*": ["../../integrations/supabase/src/*"],
+      "@trigger.dev/stripe": ["../../integrations/stripe/src/index"],
+      "@trigger.dev/stripe/*": ["../../integrations/stripe/src/*"],
+      "@trigger.dev/sendgrid": ["../../integrations/sendgrid/src/index"],
+      "@trigger.dev/sendgrid/*": ["../../integrations/sendgrid/src/*"]
+    }
+  }
+}

pnpm-workspace.yaml

@@ -5,3 +5,4 @@ packages:
   - "apps/**"
   - "examples/*"
   - "docs"
+  - "perf"