353 lines
14 KiB
C#
353 lines
14 KiB
C#
using System.Text;
|
||
using System.Text.Json;
|
||
|
||
namespace Zcbot.WindowsNode;
|
||
|
||
internal sealed class JobInboxStore(string jobsDirectory)
|
||
{
|
||
internal static readonly JsonSerializerOptions JsonOptions = new() { WriteIndented = true };
|
||
private static readonly HashSet<string> XyzPlotTypes =
|
||
["contour", "surface_3d", "ternary", "heatmap"];
|
||
|
||
// Origin 执行槽只由尚无终态的任务占用。成功但上传确认尚未落盘的任务会由
|
||
// 心跳恢复管线继续重传；上传不使用 Origin，不能反向阻塞新的绘图任务。
|
||
internal bool HasPendingJobs => new JobRepository(jobsDirectory).HasPendingJobs;
|
||
|
||
internal IReadOnlyList<RecoverableJob> ReadRecoverableJobs() =>
|
||
new JobRepository(jobsDirectory).ReadRecoverableJobs();
|
||
|
||
internal void WriteState(
|
||
RecoverableJob job,
|
||
string stage,
|
||
int progress,
|
||
string detail,
|
||
JobTransitionMode mode = JobTransitionMode.Normal) =>
|
||
new JobRepository(jobsDirectory).WriteState(job, stage, progress, detail, mode);
|
||
|
||
internal JobOfferResult Accept(JsonElement payload, AdapterCatalog adapters)
|
||
{
|
||
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() is not { Length: > 0 } capability
|
||
|| !payload.TryGetProperty("request", out var request)
|
||
|| request.ValueKind != JsonValueKind.Object
|
||
|| !payload.TryGetProperty("workspace", out var workspaceValue)
|
||
|| !TryReadWorkspace(workspaceValue, out var workspace)
|
||
|| !payload.TryGetProperty("input_transfers", out var inputTransfers)
|
||
|| !IsValidInputTransfers(inputTransfers, jobId))
|
||
{
|
||
return JobOfferResult.Reject("invalid_offer");
|
||
}
|
||
JsonElement? requestSummary = payload.TryGetProperty(
|
||
"request_summary", out var summaryValue)
|
||
&& summaryValue.ValueKind == JsonValueKind.Object
|
||
? summaryValue.Clone()
|
||
: null;
|
||
var adapter = adapters.Find(capability);
|
||
if (adapter is null || !adapter.ValidateRequest(request))
|
||
{
|
||
return JobOfferResult.Reject("unsupported_request");
|
||
}
|
||
if ((adapter.WorkspaceStateFilename is null) != (workspace is null))
|
||
{
|
||
return JobOfferResult.Reject("workspace_protocol_mismatch");
|
||
}
|
||
if (!InputsMatchTransfers(request, inputTransfers))
|
||
{
|
||
return JobOfferResult.Reject("invalid_offer");
|
||
}
|
||
var directory = PathGuard.CombineUnderRoot(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,
|
||
accepted_at = DateTimeOffset.UtcNow,
|
||
request,
|
||
request_summary = requestSummary,
|
||
workspace = workspace is null ? null : new
|
||
{
|
||
workspace_id = workspace.WorkspaceId,
|
||
source_job_id = workspace.SourceJobId,
|
||
mode = workspace.Mode,
|
||
},
|
||
input_transfers = inputTransfers,
|
||
}, JsonOptions);
|
||
AtomicFile.Write(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,
|
||
accepted_at = DateTimeOffset.UtcNow,
|
||
request,
|
||
request_summary = requestSummary,
|
||
workspace = workspace is null ? null : new
|
||
{
|
||
workspace_id = workspace.WorkspaceId,
|
||
source_job_id = workspace.SourceJobId,
|
||
mode = workspace.Mode,
|
||
},
|
||
input_transfers = inputTransfers,
|
||
}, JsonOptions);
|
||
try
|
||
{
|
||
AtomicFile.Write(requestPath, record, overwrite: false);
|
||
EnsureAcceptedState(jobId);
|
||
return JobOfferResult.Accept(jobId, leaseId, requestDigest);
|
||
}
|
||
catch (IOException)
|
||
{
|
||
return JobOfferResult.Reject("local_job_persist_failed");
|
||
}
|
||
}
|
||
|
||
internal 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)
|
||
=> new JobRepository(jobsDirectory).EnsureAcceptedState(jobId);
|
||
|
||
internal 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 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 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 == '_');
|
||
|
||
internal static bool IsValidInputTransfers(JsonElement transfers, Guid jobId)
|
||
{
|
||
if (transfers.ValueKind != JsonValueKind.Array
|
||
|| transfers.GetArrayLength() > 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)
|
||
=> new JobRepository(jobsDirectory).InputTransfers(job);
|
||
|
||
internal string InputPath(RecoverableJob job, JsonElement transfer)
|
||
=> new JobRepository(jobsDirectory).InputPath(job, transfer);
|
||
|
||
internal void WriteTerminal(
|
||
RecoverableJob job, string status, string code, string detail, bool overwrite = false)
|
||
=> new JobRepository(jobsDirectory).WriteTerminal(job, status, code, detail, overwrite);
|
||
|
||
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));
|
||
}
|
||
|
||
internal 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 static bool TryReadWorkspace(
|
||
JsonElement value, out WorkspaceBinding? workspace)
|
||
{
|
||
workspace = null;
|
||
if (value.ValueKind == JsonValueKind.Null) return true;
|
||
if (value.ValueKind != JsonValueKind.Object
|
||
|| !HasOnlyProperties(value, "workspace_id", "source_job_id", "mode")
|
||
|| !TryReadGuid(value, "workspace_id", out var workspaceId)
|
||
|| !value.TryGetProperty("mode", out var modeValue)
|
||
|| modeValue.ValueKind != JsonValueKind.String)
|
||
{
|
||
return false;
|
||
}
|
||
var mode = modeValue.GetString();
|
||
if (mode is not ("new" or "continue")) return false;
|
||
Guid? sourceJobId = null;
|
||
if (value.TryGetProperty("source_job_id", out var sourceValue)
|
||
&& sourceValue.ValueKind != JsonValueKind.Null)
|
||
{
|
||
if (!Guid.TryParse(sourceValue.GetString(), out var parsed)) return false;
|
||
sourceJobId = parsed;
|
||
}
|
||
if ((mode == "new") != (sourceJobId is null)) return false;
|
||
workspace = new WorkspaceBinding(workspaceId, sourceJobId, mode);
|
||
return true;
|
||
}
|
||
}
|
||
|
||
internal sealed record RecoverableJob(
|
||
Guid JobId,
|
||
Guid LeaseId,
|
||
string RequestDigest,
|
||
string Capability,
|
||
WorkspaceBinding? Workspace,
|
||
JsonElement? InputTransfers,
|
||
JsonElement? Terminal,
|
||
bool UploadComplete,
|
||
bool CloudTerminal);
|
||
|
||
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);
|
||
}
|