fix(dcl): MxGateway resume cursor ignores worker_sequence-0 sentinels (#128)

RealMxGatewayClient.RunEventLoopAsync stored every streamed MxEvent's
WorkerSequence as its StreamEvents resume cursor before filtering by family.
The gateway's control sentinels (ReplayGap on a resume that hit evicted
events; the opt-in StreamAttached ack) carry worker_sequence 0, so a gap
reset the cursor to 0 and the next resume became a fresh stream: no replay,
no gap signal, silent loss.

Cursor rule: advance-only (max), from real worker events (sequence > 0) only.
A ReplayGap anchors the cursor at oldest_available_sequence - 1 (the gateway's
documented gap-free resume point, valid for an empty ring too) and is logged
as a warning naming the lost interval. Sentinels never reach the value
callback.

The loop body is split into internal RunEventLoopCoreAsync(IAsyncEnumerable)
so tests drive a scripted stream; 4 of the 7 new tests fail on the old loop.

Claude-Session: https://claude.ai/code/session_01GT67ACmpMR7b9LGRPLijMi

src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/IMxGatewayClient.cs

@@ -101,7 +101,9 @@ public interface IMxGatewayClient : IAsyncDisposable
 
     /// <summary>
     /// Long-running event consumer. Invokes <paramref name="onUpdate"/> 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.
     /// </summary>
     /// <param name="onUpdate">Callback invoked per advised-tag value change.</param>

src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/RealMxGatewayClient.cs

@@ -393,11 +393,53 @@ public sealed class RealMxGatewayClient : IMxGatewayClient
     }
 
     /// <inheritdoc />
-    public async Task RunEventLoopAsync(Action<MxValueUpdate> onUpdate, CancellationToken ct = default)
+    public Task RunEventLoopAsync(Action<MxValueUpdate> onUpdate, CancellationToken ct = default)
+        => RunEventLoopCoreAsync(_session!.StreamEventsAsync(_lastSeq, ct), onUpdate, ct);
+
+    /// <summary>The worker sequence the next <see cref="RunEventLoopAsync"/> resumes after.</summary>
+    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;
+    }
+
+    /// <summary>Consumes one event stream; split out so tests can drive a scripted stream.</summary>
+    /// <remarks>
+    /// 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 <c>worker_sequence</c> 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 <c>oldest_available_sequence - 1</c>, the gateway's
+    /// documented gap-free resume point (it also holds when the ring is empty, where the
+    /// gateway reports <c>highest observed + 1</c>), so a stream that drops before any
+    /// replayed or live event arrives does not hit — and re-report — the same gap again.
+    /// </remarks>
+    internal async Task RunEventLoopCoreAsync(
+        IAsyncEnumerable<MxEvent> events, Action<MxValueUpdate> onUpdate, CancellationToken ct)
     {
-            _lastSeq = ev.WorkerSequence;
+        await foreach (var ev in events.WithCancellation(ct).ConfigureAwait(false))
+        {
+            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))

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;
+
+/// <summary>
+/// The StreamEvents resume cursor kept by <see cref="RealMxGatewayClient"/> (ScadaBridge#128).
+/// The gateway emits control sentinels with <c>worker_sequence</c> 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
+/// <c>oldest_available_sequence - 1</c>, the gateway's documented resume formula.
+/// </summary>
+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,
+    };
+
+    /// <summary>The gateway's ReplayGap sentinel: family UNSPECIFIED, worker_sequence 0.</summary>
+    private static MxEvent ReplayGapSentinel(ulong requestedAfter, ulong oldestAvailable) => new()
+    {
+        Family = MxEventFamily.Unspecified,
+        WorkerSequence = 0,
+        ReplayGap = new ReplayGap
+        {
+            RequestedAfterSequence = requestedAfter,
+            OldestAvailableSequence = oldestAvailable,
+        },
+    };
+
+    /// <summary>
+    /// The StreamAttached ack as a 0.2.1-contract client decodes it: the
+    /// <c>stream_attached</c> field is unknown to that contract, so all that is visible is a
+    /// family-UNSPECIFIED event with worker_sequence 0.
+    /// </summary>
+    private static MxEvent StreamAttachedSentinel() => new()
+    {
+        Family = MxEventFamily.Unspecified,
+        WorkerSequence = 0,
+    };
+
+    private static async IAsyncEnumerable<MxEvent> 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);
+
+    /// <summary>Real worker events of every family advance the cursor to the last sequence seen.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [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);
+    }
+
+    /// <summary>
+    /// 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.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [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);
+    }
+
+    /// <summary>A gap followed by the retained replay batch ends on the last replayed sequence.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [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);
+    }
+
+    /// <summary>The (opt-in) StreamAttached ack carries worker_sequence 0 and must not move the cursor.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [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);
+    }
+
+    /// <summary>The cursor only ever advances: a lower sequence never moves it back.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [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);
+    }
+
+    /// <summary>
+    /// Events in the evicted interval are unrecoverable; the loss is surfaced as a warning
+    /// naming the interval, never hidden.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [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);
+    }
+
+    /// <summary>Sentinels never reach the value callback.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [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>(TState state)
+                where TState : notnull => null;
+
+            public bool IsEnabled(LogLevel logLevel) => true;
+
+            public void Log<TState>(
+                LogLevel logLevel, EventId eventId, TState state, Exception? exception,
+                Func<TState, Exception?, string> formatter)
+            {
+                lock (owner.Entries)
+                    owner.Entries.Add((logLevel, formatter(state, exception)));
+            }
+        }
+    }
+}