diff --git a/docs/requirements/Component-DeploymentManager.md b/docs/requirements/Component-DeploymentManager.md index f13de138..d8b0720f 100644 --- a/docs/requirements/Component-DeploymentManager.md +++ b/docs/requirements/Component-DeploymentManager.md @@ -270,7 +270,12 @@ window the table grows for the life of the system and every unfiltered deploymen query degrades with age rather than with page size. `TerminalDeploymentRecordRetention` (default **365 days**, deliberately generous because deployment history is operator forensics) bounds it, swept opportunistically -and rate-limited on the deployment-list read path. +and rate-limited on the deployment-list read path: at most one sweep per 6 hours per +central node, enforced by the DI-singleton `TerminalPurgeRateLimiter` (a +`TimeProvider`-driven compare-and-swap claim shared by every scoped +`DeploymentService`; Gitea #90 — formerly a process-wide static). A failed sweep is +logged as a warning and swallowed — it never fails the operator's query — and still +consumes the interval, so the next attempt waits the full 6 hours. Only **terminal** rows (`Success` / `Failed`) are eligible. An `InProgress` row is never purged regardless of age: it is precisely the row the query-before-redeploy diff --git a/src/ZB.MOM.WW.ScadaBridge.DeploymentManager/DeploymentService.cs b/src/ZB.MOM.WW.ScadaBridge.DeploymentManager/DeploymentService.cs index 4b72c434..23f32ca8 100644 --- a/src/ZB.MOM.WW.ScadaBridge.DeploymentManager/DeploymentService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.DeploymentManager/DeploymentService.cs @@ -81,17 +81,11 @@ public class DeploymentService /// Batches deleted per opportunistic sweep, bounding one caller's latency cost. private const int TerminalPurgeBatchesPerSweep = 4; - /// Minimum interval between opportunistic terminal-record sweeps. - private static readonly TimeSpan TerminalPurgeMinInterval = TimeSpan.FromHours(6); - /// - /// Ticks of the last opportunistic terminal-record sweep, process-wide. - /// is scoped, so this must be static to - /// rate-limit across requests; 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). /// - private static long _lastTerminalPurgeTicks; + private readonly TerminalPurgeRateLimiter _purgeRateLimiter; /// /// Initializes a new instance of with all required dependencies. @@ -136,7 +130,8 @@ public class DeploymentService IDeploymentStatusNotifier statusNotifier, IOptions options, IOptions communicationOptions, - ILogger logger) + ILogger 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(); } /// @@ -1530,7 +1528,8 @@ public class DeploymentService /// /// Opportunistic retention sweep for TERMINAL deployment records and system /// artifact deployment records, rate-limited - /// to once every per process and bounded + /// to once every per process (via the + /// singleton ) and bounded /// to batches per call. /// /// @@ -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; diff --git a/src/ZB.MOM.WW.ScadaBridge.DeploymentManager/ServiceCollectionExtensions.cs b/src/ZB.MOM.WW.ScadaBridge.DeploymentManager/ServiceCollectionExtensions.cs index fd649372..297fccc8 100644 --- a/src/ZB.MOM.WW.ScadaBridge.DeploymentManager/ServiceCollectionExtensions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.DeploymentManager/ServiceCollectionExtensions.cs @@ -55,6 +55,11 @@ public static class ServiceCollectionExtensions // it instead of polling the database every 10 seconds. services.AddSingleton(); + // 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(); + services.AddScoped(); services.AddScoped(); services.AddScoped(); diff --git a/src/ZB.MOM.WW.ScadaBridge.DeploymentManager/TerminalPurgeRateLimiter.cs b/src/ZB.MOM.WW.ScadaBridge.DeploymentManager/TerminalPurgeRateLimiter.cs new file mode 100644 index 00000000..ea156e13 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.DeploymentManager/TerminalPurgeRateLimiter.cs @@ -0,0 +1,55 @@ +namespace ZB.MOM.WW.ScadaBridge.DeploymentManager; + +/// +/// Rate limiter for 's +/// opportunistic retention sweep: at most one sweep per . +/// +/// +/// is scoped (one per request), so the limiter must outlive it +/// to rate-limit across requests. It used to be a process-wide static field, which made +/// the sweep untestable beyond a single test per assembly (Gitea #90); it is now a DI singleton +/// registered by , driven by an +/// injectable , so each test owns its own limiter and clock. +/// +/// +public sealed class TerminalPurgeRateLimiter +{ + /// Minimum interval between opportunistic terminal-record sweeps. + public static readonly TimeSpan MinInterval = TimeSpan.FromHours(6); + + private readonly TimeProvider _timeProvider; + + /// Ticks of the last claimed sweep; 0 = never swept. + private long _lastSweepTicks; + + /// Initializes a limiter on the system clock. + public TerminalPurgeRateLimiter() + : this(TimeProvider.System) + { + } + + /// Initializes a limiter on the given clock. + /// The clock the interval is measured on. + public TerminalPurgeRateLimiter(TimeProvider timeProvider) + { + ArgumentNullException.ThrowIfNull(timeProvider); + _timeProvider = timeProvider; + } + + /// + /// Claims the next sweep if 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 rather than running a redundant pass. A claimed sweep + /// that then fails still counts: the next attempt waits a full interval. + /// + /// The clock reading the claim was made at (the sweep's cutoff basis). + /// when the caller should sweep now. + 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; + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.DeploymentManager.Tests/DeploymentServiceTests.cs b/tests/ZB.MOM.WW.ScadaBridge.DeploymentManager.Tests/DeploymentServiceTests.cs index 6a75c534..e226e85e 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.DeploymentManager.Tests/DeploymentServiceTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.DeploymentManager.Tests/DeploymentServiceTests.cs @@ -90,8 +90,8 @@ public class DeploymentServiceTests : TestKit /// /// The retention sweep also purges system artifact deployment records — one per Deploy /// Artifacts run, never read back — with the same cutoff as terminal deployment records. - /// NOTE: the sweep is rate-limited by a process-wide static, so this is the only test in - /// the assembly that may drive . + /// The rate limiter is per service unless injected (Gitea #90); its behaviour and the sweep's + /// swallowed-failure path are covered by TerminalPurgeRateLimiterTests. /// /// A task that represents the asynchronous operation. [Fact] diff --git a/tests/ZB.MOM.WW.ScadaBridge.DeploymentManager.Tests/TerminalPurgeRateLimiterTests.cs b/tests/ZB.MOM.WW.ScadaBridge.DeploymentManager.Tests/TerminalPurgeRateLimiterTests.cs new file mode 100644 index 00000000..bd859b38 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.DeploymentManager.Tests/TerminalPurgeRateLimiterTests.cs @@ -0,0 +1,229 @@ +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using NSubstitute; +using NSubstitute.ExceptionExtensions; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; +using ZB.MOM.WW.ScadaBridge.Communication; +using ZB.MOM.WW.ScadaBridge.TemplateEngine.Flattening; + +namespace ZB.MOM.WW.ScadaBridge.DeploymentManager.Tests; + +/// +/// Gitea #90: the terminal-record purge rate limiter is an injectable, clock-driven singleton +/// instead of a process-wide static, so the sweep's rate limit and its swallow-on-failure path +/// can be tested independently of every other test in the assembly. +/// +public class TerminalPurgeRateLimiterTests +{ + private static readonly DateTimeOffset Start = new(2026, 11, 9, 12, 0, 0, TimeSpan.Zero); + + private sealed class ManualClock(DateTimeOffset start) : TimeProvider + { + public DateTimeOffset Now { get; set; } = start; + + public override DateTimeOffset GetUtcNow() => Now; + } + + private sealed class CapturingLogger : ILogger + { + public List<(LogLevel Level, Exception? Exception, string Message)> Entries { get; } = new(); + + public IDisposable? BeginScope(TState state) where TState : notnull => null; + + public bool IsEnabled(LogLevel logLevel) => true; + + public void Log( + LogLevel logLevel, EventId eventId, TState state, Exception? exception, + Func formatter) + { + lock (Entries) Entries.Add((logLevel, exception, formatter(state, exception))); + } + } + + private static DeploymentService CreateService( + IDeploymentManagerRepository repo, + TerminalPurgeRateLimiter? limiter, + ILogger? logger = null) => + new( + repo, + Substitute.For(), + Substitute.For(), + Substitute.For(), + new CommunicationService( + Options.Create(new CommunicationOptions()), NullLogger.Instance), + new OperationLockManager(), + Substitute.For(), + new DiffService(), + new RevisionHashService(), + new DeploymentStatusNotifier(NullLogger.Instance), + Options.Create(new DeploymentManagerOptions()), + Options.Create(new CommunicationOptions()), + logger ?? NullLogger.Instance, + limiter); + + private static IDeploymentManagerRepository PurgingRepo(int terminal = 1, int artifact = 3) + { + var repo = Substitute.For(); + repo.PurgeTerminalDeploymentRecordsAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(terminal); + repo.PurgeSystemArtifactDeploymentRecordsAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(artifact); + return repo; + } + + /// The first claim wins; a second inside the interval loses; after it, a claim wins again. + [Fact] + public void TryClaim_OncePerInterval_OnTheInjectedClock() + { + var clock = new ManualClock(Start); + var limiter = new TerminalPurgeRateLimiter(clock); + + Assert.True(limiter.TryClaim(out var first)); + Assert.Equal(Start, first); + + clock.Now = Start + TerminalPurgeRateLimiter.MinInterval - TimeSpan.FromSeconds(1); + Assert.False(limiter.TryClaim(out _)); + + clock.Now = Start + TerminalPurgeRateLimiter.MinInterval; + Assert.True(limiter.TryClaim(out var later)); + Assert.Equal(clock.Now, later); + Assert.False(limiter.TryClaim(out _)); + } + + /// Of many concurrent claimants exactly one wins. + /// A task that represents the asynchronous operation. + [Fact] + public async Task TryClaim_ConcurrentCallers_ExactlyOneWins() + { + var limiter = new TerminalPurgeRateLimiter(new ManualClock(Start)); + using var gate = new Barrier(16); + var wins = await Task.WhenAll(Enumerable.Range(0, 16).Select(_ => Task.Run(() => + { + gate.SignalAndWait(); + return limiter.TryClaim(out var _); + }))); + + Assert.Equal(1, wins.Count(w => w)); + } + + /// + /// The limiter is per instance, not process-wide: two services with their own limiters both + /// sweep, and two sharing one (as the DI singleton makes them) sweep once. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task Sweep_RateLimitIsPerLimiter_NotProcessWide() + { + var clock = new ManualClock(Start); + + Assert.Equal(4, await CreateService(PurgingRepo(), new TerminalPurgeRateLimiter(clock)) + .TryPurgeTerminalDeploymentRecordsAsync()); + Assert.Equal(4, await CreateService(PurgingRepo(), new TerminalPurgeRateLimiter(clock)) + .TryPurgeTerminalDeploymentRecordsAsync()); + + var shared = new TerminalPurgeRateLimiter(clock); + var secondRepo = PurgingRepo(); + Assert.Equal(4, await CreateService(PurgingRepo(), shared).TryPurgeTerminalDeploymentRecordsAsync()); + Assert.Equal(0, await CreateService(secondRepo, shared).TryPurgeTerminalDeploymentRecordsAsync()); + await secondRepo.DidNotReceiveWithAnyArgs().PurgeTerminalDeploymentRecordsAsync(default, default, default); + } + + /// + /// The sweep's cutoff is taken from the injected clock: now minus the retention window. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task Sweep_CutoffIsInjectedClockMinusRetention() + { + var repo = PurgingRepo(); + await CreateService(repo, new TerminalPurgeRateLimiter(new ManualClock(Start))) + .TryPurgeTerminalDeploymentRecordsAsync(); + + var cutoff = Start - new DeploymentManagerOptions().TerminalDeploymentRecordRetention; + await repo.Received(1).PurgeTerminalDeploymentRecordsAsync(cutoff, Arg.Any(), Arg.Any()); + await repo.Received(1).PurgeSystemArtifactDeploymentRecordsAsync(cutoff, Arg.Any(), Arg.Any()); + } + + /// + /// A failing sweep never fails the caller: the exception is logged as a warning and swallowed, + /// the rows already purged are still reported, and the failed sweep still consumes the + /// interval — the next call inside it does not retry, the first call after it does. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task Sweep_Failure_IsLoggedAndSwallowed_AndWaitsOutTheInterval() + { + var clock = new ManualClock(Start); + var limiter = new TerminalPurgeRateLimiter(clock); + var logger = new CapturingLogger(); + var boom = new InvalidOperationException("deadlock victim"); + var repo = Substitute.For(); + repo.PurgeTerminalDeploymentRecordsAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(2); + repo.PurgeSystemArtifactDeploymentRecordsAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .ThrowsAsync(boom); + var service = CreateService(repo, limiter, logger); + + var purged = await service.TryPurgeTerminalDeploymentRecordsAsync(); + + Assert.Equal(2, purged); // the terminal batch that landed before the failure + var warning = Assert.Single(logger.Entries, e => e.Level == LogLevel.Warning); + Assert.Same(boom, warning.Exception); + Assert.Contains("retention sweep failed", warning.Message, StringComparison.Ordinal); + Assert.DoesNotContain(logger.Entries, e => e.Level == LogLevel.Information); + + // Inside the interval: rate-limited, no retry. + repo.ClearReceivedCalls(); + clock.Now = Start + TimeSpan.FromHours(1); + Assert.Equal(0, await service.TryPurgeTerminalDeploymentRecordsAsync()); + await repo.DidNotReceiveWithAnyArgs().PurgeTerminalDeploymentRecordsAsync(default, default, default); + + // After the interval: the sweep runs again (and fails again, still swallowed). + clock.Now = Start + TerminalPurgeRateLimiter.MinInterval; + Assert.Equal(2, await service.TryPurgeTerminalDeploymentRecordsAsync()); + await repo.ReceivedWithAnyArgs(1).PurgeTerminalDeploymentRecordsAsync(default, default, default); + Assert.Equal(2, logger.Entries.Count(e => e.Level == LogLevel.Warning)); + } + + /// A failure on the very first batch reports zero and is swallowed. + /// A task that represents the asynchronous operation. + [Fact] + public async Task Sweep_FirstBatchFailure_ReturnsZero_DoesNotThrow() + { + var repo = Substitute.For(); + repo.PurgeTerminalDeploymentRecordsAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .ThrowsAsync(new TimeoutException("SQL timeout")); + var logger = new CapturingLogger(); + + var purged = await CreateService(repo, new TerminalPurgeRateLimiter(new ManualClock(Start)), logger) + .TryPurgeTerminalDeploymentRecordsAsync(); + + Assert.Equal(0, purged); + Assert.IsType(Assert.Single(logger.Entries, e => e.Level == LogLevel.Warning).Exception); + await repo.DidNotReceiveWithAnyArgs().PurgeSystemArtifactDeploymentRecordsAsync(default, default, default); + } + + /// + /// AddDeploymentManager registers the limiter as a singleton, so every scoped DeploymentService + /// shares one rate limit. + /// + [Fact] + public void AddDeploymentManager_RegistersLimiterAsSingleton() + { + var services = new ServiceCollection(); + services.AddDeploymentManager(); + + var descriptor = Assert.Single(services, d => d.ServiceType == typeof(TerminalPurgeRateLimiter)); + Assert.Equal(ServiceLifetime.Singleton, descriptor.Lifetime); + + using var provider = services.BuildServiceProvider(); + using var a = provider.CreateScope(); + using var b = provider.CreateScope(); + Assert.Same( + a.ServiceProvider.GetRequiredService(), + b.ServiceProvider.GetRequiredService()); + } +}