Propagate cancellation tokens through External Integration Request

src/ServiceControl.Persistence.EFCore/Implementation/ExternalIntegrationRequestsDataStore.cs

@@ -5,11 +5,11 @@ namespace ServiceControl.Persistence.EFCore.Implementation;
 
 public class ExternalIntegrationRequestsDataStore : IExternalIntegrationRequestsDataStore, IHostedService
 {
-    public void Subscribe(Func<object[], Task> callback) { }
+    public void Subscribe(Func<object[], CancellationToken, Task> callback) { }
 
-    public Task StoreDispatchRequest(IEnumerable<ExternalIntegrationDispatchRequest> dispatchRequests) => Task.CompletedTask;
+    public Task StoreDispatchRequest(IEnumerable<ExternalIntegrationDispatchRequest> dispatchRequests, CancellationToken cancellationToken = default) => Task.CompletedTask;
 
-    public Task StartAsync(CancellationToken cancellationToken) => Task.CompletedTask;
+    public Task StartAsync(CancellationToken cancellationToken = default) => Task.CompletedTask;
 
-    public Task StopAsync(CancellationToken cancellationToken) => Task.CompletedTask;
+    public Task StopAsync(CancellationToken cancellationToken = default) => Task.CompletedTask;
 }

src/ServiceControl.Persistence.RavenDB/ExternalIntegrationRequestsDataStore.cs

@@ -45,9 +45,9 @@ public ExternalIntegrationRequestsDataStore(
 
         const string KeyPrefix = "ExternalIntegrationDispatchRequests";
 
-        public async Task StoreDispatchRequest(IEnumerable<ExternalIntegrationDispatchRequest> dispatchRequests)
+        public async Task StoreDispatchRequest(IEnumerable<ExternalIntegrationDispatchRequest> dispatchRequests, CancellationToken cancellationToken = default)
         {
-            using var session = await sessionProvider.OpenSession();
+            using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken);
             foreach (var dispatchRequest in dispatchRequests)
             {
                 if (dispatchRequest.Id != null)
@@ -56,13 +56,13 @@ public async Task StoreDispatchRequest(IEnumerable<ExternalIntegrationDispatchRe
                 }
 
                 dispatchRequest.Id = KeyPrefix + "/" + Guid.NewGuid();
-                await session.StoreAsync(dispatchRequest);
+                await session.StoreAsync(dispatchRequest, cancellationToken);
             }
 
-            await session.SaveChangesAsync();
+            await session.SaveChangesAsync(cancellationToken);
         }
 
-        public void Subscribe(Func<object[], Task> callback)
+        public void Subscribe(Func<object[], CancellationToken, Task> callback)
         {
             if (this.callback != null)
             {
@@ -119,7 +119,7 @@ async Task DispatchEvents(CancellationToken cancellationToken)
 
             do
             {
-                more = await TryDispatchEventBatch();
+                more = await TryDispatchEventBatch(cancellationToken);
 
                 circuitBreaker.Success();
 
@@ -132,14 +132,14 @@ async Task DispatchEvents(CancellationToken cancellationToken)
             while (!cancellationToken.IsCancellationRequested && more);
         }
 
-        async Task<bool> TryDispatchEventBatch()
+        async Task<bool> TryDispatchEventBatch(CancellationToken cancellationToken)
         {
-            using var session = await sessionProvider.OpenSession();
+            using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken);
             var awaitingDispatching = await session
                 .Query<ExternalIntegrationDispatchRequest>()
                 .Statistics(out var stats)
                 .Take(settings.ExternalIntegrationsDispatchingBatchSize)
-                .ToListAsync();
+                .ToListAsync(cancellationToken);
 
             if (awaitingDispatching.Count == 0)
             {
@@ -151,19 +151,19 @@ async Task<bool> TryDispatchEventBatch()
             var allContexts = awaitingDispatching.Select(r => r.DispatchContext).ToArray();
             logger.LogDebug("Dispatching {EventCount} events", allContexts.Length);
 
-            await callback(allContexts);
+            await callback(allContexts, cancellationToken);
 
             foreach (var dispatchedEvent in awaitingDispatching)
             {
                 session.Delete(dispatchedEvent);
             }
 
-            await session.SaveChangesAsync();
+            await session.SaveChangesAsync(cancellationToken);
 
             return true;
         }
 
-        public async Task StartAsync(CancellationToken cancellationToken)
+        public async Task StartAsync(CancellationToken cancellationToken = default)
         {
             var documentStore = await documentStoreProvider.GetDocumentStore(cancellationToken);
             subscription = documentStore
@@ -176,7 +176,7 @@ public async Task StartAsync(CancellationToken cancellationToken)
                 });
         }
 
