Implements audit data retention policy

Implements data retention policy for audit messages and saga snapshots using a background service.

This change introduces a base `RetentionCleaner` class that handles the logic for deleting expired audit data in batches.  Database-specific implementations are provided for SQL Server and PostgreSQL, leveraging their respective locking mechanisms (sp_getapplock and advisory locks) to prevent concurrent executions of the cleanup process.

Removes the registration of the `RetentionCleaner` from the base class and registers it on specific implementations.

The cleanup process deletes processed messages and saga snapshots older than the configured retention period, optimizing database space and improving query performance.

Author: johnsimons

src/ServiceControl.Audit.Persistence.Sql.Core/Abstractions/BaseAuditPersistence.cs

@@ -26,6 +26,5 @@ protected static void RegisterDataStores(IServiceCollection services, AuditSqlPe
         services.AddSingleton<IFailedAuditStorage, EFFailedAuditStorage>();
         services.AddSingleton<IAuditIngestionUnitOfWorkFactory, AuditIngestionUnitOfWorkFactory>();
         services.AddSingleton(TimeProvider.System);
-        services.AddHostedService<RetentionCleaner>();
     }
 }

src/ServiceControl.Audit.Persistence.Sql.Core/Infrastructure/RetentionCleaner.cs

@@ -8,16 +8,17 @@ namespace ServiceControl.Audit.Persistence.Sql.Core.Infrastructure;
 using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Logging;
 
-public class RetentionCleaner(
-    ILogger<RetentionCleaner> logger,
+public abstract class RetentionCleaner(
+    ILogger logger,
     TimeProvider timeProvider,
     IServiceScopeFactory serviceScopeFactory,
     AuditSqlPersisterSettings settings,
     IBodyStoragePersistence bodyPersistence) : BackgroundService
 {
+    protected const int BatchSize = 250;
+
     protected override async Task ExecuteAsync(CancellationToken stoppingToken)
     {
-        // NO-OPS per requirements
         logger.LogInformation("Starting {ServiceName}", nameof(RetentionCleaner));
 
         try
@@ -46,25 +47,53 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
 
     async Task Clean(CancellationToken stoppingToken)
     {
-        var stopwatch = Stopwatch.StartNew();
-        const int batchSize = 250;
-
         using var scope = serviceScopeFactory.CreateScope();
         var dbContext = scope.ServiceProvider.GetRequiredService<AuditDbContextBase>();
 
+        await using var transaction = await dbContext.Database.BeginTransactionAsync(stoppingToken);
+
+        if (!await TryAcquireLock(dbContext, stoppingToken))
+        {
+            logger.LogDebug("Another instance is running retention cleanup, skipping this cycle");
+            return;
+        }
+
+        var stopwatch = Stopwatch.StartNew();
         var cutoff = timeProvider.GetUtcNow().UtcDateTime - settings.AuditRetentionPeriod;
 
         var totalDeletedMessages = 0;
+        var totalDeletedSnapshots = 0;
+
+        try
+        {
+            totalDeletedMessages = await CleanProcessedMessages(dbContext, cutoff, stoppingToken);
+            totalDeletedSnapshots = await CleanSagaSnapshots(dbContext, cutoff, stoppingToken);
+            await transaction.CommitAsync(stoppingToken);
+        }
+        catch
+        {
+            await transaction.RollbackAsync(stoppingToken);
+            throw;
+        }
+
+        logger.LogInformation("Retention cleanup removed {Messages} messages and {Snapshots} saga snapshots in {Elapsed}",
+            totalDeletedMessages, totalDeletedSnapshots, stopwatch.Elapsed.ToString(@"hh\:mm\:ss"));
+    }
+
+    async Task<int> CleanProcessedMessages(AuditDbContextBase dbContext, DateTime cutoff, CancellationToken stoppingToken)
+    {
+        var totalDeleted = 0;
         int deleted;
 
         do
         {
             // Get a batch of IDs to delete so we can clean up body files
+            // Note: No OrderBy - we just need any expired records, not necessarily the oldest first.
+            // Ordering would require sorting potentially millions of rows and cause timeouts.
             var messageIdsToDelete = await dbContext.ProcessedMessages
                 .Where(m => m.ProcessedAt < cutoff)
-                .OrderBy(m => m.ProcessedAt)
                 .Select(m => m.UniqueMessageId)
-                .Take(batchSize)
+                .Take(BatchSize)
                 .ToListAsync(stoppingToken);
 
             if (messageIdsToDelete.Count == 0)
@@ -80,18 +109,23 @@ async Task Clean(CancellationToken stoppingToken)
                 .Where(m => messageIdsToDelete.Contains(m.UniqueMessageId))
                 .ExecuteDeleteAsync(stoppingToken);
 
-            totalDeletedMessages += deleted;
-        } while (deleted == batchSize);
+            totalDeleted += deleted;
+        } while (deleted == BatchSize);
 
-        var totalDeletedSnapshots = 0;
+        return totalDeleted;
+    }
+
+    async Task<int> CleanSagaSnapshots(AuditDbContextBase dbContext, DateTime cutoff, CancellationToken stoppingToken)
+    {
+        var totalDeleted = 0;
+        int deleted;
 
         do
         {
             var snapshotIdsToDelete = await dbContext.SagaSnapshots
                 .Where(s => s.ProcessedAt < cutoff)
-                .OrderBy(s => s.ProcessedAt)
                 .Select(s => s.Id)
-                .Take(batchSize)
+                .Take(BatchSize)
                 .ToListAsync(stoppingToken);
 
             if (snapshotIdsToDelete.Count == 0)
@@ -103,9 +137,11 @@ async Task Clean(CancellationToken stoppingToken)
                 .Where(s => snapshotIdsToDelete.Contains(s.Id))
                 .ExecuteDeleteAsync(stoppingToken);
 
-            totalDeletedSnapshots += deleted;
-        } while (deleted == batchSize);
+            totalDeleted += deleted;
+        } while (deleted == BatchSize);
 
-        logger.LogInformation("Retention cleanup removed {Messages} messages and {Snapshots} saga snapshots in {Elapsed}", totalDeletedMessages, totalDeletedSnapshots, stopwatch.Elapsed.ToString(@"hh\:mm\:ss"));
+        return totalDeleted;
     }
+
+    protected abstract Task<bool> TryAcquireLock(AuditDbContextBase dbContext, CancellationToken stoppingToken);
 }

