Skip to content

Continue cancellation token propagation in persistence layer #5731

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 11, 2026
Merged

Continue cancellation token propagation in persistence layer #5731

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
24 changes: 12 additions & 12 deletions src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageLifecycleDataStore.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@ namespace ServiceControl.Persistence.EFCore.Implementation;
/// </summary>
public class FailedMessageLifecycleDataStore(IServiceScopeFactory scopeFactory) : DataStoreBase(scopeFactory), IFailedMessageLifecycleDataStore
{
public async Task MarkAsArchived(string failedMessageId)
public async Task MarkAsArchived(string failedMessageId, CancellationToken cancellationToken = default)
{
await ExecuteWithDbContext(async dbContext =>
{
Expand All @@ -25,11 +25,11 @@ await dbContext.FailedMessages
.ExecuteUpdateAsync(s => s
.SetProperty(fm => fm.Status, FailedMessageStatus.Archived)
.SetProperty(fm => fm.StatusChangedAt, now)
.SetProperty(fm => fm.LastModified, now));
.SetProperty(fm => fm.LastModified, now), cancellationToken);
});
}

public async Task<bool> MarkAsResolved(string failedMessageId)
public async Task<bool> MarkAsResolved(string failedMessageId, CancellationToken cancellationToken = default)
{
return await ExecuteWithDbContext(async dbContext =>
{
Expand All @@ -45,13 +45,13 @@ public async Task<bool> MarkAsResolved(string failedMessageId)
.ExecuteUpdateAsync(s => s
.SetProperty(fm => fm.Status, FailedMessageStatus.Resolved)
.SetProperty(fm => fm.StatusChangedAt, now)
.SetProperty(fm => fm.LastModified, now));
.SetProperty(fm => fm.LastModified, now), cancellationToken);

return affected > 0;
});
}

public async Task<string[]> UnArchiveMessages(IEnumerable<string> failedMessageIds)
public async Task<string[]> UnArchiveMessages(IEnumerable<string> failedMessageIds, CancellationToken cancellationToken = default)
{
var ids = failedMessageIds
.Select(id => Guid.TryParse(id, out var guid) ? guid : Guid.Empty)
Expand All @@ -71,7 +71,7 @@ public async Task<string[]> UnArchiveMessages(IEnumerable<string> failedMessageI
var unarchivableIds = await dbContext.FailedMessages
.Where(fm => ids.Contains(fm.UniqueMessageId) && fm.Status == FailedMessageStatus.Archived)
.Select(fm => fm.UniqueMessageId)
.ToListAsync();
.ToListAsync(cancellationToken);

if (unarchivableIds.Count == 0)
{
Expand All @@ -83,13 +83,13 @@ await dbContext.FailedMessages
.ExecuteUpdateAsync(s => s
.SetProperty(fm => fm.Status, FailedMessageStatus.Unresolved)
.SetProperty(fm => fm.StatusChangedAt, now)
.SetProperty(fm => fm.LastModified, now));
.SetProperty(fm => fm.LastModified, now), cancellationToken);

return unarchivableIds.Select(id => id.ToString()).ToArray();
});
}

public async Task<string[]> UnArchiveMessagesByRange(DateTime from, DateTime to)
public async Task<string[]> UnArchiveMessagesByRange(DateTime from, DateTime to, CancellationToken cancellationToken = default)
{
return await ExecuteWithDbContext(async dbContext =>
{
Expand All @@ -101,7 +101,7 @@ public async Task<string[]> UnArchiveMessagesByRange(DateTime from, DateTime to)
&& fm.LastModified >= from
&& fm.LastModified <= to)
.Select(fm => fm.UniqueMessageId)
.ToListAsync();
.ToListAsync(cancellationToken);

if (unarchivableIds.Count == 0)
{
Expand All @@ -113,13 +113,13 @@ await dbContext.FailedMessages
.ExecuteUpdateAsync(s => s
.SetProperty(fm => fm.Status, FailedMessageStatus.Unresolved)
.SetProperty(fm => fm.StatusChangedAt, now)
.SetProperty(fm => fm.LastModified, now));
.SetProperty(fm => fm.LastModified, now), cancellationToken);

return unarchivableIds.Select(id => id.ToString()).ToArray();
});
}

