Skip to content

Propagate cancellation tokens through group data store persistence #5737

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 group data store persistence #5737

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
53 changes: 27 additions & 26 deletions src/ServiceControl.Persistence.EFCore/Implementation/GroupsDataStore.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@ namespace ServiceControl.Persistence.EFCore.Implementation;

public class GroupsDataStore(IServiceScopeFactory scopeFactory) : DataStoreBase(scopeFactory), IGroupsDataStore
{
public Task<IList<FailureGroupView>> GetUnresolvedGroupsByClassifier(string classifier, string classifierFilter) =>
public Task<IList<FailureGroupView>> GetUnresolvedGroupsByClassifier(string classifier, string classifierFilter, CancellationToken cancellationToken = default) =>
ExecuteWithDbContext(async dbContext =>
{
var groups = ByClassifier(dbContext, classifier);
Expand All @@ -22,65 +22,66 @@ public Task<IList<FailureGroupView>> 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<IList<FailureGroupView>> GetArchivedGroupsByClassifier(string classifier) =>
public Task<IList<FailureGroupView>> 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<QueryResult<FailureGroupView>> GetUnresolvedGroup(string groupId, string status, string modified) =>
ExecuteWithDbContext(dbContext => SingleGroup(dbContext, groupId, FailedMessageStatus.Unresolved, status, modified));
public Task<QueryResult<FailureGroupView>> GetUnresolvedGroup(string groupId, string status, string modified, CancellationToken cancellationToken = default) =>
ExecuteWithDbContext(dbContext => SingleGroup(dbContext, groupId, FailedMessageStatus.Unresolved, status, modified, cancellationToken));

public Task<QueryResult<FailureGroupView>> GetArchivedGroup(string groupId, string status, string modified) =>
ExecuteWithDbContext(dbContext => SingleGroup(dbContext, groupId, FailedMessageStatus.Archived, status, modified));
public Task<QueryResult<FailureGroupView>> GetArchivedGroup(string groupId, string status, string modified, CancellationToken cancellationToken = default) =>
ExecuteWithDbContext(dbContext => SingleGroup(dbContext, groupId, FailedMessageStatus.Archived, status, modified, cancellationToken));

public Task<QueryResult<IList<FailedMessageView>>> GetGroupErrors(string groupId, string status, string modified, SortInfo sortInfo, PagingInfo pagingInfo) =>
ExecuteWithDbContext(dbContext => InGroup(dbContext, groupId, status, modified).ToPagedResult(pagingInfo, sortInfo));
public Task<QueryResult<IList<FailedMessageView>>> 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<QueryStatsInfo> GetGroupErrorsCount(string groupId, string status, string modified) =>
ExecuteWithDbContext(dbContext => InGroup(dbContext, groupId, status, modified).ToQueryStatsInfo());
public Task<QueryStatsInfo> 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<int> RemoveComment(ServiceControlDbContext dbContext, string groupId) =>
static Task<int> RemoveComment(ServiceControlDbContext dbContext, string groupId, CancellationToken cancellationToken) =>
dbContext.GroupComments
.Where(groupComment => groupComment.GroupId == groupId)
.ExecuteDeleteAsync();
.ExecuteDeleteAsync(cancellationToken);

static IQueryable<FailedMessageGroupEntity> ByClassifier(ServiceControlDbContext dbContext, string classifier) =>
dbContext.FailedMessageGroups
.AsNoTracking()
.Where(group => group.Type == classifier);

static async Task<QueryResult<FailureGroupView>> SingleGroup(ServiceControlDbContext dbContext, string groupId, FailedMessageStatus baseline, string status, string modified)
static async Task<QueryResult<FailureGroupView>> SingleGroup(ServiceControlDbContext dbContext, string groupId, FailedMessageStatus baseline, string status, string modified, CancellationToken cancellationToken)
{
var groups = await dbContext.FailedMessageGroups
.AsNoTracking()
.Where(group => group.GroupId == groupId)
.AggregateGroups(WithStatus(dbContext, baseline)
.FilterByStatus(status)
.FilterByLastModifiedRange(modified))
.ToListAsync();
.ToListAsync(cancellationToken);

return new QueryResult<FailureGroupView>(groups.FirstOrDefault()!, groups.ToQueryStatsInfo());
}
Expand All @@ -97,7 +98,7 @@ static IQueryable<FailedMessageEntity> InGroup(ServiceControlDbContext dbContext
.FilterByStatus(status)
.FilterByLastModifiedRange(modified);

static async Task AttachComments(ServiceControlDbContext dbContext, IList<FailureGroupView> groups)
static async Task AttachComments(ServiceControlDbContext dbContext, IList<FailureGroupView> groups, CancellationToken cancellationToken)
{
if (groups.Count == 0)
{
Expand All @@ -109,17 +110,17 @@ static async Task AttachComments(ServiceControlDbContext dbContext, IList<Failur
var comments = await dbContext.GroupComments
.AsNoTracking()
.Where(groupComment => groupIds.Contains(groupComment.GroupId))
.ToDictionaryAsync(groupComment => groupComment.GroupId, groupComment => groupComment.Comment);
.ToDictionaryAsync(groupComment => groupComment.GroupId, groupComment => groupComment.Comment, cancellationToken);

foreach (var group in groups)
{
group.Comment = comments.GetValueOrDefault(group.Id);
}
}

static async Task<IList<FailureGroupView>> MostRecent(IQueryable<FailureGroupView> groups) =>
static async Task<IList<FailureGroupView>> MostRecent(IQueryable<FailureGroupView> groups, CancellationToken cancellationToken) =>
await groups
.OrderByDescending(group => group.Last)
.Take(FailureGroupQueries.MaxGroups)
.ToListAsync();
.ToListAsync(cancellationToken);
}
55 changes: 28 additions & 27 deletions src/ServiceControl.Persistence.RavenDB/Recoverability/GroupsDataStore.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,9 +14,9 @@ namespace ServiceControl.Persistence.RavenDB.Recoverability

class GroupsDataStore(IRavenSessionProvider sessionProvider) : IGroupsDataStore
{
public async Task<IList<FailureGroupView>> GetUnresolvedGroupsByClassifier(string classifier, string classifierFilter)
public async Task<IList<FailureGroupView>> 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<FailureGroupView, FailureGroupsViewIndex>(), v => v.Type == classifier);

if (!string.IsNullOrWhiteSpace(classifierFilter))
Expand All @@ -26,11 +26,11 @@ public async Task<IList<FailureGroupView>> 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<GroupComment, GroupCommentIndex>().Where(x => x.Id.In(commentIds))
.ToListAsync(CancellationToken.None);
.ToListAsync(cancellationToken);

foreach (var group in groups)
{
Expand All @@ -40,45 +40,45 @@ public async Task<IList<FailureGroupView>> GetUnresolvedGroupsByClassifier(strin
return groups;
}

public async Task<IList<FailureGroupView>> GetArchivedGroupsByClassifier(string classifier)
public async Task<IList<FailureGroupView>> GetArchivedGroupsByClassifier(string classifier, CancellationToken cancellationToken = default)
{
using var session = await sessionProvider.OpenSession();
using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken);
var groups = session
.Query<FailureGroupView, ArchivedGroupsViewIndex>()
.Where(v => v.Type == classifier);

var results = await groups
.OrderByDescending(x => x.Last)
.Take(200) // only show 200 groups
.ToListAsync();
.ToListAsync(cancellationToken);

return results;
}

public async Task<QueryResult<FailureGroupView>> GetUnresolvedGroup(string groupId, string status, string modified)
public async Task<QueryResult<FailureGroupView>> 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<FailureGroupView, FailureGroupsViewIndex>()
.Statistics(out var stats)
.WhereEquals(group => group.Id, groupId)
.FilterByStatusWhere(status)
.FilterByLastModifiedRange(modified)
.FirstOrDefaultAsync();
.FirstOrDefaultAsync(cancellationToken);

return new QueryResult<FailureGroupView>(document, stats.ToQueryStatsInfo());
}

public async Task<QueryResult<FailureGroupView>> GetArchivedGroup(string groupId, string status, string modified)
public async Task<QueryResult<FailureGroupView>> 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<FailureGroupView, ArchivedGroupsViewIndex>()
.Statistics(out var stats)
.WhereEquals(group => group.Id, groupId)
.FilterByStatusWhere(status)
.FilterByLastModifiedRange(modified)
.FirstOrDefaultAsync();
.FirstOrDefaultAsync(cancellationToken);

return new QueryResult<FailureGroupView>(document, stats.ToQueryStatsInfo());
}
Expand All @@ -88,10 +88,11 @@ public async Task<QueryResult<IList<FailedMessageView>>> 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<FailureGroupMessageView, FailedMessages_ByGroup>()
.Statistics(out var stats)
Expand All @@ -105,42 +106,42 @@ PagingInfo pagingInfo
.TransformToFailedMessageView();

var results = await query
.ToListAsync();
.ToListAsync(cancellationToken);

return results.ToQueryResult(stats);
}

public async Task<QueryStatsInfo> GetGroupErrorsCount(string groupId, string status, string modified)
public async Task<QueryStatsInfo> 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<FailureGroupMessageView, FailedMessages_ByGroup>()
.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<GroupComment>(MakeId(groupId))
await session.LoadAsync<GroupComment>(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}";
Expand Down
17 changes: 9 additions & 8 deletions src/ServiceControl.Persistence/IGroupsDataStore.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,22 +1,23 @@
namespace ServiceControl.Persistence
{
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Infrastructure;
using MessageFailures.Api;
using ServiceControl.Recoverability;

public interface IGroupsDataStore
{
Task<IList<FailureGroupView>> GetUnresolvedGroupsByClassifier(string classifier, string classifierFilter);
Task<IList<FailureGroupView>> GetArchivedGroupsByClassifier(string classifier);
Task<IList<FailureGroupView>> GetUnresolvedGroupsByClassifier(string classifier, string classifierFilter, CancellationToken cancellationToken = default);
Task<IList<FailureGroupView>> GetArchivedGroupsByClassifier(string classifier, CancellationToken cancellationToken = default);

Task<QueryResult<FailureGroupView>> GetUnresolvedGroup(string groupId, string status, string modified);
Task<QueryResult<FailureGroupView>> GetArchivedGroup(string groupId, string status, string modified);
Task<QueryResult<IList<FailedMessageView>>> GetGroupErrors(string groupId, string status, string modified, SortInfo sortInfo, PagingInfo pagingInfo);
Task<QueryStatsInfo> GetGroupErrorsCount(string groupId, string status, string modified);
Task<QueryResult<FailureGroupView>> GetUnresolvedGroup(string groupId, string status, string modified, CancellationToken cancellationToken = default);
Task<QueryResult<FailureGroupView>> GetArchivedGroup(string groupId, string status, string modified, CancellationToken cancellationToken = default);
Task<QueryResult<IList<FailedMessageView>>> GetGroupErrors(string groupId, string status, string modified, SortInfo sortInfo, PagingInfo pagingInfo, CancellationToken cancellationToken = default);
Task<QueryStatsInfo> 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);
}
}
4 changes: 2 additions & 2 deletions src/ServiceControl/MessageFailures/Api/ArchiveMessagesController.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,7 @@ await auditLog.AuditedOperation(user, MessageActionKind.Archive, Permissions.Err
[HttpGet]
public async Task<IActionResult> 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));

Expand All @@ -75,7 +75,7 @@ await auditLog.AuditedOperation(user, MessageActionKind.Archive, Permissions.Err
[HttpGet]
public async Task<ActionResult<FailureGroupView>> 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);

Expand Down
Loading
Loading