fix(controlplane): audit writer drops only already-persisted EventIds, never the whole SQL Server batch

SQL Server's DbUpdateException for one duplicate names every entry in the statement batch, so detaching ex.Entries dropped the whole flush. On a row-level rejection the writer now asks the database which of the batch's EventIds it already holds, drops exactly those, re-saves the rest, and falls back to row-by-row only if a non-duplicate row is still refused. Fresh context + entities per attempt. Pinned against the real central SQL Server (new SqlServerTestDatabase harness) and a faithful whole-batch interceptor. Refs #733 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
2026-10-22 13:42:29 -04:00
parent f1c52385ed
commit f7e7ba0263
3 changed files with 255 additions and 65 deletions
@@ -17,8 +17,8 @@ namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Audit;
 /// Dedup is two-layer: in-buffer (the <see cref="Dictionary{TKey, TValue}"/> below collapses
 /// duplicate EventIds before flush) and at the database via the filtered unique index
 /// <c>UX_ConfigAuditLog_EventId</c> (cross-restart safety — a retry of an already-flushed
-/// batch hits the constraint; <see cref="SaveDetachingRejectedRows"/> detaches the rejected row
-/// and re-saves, so the duplicate is dropped without losing the rest of the batch).
+/// batch hits the constraint; <see cref="SaveBatch"/> drops exactly the already-persisted EventIds
+/// and re-saves the rest, so the duplicate is dropped without losing the rest of the batch).
 ///
 /// Implements the shared <see cref="IAuditWriter"/> seam: <see cref="WriteAsync"/> is a
 /// best-effort, never-throwing entry point that simply <c>Tell</c>s this actor and returns
@@ -92,22 +92,7 @@ public sealed class AuditWriterActor : ReceiveActor, IWithTimers, IAuditWriter

        try
        {
-            using var db = _dbFactory.CreateDbContext();
-            foreach (var evt in snapshot)
-            {
-                db.ConfigAuditLogs.Add(new ConfigAuditLog
-                {
-                    Timestamp = evt.OccurredAtUtc.UtcDateTime,
-                    Principal = evt.Actor,
-                    EventType = $"{evt.Category}:{evt.Action}",
-                    NodeId = evt.SourceNode,
-                    DetailsJson = evt.DetailsJson,
-                    EventId = evt.EventId,
-                    CorrelationId = evt.CorrelationId,
-                    Outcome = evt.Outcome.ToString(),
-                });
-            }
-            var dropped = SaveDetachingRejectedRows(db);
+            var dropped = SaveBatch(snapshot);
            _log.Debug("AuditWriter flushed {Count} events ({Dropped} rejected rows dropped)", snapshot.Count - dropped, dropped);
        }
        catch (Exception ex)
@@ -116,46 +101,93 @@ public sealed class AuditWriterActor : ReceiveActor, IWithTimers, IAuditWriter
        }
    }

+    private static ConfigAuditLog ToRow(AuditEvent evt) => new()
+    {
+        Timestamp = evt.OccurredAtUtc.UtcDateTime,
+        Principal = evt.Actor,
+        EventType = $"{evt.Category}:{evt.Action}",
+        NodeId = evt.SourceNode,
+        DetailsJson = evt.DetailsJson,
+        EventId = evt.EventId,
+        CorrelationId = evt.CorrelationId,
+        Outcome = evt.Outcome.ToString(),
+    };
+
    /// <summary>
-    ///     Saves the batch, detaching and retrying past any row the database rejects (R03:2.6). One
-    ///     statement batch used to carry the whole flush: a single duplicate <c>EventId</c> (any at-least-once
-    ///     producer — the audit seam is Tell-based through a singleton proxy that re-delivers across a
-    ///     hand-over) threw for all 500 rows and the class remarks' per-row dedup claim was not what ran.
+    ///     Persists one flush, dropping ONLY the rows the database will not take (R03:2.6, fix round 1 / I4).
    /// </summary>
-    /// <param name="db">The context holding the batch.</param>
+    /// <param name="batch">The flush.</param>
    /// <returns>How many rows were dropped.</returns>
    /// <remarks>
