fix(auditlog): explicit permanent rejections on the ingest ack (IngestAck.rejected_event_ids, field 2) with a dedicated counter; reconciler ignores transient faults for abandonment; persist the served high-water mark; prune only on a non-empty enumeration (review 2026-10-21 06/PERF-1, 06/IMPL-1 adversarial follow-up)

2026-10-22 08:40:42 -04:00
parent 38167bca88
commit b0a08bb63b
23 changed files with 790 additions and 168 deletions
@@ -0,0 +1,132 @@
+using Akka.Actor;
+using Akka.TestKit.Xunit2;
+using Microsoft.Extensions.Logging.Abstractions;
+using NSubstitute;
+using NSubstitute.ExceptionExtensions;
+using ZB.MOM.WW.Audit;
+using ZB.MOM.WW.ScadaBridge.AuditLog.Central;
+using ZB.MOM.WW.ScadaBridge.AuditLog.Tests.TestSupport;
+using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
+using ZB.MOM.WW.ScadaBridge.Commons.Messages.Audit;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
+
+namespace ZB.MOM.WW.ScadaBridge.AuditLog.Tests.Central;
+
+/// <summary>
+/// Review 2026-10-21 06/IMPL-1 adversarial follow-up: central reports a row as REJECTED only for a
+/// permanent, row-intrinsic SQL failure; a database outage / timeout leaves the id in neither list,
+/// so the site never mistakes a maintenance window for a poison row. Also pins the classifier the
+/// reconciliation puller uses to decide what counts toward permanent abandonment (06/COR follow-up 3).
+/// </summary>
+public class AuditIngestRejectionTests : TestKit
+{
+    private static AuditEvent NewEvent() => ScadaBridgeAuditEventFactory.Create(
+        eventId: Guid.NewGuid(),
+        occurredAtUtc: new DateTime(2026, 5, 20, 10, 0, 0, DateTimeKind.Utc),
+        channel: AuditChannel.ApiOutbound,
+        kind: AuditKind.ApiCall,
+        status: AuditStatus.Delivered,
+        sourceSiteId: "site-a");
+
+    /// <summary>The batch reply lists only the over-width row as rejected; the timed-out row is in neither list.</summary>
+    [Fact]
+    public void Ingest_ReportsOnlyPermanentRowFailuresAsRejected()
+    {
+        var ok = NewEvent();
+        var overWidth = NewEvent();
+        var timedOut = NewEvent();
+        var repo = Substitute.For<IAuditLogRepository>();
+        repo.InsertManyIfNotExistsAsync(Arg.Any<IReadOnlyList<AuditEvent>>(), Arg.Any<TimeSpan?>(), Arg.Any<CancellationToken>())
+            .ThrowsAsync(SqlExceptionFactory.Create(2628));
+        repo.InsertIfNotExistsAsync(Arg.Is<AuditEvent>(e => e.EventId == overWidth.EventId), Arg.Any<CancellationToken>())
+            .ThrowsAsync(SqlExceptionFactory.Create(2628));
+        repo.InsertIfNotExistsAsync(Arg.Is<AuditEvent>(e => e.EventId == timedOut.EventId), Arg.Any<CancellationToken>())
+            .ThrowsAsync(SqlExceptionFactory.Create(-2));
+
+        var actor = Sys.ActorOf(Props.Create(() => new AuditLogIngestActor(
+            StubAuditRepositoryProvider.Build(repo), NullLogger<AuditLogIngestActor>.Instance)));
+        actor.Tell(new IngestAuditEventsCommand(new[] { ok, overWidth, timedOut }), TestActor);
+
+        var reply = ExpectMsg<IngestAuditEventsReply>(TimeSpan.FromSeconds(10));
+        Assert.Equal(new[] { ok.EventId }, reply.AcceptedEventIds);
+        Assert.Equal(new[] { overWidth.EventId }, reply.RejectedEventIds);
+    }
+
+    /// <summary>Classifier: row-intrinsic SQL errors are permanent; timeouts, connection loss and non-SQL faults are not.</summary>
+    [Fact]
+    public void Classifier_SeparatesRowVerdictsFromDatabaseProblems()
+    {
+        Assert.True(AuditIngestErrorClassifier.IsPermanentRowRejection(SqlExceptionFactory.Create(2628)));
+        Assert.True(AuditIngestErrorClassifier.IsPermanentRowRejection(SqlExceptionFactory.Create(515)));
+        Assert.True(AuditIngestErrorClassifier.IsPermanentRowRejection(
+            new InvalidOperationException("wrapped", SqlExceptionFactory.Create(8152))));
+        Assert.False(AuditIngestErrorClassifier.IsPermanentRowRejection(SqlExceptionFactory.Create(-2)));
+        Assert.False(AuditIngestErrorClassifier.IsPermanentRowRejection(new TimeoutException()));
+        Assert.False(AuditIngestErrorClassifier.IsPermanentRowRejection(new InvalidOperationException("bug")));
+
+        Assert.True(AuditIngestErrorClassifier.IsTransient(SqlExceptionFactory.Create(-2)));
+        Assert.True(AuditIngestErrorClassifier.IsTransient(SqlExceptionFactory.Create(1205)));
+        Assert.True(AuditIngestErrorClassifier.IsTransient(SqlExceptionFactory.Create(40613)));
+        Assert.True(AuditIngestErrorClassifier.IsTransient(new TimeoutException()));
+        Assert.True(AuditIngestErrorClassifier.IsTransient(new TaskCanceledException()));
+        Assert.True(AuditIngestErrorClassifier.IsTransient(
+            new InvalidOperationException("wrapped", new System.Net.Sockets.SocketException())));
+        Assert.False(AuditIngestErrorClassifier.IsTransient(SqlExceptionFactory.Create(2628)));
+        Assert.False(AuditIngestErrorClassifier.IsTransient(new InvalidOperationException("bug")));
+    }
+
+    private sealed class OneRowClient : IPullAuditEventsClient
+    {
+        private readonly AuditEvent _row;
+        public OneRowClient(AuditEvent row) => _row = row;
+        public System.Collections.Concurrent.ConcurrentQueue<DateTime> Since { get; } = new();
+        public Task<ZB.MOM.WW.ScadaBridge.Commons.Messages.Integration.PullAuditEventsResponse> PullAsync(
+            string siteId, DateTime sinceUtc, string? afterId, int batchSize, CancellationToken ct)
+        {
+            Since.Enqueue(sinceUtc);
+            return Task.FromResult(new ZB.MOM.WW.ScadaBridge.Commons.Messages.Integration.PullAuditEventsResponse(
+                new[] { _row }, MoreAvailable: false));
+        }
+    }
+
+    private sealed class OneSite : ISiteEnumerator
+    {
+        public Task<IReadOnlyList<SiteEntry>> EnumerateAsync(CancellationToken ct = default) =>
+            Task.FromResult<IReadOnlyList<SiteEntry>>(new[] { new SiteEntry("site-a", "http://a:8083") });
+    }
+
+    /// <summary>
+    /// A row that keeps hitting a TRANSIENT database fault is retried with the cursor held back for
+    /// far longer than the permanent-abandon threshold, and is never abandoned.
+    /// </summary>
+    [Fact]
+    public void Reconciliation_TransientFaults_NeverCountTowardAbandonment()
+    {
+        var row = NewEvent();
+        var repo = Substitute.For<IAuditLogRepository>();
+        repo.InsertManyIfNotExistsAsync(Arg.Any<IReadOnlyList<AuditEvent>>(), Arg.Any<TimeSpan?>(), Arg.Any<CancellationToken>())
+            .ThrowsAsync(SqlExceptionFactory.Create(-2));
+        repo.InsertIfNotExistsAsync(Arg.Any<AuditEvent>(), Arg.Any<CancellationToken>())
+            .ThrowsAsync(SqlExceptionFactory.Create(-2));
+        var client = new OneRowClient(row);
+
+        var actor = Sys.ActorOf(Props.Create(() => new SiteAuditReconciliationActor(
+            new OneSite(), client, StubAuditRepositoryProvider.Build(repo),
+            Microsoft.Extensions.Options.Options.Create(new SiteAuditReconciliationOptions
+            {
+                ReconciliationIntervalOverride = TimeSpan.FromMilliseconds(30),
+            }),
+            NullLogger<SiteAuditReconciliationActor>.Instance)));
+
+        AwaitCondition(() => client.Since.Count >= 12, TimeSpan.FromSeconds(10));
+        Watch(actor);
+        Sys.Stop(actor);
+        ExpectTerminated(actor);
+
+        Assert.All(client.Since, s => Assert.Equal(DateTime.MinValue, s)); // cursor held every tick
+        repo.DidNotReceive().InsertIfNotExistsAsync(
+            Arg.Is<AuditEvent>(e => e.Action.EndsWith(".ReconciliationAbandoned", StringComparison.Ordinal)),
+            Arg.Any<CancellationToken>());
+    }
+}
@@ -116,9 +116,11 @@ public class ReconciliationStatePruningTests : TestKit

        Assert.True(probe.ExpectMsg<SiteAuditTelemetryStalledChanged>(TimeSpan.FromSeconds(5)).Stalled);

