diff --git a/archreview/plans/R2-08-ldap-historyread-async-plan.md.tasks.json b/archreview/plans/R2-08-ldap-historyread-async-plan.md.tasks.json index 9f18d8f3..530ed9e9 100644 --- a/archreview/plans/R2-08-ldap-historyread-async-plan.md.tasks.json +++ b/archreview/plans/R2-08-ldap-historyread-async-plan.md.tasks.json @@ -7,7 +7,7 @@ { "id": "R2-08-T4", "subject": "LdapOpcUaUserAuthenticator: WaitAsync wall-clock bound + SemaphoreSlim concurrency cap with release-on-core-completion (turns T1 green)", "status": "completed", "blockedBy": ["R2-08-T1", "R2-08-T2"] }, { "id": "R2-08-T5", "subject": "Directory-outage circuit: fail-fast deny after N consecutive system-side failures, half-open after cooldown", "status": "completed", "blockedBy": ["R2-08-T3", "R2-08-T4"] }, { "id": "R2-08-T6", "subject": "Outage-sim integration test (real library, unroutable host, via internal HandleImpersonation) + SDK m_eventLock contract doc + docs/security.md", "status": "completed", "blockedBy": ["R2-08-T4"] }, - { "id": "R2-08-T7", "subject": "S3 sequential-serving repro test (RED): MaxObservedConcurrency >= 2 on an 8-node Raw batch — fails on current code", "status": "pending", "blockedBy": [] }, + { "id": "R2-08-T7", "subject": "S3 sequential-serving repro test (RED): MaxObservedConcurrency >= 2 on an 8-node Raw batch — fails on current code", "status": "completed", "blockedBy": [] }, { "id": "R2-08-T8", "subject": "ServerHistorianOptions: HistoryReadBatchParallelism/MaxConcurrentHistoryReadBatches/HistoryReadDeadline + Validate warnings + node-manager properties + hosted-service wiring", "status": "pending", "blockedBy": [] }, { "id": "R2-08-T9", "subject": "Async-ify ServeNode + Processed/AtTime arms (ct threading, OCE=>BadTimeout, behavior-neutral)", "status": "pending", "blockedBy": ["R2-08-T7", "R2-08-T8"] }, { "id": "R2-08-T10", "subject": "Async-ify ServeRawPaged (page read + tie-cluster over-fetch) + extract ServeEventsAsync", "status": "pending", "blockedBy": ["R2-08-T9"] }, diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadConcurrencyTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadConcurrencyTests.cs new file mode 100644 index 00000000..37599d8a --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadConcurrencyTests.cs @@ -0,0 +1,200 @@ +using System.Diagnostics; +using Microsoft.Extensions.Logging.Abstractions; +using Opc.Ua; +using Opc.Ua.Server; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using HistorianRead = ZB.MOM.WW.OtOpcUa.Core.Abstractions.HistoryReadResult; +using SdkHistoryReadResult = Opc.Ua.HistoryReadResult; + +namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests; + +/// +/// Arch-review 03/S3 — the node-manager's HistoryRead arms must serve a multi-node batch with BOUNDED +/// per-node parallelism, a per-request deadline, and a process-wide concurrent-batch limiter, rather +/// than block-bridging the gateway per node sequentially on an SDK request thread. Boots the same real +/// harness as and drives the +/// public HistoryRead with a delaying fake that records the +/// maximum observed concurrent reads (deterministic — no raw-timing dependence for the primary +/// assertions). +/// +public sealed class NodeManagerHistoryReadConcurrencyTests : IDisposable +{ + private static CancellationToken Ct => TestContext.Current.CancellationToken; + + private readonly string _pkiRoot = Path.Combine( + Path.GetTempPath(), + $"otopcua-historyread-conc-{Guid.NewGuid():N}"); + + /// + /// RED repro (03/S3): a single Raw HistoryRead naming 8 historized nodes against a 150 ms-per-read + /// backend must serve at least two nodes concurrently. On the current sequential foreach the + /// max observed concurrency is pinned at 1, so this fails. + /// + [Fact] + public async Task Raw_batch_is_served_with_bounded_parallelism() + { + var (host, server) = await BootAsync(); + var nm = server.NodeManager!; + var fake = new ConcurrencyTrackingHistorianDataSource(TimeSpan.FromMilliseconds(150)); + nm.HistorianDataSource = fake; + + var nodeIds = MaterializeHistorizedNodes(nm, count: 8); + + var details = new ReadRawModifiedDetails + { + StartTime = DateTime.UtcNow.AddHours(-1), + EndTime = DateTime.UtcNow, + NumValuesPerNode = 10, + IsReadModified = false, + }; + + var sw = Stopwatch.StartNew(); + await Task.Run(() => InvokeHistoryRead(server, nm, details, nodeIds), Ct) + .WaitAsync(TimeSpan.FromSeconds(15), Ct); + sw.Stop(); + + // Primary (deterministic): the batch was served with >1 node in flight at once. + fake.MaxObservedConcurrency.ShouldBeGreaterThanOrEqualTo(2); + + await host.DisposeAsync(); + } + + /// Materialises historized Float variables and returns their NodeIds. + private static NodeId[] MaterializeHistorizedNodes(OtOpcUaNodeManager nm, int count) + { + var ids = new NodeId[count]; + for (var i = 0; i < count; i++) + { + var key = $"eq/tag{i}"; + nm.EnsureVariable(key, parentFolderNodeId: null, displayName: $"Tag{i}", dataType: "Float", + writable: false, historianTagname: $"WW.Tag{i}"); + ids[i] = nm.TryGetVariable(key)!.NodeId; + } + + return ids; + } + + /// Issues one HistoryRead batch for all supplied node ids in order. + private static (IList Results, IList Errors) InvokeHistoryRead( + OtOpcUaSdkServer server, OtOpcUaNodeManager nm, HistoryReadDetails details, params NodeId[] nodeIds) + { + var context = new OperationContext( + new RequestHeader(), secureChannelContext: null, RequestType.HistoryRead, identity: null); + + var nodesToRead = nodeIds.Select(id => new HistoryReadValueId { NodeId = id }).ToList(); + var results = Enumerable.Repeat(null!, nodeIds.Length).ToList(); + var errors = Enumerable.Repeat(null!, nodeIds.Length).ToList(); + + nm.HistoryRead( + context, + details, + TimestampsToReturn.Both, + releaseContinuationPoints: false, + nodesToRead, + results, + errors); + + return (results, errors); + } + + /// + /// A delaying fake historian source that tracks the maximum number of reads in flight at once + /// (via interlocked live-counter) so a test can prove per-node parallelism deterministically. + /// + private sealed class ConcurrencyTrackingHistorianDataSource(TimeSpan delay) : IHistorianDataSource + { + private int _current; + private int _max; + + /// The highest number of concurrently in-flight reads observed. + public int MaxObservedConcurrency => Volatile.Read(ref _max); + + private async Task TrackAsync(CancellationToken ct) + { + var now = Interlocked.Increment(ref _current); + UpdateMax(now); + try + { + await Task.Delay(delay, ct).ConfigureAwait(false); + } + finally + { + Interlocked.Decrement(ref _current); + } + + return new HistorianRead( + new[] { new DataValueSnapshot(1.0f, StatusCodes.Good, DateTime.UtcNow, DateTime.UtcNow) }, null); + } + + private void UpdateMax(int observed) + { + int cur; + while ((cur = Volatile.Read(ref _max)) < observed && + Interlocked.CompareExchange(ref _max, observed, cur) != cur) + { + } + } + + public Task ReadRawAsync( + string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode, + CancellationToken cancellationToken) => TrackAsync(cancellationToken); + + public Task ReadProcessedAsync( + string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval, + HistoryAggregateType aggregate, CancellationToken cancellationToken) => TrackAsync(cancellationToken); + + public Task ReadAtTimeAsync( + string fullReference, IReadOnlyList timestampsUtc, CancellationToken cancellationToken) => + TrackAsync(cancellationToken); + + public Task ReadEventsAsync( + string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, + CancellationToken cancellationToken) => + Task.FromResult(new HistoricalEventsResult(Array.Empty(), null)); + + public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot(); + + public void Dispose() + { + } + } + + private async Task<(OpcUaApplicationHost Host, OtOpcUaSdkServer Server)> BootAsync() + { + var host = new OpcUaApplicationHost( + new OpcUaApplicationHostOptions + { + ApplicationName = "OtOpcUa.HistoryReadConcTest", + ApplicationUri = $"urn:OtOpcUa.HistoryReadConcTest:{Guid.NewGuid():N}", + OpcUaPort = AllocateFreePort(), + PublicHostname = "localhost", + PkiStoreRoot = _pkiRoot, + }, + NullLogger.Instance); + + var server = new OtOpcUaSdkServer(); + await host.StartAsync(server, Ct); + return (host, server); + } + + private static int AllocateFreePort() + { + using var listener = new System.Net.Sockets.TcpListener(System.Net.IPAddress.Loopback, 0); + listener.Start(); + var port = ((System.Net.IPEndPoint)listener.LocalEndpoint).Port; + listener.Stop(); + return port; + } + + /// Cleans up the PKI root directory. + public void Dispose() + { + if (Directory.Exists(_pkiRoot)) + { + try { Directory.Delete(_pkiRoot, recursive: true); } + catch { /* best-effort cleanup */ } + } + } +}