feat(webapp): RUN_OPS_SHARDS config, topology and N-way store wiring

apps/webapp/app/env.server.ts

@@ -2,6 +2,7 @@ import { z } from "zod";
 import { MachinePresetName } from "@trigger.dev/core/v3";
 import { BoolEnv } from "./utils/boolEnv";
 import { isValidDatabaseUrl } from "./utils/db";
+import { parseRunOpsShards, validateShardListAgainstNewUrl } from "~/v3/runOpsShards.server";
 import { isValidRegex } from "./utils/regex";
 import { isValidDuration } from "./services/realtime/duration.server";
 
@@ -310,6 +311,8 @@ const EnvironmentSchema = z
     RUN_OPS_DATABASE_REPLICA_DRIVER_ADAPTER: z.string().default("0"),
     RUN_OPS_LEGACY_DATABASE_WRITER_DRIVER_ADAPTER: z.string().default("0"),
     RUN_OPS_LEGACY_DATABASE_REPLICA_DRIVER_ADAPTER: z.string().default("0"),
+    // Gen-2 shard descriptors as a JSON array. Unset/"" -> [] (today). See runOpsShards.server.ts.
+    RUN_OPS_SHARDS: z.string().optional().transform(parseRunOpsShards),
     // Control-plane cache relax knobs. Unset -> defaults (DEFAULT_CP_CACHE_TTL_MS / _MAX_ENTRIES).
     CONTROL_PLANE_CACHE_TTL_MS: z.coerce.number().int().optional(),
     CONTROL_PLANE_CACHE_MAX_ENTRIES: z.coerce.number().int().optional(),
@@ -2467,6 +2470,14 @@ const EnvironmentSchema = z
         });
       }
     }
+    if (!validateShardListAgainstNewUrl(env.RUN_OPS_SHARDS, env.RUN_OPS_DATABASE_URL)) {
+      ctx.addIssue({
+        code: z.ZodIssueCode.custom,
+        path: ["RUN_OPS_SHARDS"],
+        message:
+          "RUN_OPS_SHARDS is non-empty but RUN_OPS_DATABASE_URL is unset; a shard requires the gen-1 new store",
+      });
+    }
   });
 
 export type Environment = z.infer<typeof EnvironmentSchema>;

apps/webapp/app/v3/runOpsMigration/mintShardAssignment.test.ts

@@ -473,3 +473,40 @@ describe("computeMintShard — the global override wins the complete cutover", (
     );
   });
 });
+
+describe("routableKeys bound (the shard descriptor keys this deployment can route)", () => {
+  it("drops an active key that is not routable, so the hash never returns it", () => {
+    // "z" is in the active list but not configured as a descriptor -> only "a" is selectable.
+    const ids = envIds(200);
+    for (const id of ids) {
+      const shard = computeMintShard({ id }, deps({ set: ["a", "z"] }, { routableKeys: ["a"] }));
+      expect(shard).toBe("a");
+    }
+  });
+
+  it("returns new when the active list holds only non-routable keys (fail-safe to gen-1)", () => {
+    expect(computeMintShard({ id: "env_1" }, deps({ set: ["z"] }, { routableKeys: ["a"] }))).toBe(
+      "new"
+    );
+  });
+
+  it("rejects a per-org pin to a non-routable key and falls through to the hash", () => {
+    const shard = computeMintShard(
+      { id: "env_1" },
+      deps({ set: ["a", "z"] }, { ...orgFlags({ runOpsMintShard: "z" }), routableKeys: ["a"] })
+    );
+    expect(shard).toBe("a");
+  });
+
+  it("with no routableKeys given, behaviour is unchanged", () => {
+    const ids = envIds(200);
+    for (const id of ids) {
+      const withBound = computeMintShard(
+        { id },
+        deps({ set: ["a", "b"] }, { routableKeys: ["a", "b"] })
+      );
+      const without = computeMintShard({ id }, deps({ set: ["a", "b"] }));
+      expect(withBound).toBe(without);
+    }
+  });
+});

apps/webapp/app/v3/runOpsMigration/mintShardAssignment.ts

@@ -19,6 +19,10 @@ export type MintShardDeps = {
   nowMs: number;
   graceMs: number;
   orgFeatureFlags: unknown;
+  // The shard keys this deployment can actually route (the RUN_OPS_SHARDS descriptor keys). The
+  // active set is bounded to these, so a stored key with no descriptor is never minted into.
+  // Undefined means "no bound" (today's behaviour).
+  routableKeys?: readonly string[];
   onPinRejected?: (info: { environmentId: string; pin: string; activeSet: string[] }) => void;
   onOverrideRejected?: (info: { override: string; activeSet: string[] }) => void;
 };
