- Applied changes from audit to error

- Removed duplication in EnsureStopped
- Removed log events around semaphore handling

Author: ramonsmits

src/ServiceControl.Audit/Auditing/AuditIngestion.cs

@@ -72,9 +72,7 @@ async Task EnsureStarted(CancellationToken cancellationToken)
         {
             try
             {
-                logger.Debug("Ensure started. Start/stop semaphore acquiring");
                 await startStopSemaphore.WaitAsync(cancellationToken);
-                logger.Debug("Ensure started. Start/stop semaphore acquired");
 
                 var canIngest = unitOfWorkFactory.CanIngestMore();
 
@@ -104,15 +102,13 @@ async Task EnsureStarted(CancellationToken cancellationToken)
             }
             finally
             {
-                logger.Debug("Ensure started. Start/stop semaphore releasing");
                 startStopSemaphore.Release();
-                logger.Debug("Ensure started. Start/stop semaphore released");
             }
         }
 
         async Task SetUpAndStartInfrastructure(CancellationToken cancellationToken)
         {
-            if (queueIngestor != null)
+            if (messageReceiver != null)
             {
                 logger.Debug("Infrastructure already Started");
                 return;
@@ -130,10 +126,10 @@ async Task SetUpAndStartInfrastructure(CancellationToken cancellationToken)
                     TransportTransactionMode.ReceiveOnly
                 );
 
-                queueIngestor = transportInfrastructure.Receivers[inputEndpoint];
+                messageReceiver = transportInfrastructure.Receivers[inputEndpoint];
 
                 await auditIngestor.VerifyCanReachForwardingAddress();
-                await queueIngestor.StartReceive(cancellationToken);
+                await messageReceiver.StartReceive(cancellationToken);
 
                 logger.Info(LogMessages.StartedInfrastructure);
             }
@@ -157,17 +153,19 @@ async Task StopAndTeardownInfrastructure(CancellationToken cancellationToken)
                 logger.Info("Stopping infrastructure");
                 try
                 {
-                    if (queueIngestor != null)
+                    if (messageReceiver != null)
                     {
-                        await queueIngestor.StopReceive(cancellationToken);
+                        await messageReceiver.StopReceive(cancellationToken);
                     }
                 }
                 finally
                 {
                     await transportInfrastructure.Shutdown(cancellationToken);
                 }
 
-                queueIngestor = null;
+                messageReceiver = null;
+                transportInfrastructure = null;
+
                 logger.Info(LogMessages.StoppedInfrastructure);
             }
             catch (Exception e)
@@ -181,31 +179,12 @@ async Task EnsureStopped(CancellationToken cancellationToken)
         {
             try
             {
-                logger.Info("Shutting down. Start/stop semaphore acquiring");
                 await startStopSemaphore.WaitAsync(cancellationToken);
-                logger.Info("Shutting down. Start/stop semaphore acquired");
-
-                if (queueIngestor == null)
-                {
-                    logger.Info("Shutting down. Already stopped, skipping shut down");
-                    return; //Already stopped
-                }
-
-                var stoppable = queueIngestor;
-                queueIngestor = null;
-                logger.Info("Shutting down. Infrastructure shut down commencing");
-                await stoppable.StopReceive(cancellationToken);
-                logger.Info("Shutting down. Infrastructure shut down completed");
-            }
-            catch (OperationCanceledException e) when (e.CancellationToken == cancellationToken)
-            {
-                logger.Info("StopReceive cancelled");
+                await StopAndTeardownInfrastructure(cancellationToken);
             }
             finally
             {
-                logger.Info("Shutting down. Start/stop semaphore releasing");
                 startStopSemaphore.Release();
-                logger.Info("Shutting down. Start/stop semaphore released");
             }
         }
 
@@ -318,7 +297,7 @@ public override async Task StopAsync(CancellationToken cancellationToken)
         }
 
         TransportInfrastructure transportInfrastructure;
-        IMessageReceiver queueIngestor;
+        IMessageReceiver messageReceiver;
         long consecutiveBatchFailures = 0;
 
         readonly SemaphoreSlim startStopSemaphore = new(1);

src/ServiceControl/Operations/ErrorIngestion.cs