src/ServiceControl.Audit.Persistence.Sql.PostgreSQL/Infrastructure/RetentionCleaner.cs

@@ -0,0 +1,28 @@
+namespace ServiceControl.Audit.Persistence.Sql.PostgreSQL.Infrastructure;
+
+using Core.Abstractions;
+using Core.DbContexts;
+using Core.Infrastructure;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
+
+class RetentionCleaner(
+    ILogger<RetentionCleaner> logger,
+    TimeProvider timeProvider,
+    IServiceScopeFactory serviceScopeFactory,
+    AuditSqlPersisterSettings settings,
+    IBodyStoragePersistence bodyPersistence)
+    : Core.Infrastructure.RetentionCleaner(logger, timeProvider, serviceScopeFactory, settings, bodyPersistence)
+{
+    protected override async Task<bool> TryAcquireLock(AuditDbContextBase dbContext, CancellationToken stoppingToken)
+    {
+        // Use PostgreSQL's advisory lock for distributed locking
+        // pg_try_advisory_xact_lock returns true if lock acquired, false otherwise
+        // The lock is automatically released when the transaction ends
+        var sql = "SELECT pg_try_advisory_xact_lock(hashtext('retention_cleaner'))";
+
+        var result = await dbContext.Database.SqlQueryRaw<bool>(sql).FirstOrDefaultAsync(stoppingToken);
+        return result;
+    }
+}

src/ServiceControl.Audit.Persistence.Sql.PostgreSQL/PostgreSqlAuditPersistence.cs

@@ -22,6 +22,7 @@ public void AddPersistence(IServiceCollection services)
         RegisterDataStores(services, settings);
         services.AddSingleton<IAuditFullTextSearchProvider, PostgreSqlAuditFullTextSearchProvider>();
         services.AddHostedService<KnownEndpointsReconciler>();
+        services.AddHostedService<RetentionCleaner>();
     }
 
     public void AddInstaller(IServiceCollection services)

src/ServiceControl.Audit.Persistence.Sql.SqlServer/Infrastructure/RetentionCleaner.cs

@@ -0,0 +1,36 @@
+namespace ServiceControl.Audit.Persistence.Sql.SqlServer.Infrastructure;
+
+using Core.Abstractions;
+using Core.DbContexts;
+using Core.Infrastructure;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
+
+class RetentionCleaner(
+    ILogger<RetentionCleaner> logger,
+    TimeProvider timeProvider,
+    IServiceScopeFactory serviceScopeFactory,
+    AuditSqlPersisterSettings settings,
+    IBodyStoragePersistence bodyPersistence)
+    : Core.Infrastructure.RetentionCleaner(logger, timeProvider, serviceScopeFactory, settings, bodyPersistence)
+{
+    protected override async Task<bool> TryAcquireLock(AuditDbContextBase dbContext, CancellationToken stoppingToken)
+    {
+        // Use SQL Server's sp_getapplock for distributed locking
+        // LockTimeout = 0 means return immediately if lock cannot be acquired
+        // Returns >= 0 on success, < 0 on failure
+        var sql = @"
+            DECLARE @lockResult INT;
+            EXEC @lockResult = sp_getapplock
+                @Resource = 'retention_cleaner',
+                @LockMode = 'Exclusive',
+                @LockOwner = 'Transaction',
+                @LockTimeout = 0;
+            SELECT @lockResult;
+        ";
+
+        var result = await dbContext.Database.SqlQueryRaw<int>(sql).FirstOrDefaultAsync(stoppingToken);
+        return result >= 0;
+    }
+}

src/ServiceControl.Audit.Persistence.Sql.SqlServer/SqlServerAuditPersistence.cs

@@ -22,6 +22,7 @@ public void AddPersistence(IServiceCollection services)
         RegisterDataStores(services, settings);
         services.AddSingleton<IAuditFullTextSearchProvider, SqlServerAuditFullTextSearchProvider>();
         services.AddHostedService<KnownEndpointsReconciler>();
+        services.AddHostedService<RetentionCleaner>();
     }
 
     public void AddInstaller(IServiceCollection services)