public async Task RevertRetry(string messageUniqueId)
public async Task RevertRetry(string messageUniqueId, CancellationToken cancellationToken = default)
{
await ExecuteWithDbContext(async dbContext =>
{
Expand All @@ -135,7 +135,7 @@ await dbContext.FailedMessages
.ExecuteUpdateAsync(s => s
.SetProperty(fm => fm.Status, FailedMessageStatus.Unresolved)
.SetProperty(fm => fm.StatusChangedAt, now)
.SetProperty(fm => fm.LastModified, now));
.SetProperty(fm => fm.LastModified, now), cancellationToken);
});
}
}
44 changes: 22 additions & 22 deletions src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageQueryDataStore.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,36 +12,36 @@ namespace ServiceControl.Persistence.EFCore.Implementation;

public class FailedMessageQueryDataStore(IServiceScopeFactory scopeFactory) : DataStoreBase(scopeFactory), IFailedMessageQueryDataStore
{
public Task<QueryResult<IList<FailedMessageView>>> GetFailedMessages(string? status, string? modified, string? queueAddress, PagingInfo pagingInfo, SortInfo sortInfo) =>
public Task<QueryResult<IList<FailedMessageView>>> GetFailedMessages(string? status, string? modified, string? queueAddress, PagingInfo pagingInfo, SortInfo sortInfo, CancellationToken cancellationToken = default) =>
ExecuteWithDbContext(dbContext => dbContext.FailedMessages
.AsNoTracking()
.FilterByStatus(status)
.FilterByLastModifiedRange(modified)
.FilterByQueueAddress(queueAddress)
.ToPagedResult(pagingInfo, sortInfo));
.ToPagedResult(pagingInfo, sortInfo, cancellationToken));

public Task<QueryStatsInfo> GetFailedMessagesStats(string? status, string? modified, string? queueAddress) =>
public Task<QueryStatsInfo> GetFailedMessagesStats(string? status, string? modified, string? queueAddress, CancellationToken cancellationToken = default) =>
ExecuteWithDbContext(dbContext => dbContext.FailedMessages
.AsNoTracking()
.FilterByStatus(status)
.FilterByLastModifiedRange(modified)
.FilterByQueueAddress(queueAddress)
.ToQueryStatsInfo());
.ToQueryStatsInfo(cancellationToken));

public Task<QueryResult<IList<FailedMessageView>>> GetFailedMessagesByEndpoint(string? status, string endpointName, string? modified, PagingInfo pagingInfo, SortInfo sortInfo) =>
public Task<QueryResult<IList<FailedMessageView>>> GetFailedMessagesByEndpoint(string? status, string endpointName, string? modified, PagingInfo pagingInfo, SortInfo sortInfo, CancellationToken cancellationToken = default) =>
ExecuteWithDbContext(dbContext => dbContext.FailedMessages
.AsNoTracking()
.Where(message => message.ReceivingEndpointName == endpointName)
.FilterByStatus(status)
.FilterByLastModifiedRange(modified)
.ToPagedResult(pagingInfo, sortInfo));
.ToPagedResult(pagingInfo, sortInfo, cancellationToken));

public Task<IDictionary<string, object>> GetFailedMessagesSummary() =>
public Task<IDictionary<string, object>> GetFailedMessagesSummary(CancellationToken cancellationToken = default) =>
ExecuteWithDbContext(async dbContext =>
{
var endpoints = await CountBy(dbContext, message => message.ReceivingEndpointName);
var hosts = await CountBy(dbContext, message => message.ReceivingEndpointHost);
var messageTypes = await CountBy(dbContext, message => message.MessageType);
var endpoints = await CountBy(dbContext, message => message.ReceivingEndpointName, cancellationToken);
var hosts = await CountBy(dbContext, message => message.ReceivingEndpointHost, cancellationToken);
var messageTypes = await CountBy(dbContext, message => message.MessageType, cancellationToken);

return (IDictionary<string, object>)new Dictionary<string, object>
{
Expand All @@ -51,7 +51,7 @@ public Task<IDictionary<string, object>> GetFailedMessagesSummary() =>
};
});

