Broccoli

Broccoli is a Redis-backed Python task queue for running background work with:

Priority scheduling (strict priority tiers + FIFO ordering inside each tier)
Dependency-aware tasks (depends_on)
Retries + dead-letter handling
Crash/stall recovery
Multiple worker runtimes (base, threaded, async, hybrid, gpu)
CLI tooling for operational inspection and control

It is designed for teams that want Celery-like queue behavior with a smaller, explicit codebase.

Features

Queueing and scheduling

Push tasks with explicit priorities (0 is highest priority)
FIFO ordering is preserved within each priority band
Queue operations are persisted in Redis for durability

Dependency-aware execution

A task can declare depends_on=[<task_id>, ...] for fan-out/fan-in style waits
Dependent tasks are marked waiting until all listed parents complete
Dependency release is handled automatically on parent completion

Worker execution models

Single-threaded worker (BaseWorker)
Thread pool worker (ThreadedWorker)
Asyncio worker (AsyncWorker)
Hybrid worker (HybridWorker: async dispatch + threaded execution)
GPU worker (GPUWorker: hybrid worker pinned to a GPU queue/device)

Reliability and failure handling

Built-in retry handling (max_retries, default 3)
Dead-letter capture for permanently failed tasks
Requeue dead-letter tasks for replay
Stalled task recovery (manual and startup auto-recovery)

Operations and observability

Queue depth and processing stats
Dead-letter inspection CLI commands
Health checks for Redis reachability

Architecture and Redis data model

Broccoli primarily uses Redis sorted sets and hashes:

<base>:queue — runnable tasks, scored by (priority tier + FIFO sequence)
<base>:processing — in-flight tasks, scored by processing timestamp
<base>:sequence — monotonic sequence for FIFO ordering
<task_prefix>:<task_id> — task metadata hash
dependency:<task_id> — set of blocked dependents waiting for parent
<task_prefix>:dead_letter — dead-letter task IDs with failure timestamps
dl:<task_id> — dead-letter task snapshot
result:<task_id> — result records with TTL

Requirements

Python 3.11+
Redis 6+

Installation

pip install broccoli-workers

Local development install:

pip install -e .

Quick start

1) Create a producer

from broccoli.core.task.task import Task
from broccoli.core.task.task_queue import TaskQueue

queue = TaskQueue(
    redis_url="redis://localhost:6379",
    decode_responses=True,  # set False to work with raw Redis bytes
)

task = Task(task_type="send_email", payload={"to": "user@example.com"})
queue.push(task, priority=1)
print(task.task_id)

2) Create a worker and register handlers

from broccoli.workers.threaded_worker import ThreadedWorker

worker = ThreadedWorker(redis_url="redis://localhost:6379", max_workers=4)

@worker.registry.register("send_email")
def send_email(payload):
    print(f"Sending email to {payload['to']}")
    return {"status": "sent", "to": payload["to"]}

worker.start()

Core concepts

Task model

Task fields include:

task_id: UUID (auto-generated unless provided)
task_type: handler key in registry
payload: arbitrary JSON-serializable dictionary
status: pending | waiting | in_progress | completed | failed
retries: current retry count
max_retries: retry limit
depends_on: optional list of parent task IDs
result: handler output
error: error message for failed attempts

Registry model

TaskRegistry is singleton-backed, so handlers are globally visible in-process.

You can register with a decorator:

@worker.registry.register("my_task")
def my_task(payload):
    return {"ok": True}

Or manually:

worker.registry.register_manually("my_task", my_task_handler)

Task lifecycle

Push: task metadata hash is saved.
Queue placement:
- no dependency -> pending in runnable queue
- with unresolved dependency -> waiting
Pop: worker moves task to in_progress and processing set.
Execution:
- success -> completed
- failure + retries left -> pending (requeued)
- failure + retries exhausted -> failed (dead-letter eligible)
Post-processing:
- results persisted
- task hash cleanup for terminal states
- handlers invoked

Worker types

BaseWorker

Sequential task execution
Simple and predictable behavior
Useful for debugging and low-throughput workloads

ThreadedWorker

Executes tasks concurrently using ThreadPoolExecutor
Good for mixed and blocking workloads
Config: max_workers

AsyncWorker

Uses asyncio loop with bounded concurrency
Runs task handlers via executor with timeout enforcement
Config: max_concurrent

HybridWorker

Async dispatch + thread pool execution
Suitable for high-throughput mixed workloads
Config: thread_workers, async_tasks

GPUWorker

Dedicated worker for GPU workloads (gpu_tasks:queue)
Pins execution to a selected GPU via --gpu-id
Uses HybridWorker execution model with GPU cache cleanup

Task dependencies

When pushing dependent tasks:

If all parents already completed, dependent is enqueued immediately.
If one or more parents are incomplete, dependent enters waiting and is linked under each unresolved dependency:<parent_id> set.

On parent completion, waiting tasks decrement their remaining dependency count and enqueue only when all dependencies are satisfied.

Helpful APIs:

queue.get_waiting_for(parent_id)
queue.get_waiting_tasks()

Dead-letter and recovery

Dead-letter behavior

