Skip to content

Propagate cancellation tokens through the archiving operations #5741

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
johnsimons merged 1 commit into from
Aug 12, 2026
Merged

Propagate cancellation tokens through the archiving operations #5741

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 10 additions & 10 deletions ...ServiceControl.Persistence.EFCore/Implementation/Recoverability/EFCoreArchivingManager.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);

Expand All @@ -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);

Expand All @@ -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)
Expand All @@ -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)
Expand Down
20 changes: 10 additions & 10 deletions ...rviceControl.Persistence.EFCore/Implementation/Recoverability/EFCoreUnarchivingManager.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);

Expand All @@ -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);

Expand All @@ -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);
}
}
68 changes: 34 additions & 34 deletions src/ServiceControl.Persistence.EFCore/Implementation/Recoverability/MessageArchiver.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ ILogger<MessageArchiver> 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);

Expand All @@ -45,7 +45,7 @@ public async Task ArchiveAllInGroup(string groupId, AuditUser? initiatedBy = nul
{
var dbContext = scope.ServiceProvider.GetRequiredService<ServiceControlDbContext>();

operationEntity = await GetOrCreateOperation(dbContext, groupId, ArchiveOperationType.Archive, initiatedBy, operationId);
operationEntity = await GetOrCreateOperation(dbContext, groupId, ArchiveOperationType.Archive, initiatedBy, operationId, cancellationToken);
if (operationEntity == null)
{
return;
Expand All @@ -57,27 +57,27 @@ 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
{
await using var batchScope = scopeFactory.CreateAsyncScope();
var batchDbContext = batchScope.ServiceProvider.GetRequiredService<ServiceControlDbContext>();

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);
Expand All @@ -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())
Expand All @@ -99,20 +99,20 @@ await finalizeDbContext.ArchiveOperations
o.ArchiveType == operationEntity.ArchiveType
&& o.RequestId == operationEntity.RequestId
&& o.OperationType == operationEntity.OperationType)
.ExecuteDeleteAsync();
.ExecuteDeleteAsync(cancellationToken);
}

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);

Expand All @@ -125,7 +125,7 @@ public async Task UnarchiveAllInGroup(string groupId, AuditUser? initiatedBy = n
{
var dbContext = scope.ServiceProvider.GetRequiredService<ServiceControlDbContext>();

operationEntity = await GetOrCreateOperation(dbContext, groupId, ArchiveOperationType.UnArchive, initiatedBy, operationId);
operationEntity = await GetOrCreateOperation(dbContext, groupId, ArchiveOperationType.UnArchive, initiatedBy, operationId, cancellationToken);
if (operationEntity == null)
{
return;
Expand All @@ -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<ServiceControlDbContext>();
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);
Expand All @@ -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())
Expand All @@ -177,15 +177,15 @@ await finalizeDbContext.ArchiveOperations
o.ArchiveType == operationEntity.ArchiveType
&& o.RequestId == operationEntity.RequestId
&& o.OperationType == operationEntity.OperationType)
.ExecuteDeleteAsync();
.ExecuteDeleteAsync(cancellationToken);
}

await domainEvents.Raise(new FailedMessageGroupUnarchived
{
GroupId = groupId,
GroupName = operationEntity.GroupName,
MessagesCount = operationEntity.TotalNumberOfMessages
});
}, cancellationToken);

logger.LogInformation("Unarchiving of group {GroupId} completed", groupId);
}
Expand All @@ -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<InMemoryArchive> GetArchivalOperations()
=> archivingManager.GetArchivalOperations();

async Task<ArchiveOperationEntity?> GetOrCreateOperation(ServiceControlDbContext dbContext, string groupId, ArchiveOperationType operation, AuditUser? initiatedBy, string? operationId)
async Task<ArchiveOperationEntity?> 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)
{
Expand All @@ -219,7 +219,7 @@ public IEnumerable<InMemoryArchive> 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);
Expand All @@ -246,7 +246,7 @@ public IEnumerable<InMemoryArchive> 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))
Expand Down Expand Up @@ -276,7 +276,7 @@ void AuditArchivedMessages(MessageActionKind kind, string permission, AuditUser
}
}

async Task<string[]> UpdateGroupStatusAsync(ServiceControlDbContext dbContext, string groupId, FailedMessageStatus fromStatus, FailedMessageStatus toStatus, int batchSize, CancellationToken cancellationToken = default)
async Task<string[]> 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)
Expand All @@ -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
Expand Down
Loading
Loading