fix(webapp,run-store): match batch idempotency cache hits by task and key

Cached-run lookups from all task groups are flattened into one array, so
matching by idempotency key alone could return a run from a different task
that reused the same key string. Match on taskIdentifier and key. Adds a
testcontainers test covering findRunsByIdempotencyKeys (multi-key UNION ALL,
task scoping, idempotencyKeyExpiresAt deserialization, empty short-circuit).

Author: ericallam

apps/webapp/app/v3/services/batchTriggerV3.server.ts

@@ -423,11 +423,13 @@ export class BatchTriggerV3Service extends BaseService {
     const cachedRuns = (
       await pMap(
         idempotencyKeyLookups,
-        ({ taskIdentifier, idempotencyKeys }) =>
-          this.runStore.findRunsByIdempotencyKeys(
+        async ({ taskIdentifier, idempotencyKeys }) => {
+          const rows = await this.runStore.findRunsByIdempotencyKeys(
             { runtimeEnvironmentId: environment.id, taskIdentifier, idempotencyKeys },
             this._prisma
-          ),
+          );
+          return rows.map((row) => ({ ...row, taskIdentifier }));
+        },
         { concurrency: IDEMPOTENCY_KEY_LOOKUP_CONCURRENCY }
       )
     ).flat();
@@ -438,7 +440,9 @@ export class BatchTriggerV3Service extends BaseService {
 
     const runs = await Promise.all(
       body.items.map(async (item) => {
-        const cachedRun = cachedRuns.find((r) => r.idempotencyKey === item.options?.idempotencyKey);
+        const cachedRun = cachedRuns.find(
+          (r) => r.taskIdentifier === item.task && r.idempotencyKey === item.options?.idempotencyKey
+        );
 
         if (cachedRun) {
           if (cachedRun.idempotencyKeyExpiresAt && cachedRun.idempotencyKeyExpiresAt < new Date()) {

internal-packages/run-store/src/PostgresRunStore.findRunsByIdempotencyKeys.test.ts

@@ -0,0 +1,120 @@
+import { postgresTest } from "@internal/testcontainers";
+import type { PrismaClient } from "@trigger.dev/database";
+import { describe, expect } from "vitest";
+import { PostgresRunStore } from "./PostgresRunStore.js";
+
+async function seedEnvironment(prisma: PrismaClient) {
+  const organization = await prisma.organization.create({
+    data: { title: "Test Organization", slug: "test-organization" },
+  });
+  const project = await prisma.project.create({
+    data: {
+      name: "Test Project",
+      slug: "test-project",
+      externalRef: "proj_1234",
+      organizationId: organization.id,
+    },
+  });
+  const environment = await prisma.runtimeEnvironment.create({
+    data: {
+      type: "DEVELOPMENT",
+      slug: "dev",
+      projectId: project.id,
+      organizationId: organization.id,
+      apiKey: "tr_dev_apikey",
+      pkApiKey: "pk_dev_apikey",
+      shortcode: "short_code",
+    },
+  });
+  return { organization, project, environment };
+}
+
+async function createRun(
+  prisma: PrismaClient,
+  params: {
+    runtimeEnvironmentId: string;
+    projectId: string;
+    friendlyId: string;
+    taskIdentifier: string;
+    idempotencyKey: string;
+    idempotencyKeyExpiresAt?: Date;
+  }
+) {
+  await prisma.taskRun.create({
+    data: {
+      friendlyId: params.friendlyId,
+      taskIdentifier: params.taskIdentifier,
+      idempotencyKey: params.idempotencyKey,
+      idempotencyKeyExpiresAt: params.idempotencyKeyExpiresAt ?? null,
+      payload: "{}",
+      payloadType: "application/json",
+      runtimeEnvironmentId: params.runtimeEnvironmentId,
+      projectId: params.projectId,
+      queue: `task/${params.taskIdentifier}`,
+      traceId: `trace_${params.friendlyId}`,
+      spanId: `span_${params.friendlyId}`,
+      engine: "V2",
+    },
+  });
+}
+
+describe("PostgresRunStore.findRunsByIdempotencyKeys", () => {
+  postgresTest("resolves multiple keys, scoped to (env, task)", async ({ prisma }) => {
+    const { project, environment } = await seedEnvironment(prisma);
+    const store = new PostgresRunStore({ prisma, readOnlyPrisma: prisma });
+
+    const expiresAt = new Date("2999-01-01T00:00:00.000Z");
+    await createRun(prisma, {
+      runtimeEnvironmentId: environment.id,
+      projectId: project.id,
+      friendlyId: "run_a1",
+      taskIdentifier: "task-a",
+      idempotencyKey: "idem-1",
+      idempotencyKeyExpiresAt: expiresAt,
+    });
+    await createRun(prisma, {
+      runtimeEnvironmentId: environment.id,
+      projectId: project.id,
+      friendlyId: "run_a2",
+      taskIdentifier: "task-a",
+      idempotencyKey: "idem-2",
+    });
+    await createRun(prisma, {
+      runtimeEnvironmentId: environment.id,
+      projectId: project.id,
+      friendlyId: "run_b1",
+      taskIdentifier: "task-b",
+      idempotencyKey: "idem-1",
+    });
+
+    const rows = await store.findRunsByIdempotencyKeys({
+      runtimeEnvironmentId: environment.id,
+      taskIdentifier: "task-a",
+      idempotencyKeys: ["idem-1", "idem-2", "does-not-exist"],
+    });
+
+    const byKey = new Map(rows.map((r) => [r.idempotencyKey, r]));
+    expect(rows).toHaveLength(2);
+    expect(byKey.get("idem-1")?.friendlyId).toBe("run_a1");
+    expect(byKey.get("idem-2")?.friendlyId).toBe("run_a2");
+    expect(rows.map((r) => r.friendlyId)).not.toContain("run_b1");
+    expect(byKey.get("idem-1")?.idempotencyKeyExpiresAt).toBeInstanceOf(Date);
+    expect(byKey.get("idem-1")?.idempotencyKeyExpiresAt?.toISOString()).toBe(
+      expiresAt.toISOString()
+    );
+    expect(byKey.get("idem-2")?.idempotencyKeyExpiresAt).toBeNull();
+  });
+
+  postgresTest("short-circuits on an empty key list without querying", async ({ prisma }) => {
+    const { environment } = await seedEnvironment(prisma);
+    const store = new PostgresRunStore({ prisma, readOnlyPrisma: prisma });
+
+    const rows = await store.findRunsByIdempotencyKeys({
+      runtimeEnvironmentId: environment.id,
+      taskIdentifier: "task-a",
+      idempotencyKeys: [],
+    });
+
+    expect(rows).toEqual([]);
+  });
+});