feat: retry a selection past 500 in rounds instead of refusing it

SW.Bitween.Api/Resources/Xchanges/BulkRetry.cs

@@ -32,13 +32,13 @@ public async Task<object> Handle(XchangeBulkRetry request)
         {
             await _requestContext.EnsurePermission(_dbContext, Model.Permissions.Exchanges.Operate);
 
-            var prepared = await new BulkRetryPlanner(_dbContext).Prepare(request);
+            var prepared = await new BulkRetryPlanner(_dbContext).Prepare(request, BulkRetryPlanner.Limit);
             var plan = prepared.Plan;
 
             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.");
+                    $"in one go. Send the preview's targets back in rounds of {BulkRetryPlanner.Limit}.");
 
             var retried = 0;
             foreach (var id in prepared.Targets)

SW.Bitween.Api/Resources/Xchanges/BulkRetryPlanner.cs

@@ -23,10 +23,21 @@ 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.
+    /// about the database. A bigger selection is retried in rounds of this many.
     /// </summary>
     internal const int Limit = 500;
 
+    /// <summary>
+    /// The largest selection a preview works out. Up to it, the plan lists every attempt that will
+    /// run, and the caller sends them back in rounds of <see cref="Limit"/>. Past it the caller is
+    /// asked to narrow the filter.
+    /// </summary>
+    /// <remarks>
+    /// Rounds rather than raising <see cref="Limit"/>: each round is its own request, so none of them
+    /// runs long, and a round that fails leaves the ones before it done.
+    /// </remarks>
+    internal const int MaxSelection = 5000;
+
     /// <summary>
     /// Guards the walk to the newest attempt against a cycle in the data, the same way
     /// <see cref="RetryTree"/> does. No honest chain approaches it.
@@ -44,24 +55,29 @@ internal sealed class Prepared
         internal List<string> Targets { get; init; } = new List<string>();
     }
 
-    internal async Task<Prepared> Prepare(XchangeBulkRetry request)
+    /// <param name="upTo">
+    /// The largest selection to work out: <see cref="Limit"/> for a retry, <see cref="MaxSelection"/>
+    /// for a preview.
+    /// </param>
+    internal async Task<Prepared> Prepare(XchangeBulkRetry request, int upTo)
     {
         var selection = new XchangeSelection(_dbContext);
-        var selected = await selection.Resolve(request, Limit);
+        var selected = await selection.Resolve(request, upTo);
 
-        if (selected.Count > Limit)
+        if (selected.Count > upTo)
             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
+                    // upTo + 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 selection.Count(request),
                     Limit = Limit,
+                    MaxSelection = MaxSelection,
                     OverLimit = true
                 }
             };
@@ -70,8 +86,10 @@ internal async Task<Prepared> Prepare(XchangeBulkRetry request)
         var state = await ReadState(newestAttempt.Values.Distinct().ToList());
         var reset = request.Reset;
 
-        var plan = new XchangeBulkRetryPlan { Selected = selected.Count, Limit = Limit };
+        var plan = new XchangeBulkRetryPlan { Selected = selected.Count, Limit = Limit, MaxSelection = MaxSelection };
         var targets = new List<string>();
+        // The list keeps the order; the set answers "already in it" without walking up to MaxSelection entries.
+        var targeted = new HashSet<string>();
 
         foreach (var selectedId in selected)
         {
@@ -93,11 +111,12 @@ internal async Task<Prepared> Prepare(XchangeBulkRetry request)
 
             // 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))
+            if (targeted.Add(targetId))
                 targets.Add(targetId);
         }
 
         plan.WillRetry = targets.Count;
+        plan.Targets = targets;
         plan.Properties = await ReadPromotedProperties(plan);
         return new Prepared { Plan = plan, Targets = targets };
     }

SW.Bitween.Api/Resources/Xchanges/BulkRetryPreview.cs

@@ -12,6 +12,11 @@ namespace SW.Bitween.Resources.Xchanges;
 /// 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.
+/// <para>
+/// It works out a selection past what one retry carries out, up to
+/// <see cref="BulkRetryPlanner.MaxSelection"/>, so the caller can retry it in rounds from the plan's
+/// targets.
+/// </para>
 /// </remarks>
 [HandlerName("bulkretrypreview")]
 public class BulkRetryPreview : ICommandHandler<XchangeBulkRetry, object>
@@ -31,7 +36,7 @@ public async Task<object> Handle(XchangeBulkRetry request)
         // 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);
