fix(webapp): stale-sweep stop() awaits in-flight tick before closing Redis

Devin review on PR #3754: `stop()` previously called `deps.state.close()`
immediately after `clearInterval`, but `tick()` only checks `stopped` at
entry. A tick that was already past that guard would keep making
`state.*` calls against an ioredis client that `stop()` had already
`quit()`ed — those calls would throw, the tick's own try/catch would
swallow them as `mollifier.stale_sweep.failed` warnings, and every
graceful shutdown would emit spurious noise.

Track the current tick promise as `currentTick`. `stop()` awaits it (if
present) before invoking `state.close()`, so the tick's last state call
lands BEFORE the Redis client is quit. The tick's own try/catch handles
the (unexpected) case where it rejects; the await in `stop()` is solely
for ordering.

Also drop the `instanceof MollifierStaleSweepState` guard around
`state.close()` — `close()` is part of the `StaleSweepStateStore`
contract, so unconditional invocation is correct. Test fake states
implement `close()` as a no-op.

Test: `stop() waits for an in-flight tick to finish before closing the
state` — gates a fake state's `readCursor` on a Deferred, kicks off the
interval, waits for the tick to start, then races `stop()` against the
gate. Asserts the stop promise stays unresolved while the tick is
mid-flight and that the tick's final state operation lands before
`close()`.

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

Author: d-cs

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

@@ -198,20 +198,33 @@ export function startStaleSweepInterval(
 ): StaleSweepIntervalHandle {
   let stopped = false;
   let inFlight = false;
+  // Tracks the current tick so `stop()` can await it before closing the
+  // state's Redis client. Without this, a tick that's already past the
+  // `stopped` guard at entry would continue making `state.*` calls
+  // against an ioredis client that `stop()` has already `quit()`ed,
+  // raising errors that the tick's own try/catch then logs as
+  // `mollifier.stale_sweep.failed` warnings — spurious noise on every
+  // graceful shutdown.
+  let currentTick: Promise<void> | null = null;
 
   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 run = (async () => {
+      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;
+        currentTick = null;
+      }
+    })();
+    currentTick = run;
+    await run;
   };
 
   const timer = setInterval(() => {
@@ -222,12 +235,22 @@ export function startStaleSweepInterval(
     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();
+      // Drain any tick that started before `stopped` flipped. Its
+      // `state.*` calls must land before we close the Redis client.
+      if (currentTick) {
+        try {
+          await currentTick;
+        } catch {
+          // tick has its own catch — this await is just to ensure
+          // ordering, not to surface errors that have already been
+          // logged inside the tick.
+        }
       }
+      // Close the state's underlying resource. The `close()` method is
+      // part of the `StaleSweepStateStore` contract — production's
+      // `MollifierStaleSweepState` shuts down its ioredis client; fake
+      // test states implement a no-op.
+      await deps.state.close();
     },
   };
 }

apps/webapp/test/mollifierStaleSweep.test.ts

@@ -4,7 +4,10 @@ import { MollifierBuffer } from "@trigger.dev/redis-worker";
 
 vi.mock("~/db.server", () => ({ prisma: {}, $replica: {} }));
 
-import { runStaleSweepOnce } from "~/v3/mollifier/mollifierStaleSweep.server";
+import {
+  runStaleSweepOnce,
+  startStaleSweepInterval,
+} from "~/v3/mollifier/mollifierStaleSweep.server";
 import { MollifierStaleSweepState } from "~/v3/mollifier/mollifierStaleSweepState.server";
 
 const SNAPSHOT = {
@@ -799,3 +802,119 @@ describe("MollifierStaleSweepState — direct unit tests", () => {
     },
   );
 });
