fix(auditlog): reconciliation pull serves only unacked rows (PushAbandoned/Rejected push-exit states), central pages within a tick with set-based inserts; refused push rows leave the queue after N attempts (review 2026-10-21 06/PERF-1, 06/COR-3, 06/IMPL-1)

2026-10-22 06:38:02 -04:00
parent 1da5d0bac4
commit 9e7e19c3c1
13 changed files with 873 additions and 116 deletions
@@ -51,9 +51,11 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture<MsSqlMig

    private static SiteAuditReconciliationOptions FastTickOptions(
        int batchSize = 256,
-        int stalledAfter = 2) =>
+        int stalledAfter = 2,
+        int maxPagesPerTick = 50) =>
        new()
        {
+            MaxPagesPerTick = maxPagesPerTick,
            // 100 ms tick keeps each test under a second. AwaitAssert covers
            // schedule jitter so a 100 ms tick has up to ~3 s to fire.
            ReconciliationIntervalSeconds = 300,
@@ -79,6 +81,25 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture<MsSqlMig
        private readonly HashSet<Guid> _seen = new();
        /// <summary>Gets the number of times <c>InsertIfNotExistsAsync</c> was called.</summary>
        public int InsertCallCount { get; private set; }
+        /// <summary>Gets the number of set-based <c>InsertManyIfNotExistsAsync</c> calls.</summary>
+        public int BatchCallCount { get; private set; }
+
+        /// <inheritdoc />
+        public Task<int> InsertManyIfNotExistsAsync(
+            IReadOnlyList<AuditEvent> events, TimeSpan? commandTimeout = null, CancellationToken ct = default)
+        {
+            BatchCallCount++;
+            var inserted = 0;
+            foreach (var evt in events)
+            {
+                if (_seen.Add(evt.EventId))
+                {
+                    Inserted.Add(evt);
+                    inserted++;
+                }
+            }
+            return Task.FromResult(inserted);
+        }

        /// <inheritdoc />
        public Task InsertIfNotExistsAsync(AuditEvent evt, CancellationToken ct = default)
@@ -480,9 +501,12 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture<MsSqlMig
    // 3. Tick_IngestEvents_ViaInsertIfNotExistsAsync
    // ---------------------------------------------------------------------

-    /// <summary>Verifies that pulled events are ingested through the repository insert-if-not-exists call.</summary>
+    /// <summary>
+    /// Review 2026-10-21 06/PERF-1: a pulled page is ingested with ONE set-based
+    /// <c>InsertManyIfNotExistsAsync</c> call, not one round trip per row.
+    /// </summary>
    [Fact]
-    public void Tick_IngestEvents_ViaInsertIfNotExistsAsync()
+    public void Tick_IngestEvents_ViaOneSetBasedInsert()
    {
        var sites = new StaticEnumerator(new SiteEntry("siteA", "http://siteA:8083"));
        var e1 = NewEvent("siteA");
@@ -493,16 +517,50 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture<MsSqlMig

        var actor = CreateActor(sites, client, repo, FastTickOptions());

-        AwaitAssert(() => Assert.Equal(2, repo.InsertCallCount),
+        AwaitAssert(() => Assert.Equal(2, repo.Inserted.Count),
            duration: TimeSpan.FromSeconds(3),
            interval: TimeSpan.FromMilliseconds(50));

        StopReconciliationActor(actor);

+        Assert.Equal(1, repo.BatchCallCount);
+        Assert.Equal(0, repo.InsertCallCount);
        Assert.Contains(repo.Inserted, e => e.EventId == e1.EventId);
        Assert.Contains(repo.Inserted, e => e.EventId == e2.EventId);
    }

+    /// <summary>
+    /// Review 2026-10-21 06/PERF-1: a busy but healthy site whose backlog spans several pages
+    /// is drained by paging WITHIN one tick, so the stalled latch does not fire.
+    /// </summary>
+    [Fact]
+    public void BusyHealthySite_PagesWithinOneTick_AndDoesNotLatchStalled()
+    {
+        var sites = new StaticEnumerator(new SiteEntry("siteA", "http://siteA:8083"));
+        var t = new DateTime(2026, 5, 20, 10, 0, 0, DateTimeKind.Utc);
+        var pages = Enumerable.Range(0, 3)
+            .Select(p => Enumerable.Range(0, 3).Select(i => NewEvent("siteA", t.AddSeconds(p * 10 + i))).ToArray())
+            .ToArray();
+        var client = new ScriptedPullClient().Script("siteA",
+            new PullAuditEventsResponse(pages[0], MoreAvailable: true),
+            new PullAuditEventsResponse(pages[1], MoreAvailable: true),
+            new PullAuditEventsResponse(pages[2], MoreAvailable: false));
+        var repo = new RecordingRepo();
+        var (probe, _) = SubscribeStalled();
+
+        var actor = CreateActor(sites, client, repo, FastTickOptions(stalledAfter: 1));
+
+        AwaitAssert(() => Assert.Equal(9, repo.Inserted.Count),
+            duration: TimeSpan.FromSeconds(3),
+            interval: TimeSpan.FromMilliseconds(50));
+        probe.ExpectNoMsg(TimeSpan.FromMilliseconds(500));
+        StopReconciliationActor(actor);
+
+        // The three pages were issued back to back with an advancing cursor.
+        Assert.Equal(pages[0][^1].EventId.ToString(), client.Calls[1].AfterId);
+        Assert.Equal(pages[1][^1].EventId.ToString(), client.Calls[2].AfterId);
+    }
+
    // ---------------------------------------------------------------------
    // 4. Tick_Duplicates_NotDoubleInserted (real MSSQL idempotency)
    // ---------------------------------------------------------------------
@@ -774,7 +832,9 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture<MsSqlMig
        var repo = new RecordingRepo();
        var (probe, _) = SubscribeStalled();

-        var actor = CreateActor(sites, client, repo, FastTickOptions(stalledAfter: 2));
+        // One page per tick: the backlog is still MoreAvailable when the page cap is hit, which
+        // is the genuine non-drain the stalled latch exists for.
+        var actor = CreateActor(sites, client, repo, FastTickOptions(stalledAfter: 2, maxPagesPerTick: 1));

        // Expect Stalled=true after the second non-draining tick. The probe
        // waits with its own timeout (a few seconds gives the 0 s repeat
@@ -808,7 +868,7 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture<MsSqlMig
        var repo = new RecordingRepo();
        var (probe, _) = SubscribeStalled();

-        var actor = CreateActor(sites, client, repo, FastTickOptions(stalledAfter: 2));
+        var actor = CreateActor(sites, client, repo, FastTickOptions(stalledAfter: 2, maxPagesPerTick: 1));

        // First publication is the stalled=true transition; second is the
        // back-to-draining flip. The actor publishes ONLY on transitions so we
@@ -0,0 +1,144 @@
+using Microsoft.Extensions.Logging.Abstractions;
+using Microsoft.Extensions.Options;
+using ZB.MOM.WW.Audit;
+using ZB.MOM.WW.ScadaBridge.AuditLog.Site;
+using ZB.MOM.WW.ScadaBridge.AuditLog.Tests.TestSupport;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
+
+namespace ZB.MOM.WW.ScadaBridge.AuditLog.Tests.Site;
+
+/// <summary>
+/// Forward-state lifecycle regressions for review 2026-10-21 06/PERF-1 and 06/IMPL-1: the
+/// reconciliation pull serves only rows central has NOT acknowledged (never Forwarded), the
+/// push-exit states <see cref="SiteAuditForwardStates.PushAbandoned"/> and
+/// <see cref="SiteAuditForwardStates.Rejected"/> stay owed to the pull until a covering cursor
+/// retires them, and a row central keeps refusing leaves the push queue after N attempts.
+/// </summary>
+public class SqliteAuditWriterForwardStateTests : IDisposable
+{
+    private readonly string _dbPath = Path.Combine(Path.GetTempPath(), $"audit-fwdstate-{Guid.NewGuid():N}.db");
+    private readonly CountingFailureCounter _counter = new();
+
+    /// <summary>Deletes the temporary database.</summary>
+    public void Dispose()
+    {
+        foreach (var suffix in new[] { "", "-wal", "-shm" })
+        {
+            try { File.Delete(_dbPath + suffix); } catch { /* best-effort */ }
+        }
+    }
+
+    private SqliteAuditWriter CreateWriter() => new(
+        Options.Create(new SqliteAuditWriterOptions { DatabasePath = _dbPath }),
+        NullLogger<SqliteAuditWriter>.Instance,
+        new FakeNodeIdentityProvider(),
+        failureCounter: _counter);
+
+    private static async Task<Guid> WriteAsync(SqliteAuditWriter writer, DateTime occurredAtUtc)
+    {
+        var id = Guid.NewGuid();
+        await writer.WriteAsync(ScadaBridgeAuditEventFactory.Create(
+            eventId: id,
+            occurredAtUtc: occurredAtUtc,
+            channel: AuditChannel.ApiOutbound,
+            kind: AuditKind.ApiCall,
+            status: AuditStatus.Delivered));
+        return id;
+    }
+
+    /// <summary>The pull no longer re-ships rows the push already delivered.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task ReadPendingSince_ExcludesForwardedRows()
+    {
+        await using var writer = CreateWriter();
+        var t = DateTime.UtcNow.AddMinutes(-5);
+        var forwarded = await WriteAsync(writer, t);
+        var pending = await WriteAsync(writer, t.AddSeconds(1));
+        await writer.MarkForwardedAsync(new[] { forwarded });
+
+        var served = await writer.ReadPendingSinceAsync(DateTime.MinValue, 100);
+
+        Assert.Equal(pending, Assert.Single(served).EventId);
+    }
+
+    /// <summary>A covering cursor retires only owed rows and leaves Forwarded rows alone.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task MarkReconciledUpTo_DoesNotTouchForwardedRows()
+    {
+        await using var writer = CreateWriter();
+        var t = DateTime.UtcNow.AddMinutes(-5);
+        var forwarded = await WriteAsync(writer, t);
+        var pending = await WriteAsync(writer, t.AddSeconds(1));
+        await writer.MarkForwardedAsync(new[] { forwarded });
+        var served = await writer.ReadPendingSinceAsync(DateTime.MinValue, 100);
+
+        var flipped = await writer.MarkReconciledUpToAsync(
+            served[^1].OccurredAtUtc.UtcDateTime, served[^1].EventId.ToString());
+
+        Assert.Equal(1, flipped);
+        Assert.Equal("Forwarded", writer.GetForwardStateForTests(forwarded)!.Value.State);
+        Assert.Equal("Reconciled", writer.GetForwardStateForTests(pending)!.Value.State);
+    }
+
+    /// <summary>A push-abandoned row leaves the push queue, is served by the pull, then retires.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task PushAbandoned_IsServedByThePull_ThenMarkedReconciled_AndNeverPurgedBeforeThat()
+    {
+        await using var writer = CreateWriter();
+        var old = DateTime.UtcNow.AddDays(-10);
+        var abandoned = await WriteAsync(writer, old);
+
+        await writer.MarkPushAbandonedAsync(new[] { abandoned });
+
+        Assert.Empty(await writer.ReadPendingAsync(100));
+        // Still owed: never purged on age, counted on the backlog.
+        Assert.Equal(0, await writer.PurgeExpiredAsync(DateTime.UtcNow));
+        Assert.Equal(1, (await writer.GetBacklogStatsAsync()).PendingCount);
+
+        var served = await writer.ReadPendingSinceAsync(DateTime.MinValue, 100);
+        Assert.Equal(abandoned, Assert.Single(served).EventId);
+
+        await writer.MarkReconciledUpToAsync(old, abandoned.ToString());
+
+        Assert.Equal("Reconciled", writer.GetForwardStateForTests(abandoned)!.Value.State);
+        Assert.Equal(1, await writer.PurgeExpiredAsync(DateTime.UtcNow));
+    }
+
+    /// <summary>
+    /// A row central keeps refusing bumps its attempt history every time and, at the threshold,
+    /// leaves the push queue as Rejected — unwedging the push head — while staying owed to the pull.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task RefusedRow_BumpsAttempts_ThenFlipsToRejected_UnwedgingThePushHead()
+    {
+        await using var writer = CreateWriter();
+        var t = DateTime.UtcNow.AddMinutes(-5);
+        var poison = await WriteAsync(writer, t);
+        var behind = await WriteAsync(writer, t.AddSeconds(1));
+
+        Assert.Equal(0, await writer.RecordRejectedPushAttemptsAsync(new[] { poison }, maxAttempts: 3));
+        Assert.Equal(0, await writer.RecordRejectedPushAttemptsAsync(new[] { poison }, maxAttempts: 3));
+        Assert.Equal(("Pending", 2), writer.GetForwardStateForTests(poison)!.Value);
+        Assert.Equal(poison, (await writer.ReadPendingAsync(1))[0].EventId); // still the head
+
+        Assert.Equal(1, await writer.RecordRejectedPushAttemptsAsync(new[] { poison }, maxAttempts: 3));
+
+        Assert.Equal(("Rejected", 3), writer.GetForwardStateForTests(poison)!.Value);
+        Assert.Equal(behind, Assert.Single(await writer.ReadPendingAsync(1)).EventId);
+        Assert.Contains(await writer.ReadPendingSinceAsync(DateTime.MinValue, 100), e => e.EventId == poison);
+        Assert.Equal(0, await writer.PurgeExpiredAsync(DateTime.UtcNow.AddDays(1)));
+        Assert.Equal(1, _counter.Count);
+    }
+
+    private sealed class CountingFailureCounter : IAuditWriteFailureCounter
+    {
+        private int _count;
+        public int Count => Volatile.Read(ref _count);
+        public void Increment() => Interlocked.Increment(ref _count);
+    }
+}
@@ -488,12 +488,16 @@ public class SqliteAuditWriterWriteTests

    // ----- Reconciliation pull surface ----- //

-    /// <summary>Verifies that the since-read returns pending and forwarded rows oldest first, limited to the requested count.</summary>
+    /// <summary>
+    /// Verifies that the since-read returns only rows central has NOT acknowledged, oldest first,
+    /// limited to the requested count — Forwarded rows are no longer re-shipped
+    /// (review 2026-10-21 06/PERF-1).
+    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
-    public async Task ReadPendingSinceAsync_Returns_PendingAndForwarded_OldestFirst_LimitedToN()
+    public async Task ReadPendingSinceAsync_Returns_OnlyUnackedRows_OldestFirst_LimitedToN()
    {
-        var (writer, dataSource) = CreateWriter(nameof(ReadPendingSinceAsync_Returns_PendingAndForwarded_OldestFirst_LimitedToN));
+        var (writer, dataSource) = CreateWriter(nameof(ReadPendingSinceAsync_Returns_OnlyUnackedRows_OldestFirst_LimitedToN));
        await using var _ = writer;

        var baseTime = new DateTime(2026, 5, 20, 12, 0, 0, DateTimeKind.Utc);
@@ -504,11 +508,11 @@ public class SqliteAuditWriterWriteTests
            NewEvent(occurredAtUtc: baseTime.AddSeconds(3)),
            NewEvent(occurredAtUtc: baseTime.AddSeconds(2)),
            NewEvent(occurredAtUtc: baseTime.AddSeconds(4)),
+            NewEvent(occurredAtUtc: baseTime.AddSeconds(6)),
        };
        foreach (var e in evts) await writer.WriteAsync(e);

-        // Flip half to Forwarded — they must still surface in the reconciliation pull
-        // because central hasn't confirmed they were ingested yet.
+        // Central acked these over the push: the pull has nothing to repair for them.
        await writer.MarkForwardedAsync(new[] { evts[0].EventId, evts[2].EventId });

        var rows = await writer.ReadPendingSinceAsync(sinceUtc: DateTime.MinValue, batchSize: 3);
@@ -516,7 +520,7 @@ public class SqliteAuditWriterWriteTests
        Assert.Equal(3, rows.Count);
        Assert.Equal(baseTime.AddSeconds(1), rows[0].OccurredAtUtc);
        Assert.Equal(baseTime.AddSeconds(2), rows[1].OccurredAtUtc);
-        Assert.Equal(baseTime.AddSeconds(3), rows[2].OccurredAtUtc);
+        Assert.Equal(baseTime.AddSeconds(4), rows[2].OccurredAtUtc);
    }

    /// <summary>Verifies that the since-read excludes rows older than the supplied cutoff.</summary>
@@ -740,17 +744,17 @@ public class SqliteAuditWriterWriteTests
        Assert.Equal(lateStamped.EventId, Assert.Single(reread).EventId);
    }