+        var prepared = await new BulkRetryPlanner(_dbContext).Prepare(request, BulkRetryPlanner.MaxSelection);
         return prepared.Plan;
     }
 }

SW.Bitween.IntegrationTests/Tests/RetryChainTests.cs

@@ -317,6 +317,74 @@ public async Task BulkRetryPreview_describes_the_same_work_without_doing_it()
         Assert.False(await db.Set<Xchange>().AnyAsync(x => x.RetryFor == second.Id));
     }
 
+    /// <summary>
+    /// A selection bigger than one retry is sent back in rounds of the preview's targets: the attempts
+    /// themselves, already moved to the end of their chains, so a round retries exactly those.
+    /// </summary>
+    [Fact]
+    public async Task The_preview_lists_the_attempts_a_round_retries()
+    {
+        await using var scope = _fixture.CreateScope();
+        var db = scope.ServiceProvider.GetRequiredService<BitweenDbContext>();
+        var xs = scope.ServiceProvider.GetRequiredService<XchangeService>();
+        var ctx = scope.Superuser();
+
+        var (sub, retried) = await FailedXchange(db, xs, "Targets Doc");
+        var newest = await RetryOnce(db, xs, ctx, retried.Id);
+        var other = await xs.CreateXchange(sub, new XchangeFile("{}"));
+        await db.SaveChangesAsync();
+        db.Set<XchangeResult>().Add(new XchangeResult(other.Id, null, null, exception: "boom"));
+        await db.SaveChangesAsync();
+
+        var plan = (XchangeBulkRetryPlan)await new Resources.Xchanges.BulkRetryPreview(db, ctx)
+            .Handle(new XchangeBulkRetry { Ids = [retried.Id, newest.Id, other.Id] });
+
+        Assert.Equal(new[] { newest.Id, other.Id }, plan.Targets);
+
+        var round = (XchangeBulkRetryPlan)await new Resources.Xchanges.BulkRetry(db, ctx, xs)
+            .Handle(new XchangeBulkRetry { Ids = plan.Targets });
+        await db.SaveChangesAsync();
+
+        Assert.Equal(2, round.WillRetry);
+        Assert.Empty(round.Substituted);
+        Assert.True(await db.Set<Xchange>().AnyAsync(x => x.RetryFor == newest.Id));
+        Assert.True(await db.Set<Xchange>().AnyAsync(x => x.RetryFor == other.Id));
+    }
+
+    /// <summary>
+    /// Past what one retry carries out, the preview still works the selection out so it can be
+    /// retried in rounds, while the retry itself takes no more than one round. Ids that don't exist
+    /// are enough to count with: the size is checked before any exchange is read.
+    /// </summary>
+    [Fact]
+    public async Task A_selection_past_one_retry_is_planned_whole_but_retried_a_round_at_a_time()
+    {
+        await using var scope = _fixture.CreateScope();
+        var db = scope.ServiceProvider.GetRequiredService<BitweenDbContext>();
+        var xs = scope.ServiceProvider.GetRequiredService<XchangeService>();
+        var ctx = scope.Superuser();
+        var preview = new Resources.Xchanges.BulkRetryPreview(db, ctx);
+        var limits = (XchangeBulkRetryPlan)await preview.Handle(new XchangeBulkRetry { Ids = [] });
+        Assert.True(limits.MaxSelection > limits.Limit);
+
+        static System.Collections.Generic.List<string> Ids(int count) =>
+            Enumerable.Range(0, count).Select(_ => Guid.NewGuid().ToString("N")).ToList();
+
+        var pastOneRetry = Ids(limits.Limit + 1);
+        var plan = (XchangeBulkRetryPlan)await preview.Handle(new XchangeBulkRetry { Ids = pastOneRetry });
+        Assert.False(plan.OverLimit);
+        Assert.Equal(limits.Limit + 1, plan.Skipped.Count);
+
+        var refused = await Assert.ThrowsAsync<SWValidationException>(() =>
+            new Resources.Xchanges.BulkRetry(db, ctx, xs).Handle(new XchangeBulkRetry { Ids = pastOneRetry }));
+        Assert.Contains(refused.Validations, v => v.Key == "TOO_MANY");
+
+        var pastAnyRetry = (XchangeBulkRetryPlan)await preview.Handle(
+            new XchangeBulkRetry { Ids = Ids(limits.MaxSelection + 1) });
+        Assert.True(pastAnyRetry.OverLimit);
+        Assert.Equal(limits.MaxSelection + 1, pastAnyRetry.Selected);
+    }
+
     /// <summary>
     /// 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

