A TypeScript library for managing HTTP request queues with retry and backpressure features. Built for Bun runtime using Redis for fast queue operations and PostgreSQL for durable state persistence.
- π Push-based Workers - Real-time processing with Redis pub/sub
- π Configurable Retry Strategies - Exponential, linear, fixed, or custom backoff
- π‘οΈ Backpressure Control - Rate limiting, circuit breaker, and concurrency limits
- πΎ Dual Storage - Redis for speed, PostgreSQL for durability
- π Full Observability - Request tracking, attempt logging, and statistics
- π Dead Letter Queue - Failed requests preserved for manual retry
bun add http-queue-manager
import { createQueueManager } from 'http-queue-manager';
const queue = await createQueueManager({
redis: { url: 'redis://localhost:6379' },
postgres: { connectionString: 'postgresql://user:pass@localhost:5432/queue' },
});
// Enqueue a request
const { id } = await queue.enqueue({
url: 'https://api.example.com/webhooks',
method: 'POST',
body: { event: 'user.created', data: { userId: '123' } },
});
// Handle completions
queue.onComplete((response) => {
console.log(`Request ${response.requestId} completed with status ${response.status}`);
});
// Handle failures
queue.onDead((requestId, error) => {
console.error(`Request ${requestId} failed permanently: ${error.message}`);
});
// Graceful shutdown
process.on('SIGTERM', () => queue.shutdown());
import { createQueueManager } from 'http-queue-manager';
const queue = await createQueueManager({
// Redis connection
redis: {
url: 'redis://localhost:6379',
// Or individual options:
// host: 'localhost',
// port: 6379,
// password: 'secret',
// db: 0,
keyPrefix: 'myapp:', // Optional prefix for all keys
},
// PostgreSQL connection
postgres: {
connectionString: 'postgresql://...',
// Or individual options:
// host: 'localhost',
// port: 5432,
// database: 'queue',
// user: 'postgres',
// password: 'secret',
poolSize: 10,
},
// Processing options
concurrency: 10, // Max concurrent requests
workerCount: 1, // Number of workers
// Retry configuration
retry: {
strategy: 'exponential', // 'exponential' | 'linear' | 'fixed' | 'custom'
maxRetries: 3,
baseDelay: 1000, // 1 second
maxDelay: 30000, // 30 seconds
jitter: true, // Add randomization
retryOn: [408, 429, 500, 502, 503, 504], // Status codes to retry
},
// Backpressure configuration
backpressure: {
maxConcurrency: 10,
perHostConcurrency: 5,
rateLimit: {
requestsPerSecond: 100,
burstSize: 150,
},
circuitBreaker: {
failureThreshold: 5,
successThreshold: 3,
resetTimeout: 30000,
halfOpenMaxRequests: 3,
},
},
});
// Enqueue single request
const { id, position } = await queue.enqueue({
url: 'https://api.example.com/data',
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: { key: 'value' },
priority: 75, // 0-100, higher = more urgent
maxRetries: 5, // Override default
timeout: 10000, // Request timeout in ms
scheduledFor: new Date(Date.now() + 60000), // Delay execution
});
// Batch enqueue
const results = await queue.enqueueMany([
{ url: 'https://api.example.com/1', method: 'GET' },
{ url: 'https://api.example.com/2', method: 'GET' },
]);
// Get request status
const state = await queue.getStatus(id);
// { id, status, attempts, lastAttemptAt, error, response, ... }
// Cancel pending request
const cancelled = await queue.cancel(id);
// Get statistics
const stats = await queue.getStats();
// { pending, processing, completed, failed, dead, avgProcessingTime, successRate }
// Get failed requests
const deadRequests = await queue.getDeadLetterRequests(100);
// Retry a dead request
await queue.retryDeadRequest(requestId);
// Pause processing
await queue.pause();
// Resume processing
await queue.resume();
// Graceful shutdown
await queue.shutdown();
queue.onComplete((response) => {
// response: { requestId, status, headers, body, duration, attempt, completedAt }
});
queue.onError((requestId, error, willRetry) => {
if (willRetry) {
console.log(`Request ${requestId} failed, will retry`);
}
});
queue.onRetry((requestId, attempt, nextRetryAt) => {
console.log(`Retry #${attempt} scheduled for ${nextRetryAt}`);
});
queue.onDead((requestId, error) => {
console.error(`Request ${requestId} moved to dead letter queue`);
});
# Install dependencies
bun install
# Run type check
bun run typecheck
# Run tests
bun test
# Build for production
bun run build
MIT