Request Queuing: Sequential API Calls with Promise Queue Implementation

Problem

When making multiple API calls, you often face these challenges:

• Rate limiting: APIs restrict concurrent requests
• Resource exhaustion: Too many simultaneous connections overwhelm the system
• Order dependency: Some requests must complete before others start
• Error cascading: One failure shouldn't crash the entire pipeline
• Memory pressure: Unlimited concurrent requests consume excessive memory

Without queuing, you might fire 1000 requests simultaneously, causing timeouts, 429 errors, or system crashes.

Solution

Implement a Promise Queue that:

1. Maintains a queue of pending tasks
2. Executes tasks sequentially or with controlled concurrency
3. Respects rate limits and resource constraints
4. Handles errors gracefully
5. Provides progress tracking and cancellation

Code Implementation

Basic Sequential Queue

class PromiseQueue {
  constructor(concurrency = 1) {
    this.concurrency = concurrency;
    this.running = 0;
    this.queue = [];
  }

  async add(task) {
    return new Promise((resolve, reject) => {
      this.queue.push({ task, resolve, reject });
      this.process();
    });
  }

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

    this.running++;
    const { task, resolve, reject } = this.queue.shift();

    try {
      const result = await task();
      resolve(result);
    } catch (error) {
      reject(error);
    } finally {
      this.running--;
      this.process();
    }
  }
}

// Usage
const queue = new PromiseQueue(3); // Max 3 concurrent requests

async function fetchUser(id) {
  return queue.add(() => 
    fetch(`/api/users/${id}`).then(r => r.json())
  );
}

// Queue 100 requests, only 3 run concurrently
const userIds = Array.from({ length: 100 }, (_, i) => i + 1);
const results = await Promise.all(
  userIds.map(id => fetchUser(id))
);

Advanced Queue with Features

class AdvancedPromiseQueue {
  constructor(options = {}) {
    this.concurrency = options.concurrency || 1;
    this.timeout = options.timeout || 30000;
    this.retries = options.retries || 0;
    this.retryDelay = options.retryDelay || 1000;

    this.running = 0;
    this.queue = [];
    this.completed = 0;
    this.failed = 0;
    this.results = [];
    this.errors = [];
  }

  async add(task, priority = 0) {
    return new Promise((resolve, reject) => {
      const item = { task, resolve, reject, priority, retries: 0 };

      // Insert by priority (higher priority first)
      const insertIndex = this.queue.findIndex(
        q => q.priority < priority
      );

      if (insertIndex === -1) {
        this.queue.push(item);
      } else {
        this.queue.splice(insertIndex, 0, item);
      }

      this.process();
    });
  }

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

    this.running++;
    const item = this.queue.shift();

    try {
      const result = await this.executeWithTimeout(
        item.task,
        this.timeout
      );

      item.resolve(result);
      this.results.push(result);
      this.completed++;
    } catch (error) {
      if (item.retries < this.retries) {
        item.retries++;
        await this.delay(this.retryDelay * item.retries);
        this.queue.unshift(item); // Re-queue at front
      } else {
        item.reject(error);
        this.errors.push({ task: item.task, error });
        this.failed++;
      }
    } finally {
      this.running--;
      this.process();
    }
  }

  async executeWithTimeout(task, timeout) {
    return Promise.race([
      task(),
      new Promise((_, reject) =>
        setTimeout(
          () => reject(new Error(`Task timeout after ${timeout}ms`)),
          timeout
        )
      )
    ]);
  }

  delay(ms) {
    return new Promise(resolve => setTimeout(resolve, ms));
  }

  getStats() {
    return {
      queued: this.queue.length,
      running: this.running,
      completed: this.completed,
      failed: this.failed,
      total: this.completed + this.failed + this.queue.length
    };
  }

  clear() {
    this.queue = [];
  }
}

// Usage with retries and timeout
const queue = new AdvancedPromiseQueue({
  concurrency: 5,
  timeout: 10000,
  retries: 3,
  retryDelay: 500
});

async function fetchWithQueue(url, priority = 0) {
  return queue.add(
    () => fetch(url).then(r => r.json()),
    priority
  );
}

// High priority requests
await fetchWithQueue('/api/critical', 10);

// Normal priority
await fetchWithQueue('/api/data', 0);

