Propagate cancellation tokens through transport abstractions and implementations

src/Particular.LicensingComponent/MonitoringThroughput/MonitoringThroughputHostedService.cs

@@ -26,7 +26,7 @@ public async Task StartAsync(CancellationToken cancellationToken)
     {
         logger.LogInformation("Starting {ServiceName}", nameof(MonitoringThroughputHostedService));
 
-        transportInfrastructure = await transportCustomization.CreateTransportInfrastructure(ServiceControlSettings.ServiceControlThroughputDataQueue, transportSettings, Handle, (_, __) => Task.FromResult(ErrorHandleResult.Handled), (_, __) => Task.CompletedTask);
+        transportInfrastructure = await transportCustomization.CreateTransportInfrastructure(ServiceControlSettings.ServiceControlThroughputDataQueue, transportSettings, Handle, (_, __) => Task.FromResult(ErrorHandleResult.Handled), (_, __, ___) => Task.CompletedTask, cancellationToken: cancellationToken);
         await transportInfrastructure.Receivers[ServiceControlSettings.ServiceControlThroughputDataQueue].StartReceive(cancellationToken);
     }
 

src/ServiceControl.Audit.UnitTests/.editorconfig

@@ -7,5 +7,4 @@ dotnet_diagnostic.CA2007.severity = none
 # 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.PS0013.severity = none
 dotnet_diagnostic.PS0018.severity = none

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

@@ -4,6 +4,7 @@
     using System.Collections.Generic;
     using System.Reflection;
     using System.Runtime.Loader;
+    using System.Threading;
     using System.Threading.Tasks;
     using Audit.Infrastructure.Hosting;
     using Audit.Infrastructure.Hosting.Commands;
@@ -73,7 +74,7 @@ public void CustomizeMonitoringEndpoint(EndpointConfiguration endpointConfigurat
 
         public void AddTransportForMonitoring(IServiceCollection services, TransportSettings transportSettings) => throw new NotImplementedException();
 
-        public Task ProvisionQueues(TransportSettings transportSettings, IEnumerable<string> additionalQueues)
+        public Task ProvisionQueues(TransportSettings transportSettings, IEnumerable<string> additionalQueues, CancellationToken cancellationToken = default)
         {
             QueuesCreated =
             [
@@ -85,8 +86,9 @@ public Task ProvisionQueues(TransportSettings transportSettings, IEnumerable<str
         }
 
         public Task<TransportInfrastructure> CreateTransportInfrastructure(string name, TransportSettings transportSettings,
-            OnMessage onMessage = null, OnError onError = null, Func<string, Exception, Task> onCriticalError = null,
-            TransportTransactionMode preferredTransactionMode = TransportTransactionMode.ReceiveOnly) =>
+            OnMessage onMessage = null, OnError onError = null, Func<string, Exception, CancellationToken, Task> onCriticalError = null,
+            TransportTransactionMode preferredTransactionMode = TransportTransactionMode.ReceiveOnly,
+            CancellationToken cancellationToken = default) =>
             throw new NotImplementedException();
         public string ToTransportQualifiedQueueName(string queueName) => queueName;
     }

src/ServiceControl.Audit/.editorconfig

@@ -2,10 +2,3 @@
 
 # Justification: ServiceControl app has no synchronization context
 dotnet_diagnostic.CA2007.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.PS0018.severity = none

src/ServiceControl.Audit/Auditing/AuditIngestion.cs

@@ -68,10 +68,10 @@ ILogger<AuditIngestion> logger
             );
         }
 
-        Task OnCriticalError(string failure, Exception exception)
+        Task OnCriticalError(string failure, Exception exception, CancellationToken cancellationToken)
         {
             logger.LogCritical(exception, "OnCriticalError. '{Failure}'", failure);
-            return watchdog.OnFailure(failure);
+            return watchdog.OnFailure(failure, cancellationToken);
         }
 
         async Task EnsureStarted(CancellationToken cancellationToken)
@@ -137,7 +137,8 @@ async Task SetUpAndStartInfrastructure(CancellationToken cancellationToken)
                     OnMessage,
                     errorHandlingPolicy.OnError,
                     OnCriticalError,
-                    TransportTransactionMode.ReceiveOnly
+                    TransportTransactionMode.ReceiveOnly,
+                    cancellationToken
                 );
 
                 messageReceiver = transportInfrastructure.Receivers[inputEndpoint];

src/ServiceControl.Audit/Auditing/AuditIngestionFaultPolicy.cs

