Implement manual retention sweep

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<Context>()
+            .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<Context>()
+            .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<Context>()
+            .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<RetentionPurgeStatusResponse> 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<RetentionPurgeStatusResponse>(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;
+}

src/ServiceControl.Api/Contracts/RetentionPurgeRequest.cs

@@ -0,0 +1,23 @@
+namespace ServiceControl.Api.Contracts;
+
+using System;
+
+/// <summary>
+/// Request body for <c>POST /api/maintenance/retention/purge</c>. 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.
+/// </summary>
+public class RetentionPurgeRequest
+{
+    /// <summary>
+    /// Cutoff applied to the failed-message purge. <c>null</c> means
+    /// <c>now - ErrorRetentionPeriod</c>.
+    /// </summary>
+    public DateTime? ErrorCutoff { get; set; }
+
+    /// <summary>
+    /// Cutoff applied to the event-log purge. <c>null</c> means
+    /// <c>now - EventsRetentionPeriod</c>.
+    /// </summary>
+    public DateTime? EventsCutoff { get; set; }
+}

src/ServiceControl.Api/Contracts/RetentionPurgeResponse.cs

@@ -0,0 +1,22 @@
+namespace ServiceControl.Api.Contracts;
+
+using System;
+
+/// <summary>
+/// Response body for <c>POST /api/maintenance/retention/purge</c>. The <c>Status</c> field signals the
+/// outcome: <c>started</c> (202), <c>already-running</c> (409), or
+/// <c>not-supported</c> (501).
+/// </summary>
+public class RetentionPurgeResponse
+{
+    public RetentionPurgeStatus Status { get; set; }
+
+    public DateTime? StartedAt { get; set; }
+
+    public DateTime? ErrorCutoff { get; set; }
+
+    public DateTime? EventsCutoff { get; set; }
+
+    /// <summary>A human-readable reason included when the operation is not supported.</summary>
+    public string Reason { get; set; }
+}

src/ServiceControl.Api/Contracts/RetentionPurgeStatus.cs

@@ -0,0 +1,9 @@
+namespace ServiceControl.Api.Contracts;
+
+public enum RetentionPurgeStatus
+{
+    Started,
+    AlreadyRunning,
+    NotSupported,
+    Error
+}

src/ServiceControl.Api/Contracts/RetentionPurgeStatusResponse.cs

@@ -0,0 +1,23 @@
+namespace ServiceControl.Api.Contracts;
+
+using System;
+
+/// <summary>
+/// Response body for <c>GET /api/maintenance/retention/purge/status</c>. On a persister with no sweeper
+/// (e.g. RavenDB) the endpoint returns 501 with a <see cref="Reason"/> instead.
+/// </summary>
+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; }
+
+    /// <summary>Present only on the 501 Not Implemented response.</summary>
+    public string Reason { get; set; }
+}

src/ServiceControl.Api/IRetentionApi.cs

@@ -0,0 +1,26 @@
+namespace ServiceControl.Api;
+
+using System.Threading;
+using System.Threading.Tasks;
+using Contracts;
+
+/// <summary>
+/// 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.
+/// </summary>
+public interface IRetentionApi
+{
+    /// <summary>
+    /// 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).
+    /// </summary>
+    Task<RetentionPurgeResponse> Sweep(RetentionPurgeRequest request, CancellationToken cancellationToken = default);
+
+    /// <summary>
+    /// Returns a point-in-time snapshot of purge execution state for polling.
+    /// </summary>
+    Task<RetentionPurgeStatusResponse> GetStatus(CancellationToken cancellationToken = default);
+}

src/ServiceControl.Infrastructure/Auth/MessageAction.cs

@@ -7,7 +7,8 @@ public enum MessageActionKind
     Retry,
     Archive,
     Unarchive,
-    Edit
+    Edit,
+    Delete
 }
 
 /// <summary>How the action selected the messages it acts on.</summary>

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"
             },

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);
         }
     }
+
+    /// <summary>
+    /// 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.
+    /// </summary>
+    public static async Task<T> AuditedOperation<T>(this IMessageActionAuditLog auditLog, AuditUser user,
+        MessageActionKind kind, string permission, MessageActionScope scope, string? resource,
+        int? count, string operationId, Func<CancellationToken, Task<T>> 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);
+        }
+    }
 }

src/ServiceControl.Infrastructure/Auth/Permissions.cs

@@ -58,6 +58,9 @@ public static class Permissions
     /// <summary>Event log area — viewing the event log.</summary>
     public const string ErrorEventLogView = "error:eventlog:view";
 
+    /// <summary>Retention area — manually triggering a data retention purge.</summary>
+    public const string ErrorRetentionPurge = "error:retention:purge";
+
     /// <summary>Licensing area — viewing and managing license configuration.</summary>
     public const string ErrorLicensingView = "error:licensing:view";
     /// <inheritdoc cref="ErrorLicensingView"/>

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<string, FrozenSet<string>> Roles =

src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs

@@ -40,7 +40,12 @@ protected static void RegisterDataStores(IServiceCollection services, EFPersiste
         if (settings.RunRetentionSweep)
         {
             services.AddSingleton<RetentionMetrics>();
-            services.AddHostedService<RetentionSweeper>();
+
+            // Register the sweeper as a resolvable singleton (concrete type + IRetentionSweeper) AND
+            // as a hosted service, all backed by one instance.
+            services.AddSingleton<RetentionSweeper>();
+            services.AddHostedService(sp => sp.GetRequiredService<RetentionSweeper>());
+            services.AddSingleton<IRetentionSweeper>(sp => sp.GetRequiredService<RetentionSweeper>());
         }
 
         services.AddSingleton<OperationsManager>();