Async Queue

Run a limited number of jobs at once, retry after a failure, and start the highest priority first.

JAVASCRIPT Free to use

Code

class AsyncQueue {
    constructor({ concurrency = 2, retries = 0, retryDelay = 250 } = {}) {
        if (!Number.isInteger(concurrency) || concurrency < 1) {
            throw new RangeError('concurrency must be a positive integer.');
        }
        if (!Number.isInteger(retries) || retries < 0) {
            throw new RangeError('retries cannot be negative.');
        }

        this.concurrency = concurrency;
        this.retries = retries;
        this.retryDelay = retryDelay;
        this.items = [];
        this.active = 0;
        this.delayed = 0;
        this.idleWaiters = [];
    }

    add(job, priority = 0) {
        return new Promise((resolve, reject) => {
            this.insert({ job, priority, resolve, reject, attempt: 0 });
            this.pump();
        });
    }

    idle() {
        if (this.isIdle()) {
            return Promise.resolve();
        }

        return new Promise((resolve) => {
            this.idleWaiters.push(resolve);
        });
    }

    get size() {
        return this.items.length;
    }

    get running() {
        return this.active;
    }

    isIdle() {
        return this.active === 0 && this.items.length === 0 && this.delayed === 0;
    }

    insert(item) {
        const index = this.items.findIndex((entry) => entry.priority < item.priority);
        if (index === -1) {
            this.items.push(item);
            return;
        }
        this.items.splice(index, 0, item);
    }

    pump() {
        while (this.active < this.concurrency && this.items.length > 0) {
            const item = this.items.shift();
            this.active += 1;

            this.run(item).finally(() => {
                this.active -= 1;
                this.pump();
                this.notifyIdle();
            });
        }

        this.notifyIdle();
    }

    async run(item) {
        try {
            item.resolve(await item.job());
        } catch (error) {
            if (item.attempt >= this.retries) {
                item.reject(error);
                return;
            }

            item.attempt += 1;
            this.delayed += 1;

            setTimeout(() => {
                this.delayed -= 1;
                this.insert(item);
                this.pump();
            }, this.retryDelay * item.attempt);
        }
    }

    notifyIdle() {
        if (!this.isIdle() || this.idleWaiters.length === 0) {
            return;
        }

        const waiters = this.idleWaiters;
        this.idleWaiters = [];
        waiters.forEach((resolve) => resolve());
    }
}

Quick try

const queue = new AsyncQueue({ concurrency: 2, retries: 2, retryDelay: 200 });

const later = queue.add(() => fetch('/slow').then((response) => response.json()), 1);
const first = queue.add(() => fetch('/urgent').then((response) => response.json()), 10);

const results = await Promise.all([later, first]);
await queue.idle();

What it does

The pump starts another job when a slot is free. There is no processing lock that stalls the rest of the queue.
A retry leaves the active count during the wait, then is inserted by priority.
idle waits until nothing is running or delayed. This is a browser or Node queue, not a server worker queue.

Where it fits

A practical place for this snippet in a real project.

Useful for paced requests from the UI, or a limited number of uploads at once. It is not a replacement for Redis or Supervisor on the server.

Similar snippets

JAVASCRIPT

Binary Search Tree

Insert, search, and delete while keeping values sorted. Average O(log n), worst O(n) if the tree becomes a chain.

JAVASCRIPT

Function Memoization

Wrap a function and reuse its result for the same inputs. The next call with the same values skips the work.

JAVASCRIPT

LRU Cache

Keep a fixed number of entries and evict the least recently used item when the cache is full.

Need a snippet that is not here?

Describe the problem or idea, and I will see if it belongs in the library.

Contact me Library