using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; namespace VFXReviewWorker; /// /// Main loop (§7.2): register on startup, heartbeat on its own timer (the /// cancel-signal channel), claim-poll only when idle, one job at a time. /// Workers poll dumbly — all scheduling policy (render windows, Render Now, /// urgent priority) lives server-side in the claim endpoint (§7.10). /// public sealed class WorkerService : BackgroundService { private readonly WorkerOptions _options; private readonly ApiClient _api; private readonly JobExecutor _executor; private readonly ReportSpooler _spooler; private readonly ILogger _log; private string? _machineId; private ServerConfig _config = new(); private volatile string? _currentJobId; private CancellationTokenSource? _cancelCurrentJob; private bool? _lastAeOpen; public WorkerService(WorkerOptions options, ApiClient api, JobExecutor executor, ReportSpooler spooler, ILogger log) { _options = options; _api = api; _executor = executor; _spooler = spooler; _log = log; } protected override async Task ExecuteAsync(CancellationToken stoppingToken) { _log.LogInformation("VFXReview RenderWorker {Version} starting as \"{Name}\" → {Server}", _options.WorkerVersion, _options.MachineName, _options.ServerUrl); _ = Task.Run(() => _spooler.RunAsync(stoppingToken), stoppingToken); await RegisterWithRetryAsync(stoppingToken); await _executor.ReconcileAsync(_machineId!, stoppingToken); _ = Task.Run(() => HeartbeatLoopAsync(stoppingToken), stoppingToken); // Claim loop — only while idle while (!stoppingToken.IsCancellationRequested) { try { // Preview builds drive a full AE instance, which would hijack an // artist's open session (§9.1). Simply don't ask for those jobs // while AE is running — they stay queued for a quiet moment // instead of being claimed and failed. var aeOpen = AeSession.InteractiveRunning(); if (aeOpen != _lastAeOpen) { _log.LogInformation(aeOpen ? "Interactive After Effects detected — not claiming preview jobs until it closes" : "No interactive After Effects — preview jobs enabled"); _lastAeOpen = aeOpen; } var types = aeOpen ? new[] { "AE_RENDER" } : new[] { "AE_RENDER", "PREVIEW_ONLY" }; var job = await _api.ClaimAsync(_machineId!, types, stoppingToken); if (job is not null) { _log.LogInformation("Claimed job {JobId} (attempt {Attempt}/{Max}, export {ExportId})", job.Id, job.Attempt, job.MaxAttempts, job.ExportId); _cancelCurrentJob = CancellationTokenSource.CreateLinkedTokenSource(CancellationToken.None); _currentJobId = job.Id; try { await _executor.RunAsync(job, _machineId!, _config, _cancelCurrentJob.Token, stoppingToken); } finally { _currentJobId = null; _cancelCurrentJob.Dispose(); _cancelCurrentJob = null; } continue; // immediately look for the next job } } catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) { break; } catch (Exception ex) { _log.LogWarning(ex, "Claim poll failed (server unreachable?) — retrying on normal cadence"); } try { await Task.Delay(TimeSpan.FromSeconds(_config.PollSeconds), stoppingToken); } catch (OperationCanceledException) { break; } } _log.LogInformation("RenderWorker stopping"); } private async Task RegisterWithRetryAsync(CancellationToken ct) { var backoff = TimeSpan.FromSeconds(1); while (!ct.IsCancellationRequested) { try { var reg = await _api.RegisterAsync(_options, ct); _machineId = reg.Machine.Id; _config = reg.Config; _log.LogInformation( "Registered as machine {Id} ({Name}); poll={Poll}s heartbeat={Hb}s lease={Lease}s stall={Stall}s enabled={Enabled}", reg.Machine.Id, reg.Machine.Name, _config.PollSeconds, _config.HeartbeatSeconds, _config.LeaseSeconds, _config.StallTimeoutSeconds, reg.Machine.Enabled); return; } catch (OperationCanceledException) when (ct.IsCancellationRequested) { return; } catch (Exception ex) { _log.LogWarning("Registration failed ({Error}); retrying in {Backoff}s", ex.Message, backoff.TotalSeconds); try { await Task.Delay(backoff, ct); } catch (OperationCanceledException) { return; } backoff = TimeSpan.FromSeconds(Math.Min(backoff.TotalSeconds * 2, 60)); } } } private async Task HeartbeatLoopAsync(CancellationToken ct) { while (!ct.IsCancellationRequested) { try { var resp = await _api.HeartbeatAsync(_machineId!, _currentJobId, ct); foreach (var cmd in resp.Commands) { if (cmd.Type == "CANCEL_JOB" && cmd.JobId is not null && cmd.JobId == _currentJobId) { _log.LogWarning("Heartbeat carried CANCEL_JOB for current job {JobId}", cmd.JobId); _cancelCurrentJob?.Cancel(); } } } catch (OperationCanceledException) when (ct.IsCancellationRequested) { return; } catch (Exception ex) { _log.LogDebug(ex, "Heartbeat failed (non-fatal)"); } try { await Task.Delay(TimeSpan.FromSeconds(_config.HeartbeatSeconds), ct); } catch (OperationCanceledException) { return; } } } }