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 d3b1bdbcd..bc97933b5 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 @@ -206,47 +206,73 @@ public sealed class FasterLogHistorizationOutbox : IHistorizationOutbox // here (truncate + commit), because nothing else ever would: a FIFO ack only passes records // older than the acked entry, so a corrupt record with no readable entry behind it would // otherwise pin the head (and the outbox count) forever. + // + // The scan runs OUTSIDE _state: once a backlog's head has aged out of FasterLog's in-memory window (any multi-minute + // historian outage) GetNext reads pages from the device synchronously, and holding the lock across that blocked the + // recorder's appends (index bookkeeping takes _state) on an Akka dispatcher thread, once per drain pass, exactly + // during outage recovery. The lock is taken only to snapshot the head and, when the scan met undecodable records, to + // reconcile them with the index (O(records scanned), never O(backlog)). public async ValueTask> PeekBatchAsync(int max, CancellationToken ct) { ArgumentOutOfRangeException.ThrowIfNegativeOrZero(max); - var batch = new List(Math.Min(max, 64)); - long? droppedHeadTo = null; + long from, tail; lock (_state) { - // _live and the scan are both in log order, so a forward cursor pairs each scanned record - // with its index node (by start address) in O(n) overall. - LinkedListNode? cursor = _live.First; - using FasterLogScanIterator iter = _log.Scan(_nextScanAddress, _log.TailAddress, recover: false); + from = _nextScanAddress; + tail = _log.TailAddress; + } + + var batch = new List(Math.Min(max, 64)); + List<(long Address, long Next, bool LeadsBatch)>? unreadable = null; + using (FasterLogScanIterator iter = _log.Scan(from, tail, recover: false)) + { while (batch.Count < max && iter.GetNext(out byte[] bytes, out int len, out long address, out long next)) { - while (cursor is not null && cursor.Value.Start < address) - cursor = cursor.Next; - LinkedListNode? node = cursor is not null && cursor.Value.Start == address ? cursor : null; - + ScanStepForTests?.Invoke(); if (HistorizationOutboxEntrySerializer.TryDeserialize(bytes.AsSpan(0, len), out HistorizationOutboxEntry? entry)) { batch.Add(entry); continue; } + (unreadable ??= []).Add((address, next, batch.Count == 0)); + } + } - // Count once: a node flagged at recovery (or by an earlier peek) is not re-counted. - // An unindexed record cannot be flagged, so it would count per peek — but only this - // process's own (readable) appends are ever unindexed, transiently. - if (node is null || !node.Value.Unreadable) + long? droppedHeadTo = null; + if (unreadable is not null) + { + lock (_state) + { + // _live and the scan are both in log order, so a forward cursor pairs each unreadable record with its + // index node (by start address). + LinkedListNode? cursor = _live.First; + foreach ((long address, long next, bool leadsBatch) in unreadable) { - _unreadableCount++; - if (node is not null) - node.Value = node.Value with { Unreadable = true }; - } + while (cursor is not null && cursor.Value.Start < address) + cursor = cursor.Next; + LinkedListNode? node = cursor is not null && cursor.Value.Start == address ? cursor : null; - if (batch.Count == 0 && node is not null && ReferenceEquals(node, _live.First)) - { - cursor = node.Next; - _index.Remove(node.Value.Id); - _live.RemoveFirst(); - _nextScanAddress = next; - droppedHeadTo = next; + // Count once: a node flagged at recovery (or by an earlier peek) is not re-counted. An unindexed record + // cannot be flagged, so it would count per peek — but only this process's own (readable) appends are + // ever unindexed, transiently. + if (node is null || !node.Value.Unreadable) + { + _unreadableCount++; + if (node is not null) + node.Value = node.Value with { Unreadable = true }; + } + + // Drop an unreadable record still at the HEAD, ahead of every readable entry (nothing else ever would). + // Re-checked under the lock: an ack or a drop-oldest may have moved the head while we scanned. + if (leadsBatch && node is not null && ReferenceEquals(node, _live.First)) + { + cursor = node.Next; + _index.Remove(node.Value.Id); + _live.RemoveFirst(); + _nextScanAddress = next; + droppedHeadTo = next; + } } } } @@ -260,6 +286,9 @@ public sealed class FasterLogHistorizationOutbox : IHistorizationOutbox return batch; } + /// Test seam: invoked once per record the peek scans, OUTSIDE the state lock. + internal Action? ScanStepForTests { get; set; } + /// public async ValueTask RemoveAsync(Guid id, CancellationToken ct) { 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 03c5a9c5f..e6826aa2a 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 @@ -27,6 +27,33 @@ public sealed class FasterLogHistorizationOutboxTests : IDisposable private static HistorizationOutboxEntry E(string tag, double v) => new(Guid.NewGuid(), tag, HistorianValue.FromDouble(v), 0u, new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc)); + [Fact] + public async Task A_peek_scan_does_not_hold_the_state_lock_so_appends_proceed_meanwhile() + { + // The scan may read pages from disk (a backlog aged out of the in-memory window); holding the state lock across it + // blocked the recorder's appends on an Akka dispatcher thread. + var dir = NewTempDir(); + using var o = new FasterLogHistorizationOutbox(dir, HistorizationCommitMode.PerEntry); + for (var i = 0; i < 5; i++) await o.AppendAsync(E("A", i), TestContext.Current.CancellationToken); + + bool? appendCompletedDuringScan = null; + o.ScanStepForTests = () => + { + if (appendCompletedDuringScan is not null) return; + var append = Task.Run(async () => await o.AppendAsync(E("B", 99), CancellationToken.None)); + appendCompletedDuringScan = append.Wait(TimeSpan.FromSeconds(5)); + }; + + var batch = await o.PeekBatchAsync(10, TestContext.Current.CancellationToken); + + Assert.True(appendCompletedDuringScan, "an append must not wait for a peek's scan"); + Assert.Equal(5, batch.Count(e => e.Tag == "A")); + Assert.Equal(6, await o.CountAsync(TestContext.Current.CancellationToken)); + // The concurrent append is FIFO-tracked: the next peek returns it after the five. + o.ScanStepForTests = null; + Assert.Equal("B", (await o.PeekBatchAsync(10, TestContext.Current.CancellationToken))[^1].Tag); + } + [Fact] public async Task Append_then_peek_returns_fifo() {