feat(worker): upload status in the job log, ProRes uploaded in the background
Deploy / deploy (push) Successful in 3m25s
Deploy / deploy (push) Successful in 3m25s
Preview jobs looked stuck after aerender finished: the worker was silently
uploading the review MP4s and the ProRes MOV over a slow uplink, and a
restart mid-upload threw the preview away.
- Worker 1.7.1 replaces the job's log tail with live upload status and logs
each upload's size, duration and speed.
- The ProRes MOV now uploads after finalize via a persistent queue
(HighResUploader, resumes after restart, 5 attempts) and attaches through
the new E13b endpoint POST /api/ext/render/jobs/{id}/attach-highres.
A MOV from a superseded preview is discarded.
Deploy the server before the 1.7.1 worker.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
+1
-1
@@ -1194,7 +1194,7 @@ Renders run past the cut at each end (8 + 8 by house convention). Handles ship w
|
||||
The Standard review MP4 is AE's H.264 (UHD, ~35 Mbps). ProRes itself can't be the player's source — only Safari on a Mac decodes it in a web page, and UHD ProRes LT is ~280 Mbps — so the worker (≥ 1.7.0, when `preview.highQuality` is on) adds two things after the AE render, both **best-effort** (a failure costs the extras, never the preview):
|
||||
|
||||
- **High quality MP4:** ffmpeg re-encodes the delivery MOV as full-res H.264 High, CRF 14 capped at 100 Mbps, yuv420p, faststart → `Version.hqFileUrl`. The player shows an **HQ** switch when present; switching keeps frame, play state, speed and zoom, is remembered per browser (`review:quality`), and falls back to Standard if the file won't load. Same comp as the Standard MP4, so slate and handles line up frame for frame.
|
||||
- **ProRes download:** the MOV itself is uploaded (presign kind `highres`) → `Version.highResKey/highResFilename/highResSize`. It is the version's **High Res** download in the client portal (named with the client code, `clientHighResFilename`) and a **ProRes** button on the staff review page. The shot's manually uploaded High Res remains the fallback for versions without one.
|
||||
- **ProRes download:** the MOV itself is uploaded (presign kind `highres`) → `Version.highResKey/highResFilename/highResSize`. It is the version's **High Res** download in the client portal (named with the client code, `clientHighResFilename`) and a **ProRes** button on the staff review page. The shot's manually uploaded High Res remains the fallback for versions without one. Since worker 1.7.1 the MOV uploads **in the background after finalize** (`HighResUploader`, queue in `%ProgramData%\VFXReviewWorker\highres-queue`, resumes after a restart, 5 attempts) and attaches via `POST /api/ext/render/jobs/{id}/attach-highres` (E13b); a MOV from a preview that a newer preview has superseded is discarded. While a preview job uploads, the worker replaces the job's log tail with live upload status.
|
||||
|
||||
A lease keep-alive runs during the encode and uploads (they can outlast the 300 s lease). The export timeline records what was attached.
|
||||
|
||||
|
||||
@@ -124,6 +124,7 @@ public sealed class ApiClient
|
||||
"validation" => $"api/ext/render/jobs/{jobId}/validation",
|
||||
"delivery" => $"api/ext/render/jobs/{jobId}/delivery",
|
||||
"finalize" => $"api/ext/render/jobs/{jobId}/finalize",
|
||||
"attach-highres" => $"api/ext/render/jobs/{jobId}/attach-highres",
|
||||
_ => throw new ArgumentOutOfRangeException(nameof(kind), kind, "unknown lifecycle report kind"),
|
||||
};
|
||||
var res = await _http.PostAsJsonAsync(path, payload, Json, ct);
|
||||
@@ -143,8 +144,10 @@ public sealed class ApiClient
|
||||
}
|
||||
|
||||
// E20 + PUT upload. Returns the storage key, or null when the upload failed.
|
||||
// onProgress receives the bytes sent so far, for status lines in the app.
|
||||
public async Task<string?> UploadArtifactAsync(
|
||||
string jobId, string machineId, string kind, string filePath, string contentType, CancellationToken ct)
|
||||
string jobId, string machineId, string kind, string filePath, string contentType, CancellationToken ct,
|
||||
Action<long>? onProgress = null)
|
||||
{
|
||||
try
|
||||
{
|
||||
@@ -153,6 +156,13 @@ public sealed class ApiClient
|
||||
_log.LogWarning("Artifact {Kind} not found at {Path}", kind, filePath);
|
||||
return null;
|
||||
}
|
||||
var size = new FileInfo(filePath).Length;
|
||||
var sw = System.Diagnostics.Stopwatch.StartNew();
|
||||
// Uploads can run for many minutes on a slow uplink; without these two
|
||||
// lines the log goes silent and a busy worker looks hung.
|
||||
if (kind != "log")
|
||||
_log.LogInformation("Uploading {Kind} for job {JobId}: {File} ({MB:F1} MB)",
|
||||
kind, jobId, Path.GetFileName(filePath), size / 1048576.0);
|
||||
|
||||
var presignRes = await _http.PostAsJsonAsync($"api/ext/render/jobs/{jobId}/artifact-presign", new
|
||||
{
|
||||
@@ -173,14 +183,17 @@ public sealed class ApiClient
|
||||
// presigned URL is not sent with our Authorization header.
|
||||
using var upload = new HttpClient { Timeout = TimeSpan.FromMinutes(30) };
|
||||
await using var stream = File.OpenRead(filePath);
|
||||
var content = new StreamContent(stream);
|
||||
var content = new ProgressStreamContent(stream, onProgress);
|
||||
content.Headers.ContentType = new MediaTypeHeaderValue(contentType);
|
||||
var putRes = await upload.PutAsync(presign.PresignedUrl, content, ct);
|
||||
if (!putRes.IsSuccessStatusCode)
|
||||
{
|
||||
_log.LogWarning("Upload of {Kind} failed: {Status}", kind, (int)putRes.StatusCode);
|
||||
_log.LogWarning("Upload of {Kind} failed after {Sec:F0}s: {Status}", kind, sw.Elapsed.TotalSeconds, (int)putRes.StatusCode);
|
||||
return null;
|
||||
}
|
||||
if (kind != "log")
|
||||
_log.LogInformation("Uploaded {Kind} for job {JobId} in {Sec:F0}s ({Speed:F1} MB/s)",
|
||||
kind, jobId, sw.Elapsed.TotalSeconds, size / 1048576.0 / Math.Max(0.001, sw.Elapsed.TotalSeconds));
|
||||
return presign.Key;
|
||||
}
|
||||
catch (Exception ex)
|
||||
|
||||
@@ -0,0 +1,143 @@
|
||||
using System.Text.Json;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace VFXReviewWorker;
|
||||
|
||||
/// <summary>
|
||||
/// Uploads each preview's ProRes MOV (the version's High Res download) in the
|
||||
/// background, after the job has already been finalized.
|
||||
///
|
||||
/// The MOV is by far the largest upload — hundreds of MB — and on a slow or
|
||||
/// flaky uplink it held every preview job open for 10+ minutes, during which
|
||||
/// the job looked stuck in the app and a worker restart threw the whole
|
||||
/// preview away. Now the job completes as soon as the review MP4s are up, the
|
||||
/// worker moves on to the next job, and the MOV follows; an "attach-highres"
|
||||
/// report (through the ordered spooler, so always after its finalize) attaches
|
||||
/// it to the version.
|
||||
///
|
||||
/// Pending uploads are files in highres-queue/, so a restart resumes them. The
|
||||
/// MOV itself lives on the SAN beside the EXRs, never in scratch.
|
||||
/// </summary>
|
||||
public sealed class HighResUploader
|
||||
{
|
||||
private const int MaxAttempts = 5;
|
||||
|
||||
private readonly ApiClient _api;
|
||||
private readonly ReportSpooler _spooler;
|
||||
private readonly ILogger<HighResUploader> _log;
|
||||
private readonly string _dir;
|
||||
private readonly SemaphoreSlim _wake = new(0);
|
||||
|
||||
public HighResUploader(ApiClient api, ReportSpooler spooler, ILogger<HighResUploader> log)
|
||||
{
|
||||
_api = api;
|
||||
_spooler = spooler;
|
||||
_log = log;
|
||||
_dir = Path.Combine(WorkerOptions.DataDir, "highres-queue");
|
||||
Directory.CreateDirectory(_dir);
|
||||
}
|
||||
|
||||
private sealed class Pending
|
||||
{
|
||||
public string JobId { get; set; } = "";
|
||||
public string MachineId { get; set; } = "";
|
||||
public string MovPath { get; set; } = "";
|
||||
public int Attempts { get; set; }
|
||||
}
|
||||
|
||||
public void Enqueue(string jobId, string machineId, string movPath)
|
||||
{
|
||||
var item = new Pending { JobId = jobId, MachineId = machineId, MovPath = movPath };
|
||||
var name = $"{DateTimeOffset.UtcNow.UtcTicks:D20}_{jobId}.json";
|
||||
var tmp = Path.Combine(_dir, name + ".tmp");
|
||||
File.WriteAllText(tmp, JsonSerializer.Serialize(item, WorkerOptions.JsonOpts));
|
||||
File.Move(tmp, Path.Combine(_dir, name));
|
||||
_log.LogInformation("ProRes for job {JobId} queued for background upload ({MB:F0} MB)",
|
||||
jobId, File.Exists(movPath) ? new FileInfo(movPath).Length / 1048576.0 : 0);
|
||||
_wake.Release();
|
||||
}
|
||||
|
||||
/// <summary>Long-running loop; started once by the worker service.</summary>
|
||||
public async Task RunAsync(CancellationToken ct)
|
||||
{
|
||||
var backoff = TimeSpan.FromSeconds(30);
|
||||
while (!ct.IsCancellationRequested)
|
||||
{
|
||||
var next = Directory.EnumerateFiles(_dir, "*.json").OrderBy(f => f, StringComparer.Ordinal).FirstOrDefault();
|
||||
if (next is null)
|
||||
{
|
||||
try { await _wake.WaitAsync(TimeSpan.FromSeconds(30), ct); } catch (OperationCanceledException) { break; }
|
||||
continue;
|
||||
}
|
||||
|
||||
Pending? item;
|
||||
try { item = JsonSerializer.Deserialize<Pending>(File.ReadAllText(next), WorkerOptions.JsonOpts); }
|
||||
catch { item = null; }
|
||||
if (item is null || string.IsNullOrEmpty(item.JobId))
|
||||
{
|
||||
_log.LogWarning("Discarding unreadable ProRes queue file {File}", next);
|
||||
TryDelete(next);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (!File.Exists(item.MovPath))
|
||||
{
|
||||
_log.LogWarning("ProRes for job {JobId} is gone ({Path}) — not uploading", item.JobId, item.MovPath);
|
||||
Report(item, null, $"MOV not found at {item.MovPath}");
|
||||
TryDelete(next);
|
||||
continue;
|
||||
}
|
||||
|
||||
string? key;
|
||||
try
|
||||
{
|
||||
key = await _api.UploadArtifactAsync(item.JobId, item.MachineId, "highres", item.MovPath, "video/quicktime", ct);
|
||||
}
|
||||
catch (OperationCanceledException) when (ct.IsCancellationRequested)
|
||||
{
|
||||
_log.LogInformation("Worker stopping — ProRes upload for job {JobId} resumes on next start", item.JobId);
|
||||
break;
|
||||
}
|
||||
|
||||
if (key is not null)
|
||||
{
|
||||
Report(item, key, null);
|
||||
TryDelete(next);
|
||||
backoff = TimeSpan.FromSeconds(30);
|
||||
continue;
|
||||
}
|
||||
|
||||
item.Attempts++;
|
||||
if (item.Attempts >= MaxAttempts)
|
||||
{
|
||||
_log.LogWarning("ProRes upload for job {JobId} failed {N} times — giving up", item.JobId, item.Attempts);
|
||||
Report(item, null, $"upload failed {item.Attempts} times (see the worker log)");
|
||||
TryDelete(next);
|
||||
continue;
|
||||
}
|
||||
|
||||
_log.LogWarning("ProRes upload for job {JobId} failed (attempt {N}/{Max}); retrying in {Sec}s",
|
||||
item.JobId, item.Attempts, MaxAttempts, backoff.TotalSeconds);
|
||||
try { File.WriteAllText(next, JsonSerializer.Serialize(item, WorkerOptions.JsonOpts)); } catch { /* retried anyway */ }
|
||||
try { await Task.Delay(backoff, ct); } catch (OperationCanceledException) { break; }
|
||||
backoff = TimeSpan.FromSeconds(Math.Min(backoff.TotalSeconds * 2, 600));
|
||||
}
|
||||
}
|
||||
|
||||
private void Report(Pending item, string? key, string? error)
|
||||
{
|
||||
_spooler.Enqueue("attach-highres", item.JobId, new Dictionary<string, object?>
|
||||
{
|
||||
["machineId"] = item.MachineId,
|
||||
["highResKey"] = key,
|
||||
["highResFileName"] = key is null ? null : Path.GetFileName(item.MovPath),
|
||||
["highResSize"] = key is null ? null : new FileInfo(item.MovPath).Length,
|
||||
["error"] = error,
|
||||
});
|
||||
}
|
||||
|
||||
private static void TryDelete(string path)
|
||||
{
|
||||
try { File.Delete(path); } catch { /* next pass */ }
|
||||
}
|
||||
}
|
||||
@@ -18,11 +18,14 @@ public sealed class JobExecutor
|
||||
private readonly DeliveryStage _delivery;
|
||||
private readonly BlenderRunner _blender;
|
||||
private readonly ReportSpooler _spooler;
|
||||
private readonly HighResUploader _highRes;
|
||||
private readonly ILogger<JobExecutor> _log;
|
||||
|
||||
public JobExecutor(WorkerOptions options, ApiClient api, AerenderRunner runner, PreviewStage preview,
|
||||
ValidationStage validation, DeliveryStage delivery, BlenderRunner blender, ReportSpooler spooler, ILogger<JobExecutor> log)
|
||||
ValidationStage validation, DeliveryStage delivery, BlenderRunner blender, ReportSpooler spooler,
|
||||
HighResUploader highRes, ILogger<JobExecutor> log)
|
||||
{
|
||||
_highRes = highRes;
|
||||
_options = options;
|
||||
_api = api;
|
||||
_runner = runner;
|
||||
@@ -418,8 +421,15 @@ public sealed class JobExecutor
|
||||
return; // temp AEP is deliberately kept for debugging
|
||||
}
|
||||
|
||||
// From here on aerender is done and the app would keep showing its
|
||||
// last frame line; this reports what the worker is doing instead,
|
||||
// and renews the lease through the encode and uploads.
|
||||
await using var status = new LeaseKeepAlive(_api, job.Id, machineId, TimeSpan.FromSeconds(10), shutdown);
|
||||
status.Line($"Preview rendered in {result.Seconds:F0}s — uploading review media");
|
||||
|
||||
var previewKey = await _api.UploadArtifactAsync(
|
||||
job.Id, machineId, "preview", result.Mp4Path!, "video/mp4", shutdown);
|
||||
job.Id, machineId, "preview", result.Mp4Path!, "video/mp4", shutdown,
|
||||
status.Track("preview MP4", result.Mp4Path!));
|
||||
if (previewKey is null)
|
||||
{
|
||||
// The renders are on the SAN and fine — only the upload failed,
|
||||
@@ -433,6 +443,7 @@ public sealed class JobExecutor
|
||||
});
|
||||
return;
|
||||
}
|
||||
status.Line("Preview MP4 uploaded");
|
||||
|
||||
string? thumbKey = null;
|
||||
if (result.ThumbnailPath is not null)
|
||||
@@ -443,28 +454,26 @@ public sealed class JobExecutor
|
||||
|
||||
// High quality MP4 + the MOV as the version's High Res download
|
||||
// (§9.6). Both optional: a failure here costs the extras, never the
|
||||
// preview, which is already safely uploaded.
|
||||
// preview, which is already safely uploaded. The MOV is uploaded in
|
||||
// the background after finalize (HighResUploader) — it is the big
|
||||
// one and used to hold the job open for many minutes.
|
||||
string? hqKey = null;
|
||||
string? highResKey = null;
|
||||
long? highResSize = null;
|
||||
var highResPending = false;
|
||||
if (manifest.HighQuality)
|
||||
{
|
||||
await using var keepAlive = new LeaseKeepAlive(_api, job.Id, machineId, TimeSpan.FromSeconds(60), shutdown);
|
||||
|
||||
status.Line("Encoding High quality MP4");
|
||||
var hqPath = await _preview.TryMakeHighQualityAsync(job.Id, result.MovPath!, result.Mp4Path!, shutdown);
|
||||
if (hqPath is not null)
|
||||
{
|
||||
hqKey = await _api.UploadArtifactAsync(job.Id, machineId, "preview", hqPath, "video/mp4", shutdown);
|
||||
hqKey = await _api.UploadArtifactAsync(job.Id, machineId, "preview", hqPath, "video/mp4", shutdown,
|
||||
status.Track("High quality MP4", hqPath));
|
||||
if (hqKey is null) _log.LogWarning("High quality MP4 upload failed for job {JobId}", job.Id);
|
||||
status.Line(hqKey is null ? "High quality MP4 upload FAILED (see worker log)" : "High quality MP4 uploaded");
|
||||
}
|
||||
|
||||
if (result.MovPath is not null && File.Exists(result.MovPath))
|
||||
{
|
||||
highResSize = new FileInfo(result.MovPath).Length;
|
||||
highResKey = await _api.UploadArtifactAsync(
|
||||
job.Id, machineId, "highres", result.MovPath, "video/quicktime", shutdown);
|
||||
if (highResKey is null) _log.LogWarning("ProRes upload failed for job {JobId}", job.Id);
|
||||
}
|
||||
highResPending = result.MovPath is not null && File.Exists(result.MovPath);
|
||||
if (highResPending)
|
||||
status.Line("ProRes MOV queued for background upload — completing the job now");
|
||||
}
|
||||
|
||||
_spooler.Enqueue("finalize", job.Id, new Dictionary<string, object?>
|
||||
@@ -476,9 +485,7 @@ public sealed class JobExecutor
|
||||
["previewMovKey"] = previewKey,
|
||||
["thumbnailKey"] = thumbKey,
|
||||
["hqPreviewKey"] = hqKey,
|
||||
["highResKey"] = highResKey,
|
||||
["highResFileName"] = highResKey is null ? null : Path.GetFileName(result.MovPath),
|
||||
["highResSize"] = highResKey is null ? null : highResSize,
|
||||
["highResPending"] = highResPending,
|
||||
},
|
||||
["renderStats"] = new Dictionary<string, object?>
|
||||
{
|
||||
@@ -497,6 +504,10 @@ public sealed class JobExecutor
|
||||
},
|
||||
});
|
||||
|
||||
// Enqueued after finalize: its attach report must reach the spooler
|
||||
// behind finalize, which creates the version it attaches to.
|
||||
if (highResPending) _highRes.Enqueue(job.Id, machineId, result.MovPath!);
|
||||
|
||||
// Only clean up once the outputs are safely reported.
|
||||
CleanScratch(job.Id, result.TempAep);
|
||||
}
|
||||
|
||||
@@ -1,18 +1,35 @@
|
||||
using System.Diagnostics;
|
||||
using System.Text;
|
||||
|
||||
namespace VFXReviewWorker;
|
||||
|
||||
/// <summary>
|
||||
/// Renews a job's lease with progress reports while the worker does something
|
||||
/// long that reports nothing on its own — the High quality encode and the
|
||||
/// ProRes upload can each outlast a 300 s lease, and an expired lease lets the
|
||||
/// uploads can each outlast a 300 s lease, and an expired lease lets the
|
||||
/// reaper hand the job to another machine mid-upload.
|
||||
///
|
||||
/// Each report also replaces the job's log tail with what the worker is doing
|
||||
/// now ("Uploading High quality MP4: 12.0 / 34.0 MB…"). Without it the app
|
||||
/// keeps showing aerender's last frame line and a busy job looks hung.
|
||||
/// </summary>
|
||||
internal sealed class LeaseKeepAlive : IAsyncDisposable
|
||||
{
|
||||
private readonly ApiClient _api;
|
||||
private readonly string _jobId;
|
||||
private readonly string _machineId;
|
||||
private readonly CancellationTokenSource _cts;
|
||||
private readonly SemaphoreSlim _wake = new(0);
|
||||
private readonly Task _loop;
|
||||
private readonly object _lock = new();
|
||||
private readonly List<string> _lines = new();
|
||||
private string? _current;
|
||||
|
||||
public LeaseKeepAlive(ApiClient api, string jobId, string machineId, TimeSpan interval, CancellationToken shutdown)
|
||||
{
|
||||
_api = api;
|
||||
_jobId = jobId;
|
||||
_machineId = machineId;
|
||||
_cts = CancellationTokenSource.CreateLinkedTokenSource(shutdown);
|
||||
var token = _cts.Token;
|
||||
_loop = Task.Run(async () =>
|
||||
@@ -21,10 +38,10 @@ internal sealed class LeaseKeepAlive : IAsyncDisposable
|
||||
{
|
||||
while (!token.IsCancellationRequested)
|
||||
{
|
||||
await Task.Delay(interval, token);
|
||||
// Best-effort by contract (ApiClient.ProgressAsync never throws
|
||||
// on transport errors); progress stays at 1.0 — the render is done.
|
||||
await api.ProgressAsync(jobId, machineId, 1.0, null, null, null, null, token);
|
||||
await _api.ProgressAsync(_jobId, _machineId, 1.0, null, null, null, Render(), token);
|
||||
await _wake.WaitAsync(interval, token);
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
@@ -34,10 +51,53 @@ internal sealed class LeaseKeepAlive : IAsyncDisposable
|
||||
});
|
||||
}
|
||||
|
||||
/// <summary>Adds a finished step to the status shown in the app.</summary>
|
||||
public void Line(string text)
|
||||
{
|
||||
lock (_lock)
|
||||
{
|
||||
_current = null;
|
||||
_lines.Add($"{DateTime.Now:HH:mm:ss} {text}");
|
||||
}
|
||||
_wake.Release();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Progress callback for <see cref="ApiClient.UploadArtifactAsync"/>: keeps
|
||||
/// one live line ("Uploading X: 12.0 / 34.0 MB (35%, 2.1 MB/s)") up to date.
|
||||
/// </summary>
|
||||
public Action<long> Track(string label, string filePath)
|
||||
{
|
||||
long total = File.Exists(filePath) ? new FileInfo(filePath).Length : 0;
|
||||
var sw = Stopwatch.StartNew();
|
||||
lock (_lock) _current = $"Uploading {label}: 0.0 / {Mb(total)} MB";
|
||||
_wake.Release();
|
||||
return sent =>
|
||||
{
|
||||
var pct = total > 0 ? 100.0 * sent / total : 0;
|
||||
var speed = sent / 1048576.0 / Math.Max(0.001, sw.Elapsed.TotalSeconds);
|
||||
lock (_lock) _current = $"Uploading {label}: {Mb(sent)} / {Mb(total)} MB ({pct:F0}%, {speed:F1} MB/s)";
|
||||
};
|
||||
}
|
||||
|
||||
private static string Mb(long bytes) => (bytes / 1048576.0).ToString("F1");
|
||||
|
||||
private string Render()
|
||||
{
|
||||
lock (_lock)
|
||||
{
|
||||
var sb = new StringBuilder();
|
||||
foreach (var l in _lines) sb.AppendLine(l);
|
||||
if (_current is not null) sb.AppendLine($"{DateTime.Now:HH:mm:ss} {_current}");
|
||||
return sb.ToString();
|
||||
}
|
||||
}
|
||||
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
_cts.Cancel();
|
||||
try { await _loop; } catch { /* already reported */ }
|
||||
_cts.Dispose();
|
||||
_wake.Dispose();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -37,6 +37,7 @@ builder.Logging.AddProvider(new FileLoggerProvider());
|
||||
builder.Services.AddSingleton(options);
|
||||
builder.Services.AddSingleton<ApiClient>();
|
||||
builder.Services.AddSingleton<ReportSpooler>();
|
||||
builder.Services.AddSingleton<HighResUploader>();
|
||||
builder.Services.AddSingleton<AerenderRunner>();
|
||||
builder.Services.AddSingleton<PreviewStage>();
|
||||
builder.Services.AddSingleton<ValidationStage>();
|
||||
|
||||
@@ -0,0 +1,42 @@
|
||||
using System.Net;
|
||||
|
||||
namespace VFXReviewWorker;
|
||||
|
||||
/// <summary>
|
||||
/// StreamContent that reports bytes sent, so a long upload can show progress.
|
||||
/// Length is computed from the stream, so the PUT still carries Content-Length
|
||||
/// (presigned S3 uploads reject chunked bodies).
|
||||
/// </summary>
|
||||
internal sealed class ProgressStreamContent : HttpContent
|
||||
{
|
||||
private readonly Stream _source;
|
||||
private readonly Action<long>? _onProgress;
|
||||
|
||||
public ProgressStreamContent(Stream source, Action<long>? onProgress)
|
||||
{
|
||||
_source = source;
|
||||
_onProgress = onProgress;
|
||||
}
|
||||
|
||||
protected override Task SerializeToStreamAsync(Stream stream, TransportContext? context) =>
|
||||
SerializeToStreamAsync(stream, context, CancellationToken.None);
|
||||
|
||||
protected override async Task SerializeToStreamAsync(Stream stream, TransportContext? context, CancellationToken ct)
|
||||
{
|
||||
var buffer = new byte[81920];
|
||||
long sent = 0;
|
||||
int read;
|
||||
while ((read = await _source.ReadAsync(buffer, ct)) > 0)
|
||||
{
|
||||
await stream.WriteAsync(buffer.AsMemory(0, read), ct);
|
||||
sent += read;
|
||||
_onProgress?.Invoke(sent);
|
||||
}
|
||||
}
|
||||
|
||||
protected override bool TryComputeLength(out long length)
|
||||
{
|
||||
length = _source.Length;
|
||||
return true;
|
||||
}
|
||||
}
|
||||
@@ -15,7 +15,7 @@
|
||||
every change that gets published. publish.ps1 appends the git SHA via
|
||||
SourceRevisionId, giving "1.1.0+feb1c7b".
|
||||
-->
|
||||
<Version>1.7.0</Version>
|
||||
<Version>1.7.1</Version>
|
||||
</PropertyGroup>
|
||||
|
||||
<ItemGroup>
|
||||
|
||||
@@ -15,6 +15,7 @@ public sealed class WorkerService : BackgroundService
|
||||
private readonly ApiClient _api;
|
||||
private readonly JobExecutor _executor;
|
||||
private readonly ReportSpooler _spooler;
|
||||
private readonly HighResUploader _highRes;
|
||||
private readonly ILogger<WorkerService> _log;
|
||||
|
||||
private string? _machineId;
|
||||
@@ -23,8 +24,10 @@ public sealed class WorkerService : BackgroundService
|
||||
private CancellationTokenSource? _cancelCurrentJob;
|
||||
private bool? _lastAeOpen;
|
||||
|
||||
public WorkerService(WorkerOptions options, ApiClient api, JobExecutor executor, ReportSpooler spooler, ILogger<WorkerService> log)
|
||||
public WorkerService(WorkerOptions options, ApiClient api, JobExecutor executor, ReportSpooler spooler,
|
||||
HighResUploader highRes, ILogger<WorkerService> log)
|
||||
{
|
||||
_highRes = highRes;
|
||||
_options = options;
|
||||
_api = api;
|
||||
_executor = executor;
|
||||
@@ -40,6 +43,7 @@ public sealed class WorkerService : BackgroundService
|
||||
_ = Task.Run(() => _spooler.RunAsync(stoppingToken), stoppingToken);
|
||||
|
||||
await RegisterWithRetryAsync(stoppingToken);
|
||||
_ = Task.Run(() => _highRes.RunAsync(stoppingToken), stoppingToken);
|
||||
await _api.ReportVersionAsync(_options, stoppingToken);
|
||||
await _executor.ReconcileAsync(_machineId!, stoppingToken);
|
||||
|
||||
|
||||
@@ -0,0 +1,38 @@
|
||||
import { NextRequest, NextResponse } from "next/server";
|
||||
import { isExtAuthorized } from "@/lib/render-pipeline/ext-auth";
|
||||
import { toErrorResponse } from "@/lib/render-pipeline/errors";
|
||||
import { attachHighRes, type AttachHighResInput } from "@/lib/render-pipeline/preview";
|
||||
|
||||
// ── POST /api/ext/render/jobs/{jobId}/attach-highres (E13b) ──────────────────
|
||||
//
|
||||
// The preview's ProRes MOV, uploaded in the background after finalize (worker
|
||||
// ≥ 1.7.1). Attaches it to the version as its High Res download, or records on
|
||||
// the export timeline that the upload failed.
|
||||
|
||||
export async function POST(
|
||||
req: NextRequest,
|
||||
{ params }: { params: Promise<{ jobId: string }> }
|
||||
) {
|
||||
if (!isExtAuthorized(req)) {
|
||||
return NextResponse.json({ error: "Unauthorized" }, { status: 401 });
|
||||
}
|
||||
const { jobId } = await params;
|
||||
|
||||
let body: AttachHighResInput & { machineId?: string };
|
||||
try {
|
||||
body = await req.json();
|
||||
} catch {
|
||||
return NextResponse.json({ error: "Invalid JSON" }, { status: 400 });
|
||||
}
|
||||
if (!body.machineId) {
|
||||
return NextResponse.json({ error: "machineId is required" }, { status: 400 });
|
||||
}
|
||||
|
||||
try {
|
||||
const result = await attachHighRes(jobId, body.machineId, body);
|
||||
return NextResponse.json(result);
|
||||
} catch (err) {
|
||||
const { body: errBody, status } = toErrorResponse(err);
|
||||
return NextResponse.json(errBody, { status });
|
||||
}
|
||||
}
|
||||
@@ -305,6 +305,8 @@ export interface FinalizeInput {
|
||||
highResKey?: string | null;
|
||||
highResFileName?: string | null;
|
||||
highResSize?: number | null;
|
||||
/** Worker ≥ 1.7.1 uploads the MOV after finalize and attaches it via E13b. */
|
||||
highResPending?: boolean;
|
||||
};
|
||||
renderStats?: { renderSeconds?: number; previewSeconds?: number };
|
||||
media?: {
|
||||
@@ -626,12 +628,113 @@ async function finalizePreviewTx(jobId: string, machineId: string, body: Finaliz
|
||||
function highResNote(artifacts: NonNullable<FinalizeInput["artifacts"]>): string {
|
||||
const hq = !!artifacts.hqPreviewKey;
|
||||
const mov = !!artifacts.highResKey;
|
||||
if (artifacts.highResPending) {
|
||||
return hq
|
||||
? " — High quality MP4 attached; ProRes uploading in the background"
|
||||
: " — ProRes uploading in the background (High quality encode failed or skipped)";
|
||||
}
|
||||
if (hq && mov) return " — High quality MP4 and ProRes attached";
|
||||
if (hq) return " — High quality MP4 attached (ProRes upload failed or skipped)";
|
||||
if (mov) return " — ProRes attached (High quality encode failed or skipped)";
|
||||
return " — no High quality/ProRes (disabled, worker < 1.7.0, or both failed — see the worker log)";
|
||||
}
|
||||
|
||||
export interface AttachHighResInput {
|
||||
highResKey?: string | null;
|
||||
highResFileName?: string | null;
|
||||
highResSize?: number | null;
|
||||
/** Set when the worker gave up on the upload; recorded on the timeline. */
|
||||
error?: string | null;
|
||||
}
|
||||
|
||||
/**
|
||||
* E13b — the ProRes MOV for an already-finalized preview job (worker ≥ 1.7.1
|
||||
* uploads it in the background so a slow uplink no longer holds the job
|
||||
* open). Attaches it to the export's version as its High Res download.
|
||||
*
|
||||
* The upload can land minutes later, so it only attaches while this job is
|
||||
* still the export's newest preview: a regenerated preview has newer media
|
||||
* and its own MOV on the way, and the late file is deleted instead.
|
||||
*/
|
||||
export async function attachHighRes(jobId: string, machineId: string, body: AttachHighResInput) {
|
||||
const outcome = await db.$transaction(async (tx) => {
|
||||
const job = await tx.renderJob.findUnique({
|
||||
where: { id: jobId },
|
||||
include: { export: true },
|
||||
});
|
||||
if (!job) throw new PipelineError(404, "Render job not found");
|
||||
if (job.machineId !== machineId) {
|
||||
throw new PipelineError(409, "Job is not held by this machine (lease reassigned?)");
|
||||
}
|
||||
if (job.status !== "COMPLETED") {
|
||||
throw new PipelineError(409, `Job is ${job.status} — ProRes can only attach to a finalized preview`);
|
||||
}
|
||||
const exportRow = job.export;
|
||||
const key = body.highResKey ?? null;
|
||||
const discard = key ? [key] : [];
|
||||
|
||||
const note = async (text: string) => {
|
||||
if (!exportRow) return;
|
||||
await tx.exportEvent.create({
|
||||
data: {
|
||||
exportId: exportRow.id,
|
||||
fromStatus: exportRow.status,
|
||||
toStatus: exportRow.status,
|
||||
actorType: "WORKER",
|
||||
actorId: machineId,
|
||||
note: text,
|
||||
},
|
||||
});
|
||||
};
|
||||
|
||||
if (!exportRow?.versionId) {
|
||||
return { attached: false, reason: "no version", deleteKeys: discard };
|
||||
}
|
||||
|
||||
const newer = await tx.renderJob.findFirst({
|
||||
where: { exportId: exportRow.id, type: "PREVIEW_ONLY", createdAt: { gt: job.createdAt } },
|
||||
select: { id: true },
|
||||
});
|
||||
if (newer) {
|
||||
if (key) await note("ProRes from an earlier preview arrived after a newer preview — discarded");
|
||||
return { attached: false, reason: "superseded", deleteKeys: discard };
|
||||
}
|
||||
|
||||
if (!key) {
|
||||
await note(`ProRes background upload failed — ${body.error ?? "no file"} (Standard and High quality unaffected)`);
|
||||
return { attached: false, reason: "upload failed", deleteKeys: [] as string[] };
|
||||
}
|
||||
|
||||
const version = await tx.version.findUnique({
|
||||
where: { id: exportRow.versionId },
|
||||
select: { id: true, highResKey: true },
|
||||
});
|
||||
if (!version) return { attached: false, reason: "version gone", deleteKeys: discard };
|
||||
if (version.highResKey === key) return { attached: true, alreadyApplied: true, deleteKeys: [] as string[] };
|
||||
|
||||
await tx.version.update({
|
||||
where: { id: version.id },
|
||||
data: {
|
||||
highResKey: key,
|
||||
highResFilename: body.highResFileName ?? null,
|
||||
highResSize: typeof body.highResSize === "number" ? BigInt(Math.round(body.highResSize)) : null,
|
||||
highResPrunedAt: null,
|
||||
},
|
||||
});
|
||||
const mb = typeof body.highResSize === "number" ? ` (${Math.round(body.highResSize / 1048576)} MB)` : "";
|
||||
await note(`ProRes attached${mb}`);
|
||||
return {
|
||||
attached: true,
|
||||
deleteKeys: version.highResKey ? [version.highResKey] : ([] as string[]),
|
||||
};
|
||||
});
|
||||
|
||||
// Storage is only touched once the transaction has committed.
|
||||
const { deleteKeys, ...result } = outcome;
|
||||
await deleteStorageKeys(deleteKeys);
|
||||
return { ok: true, ...result };
|
||||
}
|
||||
|
||||
/** T11 — retry the preview stage only; validated EXRs are left alone. */
|
||||
export async function retryPreview(exportId: string, actorId?: string) {
|
||||
const maxAttempts = await getConfigNumber("render.maxAttempts");
|
||||
|
||||
Reference in New Issue
Block a user