Alternative to previous commit, stop receviers and shutdown with already cancelled token. This should terminate ASAP BUT still cleanup all resources if properly implemented by transports/persisters.

Author: ramonsmits

src/ServiceControl.Audit/Auditing/AuditIngestion.cs

@@ -185,7 +185,7 @@ async Task EnsureStopped(CancellationToken cancellationToken)
             try
             {
                 await startStopSemaphore.WaitAsync(cancellationToken);
-                await StopAndTeardownInfrastructure(cancellationToken);
+                await StopAndTeardownInfrastructure(new CancellationToken(canceled: true));
             }
             finally
             {

src/ServiceControl/Operations/ErrorIngestion.cs

@@ -132,7 +132,6 @@ public override async Task StopAsync(CancellationToken cancellationToken)
         {
             try
             {
-                await stoppingToken.CancelAsync();
                 await watchdog.Stop(cancellationToken);
                 channel.Writer.Complete();
                 await base.StopAsync(cancellationToken);
@@ -262,11 +261,8 @@ async Task StopAndTeardownInfrastructure(CancellationToken cancellationToken)
             }
         }
 
-        async Task OnMessage(MessageContext messageContext, CancellationToken cancellationTokenParent)
+        async Task OnMessage(MessageContext messageContext, CancellationToken cancellationToken)
         {
-            using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationTokenParent, stoppingToken.Token);
-            var cancellationToken = cts.Token;
-
             if (settings.MessageFilter != null && settings.MessageFilter(messageContext))
             {
                 return;
@@ -297,7 +293,7 @@ async Task EnsureStopped(CancellationToken cancellationToken = default)
             try
             {
                 await startStopSemaphore.WaitAsync(cancellationToken);
-                await StopAndTeardownInfrastructure(cancellationToken);
+                await StopAndTeardownInfrastructure(new CancellationToken(canceled: true));
             }
             finally
             {
@@ -311,7 +307,6 @@ async Task EnsureStopped(CancellationToken cancellationToken = default)
         TransportInfrastructure transportInfrastructure;
         IMessageReceiver messageReceiver;
 
-        readonly CancellationTokenSource stoppingToken = new();
         readonly Settings settings;
         readonly ITransportCustomization transportCustomization;
         readonly TransportSettings transportSettings;