feat(historian-gateway): FasterLog historization outbox (PerEntry/Periodic, drop-oldest)

Claude-Session: https://claude.ai/code/session_012SDSQ3AcaXqPcBtDESBRii
2026-06-26 17:20:06 -04:00
parent 1a6eb7efe6
commit 555bd477f1
8 changed files with 536 additions and 0 deletions
@@ -0,0 +1,123 @@
+using System.Linq;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions.Historian;
+using ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Recorder;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests.Recorder;
+
+/// <summary>
+/// Durability + FIFO contract tests for the FasterLog-backed historization outbox. The
+/// remove-then-reopen (restart durability) and drop-oldest (capacity) cases are load-bearing —
+/// the outbox is the durable boundary the continuous-historization recorder acks against.
+/// </summary>
+public sealed class FasterLogHistorizationOutboxTests : IDisposable
+{
+    private readonly List<string> _dirs = new();
+
+    private string NewTempDir()
+    {
+        var dir = Path.Combine(Path.GetTempPath(), "histgw-outbox-" + Guid.NewGuid().ToString("N"));
+        Directory.CreateDirectory(dir);
+        _dirs.Add(dir);
+        return dir;
+    }
+
+    private static HistorizationOutboxEntry E(string tag, double v) =>
+        new(Guid.NewGuid(), tag, v, 192, new DateTime(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc));
+
+    [Fact]
+    public async Task Append_then_peek_returns_fifo()
+    {
+        var dir = NewTempDir();
+        using var o = new FasterLogHistorizationOutbox(dir, HistorizationCommitMode.PerEntry);
+        await o.AppendAsync(E("A", 1), TestContext.Current.CancellationToken);
+        await o.AppendAsync(E("B", 2), TestContext.Current.CancellationToken);
+        var batch = await o.PeekBatchAsync(10, TestContext.Current.CancellationToken);
+        Assert.Equal(new[] { "A", "B" }, batch.Select(b => b.Tag));
+        Assert.Equal(2, await o.CountAsync(TestContext.Current.CancellationToken));
+    }
+
+    [Fact]
+    public async Task Remove_truncates_and_survives_restart()
+    {
+        var dir = NewTempDir();
+        Guid keep;
+        {
+            using var o = new FasterLogHistorizationOutbox(dir, HistorizationCommitMode.PerEntry);
+            var a = E("A", 1);
+            var b = E("B", 2);
+            keep = b.Id;
+            await o.AppendAsync(a, TestContext.Current.CancellationToken);
+            await o.AppendAsync(b, TestContext.Current.CancellationToken);
+            await o.PeekBatchAsync(10, TestContext.Current.CancellationToken);
+            await o.RemoveAsync(a.Id, TestContext.Current.CancellationToken);   // ack A
+        }
+
+        using var reopened = new FasterLogHistorizationOutbox(dir, HistorizationCommitMode.PerEntry);
+        Assert.Equal(1, await reopened.CountAsync(TestContext.Current.CancellationToken));      // only B survives
+        var batch = await reopened.PeekBatchAsync(10, TestContext.Current.CancellationToken);
+        Assert.Equal(keep, batch[0].Id);
+    }
+
+    [Fact]
+    public async Task Capacity_full_drops_oldest_and_counts()
+    {
+        var dir = NewTempDir();
+        using var o = new FasterLogHistorizationOutbox(dir, HistorizationCommitMode.PerEntry, capacity: 2);
+        await o.AppendAsync(E("A", 1), TestContext.Current.CancellationToken);
+        await o.AppendAsync(E("B", 2), TestContext.Current.CancellationToken);
+        await o.AppendAsync(E("C", 3), TestContext.Current.CancellationToken);  // overflow -> drop oldest (A)
+        Assert.Equal(2, await o.CountAsync(TestContext.Current.CancellationToken));
+        Assert.Equal(1, o.DroppedCount);
+        var tags = (await o.PeekBatchAsync(10, TestContext.Current.CancellationToken)).Select(b => b.Tag).ToArray();
+        Assert.DoesNotContain("A", tags);
+    }
+
+    [Fact]
+    public async Task Periodic_mode_commits_and_recovers()
+    {
+        var dir = NewTempDir();
+        var a = E("A", 1);
+        var b = E("B", 2);
+        {
+            using var o = new FasterLogHistorizationOutbox(dir, HistorizationCommitMode.Periodic, commitIntervalMs: 20);
+            await o.AppendAsync(a, TestContext.Current.CancellationToken);
+            await o.AppendAsync(b, TestContext.Current.CancellationToken);
+            // Dispose flushes a final commit, making the periodic-mode appends durable.
+        }
+
+        using var reopened = new FasterLogHistorizationOutbox(dir, HistorizationCommitMode.Periodic, commitIntervalMs: 20);
+        Assert.Equal(2, await reopened.CountAsync(TestContext.Current.CancellationToken));
+        var batch = await reopened.PeekBatchAsync(10, TestContext.Current.CancellationToken);
+        Assert.Equal(new[] { a.Id, b.Id }, batch.Select(e => e.Id));
+    }
+
+    [Fact]
+    public async Task Remove_unknown_id_is_noop()
+    {
+        var dir = NewTempDir();
+        using var o = new FasterLogHistorizationOutbox(dir, HistorizationCommitMode.PerEntry);
+        await o.AppendAsync(E("A", 1), TestContext.Current.CancellationToken);
+        await o.RemoveAsync(Guid.NewGuid(), TestContext.Current.CancellationToken); // never appended -> no-op
+        Assert.Equal(1, await o.CountAsync(TestContext.Current.CancellationToken));
+    }
+
+    public void Dispose()
+    {
+        foreach (var dir in _dirs)
+        {
+            try
+            {
+                Directory.Delete(dir, recursive: true);
+            }
+            catch (IOException)
+            {
+                // Best-effort cleanup; a lingering OS handle must not fail the test run.
+            }
+            catch (UnauthorizedAccessException)
+            {
+                // Best-effort cleanup.
+            }
+        }
+    }
+}