fix(historian): incremental batch-append bookkeeping; genuine group-commit durability pin

2026-08-16 04:18:38 -04:00
parent fbe8aaad9d
commit 7320586512
3 changed files with 98 additions and 14 deletions
@@ -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.
    /// </summary>
+    /// <remarks>
+    ///     <para>
+    ///         <strong>Failure semantics — prefix, never orphan.</strong> 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
+    ///         <see cref="PeekBatchAsync"/>, ackable via <see cref="RemoveAsync"/>, counted by
+    ///         <see cref="CountAsync"/>, 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.
+    ///     </para>
+    ///     <para>
+    ///         <strong>Accepted residual.</strong> 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 <see cref="AppendAsync"/> has always had, widened
+    ///         from one entry to the batch.
+    ///     </para>
+    /// </remarks>
    /// <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>
@@ -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<LiveEntry> 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<LiveEntry> oldest = _live.First!;
@@ -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<Guid>();
+        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()
    {