Fix API's are rebase with forcing accepting all changes from master

Author: ramonsmits

src/ServiceControl.Audit/Auditing/AuditIngestion.cs

@@ -128,7 +128,7 @@ async Task SetUpAndStartInfrastructure(CancellationToken cancellationToken)
 
                 messageReceiver = transportInfrastructure.Receivers[inputEndpoint];
 
-                await auditIngestor.VerifyCanReachForwardingAddress();
+                await auditIngestor.VerifyCanReachForwardingAddress(cancellationToken);
                 await messageReceiver.StartReceive(cancellationToken);
 
                 logger.Info(LogMessages.StartedInfrastructure);
@@ -235,7 +235,7 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
 
                             recorder.Tags.Add("ingestion.batch_size", contexts.Count);
 
-                            await auditIngestor.Ingest(contexts);
+                            await auditIngestor.Ingest(contexts, stoppingToken);
                         }
 
                         consecutiveBatchFailuresCounter.Record(0);

src/ServiceControl.Audit/Auditing/AuditIngestor.cs

@@ -4,6 +4,7 @@
     using System.Collections.Generic;
     using System.Diagnostics.Metrics;
     using System.Linq;
+    using System.Threading;
     using System.Threading.Tasks;
     using Infrastructure.Settings;
     using Monitoring;
@@ -43,15 +44,15 @@ ITransportCustomization transportCustomization
             );
         }
 
-        public async Task Ingest(List<MessageContext> contexts)
+        public async Task Ingest(List<MessageContext> contexts, CancellationToken cancellationToken)
         {
-            var stored = await auditPersister.Persist(contexts);
+            var stored = await auditPersister.Persist(contexts, cancellationToken);
 
             try
             {
                 if (settings.ForwardAuditMessages)
                 {
-                    await Forward(stored, logQueueAddress);
+                    await Forward(stored, logQueueAddress, cancellationToken);
                     forwardedMessagesCounter.Add(stored.Count);
                 }
 
@@ -69,7 +70,7 @@ public async Task Ingest(List<MessageContext> contexts)
             }
         }
 
-        Task Forward(IReadOnlyCollection<MessageContext> messageContexts, string forwardingAddress)
+        Task Forward(IReadOnlyCollection<MessageContext> messageContexts, string forwardingAddress, CancellationToken cancellationToken)
         {
             var transportOperations = new TransportOperation[messageContexts.Count]; //We could allocate based on the actual number of ProcessedMessages but this should be OK
             var index = 0;
@@ -98,12 +99,11 @@ Task Forward(IReadOnlyCollection<MessageContext> messageContexts, string forward
             return anyContext != null
                 ? messageDispatcher.Value.Dispatch(
                     new TransportOperations(transportOperations),
-                    anyContext.TransportTransaction
-                )
+                    anyContext.TransportTransaction, cancellationToken)
                 : Task.CompletedTask;
         }
 
-        public async Task VerifyCanReachForwardingAddress()
+        public async Task VerifyCanReachForwardingAddress(CancellationToken cancellationToken)
         {
             if (!settings.ForwardAuditMessages)
             {
@@ -120,7 +120,7 @@ public async Task VerifyCanReachForwardingAddress()
                     )
                 );
 
-                await messageDispatcher.Value.Dispatch(transportOperations, new TransportTransaction());
+                await messageDispatcher.Value.Dispatch(transportOperations, new TransportTransaction(), cancellationToken);
             }
             catch (Exception e)
             {

src/ServiceControl.Audit/Auditing/AuditPersister.cs

@@ -3,6 +3,7 @@
     using System;
     using System.Collections.Generic;
     using System.Text.Json;
+    using System.Threading;
     using System.Threading.Tasks;
     using Infrastructure;
     using Monitoring;
@@ -21,14 +22,14 @@ class AuditPersister(IAuditIngestionUnitOfWorkFactory unitOfWorkFactory,
         IMessageSession messageSession,
         Lazy<IMessageDispatcher> messageDispatcher)
     {
-        public async Task<IReadOnlyList<MessageContext>> Persist(IReadOnlyList<MessageContext> contexts)
+        public async Task<IReadOnlyList<MessageContext>> Persist(IReadOnlyList<MessageContext> contexts, CancellationToken cancellationToken)
         {
             var storedContexts = new List<MessageContext>(contexts.Count);
             IAuditIngestionUnitOfWork unitOfWork = null;
             try
             {
                 // deliberately not using the using statement because we dispose async explicitly
-                unitOfWork = await unitOfWorkFactory.StartNew(contexts.Count);
+                unitOfWork = await unitOfWorkFactory.StartNew(contexts.Count, cancellationToken);
                 var inserts = new List<Task>(contexts.Count);
                 foreach (var context in contexts)
                 {