Add --apply-retention to the audit instance

src/ServiceControl.Audit.Persistence.RavenDB/RavenAuditRetentionApplier.cs

@@ -0,0 +1,59 @@
+namespace ServiceControl.Audit.Persistence.RavenDB
+{
+    using System;
+    using System.Threading;
+    using System.Threading.Tasks;
+    using Microsoft.Extensions.Logging;
+    using Raven.Client;
+    using Raven.Client.Documents.Operations;
+    using Raven.Client.Documents.Queries;
+
+    class RavenAuditRetentionApplier(IRavenDocumentStoreProvider documentStoreProvider, ILogger<RavenAuditRetentionApplier> logger) : IAuditRetentionApplier
+    {
+        // Ingestion sets the expiry from the ingestion time. That time is not stored; ProcessedAt is the stored time closest to it.
+        static readonly string[] Collections = ["ProcessedMessages", "SagaSnapshots"];
+
+        public async Task<long> ApplyRetention(TimeSpan retentionPeriod, IProgress<long> progress, CancellationToken cancellationToken = default)
+        {
+            var documentStore = await documentStoreProvider.GetDocumentStore(cancellationToken);
+            long updated = 0;
+
+            foreach (var collection in Collections)
+            {
+                var query = new IndexQuery
+                {
+                    Query = $$"""
+                        from {{collection}}
+                        update
+                        {
+                            this['@metadata']['@expires'] = new Date(new Date(this.ProcessedAt).getTime() + $retentionMilliseconds).toISOString();
+                        }
+                        """,
+                    QueryParameters = new Parameters
+                    {
+                        { "retentionMilliseconds", retentionPeriod.TotalMilliseconds }
+                    }
+                };
+
+                var patch = new PatchByQueryOperation(query, new QueryOperationOptions { AllowStale = true });
+                var operation = await documentStore.Operations.SendAsync(patch, token: cancellationToken);
+
+                var updatedBefore = updated;
+                operation.OnProgressChanged += (_, operationProgress) =>
+                {
+                    if (operationProgress is DeterminateProgress determinate)
+                    {
+                        progress.Report(updatedBefore + determinate.Processed);
+                    }
+                };
+
+                var result = await operation.WaitForCompletionAsync<BulkOperationResult>(cancellationToken);
+                updated += result.Total;
+
+                logger.LogInformation("Updated the expiry of {Count:N0} {Collection} documents", result.Total, collection);
+            }
+
+            return updated;
+        }
+    }
+}

src/ServiceControl.Audit.Persistence.RavenDB/RavenPersistence.cs

@@ -30,6 +30,7 @@ public void AddPersistence(IServiceCollection services)
             services.AddSingleton<ISagaHistoryDataStore>(sp => sp.GetRequiredService<RavenAuditDataStore>());
             services.AddSingleton<IAuditIngestionUnitOfWorkFactory, RavenAuditIngestionUnitOfWorkFactory>();
             services.AddSingleton<IFailedAuditStorage, RavenFailedAuditStorage>();
+            services.AddSingleton<IAuditRetentionApplier, RavenAuditRetentionApplier>();
             services.AddSingleton<IEnvironmentDataProvider, RavenEnvironmentDataProvider>();
             services.AddSingleton<IStorageIdentityProvider, RavenStorageIdentityProvider>();
             services.AddSingleton<MinimumRequiredStorageState>();

src/ServiceControl.Audit.Persistence.Tests.RavenDB/ApplyRetentionTests.cs

