From 79d41f9ee7eae95c3880373eeaa8cff68717e553 Mon Sep 17 00:00:00 2001 From: HardMax71 Date: Mon, 19 Oct 2026 01:57:56 +0200 Subject: [PATCH] fix: a replay with short waits between events no longer stalls after the first one _dispatch_next sent one event and then re-added its own APScheduler job for the next. When the wait was zero or shorter than what was left of the running job, the new run came due while that job still counted as running, so APScheduler skipped it under max_instances=1 and nothing scheduled another: the session sat in running with one event sent. One job now loops over the events and awaits each wait itself. Pause and cancel stop it through the status check, as before. --- .../services/event_replay/replay_service.py | 99 +++++++++---------- .../unit/services/test_replay_dispatch.py | 51 ++++++++++ .../services/test_replay_session_metrics.py | 7 +- 3 files changed, 99 insertions(+), 58 deletions(-) create mode 100644 backend/tests/unit/services/test_replay_dispatch.py 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"]