From 326d945d8546a0b1b69ae243401239d232ac5b92 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 9 Jul 2026 01:05:20 -0400 Subject: [PATCH] perf(site-runtime): coalesce native-alarm mirror persistence into batched flushes (P4) --- .../Actors/NativeAlarmActor.cs | 77 +++++++++++++++++-- .../Persistence/SiteStorageService.cs | 50 ++++++++++++ .../Actors/NativeAlarmActorTests.cs | 34 ++++++++ 3 files changed, 154 insertions(+), 7 deletions(-) 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]