fix(webapp): propagate Redis read failures from mollifier stale-sweep slice

`MollifierStaleSweepState.readOrgListSlice` was logging-and-returning
`{ orgs: [], total: 0 }` on pipeline-abort or per-result Redis errors.
The caller (`runStaleSweepOnce`) treated that as a clean empty cycle:
wrote cursor=0, reconciled visited envs against the empty result, and
cleared the stale-entry gauge — silencing the alerts the sweep exists
to raise.

Re-throw the underlying error instead. The interval wrapper's catch
logs `stale_sweep.failed` for the failed tick; durable cursor + counts
hash stay untouched so the gauge keeps reporting its last-known value
until a healthy tick repopulates it.

New unit test on the caller contract pins this: error propagates,
cursor stays at its seeded value, counts hash retains the seeded env,
no snapshot is reported.

Separately documents a known limitation in the .server-changes entry:
the sweep runs per-webapp-instance, so its stale-entry counter
multiplies by N webapps in HA until a distributed lease lands as a
follow-up. The system is HA-safe (Redis ops are atomic, no torn
state); only the metric output is inflated.

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

Author: d-cs

.server-changes/mollifier-drainer-replay.md

@@ -3,4 +3,4 @@ area: webapp
 type: feature
 ---
 
-Mollifier drainer replay: replay buffered entries into `engine.trigger`, stale-entry sweep, a drainer-health gauge, and run-engine cancelled/failed run APIs.
+Mollifier drainer replay: replay buffered entries into `engine.trigger`, stale-entry sweep, a drainer-health gauge, and run-engine cancelled/failed run APIs. Known limitation: stale-sweep runs per-webapp instance, so stale-entry counter metrics multiply by N webapps in HA until a distributed lease lands as follow-up.

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

@@ -101,15 +101,27 @@ export class MollifierStaleSweepState implements StaleSweepStateStore {
     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 };
+    // `pipeline.exec()` returning null is the abort-on-broken-pipe path.
+    // Surface it as a thrown error — the previous `return { orgs: [], total: 0 }`
+    // looked indistinguishable from a genuinely empty org list to the
+    // caller (`runStaleSweepOnce`), which then wrote cursor=0, reconciled
+    // visited envs against the empty result, and cleared the stale-entry
+    // gauge. That hid real Redis problems and silenced the alerts the
+    // sweep exists to raise.
+    if (!results) {
+      throw new Error("MollifierStaleSweepState.readOrgListSlice: pipeline.exec returned null");
+    }
     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 };
+      // Same reasoning as the null-result path above — propagate the
+      // failure so the sweep's interval wrapper records a failed cycle
+      // and the durable cursor / counts hash stay untouched.
+      throw lrangeErr ?? llenErr ?? new Error("MollifierStaleSweepState.readOrgListSlice failed");
     }
     return { orgs: lrangeRes ?? [], total: llenRes ?? 0 };
   }

apps/webapp/test/mollifierStaleSweep.test.ts

@@ -117,6 +117,62 @@ describe("runStaleSweepOnce — unit", () => {
     expect(snapshots).toHaveLength(1);
     expect(snapshots[0].size).toBe(0);
   });
+
+  it("surfaces readOrgListSlice failures and leaves durable state untouched", async () => {
+    // Regression: previously a Redis read failure inside
+    // `readOrgListSlice` returned `{ orgs: [], total: 0 }` and the
+    // sweep treated that as a clean empty cycle — writing cursor=0,
+    // reconciling visited envs against the empty result, and CLEARING
+    // the stale-entry gauge. That silenced the very alerts the sweep
+    // exists to raise. The fix re-throws; the caller (this function
+    // and the interval wrapper above it) must NOT mutate cursor or
+    // counts when readOrgListSlice fails.
+    const state = makeFakeState();
+    // Seed durable state so we can assert it isn't touched on failure.
+    await state.writeCursor(42);
+    await state.setEnvStaleCount("env_seed", 7);
+    await state.rebuildOrgList(["org_pre"]);
+    // Inject a failure on the very next slice read.
+    const readErr = new Error("Redis read failed");
+    let readAttempts = 0;
+    const failingState = {
+      ...state,
+      readOrgListSlice: async (start: number, count: number) => {
+        readAttempts += 1;
+        throw readErr;
+      },
+    };
+    const spies = spyDeps();
+    const buffer = {
+      listOrgs: async () => ["org_pre"],
+      listEnvsForOrg: async () => [],
+      listEntriesForEnv: async () => [],
+    } as unknown as MollifierBuffer;
+
+    await expect(
+      runStaleSweepOnce(
+        { staleThresholdMs: 60_000, maxOrgsPerPass: 10 },
+        {
+          ...spies.deps,
+          state: failingState,
+          getBuffer: () => buffer,
+          now: () => Date.now(),
+        },
+      ),
+    ).rejects.toThrow("Redis read failed");
+
+    expect(readAttempts).toBe(1);
+    // Cursor untouched (still the seeded 42, not reset to 0).
+    expect(await state.readCursor()).toBe(42);
+    // Counts hash untouched — the seeded env's count survives the
+    // failed cycle so the gauge keeps reporting its last-known value.
+    const counts = await state.readAllEnvStaleCounts();
+    expect(counts.get("env_seed")).toBe(7);
+    // No snapshot was reported because the function threw before
+    // reaching reportStaleEntrySnapshot.
+    expect(spies.snapshots).toHaveLength(0);
+    expect(spies.staleEntryCount).toBe(0);
+  });
 });
 
 describe("runStaleSweepOnce — testcontainers", () => {