perf(run-store,webapp): prime run to org off the hot path, drop the dual-write DB read

Make the hot snapshot read path do no per-request run to org DB read during pure dual-write.

resolve(runId) is now a pure cache get: the fire-and-forget replica populate is gone. The
decorator primes the run to org cache for free on every mirrored write and every Redis read
hit, so a resident run's mapping is known with zero DB. resolveAuthoritative stays as the
fail-closed backstop for the redis-only guard, and the fallback order is unchanged.

Also corrects the stale writesRedisForTransition doc comment, which still claimed the predicate
was org-blind after it became per-org presence-as-latch.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: d-cs

apps/webapp/app/v3/snapshotRunOrg.server.ts

@@ -1,7 +1,6 @@
 import { LRUCache } from "lru-cache";
-import { $replica, prisma } from "~/db.server";
+import { prisma } from "~/db.server";
 import { env } from "~/env.server";
-import { logger } from "~/services/logger.server";
 import { singleton } from "~/utils/singleton";
 
 /**
@@ -29,31 +28,38 @@ type RunOrgClient = {
 
 export type SnapshotRunOrgSource = {
   /**
-   * Cache hit or undefined. On a miss, kicks off a replica populate off-path and returns at once.
-   * Never blocks, never throws.
+   * A PURE cache get: the cached org id, or undefined on a miss. Never queries, never blocks, never
+   * throws. The hot read path calls this, so it does no work — a resident run's mapping is put here
+   * for free by `prime`, from a mirrored write or a Redis read hit, and a non-resident run never
+   * needs one. There is deliberately no off-path populate: during pure dual-write the read path must
+   * issue zero run→org DB reads fleet-wide.
    */
   resolve(runId: string): string | undefined;