console.log(queue.getStats());

Real-World Example: Batch API Processing

class APIBatcher {
  constructor(apiClient, options = {}) {
    this.apiClient = apiClient;
    this.queue = new AdvancedPromiseQueue({
      concurrency: options.concurrency || 5,
      timeout: options.timeout || 15000,
      retries: options.retries || 2
    });
  }

  async fetchUsers(userIds) {
    const tasks = userIds.map(id => ({
      id,
      promise: this.queue.add(() => 
        this.apiClient.getUser(id)
      )
    }));

    const results = await Promise.allSettled(
      tasks.map(t => t.promise)
    );

    return tasks.map((task, index) => ({
      id: task.id,
      status: results[index].status,
      data: results[index].value,
      error: results[index].reason
    }));
  }

  async processInBatches(items, batchSize = 100) {
    const batches = [];

    for (let i = 0; i < items.length; i += batchSize) {
      const batch = items.slice(i, i + batchSize);
      batches.push(this.fetchUsers(batch));
    }

    return Promise.all(batches).then(results => 
      results.flat()
    );
  }
}

// Usage
const batcher = new APIBatcher(apiClient, { concurrency: 10 });

const userIds = Array.from({ length: 5000 }, (_, i) => i + 1);
const results = await batcher.processInBatches(userIds, 500);

const successful = results.filter(r => r.status === 'fulfilled');
const failed = results.filter(r => r.status === 'rejected');

console.log(`Processed: ${successful.length}, Failed: ${failed.length}`);

Tips & Best Practices

1. Choose Right Concurrency Level

// CPU-bound: cores count
const cpuConcurrency = require('os').cpus().length;

// I/O-bound: higher is usually better
const ioConcurrency = 10-50;

// API rate limit: requests per second
const rateLimitConcurrency = Math.floor(rateLimit / timeWindow);

2. Implement Exponential Backoff

const delay = (attempt) => 
  Math.min(1000 * Math.pow(2, attempt), 30000) + 
  Math.random() * 1000;

3. Monitor Queue Health

setInterval(() => {
  const stats = queue.getStats();
  if (stats.queued > 1000) {
    console.warn('Queue backlog detected');
  }
}, 5000);

4. Handle Partial Failures

const results = await Promise.allSettled(promises);
const successful = results.filter(r => r.status === 'fulfilled');
const failed = results.filter(r => r.status === 'rejected');

5. Use Priority Queues for Mixed Workloads

// Critical requests first
await queue.add(criticalTask, priority = 100);

// Normal requests
await queue.add(normalTask, priority = 0);

// Background tasks
await queue.add(backgroundTask, priority = -100);

6. Implement Circuit Breaker Pattern

class CircuitBreaker {
  constructor(threshold = 5, timeout = 60000) {
    this.failures = 0;
    this.threshold = threshold;
    this.timeout = timeout;
    this.state = 'CLOSED'; // CLOSED, OPEN, HALF_OPEN
  }

  async execute(task) {
    if (this.state === 'OPEN') {
      throw new Error('Circuit breaker is OPEN');
    }

    try {
      const result = await task();
      this.onSuccess();
      return result;
    } catch (error) {
      this.onFailure();
      throw error;
    }
  }

  onSuccess() {
    this.failures = 0;
    this.state = 'CLOSED';
  }

  onFailure() {
    this.failures++;
    if (this.failures >= this.threshold) {
      this.state = 'OPEN';
      setTimeout(() => {
        this.state = 'HALF_OPEN';
      }, this.timeout);
    }
  }
}

7. Memory Management for Large Queues

// Stream processing instead of loading all at once
async function* processStream(items, batchSize = 100) {
  for (let i = 0; i < items.length; i += batchSize) {
    const batch = items.slice(i, i + batchSize);
    yield await queue.add(() => processBatch(batch));
  }
}

for await (const result of processStream(largeDataset)) {
  // Process one batch at a time
}

Summary

A Promise Queue is essential for:

• ✅ Respecting API rate limits
• ✅ Preventing resource exhaustion
• ✅ Handling errors gracefully
• ✅ Maintaining system stability
• ✅ Improving throughput predictability

Choose concurrency based on your constraints, implement retries for resilience, and monitor queue health in production.