Proposing state management capabilities to NCronJob

[falvarez1]

I'd like to discuss the possibility of enhancing NCronJob with state management for jobs, similar to other job orchestration frameworks. This feature would track states like Scheduled, Running, Retrying, Completed, Failed, and Cancelled. Here are the potential benefits and features:

• Enhanced Observability: Track and visualize job states in real-time, providing better insights and monitoring capabilities.
• Actionable States: Implement state change hooks that allow actions to be triggered based on state transitions. This can be used for notifications, cleanup tasks or conditional execution of other jobs (job chaining).
• Job History and Persistence: With state tracking, we can store historical data on job executions, which can be useful for auditing, debugging, and optimizing job schedules.

This enhancement would not only improve the robustness and usability of NCronJob but also bring it closer to feature parity with more advanced workflow orchestration tools.

I see state management as the core just the beginning of what we can do. Starting with basic state management, we have the opportunity to lay a solid foundation that future enhancements can build upon. We can kick things off with the essentials, and then, based on our experiences and feedback, we can expand and refine the features over time.

Would love to hear your thoughts on this idea.

[linkdotnet]

Thanks for the rich ideas!

• Enhanced Observability: I am totally on board! Currently, it is more of a black box that isn't created in somewhat enterprise-grade applications.
• Actionable States: In general, I like the idea. My gut tells me that it is a slippery slope. On the one hand, I like to keep the scheduler rather simple than complex (otherwise, it might end up as yet another Quartz/Hangfire clone). On the other hand, I do understand the need for that. To make it right there has to be a lot of flexibility in the system as well as a nice API that keeps it nice and easy for you
• Job History and Persistence: Also one of the things I ruled out, with the same explanation as above. I wanted to have a very low barrier to use the library with almost 0 config.

But all in all these are very good features. My current way of thinking is to make the NCronJob library Plug&Play ready and offer hooks for extensions packages.

For example, persistence: We would probably need multiple providers (like Postgres, SQL Server, Mongo, Cosmos, ...), so the system has to be flexible either way.

[falvarez1]

For actionable states, we certainly could hash out the design details here.
This wouldn't be another Quartznet/Hangfire. TBH every time I've tried using them I've always found something missing and would end up writing something custom or finding other solutions.

I hear you on very low barrier and easy solution and I'm on the same page. Persistence doesn't necessarily have to use external databases like you mentioned. In order to start simple, or even just keep things simple, we could use a file based write-ahead logging approach. Zero config to the end user. Maybe just a switch that says "persist to disk".
Microsoft has a project called FASTER that may be an option. Either way I think start simple, no need to set up docker-compose scripts with services that call to databases etc. just to set up the infrastructure.

[linkdotnet]

https://github.com/microsoft/FASTER sounds nice! I am very much in favor of just trying it out and in the worst scrap it again.
Alone the name of the library should be a reason to use it :D

Kidding aside - I do agree that is a good start. I had some previous exposure to job-servers in the .net landscape and being overly flexible comes with a high price. That is why I wanted to keep it as simple as possible. So yeah! You convinced me to try out this direciton.

[falvarez1]

@linkdotnet
I started a branch at experiment/state-events-system with some preliminary state management. It's still a WIP but I'd like your feedback if you get a chance.

[falvarez1]

Yes the idea was to broadcast or stream the state changes and progress to a dashboard. I don't think this would be too heavy so long as we buffer the events. Instead of sending one at a time you aggregate all events over a period, let's say one second, then flush everything in the buffer. I can think of a few other strategies to deal with this at various levels of complexity but keeping it simple is best.

[linkdotnet]

Another concept that somewhat fits "natural" here is RxJS / RxNET with a store approach.
Given our requirements for buffering and whatnot, it could be worth a shot.

The only issue is that folks normally don't like it :D

[falvarez1]

Yes it does fit well and is a great package. I've used Reactive extensions many years ago, on different platforms, however I tend to shy away from heavy packages unless absolutely necessary. I think basic buffering can be done with standard data structures.

[falvarez1]

I've used this approach in other projects successfully (simplified here to represent the concept):

public class BufferedNotifier<T> : IDisposable
{
    private readonly IMessageService<T> _messageService;
    private readonly List<T> _messageBuffer = new List<T>();
    private readonly Timer _timer;
    private readonly SemaphoreSlim _asyncLock = new SemaphoreSlim(1, 1);
    private readonly ILogger<BufferedNotifier<T>> _logger;

    public BufferedNotifier(IMessageService<T> messageService, ILogger<BufferedNotifier<T>> logger)
    {
        _messageService = messageService;
        _logger = logger;
        _timer = new Timer(async _ => await FlushMessagesAsync(), null, Timeout.Infinite, Timeout.Infinite);
    }

    public async void Notify(T message)
    {
        await HandleMessageAsync(message);
    }

