From ff518b5a91fe2d1a878908b9f124778b6c6b49fd Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Tue, 1 Sep 2026 15:38:34 +0100 Subject: [PATCH 1/4] fix(webapp,run-store): make run-ops sharding failures visible and unroutable ids a 404 Four failure modes from a manual pass over a multi-database run-ops setup, three of them silent: - A replication source whose publication carries no tables replicates nothing while boot succeeds and the service looks healthy. The client now raises a typed PublicationMisconfiguredError and the webapp counts it per source, so it alarms. - Completing a wait token with an id naming a database the deployment has no store for answered 500. Every API route builder now answers 404, matching the run routes. - An unparseable stored shard set still degrades to the previous id format, but the parse failure is now reported instead of vanishing, on both the read and write paths. - Per-shard observability, so a gradual rollout can be watched: runops_shard_routed_total and runops_read_through_source_total. --- ...ens.$waitpointFriendlyId.callback.$hash.ts | 14 +++ ...ts.tokens.$waitpointFriendlyId.complete.ts | 14 +++ .../routeBuilders/apiBuilder.server.ts | 12 ++ .../runsReplicationInstance.server.ts | 4 + .../services/runsReplicationMetrics.server.ts | 38 +++++++ .../services/runsReplicationService.server.ts | 7 ++ .../mintShardAssignment.test.ts | 33 ++++++ .../v3/runOpsMigration/mintShardAssignment.ts | 6 +- .../app/v3/runOpsMigration/mintShardGrace.ts | 37 ++++-- .../readThrough.server.test.ts | 66 +++++++++++ .../v3/runOpsMigration/readThrough.server.ts | 23 ++-- .../readThroughSourceMetric.server.test.ts | 35 ++++++ .../readThroughSourceMetric.server.ts | 42 +++++++ .../runOpsMigration/runOpsMintShard.server.ts | 22 ++++ apps/webapp/app/v3/runStore.server.ts | 19 ++++ .../test/runsReplicationMetrics.test.ts | 44 ++++++++ ...eplicationPublicationMisconfigured.test.ts | 86 ++++++++++++++ .../test/waitpointTokenUnroutableId.test.ts | 106 ++++++++++++++++++ .../client.publicationMisconfigured.test.ts | 44 ++++++++ internal-packages/replication/src/client.ts | 19 +++- internal-packages/replication/src/errors.ts | 18 +++ .../run-store/src/routingStoreMetrics.ts | 8 ++ .../src/runOpsStore.newMethods.test.ts | 1 + .../src/runOpsStore.shardMap.test.ts | 60 +++++++++- .../run-store/src/runOpsStore.ts | 12 +- 25 files changed, 743 insertions(+), 27 deletions(-) create mode 100644 apps/webapp/app/services/runsReplicationMetrics.server.ts create mode 100644 apps/webapp/app/v3/runOpsMigration/readThroughSourceMetric.server.test.ts create mode 100644 apps/webapp/app/v3/runOpsMigration/readThroughSourceMetric.server.ts create mode 100644 apps/webapp/test/runsReplicationMetrics.test.ts create mode 100644 apps/webapp/test/runsReplicationPublicationMisconfigured.test.ts create mode 100644 apps/webapp/test/waitpointTokenUnroutableId.test.ts create mode 100644 internal-packages/replication/src/client.publicationMisconfigured.test.ts diff --git a/apps/webapp/app/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.callback.$hash.ts b/apps/webapp/app/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.callback.$hash.ts index 431a7eb2582..39281bde67e 100644 --- a/apps/webapp/app/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.callback.$hash.ts +++ b/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 }); } diff --git a/apps/webapp/app/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.complete.ts b/apps/webapp/app/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.complete.ts index 950b36e873b..aff5045fbd2 100644 --- a/apps/webapp/app/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.complete.ts +++ b/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 diff --git a/apps/webapp/app/services/routeBuilders/apiBuilder.server.ts b/apps/webapp/app/services/routeBuilders/apiBuilder.server.ts index 1f0004dda1a..39e1dcef678 100644 --- a/apps/webapp/app/services/routeBuilders/apiBuilder.server.ts +++ b/apps/webapp/app/services/routeBuilders/apiBuilder.server.ts @@ -1785,6 +1785,12 @@ export function createLoaderWorkerApiRoute< return error; } + const unroutable = unroutableIdResponse(error); + if (unroutable) { + logBoundaryError("Unroutable id", error, request.url); + return unroutable; + } + logBoundaryError("Error in loader", error, request.url); return json({ error: "Internal Server Error" }, { status: 500 }); @@ -1954,6 +1960,12 @@ export function createActionWorkerApiRoute< return json({ error: error.message }, { status: error.status ?? 422 }); } + const unroutable = unroutableIdResponse(error); + if (unroutable) { + logBoundaryError("Unroutable id", error, request.url); + return unroutable; + } + logBoundaryError("Error in action", error, request.url); return json({ error: "Internal Server Error" }, { status: 500 }); diff --git a/apps/webapp/app/services/runsReplicationInstance.server.ts b/apps/webapp/app/services/runsReplicationInstance.server.ts index 7c074b1586b..f2bb8de1071 100644 --- a/apps/webapp/app/services/runsReplicationInstance.server.ts +++ b/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 diff --git a/apps/webapp/app/services/runsReplicationMetrics.server.ts b/apps/webapp/app/services/runsReplicationMetrics.server.ts new file mode 100644 index 00000000000..d8aa3dddf97 --- /dev/null +++ b/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 +): 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) +); diff --git a/apps/webapp/app/services/runsReplicationService.server.ts b/apps/webapp/app/services/runsReplicationService.server.ts index ace3029abf1..4047e831603 100644 --- a/apps/webapp/app/services/runsReplicationService.server.ts +++ b/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", () => { diff --git a/apps/webapp/app/v3/runOpsMigration/mintShardAssignment.test.ts b/apps/webapp/app/v3/runOpsMigration/mintShardAssignment.test.ts index a4e4a64d36d..16341328a31 100644 --- a/apps/webapp/app/v3/runOpsMigration/mintShardAssignment.test.ts +++ b/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"); diff --git a/apps/webapp/app/v3/runOpsMigration/mintShardAssignment.ts b/apps/webapp/app/v3/runOpsMigration/mintShardAssignment.ts index 2855f936249..1687ff76c65 100644 --- a/apps/webapp/app/v3/runOpsMigration/mintShardAssignment.ts +++ b/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 void; + +function readStoredCsv(value: unknown, key: string, onInvalid?: OnInvalidShardSet): string[] { if (typeof value !== "string") return []; try { return parseShardCsv(value); - } catch { + } catch (error) { + onInvalid?.({ key, value, error }); return []; } } @@ -84,7 +94,8 @@ 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 | null | undefined + flags: Record | null | undefined, + onInvalid?: OnInvalidShardSet ): MintShardSetResolution { const source = flags ?? {}; const flippedAtRaw = source[SET_FLIPPED_AT_KEY]; @@ -92,8 +103,11 @@ export function readMintShardSetResolution( 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, }; } @@ -105,7 +119,8 @@ export function stampMintShardSetFlip( existingFlags: Record | null | undefined, outgoingFlags: Record, nowMs: number, - graceMs: number + graceMs: number, + onInvalid?: OnInvalidShardSet ): Record { // Only act when the save actually SETS the list. Omitting it must not inject a default. if (typeof outgoingFlags[SET_KEY] !== "string") { @@ -113,11 +128,15 @@ 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, onInvalid); + const storedSet = readStoredCsv(existing[SET_KEY], SET_KEY, onInvalid); if (outgoingSet.join(",") !== storedSet.join(",")) { - const effective = effectiveMintShardSet(readMintShardSetResolution(existing), nowMs, graceMs); + const effective = effectiveMintShardSet( + readMintShardSetResolution(existing, onInvalid), + nowMs, + graceMs + ); outgoingFlags[SET_PREV_KEY] = effective.join(","); outgoingFlags[SET_FLIPPED_AT_KEY] = new Date(nowMs).toISOString(); return outgoingFlags; diff --git a/apps/webapp/app/v3/runOpsMigration/readThrough.server.test.ts b/apps/webapp/app/v3/runOpsMigration/readThrough.server.test.ts index 8a657060ef9..e644ea91fa2 100644 --- a/apps/webapp/app/v3/runOpsMigration/readThrough.server.test.ts +++ b/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([]); + } + ); }); diff --git a/apps/webapp/app/v3/runOpsMigration/readThrough.server.ts b/apps/webapp/app/v3/runOpsMigration/readThrough.server.ts index 6e1beaae62c..de37b225b85 100644 --- a/apps/webapp/app/v3/runOpsMigration/readThrough.server.ts +++ b/apps/webapp/app/v3/runOpsMigration/readThrough.server.ts @@ -25,11 +25,12 @@ import { import { logger as defaultLogger } from "~/services/logger.server"; import { resolveShard, type ShardKey } from "@trigger.dev/core/v3/isomorphic"; import { isSplitEnabled } from "./splitMode.server"; +import { recordReadThroughSource } from "./readThroughSourceMetric.server"; import { runOpsShardReplicas } from "./shardHandles.server"; type ShardSource = `shard:${string}`; -type ReadThroughSource = "new" | "legacy-replica" | ShardSource; +export type ReadThroughSource = "new" | "legacy-replica" | ShardSource; /** * `found` carries hit/miss STRUCTURALLY. `source` is open-ended once shards exist, so a @@ -56,6 +57,8 @@ type ReadThroughDeps = { logger?: { error: (m: string, meta?: Record) => void }; /** Saturation-signal emit hook: called on each legacy-replica hit. */ onLegacyReplicaRead?: (id: string) => void; + /** Per-source emit hook: called with the store that served a HIT. Defaults to the metric. */ + onSource?: (source: ReadThroughSource) => void; }; type ReadThroughRunInput = { @@ -67,7 +70,12 @@ type ReadThroughRunInput = { deps?: ReadThroughDeps; }; -function hit(source: ReadThroughSource, value: T): ReadThroughResult { +function hit( + source: ReadThroughSource, + value: T, + onSource: (source: ReadThroughSource) => void +): ReadThroughResult { + onSource(source); return { found: true, source, value }; } @@ -83,13 +91,14 @@ export async function readThroughRun( const legacyReplica = deps?.legacyReplica ?? defaultLegacyReplica; const shardReplicas = deps?.shardReplicas ?? runOpsShardReplicas; const logger = deps?.logger ?? defaultLogger; + const onSource = deps?.onSource ?? recordReadThroughSource; const splitEnabled = deps?.splitEnabled ?? (await isSplitEnabled()); // Passthrough: single plain read against the one collapsed store. if (!splitEnabled) { const v = await input.readNew(newClient); - return v != null ? hit("new", v) : miss("not-found"); + return v != null ? hit("new", v, onSource) : miss("not-found"); } // Total: an unclassifiable id resolves to "legacy" (probe rather than drop a real run). @@ -111,18 +120,18 @@ export async function readThroughRun( } // A gen-2 shard is a dedicated-schema store, exactly like `new`, so `readNew` fits. const v = await input.readNew(shardReplica); - return v != null ? hit(`shard:${shardKey}`, v) : miss("not-found"); + return v != null ? hit(`shard:${shardKey}`, v, onSource) : miss("not-found"); } if (shardKey === "new") { const v = await input.readNew(newClient); - return v != null ? hit("new", v) : miss("not-found"); + return v != null ? hit("new", v, onSource) : miss("not-found"); } if (idKind === "waitpoint") { const v = await input.readNew(newClient); if (v != null) { - return hit("new", v); + return hit("new", v, onSource); } } @@ -130,7 +139,7 @@ export async function readThroughRun( const lv = await input.readLegacy(legacyReplica); if (lv != null) { deps?.onLegacyReplicaRead?.(id); - return hit("legacy-replica", lv); + return hit("legacy-replica", lv, onSource); } if (deps?.isPastRetention?.(id)) { diff --git a/apps/webapp/app/v3/runOpsMigration/readThroughSourceMetric.server.test.ts b/apps/webapp/app/v3/runOpsMigration/readThroughSourceMetric.server.test.ts new file mode 100644 index 00000000000..920e2f8ca87 --- /dev/null +++ b/apps/webapp/app/v3/runOpsMigration/readThroughSourceMetric.server.test.ts @@ -0,0 +1,35 @@ +// The per-shard half of read-through observability: /metrics carried no `shard` label at all, so a +// cohort ramp was unobservable. A gen-2 source splits into a constant `source` plus the shard key, +// so one query sums the whole gen-2 cohort and another breaks it down per shard. +import { Registry, type RegistryContentType } from "prom-client"; +import { describe, expect, it } from "vitest"; +import { buildReadThroughSourceMetric } from "./readThroughSourceMetric.server"; + +describe("read-through source metric", () => { + it("labels a gen-2 hit with its shard key", async () => { + const register = new Registry(); + const record = buildReadThroughSourceMetric(register); + + record("shard:a"); + record("shard:a"); + record("shard:b"); + + const exposed = await register.metrics(); + expect(exposed).toContain('runops_read_through_source_total{source="shard",shard="a"} 2'); + expect(exposed).toContain('runops_read_through_source_total{source="shard",shard="b"} 1'); + }); + + it("keeps the two gen-1 sources distinguishable and shard-less", async () => { + const register = new Registry(); + const record = buildReadThroughSourceMetric(register); + + record("new"); + record("legacy-replica"); + + const exposed = await register.metrics(); + expect(exposed).toContain('runops_read_through_source_total{source="new",shard="none"} 1'); + expect(exposed).toContain( + 'runops_read_through_source_total{source="legacy-replica",shard="none"} 1' + ); + }); +}); diff --git a/apps/webapp/app/v3/runOpsMigration/readThroughSourceMetric.server.ts b/apps/webapp/app/v3/runOpsMigration/readThroughSourceMetric.server.ts new file mode 100644 index 00000000000..136049c35ce --- /dev/null +++ b/apps/webapp/app/v3/runOpsMigration/readThroughSourceMetric.server.ts @@ -0,0 +1,42 @@ +/** + * Which store actually served a read-through read: `ReadThroughSource` was an internal return type + * only, so a cohort ramp was invisible from outside the process. On the hot path, so each label + * child is resolved once and cached — `inc({ source, shard })` hashes a fresh object per call. + */ +import { Counter, type Registry, type RegistryContentType } from "prom-client"; +import { metricsRegister } from "~/metrics.server"; +import { singleton } from "~/utils/singleton"; +import type { ReadThroughSource } from "./readThrough.server"; + +const SHARD_PREFIX = "shard:"; + +export function buildReadThroughSourceMetric( + register: Registry +): (source: ReadThroughSource) => void { + const counter = new Counter({ + name: "runops_read_through_source_total", + help: "Read-through reads served, by the store that served them. `shard` carries the gen-2 shard key.", + labelNames: ["source", "shard"], + registers: [register], + }); + + const children = new Map void }>(); + + return (source) => { + let child = children.get(source); + if (child === undefined) { + // A gen-2 source is split into a constant `source` and the shard key, so one query sums the + // whole gen-2 cohort and another breaks it down per shard. + child = source.startsWith(SHARD_PREFIX) + ? counter.labels("shard", source.slice(SHARD_PREFIX.length)) + : counter.labels(source, "none"); + children.set(source, child); + } + child.inc(); + }; +} + +// singleton: module-scope Counter registration double-registers under dev HMR. +export const recordReadThroughSource = singleton("readThroughSourceMetric", () => + buildReadThroughSourceMetric(metricsRegister) +); diff --git a/apps/webapp/app/v3/runOpsMigration/runOpsMintShard.server.ts b/apps/webapp/app/v3/runOpsMigration/runOpsMintShard.server.ts index 360bc9fd863..5b3435b48b1 100644 --- a/apps/webapp/app/v3/runOpsMigration/runOpsMintShard.server.ts +++ b/apps/webapp/app/v3/runOpsMigration/runOpsMintShard.server.ts @@ -66,6 +66,27 @@ function reportOverrideRejected(info: { override: string; activeSet: string[] }) logger.error("[runOpsMintShard] override shard is not in the active set; ignoring it", info); } +// Keyed by the offending value, like the override report: one bad stored list applies to the whole +// fleet. Bounded and TTL'd because the refresh runs once per cache TTL per process, which would +// otherwise repeat this line for as long as the value stays broken. +const reportedSetParseFailures = singleton( + "runOpsMintShardReportedSetParseFailures", + () => new BoundedTtlCache(REPORT_TTL_MS, REPORT_MAX_ENTRIES) +); + +// The stored list degrades to empty on a parse failure, which is the correct fail-safe but reverts +// the whole fleet to gen-1 minting. `shard-set read failed` never covers this: the read SUCCEEDED. +function reportSetParseFailed(failure: { key: string; value: string; error: unknown }): void { + const cacheKey = `${failure.key}:${failure.value}`; + if (reportedSetParseFailures.get(cacheKey) !== undefined) return; + reportedSetParseFailures.set(cacheKey, true); + logger.error("[runOpsMintShard] stored shard set is unparseable; minting gen-1 (fail-safe)", { + key: failure.key, + value: failure.value, + error: failure.error instanceof Error ? failure.error.message : failure.error, + }); +} + /** * Which shard an environment mints new roots into. Call only after resolveRunIdMintKind has * returned "runOpsId". Returns "new" to mean a gen-1 run-ops id, which is today's behaviour. @@ -94,5 +115,6 @@ export async function resolveMintShard(environment: { onOverrideRejected: reportOverrideRejected, onReadFailed: (error) => logger.error("[runOpsMintShard] shard-set read failed; minting gen-1 (fail-safe)", { error }), + onSetParseFailed: reportSetParseFailed, }); } diff --git a/apps/webapp/app/v3/runStore.server.ts b/apps/webapp/app/v3/runStore.server.ts index 7f51a940cec..6ab9675d31c 100644 --- a/apps/webapp/app/v3/runStore.server.ts +++ b/apps/webapp/app/v3/runStore.server.ts @@ -135,9 +135,28 @@ const routingStoreMetrics: RoutingStoreMetrics = singleton("routingStoreMetrics" labelNames: ["from", "to"], registers: [metricsRegister], }); + // The only per-shard series /metrics carries, so a cohort ramp is visible per shard while it + // happens. On every routed operation: the label child is resolved once per shard and cached, + // because `inc({ shard })` hashes a fresh label object on each call. Cardinality is bounded by + // the shard alphabet plus the two reserved keys. + const shardRouted = new Counter({ + name: "runops_shard_routed_total", + help: "Run-ops store resolutions keyed by id. Fan-outs and probe hits resolve no single shard and are not counted, so read this as relative ramp signal rather than served requests.", + labelNames: ["shard"], + registers: [metricsRegister], + }); + const shardRoutedChildren = new Map void }>(); return { recordDuplicateId: (shardKeys) => duplicateId.inc({ shard_keys: shardKeys.join(",") }), recordWaitpointProbeFallback: (from, to) => probeFallback.inc({ from, to }), + recordShardRouted: (shardKey) => { + let child = shardRoutedChildren.get(shardKey); + if (child === undefined) { + child = shardRouted.labels(shardKey); + shardRoutedChildren.set(shardKey, child); + } + child.inc(); + }, }; }); diff --git a/apps/webapp/test/runsReplicationMetrics.test.ts b/apps/webapp/test/runsReplicationMetrics.test.ts new file mode 100644 index 00000000000..e6f8bf43f97 --- /dev/null +++ b/apps/webapp/test/runsReplicationMetrics.test.ts @@ -0,0 +1,44 @@ +// The alarmable half of the "publication carries no tables" failure: the client reports it, and +// this counter is what turns that report into a series an alert can fire on, per source. +import { PublicationMisconfiguredError } from "@internal/replication"; +import { Registry, type RegistryContentType } from "prom-client"; +import { describe, expect, it } from "vitest"; +import { buildRunsReplicationSourceMetrics } from "~/services/runsReplicationMetrics.server"; + +function freshRegister() { + return new Registry(); +} + +function misconfigured(publicationName: string) { + return new PublicationMisconfiguredError( + `Publication '${publicationName}' exists but has NO TABLES configured.`, + { publicationName, table: "TaskRun" } + ); +} + +describe("runs-replication source metrics", () => { + it("counts a publication misconfiguration against the source that reported it", async () => { + const register = freshRegister(); + const metrics = buildRunsReplicationSourceMetrics(register); + + metrics.recordSourceError({ sourceId: "shard-a", error: misconfigured("runs_shard_a_pub") }); + metrics.recordSourceError({ sourceId: "shard-a", error: misconfigured("runs_shard_a_pub") }); + metrics.recordSourceError({ sourceId: "new", error: misconfigured("runs_new_pub") }); + + const exposed = await register.metrics(); + expect(exposed).toContain( + 'runs_replication_publication_misconfigured_total{source="shard-a"} 2' + ); + expect(exposed).toContain('runs_replication_publication_misconfigured_total{source="new"} 1'); + }); + + it("leaves the counter alone for any other client error", async () => { + const register = freshRegister(); + const metrics = buildRunsReplicationSourceMetrics(register); + + metrics.recordSourceError({ sourceId: "legacy", error: new Error("connection terminated") }); + + const exposed = await register.metrics(); + expect(exposed).not.toContain('source="legacy"'); + }); +}); diff --git a/apps/webapp/test/runsReplicationPublicationMisconfigured.test.ts b/apps/webapp/test/runsReplicationPublicationMisconfigured.test.ts new file mode 100644 index 00000000000..cf33af14a43 --- /dev/null +++ b/apps/webapp/test/runsReplicationPublicationMisconfigured.test.ts @@ -0,0 +1,86 @@ +// End-to-end for the failure a configured-but-empty publication caused in production: the source +// replicated nothing, boot passed (a source WAS configured), and the only symptom was a log line +// every 30s. This drives the real service against a real publication with no tables and asserts the +// whole chain now produces a number: client -> onSourceError -> the /metrics counter. +import { ClickHouse } from "@internal/clickhouse"; +import { PublicationMisconfiguredError } from "@internal/replication"; +import { replicationContainerTest } from "@internal/testcontainers"; +import { setTimeout } from "node:timers/promises"; +import { Registry, type RegistryContentType } from "prom-client"; +import { buildRunsReplicationSourceMetrics } from "~/services/runsReplicationMetrics.server"; +import { RunsReplicationService } from "~/services/runsReplicationService.server"; +import { TestReplicationClickhouseFactory } from "./utils/testReplicationClickhouseFactory"; + +vi.setConfig({ testTimeout: 90_000 }); + +describe("RunsReplicationService — a source whose publication carries no tables", () => { + replicationContainerTest( + "reports the source error, which lands on the alarmable counter", + async ({ clickhouseContainer, redisOptions, postgresContainer, prisma }) => { + // The production shape: the publication exists, so the client adopts it rather than creating + // one, and it carries no tables — so this source's WAL never reaches ClickHouse. + await prisma.$executeRawUnsafe(`CREATE PUBLICATION empty_pub;`); + + const clickhouse = new ClickHouse({ + url: clickhouseContainer.getConnectionUrl(), + name: "runs-replication", + logLevel: "warn", + }); + + const register = new Registry(); + const metrics = buildRunsReplicationSourceMetrics(register); + const reported: Array<{ sourceId: string; error: unknown }> = []; + + const service = new RunsReplicationService({ + clickhouseFactory: new TestReplicationClickhouseFactory(clickhouse), + pgConnectionUrl: postgresContainer.getConnectionUri(), + serviceName: "runs-replication", + slotName: "empty_pub_slot", + publicationName: "empty_pub", + redisOptions, + flushIntervalMs: 100, + flushBatchSize: 1, + leaderLockTimeoutMs: 5000, + leaderLockExtendIntervalMs: 1000, + logLevel: "error", + sources: [ + { + id: "shard-a", + pgConnectionUrl: postgresContainer.getConnectionUri(), + slotName: "empty_pub_slot", + publicationName: "empty_pub", + originGeneration: 2, + }, + ], + onSourceError: (info) => { + reported.push(info); + metrics.recordSourceError(info); + }, + }); + + try { + await service.start(); + + const deadline = Date.now() + 20_000; + while ( + !reported.some((r) => r.error instanceof PublicationMisconfiguredError) && + Date.now() < deadline + ) { + await setTimeout(250); + } + } finally { + await service.shutdown(); + } + + const misconfigured = reported.filter( + (r) => r.error instanceof PublicationMisconfiguredError + ); + expect(misconfigured.length).toBeGreaterThan(0); + expect(misconfigured[0]?.sourceId).toBe("shard-a"); + + expect(await register.metrics()).toContain( + 'runs_replication_publication_misconfigured_total{source="shard-a"}' + ); + } + ); +}); diff --git a/apps/webapp/test/waitpointTokenUnroutableId.test.ts b/apps/webapp/test/waitpointTokenUnroutableId.test.ts new file mode 100644 index 00000000000..a34dda4d641 --- /dev/null +++ b/apps/webapp/test/waitpointTokenUnroutableId.test.ts @@ -0,0 +1,106 @@ +// `resolveShard` is pure id-shape, so any 24-char base32hex core plus `[a-z0-9]` plus "2" parses as +// a gen-2 id naming a shard — including a shard no topology configures. The run routes already +// answer 404 for that, but the waitpoint-token routes wrap their body in a catch that turns every +// non-Response error into a 500, so a caller could induce a 5xx (and trip a canary) at will. +// +// Both routes are driven for real: the routing store is a REAL RoutingRunStore (it is the thing that +// throws UnknownShardKey, from its own id-shape routing — the sub-stores are never reached), and the +// route code under test is the exported action / the handler the api-builder captured. +import { describe, expect, vi } from "vitest"; + +const H = vi.hoisted(() => ({ + handlers: [] as Array<{ config: any; handler: any }>, + store: undefined as any, +})); + +vi.mock("~/v3/runStore.server", () => ({ + runStore: new Proxy( + {}, + { + get(_t, prop) { + const store = H.store; + if (!store) throw new Error("test bug: H.store not initialised before handler ran"); + const value = store[prop]; + return typeof value === "function" ? value.bind(store) : value; + }, + } + ), +})); + +vi.mock("~/services/routeBuilders/apiBuilder.server", () => ({ + anyResource: (x: unknown) => x, + createActionApiRoute: (config: any, handler: any) => { + H.handlers.push({ config, handler }); + return { action: vi.fn(), loader: vi.fn() }; + }, +})); + +import { RoutingRunStore, type RunStore } from "@internal/run-store"; +import { WaitpointId } from "@trigger.dev/core/v3/isomorphic"; +import { action as callbackAction } from "~/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.callback.$hash"; + +// 24-char base32hex core + shard char "z" + version "2": a well-formed gen-2 id whose shard the +// router below has no store for. +const UNCONFIGURED_SHARD_WAITPOINT_ID = "waitpoint_" + "c".repeat(24) + "z2"; + +// Only "new" and "legacy" are configured, so a gen-2 id lands on UnknownShardKey inside the router. +// The sub-stores are unreachable placeholders: the throw happens before any of them is consulted. +function routerWithNoShards() { + const unreachable = new Proxy( + {}, + { + get() { + throw new Error("no sub-store should be reached for an unconfigured shard key"); + }, + } + ) as RunStore; + return new RoutingRunStore({ new: unreachable, legacy: unreachable }); +} + +async function completeHandler() { + await import("~/routes/api.v1.waitpoints.tokens.$waitpointFriendlyId.complete"); + const entry = H.handlers.find((h) => Boolean(h.config?.params?.shape?.waitpointFriendlyId)); + if (!entry) throw new Error("complete-route handler was not captured"); + return entry.handler as (args: any) => Promise; +} + +describe("waitpoint-token routes with an id naming an unconfigured shard", () => { + it("answers 404, not 500, on the token completion route", async () => { + H.store = routerWithNoShards(); + const handler = await completeHandler(); + + const thrown = await handler({ + authentication: { environment: { id: "env_1" } }, + body: { data: { ok: true } }, + params: { waitpointFriendlyId: WaitpointId.toFriendlyId(UNCONFIGURED_SHARD_WAITPOINT_ID) }, + }).then( + (response) => response, + (error) => error + ); + + expect(thrown).toBeInstanceOf(Response); + expect((thrown as Response).status).toBe(404); + // Not retryable: no number of retries makes a topology grow a store. + expect((thrown as Response).headers.get("x-should-retry")).toBe("false"); + }); + + it("answers 404, not 500, on the HTTP-callback route", async () => { + H.store = routerWithNoShards(); + + const payload = JSON.stringify({ ok: true }); + const response = await callbackAction({ + request: new Request("http://localhost/callback", { + method: "POST", + headers: { "content-type": "application/json", "content-length": String(payload.length) }, + body: payload, + }), + params: { + waitpointFriendlyId: WaitpointId.toFriendlyId(UNCONFIGURED_SHARD_WAITPOINT_ID), + hash: "whatever", + }, + context: {}, + } as any); + + expect(response.status).toBe(404); + }); +}); diff --git a/internal-packages/replication/src/client.publicationMisconfigured.test.ts b/internal-packages/replication/src/client.publicationMisconfigured.test.ts new file mode 100644 index 00000000000..0eeb5e6f0e6 --- /dev/null +++ b/internal-packages/replication/src/client.publicationMisconfigured.test.ts @@ -0,0 +1,44 @@ +// A publication that exists but carries no tables is the failure that replicated NOTHING in +// production while every log line said "healthy" and boot passed: the source IS configured, so +// the boot-time coverage assert cannot see it, and the client just retries every 30s. +// +// The client already detects it. What was missing is a way for the owner to COUNT it, which needs +// the emitted error to be distinguishable from any other client error. This drives the real +// client against a real publication with no tables and pins that type. +import { postgresAndRedisTest } from "@internal/testcontainers"; +import { LogicalReplicationClient } from "./client.js"; +import { PublicationMisconfiguredError } from "./errors.js"; + +describe("publication with no tables", () => { + postgresAndRedisTest( + "emits a typed PublicationMisconfiguredError rather than a bare client error", + async ({ postgresContainer, prisma, redisOptions }) => { + // The exact production shape: the publication exists, so the client adopts it instead of + // creating one, and validation finds it carries no tables at all. + await prisma.$executeRawUnsafe(`CREATE PUBLICATION no_tables_pub;`); + + const client = new LogicalReplicationClient({ + name: "no-tables", + slotName: "no_tables_slot", + publicationName: "no_tables_pub", + redisOptions, + table: "TaskRun", + pgConfig: { connectionString: postgresContainer.getConnectionUri() }, + }); + + const errors: unknown[] = []; + client.events.on("error", (error) => errors.push(error)); + + await client.subscribe(); + await client.shutdown(); + + const misconfigured = errors.filter((e) => e instanceof PublicationMisconfiguredError); + expect(misconfigured).toHaveLength(1); + + const error = misconfigured[0] as PublicationMisconfiguredError; + expect(error.publicationName).toBe("no_tables_pub"); + expect(error.table).toBe("TaskRun"); + expect(error.message).toContain("NO TABLES configured"); + } + ); +}); diff --git a/internal-packages/replication/src/client.ts b/internal-packages/replication/src/client.ts index 1a128a41e80..a41d9b0c7d4 100644 --- a/internal-packages/replication/src/client.ts +++ b/internal-packages/replication/src/client.ts @@ -5,7 +5,7 @@ import { tryCatch } from "@trigger.dev/core/utils"; import EventEmitter from "node:events"; import { type ClientConfig, type Connection, Client } from "pg"; import Redlock, { type Lock } from "redlock"; -import { LogicalReplicationClientError } from "./errors.js"; +import { LogicalReplicationClientError, PublicationMisconfiguredError } from "./errors.js"; import { type PgoutputMessage, getPgoutputStartReplicationSQL, @@ -609,7 +609,13 @@ export class LogicalReplicationClient { error: validationError, }); - this.events.emit("error", new LogicalReplicationClientError(validationError)); + this.events.emit( + "error", + new PublicationMisconfiguredError(validationError, { + publicationName: this.options.publicationName, + table: this.options.table, + }) + ); return false; } @@ -672,7 +678,14 @@ export class LogicalReplicationClient { async #validatePublicationConfiguration(): Promise { if (!this.client) { - return "Cannot validate publication configuration: client not connected"; + // Not a misconfiguration: a disconnected client cannot be inspected, and reporting it as + // one would count a connectivity fault on the publication series. Unreachable today, since + // both callers guard on `this.client` first. + this.logger.debug("Skipping publication validation: client not connected", { + name: this.options.name, + publicationName: this.options.publicationName, + }); + return null; } // Check if the publication has the correct table diff --git a/internal-packages/replication/src/errors.ts b/internal-packages/replication/src/errors.ts index 0521305a5e7..bd702c2bb25 100644 --- a/internal-packages/replication/src/errors.ts +++ b/internal-packages/replication/src/errors.ts @@ -3,3 +3,21 @@ export class LogicalReplicationClientError extends Error { super(message); } } + +/** + * The publication a source subscribes through exists but does not carry the table we replicate — + * commonly because it was created with no tables at all. Nothing throws and nothing stops: the + * client retries and logs, while that source replicates NOTHING, so every ClickHouse-fronted + * aggregate silently under-counts. Typed so a consumer can put a number on it and alarm. + */ +export class PublicationMisconfiguredError extends LogicalReplicationClientError { + readonly publicationName: string; + readonly table: string; + + constructor(message: string, options: { publicationName: string; table: string }) { + super(message); + this.name = "PublicationMisconfiguredError"; + this.publicationName = options.publicationName; + this.table = options.table; + } +} diff --git a/internal-packages/run-store/src/routingStoreMetrics.ts b/internal-packages/run-store/src/routingStoreMetrics.ts index d4aa454d903..85fe89f6402 100644 --- a/internal-packages/run-store/src/routingStoreMetrics.ts +++ b/internal-packages/run-store/src/routingStoreMetrics.ts @@ -4,13 +4,21 @@ * * runops_shard_duplicate_id_total — one id returned by two shards that should be disjoint * runops_waitpoint_probe_fallback_total — a waitpoint was not on the store its id named + * runops_shard_routed_total — an operation was routed to a shard, labelled by key */ export type RoutingStoreMetrics = { recordDuplicateId(shardKeys: string[]): void; recordWaitpointProbeFallback(from: string, to: string): void; + /** + * Every routed operation, keyed by the shard it landed on — the only per-shard series the + * router emits, and what makes a cohort ramp visible while it is happening rather than after. + * On the hot path, so an implementation must not build a label object per call. + */ + recordShardRouted(shardKey: string): void; }; export const noopRoutingStoreMetrics: RoutingStoreMetrics = { recordDuplicateId() {}, recordWaitpointProbeFallback() {}, + recordShardRouted() {}, }; diff --git a/internal-packages/run-store/src/runOpsStore.newMethods.test.ts b/internal-packages/run-store/src/runOpsStore.newMethods.test.ts index 8a56057a117..d7da3c76fc3 100644 --- a/internal-packages/run-store/src/runOpsStore.newMethods.test.ts +++ b/internal-packages/run-store/src/runOpsStore.newMethods.test.ts @@ -229,6 +229,7 @@ for (const topology of TOPOLOGIES) { const { router } = buildRouter(topology, { recordDuplicateId: (keys) => seen.push(keys), recordWaitpointProbeFallback() {}, + recordShardRouted() {}, }); await router.findManyWaitpointTags({ where: { environmentId: "env" } }); // A duplicate confined to {legacy, new} is the known drain-mirror case and stays silent. Once a diff --git a/internal-packages/run-store/src/runOpsStore.shardMap.test.ts b/internal-packages/run-store/src/runOpsStore.shardMap.test.ts index aff0d50764e..b107fec24fa 100644 --- a/internal-packages/run-store/src/runOpsStore.shardMap.test.ts +++ b/internal-packages/run-store/src/runOpsStore.shardMap.test.ts @@ -349,7 +349,10 @@ describe("RoutingRunStore id-to-shard-key seam", () => { }); }); -function buildNShardRouter(shardKeys: string[], opts: { aliasOf?: Record } = {}) { +function buildNShardRouter( + shardKeys: string[], + opts: { aliasOf?: Record; routed?: string[] } = {} +) { const log: Call[] = []; const newStore = fakeStore("new", log); const legacyStore = fakeStore("legacy", log); @@ -365,6 +368,15 @@ function buildNShardRouter(shardKeys: string[], opts: { aliasOf?: Record id.split(":")[0]!, + ...(opts.routed + ? { + metrics: { + recordDuplicateId() {}, + recordWaitpointProbeFallback() {}, + recordShardRouted: (key: string) => opts.routed!.push(key), + }, + } + : {}), }); return { router, log, byKey }; } @@ -384,6 +396,28 @@ describe("RoutingRunStore #distinctStores — one entry per database", () => { expect(trace(log)).toEqual(["a:findRun"]); }); + // Per-shard routing was invisible from outside the process: nothing the router emitted carried + // the shard it routed to, so a cohort ramp could only be inferred from the databases themselves. + it("counts every routed operation against the shard it landed on", async () => { + const routed: string[] = []; + const { router } = buildNShardRouter(["a", "b"], { routed }); + + await router.findRun({ id: "a:run_1" }); + await router.findRun({ id: "b:run_1" }); + await router.findRun({ id: "a:run_2" }); + + expect(routed).toEqual(["a", "b", "a"]); + }); + + it("does not count anything before an operation is routed", async () => { + const routed: string[] = []; + buildNShardRouter(["a", "b"], { routed }); + + // Constructing the router touches every configured store to build #distinctStores; that is + // not traffic, and counting it would put a boot-time step in a per-shard traffic series. + expect(routed).toEqual([]); + }); + it("counts an aliased shard's database ONCE in a sum", async () => { // "a" aliases "new": two keys, one database. const { router, log } = buildNShardRouter(["a"], { aliasOf: { a: "new" } }); @@ -555,6 +589,7 @@ describe("RoutingRunStore merge precedence and duplicate alarm", () => { metrics: { recordDuplicateId: (k: string[]) => seen.push(k), recordWaitpointProbeFallback() {}, + recordShardRouted() {}, }, seen, }; @@ -670,7 +705,15 @@ describe("RoutingRunStore countPendingWaitpoints — disjoint-sum partition", () { key: "b", store: mk("b") }, ], resolveShard: (id: string) => (id.includes(":") ? id.split(":")[0]! : "legacy"), - ...(spy ? { metrics: { recordDuplicateId: spy, recordWaitpointProbeFallback() {} } } : {}), + ...(spy + ? { + metrics: { + recordDuplicateId: spy, + recordWaitpointProbeFallback() {}, + recordShardRouted() {}, + }, + } + : {}), }); return { router, log }; } @@ -791,6 +834,7 @@ describe("RoutingRunStore waitpoint probes at N", () => { metrics: { recordDuplicateId() {}, recordWaitpointProbeFallback: (from, to) => falls.push([from, to]), + recordShardRouted() {}, }, }); await router.updateWaitpoint({ where: { id: "cuid_w1" }, data: {} } as never); @@ -910,7 +954,11 @@ describe("RoutingRunStore batch probe tolerates legitimate dual-residency", () = legacy: fakeStore("legacy", log, { batch: { id: "batch_dup", from: "legacy" } }), shards: [{ key: "a", store: fakeStore("a", log, { batch: { id: "batch_dup", from: "a" } }) }], resolveShard: (id: string) => id.split(":")[0]!, - metrics: { recordDuplicateId: (k) => seen.push(k), recordWaitpointProbeFallback() {} }, + metrics: { + recordDuplicateId: (k) => seen.push(k), + recordWaitpointProbeFallback() {}, + recordShardRouted() {}, + }, }); const batch = (await router.findBatchTaskRunById("batch_dup")) as { from: string } | null; // batchTriggerV3 writes raw to the control plane while runEngine routes by id, so this is a @@ -928,7 +976,11 @@ describe("RoutingRunStore batch probe tolerates legitimate dual-residency", () = legacy: fakeStore("legacy", log, { runs: [{ id: "dup", from: "legacy" }] }), shards: [{ key: "a", store: fakeStore("a", log, { runs: [{ id: "dup", from: "a" }] }) }], resolveShard: (id: string) => id.split(":")[0]!, - metrics: { recordDuplicateId: (k) => seen.push(k), recordWaitpointProbeFallback() {} }, + metrics: { + recordDuplicateId: (k) => seen.push(k), + recordWaitpointProbeFallback() {}, + recordShardRouted() {}, + }, }); await router.findRun({ spanId: "span_x" }); expect(seen).toEqual([["legacy", "a"]]); diff --git a/internal-packages/run-store/src/runOpsStore.ts b/internal-packages/run-store/src/runOpsStore.ts index fa923244d06..b50686465e1 100644 --- a/internal-packages/run-store/src/runOpsStore.ts +++ b/internal-packages/run-store/src/runOpsStore.ts @@ -170,11 +170,16 @@ export class RoutingRunStore implements RunStore { // it shares its target's database, and a second leg over one database double-counts a sum. // The discriminator is the DECLARATION, not object identity — the wiring layer may build a // second store object over a shared client. + // Direct map lookup rather than #shardStore: every key here comes from #precedence, so it is + // configured by construction, and #shardStore counts a ROUTED operation — building the list + // must not bump the per-shard counter at boot. + // Assigned before #distinctStores: reintroducing #shardStore below would otherwise throw on + // an unassigned #metrics at construction, in every deployment, at boot. + this.#metrics = options.metrics ?? noopRoutingStoreMetrics; + this.#distinctStores = this.#precedence .filter((key) => !aliasedKeys.has(key)) - .map((key) => ({ key, store: this.#shardStore(key) })); - - this.#metrics = options.metrics ?? noopRoutingStoreMetrics; + .map((key) => ({ key, store: this.#shards.get(key)! })); this.#logger = options.logger ?? new Logger("RoutingRunStore", "warn"); } @@ -204,6 +209,7 @@ export class RoutingRunStore implements RunStore { if (store === undefined) { throw new UnknownShardKey(key, [...this.#shards.keys()]); } + this.#metrics.recordShardRouted(key); return store; } From 9b63809d73b9293a82841e9300405ae164dec721 Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Tue, 1 Sep 2026 18:06:50 +0100 Subject: [PATCH 2/4] fix(webapp): answer not-found for unroutable ids on the remaining run routes The API route builders cover the routes built on them, but routes that handle their own errors never reach that mapping. Four bare API routes answered 500 for an id naming a database the deployment has no store for, and ten dashboard and resource routes threw straight through to an error page. Each now takes the not-found path it already has for a run that does not exist: the API routes return 404, the dashboard routes redirect or 404 as they already did. A shared helper keeps that uniform, and every one logs first, so a shard key dropped from a config meant to be append-only still alarms rather than reading as an absent run. --- apps/webapp/app/routes/@.runs.$runParam.ts | 24 ++-- .../api.v1.batches.$batchParam.results.ts | 9 ++ .../app/routes/api.v1.runs.$runId.tags.ts | 9 ++ .../api.v1.runs.$runParam.reschedule.ts | 9 ++ .../routes/api.v1.runs.$runParam.result.ts | 9 ++ ...g.projects.$projectParam.runs.$runParam.ts | 22 ++-- .../projects.v3.$projectRef.runs.$runParam.ts | 26 ++-- ...ram.realtime.v1.sessions.$sessionId.$io.ts | 9 +- ...am.realtime.v1.streams.$runId.$streamId.ts | 9 +- ...ltime.v1.streams.$runId.input.$streamId.ts | 9 +- .../resources.runs.$runParam.logs.download.ts | 32 ++--- .../app/routes/resources.runs.$runParam.ts | 118 +++++++++--------- .../resources.taskruns.$runParam.debug.ts | 32 ++--- apps/webapp/app/routes/runs.$runParam.ts | 24 ++-- .../runOpsMigration/unroutableRead.server.ts | 21 ++++ 15 files changed, 231 insertions(+), 131 deletions(-) create mode 100644 apps/webapp/app/v3/runOpsMigration/unroutableRead.server.ts diff --git a/apps/webapp/app/routes/@.runs.$runParam.ts b/apps/webapp/app/routes/@.runs.$runParam.ts index d8ff7fd49d1..cb7b34ec26c 100644 --- a/apps/webapp/app/routes/@.runs.$runParam.ts +++ b/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,20 @@ export async function loader({ params, request }: LoaderFunctionArgs) { ); } - const run = await runStore.findRun( - { - friendlyId: runParam, - }, - { - select: { - spanId: true, - runtimeEnvironmentId: true, + const run = await undefinedOnUnroutableId( + runStore.findRun( + { + friendlyId: runParam, + }, + { + select: { + spanId: true, + runtimeEnvironmentId: true, + }, }, - }, - prisma + prisma + ), + { runParam: params.runParam ?? params.runId } ); if (!run) { diff --git a/apps/webapp/app/routes/api.v1.batches.$batchParam.results.ts b/apps/webapp/app/routes/api.v1.batches.$batchParam.results.ts index 39eebc2303e..bc459aa516e 100644 --- a/apps/webapp/app/routes/api.v1.batches.$batchParam.results.ts +++ b/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 }); } diff --git a/apps/webapp/app/routes/api.v1.runs.$runId.tags.ts b/apps/webapp/app/routes/api.v1.runs.$runId.tags.ts index be717efa355..893937d6b4d 100644 --- a/apps/webapp/app/routes/api.v1.runs.$runId.tags.ts +++ b/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 }); } diff --git a/apps/webapp/app/routes/api.v1.runs.$runParam.reschedule.ts b/apps/webapp/app/routes/api.v1.runs.$runParam.reschedule.ts index dbc521a7424..e6b96dc4d31 100644 --- a/apps/webapp/app/routes/api.v1.runs.$runParam.reschedule.ts +++ b/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 }); } diff --git a/apps/webapp/app/routes/api.v1.runs.$runParam.result.ts b/apps/webapp/app/routes/api.v1.runs.$runParam.result.ts index 5df9e496c17..784b0310aff 100644 --- a/apps/webapp/app/routes/api.v1.runs.$runParam.result.ts +++ b/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 }); } diff --git a/apps/webapp/app/routes/orgs.$organizationSlug.projects.$projectParam.runs.$runParam.ts b/apps/webapp/app/routes/orgs.$organizationSlug.projects.$projectParam.runs.$runParam.ts index f2e138861e3..30d2a8bb75a 100644 --- a/apps/webapp/app/routes/orgs.$organizationSlug.projects.$projectParam.runs.$runParam.ts +++ b/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,19 @@ 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) { diff --git a/apps/webapp/app/routes/projects.v3.$projectRef.runs.$runParam.ts b/apps/webapp/app/routes/projects.v3.$projectRef.runs.$runParam.ts index 300caa27ddf..210f5d448eb 100644 --- a/apps/webapp/app/routes/projects.v3.$projectRef.runs.$runParam.ts +++ b/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,21 @@ 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, + const run = await undefinedOnUnroutableId( + runStore.findRun( + { + friendlyId: validatedParams.runParam, }, - }, - prisma + { + select: { + friendlyId: true, + spanId: true, + runtimeEnvironmentId: true, + }, + }, + prisma + ), + { runParam: params.runParam ?? params.runId } ); if (!run) { diff --git a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.sessions.$sessionId.$io.ts b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.sessions.$sessionId.$io.ts index 062b75503a6..b58add47307 100644 --- a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.sessions.$sessionId.$io.ts +++ b/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 }); diff --git a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.$streamId.ts b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.$streamId.ts index 4fbb309e2a8..8373e116202 100644 --- a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.$streamId.ts +++ b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.$streamId.ts @@ -8,6 +8,7 @@ import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server"; import { requireUserId } from "~/services/session.server"; import { EnvironmentParamSchema } from "~/utils/pathBuilder"; import { runStore } from "~/v3/runStore.server"; +import { undefinedOnUnroutableId } from "~/v3/runOpsMigration/unroutableRead.server"; const ParamsSchema = z.object({ runParam: z.string(), @@ -60,8 +61,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 }); diff --git a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.input.$streamId.ts b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.input.$streamId.ts index 7eae8f6fdcf..c53cd5c8a58 100644 --- a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.input.$streamId.ts +++ b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.input.$streamId.ts @@ -8,6 +8,7 @@ import { findEnvironmentBySlug } from "~/models/runtimeEnvironment.server"; import { requireUserId } from "~/services/session.server"; import { EnvironmentParamSchema } from "~/utils/pathBuilder"; import { runStore } from "~/v3/runStore.server"; +import { undefinedOnUnroutableId } from "~/v3/runOpsMigration/unroutableRead.server"; const ParamsSchema = z.object({ runParam: z.string(), @@ -62,8 +63,12 @@ export async function loader({ request, params }: LoaderFunctionArgs) { // Replica lag can null out a live run; a spurious 404 breaks the dashboard Agent tab input-stream // subscription (useRealtimeStream surfaces the error, no 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 }); diff --git a/apps/webapp/app/routes/resources.runs.$runParam.logs.download.ts b/apps/webapp/app/routes/resources.runs.$runParam.logs.download.ts index 649f2ef268f..d86aac6f571 100644 --- a/apps/webapp/app/routes/resources.runs.$runParam.logs.download.ts +++ b/apps/webapp/app/routes/resources.runs.$runParam.logs.download.ts @@ -15,6 +15,7 @@ import { type TraceExportContext, } from "~/v3/eventRepository/traceExport.server"; import { getMollifierBuffer } from "~/v3/mollifier/mollifierBuffer.server"; +import { undefinedOnUnroutableId } from "~/v3/runOpsMigration/unroutableRead.server"; export async function loader({ params, request }: LoaderFunctionArgs) { const user = await requireUser(request); @@ -31,20 +32,23 @@ export async function loader({ params, request }: LoaderFunctionArgs) { // Run-ops read keyed by friendlyId only (routes to the owning DB by residency). Org // membership is a control-plane concern resolved separately below — joining it here is a // cross-DB join that returns nothing once the run lives in run-ops. - let run = await runStore.findRun( - { friendlyId: parsedParams.runParam }, - { - select: { - friendlyId: true, - traceId: true, - organizationId: true, - runtimeEnvironmentId: true, - createdAt: true, - completedAt: true, - taskEventStore: true, - taskIdentifier: true, - }, - } + let run = await undefinedOnUnroutableId( + runStore.findRun( + { friendlyId: parsedParams.runParam }, + { + select: { + friendlyId: true, + traceId: true, + organizationId: true, + runtimeEnvironmentId: true, + createdAt: true, + completedAt: true, + taskEventStore: true, + taskIdentifier: true, + }, + } + ), + { runParam: parsedParams.runParam } ); // Authorize on the control-plane DB: the user must be a member of the run's org. A diff --git a/apps/webapp/app/routes/resources.runs.$runParam.ts b/apps/webapp/app/routes/resources.runs.$runParam.ts index e4328fe4b37..e4e71072978 100644 --- a/apps/webapp/app/routes/resources.runs.$runParam.ts +++ b/apps/webapp/app/routes/resources.runs.$runParam.ts @@ -12,72 +12,76 @@ import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver. import { FINAL_ATTEMPT_STATUSES, isFinalRunStatus } from "~/v3/taskStatus"; import { boundedIn } from "@trigger.dev/database"; +import { undefinedOnUnroutableId } from "~/v3/runOpsMigration/unroutableRead.server"; export type RunInspectorData = UseDataFunctionReturn; export const loader = async ({ request, params }: LoaderFunctionArgs) => { const userId = await requireUserId(request); const parsedParams = v3RunParamsSchema.pick({ runParam: true }).parse(params); - const run = await runStore.findRun( - { - friendlyId: parsedParams.runParam, - }, - { - select: { - id: true, - traceId: true, - //metadata - number: true, - taskIdentifier: true, - friendlyId: true, - isTest: true, - runTags: true, - machinePreset: true, - runtimeEnvironmentId: true, - projectId: true, - lockedById: true, - lockedToVersionId: true, - //status + duration - status: true, - startedAt: true, - createdAt: true, - updatedAt: true, - queuedAt: true, - completedAt: true, - logsDeletedAt: true, - //idempotency - idempotencyKey: true, - //delayed - delayUntil: true, - //ttl - ttl: true, - expiredAt: true, - //queue - queue: true, - concurrencyKey: true, - //schedule - scheduleId: true, - //usage - baseCostInCents: true, - costInCents: true, - usageDurationMs: true, - payload: true, - payloadType: true, - metadata: true, - metadataType: true, - maxAttempts: true, - parentTaskRun: { - select: { - friendlyId: true, + const run = await undefinedOnUnroutableId( + runStore.findRun( + { + friendlyId: parsedParams.runParam, + }, + { + select: { + id: true, + traceId: true, + //metadata + number: true, + taskIdentifier: true, + friendlyId: true, + isTest: true, + runTags: true, + machinePreset: true, + runtimeEnvironmentId: true, + projectId: true, + lockedById: true, + lockedToVersionId: true, + //status + duration + status: true, + startedAt: true, + createdAt: true, + updatedAt: true, + queuedAt: true, + completedAt: true, + logsDeletedAt: true, + //idempotency + idempotencyKey: true, + //delayed + delayUntil: true, + //ttl + ttl: true, + expiredAt: true, + //queue + queue: true, + concurrencyKey: true, + //schedule + scheduleId: true, + //usage + baseCostInCents: true, + costInCents: true, + usageDurationMs: true, + payload: true, + payloadType: true, + metadata: true, + metadataType: true, + maxAttempts: true, + parentTaskRun: { + select: { + friendlyId: true, + }, }, - }, - rootTaskRun: { - select: { - friendlyId: true, + rootTaskRun: { + select: { + friendlyId: true, + }, }, }, - }, - } + } + ), + { runParam: params.runParam ?? params.runId } ); if (!run) { diff --git a/apps/webapp/app/routes/resources.taskruns.$runParam.debug.ts b/apps/webapp/app/routes/resources.taskruns.$runParam.debug.ts index 7275e6d73aa..de9cf813d56 100644 --- a/apps/webapp/app/routes/resources.taskruns.$runParam.debug.ts +++ b/apps/webapp/app/routes/resources.taskruns.$runParam.debug.ts @@ -6,6 +6,7 @@ import { requireUserId } from "~/services/session.server"; import { engine } from "~/v3/runEngine.server"; import { runStore } from "~/v3/runStore.server"; import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver.server"; +import { undefinedOnUnroutableId } from "~/v3/runOpsMigration/unroutableRead.server"; const ParamSchema = z.object({ runParam: z.string(), @@ -18,20 +19,23 @@ export async function loader({ request, params }: LoaderFunctionArgs) { // Run-ops read keyed by friendlyId only (routes to the owning DB by residency). The // project/org-membership auth is a control-plane concern resolved separately below — // joining it here is a cross-DB join that returns nothing once the run lives in run-ops. - const run = await runStore.findRun( - { friendlyId: runParam }, - { - select: { - id: true, - engine: true, - friendlyId: true, - queue: true, - concurrencyKey: true, - queueTimestamp: true, - runtimeEnvironmentId: true, - projectId: true, - }, - } + const run = await undefinedOnUnroutableId( + runStore.findRun( + { friendlyId: runParam }, + { + select: { + id: true, + engine: true, + friendlyId: true, + queue: true, + concurrencyKey: true, + queueTimestamp: true, + runtimeEnvironmentId: true, + projectId: true, + }, + } + ), + { runParam: params.runParam ?? params.runId } ); if (!run) { diff --git a/apps/webapp/app/routes/runs.$runParam.ts b/apps/webapp/app/routes/runs.$runParam.ts index 6723f39a621..7d94462ab00 100644 --- a/apps/webapp/app/routes/runs.$runParam.ts +++ b/apps/webapp/app/routes/runs.$runParam.ts @@ -6,6 +6,7 @@ import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver. import { redirectWithErrorMessage } from "~/models/message.server"; import { requireUser } from "~/services/session.server"; import { rootPath, v3RunPath } from "~/utils/pathBuilder"; +import { undefinedOnUnroutableId } from "~/v3/runOpsMigration/unroutableRead.server"; const ParamsSchema = z.object({ runParam: z.string(), @@ -16,17 +17,20 @@ export async function loader({ params, request }: LoaderFunctionArgs) { const { runParam } = ParamsSchema.parse(params); - const run = await runStore.findRun( - { - friendlyId: runParam, - }, - { - select: { - spanId: true, - projectId: true, - runtimeEnvironmentId: true, + const run = await undefinedOnUnroutableId( + runStore.findRun( + { + friendlyId: runParam, }, - } + { + select: { + spanId: true, + projectId: true, + runtimeEnvironmentId: true, + }, + } + ), + { runParam: params.runParam ?? params.runId } ); if (!run) { diff --git a/apps/webapp/app/v3/runOpsMigration/unroutableRead.server.ts b/apps/webapp/app/v3/runOpsMigration/unroutableRead.server.ts new file mode 100644 index 00000000000..834f85f202a --- /dev/null +++ b/apps/webapp/app/v3/runOpsMigration/unroutableRead.server.ts @@ -0,0 +1,21 @@ +import { UnknownShardKey } from "@internal/run-store"; +import { logger } from "~/services/logger.server"; + +// An id naming a shard the topology has no store for cannot locate a row, so the caller's own +// not-found path is the right answer rather than a 500. Logged so a shard key dropped from a +// config meant to be append-only still alarms instead of reading as an absent run. +export async function undefinedOnUnroutableId( + read: Promise, + context: Record +): Promise { + try { + return await read; + } catch (error) { + if (error instanceof UnknownShardKey) { + logger.warn("Unroutable id treated as not found", { ...context, error: error.message }); + return undefined; + } + + throw error; + } +} From befbf7929e7b72be9172389d88b4c077d7aeaa2a Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Wed, 2 Sep 2026 10:24:03 +0100 Subject: [PATCH 3/4] fix(webapp,run-store): make the unroutable-id fallback actually catch, and count only routed traffic Two defects from review of the previous commit. undefinedOnUnroutableId took the read as a promise, but RoutingRunStore.findRun is not async and resolves the shard before it returns anything, so an unroutable id threw while the argument was still being evaluated and never reached the try block. Every dashboard and resource route it was added to still answered a server error. It now takes a thunk, and a test covers the synchronous-throw case that types cannot. runops_shard_routed_total counted every call through the generic store lookup, which probes, fan-outs and fallback legs also use, so an unrouted miss credited legacy and a waitpoint home-miss counted twice. Counting moved to the two id-routed entry points, so the series means what its name says. Tests cover an unrouted miss and a probe leg. --- apps/webapp/app/routes/@.runs.$runParam.ts | 23 ++-- ...g.projects.$projectParam.runs.$runParam.ts | 21 ++-- .../projects.v3.$projectRef.runs.$runParam.ts | 25 ++-- ...ram.realtime.v1.sessions.$sessionId.$io.ts | 4 +- ...am.realtime.v1.streams.$runId.$streamId.ts | 4 +- ...ltime.v1.streams.$runId.input.$streamId.ts | 4 +- .../resources.runs.$runParam.logs.download.ts | 31 ++--- .../app/routes/resources.runs.$runParam.ts | 117 +++++++++--------- .../resources.taskruns.$runParam.debug.ts | 31 ++--- apps/webapp/app/routes/runs.$runParam.ts | 23 ++-- .../runOpsMigration/unroutableRead.server.ts | 13 +- apps/webapp/app/v3/runStore.server.ts | 2 +- apps/webapp/test/unroutableRead.test.ts | 47 +++++++ .../src/runOpsStore.shardMap.test.ts | 24 ++++ .../run-store/src/runOpsStore.ts | 11 +- 15 files changed, 234 insertions(+), 146 deletions(-) create mode 100644 apps/webapp/test/unroutableRead.test.ts diff --git a/apps/webapp/app/routes/@.runs.$runParam.ts b/apps/webapp/app/routes/@.runs.$runParam.ts index cb7b34ec26c..dc2ba8c33a2 100644 --- a/apps/webapp/app/routes/@.runs.$runParam.ts +++ b/apps/webapp/app/routes/@.runs.$runParam.ts @@ -35,18 +35,19 @@ export async function loader({ params, request }: LoaderFunctionArgs) { } const run = await undefinedOnUnroutableId( - runStore.findRun( - { - friendlyId: runParam, - }, - { - select: { - spanId: true, - runtimeEnvironmentId: true, + () => + runStore.findRun( + { + friendlyId: runParam, + }, + { + select: { + spanId: true, + runtimeEnvironmentId: true, + }, }, - }, - prisma - ), + prisma + ), { runParam: params.runParam ?? params.runId } ); diff --git a/apps/webapp/app/routes/orgs.$organizationSlug.projects.$projectParam.runs.$runParam.ts b/apps/webapp/app/routes/orgs.$organizationSlug.projects.$projectParam.runs.$runParam.ts index 30d2a8bb75a..2cb47971979 100644 --- a/apps/webapp/app/routes/orgs.$organizationSlug.projects.$projectParam.runs.$runParam.ts +++ b/apps/webapp/app/routes/orgs.$organizationSlug.projects.$projectParam.runs.$runParam.ts @@ -17,17 +17,18 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => { const { organizationSlug, projectParam, runParam } = ParamSchema.parse(params); const run = await undefinedOnUnroutableId( - runStore.findRun( - { - friendlyId: runParam, - }, - { - select: { - projectId: true, - runtimeEnvironmentId: true, + () => + runStore.findRun( + { + friendlyId: runParam, }, - } - ), + { + select: { + projectId: true, + runtimeEnvironmentId: true, + }, + } + ), { runParam: params.runParam ?? params.runId } ); diff --git a/apps/webapp/app/routes/projects.v3.$projectRef.runs.$runParam.ts b/apps/webapp/app/routes/projects.v3.$projectRef.runs.$runParam.ts index 210f5d448eb..63475d193d4 100644 --- a/apps/webapp/app/routes/projects.v3.$projectRef.runs.$runParam.ts +++ b/apps/webapp/app/routes/projects.v3.$projectRef.runs.$runParam.ts @@ -38,19 +38,20 @@ export async function loader({ params, request }: LoaderFunctionArgs) { } const run = await undefinedOnUnroutableId( - runStore.findRun( - { - friendlyId: validatedParams.runParam, - }, - { - select: { - friendlyId: true, - spanId: true, - runtimeEnvironmentId: true, + () => + runStore.findRun( + { + friendlyId: validatedParams.runParam, }, - }, - prisma - ), + { + select: { + friendlyId: true, + spanId: true, + runtimeEnvironmentId: true, + }, + }, + prisma + ), { runParam: params.runParam ?? params.runId } ); diff --git a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.sessions.$sessionId.$io.ts b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.sessions.$sessionId.$io.ts index b58add47307..7238310372c 100644 --- a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.sessions.$sessionId.$io.ts +++ b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.sessions.$sessionId.$io.ts @@ -62,10 +62,10 @@ 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 undefinedOnUnroutableId(runStore.findRun(runWhere, runArgs, $replica), { + (await undefinedOnUnroutableId(() => runStore.findRun(runWhere, runArgs, $replica), { runParam: params.runParam, })) ?? - (await undefinedOnUnroutableId(runStore.findRunOnPrimary(runWhere, runArgs), { + (await undefinedOnUnroutableId(() => runStore.findRunOnPrimary(runWhere, runArgs), { runParam: params.runParam, })); diff --git a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.$streamId.ts b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.$streamId.ts index 8373e116202..7a398e91b67 100644 --- a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.$streamId.ts +++ b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.$streamId.ts @@ -61,10 +61,10 @@ 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 undefinedOnUnroutableId(runStore.findRun(runWhere, runArgs, $replica), { + (await undefinedOnUnroutableId(() => runStore.findRun(runWhere, runArgs, $replica), { runParam: params.runParam, })) ?? - (await undefinedOnUnroutableId(runStore.findRunOnPrimary(runWhere, runArgs), { + (await undefinedOnUnroutableId(() => runStore.findRunOnPrimary(runWhere, runArgs), { runParam: params.runParam, })); diff --git a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.input.$streamId.ts b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.input.$streamId.ts index c53cd5c8a58..be963701f4c 100644 --- a/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.input.$streamId.ts +++ b/apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.runs.$runParam.realtime.v1.streams.$runId.input.$streamId.ts @@ -63,10 +63,10 @@ export async function loader({ request, params }: LoaderFunctionArgs) { // Replica lag can null out a live run; a spurious 404 breaks the dashboard Agent tab input-stream // subscription (useRealtimeStream surfaces the error, no auto-retry). Re-read the primary on a miss. const run = - (await undefinedOnUnroutableId(runStore.findRun(runWhere, runArgs, $replica), { + (await undefinedOnUnroutableId(() => runStore.findRun(runWhere, runArgs, $replica), { runParam: params.runParam, })) ?? - (await undefinedOnUnroutableId(runStore.findRunOnPrimary(runWhere, runArgs), { + (await undefinedOnUnroutableId(() => runStore.findRunOnPrimary(runWhere, runArgs), { runParam: params.runParam, })); diff --git a/apps/webapp/app/routes/resources.runs.$runParam.logs.download.ts b/apps/webapp/app/routes/resources.runs.$runParam.logs.download.ts index d86aac6f571..86c1d5cbeb4 100644 --- a/apps/webapp/app/routes/resources.runs.$runParam.logs.download.ts +++ b/apps/webapp/app/routes/resources.runs.$runParam.logs.download.ts @@ -33,21 +33,22 @@ export async function loader({ params, request }: LoaderFunctionArgs) { // membership is a control-plane concern resolved separately below — joining it here is a // cross-DB join that returns nothing once the run lives in run-ops. let run = await undefinedOnUnroutableId( - runStore.findRun( - { friendlyId: parsedParams.runParam }, - { - select: { - friendlyId: true, - traceId: true, - organizationId: true, - runtimeEnvironmentId: true, - createdAt: true, - completedAt: true, - taskEventStore: true, - taskIdentifier: true, - }, - } - ), + () => + runStore.findRun( + { friendlyId: parsedParams.runParam }, + { + select: { + friendlyId: true, + traceId: true, + organizationId: true, + runtimeEnvironmentId: true, + createdAt: true, + completedAt: true, + taskEventStore: true, + taskIdentifier: true, + }, + } + ), { runParam: parsedParams.runParam } ); diff --git a/apps/webapp/app/routes/resources.runs.$runParam.ts b/apps/webapp/app/routes/resources.runs.$runParam.ts index e4e71072978..7cfafb673d4 100644 --- a/apps/webapp/app/routes/resources.runs.$runParam.ts +++ b/apps/webapp/app/routes/resources.runs.$runParam.ts @@ -20,67 +20,68 @@ export const loader = async ({ request, params }: LoaderFunctionArgs) => { const parsedParams = v3RunParamsSchema.pick({ runParam: true }).parse(params); const run = await undefinedOnUnroutableId( - runStore.findRun( - { - friendlyId: parsedParams.runParam, - }, - { - select: { - id: true, - traceId: true, - //metadata - number: true, - taskIdentifier: true, - friendlyId: true, - isTest: true, - runTags: true, - machinePreset: true, - runtimeEnvironmentId: true, - projectId: true, - lockedById: true, - lockedToVersionId: true, - //status + duration - status: true, - startedAt: true, - createdAt: true, - updatedAt: true, - queuedAt: true, - completedAt: true, - logsDeletedAt: true, - //idempotency - idempotencyKey: true, - //delayed - delayUntil: true, - //ttl - ttl: true, - expiredAt: true, - //queue - queue: true, - concurrencyKey: true, - //schedule - scheduleId: true, - //usage - baseCostInCents: true, - costInCents: true, - usageDurationMs: true, - payload: true, - payloadType: true, - metadata: true, - metadataType: true, - maxAttempts: true, - parentTaskRun: { - select: { - friendlyId: true, + () => + runStore.findRun( + { + friendlyId: parsedParams.runParam, + }, + { + select: { + id: true, + traceId: true, + //metadata + number: true, + taskIdentifier: true, + friendlyId: true, + isTest: true, + runTags: true, + machinePreset: true, + runtimeEnvironmentId: true, + projectId: true, + lockedById: true, + lockedToVersionId: true, + //status + duration + status: true, + startedAt: true, + createdAt: true, + updatedAt: true, + queuedAt: true, + completedAt: true, + logsDeletedAt: true, + //idempotency + idempotencyKey: true, + //delayed + delayUntil: true, + //ttl + ttl: true, + expiredAt: true, + //queue + queue: true, + concurrencyKey: true, + //schedule + scheduleId: true, + //usage + baseCostInCents: true, + costInCents: true, + usageDurationMs: true, + payload: true, + payloadType: true, + metadata: true, + metadataType: true, + maxAttempts: true, + parentTaskRun: { + select: { + friendlyId: true, + }, }, - }, - rootTaskRun: { - select: { - friendlyId: true, + rootTaskRun: { + select: { + friendlyId: true, + }, }, }, - }, - } - ), + } + ), { runParam: params.runParam ?? params.runId } ); diff --git a/apps/webapp/app/routes/resources.taskruns.$runParam.debug.ts b/apps/webapp/app/routes/resources.taskruns.$runParam.debug.ts index de9cf813d56..45aca19269a 100644 --- a/apps/webapp/app/routes/resources.taskruns.$runParam.debug.ts +++ b/apps/webapp/app/routes/resources.taskruns.$runParam.debug.ts @@ -20,21 +20,22 @@ export async function loader({ request, params }: LoaderFunctionArgs) { // project/org-membership auth is a control-plane concern resolved separately below — // joining it here is a cross-DB join that returns nothing once the run lives in run-ops. const run = await undefinedOnUnroutableId( - runStore.findRun( - { friendlyId: runParam }, - { - select: { - id: true, - engine: true, - friendlyId: true, - queue: true, - concurrencyKey: true, - queueTimestamp: true, - runtimeEnvironmentId: true, - projectId: true, - }, - } - ), + () => + runStore.findRun( + { friendlyId: runParam }, + { + select: { + id: true, + engine: true, + friendlyId: true, + queue: true, + concurrencyKey: true, + queueTimestamp: true, + runtimeEnvironmentId: true, + projectId: true, + }, + } + ), { runParam: params.runParam ?? params.runId } ); diff --git a/apps/webapp/app/routes/runs.$runParam.ts b/apps/webapp/app/routes/runs.$runParam.ts index 7d94462ab00..f6f9241a449 100644 --- a/apps/webapp/app/routes/runs.$runParam.ts +++ b/apps/webapp/app/routes/runs.$runParam.ts @@ -18,18 +18,19 @@ export async function loader({ params, request }: LoaderFunctionArgs) { const { runParam } = ParamsSchema.parse(params); const run = await undefinedOnUnroutableId( - runStore.findRun( - { - friendlyId: runParam, - }, - { - select: { - spanId: true, - projectId: true, - runtimeEnvironmentId: true, + () => + runStore.findRun( + { + friendlyId: runParam, }, - } - ), + { + select: { + spanId: true, + projectId: true, + runtimeEnvironmentId: true, + }, + } + ), { runParam: params.runParam ?? params.runId } ); diff --git a/apps/webapp/app/v3/runOpsMigration/unroutableRead.server.ts b/apps/webapp/app/v3/runOpsMigration/unroutableRead.server.ts index 834f85f202a..d209ff46ad6 100644 --- a/apps/webapp/app/v3/runOpsMigration/unroutableRead.server.ts +++ b/apps/webapp/app/v3/runOpsMigration/unroutableRead.server.ts @@ -1,15 +1,18 @@ import { UnknownShardKey } from "@internal/run-store"; import { logger } from "~/services/logger.server"; -// An id naming a shard the topology has no store for cannot locate a row, so the caller's own -// not-found path is the right answer rather than a 500. Logged so a shard key dropped from a -// config meant to be append-only still alarms instead of reading as an absent run. +// An id naming a shard with no configured store cannot locate a row, so the caller's own +// not-found path is the right answer rather than a 500. Logged so a dropped shard key still +// alarms instead of reading as an absent run. +// +// A thunk, not a promise: `RoutingRunStore.findRun` is not async and routes before returning, so +// an unroutable id throws while the argument is still being evaluated. export async function undefinedOnUnroutableId( - read: Promise, + read: () => Promise, context: Record ): Promise { try { - return await read; + return await read(); } catch (error) { if (error instanceof UnknownShardKey) { logger.warn("Unroutable id treated as not found", { ...context, error: error.message }); diff --git a/apps/webapp/app/v3/runStore.server.ts b/apps/webapp/app/v3/runStore.server.ts index 6ab9675d31c..8441f82ea1a 100644 --- a/apps/webapp/app/v3/runStore.server.ts +++ b/apps/webapp/app/v3/runStore.server.ts @@ -141,7 +141,7 @@ const routingStoreMetrics: RoutingStoreMetrics = singleton("routingStoreMetrics" // the shard alphabet plus the two reserved keys. const shardRouted = new Counter({ name: "runops_shard_routed_total", - help: "Run-ops store resolutions keyed by id. Fan-outs and probe hits resolve no single shard and are not counted, so read this as relative ramp signal rather than served requests.", + help: "Operations routed to a shard by an id that resolves to it alone. Fan-outs, probes and fallback legs resolve no single shard and are excluded.", labelNames: ["shard"], registers: [metricsRegister], }); diff --git a/apps/webapp/test/unroutableRead.test.ts b/apps/webapp/test/unroutableRead.test.ts new file mode 100644 index 00000000000..c3e6dc01254 --- /dev/null +++ b/apps/webapp/test/unroutableRead.test.ts @@ -0,0 +1,47 @@ +import { UnknownShardKey } from "@internal/run-store"; +import { describe, expect, it } from "vitest"; +import { undefinedOnUnroutableId } from "~/v3/runOpsMigration/unroutableRead.server"; + +// `RoutingRunStore.findRun` is not async: it resolves the shard and throws before returning +// anything, so these stand in for a read that fails while the call expression is still being +// evaluated rather than one that returns a rejected promise. +function throwsWhileRouting(): Promise { + throw new UnknownShardKey("q", ["legacy", "new"]); +} + +function rejectsLater(): Promise { + return Promise.reject(new UnknownShardKey("q", ["legacy", "new"])); +} + +describe("undefinedOnUnroutableId", () => { + it("returns undefined when the read throws synchronously while routing", async () => { + await expect( + undefinedOnUnroutableId(() => throwsWhileRouting(), { runParam: "run_x" }) + ).resolves.toBeUndefined(); + }); + + it("returns undefined when the read rejects", async () => { + await expect( + undefinedOnUnroutableId(() => rejectsLater(), { runParam: "run_x" }) + ).resolves.toBeUndefined(); + }); + + it("passes a successful read through untouched", async () => { + await expect(undefinedOnUnroutableId(async () => "found", { runParam: "run_x" })).resolves.toBe( + "found" + ); + }); + + it("rethrows anything that is not an unroutable id", async () => { + const boom = new Error("connection refused"); + + await expect( + undefinedOnUnroutableId( + () => { + throw boom; + }, + { runParam: "run_x" } + ) + ).rejects.toBe(boom); + }); +}); diff --git a/internal-packages/run-store/src/runOpsStore.shardMap.test.ts b/internal-packages/run-store/src/runOpsStore.shardMap.test.ts index b107fec24fa..b4d70363642 100644 --- a/internal-packages/run-store/src/runOpsStore.shardMap.test.ts +++ b/internal-packages/run-store/src/runOpsStore.shardMap.test.ts @@ -409,6 +409,30 @@ describe("RoutingRunStore #distinctStores — one entry per database", () => { expect(routed).toEqual(["a", "b", "a"]); }); + // The counter is read as per-shard traffic during a ramp, so a lookup that had to search + // rather than route must not be attributed to whichever store answered. + it("does not count an unrouted lookup that misses", async () => { + const routed: string[] = []; + const { router } = buildNShardRouter(["a", "b"], { routed }); + + await router.findRun({ spanId: "nope" }); + + expect(routed).toEqual([]); + }); + + it("does not count the legacy leg of an unrouted probe", async () => { + const routed: string[] = []; + const { router } = buildNShardRouter(["a", "b"], { routed }); + + await router.findRunsByIdempotencyKeys({ + runtimeEnvironmentId: "env_1", + taskIdentifier: "task", + idempotencyKeys: ["k1"], + }); + + expect(routed).toEqual([]); + }); + it("does not count anything before an operation is routed", async () => { const routed: string[] = []; buildNShardRouter(["a", "b"], { routed }); diff --git a/internal-packages/run-store/src/runOpsStore.ts b/internal-packages/run-store/src/runOpsStore.ts index b50686465e1..7855f75955d 100644 --- a/internal-packages/run-store/src/runOpsStore.ts +++ b/internal-packages/run-store/src/runOpsStore.ts @@ -209,6 +209,13 @@ export class RoutingRunStore implements RunStore { if (store === undefined) { throw new UnknownShardKey(key, [...this.#shards.keys()]); } + return store; + } + + // Counted separately from #shardStore, which probes, fan-outs and fallback legs also go through. + // Only a key an id resolved to on its own is traffic for that shard. + #shardStoreRoutedById(key: ShardKey): RunStore { + const store = this.#shardStore(key); this.#metrics.recordShardRouted(key); return store; } @@ -461,7 +468,7 @@ export class RoutingRunStore implements RunStore { // Route an existing run-ops id by residency. Throws on an unclassifiable id. #route(id: string): RunStore { - return this.#shardStore(this.#shardKeyOf(id)); + return this.#shardStoreRoutedById(this.#shardKeyOf(id)); } // Best-effort shard key; falls back to #idlessRouteShard when the id is absent. Classification is @@ -479,7 +486,7 @@ export class RoutingRunStore implements RunStore { } #routeOrNew(id: string | undefined): RunStore { - return this.#shardStore(this.#routeKeyOrDefault(id)); + return this.#shardStoreRoutedById(this.#routeKeyOrDefault(id)); } // WRITE routing is pure id-shape (cuid → LEGACY, run-ops id → NEW). A LEGACY-classified id is From 1e352d0866157a44cbe40d68326f953441b76219 Mon Sep 17 00:00:00 2001 From: Daniel Sutton Date: Wed, 2 Sep 2026 10:45:17 +0100 Subject: [PATCH 4/4] fix(run-store): exclude id-less operations from the routed-shard counter An operation with no id falls back to the default store rather than resolving to one, so counting it as id-routed traffic inflated the default series during a ramp. The counter now fires only when an id resolved the key. Also moves the unroutable-id helper test next to its source. --- .../v3/runOpsMigration/unroutableRead.server.test.ts} | 2 +- .../run-store/src/runOpsStore.shardMap.test.ts | 10 ++++++++++ internal-packages/run-store/src/runOpsStore.ts | 5 ++++- 3 files changed, 15 insertions(+), 2 deletions(-) rename apps/webapp/{test/unroutableRead.test.ts => app/v3/runOpsMigration/unroutableRead.server.test.ts} (94%) diff --git a/apps/webapp/test/unroutableRead.test.ts b/apps/webapp/app/v3/runOpsMigration/unroutableRead.server.test.ts similarity index 94% rename from apps/webapp/test/unroutableRead.test.ts rename to apps/webapp/app/v3/runOpsMigration/unroutableRead.server.test.ts index c3e6dc01254..c5f6f03e05f 100644 --- a/apps/webapp/test/unroutableRead.test.ts +++ b/apps/webapp/app/v3/runOpsMigration/unroutableRead.server.test.ts @@ -1,6 +1,6 @@ import { UnknownShardKey } from "@internal/run-store"; import { describe, expect, it } from "vitest"; -import { undefinedOnUnroutableId } from "~/v3/runOpsMigration/unroutableRead.server"; +import { undefinedOnUnroutableId } from "./unroutableRead.server"; // `RoutingRunStore.findRun` is not async: it resolves the shard and throws before returning // anything, so these stand in for a read that fails while the call expression is still being diff --git a/internal-packages/run-store/src/runOpsStore.shardMap.test.ts b/internal-packages/run-store/src/runOpsStore.shardMap.test.ts index b4d70363642..717cfb3f788 100644 --- a/internal-packages/run-store/src/runOpsStore.shardMap.test.ts +++ b/internal-packages/run-store/src/runOpsStore.shardMap.test.ts @@ -420,6 +420,16 @@ describe("RoutingRunStore #distinctStores — one entry per database", () => { expect(routed).toEqual([]); }); + it("does not count an id-less operation against the default store", async () => { + const routed: string[] = []; + const { router } = buildNShardRouter(["a", "b"], { routed }); + + await router.findRun({ spanId: "s1" }, undefined); + await router.runInTransaction(undefined, async () => undefined); + + expect(routed).toEqual([]); + }); + it("does not count the legacy leg of an unrouted probe", async () => { const routed: string[] = []; const { router } = buildNShardRouter(["a", "b"], { routed }); diff --git a/internal-packages/run-store/src/runOpsStore.ts b/internal-packages/run-store/src/runOpsStore.ts index 7855f75955d..fe2e9fa61a8 100644 --- a/internal-packages/run-store/src/runOpsStore.ts +++ b/internal-packages/run-store/src/runOpsStore.ts @@ -486,7 +486,10 @@ export class RoutingRunStore implements RunStore { } #routeOrNew(id: string | undefined): RunStore { - return this.#shardStoreRoutedById(this.#routeKeyOrDefault(id)); + const key = this.#routeKeyOrDefault(id); + // An absent id fell back to the default store rather than resolving anywhere, so it is not + // traffic attributable to that shard. + return typeof id === "string" ? this.#shardStoreRoutedById(key) : this.#shardStore(key); } // WRITE routing is pure id-shape (cuid → LEGACY, run-ops id → NEW). A LEGACY-classified id is