feat(galaxy): wire DiscoverPageSize and ProbeTimeoutSeconds; concurrent first-touch write lanes

Review 2026-10-21 G-5, G-6, G-8 ("wire them all"). - G-5: GatewayGalaxyHierarchySource now pages the DiscoverHierarchy RPC itself (DiscoverHierarchyRawAsync) at Repository.DiscoverPageSize. The mxaccessgw client's typed DiscoverHierarchyAsync overloads hard-code a 5000-row page and DiscoverHierarchyOptions carries no page size, so the authored knob reached nothing; the raw request takes page_size on the wire, so it is honoured here (non-positive → 5000). The loop keeps the client's repeated-page-token guard. - G-6: GalaxyDriverProbe reads ProbeTimeoutSeconds from the config and runs under min(caller budget, configured cap) — the knob can shorten a probe, never extend it past the host's budget. The TCP preflight now shares that deadline instead of hanging on the caller's token alone. - G-8: GatewayGalaxyDataWriter runs a batch as one lane per distinct tag, concurrently and bounded at 8 (MaxConcurrentWriteLanes); same-tag writes keep their authored order inside one lane and results land at their request's index. Acks stay serial by design. - GalaxyDriverForm: help text now says what the two knobs actually do. Refs #735 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
2026-10-22 06:41:47 -04:00
parent 1b3169273d
commit 6dedf5158f
8 changed files with 498 additions and 14 deletions
@@ -0,0 +1,92 @@
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.MxGateway.Contracts.Proto.Galaxy;
+using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Browse;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Browse;
+
+/// <summary>
+///     Review 2026-10-21 G-5: <c>Repository.DiscoverPageSize</c> was authored and parsed but reached
+///     nothing — the client's typed <c>DiscoverHierarchyAsync</c> hard-codes 5000. The source now pages
+///     the raw RPC itself at the configured size.
+/// </summary>
+public sealed class GatewayGalaxyHierarchySourcePagingTests
+{
+    private static GalaxyObject Obj(int id) => new() { GobjectId = id, TagName = $"Obj{id}" };
+
+    /// <summary>Every request carries the configured page size and the page tokens chain until the last page.</summary>
+    [Fact]
+    public async Task Pages_at_the_configured_size_and_chains_tokens()
+    {
+        var requests = new List<DiscoverHierarchyRequest>();
+        Task<DiscoverHierarchyReply> Discover(DiscoverHierarchyRequest req, CancellationToken ct)
+        {
+            requests.Add(req);
+            var reply = new DiscoverHierarchyReply();
+            switch (req.PageToken)
+            {
+                case "":
+                    reply.Objects.Add([Obj(1), Obj(2)]);
+                    reply.NextPageToken = "p2";
+                    break;
+                case "p2":
+                    reply.Objects.Add([Obj(3), Obj(4)]);
+                    reply.NextPageToken = "p3";
+                    break;
+                default:
+                    reply.Objects.Add(Obj(5));
+                    reply.NextPageToken = "";
+                    break;
+            }
+            return Task.FromResult(reply);
+        }
+
+        var objects = await GatewayGalaxyHierarchySource.PageAsync(Discover, pageSize: 2, CancellationToken.None);
+
+        objects.Select(o => o.GobjectId).ShouldBe([1, 2, 3, 4, 5]);
+        requests.Count.ShouldBe(3);
+        requests.ShouldAllBe(r => r.PageSize == 2);
+        requests.Select(r => r.PageToken).ShouldBe(["", "p2", "p3"]);
+    }
+
+    /// <summary>A different configured size is what goes on the wire (positive control for the knob).</summary>
+    [Fact]
+    public async Task A_different_page_size_is_sent_on_the_wire()
+    {
+        int? sent = null;
+        Task<DiscoverHierarchyReply> Discover(DiscoverHierarchyRequest req, CancellationToken ct)
+        {
+            sent = req.PageSize;
+            return Task.FromResult(new DiscoverHierarchyReply());
+        }
+
+        await GatewayGalaxyHierarchySource.PageAsync(Discover, pageSize: 250, CancellationToken.None);
+        sent.ShouldBe(250);
+    }
+
+    /// <summary>A non-positive configured size falls back to the client library's 5000 default.</summary>
+    [Theory]
+    [InlineData(0)]
+    [InlineData(-7)]
+    public void Non_positive_page_size_falls_back_to_the_default(int configured)
+        => GatewayGalaxyHierarchySource.EffectivePageSize(configured).ShouldBe(GatewayGalaxyHierarchySource.DefaultPageSize);
+
+    /// <summary>A repeated page token is refused rather than walked forever.</summary>
+    [Fact]
+    public async Task A_repeated_page_token_is_refused()
+    {
+        var calls = 0;
+        Task<DiscoverHierarchyReply> Discover(DiscoverHierarchyRequest req, CancellationToken ct)
+        {
+            calls++;
+            var reply = new DiscoverHierarchyReply { NextPageToken = "same" };
+            reply.Objects.Add(Obj(calls));
+            return Task.FromResult(reply);
+        }
+
+        var ex = await Should.ThrowAsync<InvalidOperationException>(
+            () => GatewayGalaxyHierarchySource.PageAsync(Discover, pageSize: 10, CancellationToken.None));
+        ex.Message.ShouldContain("repeated page token");
+        calls.ShouldBe(2);
+    }
+}
@@ -0,0 +1,69 @@
+using System.Diagnostics;
+using System.Net;
+using System.Net.Sockets;
+using Shouldly;
+using Xunit;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests;
+
+/// <summary>
+///     Review 2026-10-21 G-6: <c>ProbeTimeoutSeconds</c> was authored and compared for equivalence but
+///     never consumed — the probe ran on the caller's budget alone. It now caps the probe deadline
+///     (it can shorten the caller's budget, never extend it).
+/// </summary>
+[Trait("Category", "Unit")]
+public sealed class GalaxyDriverProbeTimeoutTests
+{
+    /// <summary>The effective deadline is the shorter of the caller's budget and the configured cap.</summary>
+    [Theory]
+    [InlineData(10, 30, 10)]   // caller shorter → caller wins
+    [InlineData(10, 2, 2)]     // config shorter → config caps
+    [InlineData(10, 0, 10)]    // unset config → caller
+    [InlineData(10, -1, 10)]   // nonsense config → caller
+    [InlineData(0, 5, 1)]      // no caller budget → 1 s floor beats a longer config
+    [InlineData(0, 0, 1)]      // nothing → 1 s floor
+    public void EffectiveTimeout_is_the_shorter_of_caller_and_config(int callerSeconds, int configSeconds, int expectedSeconds)
+        => GalaxyDriverProbe.EffectiveTimeout(TimeSpan.FromSeconds(callerSeconds), configSeconds)
+            .ShouldBe(TimeSpan.FromSeconds(expectedSeconds));
+
+    /// <summary>
+    ///     Against a port that accepts TCP but never speaks gRPC, a 1 s configured cap returns long
+    ///     before the caller's 20 s budget. The gRPC handshake itself does the waiting here, so this
+    ///     exercises the real deadline path, not the pure helper.
+    /// </summary>
+    [Fact]
+    public async Task Configured_cap_shortens_the_probe_against_a_silent_listener()
+    {
+        var listener = new TcpListener(IPAddress.Loopback, 0);
+        listener.Start();
+        try
+        {
+            var port = ((IPEndPoint)listener.LocalEndpoint).Port;
+            var config = $$"""
+                {
+                  "Gateway": { "Endpoint": "http://127.0.0.1:{{port}}", "ApiKeySecretRef": "dev:k", "UseTls": false },
+                  "MxAccess": { "ClientName": "probe-timeout" },
+                  "Repository": {},
+                  "Reconnect": {},
+                  "ProbeTimeoutSeconds": 1
+                }
+                """;
+
+            var sw = Stopwatch.StartNew();
+            var result = await new GalaxyDriverProbe().ProbeAsync(config, TimeSpan.FromSeconds(20), TestContext.Current.CancellationToken);
+            sw.Stop();
+
+            result.Ok.ShouldBeFalse();
+            sw.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(10), $"the 1 s configured cap must bound the probe (message: {result.Message})");
+            result.Message.ShouldNotBeNull();
+            (result.Message!.Contains("timed out", StringComparison.OrdinalIgnoreCase)
+             || result.Message.Contains("DeadlineExceeded", StringComparison.OrdinalIgnoreCase)
+             || result.Message.Contains("handshake failed", StringComparison.OrdinalIgnoreCase))
+                .ShouldBeTrue($"unexpected message: '{result.Message}'");
+        }
+        finally
+        {
+            listener.Stop();
+        }
+    }
+}
@@ -0,0 +1,168 @@
+using System.Collections.Concurrent;
+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.Runtime;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime;
+
+/// <summary>
+///     Review 2026-10-21 G-8: a batch of writes ran strictly serially, three round-trips per
+///     first-touch tag. Distinct tags now run as concurrent lanes bounded at
+///     <see cref="GatewayGalaxyDataWriter.MaxConcurrentWriteLanes"/>; same-tag writes keep their
+///     order inside one lane; results stay at their request's index.
+/// </summary>
+public sealed class GatewayGalaxyDataWriterConcurrencyTests
+{
+    private const int WaitMs = 10_000;
+
+    private static GatewayGalaxyDataWriter NewWriter()
+        => new(new GalaxyMxSession(new Config.GalaxyMxAccessOptions(ClientName: "OtOpcUa-Test")), writeUserId: 0);
+
+    /// <summary>
+    ///     Fake ops whose AddItem parks every caller until <see cref="ReleaseAt"/> callers are in
+    ///     flight at once (or a safety delay elapses), recording the peak concurrency.
+    /// </summary>
+    private sealed class GatingOps : IMxWriteOps
+    {
+        private int _inFlight;
+        private int _peak;
+        private int _nextHandle;
+        private readonly TaskCompletionSource _release = new(TaskCreationOptions.RunContinuationsAsynchronously);
+        private readonly ConcurrentDictionary<string, int> _handles = new(StringComparer.OrdinalIgnoreCase);
+
+        public int ReleaseAt { get; init; } = int.MaxValue;
+        public int Peak => Volatile.Read(ref _peak);
+        public ConcurrentQueue<(int Handle, MxValue Value)> RawWrites { get; } = new();
+        public ConcurrentQueue<string> AddItems { get; } = new();
+        public Func<int, MxCommandReply>? WriteResponder { get; init; }
+
+        /// <summary>Deterministic handle per tag when set; otherwise handles are assigned in AddItem completion order.</summary>
+        public Func<string, int>? HandleFor { get; init; }
+
+        public string SessionId => "sess";
+        public int ServerHandle => 1;
+
+        public async Task<int> AddItemAsync(string fullRef, CancellationToken ct)
+        {
+            AddItems.Enqueue(fullRef);
+            var now = Interlocked.Increment(ref _inFlight);
+            int seen;
+            do { seen = Volatile.Read(ref _peak); } while (now > seen && Interlocked.CompareExchange(ref _peak, now, seen) != seen);
+            if (now >= ReleaseAt) _release.TrySetResult();
+            await Task.WhenAny(_release.Task, Task.Delay(300, ct));
+            Interlocked.Decrement(ref _inFlight);
+            return _handles.GetOrAdd(fullRef, r => HandleFor?.Invoke(r) ?? Interlocked.Increment(ref _nextHandle));
+        }
+
+        public Task<MxCommandReply> InvokeAsync(MxCommandRequest request, CancellationToken ct)
+            => Task.FromResult(new MxCommandReply { ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok } });
+
+        public Task<MxCommandReply> WriteRawAsync(int itemHandle, MxValue value, int userId, CancellationToken ct)
+        {
+            RawWrites.Enqueue((itemHandle, value));
+            return Task.FromResult(WriteResponder?.Invoke(itemHandle) ?? GoodStatusRow());
+        }
+
+        // A reply carrying a Good MX status ROW (not an empty statuses array): the writer's static
+        // meters are process-wide, and an empty-statuses Good would increment galaxy.writes.unconfirmed
+        // under the meter tests running in parallel in another class.
+        private static MxCommandReply GoodStatusRow()
+        {
+            var reply = new MxCommandReply { ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok } };
+            reply.Statuses.Add(new MxStatusProxy { Success = 1, Category = MxStatusCategory.Ok });
+            return reply;
+        }
+    }
+
+    private static SecurityClassification Free(string _) => SecurityClassification.FreeAccess;
+
+    /// <summary>Sixteen distinct tags: lanes overlap (not serial) and never exceed the cap of eight.</summary>
+    [Fact]
+    public async Task Distinct_tags_write_concurrently_bounded_at_the_lane_cap()
+    {
+        var ops = new GatingOps { ReleaseAt = GatewayGalaxyDataWriter.MaxConcurrentWriteLanes };
+        var writes = Enumerable.Range(0, 16).Select(i => new WriteRequest($"Tag{i}.PV", (double)i)).ToArray();
+
+        var results = await NewWriter().WriteBatchForTestAsync(ops, writes, Free, CancellationToken.None)
+            .WaitAsync(TimeSpan.FromMilliseconds(WaitMs));
+
+        ops.Peak.ShouldBe(GatewayGalaxyDataWriter.MaxConcurrentWriteLanes,
+            "first-touch lanes must overlap (serial = 1) and stay bounded (unbounded > 8)");
+        results.Count.ShouldBe(16);
+        results.ShouldAllBe(r => r.StatusCode == StatusCodeMap.Good);
+        ops.AddItems.Count.ShouldBe(16, "one AddItem per distinct tag");
+        ops.RawWrites.Count.ShouldBe(16);
+    }
+
+    /// <summary>Writes to the same tag stay in input order inside their lane and share one AddItem.</summary>
+    [Fact]
+    public async Task Same_tag_writes_stay_ordered_and_share_one_handle()
+    {
+        var ops = new GatingOps { ReleaseAt = 2 };
+        var writes = new[]
+        {
+            new WriteRequest("Tank01.SP", 1.0),
+            new WriteRequest("Tank02.SP", 10.0),
+            new WriteRequest("Tank01.SP", 2.0),
+            new WriteRequest("Tank01.SP", 3.0),
+        };
+
+        var results = await NewWriter().WriteBatchForTestAsync(ops, writes, Free, CancellationToken.None)
+            .WaitAsync(TimeSpan.FromMilliseconds(WaitMs));
+
+        results.Count.ShouldBe(4);
+        results.ShouldAllBe(r => r.StatusCode == StatusCodeMap.Good);
+        ops.AddItems.Count(a => a == "Tank01.SP").ShouldBe(1, "a lane AddItems its tag once");
+        var tank01 = ops.RawWrites.Where(w => w.Handle == ops.RawWrites.First(x => x.Value.DoubleValue == 1.0).Handle)
+            .Select(w => w.Value.DoubleValue).ToArray();
+        tank01.ShouldBe([1.0, 2.0, 3.0], "same-tag writes must keep their authored order");
+    }
+
+    /// <summary>A per-tag failure lands at that request's index while the other lanes stay Good.</summary>
+    [Fact]
+    public async Task Results_land_at_their_request_index()
+    {
+        var ops = new GatingOps
+        {
+            ReleaseAt = 3,
+            HandleFor = tag => tag[0] - 'A' + 1,            // A.PV → 1, B.PV → 2, C.PV → 3
+            WriteResponder = handle => new MxCommandReply
+            {
+                ProtocolStatus = new ProtocolStatus
+                {
+                    Code = handle == 2 ? ProtocolStatusCode.MxaccessFailure : ProtocolStatusCode.Ok,
+                },
+            },
+        };
+        var writes = new[]
+        {
+            new WriteRequest("A.PV", 1.0),
+            new WriteRequest("B.PV", 2.0),   // refused by the gateway
+            new WriteRequest("C.PV", 3.0),
+        };
+
+        var results = await NewWriter().WriteBatchForTestAsync(ops, writes, Free, CancellationToken.None)
+            .WaitAsync(TimeSpan.FromMilliseconds(WaitMs));
+
+        results.Select(r => r.StatusCode).ShouldBe(
+            [StatusCodeMap.Good, StatusCodeMap.BadCommunicationError, StatusCodeMap.Good],
+            "each result must sit at its request's index regardless of lane completion order");
+    }
+
+    /// <summary>A single-tag batch takes the direct path and still produces one result per request.</summary>
+    [Fact]
+    public async Task Single_tag_batch_runs_serially_in_order()
+    {
+        var ops = new GatingOps { ReleaseAt = 1 };
+        var writes = new[] { new WriteRequest("X.PV", 1.0), new WriteRequest("X.PV", 2.0) };
+
+        var results = await NewWriter().WriteBatchForTestAsync(ops, writes, Free, CancellationToken.None)
+            .WaitAsync(TimeSpan.FromMilliseconds(WaitMs));
+
+        results.Count.ShouldBe(2);
+        ops.Peak.ShouldBe(1);
+        ops.RawWrites.Select(w => w.Value.DoubleValue).ShouldBe([1.0, 2.0]);
+    }
+}