suaveplan/openspec/archive/2026-04-24-enterprise-queue/proposal.md

OpenSpec Change: enterprise-queue

• Change ID: enterprise-queue
• Package: @suaveplan/queue
• Category: universal
• Target path: universal/streams/queue
• Status: PROPOSED
• Date: 2026-04-24
• Reference spec: /mnt/s/web/suaveplan/enterprise/queue.md

---

Summary

Deliver @suaveplan/queue on the genesis branch from scratch at enterprise quality. The package provides queue management primitives for task/job processing: a typed Queue interface, in-memory FIFO/priority implementations, QueueWorker for concurrent job processing with bounded concurrency, retry/dead-letter policies, AbortSignal cancellation, event emission for job lifecycle, OTel instrumentation, and Zod-validated configuration. This is the single-process queue substrate; @suaveplan/queue-distribution builds the distributed layer on top.

---

Current-state audit (genesis)

The package does not exist on genesis yet — fresh build. No prior source, no test results, no coverage data.

---

Available workspace deps on genesis

Package	Role
@suaveplan/error (workspace:^)	Base class for QueueError; defineCodes for QUEUE.* codes
@suaveplan/types (workspace:^)	LoggerLike, Clock, EventBusLike, IdGenerator injection interfaces
@suaveplan/events (workspace:^)	Job lifecycle event emission (job:enqueued, job:started, job:completed, job:failed, job:dead)
zod (catalog:forms)	Zod schemas for all public option and job types
@opentelemetry/api (catalog:observability)	Spans, counters, histograms on enqueue, process, and retry operations
tslib (catalog:build)	Runtime helpers

Dev deps: vitest, @vitest/coverage-istanbul, fast-check, mitata (all catalog:testing).

---

Public API surface

Queue interface

interface Queue<T> {
  readonly name: string
  readonly size: number
  enqueue(job: Job<T>): Promise<string>          // returns jobId
  dequeue(): Promise<Job<T> | undefined>
  peek(): Promise<Job<T> | undefined>
  get(jobId: string): Promise<Job<T> | undefined>
  ack(jobId: string): Promise<void>
  nack(jobId: string, reason?: string): Promise<void>
  remove(jobId: string): Promise<boolean>
  clear(): Promise<void>
  drain(): Promise<void>
}

interface Job<T> {
  id: string
  payload: T
  priority?: number           // higher = processed first; default: 0
  scheduledAt?: Date
  maxRetries?: number
  attemptCount: number
  createdAt: Date
  state: JobState
}

type JobState = "pending" | "processing" | "completed" | "failed" | "dead"

Queue implementations

// In-memory FIFO queue
function createMemoryQueue<T>(options?: MemoryQueueOptions<T>): Queue<T>

// In-memory priority queue (max-heap on job.priority)
function createPriorityQueue<T>(options?: PriorityQueueOptions<T>): Queue<T>

// Delayed job queue (jobs are not dequeued before scheduledAt)
function createDelayedQueue<T>(options?: DelayedQueueOptions<T>): Queue<T>

interface MemoryQueueOptions<T> {
  name?: string
  maxSize?: number
  clock?: Clock
  idGenerator?: IdGenerator
  bus?: EventBusLike
  logger?: LoggerLike
}

type PriorityQueueOptions<T> = MemoryQueueOptions<T>
type DelayedQueueOptions<T> = MemoryQueueOptions<T> & { pollIntervalMs?: number }

QueueWorker

function createQueueWorker<T>(options: QueueWorkerOptions<T>): QueueWorker<T>

interface QueueWorker<T> {
  start(signal?: AbortSignal): Promise<void>
  stop(): Promise<void>
  readonly stats: WorkerStats
}

interface QueueWorkerOptions<T> {
  queue: Queue<T>
  handler: JobHandler<T>
  concurrency?: number            // default: 1
  retryPolicy?: RetryPolicy
  deadLetterQueue?: Queue<T>
  logger?: LoggerLike
  clock?: Clock
}

type JobHandler<T> = (job: Job<T>, signal: AbortSignal) => Promise<void>

