Continue consistent application of optional cancellation tokens

src/ServiceControl.AcceptanceTests/Recoverability/ExternalIntegration/When_encountered_an_error.cs

@@ -3,6 +3,7 @@ namespace ServiceControl.AcceptanceTests.Recoverability.ExternalIntegration
     using System;
     using System.Collections.Generic;
     using System.Linq;
+    using System.Threading;
     using System.Threading.Tasks;
     using AcceptanceTesting;
     using AcceptanceTesting.EndpointTemplates;
@@ -79,7 +80,7 @@ class FaultyPublisher(MyContext context) : IEventPublisher
 
             public object CreateDispatchContext(IDomainEvent @event) => null;
 
-            public Task<IEnumerable<object>> PublishEventsForOwnContexts(IEnumerable<object> allContexts)
+            public Task<IEnumerable<object>> PublishEventsForOwnContexts(IEnumerable<object> allContexts, CancellationToken cancellationToken = default)
             {
                 if (!failed)
                 {

src/ServiceControl.Audit/.editorconfig

@@ -3,13 +3,14 @@
 # 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
+# Blocked on the transport abstraction. TransportCustomization.CreateTransportInfrastructure takes
+# Func<string, Exception, Task> for onCriticalError, so the critical-error path here cannot carry a
+# token until that signature gains one in the transports phase.
+[{Auditing/AuditIngestion.cs,Auditing/AuditIngestionFaultPolicy.cs,Auditing/ImportFailureCircuitBreaker.cs}]
 dotnet_diagnostic.PS0013.severity = none
-dotnet_diagnostic.PS0017.severity = none
 dotnet_diagnostic.PS0018.severity = none
-dotnet_diagnostic.PS0019.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
-dotnet_diagnostic.PS0021.severity = none

src/ServiceControl.Audit/Auditing/AuditIngestion.cs

@@ -93,12 +93,20 @@ async Task EnsureStarted(CancellationToken cancellationToken)
                     await StopAndTeardownInfrastructure(cancellationToken);
                 }
             }
