Support cancellation for error ingestion

ramonsmits · ramonsmits · commit bb515dcd34d3 · 2025-03-06T12:14:54.000+01:00
diff --git a/src/ServiceControl.Persistence.RavenDB/UnitOfWork/RavenIngestionUnitOfWork.cs b/src/ServiceControl.Persistence.RavenDB/UnitOfWork/RavenIngestionUnitOfWork.cs
@@ -1,8 +1,8 @@
 ﻿namespace ServiceControl.Persistence.RavenDB
 {
     using System.Collections.Concurrent;
+    using System.Threading;
     using System.Threading.Tasks;
-    using Raven.Client.Documents;
     using Raven.Client.Documents.Commands.Batches;
     using ServiceControl.Persistence.UnitOfWork;
 
@@ -22,12 +22,12 @@ public RavenIngestionUnitOfWork(IRavenSessionProvider sessionProvider, Expiratio
 
         internal void AddCommand(ICommandData command) => commands.Enqueue(command);
 
-        public override async Task Complete()
+        public override async Task Complete(CancellationToken cancellationToken)
         {
-            using var session = await sessionProvider.OpenSession();
+            using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken);
             // not really interested in the batch results since a batch is atomic
             session.Advanced.Defer(commands.ToArray());
-            await session.SaveChangesAsync();
+            await session.SaveChangesAsync(cancellationToken);
         }
     }
 }
diff --git a/src/ServiceControl.Persistence.Tests.RavenDB/Expiration/MessageExpiryTests.cs b/src/ServiceControl.Persistence.Tests.RavenDB/Expiration/MessageExpiryTests.cs
@@ -38,7 +38,7 @@ public async Task SingleMessageMarkedAsArchiveShouldExpire()
             {
                 await uow.Recoverability.RecordFailedProcessingAttempt(context, attempt, []);
 
-                await uow.Complete();
+                await uow.Complete(TestContext.CurrentContext.CancellationToken);
             }
 
             CompleteDatabaseOperation();
@@ -68,7 +68,7 @@ public async Task AllMessagesInUnArchivedGroupShouldNotExpire()
                 await uow.Recoverability.RecordFailedProcessingAttempt(contextA, attemptA, [new FailedMessage.FailureGroup { Id = groupIdA }]);
                 await uow.Recoverability.RecordFailedProcessingAttempt(contextB, attemptB, [new FailedMessage.FailureGroup { Id = groupIdB }]);
 
-                await uow.Complete();
+                await uow.Complete(TestContext.CurrentContext.CancellationToken);
             }
 
             CompleteDatabaseOperation();
@@ -99,7 +99,7 @@ public async Task AllMessagesInArchivedGroupShouldExpire()
             {
                 await uow.Recoverability.RecordFailedProcessingAttempt(context, attempt, [new FailedMessage.FailureGroup { Id = groupId }]);
 
-                await uow.Complete();
+                await uow.Complete(TestContext.CurrentContext.CancellationToken);
             }
 
             CompleteDatabaseOperation();
@@ -122,7 +122,7 @@ public async Task SingleMessageMarkedAsResolvedShouldExpire()
             {
                 await uow.Recoverability.RecordFailedProcessingAttempt(context, attempt, []);
 
-                await uow.Complete();
+                await uow.Complete(TestContext.CurrentContext.CancellationToken);
             }
 
             CompleteDatabaseOperation();
@@ -145,7 +145,7 @@ public async Task RetryConfirmationProcessingShouldTriggerExpiration()
             {
                 await uow.Recoverability.RecordFailedProcessingAttempt(context, attempt, []);
 
-                await uow.Complete();
+                await uow.Complete(TestContext.CurrentContext.CancellationToken);
             }
 
             CompleteDatabaseOperation();
