Building a Distributed Job Scheduler with Akka.NET

Imagine your business has to execute long-running jobs that can take anywhere from a few minutes to tens of hours to execute. You want to scale horizontally on-demand in order to parallelize job execution - Akka.NET is a perfect technology for that!

We want to:

1. Distribute jobs evenly across an arbitrary number of workers;
2. Recover mid-flight jobs in the event of worker death;
3. Have our job scheduler / coordinator survive restarts and crashes; and
4. Support scaling the system horizontally (adding new nodes) without interrupting or rebalancing jobs that are already in-progress.

We built an open source proof of concept work distribution platform for one of our Akka.NET Support customers who has this design challenge in their domain. Here’s what the dashboard looks like:

How does it work and how can it apply to your business too?

The design goes back to work I did in the banking sector years ago, where the system sat idle most of the month and then had to churn through a wall of end-of-month reports in the final two days. Jobs arrive in waves, you scale up to meet them, and you cannot afford to throw away work you have already done.

You can also watch our video “Building a Distributed Job Execution Platform with Akka.NET” here, which covers the same code this post does:

Jobs are not entities

Two things are almost always true about long-running work:

1. You should be able to add capacity on demand when many jobs are scheduled at once and
2. Once a job has started you do not want it canceled and restarted somewhere else halfway through.

A scheduler for this kind of work has one thing it can never get wrong: place new work where there is room, and leave work that is already running alone.

That requirement is what rules out the tool most people reach for first: Akka.Cluster.Sharding.

Akka.Cluster.Sharding is the built-in way to distribute stateful things across a cluster, so it looks like the obvious fit. The trouble is a lifecycle mismatch between a sharded entity and a job.

An entity actor lives essentially forever and is only busy in short bursts, so moving it costs almost nothing. Sharding leans on that: it rebalances shards to keep entities evenly distributed when the cluster scales up or down, which is exactly what you want for state. You do not want state concentrated on one node, because that node becomes a single point of failure or bottleneck.

A “job” has the opposite lifespan and requirements from an entity. Jobs have a short, finite lifespan and are intensely busy throughout their duration.

If we run jobs using Akka.Cluster.Sharding, rebalancing is what gets us in trouble:

1. Node A is 60% through a five-minute job;
2. New capacity is added to the cluster;
3. Akka.Cluster.Sharding’s ShardCoordinator determines it’s time to rebalance in order to utilize new capacity; and
4. The Shards being moved kill their child actors (your entities / jobs), which means the jobs will need to be restarted on their new destination node.

We just threw away three minutes of work to satisfy a distribution goal that doesn’t really apply to jobs in the first place. Sharding is a great tool for distributing state. It is not the tool for distributing work.

Clustered routers

The other near-miss is using Akka.Cluster router actors to distribute the work. Routers are built into Akka.Cluster and they do grow and shrink their routee set as the cluster membership changes, so they are closer to the mark.

But a router is dumb by design. It does not remember what it assigned where, it treats every job as the same size, and if the process crashes it has no idea which jobs need reassigning. It is a load balancer, not a scheduler.

So neither built-in primitive is the correct tool for the job.

What we want is a small, persistent scheduler that does three things sharding and routers cannot do together:

1. Estimate the size of each job and distribute them evenly according to available capacity on each node;
2. Never disturb a job that is already running; and
3. Reschedule and recover automatically when a worker node or coordinator dies / crashes.

Everything below is building exactly that, and Akka.NET supplies almost all of it.

Scheduler design

The scheduler is a single custom actor, a JobTracker, built on an Akka.Cluster singleton.

A singleton is an actor with exactly one instance across the whole cluster, which is what you want here: one authoritative view of the schedule, with nothing to keep in agreement across nodes. Schedulers and job trackers are the textbook case for a singleton.

The JobTracker is also persistent, so its schedule survives the actor crashing, getting rebalanced, or getting killed. When it comes back up on another node it replays its journal to rebuild state, then reconciles: it asks the receiver actors whether its schedule still matches what they are actually running, so it can catch up on anything that finished while it was being recreated.