interface RetryPolicy {
  maxRetries: number
  backoff: "constant" | "linear" | "exponential"
  initialDelayMs: number
  maxDelayMs?: number
}

interface WorkerStats {
  processed: number
  succeeded: number
  failed: number
  dead: number
  retried: number
  avgProcessingMs: number
}

Job lifecycle events (when bus provided)

"queue:job:enqueued"    // { queueName, jobId, priority }
"queue:job:started"     // { queueName, jobId, attempt }
"queue:job:completed"   // { queueName, jobId, durationMs }
"queue:job:failed"      // { queueName, jobId, error, attempt, willRetry }
"queue:job:dead"        // { queueName, jobId, reason }

Error handling

class QueueError extends SuaveplanError<QueueErrorCode>

type QueueErrorCode =
  | "QUEUE_FULL"
  | "QUEUE_JOB_NOT_FOUND"
  | "QUEUE_WORKER_ABORTED"
  | "QUEUE_HANDLER_FAILED"
  | "QUEUE_INVALID_STATE"
  | "QUEUE_DEAD_LETTER_FAILED"

function registerQueueErrorCodes(): void

Zod schemas

const JobSchema: ZodSchema
const RetryPolicySchema: ZodSchema
const MemoryQueueOptionsSchema: ZodSchema
const QueueWorkerOptionsSchema: ZodSchema

---

New additions for genesis

1. Delayed queue with injectable clock

createDelayedQueue uses the injectable Clock for scheduledAt comparisons, enabling deterministic tests without real timers.

2. Dead letter queue

When a job exceeds maxRetries, it is moved to the deadLetterQueue (if configured) with the failure reason and full retry history attached. The dead letter queue is itself a Queue<T>, allowing inspection and manual replay.

3. OTel instrumentation

• queue.enqueue — span queue.enqueue, attributes: queue.name, queue.priority
• worker.process — span queue.worker.process, attributes: queue.name, queue.job_id, queue.attempt
• Counters: queue.jobs_enqueued_total, queue.jobs_completed_total, queue.jobs_failed_total, queue.jobs_dead_total
• Histogram: queue.processing_duration_ms (attribute: queue.name)
• Gauge: queue.depth (attribute: queue.name)

4. Property-based tests (fast-check)

• FIFO invariant: dequeue order matches enqueue order for equal-priority jobs
• Priority invariant: highest-priority job is always dequeued first
• Retry count bound: job.attemptCount never exceeds maxRetries + 1 before moving to dead letter
• Worker concurrency: at most concurrency jobs are in processing state at any time

5. Benchmark suite (src/__benchmarks__/)

• Enqueue + dequeue 100k items (memory queue)
• Worker with concurrency=10 processing 1k instant jobs

---

File layout

universal/streams/queue/
  package.json
  tsconfig.json
  vitest.config.ts
  src/
    index.ts
    error-codes.ts
    constants/
      index.ts
    types/
      common.ts
      common.md
    queues/
      memory-queue/
        memory-queue.ts
        memory-queue.md
        memory-queue.types.ts
        memory-queue.test.ts
        index.ts
      priority-queue/
        priority-queue.ts
        priority-queue.md
        priority-queue.types.ts
        priority-queue.test.ts
        index.ts
      delayed-queue/
        delayed-queue.ts
        delayed-queue.md
        delayed-queue.types.ts
        delayed-queue.test.ts
        index.ts
    worker/
      queue-worker/
        queue-worker.ts
        queue-worker.md
        queue-worker.types.ts
        queue-worker.test.ts
        index.ts
    utils/
      retry-policy.ts
      retry-policy.md
      retry-policy.test.ts
    __benchmarks__/
      queue.bench.ts
  tests/
    integration.test.ts
    property.test.ts

---

Impact

Downstream packages: @suaveplan/queue-distribution extends this package with distributed worker pools. Any background-job, batch-processing, or task-scheduling feature in the monorepo uses this as the foundation.

Workspace dep prerequisites: @suaveplan/error, @suaveplan/types, @suaveplan/events must be published to Verdaccio.

Breaking changes: none (new package).