refactor(opcuaserver): async-ify ServeRawPaged + HistoryReadEvents internals (03/S3)

archreview/plans/R2-08-ldap-historyread-async-plan.md.tasks.json

@@ -10,7 +10,7 @@
     { "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": "completed", "blockedBy": [] },
     { "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": "pending", "blockedBy": ["R2-08-T9"] },
+    { "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": "pending", "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-T13", "subject": "Process-wide concurrent-batch limiter — saturation degrades to BadTooManyOperations (lazy gate, bounded wait)", "status": "pending", "blockedBy": ["R2-08-T12"] },

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

@@ -1825,9 +1825,11 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
                 continue;
             }
 
-            ServeRawPaged(
+            // Interim: still bridged sequentially per node (bounded fan-out lands in a later slice).
+            ServeRawPagedAsync(
                 handle, session, nodesToRead, results, errors,
-                details.StartTime, details.EndTime, details.NumValuesPerNode);
+                details.StartTime, details.EndTime, details.NumValuesPerNode,
+                CancellationToken.None).GetAwaiter().GetResult();
         }
     }
 
@@ -1922,10 +1924,39 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
                 continue;
             }
 
+            // Interim: still bridged sequentially per node (bounded fan-out lands in a later slice).
+            ServeEventsAsync(handle, sourceName, details, selectClauses, results, errors, CancellationToken.None)
+                .GetAwaiter().GetResult();
+        }
+    }
+
+    /// <summary>
+    ///     Serve one registered event-history notifier handle: reads the source's events over the request
+    ///     window, projects them onto the select clauses, and fills the service-level results/errors slots.
+    ///     Per-node error isolation matches <see cref="ServeNodeAsync"/> — a backend throw becomes a Bad
+    ///     status for THIS node only and never throws out of the batch; a deadline cancellation surfaces
+    ///     <c>BadTimeout</c>.
+    /// </summary>
+    /// <param name="handle">The notifier node handle; <c>handle.Index</c> indexes results/errors.</param>
+    /// <param name="sourceName">The registered historian event-source name for this notifier.</param>
+    /// <param name="details">The event-read details (window + count cap).</param>
+    /// <param name="selectClauses">The event-filter select clauses (the fields to emit, in order).</param>
+    /// <param name="results">The service-level results list to fill at <c>handle.Index</c>.</param>
+    /// <param name="errors">The service-level errors list to fill at <c>handle.Index</c>.</param>
+    /// <param name="ct">Per-request deadline token.</param>
+    private async Task ServeEventsAsync(
+        NodeHandle handle,
+        string? sourceName,
+        ReadEventDetails details,
+        SimpleAttributeOperandCollection selectClauses,
+        IList<SdkHistoryReadResult> results,
+        IList<ServiceResult> errors,
+        CancellationToken ct)
+    {
         try
         {
-                // NOT under the node-manager Lock — block-bridging the async source is safe here.
+            // NOT under the node-manager Lock — awaiting the async source is safe here.
-                var sourceResult = _historianDataSource.ReadEventsAsync(
+            var sourceResult = await _historianDataSource.ReadEventsAsync(
                 sourceName,
                 details.StartTime,
                 details.EndTime,
@@ -1934,7 +1965,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
                 // backend-default-cap sentinel instead would silently truncate a "give me everything"
                 // events read at the backend default. A positive cap passes through (clamped).
                 EventMaxEvents(details.NumValuesPerNode),
-                    CancellationToken.None).GetAwaiter().GetResult();
+                ct).ConfigureAwait(false);
 
             var historyEvent = ProjectEvents(sourceResult.Events, selectClauses);
             results[handle.Index] = new SdkHistoryReadResult
@@ -1947,10 +1978,16 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
             };
             errors[handle.Index] = ServiceResult.Good;
         }
