Sovite.Core.QueueManager (sovite v0.6.0)

Copy Markdown View Source

Schedules queued messages for delivery, like Postfix's qmgr.

Lifecycle

On startup, messages left in active/ by a crash or kill -9 go back to incoming/ (Sovite.Queue.Spool.recover/1), and the retry times of deferred/ messages are read from their records.

A message is picked up from incoming/ (right after it is queued, and by a periodic scan) or from deferred/ (when its retry time comes) and moved to active/. Its pending recipients are routed (Sovite.Core.Router) and grouped into jobs: one per destination, with at most delivery.max_recipients recipients each. Jobs run in Sovite.Core.Delivery workers.

A message with a content filter (smtp.content_filter, see Sovite.Queue.Envelope) goes to the filter instead, for all its recipients at once, with its original sender and without REQUIRETLS. When the filter offers XFORWARD, it is told the original client, HELO name, and queue ID. Its job is then done: the filter sends the message back as a new one.

Each job's results are appended to the queue file and fsynced before anything else happens, so a crash never loses a delivery result: at worst, the recipients of the jobs in flight are delivered again, which SMTP allows. When all jobs of a message are done:

  1. Failed recipients are reported to the sender (Sovite.Core.Bounce), and counted against the user who sent the message, if any (Sovite.Core.Outbound). Delivered recipients and expanded aliases are reported too, when the sender asked for it (NOTIFY=SUCCESS); a recipient with NOTIFY=NEVER, or without FAILURE, is never reported as failed.
  2. If no recipient is pending, the message is deleted.
  3. If the message is older than queue.max_lifetime, its pending recipients fail and are reported, and it is deleted.
  4. Otherwise it is deferred: a delay warning is sent if due (queue.delay_warning) for the recipients whose NOTIFY does not rule it out, and the next attempt is scheduled with exponential backoff (Sovite.Queue.Backoff).

The DSN parameters of the message (RFC 3461) go with it to servers that support DSN.

Priorities and deadlines

Jobs for the same destination run in MT-PRIORITY order (RFC 6710), and of the destinations waiting, the one with the highest priority job goes first. A message on hold (FUTURERELEASE, RFC 4865) waits in deferred/ until its release time, and its lifetime starts then. A message with a DELIVERBY deadline (RFC 2852) is retried by then at the latest; when it passes, the pending recipients fail with 5.4.7 (R), or the sender gets a delay warning (N).

ETRN

flush/2 with a domain retries the deferred messages with recipients in that domain now, for ETRN (RFC 1985). The same domain is flushed at most once a minute.

Administration

hold/2, release/2, delete/2, requeue/2, and retry/2 act on one message, like Postfix's postsuper and postqueue -i. A message being delivered is held or deleted once its running deliveries end; the deliveries that have not started are dropped. reconfigure/2 takes new options from a reloaded configuration, for the messages picked up from then on. drain/2 stops starting deliveries, for a graceful shutdown.

Limits

  • delivery.max_deliveries workers run at once.
  • delivery.destination_concurrency of them per destination.
  • delivery.destination_rate_delay between two deliveries to the same destination.

A worker that finishes a job may get the next job for the same destination and send it over the same connection. With delivery.connection_cache, a worker with no next job keeps its connection open for delivery.connection_cache_time (2 seconds), in case one comes, as Postfix's connection cache does; it still counts towards the limits meanwhile, but makes room when another destination is waiting for a free slot.

Summary

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

delete(server, queue_id)

Deletes message queue_id from any queue, without notifying anyone.

drain(server, timeout)

Stops starting deliveries, and waits up to timeout milliseconds for the running ones to end. Messages left in active/ are picked up again at the next start. Returns {:error, :timeout} if deliveries were still running.

flush(server)

Retries every deferred message now, like postqueue -f.

flush(server, node)

Retries the deferred messages with pending recipients in a domain now ({:domain, domain}), or in a domain and its subdomains ({:subdomains, domain}), as ETRN asks. Returns how many there were, or {:error, :too_soon} when the same was asked less than a minute ago.

hold(server, queue_id)

Puts message queue_id on hold (in hold/): it is not delivered until release/2 or requeue/2.

notify(server, queue_id)

Tells the queue manager that message queue_id was queued in incoming/.

opts(config, repo \\ nil)

