From 9d5088d879930188b07dc3763e53975a13d9c7cc Mon Sep 17 00:00:00 2001 From: John Simons Date: Tue, 11 Aug 2026 14:43:21 +1000 Subject: [PATCH] Propagate cancellation tokens through External Integration Request data store This change continues the effort to consistently apply optional `CancellationToken` parameters to asynchronous methods within the `IExternalIntegrationRequestsDataStore` interface, its EFCore and RavenDB implementations, and related dispatching services. Cancellation tokens are now propagated to underlying asynchronous database calls and consuming services. This ensures proper cancellation of long-running operations and further reduces cancellation analyzer debt. --- .../ExternalIntegrationRequestsDataStore.cs | 8 +++--- .../ExternalIntegrationRequestsDataStore.cs | 28 +++++++++---------- .../IExternalIntegrationRequestsDataStore.cs | 6 ++-- src/ServiceControl/.editorconfig | 5 ---- .../EventDispatcherHostedService.cs | 12 +++++--- .../IntegrationEventWriter.cs | 2 +- 6 files changed, 30 insertions(+), 31 deletions(-) diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/ExternalIntegrationRequestsDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/ExternalIntegrationRequestsDataStore.cs index dcbb12b415..2387c15fbb 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/ExternalIntegrationRequestsDataStore.cs +++ b/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 callback) { } + public void Subscribe(Func callback) { } - public Task StoreDispatchRequest(IEnumerable dispatchRequests) => Task.CompletedTask; + public Task StoreDispatchRequest(IEnumerable 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; } diff --git a/src/ServiceControl.Persistence.RavenDB/ExternalIntegrationRequestsDataStore.cs b/src/ServiceControl.Persistence.RavenDB/ExternalIntegrationRequestsDataStore.cs index 7b339732ad..9b928db0ee 100644 --- a/src/ServiceControl.Persistence.RavenDB/ExternalIntegrationRequestsDataStore.cs +++ b/src/ServiceControl.Persistence.RavenDB/ExternalIntegrationRequestsDataStore.cs @@ -45,9 +45,9 @@ public ExternalIntegrationRequestsDataStore( const string KeyPrefix = "ExternalIntegrationDispatchRequests"; - public async Task StoreDispatchRequest(IEnumerable dispatchRequests) + public async Task StoreDispatchRequest(IEnumerable 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 callback) + public void Subscribe(Func 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 TryDispatchEventBatch() + async Task TryDispatchEventBatch(CancellationToken cancellationToken) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var awaitingDispatching = await session .Query() .Statistics(out var stats) .Take(settings.ExternalIntegrationsDispatchingBatchSize) - .ToListAsync(); + .ToListAsync(cancellationToken); if (awaitingDispatching.Count == 0) { @@ -151,19 +151,19 @@ async Task 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 callback; + Func callback; bool isDisposed; readonly ILogger logger; diff --git a/src/ServiceControl.Persistence/IExternalIntegrationRequestsDataStore.cs b/src/ServiceControl.Persistence/IExternalIntegrationRequestsDataStore.cs index d1da154989..4b43e02a29 100644 --- a/src/ServiceControl.Persistence/IExternalIntegrationRequestsDataStore.cs +++ b/src/ServiceControl.Persistence/IExternalIntegrationRequestsDataStore.cs @@ -8,8 +8,8 @@ public interface IExternalIntegrationRequestsDataStore { - void Subscribe(Func callback); - Task StoreDispatchRequest(IEnumerable dispatchRequests); - Task StopAsync(CancellationToken cancellationToken); + void Subscribe(Func callback); + Task StoreDispatchRequest(IEnumerable dispatchRequests, CancellationToken cancellationToken = default); + Task StopAsync(CancellationToken cancellationToken = default); } } \ No newline at end of file diff --git a/src/ServiceControl/.editorconfig b/src/ServiceControl/.editorconfig index 3668c8ea04..0f304aa060 100644 --- a/src/ServiceControl/.editorconfig +++ b/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. -# The dispatch callback gains a token when that interface does, in the persistence phase. -[ExternalIntegrations/EventDispatcherHostedService.cs] -dotnet_diagnostic.PS0018.severity = none diff --git a/src/ServiceControl/ExternalIntegrations/EventDispatcherHostedService.cs b/src/ServiceControl/ExternalIntegrations/EventDispatcherHostedService.cs index 26b9d9c62d..3ed9c183bf 100644 --- a/src/ServiceControl/ExternalIntegrations/EventDispatcherHostedService.cs +++ b/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(); 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); } } } diff --git a/src/ServiceControl/ExternalIntegrations/IntegrationEventWriter.cs b/src/ServiceControl/ExternalIntegrations/IntegrationEventWriter.cs index 44e95ac6ea..d09eb5af2b 100644 --- a/src/ServiceControl/ExternalIntegrations/IntegrationEventWriter.cs +++ b/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;