Skip to content
bunqueue bunqueue
Get started
Get started

bunqueue Domain Layer: Sharding, Priority Queues & States

architecture · domain layer

The domain layer, no I/O.

The domain layer contains the pure business logic of bunqueue. No external dependencies, just core algorithms and data structures.

Module Structure

Section titled “Module Structure”
src/domain/
├── types/           # Type definitions
└── queue/           # Core queue logic
    ├── shard.ts             # Shard container
    ├── priorityQueue.ts     # 4-ary indexed heap
    ├── dlqShard.ts          # Dead letter queue
    ├── uniqueKeyManager.ts  # Deduplication
    ├── limiterManager.ts    # Rate/concurrency
    ├── dependencyTracker.ts # Job dependencies
    ├── temporalManager.ts   # Temporal index + delayed jobs
    ├── waiterManager.ts     # Long-poll waiters
    └── shardCounters.ts     # O(1) per-queue stats

Sharding Architecture

Section titled “Sharding Architecture”

Jobs are distributed across N shards (auto-detected from CPU cores) for parallelism:

QueueManagerN independent shards, auto-detected
queueName
→
fnv1a()
→
& SHARD_MASK
→
idx
↓
Shard 0 queues, unique, dlq, limits
Shard 1 queues, unique, dlq, limits
Shard 2 queues, unique, dlq, limits
Shard N queues, unique, dlq, limits

Shard count is a power of 2, based on CPU cores, max 64.

Shard Composition

Section titled “Shard Composition”

Each shard is a composition of managers:

Shardcomposition of managers
queues Map<string, PriorityQueue>
UniqueKeyManager deduplication with TTL
DlqShard failed job storage
LimiterManager rate and concurrency control
DependencyTracker waitingDeps + dependencyIndex
TemporalManager delayed jobs, MinHeap
stats queued, delayed, dlq
activeGroups Map, FIFO groups
waiters Array, long poll support

Priority Queue Flow

Section titled “Priority Queue Flow”

4-ary indexed heap with lazy deletion:

PriorityQueue4-ary indexed heap with lazy deletion
PUSH
1. Generate generation number, 2. add to index Map<jobId, {job, generation}>, 3. push to heap {jobId, priority, runAt, generation}, 4. bubbleUp O(log₄ n)
POP
Loop: 1. peek heap top, 2. check index for matching generation, 3. if generation mismatch, stale entry: removeTop, continue, 4. if match: removeTop, delete from index, return job O(log₄ n) amortized
REMOVE, by jobId
1. Delete from index O(1), 2. heap entry becomes stale skipped on pop, 3. compact heap when stale ratio > 20%

Job State Machine

Section titled “Job State Machine”
Job state machine
WAITING re-entered when a retryable fail triggers retry
↓
DELAYED delay > 0, becomes ready when runAt is reached
ready delay = 0
↓
ACTIVE on retryable fail, back to WAITING
↓
COMPLETED success
DLQ fail at max retries, or timeout

Dependency Resolution Flow

Section titled “Dependency Resolution Flow”
Dependency resolutionJob B, dependsOn: [A]
push B, job with dependencies
1. Push B, check: is A completed?
↓
NO add B to waitingDeps, register B in dependencyIndex[A]
YES push B to active queue
when A completes
1. Add A.id to pendingDepChecks
↓
2. Event-driven flush scheduled on the next microtask, coalescing completions from the same tick; a 30s interval acts as safety fallback only
↓
3. For each completedId, get dependencyIndex[completedId] Set<jobIds>
↓
4. For each waiting job, check all deps in completedJobs, if YES move from waitingDeps to queue

Reverse Index:

Reverse indexdependencyIndex: Map<JobId, Set<JobId>>
A
→
{B, C} B and C wait for A
D
→
{E} E waits for D

DLQ (Dead Letter Queue) Flow

Section titled “DLQ (Dead Letter Queue) Flow”
Move to DLQjob fails with attempts >= maxAttempts
DlqEntry
job original job
reason explicit_fail, max_attempts_exceeded, timeout, stalled, ttl_expired, worker_lost, unknown
error error message
attempts full history: attempt, error, duration
enteredAt timestamp
nextRetryAt if autoRetry enabled
expiresAt 7 days default
↓
DLQ maintenance, every 60s
1. Auto-retry eligible entries nextRetryAt <= now && retryCount < maxAutoRetries
2. Purge expired entries expiresAt <= now
3. Enforce maxEntries per queue 10k default, FIFO eviction when full

Rate & Concurrency Limiting

Section titled “Rate & Concurrency Limiting”
Pull requestrate and concurrency limiting
1. check rate limit, token bucket
Tokens available consume 1, proceed
No tokens return null
↓
2. check concurrency limit
active < limit increment, proceed
At limit return null
↓
3. Pop from priority queue
token bucket
capacity N tokens
refillRate N tokens/sec
tryAcquire() 1. refill based on elapsed time, 2. if tokens >= 1 consume and return true, 3. else return false

FIFO Groups

Section titled “FIFO Groups”

Ensures only one job per group processes at a time:

FIFO groupsjob with groupId: "user-123"
PULL
1. Peek job at queue head, 2. check: is groupId in activeGroups?
↓
YES job stays at the head, this pull returns no job (preserves strict per-group order)
NO pop, add to activeGroups, return job
ACK / FAIL
1. Remove groupId from activeGroups, 2. next job in the same group can now be pulled