Quickstart

Start here even if you have never used a job queue. Every step below is runnable, and each one says what just happened.

What this is for

Some work should not happen while a user waits: sending an email, resizing a photo, calling somebody else's API that is down half the time. So you write the work down and answer immediately, and a second program picks it up and does it.

That written-down list is a job queue. The hard part is not the list — it is what happens when a program dies holding a job, when the work fails, or when the power goes. nobroker is that list, kept in a file, with those cases handled.

What makes it unusual: there is no server to run. Normally a queue means installing Celery and operating a Redis box. Here there is a file, and nothing else.

Step 1 — get it

You need Python 3.11 or newer, and nothing else at all. Check what you have:

python --version
# Python 3.11.9   -- 3.11 or higher is fine

Then pick whichever suits you:

# the normal way
pip install nobroker

# or, with no install at all: copy the folder into your project
cp -r src/nobroker /path/to/your/project/

# or run it as one file, no install, from anywhere
python make.py build
python dist/nobroker.pyz --help

Step 2 — your first queue

Save this as first.py and run it. It is complete: nothing is left out, nothing needs to be running.

from nobroker import Queue

# A queue is a directory. It is created if it does not exist.
q = Queue("./jobs")

# Write down some work. The payload is any JSON-able value -- nobroker
# never looks inside it, it just keeps it safe and hands it back.
q.enqueue({"send_email_to": "ada@example.com"})
q.enqueue({"send_email_to": "grace@example.com"})

print(q.stats())
q.close()

$ python first.py
QueueStats(ready=2, delayed=0, leased=0, done=0, dead=0, total=2)

Step 3 — actually do the work

Now the other half: something that takes jobs off the list and runs them. Save this as work.py.

from nobroker import Queue, Worker

q = Queue("./jobs")

def handle(job):
    # Your code. Returning = success. Raising = failure, try again later.
    print("sending to", job.payload["send_email_to"])

# idle_timeout=1 makes it exit once the queue has been empty for a second.
# Leave it out and it runs forever, which is what you want in production.
Worker(q, handle, idle_timeout=1).run()

$ python work.py
sending to ada@example.com
sending to grace@example.com

Those two scripts are the whole idea. first.py is your web app; work.py is your background worker. They are separate programs, they can run on separate terminals at the same time, and the only thing between them is the ./jobs directory.

Step 4 — what happens when work fails

This is the part you are really buying. Make the handler fail:

from nobroker import Queue, Worker

q = Queue("./jobs", max_attempts=3)   # give up after 3 tries
q.enqueue({"charge": 500})

def handle(job):
    raise RuntimeError("the payment API is down")

Worker(q, handle, idle_timeout=1).run()

for job in q.dlq():                     # the jobs it gave up on
    print(job.id, job.attempts, job.last_error)

Nothing crashes and nothing is lost. The job is retried after 1 second, then 2, then 4 — waiting longer each time, so a service that is already struggling is not hammered. After three tries it is moved to the dead-letter queue: a holding area for work that will not succeed, with the error text kept so a person can look at it. Fix the bug, then put them all back with q.requeue_dead().

Step 5 — look at what is on disk

There is no magic here, and you can check that:

$ ls jobs/
default.000000.log   default.lock

$ nobroker --dir ./jobs inspect
   16  ENQUEUE  {"id": "a3f1...", "payload": {"send_email_to": "ada@..."}}
  184  LEASE    {"id": "a3f1...", "deadline": 1735689630.2}
  246  ACK      {"id": "a3f1...", "at": 1735689601.8}

One file, appended to and never edited. Every line is a thing that happened, in order. The queue's state is that file replayed from the top — which is why a crash can only ever lose the record that was being written, and why you can read your own queue with your eyes when something goes wrong at 3am.

Step 6 — from the terminal

A queue you cannot inspect from a terminal is a queue you cannot operate. Every command takes --dir to say which queue directory it means.

