How to fix Error: Backpressure: queue size exceeded (async operations piling up) in Node.js

// Add diagnostics to track pending async operations
const pendingOps = new Set();

function trackAsync(name, promise) {
  const id = Math.random();
  pendingOps.add(id);
  console.log(`[${name}] Pending operations: ${pendingOps.size}`);

  return promise.finally(() => pendingOps.delete(id));
}

// Use in your code
trackAsync('db-query', db.query(sql));

// BAD: Ignoring backpressure
readableStream.on('data', (chunk) => {
  writableStream.write(chunk); // No check for return value
});

// GOOD: Respecting backpressure
readableStream.on('data', (chunk) => {
  const canContinue = writableStream.write(chunk);
  if (!canContinue) {
    // Buffer is full, pause reading
    readableStream.pause();
  }
});

writableStream.on('drain', () => {
  // Buffer is empty, resume reading
  readableStream.resume();
});

// BETTER: Use pipe() which handles backpressure automatically
readableStream.pipe(writableStream);

const { pipeline } = require('stream');

pipeline(
  readableStream,
  transformStream,
  writableStream,
  (err) => {
    if (err) console.error('Pipeline failed:', err);
  }
);

// BAD: Processing all items at once
const results = await Promise.all(
  items.map(item => processItem(item))
); // Can queue thousands of operations

// GOOD: Use p-limit for concurrency control
const pLimit = require('p-limit');
const limit = pLimit(5); // Only 5 concurrent operations

const results = await Promise.all(
  items.map(item => limit(() => processItem(item)))
);

const PQueue = require('p-queue').default;

const queue = new PQueue({
  concurrency: 5,
  interval: 1000,
  intervalCap: 10 // Rate limiting: max 10 per second
});

for (const item of items) {
  queue.add(() => processItem(item));
}

await queue.onIdle(); // Wait for all to complete

const async = require('async');

const queue = async.queue(async (task) => {
  await processTask(task);
}, 5); // 5 concurrent workers

// Set a maximum queue size
const MAX_QUEUE_SIZE = 100;

function addToQueue(task) {
  if (queue.length() >= MAX_QUEUE_SIZE) {
    throw new Error('Queue size exceeded - rejecting new work');
  }
  return new Promise((resolve, reject) => {
    queue.push(task, (err, result) => {
      if (err) reject(err);
      else resolve(result);
    });
  });
}

const Queue = require('bull');

const myQueue = new Queue('work', {
  redis: { port: 6379, host: '127.0.0.1' },
  limiter: {
    max: 100, // Max 100 jobs
    duration: 1000 // per second
  }
});

myQueue.process(5, async (job) => {
  return processWork(job.data);
});

const Bottleneck = require('bottleneck');

// Limit to 10 concurrent operations, max 50 per minute
const limiter = new Bottleneck({
  maxConcurrent: 10,
  minTime: 1200, // Minimum 1.2s between operations (50/min)
  reservoir: 50, // Initial capacity
  reservoirRefreshAmount: 50,
  reservoirRefreshInterval: 60 * 1000 // Refill every minute
});

// Wrap your async functions
const rateLimitedDbQuery = limiter.wrap(async (query) => {
  return await db.execute(query);
});

// Use throughout your app
const result = await rateLimitedDbQuery('SELECT * FROM users');

class MonitoredQueue {
  constructor(maxSize = 1000, concurrency = 10) {
    this.maxSize = maxSize;
    this.queue = [];
    this.processing = 0;
    this.concurrency = concurrency;
    this.circuitOpen = false;
  }

  async add(task) {
    if (this.circuitOpen) {
      throw new Error('Circuit breaker open - queue overloaded');
    }

    if (this.queue.length >= this.maxSize) {
      this.circuitOpen = true;
      setTimeout(() => this.circuitOpen = false, 60000); // Reset after 1 min
      throw new Error('Queue overflow - circuit breaker activated');
    }

    this.queue.push(task);
    this.processNext();
  }

  async processNext() {
    if (this.processing >= this.concurrency || this.queue.length === 0) {
      return;
    }

    this.processing++;
    const task = this.queue.shift();

    try {
      await task();
    } finally {
      this.processing--;
      this.processNext();
    }
  }

  getStats() {
    return {
      queueSize: this.queue.length,
      processing: this.processing,
      circuitOpen: this.circuitOpen
    };
  }
}