Merge branch 'worktree-agent-a71052048b4ee11d4' into arch-review-remediation

src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs

@@ -148,13 +148,85 @@ public class StoreAndForwardService
     /// Cumulative count of cached-call audit-observer notifications dropped because
     /// <see cref="_observerQueue"/> was at capacity (WP2.6c). Not reset across
     /// <see cref="StartAsync"/>/<see cref="StopAsync"/> cycles — a diagnostic total for
-    /// the lifetime of this service instance.
+    /// the lifetime of this service instance. Incremented unconditionally on every drop
+    /// by <see cref="LogObserverQueueDrop"/>, independent of that method's log throttling.
     /// </summary>
     private long _observerQueueDroppedCount;
 
     /// <summary>Diagnostic counter — see <see cref="_observerQueueDroppedCount"/>.</summary>
     public long ObserverQueueDroppedCount => Interlocked.Read(ref _observerQueueDroppedCount);
 
+    /// <summary>
+    /// How often a sustained run of observer-queue drops re-logs after the first Warning
+    /// of an episode (arch-review adversarial finding F3). Without this, a stuck observer
+    /// with a large <see cref="StoreAndForwardOptions.ObserverQueueCapacity"/> floods the
+    /// log at sweep rate — one Warning per dropped item. Only the LOGGING is throttled;
+    /// <see cref="_observerQueueDroppedCount"/> still counts every drop.
+    /// </summary>
+    private static readonly TimeSpan ObserverQueueDropLogRollupInterval = TimeSpan.FromMinutes(1);
+
+    /// <summary>
+    /// <see cref="Environment.TickCount64"/> of the last observer-queue-drop Warning, or
+    /// -1 (its initial value — <c>TickCount64</c> is never negative) if none has been
+    /// logged yet this service-instance lifetime. Not reset across
+    /// <see cref="StartAsync"/>/<see cref="StopAsync"/> cycles, matching
+    /// <see cref="_observerQueueDroppedCount"/>.
+    /// </summary>
+    private long _observerQueueLastDropLogTicks = -1;
+
+    /// <summary>
+    /// Drops accumulated since <see cref="_observerQueueLastDropLogTicks"/> was last
+    /// logged — the count a rollup Warning reports before resetting to 0.
+    /// </summary>
+    private long _observerQueueDroppedSinceLastLog;
+
+    /// <summary>
+    /// Records one observer-queue drop and logs about it: a Warning for the FIRST drop of
+    /// an episode, then at most one rollup Warning per
+    /// <see cref="ObserverQueueDropLogRollupInterval"/> while drops keep happening — never
+    /// one Warning per dropped item (arch-review adversarial finding F3). Safe to call
+    /// concurrently: the log slot for an episode is claimed via a CAS on
+    /// <see cref="_observerQueueLastDropLogTicks"/>, so overlapping droppers accumulate
+    /// into <see cref="_observerQueueDroppedSinceLastLog"/> without double-logging.
+    /// </summary>
+    private void LogObserverQueueDrop()
+    {
+        var total = Interlocked.Increment(ref _observerQueueDroppedCount);
+        Interlocked.Increment(ref _observerQueueDroppedSinceLastLog);
+
+        var now = Environment.TickCount64;
+        var lastLog = Interlocked.Read(ref _observerQueueLastDropLogTicks);
+        var isFirstEverDrop = lastLog < 0;
+        var dueForRollup = !isFirstEverDrop
+            && now - lastLog >= (long)ObserverQueueDropLogRollupInterval.TotalMilliseconds;
+
+        if (!isFirstEverDrop && !dueForRollup)
+            return;
+
+        // Claims the log slot for this episode; a concurrent caller that loses the CAS
+        // simply leaves its increment above in the rollup's next count instead of logging.
+        if (Interlocked.CompareExchange(ref _observerQueueLastDropLogTicks, now, lastLog) != lastLog)
+            return;
+
+        var countSinceLog = Interlocked.Exchange(ref _observerQueueDroppedSinceLastLog, 0);
+        if (isFirstEverDrop)
+        {
+            _logger.LogWarning(
+                "Cached-call audit-observer queue exceeded its bounded capacity ({Capacity}); " +
+                "oldest pending notification dropped (total dropped: {Dropped})",
+                _options.ObserverQueueCapacity, total);
+        }
+        else
+        {
+            _logger.LogWarning(
+                "Cached-call audit-observer queue still exceeding its bounded capacity " +
+                "({Capacity}); {DroppedSinceLastLog} oldest pending notifications dropped in " +
+                "the last {IntervalMinutes} minute(s) (total dropped: {Dropped})",
+                _options.ObserverQueueCapacity, countSinceLog,
+                ObserverQueueDropLogRollupInterval.TotalMinutes, total);
+        }
+    }
+
     /// <summary>
     /// Builds a bounded, single-reader observer queue with DropOldest overflow, invoking
     /// <paramref name="onDropped"/> (if supplied) on every eviction.
@@ -417,14 +489,12 @@ public class StoreAndForwardService
         // WP2.6c: bounded + DropOldest, sized from options; a drop increments
         // _observerQueueDroppedCount (surfaced via ObserverQueueDroppedCount) and is
         // logged at Warning so a stuck observer is visible, not just silently lossy.
-        _observerQueue = CreateObserverQueue(_options.ObserverQueueCapacity, onDropped: () =>
-        {
-            Interlocked.Increment(ref _observerQueueDroppedCount);
-            _logger.LogWarning(
-                "Cached-call audit-observer queue exceeded its bounded capacity ({Capacity}); " +
-                "oldest pending notification dropped (total dropped: {Dropped})",
-                _options.ObserverQueueCapacity, Interlocked.Read(ref _observerQueueDroppedCount));
-        });
+        // F3: logging itself is rate-limited by LogObserverQueueDrop (first-drop Warning
+        // + a rollup at most once per ObserverQueueDropLogRollupInterval) — otherwise a
+        // stuck observer with a large queue floods the log at sweep rate, one Warning per
+        // dropped item. The counter is unaffected by that throttling.
+        _observerQueue = CreateObserverQueue(
+            _options.ObserverQueueCapacity, onDropped: LogObserverQueueDrop);
         _observerPump = Task.Run(async () =>
         {
             await foreach (var work in _observerQueue.Reader.ReadAllAsync())