feat(opcuaserver): process-wide HistoryRead batch limiter — flood degrades to BadTooManyOperations, not pool exhaustion (03/S3)

2026-07-13 12:18:35 -04:00
parent 888cf86655
commit 3cebfae417
3 changed files with 242 additions and 70 deletions
@@ -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"
@@ -221,6 +221,10 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
    /// </summary>
    public TimeSpan HistoryReadDeadline { get; set; } = TimeSpan.FromSeconds(60);

+    /// <summary>Lazily-created process-wide HistoryRead-batch limiter (see <see cref="MaxConcurrentHistoryReadBatches"/>).</summary>
+    private volatile SemaphoreSlim? _historyReadBatchLimiter;
+    private readonly object _historyReadBatchLimiterLock = new();
+
    private volatile IHistoryContinuationStore _historyContinuationStore = new SessionHistoryContinuationStore();

    /// <summary>
@@ -1816,25 +1820,28 @@ 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)
+        WithHistoryReadBatchLimiter(nodesToProcess, results, errors, () =>
        {
-            if (details.IsReadModified)
+            using var deadline = CreateDeadline();
+            var ct = deadline?.Token ?? CancellationToken.None;
+            var work = new List<Func<Task>>(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);
+        });
    }

    /// <inheritdoc />
@@ -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<Func<Task>>(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<Func<Task>>(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);
+        });
    }

    /// <inheritdoc />
@@ -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<DateTime>();
-        using var deadline = CreateDeadline();
-        var ct = deadline?.Token ?? CancellationToken.None;
-        var work = new List<Func<Task>>(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<Func<Task>>(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);
+        });
    }

    /// <inheritdoc />
@@ -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<Func<Task>>(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<Func<Task>>(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);
+        });
    }

    /// <summary>
@@ -2146,6 +2162,68 @@ public sealed class OtOpcUaNodeManager : CustomNodeManager2
        Task.WhenAll(tasks).GetAwaiter().GetResult();
    }

+    /// <summary>
+    ///     Run one HistoryRead arm under the process-wide concurrent-batch limiter (arch-review 03/S3). Each
+    ///     arm waits up to <c>min(HistoryReadDeadline, 5 s)</c> for a slot; on expiry every handle in the
+    ///     batch gets <c>BadTooManyOperations</c> (fast-fail under flood, brief waits under mild contention)
+    ///     and <paramref name="serve"/> is skipped. On acquisition the arm body runs and the slot is
+    ///     released in a <c>finally</c>. Combined with <see cref="HistoryReadBatchParallelism"/> this caps
+    ///     total in-flight gateway reads at
+    ///     <c>MaxConcurrentHistoryReadBatches × HistoryReadBatchParallelism</c>.
+    /// </summary>
+    /// <param name="nodesToProcess">The batch's handles — every slot is failed on limiter saturation.</param>
+    /// <param name="results">The service-level results list.</param>
+    /// <param name="errors">The service-level errors list.</param>
+    /// <param name="serve">The arm body to run once a slot is acquired.</param>
+    private void WithHistoryReadBatchLimiter(
+        List<NodeHandle> nodesToProcess,
+        IList<SdkHistoryReadResult> results,
+        IList<ServiceResult> 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();
+        }
+    }
+
+    /// <summary>
+    ///     Lazily create (double-checked) the process-wide HistoryRead-batch limiter, sized from
+    ///     <see cref="MaxConcurrentHistoryReadBatches"/> at first use so the Host's post-boot property set is
+    ///     respected (the same benign wiring race as <see cref="MaxTieClusterOverfetch"/>).
+    /// </summary>
+    /// <returns>The shared batch limiter semaphore.</returns>
+    private SemaphoreSlim GetHistoryReadBatchLimiter()
+    {
+        var gate = _historyReadBatchLimiter;
+        if (gate is not null) return gate;
+        lock (_historyReadBatchLimiterLock)
+        {
+            return _historyReadBatchLimiter ??= new SemaphoreSlim(Math.Max(1, MaxConcurrentHistoryReadBatches));
+        }
+    }
+
    /// <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
@@ -171,6 +171,55 @@ public sealed class NodeManagerHistoryReadConcurrencyTests : IDisposable
        await host.DisposeAsync();
    }

+    /// <summary>
+    ///     A saturated process-wide batch limiter degrades a second concurrent HistoryRead to
+    ///     <c>BadTooManyOperations</c> (rather than exhausting the SDK request-thread pool); once the parked
+    ///     batch completes the slot frees and a later read succeeds (03/S3).
+    /// </summary>
+    [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();
+    }
+
    /// <summary>Materialises <paramref name="count"/> historized Float variables and returns their NodeIds.</summary>
    private static NodeId[] MaterializeHistorizedNodes(OtOpcUaNodeManager nm, int count)
    {
@@ -306,6 +355,51 @@ public sealed class NodeManagerHistoryReadConcurrencyTests : IDisposable
        public void Dispose() { }
    }

+    /// <summary>A fake whose raw read parks (awaits a <see cref="TaskCompletionSource"/>) until the test
+    /// releases it — used to hold the process-wide batch limiter permit while a second batch is issued.</summary>
+    private sealed class GatedRawFake : IHistorianDataSource
+    {
+        private readonly TaskCompletionSource _gate = new(TaskCreationOptions.RunContinuationsAsynchronously);
+        private int _entered;
+
+        /// <summary>Number of raw reads that reached the backend body.</summary>
+        public int Entered => Volatile.Read(ref _entered);
+
+        /// <summary>Releases the parked read so its awaiting batch can complete.</summary>
+        public void Release() => _gate.TrySetResult();
+
+        public async Task<HistorianRead> 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<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() { }
+    }
+
    /// <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