Application Layer: Operations, Stalls & DLQ

architecture · application layer

Use cases in the application layer.

Name: bunqueue
Author: egeominotti

The application layer orchestrates all queue operations, coordinating between the client layer and domain layer: PUSH, PULL and ACK flows, stall detection, dependency resolution, and background tasks.

Module Structure

src/application/
├── queueManager.ts        # Central orchestrator
├── operations/            # PUSH, PULL, ACK, Query
├── backgroundTasks.ts     # Task orchestration
├── cleanupTasks.ts        # Memory cleanup, orphan removal
├── clientTracking.ts      # Client connection tracking
├── contextFactory.ts      # Context creation helpers
├── dependencyProcessor.ts # Dependency resolution
├── dlqManager.ts          # Dead letter queue
├── eventsManager.ts       # Event pub/sub
├── jobLogsManager.ts      # Job logs management
├── latencyTracker.ts      # Operation latency percentiles
├── lockManager.ts         # Lock management
├── lockOperations.ts      # Lock acquire/release ops
├── metricsExporter.ts     # Prometheus metrics export
├── monitoringChecks.ts    # Periodic health checks
├── stallDetection.ts      # Stall detection
├── statsManager.ts        # Queue statistics
├── taskErrorTracking.ts   # Background task circuit breaker
├── throughputTracker.ts   # Push/pull/ack rate tracking
├── types.ts               # Shared type definitions
├── webhookManager.ts      # Webhook notifications
└── workerManager.ts       # Worker tracking

QueueManager Orchestration

QueueManagercentral orchestrator

state

shards[N] paired with shardLocks[N], N auto-detected

processingShards[N] paired with processingLocks[N]

jobIndex Map<id, location>

completedJobs BoundedSet, 50k

jobResults LRU, 10k

customIdMap LRU, 50k

operations

push() operations/push.ts

pull() operations/pull.ts

ack() operations/ack.ts

query operations/queryOperations.ts

managers

DLQManager

EventsManager

WorkerManager

WebhookManager

StatsManager

JobLogsManager

background tasks

cleanup

stall

dependency

dlq

cron

PUSH Operation Flow

PUSH flowpush(queue, input)

1. Generate UUIDv7 ID, 2. check customId idempotency customIdMap, if exists return existing job

↓

3. Acquire shard write lock shardIdx = fnv1a(queue) & SHARD_MASK

↓

4. check unique key deduplication

Key available register key, continue

Key exists, strategy replace remove old, insert new

Key exists, strategy extend reset TTL, return existing

Key exists, default return existing

↓

5. check dependencies

All satisfied push to queue

Not satisfied add to waitingDeps, register in dependencyIndex

↓

6. Update jobIndex, 7. persist to SQLite buffered or durable, 8. notify waiters wake long poll, 9. broadcast 'pushed' event

PULL Operation Flow

PULL flowpull(queue, timeoutMs), runs as a loop

1. Acquire shard write lock, 2. queue paused, return null, 3. rate limit exceeded, return null, 4. concurrency at limit, return null

↓

5. dequeue loop, inspect priority-ordered candidates

TTL expired drop, try next

Not ready, delayed temporarily park, track earliest runAt, inspect next candidate

FIFO group active temporarily park, inspect work from another eligible group

Valid job pop, continue

Parked jobs remain logically queued: counters, temporal indexes and jobIndex do not change. They are restored to the heap before the shard lock is released, so a blocked root cannot hide unrelated ready work.

↓

Job found move to processing shard

No job wait for notification, event-based with timeout, then retry loop

↓

6. Create lock token if useLocks enabled, 7. update jobIndex to 'processing', 8. mark active in SQLite, 9. broadcast 'pulled' event, 10. return job with token

ACK Operation Flow

ACK flowack(jobId, result, token)

1. Verify lock token if provided, on mismatch error: token invalid

↓

2. Remove from processing shard procIdx = fnv1a(jobId) & SHARD_MASK

↓

3. release shard resources

Release unique key

Release FIFO group

Release concurrency slot

↓

4. finalize

Store result in jobResults LRU

Store result in SQLite

Update jobIndex to 'completed'

Add to completedJobs signals deps

↓

5. Add to pendingDepChecks wake dependents, 6. broadcast 'completed' event, 7. trigger webhooks

Background Tasks

Background task scheduler

Cleanup every 10s

Stall check every 5s

Dependency event-driven, 30s safety fallback

DLQ maintenance every 60s

Lock expire every 5s

Cron precise setTimeout, 60s safety fallback

Processing timeouts use one next-deadline timer keyed by each active job's startedAt + timeout. Far-future timers are safely chunked at the runtime ceiling; failed timeout transitions are logged and retried.

Stall Detection (Two-Phase)

Stall checkevery 5s, two-phase

phase 1, process previous candidates

For each job in stalledCandidates: still in processing? get stall config

↓ if confirmed stalled

stallCount < maxStalls increment + retry

stallCount >= maxStalls move to DLQ

phase 2, mark new candidates

For each job in processingShards: no heartbeat for > stallInterval 30s, add to stalledCandidates checked next tick

Why two-phase? It prevents false positives from transient delays, like a GC pause or a network hiccup.

Dependency Resolution

Dependency processorevent-driven, microtask-coalesced

0. On job completion, a flush is scheduled on the next microtask completions from the same tick are coalesced; a 30s interval is only a safety fallback for missed events

↓

1. Collect completedIds from pendingDepChecks set of jobs that completed since last flush

↓

2. For each completedId, look up dependencyIndex[completedId] returns the Set of jobIds waiting for this job

↓

3. Group by shard for efficient locking

↓

4. For each waiting job, check ALL dependencies completed completedJobs.has(depId) for all deps

↓

5. If all satisfied: remove from waitingDeps, unregister from dependencyIndex, push to active queue

Cleanup Tasks

Cleanupevery 10s

1. Refresh delayed counts in each shard

2. Compact priority queues if stale ratio > 20%, rebuild heap

3. Clean orphaned processing entries jobs stuck > 30min with no heartbeat

4. Clean stale waiting dependencies waiting > 1 hour

5. Clean expired unique keys

6. Clean orphaned job index entries

7. Remove empty queues

Event Broadcasting

Events manager

Event occurs completed, failed, progress, stalled

↓

broadcast(event)

↓

Notify all subscribers Set-based, O(1) add

Trigger matching webhooks

Wake completion waiters

Event-based waiting, no polling: waitForJobCompletion(jobId, timeout) resolves when the 'completed' event for jobId arrives.