From 2b0a433ed441208e9284323470039d7051a09cf2 Mon Sep 17 00:00:00 2001 From: John Simons Date: Tue, 11 Aug 2026 18:58:43 +1000 Subject: [PATCH] Propagate cancellation tokens through group data store persistence This change continues the effort to consistently apply optional `CancellationToken` parameters to asynchronous methods within the `IGroupsDataStore` interface, its EFCore and RavenDB implementations, and related consuming services. Cancellation tokens are now propagated to underlying asynchronous database calls, ensuring proper cancellation of long-running operations and further reducing cancellation analyzer debt. --- .../Implementation/GroupsDataStore.cs | 53 +++++++++--------- .../Recoverability/GroupsDataStore.cs | 55 ++++++++++--------- .../IGroupsDataStore.cs | 17 +++--- .../Api/ArchiveMessagesController.cs | 4 +- .../API/FailureGroupsController.cs | 10 ++-- .../Recoverability/API/GroupFetcher.cs | 2 +- .../Handlers/RetryAllInGroupHandler.cs | 2 +- 7 files changed, 73 insertions(+), 70 deletions(-) diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/GroupsDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/GroupsDataStore.cs index 488d5a527e..e31d211832 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/GroupsDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/GroupsDataStore.cs @@ -12,7 +12,7 @@ namespace ServiceControl.Persistence.EFCore.Implementation; public class GroupsDataStore(IServiceScopeFactory scopeFactory) : DataStoreBase(scopeFactory), IGroupsDataStore { - public Task> GetUnresolvedGroupsByClassifier(string classifier, string classifierFilter) => + public Task> GetUnresolvedGroupsByClassifier(string classifier, string classifierFilter, CancellationToken cancellationToken = default) => ExecuteWithDbContext(async dbContext => { var groups = ByClassifier(dbContext, classifier); @@ -22,57 +22,58 @@ public Task> GetUnresolvedGroupsByClassifier(string clas groups = groups.Where(group => group.Title == classifierFilter); } - var views = await MostRecent(groups.AggregateGroups(WithStatus(dbContext, FailedMessageStatus.Unresolved))); + var views = await MostRecent(groups.AggregateGroups(WithStatus(dbContext, FailedMessageStatus.Unresolved)), cancellationToken); - await AttachComments(dbContext, views); + await AttachComments(dbContext, views, cancellationToken); return views; }); - public Task> GetArchivedGroupsByClassifier(string classifier) => + public Task> GetArchivedGroupsByClassifier(string classifier, CancellationToken cancellationToken = default) => ExecuteWithDbContext(dbContext => MostRecent( - ByClassifier(dbContext, classifier).AggregateGroups(WithStatus(dbContext, FailedMessageStatus.Archived)))); + ByClassifier(dbContext, classifier).AggregateGroups(WithStatus(dbContext, FailedMessageStatus.Archived)), cancellationToken)); - public Task> GetUnresolvedGroup(string groupId, string status, string modified) => - ExecuteWithDbContext(dbContext => SingleGroup(dbContext, groupId, FailedMessageStatus.Unresolved, status, modified)); + public Task> GetUnresolvedGroup(string groupId, string status, string modified, CancellationToken cancellationToken = default) => + ExecuteWithDbContext(dbContext => SingleGroup(dbContext, groupId, FailedMessageStatus.Unresolved, status, modified, cancellationToken)); - public Task> GetArchivedGroup(string groupId, string status, string modified) => - ExecuteWithDbContext(dbContext => SingleGroup(dbContext, groupId, FailedMessageStatus.Archived, status, modified)); + public Task> GetArchivedGroup(string groupId, string status, string modified, CancellationToken cancellationToken = default) => + ExecuteWithDbContext(dbContext => SingleGroup(dbContext, groupId, FailedMessageStatus.Archived, status, modified, cancellationToken)); - public Task>> GetGroupErrors(string groupId, string status, string modified, SortInfo sortInfo, PagingInfo pagingInfo) => - ExecuteWithDbContext(dbContext => InGroup(dbContext, groupId, status, modified).ToPagedResult(pagingInfo, sortInfo)); + public Task>> GetGroupErrors(string groupId, string status, string modified, SortInfo sortInfo, PagingInfo pagingInfo, CancellationToken cancellationToken = default) => + ExecuteWithDbContext(dbContext => InGroup(dbContext, groupId, status, modified).ToPagedResult(pagingInfo, sortInfo, cancellationToken)); - public Task GetGroupErrorsCount(string groupId, string status, string modified) => - ExecuteWithDbContext(dbContext => InGroup(dbContext, groupId, status, modified).ToQueryStatsInfo()); + public Task GetGroupErrorsCount(string groupId, string status, string modified, CancellationToken cancellationToken = default) => + ExecuteWithDbContext(dbContext => InGroup(dbContext, groupId, status, modified).ToQueryStatsInfo(cancellationToken)); - public Task EditComment(string groupId, string comment) => + public Task EditComment(string groupId, string comment, CancellationToken cancellationToken = default) => ExecuteWithDbContext(async dbContext => { if (string.IsNullOrWhiteSpace(comment)) { - await RemoveComment(dbContext, groupId); + await RemoveComment(dbContext, groupId, cancellationToken); return; } await dbContext.UpsertAsync([groupId], () => new GroupCommentEntity { GroupId = groupId, Comment = comment }, - entity => entity.Comment = comment); + entity => entity.Comment = comment, + cancellationToken); }); - public Task DeleteComment(string groupId) => - ExecuteWithDbContext(dbContext => RemoveComment(dbContext, groupId)); + public Task DeleteComment(string groupId, CancellationToken cancellationToken = default) => + ExecuteWithDbContext(dbContext => RemoveComment(dbContext, groupId, cancellationToken)); - static Task RemoveComment(ServiceControlDbContext dbContext, string groupId) => + static Task RemoveComment(ServiceControlDbContext dbContext, string groupId, CancellationToken cancellationToken) => dbContext.GroupComments .Where(groupComment => groupComment.GroupId == groupId) - .ExecuteDeleteAsync(); + .ExecuteDeleteAsync(cancellationToken); static IQueryable ByClassifier(ServiceControlDbContext dbContext, string classifier) => dbContext.FailedMessageGroups .AsNoTracking() .Where(group => group.Type == classifier); - static async Task> SingleGroup(ServiceControlDbContext dbContext, string groupId, FailedMessageStatus baseline, string status, string modified) + static async Task> SingleGroup(ServiceControlDbContext dbContext, string groupId, FailedMessageStatus baseline, string status, string modified, CancellationToken cancellationToken) { var groups = await dbContext.FailedMessageGroups .AsNoTracking() @@ -80,7 +81,7 @@ static async Task> SingleGroup(ServiceControlDbCon .AggregateGroups(WithStatus(dbContext, baseline) .FilterByStatus(status) .FilterByLastModifiedRange(modified)) - .ToListAsync(); + .ToListAsync(cancellationToken); return new QueryResult(groups.FirstOrDefault()!, groups.ToQueryStatsInfo()); } @@ -97,7 +98,7 @@ static IQueryable InGroup(ServiceControlDbContext dbContext .FilterByStatus(status) .FilterByLastModifiedRange(modified); - static async Task AttachComments(ServiceControlDbContext dbContext, IList groups) + static async Task AttachComments(ServiceControlDbContext dbContext, IList groups, CancellationToken cancellationToken) { if (groups.Count == 0) { @@ -109,7 +110,7 @@ static async Task AttachComments(ServiceControlDbContext dbContext, IList groupIds.Contains(groupComment.GroupId)) - .ToDictionaryAsync(groupComment => groupComment.GroupId, groupComment => groupComment.Comment); + .ToDictionaryAsync(groupComment => groupComment.GroupId, groupComment => groupComment.Comment, cancellationToken); foreach (var group in groups) { @@ -117,9 +118,9 @@ static async Task AttachComments(ServiceControlDbContext dbContext, IList> MostRecent(IQueryable groups) => + static async Task> MostRecent(IQueryable groups, CancellationToken cancellationToken) => await groups .OrderByDescending(group => group.Last) .Take(FailureGroupQueries.MaxGroups) - .ToListAsync(); + .ToListAsync(cancellationToken); } diff --git a/src/ServiceControl.Persistence.RavenDB/Recoverability/GroupsDataStore.cs b/src/ServiceControl.Persistence.RavenDB/Recoverability/GroupsDataStore.cs index d12249702e..a1bbfcea58 100644 --- a/src/ServiceControl.Persistence.RavenDB/Recoverability/GroupsDataStore.cs +++ b/src/ServiceControl.Persistence.RavenDB/Recoverability/GroupsDataStore.cs @@ -14,9 +14,9 @@ namespace ServiceControl.Persistence.RavenDB.Recoverability class GroupsDataStore(IRavenSessionProvider sessionProvider) : IGroupsDataStore { - public async Task> GetUnresolvedGroupsByClassifier(string classifier, string classifierFilter) + public async Task> GetUnresolvedGroupsByClassifier(string classifier, string classifierFilter, CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var query = Queryable.Where(session.Query(), v => v.Type == classifier); if (!string.IsNullOrWhiteSpace(classifierFilter)) @@ -26,11 +26,11 @@ public async Task> GetUnresolvedGroupsByClassifier(strin var groups = await query.OrderByDescending(x => x.Last) .Take(200) - .ToListAsync(); + .ToListAsync(cancellationToken); var commentIds = groups.Select(x => MakeId(x.Id)).ToArray(); var comments = await session.Query().Where(x => x.Id.In(commentIds)) - .ToListAsync(CancellationToken.None); + .ToListAsync(cancellationToken); foreach (var group in groups) { @@ -40,9 +40,9 @@ public async Task> GetUnresolvedGroupsByClassifier(strin return groups; } - public async Task> GetArchivedGroupsByClassifier(string classifier) + public async Task> GetArchivedGroupsByClassifier(string classifier, CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var groups = session .Query() .Where(v => v.Type == classifier); @@ -50,35 +50,35 @@ public async Task> GetArchivedGroupsByClassifier(string var results = await groups .OrderByDescending(x => x.Last) .Take(200) // only show 200 groups - .ToListAsync(); + .ToListAsync(cancellationToken); return results; } - public async Task> GetUnresolvedGroup(string groupId, string status, string modified) + public async Task> GetUnresolvedGroup(string groupId, string status, string modified, CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var document = await session.Advanced .AsyncDocumentQuery() .Statistics(out var stats) .WhereEquals(group => group.Id, groupId) .FilterByStatusWhere(status) .FilterByLastModifiedRange(modified) - .FirstOrDefaultAsync(); + .FirstOrDefaultAsync(cancellationToken); return new QueryResult(document, stats.ToQueryStatsInfo()); } - public async Task> GetArchivedGroup(string groupId, string status, string modified) + public async Task> GetArchivedGroup(string groupId, string status, string modified, CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var document = await session.Advanced .AsyncDocumentQuery() .Statistics(out var stats) .WhereEquals(group => group.Id, groupId) .FilterByStatusWhere(status) .FilterByLastModifiedRange(modified) - .FirstOrDefaultAsync(); + .FirstOrDefaultAsync(cancellationToken); return new QueryResult(document, stats.ToQueryStatsInfo()); } @@ -88,10 +88,11 @@ public async Task>> GetGroupErrors( string status, string modified, SortInfo sortInfo, - PagingInfo pagingInfo + PagingInfo pagingInfo, + CancellationToken cancellationToken = default ) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var query = session.Advanced .AsyncDocumentQuery() .Statistics(out var stats) @@ -105,42 +106,42 @@ PagingInfo pagingInfo .TransformToFailedMessageView(); var results = await query - .ToListAsync(); + .ToListAsync(cancellationToken); return results.ToQueryResult(stats); } - public async Task GetGroupErrorsCount(string groupId, string status, string modified) + public async Task GetGroupErrorsCount(string groupId, string status, string modified, CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var queryResult = await session.Advanced .AsyncDocumentQuery() .WhereEquals(view => view.FailureGroupId, groupId) .FilterByStatusWhere(status) .FilterByLastModifiedRange(modified) - .GetQueryResultAsync(); + .GetQueryResultAsync(cancellationToken); return queryResult.ToQueryStatsInfo(); } - public async Task EditComment(string groupId, string comment) + public async Task EditComment(string groupId, string comment, CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var groupComment = - await session.LoadAsync(MakeId(groupId)) + await session.LoadAsync(MakeId(groupId), cancellationToken) ?? new GroupComment { Id = MakeId(groupId) }; groupComment.Comment = comment; - await session.StoreAsync(groupComment); - await session.SaveChangesAsync(); + await session.StoreAsync(groupComment, cancellationToken); + await session.SaveChangesAsync(cancellationToken); } - public async Task DeleteComment(string groupId) + public async Task DeleteComment(string groupId, CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); session.Delete(MakeId(groupId)); - await session.SaveChangesAsync(); + await session.SaveChangesAsync(cancellationToken); } public static string MakeId(string groupId) => $"GroupComment/{groupId}"; diff --git a/src/ServiceControl.Persistence/IGroupsDataStore.cs b/src/ServiceControl.Persistence/IGroupsDataStore.cs index 422082f25c..0c7493680e 100644 --- a/src/ServiceControl.Persistence/IGroupsDataStore.cs +++ b/src/ServiceControl.Persistence/IGroupsDataStore.cs @@ -1,6 +1,7 @@ namespace ServiceControl.Persistence { using System.Collections.Generic; + using System.Threading; using System.Threading.Tasks; using Infrastructure; using MessageFailures.Api; @@ -8,15 +9,15 @@ namespace ServiceControl.Persistence public interface IGroupsDataStore { - Task> GetUnresolvedGroupsByClassifier(string classifier, string classifierFilter); - Task> GetArchivedGroupsByClassifier(string classifier); + Task> GetUnresolvedGroupsByClassifier(string classifier, string classifierFilter, CancellationToken cancellationToken = default); + Task> GetArchivedGroupsByClassifier(string classifier, CancellationToken cancellationToken = default); - Task> GetUnresolvedGroup(string groupId, string status, string modified); - Task> GetArchivedGroup(string groupId, string status, string modified); - Task>> GetGroupErrors(string groupId, string status, string modified, SortInfo sortInfo, PagingInfo pagingInfo); - Task GetGroupErrorsCount(string groupId, string status, string modified); + Task> GetUnresolvedGroup(string groupId, string status, string modified, CancellationToken cancellationToken = default); + Task> GetArchivedGroup(string groupId, string status, string modified, CancellationToken cancellationToken = default); + Task>> GetGroupErrors(string groupId, string status, string modified, SortInfo sortInfo, PagingInfo pagingInfo, CancellationToken cancellationToken = default); + Task GetGroupErrorsCount(string groupId, string status, string modified, CancellationToken cancellationToken = default); - Task EditComment(string groupId, string comment); - Task DeleteComment(string groupId); + Task EditComment(string groupId, string comment, CancellationToken cancellationToken = default); + Task DeleteComment(string groupId, CancellationToken cancellationToken = default); } } diff --git a/src/ServiceControl/MessageFailures/Api/ArchiveMessagesController.cs b/src/ServiceControl/MessageFailures/Api/ArchiveMessagesController.cs index 0e61eca99f..7c5ab51504 100644 --- a/src/ServiceControl/MessageFailures/Api/ArchiveMessagesController.cs +++ b/src/ServiceControl/MessageFailures/Api/ArchiveMessagesController.cs @@ -48,7 +48,7 @@ await auditLog.AuditedOperation(user, MessageActionKind.Archive, Permissions.Err [HttpGet] public async Task GetArchiveMessageGroups(string classifier = "Exception Type and Stack Trace", CancellationToken cancellationToken = default) { - var results = await dataStore.GetArchivedGroupsByClassifier(classifier); + var results = await dataStore.GetArchivedGroupsByClassifier(classifier, cancellationToken); Response.WithDeterministicEtag(EtagHelper.CalculateEtag(results)); @@ -75,7 +75,7 @@ await auditLog.AuditedOperation(user, MessageActionKind.Archive, Permissions.Err [HttpGet] public async Task> GetGroup(string groupId, string status = default, string modified = default, CancellationToken cancellationToken = default) { - var result = await dataStore.GetArchivedGroup(groupId, status, modified); + var result = await dataStore.GetArchivedGroup(groupId, status, modified, cancellationToken); Response.WithEtag(result.QueryStats.ETag); diff --git a/src/ServiceControl/Recoverability/API/FailureGroupsController.cs b/src/ServiceControl/Recoverability/API/FailureGroupsController.cs index 18fb0c29cb..15d7db5b03 100644 --- a/src/ServiceControl/Recoverability/API/FailureGroupsController.cs +++ b/src/ServiceControl/Recoverability/API/FailureGroupsController.cs @@ -41,7 +41,7 @@ public string[] GetSupportedClassifiers() [HttpPost] public async Task EditComment(string groupId, string comment, CancellationToken cancellationToken = default) { - await store.EditComment(groupId, comment); + await store.EditComment(groupId, comment, cancellationToken); return Accepted(); } @@ -51,7 +51,7 @@ public async Task EditComment(string groupId, string comment, Can [HttpDelete] public async Task DeleteComment(string groupId, CancellationToken cancellationToken = default) { - await store.DeleteComment(groupId); + await store.DeleteComment(groupId, cancellationToken); return Accepted(); } @@ -76,7 +76,7 @@ public async Task GetAllGroups(string classifier = "Exception [HttpGet] public async Task> GetGroupErrors(string groupId, [FromQuery] SortInfo sortInfo, [FromQuery] PagingInfo pagingInfo, string status = default, string modified = default, CancellationToken cancellationToken = default) { - var results = await store.GetGroupErrors(groupId, status, modified, sortInfo, pagingInfo); + var results = await store.GetGroupErrors(groupId, status, modified, sortInfo, pagingInfo, cancellationToken); Response.WithQueryStatsAndPagingInfo(results.QueryStats, pagingInfo); return results.Results; @@ -88,7 +88,7 @@ public async Task> GetGroupErrors(string groupId, [From [HttpHead] public async Task GetGroupErrorsCount(string groupId, string status = default, string modified = default, CancellationToken cancellationToken = default) { - var results = await store.GetGroupErrorsCount(groupId, status, modified); + var results = await store.GetGroupErrorsCount(groupId, status, modified, cancellationToken); Response.WithQueryStatsInfo(results); } @@ -110,7 +110,7 @@ public async Task GetRetryHistory(CancellationToken cancellationTo [HttpGet] public async Task> GetGroup(string groupId, string status = default, string modified = default, CancellationToken cancellationToken = default) { - var result = await store.GetUnresolvedGroup(groupId, status, modified); + var result = await store.GetUnresolvedGroup(groupId, status, modified, cancellationToken); Response.WithEtag(result.QueryStats.ETag); diff --git a/src/ServiceControl/Recoverability/API/GroupFetcher.cs b/src/ServiceControl/Recoverability/API/GroupFetcher.cs index e417e49725..e72ea92458 100644 --- a/src/ServiceControl/Recoverability/API/GroupFetcher.cs +++ b/src/ServiceControl/Recoverability/API/GroupFetcher.cs @@ -20,7 +20,7 @@ public GroupFetcher(IGroupsDataStore store, IRetryHistoryDataStore retryStore, I public async Task GetGroups(string classifier, string classifierFilter, CancellationToken cancellationToken = default) { - var dbGroups = await store.GetUnresolvedGroupsByClassifier(classifier, classifierFilter); + var dbGroups = await store.GetUnresolvedGroupsByClassifier(classifier, classifierFilter, cancellationToken); var retryHistory = await retryStore.GetRetryHistory(cancellationToken); var unacknowledgedRetries = retryHistory.GetUnacknowledgedByClassifier(classifier); diff --git a/src/ServiceControl/Recoverability/Retrying/Handlers/RetryAllInGroupHandler.cs b/src/ServiceControl/Recoverability/Retrying/Handlers/RetryAllInGroupHandler.cs index ed7b3291b3..b32bfc3f7d 100644 --- a/src/ServiceControl/Recoverability/Retrying/Handlers/RetryAllInGroupHandler.cs +++ b/src/ServiceControl/Recoverability/Retrying/Handlers/RetryAllInGroupHandler.cs @@ -27,7 +27,7 @@ public async Task Handle(RetryAllInGroup message, IMessageHandlerContext context } - var group = (await dataStore.GetUnresolvedGroup(message.GroupId, null, null)).Results; + var group = (await dataStore.GetUnresolvedGroup(message.GroupId, null, null, context.CancellationToken)).Results; string originator = null; if (group?.Title != null)