@@ -0,0 +1,74 @@
+namespace ServiceControl.Audit.Persistence.Tests
+{
+    using System;
+    using System.Collections.Generic;
+    using System.Threading.Tasks;
+    using Auditing;
+    using Microsoft.Extensions.DependencyInjection;
+    using NServiceBus;
+    using NUnit.Framework;
+    using Raven.Client;
+    using SagaAudit;
+
+    [TestFixture]
+    class ApplyRetentionTests : PersistenceTestFixture
+    {
+        public override Task Setup()
+        {
+            SetSettings = s => s.AuditRetentionPeriod = IngestionRetentionPeriod;
+            return base.Setup();
+        }
+
+        [Test]
+        public async Task Sets_the_expiry_of_stored_messages_to_processed_time_plus_the_new_retention_period()
+        {
+            var message = new ProcessedMessage(
+                new Dictionary<string, string>
+                {
+                    { Headers.MessageId, "MyMessageId" },
+                    { Headers.ProcessingEndpoint, "SomeEndpoint" },
+                    { Headers.ProcessingEnded, DateTimeOffsetHelper.ToWireFormattedString(new DateTimeOffset(ProcessedAt)) }
+                },
+                new Dictionary<string, object> { { "MessageId", "MyMessageId" } });
+            var snapshot = new SagaSnapshot { SagaId = Guid.NewGuid(), ProcessedAt = ProcessedAt };
+
+            await using (var unitOfWork = await StartAuditUnitOfWork(2))
+            {
+                await unitOfWork.RecordProcessedMessage(message);
+                await unitOfWork.RecordSagaSnapshot(snapshot);
+                await unitOfWork.Complete();
+            }
+            await configuration.CompleteDBOperation();
+
+            var expiresBefore = await GetExpires<ProcessedMessage>(message.Id);
+            Assert.That(expiresBefore, Is.EqualTo(DateTime.UtcNow + IngestionRetentionPeriod).Within(TimeSpan.FromMinutes(5)), "ingestion sets the expiry from the configured retention period");
+
+            var progress = new List<long>();
+            var updated = await ServiceProvider.GetRequiredService<IAuditRetentionApplier>()
+                .ApplyRetention(NewRetentionPeriod, new SynchronousProgress(progress), TestTimeoutCancellationToken);
+
+            Assert.That(updated, Is.EqualTo(2));
+            Assert.That(await GetExpires<ProcessedMessage>(message.Id), Is.EqualTo(ProcessedAt + NewRetentionPeriod));
+            Assert.That(await GetExpires<SagaSnapshot>(snapshot.Id), Is.EqualTo(ProcessedAt + NewRetentionPeriod));
+        }
+
+        async Task<DateTime> GetExpires<T>(string id)
+        {
+            using var session = configuration.DocumentStore.OpenAsyncSession();
+            var document = await session.LoadAsync<T>(id);
+            Assert.That(document, Is.Not.Null, $"document {id} not found");
+            var expires = session.Advanced.GetMetadataFor(document).GetString(Constants.Documents.Metadata.Expires);
+            return DateTime.Parse(expires, null, System.Globalization.DateTimeStyles.AdjustToUniversal | System.Globalization.DateTimeStyles.AssumeUniversal);
+        }
+
+        // Progress<T> posts to the thread pool; this one reports inline so the test can read the list afterwards
+        sealed class SynchronousProgress(List<long> reports) : IProgress<long>
+        {
+            public void Report(long value) => reports.Add(value);
+        }
+
+        static readonly DateTime ProcessedAt = new(2026, 1, 15, 10, 30, 0, DateTimeKind.Utc);
+        static readonly TimeSpan IngestionRetentionPeriod = TimeSpan.FromDays(30);
+        static readonly TimeSpan NewRetentionPeriod = TimeSpan.FromHours(5);
+    }
+}

src/ServiceControl.Audit.Persistence/IAuditRetentionApplier.cs

@@ -0,0 +1,20 @@
+namespace ServiceControl.Audit.Persistence
+{
+    using System;
+    using System.Threading;
+    using System.Threading.Tasks;
+
+    /// <summary>
+    /// Applies the configured audit retention period to the messages and saga snapshots that are already stored.
+    /// Only a persister that stores an expiry per document at ingestion registers an implementation. A persister
+    /// that applies the retention period when it removes expired messages does not need one.
+    /// </summary>
+    public interface IAuditRetentionApplier
+    {
+        /// <summary>
+        /// Sets the expiry of every stored message and saga snapshot to its processed time plus <paramref name="retentionPeriod"/>.
+        /// Returns the number of documents that were updated.
+        /// </summary>
+        Task<long> ApplyRetention(TimeSpan retentionPeriod, IProgress<long> progress, CancellationToken cancellationToken = default);
+    }
+}

src/ServiceControl.Audit.UnitTests/Infrastructure/HostArgumentsTests.cs

@@ -0,0 +1,22 @@
+namespace ServiceControl.Audit.UnitTests.Infrastructure
+{
+    using Audit.Infrastructure.Hosting;
+    using Audit.Infrastructure.Hosting.Commands;
+    using NUnit.Framework;
+
+    [TestFixture]
+    class HostArgumentsTests
+    {
+        [Test]
+        public void No_arguments_runs_the_instance() =>
+            Assert.That(new HostArguments([]).Command, Is.EqualTo(typeof(RunCommand)));
+
+        [Test]
+        public void Apply_retention_selects_the_apply_retention_command() =>
+            Assert.That(new HostArguments(["--apply-retention"]).Command, Is.EqualTo(typeof(ApplyRetentionCommand)));
+
+        [Test]
+        public void Import_failed_audits_selects_the_import_command() =>
+            Assert.That(new HostArguments(["--import-failed-audits"]).Command, Is.EqualTo(typeof(ImportFailedAuditsCommand)));
+    }
+}

