feat(opcuaserver): bounded per-node parallel HistoryRead within a batch (03/S3)

2026-07-13 12:09:53 -04:00
parent cef7d9b281
commit 8c365da754
3 changed files with 175 additions and 14 deletions
@@ -11,7 +11,7 @@
    { "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": "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-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-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"] }
@@ -1816,6 +1816,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
        IDictionary<NodeId, NodeState> cache)
    {
        var session = context.OperationContext?.Session;
+        var work = new List<Func<Task>>(nodesToProcess.Count);
        foreach (var handle in nodesToProcess)
        {
            if (details.IsReadModified)
@@ -1825,12 +1826,13 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
                continue;
            }

-            // Interim: still bridged sequentially per node (bounded fan-out lands in a later slice).
-            ServeRawPagedAsync(
+            work.Add(() => ServeRawPagedAsync(
                handle, session, nodesToRead, results, errors,
                details.StartTime, details.EndTime, details.NumValuesPerNode,
-                CancellationToken.None).GetAwaiter().GetResult();
+                CancellationToken.None));
        }
+
+        RunBounded(work, HistoryReadBatchParallelism);
    }

    /// <inheritdoc />
@@ -1846,6 +1848,7 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
    {
        // OPC UA ProcessingInterval is a Duration in milliseconds — convert once per batch.
        var interval = TimeSpan.FromMilliseconds(details.ProcessingInterval);
+        var work = new List<Func<Task>>(nodesToProcess.Count);
        foreach (var handle in nodesToProcess)
        {
            // AggregateType is a per-node parallel collection (same length as nodesToRead, enforced by
@@ -1863,15 +1866,17 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
            // 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).
-            // Interim: still bridged sequentially per node (bounded fan-out lands in a later slice).
-            ServeNodeAsync(handle, results, errors, (source, tagname, token) => source.ReadProcessedAsync(
+            var agg = aggregate.Value;
+            work.Add(() => ServeNodeAsync(handle, results, errors, (source, tagname, token) => source.ReadProcessedAsync(
                tagname,
                details.StartTime,
                details.EndTime,
                interval,
-                aggregate.Value,
-                token), CancellationToken.None).GetAwaiter().GetResult();
+                agg,
+                token), CancellationToken.None));
        }
+
+        RunBounded(work, HistoryReadBatchParallelism);
    }

    /// <inheritdoc />
@@ -1887,14 +1892,16 @@ 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>();
+        var work = new List<Func<Task>>(nodesToProcess.Count);
        foreach (var handle in nodesToProcess)
        {
-            // Interim: still bridged sequentially per node (bounded fan-out lands in a later slice).
-            ServeNodeAsync(handle, results, errors, (source, tagname, token) => source.ReadAtTimeAsync(
+            work.Add(() => ServeNodeAsync(handle, results, errors, (source, tagname, token) => source.ReadAtTimeAsync(
                tagname,
                timestamps,
-                token), CancellationToken.None).GetAwaiter().GetResult();
+                token), CancellationToken.None));
        }
+
+        RunBounded(work, HistoryReadBatchParallelism);
    }

    /// <inheritdoc />
@@ -1911,6 +1918,7 @@ 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();

+        var work = new List<Func<Task>>(nodesToProcess.Count);
        foreach (var handle in nodesToProcess)
        {
            var idString = handle.NodeId.Identifier?.ToString();
@@ -1924,10 +1932,11 @@ 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();
+            var sn = sourceName;
+            work.Add(() => ServeEventsAsync(handle, sn, details, selectClauses, results, errors, CancellationToken.None));
        }
+
+        RunBounded(work, HistoryReadBatchParallelism);
    }

    /// <summary>
@@ -2099,6 +2108,52 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
    /// <param name="errors">The service-level errors list to fill at <c>handle.Index</c>.</param>
    /// <param name="read">Invokes the resolved data-source read with the resolved tagname; only called
    /// once the tagname is confirmed present.</param>
+    /// <summary>
+    ///     Run a batch of per-node HistoryRead work items with BOUNDED parallelism (arch-review 03/S3),
+    ///     block-bridged ONCE at the arm boundary. <b>Contract:</b> every work item must be fully
+    ///     self-contained — it fills its own results/errors slot and NEVER throws (all per-node handlers
+    ///     catch every exception, including cancellation, and surface a Bad status for that node only). That
+    ///     invariant is what makes the single <c>GetAwaiter().GetResult()</c> safe: <see cref="Task.WhenAll(Task[])"/>
+    ///     can never fault, so it cannot throw out of the synchronous SDK override. A degree ≤ 1 (or a
+    ///     single item) runs sequentially — no semaphore, no extra tasks.
+    /// </summary>
+    /// <param name="work">The per-node work items (index-disjoint writes into the shared results/errors).</param>
+    /// <param name="degree">Max concurrent work items (<see cref="HistoryReadBatchParallelism"/>).</param>
+    private static void RunBounded(IReadOnlyList<Func<Task>> work, int degree)
+    {
+        if (work.Count == 0) return;
+        if (degree <= 1 || work.Count == 1)
+        {
+            foreach (var item in work)
+                item().GetAwaiter().GetResult();
+            return;
+        }
+
+        using var gate = new SemaphoreSlim(degree);
+        var tasks = new Task[work.Count];
+        for (var i = 0; i < work.Count; i++)
+            tasks[i] = RunGatedAsync(gate, work[i]);
+
+        // Safe: every body is self-contained + never throws, so WhenAll never faults (see contract above).
+        Task.WhenAll(tasks).GetAwaiter().GetResult();
+    }
+
+    /// <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>
+    private static async Task RunGatedAsync(SemaphoreSlim gate, Func<Task> work)
+    {
+        await gate.WaitAsync().ConfigureAwait(false);
+        try
+        {
+            await work().ConfigureAwait(false);
+        }
+        finally
+        {
+            gate.Release();
+        }
+    }
+
    private async Task ServeNodeAsync(
        NodeHandle handle,
        IList<SdkHistoryReadResult> results,
@@ -61,6 +61,85 @@ public sealed class NodeManagerHistoryReadConcurrencyTests : IDisposable
        await host.DisposeAsync();
    }