On permanent failure, task IDs are stored in <task_prefix>:dead_letter and a snapshot is persisted at dl:<task_id>.

Requeue dead-letter task

queue.requeue_dead(task_id)

CLI:

broccoli dead list
broccoli dead requeue <task_id>

Recover stalled processing tasks

Manual recovery before startup:

broccoli worker start --type threaded --recover-stalled 600

Automatic startup recovery (default enabled):

broccoli worker start --type threaded --recover-on-startup
broccoli worker start --type threaded --no-recover-on-startup

CLI reference

Global

broccoli -v ...
broccoli -vv ...

Worker startup

broccoli worker start --type threaded
broccoli worker start --type async --concurrency 20
broccoli worker start --type hybrid --thread-workers 8 --async-tasks 50
broccoli worker start --type gpu --gpu-id 0
broccoli worker start --type threaded --pool --num-workers 4

Common worker flags:

--redis-url
--queue-name
--task-prefix
--worker-id
--recover-stalled
--recover-stalled-timeout
--recover-on-startup / --no-recover-on-startup
--decode-responses / --no-decode-responses
--redis-socket-timeout
--redis-socket-connect-timeout
--redis-health-check-interval
--redis-retry-on-timeout / --no-redis-retry-on-timeout
--redis-max-connections

Queue inspection

broccoli queue stats --format table
broccoli queue stats --format json
broccoli queue list --status pending --limit 20
broccoli queue get <task_id>
broccoli queue waiting <parent_id>

Dead-letter operations

broccoli dead list
broccoli dead list --format json
broccoli dead requeue <task_id>

Health

broccoli health

Environment variables

Broccoli CLI defaults can come from environment variables:

BROCCOLI_REDIS_URL (default: redis://localhost:6379)
BROCCOLI_QUEUE_NAME (default: tasks:queue)
BROCCOLI_TASK_PREFIX (default: task)
BROCCOLI_REDIS_DECODE_RESPONSES (default: true)
BROCCOLI_REDIS_SOCKET_TIMEOUT (optional, seconds)
BROCCOLI_REDIS_SOCKET_CONNECT_TIMEOUT (optional, seconds)
BROCCOLI_REDIS_HEALTH_CHECK_INTERVAL (default: 30)
BROCCOLI_REDIS_RETRY_ON_TIMEOUT (default: true)
BROCCOLI_REDIS_MAX_CONNECTIONS (optional)

Programmatic API reference

TaskQueue

constructor: TaskQueue(redis_url=..., queue_name=..., task_prefix=..., decode_responses=True, redis_config={...})
push(task, priority=0) -> task_id
pop() -> Task | None
complete(task)
fail(task)
requeue(task_id, priority=None)
requeue_dead(task_id) -> bool
recover_stalled(timeout_seconds=3600) -> int
get_task(task_id) -> Task | None
stats() -> dict
processing_stats() -> dict
get_waiting_for(task_id) -> list[str]
get_waiting_tasks() -> list[str]
is_fully_drained() -> bool

Worker hooks

All workers support lifecycle hooks:

add_completion_handler(handler)
add_failure_handler(handler)
add_pre_process_handler(handler)
add_post_process_handler(handler)
decorator aliases: on_complete, on_failure, on_pre_process, on_post_process

Operational guidance

1. Use shared registration modules

Keep task registrations in one importable module and register into each worker instance at startup.

2. Prefer idempotent handlers

Retries can execute handlers multiple times. Side effects should be safe to repeat.

3. Set realistic timeout and recovery values

Choose recover_stalled_timeout high enough to avoid recovering long-running but valid tasks.

4. Monitor dead-letter growth

A growing dead-letter set often indicates handler bugs or dependency/data issues.

5. Separate queues by workload profile

Use distinct queue names/prefixes for high-latency jobs or tenant isolation.

Troubleshooting

Worker starts but does nothing

Ensure tasks are pushed to the same queue_name the worker consumes.
Verify task_prefix alignment between producer and worker.
Confirm handlers are registered for every task_type.

`No handler registered for task type`

Register the handler in the same process that runs the worker.
Import your registration module before calling worker.start().

Tasks remain `waiting`

Check parent task status with broccoli queue get <parent_id>.
Confirm parent was completed (not failed/deleted).

Stuck processing tasks after crash

Enable startup recovery or run --recover-stalled manually.

Redis connectivity errors

Run broccoli health.
Verify Redis URL/auth/network.

Development

Install development dependencies:

pip install -e .[dev]

Run tests:

pytest

Format/lint/type-check tools are configured in pyproject.toml (black, isort, flake8, mypy).

License

MIT License. See /home/runner/work/Broccoli/Broccoli/LICENSE.

Name		Name	Last commit message	Last commit date
backend		backend
broccoli		broccoli
extra_docs		extra_docs
frontend		frontend
.gitignore		.gitignore
Dockerfile		Dockerfile
LICENSE		LICENSE
README.md		README.md
__init__.py		__init__.py
config.py		config.py
docker-compose.yaml		docker-compose.yaml
fan_out_test.py		fan_out_test.py
publish.ps1		publish.ps1
pyproject.toml		pyproject.toml
start_worker.py		start_worker.py

Latest commit

History

Folders and files

Repository files navigation