-        public async Task StopAsync(CancellationToken cancellationToken) => await DisposeAsync();
+        public async Task StopAsync(CancellationToken cancellationToken = default) => await DisposeAsync();
 
         public async ValueTask DisposeAsync()
         {
@@ -206,7 +206,7 @@ public async ValueTask DisposeAsync()
         IDisposable subscription;
         Task task;
         ManualResetEventSlim signal = new();
-        Func<object[], Task> callback;
+        Func<object[], CancellationToken, Task> callback;
         bool isDisposed;
 
         readonly ILogger<ExternalIntegrationRequestsDataStore> logger;

src/ServiceControl.Persistence/IExternalIntegrationRequestsDataStore.cs

@@ -8,8 +8,8 @@
 
     public interface IExternalIntegrationRequestsDataStore
     {
-        void Subscribe(Func<object[], Task> callback);
-        Task StoreDispatchRequest(IEnumerable<ExternalIntegrationDispatchRequest> dispatchRequests);
-        Task StopAsync(CancellationToken cancellationToken);
+        void Subscribe(Func<object[], CancellationToken, Task> callback);
+        Task StoreDispatchRequest(IEnumerable<ExternalIntegrationDispatchRequest> dispatchRequests, CancellationToken cancellationToken = default);
+        Task StopAsync(CancellationToken cancellationToken = default);
     }
 }

src/ServiceControl/.editorconfig

@@ -9,8 +9,3 @@ dotnet_diagnostic.CA2007.severity = none
 [{Operations/ErrorIngestion.cs,Operations/ErrorIngestionFaultPolicy.cs,Operations/ImportFailureCircuitBreaker.cs}]
 dotnet_diagnostic.PS0013.severity = none
 dotnet_diagnostic.PS0018.severity = none
-
-# Blocked on IExternalIntegrationRequestsDataStore.Subscribe, which takes Func<object[], Task>.
-# The dispatch callback gains a token when that interface does, in the persistence phase.
-[ExternalIntegrations/EventDispatcherHostedService.cs]
-dotnet_diagnostic.PS0018.severity = none

src/ServiceControl/ExternalIntegrations/EventDispatcherHostedService.cs

@@ -33,12 +33,12 @@ public Task StartAsync(CancellationToken cancellationToken = default)
             return Task.CompletedTask;
         }
 
-        async Task TryDispatchEventBatch(object[] allContexts)
+        async Task TryDispatchEventBatch(object[] allContexts, CancellationToken cancellationToken)
         {
             var eventsToBePublished = new List<object>();
             foreach (var publisher in eventPublishers)
             {
-                var events = await publisher.PublishEventsForOwnContexts(allContexts);
+                var events = await publisher.PublishEventsForOwnContexts(allContexts, cancellationToken);
                 eventsToBePublished.AddRange(events);
             }
 
@@ -48,7 +48,11 @@ async Task TryDispatchEventBatch(object[] allContexts)
 
                 try
                 {
-                    await messageSession.Publish(eventToBePublished);
+                    await messageSession.Publish(eventToBePublished, cancellationToken);
+                }
+                catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
+                {
+                    throw;
                 }
                 catch (Exception e)
                 {
@@ -67,7 +71,7 @@ async Task TryDispatchEventBatch(object[] allContexts)
                         m.Reason = "Failed to retrieve reason!";
                     }
 
-                    await domainEvents.Raise(m);
+                    await domainEvents.Raise(m, cancellationToken);
                 }
             }
         }

src/ServiceControl/ExternalIntegrations/IntegrationEventWriter.cs

@@ -39,7 +39,7 @@ public async Task Handle(IDomainEvent message, CancellationToken cancellationTok
                 DispatchContext = dispatchContext
             }).ToList();
 
-            await store.StoreDispatchRequest(dispatchRequests);
+            await store.StoreDispatchRequest(dispatchRequests, cancellationToken);
         }
 
         readonly IExternalIntegrationRequestsDataStore store;