Repair Backpressure & Overload Handling

I have analyzed the consumer implementation and identified an unbounded array buffer that causes memory exhaustion when the downstream database slows down.

Backpressure Implementation

const { EventEmitter } = require('events');

class BoundedQueue extends EventEmitter {
  constructor(limit = 50) {
    super();
    this.limit = limit;
    this.queue = [];
    this.isPaused = false;
  }

  async push(item, processor) {
    this.queue.push(item);
    
    // Invariant: Memory usage is bounded by this.limit
    if (this.queue.length >= this.limit && !this.isPaused) {
      this.isPaused = true;
      this.emit('pause');
    }

    // Process immediately or in background
    setImmediate(() => this.process(processor));
  }

  async process(processor) {
    if (this.queue.length === 0) return;
    
    const item = this.queue.shift();
    try {
      await processor(item);
    } finally {
      // Resume if we have dropped below the high water mark
      if (this.isPaused && this.queue.length < this.limit) {
        this.isPaused = false;
        this.emit('drain');
      }
    }
  }
}

// Integration Example
const queue = new BoundedQueue(50);
source.on('data', (chunk) => {
  queue.push(chunk, db.save);
  if (queue.isPaused) source.pause();
});

queue.on('drain', () => {
  source.resume();
});

Next steps

1. Review the drain event timing to ensure it matches your source's resume latency.
2. Add a timeout to db.save to prevent a single hanging promise from locking the queue.
3. Monitor process.memoryUsage().heapUsed during the next load test to verify the 50-item ceiling.