@azlib/queue

Provider-agnostic background task queue implementing retry policies, dead-letter routing, and dashboard query interfaces.

Capabilities

• Asynchronous task enqueue and claim workflows
• Retry policies supporting fixed, exponential, and jittered backoffs
• Dead-letter routing (DLQ) with remediation operator actions
• In-process or database-backed task state durability (SQL persistence)
• Pluggable logging for enqueue, claim, retry, ack, and dead-letter transitions
• Provider switching (e.g. Memory, AWS SQS) under a stable, uniform API

AI Agent Quick Reference

Core Exports

| Export | Type | Description | | --- | --- | --- | | createQueueService(options: QueueServiceOptions): QueueService | Function | Instantiates the core queue manager instance. | | createProviderFromEnv(env: Record<string, string>): QueueProvider | Function | Factory generating a queue provider (Memory vs AWS SQS) from env config. | | createRetryPolicyEngine(policy: RetryPolicy) | Function | Low-level calculator of next-run timestamps and backoffs. | | createConsoleQueueLogger() | Function | Simple logger output formatter for debugging task runs. |

Core Types & Signatures

• QueueService:
  • enqueue(input: EnqueueInput): Promise<EnqueueResult>
  • consume(queueName: string, worker: (task: ClaimedMessage, ctx: WorkerContext) => Promise<void>): Promise<() => void> (returns shutdown unsubscriber)
  • getDashboardQueries(): Returns service interfaces for querying status stats, attempts, and DLQ remediation.
• QueueServiceOptions:
  • provider: QueueProvider
  • defaultQueue?: string
  • retryPolicy?: RetryPolicy
  • persistence?: PersistenceConfig (durable SQL persistence from @azlib/persistence)
  • logger?: QueueLogger
• EnqueueInput:
  • idempotencyKey?: string
  • payloadRef: any (JSON-serializable data)
  • queueName?: string
• RetryPolicy:
  • maxAttempts: number
  • backoffType: "fixed" | "exponential" | "exponential-jitter"
  • baseDelayMs: number
  • maxDelayMs: number

Basic Usage (Memory Provider)

import { createQueueService } from "@azlib/queue";
import { createProviderFromEnv } from "@azlib/queue/providers";

const queue = createQueueService({
  provider: createProviderFromEnv({ QUEUE_PROVIDER: "memory" }),
  defaultQueue: "heavy-jobs",
  retryPolicy: {
    maxAttempts: 3,
    backoffType: "exponential-jitter",
    baseDelayMs: 1000,
    maxDelayMs: 30_000,
  },
});

// Enqueue a job
const result = await queue.enqueue({
  idempotencyKey: "invoice_123",
  payloadRef: { invoiceId: 123 },
});

Worker / Consumer Flow

// Start worker
const unsubscribe = await queue.consume("heavy-jobs", async (task, ctx) => {
  try {
    await processInvoice(task.payloadRef.invoiceId);
    await ctx.ack(); // success
  } catch (error) {
    await ctx.fail(error); // triggers retry policy or moves to DLQ
  }
});

// Call unsubscribe() on process shutdown

Behavioral Gotchas

• Default Logger Binding: If no logger is passed to the options, the service will fall back to using @azlib/logger automatically.
• Idempotency Guard: If SQL persistence is not configured, the default memory-based idempotency guard is transient and resets on application restart.
• DLQ State Transitions: Tasks enter the Dead-Letter Queue (DLQ) status immediately once maxAttempts are exhausted. They must be resolved manually or requeued using the Dashboard operator actions.