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

# Conflicts:
#	src/ServiceControl.Audit/Auditing/AuditIngestion.cs
#	src/ServiceControl.Audit/Auditing/AuditIngestor.cs

Author: ramonsmits

src/ServiceControl.Audit/Auditing/AuditIngestion.cs

@@ -27,7 +27,8 @@ public AuditIngestion(
             AuditIngestor auditIngestor,
             IAuditIngestionUnitOfWorkFactory unitOfWorkFactory,
             IHostApplicationLifetime applicationLifetime,
-            IngestionMetrics metrics)
+            IngestionMetrics metrics
+        )
         {
             inputEndpoint = settings.AuditQueue;
             this.transportCustomization = transportCustomization;
@@ -132,7 +133,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);
@@ -236,7 +237,7 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
                             contexts.Add(context);
                         }
 
-                        await auditIngestor.Ingest(contexts);
+                        await auditIngestor.Ingest(contexts, stoppingToken);
 
                         batchMetrics.Complete(contexts.Count);
                     }

src/ServiceControl.Audit/Auditing/AuditIngestor.cs

@@ -3,6 +3,7 @@
     using System;
     using System.Collections.Generic;
     using System.Linq;
+    using System.Threading;
     using System.Threading.Tasks;
     using Infrastructure.Settings;
     using Monitoring;
@@ -42,15 +43,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);
                 }
 
                 foreach (var context in contexts)
@@ -67,7 +68,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;
@@ -96,12 +97,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)
             {
@@ -118,7 +118,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)
                 {