diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageLifecycleDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageLifecycleDataStore.cs index 5036911a85..26c417d579 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageLifecycleDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageLifecycleDataStore.cs @@ -9,7 +9,7 @@ namespace ServiceControl.Persistence.EFCore.Implementation; /// 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 => { @@ -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 MarkAsResolved(string failedMessageId) + public async Task MarkAsResolved(string failedMessageId, CancellationToken cancellationToken = default) { return await ExecuteWithDbContext(async dbContext => { @@ -45,13 +45,13 @@ public async Task 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 UnArchiveMessages(IEnumerable failedMessageIds) + public async Task UnArchiveMessages(IEnumerable failedMessageIds, CancellationToken cancellationToken = default) { var ids = failedMessageIds .Select(id => Guid.TryParse(id, out var guid) ? guid : Guid.Empty) @@ -71,7 +71,7 @@ public async Task UnArchiveMessages(IEnumerable 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) { @@ -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 UnArchiveMessagesByRange(DateTime from, DateTime to) + public async Task UnArchiveMessagesByRange(DateTime from, DateTime to, CancellationToken cancellationToken = default) { return await ExecuteWithDbContext(async dbContext => { @@ -101,7 +101,7 @@ public async Task UnArchiveMessagesByRange(DateTime from, DateTime to) && fm.LastModified >= from && fm.LastModified <= to) .Select(fm => fm.UniqueMessageId) - .ToListAsync(); + .ToListAsync(cancellationToken); if (unarchivableIds.Count == 0) { @@ -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 => { @@ -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); }); } } \ No newline at end of file diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageQueryDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageQueryDataStore.cs index 8d0ea059d1..4aa82923bc 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageQueryDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageQueryDataStore.cs @@ -12,36 +12,36 @@ namespace ServiceControl.Persistence.EFCore.Implementation; public class FailedMessageQueryDataStore(IServiceScopeFactory scopeFactory) : DataStoreBase(scopeFactory), IFailedMessageQueryDataStore { - public Task>> GetFailedMessages(string? status, string? modified, string? queueAddress, PagingInfo pagingInfo, SortInfo sortInfo) => + public Task>> 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 GetFailedMessagesStats(string? status, string? modified, string? queueAddress) => + public Task 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>> GetFailedMessagesByEndpoint(string? status, string endpointName, string? modified, PagingInfo pagingInfo, SortInfo sortInfo) => + public Task>> 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> GetFailedMessagesSummary() => + public Task> 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)new Dictionary { @@ -51,7 +51,7 @@ public Task> GetFailedMessagesSummary() => }; }); - public Task GetLatestFailedMessageView(string failedMessageId) => + public Task GetLatestFailedMessageView(string failedMessageId, CancellationToken cancellationToken = default) => ExecuteWithDbContext(async dbContext => { if (!Guid.TryParse(failedMessageId, out var uniqueMessageId)) @@ -61,12 +61,12 @@ public Task> GetFailedMessagesSummary() => var entity = await dbContext.FailedMessages .AsNoTracking() - .SingleOrDefaultAsync(message => message.UniqueMessageId == uniqueMessageId); + .SingleOrDefaultAsync(message => message.UniqueMessageId == uniqueMessageId, cancellationToken); return entity?.ToFailedMessageView(); }); - public Task GetFailedMessage(string failedMessageId) => + public Task GetFailedMessage(string failedMessageId, CancellationToken cancellationToken = default) => ExecuteWithDbContext(async dbContext => { if (!Guid.TryParse(failedMessageId, out var uniqueMessageId)) @@ -76,37 +76,37 @@ public Task> 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 GetFailedMessagesByIds(Guid[] ids) => + public Task GetFailedMessagesByIds(Guid[] ids, CancellationToken cancellationToken = default) => ExecuteWithDbContext(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> CountBy(ServiceControlDbContext dbContext, Expression> selector) => + static async Task> CountBy(ServiceControlDbContext dbContext, Expression> selector, CancellationToken cancellationToken) => await dbContext.FailedMessages .AsNoTracking() .Where(message => message.Status == FailedMessageStatus.Unresolved) @@ -114,14 +114,14 @@ await dbContext.FailedMessages .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>> ReadGroups(ServiceControlDbContext dbContext, Guid[] uniqueMessageIds) + static async Task>> 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) diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageQueryResults.cs b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageQueryResults.cs index 9c6fe90250..c349a9a696 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageQueryResults.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageQueryResults.cs @@ -8,14 +8,14 @@ namespace ServiceControl.Persistence.EFCore.Implementation; static class FailedMessageQueryResults { - public static async Task>> ToPagedResult(this IQueryable source, PagingInfo pagingInfo, SortInfo sortInfo) + public static async Task>> ToPagedResult(this IQueryable 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 results = [.. entities.Select(entity => entity.ToFailedMessageView())]; diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageRetryDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageRetryDataStore.cs index 002f365279..12bc2c9c68 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageRetryDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/FailedMessageRetryDataStore.cs @@ -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 processCallback) => + public Task ProcessPendingRetries(DateTime periodFrom, DateTime periodTo, string queueAddress, Func processCallback, CancellationToken cancellationToken = default) => throw new NotImplementedException(); - public Task GetRetryPendingMessages(DateTime from, DateTime to, string queueAddress) => + public Task 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 GetFailedMessageBody(string uniqueMessageId) => + public Task GetFailedMessageBody(string uniqueMessageId, CancellationToken cancellationToken = default) => throw new NotImplementedException(); } diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/MessagesViewDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/MessagesViewDataStore.cs index 5cea7bc390..6c3d0fabc2 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/MessagesViewDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/MessagesViewDataStore.cs @@ -9,38 +9,38 @@ namespace ServiceControl.Persistence.EFCore.Implementation; public class MessagesViewDataStore(IServiceScopeFactory scopeFactory, IFullTextSearchDialect fullTextSearch) : DataStoreBase(scopeFactory), IMessagesViewDataStore { - public Task>> GetAllMessages(PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, DateTimeRange? timeSentRange = null) => + public Task>> 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>> GetAllMessagesForEndpoint(string endpointName, PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, DateTimeRange? timeSentRange = null) => + public Task>> 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>> GetAllMessagesByConversation(string conversationId, PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages) => + public Task>> 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>> GetAllMessagesForSearch(string searchTerms, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange? timeSentRange = null) => + public Task>> 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>> SearchEndpointMessages(string endpointName, string searchKeyword, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange? timeSentRange = null) => + public Task>> 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. diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/MessagesViewQueryResults.cs b/src/ServiceControl.Persistence.EFCore/Implementation/MessagesViewQueryResults.cs index 0dc362cd97..006a69170f 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/MessagesViewQueryResults.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/MessagesViewQueryResults.cs @@ -8,14 +8,14 @@ namespace ServiceControl.Persistence.EFCore.Implementation; static class MessagesViewQueryResults { - public static async Task>> ToPagedMessagesResult(this IQueryable source, PagingInfo pagingInfo, SortInfo sortInfo) + public static async Task>> ToPagedMessagesResult(this IQueryable 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 results = [.. entities.Select(entity => entity.ToMessagesView())]; diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/QueueAddressStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/QueueAddressStore.cs index a644efaadd..441c48bca2 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/QueueAddressStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/QueueAddressStore.cs @@ -7,7 +7,7 @@ namespace ServiceControl.Persistence.EFCore.Implementation; public class QueueAddressStore(IServiceScopeFactory scopeFactory) : DataStoreBase(scopeFactory), IQueueAddressStore { - public Task>> GetAddresses(PagingInfo pagingInfo) => + public Task>> GetAddresses(PagingInfo pagingInfo, CancellationToken cancellationToken = default) => ExecuteWithDbContext(async context => { var query = context.FailedMessages @@ -19,7 +19,7 @@ public Task>> GetAddresses(PagingInfo pagingInfo FailedMessageCount = failuresByEndpoint.Count() }); - var items = await query.Skip(pagingInfo.Offset).Take(pagingInfo.PageSize).ToListAsync(); + var items = await query.Skip(pagingInfo.Offset).Take(pagingInfo.PageSize).ToListAsync(cancellationToken); var eTag = DeterministicGuid.MakeId($"{items.Count}|{string.Join(",", items.Select(x => x.PhysicalAddress))}").ToString(); return new QueryResult>(items, new QueryStatsInfo(eTag, query.Count(), false)); diff --git a/src/ServiceControl.Persistence.RavenDB/ErrorMessagesDataStore.cs b/src/ServiceControl.Persistence.RavenDB/ErrorMessagesDataStore.cs index f7336ca754..c15f3ffe1f 100644 --- a/src/ServiceControl.Persistence.RavenDB/ErrorMessagesDataStore.cs +++ b/src/ServiceControl.Persistence.RavenDB/ErrorMessagesDataStore.cs @@ -4,6 +4,7 @@ using System.Collections.Generic; using System.IO; using System.Linq; + using System.Threading; using System.Threading.Tasks; using Editing; using Microsoft.Extensions.Logging; @@ -37,10 +38,11 @@ public async Task>> GetAllMessages( PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, - DateTimeRange timeSentRange + DateTimeRange timeSentRange, + CancellationToken cancellationToken = default ) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var query = session.Query() .IncludeSystemMessagesWhere(includeSystemMessages) .FilterBySentTimeRange(timeSentRange) @@ -50,7 +52,7 @@ DateTimeRange timeSentRange .OfType() .TransformToMessageView(); - var results = await query.ToListAsync(); + var results = await query.ToListAsync(cancellationToken); return new QueryResult>(results, stats.ToQueryStatsInfo()); } @@ -60,10 +62,11 @@ public async Task>> GetAllMessagesForEndpoint( PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, - DateTimeRange timeSentRange + DateTimeRange timeSentRange, + CancellationToken cancellationToken = default ) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var query = session.Query() .IncludeSystemMessagesWhere(includeSystemMessages) .FilterBySentTimeRange(timeSentRange) @@ -74,7 +77,7 @@ DateTimeRange timeSentRange .OfType() .TransformToMessageView(); - var results = await query.ToListAsync(); + var results = await query.ToListAsync(cancellationToken); return new QueryResult>(results, stats.ToQueryStatsInfo()); @@ -85,10 +88,11 @@ public async Task>> SearchEndpointMessages( string searchKeyword, PagingInfo pagingInfo, SortInfo sortInfo, - DateTimeRange timeSentRange + DateTimeRange timeSentRange, + CancellationToken cancellationToken = default ) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var query = session.Query() .Statistics(out var stats) .Search(x => x.Query, searchKeyword) @@ -99,7 +103,7 @@ DateTimeRange timeSentRange .OfType() .TransformToMessageView(); - var results = await query.ToListAsync(); + var results = await query.ToListAsync(cancellationToken); return new QueryResult>(results, stats.ToQueryStatsInfo()); } @@ -108,10 +112,11 @@ public async Task>> GetAllMessagesByConversation string conversationId, PagingInfo pagingInfo, SortInfo sortInfo, - bool includeSystemMessages + bool includeSystemMessages, + CancellationToken cancellationToken = default ) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var query = session.Query() .Statistics(out var stats) .Where(m => m.ConversationId == conversationId) @@ -120,7 +125,7 @@ bool includeSystemMessages .OfType() .TransformToMessageView(); - var results = await query.ToListAsync(); + var results = await query.ToListAsync(cancellationToken); return new QueryResult>(results, stats.ToQueryStatsInfo()); } @@ -129,10 +134,11 @@ public async Task>> GetAllMessagesForSearch( string searchTerms, PagingInfo pagingInfo, SortInfo sortInfo, - DateTimeRange timeSentRange + DateTimeRange timeSentRange, + CancellationToken cancellationToken = default ) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var query = session.Query() .Statistics(out var stats) .Search(x => x.Query, searchTerms) @@ -142,15 +148,15 @@ DateTimeRange timeSentRange .OfType() .TransformToMessageView(); - var results = await query.ToListAsync(); + var results = await query.ToListAsync(cancellationToken); return new QueryResult>(results, stats.ToQueryStatsInfo()); } - public async Task MarkAsArchived(string failedMessageId) + public async Task MarkAsArchived(string failedMessageId, CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); - var failedMessage = await session.LoadAsync(FailedMessageIdGenerator.MakeDocumentId(failedMessageId)); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); + var failedMessage = await session.LoadAsync(FailedMessageIdGenerator.MakeDocumentId(failedMessageId), cancellationToken); if (failedMessage.Status != FailedMessageStatus.Archived) { @@ -159,14 +165,14 @@ public async Task MarkAsArchived(string failedMessageId) expirationManager.EnableExpiration(session, failedMessage); } - await session.SaveChangesAsync(); + await session.SaveChangesAsync(cancellationToken); } - public async Task GetFailedMessagesByIds(Guid[] ids) + public async Task GetFailedMessagesByIds(Guid[] ids, CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var docIds = ids.Select(g => FailedMessageIdGenerator.MakeDocumentId(g.ToString())); - var results = await session.LoadAsync(docIds); + var results = await session.LoadAsync(docIds, cancellationToken); return results.Values.Where(x => x != null).ToArray(); } @@ -175,10 +181,11 @@ public async Task>> GetFailedMessages( string modified, string queueAddress, PagingInfo pagingInfo, - SortInfo sortInfo + SortInfo sortInfo, + 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) @@ -192,7 +199,7 @@ SortInfo sortInfo .TransformToFailedMessageView(); var results = await query - .ToListAsync(); + .ToListAsync(cancellationToken); return new QueryResult>(results, stats.ToQueryStatsInfo()); } @@ -200,16 +207,17 @@ SortInfo sortInfo public async Task GetFailedMessagesStats( string status, string modified, - string queueAddress + string queueAddress, + CancellationToken cancellationToken = default ) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var stats = await session.Advanced .AsyncDocumentQuery() .FilterByStatusWhere(status) .FilterByLastModifiedRange(modified) .FilterByQueueAddress(queueAddress) - .GetQueryResultAsync(); + .GetQueryResultAsync(cancellationToken); return stats.ToQueryStatsInfo(); } @@ -219,10 +227,11 @@ public async Task>> GetFailedMessagesByEndp string endpointName, string modified, PagingInfo pagingInfo, - SortInfo sortInfo + SortInfo sortInfo, + 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) @@ -237,14 +246,14 @@ SortInfo sortInfo .TransformToFailedMessageView(); var results = await query - .ToListAsync(); + .ToListAsync(cancellationToken); return new QueryResult>(results, stats.ToQueryStatsInfo()); } - public async Task> GetFailedMessagesSummary() + public async Task> GetFailedMessagesSummary(CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); // The facets are named after the index fields and renamed afterwards. Setting // DisplayFieldName instead makes the server resolve the field by that name, which then @@ -255,7 +264,7 @@ public async Task> GetFailedMessagesSummary() new Facet { FieldName = "Name" }, new Facet { FieldName = "Host" }, new Facet { FieldName = "MessageType" } - }).ExecuteAsync(); + }).ExecuteAsync(cancellationToken); return new Dictionary { @@ -268,17 +277,17 @@ public async Task> GetFailedMessagesSummary() static Dictionary Counts(Dictionary facetResults, string fieldName) => facetResults[fieldName].Values.ToDictionary(value => value.Range, value => (object)value.Count); - public async Task GetFailedMessage(string failedMessageId) + public async Task GetFailedMessage(string failedMessageId, CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); - var message = await session.LoadAsync(FailedMessageIdGenerator.MakeDocumentId(failedMessageId)); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); + var message = await session.LoadAsync(FailedMessageIdGenerator.MakeDocumentId(failedMessageId), cancellationToken); return message; } - public async Task GetLatestFailedMessageView(string failedMessageId) + public async Task GetLatestFailedMessageView(string failedMessageId, CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); - var message = await session.LoadAsync(FailedMessageIdGenerator.MakeDocumentId(failedMessageId)); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); + var message = await session.LoadAsync(FailedMessageIdGenerator.MakeDocumentId(failedMessageId), cancellationToken); if (message == null) { return null; @@ -336,14 +345,14 @@ FailedMessageView Map(FailedMessage message, IAsyncDocumentSession session) } - public async Task MarkAsResolved(string failedMessageId) + public async Task MarkAsResolved(string failedMessageId, CancellationToken cancellationToken = default) { var documentId = FailedMessageIdGenerator.MakeDocumentId(failedMessageId); - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); session.Advanced.UseOptimisticConcurrency = true; - var failedMessage = await session.LoadAsync(documentId); + var failedMessage = await session.LoadAsync(documentId, cancellationToken); if (failedMessage == null) { @@ -354,14 +363,14 @@ public async Task MarkAsResolved(string failedMessageId) expirationManager.EnableExpiration(session, failedMessage); - await session.SaveChangesAsync(); + await session.SaveChangesAsync(cancellationToken); return true; } - public async Task ProcessPendingRetries(DateTime periodFrom, DateTime periodTo, string queueAddress, Func processCallback) + public async Task ProcessPendingRetries(DateTime periodFrom, DateTime periodTo, string queueAddress, Func processCallback, CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var prequery = session.Advanced .AsyncDocumentQuery() .WhereEquals("Status", (int)FailedMessageStatus.RetryIssued) @@ -379,14 +388,14 @@ public async Task ProcessPendingRetries(DateTime periodFrom, DateTime periodTo, .ToQueryable() .TransformToFailedMessageView(); - await using var ie = await session.Advanced.StreamAsync(query); + await using var ie = await session.Advanced.StreamAsync(query, cancellationToken); while (await ie.MoveNextAsync()) { - await processCallback(ie.Current.Document.Id); + await processCallback(ie.Current.Document.Id, cancellationToken); } } - public async Task UnArchiveMessagesByRange(DateTime from, DateTime to) + public async Task UnArchiveMessagesByRange(DateTime from, DateTime to, CancellationToken cancellationToken = default) { const int Unresolved = (int)FailedMessageStatus.Unresolved; const int Archived = (int)FailedMessageStatus.Archived; @@ -418,8 +427,8 @@ public async Task UnArchiveMessagesByRange(DateTime from, DateTime to) RetrieveDetails = true }); - var documentStore = await documentStoreProvider.GetDocumentStore(); - var operation = await documentStore.Operations.SendAsync(patch); + var documentStore = await documentStoreProvider.GetDocumentStore(cancellationToken); + var operation = await documentStore.Operations.SendAsync(patch, token: cancellationToken); var result = await operation.WaitForCompletionAsync(); @@ -430,16 +439,16 @@ public async Task UnArchiveMessagesByRange(DateTime from, DateTime to) return ids; } - public async Task UnArchiveMessages(IEnumerable failedMessageIds) + public async Task UnArchiveMessages(IEnumerable failedMessageIds, CancellationToken cancellationToken = default) { Dictionary failedMessages; - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); session.Advanced.UseOptimisticConcurrency = true; var documentIds = failedMessageIds.Select(FailedMessageIdGenerator.MakeDocumentId); - failedMessages = await session.LoadAsync(documentIds); + failedMessages = await session.LoadAsync(documentIds, cancellationToken); foreach (var failedMessage in failedMessages.Values) { @@ -450,45 +459,45 @@ public async Task UnArchiveMessages(IEnumerable failedMessageI } } - await session.SaveChangesAsync(); + await session.SaveChangesAsync(cancellationToken); // Return the unique IDs - the dictionary keys are document ids with a prefix return failedMessages.Values.Select(x => x.UniqueMessageId).ToArray(); } - public async Task RevertRetry(string messageUniqueId) + public async Task RevertRetry(string messageUniqueId, CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var failedMessage = await session - .LoadAsync(FailedMessageIdGenerator.MakeDocumentId(messageUniqueId)); + .LoadAsync(FailedMessageIdGenerator.MakeDocumentId(messageUniqueId), cancellationToken); failedMessage?.Status = FailedMessageStatus.Unresolved; var failedMessageRetry = await session - .LoadAsync(RetryDocumentDataStore.MakeFailedMessageRetriesDocumentId(messageUniqueId)); + .LoadAsync(RetryDocumentDataStore.MakeFailedMessageRetriesDocumentId(messageUniqueId), cancellationToken); if (failedMessageRetry != null) { session.Delete(failedMessageRetry); } - await session.SaveChangesAsync(); + await session.SaveChangesAsync(cancellationToken); } - public async Task RemoveFailedMessageRetry(string uniqueMessageId) + public async Task RemoveFailedMessageRetry(string uniqueMessageId, CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); - await session.Advanced.RequestExecutor.ExecuteAsync(new DeleteDocumentCommand(RetryDocumentDataStore.MakeFailedMessageRetriesDocumentId(uniqueMessageId), null), session.Advanced.Context); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); + await session.Advanced.RequestExecutor.ExecuteAsync(new DeleteDocumentCommand(RetryDocumentDataStore.MakeFailedMessageRetriesDocumentId(uniqueMessageId), null), session.Advanced.Context, token: cancellationToken); } - public async Task GetRetryPendingMessages(DateTime from, DateTime to, string queueAddress) + public async Task GetRetryPendingMessages(DateTime from, DateTime to, string queueAddress, CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var query = session .Query() .Where(o => o.Status == FailedMessageStatus.RetryIssued && o.LastModified >= from.Ticks && o.LastModified <= to.Ticks && o.QueueAddress == queueAddress) .OfType(); int index = 0; - await using var streamResults = await session.Advanced.StreamAsync(query, out var streamQueryStatistics); + await using var streamResults = await session.Advanced.StreamAsync(query, out var streamQueryStatistics, cancellationToken); string[] ids = new string[streamQueryStatistics.TotalResults]; while (await streamResults.MoveNextAsync()) { @@ -499,7 +508,7 @@ public async Task GetRetryPendingMessages(DateTime from, DateTime to, record struct FailedMessageProjection(string UniqueMessageId); - public async Task GetFailedMessageBody(string uniqueMessageId) + public async Task GetFailedMessageBody(string uniqueMessageId, CancellationToken cancellationToken = default) { byte[] body = null; var result = await bodyStorage.TryFetch(uniqueMessageId) @@ -515,7 +524,7 @@ public async Task GetFailedMessageBody(string uniqueMessageId) // this assumption might change when we stop supporting RavenDB 3.5 but right now this is the most memory efficient way to do things // https://github.com/microsoft/Microsoft.IO.RecyclableMemoryStream#getbuffer-and-toarray using var memoryStream = new MemoryStream(); - await result.Stream.CopyToAsync(memoryStream); + await result.Stream.CopyToAsync(memoryStream, cancellationToken); body = memoryStream.ToArray(); } diff --git a/src/ServiceControl.Persistence.RavenDB/QueueAddressStore.cs b/src/ServiceControl.Persistence.RavenDB/QueueAddressStore.cs index 0b4705fa44..2b06bfa592 100644 --- a/src/ServiceControl.Persistence.RavenDB/QueueAddressStore.cs +++ b/src/ServiceControl.Persistence.RavenDB/QueueAddressStore.cs @@ -2,6 +2,7 @@ { using System.Collections.Generic; using System.Linq; + using System.Threading; using System.Threading.Tasks; using Raven.Client.Documents; using ServiceControl.MessageFailures; @@ -10,14 +11,14 @@ class QueueAddressStore(IRavenSessionProvider sessionProvider) : IQueueAddressStore { - public async Task>> GetAddresses(PagingInfo pagingInfo) + public async Task>> GetAddresses(PagingInfo pagingInfo, CancellationToken cancellationToken = default) { - using var session = await sessionProvider.OpenSession(); + using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var addresses = await session .Query() .Statistics(out var stats) .Paging(pagingInfo) - .ToListAsync(); + .ToListAsync(cancellationToken); var result = new QueryResult>(addresses, stats.ToQueryStatsInfo()); return result; diff --git a/src/ServiceControl.Persistence.Tests/Recoverability/ReturnToSenderDequeuerTests.cs b/src/ServiceControl.Persistence.Tests/Recoverability/ReturnToSenderDequeuerTests.cs index 62224a020e..7adc218149 100644 --- a/src/ServiceControl.Persistence.Tests/Recoverability/ReturnToSenderDequeuerTests.cs +++ b/src/ServiceControl.Persistence.Tests/Recoverability/ReturnToSenderDequeuerTests.cs @@ -169,13 +169,13 @@ public Task Dispatch(TransportOperations outgoingMessages, TransportTransaction class FakeErrorMessageDataStore : IFailedMessageRetryDataStore { - public Task ProcessPendingRetries(DateTime periodFrom, DateTime periodTo, string queueAddress, Func processCallback) => throw new NotImplementedException(); + public Task ProcessPendingRetries(DateTime periodFrom, DateTime periodTo, string queueAddress, Func processCallback, CancellationToken cancellationToken = default) => throw new NotImplementedException(); - public Task GetRetryPendingMessages(DateTime from, DateTime to, string queueAddress) => throw new NotImplementedException(); + public Task GetRetryPendingMessages(DateTime from, DateTime to, string queueAddress, CancellationToken cancellationToken = default) => throw new NotImplementedException(); - public Task RemoveFailedMessageRetry(string uniqueMessageId) => throw new NotImplementedException(); + public Task RemoveFailedMessageRetry(string uniqueMessageId, CancellationToken cancellationToken = default) => throw new NotImplementedException(); - public Task GetFailedMessageBody(string bodyId) => Task.FromResult(Encoding.UTF8.GetBytes(bodyId)); + public Task GetFailedMessageBody(string bodyId, CancellationToken cancellationToken = default) => Task.FromResult(Encoding.UTF8.GetBytes(bodyId)); } } } \ No newline at end of file diff --git a/src/ServiceControl.Persistence/IFailedMessageLifecycleDataStore.cs b/src/ServiceControl.Persistence/IFailedMessageLifecycleDataStore.cs index edae2a54ea..08e0fce30d 100644 --- a/src/ServiceControl.Persistence/IFailedMessageLifecycleDataStore.cs +++ b/src/ServiceControl.Persistence/IFailedMessageLifecycleDataStore.cs @@ -2,6 +2,7 @@ namespace ServiceControl.Persistence { using System; using System.Collections.Generic; + using System.Threading; using System.Threading.Tasks; /// @@ -10,10 +11,10 @@ namespace ServiceControl.Persistence /// public interface IFailedMessageLifecycleDataStore { - Task MarkAsArchived(string failedMessageId); - Task MarkAsResolved(string failedMessageId); - Task UnArchiveMessages(IEnumerable failedMessageIds); - Task UnArchiveMessagesByRange(DateTime from, DateTime to); - Task RevertRetry(string messageUniqueId); + Task MarkAsArchived(string failedMessageId, CancellationToken cancellationToken = default); + Task MarkAsResolved(string failedMessageId, CancellationToken cancellationToken = default); + Task UnArchiveMessages(IEnumerable failedMessageIds, CancellationToken cancellationToken = default); + Task UnArchiveMessagesByRange(DateTime from, DateTime to, CancellationToken cancellationToken = default); + Task RevertRetry(string messageUniqueId, CancellationToken cancellationToken = default); } } diff --git a/src/ServiceControl.Persistence/IFailedMessageQueryDataStore.cs b/src/ServiceControl.Persistence/IFailedMessageQueryDataStore.cs index c299dcf10b..12a6235524 100644 --- a/src/ServiceControl.Persistence/IFailedMessageQueryDataStore.cs +++ b/src/ServiceControl.Persistence/IFailedMessageQueryDataStore.cs @@ -3,6 +3,7 @@ namespace ServiceControl.Persistence { using System; using System.Collections.Generic; + using System.Threading; using System.Threading.Tasks; using Infrastructure; using MessageFailures.Api; @@ -10,13 +11,13 @@ namespace ServiceControl.Persistence public interface IFailedMessageQueryDataStore { - Task>> GetFailedMessages(string? status, string? modified, string? queueAddress, PagingInfo pagingInfo, SortInfo sortInfo); - Task GetFailedMessagesStats(string? status, string? modified, string? queueAddress); - Task>> GetFailedMessagesByEndpoint(string? status, string endpointName, string? modified, PagingInfo pagingInfo, SortInfo sortInfo); - Task> GetFailedMessagesSummary(); - Task GetLatestFailedMessageView(string failedMessageId); - Task GetFailedMessage(string failedMessageId); + Task>> GetFailedMessages(string? status, string? modified, string? queueAddress, PagingInfo pagingInfo, SortInfo sortInfo, CancellationToken cancellationToken = default); + Task GetFailedMessagesStats(string? status, string? modified, string? queueAddress, CancellationToken cancellationToken = default); + Task>> GetFailedMessagesByEndpoint(string? status, string endpointName, string? modified, PagingInfo pagingInfo, SortInfo sortInfo, CancellationToken cancellationToken = default); + Task> GetFailedMessagesSummary(CancellationToken cancellationToken = default); + Task GetLatestFailedMessageView(string failedMessageId, CancellationToken cancellationToken = default); + Task GetFailedMessage(string failedMessageId, CancellationToken cancellationToken = default); /// Ids with no stored failed message are skipped, so the result can be shorter than . - Task GetFailedMessagesByIds(Guid[] ids); + Task GetFailedMessagesByIds(Guid[] ids, CancellationToken cancellationToken = default); } } diff --git a/src/ServiceControl.Persistence/IFailedMessageRetryDataStore.cs b/src/ServiceControl.Persistence/IFailedMessageRetryDataStore.cs index a55178edb6..8add2b6246 100644 --- a/src/ServiceControl.Persistence/IFailedMessageRetryDataStore.cs +++ b/src/ServiceControl.Persistence/IFailedMessageRetryDataStore.cs @@ -1,13 +1,14 @@ namespace ServiceControl.Persistence { using System; + using System.Threading; using System.Threading.Tasks; public interface IFailedMessageRetryDataStore { - Task ProcessPendingRetries(DateTime periodFrom, DateTime periodTo, string queueAddress, Func processCallback); - Task GetRetryPendingMessages(DateTime from, DateTime to, string queueAddress); - Task RemoveFailedMessageRetry(string uniqueMessageId); - Task GetFailedMessageBody(string uniqueMessageId); + Task ProcessPendingRetries(DateTime periodFrom, DateTime periodTo, string queueAddress, Func processCallback, CancellationToken cancellationToken = default); + Task GetRetryPendingMessages(DateTime from, DateTime to, string queueAddress, CancellationToken cancellationToken = default); + Task RemoveFailedMessageRetry(string uniqueMessageId, CancellationToken cancellationToken = default); + Task GetFailedMessageBody(string uniqueMessageId, CancellationToken cancellationToken = default); } } diff --git a/src/ServiceControl.Persistence/IMessagesViewDataStore.cs b/src/ServiceControl.Persistence/IMessagesViewDataStore.cs index 281ca903ba..ad7ae30200 100644 --- a/src/ServiceControl.Persistence/IMessagesViewDataStore.cs +++ b/src/ServiceControl.Persistence/IMessagesViewDataStore.cs @@ -2,16 +2,17 @@ namespace ServiceControl.Persistence { using System; using System.Collections.Generic; + using System.Threading; using System.Threading.Tasks; using CompositeViews.Messages; using Infrastructure; public interface IMessagesViewDataStore { - Task>> GetAllMessages(PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, DateTimeRange timeSentRange = null); - Task>> GetAllMessagesForEndpoint(string endpointName, PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, DateTimeRange timeSentRange = null); - Task>> GetAllMessagesByConversation(string conversationId, PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages); - Task>> GetAllMessagesForSearch(string searchTerms, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange = null); - Task>> SearchEndpointMessages(string endpointName, string searchKeyword, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange = null); + Task>> GetAllMessages(PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, DateTimeRange timeSentRange = null, CancellationToken cancellationToken = default); + Task>> GetAllMessagesForEndpoint(string endpointName, PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, DateTimeRange timeSentRange = null, CancellationToken cancellationToken = default); + Task>> GetAllMessagesByConversation(string conversationId, PagingInfo pagingInfo, SortInfo sortInfo, bool includeSystemMessages, CancellationToken cancellationToken = default); + Task>> GetAllMessagesForSearch(string searchTerms, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange = null, CancellationToken cancellationToken = default); + Task>> SearchEndpointMessages(string endpointName, string searchKeyword, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange = null, CancellationToken cancellationToken = default); } } diff --git a/src/ServiceControl.Persistence/IQueueAddressStore.cs b/src/ServiceControl.Persistence/IQueueAddressStore.cs index 07ab28618e..4073504b90 100644 --- a/src/ServiceControl.Persistence/IQueueAddressStore.cs +++ b/src/ServiceControl.Persistence/IQueueAddressStore.cs @@ -1,12 +1,13 @@ namespace ServiceControl.Persistence { using System.Collections.Generic; + using System.Threading; using System.Threading.Tasks; using ServiceControl.MessageFailures; using ServiceControl.Persistence.Infrastructure; public interface IQueueAddressStore { - Task>> GetAddresses(PagingInfo pagingInfo); + Task>> GetAddresses(PagingInfo pagingInfo, CancellationToken cancellationToken = default); } } \ No newline at end of file diff --git a/src/ServiceControl.UnitTests/MessageFailures/AsyncRangeAndQueueAuditTests.cs b/src/ServiceControl.UnitTests/MessageFailures/AsyncRangeAndQueueAuditTests.cs index 6e2784070a..edf12b4bc1 100644 --- a/src/ServiceControl.UnitTests/MessageFailures/AsyncRangeAndQueueAuditTests.cs +++ b/src/ServiceControl.UnitTests/MessageFailures/AsyncRangeAndQueueAuditTests.cs @@ -159,22 +159,22 @@ internal sealed class StubErrorMessageDataStore : IFailedMessageQueryDataStore, public string[] UnArchiveMessagesResult { get; set; } = []; public FailedMessage ErrorByResult { get; set; } = new(); - public Task GetRetryPendingMessages(DateTime from, DateTime to, string queueAddress) => Task.FromResult(RetryPendingMessagesResult); - public Task RemoveFailedMessageRetry(string uniqueMessageId) => Task.CompletedTask; - public Task UnArchiveMessagesByRange(DateTime from, DateTime to) => Task.FromResult(UnArchiveByRangeResult); - public Task UnArchiveMessages(IEnumerable failedMessageIds) => Task.FromResult(UnArchiveMessagesResult); - public Task GetFailedMessage(string failedMessageId) => Task.FromResult(ErrorByResult); - public Task MarkAsArchived(string failedMessageId) => Task.CompletedTask; - - public Task GetFailedMessagesByIds(Guid[] ids) => throw new NotImplementedException(); - public Task>> GetFailedMessages(string? status, string? modified, string? queueAddress, PagingInfo pagingInfo, SortInfo sortInfo) => throw new NotImplementedException(); - public Task GetFailedMessagesStats(string? status, string? modified, string? queueAddress) => throw new NotImplementedException(); - public Task>> GetFailedMessagesByEndpoint(string? status, string endpointName, string? modified, PagingInfo pagingInfo, SortInfo sortInfo) => throw new NotImplementedException(); - public Task> GetFailedMessagesSummary() => throw new NotImplementedException(); - public Task GetLatestFailedMessageView(string failedMessageId) => throw new NotImplementedException(); - public Task MarkAsResolved(string failedMessageId) => throw new NotImplementedException(); - public Task ProcessPendingRetries(DateTime periodFrom, DateTime periodTo, string queueAddress, Func processCallback) => throw new NotImplementedException(); - public Task RevertRetry(string messageUniqueId) => throw new NotImplementedException(); - public Task GetFailedMessageBody(string uniqueMessageId) => throw new NotImplementedException(); + public Task GetRetryPendingMessages(DateTime from, DateTime to, string queueAddress, CancellationToken cancellationToken = default) => Task.FromResult(RetryPendingMessagesResult); + public Task RemoveFailedMessageRetry(string uniqueMessageId, CancellationToken cancellationToken = default) => Task.CompletedTask; + public Task UnArchiveMessagesByRange(DateTime from, DateTime to, CancellationToken cancellationToken = default) => Task.FromResult(UnArchiveByRangeResult); + public Task UnArchiveMessages(IEnumerable failedMessageIds, CancellationToken cancellationToken = default) => Task.FromResult(UnArchiveMessagesResult); + public Task GetFailedMessage(string failedMessageId, CancellationToken cancellationToken = default) => Task.FromResult(ErrorByResult); + public Task MarkAsArchived(string failedMessageId, CancellationToken cancellationToken = default) => Task.CompletedTask; + + public Task GetFailedMessagesByIds(Guid[] ids, CancellationToken cancellationToken = default) => throw new NotImplementedException(); + public Task>> GetFailedMessages(string? status, string? modified, string? queueAddress, PagingInfo pagingInfo, SortInfo sortInfo, CancellationToken cancellationToken = default) => throw new NotImplementedException(); + public Task GetFailedMessagesStats(string? status, string? modified, string? queueAddress, CancellationToken cancellationToken = default) => throw new NotImplementedException(); + public Task>> GetFailedMessagesByEndpoint(string? status, string endpointName, string? modified, PagingInfo pagingInfo, SortInfo sortInfo, CancellationToken cancellationToken = default) => throw new NotImplementedException(); + public Task> GetFailedMessagesSummary(CancellationToken cancellationToken = default) => throw new NotImplementedException(); + public Task GetLatestFailedMessageView(string failedMessageId, CancellationToken cancellationToken = default) => throw new NotImplementedException(); + public Task MarkAsResolved(string failedMessageId, CancellationToken cancellationToken = default) => throw new NotImplementedException(); + public Task ProcessPendingRetries(DateTime periodFrom, DateTime periodTo, string queueAddress, Func processCallback, CancellationToken cancellationToken = default) => throw new NotImplementedException(); + public Task RevertRetry(string messageUniqueId, CancellationToken cancellationToken = default) => throw new NotImplementedException(); + public Task GetFailedMessageBody(string uniqueMessageId, CancellationToken cancellationToken = default) => throw new NotImplementedException(); } } diff --git a/src/ServiceControl.UnitTests/MessageFailures/EditFailedMessagesControllerAuditTests.cs b/src/ServiceControl.UnitTests/MessageFailures/EditFailedMessagesControllerAuditTests.cs index bab51af05a..df63c51e38 100644 --- a/src/ServiceControl.UnitTests/MessageFailures/EditFailedMessagesControllerAuditTests.cs +++ b/src/ServiceControl.UnitTests/MessageFailures/EditFailedMessagesControllerAuditTests.cs @@ -4,6 +4,7 @@ namespace ServiceControl.UnitTests.MessageFailures; using System; using System.Collections.Generic; using System.Linq; +using System.Threading; using System.Threading.Tasks; using CompositeViews.Messages; using Microsoft.Extensions.Logging.Abstractions; @@ -63,13 +64,13 @@ sealed class StubErrorMessageDataStore : IFailedMessageQueryDataStore, IEditFail public FakeEditFailedMessagesManager EditManager { get; } = new(); public Task CreateEditFailedMessageManager() => Task.FromResult(EditManager); - public Task GetFailedMessage(string failedMessageId) => Task.FromResult(ErrorByResult); + public Task GetFailedMessage(string failedMessageId, CancellationToken cancellationToken = default) => Task.FromResult(ErrorByResult); - public Task GetFailedMessagesByIds(Guid[] ids) => throw new NotImplementedException(); - public Task>> GetFailedMessages(string? status, string? modified, string? queueAddress, PagingInfo pagingInfo, SortInfo sortInfo) => throw new NotImplementedException(); - public Task GetFailedMessagesStats(string? status, string? modified, string? queueAddress) => throw new NotImplementedException(); - public Task>> GetFailedMessagesByEndpoint(string? status, string endpointName, string? modified, PagingInfo pagingInfo, SortInfo sortInfo) => throw new NotImplementedException(); - public Task> GetFailedMessagesSummary() => throw new NotImplementedException(); - public Task GetLatestFailedMessageView(string failedMessageId) => throw new NotImplementedException(); + public Task GetFailedMessagesByIds(Guid[] ids, CancellationToken cancellationToken = default) => throw new NotImplementedException(); + public Task>> GetFailedMessages(string? status, string? modified, string? queueAddress, PagingInfo pagingInfo, SortInfo sortInfo, CancellationToken cancellationToken = default) => throw new NotImplementedException(); + public Task GetFailedMessagesStats(string? status, string? modified, string? queueAddress, CancellationToken cancellationToken = default) => throw new NotImplementedException(); + public Task>> GetFailedMessagesByEndpoint(string? status, string endpointName, string? modified, PagingInfo pagingInfo, SortInfo sortInfo, CancellationToken cancellationToken = default) => throw new NotImplementedException(); + public Task> GetFailedMessagesSummary(CancellationToken cancellationToken = default) => throw new NotImplementedException(); + public Task GetLatestFailedMessageView(string failedMessageId, CancellationToken cancellationToken = default) => throw new NotImplementedException(); } } diff --git a/src/ServiceControl/CompositeViews/Messages/GetAllMessagesApi.cs b/src/ServiceControl/CompositeViews/Messages/GetAllMessagesApi.cs index 51a9c16b78..1ab818b44c 100644 --- a/src/ServiceControl/CompositeViews/Messages/GetAllMessagesApi.cs +++ b/src/ServiceControl/CompositeViews/Messages/GetAllMessagesApi.cs @@ -19,7 +19,7 @@ public GetAllMessagesApi(IMessagesViewDataStore dataStore, Settings settings, IH protected override Task>> LocalQuery(ScatterGatherApiMessageViewWithSystemMessagesContext input, CancellationToken cancellationToken = default) { - return DataStore.GetAllMessages(input.PagingInfo, input.SortInfo, input.IncludeSystemMessages, input.TimeSentRange); + return DataStore.GetAllMessages(input.PagingInfo, input.SortInfo, input.IncludeSystemMessages, input.TimeSentRange, cancellationToken); } } } \ No newline at end of file diff --git a/src/ServiceControl/CompositeViews/Messages/GetAllMessagesForEndpointApi.cs b/src/ServiceControl/CompositeViews/Messages/GetAllMessagesForEndpointApi.cs index c7d9cfcc29..61913d996e 100644 --- a/src/ServiceControl/CompositeViews/Messages/GetAllMessagesForEndpointApi.cs +++ b/src/ServiceControl/CompositeViews/Messages/GetAllMessagesForEndpointApi.cs @@ -25,6 +25,6 @@ public GetAllMessagesForEndpointApi(IMessagesViewDataStore dataStore, Settings s { } - protected override Task>> LocalQuery(AllMessagesForEndpointContext input, CancellationToken cancellationToken = default) => DataStore.GetAllMessagesForEndpoint(input.EndpointName, input.PagingInfo, input.SortInfo, input.IncludeSystemMessages, input.TimeSentRange); + protected override Task>> LocalQuery(AllMessagesForEndpointContext input, CancellationToken cancellationToken = default) => DataStore.GetAllMessagesForEndpoint(input.EndpointName, input.PagingInfo, input.SortInfo, input.IncludeSystemMessages, input.TimeSentRange, cancellationToken); } } \ No newline at end of file diff --git a/src/ServiceControl/CompositeViews/Messages/MessagesByConversationApi.cs b/src/ServiceControl/CompositeViews/Messages/MessagesByConversationApi.cs index 22c6323adb..968a7cfc79 100644 --- a/src/ServiceControl/CompositeViews/Messages/MessagesByConversationApi.cs +++ b/src/ServiceControl/CompositeViews/Messages/MessagesByConversationApi.cs @@ -26,6 +26,6 @@ public MessagesByConversationApi(IMessagesViewDataStore dataStore, Settings sett protected override Task>> LocalQuery(MessagesByConversationContext input, CancellationToken cancellationToken = default) => DataStore.GetAllMessagesByConversation(input.ConversationId, input.PagingInfo, input.SortInfo, - input.IncludeSystemMessages); + input.IncludeSystemMessages, cancellationToken); } } \ No newline at end of file diff --git a/src/ServiceControl/CompositeViews/Messages/SearchApi.cs b/src/ServiceControl/CompositeViews/Messages/SearchApi.cs index e9fedc48a3..bbf04c4d9e 100644 --- a/src/ServiceControl/CompositeViews/Messages/SearchApi.cs +++ b/src/ServiceControl/CompositeViews/Messages/SearchApi.cs @@ -25,6 +25,6 @@ public SearchApi(IMessagesViewDataStore dataStore, Settings settings, IHttpClien } protected override Task>> LocalQuery(SearchApiContext input, CancellationToken cancellationToken = default) => - DataStore.GetAllMessagesForSearch(input.SearchQuery, input.PagingInfo, input.SortInfo, input.TimeSentRange); + DataStore.GetAllMessagesForSearch(input.SearchQuery, input.PagingInfo, input.SortInfo, input.TimeSentRange, cancellationToken); } } \ No newline at end of file diff --git a/src/ServiceControl/CompositeViews/Messages/SearchEndpointApi.cs b/src/ServiceControl/CompositeViews/Messages/SearchEndpointApi.cs index 499879c2b2..2d43a472fe 100644 --- a/src/ServiceControl/CompositeViews/Messages/SearchEndpointApi.cs +++ b/src/ServiceControl/CompositeViews/Messages/SearchEndpointApi.cs @@ -26,6 +26,6 @@ public SearchEndpointApi(IMessagesViewDataStore dataStore, Settings settings, IH } protected override Task>> LocalQuery(SearchEndpointContext input, CancellationToken cancellationToken = default) => - DataStore.SearchEndpointMessages(input.Endpoint, input.Keyword, input.PagingInfo, input.SortInfo, input.TimeSentRange); + DataStore.SearchEndpointMessages(input.Endpoint, input.Keyword, input.PagingInfo, input.SortInfo, input.TimeSentRange, cancellationToken); } } \ No newline at end of file diff --git a/src/ServiceControl/MessageFailures/Api/EditFailedMessagesController.cs b/src/ServiceControl/MessageFailures/Api/EditFailedMessagesController.cs index 3ef7cd96b5..e1fed0040d 100644 --- a/src/ServiceControl/MessageFailures/Api/EditFailedMessagesController.cs +++ b/src/ServiceControl/MessageFailures/Api/EditFailedMessagesController.cs @@ -54,7 +54,7 @@ public async Task> Edit(string failedMessageId, return Ok(new EditRetryResponse { EditIgnored = true }); } - var failedMessage = await store.GetFailedMessage(failedMessageId); + var failedMessage = await store.GetFailedMessage(failedMessageId, cancellationToken); if (failedMessage == null) { diff --git a/src/ServiceControl/MessageFailures/Api/GetAllErrorsController.cs b/src/ServiceControl/MessageFailures/Api/GetAllErrorsController.cs index 507ee6c63a..a4bc5bcd70 100644 --- a/src/ServiceControl/MessageFailures/Api/GetAllErrorsController.cs +++ b/src/ServiceControl/MessageFailures/Api/GetAllErrorsController.cs @@ -24,7 +24,8 @@ public async Task> ErrorsGet([FromQuery] PagingInfo pag modified: modified, queueAddress: queueAddress, pagingInfo, - sortInfo + sortInfo, + cancellationToken ); Response.WithQueryStatsAndPagingInfo(results.QueryStats, pagingInfo); @@ -40,7 +41,8 @@ public async Task ErrorsHead(string status, string modified, string queueAddress var queryResult = await store.GetFailedMessagesStats( status: status, modified: modified, - queueAddress: queueAddress + queueAddress: queueAddress, + cancellationToken: cancellationToken ); Response.WithQueryStatsInfo(queryResult); @@ -56,7 +58,8 @@ public async Task> ErrorsByEndpointName([FromQuery] Pag endpointName: endpointName, modified: modified, pagingInfo, - sortInfo + sortInfo, + cancellationToken ); Response.WithQueryStatsAndPagingInfo(results.QueryStats, pagingInfo); @@ -67,6 +70,6 @@ public async Task> ErrorsByEndpointName([FromQuery] Pag [Authorize(Policy = Permissions.ErrorMessagesView)] [Route("errors/summary")] [HttpGet] - public async Task> ErrorsSummary(CancellationToken cancellationToken = default) => await store.GetFailedMessagesSummary(); + public async Task> ErrorsSummary(CancellationToken cancellationToken = default) => await store.GetFailedMessagesSummary(cancellationToken); } } \ No newline at end of file diff --git a/src/ServiceControl/MessageFailures/Api/GetErrorByIdController.cs b/src/ServiceControl/MessageFailures/Api/GetErrorByIdController.cs index 65eb2ba47e..0c276e5968 100644 --- a/src/ServiceControl/MessageFailures/Api/GetErrorByIdController.cs +++ b/src/ServiceControl/MessageFailures/Api/GetErrorByIdController.cs @@ -16,7 +16,7 @@ public class GetErrorByIdController(IFailedMessageQueryDataStore store) : Contro [HttpGet] public async Task> ErrorBy(string failedMessageId, CancellationToken cancellationToken = default) { - var result = await store.GetFailedMessage(failedMessageId); + var result = await store.GetFailedMessage(failedMessageId, cancellationToken); return result == null ? NotFound() : result; } @@ -26,7 +26,7 @@ public async Task> ErrorBy(string failedMessageId, C [HttpGet] public async Task> ErrorLastBy(string failedMessageId, CancellationToken cancellationToken = default) { - var result = await store.GetLatestFailedMessageView(failedMessageId); + var result = await store.GetLatestFailedMessageView(failedMessageId, cancellationToken); return result == null ? NotFound() : result; } diff --git a/src/ServiceControl/MessageFailures/Api/QueueAddressController.cs b/src/ServiceControl/MessageFailures/Api/QueueAddressController.cs index 3aeb27e031..bec2c23ec2 100644 --- a/src/ServiceControl/MessageFailures/Api/QueueAddressController.cs +++ b/src/ServiceControl/MessageFailures/Api/QueueAddressController.cs @@ -19,7 +19,7 @@ public class QueueAddressController(IQueueAddressStore store) : ControllerBase [HttpGet] public async Task> GetAddresses([FromQuery] PagingInfo pagingInfo, CancellationToken cancellationToken = default) { - var result = await store.GetAddresses(pagingInfo); + var result = await store.GetAddresses(pagingInfo, cancellationToken); Response.WithQueryStatsAndPagingInfo(result.QueryStats, pagingInfo); diff --git a/src/ServiceControl/MessageFailures/Handlers/ArchiveMessageHandler.cs b/src/ServiceControl/MessageFailures/Handlers/ArchiveMessageHandler.cs index 1580c8b95f..8b55d03425 100644 --- a/src/ServiceControl/MessageFailures/Handlers/ArchiveMessageHandler.cs +++ b/src/ServiceControl/MessageFailures/Handlers/ArchiveMessageHandler.cs @@ -15,7 +15,7 @@ public async Task Handle(ArchiveMessage message, IMessageHandlerContext context) { var failedMessageId = message.FailedMessageId; - var failedMessage = await queryStore.GetFailedMessage(failedMessageId); + var failedMessage = await queryStore.GetFailedMessage(failedMessageId, context.CancellationToken); if (failedMessage.Status != FailedMessageStatus.Archived) { @@ -24,7 +24,7 @@ await domainEvents.Raise(new FailedMessageArchived FailedMessageId = failedMessageId }, context.CancellationToken); - await lifecycleStore.MarkAsArchived(failedMessageId); + await lifecycleStore.MarkAsArchived(failedMessageId, context.CancellationToken); var (user, operationId) = AuditHeaders.Read(context.MessageHeaders); if (!string.IsNullOrEmpty(operationId)) diff --git a/src/ServiceControl/MessageFailures/Handlers/MessageFailureResolvedHandler.cs b/src/ServiceControl/MessageFailures/Handlers/MessageFailureResolvedHandler.cs index 9663593117..508f8c14bc 100644 --- a/src/ServiceControl/MessageFailures/Handlers/MessageFailureResolvedHandler.cs +++ b/src/ServiceControl/MessageFailures/Handlers/MessageFailureResolvedHandler.cs @@ -1,5 +1,6 @@ namespace ServiceControl.MessageFailures.Handlers { + using System.Threading; using System.Threading.Tasks; using Contracts.MessageFailures; using Infrastructure.DomainEvents; @@ -14,7 +15,7 @@ class MessageFailureResolvedHandler(IFailedMessageRetryDataStore retryStore, IFa { public Task Handle(MarkPendingRetriesAsResolved message, IMessageHandlerContext context) { - Task ProcessCallback(string id) + Task ProcessCallback(string id, CancellationToken cancellationToken) { var sendOptions = new SendOptions(); sendOptions.RouteToThisEndpoint(); @@ -28,13 +29,14 @@ Task ProcessCallback(string id) message.PeriodFrom, message.PeriodTo, message.QueueAddress, - ProcessCallback + ProcessCallback, + context.CancellationToken ); } public async Task Handle(MarkPendingRetryAsResolved message, IMessageHandlerContext context) { - _ = await lifecycleStore.MarkAsResolved(message.FailedMessageId); + _ = await lifecycleStore.MarkAsResolved(message.FailedMessageId, context.CancellationToken); await domainEvents.Raise(new MessageFailureResolvedManually { diff --git a/src/ServiceControl/MessageFailures/Handlers/UnArchiveMessagesByRangeHandler.cs b/src/ServiceControl/MessageFailures/Handlers/UnArchiveMessagesByRangeHandler.cs index a26b7680ab..564e25abf3 100644 --- a/src/ServiceControl/MessageFailures/Handlers/UnArchiveMessagesByRangeHandler.cs +++ b/src/ServiceControl/MessageFailures/Handlers/UnArchiveMessagesByRangeHandler.cs @@ -14,7 +14,7 @@ class UnArchiveMessagesByRangeHandler(IFailedMessageLifecycleDataStore dataStore { public async Task Handle(UnArchiveMessagesByRange message, IMessageHandlerContext context) { - var ids = await dataStore.UnArchiveMessagesByRange(message.From, message.To); + var ids = await dataStore.UnArchiveMessagesByRange(message.From, message.To, context.CancellationToken); var (user, operationId) = AuditHeaders.Read(context.MessageHeaders); if (!string.IsNullOrEmpty(operationId)) diff --git a/src/ServiceControl/MessageFailures/Handlers/UnArchiveMessagesHandler.cs b/src/ServiceControl/MessageFailures/Handlers/UnArchiveMessagesHandler.cs index f4709d63d0..3a133e86fa 100644 --- a/src/ServiceControl/MessageFailures/Handlers/UnArchiveMessagesHandler.cs +++ b/src/ServiceControl/MessageFailures/Handlers/UnArchiveMessagesHandler.cs @@ -15,7 +15,7 @@ class UnArchiveMessagesHandler(IFailedMessageLifecycleDataStore store, IDomainEv { public async Task Handle(UnArchiveMessages messages, IMessageHandlerContext context) { - var ids = await store.UnArchiveMessages(messages.FailedMessageIds); + var ids = await store.UnArchiveMessages(messages.FailedMessageIds, context.CancellationToken); var (user, operationId) = AuditHeaders.Read(context.MessageHeaders); if (!string.IsNullOrEmpty(operationId)) diff --git a/src/ServiceControl/Recoverability/ExternalIntegration/MessageFailedPublisher.cs b/src/ServiceControl/Recoverability/ExternalIntegration/MessageFailedPublisher.cs index 9076b8e668..38ae30155d 100644 --- a/src/ServiceControl/Recoverability/ExternalIntegration/MessageFailedPublisher.cs +++ b/src/ServiceControl/Recoverability/ExternalIntegration/MessageFailedPublisher.cs @@ -29,7 +29,7 @@ protected override DispatchContext CreateDispatchRequest(MessageFailed @event) protected override async Task> PublishEvents(IEnumerable contexts, CancellationToken cancellationToken = default) { var ids = contexts.Select(x => x.FailedMessageId).ToArray(); - var results = await dataStore.GetFailedMessagesByIds(ids); + var results = await dataStore.GetFailedMessagesByIds(ids, cancellationToken); return results.Select(x => x.ToEvent()); } diff --git a/src/ServiceControl/Recoverability/Retrying/FailedMessageRetryCleaner.cs b/src/ServiceControl/Recoverability/Retrying/FailedMessageRetryCleaner.cs index 4945aedeaa..e30d4adfee 100644 --- a/src/ServiceControl/Recoverability/Retrying/FailedMessageRetryCleaner.cs +++ b/src/ServiceControl/Recoverability/Retrying/FailedMessageRetryCleaner.cs @@ -19,7 +19,7 @@ public Task Handle(MessageFailed message, CancellationToken cancellationToken = { if (message.RepeatedFailure) { - return dataStore.RemoveFailedMessageRetry(message.FailedMessageId); + return dataStore.RemoveFailedMessageRetry(message.FailedMessageId, cancellationToken); } return Task.CompletedTask; diff --git a/src/ServiceControl/Recoverability/Retrying/Handlers/PendingRetriesHandler.cs b/src/ServiceControl/Recoverability/Retrying/Handlers/PendingRetriesHandler.cs index 1190fea600..7c85eca577 100644 --- a/src/ServiceControl/Recoverability/Retrying/Handlers/PendingRetriesHandler.cs +++ b/src/ServiceControl/Recoverability/Retrying/Handlers/PendingRetriesHandler.cs @@ -20,11 +20,11 @@ public async Task Handle(RetryPendingMessages message, IMessageHandlerContext co { var messageIds = new List(); - var ids = await dataStore.GetRetryPendingMessages(message.PeriodFrom, message.PeriodTo, message.QueueAddress); + var ids = await dataStore.GetRetryPendingMessages(message.PeriodFrom, message.PeriodTo, message.QueueAddress, context.CancellationToken); foreach (var id in ids) { - await dataStore.RemoveFailedMessageRetry(id); + await dataStore.RemoveFailedMessageRetry(id, context.CancellationToken); messageIds.Add(id); } @@ -35,7 +35,7 @@ public async Task Handle(RetryPendingMessagesById message, IMessageHandlerContex { foreach (var messageUniqueId in message.MessageUniqueIds) { - await dataStore.RemoveFailedMessageRetry(messageUniqueId); + await dataStore.RemoveFailedMessageRetry(messageUniqueId, context.CancellationToken); } await SendRetryMessagesById(context, message.MessageUniqueIds); diff --git a/src/ServiceControl/Recoverability/Retrying/Handlers/RetriesHandler.cs b/src/ServiceControl/Recoverability/Retrying/Handlers/RetriesHandler.cs index 5b644b5e6d..b9c60dd5aa 100644 --- a/src/ServiceControl/Recoverability/Retrying/Handlers/RetriesHandler.cs +++ b/src/ServiceControl/Recoverability/Retrying/Handlers/RetriesHandler.cs @@ -22,7 +22,7 @@ public Task Handle(MessageFailed message, IMessageHandlerContext context) { if (message.RepeatedFailure) { - return dataStore.RemoveFailedMessageRetry(message.FailedMessageId); + return dataStore.RemoveFailedMessageRetry(message.FailedMessageId, context.CancellationToken); } return Task.CompletedTask; diff --git a/src/ServiceControl/Recoverability/Retrying/Infrastructure/ReturnToSender.cs b/src/ServiceControl/Recoverability/Retrying/Infrastructure/ReturnToSender.cs index 50c4c40d0f..38737f4de1 100644 --- a/src/ServiceControl/Recoverability/Retrying/Infrastructure/ReturnToSender.cs +++ b/src/ServiceControl/Recoverability/Retrying/Infrastructure/ReturnToSender.cs @@ -59,7 +59,7 @@ public virtual async Task HandleMessage(MessageContext message, IMessageDispatch async Task FetchFromFailedMessage(Dictionary outgoingHeaders, string messageId, string attemptMessageId, CancellationToken cancellationToken) { var uniqueMessageId = outgoingHeaders["ServiceControl.Retry.UniqueMessageId"]; - byte[] body = await errorMessageStore.GetFailedMessageBody(uniqueMessageId); + byte[] body = await errorMessageStore.GetFailedMessageBody(uniqueMessageId, cancellationToken); if (body == null) { diff --git a/src/ServiceControl/Recoverability/Retrying/Infrastructure/ReturnToSenderDequeuer.cs b/src/ServiceControl/Recoverability/Retrying/Infrastructure/ReturnToSenderDequeuer.cs index 1ad988bb9b..e2bf531c92 100644 --- a/src/ServiceControl/Recoverability/Retrying/Infrastructure/ReturnToSenderDequeuer.cs +++ b/src/ServiceControl/Recoverability/Retrying/Infrastructure/ReturnToSenderDequeuer.cs @@ -190,18 +190,13 @@ public CaptureIfMessageSendingFails(IFailedMessageLifecycleDataStore dataStore, public async Task OnError(ErrorContext errorContext, CancellationToken cancellationToken = default) { - // We are currently not propagating the cancellation token further since it would require to change - // the data store APIs and domain handlers to take a cancellation token. If this is needed it can be done - // at a later time. - _ = cancellationToken; - try { var destination = errorContext.Headers["ServiceControl.TargetEndpointAddress"]; var messageUniqueId = errorContext.Headers["ServiceControl.Retry.UniqueMessageId"]; logger.LogWarning(errorContext.Exception, "Failed to send '{UniqueMessageId}' message to '{Destination}' for retry. Attempting to revert message status to unresolved so it can be tried again", messageUniqueId, destination); - await dataStore.RevertRetry(messageUniqueId); + await dataStore.RevertRetry(messageUniqueId, cancellationToken); string reason; try