Queue manager options from the running configuration (repo is Sovite's database, for database tables). start_link/1 takes these, plus

reconfigure(server, opts)

Replaces the options with opts (from opts/2, after a config reload). The queue directory cannot change. Deliveries already planned keep the options they were planned with.

release(server, queue_id)

Releases a message from hold/ to deferred/. It is retried when its next attempt is due, at once if that has passed.

requeue(server, queue_id)

Queues message queue_id again and retries it now, from deferred/, hold/, or corrupt/ (where it goes back if it still fails verification). Its pending recipients are routed again, with the current aliases and routing tables.

retry(server, queue_id)

Retries message queue_id now, if it is deferred or waiting in incoming/. A message on hold for FUTURERELEASE stays on hold.

start_link(opts)
stats(server)

Counts of messages and deliveries, for status output and tests.

Functions

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

delete(server, queue_id)

@spec delete(GenServer.server(), String.t()) :: :ok | {:error, :not_found}

Deletes message queue_id from any queue, without notifying anyone.

drain(server, timeout)

@spec drain(GenServer.server(), timeout()) :: :ok | {:error, :timeout}

Stops starting deliveries, and waits up to timeout milliseconds for the running ones to end. Messages left in active/ are picked up again at the next start. Returns {:error, :timeout} if deliveries were still running.

flush(server)

@spec flush(GenServer.server()) :: :ok

Retries every deferred message now, like postqueue -f.

flush(server, node)

@spec flush(GenServer.server(), {:domain | :subdomains, String.t()}) ::
  {:ok, non_neg_integer()} | {:error, :too_soon}

Retries the deferred messages with pending recipients in a domain now ({:domain, domain}), or in a domain and its subdomains ({:subdomains, domain}), as ETRN asks. Returns how many there were, or {:error, :too_soon} when the same was asked less than a minute ago.

hold(server, queue_id)

@spec hold(GenServer.server(), String.t()) :: :ok | {:error, :not_found}

Puts message queue_id on hold (in hold/): it is not delivered until release/2 or requeue/2.

notify(server, queue_id)

@spec notify(GenServer.server(), String.t()) :: :ok

Tells the queue manager that message queue_id was queued in incoming/.

opts(config, repo \\ nil)

@spec opts(Sovite.Core.Config.t(), Sovite.Core.Repo.t() | nil) :: keyword()

Queue manager options from the running configuration (repo is Sovite's database, for database tables). start_link/1 takes these, plus:

  • :name - registered name, or nil for none. Defaults to this module.
  • :resolver - DNS resolver. Defaults to Sovite.DNS.default_resolver/0.
  • :port - SMTP port for MX and address-literal deliveries. Defaults to 25.
  • :client - extra Sovite.SMTP.Client.connect/3 options.
  • :scan_interval - milliseconds between scans of incoming/. Defaults to one minute.
  • :max_active - messages in memory at once. Defaults to 10000.
  • :outbound - Sovite.Core.Outbound options, if failures are counted.

reconfigure(server, opts)

@spec reconfigure(GenServer.server(), keyword()) :: :ok

Replaces the options with opts (from opts/2, after a config reload). The queue directory cannot change. Deliveries already planned keep the options they were planned with.

release(server, queue_id)

@spec release(GenServer.server(), String.t()) :: :ok | {:error, :not_found}

Releases a message from hold/ to deferred/. It is retried when its next attempt is due, at once if that has passed.

requeue(server, queue_id)

@spec requeue(GenServer.server(), String.t()) :: :ok | {:error, :not_found}

Queues message queue_id again and retries it now, from deferred/, hold/, or corrupt/ (where it goes back if it still fails verification). Its pending recipients are routed again, with the current aliases and routing tables.

retry(server, queue_id)

@spec retry(GenServer.server(), String.t()) :: :ok | {:error, :not_found}

Retries message queue_id now, if it is deferred or waiting in incoming/. A message on hold for FUTURERELEASE stays on hold.

start_link(opts)

@spec start_link(keyword()) :: GenServer.on_start()

stats(server)

@spec stats(GenServer.server()) :: %{
  active: non_neg_integer(),
  deferred: non_neg_integer(),
  deliveries: non_neg_integer(),
  ready: non_neg_integer(),
  draining: boolean()
}

Counts of messages and deliveries, for status output and tests.