@@ -158,7 +158,7 @@ public async Task RetryConfirmationProcessingShouldTriggerExpiration()
             {
                 await uow.Recoverability.RecordSuccessfulRetry(errors.Results.First().Id);
 
-                await uow.Complete();
+                await uow.Complete(TestContext.CurrentContext.CancellationToken);
             }
 
             await WaitUntil(async () => (await GetAllMessages()).Results.Count == 0, "Retry confirmation should cause message removal.");
diff --git a/src/ServiceControl.Persistence.Tests/BodyStorage/AttachmentsBodyStorageTests.cs b/src/ServiceControl.Persistence.Tests/BodyStorage/AttachmentsBodyStorageTests.cs
@@ -70,7 +70,7 @@ async Task RunTest(Func<Dictionary<string, string>, string> getIdToQuery)
                 var groups = new List<FailedMessage.FailureGroup>();
 
                 await uow.Recoverability.RecordFailedProcessingAttempt(context, processingAttempt, groups);
-                await uow.Complete();
+                await uow.Complete(cancellationSource.Token);
             }
 
             CompleteDatabaseOperation();
diff --git a/src/ServiceControl.Persistence.Tests/MonitoringDataStoreTests.cs b/src/ServiceControl.Persistence.Tests/MonitoringDataStoreTests.cs
@@ -134,7 +134,7 @@ public async Task Unit_of_work_detects_endpoint()
             {
                 await unitOfWork.Monitoring.RecordKnownEndpoint(knownEndpoint);
 
-                await unitOfWork.Complete();
+                await unitOfWork.Complete(TestContext.CurrentContext.CancellationToken);
             }
 
             CompleteDatabaseOperation();
diff --git a/src/ServiceControl.Persistence.Tests/Recoverability/RetryConfirmationProcessorTests.cs b/src/ServiceControl.Persistence.Tests/Recoverability/RetryConfirmationProcessorTests.cs
@@ -49,7 +49,7 @@ public async Task Should_handle_multiple_retry_confirmations_in_the_error_ingest
             var unitOfWork = await UnitOfWorkFactory.StartNew();
             await Processor.Process(messageContexts, unitOfWork);
 
-            Assert.DoesNotThrowAsync(() => unitOfWork.Complete());
+            Assert.DoesNotThrowAsync(() => unitOfWork.Complete(TestContext.CurrentContext.CancellationToken));
         }
 
         [Test]
@@ -71,7 +71,7 @@ public async Task Should_handle_retry_confirmation_followed_by_legacy_command()
 
             var unitOfWork = await UnitOfWorkFactory.StartNew();
             await Processor.Process(messageContexts, unitOfWork);
-            await unitOfWork.Complete();
+            await unitOfWork.Complete(TestContext.CurrentContext.CancellationToken);
 
             Assert.DoesNotThrowAsync(
                 () => Handler.Handle(CreateLegacyRetryConfirmationCommand(), new TestableInvokeHandlerContext()));
@@ -89,7 +89,7 @@ public async Task Should_handle_legacy_retry_confirmation_command_followed_by_ne
 
             var unitOfWork = await UnitOfWorkFactory.StartNew();
             await Processor.Process(messageContexts, unitOfWork);
-            Assert.DoesNotThrowAsync(() => unitOfWork.Complete());
+            Assert.DoesNotThrowAsync(() => unitOfWork.Complete(TestContext.CurrentContext.CancellationToken));
         }
 
         static MarkMessageFailureResolvedByRetry CreateLegacyRetryConfirmationCommand()