-    /// <summary>Verifies that forwarded rows are retired by reconciling up to a cursor even if they were never pulled.</summary>
+    /// <summary>
+    /// Verifies that a reconciliation cursor leaves Forwarded rows alone: central acked them over
+    /// the push, they are already purge-eligible, and the pull never served them
+    /// (review 2026-10-21 06/PERF-1).
+    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
-    public async Task MarkReconciledUpToAsync_ForwardedRowsRetire_EvenWithoutHavingBeenPulled()
+    public async Task MarkReconciledUpToAsync_LeavesForwardedRowsAlone()
    {
-        // The bound applies to PENDING rows (nothing proves central saw them but the pull).
-        // A FORWARDED row was ACKED by central through the telemetry push path, so the cursor
-        // may retire it whether or not this node has ever served it in a pull — otherwise a
-        // site node that never serves a pull would accumulate acked rows forever.
        var (writer, dataSource) = CreateWriter(
-            nameof(MarkReconciledUpToAsync_ForwardedRowsRetire_EvenWithoutHavingBeenPulled));
+            nameof(MarkReconciledUpToAsync_LeavesForwardedRowsAlone));
        await using var _w = writer;

        var t0 = new DateTime(2026, 5, 20, 12, 0, 0, DateTimeKind.Utc);
