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 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, and the user-facing storage guide.

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.

Related

• Architecture Overview - Full component map
• Domain Layer - Shards and the state machine these operations mutate
• TCP Protocol - How operations arrive over the wire
• Persistence - Where these operations are durably recorded