fix(run-store): guard every redis-only read fallback so an empty Postgres is never served

Author: d-cs

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

@@ -61,6 +61,8 @@ type ResolverOrgSource = Pick<OrgModeSource, "get" | "refresh"> &
 /** Resolves a run to its organisation. Cache-only and synchronous, undefined on a miss. */
 type ResolverRunOrgSource = {
   resolve(runId: string): string | undefined;
+  /** Bounded authoritative read, throws on failure/timeout, for the redis-only fallback gate. */
+  resolveAuthoritative?(runId: string): Promise<string>;
 };
 
 /** The census read accessors the resolver delegates to. Both synchronous and no-query. */
@@ -137,6 +139,18 @@ export function buildSnapshotStoreModeResolver(deps: {
       }
       return resolveMode(organizationId);
     },
+    // Authoritative counterpart, used only when the sync read is unresolved and some org is
+    // redis-only. Resolves run→org from the primary (bounded, throws on failure), warms the org dial
+    // so the immediate read is accurate, then answers with the org's mode. A throw propagates so the
+    // decorator fails closed.
+    readModeForAuthoritative: async (runId: string): Promise<DialMode | undefined> => {
+      if (!deps.runOrg?.resolveAuthoritative) {
+        return undefined;
+      }
+      const organizationId = await deps.runOrg.resolveAuthoritative(runId);
+      await deps.orgMode.warm?.(organizationId);
+      return resolveMode(organizationId);
+    },
     anyOrgReadEnabled: (): boolean => deps.census?.anyOrgReadEnabled() ?? false,
     anyOrgRedisOnly: (): boolean => deps.census?.anyOrgRedisOnly() ?? false,
   };
@@ -329,7 +343,10 @@ export const snapshotStoreModeResolver: SnapshotStoreModeResolver = buildSnapsho
     refresh: (organizationId) => orgModeSource().refresh(organizationId),
     warm: (organizationId) => orgModeSource().warm(organizationId),
   },
-  runOrg: { resolve: (runId) => snapshotRunOrgSource().resolve(runId) },
+  runOrg: {
+    resolve: (runId) => snapshotRunOrgSource().resolve(runId),
+    resolveAuthoritative: (runId) => snapshotRunOrgSource().resolveAuthoritative(runId),
+  },
   census: {
     anyOrgReadEnabled: () => snapshotStoreOrgCensus.anyOrgReadEnabled(),
     anyOrgRedisOnly: () => snapshotStoreOrgCensus.anyOrgRedisOnly(),

apps/webapp/test/snapshotStoreMode.test.ts

@@ -177,6 +177,67 @@ describe("the org-scoped read routing", () => {
   });
 });
 
+describe("the authoritative read position (redis-only fallback gate)", () => {
+  function buildAuth(opts: {
+    globalMode?: SnapshotStoreMode;
+    perOrg?: Record<string, SnapshotStoreMode>;
+    resolveAuthoritative?: (runId: string) => Promise<string>;
+    warm?: (organizationId: string) => Promise<void>;
+  }) {
+    return buildSnapshotStoreModeResolver({
+      globalMode: () => opts.globalMode,
+      orgMode: {
+        get: (id: string) => opts.perOrg?.[id],
+        refresh: () => {},
+        ...(opts.warm && { warm: opts.warm }),
+      },
+      runOrg: {
+        resolve: () => undefined,
+        ...(opts.resolveAuthoritative && { resolveAuthoritative: opts.resolveAuthoritative }),
+      },
+      envFloor: "off",
+    });
+  }
+
+  it("resolves the run's org authoritatively, warms the dial, and returns the org mode", async () => {
+    const warmed: string[] = [];
+    const r = buildAuth({
+      globalMode: "redis-read",
+      perOrg: { org_ro: "redis-only" },
+      resolveAuthoritative: async () => "org_ro",
+      warm: async (id) => {
+        warmed.push(id);
+      },
+    });
+    await expect(r.readModeForAuthoritative?.("run_1")).resolves.toBe("redis-only");
+    expect(warmed).toContain("org_ro");
+  });
+
+  it("returns a non-redis-only mode for a pre-cutover run so the decorator falls back", async () => {
+    const r = buildAuth({
+      globalMode: "redis-read",
+      perOrg: {},
+      resolveAuthoritative: async () => "org_pre",
+    });
+    await expect(r.readModeForAuthoritative?.("run_1")).resolves.toBe("redis-read");
+  });
+
+  it("propagates a throw from the authoritative run→org read so the decorator fails closed", async () => {
+    const r = buildAuth({
+      globalMode: "redis-read",
+      resolveAuthoritative: async () => {
+        throw new Error("run→org read timed out");
+      },
+    });
+    await expect(r.readModeForAuthoritative?.("run_1")).rejects.toThrow(/timed out/);
+  });
+
+  it("returns undefined when no authoritative run→org source is wired", async () => {
+    const r = buildAuth({ globalMode: "redis-read" });
+    await expect(r.readModeForAuthoritative?.("run_1")).resolves.toBeUndefined();
+  });
+});
+
 describe("a saved organisation dial survives a lagging replica", () => {
   type Deferred = { resolve: (v: unknown) => void; promise: Promise<unknown> };
 

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

@@ -239,4 +239,205 @@ describe("redis-only throws on a miss instead of serving empty Postgres", () =>
     ).rejects.toThrow(/redis-only/);
     expect(h.delegateTouched).not.toContain("findSnapshotCompletedWaitpointIds");
   });
