fix(webapp,run-engine): address review findings on the snapshot store wiring

The sweep handler caught its own failure, logged it and returned, which
acknowledges the job. The dead-letter path is what re-enqueues the next
occurrence, and only a throw reaches it, so one transient failure would have
stopped the sweep for good. It now rethrows after the outcome is recorded.

The save guard only covered the global dial, so an organisation could be set
past off with no connection configured and silently do nothing, which is the
exact silence the guard exists to prevent. It now covers both keys and runs on
the organisation routes as well. A companion check refuses the
per-organisation key on a global save, where nothing reads it.

Invalidation read the replica immediately after a primary write, so replica lag
could re-cache the value the write had just replaced and hold it for the whole
cache window. That is worse than not invalidating at all. Invalidation now
drops the entry and re-reads from the primary; the background warm-up keeps the
replica.

A pass cancelled at shutdown reported itself as failed. It now reports aborted.

Metric attributes are bounded at the adapter and in the engine's option type.
The values were already bounded in practice, but both sides typed them as
plain strings, so an unrecognised value now collapses to a single bucket rather
than minting a time series.

Also drops mocks from the boot test, which the injected dependencies made
unnecessary, and replaces a dial position in the run-store test that no longer
exists.

Author: d-cs

apps/webapp/app/routes/admin.api.v1.feature-flags.ts

@@ -10,7 +10,10 @@ import {
   withoutDerivedKeys,
 } from "~/v3/featureFlags.server";
 import { validatePartialFeatureFlags } from "~/v3/featureFlags";
-import { snapshotStoreFlagSaveError } from "~/v3/snapshotStoreFlagGuard.server";
+import {
+  globalOnlySnapshotStoreFlagError,
+  snapshotStoreFlagSaveError,
+} from "~/v3/snapshotStoreFlagGuard.server";
 
 export async function action({ request }: ActionFunctionArgs) {
   await requireAdminApiRequest(request);
@@ -31,6 +34,11 @@ export async function action({ request }: ActionFunctionArgs) {
       );
     }
 
+    const globalOnlyError = globalOnlySnapshotStoreFlagError(body as Record<string, unknown>);
+    if (globalOnlyError) {
+      return json({ error: globalOnlyError }, { status: 400 });
+    }
+
     const snapshotStoreError = snapshotStoreFlagSaveError(body as Record<string, unknown>, {
       redisHostConfigured: !!env.RUN_ENGINE_SNAPSHOT_STORE_REDIS_HOST,
     });

apps/webapp/app/routes/admin.api.v1.orgs.$organizationId.feature-flags.ts

@@ -6,6 +6,7 @@ import { env } from "~/env.server";
 import { prisma } from "~/db.server";
 import { requireAdminApiRequest } from "~/services/personalAccessToken.server";
 import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver.server";
+import { snapshotStoreFlagSaveError } from "~/v3/snapshotStoreFlagGuard.server";
 import { invalidateSnapshotStoreOrgMode } from "~/v3/snapshotStoreMode.server";
 import { selectMintBaselineSource, stampMintKindFlip } from "~/v3/runOpsMigration/mintFlipGrace";
 import { validatePartialFeatureFlags, withoutOrgForbiddenSnapshotKeys } from "~/v3/featureFlags";