nobroker enqueue '{"resize": "photo.jpg"}' --priority 5
nobroker stats                  # how many in each state
nobroker list --state ready     # what is waiting
nobroker work myapp.tasks:handle --concurrency 4   # run a worker
nobroker dlq                    # what failed, and why
nobroker dlq --requeue          # after you fix the bug
nobroker inspect                # the raw log, record by record
nobroker recover                # report on torn-tail repair
nobroker compact                # drop completed history, shrink the file

nobroker work myapp.tasks:handle is worth noticing: a worker is a command, not a program you have to write. The argument is module:function.

Where to go next

python make.py demo         # the end-to-end tour
python make.py test         # the test suite
python make.py bench        # throughput on your machine
python make.py check-deps   # prove there are no dependencies

Guide

Everything nobroker does, why each piece is shaped that way, and what goes wrong if you set it badly. Read it in order the first time.

The six words you need

Every queue uses these, and they are the only vocabulary in this guide. If you have read the Quickstart you have already used four of them.

The one that is not obvious is lease, and it is the reason a queue is more than a list. Keep reading.

Jobs and payloads

A job is your data — the payload — plus everything the queue knows about how its delivery has gone: how many times it has been tried, when it may next run, what went wrong last time. nobroker never looks inside the payload; it is yours.

job = q.enqueue(
    {"resize": "photo.jpg", "width": 1024},
    priority=5,          # higher leases first; ties break FIFO
    delay=30.0,          # not eligible for 30 seconds
    max_attempts=3,      # then the dead-letter queue
    job_id="order-42",   # optional: de-duplicates on this key
)

Every argument above is optional; only the payload is required.

When enqueue() returns, the job is on disk — it fsyncs before returning. Payloads must be JSON-serialisable: dicts, lists, strings, numbers, booleans, None. No bytes, no class instances, no datetime. That is the price of a log you can read with your own eyes when something has gone wrong.

q.enqueue({"charge": 500}, job_id="order-42")   # creates it
q.enqueue({"charge": 500}, job_id="order-42")   # does nothing

Leases and the visibility timeout

Why not just remove a job when a worker takes it? Because then a worker that dies mid-job takes the job with it. It is not in the queue, it was never finished, and nobody knows it existed. That is the failure a queue is supposed to prevent.

So a worker borrows the job instead. Leasing it makes it invisible to every other worker for visibility_timeout seconds, but it stays in the queue. Ack within that window and it is done and removed. Fail to — because you crashed, hung, or were killed — and the lease simply expires and the job becomes available to somebody else.

Nothing has to detect the crash. There is no heartbeat and no health check: a lease is a deadline written on the disk, and deadlines pass on their own.

jobs = q.lease(10, visibility_timeout=60.0)
for job in jobs:
    q.extend(job, 120.0)   # buy a slow handler more time
    q.ack(job)

Retries, backoff and the DLQ

When a handler raises, the job is not lost and not immediately retried — it is put back with a delay that grows each time: 1 second, then 2, then 4, then 8. That is exponential backoff, and it exists because the usual cause of failure is something else being unhealthy, and retrying instantly makes that worse.

Jitter adds randomness to those delays. Without it, an outage that fails 10,000 jobs at once produces 10,000 retries at the same instant, then again at the same instant, forever — the queue becomes a synchronised hammer aimed at the thing that was already struggling.

After max_attempts tries the job stops retrying and moves to the dead-letter queue: not deleted, just set aside with its error message so a person can look. Attempts are counted when a job is leased, not when it fails — so a worker that dies silently still burns an attempt, and a crash-looping worker cannot retry one job forever.

from nobroker import BackoffPolicy

q = Queue("./jobs", backoff=BackoffPolicy(
    base=1.0,        # delay after the first failure
    factor=2.0,      # doubling
    max_delay=300,   # ceiling, so attempt 30 is not in 2089
    jitter=0.5,      # half fixed, half random
))

for job in q.dlq():
    print(job.id, job.attempts, job.last_error)
q.requeue_dead()      # revive the whole DLQ after fixing the bug