public Task<FailedMessageView?> GetLatestFailedMessageView(string failedMessageId) =>
public Task<FailedMessageView?> GetLatestFailedMessageView(string failedMessageId, CancellationToken cancellationToken = default) =>
ExecuteWithDbContext(async dbContext =>
{
if (!Guid.TryParse(failedMessageId, out var uniqueMessageId))
Expand All @@ -61,12 +61,12 @@ public Task<IDictionary<string, object>> GetFailedMessagesSummary() =>

var entity = await dbContext.FailedMessages
.AsNoTracking()
.SingleOrDefaultAsync(message => message.UniqueMessageId == uniqueMessageId);
.SingleOrDefaultAsync(message => message.UniqueMessageId == uniqueMessageId, cancellationToken);

return entity?.ToFailedMessageView();
});

public Task<FailedMessage?> GetFailedMessage(string failedMessageId) =>
public Task<FailedMessage?> GetFailedMessage(string failedMessageId, CancellationToken cancellationToken = default) =>
ExecuteWithDbContext(async dbContext =>
{
if (!Guid.TryParse(failedMessageId, out var uniqueMessageId))
Expand All @@ -76,52 +76,52 @@ public Task<IDictionary<string, object>> GetFailedMessagesSummary() =>

var entity = await dbContext.FailedMessages
.AsNoTracking()
.SingleOrDefaultAsync(message => message.UniqueMessageId == uniqueMessageId);
.SingleOrDefaultAsync(message => message.UniqueMessageId == uniqueMessageId, cancellationToken);

if (entity == null)
{
return null;
}

var groups = await ReadGroups(dbContext, [uniqueMessageId]);
var groups = await ReadGroups(dbContext, [uniqueMessageId], cancellationToken);

return entity.ToFailedMessage(groups.GetValueOrDefault(uniqueMessageId, []));
});

public Task<FailedMessage[]> GetFailedMessagesByIds(Guid[] ids) =>
public Task<FailedMessage[]> GetFailedMessagesByIds(Guid[] ids, CancellationToken cancellationToken = default) =>
ExecuteWithDbContext<FailedMessage[]>(async dbContext =>
{
var entities = await dbContext.FailedMessages
.AsNoTracking()
.Where(message => ids.Contains(message.UniqueMessageId))
.ToListAsync();
.ToListAsync(cancellationToken);

if (entities.Count == 0)
{
return [];
}

var groups = await ReadGroups(dbContext, [.. entities.Select(entity => entity.UniqueMessageId)]);
var groups = await ReadGroups(dbContext, [.. entities.Select(entity => entity.UniqueMessageId)], cancellationToken);

return [.. entities.Select(entity => entity.ToFailedMessage(groups.GetValueOrDefault(entity.UniqueMessageId, [])))];
});

static async Task<Dictionary<string, object>> CountBy(ServiceControlDbContext dbContext, Expression<Func<FailedMessageEntity, string?>> selector) =>
static async Task<Dictionary<string, object>> CountBy(ServiceControlDbContext dbContext, Expression<Func<FailedMessageEntity, string?>> selector, CancellationToken cancellationToken) =>
await dbContext.FailedMessages
.AsNoTracking()
.Where(message => message.Status == FailedMessageStatus.Unresolved)
.Select(selector)
.Where(value => value != null && value != string.Empty)
.GroupBy(value => value)
.Select(group => new { Value = group.Key, Count = group.Count() })
.ToDictionaryAsync(row => row.Value!, row => (object)row.Count);
.ToDictionaryAsync(row => row.Value!, row => (object)row.Count, cancellationToken);

