diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GatewayGalaxyStreamAttachedAckLiveTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GatewayGalaxyStreamAttachedAckLiveTests.cs new file mode 100644 index 000000000..4a15e2ef5 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GatewayGalaxyStreamAttachedAckLiveTests.cs @@ -0,0 +1,382 @@ +using System.Diagnostics.Metrics; +using System.Runtime.CompilerServices; +using Microsoft.Extensions.Logging; +using Shouldly; +using Xunit; +using ZB.MOM.WW.MxGateway.Client; +using ZB.MOM.WW.MxGateway.Contracts; +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 live gate — the stream-attached acknowledgement (mxaccessgw#227) against a REAL gateway. The unit +/// suite () proves the driver logic over a fake; what it cannot prove is +/// that the request flag reaches a real gateway and the sentinel comes back. These do: +/// +/// the live session advertises stream-events-attached-ack and the production +/// reports it; +/// the gateway sends the sentinel first when asked, and NEVER when not asked — the +/// control that makes the driver-level assertions falsifiable; +/// through the real + : the +/// subscriber's stream carries exactly one sentinel (so it requested it), the pump's attached signal +/// completes BEFORE the first SubscribeBulk is issued, the sentinel is never dispatched or counted as +/// a received event, and a static attribute's initial value then arrives; +/// control: a subscriber that does not request the ack gets no sentinel, the pump +/// never reports one, and the driver still delivers through the old best-effort wait. +/// +/// +/// Skip-gated like the other gateway live tests: runs only when MXGW_ENDPOINT + +/// GALAXY_MXGW_API_KEY are set. Needs a gateway at mxaccessgw >= afd504d5. Captured +/// 2026-11-12 against windev 10.100.0.48:5120 (mxaccessgw 447070a5). Read-only: subscribes +/// to TestMachine_002.Tagname, a static string attribute, so the value can only come from the +/// advise's initial snapshot (no concurrent writer can supply it, unlike TestFloat). +/// +/// +[Trait("Category", "Integration")] +public sealed class GatewayGalaxyStreamAttachedAckLiveTests +{ + /// A static, read-only attribute: its only event after an advise is the initial value. + private const string StaticRef = "TestMachine_002.Tagname"; + + private const string ExpectedStaticValue = "TestMachine_002"; + + [Fact] + public async Task Live_session_advertises_the_attach_ack_capability() + { + var (endpoint, apiKey) = RequireLiveGatewayOrSkip(); + var ct = TestContext.Current.CancellationToken; + + await using var session = new GalaxyMxSession(new GalaxyMxAccessOptions(ClientName: "OtOpcUa877Caps")); + await session.ConnectAsync(BuildClientOptions(endpoint, apiKey), ct); + + var live = session.Session.ShouldNotBeNull(); + live.OpenSessionReply.Capabilities.ShouldContain(GatewayCapabilities.StreamEventsAttachedAck, + $"the gateway must advertise the capability; got [{string.Join(", ", live.OpenSessionReply.Capabilities)}]"); + live.SupportsStreamAttachedAck.ShouldBeTrue(); + new GatewayGalaxySubscriber(session).SupportsStreamAttachedAck.ShouldBeTrue( + "the production subscriber reads the capability from the live session"); + } + + /// + /// Raw-stream contract, both ways on one gateway. Requested: the first message is the sentinel + /// (no family, no body, worker_sequence 0, live_after 0 on a fresh stream). Not requested: no sentinel + /// at all — and a data event (the advise's initial value) is the positive control that the stream was + /// live long enough for a sentinel to have shown up had the gateway sent one. + /// + [Fact] + public async Task Live_gateway_sends_the_sentinel_first_when_requested_and_never_otherwise() + { + var (endpoint, apiKey) = RequireLiveGatewayOrSkip(); + var ct = TestContext.Current.CancellationToken; + var clientOptions = BuildClientOptions(endpoint, apiKey); + + // Requested. + await using (var session = new GalaxyMxSession(new GalaxyMxAccessOptions(ClientName: "OtOpcUa877RawOn"))) + { + await session.ConnectAsync(clientOptions, ct); + using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct); + cts.CancelAfter(TimeSpan.FromSeconds(20)); + await using var e = session.Session!.StreamEventsAsync(0, emitStreamAttached: true, cts.Token) + .GetAsyncEnumerator(cts.Token); + (await e.MoveNextAsync()).ShouldBeTrue("a requested stream must yield the sentinel first"); + var first = e.Current; + first.StreamAttached.ShouldNotBeNull($"the first message must be the attach sentinel; got {first}"); + first.Family.ShouldBe(MxEventFamily.Unspecified); + first.WorkerSequence.ShouldBe(0ul, "the sentinel must never look like a resume cursor"); + first.StreamAttached.LiveAfterSequence.ShouldBe(0ul, "a fresh stream filters live events at 0"); + await cts.CancelAsync(); + } + + // Not requested (control). + await using (var session = new GalaxyMxSession(new GalaxyMxAccessOptions(ClientName: "OtOpcUa877RawOff"))) + { + await session.ConnectAsync(clientOptions, ct); + var seen = new List(); + var firstData = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct); + var pump = Task.Run(async () => + { + try + { + await foreach (var ev in session.Session!.StreamEventsAsync(0, emitStreamAttached: false, cts.Token)) + { + lock (seen) seen.Add(ev); + if (ev.Family == MxEventFamily.OnDataChange) firstData.TrySetResult(); + } + } + catch (Exception) when (cts.IsCancellationRequested) { } + }, ct); + + // No ack to wait on here (that is the point): give the attach ample time before advising. + await Task.Delay(TimeSpan.FromSeconds(2), ct); + await new GatewayGalaxySubscriber(session).SubscribeBulkAsync([StaticRef], 0, ct); + await TestWait.ForAsync(firstData.Task, "positive control: the unrequested stream still carries data", + () => Describe(seen)); + await cts.CancelAsync(); + await pump; + + lock (seen) + { + seen.ShouldNotContain(ev => ev.StreamAttached != null, + $"the gateway must not send the sentinel unless asked; {Describe(seen)}"); + } + } + } + + /// + /// The production path end to end: real subscriber, real driver, real pump. Ordering is recorded by a + /// pass-through decorator, so "ack before advise" is observed, not inferred. The production 1 s + /// budget is kept, and the driver must not log the + /// missing-ack warning — so this also fails if the ack arrived only after the budget ran out. + /// + [Fact] + public async Task Live_driver_waits_for_the_ack_before_advising_and_receives_the_initial_value() + { + var (endpoint, apiKey) = RequireLiveGatewayOrSkip(); + var ct = TestContext.Current.CancellationToken; + const string clientName = "OtOpcUa877Driver"; + + using var meters = new PumpMeters(clientName); + await using var session = new GalaxyMxSession(new GalaxyMxAccessOptions(ClientName: clientName)); + await session.ConnectAsync(BuildClientOptions(endpoint, apiKey), ct); + var recorder = new RecordingSubscriber(new GatewayGalaxySubscriber(session)); + var log = new CapturingLogger(); + + await using var driver = new GalaxyDriver("live-877", Opts(endpoint, clientName), hierarchySource: null, + subscriber: recorder, logger: new TypedLogger(log)); + driver.StreamRequestWait.ShouldBe(TimeSpan.FromSeconds(1), "exercise the production budget, not a test one"); + await driver.InitializeAsync("{}", ct); + + var dispatched = new List(); + var initial = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + driver.OnDataChange += (_, e) => + { + lock (dispatched) dispatched.Add(e); + if (string.Equals(e.FullReference, StaticRef, StringComparison.OrdinalIgnoreCase)) initial.TrySetResult(e); + }; + + await driver.SubscribeAsync([StaticRef], TimeSpan.Zero, ct); + var value = await TestWait.ForAsync(initial.Task, + "the advise's initial value must arrive (the static attribute has no other event)", recorder.Describe); + + // 1. The production subscriber requested the ack: the gateway only sends it when asked (raw control). + recorder.SentinelCount.ShouldBe(1, $"exactly one sentinel on the stream; {recorder.Describe()}"); + + // 2. Ack before the first advise — observed on the wire order the decorator saw. + var calls = recorder.Snapshot(); + var ackAt = Array.FindIndex(calls, c => c.StartsWith("ack", StringComparison.Ordinal)); + var subscribeAt = Array.FindIndex(calls, c => c.StartsWith("subscribe:", StringComparison.Ordinal)); + calls[0].ShouldBe("stream", recorder.Describe()); + ackAt.ShouldBeGreaterThan(0, recorder.Describe()); + subscribeAt.ShouldBeGreaterThan(ackAt, $"the advise must follow the attach ack; {recorder.Describe()}"); + + // 3. The pump consumed the ack as a control signal. + var pump = (EventPump)driver.EventPumpInstanceForTests!; + pump.AttachAckExpected.ShouldBeTrue(); + pump.StreamAttachedReceived.ShouldBeTrue(); + pump.StreamAttached.IsCompletedSuccessfully.ShouldBeTrue(); + + // 4. Never dispatched, never counted as received. + lock (dispatched) + { + dispatched.ShouldAllBe(e => string.Equals(e.FullReference, StaticRef, StringComparison.OrdinalIgnoreCase), + "only the subscribed attribute may be dispatched"); + dispatched.Count.ShouldBeLessThanOrEqualTo(recorder.DataEventCount, + "every dispatch must come from a real data event, never the sentinel"); + } + await TestWait.UntilAsync(() => meters.StreamAttached == 1 && meters.Received == recorder.NonSentinelCount, + "the pump counts the ack once on its own counter and never as a received event", + () => $"stream_attached={meters.StreamAttached} received={meters.Received} nonSentinel={recorder.NonSentinelCount}"); + + // 5. The value is the galaxy's, and the budget was not exhausted. + value.Snapshot.Value.ShouldBe(ExpectedStaticValue); + lock (log.Entries) + { + log.Entries.ShouldNotContain(e => e.Text.Contains("attach acknowledgement", StringComparison.Ordinal), + $"the ack must arrive within the production budget; log=\n{log.AllText}"); + } + + TestContext.Current.SendDiagnosticMessage($"#877 live: {recorder.Describe()} value={value.Snapshot.Value}"); + } + + /// + /// Control for the driver test: the same driver over a subscriber that does NOT request the ack (as + /// against an older gateway). The ack-specific assertions above must fail here — no sentinel, no + /// attached signal — while the driver still works through the #865 best-effort wait. + /// + [Fact] + public async Task Live_control_without_the_request_no_ack_arrives_and_the_fallback_still_delivers() + { + var (endpoint, apiKey) = RequireLiveGatewayOrSkip(); + var ct = TestContext.Current.CancellationToken; + const string clientName = "OtOpcUa877NoAck"; + + using var meters = new PumpMeters(clientName); + await using var session = new GalaxyMxSession(new GalaxyMxAccessOptions(ClientName: clientName)); + await session.ConnectAsync(BuildClientOptions(endpoint, apiKey), ct); + session.Session!.SupportsStreamAttachedAck.ShouldBeTrue("the gateway does support it; this client opts out"); + var recorder = new RecordingSubscriber(new GatewayGalaxySubscriber(session), optOutSession: session); + + await using var driver = new GalaxyDriver("live-877-noack", Opts(endpoint, clientName), hierarchySource: null, + subscriber: recorder); + await driver.InitializeAsync("{}", ct); + var initial = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + driver.OnDataChange += (_, e) => initial.TrySetResult(e); + + await driver.SubscribeAsync([StaticRef], TimeSpan.Zero, ct); + // Positive control first: data flowed, so the stream was live long enough to have carried a sentinel. + (await TestWait.ForAsync(initial.Task, "the fallback path still delivers the initial value", recorder.Describe)) + .Snapshot.Value.ShouldBe(ExpectedStaticValue); + + recorder.SentinelCount.ShouldBe(0, $"no ack without the request; {recorder.Describe()}"); + var pump = (EventPump)driver.EventPumpInstanceForTests!; + pump.AttachAckExpected.ShouldBeFalse(); + pump.StreamAttachedReceived.ShouldBeFalse(); + pump.StreamAttached.IsCompleted.ShouldBeFalse("no ack was requested and the stream is alive"); + meters.StreamAttached.ShouldBe(0); + } + + private static GalaxyDriverOptions Opts(string endpoint, string clientName) => new( + new GalaxyGatewayOptions(endpoint, "unused-seam-path", UseTls: false), + new GalaxyMxAccessOptions(clientName), + new GalaxyRepositoryOptions(WatchDeployEvents: false), + new GalaxyReconnectOptions()); + + private static (string Endpoint, string ApiKey) RequireLiveGatewayOrSkip() + { + var endpoint = Environment.GetEnvironmentVariable("MXGW_ENDPOINT"); + var apiKey = Environment.GetEnvironmentVariable("GALAXY_MXGW_API_KEY"); + if (string.IsNullOrWhiteSpace(endpoint) || string.IsNullOrWhiteSpace(apiKey)) + Assert.Skip("Set MXGW_ENDPOINT + GALAXY_MXGW_API_KEY to run the live stream-attached-ack gate."); + return (endpoint!, apiKey!); + } + + private static MxGatewayClientOptions BuildClientOptions(string endpoint, string apiKey) => new() + { + Endpoint = new Uri(endpoint, UriKind.Absolute), + ApiKey = apiKey, + UseTls = false, + ConnectTimeout = TimeSpan.FromSeconds(10), + DefaultCallTimeout = TimeSpan.FromSeconds(30), + StreamTimeout = TimeSpan.FromSeconds(30), + }; + + private static string Describe(List seen) + { + lock (seen) return $"events=[{string.Join(" | ", seen.Select(e => e.StreamAttached != null ? "ack" : e.Family.ToString()))}]"; + } + + /// + /// Pass-through that records the order of stream request, sentinel + /// arrival and SubscribeBulk. With optOutSession set it opens the stream WITHOUT requesting the + /// ack and reports no capability — an older-gateway client against the same live gateway. + /// + private sealed class RecordingSubscriber(GatewayGalaxySubscriber inner, GalaxyMxSession? optOutSession = null) + : IGalaxySubscriber + { + private readonly Lock _gate = new(); + private readonly List _calls = []; + private int _sentinels; + private int _nonSentinels; + private int _dataEvents; + + public int SentinelCount => Volatile.Read(ref _sentinels); + + public int NonSentinelCount => Volatile.Read(ref _nonSentinels); + + public int DataEventCount => Volatile.Read(ref _dataEvents); + + public bool SupportsStreamAttachedAck => optOutSession is null && inner.SupportsStreamAttachedAck; + + public string[] Snapshot() { lock (_gate) return [.. _calls]; } + + public string Describe() => $"calls=[{string.Join(" | ", Snapshot())}] sentinels={SentinelCount} nonSentinel={NonSentinelCount}"; + + private void Record(string call) { lock (_gate) _calls.Add(call); } + + public async Task> SubscribeBulkAsync( + IReadOnlyList fullReferences, int bufferedUpdateIntervalMs, CancellationToken cancellationToken) + { + Record("subscribe:" + string.Join(",", fullReferences)); + return await inner.SubscribeBulkAsync(fullReferences, bufferedUpdateIntervalMs, cancellationToken) + .ConfigureAwait(false); + } + + public Task UnsubscribeBulkAsync(IReadOnlyList itemHandles, CancellationToken cancellationToken) + => inner.UnsubscribeBulkAsync(itemHandles, cancellationToken); + + public async IAsyncEnumerable StreamEventsAsync([EnumeratorCancellation] CancellationToken cancellationToken) + { + Record("stream"); + var source = optOutSession is null + ? inner.StreamEventsAsync(cancellationToken) + : optOutSession.Session!.StreamEventsAsync(0, emitStreamAttached: false, cancellationToken); + await foreach (var ev in source.WithCancellation(cancellationToken).ConfigureAwait(false)) + { + if (ev.StreamAttached is not null) + { + Interlocked.Increment(ref _sentinels); + Record($"ack(seq={ev.WorkerSequence},liveAfter={ev.StreamAttached.LiveAfterSequence})"); + } + else + { + Interlocked.Increment(ref _nonSentinels); + if (ev.Family == MxEventFamily.OnDataChange) Interlocked.Increment(ref _dataEvents); + Record("event:" + ev.Family); + } + yield return ev; + } + } + } + + /// Reads the pump's received / stream_attached counters for one client name only. + private sealed class PumpMeters : IDisposable + { + private readonly MeterListener _listener = new(); + private long _received; + private long _attached; + + public PumpMeters(string clientName) + { + _listener.InstrumentPublished = (instrument, listener) => + { + if (instrument.Meter.Name == EventPump.MeterName + && instrument.Name is "galaxy.events.received" or "galaxy.events.stream_attached") + listener.EnableMeasurementEvents(instrument); + }; + _listener.SetMeasurementEventCallback((instrument, value, tags, _) => + { + foreach (var tag in tags) + { + if (tag.Key != "galaxy.client" || !Equals(tag.Value, clientName)) continue; + if (instrument.Name == "galaxy.events.received") Interlocked.Add(ref _received, value); + else Interlocked.Add(ref _attached, value); + } + }); + _listener.Start(); + } + + public long Received => Interlocked.Read(ref _received); + + public long StreamAttached => Interlocked.Read(ref _attached); + + public void Dispose() => _listener.Dispose(); + } + + 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); + } + } +}