feat(opcuaserver): per-request HistoryRead deadline — hung backends surface BadTimeout (03/S3)

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"] }
   ],

src/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer/OtOpcUaNodeManager.cs

@@ -1816,6 +1816,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
         IDictionary<NodeId, NodeState> cache)
     {
         var session = context.OperationContext?.Session;
+        using var deadline = CreateDeadline();
+        var ct = deadline?.Token ?? CancellationToken.None;
         var work = new List<Func<Task>>(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<Func<Task>>(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<DateTime>();
+        using var deadline = CreateDeadline();
+        var ct = deadline?.Token ?? CancellationToken.None;
         var work = new List<Func<Task>>(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<Func<Task>>(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();
     }
 
+    /// <summary>
+    ///     Create the per-request deadline source for a HistoryRead arm (arch-review 03/S3): a
+    ///     <see cref="CancellationTokenSource"/> that fires after <see cref="HistoryReadDeadline"/>, so a
+    ///     single request can never hold an SDK request thread longer than the deadline regardless of node
+    ///     count. Returns <c>null</c> when the deadline is non-positive (unbounded — the arm then passes
+    ///     <see cref="CancellationToken.None"/>), matching the <c>ServerHistorianOptions</c> warning.
+    /// </summary>
+    /// <returns>The deadline source, or <c>null</c> for an unbounded (non-positive) deadline.</returns>
+    private CancellationTokenSource? CreateDeadline() =>
+        HistoryReadDeadline > TimeSpan.Zero ? new CancellationTokenSource(HistoryReadDeadline) : null;
+
     /// <summary>Awaits one work item under a semaphore permit, releasing it when the item completes.</summary>
     /// <param name="gate">The concurrency-limiting semaphore.</param>
     /// <param name="work">The per-node work item.</param>

tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests/NodeManagerHistoryReadConcurrencyTests.cs

@@ -140,6 +140,37 @@ public sealed class NodeManagerHistoryReadConcurrencyTests : IDisposable
         await host.DisposeAsync();
     }
 
+    /// <summary>
+    ///     A per-request deadline bounds a hung backend: reads that only complete on token cancellation
+    ///     surface <c>BadTimeout</c> for every node and the call returns within the deadline (plus slack) —
+    ///     it does not hang an SDK request thread indefinitely (03/S3).
+    /// </summary>
+    [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();
+    }
+
     /// <summary>Materialises <paramref name="count"/> historized Float variables and returns their NodeIds.</summary>
     private static NodeId[] MaterializeHistorizedNodes(OtOpcUaNodeManager nm, int count)
     {
@@ -240,6 +271,41 @@ public sealed class NodeManagerHistoryReadConcurrencyTests : IDisposable
         }
     }
 
+    /// <summary>A fake whose reads complete ONLY on token cancellation — used to prove the per-request
+    /// deadline surfaces <c>BadTimeout</c> rather than hanging.</summary>
+    private sealed class HangingHistorianDataSource : IHistorianDataSource
+    {
+        private static async Task<HistorianRead> HangAsync(CancellationToken ct)
+        {
+            await Task.Delay(Timeout.Infinite, ct).ConfigureAwait(false);
+            return new HistorianRead(Array.Empty<DataValueSnapshot>(), null);
+        }
+
+        public Task<HistorianRead> ReadRawAsync(
+            string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode,
+            CancellationToken cancellationToken) => HangAsync(cancellationToken);
+
+        public Task<HistorianRead> ReadProcessedAsync(
+            string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
+            HistoryAggregateType aggregate, CancellationToken cancellationToken) => HangAsync(cancellationToken);
+
+        public Task<HistorianRead> ReadAtTimeAsync(
+            string fullReference, IReadOnlyList<DateTime> timestampsUtc, CancellationToken cancellationToken) =>
+            HangAsync(cancellationToken);
+
+        public async Task<HistoricalEventsResult> ReadEventsAsync(
+            string? sourceName, DateTime startUtc, DateTime endUtc, int maxEvents,
+            CancellationToken cancellationToken)
+        {
+            await Task.Delay(Timeout.Infinite, cancellationToken).ConfigureAwait(false);
+            return new HistoricalEventsResult(Array.Empty<HistoricalEvent>(), null);
+        }
+
+        public HistorianHealthSnapshot GetHealthSnapshot() => NullHistorianDataSource.Instance.GetHealthSnapshot();
+
+        public void Dispose() { }
+    }
+
     /// <summary>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.</summary>
     private sealed class PerTagnameFake(Func<string, HistorianRead> rawHandler) : IHistorianDataSource