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

## Summary

Four failure modes found while exercising a multi-database run-ops setup
by hand. None stopped the service: each was unalertable, unreported, or
answered with the wrong status.

- A replication source whose Postgres publication carries no tables
replicates nothing. Boot succeeds, the service looks healthy, and every
ClickHouse-backed surface silently under-counts. There is now a counter
to alert on.
- Completing a wait token with an ID naming a database the deployment
has no store for answered a server error. It now answers not-found, like
the equivalent run routes.
- An unparseable stored shard set still degrades safely to the previous
ID format, but the parse failure went unreported. It is now reported on
the mint path.
- Per-shard observability, without which a gradual rollout cannot be
watched.

## Design

**Publication misconfiguration.** The replication client already
detected an empty publication and emitted a
`LogicalReplicationClientError` on every reconnect, which the service
logged. That is visible but not alertable, and indistinguishable from
any other client error. It now emits a `PublicationMisconfiguredError`
subclass of that error, counted per source as
`runs_replication_publication_misconfigured_total`. No boot-time check
was added: it would need a live query per source before the service
exists, so a transient blip would fail closed and refuse to boot, which
is worse than the problem being solved. The counter increments on every
retry, so a nonzero rate is the alarm and it clears when the publication
is repaired.

Repairing the publication restores replication going forward but does
not backfill rows written while it was empty, so catching this early is
the point.

**Unroutable IDs.** All seven API route builders now map this to a 404,
including the two worker builders that previously fell through to a 500.
Both new call sites log before answering, so a shard key dropped from a
config meant to be append-only still alarms rather than turning every
live token on it into a quiet not-found.

Routes that handle their own errors never reach a builder, so a sweep
covers those too: four bare API routes that answered 500 (run result,
run tags, reschedule, batch results) and ten dashboard and resource
routes that threw straight through to an error page. Each takes the
not-found path it already had for a run that does not exist, and logs
first.

**Shard-set parse failures.** The parsing module is pure by design and
its tests depend on that, so it reports through a callback and the
caller logs, matching the existing operator reports. This covers the
mint path, which every process re-reads within one cache TTL, so a bad
set alarms across the fleet. Saving one is still silent at save time:
the stamping helper takes the same callback but its caller does not yet
pass it.

**Observability.** `runops_shard_routed_total{shard}` counts ID-routed
store resolutions, and `runops_read_through_source_total{source,shard}`
records which store served a read-through. The first is deliberately a
relative ramp signal rather than a request count: fan-outs and probe
hits resolve no single shard, so they are not counted, and the help text
says so. Label children are cached rather than hashed per call, since
both sit on hot read paths.

Each fix has a test that fails if the corresponding production change is
reverted, with two exceptions worth naming: the lines binding the metric
recorders into the production service and read-through defaults are
module-scope wiring that the tests inject around.

Author: d-cs

apps/webapp/app/routes/@.runs.$runParam.ts

@@ -8,6 +8,7 @@ import { redirectWithErrorMessage } from "~/models/message.server";
 import { requireUser } from "~/services/session.server";
 import { impersonate, rootPath, v3RunPath, v3RunSpanPath } from "~/utils/pathBuilder";
 import { findBufferedRunRedirectInfo } from "~/v3/mollifier/syntheticRedirectInfo.server";