-        sites.Sites = Array.Empty<SiteEntry>();
+        // Not an empty list: an empty enumeration deliberately prunes nothing.
+        sites.Sites = new[] { new SiteEntry("survivor", "http://s:8083") };

-        var cleared = probe.ExpectMsg<SiteAuditTelemetryStalledChanged>(TimeSpan.FromSeconds(5));
+        var cleared = probe.FishForMessage<SiteAuditTelemetryStalledChanged>(
+            m => m.SiteId == "doomed", TimeSpan.FromSeconds(5));
        Assert.Equal("doomed", cleared.SiteId);
        Assert.False(cleared.Stalled);

@@ -18,7 +18,6 @@ namespace ZB.MOM.WW.ScadaBridge.AuditLog.Tests.Site;
 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()
@@ -32,8 +31,7 @@ public class SqliteAuditWriterForwardStateTests : IDisposable
    private SqliteAuditWriter CreateWriter() => new(
        Options.Create(new SqliteAuditWriterOptions { DatabasePath = _dbPath }),
        NullLogger<SqliteAuditWriter>.Instance,
-        new FakeNodeIdentityProvider(),
-        failureCounter: _counter);
+        new FakeNodeIdentityProvider());

    private static async Task<Guid> WriteAsync(SqliteAuditWriter writer, DateTime occurredAtUtc)
    {
@@ -132,13 +130,39 @@ public class SqliteAuditWriterForwardStateTests : IDisposable
        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);
