fix(site-runtime): batch native-alarm deletes into the coalesced flush transaction; docs for scripts/cache/scheduler (review 2026-10-21 01/SR-P2)

src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/NativeAlarmActor.cs

@@ -60,10 +60,12 @@ public class NativeAlarmActor : ReceiveActor
     /// source reference keeps only its LATEST transition, flushed as one batched SQLite write on
     /// a short single-shot timer rather than a write per transition. A crash loses at most one
     /// flush interval of mirror writes; the (re)subscribe snapshot swap reconciles the mirror, so
-    /// the durability trade is safe. A <see cref="PersistDelete"/> removes the ref here so a
-    /// deleted condition is not resurrected by a still-pending upsert.
+    /// the durability trade is safe. A <see cref="PersistDelete"/> records a <c>null</c> entry for
+    /// the ref (upsert/delete discriminator), replacing any pending upsert so a deleted condition
+    /// is not resurrected; a later upsert for the same ref replaces the delete. Both kinds are
+    /// applied in the one batched flush transaction (review 2026-10-21 01/SR-P2).
     /// </summary>
-    private readonly Dictionary<string, NativeAlarmTransition> _dirtyUpserts = new();
+    private readonly Dictionary<string, NativeAlarmTransition?> _dirtyUpserts = new();
 
     private ICancelable? _retryTimer;
     private ICancelable? _flushTimer;
@@ -556,6 +558,11 @@ public class NativeAlarmActor : ReceiveActor
     private void MarkDirtyUpsert(NativeAlarmTransition t)
     {
         _dirtyUpserts[t.SourceReference] = t;
+        ArmFlushTimer();
+    }
+
+    private void ArmFlushTimer()
+    {
         if (_flushTimer == null)
         {
             _flushTimer = Context.System.Scheduler.ScheduleTellOnceCancelable(
@@ -575,6 +582,8 @@ public class NativeAlarmActor : ReceiveActor
             return;
 
         var batch = _dirtyUpserts.Values
+            .Where(t => t is not null)
+            .Select(t => t!)
             .Select(t => (
                 t.SourceReference,
                 JsonSerializer.Serialize(t.Condition),
@@ -582,26 +591,24 @@ public class NativeAlarmActor : ReceiveActor
                     t.AlarmTypeName, t.Category, t.Message, t.CurrentValue, t.LimitValue, t.AckTime)),
                 t.TransitionTime))
             .ToList();
+        var deletes = _dirtyUpserts.Where(kv => kv.Value is null).Select(kv => kv.Key).ToList();
         _dirtyUpserts.Clear();
 
-        _storage.UpsertNativeAlarmsAsync(_instanceName, _source.CanonicalName, batch)
+        _storage.ApplyNativeAlarmChangesAsync(_instanceName, _source.CanonicalName, batch, deletes)
             .ContinueWith(
                 task => _logger.LogWarning(task.Exception,
-                    "Failed to persist {Count} native alarms on {Instance}", batch.Count, _instanceName),
+                    "Failed to persist {Count} native alarm changes on {Instance}",
+                    batch.Count + deletes.Count, _instanceName),
                 TaskContinuationOptions.OnlyOnFaulted);
     }
 
     private void PersistDelete(string sourceReference)
     {
-        // Drop any pending coalesced upsert for this ref so the delete is not resurrected by a
-        // still-buffered upsert on the next flush (P4).
-        _dirtyUpserts.Remove(sourceReference);
-
-        _storage.DeleteNativeAlarmAsync(_instanceName, _source.CanonicalName, sourceReference)
-            .ContinueWith(
-                task => _logger.LogWarning(task.Exception,
-                    "Failed to delete native alarm {Ref} on {Instance}", sourceReference, _instanceName),
-                TaskContinuationOptions.OnlyOnFaulted);
+        // Queue the delete in the same dirty set as upserts (a null entry replaces any pending
+        // upsert, so the delete is not resurrected) and let the coalesced flush apply it in the
+        // batched transaction — one commit for N dropped conditions instead of N.
+        _dirtyUpserts[sourceReference] = null;
+        ArmFlushTimer();
     }
 
     // ── Internal messages ──