+    /// <summary>
+    ///     Parallelism is bounded ABOVE by <c>HistoryReadBatchParallelism</c>: with the property set to 3 and
+    ///     12 nodes in flight, no more than 3 reads are ever concurrent (03/S3).
+    /// </summary>
+    [Fact]
+    public async Task Parallelism_is_bounded_above()
+    {
+        var (host, server) = await BootAsync();
+        var nm = server.NodeManager!;
+        nm.HistoryReadBatchParallelism = 3;
+        var fake = new ConcurrencyTrackingHistorianDataSource(TimeSpan.FromMilliseconds(100));
+        nm.HistorianDataSource = fake;
+
+        var nodeIds = MaterializeHistorizedNodes(nm, count: 12);
+        var details = new ReadRawModifiedDetails
+        {
+            StartTime = DateTime.UtcNow.AddHours(-1),
+            EndTime = DateTime.UtcNow,
+            NumValuesPerNode = 10,
+            IsReadModified = false,
+        };
+
+        await Task.Run(() => InvokeHistoryRead(server, nm, details, nodeIds), Ct)
+            .WaitAsync(TimeSpan.FromSeconds(15), Ct);
+
+        fake.MaxObservedConcurrency.ShouldBeGreaterThanOrEqualTo(2);
+        fake.MaxObservedConcurrency.ShouldBeLessThanOrEqualTo(3);
+
+        await host.DisposeAsync();
+    }
+
+    /// <summary>
+    ///     One node's backend throw does not poison a parallel batch: that node surfaces
+    ///     <c>BadHistoryOperationUnsupported</c> while the other seven return Good (per-node isolation holds
+    ///     under parallelism — 03/S3).
+    /// </summary>
+    [Fact]
+    public async Task One_node_failure_does_not_poison_a_parallel_batch()
+    {
+        var (host, server) = await BootAsync();
+        var nm = server.NodeManager!;
+
+        var src = DateTime.UtcNow.AddSeconds(-5);
+        var srv = DateTime.UtcNow;
+        // Node index 3's tagname throws; all others return a single good sample.
+        var fake = new PerTagnameFake(tagname => tagname == "WW.Tag3"
+            ? throw new InvalidOperationException("backend boom for Tag3")
+            : new HistorianRead(new[] { new DataValueSnapshot(1.0f, StatusCodes.Good, src, srv) }, null));
+        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 (results, errors) = await Task.Run(() => InvokeHistoryRead(server, nm, details, nodeIds), Ct)
+            .WaitAsync(TimeSpan.FromSeconds(15), Ct);
+
+        for (var i = 0; i < 8; i++)
+        {
+            if (i == 3)
+            {
+                StatusCode.IsBad(errors[i].StatusCode).ShouldBeTrue();
+                errors[i].StatusCode.Code.ShouldBe(StatusCodes.BadHistoryOperationUnsupported);
+            }
+            else
+            {
+                errors[i].StatusCode.Code.ShouldBe(StatusCodes.Good);
+                results[i].StatusCode.Code.ShouldBe(StatusCodes.Good);
+            }
+        }
+
+        await host.DisposeAsync();
+    }
+
    /// <summary>Materialises <paramref name="count"/> historized Float variables and returns their NodeIds.</summary>
    private static NodeId[] MaterializeHistorizedNodes(OtOpcUaNodeManager nm, int count)
    {
@@ -161,6 +240,33 @@ public sealed class NodeManagerHistoryReadConcurrencyTests : IDisposable
        }
    }

+    /// <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
+    {
+        public Task<HistorianRead> ReadRawAsync(
+            string fullReference, DateTime startUtc, DateTime endUtc, uint maxValuesPerNode,
+            CancellationToken cancellationToken) => Task.FromResult(rawHandler(fullReference));
+
+        public Task<HistorianRead> ReadProcessedAsync(
+            string fullReference, DateTime startUtc, DateTime endUtc, TimeSpan interval,
+            HistoryAggregateType aggregate, CancellationToken cancellationToken) =>
+            NullHistorianDataSource.Instance.ReadProcessedAsync(fullReference, startUtc, endUtc, interval, aggregate, cancellationToken);
+
+        public Task<HistorianRead> ReadAtTimeAsync(
+            string fullReference, IReadOnlyList<DateTime> timestampsUtc, CancellationToken cancellationToken) =>
+            NullHistorianDataSource.Instance.ReadAtTimeAsync(fullReference, timestampsUtc, cancellationToken);
+
+        public Task<HistoricalEventsResult> 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() { }
+    }
+
    private async Task<(OpcUaApplicationHost Host, OtOpcUaSdkServer Server)> BootAsync()
    {
        var host = new OpcUaApplicationHost(