vfxreview/RenderWorker/VFXReviewWorker/HighResUploader.cs

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 */ }
    }
}