feat(run-engine): jittered replica retry before primary fallback in getSnapshotsSince

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: d-cs

.server-changes/snapshots-since-replica-primary-fallback.md

@@ -5,4 +5,5 @@ type: improvement
 
 Run snapshot polling no longer errors or pays extra latency when the database read replica
 hasn't yet replicated the snapshot the runner is polling from
-(`RUN_ENGINE_READ_REPLICA_SNAPSHOTS_SINCE_ENABLED`): the read is served from the primary instead.
+(`RUN_ENGINE_READ_REPLICA_SNAPSHOTS_SINCE_ENABLED`): the read is briefly retried on the
+replica and served from the primary if it still hasn't caught up.

apps/webapp/app/env.server.ts

@@ -950,6 +950,8 @@ const EnvironmentSchema = z
       .default("info"),
     RUN_ENGINE_TREAT_PRODUCTION_EXECUTION_STALLS_AS_OOM: z.string().default("0"),
     RUN_ENGINE_READ_REPLICA_SNAPSHOTS_SINCE_ENABLED: z.string().default("0"),
+    RUN_ENGINE_SNAPSHOTS_SINCE_REPLICA_RETRY_MIN_MS: z.coerce.number().int().default(50),
+    RUN_ENGINE_SNAPSHOTS_SINCE_REPLICA_RETRY_MAX_MS: z.coerce.number().int().default(200),
     RUN_ENGINE_DEBOUNCE_USE_REPLICA_FOR_FAST_PATH_READ: z.string().default("0"),
 
     /** How long should the presence ttl last */

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

@@ -22,6 +22,10 @@ function createRunEngine() {
       env.RUN_ENGINE_TREAT_PRODUCTION_EXECUTION_STALLS_AS_OOM === "1",
     readReplicaSnapshotsSinceEnabled:
       env.RUN_ENGINE_READ_REPLICA_SNAPSHOTS_SINCE_ENABLED === "1",
+    readReplicaSnapshotsSinceRetryDelay: {
+      minMs: env.RUN_ENGINE_SNAPSHOTS_SINCE_REPLICA_RETRY_MIN_MS,
+      maxMs: env.RUN_ENGINE_SNAPSHOTS_SINCE_REPLICA_RETRY_MAX_MS,
+    },
     worker: {
       disabled: env.RUN_ENGINE_WORKER_ENABLED === "0",
       workers: env.RUN_ENGINE_WORKER_COUNT,

internal-packages/run-engine/src/engine/index.ts

@@ -33,6 +33,7 @@ import {
 import { Worker } from "@trigger.dev/redis-worker";
 import { assertNever } from "assert-never";
 import { EventEmitter } from "node:events";
+import { setTimeout } from "node:timers/promises";
 import { BatchQueue } from "../batch-queue/index.js";
 import type {
   BatchItem,
@@ -94,6 +95,7 @@ export class RunEngine {
   private tracer: Tracer;
   private meter: Meter;
   private snapshotsSinceReplicaMissCounter: Counter;
+  private snapshotsSinceReplicaRetryDelay: { minMs: number; maxMs: number };
   private heartbeatTimeouts: HeartbeatTimeouts;
   private repairSnapshotTimeoutMs: number;
   private batchQueue: BatchQueue;
@@ -282,10 +284,15 @@ export class RunEngine {
       "run_engine.snapshots_since.replica_miss",
       {
         description:
-          "getSnapshotsSince reads where the since snapshot was not yet on the read replica and was served from the primary",
+          "getSnapshotsSince reads where the since snapshot was not yet on the read replica, recovered via a replica retry or served from the primary",
       }
     );
 
+    this.snapshotsSinceReplicaRetryDelay = options.readReplicaSnapshotsSinceRetryDelay ?? {
+      minMs: 50,
+      maxMs: 200,
+    };
+
     const defaultHeartbeatTimeouts: HeartbeatTimeouts = {
       PENDING_EXECUTING: 60_000,
       PENDING_CANCEL: 60_000,
@@ -1948,11 +1955,36 @@ export class RunEngine {
     } catch (e) {
       if (useReplica && e instanceof ExecutionSnapshotNotFoundError) {
         // Replica lag: the runner learned this snapshot id from the writer before the
-        // replica caught up. Serve from the writer; only count/warn if the writer has it
-        // (a permanent miss is a real error, not lag).
+        // replica caught up. Give the replica one jittered retry, then serve from the
+        // writer; only count/warn if a retry succeeds (a permanent miss is a real error,
+        // not lag).
+        const { minMs, maxMs } = this.snapshotsSinceReplicaRetryDelay;
+        if (maxMs > 0) {
+          await setTimeout(minMs + Math.random() * Math.max(0, maxMs - minMs));
+          try {
+            const result = await query(this.readOnlyPrisma);
+            this.snapshotsSinceReplicaMissCounter.add(1, { outcome: "replica_retry" });
+            return result;
+          } catch (replicaRetryError) {
+            if (!(replicaRetryError instanceof ExecutionSnapshotNotFoundError)) {
+              this.logger.error("Failed to getSnapshotsSince", {
+                message:
+                  replicaRetryError instanceof Error
+                    ? replicaRetryError.message
+                    : replicaRetryError,
+                runId,
+                snapshotId,
+                retriedFromReplica: true,
+              });
+              return null;
+            }
+            // still not on the replica - fall through to the primary
+          }
+        }
+
         try {
           const result = await query(this.prisma);
-          this.snapshotsSinceReplicaMissCounter.add(1);
+          this.snapshotsSinceReplicaMissCounter.add(1, { outcome: "primary" });
           this.logger.warn("getSnapshotsSince: snapshot not yet on replica, served from primary", {
             runId,
             snapshotId,

internal-packages/run-engine/src/engine/types.ts

@@ -207,6 +207,10 @@ export type RunEngineOptions = {
    *  of the primary. Defaults to false. Callers passing an explicit `tx` always use
    *  that client regardless of this flag. */
   readReplicaSnapshotsSinceEnabled?: boolean;
+  /** Jittered delay bounds for the single replica retry `getSnapshotsSince` performs when
+   * the since snapshot is not yet on the replica, before falling back to the primary.
+   * Set maxMs to 0 to skip the replica retry and go straight to the primary. */
+  readReplicaSnapshotsSinceRetryDelay?: { minMs: number; maxMs: number };
   tracer: Tracer;
   meter?: Meter;
   logger?: Logger;