From 99f8e7932329bb8f7866f36f95e8c0965d61d31a Mon Sep 17 00:00:00 2001 From: John Simons Date: Tue, 11 Aug 2026 16:47:28 +1000 Subject: [PATCH] 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. --- .../.editorconfig | 6 ----- .../InMemoryAttachmentsBodyStorage.cs | 4 +-- .../InMemoryAuditDataStore.cs | 26 +++++++++---------- .../InMemoryAuditIngestionUnitOfWork.cs | 6 ++--- ...InMemoryAuditIngestionUnitOfWorkFactory.cs | 2 +- .../InMemoryFailedAuditStorage.cs | 14 +++++++--- .../.editorconfig | 9 +++---- .../DatabaseSetup.cs | 6 ++--- .../LicenseStatusCheck.cs | 2 +- .../RavenAttachmentsBodyStorage.cs | 4 +-- .../RavenAuditDataStore.cs | 16 ++++++------ .../RavenFailedAuditStorage.cs | 16 ++++++------ .../RavenPersistenceLifecycleHostedService.cs | 4 +-- .../RavenAuditIngestionUnitOfWork.cs | 4 +-- .../UnitOfWork/RavenAuditUnitOfWorkFactory.cs | 2 +- .../.editorconfig | 6 +++++ .../.editorconfig | 6 ----- .../BodyStorageEnricher.cs | 2 +- .../IAuditDataStore.cs | 8 +++--- .../IBodyStorage.cs | 4 +-- .../IFailedAuditStorage.cs | 6 ++--- .../IAuditIngestionUnitOfWorkFactory.cs | 2 +- src/ServiceControl.Audit/.editorconfig | 5 ---- .../Auditing/AuditIngestionFaultPolicy.cs | 2 +- .../Auditing/FailedAuditImportCustomCheck.cs | 2 +- .../Auditing/ImportFailedAudits.cs | 6 ++--- 26 files changed, 82 insertions(+), 88 deletions(-) diff --git a/src/ServiceControl.Audit.Persistence.InMemory/.editorconfig b/src/ServiceControl.Audit.Persistence.InMemory/.editorconfig index 522a310bff..ff993b49bb 100644 --- a/src/ServiceControl.Audit.Persistence.InMemory/.editorconfig +++ b/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 diff --git a/src/ServiceControl.Audit.Persistence.InMemory/InMemoryAttachmentsBodyStorage.cs b/src/ServiceControl.Audit.Persistence.InMemory/InMemoryAttachmentsBodyStorage.cs index 2408cb4bf6..9fbf8f5d34 100644 --- a/src/ServiceControl.Audit.Persistence.InMemory/InMemoryAttachmentsBodyStorage.cs +++ b/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 TryFetch(string bodyId, CancellationToken cancellationToken) + public async Task TryFetch(string bodyId, CancellationToken cancellationToken = default) { var messageBody = messageBodies.FirstOrDefault(w => w.BodyId == bodyId); diff --git a/src/ServiceControl.Audit.Persistence.InMemory/InMemoryAuditDataStore.cs b/src/ServiceControl.Audit.Persistence.InMemory/InMemoryAuditDataStore.cs index d147f5afca..81281db096 100644 --- a/src/ServiceControl.Audit.Persistence.InMemory/InMemoryAuditDataStore.cs +++ b/src/ServiceControl.Audit.Persistence.InMemory/InMemoryAuditDataStore.cs @@ -25,7 +25,7 @@ public InMemoryAuditDataStore(IBodyStorage bodyStorage) failedAuditImports = []; } - public async Task> QuerySagaHistoryById(Guid input, CancellationToken cancellationToken) + public async Task> QuerySagaHistoryById(Guid input, CancellationToken cancellationToken = default) { var sagaHistory = sagaHistories.FirstOrDefault(w => w.SagaId == input); @@ -37,7 +37,7 @@ public async Task> QuerySagaHistoryById(Guid input, Can return await Task.FromResult(new QueryResult(sagaHistory, new QueryStatsInfo(string.Empty, 1))); } - public async Task>> GetMessages(bool includeSystemMessages, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange, CancellationToken cancellationToken) + public async Task>> 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>> GetMessages(bool includeSyst return await Task.FromResult(new QueryResult>(matched, new QueryStatsInfo(string.Empty, matched.Count))); } - public async Task>> QueryMessages(string keyword, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange, CancellationToken cancellationToken) + public async Task>> QueryMessages(string keyword, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange, CancellationToken cancellationToken = default) { var messages = GetMessageIdsMatchingQuery(keyword); @@ -60,7 +60,7 @@ public async Task>> QueryMessages(string keyword return await Task.FromResult(new QueryResult>(matched, new QueryStatsInfo(string.Empty, matched.Count()))); } - public async Task>> QueryMessagesByReceivingEndpointAndKeyword(string endpoint, string keyword, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange, CancellationToken cancellationToken) + public async Task>> 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>> QueryMessagesByReceivingEndp return await Task.FromResult(new QueryResult>(matched, new QueryStatsInfo(string.Empty, matched.Count))); } - public async Task>> QueryMessagesByReceivingEndpoint(bool includeSystemMessages, string endpointName, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange, CancellationToken cancellationToken) + public async Task>> 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>> QueryMessagesByReceivingEndp return await Task.FromResult(new QueryResult>(matched, new QueryStatsInfo(string.Empty, matched.Count))); } - public async Task>> QueryMessagesByConversationId(string conversationId, PagingInfo pagingInfo, SortInfo sortInfo, CancellationToken cancellationToken) + public async Task>> 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>(matched, new QueryStatsInfo(string.Empty, matched.Count))); } - public async Task GetMessageBody(string messageId, CancellationToken cancellationToken) + public async Task 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 GetMessageBodyFromAttachments(string messageId, Canc return MessageBodyView.NotFound(); } - Task GetMessageBodyFromMetadata(string messageId) + Task GetMessageBodyFromMetadata(string messageId, CancellationToken cancellationToken) { var message = processedMessages.FirstOrDefault(pm => (pm.MessageMetadata["MessageId"] as string) == messageId); @@ -171,7 +171,7 @@ Task GetMessageBodyFromMetadata(string messageId) return Task.FromResult(MessageBodyView.FromString(body, contentType, bodySize, messageId)); } - public Task>> QueryAuditCounts(string endpointName, CancellationToken cancellationToken) + public Task>> QueryAuditCounts(string endpointName, CancellationToken cancellationToken = default) { var results = messageViews .Where(m => m.ReceivingEndpoint.Name == endpointName && !m.IsSystemMessage) @@ -187,7 +187,7 @@ public Task>> QueryAuditCounts(string endpointName return Task.FromResult(new QueryResult>(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 metadata, string key) return null; } - public Task Setup() => Task.CompletedTask; + public Task Setup(CancellationToken cancellationToken = default) => Task.CompletedTask; List messageViews; List processedMessages; diff --git a/src/ServiceControl.Audit.Persistence.InMemory/InMemoryAuditIngestionUnitOfWork.cs b/src/ServiceControl.Audit.Persistence.InMemory/InMemoryAuditIngestionUnitOfWork.cs index fa4e167227..bc51d76815 100644 --- a/src/ServiceControl.Audit.Persistence.InMemory/InMemoryAuditIngestionUnitOfWork.cs +++ b/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 body, CancellationToken cancellationToken) + public async Task RecordProcessedMessage(ProcessedMessage processedMessage, ReadOnlyMemory 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); } } \ No newline at end of file diff --git a/src/ServiceControl.Audit.Persistence.InMemory/InMemoryAuditIngestionUnitOfWorkFactory.cs b/src/ServiceControl.Audit.Persistence.InMemory/InMemoryAuditIngestionUnitOfWorkFactory.cs index fb1f3a0c66..a83db9d32e 100644 --- a/src/ServiceControl.Audit.Persistence.InMemory/InMemoryAuditIngestionUnitOfWorkFactory.cs +++ b/src/ServiceControl.Audit.Persistence.InMemory/InMemoryAuditIngestionUnitOfWorkFactory.cs @@ -7,7 +7,7 @@ class InMemoryAuditIngestionUnitOfWorkFactory(InMemoryAuditDataStore dataStore, BodyStorageEnricher bodyStorageEnricher) : IAuditIngestionUnitOfWorkFactory { - public ValueTask StartNew(int batchSize, CancellationToken cancellationToken) + public ValueTask StartNew(int batchSize, CancellationToken cancellationToken = default) { //The batchSize argument is ignored: the in-memory storage implementation doesn't support batching. return new ValueTask(new InMemoryAuditIngestionUnitOfWork(dataStore, bodyStorageEnricher)); diff --git a/src/ServiceControl.Audit.Persistence.InMemory/InMemoryFailedAuditStorage.cs b/src/ServiceControl.Audit.Persistence.InMemory/InMemoryFailedAuditStorage.cs index 887e156511..652b79983e 100644 --- a/src/ServiceControl.Audit.Persistence.InMemory/InMemoryFailedAuditStorage.cs +++ b/src/ServiceControl.Audit.Persistence.InMemory/InMemoryFailedAuditStorage.cs @@ -7,10 +7,13 @@ class InMemoryFailedAuditStorage(InMemoryAuditDataStore dataStore) : IFailedAuditStorage { - public async Task ProcessFailedMessages(Func, CancellationToken, Task> onMessage, CancellationToken cancellationToken) + public async Task ProcessFailedMessages(Func, 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 GetFailedAuditsCount() => Task.FromResult(dataStore.failedAuditImports.Count); + public Task GetFailedAuditsCount(CancellationToken cancellationToken = default) + { + cancellationToken.ThrowIfCancellationRequested(); + return Task.FromResult(dataStore.failedAuditImports.Count); + } } } diff --git a/src/ServiceControl.Audit.Persistence.RavenDB/.editorconfig b/src/ServiceControl.Audit.Persistence.RavenDB/.editorconfig index e2d491d8f7..b9cef92e2f 100644 --- a/src/ServiceControl.Audit.Persistence.RavenDB/.editorconfig +++ b/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 diff --git a/src/ServiceControl.Audit.Persistence.RavenDB/DatabaseSetup.cs b/src/ServiceControl.Audit.Persistence.RavenDB/DatabaseSetup.cs index b1d42b3d00..701f7647bc 100644 --- a/src/ServiceControl.Audit.Persistence.RavenDB/DatabaseSetup.cs +++ b/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); diff --git a/src/ServiceControl.Audit.Persistence.RavenDB/LicenseStatusCheck.cs b/src/ServiceControl.Audit.Persistence.RavenDB/LicenseStatusCheck.cs index 342013df76..935e948875 100644 --- a/src/ServiceControl.Audit.Persistence.RavenDB/LicenseStatusCheck.cs +++ b/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"; diff --git a/src/ServiceControl.Audit.Persistence.RavenDB/RavenAttachmentsBodyStorage.cs b/src/ServiceControl.Audit.Persistence.RavenDB/RavenAttachmentsBodyStorage.cs index a8f801490e..441212614c 100644 --- a/src/ServiceControl.Audit.Persistence.RavenDB/RavenAttachmentsBodyStorage.cs +++ b/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 TryFetch(string bodyId, CancellationToken cancellationToken) + public async Task 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); diff --git a/src/ServiceControl.Audit.Persistence.RavenDB/RavenAuditDataStore.cs b/src/ServiceControl.Audit.Persistence.RavenDB/RavenAuditDataStore.cs index 97f90af67c..12a88109c6 100644 --- a/src/ServiceControl.Audit.Persistence.RavenDB/RavenAuditDataStore.cs +++ b/src/ServiceControl.Audit.Persistence.RavenDB/RavenAuditDataStore.cs @@ -17,7 +17,7 @@ class RavenAuditDataStore(IRavenSessionProvider sessionProvider, DatabaseConfiguration databaseConfiguration) : IAuditDataStore { - public async Task> QuerySagaHistoryById(Guid input, CancellationToken cancellationToken) + public async Task> QuerySagaHistoryById(Guid input, CancellationToken cancellationToken = default) { using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var sagaHistory = await @@ -28,7 +28,7 @@ public async Task> QuerySagaHistoryById(Guid input, Can return sagaHistory == null ? QueryResult.Empty() : new QueryResult(sagaHistory, new QueryStatsInfo($"{stats.ResultEtag}", stats.TotalResults)); } - public async Task>> GetMessages(bool includeSystemMessages, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange, CancellationToken cancellationToken) + public async Task>> GetMessages(bool includeSystemMessages, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange, CancellationToken cancellationToken = default) { using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var results = await session.Query(GetIndexName(isFullTextSearchEnabled)) @@ -43,7 +43,7 @@ public async Task>> GetMessages(bool includeSyst return new QueryResult>(results, stats.ToQueryStatsInfo()); } - public async Task>> QueryMessages(string searchParam, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange, CancellationToken cancellationToken) + public async Task>> QueryMessages(string searchParam, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange, CancellationToken cancellationToken = default) { using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var results = await session.Query(GetIndexName(isFullTextSearchEnabled)) @@ -58,7 +58,7 @@ public async Task>> QueryMessages(string searchP return new QueryResult>(results, stats.ToQueryStatsInfo()); } - public async Task>> QueryMessagesByReceivingEndpointAndKeyword(string endpoint, string keyword, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange, CancellationToken cancellationToken) + public async Task>> QueryMessagesByReceivingEndpointAndKeyword(string endpoint, string keyword, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange, CancellationToken cancellationToken = default) { using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var results = await session.Query(GetIndexName(isFullTextSearchEnabled)) @@ -74,7 +74,7 @@ public async Task>> QueryMessagesByReceivingEndp return new QueryResult>(results, stats.ToQueryStatsInfo()); } - public async Task>> QueryMessagesByReceivingEndpoint(bool includeSystemMessages, string endpointName, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange, CancellationToken cancellationToken) + public async Task>> QueryMessagesByReceivingEndpoint(bool includeSystemMessages, string endpointName, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange, CancellationToken cancellationToken = default) { using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var results = await session.Query(GetIndexName(isFullTextSearchEnabled)) @@ -90,7 +90,7 @@ public async Task>> QueryMessagesByReceivingEndp return new QueryResult>(results, stats.ToQueryStatsInfo()); } - public async Task>> QueryMessagesByConversationId(string conversationId, PagingInfo pagingInfo, SortInfo sortInfo, CancellationToken cancellationToken) + public async Task>> QueryMessagesByConversationId(string conversationId, PagingInfo pagingInfo, SortInfo sortInfo, CancellationToken cancellationToken = default) { using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var results = await session.Query(GetIndexName(isFullTextSearchEnabled)) @@ -104,7 +104,7 @@ public async Task>> QueryMessagesByConversationI return new QueryResult>(results, stats.ToQueryStatsInfo()); } - public async Task GetMessageBody(string messageId, CancellationToken cancellationToken) + public async Task GetMessageBody(string messageId, CancellationToken cancellationToken = default) { using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var result = await session.Advanced.Attachments.GetAsync(messageId, "body", cancellationToken); @@ -123,7 +123,7 @@ public async Task GetMessageBody(string messageId, Cancellation } - public async Task>> QueryAuditCounts(string endpointName, CancellationToken cancellationToken) + public async Task>> QueryAuditCounts(string endpointName, CancellationToken cancellationToken = default) { var indexName = GetIndexName(isFullTextSearchEnabled); diff --git a/src/ServiceControl.Audit.Persistence.RavenDB/RavenFailedAuditStorage.cs b/src/ServiceControl.Audit.Persistence.RavenDB/RavenFailedAuditStorage.cs index bf7e6fe946..d99e9b6fc5 100644 --- a/src/ServiceControl.Audit.Persistence.RavenDB/RavenFailedAuditStorage.cs +++ b/src/ServiceControl.Audit.Persistence.RavenDB/RavenFailedAuditStorage.cs @@ -12,16 +12,16 @@ class RavenFailedAuditStorage(IRavenSessionProvider sessionProvider) : IFailedAuditStorage { - public async Task SaveFailedAuditImport(FailedAuditImport message) + public async Task SaveFailedAuditImport(FailedAuditImport message, CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); - await session.StoreAsync(message); - await session.SaveChangesAsync(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); + await session.StoreAsync(message, token: cancellationToken); + await session.SaveChangesAsync(cancellationToken); } public async Task ProcessFailedMessages( Func, CancellationToken, Task> onMessage, - CancellationToken cancellationToken) + CancellationToken cancellationToken = default) { using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var query = session.Query(); @@ -55,11 +55,11 @@ await onMessage(transportMessage, (token) => } } - public async Task GetFailedAuditsCount() + public async Task GetFailedAuditsCount(CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); return await session.Query() - .CountAsync(); + .CountAsync(cancellationToken); } } } \ No newline at end of file diff --git a/src/ServiceControl.Audit.Persistence.RavenDB/RavenPersistenceLifecycleHostedService.cs b/src/ServiceControl.Audit.Persistence.RavenDB/RavenPersistenceLifecycleHostedService.cs index f266850953..41b3bd0ac4 100644 --- a/src/ServiceControl.Audit.Persistence.RavenDB/RavenPersistenceLifecycleHostedService.cs +++ b/src/ServiceControl.Audit.Persistence.RavenDB/RavenPersistenceLifecycleHostedService.cs @@ -6,8 +6,8 @@ namespace ServiceControl.Audit.Persistence.RavenDB sealed class RavenPersistenceLifecycleHostedService(IRavenPersistenceLifecycle lifecycle) : IHostedService { - public Task StartAsync(CancellationToken cancellationToken) => lifecycle.Initialize(cancellationToken); + public Task StartAsync(CancellationToken cancellationToken = default) => lifecycle.Initialize(cancellationToken); - public Task StopAsync(CancellationToken cancellationToken) => lifecycle.Stop(cancellationToken); + public Task StopAsync(CancellationToken cancellationToken = default) => lifecycle.Stop(cancellationToken); } } \ No newline at end of file diff --git a/src/ServiceControl.Audit.Persistence.RavenDB/UnitOfWork/RavenAuditIngestionUnitOfWork.cs b/src/ServiceControl.Audit.Persistence.RavenDB/UnitOfWork/RavenAuditIngestionUnitOfWork.cs index 1fa64e3c34..664dd7020b 100644 --- a/src/ServiceControl.Audit.Persistence.RavenDB/UnitOfWork/RavenAuditIngestionUnitOfWork.cs +++ b/src/ServiceControl.Audit.Persistence.RavenDB/UnitOfWork/RavenAuditIngestionUnitOfWork.cs @@ -21,7 +21,7 @@ class RavenAuditIngestionUnitOfWork( IBodyStorage bodyStorage) : IAuditIngestionUnitOfWork { - public async Task RecordProcessedMessage(ProcessedMessage processedMessage, ReadOnlyMemory body, CancellationToken cancellationToken) + public async Task RecordProcessedMessage(ProcessedMessage processedMessage, ReadOnlyMemory body, CancellationToken cancellationToken = default) { processedMessage.MessageMetadata["ContentLength"] = body.Length; if (!body.IsEmpty) @@ -46,7 +46,7 @@ MetadataAsDictionary GetExpirationMetadata() => [Constants.Documents.Metadata.Expires] = DateTime.UtcNow.Add(auditRetentionPeriod) }; - public Task RecordSagaSnapshot(SagaSnapshot sagaSnapshot, CancellationToken cancellationToken) + public Task RecordSagaSnapshot(SagaSnapshot sagaSnapshot, CancellationToken cancellationToken = default) => bulkInsert.StoreAsync(sagaSnapshot, GetExpirationMetadata()); public async ValueTask DisposeAsync() diff --git a/src/ServiceControl.Audit.Persistence.RavenDB/UnitOfWork/RavenAuditUnitOfWorkFactory.cs b/src/ServiceControl.Audit.Persistence.RavenDB/UnitOfWork/RavenAuditUnitOfWorkFactory.cs index eddfba1125..0f256ab17b 100644 --- a/src/ServiceControl.Audit.Persistence.RavenDB/UnitOfWork/RavenAuditUnitOfWorkFactory.cs +++ b/src/ServiceControl.Audit.Persistence.RavenDB/UnitOfWork/RavenAuditUnitOfWorkFactory.cs @@ -13,7 +13,7 @@ class RavenAuditIngestionUnitOfWorkFactory( MinimumRequiredStorageState customCheckState) : IAuditIngestionUnitOfWorkFactory { - public async ValueTask StartNew(int batchSize, CancellationToken cancellationToken) + public async ValueTask StartNew(int batchSize, CancellationToken cancellationToken = default) { // DO NOT USE using var, will be disposed by RavenAuditIngestionUnitOfWork var lifetimeForwardedTimedCancellationSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); diff --git a/src/ServiceControl.Audit.Persistence.Tests.RavenDB/.editorconfig b/src/ServiceControl.Audit.Persistence.Tests.RavenDB/.editorconfig index d786f604c0..174320291d 100644 --- a/src/ServiceControl.Audit.Persistence.Tests.RavenDB/.editorconfig +++ b/src/ServiceControl.Audit.Persistence.Tests.RavenDB/.editorconfig @@ -6,5 +6,11 @@ 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. +# The remaining sites are private test helpers that already close over the fixture's +# TestTimeoutCancellationToken, so they need a parameter added rather than a token found. dotnet_diagnostic.PS0018.severity = none + +# Justification: the polling loop swallows OperationCanceledException on purpose, because querying an +# index at the moment it updates throws one. That is not cancellation, so there is no token to filter on. +[IndexSetupTests.cs] dotnet_diagnostic.PS0020.severity = none diff --git a/src/ServiceControl.Audit.Persistence/.editorconfig b/src/ServiceControl.Audit.Persistence/.editorconfig index 522a310bff..ff993b49bb 100644 --- a/src/ServiceControl.Audit.Persistence/.editorconfig +++ b/src/ServiceControl.Audit.Persistence/.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 diff --git a/src/ServiceControl.Audit.Persistence/BodyStorageEnricher.cs b/src/ServiceControl.Audit.Persistence/BodyStorageEnricher.cs index be9e133bf8..04e472ead3 100644 --- a/src/ServiceControl.Audit.Persistence/BodyStorageEnricher.cs +++ b/src/ServiceControl.Audit.Persistence/BodyStorageEnricher.cs @@ -12,7 +12,7 @@ public class BodyStorageEnricher(IBodyStorage bodyStorage, PersistenceSettings settings, ILogger logger) { - public async ValueTask StoreAuditMessageBody(ReadOnlyMemory body, ProcessedMessage processedMessage, CancellationToken cancellationToken) + public async ValueTask StoreAuditMessageBody(ReadOnlyMemory body, ProcessedMessage processedMessage, CancellationToken cancellationToken = default) { var bodySize = body.Length; processedMessage.MessageMetadata.Add("ContentLength", bodySize); diff --git a/src/ServiceControl.Audit.Persistence/IAuditDataStore.cs b/src/ServiceControl.Audit.Persistence/IAuditDataStore.cs index 5ed7a2fc6e..eb252e321d 100644 --- a/src/ServiceControl.Audit.Persistence/IAuditDataStore.cs +++ b/src/ServiceControl.Audit.Persistence/IAuditDataStore.cs @@ -12,13 +12,13 @@ public interface IAuditDataStore { - Task> QuerySagaHistoryById(Guid input, CancellationToken cancellationToken); + Task> QuerySagaHistoryById(Guid input, CancellationToken cancellationToken = default); Task>> GetMessages(bool includeSystemMessages, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange = null, CancellationToken cancellationToken = default); Task>> QueryMessages(string searchParam, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange = null, CancellationToken cancellationToken = default); Task>> QueryMessagesByReceivingEndpointAndKeyword(string endpoint, string keyword, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange = null, CancellationToken cancellationToken = default); Task>> QueryMessagesByReceivingEndpoint(bool includeSystemMessages, string endpointName, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange = null, CancellationToken cancellationToken = default); - Task>> QueryMessagesByConversationId(string conversationId, PagingInfo pagingInfo, SortInfo sortInfo, CancellationToken cancellationToken); - Task GetMessageBody(string messageId, CancellationToken cancellationToken); - Task>> QueryAuditCounts(string endpointName, CancellationToken cancellationToken); + Task>> QueryMessagesByConversationId(string conversationId, PagingInfo pagingInfo, SortInfo sortInfo, CancellationToken cancellationToken = default); + Task GetMessageBody(string messageId, CancellationToken cancellationToken = default); + Task>> QueryAuditCounts(string endpointName, CancellationToken cancellationToken = default); } } \ No newline at end of file diff --git a/src/ServiceControl.Audit.Persistence/IBodyStorage.cs b/src/ServiceControl.Audit.Persistence/IBodyStorage.cs index ae95cb1d3f..98b6c42e84 100644 --- a/src/ServiceControl.Audit.Persistence/IBodyStorage.cs +++ b/src/ServiceControl.Audit.Persistence/IBodyStorage.cs @@ -6,8 +6,8 @@ public interface IBodyStorage { - Task Store(string bodyId, string contentType, int bodySize, Stream bodyStream, CancellationToken cancellationToken); - Task TryFetch(string bodyId, CancellationToken cancellationToken); + Task Store(string bodyId, string contentType, int bodySize, Stream bodyStream, CancellationToken cancellationToken = default); + Task TryFetch(string bodyId, CancellationToken cancellationToken = default); } public class StreamResult diff --git a/src/ServiceControl.Audit.Persistence/IFailedAuditStorage.cs b/src/ServiceControl.Audit.Persistence/IFailedAuditStorage.cs index 86ef3aedcd..29404b6fa7 100644 --- a/src/ServiceControl.Audit.Persistence/IFailedAuditStorage.cs +++ b/src/ServiceControl.Audit.Persistence/IFailedAuditStorage.cs @@ -7,13 +7,13 @@ public interface IFailedAuditStorage { - Task SaveFailedAuditImport(FailedAuditImport message); + Task SaveFailedAuditImport(FailedAuditImport message, CancellationToken cancellationToken = default); Task ProcessFailedMessages( Func, CancellationToken, Task> onMessage, - CancellationToken cancellationToken + CancellationToken cancellationToken = default ); - Task GetFailedAuditsCount(); + Task GetFailedAuditsCount(CancellationToken cancellationToken = default); } } \ No newline at end of file diff --git a/src/ServiceControl.Audit.Persistence/UnitOfWork/IAuditIngestionUnitOfWorkFactory.cs b/src/ServiceControl.Audit.Persistence/UnitOfWork/IAuditIngestionUnitOfWorkFactory.cs index 71dceb269b..1d3470bdea 100644 --- a/src/ServiceControl.Audit.Persistence/UnitOfWork/IAuditIngestionUnitOfWorkFactory.cs +++ b/src/ServiceControl.Audit.Persistence/UnitOfWork/IAuditIngestionUnitOfWorkFactory.cs @@ -5,7 +5,7 @@ public interface IAuditIngestionUnitOfWorkFactory { - ValueTask StartNew(int batchSize, CancellationToken cancellationToken); //Throws if not enough space or some other problem preventing from writing data + ValueTask StartNew(int batchSize, CancellationToken cancellationToken = default); //Throws if not enough space or some other problem preventing from writing data bool CanIngestMore(); } } \ No newline at end of file diff --git a/src/ServiceControl.Audit/.editorconfig b/src/ServiceControl.Audit/.editorconfig index d5e4367c81..46637f1f91 100644 --- a/src/ServiceControl.Audit/.editorconfig +++ b/src/ServiceControl.Audit/.editorconfig @@ -9,8 +9,3 @@ dotnet_diagnostic.CA2007.severity = none [{Auditing/AuditIngestion.cs,Auditing/AuditIngestionFaultPolicy.cs,Auditing/ImportFailureCircuitBreaker.cs}] dotnet_diagnostic.PS0013.severity = none dotnet_diagnostic.PS0018.severity = none - -# Blocked on IFailedAuditStorage.ProcessFailedMessages, whose callback is handed its own token by the -# store. The catch filter here correctly tests that one, which PS0020 cannot see. -[Auditing/ImportFailedAudits.cs] -dotnet_diagnostic.PS0020.severity = none diff --git a/src/ServiceControl.Audit/Auditing/AuditIngestionFaultPolicy.cs b/src/ServiceControl.Audit/Auditing/AuditIngestionFaultPolicy.cs index b4952bff6c..fab0a13e1b 100644 --- a/src/ServiceControl.Audit/Auditing/AuditIngestionFaultPolicy.cs +++ b/src/ServiceControl.Audit/Auditing/AuditIngestionFaultPolicy.cs @@ -84,7 +84,7 @@ async Task DoLogging(Exception exception, FailedAuditImport failure, Cancellatio logger.LogError(exception, "Failed importing error message"); // Write to storage - await failedAuditStorage.SaveFailedAuditImport(failure); + await failedAuditStorage.SaveFailedAuditImport(failure, cancellationToken); if (!AppEnvironment.RunningInContainer) { diff --git a/src/ServiceControl.Audit/Auditing/FailedAuditImportCustomCheck.cs b/src/ServiceControl.Audit/Auditing/FailedAuditImportCustomCheck.cs index 6790858fa5..8d7afcfe84 100644 --- a/src/ServiceControl.Audit/Auditing/FailedAuditImportCustomCheck.cs +++ b/src/ServiceControl.Audit/Auditing/FailedAuditImportCustomCheck.cs @@ -11,7 +11,7 @@ class FailedAuditImportCustomCheck(IFailedAuditStorage store, ILogger PerformCheck(CancellationToken cancellationToken = default) { - var count = await store.GetFailedAuditsCount(); + var count = await store.GetFailedAuditsCount(cancellationToken); if (count > 0) { logger.LogWarning(message); diff --git a/src/ServiceControl.Audit/Auditing/ImportFailedAudits.cs b/src/ServiceControl.Audit/Auditing/ImportFailedAudits.cs index 5cb2ace0c2..c8aaf835ca 100644 --- a/src/ServiceControl.Audit/Auditing/ImportFailedAudits.cs +++ b/src/ServiceControl.Audit/Auditing/ImportFailedAudits.cs @@ -30,8 +30,6 @@ public async Task Run(CancellationToken cancellationToken = default) var succeeded = 0; var failed = 0; - // The store hands each callback its own token, distinct from this method's, so the catch - // filters below deliberately test that one rather than the outer cancellationToken. #pragma warning disable PS0021 await failedAuditStore.ProcessFailedMessages( async (transportMessage, markComplete, token) => @@ -57,9 +55,9 @@ await failedAuditStore.ProcessFailedMessages( succeeded++; logger.LogDebug("Successfully re-imported failed audit message {MessageId}", transportMessage.Id); } - catch (OperationCanceledException e) when (token.IsCancellationRequested) + catch (OperationCanceledException e) when (cancellationToken.IsCancellationRequested) { - logger.LogInformation(e, "Cancelled"); + logger.LogInformation(e, "Cancelled re-importing failed audit message {MessageId}", transportMessage.Id); } catch (Exception e) {