diff --git a/apps/webapp/app/db.server.ts b/apps/webapp/app/db.server.ts index a6a58527c..171a5b79d 100644 --- a/apps/webapp/app/db.server.ts +++ b/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( @@ -55,11 +55,9 @@ export async function $transaction( 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", + // }, ], }); diff --git a/apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.complete.ts b/apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.complete.ts index 4b3d20109..f116261cb 100644 --- a/apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.complete.ts +++ b/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 { // 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; diff --git a/apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.fail.ts b/apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.fail.ts index e0ddb26dd..8bdfb4f44 100644 --- a/apps/webapp/app/routes/api.v1.runs.$runId.tasks.$id.fail.ts +++ b/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 { // 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; diff --git a/apps/webapp/app/services/events/ingestSendEvent.server.ts b/apps/webapp/app/services/events/ingestSendEvent.server.ts index f2c5e7933..b1de0e2a9 100644 --- a/apps/webapp/app/services/events/ingestSendEvent.server.ts +++ b/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); diff --git a/apps/webapp/app/services/runs/createRun.server.ts b/apps/webapp/app/services/runs/createRun.server.ts index 34d8cd4e7..a3c789193 100644 --- a/apps/webapp/app/services/runs/createRun.server.ts +++ b/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, }, diff --git a/apps/webapp/app/services/worker.server.ts b/apps/webapp/app/services/worker.server.ts index 93fc0753a..59c5e8b52 100644 --- a/apps/webapp/app/services/worker.server.ts +++ b/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(); diff --git a/perf/src/index.ts b/perf/src/index.ts index 8c1ce8f7b..6d9e6cd2a 100644 --- a/perf/src/index.ts +++ b/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); });