broker_parrot — a self-hosted job fleet on the database you already run

2026-08-02

broker_parrot is a workflow engine that treats your database as the message bus. Its pitch is one line:

A self-hosted job fleet on the database you already run — Postgres or SQLite.

No Redis, no RabbitMQ, no Celery, no separate broker to provision, monitor, back up and reason about at 3am. Work is enqueued by inserting a row. Workers claim with SELECT … FOR UPDATE SKIP LOCKED. Crashed workers heal because a lease lapses and a sweep re-queues the row. The Python package is queue_workflows; it needs Python 3.10+, and it only needs a server if you want more than one machine.

This post is about why the design lands where it does — the no-broker bet, and the specific things you have to get right to make a database a credible queue.

Local-cluster management is the point. The engine is built to squeeze a handful of heterogeneous CPU/GPU boxes you already own: turn a machine’s worker on or off on demand, load and unload models across hosts as work shifts, and let a job land on whatever capacity is free rather than being pinned to a host that’s busy or down — without standing up a cluster scheduler.

That’s Flight Deck — a custom front-end over the engine’s telemetry, showing every box side by side with an hour of thermal, power, clock and network history each, the jobs currently on them, and a park/resume toggle per queue. You don’t have to write one; queue-broker-web ships an operator panel in pure stdlib. But the telemetry is plain rows and notifications, so a bespoke view like this stays easy.

The shape of the system is three plain processes pointed at one database. Nothing talks to anything else directly; every interaction is a row written to, or read from, that database:

Why a broker is a tax you might not need

The reflex for “I need a job queue” is Redis plus a worker framework. That’s the right call at a certain scale. But it buys you a second stateful system next to the database you already run: another thing to provision, secure, fail over, and keep consistent with your domain data. The classic bug lives in the gap between the two — you commit a row, then enqueue a job, and the process dies in between. Now you have a row with no job, or a job with no row.

If the work already lives next to your data, the database can be the queue and that gap closes: the enqueue and the domain write happen in one transaction. The cost is that you have to implement the queue mechanics carefully, which is what the engine packages up.

Making SQLite the default takes the argument to its conclusion. For a single box there is no server process at all — the queue is a file, and pip install plus configure() gets you a working DAG engine. Postgres is what you graduate to when you want a second machine.

Redis and MongoDB satisfy the queue contract but not DAG dispatch — for a DAG you keep the orchestrator on a relational backend. That’s a boundary stated in the docs rather than one you discover at runtime.

The claim: SKIP LOCKED, and a wake that can’t be lost

INSERTing a row is enqueuing the work. Claiming the next job is a single atomic statement:

UPDATE workflow_node_jobs AS j
   SET status = 'running',
       started_at = now(),
       claimed_by = %(host)s,
       lease_expires_at = {lease_expr}
 WHERE j.id = (
   SELECT c.id FROM workflow_node_jobs c
    WHERE c.queue = %(queue)s
      AND c.status = 'queued'
      AND c.project = %(project)s
      AND EXISTS (SELECT 1 FROM workflow_runs r
                   WHERE r.id = c.run_id
                     AND r.status NOT IN ('cancelled', 'failed'))
    ORDER BY {order}
    FOR UPDATE SKIP LOCKED          -- the magic words
    LIMIT 1
 )
RETURNING *;

FOR UPDATE SKIP LOCKED is the load-bearing clause: each worker locks a claimable row and skips any row another worker already holds, so N workers claim N distinct jobs with zero coordination and no lost-update races.

Only {order} and {capability} are interpolated, and only from validated integers and fixed SQL fragments — never from caller-supplied strings. A queue that builds its ordering out of user input is a SQL-injection foot-gun.

The ordering itself is where the scheduling policy lives. CPU sorts by an is_priority “run next” flag, then a priority band, then FIFO. GPU inserts one extra term:

is_priority DESC,
(required_model IS NOT DISTINCT FROM current_model) DESC,   -- warm-model affinity
priority ASC,
created_at

That second line is the whole GPU story in one clause: a job whose model is already resident on this worker sorts ahead of the priority band, so the fleet routes work to the box that already paid the multi-gigabyte load cost. A negative host_priority reverses the creation tiebreak to newest-first, which turns a spare box into an overflow worker competing for the tail of the queue instead of the head.

The other half is latency. Polling in a tight loop is wasteful; instead a row trigger fires pg_notify('node_job_ready', <queue>) inside the writer’s transaction:

CREATE TRIGGER node_job_ready_notify
    AFTER INSERT OR UPDATE OF status ON workflow_node_jobs
    FOR EACH ROW EXECUTE FUNCTION notify_node_job_ready();

