requests-batcher

An optimized, production-grade write-buffering, batching, and coalescing library for Node 22+, with zero runtime dependencies.

The "DataLoader for writes." DataLoader coalesces same-tick load(key) calls to kill the read N+1. requests-batcher does the mirror image for writes: it collapses a flood of hot-key, convergent-state updates (presence, counters, "last seen", telemetry) into a few bulk upserts — with bounded memory (backpressure) and a tunable, observable loss window.

Measured against a real Postgres: 1,541× fewer DB queries, and under hot-key collision ~57% fewer rows actually written — see Benchmark-backed results.

When to use it (and when to use a queue instead)

requests-batcher is in-process and best-effort by design. It owns one niche and is honest about its boundaries.

| If your writes are… | Use | | :--- | :--- | | Convergent state, hot keys, loss-tolerant, single process (presence, counters, heartbeats, telemetry, cache/search refresh) | ✅ requests-batcher | | Append-only events/audit you must keep, but want fewer roundtrips | ✅ requests-batcher in batch mode (coalesce:false, the default — keeps every item) | | Must-not-lose, exactly-once, strictly ordered, or cross-process / multi-node | ❌ Queue / transactional outbox (Kafka, SQS, BullMQ) |

The sweet spot is workloads where the latest value wins and a tiny crash-window of loss is acceptable — because the next update reconverges the row (eventual consistency). That loss-tolerance is exactly what makes the batching safe here.

Why use it (vs. a hand-rolled setTimeout + array)

1. Write coalescing with merge / last-write-wins — collapses evolving updates to the same row. This is the unique capability; nothing else in the ecosystem does it.
2. Flood protection via size + time triggers — maxBuffer fires under flood, flushAge governs the low-load window. Tunable RPO vs. write-amplification.
3. highWaterMark backpressure — add() pauses the producer (stream-style) when the queue is full, so OOM is impossible by construction (gotchas/0003).
4. Honest loss control — bounded, observable loss window (oldestPendingAgeMs, health()), graceful drain on shutdown, dead-letter on terminal failure. Loss is measured and never silent.
5. Clean promise boundaries — every add() resolves/rejects exactly when its batch settles, so a caller that awaits gets a real durability signal (no silent loss).

Installation

npm install requests-batcher

Quick start — the flagship case: hot-key counters / presence

This is where the library uniquely wins. A flood of updates to a small pool of hot keys (view counts, presence, "last seen") collapses into a few bulk upserts.

import { RequestBatcher } from 'requests-batcher';
import postgres from 'postgres';

const sql = postgres('postgres://localhost/db');

const presence = new RequestBatcher(
  // flushFn receives only the FINAL state per user in the window — one row each.
  async (batch) => {
    await sql`
      INSERT INTO user_presence ${sql(batch, 'id', 'status', 'lastSeen')}
      ON CONFLICT (id) DO UPDATE
        SET status = EXCLUDED.status, "lastSeen" = EXCLUDED."lastSeen"
    `;
  },
  {
    flushAge: 100,                 // micro-batch window (always < 1000ms)
    coalesce: true,                // ← merge same-key updates in-buffer
    getKey: (u) => u.id,
    coalesceStrategy: 'last-write-wins',
    acknowledgeCoalesceDropsData: true, // I understand intermediate writes are dropped
  }
);

// User 42 flips status 3× in the window…
presence.add({ id: 42, status: 'online',  lastSeen: Date.now() });
presence.add({ id: 42, status: 'typing',  lastSeen: Date.now() });
presence.add({ id: 42, status: 'idle',    lastSeen: Date.now() });
// …the DB receives ONE upsert: { id: 42, status: 'idle' }.
// metrics.coalesced += 2 (the two intermediate writes were merged away).

Measured (real Postgres, 50 hot keys, 60 s): 56.6% of requests coalesced away → 730 rows written/sec instead of 1,687 (~57% less DB write work), at the same p99 latency, 100% correctness. With more keys (less collision) the win shifts to pure roundtrip reduction (1,541× fewer queries at 1,000 keys). See §7.C.

⚠️ Coalescing drops data on purpose. coalesce:true discards intermediate same-key values — only safe on convergent state (latest-wins). It is off by default. The first time you enable it the library logs a one-time warning; pass acknowledgeCoalesceDropsData: true once you've confirmed your data is convergent. For event/audit streams that must keep every item, leave coalesce:false (batch mode).

Counter / accumulator variant (merge)

const views = new RequestBatcher(
  async (batch) => {
    // one row per key, increment already summed in-memory
    for (const { key, inc } of batch) {
      await sql`UPDATE counters SET n = n + ${inc} WHERE key = ${key}`;
    }
  },
  {
    flushAge: 200,
    coalesce: true,
    getKey: (e) => e.key,
    coalesceStrategy: 'merge',
    mergeFn: (prev, next) => ({ key: prev.key, inc: prev.inc + next.inc }),
    acknowledgeCoalesceDropsData: true,
  }
);
// 10,000 `views.add({ key: 'post:1', inc: 1 })` → one `+N` UPDATE.