Priorities and delays

Two ways to control when a job runs:

q.enqueue(payload, priority=9)   # jumps the line; higher wins
q.enqueue(payload, delay=300)   # not before five minutes from now

Ordering is (priority desc, available_at asc, insertion order) — so within one priority level it is first-in-first-out, and nothing overtakes unfairly.

Use priorities sparingly. Two or three levels is plenty. There is no ageing, which means a steady stream of priority-9 work will starve a priority-0 job indefinitely — it never becomes more urgent by waiting.

Workers

Worker is the loop you would otherwise write yourself: lease a job, run the handler, ack it if it returns, nack it if it raises, repeat. Use it unless you have a reason not to — it gets three things right that hand-written loops usually get wrong (see below).

stats = Worker(
    q, handle,
    concurrency=4,        # threads — handlers here are usually I/O bound
    poll_interval=0.1,
    max_jobs=None,        # stop after N, for recycling workers
    idle_timeout=None,    # stop after N idle seconds
).run()

The three things it gets right:

1 · Ctrl-C does not lose work. SIGINT and SIGTERM flip a flag; in-flight handlers run to completion and ack normally, and only then does the process exit. Killing a worker costs you nothing, which is what makes deploying boring.

2 · Slow handlers are not redelivered. A background thread extends the lease of anything still running, so you can keep the visibility timeout short without punishing slow jobs.

3 · One bad job cannot take the worker down. A raising handler is data, not a crash; the job is nacked and the loop continues.

run() returns WorkerStats. The field to watch is lost_leases: anything above zero means handlers are finishing after their lease expired, so work is being done twice and your visibility timeout is too short.

Several workers at once

Open the same directory from as many processes as you like, on the same machine. Nothing coordinates them and no notification is sent; each process takes a kernel file lock, reads forward from where it last got to, and finds out what everyone else did. Two workers can never be handed the same job, because the lock makes the read-decide-write sequence atomic.

# terminal 1, 2, 3 — all against the same directory
nobroker --dir ./jobs work myapp.tasks:handle

Named queues

Use a second argument to keep different kinds of work apart. Each name is an independent log with its own lock, so one directory holds emails, thumbnails and webhooks side by side without contending — and a flood of thumbnails cannot delay your password-reset mail.

emails = Queue("./jobs", "emails")
thumbs = Queue("./jobs", "thumbnails")

Compaction — keeping the file small

The log is only ever appended to, so a queue that has processed a million jobs holds a million records describing work that is finished and irrelevant. Nothing removes them on its own.

Compaction rewrites the log with only the jobs that still matter. Completed jobs are dropped; the dead-letter queue is kept, because it is a record of what failed rather than maintenance debris.

result = q.compact()
print(result.describe())
# compacted 2652568 -> 16 bytes (100.0% reclaimed), kept 0 jobs, dropped 5000 completed

Putting it together

What this looks like in a real application. A web request writes the job down and returns; a separate long-running process does the work.

# tasks.py -- shared by both sides
from nobroker import Queue

def queue():
    return Queue("/var/lib/myapp/jobs", "emails", max_attempts=5)

def send_welcome(job):
    user_id = job.payload["user_id"]
    if already_sent(user_id):        # the idempotency check
        return                       # returning = done, no duplicate
    mailer.send_welcome(user_id)
    mark_sent(user_id)

# in your web app -- fast, and durable before it answers
def signup(request):
    user = create_user(request)
    tasks.queue().enqueue(
        {"user_id": user.id},
        job_id=f"welcome-{user.id}",   # cannot be scheduled twice
    )
    return redirect("/welcome")

# the worker, as a service -- systemd, a container, or just a terminal
nobroker --dir /var/lib/myapp/jobs -q emails work tasks:send_welcome -c 4

Note the two halves of the duplicate problem being handled in the two places they belong: job_id stops the same work being scheduled twice, and the already_sent check stops it being performed twice.

Common mistakes

API reference

Nine public names, every argument, and what each one is for. New to this? Read the Quickstart first — this page assumes you know what a lease is.

