feat(webapp): mollifier stale-entry sweep + OTel signal

Without an external signal that the drainer is falling behind, a stuck
or offline drainer drives the buffer toward the entry-hash TTL line
and runs vanish silently — no PG row, no log, no dashboard
indication. Add a periodic read-only sweep over the buffer's queue
ZSETs that emits a `mollifier.stale_entries` OTel counter and a
structured `mollifier.stale_entry` warning log for each entry whose
dwell exceeds the configured threshold.

Independent of the drainer (its own gate +
`TRIGGER_MOLLIFIER_STALE_SWEEP_ENABLED`) so an entirely offline
drainer is exactly when the sweep is most useful. Defaults: interval
5min, threshold half of `entryTtlSeconds`, hard cap of 1000 entries
per env per pass.

Sweep is strictly read-only — does not remove or salvage entries.
The retention-policy question (drop the entry TTL entirely vs raise
it vs pre-TTL salvage) is intentionally deferred to a separate
change; this commit gets the signal in place first.

Tested with a real `MollifierBuffer` (testcontainers): stale entries
flagged, fresh entries left alone, multi-org scan walks every queue.
Manually verified end-to-end: with a 10s interval + 2s threshold,
each tick logs the buffered run with growing dwellMs as expected.

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

Author: d-cs

.server-changes/mollifier-stale-entry-sweep.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: feature
+---
+
+Periodic mollifier stale-entry sweep. Scans the buffer's queue ZSETs every `TRIGGER_MOLLIFIER_STALE_SWEEP_INTERVAL_MS` (default 5min); entries whose dwell exceeds `TRIGGER_MOLLIFIER_STALE_SWEEP_THRESHOLD_MS` (default half of `entryTtlSeconds`) emit a `mollifier.stale_entries` OTel counter tick plus a structured `mollifier.stale_entry` warning log. Read-only — the sweep does not remove or salvage entries; that decision is deferred to a separate retention-policy change. Gives ops a paging signal when the drainer is offline or falling behind before TTL-induced silent loss kicks in.

apps/webapp/app/entry.server.tsx

@@ -7,6 +7,7 @@ import { renderToPipeableStream } from "react-dom/server";
 import { PassThrough } from "stream";
 import * as Worker from "~/services/worker.server";
 import { initMollifierDrainerWorker } from "~/v3/mollifierDrainerWorker.server";