Everything under the tracker is plain actors:

• JobReceiver is a local, top-level actor on each node. It tracks the in-progress jobs there and spawns a JobExecutor for each one.
• JobExecutor does the actual work. In the sample it is a Task.Delay standing in for real processing, but that is where your transcription or transcode or ETL would live.
• JobSubmitter is a per-tenant actor, backed by a shard region, that submitted a job and wants progress and completion updates for its own jobs. In a multi-tenant system each submitter is a different customer.

This hierarchy deliberately mirrors the shape of Akka.Cluster.Sharding, with one difference that is the entire point: the JobTracker does not control the lifecycle of the receivers and executors.

They self-manage. New capacity joining the cluster gives the tracker new places to send future jobs, and nothing already running is touched. That is how you get the “leave running work alone” requirement for free. The live dashboard, if you are wondering how the progress bars work, is a supervisor actor that subscribes to those progress updates and turns them into Akka.Streams served as ASP.NET Core server-sent events (SSE). Fun, but not the interesting part.

Distributing work by job size

“Equally loading” the worker nodes does not mean each node gets the same number of total jobs. It means that all nodes are equally utilized, which is a function of the number of jobs and the complexity or “size” of those jobs.

A job carries a size, and computing it is a cheap, usually O(1) estimate of how expensive the job is to run.

// A job is just an id and a size. Sizing is business-specific and cheap.
public sealed record JobDefinition(JobId Id, JobSize Size);

What size means is domain-dependent and almost always something you cheaply measure or estimate before the job starts.

• Transcribing call-center audio: job size is minutes of audio.
• Transcoding video: total file size is a better proxy for complexity than playback length in minutes, because native video resolution impacts this.
• Running an ETL or financial job: the number of rows or transactions to analyze.

Without a size the scheduler treats all jobs evenly and this will inevitably form “hotspots” in a real-world system, which is inefficient and potentially error-prone.

The tracker listens for several types of inputs:

1. Commands - requests to start or stop jobs;
2. Akka.Cluster events - new nodes joining, old nodes leaving, or current nodes becoming temporarily unavailable. These all impact scheduling decisions; and
3. Job events - progress reports about the jobs themselves from the worker nodes executing them.

// Command: validated, may be rejected. Decide() turns it into events.
public interface IJobTrackerCommand : IJobTrackerDomain;

// Fact: already happened, reported from outside. Never validated, never rejected —
// "no" is not a legal answer to a thing that occurred. Integrate() turns it into events.
public interface IJobTrackerFact : IJobTrackerDomain;

// Event: what Apply() folds into state, and what gets persisted.
public interface IJobTrackerEvent : IJobTrackerDomain;

// The entire command surface is two:
public sealed record SubmitJob(JobDefinition Job, JobSubmitterId SubmitterId) : IJobTrackerCommand;
public sealed record CancelJob(JobId Id, JobSubmitterId SubmitterId, string Reason) : IJobTrackerCommand;

A command is a request to do something, so it runs through Decide and can be rejected.

A fact is something that already happened out in the world, like a node joining the cluster or a worker reporting progress; you do not get to reject a fact, so facts run through Integrate. Both paths emit events, and events are the only thing that changes state, through Apply, and the only thing that gets persisted. That split is what keeps the scheduling logic pure.

“Facts” are events too, but we’re distinguishing external events (happen outside the tracker) from internal “events” (produced by the tracker’s business rules) here.

When the facts of the cluster or jobs-in-progress change, we use that information to redistribute queued jobs or in-progress jobs that were killed because of a NodeLeft event.

public ImmutableArray<IJobTrackerEvent> Integrate(IJobTrackerFact fact, DateTimeOffset now)
{
    var events = fact switch
    {
        JobTrackerFacts.NodeJoined f         => OnNodeJoined(f, now),
        JobTrackerFacts.NodeLeft f           => OnNodeLeft(f, now),
        JobTrackerFacts.ExecutionCompleted f => OnExecutionCompleted(f, now),
        JobTrackerFacts.ProgressReported f   => OnProgressReported(f, now),
        // ...
        _ => ImmutableArray<IJobTrackerEvent>.Empty
    };

    // then place whatever can now run (and rebalance queued work if capacity grew)
    var placements = Fold(events).PlacePendingJobs(now, capacityExpanded);
    return events.AddRange(placements);
}

