perf(historian): group-commit batch append on the historization outbox

src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/IHistorizationOutbox.cs

@@ -19,6 +19,17 @@ public interface IHistorizationOutbox : IDisposable
     /// <returns>A task that represents the asynchronous operation.</returns>
     ValueTask AppendAsync(HistorizationOutboxEntry entry, CancellationToken ct);
 
+    /// <summary>
+    ///     Appends a burst of entries with ONE durability commit for the whole batch (group
+    ///     commit). Under PerEntry commit mode this is the per-batch fsync that replaces N
+    ///     per-entry fsyncs; under Periodic it is indistinguishable from N AppendAsync calls.
+    ///     An empty list is a no-op.
+    /// </summary>
+    /// <param name="entries">The value records to buffer, in append (FIFO) order.</param>
+    /// <param name="ct">Cancellation token.</param>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    ValueTask AppendBatchAsync(IReadOnlyList<HistorizationOutboxEntry> entries, CancellationToken ct);
+
     /// <summary>
     ///     Returns up to <paramref name="max"/> oldest un-acked entries in FIFO order without removing
     ///     them. Removal happens via <see cref="RemoveAsync"/> once each entry is durably written.

src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Recorder/FasterLogHistorizationOutbox.cs

@@ -99,23 +99,52 @@ public sealed class FasterLogHistorizationOutbox : IHistorizationOutbox
     {
         ArgumentNullException.ThrowIfNull(entry);
 
-        byte[] payload = HistorizationOutboxEntrySerializer.Serialize(entry);
+        // A one-entry batch is byte-for-byte the single-append path (one enqueue, one PerEntry
-        long startAddress = await _log.EnqueueAsync(payload, ct).ConfigureAwait(false);
+        // commit, one bookkeeping pass), so there is no second copy of the drop-oldest policy.
+        await AppendBatchAsync([entry], ct).ConfigureAwait(false);
+    }
+
+    /// <inheritdoc />
+    public async ValueTask AppendBatchAsync(IReadOnlyList<HistorizationOutboxEntry> entries, CancellationToken ct)
+    {
+        ArgumentNullException.ThrowIfNull(entries);
+        if (entries.Count == 0)
+            return;
+
+        // Validate before ANY enqueue: a null halfway through would otherwise leave the batch's
+        // earlier entries in the log with no in-memory index entry (invisible but not reclaimable).
+        for (int i = 0; i < entries.Count; i++)
+            ArgumentNullException.ThrowIfNull(entries[i]);
+
+        long[] startAddresses = new long[entries.Count];
+        for (int i = 0; i < entries.Count; i++)
+        {
+            byte[] payload = HistorizationOutboxEntrySerializer.Serialize(entries[i]);
+            startAddresses[i] = await _log.EnqueueAsync(payload, ct).ConfigureAwait(false);
+        }
+
         if (_commitMode == HistorizationCommitMode.PerEntry)
         {
-            // PerEntry: durable before returning. Periodic skips this — the background timer (and
+            // Group commit: ONE fsync makes the whole batch durable before returning, replacing the
-            // Dispose) commit on their cadence (accepted throughput/latency trade-off).
+            // N per-entry fsyncs a loop of AppendAsync would have cost. Periodic skips this — the
+            // background timer (and Dispose) commit on their cadence (accepted trade-off).
             await _log.CommitAsync(ct).ConfigureAwait(false);
         }
 
         long? truncateTo = null;
         lock (_state)
         {
-            LinkedListNode<LiveEntry> node = _live.AddLast(new LiveEntry(entry.Id, startAddress));
+            for (int i = 0; i < entries.Count; i++)
-            _index[entry.Id] = node;
+            {
+                HistorizationOutboxEntry entry = entries[i];
+                LinkedListNode<LiveEntry> node = _live.AddLast(new LiveEntry(entry.Id, startAddresses[i]));
+                _index[entry.Id] = node;
+            }
 
-            // Drop-oldest on overflow. The new head is the start address of whatever entry survives
+            // Drop-oldest on overflow, run ONCE for the whole batch (an over-capacity batch sheds
-            // at the front (or the tail if the log emptied); truncate to the furthest such address.
+            // every surplus entry here). The new head is the start address of whatever entry
+            // survives at the front (or the tail if the log emptied); truncate to the furthest such
+            // address, so the batch needs at most one truncate + one extra commit.
             while (_capacity > 0 && _live.Count > _capacity)
             {
                 LinkedListNode<LiveEntry> oldest = _live.First!;

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/Recorder/FasterLogHistorizationOutboxTests.cs

@@ -140,6 +140,62 @@ public sealed class FasterLogHistorizationOutboxTests : IDisposable
         Assert.Equal(new[] { a.Id, b.Id }, batch.Select(e => e.Id));
     }
 
+    [Fact]
+    public async Task AppendBatch_then_peek_returns_fifo_in_batch_order()
+    {
+        // Group commit: one fsync for the whole batch. The reopen leg is the load-bearing half — it
+        // proves the single per-batch commit actually made every entry durable, not just the last one.
+        var dir = NewTempDir();
+        var a = E("A", 1);
+        var b = E("B", 2);
+        var c = E("C", 3);
+        {
+            using var o = new FasterLogHistorizationOutbox(dir, HistorizationCommitMode.PerEntry);
+            await o.AppendBatchAsync([a, b, c], TestContext.Current.CancellationToken);
+
+            Assert.Equal(3, await o.CountAsync(TestContext.Current.CancellationToken));
+            var batch = await o.PeekBatchAsync(10, TestContext.Current.CancellationToken);
+            Assert.Equal(new[] { a.Id, b.Id, c.Id }, batch.Select(e => e.Id));
+        }
+
+        using var reopened = new FasterLogHistorizationOutbox(dir, HistorizationCommitMode.PerEntry);
+        Assert.Equal(3, await reopened.CountAsync(TestContext.Current.CancellationToken));
+        var survived = await reopened.PeekBatchAsync(10, TestContext.Current.CancellationToken);
+        Assert.Equal(new[] { a.Id, b.Id, c.Id }, survived.Select(e => e.Id));
+    }
+
+    [Fact]
+    public async Task AppendBatch_at_capacity_drops_oldest_across_the_whole_batch()
+    {
+        // The drop-oldest loop runs ONCE for the batch, so an over-capacity batch must shed every
+        // surplus entry (not just one) and count each drop.
+        var dir = NewTempDir();
+        using var o = new FasterLogHistorizationOutbox(dir, HistorizationCommitMode.PerEntry, capacity: 2);
+        var a = E("A", 1);
+        var b = E("B", 2);
+        var c = E("C", 3);
+        var d = E("D", 4);
+        await o.AppendBatchAsync([a, b, c, d], TestContext.Current.CancellationToken);
+
+        Assert.Equal(2, await o.CountAsync(TestContext.Current.CancellationToken));
+        Assert.Equal(2, o.DroppedCount);
+        var tags = (await o.PeekBatchAsync(10, TestContext.Current.CancellationToken)).Select(e => e.Tag).ToArray();
+        Assert.Equal(new[] { "C", "D" }, tags);
+    }
+
+    [Fact]
+    public async Task AppendBatch_empty_is_noop()
+    {
+        var dir = NewTempDir();
+        using var o = new FasterLogHistorizationOutbox(dir, HistorizationCommitMode.PerEntry);
+        await o.AppendAsync(E("A", 1), TestContext.Current.CancellationToken);
+
+        await o.AppendBatchAsync([], TestContext.Current.CancellationToken);
+
+        Assert.Equal(1, await o.CountAsync(TestContext.Current.CancellationToken));
+        Assert.Equal(0, o.DroppedCount);
+    }
+
     [Fact]
     public async Task Remove_unknown_id_is_noop()
     {

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

@@ -733,6 +733,37 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit
             return ValueTask.CompletedTask;
         }
 
+        public ValueTask AppendBatchAsync(IReadOnlyList<HistorizationOutboxEntry> entries, CancellationToken ct)
+        {
+            if (entries.Count == 0)
+            {
+                return ValueTask.CompletedTask;
+            }
+
+            if (_yieldOnAppend)
+            {
+                return AppendBatchYieldingAsync(entries);
+            }
+
+            foreach (HistorizationOutboxEntry entry in entries)
+            {
+                AppendCore(entry);
+            }
+
+            return ValueTask.CompletedTask;
+        }
+
+        private async ValueTask AppendBatchYieldingAsync(IReadOnlyList<HistorizationOutboxEntry> entries)
+        {
+            // Same pool-thread-completion requirement as AppendYieldingAsync below — a group commit
+            // fsyncs once for the batch, so the yield happens once, before the entries are recorded.
+            await Task.Delay(1).ConfigureAwait(false);
+            foreach (HistorizationOutboxEntry entry in entries)
+            {
+                AppendCore(entry);
+            }
+        }
+
         private async ValueTask AppendYieldingAsync(HistorizationOutboxEntry entry)
         {
             // Must complete on a POOL thread, carrying no synchronization context — that is what a real

tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ServiceCollectionExtensionsTests.cs

@@ -245,6 +245,9 @@ public sealed class ServiceCollectionExtensionsTests
         public long DroppedCount => 0;
         /// <summary>No-op append.</summary>
         public ValueTask AppendAsync(HistorizationOutboxEntry entry, CancellationToken ct) => ValueTask.CompletedTask;
+        /// <summary>No-op batch append.</summary>
+        public ValueTask AppendBatchAsync(IReadOnlyList<HistorizationOutboxEntry> entries, CancellationToken ct)
+            => ValueTask.CompletedTask;
         /// <summary>Always returns an empty batch.</summary>
         public ValueTask<IReadOnlyList<HistorizationOutboxEntry>> PeekBatchAsync(int max, CancellationToken ct)
             => ValueTask.FromResult<IReadOnlyList<HistorizationOutboxEntry>>(Array.Empty<HistorizationOutboxEntry>());