Other proven patterns

High-volume inserts (batch mode — keeps every item)

For append-only events/audit rows, leave coalescing off. The library is then a lossless (within-process, best-effort-durability) batcher: every item is written, just grouped.

const events = new RequestBatcher(
  async (batch) => { await sql`INSERT INTO events ${sql(batch, 'userId', 'type', 'ts')}`; },
  { flushAge: 100, maxBuffer: 500, maxConcurrentFlushes: 3 } // coalesce defaults to false
);

app.post('/event', (req, res) => {
  events.add(req.body)
    .then(() => res.sendStatus(201))    // resolves when THIS item's batch persisted
    .catch((err) => res.status(500).json({ error: err.message }));
});

Measured: 100× fewer DB queries (1,612 RPS in → 16 QPS out), 100% correctness (§3).

Fire-and-forget intake with bounded memory (backpressure)

If a producer doesn't await add() (e.g. an HTTP handler that responds 202 before persistence), intake can outrun the DB and grow memory without bound. Set highWaterMark and add() pauses the producer instead of buffering — OOM impossible by construction.

const ingest = new RequestBatcher(writeBatch, {
  flushAge: 50,
  highWaterMark: 10_000,        // cap items in memory (buffered + in-flight)
});

app.post('/telemetry', async (req, res) => {
  await ingest.add(req.body);   // awaits only if at capacity (backpressure shows as latency)
  res.sendStatus(202);
});

Measured: peak buffered items never exceed highWaterMark under a 60k-item flood — RAM stays flat where the unbounded path would OOM (gotchas/0003). For producers that cannot be paused (UDP, sync telemetry), opt into overflowStrategy: 'drop-oldest' | 'drop-newest' — explicit, counted shedding (metrics.dropped + onDrop) instead of OOM.

Graceful shutdown — zero loss on a planned restart

The #1 real-world loss event is rolling deploys/restarts, not crashes. Drain on signal and lose nothing.

const batcher = new RequestBatcher(writeBatch, { flushAge: 50 });
batcher.closeOnSignals(['SIGTERM', 'SIGINT'], { drainTimeoutMs: 5000 });

Measured: close({ drainTimeoutMs }) drained 1,500,000 accepted writes with 0 lost (Scenario 5C).

Observe the loss window (RPO) and alert

const h = batcher.health();
// { bufferedItemCount, oldestPendingAgeMs, maxObservedBufferAgeMs, activeFlushes, ... }
if (h.oldestPendingAgeMs > 200 || h.bufferedItemCount > 50_000) alert('RPO SLO breach');

Measured: maxObservedBufferAgeMs tracks the configured window almost exactly (7 ms at flushAge:1 → 101 ms at flushAge:100), so the data-at-risk on a crash is bounded and known. With await add(), every un-persisted item rejects — 0 silent loss (Scenario 5A/5B).

Benchmark-backed results

Full methodology, tables, and reproduction commands: benchmark/README.md. Headline figures (real Postgres on a 100-connection pool, 100 workers, 60 s/round, measured — not estimated):

| Claim | Result | Source | | :--- | :--- | :--- | | DB query reduction | 24,688 QPS → 16 QPS = 1,541× | §3 | | Hot-key write reduction | 56.6% coalesced → ~57% fewer rows written | §7.C | | Bounded memory (backpressure) | peak buffered ≤ highWaterMark under a 200k-burst flood, 0 loss (OOM impossible) | §6.A · gotchas/0003 | | Bounded memory (drop valve) | un-pausable flood capped at highWaterMark; shed items counted, written+dropped===adds | §6.B | | Live RPO probe | health() tracks the flush window under load (2/10/50 ms at flushAge 1/10/50) | §6.C | | Loss on planned drain | 0 of 1,500,000 accepted | §5C | | Silent loss with await | 0 (every at-risk item rejects) | §5B | | Reconvergence after loss | 1,000 / 1,000 keys self-heal on next update | §5D |

Reproduce locally:

npm run benchmark          # simulated DB
npm run benchmark:real     # real Postgres (needs a DB on :5439)
npm run benchmark:window   # flush-window vs batch-size tradeoff
npm run benchmark:loss     # data-loss window / crash / shutdown (RPO)
npm run benchmark:flood    # backpressure / overflow-drop / health() under flood

API reference

new RequestBatcher(flushFn, options?)

type FlushFunction<T, R> = (batch: T[]) => Promise<R> | R;

flushFn receives the (optionally coalesced) batch and persists it. Every add() whose item is in that batch resolves with flushFn's return value (or rejects if it throws after all retries).

Options

