270 lines
11 KiB
C#
270 lines
11 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)
|
||
{
|
||
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 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, allowConflict: true);
|
||
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()!;
|
||
var filename = artifact.GetProperty("filename").GetString()!;
|
||
var expectedSize = artifact.GetProperty("size_bytes").GetInt64();
|
||
var expectedDigest = artifact.GetProperty("sha256").GetString()!;
|
||
var path = Path.Combine(jobDirectory, "output", 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, 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);
|
||
}
|
||
|
||
private static async Task<byte[]?> TryCompleteAsync(
|
||
HttpClient client,
|
||
string jobDirectory,
|
||
RecoverableJob job,
|
||
JsonElement manifest,
|
||
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 }),
|
||
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)
|
||
{
|
||
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 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)
|
||
{
|
||
for (var attempt = 0; ; attempt++)
|
||
{
|
||
try
|
||
{
|
||
AtomicWrite(completionPath, responseBody);
|
||
return;
|
||
}
|
||
catch (IOException exception) when (
|
||
IsSharingViolation(exception) && attempt < SharingViolationBackoff.Length)
|
||
{
|
||
var delay = SharingViolationBackoff[attempt];
|
||
Log(Path.GetDirectoryName(completionPath)!, "WARN",
|
||
$"Completion marker busy job={job.JobId:D} "
|
||
+ $"attempt={attempt + 1} retry_ms={(int)delay.TotalMilliseconds} "
|
||
+ $"hresult=0x{exception.HResult:X8}", error: true);
|
||
await Task.Delay(delay);
|
||
}
|
||
}
|
||
}
|
||
|
||
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)
|
||
{
|
||
}
|
||
}
|
||
|
||
private static void AtomicWrite(string path, byte[] responseBody)
|
||
{
|
||
using var response = JsonDocument.Parse(responseBody);
|
||
var content = JsonSerializer.SerializeToUtf8Bytes(new
|
||
{
|
||
completed_at = DateTimeOffset.UtcNow,
|
||
response = response.RootElement,
|
||
}, JsonOptions);
|
||
var temporary = path + ".tmp-" + Guid.NewGuid().ToString("N");
|
||
try
|
||
{
|
||
using var stream = new FileStream(
|
||
temporary, FileMode.CreateNew, FileAccess.Write, FileShare.None,
|
||
4096, FileOptions.WriteThrough);
|
||
stream.Write(content);
|
||
stream.Flush(flushToDisk: true);
|
||
File.Move(temporary, path, overwrite: false);
|
||
}
|
||
finally
|
||
{
|
||
if (File.Exists(temporary)) File.Delete(temporary);
|
||
}
|
||
}
|
||
}
|