Build Your Own Async Queue

You have 200 images to resize. The resize API is fast, but it throttles at 10 concurrent requests. If you throw Promise.all(images.map(resize)) at it, you'll blast all 200 simultaneously, collect a pile of 429s, and watch your retry logic melt. If you serialize them with a for-await loop, you're using one connection at a time — maybe 5% of the throughput you could be getting.

What you want: run at most N tasks concurrently, queue the rest, drain as slots open. Like this:

const queue = new AsyncQueue(10); // max 10 in-flight at once
 
const results = await Promise.all(
  images.map(img => queue.add(() => resize(img)))
);

That's the whole API. Here's how to build it from scratch in about 60 lines of JavaScript.

What We're Building

• AsyncQueue(concurrency) — bounded concurrent execution, default 1
• .add(fn) — accepts a function (not a promise), returns a promise that resolves/rejects with the task result
• .pause() / .resume() — stop draining and restart
• .onIdle() — a promise that resolves once all running and queued tasks are done
• .size / .pending — inspect queue depth and in-flight count

The key design decision: callers pass () => fetch(url), not fetch(url). A bare promise starts executing the moment it's created — you can't delay it. A function that returns a promise only starts work when called. The queue calls it when a slot is free.

Step 1: The Skeleton

// async-queue.js
 
class AsyncQueue {
  constructor(concurrency = 1) {
    if (concurrency < 1) throw new RangeError('concurrency must be >= 1');
    this.concurrency = concurrency;
    this._running = 0;        // tasks currently executing
    this._queue = [];         // { fn, resolve, reject }
    this._paused = false;
    this._idleCallbacks = []; // subscribers to onIdle()
  }
 
  get size() { return this._queue.length; }
  get pending() { return this._running; }
}

_running tracks how many tasks are in-flight. _queue is a plain array of waiting tasks — we always take from the front (FIFO), so a plain array with shift() is fine. _idleCallbacks holds resolve functions for anyone awaiting onIdle().

Step 2: add()

class AsyncQueue {
  // ... constructor above
 
  add(fn) {
    return new Promise((resolve, reject) => {
      this._queue.push({ fn, resolve, reject });
      this._next();
    });
  }
}

Every add call creates a promise and stores its resolve and reject callbacks alongside the task function. This is how the queue later delivers results back to the specific caller — when task fn finishes, the queue calls the resolve that was captured at queue time.

_next() is called immediately after pushing. If a slot is free, the task starts right now and never enters the waiting queue long-term — shift() in _next() will pick it up in the same tick.

Step 3: _next() — the core dispatcher

class AsyncQueue {
  // ...
 
  _next() {
    while (!this._paused && this._running < this.concurrency && this._queue.length > 0) {
      const { fn, resolve, reject } = this._queue.shift();
      this._running++;
 
      Promise.resolve()
        .then(() => fn())
        .then(resolve, reject)
        .finally(() => {
          this._running--;
          this._next();
 
          if (this._running === 0 && this._queue.length === 0) {
            const cbs = this._idleCallbacks.splice(0);
            for (const cb of cbs) cb();
          }
        });
    }
  }
}

The while loop fills as many free slots as the waiting queue allows. Three conditions gate each iteration: not paused, below the concurrency cap, queue is non-empty.

Promise.resolve().then(() => fn()) looks like an unnecessary detour — why not just call fn() directly? Two reasons. First, it defers execution to a microtask, so a synchronous fn() that throws is caught by .then(resolve, reject) rather than propagating up through _next() and crashing the caller in unexpected ways. Second, it normalizes both sync and async tasks: fn() can return a plain value, a Promise, or throw synchronously, and .then() handles all three uniformly.

In .finally(): decrement _running, call _next() to fill the freed slot, then fire idle callbacks if both _running === 0 and _queue.length === 0. The splice(0) clears the array while capturing the callbacks — the queue resets cleanly even if a new onIdle() subscriber is added inside a callback.

Step 4: pause() and resume()

class AsyncQueue {
  // ...
 
  pause() {
    this._paused = true;
  }
 
  resume() {
    if (!this._paused) return;
    this._paused = false;
    this._next();
  }
}

