Joseph Doherty
0c6e5b3ace
The gateway emits a ReplayGap sentinel MxEvent at the head of a StreamEvents stream resumed via after_worker_sequence when the requested cursor predates the oldest retained event. Clients previously ignored it, silently mis-treating a lossy resume as continuous. Each client now surfaces the sentinel as a distinct, typed, non-terminal signal (never synthesized, never swallowed) so a consumer can detect the gap and re-snapshot; resume contract is after_worker_sequence = oldest_available_sequence - 1. - .NET: MxEventStreamItem (IsReplayGap/ReplayGap/Event) via new StreamEventItemsAsync + AsStreamItemsAsync extension. Build clean, 87 passed. - Go: EventResult.ReplayGap field + IsReplayGap(); ReplayGap type alias. build/vet/test clean. - Rust: EventItem enum (Event/ReplayGap); EventStream now yields Result<EventItem, Error>; CLI renders REPLAY_GAP line / replayGap JSON. fmt/check/test/clippy clean. - Python: ReplayGap dataclass; stream_events yields pb.MxEvent | ReplayGap. 131 passed. - Shared docs: ClientLibrariesDesign non-goals reframed (reconnect-replay protocol is consumable; auto-reconnect stays a non-goal); CrossLanguageSmokeMatrix resume-gap note. Java client is deferred to the windev batch (no local JRE); CLI-15 stays open until it lands. Claude-Session: https://claude.ai/code/session_01DMXXvNuPekkkrTEyPNxEkW
107 lines
3.3 KiB
Python
107 lines
3.3 KiB
Python
"""Tests for the typed ReplayGap signal on Session.stream_events (CLI-15)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import AsyncIterator
|
|
|
|
import pytest
|
|
|
|
from zb_mom_ww_mxgateway import ReplayGap, Session
|
|
from zb_mom_ww_mxgateway.generated import mxaccess_gateway_pb2 as pb
|
|
|
|
|
|
class _FakeClient:
|
|
"""Minimal client stub exposing only what Session.stream_events needs."""
|
|
|
|
def __init__(self, events: list[pb.MxEvent]) -> None:
|
|
self._events = events
|
|
self.last_request: pb.StreamEventsRequest | None = None
|
|
|
|
def stream_events_raw(
|
|
self,
|
|
request: pb.StreamEventsRequest,
|
|
) -> AsyncIterator[pb.MxEvent]:
|
|
self.last_request = request
|
|
|
|
async def _gen() -> AsyncIterator[pb.MxEvent]:
|
|
for event in self._events:
|
|
yield event
|
|
|
|
return _gen()
|
|
|
|
|
|
def _gap_sentinel(*, requested: int, oldest: int) -> pb.MxEvent:
|
|
return pb.MxEvent(
|
|
session_id="session-1",
|
|
family=pb.MX_EVENT_FAMILY_UNSPECIFIED,
|
|
replay_gap=pb.ReplayGap(
|
|
requested_after_sequence=requested,
|
|
oldest_available_sequence=oldest,
|
|
),
|
|
)
|
|
|
|
|
|
def _normal_event(worker_sequence: int) -> pb.MxEvent:
|
|
return pb.MxEvent(
|
|
session_id="session-1",
|
|
worker_sequence=worker_sequence,
|
|
family=pb.MX_EVENT_FAMILY_ON_DATA_CHANGE,
|
|
)
|
|
|
|
|
|
def _session(events: list[pb.MxEvent]) -> tuple[Session, _FakeClient]:
|
|
client = _FakeClient(events)
|
|
session = Session(client=client, session_id="session-1") # type: ignore[arg-type]
|
|
return session, client
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_replay_gap_sentinel_surfaces_as_typed_signal() -> None:
|
|
session, _client = _session(
|
|
[_gap_sentinel(requested=5, oldest=10), _normal_event(10)],
|
|
)
|
|
|
|
items = [item async for item in session.stream_events(after_worker_sequence=5)]
|
|
|
|
assert isinstance(items[0], ReplayGap)
|
|
assert items[0].requested_after_sequence == 5
|
|
assert items[0].oldest_available_sequence == 10
|
|
# Resume cursor is oldest_available_sequence - 1 so the next replayed event
|
|
# is the oldest the gateway still retains.
|
|
assert items[0].resume_after_worker_sequence == 9
|
|
|
|
# Normal events after the sentinel pass through unchanged as MxEvent.
|
|
assert isinstance(items[1], pb.MxEvent)
|
|
assert not items[1].HasField("replay_gap")
|
|
assert items[1].worker_sequence == 10
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_normal_events_are_unaffected() -> None:
|
|
session, _client = _session([_normal_event(1), _normal_event(2)])
|
|
|
|
items = [item async for item in session.stream_events()]
|
|
|
|
assert all(isinstance(item, pb.MxEvent) for item in items)
|
|
assert [item.worker_sequence for item in items] == [1, 2]
|
|
assert not any(isinstance(item, ReplayGap) for item in items)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stream_events_forwards_resume_cursor() -> None:
|
|
session, client = _session([])
|
|
|
|
async for _ in session.stream_events(after_worker_sequence=42):
|
|
pass
|
|
|
|
assert client.last_request is not None
|
|
assert client.last_request.after_worker_sequence == 42
|
|
assert client.last_request.session_id == "session-1"
|
|
|
|
|
|
def test_replay_gap_resume_cursor_never_negative() -> None:
|
|
gap = ReplayGap.from_proto(
|
|
pb.ReplayGap(requested_after_sequence=0, oldest_available_sequence=0),
|
|
)
|
|
assert gap.resume_after_worker_sequence == 0
|