+            catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+            {
+                throw;
+            }
             catch (Exception e)
             {
                 try
                 {
                     await StopAndTeardownInfrastructure(cancellationToken);
                 }
+                catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+                {
+                    throw;
+                }
                 catch (Exception teardownException)
                 {
                     throw new AggregateException(e, teardownException);
@@ -139,6 +147,10 @@ async Task SetUpAndStartInfrastructure(CancellationToken cancellationToken)
 
                 logger.LogInformation(LogMessages.StartedInfrastructure);
             }
+            catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+            {
+                throw;
+            }
             catch (Exception e)
             {
                 logger.LogError(e, "Failed to start infrastructure");
@@ -174,6 +186,10 @@ async Task StopAndTeardownInfrastructure(CancellationToken cancellationToken)
 
                 logger.LogInformation(LogMessages.StoppedInfrastructure);
             }
+            catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+            {
+                throw;
+            }
             catch (Exception e)
             {
                 logger.LogError(e, "Failed to stop infrastructure");
@@ -216,19 +232,19 @@ async Task OnMessage(MessageContext messageContext, CancellationToken cancellati
             messageIngestionMetrics.Success();
         }
 
-        public override async Task StartAsync(CancellationToken cancellationToken)
+        public override async Task StartAsync(CancellationToken cancellationToken = default)
         {
             await watchdog.Start(() => applicationLifetime.StopApplication(), cancellationToken);
             await base.StartAsync(cancellationToken);
         }
 
-        protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+        protected override async Task ExecuteAsync(CancellationToken cancellationToken = default)
         {
             try
             {
                 var contexts = new List<MessageContext>(MaxBatchSize);
 
-                while (await channel.Reader.WaitToReadAsync(stoppingToken))
+                while (await channel.Reader.WaitToReadAsync(cancellationToken))
                 {
                     // will only enter here if there is something to read.
                     try
@@ -241,22 +257,27 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
                             contexts.Add(context);
                         }
 
-                        await auditIngestor.Ingest(contexts, stoppingToken);
+                        await auditIngestor.Ingest(contexts, cancellationToken);
 
                         batchMetrics.Complete(contexts.Count);
                     }
-                    catch (Exception e)
+                    catch (OperationCanceledException e) when (cancellationToken.IsCancellationRequested)
                     {
                         // signal all message handling tasks to terminate
                         foreach (var context in contexts)
                         {
                             _ = context.GetTaskCompletionSource().TrySetException(e);
                         }
 
-                        if (e is OperationCanceledException && stoppingToken.IsCancellationRequested)
+                        logger.LogInformation(e, "Batch cancelled");
+                        break;
+                    }
+                    catch (Exception e)
+                    {
+                        // signal all message handling tasks to terminate
+                        foreach (var context in contexts)
                         {
-                            logger.LogInformation(e, "Batch cancelled");
-                            break;
+                            _ = context.GetTaskCompletionSource().TrySetException(e);
                         }
 
                         logger.LogInformation(e, "Ingesting messages failed");
@@ -268,13 +289,13 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
                 }
                 // will fall out here when writer is completed
             }
-            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
+            catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
             {
                 // ExecuteAsync cancelled
             }
         }
 
-        public override async Task StopAsync(CancellationToken cancellationToken)
+        public override async Task StopAsync(CancellationToken cancellationToken = default)
         {
             try
             {

src/ServiceControl.Audit/Auditing/AuditIngestor.cs

@@ -45,7 +45,7 @@ ILogger<AuditIngestor> logger
             );
         }
 
-        public async Task Ingest(List<MessageContext> contexts, CancellationToken cancellationToken)
+        public async Task Ingest(List<MessageContext> contexts, CancellationToken cancellationToken = default)
         {
             var stored = await auditPersister.Persist(contexts, cancellationToken);
 
@@ -61,6 +61,10 @@ public async Task Ingest(List<MessageContext> contexts, CancellationToken cancel
                     context.GetTaskCompletionSource().TrySetResult(true);
                 }
             }
+            catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+            {
+                throw;
+            }
             catch (Exception e)
             {
                 logger.LogWarning(e, "Forwarding messages failed");
@@ -101,7 +105,7 @@ Task Forward(IReadOnlyCollection<MessageContext> messageContexts, string forward
                 : Task.CompletedTask;
         }
 
-        public async Task VerifyCanReachForwardingAddress(CancellationToken cancellationToken)
+        public async Task VerifyCanReachForwardingAddress(CancellationToken cancellationToken = default)
         {
             if (!settings.ForwardAuditMessages)
             {
@@ -120,6 +124,10 @@ public async Task VerifyCanReachForwardingAddress(CancellationToken cancellation
 
                 await messageDispatcher.Value.Dispatch(transportOperations, new TransportTransaction(), cancellationToken);
             }
+            catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+            {
+                throw;
+            }
             catch (Exception e)
             {
                 throw new Exception($"Unable to write to forwarding queue {settings.AuditLogQueue}", e);

src/ServiceControl.Audit/Auditing/AuditPersister.cs

@@ -22,7 +22,7 @@ class AuditPersister(IAuditIngestionUnitOfWorkFactory unitOfWorkFactory,
         Lazy<IMessageDispatcher> messageDispatcher,
         ILogger logger)
     {
-        public async Task<IReadOnlyList<MessageContext>> Persist(IReadOnlyList<MessageContext> contexts, CancellationToken cancellationToken)
+        public async Task<IReadOnlyList<MessageContext>> Persist(IReadOnlyList<MessageContext> contexts, CancellationToken cancellationToken = default)
         {
             var storedContexts = new List<MessageContext>(contexts.Count);
             IAuditIngestionUnitOfWork unitOfWork = null;
@@ -33,7 +33,7 @@ public async Task<IReadOnlyList<MessageContext>> Persist(IReadOnlyList<MessageCo
                 var inserts = new List<Task>(contexts.Count);
                 foreach (var context in contexts)
                 {
-                    inserts.Add(ProcessMessage(context));
+                    inserts.Add(ProcessMessage(context, cancellationToken));
                 }
 
                 await Task.WhenAll(inserts);
@@ -58,6 +58,10 @@ public async Task<IReadOnlyList<MessageContext>> Persist(IReadOnlyList<MessageCo
                     storedContexts.Add(context);
                 }
             }
+            catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+            {
+                throw;
+            }
             catch (Exception e)
             {
                 logger.LogWarning(e, "Bulk insertion failed");
@@ -89,7 +93,7 @@ public async Task<IReadOnlyList<MessageContext>> Persist(IReadOnlyList<MessageCo
             return storedContexts;
         }
 
-        async Task ProcessMessage(MessageContext context)
+        async Task ProcessMessage(MessageContext context, CancellationToken cancellationToken)
         {
             if (context.Headers.TryGetValue(Headers.EnclosedMessageTypes, out var messageType)
                 && messageType == typeof(SagaUpdatedMessage).FullName)
@@ -98,7 +102,7 @@ async Task ProcessMessage(MessageContext context)
             }
             else
             {
-                await ProcessAuditMessage(context);
+                await ProcessAuditMessage(context, cancellationToken);
             }
         }
 
@@ -123,7 +127,7 @@ void ProcessSagaAuditMessage(MessageContext context)
             }
         }
 
-        async Task ProcessAuditMessage(MessageContext context)
+        async Task ProcessAuditMessage(MessageContext context, CancellationToken cancellationToken)
         {
             if (!context.Headers.TryGetValue(Headers.MessageId, out var messageId))
             {
@@ -153,11 +157,11 @@ async Task ProcessAuditMessage(MessageContext context)
 
                 foreach (var commandToEmit in commandsToEmit)
                 {
-                    await messageSession.Send(commandToEmit);
+                    await messageSession.Send(commandToEmit, cancellationToken);
                 }
 
                 await messageDispatcher.Value.Dispatch(new TransportOperations(messagesToEmit.ToArray()),
-                    new TransportTransaction()); //Do not hook into the incoming transaction
+                    new TransportTransaction(), cancellationToken); //Do not hook into the incoming transaction
 
                 logger.LogDebug("{CommandsToEmitCount} commands and {MessagesToEmitCount} control messages emitted", commandsToEmit.Count, messagesToEmit.Count);
 
@@ -174,6 +178,10 @@ await messageDispatcher.Value.Dispatch(new TransportOperations(messagesToEmit.To
                 context.Extensions.Set("AuditType", "ProcessedMessage");
                 context.Extensions.Set(auditMessage);
             }
+            catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+            {
+                throw;
+            }
             catch (Exception e)
             {
                 logger.LogWarning(e, "Processing of message '{MessageId}' failed", messageId);

src/ServiceControl.Audit/Auditing/ImportFailedAudits.cs

@@ -30,6 +30,9 @@ 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) =>
                     {
@@ -65,6 +68,7 @@ await failedAuditStore.ProcessFailedMessages(
                         }
 
                     }, cancellationToken);
+#pragma warning restore PS0021
 
             logger.LogInformation("Done re-importing failed audits. Successfully re-imported {SuccessCount} messages. Failed re-importing {FailureCount} messages", succeeded, failed);
 

src/ServiceControl.Audit/Auditing/MessagesView/GetMessages2Controller.cs

@@ -24,7 +24,7 @@ public async Task<IList<MessagesView>> GetAllMessages(
         [FromQuery(Name = "from")] string from,
         [FromQuery(Name = "to")] string to,
         string q,
-        CancellationToken cancellationToken)
+        CancellationToken cancellationToken = default)
     {
         QueryResult<IList<MessagesView>> result;
         var pagingInfo = new PagingInfo(pageSize: pageSize);