static async Task<Dictionary<Guid, List<FailedMessageGroupEntity>>> ReadGroups(ServiceControlDbContext dbContext, Guid[] uniqueMessageIds)
static async Task<Dictionary<Guid, List<FailedMessageGroupEntity>>> ReadGroups(ServiceControlDbContext dbContext, Guid[] uniqueMessageIds, CancellationToken cancellationToken)
{
var groups = await dbContext.FailedMessageGroups
.AsNoTracking()
.Where(group => uniqueMessageIds.Contains(group.FailedMessageUniqueId))
.ToListAsync();
.ToListAsync(cancellationToken);

return groups
.GroupBy(group => group.FailedMessageUniqueId)
Expand Down
6 changes: 3 additions & 3 deletions src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageQueryResults.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,14 +8,14 @@ namespace ServiceControl.Persistence.EFCore.Implementation;

static class FailedMessageQueryResults
{
public static async Task<QueryResult<IList<FailedMessageView>>> ToPagedResult(this IQueryable<FailedMessageEntity> source, PagingInfo pagingInfo, SortInfo sortInfo)
public static async Task<QueryResult<IList<FailedMessageView>>> ToPagedResult(this IQueryable<FailedMessageEntity> source, PagingInfo pagingInfo, SortInfo sortInfo, CancellationToken cancellationToken = default)
{
var stats = await source.ToQueryStatsInfo();
var stats = await source.ToQueryStatsInfo(cancellationToken);

var entities = await source
.Sort(sortInfo)
.Page(pagingInfo)
.ToListAsync();
.ToListAsync(cancellationToken);

IList<FailedMessageView> results = [.. entities.Select(entity => entity.ToFailedMessageView())];

Expand Down
8 changes: 4 additions & 4 deletions src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageRetryDataStore.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,15 +4,15 @@ namespace ServiceControl.Persistence.EFCore.Implementation;

public class FailedMessageRetryDataStore(IServiceScopeFactory scopeFactory) : DataStoreBase(scopeFactory), IFailedMessageRetryDataStore
{
public Task ProcessPendingRetries(DateTime periodFrom, DateTime periodTo, string queueAddress, Func<string, Task> processCallback) =>
public Task ProcessPendingRetries(DateTime periodFrom, DateTime periodTo, string queueAddress, Func<string, CancellationToken, Task> processCallback, CancellationToken cancellationToken = default) =>
throw new NotImplementedException();

public Task<string[]> GetRetryPendingMessages(DateTime from, DateTime to, string queueAddress) =>
public Task<string[]> GetRetryPendingMessages(DateTime from, DateTime to, string queueAddress, CancellationToken cancellationToken = default) =>
throw new NotImplementedException();

public Task RemoveFailedMessageRetry(string uniqueMessageId) =>
public Task RemoveFailedMessageRetry(string uniqueMessageId, CancellationToken cancellationToken = default) =>
throw new NotImplementedException();

public Task<byte[]> GetFailedMessageBody(string uniqueMessageId) =>
public Task<byte[]> GetFailedMessageBody(string uniqueMessageId, CancellationToken cancellationToken = default) =>
throw new NotImplementedException();
}
20 changes: 10 additions & 10 deletions src/ServiceControl.Persistence.EFCore/Implementation/MessagesViewDataStore.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,38 +9,38 @@ namespace ServiceControl.Persistence.EFCore.Implementation;

public class MessagesViewDataStore(IServiceScopeFactory scopeFactory, IFullTextSearchDialect fullTextSearch) : DataStoreBase(scopeFactory), IMessagesViewDataStore
{
public Task<QueryResult<IList<MessagesView>>> GetAllMessages(PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, DateTimeRange? timeSentRange = null) =>
public Task<QueryResult<IList<MessagesView>>> GetAllMessages(PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, DateTimeRange? timeSentRange = null, CancellationToken cancellationToken = default) =>
ExecuteWithDbContext(dbContext => dbContext.FailedMessages
.AsNoTracking()
.IncludeSystemMessagesWhere(includeSystemMessages)
.FilterBySentTimeRange(timeSentRange)
.ToPagedMessagesResult(pagingInfo, sortInfo));
.ToPagedMessagesResult(pagingInfo, sortInfo, cancellationToken));

public Task<QueryResult<IList<MessagesView>>> GetAllMessagesForEndpoint(string endpointName, PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, DateTimeRange? timeSentRange = null) =>
public Task<QueryResult<IList<MessagesView>>> GetAllMessagesForEndpoint(string endpointName, PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, DateTimeRange? timeSentRange = null, CancellationToken cancellationToken = default) =>
ExecuteWithDbContext(dbContext => dbContext.FailedMessages
.AsNoTracking()
.Where(message => message.ReceivingEndpointName == endpointName)
.IncludeSystemMessagesWhere(includeSystemMessages)
.FilterBySentTimeRange(timeSentRange)
.ToPagedMessagesResult(pagingInfo, sortInfo));
.ToPagedMessagesResult(pagingInfo, sortInfo, cancellationToken));

// includeSystemMessages is unused here: a conversation is incomplete without the system messages that took part in it.
public Task<QueryResult<IList<MessagesView>>> GetAllMessagesByConversation(string conversationId, PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages) =>
public Task<QueryResult<IList<MessagesView>>> GetAllMessagesByConversation(string conversationId, PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, CancellationToken cancellationToken = default) =>
ExecuteWithDbContext(dbContext => dbContext.FailedMessages
.AsNoTracking()
.Where(message => message.ConversationId == conversationId)
.ToPagedMessagesResult(pagingInfo, sortInfo));
.ToPagedMessagesResult(pagingInfo, sortInfo, cancellationToken));

public Task<QueryResult<IList<MessagesView>>> GetAllMessagesForSearch(string searchTerms, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange? timeSentRange = null) =>
public Task<QueryResult<IList<MessagesView>>> GetAllMessagesForSearch(string searchTerms, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange? timeSentRange = null, CancellationToken cancellationToken = default) =>
ExecuteWithDbContext(dbContext => Search(dbContext.FailedMessages.AsNoTracking(), searchTerms)
.FilterBySentTimeRange(timeSentRange)
.ToPagedMessagesResult(pagingInfo, sortInfo));
.ToPagedMessagesResult(pagingInfo, sortInfo, cancellationToken));