@@ -158,30 +158,57 @@ async Task EnsureStarted(CancellationToken cancellationToken = default)
             {
                 await startStopSemaphore.WaitAsync(cancellationToken);
 
-                if (!unitOfWorkFactory.CanIngestMore())
+                var canIngest = unitOfWorkFactory.CanIngestMore();
+
+                Logger.DebugFormat("Ensure started {0}", canIngest);
+
+                if (canIngest)
                 {
-                    if (messageReceiver != null)
-                    {
-                        var stoppable = messageReceiver;
-                        messageReceiver = null;
-                        await stoppable.StopReceive(cancellationToken);
-                        Logger.Info("Shutting down due to failed persistence health check. Infrastructure shut down completed");
-                    }
-                    return;
+                    await SetUpAndStartInfrastructure(cancellationToken);
                 }
-
-                if (messageReceiver != null)
+                else
+                {
+                    await StopAndTeardownInfrastructure(cancellationToken);
+                }
+            }
+            catch (Exception e)
+            {
+                try
                 {
-                    return; //Already started
+                    await StopAndTeardownInfrastructure(cancellationToken);
+                }
+                catch (Exception teardownException)
+                {
+                    throw new AggregateException(e, teardownException);
                 }
 
+                throw;
+            }
+            finally
+            {
+                startStopSemaphore.Release();
+            }
+        }
+
+        async Task SetUpAndStartInfrastructure(CancellationToken cancellationToken)
+        {
+            if (messageReceiver != null)
+            {
+                Logger.Debug("Infrastructure already Started");
+                return;
+            }
+
+            try
+            {
+                Logger.Info("Starting infrastructure");
                 transportInfrastructure = await transportCustomization.CreateTransportInfrastructure(
                     errorQueue,
                     transportSettings,
                     OnMessage,
                     errorHandlingPolicy.OnError,
                     OnCriticalError,
-                    TransportTransactionMode.ReceiveOnly);
+                    TransportTransactionMode.ReceiveOnly
+                );
 
                 messageReceiver = transportInfrastructure.Receivers[errorQueue];
 
@@ -192,22 +219,45 @@ async Task EnsureStarted(CancellationToken cancellationToken = default)
 
                 await messageReceiver.StartReceive(cancellationToken);
 
-                Logger.Info("Ensure started. Infrastructure started");
+                Logger.Info(LogMessages.StartedInfrastructure);
             }
-            catch
+            catch (Exception e)
             {
-                if (messageReceiver != null)
+                Logger.Error("Failed to start infrastructure", e);
+                throw;
+            }
+        }
+        async Task StopAndTeardownInfrastructure(CancellationToken cancellationToken)
+        {
+            if (transportInfrastructure == null)
+            {
+                Logger.Debug("Infrastructure already Stopped");
+                return;
+            }
+            try
+            {
+                Logger.Info("Stopping infrastructure");
+                try
+                {
+                    if (messageReceiver != null)
+                    {
+                        await messageReceiver.StopReceive(cancellationToken);
+                    }
+                }
+                finally
                 {
-                    await messageReceiver.StopReceive(cancellationToken);
+                    await transportInfrastructure.Shutdown(cancellationToken);
                 }
 
-                messageReceiver = null; // Setting to null so that it doesn't exit when it retries in line 134
+                messageReceiver = null;
+                transportInfrastructure = null;
 
-                throw;
+                Logger.Info(LogMessages.StoppedInfrastructure);
             }
-            finally
+            catch (Exception e)
             {
-                startStopSemaphore.Release();
+                Logger.Error("Failed to stop infrastructure", e);
+                throw;
             }
         }
 
@@ -238,22 +288,15 @@ async Task EnsureStopped(CancellationToken cancellationToken = default)
             try
             {
                 await startStopSemaphore.WaitAsync(cancellationToken);
-
-                if (messageReceiver == null)
-                {
-                    return; //Already stopped
-                }
-                var stoppable = messageReceiver;
-                messageReceiver = null;
-                await stoppable.StopReceive(cancellationToken);
+                await StopAndTeardownInfrastructure(cancellationToken);
             }
             finally
             {
                 startStopSemaphore.Release();
             }
         }
 
-        SemaphoreSlim startStopSemaphore = new SemaphoreSlim(1);
+        SemaphoreSlim startStopSemaphore = new(1);
         string errorQueue;
         ErrorIngestionFaultPolicy errorHandlingPolicy;
         TransportInfrastructure transportInfrastructure;
@@ -272,5 +315,11 @@ async Task EnsureStopped(CancellationToken cancellationToken = default)
         readonly IHostApplicationLifetime applicationLifetime;
 
         static readonly ILog Logger = LogManager.GetLogger<ErrorIngestion>();
+
+        internal static class LogMessages
+        {
+            internal const string StartedInfrastructure = "Started infrastructure";
+            internal const string StoppedInfrastructure = "Stopped infrastructure";
+        }
     }
 }