-    ///     Only a <see cref="DbUpdateException"/> that NAMES its entries is handled here — that is the shape a
-    ///     constraint violation takes (SQL 2601/2627 for the unique index). A failure with no entries (the
-    ///     connection died, a deadlock) is the whole-batch failure it always was and propagates to the
-    ///     caller's catch. Bounded by the batch size so a store that rejects every row still terminates.
+    ///     <para>
+    ///         <b>Why not detach <c>ex.Entries</c>.</b> SQL Server inserts the batch as one statement batch, so
+    ///         the <see cref="DbUpdateException"/> raised for ONE duplicate <c>EventId</c> names EVERY entry in
+    ///         the batch. Detaching what it names dropped the whole flush — the exact loss the fix was for
+    ///         (pinned against a real SQL Server by <c>On_sql_server_a_duplicate_event_id_costs_only_that_row</c>).
+    ///     </para>
+    ///     <para>
+    ///         So, on a rejection that names entries (a constraint violation; 2601/2627 for the unique index):
+    ///         (1) ask the database which of the batch's EventIds it already holds and drop exactly those —
+    ///         the at-least-once re-delivery the dedup contract expects; (2) re-save the rest as a batch; (3) if
+    ///         that is still rejected (some OTHER per-row constraint), fall back to row-by-row and drop only the
+    ///         rows that fail on their own. Every attempt uses a fresh context and fresh entities, so no state
+    ///         from a failed save leaks into the next one. A failure that names no entry (connection loss,
+    ///         deadlock) propagates — the whole-batch failure it always was. Bounded: at most 2 + N saves.
+    ///     </para>
    /// </remarks>
-    private int SaveDetachingRejectedRows(OtOpcUaConfigDbContext db)
+    private int SaveBatch(IReadOnlyList<AuditEvent> batch)
    {
-        var dropped = 0;
-        var attempts = 0;
-        var maxAttempts = db.ChangeTracker.Entries<ConfigAuditLog>().Count() + 1;
-        while (true)
+        if (TrySave(batch, out var first)) return 0;
+        _log.Debug(first, "AuditWriter: a batch of {Count} was rejected; dropping only already-persisted EventIds", batch.Count);
+
+        HashSet<Guid> existing;
+        using (var db = _dbFactory.CreateDbContext())
        {
-            try
-            {
-                db.SaveChanges();
-                return dropped;
-            }
-            catch (DbUpdateException ex) when (ex.Entries.Count > 0 && ++attempts < maxAttempts)
-            {
-                var duplicate = IsUniqueViolation(ex);
-                foreach (var entry in ex.Entries)
-                {
-                    var eventId = (entry.Entity as ConfigAuditLog)?.EventId;
-                    if (duplicate)
-                        _log.Debug("AuditWriter: EventId {EventId} already persisted (unique index); dropping the duplicate", eventId);
-                    else
-                        _log.Warning(ex, "AuditWriter: the database rejected EventId {EventId}; dropping that row and keeping the batch", eventId);
-                    entry.State = EntityState.Detached;
-                    dropped++;
-                }
-            }
+            var ids = batch.Select(e => (Guid?)e.EventId).ToList();
+            existing = db.ConfigAuditLogs
+                .AsNoTracking()
+                .Where(r => ids.Contains(r.EventId))
+                .Select(r => r.EventId!.Value)
+                .ToHashSet();
+        }
+        foreach (var id in existing)
+            _log.Debug("AuditWriter: EventId {EventId} already persisted (unique index); dropping the duplicate", id);
+
+        var remaining = batch.Where(e => !existing.Contains(e.EventId)).ToList();
+        var dropped = batch.Count - remaining.Count;
+        if (remaining.Count == 0 || TrySave(remaining, out _)) return dropped;
+
+        // Still rejected with no duplicate left to blame: isolate the offending row(s).
+        foreach (var evt in remaining)
+        {
+            if (TrySave([evt], out var rowError)) continue;
+            dropped++;
+            if (IsUniqueViolation(rowError!))
+                _log.Debug("AuditWriter: EventId {EventId} already persisted (unique index); dropping the duplicate", evt.EventId);
+            else
+                _log.Warning(rowError, "AuditWriter: the database rejected EventId {EventId}; dropping that row and keeping the batch", evt.EventId);
+        }
+        return dropped;
+    }
+
+    /// <summary>Saves <paramref name="events"/> on a fresh context; a rejection that names entries is reported, anything else throws.</summary>
+    /// <param name="events">The rows to save.</param>
+    /// <param name="rejection">The row-level rejection, when this returns false.</param>
+    /// <returns><see langword="true"/> when saved.</returns>
+    private bool TrySave(IReadOnlyList<AuditEvent> events, out DbUpdateException? rejection)
+    {
+        using var db = _dbFactory.CreateDbContext();
+        foreach (var evt in events) db.ConfigAuditLogs.Add(ToRow(evt));
+        try
+        {
+            db.SaveChanges();
+            rejection = null;
+            return true;
+        }
+        catch (DbUpdateException ex) when (ex.Entries.Count > 0 || IsUniqueViolation(ex))
+        {
+            rejection = ex;
+            return false;
        }
    }

