diff --git a/steward/core/alerts.py b/steward/core/alerts.py index b2df348..078747e 100644 --- a/steward/core/alerts.py +++ b/steward/core/alerts.py @@ -212,28 +212,56 @@ def _spawn(coro) -> None: task.add_done_callback(_background.discard) +def _within(txn, boundary) -> bool: + """Whether `txn` (a savepoint, or None for the outer transaction) is + `boundary` or opened inside it.""" + while txn is not None: + if txn is boundary: + return True + txn = txn.parent + return False + + def _after_commit(session: AsyncSession, fn, *args) -> None: - """Run fn(app, *args) once the session's transaction commits, and never if - it rolls back: a notification or a playbook run must describe what was - actually stored.""" + """Run fn(app, *args) once the session's outer transaction commits, and + never if what added it rolls back: a notification or a playbook run must + describe what was actually stored. + + SQLAlchemy fires after_commit and after_rollback for a savepoint too + (#5807). Releasing one, as the host-agent ingest does after its metrics, + is not the commit: a job started then raced the real commit, and an email + queued for an event not yet visible broke its foreign key. A savepoint + rolling back drops only the jobs added inside it.""" if _app is None: return info = session.info + sync = session.sync_session if _AFTER_COMMIT not in info: info[_AFTER_COMMIT] = [] app = _app - def _committed(_sync_session) -> None: + def _committed(sync_session) -> None: + if sync_session.get_nested_transaction() is not None: + return # a savepoint released: its jobs wait for the outer commit jobs, info[_AFTER_COMMIT] = info[_AFTER_COMMIT], [] - for job_fn, job_args in jobs: + for _owner, job_fn, job_args in jobs: _spawn(job_fn(app, *job_args)) - def _rolled_back(_sync_session) -> None: - info[_AFTER_COMMIT] = [] + def _rolled_back(_sync_session, previous_transaction) -> None: + # What was actually rolled back: the nearest savepoint, or the + # outer transaction (a failed flush rolls back to one of them). + boundary = previous_transaction + while boundary.parent is not None and not boundary.nested: + boundary = boundary.parent + if not boundary.nested: + info[_AFTER_COMMIT] = [] + else: + info[_AFTER_COMMIT] = [job for job in info[_AFTER_COMMIT] + if not _within(job[0], boundary)] - sa_event.listen(session.sync_session, "after_commit", _committed) - sa_event.listen(session.sync_session, "after_rollback", _rolled_back) - info[_AFTER_COMMIT].append((fn, args)) + sa_event.listen(sync, "after_commit", _committed) + sa_event.listen(sync, "after_soft_rollback", _rolled_back) + info[_AFTER_COMMIT].append((sync.get_nested_transaction(), fn, args)) LATEST_CHUNK = 1000 diff --git a/tests/integration/test_alert_after_commit_savepoints.py b/tests/integration/test_alert_after_commit_savepoints.py new file mode 100644 index 0000000..3aa527d --- /dev/null +++ b/tests/integration/test_alert_after_commit_savepoints.py @@ -0,0 +1,126 @@ +"""Integration (#5807): against real Postgres, a savepoint is not the commit. + +SQLAlchemy fires after_commit and after_rollback for a savepoint too. The +host-agent ingest releases one after recording its metrics, so an alert's +email started then, before the outer commit, and twice on 2026-10-12 its +outbox row hit the event's foreign key before the event was visible. A +savepoint rolling back (a docker persist failing) cleared every pending +notification, the outer transaction's included.""" +from __future__ import annotations + +import asyncio +import os +import uuid +from unittest.mock import patch + +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 + return create_app(testing=False) + + +async def _rule(app, tag): + from steward.alerts import service as alert_svc + from steward.core.alerts import record_metrics + from steward.models.metrics import PluginMetric + from steward.models.users import User, UserRole + res = f"sp-{tag}" + async with app.db_sessionmaker() as s: + async with s.begin(): + user = User(username=f"sp-{tag}", email=f"sp-{tag}@test", password_hash="x", + role=UserRole.admin) + s.add(user) + await record_metrics(s, [PluginMetric(source_module="host_agent", resource_name=res, + metric_name="cpu_pct", value=1.0)]) + await s.flush() + fields = await alert_svc.rule_fields( + s, name=f"hot {tag}", source_module="host_agent", resource_name=res, + metric_name="cpu_pct", operator=">", threshold=90) + await alert_svc.add_rule(s, fields, created_by=user.id) + return res + + +async def _settle(): + for _ in range(5): + await asyncio.sleep(0) + + +class Boom(Exception): + pass + + +def _recorder(): + sent = [] + + async def fake_dispatch(app_, notif_type, rule, value, event_id, note=None): + sent.append((rule.resource_name, notif_type, value)) + + return sent, fake_dispatch + + +def test_a_released_savepoint_holds_the_email_until_the_outer_commit(app): + from steward.core import alerts + sent, fake_dispatch = _recorder() + + async def _go(): + res = await _rule(app, uuid.uuid4().hex[:8]) + async with app.db_sessionmaker() as s: + async with s.begin(): + await alerts.record_metric(s, "host_agent", res, "cpu_pct", 95.0) + async with s.begin_nested(): # as the ingest's maintenance step + pass + await _settle() + assert sent == [] # the event isn't committed yet + await _settle() + assert sent == [(res, "firing", 95.0)] + + with patch.object(alerts, "_dispatch_notification", fake_dispatch): + asyncio.run(_go()) + + +def test_a_savepoint_rolled_back_after_the_alert_keeps_its_email(app): + from steward.core import alerts + sent, fake_dispatch = _recorder() + + async def _go(): + res = await _rule(app, uuid.uuid4().hex[:8]) + async with app.db_sessionmaker() as s: + async with s.begin(): + await alerts.record_metric(s, "host_agent", res, "cpu_pct", 95.0) + with pytest.raises(Boom): + async with s.begin_nested(): # as a docker persist failing + raise Boom + await _settle() + assert sent == [(res, "firing", 95.0)] + + with patch.object(alerts, "_dispatch_notification", fake_dispatch): + asyncio.run(_go()) + + +def test_an_alert_inside_a_rolled_back_savepoint_is_dropped_alone(app): + from steward.core import alerts + sent, fake_dispatch = _recorder() + + async def _go(): + kept = await _rule(app, uuid.uuid4().hex[:8]) + undone = await _rule(app, uuid.uuid4().hex[:8]) + async with app.db_sessionmaker() as s: + async with s.begin(): + await alerts.record_metric(s, "host_agent", kept, "cpu_pct", 95.0) + with pytest.raises(Boom): + async with s.begin_nested(): + async with s.begin_nested(): # released into the one undone + await alerts.record_metric(s, "host_agent", undone, "cpu_pct", 97.0) + raise Boom + await _settle() + assert sent == [(kept, "firing", 95.0)] + + with patch.object(alerts, "_dispatch_notification", fake_dispatch): + asyncio.run(_go())