fix(webapp): coalesce concurrent shard-list refreshes, and use the real txn helper

Two review findings.

The cache had a lost-update race. Two misses each issued a read, so a slower
read landing after a faster one wrote its older snapshot back into the cache
for a whole TTL. Refreshes are now single-flight: concurrent misses await one
read, and the in-flight handle is cleared on settle so a failure still lets
the next call retry. Three tests cover it with a deferred promise through the
injected reader, no mocking.

The admin action test's transaction stand-in was a reimplementation. It now
delegates to the same shared helper the production wrapper wraps, so the
transactional semantics, the nesting case and the retry behaviour are the real
ones. Only the wrapper's tracing span and infrastructure-error logging are
absent, and neither is asserted there.

Author: d-cs

apps/webapp/app/v3/runOpsMigration/mintShardAssignment.test.ts

@@ -319,6 +319,66 @@ describe("resolveMintShardWith — cache, read failure and fail-safe", () => {
     expect(await resolveMintShardWith({ id: "env_1" }, deps)).toBe("new");
   });
 
+  it("coalesces concurrent misses into ONE read", async () => {
+    // Two misses must share a single read. Otherwise a slower read landing after a faster one
+    // writes its older snapshot back into the cache for a whole TTL.
+    let release: (flags: Record<string, unknown>) => void = () => {};
+    const gate = new Promise<Record<string, unknown>>((resolve) => {
+      release = resolve;
+    });
+    const deps = wrapperDeps({ readFlags: () => gate });
+
+    const both = Promise.all([
+      resolveMintShardWith({ id: "env_1" }, deps),
+      resolveMintShardWith({ id: "env_2" }, deps),
+    ]);
+    release({ runOpsMintShardSet: "a,b" });
+    await both;
+
+    expect(deps.reads).toBe(1);
+  });
+
+  it("does not let a slower read overwrite a newer one", async () => {
+    // The slow read starts first and finishes last. Its result must not become the cached
+    // value, because the fast read already published a newer snapshot.
+    let releaseSlow: (flags: Record<string, unknown>) => void = () => {};
+    const slow = new Promise<Record<string, unknown>>((resolve) => {
+      releaseSlow = resolve;
+    });
+    let call = 0;
+    const deps = wrapperDeps({
+      readFlags: () => {
+        call++;
+        return call === 1 ? slow : Promise.resolve({ runOpsMintShardSet: "c" });
+      },
+    });
+
+    const first = resolveMintShardWith({ id: "env_1" }, deps);
+    const second = resolveMintShardWith({ id: "env_2" }, deps);
+    releaseSlow({ runOpsMintShardSet: "a" });
+    await Promise.all([first, second]);
+
+    // One read served both, so there is no second snapshot to race with.
+    expect(deps.reads).toBe(1);
+    expect(deps.cache.current?.value.resolution.set).toEqual(["a"]);
+  });
+
+  it("clears the in-flight refresh after a failure, so the next call retries", async () => {
+    let fail = true;
+    const deps = wrapperDeps({
+      readFlags: async () => {
+        if (fail) throw new Error("db down");
+        return { runOpsMintShardSet: "a,b" };
+      },
+    });
+    deps.onReadFailed = () => {};
+
+    expect(await resolveMintShardWith({ id: "env_1" }, deps)).toBe("new");
+    fail = false;
+    expect(["a", "b"]).toContain(await resolveMintShardWith({ id: "env_1" }, deps));
+    expect(deps.reads).toBe(2);
+  });
+
   it("agrees with the pure core for the same inputs", async () => {
     const deps = wrapperDeps();
     const viaWrapper = await resolveMintShardWith({ id: "env_1" }, deps);

apps/webapp/app/v3/runOpsMigration/mintShardAssignment.ts

@@ -133,11 +133,17 @@ type GlobalShardConfig = { resolution: MintShardSetResolution; override: unknown
 
 export type MintShardCache = { value: GlobalShardConfig; expiresAt: number } | undefined;
 
+type MintShardCacheHandle = {
+  current: MintShardCache;
+  // The refresh currently in flight, if any. Concurrent misses share it.
+  inFlight?: Promise<GlobalShardConfig>;
+};
+
 export type ResolveMintShardDeps = {
   // Reads the list rows. Injected so the cache and the fail-safe are testable without a
   // database, the same way computeRunIdMintKind takes its flag reader.
   readFlags: () => Promise<Record<string, unknown>>;
-  cache: { current: MintShardCache };
+  cache: MintShardCacheHandle;
   nowMs: number;
   ttlMs: number;
   graceMs: number;
@@ -155,6 +161,20 @@ export type ResolveMintShardDeps = {
 //
 // A failed read falls back to gen-1 rather than guessing a list. Guessing would move every
 // environment's placement for the length of one blip.
+async function refreshConfig(deps: ResolveMintShardDeps): Promise<GlobalShardConfig> {
+  try {
+    const flags = await deps.readFlags();
+    const config: GlobalShardConfig = {
+      resolution: readMintShardSetResolution(flags),
+      override: flags[FEATURE_FLAG.runOpsMintShardOverride],
+    };
+    deps.cache.current = { value: config, expiresAt: deps.nowMs + deps.ttlMs };
+    return config;
+  } finally {
+    deps.cache.inFlight = undefined;
+  }
+}
+
 export async function resolveMintShardWith(
   environment: { id: string; orgFeatureFlags?: unknown },
   deps: ResolveMintShardDeps
@@ -165,16 +185,13 @@ export async function resolveMintShardWith(
     config = cached.value;
   } else {
     try {
-      const flags = await deps.readFlags();
-      config = {
-        resolution: readMintShardSetResolution(flags),
-        override: flags[FEATURE_FLAG.runOpsMintShardOverride],
-      };
+      // Single-flight. Without it, two misses both read, and a slower read landing after a
+      // faster one puts its older snapshot back into the cache for a whole TTL.
+      config = await (deps.cache.inFlight ??= refreshConfig(deps));
     } catch (error) {
       deps.onReadFailed?.(error);
       return "new";
     }
-    deps.cache.current = { value: config, expiresAt: deps.nowMs + deps.ttlMs };
   }
 
   return computeMintShard(environment, {

apps/webapp/test/adminFeatureFlagsRouteAction.test.ts

@@ -2,7 +2,7 @@
 // bug surface. These drive the real exported action against a real Postgres and assert on the rows
 // it leaves behind. The only module substituted is the auth wrapper, so the handler can be called
 // without a super-admin session; the database is the genuine article, injected into db.server.
-import { boundedIn } from "@trigger.dev/database";
+import { boundedIn, $transaction as realTransaction } from "@trigger.dev/database";
 import type { PrismaClient } from "@trigger.dev/database";
 import { postgresTest } from "@internal/testcontainers";
 import { describe, expect, vi } from "vitest";
@@ -22,18 +22,23 @@ vi.mock("~/db.server", () => ({
     return db.client;
   },
   boundedIn,
-  // The real helper adds tracing around prisma.$transaction and resolves undefined when it
-  // swallows an infrastructure error. Neither is under test here, but the transactional
-  // semantics are, so this stands in with the same shape and a real interactive transaction.
-  $transaction: async (
+  // Delegates to the SAME shared implementation the production helper wraps, so the
+  // transactional semantics, the nesting case and the retry behaviour are the real ones rather
+  // than a reimplementation. Only the webapp wrapper's tracing span and its infrastructure-error
+  // logging are absent, and neither is asserted here.
+  $transaction: (
     client: PrismaClient,
     nameOrFn: unknown,
-    fnOrOptions?: unknown
-  ): Promise<unknown> => {
-    const fn = (typeof nameOrFn === "function" ? nameOrFn : fnOrOptions) as (
-      tx: PrismaClient
-    ) => Promise<unknown>;
-    return client.$transaction((tx) => fn(tx as unknown as PrismaClient));
+    fnOrOptions?: unknown,
+    options?: unknown
+  ) => {
+    const fn = (typeof nameOrFn === "function" ? nameOrFn : fnOrOptions) as Parameters<
+      typeof realTransaction
+    >[1];
+    const opts = (typeof nameOrFn === "function" ? fnOrOptions : options) as Parameters<
+      typeof realTransaction
+    >[3];
+    return realTransaction(client, fn, () => {}, opts);
   },
 }));