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
///
private long _bufferedCount;
+ ///
+ /// 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
+ /// , so the seed taken at drifts
+ /// (and goes negative after a failover). Every th
+ /// sweep re-reads the Pending count from the database.
+ ///
+ private int _sweepsSinceReseed = GaugeReseedEverySweeps - 1; // first sweep re-seeds
+
+ private const int GaugeReseedEverySweeps = 6;
+
///
/// Test seam: simulates a concurrent pre-seed
/// increment landing on
@@ -477,7 +488,7 @@ public class StoreAndForwardService
if (Interlocked.CompareExchange(ref _queueDepthProviderRegistered, 1, 0) == 0)
{
Interlocked.Add(ref _bufferedCount, pending);
- var provider = (Func)(() => Interlocked.Read(ref _bufferedCount));
+ var provider = (Func)(() => 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;
}
+ ///
+ /// 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.
+ ///
+ /// A task that represents the asynchronous operation.
+ [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());
+ }
+
/// Verifies that gauge, tracks buffered depth, across enqueue drain and park.
/// A task that represents the asynchronous operation.
[Fact]