@@ -26,7 +26,7 @@ public sealed class AuditWriterActorRejectedRowTests : ControlPlaneActorTestBase
        var dbName = Guid.NewGuid().ToString("N");
        var root = new InMemoryDatabaseRoot();
        var duplicate = Guid.NewGuid();
-        var rejecting = new RejectingDbFactory(dbName, root, new RejectOnce(duplicate));
+        var rejecting = new RejectingDbFactory(dbName, root, new AlwaysRejectBatchContaining(duplicate));
        var actor = Sys.ActorOf(AuditWriterActor.Props(rejecting));

        var ids = Enumerable.Range(0, 20).Select(_ => Guid.NewGuid()).ToList();
@@ -46,6 +46,68 @@ public sealed class AuditWriterActorRejectedRowTests : ControlPlaneActorTestBase
        persisted.ShouldContain(ids[8]);
    }

+    /// <summary>
+    ///     Fix round 1 / I4 — a FAITHFUL SQL Server shape: the provider sends the batch as one statement batch,
+    ///     so the <see cref="DbUpdateException"/> for one duplicate names EVERY entry in it. Detaching
+    ///     <c>ex.Entries</c> therefore dropped the whole flush. The writer must drop only the rows whose
+    ///     EventId is already persisted and keep the rest.
+    /// </summary>
+    [Fact]
+    public void A_whole_batch_rejection_drops_only_the_rows_already_persisted()
+    {
+        var dbName = Guid.NewGuid().ToString("N");
+        var root = new InMemoryDatabaseRoot();
+        var factory = new RejectingDbFactory(dbName, root, new RejectWholeBatchOnDuplicate());
+        var duplicate = Guid.NewGuid();
+        using (var seed = factory.CreateDbContext())
+        {
+            seed.ConfigAuditLogs.Add(new ConfigAuditLog { Principal = "earlier", EventType = "Config:Edit", EventId = duplicate });
+            seed.SaveChanges();
+        }
+        var actor = Sys.ActorOf(AuditWriterActor.Props(factory));
+
+        var ids = Enumerable.Range(0, 20).Select(_ => Guid.NewGuid()).ToList();
+        ids.Insert(7, duplicate);
+        foreach (var id in ids) actor.Tell(NewEvent(id));
+        Watch(actor);
+        actor.Tell(PoisonPill.Instance);
+        ExpectTerminated(actor);
+
+        using var db = factory.CreateDbContext();
+        var persisted = db.ConfigAuditLogs.Select(r => r.EventId).ToList();
+        persisted.Count.ShouldBe(21, "the 20 new rows land beside the one that was already there");
+        persisted.Count(id => id == duplicate).ShouldBe(1);
+        db.ConfigAuditLogs.Single(r => r.EventId == duplicate).Principal.ShouldBe("earlier");
+    }
+
+    /// <summary>The same contract against the REAL provider and the real filtered unique index.</summary>
+    [Fact]
+    [Trait("Category", "SchemaCompliance")]
+    public void On_sql_server_a_duplicate_event_id_costs_only_that_row()
+    {
+        using var sql = new SqlServerTestDatabase();
+        var duplicate = Guid.NewGuid();
+        using (var seed = sql.NewContext())
+        {
+            seed.ConfigAuditLogs.Add(new ConfigAuditLog { Principal = "earlier", EventType = "Config:Edit", EventId = duplicate });
+            seed.SaveChanges();
+        }
+        var actor = Sys.ActorOf(AuditWriterActor.Props(sql.Factory()));
+
+        var ids = Enumerable.Range(0, 20).Select(_ => Guid.NewGuid()).ToList();
+        ids.Insert(7, duplicate);
+        foreach (var id in ids) actor.Tell(NewEvent(id));
+        Watch(actor);
+        actor.Tell(PoisonPill.Instance);
+        ExpectTerminated(actor, PresenceBudget);
+
+        using var db = sql.NewContext();
+        var persisted = db.ConfigAuditLogs.Select(r => r.EventId).ToList();
+        persisted.Count.ShouldBe(21);
+        persisted.Count(id => id == duplicate).ShouldBe(1);
+        foreach (var id in ids) persisted.ShouldContain(id);
+    }
+
    /// <summary>A failure that names no entry is a whole-batch failure, exactly as before.</summary>
    [Fact]
    public void A_connection_level_failure_still_drops_the_batch_without_looping()