src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Persistence/SiteStorageService.cs

@@ -611,6 +611,84 @@ public class SiteStorageService
         await command.ExecuteNonQueryAsync();
     }
 
+    /// <summary>
+    /// Applies a coalesced set of native-alarm mirror changes for one instance/source — upserts AND
+    /// deletes — in a SINGLE connection + transaction (review 2026-10-21 01/SR-P2). Deletes used
+    /// to be one connection + autocommit (one WAL fsync + one CDC row) per dropped condition; a
+    /// snapshot swap dropping N stale conditions now costs one commit. The caller guarantees each
+    /// source reference appears at most once across <paramref name="upserts"/> and
+    /// <paramref name="deletes"/> (the actor coalesces per reference), so statement order is irrelevant.
+    /// </summary>
+    /// <param name="instanceName">Unique name of the instance owning the alarms.</param>
+    /// <param name="sourceCanonicalName">Canonical name of the source binding.</param>
+    /// <param name="upserts">The conditions to upsert.</param>
+    /// <param name="deletes">Source references whose rows are removed.</param>
+    /// <returns>A task that completes when the whole batch has been committed.</returns>
+    public async Task ApplyNativeAlarmChangesAsync(
+        string instanceName, string sourceCanonicalName,
+        IReadOnlyList<(string SourceReference, string ConditionJson, string? MetadataJson, DateTimeOffset LastTransitionAt)> upserts,
+        IReadOnlyList<string> deletes)
+    {
+        if (upserts.Count == 0 && deletes.Count == 0)
+            return;
+
+        await using var connection = OpenConnection();
+        await using var transaction = (SqliteTransaction)await connection.BeginTransactionAsync();
+
+        if (deletes.Count > 0)
+        {
+            await using var del = connection.CreateCommand();
+            del.Transaction = transaction;
+            del.CommandText = @"
+                DELETE FROM native_alarm_state
+                WHERE instance_unique_name = @name
+                  AND source_canonical_name = @source
+                  AND source_reference = @ref";
+            del.Parameters.AddWithValue("@name", instanceName);
+            del.Parameters.AddWithValue("@source", sourceCanonicalName);
+            var dRef = del.Parameters.Add("@ref", SqliteType.Text);
+            foreach (var reference in deletes)
+            {
+                dRef.Value = reference;
+                await del.ExecuteNonQueryAsync();
+            }
+        }
+
+        if (upserts.Count > 0)
+        {
+            await using var command = connection.CreateCommand();
+            command.Transaction = transaction;
+            command.CommandText = @"
+                INSERT INTO native_alarm_state
+                    (instance_unique_name, source_canonical_name, source_reference, condition_json, last_transition_at, metadata_json)
+                VALUES (@name, @source, @ref, @json, @at, @meta)
+                ON CONFLICT(instance_unique_name, source_canonical_name, source_reference) DO UPDATE SET
+                    condition_json     = excluded.condition_json,
+                    last_transition_at = excluded.last_transition_at,
+                    metadata_json      = excluded.metadata_json";
+
+            var pName = command.Parameters.Add("@name", SqliteType.Text);
+            var pSource = command.Parameters.Add("@source", SqliteType.Text);
+            var pRef = command.Parameters.Add("@ref", SqliteType.Text);
+            var pJson = command.Parameters.Add("@json", SqliteType.Text);
+            var pAt = command.Parameters.Add("@at", SqliteType.Text);
+            var pMeta = command.Parameters.Add("@meta", SqliteType.Text);
+            pName.Value = instanceName;
+            pSource.Value = sourceCanonicalName;
+
+            foreach (var row in upserts)
+            {
+                pRef.Value = row.SourceReference;
+                pJson.Value = row.ConditionJson;
+                pAt.Value = row.LastTransitionAt.ToString("O");
+                pMeta.Value = (object?)row.MetadataJson ?? DBNull.Value;
+                await command.ExecuteNonQueryAsync();
+            }
+        }
+
+        await transaction.CommitAsync();
+    }
+
     /// <summary>
     /// Returns all mirrored native alarm conditions for an instance's source binding,
     /// used to rehydrate a NativeAlarmActor on (re)start.