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)

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<Guid>(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<Guid, (TrackedOperationId Id, DateTime UpdatedAtUtc)>(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
         }
     }
 
+    /// <summary>
+    /// 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 <c>SiteCalls</c> 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.
+    /// </summary>
+    private async Task RecordCentralAcksAsync(
+        IReadOnlyList<Guid> acceptedIds,
+        Dictionary<Guid, (TrackedOperationId Id, DateTime UpdatedAtUtc)> packetVersions,
+        CancellationToken ct)
+    {
+        var versions = new Dictionary<TrackedOperationId, DateTime>();
+        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);
+        }
+    }
+
     /// <summary>
     /// Moves cached rows whose operational half can never be built out of the
     /// cached-drain queue, as <see cref="SiteAuditForwardStates.PushAbandoned"/>.

src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/ISiteCallAuditRepository.cs

@@ -42,6 +42,27 @@ public interface ISiteCallAuditRepository
     /// <returns>A task that represents the asynchronous operation.</returns>
     Task UpsertAsync(SiteCall siteCall, CancellationToken ct = default);
 
+    /// <summary>
+    /// Applies <see cref="UpsertAsync"/>'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 <see cref="UpsertAsync"/>
+    /// 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.
+    /// </summary>
+    /// <param name="siteCalls">The rows to insert or monotonically update.</param>
+    /// <param name="ct">Cancellation token.</param>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    async Task UpsertManyAsync(IReadOnlyList<SiteCall> siteCalls, CancellationToken ct = default)
+    {
+        ArgumentNullException.ThrowIfNull(siteCalls);
+        foreach (var siteCall in siteCalls)
+        {
+            await UpsertAsync(siteCall, ct).ConfigureAwait(false);
+        }
+    }
+
     /// <summary>
     /// Returns the row for the given id, or <c>null</c> if none exists.
     /// </summary>

src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Services/IOperationTrackingStore.cs

@@ -150,6 +150,15 @@ public interface IOperationTrackingStore
     /// deterministic — <c>UpdatedAtUtc ASC, TrackedOperationId ASC</c>.
     /// </para>
     /// <para>
+    /// <b>Only rows central has not already acknowledged.</b> A row whose
+    /// current version (<c>UpdatedAtUtc</c>) central acknowledged through the
+    /// cached-telemetry push — recorded via <see cref="RecordCentralAcksAsync"/> — 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.
+    /// </para>
+    /// <para>
     /// <see cref="SiteCallOperational.SourceSite"/> is left as the empty string:
     /// the site id is not a tracking-store column, and the central client re-stamps
     /// it from the <c>siteId</c> it dialed (the only authority that knows which
@@ -174,4 +183,24 @@ public interface IOperationTrackingStore
         int batchSize,
         string? afterId = null,
         CancellationToken ct = default);
+
+    /// <summary>
+    /// Records that central accepted the cached-telemetry push of each operation at the
+    /// given tracking-row version (the snapshot's <c>UpdatedAtUtc</c> the push carried),
+    /// so <see cref="ReadChangedSinceAsync"/> 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.
+    /// </summary>
+    /// <remarks>
+    /// 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).
+    /// </remarks>
+    /// <param name="ackedVersions">Operation id → the <c>UpdatedAtUtc</c> of the snapshot central accepted.</param>
+    /// <param name="ct">Cancellation token.</param>
+    /// <returns>A task that completes once the acks are queued.</returns>
+    Task RecordCentralAcksAsync(
+        IReadOnlyDictionary<TrackedOperationId, DateTime> ackedVersions,
+        CancellationToken ct = default) => Task.CompletedTask;
 }

src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/SiteCallAuditRepository.cs

@@ -169,6 +169,84 @@ public class SiteCallAuditRepository : ISiteCallAuditRepository
         }
     }
 
+    /// <summary>
+    /// 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.
+    /// </summary>
+    internal const int UpsertManyChunkSize = 128;
+
+    /// <inheritdoc />
+    public async Task UpsertManyAsync(IReadOnlyList<SiteCall> 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<object>(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();
+    }
+
+    /// <summary>Renames every <c>@Name</c> parameter reference in <paramref name="sql"/> to <c>@Name{suffix}</c>.</summary>
+    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 = @"

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);
     }
 
+    /// <summary>
+    /// 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
+    /// <c>Pinned=false</c> 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
+    /// <c>DateTime.MinValue</c>, which the idempotent upsert absorbs.
+    /// </summary>
+    private void PruneRemovedSites(IReadOnlyList<SiteEntry> sites)
+    {
+        var live = new HashSet<string>(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);
+        }
+    }
+
     /// <summary>
     /// 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.

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();
+    }
+
+    /// <summary>
+    /// 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 <c>OperationTracking</c>: 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
+    /// <c>mirrored_secrets</c>, it is not in <c>SiteLocalDbSetup.ReplicatedTables</c>.
+    /// <para>
+    /// 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.
+    /// </para>
+    /// </summary>
+    private const string CentralAckSchemaSql = """
+        CREATE TABLE IF NOT EXISTS OperationTrackingCentralAck (
+            TrackedOperationId TEXT NOT NULL PRIMARY KEY,
+            AckedUpdatedAtUtc  TEXT NOT NULL
+        );
+        """;
 
     /// <inheritdoc/>
     public async Task RecordEnqueueAsync(
@@ -271,6 +298,51 @@ public class OperationTrackingStore : IOperationTrackingStore, IAsyncDisposable,
         }, ct).ConfigureAwait(false);
     }
 
+    /// <inheritdoc/>
+    public async Task RecordCentralAcksAsync(
+        IReadOnlyDictionary<TrackedOperationId, DateTime> 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 ─────────────────────────────────────────────────────
 
     /// <summary>
@@ -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;
             """;