+import { initMollifierStaleSweepWorker } from "~/v3/mollifierStaleSweepWorker.server";
 import { bootstrap } from "./bootstrap";
 import { LocaleContextProvider } from "./components/primitives/LocaleProvider";
 import {
@@ -249,6 +250,7 @@ Worker.init().catch((error) => {
 });
 
 initMollifierDrainerWorker();
+initMollifierStaleSweepWorker();
 
 bootstrap().catch((error) => {
   logError(error);

apps/webapp/app/env.server.ts

@@ -1098,6 +1098,25 @@ const EnvironmentSchema = z
     TRIGGER_MOLLIFIER_DRAIN_SHUTDOWN_TIMEOUT_MS: z.coerce.number().int().positive().default(30_000),
     TRIGGER_MOLLIFIER_DRAIN_MAX_ORGS_PER_TICK: z.coerce.number().int().positive().default(500),
 
+    // Periodic sweep that scans buffer queue ZSETs for entries whose
+    // dwell exceeds the stale threshold. Independent of the drainer —
+    // its job is exactly to make a stuck/offline drainer visible to
+    // ops. Defaults: enabled when the mollifier is enabled, run every
+    // 5 minutes, flag entries with dwell > half of entryTtlSeconds.
+    TRIGGER_MOLLIFIER_STALE_SWEEP_ENABLED: z
+      .string()
+      .default(process.env.TRIGGER_MOLLIFIER_ENABLED ?? "0"),
+    TRIGGER_MOLLIFIER_STALE_SWEEP_INTERVAL_MS: z.coerce
+      .number()
+      .int()
+      .positive()
+      .default(5 * 60_000),
+    TRIGGER_MOLLIFIER_STALE_SWEEP_THRESHOLD_MS: z.coerce
+      .number()
+      .int()
+      .positive()
+      .optional(),
+
     BATCH_TRIGGER_PROCESS_JOB_VISIBILITY_TIMEOUT_MS: z.coerce
       .number()
       .int()

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

@@ -0,0 +1,126 @@
+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";
+
+// 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
+// unboundedly long.
+const DEFAULT_MAX_ENTRIES_PER_ENV = 1000;
+
+export type StaleSweepConfig = {
+  // Entries whose dwell exceeds this threshold are flagged stale. Set
+  // it well below `entryTtlSeconds * 1000` so ops have lead time before
+  // TTL-induced silent loss; the default (half of entryTtlSeconds)
+  // matches the cadence in the plan doc.
+  staleThresholdMs: number;
+  maxEntriesPerEnv?: number;
+};
+
+export type StaleSweepDeps = {
+  getBuffer?: () => MollifierBuffer | null;
+  recordStaleEntry?: (envId: string) => void;
+  logger?: { warn: (message: string, fields: Record<string, unknown>) => void };
+  now?: () => number;
+};
+
+export type StaleSweepResult = {
+  orgsScanned: number;
+  envsScanned: number;
+  entriesScanned: number;
+  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
+// behind well before TTL-induced loss kicks in.
+export async function runStaleSweepOnce(
+  config: StaleSweepConfig,
+  deps: StaleSweepDeps = {},
+): Promise<StaleSweepResult> {
+  const getBuffer = deps.getBuffer ?? getMollifierBuffer;
+  const recordStale = deps.recordStaleEntry ?? defaultRecordStaleEntry;
+  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) {
+    return { orgsScanned: 0, envsScanned: 0, entriesScanned: 0, staleCount: 0 };
+  }
+
+  const orgs = await buffer.listOrgs();
+  let envsScanned = 0;
+  let entriesScanned = 0;
+  let staleCount = 0;
+
+  for (const orgId of orgs) {
+    const envs = await buffer.listEnvsForOrg(orgId);
+    for (const envId of envs) {
+      envsScanned += 1;
+      const entries = await buffer.listEntriesForEnv(envId, maxEntries);
+      for (const entry of entries) {
+        entriesScanned += 1;
+        const dwellMs = now - entry.createdAt.getTime();
+        if (dwellMs > config.staleThresholdMs) {
+          recordStale(envId);
+          log.warn("mollifier.stale_entry", {
+            runId: entry.runId,
+            envId,
+            orgId,
+            dwellMs,
+            staleThresholdMs: config.staleThresholdMs,
+          });
+          staleCount += 1;
+        }
+      }
+    }
+  }
+
+  return { orgsScanned: orgs.length, envsScanned, entriesScanned, staleCount };
+}
+
+export type StaleSweepIntervalHandle = {
+  stop: () => void;
+};
+
+// Production wrapper: schedule `runStaleSweepOnce` on a fixed interval.
+// One pass at a time — if a sweep is still running when the timer fires
+// the next tick is skipped (a backed-up Redis would otherwise queue
+// overlapping sweeps that all log the same stale entries).
+export function startStaleSweepInterval(
+  config: StaleSweepConfig & { intervalMs: number },
+  deps: StaleSweepDeps = {},
+): StaleSweepIntervalHandle {
+  let stopped = false;
+  let inFlight = false;
+
+  const tick = async () => {
+    if (stopped || inFlight) return;
+    inFlight = true;
+    try {
+      await runStaleSweepOnce(config, deps);
+    } catch (err) {
+      const log = deps.logger ?? defaultLogger;
+      log.warn("mollifier.stale_sweep.failed", {
+        err: err instanceof Error ? err.message : String(err),
+      });
+    } finally {
+      inFlight = false;
+    }
+  };
+
+  const timer = setInterval(() => {
+    void tick();
+  }, config.intervalMs);
+
+  return {
+    stop: () => {
+      stopped = true;
+      clearInterval(timer);
+    },
+  };
+}

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

@@ -33,6 +33,23 @@ export function recordRealtimeBufferedSubscription(envId: string): void {
   realtimeBufferedSubscriptionsCounter.add(1, { envId });
 }
 
+// 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.
+export const staleEntriesCounter = meter.createCounter(
+  "mollifier.stale_entries",
+  {
+    description:
+      "Mollifier buffer entries whose dwell exceeds the stale threshold (per sweep pass)",
+  },
+);
+
+export function recordStaleEntry(envId: string): void {
+  staleEntriesCounter.add(1, { envId });
+}
+
 // 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/app/v3/mollifierStaleSweepWorker.server.ts

@@ -0,0 +1,54 @@
+import { env } from "~/env.server";
+import { logger } from "~/services/logger.server";
+import { signalsEmitter } from "~/services/signals.server";
+import {
+  startStaleSweepInterval,
+  type StaleSweepIntervalHandle,
+} from "./mollifier/mollifierStaleSweep.server";
+
+declare global {
+  // eslint-disable-next-line no-var
+  var __mollifierStaleSweepRegistered__: boolean | undefined;
+  // eslint-disable-next-line no-var
+  var __mollifierStaleSweepHandle__: StaleSweepIntervalHandle | undefined;
+}
+
+/**
+ * Bootstraps the mollifier stale-entry sweep.
+ *
+ * Independent of the drainer — its purpose is to alert when entries are
+ * piling up despite the drainer being supposedly healthy, so it runs
+ * any time the mollifier itself is enabled (gated separately from
+ * `TRIGGER_MOLLIFIER_DRAINER_ENABLED`). The sweep is read-only: it
+ * counts and logs stale entries but does not remove or salvage them.
+ *
+ * The Remix dev server re-evaluates `entry.server.tsx` on every change,
+ * so the registration guard + handle cache make the bootstrap
+ * idempotent across hot reloads.
+ */
+export function initMollifierStaleSweepWorker(): void {
+  if (env.TRIGGER_MOLLIFIER_STALE_SWEEP_ENABLED !== "1") return;
+  if (global.__mollifierStaleSweepRegistered__) return;
+
+  // Default the threshold to half of `entryTtlSeconds`, mirroring the
+  // plan doc's cadence. Operators wanting an earlier or later signal
+  // can set it explicitly.
+  const staleThresholdMs =
+    env.TRIGGER_MOLLIFIER_STALE_SWEEP_THRESHOLD_MS ??
+    Math.floor(env.TRIGGER_MOLLIFIER_ENTRY_TTL_S * 1000 * 0.5);
+
+  logger.debug("Initializing mollifier stale-entry sweep", {
+    intervalMs: env.TRIGGER_MOLLIFIER_STALE_SWEEP_INTERVAL_MS,
+    staleThresholdMs,
+  });
+
+  const handle = startStaleSweepInterval({
+    intervalMs: env.TRIGGER_MOLLIFIER_STALE_SWEEP_INTERVAL_MS,
+    staleThresholdMs,
+  });
+
+  signalsEmitter.on("SIGTERM", handle.stop);
+  signalsEmitter.on("SIGINT", handle.stop);
+  global.__mollifierStaleSweepRegistered__ = true;
+  global.__mollifierStaleSweepHandle__ = handle;
+}