Forging a Distributed Job Engine on MQTT 5.0: Bitmask Cron, Source Generators, and LWT Crash Recovery

The complete implementation of the Beskar.Jobs suite (Beskar.Jobs.Common, Beskar.Jobs.Client, Beskar.Jobs.Worker, Beskar.Jobs.Scheduler, and Beskar.Jobs.Generator) is open-source and available in the Beskar.Networking GitHub Repository.

From packets to orchestration: why build a job system on MQTT 5.0?

Across the previous installments of the Beskar Networking series, we built a networking stack from the ground up:

1. In Sockets on a Pipeline, we replaced NetworkStream with System.IO.Pipelines and PinnedBlockMemoryPool to move raw bytes with near-zero GC pressure.
2. In Allocations on the Wire, we built an MQTT v3.1.1 / v5.0 broker using a UTF-8 byte Trie (MqttTrieSubscriptionRouter) and ReadOnlySpan<byte> Alternate Lookups.
3. In This Is the Way: Low-Allocation Request-Response in MQTT 5.0, we turned a decoupled pub/sub broker into a bidirectional RPC fabric using MQTT 5.0 ResponseTopic and binary CorrelationData.

Once I had bidirectional RPC, retained state topics, and broker-monitored connection lifecycles running over any transport (TCP, TLS, WebSockets, QUIC, Unix Domain Sockets, Named Pipes, or In-Memory), a question naturally followed:

What happens if we build a distributed background job and workflow engine directly on top of MQTT 5.0?

Most distributed job queues in .NET (like Hangfire or Quartz.NET) are built around database polling. Workers spin in loops executing SELECT ... FOR UPDATE SKIP LOCKED against SQL tables, heartbeats are written as periodic UPDATE statements, and detecting a crashed worker requires waiting for a visibility timeout or lock lease to expire—often 30 seconds to several minutes later. Worse, if a caller wants live progress updates or the final return value of a background job, they usually have to poll the database too.

By building Beskar.Jobs natively on top of Beskar.Mqtt, we flip every one of those polling loops into an event-driven protocol primitive:

• Two-Phase RPC Job Dispatch: Instead of workers hammering a database for work, the JobScheduler claims due jobs in batches and dispatches them to capable, low-loaded workers over MQTT 5.0 Request-Response (jobs/dispatch/{queue}/{workerId}).
• Zero-Race Progress & Result Streaming: Status updates (JobStatusUpdate) are published to jobs/status/{jobId} with the MQTT Retain flag set. Clients can stream live progress percentages and strongly-typed results via await foreach (var update in client.WatchJobStatusAsync(jobId))—and even if a client subscribes after a fast job has already finished, the broker delivers the retained terminal state immediately.
• Instant Crash Failover via MQTT Last Will & Testament (LWT): Every JobWorker registers an MQTT 5.0 Will Message (IsOnline = false) when connecting. If a worker process crashes or loses network connectivity mid-job, the broker immediately publishes its Will, and the scheduler recovers and reschedules all orphaned in-flight jobs in milliseconds rather than waiting for a lease timeout.
• Compile-Time Source Generation: A Roslyn incremental generator (BeskarJobGenerator) turns [GeneratedJobHandler] attributes into static abstract descriptors, eradicating magic strings for job types and queue names across clients, workers, and multi-stage continuations.

Let’s open the hood and walk through how each piece of Beskar.Jobs was designed and built.

1. Cluster topology & compile-time wire protocol

Beskar.Jobs separates responsibilities into five focused packages:

