diff --git a/backend/app/services/event_replay/replay_service.py b/backend/app/services/event_replay/replay_service.py index 167a98f2..30d14695 100644 --- a/backend/app/services/event_replay/replay_service.py +++ b/backend/app/services/event_replay/replay_service.py @@ -97,7 +97,7 @@ async def start_session(self, session_id: str) -> ReplayOperationResult: self._schedulers[session_id] = scheduler scheduler.start() scheduler.add_job( - self._dispatch_next, + self._dispatch_events, trigger="date", run_date=datetime.now(timezone.utc), args=[session], @@ -143,7 +143,7 @@ async def resume_session(self, session_id: str) -> ReplayOperationResult: scheduler = self._schedulers.get(session_id) if scheduler: scheduler.add_job( - self._dispatch_next, + self._dispatch_events, trigger="date", run_date=datetime.now(timezone.utc), args=[session], @@ -219,63 +219,54 @@ def _report_event( attributes["replay.delay_seconds"] = delay self._tracer.start_span("replay.event", attributes=attributes).end() - async def _dispatch_next(self, session: ReplaySessionState) -> None: - if session.status != ReplayStatus.RUNNING: - return + async def _dispatch_events(self, session: ReplaySessionState) -> None: + """Replay the session's events in order, waiting out each delay, until it stops running or runs out. - event = self._pop_next_event(session.session_id) - if event is None: - if not await self._load_next_batch(session.session_id): - await self._finalize_session(session, ReplayStatus.COMPLETED) - return + One job runs the whole loop, since APScheduler skips a run of a job whose previous run hasn't returned. + """ + while session.status == ReplayStatus.RUNNING: event = self._pop_next_event(session.session_id) if event is None: - await self._finalize_session(session, ReplayStatus.COMPLETED) - return - - success = False - t0 = time.monotonic() - try: - success = await self._replay_event(session, event) - except Exception as e: - processing_time = time.monotonic() - t0 - session.errors.append( - ReplayError(timestamp=datetime.now(timezone.utc), event_id=str(event.event_id), error=str(e)) - ) - if not session.config.skip_errors: + if not await self._load_next_batch(session.session_id): + await self._finalize_session(session, ReplayStatus.COMPLETED) + return + event = self._pop_next_event(session.session_id) + if event is None: + await self._finalize_session(session, ReplayStatus.COMPLETED) + return + + success = False + t0 = time.monotonic() + try: + success = await self._replay_event(session, event) + except Exception as e: + processing_time = time.monotonic() - t0 + session.errors.append( + ReplayError(timestamp=datetime.now(timezone.utc), event_id=str(event.event_id), error=str(e)) + ) + if not session.config.skip_errors: + session.failed_events += 1 + self._report_event(session, event, False, processing_time, 0.0) + await self._finalize_session(session, ReplayStatus.FAILED) + return + else: + processing_time = time.monotonic() - t0 + + if success: + session.replayed_events += 1 + else: session.failed_events += 1 - self._report_event(session, event, False, processing_time, 0.0) - await self._finalize_session(session, ReplayStatus.FAILED) - return - else: - processing_time = time.monotonic() - t0 - - if success: - session.replayed_events += 1 - else: - session.failed_events += 1 - session.last_event_at = event.timestamp - await self._update_session_in_db(session) - - next_event = self._peek_next_event(session.session_id) - delay = 0.0 - if next_event and session.last_event_at and session.config.speed_multiplier < 100: - time_diff = (next_event.timestamp - session.last_event_at).total_seconds() - delay = max(time_diff / session.config.speed_multiplier, 0) + session.last_event_at = event.timestamp + await self._update_session_in_db(session) - self._report_event(session, event, success, processing_time, delay) + next_event = self._peek_next_event(session.session_id) + delay = 0.0 + if next_event and session.last_event_at and session.config.speed_multiplier < 100: + time_diff = (next_event.timestamp - session.last_event_at).total_seconds() + delay = max(time_diff / session.config.speed_multiplier, 0) - scheduler = self._schedulers.get(session.session_id) - if scheduler and scheduler.running and session.status == ReplayStatus.RUNNING: - scheduler.add_job( - self._dispatch_next, - trigger="date", - run_date=datetime.now(timezone.utc) + timedelta(seconds=delay), - args=[session], - id=f"dispatch_{session.session_id}", - replace_existing=True, - misfire_grace_time=None, - ) + self._report_event(session, event, success, processing_time, delay) + await asyncio.sleep(delay) def _pop_next_event(self, session_id: str) -> DomainEvent | None: idx = self._buffer_indices.get(session_id, 0) @@ -394,7 +385,7 @@ async def _replay_event(self, session: ReplaySessionState, event: DomainEvent) - except _DISPATCH_ERRORS: return False finally: - # Unexpected errors still propagate to _dispatch_next, but count against the target first + # Unexpected errors still propagate to _dispatch_events, but count against the target first attributes: dict[str, AttributeValue] = {"replay.target": config.target, "replay.target_success": success} self._tracer.start_span("replay.target", attributes=attributes).end() return True diff --git a/backend/tests/unit/services/test_replay_dispatch.py b/backend/tests/unit/services/test_replay_dispatch.py new file mode 100644 index 00000000..9a869cc5 --- /dev/null +++ b/backend/tests/unit/services/test_replay_dispatch.py @@ -0,0 +1,51 @@ +import asyncio +from collections.abc import AsyncIterator +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import pytest +import structlog +from opentelemetry import trace + +from app.db import ReplayRepository +from app.domain.enums import ReplayStatus, ReplayTarget, ReplayType +from app.domain.replay import ReplayConfig +from app.events import UnifiedProducer +from app.services.event_replay import EventReplayService +from app.services.sse import SSERedisBus +from tests.conftest import make_execution_requested_event + +pytestmark = pytest.mark.unit + +_DOCS = [make_execution_requested_event(execution_id=f"e{n}").model_dump() for n in range(25)] + + +async def _batches(*_: Any, batch_size: int, **__: Any) -> AsyncIterator[list[dict[str, Any]]]: + for start in range(0, len(_DOCS), batch_size): + yield _DOCS[start : start + batch_size] + + +@pytest.mark.parametrize("speed_multiplier", [100.0, 99.0]) +async def test_a_replay_without_waits_sends_every_event(speed_multiplier: float) -> None: + repo = AsyncMock(spec=ReplayRepository) + repo.count_events.return_value = len(_DOCS) + repo.fetch_events = MagicMock(side_effect=_batches) + service = EventReplayService( + repository=repo, + producer=AsyncMock(spec=UnifiedProducer), + tracer=trace.NoOpTracer(), + logger=structlog.get_logger("test.replay_dispatch"), + sse_bus=AsyncMock(spec=SSERedisBus), + ) + config = ReplayConfig( + replay_type=ReplayType.QUERY, target=ReplayTarget.TEST, speed_multiplier=speed_multiplier, batch_size=10 + ) + created = await service.create_session_from_config(config) + + await service.start_session(created.session_id) + async with asyncio.timeout(5): + while not service.get_session(created.session_id).status.is_terminal: + await asyncio.sleep(0.01) + + session = service.get_session(created.session_id) + assert (session.status, session.replayed_events, session.failed_events) == (ReplayStatus.COMPLETED, 25, 0) diff --git a/backend/tests/unit/services/test_replay_session_metrics.py b/backend/tests/unit/services/test_replay_session_metrics.py index 6ef85824..cef6ad52 100644 --- a/backend/tests/unit/services/test_replay_session_metrics.py +++ b/backend/tests/unit/services/test_replay_session_metrics.py @@ -129,17 +129,16 @@ async def test_dispatch_reports_each_event_and_the_completed_session( started_at=datetime.now(timezone.utc) - timedelta(seconds=4), ) first, second = make_execution_requested_event(), make_execution_requested_event() - second.timestamp = first.timestamp + timedelta(seconds=2) + second.timestamp = first.timestamp + timedelta(milliseconds=40) service._sessions[session.session_id] = session service._event_buffers[session.session_id] = [first, second] - for _ in range(3): - await service._dispatch_next(session) + await service._dispatch_events(session) spans = _spans(span_exporter) events = [attributes for name, attributes in spans if name == "replay.event"] assert [(e["replay.result"], e.get("replay.delay_seconds")) for e in events] == [ - ("replayed", 1.0), + ("replayed", pytest.approx(0.02)), ("replayed", None), ] [completed] = [attributes for name, attributes in spans if name == "replay.session_completed"]