+        Assert.Equal(1, writer.RejectedPushRowCount);
    }

-    private sealed class CountingFailureCounter : IAuditWriteFailureCounter
+    /// <summary>
+    /// The served high-water mark survives a writer restart: rows served before the restart are
+    /// retired by the first cursor that covers them after it, instead of being stranded owed
+    /// (and counted on the backlog) until something happens to re-serve them.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task ServedHighWaterMark_SurvivesRestart_SoTheNextCursorStillRetires()
    {
-        private int _count;
-        public int Count => Volatile.Read(ref _count);
-        public void Increment() => Interlocked.Increment(ref _count);
+        var t = DateTime.UtcNow.AddMinutes(-5);
+        Guid abandoned;
+        IReadOnlyList<AuditEvent> served;
+        await using (var writer = CreateWriter())
+        {
+            abandoned = await WriteAsync(writer, t);
+            await writer.MarkPushAbandonedAsync(new[] { abandoned });
+            served = await writer.ReadPendingSinceAsync(DateTime.MinValue, 100);
+            Assert.Single(served);
+        }
+
+        await using (var restarted = CreateWriter())
+        {
+            Assert.True(restarted.MaxServedRowIdForTests > 0);
+
+            var flipped = await restarted.MarkReconciledUpToAsync(
+                served[0].OccurredAtUtc.UtcDateTime, served[0].EventId.ToString());
+
+            Assert.Equal(1, flipped);
+            Assert.Equal("Reconciled", restarted.GetForwardStateForTests(abandoned)!.Value.State);
+            Assert.Equal(0, (await restarted.GetBacklogStatsAsync()).PendingCount);
+        }
    }
 }
@@ -785,27 +785,33 @@ public class SiteAuditTelemetryActorTests : TestKit
    }

    /// <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.
+    /// Review 2026-10-21 06/IMPL-1 (adversarial follow-up): only rows central EXPLICITLY lists in
+    /// <c>RejectedEventIds</c> get a refused attempt recorded. A row merely missing from the accepted
+    /// list (central DB outage / budget exhaustion) and a transport fault record nothing.
    /// </summary>
    /// <returns>A task that represents the asynchronous operation.</returns>
    [Fact]
-    public async Task Drain_RefusedRows_RecordRejectedAttempts_TransportFaultDoesNot()
+    public async Task Drain_OnlyExplicitlyRejectedRows_RecordRefusals()
    {
        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();
+        var poison = NewEvent();
+        var unanswered = NewEvent();
+        var rows = new[] { accepted, poison, unanswered };
        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>>(rows),
+                Task.FromResult<IReadOnlyList<AuditEvent>>(rows),
+                Task.FromResult<IReadOnlyList<AuditEvent>>(rows),
                Task.FromResult<IReadOnlyList<AuditEvent>>(Array.Empty<AuditEvent>()));