@@ -764,8 +768,8 @@ public class SqliteAuditWriterWriteTests

        var flipped = await writer.MarkReconciledUpToAsync(t0.AddSeconds(30), afterId: null);

-        Assert.Equal(1, flipped);
-        Assert.Equal(AuditForwardState.Reconciled.ToString(), ReadForwardState(dataSource, pushed.EventId));
+        Assert.Equal(0, flipped);
+        Assert.Equal(AuditForwardState.Forwarded.ToString(), ReadForwardState(dataSource, pushed.EventId));
        Assert.Equal(AuditForwardState.Pending.ToString(), ReadForwardState(dataSource, neverShipped.EventId));
    }

@@ -752,6 +752,76 @@ public class SiteAuditTelemetryActorTests : TestKit
        Assert.DoesNotContain(logger.Levels, l => l >= Microsoft.Extensions.Logging.LogLevel.Error);
    }

+    /// <summary>
+    /// Review 2026-10-21 06/PERF-1: against a queue that supports the push lifecycle, an orphan
+    /// cached row past grace is marked PushAbandoned (still owed to the pull), NOT Forwarded.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task CachedDrain_OrphanPastGrace_IsMarkedPushAbandoned_NotForwarded()
+    {
+        var queue = Substitute.For<ISiteAuditQueue, ZB.MOM.WW.ScadaBridge.AuditLog.Site.ISiteAuditPushLifecycle>();
+        var lifecycle = (ZB.MOM.WW.ScadaBridge.AuditLog.Site.ISiteAuditPushLifecycle)queue;
+        var orphan = NewCachedEvent(AuditKind.CachedSubmit);
+        queue.ReadPendingAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
+            .Returns(Task.FromResult<IReadOnlyList<AuditEvent>>(Array.Empty<AuditEvent>()));
+        queue.ReadPendingCachedTelemetryAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
+            .Returns(
+                Task.FromResult<IReadOnlyList<AuditEvent>>(new[] { orphan }),
+                Task.FromResult<IReadOnlyList<AuditEvent>>(Array.Empty<AuditEvent>()));
+        _trackingStore.GetStatusAsync(Arg.Any<TrackedOperationId>(), Arg.Any<CancellationToken>())
+            .Returns(Task.FromResult<TrackingStatusSnapshot?>(null));
+
+        Sys.ActorOf(Props.Create(() => new SiteAuditTelemetryActor(
+            queue, _client, Opts(), NullLogger<SiteAuditTelemetryActor>.Instance,
+            (IOperationTrackingStore?)_trackingStore)));
+
+        await AwaitAssertAsync(async () =>
+            await lifecycle.Received(1).MarkPushAbandonedAsync(
+                Arg.Is<IReadOnlyList<Guid>>(g => g.Count == 1 && g[0] == orphan.EventId),
+                Arg.Any<CancellationToken>()),
+            TimeSpan.FromSeconds(10));
+        await queue.DidNotReceive().MarkForwardedAsync(Arg.Any<IReadOnlyList<Guid>>(), Arg.Any<CancellationToken>());
+    }
+
+    /// <summary>
+    /// Review 2026-10-21 06/IMPL-1: rows central ANSWERED for but did not accept get a refused
+    /// attempt recorded (so a poison head row eventually leaves the push queue); accepted rows
+    /// do not, and a transport fault records nothing.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task Drain_RefusedRows_RecordRejectedAttempts_TransportFaultDoesNot()
+    {
+        var queue = Substitute.For<ISiteAuditQueue, ZB.MOM.WW.ScadaBridge.AuditLog.Site.ISiteAuditPushLifecycle>();
+        var lifecycle = (ZB.MOM.WW.ScadaBridge.AuditLog.Site.ISiteAuditPushLifecycle)queue;
+        var accepted = NewEvent();
+        var refused = NewEvent();
+        queue.ReadPendingAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
+            .Returns(
+                Task.FromResult<IReadOnlyList<AuditEvent>>(new[] { accepted, refused }),
+                Task.FromResult<IReadOnlyList<AuditEvent>>(new[] { accepted, refused }),
+                Task.FromResult<IReadOnlyList<AuditEvent>>(Array.Empty<AuditEvent>()));
+        _client.IngestAuditEventsAsync(Arg.Any<AuditEventBatch>(), Arg.Any<CancellationToken>())
+            .Returns(
+                _ => Task.FromException<IngestAck>(new InvalidOperationException("transport down")),
+                _ => Task.FromResult(AckAll(new[] { accepted })));
+
+        Sys.ActorOf(Props.Create(() => new SiteAuditTelemetryActor(
+            queue, _client, Opts(), NullLogger<SiteAuditTelemetryActor>.Instance,
+            (IOperationTrackingStore?)null)));
+
+        await AwaitAssertAsync(async () =>
+            await lifecycle.Received(1).RecordRejectedPushAttemptsAsync(
+                Arg.Is<IReadOnlyList<Guid>>(g => g.Count == 1 && g[0] == refused.EventId),
+                10,
+                Arg.Any<CancellationToken>()),
+            TimeSpan.FromSeconds(10));
+        await queue.Received(1).MarkForwardedAsync(
+            Arg.Is<IReadOnlyList<Guid>>(g => g.Count == 1 && g[0] == accepted.EventId),
+            Arg.Any<CancellationToken>());
+    }
+
    private sealed class CapturingLogger<T> : Microsoft.Extensions.Logging.ILogger<T>
    {
        public System.Collections.Concurrent.ConcurrentQueue<Microsoft.Extensions.Logging.LogLevel> Levels { get; } = new();