@@ -20,7 +20,7 @@ class AuditIngestionFaultPolicy
     public AuditIngestionFaultPolicy(
         IFailedAuditStorage failedAuditStorage,
         LoggingSettings settings,
-        Func<string, Exception, Task> onCriticalError,
+        Func<string, Exception, CancellationToken, Task> onCriticalError,
         IngestionMetrics metrics,
         ILogger logger)
     {

src/ServiceControl.Audit/Auditing/ImportFailureCircuitBreaker.cs

@@ -7,7 +7,7 @@
     class ImportFailureCircuitBreaker : IDisposable
     {
 
-        public ImportFailureCircuitBreaker(Func<string, Exception, Task> onCriticalError)
+        public ImportFailureCircuitBreaker(Func<string, Exception, CancellationToken, Task> onCriticalError)
         {
             this.onCriticalError = onCriticalError;
             timer = new Timer(_ => FlushHistory(), null, TimeSpan.FromSeconds(0), TimeSpan.FromSeconds(20));
@@ -28,11 +28,13 @@ public void Increment(Exception lastException)
             var result = Interlocked.Increment(ref failureCount);
             if (result > 50)
             {
-                _ = Task.Run(() => onCriticalError("Failed to import too many times", lastException));
+                // Not cancellable: the notification exists to trigger shutdown, so the token
+                // that shutdown cancels must not be able to suppress it.
+                _ = Task.Run(() => onCriticalError("Failed to import too many times", lastException, CancellationToken.None), CancellationToken.None);
             }
         }
 
-        Func<string, Exception, Task> onCriticalError;
+        Func<string, Exception, CancellationToken, Task> onCriticalError;
         Timer timer;
         long failureCount;
     }

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

@@ -33,7 +33,7 @@ public override async Task Execute(HostArguments args, Settings settings, Cancel
                     transportSettings.RunCustomChecks = false;
                     var transportCustomization = TransportFactory.Create(transportSettings);
 
-                    await transportCustomization.ProvisionQueues(transportSettings, additionalQueues);
+                    await transportCustomization.ProvisionQueues(transportSettings, additionalQueues, cancellationToken);
                 }
             }
 

src/ServiceControl.Persistence.Tests/RetryStateTests.cs

@@ -500,13 +500,14 @@ public class TestTransportCustomization : ITransportCustomization
 
             public Task<TransportInfrastructure> CreateTransportInfrastructure(string name,
                 TransportSettings transportSettings, OnMessage onMessage = null, OnError onError = null,
-                Func<string, Exception, Task> onCriticalError = null,
+                Func<string, Exception, CancellationToken, Task> onCriticalError = null,
                 NServiceBus.TransportTransactionMode preferredTransactionMode =
-                    NServiceBus.TransportTransactionMode.ReceiveOnly) => Task.FromResult(TransportInfrastructure);
+                    NServiceBus.TransportTransactionMode.ReceiveOnly,
+                CancellationToken cancellationToken = default) => Task.FromResult(TransportInfrastructure);
             public void CustomizeAuditEndpoint(NServiceBus.EndpointConfiguration endpointConfiguration, TransportSettings transportSettings) => throw new NotImplementedException();
             public void CustomizeMonitoringEndpoint(NServiceBus.EndpointConfiguration endpointConfiguration, TransportSettings transportSettings) => throw new NotImplementedException();
             public void CustomizePrimaryEndpoint(NServiceBus.EndpointConfiguration endpointConfiguration, TransportSettings transportSettings) => throw new NotImplementedException();
-            public Task ProvisionQueues(TransportSettings transportSettings, IEnumerable<string> additionalQueues) => throw new NotImplementedException();
+            public Task ProvisionQueues(TransportSettings transportSettings, IEnumerable<string> additionalQueues, CancellationToken cancellationToken = default) => throw new NotImplementedException();
             public string ToTransportQualifiedQueueName(string queueName) => queueName;
         }
 

src/ServiceControl.Transports.ASBS.Tests/.editorconfig

@@ -2,8 +2,3 @@
 
 # Justification: Test project
 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.PS0018.severity = none

src/ServiceControl.Transports.ASBS.Tests/TransportTestsConfiguration.cs

@@ -1,6 +1,7 @@
 namespace ServiceControl.Transport.Tests
 {
     using System;
+    using System.Threading;
     using System.Threading.Tasks;
     using Transports;
     using Transports.ASBS;
@@ -11,7 +12,7 @@ partial class TransportTestsConfiguration
 
         public ITransportCustomization TransportCustomization { get; private set; }
 
-        public Task Configure()
+        public Task Configure(CancellationToken cancellationToken = default)
         {
             TransportCustomization = new ASBSTransportCustomization();
             ConnectionString = Environment.GetEnvironmentVariable(ConnectionStringKey);
@@ -24,7 +25,7 @@ public Task Configure()
             return Task.CompletedTask;
         }
 
-        public Task Cleanup() => Task.CompletedTask;
+        public Task Cleanup(CancellationToken cancellationToken = default) => Task.CompletedTask;
 
         static string ConnectionStringKey = "ServiceControl_TransportTests_ASBS_ConnectionString";
     }

