Continue cancellation token propagation in audit persistence layer

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

@@ -3,9 +3,8 @@
 # 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
+# WaitForLicenseOrThrow passes only the linked cts.Token into its try, so PS0020 cannot recognise a
+# filter on the caller's token. Removing this needs the license-check exception mapping reworked, not
+# a token added, so it is deliberately scoped to the one file rather than the project.
+[LicenseStatusCheck.cs]
 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";

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);