fix(mollifier): pipeline per-tick org→env fan-out and reconcile shutdown deadlines

Two correctness/perf fixes on top of the phase-2 drainer:

1. `runOnce` was awaiting `listEnvsForOrg` serially before any pop ran.
   At the default `maxOrgsPerTick=500` and a ~1ms RTT, that's a ~500ms
   per-tick latency floor before `pLimit` even sees work. `Promise.all`
   over the org slice lets ioredis auto-pipeline the SMEMBERS into a
   single round-trip. Order is preserved so the org→envs pairing stays
   deterministic and `pickEnvForOrg` still rotates per org.

2. The SIGTERM handler is sync fire-and-forget: `drainer.stop({timeoutMs})`
   returns a promise that keeps the loop alive, but in cluster mode the
   primary process runs its own `GRACEFUL_SHUTDOWN_TIMEOUT` and will hit
   `process.exit(0)` independently. If the drainer's deadline exceeds
   the primary's, the drainer's "log a warning on timeout" turns into
   "hard exit with no log". Assert at boot that
   `MOLLIFIER_DRAIN_SHUTDOWN_TIMEOUT_MS <= GRACEFUL_SHUTDOWN_TIMEOUT - 1s`
   so a misconfig fails loud instead of disappearing at shutdown.

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

Author: d-cs

apps/webapp/app/services/worker.server.ts

@@ -142,6 +142,26 @@ export async function init() {
   try {
     const drainer = getMollifierDrainer();
     if (drainer && !global.__mollifierShutdownRegistered__) {
+      // The SIGTERM handler is sync fire-and-forget: it kicks off
+      // `drainer.stop(...)` and returns. The unresolved promise keeps the
+      // event loop alive, but in cluster mode the primary process runs its
+      // own graceful-shutdown timer (`GRACEFUL_SHUTDOWN_TIMEOUT`) and will
+      // call `process.exit(0)` independently. If the drainer's deadline
+      // exceeds the primary's, the drainer gets cut off mid-wait — which
+      // turns "log a warning on timeout" into "hard exit with no log".
+      // Reconcile the two timeouts at boot rather than discovering the
+      // misconfig from a missing warning at shutdown. Margin gives the
+      // primary room to do its own teardown after the drainer settles.
+      const SHUTDOWN_MARGIN_MS = 1_000;
+      if (
+        env.MOLLIFIER_DRAIN_SHUTDOWN_TIMEOUT_MS >=
+        env.GRACEFUL_SHUTDOWN_TIMEOUT - SHUTDOWN_MARGIN_MS
+      ) {
+        throw new Error(
+          `MOLLIFIER_DRAIN_SHUTDOWN_TIMEOUT_MS (${env.MOLLIFIER_DRAIN_SHUTDOWN_TIMEOUT_MS}) must be at least ${SHUTDOWN_MARGIN_MS}ms below GRACEFUL_SHUTDOWN_TIMEOUT (${env.GRACEFUL_SHUTDOWN_TIMEOUT}); otherwise the primary's hard exit shadows the drainer's deadline.`,
+        );
+      }
+
       // The drainer owns a polling loop and a Redis client; let it drain
       // in-flight pops on shutdown rather than tearing the process down
       // mid-handler. `init()` is called per request from entry.server.tsx,

packages/redis-worker/src/mollifier/drainer.ts

@@ -74,13 +74,20 @@ export class MollifierDrainer<TPayload = unknown> {
 
     const orgSlice = this.takeOrgSlice(orgs);
 
-    // For each picked org, pick one env from its active-envs set. The
-    // listEnvsForOrg calls are independent and could be parallelised; we
-    // do them sequentially for simplicity since they're each a fast
-    // SMEMBERS. The actual pops happen concurrently below.
+    // Fan the per-org SMEMBERS out in a single pipelined round-trip. Serial
+    // awaits would otherwise add `orgSlice.length × RTT` of dead time before
+    // pops start — at the default `maxOrgsPerTick=500` and a ~1ms ElastiCache
+    // RTT that's a ~500ms per-tick floor. ioredis auto-pipelines concurrent
+    // commands into one batch, so the burst is cheap; SMEMBERS on a small set
+    // is O(N) per org and trivial at this scale. `Promise.all` preserves
+    // order, so the org→envs pairing below stays deterministic.
+    const envsByOrg = await Promise.all(
+      orgSlice.map((orgId) => this.buffer.listEnvsForOrg(orgId)),
+    );
     const targets: string[] = [];
-    for (const orgId of orgSlice) {
-      const envsForOrg = await this.buffer.listEnvsForOrg(orgId);
+    for (let i = 0; i < orgSlice.length; i++) {
+      const orgId = orgSlice[i]!;
+      const envsForOrg = envsByOrg[i]!;
       if (envsForOrg.length === 0) continue;
       const envId = this.pickEnvForOrg(orgId, envsForOrg);
       targets.push(envId);