Alert Mail Dispatcher: Delivered, or Still on Disk

The gap this fills

My monitoring scripts alert by email, and they all quietly assume the same thing: that handing a message to the local MTA means it will eventually arrive. On a real server with postfix configured, that assumption is fine, and the MTA's queue is doing exactly the job described here, better than I ever will.

But not every box has an MTA. Containers, minimal VMs, appliances, and half the hosts in a lab have no mail agent at all. The usual workaround is opening a direct SMTP connection to the relay at send time, and that has a failure mode which should bother anyone who has read the validator's design notes: a direct connection with no queue behind it fails precisely when the network is degraded, which is precisely when the alert matters. The alert about the outage is lost to the outage. For these hosts that is the primary case rather than a corner one.

So the design center of this tool is not sending email. Sending email is a solved problem living in Python's standard library. The design center is what happens when sending fails: the message is written to a disk spool, atomically and with an fsync behind it, and a flush run retries it later. Deliver now, or persist and deliver later. The outcome the tool exists to prevent is silent loss, which means delivery is at-least-once rather than exactly-once. It will duplicate an alert before it will drop one.

What got rejected on the way here

The first sketch of this idea involved formatting SMTP commands onto a raw socket by hand, and it deserves a paragraph because rejecting it shaped everything else. Hand-rolled SMTP is a correctness trap with no compensating benefit: greeting parsing, multiline reply codes, STARTTLS negotiation, dot-stuffing, header folding, and relay-specific behaviour that smtplib and email.message already handle. "Lightweight" cannot mean reimplementing a protocol worse than the standard library that is already installed. The dependency-free constraint is real and this tool honors it. A dependency is something you install; the standard library ships with the interpreter. The lightness that matters here is operational: one file, one config, no daemon, no pip.

The second thing rejected: threshold logic. The original sketch had this tool watching logs and deciding when failure counts crossed a line. That decision belongs to the callers, and it already lives there; the validator owns its consecutive-failure counters, the drift reporter owns its diff logic. A dispatcher that also monitors is two half-tools sharing a config file. This one has a single responsibility with a crisp contract: you decide whether to alert, I deliver or durably spool, and my exit code tells you which you got.

Design decisions

Spool writes are atomic and durable: temp file, fsync, rename, fsync the directory. A flush run that wakes up mid-write must never see a half-written message. Rename within a filesystem is atomic on POSIX, so a record is either fully absent or fully present. The fsync pair pushes the file data and the directory entry through the filesystem's durability path, which extends the same property across crash and power loss as far as the storage stack honours it. Every path that writes a record goes through one function, including the attempt-counter update during flush.

An unusable record is quarantined rather than fatal. A record that fails to parse, or that parses into the wrong shape, moves to the dead-letter directory and the run continues. This reads as defensive hygiene and is the most important correction in this revision. The read originally sat outside the try block, so one truncated record raised through the loop, ended the flush process, and stranded every message behind it in sort order. The process exited 1, which is the documented code for "spool non-empty," so a poisoned spool and a busy spool looked identical from outside. On a host with no mail agent, which is the entire target environment, cron has nowhere to deliver the traceback either. Key validation happens at the same boundary rather than one frame later inside the delivery call, so both kinds of unusable record are caught in one place.

Spooled messages carry their attempt count, and exhausted ones move to a dead-letter directory. Retry forever is a lie; some messages reference relays that were decommissioned or recipients that never existed, and an immortal spool fills with them. After max_delivery_attempts, a message moves to dead_letters, still human-readable JSON, where the status command counts it loudly. Nothing is ever deleted by the tool; disposal of dead letters is a human decision.

Flush sorts failures into three classes, not two. Connection-level failures mean the relay is unreachable, which is one condition affecting every queued message, so the run stops rather than burning an attempt on all forty of them. Message-level rejections are specific to one message, so that message is charged an attempt and the run continues. The third class took a functional test to find: relay-session failures. SMTPAuthenticationError, SMTPHeloError and SMTPNotSupportedError all subclass SMTPException, so a stale credential or a relay declining STARTTLS was landing in the message-rejection branch and charging every message in the queue. At the shipped ceiling of twenty attempts on a five-minute timer, one wrong password retired an entire spool to dead letters in about ninety-five minutes, and the flush run that finished the job exited reporting a clean spool. Those three are now named in CONNECTION_LEVEL_ERRORS alongside the connection-level exceptions, and inject_defect.py can put the narrow tuple back to prove the tests still catch it.

STARTTLS by default, with a config switch rather than a code path fork. Internal relays vary. The config decides; the code has one sending function.