Which method do I want?

Queue

Queue(path, name="default", *, fsync=True,
      visibility_timeout=30.0, max_attempts=5,
      backoff=BackoffPolicy(), lock_timeout=10.0)

Adding work

enqueue(payload, *, priority=0, delay=0.0,
        max_attempts=None, job_id=None) -> Job
enqueue_many(payloads, *, priority=0, delay=0.0, …) -> list[Job]

Both fsync before returning. Raises SerializationError if the payload is not JSON-serialisable or exceeds 8 MiB.

Taking and finishing work

lease(count=1, *, visibility_timeout=None) -> list[Job]
lease_one(*, visibility_timeout=None)      -> Job | None
ack(job)                                    # -> None
nack(job, *, error=None, delay=None)     -> Job
extend(job, seconds)                        -> Job

job = q.lease_one()
if job is not None:
    try:
        do_the_work(job.payload)
        q.ack(job)                             # done, remove it
    except Exception as exc:
        q.nack(job, error=str(exc))            # retry, or DLQ if out of tries

Looking and maintaining

get(job_id)                         -> Job
stats()                             -> QueueStats
list_jobs(state=None, *, limit=None) -> list[Job]
dlq(*, limit=None)                  -> list[Job]
requeue_dead(job_id=None)           -> int
purge()                             -> int
compact()                           -> CompactionResult
close()                             # or use `with Queue(...) as q:`

print(q.stats())
# QueueStats(ready=3, delayed=1, leased=2, done=41, dead=1, total=48)

for job in q.dlq():
    print(job.id, job.attempts, job.last_error)

q.requeue_dead()          # all of them back to ready, attempts reset
q.requeue_dead("order-42")  # or just one

stats() counts by walking every job, so it is O(n) — fine to poll every second, not fine to call in a tight loop over a million jobs. purge() deletes everything, including jobs a worker is holding right now.

Job

DELAYED is deliberately not a state. A delayed job is READY with available_at in the future — one less state to keep consistent, and eligibility stays a pure function of the clock.

Worker

Worker(queue, handler, *, concurrency=1, poll_interval=0.1,
       max_jobs=None, idle_timeout=None, handle_signals=True).run()

run() blocks until stopped and returns WorkerStats(leased, succeeded, failed, dead, lost_leases). Call worker.stop() from another thread for a clean shutdown.

BackoffPolicy

BackoffPolicy(base=1.0, factor=2.0, max_delay=300.0, jitter=0.5)

How long a failed job waits before coming back. The delay for attempt n is base × factorn-1, capped at max_delay, then randomised by jitter.

from nobroker import BackoffPolicy, Queue

q = Queue("./jobs", backoff=BackoffPolicy(base=5.0, max_delay=3600))
# waits ~5s, 10s, 20s, 40s ... up to an hour

Errors

All derive from NobrokerError, so except NobrokerError catches anything the queue can raise without swallowing bugs in your own code.

A torn tail from a crash is not an error — it is repaired on open and reported through q.recovery, which has a .describe() worth logging at startup.

Design notes

The decisions, the measurements, and the three bugs worth writing down.

On disk

jobs/
  emails.000001.log     # the write-ahead log — the only durable state
  emails.current        # one line: which generation is authoritative
  emails.lock           # the cross-process lock

Everything in memory — the priority heaps, the lease table, the DLQ — is a cache of the log. There is no second source of truth to keep consistent with it, which is what makes crash recovery tractable rather than hopeful.

Replay is a pure function of the log

Nothing in the replay path reads the clock, mints a UUID, or samples jitter. Every non-deterministic value — the lease deadline, the jittered retry time, the job id — is decided once by the writer and recorded as an absolute value. Replaying a log twice therefore produces identical state, which is the property that makes the exhaustive truncation test possible at all.

Compaction without overwriting

Benchmarks

Windows 11, NVMe SSD, Python 3.11.9, via make bench. Run it on your own hardware — NVMe versus spinning rust changes the fsync rows by orders of magnitude.

