"""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. Alert email (#5507) is held: the first alert opens a row per recipient with send_after ALERT_WINDOW_S out, alerts arriving before it closes join the same rows, and the email is written when the window closes: the usual alert email for one, a list for several. Each alert's event records the outcome once every email carrying it has settled. 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 json import logging import smtplib from datetime import datetime, timedelta, timezone from email.message import EmailMessage from typing import TYPE_CHECKING, Callable from sqlalchemy import and_, insert, or_, select, text, update from steward.models.outbox import EmailOutbox, email_outbox_events 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 # Alerts that fire within this of the first go out as one email (System 156 # ruling: "change the timing on 3 to be 2 minutes"). One incident's alerts # arrive over a minute or so: the agent-offline check runs each minute. ALERT_WINDOW_S = 120 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): # Each message has one reader, whose row already names them. if len(exc.recipients) == 1: return said(*next(iter(exc.recipients.values()))) 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, 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. `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 "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 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 hold_alert(app: "Quart", alert: dict, event_id: str, now: datetime | None = None) -> int: """Queue an alert's email: join each recipient's open alert email, or open one that waits ALERT_WINDOW_S. Returns how many it went to. The event must be committed first (the link refers to it).""" from steward.core.mail import recipients cfg = app.config.get("SMTP", {}) now = now or datetime.now(timezone.utc) async with app.db_sessionmaker() as db: async with db.begin(): # One holder at a time, so two alerts at once can't each open a # window for the same reader. await db.execute(text("SELECT pg_advisory_xact_lock(hashtext('steward.email_outbox.alert'))")) to = await recipients(db, "alerts", cfg) if not to: return 0 open_rows = {r.recipient: r for r in (await db.execute( select(EmailOutbox).where(EmailOutbox.kind == "alert", EmailOutbox.state == "held", EmailOutbox.recipient.in_([r.address for r in to])) .with_for_update())).scalars()} rows = [open_rows.get(r.address) or enqueue( db, kind="alert", to=r, send_after=now + timedelta(seconds=ALERT_WINDOW_S), now=now) for r in to] await db.flush() await db.execute(insert(email_outbox_events), [ {"outbox_id": row.id, "event_id": event_id, "alert_json": json.dumps(alert)} for row in rows]) return len(to) def alert_words(cfg: dict, alerts: list[dict], to: "Recipient") -> tuple[str, str, str]: """(subject, text, html) for the alerts one email carries, oldest first: the usual alert email for one, a list of them for several (#5465).""" from steward.core.mail import alert_email, burst_email return alert_email(cfg, alerts[0], to) if len(alerts) == 1 else burst_email(cfg, alerts, to) async def write_held(app: "Quart", db, row: EmailOutbox) -> None: """Fill in a row queued without its words: an alert email, written when its window closes, in its reader's zone. Raises when it can't be.""" from zoneinfo import ZoneInfo from steward.core.mail import Recipient from steward.models.alerts import AlertEvent if row.kind != "alert": raise ValueError(f"no words to send for a {row.kind} email") alerts = [json.loads(a) for a in (await db.execute( select(email_outbox_events.c.alert_json) .join(AlertEvent, AlertEvent.id == email_outbox_events.c.event_id) .where(email_outbox_events.c.outbox_id == row.id) .order_by(AlertEvent.transitioned_at, AlertEvent.id))).scalars()] if not alerts: raise ValueError("The alerts this email carried were deleted") to = Recipient(row.recipient, ZoneInfo(row.tz), row.show_utc) 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, 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. `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 = 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: 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, db, 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 for row in rows} async def after_settle(db, rows: list[EmailOutbox]) -> None: """Record on each alert event its email's outcome, once every email carrying it has settled: sent when all were, else failed naming each reader it didn't reach and why. Same transaction as the settling.""" from steward.models.alerts import AlertEvent done = [r.id for r in rows if r.kind == "alert" and r.state in ("sent", "failed")] if not done: return link = email_outbox_events.c touched = select(link.event_id).where(link.outbox_id.in_(done)) emails: dict[str, list] = {} for event_id, state, recipient, error in (await db.execute( select(link.event_id, EmailOutbox.state, EmailOutbox.recipient, EmailOutbox.last_error) .join(EmailOutbox, EmailOutbox.id == link.outbox_id) .where(link.event_id.in_(touched)))).all(): emails.setdefault(event_id, []).append((state, recipient, error)) settled = {eid: es for eid, es in emails.items() if all(state in ("sent", "failed") for state, _, _ in es)} if not settled: return for event in (await db.execute( select(AlertEvent).where(AlertEvent.id.in_(settled)))).scalars(): failed = sorted(f"{rcpt}: {err}" for state, rcpt, err in settled[event.id] if state == "failed") event.email_sent = not failed event.email_error = "; ".join(failed) or None 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: 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) # ── What the pages show (#5509) ─────────────────────────────────────────────── KIND_LABELS = {"alert": "Alert", "report": "Weekly report", "invite": "Invite", "reset": "Password reset", "test": "Test"} # Retrying longer than this is worth a line in Settings → System's verdict. SLOW_S = 300 SHOWN_DAYS = 7 def delivery(rows: list[EmailOutbox]) -> dict: """One alert event's email, over every reader's copy, as history says it: {"state", "at", "attempts", "error"}. Gathering while any copy is held (at: when it sends); sending/retrying while any is still due (at: the next try); failed when any reader wasn't reached (error names each); else sent (at: the last acceptance).""" if held := [r for r in rows if r.state == "held"]: return {"state": "held", "at": min(r.send_after for r in held), "attempts": 0, "error": None} if due := [r for r in rows if r.state in ("pending", "sending")]: worst = max(due, key=lambda r: r.attempts) return {"state": "retrying" if worst.attempts else "sending", "at": min((r.next_attempt_at for r in due if r.next_attempt_at), default=None), "attempts": worst.attempts, "error": worst.last_error} if failed := [r for r in rows if r.state == "failed"]: return {"state": "failed", "at": max(r.first_attempt_at or r.created_at for r in failed), "attempts": max(r.attempts for r in failed), "error": "; ".join(f"{r.recipient}: {r.last_error}" for r in failed)} return {"state": "sent", "at": max((r.sent_at for r in rows if r.sent_at), default=None), "attempts": max(r.attempts for r in rows), "error": None} async def email_states(db, event_ids: list[str]) -> dict[str, dict]: """{event id: delivery(...)} for the alert events whose email went through the outbox; an event with none (no reader, or from before it) is absent.""" if not event_ids: return {} link = email_outbox_events.c by: dict[str, list] = {} for event_id, row in (await db.execute( select(link.event_id, EmailOutbox).join(EmailOutbox, EmailOutbox.id == link.outbox_id) .where(link.event_id.in_(event_ids)))).all(): by.setdefault(event_id, []).append(row) return {eid: delivery(rows) for eid, rows in by.items()} async def outgoing(db, now: datetime | None = None, limit: int = 20) -> dict: """Settings → System's "Outgoing email": every email of the last SHOWN_DAYS not yet sent (failed first, then retrying longest, then sending and gathering), how many were sent, and the problems for the page's verdict. {"rows": [EmailOutbox], "more": n, "sent": n, "problems": [str]}.""" from sqlalchemy import case, func now = now or datetime.now(timezone.utc) since = now - timedelta(days=SHOWN_DAYS) open_states = ("failed", "pending", "sending", "held") order = case({"failed": 0, "pending": 1, "sending": 2, "held": 3}, value=EmailOutbox.state) base = select(EmailOutbox).where(EmailOutbox.created_at >= since, EmailOutbox.state.in_(open_states)) rows = list((await db.execute( base.order_by(order, EmailOutbox.first_attempt_at.asc().nulls_last(), EmailOutbox.created_at.desc()).limit(limit))).scalars()) counts = dict((await db.execute( select(EmailOutbox.state, func.count()).where(EmailOutbox.created_at >= since) .group_by(EmailOutbox.state))).all()) slow = (await db.execute(select(func.count()).select_from(EmailOutbox).where( EmailOutbox.state.in_(("pending", "sending")), EmailOutbox.first_attempt_at < now - timedelta(seconds=SLOW_S)))).scalar_one() problems = [] if failed := counts.get("failed", 0): problems.append(f"{failed} email{'s' if failed != 1 else ''} failed in the last {SHOWN_DAYS} days") if slow: problems.append(f"{slow} email{'s' if slow != 1 else ''} still not accepted by the mail " f"server after {SLOW_S // 60} minutes of trying") shown = sum(counts.get(s, 0) for s in open_states) return {"rows": rows, "more": max(shown - len(rows), 0), "sent": counts.get("sent", 0), "problems": problems} async def alert_trouble(db, now: datetime | None = None) -> dict: """The dashboard Alerts widget's delivery line (#5510): alert emails of the last SHOWN_DAYS that failed, and those still retrying (tried, not yet accepted), and the line that says so ("" when neither). {"failed": n, "retrying": n, "said": str}.""" from sqlalchemy import func now = now or datetime.now(timezone.utc) rows = dict((await db.execute( select(EmailOutbox.state, func.count()) .where(EmailOutbox.kind == "alert", EmailOutbox.created_at >= now - timedelta(days=SHOWN_DAYS), or_(EmailOutbox.state == "failed", and_(EmailOutbox.state.in_(("pending", "sending")), EmailOutbox.attempts > 0))) .group_by(EmailOutbox.state))).all()) failed = rows.get("failed", 0) retrying = rows.get("pending", 0) + rows.get("sending", 0) said = [] if failed: said.append(f"{failed} alert email{'s' if failed != 1 else ''} failed in the last {SHOWN_DAYS} days") if retrying: said.append(f"{retrying} still retrying" if failed else f"{retrying} alert email{'s' if retrying != 1 else ''} still retrying") return {"failed": failed, "retrying": retrying, "said": "; ".join(said)}