766 lines
31 KiB
C#
766 lines
31 KiB
C#
using System.Text;
|
||
using System.Text.Json;
|
||
|
||
namespace Zcbot.WindowsNode;
|
||
|
||
internal sealed class JobInboxStore(string jobsDirectory)
|
||
{
|
||
private static readonly JsonSerializerOptions JsonOptions = new() { WriteIndented = true };
|
||
private static readonly HashSet<string> PlotTypes =
|
||
[
|
||
"line", "scatter", "line_scatter", "column", "bar", "grouped_column",
|
||
"y_error", "contour", "surface_3d", "ternary", "heatmap",
|
||
];
|
||
private static readonly HashSet<string> XyzPlotTypes =
|
||
["contour", "surface_3d", "ternary", "heatmap"];
|
||
private static readonly HashSet<string> JobStages =
|
||
[
|
||
"accepted",
|
||
"downloading_inputs",
|
||
"ready_to_run",
|
||
"software_running",
|
||
"uploading_outputs",
|
||
"succeeded",
|
||
"failed",
|
||
"cancelled",
|
||
];
|
||
|
||
// Origin 执行槽只由尚无终态的任务占用。成功但上传确认尚未落盘的任务会由
|
||
// 心跳恢复管线继续重传；上传不使用 Origin，不能反向阻塞新的绘图任务。
|
||
internal bool HasPendingOriginJobs => Directory.Exists(jobsDirectory)
|
||
&& ReadRecoverableJobs().Any(item => item.Terminal is null);
|
||
|
||
internal IReadOnlyList<RecoverableJob> ReadRecoverableJobs()
|
||
{
|
||
if (!Directory.Exists(jobsDirectory))
|
||
{
|
||
return [];
|
||
}
|
||
var jobs = new List<RecoverableJob>();
|
||
foreach (var requestPath in Directory.EnumerateFiles(
|
||
jobsDirectory, "request.json", SearchOption.AllDirectories))
|
||
{
|
||
try
|
||
{
|
||
using var request = JsonDocument.Parse(File.ReadAllBytes(requestPath));
|
||
var root = request.RootElement;
|
||
if (!TryReadGuid(root, "job_id", out var jobId)
|
||
|| !TryReadGuid(root, "lease_id", out var leaseId)
|
||
|| !root.TryGetProperty("request_digest", out var digestValue)
|
||
|| digestValue.GetString() is not { Length: 64 } requestDigest)
|
||
{
|
||
continue;
|
||
}
|
||
var jobDirectory = Directory.GetParent(Directory.GetParent(requestPath)!.FullName)!.FullName;
|
||
jobs.Add(new RecoverableJob(
|
||
jobId,
|
||
leaseId,
|
||
requestDigest,
|
||
root.TryGetProperty("input_transfers", out var transfers)
|
||
? transfers.Clone() : null,
|
||
ReadTerminal(Path.Combine(jobDirectory, "terminal.json")),
|
||
File.Exists(Path.Combine(jobDirectory, "upload-complete.json"))));
|
||
}
|
||
catch (Exception exception) when (
|
||
exception is JsonException or IOException or UnauthorizedAccessException)
|
||
{
|
||
}
|
||
}
|
||
return jobs;
|
||
}
|
||
|
||
internal IReadOnlyList<JobDisplaySnapshot> ReadJobSnapshots(int limit = 50)
|
||
{
|
||
if (!Directory.Exists(jobsDirectory))
|
||
{
|
||
return [];
|
||
}
|
||
var snapshots = new List<JobDisplaySnapshot>();
|
||
foreach (var requestPath in Directory.EnumerateFiles(
|
||
jobsDirectory, "request.json", SearchOption.AllDirectories))
|
||
{
|
||
try
|
||
{
|
||
using var requestDocument = JsonDocument.Parse(File.ReadAllBytes(requestPath));
|
||
var root = requestDocument.RootElement;
|
||
if (!TryReadGuid(root, "job_id", out var jobId))
|
||
{
|
||
continue;
|
||
}
|
||
var jobDirectory = Directory.GetParent(
|
||
Directory.GetParent(requestPath)!.FullName)!.FullName;
|
||
var acceptedAt = ReadDate(root, "accepted_at")
|
||
?? new DateTimeOffset(File.GetCreationTimeUtc(requestPath));
|
||
var capability = ReadString(root, "capability", "unknown");
|
||
var title = "未命名任务";
|
||
if (root.TryGetProperty("request", out var request)
|
||
&& request.TryGetProperty("operation", out var operation)
|
||
&& operation.TryGetProperty("plot", out var plot))
|
||
{
|
||
title = ReadString(plot, "title", title);
|
||
}
|
||
var inputFilename = root.TryGetProperty("input_transfers", out var transfers)
|
||
&& transfers.ValueKind == JsonValueKind.Array
|
||
? string.Join(", ", transfers.EnumerateArray()
|
||
.Select(item => ReadString(item, "filename", "-")))
|
||
: "-";
|
||
var state = ReadState(Path.Combine(jobDirectory, "state.json"));
|
||
var terminal = ReadTerminal(Path.Combine(jobDirectory, "terminal.json"));
|
||
var uploadPath = Path.Combine(jobDirectory, "upload-complete.json");
|
||
var uploadComplete = File.Exists(uploadPath);
|
||
var stage = state?.Stage ?? "accepted";
|
||
var progress = state?.Progress ?? 0;
|
||
var detail = state?.Detail ?? "任务已由本机接收";
|
||
var updatedAt = state is not null
|
||
&& state.UpdatedAt != DateTimeOffset.MinValue
|
||
&& state.UpdatedAt > acceptedAt
|
||
? state.UpdatedAt
|
||
: acceptedAt;
|
||
if (terminal is JsonElement terminalValue)
|
||
{
|
||
var terminalStatus = ReadString(terminalValue, "status", "failed");
|
||
if (terminalStatus == "succeeded" && !uploadComplete)
|
||
{
|
||
stage = "uploading_outputs";
|
||
progress = Math.Max(progress, 90);
|
||
detail = state?.Stage == "uploading_outputs"
|
||
? state.Detail
|
||
: "软件执行完成，等待上传结果";
|
||
}
|
||
else
|
||
{
|
||
stage = terminalStatus;
|
||
progress = terminalStatus == "succeeded" ? 100 : progress;
|
||
detail = TerminalDetail(terminalValue, terminalStatus, detail);
|
||
}
|
||
updatedAt = LatestWrite(updatedAt, Path.Combine(jobDirectory, "terminal.json"));
|
||
}
|
||
if (uploadComplete)
|
||
{
|
||
stage = "succeeded";
|
||
progress = 100;
|
||
detail = "结果已上传并由云端确认";
|
||
updatedAt = LatestWrite(updatedAt, uploadPath);
|
||
}
|
||
snapshots.Add(new JobDisplaySnapshot(
|
||
jobId,
|
||
capability,
|
||
title,
|
||
inputFilename,
|
||
stage,
|
||
Math.Clamp(progress, 0, 100),
|
||
detail,
|
||
acceptedAt,
|
||
updatedAt,
|
||
uploadComplete));
|
||
}
|
||
catch (Exception exception) when (
|
||
exception is JsonException or IOException or UnauthorizedAccessException)
|
||
{
|
||
}
|
||
}
|
||
return snapshots
|
||
.OrderByDescending(item => item.IsActive)
|
||
.ThenByDescending(item => item.UpdatedAt)
|
||
.Take(Math.Max(1, limit))
|
||
.ToArray();
|
||
}
|
||
|
||
internal void WriteState(RecoverableJob job, string stage, int progress, string detail)
|
||
{
|
||
if (!JobStages.Contains(stage))
|
||
{
|
||
throw new InvalidDataException("Unsupported local job stage.");
|
||
}
|
||
var path = Path.Combine(
|
||
jobsDirectory, job.JobId.ToString("D"), "state.json");
|
||
var content = JsonSerializer.SerializeToUtf8Bytes(new
|
||
{
|
||
stage,
|
||
progress = Math.Clamp(progress, 0, 100),
|
||
detail = detail[..Math.Min(detail.Length, 500)],
|
||
updated_at = DateTimeOffset.UtcNow,
|
||
}, JsonOptions);
|
||
try
|
||
{
|
||
AtomicWrite(path, content, overwrite: true);
|
||
}
|
||
catch (Exception exception) when (
|
||
exception is IOException or UnauthorizedAccessException)
|
||
{
|
||
Console.Error.WriteLine($"[WARN] Local job state update failed: {exception.Message}");
|
||
}
|
||
}
|
||
|
||
internal JobOfferResult Accept(JsonElement payload)
|
||
{
|
||
if (!TryReadGuid(payload, "job_id", out var jobId)
|
||
|| !TryReadGuid(payload, "lease_id", out var leaseId)
|
||
|| !payload.TryGetProperty("request_digest", out var digestValue)
|
||
|| digestValue.ValueKind != JsonValueKind.String
|
||
|| digestValue.GetString() is not { Length: 64 } requestDigest
|
||
|| !payload.TryGetProperty("capability", out var capabilityValue)
|
||
|| capabilityValue.GetString() != "origin.plot@v2"
|
||
|| !payload.TryGetProperty("request", out var request)
|
||
|| request.ValueKind != JsonValueKind.Object
|
||
|| !payload.TryGetProperty("input_transfers", out var inputTransfers)
|
||
|| !IsValidInputTransfers(inputTransfers, jobId))
|
||
{
|
||
return JobOfferResult.Reject("invalid_offer");
|
||
}
|
||
if (!IsValidRequest(request))
|
||
{
|
||
return JobOfferResult.Reject("unsupported_request");
|
||
}
|
||
if (!InputsMatchTransfers(request, inputTransfers))
|
||
{
|
||
return JobOfferResult.Reject("invalid_offer");
|
||
}
|
||
var directory = Path.Combine(jobsDirectory, jobId.ToString("D"));
|
||
var requestDirectory = Path.Combine(directory, "request");
|
||
var requestPath = Path.Combine(requestDirectory, "request.json");
|
||
Directory.CreateDirectory(requestDirectory);
|
||
if (File.Exists(requestPath))
|
||
{
|
||
try
|
||
{
|
||
using var existing = JsonDocument.Parse(File.ReadAllBytes(requestPath));
|
||
var root = existing.RootElement;
|
||
var sameDigest = root.TryGetProperty("request_digest", out var existingDigest)
|
||
&& existingDigest.GetString() == requestDigest;
|
||
if (!sameDigest)
|
||
{
|
||
return JobOfferResult.Reject("job_digest_conflict");
|
||
}
|
||
if (!TryReadGuid(root, "lease_id", out var existingLease)
|
||
|| existingLease != leaseId)
|
||
{
|
||
var updated = JsonSerializer.SerializeToUtf8Bytes(new
|
||
{
|
||
job_id = jobId,
|
||
lease_id = leaseId,
|
||
request_digest = requestDigest,
|
||
capability = "origin.plot@v2",
|
||
accepted_at = DateTimeOffset.UtcNow,
|
||
request,
|
||
input_transfers = inputTransfers,
|
||
}, JsonOptions);
|
||
AtomicWrite(requestPath, updated, overwrite: true);
|
||
}
|
||
EnsureAcceptedState(jobId);
|
||
return JobOfferResult.Accept(jobId, leaseId, requestDigest);
|
||
}
|
||
catch (JsonException)
|
||
{
|
||
return JobOfferResult.Reject("local_job_record_invalid");
|
||
}
|
||
}
|
||
|
||
var record = JsonSerializer.SerializeToUtf8Bytes(new
|
||
{
|
||
job_id = jobId,
|
||
lease_id = leaseId,
|
||
request_digest = requestDigest,
|
||
capability = "origin.plot@v2",
|
||
accepted_at = DateTimeOffset.UtcNow,
|
||
request,
|
||
input_transfers = inputTransfers,
|
||
}, JsonOptions);
|
||
try
|
||
{
|
||
AtomicWrite(requestPath, record, overwrite: false);
|
||
EnsureAcceptedState(jobId);
|
||
return JobOfferResult.Accept(jobId, leaseId, requestDigest);
|
||
}
|
||
catch (IOException)
|
||
{
|
||
return JobOfferResult.Reject("local_job_persist_failed");
|
||
}
|
||
}
|
||
|
||
private static JsonElement? ReadTerminal(string path)
|
||
{
|
||
if (!File.Exists(path))
|
||
{
|
||
return null;
|
||
}
|
||
using var document = JsonDocument.Parse(File.ReadAllBytes(path));
|
||
return document.RootElement.Clone();
|
||
}
|
||
|
||
private void EnsureAcceptedState(Guid jobId)
|
||
{
|
||
var path = Path.Combine(jobsDirectory, jobId.ToString("D"), "state.json");
|
||
if (File.Exists(path))
|
||
{
|
||
return;
|
||
}
|
||
var content = JsonSerializer.SerializeToUtf8Bytes(new
|
||
{
|
||
stage = "accepted",
|
||
progress = 0,
|
||
detail = "任务已由本机接收",
|
||
updated_at = DateTimeOffset.UtcNow,
|
||
}, JsonOptions);
|
||
try
|
||
{
|
||
AtomicWrite(path, content, overwrite: false);
|
||
}
|
||
catch (IOException) when (File.Exists(path))
|
||
{
|
||
}
|
||
catch (Exception exception) when (
|
||
exception is IOException or UnauthorizedAccessException)
|
||
{
|
||
Console.Error.WriteLine($"[WARN] Initial local job state failed: {exception.Message}");
|
||
}
|
||
}
|
||
|
||
private static LocalJobState? ReadState(string path)
|
||
{
|
||
if (!File.Exists(path))
|
||
{
|
||
return null;
|
||
}
|
||
try
|
||
{
|
||
using var document = JsonDocument.Parse(File.ReadAllBytes(path));
|
||
var root = document.RootElement;
|
||
var stage = ReadString(root, "stage", "accepted");
|
||
var progress = root.TryGetProperty("progress", out var progressValue)
|
||
&& progressValue.TryGetInt32(out var parsedProgress)
|
||
? parsedProgress
|
||
: 0;
|
||
return new LocalJobState(
|
||
stage,
|
||
progress,
|
||
ReadString(root, "detail", ""),
|
||
ReadDate(root, "updated_at") ?? DateTimeOffset.MinValue);
|
||
}
|
||
catch (Exception exception) when (
|
||
exception is JsonException or IOException or UnauthorizedAccessException)
|
||
{
|
||
return null;
|
||
}
|
||
}
|
||
|
||
private static string TerminalDetail(JsonElement terminal, string status, string fallback)
|
||
{
|
||
if (terminal.TryGetProperty("error", out var error))
|
||
{
|
||
var detail = ReadString(error, "detail", "");
|
||
if (!string.IsNullOrWhiteSpace(detail))
|
||
{
|
||
return detail;
|
||
}
|
||
}
|
||
return status switch
|
||
{
|
||
"succeeded" => "软件任务执行成功",
|
||
"cancelled" => "任务已取消",
|
||
"failed" => "任务执行失败",
|
||
_ => fallback,
|
||
};
|
||
}
|
||
|
||
private static DateTimeOffset LatestWrite(DateTimeOffset current, string path)
|
||
{
|
||
var writtenAt = new DateTimeOffset(File.GetLastWriteTimeUtc(path));
|
||
return writtenAt > current ? writtenAt : current;
|
||
}
|
||
|
||
private static string ReadString(JsonElement value, string name, string fallback) =>
|
||
value.TryGetProperty(name, out var property)
|
||
&& property.ValueKind == JsonValueKind.String
|
||
&& !string.IsNullOrWhiteSpace(property.GetString())
|
||
? property.GetString()!
|
||
: fallback;
|
||
|
||
private static DateTimeOffset? ReadDate(JsonElement value, string name) =>
|
||
value.TryGetProperty(name, out var property)
|
||
&& property.ValueKind == JsonValueKind.String
|
||
&& DateTimeOffset.TryParse(property.GetString(), out var parsed)
|
||
? parsed
|
||
: null;
|
||
|
||
private static void AtomicWrite(string path, byte[] content, bool overwrite)
|
||
{
|
||
Directory.CreateDirectory(Path.GetDirectoryName(path)!);
|
||
var temporaryPath = path + ".tmp-" + Guid.NewGuid().ToString("N");
|
||
try
|
||
{
|
||
using (var stream = new FileStream(
|
||
temporaryPath, FileMode.CreateNew, FileAccess.Write, FileShare.None,
|
||
bufferSize: 4096, FileOptions.WriteThrough))
|
||
{
|
||
stream.Write(content);
|
||
stream.Flush(flushToDisk: true);
|
||
}
|
||
File.Move(temporaryPath, path, overwrite);
|
||
}
|
||
finally
|
||
{
|
||
if (File.Exists(temporaryPath)) File.Delete(temporaryPath);
|
||
}
|
||
}
|
||
|
||
private static bool IsValidRequest(JsonElement request)
|
||
{
|
||
if (!HasOnlyProperties(request, "schema_version", "inputs", "operation", "outputs")
|
||
|| !request.TryGetProperty("schema_version", out var schemaVersion)
|
||
|| !schemaVersion.TryGetInt32(out var version)
|
||
|| version != 2
|
||
|| !request.TryGetProperty("inputs", out var inputs)
|
||
|| !IsValidInputBindings(inputs)
|
||
|| !request.TryGetProperty("operation", out var operation)
|
||
|| operation.ValueKind != JsonValueKind.Object
|
||
|| !HasOnlyProperties(operation, "plot")
|
||
|| !operation.TryGetProperty("plot", out var plot)
|
||
|| !IsValidPlot(plot, inputs)
|
||
|| !request.TryGetProperty("outputs", out var outputs))
|
||
{
|
||
return false;
|
||
}
|
||
return IsValidOutputs(outputs);
|
||
}
|
||
|
||
private static bool IsValidOutputs(JsonElement outputs)
|
||
{
|
||
if (outputs.ValueKind != JsonValueKind.Array || outputs.GetArrayLength() is < 1 or > 16)
|
||
{
|
||
return false;
|
||
}
|
||
var keys = new HashSet<string>(StringComparer.Ordinal);
|
||
var identities = new HashSet<string>(StringComparer.Ordinal);
|
||
foreach (var output in outputs.EnumerateArray())
|
||
{
|
||
if (output.ValueKind != JsonValueKind.Object
|
||
|| !HasOnlyProperties(output, "key", "type", "format", "options")
|
||
|| !output.TryGetProperty("key", out var keyValue)
|
||
|| keyValue.GetString() is not { } key
|
||
|| !output.TryGetProperty("type", out var typeValue)
|
||
|| typeValue.GetString() is not { } outputType
|
||
|| !output.TryGetProperty("format", out var formatValue)
|
||
|| formatValue.GetString() is not { } format)
|
||
{
|
||
return false;
|
||
}
|
||
var expectedKey = (outputType, format) switch
|
||
{
|
||
("project", "opju") => "project",
|
||
("figure", "png") => "figure_png",
|
||
("figure", "svg") => "figure_svg",
|
||
("figure", "pdf") => "figure_pdf",
|
||
_ => "",
|
||
};
|
||
if (key != expectedKey || !keys.Add(key) || !identities.Add($"{outputType}\0{format}"))
|
||
{
|
||
return false;
|
||
}
|
||
if (format == "png")
|
||
{
|
||
if (output.TryGetProperty("options", out var options)
|
||
&& (options.ValueKind != JsonValueKind.Object
|
||
|| !HasOnlyProperties(options, "dpi")
|
||
|| !options.TryGetProperty("dpi", out var dpi)
|
||
|| !dpi.TryGetInt32(out var dpiValue)
|
||
|| dpiValue is < 72 or > 1200))
|
||
{
|
||
return false;
|
||
}
|
||
}
|
||
else if (output.TryGetProperty("options", out _))
|
||
{
|
||
return false;
|
||
}
|
||
}
|
||
return true;
|
||
}
|
||
|
||
private static bool IsValidInputBindings(JsonElement inputs)
|
||
{
|
||
if (inputs.ValueKind != JsonValueKind.Array || inputs.GetArrayLength() is < 1 or > 16)
|
||
{
|
||
return false;
|
||
}
|
||
var keys = new HashSet<string>(StringComparer.Ordinal);
|
||
foreach (var input in inputs.EnumerateArray())
|
||
{
|
||
if (input.ValueKind != JsonValueKind.Object
|
||
|| !HasOnlyProperties(input, "key", "artifact_id", "selector")
|
||
|| !input.TryGetProperty("key", out var keyValue)
|
||
|| keyValue.GetString() is not { } key
|
||
|| !IsInputKey(key)
|
||
|| !keys.Add(key)
|
||
|| !input.TryGetProperty("artifact_id", out var artifactId)
|
||
|| !Guid.TryParse(artifactId.GetString(), out _)
|
||
|| input.TryGetProperty("selector", out var selector)
|
||
&& !IsValidSelector(selector))
|
||
{
|
||
return false;
|
||
}
|
||
}
|
||
return true;
|
||
}
|
||
|
||
private static bool IsValidPlot(JsonElement plot, JsonElement inputs)
|
||
{
|
||
if (plot.ValueKind != JsonValueKind.Object
|
||
|| !HasOnlyProperties(
|
||
plot, "type", "series", "template", "title", "x_axis", "y_axis", "z_axis",
|
||
"legend", "error_bars")
|
||
|| !plot.TryGetProperty("type", out var plotType)
|
||
|| plotType.GetString() is not { } plotTypeName
|
||
|| !PlotTypes.Contains(plotTypeName)
|
||
|| plot.TryGetProperty("title", out var title)
|
||
&& (title.ValueKind != JsonValueKind.String || title.GetString()!.Length > 500)
|
||
|| !plot.TryGetProperty("series", out var series)
|
||
|| series.ValueKind != JsonValueKind.Array
|
||
|| series.GetArrayLength() is < 1 or > 16
|
||
|| plot.TryGetProperty("template", out var template)
|
||
&& template.GetString() != "publication_double_column"
|
||
|| !IsValidAxis(plot, "x_axis")
|
||
|| !IsValidAxis(plot, "y_axis")
|
||
|| !IsValidAxis(plot, "z_axis")
|
||
|| !IsValidLegend(plot)
|
||
|| plot.TryGetProperty("error_bars", out _))
|
||
{
|
||
return false;
|
||
}
|
||
if ((plotTypeName == "grouped_column" && series.GetArrayLength() < 2)
|
||
|| (XyzPlotTypes.Contains(plotTypeName) && series.GetArrayLength() != 1))
|
||
{
|
||
return false;
|
||
}
|
||
var requiredRoles = XyzPlotTypes.Contains(plotTypeName)
|
||
? new[] { "x", "y", "z" }
|
||
: plotTypeName == "y_error"
|
||
? new[] { "x", "y", "y_error" }
|
||
: new[] { "x", "y" };
|
||
var allowedRoles = requiredRoles.ToHashSet(StringComparer.Ordinal);
|
||
var inputKeys = inputs.EnumerateArray()
|
||
.Select(item => item.GetProperty("key").GetString()!)
|
||
.ToHashSet(StringComparer.Ordinal);
|
||
var identities = new HashSet<string>(StringComparer.Ordinal);
|
||
var usedInputs = new HashSet<string>(StringComparer.Ordinal);
|
||
var labels = new Dictionary<string, string>(StringComparer.Ordinal);
|
||
foreach (var item in series.EnumerateArray())
|
||
{
|
||
if (item.ValueKind != JsonValueKind.Object
|
||
|| !HasOnlyProperties(item, "input", "x", "y", "z", "y_error", "label")
|
||
|| !item.TryGetProperty("input", out var input)
|
||
|| input.GetString() is not { } inputKey
|
||
|| !inputKeys.Contains(inputKey)
|
||
|| requiredRoles.Any(role =>
|
||
!item.TryGetProperty(role, out var column) || !IsColumnName(column))
|
||
|| new[] { "x", "y", "z", "y_error" }.Any(role =>
|
||
!allowedRoles.Contains(role) && item.TryGetProperty(role, out _))
|
||
|| item.TryGetProperty("label", out var label)
|
||
&& (label.ValueKind != JsonValueKind.String
|
||
|| label.GetString()!.Length is < 1 or > 200)
|
||
|| !identities.Add(string.Join(
|
||
"\0", new[] { inputKey }.Concat(requiredRoles.Select(
|
||
role => item.GetProperty(role).GetString()!)))))
|
||
{
|
||
return false;
|
||
}
|
||
usedInputs.Add(inputKey);
|
||
var yValue = item.GetProperty("y").GetString()!;
|
||
var labelKey = $"{inputKey}\0{yValue}";
|
||
var effectiveLabel = item.TryGetProperty("label", out var seriesLabel)
|
||
? seriesLabel.GetString()!
|
||
: yValue;
|
||
if (labels.TryGetValue(labelKey, out var existingLabel)
|
||
&& existingLabel != effectiveLabel)
|
||
{
|
||
return false;
|
||
}
|
||
labels[labelKey] = effectiveLabel;
|
||
}
|
||
return usedInputs.SetEquals(inputKeys);
|
||
}
|
||
|
||
private static bool IsValidSelector(JsonElement selector) =>
|
||
selector.ValueKind == JsonValueKind.Object
|
||
&& HasOnlyProperties(selector, "sheet")
|
||
&& selector.TryGetProperty("sheet", out var sheet)
|
||
&& sheet.ValueKind == JsonValueKind.String
|
||
&& sheet.GetString()!.Length is >= 1 and <= 128;
|
||
|
||
private static bool IsInputKey(string key) =>
|
||
key.Length is >= 1 and <= 32
|
||
&& key[0] is >= 'a' and <= 'z'
|
||
&& key.All(character =>
|
||
character is >= 'a' and <= 'z'
|
||
|| character is >= '0' and <= '9'
|
||
|| character == '_');
|
||
|
||
private static bool IsColumnName(JsonElement value) =>
|
||
value.ValueKind == JsonValueKind.String
|
||
&& value.GetString() is { Length: >= 1 and <= 128 };
|
||
|
||
private static bool IsValidAxis(JsonElement plot, string name)
|
||
{
|
||
if (!plot.TryGetProperty(name, out var axis)) return true;
|
||
return axis.ValueKind == JsonValueKind.Object
|
||
&& HasOnlyProperties(axis, "title", "unit", "scale")
|
||
&& (!axis.TryGetProperty("title", out var title) || title.ValueKind == JsonValueKind.String)
|
||
&& (!axis.TryGetProperty("unit", out var unit) || unit.ValueKind == JsonValueKind.String)
|
||
&& (!axis.TryGetProperty("scale", out var scale) || scale.GetString() == "linear");
|
||
}
|
||
|
||
private static bool IsValidLegend(JsonElement plot)
|
||
{
|
||
if (!plot.TryGetProperty("legend", out var legend)) return true;
|
||
return legend.ValueKind == JsonValueKind.Object
|
||
&& HasOnlyProperties(legend, "enabled", "position")
|
||
&& (!legend.TryGetProperty("enabled", out var enabled)
|
||
|| enabled.ValueKind == JsonValueKind.True)
|
||
&& (!legend.TryGetProperty("position", out var position)
|
||
|| position.GetString() == "top_right");
|
||
}
|
||
|
||
private static bool IsValidInputTransfers(JsonElement transfers, Guid jobId)
|
||
{
|
||
if (transfers.ValueKind != JsonValueKind.Array
|
||
|| transfers.GetArrayLength() is < 1 or > 16)
|
||
{
|
||
return false;
|
||
}
|
||
var keys = new HashSet<string>(StringComparer.Ordinal);
|
||
long total = 0;
|
||
foreach (var transfer in transfers.EnumerateArray())
|
||
{
|
||
if (transfer.ValueKind != JsonValueKind.Object
|
||
|| !HasOnlyProperties(
|
||
transfer, "key", "artifact_id", "filename", "size_bytes", "sha256",
|
||
"selector", "download_path")
|
||
|| !transfer.TryGetProperty("key", out var keyValue)
|
||
|| keyValue.GetString() is not { } key
|
||
|| !IsInputKey(key)
|
||
|| !keys.Add(key)
|
||
|| !transfer.TryGetProperty("artifact_id", out var artifactId)
|
||
|| !Guid.TryParse(artifactId.GetString(), out _)
|
||
|| !transfer.TryGetProperty("filename", out var filename)
|
||
|| filename.ValueKind != JsonValueKind.String
|
||
|| Path.GetFileName(filename.GetString()) != filename.GetString()
|
||
|| !transfer.TryGetProperty("size_bytes", out var size)
|
||
|| !size.TryGetInt64(out var sizeBytes)
|
||
|| sizeBytes is < 0 or > 104_857_600
|
||
|| !transfer.TryGetProperty("sha256", out var sha)
|
||
|| sha.GetString() is not { Length: 64 }
|
||
|| transfer.TryGetProperty("selector", out var selector)
|
||
&& !IsValidSelector(selector)
|
||
|| !transfer.TryGetProperty("download_path", out var downloadPath)
|
||
|| downloadPath.GetString()
|
||
!= $"/v1/software-jobs/{jobId:D}/inputs/{key}")
|
||
{
|
||
return false;
|
||
}
|
||
total += sizeBytes;
|
||
}
|
||
return total <= 536_870_912;
|
||
}
|
||
|
||
private static bool InputsMatchTransfers(JsonElement request, JsonElement transfers)
|
||
{
|
||
var bindings = request.GetProperty("inputs").EnumerateArray().ToDictionary(
|
||
item => item.GetProperty("key").GetString()!, StringComparer.Ordinal);
|
||
if (bindings.Count != transfers.GetArrayLength())
|
||
{
|
||
return false;
|
||
}
|
||
foreach (var transfer in transfers.EnumerateArray())
|
||
{
|
||
var key = transfer.GetProperty("key").GetString()!;
|
||
if (!bindings.TryGetValue(key, out var binding)
|
||
|| binding.GetProperty("artifact_id").GetString()
|
||
!= transfer.GetProperty("artifact_id").GetString())
|
||
{
|
||
return false;
|
||
}
|
||
var hasBindingSelector = binding.TryGetProperty("selector", out var bindingSelector);
|
||
var hasTransferSelector = transfer.TryGetProperty("selector", out var transferSelector);
|
||
if (hasBindingSelector != hasTransferSelector
|
||
|| hasBindingSelector
|
||
&& bindingSelector.GetProperty("sheet").GetString()
|
||
!= transferSelector.GetProperty("sheet").GetString())
|
||
{
|
||
return false;
|
||
}
|
||
}
|
||
return true;
|
||
}
|
||
|
||
internal IReadOnlyList<JsonElement> InputTransfers(RecoverableJob job)
|
||
{
|
||
if (job.InputTransfers is not JsonElement transfers
|
||
|| !IsValidInputTransfers(transfers, job.JobId))
|
||
{
|
||
throw new InvalidDataException("Stored input transfers are invalid.");
|
||
}
|
||
return transfers.EnumerateArray().Select(item => item.Clone()).ToArray();
|
||
}
|
||
|
||
internal string InputPath(RecoverableJob job, JsonElement transfer)
|
||
{
|
||
var key = transfer.GetProperty("key").GetString()!;
|
||
var filename = transfer.GetProperty("filename").GetString()!;
|
||
return Path.Combine(
|
||
jobsDirectory, job.JobId.ToString("D"), "input", key, filename);
|
||
}
|
||
|
||
internal void WriteTerminal(
|
||
RecoverableJob job, string status, string code, string detail)
|
||
{
|
||
var path = Path.Combine(jobsDirectory, job.JobId.ToString("D"), "terminal.json");
|
||
if (File.Exists(path))
|
||
{
|
||
return;
|
||
}
|
||
var content = JsonSerializer.SerializeToUtf8Bytes(new
|
||
{
|
||
job_id = job.JobId,
|
||
lease_id = job.LeaseId,
|
||
request_digest = job.RequestDigest,
|
||
status,
|
||
error = new { code, detail },
|
||
artifact_manifest = Array.Empty<object>(),
|
||
terminal_at = DateTimeOffset.UtcNow,
|
||
}, JsonOptions);
|
||
AtomicWrite(path, content, overwrite: false);
|
||
}
|
||
|
||
private static bool HasOnlyProperties(JsonElement value, params string[] allowed)
|
||
{
|
||
var names = new HashSet<string>(allowed, StringComparer.Ordinal);
|
||
return value.EnumerateObject().All(item => names.Contains(item.Name));
|
||
}
|
||
|
||
private static bool TryReadGuid(JsonElement payload, string name, out Guid value)
|
||
{
|
||
value = Guid.Empty;
|
||
return payload.TryGetProperty(name, out var property)
|
||
&& property.ValueKind == JsonValueKind.String
|
||
&& Guid.TryParse(property.GetString(), out value);
|
||
}
|
||
}
|
||
|
||
internal sealed record RecoverableJob(
|
||
Guid JobId,
|
||
Guid LeaseId,
|
||
string RequestDigest,
|
||
JsonElement? InputTransfers,
|
||
JsonElement? Terminal,
|
||
bool UploadComplete);
|
||
|
||
internal sealed record JobOfferResult(
|
||
bool Accepted, Guid JobId, Guid LeaseId, string RequestDigest, string Reason)
|
||
{
|
||
internal static JobOfferResult Accept(Guid jobId, Guid leaseId, string requestDigest) =>
|
||
new(true, jobId, leaseId, requestDigest, "");
|
||
|
||
internal static JobOfferResult Reject(string reason) =>
|
||
new(false, Guid.Empty, Guid.Empty, "", reason);
|
||
}
|