+import { undefinedOnUnroutableId } from "~/v3/runOpsMigration/unroutableRead.server";
 
 const ParamsSchema = z.object({
   runParam: z.string(),
@@ -33,17 +34,21 @@ export async function loader({ params, request }: LoaderFunctionArgs) {
     );
   }
 
-  const run = await runStore.findRun(
-    {
-      friendlyId: runParam,
-    },
-    {
-      select: {
-        spanId: true,
-        runtimeEnvironmentId: true,
-      },
-    },
-    prisma
+  const run = await undefinedOnUnroutableId(
+    () =>
+      runStore.findRun(
+        {
+          friendlyId: runParam,
+        },
+        {
+          select: {
+            spanId: true,
+            runtimeEnvironmentId: true,
+          },
+        },
+        prisma
+      ),
+    { runParam: params.runParam ?? params.runId }
   );
 
   if (!run) {

apps/webapp/app/routes/api.v1.batches.$batchParam.results.ts

@@ -5,6 +5,7 @@ import { runOpsLegacyReplica, runOpsNewReplica, runOpsSplitReadEnabled } from "~
 import { ApiBatchResultsPresenter } from "~/presenters/v3/ApiBatchResultsPresenter.server";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
+import { unroutableIdResponse } from "~/services/routeBuilders/unroutableId.server";
 
 const ParamsSchema = z.object({
   /* This is the batch friendly ID */
@@ -42,6 +43,14 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
 
       return json(result);
     } catch (error) {
+      const unroutable = unroutableIdResponse(error);
+      if (unroutable) {
+        logger.warn("Unroutable batch id on batch results", {
+          error: error instanceof Error ? error.message : error,
+        });
+        return unroutable;
+      }
+
       logger.error("Failed to load batch results", { error });
       return json({ error: "Something went wrong, please try again." }, { status: 500 });
     }

apps/webapp/app/routes/api.v1.runs.$runId.tags.ts

@@ -10,6 +10,7 @@ import { logger } from "~/services/logger.server";
 import { publishChangeRecord } from "~/services/realtime/runChangeNotifierInstance.server";
 import { mutateWithFallback } from "~/v3/mollifier/mutateWithFallback.server";
 import { runStore } from "~/v3/runStore.server";
+import { unroutableIdResponse } from "~/services/routeBuilders/unroutableId.server";
 
 // Pull the existing tags out of a buffer entry's serialised payload so
 // the buffer-path response can dedup against them, matching the
@@ -137,6 +138,14 @@ export async function action({ request, params }: ActionFunctionArgs) {
     }
     return outcome.response;
   } catch (error) {
+    const unroutable = unroutableIdResponse(error);
+    if (unroutable) {
+      logger.warn("Unroutable run id on run tags", {
+        error: error instanceof Error ? error.message : error,
+      });
+      return unroutable;
+    }
+
     logger.error("Failed to add run tags", { error });
     return json({ error: "Something went wrong, please try again." }, { status: 500 });
   }

apps/webapp/app/routes/api.v1.runs.$runParam.reschedule.ts

@@ -12,6 +12,7 @@ import { RescheduleTaskRunService } from "~/v3/services/rescheduleTaskRun.server
 import { mutateWithFallback } from "~/v3/mollifier/mutateWithFallback.server";
 import { getMollifierBuffer } from "~/v3/mollifier/mollifierBuffer.server";
 import { parseDelay } from "~/utils/delays";
+import { unroutableIdResponse } from "~/services/routeBuilders/unroutableId.server";
 
 const ParamsSchema = z.object({
   runParam: z.string(),
@@ -156,6 +157,14 @@ export async function action({ request, params }: ActionFunctionArgs) {
     if (error instanceof ServiceValidationError) {
       return json({ error: error.message }, { status: 400 });
     }
+    const unroutable = unroutableIdResponse(error);
+    if (unroutable) {
+      logger.warn("Unroutable run id on reschedule", {
+        error: error instanceof Error ? error.message : error,
+      });
+      return unroutable;
+    }
+
     logger.error("Failed to reschedule run", { error });
     return json({ error: "Something went wrong, please try again." }, { status: 500 });
   }

apps/webapp/app/routes/api.v1.runs.$runParam.result.ts

@@ -5,6 +5,7 @@ import { ApiRunResultPresenter } from "~/presenters/v3/ApiRunResultPresenter.ser
 import { runOpsLegacyReplica, runOpsNewReplica, runOpsSplitReadEnabled } from "~/db.server";
 import { authenticateApiRequest } from "~/services/apiAuth.server";
 import { logger } from "~/services/logger.server";
+import { unroutableIdResponse } from "~/services/routeBuilders/unroutableId.server";
 
 const ParamsSchema = z.object({
   /* This is the run friendly ID */
@@ -41,6 +42,14 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
 
     return json(result);
   } catch (error) {
+    const unroutable = unroutableIdResponse(error);
+    if (unroutable) {
+      logger.warn("Unroutable run id on run result", {
+        error: error instanceof Error ? error.message : error,
+      });
+      return unroutable;
+    }
+
     logger.error("Failed to load run result", { error });
     return json({ error: "Something went wrong, please try again." }, { status: 500 });
   }

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,19 @@ 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) {
+      // Same reason as the complete route: a silent 404 would hide a dropped shard key.
+      logger.warn("Unroutable waitpoint id on HTTP callback", {
+        waitpointFriendlyId: params.waitpointFriendlyId,
+        error: error instanceof Error ? error.message : error,
+      });
+      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,19 @@ 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) {
+        // Logged so a shard key dropped from an append-only config still alarms, rather than
+        // every live token on it quietly answering "not found".
+        logger.warn("Unroutable waitpoint id on token completion", {
+          waitpointFriendlyId: params.waitpointFriendlyId,
+          error: error instanceof Error ? error.message : error,
+        });
+        throw unroutable;
+      }
+
       logger.error("Failed to complete waitpoint token", {
         error:
           error instanceof Error

apps/webapp/app/routes/orgs.$organizationSlug.projects.$projectParam.runs.$runParam.ts

@@ -6,6 +6,7 @@ import { requireUserId } from "~/services/session.server";
 import { ProjectParamSchema, v3RunPath } from "~/utils/pathBuilder";
 import { runStore } from "~/v3/runStore.server";
 import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver.server";
+import { undefinedOnUnroutableId } from "~/v3/runOpsMigration/unroutableRead.server";
 
 const ParamSchema = ProjectParamSchema.extend({
   runParam: z.string(),
@@ -15,16 +16,20 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => {
   const userId = await requireUserId(request);
   const { organizationSlug, projectParam, runParam } = ParamSchema.parse(params);
 
-  const run = await runStore.findRun(
-    {
-      friendlyId: runParam,
-    },
-    {
-      select: {
-        projectId: true,
-        runtimeEnvironmentId: true,
-      },
-    }
+  const run = await undefinedOnUnroutableId(
+    () =>
+      runStore.findRun(
+        {
+          friendlyId: runParam,
+        },
+        {
+          select: {
+            projectId: true,
+            runtimeEnvironmentId: true,
+          },
+        }
+      ),
+    { runParam: params.runParam ?? params.runId }
   );
 
   if (!run) {

apps/webapp/app/routes/projects.v3.$projectRef.runs.$runParam.ts

@@ -5,6 +5,7 @@ import { requireUserId } from "~/services/session.server";
 import { v3RunSpanPath } from "~/utils/pathBuilder";
 import { runStore } from "~/v3/runStore.server";
 import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver.server";
+import { undefinedOnUnroutableId } from "~/v3/runOpsMigration/unroutableRead.server";
 
 const ParamsSchema = z.object({
   projectRef: z.string(),
@@ -36,18 +37,22 @@ export async function loader({ params, request }: LoaderFunctionArgs) {
     return new Response("Not found", { status: 404 });
   }
 
-  const run = await runStore.findRun(
-    {
-      friendlyId: validatedParams.runParam,
-    },
-    {
-      select: {
-        friendlyId: true,
-        spanId: true,
-        runtimeEnvironmentId: true,
-      },
-    },
-    prisma
+  const run = await undefinedOnUnroutableId(
+    () =>
+      runStore.findRun(
+        {
+          friendlyId: validatedParams.runParam,
+        },
+        {
+          select: {
+            friendlyId: true,
+            spanId: true,
+            runtimeEnvironmentId: true,
+          },
+        },
+        prisma
+      ),
+    { runParam: params.runParam ?? params.runId }
   );
 
   if (!run) {

apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.sessions.$sessionId.$io.ts

@@ -13,6 +13,7 @@ import {
 import { getRealtimeStreamInstance } from "~/services/realtime/v1StreamsGlobal.server";
 import { requireUserId } from "~/services/session.server";
 import { EnvironmentParamSchema } from "~/utils/pathBuilder";
+import { undefinedOnUnroutableId } from "~/v3/runOpsMigration/unroutableRead.server";
 
 const ParamsSchema = z.object({
   runParam: z.string(),
@@ -61,8 +62,12 @@ export async function loader({ request, params }: LoaderFunctionArgs) {
   // Replica lag can null out a live run; a spurious 404 breaks the dashboard Agent tab subscription
   // (useRealtimeStream surfaces the error and does not auto-retry). Re-read the primary on a miss.
   const run =
-    (await runStore.findRun(runWhere, runArgs, $replica)) ??
-    (await runStore.findRunOnPrimary(runWhere, runArgs));
+    (await undefinedOnUnroutableId(() => runStore.findRun(runWhere, runArgs, $replica), {
+      runParam: params.runParam,
+    })) ??
+    (await undefinedOnUnroutableId(() => runStore.findRunOnPrimary(runWhere, runArgs), {
+      runParam: params.runParam,
+    }));
 
   if (!run) {
     return new Response("Run not found", { status: 404 });