+
+  it("findExecutionSnapshot throws on a by-id miss at redis-only", async () => {
+    const h = missHarness({ globalMode: "redis-only" });
+
+    await expect(
+      h.decorated.findExecutionSnapshot({
+        where: { id: "snap_1", runId: "run_x" },
+        select: { createdAt: true },
+      } as never)
+    ).rejects.toThrow(/redis-only/);
+    expect(h.delegateTouched).not.toContain("findExecutionSnapshot");
+  });
+});
+
+// The unified decision: when the run→org cache is COLD (readModeFor undefined) and some org is
+// redis-only, the sync gate cannot tell whether THIS run is redis-only. It must resolve the run's
+// org authoritatively rather than either strand a redis-only run (empty Postgres) or over-throw a
+// pre-cutover run. This harness supplies that authoritative hook and a configurable Redis read.
+function authHarness(opts: {
+  globalMode: SnapshotStoreMode;
+  read: "miss" | "dangling" | "error";
+  readModeFor?: (runId: string, environmentId?: string) => SnapshotStoreMode | undefined;
+  anyOrgRedisOnly?: () => boolean;
+  anyOrgReadEnabled?: () => boolean;
+  authoritative?: (runId: string) => Promise<SnapshotStoreMode | undefined>;
+}) {
+  const danglingRead = {
+    id: "snap_head",
+    seq: 1,
+    isValid: true,
+    entry: { id: "snap_head", createdAt: new Date().toISOString() },
+    raw: "{}",
+    danglingCycle: true,
+  };
+  const redisRead = () => {
+    if (opts.read === "error") throw new Error("redis brownout");
+    return Promise.resolve(opts.read === "dangling" ? danglingRead : null);
+  };
+  const redis = new Proxy({} as RedisSnapshotStore, {
+    get: (_t, prop) => {
+      if (prop === "getSinceCreatedAt")
+        return () => (opts.read === "error" ? redisRead() : Promise.resolve({ kind: "miss" }));
+      if (prop === "getSnapshotWaitpointIds")
+        return () =>
+          opts.read === "error"
+            ? redisRead()
+            : Promise.resolve({ present: false, distinctIds: [], order: [] });
+      return () => redisRead();
+    },
+  });
+
+  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_fallback" });
+      };
+    },
+  }) as unknown as RunStore;
+
+  const authoritativeCalls: string[] = [];
+  const modeResolver: SnapshotStoreModeResolver = {
+    resolve: () => opts.globalMode,
+    ...(opts.readModeFor && {
+      readModeFor: opts.readModeFor as SnapshotStoreModeResolver["readModeFor"],
+    }),
+    ...(opts.anyOrgRedisOnly && { anyOrgRedisOnly: opts.anyOrgRedisOnly }),
+    ...(opts.anyOrgReadEnabled && { anyOrgReadEnabled: opts.anyOrgReadEnabled }),
+    ...(opts.authoritative && {
+      readModeForAuthoritative: (async (runId: string) => {
+        authoritativeCalls.push(runId);
+        return opts.authoritative!(runId);
+      }) as SnapshotStoreModeResolver["readModeForAuthoritative"],
+    }),
+  };
+
+  const decorated = new TaskRunExecutionSnapshotStore(delegate, {
+    store: redis,
+    mode: opts.globalMode,
+    modeResolver,
+    readPercent: 100,
+  });
+
+  return { decorated, delegateTouched, authoritativeCalls };
+}
+
+describe("redis-only fallback resolves a cold run→org cache authoritatively", () => {
+  it("routing branch: a cold-cache redis-only run at global dual-write THROWS, never empty PG", async () => {
+    const h = authHarness({
+      globalMode: "dual-write",
+      read: "miss",
+      readModeFor: () => undefined,
+      anyOrgRedisOnly: () => true,
+      authoritative: async () => "redis-only",
+    });
+
+    await expect(
+      h.decorated.findLatestExecutionSnapshot("run_cold", undefined, "env_a")
+    ).rejects.toThrow(/redis-only/);
+    expect(h.delegateTouched).not.toContain("findLatestExecutionSnapshot");
+    expect(h.authoritativeCalls).toContain("run_cold");
+  });
+
+  it("routing branch: a cold-cache dual-write run falls back (no over-throw)", async () => {
+    const h = authHarness({
+      globalMode: "dual-write",
+      read: "miss",
+      readModeFor: () => undefined,
+      anyOrgRedisOnly: () => true,
+      authoritative: async () => "dual-write",
+    });
+
+    const result = await h.decorated.findLatestExecutionSnapshot("run_cold", undefined, "env_a");
+    expect(result).toEqual({ id: "pg_fallback" });
+    expect(h.delegateTouched).toContain("findLatestExecutionSnapshot");
+  });
+
+  it("off/dual-write with no org redis-only falls back WITHOUT an authoritative read", async () => {
+    const h = authHarness({
+      globalMode: "dual-write",
+      read: "miss",
+      readModeFor: () => undefined,
+      anyOrgRedisOnly: () => false,
+      authoritative: async () => "redis-only",
+    });
+
+    const result = await h.decorated.findLatestExecutionSnapshot("run_x", undefined, "env_a");
+    expect(result).toEqual({ id: "pg_fallback" });
+    expect(h.authoritativeCalls).toEqual([]);
+  });
+
+  it("MISS at global redis-read with a cold-cache pre-cutover run FALLS BACK (coexistence)", async () => {
+    const h = authHarness({
+      globalMode: "redis-read",
+      read: "miss",
+      anyOrgReadEnabled: () => true,
+      readModeFor: () => undefined,
+      anyOrgRedisOnly: () => true,
+      authoritative: async () => "redis-read",
+    });
+
+    const result = await h.decorated.findLatestExecutionSnapshot("run_pre", undefined, "env_a");
+    expect(result).toEqual({ id: "pg_fallback" });
+    expect(h.delegateTouched).toContain("findLatestExecutionSnapshot");
+  });
+
+  it("MISS at global redis-read with a cold-cache redis-only run THROWS", async () => {
+    const h = authHarness({
+      globalMode: "redis-read",
+      read: "miss",
+      anyOrgReadEnabled: () => true,
+      readModeFor: () => undefined,
+      anyOrgRedisOnly: () => true,
+      authoritative: async () => "redis-only",
+    });
+
+    await expect(
+      h.decorated.findLatestExecutionSnapshot("run_ro", undefined, "env_a")
+    ).rejects.toThrow(/redis-only/);
+    expect(h.delegateTouched).not.toContain("findLatestExecutionSnapshot");
+  });
+
+  it("fails closed (THROWS) when the authoritative read itself fails and some org is redis-only", async () => {
+    const h = authHarness({
+      globalMode: "redis-read",
+      read: "miss",
+      anyOrgReadEnabled: () => true,
+      readModeFor: () => undefined,
+      anyOrgRedisOnly: () => true,
+      authoritative: async () => {
+        throw new Error("run→org read timed out");
+      },
+    });
+
+    await expect(
+      h.decorated.findLatestExecutionSnapshot("run_cold", undefined, "env_a")
+    ).rejects.toThrow(/redis-only/);
+    expect(h.delegateTouched).not.toContain("findLatestExecutionSnapshot");
+  });
+
+  it("danglingCycle at global redis-only THROWS, never empty PG", async () => {
+    const h = authHarness({ globalMode: "redis-only", read: "dangling" });
+
+    await expect(
+      h.decorated.findLatestExecutionSnapshot("run_x", undefined, "env_a")
+    ).rejects.toThrow(/redis-only/);
+    expect(h.delegateTouched).not.toContain("findLatestExecutionSnapshot");
+  });
+
+  it("danglingCycle below redis-only still falls back to Postgres", async () => {
+    const h = authHarness({
+      globalMode: "redis-read",
+      read: "dangling",
+      anyOrgReadEnabled: () => true,
+    });
+
+    const result = await h.decorated.findLatestExecutionSnapshot("run_x", undefined, "env_a");
+    expect(result).toEqual({ id: "pg_fallback" });
+    expect(h.delegateTouched).toContain("findLatestExecutionSnapshot");
+  });
 });