diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/NativeAlarmActor.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/NativeAlarmActor.cs
index 526e6757..dfa42102 100644
--- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/NativeAlarmActor.cs
+++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/NativeAlarmActor.cs
@@ -53,7 +53,21 @@ public class NativeAlarmActor : ReceiveActor
/// Buffer accumulating a snapshot replay until .
private readonly Dictionary _snapshotBuffer = new();
+ ///
+ /// 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 removes the ref here so a
+ /// deleted condition is not resurrected by a still-pending upsert.
+ ///
+ private readonly Dictionary _dirtyUpserts = new();
+
private ICancelable? _retryTimer;
+ private ICancelable? _flushTimer;
+
+ /// How long transitions coalesce before a batched persist flush (P4).
+ private readonly TimeSpan _persistFlushInterval;
/// Initializes a new for a single resolved source binding.
/// The resolved native alarm source (connection + source reference + filter).
@@ -68,6 +82,8 @@ public class NativeAlarmActor : ReceiveActor
/// Optional DI service provider used to resolve the optional
/// for alarm operational events. Fire-and-forget;
/// a logging failure never affects the mirror.
+ /// How long native-alarm transitions coalesce before a
+ /// batched SQLite persist flush (P4). Defaults to 100 ms.
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(HandleRehydration);
Receive(HandleTransition);
Receive(HandleSourceUnavailable);
Receive(HandleSubscribeResponse);
Receive(_ => SendSubscribe());
+ Receive(_ => FlushDirtyUpserts());
}
///
@@ -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)
+ ///
+ /// Records the latest transition per source reference and arms the coalesced flush timer (P4).
+ /// The actual SQLite write is deferred to so an alarm storm
+ /// collapses into one batched write instead of a write per transition.
+ ///
+ private void MarkDirtyUpsert(NativeAlarmTransition t)
{
- var json = JsonSerializer.Serialize(t.Condition);
- _storage.UpsertNativeAlarmAsync(_instanceName, _source.CanonicalName, t.SourceReference, json, t.TransitionTime)
+ _dirtyUpserts[t.SourceReference] = t;
+ if (_flushTimer == null)
+ {
+ _flushTimer = Context.System.Scheduler.ScheduleTellOnceCancelable(
+ _persistFlushInterval, Self, FlushPersistence.Instance, Self);
+ }
+ }
+
+ ///
+ /// 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.
+ ///
+ 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() { }
}
+
+ /// Self-tell that fires the coalesced native-alarm persist flush (P4).
+ private sealed class FlushPersistence
+ {
+ public static readonly FlushPersistence Instance = new();
+ private FlushPersistence() { }
+ }
}
diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Persistence/SiteStorageService.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Persistence/SiteStorageService.cs
index 7391a057..1a31deae 100644
--- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Persistence/SiteStorageService.cs
+++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Persistence/SiteStorageService.cs
@@ -506,6 +506,56 @@ public class SiteStorageService
await command.ExecuteNonQueryAsync();
}
+ ///
+ /// Batch counterpart of (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.
+ ///
+ /// Unique name of the instance owning the alarms.
+ /// Canonical name of the source binding.
+ /// The conditions to upsert (source reference, serialized condition, transition time).
+ /// A task that completes when all rows have been committed.
+ 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();
+ }
+
///
/// Removes a single mirrored native alarm condition (e.g. a return-to-normal that drops out of retention).
///
diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/NativeAlarmActorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/NativeAlarmActorTests.cs
index 3fe305a9..4bd5512e 100644
--- a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/NativeAlarmActorTests.cs
+++ b/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.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();
+
+ // 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]