using OpenNest.Api;
using OpenNest.Data;
using OpenNest.Engine.Jobs;
using OpenNest.Engine.Jobs.Remote;
namespace OpenNest.Server.Worker;
///
/// Solves the nest server's queued jobs one at a time. Each job is read from its package and
/// run through with the named built-in engine wrapped in a
/// , with no time limit; the recorded engine answers are
/// the result the desktop replays. A heartbeat keeps the job's lease; the engine's progress and
/// best plan so far are stored on the server as it runs (), and a
/// cancel request is seen within one report interval.
///
public sealed class NestJobWorker
{
private readonly RemoteNestJobs _jobs;
private readonly ILogger _logger;
private readonly TimeSpan _heartbeat;
private readonly TimeSpan _reportInterval;
public NestJobWorker(RemoteNestJobs jobs, ILogger logger, TimeSpan? heartbeat = null, TimeSpan? reportInterval = null)
{
_jobs = jobs;
_logger = logger;
_heartbeat = heartbeat ?? NestJobRules.HeartbeatInterval;
_reportInterval = reportInterval ?? NestJobRules.ProgressInterval;
}
public string WorkerId { get; } = $"{Environment.MachineName}-{Guid.NewGuid():N}";
/// Called after each heartbeat the server accepts, so a long job still shows the worker alive.
public Action? Alive { get; set; }
/// Claims and runs one job. False when there was none to run.
public async Task RunOnceAsync(CancellationToken stoppingToken)
{
var job = await _jobs.ClaimAsync(WorkerId, stoppingToken);
if (job is null)
return false;
_logger.LogInformation("Nest job {Id} ({Title}, {Engine}) started, attempt {Attempt}.",
job.Id, job.Title, job.EngineName, job.Attempts);
using var solve = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken);
var stop = StopReason.None;
using var beating = new CancellationTokenSource();
void Stop(StopReason reason)
{
if (stop == StopReason.None)
stop = reason;
solve.Cancel();
}
var heartbeat = HeartbeatAsync(job.Id, solve, reason => stop = reason, beating.Token);
var reporter = new NestJobReporter();
var reporting = ReportAsync(job.Id, reporter, Stop, beating.Token);
try
{
NestPipelineRequest request;
try
{
request = NestJobPackage.Read(await _jobs.GetPackageAsync(job.Id, solve.Token));
}
catch (FormatException ex)
{
await Report(_jobs.FailAsync(job.Id, WorkerId, $"The nest job could not be read: {ex.Message}"));
return true;
}
if (!NestingEngineRegistry.IsBuiltIn(request.EngineName))
{
await Report(_jobs.FailAsync(job.Id, WorkerId,
$"The nest server has no '{request.EngineName}' engine; only built-in engines run there."));
return true;
}
var recorder = new RecordingNestingEngine(NestingEngineRegistry.Create(request.EngineName)) { Progress = reporter };
var result = await Task.Factory.StartNew(() => NestPipeline.Run(recorder, request.EngineName, request, progress: null, solve.Token),
solve.Token, TaskCreationOptions.LongRunning, TaskScheduler.Default);
if (await _jobs.CompleteAsync(job.Id, WorkerId, recorder.Transcript.ToJson(), NestJobSummaries.From(result), CancellationToken.None))
_logger.LogInformation("Nest job {Id} finished.", job.Id);
else
_logger.LogWarning("Nest job {Id} finished after this worker lost it; the result was dropped.", job.Id);
}
catch (OperationCanceledException) when (solve.IsCancellationRequested)
{
switch (stop)
{
case StopReason.CancelRequested:
await Report(_jobs.ConfirmCancelledAsync(job.Id, WorkerId));
_logger.LogInformation("Nest job {Id} stopped on request.", job.Id);
break;
case StopReason.LeaseLost:
_logger.LogWarning("Nest job {Id} stopped: this worker no longer holds it.", job.Id);
break;
default:
// Shutting down: the job goes back to the front of the queue.
await Report(_jobs.ReleaseAsync(job.Id, WorkerId));
_logger.LogInformation("Nest job {Id} given back while the worker stops.", job.Id);
break;
}
}
catch (Exception ex) when (ex is not OperationCanceledException)
{
_logger.LogError(ex, "Nest job {Id} failed.", job.Id);
await Report(_jobs.FailAsync(job.Id, WorkerId, ex.Message));
}
finally
{
beating.Cancel();
await heartbeat;
await reporting;
}
return true;
}
private async Task ReportAsync(Guid id, NestJobReporter reporter, Action stopped, CancellationToken token)
{
var reason = await reporter.RunAsync(_jobs, id, WorkerId, _reportInterval, _logger, token);
if (reason != StopReason.None)
stopped(reason);
}
private async Task HeartbeatAsync(Guid id, CancellationTokenSource solve, Action stopped, CancellationToken token)
{
while (!token.IsCancellationRequested)
{
try { await Task.Delay(_heartbeat, token); }
catch (OperationCanceledException) { return; }
try
{
var beat = await _jobs.HeartbeatAsync(id, WorkerId, token);
if (beat is not null)
Alive?.Invoke();
if (beat is null || beat.CancelRequested)
{
stopped(beat is null ? StopReason.LeaseLost : StopReason.CancelRequested);
solve.Cancel();
return;
}
}
catch (OperationCanceledException) when (token.IsCancellationRequested) { return; }
catch (Exception ex) when (ex is IOException or HttpRequestException or TaskCanceledException)
{
// The lease outlasts several missed beats; keep solving and try again.
_logger.LogWarning("Heartbeat for nest job {Id} failed: {Message}", id, ex.Message);
}
}
}
// Reports run after the solve's token may be cancelled, so they never take it.
private async Task Report(Task report)
{
try
{
if (!await report)
_logger.LogWarning("The nest server no longer gave this worker the job it reported on.");
}
catch (Exception ex) when (ex is IOException or HttpRequestException or TaskCanceledException)
{
_logger.LogWarning("Could not report to the nest server: {Message}", ex.Message);
}
}
public enum StopReason
{
None,
CancelRequested,
LeaseLost,
}
}