Continue cancellation token propagation in audit persistence layer

This change expands the consistent application of optional `CancellationToken` parameters to asynchronous methods within the audit persistence interfaces and their InMemory and RavenDB implementations.

Cancellation tokens are now propagated to underlying asynchronous database calls and consuming services. This ensures proper cancellation of long-running operations and further reduces cancellation analyzer debt by removing suppressions from related `.editorconfig` files.

Author: johnsimons

src/ServiceControl.Audit.Persistence.InMemory/.editorconfig

@@ -2,9 +2,3 @@
 
 # Justification: ServiceControl app has no synchronization context
 dotnet_diagnostic.CA2007.severity = none
-
-# Cancellation analyzer debt. These fire because Particular.Analyzers is no longer pinned to 0.9.0.
-# They are scheduled work, not accepted exceptions: remove a line once this project has no
-# violations of that rule left, and never add a rule back to this list.
-dotnet_diagnostic.PS0003.severity = none
-dotnet_diagnostic.PS0018.severity = none

src/ServiceControl.Audit.Persistence.InMemory/InMemoryAttachmentsBodyStorage.cs

@@ -16,7 +16,7 @@ public InMemoryAttachmentsBodyStorage()
             messageBodies = [];
         }
 
-        public Task Store(string bodyId, string contentType, int bodySize, Stream bodyStream, CancellationToken cancellationToken)
+        public Task Store(string bodyId, string contentType, int bodySize, Stream bodyStream, CancellationToken cancellationToken = default)
         {
             var messageBody = messageBodies.FirstOrDefault(w => w.BodyId == bodyId);
 
@@ -44,7 +44,7 @@ public Task Store(string bodyId, string contentType, int bodySize, Stream bodySt
             return Task.CompletedTask;
         }
 
-        public async Task<StreamResult> TryFetch(string bodyId, CancellationToken cancellationToken)
+        public async Task<StreamResult> TryFetch(string bodyId, CancellationToken cancellationToken = default)
         {
             var messageBody = messageBodies.FirstOrDefault(w => w.BodyId == bodyId);
 

src/ServiceControl.Audit.Persistence.InMemory/InMemoryAuditDataStore.cs

@@ -25,7 +25,7 @@ public InMemoryAuditDataStore(IBodyStorage bodyStorage)
             failedAuditImports = [];
         }
 
-        public async Task<QueryResult<SagaHistory>> QuerySagaHistoryById(Guid input, CancellationToken cancellationToken)
+        public async Task<QueryResult<SagaHistory>> QuerySagaHistoryById(Guid input, CancellationToken cancellationToken = default)
         {
             var sagaHistory = sagaHistories.FirstOrDefault(w => w.SagaId == input);
 
@@ -37,7 +37,7 @@ public async Task<QueryResult<SagaHistory>> QuerySagaHistoryById(Guid input, Can
             return await Task.FromResult(new QueryResult<SagaHistory>(sagaHistory, new QueryStatsInfo(string.Empty, 1)));
         }
 
-        public async Task<QueryResult<IList<MessagesView>>> GetMessages(bool includeSystemMessages, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange, CancellationToken cancellationToken)
+        public async Task<QueryResult<IList<MessagesView>>> GetMessages(bool includeSystemMessages, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange, CancellationToken cancellationToken = default)
         {
             var matched = messageViews
                 .Where(w => (!w.IsSystemMessage || includeSystemMessages) &&
@@ -48,7 +48,7 @@ public async Task<QueryResult<IList<MessagesView>>> GetMessages(bool includeSyst
             return await Task.FromResult(new QueryResult<IList<MessagesView>>(matched, new QueryStatsInfo(string.Empty, matched.Count)));
         }
 
-        public async Task<QueryResult<IList<MessagesView>>> QueryMessages(string keyword, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange, CancellationToken cancellationToken)
+        public async Task<QueryResult<IList<MessagesView>>> QueryMessages(string keyword, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange, CancellationToken cancellationToken = default)
         {
             var messages = GetMessageIdsMatchingQuery(keyword);
 
@@ -60,7 +60,7 @@ public async Task<QueryResult<IList<MessagesView>>> QueryMessages(string keyword
             return await Task.FromResult(new QueryResult<IList<MessagesView>>(matched, new QueryStatsInfo(string.Empty, matched.Count())));
         }
 
-        public async Task<QueryResult<IList<MessagesView>>> QueryMessagesByReceivingEndpointAndKeyword(string endpoint, string keyword, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange, CancellationToken cancellationToken)
+        public async Task<QueryResult<IList<MessagesView>>> QueryMessagesByReceivingEndpointAndKeyword(string endpoint, string keyword, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange, CancellationToken cancellationToken = default)
         {
             var messages = GetMessageIdsMatchingQuery(keyword);
 
@@ -71,7 +71,7 @@ public async Task<QueryResult<IList<MessagesView>>> QueryMessagesByReceivingEndp
             return await Task.FromResult(new QueryResult<IList<MessagesView>>(matched, new QueryStatsInfo(string.Empty, matched.Count)));
         }
 
-        public async Task<QueryResult<IList<MessagesView>>> QueryMessagesByReceivingEndpoint(bool includeSystemMessages, string endpointName, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange, CancellationToken cancellationToken)
+        public async Task<QueryResult<IList<MessagesView>>> QueryMessagesByReceivingEndpoint(bool includeSystemMessages, string endpointName, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange, CancellationToken cancellationToken = default)
         {
             var matched = messageViews.Where(w => w.ReceivingEndpoint.Name == endpointName &&
                     (timeSentRange == null || !timeSentRange.From.HasValue || w.TimeSent >= timeSentRange.From.Value) &&
@@ -80,15 +80,15 @@ public async Task<QueryResult<IList<MessagesView>>> QueryMessagesByReceivingEndp
             return await Task.FromResult(new QueryResult<IList<MessagesView>>(matched, new QueryStatsInfo(string.Empty, matched.Count)));
         }
 
-        public async Task<QueryResult<IList<MessagesView>>> QueryMessagesByConversationId(string conversationId, PagingInfo pagingInfo, SortInfo sortInfo, CancellationToken cancellationToken)
+        public async Task<QueryResult<IList<MessagesView>>> QueryMessagesByConversationId(string conversationId, PagingInfo pagingInfo, SortInfo sortInfo, CancellationToken cancellationToken = default)
         {
             var matched = messageViews.Where(w => w.ConversationId == conversationId).ToList();
             return await Task.FromResult(new QueryResult<IList<MessagesView>>(matched, new QueryStatsInfo(string.Empty, matched.Count)));
         }
 
-        public async Task<MessageBodyView> GetMessageBody(string messageId, CancellationToken cancellationToken)
+        public async Task<MessageBodyView> GetMessageBody(string messageId, CancellationToken cancellationToken = default)
         {
-            var result = await GetMessageBodyFromMetadata(messageId);
+            var result = await GetMessageBodyFromMetadata(messageId, cancellationToken);
 
             if (!result.Found)
             {
@@ -144,7 +144,7 @@ async Task<MessageBodyView> GetMessageBodyFromAttachments(string messageId, Canc
             return MessageBodyView.NotFound();
         }
 
-        Task<MessageBodyView> GetMessageBodyFromMetadata(string messageId)
+        Task<MessageBodyView> GetMessageBodyFromMetadata(string messageId, CancellationToken cancellationToken)
         {
             var message = processedMessages.FirstOrDefault(pm => (pm.MessageMetadata["MessageId"] as string) == messageId);
 
@@ -171,7 +171,7 @@ Task<MessageBodyView> GetMessageBodyFromMetadata(string messageId)
             return Task.FromResult(MessageBodyView.FromString(body, contentType, bodySize, messageId));
         }
 
-        public Task<QueryResult<IList<AuditCount>>> QueryAuditCounts(string endpointName, CancellationToken cancellationToken)
+        public Task<QueryResult<IList<AuditCount>>> QueryAuditCounts(string endpointName, CancellationToken cancellationToken = default)
         {
             var results = messageViews
                 .Where(m => m.ReceivingEndpoint.Name == endpointName && !m.IsSystemMessage)
@@ -187,7 +187,7 @@ public Task<QueryResult<IList<AuditCount>>> QueryAuditCounts(string endpointName
             return Task.FromResult(new QueryResult<IList<AuditCount>>(results, QueryStatsInfo.Zero));
         }
 
-        public Task SaveProcessedMessage(ProcessedMessage processedMessage)
+        public Task SaveProcessedMessage(ProcessedMessage processedMessage, CancellationToken cancellationToken = default)
         {
             if (processedMessages.Any(pm => pm.Id == processedMessage.Id))
             {
@@ -200,7 +200,7 @@ public Task SaveProcessedMessage(ProcessedMessage processedMessage)
             return Task.CompletedTask;
         }
 
-        public Task SaveSagaSnapshot(SagaSnapshot sagaSnapshot)
+        public Task SaveSagaSnapshot(SagaSnapshot sagaSnapshot, CancellationToken cancellationToken = default)
         {
             var sagaHistory = sagaHistories.SingleOrDefault(sh => sh.SagaId == sagaSnapshot.SagaId);
 
@@ -240,7 +240,7 @@ object TryGet(Dictionary<string, object> metadata, string key)
             return null;
         }
 
-        public Task Setup() => Task.CompletedTask;
+        public Task Setup(CancellationToken cancellationToken = default) => Task.CompletedTask;
 
         List<MessagesView> messageViews;
         List<ProcessedMessage> processedMessages;

src/ServiceControl.Audit.Persistence.InMemory/InMemoryAuditIngestionUnitOfWork.cs

@@ -15,15 +15,15 @@ class InMemoryAuditIngestionUnitOfWork(
     {
         public ValueTask DisposeAsync() => ValueTask.CompletedTask;
 
-        public async Task RecordProcessedMessage(ProcessedMessage processedMessage, ReadOnlyMemory<byte> body, CancellationToken cancellationToken)
+        public async Task RecordProcessedMessage(ProcessedMessage processedMessage, ReadOnlyMemory<byte> body, CancellationToken cancellationToken = default)
         {
             if (!body.IsEmpty)
             {
                 await bodyStorageEnricher.StoreAuditMessageBody(body, processedMessage, cancellationToken);
             }
-            await dataStore.SaveProcessedMessage(processedMessage);
+            await dataStore.SaveProcessedMessage(processedMessage, cancellationToken);
         }
 
-        public Task RecordSagaSnapshot(SagaSnapshot sagaSnapshot, CancellationToken cancellationToken) => dataStore.SaveSagaSnapshot(sagaSnapshot);
+        public Task RecordSagaSnapshot(SagaSnapshot sagaSnapshot, CancellationToken cancellationToken = default) => dataStore.SaveSagaSnapshot(sagaSnapshot, cancellationToken);
     }
 }

src/ServiceControl.Audit.Persistence.InMemory/InMemoryAuditIngestionUnitOfWorkFactory.cs

@@ -7,7 +7,7 @@
 
     class InMemoryAuditIngestionUnitOfWorkFactory(InMemoryAuditDataStore dataStore, BodyStorageEnricher bodyStorageEnricher) : IAuditIngestionUnitOfWorkFactory
     {
-        public ValueTask<IAuditIngestionUnitOfWork> StartNew(int batchSize, CancellationToken cancellationToken)
+        public ValueTask<IAuditIngestionUnitOfWork> StartNew(int batchSize, CancellationToken cancellationToken = default)
         {
             //The batchSize argument is ignored: the in-memory storage implementation doesn't support batching.
             return new ValueTask<IAuditIngestionUnitOfWork>(new InMemoryAuditIngestionUnitOfWork(dataStore, bodyStorageEnricher));

src/ServiceControl.Audit.Persistence.InMemory/InMemoryFailedAuditStorage.cs

@@ -7,10 +7,13 @@
 
     class InMemoryFailedAuditStorage(InMemoryAuditDataStore dataStore) : IFailedAuditStorage
     {
-        public async Task ProcessFailedMessages(Func<FailedTransportMessage, Func<CancellationToken, Task>, CancellationToken, Task> onMessage, CancellationToken cancellationToken)
+        public async Task ProcessFailedMessages(Func<FailedTransportMessage, Func<CancellationToken, Task>, CancellationToken, Task> onMessage, CancellationToken cancellationToken = default)
         {
+            cancellationToken.ThrowIfCancellationRequested();
+
             foreach (var failedMessage in dataStore.failedAuditImports)
             {
+                cancellationToken.ThrowIfCancellationRequested();
                 FailedTransportMessage transportMessage = failedMessage.Message;
 
                 await onMessage(transportMessage, _ => Task.CompletedTask, cancellationToken);
@@ -19,12 +22,17 @@ public async Task ProcessFailedMessages(Func<FailedTransportMessage, Func<Cancel
             dataStore.failedAuditImports.Clear();
         }
 
-        public Task SaveFailedAuditImport(FailedAuditImport message)
+        public Task SaveFailedAuditImport(FailedAuditImport message, CancellationToken cancellationToken = default)
         {
+            cancellationToken.ThrowIfCancellationRequested();
             dataStore.failedAuditImports.Add(message);
             return Task.CompletedTask;
         }
 
-        public Task<int> GetFailedAuditsCount() => Task.FromResult(dataStore.failedAuditImports.Count);
+        public Task<int> GetFailedAuditsCount(CancellationToken cancellationToken = default)
+        {
+            cancellationToken.ThrowIfCancellationRequested();
+            return Task.FromResult(dataStore.failedAuditImports.Count);
+        }
     }
 }

src/ServiceControl.Audit.Persistence.RavenDB/.editorconfig

@@ -2,10 +2,3 @@
 
 # Justification: ServiceControl app has no synchronization context
 dotnet_diagnostic.CA2007.severity = none
-
-# Cancellation analyzer debt. These fire because Particular.Analyzers is no longer pinned to 0.9.0.
-# They are scheduled work, not accepted exceptions: remove a line once this project has no
-# violations of that rule left, and never add a rule back to this list.
-dotnet_diagnostic.PS0003.severity = none
-dotnet_diagnostic.PS0018.severity = none
-dotnet_diagnostic.PS0020.severity = none

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

@@ -19,7 +19,7 @@
 
 class DatabaseSetup(DatabaseConfiguration configuration)
 {
-    public async Task Execute(IDocumentStore documentStore, CancellationToken cancellationToken)
+    public async Task Execute(IDocumentStore documentStore, CancellationToken cancellationToken = default)
     {
         await CreateDatabase(documentStore, configuration.Name, cancellationToken);
 
@@ -66,7 +66,7 @@ async Task UpdateDatabaseSettings(IDocumentStore documentStore, string databaseN
         await documentStore.Maintenance.Server.SendAsync(new ToggleDatabasesStateOperation(databaseName, false), cancellationToken);
     }
 
-    public static async Task DeleteLegacySagaDetailsIndex(IDocumentStore documentStore, CancellationToken cancellationToken)
+    public static async Task DeleteLegacySagaDetailsIndex(IDocumentStore documentStore, CancellationToken cancellationToken = default)
     {
         // If the SagaDetailsIndex exists but does not have a .Take(50000), then we remove the current SagaDetailsIndex and
         // create a new one. If we do not remove the current one, then RavenDB will attempt to do a side-by-side migration.
@@ -82,7 +82,7 @@ public static async Task DeleteLegacySagaDetailsIndex(IDocumentStore documentSto
         }
     }
 
-    internal static async Task CreateIndexes(IDocumentStore documentStore, bool enableFreeTextSearch, CancellationToken cancellationToken)
+    internal static async Task CreateIndexes(IDocumentStore documentStore, bool enableFreeTextSearch, CancellationToken cancellationToken = default)
     {
         await DeleteLegacySagaDetailsIndex(documentStore, cancellationToken);
 

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

@@ -10,7 +10,7 @@ static class LicenseStatusCheck
 {
     record LicenseStatusFragment(string Id, string LicensedTo, string Status, bool Expired);
 
-    public static async Task WaitForLicenseOrThrow(IDocumentStore documentStore, CancellationToken cancellationToken)
+    public static async Task WaitForLicenseOrThrow(IDocumentStore documentStore, CancellationToken cancellationToken = default)
     {
         var ravenConfiguredHttpClient = documentStore.GetRequestExecutor().HttpClient;
         var licenseCheckUrl = documentStore.Urls[0].TrimEnd('/') + "/license/status";
@@ -37,9 +37,19 @@ public static async Task WaitForLicenseOrThrow(IDocumentStore documentStore, Can
                 await Task.Delay(200, cts.Token);
             }
         }
-        catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested)
+        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
         {
-            throw new InvalidOperationException("Cannot validate the current RavenDB license. Please, contact support");
+            throw;
+        }
+
+        catch (Exception exception)
+        {
+            if (exception is OperationCanceledException && cts.Token.IsCancellationRequested)
+            {
+                throw new InvalidOperationException("Cannot validate the current RavenDB license. Please, contact support");
+            }
+
+            throw;
         }
     }
 }

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

@@ -12,7 +12,7 @@ class RavenAttachmentsBodyStorage(
         int settingsMaxBodySizeToStore)
         : IBodyStorage
     {
-        public Task Store(string bodyId, string contentType, int bodySize, Stream bodyStream, CancellationToken cancellationToken)
+        public Task Store(string bodyId, string contentType, int bodySize, Stream bodyStream, CancellationToken cancellationToken = default)
         {
             if (bodySize > settingsMaxBodySizeToStore)
             {
@@ -23,7 +23,7 @@ public Task Store(string bodyId, string contentType, int bodySize, Stream bodySt
                 .StoreAsync("body", bodyStream, contentType, cancellationToken);
         }
 
-        public async Task<StreamResult> TryFetch(string bodyId, CancellationToken cancellationToken)
+        public async Task<StreamResult> TryFetch(string bodyId, CancellationToken cancellationToken = default)
         {
             using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken);
             var result = await session.Advanced.Attachments.GetAsync($"MessageBodies/{bodyId}", "body", cancellationToken);