Distributed Task Queue

A Python task-processing system built from scratch to explore the engineering behind reliable background job execution — without Celery, Redis, or RabbitMQ.

It uses multiple worker processes and a shared SQLite-backed broker, with priority scheduling, automatic retries with exponential backoff, dead-letter handling, task chaining, recurring schedules, crash recovery, and a live monitoring dashboard.

The project focuses on concurrency, coordination, failure handling, scheduling, persistence, and the design trade-offs involved in building a reliable task-processing system.

Engineering Highlights

• Multi-process worker pool for parallel task execution
• Atomic task claiming to prevent duplicate processing
• Priority-based scheduling
• Automatic retries with exponential backoff
• Dead-letter handling for permanently failed tasks
• Worker crash detection and automatic restart
• Task chaining with dependency failure propagation
• Recurring task scheduling
• CLI and live Flask monitoring dashboard

Screenshots

Live dashboard — priority queue, retries, and the dead-letter queue in action:

Task chaining + recurring schedules — a multi-step pipeline waiting on its previous step, and a recurring schedule ticking on its own:

Engineering Goals

This project was designed to explore several of the core engineering problems behind reliable background task processing:

• Parallel task execution — workers run as separate OS processes, allowing CPU-bound Python tasks to execute in parallel without sharing the Python GIL.
• Race-free task claiming — multiple workers share the same queue while an atomic database transaction prevents the same task from being claimed twice.
• Failure handling — transient failures are retried with exponential backoff, while permanently failed tasks are moved to a dead-letter state.
• Crash resilience — the manager detects and restarts worker processes that terminate unexpectedly.
• Dependency failure propagation — if a required upstream task fails permanently, dependent tasks are marked as skipped instead of waiting indefinitely.

How it works, in plain English

Think of the system as three moving parts:

1. Something submits a task. That's you, running the CLI (cli.py submit ...), clicking "Enqueue" in the web dashboard, or a recurring schedule firing on its own timer. A task is just "run this function, with these arguments" — it gets written into a shared SQLite database as a row.
2. Workers pick tasks up. A handful of worker processes are constantly asking the database "is there anything for me to do?" When one finds a task, it claims it (so no other worker can grab the same one), runs the actual Python function, and reports back whether it worked.
3. The database is the single source of truth for what happened. Success, failure, how many times something was retried, whether it's still waiting on something else — all of it lives in that one SQLite file, which is what both the CLI and the web dashboard read from.

A task moves through a small set of states over its life:

        (no dependencies)
 submit ──────────────────▶ pending ──▶ running ──▶ success
        (has dependencies)     ▲            │
 submit ──────────────────▶ waiting         └──▶ failed ──▶ (retry: back to pending)
                               │                     │
              a dependency dies│                     └──▶ (retries exhausted) ──▶ dead
                               ▼
                            skipped

• waiting → pending: only for chained tasks. A task with dependencies sits in waiting until every task it depends on has succeeded, then becomes claimable just like any other task.
• failed → pending: a transient failure (a flaky network call, for example) gets retried automatically, with an increasing delay between attempts (exponential backoff) so a struggling system isn't hammered immediately again.
• → dead: once a task has failed as many times as its max_retries allows, it stops retrying and lands in the dead-letter queue — a holding area for "this needs a human to look at it," instead of either retrying forever or silently disappearing.
• → skipped: only for chained tasks. If a task's dependency ends up dead (or itself skipped), there is no point waiting on it forever — it's marked skipped immediately so the rest of the system doesn't hang.

Architecture

                        ┌─────────────────────┐
   cli.py / dashboard ─▶│   SQLite broker      │◀── worker.py (× N processes)
   (producers)          │  tasks + schedules   │    polls, executes, reports
                        │  (WAL journal mode)  │
                        └─────────────────────┘
                                  ▲        ▲
                     spawns/      │        │ polls due schedules,
                    supervises    │        │ enqueues a Task for each
                        ┌─────────────────────┐
                        │  manager.py          │
                        │  worker pool + retry  │──▶ scheduler.py
                        │  of crashed workers   │    (background thread)
                        └─────────────────────┘

• task_queue/task.py — the Task data model and its states (see the diagram above).
• task_queue/schedule.py — the Schedule data model: "run this task every N seconds."
• task_queue/registry.py — maps a task's string name to the actual Python function, so tasks can be stored as plain JSON instead of pickled code (safer, and debuggable by just reading the database).
• task_queue/broker.py — all persistence and scheduling logic: enqueue, atomic dequeue, retry/backoff, dead-lettering, dependency promotion/skip cascades, recurring-schedule bookkeeping, stats.
• task_queue/worker.py — the loop a single worker process runs: claim a task, execute it, report the outcome.
• task_queue/manager.py — spawns the worker pool, starts the scheduler thread, handles graceful shutdown (SIGINT/SIGTERM), restarts crashed workers.
• task_queue/scheduler.py — the background thread that turns due Schedules into new Tasks (see Recurring schedules below).
• task_queue/web/ — a small Flask dashboard (polling JSON API, no websockets) to watch the queue live instead of via the CLI.
• cli.py — submit tasks (including chained ones), inspect status, list by state, requeue dead-lettered tasks, manage recurring schedules.

Why SQLite instead of Redis/RabbitMQ?

