"""Durable two-phase fan-out for multi-recipient notifications. **The failure this exists to prevent** — ``notification_outbox. recipient_person_id`` is an FK onto ``identity_persons``, so every INSERT takes a shared record lock on the recipient's person row and holds it until the transaction commits. A fan-out that enqueues N rows and sends N external messages inside ONE transaction therefore holds N person-row locks across N slow HTTP round-trips; every concurrent ``UPDATE identity_persons`` — which includes the ``last_login_at`` write on the SSO path — queues behind them until ``innodb_lock_wait_timeout`` fires as 1205. That is the 2026-06 login brown-out (``reference_outbox_fk_lock_login_outage``), and it has been re-derived independently in five more places since. **The second failure**, less obvious and worse: a crash partway through rolls back the outbox rows for messages that were ALREADY SENT. The next tick sees no record of them and sends again — so a "once a year" or "once per stage" guarantee built on those rows silently breaks. Both are fixed by the same shape, first written out in ``AssetInventoryService._dispatch_initial_notifications`` and extracted here so the next fan-out inherits it instead of re-deriving it: * **Phase 1 — enqueue, commit per row.** Every intended message is durably recorded before any of them is sent. Each person-row lock is held for sub-millisecond, never across an HTTP call. A crash after this phase leaves un-sent rows PENDING for the ``notification.outbox_retry`` worker, so delivery is still guaranteed. * **Phase 2 — deliver, commit per row.** A send failure just leaves that row PENDING for the same worker; it never rolls back a sibling that already went out. The ``next_attempt_at`` fence set in phase 1 keeps the hourly retry worker from claiming rows this drain is about to send (its picker sorts ``next_attempt_at IS NULL`` first) — without it a retry tick landing inside a multi-minute drain double-sends. Phase 2 walks its own explicit list and ignores the fence, but it **renews the fence and re-reads the row** before each send (#2240): a single fence set once in phase 1 is outlived by any drain longer than 30 minutes (500 recipients × 2 channels at ~2s per SMTP hop already is), and past that point the worker can claim a tail row while this drain still holds a stale pre-fetched ORM instance whose ``delivered_channels`` is empty — same card, twice. **The caller must not hold a lock.** This helper commits, so it has to run on a session that is not inside a ``FOR UPDATE`` window or a still-open unit of work the caller intends to roll back. """ from __future__ import annotations import logging from collections.abc import Awaitable, Callable from datetime import UTC, datetime, timedelta from typing import TYPE_CHECKING, Any from uuid import UUID from app.notifications.models import ( STATUS_DEAD, STATUS_DELIVERED, NotificationOutbox, ) if TYPE_CHECKING: # pragma: no cover from sqlalchemy.ext.asyncio import AsyncSession from app.notifications.service import NotificationEnvelope logger = logging.getLogger(__name__) # How far into the future a freshly enqueued row is pushed so the hourly # retry worker cannot claim it while phase 2 is still draining. Long # enough to outlast any realistic drain, short enough that a process # killed mid-drain still has its rows retried the same day. OUTBOX_RETRY_FENCE = timedelta(minutes=30) #: ``(done, total)`` → awaited after each delivered/failed row. See the #: ``on_progress`` paragraph in :func:`enqueue_then_drain`. ProgressHook = Callable[[int, int], Awaitable[None]] #: ``(envelope, outbox_id)`` → awaited after a row is enqueued but **before** #: that row's commit, so whatever the hook writes lands in the SAME #: transaction as the outbox row. See the ``on_enqueued`` paragraph in #: :func:`enqueue_then_drain`. EnqueuedHook = Callable[["NotificationEnvelope", UUID], Awaitable[None]] async def _revive_recipients( db: AsyncSession, ids: list[Any], label: str, ) -> None: """Re-populate the recipient rows a ``rollback()`` just expired. One bulk ``SELECT ... WHERE id IN (…)`` with ``populate_existing=True`` refreshes the identity map in place, so the ORM instances the remaining envelopes hold become usable again. Only runs on the error path. savepoint-exempt: read-only, and a failure here is logged and swallowed — it degrades to the pre-existing behaviour (the remaining rows fail and are reported as failures) rather than replacing one bad row with an exception that escapes the helper. """ from sqlalchemy import select as _select from app.identity.person_models import IdentityPerson wanted = {rid for rid in ids if rid is not None} if not wanted: return # savepoint-exempt: 只读 SELECT,失败即记日志退回原行为(剩余行照旧失败并被 # 如实计入 failed),不会让异常从这个 helper 逃出去。 try: await db.execute( _select(IdentityPerson) .where(IdentityPerson.id.in_(list(wanted))) .execution_options(populate_existing=True), ) except Exception: # noqa: BLE001 logger.exception( "%s: could not revive %d recipient row(s) after a rollback", label, len(wanted), ) async def _emit_progress( hook: ProgressHook | None, done: int, total: int, label: str, ) -> None: """Best-effort progress notification. savepoint-exempt: performs no writes of its own — the hook owns whatever it touches, and a hook that blows up is swallowed precisely so a bookkeeping error cannot abort a drain that IS delivering. The hook is responsible for leaving the session usable (ours commits per row, so a hook that rolls back only loses its own bookkeeping write). """ if hook is None: return try: await hook(done, total) except Exception: # noqa: BLE001 logger.exception("%s: progress hook failed at %d/%d", label, done, total) async def enqueue_then_drain( db: AsyncSession, envelopes: list[NotificationEnvelope], *, trace_id: str = "", actor_user_id: UUID | None = None, label: str = "fanout", on_progress: ProgressHook | None = None, on_enqueued: EnqueuedHook | None = None, ) -> dict[str, int]: """Enqueue every envelope (committing each), then deliver each. Returns ``{"enqueued": n, "delivered": n, "failed": n}``, where ``enqueued + failed == len(envelopes)`` always holds — ``failed`` counts BOTH a row that could not be written and a row that was written but could not be delivered. It used to count only the latter, so a fan-out that lost recipients in phase 1 returned ``{"enqueued": 3, "delivered": 3, "failed": 0}`` for four intended people: honest only to a caller who thought to compare ``enqueued`` against the length of the list it passed in, which is exactly the subtraction ``me_universe.wrapped_push`` had open-coded and every other caller had not. Never raises for a single recipient's failure — one bad address must not cost the other 499 recipients their message. ``on_enqueued`` (optional) is awaited with ``(envelope, outbox_id)`` after that row is enqueued and **before** its commit, so a caller's own per-recipient bookkeeping lands in the SAME transaction as the outbox row. That atomicity is the point: a ledger stamp written in a separate transaction can exist without its outbox row (or vice versa), and a caller that uses its stamp to answer "who still needs enqueueing" would then either re-enqueue a row that already went out or skip one that never did. A hook that raises fails **that one** recipient (rolled back, logged, counted) and the drain continues — same contract as an enqueue failure, because that is exactly what it is. The hook must not commit; it may only add writes to the open transaction. ``on_progress`` (optional) is awaited after each phase-2 row, with ``(done, total)``. It exists so a caller that tracks its own batch row can emit a liveness heartbeat **without** slicing the fan-out into chunks: chunking would move phase 1 inside the loop, and phase 1 being whole-batch is the entire reason a process killed mid-drain still has every remaining row sitting PENDING for the retry worker. A hook that raises is logged and ignored — a bookkeeping failure must not abort a drain that is successfully delivering. """ from app.notifications.service import NotificationService if not envelopes: return {"enqueued": 0, "delivered": 0, "failed": 0} notif = NotificationService(db) # ``(outbox_id, envelope, recipient_id)`` —— **id 是快照**,不是从实例上读的。 # phase 1 的 rollback 会 expire 掉这里已经攒下的 outbox 实例,之后读 # ``outbox.id`` 就是一次 lazy load ⇒ 异步环境下 ``MissingGreenlet``,而它会 # 从 phase 2 一路抛出这个 helper —— 违背「never raises for a single # recipient's failure」。存 id 就完全绕开了实例状态。 pending: list[tuple[UUID, NotificationEnvelope, Any]] = [] # Snapshot the recipient ids while the instances are still FRESH. A # ``rollback()`` in phase 1 expires the whole identity map, and after that # even reading ``env.recipient.id`` is a lazy load — i.e. the revive helper # could not so much as ask "which rows do I need to reload". recipient_ids: list[Any] = [ getattr(env.recipient, "id", None) for env in envelopes ] # Counted across BOTH phases — see the return-value paragraph above. failed = 0 # Phase 1 — durable record first. for i, env in enumerate(envelopes): try: outbox = await notif.enqueue(env) outbox.next_attempt_at = datetime.now(UTC) + OUTBOX_RETRY_FENCE # Read the id BEFORE the commit and before the hook: the hook # may leave the session needing a rollback, after which # ``outbox.id`` is a lazy load ⇒ ``MissingGreenlet``. outbox_id = outbox.id if on_enqueued is not None: # Deliberately inside the try and BEFORE the commit — the # caller's stamp and the outbox row must land together or # not at all. See the ``on_enqueued`` paragraph above. await on_enqueued(env, outbox_id) await db.commit() except Exception: # noqa: BLE001 # Roll back so the session stays usable for the next # recipient; a missed enqueue costs exactly one message and # is visible in the returned counts. recipient_id = recipient_ids[i] failed += 1 await db.rollback() logger.exception( "%s: failed to enqueue notification for %s", label, recipient_id, ) # ``rollback()`` expires EVERY instance in the identity map — which # includes the ``recipient`` ORM objects the envelopes we have not # processed yet are still holding. Touching an expired attribute # from async code raises ``MissingGreenlet``, so without this # revive ONE failed enqueue cascades into every remaining row # failing too: precisely the "one bad address must not cost the # other 499 recipients their message" outcome this helper exists # to prevent. (Also note the recipient id is read BEFORE the # rollback above — reading it after would be the same trap.) # 复活**全部**收件人,不只是后面那些:同一个 rollback 也 expire 了 # 已经攒进 ``pending`` 的那些 envelope 的 recipient,而 phase 2 的 # 错误分支会读 ``env.recipient.id`` —— 只复活尾部时,「第 3 行失败」 # 会让异常从 phase 2 抛出这个 helper(实测)。 await _revive_recipients(db, recipient_ids, label) continue pending.append((outbox_id, env, recipient_ids[i])) # Phase 2 — deliver; a failure leaves the row PENDING for the worker. delivered = 0 for outbox_id, env, recipient_id in pending: try: # #2240: re-claim the row immediately before delivering it. # The 30-minute fence is set ONCE, in phase 1, and never # renewed — a 500-recipient × 2-channel drain at ~2s per SMTP # hop is already past it, and any Feishu rate-limit backoff # pushes it further. Once it lapses the hourly retry worker # can claim a tail row while we still hold a PRE-FETCHED ORM # instance of it (``expire_on_commit=False``), whose # ``delivered_channels`` is stale-empty — so ``deliver`` # recomputes ``todo`` from stale state and sends the SAME card # again. Pushing the fence out per row (and re-reading what # the worker may have already delivered) closes both halves. fresh = await db.get( NotificationOutbox, outbox_id, populate_existing=True, ) if fresh is None: continue if fresh.status == STATUS_DELIVERED: # The retry worker got there first — nothing owed. (A # PARTIAL row is NOT skipped: ``deliver`` recomputes # ``todo`` from the FRESH ``delivered_channels``, so it # sends only the channels still outstanding — which is # the whole reason for re-reading rather than trusting # the pre-fetched copy.) delivered += 1 continue if fresh.status == STATUS_DEAD: failed += 1 continue fresh.next_attempt_at = datetime.now(UTC) + OUTBOX_RETRY_FENCE await db.commit() await notif.deliver( fresh, env, trace_id=trace_id, actor_user_id=actor_user_id, ) delivered += 1 await db.commit() await _emit_progress(on_progress, delivered + failed, len(pending), label) except Exception: # noqa: BLE001 # #2241: the commit used to sit AFTER this handler, outside # the try. ``deliver`` deliberately lets a 1213 deadlock # escape its savepoint, which leaves the session in a # pending-rollback state — the unconditional commit then # raised PendingRollbackError straight out of a primitive # whose docstring promises "one bad address must not cost # the other 499 recipients their message". The remaining # recipients were recoverable (the fence expires and the # retry worker picks them up 30 minutes later), but the # drain aborted and the caller — e.g. # ``run_wrapped_year_end_push`` — recorded a failed run for # one row's deadlock. failed += 1 logger.exception( "%s: failed to deliver notification to %s", label, recipient_id, ) try: await db.rollback() except Exception: # noqa: BLE001 logger.exception("%s: rollback after delivery failure failed", label) await _emit_progress(on_progress, delivered + failed, len(pending), label) return { "enqueued": len(pending), "delivered": delivered, "failed": failed, }