feat(run-engine): flag to route getSnapshotsSince through read replica (#3423)

## Summary

Adds `RUN_ENGINE_READ_REPLICA_SNAPSHOTS_SINCE_ENABLED` (default `"0"`).
When enabled, the Prisma reads inside `RunEngine.getSnapshotsSince` run
against the read-only replica client instead of the primary. Offloads
the snapshot-polling queries fired by every running task runner off the
writer.

## Why

`getSnapshotsSince` is called from the managed runner's
fetch-and-process loop (once per poll interval, plus on every
snapshot-change notification). It runs four sequential reads per call —
one `findFirst` by snapshot id, one `findMany` on snapshots with
`createdAt > X`, one raw SQL against `_completedWaitpoints`, and chunked
`findMany` on `waitpoint`. Per concurrent run, every few seconds. It's
read-only, tolerates a small amount of staleness, and is an obvious
candidate for the replica.

## Replica-lag considerations

- **Step 1 "since snapshot not found"**: if the runner just received a
snapshot id from the primary and asks the replica before it replicates,
the function throws and the caller treats the response as an error
(runner falls back to a metadata refresh). Self-correcting, not silent.
- **Step 2 missing newly-created snapshots**: the next poll's `createdAt
> sinceSnapshot.createdAt` filter still picks them up once the replica
catches up.
- **Waitpoint junction race**: the riskiest path — if a latest snapshot
is replicated but its `_completedWaitpoints` join rows aren't yet, the
runner could advance past that snapshot with `completedWaitpoints: []`.
WAL/storage-level replication replays commits in order, so in practice
both should appear atomically on the reader, but the race window is why
the flag ships disabled.

Aurora reader shrinks all three windows to single-digit ms in typical
conditions, and its storage-level replication gives atomic visibility of
committed transactions on the reader.

## Test plan

- [ ] Flip the flag on in a non-prod environment, confirm snapshot
polling behaves normally and `getSnapshotsSince` errors in Sentry stay
flat.
- [ ] Verify writer query volume drops and reader query volume rises on
the snapshot-polling queries.
- [ ] Keep an eye on `AuroraReplicaLag` (or equivalent) during rollout.

Author: ericallam

.server-changes/read-replica-snapshots-since.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: improvement
+---
+
+Add `RUN_ENGINE_READ_REPLICA_SNAPSHOTS_SINCE_ENABLED` flag (default off) to route the Prisma reads inside `RunEngine.getSnapshotsSince` through the read-only replica client. Offloads the snapshot polling queries (fired by every running task runner) from the primary. When disabled, behavior is unchanged.

apps/webapp/app/env.server.ts

@@ -835,6 +835,7 @@ const EnvironmentSchema = z
       .enum(["log", "error", "warn", "info", "debug"])
       .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"),
 
     /** How long should the presence ttl last */
     DEV_PRESENCE_SSE_TIMEOUT: z.coerce.number().int().default(30_000),

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

@@ -19,6 +19,8 @@ function createRunEngine() {
     logLevel: env.RUN_ENGINE_WORKER_LOG_LEVEL,
     treatProductionExecutionStallsAsOOM:
       env.RUN_ENGINE_TREAT_PRODUCTION_EXECUTION_STALLS_AS_OOM === "1",
+    readReplicaSnapshotsSinceEnabled:
+      env.RUN_ENGINE_READ_REPLICA_SNAPSHOTS_SINCE_ENABLED === "1",
     worker: {
       disabled: env.RUN_ENGINE_WORKER_ENABLED === "0",
       workers: env.RUN_ENGINE_WORKER_COUNT,

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

@@ -1633,7 +1633,8 @@ export class RunEngine {
     snapshotId: string;
     tx?: PrismaClientOrTransaction;
   }): Promise<RunExecutionData[] | null> {
-    const prisma = tx ?? this.prisma;
+    const prisma =
+      tx ?? (this.options.readReplicaSnapshotsSinceEnabled ? this.readOnlyPrisma : this.prisma);
 
     try {
       const snapshots = await getExecutionSnapshotsSince(prisma, runId, snapshotId);

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

@@ -145,6 +145,10 @@ export type RunEngineOptions = {
   /** Optional maximum TTL for all runs (e.g. "14d"). If set, runs without an explicit TTL
    *  will use this as their TTL, and runs with a TTL larger than this will be clamped. */
   defaultMaxTtl?: string;
+  /** When true, `getSnapshotsSince` reads through the read-only replica client instead
+   *  of the primary. Defaults to false. Callers passing an explicit `tx` always use
+   *  that client regardless of this flag. */
+  readReplicaSnapshotsSinceEnabled?: boolean;
   tracer: Tracer;
   meter?: Meter;
   logger?: Logger;