fix(webapp,run-store): make run-ops sharding failures visible and unroutable ids a 404

Four QA findings from the N-way run-ops sharding pass:

- A replication source whose publication carries no tables replicated nothing
  while boot passed and the service looked healthy. The client now emits a typed
  PublicationMisconfiguredError and the webapp counts it per source
  (runs_replication_publication_misconfigured_total), so it is alarmable.
- POST /api/v1/waitpoints/tokens/:id/complete (and the HTTP-callback route)
  answered 500 for an id naming an unconfigured shard; both now answer 404 like
  the run routes.
- An unparseable runOpsMintShardSet still degrades to gen-1 minting, but the
  parse failure is now reported through the caller's logger instead of vanishing.
- Per-shard observability: runops_shard_routed_total{shard} from the router and
  runops_read_through_source_total{source,shard} from the read-through layer.

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

Author: d-cs

apps/webapp/app/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.callback.$hash.ts

@@ -6,6 +6,7 @@ import { env } from "~/env.server";
 import { processWaitpointCompletionPacket } from "~/runEngine/concerns/waitpointCompletionPacket.server";
 import { verifyHttpCallbackHash } from "~/services/httpCallback.server";
 import { logger } from "~/services/logger.server";
+import { unroutableIdResponse } from "~/services/routeBuilders/unroutableId.server";
 import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver.server";
 import { engine } from "~/v3/runEngine.server";
 import { runStore } from "~/v3/runStore.server";
@@ -102,6 +103,12 @@ export async function action({ request, params }: ActionFunctionArgs) {
       { status: 200 }
     );
   } catch (error) {
+    // Same as the complete route: the waitpoint id comes off the URL, so an unconfigured shard
+    // key is caller-supplied input and must answer 404 rather than 500. This route is a bare
+    // Remix action, so the api-builder boundary never sees the error — answer it here.
+    const unroutable = unroutableIdResponse(error);
+    if (unroutable) return unroutable;
+
     logger.error("Failed to complete HTTP callback", { error });
     throw json({ error: "Failed to complete HTTP callback" }, { status: 500 });
   }

apps/webapp/app/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.complete.ts

@@ -10,6 +10,7 @@ import { env } from "~/env.server";
 import { logger } from "~/services/logger.server";
 import { processWaitpointCompletionPacket } from "~/runEngine/concerns/waitpointCompletionPacket.server";
 import { createActionApiRoute } from "~/services/routeBuilders/apiBuilder.server";
+import { unroutableIdResponse } from "~/services/routeBuilders/unroutableId.server";
 import { engine } from "~/v3/runEngine.server";
 import { runStore } from "~/v3/runStore.server";
 