@@ -71,24 +133,43 @@ public sealed class AuditWriterActorRejectedRowTests : ControlPlaneActorTestBase
        DetailsJson = "{}", SourceNode = "node-a", CorrelationId = Guid.NewGuid(),
    };

-    /// <summary>Throws a <see cref="DbUpdateException"/> naming the entry for one EventId, the first time it is saved.</summary>
-    private sealed class RejectOnce(Guid eventId) : ISaveChangesInterceptor
+    /// <summary>
+    ///     A row the database refuses on its own terms (not a duplicate — e.g. a per-row constraint), every
+    ///     time it is in the save. Faithful to SQL Server's batch shape: the exception names EVERY added
+    ///     entry, so only a row-by-row retry can isolate the offender.
+    /// </summary>
+    private sealed class AlwaysRejectBatchContaining(Guid eventId) : ISaveChangesInterceptor
    {
-        private int _fired;
-
        public InterceptionResult<int> SavingChanges(DbContextEventData eventData, InterceptionResult<int> result)
        {
-            if (_fired == 1 || eventData.Context is null) return result;
-            var entry = eventData.Context.ChangeTracker.Entries<ConfigAuditLog>()
-                .FirstOrDefault(e => e.Entity.EventId == eventId);
-            if (entry is null) return result;
-            _fired = 1;
-            // GetInfrastructure() hands back the pubternal InternalEntityEntry, which is the IUpdateEntry the
-            // real provider puts on the exception; the test only needs that one cast.
+            if (eventData.Context is null) return result;
+            var added = eventData.Context.ChangeTracker.Entries<ConfigAuditLog>()
+                .Where(e => e.State == EntityState.Added).ToList();
+            if (!added.Any(e => e.Entity.EventId == eventId)) return result;
 #pragma warning disable EF1001
-            var updateEntry = (IUpdateEntry)entry.GetInfrastructure();
+            var entries = added.Select(e => (IUpdateEntry)e.GetInfrastructure()).ToList();
 #pragma warning restore EF1001
-            throw new DbUpdateException("Cannot insert duplicate key row (simulated 2601)", new[] { updateEntry });
+            throw new DbUpdateException("The statement conflicted with a CHECK constraint (simulated, whole batch)", entries);
+        }
+    }
+
+    /// <summary>
+    ///     Models SQL Server's batched insert: when ANY added row's EventId is already stored, the whole save
+    ///     fails with an exception naming EVERY added entry (the InMemory provider enforces no unique index,
+    ///     so this interceptor is the constraint).
+    /// </summary>
+    private sealed class RejectWholeBatchOnDuplicate : ISaveChangesInterceptor
+    {
+        public InterceptionResult<int> SavingChanges(DbContextEventData eventData, InterceptionResult<int> result)
+        {
+            if (eventData.Context is not OtOpcUaConfigDbContext ctx) return result;
+            var added = ctx.ChangeTracker.Entries<ConfigAuditLog>().Where(e => e.State == EntityState.Added).ToList();
+            var ids = added.Select(e => e.Entity.EventId).Where(id => id != null).ToList();
+            if (ids.Count == 0 || !ctx.ConfigAuditLogs.AsNoTracking().Any(r => ids.Contains(r.EventId))) return result;
+#pragma warning disable EF1001
+            var entries = added.Select(e => (IUpdateEntry)e.GetInfrastructure()).ToList();
+#pragma warning restore EF1001
+            throw new DbUpdateException("Cannot insert duplicate key row (simulated 2601, whole batch)", entries);
        }
    }

