diff --git a/RenderPipeline2.md b/RenderPipeline2.md index b995c68..ed34a41 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. 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. +- **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. Since 1.7.2, artifacts of 16 MB or more upload as S3 multipart: `artifact-presign` with `multipart: true, size` returns `uploadId`, `partSize` (8 MB minimum) and one presigned URL per part; the worker PUTs parts on `uploadConcurrency` connections (default 16), retries each part up to 4 times, then calls `artifact-complete` (E20b) to stitch them, or with `abort: true` to discard them. A single connection from the studio to nbg1 is capped near 0.1 MB/s. 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/README.md b/RenderWorker/README.md index 30460aa..42c99d3 100644 --- a/RenderWorker/README.md +++ b/RenderWorker/README.md @@ -208,6 +208,9 @@ read back from the assembly, so a machine can never report a version it is not r `pathMappings` translate the manifest's canonical UNC paths to this machine's drive mappings. `plateSearchRoots` is optional: it is only consulted when a manifest carries no timing plate (older panel versions), to find the shot's plates for the timecode check. +`uploadConcurrency` (default 16) is how many parallel connections a file of 16 MB or more +uploads over. Each connection to object storage is capped near 0.1 MB/s from the studio, so +throughput scales with it (measured 2026-10-01: 1 → 0.13 MB/s, 8 → 0.64, 24 → 1.69 MB/s). Everything tunable (poll interval, lease, stall timeout, max attempts, validation) lives in the server's SystemConfig and arrives at registration — no per-machine tuning files. diff --git a/RenderWorker/VFXReviewWorker.Tests/MultipartUploadTests.cs b/RenderWorker/VFXReviewWorker.Tests/MultipartUploadTests.cs new file mode 100644 index 0000000..b5571c4 --- /dev/null +++ b/RenderWorker/VFXReviewWorker.Tests/MultipartUploadTests.cs @@ -0,0 +1,205 @@ +using System.Collections.Concurrent; +using System.Net; +using System.Net.Sockets; +using System.Text.Json; +using Microsoft.Extensions.Logging.Abstractions; +using VFXReviewWorker; +using Xunit; + +namespace VFXReviewWorker.Tests; + +/// +/// Large artifacts go up as parallel multipart parts: from the studio a single +/// connection to object storage is capped near 0.1 MB/s, so a ProRes MOV could +/// not finish inside the 30 min request timeout. These drive the real upload +/// code against a local fake of the server + storage. +/// +public class MultipartUploadTests : IDisposable +{ + private const int PartSize = 8 * 1024 * 1024; + + private readonly HttpListener _listener = new(); + private readonly string _base; + private readonly ConcurrentDictionary _parts = new(); + private readonly ConcurrentDictionary _failuresLeft = new(); + private readonly ConcurrentBag _completions = new(); + private int _inFlight, _maxInFlight; + private bool _multipartSupported = true; + private byte[]? _singlePut; + + public MultipartUploadTests() + { + var port = FreePort(); + _base = $"http://localhost:{port}/"; + _listener.Prefixes.Add(_base); + _listener.Start(); + _ = Task.Run(Serve); + } + + private static int FreePort() + { + var l = new TcpListener(IPAddress.Loopback, 0); + l.Start(); + var port = ((IPEndPoint)l.LocalEndpoint).Port; + l.Stop(); + return port; + } + + private async Task Serve() + { + while (_listener.IsListening) + { + HttpListenerContext ctx; + try { ctx = await _listener.GetContextAsync(); } catch { return; } + _ = Task.Run(() => Handle(ctx)); + } + } + + private async Task Handle(HttpListenerContext ctx) + { + var path = ctx.Request.Url!.AbsolutePath; + using var ms = new MemoryStream(); + await ctx.Request.InputStream.CopyToAsync(ms); + var body = ms.ToArray(); + var res = ctx.Response; + + if (path.EndsWith("/artifact-presign")) + { + var req = JsonDocument.Parse(body).RootElement; + var size = req.GetProperty("size").GetInt64(); + object payload = _multipartSupported && req.GetProperty("multipart").GetBoolean() + ? new + { + key = "highres/x.mov", + uploadId = "up1", + partSize = PartSize, + partUrls = Enumerable.Range(1, (int)((size + PartSize - 1) / PartSize)) + .Select(n => $"{_base}part/{n}").ToList(), + } + : new { key = "highres/x.mov", presignedUrl = $"{_base}single" }; + await Json(res, payload); + } + else if (path.StartsWith("/part/")) + { + var n = int.Parse(path["/part/".Length..]); + var now = Interlocked.Increment(ref _inFlight); + InterlockedMax(ref _maxInFlight, now); + await Task.Delay(50); + Interlocked.Decrement(ref _inFlight); + if (_failuresLeft.TryGetValue(n, out var left) && left > 0) + { + _failuresLeft[n] = left - 1; + res.StatusCode = 500; + res.Close(); + return; + } + _parts[n] = body; + res.Headers["ETag"] = $"\"etag-{n}\""; + res.Close(); + } + else if (path == "/single") + { + _singlePut = body; + res.Close(); + } + else if (path.EndsWith("/artifact-complete")) + { + _completions.Add(System.Text.Encoding.UTF8.GetString(body)); + await Json(res, new { ok = true }); + } + else + { + res.StatusCode = 404; + res.Close(); + } + } + + private static void InterlockedMax(ref int target, int value) + { + int seen; + while ((seen = Volatile.Read(ref target)) < value && Interlocked.CompareExchange(ref target, value, seen) != seen) { } + } + + private static async Task Json(HttpListenerResponse res, object payload) + { + var bytes = JsonSerializer.SerializeToUtf8Bytes(payload, WorkerOptions.JsonOpts); + res.ContentType = "application/json"; + await res.OutputStream.WriteAsync(bytes); + res.Close(); + } + + private ApiClient Client(int concurrency = 4) => + new(new WorkerOptions { ServerUrl = _base, ApiKey = "k", UploadConcurrency = concurrency }, + NullLogger.Instance); + + private static string TempFile(int bytes) + { + var path = Path.GetTempFileName(); + var data = new byte[bytes]; + new Random(42).NextBytes(data); + File.WriteAllBytes(path, data); + return path; + } + + [Fact] + public async Task Uploads_parts_in_parallel_and_stitches_them_in_order() + { + var file = TempFile(PartSize * 5 + 12345); + long lastProgress = 0; + var key = await Client(4).UploadArtifactAsync("job1", "m1", "highres", file, "video/quicktime", + CancellationToken.None, sent => Interlocked.Exchange(ref lastProgress, sent)); + + Assert.Equal("highres/x.mov", key); + var stitched = Enumerable.Range(1, 6).SelectMany(n => _parts[n]).ToArray(); + Assert.Equal(File.ReadAllBytes(file), stitched); + Assert.True(_maxInFlight > 1, "parts should upload concurrently"); + Assert.True(_maxInFlight <= 4, "concurrency is capped by UploadConcurrency"); + Assert.Equal(new FileInfo(file).Length, Interlocked.Read(ref lastProgress)); + + var complete = JsonDocument.Parse(Assert.Single(_completions)).RootElement; + Assert.False(complete.TryGetProperty("abort", out _)); + Assert.Equal(6, complete.GetProperty("parts").GetArrayLength()); + Assert.Equal("\"etag-1\"", complete.GetProperty("parts")[0].GetProperty("etag").GetString()); + File.Delete(file); + } + + [Fact] + public async Task A_failed_part_is_retried_on_its_own() + { + var file = TempFile(PartSize * 3); + _failuresLeft[2] = 2; + var key = await Client().UploadArtifactAsync("job1", "m1", "highres", file, "video/quicktime", CancellationToken.None); + + Assert.NotNull(key); + Assert.Equal(File.ReadAllBytes(file), Enumerable.Range(1, 3).SelectMany(n => _parts[n]).ToArray()); + File.Delete(file); + } + + [Fact] + public async Task A_part_that_keeps_failing_aborts_the_upload() + { + var file = TempFile(PartSize * 3); + _failuresLeft[3] = 99; + var key = await Client().UploadArtifactAsync("job1", "m1", "highres", file, "video/quicktime", CancellationToken.None); + + Assert.Null(key); + var abort = JsonDocument.Parse(Assert.Single(_completions)).RootElement; + Assert.True(abort.GetProperty("abort").GetBoolean()); + File.Delete(file); + } + + [Fact] + public async Task Falls_back_to_a_single_put_on_a_server_without_multipart() + { + _multipartSupported = false; + var file = TempFile(PartSize * 3); + var key = await Client().UploadArtifactAsync("job1", "m1", "highres", file, "video/quicktime", CancellationToken.None); + + Assert.Equal("highres/x.mov", key); + Assert.Equal(File.ReadAllBytes(file), _singlePut); + Assert.Empty(_completions); + File.Delete(file); + } + + public void Dispose() => _listener.Close(); +} diff --git a/RenderWorker/VFXReviewWorker/ApiClient.cs b/RenderWorker/VFXReviewWorker/ApiClient.cs index 5c7c9bb..26bf8c5 100644 --- a/RenderWorker/VFXReviewWorker/ApiClient.cs +++ b/RenderWorker/VFXReviewWorker/ApiClient.cs @@ -13,10 +13,16 @@ namespace VFXReviewWorker; public sealed class ApiClient { private readonly HttpClient _http; + private readonly WorkerOptions _options; private readonly ILogger _log; + // Presigned uploads: a separate client so the URL is not sent with our + // Authorization header, and a generous per-request timeout for media. + private static readonly HttpClient _upload = new() { Timeout = TimeSpan.FromMinutes(30) }; + public ApiClient(WorkerOptions options, ILogger log) { + _options = options; _log = log; _http = new HttpClient { @@ -170,6 +176,8 @@ public sealed class ApiClient kind, fileName = Path.GetFileName(filePath), contentType, + multipart = size >= MultipartThreshold, + size, }, Json, ct); if (!presignRes.IsSuccessStatusCode) { @@ -179,17 +187,23 @@ public sealed class ApiClient var presign = await presignRes.Content.ReadFromJsonAsync(Json, ct); if (presign is null) return null; - // Media files can be large — generous timeout, separate client so the - // 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 ProgressStreamContent(stream, onProgress); - content.Headers.ContentType = new MediaTypeHeaderValue(contentType); - var putRes = await upload.PutAsync(presign.PresignedUrl, content, ct); - if (!putRes.IsSuccessStatusCode) + if (presign.UploadId is not null && presign.PartUrls is { Count: > 0 } && presign.PartSize > 0) { - _log.LogWarning("Upload of {Kind} failed after {Sec:F0}s: {Status}", kind, sw.Elapsed.TotalSeconds, (int)putRes.StatusCode); - return null; + if (!await UploadMultipartAsync(jobId, machineId, kind, filePath, size, presign, onProgress, ct)) + return null; + } + else + { + // Single PUT: small files, or a server that predates multipart. + await using var stream = File.OpenRead(filePath); + 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 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)", @@ -203,6 +217,117 @@ public sealed class ApiClient } } + // Files at least this big go up as parallel multipart parts (E20b). + private const long MultipartThreshold = 16L * 1024 * 1024; + private const int PartAttempts = 4; + + /// + /// Uploads the parts concurrently (UploadConcurrency connections), each + /// retried on its own so a dropped connection costs one part, not the + /// file; then asks the server to stitch them. Aborts on failure so the + /// stored parts don't linger in the bucket. + /// + private async Task UploadMultipartAsync(string jobId, string machineId, string kind, string filePath, + long size, PresignResponse presign, Action? onProgress, CancellationToken ct) + { + var partUrls = presign.PartUrls!; + var partSize = presign.PartSize; + var etags = new string?[partUrls.Count]; + var sentPerPart = new long[partUrls.Count]; + long sentTotal = 0; + var next = -1; + + using var handle = File.OpenHandle(filePath, FileMode.Open, FileAccess.Read, FileShare.Read, FileOptions.Asynchronous); + using var failed = CancellationTokenSource.CreateLinkedTokenSource(ct); + + async Task Worker() + { + int i; + while ((i = Interlocked.Increment(ref next)) < partUrls.Count && !failed.IsCancellationRequested) + { + var offset = i * partSize; + var len = (int)Math.Min(partSize, size - offset); + var buffer = new byte[len]; + var read = 0; + while (read < len) read += await RandomAccess.ReadAsync(handle, buffer.AsMemory(read), offset + read, failed.Token); + + for (var attempt = 1; ; attempt++) + { + var part = i; + // Progress restarts from zero for a retried part. + Interlocked.Add(ref sentTotal, -Interlocked.Exchange(ref sentPerPart[part], 0)); + var content = new ProgressStreamContent(new MemoryStream(buffer, writable: false), sent => + { + var delta = sent - Interlocked.Exchange(ref sentPerPart[part], sent); + onProgress?.Invoke(Interlocked.Add(ref sentTotal, delta)); + }); + try + { + var res = await _upload.PutAsync(partUrls[i], content, failed.Token); + var etag = res.Headers.ETag?.Tag + ?? (res.Headers.TryGetValues("ETag", out var v) ? v.FirstOrDefault() : null); + if (res.IsSuccessStatusCode && etag is not null) + { + etags[i] = etag; + break; + } + _log.LogWarning("Part {Part}/{Count} of {Kind} for job {JobId} failed: {Status} (attempt {N}/{Max})", + i + 1, partUrls.Count, kind, jobId, (int)res.StatusCode, attempt, PartAttempts); + } + catch (Exception ex) when (!failed.IsCancellationRequested) + { + _log.LogWarning("Part {Part}/{Count} of {Kind} for job {JobId} failed: {Error} (attempt {N}/{Max})", + i + 1, partUrls.Count, kind, jobId, ex.Message, attempt, PartAttempts); + } + if (attempt >= PartAttempts) + { + failed.Cancel(); // no point finishing the other parts + return; + } + await Task.Delay(TimeSpan.FromSeconds(5 * attempt), failed.Token); + } + } + } + + var concurrency = Math.Clamp(_options.UploadConcurrency, 1, 64); + try + { + await Task.WhenAll(Enumerable.Range(0, Math.Min(concurrency, partUrls.Count)).Select(_ => Task.Run(Worker))); + } + catch (OperationCanceledException) + { + // a part gave up, or shutdown — either way the upload is aborted below + } + + var complete = !failed.IsCancellationRequested && etags.All(e => e is not null); + if (complete) + { + var res = await _http.PostAsJsonAsync($"api/ext/render/jobs/{jobId}/artifact-complete", new + { + machineId, + key = presign.Key, + uploadId = presign.UploadId, + parts = etags.Select((etag, i) => new { partNumber = i + 1, etag }), + }, Json, ct); + if (res.IsSuccessStatusCode) return true; + _log.LogWarning("Completing multipart {Kind} for job {JobId} failed: {Status} {Body}", + kind, jobId, (int)res.StatusCode, await res.Content.ReadAsStringAsync(ct)); + } + + try + { + await _http.PostAsJsonAsync($"api/ext/render/jobs/{jobId}/artifact-complete", new + { + machineId, + key = presign.Key, + uploadId = presign.UploadId, + abort = true, + }, Json, CancellationToken.None); + } + catch { /* parts expire on their own */ } + return false; + } + public Task UploadLogAsync(string jobId, string machineId, string logFilePath, CancellationToken ct) => UploadArtifactAsync(jobId, machineId, "log", logFilePath, "text/plain", ct); diff --git a/RenderWorker/VFXReviewWorker/Dtos.cs b/RenderWorker/VFXReviewWorker/Dtos.cs index 8974ab1..fd2e574 100644 --- a/RenderWorker/VFXReviewWorker/Dtos.cs +++ b/RenderWorker/VFXReviewWorker/Dtos.cs @@ -304,6 +304,11 @@ public sealed class PresignResponse public string PresignedUrl { get; set; } = ""; public string Key { get; set; } = ""; public string Url { get; set; } = ""; + // Multipart (server ≥ 1.7.2 contract) — absent from older servers, which + // only ever return PresignedUrl. + public string? UploadId { get; set; } + public long PartSize { get; set; } + public List? PartUrls { get; set; } } public sealed class ExportDetailResponse diff --git a/RenderWorker/VFXReviewWorker/VFXReviewWorker.csproj b/RenderWorker/VFXReviewWorker/VFXReviewWorker.csproj index b9295fa..7132811 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.1 + 1.7.2 diff --git a/RenderWorker/VFXReviewWorker/WorkerOptions.cs b/RenderWorker/VFXReviewWorker/WorkerOptions.cs index a87c611..732fad1 100644 --- a/RenderWorker/VFXReviewWorker/WorkerOptions.cs +++ b/RenderWorker/VFXReviewWorker/WorkerOptions.cs @@ -35,6 +35,12 @@ public sealed class WorkerOptions /// public List PlateSearchRoots { get; set; } = new(); /// + /// Parallel connections per large upload. From the studio each connection + /// to object storage is capped near 0.1 MB/s, so throughput scales with + /// this (measured: 1 → 0.13 MB/s, 8 → 0.64, 24 → 1.69). + /// + public int UploadConcurrency { get; set; } = 16; + /// /// Read from the assembly, not from config.json: a hand-maintained version /// string is exactly the thing that goes stale, and a machine reporting a /// version it is not running is worse than no version at all. diff --git a/app/api/ext/render/jobs/[jobId]/artifact-complete/route.ts b/app/api/ext/render/jobs/[jobId]/artifact-complete/route.ts new file mode 100644 index 0000000..9f27032 --- /dev/null +++ b/app/api/ext/render/jobs/[jobId]/artifact-complete/route.ts @@ -0,0 +1,62 @@ +import { NextRequest, NextResponse } from "next/server"; +import { isExtAuthorized } from "@/lib/render-pipeline/ext-auth"; +import { toErrorResponse, PipelineError } from "@/lib/render-pipeline/errors"; +import { isArtifactKey } from "@/lib/render-pipeline/artifact-kinds"; +import { db } from "@/lib/db"; +import { abortHetznerMultipartUpload, completeHetznerMultipartUpload } from "@/lib/storage"; + +// ── POST /api/ext/render/jobs/{jobId}/artifact-complete (E20b) ─────────────── +// +// Closes a multipart upload started by artifact-presign: stitches the parts +// (by part number + ETag) into the final object, or with `abort: true` throws +// the stored parts away after the worker gave up. + +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: { + machineId?: string; + key?: string; + uploadId?: string; + parts?: { partNumber?: number; etag?: string }[]; + abort?: boolean; + }; + try { + body = await req.json(); + } catch { + return NextResponse.json({ error: "Invalid JSON" }, { status: 400 }); + } + try { + if (!body.machineId) throw new PipelineError(400, "machineId is required"); + if (!body.key || !isArtifactKey(body.key)) throw new PipelineError(422, "key is not a worker artifact key"); + if (!body.uploadId) throw new PipelineError(400, "uploadId is required"); + + const job = await db.renderJob.findUnique({ where: { id: jobId }, select: { machineId: true } }); + if (!job) throw new PipelineError(404, "Render job not found"); + if (job.machineId !== body.machineId) { + throw new PipelineError(409, "Job is not held by this machine (lease reassigned?)"); + } + + if (body.abort) { + await abortHetznerMultipartUpload(body.key, body.uploadId); + return NextResponse.json({ ok: true, aborted: true }); + } + + const parts = (body.parts ?? []).filter( + (p): p is { partNumber: number; etag: string } => + typeof p.partNumber === "number" && typeof p.etag === "string" && p.etag.length > 0 + ); + if (parts.length === 0) throw new PipelineError(422, "parts are required"); + + await completeHetznerMultipartUpload(body.key, body.uploadId, parts); + return NextResponse.json({ ok: true, key: body.key, url: `/api/files/${body.key}` }); + } catch (err) { + const { body: errBody, status } = toErrorResponse(err); + return NextResponse.json(errBody, { status }); + } +} diff --git a/app/api/ext/render/jobs/[jobId]/artifact-presign/route.ts b/app/api/ext/render/jobs/[jobId]/artifact-presign/route.ts index fd326cd..2f62e05 100644 --- a/app/api/ext/render/jobs/[jobId]/artifact-presign/route.ts +++ b/app/api/ext/render/jobs/[jobId]/artifact-presign/route.ts @@ -3,7 +3,12 @@ import { randomUUID } from "crypto"; import { isExtAuthorized } from "@/lib/render-pipeline/ext-auth"; import { toErrorResponse, PipelineError } from "@/lib/render-pipeline/errors"; import { db } from "@/lib/db"; -import { generateHetznerPresignedUploadUrl, sanitizeFileName } from "@/lib/storage"; +import { ARTIFACT_KIND_FOLDERS } from "@/lib/render-pipeline/artifact-kinds"; +import { + createHetznerMultipartUpload, + generateHetznerPresignedUploadUrl, + sanitizeFileName, +} from "@/lib/storage"; // ── POST /api/ext/render/jobs/{jobId}/artifact-presign (E20) ───────────────── // @@ -11,14 +16,16 @@ import { generateHetznerPresignedUploadUrl, sanitizeFileName } from "@/lib/stora // storage folder layout: preview → videos/, thumbnail → image/, // metadata/log → renders/, highres (the delivery MOV) → highres/ beside the // manually uploaded High Res files. +// +// `multipart: true` + `size` (worker ≥ 1.7.2) returns an S3 multipart upload +// instead — an uploadId and one presigned URL per part — which the worker +// fills in parallel and closes with artifact-complete (E20b). A single PUT +// from the studio is capped near 0.1 MB/s; parallel parts scale with the +// number of connections. Older workers never ask and get the single PUT. -const KIND_FOLDERS: Record = { - preview: "videos", - thumbnail: "image", - metadata: "renders", - log: "renders", - highres: "highres", -}; +const KIND_FOLDERS = ARTIFACT_KIND_FOLDERS; +const MIN_PART = 8 * 1024 * 1024; +const MAX_PARTS = 10000; // S3 limit export async function POST( req: NextRequest, @@ -28,7 +35,14 @@ export async function POST( return NextResponse.json({ error: "Unauthorized" }, { status: 401 }); } const { jobId } = await params; - let body: { machineId?: string; kind?: string; fileName?: string; contentType?: string }; + let body: { + machineId?: string; + kind?: string; + fileName?: string; + contentType?: string; + multipart?: boolean; + size?: number; + }; try { body = await req.json(); } catch { @@ -50,6 +64,14 @@ export async function POST( } const key = `${folder}/${randomUUID()}-${sanitizeFileName(body.fileName)}`; + + if (body.multipart && typeof body.size === "number" && body.size > MIN_PART) { + const partSize = Math.max(MIN_PART, Math.ceil(body.size / MAX_PARTS)); + const partCount = Math.ceil(body.size / partSize); + const { uploadId, partUrls } = await createHetznerMultipartUpload(key, body.contentType, partCount); + return NextResponse.json({ key, url: `/api/files/${key}`, uploadId, partSize, partUrls }); + } + const presignedUrl = await generateHetznerPresignedUploadUrl(key, body.contentType); return NextResponse.json({ presignedUrl, key, url: `/api/files/${key}` }); } catch (err) { diff --git a/lib/render-pipeline/artifact-kinds.ts b/lib/render-pipeline/artifact-kinds.ts new file mode 100644 index 0000000..3855365 --- /dev/null +++ b/lib/render-pipeline/artifact-kinds.ts @@ -0,0 +1,18 @@ +/** + * Worker artifact kinds (E20) and the object-storage folder each lands in: + * preview → videos/, thumbnail → image/, metadata/log → renders/, highres (the + * delivery MOV) → highres/ beside the manually uploaded High Res files. + */ +export const ARTIFACT_KIND_FOLDERS: Record = { + preview: "videos", + thumbnail: "image", + metadata: "renders", + log: "renders", + highres: "highres", +}; + +/** True for keys artifact-presign could have issued — guards E20b. */ +export function isArtifactKey(key: string): boolean { + const folder = key.split("/")[0]; + return key.includes("/") && !key.includes("..") && Object.values(ARTIFACT_KIND_FOLDERS).includes(folder); +} diff --git a/lib/storage.ts b/lib/storage.ts index 9a92ba9..3082585 100644 --- a/lib/storage.ts +++ b/lib/storage.ts @@ -17,6 +17,10 @@ import { GetObjectCommand, ListObjectsV2Command, HeadObjectCommand, + CreateMultipartUploadCommand, + UploadPartCommand, + CompleteMultipartUploadCommand, + AbortMultipartUploadCommand, } from "@aws-sdk/client-s3"; import { getSignedUrl } from "@aws-sdk/s3-request-presigner"; import fs from "fs"; @@ -323,6 +327,65 @@ export async function generateHetznerPresignedUploadUrl( return getSignedUrl(client, command, { expiresIn }); } +/** + * Multipart upload with presigned part URLs, for the render worker's large + * files. A single PUT from the studio to nbg1 is capped near 0.1 MB/s per + * connection (188 ms away, lossy); parts uploaded in parallel scale almost + * linearly with the number of connections. + * + * Part URLs live 6 h: parts queue behind each other, and on a slow uplink the + * last ones can start long after the upload began. + */ +export async function createHetznerMultipartUpload( + key: string, + contentType: string, + partCount: number, + expiresIn: number = 6 * 3600 +): Promise<{ uploadId: string; partUrls: string[] }> { + const { client, bucket } = await buildHetznerClient(); + const created = await client.send( + new CreateMultipartUploadCommand({ Bucket: bucket, Key: key, ContentType: contentType }) + ); + if (!created.UploadId) throw new Error("Storage did not return an UploadId"); + const uploadId = created.UploadId; + const partUrls = await Promise.all( + Array.from({ length: partCount }, (_, i) => + getSignedUrl( + client, + new UploadPartCommand({ Bucket: bucket, Key: key, UploadId: uploadId, PartNumber: i + 1 }), + { expiresIn } + ) + ) + ); + return { uploadId, partUrls }; +} + +export async function completeHetznerMultipartUpload( + key: string, + uploadId: string, + parts: { partNumber: number; etag: string }[] +): Promise { + const { client, bucket } = await buildHetznerClient(); + await client.send( + new CompleteMultipartUploadCommand({ + Bucket: bucket, + Key: key, + UploadId: uploadId, + MultipartUpload: { + Parts: [...parts] + .sort((a, b) => a.partNumber - b.partNumber) + .map((p) => ({ PartNumber: p.partNumber, ETag: p.etag })), + }, + }) + ); +} + +/** Frees the parts already stored; otherwise they are billed until expiry. */ +export async function abortHetznerMultipartUpload(key: string, uploadId: string): Promise { + const { client, bucket } = await buildHetznerClient(); + await client.send(new AbortMultipartUploadCommand({ Bucket: bucket, Key: key, UploadId: uploadId })); +} + /** * In-process cache for Hetzner key existence checks. * Positive hits (key exists) are cached indefinitely for the process lifetime