1. OCE handling made false assumption that it can only happen during shutdown. Now logging info

2. Ingestion loop now has error handling to stop application when loop is terminated due to exception.

3. Added logging WRN when TrySetException fails

Author: ramonsmits

src/ServiceControl.Audit/Auditing/AuditIngestion.cs

@@ -206,49 +206,59 @@ async Task OnMessage(MessageContext messageContext, CancellationToken cancellati
 
         async Task Loop()
         {
-            var contexts = new List<MessageContext>(transportSettings.MaxConcurrency.Value);
-
-            while (await channel.Reader.WaitToReadAsync())
+            try
             {
-                // will only enter here if there is something to read.
-                try
+                var contexts = new List<MessageContext>(transportSettings.MaxConcurrency.Value);
+
+                while (await channel.Reader.WaitToReadAsync())
                 {
-                    // as long as there is something to read this will fetch up to MaximumConcurrency items
-                    while (channel.Reader.TryRead(out var context))
+                    // will only enter here if there is something to read.
+                    try
                     {
-                        contexts.Add(context);
+                        // as long as there is something to read this will fetch up to MaximumConcurrency items
+                        while (channel.Reader.TryRead(out var context))
+                        {
+                            contexts.Add(context);
+                        }
+
+                        batchSizeMeter.Mark(contexts.Count);
+                        using (batchDurationMeter.Measure())
+                        {
+                            await auditIngestor.Ingest(contexts);
+                        }
                     }
-
-                    batchSizeMeter.Mark(contexts.Count);
-                    using (batchDurationMeter.Measure())
+                    catch (OperationCanceledException e)
                     {
-                        await auditIngestor.Ingest(contexts);
+                        logger.Info("Ingesting messages failed", e);
+                        // continue loop, do nothing as we are shutting down
+                        // TODO: Assumption here is that OCE equals a shutdown which is definitely not the case.
                     }
-                }
-                catch (OperationCanceledException)
-                {
-                    //Do nothing as we are shutting down
-                    continue;
-                }
-                catch (Exception e) // show must go on
-                {
-                    if (logger.IsInfoEnabled)
+                    catch (Exception e) // show must go on
                     {
-                        logger.Info("Ingesting messages failed", e);
+                        logger.Warn("Ingesting messages failed", e);
+
+                        // signal all message handling tasks to terminate
+                        foreach (var context in contexts)
+                        {
+                            if (!context.GetTaskCompletionSource().TrySetException(e))
+                            {
+                                logger.Error("Loop TrySetException failed");
+                            }
+                        }
                     }
-
-                    // signal all message handling tasks to terminate
-                    foreach (var context in contexts)
+                    finally
                     {
-                        context.GetTaskCompletionSource().TrySetException(e);
+                        contexts.Clear();
                     }
                 }
-                finally
-                {
-                    contexts.Clear();
-                }
+                // will fall out here when writer is completed
+            }
+            catch (Exception e)
+            {
+                logger.Fatal("Loop interrupted", e);
+                applicationLifetime.StopApplication();
+                throw;
             }
-            // will fall out here when writer is completed
         }
 
         TransportInfrastructure transportInfrastructure;