diff --git a/apps/webapp/app/components/run/RunOverview.tsx b/apps/webapp/app/components/run/RunOverview.tsx index 2b32f064c..0b8fde08d 100644 --- a/apps/webapp/app/components/run/RunOverview.tsx +++ b/apps/webapp/app/components/run/RunOverview.tsx @@ -98,7 +98,9 @@ export function RunOverview({ run, trigger, showRerun, paths }: RunOverviewProps to: paths.back, text: "Runs", }} - title={`Run #${run.number}`} + title={ + typeof run.number === "number" ? `Run #${run.number}` : `Run ${run.id.slice(0, 8)}` + } /> {run.isTest && ( diff --git a/apps/webapp/app/components/runs/RunsTable.tsx b/apps/webapp/app/components/runs/RunsTable.tsx index 879bf3505..7e6fa9f0a 100644 --- a/apps/webapp/app/components/runs/RunsTable.tsx +++ b/apps/webapp/app/components/runs/RunsTable.tsx @@ -20,7 +20,7 @@ import { RunStatus } from "./RunStatuses"; type RunTableItem = { id: string; - number: number; + number: number | null; environment: { type: RuntimeEnvironmentType; }; @@ -78,7 +78,9 @@ export function RunsTable({ const path = `${runsParentPath}/${run.id}/trigger`; return ( - #{run.number} + + {typeof run.number === "number" ? `#${run.number}` : "-"} + diff --git a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.jobs.$jobParam.test/route.tsx b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.jobs.$jobParam.test/route.tsx index 1a76ae80a..e7638fb17 100644 --- a/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.jobs.$jobParam.test/route.tsx +++ b/apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.jobs.$jobParam.test/route.tsx @@ -297,7 +297,9 @@ export default function Page() { label={} description={ <> - Run #{run.number}{" "} + {typeof run.number === "number" + ? `Run #${run.number}` + : `Run ${run.id.slice(0, 8)}`} {runStatusTitle(run.status).toLocaleLowerCase()} diff --git a/apps/webapp/app/services/runs/createRun.server.ts b/apps/webapp/app/services/runs/createRun.server.ts index ee82ba887..728460ae3 100644 --- a/apps/webapp/app/services/runs/createRun.server.ts +++ b/apps/webapp/app/services/runs/createRun.server.ts @@ -44,22 +44,8 @@ export class CreateRunService { }); return await $transaction(this.#prismaClient, async (tx) => { - // Get the current max number for the given jobId - const latestJob = await tx.jobRun.findFirst({ - where: { jobId: job.id }, - orderBy: { id: "desc" }, - select: { - number: true, - }, - }); - - // Increment the number for the new execution - 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, jobId: job.id, versionId: version.id, @@ -101,7 +87,7 @@ export class CreateRunService { { id: run.id, }, - { tx, queueName: `startRun:${run.jobId}` } + { tx } ); return run; diff --git a/apps/webapp/app/services/runs/performRunExecutionV3.server.ts b/apps/webapp/app/services/runs/performRunExecutionV3.server.ts index 2cc4ee413..cda07cf40 100644 --- a/apps/webapp/app/services/runs/performRunExecutionV3.server.ts +++ b/apps/webapp/app/services/runs/performRunExecutionV3.server.ts @@ -99,7 +99,7 @@ export class PerformRunExecutionV3Service { jobKey: `job_run:EXECUTE_JOB:${run.id}`, maxAttempts: options.skipRetrying ? 1 : undefined, flags: [`rl:executions:${run.organizationId}`], - priority: run.number, + priority: run.number ?? 0, } ); } diff --git a/apps/webapp/app/services/runs/resumeRun.server.ts b/apps/webapp/app/services/runs/resumeRun.server.ts index 26e706d2c..9ecbc9944 100644 --- a/apps/webapp/app/services/runs/resumeRun.server.ts +++ b/apps/webapp/app/services/runs/resumeRun.server.ts @@ -114,9 +114,10 @@ export class ResumeRunService { }, { tx, - runAt: runAt, + runAt: runAt ?? run.createdAt, queueName: `run_resume:${run.id}`, jobKey: `run_resume:${run.id}`, + priority: run.number ?? 0, } ); } diff --git a/apps/webapp/app/services/runs/startRun.server.ts b/apps/webapp/app/services/runs/startRun.server.ts index cf5195497..62f9f0ed1 100644 --- a/apps/webapp/app/services/runs/startRun.server.ts +++ b/apps/webapp/app/services/runs/startRun.server.ts @@ -4,9 +4,10 @@ import { type IntegrationConnection, } from "@trigger.dev/database"; import type { PrismaClient, PrismaClientOrTransaction } from "~/db.server"; -import { prisma } from "~/db.server"; +import { $transaction, prisma } from "~/db.server"; import { workerQueue } from "../worker.server"; import { ResumeRunService } from "./resumeRun.server"; +import { createHash } from "node:crypto"; type FoundRun = NonNullable>>; type RunConnectionsByKey = Awaited>; @@ -58,19 +59,36 @@ export class StartRunService { : undefined ) .filter(Boolean); + const lockId = jobIdToLockId(run.jobId); - const updatedRun = await this.#prismaClient.jobRun.update({ - where: { id }, - data: { - status: "QUEUED", - queuedAt: new Date(), - runConnections: { - create: createRunConnections, - }, + await $transaction( + this.#prismaClient, + async (tx) => { + await tx.$executeRaw`SELECT pg_advisory_xact_lock(${lockId})`; + + const counter = await tx.jobCounter.upsert({ + where: { jobId: run.jobId }, + update: { lastNumber: { increment: 1 } }, + create: { jobId: run.jobId, lastNumber: 1 }, + select: { lastNumber: true }, + }); + + const updatedRun = await this.#prismaClient.jobRun.update({ + where: { id }, + data: { + number: counter.lastNumber, + status: "QUEUED", + queuedAt: new Date(), + runConnections: { + create: createRunConnections, + }, + }, + }); + + await ResumeRunService.enqueue(updatedRun, tx); }, - }); - - await ResumeRunService.enqueue(updatedRun, this.#prismaClient); + { timeout: 60000 } + ); } async #handleMissingConnections(id: string, runConnectionsByKey: RunConnectionsByKey) { @@ -217,3 +235,8 @@ async function createRunConnections(tx: PrismaClientOrTransaction, run: FoundRun function hasMissingConnections(runConnectionsByKey: RunConnectionsByKey) { return Object.values(runConnectionsByKey).some((connection) => connection.result === "missing"); } + +function jobIdToLockId(jobId: string): number { + // Convert jobId to a unique lock identifier + return parseInt(createHash("sha256").update(jobId).digest("hex").slice(0, 8), 16); +} diff --git a/packages/database/prisma/migrations/20231121144353_make_job_run_number_optional/migration.sql b/packages/database/prisma/migrations/20231121144353_make_job_run_number_optional/migration.sql new file mode 100644 index 000000000..f8979aaa8 --- /dev/null +++ b/packages/database/prisma/migrations/20231121144353_make_job_run_number_optional/migration.sql @@ -0,0 +1,2 @@ +-- AlterTable +ALTER TABLE "JobRun" ALTER COLUMN "number" DROP NOT NULL; diff --git a/packages/database/prisma/migrations/20231121154359_add_job_counter_table/migration.sql b/packages/database/prisma/migrations/20231121154359_add_job_counter_table/migration.sql new file mode 100644 index 000000000..6f7b36264 --- /dev/null +++ b/packages/database/prisma/migrations/20231121154359_add_job_counter_table/migration.sql @@ -0,0 +1,7 @@ +-- CreateTable +CREATE TABLE "JobCounter" ( + "jobId" TEXT NOT NULL, + "lastNumber" INTEGER NOT NULL DEFAULT 0, + + CONSTRAINT "JobCounter_pkey" PRIMARY KEY ("jobId") +); diff --git a/packages/database/prisma/migrations/20231121154545_seed_job_counter_tables/migration.sql b/packages/database/prisma/migrations/20231121154545_seed_job_counter_tables/migration.sql new file mode 100644 index 000000000..8bc544ef5 --- /dev/null +++ b/packages/database/prisma/migrations/20231121154545_seed_job_counter_tables/migration.sql @@ -0,0 +1,10 @@ +-- This is an empty migration. +INSERT INTO + "JobCounter" ("jobId", "lastNumber") +SELECT + "jobId", + MAX(number) +FROM + "JobRun" +GROUP BY + "jobId"; \ No newline at end of file diff --git a/packages/database/prisma/schema.prisma b/packages/database/prisma/schema.prisma index 2851393b6..3e058ad7b 100644 --- a/packages/database/prisma/schema.prisma +++ b/packages/database/prisma/schema.prisma @@ -707,7 +707,7 @@ enum PayloadType { model JobRun { id String @id @default(cuid()) - number Int + number Int? internal Boolean @default(false) job Job @relation(fields: [jobId], references: [id], onDelete: Cascade, onUpdate: Cascade) @@ -788,6 +788,11 @@ enum JobRunStatus { INVALID_PAYLOAD } +model JobCounter { + jobId String @id + lastNumber Int @default(0) +} + model JobRunAutoYieldExecution { id String @id @default(cuid()) diff --git a/perf/src/index.ts b/perf/src/index.ts index c3edeacad..498995875 100644 --- a/perf/src/index.ts +++ b/perf/src/index.ts @@ -108,8 +108,8 @@ async function mainParallel() { async function mainParallelBulk() { const batches = 1; - const concurrency = 50; - const eventsPer = 20; + const concurrency = 10; + const eventsPer = 10; console.log("Preparing perf tests..."); @@ -169,7 +169,7 @@ async function mainSerial() { } } -mainParallelBulk().catch((err) => { +main().catch((err) => { console.error(err); process.exit(1); });