fix(webapp,run-engine): bound the sweep count fields, and guard the org cache against a stale read

Two review findings, plus a correction of my own.

The engine recorded whatever count keys a pass returned, and each key is a
metric attribute, so an unrecognised one would mint a time series. Keys are now
filtered against a fixed list at the point of record, matching the allowlist the
webapp adapter already applies.

A replica read that started before an invalidation could land after the primary
read and put the superseded value back in the cache, holding it for the rest of
the window. Each organisation now carries a generation that invalidation
increments, and a load whose generation is no longer current discards its own
result.

The correction: an earlier commit made the sweep failure path rethrow, on the
stated grounds that only the dead-letter path re-enqueues the next cron
occurrence. That is wrong. The worker reschedules a cron job on the acknowledge
path as well, so returning normally always continued the chain. Throwing only
added a dead-letter entry for every transient failure. Both rethrows are
reverted and the reason is recorded at each site, because the claim reads
plausibly and would otherwise be reintroduced.

Author: d-cs

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

@@ -91,28 +91,38 @@ function createOrgModeSource(): OrgModeSource {
     ttl: env.RUN_ENGINE_SNAPSHOT_STORE_ORG_MODE_CACHE_TTL_MS ?? DEFAULT_CACHE_TTL_MS,
   });
   const inFlight = new Set<string>();
+  // A replica read that started before an invalidation can land after the primary read and put the
+  // superseded value back. A per-organisation generation lets a stale load discard its own result.
+  const generations = new Map<string, number>();
+  const generationOf = (organizationId: string) => generations.get(organizationId) ?? 0;
 
   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.
+      const generation = generationOf(organizationId) + 1;
+      generations.set(organizationId, generation);
       cache.delete(organizationId);
-      void load(organizationId, prisma);
+      void load(organizationId, prisma, generation);
     },
     refresh: (organizationId) => {
       if (inFlight.has(organizationId)) {
         return;
       }
       inFlight.add(organizationId);
 
-      void load(organizationId, $replica).finally(() => {
+      void load(organizationId, $replica, generationOf(organizationId)).finally(() => {
         inFlight.delete(organizationId);
       });
     },
   };
 
-  function load(organizationId: string, client: typeof prisma | typeof $replica) {
+  function load(
+    organizationId: string,
+    client: typeof prisma | typeof $replica,
+    generation: number
+  ) {
     return client.organization
       .findFirst({ where: { id: organizationId }, select: { featureFlags: true } })
       .then((row) => {
@@ -121,6 +131,10 @@ function createOrgModeSource(): OrgModeSource {
         const raw = (row?.featureFlags as Record<string, unknown> | null | undefined)?.[
           FEATURE_FLAG.snapshotStoreOrgMode
         ];
+        // A newer invalidation happened while this read was in flight, so its answer is stale.
+        if (generation < generationOf(organizationId)) {
+          return;
+        }
         cache.set(organizationId, cachedOrgModeFor(raw));
       })
       .catch((error) => {

apps/webapp/test/snapshotSweepRunner.test.ts

@@ -77,6 +77,8 @@ redisTest("reports failed and still releases its own lock", async ({ redisOption
       lockTtlMs: 60_000,
     });
 
+    // Resolving is deliberate: the worker reschedules a cron job on acknowledge as well as on the
+    // dead-letter path, so a failure needs no throw to keep the chain alive.
     expect((await runner(opts())).outcome).toBe("failed");
     expect(await client.get(LOCK_KEY)).toBeNull();
   } finally {

internal-packages/run-engine/src/engine/index.ts

@@ -93,6 +93,7 @@ import {
 } from "./controlPlaneResolver.js";
 import { TtlSystem } from "./systems/ttlSystem.js";
 import { WaitpointSystem } from "./systems/waitpointSystem.js";
+import { SNAPSHOT_SWEEP_COUNT_FIELDS } from "./types.js";
 import type {
   EngineWorker,
   HeartbeatTimeouts,
@@ -2976,16 +2977,16 @@ export class RunEngine {
       outcome = result.outcome;
       counts = result.counts;
     } catch (error) {
+      // Deliberately not rethrown. Both the acknowledge path and the dead-letter path reschedule a
+      // cron job, so returning here continues the chain; throwing would only add a dead-letter
+      // entry for every transient blip. The outcome metric is the signal.
       this.logger.error("sweepSnapshotOrphans threw", { error });
-      // Rethrow after the finally records the outcome. Acknowledging here would skip the
-      // dead-letter path, and that path is what re-enqueues the next occurrence, so one transient
-      // failure would stop the sweep for good.
-      throw error;
     } finally {
       globalThis.clearTimeout(abortAt);
       this.snapshotSweepPassCounter?.add(1, { outcome });
       for (const [field, value] of Object.entries(counts ?? {})) {
-        if (typeof value === "number") {
+        // Each field is a metric attribute, so an unrecognised key would mint a time series.
+        if (typeof value === "number" && SNAPSHOT_SWEEP_COUNT_FIELDS.includes(field as never)) {
           this.snapshotSweepCountsHistogram?.record(value, { field });
         }
       }

internal-packages/run-engine/src/engine/types.ts

@@ -58,6 +58,16 @@ export type SnapshotSweepOutcome =
   | "unbound"
   | "aborted";
 
+export const SNAPSHOT_SWEEP_COUNT_FIELDS = [
+  "scanned",
+  "expired",
+  "deleted",
+  "skipped",
+  "pendingDeletion",
+  "nodes",
+  "partial",
+] as const;
+
 export type SnapshotSweepCountField =
   | "scanned"
   | "expired"