From 684e48e90131580464f0d492443563711888cc37 Mon Sep 17 00:00:00 2001 From: Renn F Date: Sun, 28 Jun 2026 22:06:56 +0200 Subject: [PATCH] [F107] defer Redis bus publish until DB commit (no phantom notifications) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit deliver() and _persist_and_deliver() ran inside the caller's open transaction: the notification row was flushed but not committed, yet NOTIFICATION_SENT was published to the Redis bus immediately. A commit failure (DB hiccup, constraint, asyncpg error) rolled the row back while connected WebSocket clients had already received a push for an id that no longer existed — a phantom notification (notify_get -> NotFoundError). Added a deferred-publish (transactional-outbox) helper: defer_bus_publish enqueues the event on session.info and registers one-shot after_commit / after_rollback listeners on session.sync_session the first time it is called for that session. On commit, the after_commit listener schedules the async drain via asyncio.create_task on the running loop (the listener fires synchronously inside await AsyncSession.commit, so the loop is active); the task handles are stashed on the session so callers/tests can await them. On rollback, after_rollback drops the pending queue — a rolled-back txn emits nothing. deliver() now builds the per-recipient events up front (data materialized to strings, so deferral is safe even if the ORM object later expires) and defers each; the delivered_at DB marker stays in-tx (rolls back with the row). The bus block stays best-effort (try/except + log) so a bus-init failure never propagates or rolls back the notification row — matching the prior inline semantics. This fixes every deliver/_persist_and_deliver caller at once (the two cited in F107 plus the orchestrator + task.py deliver sites), since they all commit the session afterward (the deferred publish fires on that commit; the row is durable by the time the event goes out). --- roboco/services/notification_delivery.py | 126 ++++++++++- .../test_notification_delivery_phantom.py | 195 ++++++++++++++++++ 2 files changed, 314 insertions(+), 7 deletions(-) create mode 100644 tests/integration/test_notification_delivery_phantom.py diff --git a/roboco/services/notification_delivery.py b/roboco/services/notification_delivery.py index 711f482c..32c14ad3 100644 --- a/roboco/services/notification_delivery.py +++ b/roboco/services/notification_delivery.py @@ -9,12 +9,14 @@ Handles delivery of notifications to agents through multiple channels: Also implements the ACK system for tracking acknowledgments. """ +import asyncio from dataclasses import dataclass from datetime import UTC, datetime from typing import ClassVar, Literal from uuid import UUID -from sqlalchemy import and_, select +import structlog +from sqlalchemy import and_, event, select from sqlalchemy.ext.asyncio import AsyncSession from roboco.agents_config import ( @@ -29,6 +31,101 @@ from roboco.models.base import AgentRole, NotificationPriority, NotificationType from roboco.services.base import BaseService, NotFoundError from roboco.utils.converters import require_uuid +_log = structlog.get_logger(service="notification_delivery") + + +# ============================================================================= +# Deferred bus publish — transactional outbox (F107) +# ============================================================================= +# `deliver`/`_persist_and_deliver` run inside the caller's open transaction: +# the notification row is flushed but not committed. Publishing the +# NOTIFICATION_SENT event to the Redis bus *before* the commit created +# phantom notifications — a commit failure (DB hiccup, constraint, asyncpg +# error) rolled the row back while connected WebSocket clients had already +# received a push for an id that no longer existed. The fix defers the bus +# publish to the session's `after_commit` so a rollback drops the pending +# event: the row is durable by the time the event fires. +# +# The pending events and the scheduled drain tasks live on `session.info` so +# they are scoped to the session's lifetime (no module-global state, no +# cross-request leak). A sync `after_commit` listener schedules the async +# drain via `asyncio.create_task` (the listener runs synchronously inside +# `await AsyncSession.commit()` on the loop thread, so the running loop is +# available); `after_rollback` clears the pending queue so a rolled-back +# transaction emits nothing. + +_PENDING_PUBLISHES_KEY = "_roboco_pending_bus_publishes" +_DRAIN_TASKS_KEY = "_roboco_drain_tasks" +_DRAIN_REGISTERED_KEY = "_roboco_drain_registered" + + +async def _drain_pending_publishes(pending: list[Event]) -> None: + """Publish every deferred event best-effort once the txn has committed. + + The bus is read fresh at drain time (it may have reconnected between + deferral and commit); a disconnected bus is a silent no-op, matching the + prior inline behavior. Each publish is independent — one failure does + not drop the rest. + """ + if not pending: + return + bus = get_event_bus() + if not bus.is_connected(): + return + for ev in pending: + try: + await bus.publish(ev) + except Exception as e: # best-effort: never break the drain + _log.warning("Deferred bus publish failed", error=str(e)) + + +def _schedule_pending_publishes(session: AsyncSession) -> None: + """`after_commit` handler: hand the pending events to the running loop. + + Sync listener — runs inside `await AsyncSession.commit()`, so the event + loop is active. The created task is stashed on the session so callers / + tests can await it deterministically; in production it is fire-and-forget + (best-effort, matching the prior try/except semantics). + """ + pending = session.info.pop(_PENDING_PUBLISHES_KEY, None) + if not pending: + return + try: + loop = asyncio.get_running_loop() + except RuntimeError: # no running loop — nothing we can do, drop silently + return + task = loop.create_task(_drain_pending_publishes(pending)) + session.info.setdefault(_DRAIN_TASKS_KEY, []).append(task) + + +def _discard_pending_publishes(session: AsyncSession) -> None: + """`after_rollback` handler: a rolled-back txn emits nothing (no phantom).""" + session.info.pop(_PENDING_PUBLISHES_KEY, None) + + +def defer_bus_publish(session: AsyncSession, ev: Event) -> None: + """Enqueue a bus event to fire only after the session's transaction commits. + + Registers one-shot `after_commit` / `after_rollback` listeners on the + session the first time it is called for that session; subsequent calls + just append. The listeners are bound to the session instance and are + collected with it (no global listener accumulation). + """ + session.info.setdefault(_PENDING_PUBLISHES_KEY, []).append(ev) + if session.info.get(_DRAIN_REGISTERED_KEY): + return + session.info[_DRAIN_REGISTERED_KEY] = True + + sync_session = session.sync_session + + @event.listens_for(sync_session, "after_commit") + def _on_commit(_sync_session: object) -> None: + _schedule_pending_publishes(session) + + @event.listens_for(sync_session, "after_rollback") + def _on_rollback(_sync_session: object) -> None: + _discard_pending_publishes(session) + class EscalationError(ValueError): """Raised when an escalation can't be routed (missing chain, bad override).""" @@ -87,6 +184,13 @@ class NotificationDeliveryService(BaseService): 2. Redis pub/sub - for polling agents 3. Database - persistent storage (always) + The Redis/bus publish is deferred until the caller's transaction + commits (`defer_bus_publish`) — publishing before the commit produced + phantom notifications when the commit failed (F107). The `delivered_at` + DB marker is written inside the transaction (it rolls back with the + row if the commit fails), and the bus event fires only once the row is + durable. + Returns True if at least one delivery channel succeeded. """ notification = await self.get_notification(notification_id) @@ -96,11 +200,18 @@ class NotificationDeliveryService(BaseService): ) return False - # Mark delivery attempted + # Mark delivery attempted (in-tx — rolls back with the row on + # commit failure, so the marker and the row stay consistent). notification.delivered_at = datetime.now(UTC) await self.session.flush() - # Publish to Redis for real-time delivery + # Build the per-recipient bus events up front (the data is materialized + # to strings, so deferring is safe even if the ORM object later + # expires) and defer each to the session's after_commit. The event is + # dropped on rollback (no phantom) and fired once the row is durable. + # Best-effort: a bus-init failure is logged but never propagates — the + # notification row + delivered_at marker are already flushed, and the + # bus is a secondary delivery channel (the row is the durable store). try: bus = get_event_bus() if bus.is_connected(): @@ -113,7 +224,8 @@ class NotificationDeliveryService(BaseService): return v.value if hasattr(v, "value") else v for recipient_id in notification.to_agents: - await bus.publish( + defer_bus_publish( + self.session, Event( type=EventType.NOTIFICATION_SENT, data={ @@ -123,16 +235,16 @@ class NotificationDeliveryService(BaseService): "priority": _enum_value(notification.priority), "subject": notification.subject, }, - ) + ), ) self.log.info( - "Notification published to Redis", + "Notification bus publish deferred until commit", notification_id=str(notification_id), recipient_count=len(notification.to_agents), ) except Exception as e: self.log.warning( - "Failed to publish notification to Redis", + "Failed to defer notification bus publish", notification_id=str(notification_id), error=str(e), ) diff --git a/tests/integration/test_notification_delivery_phantom.py b/tests/integration/test_notification_delivery_phantom.py new file mode 100644 index 00000000..e1cc6353 --- /dev/null +++ b/tests/integration/test_notification_delivery_phantom.py @@ -0,0 +1,195 @@ +"""F107 — Redis bus publish must be deferred until the DB commit lands. + +`NotificationDeliveryService.deliver` historically published +``NOTIFICATION_SENT`` to the Redis event bus *before* the caller committed +the notification row. A commit failure (DB hiccup, constraint, asyncpg error) +rolled the row back but left the bus event behind — connected WebSocket +clients received a push for an id that no longer existed (a phantom +notification). The fix defers the bus publish to the session's +``after_commit`` so a rollback drops it; the row is durable by the time the +event fires. + +These tests need a real ``AsyncSession`` (the deferral uses SQLAlchemy +session commit/rollback events) plus a recording bus stand-in, so they are +integration tests against the migrated Postgres test DB. +""" + +from __future__ import annotations + +import asyncio +from typing import TYPE_CHECKING +from uuid import UUID, uuid4 + +import pytest +from roboco.db.tables import AgentTable, NotificationTable +from roboco.events import Event, EventType +from roboco.models import AgentRole, AgentStatus, NotificationPriority, NotificationType +from roboco.models.base import Team +from roboco.services.notification_delivery import get_notification_delivery_service + +if TYPE_CHECKING: + from sqlalchemy.ext.asyncio import AsyncSession + + +class _RecordingBus: + """Stand-in for StreamEventBus that records every published event. + + Mirrors the real bus surface used by ``deliver``: ``is_connected()`` + gates the publish path and ``publish`` is async. Recording lets the + tests assert exactly when (and whether) the NOTIFICATION_SENT event + fired — without a Redis stack. + """ + + def __init__(self) -> None: + self.published: list[Event] = [] + + def is_connected(self) -> bool: + return True + + async def publish(self, event: Event) -> str: + self.published.append(event) + return "recorded" + + +def _drain_tasks(session: AsyncSession) -> list[asyncio.Task[object]]: + """Pending deferred-publish drain tasks stashed on the session. + + The deferral helper stores the ``asyncio.create_task`` handles here so a + test can await them deterministically instead of racing the event loop. + """ + return list(session.info.get("_roboco_drain_tasks", [])) + + +async def _await_drain(session: AsyncSession) -> None: + """Wait for any scheduled deferred-publish tasks to finish.""" + tasks = _drain_tasks(session) + if tasks: + await asyncio.gather(*tasks, return_exceptions=True) + + +async def _seed_agents_and_notification( + db: AsyncSession, *, recipients: int +) -> tuple[UUID, NotificationTable]: + """Create a sender + N recipient agents and one flushed (uncommitted) + notification addressed to them. Returns ``(notification_id, row)``. + + Flushed only — the row lives in the session's open transaction, matching + the real pre-commit state ``deliver`` runs against. + """ + sender = AgentTable( + id=uuid4(), + name="Sender", + slug=f"sender-{uuid4().hex[:8]}", + role=AgentRole.DEVELOPER, + team=Team.BACKEND, + status=AgentStatus.ACTIVE, + model_config={}, + system_prompt="sender", + capabilities=[], + permissions={}, + metrics={}, + ) + db.add(sender) + await db.flush() + + recipient_ids: list[UUID] = [] + for i in range(recipients): + r = AgentTable( + id=uuid4(), + name=f"Recipient {i}", + slug=f"recipient-{i}-{uuid4().hex[:8]}", + role=AgentRole.QA, + team=Team.BACKEND, + status=AgentStatus.ACTIVE, + model_config={}, + system_prompt="recipient", + capabilities=[], + permissions={}, + metrics={}, + ) + db.add(r) + recipient_ids.append(r.id) + await db.flush() + + notification = NotificationTable( + type=NotificationType.REVIEW_REQUEST, + priority=NotificationPriority.NORMAL, + from_agent=sender.id, + to_agents=recipient_ids, + subject="Please review", + body="Body text", + requires_ack=True, + ) + db.add(notification) + await db.flush() + return notification.id, notification + + +@pytest.mark.asyncio +async def test_deliver_does_not_publish_before_commit( + db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch +) -> None: + """The bus event must NOT fire until the session commits (F107). + + Currently RED: ``deliver`` publishes immediately, so the bus is non-empty + before any commit — the phantom window. With the deferred-publish fix, + ``deliver`` only schedules; the event fires on commit. + """ + bus = _RecordingBus() + monkeypatch.setattr( + "roboco.services.notification_delivery.get_event_bus", lambda: bus + ) + + notif_id, _ = await _seed_agents_and_notification(db_session, recipients=2) + service = get_notification_delivery_service(db_session) + await service.deliver(notif_id) + + # Pre-commit: nothing published yet (the row is not durable). + assert bus.published == [] + + +@pytest.mark.asyncio +async def test_deliver_publishes_after_commit( + db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch +) -> None: + """Commit drains the deferred publish — one event per recipient (F107).""" + bus = _RecordingBus() + monkeypatch.setattr( + "roboco.services.notification_delivery.get_event_bus", lambda: bus + ) + + recipient_count = 2 + notif_id, _ = await _seed_agents_and_notification( + db_session, recipients=recipient_count + ) + service = get_notification_delivery_service(db_session) + await service.deliver(notif_id) + assert bus.published == [] # still nothing before commit + + await db_session.commit() + await _await_drain(db_session) + + assert len(bus.published) == recipient_count + assert all(ev.type == EventType.NOTIFICATION_SENT for ev in bus.published) + assert all(ev.data["notification_id"] == str(notif_id) for ev in bus.published) + + +@pytest.mark.asyncio +async def test_deliver_rollback_drops_phantom( + db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch +) -> None: + """A rollback instead of commit drops the pending publish — no phantom + event for a row that never became durable (F107).""" + bus = _RecordingBus() + monkeypatch.setattr( + "roboco.services.notification_delivery.get_event_bus", lambda: bus + ) + + notif_id, _ = await _seed_agents_and_notification(db_session, recipients=1) + service = get_notification_delivery_service(db_session) + await service.deliver(notif_id) + + await db_session.rollback() + await _await_drain(db_session) + + assert bus.published == []