diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs index ffad9ae9..62108a76 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs @@ -55,6 +55,13 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable // transaction and never reached once audit_event's insert throws). private const int SqliteErrorConstraint = 19; + // Max EventIds bound into a single IN (...) UPDATE by MarkForwardedAsync / + // MarkReconciledAsync. SQLite caps bound parameters per statement (999 on older + // builds, SQLITE_MAX_VARIABLE_NUMBER); a large drain batch would otherwise blow the + // ceiling as batch sizes grow. Chunk the id list into ≤500-id slices, all inside one + // transaction so the batch stays atomic (arch-review 04, P8). + private const int MarkForwardStateChunkSize = 500; + private readonly SqliteConnection _connection; // Dedicated read-only connection used by GetBacklogStatsAsync, // ReadPendingAsync, ReadPendingSinceAsync, and ReadForwardedAsync so a slow @@ -625,13 +632,11 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable { ObjectDisposedException.ThrowIf(_disposed, this); - using var cmd = _connection.CreateCommand(); // Flip the sidecar — UPDATE audit_forward_state, not the canonical // audit_event (which is append-only / write-once). Bump AttemptCount + // stamp LastAttemptUtc so operators can see how many drain passes a row - // took to forward. Build a single IN (...) parameter list so we issue - // one UPDATE per batch regardless of size. Each id is bound as its own - // parameter, so no string concatenation of user data ever enters the SQL. + // took to forward. Each id is bound as its own parameter, so no string + // concatenation of user data ever enters the SQL. // // Defensive state guard: only transition rows that are still Pending or // Forwarded (i.e. not yet Reconciled). Without this guard a mis-called @@ -641,25 +646,41 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable // WHERE ForwardState IN ($pending, $forwarded) // guard. Current callers only pass Pending IDs, so normal-path behaviour // is unchanged; the guard is purely defensive. - var sb = new System.Text.StringBuilder(); - sb.Append("UPDATE audit_forward_state SET ForwardState = $forwarded, ") - .Append("AttemptCount = AttemptCount + 1, LastAttemptUtc = $now ") - .Append("WHERE ForwardState IN ($pending, $forwarded) AND EventId IN ("); - for (int i = 0; i < eventIds.Count; i++) - { - if (i > 0) sb.Append(','); - var p = $"$id{i}"; - sb.Append(p); - cmd.Parameters.AddWithValue(p, eventIds[i].ToString()); - } - sb.Append(");"); - cmd.CommandText = sb.ToString(); - cmd.Parameters.AddWithValue("$forwarded", AuditForwardState.Forwarded.ToString()); - cmd.Parameters.AddWithValue("$pending", AuditForwardState.Pending.ToString()); - cmd.Parameters.AddWithValue("$now", DateTime.UtcNow.ToString( - "o", System.Globalization.CultureInfo.InvariantCulture)); + // + // Stamp one $now for the whole call so every chunk records the same + // LastAttemptUtc (matching the prior single-statement behaviour). + var now = DateTime.UtcNow.ToString( + "o", System.Globalization.CultureInfo.InvariantCulture); + var forwarded = AuditForwardState.Forwarded.ToString(); + var pending = AuditForwardState.Pending.ToString(); - cmd.ExecuteNonQuery(); + // Chunk into ≤MarkForwardStateChunkSize-id UPDATEs, all inside ONE + // transaction so the whole batch commits atomically (arch-review 04, P8). + using var transaction = _connection.BeginTransaction(); + for (int start = 0; start < eventIds.Count; start += MarkForwardStateChunkSize) + { + var chunk = Math.Min(MarkForwardStateChunkSize, eventIds.Count - start); + using var cmd = _connection.CreateCommand(); + cmd.Transaction = transaction; + var sb = new System.Text.StringBuilder(); + sb.Append("UPDATE audit_forward_state SET ForwardState = $forwarded, ") + .Append("AttemptCount = AttemptCount + 1, LastAttemptUtc = $now ") + .Append("WHERE ForwardState IN ($pending, $forwarded) AND EventId IN ("); + for (int i = 0; i < chunk; i++) + { + if (i > 0) sb.Append(','); + var p = $"$id{i}"; + sb.Append(p); + cmd.Parameters.AddWithValue(p, eventIds[start + i].ToString()); + } + sb.Append(");"); + cmd.CommandText = sb.ToString(); + cmd.Parameters.AddWithValue("$forwarded", forwarded); + cmd.Parameters.AddWithValue("$pending", pending); + cmd.Parameters.AddWithValue("$now", now); + cmd.ExecuteNonQuery(); + } + transaction.Commit(); return Task.CompletedTask; } } @@ -728,27 +749,39 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable { ObjectDisposedException.ThrowIf(_disposed, this); - using var cmd = _connection.CreateCommand(); // Flip the sidecar from Pending/Forwarded → Reconciled. Rows // already Reconciled are left untouched (idempotent re-call), and the // canonical audit_event row is never modified. - var sb = new System.Text.StringBuilder(); - sb.Append("UPDATE audit_forward_state SET ForwardState = $reconciled ") - .Append("WHERE ForwardState IN ($pending, $forwarded) AND EventId IN ("); - for (int i = 0; i < eventIds.Count; i++) - { - if (i > 0) sb.Append(','); - var p = $"$id{i}"; - sb.Append(p); - cmd.Parameters.AddWithValue(p, eventIds[i].ToString()); - } - sb.Append(");"); - cmd.CommandText = sb.ToString(); - cmd.Parameters.AddWithValue("$reconciled", AuditForwardState.Reconciled.ToString()); - cmd.Parameters.AddWithValue("$pending", AuditForwardState.Pending.ToString()); - cmd.Parameters.AddWithValue("$forwarded", AuditForwardState.Forwarded.ToString()); + var reconciled = AuditForwardState.Reconciled.ToString(); + var pending = AuditForwardState.Pending.ToString(); + var forwarded = AuditForwardState.Forwarded.ToString(); - cmd.ExecuteNonQuery(); + // Chunk into ≤MarkForwardStateChunkSize-id UPDATEs, all inside ONE + // transaction so the whole batch commits atomically (arch-review 04, P8). + using var transaction = _connection.BeginTransaction(); + for (int start = 0; start < eventIds.Count; start += MarkForwardStateChunkSize) + { + var chunk = Math.Min(MarkForwardStateChunkSize, eventIds.Count - start); + using var cmd = _connection.CreateCommand(); + cmd.Transaction = transaction; + var sb = new System.Text.StringBuilder(); + sb.Append("UPDATE audit_forward_state SET ForwardState = $reconciled ") + .Append("WHERE ForwardState IN ($pending, $forwarded) AND EventId IN ("); + for (int i = 0; i < chunk; i++) + { + if (i > 0) sb.Append(','); + var p = $"$id{i}"; + sb.Append(p); + cmd.Parameters.AddWithValue(p, eventIds[start + i].ToString()); + } + sb.Append(");"); + cmd.CommandText = sb.ToString(); + cmd.Parameters.AddWithValue("$reconciled", reconciled); + cmd.Parameters.AddWithValue("$pending", pending); + cmd.Parameters.AddWithValue("$forwarded", forwarded); + cmd.ExecuteNonQuery(); + } + transaction.Commit(); return Task.CompletedTask; } } diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Types/Kpi/KpiMetrics.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Types/Kpi/KpiMetrics.cs index 9c4baf14..d2230812 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Commons/Types/Kpi/KpiMetrics.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Types/Kpi/KpiMetrics.cs @@ -38,11 +38,17 @@ public static class KpiMetrics /// Pending-queue depth (rows awaiting delivery). public const string QueueDepth = "queueDepth"; + /// Stuck-row count (non-terminal older than the stuck-age threshold). + public const string StuckCount = "stuckCount"; + /// Parked-row count. public const string ParkedCount = "parkedCount"; /// Rows delivered in the last KPI interval. public const string DeliveredLastInterval = "deliveredLastInterval"; + + /// Age of the oldest pending row, in seconds. + public const string OldestPendingAgeSeconds = "oldestPendingAgeSeconds"; } /// diff --git a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Configurations/AuditLogEntityTypeConfiguration.cs b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Configurations/AuditLogEntityTypeConfiguration.cs index 70cc79d7..9b86006e 100644 --- a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Configurations/AuditLogEntityTypeConfiguration.cs +++ b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Configurations/AuditLogEntityTypeConfiguration.cs @@ -155,6 +155,12 @@ public class AuditLogEntityTypeConfiguration : IEntityTypeConfiguration e.IngestedAtUtc) .HasColumnType("datetime2(7)") .HasConversion(NullableUtcConverter) diff --git a/src/ZB.MOM.WW.ScadaBridge.NotificationOutbox/Kpi/NotificationOutboxKpiSampleSource.cs b/src/ZB.MOM.WW.ScadaBridge.NotificationOutbox/Kpi/NotificationOutboxKpiSampleSource.cs index 02393dd0..35ce1da8 100644 --- a/src/ZB.MOM.WW.ScadaBridge.NotificationOutbox/Kpi/NotificationOutboxKpiSampleSource.cs +++ b/src/ZB.MOM.WW.ScadaBridge.NotificationOutbox/Kpi/NotificationOutboxKpiSampleSource.cs @@ -37,10 +37,10 @@ public sealed class NotificationOutboxKpiSampleSource : IKpiSampleSource // Charted metrics share the public Commons catalog so source + UI trend page key // off one symbol. The uncharted internal metrics stay private here. private const string MetricQueueDepth = KpiMetrics.NotificationOutbox.QueueDepth; - private const string MetricStuckCount = "stuckCount"; + private const string MetricStuckCount = KpiMetrics.NotificationOutbox.StuckCount; private const string MetricParkedCount = KpiMetrics.NotificationOutbox.ParkedCount; private const string MetricDeliveredLastInterval = KpiMetrics.NotificationOutbox.DeliveredLastInterval; - private const string MetricOldestPendingAgeSeconds = "oldestPendingAgeSeconds"; + private const string MetricOldestPendingAgeSeconds = KpiMetrics.NotificationOutbox.OldestPendingAgeSeconds; private readonly INotificationOutboxRepository _repository; private readonly NotificationOutboxOptions _options; diff --git a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SqliteAuditWriterWriteTests.cs b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SqliteAuditWriterWriteTests.cs index c4559c09..9ee16e6c 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SqliteAuditWriterWriteTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SqliteAuditWriterWriteTests.cs @@ -590,6 +590,44 @@ public class SqliteAuditWriterWriteTests } } + /// + /// P8 (arch-review 04): a drain batch larger than SQLite's bound-parameter ceiling + /// (999 on older builds) must still flip every row. The writer chunks the id list + /// into ≤500-id UPDATEs inside one transaction, so 1200 ids in a single + /// call flip atomically — and + /// chunks the same way. + /// + [Fact] + public async Task MarkForwardedThenReconciled_LargeBatch_ChunksParameters_AllFlip() + { + var (writer, dataSource) = CreateWriter(nameof(MarkForwardedThenReconciled_LargeBatch_ChunksParameters_AllFlip)); + await using var _ = writer; + + const int count = 1200; // > 999 and > the 500 chunk size, so multiple chunks fire + var ids = new List(count); + for (int i = 0; i < count; i++) + { + var id = Guid.NewGuid(); + ids.Add(id); + await writer.WriteAsync(NewEvent(id)); + } + + // One MarkForwardedAsync call over all 1200 ids — must not throw on the + // parameter ceiling and must flip every sidecar row to Forwarded. + await writer.MarkForwardedAsync(ids); + + var afterForward = ForwardStateCounts(dataSource); + Assert.Equal(count, afterForward[AuditForwardState.Forwarded.ToString()]); + Assert.False(afterForward.ContainsKey(AuditForwardState.Pending.ToString())); + + // Same for reconciliation over the full batch. + await writer.MarkReconciledAsync(ids); + + var afterReconcile = ForwardStateCounts(dataSource); + Assert.Equal(count, afterReconcile[AuditForwardState.Reconciled.ToString()]); + Assert.False(afterReconcile.ContainsKey(AuditForwardState.Forwarded.ToString())); + } + // ----- ExecutionId (rides DetailsJson, recomposed via AsRow) ----- // [Fact] diff --git a/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/Kpi/NotificationOutboxKpiSampleSourceTests.cs b/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/Kpi/NotificationOutboxKpiSampleSourceTests.cs index 87747bc1..0084c436 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/Kpi/NotificationOutboxKpiSampleSourceTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/Kpi/NotificationOutboxKpiSampleSourceTests.cs @@ -33,6 +33,20 @@ public class NotificationOutboxKpiSampleSourceTests Assert.Equal(KpiSources.NotificationOutbox, source.Source); } + // C4 (arch-review 04): the two metric names that were inline string literals now + // live in the shared KpiMetrics catalog. Their VALUES are persisted in the KpiSample + // store and rendered by the trend UI, so a rename would silently orphan history — + // pin the exact historical strings here. + [Fact] + public void KpiMetricCatalog_NotificationOutbox_PreservesHistoricalStringValues() + { + Assert.Equal("queueDepth", KpiMetrics.NotificationOutbox.QueueDepth); + Assert.Equal("stuckCount", KpiMetrics.NotificationOutbox.StuckCount); + Assert.Equal("parkedCount", KpiMetrics.NotificationOutbox.ParkedCount); + Assert.Equal("deliveredLastInterval", KpiMetrics.NotificationOutbox.DeliveredLastInterval); + Assert.Equal("oldestPendingAgeSeconds", KpiMetrics.NotificationOutbox.OldestPendingAgeSeconds); + } + [Fact] public async Task CollectAsync_PassesCutoffsAnchoredOnCapturedAt() {