test(run-ops): add cache + distinct-db sentinel tests; drop test enumeration labels

Add a pure unit test for ControlPlaneCache covering per-slot round-trips,
null-vs-miss distinction, epoch-based invalidation, per-slot key isolation,
bounded eviction, and TTL expiry. Add a testcontainer test for
probeDistinctDatabases covering distinct clusters, same physical database
(with reason), same-cluster-different-database, and fail-closed probe failure.

Strip developer-enumeration labels from three existing test files (readThrough
step numbers, runEngineHandlers Test-X comments) and rename the run-detail
loader read-through test to drop the non-domain "shape 1" name.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: d-cs

apps/webapp/app/v3/runOpsMigration/controlPlaneCache.server.test.ts

@@ -0,0 +1,143 @@
+import { describe, expect, it } from "vitest";
+import {
+  ControlPlaneCache,
+  type ResolvedAuthenticatedEnv,
+  type ResolvedEnv,
+  type ResolvedRunLockedWorker,
+  type ResolvedWorkerVersion,
+} from "./controlPlaneCache.server";
+
+// Minimal, structurally-irrelevant stand-ins: the cache stores and returns opaque values by
+// reference, so these only need to be distinguishable objects — the slot types are exercised for
+// key routing, not field shape.
+const anEnv = { id: "env_1" } as unknown as ResolvedEnv;
+const aVersion = { worker: { id: "bw_1" } } as unknown as ResolvedWorkerVersion;
+const anAuthEnv = { id: "env_1", slug: "prod" } as unknown as ResolvedAuthenticatedEnv;
+const aLockedWorker = { lockedBy: null, lockedToVersion: null } as ResolvedRunLockedWorker;
+
+describe("ControlPlaneCache", () => {
+  it("round-trips a value through every slot", () => {
+    const cache = new ControlPlaneCache({ ttlMs: 60_000, maxEntries: 100 });
+
+    cache.setEnv("env_1", anEnv);
+    cache.setWorkerVersion("env_1:current", aVersion);
+    cache.setEnvExists("env_1", true);
+    cache.setAuthEnv("env_1", anAuthEnv);
+    cache.setLockedWorker("bw_1:v_1", aLockedWorker);
+
+    expect(cache.getEnv("env_1")).toBe(anEnv);
+    expect(cache.getWorkerVersion("env_1:current")).toBe(aVersion);
+    expect(cache.getEnvExists("env_1")).toBe(true);
+    expect(cache.getAuthEnv("env_1")).toBe(anAuthEnv);
+    expect(cache.getLockedWorker("bw_1:v_1")).toBe(aLockedWorker);
+  });
+
+  it("returns undefined for a key that was never set, in every slot", () => {
+    const cache = new ControlPlaneCache({ ttlMs: 60_000, maxEntries: 100 });
+
+    expect(cache.getEnv("missing")).toBeUndefined();
+    expect(cache.getWorkerVersion("missing")).toBeUndefined();
+    expect(cache.getEnvExists("missing")).toBeUndefined();
+    expect(cache.getAuthEnv("missing")).toBeUndefined();
+    expect(cache.getLockedWorker("missing")).toBeUndefined();
+  });
+
+  it("distinguishes a cached null (confirmed absence) from an unset miss", () => {
+    const cache = new ControlPlaneCache({ ttlMs: 60_000, maxEntries: 100 });
+
+    expect(cache.getEnv("env_2")).toBeUndefined();
+    cache.setEnv("env_2", null);
+    expect(cache.getEnv("env_2")).toBeNull();
+
+    expect(cache.getAuthEnv("env_2")).toBeUndefined();
+    cache.setAuthEnv("env_2", null);
+    expect(cache.getAuthEnv("env_2")).toBeNull();
+
+    expect(cache.getWorkerVersion("env_2:current")).toBeUndefined();
+    cache.setWorkerVersion("env_2:current", null);
+    expect(cache.getWorkerVersion("env_2:current")).toBeNull();
+
+    expect(cache.getLockedWorker("_:_")).toBeUndefined();
+    cache.setLockedWorker("_:_", null);
+    expect(cache.getLockedWorker("_:_")).toBeNull();
+  });
+
+  it("caches a false env-existence result distinctly from an unset miss", () => {
+    const cache = new ControlPlaneCache({ ttlMs: 60_000, maxEntries: 100 });
+
+    expect(cache.getEnvExists("env_3")).toBeUndefined();
+    cache.setEnvExists("env_3", false);
+    expect(cache.getEnvExists("env_3")).toBe(false);
+  });
+
+  it("invalidateEnv forces the next getEnv to miss", () => {
+    const cache = new ControlPlaneCache({ ttlMs: 60_000, maxEntries: 100 });
+
+    cache.setEnv("env_4", anEnv);
+    expect(cache.getEnv("env_4")).toBe(anEnv);
+
+    cache.invalidateEnv("env_4");
+    expect(cache.getEnv("env_4")).toBeUndefined();
+  });
+
+  it("makes a re-setEnv after invalidation readable again", () => {
+    const cache = new ControlPlaneCache({ ttlMs: 60_000, maxEntries: 100 });
+    const replacement = { id: "env_5b" } as unknown as ResolvedEnv;
+
+    cache.setEnv("env_5", anEnv);
+    cache.invalidateEnv("env_5");
+    expect(cache.getEnv("env_5")).toBeUndefined();
+
+    cache.setEnv("env_5", replacement);
+    expect(cache.getEnv("env_5")).toBe(replacement);
+  });
+
+  it("invalidateEnv is scoped to its own id", () => {
+    const cache = new ControlPlaneCache({ ttlMs: 60_000, maxEntries: 100 });
+    const other = { id: "env_keep" } as unknown as ResolvedEnv;
+
+    cache.setEnv("env_drop", anEnv);
+    cache.setEnv("env_keep", other);
+    cache.invalidateEnv("env_drop");
+
+    expect(cache.getEnv("env_drop")).toBeUndefined();
+    expect(cache.getEnv("env_keep")).toBe(other);
+  });
+
+  it("does not collide keys across slots for the same id", () => {
+    const cache = new ControlPlaneCache({ ttlMs: 60_000, maxEntries: 100 });
+
+    cache.setEnv("x", anEnv);
+    cache.setEnvExists("x", true);
+    cache.setAuthEnv("x", anAuthEnv);
+
+    expect(cache.getEnv("x")).toBe(anEnv);
+    expect(cache.getEnvExists("x")).toBe(true);
+    expect(cache.getAuthEnv("x")).toBe(anAuthEnv);
+
+    // Invalidating the env slot leaves the sibling slots for the same id intact.
+    cache.invalidateEnv("x");
+    expect(cache.getEnv("x")).toBeUndefined();
+    expect(cache.getEnvExists("x")).toBe(true);
+    expect(cache.getAuthEnv("x")).toBe(anAuthEnv);
+  });
+
+  it("evicts the oldest entry once maxEntries is exceeded", () => {
+    const cache = new ControlPlaneCache({ ttlMs: 60_000, maxEntries: 2 });
+
+    cache.setEnv("first", { id: "first" } as unknown as ResolvedEnv);
+    cache.setEnv("second", { id: "second" } as unknown as ResolvedEnv);
+    cache.setEnv("third", { id: "third" } as unknown as ResolvedEnv);
+
+    expect(cache.getEnv("first")).toBeUndefined();
+    expect(cache.getEnv("second")).toMatchObject({ id: "second" });
+    expect(cache.getEnv("third")).toMatchObject({ id: "third" });
+  });
+
+  it("treats a zero-TTL entry as immediately expired", () => {
+    const cache = new ControlPlaneCache({ ttlMs: 0, maxEntries: 100 });
+
+    cache.setEnv("env_ttl", anEnv);
+    expect(cache.getEnv("env_ttl")).toBeUndefined();
+  });
+});

