From 477aef7b27ed9eda2127064ce58b78a2be6f85c8 Mon Sep 17 00:00:00 2001 From: twotalesanimation <80506065+twotalesanimation@users.noreply.github.com> Date: Thu, 1 Oct 2026 18:13:23 +0200 Subject: [PATCH] feat(worker): upload status in the job log, ProRes uploaded in the background 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 --- RenderPipeline2.md | 2 +- RenderWorker/VFXReviewWorker/ApiClient.cs | 19 ++- .../VFXReviewWorker/HighResUploader.cs | 143 ++++++++++++++++++ RenderWorker/VFXReviewWorker/JobExecutor.cs | 47 +++--- .../VFXReviewWorker/LeaseKeepAlive.cs | 66 +++++++- RenderWorker/VFXReviewWorker/Program.cs | 1 + .../VFXReviewWorker/ProgressStreamContent.cs | 42 +++++ .../VFXReviewWorker/VFXReviewWorker.csproj | 2 +- RenderWorker/VFXReviewWorker/WorkerService.cs | 6 +- .../jobs/[jobId]/attach-highres/route.ts | 38 +++++ lib/render-pipeline/preview.ts | 103 +++++++++++++ 11 files changed, 442 insertions(+), 27 deletions(-) create mode 100644 RenderWorker/VFXReviewWorker/HighResUploader.cs create mode 100644 RenderWorker/VFXReviewWorker/ProgressStreamContent.cs create mode 100644 app/api/ext/render/jobs/[jobId]/attach-highres/route.ts diff --git a/RenderPipeline2.md b/RenderPipeline2.md index 5ff24b6..b995c68 100644 --- a/RenderPipeline2.md +++ b/RenderPipeline2.md @@ -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. diff --git a/RenderWorker/VFXReviewWorker/ApiClient.cs b/RenderWorker/VFXReviewWorker/ApiClient.cs index 022ffe0..5c7c9bb 100644 --- a/RenderWorker/VFXReviewWorker/ApiClient.cs +++ b/RenderWorker/VFXReviewWorker/ApiClient.cs @@ -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 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? 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) diff --git a/RenderWorker/VFXReviewWorker/HighResUploader.cs b/RenderWorker/VFXReviewWorker/HighResUploader.cs new file mode 100644 index 0000000..9d11c70 --- /dev/null +++ b/RenderWorker/VFXReviewWorker/HighResUploader.cs @@ -0,0 +1,143 @@ +using System.Text.Json; +using Microsoft.Extensions.Logging; + +namespace VFXReviewWorker; + +/// +/// 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. +/// +public sealed class HighResUploader +{ + private const int MaxAttempts = 5; + + private readonly ApiClient _api; + private readonly ReportSpooler _spooler; + private readonly ILogger _log; + private readonly string _dir; + private readonly SemaphoreSlim _wake = new(0); + + public HighResUploader(ApiClient api, ReportSpooler spooler, ILogger 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(); + } + + /// Long-running loop; started once by the worker service. + 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(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 + { + ["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 */ } + } +} diff --git a/RenderWorker/VFXReviewWorker/JobExecutor.cs b/RenderWorker/VFXReviewWorker/JobExecutor.cs index cc0030d..549fc4c 100644 --- a/RenderWorker/VFXReviewWorker/JobExecutor.cs +++ b/RenderWorker/VFXReviewWorker/JobExecutor.cs @@ -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 _log; public JobExecutor(WorkerOptions options, ApiClient api, AerenderRunner runner, PreviewStage preview, - ValidationStage validation, DeliveryStage delivery, BlenderRunner blender, ReportSpooler spooler, ILogger log) + ValidationStage validation, DeliveryStage delivery, BlenderRunner blender, ReportSpooler spooler, + HighResUploader highRes, ILogger 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 @@ -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 { @@ -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); } diff --git a/RenderWorker/VFXReviewWorker/LeaseKeepAlive.cs b/RenderWorker/VFXReviewWorker/LeaseKeepAlive.cs index a88d0ef..1787b91 100644 --- a/RenderWorker/VFXReviewWorker/LeaseKeepAlive.cs +++ b/RenderWorker/VFXReviewWorker/LeaseKeepAlive.cs @@ -1,18 +1,35 @@ +using System.Diagnostics; +using System.Text; + namespace VFXReviewWorker; /// /// 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. /// 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 _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 }); } + /// Adds a finished step to the status shown in the app. + public void Line(string text) + { + lock (_lock) + { + _current = null; + _lines.Add($"{DateTime.Now:HH:mm:ss} {text}"); + } + _wake.Release(); + } + + /// + /// Progress callback for : keeps + /// one live line ("Uploading X: 12.0 / 34.0 MB (35%, 2.1 MB/s)") up to date. + /// + public Action 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(); } } diff --git a/RenderWorker/VFXReviewWorker/Program.cs b/RenderWorker/VFXReviewWorker/Program.cs index 2758040..dd05530 100644 --- a/RenderWorker/VFXReviewWorker/Program.cs +++ b/RenderWorker/VFXReviewWorker/Program.cs @@ -37,6 +37,7 @@ builder.Logging.AddProvider(new FileLoggerProvider()); builder.Services.AddSingleton(options); builder.Services.AddSingleton(); builder.Services.AddSingleton(); +builder.Services.AddSingleton(); builder.Services.AddSingleton(); builder.Services.AddSingleton(); builder.Services.AddSingleton(); diff --git a/RenderWorker/VFXReviewWorker/ProgressStreamContent.cs b/RenderWorker/VFXReviewWorker/ProgressStreamContent.cs new file mode 100644 index 0000000..d9c2661 --- /dev/null +++ b/RenderWorker/VFXReviewWorker/ProgressStreamContent.cs @@ -0,0 +1,42 @@ +using System.Net; + +namespace VFXReviewWorker; + +/// +/// 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). +/// +internal sealed class ProgressStreamContent : HttpContent +{ + private readonly Stream _source; + private readonly Action? _onProgress; + + public ProgressStreamContent(Stream source, Action? 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; + } +} diff --git a/RenderWorker/VFXReviewWorker/VFXReviewWorker.csproj b/RenderWorker/VFXReviewWorker/VFXReviewWorker.csproj index 8d4243a..b9295fa 100644 --- a/RenderWorker/VFXReviewWorker/VFXReviewWorker.csproj +++ b/RenderWorker/VFXReviewWorker/VFXReviewWorker.csproj @@ -15,7 +15,7 @@ every change that gets published. publish.ps1 appends the git SHA via SourceRevisionId, giving "1.1.0+feb1c7b". --> - 1.7.0 + 1.7.1 diff --git a/RenderWorker/VFXReviewWorker/WorkerService.cs b/RenderWorker/VFXReviewWorker/WorkerService.cs index d5ebcfc..bf7a93c 100644 --- a/RenderWorker/VFXReviewWorker/WorkerService.cs +++ b/RenderWorker/VFXReviewWorker/WorkerService.cs @@ -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 _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 log) + public WorkerService(WorkerOptions options, ApiClient api, JobExecutor executor, ReportSpooler spooler, + HighResUploader highRes, ILogger 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); diff --git a/app/api/ext/render/jobs/[jobId]/attach-highres/route.ts b/app/api/ext/render/jobs/[jobId]/attach-highres/route.ts new file mode 100644 index 0000000..3738ff8 --- /dev/null +++ b/app/api/ext/render/jobs/[jobId]/attach-highres/route.ts @@ -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 }); + } +} diff --git a/lib/render-pipeline/preview.ts b/lib/render-pipeline/preview.ts index 737ff9a..fdd01fa 100644 --- a/lib/render-pipeline/preview.ts +++ b/lib/render-pipeline/preview.ts @@ -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): 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");