meowmail/smtp/queue

MeowMail — Persistent outbound message queue with retry and backoff.

Messages are spooled to disk as .eml files with envelope metadata. A background queue runner processes deferred messages with exponential backoff. The queue survives server restarts.

Types

Queue = ref object
  dir*: string               ## Spool directory path
  store*: StorageDriver      ## Flysystem-backed spool storage (atomic writes)
  entries*: seq[QueueEntry]
  lock*: Lock
  maxRetries*: int
  baseDelay*: int            ## Initial retry delay in seconds
  maxDelay*: int             ## Maximum retry delay in seconds
QueueEntry = object
  id*: string                ## Filename without extension
  path*: string              ## Spool file path, relative to the queue root
  mailFrom*: string
  rcptTo*: seq[string]
  heloName*: string
  dataOffset*: int           ## Byte offset where message data starts
  status*: QueueStatus
  retryCount*: int
  nextRetry*: times.Time     ## When to attempt next delivery
  lastAttempt*: times.Time
  lastError*: string
  created*: times.Time
QueueStatus = enum
  qsDeferred,               ## Waiting for retry
  qsActive,                 ## Currently being delivered
  qsDelivered,              ## Successfully delivered
  qsBounced,                ## Permanent failure, bounce generated
  qsFailed                   ## Unrecoverable error

Procs

proc attemptEntry(queue: Queue; entryId: string; deliverProc: proc (
    req: DeliveryRequest): DeliveryOutcome {....gcsafe.}; localStore: MaildirStore): bool {.
    ...raises: [Exception, Exception, Exception, Exception], tags: [RootEffect,
    TimeEffect, ReadDirEffect, WriteDirEffect, WriteIOEffect, ReadIOEffect],
    forbids: [].}
Attempt one delivery of the queue entry with the given id. Handles partial acceptance (bounce permanent rejections, requeue temporary ones), retry scheduling, and bounce generation on exhaustion/permanent failure. Returns false if the entry no longer exists.
proc deleteEntryFiles(queue: Queue; entry: QueueEntry) {....raises: [Exception],
    tags: [RootEffect], forbids: [].}
Delete the spool file and its metadata file. Missing files are ignored.
proc enqueue(queue: Queue; req: DeliveryRequest): string {....raises: [Exception],
    tags: [TimeEffect, RootEffect], forbids: [].}
Add a message to the queue. Returns the queue entry ID.
proc flushQueue(queue: Queue; deliverProc: proc (req: DeliveryRequest): DeliveryOutcome {.
    ...gcsafe.}; localStore: MaildirStore = nil): int {....raises: [Exception], tags: [
    TimeEffect, RootEffect, ReadDirEffect, WriteDirEffect, WriteIOEffect,
    ReadIOEffect], forbids: [].}
Force immediate delivery of every deferred entry, overriding the backoff schedule (nextRetry is reset to now). Returns the number of entries attempted. Used by meowmail queue flush and the admin API.
proc getEntry(queue: Queue; entryId: string): QueueEntry {....raises: [], tags: [],
    forbids: [].}
Look up a queue entry by ID.
proc load(queue: Queue) {....raises: [Exception, Exception],
                          tags: [RootEffect, TimeEffect], forbids: [].}
Load all spool files from the queue directory into memory.
proc markDelivered(queue: Queue; entryId: string) {....raises: [Exception],
    tags: [TimeEffect, RootEffect], forbids: [].}
Mark a queue entry as successfully delivered.
proc markFailed(queue: Queue; entryId: string; errorMsg: string) {.
    ...raises: [Exception], tags: [TimeEffect, RootEffect], forbids: [].}
Mark a queue entry as failed. If retries remain, schedule next attempt. Otherwise mark as bounced.
proc newQueue(dir: string; maxRetries = 20; baseDelay = 300; maxDelay = 86400): Queue {.
    ...raises: [ValueError, OSError, IOError],
    tags: [ReadEnvEffect, ReadIOEffect, ReadDirEffect, WriteDirEffect],
    forbids: [].}
Create a new queue backed by the given directory. Spool I/O goes through a flysystem LocalDriver rooted at dir (atomic writes, traversal-safe paths). The root directory is created if missing.
proc pending(queue: Queue): seq[QueueEntry] {....raises: [], tags: [TimeEffect],
    forbids: [].}
Return entries that are due for retry.
proc removeEntry(queue: Queue; entryId: string) {....raises: [], tags: [],
    forbids: [].}
Remove a queue entry from the in-memory list and delete its files.
proc routeBounce(localStore: MaildirStore; q: Queue;
                 envelopeFrom, rcpt, helo: string; status: DsnStatus;
                 diag: string = ""): bool {....raises: [Exception], tags: [
    TimeEffect, ReadDirEffect, WriteDirEffect, WriteIOEffect, ReadIOEffect,
    RootEffect], forbids: [].}
Generate a DSN bounce for one recipient and route it to the original sender: local senders receive it in their Maildir, remote senders get it enqueued with a null return-path (loop-safe: bounces never bounce).
proc runQueue(queue: Queue; deliverProc: proc (req: DeliveryRequest): DeliveryOutcome {.
    ...gcsafe.}; intervalMs: int = 30000; localStore: MaildirStore = nil) {.
    ...raises: [Exception], tags: [TimeEffect, RootEffect, ReadDirEffect,
                                WriteDirEffect, WriteIOEffect, ReadIOEffect],
    forbids: [].}
Background queue runner. Processes pending entries at the given interval. Temporary failures are retried with exponential backoff; permanent failures and exhausted retries generate DSN bounces to the sender. This proc runs forever (should be called from a thread).
proc saveMeta(queue: Queue; entry: QueueEntry) {....raises: [Exception],
    tags: [RootEffect], forbids: [].}
Persist queue entry metadata to a .meta file (atomic write).
proc stats(queue: Queue): tuple[total, deferred, active, delivered, bounced: int] {.
    ...raises: [], tags: [], forbids: [].}
Return queue statistics.