Three bugs worth writing down

Peers overwrote each other's records

First run of the multi-process test: four producers × 40 jobs produced 35 jobs, not 160. Four consumers leased 615 jobs from a queue of 200. The append offset was a per-process cache that the read path never refreshed, so a peer appended on top of another's records. Invisible to every single-process test.

os.open defaults to text mode on Windows

Compaction failed roughly one run in six, losing a varying number of jobs with a checksum error in a file it had just written and fsynced. A descriptor without O_BINARY silently rewrites every 0x0A as 0x0D 0x0A. JSON escapes newlines in payloads, so it only fired when a record's binary length or CRC field happened to contain one — hence intermittent, and hence surfacing at reopen rather than at write.

Compaction left stale state in memory

The heaps were rebuilt but the job table was not, so memory held jobs the log no longer mentioned. The disagreement would have vanished on the next restart — correct after a reboot, wrong until then, which is the worst failure shape there is.

Limits

• Single machine only. Kernel file locks do not work over NFS or SMB and will not coordinate two hosts.
• At-least-once, never exactly-once. Handlers must be idempotent.
• Polling, not push. A broker-less queue has nobody to send a notification. Cheap, but not free and not zero-latency.
• The whole index lives in memory. ~500 bytes per job, so a million jobs is ~500 MB.
• stats() is O(n). It counts by iterating.
• Startup replays the whole log at ~48k records/sec. Compact regularly.
• No result backend, chaining, workflows, cron DSL, or async handlers. Deliberately out of scope.
• Payloads must be JSON-serialisable. The price of a readable log.
• Windows: directory fsync does not exist, so rename durability is weaker than on POSIX.
• Clock-dependent. A large backwards clock step can delay lease reclaims by that amount.

Objections

Answered before you have to ask them.

"Why not just use sqlite3? It's stdlib too."

SQLite would give storage and transactions. It would not give any of the things this project actually is: lease semantics, visibility timeouts, backoff with jitter, fencing tokens, a dead-letter queue, or priority ordering with delayed jobs.

A job queue on SQLite is all of this same code written on top of a query layer you do not need, plus a schema, plus the BEGIN IMMEDIATE/busy_timeout dance to stop polling workers livelocking. More code, not less.

"Why not exactly-once?"

Because nobody delivers it. A worker can complete its side effect and then die before acknowledging; the job comes back and the side effect happens twice. Closing that window requires the handler's side effect and the queue's acknowledgement to commit in a single transaction — which means the queue must live inside your database, and then it is not a general-purpose queue any more.

"Is it faster than Redis?"

No. Redis will do 50,000–100,000 ops/sec on loopback. nobroker does ~1,300 durable enqueues/sec, because it calls fsync and Redis by default does not. That is not a fair fight in either direction — nobroker is paying for a guarantee Redis is not making.

• vs Redis with appendfsync always — the same promise — you are in the same order of magnitude, minus a network hop and a server.
• vs Redis default (everysec, can lose a second of acknowledged writes) — the comparable setting is fsync=False, labelled "not durable" everywhere it appears.
• On batches, enqueue_many does 57k/sec durably.

If you need 100k jobs/sec, run a broker. If you need 1,000 jobs/sec that are still there after the power cut, this is simpler and there is nothing to operate.

"Isn't a file lock per operation slow?"

It was — 44% of a non-durable enqueue, spent opening and closing the lock file. The descriptor is now held for the queue's lifetime and only the lock is taken and released, which was a 2.4× improvement with no semantic change. Both flock and msvcrt.locking associate the lock with the open file description, so it is exactly as exclusive as reopening each time.

"What if two processes compact at once?"

They cannot. Compaction runs inside the same lock as every other operation.

"What if the machine dies during compaction?"

The pointer file either got flipped or it did not. If it did, the new generation is live and complete. If it did not, the old one is still live and untouched, and the half-written new generation is inert garbage that the next compaction overwrites.