Job Queue

Table of Contents

Overview & Motivation

The job queue ensures no message is lost even if the gateway crashes mid-delivery. Every outbound message is persisted to disk before sending. Failed deliveries are automatically retried with increasing delays. Permanent errors (user blocked the bot, chat not found) are detected and moved aside to prevent infinite retry loops.

💡 Analogy — Registered Mail Service

Think of the queue like a postal service with registered mail. Every letter (message) is logged in a ledger (queue file) before leaving the post office. If delivery fails (address wrong), the postman tries again tomorrow, then next week, then next month (exponential backoff). If the recipient moved permanently (permanent error), the letter goes to the dead letter office (failed/ dir). After 5 attempts, it's flagged for human review.

Queue Storage

File-system backed queue with atomic writes via PID-stamped temp files

Recovery Engine

5 retries, 4-tier backoff [5s, 25s, 120s, 600s], 14 permanent error patterns

Retry Utilities

Generic retryAsync<T> with jitter, configurable min/max delay, abort support

Error Classification

Deep error graph traversal across 25+ transient patterns and fatal codes

Queue Storage — `delivery-queue-storage.ts`

The delivery queue stores each outbound message as a JSON file in ${stateDir}/delivery-queue/. The payload captures every field needed to retry delivery without re-resolving the recipient:

Payload Types

QueuedDeliveryPayload — the message envelope: { channel, to, accountId?, payloads[], threadId?, replyToId?, bestEffort?, gifPlayback?, forceDocument?, silent?, mirror?, gatewayClientScopes? }
QueuedDelivery — extends with queue metadata: { id, enqueuedAt, retryCount, lastAttemptAt?, lastError? }

File operations use atomic writes: the payload is first written to ${filePath}.${pid}.tmp, then renamed into place. This prevents partial reads if the process crashes mid-write.

Code Walk — Atomic Enqueue

Generate a unique ID and construct the QueuedDelivery object with retryCount: 0.

Serialize to JSON and write to a PID-stamped temporary file for crash safety.

Atomically rename the temp file to the final queue path. If rename fails, clean up the temp file.

Return the QueuedDelivery with its assigned ID for tracking.

interface QueuedDeliveryPayload {
  channel: string;
  to: string;
  accountId?: string;
  payloads: unknown[];
  threadId?: string;
  replyToId?: string;
  bestEffort?: boolean;
  gifPlayback?: boolean;
  forceDocument?: boolean;
  silent?: boolean;
  mirror?: boolean;
  gatewayClientScopes?: string[];
}

interface QueuedDelivery extends QueuedDeliveryPayload {
  id: string;
  enqueuedAt: number;
  retryCount: number;
  lastAttemptAt?: number;
  lastError?: string;
}

async function enqueueDelivery(
  payload: QueuedDeliveryPayload
): Promise<QueuedDelivery> {
  await ensureQueueDir();
  const id = crypto.randomUUID();
  const entry: QueuedDelivery = {
    ...payload,
    id,
    enqueuedAt: Date.now(),
    retryCount: 0,
  };
  const filePath = join(queueDir, `${id}.json`);
  const tmpPath = `${filePath}.${process.pid}.tmp`;

  await writeFile(tmpPath, JSON.stringify(entry, null, 2));
  await rename(tmpPath, filePath);
  return entry;
}

Recovery Engine — `delivery-queue-recovery.ts`

The recovery engine (recoverPendingDeliveries) runs on a periodic timer and processes all pending entries. Each entry is checked for eligibility: has it exceeded MAX_RETRIES (5)? Is it within its backoff window? Is the last error permanent?

Backoff Tiers

The backoff schedule uses a fixed tier system rather than pure exponential, giving operators predictable retry timing:

Tier 0 — 5 seconds (first retry)
Tier 1 — 25 seconds
Tier 2 — 120 seconds (2 minutes)
Tier 3 — 600 seconds (10 minutes, used for retries 3+)

Permanent Error Detection

