feat(webapp,run-store): route run-graph reads and writes through the run-store router

.server-changes/runops-run-graph-client-routing.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: improvement
+---
+
+Improved the reliability of how run data is read and written.

apps/webapp/app/db.server.ts

@@ -25,6 +25,7 @@ import {
   assertSplitRealtimeInterlock,
 } from "./v3/runOpsMigration/splitMode.server";
 import { computeRunOpsSplitReadEnabled } from "./v3/runOpsMigration/runOpsSplitReadGate";
+import { assertControlPlaneCoresidencyAdvisory } from "./v3/runOpsMigration/controlPlaneCoresidencySentinel.server";
 import { DATASOURCE_CONTEXT_KEY, startActiveSpan } from "./v3/tracer.server";
 import type { Span } from "@opentelemetry/api";
 import { context, trace } from "@opentelemetry/api";
@@ -188,13 +189,18 @@ export type RunOpsTopology = {
 export type SelectRunOpsTopologyConfig = {
   splitEnabled: boolean;
   legacyUrl?: string;
+  legacyReplicaUrl?: string;
   newUrl?: string;
   newReplicaUrl?: string;
 };
 export type RunOpsClientBuilders = {
   controlPlane: RunOpsClients;
   buildNewWriter: (url: string, clientType: string) => RunOpsPrismaClient;
   buildNewReplica: (url: string, clientType: string) => RunOpsPrismaClient;
+  // Legacy builders return the same PrismaClient/PrismaReplicaClient types as the control plane (no
+  // RunOpsPrismaClient double-cast needed): the legacy DB carries the full control-plane schema.
+  buildLegacyWriter: (url: string, clientType: string) => PrismaClient;
+  buildLegacyReplica: (url: string, clientType: string) => PrismaReplicaClient;
 };
 
 // Pure run-ops client selector. No env, no isSplitEnabled() — those
@@ -220,7 +226,15 @@ export function selectRunOpsTopology(
     return { newRunOps: cpFallback, legacyRunOps: controlPlane, controlPlane };
   }
 
-  const legacyRunOps = controlPlane;
+  // Track 2: build an INDEPENDENT legacy client from its own DSN instead of aliasing the control
+  // plane. legacyUrl is guaranteed present (the missing-URL branch above aliases and returns).
+  const legacyWriter = builders.buildLegacyWriter(config.legacyUrl, "run-ops-legacy-writer");
+  // Mirror the NEW replica + control-plane $replica fallback: brand a real replica (in the builder),
+  // otherwise reuse the legacy WRITER so replica reads fall back to the legacy primary — unbranded.
+  const legacyReplica: PrismaReplicaClient = config.legacyReplicaUrl
+    ? builders.buildLegacyReplica(config.legacyReplicaUrl, "run-ops-legacy-reader")
+    : legacyWriter;
+  const legacyRunOps: RunOpsClients = { writer: legacyWriter, replica: legacyReplica };
 
   const newWriter = builders.buildNewWriter(config.newUrl, "run-ops-new-writer");
   const newReplica: RunOpsPrismaClient = config.newReplicaUrl
@@ -246,10 +260,19 @@ const runOpsTopology: RunOpsTopology = singleton("runOpsTopology", () => {
   // Gate on the opt-in flag too: the distinct-DB sentinel only runs when the flag is on.
   const splitEnabled = env.RUN_OPS_SPLIT_ENABLED && !!newUrl && !!env.RUN_OPS_LEGACY_DATABASE_URL;
 
+  // Without a dedicated legacy replica URL, legacy reads fall back to the legacy WRITER (primary).
+  // Surface that so a prod misdeploy is observable instead of a silent load shift onto the primary.
+  if (splitEnabled && !env.RUN_OPS_LEGACY_DATABASE_READ_REPLICA_URL) {
+    logger.warn(
+      "RUN_OPS_LEGACY_DATABASE_READ_REPLICA_URL is unset while split is enabled; legacy reads will hit the legacy primary"
+    );
+  }
+
   return selectRunOpsTopology(
     {
       splitEnabled,
       legacyUrl: env.RUN_OPS_LEGACY_DATABASE_URL,
+      legacyReplicaUrl: env.RUN_OPS_LEGACY_DATABASE_READ_REPLICA_URL,
       newUrl,
       newReplicaUrl: env.RUN_OPS_DATABASE_READ_REPLICA_URL,
     },
@@ -268,6 +291,18 @@ const runOpsTopology: RunOpsTopology = singleton("runOpsTopology", () => {
             tagDatasourceRunOps("replica", buildRunOpsReplicaClient({ url, clientType }))
           )
         ),
+      // Legacy client shares the exact control-plane wrapper stack (the legacy DB carries the full
+      // control-plane schema); markReadReplicaClient only on a real replica URL, as with the NEW replica.
+      buildLegacyWriter: (url, clientType) =>
+        captureInfrastructureErrors(
+          tagDatasource("writer", buildWriterClient({ url, clientType }))
+        ),
+      buildLegacyReplica: (url, clientType) =>
+        markReadReplicaClient(
+          captureInfrastructureErrors(
+            tagDatasource("replica", buildReplicaClient({ url, clientType }))
+          )
+        ),
     }
   );
 });
