test(historian): pin the failed-batch-append branch; correct its loss-accounting docs

The batched ingest's append-failure catch had zero coverage, and two doc statements overstated the loss: the Warning said the burst's values were "dropped", and RecorderStatus.OutboxAppendFailures said "the value was dropped, not recorded". The outbox contract is prefix-never-orphan, so a partial-prefix failure leaves entries durably queued that drain normally. - New test pins the branch: one 3-item burst whose group commit commits a 1-entry prefix then throws => OutboxAppendFailures == 1 (per burst, not per value), TotalRecorded == 0, actor alive, and the prefix still reaches the writer via the post-failure DrainTick (drainInterval 30s, so the steady cadence cannot account for it). Falsified twice: counting per item and dropping the drain nudge each turn it red. - Second test pins the OperationCanceledException filter from its observable side - a cancel that is NOT shutdown is metered, not swallowed. Removing the 'when (_lifetime.IsCancellationRequested)' turns it red. The shutdown side has no cheap seam (the counter dies with the actor). - Docs corrected: OutboxAppendFailures counts FAILED BATCH APPENDS, one per burst whatever its size (a granularity change from the per-value era), and is not a count of values lost; TotalRecorded under-counts when a commit fails after a partial prefix. Log message says "failed to append" and names the prefix. No functional change beyond the message string. Also folds in the DrainBatchSize default alignment behind 0298a0b9: the recorder's two ctor defaults and its non-positive fallback still said 64 while ContinuousHistorizationOptions.DrainBatchSize now defaults to 1024. No test pinned the old 64.
2026-08-16 04:50:30 -04:00
parent 0298a0b944
commit 451273c860
2 changed files with 222 additions and 13 deletions
@@ -92,11 +92,17 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers

    /// <summary>A point-in-time snapshot of the recorder's counters.</summary>
    /// <param name="QueuedDepth">Un-acked entries currently held in the durable outbox.</param>
-    /// <param name="TotalRecorded">Lifetime count of values appended to the outbox.</param>
+    /// <param name="TotalRecorded">Lifetime count of values durably appended to the outbox. Credited per
+    /// BURST, on a successful group commit — so it UNDER-counts when a batch append fails after the outbox
+    /// durably committed a partial prefix: the outbox does not report how far it got, so the recorder cannot
+    /// credit that prefix. Those entries still drain; they are simply never counted here.</param>
    /// <param name="DroppedNonNumeric">Lifetime count of values dropped for not being numeric-coercible.</param>
    /// <param name="OutboxDropped">Lifetime count of entries the outbox dropped on capacity overflow.</param>
-    /// <param name="OutboxAppendFailures">Lifetime count of durable-boundary append failures (the value
-    /// was dropped, not recorded, and the actor stayed alive rather than restart-looping).</param>
+    /// <param name="OutboxAppendFailures">Lifetime count of FAILED BATCH APPENDS at the durable boundary —
+    /// ONE per failed burst whatever its size, never one per value (the granularity changed with the batched
+    /// ingest; the per-value era counted one per value). The actor stays alive rather than restart-looping.
+    /// It is NOT a count of values lost: the outbox contract is "prefix, never orphan", so a failed burst
+    /// may still have left entries durably queued that drain normally — see <see cref="IngestAsync"/>.</param>
    /// <param name="LastDrainSucceeded">Whether the most recent drain pass acked cleanly.</param>
    /// <param name="SuppressedNotPrimary">Lifetime count of values not captured because this node is
    /// the Secondary of a pair whose Primary records the same refs (Gitea #537).</param>
@@ -200,7 +206,7 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
        IHistorianValueWriter writer,
        IHistorizationOutbox outbox,
        IReadOnlyList<string> historizedRefs,
-        int drainBatchSize = 64,
+        int drainBatchSize = 1024,
        TimeSpan? drainInterval = null,
        TimeSpan? minBackoff = null,
        TimeSpan? maxBackoff = null,
@@ -232,7 +238,7 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
        IHistorianValueWriter writer,
        IHistorizationOutbox outbox,
        IReadOnlyList<string> historizedRefs,
-        int drainBatchSize = 64,
+        int drainBatchSize = 1024,
        TimeSpan? drainInterval = null,
        TimeSpan? minBackoff = null,
        TimeSpan? maxBackoff = null,
@@ -252,7 +258,10 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
            AddRef(r, r);
        }

