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 2e262a98..21955c78 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 @@ -13,7 +13,7 @@ { "id": "R2-08-T10", "subject": "Async-ify ServeRawPaged (page read + tie-cluster over-fetch) + extract ServeEventsAsync", "status": "completed", "blockedBy": ["R2-08-T9"] }, { "id": "R2-08-T11", "subject": "RunBounded fan-out in all four HistoryRead arms, single bridge per arm (turns T7 green) + upper-bound + isolation tests", "status": "completed", "blockedBy": ["R2-08-T9", "R2-08-T10"] }, { "id": "R2-08-T12", "subject": "Per-request HistoryRead deadline CTS — hung backend surfaces BadTimeout per node", "status": "completed", "blockedBy": ["R2-08-T11"] }, - { "id": "R2-08-T13", "subject": "Process-wide concurrent-batch limiter — saturation degrades to BadTooManyOperations (lazy gate, bounded wait)", "status": "pending", "blockedBy": ["R2-08-T12"] }, + { "id": "R2-08-T13", "subject": "Process-wide concurrent-batch limiter — saturation degrades to BadTooManyOperations (lazy gate, bounded wait)", "status": "completed", "blockedBy": ["R2-08-T12"] }, { "id": "R2-08-T14", "subject": "Docs (Historian.md + CLAUDE.md ServerHistorian table + security.md) + full touched-suite and whole-solution regression sweep + STATUS.md entry", "status": "pending", "blockedBy": ["R2-08-T5", "R2-08-T6", "R2-08-T13"] } ], "lastUpdated": "2026-07-12" diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs index e059d535..964c2c76 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -221,6 +221,10 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 /// public TimeSpan HistoryReadDeadline { get; set; } = TimeSpan.FromSeconds(60); + /// Lazily-created process-wide HistoryRead-batch limiter (see ). + private volatile SemaphoreSlim? _historyReadBatchLimiter; + private readonly object _historyReadBatchLimiterLock = new(); + private volatile IHistoryContinuationStore _historyContinuationStore = new SessionHistoryContinuationStore(); /// @@ -1816,25 +1820,28 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 IDictionary cache) { var session = context.OperationContext?.Session; - using var deadline = CreateDeadline(); - var ct = deadline?.Token ?? CancellationToken.None; - var work = new List>(nodesToProcess.Count); - foreach (var handle in nodesToProcess) + WithHistoryReadBatchLimiter(nodesToProcess, results, errors, () => { - if (details.IsReadModified) + using var deadline = CreateDeadline(); + var ct = deadline?.Token ?? CancellationToken.None; + var work = new List>(nodesToProcess.Count); + foreach (var handle in nodesToProcess) { - // We never serve modified-value history; mark this node unsupported and move on. - errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported; - continue; + if (details.IsReadModified) + { + // We never serve modified-value history; mark this node unsupported and move on. + errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported; + continue; + } + + work.Add(() => ServeRawPagedAsync( + handle, session, nodesToRead, results, errors, + details.StartTime, details.EndTime, details.NumValuesPerNode, + ct)); } - work.Add(() => ServeRawPagedAsync( - handle, session, nodesToRead, results, errors, - details.StartTime, details.EndTime, details.NumValuesPerNode, - ct)); - } - - RunBounded(work, HistoryReadBatchParallelism); + RunBounded(work, HistoryReadBatchParallelism); + }); } /// @@ -1850,37 +1857,40 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 { // OPC UA ProcessingInterval is a Duration in milliseconds — convert once per batch. var interval = TimeSpan.FromMilliseconds(details.ProcessingInterval); - using var deadline = CreateDeadline(); - var ct = deadline?.Token ?? CancellationToken.None; - var work = new List>(nodesToProcess.Count); - foreach (var handle in nodesToProcess) + WithHistoryReadBatchLimiter(nodesToProcess, results, errors, () => { - // AggregateType is a per-node parallel collection (same length as nodesToRead, enforced by - // the base dispatcher). handle.Index is the node's position in that collection. - var aggregateNodeId = details.AggregateType[handle.Index]; - var aggregate = MapAggregate(aggregateNodeId); - if (aggregate is null) + using var deadline = CreateDeadline(); + var ct = deadline?.Token ?? CancellationToken.None; + var work = new List>(nodesToProcess.Count); + foreach (var handle in nodesToProcess) { - errors[handle.Index] = StatusCodes.BadAggregateNotSupported; - continue; + // AggregateType is a per-node parallel collection (same length as nodesToRead, enforced by + // the base dispatcher). handle.Index is the node's position in that collection. + var aggregateNodeId = details.AggregateType[handle.Index]; + var aggregate = MapAggregate(aggregateNodeId); + if (aggregate is null) + { + errors[handle.Index] = StatusCodes.BadAggregateNotSupported; + continue; + } + + // Processed is SINGLE-SHOT (no continuation point). Unlike Raw, ReadProcessedDetails carries + // NO client count cap (NumValuesPerNode) — the bucket count is deterministic (window / interval) + // and the single-shot backend returns every bucket in one read, so there is no "full page ⇒ + // maybe more" signal to page on. Returning the complete aggregate result with a null CP is + // spec-conformant (OPC UA Part 11 lets a server return all available data in one response). + var agg = aggregate.Value; + work.Add(() => ServeNodeAsync(handle, results, errors, (source, tagname, token) => source.ReadProcessedAsync( + tagname, + details.StartTime, + details.EndTime, + interval, + agg, + token), ct)); } - // Processed is SINGLE-SHOT (no continuation point). Unlike Raw, ReadProcessedDetails carries - // NO client count cap (NumValuesPerNode) — the bucket count is deterministic (window / interval) - // and the single-shot backend returns every bucket in one read, so there is no "full page ⇒ - // maybe more" signal to page on. Returning the complete aggregate result with a null CP is - // spec-conformant (OPC UA Part 11 lets a server return all available data in one response). - var agg = aggregate.Value; - work.Add(() => ServeNodeAsync(handle, results, errors, (source, tagname, token) => source.ReadProcessedAsync( - tagname, - details.StartTime, - details.EndTime, - interval, - agg, - token), ct)); - } - - RunBounded(work, HistoryReadBatchParallelism); + RunBounded(work, HistoryReadBatchParallelism); + }); } /// @@ -1896,18 +1906,21 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 { // Snapshot the requested timestamps once — the same list is read for every node. var timestamps = details.ReqTimes?.ToList() ?? new List(); - using var deadline = CreateDeadline(); - var ct = deadline?.Token ?? CancellationToken.None; - var work = new List>(nodesToProcess.Count); - foreach (var handle in nodesToProcess) + WithHistoryReadBatchLimiter(nodesToProcess, results, errors, () => { - work.Add(() => ServeNodeAsync(handle, results, errors, (source, tagname, token) => source.ReadAtTimeAsync( - tagname, - timestamps, - token), ct)); - } + using var deadline = CreateDeadline(); + var ct = deadline?.Token ?? CancellationToken.None; + var work = new List>(nodesToProcess.Count); + foreach (var handle in nodesToProcess) + { + work.Add(() => ServeNodeAsync(handle, results, errors, (source, tagname, token) => source.ReadAtTimeAsync( + tagname, + timestamps, + token), ct)); + } - RunBounded(work, HistoryReadBatchParallelism); + RunBounded(work, HistoryReadBatchParallelism); + }); } /// @@ -1924,27 +1937,30 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 // Snapshot the select clauses once — the same filter projects every node's events. var selectClauses = details.Filter?.SelectClauses ?? new SimpleAttributeOperandCollection(); - using var deadline = CreateDeadline(); - var ct = deadline?.Token ?? CancellationToken.None; - var work = new List>(nodesToProcess.Count); - foreach (var handle in nodesToProcess) + WithHistoryReadBatchLimiter(nodesToProcess, results, errors, () => { - var idString = handle.NodeId.Identifier?.ToString(); - if (idString is null || !_eventNotifierSources.TryGetValue(idString, out var sourceName)) + using var deadline = CreateDeadline(); + var ct = deadline?.Token ?? CancellationToken.None; + var work = new List>(nodesToProcess.Count); + foreach (var handle in nodesToProcess) { - // Not a registered event-history source (plain folder / Null-source promotion) ⇒ unsupported. - // Set both errors and results explicitly on every bad path — don't rely on the SDK base - // pre-seeding results[i], so every path is self-contained and the contract is obvious. - errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported; - results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadHistoryOperationUnsupported }; - continue; + var idString = handle.NodeId.Identifier?.ToString(); + if (idString is null || !_eventNotifierSources.TryGetValue(idString, out var sourceName)) + { + // Not a registered event-history source (plain folder / Null-source promotion) ⇒ unsupported. + // Set both errors and results explicitly on every bad path — don't rely on the SDK base + // pre-seeding results[i], so every path is self-contained and the contract is obvious. + errors[handle.Index] = StatusCodes.BadHistoryOperationUnsupported; + results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadHistoryOperationUnsupported }; + continue; + } + + var sn = sourceName; + work.Add(() => ServeEventsAsync(handle, sn, details, selectClauses, results, errors, ct)); } - var sn = sourceName; - work.Add(() => ServeEventsAsync(handle, sn, details, selectClauses, results, errors, ct)); - } - - RunBounded(work, HistoryReadBatchParallelism); + RunBounded(work, HistoryReadBatchParallelism); + }); } /// @@ -2146,6 +2162,68 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 Task.WhenAll(tasks).GetAwaiter().GetResult(); } + /// + /// Run one HistoryRead arm under the process-wide concurrent-batch limiter (arch-review 03/S3). Each + /// arm waits up to min(HistoryReadDeadline, 5 s) for a slot; on expiry every handle in the + /// batch gets BadTooManyOperations (fast-fail under flood, brief waits under mild contention) + /// and is skipped. On acquisition the arm body runs and the slot is + /// released in a finally. Combined with this caps + /// total in-flight gateway reads at + /// MaxConcurrentHistoryReadBatches × HistoryReadBatchParallelism. + /// + /// The batch's handles — every slot is failed on limiter saturation. + /// The service-level results list. + /// The service-level errors list. + /// The arm body to run once a slot is acquired. + private void WithHistoryReadBatchLimiter( + List nodesToProcess, + IList results, + IList errors, + Action serve) + { + var gate = GetHistoryReadBatchLimiter(); + + // Bounded admission wait: cap the block at 5 s, and never exceed the per-request deadline. + var waitMs = HistoryReadDeadline > TimeSpan.Zero + ? Math.Min(HistoryReadDeadline.TotalMilliseconds, 5000.0) + : 5000.0; + + if (!gate.Wait(TimeSpan.FromMilliseconds(waitMs))) + { + foreach (var handle in nodesToProcess) + { + errors[handle.Index] = StatusCodes.BadTooManyOperations; + results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadTooManyOperations }; + } + return; + } + + try + { + serve(); + } + finally + { + gate.Release(); + } + } + + /// + /// Lazily create (double-checked) the process-wide HistoryRead-batch limiter, sized from + /// at first use so the Host's post-boot property set is + /// respected (the same benign wiring race as ). + /// + /// The shared batch limiter semaphore. + private SemaphoreSlim GetHistoryReadBatchLimiter() + { + var gate = _historyReadBatchLimiter; + if (gate is not null) return gate; + lock (_historyReadBatchLimiterLock) + { + return _historyReadBatchLimiter ??= new SemaphoreSlim(Math.Max(1, MaxConcurrentHistoryReadBatches)); + } + } + /// /// Create the per-request deadline source for a HistoryRead arm (arch-review 03/S3): a /// that fires after , so a diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadConcurrencyTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadConcurrencyTests.cs index f65c45f8..f8e97331 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadConcurrencyTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadConcurrencyTests.cs @@ -171,6 +171,55 @@ public sealed class NodeManagerHistoryReadConcurrencyTests : IDisposable await host.DisposeAsync(); } + /// + /// A saturated process-wide batch limiter degrades a second concurrent HistoryRead to + /// BadTooManyOperations (rather than exhausting the SDK request-thread pool); once the parked + /// batch completes the slot frees and a later read succeeds (03/S3). + /// + [Fact] + public async Task Saturated_limiter_rejects_with_BadTooManyOperations() + { + var (host, server) = await BootAsync(); + var nm = server.NodeManager!; + nm.MaxConcurrentHistoryReadBatches = 1; + nm.HistoryReadDeadline = TimeSpan.FromMilliseconds(200); + var fake = new GatedRawFake(); + nm.HistorianDataSource = fake; + + var ids = MaterializeHistorizedNodes(nm, count: 2); + var details = new ReadRawModifiedDetails + { + StartTime = DateTime.UtcNow.AddHours(-1), + EndTime = DateTime.UtcNow, + NumValuesPerNode = 10, + IsReadModified = false, + }; + + // First batch parks in the gated fake, holding the single limiter permit. + var first = Task.Run(() => InvokeHistoryRead(server, nm, details, ids[0]), Ct); + + var spin = Stopwatch.StartNew(); + while (fake.Entered == 0 && spin.Elapsed < TimeSpan.FromSeconds(5)) + await Task.Delay(10, Ct); + fake.Entered.ShouldBe(1); + + // Second concurrent batch: the limiter is saturated → all handles BadTooManyOperations. + var (_, errors2) = await Task.Run(() => InvokeHistoryRead(server, nm, details, ids[1]), Ct) + .WaitAsync(TimeSpan.FromSeconds(5), Ct); + errors2[0].StatusCode.Code.ShouldBe(StatusCodes.BadTooManyOperations); + fake.Entered.ShouldBe(1, "the rejected batch must never reach the backend"); + + // Release the parked batch; the slot frees and a later read succeeds. + fake.Release(); + await first.WaitAsync(TimeSpan.FromSeconds(10), Ct); + + var (_, errors3) = await Task.Run(() => InvokeHistoryRead(server, nm, details, ids[1]), Ct) + .WaitAsync(TimeSpan.FromSeconds(10), Ct); + errors3[0].StatusCode.Code.ShouldBe(StatusCodes.Good); + + await host.DisposeAsync(); + } + /// Materialises historized Float variables and returns their NodeIds. private static NodeId[] MaterializeHistorizedNodes(OtOpcUaNodeManager nm, int count) { @@ -306,6 +355,51 @@ public sealed class NodeManagerHistoryReadConcurrencyTests : IDisposable public void Dispose() { } } + /// A fake whose raw read parks (awaits a ) until the test + /// releases it — used to hold the process-wide batch limiter permit while a second batch is issued. + private sealed class GatedRawFake : IHistorianDataSource + { + private readonly TaskCompletionSource _gate = new(TaskCreationOptions.RunContinuationsAsynchronously); + private int _entered; + + /// Number of raw reads that reached the backend body. + public int Entered => Volatile.Read(ref _entered); + + /// Releases the parked read so its awaiting batch can complete. + public void Release() => _gate.TrySetResult(); + + public async Task ReadRawAsync( + string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode, + CancellationToken cancellationToken) + { + Interlocked.Increment(ref _entered); + // Deliberately ignore the per-request deadline token: the first batch must keep HOLDING the + // single limiter permit until the test releases it, so the second batch deterministically hits + // a saturated limiter (rather than the first batch's own deadline freeing the slot early). + await _gate.Task.ConfigureAwait(false); + return new HistorianRead( + new[] { new DataValueSnapshot(1.0f, StatusCodes.Good, DateTime.UtcNow, DateTime.UtcNow) }, null); + } + + public Task ReadProcessedAsync( + string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval, + HistoryAggregateType aggregate, CancellationToken cancellationToken) => + NullHistorianDataSource.Instance.ReadProcessedAsync(fullReference, startUtc, endUtc, interval, aggregate, cancellationToken); + + public Task ReadAtTimeAsync( + string fullReference, IReadOnlyList timestampsUtc, CancellationToken cancellationToken) => + NullHistorianDataSource.Instance.ReadAtTimeAsync(fullReference, timestampsUtc, cancellationToken); + + public Task ReadEventsAsync( + string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, + CancellationToken cancellationToken) => + NullHistorianDataSource.Instance.ReadEventsAsync(sourceName, startUtc, endUtc, maxEvents, cancellationToken); + + public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot(); + + public void Dispose() { } + } + /// A per-tagname fake that delegates each raw read to a caller-supplied func (which may throw /// to simulate a per-node backend failure). Processed/AtTime/Events delegate to the null source. private sealed class PerTagnameFake(Func rawHandler) : IHistorianDataSource