Because the notify rides the inserting transaction, there is no window where a row is queued but nobody was woken. A 1-second safety poll sits behind the LISTEN purely to cover a dropped notification — worst case one second of latency, never a stuck job. Lease reclaim flips a lapsed row back to queued, which fires the same trigger and re-wakes idle workers for free.

Liveness: one lease, five ways to die

A worker that claimed a job can crash, wedge, or get OOM-killed. How does the work come back?

The baseline is a lease. A renewer thread refreshes lease_expires_at every 10 seconds while the job runs; the lease itself is 600 seconds. Lease length is therefore decoupled from job duration — a 40-minute render and a 40-millisecond task use the same lease, because the lease answers only one question: is the claiming worker still alive? A dead worker stops renewing, the lease lapses, and the orchestrator’s reclaim sweep flips the row back to queued and bumps it toward the front. That sweep is the sole recovery path for an orphaned row.

A lapsed lease covers the worker that vanished. The harder case is the worker that’s alive but useless — a wedged CUDA kernel, a hung driver call. For that there are five more threads, each with a distinct exit code so an operator can diagnose from the code alone without log-diving:

Two details are worth pulling out, because they’re where naive versions of this go wrong.

The stall watchdog does not trust a single observation. Before tripping it runs a confirmation window — 3 samples, 1 second apart — and only fires if the GPU stays under 5% utilisation and container RAM moves less than ~5 GB across the whole window. A model load moves a lot of RAM while the GPU sits idle; without that second condition every cold start would look like a hang. Unconfirmed suspicions emit a stall_suspected event and re-arm rather than killing the job.

The GPU health watchdog replaces the wall clock for GPU work, and arms with a generous 1200-second first window that collapses to normal cadence after the first progress beat. This fixes the failure in both directions: a render wedged at 0% GPU trips within one window regardless of how young the job is, and a legitimately slow job is never killed for elapsed time alone.

All five exit with os._exit() — a hard process kill, not a Python exception. That’s deliberate: a wedged CUDA kernel won’t honour a cooperative cancel flag, and tearing the process down is the only thing that reliably frees the VRAM. This is also why one worker is one process holding one job — a hard exit kills exactly the hung work and nothing else.

The three watchdogs proper — 75, 76 and 78 — then funnel through a single policy point that decides re-queue versus fail, bumps watchdog_retries (default cap 3), and writes a forensic event row either way. The other two are deliberately not faults: 77 is a self-kill after someone else took the row, and 79 is an operator stop, which re-queues without spending a retry.

There’s one more layer, for the failure that in-process watchdogs structurally cannot catch: a hang that holds the GIL stops the daemon threads from running at all. So the orchestrator also sweeps for worker_heartbeats rows older than 30 seconds — three times the 10-second heartbeat cadence — that still own a running job, and flags the process as dead. An optional per-host supervisor reads those flags and restarts the container, with a cooldown so a crash loop can’t become a restart storm. The job itself was already recovered by ordinary lease reclaim; this second path exists to fix the box.

DAG dispatch and the durable outbox

Beyond standalone jobs the engine runs DAGs: a node finishes, and its downstream nodes become eligible once every dependency is completed or skipped. The tricky part is coupling the worker that finished a node to the dispatcher that fans out the next ones, without making fan-out a synchronous, failure-prone call.

The answer is the transactional outbox pattern. When a worker finalizes a node it writes, in one transaction, both the terminal status and a workflow_dispatch_events row:

with _db_connection() as conn, conn.cursor() as cur:
    row = node_queue.mark_completed_in_txn(cur, job_id, ...)
    if row is None:
        return "skipped"                     # already terminal — idempotent
    node_queue.enqueue_dispatch_event_in_txn(cur, job["run_id"],
                                             job["node_id"], "completed")

A separate orchestrator loop drains that outbox. So fan-out is retryable — a failing dispatch is retried next tick and poison-flagged after 10 attempts, which force-fails the run so an operator sees a clear failure instead of a silent stall. The worker is never blocked on downstream bookkeeping, and no event is ever lost, because the event and the state change commit together.

The DAG-walk logic itself is pure and unit-testable with no worker pool in sight; all the durability comes from “write the event in the same transaction as the state.”

Two job families, one database

The engine carries two independent job shapes, each with its own table and claim path:

Both share atomic enqueue-with-notify, lease and reclaim, and idempotent terminals. The ingest path takes a caller-supplied connection, so the enqueue rides the host’s own transaction — which is the entire reason to do this in a database:

