feat(webapp): alertable gauge for mollifier stale-entry signal

The mollifier.stale_entries counter from the previous commit reflects
sweep-tick events, not stable state. A single stuck entry observed
across N ticks contributes N events, so a rate() query is
proportional to (stuck-entry-count × scan-frequency), not "how many
entries are stale right now". Useful for historical views but the
wrong shape for ops alerts.

Add a companion observable gauge `mollifier.stale_entries.current`
with `{envId}` attribute. The sweep emits a per-env snapshot on each
pass (including zero counts for envs whose stale entries cleared),
and an OTel batch-observable callback exposes the latest snapshot to
the metric exporter on every scrape. Recommended alert:

  mollifier_stale_entries_current{envId=...} > 0 for 5m

The snapshot replaces (not merges) so an env that paged on a
previous sweep clears when the drainer catches up, instead of
staying latched at the last stale count. Test seam captures the
snapshot to verify per-env counts and the clear-on-drain behaviour.

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

Author: d-cs

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

@@ -1,7 +1,10 @@
 import type { MollifierBuffer } from "@trigger.dev/redis-worker";
 import { logger as defaultLogger } from "~/services/logger.server";
 import { getMollifierBuffer } from "./mollifierBuffer.server";
-import { recordStaleEntry as defaultRecordStaleEntry } from "./mollifierTelemetry.server";
+import {
+  recordStaleEntry as defaultRecordStaleEntry,
+  reportStaleEntrySnapshot as defaultReportStaleEntrySnapshot,
+} from "./mollifierTelemetry.server";
 
 // One pass of the sweep scans every env's queue ZSET. The per-env page
 // is bounded so a single pathological env can't make the sweep run
@@ -20,6 +23,7 @@ export type StaleSweepConfig = {
 export type StaleSweepDeps = {
   getBuffer?: () => MollifierBuffer | null;
   recordStaleEntry?: (envId: string) => void;
+  reportStaleEntrySnapshot?: (snapshot: Map<string, number>) => void;
   logger?: { warn: (message: string, fields: Record<string, unknown>) => void };
   now?: () => number;
 };
@@ -43,24 +47,36 @@ export async function runStaleSweepOnce(
 ): Promise<StaleSweepResult> {
   const getBuffer = deps.getBuffer ?? getMollifierBuffer;
   const recordStale = deps.recordStaleEntry ?? defaultRecordStaleEntry;
+  const reportSnapshot =
+    deps.reportStaleEntrySnapshot ?? defaultReportStaleEntrySnapshot;
   const log = deps.logger ?? defaultLogger;
   const now = (deps.now ?? Date.now)();
   const maxEntries = config.maxEntriesPerEnv ?? DEFAULT_MAX_ENTRIES_PER_ENV;
 
   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.
+    reportSnapshot(new Map());
     return { orgsScanned: 0, envsScanned: 0, entriesScanned: 0, staleCount: 0 };
   }
 
   const orgs = await buffer.listOrgs();
   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) {
     const envs = await buffer.listEnvsForOrg(orgId);
     for (const envId of envs) {
       envsScanned += 1;
+      let envStale = 0;
       const entries = await buffer.listEntriesForEnv(envId, maxEntries);
       for (const entry of entries) {
         entriesScanned += 1;
@@ -74,12 +90,16 @@ export async function runStaleSweepOnce(
             dwellMs,
             staleThresholdMs: config.staleThresholdMs,
           });
-          staleCount += 1;
+          envStale += 1;
         }
       }
+      perEnvStale.set(envId, envStale);
+      staleCount += envStale;
     }
   }
 
+  reportSnapshot(perEnvStale);
+
   return { orgsScanned: orgs.length, envsScanned, entriesScanned, staleCount };
 }
 

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

