Improve run number generation using advistory locks, and only on start

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)}`
+            }
           />
           <PageButtons>
             {run.isTest && (

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 (
               <TableRow key={run.id}>
-                <TableCell to={path}>#{run.number}</TableCell>
+                <TableCell to={path}>
+                  {typeof run.number === "number" ? `#${run.number}` : "-"}
+                </TableCell>
                 <TableCell to={path}>
                   <EnvironmentLabel environment={run.environment} />
                 </TableCell>

apps/webapp/app/routes/_app.orgs.$organizationSlug.projects.$projectParam.jobs.$jobParam.test/route.tsx

@@ -297,7 +297,9 @@ export default function Page() {
                           label={<DateTime date={run.created} />}
                           description={
                             <>
-                              Run #{run.number}{" "}
+                              {typeof run.number === "number"
+                                ? `Run #${run.number}`
+                                : `Run ${run.id.slice(0, 8)}`}
                               <span className={runStatusClassNameColor(run.status)}>
                                 {runStatusTitle(run.status).toLocaleLowerCase()}
                               </span>

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;

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,
       }
     );
   }

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,
       }
     );
   }

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<Awaited<ReturnType<typeof findRun>>>;
 type RunConnectionsByKey = Awaited<ReturnType<typeof createRunConnections>>;
@@ -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);
+}

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;

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")
+);

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";

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())
 

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);
 });