queue_workflows.register_ingest_task("run_scenario", run_scenario)

with my_pool.connection() as conn:
    my_create_scenario(conn, scenario_id)          # your domain write
    node_queue.enqueue_ingest_job(                 # and the job
        task_name="run_scenario", queue="hydraulic",
        args={"scenario_id": scenario_id}, conn=conn,
    )                                              # commit together, or not at all

The GPU angle

GPU inference shaped the parts you don’t usually find in a job queue:

• A warm-model cache, kept resident across same-model jobs, because reloading a multi-gigabyte model per job is the dominant cost.
• Affinity routing, which promotes that cache from a per-worker optimisation to a fleet-level scheduling input — see the ORDER BY above.
• A capability gate, so a worker only claims jobs whose required_model it actually knows, with an optional lane filter splitting a GPU queue into warm-model and no-model lanes.
• Capacity-aware assignment, fitting models to available VRAM and flagging overages instead of discovering them via an OOM mid-run.
• An idle supervisor for vLLM, stopping and restarting the sidecar to reclaim VRAM so an idle server doesn’t hold a card hostage.
• An operator ON/OFF plane. Hard-stop or park a single (host, queue) worker by writing one row: it re-queues its in-flight job — redistributing it, not failing it, and with no retry penalty because an operator stop isn’t a fault — frees VRAM, and stays parked until turned back on.

queue-worker-control --queue gpu --off               # hard-stop, requeue in-flight job
queue-worker-control --queue gpu --on --host host-a  # resume, no restart needed

One broker, many projects

Apps tag themselves with a project and share a single broker database, each claiming only its own rows — so a box runs one broker and everything on it pools into the same fleet, instead of standing up a queue per service. queue-broker bootstraps and inspects it; queue-broker-web serves an operator panel in pure stdlib — no build step, no framework, no separate service:

Operating it

Three process roles, all pointed at the same database, all boring:

pip install "queue_workflows @ git+https://github.com/robertziel/broker_parrot"

queue-orchestrator                      # dispatch + outbox drain + reclaim
queue-claim-worker --queue gpu          # claim and execute
queue-scheduler                         # periodic ingest ticker

The smallest useful setup needs no server:

import queue_workflows

queue_workflows.configure()          # SQLite default — a file, nothing to run
queue_workflows.db.bootstrap()       # idempotent migration chain

Graduating to a fleet is the same call with more arguments:

queue_workflows.configure(
    db_backend="pg",
    db_url_env="MY_DB_URL",
    ingest_queues=frozenset({"ingest", "hydro"}),
    project="forecast",
)

Everything domain-specific is an injected hook with a safe default, and the config module imports nothing from the rest of the engine — so a host can extend it without the engine reaching “up” into the app. The package imports nothing from any host app either, which is enforced by a test.

For observability each host samples CPU/GPU/RAM and emits a snapshot via pg_notify('hw_metrics', …), workers upsert worker_heartbeats, a flight-recorder table logs thermal, power, clock and throttle per host, and queue depth is a SELECT … GROUP BY status.

Those are the signals behind the Flight Deck view at the top of this post. The throttle traces in it are the reason the flight recorder exists: the right-hand box there logs 85 s of power throttling and 416 s of thermal throttling in the last hour — exactly the signal that tells you a job is slow because the hardware is capping itself, not because the model is big.

One deliberate design note: the project is set up to be handed to a coding agent. There’s a single configure() seam for all host wiring, the SQLite default lets an agent verify a change end-to-end without touching infrastructure, and the test suite doubles as the behavioural contract — it refuses to run against any database whose name doesn’t end in _test.

When not to do this

Being honest about the trade: a database-as-queue is excellent up to a healthy throughput ceiling and when your jobs already live next to your data. If you’re pushing millions of tiny messages per second, or you need fan-out to many independent consumers, or your jobs have nothing to do with your relational data, a purpose-built broker earns its keep. The win here is operational: one system, one backup, one set of credentials, one transactional boundary — and SKIP LOCKED doing the heavy lifting.

Licence

AGPL-3.0 — free for personal, research and open-source use, with source-available copyleft on derivatives run as a network service. A commercial licence without the AGPL obligations is available; the contact is in the repo. For the record, versions ≤ 1.0.1 shipped MIT and 1.0.2 shipped PolyForm Noncommercial, so anything already pulled stays under the terms it came with.

Code, the full design docs and ten worked operational scenarios — power-off mid-job, boot and rejoin, a wedged GPU — are on GitHub.

open source