perf(historian): bounded-parallel per-tag drain (DrainConcurrency, default 8)

src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationOptions.cs

@@ -57,6 +57,21 @@ public sealed class ContinuousHistorizationOptions
     /// <summary>Steady drain cadence in seconds (also the post-success reschedule). Defaults to 2.</summary>
     public double DrainIntervalSeconds { get; init; } = 2;
 
+    /// <summary>
+    ///     Maximum historian tags a single drain pass writes concurrently. Defaults to 8; clamped to at
+    ///     least 1 by the recorder (1 restores the fully sequential drain).
+    ///     <para>
+    ///         One peeked batch is grouped per historian tag and each group is an independent gateway
+    ///         round-trip, so a sequential pass costs <c>tags × latency</c> — with a wide batch that is the
+    ///         dominant term in how fast the outbox can drain. The window bounds how many of those
+    ///         round-trips are outstanding at once, so a large tag count cannot flood the gateway. Nothing
+    ///         about the ack changes: the pass is still acked all-or-nothing after every group has
+    ///         reported, and the Gitea #538 per-tag quarantine still keys on "another group wrote this
+    ///         pass".
+    ///     </para>
+    /// </summary>
+    public int DrainConcurrency { get; init; } = 8;
+
     /// <summary>
     ///     Maximum un-acked outbox entries before the drop-oldest capacity policy kicks in; <c>0</c>
     ///     means unbounded.
@@ -102,6 +117,8 @@ public sealed class ContinuousHistorizationOptions
             warnings.Add("ContinuousHistorization:OutboxPath is empty while historization is enabled — the durable outbox has no directory to persist to; host startup will fail when the outbox is constructed. Set an absolute path on durable storage.");
         if (string.Equals(CommitMode, "Periodic", StringComparison.OrdinalIgnoreCase) && CommitIntervalMs <= 0)
             warnings.Add($"ContinuousHistorization:CommitIntervalMs is {CommitIntervalMs} — must be > 0 in Periodic commit mode; the periodic-commit loop cannot run.");
+        if (DrainConcurrency < 1)
+            warnings.Add($"ContinuousHistorization:DrainConcurrency is {DrainConcurrency} — must be >= 1; the recorder clamps it to 1 (a fully sequential drain).");
         return warnings;
     }
 }

src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Historian/ContinuousHistorizationRecorder.cs

@@ -135,6 +135,11 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
     /// writing successfully — before its queued entries are abandoned so the FIFO head can advance.</summary>
     private readonly int _maxWriteAttempts;
 
+    /// <summary>Max per-tag groups one drain pass writes concurrently. The groups in a pass are
+    /// independent gateway round-trips, so a sequential pass costs <c>tags × latency</c>; this bounds how
+    /// many are outstanding at once so a wide batch cannot flood the gateway.</summary>
+    private readonly int _drainConcurrency;
+
     /// <summary>Consecutive failed write passes per historian tag, counted only across passes that
     /// proved the backend reachable. Reset on the tag's first success. Mutated exclusively inside
     /// <see cref="DrainOnceAsync"/>, which <see cref="_draining"/> keeps single-flight.</summary>
@@ -174,6 +179,8 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
     /// and for tests.</param>
     /// <param name="maxWriteAttempts">Gitea #538: consecutive failed write passes one tag may
     /// accumulate, while other tags are succeeding, before its entries are abandoned.</param>
+    /// <param name="drainConcurrency">Max historian tags one drain pass writes concurrently (must be
+    /// positive; 1 is the fully sequential drain).</param>
     /// <returns>Props for the actor.</returns>
     public static Props Props(
         IActorRef dependencyMux,
@@ -185,10 +192,11 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
         TimeSpan? minBackoff = null,
         TimeSpan? maxBackoff = null,
         Func<bool>? captureGate = null,
-        int maxWriteAttempts = 5) =>
+        int maxWriteAttempts = 5,
+        int drainConcurrency = 8) =>
         Akka.Actor.Props.Create(() => new ContinuousHistorizationRecorder(
             dependencyMux, writer, outbox, historizedRefs, drainBatchSize, drainInterval, minBackoff, maxBackoff,
-            captureGate, maxWriteAttempts));
+            captureGate, maxWriteAttempts, drainConcurrency));
 
     /// <summary>Initializes a new instance of the <see cref="ContinuousHistorizationRecorder"/> class.</summary>
     /// <param name="dependencyMux">The per-node dependency mux to register historized-ref interest with.</param>
@@ -204,6 +212,8 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
     /// <param name="maxWriteAttempts">Gitea #538: consecutive failed write passes one tag may
     /// accumulate, while other tags are succeeding, before its entries are abandoned. Clamped to at
     /// least 1.</param>
