478 lines
20 KiB
C#
478 lines
20 KiB
C#
using System.Net;
|
||
using System.Net.Http.Headers;
|
||
using System.Security.Cryptography;
|
||
using System.Text;
|
||
using System.Text.Json;
|
||
|
||
namespace Zcbot.WindowsNode;
|
||
|
||
internal sealed class JobOutputUploader(NodeConfig config)
|
||
{
|
||
private const long MaxDiagnosticLogBytes = 1024 * 1024;
|
||
private static readonly object DiagnosticLogLock = new();
|
||
private static readonly JsonSerializerOptions JsonOptions = new() { WriteIndented = true };
|
||
private static readonly TimeSpan[] SharingViolationBackoff =
|
||
[
|
||
TimeSpan.FromMilliseconds(500),
|
||
TimeSpan.FromSeconds(1),
|
||
TimeSpan.FromSeconds(2),
|
||
TimeSpan.FromSeconds(5),
|
||
];
|
||
|
||
internal async Task UploadAsync(
|
||
RecoverableJob job,
|
||
bool recovery,
|
||
string? workspaceStateFilename,
|
||
IReadOnlyList<string> previewOutputIds)
|
||
{
|
||
var jobDirectory = Path.Combine(
|
||
NodePaths.ForCurrentMachine().JobsDirectory, job.JobId.ToString("D"));
|
||
var completionPath = Path.Combine(jobDirectory, "upload-complete.json");
|
||
if (File.Exists(completionPath)) return;
|
||
var cloudTerminalPath = Path.Combine(jobDirectory, "cloud-terminal.json");
|
||
if (File.Exists(cloudTerminalPath)) return;
|
||
var terminalPath = Path.Combine(jobDirectory, "terminal.json");
|
||
using var terminal = JsonDocument.Parse(await File.ReadAllBytesAsync(terminalPath));
|
||
if (terminal.RootElement.GetProperty("status").GetString() != "succeeded") return;
|
||
var manifest = terminal.RootElement.GetProperty("artifact_manifest").Clone();
|
||
|
||
using var client = new HttpClient { BaseAddress = config.ServerUrl };
|
||
client.DefaultRequestHeaders.Authorization =
|
||
new AuthenticationHeaderValue("Bearer", config.NodeToken);
|
||
client.DefaultRequestHeaders.Add("X-Node-Id", config.NodeId.ToString());
|
||
client.DefaultRequestHeaders.Add("X-Lease-Id", job.LeaseId.ToString());
|
||
client.DefaultRequestHeaders.Add("X-Request-Digest", job.RequestDigest);
|
||
if (recovery)
|
||
{
|
||
var replay = await TryCompleteAsync(
|
||
client, jobDirectory, job, manifest, WorkspaceSize(jobDirectory), allowConflict: true);
|
||
if (File.Exists(cloudTerminalPath)) return;
|
||
if (replay is not null)
|
||
{
|
||
await WriteCompletionWithRetryAsync(job, completionPath, replay);
|
||
Log(jobDirectory, "OK",
|
||
$"Output upload reconciled job={job.JobId:D} source=cloud");
|
||
return;
|
||
}
|
||
}
|
||
foreach (var artifact in manifest.EnumerateArray())
|
||
{
|
||
var localId = artifact.GetProperty("artifact_id").GetString()!;
|
||
if (workspaceStateFilename is not null && !previewOutputIds.Contains(localId))
|
||
{
|
||
continue;
|
||
}
|
||
var filename = artifact.GetProperty("filename").GetString()!;
|
||
var expectedSize = artifact.GetProperty("size_bytes").GetInt64();
|
||
var expectedDigest = artifact.GetProperty("sha256").GetString()!;
|
||
var outputDirectory = workspaceStateFilename is null
|
||
? Path.Combine(jobDirectory, "output")
|
||
: Path.Combine(
|
||
NodePaths.ForCurrentMachine().WorkspacesDirectory,
|
||
job.Workspace!.WorkspaceId.ToString("D"),
|
||
"current");
|
||
var path = Path.Combine(outputDirectory, filename);
|
||
var info = new FileInfo(path);
|
||
if (!info.Exists || info.Length != expectedSize)
|
||
{
|
||
throw new InvalidDataException($"Output artifact is missing or changed: {localId}.");
|
||
}
|
||
Log(jobDirectory, "INFO",
|
||
$"Output upload phase=verify job={job.JobId:D} "
|
||
+ $"artifact={localId} file={filename}");
|
||
await using (var verify = await OpenOutputWithRetryAsync(
|
||
jobDirectory, job, localId, filename, path, "verify"))
|
||
{
|
||
var digest = Convert.ToHexString(
|
||
await SHA256.HashDataAsync(verify)).ToLowerInvariant();
|
||
if (digest != expectedDigest)
|
||
{
|
||
throw new InvalidDataException($"Output artifact digest changed: {localId}.");
|
||
}
|
||
}
|
||
Log(jobDirectory, "INFO",
|
||
$"Output upload phase=transfer job={job.JobId:D} "
|
||
+ $"artifact={localId} file={filename}");
|
||
await using var stream = await OpenOutputWithRetryAsync(
|
||
jobDirectory, job, localId, filename, path, "transfer");
|
||
using var content = new StreamContent(stream);
|
||
content.Headers.ContentType = new MediaTypeHeaderValue("application/octet-stream");
|
||
content.Headers.ContentLength = expectedSize;
|
||
content.Headers.Add("X-Content-SHA256", expectedDigest);
|
||
content.Headers.Add("X-Content-Length", expectedSize.ToString());
|
||
using var response = await client.PutAsync(
|
||
$"/v1/software-jobs/{job.JobId:D}/outputs/{Uri.EscapeDataString(localId)}",
|
||
content);
|
||
response.EnsureSuccessStatusCode();
|
||
}
|
||
|
||
var responseBody = await TryCompleteAsync(
|
||
client, jobDirectory, job, manifest, WorkspaceSize(jobDirectory), allowConflict: false)
|
||
?? throw new InvalidDataException("Output completion response is unavailable.");
|
||
await WriteCompletionWithRetryAsync(job, completionPath, responseBody);
|
||
Log(jobDirectory, "OK", $"Output upload completed job={job.JobId:D}");
|
||
}
|
||
|
||
internal void RecordDeferred(RecoverableJob job, Exception exception)
|
||
{
|
||
var jobDirectory = Path.Combine(
|
||
NodePaths.ForCurrentMachine().JobsDirectory, job.JobId.ToString("D"));
|
||
Log(jobDirectory, "WARN",
|
||
$"Output upload deferred job={job.JobId:D} "
|
||
+ $"exception={exception.GetType().Name} "
|
||
+ $"hresult=0x{exception.HResult:X8} detail={exception.Message}",
|
||
error: true);
|
||
}
|
||
|
||
internal async Task ExportAsync(RecoverableJob job, IReadOnlyList<string> outputIds)
|
||
{
|
||
var requested = outputIds.ToHashSet(StringComparer.Ordinal);
|
||
if (requested.Count == 0 || requested.Count != outputIds.Count)
|
||
{
|
||
throw new InvalidDataException("Export output identities are invalid.");
|
||
}
|
||
var jobDirectory = Path.Combine(
|
||
NodePaths.ForCurrentMachine().JobsDirectory, job.JobId.ToString("D"));
|
||
using var terminal = JsonDocument.Parse(
|
||
await File.ReadAllBytesAsync(Path.Combine(jobDirectory, "terminal.json")));
|
||
var selected = terminal.RootElement.GetProperty("artifact_manifest")
|
||
.EnumerateArray()
|
||
.Where(item => requested.Contains(item.GetProperty("artifact_id").GetString()!))
|
||
.Select(item => item.Clone())
|
||
.ToArray();
|
||
if (selected.Length != requested.Count)
|
||
{
|
||
throw new InvalidDataException("Requested local export is unavailable.");
|
||
}
|
||
using var client = new HttpClient { BaseAddress = config.ServerUrl };
|
||
client.DefaultRequestHeaders.Authorization =
|
||
new AuthenticationHeaderValue("Bearer", config.NodeToken);
|
||
client.DefaultRequestHeaders.Add("X-Node-Id", config.NodeId.ToString());
|
||
client.DefaultRequestHeaders.Add("X-Lease-Id", job.LeaseId.ToString());
|
||
client.DefaultRequestHeaders.Add("X-Request-Digest", job.RequestDigest);
|
||
var outputDirectory = Path.Combine(
|
||
NodePaths.ForCurrentMachine().WorkspacesDirectory,
|
||
job.Workspace!.WorkspaceId.ToString("D"),
|
||
"current");
|
||
foreach (var artifact in selected)
|
||
{
|
||
var localId = artifact.GetProperty("artifact_id").GetString()!;
|
||
var filename = artifact.GetProperty("filename").GetString()!;
|
||
var expectedSize = artifact.GetProperty("size_bytes").GetInt64();
|
||
var expectedDigest = artifact.GetProperty("sha256").GetString()!;
|
||
var path = Path.Combine(outputDirectory, filename);
|
||
var info = new FileInfo(path);
|
||
if (!info.Exists || info.Length != expectedSize)
|
||
{
|
||
throw new InvalidDataException($"Export output is missing or changed: {localId}.");
|
||
}
|
||
await using (var verify = await OpenOutputWithRetryAsync(
|
||
jobDirectory, job, localId, filename, path, "verify"))
|
||
{
|
||
var digest = Convert.ToHexString(
|
||
await SHA256.HashDataAsync(verify)).ToLowerInvariant();
|
||
if (digest != expectedDigest)
|
||
{
|
||
throw new InvalidDataException($"Export output digest changed: {localId}.");
|
||
}
|
||
}
|
||
await using var stream = await OpenOutputWithRetryAsync(
|
||
jobDirectory, job, localId, filename, path, "transfer");
|
||
using var content = new StreamContent(stream);
|
||
content.Headers.ContentType = new MediaTypeHeaderValue("application/octet-stream");
|
||
content.Headers.ContentLength = expectedSize;
|
||
content.Headers.Add("X-Content-SHA256", expectedDigest);
|
||
content.Headers.Add("X-Content-Length", expectedSize.ToString());
|
||
using var response = await client.PutAsync(
|
||
$"/v1/software-jobs/{job.JobId:D}/outputs/{Uri.EscapeDataString(localId)}",
|
||
content);
|
||
response.EnsureSuccessStatusCode();
|
||
}
|
||
using var completeContent = new StringContent(
|
||
JsonSerializer.Serialize(new { artifact_manifest = selected }),
|
||
Encoding.UTF8,
|
||
"application/json");
|
||
using var completeResponse = await client.PostAsync(
|
||
$"/v1/software-jobs/{job.JobId:D}/exports/complete", completeContent);
|
||
completeResponse.EnsureSuccessStatusCode();
|
||
Log(jobDirectory, "OK", $"On-demand export completed job={job.JobId:D}");
|
||
}
|
||
|
||
private static async Task<byte[]?> TryCompleteAsync(
|
||
HttpClient client,
|
||
string jobDirectory,
|
||
RecoverableJob job,
|
||
JsonElement manifest,
|
||
long? workspaceSizeBytes,
|
||
bool allowConflict)
|
||
{
|
||
Log(jobDirectory, "INFO",
|
||
$"Output upload phase=complete job={job.JobId:D} "
|
||
+ $"mode={(allowConflict ? "reconcile" : "publish")}");
|
||
using var completeContent = new StringContent(
|
||
JsonSerializer.Serialize(new
|
||
{
|
||
artifact_manifest = manifest,
|
||
workspace_size_bytes = workspaceSizeBytes,
|
||
}),
|
||
Encoding.UTF8,
|
||
"application/json");
|
||
using var completeResponse = await client.PostAsync(
|
||
$"/v1/software-jobs/{job.JobId:D}/outputs/complete", completeContent);
|
||
if (allowConflict && completeResponse.StatusCode == HttpStatusCode.Conflict)
|
||
{
|
||
var body = await completeResponse.Content.ReadAsByteArrayAsync();
|
||
if (TryReadCloudTerminal(body, out var status, out var detail))
|
||
{
|
||
WriteCloudTerminal(jobDirectory, status, detail);
|
||
Log(jobDirectory, "WARN",
|
||
$"Output upload stopped job={job.JobId:D} "
|
||
+ $"reason=cloud_terminal status={status}", error: true);
|
||
return null;
|
||
}
|
||
Log(jobDirectory, "INFO",
|
||
$"Output upload reconciliation pending job={job.JobId:D} "
|
||
+ "reason=cloud_not_complete");
|
||
return null;
|
||
}
|
||
completeResponse.EnsureSuccessStatusCode();
|
||
return await completeResponse.Content.ReadAsByteArrayAsync();
|
||
}
|
||
|
||
private static bool TryReadCloudTerminal(
|
||
byte[] responseBody, out string status, out string detail)
|
||
{
|
||
status = "failed";
|
||
detail = "云端任务已终止，本地生成的结果仍保留";
|
||
try
|
||
{
|
||
using var response = JsonDocument.Parse(responseBody);
|
||
if (!response.RootElement.TryGetProperty("detail", out var value)
|
||
|| value.ValueKind != JsonValueKind.Object
|
||
|| !value.TryGetProperty("code", out var code)
|
||
|| code.GetString() != "JOB_ALREADY_TERMINAL")
|
||
{
|
||
return false;
|
||
}
|
||
if (value.TryGetProperty("status", out var statusValue)
|
||
&& statusValue.GetString() is { Length: > 0 } cloudStatus)
|
||
{
|
||
status = cloudStatus;
|
||
}
|
||
if (value.TryGetProperty("message", out var messageValue)
|
||
&& messageValue.GetString() is { Length: > 0 } message)
|
||
{
|
||
detail = message;
|
||
}
|
||
return true;
|
||
}
|
||
catch (JsonException)
|
||
{
|
||
return false;
|
||
}
|
||
}
|
||
|
||
private static void WriteCloudTerminal(string jobDirectory, string status, string detail)
|
||
{
|
||
var path = Path.Combine(jobDirectory, "cloud-terminal.json");
|
||
if (File.Exists(path)) return;
|
||
var pending = path + ".pending";
|
||
var content = JsonSerializer.SerializeToUtf8Bytes(new
|
||
{
|
||
status,
|
||
detail,
|
||
recorded_at = DateTimeOffset.UtcNow,
|
||
}, JsonOptions);
|
||
File.WriteAllBytes(pending, content);
|
||
try
|
||
{
|
||
File.Move(pending, path, overwrite: false);
|
||
}
|
||
catch (IOException) when (File.Exists(path))
|
||
{
|
||
File.Delete(pending);
|
||
}
|
||
}
|
||
|
||
private static long? WorkspaceSize(string jobDirectory)
|
||
{
|
||
var path = Path.Combine(jobDirectory, "workspace-result.json");
|
||
if (!File.Exists(path)) return null;
|
||
using var document = JsonDocument.Parse(File.ReadAllBytes(path));
|
||
return document.RootElement.TryGetProperty("size_bytes", out var value)
|
||
&& value.TryGetInt64(out var size)
|
||
? size
|
||
: null;
|
||
}
|
||
|
||
private static async Task<FileStream> OpenOutputWithRetryAsync(
|
||
string jobDirectory,
|
||
RecoverableJob job,
|
||
string artifactId,
|
||
string filename,
|
||
string path,
|
||
string phase)
|
||
{
|
||
for (var attempt = 0; ; attempt++)
|
||
{
|
||
try
|
||
{
|
||
return new FileStream(
|
||
path, FileMode.Open, FileAccess.Read, FileShare.Read, 64 * 1024,
|
||
FileOptions.Asynchronous | FileOptions.SequentialScan);
|
||
}
|
||
catch (IOException exception) when (
|
||
IsSharingViolation(exception) && attempt < SharingViolationBackoff.Length)
|
||
{
|
||
var delay = SharingViolationBackoff[attempt];
|
||
Log(jobDirectory, "WARN",
|
||
$"Output file busy job={job.JobId:D} artifact={artifactId} "
|
||
+ $"file={filename} phase={phase} attempt={attempt + 1} "
|
||
+ $"retry_ms={(int)delay.TotalMilliseconds} "
|
||
+ $"hresult=0x{exception.HResult:X8}", error: true);
|
||
await Task.Delay(delay);
|
||
}
|
||
catch (IOException exception) when (IsSharingViolation(exception))
|
||
{
|
||
Log(jobDirectory, "WARN",
|
||
$"Output file busy job={job.JobId:D} artifact={artifactId} "
|
||
+ $"file={filename} phase={phase} retries=exhausted "
|
||
+ $"hresult=0x{exception.HResult:X8}", error: true);
|
||
throw new IOException(
|
||
$"输出文件 {filename} 在{PhaseLabel(phase)}阶段持续被占用"
|
||
+ $"（artifact={artifactId}, HRESULT=0x{exception.HResult:X8}）。",
|
||
exception);
|
||
}
|
||
}
|
||
}
|
||
|
||
private static async Task WriteCompletionWithRetryAsync(
|
||
RecoverableJob job,
|
||
string completionPath,
|
||
byte[] responseBody)
|
||
{
|
||
var jobDirectory = Path.GetDirectoryName(completionPath)!;
|
||
if (Directory.Exists(completionPath))
|
||
{
|
||
throw new InvalidDataException(
|
||
"本地完成标记路径被同名目录占用：upload-complete.json。");
|
||
}
|
||
if (File.Exists(completionPath))
|
||
{
|
||
return;
|
||
}
|
||
var pendingPath = completionPath + ".pending";
|
||
PreparePendingCompletion(pendingPath, responseBody);
|
||
for (var attempt = 0; ; attempt++)
|
||
{
|
||
if (File.Exists(completionPath))
|
||
{
|
||
return;
|
||
}
|
||
try
|
||
{
|
||
File.Move(pendingPath, completionPath, overwrite: false);
|
||
return;
|
||
}
|
||
catch (IOException exception) when (
|
||
IsSharingViolation(exception) && attempt < SharingViolationBackoff.Length)
|
||
{
|
||
var delay = SharingViolationBackoff[attempt];
|
||
Log(jobDirectory, "WARN",
|
||
$"Completion marker busy job={job.JobId:D} "
|
||
+ "operation=move_pending "
|
||
+ $"attempt={attempt + 1} retry_ms={(int)delay.TotalMilliseconds} "
|
||
+ $"hresult=0x{exception.HResult:X8}", error: true);
|
||
await Task.Delay(delay);
|
||
}
|
||
catch (IOException exception) when (IsSharingViolation(exception))
|
||
{
|
||
Log(jobDirectory, "WARN",
|
||
$"Completion marker busy job={job.JobId:D} "
|
||
+ "operation=move_pending retries=exhausted "
|
||
+ $"hresult=0x{exception.HResult:X8}", error: true);
|
||
throw new IOException(
|
||
"本地完成标记 upload-complete.json 在重命名阶段持续被占用"
|
||
+ $"（HRESULT=0x{exception.HResult:X8}）。",
|
||
exception);
|
||
}
|
||
}
|
||
}
|
||
|
||
private static void PreparePendingCompletion(string pendingPath, byte[] responseBody)
|
||
{
|
||
if (File.Exists(pendingPath))
|
||
{
|
||
try
|
||
{
|
||
using var pending = JsonDocument.Parse(File.ReadAllBytes(pendingPath));
|
||
return;
|
||
}
|
||
catch (JsonException)
|
||
{
|
||
}
|
||
catch (IOException exception) when (IsSharingViolation(exception))
|
||
{
|
||
// 上一轮已经完整落盘但暂时被系统程序扫描时，保留同一个 pending
|
||
// 文件；重试重命名不会再次触发“新文件关闭后立即扫描”。
|
||
return;
|
||
}
|
||
}
|
||
using var response = JsonDocument.Parse(responseBody);
|
||
var content = JsonSerializer.SerializeToUtf8Bytes(new
|
||
{
|
||
completed_at = DateTimeOffset.UtcNow,
|
||
response = response.RootElement,
|
||
}, JsonOptions);
|
||
using var stream = new FileStream(
|
||
pendingPath, FileMode.Create, FileAccess.Write, FileShare.None,
|
||
4096, FileOptions.WriteThrough);
|
||
stream.Write(content);
|
||
stream.Flush(flushToDisk: true);
|
||
}
|
||
|
||
private static bool IsSharingViolation(IOException exception) =>
|
||
(exception.HResult & 0xFFFF) is 32 or 33;
|
||
|
||
private static string PhaseLabel(string phase) => phase == "verify" ? "校验" : "上传";
|
||
|
||
private static void Log(
|
||
string jobDirectory,
|
||
string level,
|
||
string message,
|
||
bool error = false)
|
||
{
|
||
var consoleMessage = $"[{level}] {message}";
|
||
if (error)
|
||
{
|
||
Console.Error.WriteLine(consoleMessage);
|
||
}
|
||
else
|
||
{
|
||
Console.WriteLine(consoleMessage);
|
||
}
|
||
try
|
||
{
|
||
lock (DiagnosticLogLock)
|
||
{
|
||
var logsDirectory = Path.Combine(jobDirectory, "logs");
|
||
Directory.CreateDirectory(logsDirectory);
|
||
var path = Path.Combine(logsDirectory, "node-output-upload.log");
|
||
if (File.Exists(path) && new FileInfo(path).Length >= MaxDiagnosticLogBytes)
|
||
{
|
||
File.Move(path, path + ".1", overwrite: true);
|
||
}
|
||
File.AppendAllText(
|
||
path,
|
||
$"{DateTimeOffset.UtcNow:O} [{level}] {message}{Environment.NewLine}",
|
||
new UTF8Encoding(false));
|
||
}
|
||
}
|
||
catch (Exception exception) when (
|
||
exception is IOException or UnauthorizedAccessException)
|
||
{
|
||
}
|
||
}
|
||
|
||
}
|