From 5ab15f65f4b8af42d4640efeaa134d562555d492 Mon Sep 17 00:00:00 2001 From: John Simons Date: Tue, 11 Aug 2026 22:42:38 +1000 Subject: [PATCH] Propagate cancellation tokens through the archiving operations This change continues the effort to consistently apply optional `CancellationToken` parameters to asynchronous methods within `IArchiveMessages`, the in-memory archive and unarchive operations, the archiving and unarchiving managers, the RavenDB document managers, and both `MessageArchiver` implementations, along with the controllers and handlers that consume them. Cancellation tokens are now propagated to the underlying asynchronous database calls and to the domain events raised as an operation progresses, which retires eight `CancellationToken.None` arguments that existed only because no token reached that far. The wait-for-index calls in both RavenDB document managers now rethrow cancellation ahead of their catch-all, so a shutdown is no longer reported as an index that failed to catch up. No `.editorconfig` suppressions are removed here, because every rule in these projects still has violations from the interface groups that remain. --- .../Recoverability/EFCoreArchivingManager.cs | 20 ++--- .../EFCoreUnarchivingManager.cs | 20 ++--- .../Recoverability/MessageArchiver.cs | 68 ++++++++-------- .../Archiving/ArchiveDocumentManager.cs | 39 +++++---- .../Archiving/ArchivingManager.cs | 21 ++--- .../Archiving/MessageArchiver.cs | 81 ++++++++++--------- .../Archiving/UnarchiveDocumentManager.cs | 43 +++++----- .../Archiving/UnarchivingManager.cs | 21 ++--- .../Archiving/IArchiveMessages.cs | 9 ++- .../Archiving/InMemoryArchive.cs | 16 ++-- .../Archiving/InMemoryUnarchive.cs | 16 ++-- .../Recoverability/NoopArchiveMessages.cs | 9 ++- .../API/FailureGroupsArchiveController.cs | 2 +- .../API/FailureGroupsUnarchiveController.cs | 2 +- .../Archiving/ArchiveAllInGroupHandler.cs | 2 +- .../Archiving/UnArchiveAllInGroupHandler.cs | 2 +- 16 files changed, 193 insertions(+), 178 deletions(-) diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/Recoverability/EFCoreArchivingManager.cs b/src/ServiceControl.Persistence.EFCore/Implementation/Recoverability/EFCoreArchivingManager.cs index 7d9e29f268..35c6576c34 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/Recoverability/EFCoreArchivingManager.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/Recoverability/EFCoreArchivingManager.cs @@ -22,7 +22,7 @@ InMemoryArchive GetOrCreate(ArchiveType archiveType, string requestId) return summary; } - public Task StartArchiving(ArchiveOperationEntity operation) + public Task StartArchiving(ArchiveOperationEntity operation, CancellationToken cancellationToken = default) { var summary = GetOrCreate(operation.ArchiveType, operation.RequestId); @@ -33,10 +33,10 @@ public Task StartArchiving(ArchiveOperationEntity operation) summary.NumberOfBatches = operation.NumberOfBatches; summary.CurrentBatch = operation.CurrentBatch; - return summary.Start(); + return summary.Start(cancellationToken); } - public Task StartArchiving(string requestId, ArchiveType archiveType) + public Task StartArchiving(string requestId, ArchiveType archiveType, CancellationToken cancellationToken = default) { var summary = GetOrCreate(archiveType, requestId); @@ -47,7 +47,7 @@ public Task StartArchiving(string requestId, ArchiveType archiveType) summary.NumberOfBatches = 0; summary.CurrentBatch = 0; - return summary.Start(); + return summary.Start(cancellationToken); } public InMemoryArchive? GetStatusForArchiveOperation(string requestId, ArchiveType archiveType) @@ -56,22 +56,22 @@ public Task StartArchiving(string requestId, ArchiveType archiveType) return summary; } - public Task BatchArchived(string requestId, ArchiveType archiveType, int numberOfMessagesArchivedInBatch) + public Task BatchArchived(string requestId, ArchiveType archiveType, int numberOfMessagesArchivedInBatch, CancellationToken cancellationToken = default) { var summary = GetOrCreate(archiveType, requestId); - return summary.BatchArchived(numberOfMessagesArchivedInBatch); + return summary.BatchArchived(numberOfMessagesArchivedInBatch, cancellationToken); } - public Task ArchiveOperationFinalizing(string requestId, ArchiveType archiveType) + public Task ArchiveOperationFinalizing(string requestId, ArchiveType archiveType, CancellationToken cancellationToken = default) { var summary = GetOrCreate(archiveType, requestId); - return summary.FinalizeArchive(); + return summary.FinalizeArchive(cancellationToken); } - public Task ArchiveOperationCompleted(string requestId, ArchiveType archiveType) + public Task ArchiveOperationCompleted(string requestId, ArchiveType archiveType, CancellationToken cancellationToken = default) { var summary = GetOrCreate(archiveType, requestId); - return summary.Complete(); + return summary.Complete(cancellationToken); } public bool IsArchiveInProgressFor(string requestId) diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/Recoverability/EFCoreUnarchivingManager.cs b/src/ServiceControl.Persistence.EFCore/Implementation/Recoverability/EFCoreUnarchivingManager.cs index f9ec6a0996..e9c41c9c37 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/Recoverability/EFCoreUnarchivingManager.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/Recoverability/EFCoreUnarchivingManager.cs @@ -22,7 +22,7 @@ InMemoryUnarchive GetOrCreate(ArchiveType archiveType, string requestId) return summary; } - public Task StartUnarchiving(ArchiveOperationEntity operation) + public Task StartUnarchiving(ArchiveOperationEntity operation, CancellationToken cancellationToken = default) { var summary = GetOrCreate(operation.ArchiveType, operation.RequestId); @@ -33,10 +33,10 @@ public Task StartUnarchiving(ArchiveOperationEntity operation) summary.NumberOfBatches = operation.NumberOfBatches; summary.CurrentBatch = operation.CurrentBatch; - return summary.Start(); + return summary.Start(cancellationToken); } - public Task StartUnarchiving(string requestId, ArchiveType archiveType) + public Task StartUnarchiving(string requestId, ArchiveType archiveType, CancellationToken cancellationToken = default) { var summary = GetOrCreate(archiveType, requestId); @@ -47,24 +47,24 @@ public Task StartUnarchiving(string requestId, ArchiveType archiveType) summary.NumberOfBatches = 0; summary.CurrentBatch = 0; - return summary.Start(); + return summary.Start(cancellationToken); } - public Task BatchUnarchived(string requestId, ArchiveType archiveType, int numberOfMessagesUnarchivedInBatch) + public Task BatchUnarchived(string requestId, ArchiveType archiveType, int numberOfMessagesUnarchivedInBatch, CancellationToken cancellationToken = default) { var summary = GetOrCreate(archiveType, requestId); - return summary.BatchUnarchived(numberOfMessagesUnarchivedInBatch); + return summary.BatchUnarchived(numberOfMessagesUnarchivedInBatch, cancellationToken); } - public Task UnarchiveOperationFinalizing(string requestId, ArchiveType archiveType) + public Task UnarchiveOperationFinalizing(string requestId, ArchiveType archiveType, CancellationToken cancellationToken = default) { var summary = GetOrCreate(archiveType, requestId); - return summary.FinalizeUnarchive(); + return summary.FinalizeUnarchive(cancellationToken); } - public Task UnarchiveOperationCompleted(string requestId, ArchiveType archiveType) + public Task UnarchiveOperationCompleted(string requestId, ArchiveType archiveType, CancellationToken cancellationToken = default) { var summary = GetOrCreate(archiveType, requestId); - return summary.Complete(); + return summary.Complete(cancellationToken); } } \ No newline at end of file diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/Recoverability/MessageArchiver.cs b/src/ServiceControl.Persistence.EFCore/Implementation/Recoverability/MessageArchiver.cs index 381ebd4b8d..58ca73ded2 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/Recoverability/MessageArchiver.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/Recoverability/MessageArchiver.cs @@ -33,7 +33,7 @@ ILogger logger unarchivingManager = new EFCoreUnarchivingManager(domainEvents, operationsManager); } - public async Task ArchiveAllInGroup(string groupId, AuditUser? initiatedBy = null, string? operationId = null) + public async Task ArchiveAllInGroup(string groupId, AuditUser? initiatedBy = null, string? operationId = null, CancellationToken cancellationToken = default) { logger.LogInformation("Archiving of {GroupId} started", groupId); @@ -45,7 +45,7 @@ public async Task ArchiveAllInGroup(string groupId, AuditUser? initiatedBy = nul { var dbContext = scope.ServiceProvider.GetRequiredService(); - operationEntity = await GetOrCreateOperation(dbContext, groupId, ArchiveOperationType.Archive, initiatedBy, operationId); + operationEntity = await GetOrCreateOperation(dbContext, groupId, ArchiveOperationType.Archive, initiatedBy, operationId, cancellationToken); if (operationEntity == null) { return; @@ -57,7 +57,7 @@ public async Task ArchiveAllInGroup(string groupId, AuditUser? initiatedBy = nul } // ── Start in-memory tracking ── - await archivingManager.StartArchiving(operationEntity); + await archivingManager.StartArchiving(operationEntity, cancellationToken); // ── Batch loop ── string[] batchIds; do @@ -65,19 +65,19 @@ public async Task ArchiveAllInGroup(string groupId, AuditUser? initiatedBy = nul await using var batchScope = scopeFactory.CreateAsyncScope(); var batchDbContext = batchScope.ServiceProvider.GetRequiredService(); - operationEntity = await batchDbContext.ArchiveOperations.FindAsync(groupId, ArchiveType.FailureGroup, ArchiveOperationType.Archive) + operationEntity = await batchDbContext.ArchiveOperations.FindAsync([groupId, ArchiveType.FailureGroup, ArchiveOperationType.Archive], cancellationToken) ?? throw new InvalidOperationException($"No in progress Archive Operation found for {groupId}"); - batchIds = await UpdateGroupStatusAsync(batchDbContext, groupId, FailedMessageStatus.Unresolved, FailedMessageStatus.Archived, batchSize); - await archivingManager.BatchArchived(groupId, ArchiveType.FailureGroup, batchIds.Length); + batchIds = await UpdateGroupStatusAsync(batchDbContext, groupId, FailedMessageStatus.Unresolved, FailedMessageStatus.Archived, batchSize, cancellationToken); + await archivingManager.BatchArchived(groupId, ArchiveType.FailureGroup, batchIds.Length, cancellationToken); // Update progress tracking operationEntity.CurrentBatch++; operationEntity.NumberOfMessagesProcessed += batchIds.Length; - await batchDbContext.SaveChangesAsync(); + await batchDbContext.SaveChangesAsync(cancellationToken); // Raise batch domain event - await domainEvents.Raise(new FailedMessageGroupBatchArchived { FailedMessagesIds = batchIds }); + await domainEvents.Raise(new FailedMessageGroupBatchArchived { FailedMessagesIds = batchIds }, cancellationToken); // Per-message audit AuditArchivedMessages(MessageActionKind.Archive, Permissions.ErrorRecoverabilityGroupsArchive, auditUser, auditOperationId, batchIds); @@ -87,8 +87,8 @@ public async Task ArchiveAllInGroup(string groupId, AuditUser? initiatedBy = nul // ── Finalize ── logger.LogInformation("Archiving of group {GroupId} is complete", groupId); - await archivingManager.ArchiveOperationFinalizing(groupId, ArchiveType.FailureGroup); - await archivingManager.ArchiveOperationCompleted(groupId, ArchiveType.FailureGroup); + await archivingManager.ArchiveOperationFinalizing(groupId, ArchiveType.FailureGroup, cancellationToken); + await archivingManager.ArchiveOperationCompleted(groupId, ArchiveType.FailureGroup, cancellationToken); // Delete the operation row await using (var finalizeScope = scopeFactory.CreateAsyncScope()) @@ -99,7 +99,7 @@ await finalizeDbContext.ArchiveOperations o.ArchiveType == operationEntity.ArchiveType && o.RequestId == operationEntity.RequestId && o.OperationType == operationEntity.OperationType) - .ExecuteDeleteAsync(); + .ExecuteDeleteAsync(cancellationToken); } await domainEvents.Raise(new FailedMessageGroupArchived @@ -107,12 +107,12 @@ await domainEvents.Raise(new FailedMessageGroupArchived GroupId = groupId, GroupName = operationEntity.GroupName, MessagesCount = operationEntity.TotalNumberOfMessages - }); + }, cancellationToken); logger.LogInformation("Archiving of group {GroupId} completed", groupId); } - public async Task UnarchiveAllInGroup(string groupId, AuditUser? initiatedBy = null, string? operationId = null) + public async Task UnarchiveAllInGroup(string groupId, AuditUser? initiatedBy = null, string? operationId = null, CancellationToken cancellationToken = default) { logger.LogInformation("Unarchiving of {GroupId} started", groupId); @@ -125,7 +125,7 @@ public async Task UnarchiveAllInGroup(string groupId, AuditUser? initiatedBy = n { var dbContext = scope.ServiceProvider.GetRequiredService(); - operationEntity = await GetOrCreateOperation(dbContext, groupId, ArchiveOperationType.UnArchive, initiatedBy, operationId); + operationEntity = await GetOrCreateOperation(dbContext, groupId, ArchiveOperationType.UnArchive, initiatedBy, operationId, cancellationToken); if (operationEntity == null) { return; @@ -136,26 +136,26 @@ public async Task UnarchiveAllInGroup(string groupId, AuditUser? initiatedBy = n auditOperationId = operationEntity.OperationId; } - await unarchivingManager.StartUnarchiving(operationEntity); + await unarchivingManager.StartUnarchiving(operationEntity, cancellationToken); string[] batchIds; do { await using var batchScope = scopeFactory.CreateAsyncScope(); var batchDbContext = batchScope.ServiceProvider.GetRequiredService(); - operationEntity = await batchDbContext.ArchiveOperations.FindAsync(groupId, ArchiveType.FailureGroup, ArchiveOperationType.UnArchive) + operationEntity = await batchDbContext.ArchiveOperations.FindAsync([groupId, ArchiveType.FailureGroup, ArchiveOperationType.UnArchive], cancellationToken) ?? throw new InvalidOperationException($"No in progress Unarchive Operation found for {groupId}"); - batchIds = await UpdateGroupStatusAsync(batchDbContext, groupId, FailedMessageStatus.Archived, FailedMessageStatus.Unresolved, batchSize); + batchIds = await UpdateGroupStatusAsync(batchDbContext, groupId, FailedMessageStatus.Archived, FailedMessageStatus.Unresolved, batchSize, cancellationToken); - await unarchivingManager.BatchUnarchived(groupId, ArchiveType.FailureGroup, batchIds.Length); + await unarchivingManager.BatchUnarchived(groupId, ArchiveType.FailureGroup, batchIds.Length, cancellationToken); // Update progress tracking operationEntity.CurrentBatch++; operationEntity.NumberOfMessagesProcessed += batchIds.Length; - await batchDbContext.SaveChangesAsync(); + await batchDbContext.SaveChangesAsync(cancellationToken); // Raise batch domain event - await domainEvents.Raise(new FailedMessageGroupBatchUnarchived { FailedMessagesIds = batchIds }); + await domainEvents.Raise(new FailedMessageGroupBatchUnarchived { FailedMessagesIds = batchIds }, cancellationToken); // Per-message audit AuditArchivedMessages(MessageActionKind.Unarchive, Permissions.ErrorRecoverabilityGroupsUnarchive, auditUser, auditOperationId, batchIds); @@ -165,8 +165,8 @@ public async Task UnarchiveAllInGroup(string groupId, AuditUser? initiatedBy = n // ── Finalize ── logger.LogInformation("Unarchiving of group {GroupId} is complete", groupId); - await unarchivingManager.UnarchiveOperationFinalizing(groupId, ArchiveType.FailureGroup); - await unarchivingManager.UnarchiveOperationCompleted(groupId, ArchiveType.FailureGroup); + await unarchivingManager.UnarchiveOperationFinalizing(groupId, ArchiveType.FailureGroup, cancellationToken); + await unarchivingManager.UnarchiveOperationCompleted(groupId, ArchiveType.FailureGroup, cancellationToken); // Delete the operation row await using (var finalizeScope = scopeFactory.CreateAsyncScope()) @@ -177,7 +177,7 @@ await finalizeDbContext.ArchiveOperations o.ArchiveType == operationEntity.ArchiveType && o.RequestId == operationEntity.RequestId && o.OperationType == operationEntity.OperationType) - .ExecuteDeleteAsync(); + .ExecuteDeleteAsync(cancellationToken); } await domainEvents.Raise(new FailedMessageGroupUnarchived @@ -185,7 +185,7 @@ await domainEvents.Raise(new FailedMessageGroupUnarchived GroupId = groupId, GroupName = operationEntity.GroupName, MessagesCount = operationEntity.TotalNumberOfMessages - }); + }, cancellationToken); logger.LogInformation("Unarchiving of group {GroupId} completed", groupId); } @@ -199,18 +199,18 @@ public bool IsArchiveInProgressFor(string groupId) public void DismissArchiveOperation(string groupId, ArchiveType archiveType) => archivingManager.DismissArchiveOperation(groupId, archiveType); - public Task StartArchiving(string groupId, ArchiveType archiveType) - => archivingManager.StartArchiving(groupId, archiveType); + public Task StartArchiving(string groupId, ArchiveType archiveType, CancellationToken cancellationToken = default) + => archivingManager.StartArchiving(groupId, archiveType, cancellationToken); - public Task StartUnarchiving(string groupId, ArchiveType archiveType) - => unarchivingManager.StartUnarchiving(groupId, archiveType); + public Task StartUnarchiving(string groupId, ArchiveType archiveType, CancellationToken cancellationToken = default) + => unarchivingManager.StartUnarchiving(groupId, archiveType, cancellationToken); public IEnumerable GetArchivalOperations() => archivingManager.GetArchivalOperations(); - async Task GetOrCreateOperation(ServiceControlDbContext dbContext, string groupId, ArchiveOperationType operation, AuditUser? initiatedBy, string? operationId) + async Task GetOrCreateOperation(ServiceControlDbContext dbContext, string groupId, ArchiveOperationType operation, AuditUser? initiatedBy, string? operationId, CancellationToken cancellationToken) { - ArchiveOperationEntity? operationEntity = await dbContext.ArchiveOperations.FindAsync(groupId, ArchiveType.FailureGroup, operation); + ArchiveOperationEntity? operationEntity = await dbContext.ArchiveOperations.FindAsync([groupId, ArchiveType.FailureGroup, operation], cancellationToken); if (operationEntity != null) { @@ -219,7 +219,7 @@ public IEnumerable GetArchivalOperations() else { var targetStatus = operation == ArchiveOperationType.Archive ? FailedMessageStatus.Unresolved : FailedMessageStatus.Archived; - var (count, groupName) = await GetGroupDetails(dbContext, groupId, targetStatus); + var (count, groupName) = await GetGroupDetails(dbContext, groupId, targetStatus, cancellationToken); if (count == 0) { logger.LogWarning("No messages to {OperationType} in group {GroupId}", operation.ToString(), groupId); @@ -246,7 +246,7 @@ public IEnumerable GetArchivalOperations() try { - await dbContext.SaveChangesAsync(); + await dbContext.SaveChangesAsync(cancellationToken); logger.LogInformation("Group {GroupId} has been split into {NumberOfBatches} batches", groupId, operationEntity.NumberOfBatches); } catch (DbUpdateException ex) when (dbContext.IsDuplicateKeyException(ex)) @@ -276,7 +276,7 @@ void AuditArchivedMessages(MessageActionKind kind, string permission, AuditUser } } - async Task UpdateGroupStatusAsync(ServiceControlDbContext dbContext, string groupId, FailedMessageStatus fromStatus, FailedMessageStatus toStatus, int batchSize, CancellationToken cancellationToken = default) + async Task UpdateGroupStatusAsync(ServiceControlDbContext dbContext, string groupId, FailedMessageStatus fromStatus, FailedMessageStatus toStatus, int batchSize, CancellationToken cancellationToken) { var batchIds = await GetNextBatch(dbContext, groupId, fromStatus, batchSize) .Select(x => x.UniqueMessageId) @@ -299,7 +299,7 @@ await dbContext.FailedMessages } static async Task<(int count, string groupName)> GetGroupDetails( - ServiceControlDbContext dbContext, string groupId, FailedMessageStatus status, CancellationToken cancellationToken = default) + ServiceControlDbContext dbContext, string groupId, FailedMessageStatus status, CancellationToken cancellationToken) { var query = from fmg in dbContext.FailedMessageGroups diff --git a/src/ServiceControl.Persistence.RavenDB/Recoverability/Archiving/ArchiveDocumentManager.cs b/src/ServiceControl.Persistence.RavenDB/Recoverability/Archiving/ArchiveDocumentManager.cs index 81f822c078..e725e5215d 100644 --- a/src/ServiceControl.Persistence.RavenDB/Recoverability/Archiving/ArchiveDocumentManager.cs +++ b/src/ServiceControl.Persistence.RavenDB/Recoverability/Archiving/ArchiveDocumentManager.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; @@ -14,9 +15,9 @@ class ArchiveDocumentManager(ExpirationManager expirationManager, ILogger logger) { - public Task LoadArchiveOperation(IAsyncDocumentSession session, string groupId, ArchiveType archiveType) => session.LoadAsync(ArchiveOperation.MakeId(groupId, archiveType)); + public Task LoadArchiveOperation(IAsyncDocumentSession session, string groupId, ArchiveType archiveType, CancellationToken cancellationToken = default) => session.LoadAsync(ArchiveOperation.MakeId(groupId, archiveType), cancellationToken); - public async Task CreateArchiveOperation(IAsyncDocumentSession session, string groupId, ArchiveType archiveType, int numberOfMessages, string groupName, int batchSize, string initiatedById = null, string initiatedByName = null, string operationId = null) + public async Task CreateArchiveOperation(IAsyncDocumentSession session, string groupId, ArchiveType archiveType, int numberOfMessages, string groupName, int batchSize, string initiatedById = null, string initiatedByName = null, string operationId = null, CancellationToken cancellationToken = default) { var operation = new ArchiveOperation { @@ -34,7 +35,7 @@ public async Task CreateArchiveOperation(IAsyncDocumentSession OperationId = operationId }; - await session.StoreAsync(operation); + await session.StoreAsync(operation, cancellationToken); var documentCount = 0; var indexQuery = session.Query(new FailedMessages_ByGroup().IndexName); @@ -44,7 +45,7 @@ public async Task CreateArchiveOperation(IAsyncDocumentSession .Where(failure => failure.Status == FailedMessageStatus.Unresolved) .Select(document => document.Id); - var docs = await StreamResults(session, docQuery); + var docs = await StreamResults(session, docQuery, cancellationToken); var batches = docs .GroupBy(d => documentCount++ / batchSize); @@ -57,16 +58,16 @@ public async Task CreateArchiveOperation(IAsyncDocumentSession DocumentIds = batch.ToList() }; - await session.StoreAsync(archiveBatch); + await session.StoreAsync(archiveBatch, cancellationToken); } return operation; } - async Task> StreamResults(IAsyncDocumentSession session, IQueryable query) + async Task> StreamResults(IAsyncDocumentSession session, IQueryable query, CancellationToken cancellationToken) { var results = new List(); - await using var enumerator = await session.Advanced.StreamAsync(query); + await using var enumerator = await session.Advanced.StreamAsync(query, cancellationToken); while (await enumerator.MoveNextAsync()) { results.Add(enumerator.Current.Document); @@ -75,12 +76,12 @@ async Task> StreamResults(IAsyncDocumentSession session, IQu return results; } - public Task GetArchiveBatch(IAsyncDocumentSession session, string archiveOperationId, int batchNumber) => session.LoadAsync($"{archiveOperationId}/{batchNumber}"); + public Task GetArchiveBatch(IAsyncDocumentSession session, string archiveOperationId, int batchNumber, CancellationToken cancellationToken = default) => session.LoadAsync($"{archiveOperationId}/{batchNumber}", cancellationToken); - public async Task GetGroupDetails(IAsyncDocumentSession session, string groupId) + public async Task GetGroupDetails(IAsyncDocumentSession session, string groupId, CancellationToken cancellationToken = default) { var group = await session.Query() - .FirstOrDefaultAsync(x => x.Id == groupId); + .FirstOrDefaultAsync(x => x.Id == groupId, cancellationToken); return new GroupDetails { @@ -111,9 +112,9 @@ public void ArchiveMessageGroupBatch(IAsyncDocumentSession session, ArchiveBatch } } - public async Task WaitForIndexUpdateOfArchiveOperation(IRavenSessionProvider sessionProvider, string requestId, TimeSpan timeToWait) + public async Task WaitForIndexUpdateOfArchiveOperation(IRavenSessionProvider sessionProvider, string requestId, TimeSpan timeToWait, CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var indexQuery = session.Query(new FailedMessages_ByGroup().IndexName) .Customize(x => x.WaitForNonStaleResults(timeToWait)); @@ -123,23 +124,27 @@ public async Task WaitForIndexUpdateOfArchiveOperation(IRavenSessionProvid try { - await docQuery.AnyAsync(); + await docQuery.AnyAsync(cancellationToken); return true; } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + throw; + } catch { return false; } } - public Task UpdateArchiveOperation(IAsyncDocumentSession session, ArchiveOperation archiveOperation) => session.StoreAsync(archiveOperation); + public Task UpdateArchiveOperation(IAsyncDocumentSession session, ArchiveOperation archiveOperation, CancellationToken cancellationToken = default) => session.StoreAsync(archiveOperation, cancellationToken); - public async Task RemoveArchiveOperation(IRavenSessionProvider sessionProvider, ArchiveOperation archiveOperation) + public async Task RemoveArchiveOperation(IRavenSessionProvider sessionProvider, ArchiveOperation archiveOperation, CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); session.Advanced.Defer(new DeleteCommandData(archiveOperation.Id, null)); - await session.SaveChangesAsync(); + await session.SaveChangesAsync(cancellationToken); logger.LogInformation("Removing ArchiveOperation {ArchiveOperationId} completed", archiveOperation.Id); } diff --git a/src/ServiceControl.Persistence.RavenDB/Recoverability/Archiving/ArchivingManager.cs b/src/ServiceControl.Persistence.RavenDB/Recoverability/Archiving/ArchivingManager.cs index c98cadb608..873df977b2 100644 --- a/src/ServiceControl.Persistence.RavenDB/Recoverability/Archiving/ArchivingManager.cs +++ b/src/ServiceControl.Persistence.RavenDB/Recoverability/Archiving/ArchivingManager.cs @@ -3,6 +3,7 @@ using System; using System.Collections.Generic; using System.Linq; + using System.Threading; using System.Threading.Tasks; using Infrastructure.DomainEvents; @@ -40,7 +41,7 @@ InMemoryArchive GetOrCreate(ArchiveType archiveType, string requestId) return summary; } - public Task StartArchiving(ArchiveOperation archiveOperation) + public Task StartArchiving(ArchiveOperation archiveOperation, CancellationToken cancellationToken = default) { var summary = GetOrCreate(archiveOperation.ArchiveType, archiveOperation.RequestId); @@ -51,10 +52,10 @@ public Task StartArchiving(ArchiveOperation archiveOperation) summary.NumberOfBatches = archiveOperation.NumberOfBatches; summary.CurrentBatch = archiveOperation.CurrentBatch; - return summary.Start(); + return summary.Start(cancellationToken); } - public Task StartArchiving(string requestId, ArchiveType archiveType) + public Task StartArchiving(string requestId, ArchiveType archiveType, CancellationToken cancellationToken = default) { var summary = GetOrCreate(archiveType, requestId); @@ -65,7 +66,7 @@ public Task StartArchiving(string requestId, ArchiveType archiveType) summary.NumberOfBatches = 0; summary.CurrentBatch = 0; - return summary.Start(); + return summary.Start(cancellationToken); } public InMemoryArchive GetStatusForArchiveOperation(string requestId, ArchiveType archiveType) @@ -75,23 +76,23 @@ public InMemoryArchive GetStatusForArchiveOperation(string requestId, ArchiveTyp return summary; } - public Task BatchArchived(string requestId, ArchiveType archiveType, int numberOfMessagesArchivedInBatch) + public Task BatchArchived(string requestId, ArchiveType archiveType, int numberOfMessagesArchivedInBatch, CancellationToken cancellationToken = default) { var summary = GetOrCreate(archiveType, requestId); - return summary.BatchArchived(numberOfMessagesArchivedInBatch); + return summary.BatchArchived(numberOfMessagesArchivedInBatch, cancellationToken); } - public Task ArchiveOperationFinalizing(string requestId, ArchiveType archiveType) + public Task ArchiveOperationFinalizing(string requestId, ArchiveType archiveType, CancellationToken cancellationToken = default) { var summary = GetOrCreate(archiveType, requestId); - return summary.FinalizeArchive(); + return summary.FinalizeArchive(cancellationToken); } - public Task ArchiveOperationCompleted(string requestId, ArchiveType archiveType) + public Task ArchiveOperationCompleted(string requestId, ArchiveType archiveType, CancellationToken cancellationToken = default) { var summary = GetOrCreate(archiveType, requestId); - return summary.Complete(); + return summary.Complete(cancellationToken); } public void DismissArchiveOperation(string requestId, ArchiveType archiveType) diff --git a/src/ServiceControl.Persistence.RavenDB/Recoverability/Archiving/MessageArchiver.cs b/src/ServiceControl.Persistence.RavenDB/Recoverability/Archiving/MessageArchiver.cs index d0adadba88..022fb17d3a 100644 --- a/src/ServiceControl.Persistence.RavenDB/Recoverability/Archiving/MessageArchiver.cs +++ b/src/ServiceControl.Persistence.RavenDB/Recoverability/Archiving/MessageArchiver.cs @@ -3,6 +3,7 @@ using System; using System.Collections.Generic; using System.Linq; + using System.Threading; using System.Threading.Tasks; using Microsoft.Extensions.Logging; using RavenDB; @@ -36,20 +37,20 @@ ILogger logger unarchivingManager = new UnarchivingManager(domainEvents, operationsManager); } - public async Task ArchiveAllInGroup(string groupId, AuditUser? initiatedBy = null, string operationId = null) + public async Task ArchiveAllInGroup(string groupId, AuditUser? initiatedBy = null, string operationId = null, CancellationToken cancellationToken = default) { logger.LogInformation("Archiving of {GroupId} started", groupId); ArchiveOperation archiveOperation; - using (var session = await sessionProvider.OpenSession()) + using (var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken)) { session.Advanced.UseOptimisticConcurrency = true; // Ensure 2 messages don't split the same operation into batches at once - archiveOperation = await archiveDocumentManager.LoadArchiveOperation(session, groupId, ArchiveType.FailureGroup); + archiveOperation = await archiveDocumentManager.LoadArchiveOperation(session, groupId, ArchiveType.FailureGroup, cancellationToken); if (archiveOperation == null) { - var groupDetails = await archiveDocumentManager.GetGroupDetails(session, groupId); + var groupDetails = await archiveDocumentManager.GetGroupDetails(session, groupId, cancellationToken); if (groupDetails.NumberOfMessagesInGroup == 0) { logger.LogWarning("No messages to archive in group {GroupId}", groupId); @@ -57,8 +58,8 @@ public async Task ArchiveAllInGroup(string groupId, AuditUser? initiatedBy = nul } logger.LogInformation("Splitting group {GroupId} into batches", groupId); - archiveOperation = await archiveDocumentManager.CreateArchiveOperation(session, groupId, ArchiveType.FailureGroup, groupDetails.NumberOfMessagesInGroup, groupDetails.GroupName, batchSize, initiatedBy?.Id, initiatedBy?.Name, operationId); - await session.SaveChangesAsync(); + archiveOperation = await archiveDocumentManager.CreateArchiveOperation(session, groupId, ArchiveType.FailureGroup, groupDetails.NumberOfMessagesInGroup, groupDetails.GroupName, batchSize, initiatedBy?.Id, initiatedBy?.Name, operationId, cancellationToken); + await session.SaveChangesAsync(cancellationToken); logger.LogInformation("Group {GroupId} has been split into {NumberOfBatches} batches", groupId, archiveOperation.NumberOfBatches); } @@ -68,13 +69,13 @@ public async Task ArchiveAllInGroup(string groupId, AuditUser? initiatedBy = nul var auditUser = new AuditUser(archiveOperation.InitiatedById, archiveOperation.InitiatedByName); var auditOperationId = archiveOperation.OperationId; - await archivingManager.StartArchiving(archiveOperation); + await archivingManager.StartArchiving(archiveOperation, cancellationToken); while (archiveOperation.CurrentBatch < archiveOperation.NumberOfBatches) { - using (var batchSession = await sessionProvider.OpenSession()) + using (var batchSession = await sessionProvider.OpenSession(cancellationToken: cancellationToken)) { - var nextBatch = await archiveDocumentManager.GetArchiveBatch(batchSession, archiveOperation.Id, archiveOperation.CurrentBatch); + var nextBatch = await archiveDocumentManager.GetArchiveBatch(batchSession, archiveOperation.Id, archiveOperation.CurrentBatch, cancellationToken); if (nextBatch == null) { // We're only here in the case where Raven indexes are stale @@ -87,13 +88,13 @@ public async Task ArchiveAllInGroup(string groupId, AuditUser? initiatedBy = nul archiveDocumentManager.ArchiveMessageGroupBatch(batchSession, nextBatch); - await archivingManager.BatchArchived(archiveOperation.RequestId, archiveOperation.ArchiveType, nextBatch?.DocumentIds.Count ?? 0); + await archivingManager.BatchArchived(archiveOperation.RequestId, archiveOperation.ArchiveType, nextBatch?.DocumentIds.Count ?? 0, cancellationToken); archiveOperation = archivingManager.GetStatusForArchiveOperation(archiveOperation.RequestId, archiveOperation.ArchiveType).ToArchiveOperation(auditUser.Id, auditUser.Name, auditOperationId); - await archiveDocumentManager.UpdateArchiveOperation(batchSession, archiveOperation); + await archiveDocumentManager.UpdateArchiveOperation(batchSession, archiveOperation, cancellationToken); - await batchSession.SaveChangesAsync(); + await batchSession.SaveChangesAsync(cancellationToken); if (nextBatch != null) { @@ -103,7 +104,7 @@ public async Task ArchiveAllInGroup(string groupId, AuditUser? initiatedBy = nul await domainEvents.Raise(new FailedMessageGroupBatchArchived { FailedMessagesIds = messageIds - }); + }, cancellationToken); AuditArchivedMessages(MessageActionKind.Archive, Permissions.ErrorRecoverabilityGroupsArchive, auditUser, auditOperationId, messageIds); } @@ -116,40 +117,40 @@ await domainEvents.Raise(new FailedMessageGroupBatchArchived } logger.LogInformation("Archiving of group {GroupId} is complete. Waiting for index updates", groupId); - await archivingManager.ArchiveOperationFinalizing(archiveOperation.RequestId, archiveOperation.ArchiveType); - if (!await archiveDocumentManager.WaitForIndexUpdateOfArchiveOperation(sessionProvider, archiveOperation.RequestId, TimeSpan.FromMinutes(5)) + await archivingManager.ArchiveOperationFinalizing(archiveOperation.RequestId, archiveOperation.ArchiveType, cancellationToken); + if (!await archiveDocumentManager.WaitForIndexUpdateOfArchiveOperation(sessionProvider, archiveOperation.RequestId, TimeSpan.FromMinutes(5), cancellationToken) ) { logger.LogWarning("Archiving group {GroupId} completed but index not updated", groupId); } - await archivingManager.ArchiveOperationCompleted(archiveOperation.RequestId, archiveOperation.ArchiveType); - await archiveDocumentManager.RemoveArchiveOperation(sessionProvider, archiveOperation); + await archivingManager.ArchiveOperationCompleted(archiveOperation.RequestId, archiveOperation.ArchiveType, cancellationToken); + await archiveDocumentManager.RemoveArchiveOperation(sessionProvider, archiveOperation, cancellationToken); await domainEvents.Raise(new FailedMessageGroupArchived { GroupId = groupId, GroupName = archiveOperation.GroupName, MessagesCount = archiveOperation.TotalNumberOfMessages, - }); + }, cancellationToken); logger.LogInformation("Archiving of group {GroupId} completed", groupId); } - public async Task UnarchiveAllInGroup(string groupId, AuditUser? initiatedBy = null, string operationId = null) + public async Task UnarchiveAllInGroup(string groupId, AuditUser? initiatedBy = null, string operationId = null, CancellationToken cancellationToken = default) { logger.LogInformation("Unarchiving of {GroupId} started", groupId); UnarchiveOperation unarchiveOperation; - using (var session = await sessionProvider.OpenSession()) + using (var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken)) { session.Advanced.UseOptimisticConcurrency = true; // Ensure 2 messages don't split the same operation into batches at once - unarchiveOperation = await unarchiveDocumentManager.LoadUnarchiveOperation(session, groupId, ArchiveType.FailureGroup); + unarchiveOperation = await unarchiveDocumentManager.LoadUnarchiveOperation(session, groupId, ArchiveType.FailureGroup, cancellationToken); if (unarchiveOperation == null) { - var groupDetails = await unarchiveDocumentManager.GetGroupDetails(session, groupId); + var groupDetails = await unarchiveDocumentManager.GetGroupDetails(session, groupId, cancellationToken); if (groupDetails.NumberOfMessagesInGroup == 0) { logger.LogWarning("No messages to unarchive in group {GroupId}", groupId); @@ -158,8 +159,8 @@ public async Task UnarchiveAllInGroup(string groupId, AuditUser? initiatedBy = n } logger.LogInformation("Splitting group {GroupId} into batches", groupId); - unarchiveOperation = await unarchiveDocumentManager.CreateUnarchiveOperation(session, groupId, ArchiveType.FailureGroup, groupDetails.NumberOfMessagesInGroup, groupDetails.GroupName, batchSize, initiatedBy?.Id, initiatedBy?.Name, operationId); - await session.SaveChangesAsync(); + unarchiveOperation = await unarchiveDocumentManager.CreateUnarchiveOperation(session, groupId, ArchiveType.FailureGroup, groupDetails.NumberOfMessagesInGroup, groupDetails.GroupName, batchSize, initiatedBy?.Id, initiatedBy?.Name, operationId, cancellationToken); + await session.SaveChangesAsync(cancellationToken); logger.LogInformation("Group {GroupId} has been split into {NumberOfBatches} batches", groupId, unarchiveOperation.NumberOfBatches); } @@ -169,12 +170,12 @@ public async Task UnarchiveAllInGroup(string groupId, AuditUser? initiatedBy = n var auditUser = new AuditUser(unarchiveOperation.InitiatedById, unarchiveOperation.InitiatedByName); var auditOperationId = unarchiveOperation.OperationId; - await unarchivingManager.StartUnarchiving(unarchiveOperation); + await unarchivingManager.StartUnarchiving(unarchiveOperation, cancellationToken); while (unarchiveOperation.CurrentBatch < unarchiveOperation.NumberOfBatches) { - using var batchSession = await sessionProvider.OpenSession(); - var nextBatch = await unarchiveDocumentManager.GetUnarchiveBatch(batchSession, unarchiveOperation.Id, unarchiveOperation.CurrentBatch); + using var batchSession = await sessionProvider.OpenSession(cancellationToken: cancellationToken); + var nextBatch = await unarchiveDocumentManager.GetUnarchiveBatch(batchSession, unarchiveOperation.Id, unarchiveOperation.CurrentBatch, cancellationToken); if (nextBatch == null) { // We're only here in the case where Raven indexes are stale @@ -187,13 +188,13 @@ public async Task UnarchiveAllInGroup(string groupId, AuditUser? initiatedBy = n unarchiveDocumentManager.UnarchiveMessageGroupBatch(batchSession, nextBatch, expirationManager); - await unarchivingManager.BatchUnarchived(unarchiveOperation.RequestId, unarchiveOperation.ArchiveType, nextBatch?.DocumentIds.Count ?? 0); + await unarchivingManager.BatchUnarchived(unarchiveOperation.RequestId, unarchiveOperation.ArchiveType, nextBatch?.DocumentIds.Count ?? 0, cancellationToken); unarchiveOperation = unarchivingManager.GetStatusForUnarchiveOperation(unarchiveOperation.RequestId, unarchiveOperation.ArchiveType).ToUnarchiveOperation(auditUser.Id, auditUser.Name, auditOperationId); - await unarchiveDocumentManager.UpdateUnarchiveOperation(batchSession, unarchiveOperation); + await unarchiveDocumentManager.UpdateUnarchiveOperation(batchSession, unarchiveOperation, cancellationToken); - await batchSession.SaveChangesAsync(); + await batchSession.SaveChangesAsync(cancellationToken); if (nextBatch != null) { @@ -203,7 +204,7 @@ public async Task UnarchiveAllInGroup(string groupId, AuditUser? initiatedBy = n await domainEvents.Raise(new FailedMessageGroupBatchUnarchived { FailedMessagesIds = messageIds - }); + }, cancellationToken); AuditArchivedMessages(MessageActionKind.Unarchive, Permissions.ErrorRecoverabilityGroupsUnarchive, auditUser, auditOperationId, messageIds); } @@ -215,23 +216,23 @@ await domainEvents.Raise(new FailedMessageGroupBatchUnarchived } logger.LogInformation("Unarchiving of group {GroupId} is complete. Waiting for index updates", groupId); - await unarchivingManager.UnarchiveOperationFinalizing(unarchiveOperation.RequestId, unarchiveOperation.ArchiveType); - if (!await unarchiveDocumentManager.WaitForIndexUpdateOfUnarchiveOperation(sessionProvider, unarchiveOperation.RequestId, TimeSpan.FromMinutes(5)) + await unarchivingManager.UnarchiveOperationFinalizing(unarchiveOperation.RequestId, unarchiveOperation.ArchiveType, cancellationToken); + if (!await unarchiveDocumentManager.WaitForIndexUpdateOfUnarchiveOperation(sessionProvider, unarchiveOperation.RequestId, TimeSpan.FromMinutes(5), cancellationToken) ) { logger.LogWarning("Unarchiving group {GroupId} completed but index not updated", groupId); } logger.LogInformation("Unarchiving of group {GroupId} completed", groupId); - await unarchivingManager.UnarchiveOperationCompleted(unarchiveOperation.RequestId, unarchiveOperation.ArchiveType); - await unarchiveDocumentManager.RemoveUnarchiveOperation(sessionProvider, unarchiveOperation); + await unarchivingManager.UnarchiveOperationCompleted(unarchiveOperation.RequestId, unarchiveOperation.ArchiveType, cancellationToken); + await unarchiveDocumentManager.RemoveUnarchiveOperation(sessionProvider, unarchiveOperation, cancellationToken); await domainEvents.Raise(new FailedMessageGroupUnarchived { GroupId = groupId, GroupName = unarchiveOperation.GroupName, MessagesCount = unarchiveOperation.TotalNumberOfMessages, - }); + }, cancellationToken); } // Emits one per-message audit entry for each message in a batch, correlated to the initiating @@ -257,11 +258,11 @@ public bool IsArchiveInProgressFor(string groupId) public void DismissArchiveOperation(string groupId, ArchiveType archiveType) => archivingManager.DismissArchiveOperation(groupId, archiveType); - public Task StartArchiving(string groupId, ArchiveType archiveType) - => archivingManager.StartArchiving(groupId, archiveType); + public Task StartArchiving(string groupId, ArchiveType archiveType, CancellationToken cancellationToken = default) + => archivingManager.StartArchiving(groupId, archiveType, cancellationToken); - public Task StartUnarchiving(string groupId, ArchiveType archiveType) - => unarchivingManager.StartUnarchiving(groupId, archiveType); + public Task StartUnarchiving(string groupId, ArchiveType archiveType, CancellationToken cancellationToken = default) + => unarchivingManager.StartUnarchiving(groupId, archiveType, cancellationToken); public IEnumerable GetArchivalOperations() => archivingManager.GetArchivalOperations(); diff --git a/src/ServiceControl.Persistence.RavenDB/Recoverability/Archiving/UnarchiveDocumentManager.cs b/src/ServiceControl.Persistence.RavenDB/Recoverability/Archiving/UnarchiveDocumentManager.cs index 92615e5bf9..3759aa3a6a 100644 --- a/src/ServiceControl.Persistence.RavenDB/Recoverability/Archiving/UnarchiveDocumentManager.cs +++ b/src/ServiceControl.Persistence.RavenDB/Recoverability/Archiving/UnarchiveDocumentManager.cs @@ -3,6 +3,7 @@ using System; using System.Collections.Generic; using System.Linq; + using System.Threading; using System.Threading.Tasks; using MessageFailures; using Persistence.RavenDB; @@ -13,9 +14,9 @@ class UnarchiveDocumentManager { - public Task LoadUnarchiveOperation(IAsyncDocumentSession session, string groupId, ArchiveType archiveType) => session.LoadAsync(UnarchiveOperation.MakeId(groupId, archiveType)); + public Task LoadUnarchiveOperation(IAsyncDocumentSession session, string groupId, ArchiveType archiveType, CancellationToken cancellationToken = default) => session.LoadAsync(UnarchiveOperation.MakeId(groupId, archiveType), cancellationToken); - public async Task CreateUnarchiveOperation(IAsyncDocumentSession session, string groupId, ArchiveType archiveType, int numberOfMessages, string groupName, int batchSize, string initiatedById = null, string initiatedByName = null, string operationId = null) + public async Task CreateUnarchiveOperation(IAsyncDocumentSession session, string groupId, ArchiveType archiveType, int numberOfMessages, string groupName, int batchSize, string initiatedById = null, string initiatedByName = null, string operationId = null, CancellationToken cancellationToken = default) { var operation = new UnarchiveOperation { @@ -33,7 +34,7 @@ public async Task CreateUnarchiveOperation(IAsyncDocumentSes OperationId = operationId }; - await session.StoreAsync(operation); + await session.StoreAsync(operation, cancellationToken); var documentCount = 0; var indexQuery = session.Query(new FailedMessages_ByGroup().IndexName); @@ -43,7 +44,7 @@ public async Task CreateUnarchiveOperation(IAsyncDocumentSes .Where(failure => failure.Status == FailedMessageStatus.Archived) .Select(document => document.Id); - var docs = await StreamResults(session, docQuery); + var docs = await StreamResults(session, docQuery, cancellationToken); var batches = docs .GroupBy(d => documentCount++ / batchSize); @@ -56,16 +57,16 @@ public async Task CreateUnarchiveOperation(IAsyncDocumentSes DocumentIds = batch.ToList() }; - await session.StoreAsync(unarchiveBatch); + await session.StoreAsync(unarchiveBatch, cancellationToken); } return operation; } - async Task> StreamResults(IAsyncDocumentSession session, IQueryable query) + async Task> StreamResults(IAsyncDocumentSession session, IQueryable query, CancellationToken cancellationToken) { var results = new List(); - await using var enumerator = await session.Advanced.StreamAsync(query); + await using var enumerator = await session.Advanced.StreamAsync(query, cancellationToken); while (await enumerator.MoveNextAsync()) { results.Add(enumerator.Current.Document); @@ -74,15 +75,15 @@ async Task> StreamResults(IAsyncDocumentSession session, IQu return results; } - public Task GetUnarchiveBatch(IAsyncDocumentSession session, string unUnarchiveOperationId, int batchNumber) + public Task GetUnarchiveBatch(IAsyncDocumentSession session, string unUnarchiveOperationId, int batchNumber, CancellationToken cancellationToken = default) { - return session.LoadAsync($"{unUnarchiveOperationId}/{batchNumber}"); + return session.LoadAsync($"{unUnarchiveOperationId}/{batchNumber}", cancellationToken); } - public async Task GetGroupDetails(IAsyncDocumentSession session, string groupId) + public async Task GetGroupDetails(IAsyncDocumentSession session, string groupId, CancellationToken cancellationToken = default) { var group = await session.Query() - .FirstOrDefaultAsync(x => x.Id == groupId); + .FirstOrDefaultAsync(x => x.Id == groupId, cancellationToken); return new GroupDetails { @@ -114,9 +115,9 @@ public void UnarchiveMessageGroupBatch(IAsyncDocumentSession session, UnarchiveB } } - public async Task WaitForIndexUpdateOfUnarchiveOperation(IRavenSessionProvider sessionProvider, string requestId, TimeSpan timeToWait) + public async Task WaitForIndexUpdateOfUnarchiveOperation(IRavenSessionProvider sessionProvider, string requestId, TimeSpan timeToWait, CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var indexQuery = session.Query(new FailedMessages_ByGroup().IndexName) .Customize(x => x.WaitForNonStaleResults(timeToWait)); @@ -126,26 +127,30 @@ public async Task WaitForIndexUpdateOfUnarchiveOperation(IRavenSessionProv try { - await docQuery.AnyAsync(); + await docQuery.AnyAsync(cancellationToken); return true; } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + throw; + } catch { return false; } } - public async Task UpdateUnarchiveOperation(IAsyncDocumentSession session, UnarchiveOperation unarchiveOperation) + public async Task UpdateUnarchiveOperation(IAsyncDocumentSession session, UnarchiveOperation unarchiveOperation, CancellationToken cancellationToken = default) { - await session.StoreAsync(unarchiveOperation); + await session.StoreAsync(unarchiveOperation, cancellationToken); } - public async Task RemoveUnarchiveOperation(IRavenSessionProvider sessionProvider, UnarchiveOperation unarchiveOperation) + public async Task RemoveUnarchiveOperation(IRavenSessionProvider sessionProvider, UnarchiveOperation unarchiveOperation, CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); session.Advanced.Defer(new DeleteCommandData(unarchiveOperation.Id, null)); - await session.SaveChangesAsync(); + await session.SaveChangesAsync(cancellationToken); } public class GroupDetails diff --git a/src/ServiceControl.Persistence.RavenDB/Recoverability/Archiving/UnarchivingManager.cs b/src/ServiceControl.Persistence.RavenDB/Recoverability/Archiving/UnarchivingManager.cs index dde10ebfeb..105a4a6ed1 100644 --- a/src/ServiceControl.Persistence.RavenDB/Recoverability/Archiving/UnarchivingManager.cs +++ b/src/ServiceControl.Persistence.RavenDB/Recoverability/Archiving/UnarchivingManager.cs @@ -3,6 +3,7 @@ using System; using System.Collections.Generic; using System.Linq; + using System.Threading; using System.Threading.Tasks; using Infrastructure.DomainEvents; @@ -40,7 +41,7 @@ InMemoryUnarchive GetOrCreate(ArchiveType archiveType, string requestId) return summary; } - public Task StartUnarchiving(UnarchiveOperation archiveOperation) + public Task StartUnarchiving(UnarchiveOperation archiveOperation, CancellationToken cancellationToken = default) { var summary = GetOrCreate(archiveOperation.ArchiveType, archiveOperation.RequestId); @@ -51,10 +52,10 @@ public Task StartUnarchiving(UnarchiveOperation archiveOperation) summary.NumberOfBatches = archiveOperation.NumberOfBatches; summary.CurrentBatch = archiveOperation.CurrentBatch; - return summary.Start(); + return summary.Start(cancellationToken); } - public Task StartUnarchiving(string requestId, ArchiveType archiveType) + public Task StartUnarchiving(string requestId, ArchiveType archiveType, CancellationToken cancellationToken = default) { var summary = GetOrCreate(archiveType, requestId); @@ -65,7 +66,7 @@ public Task StartUnarchiving(string requestId, ArchiveType archiveType) summary.NumberOfBatches = 0; summary.CurrentBatch = 0; - return summary.Start(); + return summary.Start(cancellationToken); } public InMemoryUnarchive GetStatusForUnarchiveOperation(string requestId, ArchiveType archiveType) @@ -75,23 +76,23 @@ public InMemoryUnarchive GetStatusForUnarchiveOperation(string requestId, Archiv return summary; } - public Task BatchUnarchived(string requestId, ArchiveType archiveType, int numberOfMessagesArchivedInBatch) + public Task BatchUnarchived(string requestId, ArchiveType archiveType, int numberOfMessagesArchivedInBatch, CancellationToken cancellationToken = default) { var summary = GetOrCreate(archiveType, requestId); - return summary.BatchUnarchived(numberOfMessagesArchivedInBatch); + return summary.BatchUnarchived(numberOfMessagesArchivedInBatch, cancellationToken); } - public Task UnarchiveOperationFinalizing(string requestId, ArchiveType archiveType) + public Task UnarchiveOperationFinalizing(string requestId, ArchiveType archiveType, CancellationToken cancellationToken = default) { var summary = GetOrCreate(archiveType, requestId); - return summary.FinalizeUnarchive(); + return summary.FinalizeUnarchive(cancellationToken); } - public Task UnarchiveOperationCompleted(string requestId, ArchiveType archiveType) + public Task UnarchiveOperationCompleted(string requestId, ArchiveType archiveType, CancellationToken cancellationToken = default) { var summary = GetOrCreate(archiveType, requestId); - return summary.Complete(); + return summary.Complete(cancellationToken); } public void DismissArchiveOperation(string requestId, ArchiveType archiveType) diff --git a/src/ServiceControl.Persistence/Recoverability/Archiving/IArchiveMessages.cs b/src/ServiceControl.Persistence/Recoverability/Archiving/IArchiveMessages.cs index 763893e9bb..5f60b0c3d5 100644 --- a/src/ServiceControl.Persistence/Recoverability/Archiving/IArchiveMessages.cs +++ b/src/ServiceControl.Persistence/Recoverability/Archiving/IArchiveMessages.cs @@ -1,6 +1,7 @@ namespace ServiceControl.Persistence.Recoverability { using System.Collections.Generic; + using System.Threading; using System.Threading.Tasks; using ServiceControl.Infrastructure.Auth; using ServiceControl.Recoverability; @@ -10,16 +11,16 @@ /// public interface IArchiveMessages { - Task ArchiveAllInGroup(string groupId, AuditUser? initiatedBy = null, string operationId = null); - Task UnarchiveAllInGroup(string groupId, AuditUser? initiatedBy = null, string operationId = null); + Task ArchiveAllInGroup(string groupId, AuditUser? initiatedBy = null, string operationId = null, CancellationToken cancellationToken = default); + Task UnarchiveAllInGroup(string groupId, AuditUser? initiatedBy = null, string operationId = null, CancellationToken cancellationToken = default); bool IsOperationInProgressFor(string groupId, ArchiveType archiveType); bool IsArchiveInProgressFor(string groupId); void DismissArchiveOperation(string groupId, ArchiveType archiveType); - Task StartArchiving(string groupId, ArchiveType archiveType); - Task StartUnarchiving(string groupId, ArchiveType archiveType); + Task StartArchiving(string groupId, ArchiveType archiveType, CancellationToken cancellationToken = default); + Task StartUnarchiving(string groupId, ArchiveType archiveType, CancellationToken cancellationToken = default); IEnumerable GetArchivalOperations(); } diff --git a/src/ServiceControl.Persistence/Recoverability/Archiving/InMemoryArchive.cs b/src/ServiceControl.Persistence/Recoverability/Archiving/InMemoryArchive.cs index 91c9b81f22..1a77aaa457 100644 --- a/src/ServiceControl.Persistence/Recoverability/Archiving/InMemoryArchive.cs +++ b/src/ServiceControl.Persistence/Recoverability/Archiving/InMemoryArchive.cs @@ -41,7 +41,7 @@ public ArchiveProgress GetProgress() return new ArchiveProgress(roundedPercentage, TotalNumberOfMessages, NumberOfMessagesArchived, remaining); } - public Task Start() + public Task Start(CancellationToken cancellationToken = default) { ArchiveState = ArchiveState.ArchiveStarted; CompletionTime = null; @@ -52,10 +52,10 @@ public Task Start() ArchiveType = ArchiveType, Progress = GetProgress(), StartTime = Started - }, CancellationToken.None); + }, cancellationToken); } - public Task BatchArchived(int numberOfMessagesArchivedInBatch) + public Task BatchArchived(int numberOfMessagesArchivedInBatch, CancellationToken cancellationToken = default) { ArchiveState = ArchiveState.ArchiveProgressing; NumberOfMessagesArchived += numberOfMessagesArchivedInBatch; @@ -69,10 +69,10 @@ public Task BatchArchived(int numberOfMessagesArchivedInBatch) Progress = GetProgress(), StartTime = Started, Last = Last.Value - }, CancellationToken.None); + }, cancellationToken); } - public Task FinalizeArchive() + public Task FinalizeArchive(CancellationToken cancellationToken = default) { ArchiveState = ArchiveState.ArchiveFinalizing; NumberOfMessagesArchived = TotalNumberOfMessages; @@ -85,10 +85,10 @@ public Task FinalizeArchive() Progress = GetProgress(), StartTime = Started, Last = Last.Value - }, CancellationToken.None); + }, cancellationToken); } - public Task Complete() + public Task Complete(CancellationToken cancellationToken = default) { ArchiveState = ArchiveState.ArchiveCompleted; NumberOfMessagesArchived = TotalNumberOfMessages; @@ -104,7 +104,7 @@ public Task Complete() Last = Last.Value, CompletionTime = CompletionTime.Value, GroupName = GroupName - }, CancellationToken.None); + }, cancellationToken); } public bool NeedsAcknowledgement() diff --git a/src/ServiceControl.Persistence/Recoverability/Archiving/InMemoryUnarchive.cs b/src/ServiceControl.Persistence/Recoverability/Archiving/InMemoryUnarchive.cs index 0fa0b1f766..48b75179ed 100644 --- a/src/ServiceControl.Persistence/Recoverability/Archiving/InMemoryUnarchive.cs +++ b/src/ServiceControl.Persistence/Recoverability/Archiving/InMemoryUnarchive.cs @@ -41,7 +41,7 @@ public UnarchiveProgress GetProgress() return new UnarchiveProgress(roundedPercentage, TotalNumberOfMessages, NumberOfMessagesUnarchived, remaining); } - public Task Start() + public Task Start(CancellationToken cancellationToken = default) { ArchiveState = ArchiveState.ArchiveStarted; CompletionTime = null; @@ -52,10 +52,10 @@ public Task Start() ArchiveType = ArchiveType, Progress = GetProgress(), StartTime = Started - }, CancellationToken.None); + }, cancellationToken); } - public Task BatchUnarchived(int numberOfMessagesUnarchivedInBatch) + public Task BatchUnarchived(int numberOfMessagesUnarchivedInBatch, CancellationToken cancellationToken = default) { ArchiveState = ArchiveState.ArchiveProgressing; NumberOfMessagesUnarchived += numberOfMessagesUnarchivedInBatch; @@ -69,10 +69,10 @@ public Task BatchUnarchived(int numberOfMessagesUnarchivedInBatch) Progress = GetProgress(), StartTime = Started, Last = Last.Value - }, CancellationToken.None); + }, cancellationToken); } - public Task FinalizeUnarchive() + public Task FinalizeUnarchive(CancellationToken cancellationToken = default) { ArchiveState = ArchiveState.ArchiveFinalizing; NumberOfMessagesUnarchived = TotalNumberOfMessages; @@ -85,10 +85,10 @@ public Task FinalizeUnarchive() Progress = GetProgress(), StartTime = Started, Last = Last.Value - }, CancellationToken.None); + }, cancellationToken); } - public Task Complete() + public Task Complete(CancellationToken cancellationToken = default) { ArchiveState = ArchiveState.ArchiveCompleted; NumberOfMessagesUnarchived = TotalNumberOfMessages; @@ -104,7 +104,7 @@ public Task Complete() Last = Last.Value, CompletionTime = CompletionTime.Value, GroupName = GroupName - }, CancellationToken.None); + }, cancellationToken); } internal bool NeedsAcknowledgement() diff --git a/src/ServiceControl.UnitTests/Recoverability/NoopArchiveMessages.cs b/src/ServiceControl.UnitTests/Recoverability/NoopArchiveMessages.cs index b5a8a70bfe..571d27b4df 100644 --- a/src/ServiceControl.UnitTests/Recoverability/NoopArchiveMessages.cs +++ b/src/ServiceControl.UnitTests/Recoverability/NoopArchiveMessages.cs @@ -2,6 +2,7 @@ namespace ServiceControl.UnitTests.Recoverability; using System.Collections.Generic; +using System.Threading; using System.Threading.Tasks; using ServiceControl.Infrastructure.Auth; using ServiceControl.Persistence.Recoverability; @@ -11,9 +12,9 @@ sealed class NoopArchiveMessages : IArchiveMessages { public bool OperationInProgress { get; init; } - public Task ArchiveAllInGroup(string groupId, AuditUser? initiatedBy = null, string? operationId = null) => Task.CompletedTask; + public Task ArchiveAllInGroup(string groupId, AuditUser? initiatedBy = null, string? operationId = null, CancellationToken cancellationToken = default) => Task.CompletedTask; - public Task UnarchiveAllInGroup(string groupId, AuditUser? initiatedBy = null, string? operationId = null) => Task.CompletedTask; + public Task UnarchiveAllInGroup(string groupId, AuditUser? initiatedBy = null, string? operationId = null, CancellationToken cancellationToken = default) => Task.CompletedTask; public bool IsOperationInProgressFor(string groupId, ArchiveType archiveType) => OperationInProgress; @@ -23,9 +24,9 @@ public void DismissArchiveOperation(string groupId, ArchiveType archiveType) { } - public Task StartArchiving(string groupId, ArchiveType archiveType) => Task.CompletedTask; + public Task StartArchiving(string groupId, ArchiveType archiveType, CancellationToken cancellationToken = default) => Task.CompletedTask; - public Task StartUnarchiving(string groupId, ArchiveType archiveType) => Task.CompletedTask; + public Task StartUnarchiving(string groupId, ArchiveType archiveType, CancellationToken cancellationToken = default) => Task.CompletedTask; public IEnumerable GetArchivalOperations() => []; } diff --git a/src/ServiceControl/Recoverability/API/FailureGroupsArchiveController.cs b/src/ServiceControl/Recoverability/API/FailureGroupsArchiveController.cs index f0b5b957c2..938ca7143d 100644 --- a/src/ServiceControl/Recoverability/API/FailureGroupsArchiveController.cs +++ b/src/ServiceControl/Recoverability/API/FailureGroupsArchiveController.cs @@ -31,7 +31,7 @@ await auditLog.AuditedOperation(user, MessageActionKind.Archive, Permissions.ErrorRecoverabilityGroupsArchive, MessageActionScope.Group, resource: groupId, count: null, operationId: operationId, async ct => { - await archiver.StartArchiving(groupId, ArchiveType.FailureGroup); + await archiver.StartArchiving(groupId, ArchiveType.FailureGroup, ct); await bus.Send(m => { m.GroupId = groupId; }, AuditHeaders.LocalSendOptions(user, operationId), ct); }, cancellationToken); } diff --git a/src/ServiceControl/Recoverability/API/FailureGroupsUnarchiveController.cs b/src/ServiceControl/Recoverability/API/FailureGroupsUnarchiveController.cs index abf059fc6c..9def03512c 100644 --- a/src/ServiceControl/Recoverability/API/FailureGroupsUnarchiveController.cs +++ b/src/ServiceControl/Recoverability/API/FailureGroupsUnarchiveController.cs @@ -31,7 +31,7 @@ await auditLog.AuditedOperation(user, MessageActionKind.Unarchive, Permissions.ErrorRecoverabilityGroupsUnarchive, MessageActionScope.Group, resource: groupId, count: null, operationId: operationId, async ct => { - await archiver.StartUnarchiving(groupId, ArchiveType.FailureGroup); + await archiver.StartUnarchiving(groupId, ArchiveType.FailureGroup, ct); await bus.Send(m => { m.GroupId = groupId; }, AuditHeaders.LocalSendOptions(user, operationId), ct); }, cancellationToken); } diff --git a/src/ServiceControl/Recoverability/Archiving/ArchiveAllInGroupHandler.cs b/src/ServiceControl/Recoverability/Archiving/ArchiveAllInGroupHandler.cs index 69e0dd1ae2..0143ac539c 100644 --- a/src/ServiceControl/Recoverability/Archiving/ArchiveAllInGroupHandler.cs +++ b/src/ServiceControl/Recoverability/Archiving/ArchiveAllInGroupHandler.cs @@ -19,7 +19,7 @@ public async Task Handle(ArchiveAllInGroup message, IMessageHandlerContext conte var (user, operationId) = AuditHeaders.Read(context.MessageHeaders); - await archiver.ArchiveAllInGroup(message.GroupId, user, operationId); + await archiver.ArchiveAllInGroup(message.GroupId, user, operationId, context.CancellationToken); } } } diff --git a/src/ServiceControl/Recoverability/Archiving/UnArchiveAllInGroupHandler.cs b/src/ServiceControl/Recoverability/Archiving/UnArchiveAllInGroupHandler.cs index a1c6af97a2..a1a5e26b2f 100644 --- a/src/ServiceControl/Recoverability/Archiving/UnArchiveAllInGroupHandler.cs +++ b/src/ServiceControl/Recoverability/Archiving/UnArchiveAllInGroupHandler.cs @@ -19,7 +19,7 @@ public async Task Handle(UnarchiveAllInGroup message, IMessageHandlerContext con var (user, operationId) = AuditHeaders.Read(context.MessageHeaders); - await archiver.UnarchiveAllInGroup(message.GroupId, user, operationId); + await archiver.UnarchiveAllInGroup(message.GroupId, user, operationId, context.CancellationToken); } } } \ No newline at end of file