pause() flips the flag. The while condition in _next() checks it — any in-flight tasks run to completion, but nothing new gets dequeued. When resume() clears the flag, it calls _next() to start draining again.

Step 5: onIdle()

class AsyncQueue {
  // ...
 
  onIdle() {
    if (this._running === 0 && this._queue.length === 0) {
      return Promise.resolve();
    }
    return new Promise(resolve => {
      this._idleCallbacks.push(resolve);
    });
  }
}

If the queue is already empty and nothing is running, resolve immediately. Otherwise, register a callback that _next() will fire when the last task completes.

You can call onIdle() multiple times — each call pushes a new entry into _idleCallbacks, and all of them fire together. This is useful when multiple consumers are each waiting for the queue to drain.

Full Implementation

// async-queue.js
 
class AsyncQueue {
  constructor(concurrency = 1) {
    if (concurrency < 1) throw new RangeError('concurrency must be >= 1');
    this.concurrency = concurrency;
    this._running = 0;
    this._queue = [];
    this._paused = false;
    this._idleCallbacks = [];
  }
 
  get size() { return this._queue.length; }
  get pending() { return this._running; }
 
  add(fn) {
    return new Promise((resolve, reject) => {
      this._queue.push({ fn, resolve, reject });
      this._next();
    });
  }
 
  _next() {
    while (!this._paused && this._running < this.concurrency && this._queue.length > 0) {
      const { fn, resolve, reject } = this._queue.shift();
      this._running++;
 
      Promise.resolve()
        .then(() => fn())
        .then(resolve, reject)
        .finally(() => {
          this._running--;
          this._next();
          if (this._running === 0 && this._queue.length === 0) {
            const cbs = this._idleCallbacks.splice(0);
            for (const cb of cbs) cb();
          }
        });
    }
  }
 
  pause() {
    this._paused = true;
  }
 
  resume() {
    if (!this._paused) return;
    this._paused = false;
    this._next();
  }
 
  onIdle() {
    if (this._running === 0 && this._queue.length === 0) {
      return Promise.resolve();
    }
    return new Promise(resolve => {
      this._idleCallbacks.push(resolve);
    });
  }
}
 
module.exports = { AsyncQueue };

Demo: Rate-Limited API Calls

// demo.js — node demo.js
 
const { AsyncQueue } = require('./async-queue');
 
function simulatedResize(id) {
  const delay = Math.floor(Math.random() * 300) + 100;
  return new Promise(resolve =>
    setTimeout(() => resolve({ id, ms: delay }), delay)
  );
}
 
async function main() {
  const queue = new AsyncQueue(3); // 3 concurrent at most
  const imageIds = Array.from({ length: 12 }, (_, i) => i + 1);
 
  const promises = imageIds.map(id =>
    queue.add(() => simulatedResize(id)).then(result => {
      console.log(
        `image ${String(result.id).padStart(2)} done in ${result.ms}ms` +
        ` | running=${queue.pending} queued=${queue.size}`
      );
      return result;
    })
  );
 
  await queue.onIdle();
  const results = await Promise.all(promises);
  console.log(`\nProcessed ${results.length} images.`);
}
 
main();

Run it and you'll see queued count down one at a time as slots open, and the ids complete out of order — task 8 can finish before task 5 if it draws a shorter delay.

One quirk in that output: running reads 3 on almost every line. The caller's .then is queued the moment resolve is called, which is one microtask before the .finally that decrements _running. So you're observing the counter from inside the task's own completion, before its slot has been released. Log from onIdle() or a setTimeout if you want the settled numbers.

Testing

// test.js — node test.js
 
const { AsyncQueue } = require('./async-queue');
 
function assert(condition, message) {
  if (!condition) throw new Error(`FAIL: ${message}`);
  console.log(`PASS: ${message}`);
}
 
function sleep(ms) { return new Promise(r => setTimeout(r, ms)); }
 