@@ -87,6 +88,11 @@ const { action, loader } = createActionApiRoute(
       // client gets the correct status code instead of a 500, and we don't log them as errors.
       if (error instanceof Response) throw error;
 
+      // A caller-supplied id naming a shard this topology has no store for cannot be routed,
+      // so it is a 404 like an absent token — not the 500 this catch would otherwise answer.
+      const unroutable = unroutableIdResponse(error);
+      if (unroutable) throw unroutable;
+
       logger.error("Failed to complete waitpoint token", {
         error:
           error instanceof Error

apps/webapp/app/services/runsReplicationInstance.server.ts

@@ -9,6 +9,7 @@ import {
   setRunsReplicationConfiguredSources,
   setRunsReplicationGlobal,
 } from "./runsReplicationGlobal.server";
+import { runsReplicationSourceMetrics } from "./runsReplicationMetrics.server";
 import {
   RunsReplicationService,
   type RunsReplicationSource,
@@ -249,6 +250,9 @@ function initializeRunsReplicationInstance() {
     insertStrategy: env.RUN_REPLICATION_INSERT_STRATEGY,
     disablePayloadInsert: env.RUN_REPLICATION_DISABLE_PAYLOAD_INSERT === "1",
     disableErrorFingerprinting: env.RUN_REPLICATION_DISABLE_ERROR_FINGERPRINTING === "1",
+    // A source whose publication carries no usable table logs every 30s and replicates nothing.
+    // Boot cannot see it (the source IS configured), so the counter is the alarmable signal.
+    onSourceError: runsReplicationSourceMetrics.recordSourceError,
   };
 
   // Construct the SINGLE legacy source synchronously (the split gate has not resolved

apps/webapp/app/services/runsReplicationMetrics.server.ts

@@ -0,0 +1,38 @@
+/**
+ * A replication source whose publication carries no usable table replicates nothing while the
+ * service stays up and healthy: boot passes, `assertReplicationCoversSplit` only checks that a
+ * source is CONFIGURED, and the client's retry loop logs every 30s. Counting it is what makes it
+ * alarmable — a non-zero rate here means that source's runs are not reaching ClickHouse.
+ */
+import { PublicationMisconfiguredError } from "@internal/replication";
+import { Counter, type Registry, type RegistryContentType } from "prom-client";
+import { metricsRegister } from "~/metrics.server";
+import { singleton } from "~/utils/singleton";
+
+export type RunsReplicationSourceMetrics = {
+  recordSourceError(info: { sourceId: string; error: unknown }): void;
+};
+
+export function buildRunsReplicationSourceMetrics(
+  register: Registry<RegistryContentType>
+): RunsReplicationSourceMetrics {
+  const publicationMisconfigured = new Counter({
+    name: "runs_replication_publication_misconfigured_total",
+    help: "A replication source's publication does not carry the replicated table, so that source replicates nothing.",
+    labelNames: ["source"],
+    registers: [register],
+  });
+
+  return {
+    recordSourceError: ({ sourceId, error }) => {
+      if (error instanceof PublicationMisconfiguredError) {
+        publicationMisconfigured.inc({ source: sourceId });
+      }
+    },
+  };
+}
+
+// singleton: module-scope Counter registration double-registers under dev HMR.
+export const runsReplicationSourceMetrics = singleton("runsReplicationSourceMetrics", () =>
+  buildRunsReplicationSourceMetrics(metricsRegister)
+);

apps/webapp/app/services/runsReplicationService.server.ts

@@ -116,6 +116,12 @@ export type RunsReplicationServiceOptions = {
   disablePayloadInsert?: boolean;
   disableErrorFingerprinting?: boolean;
   maxPoisonStripsPerBatch?: number;
+  /**
+   * Per-source client error hook. A client error does not stop the service — a misconfigured
+   * publication just replicates nothing while the retry loop logs — so the owner needs a seam to
+   * count it on.
+   */
+  onSourceError?: (info: { sourceId: string; error: unknown }) => void;
 };
 
 type PostgresTaskRun = TaskRun & { masterQueue: string };
@@ -470,6 +476,7 @@ export class RunsReplicationService {
         sourceId: source.id,
         error,
       });
+      this.options.onSourceError?.({ sourceId: source.id, error });
     });
 
     client.events.on("start", () => {

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

@@ -314,6 +314,39 @@ describe("resolveMintShardWith — cache, read failure and fail-safe", () => {
     expect(["a", "b"]).toContain(await resolveMintShardWith({ id: "env_1" }, deps));
   });
 
+  it("reports an unparseable stored list while still degrading to gen-1", async () => {
+    // The observed failure: `runOpsMintShardSet` saved as "A,B" (uppercase) reverted the whole
+    // fleet to gen-1 minting with ZERO log lines, because the parse throw was swallowed and
+    // `onReadFailed` never fires for it — the read succeeded. The degrade is correct; silence is not.
+    const deps = wrapperDeps({ readFlags: async () => ({ runOpsMintShardSet: "A,B" }) });
+    const readFailures: unknown[] = [];
+    const parseFailures: Array<{ key: string; value: string }> = [];
+    deps.onReadFailed = (error) => readFailures.push(error);
+    deps.onSetParseFailed = ({ key, value }) => parseFailures.push({ key, value });
+
+    expect(await resolveMintShardWith({ id: "env_1" }, deps)).toBe("new");
+    expect(readFailures).toEqual([]);
+    expect(parseFailures).toEqual([{ key: "runOpsMintShardSet", value: "A,B" }]);
+  });
+
+  it("reports a reserved key in the stored list too", async () => {
+    const deps = wrapperDeps({ readFlags: async () => ({ runOpsMintShardSet: "a,legacy" }) });
+    const parseFailures: Array<{ key: string; value: string }> = [];
+    deps.onSetParseFailed = ({ key, value }) => parseFailures.push({ key, value });
+
+    expect(await resolveMintShardWith({ id: "env_1" }, deps)).toBe("new");
+    expect(parseFailures).toEqual([{ key: "runOpsMintShardSet", value: "a,legacy" }]);
+  });
+
+  it("stays silent for a stored list that parses", async () => {
+    const deps = wrapperDeps();
+    const parseFailures: unknown[] = [];
+    deps.onSetParseFailed = (failure) => parseFailures.push(failure);
+
+    expect(["a", "b"]).toContain(await resolveMintShardWith({ id: "env_1" }, deps));
+    expect(parseFailures).toEqual([]);
+  });
+
   it("returns gen-1 when the stored list is empty", async () => {
     const deps = wrapperDeps({ readFlags: async () => ({ runOpsMintShardSet: "" }) });
     expect(await resolveMintShardWith({ id: "env_1" }, deps)).toBe("new");

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

@@ -8,6 +8,7 @@ import {
   GEN_1_PIN_VALUE,
   isValidPinValue,
   readMintShardSetResolution,
+  type MintShardSetParseFailure,
   type MintShardSetResolution,
 } from "./mintShardGrace";
 
@@ -166,6 +167,9 @@ export type ResolveMintShardDeps = {
   onPinRejected?: (info: { environmentId: string; pin: string; activeSet: string[] }) => void;
   onOverrideRejected?: (info: { override: string; activeSet: string[] }) => void;
   onReadFailed?: (error: unknown) => void;
+  // A stored set that PARSED badly, which is not a read failure: onReadFailed never fires for it,
+  // yet the fleet reverts to gen-1 minting. Reported here so the operator sees the degrade.
+  onSetParseFailed?: (failure: MintShardSetParseFailure) => void;
 };
 
 // The live list is org-independent, so one process-wide entry serves every mint: one query per
@@ -180,7 +184,7 @@ async function refreshConfig(deps: ResolveMintShardDeps): Promise<GlobalShardCon
   try {
     const flags = await deps.readFlags();
     const config: GlobalShardConfig = {
-      resolution: readMintShardSetResolution(flags),
+      resolution: readMintShardSetResolution(flags, deps.onSetParseFailed),
       override: flags[FEATURE_FLAG.runOpsMintShardOverride],
     };
     deps.cache.current = { value: config, expiresAt: deps.nowMs + deps.ttlMs };

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

@@ -67,15 +67,27 @@ export function effectiveMintShardSet(
   return nowMs < r.flippedAtMs + graceMs ? r.prevSet : r.set;
 }
 
+/** Reports a stored value that could not be parsed. The module stays pure; the caller logs. */
+export type MintShardSetParseFailure = { key: string; value: string; error: unknown };
+
 // The active set lives in the control-plane database, not in the environment. A deploy rolls
 // for hours, so two pods can hold different environment values at the same time; only a shared
 // row lets every pod agree on one set. Boot may reject a bad environment value, but the mint
 // path must never throw on a bad stored value, so an unreadable list degrades to empty.
-function readStoredCsv(value: unknown): string[] {
+//
+// Degrading is right; degrading SILENTLY is not. An unparseable value (a `"A,B"` typed in
+// uppercase) reverts the whole fleet to gen-1 minting, so the failure is reported to the caller,
+// which owns the logger. The reserved-key and alphabet rejections are the same failure.
+function readStoredCsv(
+  value: unknown,
+  key: string,
+  onInvalid?: (failure: MintShardSetParseFailure) => void
+): string[] {
   if (typeof value !== "string") return [];
   try {
     return parseShardCsv(value);
-  } catch {
+  } catch (error) {
+    onInvalid?.({ key, value, error });
     return [];
   }
 }
@@ -84,16 +96,20 @@ function readStoredCsv(value: unknown): string[] {
 // timestamp can never apply, so it is dropped. A timestamp with an EMPTY prevSet is meaningful:
 // it graces a first activation, serving no shards for the window.
 export function readMintShardSetResolution(
-  flags: Record<string, unknown> | null | undefined
+  flags: Record<string, unknown> | null | undefined,
+  onInvalid?: (failure: MintShardSetParseFailure) => void
 ): MintShardSetResolution {
   const source = flags ?? {};
   const flippedAtRaw = source[SET_FLIPPED_AT_KEY];
   const parsed = typeof flippedAtRaw === "string" ? Date.parse(flippedAtRaw) : NaN;
   const flippedAtMs = Number.isNaN(parsed) ? undefined : parsed;
 
   return {
-    set: readStoredCsv(source[SET_KEY]),
-    prevSet: flippedAtMs === undefined ? undefined : readStoredCsv(source[SET_PREV_KEY]),
+    set: readStoredCsv(source[SET_KEY], SET_KEY, onInvalid),
+    prevSet:
+      flippedAtMs === undefined
+        ? undefined
+        : readStoredCsv(source[SET_PREV_KEY], SET_PREV_KEY, onInvalid),
     flippedAtMs,
   };
 }
@@ -113,8 +129,8 @@ export function stampMintShardSetFlip(
   }
 
   const existing = existingFlags ?? {};
-  const outgoingSet = readStoredCsv(outgoingFlags[SET_KEY]);
-  const storedSet = readStoredCsv(existing[SET_KEY]);
+  const outgoingSet = readStoredCsv(outgoingFlags[SET_KEY], SET_KEY);
+  const storedSet = readStoredCsv(existing[SET_KEY], SET_KEY);
 
   if (outgoingSet.join(",") !== storedSet.join(",")) {
     const effective = effectiveMintShardSet(readMintShardSetResolution(existing), nowMs, graceMs);

apps/webapp/app/v3/runOpsMigration/readThrough.server.test.ts

@@ -309,4 +309,70 @@ describe("readThroughRun (legacy replica + new DB)", () => {
       expect(throwingLegacy).not.toHaveBeenCalled();
     }
   );
+  // Which store served a read was a return value only — never emitted — so during a cohort ramp
+  // there was no way to see from outside the process where reads were landing.
+  heteroPostgresTest(
+    "emits the serving source for a gen-2 shard, the gen-1 new store and the legacy replica",
+    async ({ prisma14, prisma17 }) => {
+      const emitted: string[] = [];
+      const deps = {
+        splitEnabled: true,
+        newClient: prisma17 as unknown as PrismaReplicaClient,
+        legacyReplica: prisma14 as unknown as PrismaReplicaClient,
+        shardReplicas: new Map([["a", prisma17 as unknown as PrismaReplicaClient]]),
+        onSource: (source: string) => emitted.push(source),
+      };
+
+      await readThroughRun({
+        id: SHARD_A_RUN_ID,
+        idKind: "run",
+        environmentId: "env_1",
+        readNew: (c) => realRead(c, true),
+        readLegacy: (c) => realRead(c, false),
+        deps,
+      });
+      await readThroughRun({
+        id: NEW_RUN_ID,
+        idKind: "run",
+        environmentId: "env_1",
+        readNew: (c) => realRead(c, true),
+        readLegacy: (c) => realRead(c, false),
+        deps,
+      });
+      await readThroughRun({
+        id: LEGACY_RUN_ID,
+        idKind: "run",
+        environmentId: "env_1",
+        readNew: (c) => realRead(c, false),
+        readLegacy: (c) => realRead(c, true),
+        deps,
+      });
+
+      expect(emitted).toEqual(["shard:a", "new", "legacy-replica"]);
+    }
+  );
+
+  heteroPostgresTest(
+    "emits nothing for a miss, so a not-found cannot look like a hit",
+    async ({ prisma14, prisma17 }) => {
+      const emitted: string[] = [];
+
+      const result = await readThroughRun({
+        id: LEGACY_RUN_ID,
+        idKind: "run",
+        environmentId: "env_1",
+        readNew: (c) => realRead(c, false),
+        readLegacy: (c) => realRead(c, false),
+        deps: {
+          splitEnabled: true,
+          newClient: prisma17 as unknown as PrismaReplicaClient,
+          legacyReplica: prisma14 as unknown as PrismaReplicaClient,
+          onSource: (source: string) => emitted.push(source),
+        },
+      });
+
+      expect(result.found).toBe(false);
+      expect(emitted).toEqual([]);
+    }
+  );
 });