@@ -77,6 +78,13 @@ export async function action({ request, params }: ActionFunctionArgs) {
 
     const requestedFlags = withoutOrgForbiddenSnapshotKeys(rawRequestedFlags);
 
+    const snapshotStoreError = snapshotStoreFlagSaveError(requestedFlags, {
+      redisHostConfigured: !!env.RUN_ENGINE_SNAPSHOT_STORE_REDIS_HOST,
+    });
+    if (snapshotStoreError) {
+      return json({ error: snapshotStoreError }, { status: 400 });
+    }
+
     // Seed the flip baseline from the current GLOBAL mint flags so an org's FIRST per-org override
     // is graced from the currently-effective global kind, not the hardcoded default "cuid".
     const globalFlags = (await getGlobalFlags()) as Record<string, unknown>;

apps/webapp/app/routes/admin.api.v2.orgs.$organizationId.feature-flags.ts

@@ -6,6 +6,7 @@ import { env } from "~/env.server";
 import { prisma } from "~/db.server";
 import { requireUser } from "~/services/session.server";
 import { controlPlaneResolver } from "~/v3/runOpsMigration/controlPlaneResolver.server";
+import { snapshotStoreFlagSaveError } from "~/v3/snapshotStoreFlagGuard.server";
 import { invalidateSnapshotStoreOrgMode } from "~/v3/snapshotStoreMode.server";
 import { selectMintBaselineSource, stampMintKindFlip } from "~/v3/runOpsMigration/mintFlipGrace";
 import { flags as getGlobalFlags } from "~/v3/featureFlags.server";
@@ -146,6 +147,13 @@ export async function action({ request, params }: ActionFunctionArgs) {
 
   const requestedFlags = withoutOrgForbiddenSnapshotKeys(rawRequestedFlags);
 
+  const snapshotStoreError = snapshotStoreFlagSaveError(requestedFlags, {
+    redisHostConfigured: !!env.RUN_ENGINE_SNAPSHOT_STORE_REDIS_HOST,
+  });
+  if (snapshotStoreError) {
+    return json({ error: snapshotStoreError }, { status: 400 });
+  }
+
   // Seed the flip baseline from the current GLOBAL mint flags so an org's FIRST per-org override
   // is graced from the currently-effective global kind, not the hardcoded default "cuid".
   const globalFlags = (await getGlobalFlags()) as Record<string, unknown>;

apps/webapp/app/routes/admin.feature-flags.tsx

@@ -17,7 +17,10 @@ import {
   lockedFlagsInPayload,
   validatePartialFeatureFlags,
 } from "~/v3/featureFlags";
-import { snapshotStoreFlagSaveError } from "~/v3/snapshotStoreFlagGuard.server";
+import {
+  globalOnlySnapshotStoreFlagError,
+  snapshotStoreFlagSaveError,
+} from "~/v3/snapshotStoreFlagGuard.server";
 import { flags as getGlobalFlags, replaceGlobalFeatureFlags } from "~/v3/featureFlags.server";
 import { featuresForRequest } from "~/features.server";
 import { Button } from "~/components/primitives/Buttons";
@@ -130,6 +133,11 @@ export const action = dashboardAction(
       );
     }
 
+    const globalOnlyError = globalOnlySnapshotStoreFlagError(parsed.data.flags);
+    if (globalOnlyError) {
+      return json({ error: globalOnlyError }, { status: 400 });
+    }
+
     const snapshotStoreError = snapshotStoreFlagSaveError(parsed.data.flags, {
       redisHostConfigured: !!env.RUN_ENGINE_SNAPSHOT_STORE_REDIS_HOST,
     });

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

@@ -13,10 +13,26 @@ export function snapshotStoreFlagSaveError(
     return undefined;
   }
 
-  const mode = requested[FEATURE_FLAG.snapshotStoreMode];
-  if (typeof mode === "string" && mode !== "off") {
-    return `Cannot set ${FEATURE_FLAG.snapshotStoreMode} to "${mode}": RUN_ENGINE_SNAPSHOT_STORE_REDIS_HOST is not configured in this deployment, so the snapshot store is never constructed and the flag would have no effect.`;
+  // Both keys, because either one past `off` is equally silent without a connection.
+  for (const key of [FEATURE_FLAG.snapshotStoreMode, FEATURE_FLAG.snapshotStoreOrgMode] as const) {
+    const value = requested[key];
+    if (typeof value === "string" && value !== "off") {
+      return `Cannot set ${key} to "${value}": RUN_ENGINE_SNAPSHOT_STORE_REDIS_HOST is not configured in this deployment, so the snapshot store is never constructed and the flag would have no effect.`;
+    }
   }
 
   return undefined;
 }
+
+/**
+ * Refuses an organisation-only key on a global save. Nothing reads the global row for it, so a
+ * value saved there is inert, and an inert control an operator can set is worse than no control.
+ */
+export function globalOnlySnapshotStoreFlagError(
+  requested: Record<string, unknown>
+): string | undefined {
+  if (FEATURE_FLAG.snapshotStoreOrgMode in requested) {
+    return `${FEATURE_FLAG.snapshotStoreOrgMode} is per-organisation only; nothing reads it from the global flags, so setting it here would have no effect.`;
+  }
+  return undefined;
+}

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

@@ -3,6 +3,61 @@ import type { DecoratorMetrics, SnapshotStoreMetrics } from "@internal/run-store
 
 export type SnapshotSweepCounts = Record<string, number | boolean>;
 
+// Metric attributes must be bounded: every one of these is a time series. The store and the
+// decorator type their outcome strings loosely, so anything unrecognised collapses to "other"
+// rather than minting a series.
+const APPEND_OUTCOMES = ["written", "duplicate", "forked", "skippedNoKeyspace"] as const;
+const APPEND_TTLS = ["none", "completion", "reapplied"] as const;
+const WRITE_OUTCOMES = ["written", "staged", "post_expiry", "skipped", "failed"] as const;
+const READ_SOURCES = ["redis", "postgres"] as const;
+const SWEEP_OUTCOMES = [
+  "completed",
+  "partial",
+  "skipped_locked",
+  "failed",
+  "unbound",
+  "aborted",
+] as const;
+const SWEEP_FIELDS = [
+  "scanned",
+  "expired",
+  "deleted",
+  "skipped",
+  "pendingDeletion",
+  "nodes",
+] as const;
+
+const WRITE_SITES = [
+  "createRun",
+  "createCancelledRun",
+  "completeAttemptSuccess",
+  "expireRun",
+  "expireParkedRun",
+  "rescheduleRun",
+  "lockRunToWorker",
+  "createExecutionSnapshot",
+  "runInTransaction",
+] as const;
+const READ_METHODS = [
+  "findLatestExecutionSnapshot",
+  "findExecutionSnapshot",
+  "findManyExecutionSnapshots",
+  "findSnapshotCompletedWaitpointIds",
+  "findSnapshotCompletedWaitpointIdsWithPresence",
+] as const;
+const SNAPSHOT_OPS = [
+  "append",
+  "getById",
+  "getLatest",
+  "getSince",
+  "getSinceCreatedAt",
+  "getSnapshotWaitpointIds",
+] as const;
+
+function bounded(value: string, allowed: readonly string[]): string {
+  return allowed.includes(value) ? value : "other";
+}
+
 /**
  * Every instrument is created inside this function. At module scope they would register on every
  * boot, including deployments with no snapshot-store Redis configured.
@@ -23,37 +78,48 @@ export function createSnapshotStoreMetrics(meter: Meter) {
   const sweepCounts = meter.createHistogram("run_engine.snapshot_store.sweep_counts");
 
   const store: SnapshotStoreMetrics = {
-    recordAppend: (outcome, ttl) => appendTotal.add(1, { outcome, ttl }),
+    recordAppend: (outcome, ttl) =>
+      appendTotal.add(1, {
+        outcome: bounded(outcome, APPEND_OUTCOMES),
+        ttl: bounded(ttl, APPEND_TTLS),
+      }),
     recordEntryBytes: (bytes) => entryBytes.record(bytes),
     recordCycleKeyBytes: (bytes) => cycleKeyBytes.record(bytes),
     recordCycleCount: (count) => cycleCount.record(count),
     recordSkippedNoKeyspace: () => skippedNoKeyspace.add(1),
     recordCycleMismatch: () => cycleMismatch.add(1),
-    recordLatency: (op, ms) => opLatency.record(ms, { op }),
+    recordLatency: (op, ms) => opLatency.record(ms, { op: bounded(op, SNAPSHOT_OPS) }),
   };
 
   const decorator: DecoratorMetrics = {
     recordWrite: (site, outcome) => {
-      appendTotal.add(1, { site, outcome });
+      appendTotal.add(1, {
+        site: bounded(site, WRITE_SITES),
+        outcome: bounded(outcome, WRITE_OUTCOMES),
+      });
       if (outcome === "post_expiry") {
         postExpiryWrite.add(1);
       }
       if (outcome === "staged") {
         flushStaged.add(1);
       }
     },
-    recordAppendFailed: (site) => appendFailed.add(1, { site }),
-    recordRead: (method, source) => readSource.add(1, { method, source }),
+    recordAppendFailed: (site) => appendFailed.add(1, { site: bounded(site, WRITE_SITES) }),
+    recordRead: (method, source) =>
+      readSource.add(1, {
+        method: bounded(method, READ_METHODS),
+        source: bounded(source, READ_SOURCES),
+      }),
   };
 
   /** One emitter per pass, so a pass that throws is distinguishable from one that succeeded. */
   function recordSweepPass(outcome: string, counts?: SnapshotSweepCounts): void {
-    sweepPass.add(1, { outcome });
+    sweepPass.add(1, { outcome: bounded(outcome, SWEEP_OUTCOMES) });
     if (!counts) {
       return;
     }
     for (const [field, value] of Object.entries(counts)) {
-      if (typeof value === "number") {
+      if (typeof value === "number" && SWEEP_FIELDS.includes(field as never)) {
         sweepCounts.record(value, { field });
       }
     }

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

@@ -1,6 +1,6 @@
 import { LRUCache } from "lru-cache";
 import type { SnapshotStoreMode, SnapshotStoreModeResolver } from "@internal/run-store";
-import { $replica } from "~/db.server";
+import { $replica, prisma } from "~/db.server";
 import { env } from "~/env.server";
 import { logger } from "~/services/logger.server";
 import { singleton } from "~/utils/singleton";
@@ -40,11 +40,16 @@ type OrgModeSource = {
   get(organizationId: string): CachedOrgMode | undefined;
   /** Fire-and-forget, de-duplicated per organisation, never throws. */
   refresh(organizationId: string): void;
+  /** Re-reads from the primary after a write, so replica lag cannot re-cache the old value. */
+  invalidate(organizationId: string): void;
 };
 
+/** What resolution needs. Invalidation is a save-path concern, not a read-path one. */
+type ResolverOrgSource = Pick<OrgModeSource, "get" | "refresh">;
+
 export function buildSnapshotStoreModeResolver(deps: {
   globalMode: () => DialMode | undefined;
-  orgMode: OrgModeSource;
+  orgMode: ResolverOrgSource;
   envFloor: DialMode;
 }): SnapshotStoreModeResolver {
   return {
@@ -89,33 +94,42 @@ function createOrgModeSource(): OrgModeSource {
 
   return {
     get: (organizationId) => cache.get(organizationId),
+    invalidate: (organizationId) => {
+      // Drop first, so a resolve between now and the re-read falls back rather than serving a
+      // value the write just replaced.
+      cache.delete(organizationId);
+      void load(organizationId, prisma);
+    },
     refresh: (organizationId) => {
       if (inFlight.has(organizationId)) {
         return;
       }
       inFlight.add(organizationId);
 
-      void $replica.organization
-        .findFirst({ where: { id: organizationId }, select: { featureFlags: true } })
-        .then((row) => {
-          // Only the narrow per-org key. The blob is never passed as `overrides` for the global
-          // key, where a parsing override would win outright.
-          const raw = (row?.featureFlags as Record<string, unknown> | null | undefined)?.[
-            FEATURE_FLAG.snapshotStoreOrgMode
-          ];
-          cache.set(organizationId, cachedOrgModeFor(raw));
-        })
-        .catch((error) => {
-          logger.warn("snapshotStoreMode: organisation override refresh failed", {
-            organizationId,
-            error,
-          });
-        })
-        .finally(() => {
-          inFlight.delete(organizationId);
-        });
+      void load(organizationId, $replica).finally(() => {
+        inFlight.delete(organizationId);
+      });
     },
   };
+
+  function load(organizationId: string, client: typeof prisma | typeof $replica) {
+    return client.organization
+      .findFirst({ where: { id: organizationId }, select: { featureFlags: true } })
+      .then((row) => {
+        // Only the narrow per-org key. The blob is never passed as `overrides` for the global
+        // key, where a parsing override would win outright.
+        const raw = (row?.featureFlags as Record<string, unknown> | null | undefined)?.[
+          FEATURE_FLAG.snapshotStoreOrgMode
+        ];
+        cache.set(organizationId, cachedOrgModeFor(raw));
+      })
+      .catch((error) => {
+        logger.warn("snapshotStoreMode: organisation override read failed", {
+          organizationId,
+          error,
+        });
+      });
+  }
 }
 
 /** Built on first use, never at import: importing this module must have no side effect. */
@@ -134,5 +148,5 @@ export const snapshotStoreModeResolver: SnapshotStoreModeResolver = buildSnapsho
 
 /** Called by the organisation flag save path so the writing process sees a dial change at once. */
 export function invalidateSnapshotStoreOrgMode(organizationId: string): void {
-  orgModeSource().refresh(organizationId);
+  orgModeSource().invalidate(organizationId);
 }

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

@@ -44,6 +44,9 @@ export function buildSnapshotSweepRunner(deps: {
       }
       return { outcome: "completed", counts };
     } catch (error) {
+      if (signal.aborted) {
+        return { outcome: "aborted" };
+      }
       logger.error("snapshot orphan sweep pass failed", { error });
       return { outcome: "failed" };
     } finally {