diff --git a/Directory.Packages.props b/Directory.Packages.props index 19f5fa241..3f91892ed 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -230,8 +230,8 @@ - - + + diff --git a/nuget-packages/ZB.MOM.WW.MxGateway.Client.0.4.0.nupkg b/nuget-packages/ZB.MOM.WW.MxGateway.Client.0.4.0.nupkg new file mode 100644 index 000000000..82959d274 Binary files /dev/null and b/nuget-packages/ZB.MOM.WW.MxGateway.Client.0.4.0.nupkg differ diff --git a/nuget-packages/ZB.MOM.WW.MxGateway.Contracts.0.4.0.nupkg b/nuget-packages/ZB.MOM.WW.MxGateway.Contracts.0.4.0.nupkg new file mode 100644 index 000000000..641afa869 Binary files /dev/null and b/nuget-packages/ZB.MOM.WW.MxGateway.Contracts.0.4.0.nupkg differ diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs index 302d43020..a670da70a 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs @@ -2213,9 +2213,11 @@ public sealed class GalaxyDriver } /// - /// Upper bound on how long an advise waits for a current-session event stream (#865). Kept well under - /// Galaxy's Tier-A Read (2 s) / Subscribe (5 s) call budgets, so the "proceed anyway" fallback is - /// reachable before the resilience pipeline times the call out. + /// Upper bound on how long an advise waits for a current-session event stream (#865) — including, on a + /// gateway with the stream-attached ack, the ack itself (#877). Kept well under Galaxy's Tier-A Read (2 s) / + /// Subscribe (5 s) call budgets, so the "proceed anyway" fallback is reachable before the resilience + /// pipeline times the call out. The ack normally lands within one round trip, so a timeout on that path is + /// a fault worth a warning, not a routine fallback. /// internal TimeSpan StreamRequestWait { get; set; } = TimeSpan.FromSeconds(1); @@ -2227,9 +2229,14 @@ public sealed class GalaxyDriver /// /// A fresh StreamEvents (after_worker_sequence = 0) is never replayed, so an initial value pushed before /// the stream attaches is lost — and a slow-changing attribute (ScanState, a setpoint) may not publish - /// again for a long time. So: start the pump if there is none, and wait until its stream call has started - /// ( — this narrows, it cannot close, the attach race: the gateway - /// offers no attach acknowledgement). + /// again for a long time. So: start the pump if there is none, and wait until its stream is attached. + /// + /// + /// #877 — when the gateway advertised stream-events-attached-ack (), + /// "attached" is deterministic: the wait is on , the gateway's sentinel + /// that the stream is registered on the session, so the advise cannot lose its initial value. An older + /// gateway has no ack; the wait falls back to (the stream call has + /// started), which only narrows the attach race. /// /// /// A pump bound to an OLDER session belongs to a gateway session a reconnect replaced (the reopen → @@ -2284,6 +2291,22 @@ public sealed class GalaxyDriver try { await pump.StreamRequested.WaitAsync(remaining, cancellationToken).ConfigureAwait(false); + // AttachAckExpected is published before StreamRequested completes (#877). + if (pump.AttachAckExpected && !pump.HasEnded) + { + remaining = StreamRequestWait - elapsed.Elapsed; + if (remaining < TimeSpan.Zero) remaining = TimeSpan.Zero; + try + { + await pump.StreamAttached.WaitAsync(remaining, cancellationToken).ConfigureAwait(false); + } + catch (TimeoutException) + { + if (pump.HasEnded) ReportEndedPump(pump); + LogNoStreamAttachedAck(); + return session; + } + } if (pump.HasEnded) ReportEndedPump(pump); return session; } @@ -2292,8 +2315,10 @@ public sealed class GalaxyDriver // fall through to the log } } - else if (pump.StreamRequested.IsCompleted) + else if (pump.StreamRequested.IsCompleted + && (!pump.AttachAckExpected || pump.StreamAttached.IsCompleted)) { + if (pump.HasEnded) ReportEndedPump(pump); return session; } } @@ -2340,6 +2365,30 @@ public sealed class GalaxyDriver } private long _lastNoStreamWarningTicks; + private long _lastNoAckWarningTicks; + + /// + /// #877 — the gateway advertised the stream-attached ack but did not send it within + /// . Loud (Warning, once a minute) because the capability promises the + /// sentinel as the stream's first message: its absence means a slow/stuck attach or a gateway bug, and the + /// advise proceeding now may lose its initial value. + /// + private void LogNoStreamAttachedAck() + { + var now = Environment.TickCount64; + var last = Interlocked.Read(ref _lastNoAckWarningTicks); + if ((last == 0 || now - last >= 60_000) + && Interlocked.CompareExchange(ref _lastNoAckWarningTicks, now, last) == last) + { + _logger.LogWarning( + "GalaxyDriver {InstanceId}: the gateway advertises stream-events-attached-ack but its event stream sent no attach acknowledgement within {Wait} — advising anyway; initial values may be missed.", + _driverInstanceId, StreamRequestWait); + return; + } + _logger.LogDebug( + "GalaxyDriver {InstanceId} advising without the event stream's attach acknowledgement (warning rate-limited).", + _driverInstanceId); + } /// Warning at most once a minute while the supervisor is Healthy; Debug otherwise (an outage would flood). private void LogNoCurrentStream() diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs index 20a9a4597..4617cb2c6 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs @@ -51,6 +51,9 @@ internal sealed class EventPump : IAsyncDisposable private static readonly Counter ReplayGapsObserved = Meter.CreateCounter("galaxy.events.replay_gaps", unit: "{gap}", description: "Replay-gap sentinels received — the gateway evicted events we can never see."); + private static readonly Counter StreamAttachedAcks = + Meter.CreateCounter("galaxy.events.stream_attached", unit: "{ack}", + description: "Stream-attached sentinels received — the gateway confirmed the event stream is registered (#877)."); private static readonly Counter ValuesUndecodable = Meter.CreateCounter("galaxy.events.undecodable", unit: "{event}", description: "Data-change events carrying a value kind this driver cannot represent."); @@ -175,10 +178,38 @@ internal sealed class EventPump : IAsyncDisposable /// ⚠️ This NARROWS the stream-attach race, it does not close it: the HTTP/2 request is sent /// asynchronously and the gateway attaches the subscriber after auth, concurrently with any other /// RPC. In practice the attach wins (an advise's initial value needs a worker IPC round trip plus an - /// MXAccess advise), but the gateway offers no attach acknowledgement to await. + /// MXAccess advise). A gateway with the stream-events-attached-ack capability (mxaccessgw#227) + /// closes it: wait on when (#877). /// public Task StreamRequested => _streamRequested.Task; + private readonly TaskCompletionSource _streamAttached = new(TaskCreationOptions.RunContinuationsAsynchronously); + private volatile bool _attachAckExpected; + + /// + /// #877 — true when this pump's stream was requested WITH the gateway's stream-attached acknowledgement + /// (the subscriber reported when the stream was + /// opened). Set before completes, so a caller that awaited that first reads + /// the final value. + /// + public bool AttachAckExpected => _attachAckExpected; + + /// + /// #877 — completes when the gateway's stream-attached sentinel arrives: the stream is registered on the + /// gateway session, so every event the session publishes from then on (an advise's initial value + /// included) reaches this stream. This CLOSES the attach race that only + /// narrows. Also completes once the loop has ended, so a waiter never hangs on a dead pump — check + /// afterwards. Never completes for a stream opened without the ack + /// ( false) while it is alive: callers wait on it only when expected. + /// One pump owns exactly one stream; a reconnect replaces the pump, so every new stream starts with a + /// fresh, uncompleted signal. + /// + public Task StreamAttached => _streamAttached.Task; + + /// True once the stream-attached sentinel itself arrived (not merely the loop ending). + internal bool StreamAttachedReceived => Volatile.Read(ref _attachAckReceived); + private bool _attachAckReceived; + /// /// Marker written into the channel to wake the dispatch loop after a registration promoted /// parked early events (#859). Never dispatched itself; compared by reference. @@ -213,6 +244,9 @@ internal sealed class EventPump : IAsyncDisposable IAsyncEnumerator? enumerator = null; try { + // #877: read the capability BEFORE opening the stream — the subscriber requests the ack from the + // same session state — and publish it before StreamRequested releases a waiter. + _attachAckExpected = _subscriber.SupportsStreamAttachedAck; enumerator = _subscriber.StreamEventsAsync(ct).GetAsyncEnumerator(ct); // Start the first MoveNext BEFORE awaiting it: that synchronous step issues the RPC, which is // the moment StreamRequested reports (#865). @@ -234,6 +268,17 @@ internal sealed class EventPump : IAsyncDisposable { var ev = enumerator.Current; if (ct.IsCancellationRequested) break; + + // #877: the stream-attached sentinel is a control signal (family UNSPECIFIED, no body, + // worker_sequence 0). It is handled before anything else: it is never dispatched, never + // counted as a received/unknown event, and never touches a resume cursor (this driver + // keeps none — it always opens fresh streams — and a 0 sequence must never become one). + if (ev.StreamAttached is not null) + { + HandleStreamAttached(ev.StreamAttached); + continue; + } + EventsReceived.Add(1, _clientTag); // The replay-gap sentinel is a control signal, not data, and is handled HERE @@ -294,6 +339,8 @@ internal sealed class EventPump : IAsyncDisposable Volatile.Write(ref _ended, true); // A loop that died before issuing the RPC must not strand a StreamRequested waiter. _streamRequested.TrySetResult(); + // Nor one waiting for an attach acknowledgement that can no longer arrive (#877). + _streamAttached.TrySetResult(); if (enumerator is not null) { try { await enumerator.DisposeAsync().ConfigureAwait(false); } @@ -358,6 +405,25 @@ internal sealed class EventPump : IAsyncDisposable /// recreates the session, so a remembered sequence replayed against a fresh session /// would set the server's live watermark above every event it is about to send. /// + private void HandleStreamAttached(ZB.MOM.WW.MxGateway.Contracts.Proto.StreamAttached attached) + { + StreamAttachedAcks.Add(1, _clientTag); + if (!_attachAckExpected) + { + // An unrequested ack (capability read raced a session swap, or a gateway that sends it anyway) + // is harmless — still a control signal, still never dispatched. + _logger.LogDebug("Galaxy event stream sent a stream-attached acknowledgement that was not expected — ignoring it."); + } + else + { + _logger.LogDebug( + "Galaxy event stream attached on the gateway (live after worker_sequence {LiveAfter}).", + attached.LiveAfterSequence); + } + Volatile.Write(ref _attachAckReceived, true); + _streamAttached.TrySetResult(); + } + private void HandleReplayGap(ReplayGap gap) { ReplayGapsObserved.Add(1, _clientTag); diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/GatewayGalaxySubscriber.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/GatewayGalaxySubscriber.cs index 0d03d7ecc..d63a771a1 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/GatewayGalaxySubscriber.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/GatewayGalaxySubscriber.cs @@ -158,6 +158,15 @@ public sealed class GatewayGalaxySubscriber : IGalaxySubscriber var session = _session.Session ?? throw new InvalidOperationException( "GalaxyMxSession is not connected. StreamEventsAsync called before ConnectAsync."); - return session.StreamEventsAsync(afterWorkerSequence: 0, cancellationToken); + // #877: ask for the stream-attached sentinel only when the gateway advertised it — an older + // gateway ignores the field, so the pump would wait for an ack that never comes. + return session.StreamEventsAsync( + afterWorkerSequence: 0, + emitStreamAttached: session.SupportsStreamAttachedAck, + cancellationToken); } + + /// + /// Read from the live session's OpenSessionReply.capabilities; false while disconnected. + public bool SupportsStreamAttachedAck => _session.Session?.SupportsStreamAttachedAck ?? false; } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/IGalaxySubscriber.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/IGalaxySubscriber.cs index 427c902fa..9541496b6 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/IGalaxySubscriber.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/IGalaxySubscriber.cs @@ -38,4 +38,15 @@ public interface IGalaxySubscriber /// Cancellation token for the stream. /// An asynchronous stream of s as they arrive from the gateway. IAsyncEnumerable StreamEventsAsync(CancellationToken cancellationToken); + + /// + /// #877 — true when the stream returned by opens with the gateway's + /// stream-attached sentinel (, mxaccessgw#227): the session + /// advertised the stream-events-attached-ack capability and the subscriber requests the ack. + /// The sentinel means the stream is registered on the gateway session, so an advise issued after it + /// cannot lose its initial value to the attach race (#865). False (the default) for an older gateway + /// and for test fakes that do not model the ack — the then falls back to the + /// best-effort "stream requested" wait. + /// + bool SupportsStreamAttachedAck => false; } diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/TracedGalaxySubscriber.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/TracedGalaxySubscriber.cs index dac430f54..6e289f7dc 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/TracedGalaxySubscriber.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/TracedGalaxySubscriber.cs @@ -49,6 +49,9 @@ internal sealed class TracedGalaxySubscriber(IGalaxySubscriber inner, string cli } } + /// + public bool SupportsStreamAttachedAck => inner.SupportsStreamAttachedAck; + // Streaming RPC — one parent span covers the entire stream lifetime. Per-event // spans would dominate the trace volume at 50k tags / 1Hz; ops gets per-event // visibility through EventPump's metrics instead. diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/StreamAttachedAckTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/StreamAttachedAckTests.cs new file mode 100644 index 000000000..0bb99a9a5 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/StreamAttachedAckTests.cs @@ -0,0 +1,381 @@ +using System.Runtime.CompilerServices; +using System.Threading.Channels; +using Google.Protobuf.WellKnownTypes; +using Microsoft.Extensions.Logging; +using Shouldly; +using Xunit; +using ZB.MOM.WW.MxGateway.Contracts.Proto; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Config; +using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime; + +namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime; + +/// +/// #877 — the gateway's stream-attached acknowledgement (mxaccessgw#227). With the +/// stream-events-attached-ack capability, an advise waits until the gateway confirms the event +/// stream is REGISTERED on the session, closing the #865 attach race that the "stream requested" wait only +/// narrowed. Without it the old best-effort wait stays the fallback. +/// +/// +/// The fake models the race the real gateway has: a stream call is issued at once but the gateway +/// registers its subscriber only later (here: when the test opens the per-stream attach gate). An advise's +/// initial value pushed before that registration is lost, exactly as on the gateway — so an advise that +/// does not wait for the ack loses its value, and one that does cannot. +/// +public sealed class StreamAttachedAckTests +{ + private static GalaxyDriverOptions Opts() => new( + new GalaxyGatewayOptions("https://mxgw.test:5001", "key"), + new GalaxyMxAccessOptions("StreamAttachedAck"), + new GalaxyRepositoryOptions(WatchDeployEvents: false), + new GalaxyReconnectOptions()); + + private static MxEvent AckSentinel(ulong liveAfter = 0) => new() + { + // Shaped as the gateway emits it: no family, no body, worker_sequence 0. + Family = MxEventFamily.Unspecified, + StreamAttached = new StreamAttached { LiveAfterSequence = liveAfter }, + }; + + /// + /// Gateway-like subscriber whose stream registration is DELAYED until the test opens that stream's + /// attach gate. Records "stream:N", "attached:N" and "subscribe:<refs>" in order. + /// + private sealed class DelayedAttachSubscriber(bool supportsAck, bool sendsAck = true) : IGalaxySubscriber + { + private readonly Lock _gate = new(); + private readonly List _attachGates = []; + private Channel? _attached; + private int _nextHandle; + private int _streams; + + public bool SupportsStreamAttachedAck => supportsAck; + + public int DroppedInitialValues; + public volatile bool NextValue = true; + + public List Calls { get; } = []; + + public string[] CallsSnapshot() { lock (_gate) return [.. Calls]; } + + /// Lets stream N (1-based) register on the "gateway". + public void Attach(int stream) => GateFor(stream).TrySetResult(); + + private TaskCompletionSource GateFor(int stream) + { + lock (_gate) + { + while (_attachGates.Count < stream) + _attachGates.Add(new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously)); + return _attachGates[stream - 1]; + } + } + + public Task> SubscribeBulkAsync( + IReadOnlyList fullReferences, int bufferedUpdateIntervalMs, CancellationToken cancellationToken) + { + var results = new List(); + lock (_gate) + { + Calls.Add("subscribe:" + string.Join(",", fullReferences)); + foreach (var r in fullReferences) + { + var h = ++_nextHandle; + results.Add(new SubscribeResult { TagAddress = r, ItemHandle = h, WasSuccessful = true }); + var initial = new MxEvent + { + Family = MxEventFamily.OnDataChange, + ItemHandle = h, + WorkerSequence = (ulong)h, + Value = new MxValue { BoolValue = NextValue }, + Quality = 192, + SourceTimestamp = Timestamp.FromDateTime(DateTime.UtcNow), + }; + if (_attached is null || !_attached.Writer.TryWrite(initial)) Interlocked.Increment(ref DroppedInitialValues); + } + } + return Task.FromResult>(results); + } + + public Task UnsubscribeBulkAsync(IReadOnlyList itemHandles, CancellationToken cancellationToken) => Task.CompletedTask; + + public async IAsyncEnumerable StreamEventsAsync([EnumeratorCancellation] CancellationToken cancellationToken) + { + var n = Interlocked.Increment(ref _streams); + lock (_gate) Calls.Add("stream:" + n); + // The gateway registers the subscriber only after auth etc. — asynchronously to the call. + await GateFor(n).Task.WaitAsync(cancellationToken).ConfigureAwait(false); + var mine = Channel.CreateUnbounded(); + lock (_gate) + { + _attached = mine; + Calls.Add("attached:" + n); + } + try + { + // The gateway sends the sentinel only AFTER registration, and only when asked. + if (supportsAck && sendsAck) yield return AckSentinel(); + await foreach (var ev in mine.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false)) + { + yield return ev; + } + } + finally + { + lock (_gate) { if (ReferenceEquals(_attached, mine)) _attached = null; } + } + } + } + + private static string Describe(DelayedAttachSubscriber sub) => + $"calls=[{string.Join(" | ", sub.CallsSnapshot())}] dropped={sub.DroppedInitialValues}"; + + /// + /// The key test. Capability present: the subscribe is NOT issued while the stream is requested but not + /// yet registered; it goes out only after the ack, so its initial value arrives. Fail-before: with the + /// ack wait skipped (pre-#877 behaviour) the subscribe goes out at "stream requested", its initial value + /// is pushed to no stream, and this test fails on the ordering and the dropped count. + /// + [Fact] + public async Task Capability_present_subscribe_waits_for_the_ack_and_gets_its_initial_value() + { + var sub = new DelayedAttachSubscriber(supportsAck: true); + await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub) + { + StreamRequestWait = TimeSpan.FromSeconds(20), // generous: the gate below decides, not the budget + }; + await driver.InitializeAsync("{}", CancellationToken.None); + var delivered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + driver.OnDataChange += (_, e) => delivered.TrySetResult(e); + + var subscribe = driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None); + + await TestWait.UntilAsync(() => sub.CallsSnapshot().Contains("stream:1"), + "positive control: the subscribe must start the event stream", () => Describe(sub)); + // The stream is requested but not registered. Give a skipped wait every chance to advise now. + await Task.Delay(300, TestContext.Current.CancellationToken); + sub.CallsSnapshot().ShouldNotContain(c => c.StartsWith("subscribe:", StringComparison.Ordinal), + $"no advise may be issued before the gateway acknowledged the stream; {Describe(sub)}"); + subscribe.IsCompleted.ShouldBeFalse("the subscribe must still be waiting for the attach ack"); + + sub.Attach(1); + await TestWait.ForAsync(subscribe, "the subscribe proceeds once the ack arrives", () => Describe(sub)); + + var calls = sub.CallsSnapshot(); + Array.IndexOf(calls, "attached:1").ShouldBeLessThan( + Array.FindIndex(calls, c => c.StartsWith("subscribe:", StringComparison.Ordinal)), + $"the advise must follow the registration; {Describe(sub)}"); + (await TestWait.ForAsync(delivered.Task, "the initial value must arrive", () => Describe(sub))) + .FullReference.ShouldBe("Tank01.Level"); + sub.DroppedInitialValues.ShouldBe(0); + ((EventPump)driver.EventPumpInstanceForTests!).StreamAttachedReceived.ShouldBeTrue(); + } + + /// + /// Capability absent (an older gateway): no ack is requested or awaited — the advise proceeds as soon as + /// the stream call has started (the #865 best-effort wait), even though the stream is not registered yet. + /// That leaves the narrowed race in place (the value is dropped here, by construction of the fake); the + /// point is that the driver does not hang waiting for an ack an old gateway never sends. + /// + [Fact] + public async Task Capability_absent_falls_back_to_the_stream_requested_wait() + { + var sub = new DelayedAttachSubscriber(supportsAck: false); + await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub) + { + StreamRequestWait = TimeSpan.FromSeconds(20), + }; + await driver.InitializeAsync("{}", CancellationToken.None); + + var sw = System.Diagnostics.Stopwatch.StartNew(); + await TestWait.ForAsync( + driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None), + "without the capability the subscribe must not wait for an ack", () => Describe(sub)); + sw.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(10), "the 20 s budget must not have been spent on an ack"); + + var pump = (EventPump)driver.EventPumpInstanceForTests!; + pump.AttachAckExpected.ShouldBeFalse(); + pump.StreamAttached.IsCompleted.ShouldBeFalse("no ack was requested, none arrived"); + var calls = sub.CallsSnapshot(); + Array.IndexOf(calls, "stream:1").ShouldBe(0, $"the stream is still requested first; {Describe(sub)}"); + calls.ShouldContain(c => c.StartsWith("subscribe:", StringComparison.Ordinal)); + calls.ShouldNotContain("attached:1", "the subscribe went out before registration — the old, narrowed race"); + sub.Attach(1); // let the stream finish attaching so disposal is clean + } + + /// + /// Capability advertised but the ack never comes (a gateway bug or a stuck attach): the advise does not + /// hang — it proceeds after and logs a Warning naming the + /// missing acknowledgement. + /// + [Fact] + public async Task Missing_ack_times_out_with_a_warning_instead_of_hanging() + { + var sub = new DelayedAttachSubscriber(supportsAck: true, sendsAck: false); + var log = new CapturingLogger(); + await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub, + logger: new TypedLogger(log)) + { + StreamRequestWait = TimeSpan.FromMilliseconds(300), + }; + await driver.InitializeAsync("{}", CancellationToken.None); + sub.Attach(1); // registered, but no sentinel is sent + + await TestWait.ForAsync( + driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None), + "a missing ack must not hang the subscribe", () => Describe(sub)); + + lock (log.Entries) + { + log.Entries.ShouldContain(e => e.Level == LogLevel.Warning && e.Text.Contains("attach acknowledgement", StringComparison.Ordinal), + $"a missing ack must be logged loudly; log=\n{log.AllText}"); + } + sub.CallsSnapshot().ShouldContain(c => c.StartsWith("subscribe:", StringComparison.Ordinal)); + } + + /// + /// The sentinel is a control signal: never dispatched, never counted as a received event, and it + /// completes . Positive control: a real event after it IS dispatched. + /// + [Fact] + public async Task Ack_sentinel_is_never_dispatched() + { + var subscriber = new ChannelSubscriber(); + var registry = new SubscriptionRegistry(); + registry.Register(1, [new TagBinding("Tank.Level", ItemHandle: 7)]); + + var dispatched = 0; + var realValue = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + await using var pump = new EventPump(subscriber, registry, channelCapacity: 8, clientName: "AckTest"); + pump.OnDataChange += (_, args) => { Interlocked.Increment(ref dispatched); realValue.TrySetResult(args); }; + pump.Start(); + + // An ack shaped with OnDataChange + item handle 7 would be dispatched if it reached the channel — + // so this proves interception, not just "unknown family is dropped". + var ack = AckSentinel(liveAfter: 0); + ack.Family = MxEventFamily.OnDataChange; + ack.ItemHandle = 7; + ack.Value = new MxValue { DoubleValue = -1 }; + await subscriber.EmitAsync(ack); + await TestWait.ForAsync(pump.StreamAttached, "the ack must complete the attached signal"); + pump.StreamAttachedReceived.ShouldBeTrue(); + + await subscriber.EmitAsync(new MxEvent + { + Family = MxEventFamily.OnDataChange, + ItemHandle = 7, + Value = new MxValue { DoubleValue = 12.5 }, + Quality = 192, + SourceTimestamp = Timestamp.FromDateTime(DateTime.UtcNow), + }); + + (await TestWait.ForAsync(realValue.Task, "positive control: real events still dispatch")).Snapshot.Value.ShouldBe(12.5); + Volatile.Read(ref dispatched).ShouldBe(1, "only the real event may be dispatched, never the ack"); + } + + /// + /// A pump whose loop ends before any ack completes (so no waiter + /// hangs) WITHOUT reporting the ack as received. + /// + [Fact] + public async Task A_dead_stream_releases_ack_waiters_without_claiming_an_ack() + { + await using var pump = new EventPump( + new ThrowingSubscriber(), new SubscriptionRegistry(), onStreamFault: _ => { }); + pump.Start(); + await TestWait.ForAsync(pump.StreamAttached, "a dead pump must not strand an ack waiter"); + pump.HasEnded.ShouldBeTrue(); + pump.StreamAttachedReceived.ShouldBeFalse(); + pump.AttachAckExpected.ShouldBeTrue(); + } + + /// + /// Reconnect: the replay's new stream gets a FRESH attach signal — the first stream's ack does not + /// release it. The replay's re-advise waits for the second stream's ack, and its new initial value + /// arrives. Positive control: before stream 2 attaches, the re-advise has not been issued. + /// + [Fact] + public async Task Reconnect_gives_each_stream_a_fresh_signal() + { + var sub = new DelayedAttachSubscriber(supportsAck: true); + await using var driver = new GalaxyDriver("drv", Opts(), hierarchySource: null, subscriber: sub) + { + StreamRequestWait = TimeSpan.FromSeconds(20), + }; + await driver.InitializeAsync("{}", CancellationToken.None); + var last = new System.Collections.Concurrent.ConcurrentDictionary(); + driver.OnDataChange += (_, e) => last[e.FullReference] = e.Snapshot.Value; + + sub.Attach(1); + await driver.SubscribeAsync(["Tank01.Level"], TimeSpan.Zero, CancellationToken.None); + await TestWait.UntilAsync(() => last.TryGetValue("Tank01.Level", out var v) && v is true, + "positive control: the first stream delivers the initial value", () => Describe(sub)); + var pump1 = (EventPump)driver.EventPumpInstanceForTests!; + pump1.StreamAttached.IsCompleted.ShouldBeTrue(); + + sub.NextValue = false; + var replay = driver.InvokeReplayForTestAsync(CancellationToken.None); + await TestWait.UntilAsync(() => sub.CallsSnapshot().Contains("stream:2"), + "the replay must start a new stream", () => Describe(sub)); + var pump2 = (EventPump)driver.EventPumpInstanceForTests!; + pump2.ShouldNotBeSameAs(pump1); + pump2.StreamAttached.IsCompleted.ShouldBeFalse("a new stream starts with an uncompleted attach signal"); + + await Task.Delay(300, TestContext.Current.CancellationToken); + sub.CallsSnapshot().Count(c => c.StartsWith("subscribe:", StringComparison.Ordinal)) + .ShouldBe(1, $"the re-advise must wait for stream 2's ack; {Describe(sub)}"); + + sub.Attach(2); + await TestWait.ForAsync(replay, "the replay completes once stream 2 is acknowledged", () => Describe(sub)); + await TestWait.UntilAsync(() => last.TryGetValue("Tank01.Level", out var v) && v is false, + "the re-advised tag's new initial value must arrive on stream 2", () => Describe(sub)); + sub.DroppedInitialValues.ShouldBe(0); + } + + /// A stream fed directly by the test; reports the ack capability. + private sealed class ChannelSubscriber : IGalaxySubscriber + { + private readonly Channel _stream = Channel.CreateUnbounded(); + + public bool SupportsStreamAttachedAck => true; + + public Task> SubscribeBulkAsync( + IReadOnlyList fullReferences, int bufferedUpdateIntervalMs, CancellationToken cancellationToken) + => Task.FromResult>([]); + + public Task UnsubscribeBulkAsync(IReadOnlyList itemHandles, CancellationToken cancellationToken) => Task.CompletedTask; + + public IAsyncEnumerable StreamEventsAsync(CancellationToken cancellationToken) + => _stream.Reader.ReadAllAsync(cancellationToken); + + public ValueTask EmitAsync(MxEvent ev) => _stream.Writer.WriteAsync(ev); + } + + private sealed class ThrowingSubscriber : IGalaxySubscriber + { + public bool SupportsStreamAttachedAck => true; + + public Task> SubscribeBulkAsync( + IReadOnlyList fullReferences, int bufferedUpdateIntervalMs, CancellationToken cancellationToken) + => Task.FromResult>([]); + + public Task UnsubscribeBulkAsync(IReadOnlyList itemHandles, CancellationToken cancellationToken) => Task.CompletedTask; + + public IAsyncEnumerable StreamEventsAsync(CancellationToken cancellationToken) + => throw new InvalidOperationException("not connected"); + } + + private sealed class TypedLogger(CapturingLogger inner) : 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 (inner.Entries) inner.Log(logLevel, eventId, state, exception, formatter); + } + } +}