From 888cf86655a60a19e12ce483c738a3b846237fb1 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Mon, 13 Jul 2026 12:13:38 -0400 Subject: [PATCH] =?UTF-8?q?feat(opcuaserver):=20per-request=20HistoryRead?= =?UTF-8?q?=20deadline=20=E2=80=94=20hung=20backends=20surface=20BadTimeou?= =?UTF-8?q?t=20(03/S3)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ...-ldap-historyread-async-plan.md.tasks.json | 2 +- .../OtOpcUaNodeManager.cs | 27 ++++++-- .../NodeManagerHistoryReadConcurrencyTests.cs | 66 +++++++++++++++++++ 3 files changed, 90 insertions(+), 5 deletions(-) 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 4eb04faf..2e262a98 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 @@ -12,7 +12,7 @@ { "id": "R2-08-T9", "subject": "Async-ify ServeNode + Processed/AtTime arms (ct threading, OCE=>BadTimeout, behavior-neutral)", "status": "completed", "blockedBy": ["R2-08-T7", "R2-08-T8"] }, { "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": "pending", "blockedBy": ["R2-08-T11"] }, + { "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-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"] } ], diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs index 7539af9c..e059d535 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs @@ -1816,6 +1816,8 @@ 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) { @@ -1829,7 +1831,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 work.Add(() => ServeRawPagedAsync( handle, session, nodesToRead, results, errors, details.StartTime, details.EndTime, details.NumValuesPerNode, - CancellationToken.None)); + ct)); } RunBounded(work, HistoryReadBatchParallelism); @@ -1848,6 +1850,8 @@ 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) { @@ -1873,7 +1877,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 details.EndTime, interval, agg, - token), CancellationToken.None)); + token), ct)); } RunBounded(work, HistoryReadBatchParallelism); @@ -1892,13 +1896,15 @@ 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) { work.Add(() => ServeNodeAsync(handle, results, errors, (source, tagname, token) => source.ReadAtTimeAsync( tagname, timestamps, - token), CancellationToken.None)); + token), ct)); } RunBounded(work, HistoryReadBatchParallelism); @@ -1918,6 +1924,8 @@ 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) { @@ -1933,7 +1941,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 } var sn = sourceName; - work.Add(() => ServeEventsAsync(handle, sn, details, selectClauses, results, errors, CancellationToken.None)); + work.Add(() => ServeEventsAsync(handle, sn, details, selectClauses, results, errors, ct)); } RunBounded(work, HistoryReadBatchParallelism); @@ -2138,6 +2146,17 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2 Task.WhenAll(tasks).GetAwaiter().GetResult(); } + /// + /// Create the per-request deadline source for a HistoryRead arm (arch-review 03/S3): a + /// that fires after , so a + /// single request can never hold an SDK request thread longer than the deadline regardless of node + /// count. Returns null when the deadline is non-positive (unbounded — the arm then passes + /// ), matching the ServerHistorianOptions warning. + /// + /// The deadline source, or null for an unbounded (non-positive) deadline. + private CancellationTokenSource? CreateDeadline() => + HistoryReadDeadline > TimeSpan.Zero ? new CancellationTokenSource(HistoryReadDeadline) : null; + /// Awaits one work item under a semaphore permit, releasing it when the item completes. /// The concurrency-limiting semaphore. /// The per-node work item. 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 78bccd9a..f65c45f8 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadConcurrencyTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadConcurrencyTests.cs @@ -140,6 +140,37 @@ public sealed class NodeManagerHistoryReadConcurrencyTests : IDisposable await host.DisposeAsync(); } + /// + /// A per-request deadline bounds a hung backend: reads that only complete on token cancellation + /// surface BadTimeout for every node and the call returns within the deadline (plus slack) — + /// it does not hang an SDK request thread indefinitely (03/S3). + /// + [Fact] + public async Task Deadline_bounds_a_hung_backend() + { + var (host, server) = await BootAsync(); + var nm = server.NodeManager!; + nm.HistoryReadDeadline = TimeSpan.FromMilliseconds(500); + nm.HistorianDataSource = new HangingHistorianDataSource(); + + var nodeIds = MaterializeHistorizedNodes(nm, count: 4); + var details = new ReadRawModifiedDetails + { + StartTime = DateTime.UtcNow.AddHours(-1), + EndTime = DateTime.UtcNow, + NumValuesPerNode = 10, + IsReadModified = false, + }; + + var (_, errors) = await Task.Run(() => InvokeHistoryRead(server, nm, details, nodeIds), Ct) + .WaitAsync(TimeSpan.FromSeconds(10), Ct); + + for (var i = 0; i < 4; i++) + errors[i].StatusCode.Code.ShouldBe(StatusCodes.BadTimeout); + + await host.DisposeAsync(); + } + /// Materialises historized Float variables and returns their NodeIds. private static NodeId[] MaterializeHistorizedNodes(OtOpcUaNodeManager nm, int count) { @@ -240,6 +271,41 @@ public sealed class NodeManagerHistoryReadConcurrencyTests : IDisposable } } + /// A fake whose reads complete ONLY on token cancellation — used to prove the per-request + /// deadline surfaces BadTimeout rather than hanging. + private sealed class HangingHistorianDataSource : IHistorianDataSource + { + private static async Task HangAsync(CancellationToken ct) + { + await Task.Delay(Timeout.Infinite, ct).ConfigureAwait(false); + return new HistorianRead(Array.Empty(), null); + } + + public Task ReadRawAsync( + string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode, + CancellationToken cancellationToken) => HangAsync(cancellationToken); + + public Task ReadProcessedAsync( + string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval, + HistoryAggregateType aggregate, CancellationToken cancellationToken) => HangAsync(cancellationToken); + + public Task ReadAtTimeAsync( + string fullReference, IReadOnlyList timestampsUtc, CancellationToken cancellationToken) => + HangAsync(cancellationToken); + + public async Task ReadEventsAsync( + string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents, + CancellationToken cancellationToken) + { + await Task.Delay(Timeout.Infinite, cancellationToken).ConfigureAwait(false); + return new HistoricalEventsResult(Array.Empty(), null); + } + + 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