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 b612d301..626bb5cd 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
@@ -25,6 +25,25 @@ public interface IHistorizationOutbox : IDisposable
/// per-entry fsyncs; under Periodic it is indistinguishable from N AppendAsync calls.
/// An empty list is a no-op.
///
+ ///
+ ///
+ /// Failure semantics — prefix, never orphan. The call is not atomic. If it
+ /// throws part-way through (cancellation, an I/O fault), the entries appended before the
+ /// failure remain in the buffer and are fully tracked: they are visible to
+ /// , ackable via , counted by
+ /// , and subject to the capacity bound — exactly as if the caller
+ /// had passed only that prefix. Nothing is left buffered-but-unackable, which would be
+ /// redelivered on every drain pass and block the buffer's head from advancing. Entries
+ /// after the failure point are not appended at all; the caller may retry them.
+ ///
+ ///
+ /// Accepted residual. The batch's single durability commit runs after the
+ /// last append, so a throw before it leaves that prefix tracked but not yet committed —
+ /// served correctly for the life of the process, but absent after an unclean restart. This
+ /// is the same enqueue-then-commit window has always had, widened
+ /// from one entry to the batch.
+ ///
+ ///
/// The value records to buffer, in append (FIFO) order.
/// Cancellation token.
/// A task that represents the asynchronous operation.
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 2d6c38fe..a1daa0c9 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,8 +99,8 @@ public sealed class FasterLogHistorizationOutbox : IHistorizationOutbox
{
ArgumentNullException.ThrowIfNull(entry);
- // 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.
+ // A one-entry batch is byte-for-byte the single-append path (one enqueue, one indexing step,
+ // one PerEntry commit), so there is no second copy of the drop-oldest policy.
await AppendBatchAsync([entry], ct).ConfigureAwait(false);
}
@@ -116,11 +116,24 @@ public sealed class FasterLogHistorizationOutbox : IHistorizationOutbox
for (int i = 0; i < entries.Count; i++)
ArgumentNullException.ThrowIfNull(entries[i]);
- long[] startAddresses = new long[entries.Count];
+ // INCREMENTAL bookkeeping: each entry is indexed under _state immediately after ITS OWN
+ // enqueue returns — never accumulated and applied after the loop. That is what makes a
+ // mid-loop failure (cancellation, I/O fault) safe: entries 0..i-1 are then both physically in
+ // the log AND in _live/_index, so PeekBatchAsync (which scans the raw log) and RemoveAsync
+ // (which looks the id up in _index) agree about them, and the head can advance past them.
+ // Deferring the indexing until after the loop stranded that prefix physically-present but
+ // untracked — peekable, un-removable (RemoveAsync no-ops on an unknown id), hence redelivered
+ // to the historian on every drain pass and blocking truncation until a restart's
+ // RecoverState() re-indexed it — and it also escaped the capacity bound.
for (int i = 0; i < entries.Count; i++)
{
- byte[] payload = HistorizationOutboxEntrySerializer.Serialize(entries[i]);
- startAddresses[i] = await _log.EnqueueAsync(payload, ct).ConfigureAwait(false);
+ HistorizationOutboxEntry entry = entries[i];
+ byte[] payload = HistorizationOutboxEntrySerializer.Serialize(entry);
+ long start = await _log.EnqueueAsync(payload, ct).ConfigureAwait(false);
+ lock (_state)
+ {
+ _index[entry.Id] = _live.AddLast(new LiveEntry(entry.Id, start));
+ }
}
if (_commitMode == HistorizationCommitMode.PerEntry)
@@ -128,23 +141,26 @@ public sealed class FasterLogHistorizationOutbox : IHistorizationOutbox
// 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).
+ //
+ // Residual (documented on IHistorizationOutbox.AppendBatchAsync): a throw before this
+ // commit leaves the enqueued prefix tracked but not yet committed, exactly the exposure
+ // the single-entry path has always had between its enqueue and its commit. Such entries
+ // are served from memory until a restart, where RecoverState() rebuilds from the
+ // committed log and simply does not see them.
await _log.CommitAsync(ct).ConfigureAwait(false);
}
long? truncateTo = null;
lock (_state)
{
- 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, 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.
+ //
+ // A batch that throws mid-loop skips this pass, so _live can sit transiently above
+ // _capacity — the same convergent overflow RecoverState() documents, shed by the next
+ // append's while-loop.
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 a911a71c..ec37e82e 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
@@ -1,4 +1,5 @@
using System.Linq;
+using FASTER.core;
using Xunit;
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
using ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Recorder;
@@ -143,8 +144,11 @@ public sealed class FasterLogHistorizationOutboxTests : IDisposable
[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.
+ // Group commit: one fsync for the whole batch, and FIFO order is preserved across a restart.
+ // The reopen leg here proves ORDER and COUNT survive recovery — it deliberately does NOT
+ // prove the batch committed itself, because Dispose() unconditionally runs
+ // _log.Commit(spinWait: true) and would rescue a batch that never committed. That claim is
+ // pinned separately by AppendBatch_group_commit_is_durable_before_dispose.
var dir = NewTempDir();
var a = E("A", 1);
var b = E("B", 2);
@@ -164,6 +168,51 @@ public sealed class FasterLogHistorizationOutboxTests : IDisposable
Assert.Equal(new[] { a.Id, b.Id, c.Id }, survived.Select(e => e.Id));
}
+ [Fact]
+ public async Task AppendBatch_group_commit_is_durable_before_dispose()
+ {
+ // THE group-commit pin. Every other durability test disposes the writer before reopening, and
+ // Dispose() unconditionally commits — so an AppendBatchAsync that never called CommitAsync
+ // would still pass them. This one never disposes the writer before observing durability:
+ // it opens a SECOND, read-only FasterLog over the same directory while the writing outbox is
+ // still alive, and reads only up to CommittedUntilAddress. Anything visible there was made
+ // durable by AppendBatchAsync's own group commit and nothing else. (PerEntry mode, so no
+ // periodic-commit loop exists to commit behind the batch's back either.)
+ var dir = NewTempDir();
+ var a = E("A", 1);
+ var b = E("B", 2);
+ var c = E("C", 3);
+
+ var writer = new FasterLogHistorizationOutbox(dir, HistorizationCommitMode.PerEntry);
+ try
+ {
+ await writer.AppendBatchAsync([a, b, c], TestContext.Current.CancellationToken);
+
+ Assert.Equal(new[] { a.Id, b.Id, c.Id }, ReadCommitted(dir));
+ }
+ finally
+ {
+ writer.Dispose();
+ }
+ }
+
+ // Opens an independent read-only FasterLog over the outbox's directory and returns the ids of
+ // every record the writer has COMMITTED (BeginAddress .. CommittedUntilAddress). Uncommitted
+ // enqueues sit past CommittedUntilAddress and are therefore invisible here — which is exactly
+ // what makes this a durability observation rather than a memory one.
+ private static Guid[] ReadCommitted(string dir)
+ {
+ using var device = new ManagedLocalStorageDevice(Path.Combine(dir, "hlog.log"));
+ using var reader = new FasterLog(new FasterLogSettings { LogDevice = device, ReadOnlyMode = true });
+
+ var ids = new List();
+ using FasterLogScanIterator iter = reader.Scan(reader.BeginAddress, reader.CommittedUntilAddress, recover: false);
+ while (iter.GetNext(out byte[] bytes, out int len, out _, out _))
+ ids.Add(HistorizationOutboxEntrySerializer.Deserialize(bytes.AsSpan(0, len)).Id);
+
+ return ids.ToArray();
+ }
+
[Fact]
public async Task AppendBatch_at_capacity_drops_oldest_across_the_whole_batch()
{