From 88273693f9767005d62956d6dc4f55bfa604ea5c Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 09:35:30 -0400 Subject: [PATCH] fix(site-call-audit): reconciliation pull serves only un-acked tracking rows (node-local ack ledger), batched page upsert, prune removed sites' cursors (review 2026-10-21 06/PERF-2 + lane H follow-up) --- docs/requirements/Component-SiteCallAudit.md | 35 +++- .../Site/Telemetry/SiteAuditTelemetryActor.cs | 50 ++++++ .../Repositories/ISiteCallAuditRepository.cs | 21 +++ .../Services/IOperationTrackingStore.cs | 29 ++++ .../Repositories/SiteCallAuditRepository.cs | 78 +++++++++ .../SiteCallAuditActor.cs | 44 ++++- .../Tracking/OperationTrackingStore.cs | 94 +++++++++- .../Telemetry/SiteAuditTelemetryActorTests.cs | 48 ++++++ .../SiteCallAuditRepositoryTests.cs | 114 ++++++++++++ .../SiteCallAuditReconciliationTests.cs | 162 ++++++++++++++++++ .../TestSupport/StubRepositoryProvider.cs | 4 + .../Tracking/OperationTrackingStoreTests.cs | 120 +++++++++++++ 12 files changed, 789 insertions(+), 10 deletions(-) diff --git a/docs/requirements/Component-SiteCallAudit.md b/docs/requirements/Component-SiteCallAudit.md index d720ce8d..007cd740 100644 --- a/docs/requirements/Component-SiteCallAudit.md +++ b/docs/requirements/Component-SiteCallAudit.md @@ -135,8 +135,41 @@ Pinned)` on the EventStream (transition-only, mirroring health-observable condition rather than a silent log line, and the latch clears with `Pinned=false` once a later tick makes progress. +**The pull serves only what central has not acknowledged.** The site's +`ReadChangedSinceAsync` skips every tracking row whose *current* version +(`UpdatedAtUtc`) central already accepted through the cached-telemetry push, so in +steady state a reconciliation pull carries only genuinely lost telemetry rather than +a second copy of every lifecycle step — and a central failover, which resets the +in-memory cursors to `DateTime.MinValue`, no longer replays the whole tracking +retention window of every site. When central acks an `IngestCachedTelemetry` +packet (it commits the packet's `SiteCalls` upsert with its audit row), the site's +`SiteAuditTelemetryActor` records `(TrackedOperationId, snapshot UpdatedAtUtc)` in +the **node-local** `OperationTrackingCentralAck` table — deliberately a separate, +unregistered table, never a column on the LocalDb-replicated `OperationTracking` +table (same posture as `mirrored_secrets`): only the node that pushed saw central +accept it. The filter matches the version exactly, so any later change (even one +stamped by a clock that stepped back) is served again; acks are monotonic per row +and pruned with the tracking rows they describe. **Failover:** a site node that +has just become active holds no ack rows, so its pull serves every row once +(bounded by central's cursor); central's idempotent monotonic upsert absorbs the +duplicates and the ledger rebuilds from that node's own pushes. A lost ack only +costs one extra pull of that row. + +**Each pulled page is one batched upsert.** `ISiteCallAuditRepository.UpsertManyAsync` +runs the same two per-row statements as `UpsertAsync` (insert-if-absent, then the +ownership-checked monotonic update) for up to 128 rows per command — a 500-row page +is 4 round trips instead of 500. A chunk that faults (an insert race, or an +ownership violation) is re-applied row by row through `UpsertAsync`, which is inert +for rows already applied; the rest of the page is still applied and the first +row failure is rethrown, so the site's cursor stays put exactly as before. + +**Removed sites are pruned.** Each tick drops the cursor and pinned latch of any +site the enumerator no longer returns (a pinned one publishes its `Pinned=false` +transition first), so the dictionaries do not grow with site churn; a site that +reappears restarts from `DateTime.MinValue`. + **The drain runs off the mailbox.** A reconciliation pass is unbounded work — -every site, up to a page ceiling of network pulls each, one upsert per row — so +every site, up to a page ceiling of network pulls each, one batched upsert per page — so it runs as a background task with a `PipeTo`-delivered completion message and a single-flight guard, not as an actor message handler. A handler occupies the actor for its whole duration, so a post-outage catch-up used to park telemetry diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/Telemetry/SiteAuditTelemetryActor.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/Telemetry/SiteAuditTelemetryActor.cs index 6526bab1..4fb0253e 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/Telemetry/SiteAuditTelemetryActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/Telemetry/SiteAuditTelemetryActor.cs @@ -271,6 +271,9 @@ public class SiteAuditTelemetryActor : ReceiveActor var batch = new CachedTelemetryBatch(); var wireBytes = AuditWireBudget.InitialBytes; var emittedEventIds = new List(pending.Count); + // EventId → the tracking-row version its packet carried, so an accepted + // packet can be recorded as a central ack (review 2026-10-21 06/PERF-2). + var packetVersions = new Dictionary(pending.Count); // Rows whose operational half can never be built. They are marked // PushAbandoned so they LEAVE this queue — see AbandonUnresolvableAsync // for why that is safe and why leaving them Pending is not. @@ -338,6 +341,7 @@ public class SiteAuditTelemetryActor : ReceiveActor var packet = BuildCachedPacket(auditRow, snapshot); if (!AuditWireBudget.TryAppend(batch.Packets, packet, ref wireBytes, auditRow.EventId.ToString())) break; emittedEventIds.Add(auditRow.EventId); + packetVersions[auditRow.EventId] = (snapshot.Id, snapshot.UpdatedAtUtc); } if (lookupFailures > 0) @@ -389,6 +393,7 @@ public class SiteAuditTelemetryActor : ReceiveActor { await _queue.MarkForwardedAsync(acceptedIds, ct) .ConfigureAwait(false); + await RecordCentralAcksAsync(acceptedIds, packetVersions, ct).ConfigureAwait(false); } await RecordRefusalsAsync( @@ -413,6 +418,51 @@ public class SiteAuditTelemetryActor : ReceiveActor } } + /// + /// Tells the tracking store which tracking-row versions central has just accepted, so + /// the SiteCalls reconciliation pull stops re-serving them (review 2026-10-21 + /// 06/PERF-2). Accepted means central committed the packet's SiteCalls upsert + /// in the same transaction as its audit row, i.e. central now holds exactly what a pull + /// of that version would send. Several packets of one operation in a batch collapse to + /// the newest version. Best-effort: the store only enqueues, and a failure here just + /// leaves the rows eligible for one more pull. + /// + private async Task RecordCentralAcksAsync( + IReadOnlyList acceptedIds, + Dictionary packetVersions, + CancellationToken ct) + { + var versions = new Dictionary(); + foreach (var eventId in acceptedIds) + { + if (!packetVersions.TryGetValue(eventId, out var v)) + { + continue; + } + + if (!versions.TryGetValue(v.Id, out var existing) || v.UpdatedAtUtc > existing) + { + versions[v.Id] = v.UpdatedAtUtc; + } + } + + if (versions.Count == 0 || _trackingStore is null) + { + return; + } + + try + { + await _trackingStore.RecordCentralAcksAsync(versions, ct).ConfigureAwait(false); + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + _logger.LogDebug(ex, + "Recording {Count} central ack(s) in the tracking store failed; those rows stay eligible for the reconciliation pull.", + versions.Count); + } + } + /// /// Moves cached rows whose operational half can never be built out of the /// cached-drain queue, as . diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/ISiteCallAuditRepository.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/ISiteCallAuditRepository.cs index 90c40e1f..0180735d 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/ISiteCallAuditRepository.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/ISiteCallAuditRepository.cs @@ -42,6 +42,27 @@ public interface ISiteCallAuditRepository /// A task that represents the asynchronous operation. Task UpsertAsync(SiteCall siteCall, CancellationToken ct = default); + /// + /// Applies 's exact per-row semantics (insert-if-absent, then + /// the monotonic rank-guarded update, ownership-checked) to a whole page of rows in as + /// few round trips as the implementation can manage — the reconciliation pull's write + /// path (review 2026-10-21 06/PERF-2). Rows are applied in order; a row that faults + /// (e.g. an ownership violation) surfaces as the same exception + /// would throw — the SQL Server implementation applies the rest of the page first. The + /// default implementation simply upserts row by row and stops at the first fault. + /// + /// The rows to insert or monotonically update. + /// Cancellation token. + /// A task that represents the asynchronous operation. + async Task UpsertManyAsync(IReadOnlyList siteCalls, CancellationToken ct = default) + { + ArgumentNullException.ThrowIfNull(siteCalls); + foreach (var siteCall in siteCalls) + { + await UpsertAsync(siteCall, ct).ConfigureAwait(false); + } + } + /// /// Returns the row for the given id, or null if none exists. /// diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Services/IOperationTrackingStore.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Services/IOperationTrackingStore.cs index d175be04..d13b5309 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Services/IOperationTrackingStore.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Services/IOperationTrackingStore.cs @@ -150,6 +150,15 @@ public interface IOperationTrackingStore /// deterministic — UpdatedAtUtc ASC, TrackedOperationId ASC. /// /// + /// Only rows central has not already acknowledged. A row whose + /// current version (UpdatedAtUtc) central acknowledged through the + /// cached-telemetry push — recorded via — is + /// skipped: central already holds exactly what this read would send. The pull + /// therefore carries only lost telemetry, not a replay of every lifecycle step + /// (review 2026-10-21 06/PERF-2). The ack ledger is node-local, so a node that + /// has just become active serves every row once. + /// + /// /// is left as the empty string: /// the site id is not a tracking-store column, and the central client re-stamps /// it from the siteId it dialed (the only authority that knows which @@ -174,4 +183,24 @@ public interface IOperationTrackingStore int batchSize, string? afterId = null, CancellationToken ct = default); + + /// + /// Records that central accepted the cached-telemetry push of each operation at the + /// given tracking-row version (the snapshot's UpdatedAtUtc the push carried), + /// so stops serving that version to the + /// reconciliation pull. Monotonic per operation (an older version never replaces a + /// newer one) and best-effort: a lost ack only means the row is pulled once more, + /// which central's idempotent upsert absorbs. + /// + /// + /// The ledger lives in a NODE-LOCAL side table, never a column on the replicated + /// tracking table. The default implementation records nothing (every row stays + /// eligible for the pull — the pre-ack behaviour). + /// + /// Operation id → the UpdatedAtUtc of the snapshot central accepted. + /// Cancellation token. + /// A task that completes once the acks are queued. + Task RecordCentralAcksAsync( + IReadOnlyDictionary ackedVersions, + CancellationToken ct = default) => Task.CompletedTask; } diff --git a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/SiteCallAuditRepository.cs b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/SiteCallAuditRepository.cs index 7296ddc1..115bb61a 100644 --- a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/SiteCallAuditRepository.cs +++ b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/SiteCallAuditRepository.cs @@ -169,6 +169,84 @@ public class SiteCallAuditRepository : ISiteCallAuditRepository } } + /// + /// Rows per batched upsert command. Each row binds 15 parameters, so 128 rows stay + /// under SQL Server's 2 100-parameter limit (1 920) with headroom. + /// + internal const int UpsertManyChunkSize = 128; + + /// + public async Task UpsertManyAsync(IReadOnlyList siteCalls, CancellationToken ct = default) + { + ArgumentNullException.ThrowIfNull(siteCalls); + + // The reconciliation pull's write path (review 2026-10-21 06/PERF-2): one command + // per chunk of UpsertManyChunkSize rows instead of one round trip per row — a + // 500-row page is 4 round trips, not 500. Each row runs the SAME two statements + // UpsertAsync runs (InsertIfAbsentSql then MonotonicUpdateSql), with its parameters + // suffixed per row, so the insert-first / monotonic / ownership semantics are + // byte-identical rather than re-derived set-based. + // + // Any SqlException from a chunk — an insert race's 2601/2627 (statement-level, + // the batch runs on) or an ownership THROW 51001 (batch-terminating) — re-runs that + // chunk row by row through UpsertAsync. Re-applying rows the batch already applied + // is inert (equal rank + equal UpdatedAtUtc changes nothing) and UpsertAsync handles + // the race exactly as before. A genuinely bad row does not stop the rest of the + // page: the remaining rows are still applied and the FIRST such failure is + // rethrown at the end, so the caller sees the same exception the old per-row + // loop surfaced (and keeps its cursor), minus the collateral of the rows behind it. + System.Runtime.ExceptionServices.ExceptionDispatchInfo? firstFailure = null; + for (var offset = 0; offset < siteCalls.Count; offset += UpsertManyChunkSize) + { + var chunk = siteCalls.Skip(offset).Take(UpsertManyChunkSize).ToList(); + var sql = new System.Text.StringBuilder(); + var parameters = new List(chunk.Count * 15); + for (var i = 0; i < chunk.Count; i++) + { + var siteCall = chunk[i] ?? throw new ArgumentException("A null SiteCall row was supplied.", nameof(siteCalls)); + var rank = GetRankOrThrow(siteCall.Status); + var suffix = "_" + i.ToString(System.Globalization.CultureInfo.InvariantCulture); + sql.AppendLine(SuffixParameters(InsertIfAbsentSql, suffix)); + sql.AppendLine(SuffixParameters(MonotonicUpdateSql, suffix)); + foreach (SqlParameter p in BuildUpsertParameters(siteCall, siteCall.TrackedOperationId.Value.ToString("D"), rank)) + { + p.ParameterName += suffix; + parameters.Add(p); + } + } + + try + { + await _context.Database.ExecuteSqlRawAsync(sql.ToString(), parameters, ct).ConfigureAwait(false); + } + catch (SqlException ex) + { + _logger.LogDebug( + ex, + "SiteCallAuditRepository.UpsertManyAsync chunk of {Count} rows faulted (error {SqlErrorNumber}); re-applying it row by row.", + chunk.Count, + ex.Number); + foreach (var siteCall in chunk) + { + try + { + await UpsertAsync(siteCall, ct).ConfigureAwait(false); + } + catch (Exception rowFailure) when (rowFailure is not OperationCanceledException) + { + firstFailure ??= System.Runtime.ExceptionServices.ExceptionDispatchInfo.Capture(rowFailure); + } + } + } + } + + firstFailure?.Throw(); + } + + /// Renames every @Name parameter reference in to @Name{suffix}. + private static string SuffixParameters(string sql, string suffix) => + System.Text.RegularExpressions.Regex.Replace(sql, @"@(\w+)", "@$1" + suffix); + // Leg 1 of the upsert: create the row when it does not exist yet. Runs FIRST // so a concurrent first-write loser still has a row to update (see UpsertAsync). private const string InsertIfAbsentSql = @" diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteCallAudit/SiteCallAuditActor.cs b/src/ZB.MOM.WW.ScadaBridge.SiteCallAudit/SiteCallAuditActor.cs index ab9759e5..47786f52 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteCallAudit/SiteCallAuditActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteCallAudit/SiteCallAuditActor.cs @@ -499,6 +499,8 @@ public class SiteCallAuditActor : ReceiveActor return; } + PruneRemovedSites(sites); + if (sites.Count == 0) { return; @@ -516,6 +518,30 @@ public class SiteCallAuditActor : ReceiveActor await ReconcileSitesAsync(sites, client, repository).ConfigureAwait(false); } + /// + /// Drops the per-site cursor and pinned latch of every site the enumerator no longer + /// returns (a deleted or renamed site), so the dictionaries do not grow with site + /// churn for the singleton's lifetime. A site that was pinned publishes its + /// Pinned=false transition first, so a health subscriber does not keep showing + /// a pinned site that no longer exists. Runs on the single-flight pass that owns the + /// dictionaries. A site that later reappears simply starts again from + /// DateTime.MinValue, which the idempotent upsert absorbs. + /// + private void PruneRemovedSites(IReadOnlyList sites) + { + var live = new HashSet(sites.Select(s => s.SiteId), StringComparer.Ordinal); + foreach (var siteId in _reconciliationPinned.Keys.Where(k => !live.Contains(k)).ToList()) + { + SetReconciliationPinned(siteId, pinned: false); + _reconciliationPinned.Remove(siteId); + } + + foreach (var siteId in _reconciliationCursors.Keys.Where(k => !live.Contains(k)).ToList()) + { + _reconciliationCursors.Remove(siteId); + } + } + /// /// Reconciles every site in the tick against a single resolved repository, /// isolating per-site faults so one bad site never sinks the rest of the @@ -635,15 +661,19 @@ public class SiteCallAuditActor : ReceiveActor var maxUpdated = since; var maxAfterId = afterId; var nowUtc = DateTime.UtcNow; + + // IngestedAtUtc is the "central ingested (or last refreshed) this row" stamp — + // owned by the central actor, exactly as OnUpsertAsync does for the telemetry + // path. The whole page is written through ONE batched monotonic upsert (a few + // round trips per page instead of one per row — review 2026-10-21 06/PERF-2); + // a row already present from a prior push is a silent no-op. A fault propagates + // to the per-site catch BEFORE the cursor below moves, so the page is re-pulled. + await repository + .UpsertManyAsync(response.SiteCalls.Select(row => row with { IngestedAtUtc = nowUtc }).ToList()) + .ConfigureAwait(false); + foreach (var row in response.SiteCalls) { - // IngestedAtUtc is the "central ingested (or last refreshed) this - // row" stamp — owned by the central actor, exactly as OnUpsertAsync - // does for the telemetry path. Monotonic UpsertAsync makes a row - // already present (from a prior push) a silent no-op. - var siteCall = row with { IngestedAtUtc = nowUtc }; - await repository.UpsertAsync(siteCall).ConfigureAwait(false); - // Advance the composite max: greater timestamp wins; on a tie the // greater TrackedOperationId (ordinal, matching the site's SQLite // BINARY text collation) wins. diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Tracking/OperationTrackingStore.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Tracking/OperationTrackingStore.cs index 00563aca..ec2e37b4 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Tracking/OperationTrackingStore.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Tracking/OperationTrackingStore.cs @@ -141,7 +141,34 @@ public class OperationTrackingStore : IOperationTrackingStore, IAsyncDisposable, // which is strictly before this constructor can receive it. It stays because it // keeps a directly-constructed store (tests, tooling) self-sufficient, and the // DDL is idempotent. - private void InitializeSchema() => OperationTrackingSchema.Apply(_writeConnection); + private void InitializeSchema() + { + OperationTrackingSchema.Apply(_writeConnection); + using var cmd = _writeConnection.CreateCommand(); + cmd.CommandText = CentralAckSchemaSql; + cmd.ExecuteNonQuery(); + } + + /// + /// Node-local ledger of the tracking-row version central last acknowledged per + /// operation (review 2026-10-21 06/PERF-2). Deliberately a SEPARATE table, never a + /// column on OperationTracking: that table is LocalDb-replicated and its + /// capture triggers are built from its column list, while acks are per-node fact — + /// only the node that pushed the telemetry saw central accept it. Like + /// mirrored_secrets, it is not in SiteLocalDbSetup.ReplicatedTables. + /// + /// Failover: the standby holds no ack rows, so after it takes over its + /// reconciliation read serves every row once (bounded by central's cursor), and + /// central's idempotent monotonic upsert absorbs the duplicates. Pushes from the new + /// active node rebuild the ledger from then on. + /// + /// + private const string CentralAckSchemaSql = """ + CREATE TABLE IF NOT EXISTS OperationTrackingCentralAck ( + TrackedOperationId TEXT NOT NULL PRIMARY KEY, + AckedUpdatedAtUtc TEXT NOT NULL + ); + """; /// public async Task RecordEnqueueAsync( @@ -271,6 +298,51 @@ public class OperationTrackingStore : IOperationTrackingStore, IAsyncDisposable, }, ct).ConfigureAwait(false); } + /// + public async Task RecordCentralAcksAsync( + IReadOnlyDictionary ackedVersions, + CancellationToken ct = default) + { + ArgumentNullException.ThrowIfNull(ackedVersions); + if (ackedVersions.Count == 0) + { + return; + } + + // Rendered exactly as the write path stamps UpdatedAtUtc ("o", UTC 'Z'), so the + // read filter's text equality compares like with like. + var acks = ackedVersions + .Select(a => (Id: a.Key.ToString(), Version: DateTime + .SpecifyKind(a.Value, DateTimeKind.Utc) + .ToString("o", CultureInfo.InvariantCulture))) + .ToList(); + + // Rides the single-writer queue like a lifecycle write (enqueue and return) — + // the ack is best-effort, and FIFO keeps it behind the writes it describes. + await EnqueueAsync($"central-ack x{acks.Count}", (conn, tx) => + { + using var cmd = conn.CreateCommand(); + cmd.Transaction = tx; + // Monotonic: an older ack never rolls a newer one back. ISO-8601 "o" text + // orders lexicographically. + cmd.CommandText = """ + INSERT INTO OperationTrackingCentralAck (TrackedOperationId, AckedUpdatedAtUtc) + VALUES ($id, $version) + ON CONFLICT (TrackedOperationId) DO UPDATE + SET AckedUpdatedAtUtc = excluded.AckedUpdatedAtUtc + WHERE excluded.AckedUpdatedAtUtc > AckedUpdatedAtUtc; + """; + var idParam = cmd.Parameters.Add("$id", SqliteType.Text); + var versionParam = cmd.Parameters.Add("$version", SqliteType.Text); + foreach (var (id, version) in acks) + { + idParam.Value = id; + versionParam.Value = version; + cmd.ExecuteNonQuery(); + } + }, ct).ConfigureAwait(false); + } + // ── Queue plumbing ───────────────────────────────────────────────────── /// @@ -494,10 +566,17 @@ public class OperationTrackingStore : IOperationTrackingStore, IAsyncDisposable, cmd.Transaction = tx; // Non-terminal rows (TerminalAtUtc IS NULL) are kept regardless of // age — the operation is still in flight. + // The node-local ack ledger is pruned in the same transaction so it never + // outlives the rows it describes (it also drops acks for rows the peer's + // purge removed through replication). cmd.CommandText = """ DELETE FROM OperationTracking WHERE TerminalAtUtc IS NOT NULL AND TerminalAtUtc < $threshold; + DELETE FROM OperationTrackingCentralAck + WHERE NOT EXISTS ( + SELECT 1 FROM OperationTracking t + WHERE t.TrackedOperationId = OperationTrackingCentralAck.TrackedOperationId); """; cmd.Parameters.AddWithValue("$threshold", threshold); cmd.ExecuteNonQuery(); @@ -543,12 +622,23 @@ public class OperationTrackingStore : IOperationTrackingStore, IAsyncDisposable, var predicate = useKeyset ? "UpdatedAtUtc > $since OR (UpdatedAtUtc = $since AND TrackedOperationId > $afterId)" : "UpdatedAtUtc >= $since"; + // Rows whose CURRENT version central already acknowledged through the + // cached-telemetry push are skipped (review 2026-10-21 06/PERF-2): central holds + // exactly what this read would send, so serving them only replays every + // lifecycle step a second time — and, after a central failover reset the + // cursor, the whole retention window. Equality (not >=) on the version is the + // safe direction: any later change, even one stamped by a clock that stepped + // backwards, no longer matches and is served. cmd.CommandText = $""" SELECT TrackedOperationId, Kind, TargetSummary, Status, RetryCount, LastError, HttpStatus, CreatedAtUtc, UpdatedAtUtc, TerminalAtUtc, SourceNode FROM OperationTracking - WHERE {predicate} + WHERE ({predicate}) + AND NOT EXISTS ( + SELECT 1 FROM OperationTrackingCentralAck a + WHERE a.TrackedOperationId = OperationTracking.TrackedOperationId + AND a.AckedUpdatedAtUtc = OperationTracking.UpdatedAtUtc) ORDER BY UpdatedAtUtc ASC, TrackedOperationId ASC LIMIT $batchSize; """; diff --git a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/Telemetry/SiteAuditTelemetryActorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/Telemetry/SiteAuditTelemetryActorTests.cs index ed737846..8d4e4c43 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/Telemetry/SiteAuditTelemetryActorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/Telemetry/SiteAuditTelemetryActorTests.cs @@ -416,6 +416,54 @@ public class SiteAuditTelemetryActorTests : TestKit Assert.Equal(expectedIds, emittedEventIds); } + /// + /// Verifies that the tracking-row versions central ACCEPTED are recorded as central acks + /// in the tracking store — and only those — so the SiteCalls reconciliation pull stops + /// re-serving them (review 2026-10-21 06/PERF-2). + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task CachedDrain_AcceptedPackets_AreRecordedAsCentralAcks() + { + var acceptedRow = NewCachedEvent(AuditKind.CachedSubmit); + var refusedRow = NewCachedEvent(AuditKind.CachedSubmit); + _queue.ReadPendingAsync(Arg.Any(), Arg.Any()) + .Returns(Task.FromResult>(Array.Empty())); + _queue.ReadPendingCachedTelemetryAsync(Arg.Any(), Arg.Any()) + .Returns( + Task.FromResult>(new[] { acceptedRow, refusedRow }), + Task.FromResult>(Array.Empty())); + + var acceptedTid = new TrackedOperationId(acceptedRow.CorrelationId!.Value); + var refusedTid = new TrackedOperationId(refusedRow.CorrelationId!.Value); + var acceptedSnapshot = NewSnapshot(acceptedTid) with + { + UpdatedAtUtc = new DateTime(2026, 5, 20, 10, 0, 7, DateTimeKind.Utc), + }; + _trackingStore.GetStatusAsync(acceptedTid, Arg.Any()) + .Returns(Task.FromResult(acceptedSnapshot)); + _trackingStore.GetStatusAsync(refusedTid, Arg.Any()) + .Returns(Task.FromResult(NewSnapshot(refusedTid))); + + _client.IngestCachedTelemetryAsync(Arg.Any(), Arg.Any()) + .Returns(_ => + { + var ack = new IngestAck(); + ack.AcceptedEventIds.Add(acceptedRow.EventId.ToString()); + return Task.FromResult(ack); + }); + + CreateActorWithCachedDrain(); + + await AwaitAssertAsync(async () => + { + await _trackingStore.Received(1).RecordCentralAcksAsync( + Arg.Is>(d => + d.Count == 1 && d[acceptedTid] == acceptedSnapshot.UpdatedAtUtc), + Arg.Any()); + }, TimeSpan.FromSeconds(5)); + } + /// Verifies that a cached row with no tracking snapshot past the grace period is abandoned while a valid row still flows. /// A task that represents the asynchronous operation. [Fact] diff --git a/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/Repositories/SiteCallAuditRepositoryTests.cs b/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/Repositories/SiteCallAuditRepositoryTests.cs index a1c32413..6d263a47 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/Repositories/SiteCallAuditRepositoryTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/Repositories/SiteCallAuditRepositoryTests.cs @@ -929,6 +929,120 @@ public class SiteCallAuditRepositoryTests : IClassFixture Assert.DoesNotContain(perNode, n => n.SourceNode is null); } + // --- UpsertManyAsync (review 2026-10-21 06/PERF-2) ---------------------- + + /// + /// Verifies that a batched upsert applies exactly the per-row monotonic semantics: fresh + /// rows insert, a higher rank advances, a lower rank is a no-op, and a page larger than + /// one command chunk is fully applied — in a handful of round trips, not one per row. + /// + /// A task that represents the asynchronous operation. + [SkippableFact] + public async Task UpsertManyAsync_AppliesMonotonicSemantics_InFewRoundTrips() + { + Skip.IfNot(_fixture.Available, _fixture.SkipReason); + var site = NewSiteId(); + var now = DateTime.UtcNow; + var advance = TrackedOperationId.New(); + var regress = TrackedOperationId.New(); + await using (var seed = CreateContext()) + { + var seedRepo = new SiteCallAuditRepository(seed); + await seedRepo.UpsertAsync(NewRow(advance, site, status: "Submitted", createdAtUtc: now)); + await seedRepo.UpsertAsync(NewRow(regress, site, status: "Delivered", createdAtUtc: now, + terminal: true, terminalAtUtc: now)); + } + + var fresh = Enumerable.Range(0, 300).Select(_ => TrackedOperationId.New()).ToList(); + var page = fresh.Select(id => NewRow(id, site, status: "Attempted", retryCount: 1, createdAtUtc: now)) + .Append(NewRow(advance, site, status: "Delivered", createdAtUtc: now, terminal: true, terminalAtUtc: now)) + .Append(NewRow(regress, site, status: "Attempted", retryCount: 9, createdAtUtc: now)) + .ToList(); + + var counter = new NonQueryCounter(); + await using (var context = CreateContext(counter)) + { + await new SiteCallAuditRepository(context).UpsertManyAsync(page); + } + + Assert.True(counter.Count <= 5, $"UpsertManyAsync issued {counter.Count} commands for {page.Count} rows"); + await using var read = CreateContext(); + var stored = await read.Set().Where(s => s.SourceSite == site).ToListAsync(); + Assert.Equal(302, stored.Count); + Assert.All(stored.Where(s => fresh.Contains(s.TrackedOperationId)), s => Assert.Equal("Attempted", s.Status)); + Assert.Equal("Delivered", stored.Single(s => s.TrackedOperationId == advance).Status); + var kept = stored.Single(s => s.TrackedOperationId == regress); + Assert.Equal("Delivered", kept.Status); + Assert.Equal(0, kept.RetryCount); + } + + /// + /// Verifies that a row another site owns still fails the batch with the ownership error, + /// while every legitimate row of the page is applied (the chunk falls back to per-row). + /// + /// A task that represents the asynchronous operation. + [SkippableFact] + public async Task UpsertManyAsync_OwnershipViolation_StillThrows_AfterApplyingTheRestOfThePage() + { + Skip.IfNot(_fixture.Available, _fixture.SkipReason); + var owner = NewSiteId(); + var attacker = NewSiteId(); + var owned = TrackedOperationId.New(); + await using (var seed = CreateContext()) + { + await new SiteCallAuditRepository(seed).UpsertAsync(NewRow(owned, owner, status: "Submitted")); + } + + var good1 = TrackedOperationId.New(); + var good2 = TrackedOperationId.New(); + var page = new List + { + NewRow(good1, attacker, status: "Submitted"), + NewRow(owned, attacker, status: "Delivered", terminal: true), + NewRow(good2, attacker, status: "Submitted"), + }; + + await using (var context = CreateContext()) + { + var failure = await Assert.ThrowsAsync( + () => new SiteCallAuditRepository(context).UpsertManyAsync(page)); + Assert.Equal(51001, failure.Number); + } + + await using var read = CreateContext(); + var repo = new SiteCallAuditRepository(read); + Assert.NotNull(await repo.GetAsync(good1)); + Assert.NotNull(await repo.GetAsync(good2)); + var stored = await repo.GetAsync(owned); + Assert.Equal(owner, stored!.SourceSite); + Assert.Equal("Submitted", stored.Status); + } + + private ScadaBridgeDbContext CreateContext(DbCommandInterceptor interceptor) + { + var options = new DbContextOptionsBuilder() + .UseSqlServer(_fixture.ConnectionString) + .AddInterceptors(interceptor) + .Options; + return new ScadaBridgeDbContext(options); + } + + /// Counts the non-query commands a repository call issues. + private sealed class NonQueryCounter : DbCommandInterceptor + { + /// The number of non-query commands issued. + public int Count { get; private set; } + + /// + public override ValueTask> NonQueryExecutingAsync( + DbCommand command, CommandEventData eventData, InterceptionResult result, + CancellationToken cancellationToken = default) + { + Count++; + return base.NonQueryExecutingAsync(command, eventData, result, cancellationToken); + } + } + // --- helpers ------------------------------------------------------------ private ScadaBridgeDbContext CreateContext() diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteCallAudit.Tests/SiteCallAuditReconciliationTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteCallAudit.Tests/SiteCallAuditReconciliationTests.cs index 8be1677d..4141187d 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.SiteCallAudit.Tests/SiteCallAuditReconciliationTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.SiteCallAudit.Tests/SiteCallAuditReconciliationTests.cs @@ -231,6 +231,9 @@ public class SiteCallAuditReconciliationTests : TestKit /// Gets the number of upsert calls received. public int UpsertCallCount { get; private set; } + /// Gets the row count of every batched (page) upsert received. + public List UpsertManyPageSizes { get; } = new(); + /// public Task UpsertAsync(SiteCall siteCall, CancellationToken ct = default) { @@ -239,6 +242,23 @@ public class SiteCallAuditReconciliationTests : TestKit return Task.CompletedTask; } + /// + public Task UpsertManyAsync(IReadOnlyList siteCalls, CancellationToken ct = default) + { + lock (UpsertManyPageSizes) + { + UpsertManyPageSizes.Add(siteCalls.Count); + } + + foreach (var siteCall in siteCalls) + { + UpsertCallCount++; + Upserted[siteCall.TrackedOperationId] = siteCall; + } + + return Task.CompletedTask; + } + /// public Task GetAsync(TrackedOperationId id, CancellationToken ct = default) => Task.FromResult(Upserted.TryGetValue(id, out var row) ? row : null); @@ -706,6 +726,148 @@ public class SiteCallAuditReconciliationTests : TestKit } } + // --------------------------------------------------------------------- + // Review 2026-10-21 06/PERF-2: a pulled page is written through ONE batched + // upsert, not one repository call per row. + // --------------------------------------------------------------------- + + /// Verifies that a reconciliation page is written through a single batched upsert. + [Fact] + public void ReconciliationTick_WritesEachPageThroughOneBatchedUpsert() + { + var siteId = "siteBatch"; + var rows = Enumerable.Range(0, 3) + .Select(i => NewRow(TrackedOperationId.New(), siteId, updatedAtUtc: DateTime.UtcNow.AddSeconds(i))) + .ToArray(); + var sites = new StaticEnumerator(new SiteEntry(siteId, "http://siteBatch:8083")); + var client = new ScriptedPullClient().Script(siteId, + new PullSiteCallsResponse(rows, MoreAvailable: false)); + var repo = new RecordingRepo(); + + var actor = CreateActor(sites, client, repo, FastTickOptions()); + try + { + AwaitAssert( + () => + { + Assert.Equal(3, repo.Upserted.Count); + lock (repo.UpsertManyPageSizes) + { + Assert.Contains(3, repo.UpsertManyPageSizes); + } + }, + duration: TimeSpan.FromSeconds(3), + interval: TimeSpan.FromMilliseconds(50)); + } + finally + { + StopTicker(actor); + } + } + + /// Enumerator whose site list a test can swap between ticks. + private sealed class MutableEnumerator : ISiteEnumerator + { + private volatile IReadOnlyList _sites; + private int _calls; + + /// Creates the enumerator with an initial site list. + /// The initial sites. + public MutableEnumerator(params SiteEntry[] sites) => _sites = sites; + + /// Gets the number of enumerations performed so far. + public int Calls => Volatile.Read(ref _calls); + + /// Replaces the site list returned from the next enumeration on. + /// The new sites. + public void Set(params SiteEntry[] sites) => _sites = sites; + + /// + public Task> EnumerateAsync(CancellationToken ct = default) + { + Interlocked.Increment(ref _calls); + return Task.FromResult(_sites); + } + } + + /// + /// Verifies that a site the enumerator stops returning has its cursor and pinned latch + /// pruned: the pinned site publishes Pinned=false, and when it reappears its pull restarts + /// from (lane H follow-up to review 2026-10-21). + /// + [Fact] + public void ReconciliationTick_RemovedSite_PrunesCursorAndPinnedLatch() + { + var siteId = "siteGone"; + var ts = new DateTime(2026, 5, 20, 10, 0, 0, DateTimeKind.Utc); + var rows = new[] + { + NewRow(TrackedOperationId.New(), siteId, updatedAtUtc: ts), + NewRow(TrackedOperationId.New(), siteId, updatedAtUtc: ts), + }; + var entry = new SiteEntry(siteId, "http://siteGone:8083"); + var sites = new MutableEnumerator(entry); + var client = new LegacyIgnoresAfterIdPullClient(rows); + var recorder = new RecordingPullClient(client); + var repo = new RecordingRepo(); + // A dedicated probe, not TestActor: the re-added legacy site pins again, and that + // event must not land in StopTicker's ExpectTerminated. + var pinnedProbe = CreateTestProbe(); + Sys.EventStream.Subscribe(pinnedProbe.Ref, typeof(SiteCallReconciliationPinnedChanged)); + + var actor = CreateActor(sites, recorder, repo, new SiteCallAuditOptions + { + ReconciliationIntervalOverride = TimeSpan.FromMilliseconds(100), + ReconciliationBatchSize = 2, + }); + try + { + Assert.True(pinnedProbe.ExpectMsg(TimeSpan.FromSeconds(5)).Pinned); + + sites.Set(); + var unpinned = pinnedProbe.ExpectMsg(TimeSpan.FromSeconds(5)); + Assert.Equal(siteId, unpinned.SiteId); + Assert.False(unpinned.Pinned); + + var callsBeforeReturn = recorder.Calls.Count; + sites.Set(entry); + AwaitAssert( + () => + { + var later = recorder.Calls.Skip(callsBeforeReturn).ToList(); + Assert.NotEmpty(later); + Assert.Equal(DateTime.MinValue, later[0].SinceUtc); + }, + duration: TimeSpan.FromSeconds(3), + interval: TimeSpan.FromMilliseconds(50)); + } + finally + { + Sys.EventStream.Unsubscribe(pinnedProbe.Ref); + StopTicker(actor); + } + } + + /// Wraps a pull client and records the cursor of every pull, thread-safely. + private sealed class RecordingPullClient(IPullSiteCallsClient inner) : IPullSiteCallsClient + { + private readonly List<(DateTime SinceUtc, string? AfterId)> _calls = new(); + + /// Gets a snapshot of every pull cursor recorded so far. + public IReadOnlyList<(DateTime SinceUtc, string? AfterId)> Calls + { + get { lock (_calls) { return _calls.ToList(); } } + } + + /// + public Task PullAsync( + string siteId, DateTime sinceUtc, string? afterId, int batchSize, CancellationToken ct) + { + lock (_calls) { _calls.Add((sinceUtc, afterId)); } + return inner.PullAsync(siteId, sinceUtc, afterId, batchSize, ct); + } + } + // --------------------------------------------------------------------- // 9. The reconciliation drain runs OFF the mailbox, so ingest, // query and KPI messages are answered while a long post-outage diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteCallAudit.Tests/TestSupport/StubRepositoryProvider.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteCallAudit.Tests/TestSupport/StubRepositoryProvider.cs index 3da3b70b..2df6d72b 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.SiteCallAudit.Tests/TestSupport/StubRepositoryProvider.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.SiteCallAudit.Tests/TestSupport/StubRepositoryProvider.cs @@ -116,6 +116,10 @@ internal sealed class SerializingSiteCallAuditRepository : ISiteCallAuditReposit public Task UpsertAsync(SiteCall siteCall, CancellationToken ct = default) => RunAsync(() => _inner.UpsertAsync(siteCall, ct)); + /// + public Task UpsertManyAsync(IReadOnlyList siteCalls, CancellationToken ct = default) => + RunAsync(() => _inner.UpsertManyAsync(siteCalls, ct)); + /// public Task GetAsync(TrackedOperationId id, CancellationToken ct = default) => RunAsync(() => _inner.GetAsync(id, ct)); diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Tracking/OperationTrackingStoreTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Tracking/OperationTrackingStoreTests.cs index dca2dbfc..68389a04 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Tracking/OperationTrackingStoreTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Tracking/OperationTrackingStoreTests.cs @@ -1057,4 +1057,124 @@ public class OperationTrackingStoreTests : IDisposable Assert.Equal(System.Data.ConnectionState.Closed, connection.State); } + + // ── Central-ack filter for the reconciliation pull (review 2026-10-21 06/PERF-2) ── + + /// + /// Verifies that a row whose current version central has acknowledged through the + /// cached-telemetry push is no longer served by the reconciliation read, while + /// unacknowledged rows still are. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task ReadChangedSince_SkipsRowsCentralAckedAtTheirCurrentVersion() + { + var (store, _) = CreateStore(nameof(ReadChangedSince_SkipsRowsCentralAckedAtTheirCurrentVersion)); + await using var _store = store; + var acked = TrackedOperationId.New(); + var unacked = TrackedOperationId.New(); + await store.RecordEnqueueAsync(acked, nameof(AuditKind.ApiCallCached), "A", null, null, "node-a"); + await store.RecordEnqueueAsync(unacked, nameof(AuditKind.ApiCallCached), "B", null, null, "node-a"); + var snapshot = await store.GetStatusAsync(acked); + + await store.RecordCentralAcksAsync(new Dictionary + { + [acked] = snapshot!.UpdatedAtUtc, + }); + + var rows = await store.ReadChangedSinceAsync(DateTime.MinValue, batchSize: 100); + Assert.Equal(new[] { unacked }, rows.Select(r => r.TrackedOperationId).ToArray()); + } + + /// + /// Verifies that a row changed AFTER central acknowledged it is served again — the ack + /// covers one version, not the operation. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task ReadChangedSince_ServesRowChangedAfterItsAck() + { + var (store, _) = CreateStore(nameof(ReadChangedSince_ServesRowChangedAfterItsAck)); + await using var _store = store; + var id = TrackedOperationId.New(); + await store.RecordEnqueueAsync(id, nameof(AuditKind.ApiCallCached), "A", null, null, "node-a"); + var submitted = await store.GetStatusAsync(id); + await store.RecordCentralAcksAsync(new Dictionary + { + [id] = submitted!.UpdatedAtUtc, + }); + await Task.Delay(5); // a strictly later UpdatedAtUtc stamp + await store.RecordTerminalAsync(id, nameof(AuditStatus.Delivered), null, 200); + + var rows = await store.ReadChangedSinceAsync(DateTime.MinValue, batchSize: 100); + var row = Assert.Single(rows); + Assert.Equal(nameof(AuditStatus.Delivered), row.Status); + } + + /// + /// Verifies that an older ack never rolls back a newer one (acks are monotonic per row), + /// and that the ack ledger is pruned with the tracking rows it describes. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task CentralAcks_AreMonotonic_AndPrunedWithTheirTrackingRow() + { + var (store, dataSource) = CreateStore(nameof(CentralAcks_AreMonotonic_AndPrunedWithTheirTrackingRow)); + await using var _store = store; + var id = TrackedOperationId.New(); + await store.RecordEnqueueAsync(id, nameof(AuditKind.ApiCallCached), "A", null, null, "node-a"); + await store.RecordTerminalAsync(id, nameof(AuditStatus.Delivered), null, 200); + var terminal = await store.GetStatusAsync(id); + + await store.RecordCentralAcksAsync(new Dictionary { [id] = terminal!.UpdatedAtUtc }); + await store.RecordCentralAcksAsync(new Dictionary { [id] = terminal.CreatedAtUtc.AddSeconds(-1) }); + Assert.Empty(await store.ReadChangedSinceAsync(DateTime.MinValue, batchSize: 100)); + + // Age the terminal row past retention and purge: the ack row goes with it. + await store.FlushAsync(); + using (var connection = OpenVerifierConnection(dataSource)) + using (var cmd = connection.CreateCommand()) + { + cmd.CommandText = "UPDATE OperationTracking SET TerminalAtUtc = $old;"; + cmd.Parameters.AddWithValue("$old", DateTime.UtcNow.AddDays(-30).ToString("o")); + cmd.ExecuteNonQuery(); + } + + await store.PurgeTerminalAsync(DateTime.UtcNow.AddDays(-7)); + + using (var connection = OpenVerifierConnection(dataSource)) + using (var cmd = connection.CreateCommand()) + { + cmd.CommandText = "SELECT COUNT(*) FROM OperationTrackingCentralAck;"; + Assert.Equal(0L, (long)cmd.ExecuteScalar()!); + } + } + + /// + /// Verifies the ack ledger is its own table, not a column on the LocalDb-replicated + /// OperationTracking table (whose capture triggers are built from its column list). + /// Being node-local, a standby that takes over has no ack rows and serves every row once. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task CentralAckLedger_IsADistinctNodeLocalTable_NotAColumnOnTheReplicatedTable() + { + var (store, dataSource) = CreateStore(nameof(CentralAckLedger_IsADistinctNodeLocalTable_NotAColumnOnTheReplicatedTable)); + await using var _store = store; + await store.FlushAsync(); + + using var connection = OpenVerifierConnection(dataSource); + using var cmd = connection.CreateCommand(); + cmd.CommandText = "PRAGMA table_info(OperationTracking);"; + var columns = new List(); + using (var reader = cmd.ExecuteReader()) + { + while (reader.Read()) columns.Add(reader.GetString(1)); + } + + Assert.DoesNotContain(columns, c => c.StartsWith("Acked", StringComparison.Ordinal) + || c.Contains("CentralAck", StringComparison.Ordinal)); + cmd.CommandText = "SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = 'OperationTrackingCentralAck';"; + Assert.Equal(1L, (long)cmd.ExecuteScalar()!); + } }