From 1bcda0c6cbf76d38952c3f0e9b2f62aade8a8a80 Mon Sep 17 00:00:00 2001 From: John Simons Date: Wed, 12 Aug 2026 07:23:18 +1000 Subject: [PATCH 1/6] Propagate cancellation tokens through monitoring data store persistence This change continues the effort to consistently apply optional `CancellationToken` parameters to asynchronous methods within the `IMonitoringDataStore` interface, its EFCore and RavenDB implementations, and the controllers, handlers, and services that consume them. Cancellation tokens are now propagated to underlying asynchronous database calls, ensuring proper cancellation of long-running operations and further reducing cancellation analyzer debt. --- .../Implementation/MonitoringDataStore.cs | 29 ++++++------ .../RavenMonitoringDataStore.cs | 47 ++++++++++--------- .../IMonitoringDataStore.cs | 13 ++--- ...tEndpointSettingsSyncHostedServiceTests.cs | 12 ++--- ...rtbeatEndpointSettingsSyncHostedService.cs | 4 +- .../HeartbeatMonitoringHostedService.cs | 2 +- .../Monitoring/MonitoringDataPersister.cs | 8 ++-- .../Web/EndpointsMonitoringController.cs | 2 +- 8 files changed, 60 insertions(+), 57 deletions(-) diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/MonitoringDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/MonitoringDataStore.cs index 38f458422d..37f07b08f5 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/MonitoringDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/MonitoringDataStore.cs @@ -7,12 +7,12 @@ namespace ServiceControl.Persistence.EFCore.Implementation; public class MonitoringDataStore(IServiceScopeFactory scopeFactory) : DataStoreBase(scopeFactory), IMonitoringDataStore { - public Task CreateIfNotExists(EndpointDetails endpoint) => + public Task CreateIfNotExists(EndpointDetails endpoint, CancellationToken cancellationToken = default) => ExecuteWithDbContext(async dbContext => { var id = endpoint.GetDeterministicId(); - var exists = await dbContext.KnownEndpoints.AnyAsync(e => e.Id == id); + var exists = await dbContext.KnownEndpoints.AnyAsync(e => e.Id == id, cancellationToken); if (exists) { return; @@ -31,21 +31,21 @@ public Task CreateIfNotExists(EndpointDetails endpoint) => try { - await dbContext.SaveChangesAsync(); + await dbContext.SaveChangesAsync(cancellationToken); } catch (DbUpdateException) { // A concurrent insert with the same deterministic id may have won the race. // Anything else is a genuine failure. dbContext.Entry(knownEndpoint).State = EntityState.Detached; - if (!await dbContext.KnownEndpoints.AnyAsync(e => e.Id == id)) + if (!await dbContext.KnownEndpoints.AnyAsync(e => e.Id == id, cancellationToken)) { throw; } } }); - public Task CreateOrUpdate(EndpointDetails endpoint, IEndpointInstanceMonitoring endpointInstanceMonitoring) => + public Task CreateOrUpdate(EndpointDetails endpoint, IEndpointInstanceMonitoring endpointInstanceMonitoring, CancellationToken cancellationToken = default) => ExecuteWithDbContext(async dbContext => { var id = endpoint.GetDeterministicId(); @@ -61,24 +61,25 @@ await dbContext.UpsertAsync([id], knownEndpoint => { knownEndpoint.Monitored = endpointInstanceMonitoring.IsMonitored(id); - } + }, + cancellationToken ); }); - public Task UpdateEndpointMonitoring(EndpointDetails endpoint, bool isMonitored) => + public Task UpdateEndpointMonitoring(EndpointDetails endpoint, bool isMonitored, CancellationToken cancellationToken = default) => ExecuteWithDbContext(async dbContext => { var id = endpoint.GetDeterministicId(); await dbContext.KnownEndpoints .Where(e => e.Id == id) - .ExecuteUpdateAsync(setters => setters.SetProperty(e => e.Monitored, isMonitored)); + .ExecuteUpdateAsync(setters => setters.SetProperty(e => e.Monitored, isMonitored), cancellationToken); }); - public Task WarmupMonitoringFromPersistence(IEndpointInstanceMonitoring endpointInstanceMonitoring) => + public Task WarmupMonitoringFromPersistence(IEndpointInstanceMonitoring endpointInstanceMonitoring, CancellationToken cancellationToken = default) => ExecuteWithDbContext(async dbContext => { - await foreach (var endpoint in dbContext.KnownEndpoints.AsNoTracking().AsAsyncEnumerable()) + await foreach (var endpoint in dbContext.KnownEndpoints.AsNoTracking().AsAsyncEnumerable().WithCancellation(cancellationToken)) { var endpointDetails = new EndpointDetails { @@ -91,17 +92,17 @@ public Task WarmupMonitoringFromPersistence(IEndpointInstanceMonitoring endpoint } }); - public Task Delete(Guid endpointId) + public Task Delete(Guid endpointId, CancellationToken cancellationToken = default) { return ExecuteWithDbContext(async dbContext => { await dbContext.KnownEndpoints .Where(e => e.Id == endpointId) - .ExecuteDeleteAsync(); + .ExecuteDeleteAsync(cancellationToken); }); } - public Task> GetAllKnownEndpoints() + public Task> GetAllKnownEndpoints(CancellationToken cancellationToken = default) { return ExecuteWithDbContext>(async dbContext => await dbContext.KnownEndpoints @@ -117,6 +118,6 @@ await dbContext.KnownEndpoints HostDisplayName = e.Host, Monitored = e.Monitored }) - .ToListAsync()); + .ToListAsync(cancellationToken)); } } diff --git a/src/ServiceControl.Persistence.RavenDB/RavenMonitoringDataStore.cs b/src/ServiceControl.Persistence.RavenDB/RavenMonitoringDataStore.cs index 333da3dabb..da0b782736 100644 --- a/src/ServiceControl.Persistence.RavenDB/RavenMonitoringDataStore.cs +++ b/src/ServiceControl.Persistence.RavenDB/RavenMonitoringDataStore.cs @@ -2,6 +2,7 @@ { using System; using System.Collections.Generic; + using System.Threading; using System.Threading.Tasks; using Raven.Client.Documents; using ServiceControl.Operations; @@ -11,14 +12,14 @@ class RavenMonitoringDataStore(IRavenSessionProvider sessionProvider) : IMonitor { public static string MakeDocumentId(Guid id) => $"{KnownEndpointsCollectionName}/{id}"; - public async Task CreateIfNotExists(EndpointDetails endpoint) + public async Task CreateIfNotExists(EndpointDetails endpoint, CancellationToken cancellationToken = default) { var id = endpoint.GetDeterministicId(); var docId = MakeDocumentId(id); - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); - var knownEndpoint = await session.LoadAsync(docId); + var knownEndpoint = await session.LoadAsync(docId, cancellationToken); if (knownEndpoint != null) { @@ -32,19 +33,19 @@ public async Task CreateIfNotExists(EndpointDetails endpoint) Monitored = false }; - await session.StoreAsync(knownEndpoint, docId); + await session.StoreAsync(knownEndpoint, docId, cancellationToken); - await session.SaveChangesAsync(); + await session.SaveChangesAsync(cancellationToken); } - public async Task CreateOrUpdate(EndpointDetails endpoint, IEndpointInstanceMonitoring endpointInstanceMonitoring) + public async Task CreateOrUpdate(EndpointDetails endpoint, IEndpointInstanceMonitoring endpointInstanceMonitoring, CancellationToken cancellationToken = default) { var id = endpoint.GetDeterministicId(); var docId = MakeDocumentId(id); - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); - var knownEndpoint = await session.LoadAsync(docId); + var knownEndpoint = await session.LoadAsync(docId, cancellationToken); if (knownEndpoint == null) { @@ -55,37 +56,37 @@ public async Task CreateOrUpdate(EndpointDetails endpoint, IEndpointInstanceMoni Monitored = true }; - await session.StoreAsync(knownEndpoint, docId); + await session.StoreAsync(knownEndpoint, docId, cancellationToken); } else { knownEndpoint.Monitored = endpointInstanceMonitoring.IsMonitored(id); } - await session.SaveChangesAsync(); + await session.SaveChangesAsync(cancellationToken); } - public async Task UpdateEndpointMonitoring(EndpointDetails endpoint, bool isMonitored) + public async Task UpdateEndpointMonitoring(EndpointDetails endpoint, bool isMonitored, CancellationToken cancellationToken = default) { var id = endpoint.GetDeterministicId(); var docId = MakeDocumentId(id); - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); - var knownEndpoint = await session.LoadAsync(docId); + var knownEndpoint = await session.LoadAsync(docId, cancellationToken); if (knownEndpoint != null) { knownEndpoint.Monitored = isMonitored; - await session.SaveChangesAsync(); + await session.SaveChangesAsync(cancellationToken); } } - public async Task WarmupMonitoringFromPersistence(IEndpointInstanceMonitoring endpointInstanceMonitoring) + public async Task WarmupMonitoringFromPersistence(IEndpointInstanceMonitoring endpointInstanceMonitoring, CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); - await using var endpointsEnumerator = await session.Advanced.StreamAsync(session.Query()); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); + await using var endpointsEnumerator = await session.Advanced.StreamAsync(session.Query(), cancellationToken); while (await endpointsEnumerator.MoveNextAsync()) { @@ -95,19 +96,19 @@ public async Task WarmupMonitoringFromPersistence(IEndpointInstanceMonitoring en } } - public async Task Delete(Guid endpointId) + public async Task Delete(Guid endpointId, CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); session.Delete(MakeDocumentId(endpointId)); - await session.SaveChangesAsync(); + await session.SaveChangesAsync(cancellationToken); } - public async Task> GetAllKnownEndpoints() + public async Task> GetAllKnownEndpoints(CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var knownEndpoints = await session.Query() - .ToListAsync(); + .ToListAsync(cancellationToken); return knownEndpoints; } diff --git a/src/ServiceControl.Persistence/IMonitoringDataStore.cs b/src/ServiceControl.Persistence/IMonitoringDataStore.cs index df763ecf02..1ea67d542d 100644 --- a/src/ServiceControl.Persistence/IMonitoringDataStore.cs +++ b/src/ServiceControl.Persistence/IMonitoringDataStore.cs @@ -2,16 +2,17 @@ { using System; using System.Collections.Generic; + using System.Threading; using System.Threading.Tasks; using ServiceControl.Operations; public interface IMonitoringDataStore { - Task CreateIfNotExists(EndpointDetails endpoint); - Task CreateOrUpdate(EndpointDetails endpoint, IEndpointInstanceMonitoring endpointInstanceMonitoring); - Task UpdateEndpointMonitoring(EndpointDetails endpoint, bool isMonitored); - Task WarmupMonitoringFromPersistence(IEndpointInstanceMonitoring endpointInstanceMonitoring); - Task Delete(Guid endpointId); - Task> GetAllKnownEndpoints(); + Task CreateIfNotExists(EndpointDetails endpoint, CancellationToken cancellationToken = default); + Task CreateOrUpdate(EndpointDetails endpoint, IEndpointInstanceMonitoring endpointInstanceMonitoring, CancellationToken cancellationToken = default); + Task UpdateEndpointMonitoring(EndpointDetails endpoint, bool isMonitored, CancellationToken cancellationToken = default); + Task WarmupMonitoringFromPersistence(IEndpointInstanceMonitoring endpointInstanceMonitoring, CancellationToken cancellationToken = default); + Task Delete(Guid endpointId, CancellationToken cancellationToken = default); + Task> GetAllKnownEndpoints(CancellationToken cancellationToken = default); } } \ No newline at end of file diff --git a/src/ServiceControl.UnitTests/Monitoring/HeartbeatEndpointSettingsSyncHostedServiceTests.cs b/src/ServiceControl.UnitTests/Monitoring/HeartbeatEndpointSettingsSyncHostedServiceTests.cs index 4317f3ad17..6b8b1ca1c8 100644 --- a/src/ServiceControl.UnitTests/Monitoring/HeartbeatEndpointSettingsSyncHostedServiceTests.cs +++ b/src/ServiceControl.UnitTests/Monitoring/HeartbeatEndpointSettingsSyncHostedServiceTests.cs @@ -263,18 +263,18 @@ public Task Delete(string name, CancellationToken cancellationToken) class MockMonitoringDataStore(KnownEndpoint[] knownEndpoints) : IMonitoringDataStore { - public Task CreateIfNotExists(EndpointDetails endpoint) => throw new NotImplementedException(); + public Task CreateIfNotExists(EndpointDetails endpoint, CancellationToken cancellationToken = default) => throw new NotImplementedException(); - public Task CreateOrUpdate(EndpointDetails endpoint, IEndpointInstanceMonitoring endpointInstanceMonitoring) => + public Task CreateOrUpdate(EndpointDetails endpoint, IEndpointInstanceMonitoring endpointInstanceMonitoring, CancellationToken cancellationToken = default) => throw new NotImplementedException(); - public Task UpdateEndpointMonitoring(EndpointDetails endpoint, bool isMonitored) => + public Task UpdateEndpointMonitoring(EndpointDetails endpoint, bool isMonitored, CancellationToken cancellationToken = default) => throw new NotImplementedException(); - public Task WarmupMonitoringFromPersistence(IEndpointInstanceMonitoring endpointInstanceMonitoring) => + public Task WarmupMonitoringFromPersistence(IEndpointInstanceMonitoring endpointInstanceMonitoring, CancellationToken cancellationToken = default) => throw new NotImplementedException(); - public Task Delete(Guid endpointId) + public Task Delete(Guid endpointId, CancellationToken cancellationToken = default) { Deleted.Add(endpointId); return Task.CompletedTask; @@ -282,7 +282,7 @@ public Task Delete(Guid endpointId) public List Deleted { get; } = []; - public Task> GetAllKnownEndpoints() => + public Task> GetAllKnownEndpoints(CancellationToken cancellationToken = default) => Task.FromResult>(knownEndpoints); } } \ No newline at end of file diff --git a/src/ServiceControl/Monitoring/HeartbeatEndpointSettingsSyncHostedService.cs b/src/ServiceControl/Monitoring/HeartbeatEndpointSettingsSyncHostedService.cs index d866c78b91..48988fe30a 100644 --- a/src/ServiceControl/Monitoring/HeartbeatEndpointSettingsSyncHostedService.cs +++ b/src/ServiceControl/Monitoring/HeartbeatEndpointSettingsSyncHostedService.cs @@ -57,7 +57,7 @@ protected override async Task ExecuteAsync(CancellationToken cancellationToken = async Task PerformSync(CancellationToken cancellationToken) { - var monitorEndpoints = (await monitoringDataStore.GetAllKnownEndpoints()) + var monitorEndpoints = (await monitoringDataStore.GetAllKnownEndpoints(cancellationToken)) .Select(endpoint => endpoint.EndpointDetails.Name).Distinct().ToHashSet(); await InitialiseSettings(monitorEndpoints, cancellationToken); @@ -82,7 +82,7 @@ async Task PurgeMonitoringDataThatDoesNotNeedToBeTracked(CancellationToken cance foreach (Guid endpointId in monitorEndpointsLookup[endpointSetting.Name].SkipLast(1)) { endpointInstanceMonitoring.RemoveEndpoint(endpointId); - await monitoringDataStore.Delete(endpointId); + await monitoringDataStore.Delete(endpointId, cancellationToken); logger.LogInformation("Removed endpoint '{EndpointName}' from monitoring data", endpointSetting.Name); } } diff --git a/src/ServiceControl/Monitoring/HeartbeatMonitoringHostedService.cs b/src/ServiceControl/Monitoring/HeartbeatMonitoringHostedService.cs index 2a93f67c96..00e57af2ce 100644 --- a/src/ServiceControl/Monitoring/HeartbeatMonitoringHostedService.cs +++ b/src/ServiceControl/Monitoring/HeartbeatMonitoringHostedService.cs @@ -21,7 +21,7 @@ public HeartbeatMonitoringHostedService(IEndpointInstanceMonitoring monitor, IMo } public async Task StartAsync(CancellationToken cancellationToken = default) { - await persistence.WarmupMonitoringFromPersistence(monitor); + await persistence.WarmupMonitoringFromPersistence(monitor, cancellationToken); timer = scheduler.Schedule(CheckEndpoints, TimeSpan.Zero, TimeSpan.FromSeconds(5), e => logger.LogError(e, "Exception occurred when monitoring endpoint instances")); } diff --git a/src/ServiceControl/Monitoring/MonitoringDataPersister.cs b/src/ServiceControl/Monitoring/MonitoringDataPersister.cs index 46248bb43e..6d2ce97293 100644 --- a/src/ServiceControl/Monitoring/MonitoringDataPersister.cs +++ b/src/ServiceControl/Monitoring/MonitoringDataPersister.cs @@ -22,22 +22,22 @@ public MonitoringDataPersister(IMonitoringDataStore monitoringDataStore, IEndpoi public Task Handle(EndpointDetected domainEvent, CancellationToken cancellationToken = default) { - return _monitoringDataStore.CreateIfNotExists(domainEvent.Endpoint); + return _monitoringDataStore.CreateIfNotExists(domainEvent.Endpoint, cancellationToken); } public Task Handle(HeartbeatingEndpointDetected domainEvent, CancellationToken cancellationToken = default) { - return _monitoringDataStore.CreateOrUpdate(domainEvent.Endpoint, _endpointInstanceMonitoring); + return _monitoringDataStore.CreateOrUpdate(domainEvent.Endpoint, _endpointInstanceMonitoring, cancellationToken); } public Task Handle(MonitoringEnabledForEndpoint domainEvent, CancellationToken cancellationToken = default) { - return _monitoringDataStore.UpdateEndpointMonitoring(domainEvent.Endpoint, true); + return _monitoringDataStore.UpdateEndpointMonitoring(domainEvent.Endpoint, true, cancellationToken); } public Task Handle(MonitoringDisabledForEndpoint domainEvent, CancellationToken cancellationToken = default) { - return _monitoringDataStore.UpdateEndpointMonitoring(domainEvent.Endpoint, false); + return _monitoringDataStore.UpdateEndpointMonitoring(domainEvent.Endpoint, false, cancellationToken); } IMonitoringDataStore _monitoringDataStore; diff --git a/src/ServiceControl/Monitoring/Web/EndpointsMonitoringController.cs b/src/ServiceControl/Monitoring/Web/EndpointsMonitoringController.cs index 7a6cf742a9..8040bac1c9 100644 --- a/src/ServiceControl/Monitoring/Web/EndpointsMonitoringController.cs +++ b/src/ServiceControl/Monitoring/Web/EndpointsMonitoringController.cs @@ -55,7 +55,7 @@ public async Task DeleteEndpoint(Guid endpointId, CancellationTok return NotFound(); } - await dataStore.Delete(endpointId); + await dataStore.Delete(endpointId, cancellationToken); monitoring.RemoveEndpoint(endpointId); return NoContent(); From 98624dbe6214fba1c2c63600e7ddae14936d58d2 Mon Sep 17 00:00:00 2001 From: John Simons Date: Wed, 12 Aug 2026 08:13:15 +1000 Subject: [PATCH 2/6] Propagate cancellation tokens through endpoint settings, subscriptions and trial license persistence This change continues the effort to consistently apply optional `CancellationToken` parameters to asynchronous methods within the `IEndpointSettingsStore`, `IServiceControlSubscriptionStorage`, and `ITrialLicenseDataProvider` interfaces, along with their EFCore and RavenDB implementations. Cancellation tokens are now propagated to underlying asynchronous database calls, ensuring proper cancellation of long-running operations and further reducing cancellation analyzer debt. --- .../Implementation/EndpointSettingsStore.cs | 8 ++-- .../Implementation/SubscriptionStorage.cs | 2 +- .../TrialLicenseDataProvider.cs | 4 +- .../EndpointSettingsStore.cs | 6 +-- .../Subscriptions/RavenSubscriptionStorage.cs | 46 ++++++++++--------- .../TrialLicenseDataProvider.cs | 4 +- .../IEndpointSettingsStore.cs | 4 +- .../IServiceControlSubscriptionStorage.cs | 3 +- .../ITrialLicenseDataProvider.cs | 4 +- .../Subscriptions/SubscriptionStorage.cs | 2 +- 10 files changed, 43 insertions(+), 40 deletions(-) diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/EndpointSettingsStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/EndpointSettingsStore.cs index 93084037fe..d709c5a2d7 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/EndpointSettingsStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/EndpointSettingsStore.cs @@ -6,18 +6,18 @@ namespace ServiceControl.Persistence.EFCore.Implementation; public class EndpointSettingsStore(IServiceScopeFactory scopeFactory) : DataStoreBase(scopeFactory), IEndpointSettingsStore { - public IAsyncEnumerable GetAllEndpointSettings(CancellationToken cancellationToken) + public IAsyncEnumerable GetAllEndpointSettings(CancellationToken cancellationToken = default) => ExecuteWithDbContext(context => context.EndpointSettings.Select(row => new EndpointSettings { Name = row.Name, TrackInstances = row.TrackInstances }).AsAsyncEnumerable(), cancellationToken); - public Task UpdateEndpointSettings(EndpointSettings settings, CancellationToken token) => + public Task UpdateEndpointSettings(EndpointSettings settings, CancellationToken cancellationToken = default) => ExecuteWithDbContext(async context => await context.UpsertAsync([settings.Name], () => new EndpointSettingsEntity() { Name = settings.Name, TrackInstances = settings.TrackInstances }, entity => entity.TrackInstances = settings.TrackInstances, - token + cancellationToken )); - public Task Delete(string name, CancellationToken cancellationToken) => + public Task Delete(string name, CancellationToken cancellationToken = default) => ExecuteWithDbContext(context => context.EndpointSettings.Where(x => x.Name == name) .ExecuteDeleteAsync(cancellationToken)); } \ No newline at end of file diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/SubscriptionStorage.cs b/src/ServiceControl.Persistence.EFCore/Implementation/SubscriptionStorage.cs index e910c4d8c0..337e3fe53d 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/SubscriptionStorage.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/SubscriptionStorage.cs @@ -30,7 +30,7 @@ public SubscriptionStorage(IServiceScopeFactory scopeFactory, string endpointNam } // Subscriptions are read on demand, so there is nothing to prime at startup. - public Task Initialize() => Task.CompletedTask; + public Task Initialize(CancellationToken cancellationToken = default) => Task.CompletedTask; public async Task Subscribe(Subscriber subscriber, MessageType messageType, ContextBag context, CancellationToken cancellationToken = default) { diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/TrialLicenseDataProvider.cs b/src/ServiceControl.Persistence.EFCore/Implementation/TrialLicenseDataProvider.cs index 46aa4f4ee6..5e45712e44 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/TrialLicenseDataProvider.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/TrialLicenseDataProvider.cs @@ -5,11 +5,11 @@ namespace ServiceControl.Persistence.EFCore.Implementation; public class TrialLicenseDataProvider(IServiceScopeFactory scopeFactory) : DataStoreBase(scopeFactory), ITrialLicenseDataProvider { - public Task GetTrialEndDate(CancellationToken cancellationToken) + public Task GetTrialEndDate(CancellationToken cancellationToken = default) => ExecuteWithDbContext(context => context.GetSetting(SettingKeys.TrialEndDate, cancellationToken)); - public Task StoreTrialEndDate(DateOnly trialEndDate, CancellationToken cancellationToken) + public Task StoreTrialEndDate(DateOnly trialEndDate, CancellationToken cancellationToken = default) => ExecuteWithDbContext(context => context.StoreSetting(SettingKeys.TrialEndDate, trialEndDate, cancellationToken)); } diff --git a/src/ServiceControl.Persistence.RavenDB/EndpointSettingsStore.cs b/src/ServiceControl.Persistence.RavenDB/EndpointSettingsStore.cs index 455736495a..0792bcd8ff 100644 --- a/src/ServiceControl.Persistence.RavenDB/EndpointSettingsStore.cs +++ b/src/ServiceControl.Persistence.RavenDB/EndpointSettingsStore.cs @@ -13,7 +13,7 @@ class EndpointSettingsStore(IRavenSessionProvider sessionProvider) : IEndpointSe static string MakeDocumentId(string name) => $"{CollectionName}/{DeterministicGuid.MakeId(name)}"; - public async IAsyncEnumerable GetAllEndpointSettings([EnumeratorCancellation] CancellationToken cancellationToken) + public async IAsyncEnumerable GetAllEndpointSettings([EnumeratorCancellation] CancellationToken cancellationToken = default) { using IAsyncDocumentSession session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); await using IAsyncEnumerator> enumerator = await session @@ -26,7 +26,7 @@ public async IAsyncEnumerable GetAllEndpointSettings([Enumerat } } - public async Task Delete(string name, CancellationToken cancellationToken) + public async Task Delete(string name, CancellationToken cancellationToken = default) { string docId = MakeDocumentId(name); @@ -36,7 +36,7 @@ public async Task Delete(string name, CancellationToken cancellationToken) await session.SaveChangesAsync(cancellationToken); } - public async Task UpdateEndpointSettings(EndpointSettings settings, CancellationToken cancellationToken) + public async Task UpdateEndpointSettings(EndpointSettings settings, CancellationToken cancellationToken = default) { string docId = MakeDocumentId(settings.Name); diff --git a/src/ServiceControl.Persistence.RavenDB/Infrastructure/Subscriptions/RavenSubscriptionStorage.cs b/src/ServiceControl.Persistence.RavenDB/Infrastructure/Subscriptions/RavenSubscriptionStorage.cs index 6b8218b753..b74d41899f 100644 --- a/src/ServiceControl.Persistence.RavenDB/Infrastructure/Subscriptions/RavenSubscriptionStorage.cs +++ b/src/ServiceControl.Persistence.RavenDB/Infrastructure/Subscriptions/RavenSubscriptionStorage.cs @@ -41,15 +41,15 @@ public RavenSubscriptionStorage(IRavenSessionProvider sessionProvider, string en UpdateLookup(); } - public async Task Initialize() + public async Task Initialize(CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); - var primeSubscriptions = await LoadSubscriptions(session) ?? await MigrateSubscriptions(session, localClient); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); + var primeSubscriptions = await LoadSubscriptions(session, cancellationToken) ?? await MigrateSubscriptions(session, localClient, cancellationToken); - await SetSubscriptions(primeSubscriptions); + await SetSubscriptions(primeSubscriptions, cancellationToken); } - public async Task Subscribe(Subscriber subscriber, MessageType messageType, ContextBag context, CancellationToken cancellationToken) + public async Task Subscribe(Subscriber subscriber, MessageType messageType, ContextBag context, CancellationToken cancellationToken = default) { if (subscriber.Endpoint == localClient.Endpoint) { @@ -62,7 +62,7 @@ public async Task Subscribe(Subscriber subscriber, MessageType messageType, Cont if (AddOrUpdateSubscription(messageType, subscriber)) { - await SaveSubscriptions(); + await SaveSubscriptions(cancellationToken); } } finally @@ -71,7 +71,7 @@ public async Task Subscribe(Subscriber subscriber, MessageType messageType, Cont } } - public async Task Unsubscribe(Subscriber subscriber, MessageType messageType, ContextBag context, CancellationToken cancellationToken) + public async Task Unsubscribe(Subscriber subscriber, MessageType messageType, ContextBag context, CancellationToken cancellationToken = default) { try { @@ -89,7 +89,7 @@ public async Task Unsubscribe(Subscriber subscriber, MessageType messageType, Co if (needsSave) { - await SaveSubscriptions(); + await SaveSubscriptions(cancellationToken); } } finally @@ -98,7 +98,7 @@ public async Task Unsubscribe(Subscriber subscriber, MessageType messageType, Co } } - public Task> GetSubscriberAddressesForMessage(IEnumerable messageTypes, ContextBag context, CancellationToken cancellationToken) + public Task> GetSubscriberAddressesForMessage(IEnumerable messageTypes, ContextBag context, CancellationToken cancellationToken = default) { return Task.FromResult(messageTypes.SelectMany(x => subscriptionsLookup[x]).Distinct()); } @@ -146,12 +146,12 @@ static SubscriptionClient CreateSubscriptionClient(Subscriber subscriber) return subscriptionClient; } - async Task SaveSubscriptions() + async Task SaveSubscriptions(CancellationToken cancellationToken) { - using var session = await sessionProvider.OpenSession(); - await session.StoreAsync(subscriptions, Subscriptions.SingleDocumentId); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); + await session.StoreAsync(subscriptions, Subscriptions.SingleDocumentId, cancellationToken); UpdateLookup(); - await session.SaveChangesAsync(); + await session.SaveChangesAsync(cancellationToken); } void UpdateLookup() => @@ -180,11 +180,11 @@ string FormatId(MessageType messageType) return $"Subscriptions/{id}"; } - async Task SetSubscriptions(Subscriptions newSubscriptions) + async Task SetSubscriptions(Subscriptions newSubscriptions, CancellationToken cancellationToken) { try { - await subscriptionsLock.WaitAsync(); + await subscriptionsLock.WaitAsync(cancellationToken); subscriptions = newSubscriptions; UpdateLookup(); @@ -195,27 +195,29 @@ async Task SetSubscriptions(Subscriptions newSubscriptions) } } - static Task LoadSubscriptions(IAsyncDocumentSession session) - => session.LoadAsync(Subscriptions.SingleDocumentId); + static Task LoadSubscriptions(IAsyncDocumentSession session, CancellationToken cancellationToken) + => session.LoadAsync(Subscriptions.SingleDocumentId, cancellationToken); - async Task MigrateSubscriptions(IAsyncDocumentSession session, SubscriptionClient localClient) + async Task MigrateSubscriptions(IAsyncDocumentSession session, SubscriptionClient localClient, CancellationToken cancellationToken) { logger.LogInformation("Migrating subscriptions to new format"); var subscriptions = new Subscriptions(); - var stream = await session.Advanced.StreamAsync("Subscriptions"); + var stream = await session.Advanced.StreamAsync("Subscriptions", token: cancellationToken); while (await stream.MoveNextAsync()) { + cancellationToken.ThrowIfCancellationRequested(); + var existingSubscription = stream.Current.Document; existingSubscription.Subscribers.Remove(localClient); subscriptions.All.Add(existingSubscription.Id.Replace("Subscriptions/", string.Empty), existingSubscription); - await session.Advanced.RequestExecutor.ExecuteAsync(new DeleteDocumentCommand(stream.Current.Id, null), session.Advanced.Context); + await session.Advanced.RequestExecutor.ExecuteAsync(new DeleteDocumentCommand(stream.Current.Id, null), session.Advanced.Context, token: cancellationToken); } - await session.StoreAsync(subscriptions, Subscriptions.SingleDocumentId); - await session.SaveChangesAsync(); + await session.StoreAsync(subscriptions, Subscriptions.SingleDocumentId, cancellationToken); + await session.SaveChangesAsync(cancellationToken); return subscriptions; } diff --git a/src/ServiceControl.Persistence.RavenDB/TrialLicenseDataProvider.cs b/src/ServiceControl.Persistence.RavenDB/TrialLicenseDataProvider.cs index af9c5105bc..f0feaa2aaf 100644 --- a/src/ServiceControl.Persistence.RavenDB/TrialLicenseDataProvider.cs +++ b/src/ServiceControl.Persistence.RavenDB/TrialLicenseDataProvider.cs @@ -6,7 +6,7 @@ class TrialLicenseDataProvider(IRavenSessionProvider sessionProvider) : ITrialLicenseDataProvider { - public async Task GetTrialEndDate(CancellationToken cancellationToken) + public async Task GetTrialEndDate(CancellationToken cancellationToken = default) { using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); @@ -15,7 +15,7 @@ class TrialLicenseDataProvider(IRavenSessionProvider sessionProvider) : ITrialLi return document?.TrialEndDate; } - public async Task StoreTrialEndDate(DateOnly trialEndDate, CancellationToken cancellationToken) + public async Task StoreTrialEndDate(DateOnly trialEndDate, CancellationToken cancellationToken = default) { using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); diff --git a/src/ServiceControl.Persistence/IEndpointSettingsStore.cs b/src/ServiceControl.Persistence/IEndpointSettingsStore.cs index 2b6c52e8a1..a5b36d47c5 100644 --- a/src/ServiceControl.Persistence/IEndpointSettingsStore.cs +++ b/src/ServiceControl.Persistence/IEndpointSettingsStore.cs @@ -8,6 +8,6 @@ public interface IEndpointSettingsStore { IAsyncEnumerable GetAllEndpointSettings(CancellationToken cancellationToken = default); - Task UpdateEndpointSettings(EndpointSettings settings, CancellationToken token); - Task Delete(string name, CancellationToken cancellationToken); + Task UpdateEndpointSettings(EndpointSettings settings, CancellationToken cancellationToken = default); + Task Delete(string name, CancellationToken cancellationToken = default); } \ No newline at end of file diff --git a/src/ServiceControl.Persistence/IServiceControlSubscriptionStorage.cs b/src/ServiceControl.Persistence/IServiceControlSubscriptionStorage.cs index 09e57a38fc..abd4a6d6d8 100644 --- a/src/ServiceControl.Persistence/IServiceControlSubscriptionStorage.cs +++ b/src/ServiceControl.Persistence/IServiceControlSubscriptionStorage.cs @@ -1,10 +1,11 @@ namespace ServiceControl.Persistence { + using System.Threading; using System.Threading.Tasks; using NServiceBus.Unicast.Subscriptions.MessageDrivenSubscriptions; public interface IServiceControlSubscriptionStorage : ISubscriptionStorage { - Task Initialize(); + Task Initialize(CancellationToken cancellationToken = default); } } \ No newline at end of file diff --git a/src/ServiceControl.Persistence/ITrialLicenseDataProvider.cs b/src/ServiceControl.Persistence/ITrialLicenseDataProvider.cs index 12bf89eab4..d0dfb43628 100644 --- a/src/ServiceControl.Persistence/ITrialLicenseDataProvider.cs +++ b/src/ServiceControl.Persistence/ITrialLicenseDataProvider.cs @@ -6,7 +6,7 @@ public interface ITrialLicenseDataProvider { - Task GetTrialEndDate(CancellationToken cancellationToken); - Task StoreTrialEndDate(DateOnly trialEndDate, CancellationToken cancellationToken); + Task GetTrialEndDate(CancellationToken cancellationToken = default); + Task StoreTrialEndDate(DateOnly trialEndDate, CancellationToken cancellationToken = default); } } diff --git a/src/ServiceControl/Infrastructure/Subscriptions/SubscriptionStorage.cs b/src/ServiceControl/Infrastructure/Subscriptions/SubscriptionStorage.cs index 72117ba023..73846202f2 100644 --- a/src/ServiceControl/Infrastructure/Subscriptions/SubscriptionStorage.cs +++ b/src/ServiceControl/Infrastructure/Subscriptions/SubscriptionStorage.cs @@ -16,7 +16,7 @@ sealed class SubscriptionStorage : Feature class PrimeSubscriptions(IServiceControlSubscriptionStorage persister) : FeatureStartupTask { - protected override Task OnStart(IMessageSession session, CancellationToken cancellationToken = default) => persister?.Initialize() ?? Task.CompletedTask; + protected override Task OnStart(IMessageSession session, CancellationToken cancellationToken = default) => persister?.Initialize(cancellationToken) ?? Task.CompletedTask; protected override Task OnStop(IMessageSession session, CancellationToken cancellationToken = default) => Task.CompletedTask; } From cced5f5bf03f299b0b11f9c911fca2634ab5f1e3 Mon Sep 17 00:00:00 2001 From: John Simons Date: Wed, 12 Aug 2026 08:50:18 +1000 Subject: [PATCH 3/6] Propagate cancellation tokens through remaining persistence data stores and lifecycle operations MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This change continues the effort to consistently apply optional `CancellationToken` parameters to asynchronous methods across the remaining persistence interfaces and lifecycle operations, along with their EFCore and RavenDB implementations. Cancellation tokens are now propagated to underlying asynchronous database calls and lifecycle hooks, ensuring proper responsiveness to cancellation. This work retires the majority of cancellation analyzer suppressions across the persistence projects, leaving only the licensing data store—which belongs to the licensing phase—as an accepted exception. --- .../.editorconfig | 13 ++-- .../Implementation/BodyStorage/BodyStorage.cs | 2 +- .../Implementation/CustomCheckDataStore.cs | 16 ++--- .../Implementation/DataStoreBase.cs | 8 +-- .../Implementation/EndpointSettingsStore.cs | 10 +-- .../Implementation/EventLogDataStore.cs | 14 ++-- .../FailedErrorImportDataStore.cs | 10 +-- .../FailedMessageLifecycleDataStore.cs | 34 ++++----- .../FailedMessageQueryDataStore.cs | 44 ++++++------ .../Implementation/GroupsDataStore.cs | 30 ++++---- .../Implementation/LicensingDataStore.cs | 72 +++++++++---------- .../MessageRedirectsDataStore.cs | 20 +++--- .../Implementation/MessagesViewDataStore.cs | 20 +++--- .../Implementation/MonitoringDataStore.cs | 38 +++++----- .../Implementation/QueueAddressStore.cs | 6 +- .../Implementation/RetryBatchStore.cs | 42 +++++------ .../Implementation/RetryHistoryDataStore.cs | 28 ++++---- .../Implementation/RetryStagingStore.cs | 52 +++++++------- .../Implementation/SubscriptionStorage.cs | 12 ++-- .../TrialLicenseDataProvider.cs | 8 +-- .../Infrastructure/ExpectedLengthStream.cs | 2 +- .../IFailedMessageIngestionSqlDialect.cs | 6 +- .../Infrastructure/IRetryBatchSqlDialect.cs | 2 +- .../InsertOnlyTableReconciler.cs | 15 ++-- .../Infrastructure/OwnedStream.cs | 6 +- .../Infrastructure/RetentionSweeper.cs | 15 ++-- .../Infrastructure/SettingsStore.cs | 4 +- .../.editorconfig | 12 ++-- .../DatabaseSetup.cs | 2 +- .../LicenseStatusCheck.cs | 13 ++-- .../RavenEmbeddedPersistenceLifecycle.cs | 4 +- .../RavenExternalPersistenceLifecycle.cs | 4 +- .../RavenPersistenceLifecycleHostedService.cs | 4 +- .../.editorconfig | 16 +++-- src/ServiceControl.Persistence/.editorconfig | 8 --- 35 files changed, 297 insertions(+), 295 deletions(-) diff --git a/src/ServiceControl.Persistence.EFCore/.editorconfig b/src/ServiceControl.Persistence.EFCore/.editorconfig index 7651f47081..d8f0448f31 100644 --- a/src/ServiceControl.Persistence.EFCore/.editorconfig +++ b/src/ServiceControl.Persistence.EFCore/.editorconfig @@ -3,15 +3,12 @@ # Justification: ServiceControl app has no synchronization context dotnet_diagnostic.CA2007.severity = none -# Cancellation analyzer debt. These fire because Particular.Analyzers is no longer pinned to 0.9.0. -# They are scheduled work, not accepted exceptions: remove a line once this project has no -# violations of that rule left, and never add a rule back to this list. +# Cancellation analyzer debt, now confined to the licensing data store. ILicensingDataStore belongs to +# the licensing phase, not the persistence one, so its 15 optional-token additions land with the rest +# of Particular.LicensingComponent. Delete this block then; nothing else in the project violates a +# cancellation rule. +[Implementation/LicensingDataStore.cs] dotnet_diagnostic.PS0003.severity = none -dotnet_diagnostic.PS0004.severity = none -dotnet_diagnostic.PS0013.severity = none -dotnet_diagnostic.PS0017.severity = none -dotnet_diagnostic.PS0018.severity = none -dotnet_diagnostic.PS0019.severity = none # Disable style rules for auto-generated EF migrations [Migrations/**.cs] diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/BodyStorage.cs b/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/BodyStorage.cs index 700a978a0a..233771e24e 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/BodyStorage.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/BodyStorage/BodyStorage.cs @@ -21,7 +21,7 @@ public class BodyStorage(IServiceScopeFactory scopeFactory, IBodyStoragePersiste { public async Task TryFetch(string bodyId, CancellationToken cancellationToken = default) { - var row = await ExecuteWithDbContext(dbContext => ResolveBody(dbContext, bodyId, cancellationToken)); + var row = await ExecuteWithDbContext((dbContext, token) => ResolveBody(dbContext, bodyId, token), cancellationToken); if (row == null) { diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/CustomCheckDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/CustomCheckDataStore.cs index 182830bd92..af870ac38b 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/CustomCheckDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/CustomCheckDataStore.cs @@ -8,7 +8,7 @@ namespace ServiceControl.Persistence.EFCore.Implementation; public class CustomCheckDataStore(IServiceScopeFactory scopeFactory) : DataStoreBase(scopeFactory), ICustomChecksDataStore { - public Task UpdateCustomCheckStatus(CustomCheckDetail detail, CancellationToken cancellationToken = default) => ExecuteWithDbContext(async context => + public Task UpdateCustomCheckStatus(CustomCheckDetail detail, CancellationToken cancellationToken = default) => ExecuteWithDbContext(async (context, token) => { var status = CheckStateChange.Unchanged; @@ -44,11 +44,11 @@ await context.UpsertAsync([detail.GetDeterministicId()], entity.ReportedAt = detail.ReportedAt; entity.FailureReason = detail.FailureReason; }, - cancellationToken); + token); return status; - }); + }, cancellationToken); - public Task>> GetStats(PagingInfo paging, string? status = null, CancellationToken cancellationToken = default) => ExecuteWithDbContext(async context => + public Task>> GetStats(PagingInfo paging, string? status = null, CancellationToken cancellationToken = default) => ExecuteWithDbContext(async (context, token) => { var query = context.CustomChecks.AsQueryable().AsNoTracking(); @@ -63,7 +63,7 @@ public Task>> GetStats(PagingInfo paging, string? .OrderBy(c => c.ReportedAt) .Skip(paging.Offset) .Take(paging.PageSize) - .ToListAsync(cancellationToken); + .ToListAsync(token); return new QueryResult>(page.Select(c => new CustomCheck { @@ -74,9 +74,9 @@ public Task>> GetStats(PagingInfo paging, string? ReportedAt = c.ReportedAt, FailureReason = c.FailureReason }).ToList(), new QueryStatsInfo("", page.Count, false)); - }); + }, cancellationToken); - public Task DeleteCustomCheck(Guid id, CancellationToken cancellationToken = default) => ExecuteWithDbContext(async context => await context.CustomChecks.AsNoTracking().Where(cc => cc.Id == id).ExecuteDeleteAsync(cancellationToken)); + public Task DeleteCustomCheck(Guid id, CancellationToken cancellationToken = default) => ExecuteWithDbContext(async (context, token) => await context.CustomChecks.AsNoTracking().Where(cc => cc.Id == id).ExecuteDeleteAsync(token), cancellationToken); - public Task GetNumberOfFailedChecks(CancellationToken cancellationToken = default) => ExecuteWithDbContext(async context => await context.CustomChecks.AsNoTracking().CountAsync(p => p.Status == Status.Fail, cancellationToken)); + public Task GetNumberOfFailedChecks(CancellationToken cancellationToken = default) => ExecuteWithDbContext(async (context, token) => await context.CustomChecks.AsNoTracking().CountAsync(p => p.Status == Status.Fail, token), cancellationToken); } diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/DataStoreBase.cs b/src/ServiceControl.Persistence.EFCore/Implementation/DataStoreBase.cs index 90299c070e..5dc89f5fe0 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/DataStoreBase.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/DataStoreBase.cs @@ -16,21 +16,21 @@ public abstract class DataStoreBase(IServiceScopeFactory scopeFactory) /// /// Executes an operation with a scoped DbContext, returning a result /// - protected async Task ExecuteWithDbContext(Func> operation) + protected async Task ExecuteWithDbContext(Func> operation, CancellationToken cancellationToken = default) { await using var scope = scopeFactory.CreateAsyncScope();// Use CreateAsyncScope for async disposal var dbContext = scope.ServiceProvider.GetRequiredService(); - return await operation(dbContext); + return await operation(dbContext, cancellationToken); } /// /// Executes an operation with a scoped DbContext, without returning a result /// - protected async Task ExecuteWithDbContext(Func operation) + protected async Task ExecuteWithDbContext(Func operation, CancellationToken cancellationToken = default) { await using var scope = scopeFactory.CreateAsyncScope(); var dbContext = scope.ServiceProvider.GetRequiredService(); - await operation(dbContext); + await operation(dbContext, cancellationToken); } /// diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/EndpointSettingsStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/EndpointSettingsStore.cs index d709c5a2d7..7062260163 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/EndpointSettingsStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/EndpointSettingsStore.cs @@ -10,14 +10,14 @@ public IAsyncEnumerable GetAllEndpointSettings(CancellationTok => ExecuteWithDbContext(context => context.EndpointSettings.Select(row => new EndpointSettings { Name = row.Name, TrackInstances = row.TrackInstances }).AsAsyncEnumerable(), cancellationToken); public Task UpdateEndpointSettings(EndpointSettings settings, CancellationToken cancellationToken = default) => - ExecuteWithDbContext(async context => + ExecuteWithDbContext(async (context, token) => await context.UpsertAsync([settings.Name], () => new EndpointSettingsEntity() { Name = settings.Name, TrackInstances = settings.TrackInstances }, entity => entity.TrackInstances = settings.TrackInstances, - cancellationToken - )); + token + ), cancellationToken); public Task Delete(string name, CancellationToken cancellationToken = default) => - ExecuteWithDbContext(context => context.EndpointSettings.Where(x => x.Name == name) - .ExecuteDeleteAsync(cancellationToken)); + ExecuteWithDbContext((context, token) => context.EndpointSettings.Where(x => x.Name == name) + .ExecuteDeleteAsync(token), cancellationToken); } \ No newline at end of file diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/EventLogDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/EventLogDataStore.cs index f7403c7ff4..ffc0efa4e7 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/EventLogDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/EventLogDataStore.cs @@ -9,7 +9,7 @@ namespace ServiceControl.Persistence.EFCore.Implementation; public class EventLogDataStore(IServiceScopeFactory scopeFactory) : DataStoreBase(scopeFactory), IEventLogDataStore { public Task Add(EventLogItem logItem, CancellationToken cancellationToken = default) => - ExecuteWithDbContext(async dbContext => + ExecuteWithDbContext(async (dbContext, token) => { dbContext.EventLogItems.Add(new EventLogItemEntity { @@ -21,12 +21,12 @@ public Task Add(EventLogItem logItem, CancellationToken cancellationToken = defa EventType = logItem.EventType }); - await dbContext.SaveChangesAsync(cancellationToken); - }); + await dbContext.SaveChangesAsync(token); + }, cancellationToken); public Task>> GetEventLogItems( PagingInfo pagingInfo, string? knownVersion = null, CancellationToken cancellationToken = default) => - ExecuteWithDbContext(async dbContext => + ExecuteWithDbContext(async (dbContext, token) => { var query = dbContext.EventLogItems.AsNoTracking(); @@ -40,7 +40,7 @@ public Task>> GetEventLogItems( Newest = g.Max(e => (DateTime?)e.RaisedAt), HighestId = g.Max(e => (long?)e.Id) }) - .FirstOrDefaultAsync(cancellationToken); + .FirstOrDefaultAsync(token); var total = stats?.Total ?? 0; var version = Version(total, stats?.Newest, stats?.HighestId); @@ -71,10 +71,10 @@ public Task>> GetEventLogItems( Category = e.Category, EventType = e.EventType }) - .ToListAsync(cancellationToken); + .ToListAsync(token); return new QueryResult>(items, queryStats); - }); + }, cancellationToken); // Synthesised version ID to be used for an ETag. The highest key is the monotonic term: identity // values gap but never repeat, so an insert moves the version whatever its RaisedAt says. diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/FailedErrorImportDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/FailedErrorImportDataStore.cs index a11bf466e5..bfdfbbc09f 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/FailedErrorImportDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/FailedErrorImportDataStore.cs @@ -23,14 +23,14 @@ public class FailedErrorImportDataStore( const int BatchSize = 100; public Task QueryContainsFailedImports(CancellationToken cancellationToken = default) => - ExecuteWithDbContext(dbContext => dbContext.FailedErrorImports.AsNoTracking().AnyAsync()); + ExecuteWithDbContext((dbContext, token) => dbContext.FailedErrorImports.AsNoTracking().AnyAsync(token), cancellationToken); // Update-first, then insert. The dedupe key is deterministic, so a repeat failure updates the // existing row and concurrent writers that both miss it race only on the insert. The loser of // that race confirms the row is now present (the winner stored the same logical failure) and // otherwise rethrows, so the caller never treats a message as stored when it is not. public Task StoreFailedErrorImport(FailedErrorImport failure, CancellationToken cancellationToken = default) => - ExecuteWithDbContext(async dbContext => + ExecuteWithDbContext(async (dbContext, token) => { var uniqueMessageId = FailedErrorImport.DeriveKey(failure.Message.Headers, failure.Message.Id); var body = failure.Message.Body ?? []; @@ -39,7 +39,7 @@ public Task StoreFailedErrorImport(FailedErrorImport failure, CancellationToken if (storeExternally) { var contentType = failure.Message.Headers.GetValueOrDefault(Headers.ContentType) ?? "application/octet-stream"; - await bodyStorage.WriteBody(FailedErrorImportEntity.ExternalBodyId(uniqueMessageId), body, contentType); + await bodyStorage.WriteBody(FailedErrorImportEntity.ExternalBodyId(uniqueMessageId), body, contentType, token); } var failedAt = timeProvider.GetUtcNow().UtcDateTime; @@ -63,8 +63,8 @@ public Task StoreFailedErrorImport(FailedErrorImport failure, CancellationToken entity.Body = storedBody; entity.BodyStoredExternally = storeExternally; entity.ExceptionInfo = failure.ExceptionInfo; - }); - }); + }, token); + }, cancellationToken); // Replays oldest-first. Successful imports delete their row; failures are left in place, so the // count of failures so far is exactly the offset to the next unseen row. This walks the whole diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageLifecycleDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageLifecycleDataStore.cs index 26c417d579..e17695d8a6 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageLifecycleDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageLifecycleDataStore.cs @@ -11,7 +11,7 @@ public class FailedMessageLifecycleDataStore(IServiceScopeFactory scopeFactory) { public async Task MarkAsArchived(string failedMessageId, CancellationToken cancellationToken = default) { - await ExecuteWithDbContext(async dbContext => + await ExecuteWithDbContext(async (dbContext, token) => { if (!Guid.TryParse(failedMessageId, out var uniqueMessageId)) { @@ -25,13 +25,13 @@ await dbContext.FailedMessages .ExecuteUpdateAsync(s => s .SetProperty(fm => fm.Status, FailedMessageStatus.Archived) .SetProperty(fm => fm.StatusChangedAt, now) - .SetProperty(fm => fm.LastModified, now), cancellationToken); - }); + .SetProperty(fm => fm.LastModified, now), token); + }, cancellationToken); } public async Task MarkAsResolved(string failedMessageId, CancellationToken cancellationToken = default) { - return await ExecuteWithDbContext(async dbContext => + return await ExecuteWithDbContext(async (dbContext, token) => { if (!Guid.TryParse(failedMessageId, out var uniqueMessageId)) { @@ -45,10 +45,10 @@ public async Task MarkAsResolved(string failedMessageId, CancellationToken .ExecuteUpdateAsync(s => s .SetProperty(fm => fm.Status, FailedMessageStatus.Resolved) .SetProperty(fm => fm.StatusChangedAt, now) - .SetProperty(fm => fm.LastModified, now), cancellationToken); + .SetProperty(fm => fm.LastModified, now), token); return affected > 0; - }); + }, cancellationToken); } public async Task UnArchiveMessages(IEnumerable failedMessageIds, CancellationToken cancellationToken = default) @@ -63,7 +63,7 @@ public async Task UnArchiveMessages(IEnumerable failedMessageI return []; } - return await ExecuteWithDbContext(async dbContext => + return await ExecuteWithDbContext(async (dbContext, token) => { var now = DateTime.UtcNow; @@ -71,7 +71,7 @@ public async Task UnArchiveMessages(IEnumerable failedMessageI var unarchivableIds = await dbContext.FailedMessages .Where(fm => ids.Contains(fm.UniqueMessageId) && fm.Status == FailedMessageStatus.Archived) .Select(fm => fm.UniqueMessageId) - .ToListAsync(cancellationToken); + .ToListAsync(token); if (unarchivableIds.Count == 0) { @@ -83,15 +83,15 @@ await dbContext.FailedMessages .ExecuteUpdateAsync(s => s .SetProperty(fm => fm.Status, FailedMessageStatus.Unresolved) .SetProperty(fm => fm.StatusChangedAt, now) - .SetProperty(fm => fm.LastModified, now), cancellationToken); + .SetProperty(fm => fm.LastModified, now), token); return unarchivableIds.Select(id => id.ToString()).ToArray(); - }); + }, cancellationToken); } public async Task UnArchiveMessagesByRange(DateTime from, DateTime to, CancellationToken cancellationToken = default) { - return await ExecuteWithDbContext(async dbContext => + return await ExecuteWithDbContext(async (dbContext, token) => { var now = DateTime.UtcNow; @@ -101,7 +101,7 @@ public async Task UnArchiveMessagesByRange(DateTime from, DateTime to, && fm.LastModified >= from && fm.LastModified <= to) .Select(fm => fm.UniqueMessageId) - .ToListAsync(cancellationToken); + .ToListAsync(token); if (unarchivableIds.Count == 0) { @@ -113,15 +113,15 @@ await dbContext.FailedMessages .ExecuteUpdateAsync(s => s .SetProperty(fm => fm.Status, FailedMessageStatus.Unresolved) .SetProperty(fm => fm.StatusChangedAt, now) - .SetProperty(fm => fm.LastModified, now), cancellationToken); + .SetProperty(fm => fm.LastModified, now), token); return unarchivableIds.Select(id => id.ToString()).ToArray(); - }); + }, cancellationToken); } public async Task RevertRetry(string messageUniqueId, CancellationToken cancellationToken = default) { - await ExecuteWithDbContext(async dbContext => + await ExecuteWithDbContext(async (dbContext, token) => { if (!Guid.TryParse(messageUniqueId, out var uniqueMessageId)) { @@ -135,7 +135,7 @@ await dbContext.FailedMessages .ExecuteUpdateAsync(s => s .SetProperty(fm => fm.Status, FailedMessageStatus.Unresolved) .SetProperty(fm => fm.StatusChangedAt, now) - .SetProperty(fm => fm.LastModified, now), cancellationToken); - }); + .SetProperty(fm => fm.LastModified, now), token); + }, cancellationToken); } } \ No newline at end of file diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageQueryDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageQueryDataStore.cs index 4aa82923bc..e74536a194 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageQueryDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageQueryDataStore.cs @@ -13,35 +13,35 @@ namespace ServiceControl.Persistence.EFCore.Implementation; public class FailedMessageQueryDataStore(IServiceScopeFactory scopeFactory) : DataStoreBase(scopeFactory), IFailedMessageQueryDataStore { public Task>> GetFailedMessages(string? status, string? modified, string? queueAddress, PagingInfo pagingInfo, SortInfo sortInfo, CancellationToken cancellationToken = default) => - ExecuteWithDbContext(dbContext => dbContext.FailedMessages + ExecuteWithDbContext((dbContext, token) => dbContext.FailedMessages .AsNoTracking() .FilterByStatus(status) .FilterByLastModifiedRange(modified) .FilterByQueueAddress(queueAddress) - .ToPagedResult(pagingInfo, sortInfo, cancellationToken)); + .ToPagedResult(pagingInfo, sortInfo, token), cancellationToken); public Task GetFailedMessagesStats(string? status, string? modified, string? queueAddress, CancellationToken cancellationToken = default) => - ExecuteWithDbContext(dbContext => dbContext.FailedMessages + ExecuteWithDbContext((dbContext, token) => dbContext.FailedMessages .AsNoTracking() .FilterByStatus(status) .FilterByLastModifiedRange(modified) .FilterByQueueAddress(queueAddress) - .ToQueryStatsInfo(cancellationToken)); + .ToQueryStatsInfo(token), cancellationToken); public Task>> GetFailedMessagesByEndpoint(string? status, string endpointName, string? modified, PagingInfo pagingInfo, SortInfo sortInfo, CancellationToken cancellationToken = default) => - ExecuteWithDbContext(dbContext => dbContext.FailedMessages + ExecuteWithDbContext((dbContext, token) => dbContext.FailedMessages .AsNoTracking() .Where(message => message.ReceivingEndpointName == endpointName) .FilterByStatus(status) .FilterByLastModifiedRange(modified) - .ToPagedResult(pagingInfo, sortInfo, cancellationToken)); + .ToPagedResult(pagingInfo, sortInfo, token), cancellationToken); public Task> GetFailedMessagesSummary(CancellationToken cancellationToken = default) => - ExecuteWithDbContext(async dbContext => + ExecuteWithDbContext(async (dbContext, token) => { - var endpoints = await CountBy(dbContext, message => message.ReceivingEndpointName, cancellationToken); - var hosts = await CountBy(dbContext, message => message.ReceivingEndpointHost, cancellationToken); - var messageTypes = await CountBy(dbContext, message => message.MessageType, cancellationToken); + var endpoints = await CountBy(dbContext, message => message.ReceivingEndpointName, token); + var hosts = await CountBy(dbContext, message => message.ReceivingEndpointHost, token); + var messageTypes = await CountBy(dbContext, message => message.MessageType, token); return (IDictionary)new Dictionary { @@ -49,10 +49,10 @@ public Task> GetFailedMessagesSummary(CancellationTo [FailedMessageSummaryKeys.Hosts] = hosts, [FailedMessageSummaryKeys.MessageTypes] = messageTypes }; - }); + }, cancellationToken); public Task GetLatestFailedMessageView(string failedMessageId, CancellationToken cancellationToken = default) => - ExecuteWithDbContext(async dbContext => + ExecuteWithDbContext(async (dbContext, token) => { if (!Guid.TryParse(failedMessageId, out var uniqueMessageId)) { @@ -61,13 +61,13 @@ public Task> GetFailedMessagesSummary(CancellationTo var entity = await dbContext.FailedMessages .AsNoTracking() - .SingleOrDefaultAsync(message => message.UniqueMessageId == uniqueMessageId, cancellationToken); + .SingleOrDefaultAsync(message => message.UniqueMessageId == uniqueMessageId, token); return entity?.ToFailedMessageView(); - }); + }, cancellationToken); public Task GetFailedMessage(string failedMessageId, CancellationToken cancellationToken = default) => - ExecuteWithDbContext(async dbContext => + ExecuteWithDbContext(async (dbContext, token) => { if (!Guid.TryParse(failedMessageId, out var uniqueMessageId)) { @@ -76,35 +76,35 @@ public Task> GetFailedMessagesSummary(CancellationTo var entity = await dbContext.FailedMessages .AsNoTracking() - .SingleOrDefaultAsync(message => message.UniqueMessageId == uniqueMessageId, cancellationToken); + .SingleOrDefaultAsync(message => message.UniqueMessageId == uniqueMessageId, token); if (entity == null) { return null; } - var groups = await ReadGroups(dbContext, [uniqueMessageId], cancellationToken); + var groups = await ReadGroups(dbContext, [uniqueMessageId], token); return entity.ToFailedMessage(groups.GetValueOrDefault(uniqueMessageId, [])); - }); + }, cancellationToken); public Task GetFailedMessagesByIds(Guid[] ids, CancellationToken cancellationToken = default) => - ExecuteWithDbContext(async dbContext => + ExecuteWithDbContext(async (dbContext, token) => { var entities = await dbContext.FailedMessages .AsNoTracking() .Where(message => ids.Contains(message.UniqueMessageId)) - .ToListAsync(cancellationToken); + .ToListAsync(token); if (entities.Count == 0) { return []; } - var groups = await ReadGroups(dbContext, [.. entities.Select(entity => entity.UniqueMessageId)], cancellationToken); + var groups = await ReadGroups(dbContext, [.. entities.Select(entity => entity.UniqueMessageId)], token); return [.. entities.Select(entity => entity.ToFailedMessage(groups.GetValueOrDefault(entity.UniqueMessageId, [])))]; - }); + }, cancellationToken); static async Task> CountBy(ServiceControlDbContext dbContext, Expression> selector, CancellationToken cancellationToken) => await dbContext.FailedMessages diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/GroupsDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/GroupsDataStore.cs index e31d211832..7996849634 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/GroupsDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/GroupsDataStore.cs @@ -13,7 +13,7 @@ namespace ServiceControl.Persistence.EFCore.Implementation; public class GroupsDataStore(IServiceScopeFactory scopeFactory) : DataStoreBase(scopeFactory), IGroupsDataStore { public Task> GetUnresolvedGroupsByClassifier(string classifier, string classifierFilter, CancellationToken cancellationToken = default) => - ExecuteWithDbContext(async dbContext => + ExecuteWithDbContext(async (dbContext, token) => { var groups = ByClassifier(dbContext, classifier); @@ -22,46 +22,46 @@ public Task> GetUnresolvedGroupsByClassifier(string clas groups = groups.Where(group => group.Title == classifierFilter); } - var views = await MostRecent(groups.AggregateGroups(WithStatus(dbContext, FailedMessageStatus.Unresolved)), cancellationToken); + var views = await MostRecent(groups.AggregateGroups(WithStatus(dbContext, FailedMessageStatus.Unresolved)), token); - await AttachComments(dbContext, views, cancellationToken); + await AttachComments(dbContext, views, token); return views; - }); + }, cancellationToken); public Task> GetArchivedGroupsByClassifier(string classifier, CancellationToken cancellationToken = default) => - ExecuteWithDbContext(dbContext => MostRecent( - ByClassifier(dbContext, classifier).AggregateGroups(WithStatus(dbContext, FailedMessageStatus.Archived)), cancellationToken)); + ExecuteWithDbContext((dbContext, token) => MostRecent( + ByClassifier(dbContext, classifier).AggregateGroups(WithStatus(dbContext, FailedMessageStatus.Archived)), token), cancellationToken); public Task> GetUnresolvedGroup(string groupId, string status, string modified, CancellationToken cancellationToken = default) => - ExecuteWithDbContext(dbContext => SingleGroup(dbContext, groupId, FailedMessageStatus.Unresolved, status, modified, cancellationToken)); + ExecuteWithDbContext((dbContext, token) => SingleGroup(dbContext, groupId, FailedMessageStatus.Unresolved, status, modified, token), cancellationToken); public Task> GetArchivedGroup(string groupId, string status, string modified, CancellationToken cancellationToken = default) => - ExecuteWithDbContext(dbContext => SingleGroup(dbContext, groupId, FailedMessageStatus.Archived, status, modified, cancellationToken)); + ExecuteWithDbContext((dbContext, token) => SingleGroup(dbContext, groupId, FailedMessageStatus.Archived, status, modified, token), cancellationToken); public Task>> GetGroupErrors(string groupId, string status, string modified, SortInfo sortInfo, PagingInfo pagingInfo, CancellationToken cancellationToken = default) => - ExecuteWithDbContext(dbContext => InGroup(dbContext, groupId, status, modified).ToPagedResult(pagingInfo, sortInfo, cancellationToken)); + ExecuteWithDbContext((dbContext, token) => InGroup(dbContext, groupId, status, modified).ToPagedResult(pagingInfo, sortInfo, token), cancellationToken); public Task GetGroupErrorsCount(string groupId, string status, string modified, CancellationToken cancellationToken = default) => - ExecuteWithDbContext(dbContext => InGroup(dbContext, groupId, status, modified).ToQueryStatsInfo(cancellationToken)); + ExecuteWithDbContext((dbContext, token) => InGroup(dbContext, groupId, status, modified).ToQueryStatsInfo(token), cancellationToken); public Task EditComment(string groupId, string comment, CancellationToken cancellationToken = default) => - ExecuteWithDbContext(async dbContext => + ExecuteWithDbContext(async (dbContext, token) => { if (string.IsNullOrWhiteSpace(comment)) { - await RemoveComment(dbContext, groupId, cancellationToken); + await RemoveComment(dbContext, groupId, token); return; } await dbContext.UpsertAsync([groupId], () => new GroupCommentEntity { GroupId = groupId, Comment = comment }, entity => entity.Comment = comment, - cancellationToken); - }); + token); + }, cancellationToken); public Task DeleteComment(string groupId, CancellationToken cancellationToken = default) => - ExecuteWithDbContext(dbContext => RemoveComment(dbContext, groupId, cancellationToken)); + ExecuteWithDbContext((dbContext, token) => RemoveComment(dbContext, groupId, token), cancellationToken); static Task RemoveComment(ServiceControlDbContext dbContext, string groupId, CancellationToken cancellationToken) => dbContext.GroupComments diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/LicensingDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/LicensingDataStore.cs index 3405c3d627..d5762a6d0b 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/LicensingDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/LicensingDataStore.cs @@ -17,7 +17,7 @@ class LicensingDataStore(IServiceScopeFactory scopeFactory, TimeProvider timePro static readonly BrokerMetadata DefaultBrokerMetadata = new(null, []); public Task> GetAllEndpoints(bool includePlatformEndpoints, CancellationToken cancellationToken) => - ExecuteWithDbContext(async context => + ExecuteWithDbContext(async (context, token) => { var query = context.LicensingEndpoints.AsNoTracking(); @@ -26,13 +26,13 @@ public Task> GetAllEndpoints(bool includePlatformEndpoints query = query.Where(endpoint => !endpoint.EndpointIndicators.Contains(PlatformEndpointIndicator)); } - var rows = await query.ToListAsync(cancellationToken); + var rows = await query.ToListAsync(token); return rows.Select(ToEndpoint); - }); + }, cancellationToken); public Task GetEndpoint(EndpointIdentifier id, CancellationToken cancellationToken = default) => - ExecuteWithDbContext(async context => + ExecuteWithDbContext(async (context, token) => { var normalizedName = Normalize(id.Name); @@ -46,13 +46,13 @@ public Task> GetAllEndpoints(bool includePlatformEndpoints .Where(row => row.NormalizedName == endpoint.NormalizedName && row.ThroughputSource == endpoint.ThroughputSource) .Max(row => row.DateUtc) }) - .SingleOrDefaultAsync(cancellationToken); + .SingleOrDefaultAsync(token); return row is null ? null : ToEndpoint(row.Endpoint, row.LastCollectedDate); - }); + }, cancellationToken); public Task> GetEndpoints(IList endpointIds, CancellationToken cancellationToken) => - ExecuteWithDbContext(async context => + ExecuteWithDbContext(async (context, token) => { var normalizedNames = endpointIds.Select(id => Normalize(id.Name)).Distinct().ToList(); @@ -68,7 +68,7 @@ public Task> GetAllEndpoints(bool includePlatformEndpoints .Where(row => row.NormalizedName == endpoint.NormalizedName && row.ThroughputSource == endpoint.ThroughputSource) .Max(row => row.DateUtc) }) - .ToListAsync(cancellationToken); + .ToListAsync(token); var found = rows.ToDictionary(row => (row.Endpoint.NormalizedName, row.Endpoint.ThroughputSource)); @@ -78,10 +78,10 @@ public Task> GetAllEndpoints(bool includePlatformEndpoints : null)) .ToList() .AsEnumerable(); - }); + }, cancellationToken); public Task SaveEndpoint(Endpoint endpoint, CancellationToken cancellationToken) => - ExecuteWithDbContext(context => + ExecuteWithDbContext((context, token) => { var normalizedName = Normalize(endpoint.Id.Name); var sanitizedName = endpoint.SanitizedName ?? string.Empty; @@ -109,11 +109,11 @@ public Task SaveEndpoint(Endpoint endpoint, CancellationToken cancellationToken) row.Scope = endpoint.Scope; row.EndpointIndicators = indicators; }, - cancellationToken); - }); + token); + }, cancellationToken); public Task>> GetEndpointThroughputByQueueName(IList queueNames, CancellationToken cancellationToken) => - ExecuteWithDbContext(async context => + ExecuteWithDbContext(async (context, token) => { var results = queueNames.ToDictionary(queueName => queueName, _ => Enumerable.Empty()); @@ -143,7 +143,7 @@ public Task>> GetEndpointThrough throughput.DateUtc, throughput.MessageCount }) - .ToListAsync(cancellationToken); + .ToListAsync(token); foreach (var endpointRows in rows.GroupBy(row => (row.NormalizedSanitizedName, row.ThroughputSource))) { @@ -159,7 +159,7 @@ public Task>> GetEndpointThrough } return (IDictionary>)results; - }); + }, cancellationToken); public async Task RecordEndpointThroughput(string endpointName, ThroughputSource throughputSource, IList throughput, CancellationToken cancellationToken) { @@ -172,8 +172,8 @@ public async Task RecordEndpointThroughput(string endpointName, ThroughputSource // writer takes the update path, so a couple of attempts is enough. for (var attempt = 1; attempt <= MaxRecordAttempts; attempt++) { - var recorded = await ExecuteWithDbContext(context => - TryRecordEndpointThroughput(context, endpointName, throughputSource, throughput, cancellationToken)); + var recorded = await ExecuteWithDbContext((context, token) => + TryRecordEndpointThroughput(context, endpointName, throughputSource, throughput, token), cancellationToken); if (recorded) { @@ -247,14 +247,14 @@ static async Task TryRecordEndpointThroughput(ServiceControlDbContext cont } public Task UpdateUserIndicatorOnEndpoints(List userIndicatorUpdates, CancellationToken cancellationToken) => - ExecuteWithDbContext(async context => + ExecuteWithDbContext(async (context, token) => { var updates = userIndicatorUpdates.ToDictionary(update => Normalize(update.Name), update => update.UserIndicator); var names = updates.Keys.ToList(); var rows = await context.LicensingEndpoints .Where(endpoint => names.Contains(endpoint.NormalizedName) || names.Contains(endpoint.NormalizedSanitizedName)) - .ToListAsync(cancellationToken); + .ToListAsync(token); var indicatorBySanitizedName = new Dictionary(); @@ -278,7 +278,7 @@ public Task UpdateUserIndicatorOnEndpoints(List userIndicat var siblings = await context.LicensingEndpoints .Where(endpoint => sanitizedNames.Contains(endpoint.NormalizedSanitizedName)) - .ToListAsync(cancellationToken); + .ToListAsync(token); foreach (var sibling in siblings.Where(row => !alreadyLoaded.Contains((row.NormalizedName, row.ThroughputSource)))) { @@ -289,8 +289,8 @@ public Task UpdateUserIndicatorOnEndpoints(List userIndicat } } - await context.SaveChangesAsync(cancellationToken); - }); + await context.SaveChangesAsync(token); + }, cancellationToken); public Task IsThereThroughputForLastXDays(int days, CancellationToken cancellationToken) => IsThereThroughput(days, throughputSource: null, includeToday: false, cancellationToken); @@ -299,7 +299,7 @@ public Task IsThereThroughputForLastXDaysForSource(int days, ThroughputSou IsThereThroughput(days, throughputSource, includeToday, cancellationToken); Task IsThereThroughput(int days, ThroughputSource? throughputSource, bool includeToday, CancellationToken cancellationToken) => - ExecuteWithDbContext(context => + ExecuteWithDbContext((context, token) => { var today = DateOnly.FromDateTime(timeProvider.GetUtcNow().UtcDateTime); var from = today.AddDays(-days); @@ -314,35 +314,35 @@ Task IsThereThroughput(int days, ThroughputSource? throughputSource, bool query = query.Where(row => row.ThroughputSource == throughputSource); } - return query.AnyAsync(cancellationToken); - }); + return query.AnyAsync(token); + }, cancellationToken); public Task GetBrokerMetadata(CancellationToken cancellationToken) => - ExecuteWithDbContext(async context => - await context.GetSetting(SettingKeys.BrokerMetadata, cancellationToken) ?? DefaultBrokerMetadata); + ExecuteWithDbContext(async (context, token) => + await context.GetSetting(SettingKeys.BrokerMetadata, token) ?? DefaultBrokerMetadata, cancellationToken); public Task SaveBrokerMetadata(BrokerMetadata brokerMetadata, CancellationToken cancellationToken) => - ExecuteWithDbContext(context => context.StoreSetting(SettingKeys.BrokerMetadata, brokerMetadata, cancellationToken)); + ExecuteWithDbContext((context, token) => context.StoreSetting(SettingKeys.BrokerMetadata, brokerMetadata, token), cancellationToken); public Task GetAuditServiceMetadata(CancellationToken cancellationToken = default) => - ExecuteWithDbContext(async context => - await context.GetSetting(SettingKeys.AuditServiceMetadata, cancellationToken) ?? DefaultAuditServiceMetadata); + ExecuteWithDbContext(async (context, token) => + await context.GetSetting(SettingKeys.AuditServiceMetadata, token) ?? DefaultAuditServiceMetadata, cancellationToken); public Task SaveAuditServiceMetadata(AuditServiceMetadata auditServiceMetadata, CancellationToken cancellationToken) => - ExecuteWithDbContext(context => context.StoreSetting(SettingKeys.AuditServiceMetadata, auditServiceMetadata, cancellationToken)); + ExecuteWithDbContext((context, token) => context.StoreSetting(SettingKeys.AuditServiceMetadata, auditServiceMetadata, token), cancellationToken); public Task> GetReportMasks(CancellationToken cancellationToken) => - ExecuteWithDbContext(async context => - await context.GetSetting>(SettingKeys.ReportMasks, cancellationToken) ?? []); + ExecuteWithDbContext(async (context, token) => + await context.GetSetting>(SettingKeys.ReportMasks, token) ?? [], cancellationToken); public Task SaveReportMasks(List reportMasks, CancellationToken cancellationToken) => - ExecuteWithDbContext(context => context.StoreSetting(SettingKeys.ReportMasks, reportMasks, cancellationToken)); + ExecuteWithDbContext((context, token) => context.StoreSetting(SettingKeys.ReportMasks, reportMasks, token), cancellationToken); public Task GetLicensedEndpointDetails(CancellationToken cancellationToken) => - ExecuteWithDbContext(context => context.GetSetting(SettingKeys.LicensedEndpointDetails, cancellationToken)); + ExecuteWithDbContext((context, token) => context.GetSetting(SettingKeys.LicensedEndpointDetails, token), cancellationToken); public Task SaveLicensedEndpointDetails(LicensedEndpointDetails result, CancellationToken cancellationToken) => - ExecuteWithDbContext(context => context.StoreSetting(SettingKeys.LicensedEndpointDetails, result, cancellationToken)); + ExecuteWithDbContext((context, token) => context.StoreSetting(SettingKeys.LicensedEndpointDetails, result, token), cancellationToken); static Endpoint ToEndpoint(LicensingEndpointEntity row) => ToEndpoint(row, lastCollectedDate: null); diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/MessageRedirectsDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/MessageRedirectsDataStore.cs index 49ebd23bca..b0a5aa216b 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/MessageRedirectsDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/MessageRedirectsDataStore.cs @@ -8,11 +8,11 @@ namespace ServiceControl.Persistence.EFCore.Implementation; public class MessageRedirectsDataStore(IServiceScopeFactory scopeFactory) : DataStoreBase(scopeFactory), IMessageRedirectsDataStore { public Task> GetRedirects(CancellationToken cancellationToken = default) => - ExecuteWithDbContext>(async dbContext => + ExecuteWithDbContext>(async (dbContext, token) => { var rows = await dbContext.MessageRedirects .AsNoTracking() - .ToListAsync(cancellationToken); + .ToListAsync(token); return [.. rows.Select(row => new MessageRedirect { @@ -20,10 +20,10 @@ public Task> GetRedirects(CancellationToken cance ToPhysicalAddress = row.ToPhysicalAddress, LastModified = row.LastModified })]; - }); + }, cancellationToken); public Task AddRedirect(MessageRedirect redirect, CancellationToken cancellationToken = default) => - ExecuteWithDbContext(async dbContext => + ExecuteWithDbContext(async (dbContext, token) => { dbContext.MessageRedirects.Add(new MessageRedirectEntity { @@ -32,18 +32,18 @@ public Task AddRedirect(MessageRedirect redirect, CancellationToken cancellation LastModified = redirect.LastModified }); - await dbContext.SaveChangesAsync(cancellationToken); - }); + await dbContext.SaveChangesAsync(token); + }, cancellationToken); public Task UpdateRedirect(MessageRedirect redirect, CancellationToken cancellationToken = default) => - ExecuteWithDbContext(dbContext => dbContext.MessageRedirects + ExecuteWithDbContext((dbContext, token) => dbContext.MessageRedirects .Where(row => row.FromPhysicalAddress == redirect.FromPhysicalAddress) .ExecuteUpdateAsync(setters => setters .SetProperty(row => row.ToPhysicalAddress, redirect.ToPhysicalAddress) - .SetProperty(row => row.LastModified, redirect.LastModified), cancellationToken)); + .SetProperty(row => row.LastModified, redirect.LastModified), token), cancellationToken); public Task RemoveRedirect(MessageRedirect redirect, CancellationToken cancellationToken = default) => - ExecuteWithDbContext(dbContext => dbContext.MessageRedirects + ExecuteWithDbContext((dbContext, token) => dbContext.MessageRedirects .Where(row => row.FromPhysicalAddress == redirect.FromPhysicalAddress) - .ExecuteDeleteAsync(cancellationToken)); + .ExecuteDeleteAsync(token), cancellationToken); } diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/MessagesViewDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/MessagesViewDataStore.cs index 6c3d0fabc2..2b2d69d2db 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/MessagesViewDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/MessagesViewDataStore.cs @@ -10,37 +10,37 @@ namespace ServiceControl.Persistence.EFCore.Implementation; public class MessagesViewDataStore(IServiceScopeFactory scopeFactory, IFullTextSearchDialect fullTextSearch) : DataStoreBase(scopeFactory), IMessagesViewDataStore { public Task>> GetAllMessages(PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, DateTimeRange? timeSentRange = null, CancellationToken cancellationToken = default) => - ExecuteWithDbContext(dbContext => dbContext.FailedMessages + ExecuteWithDbContext((dbContext, token) => dbContext.FailedMessages .AsNoTracking() .IncludeSystemMessagesWhere(includeSystemMessages) .FilterBySentTimeRange(timeSentRange) - .ToPagedMessagesResult(pagingInfo, sortInfo, cancellationToken)); + .ToPagedMessagesResult(pagingInfo, sortInfo, token), cancellationToken); public Task>> GetAllMessagesForEndpoint(string endpointName, PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, DateTimeRange? timeSentRange = null, CancellationToken cancellationToken = default) => - ExecuteWithDbContext(dbContext => dbContext.FailedMessages + ExecuteWithDbContext((dbContext, token) => dbContext.FailedMessages .AsNoTracking() .Where(message => message.ReceivingEndpointName == endpointName) .IncludeSystemMessagesWhere(includeSystemMessages) .FilterBySentTimeRange(timeSentRange) - .ToPagedMessagesResult(pagingInfo, sortInfo, cancellationToken)); + .ToPagedMessagesResult(pagingInfo, sortInfo, token), cancellationToken); // includeSystemMessages is unused here: a conversation is incomplete without the system messages that took part in it. public Task>> GetAllMessagesByConversation(string conversationId, PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, CancellationToken cancellationToken = default) => - ExecuteWithDbContext(dbContext => dbContext.FailedMessages + ExecuteWithDbContext((dbContext, token) => dbContext.FailedMessages .AsNoTracking() .Where(message => message.ConversationId == conversationId) - .ToPagedMessagesResult(pagingInfo, sortInfo, cancellationToken)); + .ToPagedMessagesResult(pagingInfo, sortInfo, token), cancellationToken); public Task>> GetAllMessagesForSearch(string searchTerms, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange? timeSentRange = null, CancellationToken cancellationToken = default) => - ExecuteWithDbContext(dbContext => Search(dbContext.FailedMessages.AsNoTracking(), searchTerms) + ExecuteWithDbContext((dbContext, token) => Search(dbContext.FailedMessages.AsNoTracking(), searchTerms) .FilterBySentTimeRange(timeSentRange) - .ToPagedMessagesResult(pagingInfo, sortInfo, cancellationToken)); + .ToPagedMessagesResult(pagingInfo, sortInfo, token), cancellationToken); public Task>> SearchEndpointMessages(string endpointName, string searchKeyword, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange? timeSentRange = null, CancellationToken cancellationToken = default) => - ExecuteWithDbContext(dbContext => Search(dbContext.FailedMessages.AsNoTracking(), searchKeyword) + ExecuteWithDbContext((dbContext, token) => Search(dbContext.FailedMessages.AsNoTracking(), searchKeyword) .Where(message => message.ReceivingEndpointName == endpointName) .FilterBySentTimeRange(timeSentRange) - .ToPagedMessagesResult(pagingInfo, sortInfo, cancellationToken)); + .ToPagedMessagesResult(pagingInfo, sortInfo, token), cancellationToken); // Neither search hides system messages: a caller who searched // for something specific is not helped by hiding the message that matched it. diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/MonitoringDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/MonitoringDataStore.cs index 37f07b08f5..101131ffc0 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/MonitoringDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/MonitoringDataStore.cs @@ -8,11 +8,11 @@ namespace ServiceControl.Persistence.EFCore.Implementation; public class MonitoringDataStore(IServiceScopeFactory scopeFactory) : DataStoreBase(scopeFactory), IMonitoringDataStore { public Task CreateIfNotExists(EndpointDetails endpoint, CancellationToken cancellationToken = default) => - ExecuteWithDbContext(async dbContext => + ExecuteWithDbContext(async (dbContext, token) => { var id = endpoint.GetDeterministicId(); - var exists = await dbContext.KnownEndpoints.AnyAsync(e => e.Id == id, cancellationToken); + var exists = await dbContext.KnownEndpoints.AnyAsync(e => e.Id == id, token); if (exists) { return; @@ -31,22 +31,22 @@ public Task CreateIfNotExists(EndpointDetails endpoint, CancellationToken cancel try { - await dbContext.SaveChangesAsync(cancellationToken); + await dbContext.SaveChangesAsync(token); } catch (DbUpdateException) { // A concurrent insert with the same deterministic id may have won the race. // Anything else is a genuine failure. dbContext.Entry(knownEndpoint).State = EntityState.Detached; - if (!await dbContext.KnownEndpoints.AnyAsync(e => e.Id == id, cancellationToken)) + if (!await dbContext.KnownEndpoints.AnyAsync(e => e.Id == id, token)) { throw; } } - }); + }, cancellationToken); public Task CreateOrUpdate(EndpointDetails endpoint, IEndpointInstanceMonitoring endpointInstanceMonitoring, CancellationToken cancellationToken = default) => - ExecuteWithDbContext(async dbContext => + ExecuteWithDbContext(async (dbContext, token) => { var id = endpoint.GetDeterministicId(); await dbContext.UpsertAsync([id], @@ -62,24 +62,24 @@ await dbContext.UpsertAsync([id], { knownEndpoint.Monitored = endpointInstanceMonitoring.IsMonitored(id); }, - cancellationToken + token ); - }); + }, cancellationToken); public Task UpdateEndpointMonitoring(EndpointDetails endpoint, bool isMonitored, CancellationToken cancellationToken = default) => - ExecuteWithDbContext(async dbContext => + ExecuteWithDbContext(async (dbContext, token) => { var id = endpoint.GetDeterministicId(); await dbContext.KnownEndpoints .Where(e => e.Id == id) - .ExecuteUpdateAsync(setters => setters.SetProperty(e => e.Monitored, isMonitored), cancellationToken); - }); + .ExecuteUpdateAsync(setters => setters.SetProperty(e => e.Monitored, isMonitored), token); + }, cancellationToken); public Task WarmupMonitoringFromPersistence(IEndpointInstanceMonitoring endpointInstanceMonitoring, CancellationToken cancellationToken = default) => - ExecuteWithDbContext(async dbContext => + ExecuteWithDbContext(async (dbContext, token) => { - await foreach (var endpoint in dbContext.KnownEndpoints.AsNoTracking().AsAsyncEnumerable().WithCancellation(cancellationToken)) + await foreach (var endpoint in dbContext.KnownEndpoints.AsNoTracking().AsAsyncEnumerable().WithCancellation(token)) { var endpointDetails = new EndpointDetails { @@ -90,21 +90,21 @@ public Task WarmupMonitoringFromPersistence(IEndpointInstanceMonitoring endpoint endpointInstanceMonitoring.DetectEndpointFromPersistentStore(endpointDetails, endpoint.Monitored); } - }); + }, cancellationToken); public Task Delete(Guid endpointId, CancellationToken cancellationToken = default) { - return ExecuteWithDbContext(async dbContext => + return ExecuteWithDbContext(async (dbContext, token) => { await dbContext.KnownEndpoints .Where(e => e.Id == endpointId) - .ExecuteDeleteAsync(cancellationToken); - }); + .ExecuteDeleteAsync(token); + }, cancellationToken); } public Task> GetAllKnownEndpoints(CancellationToken cancellationToken = default) { - return ExecuteWithDbContext>(async dbContext => + return ExecuteWithDbContext>(async (dbContext, token) => await dbContext.KnownEndpoints .AsNoTracking() .Select(e => new KnownEndpoint @@ -118,6 +118,6 @@ await dbContext.KnownEndpoints HostDisplayName = e.Host, Monitored = e.Monitored }) - .ToListAsync(cancellationToken)); + .ToListAsync(token), cancellationToken); } } diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/QueueAddressStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/QueueAddressStore.cs index 441c48bca2..be2448f662 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/QueueAddressStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/QueueAddressStore.cs @@ -8,7 +8,7 @@ namespace ServiceControl.Persistence.EFCore.Implementation; public class QueueAddressStore(IServiceScopeFactory scopeFactory) : DataStoreBase(scopeFactory), IQueueAddressStore { public Task>> GetAddresses(PagingInfo pagingInfo, CancellationToken cancellationToken = default) => - ExecuteWithDbContext(async context => + ExecuteWithDbContext(async (context, token) => { var query = context.FailedMessages .GroupBy(failure => failure.FailingEndpointAddress) @@ -19,9 +19,9 @@ public Task>> GetAddresses(PagingInfo pagingInfo FailedMessageCount = failuresByEndpoint.Count() }); - var items = await query.Skip(pagingInfo.Offset).Take(pagingInfo.PageSize).ToListAsync(cancellationToken); + var items = await query.Skip(pagingInfo.Offset).Take(pagingInfo.PageSize).ToListAsync(token); var eTag = DeterministicGuid.MakeId($"{items.Count}|{string.Join(",", items.Select(x => x.PhysicalAddress))}").ToString(); return new QueryResult>(items, new QueryStatsInfo(eTag, query.Count(), false)); - }); + }, cancellationToken); } \ No newline at end of file diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/RetryBatchStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/RetryBatchStore.cs index a6d94e6520..9d88d8acb4 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/RetryBatchStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/RetryBatchStore.cs @@ -15,7 +15,7 @@ public Task CreateBatch(string retrySessionId, string requestId, RetryTy string? batchName = null, string? classifier = null, string? initiatedById = null, string? initiatedByName = null, string? operationId = null, CancellationToken cancellationToken = default) => - ExecuteWithDbContext(async dbContext => + ExecuteWithDbContext(async (dbContext, token) => { var batch = new RetryBatchEntity { @@ -37,16 +37,16 @@ public Task CreateBatch(string retrySessionId, string requestId, RetryTy dbContext.RetryBatches.Add(batch); - await dbContext.SaveChangesAsync(cancellationToken); + await dbContext.SaveChangesAsync(token); return batch.Id.ToString(); - }); + }, cancellationToken); /// /// Claims the messages for the batch. A message already claimed by another batch keeps that claim, so the batch it is staged with is whichever one got there first. /// public Task AssignMessagesToBatch(string batchId, string[] messageIds, CancellationToken cancellationToken = default) => - ExecuteWithDbContext(async dbContext => + ExecuteWithDbContext(async (dbContext, token) => { var batch = ParseBatchId(batchId); @@ -76,21 +76,21 @@ public Task AssignMessagesToBatch(string batchId, string[] messageIds, Cancellat await strategy.ExecuteAsync(async () => { - await using var transaction = await dbContext.Database.BeginTransactionAsync(cancellationToken); + await using var transaction = await dbContext.Database.BeginTransactionAsync(token); await dialect.InsertMissingRetryClaims(dbContext, claims, CancellationToken.None); - await transaction.CommitAsync(cancellationToken); + await transaction.CommitAsync(token); }); - }); + }, cancellationToken); public Task MoveBatchToStaging(string batchId, CancellationToken cancellationToken = default) { var batch = ParseBatchId(batchId); - return ExecuteWithDbContext(dbContext => dbContext.RetryBatches + return ExecuteWithDbContext((dbContext, token) => dbContext.RetryBatches .Where(row => row.Id == batch) - .ExecuteUpdateAsync(setters => setters.SetProperty(row => row.Status, RetryBatchStatus.Staging), cancellationToken)); + .ExecuteUpdateAsync(setters => setters.SetProperty(row => row.Status, RetryBatchStatus.Staging), token), cancellationToken); } // Batch ids only ever come from CreateBatch, so anything else is a programming error. @@ -100,22 +100,22 @@ static Guid ParseBatchId(string batchId) => : throw new ArgumentException($"'{batchId}' is not a retry batch id issued by this store.", nameof(batchId)); public Task>> GetOrphanedBatches(string retrySessionId, CancellationToken cancellationToken = default) => - ExecuteWithDbContext(async dbContext => + ExecuteWithDbContext(async (dbContext, token) => { var orphaned = await dbContext.RetryBatches .AsNoTracking() .Where(batch => batch.Status == RetryBatchStatus.MarkingDocuments && batch.RetrySessionId != retrySessionId) - .ToListAsync(cancellationToken); + .ToListAsync(token); - var messageCounts = await CountMessages(dbContext, [.. orphaned.Select(batch => batch.Id)], cancellationToken); + var messageCounts = await CountMessages(dbContext, [.. orphaned.Select(batch => batch.Id)], token); IList batches = [.. orphaned.Select(batch => batch.ToRetryBatch(messageCounts.GetValueOrDefault(batch.Id)))]; return new QueryResult>(batches, new QueryStatsInfo(string.Empty, batches.Count, false)); - }); + }, cancellationToken); public Task> GetAvailableBatchGroups(CancellationToken cancellationToken = default) => - ExecuteWithDbContext>(async dbContext => + ExecuteWithDbContext>(async (dbContext, token) => { var groups = await dbContext.RetryBatches .AsNoTracking() @@ -133,7 +133,7 @@ public Task> GetAvailableBatchGroups(CancellationToken ca Originator = group.Max(batch => batch.Originator), Classifier = group.Max(batch => batch.Classifier) }) - .ToListAsync(cancellationToken); + .ToListAsync(token); return [.. groups.Select(group => new RetryBatchGroup { @@ -147,14 +147,14 @@ public Task> GetAvailableBatchGroups(CancellationToken ca Originator = group.Originator, Classifier = group.Classifier })]; - }); + }, cancellationToken); public Task GetCurrentForwardingBatch(CancellationToken cancellationToken = default) => - ExecuteWithDbContext(async dbContext => + ExecuteWithDbContext(async (dbContext, token) => { var nowForwarding = await dbContext.RetryBatchNowForwarding .AsNoTracking() - .SingleOrDefaultAsync(cancellationToken); + .SingleOrDefaultAsync(token); if (nowForwarding == null) { @@ -165,8 +165,8 @@ public Task> GetAvailableBatchGroups(CancellationToken ca .AsNoTracking() .Where(batch => batch.Id == nowForwarding.RetryBatchId) .Select(batch => new ForwardingRetryBatch(batch.RequestId, batch.RetryType, batch.Originator!, batch.Classifier!)) - .SingleOrDefaultAsync(cancellationToken); - }); + .SingleOrDefaultAsync(token); + }, cancellationToken); public Task ForEachUnresolvedMessage(Func callback, CancellationToken cancellationToken = default) => ForEach(Unresolved, callback, cancellationToken); @@ -184,7 +184,7 @@ public Task ForEachUnresolvedMessageInGroup(string groupId, Func dbContext.FailedMessageGroups.Any(group => group.GroupId == groupId && group.FailedMessageUniqueId == message.UniqueMessageId)), callback, cancellationToken); Task ForEach(Func> query, Func callback, CancellationToken cancellationToken) => - ExecuteWithDbContext(dbContext => Stream(query(dbContext), callback, cancellationToken)); + ExecuteWithDbContext((dbContext, token) => Stream(query(dbContext), callback, token), cancellationToken); static IQueryable Unresolved(ServiceControlDbContext dbContext) => dbContext.FailedMessages diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/RetryHistoryDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/RetryHistoryDataStore.cs index a60ed7b963..423328625d 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/RetryHistoryDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/RetryHistoryDataStore.cs @@ -9,7 +9,7 @@ namespace ServiceControl.Persistence.EFCore.Implementation; public class RetryHistoryDataStore(IServiceScopeFactory scopeFactory) : DataStoreBase(scopeFactory), IRetryHistoryDataStore { public Task GetRetryHistory(CancellationToken cancellationToken = default) => - ExecuteWithDbContext(async dbContext => + ExecuteWithDbContext(async (dbContext, token) => { var historicOperations = await dbContext.HistoricRetryOperations .AsNoTracking() @@ -25,7 +25,7 @@ public Task GetRetryHistory(CancellationToken cancellationToken = Failed = operation.Failed, NumberOfMessagesProcessed = operation.NumberOfMessagesProcessed }) - .ToListAsync(cancellationToken); + .ToListAsync(token); var unacknowledgedOperations = await dbContext.UnacknowledgedRetryOperations .AsNoTracking() @@ -41,25 +41,25 @@ public Task GetRetryHistory(CancellationToken cancellationToken = Failed = operation.Failed, NumberOfMessagesProcessed = operation.NumberOfMessagesProcessed }) - .ToListAsync(cancellationToken); + .ToListAsync(token); return new RetryHistory { HistoricOperations = historicOperations, UnacknowledgedOperations = unacknowledgedOperations }; - }); + }, cancellationToken); public Task RecordRetryOperationCompleted(string requestId, RetryType retryType, DateTime startTime, DateTime completionTime, string originator, string classifier, bool messageFailed, int numberOfMessagesProcessed, DateTime lastProcessed, int retryHistoryDepth, CancellationToken cancellationToken = default) => - ExecuteWithDbContext(async dbContext => + ExecuteWithDbContext(async (dbContext, token) => { var strategy = dbContext.Database.CreateExecutionStrategy(); await strategy.ExecuteAsync(async () => { - await using var transaction = await dbContext.Database.BeginTransactionAsync(cancellationToken); + await using var transaction = await dbContext.Database.BeginTransactionAsync(token); dbContext.HistoricRetryOperations.Add(new HistoricRetryOperationEntity { @@ -75,27 +75,27 @@ await strategy.ExecuteAsync(async () => if (NeedsAcknowledgement(retryType)) { await RecordUnacknowledged(dbContext, requestId, retryType, startTime, completionTime, - originator, classifier, messageFailed, numberOfMessagesProcessed, lastProcessed, cancellationToken); + originator, classifier, messageFailed, numberOfMessagesProcessed, lastProcessed, token); } - await dbContext.SaveChangesAsync(cancellationToken); + await dbContext.SaveChangesAsync(token); // After the insert, so the operation just recorded competes for a place in the history. - await TrimHistory(dbContext, retryHistoryDepth, cancellationToken); + await TrimHistory(dbContext, retryHistoryDepth, token); - await transaction.CommitAsync(cancellationToken); + await transaction.CommitAsync(token); }); - }); + }, cancellationToken); public Task AcknowledgeRetryGroup(string groupId, CancellationToken cancellationToken = default) => - ExecuteWithDbContext(async dbContext => + ExecuteWithDbContext(async (dbContext, token) => { var acknowledged = await dbContext.UnacknowledgedRetryOperations .Where(operation => operation.RequestId == groupId && operation.RetryType == RetryType.FailureGroup) - .ExecuteDeleteAsync(cancellationToken); + .ExecuteDeleteAsync(token); return acknowledged > 0; - }); + }, cancellationToken); static bool NeedsAcknowledgement(RetryType retryType) => retryType is not RetryType.SingleMessage and not RetryType.MultipleMessages; diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/RetryStagingStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/RetryStagingStore.cs index eda481014e..79b508c49c 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/RetryStagingStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/RetryStagingStore.cs @@ -10,23 +10,23 @@ namespace ServiceControl.Persistence.EFCore.Implementation; public class RetryStagingStore(IServiceScopeFactory scopeFactory, TimeProvider timeProvider) : DataStoreBase(scopeFactory), IRetryStagingStore { public Task GetStagingBatch(CancellationToken cancellationToken = default) => - ExecuteWithDbContext(async dbContext => + ExecuteWithDbContext(async (dbContext, token) => { var batch = await dbContext.RetryBatches .AsNoTracking() .Where(batch => batch.Status == RetryBatchStatus.Staging) .OrderBy(batch => batch.StartTime) .ThenBy(batch => batch.Id) - .FirstOrDefaultAsync(cancellationToken); + .FirstOrDefaultAsync(token); - return batch?.ToRetryBatch(await CountMessages(dbContext, batch.Id, cancellationToken)); - }); + return batch?.ToRetryBatch(await CountMessages(dbContext, batch.Id, token)); + }, cancellationToken); public Task GetMessagesToStage(string batchId, CancellationToken cancellationToken = default) { var batch = ParseBatchId(batchId); - return ExecuteWithDbContext(async dbContext => + return ExecuteWithDbContext(async (dbContext, token) => { // A message claimed by an earlier batch is not claimed by this one, and a claim whose // message is gone drops out of the join, which is what leaves it out of the staging. @@ -44,7 +44,7 @@ public Task GetMessagesToStage(string batchId, CancellationTok message.HeadersJson, retry.StageAttempts }) - .ToListAsync(cancellationToken); + .ToListAsync(token); return rows.Select(row => new StagingMessage( row.UniqueMessageId.ToString(), @@ -53,7 +53,7 @@ public Task GetMessagesToStage(string batchId, CancellationTok row.FailingEndpointAddress, MessageHeaders.Read(row.HeadersJson), row.StageAttempts)).ToArray(); - }); + }, cancellationToken); } public Task MarkBatchAsForwarding(string batchId, string stagingId, IReadOnlyCollection stagedMessageIds, CancellationToken cancellationToken = default) @@ -62,7 +62,7 @@ public Task MarkBatchAsForwarding(string batchId, string stagingId, IReadOnlyCol var staged = ParseMessageIds(stagedMessageIds); var now = timeProvider.GetUtcNow().UtcDateTime; - return ExecuteWithDbContext(dbContext => + return ExecuteWithDbContext((dbContext, token) => InTransaction(dbContext, async token => { await dbContext.RetryBatches @@ -85,14 +85,14 @@ await dbContext.FailedMessages .SetProperty(row => row.LastModified, now), token); await PointForwarderAt(dbContext, batch, token); - }, cancellationToken)); + }, token), cancellationToken); } public Task DiscardBatch(string batchId, CancellationToken cancellationToken = default) { var batch = ParseBatchId(batchId); - return ExecuteWithDbContext(dbContext => + return ExecuteWithDbContext((dbContext, token) => InTransaction(dbContext, async token => { // Nothing was staged, so every claim of this batch is of a message that is gone. @@ -103,38 +103,38 @@ await dbContext.FailedMessageRetries await dbContext.RetryBatches .Where(row => row.Id == batch) .ExecuteDeleteAsync(token); - }, cancellationToken)); + }, token), cancellationToken); } public Task GetForwardingBatchId(CancellationToken cancellationToken = default) => - ExecuteWithDbContext(async dbContext => + ExecuteWithDbContext(async (dbContext, token) => { var nowForwarding = await dbContext.RetryBatchNowForwarding .AsNoTracking() - .SingleOrDefaultAsync(cancellationToken); + .SingleOrDefaultAsync(token); return nowForwarding?.RetryBatchId.ToString(); - }); + }, cancellationToken); public Task GetBatch(string batchId, CancellationToken cancellationToken = default) { var batch = ParseBatchId(batchId); - return ExecuteWithDbContext(async dbContext => + return ExecuteWithDbContext(async (dbContext, token) => { var entity = await dbContext.RetryBatches .AsNoTracking() - .SingleOrDefaultAsync(row => row.Id == batch, cancellationToken); + .SingleOrDefaultAsync(row => row.Id == batch, token); - return entity?.ToRetryBatch(await CountMessages(dbContext, batch, cancellationToken)); - }); + return entity?.ToRetryBatch(await CountMessages(dbContext, batch, token)); + }, cancellationToken); } public Task CompleteForwarding(string batchId, CancellationToken cancellationToken = default) { var batch = ParseBatchId(batchId); - return ExecuteWithDbContext(dbContext => + return ExecuteWithDbContext((dbContext, token) => InTransaction(dbContext, async token => { // The claims outlive the batch: they are what stops a message being staged again @@ -146,16 +146,16 @@ await dbContext.RetryBatches await dbContext.RetryBatchNowForwarding .Where(row => row.RetryBatchId == batch) .ExecuteDeleteAsync(token); - }, cancellationToken)); + }, token), cancellationToken); } public Task RecordStagingFailure(IReadOnlyCollection uniqueMessageIds, CancellationToken cancellationToken = default) { var failed = ParseMessageIds(uniqueMessageIds); - return ExecuteWithDbContext(dbContext => dbContext.FailedMessageRetries + return ExecuteWithDbContext((dbContext, token) => dbContext.FailedMessageRetries .Where(row => failed.Contains(row.UniqueMessageId)) - .ExecuteUpdateAsync(setters => setters.SetProperty(row => row.StageAttempts, 1), cancellationToken)); + .ExecuteUpdateAsync(setters => setters.SetProperty(row => row.StageAttempts, 1), token), cancellationToken); } public Task IncrementStagingAttempts(string uniqueMessageId, CancellationToken cancellationToken = default) @@ -165,9 +165,9 @@ public Task IncrementStagingAttempts(string uniqueMessageId, CancellationToken c return Task.CompletedTask; } - return ExecuteWithDbContext(dbContext => dbContext.FailedMessageRetries + return ExecuteWithDbContext((dbContext, token) => dbContext.FailedMessageRetries .Where(row => row.UniqueMessageId == message) - .ExecuteUpdateAsync(setters => setters.SetProperty(row => row.StageAttempts, row => row.StageAttempts + 1), cancellationToken)); + .ExecuteUpdateAsync(setters => setters.SetProperty(row => row.StageAttempts, row => row.StageAttempts + 1), token), cancellationToken); } public Task RemoveFromBatch(string uniqueMessageId, CancellationToken cancellationToken = default) @@ -177,9 +177,9 @@ public Task RemoveFromBatch(string uniqueMessageId, CancellationToken cancellati return Task.CompletedTask; } - return ExecuteWithDbContext(dbContext => dbContext.FailedMessageRetries + return ExecuteWithDbContext((dbContext, token) => dbContext.FailedMessageRetries .Where(row => row.UniqueMessageId == message) - .ExecuteDeleteAsync(cancellationToken)); + .ExecuteDeleteAsync(token), cancellationToken); } static async Task PointForwarderAt(ServiceControlDbContext dbContext, Guid batch, CancellationToken cancellationToken) diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/SubscriptionStorage.cs b/src/ServiceControl.Persistence.EFCore/Implementation/SubscriptionStorage.cs index 337e3fe53d..128c1543bc 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/SubscriptionStorage.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/SubscriptionStorage.cs @@ -44,11 +44,11 @@ public async Task Subscribe(Subscriber subscriber, MessageType messageType, Cont //When the subscriber is running V6 and UseLegacyMessageDrivenSubscriptionMode is enabled at the subscriber the 'subscriber.Endpoint' value is null var endpoint = subscriber.Endpoint ?? transportAddress.Split('@').First(); - await ExecuteWithDbContext(dbContext => dbContext.UpsertAsync( + await ExecuteWithDbContext((dbContext, token) => dbContext.UpsertAsync( [typeName, transportAddress], () => new SubscriptionEntity { MessageType = typeName, TransportAddress = transportAddress, Endpoint = endpoint }, entity => entity.Endpoint = endpoint, - cancellationToken)); + token), cancellationToken); InvalidateCache(typeName); } @@ -58,9 +58,9 @@ public async Task Unsubscribe(Subscriber subscriber, MessageType messageType, Co var typeName = messageType.TypeName; var transportAddress = subscriber.TransportAddress; - await ExecuteWithDbContext(dbContext => dbContext.Subscriptions + await ExecuteWithDbContext((dbContext, token) => dbContext.Subscriptions .Where(subscription => subscription.MessageType == typeName && subscription.TransportAddress == transportAddress) - .ExecuteDeleteAsync(cancellationToken)); + .ExecuteDeleteAsync(token), cancellationToken); InvalidateCache(typeName); } @@ -83,12 +83,12 @@ public async Task> GetSubscriberAddressesForMessage(IEnu async Task LoadSubscribers(string[] typeNames, CancellationToken cancellationToken) { - var stored = await ExecuteWithDbContext(dbContext => dbContext.Subscriptions + var stored = await ExecuteWithDbContext((dbContext, token) => dbContext.Subscriptions .AsNoTracking() .Where(subscription => typeNames.Contains(subscription.MessageType)) .Select(subscription => new { subscription.TransportAddress, subscription.Endpoint }) .Distinct() - .ToArrayAsync(cancellationToken)); + .ToArrayAsync(token), cancellationToken); var subscribers = stored.Select(subscription => new Subscriber(subscription.TransportAddress, subscription.Endpoint)); diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/TrialLicenseDataProvider.cs b/src/ServiceControl.Persistence.EFCore/Implementation/TrialLicenseDataProvider.cs index 5e45712e44..222f1c87ef 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/TrialLicenseDataProvider.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/TrialLicenseDataProvider.cs @@ -6,10 +6,10 @@ namespace ServiceControl.Persistence.EFCore.Implementation; public class TrialLicenseDataProvider(IServiceScopeFactory scopeFactory) : DataStoreBase(scopeFactory), ITrialLicenseDataProvider { public Task GetTrialEndDate(CancellationToken cancellationToken = default) - => ExecuteWithDbContext(context => - context.GetSetting(SettingKeys.TrialEndDate, cancellationToken)); + => ExecuteWithDbContext((context, token) => + context.GetSetting(SettingKeys.TrialEndDate, token), cancellationToken); public Task StoreTrialEndDate(DateOnly trialEndDate, CancellationToken cancellationToken = default) - => ExecuteWithDbContext(context => - context.StoreSetting(SettingKeys.TrialEndDate, trialEndDate, cancellationToken)); + => ExecuteWithDbContext((context, token) => + context.StoreSetting(SettingKeys.TrialEndDate, trialEndDate, token), cancellationToken); } diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/ExpectedLengthStream.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/ExpectedLengthStream.cs index 914c66995c..5d09f35b37 100644 --- a/src/ServiceControl.Persistence.EFCore/Infrastructure/ExpectedLengthStream.cs +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/ExpectedLengthStream.cs @@ -30,7 +30,7 @@ public override long Position public override int Read(Span buffer) => ValidateRead(stream.Read(buffer)); - public override async Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) => + public override async Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken = default) => ValidateRead(await stream.ReadAsync(buffer, offset, count, cancellationToken)); public override async ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken = default) => diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/IFailedMessageIngestionSqlDialect.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/IFailedMessageIngestionSqlDialect.cs index c43273546a..77bdd8aa19 100644 --- a/src/ServiceControl.Persistence.EFCore/Infrastructure/IFailedMessageIngestionSqlDialect.cs +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/IFailedMessageIngestionSqlDialect.cs @@ -16,16 +16,16 @@ public interface IFailedMessageIngestionSqlDialect /// attempt counter advances (unless the batch merely redelivered the attempt already stored), /// the failure window widens, and the newer attempt supplies every payload column. /// - Task UpsertFailedMessages(ServiceControlDbContext dbContext, IReadOnlyList rows, CancellationToken cancellationToken); + Task UpsertFailedMessages(ServiceControlDbContext dbContext, IReadOnlyList rows, CancellationToken cancellationToken = default); /// /// Insert if absent. The caller has already deleted these messages' group rows in the same /// transaction; if-absent keeps a concurrent writer's identical row from failing us. /// - Task InsertGroups(ServiceControlDbContext dbContext, IReadOnlyList rows, CancellationToken cancellationToken); + Task InsertGroups(ServiceControlDbContext dbContext, IReadOnlyList rows, CancellationToken cancellationToken = default); /// /// Insert if absent, never update: existing endpoints keep their Monitored flag. /// - Task InsertMissingKnownEndpoints(ServiceControlDbContext dbContext, IReadOnlyList rows, CancellationToken cancellationToken); + Task InsertMissingKnownEndpoints(ServiceControlDbContext dbContext, IReadOnlyList rows, CancellationToken cancellationToken = default); } diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/IRetryBatchSqlDialect.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/IRetryBatchSqlDialect.cs index f1c0bbbb87..e3125d255b 100644 --- a/src/ServiceControl.Persistence.EFCore/Infrastructure/IRetryBatchSqlDialect.cs +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/IRetryBatchSqlDialect.cs @@ -12,5 +12,5 @@ public interface IRetryBatchSqlDialect /// Insert if absent, never update: a message already claimed stays with the batch that claimed /// it first, so two retry requests covering the same message cannot both stage it. /// - Task InsertMissingRetryClaims(ServiceControlDbContext dbContext, IReadOnlyList rows, CancellationToken cancellationToken); + Task InsertMissingRetryClaims(ServiceControlDbContext dbContext, IReadOnlyList rows, CancellationToken cancellationToken = default); } diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/InsertOnlyTableReconciler.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/InsertOnlyTableReconciler.cs index 643b146219..2f98a350cf 100644 --- a/src/ServiceControl.Persistence.EFCore/Infrastructure/InsertOnlyTableReconciler.cs +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/InsertOnlyTableReconciler.cs @@ -14,13 +14,13 @@ public abstract class InsertOnlyTableReconciler( { protected const int BatchSize = 1000; - protected override async Task ExecuteAsync(CancellationToken stoppingToken) + protected override async Task ExecuteAsync(CancellationToken cancellationToken = default) { logger.LogInformation("Starting {ServiceName}", serviceName); try { - await Task.Delay(TimeSpan.FromSeconds(30), timeProvider, stoppingToken); + await Task.Delay(TimeSpan.FromSeconds(30), timeProvider, cancellationToken); using PeriodicTimer timer = new(TimeSpan.FromSeconds(30), timeProvider); @@ -28,15 +28,18 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken) { try { - await Reconcile(pace: true, stoppingToken); + await Reconcile(pace: true, cancellationToken); } +#pragma warning disable PS0019 // The filter already excludes OperationCanceledException, so + // cancellation propagates; PS0019 only recognises a cancellationToken guard. catch (Exception ex) when (ex is not OperationCanceledException) { logger.LogError(ex, "Error during {ServiceName} reconciliation", serviceName); } - } while (await timer.WaitForNextTickAsync(stoppingToken)); +#pragma warning restore PS0019 + } while (await timer.WaitForNextTickAsync(cancellationToken)); } - catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { logger.LogInformation("Stopping {ServiceName}", serviceName); } @@ -76,5 +79,5 @@ async Task Reconcile(bool pace, CancellationToken cancellationToken) } } - protected abstract Task ReconcileBatch(ServiceControlDbContext dbContext, CancellationToken stoppingToken); + protected abstract Task ReconcileBatch(ServiceControlDbContext dbContext, CancellationToken cancellationToken = default); } diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/OwnedStream.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/OwnedStream.cs index be265364fb..77b3f0693a 100644 --- a/src/ServiceControl.Persistence.EFCore/Infrastructure/OwnedStream.cs +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/OwnedStream.cs @@ -23,13 +23,13 @@ public override long Position public override void Flush() => stream.Flush(); - public override Task FlushAsync(CancellationToken cancellationToken) => stream.FlushAsync(cancellationToken); + public override Task FlushAsync(CancellationToken cancellationToken = default) => stream.FlushAsync(cancellationToken); public override int Read(byte[] buffer, int offset, int count) => stream.Read(buffer, offset, count); public override int Read(Span buffer) => stream.Read(buffer); - public override Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) => + public override Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken = default) => stream.ReadAsync(buffer, offset, count, cancellationToken); public override ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken = default) => @@ -43,7 +43,7 @@ public override ValueTask ReadAsync(Memory buffer, CancellationToken public override void Write(ReadOnlySpan buffer) => stream.Write(buffer); - public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) => + public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken = default) => stream.WriteAsync(buffer, offset, count, cancellationToken); public override ValueTask WriteAsync(ReadOnlyMemory buffer, CancellationToken cancellationToken = default) => diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/RetentionSweeper.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/RetentionSweeper.cs index d5e25fec02..07811c4ad3 100644 --- a/src/ServiceControl.Persistence.EFCore/Infrastructure/RetentionSweeper.cs +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/RetentionSweeper.cs @@ -24,13 +24,13 @@ public class RetentionSweeper( static readonly TimeSpan InitialDelay = TimeSpan.FromMinutes(1); static readonly TimeSpan BatchPause = TimeSpan.FromSeconds(1); - protected override async Task ExecuteAsync(CancellationToken stoppingToken) + protected override async Task ExecuteAsync(CancellationToken cancellationToken = default) { logger.LogInformation("Starting retention sweep"); try { - await Task.Delay(InitialDelay, timeProvider, stoppingToken); + await Task.Delay(InitialDelay, timeProvider, cancellationToken); using PeriodicTimer timer = new(Interval, timeProvider); @@ -38,15 +38,18 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken) { try { - await Sweep(pace: true, stoppingToken); + await Sweep(pace: true, cancellationToken); } +#pragma warning disable PS0019 // The filter already excludes OperationCanceledException, so + // cancellation propagates; PS0019 only recognises a cancellationToken guard. catch (Exception ex) when (ex is not OperationCanceledException) { logger.LogError(ex, "Error during retention sweep"); } - } while (await timer.WaitForNextTickAsync(stoppingToken)); +#pragma warning restore PS0019 + } while (await timer.WaitForNextTickAsync(cancellationToken)); } - catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { logger.LogInformation("Stopping retention sweep"); } @@ -162,11 +165,13 @@ async Task DeleteExternalBody(Guid uniqueMessageId, CancellationToken cancellati { await bodyStorage.DeleteBody(uniqueMessageId.ToString(), cancellationToken); } +#pragma warning disable PS0019 // As above: the filter excludes cancellation already. catch (Exception ex) when (ex is not OperationCanceledException) { // Retention must not stall on a missing or unavailable body. logger.LogWarning(ex, "Could not delete the external body for {UniqueMessageId} during retention", uniqueMessageId); } +#pragma warning restore PS0019 } static System.Linq.Expressions.Expression> IsExpired(DateTime cutoff) => diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/SettingsStore.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/SettingsStore.cs index 431b646f00..ab4c6813d4 100644 --- a/src/ServiceControl.Persistence.EFCore/Infrastructure/SettingsStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/SettingsStore.cs @@ -9,7 +9,7 @@ namespace ServiceControl.Persistence.EFCore.Infrastructure; // written whole, never queried by their contents, so each is stored as a JSON document. static class SettingsStore { - public static async Task GetSetting(this ServiceControlDbContext dbContext, string key, CancellationToken cancellationToken) + public static async Task GetSetting(this ServiceControlDbContext dbContext, string key, CancellationToken cancellationToken = default) { var value = await dbContext.Settings .AsNoTracking() @@ -20,7 +20,7 @@ static class SettingsStore return value is null ? default : JsonSerializer.Deserialize(value); } - public static Task StoreSetting(this ServiceControlDbContext dbContext, string key, T value, CancellationToken cancellationToken) + public static Task StoreSetting(this ServiceControlDbContext dbContext, string key, T value, CancellationToken cancellationToken = default) { var json = JsonSerializer.Serialize(value); diff --git a/src/ServiceControl.Persistence.RavenDB/.editorconfig b/src/ServiceControl.Persistence.RavenDB/.editorconfig index 5642cb469b..d84391c0d5 100644 --- a/src/ServiceControl.Persistence.RavenDB/.editorconfig +++ b/src/ServiceControl.Persistence.RavenDB/.editorconfig @@ -3,11 +3,9 @@ # Justification: ServiceControl app has no synchronization context dotnet_diagnostic.CA2007.severity = none -# Cancellation analyzer debt. These fire because Particular.Analyzers is no longer pinned to 0.9.0. -# They are scheduled work, not accepted exceptions: remove a line once this project has no -# violations of that rule left, and never add a rule back to this list. +# Cancellation analyzer debt, now confined to the licensing data store. ILicensingDataStore belongs to +# the licensing phase, not the persistence one, so its 17 optional-token additions land with the rest +# of Particular.LicensingComponent. Delete this block then; nothing else in the project violates a +# cancellation rule. +[Throughput/LicensingDataStore.cs] dotnet_diagnostic.PS0003.severity = none -dotnet_diagnostic.PS0013.severity = none -dotnet_diagnostic.PS0018.severity = none -dotnet_diagnostic.PS0020.severity = none -dotnet_diagnostic.PS0021.severity = none diff --git a/src/ServiceControl.Persistence.RavenDB/DatabaseSetup.cs b/src/ServiceControl.Persistence.RavenDB/DatabaseSetup.cs index 4b71a0a43b..147c2393ac 100644 --- a/src/ServiceControl.Persistence.RavenDB/DatabaseSetup.cs +++ b/src/ServiceControl.Persistence.RavenDB/DatabaseSetup.cs @@ -13,7 +13,7 @@ namespace ServiceControl.Persistence.RavenDB class DatabaseSetup(RavenPersisterSettings settings, IDocumentStore documentStore) { - public async Task Execute(CancellationToken cancellationToken) + public async Task Execute(CancellationToken cancellationToken = default) { await CreateDatabase(settings.DatabaseName, cancellationToken); await CreateDatabase(settings.ThroughputDatabaseName, cancellationToken); diff --git a/src/ServiceControl.Persistence.RavenDB/LicenseStatusCheck.cs b/src/ServiceControl.Persistence.RavenDB/LicenseStatusCheck.cs index 6b7ff4736e..ea0a3da4de 100644 --- a/src/ServiceControl.Persistence.RavenDB/LicenseStatusCheck.cs +++ b/src/ServiceControl.Persistence.RavenDB/LicenseStatusCheck.cs @@ -10,7 +10,7 @@ static class LicenseStatusCheck { record LicenseStatusFragment(string Id, string LicensedTo, string Status, bool Expired); - public static async Task WaitForLicenseOrThrow(IDocumentStore documentStore, CancellationToken cancellationToken) + public static async Task WaitForLicenseOrThrow(IDocumentStore documentStore, CancellationToken cancellationToken = default) { var ravenConfiguredHttpClient = documentStore.GetRequestExecutor().HttpClient; var licenseCheckUrl = documentStore.Urls[0].TrimEnd('/') + "/license/status"; @@ -20,10 +20,10 @@ public static async Task WaitForLicenseOrThrow(IDocumentStore documentStore, Can try { - while (!cts.IsCancellationRequested) + while (!cts.Token.IsCancellationRequested) { - var httpResponse = await ravenConfiguredHttpClient.GetAsync(licenseCheckUrl, cancellationToken); - var licenseStatus = await httpResponse.Content.ReadFromJsonAsync(cancellationToken); + var httpResponse = await ravenConfiguredHttpClient.GetAsync(licenseCheckUrl, cts.Token); + var licenseStatus = await httpResponse.Content.ReadFromJsonAsync(cts.Token); if (licenseStatus.Expired) { throw new InvalidOperationException("The current RavenDB license is expired. Please, contact support"); @@ -37,9 +37,14 @@ public static async Task WaitForLicenseOrThrow(IDocumentStore documentStore, Can await Task.Delay(200, cts.Token); } } + // Every call inside the try uses the linked cts.Token, so PS0020 cannot see that this filter + // tests the right thing: cancellation that is not the caller's is the 30s license-check + // timeout, and that is what the friendly message is for. +#pragma warning disable PS0020 catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested) { throw new InvalidOperationException("Cannot validate the current RavenDB license. Please, contact support"); } +#pragma warning restore PS0020 } } \ No newline at end of file diff --git a/src/ServiceControl.Persistence.RavenDB/RavenEmbeddedPersistenceLifecycle.cs b/src/ServiceControl.Persistence.RavenDB/RavenEmbeddedPersistenceLifecycle.cs index f0172acff7..34af6b3f7a 100644 --- a/src/ServiceControl.Persistence.RavenDB/RavenEmbeddedPersistenceLifecycle.cs +++ b/src/ServiceControl.Persistence.RavenDB/RavenEmbeddedPersistenceLifecycle.cs @@ -32,7 +32,7 @@ public async ValueTask GetDocumentStore(CancellationToken cancel } } - public async Task Initialize(CancellationToken cancellationToken) + public async Task Initialize(CancellationToken cancellationToken = default) { try { @@ -69,7 +69,7 @@ public async Task Initialize(CancellationToken cancellationToken) } } - public async Task Stop(CancellationToken cancellationToken) + public async Task Stop(CancellationToken cancellationToken = default) { if (database != null) { diff --git a/src/ServiceControl.Persistence.RavenDB/RavenExternalPersistenceLifecycle.cs b/src/ServiceControl.Persistence.RavenDB/RavenExternalPersistenceLifecycle.cs index a6c8d265f9..690d373b5e 100644 --- a/src/ServiceControl.Persistence.RavenDB/RavenExternalPersistenceLifecycle.cs +++ b/src/ServiceControl.Persistence.RavenDB/RavenExternalPersistenceLifecycle.cs @@ -29,7 +29,7 @@ public async ValueTask GetDocumentStore(CancellationToken cancel } } - public async Task Initialize(CancellationToken cancellationToken) + public async Task Initialize(CancellationToken cancellationToken = default) { try { @@ -59,7 +59,7 @@ public async Task Initialize(CancellationToken cancellationToken) } } - public Task Stop(CancellationToken cancellationToken) => Task.CompletedTask; + public Task Stop(CancellationToken cancellationToken = default) => Task.CompletedTask; public void Dispose() => documentStore?.Dispose(); diff --git a/src/ServiceControl.Persistence.RavenDB/RavenPersistenceLifecycleHostedService.cs b/src/ServiceControl.Persistence.RavenDB/RavenPersistenceLifecycleHostedService.cs index 57abf7d7ca..f7e59b09ad 100644 --- a/src/ServiceControl.Persistence.RavenDB/RavenPersistenceLifecycleHostedService.cs +++ b/src/ServiceControl.Persistence.RavenDB/RavenPersistenceLifecycleHostedService.cs @@ -8,8 +8,8 @@ namespace ServiceControl.Persistence.RavenDB class RavenPersistenceLifecycleHostedService(IRavenPersistenceLifecycle persistenceLifecycle) : IHostedService { - public Task StartAsync(CancellationToken cancellationToken) => persistenceLifecycle.Initialize(cancellationToken); + public Task StartAsync(CancellationToken cancellationToken = default) => persistenceLifecycle.Initialize(cancellationToken); - public Task StopAsync(CancellationToken cancellationToken) => persistenceLifecycle.Stop(cancellationToken); + public Task StopAsync(CancellationToken cancellationToken = default) => persistenceLifecycle.Stop(cancellationToken); } } \ No newline at end of file diff --git a/src/ServiceControl.Persistence.Tests/.editorconfig b/src/ServiceControl.Persistence.Tests/.editorconfig index 3fca48a527..b26fbddcce 100644 --- a/src/ServiceControl.Persistence.Tests/.editorconfig +++ b/src/ServiceControl.Persistence.Tests/.editorconfig @@ -3,10 +3,12 @@ # Justification: Test project dotnet_diagnostic.CA2007.severity = none -# Cancellation analyzer debt. These fire because Particular.Analyzers is no longer pinned to 0.9.0. -# They are scheduled work, not accepted exceptions: remove a line once this project has no -# violations of that rule left, and never add a rule back to this list. -dotnet_diagnostic.PS0003.severity = none -dotnet_diagnostic.PS0006.severity = none -dotnet_diagnostic.PS0013.severity = none -dotnet_diagnostic.PS0018.severity = none +# Justification: Test project. Demanding a CancellationToken on every [Test] method, and on the +# helpers they call, buys nothing: a test that hangs fails the run either way, and [CancelAfter] +# only bites once the code under test honours the token. Persistence integration tests that need a +# real timeout use [Test, CancelAfter(...)] with an injected token instead, which is the pattern to +# follow for new tests. These are accepted exceptions, not scheduled work. +dotnet_diagnostic.PS0003.severity = none # A parameter of type CancellationToken on a non-private delegate or method should be optional +dotnet_diagnostic.PS0006.severity = none # Pass CancellationToken.None instead of the default literal +dotnet_diagnostic.PS0013.severity = none # A Func used as a method parameter with a Task return type argument should have a CancellationToken type argument +dotnet_diagnostic.PS0018.severity = none # A task-returning method should have a CancellationToken parameter diff --git a/src/ServiceControl.Persistence/.editorconfig b/src/ServiceControl.Persistence/.editorconfig index ef7d8e541d..ff993b49bb 100644 --- a/src/ServiceControl.Persistence/.editorconfig +++ b/src/ServiceControl.Persistence/.editorconfig @@ -2,11 +2,3 @@ # Justification: ServiceControl app has no synchronization context dotnet_diagnostic.CA2007.severity = none - -# Cancellation analyzer debt. These fire because Particular.Analyzers is no longer pinned to 0.9.0. -# They are scheduled work, not accepted exceptions: remove a line once this project has no -# violations of that rule left, and never add a rule back to this list. -dotnet_diagnostic.PS0003.severity = none -dotnet_diagnostic.PS0013.severity = none -dotnet_diagnostic.PS0017.severity = none -dotnet_diagnostic.PS0018.severity = none From 3d6365bcb73b3873cef08fd9c1b170f82da7cae0 Mon Sep 17 00:00:00 2001 From: John Simons Date: Wed, 12 Aug 2026 09:02:09 +1000 Subject: [PATCH 4/6] Propagate cancellation tokens through licensing data store persistence This change continues the effort to consistently apply optional `CancellationToken` parameters to asynchronous methods within the `ILicensingDataStore` interface, along with its In-Memory, EFCore, and RavenDB implementations. Cancellation tokens are now propagated to underlying asynchronous database calls, ensuring proper responsiveness to cancellation and retiring the remaining cancellation analyzer debt for these projects. --- .../.editorconfig | 5 --- .../InMemoryLicensingDataStore.cs | 34 +++++++++---------- .../.editorconfig | 5 --- .../ILicensingDataStore.cs | 34 +++++++++---------- .../.editorconfig | 7 ---- .../Implementation/LicensingDataStore.cs | 30 ++++++++-------- .../.editorconfig | 7 ---- .../Throughput/LicensingDataStore.cs | 34 +++++++++---------- 8 files changed, 66 insertions(+), 90 deletions(-) diff --git a/src/Particular.LicensingComponent.Persistence.InMemory/.editorconfig b/src/Particular.LicensingComponent.Persistence.InMemory/.editorconfig index 8a988261dd..28e30f9f38 100644 --- a/src/Particular.LicensingComponent.Persistence.InMemory/.editorconfig +++ b/src/Particular.LicensingComponent.Persistence.InMemory/.editorconfig @@ -9,8 +9,3 @@ dotnet_diagnostic.CA2007.severity = none dotnet_diagnostic.IDE0040.severity = none csharp_style_var_elsewhere = true:error - -# Cancellation analyzer debt. These fire because Particular.Analyzers is no longer pinned to 0.9.0. -# They are scheduled work, not accepted exceptions: remove a line once this project has no -# violations of that rule left, and never add a rule back to this list. -dotnet_diagnostic.PS0003.severity = none diff --git a/src/Particular.LicensingComponent.Persistence.InMemory/InMemoryLicensingDataStore.cs b/src/Particular.LicensingComponent.Persistence.InMemory/InMemoryLicensingDataStore.cs index 46639e6305..0b7c941202 100644 --- a/src/Particular.LicensingComponent.Persistence.InMemory/InMemoryLicensingDataStore.cs +++ b/src/Particular.LicensingComponent.Persistence.InMemory/InMemoryLicensingDataStore.cs @@ -17,7 +17,7 @@ public class InMemoryLicensingDataStore : ILicensingDataStore List reportMasks = []; LicensedEndpointDetails? endpointDetails = null; - public Task> GetAllEndpoints(bool includePlatformEndpoints, CancellationToken cancellationToken) + public Task> GetAllEndpoints(bool includePlatformEndpoints, CancellationToken cancellationToken = default) { var filteredEndpoints = includePlatformEndpoints ? endpoints @@ -27,7 +27,7 @@ public Task> GetAllEndpoints(bool includePlatformEndpoints return Task.FromResult(filteredEndpoints); } - public Task GetEndpoint(EndpointIdentifier id, CancellationToken cancellationToken) + public Task GetEndpoint(EndpointIdentifier id, CancellationToken cancellationToken = default) { if (endpoints.TryGetValue(id, out var endpoint)) { @@ -39,7 +39,7 @@ public Task> GetAllEndpoints(bool includePlatformEndpoints return Task.FromResult(endpoint); } - public Task> GetEndpoints(IList endpointIds, CancellationToken cancellationToken) + public Task> GetEndpoints(IList endpointIds, CancellationToken cancellationToken = default) { var endpointLookup = endpointIds.Select(id => (id, endpoints.TryGetValue(id, out var endpoint) ? endpoint : null)); @@ -57,7 +57,7 @@ public Task> GetAllEndpoints(bool includePlatformEndpoints return Task.FromResult(endpointLookup); } - public Task SaveEndpoint(Endpoint endpoint, CancellationToken cancellationToken) + public Task SaveEndpoint(Endpoint endpoint, CancellationToken cancellationToken = default) { if (endpoints.TryGetValue(endpoint.Id, out var existingEndpoint)) { @@ -68,7 +68,7 @@ public Task SaveEndpoint(Endpoint endpoint, CancellationToken cancellationToken) return Task.CompletedTask; } - public Task>> GetEndpointThroughputByQueueName(IList queueNames, CancellationToken cancellationToken) + public Task>> GetEndpointThroughputByQueueName(IList queueNames, CancellationToken cancellationToken = default) { var result = endpoints .Where(endpoint => queueNames.Contains(endpoint.SanitizedName)) @@ -82,7 +82,7 @@ public Task>> GetEndpointThrough return Task.FromResult((IDictionary>)result); } - public async Task RecordEndpointThroughput(string endpointName, ThroughputSource throughputSource, IList throughput, CancellationToken cancellationToken) + public async Task RecordEndpointThroughput(string endpointName, ThroughputSource throughputSource, IList throughput, CancellationToken cancellationToken = default) { var id = new EndpointIdentifier(endpointName, throughputSource); if (!endpoints.TryGetValue(id, out _)) @@ -111,7 +111,7 @@ public async Task RecordEndpointThroughput(string endpointName, ThroughputSource await Task.CompletedTask; } - public async Task UpdateUserIndicatorOnEndpoints(List userIndicatorUpdates, CancellationToken cancellationToken) + public async Task UpdateUserIndicatorOnEndpoints(List userIndicatorUpdates, CancellationToken cancellationToken = default) { userIndicatorUpdates.ForEach(e => { @@ -130,12 +130,12 @@ public async Task UpdateUserIndicatorOnEndpoints(List userI await Task.CompletedTask; } - public async Task IsThereThroughputForLastXDays(int days, CancellationToken cancellationToken) => await Task.FromResult( + public async Task IsThereThroughputForLastXDays(int days, CancellationToken cancellationToken = default) => await Task.FromResult( allThroughput.Any(endpointThroughput => endpointThroughput.Value.Any( t => t.Key >= DateOnly.FromDateTime(DateTime.UtcNow).AddDays(-days) && t.Key <= DateOnly.FromDateTime(DateTime.UtcNow).AddDays(-1)))); - public async Task IsThereThroughputForLastXDaysForSource(int days, ThroughputSource throughputSource, bool includeToday, CancellationToken cancellationToken) + public async Task IsThereThroughputForLastXDaysForSource(int days, ThroughputSource throughputSource, bool includeToday, CancellationToken cancellationToken = default) { var endDate = includeToday ? DateOnly.FromDateTime(DateTime.UtcNow) : DateOnly.FromDateTime(DateTime.UtcNow).AddDays(-1); @@ -147,33 +147,33 @@ public async Task IsThereThroughputForLastXDaysForSource(int days, Through List GetAllConnectedEndpoints(string name) => endpoints.Where(w => w.SanitizedName == name || w.Id.Name == name).ToList(); - public Task GetBrokerMetadata(CancellationToken cancellationToken) => Task.FromResult(brokerMetadata); + public Task GetBrokerMetadata(CancellationToken cancellationToken = default) => Task.FromResult(brokerMetadata); - public Task SaveBrokerMetadata(BrokerMetadata brokerMetadata, CancellationToken cancellationToken) + public Task SaveBrokerMetadata(BrokerMetadata brokerMetadata, CancellationToken cancellationToken = default) { this.brokerMetadata = brokerMetadata; return Task.CompletedTask; } - public Task GetAuditServiceMetadata(CancellationToken cancellationToken) => Task.FromResult(auditServiceMetadata); + public Task GetAuditServiceMetadata(CancellationToken cancellationToken = default) => Task.FromResult(auditServiceMetadata); - public Task SaveAuditServiceMetadata(AuditServiceMetadata auditServiceMetadata, CancellationToken cancellationToken) + public Task SaveAuditServiceMetadata(AuditServiceMetadata auditServiceMetadata, CancellationToken cancellationToken = default) { this.auditServiceMetadata = auditServiceMetadata; return Task.CompletedTask; } - public Task> GetReportMasks(CancellationToken cancellationToken) => Task.FromResult(reportMasks); + public Task> GetReportMasks(CancellationToken cancellationToken = default) => Task.FromResult(reportMasks); - public Task SaveReportMasks(List reportMasks, CancellationToken cancellationToken) + public Task SaveReportMasks(List reportMasks, CancellationToken cancellationToken = default) { this.reportMasks = reportMasks; return Task.CompletedTask; } - public Task GetLicensedEndpointDetails(CancellationToken cancellationToken) => Task.FromResult(endpointDetails); + public Task GetLicensedEndpointDetails(CancellationToken cancellationToken = default) => Task.FromResult(endpointDetails); - public Task SaveLicensedEndpointDetails(LicensedEndpointDetails result, CancellationToken cancellationToken) + public Task SaveLicensedEndpointDetails(LicensedEndpointDetails result, CancellationToken cancellationToken = default) { endpointDetails = result; return Task.CompletedTask; diff --git a/src/Particular.LicensingComponent.Persistence/.editorconfig b/src/Particular.LicensingComponent.Persistence/.editorconfig index eae5d84087..ff993b49bb 100644 --- a/src/Particular.LicensingComponent.Persistence/.editorconfig +++ b/src/Particular.LicensingComponent.Persistence/.editorconfig @@ -2,8 +2,3 @@ # Justification: ServiceControl app has no synchronization context dotnet_diagnostic.CA2007.severity = none - -# Cancellation analyzer debt. These fire because Particular.Analyzers is no longer pinned to 0.9.0. -# They are scheduled work, not accepted exceptions: remove a line once this project has no -# violations of that rule left, and never add a rule back to this list. -dotnet_diagnostic.PS0003.severity = none diff --git a/src/Particular.LicensingComponent.Persistence/ILicensingDataStore.cs b/src/Particular.LicensingComponent.Persistence/ILicensingDataStore.cs index 3a73a50891..135115d51d 100644 --- a/src/Particular.LicensingComponent.Persistence/ILicensingDataStore.cs +++ b/src/Particular.LicensingComponent.Persistence/ILicensingDataStore.cs @@ -4,39 +4,39 @@ public interface ILicensingDataStore { - Task> GetAllEndpoints(bool includePlatformEndpoints, CancellationToken cancellationToken); + Task> GetAllEndpoints(bool includePlatformEndpoints, CancellationToken cancellationToken = default); - Task GetEndpoint(string endpointName, ThroughputSource throughputSource, CancellationToken cancellationToken) => + Task GetEndpoint(string endpointName, ThroughputSource throughputSource, CancellationToken cancellationToken = default) => GetEndpoint(new EndpointIdentifier(endpointName, throughputSource), cancellationToken); Task GetEndpoint(EndpointIdentifier id, CancellationToken cancellationToken = default); - Task> GetEndpoints(IList endpointIds, CancellationToken cancellationToken); + Task> GetEndpoints(IList endpointIds, CancellationToken cancellationToken = default); - Task SaveEndpoint(Endpoint endpoint, CancellationToken cancellationToken); + Task SaveEndpoint(Endpoint endpoint, CancellationToken cancellationToken = default); - Task>> GetEndpointThroughputByQueueName(IList queueNames, CancellationToken cancellationToken); + Task>> GetEndpointThroughputByQueueName(IList queueNames, CancellationToken cancellationToken = default); - Task RecordEndpointThroughput(string endpointName, ThroughputSource throughputSource, DateOnly date, long messageCount, CancellationToken cancellationToken) => + Task RecordEndpointThroughput(string endpointName, ThroughputSource throughputSource, DateOnly date, long messageCount, CancellationToken cancellationToken = default) => RecordEndpointThroughput(endpointName, throughputSource, [new EndpointDailyThroughput(date, messageCount)], cancellationToken); - Task RecordEndpointThroughput(string endpointName, ThroughputSource throughputSource, IList throughput, CancellationToken cancellationToken); + Task RecordEndpointThroughput(string endpointName, ThroughputSource throughputSource, IList throughput, CancellationToken cancellationToken = default); - Task UpdateUserIndicatorOnEndpoints(List userIndicatorUpdates, CancellationToken cancellationToken); + Task UpdateUserIndicatorOnEndpoints(List userIndicatorUpdates, CancellationToken cancellationToken = default); - Task IsThereThroughputForLastXDays(int days, CancellationToken cancellationToken); - Task IsThereThroughputForLastXDaysForSource(int days, ThroughputSource throughputSource, bool includeToday, CancellationToken cancellationToken); + Task IsThereThroughputForLastXDays(int days, CancellationToken cancellationToken = default); + Task IsThereThroughputForLastXDaysForSource(int days, ThroughputSource throughputSource, bool includeToday, CancellationToken cancellationToken = default); - Task GetBrokerMetadata(CancellationToken cancellationToken); + Task GetBrokerMetadata(CancellationToken cancellationToken = default); - Task SaveBrokerMetadata(BrokerMetadata brokerMetadata, CancellationToken cancellationToken); + Task SaveBrokerMetadata(BrokerMetadata brokerMetadata, CancellationToken cancellationToken = default); Task GetAuditServiceMetadata(CancellationToken cancellationToken = default); - Task SaveAuditServiceMetadata(AuditServiceMetadata auditServiceMetadata, CancellationToken cancellationToken); - Task> GetReportMasks(CancellationToken cancellationToken); - Task SaveReportMasks(List reportMasks, CancellationToken cancellationToken); + Task SaveAuditServiceMetadata(AuditServiceMetadata auditServiceMetadata, CancellationToken cancellationToken = default); + Task> GetReportMasks(CancellationToken cancellationToken = default); + Task SaveReportMasks(List reportMasks, CancellationToken cancellationToken = default); - Task GetLicensedEndpointDetails(CancellationToken cancellationToken); - Task SaveLicensedEndpointDetails(LicensedEndpointDetails result, CancellationToken cancellationToken); + Task GetLicensedEndpointDetails(CancellationToken cancellationToken = default); + Task SaveLicensedEndpointDetails(LicensedEndpointDetails result, CancellationToken cancellationToken = default); } \ No newline at end of file diff --git a/src/ServiceControl.Persistence.EFCore/.editorconfig b/src/ServiceControl.Persistence.EFCore/.editorconfig index d8f0448f31..fc68ac3228 100644 --- a/src/ServiceControl.Persistence.EFCore/.editorconfig +++ b/src/ServiceControl.Persistence.EFCore/.editorconfig @@ -3,13 +3,6 @@ # Justification: ServiceControl app has no synchronization context dotnet_diagnostic.CA2007.severity = none -# Cancellation analyzer debt, now confined to the licensing data store. ILicensingDataStore belongs to -# the licensing phase, not the persistence one, so its 15 optional-token additions land with the rest -# of Particular.LicensingComponent. Delete this block then; nothing else in the project violates a -# cancellation rule. -[Implementation/LicensingDataStore.cs] -dotnet_diagnostic.PS0003.severity = none - # Disable style rules for auto-generated EF migrations [Migrations/**.cs] dotnet_diagnostic.IDE0065.severity = none diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/LicensingDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/LicensingDataStore.cs index d5762a6d0b..36538858c3 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/LicensingDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/LicensingDataStore.cs @@ -16,7 +16,7 @@ class LicensingDataStore(IServiceScopeFactory scopeFactory, TimeProvider timePro static readonly AuditServiceMetadata DefaultAuditServiceMetadata = new([], []); static readonly BrokerMetadata DefaultBrokerMetadata = new(null, []); - public Task> GetAllEndpoints(bool includePlatformEndpoints, CancellationToken cancellationToken) => + public Task> GetAllEndpoints(bool includePlatformEndpoints, CancellationToken cancellationToken = default) => ExecuteWithDbContext(async (context, token) => { var query = context.LicensingEndpoints.AsNoTracking(); @@ -51,7 +51,7 @@ public Task> GetAllEndpoints(bool includePlatformEndpoints return row is null ? null : ToEndpoint(row.Endpoint, row.LastCollectedDate); }, cancellationToken); - public Task> GetEndpoints(IList endpointIds, CancellationToken cancellationToken) => + public Task> GetEndpoints(IList endpointIds, CancellationToken cancellationToken = default) => ExecuteWithDbContext(async (context, token) => { var normalizedNames = endpointIds.Select(id => Normalize(id.Name)).Distinct().ToList(); @@ -80,7 +80,7 @@ public Task> GetAllEndpoints(bool includePlatformEndpoints .AsEnumerable(); }, cancellationToken); - public Task SaveEndpoint(Endpoint endpoint, CancellationToken cancellationToken) => + public Task SaveEndpoint(Endpoint endpoint, CancellationToken cancellationToken = default) => ExecuteWithDbContext((context, token) => { var normalizedName = Normalize(endpoint.Id.Name); @@ -112,7 +112,7 @@ public Task SaveEndpoint(Endpoint endpoint, CancellationToken cancellationToken) token); }, cancellationToken); - public Task>> GetEndpointThroughputByQueueName(IList queueNames, CancellationToken cancellationToken) => + public Task>> GetEndpointThroughputByQueueName(IList queueNames, CancellationToken cancellationToken = default) => ExecuteWithDbContext(async (context, token) => { var results = queueNames.ToDictionary(queueName => queueName, _ => Enumerable.Empty()); @@ -161,7 +161,7 @@ public Task>> GetEndpointThrough return (IDictionary>)results; }, cancellationToken); - public async Task RecordEndpointThroughput(string endpointName, ThroughputSource throughputSource, IList throughput, CancellationToken cancellationToken) + public async Task RecordEndpointThroughput(string endpointName, ThroughputSource throughputSource, IList throughput, CancellationToken cancellationToken = default) { if (throughput.Count == 0) { @@ -246,7 +246,7 @@ static async Task TryRecordEndpointThroughput(ServiceControlDbContext cont }); } - public Task UpdateUserIndicatorOnEndpoints(List userIndicatorUpdates, CancellationToken cancellationToken) => + public Task UpdateUserIndicatorOnEndpoints(List userIndicatorUpdates, CancellationToken cancellationToken = default) => ExecuteWithDbContext(async (context, token) => { var updates = userIndicatorUpdates.ToDictionary(update => Normalize(update.Name), update => update.UserIndicator); @@ -292,10 +292,10 @@ public Task UpdateUserIndicatorOnEndpoints(List userIndicat await context.SaveChangesAsync(token); }, cancellationToken); - public Task IsThereThroughputForLastXDays(int days, CancellationToken cancellationToken) => + public Task IsThereThroughputForLastXDays(int days, CancellationToken cancellationToken = default) => IsThereThroughput(days, throughputSource: null, includeToday: false, cancellationToken); - public Task IsThereThroughputForLastXDaysForSource(int days, ThroughputSource throughputSource, bool includeToday, CancellationToken cancellationToken) => + public Task IsThereThroughputForLastXDaysForSource(int days, ThroughputSource throughputSource, bool includeToday, CancellationToken cancellationToken = default) => IsThereThroughput(days, throughputSource, includeToday, cancellationToken); Task IsThereThroughput(int days, ThroughputSource? throughputSource, bool includeToday, CancellationToken cancellationToken) => @@ -317,31 +317,31 @@ Task IsThereThroughput(int days, ThroughputSource? throughputSource, bool return query.AnyAsync(token); }, cancellationToken); - public Task GetBrokerMetadata(CancellationToken cancellationToken) => + public Task GetBrokerMetadata(CancellationToken cancellationToken = default) => ExecuteWithDbContext(async (context, token) => await context.GetSetting(SettingKeys.BrokerMetadata, token) ?? DefaultBrokerMetadata, cancellationToken); - public Task SaveBrokerMetadata(BrokerMetadata brokerMetadata, CancellationToken cancellationToken) => + public Task SaveBrokerMetadata(BrokerMetadata brokerMetadata, CancellationToken cancellationToken = default) => ExecuteWithDbContext((context, token) => context.StoreSetting(SettingKeys.BrokerMetadata, brokerMetadata, token), cancellationToken); public Task GetAuditServiceMetadata(CancellationToken cancellationToken = default) => ExecuteWithDbContext(async (context, token) => await context.GetSetting(SettingKeys.AuditServiceMetadata, token) ?? DefaultAuditServiceMetadata, cancellationToken); - public Task SaveAuditServiceMetadata(AuditServiceMetadata auditServiceMetadata, CancellationToken cancellationToken) => + public Task SaveAuditServiceMetadata(AuditServiceMetadata auditServiceMetadata, CancellationToken cancellationToken = default) => ExecuteWithDbContext((context, token) => context.StoreSetting(SettingKeys.AuditServiceMetadata, auditServiceMetadata, token), cancellationToken); - public Task> GetReportMasks(CancellationToken cancellationToken) => + public Task> GetReportMasks(CancellationToken cancellationToken = default) => ExecuteWithDbContext(async (context, token) => await context.GetSetting>(SettingKeys.ReportMasks, token) ?? [], cancellationToken); - public Task SaveReportMasks(List reportMasks, CancellationToken cancellationToken) => + public Task SaveReportMasks(List reportMasks, CancellationToken cancellationToken = default) => ExecuteWithDbContext((context, token) => context.StoreSetting(SettingKeys.ReportMasks, reportMasks, token), cancellationToken); - public Task GetLicensedEndpointDetails(CancellationToken cancellationToken) => + public Task GetLicensedEndpointDetails(CancellationToken cancellationToken = default) => ExecuteWithDbContext((context, token) => context.GetSetting(SettingKeys.LicensedEndpointDetails, token), cancellationToken); - public Task SaveLicensedEndpointDetails(LicensedEndpointDetails result, CancellationToken cancellationToken) => + public Task SaveLicensedEndpointDetails(LicensedEndpointDetails result, CancellationToken cancellationToken = default) => ExecuteWithDbContext((context, token) => context.StoreSetting(SettingKeys.LicensedEndpointDetails, result, token), cancellationToken); static Endpoint ToEndpoint(LicensingEndpointEntity row) => ToEndpoint(row, lastCollectedDate: null); diff --git a/src/ServiceControl.Persistence.RavenDB/.editorconfig b/src/ServiceControl.Persistence.RavenDB/.editorconfig index d84391c0d5..ff993b49bb 100644 --- a/src/ServiceControl.Persistence.RavenDB/.editorconfig +++ b/src/ServiceControl.Persistence.RavenDB/.editorconfig @@ -2,10 +2,3 @@ # Justification: ServiceControl app has no synchronization context dotnet_diagnostic.CA2007.severity = none - -# Cancellation analyzer debt, now confined to the licensing data store. ILicensingDataStore belongs to -# the licensing phase, not the persistence one, so its 17 optional-token additions land with the rest -# of Particular.LicensingComponent. Delete this block then; nothing else in the project violates a -# cancellation rule. -[Throughput/LicensingDataStore.cs] -dotnet_diagnostic.PS0003.severity = none diff --git a/src/ServiceControl.Persistence.RavenDB/Throughput/LicensingDataStore.cs b/src/ServiceControl.Persistence.RavenDB/Throughput/LicensingDataStore.cs index aecffa14d7..4c76b83e64 100644 --- a/src/ServiceControl.Persistence.RavenDB/Throughput/LicensingDataStore.cs +++ b/src/ServiceControl.Persistence.RavenDB/Throughput/LicensingDataStore.cs @@ -29,7 +29,7 @@ class LicensingDataStore( static readonly BrokerMetadata DefaultBrokerMetadata = new(null, []); static readonly ReportConfigurationDocument DefaultReportConfiguration = new(); - public async Task> GetAllEndpoints(bool includePlatformEndpoints, CancellationToken cancellationToken) + public async Task> GetAllEndpoints(bool includePlatformEndpoints, CancellationToken cancellationToken = default) { var store = await storeProvider.GetDocumentStore(cancellationToken); using IAsyncDocumentSession session = store.OpenAsyncSession(databaseConfiguration.Name); @@ -45,7 +45,7 @@ public async Task> GetAllEndpoints(bool includePlatformEnd return documents.Select(document => document.ToEndpoint()); } - public async Task GetEndpoint(EndpointIdentifier id, CancellationToken cancellationToken) + public async Task GetEndpoint(EndpointIdentifier id, CancellationToken cancellationToken = default) { var store = await storeProvider.GetDocumentStore(cancellationToken); using IAsyncDocumentSession session = store.OpenAsyncSession(databaseConfiguration.Name); @@ -70,7 +70,7 @@ public async Task> GetAllEndpoints(bool includePlatformEnd return endpoint; } - public async Task> GetEndpoints(IList endpointIds, CancellationToken cancellationToken) + public async Task> GetEndpoints(IList endpointIds, CancellationToken cancellationToken = default) { var documentIds = endpointIds.Select(id => id.GenerateDocumentId()); @@ -110,7 +110,7 @@ public async Task> GetAllEndpoints(bool includePlatformEnd }); } - public async Task SaveEndpoint(Endpoint endpoint, CancellationToken cancellationToken) + public async Task SaveEndpoint(Endpoint endpoint, CancellationToken cancellationToken = default) { var document = endpoint.ToEndpointDocument(); @@ -121,7 +121,7 @@ public async Task SaveEndpoint(Endpoint endpoint, CancellationToken cancellation await session.SaveChangesAsync(cancellationToken); } - public async Task>> GetEndpointThroughputByQueueName(IList queueNames, CancellationToken cancellationToken) + public async Task>> GetEndpointThroughputByQueueName(IList queueNames, CancellationToken cancellationToken = default) { var results = queueNames.ToDictionary(queueName => queueName, _ => new List() as IEnumerable); @@ -156,7 +156,7 @@ public async Task>> GetEndpointT return results; } - public async Task RecordEndpointThroughput(string endpointName, ThroughputSource throughputSource, IList throughput, CancellationToken cancellationToken) + public async Task RecordEndpointThroughput(string endpointName, ThroughputSource throughputSource, IList throughput, CancellationToken cancellationToken = default) { if (!throughput.Any()) { @@ -181,7 +181,7 @@ public async Task RecordEndpointThroughput(string endpointName, ThroughputSource await session.SaveChangesAsync(cancellationToken); } - public async Task UpdateUserIndicatorOnEndpoints(List userIndicatorUpdates, CancellationToken cancellationToken) + public async Task UpdateUserIndicatorOnEndpoints(List userIndicatorUpdates, CancellationToken cancellationToken = default) { var updates = userIndicatorUpdates.ToDictionary(u => u.Name, u => u.UserIndicator); var store = await storeProvider.GetDocumentStore(cancellationToken); @@ -230,7 +230,7 @@ public async Task UpdateUserIndicatorOnEndpoints(List userI await session.SaveChangesAsync(cancellationToken); } - public async Task IsThereThroughputForLastXDays(int days, CancellationToken cancellationToken) + public async Task IsThereThroughputForLastXDays(int days, CancellationToken cancellationToken = default) { var store = await storeProvider.GetDocumentStore(cancellationToken); using IAsyncDocumentSession session = store.OpenAsyncSession(databaseConfiguration.Name); @@ -240,7 +240,7 @@ public async Task IsThereThroughputForLastXDays(int days, CancellationToke return result; } - public async Task IsThereThroughputForLastXDaysForSource(int days, ThroughputSource throughputSource, bool includeToday, CancellationToken cancellationToken) + public async Task IsThereThroughputForLastXDaysForSource(int days, ThroughputSource throughputSource, bool includeToday, CancellationToken cancellationToken = default) { var store = await storeProvider.GetDocumentStore(cancellationToken); using IAsyncDocumentSession session = store.OpenAsyncSession(databaseConfiguration.Name); @@ -265,7 +265,7 @@ static async Task IsThereThroughputForLastXDaysInternal(IRavenQueryable timeSeries.Results).Any(); } - public async Task GetBrokerMetadata(CancellationToken cancellationToken) + public async Task GetBrokerMetadata(CancellationToken cancellationToken = default) { var store = await storeProvider.GetDocumentStore(cancellationToken); using IAsyncDocumentSession session = store.OpenAsyncSession(databaseConfiguration.Name); @@ -273,7 +273,7 @@ public async Task GetBrokerMetadata(CancellationToken cancellati return await session.LoadAsync(BrokerMetadataDocumentId, cancellationToken) ?? DefaultBrokerMetadata; } - public async Task SaveBrokerMetadata(BrokerMetadata brokerMetadata, CancellationToken cancellationToken) + public async Task SaveBrokerMetadata(BrokerMetadata brokerMetadata, CancellationToken cancellationToken = default) { var store = await storeProvider.GetDocumentStore(cancellationToken); using IAsyncDocumentSession session = store.OpenAsyncSession(databaseConfiguration.Name); @@ -282,7 +282,7 @@ public async Task SaveBrokerMetadata(BrokerMetadata brokerMetadata, Cancellation await session.SaveChangesAsync(cancellationToken); } - public async Task GetAuditServiceMetadata(CancellationToken cancellationToken) + public async Task GetAuditServiceMetadata(CancellationToken cancellationToken = default) { var store = await storeProvider.GetDocumentStore(cancellationToken); using IAsyncDocumentSession session = store.OpenAsyncSession(databaseConfiguration.Name); @@ -290,7 +290,7 @@ public async Task GetAuditServiceMetadata(CancellationToke return await session.LoadAsync(AuditServiceMetadataDocumentId, cancellationToken) ?? DefaultAuditServiceMetadata; } - public async Task SaveAuditServiceMetadata(AuditServiceMetadata auditServiceMetadata, CancellationToken cancellationToken) + public async Task SaveAuditServiceMetadata(AuditServiceMetadata auditServiceMetadata, CancellationToken cancellationToken = default) { var store = await storeProvider.GetDocumentStore(cancellationToken); using IAsyncDocumentSession session = store.OpenAsyncSession(databaseConfiguration.Name); @@ -299,7 +299,7 @@ public async Task SaveAuditServiceMetadata(AuditServiceMetadata auditServiceMeta await session.SaveChangesAsync(cancellationToken); } - public async Task> GetReportMasks(CancellationToken cancellationToken) + public async Task> GetReportMasks(CancellationToken cancellationToken = default) { var store = await storeProvider.GetDocumentStore(cancellationToken); using IAsyncDocumentSession session = store.OpenAsyncSession(databaseConfiguration.Name); @@ -309,7 +309,7 @@ public async Task> GetReportMasks(CancellationToken cancellationTok return config.MaskedStrings; } - public async Task SaveReportMasks(List reportMasks, CancellationToken cancellationToken) + public async Task SaveReportMasks(List reportMasks, CancellationToken cancellationToken = default) { var store = await storeProvider.GetDocumentStore(cancellationToken); using IAsyncDocumentSession session = store.OpenAsyncSession(databaseConfiguration.Name); @@ -318,7 +318,7 @@ public async Task SaveReportMasks(List reportMasks, CancellationToken ca await session.SaveChangesAsync(cancellationToken); } - public async Task GetLicensedEndpointDetails(CancellationToken cancellationToken) + public async Task GetLicensedEndpointDetails(CancellationToken cancellationToken = default) { var store = await storeProvider.GetDocumentStore(cancellationToken); using IAsyncDocumentSession session = store.OpenAsyncSession(databaseConfiguration.Name); @@ -326,7 +326,7 @@ public async Task SaveReportMasks(List reportMasks, CancellationToken ca return await session.LoadAsync(LicencedEndpointDetailsDocumentId, cancellationToken); } - public async Task SaveLicensedEndpointDetails(LicensedEndpointDetails result, CancellationToken cancellationToken) + public async Task SaveLicensedEndpointDetails(LicensedEndpointDetails result, CancellationToken cancellationToken = default) { var store = await storeProvider.GetDocumentStore(cancellationToken); using IAsyncDocumentSession session = store.OpenAsyncSession(databaseConfiguration.Name); From 399ff72eb2c4a501f8b629dee3515cc491431ed4 Mon Sep 17 00:00:00 2001 From: John Simons Date: Wed, 12 Aug 2026 09:29:20 +1000 Subject: [PATCH 5/6] Clean up cancellation analyzer suppression comment in licensing status check This change removes a redundant comment explaining the PS0020 analyzer suppression within the RavenDB licensing status check, continuing the effort to clean up cancellation-related debt. --- src/ServiceControl.Persistence.RavenDB/LicenseStatusCheck.cs | 3 --- 1 file changed, 3 deletions(-) diff --git a/src/ServiceControl.Persistence.RavenDB/LicenseStatusCheck.cs b/src/ServiceControl.Persistence.RavenDB/LicenseStatusCheck.cs index ea0a3da4de..58a65e87dc 100644 --- a/src/ServiceControl.Persistence.RavenDB/LicenseStatusCheck.cs +++ b/src/ServiceControl.Persistence.RavenDB/LicenseStatusCheck.cs @@ -37,9 +37,6 @@ public static async Task WaitForLicenseOrThrow(IDocumentStore documentStore, Can await Task.Delay(200, cts.Token); } } - // Every call inside the try uses the linked cts.Token, so PS0020 cannot see that this filter - // tests the right thing: cancellation that is not the caller's is the 30s license-check - // timeout, and that is what the friendly message is for. #pragma warning disable PS0020 catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested) { From 2940c40dd3e8fad2531381f2b1fd272840746efb Mon Sep 17 00:00:00 2001 From: John Simons Date: Wed, 12 Aug 2026 13:47:22 +1000 Subject: [PATCH 6/6] Propagate cancellation tokens through failed message retry data store persistence This change continues the effort to consistently apply optional `CancellationToken` parameters to asynchronous methods within the `IFailedMessageRetryDataStore` EFCore implementation. Cancellation tokens are now propagated to underlying asynchronous database calls, ensuring proper responsiveness to cancellation and further reducing cancellation analyzer debt. --- .../FailedMessageRetryDataStore.cs | 28 +++++++++---------- 1 file changed, 14 insertions(+), 14 deletions(-) diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageRetryDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageRetryDataStore.cs index f50e2b8be6..a2850f9ddd 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageRetryDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageRetryDataStore.cs @@ -10,19 +10,19 @@ namespace ServiceControl.Persistence.EFCore.Implementation; public class FailedMessageRetryDataStore(IServiceScopeFactory scopeFactory, IBodyStorage bodyStorage) : DataStoreBase(scopeFactory), IFailedMessageRetryDataStore { - public Task RemoveFailedMessageRetry(string uniqueMessageId, CancellationToken cancellationToken) => - ExecuteWithDbContext(async dbContext => + public Task RemoveFailedMessageRetry(string uniqueMessageId, CancellationToken cancellationToken = default) => + ExecuteWithDbContext(async (dbContext, token) => { if (Guid.TryParse(uniqueMessageId, out var id)) { await dbContext.FailedMessageRetries .Where(r => r.UniqueMessageId == id) - .ExecuteDeleteAsync(cancellationToken: cancellationToken); + .ExecuteDeleteAsync(token); } - }); + }, cancellationToken); - public Task GetRetryPendingMessages(DateTime from, DateTime to, string queueAddress, CancellationToken cancellationToken) => - ExecuteWithDbContext(async dbContext => + public Task GetRetryPendingMessages(DateTime from, DateTime to, string queueAddress, CancellationToken cancellationToken = default) => + ExecuteWithDbContext(async (dbContext, token) => { var normalizedQueueAddress = queueAddress?.ToLowerInvariant(); var ids = await dbContext.FailedMessages @@ -32,13 +32,13 @@ public Task GetRetryPendingMessages(DateTime from, DateTime to, string && ((m.FailingEndpointAddress == null && normalizedQueueAddress == null) || (m.FailingEndpointAddress != null && m.FailingEndpointAddress.ToLower() == normalizedQueueAddress))) .Select(m => m.UniqueMessageId.ToString()) - .ToListAsync(cancellationToken: cancellationToken); + .ToListAsync(token); return ids.ToArray(); - }); + }, cancellationToken); - public Task ProcessPendingRetries(DateTime periodFrom, DateTime periodTo, string queueAddress, Func processCallback, CancellationToken cancellationToken) => - ExecuteWithDbContext(async dbContext => + public Task ProcessPendingRetries(DateTime periodFrom, DateTime periodTo, string queueAddress, Func processCallback, CancellationToken cancellationToken = default) => + ExecuteWithDbContext(async (dbContext, token) => { var query = dbContext.FailedMessages .AsNoTracking() @@ -47,13 +47,13 @@ public Task ProcessPendingRetries(DateTime periodFrom, DateTime periodTo, string .FilterByQueueAddress(queueAddress) .Select(m => m.UniqueMessageId.ToString()); - await foreach (var uniqueMessageId in query.AsAsyncEnumerable().WithCancellation(cancellationToken)) + await foreach (var uniqueMessageId in query.AsAsyncEnumerable().WithCancellation(token)) { - await processCallback(uniqueMessageId, cancellationToken); + await processCallback(uniqueMessageId, token); } - }); + }, cancellationToken); - public async Task GetFailedMessageBody(string uniqueMessageId, CancellationToken cancellationToken) + public async Task GetFailedMessageBody(string uniqueMessageId, CancellationToken cancellationToken = default) { var result = await bodyStorage.TryFetch(uniqueMessageId, cancellationToken) ?? throw new InvalidOperationException("IBodyStorage.TryFetch result cannot be null");