-        _drainBatchSize = drainBatchSize > 0 ? drainBatchSize : 64;
+        // Both this fallback and the ctor/Props defaults above must track ContinuousHistorizationOptions
+        // .DrainBatchSize (1024): production always passes the option through, so a divergent literal here
+        // only ever shows up as a quietly different batch size in tests and in degenerate-input callers.
+        _drainBatchSize = drainBatchSize > 0 ? drainBatchSize : 1024;
        _drainInterval = drainInterval is { } di && di > TimeSpan.Zero ? di : TimeSpan.FromSeconds(2);
        _minBackoff = minBackoff is { } mb && mb > TimeSpan.Zero ? mb : TimeSpan.FromSeconds(1);
        _maxBackoff = maxBackoff is { } xb && xb > TimeSpan.Zero ? xb : TimeSpan.FromSeconds(30);
@@ -339,11 +348,14 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
    ///         #491 note below).
    ///     </para>
    ///     <para>
-    ///         <b>Append failure is batch-wide.</b> A failed group commit is metered once
-    ///         (<c>OutboxAppendFailures</c>), logged by category, and the burst's values are dropped — the
-    ///         same never-restart-loop policy the per-value path had, applied to the batch. Note the outbox
-    ///         contract makes any partially-appended prefix fully tracked and ackable, so the trailing drain
-    ///         nudge still flushes it; nothing is left buffered-but-unackable.
+    ///         <b>Append failure is batch-wide, and it is NOT a whole-burst loss.</b> A failed group commit
+    ///         is metered ONCE (<c>OutboxAppendFailures</c> — one per failed burst whatever its size, never
+    ///         one per value) and logged by category, keeping the same never-restart-loop policy the
+    ///         per-value path had. What is actually lost is only what the outbox never committed: its
+    ///         contract is "prefix, never orphan", so anything it durably appended before the failure stays
+    ///         fully tracked and ackable and the trailing drain nudge flushes it normally — nothing is left
+    ///         buffered-but-unackable. That prefix does NOT credit <c>TotalRecorded</c>, because the outbox
+    ///         does not report how far it got; the counter under-counts rather than over-claims.
    ///     </para>
    /// </remarks>
    /// <param name="items">The burst's matched dependency values, in fan-out order.</param>
@@ -433,9 +445,20 @@ public sealed class ContinuousHistorizationRecorder : ReceiveActor, IWithTimers
            // out of the handler — that would trip Akka supervision into a restart, and under a persistent
            // disk fault the actor would restart-loop (re-register → next burst → append fails → restart →
            // …). Mirror the drain path's catch-all: meter the failure (category only, no value content)
-            // and drop this burst's values.
+            // and carry on.
+            //
+            // The meter is ONE per failed burst, whatever the burst's size — a deliberate granularity change
+            // from the per-value era, where it was one per value. And it is NOT a count of values lost: the
+            // outbox commits a PREFIX or nothing (never an orphan), so an unknown number of this burst's
+            // entries may already be durable. Hence the message says "failed to append", not "dropped" —
+            // claiming total loss here would be wrong about the prefix the drain nudge below is about to
+            // flush. TotalRecorded is likewise not credited for that prefix: the outbox does not report how
+            // far it got, so the counter under-counts rather than over-claims.
            _outboxAppendFailures++;
-            _log.Warning("ContinuousHistorization: outbox batch append failed ({Exception}); {Count} value(s) dropped.",
+            _log.Warning(
+                "ContinuousHistorization: outbox batch append failed ({Exception}); {Count} value(s) in the "
+                + "burst failed to append. Any prefix the outbox committed before the failure stays queued "
+                + "and drains normally; the remainder are lost.",
                ex.GetType().Name, entries.Count);
        }

@@ -937,6 +937,140 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit
        mux.ExpectNoMsg(TimeSpan.FromMilliseconds(300));   // no PreStart re-register == no restart
    }