+        var explicitAck = AckAll(new[] { accepted });
+        explicitAck.RejectedEventIds.Add(poison.EventId.ToString());
        _client.IngestAuditEventsAsync(Arg.Any<AuditEventBatch>(), Arg.Any<CancellationToken>())
            .Returns(
                _ => Task.FromException<IngestAck>(new InvalidOperationException("transport down")),
-                _ => Task.FromResult(AckAll(new[] { accepted })));
+                _ => Task.FromResult(AckAll(new[] { accepted })),   // partial ack, no verdict (old central / DB outage)
+                _ => Task.FromResult(explicitAck));

        Sys.ActorOf(Props.Create(() => new SiteAuditTelemetryActor(
            queue, _client, Opts(), NullLogger<SiteAuditTelemetryActor>.Instance,
@@ -813,13 +819,12 @@ public class SiteAuditTelemetryActorTests : TestKit

        await AwaitAssertAsync(async () =>
            await lifecycle.Received(1).RecordRejectedPushAttemptsAsync(
-                Arg.Is<IReadOnlyList<Guid>>(g => g.Count == 1 && g[0] == refused.EventId),
+                Arg.Is<IReadOnlyList<Guid>>(g => g.Count == 1 && g[0] == poison.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>());
+            TimeSpan.FromSeconds(15));
+        await lifecycle.Received(1).RecordRejectedPushAttemptsAsync(
+            Arg.Any<IReadOnlyList<Guid>>(), Arg.Any<int>(), Arg.Any<CancellationToken>());
    }

    private sealed class CapturingLogger<T> : Microsoft.Extensions.Logging.ILogger<T>
@@ -0,0 +1,35 @@
+using System.Reflection;
+using Microsoft.Data.SqlClient;
+
+namespace ZB.MOM.WW.ScadaBridge.AuditLog.Tests.TestSupport;
+
+/// <summary>
+/// Fabricates a <see cref="SqlException"/> with a given error number via the driver's internal
+/// <c>CreateException</c> factory (the type has no public constructor) — same technique as
+/// ExternalSystemGateway.Tests' <c>FabricateSqlException</c>.
+/// </summary>
+internal static class SqlExceptionFactory
+{
+    /// <summary>Creates a <see cref="SqlException"/> carrying one error with <paramref name="number"/>.</summary>
+    /// <param name="number">The SQL Server error number.</param>
+    /// <param name="message">The error message.</param>
+    /// <returns>The fabricated exception.</returns>
+    public static SqlException Create(int number, string message = "simulated")
+    {
+        var errorCtor = typeof(SqlError).GetConstructors(BindingFlags.Instance | BindingFlags.NonPublic)
+            .First(c => c.GetParameters().Length == 8 && c.GetParameters()[7].ParameterType == typeof(Exception));
+        var sqlError = (SqlError)errorCtor.Invoke(
+            new object?[] { number, (byte)0, (byte)0, "server", message, "procedure", 0, null });
+
+        var collection = (SqlErrorCollection)typeof(SqlErrorCollection)
+            .GetConstructors(BindingFlags.Instance | BindingFlags.NonPublic).First()
+            .Invoke(Array.Empty<object?>());
+        typeof(SqlErrorCollection).GetMethod("Add", BindingFlags.Instance | BindingFlags.NonPublic)!
+            .Invoke(collection, new object?[] { sqlError });
+
+        var create = typeof(SqlException).GetMethod("CreateException",
+            BindingFlags.Static | BindingFlags.NonPublic,
+            new[] { typeof(SqlErrorCollection), typeof(string) })!;
+        return (SqlException)create.Invoke(null, new object?[] { collection, "6.0.0" })!;
+    }
+}
@@ -259,6 +259,23 @@ public class CentralControlDtoMapperTests
        Assert.Equal(ids, roundTripped);
    }

+    /// <summary>Verifies the additive rejected-id list round-trips and reads empty when absent.</summary>
+    [Fact]
+    public void IngestAck_RejectedIds_RoundTrip_AndAbsentFieldReadsEmpty()
+    {
+        // Review 2026-10-21 06/IMPL-1: additive field 2. An older central never sets it, so the
+        // site must read an EMPTY list (never a refusal) from a pre-field ack.
+        var accepted = new[] { Guid.NewGuid() };
+        var rejected = new[] { Guid.NewGuid(), Guid.NewGuid() };
+
+        var wire = Wire(CentralControlDtoMapper.ToIngestAck(accepted, rejected));
+
+        Assert.Equal(accepted, CentralControlDtoMapper.FromIngestAck(wire));
+        Assert.Equal(rejected, CentralControlDtoMapper.RejectedFromIngestAck(wire));
+        Assert.Empty(CentralControlDtoMapper.RejectedFromIngestAck(
+            Wire(CentralControlDtoMapper.ToIngestAck(accepted))));
+    }
+
    /// <summary>Verifies that an ingest acknowledgement with nothing accepted stays empty.</summary>
    [Fact]
    public void IngestAck_RoundTrip_NothingAccepted_StaysEmpty()