diff --git a/clients/dotnet/README.md b/clients/dotnet/README.md index 71ecb15..ebdd94c 100644 --- a/clients/dotnet/README.md +++ b/clients/dotnet/README.md @@ -84,6 +84,48 @@ messages. `MxGatewaySession.OpenSessionReply` keeps the raw session-open reply available, and command helpers have `*RawAsync` variants when callers need the complete `MxCommandReply`. +### Event Streaming And Replay Gaps + +`StreamEventsAsync(afterWorkerSequence)` yields raw generated `MxEvent` +messages. Passing a non-zero `afterWorkerSequence` resumes a session's event +stream after a known worker sequence — this is the reconnect cursor. If that +cursor is *stale* — older than the oldest event the gateway still retains in the +session replay ring — the events in between were evicted and cannot be replayed. +The gateway signals this by emitting a single **replay-gap sentinel** at the head +of the resumed stream: an `MxEvent` with its `ReplayGap` field set, `Family` +unspecified, and no body. It means "you missed events — discard local state and +re-snapshot." + +Rather than force callers to inspect the raw sentinel, the client exposes a +typed surface, `StreamEventItemsAsync`, which yields `MxEventStreamItem` values: + +```csharp +await foreach (MxEventStreamItem item in session.StreamEventItemsAsync( + afterWorkerSequence: lastSeenSequence)) +{ + if (item.IsReplayGap) + { + // We missed events: throw away local state and re-snapshot. + ReplayGap gap = item.ReplayGap!; + // Resume without incurring another gap: + lastSeenSequence = gap.OldestAvailableSequence - 1; + await ReSnapshotAsync(); + continue; + } + + HandleEvent(item.Event); // normal MXAccess event, IsReplayGap == false + lastSeenSequence = item.Event.WorkerSequence; +} +``` + +The typed surface never synthesizes or drops events — it only makes the +gateway's own sentinel observable. Normal events pass through with +`IsReplayGap == false` and `ReplayGap == null`. The gap is only ever produced by +`StreamEvents`; the diagnostic drain path never emits it. If you already consume +the raw `StreamEventsAsync` (or the client-level stream), the +`AsStreamItemsAsync()` extension projects any `IAsyncEnumerable` into +the same `MxEventStreamItem` surface. + For alarms, the client exposes `QueryActiveAlarmsAsync` (one-shot snapshot of the active alarms the gateway's central monitor currently holds), `StreamAlarmsAsync` (server-streaming feed of alarm-state-change messages diff --git a/clients/dotnet/ZB.MOM.WW.MxGateway.Client.Tests/MxGatewayClientSessionTests.cs b/clients/dotnet/ZB.MOM.WW.MxGateway.Client.Tests/MxGatewayClientSessionTests.cs index d827be1..1de4290 100644 --- a/clients/dotnet/ZB.MOM.WW.MxGateway.Client.Tests/MxGatewayClientSessionTests.cs +++ b/clients/dotnet/ZB.MOM.WW.MxGateway.Client.Tests/MxGatewayClientSessionTests.cs @@ -215,6 +215,55 @@ public sealed class MxGatewayClientSessionTests Assert.Equal("session-fixture", request.SessionId); } + /// + /// Verifies that a reconnect-replay gap sentinel is surfaced as a typed + /// with the gap populated, while normal + /// events pass through unchanged with IsReplayGap false. + /// + [Fact] + public async Task StreamEventItemsAsync_SurfacesReplayGapSentinel() + { + FakeGatewayTransport transport = CreateTransport(); + transport.AddEvent(new MxEvent + { + SessionId = "session-fixture", + ReplayGap = new ReplayGap + { + RequestedAfterSequence = 5, + OldestAvailableSequence = 42, + }, + }); + transport.AddEvent(new MxEvent + { + SessionId = "session-fixture", + Family = MxEventFamily.OnDataChange, + WorkerSequence = 42, + }); + await using MxGatewayClient client = CreateClient(transport); + MxGatewaySession session = await client.OpenSessionAsync(); + + List items = []; + await foreach (MxEventStreamItem item in session.StreamEventItemsAsync(afterWorkerSequence: 5)) + { + items.Add(item); + } + + Assert.Equal(2, items.Count); + + MxEventStreamItem gap = items[0]; + Assert.True(gap.IsReplayGap); + Assert.NotNull(gap.ReplayGap); + Assert.Equal(5UL, gap.ReplayGap!.RequestedAfterSequence); + Assert.Equal(42UL, gap.ReplayGap.OldestAvailableSequence); + Assert.Same(gap.ReplayGap, gap.Event.ReplayGap); + + MxEventStreamItem normal = items[1]; + Assert.False(normal.IsReplayGap); + Assert.Null(normal.ReplayGap); + Assert.Equal(42UL, normal.Event.WorkerSequence); + Assert.Equal(MxEventFamily.OnDataChange, normal.Event.Family); + } + /// Verifies that close is explicit and idempotent. [Fact] public async Task CloseAsync_IsExplicitAndIdempotent() diff --git a/clients/dotnet/ZB.MOM.WW.MxGateway.Client/MxEventStreamExtensions.cs b/clients/dotnet/ZB.MOM.WW.MxGateway.Client/MxEventStreamExtensions.cs new file mode 100644 index 0000000..67b376a --- /dev/null +++ b/clients/dotnet/ZB.MOM.WW.MxGateway.Client/MxEventStreamExtensions.cs @@ -0,0 +1,40 @@ +using System.Runtime.CompilerServices; +using ZB.MOM.WW.MxGateway.Contracts.Proto; + +namespace ZB.MOM.WW.MxGateway.Client; + +/// +/// Extension methods that project a raw stream into the +/// typed surface, making the gateway's +/// reconnect-replay gap sentinel observable. +/// +public static class MxEventStreamExtensions +{ + /// + /// Projects a raw stream (e.g. + /// or + /// ) into typed + /// values. Normal events pass through with + /// false; the gateway's + /// reconnect-replay gap sentinel is surfaced with + /// true and + /// populated. The stream is + /// forwarded faithfully — no event is synthesized or dropped. + /// + /// The raw event stream to wrap. + /// Cancellation token for the enumeration. + /// The same events, each wrapped as an . + public static async IAsyncEnumerable AsStreamItemsAsync( + this IAsyncEnumerable source, + [EnumeratorCancellation] CancellationToken cancellationToken = default) + { + ArgumentNullException.ThrowIfNull(source); + + await foreach (MxEvent gatewayEvent in source + .WithCancellation(cancellationToken) + .ConfigureAwait(false)) + { + yield return MxEventStreamItem.From(gatewayEvent); + } + } +} diff --git a/clients/dotnet/ZB.MOM.WW.MxGateway.Client/MxEventStreamItem.cs b/clients/dotnet/ZB.MOM.WW.MxGateway.Client/MxEventStreamItem.cs new file mode 100644 index 0000000..c2f79df --- /dev/null +++ b/clients/dotnet/ZB.MOM.WW.MxGateway.Client/MxEventStreamItem.cs @@ -0,0 +1,82 @@ +using ZB.MOM.WW.MxGateway.Contracts.Proto; + +namespace ZB.MOM.WW.MxGateway.Client; + +/// +/// One item yielded by the typed event stream. It is either a normal MXAccess +/// or a reconnect-replay gap sentinel. +/// +/// +/// +/// The gateway emits a single sentinel at the head of a +/// StreamEvents stream that was resumed via +/// StreamEventsRequest.after_worker_sequence when the requested sequence +/// predates the oldest event still retained in the session replay ring — i.e. +/// events were evicted and cannot be replayed. On that sentinel the +/// MxEvent.replay_gap field is set, is +/// , the body oneof is unset, and +/// no per-item fields are populated. +/// +/// +/// This wrapper makes that sentinel observable instead of forcing the consumer +/// to inspect the raw . It never synthesizes or swallows an +/// event: the gap is exactly the gateway's own sentinel, exposed with +/// set and populated. Every +/// other event flows through unchanged with false. +/// +/// +/// When is the consumer has +/// missed events and MUST discard local state and re-snapshot. To resume without +/// incurring another gap, reconnect with +/// after_worker_sequence = ReplayGap.OldestAvailableSequence - 1. +/// +/// +public sealed class MxEventStreamItem +{ + private MxEventStreamItem(MxEvent gatewayEvent, ReplayGap? replayGap) + { + Event = gatewayEvent; + ReplayGap = replayGap; + } + + /// + /// The underlying raw . For a normal event this is the + /// MXAccess event itself; for a replay-gap item this is the gateway's + /// sentinel event whose only meaningful payload is . + /// Never . + /// + public MxEvent Event { get; } + + /// + /// The reconnect-replay gap payload when this item is a gap sentinel; + /// otherwise . Read + /// and + /// to learn + /// which events were lost. + /// + public ReplayGap? ReplayGap { get; } + + /// + /// when this item is a reconnect-replay gap sentinel: + /// the consumer missed events and must discard local state and re-snapshot. + /// Resume without another gap by reconnecting with + /// after_worker_sequence = ReplayGap.OldestAvailableSequence - 1. + /// + public bool IsReplayGap => ReplayGap is not null; + + /// + /// Wraps a raw stream as a typed item, classifying it + /// as a replay-gap sentinel when MxEvent.replay_gap is present. + /// + /// The raw event from the StreamEvents stream. + /// A typed item exposing either the normal event or the replay gap. + internal static MxEventStreamItem From(MxEvent gatewayEvent) + { + ArgumentNullException.ThrowIfNull(gatewayEvent); + + // For a message-typed proto3 field, presence is a non-null reference. + return gatewayEvent.ReplayGap is { } replayGap + ? new MxEventStreamItem(gatewayEvent, replayGap) + : new MxEventStreamItem(gatewayEvent, replayGap: null); + } +} diff --git a/clients/dotnet/ZB.MOM.WW.MxGateway.Client/MxGatewaySession.cs b/clients/dotnet/ZB.MOM.WW.MxGateway.Client/MxGatewaySession.cs index e0c6f24..1740a96 100644 --- a/clients/dotnet/ZB.MOM.WW.MxGateway.Client/MxGatewaySession.cs +++ b/clients/dotnet/ZB.MOM.WW.MxGateway.Client/MxGatewaySession.cs @@ -875,6 +875,34 @@ public sealed class MxGatewaySession : IAsyncDisposable cancellationToken); } + /// + /// Streams events as typed values, surfacing + /// the gateway's reconnect-replay gap sentinel as an observable, typed signal. + /// + /// + /// When resuming with a stale (older than + /// the oldest event still retained in the session replay ring), the gateway emits + /// a single gap sentinel at the head of the stream. It arrives here as an item + /// with true and + /// populated, meaning the consumer + /// missed events and MUST discard local state and re-snapshot. To resume without + /// incurring another gap, reconnect with + /// afterWorkerSequence = item.ReplayGap.OldestAvailableSequence - 1. + /// All other events pass through with + /// false. Use when raw generated + /// messages are needed instead. + /// + /// The sequence number to stream from. Defaults to 0. + /// Cancellation token. + /// An async enumerable of typed event items. + public IAsyncEnumerable StreamEventItemsAsync( + ulong afterWorkerSequence = 0, + CancellationToken cancellationToken = default) + { + return StreamEventsAsync(afterWorkerSequence, cancellationToken) + .AsStreamItemsAsync(cancellationToken); + } + /// /// Closes the session and releases resources. /// diff --git a/clients/go/README.md b/clients/go/README.md index 1c66881..2adfec4 100644 --- a/clients/go/README.md +++ b/clients/go/README.md @@ -92,6 +92,44 @@ goroutine cleanup. Raw protobuf messages remain available through the `errors.As` for `GatewayError`, `CommandError`, and `MxAccessError`; command errors preserve the raw reply. +### Reconnect-replay gap + +Each `EventResult` carries exactly one of `Event`, `ReplayGap`, or `Err`. When +you resume a stream with `EventsAfter`/`SubscribeEventsAfter` and a non-zero +`afterWorkerSequence`, the gateway replays buffered events from that point. If +the requested sequence predates the oldest event still retained in its replay +ring, it delivers a single **replay-gap sentinel** at the head of the resumed +stream: `res.ReplayGap` is non-nil (`res.Event` is nil, `res.IsReplayGap()` is +true) and normal events follow it. + +A gap means events were lost, so any locally cached tag/alarm state is now +stale. On seeing it, discard your cached state and re-snapshot. To resume +without provoking another gap, reconnect from just before the oldest retained +sequence: + +```go +for res := range events { + switch { + case res.Err != nil: + // terminal: stream ended (see ErrSlowConsumer for the overflow case) + return res.Err + case res.IsReplayGap(): + gap := res.ReplayGap + log.Printf("replay gap: requested after %d, oldest available %d; re-snapshotting", + gap.GetRequestedAfterSequence(), gap.GetOldestAvailableSequence()) + resnapshot() + // to resume cleanly, reconnect with: + // session.EventsAfter(ctx, gap.GetOldestAvailableSequence()-1) + default: + handle(res.Event) + } +} +``` + +The gateway sets `ReplayGap` only on `StreamEvents` results (never on a fresh, +non-resumed stream and never on `DrainEvents`). The client makes the gateway's +sentinel typed and observable; it never synthesizes or swallows it. + For alarms, the package exposes `Client.QueryActiveAlarms` for one-shot snapshots, `Client.StreamAlarms` for the server-streaming feed, and `Client.AcknowledgeAlarm` to ack an alarm by full reference. The streaming diff --git a/clients/go/mxgateway/client_session_test.go b/clients/go/mxgateway/client_session_test.go index fa1f85b..172d268 100644 --- a/clients/go/mxgateway/client_session_test.go +++ b/clients/go/mxgateway/client_session_test.go @@ -200,6 +200,63 @@ func TestEventsSlowConsumerYieldsErrSlowConsumerBeforeClose(t *testing.T) { } } +func TestEventsSurfacesReplayGapSentinelAsTypedSignal(t *testing.T) { + fake := &fakeGatewayServer{ + streamStarted: make(chan struct{}), + streamReplayGap: &pb.ReplayGap{ + RequestedAfterSequence: 5, + OldestAvailableSequence: 42, + }, + } + client, cleanup := newBufconnClient(t, fake) + defer cleanup() + session := NewSessionForID(client, "session-1") + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + events, err := session.EventsAfter(ctx, 5) + if err != nil { + t.Fatalf("EventsAfter() error = %v", err) + } + <-fake.streamStarted + + // First result must be the typed replay-gap signal, not a normal event. + first := <-events + if first.Err != nil { + t.Fatalf("first result error = %v", first.Err) + } + if !first.IsReplayGap() { + t.Fatalf("first result IsReplayGap() = false, want true") + } + if first.Event != nil { + t.Fatalf("replay-gap result carried a non-nil Event %+v; want the sentinel to clear Event", first.Event) + } + if got := first.ReplayGap.GetRequestedAfterSequence(); got != 5 { + t.Fatalf("ReplayGap.RequestedAfterSequence = %d, want 5", got) + } + if got := first.ReplayGap.GetOldestAvailableSequence(); got != 42 { + t.Fatalf("ReplayGap.OldestAvailableSequence = %d, want 42", got) + } + + // Normal events after the sentinel are unaffected: Event set, ReplayGap nil. + second := <-events + if second.Err != nil { + t.Fatalf("second result error = %v", second.Err) + } + if second.IsReplayGap() { + t.Fatalf("second result IsReplayGap() = true, want false for a normal event") + } + if second.Event == nil { + t.Fatal("second result Event = nil, want a normal event") + } + if got := second.Event.GetWorkerSequence(); got != 1 { + t.Fatalf("normal event worker sequence = %d, want 1", got) + } + if second.Event.GetFamily() != pb.MxEventFamily_MX_EVENT_FAMILY_ON_DATA_CHANGE { + t.Fatalf("normal event family = %s, want ON_DATA_CHANGE", second.Event.GetFamily()) + } +} + func TestSessionHelpersBuildCommandsAndExposeRawReply(t *testing.T) { fake := &fakeGatewayServer{ invokeReply: &pb.MxCommandReply{ @@ -643,6 +700,7 @@ type fakeGatewayServer struct { streamStarted chan struct{} streamDone chan struct{} streamEventCount int + streamReplayGap *pb.ReplayGap invokeReply *pb.MxCommandReply invokeRequest *pb.MxCommandRequest } @@ -691,6 +749,16 @@ func (s *fakeGatewayServer) StreamEvents(req *pb.StreamEventsRequest, stream grp if s.streamStarted != nil { close(s.streamStarted) } + if s.streamReplayGap != nil { + // Emit the reconnect-replay gap sentinel at the head of the resumed + // stream: family UNSPECIFIED, body unset, only replay_gap populated. + if err := stream.Send(&pb.MxEvent{ + SessionId: req.GetSessionId(), + ReplayGap: s.streamReplayGap, + }); err != nil { + return err + } + } eventCount := s.streamEventCount if eventCount == 0 { eventCount = 1 diff --git a/clients/go/mxgateway/session.go b/clients/go/mxgateway/session.go index e61cf46..c58e00f 100644 --- a/clients/go/mxgateway/session.go +++ b/clients/go/mxgateway/session.go @@ -27,14 +27,39 @@ const eventBufferSize = 16 // non-blockingly on overflow, even when all data slots are full. const eventBufferReservedSlots = 1 -// EventResult carries either the next ordered event or a terminal stream error. +// EventResult carries the next ordered event, a replay-gap signal, or a +// terminal stream error. Exactly one of Event, ReplayGap, or Err is set on any +// delivered result. type EventResult struct { - // Event is the next event from the stream when Err is nil. + // Event is the next MXAccess event from the stream when both ReplayGap and + // Err are nil. Event *MxEvent + // ReplayGap, when non-nil, is the gateway's reconnect-replay gap sentinel: it + // is delivered at the head of a resumed stream (one opened with a non-zero + // after_worker_sequence via EventsAfter/SubscribeEventsAfter) when the + // requested sequence predates the oldest event still retained in the replay + // ring, so the events in between were lost. + // + // It is a non-terminal, observable signal — the stream continues with normal + // events after it, and Event is nil on a gap result so a gap is never + // mistaken for a normal MXAccess event. On seeing a gap the consumer must + // discard any locally cached tag/alarm state and re-snapshot. To resume + // cleanly (without provoking another gap), reconnect with EventsAfter using + // afterWorkerSequence = ReplayGap.GetOldestAvailableSequence() - 1. + // + // The gateway sets ReplayGap only on StreamEvents results, never on a normal + // (non-resumed) stream and never on DrainEvents. + ReplayGap *ReplayGap // Err is the terminal stream error; when non-nil no further results follow. Err error } +// IsReplayGap reports whether this result carries the gateway's reconnect-replay +// gap sentinel rather than a normal event or a terminal error. +func (r EventResult) IsReplayGap() bool { + return r.ReplayGap != nil +} + // EventSubscription owns a running gateway event stream. type EventSubscription struct { results <-chan EventResult @@ -736,7 +761,15 @@ func (s *Session) subscribeEventsAfter(ctx context.Context, afterWorkerSequence for { event, err := stream.Recv() if err == nil { - if !sendEventResult(streamCtx, results, EventResult{Event: event}, cancelWhenResultBufferFull, cancel) { + result := EventResult{Event: event} + // The gateway marks a reconnect-replay gap with a sentinel MxEvent + // carrying replay_gap (family UNSPECIFIED, body unset). Surface it + // as a distinct typed signal rather than a normal event: clear + // Event so consumers never process the sentinel as a data change. + if gap := event.GetReplayGap(); gap != nil { + result = EventResult{ReplayGap: gap} + } + if !sendEventResult(streamCtx, results, result, cancelWhenResultBufferFull, cancel) { return } continue diff --git a/clients/go/mxgateway/types.go b/clients/go/mxgateway/types.go index 26eba2e..8b9f402 100644 --- a/clients/go/mxgateway/types.go +++ b/clients/go/mxgateway/types.go @@ -30,6 +30,12 @@ type ( MxCommand = pb.MxCommand // MxEvent is one ordered event delivered on a session event stream. MxEvent = pb.MxEvent + // ReplayGap is the gateway sentinel payload signalling that a resumed event + // stream skipped past the oldest event still retained in the replay ring. + // RequestedAfterSequence is the after_worker_sequence the client resumed + // from; OldestAvailableSequence is the oldest sequence the gateway can still + // replay. See EventResult.ReplayGap for consumption guidance. + ReplayGap = pb.ReplayGap // MxValue is the protobuf representation of an MXAccess value. MxValue = pb.MxValue // Value is an alias for MxValue retained for symmetry with other clients. diff --git a/clients/python/README.md b/clients/python/README.md index 3833809..ef8409f 100644 --- a/clients/python/README.md +++ b/clients/python/README.md @@ -105,6 +105,40 @@ terminate the stream. Canceling a Python task cancels the client-side gRPC call or stream wait. It does not abort an in-flight MXAccess COM call inside the worker process. +### Event streaming and reconnect gaps + +`Session.stream_events()` yields an async iterator whose items are either a +normal `MxEvent` or a `ReplayGap`. Track the `worker_sequence` of the last event +you processed and pass it back as `after_worker_sequence` to resume after a +disconnect: + +```python +from zb_mom_ww_mxgateway import ReplayGap + +cursor = 0 +async for item in session.stream_events(after_worker_sequence=cursor): + if isinstance(item, ReplayGap): + # The gateway dropped events between item.requested_after_sequence and + # item.oldest_available_sequence — they are gone from the replay ring. + # Discard local tag/alarm state and re-snapshot (e.g. read_bulk / + # query_active_alarms), then resume without another gap: + cursor = item.resume_after_worker_sequence # oldest_available_sequence - 1 + continue + # Normal MXAccess event. + cursor = item.worker_sequence + handle(item) +``` + +`ReplayGap` is the gateway's reconnect-replay gap sentinel made typed and +observable. It is delivered only at the head of a stream resumed with a non-zero +`after_worker_sequence` when the requested cursor predates the oldest retained +event. It is a **non-terminal** signal — the stream continues with normal events +after it — and it is never yielded as an `MxEvent`, so it can never be mistaken +for a real MXAccess event. The client does not synthesize or swallow it; the +gateway only sets it on `StreamEvents` results (never on a fresh stream or a +`DrainEvents` reply). `GatewayClient.stream_events_raw` remains the raw protobuf +stream (the sentinel arrives there as an `MxEvent` with `replay_gap` set). + ## Write Semantics And Common Pitfalls These are MXAccess parity behaviors that surprise new callers. The gateway diff --git a/clients/python/src/zb_mom_ww_mxgateway/__init__.py b/clients/python/src/zb_mom_ww_mxgateway/__init__.py index 49bace6..0d9943d 100644 --- a/clients/python/src/zb_mom_ww_mxgateway/__init__.py +++ b/clients/python/src/zb_mom_ww_mxgateway/__init__.py @@ -9,6 +9,7 @@ from .generated.galaxy_repository_pb2 import ( GalaxyObject, WatchDeployEventsRequest, ) +from .events import ReplayGap from .errors import ( MxAccessError, MxGatewayAuthenticationError, @@ -43,6 +44,7 @@ __all__ = [ "MxGatewayTransportError", "MxGatewayWorkerError", "MxValueView", + "ReplayGap", "Session", "WatchDeployEventsRequest", "__version__", diff --git a/clients/python/src/zb_mom_ww_mxgateway/events.py b/clients/python/src/zb_mom_ww_mxgateway/events.py new file mode 100644 index 0000000..fe074de --- /dev/null +++ b/clients/python/src/zb_mom_ww_mxgateway/events.py @@ -0,0 +1,63 @@ +"""Typed event-stream signals for the MXAccess Gateway Python client.""" + +from __future__ import annotations + +from dataclasses import dataclass + +from .generated import mxaccess_gateway_pb2 as pb + + +@dataclass(frozen=True, slots=True) +class ReplayGap: + """Reconnect-replay gap signal surfaced on a resumed event stream. + + The gateway emits this at the head of a stream resumed with + :meth:`Session.stream_events`'s ``after_worker_sequence`` cursor when the + requested sequence predates the oldest event still retained in the gateway's + replay ring. That means the events between ``requested_after_sequence`` and + ``oldest_available_sequence`` were dropped from the ring and can no longer be + replayed — the client has an unrecoverable hole in its event history. + + ``ReplayGap`` is a *non-terminal, observable* signal: the stream keeps + delivering normal :class:`~zb_mom_ww_mxgateway.generated.mxaccess_gateway_pb2.MxEvent` + values after it. :meth:`Session.stream_events` yields it as a distinct type + (never as an ``MxEvent``) so a consumer can branch on + ``isinstance(item, ReplayGap)`` and never mistake a gap for a real MXAccess + event. The client neither synthesizes nor swallows the gateway's sentinel — + it only makes that sentinel typed and observable. + + On seeing a gap the consumer must discard any locally cached tag/alarm state + and re-snapshot (for example via :meth:`Session.read_bulk` or + :meth:`~zb_mom_ww_mxgateway.GatewayClient.query_active_alarms`). To resume + the stream without provoking another gap, reconnect with + ``after_worker_sequence = gap.resume_after_worker_sequence`` (that is, + ``oldest_available_sequence - 1``) so the next replayed event is the oldest + the gateway still retains. + + The gateway sets this only on ``StreamEvents`` results — never on a normal + (non-resumed) stream and never on a ``DrainEvents`` reply. + """ + + requested_after_sequence: int + """The ``after_worker_sequence`` cursor the resumed stream was opened with.""" + + oldest_available_sequence: int + """Oldest worker sequence the gateway can still replay.""" + + @classmethod + def from_proto(cls, gap: pb.ReplayGap) -> "ReplayGap": + """Build a :class:`ReplayGap` from the generated ``ReplayGap`` message.""" + return cls( + requested_after_sequence=gap.requested_after_sequence, + oldest_available_sequence=gap.oldest_available_sequence, + ) + + @property + def resume_after_worker_sequence(self) -> int: + """``after_worker_sequence`` to resume the stream without another gap. + + Equal to ``oldest_available_sequence - 1`` so the next event the gateway + replays is ``oldest_available_sequence`` — the oldest it still retains. + Clamped at ``0`` so it is never negative. + """ + return max(self.oldest_available_sequence - 1, 0) diff --git a/clients/python/src/zb_mom_ww_mxgateway/session.py b/clients/python/src/zb_mom_ww_mxgateway/session.py index 1302025..973ba9a 100644 --- a/clients/python/src/zb_mom_ww_mxgateway/session.py +++ b/clients/python/src/zb_mom_ww_mxgateway/session.py @@ -5,6 +5,7 @@ from __future__ import annotations from collections.abc import AsyncIterator, Sequence from .errors import ensure_mxaccess_success +from .events import ReplayGap from .generated import mxaccess_gateway_pb2 as pb from .values import MxValueInput, to_mx_value @@ -572,14 +573,57 @@ class Session: self, *, after_worker_sequence: int = 0, - ) -> AsyncIterator[pb.MxEvent]: - """Return an async iterator of `MxEvent` messages for this session.""" - return self.client.stream_events_raw( + ) -> AsyncIterator[pb.MxEvent | ReplayGap]: + """Return an async iterator over this session's `MxEvent` stream. + + Each yielded item is either a normal :class:`~...MxEvent` or a + :class:`ReplayGap`. Branch on ``isinstance(item, ReplayGap)`` — a gap is + never delivered as an ``MxEvent`` so it cannot be mistaken for a real + MXAccess event. + + Pass a non-zero *after_worker_sequence* to resume a previously observed + stream. If that cursor predates the oldest event the gateway still + retains in its replay ring, the stream opens with a single + :class:`ReplayGap` sentinel (events in the gap were dropped and cannot be + replayed), then continues with normal events. On a gap, discard locally + cached state, re-snapshot, and — to resume without another gap — + reconnect with ``after_worker_sequence = gap.resume_after_worker_sequence``. + See :class:`ReplayGap` for the full semantics. The underlying protobuf + stream is available raw via ``GatewayClient.stream_events_raw``. + """ + raw = self.client.stream_events_raw( pb.StreamEventsRequest( session_id=self.session_id, after_worker_sequence=after_worker_sequence, ), ) + return _surface_replay_gaps(raw) + + +async def _surface_replay_gaps( + raw: AsyncIterator[pb.MxEvent], +) -> AsyncIterator[pb.MxEvent | ReplayGap]: + """Map the gateway's ``replay_gap`` sentinel event to a typed :class:`ReplayGap`. + + Normal events pass through unchanged. The sentinel (``replay_gap`` set, + ``family`` unspecified, body unset) is converted to a distinct + :class:`ReplayGap` so a consumer can branch on it without inspecting proto + presence, and is never yielded as an ``MxEvent``. The sentinel is forwarded + faithfully — it is neither dropped nor turned into a normal event. + + Closing this generator (``aclose``) propagates to *raw* so the underlying + gRPC call is cancelled, preserving the raw stream's cancel-on-stop contract. + """ + try: + async for event in raw: + if event.HasField("replay_gap"): + yield ReplayGap.from_proto(event.replay_gap) + else: + yield event + finally: + aclose = getattr(raw, "aclose", None) + if aclose is not None: + await aclose() def _ensure_bulk_size(name: str, count: int) -> None: diff --git a/clients/python/tests/test_replay_gap.py b/clients/python/tests/test_replay_gap.py new file mode 100644 index 0000000..472e9b4 --- /dev/null +++ b/clients/python/tests/test_replay_gap.py @@ -0,0 +1,106 @@ +"""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 diff --git a/clients/rust/README.md b/clients/rust/README.md index 4548fee..e27f707 100644 --- a/clients/rust/README.md +++ b/clients/rust/README.md @@ -137,6 +137,47 @@ redaction. Per-item bulk failures are reported inside each result entry `invoke_raw` / `client.invoke_raw` escape hatch performs neither check and returns the unvalidated reply. +## Event Streaming And Reconnect-Replay Gaps + +`session.events()` / `session.events_after(after_worker_sequence)` (and the +lower-level `client.stream_events`) return an `EventStream` that yields +`EventItem` values, not bare `MxEvent`s: + +```rust +use zb_mom_ww_mxgateway_client::EventItem; + +let mut stream = session.events_after(cursor).await?; +while let Some(item) = stream.next().await { + match item? { + EventItem::Event(event) => { /* apply the MXAccess change */ } + EventItem::ReplayGap(gap) => { + // Recent history was evicted — discard local state and re-snapshot, + // then resume without provoking another gap: + let resume = gap.oldest_available_sequence.saturating_sub(1); + stream = session.events_after(resume).await?; + } + } +} +``` + +Almost every item is a normal `EventItem::Event`. `EventItem::ReplayGap` is a +faithful, typed surfacing of the gateway's reconnect-replay gap sentinel — the +client does not synthesize it. The gateway emits the sentinel at most once, at +the head of a stream **resumed** via `events_after` (`after_worker_sequence`) +when the requested sequence is older than the oldest event still retained in the +session's replay ring: events in the open interval +`(requested_after_sequence, oldest_available_sequence)` were evicted and cannot +be replayed. A `ReplayGap` therefore means "you missed events — discard any +local state and re-snapshot." To resume without a second gap, reconnect with +`events_after(gap.oldest_available_sequence - 1)`, which replays starting at the +first still-retained event. A stream opened from the beginning +(`session.events()` / `events_after(0)`) never produces a `ReplayGap`. + +`EventItem` provides `as_event()`, `into_event()`, and `replay_gap()` accessors +for callers that prefer not to `match`. The `mxgw-cli stream-events` subcommand +renders the sentinel as a distinct `REPLAY_GAP …` line (or a `replayGap` JSON +object under `--json` / `--jsonl`). + ## Write Semantics And Common Pitfalls These are MXAccess parity behaviors that surprise new callers. The gateway diff --git a/clients/rust/crates/mxgw-cli/src/main.rs b/clients/rust/crates/mxgw-cli/src/main.rs index 60d7b21..8a84ffe 100644 --- a/clients/rust/crates/mxgw-cli/src/main.rs +++ b/clients/rust/crates/mxgw-cli/src/main.rs @@ -28,8 +28,8 @@ use zb_mom_ww_mxgateway_client::generated::mxaccess_gateway::v1::{ WriteSecuredBulkEntry, }; use zb_mom_ww_mxgateway_client::{ - next_correlation_id, ApiKey, ClientOptions, Error, GalaxyClient, GatewayClient, MxValue, - MxValueProjection, CLIENT_VERSION, GATEWAY_PROTOCOL_VERSION, WORKER_PROTOCOL_VERSION, + next_correlation_id, ApiKey, ClientOptions, Error, EventItem, GalaxyClient, GatewayClient, + MxValue, MxValueProjection, CLIENT_VERSION, GATEWAY_PROTOCOL_VERSION, WORKER_PROTOCOL_VERSION, }; const MAX_AGGREGATE_EVENTS: usize = 10_000; @@ -886,17 +886,43 @@ async fn dispatch(command: Command) -> Result<(), Error> { let mut events: Vec = Vec::new(); let mut event_count = 0usize; while event_count < max_events { - let Some(event) = stream.next().await else { + let Some(item) = stream.next().await else { break; }; - let event = event?; + let item = item?; event_count += 1; - if jsonl { - println!("{}", event_to_json(&event)); - } else if json { - events.push(event_to_json(&event)); - } else { - println!("{} {}", event.worker_sequence, event.family); + match item { + EventItem::Event(event) => { + if jsonl { + println!("{}", event_to_json(&event)); + } else if json { + events.push(event_to_json(&event)); + } else { + println!("{} {}", event.worker_sequence, event.family); + } + } + // Reconnect-replay gap sentinel: recent history was evicted + // before this resumed stream could replay it. Render it as a + // distinct row so the caller can re-snapshot and resume with + // `oldest_available_sequence - 1`. + EventItem::ReplayGap(gap) => { + let value = json!({ + "replayGap": { + "requestedAfterSequence": gap.requested_after_sequence, + "oldestAvailableSequence": gap.oldest_available_sequence, + } + }); + if jsonl { + println!("{value}"); + } else if json { + events.push(value); + } else { + println!( + "REPLAY_GAP requested_after={} oldest_available={}", + gap.requested_after_sequence, gap.oldest_available_sequence + ); + } + } } } if json { diff --git a/clients/rust/src/client.rs b/clients/rust/src/client.rs index 493503c..aa3e9b1 100644 --- a/clients/rust/src/client.rs +++ b/clients/rust/src/client.rs @@ -18,7 +18,7 @@ use crate::generated::mxaccess_gateway::v1::mx_access_gateway_client::MxAccessGa use crate::generated::mxaccess_gateway::v1::{ AcknowledgeAlarmReply, AcknowledgeAlarmRequest, ActiveAlarmSnapshot, AlarmFeedMessage, CloseSessionReply, CloseSessionRequest, MxCommandReply, MxCommandRequest, MxEvent, - OpenSessionReply, OpenSessionRequest, QueryActiveAlarmsRequest, StreamAlarmsRequest, + OpenSessionReply, OpenSessionRequest, QueryActiveAlarmsRequest, ReplayGap, StreamAlarmsRequest, StreamEventsRequest, }; use crate::options::{build_tls_config, ClientOptions}; @@ -28,11 +28,120 @@ use crate::session::Session; /// [`GatewayClient`] uses internally. pub type RawGatewayClient = MxAccessGatewayClient>; -/// Pinned, boxed [`MxEvent`] stream returned by -/// [`GatewayClient::stream_events`]. Errors are pre-mapped from -/// `tonic::Status` to [`Error`]; dropping the stream cancels the call. +/// One item yielded by the per-session event stream returned by +/// [`GatewayClient::stream_events`]. +/// +/// Almost every item is an ordinary MXAccess event ([`EventItem::Event`]). +/// The one exception is the reconnect-replay gap sentinel +/// ([`EventItem::ReplayGap`]): the gateway emits it at most once, at the head +/// of a stream that was *resumed* via +/// [`Session::events_after`](crate::session::Session::events_after) +/// (`StreamEventsRequest.after_worker_sequence`) when the requested sequence is +/// older than the oldest event still retained in the session replay ring — i.e. +/// events were evicted and cannot be replayed. +/// +/// The client does **not** synthesize this signal: it faithfully forwards the +/// gateway's sentinel `MxEvent` (whose `replay_gap` field is set), and only +/// makes it a distinct, typed variant so consumers can `match` on it instead of +/// inspecting a field on a value that otherwise looks like a normal event. +/// +/// # Reacting to a gap +/// +/// A [`EventItem::ReplayGap`] means "you missed events — discard any local +/// state and re-snapshot." The events in the open interval +/// `(requested_after_sequence, oldest_available_sequence)` are gone. To resume +/// the stream without provoking another gap, reconnect with +/// [`Session::events_after`](crate::session::Session::events_after) passing +/// `oldest_available_sequence - 1`, which replays starting at the first still +/// retained event (`oldest_available_sequence`): +/// +/// ```no_run +/// # use zb_mom_ww_mxgateway_client::{EventItem, Session}; +/// # use futures_util::StreamExt; +/// # async fn run(session: Session, cursor: u64) -> Result<(), zb_mom_ww_mxgateway_client::Error> { +/// let mut stream = session.events_after(cursor).await?; +/// while let Some(item) = stream.next().await { +/// match item? { +/// EventItem::Event(event) => { +/// let _ = event; // apply the change +/// } +/// EventItem::ReplayGap(gap) => { +/// // Local state is stale — re-snapshot, then resume without a gap. +/// let resume_cursor = gap.oldest_available_sequence.saturating_sub(1); +/// stream = session.events_after(resume_cursor).await?; +/// } +/// } +/// } +/// # Ok(()) +/// # } +/// ``` +// The `Event` variant is the hot path (nearly every stream item) and is the +// large one; the rare `ReplayGap` sentinel is small. Boxing `Event` to equalize +// the variants would add a heap allocation to every streamed event — a +// regression versus the prior `Result` surface, which already +// moved `MxEvent` by value. Keep the common path allocation-free. +#[allow(clippy::large_enum_variant)] +#[derive(Clone, Debug, PartialEq)] +pub enum EventItem { + /// A normal MXAccess event forwarded from the worker. + Event(MxEvent), + /// The reconnect-replay gap sentinel — recent event history was evicted + /// before this resumed stream could replay it. See [`EventItem`] for how + /// to react. + ReplayGap(ReplayGap), +} + +impl EventItem { + /// Classify an incoming `MxEvent` into the typed stream item. + /// + /// A present `replay_gap` promotes the event to [`EventItem::ReplayGap`]; + /// otherwise it is an ordinary [`EventItem::Event`]. The sentinel is never + /// dropped and never surfaced as a normal event. + fn from_event(mut event: MxEvent) -> Self { + match event.replay_gap.take() { + Some(gap) => EventItem::ReplayGap(gap), + None => EventItem::Event(event), + } + } + + /// Borrow the inner [`MxEvent`] when this item is a normal event, or + /// `None` when it is the [`EventItem::ReplayGap`] sentinel. + #[must_use] + pub fn as_event(&self) -> Option<&MxEvent> { + match self { + EventItem::Event(event) => Some(event), + EventItem::ReplayGap(_) => None, + } + } + + /// Borrow the [`ReplayGap`] when this item is the reconnect-replay gap + /// sentinel, or `None` for a normal event. + #[must_use] + pub fn replay_gap(&self) -> Option<&ReplayGap> { + match self { + EventItem::ReplayGap(gap) => Some(gap), + EventItem::Event(_) => None, + } + } + + /// Consume the item and return the inner [`MxEvent`] when it is a normal + /// event, or `None` for the [`EventItem::ReplayGap`] sentinel. + #[must_use] + pub fn into_event(self) -> Option { + match self { + EventItem::Event(event) => Some(event), + EventItem::ReplayGap(_) => None, + } + } +} + +/// Pinned, boxed [`EventItem`] stream returned by +/// [`GatewayClient::stream_events`]. Each item is either a normal +/// [`EventItem::Event`] or the [`EventItem::ReplayGap`] reconnect-replay +/// sentinel. Errors are pre-mapped from `tonic::Status` to [`Error`]; dropping +/// the stream cancels the call. pub type EventStream = - std::pin::Pin> + Send + 'static>>; + std::pin::Pin> + Send + 'static>>; /// Pinned, boxed [`ActiveAlarmSnapshot`] stream returned by /// [`GatewayClient::query_active_alarms`]. Errors are pre-mapped from @@ -190,8 +299,12 @@ impl GatewayClient { /// Open the server-streaming `StreamEvents` RPC. /// - /// The returned [`EventStream`] yields `MxEvent` messages as the worker - /// produces them. Dropping the stream cancels the gRPC call cooperatively. + /// The returned [`EventStream`] yields [`EventItem`] values as the worker + /// produces them: ordinary MXAccess events as [`EventItem::Event`], and the + /// gateway's reconnect-replay gap sentinel — set only on resumed streams + /// whose requested sequence predates the retained replay history — as + /// [`EventItem::ReplayGap`]. Dropping the stream cancels the gRPC call + /// cooperatively. /// /// # Errors /// @@ -201,7 +314,7 @@ impl GatewayClient { let mut client = self.inner.clone(); let response = client.stream_events(self.stream_request(request)).await?; let stream = futures_util::StreamExt::map(response.into_inner(), |result| { - result.map_err(Error::from) + result.map(EventItem::from_event).map_err(Error::from) }); Ok(Box::pin(stream)) diff --git a/clients/rust/src/lib.rs b/clients/rust/src/lib.rs index 44acce7..50d4306 100644 --- a/clients/rust/src/lib.rs +++ b/clients/rust/src/lib.rs @@ -24,12 +24,14 @@ pub mod version; #[doc(inline)] pub use auth::{ApiKey, AuthInterceptor}; #[doc(inline)] -pub use client::{AlarmFeedStream, EventStream, GatewayClient}; +pub use client::{AlarmFeedStream, EventItem, EventStream, GatewayClient}; #[doc(inline)] pub use error::{CommandError, Error, MxAccessError}; #[doc(inline)] pub use galaxy::{DeployEventStream, GalaxyClient}; #[doc(inline)] +pub use generated::mxaccess_gateway::v1::ReplayGap; +#[doc(inline)] pub use options::ClientOptions; #[doc(inline)] pub use session::{next_correlation_id, Session}; diff --git a/clients/rust/src/session.rs b/clients/rust/src/session.rs index bd04339..63f8189 100644 --- a/clients/rust/src/session.rs +++ b/clients/rust/src/session.rs @@ -630,6 +630,13 @@ impl Session { /// Open the per-session event stream from the beginning. /// + /// The returned [`EventStream`] yields [`EventItem`](crate::EventItem) + /// values — normal MXAccess events as + /// [`EventItem::Event`](crate::EventItem::Event). A stream opened from the + /// beginning never produces a + /// [`EventItem::ReplayGap`](crate::EventItem::ReplayGap); that sentinel + /// only appears on a resumed stream (see [`Session::events_after`]). + /// /// # Errors /// /// Returns the `tonic::Status` mapped through [`Error::from`] when the @@ -642,6 +649,15 @@ impl Session { /// `worker_sequence` is greater than `after_worker_sequence`. Pass `0` /// to receive every buffered event. /// + /// If `after_worker_sequence` predates the oldest event still retained in + /// the gateway's replay ring, the stream opens with a single + /// [`EventItem::ReplayGap`](crate::EventItem::ReplayGap) sentinel: recent + /// history was evicted and cannot be replayed, so the caller must discard + /// any local state and re-snapshot. To resume without provoking another + /// gap, call this method again with + /// `gap.oldest_available_sequence - 1`. See + /// [`EventItem`](crate::EventItem) for the full contract. + /// /// # Errors /// /// Same conditions as [`Session::events`]. diff --git a/clients/rust/tests/client_behavior.rs b/clients/rust/tests/client_behavior.rs index ae3f3d7..477fa81 100644 --- a/clients/rust/tests/client_behavior.rs +++ b/clients/rust/tests/client_behavior.rs @@ -27,13 +27,13 @@ use zb_mom_ww_mxgateway_client::generated::mxaccess_gateway::v1::{ CloseSessionReply, CloseSessionRequest, MxCommandKind, MxCommandReply, MxDataType, MxEvent, MxEventFamily, MxSparseArray, MxSparseElement, MxStatusCategory, MxStatusProxy, MxStatusSource, MxValue, OnAlarmTransitionEvent, OpenSessionReply, OpenSessionRequest, ProtocolStatus, - ProtocolStatusCode, QueryActiveAlarmsRequest, RegisterReply, SessionState, StreamAlarmsRequest, - StreamEventsRequest, SubscribeResult, Write2BulkEntry, WriteBulkEntry, WriteCommand, - WriteSecured2BulkEntry, WriteSecuredBulkEntry, + ProtocolStatusCode, QueryActiveAlarmsRequest, RegisterReply, ReplayGap, SessionState, + StreamAlarmsRequest, StreamEventsRequest, SubscribeResult, Write2BulkEntry, WriteBulkEntry, + WriteCommand, WriteSecured2BulkEntry, WriteSecuredBulkEntry, }; use zb_mom_ww_mxgateway_client::{ - next_correlation_id, ApiKey, ClientOptions, CommandError, Error, GatewayClient, MxStatus, - MxValue as ClientMxValue, MxValueProjection, + next_correlation_id, ApiKey, ClientOptions, CommandError, Error, EventItem, GatewayClient, + MxStatus, MxValue as ClientMxValue, MxValueProjection, }; #[tokio::test] @@ -128,8 +128,28 @@ async fn event_stream_preserves_order_and_drop_cancels_server_stream() { .await .unwrap(); - assert_eq!(stream.next().await.unwrap().unwrap().worker_sequence, 1); - assert_eq!(stream.next().await.unwrap().unwrap().worker_sequence, 2); + assert_eq!( + stream + .next() + .await + .unwrap() + .unwrap() + .as_event() + .unwrap() + .worker_sequence, + 1 + ); + assert_eq!( + stream + .next() + .await + .unwrap() + .unwrap() + .as_event() + .unwrap() + .worker_sequence, + 2 + ); drop(stream); for _ in 0..20 { @@ -142,6 +162,55 @@ async fn event_stream_preserves_order_and_drop_cancels_server_stream() { assert!(state.stream_dropped.load(Ordering::SeqCst)); } +#[tokio::test] +async fn replay_gap_sentinel_surfaces_as_typed_event_item() { + let state = Arc::new(FakeState::default()); + // Script a resumed stream: the reconnect-replay gap sentinel at the head + // (family UNSPECIFIED, no body, `replay_gap` set) followed by a normal + // event. The client must promote the sentinel to `EventItem::ReplayGap` + // and leave the following event as a normal `EventItem::Event`. + *state.stream_events_script.lock().await = Some(vec![ + MxEvent { + replay_gap: Some(ReplayGap { + requested_after_sequence: 5, + oldest_available_sequence: 42, + }), + ..MxEvent::default() + }, + event(42), + ]); + let endpoint = spawn_fake_gateway(state.clone()).await; + let client = GatewayClient::connect(ClientOptions::new(endpoint)) + .await + .unwrap(); + + let mut stream = client + .stream_events(StreamEventsRequest { + session_id: "session-fixture".to_owned(), + after_worker_sequence: 5, + }) + .await + .unwrap(); + + // First item is the typed gap sentinel, not a normal event. + let first = stream.next().await.unwrap().unwrap(); + match &first { + EventItem::ReplayGap(gap) => { + assert_eq!(gap.requested_after_sequence, 5); + assert_eq!(gap.oldest_available_sequence, 42); + } + EventItem::Event(_) => panic!("expected a ReplayGap sentinel, got a normal event"), + } + // Accessor helpers reflect the variant. + assert!(first.as_event().is_none()); + assert_eq!(first.replay_gap().unwrap().oldest_available_sequence, 42); + + // The normal event that follows is unaffected. + let second = stream.next().await.unwrap().unwrap(); + assert_eq!(second.as_event().unwrap().worker_sequence, 42); + assert!(second.replay_gap().is_none()); +} + #[tokio::test] async fn acknowledge_alarm_returns_reply_with_native_status() { let state = Arc::new(FakeState::default()); @@ -672,6 +741,11 @@ struct FakeState { /// handler to emit a synthetic ConditionRefresh -> snapshot_complete /// -> transition sequence. stream_alarms_script: Mutex>>, + /// Optional per-test override that pins the fake's `StreamEvents` + /// handler to emit a scripted `MxEvent` sequence (e.g. a `replay_gap` + /// sentinel followed by a normal event). When `None`, the handler falls + /// back to the default `event(1)` / `event(2)` pair. + stream_events_script: Mutex>>, } /// Per-test override for the fake's `Invoke` handler. @@ -921,9 +995,12 @@ impl MxAccessGateway for FakeGateway { &self, _request: Request, ) -> Result, Status> { - let (sender, receiver) = mpsc::channel(4); - sender.send(Ok(event(1))).await.unwrap(); - sender.send(Ok(event(2))).await.unwrap(); + let script = self.state.stream_events_script.lock().await.take(); + let events = script.unwrap_or_else(|| vec![event(1), event(2)]); + let (sender, receiver) = mpsc::channel(events.len().max(1)); + for event in events { + sender.send(Ok(event)).await.unwrap(); + } Ok(Response::new(DropAwareStream { inner: ReceiverStream::new(receiver), diff --git a/docs/ClientLibrariesDesign.md b/docs/ClientLibrariesDesign.md index c29fb1d..d6f9326 100644 --- a/docs/ClientLibrariesDesign.md +++ b/docs/ClientLibrariesDesign.md @@ -63,12 +63,29 @@ Goals: Non-goals for v1: -- client-side reconnectable sessions, -- client-side event replay, +- automatic client-side session reconnection (the library never transparently + re-opens a dropped stream on the consumer's behalf — reconnect timing and + policy stay an application concern), +- client-side event replay buffering (the gateway owns the replay ring; the + client does not retain its own event history), - client-side command batching, - synthetic MXAccess events, - hiding MXAccess handles behind opaque client-only handles. +The gateway's reconnect-replay *protocol* is, however, consumable from every +client: each exposes the `after_worker_sequence` resume cursor on +`StreamEvents` and surfaces the gateway's `ReplayGap` sentinel as a distinct, +typed, non-terminal signal (CLI-15) so a consumer can detect an evicted-history +gap and re-snapshot. Per the no-synthesized-events invariant the client never +fabricates or swallows the sentinel — it only makes the gateway's own signal +observable. The reaction contract is identical across languages: on a gap, +discard cached state, re-snapshot, and resume with +`after_worker_sequence = oldest_available_sequence - 1`. The per-language +surface follows each idiom (.NET `MxEventStreamItem.IsReplayGap` via +`StreamEventItemsAsync`; Go `EventResult.ReplayGap`/`IsReplayGap()`; Rust +`EventItem::ReplayGap`; Python yields a typed `ReplayGap` from +`stream_events`; Java — pending, tracked with the JDK-17/windows client pass). + ## Public Client Concepts All languages should expose the same core concepts, using idiomatic naming: diff --git a/docs/CrossLanguageSmokeMatrix.md b/docs/CrossLanguageSmokeMatrix.md index e6acb2d..af7c374 100644 --- a/docs/CrossLanguageSmokeMatrix.md +++ b/docs/CrossLanguageSmokeMatrix.md @@ -30,6 +30,14 @@ Each client entry defines commands for the same required operation sequence: The optional `write` command is documented separately because writing changes provider state and should only run when the operator supplies a safe test value. +When `stream-events` is resumed with an `after_worker_sequence` cursor that +predates the oldest event still in the gateway's replay ring, the gateway emits a +single `ReplayGap` sentinel at the head of the stream. Every client surfaces this +as a distinct, typed, non-terminal signal (see each client README); the resume +contract is `after_worker_sequence = oldest_available_sequence - 1`. The default +smoke sequence opens a fresh stream (no cursor) and does not exercise the gap +path; a resume-with-gap fixture case is tracked separately (TST-24). + ## Integration Gate Cross-language smoke execution is opt-in. Runners should skip the matrix unless