@@ -35,9 +35,10 @@ export function recordRealtimeBufferedSubscription(envId: string): void {
 
 // Counts buffer entries that have been waiting in the queue ZSET longer
 // than the configured stale threshold (typically half of entryTtlSeconds).
-// Climbing in lockstep with the queue depth means the drainer is offline
-// or falling behind — alerting hooks into this counter give ops a paging
-// signal before TTL-induced silent loss kicks in.
+// Useful for historical "stale events over time" views, but not directly
+// alertable on its own — a single stuck entry observed by N sweep ticks
+// adds N to the counter, so `rate()` over an alerting window reflects
+// (entries × ticks), not "entries that are stale right now".
 export const staleEntriesCounter = meter.createCounter(
   "mollifier.stale_entries",
   {
@@ -50,6 +51,41 @@ export function recordStaleEntry(envId: string): void {
   staleEntriesCounter.add(1, { envId });
 }
 
+// Alertable signal: the count of stale entries observed by the latest
+// sweep, per env. The sweep snapshots the full per-env picture on each
+// pass (including zeros for envs that no longer have any stale entries)
+// so an env that was paging can clear when the drainer catches up
+// instead of staying latched. Recommended alert:
+//   mollifier_stale_entries_current{envId=...} > 0 for 5m
+export const staleEntriesGauge = meter.createObservableGauge(
+  "mollifier.stale_entries.current",
+  {
+    description:
+      "Buffer entries whose dwell exceeds the stale threshold, as observed by the latest sweep pass",
+  },
+);
+
+const latestStaleSnapshot = new Map<string, number>();
+
+export function reportStaleEntrySnapshot(snapshot: Map<string, number>): void {
+  // Replace, don't merge — envs absent from the new snapshot have either
+  // drained or no longer exist; leaving their last value cached would
+  // keep alerts latched forever.
+  latestStaleSnapshot.clear();
+  for (const [envId, count] of snapshot) {
+    latestStaleSnapshot.set(envId, count);
+  }
+}
+
+meter.addBatchObservableCallback(
+  (result) => {
+    for (const [envId, count] of latestStaleSnapshot) {
+      result.observe(staleEntriesGauge, count, { envId });
+    }
+  },
+  [staleEntriesGauge],
+);
+
 // Electric SQL's shape-stream protocol adds a `handle=` query param on
 // every reconnect after the initial GET. Gating the realtime-buffered
 // log/counter on its absence keeps the signal at one tick per

apps/webapp/test/mollifierStaleSweep.test.ts

@@ -15,14 +15,22 @@ const SNAPSHOT = {
 
 function spyDeps() {
   const recordedStaleEnvIds: string[] = [];
+  const snapshots: Array<Map<string, number>> = [];
   const warnings: Array<{ message: string; fields: Record<string, unknown> }> = [];
   return {
     recordedStaleEnvIds,
+    snapshots,
     warnings,
     deps: {
       recordStaleEntry: (envId: string) => {
         recordedStaleEnvIds.push(envId);
       },
+      reportStaleEntrySnapshot: (snapshot: Map<string, number>) => {
+        // Clone so post-sweep assertions see what was reported *at that
+        // call site*, not whatever subsequent passes mutate the source
+        // map into.
+        snapshots.push(new Map(snapshot));
+      },
       logger: {
         warn: (message: string, fields: Record<string, unknown>) => {
           warnings.push({ message, fields });
@@ -37,7 +45,7 @@ describe("runStaleSweepOnce — unit", () => {
     // Mirrors the prod gate: if TRIGGER_MOLLIFIER_ENABLED=0 the buffer
     // singleton is null and the sweep is a no-op. We don't want it to
     // emit a metric (or throw) just because mollifier is disabled.
-    const { deps, recordedStaleEnvIds, warnings } = spyDeps();
+    const { deps, recordedStaleEnvIds, warnings, snapshots } = spyDeps();
     const result = await runStaleSweepOnce(
       { staleThresholdMs: 1000 },
       { ...deps, getBuffer: () => null },
@@ -50,6 +58,10 @@ describe("runStaleSweepOnce — unit", () => {
     });
     expect(recordedStaleEnvIds).toEqual([]);
     expect(warnings).toEqual([]);
+    // An empty snapshot is still reported so any previously-paging env
+    // (from a prior sweep before mollifier was disabled) clears.
+    expect(snapshots).toHaveLength(1);
+    expect(snapshots[0].size).toBe(0);
   });
 });
 
@@ -86,7 +98,7 @@ describe("runStaleSweepOnce — testcontainers", () => {
         // the threshold without actually waiting in real time.
         const futureNow = Date.now() + 5 * 60 * 1000;
 
-        const { deps, recordedStaleEnvIds, warnings } = spyDeps();
+        const { deps, recordedStaleEnvIds, warnings, snapshots } = spyDeps();
         const result = await runStaleSweepOnce(
           { staleThresholdMs: 60 * 1000 },
           {
@@ -110,6 +122,42 @@ describe("runStaleSweepOnce — testcontainers", () => {
           expect(w.fields.staleThresholdMs).toBe(60 * 1000);
           expect(w.fields.dwellMs).toBeGreaterThan(60 * 1000);
         }
+        // Snapshot drives the alertable gauge — env_a has 2 stale
+        // entries, env_b has 1. Both must appear so a future alert can
+        // identify which env is paging.
+        expect(snapshots).toHaveLength(1);
+        expect(Object.fromEntries(snapshots[0])).toEqual({
+          env_a: 2,
+          env_b: 1,
+        });
+      } finally {
+        await buffer.close();
+      }
+    },
+  );
+
+  redisTest(
+    "snapshot reports zero for envs that have entries but none stale (clears latched alerts)",
+    async ({ redisOptions }) => {
+      // Critical for alert behaviour: a previous sweep reported env_a
+      // stale, alert fired, drainer caught up. The next sweep must
+      // report `env_a -> 0` so the gauge drops below the alert
+      // threshold instead of staying latched at the last stale value.
+      const buffer = new MollifierBuffer({ redisOptions, entryTtlSeconds: 60 });
+      try {
+        await buffer.accept({
+          runId: "run_just_arrived",
+          envId: "env_a",
+          orgId: "org_1",
+          payload: JSON.stringify(SNAPSHOT),
+        });
+        const { deps, snapshots } = spyDeps();
+        await runStaleSweepOnce(
+          { staleThresholdMs: 60 * 1000 },
+          { ...deps, getBuffer: () => buffer },
+        );
+        expect(snapshots).toHaveLength(1);
+        expect(Object.fromEntries(snapshots[0])).toEqual({ env_a: 0 });
       } finally {
         await buffer.close();
       }