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.

---

Table of contents

• Features
• Architecture and Redis data model
• Requirements
• Installation
• Quick start
• Core concepts
• Task lifecycle
• Worker types
• Task dependencies
• Dead-letter and recovery
• CLI reference
• Environment variables
• Programmatic API reference
• Operational guidance
• Troubleshooting
• Development
• License

---

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

1. Push: task metadata hash is saved.
2. Queue placement:
   • no dependency -> pending in runnable queue
   • with unresolved dependency -> waiting
3. Pop: worker moves task to in_progress and processing set.
4. Execution:
   • success -> completed
   • failure + retries left -> pending (requeued)
   • failure + retries exhausted -> failed (dead-letter eligible)
5. 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 reads only the Redis URL (which may include a password) from the environment:

• BROCCOLI_REDIS_URL, else REDIS_URL (default: redis://localhost:6379)

It is also read from a .env file in the working directory (or the path in BROCCOLI_ENV_FILE); real environment variables take precedence. Example .env:

REDIS_URL=redis://:yourpassword@localhost:6379/0

All other settings (queue name, timeouts, retries, worker counts, ...) are fields on BroccoliConfig in broccoli/config.py. Change them in code:

from broccoli.config import get_config, set_config

set_config(get_config().with_overrides(queue_name="jobs:queue", task_timeout=900))

---

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.