diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/IHistorizationOutbox.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/IHistorizationOutbox.cs
index a9a4f1a0..b612d301 100644
--- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/IHistorizationOutbox.cs
+++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Abstractions/Historian/IHistorizationOutbox.cs
@@ -19,6 +19,17 @@ public interface IHistorizationOutbox : IDisposable
/// A task that represents the asynchronous operation.
ValueTask AppendAsync(HistorizationOutboxEntry entry, CancellationToken ct);
+ ///
+ /// 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.
+ ///
+ /// The value records to buffer, in append (FIFO) order.
+ /// Cancellation token.
+ /// A task that represents the asynchronous operation.
+ ValueTask AppendBatchAsync(IReadOnlyList entries, CancellationToken ct);
+
///
/// Returns up to oldest un-acked entries in FIFO order without removing
/// them. Removal happens via once each entry is durably written.
diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Recorder/FasterLogHistorizationOutbox.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Recorder/FasterLogHistorizationOutbox.cs
index 21398c91..2d6c38fe 100644
--- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway/Recorder/FasterLogHistorizationOutbox.cs
+++ b/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);
- long startAddress = await _log.EnqueueAsync(payload, ct).ConfigureAwait(false);
+ // A one-entry batch is byte-for-byte the single-append path (one enqueue, one PerEntry
+ // commit, one bookkeeping pass), so there is no second copy of the drop-oldest policy.
+ await AppendBatchAsync([entry], ct).ConfigureAwait(false);
+ }
+
+ ///
+ public async ValueTask AppendBatchAsync(IReadOnlyList 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
- // Dispose) commit on their cadence (accepted throughput/latency trade-off).
+ // Group commit: ONE fsync makes the whole batch durable before returning, replacing the
+ // 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 node = _live.AddLast(new LiveEntry(entry.Id, startAddress));
- _index[entry.Id] = node;
+ for (int i = 0; i < entries.Count; i++)
+ {
+ HistorizationOutboxEntry entry = entries[i];
+ LinkedListNode 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
- // at the front (or the tail if the log emptied); truncate to the furthest such address.
+ // Drop-oldest on overflow, run ONCE for the whole batch (an over-capacity batch sheds
+ // 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 oldest = _live.First!;
diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/Recorder/FasterLogHistorizationOutboxTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/Recorder/FasterLogHistorizationOutboxTests.cs
index 33b0384c..a911a71c 100644
--- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/Recorder/FasterLogHistorizationOutboxTests.cs
+++ b/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()
{
diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderTests.cs
index e9cd3145..78dc4e51 100644
--- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Historian/ContinuousHistorizationRecorderTests.cs
+++ b/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 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 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
diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ServiceCollectionExtensionsTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ServiceCollectionExtensionsTests.cs
index 0b8965c7..37111fd0 100644
--- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ServiceCollectionExtensionsTests.cs
+++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/ServiceCollectionExtensionsTests.cs
@@ -245,6 +245,9 @@ public sealed class ServiceCollectionExtensionsTests
public long DroppedCount => 0;
/// No-op append.
public ValueTask AppendAsync(HistorizationOutboxEntry entry, CancellationToken ct) => ValueTask.CompletedTask;
+ /// No-op batch append.
+ public ValueTask AppendBatchAsync(IReadOnlyList entries, CancellationToken ct)
+ => ValueTask.CompletedTask;
/// Always returns an empty batch.
public ValueTask> PeekBatchAsync(int max, CancellationToken ct)
=> ValueTask.FromResult>(Array.Empty());