SW.Bitween.Sdk/Model/Xchange.cs

@@ -113,16 +113,29 @@
         /// <summary>Distinct exchanges that will be (or were) retried.</summary>
         public int WillRetry { get; set; }
 
-        /// <summary>The largest selection this endpoint will carry out in one request.</summary>
+        /// <summary>
+        /// The largest selection one retry request carries out. A bigger one is retried in rounds of
+        /// this many, from <see cref="Targets"/>.
+        /// </summary>
         public int Limit { get; set; }
 
+        /// <summary>The largest selection a preview works out; past it, narrow the selection.</summary>
+        public int MaxSelection { get; set; }
+
         /// <summary>
-        /// <c>true</c> when the selection is past <see cref="Limit"/>. 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.
+        /// <c>true</c> when the selection is past <see cref="MaxSelection"/> for a preview, or past
+        /// <see cref="Limit"/> for a retry. 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.
         /// </summary>
         public bool OverLimit { get; set; }
 
+        /// <summary>
+        /// The attempts that will run: each selection's newest attempt, once per chain, minus the
+        /// skipped. Sent back as <c>ids</c> in rounds of <see cref="Limit"/> when there are
+        /// more than one retry carries out.
+        /// </summary>
+        public List<string> Targets { get; set; } = new List<string>();
+
         /// <summary>
         /// Selections that had already been retried, and the later attempt standing in for each.
         /// </summary>
