perf(redis-worker): pipeline HGETALLs in MollifierBuffer.listEntriesForEnv

The stale-sweep calls listEntriesForEnv with maxCount=1000 per env per pass.
The previous implementation issued one HGETALL per runId returned by LRANGE,
serially — ~1000 round-trips per env per pass. At any meaningful drainer
backlog this dominated sweep wall-time and made the sweep run longer than its
interval (the inFlight guard then drops every subsequent tick until it
finishes).

Refactor the per-entry fetch into a single ioredis pipeline so one network
batch covers every runId LRANGE returned. Behaviour is preserved end-to-end:

- Empty/missing entry hashes are still skipped (concurrent drainer ack/fail
  between LRANGE and our HGETALL DELs the hash — a runId on the queue with
  no backing hash is an expected race outcome, not an error).
- Per-entry parse failures are logged and skipped without aborting the batch.
- Per-entry pipeline errors are logged and skipped (pipeline.exec does not
  abort on individual command errors).

Existing tests for happy-path + empty queue + maxCount<=0 unchanged. Added a
new test characterising the torn-down-entry race (`mollifier:entries:r_a`
DEL'd between accept and listEntriesForEnv) — passes against both serial and
pipelined implementations, so it acts as a safety net for this refactor and
future ones.

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

Author: d-cs

.changeset/mollifier-buffer-pipeline-list-entries.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/redis-worker": patch
+---
+
+Pipeline the per-entry `HGETALL` fetches in `MollifierBuffer.listEntriesForEnv`. The previous serial implementation issued one Redis round-trip per runId returned by `LRANGE`, which dominated stale-sweep wall-time at any meaningful backlog (at the sweep's default maxCount=1000, this is ~1000 RTTs per env per pass). Behaviour is unchanged — entries are still skipped when the entry hash has been torn down by a concurrent drainer ack/fail between the LRANGE and the HGETALL.

packages/redis-worker/src/mollifier/buffer.test.ts

@@ -2341,6 +2341,41 @@ describe("MollifierBuffer.listEntriesForEnv", () => {
     }
   });
 
+  redisTest(
+    "skips entries whose hash was torn down between LRANGE and HGETALL (concurrent drainer ack/fail race)",
+    { timeout: 20_000 },
+    async ({ redisContainer }) => {
+      // The drainer can RPOP + ack/fail an entry between our LRANGE and
+      // the per-runId HGETALL — its DEL of the entry hash races our read.
+      // listEntriesForEnv must tolerate this: skip the runId, return
+      // every other entry. This is exercised here by simulating the race:
+      // LPUSH a runId onto the queue without an accompanying entry hash.
+      const buffer = new MollifierBuffer({
+        redisOptions: {
+          host: redisContainer.getHost(),
+          port: redisContainer.getPort(),
+          password: redisContainer.getPassword(),
+        },
+        logger: new Logger("test", "log"),
+      });
+
+      try {
+        await buffer.accept({ runId: "r_a", envId: "env_race", orgId: "org_1", payload: "{}" });
+        await buffer.accept({ runId: "r_b", envId: "env_race", orgId: "org_1", payload: "{}" });
+
+        // Tear down r_a's hash to simulate the drainer winning the race.
+        // The runId stays on the queue LIST but its entry hash is gone —
+        // listEntriesForEnv must tolerate the missing HGETALL result.
+        await buffer["redis"].del("mollifier:entries:r_a");
+
+        const entries = await buffer.listEntriesForEnv("env_race", 10);
+        expect(entries.map((e) => e.runId).sort()).toEqual(["r_b"]);
+      } finally {
+        await buffer.close();
+      }
+    },
+  );
+
   redisTest("maxCount <= 0 returns empty without hitting redis", { timeout: 20_000 }, async ({ redisContainer }) => {
     const buffer = new MollifierBuffer({
       redisOptions: {

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

@@ -251,20 +251,53 @@ export class MollifierBuffer {
   // Used by the stale-sweep to compute per-entry dwell time, so order is
   // immaterial — LRANGE returns them newest-first (LPUSH head) but the
   // caller scans the whole window. Non-destructive: the drainer still
-  // RPOPs these entries in FIFO order. Each entry hash is fetched
-  // separately; a `null` from getEntry (entry torn down by a concurrent
-  // drainer ack/fail between LRANGE and HGETALL) is skipped.
+  // RPOPs these entries in FIFO order.
+  //
+  // The entry HGETALLs are issued in a single pipelined batch (one
+  // network round-trip instead of N) — at the stale-sweep's default
+  // maxCount=1000 the serial implementation cost ~1000 RTTs per env,
+  // which dominated sweep wall-time at any meaningful backlog.
+  //
+  // A missing entry (empty hash) is skipped: the drainer's RPOP+DEL of
+  // the entry hash can race our LRANGE→HGETALL window, so a runId on
+  // the queue with no backing hash is an expected concurrency outcome,
+  // not an error.
   async listEntriesForEnv(envId: string, maxCount: number): Promise<BufferEntry[]> {
     if (maxCount <= 0) return [];
     const runIds = await this.redis.lrange(
       `mollifier:queue:${envId}`,
       0,
       maxCount - 1,
     );
-    const entries: BufferEntry[] = [];
+    if (runIds.length === 0) return [];
+
+    const pipeline = this.redis.pipeline();
     for (const runId of runIds) {
-      const entry = await this.getEntry(runId);
-      if (entry) entries.push(entry);
+      pipeline.hgetall(`mollifier:entries:${runId}`);
+    }
+    const results = await pipeline.exec();
+    if (!results) return [];
+
+    const entries: BufferEntry[] = [];
+    for (let i = 0; i < results.length; i++) {
+      const [err, raw] = results[i] as [Error | null, Record<string, string> | null];
+      if (err) {
+        this.logger.error("MollifierBuffer.listEntriesForEnv: hgetall failed", {
+          runId: runIds[i],
+          err: err.message,
+        });
+        continue;
+      }
+      if (!raw || Object.keys(raw).length === 0) continue;
+      const parsed = BufferEntrySchema.safeParse(raw);
+      if (!parsed.success) {
+        this.logger.error("MollifierBuffer.listEntriesForEnv: invalid entry shape", {
+          runId: runIds[i],
+          errors: parsed.error.flatten(),
+        });
+        continue;
+      }
+      entries.push(parsed.data);
     }
     return entries;
   }