+  /**
+   * Records a run→org mapping the caller learned for free — from a mirrored write or a Redis read
+   * hit — so a later `resolve` is a pure hit with no DB read. In-memory, immutable (run→org never
+   * changes), no TTL, no invalidation. Fire-and-forget: never queries, never throws.
+   */
+  prime(runId: string, organizationId: string): void;
   /**
    * Awaits a bounded primary read, caches, and returns the org id. Throws on timeout, client
-   * failure, or a run with no organisation, so a caller can fail closed.
+   * failure, or a run with no organisation, so a caller can fail closed. This is the redis-only
+   * fallback gate's last leg, reached only when the sync cache is cold AND some org is redis-only.
    */
   resolveAuthoritative(runId: string): Promise<string>;
 };
 
 export function createSnapshotRunOrgSource(clients?: {
   primary: RunOrgClient;
-  replica: RunOrgClient;
 }): SnapshotRunOrgSource {
   const primaryClient = (clients?.primary ?? prisma) as RunOrgClient;
-  const replicaClient = (clients?.replica ?? $replica) as RunOrgClient;
   // No ttl: run→org is immutable, so a cached mapping never goes stale.
   const cache = new LRUCache<string, string>({
     max: env.RUN_ENGINE_SNAPSHOT_STORE_RUN_ORG_CACHE_MAX ?? DEFAULT_CACHE_MAX,
   });
-  const inFlight = new Set<string>();
 
-  async function read(runId: string, client: RunOrgClient): Promise<string> {
-    return client.taskRun
+  async function read(runId: string): Promise<string> {
+    return primaryClient.taskRun
       .findFirst({
         where: { id: runId },
         select: { runtimeEnvironment: { select: { organizationId: true } } },
@@ -70,21 +76,12 @@ export function createSnapshotRunOrgSource(clients?: {
 
   return {
     resolve(runId) {
-      const cached = cache.get(runId);
-      if (cached !== undefined) {
-        return cached;
-      }
-      if (!inFlight.has(runId)) {
-        inFlight.add(runId);
-        void read(runId, replicaClient)
-          .catch((error) => {
-            logger.warn("snapshotRunOrg: run→org populate failed", { runId, error });
-          })
-          .finally(() => {
-            inFlight.delete(runId);
-          });
-      }
-      return undefined;
+      return cache.get(runId);
+    },
+    prime(runId, organizationId) {
+      // run→org is immutable, so this can only ever confirm what is already there; setting keeps the
+      // active run hot in the LRU, which is exactly the run whose mapping the read path will want.
+      cache.set(runId, organizationId);
     },
     async resolveAuthoritative(runId) {
       const cached = cache.get(runId);
@@ -104,7 +101,7 @@ export function createSnapshotRunOrgSource(clients?: {
       });
 
       try {
-        return await Promise.race([read(runId, primaryClient), deadline]);
+        return await Promise.race([read(runId), deadline]);
       } finally {
         if (timer) clearTimeout(timer);
       }

apps/webapp/app/v3/snapshotStoreMode.server.ts

@@ -103,6 +103,8 @@ type ResolverOrgSource = {
 /** Resolves a run to its organisation. Cache-only and synchronous, undefined on a miss. */
 type ResolverRunOrgSource = {
   resolve(runId: string): string | undefined;
+  /** Records an immutable run→org mapping learned off a mirrored write or a Redis read hit. */
+  prime?(runId: string, organizationId: string): void;
   /** Bounded authoritative read, throws on failure/timeout, for the redis-only fallback gate. */
   resolveAuthoritative?(runId: string): Promise<string>;
 };
@@ -194,6 +196,11 @@ export function buildSnapshotStoreModeResolver(deps: {
     },
     anyOrgReadEnabled: (): boolean => deps.census?.anyOrgReadEnabled() ?? false,
     anyOrgRedisOnly: (): boolean => deps.census?.anyOrgRedisOnly() ?? false,
+    // The decorator hands back a run→org mapping it learned on a mirrored write or a Redis read hit.
+    // Recorded in-memory so a later readModeFor is a pure hit; absent hook is a no-op.
+    prime: (runId: string, organizationId: string): void => {
+      deps.runOrg?.prime?.(runId, organizationId);
+    },
   };
 }
 
@@ -237,6 +244,7 @@ export const snapshotStoreModeResolver: SnapshotStoreModeResolver = buildSnapsho
   },
   runOrg: {
     resolve: (runId) => snapshotRunOrgSource().resolve(runId),
+    prime: (runId, organizationId) => snapshotRunOrgSource().prime(runId, organizationId),
     resolveAuthoritative: (runId) => snapshotRunOrgSource().resolveAuthoritative(runId),
   },
   census: {

apps/webapp/test/snapshotRunOrg.server.test.ts

@@ -33,77 +33,77 @@ function fakeClient(opts: {
 const tick = () => new Promise((resolve) => setTimeout(resolve, 10));
 
 describe("snapshot run→org source", () => {
-  it("returns undefined on a cold miss, then serves the org id once the populate settles", async () => {
-    const replica = fakeClient({ mapping: { run_a: "org_a" } });
-    const source = createSnapshotRunOrgSource({ primary: replica, replica });
+  it("resolve is a pure cache get: a cold miss is undefined and never queries", async () => {
+    const primary = fakeClient({ mapping: { run_a: "org_a" } });
+    const source = createSnapshotRunOrgSource({ primary });
 
     expect(source.resolve("run_a")).toBeUndefined();
 
     await tick();
 
-    expect(source.resolve("run_a")).toBe("org_a");
+    // No off-path populate exists anymore, so a miss stays a miss and the DB is never touched.
+    expect(source.resolve("run_a")).toBeUndefined();
+    expect(primary.calls).toBe(0);
   });
 
-  it("does not start a second populate while one is in flight", async () => {
-    const replica = fakeClient({ mapping: { run_a: "org_a" }, delayMs: 20 });
-    const source = createSnapshotRunOrgSource({ primary: replica, replica });
+  it("prime makes a later resolve a pure hit, with no query", async () => {
+    const primary = fakeClient({ mapping: { run_a: "org_a" } });
+    const source = createSnapshotRunOrgSource({ primary });
+
+    source.prime("run_a", "org_a");
+
+    expect(source.resolve("run_a")).toBe("org_a");
+    await tick();
+    expect(primary.calls).toBe(0);
+  });
 
-    source.resolve("run_a");
-    source.resolve("run_a");
-    source.resolve("run_a");
+  it("prime is idempotent and never queries, however many times it is called", async () => {
+    const primary = fakeClient({ mapping: { run_a: "org_a" } });
+    const source = createSnapshotRunOrgSource({ primary });
 
-    await new Promise((resolve) => setTimeout(resolve, 40));
+    source.prime("run_a", "org_a");
+    source.prime("run_a", "org_a");
+    source.prime("run_a", "org_a");
 
-    expect(replica.calls).toBe(1);
     expect(source.resolve("run_a")).toBe("org_a");
+    expect(primary.calls).toBe(0);
   });
 
   it("resolveAuthoritative returns the org id on success and caches it", async () => {
     const primary = fakeClient({ mapping: { run_a: "org_a" } });
-    const source = createSnapshotRunOrgSource({ primary, replica: primary });
+    const source = createSnapshotRunOrgSource({ primary });
 
     await expect(source.resolveAuthoritative("run_a")).resolves.toBe("org_a");
     expect(source.resolve("run_a")).toBe("org_a");
   });
 
+  it("resolveAuthoritative serves a primed mapping without querying", async () => {
+    const primary = fakeClient({ mapping: { run_a: "org_a" } });
+    const source = createSnapshotRunOrgSource({ primary });
+
+    source.prime("run_a", "org_a");
+
+    await expect(source.resolveAuthoritative("run_a")).resolves.toBe("org_a");
+    expect(primary.calls).toBe(0);
+  });
+
   it("resolveAuthoritative throws when the run has no organization", async () => {
     const primary = fakeClient({ mapping: {} });
-    const source = createSnapshotRunOrgSource({ primary, replica: primary });
+    const source = createSnapshotRunOrgSource({ primary });
 
     await expect(source.resolveAuthoritative("run_missing")).rejects.toThrow();
   });
 
   it("resolveAuthoritative throws when the client rejects", async () => {
     const primary = fakeClient({ reject: true });
-    const source = createSnapshotRunOrgSource({ primary, replica: primary });
+    const source = createSnapshotRunOrgSource({ primary });
 
     await expect(source.resolveAuthoritative("run_a")).rejects.toThrow();
   });
 
-  it("resolve stays silent and releases in-flight when findFirst throws synchronously", async () => {
-    let calls = 0;
-    const throwing = {
-      taskRun: {
-        findFirst() {
-          calls++;
-          throw new Error("sync boom");
-        },
-      },
-    } as unknown as NonNullable<Parameters<typeof createSnapshotRunOrgSource>[0]>["replica"];
-    const source = createSnapshotRunOrgSource({ primary: throwing, replica: throwing });
-
-    expect(() => source.resolve("run_a")).not.toThrow();
-
-    await tick();
-
-    // In-flight was released, so a fresh miss starts a new populate rather than wedging forever.
-    expect(() => source.resolve("run_a")).not.toThrow();
-    expect(calls).toBe(2);
-  });
-
   it("resolveAuthoritative throws when the read exceeds the deadline", async () => {
     const primary = fakeClient({ mapping: { run_a: "org_a" }, delayMs: 2000 });
-    const source = createSnapshotRunOrgSource({ primary, replica: primary });
+    const source = createSnapshotRunOrgSource({ primary });
 
     await expect(source.resolveAuthoritative("run_a")).rejects.toThrow(/deadline|exceed/i);
   });

internal-packages/run-store/src/taskRunExecutionSnapshotStore.priming.test.ts

@@ -0,0 +1,155 @@
+// The read path must do NO run→org DB read during pure dual-write, and the mappings it needs must
+// arrive for free: the decorator primes the run→org cache on every mirrored write and every Redis
+// read hit. A counting fake stands in for the run→org source — `readModeFor` is a pure in-memory
+// lookup (the off-path populate is gone), and only `readModeForAuthoritative` would ever touch the
+// DB, so counting its calls counts run→org DB reads.
+import { describe, expect, it } from "vitest";
+import {
+  TaskRunExecutionSnapshotStore,
+  type SnapshotStoreMode,
+  type SnapshotStoreModeResolver,
+} from "./taskRunExecutionSnapshotStore.js";
+import type { RedisSnapshotStore, SnapshotRead } from "./redisSnapshotStore.js";
+import type { RunStore } from "./types.js";
+
+const SCOPE = {
+  environmentId: "env_1",
+  environmentType: "PRODUCTION",
+  projectId: "proj_1",
+  organizationId: "org_a",
+} as const;
+
+function harness(opts: {
+  globalMode: SnapshotStoreMode;
+  anyOrgReadEnabled?: boolean;
+  anyOrgRedisOnly?: boolean;
+  /** What Redis returns from getLatest, when a read routes to it. */
+  latest?: SnapshotRead | null;
+}) {
+  // The run→org cache the decorator primes into, plus a DB-read counter. `readModeFor` reads it in
+  // memory; only the authoritative leg counts as a DB read, and nothing here calls it unless asked.
+  const runOrg = new Map<string, string>();
+  let dbReads = 0;
+
+  const redis = new Proxy({} as RedisSnapshotStore, {
+    get: (_t, prop) => {
+      if (prop === "getLatest") return () => Promise.resolve(opts.latest ?? null);
+      return () => Promise.resolve({ outcome: "written", seq: 1 });
+    },
+  });
+
+  const delegateTouched: string[] = [];
+  const delegate = new Proxy({} as Record<string, unknown>, {
+    get: (_t, prop) => {
+      return (...__: unknown[]) => {
+        delegateTouched.push(String(prop));
+        return Promise.resolve({ id: "pg_row" });
+      };
+    },
+  }) as unknown as RunStore;
+
+  const modeResolver: SnapshotStoreModeResolver = {
+    resolve: () => opts.globalMode,
+    prime: (runId: string, organizationId: string) => {
+      runOrg.set(runId, organizationId);
+    },
+    readModeFor: (runId: string) => {
+      const org = runOrg.get(runId);
+      return org ? opts.globalMode : undefined;
+    },
+    readModeForAuthoritative: async (runId: string) => {
+      dbReads++;
+      const org = runOrg.get(runId);
+      return org ? opts.globalMode : undefined;
+    },
+    anyOrgReadEnabled: () => opts.anyOrgReadEnabled ?? false,
+    anyOrgRedisOnly: () => opts.anyOrgRedisOnly ?? false,
+  };
+
+  const decorated = new TaskRunExecutionSnapshotStore(delegate, {
+    store: redis,
+    mode: opts.globalMode,
+    modeResolver,
+    readPercent: 100,
+  });
+
+  return { decorated, runOrg, delegateTouched, dbReads: () => dbReads };
+}
+
+function completionForOrg(organizationId: string) {
+  return {
+    completedAt: new Date(),
+    outputType: "application/json",
+    usageDurationMs: 1,
+    costInCents: 0,
+    snapshot: {
+      id: "snap_1",
+      executionStatus: "FINISHED" as const,
+      description: "done",
+      runStatus: "COMPLETED_SUCCESSFULLY" as const,
+      attemptNumber: 1,
+      ...SCOPE,
+      organizationId,
+    },
+  };
+}
+
+function redisRead(organizationId: string): SnapshotRead {
+  return {
+    id: "snap_1",
+    seq: 1,
+    isValid: true,
+    raw: "{}",
+    entry: {
+      id: "snap_1",
+      runId: "run_1",
+      organizationId,
+      executionStatus: "EXECUTING",
+      createdAt: new Date().toISOString(),
+    },
+  };
+}
+
+describe("priming the run→org cache off the hot path", () => {
+  it("a mirrored transition primes the run→org mapping, with no run→org DB read", async () => {
+    const h = harness({ globalMode: "dual-write" });
+
+    await h.decorated.completeAttemptSuccess("run_1", completionForOrg("org_a"), {
+      select: { id: true },
+    });
+
+    expect(h.runOrg.get("run_1")).toBe("org_a");
+    expect(h.dbReads()).toBe(0);
+    // After the write, the run's mode resolves from the primed cache: a pure hit, still no DB read.
+    expect(h.decorated.modeForTest("org_a")).toBe("dual-write");
+    expect(h.dbReads()).toBe(0);
+  });
+
+  it("a Redis read hit primes the run→org mapping", async () => {
+    const h = harness({
+      globalMode: "redis-read",
+      anyOrgReadEnabled: true,
+      latest: redisRead("org_a"),
+    });
+
+    const latest = await h.decorated.findLatestExecutionSnapshot("run_1");
+
+    expect(latest).not.toBeNull();
+    expect(h.runOrg.get("run_1")).toBe("org_a");
+  });
+
+  it("a pure dual-write read issues ZERO run→org DB calls", async () => {
+    const h = harness({
+      globalMode: "dual-write",
+      anyOrgReadEnabled: false,
+      anyOrgRedisOnly: false,
+    });
+
+    const latest = await h.decorated.findLatestExecutionSnapshot("run_cold", undefined, "env_1");
+
+    // Fell back to Postgres, and the authoritative (DB) leg was never consulted.
+    expect(latest).toEqual({ id: "pg_row" });
+    expect(h.delegateTouched).toContain("findLatestExecutionSnapshot");
+    expect(h.dbReads()).toBe(0);
+  });
+});