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
This commit is contained in:
Joseph Doherty
@@ -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>
|
||||
|
||||
@@ -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)
|
||||
{
|
||||
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))
|
||||
|
||||
+191
@@ -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)));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user