async function run() {
  // Concurrency cap is enforced — and actually reached.
  // `<= 2` alone would also pass if the queue ran everything serially.
  const q1 = new AsyncQueue(2);
  let maxConcurrent = 0, concurrent = 0;
  const started = [];
 
  const tasks = Array.from({ length: 8 }, (_, i) =>
    q1.add(async () => {
      started.push(i);
      concurrent++;
      maxConcurrent = Math.max(maxConcurrent, concurrent);
      // Later tasks finish sooner, so completion order != start order
      await sleep(40 - i * 4);
      concurrent--;
      return i;
    })
  );
 
  const results = await Promise.all(tasks);
  assert(maxConcurrent === 2, `concurrency cap reached and never exceeded (max: ${maxConcurrent})`);
  // FIFO has to be measured at dequeue time. `results` is ordered by
  // Promise.all regardless of what the queue did, so it proves nothing.
  assert(started.join(',') === '0,1,2,3,4,5,6,7', 'tasks are dequeued in FIFO order');
  assert(results.join(',') === '0,1,2,3,4,5,6,7', 'each caller gets its own task result');
 
  // Rejection propagates to the right promise and does not stop the queue
  const q2 = new AsyncQueue(2);
  const fail = q2.add(() => Promise.reject(new Error('boom')));
  const ok   = q2.add(() => Promise.resolve(42));
  const [r1, r2] = await Promise.allSettled([fail, ok]);
  assert(r1.status === 'rejected' && r1.reason.message === 'boom', 'rejection reaches correct caller');
  assert(r2.status === 'fulfilled' && r2.value === 42, 'queue keeps running after a rejection');
 
  // pause/resume
  const q3 = new AsyncQueue(1);
  q3.pause();
  const log = [];
  q3.add(() => { log.push('A'); return Promise.resolve(); });
  q3.add(() => { log.push('B'); return Promise.resolve(); });
  assert(log.length === 0, 'no tasks run while paused');
  q3.resume();
  await q3.onIdle();
  assert(log.join('') === 'AB', 'tasks run in FIFO order after resume');
 
  // onIdle resolves immediately on an empty queue
  const q4 = new AsyncQueue(1);
  const t0 = Date.now();
  await q4.onIdle();
  assert(Date.now() - t0 < 10, 'onIdle resolves immediately when already idle');
 
  // Multiple onIdle() subscribers all fire
  const q5 = new AsyncQueue(1);
  q5.add(() => sleep(30));
  const [a, b] = await Promise.all([q5.onIdle(), q5.onIdle()]);
  assert(a === undefined && b === undefined, 'multiple onIdle() subscribers all resolve');
 
  console.log('\nAll tests passed.');
}
 
run().catch(err => { console.error(err); process.exit(1); });

What We Skipped

Priority queues: All tasks above are FIFO. Real workloads often need priority tiers — batch jobs at low priority, user-facing requests at high priority. add(fn, { priority: 10 }) enqueues with that priority, and _queue.shift() becomes "take the highest-priority waiting task" — either a sorted insert on the existing array or a proper heap if queue depth gets large enough for the O(n) insert to matter. p-queue supports this out of the box if you'd rather not.

Rate limiting: Concurrency controls simultaneous tasks. Rate limiting controls frequency — "no more than 10 tasks per second". These are different constraints. A task that completes in 50ms uses its concurrency slot for 50ms; under rate limiting, it still consumes one of your 10/sec allowance even if it finishes instantly. Implementing rate limiting requires a token bucket or sliding-window timer in _next(), not just a counter.

Task cancellation: add(fn, { signal }) — if the AbortSignal fires before the task starts executing, reject the promise without calling fn. Check signal?.aborted in _next() immediately before calling fn(). If it fires during execution, that's the task's responsibility to handle internally. On cancellation, also call _next() — the cancelled task freed a logical slot.

Backpressure: If producers add tasks faster than the queue drains, _queue grows without bound. Add a maxSize constructor option. When _queue.length >= maxSize, add() returns a rejected promise or awaits a secondary "add backpressure" queue. Useful when callers are tight loops that could buffer unbounded work.

Error recovery / retry: The queue doesn't retry failed tasks. For automatic retry, wrap the task: queue.add(() => withRetry(fn, { retries: 3 })). The retry logic is fully external to the queue — that's a feature, not a limitation. The queue's job is concurrency control, not error policy.

The bounded-pool-plus-work-queue structure appears everywhere in systems programming: HTTP connection pools, thread pools, database connection pools, browser resource schedulers. Once you recognize the pattern, you see it in most infrastructure code.