isPermanentDeliveryError(error) matches against 14 regex patterns that indicate non-recoverable failures. These include:

Chat/conversation not found
User not found or deactivated
Bot was blocked by user
Forbidden — insufficient permissions
Message too long / invalid format

When a permanent error is detected, the entry is moved to ${stateDir}/delivery-queue/failed/ via moveToFailed() and no further retries occur.

✅ Recovery Summary

After each recovery cycle, a RecoverySummary is emitted: { recovered, failed, skippedMaxRetries, deferredBackoff }. This is logged and also exposed via the /api/queue/status health endpoint for monitoring dashboards.

Retry Utilities — `retry.ts`

The generic retryAsync<T>(fn, attemptsOrOptions) function wraps any async operation with configurable retry behavior. It is used throughout the codebase, not just by the delivery queue.

Defaults: 3 attempts, 300 ms min delay, 30 s max delay, 0 jitter
Exponential factor: 2^attempt multiplier on base delay
Custom hooks: shouldRetry?(error) for selective retry, retryAfterMs?(error) to honor server-sent Retry-After headers, onRetry?(error, attempt) for logging
Label: optional label string included in log messages

Backoff Policy — `backoff.ts`

The BackoffPolicy interface defines the four parameters that control delay growth. The computeBackoff(policy, attempt) function applies the formula: min(initialMs × factor^attempt + jitter, maxMs).

sleepWithAbort(ms, abortSignal?) wraps the delay in a promise that rejects immediately if the abort signal fires, enabling clean cancellation of in-flight backoff waits during shutdown.

Error Classification — `unhandled-rejections.ts`

The error classification module performs deep graph traversal on error objects. An error is "transient" if any node in its cause chain matches a known transient pattern. The module defines:

25+ transient network codes: ECONNRESET, ECONNREFUSED, ENOTFOUND, ETIMEDOUT, EPIPE, EHOSTUNREACH, EAI_AGAIN, plus all UND_* undici error codes
Transient SQLite codes: SQLITE_BUSY, SQLITE_LOCKED, SQLITE_CANTOPEN
Fatal codes: ERR_OUT_OF_MEMORY, ERR_WORKER_OUT_OF_MEMORY
Config codes: application-specific config validation failures

isTransientNetworkError(err) walks the full error cause chain (including .cause, .errors arrays for AggregateError, and .reason for promise rejections) and returns true if any node matches.

⚠️ Deep Traversal Warning

The error graph traversal has a depth limit of 10 and a visited-set guard to prevent infinite loops with circular error references. If an error chain exceeds 10 levels, it is treated as transient (fail-open) to avoid dropping messages.

Delivery Recovery Flowchart

The following diagram shows the complete recovery loop, from loading pending entries through eligibility checks, backoff computation, and the final delivery attempt with its three possible outcomes.

Click to expand

flowchart TD
    A["recoverPendingDeliveries()"] --> B["loadPendingDeliveries()"]
    B --> C{"For each entry"}
    C --> D{"retryCount >= MAX_RETRIES?"}
    D -->|yes| E["moveToFailed(entry)"]
    D -->|no| F{"isPermanentDeliveryError?"}
    F -->|yes| E
    F -->|no| G{"isEntryEligibleForRecoveryRetry?"}
    G -->|backoff not elapsed| H["deferredBackoff++"]
    G -->|eligible| I["attemptDelivery(entry)"]
    I -->|success| J["ackDelivery(entry.id)"]
    I -->|transient error| K["entry.retryCount++\nentry.lastError = err"]
    I -->|permanent error| E
    J --> L["recovered++"]
    K --> M["save updated entry"]
    E --> N["failed++"]
    C -->|done| O["emit RecoverySummary"]

Queue Item Lifecycle

Each queue item transitions through a well-defined set of states from initial enqueue to final resolution. Items can be delivered successfully, moved to the failed directory after exhausting retries or hitting a permanent error, or remain in the retrying loop within backoff windows.

Click to expand

