fix(store-and-forward): re-seed queue-depth gauge from DB on sweep, clamp at 0 (review 2026-10-21 02/C3)

This commit is contained in:
Joseph Doherty
2026-10-22 08:28:40 -04:00
parent 76f7960a06
commit d3cba15699
2 changed files with 60 additions and 1 deletions
@@ -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)
{
@@ -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]