feat(redis-worker): per-env batched pop in MollifierDrainer

Adds a drainBatchSize option (default 1, preserves existing behaviour)
that lets the drainer pop up to N entries from each chosen env per tick
and dispatch them all through the shared concurrency-bounded limiter.
Org/env fairness is preserved — the per-tick env selection is
unchanged, only the in-env pop count grows. Wires
TRIGGER_MOLLIFIER_DRAIN_BATCH_SIZE through the webapp (default 50).

For a single-env burst of K entries with K > 1, drain time drops from
K × tick_time to ceil(K / drainBatchSize) × tick_time with handler
parallelism capped at concurrency. Heavy single-tenant tails go from
minutes to tens of seconds without changing PG load characteristics.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: d-cs

.changeset/mollifier-drain-batch-size.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/redis-worker": patch
+---
+
+`MollifierDrainer` now accepts a `drainBatchSize` option that controls how many entries it pops from a single env per tick. Default remains 1 (one pop per env per tick — previous behaviour). Setting it higher lets a single-env burst drain at handler-parallelism speed instead of one entry per ~50ms tick: the drainer pops up to `drainBatchSize` from the picked env and dispatches all popped entries through the shared `concurrency`-bounded limiter. Org/env fairness is unchanged — the per-tick env selection is unaffected.

.server-changes/mollifier-drain-batch-size.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: improvement
+---
+
+Wire `TRIGGER_MOLLIFIER_DRAIN_BATCH_SIZE` (default 50) into the drainer so single-env bursts drain at the full `DRAIN_CONCURRENCY` budget instead of one pop per ~50ms tick. For a 20k-trigger burst on one env this cuts drain time from minutes to ~tens of seconds; smaller bursts (e.g. 50 on one env) drop from ~2.5s to ~50–100ms tail.

apps/webapp/app/env.server.ts

@@ -1101,6 +1101,16 @@ const EnvironmentSchema = z
     TRIGGER_MOLLIFIER_DRAIN_MAX_ATTEMPTS: z.coerce.number().int().positive().default(3),
     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),
+    // Per-env per-tick pop cap. The drainer rotates one env per org per
+    // tick; this bounds how many entries it pops from that env before
+    // dispatching them through the shared `DRAIN_CONCURRENCY`-bounded
+    // limiter. Default matches `DRAIN_CONCURRENCY` so a single-env burst
+    // uses the full handler-parallelism budget — for 20k buffered on one
+    // env this is the difference between ~17m (one-pop-per-tick × ~50ms)
+    // and ~20s (400 ticks × concurrent engine.trigger). Org/env fairness
+    // is preserved because the per-tick env selection is unchanged; only
+    // the in-env pop count grows.
+    TRIGGER_MOLLIFIER_DRAIN_BATCH_SIZE: z.coerce.number().int().positive().default(50),
     // Periodic sweep that scans buffer queue LISTs for entries whose
     // dwell exceeds the stale threshold. Independent of the drainer —
     // its job is exactly to make a stuck/offline drainer visible to

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

@@ -72,6 +72,7 @@ function initializeMollifierDrainer(): MollifierDrainer<MollifierSnapshot> {
   logger.debug("Initializing mollifier drainer", {
     concurrency: env.TRIGGER_MOLLIFIER_DRAIN_CONCURRENCY,
     maxAttempts: env.TRIGGER_MOLLIFIER_DRAIN_MAX_ATTEMPTS,
+    drainBatchSize: env.TRIGGER_MOLLIFIER_DRAIN_BATCH_SIZE,
   });
 
   const drainer = new MollifierDrainer<MollifierSnapshot>({
@@ -81,6 +82,7 @@ function initializeMollifierDrainer(): MollifierDrainer<MollifierSnapshot> {
     concurrency: env.TRIGGER_MOLLIFIER_DRAIN_CONCURRENCY,
     maxAttempts: env.TRIGGER_MOLLIFIER_DRAIN_MAX_ATTEMPTS,
     maxOrgsPerTick: env.TRIGGER_MOLLIFIER_DRAIN_MAX_ORGS_PER_TICK,
+    drainBatchSize: env.TRIGGER_MOLLIFIER_DRAIN_BATCH_SIZE,
     isRetryable: isRetryablePgError,
   });