@@ -94,7 +98,17 @@ function hrwSelect(environmentId: string, activeSet: string[]): string {
 // would leak the drain the active list performs, and throwing would fail customer triggers
 // whenever a pinned shard drains.
 export function computeMintShard(environment: { id: string }, deps: MintShardDeps): ShardKey {
-  const activeSet = effectiveMintShardSet(deps.resolution, deps.nowMs, deps.graceMs);
+  const rawActiveSet = effectiveMintShardSet(deps.resolution, deps.nowMs, deps.graceMs);
+  // Empty check BEFORE the bound, so an unconfigured deployment returns "new" exactly as today.
+  if (rawActiveSet.length === 0) {
+    return "new";
+  }
+
+  // Bound the active set to the keys this deployment can route. A stored key with no descriptor is
+  // dropped, never minted into. If nothing survives, fall back to gen-1 (fail-safe, never a throw).
+  const activeSet = deps.routableKeys
+    ? rawActiveSet.filter((key) => deps.routableKeys!.includes(key))
+    : rawActiveSet;
   if (activeSet.length === 0) {
     return "new";
   }
@@ -148,6 +162,7 @@ export type ResolveMintShardDeps = {
   ttlMs: number;
   graceMs: number;
   orgFeatureFlags: unknown;
+  routableKeys?: readonly string[];
   onPinRejected?: (info: { environmentId: string; pin: string; activeSet: string[] }) => void;
   onOverrideRejected?: (info: { override: string; activeSet: string[] }) => void;
   onReadFailed?: (error: unknown) => void;
@@ -200,6 +215,7 @@ export async function resolveMintShardWith(
     nowMs: deps.nowMs,
     graceMs: deps.graceMs,
     orgFeatureFlags: deps.orgFeatureFlags,
+    routableKeys: deps.routableKeys,
     onPinRejected: deps.onPinRejected,
     onOverrideRejected: deps.onOverrideRejected,
   });

apps/webapp/app/v3/runOpsMigration/runOpsMintShard.server.ts

@@ -84,6 +84,8 @@ export async function resolveMintShard(environment: {
     ttlMs: env.RUN_OPS_MINT_FLAG_CACHE_TTL_MS,
     graceMs: env.RUN_OPS_MINT_FLIP_GRACE_MS,
     orgFeatureFlags: environment.orgFeatureFlags,
+    // Bound the active list to the shards this deployment can actually route.
+    routableKeys: env.RUN_OPS_SHARDS.map((shard) => shard.key),
     onPinRejected: reportPinRejected,
     onOverrideRejected: reportOverrideRejected,
     onReadFailed: (error) =>

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

@@ -0,0 +1,80 @@
+import { env } from "~/env.server";
+import type { RunOpsShardKnobs } from "~/v3/runOpsShards.server";
+
+// Pool configuration for one run-ops store (writer + replica). Kept separate from db.server (which
+// ~156 tests mock wholesale) so a new export breaks no mock.
+export type ResolvedPoolKnobs = {
+  writerPoolTimeout: number;
+  writerConnectionTimeout: number;
+  writerDriverAdapter: boolean;
+  connectionLimit: number;
+  replicaConnectionLimit: number;
+  replicaPoolTimeout: number;
+  replicaConnectionTimeout: number;
+  replicaDriverAdapter: boolean;
+};
+
+type Role = "new" | "legacy";
+
+// PURE: overlay a gen-2 shard's descriptor knobs on a role's resolved defaults. This holds the only
+// logic (per-field override), so a test drives it with literal defaults and literal overrides —
+// no env import, no circular assertion against the same env expression the impl reads.
+export function applyPoolKnobOverrides(
+  defaults: ResolvedPoolKnobs,
+  k?: RunOpsShardKnobs
+): ResolvedPoolKnobs {
+  return {
+    writerPoolTimeout: k?.writerPoolTimeout ?? defaults.writerPoolTimeout,
+    writerConnectionTimeout: k?.writerConnectionTimeout ?? defaults.writerConnectionTimeout,
+    writerDriverAdapter: k?.writerDriverAdapter ?? defaults.writerDriverAdapter,
+    connectionLimit: k?.connectionLimit ?? defaults.connectionLimit,
+    replicaConnectionLimit: k?.replicaConnectionLimit ?? defaults.replicaConnectionLimit,
+    replicaPoolTimeout: k?.replicaPoolTimeout ?? defaults.replicaPoolTimeout,
+    replicaConnectionTimeout: k?.replicaConnectionTimeout ?? defaults.replicaConnectionTimeout,
+    replicaDriverAdapter: k?.replicaDriverAdapter ?? defaults.replicaDriverAdapter,
+  };
+}
+
+// The env-derived defaults for a role, reproducing today's run-ops builder expressions exactly. A
+// flat mapping (no logic), verified by inspection against the former builders. Transaction
+// resilience is a SEPARATE mechanism (resolveTransactionResilience) and is not here.
+function poolKnobDefaults(role: Role): ResolvedPoolKnobs {
+  if (role === "legacy") {
+    return {
+      writerPoolTimeout:
+        env.RUN_OPS_LEGACY_DATABASE_WRITER_POOL_TIMEOUT ?? env.DATABASE_POOL_TIMEOUT,
+      writerConnectionTimeout:
+        env.RUN_OPS_LEGACY_DATABASE_WRITER_CONNECTION_TIMEOUT ?? env.DATABASE_CONNECTION_TIMEOUT,
+      writerDriverAdapter: env.RUN_OPS_LEGACY_DATABASE_WRITER_DRIVER_ADAPTER === "1",
+      connectionLimit: env.DATABASE_CONNECTION_LIMIT,
+      replicaConnectionLimit: env.DATABASE_CONNECTION_LIMIT,
+      replicaPoolTimeout:
+        env.RUN_OPS_LEGACY_DATABASE_READ_REPLICA_POOL_TIMEOUT ?? env.DATABASE_POOL_TIMEOUT,
+      replicaConnectionTimeout:
+        env.RUN_OPS_LEGACY_DATABASE_READ_REPLICA_CONNECTION_TIMEOUT ??
+        env.DATABASE_CONNECTION_TIMEOUT,
+      replicaDriverAdapter: env.RUN_OPS_LEGACY_DATABASE_REPLICA_DRIVER_ADAPTER === "1",
+    };
+  }
+
+  return {
+    writerPoolTimeout: env.RUN_OPS_DATABASE_WRITER_POOL_TIMEOUT ?? env.DATABASE_POOL_TIMEOUT,
+    writerConnectionTimeout:
+      env.RUN_OPS_DATABASE_WRITER_CONNECTION_TIMEOUT ?? env.DATABASE_CONNECTION_TIMEOUT,
+    writerDriverAdapter: env.RUN_OPS_DATABASE_WRITER_DRIVER_ADAPTER === "1",
+    connectionLimit: env.DATABASE_CONNECTION_LIMIT,
+    replicaConnectionLimit:
+      env.RUN_OPS_DATABASE_READ_REPLICA_CONNECTION_LIMIT ?? env.DATABASE_CONNECTION_LIMIT,
+    replicaPoolTimeout: env.RUN_OPS_DATABASE_READ_REPLICA_POOL_TIMEOUT ?? env.DATABASE_POOL_TIMEOUT,
+    replicaConnectionTimeout:
+      env.RUN_OPS_DATABASE_READ_REPLICA_CONNECTION_TIMEOUT ?? env.DATABASE_CONNECTION_TIMEOUT,
+    replicaDriverAdapter: env.RUN_OPS_DATABASE_REPLICA_DRIVER_ADAPTER === "1",
+  };
+}
+
+export function resolveRunOpsPoolKnobs(
+  role: Role,
+  descriptorKnobs?: RunOpsShardKnobs
+): ResolvedPoolKnobs {
+  return applyPoolKnobOverrides(poolKnobDefaults(role), descriptorKnobs);
+}

apps/webapp/app/v3/runOpsShardTable.ts

@@ -0,0 +1,28 @@
+// Pure boot-table helpers. Dependency-free (no db.server, no env) so a test of these two string
+// functions never constructs a Prisma client. db.server imports them for the boot log.
+
+// A host:port/db address, with NO username and NO query params — never a secret, and deliberately
+// NOT an identity claim (two DSNs can share an address yet be different databases; that proof is the
+// distinctness sentinel's, not this line's). Same tuple sameDatabaseTarget compares, kept in step.
+export function runOpsAddressFingerprint(url: string): string {
+  try {
+    const u = new URL(url);
+    return `${u.hostname}:${u.port || "5432"}${u.pathname}`;
+  } catch {
+    return "unparseable";
+  }
+}
+
+export type RunOpsShardTableRow = { key: string; fingerprint: string; role: string };
+
+// The resolved shard table for the boot log: one row per descriptor. An alias reports its role and
+// carries no address (it shares the new store's pool).
+export function buildRunOpsShardTable(
+  descriptors: Array<{ key: string; url?: string; aliasOf?: "new" }>
+): RunOpsShardTableRow[] {
+  return descriptors.map((d) =>
+    d.aliasOf
+      ? { key: d.key, fingerprint: "alias(new)", role: "alias(new)" }
+      : { key: d.key, fingerprint: runOpsAddressFingerprint(d.url ?? ""), role: "shard" }
+  );
+}

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

@@ -0,0 +1,124 @@
+import { z } from "zod";
+import { isValidShardChar } from "@trigger.dev/core/v3/isomorphic";
+import { isValidDatabaseUrl } from "~/utils/db";
+
+const KnobsSchema = z
+  .object({
+    writerPoolTimeout: z.number().int().optional(),
+    writerConnectionTimeout: z.number().int().optional(),
+    writerDriverAdapter: z.boolean().optional(),
+    connectionLimit: z.number().int().optional(),
+    replicaConnectionLimit: z.number().int().optional(),
+    replicaPoolTimeout: z.number().int().optional(),
+    replicaConnectionTimeout: z.number().int().optional(),
+    replicaDriverAdapter: z.boolean().optional(),
+    transactionMaxWaitMs: z.number().int().optional(),
+    transactionStartRetryEnabled: z.boolean().optional(),
+    transactionStartRetryMaxAttempts: z.number().int().optional(),
+    transactionStartRetryBackoffMinMs: z.number().int().optional(),
+    transactionStartRetryBackoffMaxMs: z.number().int().optional(),
+    transactionStartRetryBudgetPerSec: z.number().int().optional(),
+    transactionStartRetryBudgetBurst: z.number().int().optional(),
+  })
+  .strict();
+export type RunOpsShardKnobs = z.infer<typeof KnobsSchema>;
+
+const ReplicationSchema = z.object({
+  slotName: z.string().min(1),
+  publicationName: z.string().min(1),
+  originGeneration: z.number().int().min(2).max(255),
+});
+
+const DescriptorSchema = z
+  .object({
+    key: z.string().refine(isValidShardChar, "shard key must be a single [a-z0-9] char"),
+    region: z.string().min(1),
+    url: z.string().refine(isValidDatabaseUrl, "url is invalid").optional(),
+    replicaUrl: z.string().refine(isValidDatabaseUrl, "replicaUrl is invalid").optional(),
+    directUrl: z.string().refine(isValidDatabaseUrl, "directUrl is invalid").optional(),
+    replication: ReplicationSchema.optional(),
+    knobs: KnobsSchema.optional(),
+    aliasOf: z.literal("new").optional(),
+  })
+  .strict()
+  .superRefine((d, ctx) => {
+    const hasUrl = d.url !== undefined;
+    const hasAlias = d.aliasOf !== undefined;
+    if (hasUrl === hasAlias) {
+      ctx.addIssue({
+        code: z.ZodIssueCode.custom,
+        message: "exactly one of url or aliasOf is required",
+      });
+    }
+    if (!hasAlias && d.replication === undefined) {
+      ctx.addIssue({
+        code: z.ZodIssueCode.custom,
+        message: "replication is required unless aliasOf is set",
+      });
+    }
+  });
+
+export type RunOpsShardDescriptor = z.infer<typeof DescriptorSchema>;
+
+// Boot-validated transform, in the style of parseMachinePresetCsv. Undefined and "" both mean the
+// off state and resolve to []. The undefined guard is load-bearing: an unguarded JSON.parse would
+// kill every single-DB boot, which never sets this variable.
+export function parseRunOpsShards(
+  raw: string | undefined,
+  ctx: z.RefinementCtx
+): RunOpsShardDescriptor[] {
+  if (raw === undefined || raw.trim() === "") return [];
+
+  let parsed: unknown;
+  try {
+    parsed = JSON.parse(raw);
+  } catch {
+    ctx.addIssue({ code: z.ZodIssueCode.custom, message: "RUN_OPS_SHARDS is not valid JSON" });
+    return z.NEVER;
+  }
+
+  const result = z.array(DescriptorSchema).safeParse(parsed);
+  if (!result.success) {
+    for (const issue of result.error.issues) {
+      ctx.addIssue({
+        code: z.ZodIssueCode.custom,
+        message: `RUN_OPS_SHARDS[${issue.path.join(".")}]: ${issue.message}`,
+      });
+    }
+    return z.NEVER;
+  }
+
+  const keys = new Set<string>();
+  const gens = new Set<number>();
+  for (const d of result.data) {
+    if (keys.has(d.key)) {
+      ctx.addIssue({
+        code: z.ZodIssueCode.custom,
+        message: `RUN_OPS_SHARDS: duplicate key ${d.key}`,
+      });
+      return z.NEVER;
+    }
+    keys.add(d.key);
+    if (d.replication) {
+      if (gens.has(d.replication.originGeneration)) {
+        ctx.addIssue({
+          code: z.ZodIssueCode.custom,
+          message: `RUN_OPS_SHARDS: duplicate originGeneration ${d.replication.originGeneration}`,
+        });
+        return z.NEVER;
+      }
+      gens.add(d.replication.originGeneration);
+    }
+  }
+
+  return result.data;
+}
+
+// A non-empty shard list requires the gen-1 new store, because gen-1 v1 ids resolve to "new"
+// forever (append-only). Pure so the boot refinement and its test share one rule.
+export function validateShardListAgainstNewUrl(
+  shards: RunOpsShardDescriptor[],
+  newUrl: string | undefined
+): boolean {
+  return shards.length === 0 || !!newUrl;
+}