stateDiagram-v2
    [*] --> Queued: enqueueDelivery()
    Queued --> Attempting: recovery cycle picks up
    Attempting --> Delivered: success
    Attempting --> Retrying: transient error
    Attempting --> PermanentFailure: permanent error detected
    Retrying --> BackoffWait: compute next delay
    BackoffWait --> Attempting: backoff elapsed
    Retrying --> MaxRetriesExceeded: retryCount >= 5
    MaxRetriesExceeded --> MovedToFailed: moveToFailed()
    PermanentFailure --> MovedToFailed: moveToFailed()
    Delivered --> [*]: ackDelivery()
    MovedToFailed --> [*]: archived in failed/ dir

Recovery Loop Implementation

Code Walk — Recovery with Backoff

Load all pending JSON files from the queue directory.

Filter out entries that are still within their backoff window.

Attempt delivery. On success, acknowledge (delete file). On failure, update retry count.

Emit a RecoverySummary with counts for monitoring.

const MAX_RETRIES = 5;
const BACKOFF_TIERS_MS = [5_000, 25_000, 120_000, 600_000];

interface RecoverySummary {
  recovered: number;
  failed: number;
  skippedMaxRetries: number;
  deferredBackoff: number;
}

function computeBackoffMs(retryCount: number, baseMs?: number): number {
  const tier = Math.min(retryCount, BACKOFF_TIERS_MS.length - 1);
  return baseMs ?? BACKOFF_TIERS_MS[tier];
}

async function recoverPendingDeliveries(
  opts?: { signal?: AbortSignal }
): Promise<RecoverySummary> {
  const entries = await loadPendingDeliveries();
  const summary: RecoverySummary = {
    recovered: 0, failed: 0,
    skippedMaxRetries: 0, deferredBackoff: 0,
  };

  for (const entry of entries) {
    if (opts?.signal?.aborted) break;

    if (entry.retryCount >= MAX_RETRIES) {
      await moveToFailed(entry);
      summary.skippedMaxRetries++;
      continue;
    }

    if (entry.lastError && isPermanentDeliveryError(entry.lastError)) {
      await moveToFailed(entry);
      summary.failed++;
      continue;
    }

    const backoff = computeBackoffMs(entry.retryCount);
    const elapsed = Date.now() - (entry.lastAttemptAt ?? entry.enqueuedAt);
    if (elapsed < backoff) {
      summary.deferredBackoff++;
      continue;
    }

    try {
      await attemptDelivery(entry);
      await ackDelivery(entry.id);
      summary.recovered++;
    } catch (err) {
      entry.retryCount++;
      entry.lastAttemptAt = Date.now();
      entry.lastError = String(err);
      await saveEntry(entry);
    }
  }
  return summary;
}

Operational Notes

ℹ️ Queue Health Monitoring

The /api/queue/status endpoint returns the current queue depth, number of items in backoff, and the last RecoverySummary. Set up alerts on skippedMaxRetries > 0 to catch delivery failures early.

ℹ️ Manual Retry

Items in failed/ can be moved back to the queue directory for manual re-processing. Reset the retryCount to 0 and remove lastError before moving the file. The next recovery cycle will pick it up automatically.

Overview & Motivation

Queue Storage

Recovery Engine

Retry Utilities

Error Classification

Queue Storage — delivery-queue-storage.ts

Payload Types

Code Walk — Atomic Enqueue

Recovery Engine — delivery-queue-recovery.ts

Backoff Tiers

Permanent Error Detection

Retry Utilities — retry.ts

Backoff Policy — backoff.ts

Error Classification — unhandled-rejections.ts

Delivery Recovery Flowchart

Queue Item Lifecycle

Recovery Loop Implementation

Code Walk — Recovery with Backoff

Operational Notes

Queue Storage — `delivery-queue-storage.ts`

Recovery Engine — `delivery-queue-recovery.ts`

Retry Utilities — `retry.ts`

Backoff Policy — `backoff.ts`

Error Classification — `unhandled-rejections.ts`