diff --git a/src/ServiceControl.Persistence/UnitOfWork/FallbackIngestionUnitOfWork.cs b/src/ServiceControl.Persistence/UnitOfWork/FallbackIngestionUnitOfWork.cs
@@ -1,5 +1,6 @@
 ﻿namespace ServiceControl.Persistence.UnitOfWork
 {
+    using System.Threading;
     using System.Threading.Tasks;
 
     // HINT: This allows an implementor to provide only part of the implementation and allow the other part
@@ -18,7 +19,11 @@ public FallbackIngestionUnitOfWork(IIngestionUnitOfWork primary, IIngestionUnitO
             Recoverability = primary.Recoverability ?? fallback.Recoverability;
         }
 
-        public override Task Complete() => Task.WhenAll(primary.Complete(), fallback.Complete());
+        public override Task Complete(CancellationToken cancellationToken)
+            => Task.WhenAll(
+                primary.Complete(cancellationToken),
+                fallback.Complete(cancellationToken)
+            );
 
         protected override void Dispose(bool disposing)
         {
diff --git a/src/ServiceControl.Persistence/UnitOfWork/IIngestionUnitOfWork.cs b/src/ServiceControl.Persistence/UnitOfWork/IIngestionUnitOfWork.cs
@@ -1,12 +1,13 @@
 ﻿namespace ServiceControl.Persistence.UnitOfWork
 {
     using System;
+    using System.Threading;
     using System.Threading.Tasks;
 
     public interface IIngestionUnitOfWork : IDisposable
     {
         IMonitoringIngestionUnitOfWork Monitoring { get; }
         IRecoverabilityIngestionUnitOfWork Recoverability { get; }
-        Task Complete();
+        Task Complete(CancellationToken cancellationToken);
     }
 }
diff --git a/src/ServiceControl.Persistence/UnitOfWork/IngestionUnitOfWorkBase.cs b/src/ServiceControl.Persistence/UnitOfWork/IngestionUnitOfWorkBase.cs
@@ -1,6 +1,7 @@
 ﻿namespace ServiceControl.Persistence.UnitOfWork
 {
     using System;
+    using System.Threading;
     using System.Threading.Tasks;
 
     public abstract class IngestionUnitOfWorkBase : IIngestionUnitOfWork
@@ -19,6 +20,6 @@ public void Dispose()
 
         public IMonitoringIngestionUnitOfWork Monitoring { get; protected set; }
         public IRecoverabilityIngestionUnitOfWork Recoverability { get; protected set; }
-        public virtual Task Complete() => Task.CompletedTask;
+        public virtual Task Complete(CancellationToken cancellationToken) => Task.CompletedTask;
     }
 }
diff --git a/src/ServiceControl/Operations/ErrorIngestion.cs b/src/ServiceControl/Operations/ErrorIngestion.cs
@@ -96,7 +96,7 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
                         batchSizeMeter.Mark(contexts.Count);
                         using (batchDurationMeter.Measure())
                         {
-                            await ingestor.Ingest(contexts);
+                            await ingestor.Ingest(contexts, stoppingToken);
                         }
                     }
                     catch (Exception e)
@@ -214,7 +214,7 @@ async Task SetUpAndStartInfrastructure(CancellationToken cancellationToken)
 
                 if (settings.ForwardErrorMessages)
                 {
-                    await ingestor.VerifyCanReachForwardingAddress();
+                    await ingestor.VerifyCanReachForwardingAddress(cancellationToken);
                 }
 
                 await messageReceiver.StartReceive(cancellationToken);
diff --git a/src/ServiceControl/Operations/ErrorIngestor.cs b/src/ServiceControl/Operations/ErrorIngestor.cs
@@ -4,6 +4,7 @@
     using System.Collections.Generic;
     using System.Diagnostics;
     using System.Linq;
+    using System.Threading;
     using System.Threading.Tasks;
     using Contracts.Operations;
     using Infrastructure.DomainEvents;
@@ -49,7 +50,7 @@ public ErrorIngestor(Metrics metrics,
             logQueueAddress = new UnicastAddressTag(transportCustomization.ToTransportQualifiedQueueName(this.settings.ErrorLogQueue));
         }
 
-        public async Task Ingest(List<MessageContext> contexts)
+        public async Task Ingest(List<MessageContext> contexts, CancellationToken cancellationToken)
         {
             var failedMessages = new List<MessageContext>(contexts.Count);
             var retriedMessages = new List<MessageContext>(contexts.Count);
@@ -67,7 +68,7 @@ public async Task Ingest(List<MessageContext> contexts)
             }
 
 
