From 6cbf763719ba5eebb339947512a4ecafd395ac8c Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 10:17:30 -0400 Subject: [PATCH] fix(alarms): engine cache prune + bounded dispose, source Warning, cached dead-letter depth, script-log rate limit (E3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - R05:1/12 ScriptedAlarmEngine: _valueCache is cleared in LoadAsync (it is re-seeded for the referenced paths) so retired tags no longer accrete across deploys; OnUpstreamChange no longer writes the cache after Dispose. - R05:3 ScriptedAlarmEngine.Dispose drains in-flight work for at most 5 s (DisposeDrainTimeout), then cancels it through a shutdown token every DrainDirtyAsync/ShelvingCheckAsync now runs on, logs a Warning and gives one second of grace — a stalled IAlarmStateStore.SaveAsync can no longer hang host shutdown. A save that completes inside the bound is still drained (positive control test). - R05:6 ScriptedAlarmSource logs a throwing OnAlarmEvent subscriber at Warning (naming the alarm) instead of swallowing it. - R05:7 LocalDbStoreAndForwardSink keeps a cached dead-letter counter in step with every dead-letter/revive/purge path and resyncs it with the queued counter on the same cadence; GetStatus no longer runs a COUNT(*) scan. - R05:9 ScriptLogRateLimiter (20 per source per minute, fixed window, one summary entry naming the dropped count when the window rolls) behind ScriptLogTopicSink (key = script|vtag|alarm) and ScriptLogCompanionSink (key = script name). Refs #734 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT --- .../LocalDbStoreAndForwardSink.cs | 59 +++++-- .../ScriptedAlarmEngine.cs | 53 +++++- .../ScriptedAlarmSource.cs | 19 ++- .../ScriptLogCompanionSink.cs | 23 ++- .../ScriptLogRateLimiter.cs | 123 ++++++++++++++ .../ScriptLogTopicSink.cs | 49 +++++- .../LocalDbStoreAndForwardSinkTests.cs | 34 ++++ .../ScriptedAlarmEngineLifecycleTests.cs | 159 ++++++++++++++++++ .../ScriptedAlarmSourceLoggingTests.cs | 59 +++++++ .../ScriptLogRateLimiterTests.cs | 137 +++++++++++++++ 10 files changed, 683 insertions(+), 32 deletions(-) create mode 100644 src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptLogRateLimiter.cs create mode 100644 tests/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests/ScriptedAlarmEngineLifecycleTests.cs create mode 100644 tests/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests/ScriptedAlarmSourceLoggingTests.cs create mode 100644 tests/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests/ScriptLogRateLimiterTests.cs diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian/LocalDbStoreAndForwardSink.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian/LocalDbStoreAndForwardSink.cs index 8a9b9ec4..6d239f75 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian/LocalDbStoreAndForwardSink.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian/LocalDbStoreAndForwardSink.cs @@ -92,6 +92,12 @@ public sealed class LocalDbStoreAndForwardSink : IAlarmHistorianSink, IDisposabl private bool? _lastGateDecision; private long _queuedRowCount; + // Cached dead-letter depth (R05:7). GetStatus used to run a synchronous COUNT(*) table scan per + // call — one per /healthz probe and per AdminUI poll on the shared LocalDb file. Kept in step by + // every path that flips dead_lettered (drain outcomes, corrupt rows, RetryDeadLettered, retention + // purge) and resynced from storage on the same cadence as _queuedRowCount, so a row the pair peer + // replicates in is visible after the next resync rather than never. + private long _deadLetterRowCount; // Probe counter — incremented every time we actually issue a real COUNT(*) for // capacity enforcement. Public for test instrumentation only. private long _capacityProbeCount; @@ -145,9 +151,16 @@ public sealed class LocalDbStoreAndForwardSink : IAlarmHistorianSink, IDisposabl _clock = clock ?? (() => DateTime.UtcNow); _drainGate = drainGate ?? (static () => true); - _queuedRowCount = ProbeQueuedRowCount(); + _queuedRowCount = ProbeRowCount(deadLettered: false); + _deadLetterRowCount = ProbeRowCount(deadLettered: true); } + /// Test-only: re-sync both cached counters (queued + dead-lettered) from storage — the same + /// resync the enqueue cadence drives every enqueues. + /// Cancellation token. + /// A task that completes once both counters match storage. + public Task DebugResyncCountersAsync(CancellationToken ct) => ResyncRowCountsAsync(ct); + /// /// Start the background drain worker. Not started automatically so tests can /// drive deterministically. @@ -260,7 +273,7 @@ public sealed class LocalDbStoreAndForwardSink : IAlarmHistorianSink, IDisposabl // has a second way to fall behind that no local code path can observe. if (enqueuesSinceResync >= ResyncEnqueueInterval) { - await ResyncQueuedRowCountAsync(ct).ConfigureAwait(false); + await ResyncRowCountsAsync(ct).ConfigureAwait(false); cached = Interlocked.Read(ref _queuedRowCount); Interlocked.Exchange(ref _enqueuesSinceResync, 0); } @@ -273,22 +286,26 @@ public sealed class LocalDbStoreAndForwardSink : IAlarmHistorianSink, IDisposabl await EnforceCapacityAsync(ct).ConfigureAwait(false); } - /// Synchronously query COUNT(*) of non-dead-lettered rows. Used at startup. - private long ProbeQueuedRowCount() + /// Synchronously query COUNT(*) of queued or dead-lettered rows. Used at startup. + private long ProbeRowCount(bool deadLettered) { Interlocked.Increment(ref _capacityProbeCount); using var conn = _db.CreateConnection(); using var cmd = conn.CreateCommand(); - cmd.CommandText = "SELECT COUNT(*) FROM alarm_sf_events WHERE dead_lettered = 0"; + cmd.CommandText = deadLettered + ? "SELECT COUNT(*) FROM alarm_sf_events WHERE dead_lettered = 1" + : "SELECT COUNT(*) FROM alarm_sf_events WHERE dead_lettered = 0"; return (long)(cmd.ExecuteScalar() ?? 0L); } - /// Re-sync the in-memory counter from storage (async path). - private async Task ResyncQueuedRowCountAsync(CancellationToken ct) + /// Re-sync both in-memory counters from storage (async path). + private async Task ResyncRowCountsAsync(CancellationToken ct) { Interlocked.Increment(ref _capacityProbeCount); var live = await CountAsync("dead_lettered = 0", ct).ConfigureAwait(false); Interlocked.Exchange(ref _queuedRowCount, live); + var dead = await CountAsync("dead_lettered = 1", ct).ConfigureAwait(false); + Interlocked.Exchange(ref _deadLetterRowCount, dead); } private async Task CountAsync(string predicate, CancellationToken ct) @@ -356,6 +373,7 @@ public sealed class LocalDbStoreAndForwardSink : IAlarmHistorianSink, IDisposabl await corruptTx.CommitAsync(ct).ConfigureAwait(false); } Interlocked.Add(ref _queuedRowCount, -corruptIds.Count); + Interlocked.Add(ref _deadLetterRowCount, corruptIds.Count); _logger.Warning( "Dead-lettered {Count} historian queue row(s) with un-deserializable payload", corruptIds.Count); @@ -405,6 +423,7 @@ public sealed class LocalDbStoreAndForwardSink : IAlarmHistorianSink, IDisposabl } int rowsLeavingQueue = 0; + int rowsDeadLettered = 0; await using (var tx = await _db.BeginTransactionAsync(ct).ConfigureAwait(false)) { for (var i = 0; i < outcomes.Count; i++) @@ -423,6 +442,7 @@ public sealed class LocalDbStoreAndForwardSink : IAlarmHistorianSink, IDisposabl case HistorianWriteOutcome.PermanentFail: await DeadLetterRowAsync(tx, id, $"permanent fail at {_clock():O}", ct).ConfigureAwait(false); rowsLeavingQueue++; + rowsDeadLettered++; break; case HistorianWriteOutcome.RetryPlease: // finding 002: cap retries so a perpetually-RetryPlease (poison) @@ -434,6 +454,7 @@ public sealed class LocalDbStoreAndForwardSink : IAlarmHistorianSink, IDisposabl { await DeadLetterRowAsync(tx, id, $"max attempts ({_maxAttempts}) exceeded", ct).ConfigureAwait(false); rowsLeavingQueue++; + rowsDeadLettered++; } else { @@ -449,6 +470,8 @@ public sealed class LocalDbStoreAndForwardSink : IAlarmHistorianSink, IDisposabl // cap (finding 002) — keep the counter aligned. if (rowsLeavingQueue > 0) Interlocked.Add(ref _queuedRowCount, -rowsLeavingQueue); + if (rowsDeadLettered > 0) + Interlocked.Add(ref _deadLetterRowCount, rowsDeadLettered); var acks = outcomes.Count(o => o == HistorianWriteOutcome.Ack); lock (_statusLock) @@ -523,13 +546,9 @@ public sealed class LocalDbStoreAndForwardSink : IAlarmHistorianSink, IDisposabl var queued = Interlocked.Read(ref _queuedRowCount); if (queued < 0) queued = 0; - long deadlettered; - using (var conn = _db.CreateConnection()) - using (var cmd = conn.CreateCommand()) - { - cmd.CommandText = "SELECT COUNT(*) FROM alarm_sf_events WHERE dead_lettered = 1"; - deadlettered = (long)(cmd.ExecuteScalar() ?? 0L); - } + // Cached, never a live scan (R05:7) — see the field's remarks for the resync cadence. + var deadlettered = Interlocked.Read(ref _deadLetterRowCount); + if (deadlettered < 0) deadlettered = 0; DateTime? lastDrain, lastSuccess; string? lastError; @@ -568,7 +587,11 @@ public sealed class LocalDbStoreAndForwardSink : IAlarmHistorianSink, IDisposabl var revived = cmd.ExecuteNonQuery(); // Dead-lettered rows rejoin the non-dead-lettered queue — keep the in-memory // counter aligned. - if (revived > 0) Interlocked.Add(ref _queuedRowCount, revived); + if (revived > 0) + { + Interlocked.Add(ref _queuedRowCount, revived); + Interlocked.Add(ref _deadLetterRowCount, -revived); + } return revived; } @@ -646,8 +669,9 @@ public sealed class LocalDbStoreAndForwardSink : IAlarmHistorianSink, IDisposabl Interlocked.Increment(ref _capacityProbeCount); var count = await CountAsync("dead_lettered = 0", ct).ConfigureAwait(false); - // Resync the in-memory counter while we have a fresh number. + // Resync the in-memory counters while we have a fresh number. Interlocked.Exchange(ref _queuedRowCount, count); + Interlocked.Exchange(ref _deadLetterRowCount, await CountAsync("dead_lettered = 1", ct).ConfigureAwait(false)); if (count < _capacity) return; var toEvict = count - _capacity + 1; @@ -684,7 +708,10 @@ public sealed class LocalDbStoreAndForwardSink : IAlarmHistorianSink, IDisposabl ct).ConfigureAwait(false); if (purged > 0) + { + Interlocked.Add(ref _deadLetterRowCount, -purged); _logger.Information("Purged {Count} dead-lettered row(s) past retention window", purged); + } } private void BumpBackoff() => _backoffIndex = Math.Min(_backoffIndex + 1, BackoffLadder.Length - 1); diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms/ScriptedAlarmEngine.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms/ScriptedAlarmEngine.cs index 808935ae..9438132a 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms/ScriptedAlarmEngine.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms/ScriptedAlarmEngine.cs @@ -170,6 +170,24 @@ public sealed class ScriptedAlarmEngine : IDisposable => _scratchByAlarmId.TryGetValue(alarmId, out var s) ? s.Context : null; private readonly ConcurrentDictionary _valueCache = new(StringComparer.Ordinal); + + /// Test hook — the number of paths currently held in the value cache. + internal int ValueCacheCountForTest => _valueCache.Count; + + /// Test hook — whether is currently held in the value cache. + internal bool ValueCacheContainsForTest(string path) => _valueCache.ContainsKey(path); + + /// + /// How long waits for in-flight background work (a re-evaluation or shelving + /// check mid-SaveAsync) before cancelling it through and + /// returning. 5 s, like PollGroupEngine's drain; a hung store must not hang host shutdown. + /// + internal TimeSpan DisposeDrainTimeout { get; set; } = TimeSpan.FromSeconds(5); + + /// Cancelled by once the drain bound expires — every background task + /// (DrainDirtyAsync, ShelvingCheckAsync) runs on its token, so a store call that honours + /// cancellation returns instead of outliving the engine. + private readonly CancellationTokenSource _shutdownCts = new(); private readonly Dictionary> _alarmsReferencing = new(StringComparer.Ordinal); // tag path -> alarm ids @@ -281,6 +299,10 @@ public sealed class ScriptedAlarmEngine : IDisposable UnsubscribeFromUpstream(); _alarms.Clear(); _alarmsReferencing.Clear(); + // The value cache is re-seeded below for exactly the paths this generation references. It + // used to survive reloads untouched, so every path a retired alarm had ever referenced stayed + // cached forever — a slow leak over many deploys with a churning tag set. + _valueCache.Clear(); // Drop the prior generation's per-alarm scratch buffers — definitions may // have changed (different Inputs, different Logger), so any reuse would be // unsafe. @@ -573,8 +595,8 @@ public sealed class ScriptedAlarmEngine : IDisposable /// The new data value snapshot. internal void OnUpstreamChange(string path, DataValueSnapshot value) { + if (_disposed) return; // a disposed engine neither caches nor queues — nothing reads it again _valueCache[path] = value; - if (_disposed) return; // don't queue new work against a disposing engine if (!_alarmsReferencing.TryGetValue(path, out var alarmIds)) return; bool startDrain; @@ -589,7 +611,7 @@ public sealed class ScriptedAlarmEngine : IDisposable // Started outside the lock: DrainDirtyAsync runs synchronously up to its // first real await, and nothing it does should happen under _dirtyLock. - if (startDrain) TrackBackgroundTask(() => DrainDirtyAsync(CancellationToken.None)); + if (startDrain) TrackBackgroundTask(() => DrainDirtyAsync(_shutdownCts.Token)); } /// @@ -732,6 +754,11 @@ public sealed class ScriptedAlarmEngine : IDisposable } finally { _evalGate.Release(); } } + catch (OperationCanceledException) when (_shutdownCts.IsCancellationRequested) + { + // Dispose cut the drain short after its bound — expected, not a failure. + return; + } catch (Exception ex) { _engineLogger.Error(ex, "ScriptedAlarmEngine reevaluate failed"); @@ -1048,7 +1075,7 @@ public sealed class ScriptedAlarmEngine : IDisposable { if (_disposed) return; var ids = _alarms.Keys.ToArray(); - TrackBackgroundTask(() => ShelvingCheckAsync(ids, CancellationToken.None)); + TrackBackgroundTask(() => ShelvingCheckAsync(ids, _shutdownCts.Token)); } /// @@ -1187,6 +1214,10 @@ public sealed class ScriptedAlarmEngine : IDisposable } finally { _evalGate.Release(); } } + catch (OperationCanceledException) when (_shutdownCts.IsCancellationRequested) + { + return; // Dispose cut the check short after its bound + } catch (Exception ex) { _engineLogger.Warning(ex, "ScriptedAlarmEngine shelving-check failed"); @@ -1227,11 +1258,25 @@ public sealed class ScriptedAlarmEngine : IDisposable // flight can outlive the engine and write to a (possibly disposed) store // after Dispose() has returned. The tasks re-check _disposed after // acquiring the gate and bail out, but the await still has to complete. + // BOUNDED (R05:3): the drained tasks end in IAlarmStateStore.SaveAsync, and a stalled store + // (connection stall, lock, replication hiccup) used to hang this — and the host shutdown that + // calls it — indefinitely. Wait DisposeDrainTimeout; past it, cancel the tasks' shared token (a + // store that honours cancellation returns promptly), log, and give them one short grace period. Task[] toAwait; lock (_inFlightLock) { toAwait = [.. _inFlight]; } if (toAwait.Length > 0) { - try { Task.WhenAll(toAwait).GetAwaiter().GetResult(); } + var all = Task.WhenAll(toAwait); + try { all.WaitAsync(DisposeDrainTimeout).GetAwaiter().GetResult(); } + catch (TimeoutException) + { + _engineLogger.Warning( + "ScriptedAlarmEngine dispose: {Count} background task(s) still running after {Timeout} — cancelling them (a store call is stalled)", + toAwait.Length, DisposeDrainTimeout); + _shutdownCts.Cancel(); + try { all.WaitAsync(TimeSpan.FromSeconds(1)).GetAwaiter().GetResult(); } + catch (Exception) { /* cancelled, faulted, or still stuck — we are leaving either way */ } + } catch (Exception ex) { // Background task failures already logged inside ReevaluateAsync / diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms/ScriptedAlarmSource.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms/ScriptedAlarmSource.cs index dc76923c..9d44aee5 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms/ScriptedAlarmSource.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms/ScriptedAlarmSource.cs @@ -1,4 +1,5 @@ using System.Collections.Concurrent; +using Serilog; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; namespace ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms; @@ -23,19 +24,27 @@ namespace ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms; /// adapter doesn't maintain per-subscription state beyond the filter set — it /// checks each emission against every live subscription. /// +/// +/// A throwing subscriber is logged at Warning (naming the alarm) and +/// does not crash the adapter — the same posture as the engine's own FireEvent. It used to +/// be swallowed silently: a lost alarm event with no trace. +/// /// public sealed class ScriptedAlarmSource : IAlarmSource, IDisposable { private readonly ScriptedAlarmEngine _engine; + private readonly ILogger _logger; private readonly ConcurrentDictionary _subscriptions = new(StringComparer.Ordinal); private bool _disposed; /// Initializes a new instance of the class. /// The scripted alarm engine to expose. - public ScriptedAlarmSource(ScriptedAlarmEngine engine) + /// Where a throwing subscriber is reported; null ⇒ Serilog's global logger. + public ScriptedAlarmSource(ScriptedAlarmEngine engine, ILogger? logger = null) { _engine = engine ?? throw new ArgumentNullException(nameof(engine)); + _logger = logger ?? Log.Logger; _engine.OnEvent += OnEngineEvent; } @@ -98,7 +107,13 @@ public sealed class ScriptedAlarmSource : IAlarmSource, IDisposable Severity: evt.Severity, SourceTimestampUtc: evt.TimestampUtc); try { OnAlarmEvent?.Invoke(this, payload); } - catch { /* subscriber exceptions don't crash the adapter */ } + catch (Exception ex) + { + // Subscriber exceptions don't crash the adapter — but they are not silent either: a lost + // alarm event must leave a trace the operator can find. + _logger.Warning(ex, "ScriptedAlarmSource OnAlarmEvent subscriber threw for {AlarmId} ({Emission})", + evt.AlarmId, evt.Emission); + } } } diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptLogCompanionSink.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptLogCompanionSink.cs index fb1aff35..84bb0e04 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptLogCompanionSink.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptLogCompanionSink.cs @@ -24,22 +24,32 @@ namespace ZB.MOM.WW.OtOpcUa.Core.Scripting; /// Original exception (if any) is attached so the main log's diagnostics keep /// the full stack trace. /// +/// +/// Rate-limited per script. A predicate that throws on every push would otherwise mirror +/// one Warning per push into the main log; a keyed by script +/// name admits 20 per minute and mirrors one summary line for the rest when the window rolls. The +/// scripts-*.log file sink is unaffected and keeps the full stream. +/// /// public sealed class ScriptLogCompanionSink : ILogEventSink { private readonly ILogger _mainLogger; private readonly LogEventLevel _minMirrorLevel; + private readonly ScriptLogRateLimiter _limiter; /// Initializes a new instance of the ScriptLogCompanionSink. /// The companion logger to mirror events to. /// The minimum log level to mirror (default: Error). - public ScriptLogCompanionSink(ILogger mainLogger, LogEventLevel minMirrorLevel = LogEventLevel.Error) + /// Per-script rate limiter; null ⇒ the default 20 per minute. + public ScriptLogCompanionSink(ILogger mainLogger, LogEventLevel minMirrorLevel = LogEventLevel.Error, + ScriptLogRateLimiter? limiter = null) { _mainLogger = mainLogger ?? throw new ArgumentNullException(nameof(mainLogger)); _minMirrorLevel = minMirrorLevel; + _limiter = limiter ?? new ScriptLogRateLimiter(); } - /// Emits a log event to the companion logger if it meets the minimum level. + /// Emits a log event to the companion logger if it meets the minimum level and the script is within its per-minute budget. /// The log event to emit. public void Emit(LogEvent logEvent) { @@ -54,6 +64,15 @@ public sealed class ScriptLogCompanionSink : ILogEventSink scriptName = s; } + var admission = _limiter.Admit(scriptName); + if (admission.SuppressedInPreviousWindow > 0) + { + _mainLogger.Warning( + "[Script] {ScriptName}: {SuppressedCount} further script error(s) were suppressed in the previous {WindowSeconds:F0} s; see scripts-*.log", + scriptName, admission.SuppressedInPreviousWindow, _limiter.Window.TotalSeconds); + } + if (!admission.Admitted) return; + var rendered = logEvent.RenderMessage(); if (logEvent.Exception is not null) { diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptLogRateLimiter.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptLogRateLimiter.cs new file mode 100644 index 00000000..17a6f2f6 --- /dev/null +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptLogRateLimiter.cs @@ -0,0 +1,123 @@ +namespace ZB.MOM.WW.OtOpcUa.Core.Scripting; + +/// +/// Per-source fixed-window rate limit for the script-log fan-out: the first +/// entries a source emits in a are +/// admitted, the rest are counted and dropped, and the count is handed back exactly once — on the first +/// admission after the window rolls — so the caller can emit one summary line in the dropped entries' +/// place. Nothing is silently lost (the operator sees "N suppressed"), and one chatty script cannot +/// shed every other script's entries out of the bounded telemetry channels (drop-oldest) or flood the +/// cluster bus / main log. +/// +/// +/// Used by (key = script + virtual-tag + alarm identity), +/// (key = script name) and the virtual-tag actor's per-failure +/// log (one key per tag). Thread-safe; Serilog invokes a sink on whichever thread logged. Idle sources +/// are swept every admissions so a churning tag set cannot grow the table +/// without bound. +/// +public sealed class ScriptLogRateLimiter +{ + /// Entries admitted per source per window by default: 20. + public const int DefaultLimitPerWindow = 20; + + /// The default window: one minute. + public static readonly TimeSpan DefaultWindow = TimeSpan.FromMinutes(1); + + private const int SweepEvery = 256; + + private readonly Dictionary _sources = new(StringComparer.Ordinal); + private readonly int _limit; + private readonly TimeSpan _window; + private readonly Func _clock; + private int _admitsSinceSweep; + + /// Initializes a limiter. + /// Entries admitted per source per window; values below 1 are treated as 1. + /// The fixed window; null ⇒ . + /// UTC clock; null ⇒ . + public ScriptLogRateLimiter(int limitPerWindow = DefaultLimitPerWindow, TimeSpan? window = null, Func? clock = null) + { + _limit = Math.Max(1, limitPerWindow); + _window = window is { } w && w > TimeSpan.Zero ? w : DefaultWindow; + _clock = clock ?? (static () => DateTime.UtcNow); + } + + /// The window length. + public TimeSpan Window => _window; + + /// + /// Ask whether 's next entry may be emitted. When the window has rolled + /// since the last call, carries the + /// number of entries dropped in the window that just closed (zero otherwise) — the caller emits one + /// summary for them BEFORE the admitted entry. + /// + /// Identity of the emitting script instance. + /// The admission decision. + public ScriptLogAdmission Admit(string sourceKey) + { + ArgumentNullException.ThrowIfNull(sourceKey); + var now = _clock(); + lock (_sources) + { + if (++_admitsSinceSweep >= SweepEvery) + { + _admitsSinceSweep = 0; + Sweep(now); + } + + if (!_sources.TryGetValue(sourceKey, out var source)) + { + source = new Source { WindowStart = now }; + _sources[sourceKey] = source; + } + + long summary = 0; + if (now - source.WindowStart >= _window) + { + summary = source.Suppressed; + source.WindowStart = now; + source.Admitted = 0; + source.Suppressed = 0; + } + + source.LastSeen = now; + if (source.Admitted < _limit) + { + source.Admitted++; + return new ScriptLogAdmission(true, summary); + } + + source.Suppressed++; + return new ScriptLogAdmission(false, summary); + } + } + + private void Sweep(DateTime now) + { + var idle = now - (_window + _window); + List? dead = null; + foreach (var (key, source) in _sources) + { + // A source that still has a pending summary is kept: its next entry must report the count. + if (source.LastSeen < idle && source.Suppressed == 0) + (dead ??= []).Add(key); + } + if (dead is null) return; + foreach (var key in dead) _sources.Remove(key); + } + + private sealed class Source + { + public DateTime WindowStart; + public DateTime LastSeen; + public int Admitted; + public long Suppressed; + } +} + +/// Result of . +/// Whether the entry may be emitted. +/// Entries dropped in the window that just closed — non-zero at +/// most once per window roll; the caller emits one summary line for them. +public readonly record struct ScriptLogAdmission(bool Admitted, long SuppressedInPreviousWindow); diff --git a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptLogTopicSink.cs b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptLogTopicSink.cs index fe9e7f3e..a7b8f03e 100644 --- a/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptLogTopicSink.cs +++ b/src/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting/ScriptLogTopicSink.cs @@ -25,25 +25,37 @@ namespace ZB.MOM.WW.OtOpcUa.Core.Scripting; /// property so pipelines that pre-date this sink continue to work correctly. If /// neither property is present ScriptId defaults to "unknown". /// +/// +/// Rate-limited per source. A script that logs at Information on every evaluation of a 1 Hz +/// input put one rendered string per second per script onto the cluster bus; 200 such scripts shed +/// everyone's entries out of the bounded telemetry channels. A +/// keyed by the entry's identity (script + virtual tag + alarm) admits 20 entries per minute per +/// source and publishes one summary entry naming the dropped count when the window rolls. +/// /// public sealed class ScriptLogTopicSink : ILogEventSink { private readonly IScriptLogPublisher _publisher; private readonly LogEventLevel _min; + private readonly ScriptLogRateLimiter _limiter; /// Initializes a new instance of the class. /// The publisher that routes entries to the cluster bus. Must not be null. /// Minimum log level to forward. Events below this level are discarded (default: ). + /// Per-source rate limiter; null ⇒ the default 20 per minute. /// Thrown when is null. - public ScriptLogTopicSink(IScriptLogPublisher publisher, LogEventLevel min = LogEventLevel.Information) + public ScriptLogTopicSink(IScriptLogPublisher publisher, LogEventLevel min = LogEventLevel.Information, + ScriptLogRateLimiter? limiter = null) { _publisher = publisher ?? throw new ArgumentNullException(nameof(publisher)); _min = min; + _limiter = limiter ?? new ScriptLogRateLimiter(); } /// /// Converts the to a and - /// publishes it, unless the event is below the configured minimum level. + /// publishes it, unless the event is below the configured minimum level or the source is over its + /// per-minute budget. /// /// The Serilog event to process. Silently ignored when null. public void Emit(LogEvent logEvent) @@ -56,15 +68,36 @@ public sealed class ScriptLogTopicSink : ILogEventSink ? s : null; + var scriptId = P(ScriptLoggerFactory.ScriptIdProperty) + ?? P(ScriptLoggerFactory.ScriptNameProperty) + ?? "unknown"; + var virtualTagId = P(ScriptLoggerFactory.VirtualTagIdProperty); + var alarmId = P(ScriptLoggerFactory.AlarmIdProperty); + var equipmentId = P(ScriptLoggerFactory.EquipmentIdProperty); + + var admission = _limiter.Admit(string.Concat(scriptId, "|", virtualTagId, "|", alarmId)); + if (admission.SuppressedInPreviousWindow > 0) + { + _publisher.Publish(new ScriptLogEntry( + ScriptId: scriptId, + Level: nameof(LogEventLevel.Warning), + Message: $"{admission.SuppressedInPreviousWindow} script-log entr{(admission.SuppressedInPreviousWindow == 1 ? "y" : "ies")} " + + $"from this source were suppressed in the previous {_limiter.Window.TotalSeconds:F0} s " + + $"(limit {ScriptLogRateLimiter.DefaultLimitPerWindow}/window); see scripts-*.log for the full stream", + TimestampUtc: logEvent.Timestamp.UtcDateTime, + VirtualTagId: virtualTagId, + AlarmId: alarmId, + EquipmentId: equipmentId)); + } + if (!admission.Admitted) return; + _publisher.Publish(new ScriptLogEntry( - ScriptId: P(ScriptLoggerFactory.ScriptIdProperty) - ?? P(ScriptLoggerFactory.ScriptNameProperty) - ?? "unknown", + ScriptId: scriptId, Level: logEvent.Level.ToString(), Message: logEvent.RenderMessage(), TimestampUtc: logEvent.Timestamp.UtcDateTime, - VirtualTagId: P(ScriptLoggerFactory.VirtualTagIdProperty), - AlarmId: P(ScriptLoggerFactory.AlarmIdProperty), - EquipmentId: P(ScriptLoggerFactory.EquipmentIdProperty))); + VirtualTagId: virtualTagId, + AlarmId: alarmId, + EquipmentId: equipmentId)); } } diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian.Tests/LocalDbStoreAndForwardSinkTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian.Tests/LocalDbStoreAndForwardSinkTests.cs index 9c4c6b11..97d68679 100644 --- a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian.Tests/LocalDbStoreAndForwardSinkTests.cs +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.AlarmHistorian.Tests/LocalDbStoreAndForwardSinkTests.cs @@ -342,6 +342,40 @@ public sealed class LocalDbStoreAndForwardSinkTests : IDisposable sink.GetStatus().DeadLetterDepth.ShouldBe(0, "purged past retention"); } + /// + /// Review R05:7 — GetStatus no longer runs a COUNT(*) table scan per call for the + /// dead-letter depth: it reads a cached counter kept in step by every dead-letter / revive / purge + /// path (the other tests in this class pin those), and resynced from storage on the same cadence as + /// the queued-row counter. A row flipped underneath the sink (the peer replicating a dead-letter in) + /// is therefore invisible until the next resync — that is the documented trade, and the resync is + /// what makes it bounded rather than permanent. + /// + [Fact] + public async Task DeadLetterDepth_is_served_from_a_cached_counter_that_the_periodic_resync_corrects() + { + var writer = new FakeWriter(); + using var sink = new LocalDbStoreAndForwardSink(Db, writer, _log); + await sink.EnqueueAsync(Event("A1"), CancellationToken.None); + await sink.EnqueueAsync(Event("A2"), CancellationToken.None); + sink.GetStatus().DeadLetterDepth.ShouldBe(0); + var probesBefore = sink.DebugCapacityProbeCount; + + // Drift the table underneath the sink, exactly as a replicated dead-letter from the pair peer would. + await Db.ExecuteAsync( + "UPDATE alarm_sf_events SET dead_lettered = 1 WHERE id IN (SELECT id FROM alarm_sf_events LIMIT 1)", + null, CancellationToken.None); + + // Cached: GetStatus did not go to the table, so it still reports the pre-drift depth… + sink.GetStatus().DeadLetterDepth.ShouldBe(0, "GetStatus serves the cached counter, not a live COUNT(*)"); + sink.DebugCapacityProbeCount.ShouldBe(probesBefore, "GetStatus must not probe storage"); + + // …and the resync (the one the enqueue cadence drives) brings both counters back in line. + await sink.DebugResyncCountersAsync(CancellationToken.None); + var status = sink.GetStatus(); + status.DeadLetterDepth.ShouldBe(1); + status.QueueDepth.ShouldBe(1); + } + /// Verifies that RetryDeadLettered requeues dead-lettered rows for retry. [Fact] public async Task RetryDeadLettered_requeues_for_retry() diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests/ScriptedAlarmEngineLifecycleTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests/ScriptedAlarmEngineLifecycleTests.cs new file mode 100644 index 00000000..7afea07a --- /dev/null +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests/ScriptedAlarmEngineLifecycleTests.cs @@ -0,0 +1,159 @@ +using Serilog; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Core.Scripting; + +namespace ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests; + +/// +/// Review R05:1 / R05:12 / R05:3 — the engine's value cache is pruned on every reload (it used to keep +/// every path ever referenced, forever), is no longer written after Dispose, and Dispose's in-flight +/// drain is bounded (a hung store no longer hangs host shutdown). +/// +[Trait("Category", "Unit")] +public sealed class ScriptedAlarmEngineLifecycleTests +{ + private static ScriptedAlarmDefinition Alarm(string id, string predicate) => + new(AlarmId: id, EquipmentPath: "Plant/Line1/Reactor", AlarmName: id, Kind: AlarmKind.AlarmCondition, + Severity: AlarmSeverity.High, MessageTemplate: "condition", PredicateScriptSource: predicate); + + private static ScriptedAlarmEngine Build(FakeUpstream up, IAlarmStateStore? store = null) + { + var logger = new LoggerConfiguration().CreateLogger(); + return new ScriptedAlarmEngine(up, store ?? new InMemoryAlarmStateStore(), new ScriptLoggerFactory(logger), logger); + } + + [Fact] + public async Task Reload_prunes_value_cache_entries_for_paths_the_new_alarm_set_no_longer_references() + { + var up = new FakeUpstream(); + up.Set("Temp", 50); + up.Set("Pressure", 5); + using var eng = Build(up); + var ct = TestContext.Current.CancellationToken; + + await eng.LoadAsync( + [Alarm("a", """return (int)ctx.GetTag("Temp").Value > 100 || (int)ctx.GetTag("Pressure").Value > 10;""")], ct); + eng.ValueCacheCountForTest.ShouldBe(2); + + // The redeploy drops Pressure from every predicate: its entry must go with it. + await eng.LoadAsync([Alarm("a", """return (int)ctx.GetTag("Temp").Value > 100;""")], ct); + eng.ValueCacheCountForTest.ShouldBe(1); + eng.ValueCacheContainsForTest("Pressure").ShouldBeFalse(); + eng.ValueCacheContainsForTest("Temp").ShouldBeTrue(); + } + + [Fact] + public async Task A_push_after_Dispose_does_not_write_the_value_cache() + { + var up = new FakeUpstream(); + up.Set("Temp", 50); + var eng = Build(up); + await eng.LoadAsync([Alarm("a", """return (int)ctx.GetTag("Temp").Value > 100;""")], + TestContext.Current.CancellationToken); + eng.Dispose(); + eng.ValueCacheCountForTest.ShouldBe(1); + + eng.OnUpstreamChange("Late", new DataValueSnapshot(1, 0u, DateTime.UtcNow, DateTime.UtcNow)); + + eng.ValueCacheContainsForTest("Late").ShouldBeFalse("a disposed engine must not grow its cache"); + } + + [Fact] + public async Task Dispose_gives_up_on_a_store_that_never_returns_after_the_drain_bound_and_cancels_it() + { + var up = new FakeUpstream(); + up.Set("Temp", 50); + var store = new NeverReturningSaveStore(); + var eng = Build(up, store); + eng.DisposeDrainTimeout = TimeSpan.FromMilliseconds(300); + await eng.LoadAsync([Alarm("A", """return (int)ctx.GetTag("Temp").Value > 100;""")], + TestContext.Current.CancellationToken); + + store.BlockNext = true; + up.Push("Temp", 150); // Activated → SaveAsync → blocks forever (until cancelled) + await WaitForAsync(() => store.SaveInProgress); + + var sw = System.Diagnostics.Stopwatch.StartNew(); + eng.Dispose(); + sw.Stop(); + + sw.Elapsed.ShouldBeLessThan(TimeSpan.FromSeconds(5), "the drain is bounded; a hung store cannot hang shutdown"); + await WaitForAsync(() => store.ObservedCancellation, timeoutMs: 2000); + store.ObservedCancellation.ShouldBeTrue("the in-flight save was cancelled through the engine's shutdown token"); + } + + [Fact] + public async Task Dispose_still_waits_for_a_save_that_completes_inside_the_bound() + { + // Positive control for the bound: a save that finishes promptly is drained, not cut off. + var up = new FakeUpstream(); + up.Set("Temp", 50); + var store = new NeverReturningSaveStore { ReleaseAfter = TimeSpan.FromMilliseconds(100) }; + var eng = Build(up, store); + eng.DisposeDrainTimeout = TimeSpan.FromSeconds(5); + await eng.LoadAsync([Alarm("A", """return (int)ctx.GetTag("Temp").Value > 100;""")], + TestContext.Current.CancellationToken); + + store.BlockNext = true; + up.Push("Temp", 150); + await WaitForAsync(() => store.SaveInProgress); + + eng.Dispose(); + + store.SaveInProgress.ShouldBeFalse(); + store.ObservedCancellation.ShouldBeFalse("a save that completes within the bound is never cancelled"); + store.Saved.ShouldContain("A"); + } + + private static async Task WaitForAsync(Func cond, int timeoutMs = 5000) + { + var deadline = DateTime.UtcNow.AddMilliseconds(timeoutMs); + while (DateTime.UtcNow < deadline) + { + if (cond()) return; + await Task.Delay(20); + } + throw new TimeoutException("Condition did not become true in time"); + } + + /// A store whose next SaveAsync blocks until cancelled (or until ). + private sealed class NeverReturningSaveStore : IAlarmStateStore + { + private readonly InMemoryAlarmStateStore _inner = new(); + public volatile bool BlockNext; + public volatile bool SaveInProgress; + public volatile bool ObservedCancellation; + public TimeSpan? ReleaseAfter { get; set; } + public List Saved { get; } = []; + + public Task LoadAsync(string alarmId, CancellationToken ct) => _inner.LoadAsync(alarmId, ct); + public Task> LoadAllAsync(CancellationToken ct) => _inner.LoadAllAsync(ct); + public Task RemoveAsync(string alarmId, CancellationToken ct) => _inner.RemoveAsync(alarmId, ct); + + public async Task SaveAsync(AlarmConditionState state, CancellationToken ct) + { + if (BlockNext) + { + BlockNext = false; + SaveInProgress = true; + try + { + await Task.Delay(ReleaseAfter ?? Timeout.InfiniteTimeSpan, ct).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + ObservedCancellation = true; + throw; + } + finally + { + SaveInProgress = false; + } + } + lock (Saved) Saved.Add(state.AlarmId); + await _inner.SaveAsync(state, ct).ConfigureAwait(false); + } + } +} diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests/ScriptedAlarmSourceLoggingTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests/ScriptedAlarmSourceLoggingTests.cs new file mode 100644 index 00000000..94c0408c --- /dev/null +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests/ScriptedAlarmSourceLoggingTests.cs @@ -0,0 +1,59 @@ +using Serilog; +using Serilog.Core; +using Serilog.Events; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Core.Abstractions; +using ZB.MOM.WW.OtOpcUa.Core.Scripting; + +namespace ZB.MOM.WW.OtOpcUa.Core.ScriptedAlarms.Tests; + +/// Review R05:6 — a throwing subscriber used to be +/// swallowed with no diagnostic: a lost alarm event with no trace. It is now logged at Warning. +[Trait("Category", "Unit")] +public sealed class ScriptedAlarmSourceLoggingTests +{ + private sealed class CapturingSink : ILogEventSink + { + public List Events { get; } = []; + public void Emit(LogEvent logEvent) { lock (Events) Events.Add(logEvent); } + } + + [Fact] + public async Task A_throwing_subscriber_is_logged_at_Warning_and_does_not_crash_the_adapter() + { + var sinkCapture = new CapturingSink(); + var logger = new LoggerConfiguration().MinimumLevel.Verbose().WriteTo.Sink(sinkCapture).CreateLogger(); + var up = new FakeUpstream(); + up.Set("Temp", 50); + var engine = new ScriptedAlarmEngine(up, new InMemoryAlarmStateStore(), new ScriptLoggerFactory(logger), logger); + await engine.LoadAsync( + [new ScriptedAlarmDefinition("Plant/Line1::HighTemp", "Plant/Line1", "HighTemp", AlarmKind.LimitAlarm, + AlarmSeverity.High, "Temp {Temp}C", """return (int)ctx.GetTag("Temp").Value > 100;""")], + TestContext.Current.CancellationToken); + using var _e = engine; + using var source = new ScriptedAlarmSource(engine, logger); + await source.SubscribeAlarmsAsync(Array.Empty(), TestContext.Current.CancellationToken); + source.OnAlarmEvent += (_, _) => throw new InvalidOperationException("subscriber broke"); + + up.Push("Temp", 150); + + var deadline = DateTime.UtcNow.AddSeconds(10); + while (DateTime.UtcNow < deadline) + { + lock (sinkCapture.Events) + { + if (sinkCapture.Events.Any(e => e.Level == LogEventLevel.Warning && e.Exception?.Message == "subscriber broke")) + break; + } + await Task.Delay(25); + } + + LogEvent warning; + lock (sinkCapture.Events) + warning = sinkCapture.Events.Single(e => e.Level == LogEventLevel.Warning && e.Exception?.Message == "subscriber broke"); + warning.RenderMessage().ShouldContain("Plant/Line1::HighTemp"); + // The adapter survived: the engine's state still moved. + engine.GetState("Plant/Line1::HighTemp")!.Active.ShouldBe(AlarmActiveState.Active); + } +} diff --git a/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests/ScriptLogRateLimiterTests.cs b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests/ScriptLogRateLimiterTests.cs new file mode 100644 index 00000000..3f560e40 --- /dev/null +++ b/tests/Core/ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests/ScriptLogRateLimiterTests.cs @@ -0,0 +1,137 @@ +using Serilog; +using Serilog.Core; +using Serilog.Events; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Logging; + +namespace ZB.MOM.WW.OtOpcUa.Core.Scripting.Tests; + +/// +/// Per-source rate limit on the script-log fan-out (review R05:9 / R02:P6): a script that logs — or +/// fails — on every evaluation of a 1 Hz input used to put one entry per second per script onto the +/// cluster bus and the main log. The limiter admits the first 20 entries per source per minute, drops +/// the rest, and surfaces one summary line (with the dropped count) when the next entry arrives after +/// the window rolled — so nothing is silently lost, and a chatty script cannot shed everyone else's +/// entries out of the bounded telemetry channels. +/// +[Trait("Category", "Unit")] +public sealed class ScriptLogRateLimiterTests +{ + private sealed class Clock + { + public DateTime Now = new(2026, 1, 1, 12, 0, 0, DateTimeKind.Utc); + public void Advance(TimeSpan by) => Now += by; + } + + [Fact] + public void Admits_the_first_N_per_window_then_suppresses_and_summarises_once_the_window_rolls() + { + var clock = new Clock(); + var limiter = new ScriptLogRateLimiter(limitPerWindow: 3, window: TimeSpan.FromMinutes(1), clock: () => clock.Now); + + limiter.Admit("s1").Admitted.ShouldBeTrue(); + limiter.Admit("s1").Admitted.ShouldBeTrue(); + limiter.Admit("s1").Admitted.ShouldBeTrue(); + limiter.Admit("s1").Admitted.ShouldBeFalse(); + limiter.Admit("s1").Admitted.ShouldBeFalse(); + limiter.Admit("s1").SuppressedInPreviousWindow.ShouldBe(0, "no summary until the window rolls"); + + clock.Advance(TimeSpan.FromMinutes(1)); + var first = limiter.Admit("s1"); + first.Admitted.ShouldBeTrue("a fresh window admits again"); + first.SuppressedInPreviousWindow.ShouldBe(3, "the caller emits one summary for the three dropped entries"); + limiter.Admit("s1").SuppressedInPreviousWindow.ShouldBe(0, "the summary is reported exactly once"); + } + + [Fact] + public void Sources_are_independent() + { + var limiter = new ScriptLogRateLimiter(limitPerWindow: 1); + limiter.Admit("a").Admitted.ShouldBeTrue(); + limiter.Admit("a").Admitted.ShouldBeFalse(); + limiter.Admit("b").Admitted.ShouldBeTrue("a chatty source must not shed another source's entries"); + } + + [Fact] + public void Defaults_are_20_per_minute() + { + ScriptLogRateLimiter.DefaultLimitPerWindow.ShouldBe(20); + ScriptLogRateLimiter.DefaultWindow.ShouldBe(TimeSpan.FromMinutes(1)); + var limiter = new ScriptLogRateLimiter(); + for (var i = 0; i < 20; i++) limiter.Admit("s").Admitted.ShouldBeTrue(); + limiter.Admit("s").Admitted.ShouldBeFalse(); + } + + // ── the two sinks ──────────────────────────────────────────────────────────────────────── + + private sealed class FakePublisher : IScriptLogPublisher + { + public List Published { get; } = []; + public void Publish(ScriptLogEntry entry) => Published.Add(entry); + } + + private static LogEvent Evt(LogEventLevel level, string template, params (string Key, string Val)[] props) + { + var tmpl = new Serilog.Parsing.MessageTemplateParser().Parse(template); + var logProps = props.Select(p => new LogEventProperty(p.Key, new ScalarValue(p.Val))); + return new LogEvent(DateTimeOffset.UtcNow, level, exception: null, tmpl, logProps); + } + + [Fact] + public void Topic_sink_rate_limits_per_source_and_publishes_a_summary_entry() + { + var clock = new Clock(); + var publisher = new FakePublisher(); + var sink = new ScriptLogTopicSink(publisher, LogEventLevel.Information, + new ScriptLogRateLimiter(limitPerWindow: 2, clock: () => clock.Now)); + + for (var i = 0; i < 5; i++) + sink.Emit(Evt(LogEventLevel.Warning, "chatty", (ScriptLoggerFactory.ScriptIdProperty, "S1"), (ScriptLoggerFactory.VirtualTagIdProperty, "V1"))); + sink.Emit(Evt(LogEventLevel.Warning, "other", (ScriptLoggerFactory.ScriptIdProperty, "S2"))); + + publisher.Published.Count(e => e.ScriptId == "S1").ShouldBe(2, "the limit is per source"); + publisher.Published.Count(e => e.ScriptId == "S2").ShouldBe(1, "another source is unaffected"); + + clock.Advance(TimeSpan.FromMinutes(1)); + sink.Emit(Evt(LogEventLevel.Warning, "chatty again", (ScriptLoggerFactory.ScriptIdProperty, "S1"), (ScriptLoggerFactory.VirtualTagIdProperty, "V1"))); + + var s1 = publisher.Published.Where(e => e.ScriptId == "S1").ToList(); + s1.Count.ShouldBe(4, "2 admitted + 1 summary + 1 admitted in the new window"); + var summary = s1[2]; + summary.Message.ShouldContain("3"); + summary.Message.ShouldContain("suppressed"); + summary.VirtualTagId.ShouldBe("V1", "the summary carries the source's identity so the page attributes it"); + summary.Level.ShouldBe("Warning"); + s1[3].Message.ShouldBe("chatty again"); + } + + private sealed class CapturingSink : ILogEventSink + { + public List Events { get; } = []; + public void Emit(LogEvent logEvent) => Events.Add(logEvent); + } + + [Fact] + public void Companion_sink_rate_limits_per_script_and_mirrors_a_summary_line() + { + var clock = new Clock(); + var main = new CapturingSink(); + var mainLogger = new LoggerConfiguration().MinimumLevel.Verbose().WriteTo.Sink(main).CreateLogger(); + var sink = new ScriptLogCompanionSink(mainLogger, LogEventLevel.Error, + new ScriptLogRateLimiter(limitPerWindow: 2, clock: () => clock.Now)); + + for (var i = 0; i < 5; i++) + sink.Emit(Evt(LogEventLevel.Error, "boom", (ScriptLoggerFactory.ScriptNameProperty, "HighTemp"))); + main.Events.Count.ShouldBe(2); + + clock.Advance(TimeSpan.FromMinutes(1)); + sink.Emit(Evt(LogEventLevel.Error, "boom", (ScriptLoggerFactory.ScriptNameProperty, "HighTemp"))); + + main.Events.Count.ShouldBe(4); + var summary = main.Events[2].RenderMessage(); + summary.ShouldContain("HighTemp"); + summary.ShouldContain("3"); + summary.ShouldContain("suppressed"); + } +}