perf(site-runtime): coalesce native-alarm mirror persistence into batched flushes (P4)

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

@@ -53,7 +53,21 @@ public class NativeAlarmActor : ReceiveActor
     /// <summary>Buffer accumulating a snapshot replay until <see cref="AlarmTransitionKind.SnapshotComplete"/>.</summary>
     private readonly Dictionary<string, NativeAlarmTransition> _snapshotBuffer = new();
 
+    /// <summary>
+    /// Pending mirror upserts coalesced by source reference (P4): under an alarm storm each
+    /// 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.
+    /// </summary>
+    private readonly Dictionary<string, NativeAlarmTransition> _dirtyUpserts = new();
+
     private ICancelable? _retryTimer;
+    private ICancelable? _flushTimer;
+
+    /// <summary>How long transitions coalesce before a batched persist flush (P4).</summary>
+    private readonly TimeSpan _persistFlushInterval;
 
     /// <summary>Initializes a new <see cref="NativeAlarmActor"/> for a single resolved source binding.</summary>
     /// <param name="source">The resolved native alarm source (connection + source reference + filter).</param>
@@ -68,6 +82,8 @@ public class NativeAlarmActor : ReceiveActor
     /// <param name="serviceProvider">Optional DI service provider used to resolve the optional
     /// <see cref="ISiteEventLogger"/> for <c>alarm</c> operational events. Fire-and-forget;
     /// a logging failure never affects the mirror.</param>
+    /// <param name="persistFlushInterval">How long native-alarm transitions coalesce before a
+    /// batched SQLite persist flush (P4). Defaults to 100 ms.</param>
     public NativeAlarmActor(
         ResolvedNativeAlarmSource source,
         string instanceName,
@@ -77,7 +93,8 @@ public class NativeAlarmActor : ReceiveActor
         SiteRuntimeOptions options,
         ILogger logger,
         AlarmKind nativeKind = AlarmKind.NativeOpcUa,
-        IServiceProvider? serviceProvider = null)
+        IServiceProvider? serviceProvider = null,
+        TimeSpan? persistFlushInterval = null)
     {
         _source = source;
         _instanceName = instanceName;
@@ -88,12 +105,14 @@ public class NativeAlarmActor : ReceiveActor
         _logger = logger;
         _nativeKind = nativeKind;
         _serviceProvider = serviceProvider;
+        _persistFlushInterval = persistFlushInterval ?? TimeSpan.FromMilliseconds(100);
 
         Receive<RehydrationCompleted>(HandleRehydration);
         Receive<NativeAlarmTransitionUpdate>(HandleTransition);
         Receive<NativeAlarmSourceUnavailable>(HandleSourceUnavailable);
         Receive<SubscribeAlarmsResponse>(HandleSubscribeResponse);
         Receive<RetrySubscribe>(_ => SendSubscribe());
+        Receive<FlushPersistence>(_ => FlushDirtyUpserts());
     }
 
     /// <inheritdoc />
@@ -117,6 +136,11 @@ public class NativeAlarmActor : ReceiveActor
     protected override void PostStop()
     {
         _retryTimer?.Cancel();
+        _flushTimer?.Cancel();
+        // Best-effort final flush of any coalesced upserts so a graceful stop does not lose the
+        // last ≤ flush-interval of mirror writes (P4). Fire-and-forget — a failed final write is
+        // reconciled by the next (re)subscribe snapshot swap.
+        FlushDirtyUpserts();
         base.PostStop();
     }
 
@@ -225,7 +249,7 @@ public class NativeAlarmActor : ReceiveActor
         foreach (var (sourceRef, t) in _snapshotBuffer)
         {
             _alarms[sourceRef] = t;
-            PersistUpsert(t);
+            MarkDirtyUpsert(t);
             // A snapshot replay is a re-sync of the source's current
             // active set on (re)subscribe, NOT a live transition — surface it
             // upward for the DebugView but do NOT re-log an `alarm` operational
@@ -249,7 +273,7 @@ public class NativeAlarmActor : ReceiveActor
         }
 
         _alarms[t.SourceReference] = t;
-        PersistUpsert(t);
+        MarkDirtyUpsert(t);
         Emit(t, t.Condition);
 
         // Retention: a resolved condition (inactive AND acknowledged) drops out of
@@ -428,18 +452,50 @@ public class NativeAlarmActor : ReceiveActor
             "alarm", severity, _instanceName, $"NativeAlarmActor:{_source.CanonicalName}", message);
     }
 
