## Summary
Adds `drainBatchSize` to `MollifierDrainer` (default `1` — preserves
existing behaviour) and wires `TRIGGER_MOLLIFIER_DRAIN_BATCH_SIZE`
through the webapp (default `50`). Each tick the drainer now pops up to
`drainBatchSize` from each chosen env, then dispatches every popped
entry through the shared `concurrency`-bounded `pLimit`. Per-org/per-env
fairness is unchanged — only the in-env pop count grows.
Pre-existing behaviour was one pop per env per tick. For a single-env
burst that single-flighted the drain at the per-tick floor of `pop +
engine.trigger ≈ 50–60 ms`. With buffer entries piling up under a
real-world tenant burst that's tens of minutes of tail latency to fully
materialise — even though PG itself could comfortably sustain the
writes.
## Why this matters — heavy-tail illustration
Scenario: 100 customers in one window — 94 fire 20 triggers each, 5 fire
100, 1 fires 1000. Gate at `THRESHOLD=10/s`, `HOLD_MS=500`. First 10 of
each burst hit PG directly; the rest buffer.
| Customers | Triggers each | PG direct | Buffered each | Total buffered
|
|---|---|---|---|---|
| 94 small | 20 | 10 | 10 | 940 |
| 5 medium | 100 | 10 | 90 | 450 |
| 1 heavy | 1000 | 10 | 990 | 990 |
**With `DRAIN_BATCH_SIZE=50`, `DRAIN_CONCURRENCY=50`, ~50 ms
`engine.trigger`:**
| Tick | Pops | Dispatch waves | Wall-clock |
|---|---|---|---|
| 1 | 94×10 + 5×50 + 1×50 = 1 240 | 25 × 50 ms | ~1 300 ms (94 smalls
done) |
| 2 | 5×40 + 1×50 = 250 | 5 × 50 ms | ~300 ms (5 mediums done) |
| 3–20 | heavy alone, 50/tick | 1 × 50 ms | ~100 ms each |
| Customer class | Buffered fully drained |
|---|---|
| 94 small | **~1.3 s** |
| 5 medium | **~1.6 s** |
| 1 heavy | **~3.4 s** |
**Without batching (one pop per env per tick — current behaviour):**
| Customer class | Buffered fully drained |
|---|---|
| 94 small | ~500 ms |
| 5 medium | ~4.5 s |
| 1 heavy | **~49 s** |
So the heavy single-tenant tail drops from ~49 s to ~3.4 s (~14× faster)
without changing PG load characteristics. Smalls go up slightly in this
scenario (500 ms → 1.3 s) because all 100 envs share one tick's dispatch
queue — that's the trade we accept for the heavy tail; the worst-case
small wait is still inside one tick. PG load is identical either way (50
concurrent inserts at a time, capped by `DRAIN_CONCURRENCY`).
## What changed
**`packages/redis-worker`**
- New `drainBatchSize` option (default 1 — full backward compat).
- `runOnce()` refactored to pop per-env batches in parallel, then
dispatch all popped entries through the existing global `pLimit`.
Mid-batch pop failure aborts only that env's batch and counts as one
failure (same semantic as the old per-env path).
- Removed the now-unused `processOneFromEnv` helper.
**`apps/webapp`**
- `TRIGGER_MOLLIFIER_DRAIN_BATCH_SIZE` env var (default 50, matching
`DRAIN_CONCURRENCY`).
- Wired into `mollifierDrainer.server.ts`.
**Test cloud config** (separate cloud PR):
`TRIGGER_MOLLIFIER_DRAIN_BATCH_SIZE="50"` on the worker service.
Production rollout deferred until we've watched it on test cloud.
## Test plan
- [x] All 25 stub-based drainer tests pass (18 pre-existing + 7 new). 7
new tests under `MollifierDrainer.drainBatchSize`:
- pops up to `drainBatchSize` across ticks
- global `concurrency` cap still holds when batch > concurrency
- mid-batch pop failure isolation
- multi-env batch fan-out in one tick
- **hierarchical org fairness preserved at `drainBatchSize > 1`**
(load-bearing — guards against future regressions to
per-env-instead-of-per-org rotation)
- mixed success/failure accounting in a batched tick
- bounded pops on empty queue (no Lua spam past `drainBatchSize`)
- [x] All pre-existing tests still pass unchanged at default
`drainBatchSize=1` → backward-compat locked.
- [x] `pnpm run build --filter @trigger.dev/redis-worker` clean.
- [x] `pnpm run typecheck --filter webapp` clean.
- [x] `redisTest` block (real Redis via testcontainers) — couldn't run
locally on this branch due to testcontainers runtime discovery; will
validate in CI.
- [ ] Test-cloud smoke after cloud PR lands: fire `burst 50` against a
flagged env and confirm the 50th entry's drain time drops from ~2.5 s to
<200 ms.
## Notes
- Per-tick memory bound: `maxOrgsPerTick × drainBatchSize` entries can
sit in the JS pLimit queue between pop and dispatch. At defaults that's
`500 × 50 = 25 000` × ~5 KB snapshot ≈ ~125 MB worst case per worker —
well within headroom.
- The pre-batch model's strict per-env throughput cap of `1/tick` is
documented as the fairness baseline elsewhere. Org-level fairness is
what callers actually rely on; this change does not weaken that.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
---------
Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Daniel Sutton
139eede41c