| Option | Type | Default | Description | | :--- | :--- | :--- | :--- | | flushAge | number | 50 | Max age (ms) of buffered items before a flush. Must be < 1000. 0 = next-tick burst mode (setImmediate): collapses a synchronous burst into one flush at ~0 added latency. | | maxBuffer | number | 1000 | Flush immediately when this many items are buffered. | | maxConcurrentFlushes | number | 5 | Cap on concurrent flushFn calls. | | highWaterMark | number | undefined | Max items in memory (buffered + in-flight). When reached, add() pauses the caller — bounds RAM so OOM is impossible. Unset = unbounded (legacy). | | overflowStrategy | 'backpressure' \| 'drop-oldest' \| 'drop-newest' | 'backpressure' | Behavior at highWaterMark. drop-* shed one item (counted in metrics.dropped + onDrop) for un-pausable producers. Requires highWaterMark. | | maxRetries | number | 3 | Retry attempts for a failed batch (exponential backoff). | | retryDelay | number | 100 | Initial retry delay (ms). | | coalesce | boolean | false | Enable in-buffer write coalescing. Drops intermediate same-key writes — convergent state only. | | acknowledgeCoalesceDropsData | boolean | false | Suppress the one-time coalesce safety warning. | | getKey | (item) => string \| number | — | Key extractor. Required when coalesce is true. | | coalesceStrategy | 'last-write-wins' \| 'merge' | 'last-write-wins' | How to resolve same-key writes. | | mergeFn | (prev, next) => T | — | Required for 'merge'. | | onFlush | (batch, result) => void | — | Success hook. | | onError | (error, batch) => void | — | Terminal-failure hook (after retries). | | onDeadLetter | (items, error) => void | — | Receives the actual items of a batch that exhausted retries, so you can persist them elsewhere (no silent loss). Counted in metrics.deadLettered. | | onDrop | (items) => void | — | Receives items shed by an overflow strategy. Counted in metrics.dropped. |

Methods & properties

class RequestBatcher<T, R = any> {
  add(item: T): Promise<R>;                 // resolves/rejects when its batch settles
  flush(): Promise<void>;                    // flush the current buffer now
  close(opts?: { drainTimeoutMs?: number }): Promise<void>;     // graceful drain, bounded
  closeOnSignals(signals?: NodeJS.Signals[], // drain on SIGTERM/SIGINT; returns an unregister fn
                 opts?: { drainTimeoutMs?: number }): () => void;
  drain(): Promise<void>;                    // resolves when capacity is free (stream 'drain')
  health(): HealthSnapshot;                  // one-shot RPO/health snapshot for SLO alerts

  readonly pendingCount: number;             // items in the buffer right now
  readonly bufferedItemCount: number;        // items in memory (buffered + in-flight); ≤ highWaterMark
  readonly oldestPendingAgeMs: number;       // age of the oldest buffered item — current RPO; 0 if empty
  readonly writableNeedDrain: boolean;       // true at/over highWaterMark (mirrors stream)
  readonly activeFlushesCount: number;
  readonly isClosed: boolean;
  readonly metrics: BatcherMetrics;
}

Metrics — the loss ledger

Every item is accounted for. Balance invariant: written + coalesced + dropped + deadLettered === adds.

interface BatcherMetrics {
  adds: number;                  // items submitted via add()
  flushes: number;               // batch flushes attempted
  written: number;               // items successfully persisted
  coalesced: number;             // items merged away in-buffer (intentional, convergent only)
  dropped: number;               // items shed by an overflow strategy (onDrop)
  deadLettered: number;          // items handed back via onDeadLetter after retries
  failed: number;                // legacy alias of deadLettered
  retries: number;               // retry attempts made
  backpressured: number;         // add() calls that had to pause for capacity
  maxObservedBufferAgeMs: number;// worst buffer age seen at flush time — largest RPO window
  individualWrites: number;      // reserved for error-isolation splitting (not yet implemented)
}

Resilience

• Retries with exponential backoff (maxRetries, retryDelay) are applied to every failed batch.
• Dead-letter on terminal failure: when a batch exhausts its retries, onDeadLetter(items, error) hands back the actual items (and each add() promise rejects), so loss becomes handled, not silent.
• Error-isolation batch-splitting (retry a failing batch as smaller chunks to isolate one bad row) is not yet implemented — the individualWrites metric is reserved for it. Today a permanently-failing batch dead-letters as a whole.

Design notes

• Single-file ESM, zero runtime dependencies. Compiles to one ES module.
• Event-loop friendly. close() clears all timers, flushes remaining items, and settles every pending promise — no dangling handles in tests or shutdown.
• Two failure axes, handled independently. Memory is bounded by highWaterMark (OOM impossible); data loss is bounded by the flush window + drain-on-shutdown (RPO small, known, observable). Durability beyond best-effort stays the queue's job — by design.

See DEVELOPMENT_PLAN.md for the full roadmap and rationale, and gotchas/ for hard-won measurement and design lessons.