From d3cba156994664453f18da022114e2eaa351c1a9 Mon Sep 17 00:00:00 2001
From: Joseph Doherty <dohejw01@gmail.com>
Date: Thu, 22 Oct 2026 08:28:40 -0400
Subject: [PATCH] fix(store-and-forward): re-seed queue-depth gauge from DB on
 sweep, clamp at 0 (review 2026-10-21 02/C3)

---
 .../StoreAndForwardService.cs                 | 30 +++++++++++++++++-
 .../QueueDepthGaugeTests.cs                   | 31 +++++++++++++++++++
 2 files changed, 60 insertions(+), 1 deletion(-)

diff --git a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs
index f2ea0a85..66c9532f 100644
--- a/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs
+++ b/src/ZB.MOM.WW.ScadaBridge.StoreAndForward/StoreAndForwardService.cs
@@ -283,6 +283,17 @@ public class StoreAndForwardService
     /// </summary>
     private long _bufferedCount;
 
+    /// <summary>
+    /// Sweeps run since the last gauge re-seed. Review 2026-10-21 02/C3: rows that arrive by
+    /// LocalDb CDC (standby) or are retried/parked/discarded on the peer never touch
+    /// <see cref="_bufferedCount"/>, so the seed taken at <see cref="StartAsync"/> drifts
+    /// (and goes negative after a failover). Every <see cref="GaugeReseedEverySweeps"/>th
+    /// sweep re-reads the Pending count from the database.
+    /// </summary>
+    private int _sweepsSinceReseed = GaugeReseedEverySweeps - 1; // first sweep re-seeds
+
+    private const int GaugeReseedEverySweeps = 6;
+
     /// <summary>
     /// Test seam: simulates a concurrent pre-seed
     /// <see cref="BufferAsync"/> increment landing on <see cref="_bufferedCount"/>
@@ -477,7 +488,7 @@ public class StoreAndForwardService
         if (Interlocked.CompareExchange(ref _queueDepthProviderRegistered, 1, 0) == 0)
         {
             Interlocked.Add(ref _bufferedCount, pending);
-            var provider = (Func<long>)(() => Interlocked.Read(ref _bufferedCount));
+            var provider = (Func<long>)(() => Math.Max(0, Interlocked.Read(ref _bufferedCount)));
             _queueDepthProvider = provider;
             ScadaBridgeTelemetry.SetQueueDepthProvider(provider);
         }
@@ -831,10 +842,27 @@ public class StoreAndForwardService
         }
     }
 
+    private async Task ReseedBufferedCountAsync()
+    {
+        try
+        {
+            if (++_sweepsSinceReseed < GaugeReseedEverySweeps) return;
+            _sweepsSinceReseed = 0;
+            var pending = await _storage.GetMessageCountByStatusAsync(StoreAndForwardMessageStatus.Pending);
+            Interlocked.Exchange(ref _bufferedCount, Math.Max(0, pending));
+        }
+        catch (Exception ex)
+        {
+            _logger.LogDebug(ex, "S&F queue-depth gauge re-seed failed; keeping the cached count");
+        }
+    }
+
     private async Task RunSweepOwnedAsync()
     {
         try
         {
+            await ReseedBufferedCountAsync();
+
             var gate = _deliveryGate;
             if (gate != null)
             {
diff --git a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/QueueDepthGaugeTests.cs b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/QueueDepthGaugeTests.cs
index f5d3e801..a3b8ba35 100644
--- a/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/QueueDepthGaugeTests.cs
+++ b/tests/ZB.MOM.WW.ScadaBridge.StoreAndForward.Tests/QueueDepthGaugeTests.cs
@@ -97,6 +97,37 @@ public class QueueDepthGaugeTests : IAsyncLifetime, IDisposable
         return observed;
     }
 
+    /// <summary>
+    /// Review 2026-10-21 02/C3: rows that arrive without passing through this node's counter
+    /// paths (LocalDb CDC on a standby) are picked up by the sweep's periodic re-seed.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task Gauge_ReseedsFromDatabase_OnSweep()
+    {
+        Assert.Equal(0, ReadQueueDepthGauge());
+        for (var i = 0; i < 3; i++)
+        {
+            await _storage.EnqueueAsync(new StoreAndForwardMessage
+            {
+                Id = $"cdc{i}",
+                Category = StoreAndForwardCategory.ExternalSystem,
+                Target = "t",
+                PayloadJson = "{}",
+                MaxRetries = 50,
+                RetryIntervalMs = 3_600_000,
+                CreatedAt = DateTimeOffset.UtcNow,
+                LastAttemptAt = DateTimeOffset.UtcNow,
+                Status = StoreAndForwardMessageStatus.Pending,
+            });
+        }
+        Assert.Equal(0, ReadQueueDepthGauge()); // counter never saw them
+
+        await _service.RetryPendingMessagesAsync();
+
+        Assert.Equal(3, ReadQueueDepthGauge());
+    }
+
     /// <summary>Verifies that gauge, tracks buffered depth, across enqueue drain and park.</summary>
     /// <returns>A task that represents the asynchronous operation.</returns>
     [Fact]