+    /// <summary>
+    ///     A failed group commit is metered ONCE for the whole burst, credits nothing to
+    ///     <c>TotalRecorded</c>, leaves the actor alive, and STILL nudges the drain — which flushes whatever
+    ///     prefix the outbox durably committed before it failed.
+    /// </summary>
+    /// <remarks>
+    ///     Pins the batched ingest's append-failure catch branch, which had no coverage at all: every other
+    ///     test here uses an outbox that always succeeds, so a catch that miscounted, swallowed the drain
+    ///     nudge, or rethrew into supervision would go unnoticed.
+    ///     <para>
+    ///         Three separate claims, each of which is a distinct way the branch can be wrong:
+    ///     </para>
+    ///     <para>
+    ///         (1) <c>OutboxAppendFailures == 1</c> for a THREE-item burst. The counter is per failed batch,
+    ///         not per value — a deliberate granularity change from the per-value era — so a catch that
+    ///         credited <c>entries.Count</c> would report 3 and make an operator read one bad fsync as three
+    ///         separate faults.
+    ///     </para>
+    ///     <para>
+    ///         (2) <c>TotalRecorded == 0</c> even though entry "A" IS durable. The outbox does not report how
+    ///         far it got, so the recorder deliberately under-counts rather than over-claims; this asserts
+    ///         the documented direction of the inaccuracy.
+    ///     </para>
+    ///     <para>
+    ///         (3) The durable prefix reaches the writer. This is the load-bearing half — it is the positive
+    ///         observation that the post-failure <c>DrainTick</c> was sent, and it is why the log message
+    ///         must not claim the burst was "dropped". <c>drainInterval</c> is set to 30 s so the steady
+    ///         cadence cannot fire inside the test: the only thing that can produce a drain here is the
+    ///         nudge. Deleting <c>self.Tell(DrainTick.Instance)</c> from the failure path (or returning from
+    ///         the catch) leaves "A" queued forever and turns this red.
+    ///     </para>
+    /// </remarks>
+    [Fact]
+    public async Task A_failed_batch_append_counts_one_failure_and_still_ticks_the_drain()
+    {
+        var mux = CreateTestProbe();
+        var writer = new FakeValueWriter { Succeed = true };
+        var outbox = new InMemoryOutbox
+        {
+            // The partial-prefix shape: "A" commits durably, then the commit fails — so "B" and "C" are
+            // genuinely lost while "A" is merely un-drained.
+            AppendBatchFault = new IOException("simulated fsync failure (disk full)"),
+            AppendBatchFaultAfter = 1,
+        };
+
+        IActorRef rec = Sys.ActorOf(ContinuousHistorizationRecorder.Props(
+            mux.Ref, writer, outbox, historizedRefs: new[] { "A", "B", "C" },
+            drainInterval: TimeSpan.FromSeconds(30)));
+        mux.ExpectMsg<DependencyMuxActor.RegisterInterest>();
+
+        var stamp = new DateTime(2026, 8, 15, 13, 0, 0, DateTimeKind.Utc);
+        rec.Tell(new VirtualTagActor.DependencyValueBatch(new[]
+        {
+            new VirtualTagActor.DependencyValueChanged("A", 1.0, stamp),
+            new VirtualTagActor.DependencyValueChanged("B", 2.0, stamp),
+            new VirtualTagActor.DependencyValueChanged("C", 3.0, stamp),
+        }));
+
+        // Answering GetStatus at all is itself the "actor is still alive" observation — an append failure
+        // that reached supervision would have restarted (or stopped) it.
+        await AwaitAssertAsync(
+            async () =>
+            {
+                var status = await rec.Ask<ContinuousHistorizationRecorder.RecorderStatus>(
+                    ContinuousHistorizationRecorder.GetStatus.Instance, TimeSpan.FromSeconds(3));
+                Assert.Equal(1, status.OutboxAppendFailures);
+                Assert.Equal(0, status.TotalRecorded);
+            },
+            TimeSpan.FromSeconds(10));
+
+        await AwaitAssertAsync(
+            () =>
+            {
+                Assert.True(outbox.PeekBatchCallCount >= 1, "the failure must still have nudged a drain");
+                Assert.Contains(writer.Snapshot(), w => w.Tag == "A" && w.Value == 1.0);
+                return Task.CompletedTask;
+            },
+            TimeSpan.FromSeconds(10));
+
+        // What never committed stays gone — the prefix drained, the remainder did not resurrect.
+        Assert.DoesNotContain(writer.Snapshot(), w => w.Tag == "B");
+        Assert.DoesNotContain(writer.Snapshot(), w => w.Tag == "C");
+        mux.ExpectNoMsg(TimeSpan.FromMilliseconds(300));   // no PreStart re-register == no restart
+    }
+
+    /// <summary>
+    ///     An <see cref="OperationCanceledException"/> that is NOT shutdown is metered as an append failure,
+    ///     not swallowed.
+    /// </summary>
+    /// <remarks>
+    ///     The catch that quietly discards a cancelled append is filtered on
+    ///     <c>when (_lifetime.IsCancellationRequested)</c> — i.e. only during the actor's own shutdown.
+    ///     Drop the filter and every cancelled append becomes invisible: no metric, and no drain nudge for
+    ///     the prefix. This drives the observable side of that filter (the lifetime is open here, so the
+    ///     general catch must take it).
+    ///     <para>
+    ///         The shutdown side itself is deliberately not tested: <c>_lifetime</c> is cancelled in
+    ///         <c>PostStop</c>, so reaching that branch means racing an append against a stopping actor —
+    ///         and once it has stopped there is no <c>GetStatus</c> to read the counter from. There is no
+    ///         cheap seam for it; the filter's correctness is pinned from this side instead.
+    ///     </para>
+    /// </remarks>
+    [Fact]
+    public async Task A_cancellation_that_is_not_shutdown_is_metered_as_an_append_failure()
+    {
+        var mux = CreateTestProbe();
+        var writer = new FakeValueWriter { Succeed = true };
+        var outbox = new InMemoryOutbox
+        {
+            AppendBatchFault = new OperationCanceledException("not a shutdown"),
+            AppendBatchFaultAfter = 0,
+        };
+
+        IActorRef rec = Sys.ActorOf(ContinuousHistorizationRecorder.Props(
+            mux.Ref, writer, outbox, historizedRefs: new[] { "A" },
+            drainInterval: TimeSpan.FromSeconds(30)));
+        mux.ExpectMsg<DependencyMuxActor.RegisterInterest>();
+
+        rec.Tell(new VirtualTagActor.DependencyValueChanged(
+            "A", 1.0, new DateTime(2026, 8, 15, 14, 0, 0, DateTimeKind.Utc)));
+
+        await AwaitAssertAsync(
+            async () =>
+            {
+                var status = await rec.Ask<ContinuousHistorizationRecorder.RecorderStatus>(
+                    ContinuousHistorizationRecorder.GetStatus.Instance, TimeSpan.FromSeconds(3));
+                Assert.Equal(1, status.OutboxAppendFailures);
+                Assert.Equal(0, status.TotalRecorded);
+            },
+            TimeSpan.FromSeconds(10));
+
+        mux.ExpectNoMsg(TimeSpan.FromMilliseconds(300));   // no PreStart re-register == no restart
+    }
+
    /// <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
