diff --git a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/IMxGatewayClient.cs b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/IMxGatewayClient.cs index e91f0bae..0b1656e6 100644 --- a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/IMxGatewayClient.cs +++ b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/IMxGatewayClient.cs @@ -101,7 +101,9 @@ public interface IMxGatewayClient : IAsyncDisposable /// /// Long-running event consumer. Invokes for each advised-tag - /// data change. Resumes from the last delivered worker sequence on reconnect. Completes + /// data change. Re-invoking it on the same client resumes after the last real worker + /// sequence seen (gateway control sentinels never move that cursor; a replay gap is logged). + /// The adapter's reconnect builds a new client and session, which streams fresh. Completes /// (or throws) when the stream ends — the adapter treats that as a disconnect. /// /// Callback invoked per advised-tag value change. diff --git a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/RealMxGatewayClient.cs b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/RealMxGatewayClient.cs index f611f2f9..7e0f0261 100644 --- a/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/RealMxGatewayClient.cs +++ b/src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/RealMxGatewayClient.cs @@ -393,11 +393,53 @@ public sealed class RealMxGatewayClient : IMxGatewayClient } /// - public async Task RunEventLoopAsync(Action onUpdate, CancellationToken ct = default) + public Task RunEventLoopAsync(Action onUpdate, CancellationToken ct = default) + => RunEventLoopCoreAsync(_session!.StreamEventsAsync(_lastSeq, ct), onUpdate, ct); + + /// The worker sequence the next resumes after. + internal ulong ResumeCursor => _lastSeq; + + private void AdvanceResumeCursor(ulong sequence) { - await foreach (var ev in _session!.StreamEventsAsync(_lastSeq, ct).ConfigureAwait(false)) + if (sequence > _lastSeq) + _lastSeq = sequence; + } + + /// Consumes one event stream; split out so tests can drive a scripted stream. + /// + /// Resume-cursor rule (ScadaBridge#128): the cursor only ever advances, and only from + /// real worker events. The gateway's control sentinels — the ReplayGap sentinel on a + /// resume that hit evicted events, and the opt-in StreamAttached ack this adapter does not + /// request — carry worker_sequence 0; storing that would turn the next resume into + /// a fresh stream (no replay, no gap signal) and silently lose everything in between. + /// A ReplayGap anchors the cursor at oldest_available_sequence - 1, the gateway's + /// documented gap-free resume point (it also holds when the ring is empty, where the + /// gateway reports highest observed + 1), so a stream that drops before any + /// replayed or live event arrives does not hit — and re-report — the same gap again. + /// + internal async Task RunEventLoopCoreAsync( + IAsyncEnumerable events, Action onUpdate, CancellationToken ct) + { + await foreach (var ev in events.WithCancellation(ct).ConfigureAwait(false)) { - _lastSeq = ev.WorkerSequence; + if (ev.ReplayGap is { } gap) + { + // Events in (requested, oldest) were evicted from the gateway's replay ring and + // are unrecoverable: tags whose only change fell in that window keep a stale + // value until they next change. Surface it — never hide the loss. + _logger.LogWarning( + "MxGateway event stream resumed after worker sequence {RequestedAfterSequence} but the gateway only retains events from {OldestAvailableSequence}; the events in between were lost and subscribed values may be stale until they next change", + gap.RequestedAfterSequence, gap.OldestAvailableSequence); + if (gap.OldestAvailableSequence > 0) + AdvanceResumeCursor(gap.OldestAvailableSequence - 1); + continue; + } + + // A 0 sequence is a control sentinel (e.g. StreamAttached), never a worker event. + if (ev.WorkerSequence == 0) + continue; + + AdvanceResumeCursor(ev.WorkerSequence); if (ev.Family != MxEventFamily.OnDataChange) continue; if (!_handleToTag.TryGetValue(ev.ItemHandle, out var tag)) diff --git a/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/RealMxGatewayClientResumeCursorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/RealMxGatewayClientResumeCursorTests.cs new file mode 100644 index 00000000..a23ed330 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests/Adapters/RealMxGatewayClientResumeCursorTests.cs @@ -0,0 +1,191 @@ +using Microsoft.Extensions.Logging; +using ZB.MOM.WW.MxGateway.Contracts.Proto; +using ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Adapters; + +namespace ZB.MOM.WW.ScadaBridge.DataConnectionLayer.Tests.Adapters; + +/// +/// The StreamEvents resume cursor kept by (ScadaBridge#128). +/// The gateway emits control sentinels with worker_sequence 0 — the ReplayGap sentinel +/// on a resume that hit evicted events and (opt-in only) the StreamAttached ack. Neither is a +/// worker event, so neither may move the cursor; a ReplayGap instead anchors it at +/// oldest_available_sequence - 1, the gateway's documented resume formula. +/// +public class RealMxGatewayClientResumeCursorTests +{ + private static MxEvent DataChange(ulong seq, int itemHandle = 999) => new() + { + Family = MxEventFamily.OnDataChange, + WorkerSequence = seq, + ItemHandle = itemHandle, + }; + + private static MxEvent WriteComplete(ulong seq) => new() + { + Family = MxEventFamily.OnWriteComplete, + WorkerSequence = seq, + }; + + /// The gateway's ReplayGap sentinel: family UNSPECIFIED, worker_sequence 0. + private static MxEvent ReplayGapSentinel(ulong requestedAfter, ulong oldestAvailable) => new() + { + Family = MxEventFamily.Unspecified, + WorkerSequence = 0, + ReplayGap = new ReplayGap + { + RequestedAfterSequence = requestedAfter, + OldestAvailableSequence = oldestAvailable, + }, + }; + + /// + /// The StreamAttached ack as a 0.2.1-contract client decodes it: the + /// stream_attached field is unknown to that contract, so all that is visible is a + /// family-UNSPECIFIED event with worker_sequence 0. + /// + private static MxEvent StreamAttachedSentinel() => new() + { + Family = MxEventFamily.Unspecified, + WorkerSequence = 0, + }; + + private static async IAsyncEnumerable Stream(params MxEvent[] events) + { + foreach (var ev in events) + { + await Task.Yield(); + yield return ev; + } + } + + private static Task Run(RealMxGatewayClient client, params MxEvent[] events) => + client.RunEventLoopCoreAsync(Stream(events), _ => { }, CancellationToken.None); + + /// Real worker events of every family advance the cursor to the last sequence seen. + /// A task that represents the asynchronous operation. + [Fact] + public async Task RealEvents_AdvanceCursor() + { + var client = new RealMxGatewayClient(null); + + await Run(client, DataChange(1), WriteComplete(2), DataChange(3)); + + Assert.Equal(3UL, client.ResumeCursor); + } + + /// + /// The issue's case: a resume that hits a gap with nothing retained (age eviction or a + /// disabled ring) yields ONLY the ReplayGap sentinel. Resetting the cursor to 0 would turn + /// the next reconnect into a fresh stream — no replay and no gap signal. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task ReplayGapSentinel_AloneOnStream_AnchorsCursorAtOldestMinusOne() + { + var client = new RealMxGatewayClient(null); + await Run(client, DataChange(9), DataChange(10)); + + // Ring empty: the gateway reports the next deliverable sequence (highest observed + 1). + await Run(client, ReplayGapSentinel(requestedAfter: 10, oldestAvailable: 100)); + + Assert.Equal(99UL, client.ResumeCursor); + } + + /// A gap followed by the retained replay batch ends on the last replayed sequence. + /// A task that represents the asynchronous operation. + [Fact] + public async Task ReplayGapSentinel_FollowedByReplay_CursorEndsOnLastReplayedEvent() + { + var client = new RealMxGatewayClient(null); + await Run(client, DataChange(10)); + + await Run(client, ReplayGapSentinel(10, 50), DataChange(50), DataChange(51)); + + Assert.Equal(51UL, client.ResumeCursor); + } + + /// The (opt-in) StreamAttached ack carries worker_sequence 0 and must not move the cursor. + /// A task that represents the asynchronous operation. + [Fact] + public async Task StreamAttachedSentinel_DoesNotResetCursor() + { + var client = new RealMxGatewayClient(null); + await Run(client, DataChange(7)); + + await Run(client, StreamAttachedSentinel()); + + Assert.Equal(7UL, client.ResumeCursor); + } + + /// The cursor only ever advances: a lower sequence never moves it back. + /// A task that represents the asynchronous operation. + [Fact] + public async Task Cursor_NeverMovesBackwards() + { + var client = new RealMxGatewayClient(null); + + await Run(client, DataChange(20), DataChange(5), ReplayGapSentinel(1, 3)); + + Assert.Equal(20UL, client.ResumeCursor); + } + + /// + /// Events in the evicted interval are unrecoverable; the loss is surfaced as a warning + /// naming the interval, never hidden. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task ReplayGapSentinel_LogsWarningNamingTheLostInterval() + { + var provider = new CapturingLoggerProvider(); + using var factory = LoggerFactory.Create(b => b.AddProvider(provider)); + var client = new RealMxGatewayClient(factory); + + await Run(client, ReplayGapSentinel(10, 50)); + + var warning = Assert.Single(provider.Entries, e => e.Level == LogLevel.Warning); + Assert.Contains("10", warning.Message); + Assert.Contains("50", warning.Message); + } + + /// Sentinels never reach the value callback. + /// A task that represents the asynchronous operation. + [Fact] + public async Task Sentinels_AreNotDeliveredAsValueUpdates() + { + var client = new RealMxGatewayClient(null); + var updates = 0; + + await client.RunEventLoopCoreAsync( + Stream(StreamAttachedSentinel(), ReplayGapSentinel(1, 5)), _ => updates++, CancellationToken.None); + + Assert.Equal(0, updates); + } + + private sealed class CapturingLoggerProvider : ILoggerProvider + { + public List<(LogLevel Level, string Message)> Entries { get; } = new(); + + public ILogger CreateLogger(string categoryName) => new Capturing(this); + + public void Dispose() + { + } + + private sealed class Capturing(CapturingLoggerProvider owner) : ILogger + { + public IDisposable? BeginScope(TState state) + where TState : notnull => null; + + public bool IsEnabled(LogLevel logLevel) => true; + + public void Log( + LogLevel logLevel, EventId eventId, TState state, Exception? exception, + Func formatter) + { + lock (owner.Entries) + owner.Entries.Add((logLevel, formatter(state, exception))); + } + } + } +}