+        catch (OperationCanceledException)
+        {
+            // The per-request deadline elapsed while reading this notifier — BadTimeout for THIS node only.
+            errors[handle.Index] = StatusCodes.BadTimeout;
+            results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadTimeout };
+        }
         catch (Exception ex)
         {
             // One node's backend failure must not throw out of the batch — surface Bad for THIS node
-                // only. This manager carries no ILogger, so log via the SDK's static trace (see ServeNode).
+            // only. This manager carries no ILogger, so log via the SDK's static trace (see ServeNodeAsync).
 #pragma warning disable CS0618 // Type or member is obsolete
             Utils.LogError(ex, "OtOpcUaNodeManager: HistoryReadEvents failed for node {0}", handle.NodeId);
 #pragma warning restore CS0618
@@ -1958,7 +1995,6 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
             results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadHistoryOperationUnsupported };
         }
     }
-    }
 
     /// <summary>Clamp a <see cref="uint"/> request cap to a non-negative <see cref="int"/> for the
     /// event-read surface (whose <c>maxEvents</c> is signed): values above <see cref="int.MaxValue"/>
@@ -2149,7 +2185,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
     /// <param name="startTimeUtc">The request window's (inclusive) lower bound, used for a fresh read.</param>
     /// <param name="endUtc">The (inclusive) upper bound of the read window; unchanged across pages.</param>
     /// <param name="numValuesPerNode">The client's per-page cap; <c>0</c> means "all values, no paging".</param>
-    private void ServeRawPaged(
+    private async Task ServeRawPagedAsync(
         NodeHandle handle,
         ISession? session,
         IList<HistoryReadValueId> nodesToRead,
@@ -2157,7 +2193,8 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
         IList<ServiceResult> errors,
         DateTime startTimeUtc,
         DateTime endUtc,
-        uint numValuesPerNode)
+        uint numValuesPerNode,
+        CancellationToken ct)
     {
         var inboundCp = nodesToRead[handle.Index].ContinuationPoint;
 
@@ -2199,10 +2236,10 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
                 startUtc = startTimeUtc;
             }
 
-            // HistoryRead is NOT under the node-manager Lock — block-bridging the async source is safe.
+            // HistoryRead is NOT under the node-manager Lock — awaiting the async source is safe.
-            var sourceResult = HistorianDataSource
+            var sourceResult = await HistorianDataSource
-                .ReadRawAsync(tagname, startUtc, endUtc, numValuesPerNode, CancellationToken.None)
+                .ReadRawAsync(tagname, startUtc, endUtc, numValuesPerNode, ct)
-                .GetAwaiter().GetResult();
+                .ConfigureAwait(false);
 
             var backendFull = HistoryPaging.IsFullPage(sourceResult.Samples.Count, numValuesPerNode);
 
@@ -2232,9 +2269,9 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
                 // Compute in uint so +1 can never wrap: Math.Max(1,…) also guards against a zero/negative
                 // config value slipping through (Validate() warns but does not block startup).
                 var overfetchCap = (uint)Math.Max(1, MaxTieClusterOverfetch) + 1u;
-                var cluster = HistorianDataSource
+                var cluster = (await HistorianDataSource
-                    .ReadRawAsync(tagname, startUtc, startUtc, overfetchCap, CancellationToken.None)
+                    .ReadRawAsync(tagname, startUtc, startUtc, overfetchCap, ct)
-                    .GetAwaiter().GetResult().Samples;
+                    .ConfigureAwait(false)).Samples;
 
                 // Absurd burst: more ties than we're willing to buffer in memory. Preserve today's loud-fail
                 // for that node rather than over-fetch an unbounded cluster; the operator's remedy is a
@@ -2306,6 +2343,12 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
             };
             errors[handle.Index] = ServiceResult.Good;
         }
+        catch (OperationCanceledException)
+        {
+            // The per-request deadline elapsed while paging this node — BadTimeout for THIS node only.
+            errors[handle.Index] = StatusCodes.BadTimeout;
+            results[handle.Index] = new SdkHistoryReadResult { StatusCode = StatusCodes.BadTimeout };
+        }
         catch (Exception ex)
         {
             // One node's backend failure must not throw out of the batch — Bad for THIS node only.