Fixed sparodically failed run creations

- Use a better way of getting the latest job run number to increment
- Make the CreateRunService transaction more reliable
- Invoke dispatchers in parallel
- No longer swallow prisma errors in $transaction

apps/webapp/app/db.server.ts

@@ -31,7 +31,7 @@ export type PrismaTransactionOptions = {
   /**  Sets the transaction isolation level. By default this is set to the value currently configured in your database. */
   isolationLevel?: Prisma.TransactionIsolationLevel;
 
-  rethrowPrismaErrors?: boolean;
+  swallowPrismaErrors?: boolean;
 };
 
 export async function $transaction<R>(
@@ -55,11 +55,9 @@ export async function $transaction<R>(
         name: error.name,
       });
 
-      if (options?.rethrowPrismaErrors) {
-        throw error;
+      if (options?.swallowPrismaErrors) {
+        return;
       }
-
-      return;
     }
 
     throw error;
@@ -124,6 +122,10 @@ function getClient() {
         emit: "stdout",
         level: "warn",
       },
+      // {
+      //   emit: "stdout",
+      //   level: "query",
+      // },
     ],
   });
 

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

@@ -3,7 +3,7 @@ import { json } from "@remix-run/server-runtime";
 import type { CompleteTaskBodyOutput, ServerTask } from "@trigger.dev/core";
 import { CompleteTaskBodyInputSchema } from "@trigger.dev/core";
 import { z } from "zod";
-import { $transaction, PrismaClient, prisma } from "~/db.server";
+import { PrismaClient, prisma } from "~/db.server";
 import { taskWithAttemptsToServerTask } from "~/models/task.server";
 import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
