perf(webapp): shard the mollifier stale sweep to bound per-tick work

The previous sweep was unbounded along two dimensions: every tick walked
every org and every env (via buffer.listOrgs + listEnvsForOrg). At the
sweep's default per-env entry cap of 1000, an incident-scale fan-out
gave O(orgs * envs * 1000) Redis round-trips per tick — running far
longer than the 5-minute interval and triggering the inFlight guard
to drop every subsequent tick until the slow pass finished.

Shard the work via a durable cursor:

- New file `mollifierStaleSweepState.server.ts` owns three Redis keys
  (`mollifier:stale_sweep:{cursor,org_list,counts}`), all under the
  mollifier namespace but separated from the buffer's own state. The
  state class has its own Redis client; the buffer's existing
  `MollifierBuffer` API surface is untouched.
- On `cursor === 0` the org list is rebuilt by snapshotting
  `buffer.listOrgs()` into the frozen LIST — the cycle's frozen view.
- Each tick consumes up to `maxOrgsPerPass` orgs (default 100),
  processes them, and advances the cursor. When the cursor reaches the
  end of the LIST it wraps to 0; the next tick rebuilds and starts the
  next cycle.
- The per-env counts HASH is the source of truth for the gauge
  snapshot. Visiting an env with zero stale entries HDEL's its hash
  field — gauge clears immediately on revisit. Envs not revisited this
  tick keep their last-known value (durability across ticks AND across
  webapp restarts), accepting a worst-case lag of one full cursor cycle
  before a no-longer-stale env clears.

Snapshot contract change: only envs with non-zero stale counts appear
in the reported `Map`. The telemetry layer (`mollifierTelemetry.server.ts`
`reportStaleEntrySnapshot`) sums values, so absence is equivalent to
zero for the gauge — the alert behaviour is unchanged.

Tests:
- New: "shards work across ticks: cursor advances by maxOrgsPerPass and
  wraps after a full cycle" — drives a 5-org fixture with cap=2,
  asserts the cursor's three-tick progression and wrap.
- New: "clears an env from the durable snapshot on revisit when it has
  entries but none currently stale" — same entry flips
  stale→not-stale by varying the sweep's `now`, asserts HDEL on
  revisit.
- Existing tests updated to inject `state`; one assertion shape
  rewritten ("snapshot omits envs that have entries but none stale")
  to match the new HDEL semantics.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: d-cs

apps/webapp/app/v3/mollifier/mollifierStaleSweep.server.ts

@@ -1,15 +1,22 @@
 import type { MollifierBuffer } from "@trigger.dev/redis-worker";
 import { logger as defaultLogger } from "~/services/logger.server";
 import { getMollifierBuffer } from "./mollifierBuffer.server";
+import { MollifierStaleSweepState, type StaleSweepStateStore } from "./mollifierStaleSweepState.server";
 import {
   recordStaleEntry as defaultRecordStaleEntry,
   reportStaleEntrySnapshot as defaultReportStaleEntrySnapshot,
 } from "./mollifierTelemetry.server";
 
-// One pass of the sweep scans every env's queue LIST. The per-env page
-// is bounded so a single pathological env can't make the sweep run
-// unboundedly long.
+// One pass of the sweep scans a bounded slice of orgs from the buffer's
+// queue LIST, identified by a durable cursor in Redis. Per-env entry
+// scan is also bounded so a single pathological env can't extend the
+// pass.
 const DEFAULT_MAX_ENTRIES_PER_ENV = 1000;
