8014a0e770
Jobs are prompts or watchers. A reminder is a prompt the agent passes on, so
it lands in the transcript. Stored reminder jobs load as prompts
("Remind me: <text>"). Also records the Coolify deploy target in CLAUDE.md.
494 lines
22 KiB
C#
494 lines
22 KiB
C#
using Microsoft.Extensions.Logging;
|
|
using Volcanic.AgenticCore.Abstractions.Processes;
|
|
|
|
namespace Volcanic.AgenticCore.Extensions.Scheduler;
|
|
|
|
/// <summary>
|
|
/// The process-wide scheduler: loads <c>jobs.json</c>, arms one <see cref="TimeProvider"/> timer for the soonest job
|
|
/// (capped at 1 h, re-armed on every tick, create, cancel, fire and zone change), and fires jobs through the host's
|
|
/// <see cref="IJobDelivery"/>. A fire is tracked in flight and never awaited by the tick; an occurrence that comes due
|
|
/// while the job is still in flight is skipped. A watcher's script run touches only the runner, <c>IsDeliverable</c>,
|
|
/// <c>WorkspaceFor</c> and <c>TimeZoneFor</c>: a silent run makes no delivery call at all.
|
|
/// </summary>
|
|
public sealed class JobScheduler : IAsyncDisposable
|
|
{
|
|
public static readonly TimeSpan MaxTimerDelay = TimeSpan.FromHours(1);
|
|
public static readonly TimeSpan LateThreshold = TimeSpan.FromSeconds(60);
|
|
public static readonly TimeSpan StopGrace = TimeSpan.FromSeconds(10);
|
|
private const string IdAlphabet = "abcdefghjkmnpqrstuvwxyz23456789";
|
|
|
|
private readonly SchedulerOptions _options;
|
|
private readonly IScriptRunner _runner;
|
|
private readonly ISecretRedactor _redactor;
|
|
private readonly TimeProvider _time;
|
|
private readonly ILogger _logger;
|
|
private readonly JobStore _store;
|
|
private readonly SemaphoreSlim _saveGate = new(1, 1);
|
|
private readonly Lock _lock = new();
|
|
private readonly Dictionary<string, Entry> _jobs = new(StringComparer.Ordinal);
|
|
private readonly Dictionary<string, Fire> _inFlight = new(StringComparer.Ordinal);
|
|
private readonly CancellationTokenSource _stopping = new();
|
|
private IJobDelivery? _delivery;
|
|
private ITimer? _timer;
|
|
private DateTimeOffset? _wakeAt;
|
|
private bool _stopped;
|
|
private Task _pendingTick = Task.CompletedTask;
|
|
private int _firesStarted;
|
|
|
|
public JobScheduler(SchedulerOptions options, IScriptRunner runner, ISecretRedactor redactor, TimeProvider time, ILogger<JobScheduler> logger)
|
|
{
|
|
ArgumentNullException.ThrowIfNull(options);
|
|
if (string.IsNullOrWhiteSpace(options.StorePath)) throw new ArgumentException("SchedulerOptions.StorePath is required.", nameof(options));
|
|
_options = options;
|
|
_runner = runner;
|
|
_redactor = redactor;
|
|
_time = time;
|
|
_logger = logger;
|
|
_store = new JobStore(options.StorePath, time, logger);
|
|
}
|
|
|
|
public bool IsStarted { get { lock (_lock) return _delivery is not null && !_stopped; } }
|
|
public DateTimeOffset Now => _time.GetUtcNow();
|
|
public string StorePath => _store.Path;
|
|
|
|
/// <summary>When the timer fires next (tests).</summary>
|
|
internal DateTimeOffset? NextWakeAt { get { lock (_lock) return _wakeAt; } }
|
|
|
|
/// <summary>How many fires have been started (tests: a skipped occurrence doesn't count).</summary>
|
|
internal int FiresStarted { get { lock (_lock) return _firesStarted; } }
|
|
|
|
/// <summary>The latest queued tick, without the fires it started (tests; unlike <see cref="IdleAsync"/> it doesn't wait for held fires).</summary>
|
|
internal Task PendingTick { get { lock (_lock) return _pendingTick; } }
|
|
|
|
/// <summary>Load the store, fire overdue one-offs (late), compute cron occurrences strictly after now (no catch-up), arm the timer.</summary>
|
|
public async Task StartAsync(IJobDelivery delivery, CancellationToken cancellationToken)
|
|
{
|
|
ArgumentNullException.ThrowIfNull(delivery);
|
|
var jobs = await _store.LoadAsync(cancellationToken).ConfigureAwait(false);
|
|
var now = _time.GetUtcNow();
|
|
lock (_lock)
|
|
{
|
|
if (_delivery is not null || _stopped) throw new InvalidOperationException("The scheduler was already started.");
|
|
_delivery = delivery;
|
|
foreach (var job in jobs)
|
|
{
|
|
var entry = new Entry(job);
|
|
if (job.Schedule.At is { } at) entry.Next = at;
|
|
else
|
|
{
|
|
try
|
|
{
|
|
entry.Next = NextCron(job, now);
|
|
var missed = Schedules.NextCron(job.Schedule.Cron!, job.LastRunAt ?? job.CreatedAt, delivery.TimeZoneFor(job.Target));
|
|
if (missed is { } m && m < now)
|
|
_logger.LogInformation("Job {Id} ({Name}): occurrences missed while the bot was down are skipped; next {Next}", job.Id, job.Name, entry.Next);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
// One bad entry never keeps the host down: it stays listed but unarmed until a zone change recomputes it.
|
|
entry.Next = null;
|
|
_logger.LogError(ex, "Job {Id} ({Name}): computing the next occurrence on start failed; the job isn't armed", job.Id, job.Name);
|
|
}
|
|
}
|
|
_jobs[job.Id] = entry;
|
|
}
|
|
_timer = _time.CreateTimer(_ => OnTimer(), null, Timeout.InfiniteTimeSpan, Timeout.InfiniteTimeSpan);
|
|
}
|
|
Tick(); // overdue one-offs fire now
|
|
}
|
|
|
|
/// <summary>Disarm, kill running scripts, and wait at most <see cref="StopGrace"/> for in-flight fires.</summary>
|
|
public async Task StopAsync(CancellationToken cancellationToken)
|
|
{
|
|
Fire[] fires;
|
|
ITimer? timer;
|
|
lock (_lock)
|
|
{
|
|
if (_stopped) return;
|
|
_stopped = true;
|
|
fires = [.. _inFlight.Values];
|
|
timer = _timer;
|
|
}
|
|
timer?.Dispose();
|
|
await _stopping.CancelAsync().ConfigureAwait(false); // every fire's token is linked to this one
|
|
try
|
|
{
|
|
await Task.WhenAll(fires.Select(f => f.Task)).WaitAsync(StopGrace, _time, cancellationToken).ConfigureAwait(false);
|
|
}
|
|
catch (TimeoutException)
|
|
{
|
|
_logger.LogWarning("Scheduler stop: {Count} job(s) still running after {Grace}", fires.Count(f => !f.Task.IsCompleted), StopGrace);
|
|
}
|
|
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) { }
|
|
}
|
|
|
|
public ValueTask DisposeAsync() => new(StopAsync(CancellationToken.None));
|
|
|
|
public IReadOnlyList<JobView> List(string target)
|
|
{
|
|
lock (_lock)
|
|
return _jobs.Values.Where(e => e.Job.Target == target)
|
|
.Select(e => new JobView(e.Job, e.Next))
|
|
.OrderBy(v => v.NextRunAt ?? DateTimeOffset.MaxValue).ThenBy(v => v.Job.CreatedAt)
|
|
.ToList();
|
|
}
|
|
|
|
public TimeZoneInfo TimeZoneFor(string target) => Volatile.Read(ref _delivery)?.TimeZoneFor(target) ?? TimeZoneInfo.Local;
|
|
|
|
/// <summary>Validate (an offset-less <c>at</c> is read in the target's zone), persist, arm. Throws <see cref="JobValidationException"/>.</summary>
|
|
public async Task<JobView> CreateAsync(string target, NewJob job, CancellationToken cancellationToken)
|
|
{
|
|
ArgumentNullException.ThrowIfNull(job);
|
|
if (!IsStarted) throw new InvalidOperationException("The scheduler isn't running.");
|
|
var zone = TimeZoneFor(target);
|
|
var now = _time.GetUtcNow();
|
|
var schedule = JobValidation.ValidateNew(job, now, zone);
|
|
ScheduledJob created;
|
|
DateTimeOffset? next;
|
|
lock (_lock)
|
|
{
|
|
if (_jobs.Values.Count(e => e.Job.Target == target) >= _options.MaxJobsPerTarget)
|
|
throw new JobValidationException($"This chat already has {_options.MaxJobsPerTarget} scheduled jobs, the maximum. Cancel one first.");
|
|
created = new ScheduledJob
|
|
{
|
|
Id = NewId(), Target = target, Name = job.Name.Trim(), Kind = job.Kind, Schedule = schedule,
|
|
Text = job.Kind == JobKind.Watcher ? null : job.Text, Command = job.Kind == JobKind.Watcher ? job.Command : null,
|
|
TimeoutSeconds = job.TimeoutSeconds, CreatedAt = now,
|
|
};
|
|
next = schedule.At ?? Schedules.NextCron(schedule.Cron!, now, zone);
|
|
_jobs[created.Id] = new Entry(created) { Next = next };
|
|
}
|
|
await PersistAsync(cancellationToken).ConfigureAwait(false);
|
|
Arm();
|
|
return new JobView(created, next);
|
|
}
|
|
|
|
/// <summary>Remove a job of <paramref name="target"/>. A running watcher is killed and its outcome dropped; a prompt run in progress isn't aborted. False: unknown, or another target's.</summary>
|
|
public async Task<bool> CancelAsync(string target, string id, CancellationToken cancellationToken) =>
|
|
await CancelJobAsync(target, id, cancellationToken).ConfigureAwait(false) is not null;
|
|
|
|
/// <summary><see cref="CancelAsync"/> returning the removed job (null: unknown, or another target's); looked up and removed in one step.</summary>
|
|
public async Task<ScheduledJob?> CancelJobAsync(string target, string id, CancellationToken cancellationToken)
|
|
{
|
|
Fire? running;
|
|
ScheduledJob removed;
|
|
lock (_lock)
|
|
{
|
|
if (!_jobs.TryGetValue(id, out var entry) || entry.Job.Target != target) return null;
|
|
removed = entry.Job;
|
|
_jobs.Remove(id);
|
|
_inFlight.TryGetValue(id, out running);
|
|
}
|
|
if (running is not null) await TryCancelAsync(running.Cancel).ConfigureAwait(false);
|
|
await PersistAsync(cancellationToken).ConfigureAwait(false);
|
|
Arm();
|
|
return removed;
|
|
}
|
|
|
|
/// <summary>The target's zone changed: cron jobs get their next occurrence in the new zone (strictly after now), one-offs keep their instant.</summary>
|
|
public Task<ZoneChange> TimeZoneChangedAsync(string target, CancellationToken cancellationToken)
|
|
{
|
|
var now = _time.GetUtcNow();
|
|
var zone = TimeZoneFor(target);
|
|
int cron = 0, oneOffs = 0;
|
|
lock (_lock)
|
|
{
|
|
foreach (var entry in _jobs.Values.Where(e => e.Job.Target == target))
|
|
{
|
|
if (entry.Job.Schedule.Cron is { } expression)
|
|
{
|
|
cron++;
|
|
entry.Next = Schedules.NextCron(expression, now, zone);
|
|
}
|
|
else oneOffs++;
|
|
}
|
|
}
|
|
Arm();
|
|
return Task.FromResult(new ZoneChange(cron, oneOffs));
|
|
}
|
|
|
|
/// <summary>Wait for the pending tick and every fire in flight (tests).</summary>
|
|
internal async Task IdleAsync()
|
|
{
|
|
while (true)
|
|
{
|
|
Task tick;
|
|
Task[] fires;
|
|
lock (_lock)
|
|
{
|
|
tick = _pendingTick;
|
|
fires = _inFlight.Values.Select(f => f.Task).ToArray();
|
|
}
|
|
if (tick.IsCompleted && fires.All(f => f.IsCompleted)) return;
|
|
await Task.WhenAll(fires.Append(tick)).WaitAsync(TimeSpan.FromSeconds(10)).ConfigureAwait(false);
|
|
}
|
|
}
|
|
|
|
// ---- Timer and ticks ----
|
|
|
|
/// <summary>The timer callback: queues a tick behind the previous one (it may run synchronously inside <c>ITimer.Change</c>).</summary>
|
|
private void OnTimer()
|
|
{
|
|
lock (_lock)
|
|
{
|
|
if (_stopped) return;
|
|
_pendingTick = _pendingTick.ContinueWith(_ => Tick(), CancellationToken.None, TaskContinuationOptions.None, TaskScheduler.Default);
|
|
}
|
|
}
|
|
|
|
/// <summary>Starts every due job that isn't in flight, advances each due job's next occurrence, re-arms. Never awaits a fire.</summary>
|
|
private void Tick()
|
|
{
|
|
try
|
|
{
|
|
var now = _time.GetUtcNow();
|
|
lock (_lock)
|
|
{
|
|
if (_stopped || _delivery is null) return;
|
|
foreach (var entry in _jobs.Values)
|
|
{
|
|
if (entry.Next is not { } next || next > now) continue;
|
|
if (_inFlight.ContainsKey(entry.Job.Id))
|
|
_logger.LogInformation("Job {Id} ({Name}) is still running; skipped the occurrence due {Due}", entry.Job.Id, entry.Job.Name, next);
|
|
else StartFire(entry.Job, next, now);
|
|
// Cron: the next occurrence strictly after now. One-off: none (it's removed once its fire completes).
|
|
entry.Next = null;
|
|
if (entry.Job.Schedule.Cron is null) continue;
|
|
try
|
|
{
|
|
entry.Next = NextCron(entry.Job, now);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
// Left unarmed rather than due forever (a zero-delay timer loop); a restart or zone change recomputes it.
|
|
_logger.LogError(ex, "Job {Id} ({Name}): computing the next occurrence failed", entry.Job.Id, entry.Job.Name);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
finally
|
|
{
|
|
Arm();
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Computes the wake time and applies it to the timer under the lock, so two concurrent arms can't apply an older
|
|
/// wake after a newer one. A synchronous callback from <c>Change</c> (zero delay) is safe: <see cref="Lock"/> is
|
|
/// re-entrant and <see cref="OnTimer"/> only queues a tick.
|
|
/// </summary>
|
|
private void Arm()
|
|
{
|
|
lock (_lock)
|
|
{
|
|
if (_stopped || _timer is null) return;
|
|
var now = _time.GetUtcNow();
|
|
var soonest = _jobs.Values.Select(e => e.Next).Where(n => n is not null).DefaultIfEmpty(null).Min();
|
|
var wake = soonest is { } s && s - now < MaxTimerDelay ? s : now + MaxTimerDelay;
|
|
if (wake < now) wake = now;
|
|
_wakeAt = wake;
|
|
try
|
|
{
|
|
_timer.Change(wake - now, Timeout.InfiniteTimeSpan);
|
|
}
|
|
catch (ObjectDisposedException)
|
|
{
|
|
// Stopped meanwhile.
|
|
}
|
|
}
|
|
}
|
|
|
|
/// <summary>Under the lock. The next cron occurrence strictly after <paramref name="after"/> in the target's current zone.</summary>
|
|
private DateTimeOffset? NextCron(ScheduledJob job, DateTimeOffset after) =>
|
|
Schedules.NextCron(job.Schedule.Cron!, after, _delivery!.TimeZoneFor(job.Target));
|
|
|
|
// ---- Firing ----
|
|
|
|
/// <summary>Called under the lock: registers the fire in flight before it can complete.</summary>
|
|
private void StartFire(ScheduledJob job, DateTimeOffset dueAt, DateTimeOffset now)
|
|
{
|
|
var cancel = CancellationTokenSource.CreateLinkedTokenSource(_stopping.Token);
|
|
var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
|
|
var task = RunFireAsync(gate.Task, job, dueAt, now, cancel);
|
|
_inFlight[job.Id] = new Fire(task, cancel);
|
|
_firesStarted++;
|
|
gate.SetResult();
|
|
}
|
|
|
|
private async Task RunFireAsync(Task gate, ScheduledJob job, DateTimeOffset dueAt, DateTimeOffset now, CancellationTokenSource cancel)
|
|
{
|
|
await gate.ConfigureAwait(false);
|
|
try
|
|
{
|
|
await FireAsync(job, dueAt, now, cancel.Token).ConfigureAwait(false);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogError(ex, "Job {Id} ({Name}) failed to fire", job.Id, job.Name);
|
|
}
|
|
finally
|
|
{
|
|
lock (_lock) _inFlight.Remove(job.Id);
|
|
cancel.Dispose();
|
|
}
|
|
}
|
|
|
|
private async Task FireAsync(ScheduledJob job, DateTimeOffset dueAt, DateTimeOffset now, CancellationToken scriptToken)
|
|
{
|
|
var delivery = _delivery!;
|
|
if (!delivery.IsDeliverable(job.Target))
|
|
{
|
|
// Kept: a cron job has already advanced; a one-off fires again after the next start.
|
|
_logger.LogWarning("Job {Id} ({Name}): target {Target} can't be served any more; skipped", job.Id, job.Name, job.Target);
|
|
return;
|
|
}
|
|
var zone = delivery.TimeZoneFor(job.Target);
|
|
DateTimeOffset? lateDue = now - dueAt > LateThreshold ? dueAt : null;
|
|
JobResult? result;
|
|
switch (job.Kind)
|
|
{
|
|
case JobKind.Prompt:
|
|
result = await DeliverAsync(job, ct => delivery.RunPromptAsync(job.Target, JobMessages.Prompt(job, lateDue, zone), ct)).ConfigureAwait(false)
|
|
? new JobResult(true, null, false, null, _time.GetUtcNow())
|
|
: null;
|
|
break;
|
|
default:
|
|
result = await RunWatcherAsync(job, delivery, zone, scriptToken).ConfigureAwait(false);
|
|
break;
|
|
}
|
|
if (result is not null) await RecordAsync(job.Id, result).ConfigureAwait(false);
|
|
}
|
|
|
|
/// <summary>Null when the run or its delivery was cancelled (job cancelled or scheduler stopping): the outcome is dropped.</summary>
|
|
private async Task<JobResult?> RunWatcherAsync(ScheduledJob job, IJobDelivery delivery, TimeZoneInfo zone, CancellationToken scriptToken)
|
|
{
|
|
var timeout = job.TimeoutSeconds is { } seconds ? TimeSpan.FromSeconds(seconds) : _options.DefaultScriptTimeout;
|
|
var request = new ScriptRequest(job.Command!, delivery.WorkspaceFor(job.Target), timeout, job.Target, job.Id, ZoneNames.Iana(zone));
|
|
ScriptOutcome outcome;
|
|
try
|
|
{
|
|
outcome = await _runner.RunAsync(request, scriptToken).ConfigureAwait(false);
|
|
}
|
|
catch (OperationCanceledException) when (scriptToken.IsCancellationRequested)
|
|
{
|
|
return null;
|
|
}
|
|
if (scriptToken.IsCancellationRequested || !Exists(job.Id)) return null;
|
|
|
|
var at = _time.GetUtcNow();
|
|
switch (outcome)
|
|
{
|
|
case ScriptOutcome.Output output:
|
|
// The condition is met: exactly one agent prompt. Redacted: watcher stdout isn't a tool result.
|
|
return await DeliverAsync(job, ct => delivery.RunPromptAsync(job.Target, JobMessages.WatcherMatch(job, _redactor.Redact(output.Text)), ct)).ConfigureAwait(false)
|
|
? new JobResult(true, 0, false, null, at)
|
|
: null;
|
|
case ScriptOutcome.Failed failed:
|
|
var result = new JobResult(false, failed.ExitCode, failed.TimedOut, _redactor.Redact(failed.StderrTail), at);
|
|
if (!IsRepeat(LastResult(job.Id), result)
|
|
&& !await DeliverAsync(job, ct => delivery.SendTextAsync(job.Target, JobMessages.Failure(job, result), ct)).ConfigureAwait(false))
|
|
return null; // the alert was cut off by shutdown: not recorded, so the next failure alerts again
|
|
return result;
|
|
default:
|
|
return new JobResult(true, 0, false, null, at); // silent: nothing but LastRunAt/LastResult
|
|
}
|
|
}
|
|
|
|
/// <summary>The previous run failed the same way (exit code and timeout): the alert isn't sent again.</summary>
|
|
private static bool IsRepeat(JobResult? previous, JobResult current) =>
|
|
previous is { Ok: false } p && p.ExitCode == current.ExitCode && p.TimedOut == current.TimedOut;
|
|
|
|
/// <summary>
|
|
/// A delivery exception is logged and swallowed; the job's state still advances (no redelivery queue). False only
|
|
/// when shutdown interrupted the delivery: the fire then isn't recorded, so a one-off stays stored and fires again
|
|
/// after the next start (at-least-once).
|
|
/// </summary>
|
|
private async Task<bool> DeliverAsync(ScheduledJob job, Func<CancellationToken, Task> deliver)
|
|
{
|
|
try
|
|
{
|
|
await deliver(_stopping.Token).ConfigureAwait(false); // a prompt run isn't aborted by a cancel, only by shutdown
|
|
return true;
|
|
}
|
|
catch (OperationCanceledException) when (_stopping.IsCancellationRequested)
|
|
{
|
|
_logger.LogInformation("Job {Id} ({Name}): delivery interrupted by shutdown; kept for the next start", job.Id, job.Name);
|
|
return false;
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogError(ex, "Job {Id} ({Name}): delivery to {Target} failed", job.Id, job.Name, job.Target);
|
|
return true;
|
|
}
|
|
}
|
|
|
|
/// <summary>Update and persist the run fields; a one-off is removed now that its fire has completed.</summary>
|
|
private async Task RecordAsync(string id, JobResult result)
|
|
{
|
|
lock (_lock)
|
|
{
|
|
if (!_jobs.TryGetValue(id, out var entry)) return; // cancelled meanwhile
|
|
if (entry.Job.Schedule.At is not null) _jobs.Remove(id);
|
|
else entry.Job = entry.Job with { LastRunAt = result.At, LastResult = result };
|
|
}
|
|
try
|
|
{
|
|
await PersistAsync(CancellationToken.None).ConfigureAwait(false);
|
|
}
|
|
catch (Exception ex) when (ex is IOException or UnauthorizedAccessException)
|
|
{
|
|
_logger.LogError(ex, "Saving the job store failed");
|
|
}
|
|
Arm();
|
|
}
|
|
|
|
/// <summary>Saves the current jobs. The snapshot is taken inside the gate, so a later save never writes older state.</summary>
|
|
private async Task PersistAsync(CancellationToken cancellationToken)
|
|
{
|
|
await _saveGate.WaitAsync(cancellationToken).ConfigureAwait(false);
|
|
try
|
|
{
|
|
List<ScheduledJob> snapshot;
|
|
lock (_lock) snapshot = [.. _jobs.Values.Select(e => e.Job).OrderBy(j => j.CreatedAt).ThenBy(j => j.Id, StringComparer.Ordinal)];
|
|
await _store.SaveAsync(snapshot, cancellationToken).ConfigureAwait(false);
|
|
}
|
|
finally
|
|
{
|
|
_saveGate.Release();
|
|
}
|
|
}
|
|
|
|
private static async Task TryCancelAsync(CancellationTokenSource cancel)
|
|
{
|
|
try
|
|
{
|
|
await cancel.CancelAsync().ConfigureAwait(false);
|
|
}
|
|
catch (ObjectDisposedException)
|
|
{
|
|
// The fire completed meanwhile.
|
|
}
|
|
}
|
|
|
|
private bool Exists(string id) { lock (_lock) return _jobs.ContainsKey(id); }
|
|
|
|
private JobResult? LastResult(string id) { lock (_lock) return _jobs.TryGetValue(id, out var e) ? e.Job.LastResult : null; }
|
|
|
|
/// <summary>Under the lock. Not reused while a cancelled job with that id is still in flight.</summary>
|
|
private string NewId()
|
|
{
|
|
string id;
|
|
do id = new string(Random.Shared.GetItems(IdAlphabet.AsSpan(), 5));
|
|
while (_jobs.ContainsKey(id) || _inFlight.ContainsKey(id));
|
|
return id;
|
|
}
|
|
|
|
private sealed class Entry(ScheduledJob job)
|
|
{
|
|
public ScheduledJob Job { get; set; } = job;
|
|
public DateTimeOffset? Next { get; set; }
|
|
}
|
|
|
|
private sealed record Fire(Task Task, CancellationTokenSource Cancel);
|
|
}
|