refactor(deployment): injectable terminal-purge rate limiter; test the swallowed failure (ScadaBridge#90)

DeploymentService.TryPurgeTerminalDeploymentRecordsAsync rate-limited its
opportunistic retention sweep with a process-wide static tick field, so only
one test in the assembly could drive the sweep and its swallow-on-failure path
was untested.

The limit now lives in TerminalPurgeRateLimiter: a TimeProvider-driven
compare-and-swap claim (MinInterval 6 h) registered by AddDeploymentManager as
a DI singleton, so every scoped DeploymentService still shares one limit per
node. DeploymentService takes it as an optional trailing constructor parameter;
a hand-built instance gets its own limiter. The sweep's cutoff now comes from
the same clock. Behaviour is otherwise unchanged.

Evidence (TerminalPurgeRateLimiterTests):
- claim once per interval on an injected clock; 16 concurrent claimants ->
  exactly one wins;
- two services with their own limiters both sweep, two sharing one sweep once;
- cutoff = injected now - retention;
- a failing artifact batch is logged as a warning with the exception and
  swallowed, the already-purged terminal rows are still reported, the next
  call inside the interval does not retry and the first call after it does;
  a first-batch failure returns 0 and skips the artifact purge;
- AddDeploymentManager registers the limiter as a singleton shared across scopes.
Mutation-checked: a process-wide static limiter fails 5 tests (incl. the
existing artifact-purge test); narrowing the catch fails both failure tests.

Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
This commit is contained in:
Joseph Doherty
2026-11-09 13:20:46 -05:00
parent 6070cc2b9d
commit 51d2f6c906
6 changed files with 311 additions and 23 deletions
@@ -81,17 +81,11 @@ public class DeploymentService
/// <summary>Batches deleted per opportunistic sweep, bounding one caller's latency cost.</summary>
private const int TerminalPurgeBatchesPerSweep = 4;
/// <summary>Minimum interval between opportunistic terminal-record sweeps.</summary>
private static readonly TimeSpan TerminalPurgeMinInterval = TimeSpan.FromHours(6);
/// <summary>
/// Ticks of the last opportunistic terminal-record sweep, process-wide.
/// <see cref="DeploymentService"/> is scoped, so this must be static to
/// rate-limit across requests; <see cref="Interlocked"/> makes the
/// claim-the-sweep step a compare-and-swap so concurrent readers do not all
/// sweep at once.
/// Rate limiter for the opportunistic terminal-record sweep. A DI singleton shared by every
/// scoped instance (Gitea #90 — it was a process-wide static).
/// </summary>
private static long _lastTerminalPurgeTicks;
private readonly TerminalPurgeRateLimiter _purgeRateLimiter;
/// <summary>
/// Initializes a new instance of <see cref="DeploymentService"/> with all required dependencies.
@@ -136,7 +130,8 @@ public class DeploymentService
IDeploymentStatusNotifier statusNotifier,
IOptions<DeploymentManagerOptions> options,
IOptions<CommunicationOptions> communicationOptions,
ILogger<DeploymentService> logger)
ILogger<DeploymentService> logger,
TerminalPurgeRateLimiter? purgeRateLimiter = null)
{
_repository = repository;
_siteRepository = siteRepository;
@@ -151,6 +146,9 @@ public class DeploymentService
_options = options.Value;
_commOptions = communicationOptions.Value;
_logger = logger;
// AddDeploymentManager registers the shared singleton; a hand-built instance
// (tests, tools) gets its own limiter.
_purgeRateLimiter = purgeRateLimiter ?? new TerminalPurgeRateLimiter();
}
/// <summary>
@@ -1530,7 +1528,8 @@ public class DeploymentService
/// <summary>
/// Opportunistic retention sweep for TERMINAL deployment records and system
/// artifact deployment records, rate-limited
/// to once every <see cref="TerminalPurgeMinInterval"/> per process and bounded
/// to once every <see cref="TerminalPurgeRateLimiter.MinInterval"/> per process (via the
/// singleton <see cref="TerminalPurgeRateLimiter"/>) and bounded
/// to <see cref="TerminalPurgeBatchesPerSweep"/> batches per call.
///
/// <para>
@@ -1554,14 +1553,9 @@ public class DeploymentService
if (retention <= TimeSpan.Zero)
return 0;
var now = DateTimeOffset.UtcNow;
var last = Interlocked.Read(ref _lastTerminalPurgeTicks);
if (last != 0 && now.UtcTicks - last < TerminalPurgeMinInterval.Ticks)
return 0;
// Claim the sweep. A loser of this CAS returns immediately rather than
// running a redundant second pass.
if (Interlocked.CompareExchange(ref _lastTerminalPurgeTicks, now.UtcTicks, last) != last)
// Claim the sweep. Within the interval, or a loser of a concurrent claim,
// returns immediately rather than running a redundant pass.
if (!_purgeRateLimiter.TryClaim(out var now))
return 0;
var cutoff = now - retention;
@@ -1604,7 +1598,7 @@ public class DeploymentService
{
_logger.LogWarning(ex,
"Terminal deployment-record retention sweep failed; retrying after {Interval}.",
TerminalPurgeMinInterval);
TerminalPurgeRateLimiter.MinInterval);
}
return purged;
@@ -55,6 +55,11 @@ public static class ServiceCollectionExtensions
// it instead of polling the database every 10 seconds.
services.AddSingleton<IDeploymentStatusNotifier, DeploymentStatusNotifier>();
// Rate limiter for DeploymentService's opportunistic terminal-record retention
// sweep. Singleton because DeploymentService is scoped and the limit must hold
// across requests (Gitea #90 — formerly a process-wide static field).
services.TryAddSingleton<TerminalPurgeRateLimiter>();
services.AddScoped<IFlatteningPipeline, FlatteningPipeline>();
services.AddScoped<DeploymentService>();
services.AddScoped<ArtifactDeploymentService>();
@@ -0,0 +1,55 @@
namespace ZB.MOM.WW.ScadaBridge.DeploymentManager;
/// <summary>
/// Rate limiter for <see cref="DeploymentService.TryPurgeTerminalDeploymentRecordsAsync"/>'s
/// opportunistic retention sweep: at most one sweep per <see cref="MinInterval"/>.
///
/// <para>
/// <see cref="DeploymentService"/> is scoped (one per request), so the limiter must outlive it
/// to rate-limit across requests. It used to be a process-wide <c>static</c> field, which made
/// the sweep untestable beyond a single test per assembly (Gitea #90); it is now a DI singleton
/// registered by <see cref="ServiceCollectionExtensions.AddDeploymentManager"/>, driven by an
/// injectable <see cref="TimeProvider"/>, so each test owns its own limiter and clock.
/// </para>
/// </summary>
public sealed class TerminalPurgeRateLimiter
{
/// <summary>Minimum interval between opportunistic terminal-record sweeps.</summary>
public static readonly TimeSpan MinInterval = TimeSpan.FromHours(6);
private readonly TimeProvider _timeProvider;
/// <summary>Ticks of the last claimed sweep; 0 = never swept.</summary>
private long _lastSweepTicks;
/// <summary>Initializes a limiter on the system clock.</summary>
public TerminalPurgeRateLimiter()
: this(TimeProvider.System)
{
}
/// <summary>Initializes a limiter on the given clock.</summary>
/// <param name="timeProvider">The clock the interval is measured on.</param>
public TerminalPurgeRateLimiter(TimeProvider timeProvider)
{
ArgumentNullException.ThrowIfNull(timeProvider);
_timeProvider = timeProvider;
}
/// <summary>
/// Claims the next sweep if <see cref="MinInterval"/> has elapsed since the last claimed one.
/// The claim is a compare-and-swap, so of several concurrent callers exactly one wins and the
/// rest return <see langword="false"/> rather than running a redundant pass. A claimed sweep
/// that then fails still counts: the next attempt waits a full interval.
/// </summary>
/// <param name="now">The clock reading the claim was made at (the sweep's cutoff basis).</param>
/// <returns><see langword="true"/> when the caller should sweep now.</returns>
public bool TryClaim(out DateTimeOffset now)
{
now = _timeProvider.GetUtcNow();
var last = Interlocked.Read(ref _lastSweepTicks);
if (last != 0 && now.UtcTicks - last < MinInterval.Ticks)
return false;
return Interlocked.CompareExchange(ref _lastSweepTicks, now.UtcTicks, last) == last;
}
}