@@ -281,8 +316,17 @@ export const runOpsNewPrisma: PrismaClient = runOpsTopology.newRunOps
   .writer as unknown as PrismaClient;
 export const runOpsNewReplica: PrismaReplicaClient = runOpsTopology.newRunOps
   .replica as unknown as PrismaReplicaClient;
+// Track 2: under split-on these point at the INDEPENDENT legacy client (its own DSN); under split-off
+// or missing URLs they still alias the control-plane client, so single-DB installs are unchanged.
 export const runOpsLegacyPrisma: PrismaClient = runOpsTopology.legacyRunOps.writer;
 export const runOpsLegacyReplica: PrismaReplicaClient = runOpsTopology.legacyRunOps.replica;
+// Branded legacy handles typed as RunOpsPrismaClient for the run-store boundary — same underlying
+// legacy writer/replica as runOpsLegacyPrisma/runOpsLegacyReplica above, but carrying the run-ops
+// brand so the guard classifies provably-legacy access as `runops`, not `cp`.
+export const runOpsLegacyPrismaClient: RunOpsPrismaClient = runOpsTopology.legacyRunOps
+  .writer as unknown as RunOpsPrismaClient;
+export const runOpsLegacyReplicaClient: RunOpsPrismaClient = runOpsTopology.legacyRunOps
+  .replica as unknown as RunOpsPrismaClient;
 
 export const runOpsSplitReadEnabled: boolean = computeRunOpsSplitReadEnabled({
   newReplica: runOpsNewReplicaClient,
@@ -295,8 +339,8 @@ export const runOpsSplitReadEnabled: boolean = computeRunOpsSplitReadEnabled({
 
 // Boot-time interlock: if the flag is on but the distinct-DB sentinel does not
 // confirm two physically-distinct run-ops DBs, refuse to enable split (data-loss
-// interlock). Async, so it cannot live in the synchronous singleton factory —
-// call it from the eager-boot path before any run-ops routing is wired.
+// interlock). Async, so it cannot live in the synchronous singleton factory — called
+// fire-and-forget from the eager-boot path (routing is wired synchronously at module load).
 export async function assertRunOpsSplitSentinel(): Promise<void> {
   if (!env.RUN_OPS_SPLIT_ENABLED) return;
   // Realtime interlock (synchronous): Electric replicates only from the control-plane
@@ -312,6 +356,9 @@ export async function assertRunOpsSplitSentinel(): Promise<void> {
       "RUN_OPS_SPLIT_ENABLED is on but the distinct-DB sentinel did not confirm two physically-distinct run-ops DBs; refusing to enable split (data-loss interlock)."
     );
   }
+  // Advisory-only (T2.3): observe legacy vs control-plane co-residency. Emits a metric + log and only
+  // throws when RUN_OPS_EXPECT_CONTROL_PLANE_SPLIT is on AND co-residency is positively confirmed.
+  await assertControlPlaneCoresidencyAdvisory();
 }
 
 function getClient() {

apps/webapp/app/env.server.ts

@@ -138,8 +138,10 @@ const EnvironmentSchema = z
       .string()
       .refine(isValidDatabaseUrl, "RUN_OPS_DATABASE_URL is invalid")
       .optional(),
-    // The LEGACY run-ops DB (the control-plane DB during the transition). When unset, legacy
-    // run-ops reuses the existing DATABASE_URL (legacy run-ops == control-plane DB initially).
+    // The LEGACY run-ops DB. Now a CONNECTED DSN (Track 2): when split is on and this is set it builds
+    // an INDEPENDENT legacy Prisma client, no longer an alias of the control-plane client (nor merely
+    // the sentinel's probe target). Unset -> legacy reuses the control-plane client / DATABASE_URL, so
+    // single-DB and self-host installs boot byte-identical.
     RUN_OPS_LEGACY_DATABASE_URL: z
       .string()
       .refine(isValidDatabaseUrl, "RUN_OPS_LEGACY_DATABASE_URL is invalid")
@@ -151,6 +153,22 @@ const EnvironmentSchema = z
       .string()
       .refine(isValidDatabaseUrl, "RUN_OPS_DATABASE_READ_REPLICA_URL is invalid")
       .optional(),
+    // The LEGACY run-ops DB read replica (Track 2). Unset -> the legacy replica handle falls back to the
+    // legacy WRITER (as $replica does with no CP replica). Set in production so legacy reads hit the reader.
+    RUN_OPS_LEGACY_DATABASE_READ_REPLICA_URL: z
+      .string()
+      .refine(isValidDatabaseUrl, "RUN_OPS_LEGACY_DATABASE_READ_REPLICA_URL is invalid")
+      .optional(),
+    // Direct DSN for applying the full @trigger.dev/database migrations to the LEGACY run-ops DB, keeping
+    // its schema current after the control plane moves off it. Direct, not pooled — migrations never run
+    // over a pooler. Optional; unset -> the entrypoint's legacy migrate step is skipped.
+    RUN_OPS_LEGACY_DIRECT_URL: z
+      .string()
+      .refine(isValidDatabaseUrl, "RUN_OPS_LEGACY_DIRECT_URL is invalid")
+      .optional(),
+    // Advisory control-plane co-residency sentinel enforcement (Track 2, T2.3). Default OFF; the advisory
+    // arm always emits its metric, this only turns a still-co-resident pair into a hard boot failure.
+    RUN_OPS_EXPECT_CONTROL_PLANE_SPLIT: BoolEnv.default(false),
     // --- Control-plane datasource repoint. Additive-only. ---
     // Optional control-plane DB. Unset (self-host/single-DB) -> getClient()/getReplicaClient() fall back to
     // DATABASE_URL/DATABASE_READ_REPLICA_URL, so boot is byte-identical. When set, these point at the
@@ -1618,6 +1636,14 @@ const EnvironmentSchema = z
     RUN_REPLICATION_DISABLE_PAYLOAD_INSERT: z.string().default("0"),
     RUN_REPLICATION_DISABLE_ERROR_FINGERPRINTING: z.string().default("0"),
 
+    // Connection URL for the LEGACY runs-replication source (the runs-CDC slot on the legacy runs DB, plus
+    // the admin recovery route). Direct, not pooled: replication can't run over a pooler. Optional; unset ->
+    // falls back to DATABASE_URL, so nothing changes today.
+    RUN_REPLICATION_LEGACY_DATABASE_URL: z
+      .string()
+      .refine(isValidDatabaseUrl, "RUN_REPLICATION_LEGACY_DATABASE_URL is invalid")
+      .optional(),
+
     // --- Run-ops DB split — second replication source (the NEW dedicated run-ops DB). ---
     // Cloud-only; only consulted when isSplitEnabled() is true. Self-host never sets these.
     // Connection URL for the run-ops DB used by the runs-replication source. Required when the split is
@@ -1658,6 +1684,12 @@ const EnvironmentSchema = z
     SESSION_REPLICATION_PUBLICATION_NAME: z
       .string()
       .default("sessions_to_clickhouse_v1_publication"),
+    // Connection URL for the sessions-replication slot. Direct, not pooled: replication can't run over a
+    // pooler. Optional; unset -> falls back to DATABASE_URL, so nothing changes today.
+    SESSION_REPLICATION_DATABASE_URL: z
+      .string()
+      .refine(isValidDatabaseUrl, "SESSION_REPLICATION_DATABASE_URL is invalid")
+      .optional(),
     SESSION_REPLICATION_MAX_FLUSH_CONCURRENCY: z.coerce.number().int().default(1),
     SESSION_REPLICATION_FLUSH_INTERVAL_MS: z.coerce.number().int().default(1000),
     SESSION_REPLICATION_FLUSH_BATCH_SIZE: z.coerce.number().int().default(100),

apps/webapp/app/models/waitpointTag.server.ts

@@ -1,5 +1,5 @@
 import { Prisma } from "@trigger.dev/database";
-import { prisma } from "~/db.server";
+import { runStore } from "~/v3/runStore.server";
 
 export const MAX_TAGS_PER_WAITPOINT = 10;
 const MAX_RETRIES = 3;
@@ -19,19 +19,10 @@ export async function createWaitpointTag({
 
   while (attempts < MAX_RETRIES) {
     try {
-      return await prisma.waitpointTag.upsert({
-        where: {
-          environmentId_name: {
-            environmentId,
-            name: tag,
-          },
-        },
-        create: {
-          name: tag,
-          environmentId,
-          projectId,
-        },
-        update: {},
+      return await runStore.upsertWaitpointTag({
+        environmentId,
+        name: tag,
+        projectId,
       });
     } catch (error) {
       if (error instanceof Prisma.PrismaClientKnownRequestError && error.code === "P2002") {

apps/webapp/app/presenters/v3/ApiRunResultPresenter.server.ts

@@ -2,13 +2,14 @@ import type { TaskRunExecutionResult } from "@trigger.dev/core/v3";
 import type { PrismaClientOrTransaction, PrismaReplicaClient } from "~/db.server";
 import { executionResultForTaskRun } from "~/models/taskRun.server";
 import type { AuthenticatedEnvironment } from "~/services/apiAuth.server";
-import { readThroughRun } from "~/v3/runOpsMigration/readThrough.server";
+import { runStore as defaultRunStore } from "~/v3/runStore.server";
 import { BasePresenter } from "./basePresenter.server";
 
 type ApiRunResultReadThroughDeps = {
   splitEnabled?: boolean;
   newClient?: PrismaReplicaClient;
-  // LEGACY RUN-OPS READ REPLICA ONLY (never a writer/primary); defaults to this._replica.
+  // LEGACY RUN-OPS READ REPLICA ONLY (never a writer/primary); defaults to runOpsLegacyReplica
+  // (the Aurora legacy read replica), never the control-plane replica.
   legacyReplica?: PrismaReplicaClient;
   isPastRetention?: (runId: string) => boolean;
 };
@@ -17,7 +18,8 @@ export class ApiRunResultPresenter extends BasePresenter {
   constructor(
     prisma?: PrismaClientOrTransaction,
     replica?: PrismaClientOrTransaction,
-    private readonly _readThrough?: ApiRunResultReadThroughDeps
+    private readonly _readThrough?: ApiRunResultReadThroughDeps,
+    private readonly runStore = defaultRunStore
   ) {
     super(prisma, replica);
   }
@@ -27,31 +29,14 @@ export class ApiRunResultPresenter extends BasePresenter {
     env: AuthenticatedEnvironment
   ): Promise<TaskRunExecutionResult | undefined> {
     return this.traceWithEnv("call", env, async (span) => {
-      const findRun = (client: PrismaReplicaClient) =>
-        client.taskRun.findFirst({
-          where: { friendlyId, runtimeEnvironmentId: env.id },
-          include: { attempts: { orderBy: { createdAt: "desc" } } },
-        });
-
-      // Single-run result poll routed through run-ops read-through. Split on: primary store first,
-      // then the secondary read replica for runs that miss on new; past-retention ids return
-      // undefined -> the route's normal 404. Split off (single-DB / self-host): readThroughRun does
-      // one plain findFirst against the single client (passthrough).
-      const result = await readThroughRun({
-        runId: friendlyId,
-        environmentId: env.id,
-        readNew: findRun,
-        readLegacy: findRun,
-        deps: {
-          splitEnabled: this._readThrough?.splitEnabled,
-          newClient: this._readThrough?.newClient ?? (this._prisma as PrismaReplicaClient),
-          legacyReplica: this._readThrough?.legacyReplica ?? (this._replica as PrismaReplicaClient),
-          isPastRetention: this._readThrough?.isPastRetention,
-        },
-      });
-
-      const taskRun =
-        result.source === "new" || result.source === "legacy-replica" ? result.value : undefined;
+      // Single-run result poll routed through the run store, which selects the owning DB by
+      // run-id residency (id shape): a run-ops (NEW) id reads the new store, a cuid (LEGACY) id
+      // reads the legacy store. Single-DB / self-host collapses to one plain findFirst against the
+      // one store (passthrough). The identical TaskRun(+attempts) lookup runs inside the router.
+      const taskRun = await this.runStore.findRun(
+        { friendlyId, runtimeEnvironmentId: env.id },
+        { include: { attempts: { orderBy: { createdAt: "desc" } } } }
+      );
 
       if (!taskRun) {
         return undefined;

apps/webapp/app/presenters/v3/ApiWaitpointPresenter.server.ts

@@ -5,11 +5,11 @@ import { BasePresenter } from "./basePresenter.server";
 import { waitpointStatusToApiStatus } from "./WaitpointListPresenter.server";
 import { generateHttpCallbackUrl } from "~/services/httpCallback.server";
 import type { PrismaClientOrTransaction, PrismaReplicaClient } from "~/db.server";
-import { readThroughRun } from "~/v3/runOpsMigration/readThrough.server";
+import { runStore as defaultRunStore } from "~/v3/runStore.server";
 
-// When omitted, clients default to the inherited _replica handle => passthrough reads the
-// replica exactly as today. isPastRetention is injectable for tests. Typed PrismaReplicaClient
-// to match readThroughRun's readNew/readLegacy + deps.
+// Retained only to preserve the public constructor signature the route passes. Run-ops routing
+// (NEW vs LEGACY residency, replica reads) is now handled inside the injected `runStore`, so
+// these deps are no longer consulted for the read.
 type ApiWaitpointPresenterReadThroughDeps = {
   newClient?: PrismaReplicaClient;
   legacyReplica?: PrismaReplicaClient;
@@ -21,7 +21,8 @@ export class ApiWaitpointPresenter extends BasePresenter {
   constructor(
     prismaClient?: PrismaClientOrTransaction,
     replicaClient?: PrismaClientOrTransaction,
-    private readonly readThroughDeps?: ApiWaitpointPresenterReadThroughDeps
+    private readonly readThroughDeps?: ApiWaitpointPresenterReadThroughDeps,
+    private readonly runStore = defaultRunStore
   ) {
     super(prismaClient, replicaClient);
   }
@@ -39,56 +40,32 @@ export class ApiWaitpointPresenter extends BasePresenter {
     waitpointId: string
   ) {
     return this.trace("call", async (span) => {
-      // Public waitpoint retrieve. Split on: new run-ops client first, then the LEGACY
-      // RUN-OPS READ REPLICA ONLY on a new-probe miss — never the legacy primary.
-      // Split off (single-DB / self-host): one plain waitpoint.findFirst against the replica
-      // (passthrough). The waitpointId is the residency-classifiable run-ops id (the route
-      // pre-decodes the friendlyId via WaitpointId.toId).
-      const hydrate = (client: PrismaReplicaClient) =>
-        client.waitpoint.findFirst({
-          where: {
-            id: waitpointId,
-            environmentId: environment.id,
-          },
-          select: {
-            id: true,
-            friendlyId: true,
-            type: true,
-            status: true,
-            idempotencyKey: true,
-            userProvidedIdempotencyKey: true,
-            idempotencyKeyExpiresAt: true,
-            inactiveIdempotencyKey: true,
-            output: true,
-            outputType: true,
-            outputIsError: true,
-            completedAfter: true,
-            completedAt: true,
-            createdAt: true,
-            tags: true,
-          },
-        });
-
-      const result = await readThroughRun({
-        runId: waitpointId,
-        environmentId: environment.id,
-        readNew: (client) => hydrate(client),
-        readLegacy: (replica) => hydrate(replica),
-        deps: {
-          splitEnabled: this.readThroughDeps?.splitEnabled,
-          // Default both clients to the inherited _replica handle (declared
-          // PrismaClientOrTransaction but $replica at runtime) so passthrough reads the replica
-          // as today; split mode injects a distinct newClient.
-          newClient: this.readThroughDeps?.newClient ?? (this._replica as PrismaReplicaClient),
-          legacyReplica:
-            this.readThroughDeps?.legacyReplica ?? (this._replica as PrismaReplicaClient),
-          isPastRetention: this.readThroughDeps?.isPastRetention,
+      // The store routes by the waitpointId's residency (id shape) and reads the owning
+      // store's replica. waitpointId is pre-decoded from the friendlyId via WaitpointId.toId.
+      const waitpoint = await this.runStore.findWaitpoint({
+        where: {
+          id: waitpointId,
+          environmentId: environment.id,
+        },
+        select: {
+          id: true,
+          friendlyId: true,
+          type: true,
+          status: true,
+          idempotencyKey: true,
+          userProvidedIdempotencyKey: true,
+          idempotencyKeyExpiresAt: true,
+          inactiveIdempotencyKey: true,
+          output: true,
+          outputType: true,
+          outputIsError: true,
+          completedAfter: true,
+          completedAt: true,
+          createdAt: true,
+          tags: true,
         },
       });
 
-      const waitpoint =
-        result.source === "new" || result.source === "legacy-replica" ? result.value : null;
-
       if (!waitpoint) {
         logger.error(`WaitpointPresenter: Waitpoint not found`, {
           id: waitpointId,