@@ -233,23 +246,23 @@
 
     public class XchangeRow
     {
         public string Id { get; set; }
         public int? SubscriptionId { get; set; }
         public string SubscriptionName { get; set; }
         public int DocumentId { get; set; }
         public string DocumentName { get; set; }
         public string HandlerId { get; set; }
         public string MapperId { get; set; }
         public string[] References { get; set; }
         public bool? Status { get; set; }
         public int StatusFilter { get; set; }
         public string StatusString { get; set; }
         public string Exception { get; set; }
         public DateTime? DeliveredOn { get; set; }
         public DateTime? FinishedOn { get; set; }
         public DateTime? AggregatedOn { get; set; }
         public DateTime StartedOn { get; set; }
         public string InputFileName { get; set; }
         public int InputFileSize { get; set; }
         public string InputFileHash { get; set; }
         public string OutputFileName { get; set; }

SW.Bitween.Web/ClientApp/src/api/http/exchanges.ts

@@ -79,7 +79,9 @@ interface RawBulkRetryPlan {
   selected: number;
   willRetry: number;
   limit: number;
+  maxSelection: number;
   overLimit: boolean;
+  targets: string[] | null;
   substituted: { selectedId: string; retryId: string }[] | null;
   skipped: { id: string; reason: string }[] | null;
   properties: Record<string, Record<string, string | null>> | null;
@@ -262,7 +264,9 @@ async function postBulkRetry(
     selected: plan.selected,
     willRetry: plan.willRetry,
     limit: plan.limit,
+    maxSelection: plan.maxSelection,
     overLimit: plan.overLimit,
+    targets: plan.targets ?? [],
     substituted: plan.substituted ?? [],
     skipped: plan.skipped ?? [],
     properties: plan.properties ?? {},

SW.Bitween.Web/ClientApp/src/api/types.ts

@@ -1544,9 +1544,14 @@ export interface ExportFilesCheck {
 export interface BulkRetryPlan {
   selected: number;
   willRetry: number;
+  /** The most one retry request carries out; a bigger selection goes in rounds of this many. */
   limit: number;
-  /** True when the selection is past `limit`; nothing else is filled in. */
+  /** The most a preview works out. */
+  maxSelection: number;
+  /** True when the selection is past `maxSelection` (or, for a retry, past `limit`); nothing else is filled in. */
   overLimit: boolean;
+  /** The attempts that will run - what the rounds send back as ids when `selected` is past `limit`. */
+  targets: string[];
   substituted: RetrySubstitution[];
   skipped: RetrySkip[];
   /**

SW.Bitween.Web/ClientApp/src/pages/exchanges/ExchangesPage.tsx

@@ -1,9 +1,10 @@
-import { useCallback, useEffect, useMemo, useState } from "react";
+import { useCallback, useEffect, useMemo, useRef, useState } from "react";
 import { Link, useSearchParams } from "react-router";
 import { keepPreviousData, useMutation, useQuery, useQueryClient } from "@tanstack/react-query";
 import { ArrowUp, Download, Pause, Play, Plus, RefreshCw, RotateCcw, TriangleAlert, X } from "lucide-react";
 import {
   api,
+  type BulkRetryPlan,
   type BulkRetrySelection,
   type ExchangeStatus,
   type ExportFilesCheck,
@@ -304,20 +305,74 @@ export function ExchangesPage() {
     staleTime: 30_000,
   });
 
+  /** How far a retry in rounds has got. */
+  const [bulkProgress, setBulkProgress] = useState<{ done: number; total: number } | null>(null);
+  /** Set when the confirm is closed mid-way: the rounds stop after the one under way. */
+  const stopRounds = useRef(false);
+  /**
+   * The run the open confirm belongs to. Closing it moves this on, so a run that ends afterwards
+   * reports what it did but can't close a confirm opened since, or clear a selection made since.
+   */
+  const bulkRun = useRef(0);
+
+  /**
+   * A selection past what one request carries out: the preview listed every attempt that will run,
+   * and they go back as ids, a round at a time. No request runs long, and a round that fails leaves
+   * the ones before it done. The attempts, not the selection, so a round can't reach past them to an
+   * attempt an earlier round just made.
+   */
+  async function retryInRounds(planned: BulkRetryPlan, reset: boolean) {
+    const done = { ...planned, willRetry: 0, skipped: [...planned.skipped], stoppedAt: null as number | null };
+    const total = planned.targets.length;
+    stopRounds.current = false;
+    setBulkProgress({ done: 0, total });
+    for (let at = 0; at < total; at += planned.limit) {
+      if (stopRounds.current) {
+        done.stoppedAt = at;
+        break;
+      }
+      let round: BulkRetryPlan;
+      try {
+        round = await api.bulkRetryExchanges({ ids: planned.targets.slice(at, at + planned.limit) }, { reset });
+      } catch (e) {
+        throw new Error(
+          `${done.willRetry.toLocaleString()} of ${total.toLocaleString()} were retried, then a round failed` +
+            ` (${e instanceof Error ? e.message : String(e)}). The rest were not retried.`,
+        );
+      }
+      done.willRetry += round.willRetry;
+      done.skipped.push(...round.skipped);
+      setBulkProgress({ done: Math.min(at + planned.limit, total), total });
+    }
+    return done;
+  }
+
   const bulkRetry = useMutation({
-    mutationFn: (reset: boolean) => api.bulkRetryExchanges(selection, { reset }),
-    onSuccess: (done) => {
-      setBulkConfirm(false);
-      setBulkReset(false);
-      clearSelection();
+    mutationFn: (reset: boolean) =>
+      plan && plan.selected > plan.limit
+        ? retryInRounds(plan, reset)
+        : api.bulkRetryExchanges(selection, { reset }).then((done) => ({ ...done, stoppedAt: null })),
+    onMutate: () => ({ run: ++bulkRun.current }),
+    onSuccess: (done, _reset, started) => {
+      setBulkProgress(null);
+      if (started?.run === bulkRun.current) {
+        setBulkConfirm(false);
+        setBulkReset(false);
+        clearSelection();
+      }
       setBulkResult(
         `${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` : "") +
+          (done.stoppedAt != null
+            ? `. Stopped after ${done.stoppedAt.toLocaleString()} of ${done.targets.length.toLocaleString()}; the rest were not retried`
+            : "") +
           ".",
       );
       void queryClient.invalidateQueries({ queryKey: keys.exchanges.all });
     },
+    // The list changed as far as the rounds got.
+    onError: () => void queryClient.invalidateQueries({ queryKey: keys.exchanges.all }),
   });
 
   // Asked first, so someone hears that some exchanges are past the files storage keeps before choosing to
@@ -654,10 +709,18 @@ export function ExchangesPage() {
           plan={plan ?? null}
           planLoading={planLoading}
           busy={bulkRetry.isPending}
+          progress={bulkProgress}
+          error={bulkRetry.error?.message ?? null}
           onConfirm={(reset) => bulkRetry.mutate(reset)}
           onResetChange={setBulkReset}
           onClose={() => {
+            stopRounds.current = true;
+            bulkRun.current++;
             setBulkConfirm(false);
+            setBulkProgress(null);
+            // Not while rounds are still under way: Retry would be offered again, and a second run
+            // beside the first would retry what the first just did.
+            if (bulkRetry.isError) bulkRetry.reset();
             // The dialog starts unticked each time it opens, so the mirrored copy has to as well.
             setBulkReset(false);
           }}