Backpressure Management

Prevent OOM crashes when producers outpace consumers.

When to Use This Skill

• Database writes slower than event ingestion

• Memory filling up with queued items

• Need to handle traffic spikes gracefully

• Want to drop low-priority data under load

Core Concepts

• Bounded buffer - Fixed-size queue prevents unbounded growth

• Watermarks - Thresholds trigger state changes

• Strategies - Block, drop oldest, drop newest, or sample

• Adaptive flushing - Adjust rate based on downstream health

State Machine

NORMAL (< 50%) → ELEVATED (50-80%) → CRITICAL (80-100%) → BLOCKED (100%) ↑ │ └──────────────────────────────────────────────────────────┘ (buffer drains)

TypeScript Implementation

Types

// types.ts export enum BackpressureState { NORMAL = 'normal', ELEVATED = 'elevated', CRITICAL = 'critical', BLOCKED = 'blocked', DRAINING = 'draining', }

export enum BackpressureStrategy { BLOCK = 'block', DROP_OLDEST = 'drop_oldest', DROP_NEWEST = 'drop_newest', SAMPLE = 'sample', }

export interface BackpressureConfig { maxBufferSize: number; highWatermark: number; // 0-1 lowWatermark: number; // 0-1 strategy: BackpressureStrategy; sampleRate?: number; maxBlockTimeMs?: number; batchSize: number; minFlushIntervalMs: number; maxFlushIntervalMs: number; targetLatencyMs: number; }

export interface FlushResult { success: number; failed: number; errors: Error[]; }

export type FlushFunction<T> = (items: T[]) => Promise<FlushResult>;

Bounded Buffer

// buffer.ts export class BoundedBuffer<T> { private items: T[] = [];

constructor(private readonly maxSize: number) {}

get size(): number { return this.items.length; } get capacity(): number { return this.maxSize; } get utilization(): number { return this.items.length / this.maxSize; }

isFull(): boolean { return this.items.length >= this.maxSize; } isEmpty(): boolean { return this.items.length === 0; }

push(item: T): boolean { if (this.isFull()) return false; this.items.push(item); return true; }

pushWithEviction(item: T): T | null { const evicted = this.isFull() ? this.items.shift() ?? null : null; this.items.push(item); return evicted; }

take(count: number): T[] { return this.items.splice(0, Math.min(count, this.items.length)); }

clear(): T[] { const all = this.items; this.items = []; return all; } }

Backpressure Controller

// controller.ts import { BoundedBuffer } from './buffer'; import { BackpressureState, BackpressureStrategy, BackpressureConfig, FlushFunction, } from './types';

const DEFAULT_CONFIG: BackpressureConfig = { maxBufferSize: 10000, highWatermark: 0.8, lowWatermark: 0.5, strategy: BackpressureStrategy.DROP_OLDEST, batchSize: 100, minFlushIntervalMs: 100, maxFlushIntervalMs: 30000, targetLatencyMs: 500, };