src/ServiceControl.Audit/Infrastructure/Hosting/Commands/ApplyRetentionCommand.cs

@@ -0,0 +1,67 @@
+namespace ServiceControl.Audit.Infrastructure.Hosting.Commands
+{
+    using System;
+    using System.Threading;
+    using System.Threading.Tasks;
+    using Microsoft.Extensions.DependencyInjection;
+    using Microsoft.Extensions.Hosting;
+    using Microsoft.Extensions.Logging;
+    using Persistence;
+    using ServiceControl.Infrastructure;
+    using Settings;
+
+    /// <summary>
+    /// Applies the configured audit retention period to the stored messages. Run it after a change of
+    /// <c>ServiceControl.Audit/AuditRetentionPeriod</c>: ingestion sets the expiry of each message when it is
+    /// stored, so the stored messages keep the old retention period until this command rewrites their expiry.
+    /// </summary>
+    class ApplyRetentionCommand : AbstractCommand
+    {
+        public override async Task Execute(HostArguments args, Settings settings, CancellationToken cancellationToken = default)
+        {
+            var logger = LoggerUtil.CreateStaticLogger<ApplyRetentionCommand>();
+
+            var persistenceConfiguration = PersistenceConfigurationFactory.LoadPersistenceConfiguration(settings);
+            var persistenceSettings = persistenceConfiguration.BuildPersistenceSettings(settings);
+
+            var hostBuilder = Host.CreateApplicationBuilder();
+            hostBuilder.Services.AddPersistence(persistenceSettings, persistenceConfiguration);
+
+            using var tokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
+            Console.CancelKeyPress += (_, e) =>
+            {
+                e.Cancel = true;
+                tokenSource.Cancel();
+            };
+
+            using var host = hostBuilder.Build();
+            await host.StartAsync(tokenSource.Token);
+
+            try
+            {
+                var applier = host.Services.GetService<IAuditRetentionApplier>();
+
+                if (applier == null)
+                {
+                    logger.LogInformation("The {PersistenceType} persister applies the retention period when it removes expired messages, so the stored messages need no update", settings.PersistenceType);
+                    return;
+                }
+
+                logger.LogInformation("Applying the audit retention period of {RetentionPeriod} to the stored messages", settings.AuditRetentionPeriod);
+
+                var progress = new Progress<long>(count => logger.LogInformation("{Count:N0} documents updated so far", count));
+                var updated = await applier.ApplyRetention(settings.AuditRetentionPeriod, progress, tokenSource.Token);
+
+                logger.LogInformation("Done. Updated the expiry of {Count:N0} documents", updated);
+            }
+            catch (OperationCanceledException e) when (tokenSource.Token.IsCancellationRequested)
+            {
+                logger.LogInformation(e, "Cancelled");
+            }
+            finally
+            {
+                await host.StopAsync(CancellationToken.None);
+            }
+        }
+    }
+}

src/ServiceControl.Audit/Infrastructure/Hosting/Help.txt

@@ -10,6 +10,19 @@ The REST API, message importers and the background document expiry are all disab
 
 This mode is only supported when run interactively.
 
+APPLY RETENTION
+
+   ServiceControl.Audit.exe --apply-retention
+
+Applies the configured audit retention period (ServiceControl.Audit/AuditRetentionPeriod) to the messages and saga
+snapshots that are already stored. Ingestion sets the expiry of each message from the retention period that is
+configured at that moment, so a change of the retention period only affects new messages until this command runs.
+The new expiry of each message is its processed time plus the retention period. Messages that are already past the new
+expiry are removed by the next expiration run.
+
+Stop the instance before running this command. A persister that applies the retention period when it removes expired
+messages needs no update, and the command reports that.
+
 SERVICE INSTALL AND UNINSTALL AND CONFIGURATION OPTIONS
 
 As of Service Control 1.7 the command line  uninstall and install switches have been removed.

src/ServiceControl.Audit/Infrastructure/Hosting/HostArguments.cs

@@ -53,6 +53,15 @@ public HostArguments(string[] args)
                 }
             };
 
+            var applyRetentionOptions = new OptionSet
+            {
+                {
+                    "apply-retention",
+                    "Apply the configured audit retention period to the stored messages",
+                    s => Command = typeof(ApplyRetentionCommand)
+                }
+            };
+
             try
             {
                 externalInstallerOptions.Parse(args);
@@ -76,6 +85,13 @@ public HostArguments(string[] args)
                     return;
                 }
 
+                applyRetentionOptions.Parse(args);
+
+                if (Command == typeof(ApplyRetentionCommand))
+                {
+                    return;
+                }
+
                 defaultOptions.Parse(args);
             }
             catch (Exception e)