TaskQueue

A robust, TypeScript-based task queue implementation with concurrency control, deduplication, and automatic retry capabilities with exponential backoff.

🚀 Features

• Concurrency Control — Limit the number of simultaneous operations
• Request Deduplication — Automatically deduplicate tasks by key
• Automatic Retries — Configurable retry mechanism with exponential backoff
• Jitter Support — Adds randomness to retry delays to prevent “thundering herd” issues
• Comprehensive Metrics — Detailed statistics for queue performance monitoring
• Graceful Shutdown — Safely cancel pending and complete active tasks
• Type Safety — Fully typed implementation with generics for ergonomic and reliable use

📦 Installation

npm i async-scheduler-js

or

yarn add async-scheduler-js

💡 Usage

Basic Example

import { TaskQueue } from 'async-scheduler-js';

// Create a queue with concurrency of 3
const queue = new TaskQueue<string>(3);

// Schedule tasks
const result1 = await queue.schedule('task-1', async (key) => {
    return `Result for ${key}`;
});

const result2 = await queue.schedule('task-2', async (key) => {
    return `Result for ${key}`;
});

Example with Retry Configuration

const queue = new TaskQueue<number>(5, {
    maxRetries: 5,
    baseDelay: 1000,
    baseJitter: 500
});

const result = await queue.schedule('expensive-operation', async (key) => {
    // This will be retried up to 5 times with exponential backoff
    return await performExpensiveOperation(key);
});

🧭 API Reference

Constructor

new TaskQueue<T>(concurrency: number, options?: ITaskQueueOptions)

Parameters:

• concurrency: Maximum number of concurrent operations
• options: Optional configuration
  • maxRetries: Maximum retry attempts (default: 3)
  • baseDelay: Base delay for exponential backoff, in milliseconds (default: 1000)
  • baseJitter: Jitter range, in milliseconds (default: 100)

Methods

schedule(key: string, task: (key: string) => Promise<T>): Promise<T>

Schedules a task for execution.
If a task with the same key is already queued, returns the existing promise.

Parameters:

• key: Unique identifier for the task (used for deduplication)
• task: Async function that performs the actual work

Returns:
A Promise that resolves with the task result.

shutdown(): Promise<void>

Initiates a graceful shutdown.

• Prevents new tasks from being scheduled
• Cancels all pending tasks
• Waits for active tasks to complete

awaiter(ms: number): Promise<void>

Utility helper that resolves after a specified number of milliseconds.

📊 Properties

Queue State

• isShutdown: Indicates whether the queue is shutting down or has already shut down
• pending: Current number of pending operations

Statistics

• enqueued: Total tasks enqueued
• deduped: Number of deduplicated tasks (same key)
• started: Started executions
• retried: Total retry attempts
• succeeded: Successfully completed tasks
• errored: Tasks that failed (including retries)
• permanentError: Tasks that failed after maximum retries
• cancelled: Tasks cancelled during shutdown

🧩 Internal Types

export const enum eOperationStatus {
    Pending,
    Running,
    Retrying,
    Done
}

export interface ITaskQueueOptions {
    maxRetries?: number;
    baseDelay?: number;
    baseJitter?: number;
}

export interface IOperation<T> {
    promise: Promise<T>;
    resolve: (value: T) => void;
    reject: (reason?: any) => void;
    key: string;
    task: (key: string) => Promise<T>;
    attempts: number;
    status: eOperationStatus;
    timer?: NodeJS.Timeout;
}

🔁 Retry Mechanism

The queue implements exponential backoff with jitter:

• Base Delay — configurable initial delay (default: 1000ms)
• Exponential Backoff — each retry multiplies the delay (×2 per attempt)
• Jitter — random delay added to prevent synchronized retries

Delay formula:

baseDelay * 2^(attempt - 1) + random(0, baseJitter)

⚠️ Error Handling

• Temporary failures are retried up to maxRetries
• Permanent failures (after maximum retries) reject the promise
• During shutdown, pending tasks are cancelled and their promises rejected

🌐 Example: API Rate Limiting

const apiQueue = new TaskQueue<any>(2, {
    maxRetries: 3,
    baseDelay: 2000
});

async function fetchWithRetry(url: string) {
    return apiQueue.schedule(url, async (key) => {
        const response = await fetch(key);
        if (!response.ok) throw new Error(`HTTP ${response.status}`);
        return response.json();
    });
}

// Usage — limited to 2 concurrent requests
const results = await Promise.all([
    fetchWithRetry('https://api.example.com/data1'),
    fetchWithRetry('https://api.example.com/data2'),
    fetchWithRetry('https://api.example.com/data3')
]);

📈 Performance Monitoring

const queue = new TaskQueue(3);

// Monitor queue performance
setInterval(() => {
    console.log({
        active: queue.activeCount,
        pending: queue.pending,
        succeeded: queue.succeeded,
        errored: queue.errored,
        deduped: queue.deduped
    });
}, 5000);

🔗 Links

• 📘 Documentation
• 🧩 NPM Package
• 🧠 Issues & Feedback
• 📜 License