refactor(run-store,run-engine,webapp): make redis-read reads org-gated, drop the percentage

Reads at redis-read were gated by a per-run hash percentage that production
always pinned to 100, so the sampling never ran. The organisation is the ramp
unit, so an org at redis-read now reads every one of its runs from Redis, with
Postgres as the fallback on a miss or error. Removes the readPercent option, the
hash sampling, and the tests that only exercised partial percentages.

Author: d-cs

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

@@ -114,9 +114,7 @@ const instance = singleton<Instance | undefined>("snapshotStoreInstance", () =>
         store: redisSnapshotStore,
         modeResolver: snapshotStoreModeResolver,
         halted: snapshotStoreHalted,
-        // Pinned: the field defaults to 0, which would mean no read ever reaches Redis. The
-        // organisation is the ramp unit, so there is no percentage to ramp.
-        readPercent: 100,
+        // The organisation is the ramp unit.
         metrics: metrics.decorator,
         onAppendFailure: async (args) => {
           const enqueue = getSnapshotRepairEnqueuer();

internal-packages/run-engine/src/engine/tests/helpers/decoratedStore.ts

@@ -32,7 +32,6 @@ export function buildDecoratedStore(opts: {
   prisma: PrismaClient;
   redisOptions: RedisOptions;
   mode: SnapshotStoreMode;
-  readPercent?: number;
   faults?: SnapshotFaultInjector;
   onAppendFailure?: SnapshotRepairEnqueuer;
 }): DecoratedStoreHarness {
@@ -50,7 +49,6 @@ export function buildDecoratedStore(opts: {
     {
       store: redis,
       mode: opts.mode,
-      readPercent: opts.readPercent ?? 100,
       ...(opts.faults && { faults: opts.faults }),
       onAppendFailure: async (args) => {
         repairs.push(args);

internal-packages/run-engine/src/engine/tests/snapshotStoreChaos.test.ts

@@ -112,7 +112,6 @@ describe("snapshot store crash boundaries", () => {
         prisma,
         redisOptions,
         mode: "redis-read",
-        readPercent: 100,
         faults: faults.hook,
       });
       const engine = new RunEngine(engineOptions(prisma, redisOptions, harness, 200) as never);
@@ -170,7 +169,6 @@ describe("snapshot store crash boundaries", () => {
         prisma,
         redisOptions,
         mode: "redis-read",
-        readPercent: 100,
         faults: faults.hook,
       });
       const engine = new RunEngine(engineOptions(prisma, redisOptions, harness, 200) as never);
@@ -286,7 +284,6 @@ describe("snapshot store crash boundaries", () => {
         prisma,
         redisOptions,
         mode: "redis-read",
-        readPercent: 100,
         faults: faults.hook,
       });
       const engine = new RunEngine(engineOptions(prisma, redisOptions, harness, 200) as never);

internal-packages/run-engine/src/engine/tests/snapshotStoreReadGate.test.ts

@@ -60,7 +60,6 @@ describe("snapshot store read gate", () => {
         prisma,
         redisOptions,
         mode: "redis-read",
-        readPercent: 100,
       });
       const engine = new RunEngine(engineOptions(prisma, redisOptions, harness) as never);
 
@@ -116,7 +115,6 @@ describe("snapshot store read gate", () => {
         prisma,
         redisOptions,
         mode: "redis-read",
-        readPercent: 100,
       });
       const engine = new RunEngine(engineOptions(prisma, redisOptions, harness) as never);
 
@@ -162,7 +160,6 @@ describe("snapshot store read gate", () => {
         prisma,
         redisOptions,
         mode: "redis-read",
-        readPercent: 100,
       });
       const engine = new RunEngine(engineOptions(prisma, redisOptions, harness) as never);
 
@@ -219,7 +216,6 @@ describe("snapshot store read gate", () => {
         prisma,
         redisOptions,
         mode: "redis-read",
-        readPercent: 100,
       });
       const engine = new RunEngine(engineOptions(prisma, redisOptions, harness) as never);
 
@@ -287,7 +283,6 @@ describe("snapshot store read gate", () => {
         prisma,
         redisOptions,
         mode: "redis-read",
-        readPercent: 100,
       });
       const engineOn = new RunEngine(engineOptions(prisma, redisOptions, on) as never);
       try {

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

@@ -44,7 +44,6 @@ describe("a checkpoint on a snapshot served from Redis", () => {
       const decorated = new TaskRunExecutionSnapshotStore(plain as unknown as RunStore, {
         store: redis,
         mode: "redis-read",
-        readPercent: 100,
       });
       try {
         const env = await seedSnapshotEnvironment(prisma);
@@ -121,7 +120,6 @@ describe("a checkpoint on a snapshot served from Redis", () => {
       const decorated = new TaskRunExecutionSnapshotStore(store as unknown as RunStore, {
         store: redis,
         mode: "redis-only",
-        readPercent: 100,
       });
       try {
         const env = await seedSnapshotEnvironment(prisma);
@@ -191,7 +189,6 @@ describe("a checkpoint on a snapshot served from Redis", () => {
       const decorated = new TaskRunExecutionSnapshotStore(plain as unknown as RunStore, {
         store: redis,
         mode: "redis-read",
-        readPercent: 100,
         metrics: {
           recordWrite: () => {},
           recordAppendFailed: () => {},

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

@@ -22,7 +22,6 @@ function storeWith(options: {
     mode: options.mode,
     ...(options.resolver && { modeResolver: options.resolver }),
     ...(options.halted !== undefined && { halted: () => options.halted === true }),
-    readPercent: 100,
   });
 }
 

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

@@ -210,7 +210,7 @@ describe("a Redis that stops answering", () => {
       const redis = new RedisSnapshotStore({ redisOptions, completedTtlMs: COMPLETED_TTL_MS });
       const decorated = new TaskRunExecutionSnapshotStore(
         new PostgresRunStore({ prisma, readOnlyPrisma: prisma }) as unknown as RunStore,
-        { store: redis, mode: "redis-read", readPercent: 100 }
+        { store: redis, mode: "redis-read" }
       );
       try {
         const env = await seedSnapshotEnvironment(prisma);
@@ -243,7 +243,7 @@ describe("a Redis that stops answering", () => {
     });
     const decorated = new TaskRunExecutionSnapshotStore(
       new PostgresRunStore({ prisma, readOnlyPrisma: prisma }) as unknown as RunStore,
-      { store: redis, mode: "redis-read", readPercent: 100 }
+      { store: redis, mode: "redis-read" }
     );
     try {
       const env = await seedSnapshotEnvironment(prisma);
@@ -285,7 +285,6 @@ describe("a Redis that stops answering", () => {
       const decorated = new TaskRunExecutionSnapshotStore(plain as unknown as RunStore, {
         store: redis,
         mode: "redis-read",
-        readPercent: 100,
       });
       try {
         const env = await seedSnapshotEnvironment(prisma);
@@ -365,7 +364,6 @@ describe("a Redis that stops answering", () => {
       const decorated = new TaskRunExecutionSnapshotStore(plain as unknown as RunStore, {
         store: redis,
         mode: "redis-read",
-        readPercent: 100,
         metrics: {
           recordWrite: () => {},
           recordAppendFailed: () => {},

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

@@ -16,7 +16,6 @@ function harness(opts: {
   globalMode: SnapshotStoreMode;
   readModeFor?: (runId: string, environmentId?: string) => SnapshotStoreMode;
   anyOrgReadEnabled?: () => boolean;
-  readPercent?: number;
 }) {
   const redisTouched: string[] = [];
   const redis = new Proxy({} as RedisSnapshotStore, {
@@ -55,7 +54,6 @@ function harness(opts: {
     store: redis,
     mode: opts.globalMode,
     modeResolver,
-    readPercent: opts.readPercent ?? 100,
   });
 
   return {
@@ -133,25 +131,23 @@ describe("org-scoped read routing", () => {
 describe("resolver is transparent to routing when no per-org override applies", () => {
   const ids = Array.from({ length: 500 }, (_, n) => `run_cohort_${n}_${n * 7919}`);
 
-  function plain(mode: SnapshotStoreMode, readPercent: number): CohortProbe {
+  function plain(mode: SnapshotStoreMode): CohortProbe {
     return new TaskRunExecutionSnapshotStore({} as unknown as RunStore, {
       store: {} as never,
       mode,
-      readPercent,
     }) as unknown as CohortProbe;
   }
 
   it("routes every run exactly as the global dial did before per-org reads existed", () => {
     // A resolver that always answers the global mode, with no org read-enabled. The short-circuit
-    // does not fire (global IS a read position), effective === global, and the hash path is
-    // untouched, so the population must route identically to a store with no resolver at all.
+    // does not fire (global IS a read position) and effective === global, so the population must
+    // route identically to a store with no resolver at all.
     const withResolver = harness({
       globalMode: "redis-read",
-      readPercent: 50,
       anyOrgReadEnabled: () => false,
       readModeFor: () => "redis-read",
     }).probe;
-    const withoutResolver = plain("redis-read", 50);
+    const withoutResolver = plain("redis-read");
 
     for (const id of ids) {
       expect(withResolver.readsFromRedis(id)).toBe(withoutResolver.readsFromRedis(id));

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

@@ -70,7 +70,6 @@ function harness(opts: {
     store: redis,
     mode: opts.globalMode,
     modeResolver,
-    readPercent: 100,
   });
 
   return { decorated, runOrg, delegateTouched, dbReads: () => dbReads };

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

@@ -45,7 +45,6 @@ function harness(global: SnapshotStoreMode, perOrg: SnapshotStoreMode) {
     store: redis,
     mode: global,
     modeResolver,
-    readPercent: 100,
   });
 
   return { decorated, appends };