Jobs That Survive a Crash: A Durable Work Queue on Zaris Streams

Almost every service grows a background queue. A request comes in, you do the fast part inline — validate, write a row, return 202 Accepted — and you hand the slow part to a worker: render the PDF, charge the card, send the email, re-index the document. Enqueuing is the easy half. The hard half is the promise you have quietly made to the caller: this job will actually run, even if a worker falls over in the middle of it.

That promise is where naive queues break. The usual first cut is a list: push a job on one end, have workers pop from the other. It works beautifully until a worker pops a job, begins the work, and then the process is killed — a deploy, an OOM, a node reboot. The job was removed from the list the instant it was popped, so it exists nowhere now except in the memory of a process that no longer exists. No error, no retry, no trace. The caller got their 202 and the work silently never happened.

A durable work queue needs three properties a pop-from-a-list queue does not have: a job is delivered to exactly one worker at a time, it stays owned-but-unfinished until that worker explicitly confirms completion, and if the worker never confirms, the job becomes claimable by another worker. Zaris streams with consumer groups give you all three as first-class server-side behaviour. This post builds the queue on them, verified against the shipping native .NET API.

Why a stream, and not a list

A Zaris stream is an append-only, id-ordered log: every entry gets a monotonic id, entries are never reordered, and — unlike the other collections — the log persists even while it is being drained. On its own that is just a log. What turns it into a work queue is the consumer group.

When a group reads with the "give me new messages" cursor, the server does three things atomically: it hands each entry to exactly one consumer in the group, it advances the group's cursor so no other consumer sees that entry, and it records the entry in that consumer's pending entries list (the PEL) — a per-group ledger of "delivered, but not yet acknowledged." The entry does not leave the PEL when it is read. It leaves only when the consumer calls acknowledge. Until then, it is owned but unfinished, and the server knows exactly who owns it and for how long.

That single difference is the whole reliability story. With a list, the acknowledgement is the pop — read and commit are the same irreversible step, so a crash after the pop loses the job. With a stream group, read and acknowledge are two separate steps with the work in between, and anything still sitting in the PEL is, by definition, a job that was started and never finished. Recovering lost work becomes a question you can actually ask the store: what is still pending, and how long has it been idle?

The connection string and wiring

Nothing about the queue is special at the connection layer. Point the app at a store by name, the same as any other Zaris feature:

// appsettings.json
{
  "ConnectionStrings": {
    "Jobs": "zariss://zaris-a.example.com:7861,zaris-b.example.com:7861/?store=jobs&token=env:ZARIS_TOKEN"
  }
}

zariss:// is the TLS scheme (the extra s, same convention as https); list more than one host and the client fails over between them. Register from configuration and you get an IZarisClientProvider to resolve a store from:

using Clustron.Zaris.Client.DependencyInjection;

builder.Services.AddClustronZarisFromConnectionStrings(builder.Configuration);

The stream API lives on the full client surface, IZaris, as .Streams. Everything below goes through that one sub-client.

Enqueuing a job

Producing a job is a single append. A stream entry is an ordered set of field/value pairs — both binary — so you put whatever your worker needs to do the work into the fields. The server assigns the id; you pass * to mean "next id, please."

using System.Text;
using Clustron.Zaris.Client;

public sealed class JobProducer
{
    private readonly IZarisClientProvider _provider;
    private const string Store = "Jobs";
    private const string Stream = "jobs:email";

    public JobProducer(IZarisClientProvider provider) => _provider = provider;

    // Enqueue one job. Returns the stream id the server assigned — a durable receipt.
    public async Task<string> EnqueueAsync(string to, string template, CancellationToken ct = default)
    {
        var client = await _provider.GetAsync(Store);
        var streams = ((IZaris)client).Streams;

        var fields = new[]
        {
            new KeyValuePair<string, byte[]>("to",       Encoding.UTF8.GetBytes(to)),
            new KeyValuePair<string, byte[]>("template", Encoding.UTF8.GetBytes(template)),
        };

        // id "*" = let the server assign the next monotonic id.
        var result = await streams.AddAsync(Stream, fields, id: "*", ct: ct);
        return result.Value.ToString();   // e.g. "1730000000000-0"
    }
}

AddAsync returns the assigned StreamId as its result. That id is a real receipt: it is monotonic, it is unique, and you can hand it back to the caller or log it so a job is traceable from enqueue to completion. The append is replicated like any other Zaris write, so the job survives a node failover the moment this call returns.

The worker: read, work, acknowledge

