dohertj2/ScadaBridge
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

chore(data-layer): low-severity cleanups — SQLite param chunking, KPI metric catalog, computed-column predicate guard

Browse Source
This commit is contained in:
Joseph Doherty
2026-07-09 09:32:04 -04:00
parent b58d810dbe
commit acd8a6b8b5
6 changed files with 138 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs
+72 -39
View File
@@ -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;
}
}
src/ZB.MOM.WW.ScadaBridge.Commons/Types/Kpi/KpiMetrics.cs
+6
View File
@@ -38,11 +38,17 @@ public static class KpiMetrics
/// <summary>Pending-queue depth (rows awaiting delivery).</summary>
public const string QueueDepth = "queueDepth";
/// <summary>Stuck-row count (non-terminal older than the stuck-age threshold).</summary>
public const string StuckCount = "stuckCount";
/// <summary>Parked-row count.</summary>
public const string ParkedCount = "parkedCount";
/// <summary>Rows delivered in the last KPI interval.</summary>
public const string DeliveredLastInterval = "deliveredLastInterval";
/// <summary>Age of the oldest pending row, in seconds.</summary>
public const string OldestPendingAgeSeconds = "oldestPendingAgeSeconds";
}
/// <summary>
src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Configurations/AuditLogEntityTypeConfiguration.cs
+6
View File
@@ -155,6 +155,12 @@ public class AuditLogEntityTypeConfiguration : IEntityTypeConfiguration<AuditLog
// rejected at CREATE. It is not indexed, so non-persistence costs nothing.
// Routed through the nullable UTC converter so the materialised value
// carries Kind=Utc.
//
// NEVER use IngestedAtUtc in a WHERE predicate — it is non-persisted
// (stored:false), so any filter on it forces a full-table JSON_VALUE parse of
// DetailsJson on every row (no index can back it). Read-side projection only
// (arch-review 04, P7). Filter on OccurredAtUtc (the partition-aligned clustered
// key) instead.
builder.Property(e => e.IngestedAtUtc)
.HasColumnType("datetime2(7)")
.HasConversion(NullableUtcConverter)
src/ZB.MOM.WW.ScadaBridge.NotificationOutbox/Kpi/NotificationOutboxKpiSampleSource.cs
+2 -2
View File
@@ -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;
tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SqliteAuditWriterWriteTests.cs
+38
View File
@@ -590,6 +590,44 @@ public class SqliteAuditWriterWriteTests
}
}
/// <summary>
/// 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
/// <see cref="SqliteAuditWriter.MarkForwardedAsync"/> call flip atomically — and
/// <see cref="SqliteAuditWriter.MarkReconciledAsync"/> chunks the same way.
/// </summary>
[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<Guid>(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]
tests/ZB.MOM.WW.ScadaBridge.NotificationOutbox.Tests/Kpi/NotificationOutboxKpiSampleSourceTests.cs
+14
View File
@@ -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()
{