feat(email): the report, invites, resets and test sends go through the outbox; only it talks SMTP (#5508)

#5508 "Every other email through the outbox: weekly report, invites,
password resets, and the test buttons" (milestone 501).

- outbox.send_each queues one email per reader (claimed, so the sender
  doesn't take it too) and tries them at once, for a page that answers
  with what happened; outbox.said words the result. attempt takes a cfg
  for a test of the form's unsaved settings and returns the settled rows.
- Weekly report (scheduled and Send now) and invites: queued, tried at
  once; one the server couldn't take yet is retried for an hour. An invite
  that didn't go shows its link with the reason, as before.
- Password reset: queued in the reset link's own transaction; the sender
  delivers it on its next tick.
- Settings -> Notifications test email and a rule's Send test: queued and
  recorded with retry=False, tried once (System 156 ruling).
- Removed mail.send, notifications._send_email and _smtp_send_sync;
  dispatch_notifications is webhook only. A guard asserts only the outbox
  (and smtp_open itself) opens an SMTP connection, with a self-test.

Also fixes run 8941 (#5507's integration test): its Also-send-to
addresses were "@test", which is_real_address rightly drops; they are
example.com now.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

steward/alerts/service.py

@@ -266,9 +266,9 @@ async def active_count(db) -> int:
 
 def notify_targets(smtp: dict, webhook: dict, accounts: int = 0) -> list[str]:
     """Where an alert's notifications go, in words, by the same tests
-    core.notifications.dispatch_notifications sends by: email when a server
-    is set and `accounts` (that get alerts) or Also send to addresses exist,
-    a webhook when its URL is. Empty: nobody (#5315, milestone 495)."""
+    alerts are sent by: email when a server is set and `accounts` (that get
+    alerts) or Also send to addresses exist (core.outbox.hold_alert), a
+    webhook when its URL is (core.notifications.dispatch_notifications). Empty: nobody (#5315, milestone 495)."""
     from urllib.parse import urlsplit
     out = []
     recipients = smtp.get("recipients") or []
@@ -752,7 +752,8 @@ async def send_test(app, rule_id: str) -> dict:
     failed; lines say what happened to each channel. LookupError for no rule.
     """
     from urllib.parse import urlsplit
-    from steward.core.mail import recipients
+    from steward.core import outbox
+    from steward.core.mail import alert_email, recipients
     from steward.core.notifications import dispatch_notifications
     smtp, webhook = app.config.get("SMTP", {}), app.config.get("WEBHOOK", {})
     async with app.db_sessionmaker() as db:
@@ -761,12 +762,14 @@ async def send_test(app, rule_id: str) -> dict:
             raise LookupError(rule_id)
         alert = await test_alert(db, rule)
         to = await recipients(db, "alerts", smtp) if smtp.get("host") else []
-    results = await dispatch_notifications(smtp, webhook, alert, to)
+    results = await dispatch_notifications(webhook, alert)
     lines = []
-    if email := results.get("email"):
-        who = ", ".join(r.address for r in to)
-        lines.append(f"Email sent to {who}." if email["sent"]
-                     else f"Email to {who} failed: {email['error']}")
+    if to:
+        # Queued and recorded like any email, tried once, not retried.
+        ok, said = outbox.said(await outbox.send_each(
+            app, "test", to, lambda r: alert_email(smtp, alert, r), retry=False))
+        results["email"] = {"sent": ok, "error": None if ok else said}
+        lines.append(f"Email: {said}")
     if hook := results.get("webhook"):
         host = urlsplit(webhook["url"]).hostname or webhook["url"]
         lines.append(f"Webhook to {host} delivered." if hook["sent"]

steward/auth/routes.py

@@ -436,7 +436,7 @@ async def forgot_post():
     accounts; every request from an address counts toward its limit."""
     from quart import current_app
     from steward.auth.throttle import reset_throttle
-    from steward.core import mail
+    from steward.core import mail, outbox
     from steward.core.audit import log_audit
     from steward.core.formatting import zone
     from steward.core.password_resets import account_for, issue, link
@@ -454,16 +454,15 @@ async def forgot_post():
             user = await account_for(db, name)
             if user is not None:
                 reset, key = await issue(db, user)
-                to, username, user_id, tz = user.email, user.username, user.id, user.time_zone
+                username, user_id = user.username, user.id
+                to = mail.Recipient(user.email, zone(user.time_zone), False, username)
+                subject, text, html = mail.reset_email(username, reset.expires_at,
+                                                       link(public_base_url(request), key), to.tz)
+                # Sent by the outbox, with the link's own transaction; the page
+                # says the same either way, and the queue records how it went.
+                outbox.enqueue(db, kind="reset", to=to, subject=subject, text=text, html=html)
     if user is None:
         return await _forgot_page(sent=True)
-    cfg = current_app.config.get("SMTP", {})
-    subject, text, html = mail.reset_email(username, reset.expires_at,
-                                           link(public_base_url(request), key), zone(tz))
-    try:
-        await mail.send(cfg, [mail.message(cfg, to, subject, text, html)])
-    except Exception as exc:                # the page still says the same; the log says why
-        current_app.logger.warning("Password reset email to %s failed: %s", to, exc)
     await log_audit(current_app, user_id, username, "user.password_reset_requested", "user", user_id,
                     detail={"address": ip})
     return await _forgot_page(sent=True)

steward/core/alerts.py

@@ -522,7 +522,7 @@ async def _dispatch_notification(
             await outbox.hold_alert(app, alert_data, event_id)
         except Exception:
             logger.exception("Alert %s: its email could not be queued", event_id)
-    results = await dispatch_notifications({}, app.config.get("WEBHOOK", {}), alert_data)
+    results = await dispatch_notifications(app.config.get("WEBHOOK", {}), alert_data)
 
     webhook_r = results.get("webhook")
     if not webhook_r:

steward/core/mail.py

@@ -163,13 +163,6 @@ async def audience(db) -> dict:
     return out
 
 
-async def send(cfg: dict, messages: list[EmailMessage]) -> None:
-    """Send every message over one SMTP connection, off the event loop."""
-    import asyncio
-    from steward.core.notifications import _smtp_send_sync
-    await asyncio.get_running_loop().run_in_executor(None, _smtp_send_sync, cfg, messages)
-
-
 def _render(name: str, **ctx) -> tuple[str, str]:
     ctx = {"c": palette(), **ctx}
     return (_env.get_template(f"{name}.txt").render(ctx),

steward/core/notifications.py

@@ -4,7 +4,6 @@ import logging
 import smtplib
 import ssl
 import types
-from email.message import EmailMessage
 
 import httpx
 from jinja2.sandbox import SandboxedEnvironment
@@ -12,20 +11,11 @@ from jinja2.sandbox import SandboxedEnvironment
 logger = logging.getLogger(__name__)
 
 
-async def dispatch_notifications(
-    smtp_cfg: dict,
-    webhook_cfg: dict,
-    alert_data: dict,
-    recipients: list = (),
-) -> dict:
-    """Dispatch all configured notification channels: email to each of
-    `recipients` (mail.Recipient, from mail.recipients) when a server is set.
-    Returns results dict."""
+async def dispatch_notifications(webhook_cfg: dict, alert_data: dict) -> dict:
+    """Post an alert's webhook when one is set; {"webhook": result} or {}.
+    Email goes through the outbox instead (core.outbox, #5506)."""
     results = {}
 
-    if smtp_cfg.get("host") and recipients:
-        results["email"] = await _send_email(smtp_cfg, alert_data, recipients)
-
     if webhook_cfg.get("url"):
         results["webhook"] = await _send_webhook(webhook_cfg, alert_data)
 
@@ -43,21 +33,10 @@ def smtp_mode(port, tls) -> str:
     return "starttls" if tls else "plain"
 
 
-async def _send_email(cfg: dict, alert: dict, recipients: list) -> dict:
-    """Each recipient their own copy, in their own zone, over one connection."""
-    from steward.core.mail import alert_email, message, send
-    try:
-        messages = [message(cfg, to.address, *alert_email(cfg, alert, to)) for to in recipients]
-        await send(cfg, messages)
-        return {"sent": True, "error": None}
-    except Exception as exc:
-        logger.error(f"Email notification failed: {exc}")
-        return {"sent": False, "error": str(exc)}
-
-
 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`)."""
+    bounded by SMTP_TIMEOUT_S. The caller closes it (use it in `with`).
+    Only core.outbox.deliver_sync opens one: every email is a queued row."""
     port = int(cfg.get("port") or 587)
     mode = smtp_mode(port, cfg.get("tls", True))
     ctx = ssl.create_default_context()
@@ -76,12 +55,6 @@ def smtp_open(cfg: dict) -> smtplib.SMTP:
     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)
-
-
 def _json_str(value) -> str:
     """A value as it may sit between the quotes of a JSON string."""
     return json.dumps("" if value is None else str(value))[1:-1]

steward/core/outbox.py

@@ -33,7 +33,7 @@ import logging
 import smtplib
 from datetime import datetime, timedelta, timezone
 from email.message import EmailMessage
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Callable
 
 from sqlalchemy import insert, or_, select, text, update
 
@@ -104,18 +104,22 @@ def reason(exc: BaseException) -> str:
 
 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,
+            send_after: datetime | None = None, retry: bool = True, claimed: bool = False,
             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."""
+    due at once. `claimed` marks it as being sent by the caller, who will
+    `attempt` it right after committing, so the sender doesn't take it too;
+    a claim the caller never settles goes back to the queue (STALE_CLAIM_S).
+    `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)
+                      state="held" if send_after else "sending" if claimed else "pending",
+                      send_after=send_after, next_attempt_at=send_after or now,
+                      claimed_at=now if claimed else None)
     db.add(row)
     return row
 
@@ -219,14 +223,18 @@ async def write_held(app: "Quart", db, row: EmailOutbox) -> None:
     row.subject, row.text, row.html = alert_words(app.config.get("SMTP", {}), alerts, to)
 
 
-async def attempt(app: "Quart", ids: list[str], now: datetime | None = None) -> dict[str, str]:
+async def attempt(app: "Quart", ids: list[str], now: datetime | None = None,
+                  cfg: dict | None = None) -> dict[str, EmailOutbox]:
     """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}."""
+    is held while the mail server is talked to. `cfg` replaces the saved mail
+    settings for this one try (a test of the form's unsaved values; such a
+    row has retry=False, so nothing later needs them). Returns {id: row}
+    as settled."""
     from steward.core.mail import message
     if not ids:
         return {}
-    cfg = app.config.get("SMTP", {})
+    cfg = cfg if cfg is not None else app.config.get("SMTP", {})
     outcomes: dict[str, BaseException | None] = {}
     sendable: list[tuple[str, EmailMessage]] = []
     async with app.db_sessionmaker() as db:
@@ -257,7 +265,7 @@ async def attempt(app: "Quart", ids: list[str], now: datetime | None = None) ->
                 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}
+        return {row.id: row for row in rows}
 
 
 async def after_settle(db, rows: list[EmailOutbox]) -> None:
@@ -315,6 +323,36 @@ async def send_due(app: "Quart") -> None:
     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"))
+        rows = await attempt(app, ids, now)
+        logger.info("email outbox: %d tried, %d sent", len(rows),
+                    sum(1 for r in rows.values() if r.state == "sent"))
+
+
+async def send_each(app: "Quart", kind: str, to: list, words: Callable[..., tuple],
+                    *, retry: bool = True, cfg: dict | None = None) -> list[EmailOutbox]:
+    """Queue one email per reader in `to` (mail.Recipient or address),
+    written by `words(reader)` -> (subject, text, html), and try them at
+    once, for a page that answers with what happened (see `said`). Returns
+    the rows as settled: a temporary failure is left pending for the sender
+    to retry, unless retry=False (a test). `cfg` as for `attempt`."""
+    async with app.db_sessionmaker() as db:
+        async with db.begin():
+            rows = [enqueue(db, kind=kind, to=r, subject=s, text=t, html=h, retry=retry, claimed=True)
+                    for r in to for s, t, h in [words(r)]]
+            await db.flush()
+            ids = [r.id for r in rows]
+    done = await attempt(app, ids, cfg=cfg)
+    return [done[i] for i in ids if i in done]
+
+
+def said(rows: list[EmailOutbox]) -> tuple[bool, str]:
+    """(all sent, what happened) for rows just tried: who it went to, who
+    refused and why, and who Steward is still trying."""
+    sent = [r.recipient for r in rows if r.state == "sent"]
+    failed = [f"{r.recipient}: {r.last_error}" for r in rows if r.state == "failed"]
+    later = [f"{r.recipient} ({r.last_error})" for r in rows if r.state == "pending"]
+    parts = ([f"Sent to {', '.join(sent)}."] if sent else []) + \
+        ([f"Not sent to {'; '.join(failed)}."] if failed else []) + \
+        ([f"Not sent yet to {'; '.join(later)}; Steward will try again for up to an hour."]
+         if later else [])
+    return bool(rows) and not failed and not later, " ".join(parts)

steward/core/reports.py

@@ -11,7 +11,7 @@ from sqlalchemy import and_, case, func, select
 
 from steward.alerts import service as alert_svc
 from steward.core.formatting import zone
-from steward.core.mail import message, recipients, report_email, send
+from steward.core.mail import recipients, report_email
 from steward.core.settings import get_setting
 from steward.models.alerts import AlertEvent
 from steward.models.metrics import PluginMetric
@@ -134,13 +134,13 @@ async def send_report(app) -> tuple[bool, str]:
         logger.exception("Failed to build report data")
         return False, f"Data collection failed: {exc}"
 
-    try:
-        await send(smtp_cfg, [message(smtp_cfg, r.address, *report_email(smtp_cfg, data, r)) for r in to])
-    except Exception as exc:
-        logger.error("Failed to send weekly report: %s", exc)
-        return False, f"Send failed: {exc}"
-    logger.info("Weekly report sent to %d address(es)", len(to))
-    return True, f"Report sent to {', '.join(r.address for r in to)}."
+    from steward.core import outbox
+    # Queued, then tried at once so the button can say what happened; one
+    # the server didn't take yet stays queued and is retried for an hour.
+    ok, said = outbox.said(await outbox.send_each(
+        app, "report", to, lambda r: report_email(smtp_cfg, data, r)))
+    logger.info("Weekly report: %s", said)
+    return ok, said
 
 
 def report_due(now: datetime, day: int, hour: int, tz) -> bool:

steward/settings/routes.py

@@ -319,7 +319,8 @@ async def test_email():
     the zone chosen for them. A blank password uses the saved one."""
     from steward.core.formatting import zone
     from steward.core.mail import Recipient, alert_email, is_real_address, pick_recipients
-    from steward.core.notifications import _send_email, smtp_mode
+    from steward.core import outbox
+    from steward.core.notifications import smtp_mode
     from steward.models.users import User
     form = await request.form
     async with current_app.db_sessionmaker() as db:
@@ -345,11 +346,12 @@ async def test_email():
             "email", False, "Your account has no email address mail can reach, and no Also send to "
                             "addresses are entered. Set your address on the Account page.")
     alert = _sample_alert()
-    sent = await _send_email(cfg, alert, to)
+    # Queued and recorded like any email, tried once with the form's values.
+    ok, said = outbox.said(await outbox.send_each(
+        current_app, "test", to, lambda r: alert_email(cfg, alert, r), retry=False, cfg=cfg))
     how = {"ssl": "implicit TLS", "starttls": "STARTTLS", "plain": "no TLS"}[smtp_mode(port, cfg["tls"])]
-    result = (f"Sent to {', '.join(r.address for r in to)} via {cfg['host']}:{port} ({how})."
-              if sent["sent"] else f"{cfg['host']}:{port} ({how}): {sent['error']}")
-    return await _notify_test_result("email", sent["sent"], result, alert_email(cfg, alert, to[0])[1])
+    result = f"{said[:-1]} via {cfg['host']}:{port} ({how})." if ok else f"{cfg['host']}:{port} ({how}): {said}"
+    return await _notify_test_result("email", ok, result, alert_email(cfg, alert, to[0])[1])
 
 
 # ── Ansible Sources ───────────────────────────────────────────────────────────

steward/settings/users.py

@@ -185,12 +185,15 @@ async def _deliver(invite: UserInvite, key: str) -> str | None:
     url = invites.link(public_base_url(request), key)
     subject, text, html = mail.invite_email(session.get("username", "An admin"), invite.role.value,
                                             invite.expires_at, url, zone(session.get("tz")))
-    try:
-        await mail.send(cfg, [mail.message(cfg, invite.email, subject, text, html)])
-    except Exception as exc:
-        current_app.logger.warning("Invite email to %s failed: %s", invite.email, exc)
-        return f"Sending to {invite.email} failed ({exc})."
-    return None
+    from steward.core import outbox
+    (row,) = await outbox.send_each(current_app, "invite", [invite.email],
+                                    lambda _: (subject, text, html))
+    if row.state == "sent":
+        return None
+    if row.state == "pending":
+        return (f"Sending to {invite.email} didn't go through yet ({row.last_error}); "
+                f"Steward will keep trying for up to an hour.")
+    return f"Sending to {invite.email} failed ({row.last_error})."
 
 
 async def _link_page(invite: UserInvite, key: str, why: str):