diff --git a/src/ServiceControl.AcceptanceTests/WebApi/When_triggering_a_manual_retention_sweep.cs b/src/ServiceControl.AcceptanceTests/WebApi/When_triggering_a_manual_retention_sweep.cs new file mode 100644 index 0000000000..4b88c00bf2 --- /dev/null +++ b/src/ServiceControl.AcceptanceTests/WebApi/When_triggering_a_manual_retention_sweep.cs @@ -0,0 +1,160 @@ +namespace ServiceControl.AcceptanceTests.WebApi; + +using System; +using System.Net; +using System.Net.Http; +using System.Net.Http.Json; +using System.Text.Json; +using System.Threading; +using System.Threading.Tasks; +using AcceptanceTesting; +using NServiceBus.AcceptanceTesting; +using NUnit.Framework; +using ServiceControl.Api.Contracts; + +class When_triggering_a_manual_retention_sweep : AcceptanceTest +{ + [Test] + public async Task Should_be_available_on_efcore_persisters() + { + if (StorageConfiguration.PersistenceType == "RavenDB") + { + Assert.Ignore("RavenDB has no sweeper — covered by Should_return_501_on_a_ravendb_backed_instance."); + return; + } + + HttpStatusCode started = default; + HttpStatusCode invalidCutoff = default; + + await Define() + .Done(async _ => + { + // Trigger a sweep with a past UTC cutoff. The delete work runs in the background, + // so the call returns 202 Accepted immediately. + using var response = await HttpClient.PostAsJsonAsync( + "/api/retention/sweep", + new RetentionSweepRequest { ErrorCutoff = DateTime.UtcNow.AddDays(-30) }, + SerializerOptions); + + started = response.StatusCode; + + // A future-dated cutoff is rejected with 400. + using var badRequest = await HttpClient.PostAsJsonAsync( + "/api/retention/sweep", + new RetentionSweepRequest { ErrorCutoff = DateTime.UtcNow.AddDays(1) }, + SerializerOptions); + + invalidCutoff = badRequest.StatusCode; + + return true; + }) + .Run(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(started, Is.EqualTo(HttpStatusCode.Accepted), "the sweep should start in the background"); + Assert.That(invalidCutoff, Is.EqualTo(HttpStatusCode.BadRequest), "a future cutoff must be rejected"); + } + + // The status endpoint must report the run, and the background sweep must complete. + var status = await WaitUntilSweepFinishes(); + Assert.That(status.IsRunning, Is.False, "the background sweep must complete"); + Assert.That(status.LastStartedAt, Is.Not.Null); + } + + [Test] + public async Task Should_report_background_completion_via_status() + { + if (StorageConfiguration.PersistenceType == "RavenDB") + { + Assert.Ignore("RavenDB has no sweeper — covered by Should_return_501_on_a_ravendb_backed_instance."); + return; + } + + await Define() + .Done(async _ => + { + using var response = await HttpClient.PostAsJsonAsync( + "/api/retention/sweep", + new RetentionSweepRequest { ErrorCutoff = DateTime.UtcNow.AddDays(-30) }, + SerializerOptions); + + return response.StatusCode == HttpStatusCode.Accepted; + }) + .Run(); + + var status = await WaitUntilSweepFinishes(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(status.IsRunning, Is.False); + Assert.That(status.LastFinishedAt, Is.Not.Null, "a completed run records its finish time"); + } + } + + [Test] + public async Task Should_return_501_on_a_ravendb_backed_instance() + { + if (StorageConfiguration.PersistenceType != "RavenDB") + { + Assert.Ignore("EFCore persisters support the sweep — covered by the efcore tests."); + return; + } + + HttpStatusCode postStatus = default; + HttpStatusCode getStatus = default; + + await Define() + .Done(async _ => + { + using var response = await HttpClient.PostAsJsonAsync( + "/api/retention/sweep", + new RetentionSweepRequest { ErrorCutoff = DateTime.UtcNow.AddDays(-30) }, + SerializerOptions); + + postStatus = response.StatusCode; + + using var status = await HttpClient.GetAsync("/api/retention/sweep/status"); + + getStatus = status.StatusCode; + + return true; + }) + .Run(); + + using (Assert.EnterMultipleScope()) + { + // RavenDB retention is the server-side @expires bundle; there is no cutoff-based sweeper + // to trigger, so the optional IRetentionSweeper resolution is absent and both verbs + // return 501 Not Implemented. + Assert.That(postStatus, Is.EqualTo(HttpStatusCode.NotImplemented), "POST must report not-supported on RavenDB"); + Assert.That(getStatus, Is.EqualTo(HttpStatusCode.NotImplemented), "GET status must report not-supported on RavenDB"); + } + } + + async Task WaitUntilSweepFinishes(TimeSpan? timeout = null) + { + var deadline = DateTime.UtcNow + (timeout ?? TimeSpan.FromSeconds(30)); + + while (DateTime.UtcNow < deadline) + { + using var response = await HttpClient.GetAsync("/api/retention/sweep/status"); + + if (response.StatusCode == HttpStatusCode.OK) + { + var status = await response.Content.ReadFromJsonAsync(SerializerOptions); + + if (status is { IsRunning: false }) + { + return status; + } + } + + await Task.Delay(TimeSpan.FromMilliseconds(200)); + } + + throw new Exception("The manual retention sweep did not finish within the timeout."); + } + + class Context : ScenarioContext; +} \ No newline at end of file diff --git a/src/ServiceControl.Api/Contracts/RetentionSweepRequest.cs b/src/ServiceControl.Api/Contracts/RetentionSweepRequest.cs new file mode 100644 index 0000000000..48a352a1bf --- /dev/null +++ b/src/ServiceControl.Api/Contracts/RetentionSweepRequest.cs @@ -0,0 +1,23 @@ +namespace ServiceControl.Api.Contracts; + +using System; + +/// +/// Request body for POST /api/retention/sweep. Both cutoffs are optional; when omitted +/// the corresponding sub-sweep derives its cutoff from the configured retention period, as the +/// scheduled hourly sweep does. A bare future-dated cutoff is rejected. +/// +public class RetentionSweepRequest +{ + /// + /// Cutoff applied to the failed-message sweep. null means + /// now - ErrorRetentionPeriod. + /// + public DateTime? ErrorCutoff { get; set; } + + /// + /// Cutoff applied to the event-log sweep. null means + /// now - EventsRetentionPeriod. + /// + public DateTime? EventsCutoff { get; set; } +} \ No newline at end of file diff --git a/src/ServiceControl.Api/Contracts/RetentionSweepResponse.cs b/src/ServiceControl.Api/Contracts/RetentionSweepResponse.cs new file mode 100644 index 0000000000..7c68fe0c4b --- /dev/null +++ b/src/ServiceControl.Api/Contracts/RetentionSweepResponse.cs @@ -0,0 +1,22 @@ +namespace ServiceControl.Api.Contracts; + +using System; + +/// +/// Response body for POST /api/retention/sweep. The Status field signals the +/// outcome: started (202), already-running (409), or +/// not-supported (501). +/// +public class RetentionSweepResponse +{ + public string Status { get; set; } + + public DateTime? StartedAt { get; set; } + + public DateTime? ErrorCutoff { get; set; } + + public DateTime? EventsCutoff { get; set; } + + /// A human-readable reason included when the operation is not supported. + public string Reason { get; set; } +} \ No newline at end of file diff --git a/src/ServiceControl.Api/Contracts/RetentionSweepStatus.cs b/src/ServiceControl.Api/Contracts/RetentionSweepStatus.cs new file mode 100644 index 0000000000..6e58c8f298 --- /dev/null +++ b/src/ServiceControl.Api/Contracts/RetentionSweepStatus.cs @@ -0,0 +1,25 @@ +namespace ServiceControl.Api.Contracts; + +using System; + +/// +/// Response body for GET /api/retention/sweep/status. On a persister with no sweeper +/// (e.g. RavenDB) the endpoint returns 501 with a instead. +/// +public class RetentionSweepStatus +{ + public bool IsRunning { get; set; } + + public DateTime? LastStartedAt { get; set; } + + public DateTime? LastFinishedAt { get; set; } + + public DateTime? LastErrorCutoff { get; set; } + + public DateTime? LastEventsCutoff { get; set; } + + public string LastError { get; set; } + + /// Present only on the 501 Not Implemented response. + public string Reason { get; set; } +} \ No newline at end of file diff --git a/src/ServiceControl.Api/IRetentionApi.cs b/src/ServiceControl.Api/IRetentionApi.cs new file mode 100644 index 0000000000..d497d716a7 --- /dev/null +++ b/src/ServiceControl.Api/IRetentionApi.cs @@ -0,0 +1,26 @@ +namespace ServiceControl.Api; + +using System.Threading; +using System.Threading.Tasks; +using Contracts; + +/// +/// Manual retention-sweep API. The implementation resolves the persister's sweeper +/// optionally: when no sweeper is registered (e.g. RavenDB, which uses server-side document +/// expiration) the operations report that the feature is not supported rather than silently +/// no-op'ing. +/// +public interface IRetentionApi +{ + /// + /// Starts a manual retention sweep with caller-supplied cutoffs. The delete work runs in + /// the background on a host-lifetime token; this method returns as soon as the run is + /// accepted (or refused because one is already running / unsupported). + /// + Task SweepAsync(RetentionSweepRequest request, CancellationToken cancellationToken = default); + + /// + /// Returns a point-in-time snapshot of sweep execution state for polling. + /// + Task GetStatusAsync(CancellationToken cancellationToken = default); +} \ No newline at end of file diff --git a/src/ServiceControl.Infrastructure/Auth/Permissions.cs b/src/ServiceControl.Infrastructure/Auth/Permissions.cs index 04f6582c14..b2af5c1c62 100644 --- a/src/ServiceControl.Infrastructure/Auth/Permissions.cs +++ b/src/ServiceControl.Infrastructure/Auth/Permissions.cs @@ -58,6 +58,9 @@ public static class Permissions /// Event log area — viewing the event log. public const string ErrorEventLogView = "error:eventlog:view"; + /// Retention area — manually triggering a data retention sweep. + public const string ErrorRetentionSweep = "error:retention:sweep"; + /// Licensing area — viewing and managing license configuration. public const string ErrorLicensingView = "error:licensing:view"; /// diff --git a/src/ServiceControl.Infrastructure/Auth/RolePermissions.cs b/src/ServiceControl.Infrastructure/Auth/RolePermissions.cs index 1263d43f1e..ad75a279d6 100644 --- a/src/ServiceControl.Infrastructure/Auth/RolePermissions.cs +++ b/src/ServiceControl.Infrastructure/Auth/RolePermissions.cs @@ -63,6 +63,7 @@ public static class RolePermissions Permissions.ErrorRedirectsManage, Permissions.ErrorThroughputView, Permissions.ErrorThroughputManage, + Permissions.ErrorRetentionSweep, ]; public static readonly FrozenDictionary> Roles = diff --git a/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs b/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs index e88ea72bfc..3002987a87 100644 --- a/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs +++ b/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs @@ -40,7 +40,12 @@ protected static void RegisterDataStores(IServiceCollection services, EFPersiste if (settings.RunRetentionSweep) { services.AddSingleton(); - services.AddHostedService(); + + // Register the sweeper as a resolvable singleton (concrete type + IRetentionSweeper) AND + // as a hosted service, all backed by one instance. + services.AddSingleton(); + services.AddHostedService(sp => sp.GetRequiredService()); + services.AddSingleton(sp => sp.GetRequiredService()); } services.AddSingleton(); diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/RetentionSweeper.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/RetentionSweeper.cs index 5396e060da..33e33ad681 100644 --- a/src/ServiceControl.Persistence.EFCore/Infrastructure/RetentionSweeper.cs +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/RetentionSweeper.cs @@ -5,6 +5,7 @@ namespace ServiceControl.Persistence.EFCore.Infrastructure; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; using ServiceControl.MessageFailures; +using ServiceControl.Persistence; using ServiceControl.Persistence.EFCore.Abstractions; using ServiceControl.Persistence.EFCore.DbContexts; using ServiceControl.Persistence.EFCore.Entities; @@ -13,19 +14,39 @@ namespace ServiceControl.Persistence.EFCore.Infrastructure; // Deletes rows once they age past their retention period. // Runs hourly, in bounded batches so it never holds a large delete, and recomputes the cutoffs on // every run so a changed retention setting takes effect without rewriting any row. +// +// A manual sweep can be triggered via the API (see IRetentionSweeper / IRetentionApi) with +// caller-supplied cutoffs. public class RetentionSweeper( ILogger logger, TimeProvider timeProvider, IServiceScopeFactory serviceScopeFactory, IBodyStoragePersistence bodyStorage, RetentionMetrics metrics, - EFPersisterSettings settings) : BackgroundService + EFPersisterSettings settings, + IHostApplicationLifetime hostApplicationLifetime) : BackgroundService, IRetentionSweeper { const int BatchSize = 1000; static readonly TimeSpan Interval = TimeSpan.FromHours(1); static readonly TimeSpan InitialDelay = TimeSpan.FromMinutes(1); static readonly TimeSpan BatchPause = TimeSpan.FromSeconds(1); + // Single-flight guard shared by the hourly timer path and the manual API path so two sweeps + // never overlap. Precedent: ExternalIntegrationRequestsDataStore.drainLock. + readonly SemaphoreSlim sweepLock = new(1, 1); + + // Status snapshot for GET /api/retention/sweep/status polling. Volatile reads/writes are + // sufficient here: the fields are written under sweepLock (or once at start) and read + // lock-free for status reporting, which only needs an eventually-consistent snapshot. + volatile bool isRunning; + DateTime? lastStartedAt; + DateTime? lastFinishedAt; + DateTime? lastErrorCutoff; + DateTime? lastEventsCutoff; + string? lastError; + + public RetentionSweepConfig Config => new(settings.ErrorRetentionPeriod, settings.EventsRetentionPeriod); + protected override async Task ExecuteAsync(CancellationToken cancellationToken = default) { logger.LogInformation("Starting retention sweep"); @@ -40,7 +61,7 @@ protected override async Task ExecuteAsync(CancellationToken cancellationToken = { try { - await Sweep(pace: true, cancellationToken); + await Sweep(errorCutoff: null, eventsCutoff: null, pace: true, cancellationToken); } #pragma warning disable PS0019 // The filter already excludes OperationCanceledException, so // cancellation propagates; PS0019 only recognises a cancellationToken guard. @@ -58,13 +79,77 @@ protected override async Task ExecuteAsync(CancellationToken cancellationToken = } // Runs a full sweep immediately, bypassing the timer and the inter-batch pause. - // Intended for tests that need the effect without waiting for the hourly loop. - public Task SweepNow(CancellationToken cancellationToken = default) => Sweep(pace: false, cancellationToken); + // Intended for tests that need the effect without waiting for the hourly loop. Uses the + // default cutoff derivation (now - retention period). + public Task SweepNow(CancellationToken cancellationToken = default) => + Sweep(errorCutoff: null, eventsCutoff: null, pace: false, cancellationToken); + + public ManualSweepAttempt TryStartManualSweep(DateTime? errorCutoff, DateTime? eventsCutoff, CancellationToken cancellationToken = default) + { + // Try to acquire the single-flight lock without waiting if a scheduled or manual sweep is + // already running (holding the lock) + if (!sweepLock.Wait(0, cancellationToken)) + { + return new ManualSweepAttempt(ManualSweepOutcome.AlreadyRunning, lastStartedAt, errorCutoff, eventsCutoff); + } + + // Lock acquired on this thread. The background task owns it from here and releases it when + // the sweep body completes (SemaphoreSlim is not thread-affine, so releasing from the + // background thread is safe). isRunning is set now so a concurrent manual call sees it. + isRunning = true; + lastStartedAt = timeProvider.GetUtcNow().UtcDateTime; + lastErrorCutoff = errorCutoff; + lastEventsCutoff = eventsCutoff; + lastError = null; + + _ = SweepWithoutAcquiringLock(); + + return new ManualSweepAttempt(ManualSweepOutcome.Started, lastStartedAt, errorCutoff, eventsCutoff); + + async Task SweepWithoutAcquiringLock() + { + try + { + // if the caller doesn't hand over a real cancellation token then use the application lifetime. + await SweepBody(errorCutoff, eventsCutoff, false, cancellationToken.CanBeCanceled ? cancellationToken : hostApplicationLifetime.ApplicationStopping); + lastFinishedAt = timeProvider.GetUtcNow().UtcDateTime; + } + finally + { + isRunning = false; + sweepLock.Release(); + } + } + } + + public RetentionSweepStatus GetStatus() => new(isRunning, lastStartedAt, lastFinishedAt, lastErrorCutoff, lastEventsCutoff, lastError); + + async Task Sweep(DateTime? errorCutoff, DateTime? eventsCutoff, bool pace, CancellationToken cancellationToken) + { + await sweepLock.WaitAsync(cancellationToken); + isRunning = true; + lastStartedAt = timeProvider.GetUtcNow().UtcDateTime; + lastErrorCutoff = errorCutoff; + lastEventsCutoff = eventsCutoff; + lastError = null; + try + { + await SweepBody(errorCutoff, eventsCutoff, pace, cancellationToken); + lastFinishedAt = timeProvider.GetUtcNow().UtcDateTime; + } + finally + { + isRunning = false; + sweepLock.Release(); + } + } - async Task Sweep(bool pace, CancellationToken cancellationToken) + // The three sub-sweeps, isolated from lock management so both the locked Sweep path and the + // manual background path (which already holds the lock) share one implementation. + async Task SweepBody(DateTime? errorCutoff, DateTime? eventsCutoff, bool pace, CancellationToken cancellationToken) { - await RunPass(RetentionEntity.FailedMessages, token => SweepFailedMessages(pace, token), cancellationToken); - await RunPass(RetentionEntity.EventLog, token => SweepEventLogItems(pace, token), cancellationToken); + await RunPass(RetentionEntity.FailedMessages, token => SweepFailedMessages(pace, errorCutoff, token), cancellationToken); + await RunPass(RetentionEntity.EventLog, token => SweepEventLogItems(pace, eventsCutoff, token), cancellationToken); await RunPass(RetentionEntity.GroupComments, SweepOrphanedGroupComments, cancellationToken); } @@ -105,10 +190,10 @@ async Task SweepOrphanedGroupComments(CancellationToken cancellationToken) } // Event log items are insert-only and carry no external bodies, so each batch is a single - // ordered DELETE. - async Task SweepEventLogItems(bool pace, CancellationToken cancellationToken) + // ordered DELETE. A caller-supplied cutoff overrides the default derivation. + async Task SweepEventLogItems(bool pace, DateTime? eventsCutoff, CancellationToken cancellationToken) { - var cutoff = timeProvider.GetUtcNow().UtcDateTime - settings.EventsRetentionPeriod; + var cutoff = eventsCutoff ?? (timeProvider.GetUtcNow().UtcDateTime - settings.EventsRetentionPeriod); while (!cancellationToken.IsCancellationRequested) { @@ -135,9 +220,9 @@ async Task SweepEventLogItems(bool pace, CancellationToken cancellationToken) } } - async Task SweepFailedMessages(bool pace, CancellationToken cancellationToken) + async Task SweepFailedMessages(bool pace, DateTime? errorCutoff, CancellationToken cancellationToken) { - var cutoff = timeProvider.GetUtcNow().UtcDateTime - settings.ErrorRetentionPeriod; + var cutoff = errorCutoff ?? (timeProvider.GetUtcNow().UtcDateTime - settings.ErrorRetentionPeriod); while (!cancellationToken.IsCancellationRequested) { @@ -195,4 +280,4 @@ async Task SweepFailedMessages(bool pace, CancellationToken cancellationToken) static System.Linq.Expressions.Expression> IsExpired(DateTime cutoff) => failedMessage => (failedMessage.Status == FailedMessageStatus.Resolved || failedMessage.Status == FailedMessageStatus.Archived) && failedMessage.StatusChangedAt < cutoff; -} +} \ No newline at end of file diff --git a/src/ServiceControl.Persistence.Tests/EFCore/RetentionSweepTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/RetentionSweepTests.cs index 761e624e12..aa4249ead8 100644 --- a/src/ServiceControl.Persistence.Tests/EFCore/RetentionSweepTests.cs +++ b/src/ServiceControl.Persistence.Tests/EFCore/RetentionSweepTests.cs @@ -400,4 +400,142 @@ async Task> GetRemainingMarkers() var items = (await EventLogDataStore.GetEventLogItems(new PagingInfo(page: 1, pageSize: 100))).Results; return [.. items.Select(i => i.Description)]; } + + IRetentionSweeper GetSweeper() => ServiceProvider.GetRequiredService(); + + async Task WaitForManualSweepToFinish() + { + var sweeper = GetSweeper(); + await WaitUntil(() => Task.FromResult(!sweeper.GetStatus().IsRunning), + "the manual sweep to finish"); + } + + [Test] + public async Task Manual_sweep_uses_the_caller_supplied_error_cutoff_to_delete_early() + { + // 20 days old is within the 30 day configured retention, so the scheduled sweep would keep it. + // A caller-supplied cutoff of 15 days ago is earlier than the message, so the manual sweep deletes it. + var message = await SeedFailedMessage(FailedMessageStatus.Resolved, Now.AddDays(-20)); + + var attempt = GetSweeper().TryStartManualSweep(Now.AddDays(-15), null); + + await WaitForManualSweepToFinish(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(attempt.Outcome, Is.EqualTo(ManualSweepOutcome.Started)); + Assert.That(await FindFailedMessage(message), Is.Null, + "the caller-supplied cutoff overrides the configured retention derivation"); + } + } + + [Test] + public async Task Manual_sweep_uses_the_caller_supplied_events_cutoff() + { + EFSettings.EventsRetentionPeriod = TimeSpan.FromDays(14); + + // 10 days old is within the 14 day configured events retention; a caller cutoff of 5 days ago deletes it. + await Store(EventLogRow("to-delete", Now.AddDays(-10))); + await Store(EventLogRow("to-keep", Now.AddDays(-3))); + + GetSweeper().TryStartManualSweep(null, Now.AddDays(-5)); + + await WaitForManualSweepToFinish(); + + var remaining = await GetRemainingMarkers(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(remaining, Does.Not.Contain("to-delete")); + Assert.That(remaining, Does.Contain("to-keep")); + } + } + + [Test] + public async Task Manual_sweep_with_null_cutoffs_keeps_the_default_derivation() + { + // No cutoff supplied => derive from settings as the scheduled path does. 29 days old is within 30 days. + var withinRetention = await SeedFailedMessage(FailedMessageStatus.Resolved, Now.AddDays(-29)); + var pastRetention = await SeedFailedMessage(FailedMessageStatus.Resolved, Now.AddDays(-31)); + + GetSweeper().TryStartManualSweep(null, null); + + await WaitForManualSweepToFinish(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(await FindFailedMessage(withinRetention), Is.Not.Null); + Assert.That(await FindFailedMessage(pastRetention), Is.Null); + } + } + + [Test] + public async Task Manual_sweep_runs_in_the_background_and_reports_status() + { + await SeedFailedMessage(FailedMessageStatus.Resolved, Now.AddDays(-31)); + + var sweeper = GetSweeper(); + var attempt = sweeper.TryStartManualSweep(Now.AddDays(-30), null); + + Assert.That(attempt.Outcome, Is.EqualTo(ManualSweepOutcome.Started)); + Assert.That(attempt.StartedAt, Is.Not.Null); + + await WaitForManualSweepToFinish(); + + var status = sweeper.GetStatus(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(status.IsRunning, Is.False); + Assert.That(status.LastStartedAt, Is.Not.Null); + Assert.That(status.LastFinishedAt, Is.Not.Null); + Assert.That(status.LastErrorCutoff, Is.Not.Null); + Assert.That(status.LastError, Is.Null); + } + } + + [Test] + public async Task A_second_manual_sweep_is_refused_while_one_is_running() + { + // Seed enough rows to force multiple delete batches so the first sweep is still running when the + // second, synchronous call is made. The single-flight lock is held from the moment the first call + // returns Started until the background body completes. + var rows = new List(); + for (var i = 0; i < 1500; i++) + { + rows.Add(new FailedMessageEntity + { + UniqueMessageId = Guid.NewGuid(), + Status = FailedMessageStatus.Archived, + StatusChangedAt = Now.AddDays(-31), + LastModified = Now.AddDays(-31), + NumberOfProcessingAttempts = 1, + FirstTimeOfFailure = Now.AddDays(-31), + LastTimeOfFailure = Now.AddDays(-31), + LastAttemptedAt = Now.AddDays(-31), + IsSystemMessage = false, + HeadersJson = "{}", + BodyStoredExternally = false, + BodySize = 0, + FailingEndpointAddress = "Shipping" + }); + } + + await Store([.. rows]); + + var sweeper = GetSweeper(); + var first = sweeper.TryStartManualSweep(Now.AddDays(-30), null); + // Immediately request a second sweep on the same thread while the first is still deleting. + var second = sweeper.TryStartManualSweep(Now.AddDays(-30), null); + + await WaitForManualSweepToFinish(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(first.Outcome, Is.EqualTo(ManualSweepOutcome.Started), + "the first call should start the sweep"); + Assert.That(second.Outcome, Is.EqualTo(ManualSweepOutcome.AlreadyRunning), + "a second sweep must not run in parallel with the first"); + } + } } diff --git a/src/ServiceControl.Persistence/IRetentionSweeper.cs b/src/ServiceControl.Persistence/IRetentionSweeper.cs new file mode 100644 index 0000000000..3f7b3d58fe --- /dev/null +++ b/src/ServiceControl.Persistence/IRetentionSweeper.cs @@ -0,0 +1,63 @@ +namespace ServiceControl.Persistence; + +using System; +using System.Threading; +using System.Threading.Tasks; + +/// +/// A persister-agnostic retention sweep operation. Only persisters that actually scan and +/// delete aged rows register this interface (e.g. the EFCore SQL persisters). RavenDB does +/// not — its retention is the server-side @expires bundle stamped per-document at +/// write time — so the interface is resolved optionally by the API, which returns +/// 501 Not Implemented when no registration is present. +/// +public interface IRetentionSweeper +{ + /// + /// The retention periods and minimum-age rules in force for this instance. + /// + RetentionSweepConfig Config { get; } + + /// + /// Starts a full retention sweep on a background task tied to the host lifetime (not the + /// caller's request token), using the caller-supplied cutoffs. When a cutoff is + /// null the corresponding sub-sweep derives its cutoff from the configured + /// retention period as the scheduled path does. + /// + /// A snapshot describing the run that was started; never throws for "already running" + /// — that is reported in the returned status. + ManualSweepAttempt TryStartManualSweep(DateTime? errorCutoff, DateTime? eventsCutoff, CancellationToken cancellationToken = default); + + /// + /// A point-in-time snapshot of sweep execution state for status polling. + /// + RetentionSweepStatus GetStatus(); +} + +/// Configuration describing the retention rules in force. +public sealed record RetentionSweepConfig(TimeSpan ErrorRetentionPeriod, TimeSpan EventsRetentionPeriod); + +/// The outcome of a manual sweep start request. +public enum ManualSweepOutcome +{ + /// The sweep was started on a background task. + Started, + /// A sweep is already running; the caller should poll . + AlreadyRunning +} + +/// The result of a call. +public sealed record ManualSweepAttempt( + ManualSweepOutcome Outcome, + DateTime? StartedAt, + DateTime? ErrorCutoff, + DateTime? EventsCutoff); + +/// A point-in-time snapshot of sweep execution state. +public sealed record RetentionSweepStatus( + bool IsRunning, + DateTime? LastStartedAt, + DateTime? LastFinishedAt, + DateTime? LastErrorCutoff, + DateTime? LastEventsCutoff, + string? LastError); \ No newline at end of file diff --git a/src/ServiceControl.UnitTests/ApprovalFiles/APIApprovals.HttpApiRoutes.approved.txt b/src/ServiceControl.UnitTests/ApprovalFiles/APIApprovals.HttpApiRoutes.approved.txt index 264be98787..475a694bae 100644 --- a/src/ServiceControl.UnitTests/ApprovalFiles/APIApprovals.HttpApiRoutes.approved.txt +++ b/src/ServiceControl.UnitTests/ApprovalFiles/APIApprovals.HttpApiRoutes.approved.txt @@ -74,4 +74,6 @@ GET /redirects => ServiceControl.MessageRedirects.Api.MessageRedirectsController POST /redirects => ServiceControl.MessageRedirects.Api.MessageRedirectsController:NewRedirects(MessageRedirectRequest request, CancellationToken cancellationToken) DELETE /redirects/{messageRedirectId:guid} => ServiceControl.MessageRedirects.Api.MessageRedirectsController:DeleteRedirect(Guid messageRedirectId, CancellationToken cancellationToken) PUT /redirects/{messageRedirectId:guid} => ServiceControl.MessageRedirects.Api.MessageRedirectsController:UpdateRedirect(Guid messageRedirectId, MessageRedirectRequest request, CancellationToken cancellationToken) +POST /retention/sweep => ServiceControl.Retention.Api.RetentionController:Sweep(RetentionSweepRequest request, CancellationToken cancellationToken) +GET /retention/sweep/status => ServiceControl.Retention.Api.RetentionController:Status(CancellationToken cancellationToken) GET /sagas/{id} => ServiceControl.SagaAudit.SagasController:Sagas(PagingInfo pagingInfo, Guid id, CancellationToken cancellationToken) diff --git a/src/ServiceControl/Infrastructure/Api/RetentionApi.cs b/src/ServiceControl/Infrastructure/Api/RetentionApi.cs new file mode 100644 index 0000000000..0d3e5649d6 --- /dev/null +++ b/src/ServiceControl/Infrastructure/Api/RetentionApi.cs @@ -0,0 +1,137 @@ +namespace ServiceControl.Infrastructure.Api; + +using System; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using ServiceBus.Management.Infrastructure.Settings; +using ServiceControl.Api; +using ServiceControl.Api.Contracts; + +// Manual retention-sweep API. The persister's IRetentionSweeper is resolved *optionally* so the +// same controller/route is mapped on every persister: EFCore registers it and gets 202/409/200; +// RavenDB registers nothing (its retention is the server-side @expires bundle) and gets 501. +class RetentionApi(IServiceProvider serviceProvider, Settings settings) : IRetentionApi +{ + const string NotSupportedReason = "The current storage has no retention sweeper."; + + public Task SweepAsync(RetentionSweepRequest request, CancellationToken cancellationToken = default) + { + // Resolve the sweeper lazily and optionally — never required via constructor injection, or + // a RavenDB-backed instance would throw at resolve time. Absent => 501 Not Implemented. + var sweeper = serviceProvider.GetService(); + if (sweeper is null) + { + return Task.FromResult(NotSupported()); + } + + // Maintenance mode refuses mutating operations; a sweep while the DB is being maintained + // would contend with the maintenance work. + if (settings.PersisterSpecificSettings?.MaintenanceMode == true) + { + return Task.FromResult(new RetentionSweepResponse { Status = "maintenance", Reason = "The instance is in maintenance mode." }); + } + + request ??= new RetentionSweepRequest(); + + // Cutoffs must be UTC and in the past. A future cutoff would delete nothing and is almost + // certainly a caller mistake, so it is rejected rather than clamped. + if (TryValidateCutoff(request.ErrorCutoff, out var errorCutoff, out var error) is false) + { + return Task.FromResult(Invalid(error)); + } + + if (TryValidateCutoff(request.EventsCutoff, out var eventsCutoff, out error) is false) + { + return Task.FromResult(Invalid(error)); + } + + var attempt = sweeper.TryStartManualSweep(errorCutoff, eventsCutoff, cancellationToken); + + return Task.FromResult(attempt.Outcome switch + { + ServiceControl.Persistence.ManualSweepOutcome.Started => new RetentionSweepResponse + { + Status = "started", + StartedAt = attempt.StartedAt, + ErrorCutoff = attempt.ErrorCutoff, + EventsCutoff = attempt.EventsCutoff + }, + ServiceControl.Persistence.ManualSweepOutcome.AlreadyRunning => new RetentionSweepResponse + { + Status = "already-running", + StartedAt = attempt.StartedAt + }, + _ => new RetentionSweepResponse + { + Status = "already-running", + StartedAt = attempt.StartedAt + } + }); + } + + public Task GetStatusAsync(CancellationToken cancellationToken = default) + { + var sweeper = serviceProvider.GetService(); + if (sweeper is null) + { + return Task.FromResult(new RetentionSweepStatus { Reason = NotSupportedReason }); + } + + // Map the persister's status record onto the API contract DTO (the two share a name but + // live in different namespaces: ServiceControl.Persistence vs ServiceControl.Api.Contracts). + ServiceControl.Persistence.RetentionSweepStatus status = sweeper.GetStatus(); + + return Task.FromResult(new RetentionSweepStatus + { + IsRunning = status.IsRunning, + LastStartedAt = status.LastStartedAt, + LastFinishedAt = status.LastFinishedAt, + LastErrorCutoff = status.LastErrorCutoff, + LastEventsCutoff = status.LastEventsCutoff, + LastError = status.LastError + }); + } + + static bool TryValidateCutoff(DateTime? supplied, out DateTime? validated, out string error) + { + if (supplied is null) + { + validated = null; + error = null; + return true; + } + + var value = supplied.Value; + + if (value.Kind != DateTimeKind.Utc) + { + validated = null; + error = "Cutoffs must be specified as UTC DateTime values."; + return false; + } + + if (value > DateTime.UtcNow) + { + validated = null; + error = "Cutoffs must not be in the future."; + return false; + } + + validated = value; + error = null; + return true; + } + + static RetentionSweepResponse NotSupported() => new() + { + Status = "not-supported", + Reason = NotSupportedReason + }; + + static RetentionSweepResponse Invalid(string reason) => new() + { + Status = "invalid-cutoff", + Reason = reason + }; +} \ No newline at end of file diff --git a/src/ServiceControl/Retention/Api/RetentionController.cs b/src/ServiceControl/Retention/Api/RetentionController.cs new file mode 100644 index 0000000000..2f0373872e --- /dev/null +++ b/src/ServiceControl/Retention/Api/RetentionController.cs @@ -0,0 +1,51 @@ +namespace ServiceControl.Retention.Api; + +using System.Threading; +using System.Threading.Tasks; +using Infrastructure.Auth; +using Microsoft.AspNetCore.Authorization; +using Microsoft.AspNetCore.Mvc; +using ServiceControl.Api; +using ServiceControl.Api.Contracts; + +// Manual retention-sweep endpoint. Lives only on the primary error instance (the sweeper is only +// registered there). On a RavenDB-backed instance IRetentionSweeper is not registered, so the +// IRetentionApi implementation returns a "not-supported" status that this controller maps to 501. +[ApiController] +[Route("api")] +public class RetentionController(IRetentionApi retentionApi) : ControllerBase +{ + // Starts a full retention sweep with caller-supplied cutoffs. The delete work runs in the + // background on a host-lifetime token; this returns as soon as the run is accepted (202), + // already running (409), in maintenance mode (503), unsupported by the persister (501), or + // the cutoff was invalid (400). + [Authorize(Policy = Permissions.ErrorRetentionSweep)] + [Route("retention/sweep")] + [HttpPost] + public async Task Sweep([FromBody] RetentionSweepRequest request, CancellationToken cancellationToken = default) + { + var response = await retentionApi.SweepAsync(request ?? new RetentionSweepRequest(), cancellationToken); + + return response.Status switch + { + "started" => Accepted(response), + "already-running" => Conflict(response), + "maintenance" => StatusCode(503, response), + "not-supported" => StatusCode(501, response), + "invalid-cutoff" => BadRequest(response), + _ => Ok(response) + }; + } + + // Polls the execution state of the most recent sweep. + [Authorize(Policy = Permissions.ErrorRetentionSweep)] + [Route("retention/sweep/status")] + [HttpGet] + public async Task Status(CancellationToken cancellationToken = default) + { + var status = await retentionApi.GetStatusAsync(cancellationToken); + + // A reason is present only when the persister has no sweeper (e.g. RavenDB). + return status.Reason is not null ? StatusCode(501, status) : Ok(status); + } +} \ No newline at end of file diff --git a/src/ServiceControl/ServiceControlApiHostBuilderExtensions.cs b/src/ServiceControl/ServiceControlApiHostBuilderExtensions.cs index 9cb637acff..ad678bcef7 100644 --- a/src/ServiceControl/ServiceControlApiHostBuilderExtensions.cs +++ b/src/ServiceControl/ServiceControlApiHostBuilderExtensions.cs @@ -12,6 +12,7 @@ public static void AddServiceControlApis(this IHostApplicationBuilder hostBuilde hostBuilder.Services.AddSingleton(); hostBuilder.Services.AddSingleton(); hostBuilder.Services.AddSingleton(); + hostBuilder.Services.AddSingleton(); } } } \ No newline at end of file