+
+describe("startStaleSweepInterval — lifecycle", () => {
+  it("stop() waits for an in-flight tick to finish before closing the state", async () => {
+    // Devin's BUG report on PR #3754: `stop()` previously called
+    // `deps.state.close()` immediately after `clearInterval`, but the
+    // `tick` function only checks `stopped` at entry. A tick that was
+    // already past that check would keep making `state.*` Redis calls
+    // against a now-closed ioredis client, throw, get caught by tick's
+    // own try/catch, and log a `mollifier.stale_sweep.failed` warning
+    // for every graceful shutdown.
+    //
+    // The fix tracks the current tick promise so `stop()` can await it
+    // before closing. This test pins that order by gating one of the
+    // tick's state calls on a Deferred — until we resolve it, the tick
+    // can't progress, and `stop()` must hang in the meantime.
+    let resolveGate: () => void = () => {};
+    const gate = new Promise<void>((r) => {
+      resolveGate = r;
+    });
+
+    const callOrder: string[] = [];
+    let closeCalled = false;
+    const state = {
+      readCursor: async () => {
+        callOrder.push("readCursor:start");
+        await gate;
+        callOrder.push("readCursor:end");
+        return 0;
+      },
+      writeCursor: async () => {
+        callOrder.push("writeCursor");
+      },
+      rebuildOrgList: async () => {
+        callOrder.push("rebuildOrgList");
+      },
+      readOrgListSlice: async () => {
+        callOrder.push("readOrgListSlice");
+        // Return zero orgs so the org loop is a no-op — we only care
+        // about ordering of state calls vs close, not the work.
+        return { orgs: [] as string[], total: 0 };
+      },
+      setEnvStaleCount: async () => {
+        callOrder.push("setEnvStaleCount");
+      },
+      readAllEnvStaleCounts: async () => {
+        callOrder.push("readAllEnvStaleCounts");
+        return new Map<string, number>();
+      },
+      markEnvVisited: async () => {
+        callOrder.push("markEnvVisited");
+      },
+      reconcileVisited: async () => {
+        callOrder.push("reconcileVisited");
+      },
+      clearAll: async () => {
+        callOrder.push("clearAll");
+      },
+      close: async () => {
+        callOrder.push("close");
+        closeCalled = true;
+      },
+    };
+
+    const fakeBuffer = {
+      listOrgs: async () => [],
+      listEnvsForOrg: async () => [],
+      listEntriesForEnv: async () => [],
+    } as any;
+
+    const handle = startStaleSweepInterval(
+      {
+        intervalMs: 20,
+        staleThresholdMs: 60_000,
+        maxOrgsPerPass: 10,
+      },
+      {
+        state,
+        getBuffer: () => fakeBuffer,
+        recordStaleEntry: () => {},
+        reportStaleEntrySnapshot: () => {},
+        logger: { warn: () => {} },
+        now: () => Date.now(),
+      },
+    );
+
+    // Wait for the interval to fire one tick. The tick will start, call
+    // readCursor, and then block on `gate`.
+    await new Promise((r) => setTimeout(r, 80));
+    expect(callOrder).toContain("readCursor:start");
+    expect(closeCalled).toBe(false);
+
+    // Call stop() concurrently — its promise MUST NOT resolve while the
+    // tick is still mid-flight.
+    let stopResolved = false;
+    const stopPromise = handle.stop().then(() => {
+      stopResolved = true;
+    });
+    await new Promise((r) => setTimeout(r, 50));
+    expect(stopResolved).toBe(false);
+    expect(closeCalled).toBe(false);
+
+    // Release the gate. The tick can now finish, and only then should
+    // stop() resolve and close the state.
+    resolveGate();
+    await stopPromise;
+    expect(stopResolved).toBe(true);
+    expect(closeCalled).toBe(true);
+
+    // The tick's readCursor:end MUST appear before the close — otherwise
+    // we closed the Redis client out from under an in-flight tick.
+    expect(callOrder.indexOf("readCursor:end")).toBeGreaterThan(-1);
+    expect(callOrder.indexOf("close")).toBeGreaterThan(
+      callOrder.indexOf("readCursor:end"),
+    );
+  });
+});