Tasks need to survive a worker crash, and multiple processes need a consistent view of the same queue — a plain queue.Queue only lives inside one process. SQLite with WAL journaling gives shared, crash-safe storage without asking anyone running this project to stand up an extra service. The one place concurrent access is genuinely tricky — atomically claiming a task so two workers can never grab the same one — is handled with an explicit transaction rather than a hand-rolled lock (see the docstring on Broker.dequeue).

Why the scheduler is a thread, not its own process

Everything else in this project that needs real concurrency (running task code) is a separate OS process, on purpose. The scheduler is different: all it does is periodically ask the database "is anything due?" and, if so, insert a row — cheap, infrequent, I/O-bound work with no CPU-heavy code running inside it. A background thread gets genuine concurrency with the worker processes for that kind of work, without asking whoever runs this project to babysit a fourth terminal window just to keep recurring tasks alive.

Getting started

python -m venv .venv && source .venv/bin/activate   # optional but recommended
pip install -r requirements.txt

# Terminal 1: start the worker pool + scheduler thread (defaults to 4 workers)
python run_workers.py --workers 4

# Terminal 2: submit some tasks
python cli.py submit add --args '[2, 3]'
python cli.py submit slow_task --kwargs '{"seconds": 3}' --priority 5
python cli.py submit flaky_task --kwargs '{"fail_probability": 0.7}'
python cli.py submit always_fails --max-retries 2

# Terminal 3: watch it happen in the browser
python task_queue/web/app.py
# -> http://127.0.0.1:5000

Or inspect everything from the CLI instead of the browser:

python cli.py stats
python cli.py list --status dead
python cli.py status <task_id>
python cli.py requeue <task_id>     # give a dead-lettered task a fresh set of attempts

Task chaining (workflows)

Submit a task with --depends-on and it won't run until every task it depends on has succeeded:

STEP1=$(python cli.py submit fetch_data --kwargs '{"source": "warehouse-db"}')
# copy the printed task id, then:
python cli.py submit process_data --depends-on <step1-task-id>

You can chain more than one dependency by comma-separating ids (--depends-on id1,id2) — the task waits until all of them succeed.

If a dependency ends up dead-lettered (retries exhausted), the waiting task is automatically marked skipped instead of waiting forever — check python cli.py list --status skipped to see anything that got cut short this way, and why, via python cli.py status <id>.

Cycles aren't a concern here: a task can only depend on ids that already exist at the moment it's created, so a chain can only ever point backwards in time — there's no graph-traversal cycle detection needed because a cycle is structurally impossible to create.

Recurring Schedules

# Runs `heartbeat` once immediately, then every 30 seconds, forever
python cli.py schedule-add heartbeat --interval 30

python cli.py schedule-list
python cli.py schedule-disable <schedule_id>   # pause without deleting
python cli.py schedule-enable <schedule_id>    # resume

Each time a schedule fires, it enqueues a brand new, ordinary Task — so a recurring task gets exactly the same retries, backoff, and dead-lettering as anything submitted by hand. Timing is checked every couple of seconds (see SCHEDULER_POLL_INTERVAL_SECONDS below), so an interval of, say, 3 seconds will drift by a second or two rather than firing on the exact millisecond — fine for the kind of periodic housekeeping this is meant for, not intended for sub-second precision.

Defining your own tasks

# examples/sample_tasks.py
from task_queue.registry import task

@task("send_email")
def send_email(to, subject):
    ...

Any process that needs to run a task (a worker, or the dashboard for its task-name dropdown) must import the module the task is defined in first, so the registry is populated — see how run_workers.py imports examples.sample_tasks before starting the pool.

Running the tests

python -m unittest discover -s tests -v

Tests cover the logic that actually matters: priority ordering, that a claimed task can't be double-claimed, backoff scheduling, the transition into the dead-letter queue, dependency promotion/skip cascades for chained tasks, and recurring-schedule firing/pausing.

Configuration

Every tunable (poll interval, retry count, backoff timing, worker count, scheduler poll interval, dashboard port) lives in task_queue/config.py and can be overridden with environment variables, e.g.:

TASKQUEUE_WORKER_COUNT=8 TASKQUEUE_MAX_RETRIES=5 python run_workers.py

Design Trade-offs and Limitations

This implementation is intentionally scoped as a single-host portfolio system rather than a production replacement for systems such as Celery or Airflow.

• SQLite's single-writer model is appropriate for this project's scope, but would become a bottleneck under high write concurrency and is not suitable as a shared broker across multiple hosts. A production-scale deployment would typically use a dedicated broker such as Redis or RabbitMQ.
• No task result expiry — completed tasks stay in the table forever. Fine for a portfolio project; a real deployment would need a cleanup job.
• No auth on the dashboard/API — it's meant to run locally or behind something else that handles auth.
• Tasks are only as safe as the code they run — this project does not sandbox task execution.
• Chains support "wait for all of these to succeed," not a full branching DAG with fan-out/fan-in visualization — you can build a DAG by chaining tasks pairwise, but there's no graph view of it in the dashboard yet.
• Recurring-schedule timing has a few seconds of jitter, bounded by the scheduler's poll interval — not meant for sub-second precision.

Future Improvements

• Swap the SQLite broker for Redis behind the same Broker interface
• Websocket-based dashboard instead of polling
• A graph view of task chains in the dashboard
• Task cancellation for anything still pending/waiting
• Basic auth in front of the dashboard/API