diff --git a/SW.Bitween.Api/Controllers/XchangeExportController.cs b/SW.Bitween.Api/Controllers/XchangeExportController.cs
index 866efb3e..264e7981 100644
--- a/SW.Bitween.Api/Controllers/XchangeExportController.cs
+++ b/SW.Bitween.Api/Controllers/XchangeExportController.cs
@@ -57,10 +57,22 @@ public class XchangeExportController(BitweenDbContext dbContext, RequestContext
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.
+ /// The largest zip built: a file that would take it past this is left out, and so are the exchanges
+ /// after the zip reaches it, with missing.txt saying so. Well inside the 2 GB of temporary disk the
+ /// Helm chart gives a node, with keeping it to one zip at a time.
+ /// Set lower by tests.
///
- public const long MaxBytes = 1L << 30;
+ public static long MaxBytes { get; internal set; } = 1L << 30;
+
+ ///
+ /// One export built in the background on a node at a time: each is up to on
+ /// its temporary disk, and two at once could fill it. Exports asked directly stream to the browser
+ /// instead, and take no disk.
+ ///
+ internal static readonly SemaphoreSlim BuildingInBackground = new(1, 1);
+
+ /// How often an export waiting its turn says so, which keeps its time limit from running out. Set shorter by tests.
+ internal static TimeSpan SayWaitingEvery { get; set; } = TimeSpan.FromSeconds(5);
public const string MissingList = "missing.txt";
@@ -102,7 +114,32 @@ public async Task Post([FromBody] XchangeFilesExport request)
var background = Services.Background.BackgroundRequestScope.Current;
var (ids, refusal) = await Resolve(request, background != null ? MaxInBackground : Limit);
if (refusal != null) return refusal;
+ if (background == null) return await Build(ids, background);
+
+ // Its turn on this node's temporary disk. Cancelled while it waits, it builds no zip but an
+ // empty one whose missing.txt says so, as a cancel part way does.
+ var held = false;
+ try
+ {
+ while (!(held = await BuildingInBackground.WaitAsync(SayWaitingEvery, background.CancelRequested)))
+ background.Progress(0, ids.Count, "Waiting for another export on this server to finish");
+ }
+ catch (OperationCanceledException)
+ {
+ }
+ try
+ {
+ return await Build(ids, background);
+ }
+ finally
+ {
+ if (held) BuildingInBackground.Release();
+ }
+ }
+
+ private async Task Build(List ids, Services.Background.BackgroundRequestScope background)
+ {
var xchanges = await dbContext.Set().AsNoTracking()
.Where(x => ids.Contains(x.Id)).OrderByDescending(x => x.StartedOn).ToListAsync();
var results = await dbContext.Set().AsNoTracking()
@@ -147,7 +184,7 @@ await dbContext.RecordNowAsync(requestContext, OperatorAction.Export,
if (body.Written > MaxBytes)
{
- missing.Add($"The zip reached {MaxBytes / (1024 * 1024 * 1024)} GB after {exported} of {xchanges.Count} exchanges, " +
+ missing.Add($"The zip reached {Size(MaxBytes)} after {exported} of {xchanges.Count} exchanges, " +
"the most one export builds; the rest aren't here. Export them separately.");
break;
}
@@ -157,14 +194,14 @@ await dbContext.RecordNowAsync(requestContext, OperatorAction.Export,
var mainValue = mainProperty == null ? null : promoted.GetValueOrDefault(xchange.Id)?.Properties?.GetValueOrDefault(mainProperty);
var folder = ExchangeArchive.BaseName(mainValue, xchange.Id);
- await Add(zip, xchange, XchangeFileType.Input, $"{folder}/input",
+ await Add(zip, body, xchange, XchangeFileType.Input, $"{folder}/input",
xchange.InputName, xchange.InputSize, xchange.InputContentType, missing, aborted);
if (!results.TryGetValue(xchange.Id, out var result)) continue;
- await Add(zip, xchange, XchangeFileType.Output, $"{folder}/mapped",
+ await Add(zip, body, xchange, XchangeFileType.Output, $"{folder}/mapped",
result.OutputName, result.OutputSize, result.OutputContentType, missing, aborted,
result.NotStoredReason(XchangeFileType.Output));
- await Add(zip, xchange, XchangeFileType.Response, $"{folder}/handled",
+ await Add(zip, body, xchange, XchangeFileType.Response, $"{folder}/handled",
result.ResponseName, result.ResponseSize, result.ResponseContentType, missing, aborted,
result.NotStoredReason(XchangeFileType.Response));
}
@@ -188,8 +225,8 @@ await list.WriteAsync(Encoding.UTF8.GetBytes(
/// gives. A file the exchange never had is left out; one storage no longer has, or never got because it
/// couldn't be stored after the delivery (), goes on the missing list.
///
- private async Task Add(ZipArchive zip, Xchange xchange, XchangeFileType type, string path, string name, int size,
- string contentType, List missing, CancellationToken aborted, string notStored = null)
+ private async Task Add(ZipArchive zip, HeldSyncWrites body, Xchange xchange, XchangeFileType type, string path,
+ string name, int size, string contentType, List missing, CancellationToken aborted, string notStored = null)
{
if (notStored != null)
{
@@ -200,6 +237,15 @@ private async Task Add(ZipArchive zip, Xchange xchange, XchangeFileType type, st
if (size == 0) return;
path += ExtensionOf(name, contentType);
+ // Checked before it is read rather than after: one large file could otherwise take the zip, and
+ // the disk it is built on, well past the limit. Its stored size; compressed, it may be less.
+ if (body.Written + size > MaxBytes)
+ {
+ missing.Add($"{path}: left out, since it would take the zip past {Size(MaxBytes)}, the most one export builds. " +
+ "Export this exchange on its own.");
+ return;
+ }
+
Stream file;
try
{
@@ -234,6 +280,9 @@ private async Task Add(ZipArchive zip, Xchange xchange, XchangeFileType type, st
}
}
+ private static string Size(long bytes) =>
+ bytes >= 1L << 30 ? $"{bytes / (1L << 30)} GB" : bytes >= 1L << 20 ? $"{bytes / (1L << 20)} MB" : $"{bytes:n0} bytes";
+
/// The file's own extension when it has a plain one, or else one for its content type.
private static string ExtensionOf(string name, string contentType)
{
diff --git a/SW.Bitween.Api/Resources/XchangeExplanations/Explain.cs b/SW.Bitween.Api/Resources/XchangeExplanations/Explain.cs
index c123eb91..2dcf95a3 100644
--- a/SW.Bitween.Api/Resources/XchangeExplanations/Explain.cs
+++ b/SW.Bitween.Api/Resources/XchangeExplanations/Explain.cs
@@ -83,7 +83,7 @@ public async Task Handle(XchangeExplanationRequest r
await AddFile(files, masking, xchange, XchangeFileType.Output, result.OutputSize, result.OutputContentType);
await AddFile(files, masking, xchange, XchangeFileType.Response, result.ResponseSize, result.ResponseContentType);
var chain = await Chain(xchange);
- await SeeOtherAttempts(chain, xchange.Id, result.Exception, masking);
+ var sendable = await SeeOtherAttempts(chain, xchange.Id, result.Exception, masking);
sent.AppendLine("Failed step: " + FailedStep(xchange, result));
if (subscription is not null)
@@ -113,7 +113,7 @@ public async Task Handle(XchangeExplanationRequest r
{
sent.AppendLine();
sent.AppendLine("");
- foreach (var line in ChainLines(chain, xchange.Id, masking)) sent.AppendLine(line);
+ foreach (var line in ChainLines(chain, xchange.Id, masking, sendable)) sent.AppendLine(line);
sent.AppendLine("");
}
@@ -278,22 +278,45 @@ from r in results.DefaultIfEmpty()
///
/// Reads the documents of the other attempts whose error differs from this one's, so masking
/// knows their values too: a later attempt's response can hold what this one's documents don't.
+ /// Returns the attempts whose error line may be sent: this one, those with its very line, and
+ /// those whose documents were all read and masked. Any other's line is withheld, not guessed at.
///
- async Task SeeOtherAttempts(RetryChain chain, string thisId, string? thisError, Masking masking)
+ async Task> SeeOtherAttempts(RetryChain chain, string thisId, string? thisError, Masking masking)
{
var thisLine = WholeFirstLine(thisError ?? "");
- var ids = chain.Attempts
- .Where(a => a.Id != thisId && !string.IsNullOrWhiteSpace(a.Error) && WholeFirstLine(a.Error) != thisLine)
- .Select(a => a.Id).Take(OtherAttemptsRead).ToList();
- if (ids.Count == 0) return;
+ var sendable = chain.Attempts
+ .Where(a => a.Id == thisId || string.IsNullOrWhiteSpace(a.Error) || WholeFirstLine(a.Error) == thisLine)
+ .Select(a => a.Id).ToHashSet();
+ var ids = chain.Attempts.Where(a => !sendable.Contains(a.Id)).Select(a => a.Id).Take(OtherAttemptsRead).ToList();
+ if (ids.Count == 0) return sendable;
var xchanges = await dbContext.Set().AsNoTracking().Where(x => ids.Contains(x.Id)).ToListAsync();
var results = await dbContext.Set().AsNoTracking().Where(r => ids.Contains(r.Id)).ToDictionaryAsync(r => r.Id);
foreach (var x in xchanges)
{
if (!results.TryGetValue(x.Id, out var r)) continue;
- var ignored = new List<(string, string)>();
- await AddFile(ignored, masking, x, XchangeFileType.Output, r.OutputSize, r.OutputContentType);
- await AddFile(ignored, masking, x, XchangeFileType.Response, r.ResponseSize, r.ResponseContentType);
+ if (await Seen(masking, x, XchangeFileType.Output, r.OutputSize, r.OutputContentType) &&
+ await Seen(masking, x, XchangeFileType.Response, r.ResponseSize, r.ResponseContentType))
+ sendable.Add(x.Id);
+ }
+ return sendable;
+ }
+
+ ///
+ /// Masks one of another attempt's documents, for its values only: whether they are now known.
+ /// Not when it is too large, gone, unreadable, or in a format masking can't read.
+ ///
+ async Task Seen(Masking masking, Xchange xchange, XchangeFileType type, int size, string? contentType)
+ {
+ if (size <= 0) return true;
+ if (size > MaxMaskedFileBytes) return false;
+ try
+ {
+ return masking.Document(await xchangeService.GetFile(xchange, type), contentType) != null;
+ }
+ catch (Exception)
+ {
+ // Storage that can't give it back now costs this attempt's line, not the explanation.
+ return false;
}
}
@@ -301,7 +324,7 @@ async Task SeeOtherAttempts(RetryChain chain, string thisId, string? thisError,
/// The chain as the model reads it: one line per attempt, its outcome and its error's first line,
/// with a run of the same outcome and error folded into one line.
///
- internal static IEnumerable ChainLines(RetryChain chain, string thisId, Masking masking)
+ internal static IEnumerable ChainLines(RetryChain chain, string thisId, Masking masking, ISet sendable)
{
string Describe(Attempt a)
{
@@ -309,6 +332,7 @@ string Describe(Attempt a)
if (a.Success == true && !a.ResponseBad) return "succeeded";
var what = a.ResponseBad ? "delivered, bad response" : "failed";
if (string.IsNullOrWhiteSpace(a.Error)) return what;
+ if (!sendable.Contains(a.Id)) return $"{what}, with a different error, not sent: its documents couldn't be read to mask it";
// Masked whole, then cut: a cut first could leave half a value masking can't recognise.
var line = masking.Error(WholeFirstLine(a.Error));
return $"{what}: {(line.Length > MaxLineLength ? line[..MaxLineLength] + " ..." : line)}";
@@ -333,20 +357,17 @@ string Describe(Attempt a)
}
///
- /// An error's first line. Only characters are read, so a first line that
- /// runs past them is cut back to before its last, possibly partial, word - and before a quote it
- /// opened and never closed - rather than send half a value masking can't recognise.
+ /// An error's first line. Only characters are read, and a first line that
+ /// runs past them is cut somewhere masking can't see the whole of: a name's first word looks like
+ /// any other. So only what comes before its first ": " - the exception's type - is kept.
///
internal static string WholeFirstLine(string text)
{
var newline = text.IndexOf('\n');
if (newline >= 0) return text[..newline].Trim();
if (text.Length < ErrorRead) return text.Trim();
- var end = Math.Max(text.LastIndexOf(' '), 0);
- foreach (var quote in new[] { '\'', '"' })
- if (text[..end].Count(c => c == quote) % 2 == 1)
- end = text.LastIndexOf(quote, end - 1);
- return (end > 0 ? text[..end].Trim() + " " : "") + "...";
+ var type = text.IndexOf(": ", StringComparison.Ordinal);
+ return (type > 0 && type < 200 ? text[..type] + ": " : "") + "(the rest of this line is too long to send)";
}
static string? ContentTypeOf(Document? document) => document?.DocumentFormat switch
diff --git a/SW.Bitween.Api/Resources/Xchanges/BulkRetry.cs b/SW.Bitween.Api/Resources/Xchanges/BulkRetry.cs
index 3cfbc6ef..251b113e 100644
--- a/SW.Bitween.Api/Resources/Xchanges/BulkRetry.cs
+++ b/SW.Bitween.Api/Resources/Xchanges/BulkRetry.cs
@@ -69,38 +69,67 @@ public async Task