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 7651f47081..fc68ac3228 100644 --- a/src/ServiceControl.Persistence.EFCore/.editorconfig +++ b/src/ServiceControl.Persistence.EFCore/.editorconfig @@ -3,16 +3,6 @@ # 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.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] dotnet_diagnostic.IDE0065.severity = none 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 93084037fe..7062260163 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) => - ExecuteWithDbContext(async context => + public Task UpdateEndpointSettings(EndpointSettings settings, CancellationToken cancellationToken = default) => + ExecuteWithDbContext(async (context, token) => 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) => - ExecuteWithDbContext(context => context.EndpointSettings.Where(x => x.Name == name) - .ExecuteDeleteAsync(cancellationToken)); + public Task Delete(string name, CancellationToken cancellationToken = default) => + 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/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"); 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..36538858c3 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/LicensingDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/LicensingDataStore.cs @@ -16,8 +16,8 @@ class LicensingDataStore(IServiceScopeFactory scopeFactory, TimeProvider timePro static readonly AuditServiceMetadata DefaultAuditServiceMetadata = new([], []); static readonly BrokerMetadata DefaultBrokerMetadata = new(null, []); - public Task> GetAllEndpoints(bool includePlatformEndpoints, CancellationToken cancellationToken) => - ExecuteWithDbContext(async context => + public Task> GetAllEndpoints(bool includePlatformEndpoints, CancellationToken cancellationToken = default) => + 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 => + public Task> GetEndpoints(IList endpointIds, CancellationToken cancellationToken = default) => + 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 => + public Task SaveEndpoint(Endpoint endpoint, CancellationToken cancellationToken = default) => + 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 => + public Task>> GetEndpointThroughputByQueueName(IList queueNames, CancellationToken cancellationToken = default) => + 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,9 +159,9 @@ 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) { @@ -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) { @@ -246,15 +246,15 @@ static async Task TryRecordEndpointThroughput(ServiceControlDbContext cont }); } - public Task UpdateUserIndicatorOnEndpoints(List userIndicatorUpdates, CancellationToken cancellationToken) => - ExecuteWithDbContext(async context => + public Task UpdateUserIndicatorOnEndpoints(List userIndicatorUpdates, CancellationToken cancellationToken = default) => + 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,17 +289,17 @@ public Task UpdateUserIndicatorOnEndpoints(List userIndicat } } - await context.SaveChangesAsync(cancellationToken); - }); + 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) => - 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); + 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) => - ExecuteWithDbContext(context => context.StoreSetting(SettingKeys.BrokerMetadata, brokerMetadata, 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 => - 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)); + public Task SaveAuditServiceMetadata(AuditServiceMetadata auditServiceMetadata, CancellationToken cancellationToken = default) => + ExecuteWithDbContext((context, token) => context.StoreSetting(SettingKeys.AuditServiceMetadata, auditServiceMetadata, token), cancellationToken); - public Task> GetReportMasks(CancellationToken cancellationToken) => - ExecuteWithDbContext(async context => - await context.GetSetting>(SettingKeys.ReportMasks, 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) => - ExecuteWithDbContext(context => context.StoreSetting(SettingKeys.ReportMasks, reportMasks, cancellationToken)); + public Task SaveReportMasks(List reportMasks, CancellationToken cancellationToken = default) => + ExecuteWithDbContext((context, token) => context.StoreSetting(SettingKeys.ReportMasks, reportMasks, token), cancellationToken); - public Task GetLicensedEndpointDetails(CancellationToken cancellationToken) => - ExecuteWithDbContext(context => context.GetSetting(SettingKeys.LicensedEndpointDetails, cancellationToken)); + public Task GetLicensedEndpointDetails(CancellationToken cancellationToken = default) => + ExecuteWithDbContext((context, token) => context.GetSetting(SettingKeys.LicensedEndpointDetails, token), cancellationToken); - public Task SaveLicensedEndpointDetails(LicensedEndpointDetails result, CancellationToken cancellationToken) => - ExecuteWithDbContext(context => context.StoreSetting(SettingKeys.LicensedEndpointDetails, result, 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.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 38f458422d..101131ffc0 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) => - ExecuteWithDbContext(async dbContext => + public Task CreateIfNotExists(EndpointDetails endpoint, CancellationToken cancellationToken = default) => + ExecuteWithDbContext(async (dbContext, token) => { var id = endpoint.GetDeterministicId(); - var exists = await dbContext.KnownEndpoints.AnyAsync(e => e.Id == id); + var exists = await dbContext.KnownEndpoints.AnyAsync(e => e.Id == id, token); if (exists) { return; @@ -31,22 +31,22 @@ public Task CreateIfNotExists(EndpointDetails endpoint) => try { - await dbContext.SaveChangesAsync(); + 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)) + if (!await dbContext.KnownEndpoints.AnyAsync(e => e.Id == id, token)) { throw; } } - }); + }, cancellationToken); - public Task CreateOrUpdate(EndpointDetails endpoint, IEndpointInstanceMonitoring endpointInstanceMonitoring) => - ExecuteWithDbContext(async dbContext => + public Task CreateOrUpdate(EndpointDetails endpoint, IEndpointInstanceMonitoring endpointInstanceMonitoring, CancellationToken cancellationToken = default) => + ExecuteWithDbContext(async (dbContext, token) => { var id = endpoint.GetDeterministicId(); await dbContext.UpsertAsync([id], @@ -61,24 +61,25 @@ await dbContext.UpsertAsync([id], knownEndpoint => { knownEndpoint.Monitored = endpointInstanceMonitoring.IsMonitored(id); - } + }, + token ); - }); + }, cancellationToken); - public Task UpdateEndpointMonitoring(EndpointDetails endpoint, bool isMonitored) => - ExecuteWithDbContext(async dbContext => + public Task UpdateEndpointMonitoring(EndpointDetails endpoint, bool isMonitored, CancellationToken cancellationToken = default) => + ExecuteWithDbContext(async (dbContext, token) => { 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), token); + }, cancellationToken); - public Task WarmupMonitoringFromPersistence(IEndpointInstanceMonitoring endpointInstanceMonitoring) => - ExecuteWithDbContext(async dbContext => + public Task WarmupMonitoringFromPersistence(IEndpointInstanceMonitoring endpointInstanceMonitoring, CancellationToken cancellationToken = default) => + ExecuteWithDbContext(async (dbContext, token) => { - await foreach (var endpoint in dbContext.KnownEndpoints.AsNoTracking().AsAsyncEnumerable()) + await foreach (var endpoint in dbContext.KnownEndpoints.AsNoTracking().AsAsyncEnumerable().WithCancellation(token)) { var endpointDetails = new EndpointDetails { @@ -89,21 +90,21 @@ public Task WarmupMonitoringFromPersistence(IEndpointInstanceMonitoring endpoint endpointInstanceMonitoring.DetectEndpointFromPersistentStore(endpointDetails, endpoint.Monitored); } - }); + }, cancellationToken); - public Task Delete(Guid endpointId) + 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(); - }); + .ExecuteDeleteAsync(token); + }, cancellationToken); } - public Task> GetAllKnownEndpoints() + public Task> GetAllKnownEndpoints(CancellationToken cancellationToken = default) { - return ExecuteWithDbContext>(async dbContext => + return ExecuteWithDbContext>(async (dbContext, token) => await dbContext.KnownEndpoints .AsNoTracking() .Select(e => new KnownEndpoint @@ -117,6 +118,6 @@ await dbContext.KnownEndpoints HostDisplayName = e.Host, Monitored = e.Monitored }) - .ToListAsync()); + .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 e910c4d8c0..128c1543bc 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) { @@ -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 46aa4f4ee6..222f1c87ef 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) - => ExecuteWithDbContext(context => - context.GetSetting(SettingKeys.TrialEndDate, cancellationToken)); + public Task GetTrialEndDate(CancellationToken cancellationToken = default) + => ExecuteWithDbContext((context, token) => + context.GetSetting(SettingKeys.TrialEndDate, token), cancellationToken); - public Task StoreTrialEndDate(DateOnly trialEndDate, CancellationToken cancellationToken) - => ExecuteWithDbContext(context => - context.StoreSetting(SettingKeys.TrialEndDate, trialEndDate, cancellationToken)); + public Task StoreTrialEndDate(DateOnly trialEndDate, CancellationToken cancellationToken = default) + => 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..ff993b49bb 100644 --- a/src/ServiceControl.Persistence.RavenDB/.editorconfig +++ b/src/ServiceControl.Persistence.RavenDB/.editorconfig @@ -2,12 +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.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/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/LicenseStatusCheck.cs b/src/ServiceControl.Persistence.RavenDB/LicenseStatusCheck.cs index 6b7ff4736e..58a65e87dc 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,11 @@ public static async Task WaitForLicenseOrThrow(IDocumentStore documentStore, Can await Task.Delay(200, cts.Token); } } +#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/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.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.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); 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.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 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/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.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.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/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; } 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();