Scheduler built with observables v2 (follow-up)

Question

It would probably be too easy if the old Scheduler worked as intended being that simple. But an eye-openig review showed that from time to time it'll miss one second due to 14ms timer incaccuracy. This of course would have bad implications on the schedules (they wouldn't fire if their second was skipped).

---

I spent sometime experimenting with various solutions to fix the missing second bug and I think I found a solution that works nicely. I didn't want it to be too complex, I'm not (currently) writing any systems essential for life. It just should make sure nothing is missing. Without it the observable will accumulate all published values and the first subscriber will recieve them all. We don't want this to happen because we are interested at what's now and now in the history.

---

So how does the Scheduler work? I turned it into a hot observable with Publish + Connect. I want it to be ticking regardles if there is anyone listening.

The new Scheduler also can check how many jobs are already running and decline starting a new one.

How did I fix the missing second issue? The Scheduler is publishing timestamps the same way it did before but now it additionally checks if the difference between the last and the current timestamp is greater then 1s. If it is, it publishes the missing second before publishing the current one. This is achieved by switching from Select to SelectMany. Two intervals published virtually at the same moment don't really matter. More important is to guarantee that nothing is left out and that jobs at any second can be triggered. I think it's extremely unlikely that this would cause any issues ever.

public class Scheduler : IDisposable
{
    public const int UnlimitedJobParallelism = -1;
    
    private const int ZeroJobCounter = 0;

    private IConnectableObservable<DateTime> _scheduler;

    private IDisposable _disconnect;

    private readonly ConcurrentDictionary<Job, int> _jobCounters;

    public Scheduler(TimeSpan period, IDateTime dateTime)
    {
        var lastTimestamp = dateTime.Now();
        _scheduler =
            Observable
                .Interval(period)
                .SelectMany(_ =>
                {
                    // If we missed one second due to time inaccuracy, 
                    // this makes sure to publish the missing second too
                    // so that all jobs at that second can also be triggered.
                    var now = dateTime.Now();
                    var timestampDifference = now.Second - lastTimestamp.Second;
                    var timestamps =
                        Enumerable
                            // When starting the scheduler it might occur 
                            // that the difference is a negative number (-59).
                            // If this is the case, make it 1.
                            .Range(1, timestampDifference < 0 ? 1 : timestampDifference)
                            .Select(second => lastTimestamp.AddSeconds(second));
                    lastTimestamp = now;
                    return timestamps.Dump();
                })
                // Share the actual data values and not just the observable instance.
                .Publish();

        // Turn this into a hot observable and start ticking
        // regardless if there is anything listening.
        _disconnect = _scheduler.Connect();
        _jobCounters = new ConcurrentDictionary<Job, int>();
    }

    public Scheduler()
    : this(TimeSpan.FromSeconds(1), new LocalDateTime())
    { }

    public IDisposable Schedule(Job job)
    {
        var cronExpression = CronExpression.Parse(job.CronExpression);
        var unschedule =
            _scheduler
                .Where(cronExpression.Contains)
                .Subscribe(async timestamp =>
                {
                    if (CanExecute(job))
                    {
                        _jobCounters.AddOrUpdate(job, 1, (j, c) => c + 1);
                        await job
                            .ExecuteAsync(timestamp)
                            .ContinueWith(_ =>
                                _jobCounters
                                    .AddOrUpdate(job, 1, (j, c) => c - 1)
                            );
                    }
                    else
                    {
                        $"'{job.Name}' is alredy running!".Dump();
                    }
                });

        return Disposable.Create(() =>
        {
            _jobCounters.TryRemove(job, out var _);
            unschedule.Dispose();
        });
    }

    private bool CanExecute(Job job)
    {
        // Reversed the second condition to make the left side consistent.
        return
            job.MaxDegreeOfParallelism == UnlimitedJobParallelism ||
            job.MaxDegreeOfParallelism > _jobCounters.GetOrAdd(job, j => ZeroJobCounter);
    }

    public void Dispose()
    {
        // Stop ticking.
        _disconnect.Dispose();
    }
}

---

There are also two new helpers. An extensions for the scheduler and the Job class.

public static class SchedulerExtensions
{
    public static IDisposable Schedule(this Scheduler scheduler, string cronExpression, Func<DateTime, Task> execute, int maxDegreeOfParallelism = 1)
    {
        return scheduler.Schedule(new Job
        {
            CronExpression = cronExpression,
            ExecuteAsync = execute,
            MaxDegreeOfParallelism = maxDegreeOfParallelism
        });
    }
}

public class Job
{
    public string Name { get; set; }

    public string CronExpression { get; set; }

    public Func<DateTime, Task> ExecuteAsync { get; set; }

    public int MaxDegreeOfParallelism { get; set; } = 1;
}

---

Example

The usage didn't change much, just the signatures.

void Main()
{
    var scheduler = new Scheduler();

    // Let the scheduler tick for a moment...
    Thread.Sleep(3300);

    scheduler.Schedule("0/1 * * * * * *", async schedule =>
    {
        Console.WriteLine($"DEBUG: {schedule} - {schedule:ss.fff} [{Thread.CurrentThread.ManagedThreadId}]");
        await Task.Delay(1100);
    }, maxDegreeOfParallelism: Scheduler.UnlimitedJobParallelism);

    scheduler.Schedule("0/2 * * * * * *", async schedule =>
    {
        Console.WriteLine($"ACTION-1: {schedule} - {schedule:ss.fff}");
        await Task.Delay(1500);

    }, maxDegreeOfParallelism: Scheduler.UnlimitedJobParallelism);

    scheduler.Schedule(new Job
    {
        Name = "Test-job",
        CronExpression = "0/3 * * * * * *",
        ExecuteAsync = async schedule =>
        {
            Console.WriteLine($"ACTION-2: {schedule} - {schedule:ss.fff}");
            await Task.Delay(4000);
        },
        MaxDegreeOfParallelism = 1
    });
}