What is backpressure and how do you handle it?

// Backpressure on a fake clock: a producer bursts at 150 msg/s for 10 s, then sends
// 50 msg/s; the consumer can do 100 msg/s. Four ways to handle the gap.
const SECONDS = 20, CAPACITY = 100, BOUND = 100;
const offered = (t) => (t < 10 ? 150 : 50);

function simulate(strategy) {
  const queue = []; // FIFO of [enqueueSecond, count]
  let depth = 0, dropped = 0, backlog = 0, maxDepth = 0, maxWait = 0, maxBacklog = 0, done = 0, offeredTotal = 0;
  const depths = [];
  for (let t = 0; t < SECONDS; t++) {
    let arriving = offered(t);
    offeredTotal += arriving;
    if (strategy === "block" || strategy === "credits") { arriving += backlog; backlog = 0; }
    let room = strategy === "unbounded" ? Infinity : BOUND + CAPACITY - depth; // what fits this second
    if (strategy === "credits") room = Math.min(room, CAPACITY); // consumer grants credits = its rate
    const accepted = Math.min(arriving, room);
    if (strategy === "drop") dropped += arriving - accepted;
    else backlog += arriving - accepted; // block / credits: producer holds it (upstream waits)
    if (accepted > 0) { queue.push([t, accepted]); depth += accepted; }
    let budget = CAPACITY; // consumer drains up to 100 this second
    while (budget > 0 && queue.length) {
      const seg = queue[0];
      const n = Math.min(budget, seg[1]);
      seg[1] -= n; budget -= n; depth -= n; done += n;
      maxWait = Math.max(maxWait, t - seg[0]);
      if (seg[1] === 0) queue.shift();
    }
    maxDepth = Math.max(maxDepth, depth);
    maxBacklog = Math.max(maxBacklog, backlog);
    depths.push(depth);
  }
  return { depths, maxDepth, maxWait, dropped, maxBacklog, done, offeredTotal, left: depth + backlog };
}

console.log(`offered: 150/s for t=0..9, 50/s for t=10..19; consumer 100/s; bound ${BOUND}`);
console.log("queue depth at the end of each second:");
const rows = [];
for (const s of ["unbounded", "drop", "block", "credits"]) {
  const r = simulate(s);
  rows.push([s, r]);
  console.log(`  ${s.padEnd(10)} ${r.depths.map((d) => String(d).padStart(3)).join(" ")}`);
}
console.log("\nstrategy    max depth  max wait  dropped  max producer backlog  processed  still pending");
for (const [s, r] of rows)
  console.log(`  ${s.padEnd(10)} ${String(r.maxDepth).padStart(9)} ${String(r.maxWait + " s").padStart(9)} ${String(r.dropped).padStart(8)} ${String(r.maxBacklog).padStart(21)} ${String(r.done).padStart(10)} ${String(r.left).padStart(14)}`);

console.log("\nsustained overload with an unbounded queue (in 120/s, out 100/s, 2 KB per message):");
for (const minutes of [1, 10, 60]) {
  const queued = 20 * 60 * minutes;
  console.log(`  after ${String(minutes).padStart(2)} min: ${String(queued).padStart(6)} queued, ${String((queued * 2) / 1000).padStart(5)} MB, newest waits ${queued / 100} s`);
}

// Real Node.js stream backpressure: write() returns false once the buffer reaches
// highWaterMark; a well-behaved producer waits for 'drain'. Node 22 default HWM shown.
const { Writable } = require("node:stream");
const { once } = require("node:events");

const CHUNK = Buffer.alloc(16 * 1024); // 16 KiB per write
const TOTAL = 20; // 320 KiB in all

function slowSink() {
  // A consumer that finishes one chunk per event-loop turn (setImmediate).
  return new Writable({ write(chunk, enc, cb) { setImmediate(cb); } });
}

console.log(`node ${process.version.split(".")[0]}: default writableHighWaterMark = ${slowSink().writableHighWaterMark} bytes`);

async function ignoring() {
  const w = slowSink();
  let firstFalse = -1, peak = 0;
  for (let i = 0; i < TOTAL; i++) {
    const ok = w.write(CHUNK);
    if (!ok && firstFalse < 0) firstFalse = i + 1;
    peak = Math.max(peak, w.writableLength);
  }
  w.end(); await once(w, "finish");
  console.log(`ignore false: write() first returned false on write #${firstFalse}, kept writing; peak buffered = ${peak / 1024} KiB`);
}

async function respecting() {
  const w = slowSink();
  let peak = 0, waits = 0;
  for (let i = 0; i < TOTAL; i++) {
    const ok = w.write(CHUNK);
    peak = Math.max(peak, w.writableLength);
    if (!ok) { waits++; await once(w, "drain"); }
  }
  w.end(); await once(w, "finish");
  console.log(`respect false: paused for 'drain' ${waits} times; peak buffered = ${peak / 1024} KiB`);
}

async function piped() {
  const { Readable, pipeline } = require("node:stream");
  let produced = 0, peak = 0;
  const src = new Readable({
    read() { produced++; this.push(produced <= TOTAL ? CHUNK : null); },
  });
  const w = slowSink();
  await new Promise((res, rej) => {
    const orig = w.write.bind(w);
    w.write = (c, ...a) => { const r = orig(c, ...a); peak = Math.max(peak, w.writableLength); return r; };
    pipeline(src, w, (err) => (err ? rej(err) : res()));
  });
  console.log(`pipeline():   ${TOTAL} chunks moved; peak buffered in the sink = ${peak / 1024} KiB`);
}

(async () => { await ignoring(); await respecting(); await piped(); })();