From 6dedf5158f0068fd99ed14ba94ea06ea312d0911 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 06:41:47 -0400 Subject: [PATCH] feat(galaxy): wire DiscoverPageSize and ProbeTimeoutSeconds; concurrent first-touch write lanes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .../Browse/GatewayGalaxyHierarchySource.cs | 67 ++++++- .../GalaxyDriver.cs | 3 +- .../GalaxyDriverProbe.cs | 31 +++- .../Runtime/GatewayGalaxyDataWriter.cs | 78 +++++++- .../Drivers/Forms/GalaxyDriverForm.razor | 4 +- ...GatewayGalaxyHierarchySourcePagingTests.cs | 92 ++++++++++ .../GalaxyDriverProbeTimeoutTests.cs | 69 +++++++ ...GatewayGalaxyDataWriterConcurrencyTests.cs | 168 ++++++++++++++++++ 8 files changed, 498 insertions(+), 14 deletions(-) create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Browse/GatewayGalaxyHierarchySourcePagingTests.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/GalaxyDriverProbeTimeoutTests.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GatewayGalaxyDataWriterConcurrencyTests.cs diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Browse/GatewayGalaxyHierarchySource.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Browse/GatewayGalaxyHierarchySource.cs index bafabb59..68899bb8 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Browse/GatewayGalaxyHierarchySource.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Browse/GatewayGalaxyHierarchySource.cs @@ -5,22 +5,83 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Browse; /// /// Default wrapping the gateway's -/// . Pages internally via the client's overload. +/// . Pages the DiscoverHierarchy RPC itself, +/// at the configured Repository.DiscoverPageSize: the client's typed +/// DiscoverHierarchyAsync overloads hard-code a 5000-row page +/// (DiscoverHierarchyOptions carries no page size), so the knob the AdminUI +/// authors and the factory parses used to reach nothing. The raw RPC takes the page +/// size on the wire (DiscoverHierarchyRequest.page_size), so it is honoured here. /// public sealed class GatewayGalaxyHierarchySource : IGalaxyHierarchySource { + /// The page size used when the configured one is not positive — the client library's own default. + public const int DefaultPageSize = 5000; + private readonly GalaxyRepositoryClient _client; + private readonly int _pageSize; /// /// Initializes a new instance of the class. /// /// The gateway's Galaxy repository client for discovering the object hierarchy. - public GatewayGalaxyHierarchySource(GalaxyRepositoryClient client) + /// + /// Objects per DiscoverHierarchy page (Repository.DiscoverPageSize). A value that is + /// not positive falls back to . + /// + public GatewayGalaxyHierarchySource(GalaxyRepositoryClient client, int pageSize = DefaultPageSize) { _client = client ?? throw new ArgumentNullException(nameof(client)); + _pageSize = EffectivePageSize(pageSize); } + /// The page size this source sends on every DiscoverHierarchy request. + public int PageSize => _pageSize; + + /// Clamp a configured page size to something the gateway accepts: non-positive → . + /// The configured value. + /// The page size to send. + public static int EffectivePageSize(int configured) => configured > 0 ? configured : DefaultPageSize; + /// public Task> GetHierarchyAsync(CancellationToken cancellationToken) - => _client.DiscoverHierarchyAsync(cancellationToken); + => PageAsync(_client.DiscoverHierarchyRawAsync, _pageSize, cancellationToken); + + /// + /// Walk every DiscoverHierarchy page at and concatenate the + /// objects. Mirrors the client library's own loop (including its repeated-page-token guard, + /// which turns a misbehaving gateway into an error rather than an endless walk), with the + /// RPC behind a delegate so the loop is unit-testable without a channel. + /// + /// The raw DiscoverHierarchy RPC. + /// Objects per page; sent on every request. + /// Cancellation token for the walk. + /// Every object across all pages, in gateway order. + internal static async Task> PageAsync( + Func> discover, + int pageSize, + CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(discover); + var effective = EffectivePageSize(pageSize); + + var objects = new List(); + var seenPageTokens = new HashSet(StringComparer.Ordinal); + var pageToken = string.Empty; + do + { + var request = new DiscoverHierarchyRequest { PageSize = effective, PageToken = pageToken }; + var reply = await discover(request, cancellationToken).ConfigureAwait(false); + + objects.AddRange(reply.Objects); + pageToken = reply.NextPageToken ?? string.Empty; + if (!string.IsNullOrWhiteSpace(pageToken) && !seenPageTokens.Add(pageToken)) + { + throw new InvalidOperationException( + $"Galaxy DiscoverHierarchy returned a repeated page token '{pageToken}' — refusing to loop."); + } + } + while (!string.IsNullOrWhiteSpace(pageToken)); + + return objects; + } } 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 f2768291..8cb4bcb5 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs @@ -1821,7 +1821,8 @@ public sealed class GalaxyDriver _ownedRepositoryClient ??= GalaxyRepositoryClient.Create( await BuildClientOptionsAsync(_options.Gateway, cancellationToken).ConfigureAwait(false)); return new TracedGalaxyHierarchySource( - new GatewayGalaxyHierarchySource(_ownedRepositoryClient), _options.MxAccess.ClientName); + new GatewayGalaxyHierarchySource(_ownedRepositoryClient, _options.Repository.DiscoverPageSize), + _options.MxAccess.ClientName); } /// diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriverProbe.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriverProbe.cs index a4010d07..81807500 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriverProbe.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriverProbe.cs @@ -54,12 +54,18 @@ public sealed class GalaxyDriverProbe : IDriverProbe if (string.IsNullOrWhiteSpace(host) || port <= 0) return new(false, "Config has no host/port to probe.", null); + // The config's own ProbeTimeoutSeconds caps the probe deadline. It can only tighten the + // caller's budget, never extend it — the caller's token fires at `timeout` regardless. + timeout = EffectiveTimeout(timeout, opts.ProbeTimeoutSeconds); + using var probeCts = CancellationTokenSource.CreateLinkedTokenSource(ct); + probeCts.CancelAfter(timeout); + // --- TCP preflight: fast-fail for closed ports / unreachable hosts --- var sw = Stopwatch.StartNew(); try { using var socket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp); - await socket.ConnectAsync(host, port, ct); + await socket.ConnectAsync(host, port, probeCts.Token); } catch (SocketException ex) { @@ -86,10 +92,7 @@ public sealed class GalaxyDriverProbe : IDriverProbe { client = GalaxyRepositoryClient.Create(BuildProbeClientOptions(opts.Gateway, timeout)); - using var deadlineCts = CancellationTokenSource.CreateLinkedTokenSource(ct); - deadlineCts.CancelAfter(timeout); - - await client.TestConnectionAsync(deadlineCts.Token).ConfigureAwait(false); + await client.TestConnectionAsync(probeCts.Token).ConfigureAwait(false); sw.Stop(); // No RpcException ⇒ the gateway answered the gRPC call successfully. @@ -157,6 +160,24 @@ public sealed class GalaxyDriverProbe : IDriverProbe _ => (false, $"Reachable at {host}:{port} but gateway gRPC handshake failed: {code}"), }; + /// + /// The deadline the probe actually runs under: the caller's budget, tightened by the + /// config's when that is positive and + /// shorter. The knob was authored by the AdminUI and parsed by the factory but consumed by + /// nothing — the probe always ran on the caller's budget alone. A non-positive caller budget + /// is treated as the config value (or 1 s when that is unset too), as before. + /// + /// The budget the host handed the probe. + /// The config's ProbeTimeoutSeconds; ignored when not positive. + /// The effective probe deadline. + internal static TimeSpan EffectiveTimeout(TimeSpan callerTimeout, int configuredSeconds) + { + var caller = callerTimeout > TimeSpan.Zero ? callerTimeout : TimeSpan.FromSeconds(1); + if (configuredSeconds <= 0) return caller; + var configured = TimeSpan.FromSeconds(configuredSeconds); + return configured < caller ? configured : caller; + } + /// /// Builds the gRPC client options for the probe ping from the gateway config WITHOUT /// resolving the API-key secret ref (the host owns secret resolution; the probe sends the diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/GatewayGalaxyDataWriter.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/GatewayGalaxyDataWriter.cs index a472f269..957459df 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/GatewayGalaxyDataWriter.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/GatewayGalaxyDataWriter.cs @@ -120,15 +120,87 @@ public sealed class GatewayGalaxyDataWriter : IGalaxyDataWriter // Bind the per-batch MX write operations behind the internal IMxWriteOps seam (see below) so the // whole write pipeline is unit-testable without the sealed SDK session (archreview 06/S-1). var ops = new SessionMxWriteOps(session, serverHandle); + return await WriteBatchAsync(ops, writes, securityResolver, cancellationToken).ConfigureAwait(false); + } + /// + /// Upper bound on write lanes in flight at once. A first-touch write is three round-trips + /// (AddItem → AdviseSupervisory → Write) and the batch used to run them strictly serially; + /// distinct tags are independent, so their lanes now overlap, bounded so a large batch cannot + /// flood the gateway's STA pump. Acks stay serial by design (see GalaxyDriver.AcknowledgeAsync). + /// + internal const int MaxConcurrentWriteLanes = 8; + + /// Test seam: drive the real batch pipeline against a fake . + /// The (fake) MX write operations. + /// The batch. + /// Per-tag classification resolver. + /// The cancellation token. + /// One result per request, in input order. + internal Task> WriteBatchForTestAsync( + IMxWriteOps ops, IReadOnlyList writes, + Func securityResolver, CancellationToken ct) + => WriteBatchAsync(ops, writes, securityResolver, ct); + + /// + /// Run a batch as one lane per distinct tag (OrdinalIgnoreCase): writes to the SAME tag + /// stay in input order inside their lane, lanes run concurrently up to + /// , and results land at their request's index. + /// A single-tag batch takes the direct path with no gate. + /// + private async Task> WriteBatchAsync( + IMxWriteOps ops, IReadOnlyList writes, + Func securityResolver, CancellationToken ct) + { var results = new WriteResult[writes.Count]; + if (writes.Count == 0) return results; + + // Group request indices by tag, first-seen order preserved so the lane start order is + // deterministic and a same-tag sequence keeps its authored order. + var lanes = new Dictionary>(StringComparer.OrdinalIgnoreCase); + var laneOrder = new List>(); for (var i = 0; i < writes.Count; i++) { - results[i] = await WriteOneAsync(ops, writes[i], - securityResolver(writes[i].FullReference), cancellationToken) - .ConfigureAwait(false); + if (!lanes.TryGetValue(writes[i].FullReference, out var lane)) + { + lane = []; + lanes[writes[i].FullReference] = lane; + laneOrder.Add(lane); + } + lane.Add(i); } + + if (laneOrder.Count == 1) + { + await RunLaneAsync(laneOrder[0]).ConfigureAwait(false); + return results; + } + + using var gate = new SemaphoreSlim(MaxConcurrentWriteLanes, MaxConcurrentWriteLanes); + var tasks = new Task[laneOrder.Count]; + for (var l = 0; l < laneOrder.Count; l++) + { + var lane = laneOrder[l]; + tasks[l] = RunGatedAsync(lane); + } + await Task.WhenAll(tasks).ConfigureAwait(false); return results; + + async Task RunGatedAsync(List lane) + { + await gate.WaitAsync(ct).ConfigureAwait(false); + try { await RunLaneAsync(lane).ConfigureAwait(false); } + finally { gate.Release(); } + } + + async Task RunLaneAsync(List lane) + { + foreach (var i in lane) + { + results[i] = await WriteOneAsync(ops, writes[i], + securityResolver(writes[i].FullReference), ct).ConfigureAwait(false); + } + } } /// Test seam: drive the real single-write pipeline against a fake diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Drivers/Forms/GalaxyDriverForm.razor b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Drivers/Forms/GalaxyDriverForm.razor index 17468d7d..705f70c7 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Drivers/Forms/GalaxyDriverForm.razor +++ b/src/Server/ZB.MOM.WW.OtOpcUa.AdminUI/Components/Shared/Drivers/Forms/GalaxyDriverForm.razor @@ -92,7 +92,7 @@
-
Default 5000 objects per page.
+
Objects per DiscoverHierarchy page the driver asks the gateway for. Default 5000.
@@ -138,7 +138,7 @@
-
Max 60. Used by Test Connect. Default 30.
+
Caps the Test Connect probe deadline (the probe's own budget still applies, so this only shortens it). Max 60. Default 30.
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Browse/GatewayGalaxyHierarchySourcePagingTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Browse/GatewayGalaxyHierarchySourcePagingTests.cs new file mode 100644 index 00000000..d0767592 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Browse/GatewayGalaxyHierarchySourcePagingTests.cs @@ -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; + +/// +/// Review 2026-10-21 G-5: Repository.DiscoverPageSize was authored and parsed but reached +/// nothing — the client's typed DiscoverHierarchyAsync hard-codes 5000. The source now pages +/// the raw RPC itself at the configured size. +/// +public sealed class GatewayGalaxyHierarchySourcePagingTests +{ + private static GalaxyObject Obj(int id) => new() { GobjectId = id, TagName = $"Obj{id}" }; + + /// Every request carries the configured page size and the page tokens chain until the last page. + [Fact] + public async Task Pages_at_the_configured_size_and_chains_tokens() + { + var requests = new List(); + Task 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"]); + } + + /// A different configured size is what goes on the wire (positive control for the knob). + [Fact] + public async Task A_different_page_size_is_sent_on_the_wire() + { + int? sent = null; + Task Discover(DiscoverHierarchyRequest req, CancellationToken ct) + { + sent = req.PageSize; + return Task.FromResult(new DiscoverHierarchyReply()); + } + + await GatewayGalaxyHierarchySource.PageAsync(Discover, pageSize: 250, CancellationToken.None); + sent.ShouldBe(250); + } + + /// A non-positive configured size falls back to the client library's 5000 default. + [Theory] + [InlineData(0)] + [InlineData(-7)] + public void Non_positive_page_size_falls_back_to_the_default(int configured) + => GatewayGalaxyHierarchySource.EffectivePageSize(configured).ShouldBe(GatewayGalaxyHierarchySource.DefaultPageSize); + + /// A repeated page token is refused rather than walked forever. + [Fact] + public async Task A_repeated_page_token_is_refused() + { + var calls = 0; + Task 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( + () => GatewayGalaxyHierarchySource.PageAsync(Discover, pageSize: 10, CancellationToken.None)); + ex.Message.ShouldContain("repeated page token"); + calls.ShouldBe(2); + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/GalaxyDriverProbeTimeoutTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/GalaxyDriverProbeTimeoutTests.cs new file mode 100644 index 00000000..1f5fbb78 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/GalaxyDriverProbeTimeoutTests.cs @@ -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; + +/// +/// Review 2026-10-21 G-6: ProbeTimeoutSeconds 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). +/// +[Trait("Category", "Unit")] +public sealed class GalaxyDriverProbeTimeoutTests +{ + /// The effective deadline is the shorter of the caller's budget and the configured cap. + [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)); + + /// + /// 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. + /// + [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(); + } + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GatewayGalaxyDataWriterConcurrencyTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GatewayGalaxyDataWriterConcurrencyTests.cs new file mode 100644 index 00000000..dff0f665 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/GatewayGalaxyDataWriterConcurrencyTests.cs @@ -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; + +/// +/// 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 +/// ; same-tag writes keep their +/// order inside one lane; results stay at their request's index. +/// +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); + + /// + /// Fake ops whose AddItem parks every caller until callers are in + /// flight at once (or a safety delay elapses), recording the peak concurrency. + /// + private sealed class GatingOps : IMxWriteOps + { + private int _inFlight; + private int _peak; + private int _nextHandle; + private readonly TaskCompletionSource _release = new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly ConcurrentDictionary _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 AddItems { get; } = new(); + public Func? WriteResponder { get; init; } + + /// Deterministic handle per tag when set; otherwise handles are assigned in AddItem completion order. + public Func? HandleFor { get; init; } + + public string SessionId => "sess"; + public int ServerHandle => 1; + + public async Task 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 InvokeAsync(MxCommandRequest request, CancellationToken ct) + => Task.FromResult(new MxCommandReply { ProtocolStatus = new ProtocolStatus { Code = ProtocolStatusCode.Ok } }); + + public Task 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; + + /// Sixteen distinct tags: lanes overlap (not serial) and never exceed the cap of eight. + [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); + } + + /// Writes to the same tag stay in input order inside their lane and share one AddItem. + [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"); + } + + /// A per-tag failure lands at that request's index while the other lanes stay Good. + [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"); + } + + /// A single-tag batch takes the direct path and still produces one result per request. + [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]); + } +}