Oxidizer ALPHA

A distributed DAG execution engine built with Python and Rust (via PyO3 / Maturin). Oxidizer orchestrates multi-tier data pipelines across workers using Redis Streams for messaging, Redis JSON for state, and S3/MinIO for config storage.

Supports both batch DAG runs and live (streaming) topologies. Live nodes use a stateless batch+republish model — each worker processes one batch, re-publishes the task back to the stream, and any worker can claim it next. Redistribution across workers is automatic; scaling requires zero coordination.

Architecture

All I/O (Redis, S3, stream reads/writes) runs in async Rust via Tokio. Python only handles business logic.

Project Structure

oxidizer/
├── examples/                   # Self-contained pipeline examples
│   ├── Dockerfile              # Shared multi-stage build
│   ├── main.py                 # Controller entrypoint
│   ├── api.py                  # API entrypoint
│   ├── skeleton/               # Minimal starter template (batch)
│   ├── batch_etl/              # E-Commerce ETL pipeline (batch)
│   ├── live_events/            # IoT sensor monitoring (live)
│   └── mixed_pipeline/         # Fintech transactions (mixed)
├── oxidizer/                   # Python package
│   ├── ui.html                 # Dashboard UI (served by Microscope)
│   └── *.py                    # Core modules
├── rust/src/                   # Rust source (PyO3 extension)
├── tests/                      # Pytest + Rust unit tests
├── Cargo.toml
├── pyproject.toml
└── requirements.txt

Each example folder contains its own docker-compose.yml, worker file(s), and YAML config. See examples/README.md for details.

Prerequisites

• Python 3.10+
• Rust 1.70+ (for local builds)
• Docker & Docker Compose (for containerized runs)
• Redis (with JSON module) and MinIO (or S3-compatible store)

Quickstart — Docker Compose

The fastest way to run the full stack (using the skeleton example):

docker compose -f examples/skeleton/docker-compose.yml up --build -d

Each example starts the following core services:

Some examples add additional workers (e.g. dedicated stream workers). See each example's docker-compose.yml for specifics.

Once running, the dashboard is at http://localhost:8000, the API at http://localhost:8000/health, the MCP endpoint at http://localhost:8000/mcp, and the MinIO console at http://localhost:9001.

To stop:

docker compose -f examples/skeleton/docker-compose.yml down

See examples/README.md for all four examples and their deployment instructions.

Quickstart — Local Development

1. Install

python -m venv .venv && source .venv/bin/activate
pip install maturin
maturin develop          # builds the Rust extension in-place
pip install -r requirements.txt

2. Controller (Oxidizer)

The controller builds a DAG from an alloy config, splits it into execution tiers, and dispatches each node to a worker stream.

from oxidizer import configure_logging
from oxidizer.catalyst import Catalyst, CatalystConnection
from oxidizer.alloy import Alloy, AlloyConnection
from oxidizer.oxidizer import Oxidizer

configure_logging()  # structlog + Rust pyo3_log bridge

catalyst = Catalyst(connection=CatalystConnection(url="localhost"))
alloy = Alloy(connection=AlloyConnection(endpoint="localhost:9000"))

oxidizer = Oxidizer(catalyst=catalyst, alloy=alloy)

# Submit a DAG run by alloy name (non-blocking — queued to the alloy stream)
oxidizer.submit_run("example")

# Start the controller loop (blocks forever, dispatches runs as they arrive)
oxidizer.start()

3. Worker (Reagent)

Workers consume tasks from the oxidizer stream. Define your processing logic with the @reagent.react() decorator:

from oxidizer.catalyst import Catalyst, CatalystConnection
from oxidizer.reagent import Reagent

catalyst = Catalyst(connection=CatalystConnection(url="localhost"))
reagent = Reagent(catalyst=catalyst)

@reagent.react()
def process(data: dict, context: dict):
    node_id = context["node_id"]
    run_id = context["run_id"]
    print(f"Processing {node_id} for run {run_id}")
    # data: dict mapping alias → resolved upstream records
    # context: run_id, node_id, alloy, layer, node_config, connections
    return {"data": data}

The decorated function receives (data, context):

• data: dict mapping alias → resolved upstream records (tiered retrieval handled by the framework)
• context: dict with keys run_id, node_id, alloy, layer, node_config, connections

For dedicated streams (e.g. routing specific nodes to specialized workers):

@reagent.react(dedicated_stream="gold_summary_stream")
def process(data: dict, context: dict):
    ...

4. UI / API / MCP (Microscope)

Microscope is a FastMCP server that exposes HTTP endpoints for dashboards and curl, plus MCP tools, resources, and prompts for AI agent integration.

from oxidizer.catalyst import Catalyst, CatalystConnection
from oxidizer.alloy import Alloy, AlloyConnection
from oxidizer.microscope import Microscope

catalyst = Catalyst(connection=CatalystConnection(url="localhost"))
alloy = Alloy(connection=AlloyConnection(endpoint="localhost:9000"))

microscope = Microscope(catalyst=catalyst, alloy=alloy)
microscope.run(host="0.0.0.0", port=8000)

See API Reference for the full list of HTTP endpoints, MCP tools, resources, and prompts.

Live Topologies

Oxidizer supports live (streaming) topologies alongside batch DAG runs. A live topology deploys long-running nodes that continuously process data from Redis Streams using a batch+republish model — each worker processes one batch, re-publishes the task, and any worker can claim it next.

# Deploy a live topology
curl -X POST http://localhost:8000/topology/deploy \
  -H 'Content-Type: application/json' \
  -d '{"alloy_name": "example"}'

# Check status
curl http://localhost:8000/topology/{run_id}/status

# Graceful stop
curl -X POST http://localhost:8000/topology/{run_id}/stop

See Live Topologies for the full architecture, node pause/unpause, rebalancing, and crash recovery details.

Documentation

Detailed documentation lives in the docs/ folder:

Testing

# Rust unit tests (no external services needed)
cargo test

# Python tests (no external services needed)
python -m pytest tests/ -v

License

TBD