Same locking discipline as the bash tools. One flush at a time, enforced with fcntl, for the same reason the reliability library enforces it with flock: overlapping runs mutating shared state produce no symptom at all until the symptom is a counter that has been silently overwritten.

The code

#!/usr/bin/env python3
"""
smtp_alert_dispatcher.py
Deliver alert email now, or spool to disk and deliver later. Never neither.

Commands:
    send   --subject S [--body B | body on stdin]
    flush
    status

Exit codes: 0 delivered (or flush/status clean), 1 spooled for retry or
spool non-empty, 2 configuration error or dead letters present.
"""

import argparse
import email.utils
import fcntl
import json
import os
import smtplib
import ssl
import sys
import time
import uuid
from email.message import EmailMessage
from pathlib import Path

CONFIGURATION_PATH = Path("/etc/smtp_alert_dispatcher/config.json")

DEFAULT_CONFIGURATION = {
    "relay_host": "mailrelay.example.net",
    "relay_port": 587,
    "use_starttls": True,
    "smtp_username": "",
    "smtp_password_file": "",
    "sender_address": "alerts@example.net",
    "recipient_addresses": ["storage-alerts@example.net"],
    "connect_timeout_seconds": 10,
    "max_delivery_attempts": 20,
    "spool_directory": "/var/spool/smtp_alert_dispatcher/pending",
    "dead_letter_directory": "/var/spool/smtp_alert_dispatcher/dead_letters",
    "lock_file": "/var/lock/smtp_alert_dispatcher.lock",
}

# Failures of the relay session rather than of one message. Authentication,
# HELO and STARTTLS refusals all subclass SMTPException, so unless they are
# named here they fall into the message-rejection branch below and one wrong
# credential charges an attempt against every message in the spool.
CONNECTION_LEVEL_ERRORS = (
    smtplib.SMTPServerDisconnected,
    smtplib.SMTPConnectError,
    smtplib.SMTPHeloError,
    smtplib.SMTPAuthenticationError,
    smtplib.SMTPNotSupportedError,
)

REQUIRED_RECORD_KEYS = ("subject", "body", "attempt_count")


def load_configuration():
    configuration = dict(DEFAULT_CONFIGURATION)
    if CONFIGURATION_PATH.exists():
        with CONFIGURATION_PATH.open() as configuration_handle:
            configuration.update(json.load(configuration_handle))
    for directory_key in ("spool_directory", "dead_letter_directory"):
        Path(configuration[directory_key]).mkdir(parents=True, exist_ok=True)
    # The lock file's directory is bootstrapped too. On a minimal container
    # /var/lock may not exist, and flush would then raise before it could
    # acquire the lock, which is a failure with no spool and no signal.
    Path(configuration["lock_file"]).parent.mkdir(parents=True, exist_ok=True)
    return configuration


def fsync_directory(directory_path):
    """Force a directory entry to disk so a rename survives power loss.

    rename() is atomic in the POSIX namespace, but atomicity is not
    durability. Without this the kernel may have the new name only in the
    page cache, and a power loss can leave the entry missing or the file
    zero length after reboot.
    """
    directory_descriptor = os.open(str(directory_path), os.O_RDONLY)
    try:
        os.fsync(directory_descriptor)
    finally:
        os.close(directory_descriptor)


def write_record_atomically(final_path, message_record):
    """Write a spool record so a reader never sees a partial file.

    Write to a temporary name the *.json glob does not match, fsync the
    file contents, rename over the target, then fsync the directory.
    os.replace rather than Path.rename because this is also the update
    path, and the destination already exists when an attempt is charged.
    """
    temporary_path = final_path.with_suffix(".tmp")
    with open(temporary_path, "w") as record_handle:
        record_handle.write(json.dumps(message_record, indent=2))
        record_handle.flush()
        os.fsync(record_handle.fileno())
    os.replace(temporary_path, final_path)
    fsync_directory(final_path.parent)
    return final_path


def load_spool_record(pending_path):
    """Read one spool record, raising ValueError if it cannot be delivered.

    Checking the keys here rather than letting deliver_message raise KeyError
    keeps every unusable-record failure in one place. A record that parses
    and a record that parses into the wrong shape fail the run identically.
    """
    try:
        message_record = json.loads(pending_path.read_text())
    except json.JSONDecodeError as parse_error:
        raise ValueError(f"unparseable: {parse_error}") from parse_error
    missing_keys = [key for key in REQUIRED_RECORD_KEYS if key not in message_record]
    if missing_keys:
        raise ValueError(f"missing keys: {', '.join(missing_keys)}")
    return message_record