+// Max orgs visited per tick. Together with `maxEntriesPerEnv` this
+// caps Redis traffic per pass. One "cycle" (visiting every org once)
+// takes `ceil(N_orgs / cap)` ticks, after which the cursor wraps and a
+// fresh org list is taken.
+const DEFAULT_MAX_ORGS_PER_PASS = 100;
 
 export type StaleSweepConfig = {
   // Entries whose dwell exceeds this threshold are flagged stale. Set
@@ -18,10 +25,21 @@ export type StaleSweepConfig = {
   // matches the cadence in the plan doc.
   staleThresholdMs: number;
   maxEntriesPerEnv?: number;
+  // Hard cap on orgs visited per tick. Bounds the per-pass Redis traffic
+  // and wall-time. Default 100 — at typical fleet sizes one or two
+  // ticks cover everyone; under incident-scale fan-out a full cycle
+  // takes a handful of ticks (~minutes) which is still well below the
+  // staleness signal latency that ops cares about.
+  maxOrgsPerPass?: number;
 };
 
 export type StaleSweepDeps = {
   getBuffer?: () => MollifierBuffer | null;
+  // Durable cursor + per-env counts hash. Required: the sweep is
+  // useless without persistent state across ticks. The webapp wires up
+  // a real `MollifierStaleSweepState`; tests pass one constructed
+  // against the test container.
+  state: StaleSweepStateStore;
   // No `envId` arg — `envId` is a high-cardinality metric attribute and
   // is intentionally not emitted as a metric label. The structured warn
   // log below carries envId for forensic drill-down.
@@ -38,15 +56,32 @@ export type StaleSweepResult = {
   staleCount: number;
 };
 
-// Walks orgs → envs → entries, emitting an OTel counter tick and a
-// structured warning log for each buffer entry whose dwell exceeds the
-// stale threshold. Read-only: the sweep does NOT remove or salvage
-// entries; that decision is deferred to a separate retention-policy
-// change. The signal here exists so ops sees the drainer falling
+// Walks a bounded slice of `orgs → envs → entries`, emitting an OTel
+// counter tick and a structured warning log for each buffer entry whose
+// dwell exceeds the stale threshold. Read-only on the buffer's own
+// state; writes only to the sweep's three dedicated keys
+// (`mollifier:stale_sweep:*`). The sweep does NOT remove or salvage
+// buffer entries; that decision is deferred to a separate retention-
+// policy change. The signal here exists so ops sees the drainer falling
 // behind well before TTL-induced loss kicks in.
+//
+// Sharding contract:
+// - Cursor starts at 0. On cursor=0 the org list is refreshed by
+//   snapshotting `buffer.listOrgs()` into the durable LIST — that is
+//   the cycle's frozen view of orgs to visit.
+// - Each tick consumes up to `maxOrgsPerPass` orgs from the LIST,
+//   advances the cursor, and persists.
+// - When the cursor reaches the end of the LIST it wraps to 0; the next
+//   tick rebuilds the org list, capturing any orgs that joined the
+//   buffer mid-cycle.
+// - The per-env counts HASH carries over across ticks: an env visited
+//   on tick N and not revisited until tick N+M keeps its last-known
+//   stale count in the gauge for that window. This is the price of
+//   sharding — accepted because the alternative (re-scan everything
+//   every tick) does not bound work.
 export async function runStaleSweepOnce(
   config: StaleSweepConfig,
-  deps: StaleSweepDeps = {},
+  deps: StaleSweepDeps,
 ): Promise<StaleSweepResult> {
   const getBuffer = deps.getBuffer ?? getMollifierBuffer;
   const recordStale = deps.recordStaleEntry ?? defaultRecordStaleEntry;
@@ -55,27 +90,40 @@ export async function runStaleSweepOnce(
   const log = deps.logger ?? defaultLogger;
   const now = (deps.now ?? Date.now)();
   const maxEntries = config.maxEntriesPerEnv ?? DEFAULT_MAX_ENTRIES_PER_ENV;
+  const maxOrgsPerPass = config.maxOrgsPerPass ?? DEFAULT_MAX_ORGS_PER_PASS;
 
   const buffer = getBuffer();
   if (!buffer) {
     // Replace any previous snapshot with empty so a previously-paging
     // env doesn't stay latched if mollifier is turned off mid-flight.
+    // Also clear the durable state so a re-enable starts from a clean
+    // slate instead of resuming on a stale cursor.
+    await deps.state.clearAll();
     reportSnapshot(new Map());
     return { orgsScanned: 0, envsScanned: 0, entriesScanned: 0, staleCount: 0 };
   }
 
-  const orgs = await buffer.listOrgs();
+  let cursor = await deps.state.readCursor();
+  if (cursor === 0) {
+    // Fresh cycle — capture the current set of orgs into the frozen
+    // LIST. Any orgs that join after this snapshot wait until the next
+    // cycle to be visited. Acceptable for an observational sweep; the
+    // staleness signal would only fire on entries that have been
+    // dwelling for `staleThresholdMs` anyway, so they're not new.
+    const orgs = await buffer.listOrgs();
+    await deps.state.rebuildOrgList(orgs);
+  }
+
+  const { orgs: slice, total } = await deps.state.readOrgListSlice(
+    cursor,
+    maxOrgsPerPass,
+  );
+
   let envsScanned = 0;
   let entriesScanned = 0;
   let staleCount = 0;
-  // Tracks the stale count per env this pass. Includes zero counts for
-  // envs that have entries but none stale — that's what lets the gauge
-  // drop back to 0 when the drainer catches up. Envs absent from this
-  // map are also absent from the new snapshot, clearing any latched
-  // alerts on envs that have fully drained.
-  const perEnvStale = new Map<string, number>();
-
-  for (const orgId of orgs) {
+
+  for (const orgId of slice) {
     const envs = await buffer.listEnvsForOrg(orgId);
     for (const envId of envs) {
       envsScanned += 1;
@@ -96,18 +144,31 @@ export async function runStaleSweepOnce(
           envStale += 1;
         }
       }
-      perEnvStale.set(envId, envStale);
+      // Persist the per-env count to the durable hash. HSET when stale
+      // > 0, HDEL when it dropped back to zero — the hash is the source
+      // of truth for the gauge snapshot below.
+      await deps.state.setEnvStaleCount(envId, envStale);
       staleCount += envStale;
     }
   }
 
-  reportSnapshot(perEnvStale);
+  // Advance the cursor. If the slice consumed the end of the LIST, wrap
+  // to 0 so the next tick rebuilds the org list and starts a new cycle.
+  const advanced = cursor + slice.length;
+  const newCursor = advanced >= total ? 0 : advanced;
+  await deps.state.writeCursor(newCursor);
+
+  // Emit the snapshot from the durable hash, which carries values for
+  // envs visited in earlier ticks too. This is what makes the gauge
+  // stable across ticks (and across webapp restarts).
+  const snapshot = await deps.state.readAllEnvStaleCounts();
+  reportSnapshot(snapshot);
 
-  return { orgsScanned: orgs.length, envsScanned, entriesScanned, staleCount };
+  return { orgsScanned: slice.length, envsScanned, entriesScanned, staleCount };
 }
 
 export type StaleSweepIntervalHandle = {
-  stop: () => void;
+  stop: () => Promise<void>;
 };
 
 // Production wrapper: schedule `runStaleSweepOnce` on a fixed interval.
@@ -116,7 +177,7 @@ export type StaleSweepIntervalHandle = {
 // overlapping sweeps that all log the same stale entries).
 export function startStaleSweepInterval(
   config: StaleSweepConfig & { intervalMs: number },
-  deps: StaleSweepDeps = {},
+  deps: StaleSweepDeps,
 ): StaleSweepIntervalHandle {
   let stopped = false;
   let inFlight = false;
@@ -141,9 +202,15 @@ export function startStaleSweepInterval(
   }, config.intervalMs);
 
   return {
-    stop: () => {
+    stop: async () => {
       stopped = true;
       clearInterval(timer);
+      // Close the durable-state Redis client if the deps own a real
+      // `MollifierStaleSweepState`. Tests may inject a fake without a
+      // `close()`; guard accordingly.
+      if (deps.state instanceof MollifierStaleSweepState) {
+        await deps.state.close();
+      }
     },
   };
 }

apps/webapp/app/v3/mollifier/mollifierStaleSweepState.server.ts

@@ -0,0 +1,124 @@
+import { createRedisClient, type Redis, type RedisOptions } from "@internal/redis";
+import { Logger } from "@trigger.dev/core/logger";
+
+// Durable per-tick state for the sharded stale sweep. Three Redis keys,
+// all in the `mollifier:` namespace alongside the buffer's own state:
+//
+//   mollifier:stale_sweep:cursor    STRING  next position in org_list (0 = fresh cycle)
+//   mollifier:stale_sweep:org_list  LIST    org IDs frozen at the start of the cycle
+//   mollifier:stale_sweep:counts    HASH    envId -> last-known stale count
+//
+// The state survives webapp restarts: a restarted process picks up the
+// cursor where the previous one left off and re-emits the last-known
+// gauge values immediately, rather than blinking to zero until the next
+// cycle visits each env.
+//
+// Storage is owned by this class rather than added to MollifierBuffer
+// because the keys are sweep-internal — the buffer abstracts the
+// drainer/queue state, this abstracts sweep state. They share a
+// namespace prefix but no API surface.
+
+export interface StaleSweepStateStore {
+  readCursor(): Promise<number>;
+  writeCursor(value: number): Promise<void>;
+  /** Replaces the cycle's frozen org_list. Called at cursor=0. */
+  rebuildOrgList(orgs: string[]): Promise<void>;
+  /** Returns up to `count` org IDs starting at `start`, plus the LIST's total length. */
+  readOrgListSlice(start: number, count: number): Promise<{ orgs: string[]; total: number }>;
+  /** HSET when count > 0, HDEL when count === 0 (so the snapshot reflects current truth). */
+  setEnvStaleCount(envId: string, count: number): Promise<void>;
+  readAllEnvStaleCounts(): Promise<Map<string, number>>;
+  clearAll(): Promise<void>;
+  close(): Promise<void>;
+}
+
+const CURSOR_KEY = "mollifier:stale_sweep:cursor";
+const ORG_LIST_KEY = "mollifier:stale_sweep:org_list";
+const COUNTS_KEY = "mollifier:stale_sweep:counts";
+
+export class MollifierStaleSweepState implements StaleSweepStateStore {
+  private readonly redis: Redis;
+  private readonly logger: Logger;
+
+  constructor(options: { redisOptions: RedisOptions; logger?: Logger }) {
+    this.logger = options.logger ?? new Logger("MollifierStaleSweepState", "debug");
+    this.redis = createRedisClient(
+      { ...options.redisOptions, maxRetriesPerRequest: 20 },
+      {
+        onError: (error) => {
+          this.logger.error("MollifierStaleSweepState redis client error:", { error });
+        },
+      },
+    );
+  }
+
+  async readCursor(): Promise<number> {
+    const raw = await this.redis.get(CURSOR_KEY);
+    if (raw === null) return 0;
+    const n = Number.parseInt(raw, 10);
+    return Number.isFinite(n) && n >= 0 ? n : 0;
+  }
+
+  async writeCursor(value: number): Promise<void> {
+    await this.redis.set(CURSOR_KEY, String(value));
+  }
+
+  async rebuildOrgList(orgs: string[]): Promise<void> {
+    // DEL + RPUSH in a pipeline — close enough to atomic for an
+    // observational sweep (the inFlight guard at startStaleSweepInterval
+    // serialises sweep passes; nothing else writes these keys).
+    const pipeline = this.redis.pipeline();
+    pipeline.del(ORG_LIST_KEY);
+    if (orgs.length > 0) {
+      pipeline.rpush(ORG_LIST_KEY, ...orgs);
+    }
+    await pipeline.exec();
+  }
+
+  async readOrgListSlice(
+    start: number,
+    count: number,
+  ): Promise<{ orgs: string[]; total: number }> {
+    const pipeline = this.redis.pipeline();
+    pipeline.lrange(ORG_LIST_KEY, start, start + count - 1);
+    pipeline.llen(ORG_LIST_KEY);
+    const results = await pipeline.exec();
+    if (!results) return { orgs: [], total: 0 };
+    const [lrangeErr, lrangeRes] = results[0] as [Error | null, string[] | null];
+    const [llenErr, llenRes] = results[1] as [Error | null, number | null];
+    if (lrangeErr || llenErr) {
+      this.logger.error("MollifierStaleSweepState.readOrgListSlice failed", {
+        lrangeErr: lrangeErr?.message,
+        llenErr: llenErr?.message,
+      });
+      return { orgs: [], total: 0 };
+    }
+    return { orgs: lrangeRes ?? [], total: llenRes ?? 0 };
+  }
+
+  async setEnvStaleCount(envId: string, count: number): Promise<void> {
+    if (count > 0) {
+      await this.redis.hset(COUNTS_KEY, envId, String(count));
+    } else {
+      await this.redis.hdel(COUNTS_KEY, envId);
+    }
+  }
+
+  async readAllEnvStaleCounts(): Promise<Map<string, number>> {
+    const raw = await this.redis.hgetall(COUNTS_KEY);
+    const out = new Map<string, number>();
+    for (const [envId, value] of Object.entries(raw)) {
+      const n = Number.parseInt(value, 10);
+      if (Number.isFinite(n)) out.set(envId, n);
+    }
+    return out;
+  }
+
+  async clearAll(): Promise<void> {
+    await this.redis.del(CURSOR_KEY, ORG_LIST_KEY, COUNTS_KEY);
+  }
+
+  async close(): Promise<void> {
+    await this.redis.quit();
+  }
+}

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

@@ -5,6 +5,7 @@ import {
   startStaleSweepInterval,
   type StaleSweepIntervalHandle,
 } from "./mollifier/mollifierStaleSweep.server";
+import { MollifierStaleSweepState } from "./mollifier/mollifierStaleSweepState.server";
 
 declare global {
   // eslint-disable-next-line no-var
@@ -35,13 +36,38 @@ export function initMollifierStaleSweepWorker(): void {
     staleThresholdMs: env.TRIGGER_MOLLIFIER_STALE_SWEEP_THRESHOLD_MS,
   });
 
-  const handle = startStaleSweepInterval({
-    intervalMs: env.TRIGGER_MOLLIFIER_STALE_SWEEP_INTERVAL_MS,
-    staleThresholdMs: env.TRIGGER_MOLLIFIER_STALE_SWEEP_THRESHOLD_MS,
+  // Construct the sweep's durable-state Redis client using the same
+  // mollifier-Redis credentials as the buffer. Keeping this client
+  // separate from the buffer's own client keeps state ownership clean:
+  // the buffer abstracts queue/entry state, this abstracts sweep state.
+  const state = new MollifierStaleSweepState({
+    redisOptions: {
+      keyPrefix: "",
+      host: env.TRIGGER_MOLLIFIER_REDIS_HOST,
+      port: env.TRIGGER_MOLLIFIER_REDIS_PORT,
+      username: env.TRIGGER_MOLLIFIER_REDIS_USERNAME,
+      password: env.TRIGGER_MOLLIFIER_REDIS_PASSWORD,
+      enableAutoPipelining: true,
+      ...(env.TRIGGER_MOLLIFIER_REDIS_TLS_DISABLED === "true" ? {} : { tls: {} }),
+    },
   });
 
-  signalsEmitter.on("SIGTERM", handle.stop);
-  signalsEmitter.on("SIGINT", handle.stop);
+  const handle = startStaleSweepInterval(
+    {
+      intervalMs: env.TRIGGER_MOLLIFIER_STALE_SWEEP_INTERVAL_MS,
+      staleThresholdMs: env.TRIGGER_MOLLIFIER_STALE_SWEEP_THRESHOLD_MS,
+    },
+    { state },
+  );
+
+  // `handle.stop` is now async (it closes the Redis client). The signals
+  // emitter swallows promise rejections from listeners, so wrap it in a
+  // void-returning shim to be explicit about discarding the promise.
+  const onShutdown = (): void => {
+    void handle.stop();
+  };
+  signalsEmitter.on("SIGTERM", onShutdown);
+  signalsEmitter.on("SIGINT", onShutdown);
   global.__mollifierStaleSweepRegistered__ = true;
   global.__mollifierStaleSweepHandle__ = handle;
 }