diff --git a/SW.Bitween.Api/Controllers/BackgroundRequestsController.cs b/SW.Bitween.Api/Controllers/BackgroundRequestsController.cs
new file mode 100644
index 00000000..24ae6f55
--- /dev/null
+++ b/SW.Bitween.Api/Controllers/BackgroundRequestsController.cs
@@ -0,0 +1,168 @@
+using System;
+using System.Linq;
+using System.Security.Claims;
+using System.Threading.Tasks;
+using Microsoft.AspNetCore.Authorization;
+using Microsoft.AspNetCore.Http;
+using Microsoft.AspNetCore.Mvc;
+using Microsoft.EntityFrameworkCore;
+using SW.Bitween.Domain;
+using SW.Bitween.Services.Background;
+using SW.PrimitiveTypes;
+
+namespace SW.Bitween.Controllers;
+
+///
+/// The signed-in user's background requests (see ): which
+/// there are, the answer of one, and cancelling one. Nobody sees anyone else's.
+///
+///
+/// A controller rather than CqApi resources, because an answer goes back exactly as the request it
+/// stands in for would have sent it: its status, its content type, its bytes.
+///
+[ApiController]
+[Authorize]
+[Route("api/background")]
+public class BackgroundRequestsController(BitweenDbContext db, BackgroundRequests requests, ICloudFilesService cloudFiles,
+ BackgroundDownloads downloads) : ControllerBase
+{
+ /// Those still running, and those that finished within the half hour answers are kept.
+ [HttpGet]
+ public async Task List()
+ {
+ var owner = Owner;
+ if (owner == null) return Unauthorized();
+ await requests.ForgetExpiredAsync(db);
+
+ var unfinished = await db.Set()
+ .Where(r => r.Owner == owner &&
+ (r.Status == BackgroundRequestStatus.Waiting || r.Status == BackgroundRequestStatus.Running))
+ .ToListAsync();
+ await BackgroundRequests.EndStoppedAsync(db, unfinished);
+
+ var now = DateTime.UtcNow;
+ var since = now - BackgroundRequests.KeptFor;
+ var rows = await db.Set().AsNoTracking()
+ .Where(r => r.Owner == owner && r.StartedOn > now.AddDays(-1) && (r.FinishedOn == null || r.FinishedOn > since))
+ .OrderByDescending(r => r.StartedOn)
+ .Take(20)
+ .Select(r => new
+ {
+ r.Id, r.Label, r.Page, r.Path, r.Status, r.StartedOn, r.FinishedOn, r.ResultStatus, r.Done, r.Total,
+ r.Summary
+ })
+ .ToListAsync();
+
+ return Ok(rows.Select(r => new
+ {
+ r.Id,
+ r.Label,
+ r.Page,
+ r.Path,
+ Status = r.Status.ToString(),
+ r.StartedOn,
+ r.FinishedOn,
+ r.ResultStatus,
+ r.Done,
+ r.Total,
+ r.Summary,
+ }));
+ }
+
+ /// The answer, as the request would have given it; or why there is none.
+ [HttpGet("{id}/result")]
+ public async Task Result(string id)
+ {
+ var owner = Owner;
+ if (owner == null) return Unauthorized();
+ await requests.ForgetExpiredAsync(db);
+
+ var row = await db.Set().FirstOrDefaultAsync(r => r.Id == id && r.Owner == owner);
+ if (row == null)
+ return Refusal(StatusCodes.Status404NotFound, "BACKGROUND_EXPIRED",
+ "This answer is no longer kept. Open the page again to ask again.");
+
+ if (!row.Finished) await BackgroundRequests.EndStoppedAsync(db, [row]);
+
+ // Work cancelled part way can have an answer too: what it did before it stopped.
+ if (row.Answered)
+ {
+ Response.StatusCode = row.ResultStatus ?? StatusCodes.Status200OK;
+ if (row.ResultContentType != null) Response.ContentType = row.ResultContentType;
+ if (row.ResultDisposition != null) Response.Headers.ContentDisposition = row.ResultDisposition;
+ if (row.ResultKey == null) await Response.Body.WriteAsync(row.Result ?? []);
+ else
+ {
+ await using var kept = await cloudFiles.OpenReadAsync(row.ResultKey);
+ await kept.CopyToAsync(Response.Body);
+ }
+
+ return new EmptyResult();
+ }
+
+ switch (row.Status)
+ { case BackgroundRequestStatus.Cancelled:
+ return Refusal(StatusCodes.Status410Gone, "BACKGROUND_CANCELLED", "It was cancelled.");
+ case BackgroundRequestStatus.TimedOut:
+ return Refusal(StatusCodes.Status504GatewayTimeout, "BACKGROUND_TIMED_OUT",
+ $"It went {BackgroundRequests.Limit.TotalMinutes:n0} minutes without getting further and was stopped.");
+ case BackgroundRequestStatus.Stopped:
+ return Refusal(StatusCodes.Status503ServiceUnavailable, "BACKGROUND_STOPPED",
+ "The server running it stopped before it finished, as a deploy or restart does. Ask again.");
+ default:
+ return Refusal(StatusCodes.Status409Conflict, "BACKGROUND_RUNNING", "It is still running.");
+ }
+ }
+
+ ///
+ /// A link to the file one of the caller's answered with - an export's zip - for the browser to download
+ /// the ordinary way, straight to disk. It opens for .
+ ///
+ [HttpPost("{id}/link")]
+ public async Task Link(string id)
+ {
+ var owner = Owner;
+ if (owner == null) return Unauthorized();
+ await requests.ForgetExpiredAsync(db);
+
+ var row = await db.Set().AsNoTracking().FirstOrDefaultAsync(r => r.Id == id && r.Owner == owner);
+ if (row == null)
+ return Refusal(StatusCodes.Status404NotFound, "BACKGROUND_EXPIRED", "This file is no longer kept. Export it again.");
+ if (row.ResultKey == null || row.ResultStatus is not (>= 200 and < 300))
+ return Refusal(StatusCodes.Status409Conflict, "BACKGROUND_NO_FILE", "It has no file to download.");
+ return Ok(new { url = downloads.LinkTo(id) });
+ }
+
+ /// The file a link from opens. No sign-in: the link's seal is what lets it in.
+ [AllowAnonymous]
+ [HttpGet("{id}/file")]
+ public async Task Download(string id, [FromQuery] string expires, [FromQuery] string seal)
+ {
+ // The same answer for a bad seal as for a missing file, so a link can't be used to probe.
+ if (!downloads.Opens(id, expires, seal)) return NotFound();
+ var row = await db.Set().AsNoTracking().FirstOrDefaultAsync(r => r.Id == id);
+ if (row?.ResultKey == null) return NotFound();
+
+ if (row.ResultDisposition != null) Response.Headers.ContentDisposition = row.ResultDisposition;
+ return File(await cloudFiles.OpenReadAsync(row.ResultKey), row.ResultContentType ?? "application/octet-stream");
+ }
+
+ /// Stops one of the caller's, wherever it runs.
+ [HttpDelete("{id}")]
+ public async Task Cancel(string id)
+ {
+ var owner = Owner;
+ if (owner == null) return Unauthorized();
+ return await requests.CancelAsync(db, owner, id) ? NoContent() : NotFound();
+ }
+
+ string Owner => User.FindFirst(ClaimTypes.NameIdentifier)?.Value;
+
+ /// A refusal in the shape the UI reads one: the code, with what to tell the person.
+ static IActionResult Refusal(int status, string code, string message) => new ContentResult
+ {
+ StatusCode = status,
+ ContentType = "application/json",
+ Content = System.Text.Encoding.UTF8.GetString(BackgroundRequests.Problem(code, message)),
+ };
+}
diff --git a/SW.Bitween.Api/Controllers/XchangeExportController.cs b/SW.Bitween.Api/Controllers/XchangeExportController.cs
index 22d32355..866efb3e 100644
--- a/SW.Bitween.Api/Controllers/XchangeExportController.cs
+++ b/SW.Bitween.Api/Controllers/XchangeExportController.cs
@@ -44,11 +44,24 @@ public class XchangeExportController(BitweenDbContext dbContext, RequestContext
: ControllerBase
{
///
- /// The most exchanges one zip takes - the bulk retry's ceiling, for the same reason: every file is
- /// read from storage in sequence, inside the one request.
+ /// The most exchanges one zip takes when asked directly - the bulk retry's ceiling, for the same
+ /// reason: every file is read from storage in sequence, inside the one request.
///
public const int Limit = 500;
+ ///
+ /// The most one zip takes carried on in the background, as the admin UI asks for it. The zip is built
+ /// on the node's temporary disk and kept in storage, and the browser downloads it from a link, straight
+ /// to disk; is what stops a zip of large files first.
+ ///
+ public const int MaxInBackground = 50_000;
+
+ ///
+ /// The largest zip built: past it, the exchanges not yet in it are left out, and missing.txt says so.
+ /// Well inside the 2 GB of temporary disk the Helm chart gives a node.
+ ///
+ public const long MaxBytes = 1L << 30;
+
public const string MissingList = "missing.txt";
private static readonly Regex Extension = new(@"^\.[A-Za-z0-9]{1,10}$");
@@ -56,12 +69,13 @@ public class XchangeExportController(BitweenDbContext dbContext, RequestContext
///
/// What an export of would hold, asked before it's downloaded: how many
/// exchanges it takes, and how many are older than the bucket's rule keeps their files. Refuses what
- /// the export itself would refuse.
+ /// the export in the background would refuse, past .
///
[HttpPost("check")]
public async Task Check([FromBody] XchangeFilesExport request)
{
- var (ids, refusal) = await Resolve(request);
+ // Checked against what the admin UI's export then takes, in the background.
+ var (ids, refusal) = await Resolve(request, MaxInBackground);
if (refusal != null) return refusal;
var xchanges = await dbContext.Set().AsNoTracking().Where(x => ids.Contains(x.Id)).ToListAsync();
@@ -85,7 +99,8 @@ public async Task Check([FromBody] XchangeFilesExport request)
[HttpPost]
public async Task Post([FromBody] XchangeFilesExport request)
{
- var (ids, refusal) = await Resolve(request);
+ var background = Services.Background.BackgroundRequestScope.Current;
+ var (ids, refusal) = await Resolve(request, background != null ? MaxInBackground : Limit);
if (refusal != null) return refusal;
var xchanges = await dbContext.Set().AsNoTracking()
@@ -118,8 +133,26 @@ await dbContext.RecordNowAsync(requestContext, OperatorAction.Export,
await using (var zip = await ZipArchive.CreateAsync(body, ZipArchiveMode.Create, leaveOpen: true,
entryNameEncoding: null, aborted))
{
+ // In the background, as the admin UI asks for it: says how far it has got, and a cancel
+ // ends it between exchanges - with what it took so far, and missing.txt saying so.
+ var exported = 0;
foreach (var xchange in xchanges)
{
+ background?.Progress(exported, xchanges.Count, $"{exported:n0} of {xchanges.Count:n0} exchanges");
+ if (background?.CancelRequested.IsCancellationRequested == true)
+ {
+ missing.Add($"The export was cancelled after {exported} of {xchanges.Count} exchanges; the rest aren't here.");
+ break;
+ }
+
+ if (body.Written > MaxBytes)
+ {
+ missing.Add($"The zip reached {MaxBytes / (1024 * 1024 * 1024)} GB after {exported} of {xchanges.Count} exchanges, " +
+ "the most one export builds; the rest aren't here. Export them separately.");
+ break;
+ }
+
+ exported++;
var mainProperty = mainProperties.GetValueOrDefault(xchange.DocumentId);
var mainValue = mainProperty == null ? null : promoted.GetValueOrDefault(xchange.Id)?.Properties?.GetValueOrDefault(mainProperty);
var folder = ExchangeArchive.BaseName(mainValue, xchange.Id);
@@ -136,6 +169,8 @@ await Add(zip, xchange, XchangeFileType.Response, $"{folder}/handled",
result.NotStoredReason(XchangeFileType.Response));
}
+ background?.Progress(exported, xchanges.Count, $"{exported:n0} of {xchanges.Count:n0} exchanges");
+
if (missing.Count > 0)
{
await using var list = await zip.CreateEntry(MissingList).OpenAsync(aborted);
@@ -215,8 +250,8 @@ var type when type.StartsWith("text/") => ".txt",
};
}
- /// The exchanges selects, or why it can't be exported.
- private async Task<(List Ids, IActionResult Refusal)> Resolve(XchangeFilesExport request)
+ /// The exchanges selects, up to , or why it can't be exported.
+ private async Task<(List Ids, IActionResult Refusal)> Resolve(XchangeFilesExport request, int limit)
{
try
{
@@ -228,13 +263,13 @@ var type when type.StartsWith("text/") => ".txt",
}
var selection = new XchangeSelection(dbContext);
- var ids = await selection.Resolve(request, Limit);
+ var ids = await selection.Resolve(request, limit);
if (ids.Count == 0) return (null, Refuse("NOTHING_SELECTED", "Select at least one exchange to export."));
- if (ids.Count > Limit)
+ if (ids.Count > limit)
{
- var count = string.IsNullOrWhiteSpace(request.Filter) ? ids.Count : await selection.Count(request);
+ var count = string.IsNullOrWhiteSpace(request.Filter) ? ids.Count : await selection.Count(request, limit);
return (null, Refuse("TOO_MANY",
- $"{count:n0} exchanges is more than the {Limit} one export can take. Narrow the filter and export the rest after."));
+ $"{count:n0} exchanges is more than the {limit:n0} one export can take. Narrow the filter and export the rest after."));
}
return (ids, null);
@@ -255,6 +290,9 @@ private sealed class HeldSyncWrites(Stream response) : Stream
{
private readonly MemoryStream _held = new();
+ /// How many bytes of zip have been written so far.
+ public long Written { get; private set; }
+
public override bool CanRead => false;
public override bool CanSeek => false;
public override bool CanWrite => true;
@@ -266,8 +304,17 @@ public override long Position
set => throw new NotSupportedException();
}
- public override void Write(byte[] buffer, int offset, int count) => _held.Write(buffer, offset, count);
- public override void Write(ReadOnlySpan buffer) => _held.Write(buffer);
+ public override void Write(byte[] buffer, int offset, int count)
+ {
+ _held.Write(buffer, offset, count);
+ Written += count;
+ }
+
+ public override void Write(ReadOnlySpan buffer)
+ {
+ _held.Write(buffer);
+ Written += buffer.Length;
+ }
/// Nothing to do until the next asynchronous write or flush.
public override void Flush()
@@ -278,6 +325,7 @@ public override async ValueTask WriteAsync(ReadOnlyMemory buffer, Cancella
{
await SendHeld(cancellationToken);
await response.WriteAsync(buffer, cancellationToken);
+ Written += buffer.Length;
}
public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) =>
diff --git a/SW.Bitween.Api/Data/BitweenDbContext.cs b/SW.Bitween.Api/Data/BitweenDbContext.cs
index 1b569444..6f8c25c8 100644
--- a/SW.Bitween.Api/Data/BitweenDbContext.cs
+++ b/SW.Bitween.Api/Data/BitweenDbContext.cs
@@ -40,12 +40,15 @@ public class BitweenDbContext(DbContextOptions options, RequestContext requestCo
///
/// Every provider's context counts its commands against the exchange stage running, so the
- /// round trips an exchange costs are measured rather than estimated (see DbCommandTally).
+ /// round trips an exchange costs are measured rather than estimated (see DbCommandTally),
+ /// and lets a background request's commands run past the usual limit (see
+ /// BackgroundRequestScope).
///
protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder)
{
base.OnConfiguring(optionsBuilder);
- optionsBuilder.AddInterceptors(Services.Metrics.DbCommandCounter.Instance);
+ optionsBuilder.AddInterceptors(Services.Metrics.DbCommandCounter.Instance,
+ Services.Background.BackgroundRequestCommands.Instance);
}
protected override void OnModelCreating(ModelBuilder modelBuilder)
@@ -395,6 +398,25 @@ protected override void OnModelCreating(ModelBuilder modelBuilder)
b.Property(u => u.KeyName).HasMaxLength(500);
});
+ modelBuilder.Entity(b =>
+ {
+ b.ToTable("BackgroundRequests");
+ b.Property(p => p.Id).HasMaxLength(32).IsUnicode(false);
+ b.Property(p => p.Owner).IsRequired().HasMaxLength(50);
+ b.Property(p => p.Path).IsRequired().HasMaxLength(4000);
+ b.Property(p => p.Label).HasMaxLength(200);
+ b.Property(p => p.Page).HasMaxLength(2000);
+ b.Property(p => p.Node).HasMaxLength(200).IsUnicode(false);
+ b.Property(p => p.ResultContentType).HasMaxLength(200).IsUnicode(false);
+ b.Ignore(p => p.Finished);
+ b.Ignore(p => p.Answered);
+ b.Property(p => p.Summary).HasMaxLength(500);
+ b.Property(p => p.ResultDisposition).HasMaxLength(500);
+ b.Property(p => p.ResultKey).HasMaxLength(500).IsUnicode(false);
+ b.HasIndex(p => new { p.Owner, p.StartedOn });
+ b.HasIndex(p => p.StartedOn);
+ });
+
modelBuilder.Entity(b =>
{
b.ToTable("ClusterNodes");
diff --git a/SW.Bitween.Api/Domain/BackgroundRequest.cs b/SW.Bitween.Api/Domain/BackgroundRequest.cs
new file mode 100644
index 00000000..4fa3d854
--- /dev/null
+++ b/SW.Bitween.Api/Domain/BackgroundRequest.cs
@@ -0,0 +1,127 @@
+using System;
+
+namespace SW.Bitween.Domain;
+
+///
+/// Work the admin UI asked for that took longer than it would wait for - prod's list to bring from,
+/// a bulk retry, a zip of exchange files - carried on without it. The answer is kept until the UI
+/// collects it, so any node can hand it over, whichever ran it.
+///
+///
+/// Only what the UI asks to carry on becomes one; see
+/// .
+///
+public class BackgroundRequest
+{
+ private BackgroundRequest()
+ {
+ }
+
+ public BackgroundRequest(string id, string owner, string path, string label, string page, string node)
+ {
+ Id = id;
+ Owner = owner;
+ Path = path;
+ Label = label;
+ Page = page;
+ Node = node;
+ Status = BackgroundRequestStatus.Waiting;
+ StartedOn = DateTime.UtcNow;
+ }
+
+ public string Id { get; private set; }
+
+ /// The account that asked, by its name identifier. Nobody else sees it or its answer.
+ public string Owner { get; private set; }
+
+ /// What was asked: the method, path and query, as the API received them.
+ public string Path { get; private set; }
+
+ /// What the UI called it, for the header.
+ public string Label { get; private set; }
+
+ /// The UI address it was asked from, which is where its answer is shown.
+ public string Page { get; private set; }
+
+ /// The node running it, as its heartbeat names it.
+ public string Node { get; private set; }
+
+ public BackgroundRequestStatus Status { get; private set; }
+ public DateTime StartedOn { get; private set; }
+ public DateTime? FinishedOn { get; private set; }
+
+ /// How far it has got, for work that says: done of total, and in words.
+ public int? Done { get; private set; }
+
+ public int? Total { get; private set; }
+ public string Summary { get; private set; }
+
+ /// The answer's HTTP status; null until there is one.
+ public int? ResultStatus { get; private set; }
+
+ public string ResultContentType { get; private set; }
+
+ /// The answer's Content-Disposition, for one that is a file to save.
+ public string ResultDisposition { get; private set; }
+
+ /// A small answer, kept here.
+ public byte[] Result { get; private set; }
+
+ /// A large one - a zip - kept in storage under this key instead.
+ public string ResultKey { get; private set; }
+
+ public bool Finished => Status is not (BackgroundRequestStatus.Waiting or BackgroundRequestStatus.Running);
+
+ /// Whether there is an answer to hand over, which work cancelled part way can have too.
+ public bool Answered => ResultStatus != null;
+
+ public void Run() => Status = BackgroundRequestStatus.Running;
+
+ public void Progress(int done, int total, string summary)
+ {
+ Done = done;
+ Total = total;
+ Summary = summary;
+ }
+
+ ///
+ /// What the request answered, whatever it answered: an error is an answer too. Work its owner
+ /// cancelled, that stopped where it was and said how far it got, ends cancelled with its answer.
+ ///
+ public void Answer(int status, string contentType, string disposition, byte[] body, string key,
+ bool cancelled = false)
+ {
+ ResultStatus = status;
+ ResultContentType = contentType;
+ ResultDisposition = disposition;
+ Result = body;
+ ResultKey = key;
+ End(cancelled ? BackgroundRequestStatus.Cancelled : BackgroundRequestStatus.Done);
+ }
+
+ public void End(BackgroundRequestStatus status)
+ {
+ Status = status;
+ FinishedOn = DateTime.UtcNow;
+ }
+}
+
+public enum BackgroundRequestStatus
+{
+ /// Started, waiting for one of the node's places to run in.
+ Waiting = 0,
+
+ Running = 1,
+
+ /// Answered; the answer is kept until it expires.
+ Done = 2,
+
+ /// Its owner cancelled it.
+ Cancelled = 3,
+
+ /// Ran past the longest a background request may go without progress.
+ TimedOut = 4,
+
+ /// The node running it stopped before it finished, as a deploy or restart does.
+ Stopped = 5,
+}
diff --git a/SW.Bitween.Api/Resources/Xchanges/BulkRetry.cs b/SW.Bitween.Api/Resources/Xchanges/BulkRetry.cs
index 65913c5f..3cfbc6ef 100644
--- a/SW.Bitween.Api/Resources/Xchanges/BulkRetry.cs
+++ b/SW.Bitween.Api/Resources/Xchanges/BulkRetry.cs
@@ -1,8 +1,10 @@
-using System.Linq;
+using System.Collections.Generic;
+using System.Linq;
using System.Threading.Tasks;
using Microsoft.EntityFrameworkCore;
using SW.Bitween.Domain;
using SW.Bitween.Model;
+using SW.Bitween.Services.Background;
using SW.PrimitiveTypes;
namespace SW.Bitween.Resources.Xchanges
@@ -12,9 +14,30 @@ namespace SW.Bitween.Resources.Xchanges
/// matches. Returns what it did: answers the same question
/// beforehand, so the caller can show it and be sure the two agree.
///
+ ///
+ /// Asked to carry on in the background, as the admin UI asks it, it takes the whole selection a
+ /// preview works out - up to - saving as it goes, saying
+ /// how far it has got, and stopping between exchanges when cancelled. It goes at the pipeline's pace:
+ /// no more than of its exchanges wait to be processed at once, so a retry of
+ /// hundreds of thousands never floods the queues, and the partners behind them, and the day's own
+ /// traffic is never more than that many behind it. Asked directly, it keeps to
+ /// , as it always has.
+ ///
[HandlerName("bulkretry")]
public class BulkRetry : ICommandHandler
{
+ ///
+ /// How many exchanges go into one save. Each save is its own transaction, so a retry that
+ /// stops part way keeps what it did; and no more than this many are held in memory at once.
+ ///
+ private const int SaveEvery = 100;
+
+ /// The most of a background retry's exchanges waiting to be processed at once. Set lower by tests.
+ internal static int Waiting { get; set; } = 1_000;
+
+ /// How often a retry held back by looks again. Set shorter by tests.
+ internal static System.TimeSpan LookEvery { get; set; } = System.TimeSpan.FromSeconds(1);
+
private readonly BitweenDbContext _dbContext;
private readonly RequestContext _requestContext;
private readonly XchangeService _xchangeService;
@@ -32,102 +55,182 @@ public async Task Handle(XchangeBulkRetry request)
{
await _requestContext.EnsurePermission(_dbContext, Model.Permissions.Exchanges.Operate);
- var prepared = await new BulkRetryPlanner(_dbContext).Prepare(request, BulkRetryPlanner.Limit);
+ var background = BackgroundRequestScope.Current;
+ var upTo = background != null ? BulkRetryPlanner.MaxSelection : BulkRetryPlanner.Limit;
+ var prepared = await new BulkRetryPlanner(_dbContext).Prepare(request, upTo, listTargets: background == null);
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. Send the preview's targets back in rounds of {BulkRetryPlanner.Limit}.");
+ throw new SWValidationException("TOO_MANY", background != null
+ ? $"{plan.Selected:n0} exchanges is more than the {BulkRetryPlanner.MaxSelection:n0} one retry " +
+ "takes. Narrow the selection."
+ : $"{plan.Selected:n0} exchanges is more than the {BulkRetryPlanner.Limit} this can retry " +
+ $"in one go. Send the preview's targets back in rounds of {BulkRetryPlanner.Limit}.");
+ var total = prepared.Targets.Count;
var retried = 0;
+ var handled = 0;
+ // The retries this run made that the pipeline hasn't finished, and this save's targets.
+ var unfinished = new List();
+ var batch = new List();
+ background?.Progress(0, total, Said(0, plan));
foreach (var id in prepared.Targets)
{
- var xchange = await _dbContext.Set().AsNoTracking()
- .FirstOrDefaultAsync(x => x.Id == id);
- if (xchange == null)
+ // Between exchanges only, so the one under way is either done or not started; and room
+ // for a whole save at its start, so what waits never passes Waiting.
+ if (background?.CancelRequested.IsCancellationRequested == true ||
+ background != null && handled % SaveEvery == 0 &&
+ !await RoomFor(unfinished, background, handled, total, retried, plan))
{
- plan.Skipped.Add(new XchangeRetrySkip
- {
- Id = id,
- Reason = "This exchange no longer exists."
- });
- continue;
+ plan.StoppedAfter = handled;
+ break;
}
- var subscription = await _dbContext.Subscriptions()
- .FirstOrDefaultAsync(s => s.Id == xchange.SubscriptionId);
-
- if (request.Reset && subscription == null)
+ // Only those it retried: a skipped one's retry, if it has one, is someone else's.
+ if (await RetryOne(id, request, plan))
{
- // 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;
+ retried++;
+ batch.Add(id);
}
- // 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)
+ handled++;
+
+ if (handled % SaveEvery != 0) continue;
+ await _dbContext.SaveChangesAsync();
+ _dbContext.ChangeTracker.Clear();
+ if (background != null) unfinished.AddRange(await RetriesOf(batch));
+ batch.Clear();
+ background?.Progress(handled, total, Said(retried, plan));
+ }
+
+ if (retried > 0)
+ _dbContext.Record(_requestContext, OperatorAction.BulkRetry, $"{retried} exchanges",
+ $"{retried} retried, {plan.SkippedCount} skipped" +
+ (plan.StoppedAfter is { } stopped ? $", cancelled after {stopped} of {total}" : ""));
+ await _dbContext.SaveChangesAsync();
+
+ plan.WillRetry = retried;
+ background?.Progress(handled, total, Said(retried, plan));
+ return plan;
+ }
+
+ /// How far it has got, in the words the header shows.
+ private static string Said(int retried, XchangeBulkRetryPlan plan) =>
+ $"{retried:n0} retried, {plan.SkippedCount:n0} skipped";
+
+ ///
+ /// Waits until a save's worth more of this run's retries can join those still to be processed
+ /// without passing . Each one
+ /// the pipeline finishes is progress, which keeps the retry's limit from running out; a pipeline
+ /// that finishes none for the whole limit - stopped, or a paused subscription holding them - ends
+ /// it there. False when its owner cancels it meanwhile.
+ ///
+ private async Task RoomFor(List unfinished, BackgroundRequestScope background, int handled,
+ int total, int retried, XchangeBulkRetryPlan plan)
+ {
+ while (unfinished.Count > 0 && unfinished.Count + SaveEvery > Waiting)
+ {
+ var finished = (await _dbContext.Set().AsNoTracking()
+ .Where(r => unfinished.Contains(r.Id)).Select(r => r.Id).ToListAsync()).ToHashSet();
+ if (finished.Count > 0)
{
- plan.Skipped.Add(new XchangeRetrySkip
- {
- Id = id,
- Reason = "Its input document could not be read, so there is nothing to re-send."
- });
+ unfinished.RemoveAll(finished.Contains);
+ background.Progress(handled, total,
+ $"{Said(retried, plan)} · waiting for the pipeline, {unfinished.Count:n0} to go");
continue;
}
- try
+ if (background.CancelRequested.IsCancellationRequested) return false;
+ await Task.Delay(LookEvery, background.Stopped);
+ }
+
+ return true;
+ }
+
+ /// The retries just saved for .
+ private Task> RetriesOf(List targets) =>
+ _dbContext.Set().AsNoTracking()
+ .Where(x => targets.Contains(x.RetryFor)).Select(x => x.Id).ToListAsync();
+
+ /// Retries one target, or says in the plan why it didn't; true when it did.
+ private async Task RetryOne(string id, XchangeBulkRetry request, XchangeBulkRetryPlan plan)
+ {
+ var xchange = await _dbContext.Set().AsNoTracking()
+ .FirstOrDefaultAsync(x => x.Id == id);
+ if (xchange == null)
+ {
+ BulkRetryPlanner.Skip(plan, new XchangeRetrySkip
{
- 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);
- }
- }
- catch (SWValidationException e) when (
- e.Validations.Any(v => v.Key == "ALREADY_RETRIED"))
+ Id = id,
+ Reason = "This exchange no longer exists."
+ });
+ return false;
+ }
+
+ var subscription = await _dbContext.Subscriptions()
+ .FirstOrDefaultAsync(s => s.Id == xchange.SubscriptionId);
+
+ 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.
+ BulkRetryPlanner.Skip(plan, new XchangeRetrySkip
{
- // 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;
- }
+ Id = id,
+ Reason = "Its properties cannot be reset: the subscription no longer exists."
+ });
+ return false;
+ }
- retried++;
+ // 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)
+ {
+ BulkRetryPlanner.Skip(plan, new XchangeRetrySkip
+ {
+ Id = id,
+ Reason = "Its input document could not be read, so there is nothing to re-send."
+ });
+ return false;
}
- if (retried > 0)
- _dbContext.Record(_requestContext, OperatorAction.BulkRetry, $"{retried} exchanges",
- $"{retried} retried, {plan.Skipped?.Count ?? 0} skipped");
- await _dbContext.SaveChangesAsync();
+ try
+ {
+ 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);
+ }
+ }
+ catch (SWValidationException e) when (
+ e.Validations.Any(v => v.Key == "ALREADY_RETRIED"))
+ {
+ // 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.
+ BulkRetryPlanner.Skip(plan, new XchangeRetrySkip
+ {
+ Id = id,
+ Reason = "It was retried by someone else a moment ago. Its own retry can be retried instead."
+ });
+ return false;
+ }
- plan.WillRetry = retried;
- return plan;
+ return true;
}
}
}
diff --git a/SW.Bitween.Api/Resources/Xchanges/BulkRetryPlanner.cs b/SW.Bitween.Api/Resources/Xchanges/BulkRetryPlanner.cs
index 8fc46cf0..974ff07b 100644
--- a/SW.Bitween.Api/Resources/Xchanges/BulkRetryPlanner.cs
+++ b/SW.Bitween.Api/Resources/Xchanges/BulkRetryPlanner.cs
@@ -20,23 +20,28 @@ namespace SW.Bitween.Resources.Xchanges;
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. A bigger selection is retried in rounds of this many.
+ /// The largest selection a retry asked directly carries 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. A bigger selection is retried in rounds of this many, or in the background.
///
internal const int Limit = 500;
///
- /// 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 . Past it the caller is
- /// asked to narrow the filter.
+ /// The largest selection a preview works out, and a retry in the background carries out - as the
+ /// admin UI asks it. Past it the caller is asked to narrow the selection.
///
///
- /// Rounds rather than raising : each round is its own request, so none of them
- /// runs long, and a round that fails leaves the ones before it done.
+ /// A retry this size doesn't put it all on the queues at once: it keeps 's
+ /// window of exchanges waiting at most, so it goes at the pipeline's pace.
///
- internal const int MaxSelection = 5000;
+ internal const int MaxSelection = 500_000;
+
+ ///
+ /// How many substitutions, and how many skips, a plan names; the rest are counted. A selection of
+ /// hundreds of thousands would otherwise answer with tens of megabytes no one reads past the top.
+ ///
+ internal const int Listed = 1_000;
///
/// Guards the walk to the newest attempt against a cycle in the data, the same way
@@ -59,7 +64,11 @@ internal sealed class Prepared
/// The largest selection to work out: for a retry,
/// for a preview.
///
- internal async Task Prepare(XchangeBulkRetry request, int upTo)
+ ///
+ /// Whether the plan lists , for a caller retrying in rounds;
+ /// one retrying in the background doesn't need them.
+ ///
+ internal async Task Prepare(XchangeBulkRetry request, int upTo, bool listTargets = true)
{
var selection = new XchangeSelection(_dbContext);
var selected = await selection.Resolve(request, upTo);
@@ -75,7 +84,7 @@ internal async Task Prepare(XchangeBulkRetry request, int upTo)
// 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),
+ : await selection.Count(request, upTo),
Limit = Limit,
MaxSelection = MaxSelection,
OverLimit = true
@@ -86,7 +95,10 @@ internal async Task Prepare(XchangeBulkRetry request, int upTo)
var state = await ReadState(newestAttempt.Values.Distinct().ToList());
var reset = request.Reset;
- var plan = new XchangeBulkRetryPlan { Selected = selected.Count, Limit = Limit, MaxSelection = MaxSelection };
+ var plan = new XchangeBulkRetryPlan
+ {
+ Selected = selected.Count, Limit = Limit, MaxSelection = MaxSelection, Listed = Listed
+ };
var targets = new List();
// The list keeps the order; the set answers "already in it" without walking up to MaxSelection entries.
var targeted = new HashSet();
@@ -96,7 +108,7 @@ internal async Task Prepare(XchangeBulkRetry request, int upTo)
var targetId = newestAttempt[selectedId];
if (targetId != selectedId)
- plan.Substituted.Add(new XchangeRetrySubstitution
+ Substitute(plan, new XchangeRetrySubstitution
{
SelectedId = selectedId,
RetryId = targetId
@@ -105,7 +117,7 @@ internal async Task Prepare(XchangeBulkRetry request, int upTo)
var reason = WhyNot(state.GetValueOrDefault(targetId), targetId != selectedId, reset);
if (reason != null)
{
- plan.Skipped.Add(new XchangeRetrySkip { Id = targetId, Reason = reason });
+ Skip(plan, new XchangeRetrySkip { Id = targetId, Reason = reason });
continue;
}
@@ -116,11 +128,25 @@ internal async Task Prepare(XchangeBulkRetry request, int upTo)
}
plan.WillRetry = targets.Count;
- plan.Targets = targets;
+ if (listTargets) plan.Targets = targets;
plan.Properties = await ReadPromotedProperties(plan);
return new Prepared { Plan = plan, Targets = targets };
}
+ /// Counts a skip, and names it while fewer than are named.
+ internal static void Skip(XchangeBulkRetryPlan plan, XchangeRetrySkip skip)
+ {
+ plan.SkippedCount++;
+ if (plan.Skipped.Count < Listed) plan.Skipped.Add(skip);
+ }
+
+ /// Counts a substitution, and names it while fewer than are named.
+ private static void Substitute(XchangeBulkRetryPlan plan, XchangeRetrySubstitution substitution)
+ {
+ plan.SubstitutedCount++;
+ if (plan.Substituted.Count < Listed) plan.Substituted.Add(substitution);
+ }
+
///
/// 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
diff --git a/SW.Bitween.Api/Resources/Xchanges/BulkRetryPreview.cs b/SW.Bitween.Api/Resources/Xchanges/BulkRetryPreview.cs
index b9cc69bf..389a7c75 100644
--- a/SW.Bitween.Api/Resources/Xchanges/BulkRetryPreview.cs
+++ b/SW.Bitween.Api/Resources/Xchanges/BulkRetryPreview.cs
@@ -36,7 +36,8 @@ public async Task 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, BulkRetryPlanner.MaxSelection);
+ var prepared = await new BulkRetryPlanner(_dbContext).Prepare(request, BulkRetryPlanner.MaxSelection,
+ listTargets: Services.Background.BackgroundRequestScope.Current == null);
return prepared.Plan;
}
}
diff --git a/SW.Bitween.Api/Resources/Xchanges/XchangeSelection.cs b/SW.Bitween.Api/Resources/Xchanges/XchangeSelection.cs
index ef1335d8..5ead4642 100644
--- a/SW.Bitween.Api/Resources/Xchanges/XchangeSelection.cs
+++ b/SW.Bitween.Api/Resources/Xchanges/XchangeSelection.cs
@@ -1,3 +1,4 @@
+using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
@@ -36,16 +37,19 @@ internal async Task> Resolve(IXchangeSelection request, int limit)
.ToListAsync();
}
- /// How many exchanges a filter's selection really holds, for saying how far past a limit it is.
- internal async Task Count(IXchangeSelection request)
+ ///
+ /// How many exchanges a filter's selection really holds, for saying how far past
+ /// it is: counted to ten times past it at most.
+ ///
+ internal async Task Count(IXchangeSelection request, int limit)
{
var exclude = request.ExcludeIds ?? new List();
return await (await Query(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)
+ // matching row. The number is only being used to say "too many", so it stops well past
+ // the limit - far enough that the number said is never one under it.
+ .Take(Math.Max(Search.CountCap, limit * 10) + 1)
.CountAsync();
}
diff --git a/SW.Bitween.Api/Services/Background/AskedRequest.cs b/SW.Bitween.Api/Services/Background/AskedRequest.cs
new file mode 100644
index 00000000..b23e92b9
--- /dev/null
+++ b/SW.Bitween.Api/Services/Background/AskedRequest.cs
@@ -0,0 +1,151 @@
+using System.IO;
+using System.Linq;
+using System.Threading.Tasks;
+using System.Net;
+using System.Security.Claims;
+using System.Threading;
+using Microsoft.AspNetCore.Authentication;
+using Microsoft.AspNetCore.Http;
+using Microsoft.AspNetCore.Http.Features;
+using Microsoft.AspNetCore.Routing;
+
+namespace SW.Bitween.Services.Background;
+
+///
+/// What a request asked, copied off it, so it can be answered after the request that brought it has
+/// ended and its context has been handed to the next caller.
+///
+///
+/// Taken once routing, authentication and authorization have run, so a new context made from it
+/// carries their outcome - the endpoint, the user, the markers the endpoint middleware checks for -
+/// and only what follows them runs again. That is everything from the request context onwards: the
+/// handler, model binding and serialization, the same as for any request.
+///
+internal sealed class AskedRequest
+{
+ string method, scheme, protocol, traceId;
+ HostString host;
+ PathString pathBase, path;
+ QueryString query;
+ HeaderDictionary headers;
+ ClaimsPrincipal user;
+ (object Key, object Value)[] markers;
+ Endpoint endpoint;
+ RouteValueDictionary routeValues;
+ IAuthenticateResultFeature authenticated;
+ IPAddress remoteIp, localIp;
+ int remotePort, localPort;
+ byte[] sent;
+
+ /// The largest body a request carried on in the background may send: a selection, not a file.
+ public const int MaxBody = 4 * 1024 * 1024;
+
+ ///
+ /// What asked, its body read whole; null for a body too big to carry,
+ /// which is then asked the usual way.
+ ///
+ public static async Task OfAsync(HttpContext context)
+ {
+ var request = context.Request;
+ byte[] sent = [];
+ // Any body but one known to be empty: HTTP/2 can send one with neither a length nor chunks.
+ if (request.ContentLength is not 0)
+ {
+ if (request.ContentLength is > MaxBody) return null;
+ // Rewound after, so a body too big to carry can still be read by the request as usual.
+ request.EnableBuffering();
+ // No more than one past the most it carries, whatever the body says of itself.
+ using var copy = new MemoryStream();
+ var chunk = new byte[81920];
+ int got;
+ while (copy.Length <= MaxBody &&
+ (got = await request.Body.ReadAsync(chunk, context.RequestAborted)) > 0)
+ copy.Write(chunk, 0, got);
+ request.Body.Position = 0;
+ if (copy.Length > MaxBody) return null;
+ sent = copy.ToArray();
+ }
+
+ return new AskedRequest
+ {
+ sent = sent,
+ method = request.Method,
+ scheme = request.Scheme,
+ protocol = request.Protocol,
+ traceId = context.TraceIdentifier,
+ host = request.Host,
+ pathBase = request.PathBase,
+ path = request.Path,
+ query = request.QueryString,
+ headers = new HeaderDictionary(request.Headers.ToDictionary(h => h.Key, h => h.Value)),
+ user = context.User,
+ // Only the "this middleware ran" markers: the endpoint middleware refuses an endpoint that
+ // asks for authorization when the authorization middleware has not marked the request. The
+ // rest of Items belongs to the request it was set on.
+ markers = context.Items
+ .Where(i => i.Key is string key && key.StartsWith("__") && key.EndsWith("MiddlewareWithEndpointInvoked"))
+ .Select(i => (i.Key, i.Value))
+ .ToArray(),
+ endpoint = context.GetEndpoint(),
+ routeValues = new RouteValueDictionary(request.RouteValues),
+ authenticated = context.Features.Get(),
+ remoteIp = context.Connection.RemoteIpAddress,
+ localIp = context.Connection.LocalIpAddress,
+ remotePort = context.Connection.RemotePort,
+ localPort = context.Connection.LocalPort,
+ };
+ }
+
+ /// The path and query, as asked.
+ public string PathAndQuery => (pathBase + path + query).ToString();
+
+ /// Whether it only reads, and so can be asked again from the start.
+ public bool IsRead => HttpMethods.IsGet(method);
+
+ /// What it asked, as the list of background requests says it: the path, after the method unless a read.
+ public string Described => IsRead ? PathAndQuery : $"{method} {PathAndQuery}";
+
+ ///
+ /// A new context asking the same thing, writing its answer to , and told it
+ /// was aborted when is cancelled.
+ ///
+ public DefaultHttpContext NewContext(System.IServiceProvider services, Stream body, CancellationToken aborted, int attempt)
+ {
+ var features = new FeatureCollection();
+ features.Set(new HttpRequestFeature
+ {
+ Method = method,
+ Scheme = scheme,
+ Protocol = protocol,
+ PathBase = pathBase.Value ?? "",
+ Path = path.Value ?? "",
+ QueryString = query.Value ?? "",
+ RawTarget = PathAndQuery,
+ Headers = new HeaderDictionary(headers.ToDictionary(h => h.Key, h => h.Value)),
+ Body = new MemoryStream(sent, writable: false),
+ });
+ features.Set(new HttpResponseFeature());
+ features.Set(new StreamResponseBodyFeature(body));
+ features.Set(new HttpRequestLifetimeFeature { RequestAborted = aborted });
+ features.Set(new HttpConnectionFeature
+ {
+ RemoteIpAddress = remoteIp,
+ LocalIpAddress = localIp,
+ RemotePort = remotePort,
+ LocalPort = localPort,
+ });
+ if (authenticated != null) features.Set(authenticated);
+
+ var context = new DefaultHttpContext(features)
+ {
+ User = user,
+ RequestServices = services,
+ TraceIdentifier = $"{traceId}:{attempt}",
+ };
+ context.Request.Host = host;
+ foreach (var (key, value) in markers) context.Items[key] = value;
+ context.SetEndpoint(endpoint);
+ context.Request.RouteValues = new RouteValueDictionary(routeValues);
+ return context;
+ }
+}
diff --git a/SW.Bitween.Api/Services/Background/BackgroundDownloads.cs b/SW.Bitween.Api/Services/Background/BackgroundDownloads.cs
new file mode 100644
index 00000000..1d4cd6a1
--- /dev/null
+++ b/SW.Bitween.Api/Services/Background/BackgroundDownloads.cs
@@ -0,0 +1,48 @@
+using System;
+using System.Globalization;
+using System.Security.Cryptography;
+using System.Text;
+using Microsoft.Extensions.Configuration;
+
+namespace SW.Bitween.Services.Background;
+
+///
+/// Links to a background request's file - an export's zip - that a browser can download the ordinary
+/// way, straight to disk, rather than holding it in memory first: a browser following a link can't send
+/// the sign-in header.
+///
+///
+/// A link carries its expiry and a seal over the request's id and that expiry, made with a secret
+/// derived from Token:Key , as are: any node can check it, and changing a
+/// character of it breaks it. Only the request's owner is handed one, and it opens that one file for
+/// .
+///
+public class BackgroundDownloads(IConfiguration configuration)
+{
+ /// How long a link opens its file: long enough to start the download it was asked for.
+ public static readonly TimeSpan Lasts = TimeSpan.FromMinutes(10);
+
+ private readonly byte[] key = HKDF.DeriveKey(HashAlgorithmName.SHA256,
+ Encoding.UTF8.GetBytes(configuration["Token:Key"] ?? string.Empty), 32,
+ info: Encoding.UTF8.GetBytes("bitween:background-downloads:v1"));
+
+ /// The address, under /api , that downloads 's file until it expires.
+ public string LinkTo(string id)
+ {
+ var expires = DateTimeOffset.UtcNow.Add(Lasts).ToUnixTimeSeconds().ToString(CultureInfo.InvariantCulture);
+ return $"/api/background/{id}/file?expires={expires}&seal={Seal(id, expires)}";
+ }
+
+ /// Whether a link's expiry and seal open 's file now.
+ public bool Opens(string id, string expires, string seal)
+ {
+ if (string.IsNullOrEmpty(seal) || !long.TryParse(expires, NumberStyles.None, CultureInfo.InvariantCulture, out var at) ||
+ DateTimeOffset.UtcNow.ToUnixTimeSeconds() > at)
+ return false;
+ return CryptographicOperations.FixedTimeEquals(Encoding.ASCII.GetBytes(Seal(id, expires)), Encoding.ASCII.GetBytes(seal));
+ }
+
+ private string Seal(string id, string expires) =>
+ Convert.ToBase64String(HMACSHA256.HashData(key, Encoding.UTF8.GetBytes($"{id}|{expires}")))
+ .TrimEnd('=').Replace('+', '-').Replace('/', '_');
+}
diff --git a/SW.Bitween.Api/Services/Background/BackgroundRequestHub.cs b/SW.Bitween.Api/Services/Background/BackgroundRequestHub.cs
new file mode 100644
index 00000000..5117ad8e
--- /dev/null
+++ b/SW.Bitween.Api/Services/Background/BackgroundRequestHub.cs
@@ -0,0 +1,76 @@
+using System.Threading.Tasks;
+using Microsoft.AspNetCore.Authorization;
+using Microsoft.AspNetCore.SignalR;
+using SW.Bitween.Domain;
+using SW.PrimitiveTypes;
+
+namespace SW.Bitween.Services.Background;
+
+///
+/// Where the admin UI hears that one of its background requests changed. It only listens: what it
+/// does about it - collect the answer, cancel - goes through /api/background .
+///
+///
+/// Each connection is its signed-in user's, by the token's name identifier, which is what
+/// sends to; a node knows only the connections it holds, so every
+/// change is also broadcast for the others to pass on.
+///
+[Authorize]
+public class BackgroundRequestHub : Hub
+{
+ public const string Path = "/api/background/hub";
+
+ /// The message: { id, status } .
+ public const string Changed = "changed";
+}
+
+/// A background request changed, said to every node, for each to tell its own connections.
+public class BackgroundRequestChanged
+{
+ public string Id { get; set; }
+ public string Owner { get; set; }
+ public BackgroundRequestStatus Status { get; set; }
+
+ /// How far it got, for work that counts its steps.
+ public int? Done { get; set; }
+
+ public int? Total { get; set; }
+ public string Summary { get; set; }
+
+ /// The that sent it, which told its own already.
+ public string Node { get; set; }
+
+ public static BackgroundRequestChanged Of(BackgroundRequest row, string node) => new()
+ {
+ Id = row.Id,
+ Owner = row.Owner,
+ Status = row.Status,
+ Done = row.Done,
+ Total = row.Total,
+ Summary = row.Summary,
+ Node = node,
+ };
+}
+
+/// Its owner cancelled a background request, said to every node; the one running it stops it.
+public class BackgroundRequestCancel
+{
+ public string Id { get; set; }
+}
+
+public class BackgroundRequestChangedListener(BackgroundRequests requests) : IListen
+{
+ public Task Process(BackgroundRequestChanged message) =>
+ message == null || message.Node == requests.Node
+ ? Task.CompletedTask
+ : requests.PushAsync(message);
+}
+
+public class BackgroundRequestCancelListener(BackgroundRequests requests) : IListen
+{
+ public Task Process(BackgroundRequestCancel message)
+ {
+ if (message?.Id != null) requests.CancelHere(message.Id);
+ return Task.CompletedTask;
+ }
+}
diff --git a/SW.Bitween.Api/Services/Background/BackgroundRequestMiddleware.cs b/SW.Bitween.Api/Services/Background/BackgroundRequestMiddleware.cs
new file mode 100644
index 00000000..71a1f0d8
--- /dev/null
+++ b/SW.Bitween.Api/Services/Background/BackgroundRequestMiddleware.cs
@@ -0,0 +1,136 @@
+using System;
+using System.Linq;
+using System.Security.Claims;
+using System.Threading.Tasks;
+using Microsoft.AspNetCore.Http;
+using Microsoft.Net.Http.Headers;
+
+namespace SW.Bitween.Services.Background;
+
+///
+/// Lets work that takes long go on in the background instead of holding its caller: a GET or POST under
+/// /api whose caller sent Prefer: respond-async is answered as usual when it answers within
+/// the caller's wait (three seconds unless it says), and otherwise with 202 Accepted and the
+/// id of a background request that goes on running. The admin UI asks it only of the work that can
+/// take long - prod's list to bring from, bringing, bulk retry, exporting files. Every other request
+/// passes through untouched - the UI's other pages, the CLI, scripts, another Bitween - so they keep the
+/// limits they always had.
+///
+///
+/// Sits after authentication and authorization, whose outcome the background request keeps, and before
+/// the request context is filled in, which the background request does again for itself; see
+/// .
+///
+public class BackgroundRequestMiddleware(RequestDelegate next, BackgroundRequests requests)
+{
+ /// What the UI calls the request, for the header. URL-encoded, being free text.
+ public const string LabelHeader = "X-Bitween-Label";
+
+ /// The UI address the request was made from. URL-encoded.
+ public const string PageHeader = "X-Bitween-Page";
+
+ static readonly TimeSpan DefaultWait = TimeSpan.FromSeconds(3);
+ static readonly TimeSpan LongestWait = TimeSpan.FromSeconds(30);
+
+ public async Task InvokeAsync(HttpContext context)
+ {
+ if (!Wanted(context, out var wait, out var owner))
+ {
+ await next(context);
+ return;
+ }
+
+ var asked = await AskedRequest.OfAsync(context);
+ if (asked == null)
+ {
+ await next(context);
+ return;
+ }
+
+ var work = requests.Start(asked, next, owner, Header(context, LabelHeader), Header(context, PageHeader));
+
+ BackgroundRequests.Answer answer = null;
+ if (await Task.WhenAny(work.AnsweredHere.Task, Task.Delay(wait, context.RequestAborted)) == work.AnsweredHere.Task)
+ answer = await work.AnsweredHere.Task;
+ else if (context.RequestAborted.IsCancellationRequested)
+ {
+ // The caller left before it was answered, so nobody will ask for the answer.
+ work.Cancel();
+ return;
+ }
+
+ if (answer == null && !work.TryMove())
+ // Answered between the wait running out and the move: answered as it came.
+ answer = await work.AnsweredHere.Task;
+
+ if (answer != null)
+ {
+ await using (answer.Body) await WriteAsync(context, answer);
+ return;
+ }
+
+ var kept = await requests.KeepAsync(work);
+ context.Response.StatusCode = StatusCodes.Status202Accepted;
+ context.Response.Headers[HeaderNames.Location] = $"/api/background/{kept.Id}";
+ context.Response.Headers["Preference-Applied"] = "respond-async";
+ await context.Response.WriteAsJsonAsync(new { id = kept.Id, status = kept.Status.ToString() });
+ }
+
+ static bool Wanted(HttpContext context, out TimeSpan wait, out string owner)
+ {
+ wait = DefaultWait;
+ owner = context.User.FindFirst(ClaimTypes.NameIdentifier)?.Value;
+ if (!(HttpMethods.IsGet(context.Request.Method) || HttpMethods.IsPost(context.Request.Method)) || owner == null ||
+ !context.Request.Path.StartsWithSegments("/api") ||
+ context.Request.Path.StartsWithSegments("/api/background"))
+ return false;
+
+ var asked = false;
+ foreach (var preference in context.Request.Headers["Prefer"]
+ .SelectMany(v => (v ?? "").Split(',', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries)))
+ {
+ if (preference.Equals("respond-async", StringComparison.OrdinalIgnoreCase)) asked = true;
+ else if (preference.StartsWith("wait=", StringComparison.OrdinalIgnoreCase) &&
+ int.TryParse(preference["wait=".Length..], out var seconds) && seconds >= 0)
+ wait = TimeSpan.FromSeconds(seconds) < LongestWait ? TimeSpan.FromSeconds(seconds) : LongestWait;
+ }
+
+ return asked;
+ }
+
+ static async Task WriteAsync(HttpContext context, BackgroundRequests.Answer answer)
+ {
+ switch (answer.Kind)
+ {
+ case BackgroundRequests.AnswerKind.Answered:
+ context.Response.StatusCode = answer.Status;
+ foreach (var header in answer.Headers)
+ if (!header.Key.Equals(HeaderNames.ContentLength, StringComparison.OrdinalIgnoreCase) &&
+ !header.Key.Equals(HeaderNames.TransferEncoding, StringComparison.OrdinalIgnoreCase))
+ context.Response.Headers[header.Key] = header.Value;
+ await answer.Body.CopyToAsync(context.Response.Body);
+ break;
+ case BackgroundRequests.AnswerKind.TimedOut:
+ // Not reachable within a wait of seconds; answered the way the database would have.
+ context.Response.StatusCode = StatusCodes.Status504GatewayTimeout;
+ break;
+ default:
+ context.Response.StatusCode = StatusCodes.Status500InternalServerError;
+ break;
+ }
+ }
+
+ static string Header(HttpContext context, string name)
+ {
+ var value = context.Request.Headers[name].FirstOrDefault();
+ if (string.IsNullOrWhiteSpace(value)) return null;
+ try
+ {
+ return Uri.UnescapeDataString(value);
+ }
+ catch (UriFormatException)
+ {
+ return value;
+ }
+ }
+}
diff --git a/SW.Bitween.Api/Services/Background/BackgroundRequestScope.cs b/SW.Bitween.Api/Services/Background/BackgroundRequestScope.cs
new file mode 100644
index 00000000..10c3e30c
--- /dev/null
+++ b/SW.Bitween.Api/Services/Background/BackgroundRequestScope.cs
@@ -0,0 +1,220 @@
+using System;
+using System.Collections.Concurrent;
+using System.Data.Common;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.EntityFrameworkCore.Diagnostics;
+
+namespace SW.Bitween.Services.Background;
+
+///
+/// The background request running on this flow, if one is: how long its database commands may take,
+/// what stops them, and where work that counts its steps says how far it has got.
+///
+///
+///
+/// A request someone is waiting on keeps the connection's own limit, thirty seconds by default. A
+/// background request has nobody waiting, so its commands may use whatever is left of its own limit
+/// instead; and stopping it cancels the command the database is running, rather than only the wait
+/// for it. Opened like , and read by
+/// .
+///
+///
+/// Work that reports - a bulk retry, an export - is different in two ways. Each
+/// report moves its limit on, since work still making progress isn't stuck. And its owner's cancel
+/// doesn't stop it mid-step: it sees between steps and ends there, so
+/// what it has done is saved and said.
+///
+///
+public sealed class BackgroundRequestScope
+{
+ static readonly AsyncLocal current = new();
+
+ readonly ConcurrentDictionary running = new();
+ readonly Action progressed;
+
+ BackgroundRequestScope(DateTime deadline, CancellationToken stopped, CancellationToken cancelRequested,
+ Action progressed)
+ {
+ Deadline = deadline;
+ Stopped = stopped;
+ CancelRequested = cancelRequested;
+ this.progressed = progressed;
+ stopped.Register(CancelRunning);
+ }
+
+ /// The background request running on this flow, if one is.
+ public static BackgroundRequestScope Current => current.Value;
+
+ ///
+ /// Makes what runs on this flow from here on a background request's. Called at the top of an
+ /// async method, it covers that method and everything it awaits, and ends with it.
+ ///
+ public static BackgroundRequestScope Open(DateTime deadline, CancellationToken stopped,
+ CancellationToken cancelRequested = default, Action progressed = null)
+ {
+ var scope = new BackgroundRequestScope(deadline, stopped, cancelRequested, progressed);
+ current.Value = scope;
+ return scope;
+ }
+
+ public DateTime Deadline { get; private set; }
+
+ /// Stops it where it is: once the deadline passes, or its owner cancels work that doesn't count its steps.
+ public CancellationToken Stopped { get; }
+
+ /// Its owner cancelled it. Work that counts its steps checks this between them, and ends there.
+ public CancellationToken CancelRequested { get; }
+
+ /// Whether it has said how far it got, which makes a cancel wait for the step it's on.
+ public bool Counts { get; private set; }
+
+ ///
+ /// Says how far it has got, and that it is getting somewhere: the limit starts again from now.
+ /// Cheap enough to call per step; it is passed on at most once a second.
+ ///
+ public void Progress(int done, int total, string summary = null)
+ {
+ Counts = true;
+ Deadline = DateTime.UtcNow + BackgroundRequests.Limit;
+ progressed?.Invoke(done, total, summary);
+ }
+
+ internal void Starting(DbCommand command)
+ {
+ Stopped.ThrowIfCancellationRequested();
+ command.CommandTimeout = Math.Max(1, (int)Math.Ceiling((Deadline - DateTime.UtcNow).TotalSeconds));
+ running.TryAdd(command, 0);
+ // A stop landing between the check and the add found nothing to cancel.
+ if (Stopped.IsCancellationRequested) Cancel(command);
+ }
+
+ internal void Ended(DbCommand command) => running.TryRemove(command, out _);
+
+ void CancelRunning()
+ {
+ foreach (var command in running.Keys) Cancel(command);
+ }
+
+ static void Cancel(DbCommand command)
+ {
+ try
+ {
+ command.Cancel();
+ }
+ catch
+ {
+ // Already finished, or the provider can't; the deadline still applies.
+ }
+ }
+}
+
+///
+/// Applies the background request open on the calling flow to each command EF sends; nothing when
+/// none is open. A command that reads stays covered until its reader is disposed, since a large read
+/// is still running after the command has returned.
+///
+public sealed class BackgroundRequestCommands : DbCommandInterceptor
+{
+ public static readonly BackgroundRequestCommands Instance = new();
+
+ BackgroundRequestCommands()
+ {
+ }
+
+ static void Starting(DbCommand command) => BackgroundRequestScope.Current?.Starting(command);
+ static void Ended(DbCommand command) => BackgroundRequestScope.Current?.Ended(command);
+
+ public override InterceptionResult ReaderExecuting(DbCommand command, CommandEventData eventData,
+ InterceptionResult result)
+ {
+ Starting(command);
+ return result;
+ }
+
+ public override ValueTask> ReaderExecutingAsync(DbCommand command,
+ CommandEventData eventData, InterceptionResult result, CancellationToken cancellationToken = default)
+ {
+ Starting(command);
+ return ValueTask.FromResult(result);
+ }
+
+ public override InterceptionResult DataReaderDisposing(DbCommand command, DataReaderDisposingEventData eventData,
+ InterceptionResult result)
+ {
+ Ended(command);
+ return result;
+ }
+
+ public override InterceptionResult NonQueryExecuting(DbCommand command, CommandEventData eventData,
+ InterceptionResult result)
+ {
+ Starting(command);
+ return result;
+ }
+
+ public override ValueTask> NonQueryExecutingAsync(DbCommand command,
+ CommandEventData eventData, InterceptionResult result, CancellationToken cancellationToken = default)
+ {
+ Starting(command);
+ return ValueTask.FromResult(result);
+ }
+
+ public override int NonQueryExecuted(DbCommand command, CommandExecutedEventData eventData, int result)
+ {
+ Ended(command);
+ return result;
+ }
+
+ public override ValueTask NonQueryExecutedAsync(DbCommand command, CommandExecutedEventData eventData,
+ int result, CancellationToken cancellationToken = default)
+ {
+ Ended(command);
+ return ValueTask.FromResult(result);
+ }
+
+ public override InterceptionResult ScalarExecuting(DbCommand command, CommandEventData eventData,
+ InterceptionResult result)
+ {
+ Starting(command);
+ return result;
+ }
+
+ public override ValueTask> ScalarExecutingAsync(DbCommand command,
+ CommandEventData eventData, InterceptionResult result, CancellationToken cancellationToken = default)
+ {
+ Starting(command);
+ return ValueTask.FromResult(result);
+ }
+
+ public override object ScalarExecuted(DbCommand command, CommandExecutedEventData eventData, object result)
+ {
+ Ended(command);
+ return result;
+ }
+
+ public override ValueTask ScalarExecutedAsync(DbCommand command, CommandExecutedEventData eventData,
+ object result, CancellationToken cancellationToken = default)
+ {
+ Ended(command);
+ return ValueTask.FromResult(result);
+ }
+
+ public override void CommandFailed(DbCommand command, CommandErrorEventData eventData) => Ended(command);
+
+ public override Task CommandFailedAsync(DbCommand command, CommandErrorEventData eventData,
+ CancellationToken cancellationToken = default)
+ {
+ Ended(command);
+ return Task.CompletedTask;
+ }
+
+ public override void CommandCanceled(DbCommand command, CommandEndEventData eventData) => Ended(command);
+
+ public override Task CommandCanceledAsync(DbCommand command, CommandEndEventData eventData,
+ CancellationToken cancellationToken = default)
+ {
+ Ended(command);
+ return Task.CompletedTask;
+ }
+}
diff --git a/SW.Bitween.Api/Services/Background/BackgroundRequests.cs b/SW.Bitween.Api/Services/Background/BackgroundRequests.cs
new file mode 100644
index 00000000..cfc78195
--- /dev/null
+++ b/SW.Bitween.Api/Services/Background/BackgroundRequests.cs
@@ -0,0 +1,615 @@
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.IO;
+using System.Linq;
+using System.Text;
+using System.Text.Json;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.AspNetCore.Http;
+using Microsoft.AspNetCore.SignalR;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
+using Microsoft.Net.Http.Headers;
+using SW.Bitween.Domain;
+using SW.Bitween.Services.Cluster;
+using SW.PrimitiveTypes;
+
+namespace SW.Bitween.Services.Background;
+
+///
+/// Runs the work the admin UI asks to carry on - prod's list to bring from, a bulk retry, a zip of
+/// exchange files - and keeps going with whatever outlasts the wait the UI asked for: that becomes a
+/// , whose answer is kept for the UI to collect, and whose owner is told
+/// when it changes - by SignalR, on whichever node holds their connection.
+///
+///
+///
+/// A request starts the moment it arrives, so one that answers in time is answered exactly as it would
+/// have been. What it may not do is run in the background beyond on a node, or
+/// of one person's: they are long database work, on the database exchanges are
+/// processed on. A read that would go past either is stopped and asked again once a place frees up; a
+/// write can't be asked twice, so it goes on regardless.
+///
+///
+/// A singleton, and has to be: it holds the places, and the requests running here.
+///
+///
+public sealed class BackgroundRequests(IServiceScopeFactory scopes, IHubContext hub,
+ ICloudFilesService cloudFiles, ILogger logger)
+{
+ /// The longest a background request may go without saying it got further.
+ public static readonly TimeSpan Limit = TimeSpan.FromMinutes(10);
+
+ /// How long an answer is kept for its owner to collect.
+ public static readonly TimeSpan KeptFor = TimeSpan.FromMinutes(30);
+
+ public const int MaxRunning = 10;
+ public const int MaxPerOwner = 3;
+
+ /// Answers up to this size are kept in the database; larger ones, and files, in storage.
+ public const int KeptInline = 1024 * 1024;
+
+ /// Where answers kept in storage go, each under its request's id.
+ public const string StoragePrefix = "background-requests/";
+
+ static readonly TimeSpan ProgressEvery = TimeSpan.FromSeconds(1);
+
+ /// This instance, among those a change is broadcast to; unique even for two in one process.
+ public string Node { get; } = $"{NodeHeartbeat.NodeName}/{Guid.NewGuid():N}";
+
+ readonly SemaphoreSlim places = new(MaxRunning, MaxRunning);
+ readonly ConcurrentDictionary ownerPlaces = new();
+ readonly ConcurrentDictionary background = new();
+
+ /// Starts answering , by running what follows in the pipeline.
+ internal Work Start(AskedRequest asked, RequestDelegate next, string owner, string label, string page)
+ {
+ var work = new Work(asked, next, owner, label, page);
+ // Its own flow from the start: nothing set on the request's - the context accessor's holder
+ // above all, which is cleared when the request ends - may reach it.
+ using (ExecutionContext.SuppressFlow())
+ _ = Task.Run(() => RunAsync(work));
+ return work;
+ }
+
+ ///
+ /// Keeps going after its caller stopped waiting: records it, and gives it a
+ /// place - or, for a read, stops it until one frees up.
+ ///
+ internal async Task KeepAsync(Work work)
+ {
+ try
+ {
+ await using var scope = scopes.CreateAsyncScope();
+ var db = scope.ServiceProvider.GetRequiredService();
+ await ForgetExpiredAsync(db);
+
+ var row = new BackgroundRequest(work.Id, work.Owner, Clip(work.Asked.Described, 4000),
+ Clip(work.Label, 200), Clip(work.Page, 2000), NodeHeartbeat.NodeName);
+ if (TryTakePlaces(work) || !work.Asked.IsRead) row.Run();
+ else work.Requeue();
+ if (work.Total is { } total) row.Progress(work.Done, total, Clip(work.Summary, 500));
+
+ background[work.Id] = work;
+ db.Add(row);
+ await db.SaveChangesAsync();
+ work.Kept.TrySetResult(true);
+ await NotifyAsync(row);
+ return row;
+ }
+ catch (Exception ex)
+ {
+ // Nobody could collect it, so it must not go on.
+ work.Cancel(hard: true);
+ background.TryRemove(work.Id, out _);
+ ReleasePlaces(work);
+ work.Kept.TrySetResult(false);
+ logger.LogError(ex, "Couldn't keep a background request for {Path}.", work.Asked.Described);
+ throw;
+ }
+ }
+
+ /// Stops a request running on this node; false when it isn't running here.
+ public bool CancelHere(string id)
+ {
+ if (!background.TryGetValue(id, out var work)) return false;
+ work.Cancel();
+ return true;
+ }
+
+ ///
+ /// Stops one of 's requests, wherever it runs. False when there is no such
+ /// request of theirs.
+ ///
+ ///
+ /// Work that counts its steps ends itself, at the next one, and says how far it got; anything else
+ /// ends here and now.
+ ///
+ public async Task CancelAsync(BitweenDbContext db, string owner, string id)
+ {
+ var row = await db.Set().FirstOrDefaultAsync(r => r.Id == id && r.Owner == owner);
+ if (row == null) return false;
+ if (row.Finished) return true;
+
+ // Here, the run itself says whether it counts: its first count may not be saved yet.
+ if (background.TryGetValue(id, out var work))
+ {
+ work.Cancel();
+ if (work.Counts) return true;
+ }
+ else
+ {
+ await BroadcastAsync(new BackgroundRequestCancel { Id = id });
+ if (row.Total != null) return true;
+ }
+
+ row.End(BackgroundRequestStatus.Cancelled);
+ await db.SaveChangesAsync();
+ await NotifyAsync(row);
+ return true;
+ }
+
+ ///
+ /// Ends whichever of these is still running on a node that has gone, or has restarted since it
+ /// started it: those will never finish.
+ ///
+ public static async Task EndStoppedAsync(BitweenDbContext db, IReadOnlyCollection rows)
+ {
+ var unfinished = rows.Where(r => !r.Finished).ToList();
+ if (unfinished.Count == 0) return;
+
+ var names = unfinished.Select(r => r.Node).Distinct().ToList();
+ var nodes = await db.Set().AsNoTracking()
+ .Where(n => names.Contains(n.Id)).ToDictionaryAsync(n => n.Id);
+ var now = DateTime.UtcNow;
+ var ended = false;
+ foreach (var row in unfinished)
+ {
+ var stopped = nodes.TryGetValue(row.Node, out var node)
+ ? node.LastSeenOn < now - NodeHeartbeat.GoneAfter || node.StartedOn > row.StartedOn
+ // A node writes its first beat as it starts; a request that new gets the same grace.
+ : row.StartedOn < now - NodeHeartbeat.GoneAfter;
+ if (!stopped) continue;
+ row.End(BackgroundRequestStatus.Stopped);
+ ended = true;
+ }
+
+ if (ended) await db.SaveChangesAsync();
+ }
+
+ ///
+ /// Deletes what is no longer kept - finished longer ago than , or started more
+ /// than a day ago - with the answers it kept in storage. On every read of them too, so nothing is
+ /// served past that when no new request comes along to clear it.
+ ///
+ internal async Task ForgetExpiredAsync(BitweenDbContext db)
+ {
+ var now = DateTime.UtcNow;
+ var expired = db.Set()
+ .Where(r => r.StartedOn < now.AddDays(-1) || (r.FinishedOn != null && r.FinishedOn < now - KeptFor));
+ foreach (var key in await expired.Where(r => r.ResultKey != null).Select(r => r.ResultKey).ToListAsync())
+ {
+ try
+ {
+ await cloudFiles.DeleteAsync(key);
+ }
+ catch (Exception ex)
+ {
+ logger.LogWarning(ex, "Couldn't delete a background request's answer, {Key}.", key);
+ }
+ }
+
+ await expired.ExecuteDeleteAsync();
+ }
+
+ async Task RunAsync(Work work)
+ {
+ Answer answer = null;
+ try
+ {
+ answer = await AttemptAsync(work, work.FirstAttempt);
+ if (work.TryAnswerHere())
+ {
+ work.AnsweredHere.TrySetResult(answer);
+ answer = null;
+ return;
+ }
+
+ if (!await work.Kept.Task) return;
+ try
+ {
+ // A read stopped for want of a place rather than finished: asked again once one frees up.
+ if (answer.Kind == AnswerKind.Requeued)
+ {
+ try
+ {
+ await TakePlacesAsync(work);
+ }
+ catch (OperationCanceledException)
+ {
+ await SaveAsync(work, Answer.Cancelled);
+ return;
+ }
+
+ await SetRunningAsync(work);
+ using var attempt = new CancellationTokenSource();
+ answer = await AttemptAsync(work, attempt);
+ }
+
+ await SaveAsync(work, answer);
+ }
+ finally
+ {
+ ReleasePlaces(work);
+ background.TryRemove(work.Id, out _);
+ }
+ }
+ catch (Exception ex)
+ {
+ logger.LogError(ex, "A background request for {Path} failed.", work.Asked.Described);
+ // Its caller still waits, and is told; or it moved, and its record says it failed.
+ if (work.TryAnswerHere()) work.AnsweredHere.TrySetException(ex);
+ else if (work.IsMoved && await work.Kept.Task)
+ {
+ try
+ {
+ await SaveAsync(work, Answer.Failed);
+ }
+ catch (Exception saving)
+ {
+ logger.LogError(saving, "Couldn't record that a background request failed.");
+ }
+ }
+ }
+ finally
+ {
+ if (answer?.Body != null) await answer.Body.DisposeAsync();
+ }
+ }
+
+ ///
+ /// Asks the request again, in a scope of its own, with to get anywhere and
+ /// able to end it sooner.
+ ///
+ async Task AttemptAsync(Work work, CancellationTokenSource stop)
+ {
+ work.Attempt = stop;
+ stop.CancelAfter(Limit);
+ BackgroundRequestScope.Open(DateTime.UtcNow + Limit, stop.Token, work.CancelRequested,
+ (done, total, summary) => Progressed(work, stop, done, total, summary));
+
+ await using var scope = scopes.CreateAsyncScope();
+ // On disk rather than in memory: an answer can be a zip of hundreds of exchanges' files.
+ var body = new FileStream(Path.Combine(Path.GetTempPath(), $"bitween-background-{work.Id}-{work.Attempts + 1}"),
+ FileMode.Create, FileAccess.ReadWrite, FileShare.None, 81920, FileOptions.DeleteOnClose | FileOptions.Asynchronous);
+ var context = work.Asked.NewContext(scope.ServiceProvider, body, stop.Token, ++work.Attempts);
+ var accessor = scope.ServiceProvider.GetService();
+ if (accessor != null) accessor.HttpContext = context;
+
+ try
+ {
+ await work.Next(context);
+ await context.Response.CompleteAsync();
+ }
+ catch (Exception) when (stop.IsCancellationRequested)
+ {
+ await body.DisposeAsync();
+ return work.CancelRequested.IsCancellationRequested ? Answer.Cancelled
+ : work.Requeued ? Answer.Requeued
+ : Answer.TimedOut;
+ }
+ catch (Exception ex)
+ {
+ await body.DisposeAsync();
+ logger.LogError(ex, "A background request for {Path} threw.", work.Asked.Described);
+ return Answer.Failed;
+ }
+ finally
+ {
+ if (accessor != null) accessor.HttpContext = null;
+ }
+
+ body.Position = 0;
+ // The answer the web host would have rewritten on its way out; see Startup.
+ var status = context.Response.StatusCode == StatusCodes.Status401Unauthorized &&
+ context.User.Identity?.IsAuthenticated == true
+ ? StatusCodes.Status403Forbidden
+ : context.Response.StatusCode;
+ return new Answer(AnswerKind.Answered, status, context.Response.ContentType,
+ context.Response.Headers[HeaderNames.ContentDisposition].FirstOrDefault(), context.Response.Headers, body);
+ }
+
+ ///
+ /// Work said how far it got: it is still going, so its limit starts again; and its owner hears,
+ /// once a second at most.
+ ///
+ void Progressed(Work work, CancellationTokenSource stop, int done, int total, string summary)
+ {
+ work.Counts = true;
+ try
+ {
+ stop.CancelAfter(Limit);
+ }
+ catch (ObjectDisposedException)
+ {
+ // The attempt it was for is over.
+ }
+
+ work.Done = done;
+ work.Total = total;
+ work.Summary = summary;
+ var now = DateTime.UtcNow;
+ if (now - work.ToldOn < ProgressEvery || !work.Kept.Task.IsCompletedSuccessfully || !work.Kept.Task.Result)
+ return;
+ work.ToldOn = now;
+
+ // Off the work's own flow, so it neither waits for this nor has its limits applied to it.
+ using (ExecutionContext.SuppressFlow())
+ _ = Task.Run(async () =>
+ {
+ try
+ {
+ await using var scope = scopes.CreateAsyncScope();
+ var db = scope.ServiceProvider.GetRequiredService();
+ var row = await db.Set().FindAsync(work.Id);
+ if (row == null || row.Finished) return;
+ row.Progress(done, total, Clip(summary, 500));
+ await db.SaveChangesAsync();
+ await NotifyAsync(row);
+ }
+ catch (Exception ex)
+ {
+ logger.LogWarning(ex, "Couldn't record how far a background request got.");
+ }
+ });
+ }
+
+ async Task SaveAsync(Work work, Answer answer)
+ {
+ await using var scope = scopes.CreateAsyncScope();
+ var db = scope.ServiceProvider.GetRequiredService();
+ var row = await db.Set().FindAsync(work.Id);
+ // Gone, or ended meanwhile: nothing to tell anyone.
+ if (row == null || row.Finished) return;
+ if (work.Total is { } total) row.Progress(work.Done, total, Clip(work.Summary, 500));
+
+ switch (answer.Kind)
+ {
+ case AnswerKind.Answered:
+ var file = answer.Disposition != null || answer.Body.Length > KeptInline;
+ string key = null;
+ byte[] inline = null;
+ if (file)
+ {
+ key = StoragePrefix + work.Id;
+ await cloudFiles.WriteAsync(answer.Body, new WriteFileSettings
+ {
+ Key = key,
+ ContentType = answer.ContentType ?? "application/octet-stream"
+ });
+ }
+ else
+ {
+ using var copy = new MemoryStream();
+ await answer.Body.CopyToAsync(copy);
+ inline = copy.ToArray();
+ }
+
+ row.Answer(answer.Status, answer.ContentType, answer.Disposition, inline, key,
+ cancelled: work.CancelRequested.IsCancellationRequested);
+ break;
+ case AnswerKind.Failed:
+ row.Answer(StatusCodes.Status500InternalServerError, "application/json", null,
+ Problem("BACKGROUND_FAILED", "It failed while running. The log of the node that ran it says why."), null);
+ break;
+ case AnswerKind.TimedOut:
+ row.End(BackgroundRequestStatus.TimedOut);
+ break;
+ default:
+ row.End(BackgroundRequestStatus.Cancelled);
+ break;
+ }
+
+ await db.SaveChangesAsync();
+ await NotifyAsync(row);
+ }
+
+ async Task SetRunningAsync(Work work)
+ {
+ await using var scope = scopes.CreateAsyncScope();
+ var db = scope.ServiceProvider.GetRequiredService();
+ var row = await db.Set().FindAsync(work.Id);
+ if (row == null || row.Finished) return;
+ row.Run();
+ await db.SaveChangesAsync();
+ await NotifyAsync(row);
+ }
+
+ /// A body as the UI reads a refusal: the code, with what to tell the person.
+ internal static byte[] Problem(string code, string message) =>
+ Encoding.UTF8.GetBytes(JsonSerializer.Serialize(new Dictionary { [code] = [message] }));
+
+ bool TryTakePlaces(Work work)
+ {
+ var owners = ownerPlaces.GetOrAdd(work.Owner, _ => new SemaphoreSlim(MaxPerOwner, MaxPerOwner));
+ if (!owners.Wait(0)) return false;
+ if (places.Wait(0))
+ {
+ work.HoldsPlaces = true;
+ return true;
+ }
+
+ owners.Release();
+ return false;
+ }
+
+ async Task TakePlacesAsync(Work work)
+ {
+ var owners = ownerPlaces.GetOrAdd(work.Owner, _ => new SemaphoreSlim(MaxPerOwner, MaxPerOwner));
+ await owners.WaitAsync(work.CancelRequested);
+ try
+ {
+ await places.WaitAsync(work.CancelRequested);
+ }
+ catch
+ {
+ owners.Release();
+ throw;
+ }
+
+ work.HoldsPlaces = true;
+ }
+
+ void ReleasePlaces(Work work)
+ {
+ if (!work.HoldsPlaces) return;
+ work.HoldsPlaces = false;
+ places.Release();
+ ownerPlaces[work.Owner].Release();
+ }
+
+ ///
+ /// Tells the owner's connections here, then every other node, whose listener tells theirs. Losing
+ /// either loses only the push: the UI asks again when it reconnects.
+ ///
+ internal async Task NotifyAsync(BackgroundRequest row)
+ {
+ var change = BackgroundRequestChanged.Of(row, Node);
+ await PushAsync(change);
+ await BroadcastAsync(change);
+ }
+
+ internal async Task PushAsync(BackgroundRequestChanged change)
+ {
+ try
+ {
+ await hub.Clients.User(change.Owner).SendAsync(BackgroundRequestHub.Changed, new
+ {
+ id = change.Id,
+ status = change.Status.ToString(),
+ done = change.Done,
+ total = change.Total,
+ summary = change.Summary,
+ });
+ }
+ catch (Exception ex)
+ {
+ logger.LogWarning(ex, "Couldn't tell a browser about a background request.");
+ }
+ }
+
+ // Generic, not object: SW-Bus names a broadcast by the type it is called with, and a listener
+ // only hears its own type's name. Called with object, every node dropped it without a word.
+ async Task BroadcastAsync(TMessage message)
+ {
+ try
+ {
+ await using var scope = scopes.CreateAsyncScope();
+ await scope.ServiceProvider.GetRequiredService().Broadcast(message);
+ }
+ catch (Exception ex)
+ {
+ logger.LogWarning(ex, "Couldn't tell the other nodes about a background request.");
+ }
+ }
+
+ static string Clip(string value, int length) =>
+ value == null || value.Length <= length ? value : value[..length];
+
+ internal enum AnswerKind
+ {
+ Answered,
+ Failed,
+ Cancelled,
+ TimedOut,
+ Requeued,
+ }
+
+ /// What an attempt came to. is a temporary file, deleted when disposed.
+ internal sealed record Answer(AnswerKind Kind, int Status = 0, string ContentType = null, string Disposition = null,
+ IHeaderDictionary Headers = null, Stream Body = null)
+ {
+ public static readonly Answer Failed = new(AnswerKind.Failed);
+ public static readonly Answer Cancelled = new(AnswerKind.Cancelled);
+ public static readonly Answer TimedOut = new(AnswerKind.TimedOut);
+ public static readonly Answer Requeued = new(AnswerKind.Requeued);
+ }
+
+ /// One request, from when it arrives until it is answered or ends.
+ internal sealed class Work(AskedRequest asked, RequestDelegate next, string owner, string label, string page)
+ {
+ const int Waited = 0, Moved = 1, Answered = 2;
+ int state;
+ readonly CancellationTokenSource cancelRequested = new();
+
+ public string Id { get; } = Guid.NewGuid().ToString("N");
+ public AskedRequest Asked { get; } = asked;
+ public RequestDelegate Next { get; } = next;
+ public string Owner { get; } = owner;
+ public string Label { get; } = label;
+ public string Page { get; } = page;
+ public int Attempts { get; set; }
+ public bool HoldsPlaces { get; set; }
+ public bool Requeued { get; private set; }
+
+ /// It has said how far it got, so a cancel lets it end at its next step.
+ public bool Counts { get; set; }
+
+ public int Done { get; set; }
+ public int? Total { get; set; }
+ public string Summary { get; set; }
+ public DateTime ToldOn { get; set; }
+
+ /// Its owner's cancel, or its caller leaving before it moved to the background.
+ public CancellationToken CancelRequested => cancelRequested.Token;
+
+ /// What stops the first attempt: a cancel, a requeue, or the limit.
+ public CancellationTokenSource FirstAttempt { get; } = new();
+
+ /// What stops the attempt under way.
+ public CancellationTokenSource Attempt { get; set; }
+
+ /// The answer, when it came while the caller still waited.
+ public TaskCompletionSource AnsweredHere { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
+
+ /// True once its record exists, false when keeping it failed.
+ public TaskCompletionSource Kept { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously);
+
+ /// Claims the answer for the caller still waiting; false once it moved to the background.
+ public bool TryAnswerHere() => Interlocked.CompareExchange(ref state, Answered, Waited) == Waited;
+
+ /// Claims it for the background; false once its caller has been answered.
+ public bool TryMove() => Interlocked.CompareExchange(ref state, Moved, Waited) == Waited;
+
+ public bool IsMoved => Volatile.Read(ref state) == Moved;
+
+ ///
+ /// Stops it: where it is, unless it counts its steps - then at the next one, unless
+ /// .
+ ///
+ public void Cancel(bool hard = false)
+ {
+ cancelRequested.Cancel();
+ if (Counts && !hard) return;
+ FirstAttempt.Cancel();
+ try
+ {
+ Attempt?.Cancel();
+ }
+ catch (ObjectDisposedException)
+ {
+ // That attempt is over.
+ }
+ }
+
+ /// Stops the first attempt, to ask again once there is a place for it.
+ public void Requeue()
+ {
+ Requeued = true;
+ FirstAttempt.Cancel();
+ }
+ }
+}
diff --git a/SW.Bitween.HttpTests/BackgroundRequestTests.cs b/SW.Bitween.HttpTests/BackgroundRequestTests.cs
new file mode 100644
index 00000000..f5adf189
--- /dev/null
+++ b/SW.Bitween.HttpTests/BackgroundRequestTests.cs
@@ -0,0 +1,380 @@
+using System;
+using System.Linq;
+using System.Net;
+using System.Net.Http;
+using System.Net.Http.Headers;
+using System.Net.Http.Json;
+using System.Security.Claims;
+using System.Text;
+using System.Text.Json;
+using System.Text.Json.Nodes;
+using System.Threading.Tasks;
+using Microsoft.AspNetCore.Http.Connections;
+using Microsoft.AspNetCore.Mvc;
+using Microsoft.AspNetCore.Mvc.Testing;
+using Microsoft.AspNetCore.SignalR.Client;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.IdentityModel.JsonWebTokens;
+using Microsoft.IdentityModel.Tokens;
+using Xunit;
+
+namespace SW.Bitween.HttpTests;
+
+///
+/// The admin UI's long reads, carried on in the background when it sends Prefer: respond-async .
+/// On a node whose database commands time out after two seconds, so that "longer than the usual
+/// limit" takes seconds to show rather than half a minute.
+///
+[Collection("Http")]
+public class BackgroundRequestTests(HttpFixture fixture) : IAsyncLifetime
+{
+ WebApplicationFactory node = null!;
+
+ public Task InitializeAsync()
+ {
+ node = fixture.SecondNode(
+ services => services.AddControllers().AddApplicationPart(typeof(SlowReadsController).Assembly),
+ settings => settings["ConnectionStrings:BitweenDb"] += ";Command Timeout=2");
+ return Task.CompletedTask;
+ }
+
+ public async Task DisposeAsync() => await node.DisposeAsync();
+
+ [Fact]
+ public async Task A_read_answered_within_the_wait_is_answered_as_usual()
+ {
+ using var admin = await HttpFixture.AdminOf(node);
+ var before = (await RowsAsync()).Length;
+
+ var response = await admin.SendAsync(Read("/api/slowreads/0", wait: 5));
+
+ Assert.Equal(HttpStatusCode.OK, response.StatusCode);
+ Assert.Equal(0, (double)(await Api.Json(response))["seconds"]!);
+ Assert.Equal(before, (await RowsAsync()).Length);
+ }
+
+ [Fact]
+ public async Task A_read_that_does_not_ask_keeps_the_usual_limit()
+ {
+ using var admin = await HttpFixture.AdminOf(node);
+
+ var response = await admin.SendAsync(Read("/api/slowreads/3"));
+
+ Assert.Equal(HttpStatusCode.InternalServerError, response.StatusCode);
+ }
+
+ [Fact]
+ public async Task A_read_that_outlasts_the_wait_runs_past_the_usual_limit_and_its_answer_is_kept()
+ {
+ using var admin = await HttpFixture.AdminOf(node);
+
+ var accepted = await admin.SendAsync(Read("/api/slowreads/3", wait: 0, label: "Exchanges: failed, last 7 days",
+ page: "/exchanges?status=failed"));
+
+ Assert.Equal(HttpStatusCode.Accepted, accepted.StatusCode);
+ var id = (string)(await Api.Json(accepted))["id"]!;
+ Assert.Equal($"/api/background/{id}", accepted.Headers.Location?.ToString());
+
+ var row = await FinishedAsync(admin, id);
+ Assert.Equal("Done", (string)row["status"]!);
+ Assert.Equal("Exchanges: failed, last 7 days", (string)row["label"]!);
+ Assert.Equal("/exchanges?status=failed", (string)row["page"]!);
+ Assert.Equal("/api/slowreads/3", (string)row["path"]!);
+
+ var result = await admin.GetAsync($"/api/background/{id}/result");
+ Assert.Equal(3, (double)(await Api.Json(result))["seconds"]!);
+ }
+
+ [Fact]
+ public async Task An_api_read_in_the_background_answers_as_it_would_have_directly()
+ {
+ using var admin = await HttpFixture.AdminOf(node);
+
+ // A list, and a refusal CqApi turns into a 404 of its own.
+ foreach (var path in new[] { "/api/subscriptions", "/api/subscriptions/987654" })
+ {
+ var direct = await admin.GetAsync(path);
+
+ // Made slow by holding its table: the read waits for the lock, past the caller's wait.
+ var accepted = await fixture.InDbAsync(async db =>
+ {
+ await using var transaction = await db.Database.BeginTransactionAsync();
+ await db.Database.ExecuteSqlRawAsync("LOCK TABLE infolink.subscription IN ACCESS EXCLUSIVE MODE");
+ var response = await admin.SendAsync(Read(path, wait: 1));
+ await transaction.RollbackAsync();
+ return response;
+ });
+
+ Assert.Equal(HttpStatusCode.Accepted, accepted.StatusCode);
+ var id = (string)(await Api.Json(accepted))["id"]!;
+ Assert.Equal("Done", (string)(await FinishedAsync(admin, id))["status"]!);
+ var kept = await admin.GetAsync($"/api/background/{id}/result");
+ Assert.Equal(direct.StatusCode, kept.StatusCode);
+ Assert.Equal(direct.Content.Headers.ContentType?.MediaType, kept.Content.Headers.ContentType?.MediaType);
+ Assert.Equal(await direct.Content.ReadAsStringAsync(), await kept.Content.ReadAsStringAsync());
+ }
+ }
+
+ [Fact]
+ public async Task Cancelling_stops_the_query_the_database_is_running()
+ {
+ using var admin = await HttpFixture.AdminOf(node);
+ var id = (string)(await Api.Json(await admin.SendAsync(Read("/api/slowreads/60", wait: 0))))["id"]!;
+ await SleepingAsync(1);
+
+ Assert.Equal(HttpStatusCode.NoContent, (await admin.DeleteAsync($"/api/background/{id}")).StatusCode);
+
+ await SleepingAsync(0);
+ Assert.Equal("Cancelled", (string)(await FinishedAsync(admin, id))["status"]!);
+ Assert.Equal(HttpStatusCode.Gone, (await admin.GetAsync($"/api/background/{id}/result")).StatusCode);
+ }
+
+ [Fact]
+ public async Task Three_of_a_persons_run_at_once_and_the_next_waits_its_turn()
+ {
+ using var admin = await HttpFixture.AdminOf(node);
+
+ var started = await Task.WhenAll(Enumerable.Range(0, 4).Select(async _ =>
+ await Api.Json(await admin.SendAsync(Read("/api/slowreads/2", wait: 0)))));
+
+ Assert.Equal(["Running", "Running", "Running", "Waiting"], started.Select(s => (string)s["status"]!).Order());
+ foreach (var request in started)
+ Assert.Equal("Done", (string)(await FinishedAsync(admin, (string)request["id"]!))["status"]!);
+ }
+
+ [Fact]
+ public async Task Its_owner_hears_it_finish_on_another_node()
+ {
+ using var admin = await HttpFixture.AdminOf(node);
+ var heard = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+ await using var hub = new HubConnectionBuilder()
+ .WithUrl("https://localhost/api/background/hub", options =>
+ {
+ options.Transports = HttpTransportType.LongPolling;
+ options.HttpMessageHandlerFactory = _ => fixture.App.Server.CreateHandler();
+ options.AccessTokenProvider = () => Task.FromResult(admin.DefaultRequestHeaders.Authorization!.Parameter);
+ })
+ .Build();
+ hub.On("changed", change =>
+ {
+ if (change.GetProperty("status").GetString() == "Done") heard.TrySetResult(change.GetProperty("id").GetString()!);
+ });
+ await hub.StartAsync();
+
+ var id = (string)(await Api.Json(await admin.SendAsync(Read("/api/slowreads/1", wait: 0))))["id"]!;
+
+ Assert.Equal(id, await heard.Task.WaitAsync(TimeSpan.FromSeconds(30)));
+ }
+
+ [Fact]
+ public async Task Nobody_else_sees_it_or_its_answer()
+ {
+ using var admin = await HttpFixture.AdminOf(node);
+ var id = (string)(await Api.Json(await admin.SendAsync(Read("/api/slowreads/1", wait: 0))))["id"]!;
+
+ using var other = HttpFixture.ClientOf(node);
+ other.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Bearer", TokenFor("987654"));
+
+ Assert.DoesNotContain((await Api.Json(await other.GetAsync("/api/background"))).AsArray(),
+ r => (string)r!["id"]! == id);
+ Assert.Equal(HttpStatusCode.NotFound, (await other.GetAsync($"/api/background/{id}/result")).StatusCode);
+ Assert.Equal(HttpStatusCode.NotFound, (await other.DeleteAsync($"/api/background/{id}")).StatusCode);
+ Assert.Equal("Done", (string)(await FinishedAsync(admin, id))["status"]!);
+ }
+
+ [Fact]
+ public async Task One_on_a_node_that_has_gone_shows_as_stopped()
+ {
+ using var admin = await HttpFixture.AdminOf(node);
+ var real = (string)(await Api.Json(await admin.SendAsync(Read("/api/slowreads/0", wait: 0))))["id"]!;
+ await FinishedAsync(admin, real);
+ var owner = (await RowsAsync()).Single(r => r.Id == real).Owner;
+
+ var id = Guid.NewGuid().ToString("N");
+ var startedOn = DateTime.UtcNow.AddMinutes(-5);
+ await fixture.InDbAsync(db => db.Database.ExecuteSqlInterpolatedAsync(
+ $"INSERT INTO infolink.background_request (id, owner, path, node, status, started_on) VALUES ({id}, {owner}, {"/api/xchanges"}, {"gone-node:1"}, 1, {startedOn})"));
+
+ var row = (await Api.Json(await admin.GetAsync("/api/background"))).AsArray().Single(r => (string)r!["id"]! == id)!;
+ Assert.Equal("Stopped", (string)row["status"]!);
+ Assert.Equal(HttpStatusCode.ServiceUnavailable, (await admin.GetAsync($"/api/background/{id}/result")).StatusCode);
+ }
+
+ [Fact]
+ public async Task Work_that_counts_its_steps_says_how_far_it_got_and_a_cancel_keeps_what_it_did()
+ {
+ using var admin = await HttpFixture.AdminOf(node);
+ var request = new HttpRequestMessage(HttpMethod.Post, "/api/slowreads/steps") { Content = JsonContent.Create(new { count = 100 }) };
+ request.Headers.TryAddWithoutValidation("Prefer", "respond-async, wait=0");
+ var accepted = await admin.SendAsync(request);
+ Assert.Equal(HttpStatusCode.Accepted, accepted.StatusCode);
+ var id = (string)(await Api.Json(accepted))["id"]!;
+
+ // Said once a second at most, so seen within a few.
+ JsonNode? row = null;
+ for (var i = 0; i < 50 && (row == null || (int?)row["done"] is null or 0); i++)
+ {
+ await Task.Delay(100);
+ row = (await Api.Json(await admin.GetAsync("/api/background"))).AsArray().Single(r => (string)r!["id"]! == id);
+ }
+ Assert.Equal(100, (int)row!["total"]!);
+ Assert.True((int)row["done"]! > 0);
+ Assert.Equal("POST /api/slowreads/steps", (string)row["path"]!);
+
+ Assert.Equal(HttpStatusCode.NoContent, (await admin.DeleteAsync($"/api/background/{id}")).StatusCode);
+
+ var ended = await FinishedAsync(admin, id);
+ Assert.Equal("Cancelled", (string)ended["status"]!);
+ var result = await admin.GetAsync($"/api/background/{id}/result");
+ var stoppedAt = (int)(await Api.Json(result))["done"]!;
+ Assert.InRange(stoppedAt, 1, 99);
+ }
+
+ [Fact]
+ public async Task An_export_in_the_background_is_kept_in_storage_and_handed_over_as_the_file()
+ {
+ var (admin, urlName, _, partnerId, subscriptionId) = await Api.GatewayAsync(fixture);
+ var key = await Api.NewKeyAsync(admin);
+ await Api.SetKeysAsync(admin, partnerId, ("export", key));
+ foreach (var n in new[] { 1, 2 })
+ {
+ using var partner = fixture.Client();
+ using var call = new HttpRequestMessage(HttpMethod.Post, $"/api/gateway/{urlName}/async")
+ {
+ Content = new StringContent($"{{\"order\":{n}}}", Encoding.UTF8, "application/json")
+ };
+ call.Headers.Add("partnerkey", key);
+ Assert.Equal(HttpStatusCode.Accepted, (await partner.SendAsync(call)).StatusCode);
+ }
+ // An async call's exchange is made off the call, so it is waited for.
+ var ids = new System.Collections.Generic.List();
+ for (var i = 0; i < 100 && ids.Count < 2; i++)
+ {
+ if (i > 0) await Task.Delay(100);
+ ids = await fixture.InDbAsync(db => db.Set().Where(x => x.SubscriptionId == subscriptionId)
+ .Select(x => x.Id).ToListAsync());
+ }
+ Assert.Equal(2, ids.Count);
+
+ using var owner = await HttpFixture.AdminOf(node);
+ var request = new HttpRequestMessage(HttpMethod.Post, "/api/xchanges/export") { Content = JsonContent.Create(new { ids }) };
+ request.Headers.TryAddWithoutValidation("Prefer", "respond-async, wait=0");
+ var accepted = await owner.SendAsync(request);
+ Assert.Equal(HttpStatusCode.Accepted, accepted.StatusCode);
+ var id = (string)(await Api.Json(accepted))["id"]!;
+ Assert.Equal("Done", (string)(await FinishedAsync(owner, id))["status"]!);
+
+ var kept = await fixture.InDbAsync(db => db.Set().AsNoTracking().SingleAsync(r => r.Id == id));
+ Assert.Equal($"background-requests/{id}", kept.ResultKey);
+ Assert.Null(kept.Result);
+
+ // Downloaded from a link the owner is given, by a browser that sends no sign-in.
+ var link = (string)(await Api.Json(await owner.PostAsync($"/api/background/{id}/link", null)))["url"]!;
+ using var browser = HttpFixture.ClientOf(node);
+ var file = await browser.GetAsync(link);
+ Assert.Equal(HttpStatusCode.OK, file.StatusCode);
+ Assert.Equal("application/zip", file.Content.Headers.ContentType?.MediaType);
+ Assert.StartsWith("exchanges-", file.Content.Headers.ContentDisposition?.FileName?.Trim('"'));
+ using var zip = new System.IO.Compression.ZipArchive(await file.Content.ReadAsStreamAsync());
+ Assert.Equal(2, zip.Entries.Count(e => e.FullName.Contains("/input")));
+
+ // A link changed by a character, or to another request, opens nothing.
+ Assert.Equal(HttpStatusCode.NotFound, (await browser.GetAsync(link[..^1] + (link[^1] == 'A' ? 'B' : 'A'))).StatusCode);
+ Assert.Equal(HttpStatusCode.NotFound, (await browser.GetAsync(link.Replace(id, Guid.NewGuid().ToString("N")))).StatusCode);
+ // And nobody else is handed one.
+ using var other = HttpFixture.ClientOf(node);
+ other.DefaultRequestHeaders.Authorization = new AuthenticationHeaderValue("Bearer", TokenFor("987654"));
+ Assert.Equal(HttpStatusCode.NotFound, (await other.PostAsync($"/api/background/{id}/link", null)).StatusCode);
+ }
+
+ static HttpRequestMessage Read(string path, int? wait = null, string? label = null, string? page = null)
+ {
+ var request = new HttpRequestMessage(HttpMethod.Get, path);
+ if (wait != null) request.Headers.TryAddWithoutValidation("Prefer", $"respond-async, wait={wait}");
+ if (label != null) request.Headers.Add("X-Bitween-Label", Uri.EscapeDataString(label));
+ if (page != null) request.Headers.Add("X-Bitween-Page", Uri.EscapeDataString(page));
+ return request;
+ }
+
+ /// The request, from the list, once it has stopped running.
+ static async Task FinishedAsync(HttpClient client, string id)
+ {
+ for (var i = 0; i < 300; i++)
+ {
+ var row = (await Api.Json(await client.GetAsync("/api/background"))).AsArray()
+ .FirstOrDefault(r => (string)r!["id"]! == id);
+ if (row != null && (string)row["status"]! is not ("Waiting" or "Running")) return row;
+ await Task.Delay(100);
+ }
+
+ throw new TimeoutException($"Background request {id} didn't finish.");
+ }
+
+ Task RowsAsync() =>
+ fixture.InDbAsync(db => db.Set().AsNoTracking().ToArrayAsync());
+
+ /// Waits until the database is running of the slow reads' sleeps.
+ async Task SleepingAsync(int count)
+ {
+ for (var i = 0; i < 100 && await SleepsAsync() != count; i++) await Task.Delay(100);
+ Assert.Equal(count, await SleepsAsync());
+ }
+
+ Task SleepsAsync() => fixture.InDbAsync(db => db.Database
+ .SqlQueryRaw("SELECT count(*)::int AS \"Value\" FROM pg_stat_activity WHERE state = 'active' AND query LIKE 'SELECT pg_sleep(%'")
+ .SingleAsync());
+
+ /// A sign-in for an account that isn't the administrator, signed as this install signs them.
+ string TokenFor(string accountId)
+ {
+ var configuration = node.Services.GetRequiredService();
+ return new JsonWebTokenHandler().CreateToken(new SecurityTokenDescriptor
+ {
+ Subject = new ClaimsIdentity([new Claim(ClaimTypes.NameIdentifier, accountId)]),
+ Issuer = configuration["Token:Issuer"],
+ Audience = configuration["Token:Audience"],
+ Expires = DateTime.UtcNow.AddMinutes(5),
+ SigningCredentials = new SigningCredentials(
+ new SymmetricSecurityKey(Encoding.UTF8.GetBytes(configuration["Token:Key"]!)), SecurityAlgorithms.HmacSha256),
+ });
+ }
+}
+
+/// A read as slow as asked: it sleeps in the database, as one on a big table waits on it.
+[ApiController]
+[Route("api/slowreads")]
+public class SlowReadsController(BitweenDbContext db) : ControllerBase
+{
+ [HttpGet("{seconds:double}")]
+ public async Task Get(double seconds)
+ {
+ await db.Database.ExecuteSqlRawAsync("SELECT pg_sleep({0})", seconds);
+ return Ok(new { seconds });
+ }
+
+ ///
+ /// Work that counts its steps, as a bulk retry does: a fifth of a second each, saying how far it got,
+ /// and stopping between steps once its owner cancels it.
+ ///
+ [HttpPost("steps")]
+ public async Task Steps([FromBody] StepsRequest request)
+ {
+ var background = Services.Background.BackgroundRequestScope.Current;
+ var done = 0;
+ for (; done < request.Count; done++)
+ {
+ background?.Progress(done, request.Count, $"{done} of {request.Count}");
+ if (background?.CancelRequested.IsCancellationRequested == true) break;
+ await db.Database.ExecuteSqlRawAsync("SELECT pg_sleep(0.2)");
+ }
+
+ return Ok(new { done });
+ }
+
+ public class StepsRequest
+ {
+ public int Count { get; set; }
+ }
+}
diff --git a/SW.Bitween.HttpTests/HttpFixture.cs b/SW.Bitween.HttpTests/HttpFixture.cs
index 3cd04ec4..6ff20ba1 100644
--- a/SW.Bitween.HttpTests/HttpFixture.cs
+++ b/SW.Bitween.HttpTests/HttpFixture.cs
@@ -48,9 +48,13 @@ public sealed class HttpFixture : IAsyncLifetime
/// is. The caller disposes it.
///
/// Changes to the node's services, as a test needs them.
- public WebApplicationFactory SecondNode(Action? services = null)
+ /// Changes to the node's settings, made to a copy of the fixture's.
+ public WebApplicationFactory SecondNode(Action? services = null,
+ Action>? own = null)
{
- var node = new BitweenApp(settings, services);
+ var nodeSettings = new Dictionary(settings);
+ own?.Invoke(nodeSettings);
+ var node = new BitweenApp(nodeSettings, services);
using var _ = node.CreateClient();
return node;
}
diff --git a/SW.Bitween.HttpTests/SW.Bitween.HttpTests.csproj b/SW.Bitween.HttpTests/SW.Bitween.HttpTests.csproj
index 25a6a55a..085b81e5 100644
--- a/SW.Bitween.HttpTests/SW.Bitween.HttpTests.csproj
+++ b/SW.Bitween.HttpTests/SW.Bitween.HttpTests.csproj
@@ -8,6 +8,7 @@
+
diff --git a/SW.Bitween.IntegrationTests/Tests/BulkRetryInBackgroundTests.cs b/SW.Bitween.IntegrationTests/Tests/BulkRetryInBackgroundTests.cs
new file mode 100644
index 00000000..23f63f9c
--- /dev/null
+++ b/SW.Bitween.IntegrationTests/Tests/BulkRetryInBackgroundTests.cs
@@ -0,0 +1,184 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+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.Bitween.Services.Background;
+using SW.PrimitiveTypes;
+using Xunit;
+
+namespace SW.Bitween.IntegrationTests.Tests;
+
+///
+/// A bulk retry the admin UI carries on in the background: the whole selection at once rather than in
+/// rounds of 500, saved as it goes, saying how far it got, and stopping between exchanges when cancelled.
+///
+[Collection("Bitween")]
+public class BulkRetryInBackgroundTests(BitweenFixture fixture)
+{
+ /// One more than a retry asked directly takes.
+ const int PastOneRound = 501;
+
+ [Fact]
+ public async Task Asked_directly_it_still_takes_one_round_at_most()
+ {
+ await using var scope = fixture.CreateScope();
+ var (db, xs, ctx) = Services(scope);
+ var ids = await FailedXchangesAsync(db, xs, PastOneRound);
+
+ var refused = await Assert.ThrowsAsync(() =>
+ new Resources.Xchanges.BulkRetry(db, ctx, xs).Handle(new XchangeBulkRetry { Ids = ids }));
+
+ Assert.Contains(refused.Validations, v => v.Key == "TOO_MANY");
+ }
+
+ [Fact]
+ public async Task In_the_background_it_takes_the_whole_selection_and_says_how_far_it_got()
+ {
+ await using var scope = fixture.CreateScope();
+ var (db, xs, ctx) = Services(scope);
+ var ids = await FailedXchangesAsync(db, xs, PastOneRound);
+ var said = new List<(int Done, int Total, string Summary)>();
+
+ using var stop = new CancellationTokenSource();
+ BackgroundRequestScope.Open(DateTime.UtcNow + BackgroundRequests.Limit, stop.Token, default,
+ (done, total, summary) => said.Add((done, total, summary)));
+ var plan = (XchangeBulkRetryPlan)await new Resources.Xchanges.BulkRetry(db, ctx, xs)
+ .Handle(new XchangeBulkRetry { Ids = ids });
+
+ Assert.Equal(PastOneRound, plan.WillRetry);
+ Assert.Null(plan.StoppedAfter);
+ Assert.Equal(PastOneRound, await db.Set().CountAsync(x => ids.Contains(x.RetryFor)));
+ // Once at the start, after each save of a hundred, and at the end.
+ Assert.Equal([0, 100, 200, 300, 400, 500, PastOneRound], said.Select(s => s.Done));
+ Assert.All(said, s => Assert.Equal(PastOneRound, s.Total));
+ Assert.Equal($"{PastOneRound} retried, 0 skipped", said[^1].Summary);
+ }
+
+ [Fact]
+ public async Task Cancelled_it_stops_between_exchanges_and_keeps_what_it_did()
+ {
+ await using var scope = fixture.CreateScope();
+ var (db, xs, ctx) = Services(scope);
+ var ids = await FailedXchangesAsync(db, xs, 250);
+
+ using var stop = new CancellationTokenSource();
+ using var cancel = new CancellationTokenSource();
+ // Cancelled by its owner once the first hundred are saved, as the header's Cancel would.
+ BackgroundRequestScope.Open(DateTime.UtcNow + BackgroundRequests.Limit, stop.Token, cancel.Token,
+ (done, _, _) => { if (done >= 100) cancel.Cancel(); });
+ var plan = (XchangeBulkRetryPlan)await new Resources.Xchanges.BulkRetry(db, ctx, xs)
+ .Handle(new XchangeBulkRetry { Ids = ids });
+
+ Assert.Equal(100, plan.StoppedAfter);
+ Assert.Equal(100, plan.WillRetry);
+ Assert.Equal(100, await db.Set().CountAsync(x => ids.Contains(x.RetryFor)));
+ }
+
+ [Fact]
+ public async Task It_goes_at_the_pipelines_pace_holding_back_while_too_many_wait()
+ {
+ var (waiting, lookEvery) = (Resources.Xchanges.BulkRetry.Waiting, Resources.Xchanges.BulkRetry.LookEvery);
+ Resources.Xchanges.BulkRetry.Waiting = 150;
+ Resources.Xchanges.BulkRetry.LookEvery = TimeSpan.FromMilliseconds(50);
+ try
+ {
+ await using var scope = fixture.CreateScope();
+ var (db, xs, ctx) = Services(scope);
+ var ids = await FailedXchangesAsync(db, xs, 250);
+ var said = new List();
+
+ using var stop = new CancellationTokenSource();
+ var retrying = Task.Run(async () =>
+ {
+ BackgroundRequestScope.Open(DateTime.UtcNow + BackgroundRequests.Limit, stop.Token, default,
+ (_, _, summary) => { lock (said) said.Add(summary); });
+ return (XchangeBulkRetryPlan)await new Resources.Xchanges.BulkRetry(db, ctx, xs)
+ .Handle(new XchangeBulkRetry { Ids = ids });
+ });
+
+ // A second save of a hundred would make 200 waiting, past the 150 allowed: it holds at 100.
+ await using var check = fixture.CreateScope();
+ var checkDb = check.ServiceProvider.GetRequiredService();
+ for (var i = 0; i < 100 && await Retries(checkDb, ids) < 100; i++) await Task.Delay(50);
+ await Task.Delay(300);
+ Assert.Equal(100, await Retries(checkDb, ids));
+ Assert.False(retrying.IsCompleted);
+
+ // The pipeline finishes what it sends, and it carries on to the end.
+ var processed = new HashSet();
+ for (var i = 0; i < 300 && !retrying.IsCompleted; i++)
+ {
+ var made = await checkDb.Set().Where(x => ids.Contains(x.RetryFor)).Select(x => x.Id).ToListAsync();
+ checkDb.AddRange(made.Where(processed.Add).Select(id => new XchangeResult(id, null, null, exception: "processed")));
+ await checkDb.SaveChangesAsync();
+ await Task.Delay(100);
+ }
+
+ var plan = await retrying.WaitAsync(TimeSpan.FromSeconds(30));
+ Assert.Equal(250, plan.WillRetry);
+ Assert.Equal(250, await Retries(checkDb, ids));
+ Assert.Contains(said, s => s.Contains("waiting for the pipeline"));
+ }
+ finally
+ {
+ (Resources.Xchanges.BulkRetry.Waiting, Resources.Xchanges.BulkRetry.LookEvery) = (waiting, lookEvery);
+ }
+ }
+
+ [Fact]
+ public async Task A_plan_counts_every_skip_and_names_the_first_thousand()
+ {
+ await using var scope = fixture.CreateScope();
+ var (db, xs, ctx) = Services(scope);
+ var ids = await FailedXchangesAsync(db, xs, 1_005);
+ // Succeeded since, so all of them are skipped.
+ await db.Set().Where(r => ids.Contains(r.Id))
+ .ExecuteUpdateAsync(u => u.SetProperty(r => r.Success, true));
+
+ var plan = (XchangeBulkRetryPlan)await new Resources.Xchanges.BulkRetryPreview(db, ctx)
+ .Handle(new XchangeBulkRetry { Ids = ids });
+
+ Assert.Equal(1_005, plan.SkippedCount);
+ Assert.Equal(1_000, plan.Skipped.Count);
+ Assert.Equal(1_000, plan.Listed);
+ Assert.Equal(0, plan.WillRetry);
+ }
+
+ static Task Retries(BitweenDbContext db, List ids) =>
+ db.Set().CountAsync(x => ids.Contains(x.RetryFor));
+
+ static (BitweenDbContext Db, XchangeService Xs, RequestContext Ctx) Services(AsyncServiceScope scope) =>
+ (scope.ServiceProvider.GetRequiredService(),
+ scope.ServiceProvider.GetRequiredService(),
+ scope.Superuser());
+
+ /// Exchanges that failed, on a subscription of their own, by id.
+ static async Task> FailedXchangesAsync(BitweenDbContext db, XchangeService xs, int count)
+ {
+ var name = $"Bulk background {Guid.NewGuid():N}"[..30];
+ 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 ids = new List();
+ for (var i = 0; i < count; i++)
+ {
+ var xchange = await xs.CreateXchange(sub, new XchangeFile($"{{\"n\":{i}}}"));
+ db.Set().Add(new XchangeResult(xchange.Id, null, null, exception: "boom"));
+ ids.Add(xchange.Id);
+ }
+
+ await db.SaveChangesAsync();
+ db.ChangeTracker.Clear();
+ return ids;
+ }
+}
diff --git a/SW.Bitween.MsSql/Migrations/20261101090252_AddBackgroundRequests.Designer.cs b/SW.Bitween.MsSql/Migrations/20261101090252_AddBackgroundRequests.Designer.cs
new file mode 100644
index 00000000..984d66fa
--- /dev/null
+++ b/SW.Bitween.MsSql/Migrations/20261101090252_AddBackgroundRequests.Designer.cs
@@ -0,0 +1,3573 @@
+//
+using System;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore.Infrastructure;
+using Microsoft.EntityFrameworkCore.Metadata;
+using Microsoft.EntityFrameworkCore.Migrations;
+using Microsoft.EntityFrameworkCore.Storage.ValueConversion;
+using SW.Bitween.MsSql;
+
+#nullable disable
+
+namespace SW.Bitween.MsSql.Migrations
+{
+ [DbContext(typeof(BitweenDbContext))]
+ [Migration("20261101090252_AddBackgroundRequests")]
+ partial class AddBackgroundRequests
+ {
+ ///
+ protected override void BuildTargetModel(ModelBuilder modelBuilder)
+ {
+#pragma warning disable 612, 618
+ modelBuilder
+ .HasAnnotation("ProductVersion", "9.0.20")
+ .HasAnnotation("Relational:MaxIdentifierLength", 128);
+
+ SqlServerModelBuilderExtensions.UseIdentityColumns(modelBuilder);
+
+ modelBuilder.HasSequence("DocumentIds");
+
+ modelBuilder.Entity("SW.Bitween.Domain.Accounts.Account", b =>
+ {
+ b.Property("Id")
+ .ValueGeneratedOnAdd()
+ .HasColumnType("int");
+
+ SqlServerPropertyBuilderExtensions.UseIdentityColumn(b.Property("Id"));
+
+ b.Property("CreatedBy")
+ .HasColumnType("nvarchar(max)");
+
+ b.Property("CreatedOn")
+ .HasColumnType("datetime2");
+
+ b.Property("Deleted")
+ .HasColumnType("bit");
+
+ b.Property("Disabled")
+ .HasColumnType("bit");
+
+ b.Property("DisplayName")
+ .IsRequired()
+ .HasMaxLength(200)
+ .HasColumnType("nvarchar(200)");
+
+ b.Property("Email")
+ .HasMaxLength(200)
+ .IsUnicode(false)
+ .HasColumnType("varchar(200)");
+
+ b.Property("EmailProvider")
+ .HasColumnType("tinyint");
+
+ b.Property("FailedLoginCount")
+ .HasColumnType("int");
+
+ b.Property("Kind")
+ .HasColumnType("tinyint");
+
+ b.Property("LastSignInOn")
+ .HasColumnType("datetime2");
+
+ b.Property("LockoutEnd")
+ .HasColumnType("datetime2");
+
+ b.Property("LoginMethods")
+ .HasColumnType("tinyint");
+
+ b.Property("MicrosoftIdentity")
+ .HasColumnType("nvarchar(max)");
+
+ b.Property("ModifiedBy")
+ .HasColumnType("nvarchar(max)");
+
+ b.Property("ModifiedOn")
+ .HasColumnType("datetime2");
+
+ b.Property("MustChangePassword")
+ .HasColumnType("bit");
+
+ b.Property("OidcIdentity")
+ .HasColumnType("nvarchar(max)");
+
+ b.Property("Password")
+ .HasMaxLength(500)
+ .IsUnicode(false)
+ .HasColumnType("varchar(500)");
+
+ b.Property("Role")
+ .HasColumnType("int");
+
+ b.HasKey("Id");
+
+ b.HasIndex("Email")
+ .IsUnique()
+ .HasFilter("[Email] IS NOT NULL");
+
+ b.ToTable("Accounts", (string)null);
+
+ b.HasData(
+ new
+ {
+ Id = 9999,
+ CreatedOn = new DateTime(2022, 1, 1, 0, 0, 0, 0, DateTimeKind.Utc),
+ Deleted = false,
+ Disabled = false,
+ DisplayName = "Admin",
+ Email = "admin@Bitween.systems",
+ EmailProvider = (byte)0,
+ FailedLoginCount = 0,
+ Kind = (byte)0,
+ LoginMethods = (byte)2,
+ MustChangePassword = true,
+ Password = "$SWHASH$V1$10000$VQCi48eitH4Ml5juvBMOFZrMdQwBbhuIQVXe6RR7qJdDF2bJ",
+ Role = 0
+ });
+ });
+
+ modelBuilder.Entity("SW.Bitween.Domain.Accounts.AccountLink", b =>
+ {
+ b.Property("Id")
+ .ValueGeneratedOnAdd()
+ .HasColumnType("int");
+
+ SqlServerPropertyBuilderExtensions.UseIdentityColumn(b.Property("Id"));
+
+ b.Property("AccountId")
+ .HasColumnType("int");
+
+ b.Property("CreatedOn")
+ .HasColumnType("datetime2");
+
+ b.Property("EmailError")
+ .HasMaxLength(500)
+ .HasColumnType("nvarchar(500)");
+
+ b.Property("EmailedOn")
+ .HasColumnType("datetime2");
+
+ b.Property("ExpiresOn")
+ .HasColumnType("datetime2");
+
+ b.Property("Hash")
+ .IsRequired()
+ .HasMaxLength(64)
+ .IsUnicode(false)
+ .HasColumnType("varchar(64)");
+
+ b.Property("Purpose")
+ .HasColumnType("tinyint");
+
+ b.HasKey("Id");
+
+ b.HasIndex("AccountId");
+
+ b.HasIndex("Hash")
+ .IsUnique();
+
+ b.ToTable("AccountLinks", (string)null);
+ });
+
+ modelBuilder.Entity("SW.Bitween.Domain.Accounts.AccountRoleLink", b =>
+ {
+ b.Property("AccountId")
+ .HasColumnType("int");
+
+ b.Property("RoleId")
+ .HasColumnType("int");
+
+ b.HasKey("AccountId", "RoleId");
+
+ b.HasIndex("RoleId");
+
+ b.ToTable("AccountRoles", (string)null);
+ });
+
+ modelBuilder.Entity("SW.Bitween.Domain.Accounts.RefreshToken", b =>
+ {
+ b.Property("Id")
+ .HasMaxLength(50)
+ .IsUnicode(false)
+ .HasColumnType("varchar(50)");
+
+ b.Property("AccountId")
+ .HasColumnType("int");
+
+ b.Property("CreatedOn")
+ .HasColumnType("datetime2");
+
+ b.Property("LoginMethod")
+ .HasColumnType("tinyint");
+
+ b.Property("SupersededOn")
+ .HasColumnType("datetime2");
+
+ b.HasKey("Id");
+
+ b.HasIndex("AccountId");
+
+ b.ToTable("RefreshTokens", (string)null);
+ });
+
+ modelBuilder.Entity("SW.Bitween.Domain.Accounts.Role", b =>
+ {
+ b.Property("Id")
+ .ValueGeneratedOnAdd()
+ .HasColumnType("int");
+
+ SqlServerPropertyBuilderExtensions.UseIdentityColumn(b.Property("Id"));
+
+ b.Property("CreatedBy")
+ .HasColumnType("nvarchar(max)");
+
+ b.Property("CreatedOn")
+ .HasColumnType("datetime2");
+
+ b.Property("Description")
+ .HasMaxLength(500)
+ .HasColumnType("nvarchar(500)");
+
+ b.Property("IsSystem")
+ .HasColumnType("bit");
+
+ b.Property("ModifiedBy")
+ .HasColumnType("nvarchar(max)");
+
+ b.Property("ModifiedOn")
+ .HasColumnType("datetime2");
+
+ b.Property("Name")
+ .IsRequired()
+ .HasMaxLength(100)
+ .HasColumnType("nvarchar(100)");
+
+ b.Property("Permissions")
+ .HasColumnType("nvarchar(max)");
+
+ b.HasKey("Id");
+
+ b.HasIndex("Name")
+ .IsUnique();
+
+ b.ToTable("Roles", (string)null);
+
+ b.HasData(
+ new
+ {
+ Id = 1,
+ CreatedOn = new DateTime(2022, 1, 1, 0, 0, 0, 0, DateTimeKind.Utc),
+ Description = "Full access to everything, including members, roles and settings.",
+ IsSystem = true,
+ Name = "Administrator",
+ Permissions = "[]"
+ },
+ new
+ {
+ Id = 2,
+ CreatedOn = new DateTime(2022, 1, 1, 0, 0, 0, 0, DateTimeKind.Utc),
+ Description = "Runs and configures integrations. Can't manage members, roles or settings.",
+ IsSystem = true,
+ Name = "Member",
+ Permissions = "[]"
+ },
+ new
+ {
+ Id = 3,
+ CreatedOn = new DateTime(2022, 1, 1, 0, 0, 0, 0, DateTimeKind.Utc),
+ Description = "Read-only access to integrations, exchanges and configuration.",
+ IsSystem = true,
+ Name = "Viewer",
+ Permissions = "[]"
+ });
+ });
+
+ modelBuilder.Entity("SW.Bitween.Domain.Accounts.ServiceToken", b =>
+ {
+ b.Property("Id")
+ .ValueGeneratedOnAdd()
+ .HasColumnType("int");
+
+ SqlServerPropertyBuilderExtensions.UseIdentityColumn(b.Property("Id"));
+
+ b.Property("AccountId")
+ .HasColumnType("int");
+
+ b.Property("CreatedBy")
+ .HasColumnType("nvarchar(max)");
+
+ b.Property("CreatedOn")
+ .HasColumnType("datetime2");
+
+ b.Property("ExpiresOn")
+ .HasColumnType("datetime2");
+
+ b.Property("Hash")
+ .IsRequired()
+ .HasMaxLength(64)
+ .IsUnicode(false)
+ .HasColumnType("varchar(64)");
+
+ b.Property("LastUsedFrom")
+ .HasMaxLength(64)
+ .IsUnicode(false)
+ .HasColumnType("varchar(64)");
+
+ b.Property("LastUsedOn")
+ .HasColumnType("datetime2");
+
+ b.Property("ModifiedBy")
+ .HasColumnType("nvarchar(max)");
+
+ b.Property("ModifiedOn")
+ .HasColumnType("datetime2");
+
+ b.Property("Name")
+ .IsRequired()
+ .HasMaxLength(100)
+ .HasColumnType("nvarchar(100)");
+
+ b.Property("RevokedBy")
+ .HasMaxLength(200)
+ .HasColumnType("nvarchar(200)");
+
+ b.Property("RevokedOn")
+ .HasColumnType("datetime2");
+
+ b.Property("Shown")
+ .IsRequired()
+ .HasMaxLength(20)
+ .IsUnicode(false)
+ .HasColumnType("varchar(20)");
+
+ b.HasKey("Id");
+
+ b.HasIndex("AccountId");
+
+ b.HasIndex("Hash")
+ .IsUnique();
+
+ b.ToTable("ServiceTokens", (string)null);
+ });
+
+ modelBuilder.Entity("SW.Bitween.Domain.AdapterDraft", b =>
+ {
+ b.Property("Id")
+ .ValueGeneratedOnAdd()
+ .HasColumnType("int");
+
+ SqlServerPropertyBuilderExtensions.UseIdentityColumn(b.Property("Id"));
+
+ b.Property("AdapterId")
+ .IsRequired()
+ .HasMaxLength(200)
+ .IsUnicode(false)
+ .HasColumnType("varchar(200)");
+
+ b.Property("BaseVersion")
+ .HasMaxLength(50)
+ .IsUnicode(false)
+ .HasColumnType("varchar(50)");
+
+ b.Property("CreatedBy")
+ .HasMaxLength(200)
+ .HasColumnType("nvarchar(200)");
+
+ b.Property("CreatedOn")
+ .HasColumnType("datetime2");
+
+ b.Property("FilesHash")
+ .IsRequired()
+ .HasMaxLength(64)
+ .IsUnicode(false)
+ .HasColumnType("varchar(64)");
+
+ b.Property("FilesJson")
+ .IsRequired()
+ .HasColumnType("nvarchar(max)");
+
+ b.Property("Kind")
+ .HasMaxLength(20)
+ .IsUnicode(false)
+ .HasColumnType("varchar(20)");
+
+ b.Property("Language")
+ .IsRequired()
+ .HasMaxLength(20)
+ .IsUnicode(false)
+ .HasColumnType("varchar(20)");
+
+ b.Property("ModifiedBy")
+ .HasMaxLength(200)
+ .HasColumnType("nvarchar(200)");
+
+ b.Property("ModifiedOn")
+ .HasColumnType("datetime2");
+
+ b.HasKey("Id");
+
+ b.HasIndex("AdapterId");
+
+ b.ToTable("AdapterDrafts", (string)null);
+ });
+
+ modelBuilder.Entity("SW.Bitween.Domain.AdapterRelease", b =>
+ {
+ b.Property("Id")
+ .ValueGeneratedOnAdd()
+ .HasColumnType("bigint");
+
+ SqlServerPropertyBuilderExtensions.UseIdentityColumn(b.Property("Id"));
+
+ b.Property("AccountId")
+ .HasMaxLength(100)
+ .IsUnicode(false)
+ .HasColumnType("varchar(100)");
+
+ b.Property("Action")
+ .IsRequired()
+ .HasMaxLength(20)
+ .IsUnicode(false)
+ .HasColumnType("varchar(20)");
+
+ b.Property("AdapterId")
+ .IsRequired()
+ .HasMaxLength(200)
+ .IsUnicode(false)
+ .HasColumnType("varchar(200)");
+
+ b.Property("DraftId")
+ .HasColumnType("int");
+
+ b.Property("OccurredOn")
+ .HasColumnType("datetime2");
+
+ b.Property("Version")
+ .IsRequired()
+ .HasMaxLength(50)
+ .IsUnicode(false)
+ .HasColumnType("varchar(50)");
+
+ b.HasKey("Id");
+
+ b.HasIndex("AdapterId", "OccurredOn");
+
+ b.ToTable("AdapterReleases", (string)null);
+ });
+
+ modelBuilder.Entity("SW.Bitween.Domain.AdapterSourceAccess", b =>
+ {
+ b.Property("Id")
+ .ValueGeneratedOnAdd()
+ .HasColumnType("bigint");
+
+ SqlServerPropertyBuilderExtensions.UseIdentityColumn(b.Property("Id"));
+
+ b.Property("AccountId")
+ .HasMaxLength(100)
+ .IsUnicode(false)
+ .HasColumnType("varchar(100)");
+
+ b.Property("AdapterId")
+ .IsRequired()
+ .HasMaxLength(200)
+ .IsUnicode(false)
+ .HasColumnType("varchar(200)");
+
+ b.Property("OccurredOn")
+ .HasColumnType("datetime2");
+
+ b.Property("Path")
+ .HasMaxLength(500)
+ .HasColumnType("nvarchar(500)");
+
+ b.Property("Version")
+ .HasMaxLength(50)
+ .IsUnicode(false)
+ .HasColumnType("varchar(50)");
+
+ b.HasKey("Id");
+
+ b.HasIndex("AdapterId", "OccurredOn");
+
+ b.ToTable("AdapterSourceAccesses", (string)null);
+ });
+
+ modelBuilder.Entity("SW.Bitween.Domain.ApiKeyUse", b =>
+ {
+ b.Property("PartnerId")
+ .HasColumnType("int");
+
+ b.Property("KeyName")
+ .HasMaxLength(500)
+ .HasColumnType("nvarchar(500)");
+
+ b.Property("LastUsedOn")
+ .HasColumnType("datetime2");
+
+ b.HasKey("PartnerId", "KeyName");
+
+ b.ToTable("ApiKeyUses", (string)null);
+ });
+
+ modelBuilder.Entity("SW.Bitween.Domain.AuditEntry", b =>
+ {
+ b.Property("Id")
+ .HasMaxLength(32)
+ .IsUnicode(false)
+ .HasColumnType("varchar(32)");
+
+ b.Property("Changes")
+ .HasColumnType("nvarchar(max)");
+
+ b.Property("CorrelationId")
+ .IsRequired()
+ .HasMaxLength(36)
+ .IsUnicode(false)
+ .HasColumnType("varchar(36)");
+
+ b.Property("EntityKey")
+ .HasMaxLength(200)
+ .HasColumnType("nvarchar(200)");
+
+ b.Property("EntityName")
+ .IsRequired()
+ .HasMaxLength(200)
+ .HasColumnType("nvarchar(200)");
+
+ b.Property("OccurredOn")
+ .HasColumnType("datetime2");
+
+ b.Property("Sequence")
+ .HasColumnType("int");
+
+ b.Property("State")
+ .IsRequired()
+ .HasMaxLength(10)
+ .IsUnicode(false)
+ .HasColumnType("varchar(10)");
+
+ b.Property("UserId")
+ .HasMaxLength(50)
+ .IsUnicode(false)
+ .HasColumnType("varchar(50)");
+
+ b.HasKey("Id");
+
+ b.HasIndex("CorrelationId");
+
+ b.HasIndex("OccurredOn");
+
+ b.HasIndex("EntityName", "EntityKey", "OccurredOn");
+
+ b.ToTable("AuditEntries", (string)null);
+ });
+
+ modelBuilder.Entity("SW.Bitween.Domain.BackgroundRequest", b =>
+ {
+ b.Property("Id")
+ .HasMaxLength(32)
+ .IsUnicode(false)
+ .HasColumnType("varchar(32)");
+
+ b.Property("Done")
+ .HasColumnType("int");
+
+ b.Property("FinishedOn")
+ .HasColumnType("datetime2");
+
+ b.Property("Label")
+ .HasMaxLength(200)
+ .HasColumnType("nvarchar(200)");
+
+ b.Property("Node")
+ .HasMaxLength(200)
+ .IsUnicode(false)
+ .HasColumnType("varchar(200)");
+
+ b.Property("Owner")
+ .IsRequired()
+ .HasMaxLength(50)
+ .HasColumnType("nvarchar(50)");
+
+ b.Property("Page")
+ .HasMaxLength(2000)
+ .HasColumnType("nvarchar(2000)");
+
+ b.Property("Path")
+ .IsRequired()
+ .HasMaxLength(4000)
+ .HasColumnType("nvarchar(4000)");
+
+ b.Property("Result")
+ .HasColumnType("varbinary(max)");
+
+ b.Property("ResultContentType")
+ .HasMaxLength(200)
+ .IsUnicode(false)
+ .HasColumnType("varchar(200)");
+
+ b.Property("ResultDisposition")
+ .HasMaxLength(500)
+ .HasColumnType("nvarchar(500)");
+
+ b.Property("ResultKey")
+ .HasMaxLength(500)
+ .IsUnicode(false)
+ .HasColumnType("varchar(500)");
+
+ b.Property("ResultStatus")
+ .HasColumnType("int");
+
+ b.Property("StartedOn")
+ .HasColumnType("datetime2");
+
+ b.Property("Status")
+ .HasColumnType("int");
+
+ b.Property("Summary")
+ .HasMaxLength(500)
+ .HasColumnType("nvarchar(500)");
+
+ b.Property("Total")
+ .HasColumnType("int");
+
+ b.HasKey("Id");
+
+ b.HasIndex("StartedOn");
+
+ b.HasIndex("Owner", "StartedOn");
+
+ b.ToTable("BackgroundRequests", (string)null);
+ });
+
+ modelBuilder.Entity("SW.Bitween.Domain.Cluster.ClusterLease", b =>
+ {
+ b.Property("Id")
+ .HasMaxLength(200)
+ .IsUnicode(false)
+ .HasColumnType("varchar(200)");
+
+ b.Property("AcquiredOn")
+ .HasColumnType("datetime2");
+
+ b.Property("OwnerNode")
+ .HasMaxLength(200)
+ .IsUnicode(false)
+ .HasColumnType("varchar(200)");
+
+ b.Property("Term")
+ .HasColumnType("bigint");
+
+ b.HasKey("Id");
+
+ b.ToTable("ClusterLeases", (string)null);
+ });
+
+ modelBuilder.Entity("SW.Bitween.Domain.Cluster.ClusterNode", b =>
+ {
+ b.Property("Id")
+ .HasMaxLength(200)
+ .IsUnicode(false)
+ .HasColumnType("varchar(200)");
+
+ b.Property("DataSources")
+ .HasColumnType("bit");
+
+ b.Property("Host")
+ .HasMaxLength(200)
+ .IsUnicode(false)
+ .HasColumnType("varchar(200)");
+
+ b.Property("LastSeenOn")
+ .HasColumnType("datetime2");
+
+ b.Property("OutboxPublishedLastHour")
+ .HasColumnType("int");
+
+ b.Property("ResidentHealth")
+ .HasMaxLength(4000)
+ .HasColumnType("nvarchar(4000)");
+
+ b.Property("Runtimes")
+ .HasMaxLength(100)
+ .IsUnicode(false)
+ .HasColumnType("varchar(100)");
+
+ b.Property("StartedOn")
+ .HasColumnType("datetime2");
+
+ b.Property("Version")
+ .HasMaxLength(50)
+ .IsUnicode(false)
+ .HasColumnType("varchar(50)");
+
+ b.HasKey("Id");
+
+ b.HasIndex("LastSeenOn");
+
+ b.ToTable("ClusterNodes", (string)null);
+ });
+
+ modelBuilder.Entity("SW.Bitween.Domain.Connections.Connection", b =>
+ {
+ b.Property("Id")
+ .ValueGeneratedOnAdd()
+ .HasColumnType("int");
+
+ SqlServerPropertyBuilderExtensions.UseIdentityColumn(b.Property("Id"));
+
+ b.Property("AccessToken")
+ .HasColumnType("nvarchar(max)");
+
+ b.Property("AccessTokenExpiresOn")
+ .HasColumnType("datetime2");
+
+ b.Property("AuthorizeParameters")
+ .HasMaxLength(2000)
+ .HasColumnType("nvarchar(2000)");
+
+ b.Property("AuthorizeUrl")
+ .HasMaxLength(1000)
+ .HasColumnType("nvarchar(1000)");
+
+ b.Property("AuthorizedBy")
+ .HasMaxLength(200)
+ .HasColumnType("nvarchar(200)");
+
+ b.Property("AuthorizedOn")
+ .HasColumnType("datetime2");
+
+ b.Property("ClientId")
+ .HasMaxLength(500)
+ .HasColumnType("nvarchar(500)");
+
+ b.Property("ClientSecret")
+ .HasColumnType("nvarchar(max)");
+
+ b.Property("CreatedBy")
+ .HasColumnType("nvarchar(max)");
+
+ b.Property("CreatedOn")
+ .HasColumnType("datetime2");
+
+ b.Property("Grant")
+ .HasColumnType("int");
+
+ b.Property("LastError")
+ .HasMaxLength(2000)
+ .HasColumnType("nvarchar(2000)");
+
+ b.Property("ModifiedBy")
+ .HasColumnType("nvarchar(max)");
+
+ b.Property("ModifiedOn")
+ .HasColumnType("datetime2");
+
+ b.Property("Name")
+ .IsRequired()
+ .HasMaxLength(100)
+ .IsUnicode(false)
+ .HasColumnType("varchar(100)");
+
+ b.Property("Password")
+ .HasColumnType("nvarchar(max)");
+
+ b.Property("RefreshToken")
+ .HasColumnType("nvarchar(max)");
+
+ b.Property("RenewingUntil")
+ .HasColumnType("datetime2");
+
+ b.Property("Scopes")
+ .HasMaxLength(2000)
+ .HasColumnType("nvarchar(2000)");
+
+ b.Property("Status")
+ .HasColumnType("int");
+
+ b.Property("TokenAuth")
+ .HasMaxLength(20)
+ .IsUnicode(false)
+ .HasColumnType("varchar(20)");
+
+ b.Property("TokenUrl")
+ .HasMaxLength(1000)
+ .HasColumnType("nvarchar(1000)");
+
+ b.Property("Username")
+ .HasMaxLength(500)
+ .HasColumnType("nvarchar(500)");
+
+ b.HasKey("Id");
+
+ b.HasIndex("Name")
+ .IsUnique();
+
+ b.ToTable("Connections", (string)null);
+ });
+
+ modelBuilder.Entity("SW.Bitween.Domain.DataSources.AdapterState", b =>
+ {
+ b.Property("AdapterId")
+ .HasMaxLength(200)
+ .IsUnicode(false)
+ .HasColumnType("varchar(200)");
+
+ b.Property("InstanceKey")
+ .HasMaxLength(200)
+ .IsUnicode(false)
+ .HasColumnType("varchar(200)");
+
+ b.Property("Name")
+ .HasMaxLength(200)
+ .IsUnicode(false)
+ .HasColumnType("varchar(200)");
+
+ b.Property("UpdatedOn")
+ .HasColumnType("datetime2");
+
+ b.Property("Value")
+ .HasMaxLength(8000)
+ .HasColumnType("nvarchar(max)");
+
+ b.HasKey("AdapterId", "InstanceKey", "Name");
+
+ b.ToTable("AdapterStates", (string)null);
+ });
+
+ modelBuilder.Entity("SW.Bitween.Domain.DataSources.DataSource", b =>
+ {
+ b.Property("Id")
+ .ValueGeneratedOnAdd()
+ .HasColumnType("int");
+
+ SqlServerPropertyBuilderExtensions.UseIdentityColumn(b.Property("Id"));
+
+ b.Property("AdapterId")
+ .IsRequired()
+ .HasMaxLength(200)
+ .IsUnicode(false)
+ .HasColumnType("varchar(200)");
+
+ b.Property("ConsecutiveFailures")
+ .HasColumnType("int");
+
+ b.Property("CpuLimitSamples")
+ .HasColumnType("int");
+
+ b.Property("CpuPercentLimit")
+ .HasColumnType("float");
+
+ b.Property("CreatedBy")
+ .HasColumnType("nvarchar(max)");
+
+ b.Property("CreatedOn")
+ .HasColumnType("datetime2");
+
+ b.Property("DeduplicationWindowDays")
+ .HasColumnType("int");
+
+ b.Property("HardMemoryLimitMb")
+ .HasColumnType("int");
+
+ b.Property("Inactive")
+ .HasColumnType("bit");
+
+ b.Property("Kind")
+ .HasColumnType("int");
+
+ b.Property("LastException")
+ .HasColumnType("nvarchar(max)");
+
+ b.Property("LastHeartbeatOn")
+ .HasColumnType("datetime2");
+
+ b.Property("LastKnownState")
+ .HasMaxLength(100)
+ .IsUnicode(false)
+ .HasColumnType("varchar(100)");
+
+ b.Property("ModifiedBy")
+ .HasColumnType("nvarchar(max)");
+
+ b.Property("ModifiedOn")
+ .HasColumnType("datetime2");
+
+ b.Property("Name")
+ .IsRequired()
+ .HasMaxLength(200)
+ .HasColumnType("nvarchar(200)");
+
+ b.Property("OwnedByNode")
+ .HasMaxLength(200)
+ .IsUnicode(false)
+ .HasColumnType("varchar(200)");
+
+ b.Property("Placement")
+ .HasColumnType("int");
+
+ b.Property("Properties")
+ .HasColumnType("nvarchar(max)");
+
+ b.Property("SecretProperties")
+ .HasColumnType("nvarchar(max)");
+
+ b.Property