def build_message_record(subject_text, body_text, attempt_count=1):
    """Create a record carrying a stable identity across retries.

    message_id and date are generated once, here, and reused on every
    later delivery attempt. Delivery is at-least-once: a message can be
    accepted by the relay and still be retried if the connection drops
    before the spool file is unlinked. Without a stable Message-ID each
    retry arrives as a distinct message that no client can thread or
    collapse, so a retry storm becomes inbox noise during exactly the
    incident the alert exists to report.
    """
    return {
        "created_at": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
        "message_id": email.utils.make_msgid(),
        "date": email.utils.formatdate(localtime=True),
        "subject": subject_text,
        "body": body_text,
        "attempt_count": attempt_count,
    }


def read_smtp_password(configuration):
    password_file = configuration["smtp_password_file"]
    if not password_file:
        return ""
    return Path(password_file).read_text().strip()


def deliver_message(configuration, message_record):
    """Attempt one SMTP delivery. Raises on any failure; caller spools.

    Message-ID and Date come from the record rather than being generated
    per attempt, so a redelivery is recognisably the same message. Records
    spooled by an earlier version carry neither, so both fall back to a
    freshly generated value rather than raising.
    """
    outgoing_message = EmailMessage()
    outgoing_message["From"] = configuration["sender_address"]
    outgoing_message["To"] = ", ".join(configuration["recipient_addresses"])
    outgoing_message["Subject"] = message_record["subject"]
    outgoing_message["Message-ID"] = message_record.get(
        "message_id") or email.utils.make_msgid()
    outgoing_message["Date"] = message_record.get(
        "date") or email.utils.formatdate(localtime=True)
    outgoing_message.set_content(message_record["body"])

    with smtplib.SMTP(
        configuration["relay_host"],
        configuration["relay_port"],
        timeout=configuration["connect_timeout_seconds"],
    ) as smtp_connection:
        if configuration["use_starttls"]:
            smtp_connection.starttls(context=ssl.create_default_context())
        if configuration["smtp_username"]:
            smtp_connection.login(
                configuration["smtp_username"],
                read_smtp_password(configuration),
            )
        smtp_connection.send_message(outgoing_message)


def spool_message(configuration, message_record):
    """Persist a prepared record atomically."""
    spool_directory = Path(configuration["spool_directory"])
    final_path = spool_directory / f"{int(time.time())}_{uuid.uuid4().hex}.json"
    return write_record_atomically(final_path, message_record)


def command_send(configuration, arguments):
    if arguments.body is not None:
        body_text = arguments.body
    elif not sys.stdin.isatty():
        body_text = sys.stdin.read()
    else:
        # Reading a tty here waits forever and looks like a hang rather
        # than a usage error, which is the wrong thing to do to someone
        # testing the command by hand during an incident.
        print("no body: pass --body or pipe the body on stdin",
              file=sys.stderr)
        return 2

    message_record = build_message_record(arguments.subject, body_text)
    try:
        deliver_message(configuration, message_record)
        print("delivered")
        return 0
    except (smtplib.SMTPException, OSError) as delivery_error:
        spooled_path = spool_message(configuration, message_record)
        print(f"spooled: {delivery_error} -> {spooled_path.name}", file=sys.stderr)
        return 1


