diff --git a/steward/app.py b/steward/app.py index 436f159..6551942 100644 --- a/steward/app.py +++ b/steward/app.py @@ -438,7 +438,21 @@ def _register_core_tasks(app: Quart) -> None: from .core.self_monitor import record_self_metrics await record_self_metrics(app) + async def run_email_outbox(): + from .core.outbox import send_due + await send_due(app) + + from .core.outbox import SENDER_TICK_S app._task_registry.extend([ + # Every email is a row until the mail server accepts it (#5506); this + # sends what is due and retries what failed. On startup too, so what a + # restart left unsent goes out at once. + ScheduledTask( + name="email_outbox", + coro_factory=run_email_outbox, + interval_seconds=SENDER_TICK_S, + run_on_startup=True, + ), ScheduledTask( name="weekly_report_check", coro_factory=run_report_check, diff --git a/steward/core/cleanup.py b/steward/core/cleanup.py index 3640bbd..6b9d0ec 100644 --- a/steward/core/cleanup.py +++ b/steward/core/cleanup.py @@ -12,6 +12,7 @@ from steward.models.alerts import AlertEvent from steward.models.ansible import AnsibleRun from steward.models.audit import AuditEvent from steward.models.monitors import MonitorResult +from steward.models.outbox import EmailOutbox if TYPE_CHECKING: from quart import Quart @@ -23,7 +24,8 @@ logger = logging.getLogger(__name__) # own with prune_by_age() in setup(), so the core never imports plugin models. # Tables kept another way are not listed: plugin_metrics and the docker tables # are rolled up (below), and the audit log has its own, longer window. -_AGED: list = [MonitorResult.checked_at, AnsibleRun.started_at, AlertEvent.transitioned_at] +_AGED: list = [MonitorResult.checked_at, AnsibleRun.started_at, AlertEvent.transitioned_at, + EmailOutbox.created_at] def prune_by_age(column) -> None: diff --git a/steward/core/mail.py b/steward/core/mail.py index 87ae983..03511f7 100644 --- a/steward/core/mail.py +++ b/steward/core/mail.py @@ -180,14 +180,15 @@ def sender(cfg: dict) -> str: return cfg.get("from") or cfg.get("username") or "steward@localhost" -def message(cfg: dict, to: str, subject: str, text: str, html: str) -> EmailMessage: +def message(cfg: dict, to: str, subject: str, text: str, html: str | None = None) -> EmailMessage: msg = EmailMessage() # A header may not hold a line break; a rule name can (MCP takes any string). msg["Subject"] = " ".join(subject.split()) msg["From"] = sender(cfg) msg["To"] = to msg.set_content(text) - msg.add_alternative(html, subtype="html") + if html: + msg.add_alternative(html, subtype="html") return msg diff --git a/steward/core/notifications.py b/steward/core/notifications.py index cc1f17e..814fcda 100644 --- a/steward/core/notifications.py +++ b/steward/core/notifications.py @@ -55,7 +55,9 @@ async def _send_email(cfg: dict, alert: dict, recipients: list) -> dict: return {"sent": False, "error": str(exc)} -def _smtp_send_sync(cfg: dict, messages: list[EmailMessage]) -> None: +def smtp_open(cfg: dict) -> smtplib.SMTP: + """A connection to the mail server, TLS and login done, every step + bounded by SMTP_TIMEOUT_S. The caller closes it (use it in `with`).""" port = int(cfg.get("port") or 587) mode = smtp_mode(port, cfg.get("tls", True)) ctx = ssl.create_default_context() @@ -63,11 +65,19 @@ def _smtp_send_sync(cfg: dict, messages: list[EmailMessage]) -> None: smtp = smtplib.SMTP_SSL(cfg["host"], port, timeout=SMTP_TIMEOUT_S, context=ctx) else: smtp = smtplib.SMTP(cfg["host"], port, timeout=SMTP_TIMEOUT_S) - with smtp: + try: if mode == "starttls": smtp.starttls(context=ctx) if cfg.get("username"): smtp.login(cfg["username"], cfg.get("password", "")) + except BaseException: + smtp.close() + raise + return smtp + + +def _smtp_send_sync(cfg: dict, messages: list[EmailMessage]) -> None: + with smtp_open(cfg) as smtp: for msg in messages: smtp.send_message(msg) diff --git a/steward/core/outbox.py b/steward/core/outbox.py new file mode 100644 index 0000000..973d979 --- /dev/null +++ b/steward/core/outbox.py @@ -0,0 +1,233 @@ +"""Every email Steward sends goes out through here (#5506). + +An email is a row in email_outbox until the mail server accepts it (System +156 ruling, "Every email goes through a queue on the record"). The sender, +send_due, runs every SENDER_TICK_S. Each tick it: +- claims what is due; +- sends the batch over one connection; +- settles each row on what the server said. + +How a row settles: +- **accepted:** sent. +- **refused outright** (a 5xx reply, e.g. "no such mailbox"): failed at once, + with the server's reason. Retrying would only repeat the refusal. +- **temporary** (no connection, a timeout, a 4xx reply): tried again after + BACKOFF_S, and once more at the hour. If the server still hasn't taken it + by then, it is failed with the last error. +- **a test** (retry=False): settles on its first answer either way. + +Nothing is kept in memory. A row left behind by a restart is still pending +(or held, or claimed by a run that died), and the next tick picks it up. +""" +from __future__ import annotations + +import asyncio +import logging +import smtplib +from datetime import datetime, timedelta, timezone +from email.message import EmailMessage +from typing import TYPE_CHECKING + +from sqlalchemy import or_, select, update + +from steward.models.outbox import EmailOutbox + +if TYPE_CHECKING: + from quart import Quart + from steward.core.mail import Recipient + +logger = logging.getLogger(__name__) + +SENDER_TICK_S = 15 +# Waits after the 1st, 2nd, … temporary failure; every later wait is the last. +BACKOFF_S = (30, 60, 120, 300, 600, 900) +# An email the server hasn't accepted this long after its first attempt is +# failed (operator ruling: "retry fails at an hour"). +GIVE_UP_S = 3600 +# A claim older than this belongs to a run that died mid-send: back to pending. +STALE_CLAIM_S = 300 +BATCH = 50 + + +def backoff(attempts: int) -> int: + """Seconds to wait after the `attempts`-th failed try.""" + return BACKOFF_S[min(max(attempts, 1), len(BACKOFF_S)) - 1] + + +def classify(exc: BaseException) -> str: + """"permanent" when the mail server refused it (5xx), or the message + can't be sent as written; "temporary" when trying later may work: no + connection, a timeout, a dropped line, or a 4xx "try again".""" + if isinstance(exc, smtplib.SMTPRecipientsRefused): + codes = [code for code, _ in exc.recipients.values()] + return "permanent" if codes and all(500 <= c < 600 for c in codes) else "temporary" + if isinstance(exc, smtplib.SMTPResponseException): + return "permanent" if 500 <= exc.smtp_code < 600 else "temporary" + if isinstance(exc, smtplib.SMTPNotSupportedError): + return "permanent" # e.g. STARTTLS asked of a server without it + if isinstance(exc, (OSError, TimeoutError)): + return "temporary" # SMTPServerDisconnected, refused, unreachable, TLS + return "permanent" # a message we couldn't build: trying again repeats it + + +def reason(exc: BaseException) -> str: + """What went wrong, as the server said it where it said anything.""" + def said(code, msg) -> str: + text = msg.decode(errors="replace") if isinstance(msg, bytes) else str(msg) + return f"{code} {' '.join(text.split())}" + if isinstance(exc, smtplib.SMTPRecipientsRefused): + return "; ".join(f"{to}: {said(c, m)}" for to, (c, m) in exc.recipients.items()) + if isinstance(exc, smtplib.SMTPResponseException): + return said(exc.smtp_code, exc.smtp_error) + if isinstance(exc, TimeoutError): + return "The mail server didn't answer in time" + if isinstance(exc, ConnectionRefusedError): + return "The mail server refused the connection" + if isinstance(exc, smtplib.SMTPServerDisconnected): + return f"The mail server closed the connection{': ' + str(exc) if str(exc) else ''}" + return str(exc) or type(exc).__name__ + + +def enqueue(db, *, kind: str, to: "Recipient | str", subject: str | None = None, + text: str | None = None, html: str | None = None, + send_after: datetime | None = None, retry: bool = True, + now: datetime | None = None) -> EmailOutbox: + """Add one email to the queue in `db`'s transaction (the caller commits). + `send_after` holds it until then (an alert burst, #5507); otherwise it is + due at once. `to` is a mail.Recipient, or a bare address read in UTC.""" + now = now or datetime.now(timezone.utc) + address, tz, show_utc = (to, "UTC", False) if isinstance(to, str) else \ + (to.address, getattr(to.tz, "key", None) or "UTC", to.show_utc) + row = EmailOutbox(kind=kind, recipient=address, tz=tz, show_utc=show_utc, + subject=subject, text=text, html=html, retry=retry, created_at=now, + state="held" if send_after else "pending", send_after=send_after, + next_attempt_at=send_after or now) + db.add(row) + return row + + +def settle(row: EmailOutbox, outcome: BaseException | None, now: datetime) -> None: + """Record one attempt's outcome on `row`: None is the server's acceptance.""" + row.attempts += 1 + row.first_attempt_at = row.first_attempt_at or now + row.claimed_at = None + if outcome is None: + row.state, row.sent_at, row.last_error, row.next_attempt_at = "sent", now, None, None + return + row.last_error = reason(outcome) + deadline = row.first_attempt_at + timedelta(seconds=GIVE_UP_S) + if classify(outcome) == "permanent" or not row.retry or now >= deadline: + row.state, row.next_attempt_at = "failed", None + if now >= deadline and classify(outcome) == "temporary" and row.retry: + row.last_error += " (gave up after an hour of retries)" + return + # Retry; the last try lands at the hour, not after it. + row.state = "pending" + row.next_attempt_at = min(now + timedelta(seconds=backoff(row.attempts)), deadline) + + +def deliver_sync(cfg: dict, messages: list[EmailMessage]) -> list[BaseException | None]: + """Send each message over one connection; per message, None when the + server accepted it, else what went wrong. A failure to connect, or a + line dropped mid-batch, is every unsent message's outcome.""" + from steward.core.notifications import smtp_open + outcomes: list[BaseException | None] = [] + try: + with smtp_open(cfg) as smtp: + for msg in messages: + try: + smtp.send_message(msg) + outcomes.append(None) + except smtplib.SMTPServerDisconnected as exc: + outcomes.append(exc) + break + except Exception as exc: + outcomes.append(exc) + except Exception as exc: + outcomes += [exc] * (len(messages) - len(outcomes)) + return outcomes + [smtplib.SMTPServerDisconnected("connection lost")] * (len(messages) - len(outcomes)) + + +async def write_held(app: "Quart", row: EmailOutbox) -> None: + """Fill in a row queued without its words: an alert burst, written when + its window closes (#5507). Raises when it can't be written.""" + raise ValueError(f"no words to send for a {row.kind} email") + + +async def attempt(app: "Quart", ids: list[str], now: datetime | None = None) -> dict[str, str]: + """Try to send the claimed rows `ids` once, and settle each. The words + are written and kept first, so a retry sends the same email; no session + is held while the mail server is talked to. Returns {id: state after}.""" + from steward.core.mail import message + if not ids: + return {} + cfg = app.config.get("SMTP", {}) + outcomes: dict[str, BaseException | None] = {} + sendable: list[tuple[str, EmailMessage]] = [] + async with app.db_sessionmaker() as db: + for row in (await db.execute(select(EmailOutbox).where(EmailOutbox.id.in_(ids)))).scalars(): + if not cfg.get("host"): + outcomes[row.id] = smtplib.SMTPNotSupportedError( + "No mail server is set (Settings → Notifications)") + continue + try: + if row.subject is None: + await write_held(app, row) + sendable.append((row.id, message(cfg, row.recipient, row.subject, row.text, row.html))) + except Exception as exc: + logger.exception("email %s: could not be written", row.id) + outcomes[row.id] = exc + await db.commit() + if sendable: + sent = await asyncio.get_running_loop().run_in_executor( + None, deliver_sync, cfg, [m for _, m in sendable]) + outcomes.update({rid: out for (rid, _), out in zip(sendable, sent)}) + now = now or datetime.now(timezone.utc) + async with app.db_sessionmaker() as db: + async with db.begin(): + rows = list((await db.execute( + select(EmailOutbox).where(EmailOutbox.id.in_(ids)))).scalars()) + for row in rows: + settle(row, outcomes.get(row.id, RuntimeError("not tried")), now) + if row.state == "failed": + logger.warning("email %s to %s failed: %s", row.kind, row.recipient, row.last_error) + await after_settle(db, rows) + return {row.id: row.state for row in rows} + + +async def after_settle(db, rows: list[EmailOutbox]) -> None: + """What else follows from these rows' new states, in the same + transaction: an alert event's email outcome (#5507).""" + + +async def claim_due(app: "Quart", now: datetime, limit: int = BATCH) -> list[str]: + """Mark what is due as being sent, and return its ids: pending rows whose + time has come, held rows whose window has closed, and rows a dead run + had claimed. SKIP LOCKED, so two senders never claim one row.""" + stale = now - timedelta(seconds=STALE_CLAIM_S) + async with app.db_sessionmaker() as db: + async with db.begin(): + await db.execute(update(EmailOutbox) + .where(EmailOutbox.state == "sending", EmailOutbox.claimed_at < stale) + .values(state="pending", claimed_at=None, next_attempt_at=now)) + ids = list((await db.execute( + select(EmailOutbox.id) + .where(or_(EmailOutbox.state == "pending", EmailOutbox.state == "held"), + EmailOutbox.next_attempt_at <= now) + .order_by(EmailOutbox.next_attempt_at) + .limit(limit) + .with_for_update(skip_locked=True))).scalars()) + if ids: + await db.execute(update(EmailOutbox).where(EmailOutbox.id.in_(ids)) + .values(state="sending", claimed_at=now)) + return ids + + +async def send_due(app: "Quart") -> None: + """The sender's tick: claim what is due and try it once.""" + now = datetime.now(timezone.utc) + ids = await claim_due(app, now) + if ids: + states = await attempt(app, ids, now) + logger.info("email outbox: %d tried, %d sent", len(states), + sum(1 for s in states.values() if s == "sent")) diff --git a/steward/migrations/versions/0042_email_outbox.py b/steward/migrations/versions/0042_email_outbox.py new file mode 100644 index 0000000..3c60e45 --- /dev/null +++ b/steward/migrations/versions/0042_email_outbox.py @@ -0,0 +1,57 @@ +"""Outgoing email as rows until the mail server accepts them (#5506), and +which alert events each one carries (#5507). + +Revision ID: 0042_email_outbox +Revises: 0041_alert_rule_covers +Create Date: 2026-10-09 +""" +from typing import Sequence, Union +from alembic import op +import sqlalchemy as sa + +revision: str = "0042_email_outbox" +down_revision: Union[str, None] = "0041_alert_rule_covers" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "email_outbox", + sa.Column("id", sa.String(36), primary_key=True), + sa.Column("kind", sa.String(16), nullable=False), + sa.Column("recipient", sa.String(320), nullable=False), + sa.Column("tz", sa.String(64), nullable=False, server_default="UTC"), + sa.Column("show_utc", sa.Boolean, nullable=False, server_default=sa.false()), + sa.Column("subject", sa.Text, nullable=True), + sa.Column("text", sa.Text, nullable=True), + sa.Column("html", sa.Text, nullable=True), + sa.Column("state", sa.String(10), nullable=False, server_default="pending"), + sa.Column("retry", sa.Boolean, nullable=False, server_default=sa.true()), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("send_after", sa.DateTime(timezone=True), nullable=True), + sa.Column("next_attempt_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("first_attempt_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("attempts", sa.Integer, nullable=False, server_default="0"), + sa.Column("claimed_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("last_error", sa.Text, nullable=True), + sa.Column("sent_at", sa.DateTime(timezone=True), nullable=True), + ) + op.create_index("ix_email_outbox_state_due", "email_outbox", ["state", "next_attempt_at"]) + op.create_index("ix_email_outbox_created", "email_outbox", ["created_at"]) + op.create_table( + "email_outbox_events", + sa.Column("outbox_id", sa.String(36), + sa.ForeignKey("email_outbox.id", ondelete="CASCADE"), primary_key=True), + sa.Column("event_id", sa.String(36), + sa.ForeignKey("alert_events.id", ondelete="CASCADE"), primary_key=True), + ) + op.create_index("ix_email_outbox_events_event_id", "email_outbox_events", ["event_id"]) + + +def downgrade() -> None: + op.drop_index("ix_email_outbox_events_event_id", table_name="email_outbox_events") + op.drop_table("email_outbox_events") + op.drop_index("ix_email_outbox_created", table_name="email_outbox") + op.drop_index("ix_email_outbox_state_due", table_name="email_outbox") + op.drop_table("email_outbox") diff --git a/steward/models/__init__.py b/steward/models/__init__.py index 78e7186..a5ceb45 100644 --- a/steward/models/__init__.py +++ b/steward/models/__init__.py @@ -12,6 +12,7 @@ from .ansible_inventory import AnsibleTarget, AnsibleGroup, ansible_target_group from .ansible_schedule import AnsibleSchedule from .settings import AppSetting from .dashboard import Dashboard, DashboardWidget, DashboardShareToken +from .outbox import EmailOutbox, email_outbox_events __all__ = [ "Base", "User", "ApiToken", "UserInvite", "PasswordReset", @@ -24,4 +25,5 @@ __all__ = [ "AnsibleSchedule", "AppSetting", "Dashboard", "DashboardWidget", "DashboardShareToken", + "EmailOutbox", "email_outbox_events", ] diff --git a/steward/models/outbox.py b/steward/models/outbox.py new file mode 100644 index 0000000..3494272 --- /dev/null +++ b/steward/models/outbox.py @@ -0,0 +1,68 @@ +"""Outgoing email, one row per message per recipient, until the mail server +accepts it (#5506). Sent by steward/core/outbox.py; the System 156 ruling +"Every email goes through a queue on the record" is why it is a table and +not something held in memory: a restart picks up whatever is unsent. +""" +from __future__ import annotations + +from datetime import datetime, timezone + +from sqlalchemy import (Boolean, Column, DateTime, ForeignKey, Index, Integer, String, Table, + Text) +from sqlalchemy.orm import Mapped, mapped_column + +from steward.core.ids import time_ordered_id + +from .base import Base + +# held: an alert email gathering its burst until send_after (#5507) +# pending: due at next_attempt_at, first try or a retry after a temporary failure +# sending: claimed by a sender; back to pending if the claim goes stale +# sent: the mail server accepted it +# failed: refused outright, a test that didn't go, or an hour of retries +STATES = ("held", "pending", "sending", "sent", "failed") + + +class EmailOutbox(Base): + __tablename__ = "email_outbox" + + id: Mapped[str] = mapped_column(String(36), primary_key=True, default=time_ordered_id) + kind: Mapped[str] = mapped_column(String(16), nullable=False) # alert | report | invite | reset | test + recipient: Mapped[str] = mapped_column(String(320), nullable=False) + # How the recipient reads times, kept so an alert email rendered when its + # burst closes is still written in their zone. + tz: Mapped[str] = mapped_column(String(64), nullable=False, default="UTC") + show_utc: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + # Rendered at enqueue, or (an alert burst) when the window closes; a retry + # sends the same words. + subject: Mapped[str | None] = mapped_column(Text, nullable=True) + text: Mapped[str | None] = mapped_column(Text, nullable=True) + html: Mapped[str | None] = mapped_column(Text, nullable=True) + state: Mapped[str] = mapped_column(String(10), nullable=False, default="pending") + retry: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc)) + send_after: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + next_attempt_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + first_attempt_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + attempts: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + claimed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + last_error: Mapped[str | None] = mapped_column(Text, nullable=True) + sent_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + + __table_args__ = ( + # The sender's question each tick: what is due, in each open state. + Index("ix_email_outbox_state_due", "state", "next_attempt_at"), + Index("ix_email_outbox_created", "created_at"), + ) + + +# Which alert events an email carries (#5507): a burst carries several, and +# each event is carried by one email per recipient. +email_outbox_events = Table( + "email_outbox_events", Base.metadata, + Column("outbox_id", String(36), ForeignKey("email_outbox.id", ondelete="CASCADE"), + primary_key=True), + Column("event_id", String(36), ForeignKey("alert_events.id", ondelete="CASCADE"), + primary_key=True, index=True), +) diff --git a/tests/core/test_outbox.py b/tests/core/test_outbox.py new file mode 100644 index 0000000..092bab6 --- /dev/null +++ b/tests/core/test_outbox.py @@ -0,0 +1,162 @@ +"""The email outbox's decisions (#5506): which failures are worth another +try, how long to wait, when to give up, and what one connection's batch +reports per message. The table and the sender's ticks are in +tests/integration/test_email_outbox.py.""" +import smtplib +from datetime import datetime, timedelta, timezone +from email.message import EmailMessage +from zoneinfo import ZoneInfo + +import pytest + +from steward.core import mail, outbox +from steward.models.outbox import EmailOutbox + +NOW = datetime(2026, 10, 9, 12, 0, tzinfo=timezone.utc) + + +def _row(**kw) -> EmailOutbox: + base = dict(kind="report", recipient="ann@example.com", retry=True, attempts=0, + state="sending", first_attempt_at=None, claimed_at=NOW) + return EmailOutbox(**{**base, **kw}) + + +@pytest.mark.parametrize("exc, kind", [ + (smtplib.SMTPRecipientsRefused({"a@x": (550, b"no such user")}), "permanent"), + (smtplib.SMTPRecipientsRefused({"a@x": (550, b"no"), "b@x": (451, b"later")}), "temporary"), + (smtplib.SMTPDataError(554, b"rejected as spam"), "permanent"), + (smtplib.SMTPSenderRefused(553, b"not yours", "steward@x"), "permanent"), + (smtplib.SMTPAuthenticationError(535, b"bad credentials"), "permanent"), + (smtplib.SMTPDataError(451, b"greylisted, try later"), "temporary"), + (smtplib.SMTPServerDisconnected("gone"), "temporary"), + (ConnectionRefusedError(111, "refused"), "temporary"), + (TimeoutError(), "temporary"), + (smtplib.SMTPNotSupportedError("STARTTLS extension not supported"), "permanent"), + (ValueError("no words"), "permanent"), +]) +def test_a_refusal_is_final_and_anything_that_may_pass_is_retried(exc, kind): + assert outbox.classify(exc) == kind + + +def test_backoff_grows_then_holds_at_its_last_step(): + assert [outbox.backoff(n) for n in range(1, 9)] == [30, 60, 120, 300, 600, 900, 900, 900] + + +def test_reason_is_what_the_server_said(): + assert outbox.reason(smtplib.SMTPDataError(554, b"5.7.1 rejected\r\n as spam")) \ + == "554 5.7.1 rejected as spam" + assert outbox.reason(smtplib.SMTPRecipientsRefused({"a@x": (550, b"no such user")})) \ + == "a@x: 550 no such user" + assert outbox.reason(TimeoutError()) == "The mail server didn't answer in time" + + +def test_accepted_is_sent(): + row = _row() + outbox.settle(row, None, NOW) + assert (row.state, row.sent_at, row.attempts, row.claimed_at) == ("sent", NOW, 1, None) + assert row.first_attempt_at == NOW and row.next_attempt_at is None + + +def test_a_refusal_fails_at_once_with_the_servers_words(): + row = _row() + outbox.settle(row, smtplib.SMTPRecipientsRefused({"ann@example.com": (550, b"no such user")}), NOW) + assert row.state == "failed" and row.next_attempt_at is None + assert row.last_error == "ann@example.com: 550 no such user" + + +def test_a_temporary_failure_waits_and_tries_again(): + row = _row() + outbox.settle(row, smtplib.SMTPDataError(451, b"try later"), NOW) + assert row.state == "pending" and row.next_attempt_at == NOW + timedelta(seconds=30) + later = NOW + timedelta(seconds=30) + outbox.settle(row, ConnectionRefusedError(), later) + assert row.attempts == 2 and row.next_attempt_at == later + timedelta(seconds=60) + assert row.first_attempt_at == NOW + + +def test_the_last_try_lands_at_the_hour_and_then_it_fails(): + row = _row(attempts=7, first_attempt_at=NOW) + near = NOW + timedelta(seconds=3300) + outbox.settle(row, TimeoutError(), near) + # 900 s would overshoot the hour: the try is pulled back to it. + assert row.state == "pending" and row.next_attempt_at == NOW + timedelta(seconds=3600) + outbox.settle(row, TimeoutError(), NOW + timedelta(seconds=3600)) + assert row.state == "failed" + assert row.last_error == "The mail server didn't answer in time (gave up after an hour of retries)" + + +def test_a_test_send_settles_on_its_first_answer(): + row = _row(kind="test", retry=False) + outbox.settle(row, ConnectionRefusedError(), NOW) + assert row.state == "failed" and row.last_error == "The mail server refused the connection" + + +def test_enqueue_keeps_the_readers_zone_and_holds_until_send_after(): + added = [] + + class _Db: + add = added.append + + ann = mail.Recipient("ann@example.com", ZoneInfo("America/New_York"), True, "ann") + now_row = outbox.enqueue(_Db, kind="report", to=ann, subject="s", text="t", now=NOW) + assert (now_row.state, now_row.next_attempt_at, now_row.tz, now_row.show_utc) == \ + ("pending", NOW, "America/New_York", True) + held = outbox.enqueue(_Db, kind="alert", to="bob@example.com", + send_after=NOW + timedelta(minutes=2), now=NOW) + assert (held.state, held.next_attempt_at, held.tz) == ("held", NOW + timedelta(minutes=2), "UTC") + assert added == [now_row, held] + + +class _FakeSMTP: + def __init__(self, answers): + self.answers, self.sent = list(answers), [] + + def __enter__(self): + return self + + def __exit__(self, *exc): + return False + + def send_message(self, msg): + answer = self.answers.pop(0) + if answer is not None: + raise answer + self.sent.append(msg) + + +def _msgs(n): + out = [] + for i in range(n): + m = EmailMessage() + m["Subject"] = f"m{i}" + out.append(m) + return out + + +def test_one_connection_reports_each_message(monkeypatch): + from steward.core import notifications + smtp = _FakeSMTP([None, smtplib.SMTPRecipientsRefused({"b@x": (550, b"no")}), None]) + monkeypatch.setattr(notifications, "smtp_open", lambda cfg: smtp) + out = outbox.deliver_sync({}, _msgs(3)) + assert out[0] is None and isinstance(out[1], smtplib.SMTPRecipientsRefused) and out[2] is None + assert len(smtp.sent) == 2 + + +def test_a_dropped_line_is_every_unsent_messages_outcome(monkeypatch): + from steward.core import notifications + gone = smtplib.SMTPServerDisconnected("gone") + monkeypatch.setattr(notifications, "smtp_open", lambda cfg: _FakeSMTP([None, gone])) + out = outbox.deliver_sync({}, _msgs(4)) + assert out[0] is None and out[1] is gone + assert all(isinstance(o, smtplib.SMTPServerDisconnected) for o in out[2:]) and len(out) == 4 + + +def test_no_connection_is_every_messages_outcome(monkeypatch): + from steward.core import notifications + refused = ConnectionRefusedError(111, "refused") + + def _open(cfg): + raise refused + + monkeypatch.setattr(notifications, "smtp_open", _open) + assert outbox.deliver_sync({}, _msgs(2)) == [refused, refused] diff --git a/tests/integration/test_email_outbox.py b/tests/integration/test_email_outbox.py new file mode 100644 index 0000000..9e08c44 --- /dev/null +++ b/tests/integration/test_email_outbox.py @@ -0,0 +1,108 @@ +"""Integration: the email outbox's sender against Postgres (#5506). + +A due row is sent on the sender's first tick, a refusal fails it, a +temporary failure leaves it waiting for its retry, a held row waits for its +window, and a claim left by a run that died goes back to the queue. Only the +SMTP conversation itself is replaced. +""" +from __future__ import annotations + +import asyncio +import os +import smtplib +import uuid +from datetime import datetime, timedelta, timezone + +import pytest + +pytestmark = pytest.mark.integration + + +@pytest.fixture +def app(): + if not os.environ.get("STEWARD_DATABASE_URL"): + pytest.skip("needs Postgres") + from steward.app import create_app + app = create_app(testing=False) + app.config["SMTP"] = {"host": "mx.test", "port": 587, "from": "steward@test"} + return app + + +def test_the_sender_settles_each_row_on_what_the_server_said(app, monkeypatch): + from sqlalchemy import select + from steward.core import outbox + from steward.models.outbox import EmailOutbox + + tag = uuid.uuid4().hex[:8] + now = datetime.now(timezone.utc) + to = {k: f"{k}-{tag}@test" for k in ("ok", "refused", "later", "held", "stale")} + tried: list[str] = [] + + def fake_deliver(cfg, messages): + out = [] + for m in messages: + tried.append(m["To"]) + if m["To"] == to["refused"]: + out.append(smtplib.SMTPRecipientsRefused({m["To"]: (550, b"no such user")})) + elif m["To"] == to["later"]: + out.append(smtplib.SMTPDataError(451, b"greylisted")) + else: + out.append(None) + return out + + monkeypatch.setattr(outbox, "deliver_sync", fake_deliver) + + async def _go(): + async with app.db_sessionmaker() as s: + async with s.begin(): + for k in ("ok", "refused", "later"): + outbox.enqueue(s, kind="report", to=to[k], subject=f"s {k}", text="t", html="

t

") + outbox.enqueue(s, kind="report", to=to["held"], subject="s", text="t", + send_after=now + timedelta(minutes=2)) + # Claimed by a sender that died ten minutes ago. + dead = outbox.enqueue(s, kind="report", to=to["stale"], subject="s", text="t") + dead.state, dead.claimed_at = "sending", now - timedelta(minutes=10) + await outbox.send_due(app) + async with app.db_sessionmaker() as s: + rows = (await s.execute(select(EmailOutbox) + .where(EmailOutbox.recipient.in_(to.values())))).scalars().all() + return {r.recipient: r for r in rows} + + rows = asyncio.run(_go()) + by = {k: rows[to[k]] for k in to} + assert by["ok"].state == "sent" and by["ok"].sent_at is not None and by["ok"].attempts == 1 + assert by["refused"].state == "failed" + assert by["refused"].last_error == f"{to['refused']}: 550 no such user" + later = by["later"] + assert later.state == "pending" and later.attempts == 1 and later.last_error == "451 greylisted" + assert later.next_attempt_at > later.first_attempt_at + # The held row is not due for two minutes; nothing went to it. + assert by["held"].state == "held" and by["held"].attempts == 0 and to["held"] not in tried + # The dead run's claim was released and the row sent in the same tick. + assert by["stale"].state == "sent" and by["stale"].claimed_at is None + assert sorted(t for t in tried if t.endswith(f"{tag}@test")) == sorted( + to[k] for k in ("ok", "refused", "later", "stale")) + + +def test_with_no_mail_server_a_row_fails_and_says_where_to_set_one(app, monkeypatch): + from sqlalchemy import select + from steward.core import outbox + from steward.models.outbox import EmailOutbox + + app.config["SMTP"] = {} + monkeypatch.setattr(outbox, "deliver_sync", lambda cfg, msgs: pytest.fail("nothing to send to")) + addr = f"nomx-{uuid.uuid4().hex[:8]}@test" + + async def _go(): + async with app.db_sessionmaker() as s: + async with s.begin(): + row = outbox.enqueue(s, kind="report", to=addr, subject="s", text="t") + await s.flush() + rid = row.id + await outbox.attempt(app, [rid]) + async with app.db_sessionmaker() as s: + return (await s.execute(select(EmailOutbox).where(EmailOutbox.recipient == addr))).scalar_one() + + row = asyncio.run(_go()) + assert row.state == "failed" + assert row.last_error == "No mail server is set (Settings → Notifications)"