+    /// <param name="drainConcurrency">Max historian tags one drain pass writes concurrently. Clamped to
+    /// at least 1; 1 is the fully sequential drain.</param>
     public ContinuousHistorizationRecorder(
         IActorRef dependencyMux,
         IHistorianValueWriter writer,
@@ -214,7 +224,8 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
         TimeSpan? minBackoff = null,
         TimeSpan? maxBackoff = null,
         Func<bool>? captureGate = null,
-        int maxWriteAttempts = 5)
+        int maxWriteAttempts = 5,
+        int drainConcurrency = 8)
     {
         _dependencyMux = dependencyMux ?? throw new ArgumentNullException(nameof(dependencyMux));
         _writer = writer ?? throw new ArgumentNullException(nameof(writer));
@@ -238,6 +249,7 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
         // gate), and a test that does not care about the role.
         _captureGate = captureGate ?? (static () => true);
         _maxWriteAttempts = maxWriteAttempts > 0 ? maxWriteAttempts : 1;
+        _drainConcurrency = drainConcurrency > 0 ? drainConcurrency : 1;
 
         ReceiveAsync<VirtualTagActor.DependencyValueChanged>(OnValueChangedAsync);
         ReceiveAsync<VirtualTagActor.DependencyValueBatch>(OnValueBatchAsync);
@@ -537,35 +549,68 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
             // id, so partial per-tag acking could drop un-written older entries; the ack stays
             // all-or-nothing — only when nothing is left blocking do we truncate the whole peeked
             // prefix (one commit).
-            var succeeded = 0;
+            //
-            var failedGroups = new List<(string Tag, int Count)>();
+            // The groups are INDEPENDENT gateway round-trips against distinct historian tags, so they run
-            foreach (IGrouping<string, HistorizationOutboxEntry> group in
+            // concurrently under a bounded window rather than one after another: a sequential pass costs
-                     batch.GroupBy(e => e.Tag, StringComparer.Ordinal))
+            // `tags × latency`, which is the dominant term in how fast a wide batch can drain. The window
-            {
+            // (_drainConcurrency) is what stops a large tag count from flooding the gateway; a window of 1
-                var values = group
+            // is exactly the old sequential loop.
-                    .Select(e => new HistorizationValue(e.TimestampUtc, e.NumericValue, e.Quality))
+            var groups = batch
-                    .ToList();
+                .GroupBy(e => e.Tag, StringComparer.Ordinal)
+                .Select(g => (
+                    Key: g.Key,
+                    Values: (IReadOnlyList<HistorizationValue>)g
+                        .Select(e => new HistorizationValue(e.TimestampUtc, e.NumericValue, e.Quality))
+                        .ToList()))
+                .ToList();
 
-                bool ok;
+            using var window = new SemaphoreSlim(_drainConcurrency);
+            var tasks = groups.Select(async g =>
+            {
+                await window.WaitAsync(ct).ConfigureAwait(false);
                 try
                 {
-                    ok = await _writer.WriteLiveValuesAsync(group.Key, values, ct).ConfigureAwait(false);
+                    bool ok;
-                }
+                    try
-                catch
+                    {
-                {
+                        ok = await _writer.WriteLiveValuesAsync(g.Key, g.Values, ct).ConfigureAwait(false);
-                    // The writer is non-throwing by contract; guard defensively so one tag's fault
+                    }
-                    // cannot fail the whole drain task.
+                    catch
-                    ok = false;
+                    {
-                }
+                        // The writer is non-throwing by contract; guard defensively so one tag's fault
+                        // cannot fail the whole drain task (and, now, cannot fault the WhenAll either).
+                        ok = false;
+                    }
 
+                    return (Tag: g.Key, Count: g.Values.Count, Ok: ok);
+                }
+                finally
+                {
+                    window.Release();
+                }
+            }).ToList();
+
+            // Task.WhenAll waits for EVERY task to complete before it faults, so no lambda can still be
+            // touching `window` when the using-scope disposes it — including on the cancellation path,
+            // where the not-yet-entered tasks fault out of WaitAsync without reaching the finally.
+            (string Tag, int Count, bool Ok)[] outcomes = await Task.WhenAll(tasks).ConfigureAwait(false);
+
+            // Fold the outcomes on the single logical drain flow — NEVER inside the parallel lambdas.
+            // _tagWriteFailures is actor state (kept single-flight by _draining); mutating it from N
+            // concurrent tasks would corrupt the Gitea #538 counters. Nothing below observes completion
+            // ORDER, so the fold is deterministic whatever order the writes finished in.
+            var succeeded = 0;
+            var failedGroups = new List<(string Tag, int Count)>();
+            foreach ((string tag, int count, bool ok) in outcomes)
+            {
                 if (ok)
                 {
                     succeeded++;
-                    _tagWriteFailures.Remove(group.Key);
+                    _tagWriteFailures.Remove(tag);
                 }
                 else
                 {
-                    failedGroups.Add((group.Key, values.Count));
+                    failedGroups.Add((tag, count));
                 }
             }
 

src/Server/ZB.MOM.WW.OtOpcUa.Runtime/ServiceCollectionExtensions.cs

@@ -477,7 +477,10 @@ public static class ServiceCollectionExtensions
                             captureGate: resolver.GetService<IHistorizationRoleView>() is { } histRoleView
                                 ? () => histRoleView.ShouldRecordHistorizedValues
                                 : null,
-                            maxWriteAttempts: continuousOptions.MaxWriteAttempts),
+                            maxWriteAttempts: continuousOptions.MaxWriteAttempts,
+                            // Bounded-parallel per-tag drain: the groups in one pass are independent
+                            // gateway round-trips, so a sequential pass costs `tags × latency`.
+                            drainConcurrency: continuousOptions.DrainConcurrency),
                         ContinuousHistorizationRecorderActorName);
                     registry.Register<ContinuousHistorizationRecorderKey>(continuousRecorder);
                 }

tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderTests.cs

@@ -607,6 +607,189 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit
         }, TimeSpan.FromSeconds(10));
     }
 
+    // ---------------- Bounded-parallel drain (DrainConcurrency) ----------------
+
+    /// <summary>
+    ///     A drain pass writes distinct historian tags CONCURRENTLY, bounded by <c>DrainConcurrency</c>.
+    /// </summary>
+    /// <remarks>
+    ///     The tags in one peeked batch are independent gateway round-trips, so a sequential foreach makes
+    ///     a pass cost <c>tags × latency</c>. The load-bearing observation is a POSITIVE one — two writes
+    ///     genuinely in flight at the same instant — which is why the writer double blocks every call on a
+    ///     TCS and records the peak in-flight count. Measuring wall-clock time instead would be a flake
+    ///     generator and would not distinguish "parallel" from "fast".
+    ///     <para>
+    ///         The values arrive as ONE <see cref="VirtualTagActor.DependencyValueBatch"/> deliberately:
+    ///         the per-value handler nudges a DrainTick after each append, so three separate messages could
+    ///         start the drain on a batch of one and the pass would hold a single group — nothing to
+    ///         parallelise. A batch message appends all three before any DrainTick is dequeued.
+    ///     </para>
+    /// </remarks>
+    [Fact]
+    public async Task Drain_writes_distinct_tags_concurrently_within_the_window()
+    {
+        var mux = CreateTestProbe();
+        var writer = new GatingValueWriter();
+        var outbox = new InMemoryOutbox();
+
+        IActorRef rec = Sys.ActorOf(ContinuousHistorizationRecorder.Props(
+            mux.Ref, writer, outbox, historizedRefs: new[] { "A", "B", "C" },
+            drainInterval: TimeSpan.FromMilliseconds(50),
+            drainConcurrency: 8));
+
+        rec.Tell(new VirtualTagActor.DependencyValueBatch(new[]
+        {
+            new VirtualTagActor.DependencyValueChanged("A", 1.0, DateTime.UtcNow),
+            new VirtualTagActor.DependencyValueChanged("B", 2.0, DateTime.UtcNow),
+            new VirtualTagActor.DependencyValueChanged("C", 3.0, DateTime.UtcNow),
+        }));
+
+        await AwaitAssertAsync(
+            () =>
+            {
+                Assert.True(
+                    writer.MaxObservedConcurrency >= 2,
+                    $"expected overlapping writes; peak in-flight was {writer.MaxObservedConcurrency}");
+                return Task.CompletedTask;
+            },
+            TimeSpan.FromSeconds(10));
+
+        writer.Release();
+
+        // …and the pass still acks all-or-nothing: the whole peeked prefix leaves the outbox.
+        await AwaitAssertAsync(async () =>
+        {
+            var status = await rec.Ask<ContinuousHistorizationRecorder.RecorderStatus>(
+                ContinuousHistorizationRecorder.GetStatus.Instance, TimeSpan.FromSeconds(3));
+            Assert.Equal(0, status.QueuedDepth);
+            Assert.Equal(0, status.AbandonedEntries);
+        }, TimeSpan.FromSeconds(10));
+    }
+
+    /// <summary>
+    ///     Falsification control: <c>DrainConcurrency = 1</c> is exactly the old sequential drain.
+    /// </summary>
+    /// <remarks>
+    ///     Same fixture as the test above, so the only difference is the window size. Without this an
+    ///     implementation that ignored the option entirely (always unbounded) would pass the concurrency
+    ///     test and nothing would pin the bound.
+    /// </remarks>
+    [Fact]
+    public async Task Drain_concurrency_of_one_is_sequential()
+    {
+        var mux = CreateTestProbe();
+        var writer = new GatingValueWriter();
+        var outbox = new InMemoryOutbox();
+
+        IActorRef rec = Sys.ActorOf(ContinuousHistorizationRecorder.Props(
+            mux.Ref, writer, outbox, historizedRefs: new[] { "A", "B", "C" },
+            drainInterval: TimeSpan.FromMilliseconds(50),
+            drainConcurrency: 1));
+
+        rec.Tell(new VirtualTagActor.DependencyValueBatch(new[]
+        {
+            new VirtualTagActor.DependencyValueChanged("A", 1.0, DateTime.UtcNow),
+            new VirtualTagActor.DependencyValueChanged("B", 2.0, DateTime.UtcNow),
+            new VirtualTagActor.DependencyValueChanged("C", 3.0, DateTime.UtcNow),
+        }));
+
+        // The first write is in flight and blocked…
+        await AwaitAssertAsync(
+            () =>
+            {
+                Assert.True(writer.CallCount >= 1, "the drain must have started a write");
+                return Task.CompletedTask;
+            },
+            TimeSpan.FromSeconds(10));
+
+        // …and stays alone: with a window of 8 the other two groups would have entered by now.
+        await Task.Delay(TimeSpan.FromMilliseconds(300));
+        Assert.Equal(1, writer.MaxObservedConcurrency);
+        Assert.Equal(1, writer.CallCount);
+
+        writer.Release();
+
+        await AwaitAssertAsync(async () =>
+        {
+            var status = await rec.Ask<ContinuousHistorizationRecorder.RecorderStatus>(
+                ContinuousHistorizationRecorder.GetStatus.Instance, TimeSpan.FromSeconds(3));
+            Assert.Equal(0, status.QueuedDepth);
+        }, TimeSpan.FromSeconds(10));
+        Assert.Equal(1, writer.MaxObservedConcurrency);
+    }
+
+    /// <summary>
+    ///     <see cref="IHistorianValueWriter"/> double that BLOCKS every call until <see cref="Release"/>,
+    ///     recording the peak number of calls in flight simultaneously. That peak is what distinguishes a
+    ///     bounded-parallel drain from a sequential one; a writer that returns synchronously (like
+    ///     <see cref="FakeValueWriter"/>) can never observe overlap however parallel the caller is.
+    /// </summary>
+    private sealed class GatingValueWriter : IHistorianValueWriter
+    {
+        private readonly Lock _gate = new();
+        private readonly TaskCompletionSource _release = new(TaskCreationOptions.RunContinuationsAsynchronously);
+        private int _inFlight;
+        private int _maxObserved;
+        private int _calls;
+
+        /// <summary>Peak number of <see cref="WriteLiveValuesAsync"/> calls observed in flight at once.</summary>
+        public int MaxObservedConcurrency
+        {
+            get
+            {
+                lock (_gate)
+                {
+                    return _maxObserved;
+                }
+            }
+        }
+
+        /// <summary>Lifetime count of <see cref="WriteLiveValuesAsync"/> invocations.</summary>
+        public int CallCount
+        {
+            get
+            {
+                lock (_gate)
+                {
+                    return _calls;
+                }
+            }
+        }
+
+        /// <summary>Unblocks every pending and future call.</summary>
+        public void Release() => _release.TrySetResult();
+
+        public async Task<bool> WriteLiveValuesAsync(
+            string tag, IReadOnlyList<HistorizationValue> values, CancellationToken ct)
+        {
+            lock (_gate)
+            {
+                _calls++;
+                _inFlight++;
+                if (_inFlight > _maxObserved)
+                {
+                    _maxObserved = _inFlight;
+                }
+            }
+
+            try
+            {
+                // A bounded wait rather than an unbounded one: a regression that never releases fails the
+                // assertion rather than hanging the whole test run.
+                await _release.Task.WaitAsync(TimeSpan.FromSeconds(30), ct).ConfigureAwait(false);
+            }
+            finally
+            {
+                lock (_gate)
+                {
+                    _inFlight--;
+                }
+            }
+
+            return true;
+        }
+    }
+
     /// <summary>In-memory <see cref="IHistorianValueWriter"/> double: records every value written and
     /// returns <see cref="Succeed"/> as the ack. Thread-safe — the recorder drains off the actor thread.</summary>
     private sealed class FakeValueWriter : IHistorianValueWriter