fix: a replay with short waits between events no longer stalls after the first one

backend/app/services/event_replay/replay_service.py

@@ -97,7 +97,7 @@
         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 @@
         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 @@
             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 @@
         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

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)

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"]