-    private void PersistUpsert(NativeAlarmTransition t)
+    /// <summary>
+    /// Records the latest transition per source reference and arms the coalesced flush timer (P4).
+    /// The actual SQLite write is deferred to <see cref="FlushDirtyUpserts"/> so an alarm storm
+    /// collapses into one batched write instead of a write per transition.
+    /// </summary>
+    private void MarkDirtyUpsert(NativeAlarmTransition t)
     {
-        var json = JsonSerializer.Serialize(t.Condition);
+        _dirtyUpserts[t.SourceReference] = t;
-        _storage.UpsertNativeAlarmAsync(_instanceName, _source.CanonicalName, t.SourceReference, json, t.TransitionTime)
+        if (_flushTimer == null)
+        {
+            _flushTimer = Context.System.Scheduler.ScheduleTellOnceCancelable(
+                _persistFlushInterval, Self, FlushPersistence.Instance, Self);
+        }
+    }
+
+    /// <summary>
+    /// Persists all coalesced upserts in one batched transaction (P4), then disarms the flush
+    /// timer. Fire-and-forget with OnlyOnFaulted logging — a failed write never blocks the actor;
+    /// the mirror reconciles on the next (re)subscribe snapshot.
+    /// </summary>
+    private void FlushDirtyUpserts()
+    {
+        _flushTimer = null;
+        if (_dirtyUpserts.Count == 0)
+            return;
+
+        var batch = _dirtyUpserts.Values
+            .Select(t => (t.SourceReference, JsonSerializer.Serialize(t.Condition), t.TransitionTime))
+            .ToList();
+        _dirtyUpserts.Clear();
+
+        _storage.UpsertNativeAlarmsAsync(_instanceName, _source.CanonicalName, batch)
             .ContinueWith(
                 task => _logger.LogWarning(task.Exception,
-                    "Failed to persist native alarm {Ref} on {Instance}", t.SourceReference, _instanceName),
+                    "Failed to persist {Count} native alarms on {Instance}", batch.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,
@@ -456,4 +512,11 @@ public class NativeAlarmActor : ReceiveActor
         public static readonly RetrySubscribe Instance = new();
         private RetrySubscribe() { }
     }
+
+    /// <summary>Self-tell that fires the coalesced native-alarm persist flush (P4).</summary>
+    private sealed class FlushPersistence
+    {
+        public static readonly FlushPersistence Instance = new();
+        private FlushPersistence() { }
+    }
 }

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

@@ -506,6 +506,56 @@ public class SiteStorageService
         await command.ExecuteNonQueryAsync();
     }
 
+    /// <summary>
+    /// Batch counterpart of <see cref="UpsertNativeAlarmAsync"/> (P4): upserts a set of mirrored
+    /// native alarm conditions for one instance/source in a SINGLE connection + transaction with
+    /// one prepared command re-bound per row. Used by the NativeAlarmActor's coalesced flush so an
+    /// alarm storm collapses into one batched write instead of a write per transition.
+    /// </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="rows">The conditions to upsert (source reference, serialized condition, transition time).</param>
+    /// <returns>A task that completes when all rows have been committed.</returns>
+    public async Task UpsertNativeAlarmsAsync(
+        string instanceName, string sourceCanonicalName,
+        IReadOnlyList<(string SourceReference, string ConditionJson, DateTimeOffset LastTransitionAt)> rows)
+    {
+        if (rows.Count == 0)
+            return;
+
+        await using var connection = new SqliteConnection(_connectionString);
+        await connection.OpenAsync();
+        await using var transaction = (SqliteTransaction)await connection.BeginTransactionAsync();
+
+        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)
+            VALUES (@name, @source, @ref, @json, @at)
+            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";
+
+        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);
+        pName.Value = instanceName;
+        pSource.Value = sourceCanonicalName;
+
+        foreach (var row in rows)
+        {
+            pRef.Value = row.SourceReference;
+            pJson.Value = row.ConditionJson;
+            pAt.Value = row.LastTransitionAt.ToString("O");
+            await command.ExecuteNonQueryAsync();
+        }
+
+        await transaction.CommitAsync();
+    }
+
     /// <summary>
     /// Removes a single mirrored native alarm condition (e.g. a return-to-normal that drops out of retention).
     /// </summary>

tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/NativeAlarmActorTests.cs

@@ -150,6 +150,40 @@ public class NativeAlarmActorTests : TestKit, IDisposable
         instance.ExpectNoMsg(TimeSpan.FromMilliseconds(300));
     }
 
+    // ── PLAN-03 Task 20 (P4): coalesced batched persistence ─────────────────
+
+    private IActorRef SpawnWithFlush(IActorRef instanceActor, IActorRef dclManager, TimeSpan flushInterval) =>
+        ActorOf(Props.Create(() => new NativeAlarmActor(
+            Source(), "inst", instanceActor, dclManager, _storage, _options,
+            NullLogger<NativeAlarmActor>.Instance, AlarmKind.NativeOpcUa, null, flushInterval)));
+
+    private static void DeliverLiveTransition(IActorRef actor, string sourceRef, int severity, DateTimeOffset time) =>
+        actor.Tell(new NativeAlarmTransitionUpdate("Opc", Transition(
+            sourceRef, AlarmTransitionKind.Raise,
+            new AlarmConditionState(true, false, null, AlarmShelveState.Unshelved, false, severity), time)));
+
+    [Fact]
+    public async Task AlarmStorm_CoalescesPersistence_LatestPerSourceRefWins()
+    {
+        var instance = CreateTestProbe();
+        var dcl = CreateTestProbe();
+        var actor = SpawnWithFlush(instance.Ref, dcl.Ref, TimeSpan.FromMilliseconds(100));
+        dcl.ExpectMsg<SubscribeAlarmsRequest>();
+
+        // 50 live transitions across 5 source refs (strictly increasing time per ref so none
+        // is dropped as stale) — the coalesced flush must collapse them to one row per ref.
+        var baseTime = DateTimeOffset.UtcNow;
+        for (var i = 0; i < 50; i++)
+            DeliverLiveTransition(actor, $"ref-{i % 5}", severity: i, baseTime.AddMilliseconds(i));
+
+        await Task.Delay(500); // > flush interval
+
+        var rows = await _storage.GetNativeAlarmsAsync("inst", Source().CanonicalName);
+        Assert.Equal(5, rows.Count);
+        // The LAST write for ref-4 carries severity 49 (i = 49, 49 % 5 == 4).
+        Assert.Contains(rows, r => r.ConditionJson.Contains("49"));
+    }
+
     // ── M1.5: site event log `alarm` category ──────────────────────────────
 
     [Fact]