• Beskar.Jobs.Common: Core models, state machine enums, the IJobStorage persistence contract (InMemoryJobStorage), the bitmask CronSchedule engine, OpenTelemetry metrics (JobMetrics), and source-generated topic routers (JobTopics).
• Beskar.Jobs.Generator: The Roslyn incremental source generator ([GeneratedJobHandler]).
• Beskar.Jobs.Client: Lightweight producer client for enqueuing immediate, delayed, or DAG continuation jobs, managing recurring Cron schedules, canceling jobs, and streaming real-time progress.
• Beskar.Jobs.Worker: Execution node managing concurrency slots, handler invocation, timeout/cancellation tokens, heartbeats, and LWT registration.
• Beskar.Jobs.Scheduler: Orchestration coordinator running the Cron evaluator, IJobStorage state transitions, capability-aware worker load balancing, retry backoff policies, and LWT crash recovery.

  +-------------+        1. Enqueue / Cron / Cancel (MQTT v5 RPC)        +--------------------+
  |  JobClient  | -----------------------------------------------------> |                    |
  +-------------+ <----------------------------------------------------- |                    |
                  5. Retained Status Stream (await foreach on status/+)  |                    |
                                                                         | Beskar.Mqtt.Server |
  +--------------+       2. Commands, Events & Worker Heartbeats         | (MQTT v5.0 Broker) |
  | JobScheduler | <---------------------------------------------------> |                    |
  |  + Storage   |                                                       |                    |
  |  + Cron      |       3. Two-Phase Claim Dispatch RPC                 |                    |
  +--------------+ ----------------------------------------------------> |                    |
                                                                         +--------------------+
                                                                            |              ^
                                           3. Dispatch (dispatch/queue/id)  |              | 4. Events, Progress
                                                                            v              |    & LWT Heartbeats
                                                                      +--------------------------+
                                                                      | JobWorker Pool (#1..#N)  |
                                                                      +--------------------------+

Zero-allocation topic parsing with [GeneratedMqttTopic]

Every message exchanged between JobClient, JobScheduler, and JobWorker flows over a deterministic MQTT topic hierarchy. Rather than using string.Split('/') or regular expressions on incoming topics, JobTopics.cs uses our compile-time [GeneratedMqttTopic] source generator to emit span-based parsers and formatters:

using Beskar.Mqtt.Common.Generators;

namespace Beskar.Jobs.Common.Topics;

public static partial class JobTopics
{
   public const string EnqueueCommandTopic = "jobs/commands/enqueue";
   public const string RecurringCommandTopic = "jobs/commands/recurring";
   public const string CancelCommandWildcard = "jobs/commands/cancel/+";
   public const string EventsWildcard = "jobs/events/+/+";
   public const string WorkerHeartbeatWildcard = "jobs/workers/+/heartbeat";
   public const string StatusWildcard = "jobs/status/+";

   [GeneratedMqttTopic("jobs/commands/cancel/{jobId}")]
   public static partial bool TryParseCancelCommand(ReadOnlySpan<char> topic, out string jobId);

   [GeneratedMqttTopic("jobs/dispatch/{queue}/{workerId}")]
   public static partial bool TryParseDispatch(ReadOnlySpan<char> topic, out string queue, out string workerId);

   [GeneratedMqttTopic("jobs/status/{jobId}")]
   public static partial bool TryParseStatus(ReadOnlySpan<char> topic, out string jobId);

   [GeneratedMqttTopic("jobs/events/{jobId}/{eventType}")]
   public static partial bool TryParseWorkerEvent(ReadOnlySpan<char> topic, out string jobId, out string eventType);

   [GeneratedMqttTopic("jobs/workers/{workerId}/heartbeat")]
   public static partial bool TryParseWorkerHeartbeat(ReadOnlySpan<char> topic, out string workerId);
}

When the scheduler or worker receives an MQTT packet, it slices context.Message.Topic.AsSpan() and extracts route parameters (jobId, queue, workerId, eventType) in a single forward scan without allocating intermediate strings or arrays.

2. Eradicating magic strings: BeskarJobGenerator and static abstract descriptors

One of my biggest pet peeves in background job frameworks is stringly-typed wiring. You register a handler under "media.video.transcode" on queue "media", and three months later someone enqueues "media.video.transcoder" on "default"—resulting in a silent runtime failure or a dead-lettered job.

To make job routing type-safe without reflection overhead at runtime, we combine C# static abstract interface members with a Roslyn Incremental Source Generator.

First, we define IGeneratedJobDescriptor:

namespace Beskar.Jobs.Common.Interfaces;

public interface IGeneratedJobDescriptor
{
   public static abstract string GeneratedJobType { get; }
   public static abstract string GeneratedQueue { get; }
}

Next, when you write a job handler, you decorate your partial class with [GeneratedJobHandler] and implement either IJobHandler<TArgs> (for void jobs) or IJobHandlerWithResult<TArgs, TResult> (for jobs that return a typed result):

[GeneratedJobHandler("media.video.transcode", Queue = "media")]
public sealed partial class VideoTranscodeJobHandler
   : IJobHandlerWithResult<VideoTranscodeArgs, VideoTranscodeResult>
{
   public async ValueTask<VideoTranscodeResult> ExecuteAsync(
      IJobExecutionContext<VideoTranscodeArgs> context,
      CancellationToken ct = default)
   {
      await context.ReportProgressAsync(50, $"Encoding {context.Arguments.SourceFile}...", ct);
      await Task.Delay(150, ct);
      return new VideoTranscodeResult($"/cdn/{context.Arguments.SourceFile}.mp4", 4_194_304);
   }
}

Inside the Roslyn Incremental Generator

At compile time, BeskarJobGenerator hooks into SyntaxProvider.ForAttributeWithMetadataName. In BeskarJobGenerator.Transform.cs, we inspect the Roslyn INamedTypeSymbol and validate four invariants up front:

1. JOBGEN001: The target type—and every enclosing parent type if nested—must be declared partial.
2. JOBGEN002: The jobType constructor argument cannot be empty or whitespace.
3. JOBGEN003: The Queue named argument cannot be empty or whitespace.
4. JOBGEN004: The class must implement IJobHandler<TArgs> or IJobHandlerWithResult<TArgs, TResult>.

Crucially, the transform step extracts a pure value-equatable record (GeneratedJobHandlerModel) containing no ISymbol or SyntaxNode references, ensuring Roslyn's incremental caching works without leaking compilation trees in IDE memory.

Then, in BeskarJobGenerator.Rendering.cs, we use CodeTextWriter from Beskar.CodeGeneration initialized over a stackalloc char[512] buffer to emit the descriptor implementation:

private static string GenerateDescriptorMembers(GeneratedJobHandlerModel model)
{
   Span<char> initialBuffer = stackalloc char[512];
   var writer = new CodeTextWriter(initialBuffer, stackalloc char[64]);

   try
   {
      var escapedJobType = SymbolDisplay.FormatLiteral(model.JobType, true);
      var escapedQueue = SymbolDisplay.FormatLiteral(model.Queue, true);

      writer.WriteLineInterpolated($"public const string JobType = {escapedJobType};");
      writer.WriteLineInterpolated($"public const string DefaultQueue = {escapedQueue};");
      writer.WriteLine("public static string GeneratedJobType => JobType;");
      writer.WriteLine("public static string GeneratedQueue => DefaultQueue;");

      return writer.ToString();
   }
   finally
   {
      writer.Dispose();
   }
}

Because VideoTranscodeJobHandler now implements IGeneratedJobDescriptor, the generic constraints on JobWorkerBuilder and JobClient can resolve THandler.GeneratedJobType and THandler.GeneratedQueue directly at JIT compile time:

// Worker registration: automatically registers "media.video.transcode" AND subscribes to queue "media"
var worker = JobWorkerBuilder.Create()
   .RegisterGeneratedHandlerWithResult<VideoTranscodeJobHandler, VideoTranscodeArgs, VideoTranscodeResult>()
   .UseTcp(endPoint)
   .Build();

// Client enqueue: zero string literals required!
var response = await client.EnqueueGeneratedWithResultAsync<
   VideoTranscodeJobHandler,
   VideoTranscodeArgs,
   VideoTranscodeResult>(new VideoTranscodeArgs("intro.mov", "av1"));

3. A zero-allocation bitmask Cron & interval engine (CronSchedule)

A job scheduler needs a fast, rock-solid Cron evaluator. Most Cron libraries allocate arrays of valid values per field or build AST objects on the heap during parsing, and then step minute-by-minute when calculating the next occurrence.

In CronSchedule.cs, I designed a zero-allocation bitmask evaluator packed into a readonly struct that supports:

• 5-field standard Cron (*/15 9-17 * * 1-5)
• 6-field seconds-precision Cron (0,30 * * * * *)
• Standard macros (@yearly, @monthly, @weekly, @daily, @hourly)
• Sub-second & custom intervals (@every 250ms, @every 5s, @every 15m)

Packing a full Cron expression into 22 bytes of bitmasks

Every calendar field has a small, bounded domain:

• Seconds (0..59) -> fits in a 64-bit ulong (SecondsMask)
• Minutes (0..59) -> fits in a 64-bit ulong (MinutesMask)
• Hours (0..23) -> fits in a 32-bit uint (HoursMask)
• Days of Month (1..31) -> fits in a 32-bit uint (DaysOfMonthMask)
• Months (1..12) -> fits in a 16-bit ushort (MonthsMask)
• Days of Week (0..6, with 7 normalized to 0 for Sunday) -> fits in an 8-bit byte (DaysOfWeekMask)

When parsing an expression in CronSchedule.TryParse(ReadOnlySpan<char> expression, out CronSchedule schedule), we tokenize the span into a stack-allocated Span<Range> ranges = stackalloc Range[7] and parse each comma-separated segment (1-5, */10, JAN-MAR, MON-FRI) directly into its corresponding bitmask:

public static bool TryParse(ReadOnlySpan<char> expression, out CronSchedule schedule)
{
   schedule = default;
   expression = expression.Trim();
   if (expression.IsEmpty) return false;

   if (expression[0] == '@')
   {
      return TryParseMacro(expression, out schedule);
   }

   // Split into up to 7 tokens by whitespace on the stack
   Span<Range> ranges = stackalloc Range[7];
   var fieldCount = SplitWhitespace(expression, ranges);
   if (fieldCount is not (5 or 6)) return false;

   var offset = 0;
   var secondsMask = 1UL; // Default second = 0 for 5-field cron

   if (fieldCount == 6)
   {
      if (!TryParseField(expression[ranges[0]], 0, 59, FieldKind.Numeric, out secondsMask, out _))
         return false;
      offset = 1;
   }

   if (!TryParseField(expression[ranges[offset]], 0, 59, FieldKind.Numeric, out var minutesMask, out _))
      return false;
   if (!TryParseField(expression[ranges[offset + 1]], 0, 23, FieldKind.Numeric, out var hoursMask64, out _))
      return false;
   if (!TryParseField(expression[ranges[offset + 2]], 1, 31, FieldKind.Numeric, out var domMask64, out var isDomWildcard))
      return false;
   if (!TryParseField(expression[ranges[offset + 3]], 1, 12, FieldKind.Month, out var monthsMask64, out _))
      return false;
   if (!TryParseField(expression[ranges[offset + 4]], 0, 7, FieldKind.DayOfWeek, out var dowMask64, out var isDowWildcard))
      return false;

   // Map Sunday = 7 to bit 0
   if ((dowMask64 & (1UL << 7)) != 0)
   {
      dowMask64 = (dowMask64 & ~(1UL << 7)) | 1UL;
   }

   schedule = new CronSchedule(
      secondsMask, minutesMask, (uint)hoursMask64,
      (uint)domMask64, (ushort)monthsMask64, (byte)dowMask64,
      isDomWildcard, isDowWildcard);
   return true;
}

Hardware-intrinsic bit jumping (BitOperations.TrailingZeroCount)

Once a schedule is represented as bitmasks, checking if a timestamp matches (IsMatch) is just five bit-shifts. Even better, finding the next occurrence (GetNextOccurrence) doesn't need to loop second-by-second or minute-by-minute.

Instead, we mask off all bits below the current unit (mask & (~0UL << startBit)) and use the CPU's hardware TZCNT / BSF instruction via BitOperations.TrailingZeroCount to jump straight to the next enabled bit in O(1) time:

private static int NextSetBit32(uint mask, int startBit, int maxBit)
{
   if (startBit > maxBit) return -1;
   var shifted = mask & (~0U << startBit);
   if (shifted == 0) return -1;
   var bit = BitOperations.TrailingZeroCount(shifted);
   return bit <= maxBit ? bit : -1;
}

private static int NextSetBit64(ulong mask, int startBit, int maxBit)
{
   if (startBit > maxBit) return -1;
   var shifted = mask & (~0UL << startBit);
   if (shifted == 0) return -1;
   var bit = BitOperations.TrailingZeroCount(shifted);
   return bit <= maxBit ? bit : -1;
}

If no valid second remains in the current minute (nextSecond < 0), we increment the minute, reset second = 0, and jump to the next set minute bit. This makes computing the next execution timestamp across thousands of recurring schedules virtually instantaneous and completely allocation-free.

4. Capability-aware load balancing & two-phase RPC dispatch

In a distributed cluster, workers are rarely homogeneous. Some worker nodes might have GPUs subscribed to the "media" queue for video transcoding, while others handle "billing" and "notifications" with different concurrency limits.

Every JobWorker advertises its live state by publishing a retained WorkerHeartbeatPayload to jobs/workers/{workerId}/heartbeat:

• IsOnline: true while active, false on graceful stop or broker LWT trigger.
• Queues: Subscribed queue names (e.g., ["media", "default"]).
• SupportedJobTypes: Registered handler types (e.g., ["media.video.transcode", "media.image.thumbnail"]).
• ActiveJobs & MaxConcurrency: Current in-flight job count and maximum concurrency slots.

The 4-stage eligibility filter & round-robin tie-breaker

When JobScheduler claims a batch of due jobs from IJobStorage, it queries WorkerRegistry.SelectCandidateWorkers(queue, jobType). The registry filters for online workers with free concurrency slots that match both the queue and the job type, and then sorts them by lowest load factor (ActiveJobs / MaxConcurrency) with an atomic round-robin hash tie-breaker:

public List<string> SelectCandidateWorkers(string queue, string jobType)
{
   lock (_lock)
   {
      var candidates = new List<WorkerNodeState>();
      foreach (var w in _workers.Values)
      {
         if (!w.IsOnline || w.AvailableSlots <= 0) continue;
         if (!w.Queues.Contains(queue)) continue;
         if (w.SupportedJobTypes.Count > 0 && !w.SupportedJobTypes.Contains(jobType)) continue;

         candidates.Add(w);
      }

      if (candidates.Count == 0) return [];

      var rr = _roundRobinCounter++;
      candidates.Sort((a, b) =>
      {
         var cmp = a.LoadFactor.CompareTo(b.LoadFactor);
         if (cmp != 0) return cmp;

         var hashA = ((uint)StringComparer.Ordinal.GetHashCode(a.WorkerId) + rr) & 0x7FFFFFFF;
         var hashB = ((uint)StringComparer.Ordinal.GetHashCode(b.WorkerId) + rr) & 0x7FFFFFFF;
         return hashA.CompareTo(hashB);
      });

      var ids = new List<string>(candidates.Count);
      foreach (var c in candidates) ids.Add(c.WorkerId);
      return ids;
   }
}

Why does the round-robin hash matter? Suppose you have four idle workers (LoadFactor == 0.0) and a burst of 20 jobs arrives in a single scheduler tick. Without combining the worker ID's hash with _roundRobinCounter++, Sort would return the exact same worker order every time. By rotating the tie-breaker atomically and optimistically incrementing the chosen worker's ActiveJobs count as soon as a dispatch claim is sent, consecutive jobs fan out evenly across the entire worker pool.

Two-phase claim & lock-free concurrency slot reservation

Once the scheduler picks a candidate worker, it doesn't just fire-and-forget the job. What if the worker's last concurrency slot filled up a microsecond ago, or its socket just dropped?

Instead, JobScheduler performs a two-phase claim using our MQTT 5.0 Request-Response RPC (MqttClient.RequestAsync) on jobs/dispatch/{queue}/{workerId}. On the receiving side, JobWorker.HandleDispatchRequestAsync uses a lock-free Interlocked.CompareExchange loop to atomically reserve a concurrency slot before replying Accepted = true:

// Atomically reserve a concurrency slot on the worker
while (true)
{
   var current = Volatile.Read(ref _activeJobs);
   if (current >= Options.MaxConcurrency)
   {
      var busyReject = new JobDispatchResponse
      {
         Accepted = false,
         JobId = request.JobId,
         WorkerId = Options.WorkerId,
         RejectReason = $"Worker '{Options.WorkerId}' is at maximum concurrency ({Options.MaxConcurrency})."
      };

      await context.RespondAsync(Options.Serializer.Serialize(busyReject), QualityOfServiceType.AtLeastOnce, ct);
      return;
   }

   if (Interlocked.CompareExchange(ref _activeJobs, current + 1, current) == current)
   {
      break;
   }
}

var acceptResponse = new JobDispatchResponse
{
   Accepted = true,
   JobId = request.JobId,
   WorkerId = Options.WorkerId
};

await context.RespondAsync(Options.Serializer.Serialize(acceptResponse), QualityOfServiceType.AtLeastOnce, ct);

var executionTask = Task.Run(() => ExecuteClaimedJobAsync(request, executor), CancellationToken.None);
_runningJobTasks[request.JobId] = executionTask;

If a worker rejects the claim (or if the RPC times out), the scheduler decrements its optimistic reservation and immediately tries the next candidate worker in the list. If no worker accepts the job during the pass, the job is cleanly released back to JobStatus.Enqueued.

5. Instant worker crash recovery via MQTT 5.0 Last Will & Testament

What happens if a worker accepts a job, begins executing it (JobStatus.Running), and then the worker process segfaults, gets OOM-killed by Kubernetes, or loses power?

In traditional job queues, that job sits stuck in Running until a background reaper thread notices its lock lease has expired minutes later. With MQTT 5.0, the broker itself is already monitoring the TCP/QUIC socket of every connected worker.

When JobWorker.StartAsync connects to the broker, it attaches a retained Last Will & Testament (LWT) payload with IsOnline = false on its heartbeat topic (jobs/workers/{workerId}/heartbeat):

var offlineWill = new WorkerHeartbeatPayload
{
   WorkerId = Options.WorkerId,
   IsOnline = false,
   Queues = [.. Options.Queues],
   SupportedJobTypes = [.. _handlers.Keys],
   ActiveJobs = 0,
   MaxConcurrency = Options.MaxConcurrency,
   TimestampUtc = DateTimeOffset.UtcNow
};

var connectOptions = new ConnectOptions
{
   EndPoint = Options.EndPoint!,
   ProtocolVersion = MqttProtocolVersion.V50,
   ClientIdUtf8Bytes = Encoding.UTF8.GetBytes(Options.WorkerId),
   CleanSession = true,
   KeepAlivePeriod = Options.KeepAlivePeriodSeconds,
   HasWill = true,
   WillTopicUtf8Bytes = JobTopics.FormatWorkerHeartbeatToBytes(Options.WorkerId),
   WillPayload = new ReadOnlySequence<byte>(Options.Serializer.Serialize(offlineWill)),
   WillQualityOfService = QualityOfServiceType.AtLeastOnce,
   WillRetain = true
};

If the worker disconnects gracefully via StopAsync(), it sends a normal MQTT DISCONNECT packet, which suppresses the Will message. However, if the worker crashes or its socket aborts unexpectedly, Beskar.Mqtt.Server immediately publishes the worker's Will message to jobs/workers/{workerId}/heartbeat.

Inside JobScheduler.HandleWorkerHeartbeatAsync, the scheduler receives that IsOnline = false heartbeat, marks the worker offline in WorkerRegistry, queries Storage.GetActiveJobsByWorkerAsync(workerId) for all in-flight jobs assigned to that node, fires Events.OnWorkerCrashed, and routes every orphaned job through the retry/backoff pipeline:

// Worker reported offline (or broker fired Last Will and Testament on unexpected disconnect!)
var orphanedJobs = await Storage.GetActiveJobsByWorkerAsync(heartbeat.WorkerId, ct);
if (!transitionedToOffline && orphanedJobs.Count == 0)
{
   return;
}

if (orphanedJobs.Count > 0)
{
   JobMetrics.RecordWorkerCrash(heartbeat.WorkerId);

   if (Events.OnWorkerCrashed.Count > 0)
   {
      await Events.OnWorkerCrashed.ExecuteAsync(
         new JobSchedulerWorkerCrashedContext(heartbeat.WorkerId, orphanedJobs),
         cancellationToken: ct);
   }

   foreach (var orphan in orphanedJobs)
   {
      await HandleJobAttemptFailureAsync(
         orphan.JobId,
         $"Worker '{heartbeat.WorkerId}' disconnected unexpectedly (LWT crash recovery).",
         ct,
         orphan.Attempt,
         heartbeat.WorkerId);
   }

   SignalWakeup();
}

To make this deterministically testable in our test suite (JobRetryCancellationAndCrashTests), JobWorker even exposes a SimulateCrashAsync() method that cancels local tokens and disposes the underlying IMqttClient transport without sending an MQTT DISCONNECT packet—allowing us to verify sub-second LWT failover from worker-1 to worker-2 in integration tests.

6. Multi-stage DAG continuations, overlap policies & retained streaming

Real-world background processing rarely stops at a single isolated task. Often you need a multi-stage pipeline:

1. Stage 1 (orders queue): Validate order & reserve inventory.
2. Stage 2 (billing queue): Render invoice PDF.
3. Stage 3 (notifications queue): Email receipt to the customer.

Parent-to-child DAG continuations

In Beskar.Jobs, any job can specify a ParentJobId. When enqueued with a ParentJobId whose parent has not yet succeeded, JobScheduler holds the child job in JobStatus.Scheduled. Furthermore, IJobStorage.ClaimDueJobsAsync only claims jobs whose ParentJobId is null (or already unlocked to Enqueued).

As soon as a parent job transitions to JobStatus.Succeeded, the scheduler runs TriggerContinuationJobsAsync(parentJobId):

• All dependent children waiting on parentJobId are atomically transitioned from Scheduled to Enqueued using optimistic concurrency (Version), supporting both linear chains (Parent -> Child -> Grandchild) and parallel fan-out (1 Parent -> N Children).
• What if a client enqueues a continuation at the exact same millisecond that the parent job finishes on a worker? In JobScheduler.EnqueueAsync, we perform a double-check after inserting the child record: if the parent reached a terminal state while the child was being persisted, the scheduler unlocks (or cascades cancellation to) the child immediately so it can never be stranded.
• Conversely, if a parent job fails permanently (DeadLettered) or is Cancelled, HandleFailedOrCancelledParentContinuationsAsync recursively cascades JobStatus.Cancelled down the entire continuation tree.

Reusing continuations for Cron OverlapPolicy.QueueNext

One of my favorite design synergies in JobScheduler is how recurring Cron jobs handle overlapping runs. When a Cron schedule triggers while its previous job instance (rec.LastJobId) is still running, OverlapPolicy gives you three choices:

• AllowConcurrent: Spawn the new run immediately alongside the previous one.
• SkipIfRunning: Skip spawning a duplicate instance and advance NextRunAtUtc.
• QueueNext: Enqueue the new occurrence with ParentJobId = lastJob.JobId! By linking the new recurring instance as a continuation of the currently running instance, it automatically waits in JobStatus.Scheduled and unlocks the instant the previous run finishes.

Race-free progress streaming with WatchJobStatusAsync

Finally, how do clients observe jobs in real time without polling?

Whenever a job changes state or a handler calls await context.ReportProgressAsync(percent, message, ct), a JobStatusUpdate with a monotonically increasing SequenceNumber is published to jobs/status/{jobId} with Retain = true.

On JobClient, WatchJobStatusAsync wraps MqttClient.SubscribeStream into a clean IAsyncEnumerable<JobStatusUpdate>:

public async IAsyncEnumerable<JobStatusUpdate> WatchJobStatusAsync(
   string jobId,
   bool stopOnTerminal = true,
   [EnumeratorCancellation] CancellationToken ct = default)
{
   ArgumentException.ThrowIfNullOrWhiteSpace(jobId);

   var statusTopic = JobTopics.FormatStatus(jobId);
   var stream = MqttClient.SubscribeStream(
      statusTopic,
      bytes => Options.Serializer.Deserialize<JobStatusUpdate>(bytes.Span),
      QualityOfServiceType.AtLeastOnce,
      ct: ct);

   await foreach (var update in stream)
   {
      if (update is null) continue;

      yield return update;

      if (stopOnTerminal && update.IsTerminal)
      {
         yield break;
      }
   }
}

Because the status topic is retained on the MQTT broker, there is zero race condition between calling EnqueueGeneratedWithResultAsync and subscribing with WatchJobStatusAsync (or WaitForJobResultAsync<TResult>). Even if the worker finishes the job in 0.2 milliseconds before the client's SUBSCRIBE packet reaches the broker, the broker immediately delivers the retained Succeeded snapshot containing the serialized ResultPayload, yields it to the caller, and terminates the await foreach loop cleanly.

7. Putting it all together

Here is a complete, self-contained example spinning up the MQTT v5.0 broker, scheduler, worker, and client, enqueuing a source-generated job, and streaming its progress and strongly-typed result:

using System.Net;
using Beskar.Jobs.Client;
using Beskar.Jobs.Common.Attributes;
using Beskar.Jobs.Common.Enums;
using Beskar.Jobs.Common.Interfaces;
using Beskar.Jobs.Scheduler;
using Beskar.Jobs.Worker;
using Beskar.Mqtt.Server;

var endPoint = new IPEndPoint(IPAddress.Loopback, 18881);

// 1. Start the Beskar MQTT v5.0 Broker
var broker = MqttServerFactory.CreateBuilder()
   .WithDefaultClientIdGenerator()
   .UseTcp(endPoint.Port)
   .Build();
await broker.StartAsync();

// 2. Start the Job Scheduler
await using var scheduler = JobSchedulerBuilder.Create()
   .WithTickInterval(TimeSpan.FromMilliseconds(25))
   .UseTcp(endPoint)
   .Build();
await scheduler.StartAsync();

// 3. Start a Job Worker with our source-generated handler
await using var worker = JobWorkerBuilder.Create()
   .WithWorkerId("worker-media-01")
   .WithMaxConcurrency(4)
   .RegisterGeneratedHandlerWithResult<ImageResizeJobHandler, ImageResizeArgs, ImageResizeResult>()
   .UseTcp(endPoint)
   .Build();
await worker.StartAsync();

// 4. Connect a Job Client, enqueue a job, and stream live progress
await using var client = JobClientBuilder.Create()
   .UseTcp(endPoint)
   .Build();
await client.ConnectAsync();

var enqueueResult = await client.EnqueueGeneratedWithResultAsync<
   ImageResizeJobHandler,
   ImageResizeArgs,
   ImageResizeResult>(new ImageResizeArgs("hero-banner.png", 1920, 1080));

var jobId = enqueueResult.Success!.JobId;

await foreach (var update in client.WatchJobStatusAsync(jobId))
{
   Console.WriteLine($"[{update.Status,-11}] {update.ProgressPercent,3}% - {update.Message}");

   if (update is { Status: JobStatus.Succeeded, ResultPayload: not null })
   {
      var result = client.Options.Serializer.Deserialize<ImageResizeResult>(update.ResultPayload);
      Console.WriteLine($"Completed -> {result?.OutputPath} ({result?.OutputBytes:N0} bytes)");
   }
}

await broker.StopAsync();

public sealed record ImageResizeArgs(string FileName, int Width, int Height);
public sealed record ImageResizeResult(string OutputPath, long OutputBytes);

[GeneratedJobHandler("media.image.resize", Queue = "media")]
public sealed partial class ImageResizeJobHandler : IJobHandlerWithResult<ImageResizeArgs, ImageResizeResult>
{
   public async ValueTask<ImageResizeResult> ExecuteAsync(
      IJobExecutionContext<ImageResizeArgs> context,
      CancellationToken ct = default)
   {
      await context.ReportProgressAsync(50, $"Resizing {context.Arguments.FileName}...", ct);
      await Task.Delay(100, ct);
      return new ImageResizeResult($"/cdn/{context.Arguments.FileName}.webp", 148_920);
   }
}

Conclusion: layered architecture pays compound interest

Building Beskar.Jobs was one of those rare moments in software engineering where every low-level primitive we built earlier in the stack paid immediate compound interest:

• Because Beskar.Mqtt already had zero-allocation topic routing and native MQTT 5.0 Request-Response RPC, two-phase job claim dispatch took just a few dozen lines of coordinator code.
• Because MQTT 5.0 natively supports Retained Messages and Last Will & Testament, we achieved race-free progress streaming and instant worker crash recovery without database polling loops.
• Because Beskar.CodeGeneration and our Roslyn incremental generator patterns were already battle-tested, [GeneratedJobHandler] eliminated magic strings across the entire API surface.
• And by keeping persistence behind a clean 12-method IJobStorage interface with optimistic concurrency, you can run InMemoryJobStorage for ultra-fast ephemeral clusters or drop in PostgreSQL, SQLite, or Redis without touching a single line of job orchestration logic.

Keep your bitmasks tight, your source generators incremental, and your workers resilient!

• GitHub: github.com/MarvinDrude/Beskar.Networking
• Documentation: Beskar.Jobs Architecture & Guides
• GitHub: github.com/MarvinDrude/Beskar.Memory