def command_flush(configuration):
    # Append rather than write: "w" truncates the lock file on every run,
    # which is a pointless mutation of a file that exists only to be held.
    lock_handle = open(configuration["lock_file"], "a")
    try:
        fcntl.flock(lock_handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except BlockingIOError:
        lock_handle.close()
        print("another flush is running", file=sys.stderr)
        return 0

    pending_paths = sorted(Path(configuration["spool_directory"]).glob("*.json"))
    dead_letter_directory = Path(configuration["dead_letter_directory"])
    for pending_path in pending_paths:
        try:
            message_record = load_spool_record(pending_path)
        except (ValueError, OSError) as record_error:
            # This read sat outside the try below until this revision. One
            # unusable record then raised through the loop, killed the run,
            # and stranded every message behind it in sort order, while the
            # process exited 1, which is the code for an ordinary non-empty
            # spool. A poisoned spool and a busy spool looked identical.
            os.replace(pending_path, dead_letter_directory / pending_path.name)
            fsync_directory(dead_letter_directory)
            fsync_directory(pending_path.parent)
            print(f"unusable spool record, dead lettered: "
                  f"{pending_path.name}: {record_error}", file=sys.stderr)
            continue
        try:
            deliver_message(configuration, message_record)
            pending_path.unlink()
            print(f"delivered from spool: {pending_path.name}")
        except CONNECTION_LEVEL_ERRORS as session_error:
            # Failure of the relay session, not of this message: one
            # condition affecting every queued message, so stop the run
            # rather than charging an attempt against all of them.
            print(f"relay session failed, stopping flush: {session_error}",
                  file=sys.stderr)
            break
        except smtplib.SMTPException:
            # Message-level rejection: charge this message, continue the run.
            # Ordering note: SMTPException subclasses OSError since Python
            # 3.4, so the bare OSError handler below MUST come after this
            # one or this branch is unreachable and nothing ever dead
            # letters. Found by the functional gate, not by review.
            message_record["attempt_count"] += 1
            # Same atomic path as the initial spool. Writing in place here
            # would mean a power loss mid-update truncates a live spool
            # file and loses the message outright, which is the one
            # outcome this tool exists to prevent.
            write_record_atomically(pending_path, message_record)
            if message_record["attempt_count"] >= configuration["max_delivery_attempts"]:
                dead_path = dead_letter_directory / pending_path.name
                os.replace(pending_path, dead_path)
                fsync_directory(dead_path.parent)
                fsync_directory(pending_path.parent)
                print(f"dead lettered after {message_record['attempt_count']} "
                      f"attempts: {pending_path.name}", file=sys.stderr)
        except OSError as connection_error:
            # Socket-level failure (refused, timeout, unreachable): stop.
            print(f"relay unreachable, stopping flush: {connection_error}",
                  file=sys.stderr)
            break
    remaining_count = len(list(Path(configuration["spool_directory"]).glob("*.json")))
    fcntl.flock(lock_handle, fcntl.LOCK_UN)
    lock_handle.close()
    return 1 if remaining_count else 0


def command_status(configuration):
    pending_count = len(list(Path(configuration["spool_directory"]).glob("*.json")))
    dead_count = len(list(Path(configuration["dead_letter_directory"]).glob("*.json")))
    print(f"pending: {pending_count}  dead_letters: {dead_count}")
    if dead_count:
        return 2
    return 1 if pending_count else 0


def main():
    argument_parser = argparse.ArgumentParser(description=__doc__)
    subcommand_parsers = argument_parser.add_subparsers(dest="command", required=True)
    send_parser = subcommand_parsers.add_parser("send")
    send_parser.add_argument("--subject", required=True)
    send_parser.add_argument("--body")
    subcommand_parsers.add_parser("flush")
    subcommand_parsers.add_parser("status")
    arguments = argument_parser.parse_args()

    try:
        configuration = load_configuration()
    except (OSError, json.JSONDecodeError) as configuration_error:
        print(f"configuration error: {configuration_error}", file=sys.stderr)
        return 2

    if arguments.command == "send":
        return command_send(configuration, arguments)
    if arguments.command == "flush":
        return command_flush(configuration)
    return command_status(configuration)


if __name__ == "__main__":
    sys.exit(main())

Deployment and integration

# Callers alert like this; exit code 1 means spooled, not lost
printf '%s\n' "$alert_body" | smtp_alert_dispatcher.py send \
    --subject "ALERT: connectivity failures on ${target_host}"

# Flush the spool every five minutes; harmless no-op when empty
*/5 * * * * root /usr/local/bin/smtp_alert_dispatcher.py flush

# Watch for dead letters in whatever check wraps everything else
/usr/local/bin/smtp_alert_dispatcher.py status

The bash tools swap their mail invocation for the send command and gain the spool for free. Worth stating the resulting boundary honestly: on hosts that have a real MTA, keep using the MTA, whose queue is more battle-tested than this will ever be. This tool serves the hosts that lack one, and it should compete with "raw socket and hope" rather than with postfix.

Known limitations

Plain text only, no attachments, by intent; an alert with an attachment is a report, and reports have other channels. The SMTP password lives in a root-readable file rather than a secrets manager, which is a deliberate floor for this deployment tier and the obvious thing to replace first in an environment that already runs a secret store.

Retry timing belongs to the flush schedule rather than to the message. A spooled record carries an attempt count and no next-attempt timestamp, so worst-case delivery latency is a function of the cron interval, and there is no per-message backoff or jitter. Against a relay that is simply down, that is acceptable at this scale. Against a relay that is rate-limiting, it is the wrong shape.

Two failure paths stay deliberately unhandled. SMTPDataError and other transient 4xx replies land in the message-rejection branch and are charged an attempt, which suits a bad recipient better than it suits a busy relay. And on the send path, a delivery failure followed by a spool-write failure has no second fallback: the spool exception escapes and the process exits 1, the same code that means the message was spooled. A full or unwritable spool directory therefore reports durable storage it failed to achieve. Giving that its own exit code is the next change to this tool.

Spool growth during a very long relay outage is bounded only by disk; a spool-size warning in status is the cheap fix and is on the roadmap. Delivery is at-least-once: a crash after the relay accepts but before the spool file is unlinked can produce a duplicate alert on the next flush, and for alerting, duplicate beats lost by a wide margin.