    private async Task HandleMessageAsync(T message)
    {
        _logger.LogTrace("Handling message");

        if (!_messageService.IsConnected)
        {
            _logger.LogWarning("The message service connection is not open.");
            return;
        }

        await _asyncLock.WaitAsync().ConfigureAwait(false);
        try
        {
            _messageBuffer.Add(message);

            if (_messageBuffer.Count == 1)
            {
                _timer.Change(TimeSpan.FromSeconds(1), Timeout.InfiniteTimeSpan);
            }
        }
        finally
        {
            _asyncLock.Release();
        }
    }

    private async Task FlushMessagesAsync()
    {
        await _asyncLock.WaitAsync();
        List<T> messagesToSend;

        try
        {
            if (_messageBuffer.Count == 0)
                return;

            messagesToSend = [.._messageBuffer];
            _messageBuffer.Clear();
        }
        finally
        {
            _asyncLock.Release();
        }

        if (_messageService.IsConnected)
        {
            try
            {
                await _messageService.SendBatchMessageAsync("ReceiveMessagesBatch", messagesToSend);
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "An error occurred while sending batched messages");
            }
        }
    }

    public void Dispose()
    {
        _timer?.Dispose();
        _asyncLock?.Dispose();
    }
}

[linkdotnet]

Yes it does fit well and is a great package. I've used Reactive extensions many years ago, on different platforms, however I tend to shy away from heavy packages unless absolutely necessary. I think basic buffering can be done with standard data structures.

Fair point. Yes - it doesn't necessarily call for such heavy calibers here. Once you worked with such libraries the flow of data is somewhat nice to follow.

But for now - let's start with a unified mediator.

[falvarez1]

@linkdotnet I created this branch to start working on designing the job commands capabilities to support the actions that would be performed on the Dashboard. So far supports:

• Cancel in-progress JobRun
• Pause JobRun
• Resume JobRun
• Pause Job Queue
• Resume Job Queue
• Cancel all in-progress Job Runs for a given Job type

Job Commands Management WIP branch

Currently focusing on making sure the basic commands work well.
Looking for feedback

[falvarez1]

@linkdotnet I’m jumping back into this! I see there have been some changes since I last worked on the code, so I’ll need to review them. I’d love to get the dashboard finalized and fully functional. Time to dust off my dashboard branch and get cracking! 🚀

[linkdotnet]

Hey, welcome back. Unfortunately, I noticed your message was a bit too late.

The majority of changes are in test-setup. @nulltoken added some observability, which made the Channel approach we used earlier obsolete. We discovered some issues with IDisposable and made smaller changes to the API surface as well. The majority of work was done by @nulltoken - I don't want to take much credit here :D.

At its core, the behavior is somewhat the same (hopefully :D)

[nulltoken]

@falvarez1 Some additional context:

Observability was added to scratch a personal itch (#147).

It eventually allowed to tweak the way we write tests (some remaining work still in flight in #209 & #199).

Observability feature isn't done yet and ExecutionProgress will need to be enhanced to make #36 possible.

@linkdotnet, for instance, was willing to potentially surface the IJobExecutionContext through it. However, in my mind, the ExecutionProgress should be an immutable snapshot (as a consumer of the IObserver interface, I wouldn't want a process updating previous states behind my back ;-)).

As I needn't it yet, I deferred that work to finish updating the test suite. However, we will eventually need new props exposed (but maybe a little brain work should be done before in order to define what identifies a job).

While working on Observability, the concept of Orchestration emerged. It's more or less an abstract wrapper around the execution of a job and of all the dependents jobs it's linked to.

An orchestration is uniquely identifed by an identifier. All jobs belonging to an orchestration share
this same `CorrelationId`.

From a timeline perspective, an orchestration starts before the root job that initiated it
and completes when **all** of its leaf jobs have reached a final state.

As the discussion is about state management, an additional random thought: Currently we're rely on an impossible date to disable jobs (Feb, 31st). Although that works, that looks like a bit of a hack. I wonder if we shouldn't switch to a simple more straightforward internal boolean. Thoughts?

[linkdotnet]

Very good summary and overall points @nulltoken

We had some major discussions around Jobs and state here: #137

However, in my mind, the ExecutionProgress should be an immutable snapshot

Yeah. There are many places where this might be useful but it would increase overall complexity to some degree (for example, Passed-in arguments to a job are completely mutable, which might be surprising).

I wonder if we shouldn't switch to a simple more straightforward internal boolean

Indeed. But that goes into the same discussion as #137: How do we structure the entities we have? What does what and so on. Currently, the easiest would be that boolean on the JobDefinition, but that would be the quick-fix (as 31st of Feb. is).

PS: @falvarez1 and I started to do a discussion on the Matrix Channel. We could "resurrect" the channel.

[nulltoken]

However, in my mind, the ExecutionProgress should be an immutable snapshot

Yeah. There are many places where this might be useful but it would increase overall complexity to some degree (for example, Passed-in arguments to a job are completely mutable, which might be surprising).

Hence, one of the reasons why I deferred the task until now ;-)
That looks like a case of "let's not do it yet. At least, until there's a real use case demanding it"

We had some major discussions around Jobs and state here: #137

I intend to go back to #137 once I'm done with #209 and #199