A worker belongs to a consumer group and reads under its own consumer name. Create the group once (any worker can do it idempotently at start-up), then loop: read new messages with the > cursor, do the work, acknowledge. The three steps are deliberately separate, and the gap between step two and step three is where durability lives.

using System.Text;
using Clustron.Zaris.Client;

public sealed class EmailWorker : BackgroundService
{
    private readonly IZarisClientProvider _provider;
    private readonly IEmailSender _email;
    private const string Store    = "Jobs";
    private const string Stream   = "jobs:email";
    private const string Group    = "senders";
    private readonly string _consumer = $"sender-{Environment.MachineName}-{Guid.NewGuid():N}";

    public EmailWorker(IZarisClientProvider provider, IEmailSender email)
    {
        _provider = provider;
        _email = email;
    }

    protected override async Task ExecuteAsync(CancellationToken ct)
    {
        var client = await _provider.GetAsync(Store);
        var streams = ((IZaris)client).Streams;

        // Create the group starting at "$" (only jobs enqueued from now on).
        // mkStream: true creates the stream if the first producer has not run yet.
        // It is a no-op if the group already exists, so every worker can call it.
        await streams.GroupCreateAsync(Stream, Group, startId: "$", mkStream: true, ct: ct);

        while (!ct.IsCancellationRequested)
        {
            // ">" = deliver messages no other consumer in this group has seen.
            // Each delivered entry is recorded in THIS consumer's pending list.
            var read = await streams.ReadGroupAsync(Stream, Group, _consumer, id: ">", count: 10, ct: ct);

            if (!read.IsSuccess || read.Value.Count == 0)
            {
                // No native blocking read on the typed client yet — poll politely.
                await Task.Delay(TimeSpan.FromMilliseconds(500), ct);
                continue;
            }

            foreach (var entry in read.Value)
            {
                var to       = Encoding.UTF8.GetString(Field(entry, "to"));
                var template = Encoding.UTF8.GetString(Field(entry, "template"));

                await _email.SendAsync(to, template, ct);   // the actual work

                // Only now does the job leave the pending list. A crash before
                // this line leaves the entry claimable by another worker.
                await streams.AckAsync(Stream, Group, new[] { entry.Id.ToString() }, ct);
            }
        }
    }

    private static byte[] Field(StreamEntry entry, string name)
    {
        foreach (var f in entry.Fields)
            if (f.Key == name) return f.Value;
        return Array.Empty<byte>();
    }
}

Three things are load-bearing here:

• ReadGroupAsync with id: ">" is the dividing line between consumers. Each entry it returns has been handed to this consumer and this consumer only, and recorded in its PEL. Start ten workers in the same group and the stream is partitioned across them automatically — no coordination code, no locks, the server does the hand-out. (Pass any id other than > and you re-read this consumer's own pending history instead — useful for resuming after a transient hiccup without losing your place.)
• The work happens between read and ack, on purpose. If SendAsync throws or the process dies before AckAsync, the entry stays in the PEL. It is not redelivered to a new > read — the group's cursor has already moved past it — but it is still owned, still counted as pending, and now recoverable. This is the difference between "at-most-once" (the list) and "at-least-once" (the stream group).
• startId: "$" means "only future jobs." Use "0" instead if a freshly created group should also pick up entries already sitting in the stream. For a long-running queue, "$" is usually what you want; existing jobs are handled by whichever group was already running.

Recovering a crashed worker's jobs

The PEL is what makes recovery possible, but something has to act on it. A worker that owned a job and died will never acknowledge it, so its entries sit pending forever unless another worker takes them over. That is exactly what auto-claim does: it scans the group's pending list for entries that have been idle longer than a threshold — idle meaning delivered this long ago and still not acknowledged — and transfers ownership of them to the calling consumer, handing back the entries so it can redo the work.

Run this on a timer (or at the top of the worker loop) and stalled jobs heal themselves:

// Reclaim jobs that have been owned-but-unfinished for over a minute.
// A live worker that is simply slow will have refreshed its idle time by
// acking or re-reading; only genuinely abandoned jobs cross the threshold.
public async Task RecoverStalledAsync(IStreamClient streams, CancellationToken ct)
{
    var result = await streams.AutoClaimAsync(
        Stream, Group, _consumer,
        minIdleMs: 60_000,   // older than 60s of silence = abandoned
        start: "0",          // scan the PEL from the beginning
        count: 50,
        ct: ct);

    foreach (var entry in result.Claimed)
    {
        // These are now OURS. Redo the work, then acknowledge.
        var to       = Encoding.UTF8.GetString(Field(entry, "to"));
        var template = Encoding.UTF8.GetString(Field(entry, "template"));

        await _email.SendAsync(to, template, ct);
        await streams.AckAsync(Stream, Group, new[] { entry.Id.ToString() }, ct);
    }

    // result.Cursor is "0-0" when the scan has swept the whole PEL;
    // otherwise pass it back as `start` to continue.
}

AutoClaimAsync returns the entries it claimed, a cursor to continue a large sweep, and the ids of any entries whose underlying data had already been deleted (so you can drop them from your own tracking). The idle threshold is the one tuning knob that matters: set it comfortably longer than your slowest legitimate job, so a worker that is merely slow is never robbed of a job it is still working on. If you want to inspect the backlog before acting — how many are pending, who owns them, how long they have been idle — PendingSummaryAsync and PendingAsync report exactly that, and InfoGroupsAsync gives you the group's lag for a dashboard.

The queue, end to end

The producer appends and is done — it never waits for a worker. The group hands each job to one worker and writes it to that worker's pending list. Worker one finishes and acknowledges; the job leaves the pending list and is truly complete. Worker two crashes before it can acknowledge, so its job stays in the pending list, owned by a consumer that is gone. After the idle threshold passes, an auto-claim sweep finds it, transfers it to a living worker, and the job runs. Nothing was lost, and no job ran twice to completion — because completion is the acknowledgement, and the acknowledgement only happens once.

Trade-offs, and when to reach for something else

This is at-least-once, not exactly-once. A job can be delivered more than once — most obviously when a worker finishes the work, crashes before it acknowledges, and the entry is later reclaimed and redone. That is the correct and unavoidable trade for durability: the only way to never lose a job is to be willing to occasionally repeat one. So make your job handlers idempotent — key the side effect on the job's stream id or a natural key (an email dedup key, an INSERT ... ON CONFLICT, a charge-idempotency token) so that doing the work twice has the same effect as doing it once. If your work genuinely cannot be made idempotent, a queue is the wrong shape for it and you want a transaction.

The typed client polls; it does not block. The native .NET ReadGroupAsync returns immediately with whatever is available, so an idle worker loops with a short delay rather than parking on the server until a job arrives. That is a little less efficient than a true blocking read and adds a touch of latency floor equal to your poll interval. Keep the interval modest (a few hundred milliseconds is usually fine), or, if you are driving the queue over the Redis protocol, XREADGROUP with BLOCK is available there. For most background work — emails, renders, re-indexing — a sub-second poll is invisible.

Something has to run the recovery sweep. The PEL makes lost work recoverable, but auto-claim is not automatic: a worker has to call it. If every worker dies at once and none restarts, pending jobs wait until something comes back to claim them. In practice you run the sweep on a timer inside each worker, so as long as any worker is alive, stalled jobs heal. Set the idle threshold above your slowest legitimate job so you never reclaim work that is still in progress.

Trim the log, or it grows forever. A stream is append-only; acknowledging a job removes it from the group's pending list but does not delete the entry from the log. For a steady-state queue, cap the log with a trim — AddAsync takes a maxLen to trim in the same call, or call TrimByMinIdAsync periodically — so the stream stays bounded. Keep enough history to cover your longest possible recovery window and no more.

When a stream is the wrong tool. If you do not need durability at all — the work is cheap to lose and cheaper to recompute — a plain list queue is simpler and faster. If you need every consumer to see every message rather than dividing work among them, that is pub/sub fan-out, not a consumer group. And if "process this" really means "coordinate exclusive access to a resource," reach for a lock or a lease, not a queue. The stream group is specifically the right answer when the requirement is every job runs to completion exactly once observable, even across crashes, and the work divides across a pool of workers.

Wrap-up

The trap in background work is that enqueuing feels like the job is safe, when the real risk is everything that happens after a worker picks it up. A pop-from-a-list queue destroys the job at the moment of delivery, so a crash mid-work loses it silently — the caller's 202 was a promise the system could not keep. A Zaris stream with a consumer group keeps delivery and completion as two separate acts with the work in between: the server hands each job to exactly one worker, holds it in a pending list until that worker acknowledges, and lets another worker reclaim it if the first never does. You get at-least-once delivery, automatic work-sharing across a pool, and self-healing recovery, with no coordination code of your own — just append, read, work, acknowledge, and a periodic claim to sweep up the stragglers.

Make the handlers idempotent, bound the log with a trim, and set the idle threshold above your slowest job, and you have a work queue that keeps the promise you made when you returned that 202. The Streams sample in the Zaris samples repository has the append-and-group mechanics this post builds on, in executable form.