Merge remote-tracking branch 'origin/main' into fix/run-ops-ulid-run-id

# Conflicts:
#	internal-packages/run-store/src/runOpsStore.test.ts
#	internal-packages/run-store/src/runOpsStore.ts

Author: d-cs

.server-changes/fix-run-store-routing-read-your-writes.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: fix
+---
+
+Fix run store routing dropping the caller's read client, which downgraded read-your-writes reads (execution snapshots, waitpoints, batches) to the read replica and could fail dequeues under replica lag

apps/webapp/app/db.server.ts

@@ -8,6 +8,7 @@ import {
   type PrismaTransactionOptions,
 } from "@trigger.dev/database";
 import { RunOpsPrismaClient } from "@internal/run-ops-database";
+import { markReadReplicaClient } from "@internal/run-store";
 import invariant from "tiny-invariant";
 import { z } from "zod";
 import { env } from "./env.server";
@@ -170,7 +171,11 @@ export const prisma = singleton("prisma", () =>
 
 export const $replica: PrismaReplicaClient = singleton("replica", () => {
   const replica = getReplicaClient();
-  return replica ? captureInfrastructureErrors(tagDatasource("replica", replica)) : prisma;
+  // Brand ONLY a real replica so the run-store routing layer keeps replica reads off the primary.
+  // No replica configured → fall back to the writer `prisma`, which must stay UNBRANDED.
+  return replica
+    ? markReadReplicaClient(captureInfrastructureErrors(tagDatasource("replica", replica)))
+    : prisma;
 });
 
 export type RunOpsClients = { writer: PrismaClient; replica: PrismaReplicaClient };
@@ -254,9 +259,14 @@ const runOpsTopology: RunOpsTopology = singleton("runOpsTopology", () => {
         captureInfraErrorsRunOps(
           tagDatasourceRunOps("writer", buildRunOpsWriterClient({ url, clientType }))
         ),
+      // Brand the run-ops replica (only built for a real replica URL) so routed replica reads stay
+      // off the primary. When no replica URL is set, selectRunOpsTopology reuses the writer here —
+      // which this callback never touches, so the writer stays unbranded.
       buildNewReplica: (url, clientType) =>
-        captureInfraErrorsRunOps(
-          tagDatasourceRunOps("replica", buildRunOpsReplicaClient({ url, clientType }))
+        markReadReplicaClient(
+          captureInfraErrorsRunOps(
+            tagDatasourceRunOps("replica", buildRunOpsReplicaClient({ url, clientType }))
+          )
         ),
     }
   );

internal-packages/run-store/src/PostgresRunStore.ts

@@ -507,6 +507,13 @@ export class PostgresRunStore implements RunStore {
     this.schemaVariant = options.schemaVariant ?? "legacy";
   }
 
+  // The writer handle in read-client form, so the routing layer can honor a caller-passed client
+  // (read-your-writes) with THIS store's own primary instead of leaking the caller's client across
+  // DBs. Cast mirrors runInTransaction: the generated clients differ only in delegates reads use.
+  get primaryReadClient(): ReadClient {
+    return this.prisma as unknown as ReadClient;
+  }
+
   // Open ONE interactive transaction on this store's OWN writer client and run `fn` against THIS store
   // (so subclass overrides survive) with the tx as the client to thread into the inner writes. `runId`
   // is ignored here — a single store has one connection — but is in the contract so the router can

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

@@ -1,3 +1,4 @@
 export * from "./types.js";
 export * from "./PostgresRunStore.js";
 export * from "./runOpsStore.js";
+export * from "./readReplicaClient.js";

internal-packages/run-store/src/readReplicaClient.ts

@@ -0,0 +1,26 @@
+// A writer and a read-replica Prisma client are structurally identical at runtime (a replica is a
+// `new PrismaClient(...)` too, so it also exposes `$transaction`). The routing layer therefore
+// cannot tell a caller-passed replica from a writer by shape, yet it must — a writer/tx read has to
+// reach the owning store's PRIMARY (read-your-writes) while a replica read stays on a replica (read
+// scaling). So the client builder brands replica handles and the routing store reads the brand.
+export const READ_REPLICA_BRAND: unique symbol = Symbol.for("trigger.dev/run-store/read-replica");
+
+// Brand a replica client (returns the same object). MUST only be called on a genuine replica: the
+// routing layer trusts the brand to mean "do not escalate this read to the primary". An unbranded
+// replica just escalates as before — a scaling miss, never a correctness bug.
+export function markReadReplicaClient<T extends object>(client: T): T {
+  try {
+    (client as Record<symbol, unknown>)[READ_REPLICA_BRAND] = true;
+  } catch {
+    // Frozen/exotic clients may reject the assignment; only costs the optimization, not correctness.
+  }
+  return client;
+}
+
+export function isReadReplicaClient(client: unknown): boolean {
+  return (
+    !!client &&
+    typeof client === "object" &&
+    (client as Record<symbol, unknown>)[READ_REPLICA_BRAND] === true
+  );
+}