Implementing Backpressure with Message Credits

When the data is not stream-shaped — record batches, rendered tiles, search hits — the platform will not throttle the producer for you. A credit window is the smallest mechanism that will, and it fits in about sixty lines shared by both threads.

This is the hand-written path from Streaming & Backpressure Across Threads, part of High-Performance Computation Patterns. The design goal is a hard ceiling on in-flight memory that does not depend on how fast either thread happens to run.

The Channel, Both Halves

One shared module, imported by the page and the worker, so the message shapes cannot drift.

// credit-channel.ts — shared by both threads
export type ChannelMessage<T> =
  | { kind: 'CHUNK'; seq: number; payload: T }
  | { kind: 'CREDIT'; n: number }
  | { kind: 'END'; count: number }
  | { kind: 'ERROR'; message: string }
  | { kind: 'CANCEL' };

export class CreditProducer<T> {
  #credits = 0;
  #wake: (() => void) | null = null;
  #cancelled = false;
  #seq = 0;

  constructor(private post: (m: ChannelMessage<T>, transfer?: Transferable[]) => void) {}

  /** Called from the message listener on the producer's side. */
  handle(message: ChannelMessage<T>): void {
    if (message.kind === 'CREDIT') { this.#credits += message.n; this.#wake?.(); this.#wake = null; }
    if (message.kind === 'CANCEL') { this.#cancelled = true; this.#wake?.(); this.#wake = null; }
  }

  async send(payload: T, transfer?: Transferable[]): Promise<boolean> {
    if (this.#credits === 0) await new Promise<void>((r) => { this.#wake = r; });
    if (this.#cancelled) return false;          // stop the loop cooperatively
    this.#credits--;
    this.post({ kind: 'CHUNK', seq: this.#seq++, payload }, transfer);
    return true;
  }

  end(): void { this.post({ kind: 'END', count: this.#seq }); }
}

// credit-consumer.ts — the other half
export class CreditConsumer<T> {
  #outstanding = 0;

  constructor(
    private post: (m: ChannelMessage<T>) => void,
    private onChunk: (payload: T, seq: number) => Promise<void> | void,
    private window = 4,
  ) {
    this.post({ kind: 'CREDIT', n: this.window });   // prime the pipeline
  }

  async handle(message: ChannelMessage<T>): Promise<void> {
    if (message.kind !== 'CHUNK') return;
    this.#outstanding++;
    try {
      await this.onChunk(message.payload, message.seq);
    } finally {
      this.#outstanding--;
      this.post({ kind: 'CREDIT', n: 1 });            // ALWAYS, even on failure
    }
  }

  get depth(): number { return this.#outstanding; }
}

The producer’s send is the only place that can block, and it blocks by parking on a promise rather than spinning — a parked worker uses no CPU, so a stalled consumer costs nothing but latency.

Wiring It Up

// worker.ts
import { CreditProducer, type ChannelMessage } from './credit-channel';

const producer = new CreditProducer<Float64Array>(
  (m, transfer) => self.postMessage(m, transfer ?? []));

self.addEventListener('message', (e: MessageEvent<ChannelMessage<Float64Array>>) =>
  producer.handle(e.data));

export async function run(source: AsyncIterable<Float64Array>): Promise<void> {
  try {
    for await (const batch of source) {
      const proceed = await producer.send(batch, [batch.buffer]);
      if (!proceed) return;                        // cancelled
    }
    producer.end();
  } catch (error) {
    self.postMessage({ kind: 'ERROR', message: (error as Error).message });
  }
}

// page.ts
const consumer = new CreditConsumer<Float64Array>(
  (m) => worker.postMessage(m),
  async (batch, seq) => { await renderBatch(batch, seq); },
  4,                                              // window
);

worker.addEventListener('message', (e: MessageEvent<ChannelMessage<Float64Array>>) => {
  if (e.data.kind === 'ERROR') return fail(new Error(e.data.message));
  if (e.data.kind === 'END') return complete(e.data.count);
  void consumer.handle(e.data);
});

Note what the consumer’s handle does not do: it does not check whether the job is still relevant. That check belongs inside onChunk, after the credit accounting is already committed by the try/finally — otherwise an early return skips the credit and shrinks the window permanently.

Sizing the Window

The arithmetic is short. Let L be the round-trip latency of the boundary (0.2–1 ms in practice), C the consumer’s cost per chunk, and P the producer’s cost per chunk. To keep the producer busy, the window must cover the time between sending a chunk and getting its credit back:

window ≥ (L + C) / P        rounded up, minimum 2

A consumer that renders in 14 ms against a producer that parses in 2 ms wants a window of about 8; a consumer that costs 1 ms against a producer that costs 20 ms is satisfied by 2. Peak memory is window × chunkSize either way, so the second case is not merely adequate — it is strictly better, and any larger window buys nothing.

Where chunk sizes vary by an order of magnitude within a job, count bytes instead of messages: grant a 16 MB budget, decrement by byteLength when sending, and return the same number on completion. The invariant becomes “at most 16 MB in flight” regardless of how the producer chose to slice the data.

Metrics That Catch Regressions

Two numbers, exported per job, distinguish every failure this protocol can have.

Depth (sent − acknowledged, on the producer) is the memory ceiling in action. Healthy pipelines oscillate just below the window; a depth pinned at the window for seconds means the consumer has stalled; a depth stuck at zero means the producer is the bottleneck and the window is irrelevant.

Park time (total milliseconds the producer spent awaiting credit) is the tuning signal. Near zero means the window is at least large enough. A large fraction of job time means either the consumer is genuinely the bottleneck — in which case a bigger window will not help, only more consumers will — or the window is too small for the latency.

// producer instrumentation, five lines
let parkedMs = 0;
async function awaitCredit() {
  const t0 = performance.now();
  await park();
  parkedMs += performance.now() - t0;
}
// on END: self.postMessage({ kind: 'STATS', parkedMs, chunks: seq });

Log both with the job’s duration and chunk count. A regression in park time with unchanged depth is almost always a consumer that got slower — a new render path, a larger viewport — and it shows up here days before anyone files a bug about the page feeling sluggish.

One Channel Per Job

A single worker often runs several jobs over its lifetime, and sharing one credit counter between them creates a subtle coupling: a stalled job’s outstanding chunks consume the window that a newly started job needs. Giving each job its own MessageChannel removes the interference and simplifies teardown.

// page.ts — a port per job, closed when the job ends
export function startJob(spec: JobSpec) {
  const { port1, port2 } = new MessageChannel();
  const consumer = new CreditConsumer<Float64Array>(
    (m) => port1.postMessage(m), (batch, seq) => renderBatch(batch, seq), 4);

  port1.onmessage = (e) => {
    if (e.data.kind === 'END' || e.data.kind === 'ERROR') { port1.close(); return finish(e.data); }
    void consumer.handle(e.data);
  };

  worker.postMessage({ kind: 'START', spec, port: port2 }, [port2]);
  return () => port1.postMessage({ kind: 'CANCEL' });     // the cancel handle
}

Each job now has an independent window, an independent depth metric, and a teardown that is one close() rather than a filter over shared state. The cost is one MessageChannel per job — a few hundred bytes — and the discipline of closing it on every terminal path, including cancellation.

This also makes the protocol testable in isolation. A test can drive a CreditProducer and a CreditConsumer through a plain MessageChannel in a single thread, with no worker at all, and assert that depth never exceeds the window under an artificially slow consumer. That test catches the credit-accounting bugs below far more reliably than any integration test.

Gotchas

Returning credit before the work. The single most common error, and it silently restores unbounded queueing. The credit represents freed capacity; sending it on arrival means the producer refills at message rate.

Skipping the credit on an early return. if (job !== currentJob) return; inside the chunk handler leaks one credit per stale chunk. Put every early return inside the try, so the finally still runs.

Cancelling a parked producer. A producer awaiting credit is not running its loop, so a cancelled flag it only checks at the top of the loop is never read. The park promise must also be resolved by the cancel message — the dashed edge in the state diagram.

Assuming ordering across a pool. Sequence numbers are in the protocol above for a reason: with more than one producer, chunks arrive out of order, and a consumer that appends blindly produces scrambled output. Either reassemble by seq or design the consumer to be order-independent.

Priming with a window the consumer cannot honour. The initial grant should match what the consumer can actually hold; a window of 64 chosen “to be safe” means 64 chunks may arrive before the first credit returns, which is precisely the unbounded behaviour in miniature.

Forgetting that transfer detaches. producer.send(batch, [batch.buffer]) detaches batch immediately. A producer that reuses one scratch array across iterations gets an empty buffer on the second send — allocate per chunk, or run a small pool of buffers the consumer returns explicitly.

Performance Note

Streaming 12,000 record batches of 256 kB from a worker to a rendering page on a 2023 laptop: with no flow control, peak process memory reached 740 MB and the tab dropped frames throughout. With a window of 4, peak in-flight memory was 1 MB, total wall clock was within 2% of the uncontrolled run, and frame pacing was steady. The window costs nothing in throughput because the consumer was always the limit — it only changes where the un-rendered data waits.