using System.Text.Json; using Microsoft.Extensions.Logging; namespace VFXReviewWorker; /// /// Executes one claimed job end-to-end: writes crash-recovery state, runs /// aerender, uploads the full log, and enqueues the durable complete/fail /// report. One job at a time per machine (§7.2). /// public sealed class JobExecutor { private readonly WorkerOptions _options; private readonly ApiClient _api; private readonly AerenderRunner _runner; private readonly PreviewStage _preview; private readonly ValidationStage _validation; private readonly DeliveryStage _delivery; private readonly BlenderRunner _blender; private readonly ReportSpooler _spooler; private readonly ILogger _log; public JobExecutor(WorkerOptions options, ApiClient api, AerenderRunner runner, PreviewStage preview, ValidationStage validation, DeliveryStage delivery, BlenderRunner blender, ReportSpooler spooler, ILogger log) { _options = options; _api = api; _runner = runner; _preview = preview; _validation = validation; _delivery = delivery; _blender = blender; _spooler = spooler; _log = log; } public async Task RunAsync(ClaimedJob job, string machineId, ServerConfig config, CancellationToken cancelJob, CancellationToken shutdown) { if (job.Type == "PREVIEW_ONLY") { await RunPreviewAsync(job, machineId, config, cancelJob, shutdown); return; } if (job.Type == "DELIVERY_BUILD") { await RunDeliveryAsync(job, machineId, cancelJob, shutdown); return; } if (job.Type == "BLENDER_RENDER") { await RunBlenderAsync(job, machineId, config, cancelJob, shutdown); return; } var manifest = job.AsRenderManifest(); var mapper = new PathMapper(_options.PathMappings); var outputDirLocal = mapper.Map(manifest.OutputDir); WriteState(new CurrentJobState { JobId = job.Id, ExportId = job.ExportId, MachineId = machineId, OutputDirLocal = outputDirLocal, ClaimedAt = DateTimeOffset.UtcNow, }); try { var result = await _runner.RunAsync(job, manifest, machineId, outputDirLocal, config, cancelJob, shutdown); var logKey = await _api.UploadLogAsync(job.Id, machineId, result.FullLogPath, shutdown); if (result.Success) { long totalBytes = 0; int fileCount = 0; // Prefix and extension both: the preview MOV/MP4 for this // version share the basename and live in this same folder, and // counting them as frames would inflate the EXR stats. var pattern = SequencePattern.FromOutputPattern(manifest.OutputPattern); try { foreach (var f in Directory.EnumerateFiles(outputDirLocal)) { if (!pattern.MatchesOutputFile(Path.GetFileName(f))) continue; fileCount++; totalBytes += new FileInfo(f).Length; } } catch { /* stats are best-effort in Phase 2; validation owns this in Phase 3 */ } if (fileCount == 0) { // aerender can exit 0 having rendered nothing (bad frame // range, template mismatch, silently swallowed error). // Until Phase 3 validation exists, an empty output dir must // never become READY_FOR_QC. _log.LogError("Job {JobId}: aerender exited 0 but no \"{Prefix}*{Ext}\" files exist in {Dir} — reporting failure", job.Id, pattern.Prefix, pattern.Extension, outputDirLocal); _spooler.Enqueue("fail", job.Id, new Dictionary { ["machineId"] = machineId, ["stage"] = "RENDER", ["exitCode"] = 0, ["errorMessage"] = "aerender exited 0 but produced no output files in " + outputDirLocal, ["logTail"] = result.LogTail, ["logFileKey"] = logKey, ["retryable"] = false, }); return; } _log.LogInformation("Job {JobId} complete: {Frames} frames in {Sec:F0}s ({Files} files, {Bytes} bytes)", job.Id, manifest.TotalFrames, result.RenderSeconds, fileCount, totalBytes); _spooler.Enqueue("complete-render", job.Id, new Dictionary { ["machineId"] = machineId, ["renderSeconds"] = (int)result.RenderSeconds, ["logFileKey"] = logKey, ["logTail"] = result.LogTail, ["exrFileCount"] = fileCount, ["exrTotalBytes"] = totalBytes, }); // Validation runs here, while we still hold the job and the // files are local. It is queued behind complete-render, and the // spooler's in-order replay is what guarantees the server has // moved the Export to VALIDATING before this lands (§8). if (config.ValidationEnabled) { await RunValidationAsync(job, manifest, machineId, outputDirLocal, config, shutdown); } } else if (result.ErrorMessage == "Cancelled") { // Ack the cancel so the server clears the CANCEL_JOB command _log.LogInformation("Job {JobId} cancelled — acknowledging", job.Id); _spooler.Enqueue("fail", job.Id, new Dictionary { ["machineId"] = machineId, ["stage"] = "RENDER", ["errorMessage"] = "Cancelled by server", ["logTail"] = result.LogTail, ["logFileKey"] = logKey, ["retryable"] = false, }); } else { _log.LogError("Job {JobId} failed (exit {Exit}, retryable {Retryable}): {Error}", job.Id, result.ExitCode, result.Retryable, result.ErrorMessage); _spooler.Enqueue("fail", job.Id, new Dictionary { ["machineId"] = machineId, ["stage"] = "RENDER", ["exitCode"] = result.ExitCode, ["errorMessage"] = result.ErrorMessage, ["logTail"] = result.LogTail, ["logFileKey"] = logKey, ["retryable"] = result.Retryable, }); } } catch (Exception ex) { _log.LogError(ex, "Unexpected executor error for job {JobId}", job.Id); _spooler.Enqueue("fail", job.Id, new Dictionary { ["machineId"] = machineId, ["stage"] = "RENDER", ["errorMessage"] = $"Worker exception: {ex.Message}", ["retryable"] = true, }); } finally { ClearState(); } } /// /// BLENDER_RENDER job: a standalone render with no Export behind it, so /// there is nothing to validate, preview or QC — it finishes at /// complete-render and that is the whole lifecycle. /// private async Task RunBlenderAsync(ClaimedJob job, string machineId, ServerConfig config, CancellationToken cancelJob, CancellationToken shutdown) { var manifest = job.AsBlenderManifest(); WriteState(new CurrentJobState { JobId = job.Id, ExportId = null, MachineId = machineId, OutputDirLocal = manifest.OutputPath ?? "", ClaimedAt = DateTimeOffset.UtcNow, }); try { var result = await _blender.RunAsync(job, manifest, machineId, config, cancelJob, shutdown); var logKey = string.IsNullOrEmpty(result.FullLogPath) ? null : await _api.UploadLogAsync(job.Id, machineId, result.FullLogPath, shutdown); if (result.Success) { _log.LogInformation("Blender job {JobId} complete: {Frames} frames in {Sec:F0}s", job.Id, manifest.TotalFrames, result.RenderSeconds); _spooler.Enqueue("complete-render", job.Id, new Dictionary { ["machineId"] = machineId, ["renderSeconds"] = (int)result.RenderSeconds, ["logFileKey"] = logKey, ["logTail"] = result.LogTail, }); return; } _spooler.Enqueue("fail", job.Id, new Dictionary { ["machineId"] = machineId, ["stage"] = "BLENDER", ["exitCode"] = result.ExitCode, ["errorMessage"] = result.ErrorMessage, ["logTail"] = result.LogTail, ["logFileKey"] = logKey, ["retryable"] = result.ErrorMessage == "Cancelled" ? false : result.Retryable, }); } catch (Exception ex) { _log.LogError(ex, "Unexpected Blender error for job {JobId}", job.Id); _spooler.Enqueue("fail", job.Id, new Dictionary { ["machineId"] = machineId, ["stage"] = "BLENDER", ["errorMessage"] = $"Worker exception: {ex.Message}", ["retryable"] = true, }); } finally { ClearState(); } } /// /// DELIVERY_BUILD job: build the dated package folder from already-QC'd /// exports. No AE involved — folders, links and verification only. /// private async Task RunDeliveryAsync(ClaimedJob job, string machineId, CancellationToken cancelJob, CancellationToken shutdown) { var manifest = job.AsDeliveryManifest(); WriteState(new CurrentJobState { JobId = job.Id, ExportId = null, MachineId = machineId, OutputDirLocal = new PathMapper(_options.PathMappings).Map(manifest.PackageDir), ClaimedAt = DateTimeOffset.UtcNow, }); try { var result = await _delivery.RunAsync(job, manifest, machineId, cancelJob, shutdown); _log.LogInformation("Delivery {Name}: {Ok} ({Verified}/{Total} shots verified)", manifest.PackageName, result.Ok ? "READY" : "FAILED", result.Items.Count(i => i.Status == "VERIFIED"), manifest.Items.Count); string? reportKey = null; if (result.ReportPath is not null) { reportKey = await _api.UploadArtifactAsync(job.Id, machineId, "report", result.ReportPath, "text/csv", shutdown); } _spooler.Enqueue("delivery", job.Id, new Dictionary { ["machineId"] = machineId, ["ok"] = result.Ok, ["rootPath"] = result.PackageDir, ["reportKey"] = reportKey, ["items"] = result.Items.Select(i => new Dictionary { ["itemId"] = i.ItemId, ["status"] = i.Status, ["exrPath"] = i.ExrPath, ["movPath"] = i.MovPath, ["fileCount"] = i.FileCount, ["totalBytes"] = i.TotalBytes, ["checksum"] = i.Checksum, ["linkMode"] = i.LinkMode, ["errorMessage"] = i.ErrorMessage, }).ToList(), ["manifest"] = new Dictionary { ["packageDir"] = result.PackageDir, ["reportPath"] = result.ReportPath, ["files"] = result.Files, }, ["errorMessage"] = result.Ok ? null : string.Join("; ", result.Items.Where(i => i.Status == "FAILED").Select(i => $"{i.ShotCode}: {i.ErrorMessage}").Take(5)), }); } catch (Exception ex) { _log.LogError(ex, "Unexpected delivery error for job {JobId}", job.Id); _spooler.Enqueue("delivery", job.Id, new Dictionary { ["machineId"] = machineId, ["ok"] = false, ["items"] = new List(), ["errorMessage"] = $"Worker exception: {ex.Message}", }); } finally { ClearState(); } } /// /// Runs the validation engine and spools the E12 report. Deliberately /// non-fatal on its own errors: a crash in the checks must not lose a /// finished render, so it is reported as a failed validation run rather /// than being allowed to escape. /// private async Task RunValidationAsync( ClaimedJob job, RenderManifest manifest, string machineId, string outputDirLocal, ServerConfig config, CancellationToken shutdown) { var started = DateTimeOffset.UtcNow; try { // Keep the lease alive: the job stays open through validation. await _api.ProgressAsync(job.Id, machineId, 1.0, manifest.FrameEnd, manifest.TotalFrames, 0, "Validating rendered sequence…", shutdown); var outcome = await Task.Run(() => _validation.Run(manifest, outputDirLocal, config), shutdown); var seconds = (DateTimeOffset.UtcNow - started).TotalSeconds; var failed = outcome.Checks.Where(c => c.Status == "FAIL").Select(c => c.CheckName).ToList(); if (failed.Count > 0) { _log.LogWarning("Job {JobId} validation FAILED in {Sec:F1}s: {Checks}", job.Id, seconds, string.Join(", ", failed)); } else { _log.LogInformation("Job {JobId} validation passed in {Sec:F1}s ({Count} checks, {Files} files)", job.Id, seconds, outcome.Checks.Count, outcome.FileCount); } _spooler.Enqueue("validation", job.Id, new Dictionary { ["machineId"] = machineId, ["checks"] = outcome.Checks, ["checksum"] = outcome.Checksum, ["checksumAlgo"] = "xxh64", ["exrFileCount"] = outcome.FileCount, ["exrTotalBytes"] = outcome.TotalBytes, ["validationSeconds"] = (int)seconds, }); } catch (Exception ex) { _log.LogError(ex, "Validation engine error for job {JobId}", job.Id); _spooler.Enqueue("validation", job.Id, new Dictionary { ["machineId"] = machineId, ["checks"] = new[] { ValidationCheck.Fail("engine_error", "validation completes", "worker exception", $"The validation engine threw: {ex.Message}"), }, }); } } /// /// PREVIEW_ONLY job: rebuild the shot around the rendered EXRs with slate /// and burn-ins, render the delivery MOV + review MP4, upload the review /// media, and finalize (which registers the internal-only Version). /// private async Task RunPreviewAsync(ClaimedJob job, string machineId, ServerConfig config, CancellationToken cancelJob, CancellationToken shutdown) { var manifest = job.AsPreviewManifest(); WriteState(new CurrentJobState { JobId = job.Id, ExportId = job.ExportId, MachineId = machineId, OutputDirLocal = new PathMapper(_options.PathMappings).Map(manifest.OutputDir), ClaimedAt = DateTimeOffset.UtcNow, }); try { var result = await _preview.RunAsync(job, manifest, machineId, config, cancelJob, shutdown); if (!result.Success) { _log.LogError("Preview job {JobId} failed (retryable {Retryable}): {Error}", job.Id, result.Retryable, result.ErrorMessage); _spooler.Enqueue("fail", job.Id, new Dictionary { ["machineId"] = machineId, ["stage"] = "PREVIEW", ["errorMessage"] = result.ErrorMessage, ["retryable"] = result.Retryable, }); return; // temp AEP is deliberately kept for debugging } var previewKey = await _api.UploadArtifactAsync( job.Id, machineId, "preview", result.Mp4Path!, "video/mp4", shutdown); if (previewKey is null) { // The renders are on the SAN and fine — only the upload failed, // so this is worth retrying without rebuilding anything. _spooler.Enqueue("fail", job.Id, new Dictionary { ["machineId"] = machineId, ["stage"] = "PREVIEW", ["errorMessage"] = "Preview MP4 upload failed", ["retryable"] = true, }); return; } string? thumbKey = null; if (result.ThumbnailPath is not null) { thumbKey = await _api.UploadArtifactAsync( job.Id, machineId, "thumbnail", result.ThumbnailPath, "image/jpeg", shutdown); } _spooler.Enqueue("finalize", job.Id, new Dictionary { ["machineId"] = machineId, ["artifacts"] = new Dictionary { ["deliveryMovPath"] = manifest.MovOutput, ["previewMovKey"] = previewKey, ["thumbnailKey"] = thumbKey, }, ["renderStats"] = new Dictionary { ["previewSeconds"] = (int)result.Seconds, }, ["media"] = new Dictionary { ["width"] = result.Build?.Width, ["height"] = result.Build?.Height, ["frameCount"] = result.Build?.FrameCount, ["fps"] = manifest.Fps, ["fileName"] = Path.GetFileName(result.Mp4Path!), }, }); // Only clean up once the outputs are safely reported. CleanScratch(job.Id, result.TempAep); } catch (Exception ex) { _log.LogError(ex, "Unexpected preview error for job {JobId}", job.Id); _spooler.Enqueue("fail", job.Id, new Dictionary { ["machineId"] = machineId, ["stage"] = "PREVIEW", ["errorMessage"] = $"Worker exception: {ex.Message}", ["retryable"] = true, }); } finally { ClearState(); } } private static void CleanScratch(string jobId, string? tempAep) { PreviewStage.TryDelete(Path.Combine(WorkerOptions.ScratchDir, $"job_{jobId}_context.json")); PreviewStage.TryDelete(Path.Combine(WorkerOptions.ScratchDir, $"job_{jobId}_result.json")); PreviewStage.TryDelete(Path.Combine(WorkerOptions.ScratchDir, $"job_{jobId}_run.jsx")); PreviewStage.TryDelete(Path.Combine(WorkerOptions.ScratchDir, $"job_{jobId}_thumb.jpg")); if (tempAep is not null) PreviewStage.TryDelete(tempAep); } /// /// Crash recovery (§7.5): on service start, if current-job.json names a /// job, ask the server what became of it. Still ours (CLAIMED/RUNNING) → /// we crashed mid-render; report a retryable fail so the server requeues /// (never resume a partial aerender). Reassigned/finished → discard state; /// the next attempt's PrepareOutputDir clears any orphaned frames. /// public async Task ReconcileAsync(string machineId, CancellationToken ct) { var state = ReadState(); if (state is null) return; _log.LogWarning("Found crash-recovery state for job {JobId} — reconciling", state.JobId); try { var stillOurs = false; if (!string.IsNullOrEmpty(state.ExportId)) { var export = await _api.GetExportAsync(state.ExportId!, ct); var job = export?.RenderJobs.FirstOrDefault(j => j.Id == state.JobId); stillOurs = job?.Status is "CLAIMED" or "RUNNING"; } if (stillOurs) { _log.LogWarning("Server still shows job {JobId} as ours — reporting crash for requeue", state.JobId); _spooler.Enqueue("fail", state.JobId, new Dictionary { ["machineId"] = machineId, ["stage"] = "RENDER", ["errorMessage"] = "Worker restarted mid-render (crash/power loss); never resuming a partial render", ["retryable"] = true, }); } } catch (Exception ex) { _log.LogWarning(ex, "Reconcile failed for job {JobId}; discarding local state (reaper covers the lease)", state.JobId); } ClearState(); } private static void WriteState(CurrentJobState state) { Directory.CreateDirectory(WorkerOptions.DataDir); File.WriteAllText(WorkerOptions.StateFilePath, JsonSerializer.Serialize(state, WorkerOptions.JsonOpts)); } private static CurrentJobState? ReadState() { try { if (!File.Exists(WorkerOptions.StateFilePath)) return null; return JsonSerializer.Deserialize(File.ReadAllText(WorkerOptions.StateFilePath), WorkerOptions.JsonOpts); } catch { return null; } } private static void ClearState() { try { File.Delete(WorkerOptions.StateFilePath); } catch { } } }