-            var storedFailed = await PersistFailedMessages(failedMessages, retriedMessages);
+            var storedFailed = await PersistFailedMessages(failedMessages, retriedMessages, cancellationToken);
 
             try
             {
@@ -89,7 +90,7 @@ public async Task Ingest(List<MessageContext> contexts)
                     {
                         Logger.Debug($"Forwarding {storedFailed.Count} messages");
                     }
-                    await Forward(storedFailed);
+                    await Forward(storedFailed, cancellationToken);
                     if (Logger.IsDebugEnabled)
                     {
                         Logger.Debug("Forwarded messages");
@@ -113,7 +114,7 @@ public async Task Ingest(List<MessageContext> contexts)
             }
         }
 
-        async Task<IReadOnlyList<MessageContext>> PersistFailedMessages(List<MessageContext> failedMessageContexts, List<MessageContext> retriedMessageContexts)
+        async Task<IReadOnlyList<MessageContext>> PersistFailedMessages(List<MessageContext> failedMessageContexts, List<MessageContext> retriedMessageContexts, CancellationToken cancellationToken)
         {
             var stopwatch = Stopwatch.StartNew();
 
@@ -130,7 +131,7 @@ async Task<IReadOnlyList<MessageContext>> PersistFailedMessages(List<MessageCont
 
                 using (bulkInsertDurationMeter.Measure())
                 {
-                    await unitOfWork.Complete();
+                    await unitOfWork.Complete(cancellationToken);
                 }
                 return storedFailedMessageContexts;
             }
@@ -154,7 +155,7 @@ async Task<IReadOnlyList<MessageContext>> PersistFailedMessages(List<MessageCont
             }
         }
 
-        Task Forward(IReadOnlyCollection<MessageContext> messageContexts)
+        Task Forward(IReadOnlyCollection<MessageContext> messageContexts, 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;
@@ -177,11 +178,11 @@ Task Forward(IReadOnlyCollection<MessageContext> messageContexts)
             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)
         {
             try
             {
@@ -193,7 +194,7 @@ public async Task VerifyCanReachForwardingAddress()
                     )
                 );
 
