From 9312e41d4d455885dec7ca226c850c610004753d Mon Sep 17 00:00:00 2001 From: John Simons Date: Tue, 11 Aug 2026 14:22:17 +1000 Subject: [PATCH] Propagate cancellation tokens through persistence interfaces and implementations This change continues the effort to consistently apply optional `CancellationToken` parameters to asynchronous methods within the persistence data store interfaces (`IRetryBatchStore`, `IRetryHistoryDataStore`, `IRetryStagingStore`) and their EFCore and RavenDB implementations. The cancellation tokens are now propagated to underlying asynchronous database calls and to consuming services. This ensures proper cancellation of long-running database operations and resolves related static analysis warnings, further reducing cancellation analyzer debt. --- .../Implementation/RetryBatchStore.cs | 61 ++++++++------- .../Implementation/RetryHistoryDataStore.cs | 35 +++++---- .../Implementation/RetryStagingStore.cs | 78 +++++++++---------- .../Recoverability/RetryHistoryDataStore.cs | 28 +++---- .../RetryDocumentDataStore.cs | 76 +++++++++--------- .../RetryStagingStore.cs | 67 ++++++++-------- .../EFCore/RetryBatchStoreTests.cs | 5 +- .../IRetryBatchStore.cs | 22 +++--- .../IRetryHistoryDataStore.cs | 8 +- .../IRetryStagingStore.cs | 20 ++--- src/ServiceControl/.editorconfig | 9 --- .../API/FailureGroupsController.cs | 2 +- .../Recoverability/API/GroupFetcher.cs | 4 +- .../API/UnacknowledgedGroupsController.cs | 2 +- .../Retrying/History/StoreHistoryHandler.cs | 3 +- .../Recoverability/Retrying/RetriesGateway.cs | 36 ++++----- .../Retrying/RetryDocumentManager.cs | 6 +- .../Recoverability/Retrying/RetryProcessor.cs | 18 ++--- 18 files changed, 242 insertions(+), 238 deletions(-) diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/RetryBatchStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/RetryBatchStore.cs index 11fa9b8681..a6d94e6520 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/RetryBatchStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/RetryBatchStore.cs @@ -13,7 +13,8 @@ public class RetryBatchStore(IServiceScopeFactory scopeFactory, IRetryBatchSqlDi public Task CreateBatch(string retrySessionId, string requestId, RetryType retryType, string[] failedMessageRetryIds, string originator, DateTime startTime, DateTime? last = null, string? batchName = null, string? classifier = null, - string? initiatedById = null, string? initiatedByName = null, string? operationId = null) => + string? initiatedById = null, string? initiatedByName = null, string? operationId = null, + CancellationToken cancellationToken = default) => ExecuteWithDbContext(async dbContext => { var batch = new RetryBatchEntity @@ -36,7 +37,7 @@ public Task CreateBatch(string retrySessionId, string requestId, RetryTy dbContext.RetryBatches.Add(batch); - await dbContext.SaveChangesAsync(); + await dbContext.SaveChangesAsync(cancellationToken); return batch.Id.ToString(); }); @@ -44,7 +45,7 @@ public Task CreateBatch(string retrySessionId, string requestId, RetryTy /// /// 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) => + public Task AssignMessagesToBatch(string batchId, string[] messageIds, CancellationToken cancellationToken = default) => ExecuteWithDbContext(async dbContext => { var batch = ParseBatchId(batchId); @@ -75,21 +76,21 @@ public Task AssignMessagesToBatch(string batchId, string[] messageIds) => await strategy.ExecuteAsync(async () => { - await using var transaction = await dbContext.Database.BeginTransactionAsync(); + await using var transaction = await dbContext.Database.BeginTransactionAsync(cancellationToken); await dialect.InsertMissingRetryClaims(dbContext, claims, CancellationToken.None); - await transaction.CommitAsync(); + await transaction.CommitAsync(cancellationToken); }); }); - public Task MoveBatchToStaging(string batchId) + public Task MoveBatchToStaging(string batchId, CancellationToken cancellationToken = default) { var batch = ParseBatchId(batchId); return ExecuteWithDbContext(dbContext => dbContext.RetryBatches .Where(row => row.Id == batch) - .ExecuteUpdateAsync(setters => setters.SetProperty(row => row.Status, RetryBatchStatus.Staging))); + .ExecuteUpdateAsync(setters => setters.SetProperty(row => row.Status, RetryBatchStatus.Staging), cancellationToken)); } // Batch ids only ever come from CreateBatch, so anything else is a programming error. @@ -98,22 +99,22 @@ static Guid ParseBatchId(string batchId) => ? parsed : throw new ArgumentException($"'{batchId}' is not a retry batch id issued by this store.", nameof(batchId)); - public Task>> GetOrphanedBatches(string retrySessionId) => + public Task>> GetOrphanedBatches(string retrySessionId, CancellationToken cancellationToken = default) => ExecuteWithDbContext(async dbContext => { var orphaned = await dbContext.RetryBatches .AsNoTracking() .Where(batch => batch.Status == RetryBatchStatus.MarkingDocuments && batch.RetrySessionId != retrySessionId) - .ToListAsync(); + .ToListAsync(cancellationToken); - var messageCounts = await CountMessages(dbContext, [.. orphaned.Select(batch => batch.Id)]); + var messageCounts = await CountMessages(dbContext, [.. orphaned.Select(batch => batch.Id)], cancellationToken); IList batches = [.. orphaned.Select(batch => batch.ToRetryBatch(messageCounts.GetValueOrDefault(batch.Id)))]; return new QueryResult>(batches, new QueryStatsInfo(string.Empty, batches.Count, false)); }); - public Task> GetAvailableBatchGroups() => + public Task> GetAvailableBatchGroups(CancellationToken cancellationToken = default) => ExecuteWithDbContext>(async dbContext => { var groups = await dbContext.RetryBatches @@ -132,7 +133,7 @@ public Task> GetAvailableBatchGroups() => Originator = group.Max(batch => batch.Originator), Classifier = group.Max(batch => batch.Classifier) }) - .ToListAsync(); + .ToListAsync(cancellationToken); return [.. groups.Select(group => new RetryBatchGroup { @@ -148,12 +149,12 @@ public Task> GetAvailableBatchGroups() => })]; }); - public Task GetCurrentForwardingBatch() => + public Task GetCurrentForwardingBatch(CancellationToken cancellationToken = default) => ExecuteWithDbContext(async dbContext => { var nowForwarding = await dbContext.RetryBatchNowForwarding .AsNoTracking() - .SingleOrDefaultAsync(); + .SingleOrDefaultAsync(cancellationToken); if (nowForwarding == null) { @@ -164,45 +165,45 @@ public Task> GetAvailableBatchGroups() => .AsNoTracking() .Where(batch => batch.Id == nowForwarding.RetryBatchId) .Select(batch => new ForwardingRetryBatch(batch.RequestId, batch.RetryType, batch.Originator!, batch.Classifier!)) - .SingleOrDefaultAsync(); + .SingleOrDefaultAsync(cancellationToken); }); - public Task ForEachUnresolvedMessage(Func callback) => - ForEach(Unresolved, callback); + public Task ForEachUnresolvedMessage(Func callback, CancellationToken cancellationToken = default) => + ForEach(Unresolved, callback, cancellationToken); - public Task ForEachUnresolvedMessageForEndpoint(string endpoint, Func callback) => + public Task ForEachUnresolvedMessageForEndpoint(string endpoint, Func callback, CancellationToken cancellationToken = default) => ForEach(dbContext => Unresolved(dbContext) - .Where(message => message.ReceivingEndpointName == endpoint), callback); + .Where(message => message.ReceivingEndpointName == endpoint), callback, cancellationToken); - public Task ForEachMessageForQueueAddress(string failedQueueAddress, FailedMessageStatus status, Func callback) => + public Task ForEachMessageForQueueAddress(string failedQueueAddress, FailedMessageStatus status, Func callback, CancellationToken cancellationToken = default) => ForEach(dbContext => Unresolved(dbContext) - .Where(message => message.FailingEndpointAddress == failedQueueAddress && message.Status == status), callback); + .Where(message => message.FailingEndpointAddress == failedQueueAddress && message.Status == status), callback, cancellationToken); - public Task ForEachUnresolvedMessageInGroup(string groupId, Func callback) => + public Task ForEachUnresolvedMessageInGroup(string groupId, Func callback, CancellationToken cancellationToken = default) => ForEach(dbContext => Unresolved(dbContext) - .Where(message => dbContext.FailedMessageGroups.Any(group => group.GroupId == groupId && group.FailedMessageUniqueId == message.UniqueMessageId)), callback); + .Where(message => dbContext.FailedMessageGroups.Any(group => group.GroupId == groupId && group.FailedMessageUniqueId == message.UniqueMessageId)), callback, cancellationToken); - Task ForEach(Func> query, Func callback) => - ExecuteWithDbContext(dbContext => Stream(query(dbContext), callback)); + Task ForEach(Func> query, Func callback, CancellationToken cancellationToken) => + ExecuteWithDbContext(dbContext => Stream(query(dbContext), callback, cancellationToken)); static IQueryable Unresolved(ServiceControlDbContext dbContext) => dbContext.FailedMessages .AsNoTracking() .Where(message => message.Status == FailedMessageStatus.Unresolved); - static async Task Stream(IQueryable messages, Func callback) + static async Task Stream(IQueryable messages, Func callback, CancellationToken cancellationToken) { var rows = messages .Select(message => new { message.UniqueMessageId, message.LastTimeOfFailure }) .AsAsyncEnumerable(); - await foreach (var row in rows) + await foreach (var row in rows.WithCancellation(cancellationToken)) { - await callback(row.UniqueMessageId.ToString(), row.LastTimeOfFailure); + await callback(row.UniqueMessageId.ToString(), row.LastTimeOfFailure, cancellationToken); } } - static async Task> CountMessages(ServiceControlDbContext dbContext, Guid[] batchIds) + static async Task> CountMessages(ServiceControlDbContext dbContext, Guid[] batchIds, CancellationToken cancellationToken) { if (batchIds.Length == 0) { @@ -214,6 +215,6 @@ static async Task> CountMessages(ServiceControlDbContext d .Where(retry => batchIds.Contains(retry.RetryBatchId)) .GroupBy(retry => retry.RetryBatchId) .Select(group => new { RetryBatchId = group.Key, MessageCount = group.Count() }) - .ToDictionaryAsync(row => row.RetryBatchId, row => row.MessageCount); + .ToDictionaryAsync(row => row.RetryBatchId, row => row.MessageCount, cancellationToken); } } diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/RetryHistoryDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/RetryHistoryDataStore.cs index 4170230ade..a60ed7b963 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/RetryHistoryDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/RetryHistoryDataStore.cs @@ -8,7 +8,7 @@ namespace ServiceControl.Persistence.EFCore.Implementation; public class RetryHistoryDataStore(IServiceScopeFactory scopeFactory) : DataStoreBase(scopeFactory), IRetryHistoryDataStore { - public Task GetRetryHistory() => + public Task GetRetryHistory(CancellationToken cancellationToken = default) => ExecuteWithDbContext(async dbContext => { var historicOperations = await dbContext.HistoricRetryOperations @@ -25,7 +25,7 @@ public Task GetRetryHistory() => Failed = operation.Failed, NumberOfMessagesProcessed = operation.NumberOfMessagesProcessed }) - .ToListAsync(); + .ToListAsync(cancellationToken); var unacknowledgedOperations = await dbContext.UnacknowledgedRetryOperations .AsNoTracking() @@ -41,7 +41,7 @@ public Task GetRetryHistory() => Failed = operation.Failed, NumberOfMessagesProcessed = operation.NumberOfMessagesProcessed }) - .ToListAsync(); + .ToListAsync(cancellationToken); return new RetryHistory { @@ -51,14 +51,15 @@ public Task GetRetryHistory() => }); public Task RecordRetryOperationCompleted(string requestId, RetryType retryType, DateTime startTime, DateTime completionTime, - string originator, string classifier, bool messageFailed, int numberOfMessagesProcessed, DateTime lastProcessed, int retryHistoryDepth) => + string originator, string classifier, bool messageFailed, int numberOfMessagesProcessed, DateTime lastProcessed, int retryHistoryDepth, + CancellationToken cancellationToken = default) => ExecuteWithDbContext(async dbContext => { var strategy = dbContext.Database.CreateExecutionStrategy(); await strategy.ExecuteAsync(async () => { - await using var transaction = await dbContext.Database.BeginTransactionAsync(); + await using var transaction = await dbContext.Database.BeginTransactionAsync(cancellationToken); dbContext.HistoricRetryOperations.Add(new HistoricRetryOperationEntity { @@ -74,24 +75,24 @@ await strategy.ExecuteAsync(async () => if (NeedsAcknowledgement(retryType)) { await RecordUnacknowledged(dbContext, requestId, retryType, startTime, completionTime, - originator, classifier, messageFailed, numberOfMessagesProcessed, lastProcessed); + originator, classifier, messageFailed, numberOfMessagesProcessed, lastProcessed, cancellationToken); } - await dbContext.SaveChangesAsync(); + await dbContext.SaveChangesAsync(cancellationToken); // After the insert, so the operation just recorded competes for a place in the history. - await TrimHistory(dbContext, retryHistoryDepth); + await TrimHistory(dbContext, retryHistoryDepth, cancellationToken); - await transaction.CommitAsync(); + await transaction.CommitAsync(cancellationToken); }); }); - public Task AcknowledgeRetryGroup(string groupId) => + public Task AcknowledgeRetryGroup(string groupId, CancellationToken cancellationToken = default) => ExecuteWithDbContext(async dbContext => { var acknowledged = await dbContext.UnacknowledgedRetryOperations .Where(operation => operation.RequestId == groupId && operation.RetryType == RetryType.FailureGroup) - .ExecuteDeleteAsync(); + .ExecuteDeleteAsync(cancellationToken); return acknowledged > 0; }); @@ -101,10 +102,10 @@ static bool NeedsAcknowledgement(RetryType retryType) => static async Task RecordUnacknowledged(ServiceControlDbContext dbContext, string requestId, RetryType retryType, DateTime startTime, DateTime completionTime, string originator, string classifier, bool messageFailed, - int numberOfMessagesProcessed, DateTime lastProcessed) + int numberOfMessagesProcessed, DateTime lastProcessed, CancellationToken cancellationToken) { var unacknowledged = await dbContext.UnacknowledgedRetryOperations - .SingleOrDefaultAsync(operation => operation.RequestId == requestId && operation.RetryType == retryType); + .SingleOrDefaultAsync(operation => operation.RequestId == requestId && operation.RetryType == retryType, cancellationToken); if (unacknowledged == null) { @@ -121,11 +122,11 @@ static async Task RecordUnacknowledged(ServiceControlDbContext dbContext, string unacknowledged.NumberOfMessagesProcessed = numberOfMessagesProcessed; } - static async Task TrimHistory(ServiceControlDbContext dbContext, int retryHistoryDepth) + static async Task TrimHistory(ServiceControlDbContext dbContext, int retryHistoryDepth, CancellationToken cancellationToken) { if (retryHistoryDepth <= 0) { - await dbContext.HistoricRetryOperations.ExecuteDeleteAsync(); + await dbContext.HistoricRetryOperations.ExecuteDeleteAsync(cancellationToken); return; } @@ -136,7 +137,7 @@ static async Task TrimHistory(ServiceControlDbContext dbContext, int retryHistor .ThenByDescending(operation => operation.Id) .Skip(retryHistoryDepth - 1) .Select(operation => new { operation.CompletionTime, operation.Id }) - .FirstOrDefaultAsync(); + .FirstOrDefaultAsync(cancellationToken); if (cutoff == null) { @@ -146,6 +147,6 @@ static async Task TrimHistory(ServiceControlDbContext dbContext, int retryHistor await dbContext.HistoricRetryOperations .Where(operation => operation.CompletionTime < cutoff.CompletionTime || (operation.CompletionTime == cutoff.CompletionTime && operation.Id < cutoff.Id)) - .ExecuteDeleteAsync(); + .ExecuteDeleteAsync(cancellationToken); } } diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/RetryStagingStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/RetryStagingStore.cs index 467871666e..eda481014e 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/RetryStagingStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/RetryStagingStore.cs @@ -9,7 +9,7 @@ namespace ServiceControl.Persistence.EFCore.Implementation; public class RetryStagingStore(IServiceScopeFactory scopeFactory, TimeProvider timeProvider) : DataStoreBase(scopeFactory), IRetryStagingStore { - public Task GetStagingBatch() => + public Task GetStagingBatch(CancellationToken cancellationToken = default) => ExecuteWithDbContext(async dbContext => { var batch = await dbContext.RetryBatches @@ -17,12 +17,12 @@ public class RetryStagingStore(IServiceScopeFactory scopeFactory, TimeProvider t .Where(batch => batch.Status == RetryBatchStatus.Staging) .OrderBy(batch => batch.StartTime) .ThenBy(batch => batch.Id) - .FirstOrDefaultAsync(); + .FirstOrDefaultAsync(cancellationToken); - return batch?.ToRetryBatch(await CountMessages(dbContext, batch.Id)); + return batch?.ToRetryBatch(await CountMessages(dbContext, batch.Id, cancellationToken)); }); - public Task GetMessagesToStage(string batchId) + public Task GetMessagesToStage(string batchId, CancellationToken cancellationToken = default) { var batch = ParseBatchId(batchId); @@ -44,7 +44,7 @@ public Task GetMessagesToStage(string batchId) message.HeadersJson, retry.StageAttempts }) - .ToListAsync(); + .ToListAsync(cancellationToken); return rows.Select(row => new StagingMessage( row.UniqueMessageId.ToString(), @@ -56,67 +56,67 @@ public Task GetMessagesToStage(string batchId) }); } - public Task MarkBatchAsForwarding(string batchId, string stagingId, IReadOnlyCollection stagedMessageIds) + public Task MarkBatchAsForwarding(string batchId, string stagingId, IReadOnlyCollection stagedMessageIds, CancellationToken cancellationToken = default) { var batch = ParseBatchId(batchId); var staged = ParseMessageIds(stagedMessageIds); var now = timeProvider.GetUtcNow().UtcDateTime; return ExecuteWithDbContext(dbContext => - InTransaction(dbContext, async () => + InTransaction(dbContext, async token => { await dbContext.RetryBatches .Where(row => row.Id == batch) .ExecuteUpdateAsync(setters => setters .SetProperty(row => row.Status, RetryBatchStatus.Forwarding) - .SetProperty(row => row.StagingId, stagingId)); + .SetProperty(row => row.StagingId, stagingId), token); // The batch keeps only what it staged, so its message count is what the forwarder is // told to expect. The claims dropped here are of messages that no longer exist. await dbContext.FailedMessageRetries .Where(row => row.RetryBatchId == batch && !staged.Contains(row.UniqueMessageId)) - .ExecuteDeleteAsync(); + .ExecuteDeleteAsync(token); await dbContext.FailedMessages .Where(row => staged.Contains(row.UniqueMessageId)) .ExecuteUpdateAsync(setters => setters .SetProperty(row => row.Status, FailedMessageStatus.RetryIssued) .SetProperty(row => row.StatusChangedAt, now) - .SetProperty(row => row.LastModified, now)); + .SetProperty(row => row.LastModified, now), token); - await PointForwarderAt(dbContext, batch); - })); + await PointForwarderAt(dbContext, batch, token); + }, cancellationToken)); } - public Task DiscardBatch(string batchId) + public Task DiscardBatch(string batchId, CancellationToken cancellationToken = default) { var batch = ParseBatchId(batchId); return ExecuteWithDbContext(dbContext => - InTransaction(dbContext, async () => + InTransaction(dbContext, async token => { // Nothing was staged, so every claim of this batch is of a message that is gone. await dbContext.FailedMessageRetries .Where(row => row.RetryBatchId == batch) - .ExecuteDeleteAsync(); + .ExecuteDeleteAsync(token); await dbContext.RetryBatches .Where(row => row.Id == batch) - .ExecuteDeleteAsync(); - })); + .ExecuteDeleteAsync(token); + }, cancellationToken)); } - public Task GetForwardingBatchId() => + public Task GetForwardingBatchId(CancellationToken cancellationToken = default) => ExecuteWithDbContext(async dbContext => { var nowForwarding = await dbContext.RetryBatchNowForwarding .AsNoTracking() - .SingleOrDefaultAsync(); + .SingleOrDefaultAsync(cancellationToken); return nowForwarding?.RetryBatchId.ToString(); }); - public Task GetBatch(string batchId, CancellationToken cancellationToken) + public Task GetBatch(string batchId, CancellationToken cancellationToken = default) { var batch = ParseBatchId(batchId); @@ -130,35 +130,35 @@ await dbContext.RetryBatches }); } - public Task CompleteForwarding(string batchId) + public Task CompleteForwarding(string batchId, CancellationToken cancellationToken = default) { var batch = ParseBatchId(batchId); return ExecuteWithDbContext(dbContext => - InTransaction(dbContext, async () => + InTransaction(dbContext, async token => { // The claims outlive the batch: they are what stops a message being staged again // before its retry is confirmed. await dbContext.RetryBatches .Where(row => row.Id == batch) - .ExecuteDeleteAsync(); + .ExecuteDeleteAsync(token); await dbContext.RetryBatchNowForwarding .Where(row => row.RetryBatchId == batch) - .ExecuteDeleteAsync(); - })); + .ExecuteDeleteAsync(token); + }, cancellationToken)); } - public Task RecordStagingFailure(IReadOnlyCollection uniqueMessageIds) + public Task RecordStagingFailure(IReadOnlyCollection uniqueMessageIds, CancellationToken cancellationToken = default) { var failed = ParseMessageIds(uniqueMessageIds); return ExecuteWithDbContext(dbContext => dbContext.FailedMessageRetries .Where(row => failed.Contains(row.UniqueMessageId)) - .ExecuteUpdateAsync(setters => setters.SetProperty(row => row.StageAttempts, 1))); + .ExecuteUpdateAsync(setters => setters.SetProperty(row => row.StageAttempts, 1), cancellationToken)); } - public Task IncrementStagingAttempts(string uniqueMessageId) + public Task IncrementStagingAttempts(string uniqueMessageId, CancellationToken cancellationToken = default) { if (!Guid.TryParse(uniqueMessageId, out var message)) { @@ -167,10 +167,10 @@ public Task IncrementStagingAttempts(string uniqueMessageId) return ExecuteWithDbContext(dbContext => dbContext.FailedMessageRetries .Where(row => row.UniqueMessageId == message) - .ExecuteUpdateAsync(setters => setters.SetProperty(row => row.StageAttempts, row => row.StageAttempts + 1))); + .ExecuteUpdateAsync(setters => setters.SetProperty(row => row.StageAttempts, row => row.StageAttempts + 1), cancellationToken)); } - public Task RemoveFromBatch(string uniqueMessageId) + public Task RemoveFromBatch(string uniqueMessageId, CancellationToken cancellationToken = default) { if (!Guid.TryParse(uniqueMessageId, out var message)) { @@ -179,35 +179,35 @@ public Task RemoveFromBatch(string uniqueMessageId) return ExecuteWithDbContext(dbContext => dbContext.FailedMessageRetries .Where(row => row.UniqueMessageId == message) - .ExecuteDeleteAsync()); + .ExecuteDeleteAsync(cancellationToken)); } - static async Task PointForwarderAt(ServiceControlDbContext dbContext, Guid batch) + static async Task PointForwarderAt(ServiceControlDbContext dbContext, Guid batch, CancellationToken cancellationToken) { var updated = await dbContext.RetryBatchNowForwarding .Where(row => row.Id == RetryBatchNowForwardingEntity.SingleRowId) - .ExecuteUpdateAsync(setters => setters.SetProperty(row => row.RetryBatchId, batch)); + .ExecuteUpdateAsync(setters => setters.SetProperty(row => row.RetryBatchId, batch), cancellationToken); if (updated == 0) { dbContext.RetryBatchNowForwarding.Add(new RetryBatchNowForwardingEntity { RetryBatchId = batch }); - await dbContext.SaveChangesAsync(); + await dbContext.SaveChangesAsync(cancellationToken); } } - static Task CountMessages(ServiceControlDbContext dbContext, Guid batch, CancellationToken cancellationToken = default) => + static Task CountMessages(ServiceControlDbContext dbContext, Guid batch, CancellationToken cancellationToken) => dbContext.FailedMessageRetries .AsNoTracking() .CountAsync(retry => retry.RetryBatchId == batch, cancellationToken); - static Task InTransaction(ServiceControlDbContext dbContext, Func operations) => + static Task InTransaction(ServiceControlDbContext dbContext, Func operations, CancellationToken cancellationToken) => dbContext.Database.CreateExecutionStrategy().ExecuteAsync(async () => { - await using var transaction = await dbContext.Database.BeginTransactionAsync(); + await using var transaction = await dbContext.Database.BeginTransactionAsync(cancellationToken); - await operations(); + await operations(cancellationToken); - await transaction.CommitAsync(); + await transaction.CommitAsync(cancellationToken); }); /// diff --git a/src/ServiceControl.Persistence.RavenDB/Recoverability/RetryHistoryDataStore.cs b/src/ServiceControl.Persistence.RavenDB/Recoverability/RetryHistoryDataStore.cs index 7e270a36af..b93e1c8df8 100644 --- a/src/ServiceControl.Persistence.RavenDB/Recoverability/RetryHistoryDataStore.cs +++ b/src/ServiceControl.Persistence.RavenDB/Recoverability/RetryHistoryDataStore.cs @@ -1,6 +1,7 @@ namespace ServiceControl.Persistence.RavenDB.Recoverability { using System; + using System.Threading; using System.Threading.Tasks; using ServiceControl.Recoverability; @@ -8,10 +9,10 @@ class RetryHistoryDataStore(IRavenSessionProvider sessionProvider) : IRetryHisto { const string DocumentId = "RetryOperations/History"; - public async Task GetRetryHistory() + public async Task GetRetryHistory(CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); - var retryHistory = await session.LoadAsync(DocumentId); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); + var retryHistory = await session.LoadAsync(DocumentId, cancellationToken); retryHistory ??= new(); @@ -19,10 +20,11 @@ public async Task GetRetryHistory() } public async Task RecordRetryOperationCompleted(string requestId, RetryType retryType, DateTime startTime, DateTime completionTime, - string originator, string classifier, bool messageFailed, int numberOfMessagesProcessed, DateTime lastProcessed, int retryHistoryDepth) + string originator, string classifier, bool messageFailed, int numberOfMessagesProcessed, DateTime lastProcessed, int retryHistoryDepth, + CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); - var retryHistory = await session.LoadAsync(DocumentId) ?? new(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); + var retryHistory = await session.LoadAsync(DocumentId, cancellationToken) ?? new(); retryHistory.AddToUnacknowledged(new UnacknowledgedRetryOperation { @@ -48,20 +50,20 @@ public async Task RecordRetryOperationCompleted(string requestId, RetryType retr NumberOfMessagesProcessed = numberOfMessagesProcessed }, retryHistoryDepth); - await session.StoreAsync(retryHistory, DocumentId); - await session.SaveChangesAsync(); + await session.StoreAsync(retryHistory, DocumentId, cancellationToken); + await session.SaveChangesAsync(cancellationToken); } - public async Task AcknowledgeRetryGroup(string groupId) + public async Task AcknowledgeRetryGroup(string groupId, CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); - var retryHistory = await session.LoadAsync(DocumentId); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); + var retryHistory = await session.LoadAsync(DocumentId, cancellationToken); if (retryHistory != null) { if (retryHistory.Acknowledge(groupId, RetryType.FailureGroup)) { - await session.StoreAsync(retryHistory, DocumentId); - await session.SaveChangesAsync(); + await session.StoreAsync(retryHistory, DocumentId, cancellationToken); + await session.SaveChangesAsync(cancellationToken); return true; } diff --git a/src/ServiceControl.Persistence.RavenDB/RetryDocumentDataStore.cs b/src/ServiceControl.Persistence.RavenDB/RetryDocumentDataStore.cs index 42629c6b89..162565009e 100644 --- a/src/ServiceControl.Persistence.RavenDB/RetryDocumentDataStore.cs +++ b/src/ServiceControl.Persistence.RavenDB/RetryDocumentDataStore.cs @@ -3,6 +3,7 @@ using System; using System.Collections.Generic; using System.Linq; + using System.Threading; using System.Threading.Tasks; using MessageFailures; using Microsoft.Extensions.Logging; @@ -17,7 +18,7 @@ class RetryDocumentDataStore(IRavenSessionProvider sessionProvider, IRavenDocumentStoreProvider documentStoreProvider, ILogger logger) : IRetryBatchStore { - public async Task AssignMessagesToBatch(string batchId, string[] messageIds) + public async Task AssignMessagesToBatch(string batchId, string[] messageIds, CancellationToken cancellationToken = default) { var commands = new ICommandData[messageIds.Length]; @@ -26,17 +27,17 @@ public async Task AssignMessagesToBatch(string batchId, string[] messageIds) commands[i] = CreateFailedMessageRetryDocument(batchId, messageIds[i]); } - using var session = await sessionProvider.OpenSession(); - var documentStore = await documentStoreProvider.GetDocumentStore(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); + var documentStore = await documentStoreProvider.GetDocumentStore(cancellationToken); var batch = new SingleNodeBatchCommand(documentStore.Conventions, session.Advanced.Context, commands); - await session.Advanced.RequestExecutor.ExecuteAsync(batch, session.Advanced.Context); + await session.Advanced.RequestExecutor.ExecuteAsync(batch, session.Advanced.Context, token: cancellationToken); } - public async Task MoveBatchToStaging(string batchId) + public async Task MoveBatchToStaging(string batchId, CancellationToken cancellationToken = default) { try { - var documentStore = await documentStoreProvider.GetDocumentStore(); + var documentStore = await documentStoreProvider.GetDocumentStore(cancellationToken); await documentStore.Operations.SendAsync(new PatchOperation(batchId, null, new PatchRequest { Script = @"this.Status = args.Status", @@ -44,7 +45,7 @@ public async Task MoveBatchToStaging(string batchId) { {"Status", (int)RetryBatchStatus.Staging } } - })); + }), token: cancellationToken); } catch (ConcurrencyException) { @@ -55,11 +56,12 @@ public async Task MoveBatchToStaging(string batchId) public async Task CreateBatch(string retrySessionId, string requestId, RetryType retryType, string[] failedMessageRetryIds, string originator, DateTime startTime, DateTime? last = null, string batchName = null, string classifier = null, - string initiatedById = null, string initiatedByName = null, string operationId = null) + string initiatedById = null, string initiatedByName = null, string operationId = null, + CancellationToken cancellationToken = default) { var batchId = MakeDocumentId(Guid.NewGuid().ToString()); failedMessageRetryIds = failedMessageRetryIds.Select(MakeFailedMessageRetriesDocumentId).ToArray(); - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); await session.StoreAsync(new RetryBatch { Id = batchId, @@ -77,31 +79,31 @@ await session.StoreAsync(new RetryBatch InitiatedById = initiatedById, InitiatedByName = initiatedByName, OperationId = operationId - }); - await session.SaveChangesAsync(); + }, cancellationToken); + await session.SaveChangesAsync(cancellationToken); return batchId; } - public async Task>> GetOrphanedBatches(string retrySessionId) + public async Task>> GetOrphanedBatches(string retrySessionId, CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var orphanedBatches = await session .Query() .Where(b => b.Status == RetryBatchStatus.MarkingDocuments && b.RetrySessionId != retrySessionId) .Statistics(out var stats) - .ToListAsync(); + .ToListAsync(cancellationToken); return orphanedBatches.Select(batch => batch.ToContract()).ToList().ToQueryResult(stats); } - public async Task> GetAvailableBatchGroups() + public async Task> GetAvailableBatchGroups(CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var results = await session.Query() .Where(b => b.HasStagingBatches || b.HasForwardingBatches) - .ToListAsync(); + .ToListAsync(cancellationToken); return results; } @@ -121,9 +123,9 @@ static ICommandData CreateFailedMessageRetryDocument(string batchId, string mess return new PatchCommandData(MakeFailedMessageRetriesDocumentId(messageId), null, patch: new PatchRequest { Script = "" }, patchIfMissing: patchRequest); } - public async Task ForEachUnresolvedMessage(Func callback) + public async Task ForEachUnresolvedMessage(Func callback, CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var query = session.Query() .Where(d => d.Status == FailedMessageStatus.Unresolved) .Select(m => new @@ -132,17 +134,17 @@ public async Task ForEachUnresolvedMessage(Func callback LatestTimeOfFailure = m.TimeOfFailure }); - await using var stream = await session.Advanced.StreamAsync(query); + await using var stream = await session.Advanced.StreamAsync(query, cancellationToken); while (await stream.MoveNextAsync()) { var current = stream.Current.Document; - await callback(current.UniqueMessageId, current.LatestTimeOfFailure); + await callback(current.UniqueMessageId, current.LatestTimeOfFailure, cancellationToken); } } - public async Task ForEachUnresolvedMessageForEndpoint(string endpoint, Func callback) + public async Task ForEachUnresolvedMessageForEndpoint(string endpoint, Func callback, CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var query = session.Query() .Where(d => d.Status == FailedMessageStatus.Unresolved) .Where(m => m.ReceivingEndpointName == endpoint) @@ -152,17 +154,17 @@ public async Task ForEachUnresolvedMessageForEndpoint(string endpoint, Func callback) + public async Task ForEachMessageForQueueAddress(string failedQueueAddress, FailedMessageStatus status, Func callback, CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var query = session.Query() .Where(d => d.Status == FailedMessageStatus.Unresolved) .Where(m => m.QueueAddress == failedQueueAddress && m.Status == status) @@ -172,17 +174,17 @@ public async Task ForEachMessageForQueueAddress(string failedQueueAddress, Faile LatestTimeOfFailure = m.TimeOfFailure }); - await using var stream = await session.Advanced.StreamAsync(query); + await using var stream = await session.Advanced.StreamAsync(query, cancellationToken); while (await stream.MoveNextAsync()) { var current = stream.Current.Document; - await callback(current.UniqueMessageId, current.LatestTimeOfFailure); + await callback(current.UniqueMessageId, current.LatestTimeOfFailure, cancellationToken); } } - public async Task ForEachUnresolvedMessageInGroup(string groupId, Func callback) + public async Task ForEachUnresolvedMessageInGroup(string groupId, Func callback, CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var query = session.Query() .Where(d => d.Status == FailedMessageStatus.Unresolved) .Where(m => m.FailureGroupId == groupId) @@ -192,26 +194,26 @@ public async Task ForEachUnresolvedMessageInGroup(string groupId, Func GetCurrentForwardingBatch() + public async Task GetCurrentForwardingBatch(CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var nowForwarding = await session.Include(r => r.RetryBatchId) - .LoadAsync(NowForwardingDocumentId); + .LoadAsync(NowForwardingDocumentId, cancellationToken); if (nowForwarding == null) { return null; } - var batch = await session.LoadAsync(nowForwarding.RetryBatchId); + var batch = await session.LoadAsync(nowForwarding.RetryBatchId, cancellationToken); return batch == null ? null diff --git a/src/ServiceControl.Persistence.RavenDB/RetryStagingStore.cs b/src/ServiceControl.Persistence.RavenDB/RetryStagingStore.cs index 4fa6b69ad6..e6e43b123e 100644 --- a/src/ServiceControl.Persistence.RavenDB/RetryStagingStore.cs +++ b/src/ServiceControl.Persistence.RavenDB/RetryStagingStore.cs @@ -18,35 +18,35 @@ class RetryStagingStore( ExpirationManager expirationManager, ILogger logger) : IRetryStagingStore { - public async Task GetStagingBatch() + public async Task GetStagingBatch(CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var batch = await session.Query() - .FirstOrDefaultAsync(b => b.Status == RetryBatchStatus.Staging); + .FirstOrDefaultAsync(b => b.Status == RetryBatchStatus.Staging, cancellationToken); return batch?.ToContract(); } - public async Task GetMessagesToStage(string batchId) + public async Task GetMessagesToStage(string batchId, CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); - var batch = await session.LoadAsync(batchId); + var batch = await session.LoadAsync(batchId, cancellationToken); if (batch == null) { return []; } - var retries = await session.LoadAsync(batch.FailureRetries); + var retries = await session.LoadAsync(batch.FailureRetries, cancellationToken); // A message claimed by an earlier batch keeps that claim, so this batch does not stage it. var claims = retries.Values .Where(retry => retry != null && retry.RetryBatchId == batchId) .ToArray(); - var messages = await session.LoadAsync(claims.Select(claim => claim.FailedMessageId)); + var messages = await session.LoadAsync(claims.Select(claim => claim.FailedMessageId), cancellationToken); return claims .Select(claim => new { Claim = claim, Message = messages[claim.FailedMessageId] }) @@ -68,11 +68,11 @@ static StagingMessage ToStagingMessage(FailedMessage message, int stageAttempts) stageAttempts); } - public async Task MarkBatchAsForwarding(string batchId, string stagingId, IReadOnlyCollection stagedMessageIds) + public async Task MarkBatchAsForwarding(string batchId, string stagingId, IReadOnlyCollection stagedMessageIds, CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); - var batch = await session.LoadAsync(batchId); + var batch = await session.LoadAsync(batchId, cancellationToken); if (batch == null) { @@ -95,30 +95,31 @@ public async Task MarkBatchAsForwarding(string batchId, string stagingId, IReadO await session.StoreAsync( new RetryBatchNowForwarding { RetryBatchId = batchId }, - RetryDocumentDataStore.NowForwardingDocumentId); + RetryDocumentDataStore.NowForwardingDocumentId, + cancellationToken); - await session.SaveChangesAsync(); + await session.SaveChangesAsync(cancellationToken); } - public async Task DiscardBatch(string batchId) + public async Task DiscardBatch(string batchId, CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); session.Delete(batchId); - await session.SaveChangesAsync(); + await session.SaveChangesAsync(cancellationToken); } - public async Task GetForwardingBatchId() + public async Task GetForwardingBatchId(CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); - var nowForwarding = await session.LoadAsync(RetryDocumentDataStore.NowForwardingDocumentId); + var nowForwarding = await session.LoadAsync(RetryDocumentDataStore.NowForwardingDocumentId, cancellationToken); return nowForwarding?.RetryBatchId; } - public async Task GetBatch(string batchId, CancellationToken cancellationToken) + public async Task GetBatch(string batchId, CancellationToken cancellationToken = default) { using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); @@ -127,17 +128,17 @@ public async Task GetForwardingBatchId() return batch?.ToContract(); } - public async Task CompleteForwarding(string batchId) + public async Task CompleteForwarding(string batchId, CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); session.Delete(batchId); session.Delete(RetryDocumentDataStore.NowForwardingDocumentId); - await session.SaveChangesAsync(); + await session.SaveChangesAsync(cancellationToken); } - public async Task RecordStagingFailure(IReadOnlyCollection uniqueMessageIds) + public async Task RecordStagingFailure(IReadOnlyCollection uniqueMessageIds, CancellationToken cancellationToken = default) { var commands = uniqueMessageIds .Select(ICommandData (uniqueMessageId) => new PatchCommandData( @@ -152,11 +153,11 @@ public async Task RecordStagingFailure(IReadOnlyCollection uniqueMessage try { - using var session = await sessionProvider.OpenSession(); - var documentStore = await documentStoreProvider.GetDocumentStore(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); + var documentStore = await documentStoreProvider.GetDocumentStore(cancellationToken); var batch = new SingleNodeBatchCommand(documentStore.Conventions, session.Advanced.Context, commands); - await session.Advanced.RequestExecutor.ExecuteAsync(batch, session.Advanced.Context); + await session.Advanced.RequestExecutor.ExecuteAsync(batch, session.Advanced.Context, token: cancellationToken); } catch (ConcurrencyException) { @@ -164,15 +165,15 @@ public async Task RecordStagingFailure(IReadOnlyCollection uniqueMessage } } - public async Task IncrementStagingAttempts(string uniqueMessageId) + public async Task IncrementStagingAttempts(string uniqueMessageId, CancellationToken cancellationToken = default) { try { - var documentStore = await documentStoreProvider.GetDocumentStore(); + var documentStore = await documentStoreProvider.GetDocumentStore(cancellationToken); await documentStore.Operations.SendAsync(new PatchOperation( RetryDocumentDataStore.MakeFailedMessageRetriesDocumentId(uniqueMessageId), null, - new PatchRequest { Script = "this.StageAttempts += 1" })); + new PatchRequest { Script = "this.StageAttempts += 1" }), token: cancellationToken); } catch (ConcurrencyException) { @@ -180,13 +181,13 @@ await documentStore.Operations.SendAsync(new PatchOperation( } } - public async Task RemoveFromBatch(string uniqueMessageId) + public async Task RemoveFromBatch(string uniqueMessageId, CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); await session.Advanced.RequestExecutor.ExecuteAsync( new DeleteDocumentCommand(RetryDocumentDataStore.MakeFailedMessageRetriesDocumentId(uniqueMessageId), null), - session.Advanced.Context); + session.Advanced.Context, token: cancellationToken); } PatchRequest RetryIssuedPatch() diff --git a/src/ServiceControl.Persistence.Tests/EFCore/RetryBatchStoreTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/RetryBatchStoreTests.cs index 1746a49f1d..6ab2896ae8 100644 --- a/src/ServiceControl.Persistence.Tests/EFCore/RetryBatchStoreTests.cs +++ b/src/ServiceControl.Persistence.Tests/EFCore/RetryBatchStoreTests.cs @@ -3,6 +3,7 @@ namespace ServiceControl.Persistence.Tests; using System; using System.Collections.Generic; using System.Linq; +using System.Threading; using System.Threading.Tasks; using Microsoft.EntityFrameworkCore; using NUnit.Framework; @@ -239,11 +240,11 @@ async Task> ClaimedBy(string batchId) return [.. claimed.Select(uniqueMessageId => uniqueMessageId.ToString())]; } - static async Task> Collect(Func, Task> stream) + static async Task> Collect(Func, Task> stream) { var streamed = new List(); - await stream((uniqueMessageId, _) => + await stream((uniqueMessageId, _, _) => { streamed.Add(uniqueMessageId); return Task.CompletedTask; diff --git a/src/ServiceControl.Persistence/IRetryBatchStore.cs b/src/ServiceControl.Persistence/IRetryBatchStore.cs index 61cc4a7b1d..1d1372e882 100644 --- a/src/ServiceControl.Persistence/IRetryBatchStore.cs +++ b/src/ServiceControl.Persistence/IRetryBatchStore.cs @@ -2,6 +2,7 @@ namespace ServiceControl.Persistence { using System; using System.Collections.Generic; + using System.Threading; using System.Threading.Tasks; using Infrastructure; using ServiceControl.MessageFailures; @@ -11,20 +12,21 @@ public interface IRetryBatchStore Task CreateBatch(string retrySessionId, string requestId, RetryType retryType, string[] failedMessageRetryIds, string originator, DateTime startTime, DateTime? last = null, string batchName = null, string classifier = null, - string initiatedById = null, string initiatedByName = null, string operationId = null); + string initiatedById = null, string initiatedByName = null, string operationId = null, + CancellationToken cancellationToken = default); - Task AssignMessagesToBatch(string batchId, string[] messageIds); + Task AssignMessagesToBatch(string batchId, string[] messageIds, CancellationToken cancellationToken = default); - Task MoveBatchToStaging(string batchId); + Task MoveBatchToStaging(string batchId, CancellationToken cancellationToken = default); - Task>> GetOrphanedBatches(string retrySessionId); - Task> GetAvailableBatchGroups(); + Task>> GetOrphanedBatches(string retrySessionId, CancellationToken cancellationToken = default); + Task> GetAvailableBatchGroups(CancellationToken cancellationToken = default); - Task GetCurrentForwardingBatch(); + Task GetCurrentForwardingBatch(CancellationToken cancellationToken = default); - Task ForEachUnresolvedMessage(Func callback); - Task ForEachUnresolvedMessageForEndpoint(string endpoint, Func callback); - Task ForEachMessageForQueueAddress(string failedQueueAddress, FailedMessageStatus status, Func callback); - Task ForEachUnresolvedMessageInGroup(string groupId, Func callback); + Task ForEachUnresolvedMessage(Func callback, CancellationToken cancellationToken = default); + Task ForEachUnresolvedMessageForEndpoint(string endpoint, Func callback, CancellationToken cancellationToken = default); + Task ForEachMessageForQueueAddress(string failedQueueAddress, FailedMessageStatus status, Func callback, CancellationToken cancellationToken = default); + Task ForEachUnresolvedMessageInGroup(string groupId, Func callback, CancellationToken cancellationToken = default); } } diff --git a/src/ServiceControl.Persistence/IRetryHistoryDataStore.cs b/src/ServiceControl.Persistence/IRetryHistoryDataStore.cs index ec3d2c4cb7..a16b7f4d45 100644 --- a/src/ServiceControl.Persistence/IRetryHistoryDataStore.cs +++ b/src/ServiceControl.Persistence/IRetryHistoryDataStore.cs @@ -1,14 +1,16 @@ namespace ServiceControl.Persistence { using System; + using System.Threading; using System.Threading.Tasks; using ServiceControl.Recoverability; public interface IRetryHistoryDataStore { - Task GetRetryHistory(); + Task GetRetryHistory(CancellationToken cancellationToken = default); Task RecordRetryOperationCompleted(string requestId, RetryType retryType, DateTime startTime, DateTime completionTime, - string originator, string classifier, bool messageFailed, int numberOfMessagesProcessed, DateTime lastProcessed, int retryHistoryDepth); - Task AcknowledgeRetryGroup(string groupId); + string originator, string classifier, bool messageFailed, int numberOfMessagesProcessed, DateTime lastProcessed, int retryHistoryDepth, + CancellationToken cancellationToken = default); + Task AcknowledgeRetryGroup(string groupId, CancellationToken cancellationToken = default); } } diff --git a/src/ServiceControl.Persistence/IRetryStagingStore.cs b/src/ServiceControl.Persistence/IRetryStagingStore.cs index da45fb97b1..73669a0967 100644 --- a/src/ServiceControl.Persistence/IRetryStagingStore.cs +++ b/src/ServiceControl.Persistence/IRetryStagingStore.cs @@ -14,52 +14,52 @@ public interface IRetryStagingStore /// /// The batch waiting to be staged, or null when there is nothing to stage. /// - Task GetStagingBatch(); + Task GetStagingBatch(CancellationToken cancellationToken = default); /// /// The messages the batch still holds. A message that has since been deleted is not returned, /// so the result can be shorter than what the batch claimed. /// - Task GetMessagesToStage(string batchId); + Task GetMessagesToStage(string batchId, CancellationToken cancellationToken = default); /// /// Hands the batch to the forwarder: the batch keeps only the messages that were staged, those /// messages become , and the batch /// becomes the one being forwarded. /// - Task MarkBatchAsForwarding(string batchId, string stagingId, IReadOnlyCollection stagedMessageIds); + Task MarkBatchAsForwarding(string batchId, string stagingId, IReadOnlyCollection stagedMessageIds, CancellationToken cancellationToken = default); /// /// Drops a batch that has nothing left to stage, because every message it covered was claimed /// by an earlier batch. /// - Task DiscardBatch(string batchId); + Task DiscardBatch(string batchId, CancellationToken cancellationToken = default); /// /// The batch being forwarded, or null when none is. Outlives the batch it names, so the batch /// itself can be gone by the time it is asked for. /// - Task GetForwardingBatchId(); + Task GetForwardingBatchId(CancellationToken cancellationToken = default); - Task GetBatch(string batchId, CancellationToken cancellationToken); + Task GetBatch(string batchId, CancellationToken cancellationToken = default); /// /// Removes the forwarded batch and the pointer to it. Tolerates a batch that is already gone, /// so a pointer left behind by a premature shutdown can always be cleared. /// - Task CompleteForwarding(string batchId); + Task CompleteForwarding(string batchId, CancellationToken cancellationToken = default); /// /// Records that the whole batch failed to reach the transport, so the next attempt at these /// messages knows it is a retry of a retry. /// - Task RecordStagingFailure(IReadOnlyCollection uniqueMessageIds); + Task RecordStagingFailure(IReadOnlyCollection uniqueMessageIds, CancellationToken cancellationToken = default); - Task IncrementStagingAttempts(string uniqueMessageId); + Task IncrementStagingAttempts(string uniqueMessageId, CancellationToken cancellationToken = default); /// /// Releases the message from its batch, leaving it unresolved for a later retry request. /// - Task RemoveFromBatch(string uniqueMessageId); + Task RemoveFromBatch(string uniqueMessageId, CancellationToken cancellationToken = default); } } diff --git a/src/ServiceControl/.editorconfig b/src/ServiceControl/.editorconfig index e2f03528c2..3668c8ea04 100644 --- a/src/ServiceControl/.editorconfig +++ b/src/ServiceControl/.editorconfig @@ -14,12 +14,3 @@ dotnet_diagnostic.PS0018.severity = none # The dispatch callback gains a token when that interface does, in the persistence phase. [ExternalIntegrations/EventDispatcherHostedService.cs] dotnet_diagnostic.PS0018.severity = none - -# Blocked on IRetryBatchStore. BulkRetryRequest.Invoke forwards a Func straight -# into ForEachUnresolvedMessage and friends, so the callback gains a token when that interface does. -# PS0004 is a separate matter: AssignMessagesToBatch is private and would take a required token, but its -# trailing optional parameters make that illegal, so the default stays until the parameters are reordered. -[Recoverability/Retrying/RetriesGateway.cs] -dotnet_diagnostic.PS0004.severity = none -dotnet_diagnostic.PS0013.severity = none -dotnet_diagnostic.PS0018.severity = none diff --git a/src/ServiceControl/Recoverability/API/FailureGroupsController.cs b/src/ServiceControl/Recoverability/API/FailureGroupsController.cs index 2a33edc330..18fb0c29cb 100644 --- a/src/ServiceControl/Recoverability/API/FailureGroupsController.cs +++ b/src/ServiceControl/Recoverability/API/FailureGroupsController.cs @@ -98,7 +98,7 @@ public async Task GetGroupErrorsCount(string groupId, string status = default, s [HttpGet] public async Task GetRetryHistory(CancellationToken cancellationToken = default) { - var retryHistory = await retryStore.GetRetryHistory(); + var retryHistory = await retryStore.GetRetryHistory(cancellationToken); Response.WithDeterministicEtag(retryHistory.GetHistoryOperationsUniqueIdentifier()); diff --git a/src/ServiceControl/Recoverability/API/GroupFetcher.cs b/src/ServiceControl/Recoverability/API/GroupFetcher.cs index d97b930fe7..e417e49725 100644 --- a/src/ServiceControl/Recoverability/API/GroupFetcher.cs +++ b/src/ServiceControl/Recoverability/API/GroupFetcher.cs @@ -21,7 +21,7 @@ public GroupFetcher(IGroupsDataStore store, IRetryHistoryDataStore retryStore, I public async Task GetGroups(string classifier, string classifierFilter, CancellationToken cancellationToken = default) { var dbGroups = await store.GetUnresolvedGroupsByClassifier(classifier, classifierFilter); - var retryHistory = await retryStore.GetRetryHistory(); + var retryHistory = await retryStore.GetRetryHistory(cancellationToken); var unacknowledgedRetries = retryHistory.GetUnacknowledgedByClassifier(classifier); var openRetryAcknowledgements = MapAcksToOpenGroups(dbGroups, unacknowledgedRetries); @@ -34,7 +34,7 @@ public async Task GetGroups(string classifier, string classifi openGroups = MapOpenGroups(openGroups, archiver.GetArchivalOperations()).ToList(); openGroups = openGroups.Where(group => !closedGroups.Any(closedGroup => closedGroup.Id == group.Id)).ToList(); - var currentForwardingBatch = await retryBatchStore.GetCurrentForwardingBatch(); + var currentForwardingBatch = await retryBatchStore.GetCurrentForwardingBatch(cancellationToken); MakeSureForwardingBatchIsIncludedAsOpen(classifier, currentForwardingBatch, openGroups); var groups = openGroups.Union(closedGroups); diff --git a/src/ServiceControl/Recoverability/API/UnacknowledgedGroupsController.cs b/src/ServiceControl/Recoverability/API/UnacknowledgedGroupsController.cs index 7d60ef4254..c6fefcd378 100644 --- a/src/ServiceControl/Recoverability/API/UnacknowledgedGroupsController.cs +++ b/src/ServiceControl/Recoverability/API/UnacknowledgedGroupsController.cs @@ -23,7 +23,7 @@ public async Task AcknowledgeOperation(string groupId, Cancellati return Ok(); } - var success = await retryStore.AcknowledgeRetryGroup(groupId); + var success = await retryStore.AcknowledgeRetryGroup(groupId, cancellationToken); if (success) { diff --git a/src/ServiceControl/Recoverability/Retrying/History/StoreHistoryHandler.cs b/src/ServiceControl/Recoverability/Retrying/History/StoreHistoryHandler.cs index 2a7db48314..61d31f3b6d 100644 --- a/src/ServiceControl/Recoverability/Retrying/History/StoreHistoryHandler.cs +++ b/src/ServiceControl/Recoverability/Retrying/History/StoreHistoryHandler.cs @@ -26,7 +26,8 @@ public Task Handle(RetryOperationCompleted message, CancellationToken cancellati message.Failed, message.NumberOfMessagesProcessed, message.Last, - settings.RetryHistoryDepth); + settings.RetryHistoryDepth, + cancellationToken); } readonly IRetryHistoryDataStore store; diff --git a/src/ServiceControl/Recoverability/Retrying/RetriesGateway.cs b/src/ServiceControl/Recoverability/Retrying/RetriesGateway.cs index 0d75482606..dbb53b04c5 100644 --- a/src/ServiceControl/Recoverability/Retrying/RetriesGateway.cs +++ b/src/ServiceControl/Recoverability/Retrying/RetriesGateway.cs @@ -30,7 +30,7 @@ public async Task StartRetryForSingleMessage(string uniqueMessageId, AuditUser? var numberOfMessages = 1; await operationManager.Preparing(requestId, retryType, numberOfMessages, cancellationToken); - await AssignMessagesToBatch(requestId, retryType, new[] { uniqueMessageId }, DateTime.UtcNow, initiatedBy: initiatedBy, operationId: operationId, cancellationToken: cancellationToken); + await AssignMessagesToBatch(requestId, retryType, new[] { uniqueMessageId }, DateTime.UtcNow, cancellationToken, initiatedBy: initiatedBy, operationId: operationId); await operationManager.PreparedBatch(requestId, retryType, numberOfMessages, cancellationToken); } @@ -43,11 +43,11 @@ public async Task StartRetryForMessageSelection(string[] uniqueMessageIds, Audit var numberOfMessages = uniqueMessageIds.Length; await operationManager.Preparing(requestId, retryType, numberOfMessages, cancellationToken); - await AssignMessagesToBatch(requestId, retryType, uniqueMessageIds, DateTime.UtcNow, initiatedBy: initiatedBy, operationId: operationId, cancellationToken: cancellationToken); + await AssignMessagesToBatch(requestId, retryType, uniqueMessageIds, DateTime.UtcNow, cancellationToken, initiatedBy: initiatedBy, operationId: operationId); await operationManager.PreparedBatch(requestId, retryType, numberOfMessages, cancellationToken); } - async Task AssignMessagesToBatch(string requestId, RetryType retryType, string[] messageIds, DateTime startTime, DateTime? last = null, string originator = null, string batchName = null, string classifier = null, AuditUser? initiatedBy = null, string operationId = null, CancellationToken cancellationToken = default) + async Task AssignMessagesToBatch(string requestId, RetryType retryType, string[] messageIds, DateTime startTime, CancellationToken cancellationToken, DateTime? last = null, string originator = null, string batchName = null, string classifier = null, AuditUser? initiatedBy = null, string operationId = null) { if (messageIds == null || !messageIds.Any()) { @@ -57,11 +57,11 @@ async Task AssignMessagesToBatch(string requestId, RetryType retryType, string[] var failedMessageRetryIds = messageIds.ToArray(); - var batchId = await store.CreateBatch(RetryDocumentManager.RetrySessionId, requestId, retryType, failedMessageRetryIds, originator, startTime, last, batchName, classifier, initiatedBy?.Id, initiatedBy?.Name, operationId); + var batchId = await store.CreateBatch(RetryDocumentManager.RetrySessionId, requestId, retryType, failedMessageRetryIds, originator, startTime, last, batchName, classifier, initiatedBy?.Id, initiatedBy?.Name, operationId, cancellationToken); logger.LogInformation("Created Batch '{BatchDocumentId}' with {BatchMessageCount} messages for '{BatchName}'", batchId, messageIds.Length, batchName); - await store.AssignMessagesToBatch(batchId, messageIds); + await store.AssignMessagesToBatch(batchId, messageIds, cancellationToken); await MoveBatchToStaging(batchId, cancellationToken); @@ -69,7 +69,7 @@ async Task AssignMessagesToBatch(string requestId, RetryType retryType, string[] } // Needs to be overridable by a test - protected virtual Task MoveBatchToStaging(string batchId, CancellationToken cancellationToken = default) => store.MoveBatchToStaging(batchId); + protected virtual Task MoveBatchToStaging(string batchId, CancellationToken cancellationToken = default) => store.MoveBatchToStaging(batchId, cancellationToken); public async Task ProcessNextBulkRetry(CancellationToken cancellationToken = default) // Invoked from BulkRetryBatchCreationHostedService in schedule @@ -96,7 +96,7 @@ async Task ProcessRequest(BulkRetryRequest request, CancellationToken cancellati for (var i = 0; i < batches.Count; i++) { - await AssignMessagesToBatch(request.RequestId, request.RetryType, batches[i], request.StartTime, latestAttempt, request.Originator, GetBatchName(i + 1, batches.Count, request.Originator), request.Classifier, request.InitiatedBy, request.OperationId, cancellationToken); + await AssignMessagesToBatch(request.RequestId, request.RetryType, batches[i], request.StartTime, cancellationToken, latestAttempt, request.Originator, GetBatchName(i + 1, batches.Count, request.Originator), request.Classifier, request.InitiatedBy, request.OperationId); numberOfMessagesAdded += batches[i].Length; await operationManager.PreparedBatch(request.RequestId, request.RetryType, numberOfMessagesAdded, cancellationToken); @@ -175,7 +175,7 @@ public BulkRetryRequest( OperationId = operationId; } - protected abstract Task Invoke(IRetryBatchStore store, Func callback); + protected abstract Task Invoke(IRetryBatchStore store, Func callback, CancellationToken cancellationToken = default); public async Task, DateTime>> GetRequestedBatches(IRetryBatchStore store, CancellationToken cancellationToken = default) { @@ -183,7 +183,7 @@ public async Task, DateTime>> GetRequestedBatches(IRetryBat var currentBatch = new HashSet(StringComparer.OrdinalIgnoreCase); var latestAttempt = DateTime.MinValue; - Task Process(string uniqueMessageId, DateTime latestTimeOfFailure) + Task Process(string uniqueMessageId, DateTime latestTimeOfFailure, CancellationToken cancellationToken) { currentBatch.Add(uniqueMessageId); @@ -203,7 +203,7 @@ Task Process(string uniqueMessageId, DateTime latestTimeOfFailure) return Task.FromResult(response); } - await Invoke(store, Process); + await Invoke(store, Process, cancellationToken); if (currentBatch.Count > 0) { @@ -221,9 +221,9 @@ public RetryForAllMessages(AuditUser? initiatedBy = null, string operationId = n { } - protected override Task Invoke(IRetryBatchStore store, Func callback) + protected override Task Invoke(IRetryBatchStore store, Func callback, CancellationToken cancellationToken = default) { - return store.ForEachUnresolvedMessage(callback); + return store.ForEachUnresolvedMessage(callback, cancellationToken); } } @@ -236,9 +236,9 @@ class RetryForEndpoint : BulkRetryRequest Endpoint = endpoint; } - protected override Task Invoke(IRetryBatchStore store, Func callback) + protected override Task Invoke(IRetryBatchStore store, Func callback, CancellationToken cancellationToken = default) { - return store.ForEachUnresolvedMessageForEndpoint(Endpoint, callback); + return store.ForEachUnresolvedMessageForEndpoint(Endpoint, callback, cancellationToken); } } @@ -255,9 +255,9 @@ public RetryForFailureGroup(string groupId, string groupTitle, string groupType, GroupTitle = groupTitle; } - protected override Task Invoke(IRetryBatchStore store, Func callback) + protected override Task Invoke(IRetryBatchStore store, Func callback, CancellationToken cancellationToken = default) { - return store.ForEachUnresolvedMessageInGroup(GroupId, callback); + return store.ForEachUnresolvedMessageInGroup(GroupId, callback, cancellationToken); } } @@ -278,9 +278,9 @@ public RetryForFailedQueueAddress( Status = status; } - protected override Task Invoke(IRetryBatchStore store, Func callback) + protected override Task Invoke(IRetryBatchStore store, Func callback, CancellationToken cancellationToken = default) { - return store.ForEachMessageForQueueAddress(FailedQueueAddress, Status, callback); + return store.ForEachMessageForQueueAddress(FailedQueueAddress, Status, callback, cancellationToken); } } } diff --git a/src/ServiceControl/Recoverability/Retrying/RetryDocumentManager.cs b/src/ServiceControl/Recoverability/Retrying/RetryDocumentManager.cs index 0315e65297..6c67c78268 100644 --- a/src/ServiceControl/Recoverability/Retrying/RetryDocumentManager.cs +++ b/src/ServiceControl/Recoverability/Retrying/RetryDocumentManager.cs @@ -20,7 +20,7 @@ public RetryDocumentManager(IHostApplicationLifetime applicationLifetime, IRetry public async Task AdoptOrphanedBatches(CancellationToken cancellationToken = default) { - var orphanedBatches = await store.GetOrphanedBatches(RetrySessionId); + var orphanedBatches = await store.GetOrphanedBatches(RetrySessionId, cancellationToken); logger.LogInformation("Found {OrphanedBatchCount} orphaned retry batches from previous sessions", orphanedBatches.Results.Count); @@ -47,11 +47,11 @@ await Task.WhenAll(orphanedBatches.Results.Select(b => Task.Run(async () => return orphanedBatches.QueryStats.IsStale || orphanedBatches.Results.Any(); } - public virtual Task MoveBatchToStaging(string batchId, CancellationToken cancellationToken = default) => store.MoveBatchToStaging(batchId); + public virtual Task MoveBatchToStaging(string batchId, CancellationToken cancellationToken = default) => store.MoveBatchToStaging(batchId, cancellationToken); public async Task RebuildRetryOperationState(CancellationToken cancellationToken = default) { - var stagingBatchGroups = await store.GetAvailableBatchGroups(); + var stagingBatchGroups = await store.GetAvailableBatchGroups(cancellationToken); foreach (var group in stagingBatchGroups) { diff --git a/src/ServiceControl/Recoverability/Retrying/RetryProcessor.cs b/src/ServiceControl/Recoverability/Retrying/RetryProcessor.cs index c8d231b6be..c61b08adcf 100644 --- a/src/ServiceControl/Recoverability/Retrying/RetryProcessor.cs +++ b/src/ServiceControl/Recoverability/Retrying/RetryProcessor.cs @@ -54,7 +54,7 @@ async Task MoveStagedBatchesToForwardingBatch(CancellationToken cancellati isRecoveringFromPrematureShutdown = false; - var stagingBatch = await store.GetStagingBatch(); + var stagingBatch = await store.GetStagingBatch(cancellationToken); if (stagingBatch != null) { @@ -85,7 +85,7 @@ async Task ForwardCurrentBatch(CancellationToken cancellationToken) { logger.LogDebug("Looking for batch to forward"); - var forwardingBatchId = await store.GetForwardingBatchId(); + var forwardingBatchId = await store.GetForwardingBatchId(cancellationToken); if (forwardingBatchId == null) { @@ -112,7 +112,7 @@ async Task ForwardCurrentBatch(CancellationToken cancellationToken) logger.LogDebug("Removing forwarding pointer"); - await store.CompleteForwarding(forwardingBatchId); + await store.CompleteForwarding(forwardingBatchId, cancellationToken); return true; } @@ -159,12 +159,12 @@ async Task Stage(RetryBatch stagingBatch, CancellationToken cancellationTok { var stagingId = Guid.NewGuid().ToString(); - var messagesToStage = await store.GetMessagesToStage(stagingBatch.Id); + var messagesToStage = await store.GetMessagesToStage(stagingBatch.Id, cancellationToken); if (messagesToStage.Length == 0) { logger.LogInformation("Retry batch {RetryBatchId} cancelled as it has no messages left to stage", stagingBatch.Id); - await store.DiscardBatch(stagingBatch.Id); + await store.DiscardBatch(stagingBatch.Id, cancellationToken); return 0; } @@ -195,7 +195,7 @@ await domainEvents.Raise(new MessagesSubmittedForRetry }, cancellationToken); } - await store.MarkBatchAsForwarding(stagingBatch.Id, stagingId, [.. stageAttemptsById.Keys]); + await store.MarkBatchAsForwarding(stagingBatch.Id, stagingId, [.. stageAttemptsById.Keys], cancellationToken); logger.LogInformation("Retry batch {RetryBatchId} staged with Staging Id {StagingId} and {RetryFailureCount} matching failure retries", stagingBatch.Id, stagingId, messagesToStage.Length); return messagesToStage.Length; @@ -265,7 +265,7 @@ async Task BatchDispatchToTransport(string batchId, TransportOperation[] transpo { logger.LogWarning(e, "Attempt 1 of {MaxStagingAttempts} to stage the {MessageCount} messages of retry batch {RetryBatchId} failed", MaxStagingAttempts, messages.Count, batchId); - await store.RecordStagingFailure([.. messages.Select(message => message.UniqueMessageId)]); + await store.RecordStagingFailure([.. messages.Select(message => message.UniqueMessageId)], cancellationToken); throw new RetryStagingException(e); } @@ -291,13 +291,13 @@ async Task TryStageMessage(TransportOperation transportOperation, IReadOnlyDicti { logger.LogWarning(e, "Attempt {StagingRetryAttempt} of {StagingRetryLimit} to stage a retry message {RetryMessageId} failed", incrementedAttempts, MaxStagingAttempts, uniqueMessageId); - await store.IncrementStagingAttempts(uniqueMessageId); + await store.IncrementStagingAttempts(uniqueMessageId, cancellationToken); } else { logger.LogError(e, "Retry message {RetryMessageId} reached its staging retry limit ({StagingRetryLimit}) and is going to be removed from the batch", uniqueMessageId, MaxStagingAttempts); - await store.RemoveFromBatch(uniqueMessageId); + await store.RemoveFromBatch(uniqueMessageId, cancellationToken); await domainEvents.Raise(new MessageFailedInStaging {