Consumer Pool Patterns in JavaScript

Multiple workers pulling from a shared queue — with variants for blocking producers when the queue gets too full and deduplicating work at enqueue time to keep the queue from blowing up.

Frontier-based async consumer pool — multiple cooperative consumers pulling from a shared queue.

This creates n workers to implicitly address execution backpressure, not discovery backpressure. It can limit in-flight work, but cannot limit queue growth.

JavaScript

const queue = [seedItem];
let qIndex = 0;

async function worker(workerId) {
  while (true) {
    if (qIndex >= queue.length) return;

const item = queue[qIndex++];
    try {
      // `process` can push more items into `queue`
      await process(item, workerId, queue);
    } catch (err) {
      onError(err, item, workerId);
    }
  }
}

const n = Math.max(1, concurrency | 0);
await Promise.all(Array.from({ length: n }, (_, i) => worker(i)));

return queue;

Block Producers when queue is full

JavaScript

const queue = [seedItem];
let qIndex = 0;

// producers waiting for space
const waiters = [];

function notifySpace() {
  const resolve = waiters.shift();
  if (resolve) resolve();
}

async function enqueue(item) {
  // block while queue is "full"
  while (queue.length - qIndex >= maxQueueSize) {
    await new Promise(resolve => waiters.push(resolve));
  }
  queue.push(item);
}

async function worker(workerId) {
  while (true) {
    if (qIndex >= queue.length) return;

const item = queue[qIndex++];
    notifySpace(); // consuming frees capacity

try {
      // process can enqueue more work
      await process(item, workerId, enqueue);
    } catch (err) {
      onError(err, item, workerId);
    }
  }
}

const n = Math.max(1, concurrency | 0);
await Promise.all(Array.from({ length: n }, (_, i) => worker(i)));

return queue;

Avoid duplicate work

The important thing to note is that deduping happens at enqueue time; otherwise the queue can blow up.

JavaScript

const queue = [seedItem];
let qIndex = 0;

// "seen" tracks items already enqueued
const seen = new Set([seedItem]);

function enqueue(item) {
  if (seen.has(item)) return false;
  seen.add(item);
  queue.push(item);
  return true;
}

async function worker(workerId) {
  while (true) {
    if (qIndex >= queue.length) return;

const item = queue[qIndex++];

try {
      // process can enqueue more items via `enqueue`
      await process(item, workerId, enqueue);
    } catch (err) {
      onError(err, item, workerId);
    }
  }
}

const n = Math.max(1, concurrency | 0);
await Promise.all(Array.from({ length: n }, (_, i) => worker(i)));

return queue;        // all discovered items
// return seen;      // alternatively, return deduped set