src/ServiceControl.Transports.ASBS/.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.PS0004.severity = none
-dotnet_diagnostic.PS0017.severity = none

src/ServiceControl.Transports.ASBS/ASBSTransportCustomization.cs

@@ -3,6 +3,7 @@
     using System.Collections.Generic;
     using System.Linq;
     using System.Text.Json;
+    using System.Threading;
     using System.Threading.Tasks;
     using Azure.Messaging.ServiceBus;
     using Azure.Messaging.ServiceBus.Administration;
@@ -99,9 +100,9 @@ protected override void AddTransportForMonitoringCore(IServiceCollection service
             services.AddHostedService(provider => provider.GetRequiredService<IProvideQueueLength>());
         }
 
-        public override async Task ProvisionQueues(TransportSettings transportSettings, IEnumerable<string> additionalQueues)
+        public override async Task ProvisionQueues(TransportSettings transportSettings, IEnumerable<string> additionalQueues, CancellationToken cancellationToken = default)
         {
-            await base.ProvisionQueues(transportSettings, additionalQueues);
+            await base.ProvisionQueues(transportSettings, additionalQueues, cancellationToken);
 
             if (transportSettings.EventTypesPublished.Count == 0)
             {
@@ -132,7 +133,7 @@ async Task CreateTopic(string publishedTopic)
 
                 try
                 {
-                    await managementClient.CreateTopicAsync(topicToPublishTo).ConfigureAwait(false);
+                    await managementClient.CreateTopicAsync(topicToPublishTo, cancellationToken).ConfigureAwait(false);
                 }
                 catch (ServiceBusException sbe) when (sbe.Reason == ServiceBusFailureReason.MessagingEntityAlreadyExists || sbe.IsTransient)
                 {

src/ServiceControl.Transports.ASBS/AzureQuery.cs

@@ -201,7 +201,7 @@ public string ExtractServiceBusName()
 
     public override async IAsyncEnumerable<QueueThroughput> GetThroughputPerDay(IBrokerQueue brokerQueue,
         DateOnly startDate,
-        [EnumeratorCancellation] CancellationToken cancellationToken)
+        [EnumeratorCancellation] CancellationToken cancellationToken = default)
     {
         logger.LogInformation($"Gathering metrics for \"{brokerQueue.QueueName}\" queue");
 
@@ -249,7 +249,7 @@ public override async IAsyncEnumerable<QueueThroughput> GetThroughputPerDay(IBro
     }
 
     async Task<IReadOnlyList<MonitorMetricValue>> GetMetrics(string queueName, DateOnly startTime, DateOnly endTime,
-        CancellationToken cancellationToken = default)
+        CancellationToken cancellationToken)
     {
         var options = new ArmResourceGetMonitorMetricsOptions()
         {
@@ -272,7 +272,7 @@ async Task<IReadOnlyList<MonitorMetricValue>> GetMetrics(string queueName, DateO
     }
 
     public override async IAsyncEnumerable<IBrokerQueue> GetQueueNames(
-        [EnumeratorCancellation] CancellationToken cancellationToken)
+        [EnumeratorCancellation] CancellationToken cancellationToken = default)
     {
         var validNamespaces = await GetValidNamespaceNames(cancellationToken);
 
@@ -310,7 +310,7 @@ public override async IAsyncEnumerable<IBrokerQueue> GetQueueNames(
         { ArmEnvironment.AzureChina, "servicebus.chinacloudapi.cn" },
     };
 
-    async Task<HashSet<string>> GetValidNamespaceNames(CancellationToken cancellationToken = default)
+    async Task<HashSet<string>> GetValidNamespaceNames(CancellationToken cancellationToken)
     {
         var validNamespaces = new HashSet<string>(StringComparer.OrdinalIgnoreCase) { serviceBusName };
 
@@ -347,7 +347,7 @@ async Task<HashSet<string>> GetValidNamespaceNames(CancellationToken cancellatio
     ];
 
     protected override async Task<(bool Success, List<string> Errors)> TestConnectionCore(
-        CancellationToken cancellationToken)
+        CancellationToken cancellationToken = default)
     {
         await foreach (IBrokerQueue brokerQueue in GetQueueNames(cancellationToken))
         {

src/ServiceControl.Transports.ASBS/QueueLengthProvider.cs

@@ -27,24 +27,24 @@ public override void TrackEndpointInputQueue(EndpointToQueueMapping queueToTrack
                 (id, old) => queueToTrack.EndpointName
             );
 
-        protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+        protected override async Task ExecuteAsync(CancellationToken cancellationToken = default)
         {
-            while (!stoppingToken.IsCancellationRequested)
+            while (!cancellationToken.IsCancellationRequested)
             {
                 try
                 {
                     logger.LogDebug("Waiting for next interval");
-                    await Task.Delay(queryDelayInterval, stoppingToken);
+                    await Task.Delay(queryDelayInterval, cancellationToken);
 
                     logger.LogDebug("Querying management client");
 
-                    var queueRuntimeInfos = await GetQueueList(stoppingToken);
+                    var queueRuntimeInfos = await GetQueueList(cancellationToken);
 
                     logger.LogDebug("Retrieved details of {QueueCount} queues", queueRuntimeInfos.Count);
 
                     UpdateAllQueueLengths(queueRuntimeInfos);
                 }
-                catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
+                catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
                 {
                     // no-op
                 }

src/ServiceControl.Transports.ASQ.Tests/.editorconfig

@@ -2,8 +2,3 @@
 
 # Justification: Test project
 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.PS0018.severity = none

src/ServiceControl.Transports.ASQ.Tests/TransportTestsConfiguration.cs

@@ -1,6 +1,7 @@
 namespace ServiceControl.Transport.Tests
 {
     using System;
+    using System.Threading;
     using System.Threading.Tasks;
     using Transports;
     using Transports.ASQ;
@@ -11,7 +12,7 @@ partial class TransportTestsConfiguration
 
         public ITransportCustomization TransportCustomization { get; private set; }
 
-        public Task Configure()
+        public Task Configure(CancellationToken cancellationToken = default)
         {
             TransportCustomization = new ASQTransportCustomization();
             ConnectionString = Environment.GetEnvironmentVariable(ConnectionStringKey);
@@ -24,7 +25,7 @@ public Task Configure()
             return Task.CompletedTask;
         }
 
-        public Task Cleanup() => Task.CompletedTask;
+        public Task Cleanup(CancellationToken cancellationToken = default) => Task.CompletedTask;
 
         static string ConnectionStringKey = "ServiceControl_TransportTests_ASQ_ConnectionString";
     }

src/ServiceControl.Transports.ASQ/.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.PS0017.severity = none

src/ServiceControl.Transports.ASQ/QueueLengthProvider.cs

@@ -33,19 +33,19 @@ public override void TrackEndpointInputQueue(EndpointToQueueMapping queueToTrack
             queueLengths.AddOrUpdate(queueToTrack, _ => emptyQueueLength, (_, existingQueueLength) => existingQueueLength);
         }
 
-        protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+        protected override async Task ExecuteAsync(CancellationToken cancellationToken = default)
         {
-            while (!stoppingToken.IsCancellationRequested)
+            while (!cancellationToken.IsCancellationRequested)
             {
                 try
                 {
-                    await FetchQueueSizes(stoppingToken);
+                    await FetchQueueSizes(cancellationToken);
 
                     UpdateQueueLengthStore();
 
-                    await Task.Delay(QueryDelayInterval, stoppingToken);
+                    await Task.Delay(QueryDelayInterval, cancellationToken);
                 }
-                catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
+                catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
                 {
                     // no-op
                 }

src/ServiceControl.Transports.IBMMQ.Tests/TransportTestsConfiguration.cs

@@ -1,6 +1,7 @@
 namespace ServiceControl.Transport.Tests;
 
 using System;
+using System.Threading;
 using System.Threading.Tasks;
 using Transports;
 using Transports.IBMMQ;
@@ -21,7 +22,7 @@ class TransportTestsConfiguration
 
     public ITransportCustomization TransportCustomization { get; private set; }
 
-    public Task Configure()
+    public Task Configure(CancellationToken cancellationToken = default)
     {
         TransportCustomization = new TestIBMMQTransportCustomization();
         ConnectionString = Environment.GetEnvironmentVariable(ConnectionStringKey)
@@ -35,7 +36,7 @@ public Task Configure()
         return Task.CompletedTask;
     }
 
-    public Task Cleanup() => Task.CompletedTask;
+    public Task Cleanup(CancellationToken cancellationToken = default) => Task.CompletedTask;
 
     static string ConnectionStringKey = "ServiceControl_TransportTests_IBMMQ_ConnectionString";
 }