-                await messageDispatcher.Value.Dispatch(transportOperations, new TransportTransaction());
+                await messageDispatcher.Value.Dispatch(transportOperations, new TransportTransaction(), cancellationToken);
             }
             catch (Exception e)
             {
diff --git a/src/ServiceControl/Operations/ImportFailedErrors.cs b/src/ServiceControl/Operations/ImportFailedErrors.cs
@@ -16,17 +16,23 @@ public async Task Run(CancellationToken cancellationToken = default)
         {
             if (settings.ForwardErrorMessages)
             {
-                await errorIngestor.VerifyCanReachForwardingAddress();
+                await errorIngestor.VerifyCanReachForwardingAddress(cancellationToken);
             }
 
             await store.ProcessFailedErrorImports(async transportMessage =>
             {
-                var messageContext = new MessageContext(transportMessage.Id, transportMessage.Headers, transportMessage.Body, EmptyTransaction, settings.ErrorQueue, EmptyContextBag);
-                var taskCompletionSource =
-                    new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+                var messageContext = new MessageContext(
+                    transportMessage.Id,
+                    transportMessage.Headers,
+                    transportMessage.Body,
+                    EmptyTransaction,
+                    settings.ErrorQueue,
+                    EmptyContextBag
+                );
+                var taskCompletionSource = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
                 messageContext.SetTaskCompletionSource(taskCompletionSource);
 
-                await errorIngestor.Ingest([messageContext]);
+                await errorIngestor.Ingest([messageContext], cancellationToken);
                 await taskCompletionSource.Task;
             }, cancellationToken);
         }

Original file line number	Diff line number	Diff line change
`@@ -1,8 +1,8 @@`
`1`	`1`	`namespace ServiceControl.Persistence.RavenDB`
`2`	`2`	`{`
`3`	`3`	`using System.Collections.Concurrent;`
	`4`	`+ using System.Threading;`
`4`	`5`	`using System.Threading.Tasks;`
`5`		`- using Raven.Client.Documents;`
`6`	`6`	`using Raven.Client.Documents.Commands.Batches;`
`7`	`7`	`using ServiceControl.Persistence.UnitOfWork;`
`8`	`8`
`@@ -22,12 +22,12 @@ public RavenIngestionUnitOfWork(IRavenSessionProvider sessionProvider, Expiratio`
`22`	`22`
`23`	`23`	`internal void AddCommand(ICommandData command) => commands.Enqueue(command);`
`24`	`24`
`25`		`- public override async Task Complete()`
	`25`	`+ public override async Task Complete(CancellationToken cancellationToken)`
`26`	`26`	`{`
`27`		`- using var session = await sessionProvider.OpenSession();`
	`27`	`+ using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken);`
`28`	`28`	`// not really interested in the batch results since a batch is atomic`
`29`	`29`	`session.Advanced.Defer(commands.ToArray());`
`30`		`- await session.SaveChangesAsync();`
	`30`	`+ await session.SaveChangesAsync(cancellationToken);`
`31`	`31`	`}`
`32`	`32`	`}`
`33`	`33`	`}`
Original file line number	Diff line number	Diff line change
`@@ -38,7 +38,7 @@ public async Task SingleMessageMarkedAsArchiveShouldExpire()`
`38`	`38`	`{`
`39`	`39`	`await uow.Recoverability.RecordFailedProcessingAttempt(context, attempt, []);`
`40`	`40`
`41`		`- await uow.Complete();`
	`41`	`+ await uow.Complete(TestContext.CurrentContext.CancellationToken);`
`42`	`42`	`}`
`43`	`43`
`44`	`44`	`CompleteDatabaseOperation();`
`@@ -68,7 +68,7 @@ public async Task AllMessagesInUnArchivedGroupShouldNotExpire()`
`68`	`68`	`await uow.Recoverability.RecordFailedProcessingAttempt(contextA, attemptA, [new FailedMessage.FailureGroup { Id = groupIdA }]);`
`69`	`69`	`await uow.Recoverability.RecordFailedProcessingAttempt(contextB, attemptB, [new FailedMessage.FailureGroup { Id = groupIdB }]);`
`70`	`70`
`71`		`- await uow.Complete();`
	`71`	`+ await uow.Complete(TestContext.CurrentContext.CancellationToken);`
`72`	`72`	`}`
`73`	`73`
`74`	`74`	`CompleteDatabaseOperation();`
`@@ -99,7 +99,7 @@ public async Task AllMessagesInArchivedGroupShouldExpire()`
`99`	`99`	`{`
`100`	`100`	`await uow.Recoverability.RecordFailedProcessingAttempt(context, attempt, [new FailedMessage.FailureGroup { Id = groupId }]);`
`101`	`101`
`102`		`- await uow.Complete();`
	`102`	`+ await uow.Complete(TestContext.CurrentContext.CancellationToken);`
`103`	`103`	`}`
`104`	`104`
`105`	`105`	`CompleteDatabaseOperation();`
`@@ -122,7 +122,7 @@ public async Task SingleMessageMarkedAsResolvedShouldExpire()`
`122`	`122`	`{`
`123`	`123`	`await uow.Recoverability.RecordFailedProcessingAttempt(context, attempt, []);`
`124`	`124`
`125`		`- await uow.Complete();`
	`125`	`+ await uow.Complete(TestContext.CurrentContext.CancellationToken);`
`126`	`126`	`}`
`127`	`127`
`128`	`128`	`CompleteDatabaseOperation();`
`@@ -145,7 +145,7 @@ public async Task RetryConfirmationProcessingShouldTriggerExpiration()`
`145`	`145`	`{`
`146`	`146`	`await uow.Recoverability.RecordFailedProcessingAttempt(context, attempt, []);`
`147`	`147`
`148`		`- await uow.Complete();`
	`148`	`+ await uow.Complete(TestContext.CurrentContext.CancellationToken);`
`149`	`149`	`}`
`150`	`150`
`151`	`151`	`CompleteDatabaseOperation();`
`@@ -158,7 +158,7 @@ public async Task RetryConfirmationProcessingShouldTriggerExpiration()`
`158`	`158`	`{`
`159`	`159`	`await uow.Recoverability.RecordSuccessfulRetry(errors.Results.First().Id);`
`160`	`160`
`161`		`- await uow.Complete();`
	`161`	`+ await uow.Complete(TestContext.CurrentContext.CancellationToken);`
`162`	`162`	`}`
`163`	`163`
`164`	`164`	`await WaitUntil(async () => (await GetAllMessages()).Results.Count == 0, "Retry confirmation should cause message removal.");`
Original file line number	Diff line number	Diff line change
`@@ -70,7 +70,7 @@ async Task RunTest(Func<Dictionary<string, string>, string> getIdToQuery)`
`70`	`70`	`var groups = new List<FailedMessage.FailureGroup>();`
`71`	`71`
`72`	`72`	`await uow.Recoverability.RecordFailedProcessingAttempt(context, processingAttempt, groups);`
`73`		`- await uow.Complete();`
	`73`	`+ await uow.Complete(cancellationSource.Token);`
`74`	`74`	`}`
`75`	`75`
`76`	`76`	`CompleteDatabaseOperation();`
Original file line number	Diff line number	Diff line change
`@@ -134,7 +134,7 @@ public async Task Unit_of_work_detects_endpoint()`
`134`	`134`	`{`
`135`	`135`	`await unitOfWork.Monitoring.RecordKnownEndpoint(knownEndpoint);`
`136`	`136`
`137`		`- await unitOfWork.Complete();`
	`137`	`+ await unitOfWork.Complete(TestContext.CurrentContext.CancellationToken);`
`138`	`138`	`}`
`139`	`139`
`140`	`140`	`CompleteDatabaseOperation();`
Original file line number	Diff line number	Diff line change
`@@ -1,5 +1,6 @@`
`1`	`1`	`namespace ServiceControl.Persistence.UnitOfWork`
`2`	`2`	`{`
	`3`	`+ using System.Threading;`
`3`	`4`	`using System.Threading.Tasks;`
`4`	`5`
`5`	`6`	`// HINT: This allows an implementor to provide only part of the implementation and allow the other part`
`@@ -18,7 +19,11 @@ public FallbackIngestionUnitOfWork(IIngestionUnitOfWork primary, IIngestionUnitO`
`18`	`19`	`Recoverability = primary.Recoverability ?? fallback.Recoverability;`
`19`	`20`	`}`
`20`	`21`
`21`		`- public override Task Complete() => Task.WhenAll(primary.Complete(), fallback.Complete());`
	`22`	`+ public override Task Complete(CancellationToken cancellationToken)`
	`23`	`+ => Task.WhenAll(`
	`24`	`+ primary.Complete(cancellationToken),`
	`25`	`+ fallback.Complete(cancellationToken)`
	`26`	`+ );`
`22`	`27`
`23`	`28`	`protected override void Dispose(bool disposing)`
`24`	`29`	`{`
Original file line number	Diff line number	Diff line change
`@@ -1,12 +1,13 @@`
`1`	`1`	`namespace ServiceControl.Persistence.UnitOfWork`
`2`	`2`	`{`
`3`	`3`	`using System;`
	`4`	`+ using System.Threading;`
`4`	`5`	`using System.Threading.Tasks;`
`5`	`6`
`6`	`7`	`public interface IIngestionUnitOfWork : IDisposable`
`7`	`8`	`{`
`8`	`9`	`IMonitoringIngestionUnitOfWork Monitoring { get; }`
`9`	`10`	`IRecoverabilityIngestionUnitOfWork Recoverability { get; }`
`10`		`- Task Complete();`
	`11`	`+ Task Complete(CancellationToken cancellationToken);`
`11`	`12`	`}`
`12`	`13`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1,6 +1,7 @@`
`1`	`1`	`namespace ServiceControl.Persistence.UnitOfWork`
`2`	`2`	`{`
`3`	`3`	`using System;`
	`4`	`+ using System.Threading;`
`4`	`5`	`using System.Threading.Tasks;`
`5`	`6`
`6`	`7`	`public abstract class IngestionUnitOfWorkBase : IIngestionUnitOfWork`
`@@ -19,6 +20,6 @@ public void Dispose()`
`19`	`20`
`20`	`21`	`public IMonitoringIngestionUnitOfWork Monitoring { get; protected set; }`
`21`	`22`	`public IRecoverabilityIngestionUnitOfWork Recoverability { get; protected set; }`
`22`		`- public virtual Task Complete() => Task.CompletedTask;`
	`23`	`+ public virtual Task Complete(CancellationToken cancellationToken) => Task.CompletedTask;`
`23`	`24`	`}`
`24`	`25`	`}`
Original file line number	Diff line number	Diff line change
`@@ -96,7 +96,7 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)`
`96`	`96`	`batchSizeMeter.Mark(contexts.Count);`
`97`	`97`	`using (batchDurationMeter.Measure())`
`98`	`98`	`{`
`99`		`- await ingestor.Ingest(contexts);`
	`99`	`+ await ingestor.Ingest(contexts, stoppingToken);`
`100`	`100`	`}`
`101`	`101`	`}`
`102`	`102`	`catch (Exception e)`
`@@ -214,7 +214,7 @@ async Task SetUpAndStartInfrastructure(CancellationToken cancellationToken)`
`214`	`214`
`215`	`215`	`if (settings.ForwardErrorMessages)`
`216`	`216`	`{`
`217`		`- await ingestor.VerifyCanReachForwardingAddress();`
	`217`	`+ await ingestor.VerifyCanReachForwardingAddress(cancellationToken);`
`218`	`218`	`}`
`219`	`219`
`220`	`220`	`await messageReceiver.StartReceive(cancellationToken);`
Original file line number	Diff line number	Diff line change
`@@ -4,6 +4,7 @@`
`4`	`4`	`using System.Collections.Generic;`
`5`	`5`	`using System.Diagnostics;`
`6`	`6`	`using System.Linq;`
	`7`	`+ using System.Threading;`
`7`	`8`	`using System.Threading.Tasks;`
`8`	`9`	`using Contracts.Operations;`
`9`	`10`	`using Infrastructure.DomainEvents;`
`@@ -49,7 +50,7 @@ public ErrorIngestor(Metrics metrics,`
`49`	`50`	`logQueueAddress = new UnicastAddressTag(transportCustomization.ToTransportQualifiedQueueName(this.settings.ErrorLogQueue));`
`50`	`51`	`}`
`51`	`52`
`52`		`- public async Task Ingest(List<MessageContext> contexts)`
	`53`	`+ public async Task Ingest(List<MessageContext> contexts, CancellationToken cancellationToken)`
`53`	`54`	`{`
`54`	`55`	`var failedMessages = new List<MessageContext>(contexts.Count);`
`55`	`56`	`var retriedMessages = new List<MessageContext>(contexts.Count);`
`@@ -67,7 +68,7 @@ public async Task Ingest(List<MessageContext> contexts)`
`67`	`68`	`}`
`68`	`69`
`69`	`70`
`70`		`- var storedFailed = await PersistFailedMessages(failedMessages, retriedMessages);`
	`71`	`+ var storedFailed = await PersistFailedMessages(failedMessages, retriedMessages, cancellationToken);`
`71`	`72`
`72`	`73`	`try`
`73`	`74`	`{`
`@@ -89,7 +90,7 @@ public async Task Ingest(List<MessageContext> contexts)`
`89`	`90`	`{`
`90`	`91`	`Logger.Debug($"Forwarding {storedFailed.Count} messages");`
`91`	`92`	`}`
`92`		`- await Forward(storedFailed);`
	`93`	`+ await Forward(storedFailed, cancellationToken);`
`93`	`94`	`if (Logger.IsDebugEnabled)`
`94`	`95`	`{`
`95`	`96`	`Logger.Debug("Forwarded messages");`
`@@ -113,7 +114,7 @@ public async Task Ingest(List<MessageContext> contexts)`
`113`	`114`	`}`
`114`	`115`	`}`
`115`	`116`
`116`		`- async Task<IReadOnlyList<MessageContext>> PersistFailedMessages(List<MessageContext> failedMessageContexts, List<MessageContext> retriedMessageContexts)`
	`117`	`+ async Task<IReadOnlyList<MessageContext>> PersistFailedMessages(List<MessageContext> failedMessageContexts, List<MessageContext> retriedMessageContexts, CancellationToken cancellationToken)`
`117`	`118`	`{`
`118`	`119`	`var stopwatch = Stopwatch.StartNew();`
`119`	`120`
`@@ -130,7 +131,7 @@ async Task<IReadOnlyList<MessageContext>> PersistFailedMessages(List<MessageCont`
`130`	`131`
`131`	`132`	`using (bulkInsertDurationMeter.Measure())`
`132`	`133`	`{`
`133`		`- await unitOfWork.Complete();`
	`134`	`+ await unitOfWork.Complete(cancellationToken);`
`134`	`135`	`}`
`135`	`136`	`return storedFailedMessageContexts;`
`136`	`137`	`}`
`@@ -154,7 +155,7 @@ async Task<IReadOnlyList<MessageContext>> PersistFailedMessages(List<MessageCont`
`154`	`155`	`}`
`155`	`156`	`}`
`156`	`157`
`157`		`- Task Forward(IReadOnlyCollection<MessageContext> messageContexts)`
	`158`	`+ Task Forward(IReadOnlyCollection<MessageContext> messageContexts, CancellationToken cancellationToken)`
`158`	`159`	`{`
`159`	`160`	`var transportOperations = new TransportOperation[messageContexts.Count]; //We could allocate based on the actual number of ProcessedMessages but this should be OK`
`160`	`161`	`var index = 0;`
`@@ -177,11 +178,11 @@ Task Forward(IReadOnlyCollection<MessageContext> messageContexts)`
`177`	`178`	`return anyContext != null`
`178`	`179`	`? messageDispatcher.Value.Dispatch(`
`179`	`180`	`new TransportOperations(transportOperations),`
`180`		`- anyContext.TransportTransaction)`
	`181`	`+ anyContext.TransportTransaction, cancellationToken)`
`181`	`182`	`: Task.CompletedTask;`
`182`	`183`	`}`
`183`	`184`
`184`		`- public async Task VerifyCanReachForwardingAddress()`
	`185`	`+ public async Task VerifyCanReachForwardingAddress(CancellationToken cancellationToken)`
`185`	`186`	`{`
`186`	`187`	`try`
`187`	`188`	`{`
`@@ -193,7 +194,7 @@ public async Task VerifyCanReachForwardingAddress()`
`193`	`194`	`)`
`194`	`195`	`);`
`195`	`196`
`196`		`- await messageDispatcher.Value.Dispatch(transportOperations, new TransportTransaction());`
	`197`	`+ await messageDispatcher.Value.Dispatch(transportOperations, new TransportTransaction(), cancellationToken);`
`197`	`198`	`}`
`198`	`199`	`catch (Exception e)`
`199`	`200`	`{`