diff --git a/src/ServiceControl.AcceptanceTests/WebApi/When_triggering_a_manual_retention_purge.cs b/src/ServiceControl.AcceptanceTests/WebApi/When_triggering_a_manual_retention_purge.cs new file mode 100644 index 0000000000..da514f32db --- /dev/null +++ b/src/ServiceControl.AcceptanceTests/WebApi/When_triggering_a_manual_retention_purge.cs @@ -0,0 +1,165 @@ +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_purge : 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; + RetentionPurgeStatusResponse completion = null; + + await Define() + .Done(async _ => + { + // Trigger a purge 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/maintenance/retention/purge", + new RetentionPurgeRequest { ErrorCutoff = DateTime.UtcNow.AddDays(-30) }, + SerializerOptions); + + started = response.StatusCode; + + // A future-dated cutoff is rejected with 400. + using var badRequest = await HttpClient.PostAsJsonAsync( + "/api/maintenance/retention/purge", + new RetentionPurgeRequest { ErrorCutoff = DateTime.UtcNow.AddDays(1) }, + SerializerOptions); + + invalidCutoff = badRequest.StatusCode; + + // The status endpoint must report the run, and the background purge must complete. + completion = await WaitUntilPurgeFinishes(); + + return true; + }) + .Run(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(started, Is.EqualTo(HttpStatusCode.Accepted), "the purge should start in the background"); + Assert.That(invalidCutoff, Is.EqualTo(HttpStatusCode.BadRequest), "a future cutoff must be rejected"); + Assert.That(completion, Is.Not.Null, "the background purge must complete"); + Assert.That(completion.IsRunning, Is.False, "the background purge must complete"); + Assert.That(completion.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; + } + + RetentionPurgeStatusResponse completion = null; + + await Define() + .Done(async _ => + { + using var response = await HttpClient.PostAsJsonAsync( + "/api/maintenance/retention/purge", + new RetentionPurgeRequest { ErrorCutoff = DateTime.UtcNow.AddDays(-30) }, + SerializerOptions); + + completion = await WaitUntilPurgeFinishes(); + return response.StatusCode == HttpStatusCode.Accepted; + }) + .Run(); + + + using (Assert.EnterMultipleScope()) + { + Assert.That(completion, Is.Not.Null, "the background purge must complete"); + Assert.That(completion.IsRunning, Is.False); + Assert.That(completion.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 purge — covered by the efcore tests."); + return; + } + + HttpStatusCode postStatus = default; + HttpStatusCode getStatus = default; + + await Define() + .Done(async _ => + { + using var response = await HttpClient.PostAsJsonAsync( + "/api/maintenance/retention/purge", + new RetentionPurgeRequest { ErrorCutoff = DateTime.UtcNow.AddDays(-30) }, + SerializerOptions); + + postStatus = response.StatusCode; + + using var status = await HttpClient.GetAsync("/api/maintenance/retention/purge/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 WaitUntilPurgeFinishes(TimeSpan? timeout = null) + { + var deadline = DateTime.UtcNow + (timeout ?? TimeSpan.FromSeconds(30)); + + while (DateTime.UtcNow < deadline) + { + using var response = await HttpClient.GetAsync("/api/maintenance/retention/purge/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 purge did not finish within the timeout."); + } + + class Context : ScenarioContext; +} \ No newline at end of file diff --git a/src/ServiceControl.Api/Contracts/RetentionPurgeRequest.cs b/src/ServiceControl.Api/Contracts/RetentionPurgeRequest.cs new file mode 100644 index 0000000000..ec24bca4b8 --- /dev/null +++ b/src/ServiceControl.Api/Contracts/RetentionPurgeRequest.cs @@ -0,0 +1,23 @@ +namespace ServiceControl.Api.Contracts; + +using System; + +/// +/// Request body for POST /api/maintenance/retention/purge. Both cutoffs are optional; when omitted +/// the corresponding purge derives its cutoff from the configured retention period, as the +/// scheduled hourly sweep does. A bare future-dated cutoff is rejected. +/// +public class RetentionPurgeRequest +{ + /// + /// Cutoff applied to the failed-message purge. null means + /// now - ErrorRetentionPeriod. + /// + public DateTime? ErrorCutoff { get; set; } + + /// + /// Cutoff applied to the event-log purge. null means + /// now - EventsRetentionPeriod. + /// + public DateTime? EventsCutoff { get; set; } +} \ No newline at end of file diff --git a/src/ServiceControl.Api/Contracts/RetentionPurgeResponse.cs b/src/ServiceControl.Api/Contracts/RetentionPurgeResponse.cs new file mode 100644 index 0000000000..bcf9edf47a --- /dev/null +++ b/src/ServiceControl.Api/Contracts/RetentionPurgeResponse.cs @@ -0,0 +1,22 @@ +namespace ServiceControl.Api.Contracts; + +using System; + +/// +/// Response body for POST /api/maintenance/retention/purge. The Status field signals the +/// outcome: started (202), already-running (409), or +/// not-supported (501). +/// +public class RetentionPurgeResponse +{ + public RetentionPurgeStatus 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/RetentionPurgeStatus.cs b/src/ServiceControl.Api/Contracts/RetentionPurgeStatus.cs new file mode 100644 index 0000000000..c8b000d5da --- /dev/null +++ b/src/ServiceControl.Api/Contracts/RetentionPurgeStatus.cs @@ -0,0 +1,9 @@ +namespace ServiceControl.Api.Contracts; + +public enum RetentionPurgeStatus +{ + Started, + AlreadyRunning, + NotSupported, + Error +} \ No newline at end of file diff --git a/src/ServiceControl.Api/Contracts/RetentionPurgeStatusResponse.cs b/src/ServiceControl.Api/Contracts/RetentionPurgeStatusResponse.cs new file mode 100644 index 0000000000..716c666fdf --- /dev/null +++ b/src/ServiceControl.Api/Contracts/RetentionPurgeStatusResponse.cs @@ -0,0 +1,23 @@ +namespace ServiceControl.Api.Contracts; + +using System; + +/// +/// Response body for GET /api/maintenance/retention/purge/status. On a persister with no sweeper +/// (e.g. RavenDB) the endpoint returns 501 with a instead. +/// +public class RetentionPurgeStatusResponse +{ + 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; } + + /// 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..d1985c1870 --- /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-purge 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 +/// failing. +/// +public interface IRetentionApi +{ + /// + /// Starts a manual retention purge 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 Sweep(RetentionPurgeRequest request, CancellationToken cancellationToken = default); + + /// + /// Returns a point-in-time snapshot of purge execution state for polling. + /// + Task GetStatus(CancellationToken cancellationToken = default); +} \ No newline at end of file diff --git a/src/ServiceControl.Infrastructure/Auth/MessageAction.cs b/src/ServiceControl.Infrastructure/Auth/MessageAction.cs index 37deffca8d..f8463e5eac 100644 --- a/src/ServiceControl.Infrastructure/Auth/MessageAction.cs +++ b/src/ServiceControl.Infrastructure/Auth/MessageAction.cs @@ -7,7 +7,8 @@ public enum MessageActionKind Retry, Archive, Unarchive, - Edit + Edit, + Delete } /// How the action selected the messages it acts on. diff --git a/src/ServiceControl.Infrastructure/Auth/MessageActionAuditLog.cs b/src/ServiceControl.Infrastructure/Auth/MessageActionAuditLog.cs index 958724d309..48e6df9149 100644 --- a/src/ServiceControl.Infrastructure/Auth/MessageActionAuditLog.cs +++ b/src/ServiceControl.Infrastructure/Auth/MessageActionAuditLog.cs @@ -71,7 +71,7 @@ static string BuildEcsEvent(AuditUser user, MessageActionKind kind, string permi { kind = "event", category = new[] { "configuration" }, - type = new[] { kind == MessageActionKind.Archive ? "deletion" : "change" }, + type = new[] { kind is MessageActionKind.Archive or MessageActionKind.Delete ? "deletion" : "change" }, action = permission, outcome = success ? "success" : "failure" }, diff --git a/src/ServiceControl.Infrastructure/Auth/MessageActionAuditLogExtensions.cs b/src/ServiceControl.Infrastructure/Auth/MessageActionAuditLogExtensions.cs index f91aa3fb46..157c16b841 100644 --- a/src/ServiceControl.Infrastructure/Auth/MessageActionAuditLogExtensions.cs +++ b/src/ServiceControl.Infrastructure/Auth/MessageActionAuditLogExtensions.cs @@ -39,4 +39,37 @@ public static async Task AuditedOperation(this IMessageActionAuditLog auditLog, auditLog.Operation(user, kind, permission, scope, resource, count, operationId, success); } } + + /// + /// Executes a message action and records the operation-level audit entry with the actual + /// outcome: success when the action completed, failure when it threw (the exception is + /// rethrown). Logging after the action keeps the trail truthful — an entry written before the + /// send would claim success for an operation the transport may have rejected. + /// + public static async Task AuditedOperation(this IMessageActionAuditLog auditLog, AuditUser user, + MessageActionKind kind, string permission, MessageActionScope scope, string? resource, + int? count, string operationId, Func> action, + CancellationToken cancellationToken = default) + { + var success = true; + try + { + return await action(cancellationToken).ConfigureAwait(false); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + // A cancelled operation did not complete, so it is recorded as a failure like any other. + success = false; + throw; + } + catch + { + success = false; + throw; + } + finally + { + auditLog.Operation(user, kind, permission, scope, resource, count, operationId, success); + } + } } diff --git a/src/ServiceControl.Infrastructure/Auth/Permissions.cs b/src/ServiceControl.Infrastructure/Auth/Permissions.cs index 04f6582c14..45e9cec07f 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 purge. + public const string ErrorRetentionPurge = "error:retention:purge"; + /// 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..b1c273ee27 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.ErrorRetentionPurge, ]; 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..a5df9ae33b 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,36 @@ 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/maintenance/retention/purge/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; + protected override async Task ExecuteAsync(CancellationToken cancellationToken = default) { logger.LogInformation("Starting retention sweep"); @@ -40,7 +58,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 +76,84 @@ 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(RetentionSweepStatus.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; + + _ = SweepWithoutAcquiringLock(); + + return new ManualSweepAttempt(RetentionSweepStatus.Started, lastStartedAt, errorCutoff, eventsCutoff); + + async Task SweepWithoutAcquiringLock() + { + using var cancellation = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, hostApplicationLifetime.ApplicationStopping); + try + { + // if the caller doesn't hand over a real cancellation token then use the application lifetime. + await SweepBody(errorCutoff, eventsCutoff, false, cancellation.Token); + lastFinishedAt = timeProvider.GetUtcNow().UtcDateTime; + } + catch (OperationCanceledException) when (cancellation.Token.IsCancellationRequested) + { + //smother this exception, cancelling + } + catch (Exception e) + { + logger.LogError(e, "Error during retention sweep"); + } + finally + { + isRunning = false; + sweepLock.Release(); + } + } + } + + public RetentionSweepCurrentStatus GetStatus() => new(isRunning, lastStartedAt, lastFinishedAt, lastErrorCutoff, lastEventsCutoff); + + 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; + 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 +194,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 +224,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 +284,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..f20f506235 100644 --- a/src/ServiceControl.Persistence.Tests/EFCore/RetentionSweepTests.cs +++ b/src/ServiceControl.Persistence.Tests/EFCore/RetentionSweepTests.cs @@ -400,4 +400,141 @@ 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(RetentionSweepStatus.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(RetentionSweepStatus.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); + } + } + + [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(RetentionSweepStatus.Started), + "the first call should start the sweep"); + Assert.That(second.Outcome, Is.EqualTo(RetentionSweepStatus.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..96b52112c9 --- /dev/null +++ b/src/ServiceControl.Persistence/IRetentionSweeper.cs @@ -0,0 +1,25 @@ +namespace ServiceControl.Persistence; + +using System; +using System.Threading; + +/// +/// 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 +{ + /// + /// Starts a full retention sweep on a background task + /// + /// A snapshot describing the run that was started; will respond with a status of AlreadyRunning if there is already a sweep running. + ManualSweepAttempt TryStartManualSweep(DateTime? errorCutoff, DateTime? eventsCutoff, CancellationToken cancellationToken = default); + + /// + /// A point-in-time snapshot of current sweep execution state for status polling. + /// + RetentionSweepCurrentStatus GetStatus(); +} \ No newline at end of file diff --git a/src/ServiceControl.Persistence/ManualSweepAttempt.cs b/src/ServiceControl.Persistence/ManualSweepAttempt.cs new file mode 100644 index 0000000000..451f59c66d --- /dev/null +++ b/src/ServiceControl.Persistence/ManualSweepAttempt.cs @@ -0,0 +1,10 @@ +namespace ServiceControl.Persistence; + +using System; + +/// The result of a call. +public sealed record ManualSweepAttempt( + RetentionSweepStatus Outcome, + DateTime? StartedAt, + DateTime? ErrorCutoff, + DateTime? EventsCutoff); \ No newline at end of file diff --git a/src/ServiceControl.Persistence/RetentionSweepCurrentStatus.cs b/src/ServiceControl.Persistence/RetentionSweepCurrentStatus.cs new file mode 100644 index 0000000000..2bc32143ea --- /dev/null +++ b/src/ServiceControl.Persistence/RetentionSweepCurrentStatus.cs @@ -0,0 +1,11 @@ +namespace ServiceControl.Persistence; + +using System; + +/// A point-in-time snapshot of sweep execution state. +public sealed record RetentionSweepCurrentStatus( + bool IsRunning, + DateTime? LastStartedAt, + DateTime? LastFinishedAt, + DateTime? LastErrorCutoff, + DateTime? LastEventsCutoff); \ No newline at end of file diff --git a/src/ServiceControl.Persistence/RetentionSweepStatus.cs b/src/ServiceControl.Persistence/RetentionSweepStatus.cs new file mode 100644 index 0000000000..b7a1f9ca52 --- /dev/null +++ b/src/ServiceControl.Persistence/RetentionSweepStatus.cs @@ -0,0 +1,10 @@ +namespace ServiceControl.Persistence; + +/// The outcome of a sweep start request. +public enum RetentionSweepStatus +{ + /// The sweep was started on a background task. + Started, + /// A sweep is already running. + AlreadyRunning +} \ 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..69498d4d04 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/purge => ServiceControl.Retention.Api.SystemMaintenanceController:Purge(RetentionPurgeRequest request, CancellationToken cancellationToken) +GET /retention/purge/status => ServiceControl.Retention.Api.SystemMaintenanceController: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..44d3c6fa71 --- /dev/null +++ b/src/ServiceControl/Infrastructure/Api/RetentionApi.cs @@ -0,0 +1,101 @@ +namespace ServiceControl.Infrastructure.Api; + +using System; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using Persistence; +using ServiceControl.Api; +using ServiceControl.Api.Contracts; +using RetentionSweepStatus = Persistence.RetentionSweepStatus; + +// Manual retention-purge 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) : IRetentionApi +{ + public const string NotSupportedReason = "The currently configured storage has no retention sweeper."; + + + public Task Sweep(RetentionPurgeRequest 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((RetentionPurgeResponse)new() { Status = RetentionPurgeStatus.NotSupported, Reason = NotSupportedReason }); + } + + + request ??= new RetentionPurgeRequest(); + + // Cutoffs must be UTC and in the past. A future cutoff would delete everything and is almost + // certainly a caller mistake, so it is rejected rather than clamped. + if (!TryValidateCutoff(request.ErrorCutoff, out var errorCutoff, out var error) + || !TryValidateCutoff(request.EventsCutoff, out var eventsCutoff, out error)) + { + return Task.FromResult((RetentionPurgeResponse)new() { Status = RetentionPurgeStatus.Error, Reason = error }); + } + + var attempt = sweeper.TryStartManualSweep(errorCutoff, eventsCutoff, cancellationToken); + + return Task.FromResult(attempt.Outcome == RetentionSweepStatus.Started + ? new RetentionPurgeResponse { Status = RetentionPurgeStatus.Started, StartedAt = attempt.StartedAt, ErrorCutoff = attempt.ErrorCutoff, EventsCutoff = attempt.EventsCutoff } + : new RetentionPurgeResponse { Status = RetentionPurgeStatus.AlreadyRunning, StartedAt = attempt.StartedAt }); + } + + public Task GetStatus(CancellationToken cancellationToken = default) + { + var sweeper = serviceProvider.GetService(); + if (sweeper is null) + { + return Task.FromResult(new RetentionPurgeStatusResponse { 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). + var status = sweeper.GetStatus(); + + return Task.FromResult(new RetentionPurgeStatusResponse + { + IsRunning = status.IsRunning, + LastStartedAt = status.LastStartedAt, + LastFinishedAt = status.LastFinishedAt, + LastErrorCutoff = status.LastErrorCutoff, + LastEventsCutoff = status.LastEventsCutoff + }); + } + + 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.Unspecified) + { + validated = null; + error = "Cutoffs must be specified as UTC DateTime values."; + return false; + } + + value = value.ToUniversalTime(); + + if (value > DateTime.UtcNow) + { + validated = null; + error = "Cutoffs must not be in the future."; + return false; + } + + validated = value; + error = null; + return true; + } +} \ No newline at end of file diff --git a/src/ServiceControl/Retention/Api/SystemMaintenanceController.cs b/src/ServiceControl/Retention/Api/SystemMaintenanceController.cs new file mode 100644 index 0000000000..9b2496dc30 --- /dev/null +++ b/src/ServiceControl/Retention/Api/SystemMaintenanceController.cs @@ -0,0 +1,59 @@ +namespace ServiceControl.Retention.Api; + +using System; +using System.Threading; +using System.Threading.Tasks; +using Infrastructure.WebApi; +using Microsoft.AspNetCore.Authorization; +using Microsoft.AspNetCore.Mvc; +using ServiceControl.Api; +using ServiceControl.Api.Contracts; +using ServiceControl.Infrastructure.Auth; + +// Manual retention-purge 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/maintenance")] +public class SystemMaintenanceController(IRetentionApi retentionApi, ICurrentUserAccessor userAccessor, IMessageActionAuditLog auditLog) : ControllerBase +{ + // Starts a full retention purge 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), unsupported by the persister (501), or the cutoff was invalid (400). + [Authorize(Policy = Permissions.ErrorRetentionPurge)] + [Route("retention/purge")] + [HttpPost] + public async Task Purge([FromBody] RetentionPurgeRequest request, CancellationToken cancellationToken = default) + { + var user = userAccessor.Resolve(User); + var operationId = this.AuditOperationId(); + RetentionPurgeResponse response = null; + return await auditLog.AuditedOperation(user, MessageActionKind.Delete, Permissions.ErrorRetentionPurge, MessageActionScope.Range, null, null, operationId, async ct => + { + response = await retentionApi.Sweep(request ?? new RetentionPurgeRequest(), ct); + return ToActionResult(response); + }, cancellationToken); + + IActionResult ToActionResult(RetentionPurgeResponse retentionPurgeResponse) => + retentionPurgeResponse.Status switch + { + RetentionPurgeStatus.Started => Accepted(retentionPurgeResponse), + RetentionPurgeStatus.AlreadyRunning => Conflict(retentionPurgeResponse), + RetentionPurgeStatus.NotSupported => StatusCode(501, retentionPurgeResponse), + RetentionPurgeStatus.Error => BadRequest(retentionPurgeResponse), + _ => throw new ArgumentOutOfRangeException(nameof(retentionPurgeResponse.Status), retentionPurgeResponse.Status, "Unexpected retention purge status.") + }; + } + + // Polls the execution state of the most recent sweep/purge operation. + [Authorize(Policy = Permissions.ErrorRetentionPurge)] + [Route("retention/purge/status")] + [HttpGet] + public async Task Status(CancellationToken cancellationToken = default) + { + var status = await retentionApi.GetStatus(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