Six small backlog items

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

@@ -260,7 +260,7 @@ private async Task<List<string>> ResolveSelection(XchangeBulkRetry request)
 
         // 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)
+        return await (await SelectionQuery(request))
             .Where(r => !exclude.Contains(r.Id))
             .Select(r => r.Id)
             .Take(Limit + 1)
@@ -270,7 +270,7 @@ private async Task<List<string>> ResolveSelection(XchangeBulkRetry request)
     private async Task<int> CountSelection(XchangeBulkRetry request)
     {
         var exclude = request.ExcludeIds ?? new List<string>();
-        return await SelectionQuery(request)
+        return await (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
@@ -290,10 +290,13 @@ private async Task<int> CountSelection(XchangeBulkRetry request)
     /// 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.
     /// </remarks>
-    private IQueryable<XchangeRow> SelectionQuery(XchangeBulkRetry request)
+    private async Task<IQueryable<XchangeRow>> SelectionQuery(XchangeBulkRetry request)
     {
         var searchyRequest = new SearchyRequest(request.Filter);
         searchyRequest.DatesToUtc();
+        // Async, and inside here rather than in the two callers, so a "select all matching" over a
+        // run selects exactly the rows the list showed for it.
+        await XchangeFilters.ResolveReceiveAttemptFilterAsync(searchyRequest, _dbContext);
 
         var query = from xchange in _dbContext.Set<Xchange>()
                     join result in _dbContext.Set<XchangeResult>() on xchange.Id equals result.Id into xr

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

@@ -34,6 +34,7 @@ public async Task<object> Handle(SearchyRequest searchyRequest, bool lookup = fa
                 await requestContext.EnsurePermission(dbContext, Model.Permissions.Exchanges.View, Model.Permissions.Dashboard.View);
 
             searchyRequest.DatesToUtc();
+            await XchangeFilters.ResolveReceiveAttemptFilterAsync(searchyRequest, dbContext);
             await using var dr = await dbContext.Database.BeginTransactionAsync(IsolationLevel.ReadUncommitted);
 
             var query = from xchange in dbContext.Set<Xchange>()

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

@@ -1,4 +1,5 @@
 using System.Linq;
+using System.Threading.Tasks;
 using Microsoft.EntityFrameworkCore;
 using SW.Bitween.Domain;
 using SW.Bitween.Model;
@@ -19,6 +20,58 @@ namespace SW.Bitween.Resources.Xchanges;
 /// </remarks>
 internal static class XchangeFilters
 {
+    /// <summary>
+    /// Turns a <c>ReceiveAttemptId</c> filter into the <c>Id</c> filter the rest of the pipeline
+    /// already understands, by reading the run's own record of the exchanges it created.
+    /// </summary>
+    /// <remarks>
+    /// Separate from <see cref="ApplySpecialFilters"/>, and async, because the run has to be read
+    /// before the query can be built. <c>ReceiveAttempt.ExchangeIds</c> is persisted as a single
+    /// separator-delimited string rather than an array (see <c>BitweenDbContext</c>), so no
+    /// subquery can reach inside it — and with three database providers in the solution there is
+    /// no one translation that would.
+    ///
+    /// It rewrites rather than filters here so the result stays identical to the long
+    /// <c>?ids=a,b,c</c> URL this replaces, relatives and all — the only thing that changes is that
+    /// the ids no longer have to fit in a web address.
+    /// </remarks>
+    internal static async Task ResolveReceiveAttemptFilterAsync(SearchyRequest searchyRequest,
+        BitweenDbContext dbContext)
+    {
+        var condition = searchyRequest.Conditions.FirstOrDefault();
+        if (condition == null)
+            return;
+
+        var attemptFilters = condition.Filters.Where(f => f.Field == "ReceiveAttemptId").ToList();
+        foreach (var attemptFilter in attemptFilters)
+        {
+            if (!int.TryParse(attemptFilter.Value?.ToString(), out var attemptId))
+                throw new SWValidationException("NOT_SUPPORTED",
+                    $"'{attemptFilter.Value}' is not a receive attempt id.");
+
+            var exchangeIds = await dbContext.Set<ReceiveAttempt>().AsNoTracking()
+                .Where(a => a.Id == attemptId)
+                .Select(a => a.ExchangeIds)
+                .SingleOrDefaultAsync();
+
+            if (exchangeIds == null)
+                throw new SWValidationException("NOT_FOUND",
+                    $"Run {attemptId} no longer exists. Its history may have been cleaned up.");
+
+            // A run that created nothing leaves an empty array, and the Contains branch below reads
+            // that as "matches no row" — which is the honest answer. Dropping the filter instead
+            // would widen the selection to every exchange, and bulk retry acts on whatever this
+            // selects.
+            condition.Filters.Add(new SearchyFilter
+            {
+                Field = "Id",
+                Rule = SearchyRule.Contains,
+                ValueStringArray = exchangeIds,
+            });
+            condition.Filters.Remove(attemptFilter);
+        }
+    }
+
     /// <summary>
     /// Applies the special filters and removes them from <paramref name="searchyRequest"/>, leaving
     /// the plain per-column ones for Searchy to handle.

SW.Bitween.Api/Services/XchangeService.cs

@@ -394,11 +394,22 @@ private async Task<XchangeFile> RunHandler(Xchange xchange, XchangeFile xchangeF
 
     private async Task AddFile(string xchangeId, XchangeFileType type, XchangeFile file)
     {
-        await cloudFiles.WriteTextAsync(file.Data, new WriteFileSettings
+        var key = GetFileKey(xchangeId, type);
+        try
         {
-            Public = !BitweenSettings.AreXChangeFilesPrivate,
-            Key = GetFileKey(xchangeId, type)
-        });
+            await cloudFiles.WriteTextAsync(file.Data, new WriteFileSettings
+            {
+                Public = !BitweenSettings.AreXChangeFilesPrivate,
+                Key = key
+            });
+            logger.LogDebug("Wrote the {FileType} file of xchange {XchangeId} to {Key}.", type, xchangeId, key);
+        }
+        catch (Exception ex)
+        {
+            logger.LogError(ex, "Could not write the {FileType} file of xchange {XchangeId} to {Key}.",
+                type, xchangeId, key);
+            throw;
+        }
     }
 
     public string GetFileUrl(string xchangeId, XchangeFileType type)
@@ -415,23 +426,29 @@ public string GetFileKey(string xchangeId, int? fileSize, XchangeFileType type)
     {
         if (fileSize is null or 0)
             return null;
-        var key = $"{BitweenSettings.DocumentPrefix}/{xchangeId}/{type.ToString().ToLower()}";
-        logger.LogInformation($"the file key is:'{key}'");
-        return key;
+        return $"{BitweenSettings.DocumentPrefix}/{xchangeId}/{type.ToString().ToLower()}";
     }
 
     private string GetFileKey(string xchangeId, XchangeFileType type)
     {
-        var key = $"{BitweenSettings.DocumentPrefix}/{xchangeId}/{type.ToString().ToLower()}";
-        logger.LogInformation($"the file key is:'{key}'");
-        return key;
+        return $"{BitweenSettings.DocumentPrefix}/{xchangeId}/{type.ToString().ToLower()}";
     }
 
     public async Task<string> GetFile(string xchangeId, XchangeFileType type)
     {
-        await using var cloudStream = await cloudFiles.OpenReadAsync(GetFileKey(xchangeId, type));
-        using var reader = new StreamReader(cloudStream);
-        return await reader.ReadToEndAsync();
+        var key = GetFileKey(xchangeId, type);
+        try
+        {
+            await using var cloudStream = await cloudFiles.OpenReadAsync(key);
+            using var reader = new StreamReader(cloudStream);
+            return await reader.ReadToEndAsync();
+        }
+        catch (Exception ex)
+        {
+            logger.LogError(ex, "Could not read the {FileType} file of xchange {XchangeId} from {Key}.",
+                type, xchangeId, key);
+            throw;
+        }
     }
 
     private async Task Process(XchangeMessage message)

SW.Bitween.IntegrationTests/Tests/ReceiveAttemptFilterTests.cs

@@ -0,0 +1,121 @@
+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;
+
+/// <summary>
+/// The <c>ReceiveAttemptId</c> search filter: "the exchanges one run created".
+/// </summary>
+/// <remarks>
+/// It exists so the UI can link to a run's exchanges without naming every one of them in the URL,
+/// which put a ceiling on how big a run could be linked to. The filter is resolved before the
+/// query rather than as a subquery, because <c>ReceiveAttempt.ExchangeIds</c> is stored as one
+/// separator-delimited string and no provider can see inside it.
+/// </remarks>
+[Collection("Bitween")]
+public class ReceiveAttemptFilterTests(BitweenFixture fixture)
+{
+    private static SearchyRequest ForAttempt(int attemptId) =>
+        new("ReceiveAttemptId", SearchyRule.EqualsTo, attemptId.ToString());
+
+    /// <summary>A subscription with one exchange, and a run that claims to have created it.</summary>
+    private static async Task<(int AttemptId, string XchangeId)> ARunThatCreated(
+        BitweenDbContext db, XchangeService xs, string name)
+    {
+        var doc = new Document(null, name, DocumentFormat.Json);
+        db.Set<Document>().Add(doc);
+        await db.SaveChangesAsync();
+
+        var sub = new Subscription(name, doc.Id) { Inactive = false };
+        db.Set<Subscription>().Add(sub);
+        await db.SaveChangesAsync();
+
+        var xchange = await xs.CreateXchange(sub, new XchangeFile("{}"));
+        await db.SaveChangesAsync();
+
+        var attempt = new ReceiveAttempt
+        {
+            SubscriptionId = sub.Id,
+            StartedOn = System.DateTime.UtcNow,
+            FinishedOn = System.DateTime.UtcNow,
+            Outcome = ReceiveOutcome.Received,
+            ExchangeIds = [xchange.Id],
+        };
+        db.Set<ReceiveAttempt>().Add(attempt);
+        await db.SaveChangesAsync();
+
+        return (attempt.Id, xchange.Id);
+    }
+
+    private static async Task<string[]> Search(AsyncServiceScope scope, SearchyRequest request)
+    {
+        var db = scope.ServiceProvider.GetRequiredService<BitweenDbContext>();
+        var xs = scope.ServiceProvider.GetRequiredService<XchangeService>();
+        var response = (SearchyResponse<XchangeRow>)await new Resources.Xchanges.Search(
+            db, xs, scope.Superuser()).Handle(request);
+        return response.Result.Select(r => r.Id).ToArray();
+    }
+
+    [Fact]
+    public async Task Returns_the_run_s_own_exchanges_and_nobody_else_s()
+    {
+        await using var scope = fixture.CreateScope();
+        var db = scope.ServiceProvider.GetRequiredService<BitweenDbContext>();
+        var xs = scope.ServiceProvider.GetRequiredService<XchangeService>();
+
+        var mine = await ARunThatCreated(db, xs, "Attempt filter mine");
+        // A second run, so "returns everything" cannot pass by accident.
+        var theirs = await ARunThatCreated(db, xs, "Attempt filter theirs");
+
+        var found = await Search(scope, ForAttempt(mine.AttemptId));
+
+        Assert.Equal([mine.XchangeId], found);
+        Assert.DoesNotContain(theirs.XchangeId, found);
+    }
+
+    [Fact]
+    public async Task A_run_that_created_nothing_matches_nothing_rather_than_everything()
+    {
+        await using var scope = fixture.CreateScope();
+        var db = scope.ServiceProvider.GetRequiredService<BitweenDbContext>();
+        var xs = scope.ServiceProvider.GetRequiredService<XchangeService>();
+
+        // Exchanges exist; this run just is not responsible for any of them. Dropping the filter
+        // instead of honouring it would return all of them — and bulk retry acts on this selection.
+        await ARunThatCreated(db, xs, "Attempt filter bystander");
+
+        var empty = new ReceiveAttempt
+        {
+            SubscriptionId = 1,
+            StartedOn = System.DateTime.UtcNow,
+            FinishedOn = System.DateTime.UtcNow,
+            Outcome = ReceiveOutcome.NoNewData,
+            ExchangeIds = [],
+        };
+        db.Set<ReceiveAttempt>().Add(empty);
+        await db.SaveChangesAsync();
+
+        Assert.Empty(await Search(scope, ForAttempt(empty.Id)));
+    }
+
+    [Fact]
+    public async Task A_run_that_no_longer_exists_is_refused()
+    {
+        await using var scope = fixture.CreateScope();
+
+        // Refused rather than ignored, for the same reason: a silently dropped filter widens the
+        // selection to every exchange. ReceiveAttemptCleanupJob does delete old runs, so a stale
+        // link is a real case, not a hypothetical one.
+        var ex = await Assert.ThrowsAsync<SWValidationException>(
+            () => Search(scope, ForAttempt(int.MaxValue)));
+
+        Assert.Contains(ex.Validations, v => v.Key == "NOT_FOUND");
+    }
+}

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

@@ -181,6 +181,10 @@ function buildExchangeFilters(query: ExchangeQuery): URLSearchParams {
     const ids = query.ids.split(/[\s,|]+/).filter(Boolean);
     params.append("filter", `Id:4:text|${ids.join("|")}`);
   }
+  // The server turns this into the same Id filter above, from the run's own record of what it
+  // created — so a run of any size is one short parameter rather than every id in the URL.
+  if (query.receiveAttemptId !== undefined)
+    params.append("filter", `ReceiveAttemptId:1:${query.receiveAttemptId}`);
   if (query.correlationId?.trim()) params.append("filter", `CorrelationId:1:${query.correlationId.trim()}`);
   if (query.latest) params.append("filter", "LatestOnly:1:true");
   // PromotedPropertiesRaw is stored as "key:value,key:value", so prefixing the key turns

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

@@ -1126,6 +1126,12 @@ export interface ExchangeQuery {
   informationTypeId?: number;
   /** Comma/pipe/newline separated; matches id, retryFor OR aggregationXchangeId. */
   ids?: string;
+  /**
+   * The exchanges one run of a scheduled job created. The server reads the run's own list of
+   * them, so this says the same thing as spelling every id into `ids` — at any size, and
+   * without putting them all in the URL.
+   */
+  receiveAttemptId?: number;
   correlationId?: string;
   /**
    * Only the newest attempt of each retry chain. A chain is one piece of work however many

SW.Bitween.Web/ClientApp/src/components/config/RetryPolicyDialog.tsx

@@ -0,0 +1,74 @@
+import { useState, type FormEvent } from "react";
+import { useNavigate } from "react-router";
+import { useMutation, useQueryClient } from "@tanstack/react-query";
+import { api } from "../../api";
+import { Button, FormError } from "../ui/basics";
+import { Field, TextInput } from "../ui/forms";
+import { Dialog } from "../ui/overlays";
+import { keys } from "../../api/queryKeys";
+
+/**
+ * Creating a retry policy, as one component.
+ *
+ * A policy is born with nothing but a name — its budget, groups and alert routing are
+ * configured on its own page, which is too much to ask for inside a dialog. So the two
+ * callers want opposite things once it exists, and `onCreated` is what tells them apart:
+ * the policies list sends you to the new policy's page to carry on configuring it, while
+ * a subscription attaching one wants to stay where it is with the policy now selected.
+ */
+export function CreateRetryPolicyDialog({
+  onClose,
+  /** Given, the new policy is handed back and the dialog closes. Omitted, we navigate to it. */
+  onCreated,
+}: {
+  onClose: () => void;
+  onCreated?: (id: number) => void;
+}) {
+  const navigate = useNavigate();
+  const queryClient = useQueryClient();
+  const [name, setName] = useState("");
+
+  const create = useMutation({
+    mutationFn: () => api.createRetryPolicy({ name }),
+    onSuccess: (policy) => {
+      void queryClient.invalidateQueries({ queryKey: keys.retryPolicies.all });
+      if (onCreated) {
+        onCreated(policy.id);
+        onClose();
+        return;
+      }
+      // replace, not push: the ?new=1 entry this dialog opened on is still behind us,
+      // and Back onto it would reopen the form that was just submitted.
+      navigate(`/retry-policies/${policy.id}`, { replace: true });
+    },
+  });
+
+  const submit = (e: FormEvent) => {
+    e.preventDefault();
+    create.mutate();
+  };
+
+  return (
+    <Dialog title="New retry policy" onClose={onClose}>
+      <form onSubmit={submit} className="space-y-4">
+        <Field label="Name" htmlFor="nrp-name" hint="Groups and budgets are added on the policy's page.">
+          <TextInput
+            id="nrp-name"
+            required
+            autoFocus
+            value={name}
+            onChange={(e) => setName(e.target.value)}
+            placeholder="e.g. Transient failures"
+          />
+        </Field>
+        <FormError>{create.error?.message}</FormError>
+        <div className="flex justify-end gap-2">
+          <Button onClick={onClose}>Cancel</Button>
+          <Button type="submit" variant="primary" busy={create.isPending}>
+            Create policy
+          </Button>
+        </div>
+      </form>
+    </Dialog>
+  );
+}