@@ -1097,6 +1231,34 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit
        private int _appendBatchCalls;

        private readonly bool _yieldOnAppend;
+        private int _peekCalls;
+
+        /// <summary>When non-null, <see cref="AppendBatchAsync"/> durably appends the first
+        /// <see cref="AppendBatchFaultAfter"/> entries and then throws this — the production outbox's
+        /// "prefix, never orphan" contract on its failure path (an fsync hitting disk-full mid-commit).
+        /// The throw happens AFTER a pool-thread yield, as a real group-commit fsync does, so the
+        /// recorder's <c>ConfigureAwait(false)</c> continuation runs its catch block off the actor
+        /// thread — the Gitea #491 shape, on the failure path.</summary>
+        public Exception? AppendBatchFault { get; init; }
+
+        /// <summary>How many of the batch's entries are durably committed before
+        /// <see cref="AppendBatchFault"/> is thrown. 0 = nothing committed; a positive value models the
+        /// PARTIAL-PREFIX failure, where some entries are durable and will drain even though the append
+        /// as a whole reported failure.</summary>
+        public int AppendBatchFaultAfter { get; init; }
+
+        /// <summary>Lifetime count of <see cref="PeekBatchAsync"/> calls — one per drain pass, so a
+        /// non-zero value is positive evidence that a drain actually RAN.</summary>
+        public int PeekBatchCallCount
+        {
+            get
+            {
+                lock (_gate)
+                {
+                    return _peekCalls;
+                }
+            }
+        }

        /// <param name="capacity">Drop-oldest capacity; 0 means unbounded.</param>
        /// <param name="yieldOnAppend">
@@ -1175,6 +1337,11 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit
                _appendBatchCalls++;
            }

+            if (AppendBatchFault is not null)
+            {
+                return AppendBatchFaultingAsync(entries);
+            }
+
            if (entries.Count == 0)
            {
                return ValueTask.CompletedTask;
@@ -1193,6 +1360,24 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit
            return ValueTask.CompletedTask;
        }

+        private async ValueTask AppendBatchFaultingAsync(IReadOnlyList<HistorizationOutboxEntry> entries)
+        {
+            // Complete on a POOL thread before faulting (the same requirement AppendYieldingAsync documents):
+            // a real group commit fsyncs, so the recorder's ConfigureAwait(false) continuation — here, its
+            // CATCH block — resumes with no ActorContext installed. A synchronous throw would run the catch
+            // inline on the actor thread and would not exercise that.
+            await Task.Delay(1).ConfigureAwait(false);
+
+            // "Prefix, never orphan": whatever was committed before the failure is durably tracked and
+            // ackable, which is exactly what makes the post-failure drain nudge meaningful.
+            for (var i = 0; i < AppendBatchFaultAfter && i < entries.Count; i++)
+            {
+                AppendCore(entries[i]);
+            }
+
+            throw AppendBatchFault!;
+        }
+
        private async ValueTask AppendBatchYieldingAsync(IReadOnlyList<HistorizationOutboxEntry> entries)
        {
            // Same pool-thread-completion requirement as AppendYieldingAsync below — a group commit
@@ -1235,6 +1420,7 @@ public sealed class ContinuousHistorizationRecorderTests : TestKit
        {
            lock (_gate)
            {
+                _peekCalls++;
                IReadOnlyList<HistorizationOutboxEntry> batch = _entries.Take(max).ToArray();
                return ValueTask.FromResult(batch);
            }