@@ -0,0 +1,77 @@
+using Microsoft.Data.SqlClient;
+using Microsoft.EntityFrameworkCore;
+using ZB.MOM.WW.OtOpcUa.Configuration;
+
+namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Harness;
+
+/// <summary>
+///     A migrated, throw-away config database on the central SQL Server, for the few ControlPlane behaviours
+///     the InMemory provider cannot model (a SQL Server <c>DbUpdateException</c> names the WHOLE statement
+///     batch; collation-ordered reads). Same server / credentials / env overrides as
+///     <c>Configuration.Tests.SchemaComplianceFixture</c> (<c>OTOPCUA_CONFIG_TEST_SERVER</c>,
+///     <c>OTOPCUA_CONFIG_TEST_SA_PASSWORD</c>; default the shared docker host's dev SA), and the same database
+///     name shape, so that suite's 24 h orphan reaper also reclaims a database a killed run leaks here.
+/// </summary>
+public sealed class SqlServerTestDatabase : IDisposable
+{
+    private const string DefaultServer = "10.100.0.35,14330";
+    private const string DefaultSaPassword = "OtOpcUaDev_2026!";
+
+    /// <summary>Gets the connection string of the migrated database.</summary>
+    public string ConnectionString { get; }
+
+    private readonly string _databaseName;
+
+    /// <summary>Creates and migrates the database.</summary>
+    public SqlServerTestDatabase()
+    {
+        var server = Environment.GetEnvironmentVariable("OTOPCUA_CONFIG_TEST_SERVER") ?? DefaultServer;
+        var saPassword = Environment.GetEnvironmentVariable("OTOPCUA_CONFIG_TEST_SA_PASSWORD") ?? DefaultSaPassword;
+        _databaseName = $"OtOpcUaConfig_Test_{DateTime.UtcNow:yyyyMMddHHmmss}_{Guid.NewGuid():N}";
+        ConnectionString =
+            $"Server={server};Database={_databaseName};User Id=sa;Password={saPassword};TrustServerCertificate=True;Encrypt=False;";
+        using var ctx = NewContext();
+        ctx.Database.Migrate();
+    }
+
+    /// <summary>A new context on the database.</summary>
+    /// <returns>The context.</returns>
+    public OtOpcUaConfigDbContext NewContext() =>
+        new(new DbContextOptionsBuilder<OtOpcUaConfigDbContext>().UseSqlServer(ConnectionString).Options);
+
+    /// <summary>A factory over <see cref="NewContext"/>.</summary>
+    /// <returns>The factory.</returns>
+    public IDbContextFactory<OtOpcUaConfigDbContext> Factory() => new ContextFactory(this);
+
+    /// <summary>Drops the database (best effort; a leak is reaped by the Configuration suite).</summary>
+    public void Dispose()
+    {
+        try
+        {
+            SqlConnection.ClearAllPools();
+            var master = new SqlConnectionStringBuilder(ConnectionString) { InitialCatalog = "master" }.ConnectionString;
+            using var conn = new SqlConnection(master);
+            conn.Open();
+            using var cmd = conn.CreateCommand();
+            cmd.CommandText = @"
+DECLARE @sql nvarchar(max);
+IF DB_ID(@name) IS NOT NULL
+BEGIN
+    SET @sql = N'ALTER DATABASE ' + QUOTENAME(@name) + N' SET SINGLE_USER WITH ROLLBACK IMMEDIATE; ' +
+               N'DROP DATABASE ' + QUOTENAME(@name) + N';';
+    EXEC sp_executesql @sql;
+END";
+            cmd.Parameters.Add(new SqlParameter("@name", _databaseName));
+            cmd.ExecuteNonQuery();
+        }
+        catch (Exception)
+        {
+            // Best effort — see the class remarks.
+        }
+    }
+
+    private sealed class ContextFactory(SqlServerTestDatabase owner) : IDbContextFactory<OtOpcUaConfigDbContext>
+    {
+        public OtOpcUaConfigDbContext CreateDbContext() => owner.NewContext();
+    }
+}