dohertj2/lmxopcua
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases Wiki Activity

test(galaxy): live gate for the stream-attached ack against a real gateway (#877)
v2-ci / unit-tests (push) Failing after 10m42s
Details

Browse Source 
Adds GatewayGalaxyStreamAttachedAckLiveTests (skip-gated on MXGW_ENDPOINT +
GALAXY_MXGW_API_KEY, like the other gateway live tests). Proves what the unit
suite cannot: the request flag reaches a real gateway and the sentinel comes back.

- live session advertises stream-events-attached-ack; GatewayGalaxySubscriber reports it
- raw stream: sentinel is the first message when requested (family Unspecified,
  worker_sequence 0, live_after 0) and never sent when not requested (control,
  with a data event as the positive control)
- real GalaxyDriver + EventPump + GatewayGalaxySubscriber behind a pass-through
  recorder: exactly one sentinel, ack observed before the first SubscribeBulk,
  pump StreamAttached completed, sentinel never dispatched nor counted as
  received (galaxy.events.stream_attached=1, received == non-sentinel events),
  no missing-ack warning within the production 1 s budget, and the static
  attribute TestMachine_002.Tagname delivers its initial value
- control: a subscriber that opts out gets no sentinel and the pump never
  reports one, while the #865 fallback still delivers

Run 2026-11-12 against windev 10.100.0.48:5120 (mxaccessgw 447070a5): 4/4.
Mutation checks, each caught: subscriber not requesting the ack (sentinel
count 0); driver skipping the ack wait (wire order became stream | subscribe |
ack, 2/2 runs, so the race is real on a live gateway); pump not intercepting
the sentinel (received=2 vs 1 non-sentinel).

Claude-Session: https://claude.ai/code/session_01GT67ACmpMR7b9LGRPLijMi
This commit is contained in:
Joseph Doherty
2026-11-12 08:08:55 -05:00
parent f264977fb0
commit a28adba2fd
1 changed files with 382 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GatewayGalaxyStreamAttachedAckLiveTests.cs
+382
View File
@@ -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;
/// <summary>
/// #877 live gate — the stream-attached acknowledgement (mxaccessgw#227) against a REAL gateway. The unit
/// suite (<see cref="StreamAttachedAckTests"/>) 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:
/// <list type="number">
/// <item><description>the live session advertises <c>stream-events-attached-ack</c> and the production
/// <see cref="GatewayGalaxySubscriber"/> reports it;</description></item>
/// <item><description>the gateway sends the sentinel first when asked, and NEVER when not asked — the
/// control that makes the driver-level assertions falsifiable;</description></item>
/// <item><description>through the real <see cref="GalaxyDriver"/> + <see cref="EventPump"/>: 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;</description></item>
/// <item><description>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.</description></item>
/// </list>
/// <para>
/// Skip-gated like the other gateway live tests: runs only when <c>MXGW_ENDPOINT</c> +
/// <c>GALAXY_MXGW_API_KEY</c> are set. Needs a gateway at mxaccessgw &gt;= <c>afd504d5</c>. Captured
/// 2026-11-12 against windev <c>10.100.0.48:5120</c> (mxaccessgw <c>447070a5</c>). Read-only: subscribes
/// to <c>TestMachine_002.Tagname</c>, a static string attribute, so the value can only come from the
/// advise's initial snapshot (no concurrent writer can supply it, unlike <c>TestFloat</c>).
/// </para>
/// </summary>
[Trait("Category", "Integration")]
public sealed class GatewayGalaxyStreamAttachedAckLiveTests
{
/// <summary>A static, read-only attribute: its only event after an advise is the initial value.</summary>
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");
}
/// <summary>
/// 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.
/// </summary>
[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<MxEvent>();
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)}");
}
}
}
/// <summary>
/// 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
/// <see cref="GalaxyDriver.StreamRequestWait"/> 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.
/// </summary>
[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<DataChangeEventArgs>();
var initial = new TaskCompletionSource<DataChangeEventArgs>(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}");
}
/// <summary>
/// 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.
/// </summary>
[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<DataChangeEventArgs>(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<MxEvent> seen)
{
lock (seen) return $"events=[{string.Join(" | ", seen.Select(e => e.StreamAttached != null ? "ack" : e.Family.ToString()))}]";
}
/// <summary>
/// Pass-through <see cref="IGalaxySubscriber"/> that records the order of stream request, sentinel
/// arrival and SubscribeBulk. With <c>optOutSession</c> set it opens the stream WITHOUT requesting the
/// ack and reports no capability — an older-gateway client against the same live gateway.
/// </summary>
private sealed class RecordingSubscriber(GatewayGalaxySubscriber inner, GalaxyMxSession? optOutSession = null)
: IGalaxySubscriber
{
private readonly Lock _gate = new();
private readonly List<string> _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<IReadOnlyList<SubscribeResult>> SubscribeBulkAsync(
IReadOnlyList<string> fullReferences, int bufferedUpdateIntervalMs, CancellationToken cancellationToken)
{
Record("subscribe:" + string.Join(",", fullReferences));
return await inner.SubscribeBulkAsync(fullReferences, bufferedUpdateIntervalMs, cancellationToken)
.ConfigureAwait(false);
}
public Task UnsubscribeBulkAsync(IReadOnlyList<int> itemHandles, CancellationToken cancellationToken)
=> inner.UnsubscribeBulkAsync(itemHandles, cancellationToken);
public async IAsyncEnumerable<MxEvent> 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;
}
}
}
/// <summary>Reads the pump's received / stream_attached counters for one client name only.</summary>
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<long>((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<GalaxyDriver>
{
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 (inner.Entries) inner.Log(logLevel, eventId, state, exception, formatter);
}
}
}