A thin adapter turns real Akka.Cluster membership events, like MemberUp / MemberRemoved, into NodeJoined and NodeLeft - we did this mostly to make things more testable. You certainly do not need to do it that way.

The placement itself, PlacePendingJobs above, comes down to one method:

private Address? SelectDispatchNode(JobDefinition job,
    IReadOnlyDictionary<Address, JobSize> inUse,
    IReadOnlyDictionary<Address, JobSize> queuedSize,
    Address? preferredNode = null)
{
    var eligible = Nodes.Values.Where(n => n.IsEligible).ToList();
    if (eligible.Count == 0) return null;

    JobSize Load(NodeStatus n) => inUse[n.NodeAddress] + queuedSize[n.NodeAddress];

    var fitting = eligible
        .Where(n => n.MaximumCapacity >= job.Size)   // big enough to hold it
        .OrderBy(Load)                                // least-loaded first
        .ThenBy(n => Equals(n.NodeAddress, preferredNode) ? 0 : 1)
        .ThenBy(n => n.NodeAddress.ToString(), StringComparer.Ordinal)
        .ToList();

    if (fitting.Count > 0) return fitting[0].NodeAddress;

    // oversized: nothing's big enough — roomiest node, and it runs there exclusively
    return eligible
        .OrderByDescending(n => n.MaximumCapacity)
        .ThenBy(Load)
        .First().NodeAddress;
}

The load on a node is:

1. Node’s currently executing work plus
2. Its future reserved work

So the tracker picks the node with the most real spare capacity, not the one running the fewest jobs right now.

A job that fits goes to the least-loaded eligible node. A job bigger than any single node’s capacity is oversized, so it goes to the roomiest node and runs there exclusively once that node drains.

This is deliberately a simple algorithm. You could get much fancier: gap-detection to slip small jobs onto a busy node while a big one waits, reprioritization so rescheduled jobs jump the queue, and all of it layers on top of this. But that’s all out of scope for a proof of concept.

Separating scheduler state from the actor

That state, the pending jobs, the tracked jobs, and the known nodes with their capacity, lives in one immutable C# record. This is a pattern we recommend in Akka.NET bootcamp and elsewhere, such as in our “Akka.NET Application Design” videos.

We recommend it because it vastly simplifies testing and keeps your actor relatively thin and infrastructure-focused.

[Fact]
public void Submitting_a_job_that_fits_schedules_it_immediately()
{
    var tracker = new Tracker();
    tracker.Send(Joining("node-a", 100));

    tracker.Send(new SubmitJob(Job("job-1", 40), Submitter));

    Assert.Equal(JobStatus.Running, tracker.Job("job-1").Status);
    Assert.Equal(new JobSize(40), tracker.Node(Node("node-a")).CapacityInUse);
}

This test would be much harder to author if we had to spin up a real cluster, have multiple nodes join, schedule real jobs, and so on. Much better to have it as a simple record type with some extension methods for applying the business rules.

Wrapping up

I recently wrote a post about “Managing Long-Running Tasks Inside Akka.NET Actors”; that’s a useful post to visit too, but what we built here is a much more ambitious, production-grade abstraction that can sit atop the one mentioned in that post.

There are lots of tools, in the .NET space and elsewhere, that all take aim at job execution. My experience working across sectors like banking, multi-media processing, bulk data import / export, simulations, and others is that most companies usually end up having to roll something domain-specific in-house due to each organization’s unique constraints and requirements.

The sample we put together and made available on GitHub gives you a good starting point to attempt this using Akka.NET from day one.

And if you want some help applying this type of design to your organization, that’s what our Akka.NET Support plans and consulting services are for. Schedule a free consultation with us.