bvandeusenandClaude Opus 5.5
a68990db70
SQLAlchemy fires after_commit and after_rollback for a savepoint too. The host-agent ingest releases one after recording its metrics, so an alert raised by a sample started its email before the outer transaction committed; twice on 2026-10-12 the outbox row hit the event's foreign key first and the email was lost. A savepoint rolling back (a docker persist failing) cleared every pending notification, the outer one's included. _after_commit now runs its jobs only on the outermost commit, and a rolled-back savepoint drops only the jobs added inside it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
127 lines
4.4 KiB
Python
127 lines
4.4 KiB
Python
"""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())
|