diff --git a/SW.Bitween.Api/Resources/Xchanges/BulkRetry.cs b/SW.Bitween.Api/Resources/Xchanges/BulkRetry.cs index 49065a29..68b1ca52 100644 --- a/SW.Bitween.Api/Resources/Xchanges/BulkRetry.cs +++ b/SW.Bitween.Api/Resources/Xchanges/BulkRetry.cs @@ -1,67 +1,130 @@ -using System.Linq; +using System.Linq; using System.Threading.Tasks; using Microsoft.EntityFrameworkCore; using SW.Bitween.Domain; -using SW.Bitween.Domain.Accounts; using SW.Bitween.Model; using SW.PrimitiveTypes; namespace SW.Bitween.Resources.Xchanges { + /// + /// Retries a selection of exchanges — either a hand-picked list of ids or everything a filter + /// matches. Returns what it did: answers the same question + /// beforehand, so the caller can show it and be sure the two agree. + /// [HandlerName("bulkretry")] public class BulkRetry : ICommandHandler { private readonly BitweenDbContext _dbContext; + private readonly RequestContext _requestContext; private readonly XchangeService _xchangeService; - public BulkRetry(BitweenDbContext dbContext, XchangeService xchangeService) + public BulkRetry(BitweenDbContext dbContext, RequestContext requestContext, + XchangeService xchangeService) { _dbContext = dbContext; + _requestContext = requestContext; _xchangeService = xchangeService; } public async Task Handle(XchangeBulkRetry request) { - var scheduledIds = await _dbContext.Set() - .Where(d => request.Ids.Contains(d.Id)) - .Select(d => d.Id) - .ToListAsync(); + await _requestContext.EnsurePermission(_dbContext, Model.Permissions.Exchanges.Operate); - var xchanges = await _dbContext.Set() - .Where(c => request.Ids.Contains(c.Id) && !scheduledIds.Contains(c.Id)).AsNoTracking() - .ToListAsync(); + var prepared = await new BulkRetryPlanner(_dbContext).Prepare(request); + var plan = prepared.Plan; - foreach (var xchange in xchanges) + if (plan.OverLimit) + throw new SWValidationException("TOO_MANY", + $"{plan.Selected:n0} exchanges is more than the {BulkRetryPlanner.Limit} this can retry " + + "in one go. Narrow the filter and retry the rest after."); + + var retried = 0; + foreach (var id in prepared.Targets) { - var inputFileData = await _xchangeService.GetFile(xchange.Id, XchangeFileType.Input); - var xchangeFile = new XchangeFile(inputFileData, xchange.InputName); + var xchange = await _dbContext.Set().AsNoTracking() + .FirstOrDefaultAsync(x => x.Id == id); + if (xchange == null) + { + plan.Skipped.Add(new XchangeRetrySkip + { + Id = id, + Reason = "This exchange no longer exists." + }); + continue; + } + var subscription = await _dbContext.Subscriptions() .FirstOrDefaultAsync(s => s.Id == xchange.SubscriptionId); - - if (request.Reset) + + if (request.Reset && subscription == null) + { + // Reported per exchange rather than thrown. Throwing meant one exchange whose + // subscription had since been deleted took the whole selection down with it, + // and the caller could not tell which one. + plan.Skipped.Add(new XchangeRetrySkip + { + Id = id, + Reason = "Its properties cannot be reset: the subscription no longer exists." + }); + continue; + } + + // The tolerant read, not GetFile: a retry re-sends the original input, so an + // exchange whose input has been deleted or expired cannot be retried — and one of + // those in a selection of five hundred must not take the other 499 with it. + var xchangeFile = await _xchangeService.ReadInputFile(xchange); + if (xchangeFile == null) + { + plan.Skipped.Add(new XchangeRetrySkip + { + Id = id, + Reason = "Its input document could not be read, so there is nothing to re-send." + }); + continue; + } + + try { - if (subscription == null) - throw new SWValidationException("SUBSCRIPTION_NOT_FOUND", - "Cant reset properties, subscription doesnt exist anymore"); - await _xchangeService.CreateXchange(subscription, xchange, xchangeFile, - manualRetry: true); + if (request.Reset) + { + await _xchangeService.CreateXchange(subscription, xchange, xchangeFile, + manualRetry: true); + } + else + { + // Null when the subscription has since been deleted, which a document-only + // exchange also has from the start. The single-exchange retry has always allowed + // for it; without the same here, one such id in a selection threw and took the + // whole bulk retry down with it. + await _xchangeService.CreateXchange(xchange, xchangeFile, subscription?.WorkGroup, + manualRetry: true); + } } - else + catch (SWValidationException e) when ( + e.Validations.Any(v => v.Key == "ALREADY_RETRIED")) { - - // Null when the subscription has since been deleted, which a document-only - // exchange also has from the start. The single-exchange retry has always allowed - // for it; without the same here, one such id in a selection threw and took the - // whole bulk retry down with it. - await _xchangeService.CreateXchange(xchange, xchangeFile, subscription?.WorkGroup, - manualRetry: true); + // The planner resolves every selection to the end of its chain, so this is not + // reachable by choosing badly — it means someone retried this attempt in the + // moment between the plan being worked out and it being carried out. Reported + // like the other per-exchange refusals rather than thrown, so one racing + // operator cannot cancel another's whole recovery. + plan.Skipped.Add(new XchangeRetrySkip + { + Id = id, + Reason = "It was retried by someone else a moment ago. Its own retry can be retried instead." + }); + continue; } + + retried++; } await _dbContext.SaveChangesAsync(); - return null; + plan.WillRetry = retried; + return plan; } } -} \ No newline at end of file +} diff --git a/SW.Bitween.Api/Resources/Xchanges/BulkRetryPlanner.cs b/SW.Bitween.Api/Resources/Xchanges/BulkRetryPlanner.cs new file mode 100644 index 00000000..7c65bd9c --- /dev/null +++ b/SW.Bitween.Api/Resources/Xchanges/BulkRetryPlanner.cs @@ -0,0 +1,327 @@ +using System.Collections.Generic; +using System.Linq; +using System.Threading.Tasks; +using Microsoft.EntityFrameworkCore; +using SW.EfCoreExtensions; +using SW.Bitween.Domain; +using SW.Bitween.Model; +using SW.PrimitiveTypes; + +namespace SW.Bitween.Resources.Xchanges; + +/// +/// Works out what a bulk retry would do: which exchanges the selection comes to, which of them +/// have already been retried and so hand over to a later attempt, and which cannot be retried at +/// all. +/// +/// +/// Shared by the preview and the retry itself so that what someone confirms is what runs. Built by +/// hand rather than injected, like RetryGroupBudget — it is a piece of one request's work, +/// not a service. +/// +internal sealed class BulkRetryPlanner +{ + /// + /// The largest selection one request will carry out. Every exchange retried means reading its + /// input back out of storage and writing a new exchange, in sequence, inside the one request — + /// so the ceiling is about what can finish before something upstream gives up waiting, not + /// about the database. Past it the caller is asked to narrow the filter. + /// + internal const int Limit = 500; + + /// + /// Guards the walk to the newest attempt against a cycle in the data, the same way + /// does. No honest chain approaches it. + /// + private const int MaxDepth = 100; + + private readonly BitweenDbContext _dbContext; + + internal BulkRetryPlanner(BitweenDbContext dbContext) => _dbContext = dbContext; + + /// The plan to show or carry out, and the exchanges it would actually retry. + internal sealed class Prepared + { + internal XchangeBulkRetryPlan Plan { get; init; } + internal List Targets { get; init; } = new List(); + } + + internal async Task Prepare(XchangeBulkRetry request) + { + var selected = await ResolveSelection(request); + + if (selected.Count > Limit) + return new Prepared + { + Plan = new XchangeBulkRetryPlan + { + // A hand-picked list is already counted. A filter's selection was cut off at + // Limit + 1 to notice it was too long without reading it all, so that one is + // counted properly now — the caller is told how far past the line it is rather + // than just "501", and never a count of rows it did not ask about. + Selected = string.IsNullOrWhiteSpace(request.Filter) + ? selected.Count + : await CountSelection(request), + Limit = Limit, + OverLimit = true + } + }; + + var newestAttempt = await FindNewestAttempts(selected); + var state = await ReadState(newestAttempt.Values.Distinct().ToList()); + var reset = request.Reset; + + var plan = new XchangeBulkRetryPlan { Selected = selected.Count, Limit = Limit }; + var targets = new List(); + + foreach (var selectedId in selected) + { + var targetId = newestAttempt[selectedId]; + + if (targetId != selectedId) + plan.Substituted.Add(new XchangeRetrySubstitution + { + SelectedId = selectedId, + RetryId = targetId + }); + + var reason = WhyNot(state.GetValueOrDefault(targetId), targetId != selectedId, reset); + if (reason != null) + { + plan.Skipped.Add(new XchangeRetrySkip { Id = targetId, Reason = reason }); + continue; + } + + // Two selections in the same chain — an exchange and its own retry, say — come to the + // same attempt, which is retried once. + if (!targets.Contains(targetId)) + targets.Add(targetId); + } + + plan.WillRetry = targets.Count; + plan.Properties = await ReadPromotedProperties(plan); + return new Prepared { Plan = plan, Targets = targets }; + } + + /// + /// Why this attempt will be left alone, or null when it will be retried. The wording + /// says whose fault it is: the exchange the caller picked, or the later attempt standing in + /// for it. + /// + private static string WhyNot(SelectionState state, bool substituted, bool reset) + { + var subject = substituted ? "Its newest attempt" : "It"; + + if (state == null) + return "This exchange no longer exists."; + + // Only when the caller asked for the subscription's current configuration, which there is + // no way to read once the subscription is gone. Checked here as well as at the moment of + // retrying, because a plan that promises to retry something the retry then skips makes the + // confirmation worthless — the whole point of showing it is that it is what will happen. + if (reset && state.SubscriptionMissing) + return $"{subject} cannot have its properties re-resolved: the subscription no longer exists."; + + if (state.ScheduledRetryOn != null) + return $"{subject} already has an auto-retry scheduled. Run that now instead."; + + // An exchange with no result is deliberately not skipped as "still running". That is what + // a broker outage leaves behind — hundreds of exchanges that were never processed and never + // will be — and retrying those in bulk is the main thing a wide selection is for. Refusing + // them would take the recovery path away to protect against double-running work that, in + // the case anyone actually selects in bulk, is not running at all. + + if (state.Status == true && state.ResponseBad != true) + return $"{subject} succeeded, so there is nothing to retry."; + + return null; + } + + /// + /// The end of each selection's chain: for an exchange already retried, the attempt a retry + /// would actually run, since the selected one is refused a second retry. + /// + /// + /// One query per round for the whole selection at once, rather than a walk per exchange — + /// five hundred selections down a chain of four is four queries, not two thousand. + /// + private async Task> FindNewestAttempts(List selected) + { + var newest = selected.ToDictionary(id => id, id => id); + + for (var depth = 0; depth < MaxDepth; depth++) + { + // Where each selection has got to so far. Asking about that frontier — rather than + // about one level of the tree, tracking which nodes the walk had already seen — is what + // keeps selections that sit in the same chain consistent with each other. With a seen + // set, a selection lagging behind another stopped dead on a node the leader had already + // stepped over, and was reported as "already retried, its newest attempt runs instead" + // naming an attempt that had itself been retried. Retrying that then threw + // ALREADY_RETRIED and took the whole bulk retry with it. + var frontier = newest.Values.Distinct().ToList(); + + var children = await _dbContext.Set().AsNoTracking() + .Where(x => frontier.Contains(x.RetryFor)) + .Select(x => new { x.Id, x.RetryFor, x.StartedOn }) + .ToListAsync(); + + if (children.Count == 0) break; + + // One retry per exchange is the rule, but exchanges retried before it was enforced can + // have more than one. Follow the newest, and let the retry tree in the drawer be where + // the fork is shown — a bulk retry has to choose something, and the newest attempt is + // the one that reflects where the work actually got to. + var step = children + .GroupBy(c => c.RetryFor) + .ToDictionary(g => g.Key, g => g.OrderByDescending(c => c.StartedOn).First().Id); + + var advanced = false; + foreach (var key in newest.Keys.ToList()) + if (step.TryGetValue(newest[key], out var child) && child != newest[key]) + { + newest[key] = child; + advanced = true; + } + + // Nothing moved, so every selection is at the end of its chain. This is also the way + // out of a cycle in the data, which would otherwise keep advancing until MaxDepth. + if (!advanced) break; + } + + return newest; + } + + /// + /// What the payload promoted, for each exchange the plan names on either side of a + /// substitution or in a skip — so the caller can describe them the way the exchange list does + /// rather than by id alone. + /// + private async Task>> ReadPromotedProperties( + XchangeBulkRetryPlan plan) + { + var named = plan.Substituted.Select(s => s.SelectedId) + .Concat(plan.Substituted.Select(s => s.RetryId)) + .Concat(plan.Skipped.Select(s => s.Id)) + .Distinct() + .ToList(); + + if (named.Count == 0) + return new Dictionary>(); + + var rows = await _dbContext.Set().AsNoTracking() + .Where(p => named.Contains(p.Id)) + .ToListAsync(); + + return rows.ToDictionary(r => r.Id, r => (IDictionary)r.Properties.ToDictionary()); + } + + private sealed class SelectionState + { + public string Id { get; set; } + public bool? Status { get; set; } + public bool? ResponseBad { get; set; } + public System.DateTime? ScheduledRetryOn { get; set; } + + /// + /// Covers an exchange that never had a subscription — a document-only one — as well as one + /// whose subscription has since been deleted. Both leave a reset with nothing to read. + /// + public bool SubscriptionMissing { get; set; } + } + + private async Task> ReadState(List ids) + { + var rows = await ( + from xchange in _dbContext.Set() + join result in _dbContext.Set() on xchange.Id equals result.Id into xr + from result in xr.DefaultIfEmpty() + join delayed in _dbContext.Set() on xchange.Id equals delayed.Id into dr + from delayed in dr.DefaultIfEmpty() + where ids.Contains(xchange.Id) + select new SelectionState + { + Id = xchange.Id, + Status = result.Success, + ResponseBad = result.ResponseBad, + ScheduledRetryOn = delayed != null ? delayed.On : (System.DateTime?)null, + SubscriptionMissing = xchange.SubscriptionId == null || + !_dbContext.Set().Any(sub => sub.Id == xchange.SubscriptionId) + }).AsNoTracking().ToListAsync(); + + return rows.ToDictionary(r => r.Id); + } + + private async Task> ResolveSelection(XchangeBulkRetry request) + { + if (string.IsNullOrWhiteSpace(request.Filter)) + return (request.Ids ?? new List()).Where(id => id != null).Distinct().ToList(); + + var exclude = request.ExcludeIds ?? new List(); + + // One past the limit is all it takes to know the selection is too big, and stops a + // "select all" over a wide filter from reading a million ids to refuse them. + return await SelectionQuery(request) + .Where(r => !exclude.Contains(r.Id)) + .Select(r => r.Id) + .Take(Limit + 1) + .ToListAsync(); + } + + private async Task CountSelection(XchangeBulkRetry request) + { + var exclude = request.ExcludeIds ?? new List(); + return await SelectionQuery(request) + .Where(r => !exclude.Contains(r.Id)) + // Same reason the search caps its own count: counting every match has to visit every + // matching row. The number is only being used to say "too many", so stopping early + // costs the caller nothing. + .Take(Search.CountCap + 1) + .CountAsync(); + } + + /// + /// The exchanges a "select all matching" came from, filtered exactly as the search would have + /// filtered them. + /// + /// + /// The projection carries the columns the exchange list can filter on (see + /// buildExchangeQuery in the client) and nothing else, so this stays translatable and + /// composable — the search's own projection cannot be reused for that, as it builds file URLs + /// in C#. Filtering on a column that is not here would silently match nothing, so a new filter + /// on the list needs a column here too. + /// + private IQueryable SelectionQuery(XchangeBulkRetry request) + { + var searchyRequest = new SearchyRequest(request.Filter); + searchyRequest.DatesToUtc(); + + var query = from xchange in _dbContext.Set() + join result in _dbContext.Set() on xchange.Id equals result.Id into xr + from result in xr.DefaultIfEmpty() + join agg in _dbContext.Set() on xchange.Id equals agg.Id into xa + from agg in xa.DefaultIfEmpty() + join promoted in _dbContext.Set() on xchange.Id equals promoted.Id into xp + from promoted in xp.DefaultIfEmpty() + join subscriber in _dbContext.Set() on xchange.SubscriptionId equals subscriber.Id into xs + from subscriber in xs.DefaultIfEmpty() + select new XchangeRow + { + Id = xchange.Id, + SubscriptionId = xchange.SubscriptionId, + DocumentId = xchange.DocumentId, + StartedOn = xchange.StartedOn, + CorrelationId = xchange.CorrelationId, + Status = result.Success, + ResponseBad = result.ResponseBad, + RetryFor = xchange.RetryFor, + AggregationXchangeId = agg.AggregationXchangeId, + PromotedPropertiesRaw = promoted.PropertiesRaw, + // Same fallback the search makes for exchanges written before the column + // existed, so a partner filter selects the same rows it listed. + PartnerId = xchange.PartnerId ?? subscriber.PartnerId + }; + + query = query.ApplySpecialFilters(searchyRequest, _dbContext); + return query.AsNoTracking().Search(searchyRequest.Conditions); + } +} diff --git a/SW.Bitween.Api/Resources/Xchanges/BulkRetryPreview.cs b/SW.Bitween.Api/Resources/Xchanges/BulkRetryPreview.cs new file mode 100644 index 00000000..dc77a50f --- /dev/null +++ b/SW.Bitween.Api/Resources/Xchanges/BulkRetryPreview.cs @@ -0,0 +1,37 @@ +using System.Threading.Tasks; +using SW.Bitween.Model; +using SW.PrimitiveTypes; + +namespace SW.Bitween.Resources.Xchanges; + +/// +/// What would do with the same request, without doing any of it. +/// +/// +/// A selection is rarely just itself: exchanges already retried hand over to their newest attempt, +/// ones that have since succeeded or are still running drop out, and two selections in one chain +/// come to the same attempt. Retrying is not undoable, so the caller gets to show all of that and +/// be told the number before anyone commits to it. +/// +[HandlerName("bulkretrypreview")] +public class BulkRetryPreview : ICommandHandler +{ + private readonly BitweenDbContext _dbContext; + private readonly RequestContext _requestContext; + + public BulkRetryPreview(BitweenDbContext dbContext, RequestContext requestContext) + { + _dbContext = dbContext; + _requestContext = requestContext; + } + + public async Task Handle(XchangeBulkRetry request) + { + // Reads nothing a retryer could not already see, but it is the retry dialog's own call and + // describes an action only they can take, so it is gated with the action. + await _requestContext.EnsurePermission(_dbContext, Model.Permissions.Exchanges.Operate); + + var prepared = await new BulkRetryPlanner(_dbContext).Prepare(request); + return prepared.Plan; + } +} diff --git a/SW.Bitween.Api/Resources/Xchanges/Retry.cs b/SW.Bitween.Api/Resources/Xchanges/Retry.cs index 3c76cd8f..5e37c016 100644 --- a/SW.Bitween.Api/Resources/Xchanges/Retry.cs +++ b/SW.Bitween.Api/Resources/Xchanges/Retry.cs @@ -10,16 +10,21 @@ namespace SW.Bitween.Resources.Xchanges public class Retry : ICommandHandler { private readonly BitweenDbContext dbContext; + private readonly RequestContext requestContext; private readonly XchangeService xchangeService; - public Retry(BitweenDbContext dbContext, XchangeService xchangeService) + public Retry(BitweenDbContext dbContext, RequestContext requestContext, + XchangeService xchangeService) { this.dbContext = dbContext; + this.requestContext = requestContext; this.xchangeService = xchangeService; } public async Task Handle(string key, XchangeRetry xchangeRetry) { + await requestContext.EnsurePermission(dbContext, Model.Permissions.Exchanges.Operate); + if (await dbContext.Set().AnyAsync(d => d.Id == key)) throw new SWValidationException("AUTO_RETRY_SCHEDULED", "An auto-retry is already scheduled for this exchange. Use \"Run Now\" to execute it immediately instead of retrying manually."); diff --git a/SW.Bitween.Api/Resources/Xchanges/RetryTree.cs b/SW.Bitween.Api/Resources/Xchanges/RetryTree.cs new file mode 100644 index 00000000..329f6cf2 --- /dev/null +++ b/SW.Bitween.Api/Resources/Xchanges/RetryTree.cs @@ -0,0 +1,150 @@ +using System.Collections.Generic; +using System.Linq; +using System.Threading.Tasks; +using Microsoft.EntityFrameworkCore; +using SW.Bitween.Domain; +using SW.Bitween.Model; +using SW.PrimitiveTypes; + +namespace SW.Bitween.Resources.Xchanges; + +/// +/// Every attempt in one exchange's retry chain: back to the original, and forward through each +/// retry made from it. +/// +/// +/// +/// Its own endpoint rather than part of the exchange search, because a chain cannot be had from the +/// same query that lists exchanges — it takes a walk of unknown length, and the search is already +/// the expensive one. Callers avoid asking altogether for the common case: and come back with every row, and +/// an exchange with neither has no chain to fetch. +/// +/// +/// Walked iteratively, a level at a time, rather than as a recursive query — the three supported +/// databases would each need their own SQL for that, to save queries on a chain that a retry budget +/// keeps to single digits in practice. Both walks are indexed lookups on RetryFor. +/// +/// +[HandlerName("retrytree")] +public class RetryTree : IQueryHandler +{ + /// + /// How far the walk goes in each direction before giving up and saying so. Far past any real + /// chain — a retry policy's budget is the practical limit — so this exists to keep a cycle in + /// the data from becoming an endless loop, not to shorten honest answers. + /// + private const int MaxDepth = 100; + + /// + /// How many attempts one answer carries. Depth alone does not bound the work: an exchange + /// retried many times over before the one-retry rule existed has all of those as direct + /// children, and every one would be loaded and serialized. Far above any real chain, so + /// reaching it means the data is unusual — and the answer then says it is partial rather than + /// quietly costing the database more the older the exchange is. + /// + private const int MaxNodes = 500; + + private readonly BitweenDbContext _dbContext; + private readonly RequestContext _requestContext; + + public RetryTree(BitweenDbContext dbContext, RequestContext requestContext) + { + _dbContext = dbContext; + _requestContext = requestContext; + } + + public async Task Handle(XchangeRetryTreeRequest request) + { + await _requestContext.EnsurePermission(_dbContext, Model.Permissions.Exchanges.View); + + if (string.IsNullOrWhiteSpace(request?.Id)) + throw new SWValidationException("ID_REQUIRED", "Which exchange's retries?"); + + var truncated = false; + + // Up to the original. Each step reads one pointer, so the exchange asked about does not + // have to be the newest attempt — opening any of them shows the same chain. + var rootId = request.Id; + var ancestors = new HashSet { rootId }; + for (var depth = 0; depth < MaxDepth; depth++) + { + var parent = await _dbContext.Set().AsNoTracking() + .Where(x => x.Id == rootId) + .Select(x => x.RetryFor) + .FirstOrDefaultAsync(); + + // Also covers the exchange not existing at all: no row, so no parent, and the walk + // down then finds nothing either — an empty tree rather than a 404, because a caller + // asking about an exchange it is looking at wants the chain, not another error to + // handle. + if (parent == null || !ancestors.Add(parent)) + break; + + rootId = parent; + if (depth == MaxDepth - 1) truncated = true; + } + + // Then down from the original, taking a whole level per query. From the root rather than + // from the exchange asked about, so an exchange that forked before one-retry-per-exchange + // was enforced shows both of its branches instead of only the one that leads here. + var ids = new List { rootId }; + // Its own visited set: the walk down passes back through the ancestors the walk up just + // collected, so sharing one would drop the whole stretch between the original and the + // exchange asked about. + var visited = new HashSet { rootId }; + var level = new List { rootId }; + for (var depth = 0; depth < MaxDepth && level.Count > 0; depth++) + { + var children = await _dbContext.Set().AsNoTracking() + .Where(x => level.Contains(x.RetryFor)) + .Select(x => x.Id) + .ToListAsync(); + + level = children.Where(visited.Add).ToList(); + + if (ids.Count + level.Count > MaxNodes) + { + ids.AddRange(level.Take(MaxNodes - ids.Count)); + truncated = true; + break; + } + + ids.AddRange(level); + + if (depth == MaxDepth - 1 && level.Count > 0) truncated = true; + } + + var nodes = await ( + from xchange in _dbContext.Set() + join result in _dbContext.Set() on xchange.Id equals result.Id into xr + from result in xr.DefaultIfEmpty() + join delayed in _dbContext.Set() on xchange.Id equals delayed.Id into dr + from delayed in dr.DefaultIfEmpty() + join promoted in _dbContext.Set() on xchange.Id equals promoted.Id into xp + from promoted in xp.DefaultIfEmpty() + where ids.Contains(xchange.Id) + orderby xchange.StartedOn + select new XchangeRetryNode + { + Id = xchange.Id, + RetryFor = xchange.RetryFor, + StartedOn = xchange.StartedOn, + FinishedOn = result.FinishedOn, + Status = result.Success, + ResponseBad = result.ResponseBad, + Exception = result.Exception, + ManualRetry = xchange.ManualRetry, + ScheduledRetryOn = delayed != null ? delayed.On : (System.DateTime?)null, + RetryBlockedReason = result.RetryBlockedReason, + PromotedProperties = promoted == null ? null : promoted.Properties.ToDictionary() + }).AsNoTracking().ToListAsync(); + + return new XchangeRetryTree + { + RootId = rootId, + Nodes = nodes, + Truncated = truncated + }; + } +} diff --git a/SW.Bitween.Api/Resources/Xchanges/Search.cs b/SW.Bitween.Api/Resources/Xchanges/Search.cs index a67af1e9..c12c5ebd 100644 --- a/SW.Bitween.Api/Resources/Xchanges/Search.cs +++ b/SW.Bitween.Api/Resources/Xchanges/Search.cs @@ -96,91 +96,30 @@ from delayedRetry in drGroup.DefaultIfEmpty() RetryBlockedReason = result.RetryBlockedReason }; - var condition = searchyRequest.Conditions.FirstOrDefault(); - if (condition != null) - { - var idFilters = condition.Filters.Where(f => f.Field == "Id").ToList(); - foreach (var idFilter in idFilters) - { - var value = idFilter.Value.ToString(); - switch (idFilter.Rule) - { - case SearchyRule.EqualsTo: - query = query.Where(i => - i.Id == value || i.RetryFor == value || i.AggregationXchangeId == value); - break; - case SearchyRule.Contains: - { - var valueAsArray = idFilter.ValueStringArray; - query = query.Where(i => - valueAsArray.Any(v => i.RetryFor == v) || - valueAsArray.Any(v => i.AggregationXchangeId == v) || - valueAsArray.Any(v => i.Id == v) - ); - break; - } - - - default: - throw new SWValidationException("NOT_SUPPORTED", "Search query not supported"); - } - - condition.Filters.Remove(idFilter); - } - - var statusFilters = condition.Filters.Where(f => f.Field == "StatusFilter").ToList(); - foreach (var statusFilter in statusFilters) - { - switch (statusFilter.Value) - { - case "0": - // "Still running" means no result row exists yet. Asking for it as - // Status == null reads as `x0.success IS NULL` on the left join, and - // Postgres cannot estimate that: it guesses one row, plans every join - // above it for one row, and picks per-row sequential scans of the small - // side tables. Measured on 1M exchanges that was 22.8s for 25 rows. - // NOT EXISTS asks the same question as an anti-join, which it can - // estimate — 34ms. Equivalent because success is NOT NULL, so a result - // row can never itself carry a null status. - query = query.Where(i => - !dbContext.Set().Any(r => r.Id == i.Id)); - break; - case "1": - query = query.Where(i => i.Status == true && i.ResponseBad != true); - break; - - case "2": - query = query.Where(i => i.Status == true && i.ResponseBad == true); - break; - - case "3": - query = query.Where(i => i.Status == false); - break; - } - - condition.Filters.Remove(statusFilter); - } - - var propertiesFilters = condition.Filters - .Where(f => f.Field == "PromotedPropertiesRaw").ToList(); - foreach (var propertyFilter in propertiesFilters) - { - var value = propertyFilter.Value.ToString()!.ToLower(); - - // Both sides lower-cased at query time. Promoted values keep the case the - // payload had (see FilterService), so the column has to be folded here for - // the search to stay case-insensitive. No index is lost: a Contains is a - // leading-wildcard LIKE, which the b-tree on this column could never serve. - query = query.Where(i => i.PromotedPropertiesRaw.ToLower().Contains(value)); - condition.Filters.Remove(propertyFilter); - } - } + query = query.ApplySpecialFilters(searchyRequest, dbContext); var s = query.OrderByDescending(p => p.StartedOn).AsNoTracking().Search(searchyRequest.Conditions, searchyRequest.Sorts, searchyRequest.PageSize, searchyRequest.PageIndex); var r = await s.ToListAsync(); + // Which of these have already been retried, so the client can tell a spent exchange + // from a retryable one without asking about each row's chain. Asked separately rather + // than as a subquery in the projection above, because TotalCount reuses that query and + // an exact count already costs more than fetching the rows does — this way neither + // plan changes. One indexed lookup over RetryFor for the page's worth of ids. + if (r.Count > 0) + { + var pageIds = r.Select(row => row.Id).ToList(); + var retriedIds = await dbContext.Set().AsNoTracking() + .Where(x => pageIds.Contains(x.RetryFor)) + .Select(x => x.RetryFor) + .Distinct() + .ToListAsync(); + foreach (var row in r) + row.HasRetry = retriedIds.Contains(row.Id); + } + var searchyResponse = new SearchyResponse { Result = r, diff --git a/SW.Bitween.Api/Resources/Xchanges/XchangeFilters.cs b/SW.Bitween.Api/Resources/Xchanges/XchangeFilters.cs new file mode 100644 index 00000000..395e3639 --- /dev/null +++ b/SW.Bitween.Api/Resources/Xchanges/XchangeFilters.cs @@ -0,0 +1,119 @@ +using System.Linq; +using Microsoft.EntityFrameworkCore; +using SW.Bitween.Domain; +using SW.Bitween.Model; +using SW.PrimitiveTypes; + +namespace SW.Bitween.Resources.Xchanges; + +/// +/// The exchange filters that cannot be handed to Searchy as-is, because they do not correspond to +/// one column: an id that should also match relatives, a status assembled from two columns and the +/// absence of a result row, a promoted-property substring that has to be case-folded. +/// +/// +/// Shared by the search and by bulk retry, which needs "every exchange this filter matches" to mean +/// exactly the set the person was looking at when they chose it. The two drifting apart would be +/// invisible until a bulk retry quietly acted on a different set of exchanges than the one on +/// screen. +/// +internal static class XchangeFilters +{ + /// + /// Applies the special filters and removes them from , leaving + /// the plain per-column ones for Searchy to handle. + /// + internal static IQueryable ApplySpecialFilters(this IQueryable query, + SearchyRequest searchyRequest, BitweenDbContext dbContext) + { + var condition = searchyRequest.Conditions.FirstOrDefault(); + if (condition == null) + return query; + + var idFilters = condition.Filters.Where(f => f.Field == "Id").ToList(); + foreach (var idFilter in idFilters) + { + var value = idFilter.Value.ToString(); + switch (idFilter.Rule) + { + case SearchyRule.EqualsTo: + query = query.Where(i => + i.Id == value || i.RetryFor == value || i.AggregationXchangeId == value); + break; + case SearchyRule.Contains: + { + var valueAsArray = idFilter.ValueStringArray; + query = query.Where(i => + valueAsArray.Any(v => i.RetryFor == v) || + valueAsArray.Any(v => i.AggregationXchangeId == v) || + valueAsArray.Any(v => i.Id == v) + ); + break; + } + + + default: + throw new SWValidationException("NOT_SUPPORTED", "Search query not supported"); + } + + condition.Filters.Remove(idFilter); + } + + var statusFilters = condition.Filters.Where(f => f.Field == "StatusFilter").ToList(); + foreach (var statusFilter in statusFilters) + { + switch (statusFilter.Value) + { + case "0": + // "Still running" means no result row exists yet. Asking for it as + // Status == null reads as `x0.success IS NULL` on the left join, and + // Postgres cannot estimate that: it guesses one row, plans every join + // above it for one row, and picks per-row sequential scans of the small + // side tables. Measured on 1M exchanges that was 22.8s for 25 rows. + // NOT EXISTS asks the same question as an anti-join, which it can + // estimate — 34ms. Equivalent because success is NOT NULL, so a result + // row can never itself carry a null status. + query = query.Where(i => + !dbContext.Set().Any(r => r.Id == i.Id)); + break; + case "1": + query = query.Where(i => i.Status == true && i.ResponseBad != true); + break; + + case "2": + query = query.Where(i => i.Status == true && i.ResponseBad == true); + break; + + case "3": + query = query.Where(i => i.Status == false); + break; + + default: + // The filter is removed below whether or not it matched, so falling through + // here used to drop it silently and widen the selection to everything. A + // search returning too much is merely wrong; bulk retry runs over whatever + // this selects, so an unreadable status has to be refused rather than ignored. + throw new SWValidationException("NOT_SUPPORTED", + $"'{statusFilter.Value}' is not an exchange status."); + } + + condition.Filters.Remove(statusFilter); + } + + var propertiesFilters = condition.Filters + .Where(f => f.Field == "PromotedPropertiesRaw").ToList(); + foreach (var propertyFilter in propertiesFilters) + { + var value = propertyFilter.Value.ToString()!.ToLower(); + + // Both sides lower-cased at query time. Promoted values keep the case the + // payload had (see FilterService), so the column has to be folded here for + // the search to stay case-insensitive. No index is lost: a Contains is a + // leading-wildcard LIKE, which the b-tree on this column could never serve. + query = query.Where(i => i.PromotedPropertiesRaw.ToLower().Contains(value)); + condition.Filters.Remove(propertyFilter); + } + + return query; + } +} diff --git a/SW.Bitween.Api/Services/XchangeService.cs b/SW.Bitween.Api/Services/XchangeService.cs index 91cb05ff..4630fc0c 100644 --- a/SW.Bitween.Api/Services/XchangeService.cs +++ b/SW.Bitween.Api/Services/XchangeService.cs @@ -90,6 +90,7 @@ public async Task SubmitFilterXchange(int documentId, XchangeFile file, string[] public async Task CreateXchange(Xchange xchange, XchangeFile file, WorkGroup workGroup, bool manualRetry = false) { + await EnsureNotAlreadyRetried(xchange.Id); var newXchange = new Xchange(xchange, file, workGroup, manualRetry); await AddFile(newXchange.Id, XchangeFileType.Input, file); _dbContext.Add(newXchange); @@ -98,6 +99,7 @@ public async Task CreateXchange(Xchange xchange, XchangeFile file, WorkGroup wor public async Task CreateXchange(Subscription subscription, Xchange xchange, XchangeFile file, string[] references = null, Dictionary groupAttemptCounts = null, bool manualRetry = false) { + await EnsureNotAlreadyRetried(xchange.Id); var partnerId = xchange.PartnerId ?? subscription.PartnerId; var partner = partnerId.HasValue ? await _dbContext.FindAsync(partnerId.Value) : null; var globalAdapterValuesSets = await _BitweenCache.ListGlobalAdapterValuesSetsAsync(); @@ -176,6 +178,21 @@ public async Task ExecuteDelayedRetry(DelayedRetry delayedRetry) return false; } + var alreadyRetried = await FindRetryOf(xchange.Id); + if (alreadyRetried != null) + { + // Reached only if a manual retry got in first — the endpoint refuses that while a + // retry is scheduled, so it takes a race to arrive here. Dropped like the cases below + // rather than left to throw: an exception here would leave the schedule in place and + // the job would pick the same impossible retry up again on every pass, forever. + _dbContext.Remove(delayedRetry); + + var retried = await _dbContext.FindAsync(xchange.Id); + retried?.SetRetryBlocked( + $"The scheduled retry was dropped: this exchange had already been retried, as {alreadyRetried}."); + return false; + } + var inputFile = await ReadInputFile(xchange); if (inputFile == null) { @@ -198,7 +215,7 @@ public async Task ExecuteDelayedRetry(DelayedRetry delayedRetry) /// The original input, or null when it cannot be read — deleted from storage, expired by a /// lifecycle rule, or storage itself unavailable. /// - private async Task ReadInputFile(Xchange xchange) + public async Task ReadInputFile(Xchange xchange) { try { @@ -652,6 +669,38 @@ private async Task TryScheduleAutoRetry(Xchange xchange, XchangeResultType resul decision.MatchedGroup.Budget!.MaxAttemptsTotal); } + /// + /// The retry, if any, already made from . + /// + private Task FindRetryOf(string xchangeId) => + _dbContext.Set().AsNoTracking() + .Where(x => x.RetryFor == xchangeId) + .Select(x => x.Id) + .FirstOrDefaultAsync(); + + /// + /// An exchange gets at most one retry, so that the attempts made from one original form a + /// single chain that can be read end to end. Retrying an exchange that already has one would + /// fork it: two attempts from the same starting point, neither of them the current state of + /// anything, and no way to say which one "the retry" of the original was. + /// + /// + /// Enforced here rather than at each endpoint so that every way of asking for a retry — by + /// hand, in bulk, or by a retry policy coming due — is held to it. Not enforced by a unique + /// index as well: exchanges retried before this rule existed can already have forked, and an + /// index that will not create over the data it inherits is worse than no index. Two retries of + /// the same exchange committed at the very same moment can therefore still both pass this + /// check; the loser is a duplicate attempt, which the tree then shows as a fork. + /// + private async Task EnsureNotAlreadyRetried(string xchangeId) + { + var existing = await FindRetryOf(xchangeId); + if (existing != null) + throw new SWValidationException("ALREADY_RETRIED", + $"This exchange has already been retried, as exchange {existing}. Retry that attempt " + + "instead — an exchange is only retried once, so that its attempts stay a single chain."); + } + private async Task CountRetryChainDepth(Xchange xchange) { var depth = 0; diff --git a/SW.Bitween.IntegrationTests/Tests/DelayedRetriesTests.cs b/SW.Bitween.IntegrationTests/Tests/DelayedRetriesTests.cs index 9745d12e..8a63f2df 100644 --- a/SW.Bitween.IntegrationTests/Tests/DelayedRetriesTests.cs +++ b/SW.Bitween.IntegrationTests/Tests/DelayedRetriesTests.cs @@ -1,4 +1,4 @@ -using System; +using System; using System.Collections.Generic; using System.Linq; using System.Threading.Tasks; @@ -59,7 +59,7 @@ public async Task Retry_throws_when_auto_retry_already_scheduled() db.Set().Add(new DelayedRetry { Id = xchange.Id, On = DateTime.UtcNow.AddMinutes(5) }); await db.SaveChangesAsync(); - var retry = new SW.Bitween.Resources.Xchanges.Retry(db, xs); + var retry = new SW.Bitween.Resources.Xchanges.Retry(db, scope.Superuser(), xs); await Assert.ThrowsAsync(() => retry.Handle(xchange.Id, new XchangeRetry { Reset = false })); @@ -73,7 +73,7 @@ public async Task Retry_succeeds_when_no_auto_retry_scheduled() var xs = scope.ServiceProvider.GetRequiredService(); var (_, _, xchange) = await CreateSubscriptionWithXchange(db, xs, "Retry OK Doc"); - var retry = new SW.Bitween.Resources.Xchanges.Retry(db, xs); + var retry = new SW.Bitween.Resources.Xchanges.Retry(db, scope.Superuser(), xs); await retry.Handle(xchange.Id, new XchangeRetry { Reset = false }); var retryXchange = await db.Set().FirstOrDefaultAsync(x => x.RetryFor == xchange.Id); @@ -93,7 +93,7 @@ public async Task BulkRetry_skips_ids_with_scheduled_auto_retry_and_processes_ot db.Set().Add(new DelayedRetry { Id = xchangeScheduled.Id, On = DateTime.UtcNow.AddMinutes(5) }); await db.SaveChangesAsync(); - var bulkRetry = new SW.Bitween.Resources.Xchanges.BulkRetry(db, xs); + var bulkRetry = new SW.Bitween.Resources.Xchanges.BulkRetry(db, scope.Superuser(), xs); await bulkRetry.Handle(new XchangeBulkRetry { Reset = false, diff --git a/SW.Bitween.IntegrationTests/Tests/RetryChainTests.cs b/SW.Bitween.IntegrationTests/Tests/RetryChainTests.cs new file mode 100644 index 00000000..ba9ddd23 --- /dev/null +++ b/SW.Bitween.IntegrationTests/Tests/RetryChainTests.cs @@ -0,0 +1,431 @@ +using System; +using System.Linq; +using System.Threading.Tasks; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; +using SW.Bitween.Domain; +using SW.Bitween.IntegrationTests.Fixtures; +using SW.Bitween.Model; +using SW.PrimitiveTypes; +using Xunit; + +namespace SW.Bitween.IntegrationTests.Tests; + +/// +/// One exchange, at most one retry: the attempts made from an original form a chain that can be +/// read end to end, and retrying anything but its newest attempt is refused. +/// +[Collection("Bitween")] +public class RetryChainTests +{ + private readonly BitweenFixture _fixture; + + public RetryChainTests(BitweenFixture fixture) + { + _fixture = fixture; + } + + private static async Task<(Subscription sub, Xchange xchange)> FailedXchange( + BitweenDbContext db, XchangeService xs, string name) + { + var doc = new Document(null, name, DocumentFormat.Json); + db.Set().Add(doc); + await db.SaveChangesAsync(); + + var sub = new Subscription(name, doc.Id) { Inactive = false }; + db.Set().Add(sub); + await db.SaveChangesAsync(); + + var xchange = await xs.CreateXchange(sub, new XchangeFile("{}")); + await db.SaveChangesAsync(); + + db.Set().Add(new XchangeResult(xchange.Id, null, null, exception: "boom")); + await db.SaveChangesAsync(); + + return (sub, xchange); + } + + /// The retry made from , failed so it can be retried in turn. + private static async Task RetryOnce(BitweenDbContext db, XchangeService xs, + RequestContext ctx, string id) + { + await new Resources.Xchanges.Retry(db, ctx, xs).Handle(id, new XchangeRetry { Reset = false }); + await db.SaveChangesAsync(); + + var child = await db.Set().FirstAsync(x => x.RetryFor == id); + db.Set().Add(new XchangeResult(child.Id, null, null, exception: "boom again")); + await db.SaveChangesAsync(); + return child; + } + + // ─── The invariant ──────────────────────────────────────────────────────── + + [Fact] + public async Task Retry_refuses_an_exchange_that_has_already_been_retried() + { + await using var scope = _fixture.CreateScope(); + var db = scope.ServiceProvider.GetRequiredService(); + var xs = scope.ServiceProvider.GetRequiredService(); + var ctx = scope.Superuser(); + + var (_, xchange) = await FailedXchange(db, xs, "Chain Refuse Doc"); + var second = await RetryOnce(db, xs, ctx, xchange.Id); + + var error = await Assert.ThrowsAsync(() => + new Resources.Xchanges.Retry(db, ctx, xs).Handle(xchange.Id, new XchangeRetry())); + + Assert.Contains(second.Id, error.Message); + + // Still exactly one retry from it — the point of the rule. + Assert.Equal(1, await db.Set().CountAsync(x => x.RetryFor == xchange.Id)); + } + + [Fact] + public async Task Retry_allows_the_newest_attempt_in_a_chain() + { + await using var scope = _fixture.CreateScope(); + var db = scope.ServiceProvider.GetRequiredService(); + var xs = scope.ServiceProvider.GetRequiredService(); + var ctx = scope.Superuser(); + + var (_, xchange) = await FailedXchange(db, xs, "Chain Deepen Doc"); + var second = await RetryOnce(db, xs, ctx, xchange.Id); + var third = await RetryOnce(db, xs, ctx, second.Id); + + Assert.Equal(second.Id, third.RetryFor); + } + + [Fact] + public async Task A_scheduled_retry_of_an_already_retried_exchange_is_dropped_with_a_reason() + { + await using var scope = _fixture.CreateScope(); + var db = scope.ServiceProvider.GetRequiredService(); + var xs = scope.ServiceProvider.GetRequiredService(); + var ctx = scope.Superuser(); + + var (_, xchange) = await FailedXchange(db, xs, "Chain Scheduled Doc"); + var second = await RetryOnce(db, xs, ctx, xchange.Id); + + // Takes a race to arrive at: the manual retry got in after the policy scheduled one. What + // matters is that the job drops it rather than throwing, which would leave the schedule in + // place to fail again on every pass. + var delayed = new DelayedRetry { Id = xchange.Id, On = DateTime.UtcNow.AddMinutes(-1) }; + db.Set().Add(delayed); + await db.SaveChangesAsync(); + + Assert.False(await xs.ExecuteDelayedRetry(delayed)); + await db.SaveChangesAsync(); + + Assert.False(await db.Set().AnyAsync(d => d.Id == xchange.Id)); + Assert.Equal(1, await db.Set().CountAsync(x => x.RetryFor == xchange.Id)); + + var result = await db.Set().AsNoTracking().FirstAsync(r => r.Id == xchange.Id); + Assert.Contains(second.Id, result.RetryBlockedReason); + } + + // ─── Bulk retry follows the chain ───────────────────────────────────────── + + [Fact] + public async Task BulkRetry_retries_the_newest_attempt_of_an_already_retried_selection() + { + await using var scope = _fixture.CreateScope(); + var db = scope.ServiceProvider.GetRequiredService(); + var xs = scope.ServiceProvider.GetRequiredService(); + var ctx = scope.Superuser(); + + var (_, xchange) = await FailedXchange(db, xs, "Bulk Chain Doc"); + var second = await RetryOnce(db, xs, ctx, xchange.Id); + var third = await RetryOnce(db, xs, ctx, second.Id); + + // Selecting the original, which is two attempts out of date. + var plan = (XchangeBulkRetryPlan)await new Resources.Xchanges.BulkRetry(db, ctx, xs) + .Handle(new XchangeBulkRetry { Ids = [xchange.Id] }); + await db.SaveChangesAsync(); + + Assert.Equal(1, plan.WillRetry); + var substitution = Assert.Single(plan.Substituted); + Assert.Equal(xchange.Id, substitution.SelectedId); + Assert.Equal(third.Id, substitution.RetryId); + + Assert.True(await db.Set().AnyAsync(x => x.RetryFor == third.Id)); + Assert.Equal(1, await db.Set().CountAsync(x => x.RetryFor == xchange.Id)); + } + + [Fact] + public async Task BulkRetry_retries_a_chain_once_when_two_of_its_attempts_are_selected() + { + await using var scope = _fixture.CreateScope(); + var db = scope.ServiceProvider.GetRequiredService(); + var xs = scope.ServiceProvider.GetRequiredService(); + var ctx = scope.Superuser(); + + var (_, xchange) = await FailedXchange(db, xs, "Bulk Dedupe Doc"); + var second = await RetryOnce(db, xs, ctx, xchange.Id); + + var plan = (XchangeBulkRetryPlan)await new Resources.Xchanges.BulkRetry(db, ctx, xs) + .Handle(new XchangeBulkRetry { Ids = [xchange.Id, second.Id] }); + await db.SaveChangesAsync(); + + Assert.Equal(2, plan.Selected); + Assert.Equal(1, plan.WillRetry); + Assert.Equal(1, await db.Set().CountAsync(x => x.RetryFor == second.Id)); + } + + /// + /// Every selection in one chain has to resolve to the same end of it. The first version walked + /// the tree a level at a time with a set of nodes already seen, which made a selection lagging + /// behind another stop on a node the leader had stepped over — so it was reported as handing + /// over to an attempt that had itself been retried, and retrying that threw ALREADY_RETRIED and + /// took the whole selection down. Needs three or more attempts and two selections at different + /// depths to show up at all. + /// + [Fact] + public async Task BulkRetry_resolves_every_selection_in_one_chain_to_the_same_end() + { + await using var scope = _fixture.CreateScope(); + var db = scope.ServiceProvider.GetRequiredService(); + var xs = scope.ServiceProvider.GetRequiredService(); + var ctx = scope.Superuser(); + + var (_, first) = await FailedXchange(db, xs, "Bulk Lagging Doc"); + var second = await RetryOnce(db, xs, ctx, first.Id); + var third = await RetryOnce(db, xs, ctx, second.Id); + var fourth = await RetryOnce(db, xs, ctx, third.Id); + + // The oldest and one in the middle: two selections, three and one attempts out of date. + var plan = (XchangeBulkRetryPlan)await new Resources.Xchanges.BulkRetry(db, ctx, xs) + .Handle(new XchangeBulkRetry { Ids = [first.Id, third.Id] }); + await db.SaveChangesAsync(); + + Assert.Equal(2, plan.Selected); + Assert.Equal(1, plan.WillRetry); + Assert.All(plan.Substituted, sub => Assert.Equal(fourth.Id, sub.RetryId)); + Assert.Equal(2, plan.Substituted.Count); + + // And only the end of the chain grew. + Assert.Equal(1, await db.Set().CountAsync(x => x.RetryFor == fourth.Id)); + foreach (var stale in new[] { first.Id, second.Id, third.Id }) + Assert.Equal(1, await db.Set().CountAsync(x => x.RetryFor == stale)); + } + + [Fact] + public async Task BulkRetry_skips_a_chain_whose_newest_attempt_succeeded() + { + await using var scope = _fixture.CreateScope(); + var db = scope.ServiceProvider.GetRequiredService(); + var xs = scope.ServiceProvider.GetRequiredService(); + var ctx = scope.Superuser(); + + var (_, xchange) = await FailedXchange(db, xs, "Bulk Succeeded Doc"); + + await new Resources.Xchanges.Retry(db, ctx, xs).Handle(xchange.Id, new XchangeRetry()); + await db.SaveChangesAsync(); + var second = await db.Set().FirstAsync(x => x.RetryFor == xchange.Id); + db.Set().Add(new XchangeResult(second.Id, null, null)); + await db.SaveChangesAsync(); + + var plan = (XchangeBulkRetryPlan)await new Resources.Xchanges.BulkRetry(db, ctx, xs) + .Handle(new XchangeBulkRetry { Ids = [xchange.Id] }); + await db.SaveChangesAsync(); + + Assert.Equal(0, plan.WillRetry); + Assert.Contains("succeeded", Assert.Single(plan.Skipped).Reason); + Assert.False(await db.Set().AnyAsync(x => x.RetryFor == second.Id)); + } + + [Fact] + public async Task BulkRetry_selects_every_exchange_a_filter_matches_minus_the_exclusions() + { + await using var scope = _fixture.CreateScope(); + var db = scope.ServiceProvider.GetRequiredService(); + var xs = scope.ServiceProvider.GetRequiredService(); + var ctx = scope.Superuser(); + + // Three failures on one subscription — more than a page would show, in miniature. + var (sub, first) = await FailedXchange(db, xs, "Bulk Filter Doc"); + var second = await xs.CreateXchange(sub, new XchangeFile("{}")); + var third = await xs.CreateXchange(sub, new XchangeFile("{}")); + await db.SaveChangesAsync(); + db.Set().Add(new XchangeResult(second.Id, null, null, exception: "boom")); + db.Set().Add(new XchangeResult(third.Id, null, null, exception: "boom")); + await db.SaveChangesAsync(); + + var plan = (XchangeBulkRetryPlan)await new Resources.Xchanges.BulkRetry(db, ctx, xs) + .Handle(new XchangeBulkRetry + { + // The same filter string the exchange list builds, so "select all matching" means + // the set that was on screen. + Filter = $"filter=SubscriptionId:1:{sub.Id}&filter=StatusFilter:1:3", + ExcludeIds = [third.Id] + }); + await db.SaveChangesAsync(); + + Assert.Equal(2, plan.Selected); + Assert.Equal(2, plan.WillRetry); + Assert.True(await db.Set().AnyAsync(x => x.RetryFor == first.Id)); + Assert.True(await db.Set().AnyAsync(x => x.RetryFor == second.Id)); + Assert.False(await db.Set().AnyAsync(x => x.RetryFor == third.Id)); + } + + [Fact] + public async Task BulkRetryPreview_describes_the_same_work_without_doing_it() + { + await using var scope = _fixture.CreateScope(); + var db = scope.ServiceProvider.GetRequiredService(); + var xs = scope.ServiceProvider.GetRequiredService(); + var ctx = scope.Superuser(); + + var (_, xchange) = await FailedXchange(db, xs, "Preview Doc"); + var second = await RetryOnce(db, xs, ctx, xchange.Id); + + var plan = (XchangeBulkRetryPlan)await new Resources.Xchanges.BulkRetryPreview(db, ctx) + .Handle(new XchangeBulkRetry { Ids = [xchange.Id] }); + + Assert.Equal(1, plan.WillRetry); + Assert.Equal(second.Id, Assert.Single(plan.Substituted).RetryId); + Assert.False(await db.Set().AnyAsync(x => x.RetryFor == second.Id)); + } + + /// + /// Retrying is an operator's action, and until this was added the endpoints took anyone's word + /// for it: the UI hid the button without a permission, but the API accepted the call from any + /// signed-in account. PermissionGuardTests could not have caught it — those exercise + /// EnsurePermission itself, and the gap was in never calling it. + /// + [Fact] + public async Task Retrying_is_refused_to_an_account_that_cannot_operate_exchanges() + { + await using var scope = _fixture.CreateScope(); + var db = scope.ServiceProvider.GetRequiredService(); + var xs = scope.ServiceProvider.GetRequiredService(); + + var (_, xchange) = await FailedXchange(db, xs, "Retry Guard Perm Doc"); + + var viewerId = await scope.AsNewViewer($"retry-guard-{System.Guid.NewGuid():N}"); + var viewer = scope.As(viewerId); + + await Assert.ThrowsAsync(() => + new Resources.Xchanges.Retry(db, viewer, xs).Handle(xchange.Id, new XchangeRetry())); + + await Assert.ThrowsAsync(() => + new Resources.Xchanges.BulkRetry(db, viewer, xs) + .Handle(new XchangeBulkRetry { Ids = [xchange.Id] })); + + await Assert.ThrowsAsync(() => + new Resources.Xchanges.BulkRetryPreview(db, viewer) + .Handle(new XchangeBulkRetry { Ids = [xchange.Id] })); + + Assert.False(await db.Set().AnyAsync(x => x.RetryFor == xchange.Id)); + } + + /// + /// The plan has to answer the question actually being asked. Re-resolving properties needs the + /// subscription, so for an exchange that has none the answer differs with the choice — and a + /// plan that promised a retry the retry then skipped would make the confirmation worthless. + /// + [Fact] + public async Task BulkRetryPreview_answers_for_the_reset_that_was_asked_about() + { + await using var scope = _fixture.CreateScope(); + var db = scope.ServiceProvider.GetRequiredService(); + var xs = scope.ServiceProvider.GetRequiredService(); + var ctx = scope.Superuser(); + + // A document-only exchange: no subscription from the start, which is also what an exchange + // whose subscription was later deleted looks like. + var doc = new Document(null, "Preview Reset Doc", DocumentFormat.Json); + db.Set().Add(doc); + await db.SaveChangesAsync(); + + var orphan = await xs.CreateXchange(doc, WorkGroup.None, new XchangeFile("{}")); + await db.SaveChangesAsync(); + db.Set().Add(new XchangeResult(orphan.Id, null, null, exception: "boom")); + await db.SaveChangesAsync(); + + var preview = new Resources.Xchanges.BulkRetryPreview(db, ctx); + + var plain = (XchangeBulkRetryPlan)await preview.Handle( + new XchangeBulkRetry { Ids = [orphan.Id], Reset = false }); + Assert.Equal(1, plain.WillRetry); + Assert.Empty(plain.Skipped); + + var withReset = (XchangeBulkRetryPlan)await preview.Handle( + new XchangeBulkRetry { Ids = [orphan.Id], Reset = true }); + Assert.Equal(0, withReset.WillRetry); + Assert.Contains("subscription no longer exists", Assert.Single(withReset.Skipped).Reason); + + // And the retry itself agrees with the plan that described it. + var done = (XchangeBulkRetryPlan)await new Resources.Xchanges.BulkRetry(db, ctx, xs) + .Handle(new XchangeBulkRetry { Ids = [orphan.Id], Reset = true }); + await db.SaveChangesAsync(); + Assert.Equal(0, done.WillRetry); + Assert.False(await db.Set().AnyAsync(x => x.RetryFor == orphan.Id)); + } + + /// + /// A status the filter cannot read used to be dropped, which widened the selection to every + /// exchange there is — harmless in a search, but bulk retry runs over whatever this selects. + /// + [Fact] + public async Task BulkRetry_refuses_a_filter_carrying_an_unreadable_status() + { + await using var scope = _fixture.CreateScope(); + var db = scope.ServiceProvider.GetRequiredService(); + var xs = scope.ServiceProvider.GetRequiredService(); + var ctx = scope.Superuser(); + + var (_, xchange) = await FailedXchange(db, xs, "Bad Status Doc"); + + await Assert.ThrowsAsync(() => + new Resources.Xchanges.BulkRetry(db, ctx, xs) + .Handle(new XchangeBulkRetry { Filter = "filter=StatusFilter:1:9" })); + + Assert.False(await db.Set().AnyAsync(x => x.RetryFor == xchange.Id)); + } + + // ─── Reading the chain back ─────────────────────────────────────────────── + + [Fact] + public async Task RetryTree_returns_the_whole_chain_whichever_attempt_is_asked_about() + { + await using var scope = _fixture.CreateScope(); + var db = scope.ServiceProvider.GetRequiredService(); + var xs = scope.ServiceProvider.GetRequiredService(); + var ctx = scope.Superuser(); + + var (_, xchange) = await FailedXchange(db, xs, "Tree Doc"); + var second = await RetryOnce(db, xs, ctx, xchange.Id); + var third = await RetryOnce(db, xs, ctx, second.Id); + + foreach (var asked in new[] { xchange.Id, second.Id, third.Id }) + { + var tree = (XchangeRetryTree)await new Resources.Xchanges.RetryTree(db, ctx) + .Handle(new XchangeRetryTreeRequest { Id = asked }); + + Assert.Equal(xchange.Id, tree.RootId); + Assert.False(tree.Truncated); + Assert.Equal([xchange.Id, second.Id, third.Id], tree.Nodes.Select(n => n.Id)); + Assert.Equal([null, xchange.Id, second.Id], tree.Nodes.Select(n => n.RetryFor)); + Assert.All(tree.Nodes.Skip(1), n => Assert.True(n.ManualRetry)); + } + } + + [Fact] + public async Task Exchange_search_reports_which_rows_have_been_retried() + { + await using var scope = _fixture.CreateScope(); + var db = scope.ServiceProvider.GetRequiredService(); + var xs = scope.ServiceProvider.GetRequiredService(); + var ctx = scope.Superuser(); + + var (sub, xchange) = await FailedXchange(db, xs, "HasRetry Doc"); + var second = await RetryOnce(db, xs, ctx, xchange.Id); + + var response = (SearchyResponse)await new Resources.Xchanges.Search(db, xs, ctx) + .Handle(new SearchyRequest($"filter=SubscriptionId:1:{sub.Id}") { PageSize = 50 }); + + Assert.True(response.Result.Single(r => r.Id == xchange.Id).HasRetry); + Assert.False(response.Result.Single(r => r.Id == second.Id).HasRetry); + Assert.Equal(xchange.Id, response.Result.Single(r => r.Id == second.Id).RetryFor); + } +} diff --git a/SW.Bitween.IntegrationTests/Tests/RetryJobTests.cs b/SW.Bitween.IntegrationTests/Tests/RetryJobTests.cs index 7f9bbf12..0966ce3c 100644 --- a/SW.Bitween.IntegrationTests/Tests/RetryJobTests.cs +++ b/SW.Bitween.IntegrationTests/Tests/RetryJobTests.cs @@ -1,4 +1,4 @@ -using System; +using System; using System.Linq; using System.Threading.Tasks; using Microsoft.EntityFrameworkCore; @@ -206,7 +206,7 @@ public async Task BulkRetry_handles_an_exchange_with_no_subscription() // One selection containing both. This threw before, so the whole bulk retry failed — including // for the exchanges that were perfectly retryable. - await new Resources.Xchanges.BulkRetry(db, xs).Handle(new XchangeBulkRetry + await new Resources.Xchanges.BulkRetry(db, scope.Superuser(), xs).Handle(new XchangeBulkRetry { Ids = [orphan.Id, healthy.Id], Reset = false diff --git a/SW.Bitween.Sdk/Model/Xchange.cs b/SW.Bitween.Sdk/Model/Xchange.cs index e29dbbc8..94f5032d 100644 --- a/SW.Bitween.Sdk/Model/Xchange.cs +++ b/SW.Bitween.Sdk/Model/Xchange.cs @@ -41,7 +41,138 @@ public class XchangeRetry public class XchangeBulkRetry : XchangeRetry { + /// + /// The exchanges the caller picked by hand. Ignored when is set. + /// public List Ids { get; set; } + + /// + /// A whole filter's worth of exchanges instead of a hand-picked list, as the same + /// query-string fragment the exchange search takes (filter=StatusFilter:1:3&filter=PartnerId:1:7). + /// Sent when someone chose "select all matching", so the selection is not limited to + /// the rows one page happened to show. + /// + public string Filter { get; set; } + + /// + /// Exchanges to leave out of a selection — the rows unticked after + /// selecting everything. + /// + public List ExcludeIds { get; set; } + } + + /// + /// What a bulk retry is about to do, so it can be shown before it is run. Returned by the + /// preview and again by the retry itself, where the counts are what actually happened. + /// + public class XchangeBulkRetryPlan + { + /// How many exchanges the selection came to, before resolving any chains. + public int Selected { get; set; } + + /// Distinct exchanges that will be (or were) retried. + public int WillRetry { get; set; } + + /// The largest selection this endpoint will carry out in one request. + public int Limit { get; set; } + + /// + /// true when the selection is past . Nothing else is filled in + /// then — resolving thousands of chains to describe a request that will be refused + /// costs more than the answer is worth. + /// + public bool OverLimit { get; set; } + + /// + /// Selections that had already been retried, and the later attempt standing in for each. + /// + public List Substituted { get; set; } = new List(); + + /// Selections nothing will be done about, each with the reason. + public List Skipped { get; set; } = new List(); + + /// + /// Promoted properties for every exchange named in or + /// , keyed by id, so the caller can name them the way the exchange + /// list does. One lookup rather than a copy per entry, since one attempt is often named by + /// several selections. Exchanges whose information type promotes nothing are absent. + /// + public Dictionary> Properties { get; set; } = + new Dictionary>(); + } + + /// + /// A selected exchange that had already been retried, paired with the newest attempt in its + /// chain — the one a retry actually runs, since retrying the selected one again is refused. + /// + public class XchangeRetrySubstitution + { + public string SelectedId { get; set; } + public string RetryId { get; set; } + } + + public class XchangeRetrySkip + { + /// + /// The exchange the skip is about: the one selected, or — once it had already been + /// retried — the later attempt that stood in for it. + /// + public string Id { get; set; } + + public string Reason { get; set; } + } + + public class XchangeRetryTreeRequest + { + /// Any exchange in the chain — the answer is the same whichever one is asked about. + public string Id { get; set; } + } + + /// + /// Every attempt related to one exchange: the original, and each retry descended from it. + /// + public class XchangeRetryTree + { + /// The original attempt everything in descends from. + public string RootId { get; set; } + + /// + /// Flat, oldest first. Each node names its parent, so the caller rebuilds the shape + /// without the server having to pick a rendering. + /// + public List Nodes { get; set; } = new List(); + + /// + /// true when the walk stopped at its depth limit, so the chain shown is only + /// the part nearest the exchange asked about. + /// + public bool Truncated { get; set; } + } + + public class XchangeRetryNode + { + public string Id { get; set; } + public string RetryFor { get; set; } + + /// + /// What the payload promoted, so an attempt can be named the way the exchange list names + /// it. An id identifies an exchange but says nothing about which one it is. + /// + public IDictionary PromotedProperties { get; set; } + public DateTime StartedOn { get; set; } + public DateTime? FinishedOn { get; set; } + + /// null while the exchange is still running. + public bool? Status { get; set; } + + public bool? ResponseBad { get; set; } + public string Exception { get; set; } + + /// true when a person asked for this attempt rather than a retry policy. + public bool ManualRetry { get; set; } + + public DateTime? ScheduledRetryOn { get; set; } + public string RetryBlockedReason { get; set; } } public class XchangeGetResultResponse @@ -100,5 +231,17 @@ public class XchangeRow /// Why the retry policy declined to schedule another attempt, when it declined. public string RetryBlockedReason { get; set; } + + /// + /// true when something has already been retried from this exchange. An exchange + /// gets at most one retry, so this is also what makes it un-retryable: the row's own + /// Retry action gives way to a link to the later attempt. + /// + /// + /// Carried on the row rather than left to the retry-tree endpoint so that the common + /// case — an exchange with no retries either side of it — needs no second request at + /// all. Costs one index lookup per returned row, on the index over RetryFor. + /// + public bool HasRetry { get; set; } } } \ No newline at end of file diff --git a/SW.Bitween.Web/ClientApp/e2e/exchanges.spec.ts b/SW.Bitween.Web/ClientApp/e2e/exchanges.spec.ts index d2bff956..0224089e 100644 --- a/SW.Bitween.Web/ClientApp/e2e/exchanges.spec.ts +++ b/SW.Bitween.Web/ClientApp/e2e/exchanges.spec.ts @@ -79,3 +79,94 @@ test("queue health page loads with live consumer data", async ({ page }) => { await expect(page.getByText("v3.local.bitween").first()).toBeVisible({ timeout: 10000 }); await expect(page.getByText("undefined")).toHaveCount(0); }); + +/** + * An exchange is retried at most once, so a retried one stops offering Retry and hands over to + * the attempt that can be retried. Built through the UI rather than pinned to particular ids, + * since the retry has to exist for the state to be real. + */ +test("a retried exchange shows its chain and sends you to the newest attempt", async ({ page }) => { + test.setTimeout(60000); + await page.goto("exchanges?status=failed"); + await expect(page.getByRole("row").nth(1)).toBeVisible({ timeout: 15000 }); + await page.getByLabel("Refresh interval").selectOption("0"); + + // Whichever failed exchange has not been retried yet. Most have not, but this database + // accumulates chains as the suite runs, and an already-retried one has no Retry button to + // press — which is the very thing under test further down. + let retried: string | null = null; + const failedRows = page.getByRole("row").filter({ hasText: "Failed" }); + // The whole page, not the first few rows: this database accumulates chains as the suite runs, + // and a run that happened to leave several retried exchanges at the top would otherwise fail + // here before reaching what the test is about. Newest first, and every retry lands at the top + // as a fresh un-retried leaf, so a page is far more than enough. + const candidates = await failedRows.count(); + for (let i = 0; i < candidates && retried === null; i++) { + const row = failedRows.nth(i); + const label = await row.locator("input[type=checkbox]").getAttribute("aria-label"); + await row.locator("td").last().click(); + if (await page.getByRole("button", { name: "Retry…" }).isVisible()) { + retried = label!.replace("Select ", ""); + break; + } + await row.locator("td").last().click(); // collapse and try the next one + } + expect( + retried, + `no un-retried failed exchange among the ${candidates} on this page`, + ).not.toBeNull(); + + await page.getByRole("button", { name: "Retry…" }).click(); + await page.getByRole("dialog").getByRole("button", { name: "Retry" }).click(); + await expect(page.getByText(/Retry started/)).toBeVisible({ timeout: 15000 }); + + // Open that same exchange again: it is spent now. + await page.goto(`exchanges?ids=${retried}`); + const row = page.locator(`tr:has(input[aria-label="Select ${retried}"])`); + await expect(row).toBeVisible({ timeout: 15000 }); + await row.locator("td").last().click(); + + await expect(page.getByText("Already retried")).toBeVisible({ timeout: 10000 }); + await expect(page.getByRole("button", { name: "Retry…" })).toHaveCount(0); + await expect(page.getByRole("link", { name: /Open the newest attempt/ })).toBeVisible(); + + // And the chain itself, with this exchange marked in it. + await expect(page.getByText(/Retry chain · \d+ attempts/)).toBeVisible(); + await expect(page.getByText("You are here")).toBeVisible(); +}); + +/** + * A selection has to be able to mean "everything this filter matches", or a 200-exchange + * recovery is 8 pages of ticking boxes. Stops at the confirm — what it says is the point, and + * running it would retry the whole filter. + */ +test("select all matching covers the whole filter, and the confirm says what will run", async ({ page }) => { + await page.goto("exchanges?status=failed"); + await expect(page.getByRole("row").nth(1)).toBeVisible({ timeout: 15000 }); + await page.getByLabel("Refresh interval").selectOption("0"); + + await page.getByRole("checkbox", { name: "Select all on this page" }).check(); + const offer = page.getByRole("button", { name: /Select all [\d,]+\+? matching this filter/ }); + await expect(offer).toBeVisible(); + await offer.click(); + + await expect(page.getByText(/everything this filter matches/)).toBeVisible(); + + // Unticking a row in this mode records an exclusion rather than dropping out of it. + const before = await page.locator("text=/^[\\d,]+\\+? selected/").first().innerText(); + await page.getByRole("checkbox", { name: /^Select (?!all\b)/ }).first().uncheck(); + await expect(page.getByText(/1 unticked/)).toBeVisible(); + expect(await page.locator("text=/^[\\d,]+\\+? selected/").first().innerText()).not.toBe(before); + + // The confirm describes the selection the server resolved, not the rows on screen. + await page.getByRole("button", { name: "Retry selected…" }).click(); + const dialog = page.getByRole("dialog"); + await expect(dialog).toBeVisible(); + await expect(dialog.getByText(/Retry [\d,]+ exchanges\?/)).toBeVisible({ timeout: 15000 }); + // Either it is within the cap and says how many will run, or it is past it and refuses. + await expect(dialog.getByText(/will run again|more than the [\d,]+ a single retry|Nothing here can be retried/)).toBeVisible({ timeout: 15000 }); + + // By text, not accessible name: the dialog's own × is also called "Close". + await dialog.locator("button", { hasText: /^(Cancel|Close)$/ }).click(); + await expect(dialog).toHaveCount(0); +}); diff --git a/SW.Bitween.Web/ClientApp/src/api/client.ts b/SW.Bitween.Web/ClientApp/src/api/client.ts index 957765dc..26b64e52 100644 --- a/SW.Bitween.Web/ClientApp/src/api/client.ts +++ b/SW.Bitween.Web/ClientApp/src/api/client.ts @@ -14,6 +14,9 @@ import type { DashboardData, ExchangeQuery, ExchangeRow, + BulkRetryPlan, + BulkRetrySelection, + RetryTree, GlobalValuesSetDetail, GlobalValuesSetRow, InformationType, @@ -385,7 +388,9 @@ export interface ApiClient { */ retryExchange(id: string, opts: { reset: boolean }): Promise<{ id: string }>; /** Retries many; exchanges with a pending auto-retry are skipped, not failed. */ - bulkRetryExchanges(ids: string[], opts: { reset: boolean }): Promise<{ retried: number; skipped: number }>; + bulkRetryExchanges(selection: BulkRetrySelection, opts: { reset: boolean }): Promise; + previewBulkRetry(selection: BulkRetrySelection, opts: { reset: boolean }): Promise; + getRetryTree(id: string): Promise; /** Manually injects a payload, addressed at a subscription or an information type. */ createExchange(input: { target: "subscription" | "informationType"; diff --git a/SW.Bitween.Web/ClientApp/src/api/http/exchanges.ts b/SW.Bitween.Web/ClientApp/src/api/http/exchanges.ts index 6be40cf9..b8dfba6b 100644 --- a/SW.Bitween.Web/ClientApp/src/api/http/exchanges.ts +++ b/SW.Bitween.Web/ClientApp/src/api/http/exchanges.ts @@ -1,9 +1,12 @@ import type { ApiClient } from "../client"; import type { + BulkRetryPlan, + BulkRetrySelection, ExchangeQuery, ExchangeRow, ExchangeStatus, Paged, + RetryTree, ScheduledRetryQuery, ScheduledRetryRow, } from "../types"; @@ -40,7 +43,39 @@ interface RawXchangeRow { correlationId: string | null; partnerId: number | null; scheduledRetryOn: string | null; + hasRetry: boolean; } + +interface RawRetryNode { + id: string; + retryFor: string | null; + promotedProperties: Record | null; + startedOn: string; + finishedOn: string | null; + status: boolean | null; + responseBad: boolean | null; + exception: string | null; + manualRetry: boolean; + scheduledRetryOn: string | null; + retryBlockedReason: string | null; +} + +interface RawRetryTree { + rootId: string; + nodes: RawRetryNode[]; + truncated: boolean; +} + +interface RawBulkRetryPlan { + selected: number; + willRetry: number; + limit: number; + overLimit: boolean; + substituted: { selectedId: string; retryId: string }[] | null; + skipped: { id: string; reason: string }[] | null; + properties: Record> | null; +} + interface RawDelayedRetryRow { id: string; on: string; @@ -84,6 +119,7 @@ const toExchangeRow = (raw: RawXchangeRow, partnerNameById: Map) finishedOn: raw.finishedOn, correlationId: raw.correlationId, retryFor: raw.retryFor, + hasRetry: raw.hasRetry, aggregationXchangeId: raw.aggregationXchangeId, scheduledRetryOn: raw.scheduledRetryOn, exception: raw.exception, @@ -124,7 +160,11 @@ const toScheduledRetryRow = (raw: RawDelayedRetryRow): ScheduledRetryRow => ({ * comparisons (`GreaterThanOrEquals`/`LessThanOrEquals`, rules 6/8) go * through a different code path and work correctly — use those instead. */ -function buildExchangeQuery(query: ExchangeQuery): string { +/** + * Just the filters, without paging — what "select all matching" sends, so a bulk retry acts on + * the same set the list was showing rather than on the 25 rows that happened to be on screen. + */ +function buildExchangeFilters(query: ExchangeQuery): URLSearchParams { const params = new URLSearchParams(); if (query.status) params.append("filter", `StatusFilter:1:${STATUS_FILTER[query.status]}`); if (query.subscriptionId !== undefined) params.append("filter", `SubscriptionId:1:${query.subscriptionId}`); @@ -144,6 +184,11 @@ function buildExchangeQuery(query: ExchangeQuery): string { if (propertyTerm) params.append("filter", `PromotedPropertiesRaw:4:${propertyTerm}`); if (query.from) params.append("filter", `StartedOn:6:${query.from}`); if (query.to) params.append("filter", `StartedOn:8:${query.to}`); + return params; +} + +function buildExchangeQuery(query: ExchangeQuery): string { + const params = buildExchangeFilters(query); params.set("page", String(Math.floor(query.offset / query.limit))); params.set("size", String(query.limit)); return searchyQueryString(params); @@ -161,6 +206,38 @@ function buildScheduledRetryQuery(query: ScheduledRetryQuery): string { return searchyQueryString(params); } +/** + * Both bulk-retry endpoints take the same request, so the preview cannot describe a different + * selection than the retry acts on. The filter goes over as the same query-string fragment the + * list itself sends, percent-encoded the way the backend's parser expects (see + * `searchyQueryString`), and the backend re-runs it — the client never has to enumerate ids it + * has not loaded. + */ +async function postBulkRetry( + url: string, + selection: BulkRetrySelection, + reset: boolean, +): Promise { + const body = + "ids" in selection + ? { ids: selection.ids } + : { + filter: searchyQueryString(buildExchangeFilters(selection.matching)), + excludeIds: selection.excludeIds, + }; + + const plan = await post(url, { ...body, reason: "Bulk retry", reset }); + return { + selected: plan.selected, + willRetry: plan.willRetry, + limit: plan.limit, + overLimit: plan.overLimit, + substituted: plan.substituted ?? [], + skipped: plan.skipped ?? [], + properties: plan.properties ?? {}, + }; +} + async function partnerNameMap(): Promise> { const partners = await partnerMethods.listPartners(); return new Map(partners.map((p) => [p.id, p.name])); @@ -190,20 +267,38 @@ export const exchangeMethods = { return { id: res.result[0]?.id ?? id }; }, - async bulkRetryExchanges(ids: string[], { reset }: { reset: boolean }): Promise<{ retried: number; skipped: number }> { - // BulkRetry.cs silently skips ids that already have a scheduled auto-retry - // and returns null — mirror its exact skip rule ourselves beforehand so we - // can report real counts back to the caller. - const idFilter = `Id:4:text|${ids.join("|")}`; - const current = await get>( - `/xchanges?filter=${encodeURIComponent(idFilter)}&size=${ids.length}`, - ); - // The Id filter also matches retryFor/aggregationXchangeId — narrow back - // down to exactly the requested ids. - const byId = new Map(current.result.filter((r) => ids.includes(r.id)).map((r) => [r.id, r])); - const skipped = ids.filter((id) => byId.get(id)?.scheduledRetryOn != null).length; - await post("/xchanges/bulkretry", { ids, reason: "Bulk retry", reset }); - return { retried: ids.length - skipped, skipped }; + /** + * Every attempt related to one exchange. Worth asking for only when the row says there is + * something to see — `retryFor` or `hasRetry` — since most exchanges have neither. + */ + async getRetryTree(id: string): Promise { + const raw = await get(`/xchanges/retrytree?id=${encodeURIComponent(id)}`); + return { + rootId: raw.rootId, + truncated: raw.truncated, + attempts: (raw.nodes ?? []).map((n) => ({ + id: n.id, + retryFor: n.retryFor, + promotedProperties: n.promotedProperties, + startedOn: n.startedOn, + finishedOn: n.finishedOn, + status: deriveStatus(n), + exception: n.exception, + manualRetry: n.manualRetry, + scheduledRetryOn: n.scheduledRetryOn, + retryBlockedReason: n.retryBlockedReason, + })), + }; + }, + + /** What a bulk retry would do, so it can be shown before anyone commits to it. */ + previewBulkRetry(selection: BulkRetrySelection, { reset }: { reset: boolean }): Promise { + return postBulkRetry("/xchanges/bulkretrypreview", selection, reset); + }, + + /** Runs the retry and reports what it actually did, in the same shape as the preview. */ + bulkRetryExchanges(selection: BulkRetrySelection, { reset }: { reset: boolean }): Promise { + return postBulkRetry("/xchanges/bulkretry", selection, reset); }, async createExchange(input: { diff --git a/SW.Bitween.Web/ClientApp/src/api/queryKeys.ts b/SW.Bitween.Web/ClientApp/src/api/queryKeys.ts index 7c622260..11060544 100644 --- a/SW.Bitween.Web/ClientApp/src/api/queryKeys.ts +++ b/SW.Bitween.Web/ClientApp/src/api/queryKeys.ts @@ -124,6 +124,13 @@ export const keys = { all: ["exchanges"] as const, search: (params: string) => ["exchanges", "search", params] as const, document: (key: string | null) => ["exchanges", "document", key] as const, + /** + * A chain never changes above the exchange asked about, and only grows below it, so this is + * safe to keep and to prefetch on hover. + */ + retryTree: (id: string) => ["exchanges", "retryTree", id] as const, + /** Keyed by the serialized selection, since that is the whole of what the plan depends on. */ + bulkRetryPreview: (selection: string) => ["exchanges", "bulkRetryPreview", selection] as const, }, scheduledRetries: { diff --git a/SW.Bitween.Web/ClientApp/src/api/types.ts b/SW.Bitween.Web/ClientApp/src/api/types.ts index 6602c861..22f415a8 100644 --- a/SW.Bitween.Web/ClientApp/src/api/types.ts +++ b/SW.Bitween.Web/ClientApp/src/api/types.ts @@ -845,6 +845,12 @@ export interface ExchangeRow { correlationId: string | null; /** Set when this exchange is a retry of another one. */ retryFor: string | null; + /** + * True when a retry has already been made from this exchange. An exchange gets at most one + * retry, so this is also what makes it un-retryable — the newest attempt in the chain is the + * one to act on. + */ + hasRetry: boolean; /** Set when this exchange was rolled up into an aggregation exchange. */ aggregationXchangeId: string | null; /** A pending auto-retry, when the retry policy scheduled one. */ @@ -860,6 +866,68 @@ export interface ExchangeRow { }; } +/** One attempt in a retry chain. */ +export interface RetryTreeNode { + id: string; + /** null on the original attempt. */ + retryFor: string | null; + /** What the payload promoted — how an attempt is named, the same as in the exchange list. */ + promotedProperties: Record | null; + startedOn: string; + finishedOn: string | null; + status: ExchangeStatus; + exception: string | null; + /** True when a person asked for this attempt rather than a retry policy. */ + manualRetry: boolean; + scheduledRetryOn: string | null; + retryBlockedReason: string | null; +} + +/** Every attempt related to one exchange, oldest first. */ +export interface RetryTree { + rootId: string; + attempts: RetryTreeNode[]; + /** True when the chain is longer than the server would walk. */ + truncated: boolean; +} + +/** + * Which exchanges a bulk retry is about: the rows someone ticked, or a whole filter's worth + * minus the ones they unticked. + */ +export type BulkRetrySelection = + | { ids: string[] } + /** Everything the list's current filters match, minus the rows unticked after selecting all. */ + | { matching: ExchangeQuery; excludeIds: string[] }; + +/** A selection that had already been retried, and the later attempt standing in for it. */ +export interface RetrySubstitution { + selectedId: string; + retryId: string; +} + +export interface RetrySkip { + /** The exchange the skip is about — the one selected, or the attempt that stood in for it. */ + id: string; + reason: string; +} + +/** What a bulk retry will do, or (returned by the retry itself) what it did. */ +export interface BulkRetryPlan { + selected: number; + willRetry: number; + limit: number; + /** True when the selection is past `limit`; nothing else is filled in. */ + overLimit: boolean; + substituted: RetrySubstitution[]; + skipped: RetrySkip[]; + /** + * Promoted properties for every exchange the plan names, keyed by id, so they can be named the + * way the exchange list names them. Absent for exchanges that promote nothing. + */ + properties: Record | undefined>; +} + export interface ExchangeQuery { status?: ExchangeStatus; subscriptionId?: number; diff --git a/SW.Bitween.Web/ClientApp/src/components/config/__tests__/promotedProps.test.ts b/SW.Bitween.Web/ClientApp/src/components/config/__tests__/promotedProps.test.ts new file mode 100644 index 00000000..6d97b74f --- /dev/null +++ b/SW.Bitween.Web/ClientApp/src/components/config/__tests__/promotedProps.test.ts @@ -0,0 +1,28 @@ +import { describe, expect, it } from "vitest"; +import { namesSomething } from "../shared"; + +/** + * Whether promoted properties can stand in for an exchange's identity. An information type can + * promote paths a payload never filled, and those name every exchange of that type equally — so + * the callers fall back to the id, which is at least this exchange's own. + */ +describe("namesSomething", () => { + it("is false when there are no promoted properties at all", () => { + expect(namesSomething(null)).toBe(false); + expect(namesSomething({})).toBe(false); + }); + + it("is false when every promoted path resolved to nothing", () => { + // What the backend sends for a type that promotes three paths the payload did not carry: + // unresolved values arrive as null rather than as an empty string. + expect(namesSomething({ merchant: null, orderRef: null, destination: null })).toBe(false); + expect(namesSomething({ merchant: "", orderRef: "" })).toBe(false); + }); + + it("is true as soon as one carries a value", () => { + expect(namesSomething({ merchant: "Acme" })).toBe(true); + // Partial information is still information, so the empty siblings stay on show beside it. + expect(namesSomething({ merchant: "Acme", orderRef: null })).toBe(true); + expect(namesSomething({ trackingNo: "0" })).toBe(true); + }); +}); diff --git a/SW.Bitween.Web/ClientApp/src/components/config/shared.tsx b/SW.Bitween.Web/ClientApp/src/components/config/shared.tsx index cfae1019..81dfb7fe 100644 --- a/SW.Bitween.Web/ClientApp/src/components/config/shared.tsx +++ b/SW.Bitween.Web/ClientApp/src/components/config/shared.tsx @@ -207,6 +207,15 @@ export function ExchangeStatusBadge({ * the width empty, so what the exchange actually *was* never made it to the * screen. */ +/** + * Promoted properties only name an exchange when at least one of them carries a value. An + * information type can promote three paths that a payload never filled, and + * "merchant= orderRef= destination=" then names every exchange of that type equally — so a + * caller with room for one identity is better off showing the id. + */ +export const namesSomething = (properties: Record | null) => + properties != null && Object.values(properties).some((v) => v != null && v !== ""); + export function PromotedProps({ properties, max = 3, @@ -226,7 +235,12 @@ export function PromotedProps({ // that resolved to nothing arrives as a null value rather than as an empty // string. Normalise once, here, so nothing downstream has to keep asking. const entries: [string, string][] = Object.entries(properties ?? {}).map(([k, v]) => [k, v ?? ""]); - if (entries.length === 0) + + // Keys whose values are all empty are treated like no promoted properties at all. An + // information type can promote three paths that a payload never filled, and + // "merchant= orderRef= destination=" then names every exchange of that type equally — three + // chips that say which fields exist and nothing about which record this is. + if (!namesSomething(properties)) return fallbackId ? ( {fallbackId.slice(0, 8)}… diff --git a/SW.Bitween.Web/ClientApp/src/pages/exchanges/ExchangeDrawer.tsx b/SW.Bitween.Web/ClientApp/src/pages/exchanges/ExchangeDrawer.tsx index 132f65e0..218b557c 100644 --- a/SW.Bitween.Web/ClientApp/src/pages/exchanges/ExchangeDrawer.tsx +++ b/SW.Bitween.Web/ClientApp/src/pages/exchanges/ExchangeDrawer.tsx @@ -11,6 +11,7 @@ import { formatDateTime, duration, timeUntil } from "../../lib/dates"; import { formatDocument } from "../../lib/documentPreview"; import { useSubscriptionsCache } from "../../components/config/shared"; import { RetryDialog, journeyStages, type JourneyStage } from "./shared"; +import { RetryChain, hasRetryChain, newestAttempt, retryTreeQuery } from "./RetryChain"; import { keys } from "../../api/queryKeys"; const STAGE_TONES: Record = { @@ -208,6 +209,11 @@ export function ExchangeDrawer({ x }: { x: ExchangeRow }) { enabled: activeKey !== null, }); + // The same query the chain below reads, so asking here costs nothing extra — and only asked + // for at all when the row says there is a chain to read. + const { data: chain } = useQuery({ ...retryTreeQuery(x.id), enabled: hasRetryChain(x) }); + const newest = chain ? newestAttempt(chain, x.id) : null; + const [confirming, setConfirming] = useState(false); const [actionError, setActionError] = useState(null); const [startedId, setStartedId] = useState(null); @@ -301,6 +307,9 @@ export function ExchangeDrawer({ x }: { x: ExchangeRow }) { )} + {/* — the attempts this exchange belongs to, when it belongs to any — */} + {hasRetryChain(x) && } + {/* — metadata — */}
{/* The id lives here rather than in the row: it identifies a record you @@ -347,16 +356,6 @@ export function ExchangeDrawer({ x }: { x: ExchangeRow }) { Retries & aggregation family - {x.retryFor && ( - - - {x.retryFor} - - - )} {isRollUp && ( {/* The Id filter matches AggregationXchangeId as well as Id, so this one @@ -413,12 +412,29 @@ export function ExchangeDrawer({ x }: { x: ExchangeRow }) { ) : ( - (x.status === "failed" || x.status === "badResponse") && ( + (x.status === "failed" || x.status === "badResponse") && + /* An exchange is retried at most once, so a spent one offers the way on to the + attempt that can be retried instead of a button that would be refused. */ + (x.hasRetry ? ( + <> + + Already retried + + {newest && newest.id !== x.id && ( + + Open the newest attempt to retry from there + + )} + + ) : ( - ) + )) )} {actionError &&

{actionError}

} diff --git a/SW.Bitween.Web/ClientApp/src/pages/exchanges/ExchangesPage.tsx b/SW.Bitween.Web/ClientApp/src/pages/exchanges/ExchangesPage.tsx index ae2a800e..b997ae32 100644 --- a/SW.Bitween.Web/ClientApp/src/pages/exchanges/ExchangesPage.tsx +++ b/SW.Bitween.Web/ClientApp/src/pages/exchanges/ExchangesPage.tsx @@ -1,8 +1,8 @@ -import { Fragment, useMemo, useState } from "react"; +import { Fragment, useEffect, useMemo, useState } from "react"; import { Link, useSearchParams } from "react-router"; import { keepPreviousData, useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; import { ChevronDown, ChevronRight, Plus, RotateCcw, X } from "lucide-react"; -import { api, type ExchangeQuery, type ExchangeStatus } from "../../api"; +import { api, type BulkRetrySelection, type ExchangeQuery, type ExchangeStatus } from "../../api"; import { Can } from "../../auth/guards"; import { PageHeader } from "../../components/layout/PageHeader"; import { Badge, Button, EmptyState, LoadingBlock } from "../../components/ui/basics"; @@ -11,6 +11,7 @@ import { SearchSelect } from "../../components/ui/SearchSelect"; import { useSubscriptionsCache } from "../../components/config/shared"; import { timeAgo, timeUntil, duration } from "../../lib/dates"; import { ExchangeDrawer } from "./ExchangeDrawer"; +import { hasRetryChain, retryTreeQuery } from "./RetryChain"; import { JourneyStrip, RetryDialog, STATUS_LABELS, StatusBadge } from "./shared"; import { PromotedProps } from "../../components/config/shared"; import { keys } from "../../api/queryKeys"; @@ -57,7 +58,16 @@ export function ExchangesPage() { const [refreshMs, setRefreshMs] = useState(15_000); const [open, setOpen] = useState>(new Set()); const [selected, setSelected] = useState>(new Set()); + /** + * "Select all matching" holds the filter rather than a list of ids, so a selection is not + * limited to the 25 rows one page happened to load. Unticking a row in this mode records an + * exclusion instead of removing an id. + */ + const [allMatching, setAllMatching] = useState(false); + const [excluded, setExcluded] = useState>(new Set()); const [bulkConfirm, setBulkConfirm] = useState(false); + /** Mirrors the confirm dialog's own choice, because the plan depends on it. */ + const [bulkReset, setBulkReset] = useState(false); const [bulkResult, setBulkResult] = useState(null); const queryClient = useQueryClient(); @@ -127,26 +137,70 @@ export function ExchangesPage() { return next; }); + const toggle = (set: Set, id: string) => { + const next = new Set(set); + if (next.has(id)) next.delete(id); + else next.add(id); + return next; + }; + const toggleSelected = (id: string) => - setSelected((prev) => { - const next = new Set(prev); - if (next.has(id)) next.delete(id); - else next.add(id); - return next; - }); + allMatching + ? setExcluded((prev) => toggle(prev, id)) + : setSelected((prev) => toggle(prev, id)); + + const clearSelection = () => { + setSelected(new Set()); + setExcluded(new Set()); + setAllMatching(false); + }; const rows = data?.result ?? []; const total = data?.total ?? 0; const totalIsCapped = total > COUNT_CAP; - const allOnPageSelected = rows.length > 0 && rows.every((r) => selected.has(r.id)); + + const isSelected = (id: string) => (allMatching ? !excluded.has(id) : selected.has(id)); + const allOnPageSelected = rows.length > 0 && rows.every((r) => isSelected(r.id)); + const selectedCount = allMatching ? Math.max(0, total - excluded.size) : selected.size; + + /** What the retry acts on: the ticked rows, or the filter itself minus the unticked ones. */ + const selection: BulkRetrySelection = allMatching + ? { matching: query, excludeIds: [...excluded] } + : { ids: [...selected] }; + + /** + * A selection means a set of exchanges, and changing the filters changes which set that is — + * for "all matching" it would silently come to mean something else entirely. Paging is not a + * filter, so ticking rows across pages still accumulates. + */ + const filterKey = FILTER_KEYS.map((k) => searchParams.get(k) ?? "").join("\u0000"); + useEffect(clearSelection, [filterKey]); + + /** + * What the retry would do, asked for while the confirm is open. A selection is rarely just + * itself — anything already retried hands over to its newest attempt — so this is shown + * before anyone commits rather than reported afterwards. + */ + const { data: plan, isFetching: planLoading } = useQuery({ + queryKey: keys.exchanges.bulkRetryPreview(JSON.stringify({ selection, reset: bulkReset })), + queryFn: () => api.previewBulkRetry(selection, { reset: bulkReset }), + enabled: bulkConfirm && selectedCount > 0, + staleTime: 30_000, + }); const bulkRetry = useMutation({ - mutationFn: (reset: boolean) => api.bulkRetryExchanges([...selected], { reset }), - onSuccess: ({ retried, skipped }) => { + mutationFn: (reset: boolean) => api.bulkRetryExchanges(selection, { reset }), + onSuccess: (done) => { setBulkConfirm(false); - setSelected(new Set()); + setBulkReset(false); + clearSelection(); setBulkResult( - `${retried} retr${retried === 1 ? "y" : "ies"} started${skipped > 0 ? `, ${skipped} skipped (auto-retry already scheduled)` : ""}.`, + `${done.willRetry.toLocaleString()} retr${done.willRetry === 1 ? "y" : "ies"} started` + + (done.substituted.length > 0 + ? `, ${done.substituted.length} continuing an existing chain` + : "") + + (done.skipped.length > 0 ? `, ${done.skipped.length} skipped` : "") + + ".", ); void queryClient.invalidateQueries({ queryKey: keys.exchanges.all }); }, @@ -357,14 +411,17 @@ export function ExchangesPage() { aria-label="Select all on this page" className="size-3.5 cursor-pointer accent-crimson-600" checked={allOnPageSelected} - onChange={() => + onChange={() => { + // In "all matching" mode the box stands for the whole filter, so + // clearing it drops the whole selection rather than excluding 25 rows. + if (allMatching) return clearSelection(); setSelected((prev) => { const next = new Set(prev); if (allOnPageSelected) rows.forEach((r) => next.delete(r.id)); else rows.forEach((r) => next.add(r.id)); return next; - }) - } + }); + }} /> @@ -383,6 +440,12 @@ export function ExchangesPage() { toggleOpen(x.id)} + /* Hovering a row is most of a second's head start on opening it, which is + enough that its retry chain is already there when the drawer renders. + Only for rows that have one — most exchanges do not. */ + onMouseEnter={() => { + if (hasRetryChain(x)) void queryClient.prefetchQuery(retryTreeQuery(x.id)); + }} className="cursor-pointer border-b border-ink-50 transition-colors last:border-0 hover:bg-ink-50/60" > e.stopPropagation()}> @@ -391,7 +454,7 @@ export function ExchangesPage() { type="checkbox" aria-label={`Select ${x.id}`} className="size-3.5 cursor-pointer accent-crimson-600" - checked={selected.has(x.id)} + checked={isSelected(x.id)} onChange={() => toggleSelected(x.id)} /> @@ -514,29 +577,65 @@ export function ExchangesPage() { )} {/* — bulk action bar — */} - {selected.size > 0 && ( -
- - {selected.size} selected - - - - - + {selectedCount > 0 && ( +
+
+ + + {totalIsCapped && allMatching ? `${COUNT_CAP.toLocaleString()}+` : selectedCount.toLocaleString()} + {" "} + selected + {allMatching && — everything this filter matches} + {allMatching && excluded.size > 0 && ( + , {excluded.size} unticked + )} + + + + + +
+ + {/* The whole page is ticked but there is more behind it — the one moment where + "select all matching" is what someone actually wants, so it is offered there + rather than living permanently in the toolbar. */} + {!allMatching && allOnPageSelected && total > rows.length && ( +

+ Only the {rows.length} rows on this page.{" "} + +

+ )}
)} {bulkConfirm && ( bulkRetry.mutate(reset)} - onClose={() => setBulkConfirm(false)} + onResetChange={setBulkReset} + onClose={() => { + setBulkConfirm(false); + // The dialog starts unticked each time it opens, so the mirrored copy has to as well. + setBulkReset(false); + }} /> )}
diff --git a/SW.Bitween.Web/ClientApp/src/pages/exchanges/RetryChain.tsx b/SW.Bitween.Web/ClientApp/src/pages/exchanges/RetryChain.tsx new file mode 100644 index 00000000..a784e617 --- /dev/null +++ b/SW.Bitween.Web/ClientApp/src/pages/exchanges/RetryChain.tsx @@ -0,0 +1,189 @@ +import { Link, useNavigate } from "react-router"; +import { useQuery } from "@tanstack/react-query"; +import { api, type RetryTree, type RetryTreeNode } from "../../api"; +import { keys } from "../../api/queryKeys"; +import { Badge } from "../../components/ui/basics"; +import { PromotedProps, namesSomething } from "../../components/config/shared"; +import { timeAgo, timeUntil } from "../../lib/dates"; +import { StatusBadge } from "./shared"; + +/** + * The whole chain is worth asking for only when the row says there is something in it. Both + * facts come back with every exchange row, so an exchange that was never retried and is not + * itself a retry — most of them — costs no request at all. + */ +export const hasRetryChain = (x: { retryFor: string | null; hasRetry: boolean }) => + x.retryFor !== null || x.hasRetry; + +export const retryTreeQuery = (id: string) => ({ + queryKey: keys.exchanges.retryTree(id), + queryFn: () => api.getRetryTree(id), + /** + * A chain is settled history above the exchange asked about and only ever grows below it, so + * a held copy cannot be wrong about what it shows — at worst it is missing an attempt someone + * has just started, which invalidating on retry covers. + */ + staleTime: 60_000, +}); + +/** The end of the chain below `fromId` — the attempt a retry would actually run. */ +export function newestAttempt(tree: RetryTree, fromId: string): RetryTreeNode | null { + let current = tree.attempts.find((a) => a.id === fromId) ?? null; + if (!current) return null; + + for (;;) { + // Newest first, so a chain that forked before one-retry-per-exchange was enforced resolves + // the same way the backend resolves it. + const children = tree.attempts + .filter((a) => a.retryFor === current!.id) + .sort((a, b) => b.startedOn.localeCompare(a.startedOn)); + if (children.length === 0) return current; + current = children[0]; + } +} + +const childrenOf = (tree: RetryTree, id: string | null) => + tree.attempts + .filter((a) => a.retryFor === id) + .sort((a, b) => a.startedOn.localeCompare(b.startedOn)); + +function Attempt({ + node, + tree, + currentId, + depth, +}: { + node: RetryTreeNode; + tree: RetryTree; + currentId: string; + depth: number; +}) { + const isCurrent = node.id === currentId; + const children = childrenOf(tree, node.id); + const navigate = useNavigate(); + + return ( + <> +
  • navigate(`/exchanges?ids=${encodeURIComponent(node.id)}`)} + className={`flex flex-wrap items-center gap-x-2 gap-y-1 rounded-md px-2 py-1.5 ${ + isCurrent ? "bg-ink-100/70" : "cursor-pointer hover:bg-ink-50" + }`} + style={{ marginLeft: depth * 14 }} + > + {/* The attempt number carries the link, rather than the identity on the right: promoted + properties can open a panel of their own, and a button inside a link is neither. */} + {isCurrent ? ( + + Attempt {depth + 1} + + ) : ( + + Attempt {depth + 1} + + )} + + + {node.retryFor === null ? "Original" : node.manualRetry ? "By hand" : "Auto"} + + {/* Without this, a fork reads as a mistake: two branches put two different exchanges at + the same depth, so the same attempt number appears twice and looks like one exchange + retried twice over. Only reachable in exchanges retried before the rule existed. */} + {children.length > 1 && ( + + {children.length} retries from here + + )} + {node.scheduledRetryOn && ( + + Auto-retry {timeUntil(node.scheduledRetryOn)} + + )} + + {timeAgo(node.startedOn)} + + + {/* Named the way the exchange list names a row — the promoted properties are what + someone recognises an exchange by, and the id falls back in when there are none. */} + + {namesSomething(node.promotedProperties) ? ( + + ) : ( + {node.id} + )} + {isCurrent && ( + + You are here + + )} + +
    • + {children.map((child) => ( + + ))} + + ); +} + +const ordinal = (n: number) => { + const names = ["first", "second", "third", "fourth", "fifth"]; + return names[n - 1] ?? `${n}th`; +}; + +/** + * Every attempt made at one piece of work, in order, with the exchange being looked at marked. + * An exchange is retried at most once, so this reads as a chain; exchanges retried before that + * rule was enforced can fork, and those show as branches rather than being hidden. + */ +export function RetryChain({ id }: { id: string }) { + const { data: tree, isLoading } = useQuery(retryTreeQuery(id)); + + // A line rather than a spinner block: this sits inside an already-rendered drawer, and it is + // usually filled in before anyone looks at it — the row is prefetched on hover. + if (isLoading) + return

    Loading the retry chain…

    ; + if (!tree || tree.attempts.length < 2) return null; + + const root = tree.attempts.find((a) => a.id === tree.rootId) ?? tree.attempts[0]; + + return ( +
    +

    + Retry chain · {tree.attempts.length} attempts +

    +
      + +
    + {tree.truncated && ( +

    + Only the attempts nearest this one are shown — the chain is longer than this view walks. +

    + )} +
    + ); +} diff --git a/SW.Bitween.Web/ClientApp/src/pages/exchanges/shared.tsx b/SW.Bitween.Web/ClientApp/src/pages/exchanges/shared.tsx index 57559fb6..bb4313f4 100644 --- a/SW.Bitween.Web/ClientApp/src/pages/exchanges/shared.tsx +++ b/SW.Bitween.Web/ClientApp/src/pages/exchanges/shared.tsx @@ -1,7 +1,8 @@ import { useState } from "react"; -import { Check, Copy } from "lucide-react"; -import type { ExchangeRow, ExchangeStatus } from "../../api"; +import { ArrowRight, Check, Copy } from "lucide-react"; +import type { BulkRetryPlan, ExchangeRow, ExchangeStatus } from "../../api"; import { Badge, Button } from "../../components/ui/basics"; +import { PromotedProps, namesSomething } from "../../components/config/shared"; import { Checkbox } from "../../components/ui/forms"; import { Dialog } from "../../components/ui/overlays"; @@ -112,45 +113,171 @@ export function XchangeId({ id, className = "" }: { id: string; className?: stri } /** - * Shared confirm for single and bulk retries — carries the "reset adapter - * properties" choice that decides whether the retry re-resolves config. + * How the plan lists name an exchange: the promoted properties the exchange list names it by, + * and enough of the id to tell two of them apart. + * + * The id stays because a substitution puts two exchanges side by side, and an information type + * whose promoted paths resolved to nothing gives both of them the same chips — "trackingNo= → + * trackingNo=" says which fields exist and nothing about which exchanges these are. + */ +function ExchangeIdentity({ + id, + properties, +}: { + id: string; + properties: Record | null; +}) { + // Properties that carry no values name nothing, so they are left out entirely rather than + // shown as a row of empty chips next to an identical row of empty chips. + if (!namesSomething(properties)) + return ( + + {id} + + ); + + return ( + + + + {id.slice(0, 8)}… + + + ); +} + +/** + * Shared confirm for single and bulk retries — carries the "reset adapter properties" choice + * that decides whether the retry re-resolves config. + * + * A bulk retry also passes the `plan` the server worked out for the same selection, because a + * selection is rarely just itself: exchanges already retried hand over to their newest attempt, + * ones that have since succeeded drop out, and two selections in one chain come to the same + * attempt. All of that is shown before anyone commits, since a retry cannot be taken back. */ export function RetryDialog({ count, + plan, + planLoading = false, busy, onConfirm, + onResetChange, onClose, }: { count: number; + /** Bulk retries only — a single retry has nothing to resolve. */ + plan?: BulkRetryPlan | null; + planLoading?: boolean; busy: boolean; onConfirm: (reset: boolean) => void; + /** + * Bulk retries only: the plan depends on this choice — re-resolving properties is impossible + * for an exchange whose subscription is gone — so the caller has to be able to ask again. + */ + onResetChange?: (reset: boolean) => void; onClose: () => void; }) { const [reset, setReset] = useState(false); + const bulk = count !== 1; + const nothingToDo = plan != null && !plan.overLimit && plan.willRetry === 0; + return (
    -

    - The original input document{count === 1 ? "" : "s"} will run through the pipeline again as{" "} - {count === 1 ? "a new exchange" : "new exchanges"}. - {count > 1 && " Exchanges that already have a pending auto-retry are skipped."} -

    - setReset(e.target.checked)} - /> + {plan?.overLimit ? ( +

    + {plan.selected.toLocaleString()} exchanges match this filter, which is more than the{" "} + {plan.limit.toLocaleString()} a single retry will carry out. Narrow the filter — by + status, partner or date — and retry the rest after. +

    + ) : ( + <> +

    + {bulk && planLoading + ? "Working out what will run…" + : plan + ? plan.willRetry === 0 + ? "Nothing here can be retried." + : `${plan.willRetry.toLocaleString()} ${ + plan.willRetry === 1 ? "exchange" : "exchanges" + } will run again — the original input document goes back through the pipeline as a new exchange.` + : `The original input document${count === 1 ? "" : "s"} will run through the pipeline again as ${ + count === 1 ? "a new exchange" : "new exchanges" + }.`} +

    + + {plan != null && plan.substituted.length > 0 && ( +
    +

    + {plan.substituted.length.toLocaleString()} of + these {plan.substituted.length === 1 ? "has" : "have"} already been retried. An + exchange is only retried once, so{" "} + {plan.substituted.length === 1 ? "its newest attempt runs" : "their newest attempts run"}{" "} + instead. +

    +
    + + Show which + +
      + {plan.substituted.map((s) => ( +
    • + + + +
      • + ))} +
    +
    +
    + )} + + {plan != null && plan.skipped.length > 0 && ( +
    +

    + {plan.skipped.length.toLocaleString()}{" "} + will be skipped. +

    +
    + + Show why + +
      + {plan.skipped.map((s) => ( +
    • + + — {s.reason} +
      • + ))} +
    +
    +
    + )} + + { + setReset(e.target.checked); + onResetChange?.(e.target.checked); + }} + /> + + )} +
    - + {!plan?.overLimit && !nothingToDo && ( + + )}
    diff --git a/SW.Bitween.Web/ClientApp/src/pages/scheduled-retries/ScheduledRetriesPage.tsx b/SW.Bitween.Web/ClientApp/src/pages/scheduled-retries/ScheduledRetriesPage.tsx index 935a0cca..2e369c64 100644 --- a/SW.Bitween.Web/ClientApp/src/pages/scheduled-retries/ScheduledRetriesPage.tsx +++ b/SW.Bitween.Web/ClientApp/src/pages/scheduled-retries/ScheduledRetriesPage.tsx @@ -157,7 +157,9 @@ export function ScheduledRetriesPage() { // Same identity rule as the Exchanges list: what it carries first, // the id only as a link out. header: "Properties", - cell: (r) => , + // With a fallback id, so a row whose promoted paths resolved to nothing keeps an + // identity of its own rather than falling back to a bare dash. + cell: (r) => , }, { header: "Information type",