@@ -86,8 +86,8 @@ export class CompleteRunTaskService {
   ): Promise<ServerTask | undefined> {
     // Using a transaction, we'll first check to see if the task already exists and return if if it does
     // If it doesn't exist, we'll create it and return it
-    const task = await this.#prismaClient.$transaction(async (prisma) => {
-      const existingTask = await prisma.task.findUnique({
+    const task = await this.#prismaClient.$transaction(async (tx) => {
+      const existingTask = await tx.task.findUnique({
         where: {
           id,
         },
@@ -129,35 +129,31 @@ export class CompleteRunTaskService {
         return existingTask;
       }
 
-      const task = await $transaction(prisma, async (tx) => {
-        if (existingTask.attempts.length === 1) {
-          await tx.taskAttempt.update({
-            where: {
-              id: existingTask.attempts[0].id,
-            },
-            data: {
-              status: "COMPLETED",
-            },
-          });
-        }
-
-        return await tx.task.update({
+      if (existingTask.attempts.length === 1) {
+        await tx.taskAttempt.update({
           where: {
-            id,
+            id: existingTask.attempts[0].id,
           },
           data: {
             status: "COMPLETED",
-            output: taskBody.output ?? undefined,
-            completedAt: new Date(),
-            outputProperties: taskBody.properties,
-          },
-          include: {
-            attempts: true,
           },
         });
-      });
+      }
 
-      return task;
+      return await tx.task.update({
+        where: {
+          id,
+        },
+        data: {
+          status: "COMPLETED",
+          output: taskBody.output ?? undefined,
+          completedAt: new Date(),
+          outputProperties: taskBody.properties,
+        },
+        include: {
+          attempts: true,
+        },
+      });
     });
 
     return task ? taskWithAttemptsToServerTask(task) : undefined;

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

@@ -2,7 +2,7 @@ import type { ActionArgs } from "@remix-run/server-runtime";
 import { json } from "@remix-run/server-runtime";
 import { FailTaskBodyInput, FailTaskBodyInputSchema, ServerTask } from "@trigger.dev/core";
 import { z } from "zod";
-import { $transaction, PrismaClient, prisma } from "~/db.server";
+import { PrismaClient, prisma } from "~/db.server";
 import { taskWithAttemptsToServerTask } from "~/models/task.server";
 import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
@@ -86,8 +86,8 @@ export class FailRunTaskService {
   ): Promise<ServerTask | undefined> {
     // Using a transaction, we'll first check to see if the task already exists and return if if it does
     // If it doesn't exist, we'll create it and return it
-    const task = await this.#prismaClient.$transaction(async (prisma) => {
-      const existingTask = await prisma.task.findUnique({
+    const task = await this.#prismaClient.$transaction(async (tx) => {
+      const existingTask = await tx.task.findUnique({
         where: {
           id,
         },
@@ -129,35 +129,31 @@ export class FailRunTaskService {
         return existingTask;
       }
 
-      const task = await $transaction(prisma, async (tx) => {
-        if (existingTask.attempts.length === 1) {
-          await tx.taskAttempt.update({
-            where: {
-              id: existingTask.attempts[0].id,
-            },
-            data: {
-              status: "ERRORED",
-              error: formatError(taskBody.error),
-            },
-          });
-        }
-
-        return await prisma.task.update({
+      if (existingTask.attempts.length === 1) {
+        await tx.taskAttempt.update({
           where: {
-            id,
+            id: existingTask.attempts[0].id,
           },
           data: {
             status: "ERRORED",
-            output: taskBody.error ?? undefined,
-            completedAt: new Date(),
-          },
-          include: {
-            attempts: true,
+            error: formatError(taskBody.error),
           },
         });
-      });
+      }
 
-      return task;
+      return await tx.task.update({
+        where: {
+          id,
+        },
+        data: {
+          status: "ERRORED",
+          output: taskBody.error ?? undefined,
+          completedAt: new Date(),
+        },
+        include: {
+          attempts: true,
+        },
+      });
     });
 
     return task ? taskWithAttemptsToServerTask(task) : undefined;

apps/webapp/app/services/events/ingestSendEvent.server.ts

@@ -34,77 +34,55 @@ export class IngestSendEvent {
     try {
       const deliverAt = this.#calculateDeliverAt(options);
 
-      return await $transaction(
-        this.#prismaClient,
-        async (tx) => {
-          const externalAccount = options?.accountId
-            ? await tx.externalAccount.upsert({
-                where: {
-                  environmentId_identifier: {
-                    environmentId: environment.id,
-                    identifier: options.accountId,
-                  },
-                },
-                create: {
+      return await $transaction(this.#prismaClient, async (tx) => {
+        const externalAccount = options?.accountId
+          ? await tx.externalAccount.upsert({
+              where: {
+                environmentId_identifier: {
                   environmentId: environment.id,
-                  organizationId: environment.organizationId,
                   identifier: options.accountId,
                 },
-                update: {},
-              })
-            : undefined;
+              },
+              create: {
+                environmentId: environment.id,
+                organizationId: environment.organizationId,
+                identifier: options.accountId,
+              },
+              update: {},
+            })
+          : undefined;
 
-          // Create a new event in the database
-          const eventLog = await tx.eventRecord.create({
-            data: {
-              organization: {
-                connect: {
-                  id: environment.organizationId,
-                },
-              },
-              project: {
-                connect: {
-                  id: environment.projectId,
-                },
-              },
-              environment: {
-                connect: {
-                  id: environment.id,
-                },
-              },
-              eventId: event.id,
-              name: event.name,
-              timestamp: event.timestamp ?? new Date(),
-              payload: event.payload ?? {},
-              context: event.context ?? {},
-              source: event.source ?? "trigger.dev",
-              sourceContext,
-              deliverAt: deliverAt,
-              externalAccount: externalAccount
-                ? {
-                    connect: {
-                      id: externalAccount.id,
-                    },
-                  }
-                : {},
+        // Create a new event in the database
+        const eventLog = await tx.eventRecord.create({
+          data: {
+            organizationId: environment.organizationId,
+            projectId: environment.projectId,
+            environmentId: environment.id,
+            eventId: event.id,
+            name: event.name,
+            timestamp: event.timestamp ?? new Date(),
+            payload: event.payload ?? {},
+            context: event.context ?? {},
+            source: event.source ?? "trigger.dev",
+            sourceContext,
+            deliverAt: deliverAt,
+            externalAccountId: externalAccount ? externalAccount.id : undefined,
+          },
+        });
+
+        if (this.deliverEvents) {
+          // Produce a message to the event bus
+          await workerQueue.enqueue(
+            "deliverEvent",
+            {
+              id: eventLog.id,
             },
-          });
+            { runAt: eventLog.deliverAt, tx, jobKey: `event:${eventLog.id}` }
+          );
+        }
 
-          if (this.deliverEvents) {
-            // Produce a message to the event bus
-            await workerQueue.enqueue(
-              "deliverEvent",
-              {
-                id: eventLog.id,
-              },
-              { runAt: eventLog.deliverAt, tx, jobKey: `event:${eventLog.id}` }
-            );
-          }
-
-          return eventLog;
-        },
-        { rethrowPrismaErrors: true }
-      );
+        return eventLog;
+      });
     } catch (error) {
       const prismaError = PrismaErrorSchema.safeParse(error);
 

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

@@ -42,29 +42,32 @@ export class CreateRunService {
 
     return await $transaction(this.#prismaClient, async (tx) => {
       // Get the current max number for the given jobId
-      const currentMaxNumber = await tx.jobRun.aggregate({
+      const latestJob = await tx.jobRun.findFirst({
         where: { jobId: job.id },
-        _max: { number: true },
+        orderBy: { id: "desc" },
+        select: {
+          number: true,
+        },
       });
 
       // Increment the number for the new execution
-      const newNumber = (currentMaxNumber._max.number ?? 0) + 1;
+      const newNumber = (latestJob?.number ?? 0) + 1;
 
       // Create the new execution with the incremented number
       const run = await tx.jobRun.create({
         data: {
           number: newNumber,
           preprocess: version.preprocessRuns,
-          job: { connect: { id: job.id } },
-          version: { connect: { id: version.id } },
-          event: { connect: { id: eventId } },
-          environment: { connect: { id: environment.id } },
-          organization: { connect: { id: environment.organizationId } },
-          project: { connect: { id: environment.projectId } },
-          endpoint: { connect: { id: endpoint.id } },
-          queue: { connect: { id: jobQueue.id } },
-          externalAccount: eventRecord.externalAccountId
-            ? { connect: { id: eventRecord.externalAccountId } }
+          jobId: job.id,
+          versionId: version.id,
+          eventId: eventId,
+          environmentId: environment.id,
+          organizationId: environment.organizationId,
+          projectId: environment.projectId,
+          endpointId: endpoint.id,
+          queueId: jobQueue.id,
+          externalAccountId: eventRecord.externalAccountId
+            ? eventRecord.externalAccountId
             : undefined,
           isTest: eventRecord.isTest,
         },

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

@@ -161,7 +161,8 @@ function getWorkerQueue() {
     tasks: {
       "events.invokeDispatcher": {
         priority: 0, // smaller number = higher priority
-        maxAttempts: 3,
+        maxAttempts: 6,
+        queueName: (payload) => `dispatcher:${payload.id}`, // use a queue for a dispatcher so runs are created sequentially
         handler: async (payload, job) => {
           const service = new InvokeDispatcherService();
 

perf/src/index.ts

@@ -56,6 +56,32 @@ async function main() {
   // }
 }
 
+async function mainParallel() {
+  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 Promise.all([sendEvent(), sendEvent(), sendEvent(), sendEvent(), sendEvent()]);
+
+    await new Promise((resolve) => setTimeout(resolve, 250));
+  }
+
+  // console.log("Sending 30 events...");
+  // for (let i = 0; i < 30; i++) {
+  //   await sendEvent();
+  // }
+}
+
 async function mainLong() {
   console.log("Preparing long perf tests...");
 
@@ -95,7 +121,7 @@ async function mainSerial() {
   }
 }
 
-mainSerial().catch((err) => {
+mainParallel().catch((err) => {
   console.error(err);
   process.exit(1);
 });