export class BackpressureController<T> { private buffer: BoundedBuffer<T>; private state: BackpressureState = BackpressureState.NORMAL; private config: BackpressureConfig; private flushFn: FlushFunction<T>; private flushInterval: NodeJS.Timeout | null = null; private currentFlushIntervalMs: number; private running = false;

// Metrics private eventsAccepted = 0; private eventsDropped = 0; private eventsFlushed = 0; private lastFlushLatencyMs = 0;

constructor(flushFn: FlushFunction<T>, config: Partial<BackpressureConfig> = {}) { this.config = { ...DEFAULT_CONFIG, ...config }; this.buffer = new BoundedBuffer(this.config.maxBufferSize); this.flushFn = flushFn; this.currentFlushIntervalMs = this.config.minFlushIntervalMs; }

start(): void { if (this.running) return; this.running = true; this.scheduleFlush(); }

stop(): void { this.running = false; if (this.flushInterval) { clearTimeout(this.flushInterval); this.flushInterval = null; } }

async push(item: T): Promise<boolean> { switch (this.config.strategy) { case BackpressureStrategy.BLOCK: if (this.state === BackpressureState.BLOCKED) { const waited = await this.waitForSpace(); if (!waited) { this.eventsDropped++; return false; } } break;

  case BackpressureStrategy.DROP_NEWEST:
    if (this.buffer.isFull()) {
      this.eventsDropped++;
      return false;
    }
    break;

  case BackpressureStrategy.DROP_OLDEST:
    if (this.buffer.isFull()) {
      this.buffer.pushWithEviction(item);
      this.eventsDropped++;
      this.eventsAccepted++;
      this.updateState();
      return true;
    }
    break;

  case BackpressureStrategy.SAMPLE:
    if (this.state !== BackpressureState.NORMAL) {
      const sampleRate = this.config.sampleRate || 10;
      if (this.eventsAccepted % sampleRate !== 0) {
        this.eventsDropped++;
        return false;
      }
    }
    break;
}

const accepted = this.buffer.push(item);
if (accepted) {
  this.eventsAccepted++;
} else {
  this.eventsDropped++;
}

this.updateState();
return accepted;

}

async drain(): Promise<void> { this.state = BackpressureState.DRAINING; while (!this.buffer.isEmpty()) { await this.flush(); } }

getMetrics() { return { state: this.state, bufferSize: this.buffer.size, bufferUtilization: this.buffer.utilization, eventsAccepted: this.eventsAccepted, eventsDropped: this.eventsDropped, eventsFlushed: this.eventsFlushed, lastFlushLatencyMs: this.lastFlushLatencyMs, }; }

private async flush(): Promise<void> { if (this.buffer.isEmpty()) return;

const batch = this.buffer.take(this.config.batchSize);
const startTime = Date.now();

try {
  const result = await this.flushFn(batch);
  this.eventsFlushed += result.success;
  this.lastFlushLatencyMs = Date.now() - startTime;
  this.adaptFlushInterval();
} catch (error) {
  console.error('[Backpressure] Flush error:', error);
}

this.updateState();

}

private scheduleFlush(): void { if (!this.running) return; this.flushInterval = setTimeout(async () => { await this.flush(); this.scheduleFlush(); }, this.currentFlushIntervalMs); }

private adaptFlushInterval(): void { const { targetLatencyMs, minFlushIntervalMs, maxFlushIntervalMs } = this.config;

if (this.lastFlushLatencyMs > targetLatencyMs * 1.5) {
  this.currentFlushIntervalMs = Math.min(
    this.currentFlushIntervalMs * 1.5,
    maxFlushIntervalMs
  );
} else if (this.lastFlushLatencyMs < targetLatencyMs * 0.5) {
  this.currentFlushIntervalMs = Math.max(
    this.currentFlushIntervalMs * 0.8,
    minFlushIntervalMs
  );
}

}

private updateState(): void { const util = this.buffer.utilization;

if (this.state === BackpressureState.DRAINING) return;

if (util >= 1.0) {
  this.state = BackpressureState.BLOCKED;
} else if (util >= this.config.highWatermark) {
  this.state = BackpressureState.CRITICAL;
} else if (util >= this.config.lowWatermark) {
  this.state = BackpressureState.ELEVATED;
} else {
  this.state = BackpressureState.NORMAL;
}

}

private async waitForSpace(): Promise<boolean> { const maxWait = this.config.maxBlockTimeMs || 5000; const startTime = Date.now();

while (Date.now() - startTime < maxWait) {
  if (!this.buffer.isFull()) return true;
  await new Promise(r => setTimeout(r, 50));
}
return false;

} }

Usage Examples

// Create controller const controller = new BackpressureController( async (items) => { const result = await db.batchInsert('events', items); return { success: result.inserted, failed: 0, errors: [] }; }, { strategy: BackpressureStrategy.DROP_OLDEST, maxBufferSize: 10000, batchSize: 100, } );

// Start processing controller.start();

// Push events await controller.push(event);

// On shutdown await controller.drain(); controller.stop();

Strategy Selection

Strategy Use Case Trade-off

BLOCK Critical data Producers slow down

DROP_OLDEST Time-series Lose historical data

DROP_NEWEST Batch jobs Reject new work

SAMPLE Telemetry Statistical accuracy

Best Practices

• Size buffers for memory - Don't exceed available RAM

• Match strategy to data - Critical data = BLOCK

• Monitor drop rates - Alert on high drops

• Drain on shutdown - Don't lose buffered data

• Combine with circuit breaker - Protect flush function

Common Mistakes

• Unbounded queues (OOM crash)

• No metrics on drops

• Not draining on shutdown

• Wrong strategy for data criticality

• No adaptive rate adjustment

Related Skills

• Circuit Breaker

• Background Jobs

• Health Checks