# Application Layer: Operations, Stalls & DLQ
bunqueue application layer: PUSH/PULL/ACK operations, stall detection, dependency resolution, DLQ management, and job lifecycle flows.
Canonical: https://bunqueue.dev/architecture/application-layer/
---
architecture · application layer
Use cases in the application layer.
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.
The server selects one application manager at startup. `QueueManager` owns the
synchronous memory/SQLite path shown in the diagrams below;
`PostgresQueueManager` exposes the same handler-facing operations but commits
state through database transactions and refreshes a bounded compatibility
projection. See [Storage backends](/guide/databases/) for the multi-broker path.
## Module Structure
```
src/application/
├── queueManager.ts # Central orchestrator
├── postgresQueueManager.ts # PostgreSQL manager facade
├── postgres-queue-manager/ # Transactional operations and local projection
├── 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
```
`PostgresQueueManager` replaces the base delivery and lifecycle operations at
the same handler boundary. Admissions and terminal transitions commit job
state, ownership, results, and durable events together; pulls claim ordered
rows with `FOR UPDATE SKIP LOCKED`; ACK/FAIL validates database-clock leases and
broker-session tokens. A bounded local projection serves compatibility reads
and is repaired from the durable outbox plus polling. The complete transaction,
lease, and replay model is documented in the repository reference
`docs/features/postgres-multibroker.md`, the [architecture overview](/architecture/),
and the user-facing [storage guide](/guide/databases/).
## QueueManager Orchestration
QueueManagermemory / SQLite 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 flowmemory / SQLite push(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 configured SQLite buffered or durable; no-op in memory mode, 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. promote due secondary-index entries
↓
4. select through secondary scheduling indexes
TTL expired drop, try next
Ungrouped ready serve before grouped work
Grouped ready priority/FIFO lane, round-robin cursor, check group limits
No candidate track earliest delayed/rate-window wake-up
The authoritative priority heap and secondary ungrouped, delayed and per-group indexes change under the same synchronous shard lock. A blocked group requires no heap scan or temporary reinsertion.
↓
Job found move to processing shard
No job wait for notification, event-based with timeout, then retry loop
↓
5. Acquire queue and group capacity, 6. remove authoritative job, 7. advance group cursor, 8. create lock token if useLocks enabled, 9. update jobIndex to 'processing', 10. persist/broadcast active, 11. 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 active group capacity
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.
:::tip[Related]
- [Architecture Overview](/architecture/) - Full component map
- [Domain Layer](/architecture/domain-layer/) - Shards and the state machine these operations mutate
- [TCP Protocol](/architecture/tcp-protocol/) - How operations arrive over the wire
- [Persistence](/architecture/persistence/) - Where these operations are durably recorded
:::