public Task<QueryResult<IList<MessagesView>>> SearchEndpointMessages(string endpointName, string searchKeyword, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange? timeSentRange = null) =>
public Task<QueryResult<IList<MessagesView>>> SearchEndpointMessages(string endpointName, string searchKeyword, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange? timeSentRange = null, CancellationToken cancellationToken = default) =>
ExecuteWithDbContext(dbContext => Search(dbContext.FailedMessages.AsNoTracking(), searchKeyword)
.Where(message => message.ReceivingEndpointName == endpointName)
.FilterBySentTimeRange(timeSentRange)
.ToPagedMessagesResult(pagingInfo, sortInfo));
.ToPagedMessagesResult(pagingInfo, sortInfo, cancellationToken));

// Neither search hides system messages: a caller who searched
// for something specific is not helped by hiding the message that matched it.
Expand Down
6 changes: 3 additions & 3 deletions src/ServiceControl.Persistence.EFCore/Implementation/MessagesViewQueryResults.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,14 +8,14 @@ namespace ServiceControl.Persistence.EFCore.Implementation;

static class MessagesViewQueryResults
{
public static async Task<QueryResult<IList<MessagesView>>> ToPagedMessagesResult(this IQueryable<FailedMessageEntity> source, PagingInfo pagingInfo, SortInfo sortInfo)
public static async Task<QueryResult<IList<MessagesView>>> ToPagedMessagesResult(this IQueryable<FailedMessageEntity> source, PagingInfo pagingInfo, SortInfo sortInfo, CancellationToken cancellationToken = default)
{
var stats = await source.ToQueryStatsInfo();
var stats = await source.ToQueryStatsInfo(cancellationToken);

var entities = await source
.SortMessages(sortInfo)
.Page(pagingInfo)
.ToListAsync();
.ToListAsync(cancellationToken);

IList<MessagesView> results = [.. entities.Select(entity => entity.ToMessagesView())];

Expand Down
Loading
Loading