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;
var prefix = AerenderRunner.OutputPatternPrefix(manifest.OutputPattern);
try
{
foreach (var f in Directory.EnumerateFiles(outputDirLocal))
{
if (!Path.GetFileName(f).StartsWith(prefix, StringComparison.OrdinalIgnoreCase)) 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}*\" files exist in {Dir} — reporting failure",
job.Id, prefix, 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