apps/webapp/app/v3/runOpsMigration/readThrough.server.test.ts

@@ -40,7 +40,7 @@ function toHttpish<T>(result: ReadThroughResult<T>): { status: number; value?: T
 
 describe("readThroughRun (legacy replica + new DB)", () => {
   heteroPostgresTest(
-    "Step 1: old in-retention run is served from the legacy REPLICA, never a primary",
+    "old in-retention run is served from the legacy REPLICA, never a primary",
     async ({ prisma14, prisma17 }) => {
       // legacy hit, new miss. The layer has NO legacy-writer handle at all — the
       // read resolving through `legacyReplica` (prisma14) IS the structural guarantee
@@ -63,7 +63,7 @@ describe("readThroughRun (legacy replica + new DB)", () => {
   );
 
   heteroPostgresTest(
-    "Step 3: post-termination past-retention returns the normal not-found surface",
+    "post-termination past-retention returns the normal not-found surface",
     async ({ prisma14, prisma17 }) => {
       const pastRetentionResult = await readThroughRun({
         runId: LEGACY_RUN_ID,
@@ -102,7 +102,7 @@ describe("readThroughRun (legacy replica + new DB)", () => {
   );
 
   heteroPostgresTest(
-    "Step 4: single-DB passthrough — only readNew runs, legacy never touched",
+    "single-DB passthrough — only readNew runs, legacy never touched",
     async ({ prisma14, prisma17 }) => {
       const throwingLegacy = vi.fn(async (): Promise<{ marker: number } | null> => {
         throw new Error("readLegacy must never run in single-DB mode");
@@ -128,7 +128,7 @@ describe("readThroughRun (legacy replica + new DB)", () => {
   );
 
   heteroPostgresTest(
-    "Step 5: new-residency fast-path — legacy replica is never touched",
+    "new-residency fast-path — legacy replica is never touched",
     async ({ prisma14, prisma17 }) => {
       const throwingLegacy = vi.fn(async (): Promise<{ marker: number } | null> => {
         throw new Error("readLegacy must never run for a NEW-residency id");

…Loaders.controlPlane.readthrough.test.ts …Loaders.controlPlane.readthrough.test.tsapps/webapp/test/shape1RunDetailLoaders.controlPlane.readthrough.test.ts renamed to apps/webapp/test/runDetailLoaders.controlPlane.readthrough.test.ts apps/webapp/test/shape1RunDetailLoaders.controlPlane.readthrough.test.ts renamed to apps/webapp/test/runDetailLoaders.controlPlane.readthrough.test.ts

@@ -1,4 +1,4 @@
-// Dedicated run-ops proof: Shape-1 run-detail loaders read the run by friendlyId on the dedicated
+// Dedicated run-ops proof: the run-detail page loaders read the run by friendlyId on the dedicated
 // run-ops client (PG17, subset schema with no control-plane tables), then authorize membership +
 // resolve env on PG14. Neither DB joins the other.
 import { heteroRunOpsPostgresTest } from "@internal/testcontainers";
@@ -48,8 +48,8 @@ async function seedAll(prisma: PrismaClient) {
   return { organization, project, environment, member, stranger };
 }
 
-// [TEST-NEWSEED] The run lives on the dedicated run-ops client; its control-plane FKs are synthetic
-// scalar ids pointing at rows that exist only on PG14 (the dedicated DB has no such tables).
+// The run lives on the dedicated run-ops client; its control-plane FKs are synthetic scalar ids
+// pointing at rows that exist only on PG14 (the dedicated DB has no such tables).
 async function seedKsuidRun(prisma17: RunOpsPrismaClient, cp: Awaited<ReturnType<typeof seedAll>>) {
   const k = n++;
   return prisma17.taskRun.create({
@@ -61,10 +61,10 @@ async function seedKsuidRun(prisma17: RunOpsPrismaClient, cp: Awaited<ReturnType
       runtimeEnvironmentId: cp.environment.id,
       projectId: cp.project.id,
       organizationId: cp.organization.id,
-      taskIdentifier: "shape1-task",
+      taskIdentifier: "run-detail-task",
       payload: "{}",
       payloadType: "application/json",
-      queue: "task/shape1-task",
+      queue: "task/run-detail-task",
       idempotencyKey: "idem-1",
       spanId: `sp_${k}`,
       traceId: `tr_${k}`,
@@ -89,7 +89,7 @@ function wire(prisma14: PrismaClient, prisma17: RunOpsPrismaClient) {
   return { runStore, resolver };
 }
 
-describe("Shape-1 run-detail loaders cross-DB read-through (dedicated run-ops client)", () => {
+describe("run-detail loaders cross-DB read-through (dedicated run-ops client)", () => {
   heteroRunOpsPostgresTest(
     "ksuid run resolves: friendlyId read on the dedicated run-ops DB + membership/env auth on PG14 (resources.runs.$runParam shape)",
     async ({ prisma14, prisma17 }) => {

apps/webapp/test/runEngineHandlers.test.ts

@@ -151,7 +151,7 @@ function failure(index: number, errorCode: string, extra?: Record<string, unknow
 }
 
 describe("runEngineHandlers read-through", () => {
-  // Test A: a NEW run resolves via read-through against the new store.
+  // A NEW run resolves via read-through against the new store.
   containerTest("event read resolves a NEW run via read-through", async ({ prisma }) => {
     const store = new PostgresRunStore({ prisma, readOnlyPrisma: prisma });
     const { organization, project, environment } = await seedEnvironment(prisma, "a");
@@ -181,7 +181,7 @@ describe("runEngineHandlers read-through", () => {
     expect(run!.taskEventStore).toBe("taskEvent");
   });
 
-  // Test C: single-DB short-circuit — readLegacy must never be invoked.
+  // Single-DB short-circuit — readLegacy must never be invoked.
   containerTest("single-DB short-circuit never touches a legacy handle", async ({ prisma }) => {
     const store = new PostgresRunStore({ prisma, readOnlyPrisma: prisma });
     const { organization, project, environment } = await seedEnvironment(prisma, "c");
@@ -237,8 +237,8 @@ describe("runEngineHandlers read-through", () => {
 });
 
 describe("runEngineHandlers read-through cross-version", () => {
-  // Test B (heterogeneous recast): an OLD in-retention run is served off the LEGACY
-  // REPLICA only, and the legacy primary/writer is structurally absent.
+  // An OLD in-retention run is served off the LEGACY REPLICA only, and the legacy
+  // primary/writer is structurally absent.
   heteroPostgresTest(
     "event read resolves an OLD in-retention run via the legacy replica",
     async ({ prisma14, prisma17 }) => {
@@ -378,7 +378,7 @@ describe("runEngineHandlers batch completion", () => {
     expect(batch.processingCompletedAt).toBeNull();
   });
 
-  // Test E: callback retry is idempotent via skipDuplicates.
+  // Callback retry is idempotent via skipDuplicates.
   containerTest("batch txn is idempotent on callback retry", async ({ prisma }) => {
     const { environment } = await seedEnvironment(prisma, "e");
     const batchId = "e".repeat(25);
@@ -404,7 +404,7 @@ describe("runEngineHandlers batch completion", () => {
     expect(errors).toHaveLength(2);
   });
 
-  // Test I: aggregate fast-path collapses same-errorCode failures to one row.
+  // Aggregate fast-path collapses same-errorCode failures to one row.
   containerTest("aggregate fast-path collapses queue-size-limit failures", async ({ prisma }) => {
     const { environment } = await seedEnvironment(prisma, "i");
     const batchId = "f".repeat(25);
@@ -510,8 +510,8 @@ describe("runEngineHandlers batch residency routing", () => {
     expect(writer).toBe(prisma);
   });
 
-  // Test G (heterogeneous recast): a legacy-resident batch (row only on the legacy DB) commits on
-  // the LEGACY writer; the NEW DB is left with zero rows for the batch.
+  // A legacy-resident batch (row only on the legacy DB) commits on the LEGACY writer;
+  // the NEW DB is left with zero rows for the batch.
   heteroPostgresTest(
     "legacy-resident batch routes to the LEGACY writer, new DB untouched",
     async ({ prisma14, prisma17 }) => {
@@ -617,8 +617,7 @@ describe("runEngineHandlers batch residency routing", () => {
     }
   );
 
-  // Test H (heterogeneous recast): a new batch (row only on the new DB) commits on the NEW
-  // writer; the LEGACY DB is untouched.
+  // A new batch (row only on the new DB) commits on the NEW writer; the LEGACY DB is untouched.
   heteroPostgresTest(
     "new batch routes to the NEW writer, legacy DB untouched",
     async ({ prisma14, prisma17 }) => {

apps/webapp/test/v3/runOpsMigration/distinctDbSentinel.server.test.ts

@@ -0,0 +1,62 @@
+import { heteroPostgresTest } from "@internal/testcontainers";
+import { PrismaClient } from "@trigger.dev/database";
+import { describe, expect, vi } from "vitest";
+import { probeDistinctDatabases } from "~/v3/runOpsMigration/distinctDbSentinel.server";
+
+// Spinning up two separate postgres clusters and probing each can exceed the 5s default.
+vi.setConfig({ testTimeout: 60_000 });
+
+function urlWithDatabase(uri: string, database: string): string {
+  const url = new URL(uri);
+  url.pathname = `/${database}`;
+  return url.toString();
+}
+
+describe("probeDistinctDatabases", () => {
+  heteroPostgresTest(
+    "reports distinct for two separate physical clusters",
+    async ({ uri14, uri17 }) => {
+      const result = await probeDistinctDatabases(uri14, uri17);
+      expect(result).toEqual({ distinct: true });
+    }
+  );
+
+  heteroPostgresTest(
+    "reports NOT distinct, citing the same physical database, when both URLs point at it",
+    async ({ uri14 }) => {
+      const result = await probeDistinctDatabases(uri14, uri14);
+      expect(result.distinct).toBe(false);
+      if (result.distinct === false) {
+        expect(result.reason).toMatch(/same physical database/i);
+      }
+    }
+  );
+
+  heteroPostgresTest(
+    "reports distinct for two databases in the SAME cluster",
+    async ({ postgresContainer14, uri14 }) => {
+      const otherDb = `sentinel_other_${Date.now()}`;
+      const admin = new PrismaClient({
+        datasources: { db: { url: urlWithDatabase(postgresContainer14.getConnectionUri(), "postgres") } },
+      });
+      try {
+        await admin.$executeRawUnsafe(`CREATE DATABASE "${otherDb}"`);
+      } finally {
+        await admin.$disconnect();
+      }
+
+      const otherUrl = urlWithDatabase(uri14, otherDb);
+      const result = await probeDistinctDatabases(uri14, otherUrl);
+      expect(result).toEqual({ distinct: true });
+    }
+  );
+
+  heteroPostgresTest(
+    "fails closed to NOT distinct when a probe cannot reach a database",
+    async ({ uri14 }) => {
+      const unreachable = "postgresql://nobody:nobody@127.0.0.1:1/does_not_exist";
+      const result = await probeDistinctDatabases(uri14, unreachable);
+      expect(result.distinct).toBe(false);
+    }
+  );
+});