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