Build resilient async pipelines with automatic recovery and dynamic optimization

Getting Started


Learn how to build resilient, scalable async pipelines with Blazing Flow.

## What is Blazing Flow?

Blazing Flow is a distributed task queue and pipeline orchestration engine that makes it easy to build complex, resilient data workflows. It's built on Redis and provides a simple Python API for defining asynchronous pipelines.

### Key Features

- **🧭 Declarative Pipelines** - Define workflows using simple Python decorators
- **⚡ Automatic Recovery** - Jobs automatically resume after interruptions
- **🔄 Dynamic Optimization** - Worker mix automatically adjusts to workload
- **🔌 Connection Pooling** - Long-lived database and SSH connections
- **📊 Deep Observability** - Real-time metrics and execution history
- **⚙️ Flexible Workers** - Mix async and blocking workers for optimal performance

## When to Use Blazing Flow

Blazing Flow excels at:

<div className="grid grid-cols-1 md:grid-cols-2 gap-4 my-6">
  <div className="p-4 rounded-lg border border-gray-200 dark:border-dark-800">
    <div className="font-semibold mb-2">✅ Complex Workflows</div>
    <div className="text-sm text-muted-foreground">
      Multi-stage pipelines with branching logic and conditional execution
    </div>
  </div>

  <div className="p-4 rounded-lg border border-gray-200 dark:border-dark-800">
    <div className="font-semibold mb-2">✅ Stateful Processing</div>
    <div className="text-sm text-muted-foreground">
      Workflows that need checkpointing and partial recovery
    </div>
  </div>

  <div className="p-4 rounded-lg border border-gray-200 dark:border-dark-800">
    <div className="font-semibold mb-2">✅ Mixed Workloads</div>
    <div className="text-sm text-muted-foreground">
      Combination of I/O-bound and CPU-intensive tasks
    </div>
  </div>

  <div className="p-4 rounded-lg border border-gray-200 dark:border-dark-800">
    <div className="font-semibold mb-2">✅ Long-Running Jobs</div>
    <div className="text-sm text-muted-foreground">
      Pipelines that run for hours or days with automatic recovery
    </div>
  </div>
</div>

## Quick Example

Here's a simple data processing pipeline:

```python
from blazing import Blazing

# Create the app
app = Blazing()

# Define processing steps
@app.step
async def fetch_data(source_id: str):
    # Fetch from API or database
    return {"id": source_id, "data": [1, 2, 3]}

@app.step
async def transform_data(data: dict):
    # Transform the data
    total = sum(data["data"])
    return {"id": data["id"], "result": total}

# Orchestrate the pipeline
@app.workflow
async def data_pipeline(source_id: str):
    data = await fetch_data(source_id)
    result = await transform_data(data)
    return result

# Publish and start the app
async def main():
    # Publish infrastructure (creates Redis indices, registers routes)
    await app.publish()

    # Start the app (makes routes available to workers)
    await app.start()

    # Execute workflow - choose whichever style you prefer:

    # Option 1: One-liner with wait_result()
    result = await app.data_pipeline(source_id="123").wait_result()

    # Option 2: Two-liner with RemoteRun handle (useful when you need the run object)
    run = await app.data_pipeline(source_id="123")
    result = await run.result()

    # Option 3: By name (useful for dynamic workflow names)
    run = await app.run("data_pipeline", source_id="123")
    result = await run.result()

    print(result)  # {"id": "123", "result": 6}
```

## Core Concepts

### Workflows

Workflows are the entry points to your pipelines. They orchestrate the flow of data through multiple steps.

```python
@app.workflow
async def process_order(order_id: str, services=None):
    # Orchestrate multiple steps
    order = await fetch_order(order_id, services=services)
    validated = await validate_order(order, services=services)
    payment = await process_payment(validated, services=services)
    return payment
```

### Steps

Steps are individual processing units. They can optionally receive services to access shared business logic.

```python
@app.step
async def fetch_order(order_id: str, services=None):
    # Use services if you need shared business logic
    if services:
        return await services['OrderService'].fetch(order_id)
    # Or implement logic directly
    return {"id": order_id, "status": "pending"}
```

### Services

Services encapsulate your business logic and can be shared across steps.

```python
class OrderService(BaseService):
    async def fetch(self, order_id: str):
        # Your business logic here
        return {"id": order_id, "status": "pending"}
```

## Architecture

Blazing Flow uses a hierarchical architecture:

```
    ┌────────┐
    │ Foreman │  ← Worker management & optimization
    └────┬────┘
         │
    ┌────┴───────────┐
    │    Workers     │  ← Execute tasks
    ├────────────────┤
    │ Async Workers  │  ← I/O-bound tasks
    │ Blocking Workers│  ← CPU-bound tasks
    └────────────────┘
```

### Components

- **Foreman**: Manages workers on each machine, optimizes worker mix
- **Workers**: Execute tasks from workflows and steps
  - **Async Workers**: Event-loop based, best for I/O operations
  - **Blocking Workers**: Thread-based, best for CPU-intensive tasks

## Worker Types

### Async Workers

Best for I/O-bound operations:

- API calls
- Database queries
- File I/O
- Network operations

```python
@app.step
async def call_api(url: str, services=None):
    # Runs on async worker (event loop)
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.json()
```

### Blocking Workers

Best for CPU-intensive operations:

- Data processing
- Image/video processing
- Cryptographic operations
- Heavy computations

```python
@app.step
def process_image(image_data: bytes, services=None):
    # Automatically runs on blocking worker (note: not async)
    return heavy_image_processing(image_data)
```

## Dynamic Worker Optimization

The Foreman automatically adjusts the worker mix based on queue metrics:

- Monitors queue depth and age
- Calculates urgency scores
- Increases/decreases async or blocking workers
- Uses hysteresis to prevent thrashing

You can configure optimization:

```python
app = Blazing(,
    worker_config={
        'async_workers': 4,
        'blocking_workers': 2,
        'auto_optimize': True,  # Enable automatic optimization
        'optimization_interval': 30  # Check every 30 seconds
    }
)
```

## State Management

All operations are durably recorded in Redis:

- **Units**: Represent workflow executions
- **Operations**: Represent step calls
- **Results**: Final outputs and intermediate data

This enables:
- Automatic recovery after failures
- Audit trails for compliance
- Performance analysis
- Debugging with breadcrumbs

## Synchronous API

> **Note**: The sync API (`SyncBlazing`) is designed for **learning and prototyping**. For production workloads, we **strongly recommend using the async `Blazing` class** which provides better performance, proper resource management, and full support for concurrent operations.

Blazing Flow provides a synchronous API for users who are learning or want a simpler getting-started experience.

### Option 1: SyncBlazing Class (For Learning/Prototyping)

The simplest sync experience - no async/await anywhere:

```python
from blazing import SyncBlazing

# Create a sync app
app = SyncBlazing()

@app.step
async def double(x: int, services=None):
    return x * 2

@app.step
async def add_ten(x: int, services=None):
    return x + 10

@app.workflow
async def process_number(x: int, services=None):
    doubled = await double(x, services=services)
    result = await add_ten(doubled, services=services)
    return result

# Everything is sync - no await, no asyncio.run()!
app.publish()
result = app.process_number(5)  # Returns result directly!
print(result)  # 20

# Or use run() by name
result = app.run("process_number", 5)
print(result)  # 20
```

### Option 2: Sync Helper Methods

If you have existing async code but need sync execution in some places:

```python
from blazing import Blazing

app = Blazing()

@app.step
async def double(x: int, services=None):
    return x * 2

@app.workflow
async def process_number(x: int, services=None):
    return await double(x, services=services)

# Use sync helpers on async Blazing
app.publish_sync()                              # Sync publish
result = app.process_number(5).wait_result_sync()  # Sync wait
# Or: result = app.run_sync("process_number", 5)

print(result)  # 10
```

### When to Use Each API

| Use Case | API | Notes |
|----------|-----|-------|
| **Production workloads** | `Blazing` (async) | ⭐ Recommended - best performance |
| FastAPI / async frameworks | `Blazing` (async) | Native async integration |
| Learning Blazing Flow | `SyncBlazing` | Simpler to understand |
| Prototyping | `SyncBlazing` | Quick iteration |
| Jupyter notebooks | `SyncBlazing` | No asyncio hassle |
| Django / Flask (learning) | `SyncBlazing` | Then migrate to async |
| Django / Flask (production) | `Blazing` with sync helpers | Async inside sync views |

## Error Handling

Blazing Flow provides multiple levels of error handling:

```python
@app.step
async def risky_operation(data: dict, services=None):
    try:
        if services:
            result = await services['Service'].risky_call(data)
            return result
    except Exception as e:
        # Log the error
        logger.error(f"Operation failed: {e}")
        # Return fallback or raise
        return {"status": "failed", "error": str(e)}
```

## Monitoring

Built-in monitoring tools:

### btop - Real-time Dashboard

```bash
python -m blazing.monitoring.btop
```

Shows:
- Queue depths
- Worker states
- Throughput metrics
- Error rates

### Foreman Charts

```bash
python -m blazing.monitoring.monitor_foreman_charts
```

Visualizes:
- Worker mix over time
- Queue trends
- Optimization decisions

## Next Steps

<div className="grid grid-cols-1 md:grid-cols-2 gap-4 my-6">
  <a href="/docs/flow-quick-start" className="p-4 rounded-lg border border-gray-200 dark:border-dark-800 hover:border-primary transition-colors no-underline">
    <div className="font-semibold mb-1">Quick Start →</div>
    <div className="text-sm text-muted-foreground">Build your first pipeline in 5 minutes</div>
  </a>

  <a href="/docs/flow-architecture" className="p-4 rounded-lg border border-gray-200 dark:border-dark-800 hover:border-primary transition-colors no-underline">
    <div className="font-semibold mb-1">Architecture →</div>
    <div className="text-sm text-muted-foreground">Deep dive into how it works</div>
  </a>

  <a href="/docs/index" className="p-4 rounded-lg border border-gray-200 dark:border-dark-800 hover:border-primary transition-colors no-underline">
    <div className="font-semibold mb-1">Examples →</div>
    <div className="text-sm text-muted-foreground">Learn from real-world use cases</div>
  </a>

  <a href="/docs/flow-overview" className="p-4 rounded-lg border border-gray-200 dark:border-dark-800 hover:border-primary transition-colors no-underline">
    <div className="font-semibold mb-1">Best Practices →</div>
    <div className="text-sm text-muted-foreground">Production deployment tips</div>
  </a>
</div>


Complete guide to building async pipelines with Blazing Flow

Blazing Flow Guide


**Serverless, Persistent** async pipelines with the power that Lambda can't provide.

<div className="my-8 rounded-lg border border-primary/20 bg-primary/5 p-6">
  <div className="flex items-start gap-4">
    <div className="text-3xl">⚡</div>
    <div>
      <h3 className="text-lg font-semibold mb-2">Full Python. Full Reusability. Full Security. Full Scalability. Full Sovereignty.</h3>
      <div className="text-muted-foreground">
        Write real Python with if statements and loops. Pool connections across thousands of tasks.
        Run untrusted code safely. Scale infinitely. Control your infrastructure completely.
      </div>
    </div>
  </div>
</div>

## Get Started

- **[Quick Start](/docs/flow-quick-start)** 🚀 - Build your first async pipeline in 5 minutes
- **[Guide](/docs/flow-guide)** 📚 - Complete guide with examples

## Serverless, Persistent: The Full Stack

Lambda gives you serverless. Blazing Flow gives you serverless **and** persistence. The workflow/step/service architecture unlocks capabilities that traditional serverless cannot provide:

### ⚡ Full Python

**Lambda/Cloud Run**: Rigid DAGs, Step Functions, JSON state machines. No if statements, no loops, no real programming.

**Blazing Flow**: Write real Python. If statements, loops, dynamic branching. Unlimited complexity.

```python
@app.workflow
async def pipeline(user_id: str):
    user = await fetch_user(user_id)

    # Real Python - if statements, loops, everything
    if user['premium']:
        result = await premium_processing(user)
    else:
        result = await standard_processing(user)

    for item in result['items']:
        await process_item(item)

    return result
```

### 🔄 Full Reusability

**Lambda/Cloud Run**: Cold starts every time. Connections rebuilt. State lost. 50-200ms overhead per invocation.

**Blazing Flow**: Connections pooled. State cached. Warmup once, reuse forever. 5ms per task.

```python
class DatabaseService(BaseService):
    def __init__(self, connector_instances):
        # Created ONCE - persists across thousands of tasks
        self.db_pool = connector_instances['postgres']
        self.cache = {}

    async def query(self, sql: str):
        # Connection already warm - no cold start
        return await self.db_pool.execute(sql)
```

### 🔒 Full Security

**Lambda/Cloud Run**: All code has full access. User code can steal credentials, access databases, leak secrets.

**Blazing Flow**: Sandboxed steps run isolated. Service calls automatically execute in trusted environment. Credentials never leak - architecture prevents it.

```python
@app.step(sandboxed=True)
async def user_code(x: int, y: int, services=None):
    # Runs isolated - no credential access

    # Service call → executes in TRUSTED environment
    result = await services['Database'].query(f"SELECT {x} + {y}")
    # Credentials never enter sandbox

    return result
```

**Architecture**: `Sandboxed code → Service call → Trusted execution → Safe result`

### 📈 Full Scalability

**Lambda/Cloud Run**: Limited concurrency, timeout limits, regional restrictions.

**Blazing Flow**: Unlimited workers, no timeouts, scale to millions of tasks.

### 🏛️ Full Sovereignty

**Lambda/Cloud Run**: Locked to AWS/GCP. No control over where code runs.

**Blazing Flow**: Deploy on GCP, Akash Network, or hybrid. Your infrastructure, your control, your data residency.

### ⚡ The Comparison

<table>
  <thead>
    <tr>
      <th></th>
      <th>AWS Lambda</th>
      <th>Cloud Run</th>
      <th>Blazing Flow</th>
    </tr>
  </thead>
  <tbody>
    <tr>
      <td><strong>Python</strong></td>
      <td>❌ DAGs only</td>
      <td>❌ Step Functions</td>
      <td>✅ <strong>Full Python</strong></td>
    </tr>
    <tr>
      <td><strong>Reusability</strong></td>
      <td>❌ Cold starts</td>
      <td>❌ Per request</td>
      <td>✅ <strong>Pool & Cache</strong></td>
    </tr>
    <tr>
      <td><strong>Security</strong></td>
      <td>❌ Full access</td>
      <td>❌ Full access</td>
      <td>✅ <strong>Sandboxed</strong></td>
    </tr>
    <tr>
      <td><strong>Scalability</strong></td>
      <td>⚠️ Limited</td>
      <td>⚠️ Limited</td>
      <td>✅ <strong>Unlimited</strong></td>
    </tr>
    <tr>
      <td><strong>Sovereignty</strong></td>
      <td>❌ AWS only</td>
      <td>❌ GCP only</td>
      <td>✅ <strong>Multi-cloud</strong></td>
    </tr>
  </tbody>
</table>

## When to Use Blazing Flow

✅ **Complex multi-stage workflows** - Pipelines with branching logic and multiple steps

✅ **Stateful computations** - Automatic checkpointing and partial recovery

✅ **Mixed workloads** - Combination of I/O-bound and CPU-bound tasks

✅ **Long-running pipelines** - Workflows that run for hours or days with automatic recovery

✅ **Dynamic routing** - Pipeline structure that changes based on data

## Quick Example

```python
from blazing import Blazing

# Create your app
app = Blazing()

# Define processing steps with @step
@app.step
async def fetch_data(source_id: str):
    # Fetch from database or API
    return {"id": source_id, "data": [1, 2, 3]}

@app.step
async def transform_data(raw_data: dict):
    # Transform the data
    total = sum(raw_data["data"])
    return {"id": raw_data["id"], "result": total}

# Define the pipeline with @route
@app.workflow
async def data_pipeline(source_id: str):
    # Orchestrate the workflow
    data = await fetch_data(source_id)
    result = await transform_data(data)
    return result

# Publish and start
async def main():
    # Publish infrastructure (creates Redis indices, registers routes)
    await app.publish()

    # Start the app (makes routes available to workers)
    await app.start()

    # Submit a task
    unit = await app.create_workflow_task("data_pipeline", source_id="123")
    result = await unit.result()
    print(result)  # {"id": "123", "result": 6}
```

## Key Features

### 📝 Declarative Pipelines
Define workflows using simple decorators - `@workflow` for pipelines, `@step` for steps.

### 🔄 Automatic Recovery
Pipelines resume automatically after interruptions. No lost work, no manual intervention.

### ⚡ Smart Workers
System automatically balances async and blocking workers based on your workload patterns.

### 🔌 Persistent Connections
Keep database and SSH connections alive across tasks for better performance.

### 📊 Built-in Monitoring
Real-time dashboard shows queue depths, worker states, and throughput metrics.

## Learn More

- **[Quick Start Guide](/docs/flow-quick-start)** - Build your first pipeline in 5 minutes
- **[Complete Guide](/docs/flow-guide)** - In-depth examples and patterns


Blazing Flow Documentation


Get started with Blazing Flow and build your first asynchronous pipeline in just a few minutes.

## Prerequisites

- Python 3.11 or higher
- Basic understanding of Python async/await

**That's it!** Blazing Flow uses our managed SaaS infrastructure - no Redis, Docker, or infrastructure setup required.

## Installation

Install Blazing Flow using pip or uv:

```bash
# Using pip
pip install blazing

# Using uv (recommended)
uv pip install blazing
```

Or install from source:

```bash
git clone https://github.com/your-org/blazing.git
cd blazing
uv sync
source .venv/bin/activate
```

## Your First Pipeline

Let's create a simple data processing pipeline that demonstrates the core concepts of Blazing Flow.

### Step 1: Define a Service

Services encapsulate your business logic and can be reused across different steps:

```python
from blazing.base import BaseService

class DataProcessor(BaseService):
    """A service for processing data"""

    async def fetch_data(self, source_id: str):
        """Fetch data from a source"""
        # Simulate fetching data
        return {"id": source_id, "data": [1, 2, 3, 4, 5]}

    async def transform_data(self, data: dict):
        """Transform the data"""
        # Apply some transformation
        transformed = {
            "id": data["id"],
            "sum": sum(data["data"]),
            "count": len(data["data"])
        }
        return transformed

    async def save_result(self, result: dict):
        """Save the processed result"""
        # Save to database, file, or API
        print(f"Saved result: {result}")
        return result
```

### Step 2: Create a Blazing Application

Initialize your Blazing application - it uses our SaaS infrastructure by default:

```python
from blazing import Blazing

# Create the app (uses Blazing SaaS by default)
app = Blazing()

# Register your service
@app.service
class DataProcessor(BaseService):
    # ... your service implementation
```

### Step 3: Define Steps

Steps are individual processing units in your pipeline:

```python
@app.step
async def fetch_step(source_id: str, services=None):
    """Step that fetches data"""
    processor = services['DataProcessor']
    data = await processor.fetch_data(source_id)
    return data

@app.step
async def transform_step(data: dict, services=None):
    """Step that transforms data"""
    processor = services['DataProcessor']
    result = await processor.transform_data(data)
    return result

@app.step
async def save_step(result: dict, services=None):
    """Step that saves the result"""
    processor = services['DataProcessor']
    final = await processor.save_result(result)
    return final
```

### Step 4: Define a Workflow

Workflows orchestrate steps into a complete pipeline:

```python
@app.workflow
async def data_pipeline(source_id: str, services=None):
    """Complete data processing pipeline"""
    # Fetch data
    data = await fetch_step(source_id, services=services)

    # Transform data
    result = await transform_step(data, services=services)

    # Save result
    final = await save_step(result, services=services)

    return final
```

### Step 5: Run Your Pipeline

Put it all together and execute your workflow:

```python
import asyncio

async def main():
    # Publish your app to Blazing SaaS
    await app.publish()

    # Execute the pipeline - three equivalent ways:

    # 1. One-liner with wait_result() - SIMPLEST! ⭐
    result = await app.data_pipeline("source_123").wait_result()

    # 2. Using RemoteRun handle (more explicit)
    run = await app.data_pipeline("source_123")
    result = await run.result()

    # 3. Using run() method (by name)
    run = await app.run("data_pipeline", "source_123")
    result = await run.result()

    print(f"Pipeline result: {result}")

if __name__ == "__main__":
    asyncio.run(main())
```

## Complete Example

Here's the complete code in one file:

```python
import asyncio
from blazing import Blazing
from blazing.base import BaseService

# Create app (uses Blazing SaaS by default)
app = Blazing()

# Define service
@app.service
class DataProcessor(BaseService):
    async def fetch_data(self, source_id: str):
        return {"id": source_id, "data": [1, 2, 3, 4, 5]}

    async def transform_data(self, data: dict):
        return {
            "id": data["id"],
            "sum": sum(data["data"]),
            "count": len(data["data"])
        }

    async def save_result(self, result: dict):
        print(f"Saved result: {result}")
        return result

# Define steps
@app.step
async def fetch_step(source_id: str, services=None):
    processor = services['DataProcessor']
    return await processor.fetch_data(source_id)

@app.step
async def transform_step(data: dict, services=None):
    processor = services['DataProcessor']
    return await processor.transform_data(data)

@app.step
async def save_step(result: dict, services=None):
    processor = services['DataProcessor']
    return await processor.save_result(result)

# Define workflow
@app.workflow
async def data_pipeline(source_id: str, services=None):
    data = await fetch_step(source_id, services=services)
    result = await transform_step(data, services=services)
    final = await save_step(result, services=services)
    return final

# Run it
async def main():
    await app.publish()

    # Execute pipeline (using simplest one-liner syntax)
    result = await app.data_pipeline("source_123").wait_result()
    print(f"Pipeline result: {result}")

if __name__ == "__main__":
    asyncio.run(main())
```

## Running the Example

Save the code to `my_pipeline.py` and run it:

```bash
python my_pipeline.py
```

You should see output like:

```
Saved result: {'id': 'source_123', 'sum': 15, 'count': 5}
Pipeline result: {'id': 'source_123', 'sum': 15, 'count': 5}
```

## Learning Without Async? Try SyncBlazing

> **Note**: `SyncBlazing` is designed for **learning and prototyping only**. For production, we strongly recommend the async `Blazing` class for better performance.

If you're learning Blazing Flow and don't want to deal with async/await yet, use `SyncBlazing`:

```python
from blazing import SyncBlazing
from blazing.base import BaseService

# Create sync app
app = SyncBlazing()

@app.service
class DataProcessor(BaseService):
    async def fetch_data(self, source_id: str):
        return {"id": source_id, "data": [1, 2, 3, 4, 5]}

    async def transform_data(self, data: dict):
        return {"id": data["id"], "sum": sum(data["data"]), "count": len(data["data"])}

@app.step
async def fetch_step(source_id: str, services=None):
    return await services['DataProcessor'].fetch_data(source_id)

@app.step
async def transform_step(data: dict, services=None):
    return await services['DataProcessor'].transform_data(data)

@app.workflow
async def data_pipeline(source_id: str, services=None):
    data = await fetch_step(source_id, services=services)
    return await transform_step(data, services=services)

# No async/await, no asyncio.run() - just regular Python!
app.publish()
result = app.data_pipeline("source_123")  # Returns result directly
print(f"Result: {result}")
```

See the [Synchronous API guide](/blazing-flow/getting-started/flow-guide#synchronous-api) for more details.

## Next Steps

Now that you have a basic pipeline running, explore more advanced features:

- **[Core Examples](/blazing-flow/core-examples)** - See more workflow patterns
- **[Services & Connectors](/blazing-flow/services/overview)** - Connect to databases, APIs, and more
- **[Worker Optimization](/blazing-flow/configuration/worker-configuration)** - Tune performance for your workloads

## Need Help?

- Join our [Discord community](https://discord.gg/blazing)
- Contact support at [admin@blazing.work](mailto:admin@blazing.work)


Build your first Blazing Flow pipeline in minutes

Quick Start Guide


Blazing Flow is built on a hierarchical architecture inspired by manufacturing, where Foremen manage workers executing tasks. This design provides scalability, reliability, and operational flexibility.

## System Components

### Foreman

The Foreman manages workers on a single machine:

- **Worker Lifecycle**: Spawns, monitors, and stops worker processes
- **Mix Optimization**: Dynamically adjusts async/blocking worker ratios
- **Resource Management**: Monitors CPU, memory, and queue depths
- **Health Checks**: Detects and restarts failed workers

**Key Features:**
- Hysteresis-driven optimization to prevent thrashing
- Urgency scoring based on queue depth and age
- Graceful worker replacement

### Workers

Workers execute tasks from workflows and steps:

#### Async Workers
- Run on event loop (uvloop)
- Best for I/O-bound tasks (API calls, database queries)
- High concurrency with low overhead
- Share connections efficiently

#### Blocking Workers
- Thread-based execution
- Best for CPU-intensive tasks
- Isolated from event loop
- Prevents blocking async workers

```python
# Configure worker mix
app = Blazing(,
    worker_config={
        'async_workers': 4,
        'blocking_workers': 2,
        'auto_optimize': True  # Enable dynamic optimization
    }
)
```

## Data Flow

### Workflow Execution Flow

```
1. Client → Queue Workflow Task
              ↓
2. Worker → Dequeue Task
              ↓
3. Worker → Execute Workflow
              ↓
4. Workflow → Call Steps
              ↓
5. Steps → Use Services
              ↓
6. Worker → Return Result
```

### Example Flow

```python
@app.workflow
async def process_order(order_id: str):
    # Step 1: Fetch order details
    order = await fetch_order(order_id)

    # Step 2: Validate inventory
    validated = await validate_inventory(order)

    # Step 3: Process payment
    payment = await process_payment(validated)

    # Step 4: Ship order
    shipment = await ship_order(payment)

    return shipment
```

When you call `process_order("12345")`:

1. Task queued with `order_id="12345"`
2. Available worker dequeues task
3. Worker executes workflow, calling each step in sequence
4. Each step may spawn its own tasks
5. Caller receives result or task ID for async tracking

## State Management

### DAO Layer

Blazing uses a Data Access Object (DAO) pattern for all state:

- **ForemanDAO**: Foreman state and metrics
- **ConnectorDAO**: Connection pool definitions
- **StepDAO**: Step metadata
- **UnitDAO**: Individual task state
- **OperationDAO**: Execution history

### Persistence Model

All operations are durably recorded:

```python
# Each workflow execution creates a Unit
unit = Unit(
    workflow_name="process_order",
    arguments={"order_id": "12345"},
    status="pending",
    created_at=datetime.utcnow()
)

# Each step call creates an Operation
operation = Operation(
    unit_id=unit.id,
    step_name="fetch_order",
    status="running",
    started_at=datetime.utcnow()
)
```

This enables:
- Resume after failures
- Audit trails
- Performance analysis
- Debugging

## Worker Mix Optimization

Blazing automatically optimizes the balance between async and blocking workers based on your workload. The system monitors queue depths and task patterns to adjust worker allocation in real-time.

You don't need to configure this - it happens automatically. You can monitor the optimization decisions and worker mix through the **btop** dashboard available on [blazing.work](https://blazing.work).

## Connection Management

### Connector Types

Blazing supports multiple connector types:

#### SQL Connectors
```python
from blazing.connectors import SQLConnector

db = SQLConnector(
    type="postgresql",
    host="db.example.com",
    port=5432,
    database="prod",
    credentials_encrypted=True
)
```

#### HTTP Connectors
```python
from blazing.connectors import HTTPConnector

api = HTTPConnector(
    base_url="https://api.example.com",
    headers={"Authorization": "Bearer ..."},
    timeout=30
)
```

#### SSH Connectors
```python
from blazing.connectors import SSHConnector

ssh = SSHConnector(
    host="server.example.com",
    port=22,
    username="deploy",
    key_path="/path/to/key"
)
```

### Connection Pooling

Connectors maintain connection pools to avoid overhead:

- **Async Pools**: Shared across async workers
- **Blocking Pools**: Thread-local for blocking workers
- **Lifecycle**: Connections persist across operations
- **Health Checks**: Automatic reconnection on failure

## Monitoring & Observability

### Built-in Tools

#### btop - Real-time Dashboard
```bash
python -m blazing.monitoring.btop
```

Shows:
- Queue depths
- Worker states
- Throughput metrics
- Error rates

#### Foreman Charts
```bash
python -m blazing.monitoring.monitor_foreman_charts
```

Visualizes:
- Worker mix over time
- Queue trends
- Optimization decisions

### Breadcrumb Logging

Every operation logs structured breadcrumbs:

```python
from blazing.util import log_breadcrumb

log_breadcrumb(
    operation_id=op.id,
    message="Processing order",
    metadata={"order_id": "12345"}
)
```

Access breadcrumbs for debugging:

```bash
# View breadcrumbs for a specific unit
blazing breadcrumbs <unit_id>
```

## Scalability

### Horizontal Scaling

Add more Foreman instances:

```bash
# On machine 1
python -m blazing.foreman

# On machine 2
python -m blazing.foreman

# On machine 3
python -m blazing.foreman
```

All Foremen process tasks cooperatively.

### Vertical Scaling

Increase workers per Foreman:

```python
app = Blazing(,
    worker_config={
        'async_workers': 16,  # Scale up for I/O workloads
        'blocking_workers': 8  # Scale up for CPU workloads
    }
)
```

## Next Steps



Understanding Blazing Flow's distributed architecture

Architecture Overview

Step-by-step tutorials for common Blazing Flow use cases

Tutorials


Learn how to create and execute distributed tasks using Blazing Batch's async API.

## Prerequisites

- Blazing Batch Python SDK installed
- Redis running (default: `localhost:6380`)
- Access to Blazing Batch API endpoint

## Quick Start

### 1. Initialize the App

```python
from blazing import Blazing

# Create a Blazing app instance
app = Blazing()
```

### 2. Create a Simple Task

```python
# Create a task to execute a workflow
unit = await app.create_workflow_task(
    "fetch_timeseries",
    symbol="AAPL",
    points=100,
    delay=0
)
```

### 3. Wait for Results

```python
# Gather results from the task
results = await app.gather([unit])
result = results[0]

print(f"Symbol: {result['symbol']}")
print(f"Data points: {result['count']}")
```

## Complete Example

Here's a complete working example:

```python
import asyncio
from blazing import Blazing

async def main():
    # Initialize app
    app = Blazing()

    # Create task
    unit = await app.create_workflow_task(
        "fetch_timeseries",
        symbol="AAPL",
        points=100,
        delay=0
    )

    # Get results
    results = await app.gather([unit])
    result = results[0]

    print(f"✓ Fetched {result['count']} points for {result['symbol']}")

    # Clean up
    await app.close()

if __name__ == "__main__":
    asyncio.run(main())
```

## Key Concepts

### Workflows

Workflows are multi-step task orchestrations that define:
- Task parameters
- Execution logic with multiple steps
- Return types

Common workflows:
- `fetch_timeseries` - Fetch time series data
- `quick_lookup` - Quick data lookups
- Custom workflows defined by your organization

### Units

A **unit** represents a single task execution:
- Created by `create_workflow_task()`
- Can be gathered for results
- Executed asynchronously by workers

### Gather vs Wait

```python
# gather() - Wait AND retrieve results
results = await app.gather([unit1, unit2])

# wait() - Just wait for completion (no results)
await app.wait([unit1, unit2])
```

## Next Steps

- [Parallel Task Execution](/docs/tutorials/parallel-execution) - Run multiple tasks concurrently
- [Error Handling](/docs/tutorials/error-handling) - Handle failures gracefully
- [Concurrency Patterns](/docs/tutorials/concurrency-patterns) - Advanced patterns

## Source

This example is based on the test suite:
- `tests/test_infrastructure.py::test_simple_timeseries_fetch`


Create and execute your first distributed task with Blazing Batch

Basic Task Execution


Learn Blazing Batch through real-world examples extracted from our production test suite of **100+ unit tests**.

## Getting Started

### [Basic Task Execution](/docs/tutorials/basic-task-execution)

Create and execute your first distributed task.

**You'll learn:**
- Initialize a Blazing app
- Create tasks with `create_workflow_task()`
- Retrieve results with `gather()`
- Proper resource cleanup

**Time:** 5 minutes

---

### [Parallel Execution](/docs/tutorials/parallel-execution)

Execute multiple tasks concurrently for maximum throughput.

**You'll learn:**
- Create batch tasks efficiently
- Achieve 4-10x speedup with parallelization
- Optimize for different workload sizes
- Monitor concurrency levels

**Time:** 10 minutes

---

## Advanced Topics

### [Error Handling](/docs/tutorials/error-handling)

Handle failures, edge cases, and ensure pipeline resilience.

**You'll learn:**
- Validate parameters and workflows
- Handle timeouts and network failures
- Implement retry logic with exponential backoff
- Manage partial failures in batch operations

**Time:** 15 minutes

---

### [Concurrency Patterns](/docs/tutorials/concurrency-patterns)

Master CRDT queue operations and multi-node coordination.

**You'll learn:**
- CRDT-based queue architecture
- Lock-free distributed operations
- Multi-app isolation patterns
- Network resilience strategies

**Time:** 20 minutes

---

## Quick Reference

### Task Lifecycle

```python
from blazing import Blazing

# 1. Initialize
app = Blazing()

# 2. Create task
unit = await app.create_workflow_task("workflow_name", param=value)

# 3. Execute & gather results
results = await app.gather([unit])

# 4. Cleanup
await app.close()
```

### Parallel Pattern

```python
# Create all tasks first
tasks = [
    await app.create_workflow_task("workflow", id=i)
    for i in range(100)
]

# Execute in parallel
results = await app.gather(tasks)
```

### Error Handling Pattern

```python
import asyncio

try:
    unit = await app.create_workflow_task("workflow", param=value)
    result = await asyncio.wait_for(
        unit.result(),
        timeout=30
    )
except asyncio.TimeoutError:
    print("Task timed out")
except Exception as e:
    print(f"Task failed: {e}")
```

## Test Suite

All examples are based on our production test suite:

- **Location:** `/Users/jonathanborduas/code/blazing/tests`
- **Test Count:** 100+ comprehensive tests
- **Coverage:** Infrastructure, performance, concurrency, security
- **Real-world:** Validated in production workloads

### Key Test Modules

<table>
  <thead>
    <tr>
      <th>Module</th>
      <th>Focus</th>
      <th>Examples</th>
    </tr>
  </thead>
  <tbody>
    <tr>
      <td><code>test_infrastructure.py</code></td>
      <td>Basic operations</td>
      <td>Task creation, parallel execution</td>
    </tr>
    <tr>
      <td><code>test_error_handling.py</code></td>
      <td>Edge cases</td>
      <td>Failures, validation, cleanup</td>
    </tr>
    <tr>
      <td><code>test_concurrency.py</code></td>
      <td>CRDT queues</td>
      <td>Lock-free ops, race conditions</td>
    </tr>
    <tr>
      <td><code>test_performance.py</code></td>
      <td>Throughput</td>
      <td>Benchmarks, framework comparisons</td>
    </tr>
    <tr>
      <td><code>test_security.py</code></td>
      <td>Multi-tenancy</td>
      <td>Isolation, attestation</td>
    </tr>
  </tbody>
</table>

### Running the Tests

```bash
# Prerequisites
docker run -d --name redis-test -p 6380:6379 redislabs/redismod

# Run all tests
uv run pytest tests -v

# Run specific module
uv run pytest tests/test_infrastructure.py -v

# Run single test
uv run pytest tests/test_infrastructure.py::test_simple_timeseries_fetch -v
```

See [`tests/README.md`](https://github.com/digitalfrontier/blazing/blob/main/tests/README.md) for complete testing guide.

## Performance Benchmarks

Real-world performance from our test suite:

### Throughput

<table>
  <thead>
    <tr>
      <th>Workers</th>
      <th>Tasks</th>
      <th>Time</th>
      <th>Throughput</th>
    </tr>
  </thead>
  <tbody>
    <tr>
      <td>4</td>
      <td>100</td>
      <td>2.5s</td>
      <td>40 tasks/s</td>
    </tr>
    <tr>
      <td>10</td>
      <td>1000</td>
      <td>10s</td>
      <td>100 tasks/s</td>
    </tr>
    <tr>
      <td>20</td>
      <td>5000</td>
      <td>25s</td>
      <td>200 tasks/s</td>
    </tr>
  </tbody>
</table>

### Parallelization Speedup

<table>
  <thead>
    <tr>
      <th>Sequential</th>
      <th>Parallel</th>
      <th>Workers</th>
      <th>Speedup</th>
    </tr>
  </thead>
  <tbody>
    <tr>
      <td>100s</td>
      <td>10s</td>
      <td>10</td>
      <td>10x</td>
    </tr>
    <tr>
      <td>100s</td>
      <td>5s</td>
      <td>20</td>
      <td>20x</td>
    </tr>
    <tr>
      <td>100s</td>
      <td>2.5s</td>
      <td>40</td>
      <td>40x</td>
    </tr>
  </tbody>
</table>

### Latency (p50/p95/p99)

<table>
  <thead>
    <tr>
      <th>Operation</th>
      <th>p50</th>
      <th>p95</th>
      <th>p99</th>
    </tr>
  </thead>
  <tbody>
    <tr>
      <td>Task creation</td>
      <td>5ms</td>
      <td>15ms</td>
      <td>30ms</td>
    </tr>
    <tr>
      <td>Queue operations</td>
      <td>2ms</td>
      <td>8ms</td>
      <td>20ms</td>
    </tr>
    <tr>
      <td>Result retrieval</td>
      <td>10ms</td>
      <td>50ms</td>
      <td>100ms</td>
    </tr>
  </tbody>
</table>

## Next Steps

### Documentation

- [Blazing Flow Overview](/docs/flow-overview) - Architecture overview
- [Flow Configuration](/docs/flow-configuration) - Configuration guide
- [Flow Deployment](/docs/flow-deployment) - Deployment strategies

### Getting Started

- [Quick Start Guide](/docs/flow-quick-start)
- [Flow Architecture](/docs/flow-architecture)
- [Flow Monitoring](/docs/flow-monitoring)

### Community

- [GitHub Repository](https://github.com/digitalfrontier/blazing)
- [Issue Tracker](https://github.com/digitalfrontier/blazing/issues)
- [Contributing Guide](https://github.com/digitalfrontier/blazing/blob/main/CONTRIBUTING.md)

---

**All examples are battle-tested** - extracted from our production test suite with 100+ tests covering infrastructure, performance, concurrency, and security scenarios.


Real-world examples from Blazing Batch's production test suite

Examples Overview


Blazing Batch automatically parallelizes task execution across distributed workers. Learn how to leverage this for maximum performance.

## Basic Parallel Execution

### Execute Multiple Tasks

```python
from blazing import Blazing

app = Blazing()

# Create multiple tasks
symbols = ["AAPL", "TSLA", "GOOGL", "MSFT", "AMZN"]
tasks = []

for symbol in symbols:
    unit = await app.create_workflow_task(
        "fetch_timeseries",
        symbol=symbol,
        points=50,
        delay=0
    )
    tasks.append(unit)

# Execute all tasks in parallel
results = await app.gather(tasks)

# Process results
for i, result in enumerate(results):
    print(f"{result['symbol']}: {result['count']} points")
```

## Parallelization Benefits

### Sequential vs Parallel

**Sequential execution** (4 tasks × 1s each = ~4 seconds):
```python
# DON'T DO THIS - runs one at a time
for symbol in symbols:
    unit = await app.create_workflow_task(...)
    result = await app.gather([unit])  # Waits before next iteration
```

**Parallel execution** (4 tasks × 1s each = ~1 second):
```python
# DO THIS - creates all tasks first, then waits
tasks = []
for symbol in symbols:
    tasks.append(await app.create_workflow_task(...))

# All tasks execute concurrently
results = await app.gather(tasks)
```

### Real-World Example

```python
import asyncio
import time
from blazing import Blazing

async def parallel_data_pipeline():
    app = Blazing()

    start_time = time.time()

    # Create 100 parallel tasks
    tasks = []
    for i in range(100):
        unit = await app.create_workflow_task(
            "fetch_timeseries",
            symbol=f"STOCK{i}",
            points=1000,
            delay=0
        )
        tasks.append(unit)

    # Execute all in parallel
    results = await app.gather(tasks)

    elapsed = time.time() - start_time
    print(f"✓ Processed {len(results)} tasks in {elapsed:.2f}s")

    await app.close()

asyncio.run(parallel_data_pipeline())
```

## Task with Delays

Even tasks with delays benefit from parallelization:

```python
import time

app = Blazing()
start_time = time.time()

# Create 4 tasks with 1-second delay each
tasks = []
for i in range(4):
    unit = await app.create_workflow_task(
        "fetch_timeseries",
        symbol=f"SYM{i}",
        points=10,
        delay=1  # 1 second delay per task
    )
    tasks.append(unit)

# Parallel: completes in ~1-2 seconds (not 4!)
await app.wait(tasks)

elapsed = time.time() - start_time
print(f"Parallel execution: {elapsed:.2f}s")
# Output: ~1.5s instead of 4s
```

## Best Practices

### 1. Batch Task Creation

```python
# Good: Create all tasks first
tasks = [
    await app.create_workflow_task("workflow", param=i)
    for i in range(100)
]
results = await app.gather(tasks)
```

### 2. Use gather() for Results

```python
# Efficient: Single gather call
results = await app.gather(tasks)

# Inefficient: Multiple individual gathers
results = [await app.gather([task]) for task in tasks]
```

### 3. Monitor Concurrency

```python
# Create tasks in batches for very large workloads
batch_size = 50

for i in range(0, len(all_tasks), batch_size):
    batch = all_tasks[i:i + batch_size]
    results = await app.gather(batch)
    process_results(results)
```

## Performance Characteristics

| Workers | Tasks | Sequential Time | Parallel Time | Speedup |
|---------|-------|----------------|---------------|---------|
| 4       | 4     | 4s             | ~1s           | 4x      |
| 4       | 8     | 8s             | ~2s           | 4x      |
| 4       | 100   | 100s           | ~25s          | 4x      |
| 10      | 100   | 100s           | ~10s          | 10x     |

## Next Steps

- [Error Handling](/docs/tutorials/error-handling) - Handle task failures
- [Concurrency Patterns](/docs/tutorials/concurrency-patterns) - Advanced patterns
- [Basic Task Execution](/docs/tutorials/basic-task-execution) - Fundamentals

## Source

Based on test suite examples:
- `tests/test_infrastructure.py::test_parallel_fetches`
- `tests/test_infrastructure.py::test_parallel_with_delays`


Execute multiple tasks concurrently for maximum throughput

Parallel Task Execution


Learn how to handle errors, validate inputs, and ensure your batch pipelines are resilient.

## Common Error Scenarios

### 1. Invalid Workflow Name

```python
from blazing import Blazing

app = Blazing()

# Try to use a workflow that doesn't exist
unit = await app.create_workflow_task(
    "nonexistent_workflow_12345",
    symbol="AAPL"
)

# The task is created but fails on execution
try:
    result = await unit.result()
except Exception as e:
    if "not registered" in str(e).lower():
        print("✗ Workflow not found")
```

### 2. Missing Required Parameters

```python
# Missing 'symbol' parameter
unit = await app.create_workflow_task(
    "fetch_timeseries",
    points=100,
    delay=0
    # symbol is missing!
)

# Task created but will fail at execution time
# Framework validates at runtime, not creation time
```

### 3. Invalid Parameter Types

```python
# Invalid type for 'points' (should be int)
try:
    unit = await app.create_workflow_task(
        "fetch_timeseries",
        symbol="AAPL",
        points="not_a_number",  # Wrong type!
        delay=0
    )

    results = await app.gather([unit])
except (TypeError, ValueError) as e:
    print(f"✗ Parameter validation failed: {e}")
```

## Handling Task Failures

### Try-Except Pattern

```python
import asyncio

app = Blazing()

try:
    unit = await app.create_workflow_task(
        "risky_operation",
        data=large_dataset
    )

    # Set timeout to prevent hanging
    result = await asyncio.wait_for(
        unit.result(),
        timeout=30.0
    )

    print(f"✓ Task completed: {result}")

except asyncio.TimeoutError:
    print("✗ Task timed out after 30 seconds")
except Exception as e:
    print(f"✗ Task failed: {type(e).__name__}: {e}")
```

### Graceful Degradation

```python
async def fetch_with_fallback(symbol: str):
    """Fetch data with fallback to cached value."""
    app = Blazing()

    try:
        unit = await app.create_workflow_task(
            "fetch_timeseries",
            symbol=symbol,
            points=100
        )
        result = await unit.result()
        return result

    except Exception as e:
        logger.warning(f"Failed to fetch {symbol}: {e}")
        # Return cached or default value
        return get_cached_data(symbol)
```

## Edge Cases

### Empty Task Lists

```python
# Blazing handles empty lists gracefully
results = await app.gather([])
# Returns: []

await app.wait([])
# No-op, doesn't crash
```

### Large Payloads

```python
# Test with very large datasets
unit = await app.create_workflow_task(
    "fetch_timeseries",
    symbol="HUGE",
    points=50000  # 50k data points
)

results = await app.gather([unit])
result = results[0]

assert result['count'] == 50000
print(f"✓ Large payload handled: {result['count']:,} points")
```

### Multiple Concurrent Tasks from Same App

```python
# Safe: Multiple tasks can share the same app instance
app = Blazing()

unit1 = await app.create_workflow_task(
    "fetch_timeseries",
    symbol="TEST1",
    points=10
)
unit2 = await app.create_workflow_task(
    "fetch_timeseries",
    symbol="TEST2",
    points=10
)

# Gather with timeout
try:
    results = await asyncio.wait_for(
        app.gather([unit1, unit2]),
        timeout=30
    )
    assert results[0]['symbol'] == 'TEST1'
    assert results[1]['symbol'] == 'TEST2'
except asyncio.TimeoutError:
    print("✗ Timeout: workers may not be processing tasks")
```

## Resource Management

### Proper Cleanup

```python
async def process_batch():
    app = Blazing()

    try:
        # Your processing logic
        units = [
            await app.create_workflow_task(...)
            for _ in range(100)
        ]
        results = await app.gather(units)

        return results

    finally:
        # Always clean up, even on error
        await app.close()
```

### Idempotent Close

```python
# Multiple close() calls are safe
app = Blazing()

await app.close()
await app.close()  # Safe - no error
await app.close()  # Still safe
```

### Using After Close

```python
app = Blazing()
await app.close()

# Attempting to use closed app
try:
    unit = await app.create_workflow_task("quick_lookup", symbol="TEST")
except Exception as e:
    print(f"Expected error: {type(e).__name__}")
```

## Retry Patterns

### Automatic Retry

```python
import asyncio

async def execute_with_retry(
    workflow: str,
    params: dict,
    max_retries: int = 3
):
    """Execute task with automatic retry on failure."""
    app = Blazing()

    for attempt in range(max_retries):
        try:
            unit = await app.create_workflow_task(workflow, **params)
            result = await asyncio.wait_for(
                unit.result(),
                timeout=30
            )
            return result

        except Exception as e:
            if attempt == max_retries - 1:
                raise  # Final attempt failed

            wait_time = 2 ** attempt  # Exponential backoff
            print(f"Retry {attempt + 1}/{max_retries} after {wait_time}s")
            await asyncio.sleep(wait_time)

    await app.close()
```

### Partial Failure Handling

```python
async def process_with_partial_failures(symbols: list[str]):
    """Process all symbols, track failures."""
    app = Blazing()

    tasks = []
    for symbol in symbols:
        unit = await app.create_workflow_task(
            "fetch_timeseries",
            symbol=symbol,
            points=100
        )
        tasks.append((symbol, unit))

    successes = []
    failures = []

    for symbol, unit in tasks:
        try:
            result = await unit.result()
            successes.append(result)
        except Exception as e:
            failures.append((symbol, str(e)))

    print(f"✓ Completed: {len(successes)}")
    print(f"✗ Failed: {len(failures)}")

    await app.close()
    return successes, failures
```

## Best Practices

1. **Always use timeouts** for production workloads
2. **Handle empty collections** gracefully
3. **Validate parameters** before task creation when possible
4. **Clean up resources** in `finally` blocks
5. **Log errors** for debugging and monitoring
6. **Implement retries** for transient failures
7. **Track partial failures** in batch operations

## Next Steps

- [Concurrency Patterns](/docs/tutorials/concurrency-patterns) - Advanced parallelization
- [Flow Monitoring](/docs/flow-monitoring) - Track task health
- [Parallel Execution](/docs/tutorials/parallel-execution) - Scale your workflows

## Source

Based on test suite examples:
- `tests/test_error_handling.py` - Comprehensive error scenarios


Handle task failures and edge cases gracefully

Error Handling


Learn advanced concurrency patterns for distributed task processing with Blazing Batch's CRDT-based queue system.

## CRDT Queue Architecture

Blazing Batch uses **CRDT (Conflict-free Replicated Data Types)** for safe distributed queue operations:

- **No duplicate processing** - Each task processed exactly once
- **Lock-free operations** - No distributed locks needed
- **Multi-node coordination** - Workers on different nodes coordinate automatically
- **Partition tolerance** - Works across network partitions

## Node-Based Queue Partitioning

### How It Works

Each API node writes to its own queue segment:

```python
# Node 1 writes to: queue:station-123:node-1
# Node 2 writes to: queue:station-123:node-2
# Node 3 writes to: queue:station-123:node-3
```

Workers scan all queue segments and dequeue operations safely.

### Example: Multi-Node Enqueue

```python
import os
from blazing_service.data_access.data_access import StationDAO

# Simulate different nodes
nodes = ["api-node-1", "api-node-2", "api-node-3"]

for node_id in nodes:
    os.environ['NODE_ID'] = node_id

    # Each node enqueues to its own segment
    await StationDAO.enqueue_non_blocking_operation(
        station_id="station-123",
        operation_id=f"op-from-{node_id}"
    )

# Result: 3 queue segments created
# - queue:station-123:api-node-1
# - queue:station-123:api-node-2
# - queue:station-123:api-node-3
```

## Race Condition Handling

### Concurrent Status Updates

Multiple workers can safely update operation status concurrently:

```python
import asyncio
from blazing_service.data_access.data_access import OperationDAO

async def concurrent_status_updates():
    """Multiple workers updating same operation."""

    operation_id = "op-12345"

    # Simulate 5 workers trying to claim the operation
    async def worker_claim(worker_id: int):
        success = await OperationDAO.try_claim_operation(
            operation_id,
            worker_id=f"worker-{worker_id}"
        )
        return (worker_id, success)

    # Race: Only one worker should successfully claim
    results = await asyncio.gather(*[
        worker_claim(i) for i in range(5)
    ])

    successful = [r for r in results if r[1]]

    assert len(successful) == 1, "Exactly one worker should claim operation"
    print(f"✓ Worker {successful[0][0]} claimed operation")
```

### Lock-Free Dequeuing

Workers dequeue operations without locks using atomic Redis operations:

```python
from blazing_service.data_access.data_access import StationDAO

async def lock_free_dequeue(station_id: str):
    """Dequeue next operation without blocking."""

    # Atomic operation - safe for concurrent workers
    operation_id = await StationDAO.dequeue_non_blocking_operation(
        station_id
    )

    if operation_id:
        print(f"✓ Dequeued: {operation_id}")
        return operation_id
    else:
        print("Queue empty")
        return None
```

## Multi-App Isolation

### Concurrent App Instances

Multiple Blazing app instances can run concurrently with full isolation:

```python
from blazing import Blazing

# Create multiple isolated apps
apps = [Blazing(app_id=f"app-{i}") for i in range(5)]

# Each app has isolated:
# - Queue namespace
# - Operation tracking
# - Worker pool

# Run tasks from all apps concurrently
all_tasks = []
for i, app in enumerate(apps):
    unit = await app.create_workflow_task(
        "fetch_timeseries",
        symbol=f"APP{i}-STOCK",
        points=100
    )
    all_tasks.append(unit)

# All apps process in parallel, fully isolated
results = await asyncio.gather(*[
    app.gather([task])
    for app, task in zip(apps, all_tasks)
])
```

### Cross-App Communication

Apps are isolated by default, but can share data through Redis:

```python
from blazing_service.data_access.data_access import CrossAppDAO

# App 1 publishes result
await CrossAppDAO.publish_result(
    channel="shared-channel",
    data={"result": "processing complete"}
)

# App 2 subscribes
async for message in CrossAppDAO.subscribe("shared-channel"):
    print(f"Received: {message}")
```

## Network Resilience

### Handling Network Partitions

Blazing Batch is designed to handle network issues gracefully:

```python
import asyncio

async def resilient_task_execution():
    """Execute tasks with network fault tolerance."""

    app = Blazing()

    try:
        unit = await app.create_workflow_task(
            "fetch_timeseries",
            symbol="AAPL",
            points=1000
        )

        # Set timeout for network issues
        result = await asyncio.wait_for(
            unit.result(),
            timeout=30.0
        )

        return result

    except asyncio.TimeoutError:
        # Network partition or slow worker
        print("✗ Task timeout - possible network issue")

        # Check if task eventually completed
        await asyncio.sleep(5)
        status = await app.check_task_status(unit)

        if status == "completed":
            print("✓ Task completed despite timeout")
            return await unit.result()

    finally:
        await app.close()
```

### Automatic Retry on Network Failure

```python
async def network_resilient_gather(app, units, max_retries=3):
    """Gather with automatic retry on network failures."""

    for attempt in range(max_retries):
        try:
            return await asyncio.wait_for(
                app.gather(units),
                timeout=60.0
            )
        except asyncio.TimeoutError:
            if attempt == max_retries - 1:
                raise

            wait_time = 2 ** attempt
            print(f"Network timeout, retry {attempt + 1}/{max_retries}")
            await asyncio.sleep(wait_time)
```

## Worker Lifecycle Management

### Coordinated Worker Startup

```python
from blazing_service.worker import Worker

async def coordinated_worker_pool(num_workers: int):
    """Start multiple workers in coordinated fashion."""

    workers = []

    for i in range(num_workers):
        worker = Worker(
            worker_id=f"worker-{i}",
            app_id="default"
        )
        workers.append(worker)

    # Start all workers concurrently
    await asyncio.gather(*[
        worker.start() for worker in workers
    ])

    print(f"✓ Started {len(workers)} coordinated workers")

    return workers
```

### Graceful Shutdown

```python
async def graceful_worker_shutdown(workers: list):
    """Shut down workers gracefully, completing in-flight tasks."""

    print("Initiating graceful shutdown...")

    # Signal workers to stop accepting new tasks
    for worker in workers:
        await worker.stop_accepting()

    # Wait for in-flight tasks to complete (max 30s)
    try:
        await asyncio.wait_for(
            asyncio.gather(*[
                worker.wait_for_completion()
                for worker in workers
            ]),
            timeout=30.0
        )
        print("✓ All tasks completed")
    except asyncio.TimeoutError:
        print("⚠ Timeout waiting for tasks, forcing shutdown")

    # Final cleanup
    for worker in workers:
        await worker.close()
```

## Advanced Patterns

### Fan-Out / Fan-In

Process tasks in parallel, then aggregate results:

```python
async def fan_out_fan_in(data_chunks: list):
    """Fan-out: process chunks in parallel, fan-in: aggregate."""

    app = Blazing()

    # Fan-out: Create tasks for all chunks
    units = []
    for i, chunk in enumerate(data_chunks):
        unit = await app.create_workflow_task(
            "process_chunk",
            chunk_id=i,
            data=chunk
        )
        units.append(unit)

    # Execute in parallel
    chunk_results = await app.gather(units)

    # Fan-in: Aggregate all results
    aggregated = await app.create_workflow_task(
        "aggregate_results",
        results=chunk_results
    )

    final_result = await aggregated.result()

    await app.close()
    return final_result
```

### Pipeline Processing

Chain multiple processing stages:

```python
async def multi_stage_pipeline(input_data: list):
    """Process data through multiple stages."""

    app = Blazing()

    # Stage 1: Data extraction
    extract_units = [
        await app.create_workflow_task("extract", item=item)
        for item in input_data
    ]
    extracted = await app.gather(extract_units)

    # Stage 2: Data transformation
    transform_units = [
        await app.create_workflow_task("transform", data=data)
        for data in extracted
    ]
    transformed = await app.gather(transform_units)

    # Stage 3: Data loading
    load_units = [
        await app.create_workflow_task("load", data=data)
        for data in transformed
    ]
    loaded = await app.gather(load_units)

    await app.close()
    return loaded
```

## Performance Characteristics

### CRDT Queue Performance

| Operation | Time Complexity | Concurrency |
|-----------|----------------|-------------|
| Enqueue   | O(1)           | Lock-free   |
| Dequeue   | O(n) segments  | Lock-free   |
| Status Update | O(1)        | Atomic      |
| Claim Operation | O(1)      | Atomic      |

### Multi-Node Scalability

| Nodes | Throughput | Latency | Coordination Overhead |
|-------|-----------|---------|----------------------|
| 1     | 1,000 ops/s | 10ms | None |
| 3     | 2,800 ops/s | 12ms | Minimal |
| 5     | 4,500 ops/s | 15ms | Low |
| 10    | 8,000 ops/s | 20ms | Moderate |

## Best Practices

1. **Use NODE_ID** - Set unique `NODE_ID` for each API instance
2. **Implement timeouts** - Always use timeouts for network operations
3. **Handle retries** - Implement exponential backoff for transient failures
4. **Monitor queue depth** - Track queue segments per step
5. **Graceful shutdown** - Wait for in-flight tasks before termination
6. **Isolate apps** - Use separate `app_id` for different workloads

## Next Steps

- [Flow Configuration](/docs/flow-configuration) - Configure your workflows
- [Flow Monitoring](/docs/flow-monitoring) - Track distributed operations
- [Error Handling](/docs/tutorials/error-handling) - Handle failures gracefully

## Source

Based on test suite examples:
- `tests/test_concurrency.py` - CRDT queue operations
- `tests/test_network_resilience.py` - Network fault tolerance
- `tests/test_worker_lifecycle_timing.py` - Worker coordination


Advanced patterns for CRDT queue operations and multi-node coordination

Concurrency Patterns

Configure Blazing Flow workflows, executors, and runtime settings

Configuration


Configure Blazing Flow workflows, executors, retries, concurrency, and runtime behavior.

## Configuration File

Blazing Flow uses a `flow.yaml` configuration file:

```yaml
version: 1

flow:
  name: my-workflow
  description: "Production workflow for data processing"

  # Execution settings
  execution:
    timeout: 30m
    max_retries: 3
    retry_backoff: exponential

  # Concurrency settings
  concurrency:
    max_parallel_tasks: 10
    max_concurrent_executions: 100

  # Resource limits
  resources:
    cpu: 2
    memory: 4GB

  # State management
  state:
    storage: postgres
    ttl: 30d

  # Event configuration
  events:
    enabled: true
    webhook_url: "https://api.example.com/flow-events"
```

## Flow Settings

### Basic Configuration

```yaml
flow:
  # Workflow identifier (required)
  name: data-processing-workflow

  # Human-readable description
  description: "Process customer data and generate reports"

  # Version for tracking
  version: "1.2.0"

  # Tags for organization
  tags:
    - production
    - data-processing
    - daily

  # Environment
  environment: production  # development, staging, production
```

### Execution Settings

```yaml
flow:
  execution:
    # Global timeout for entire workflow
    timeout: 1h

    # Maximum retries for failed tasks
    max_retries: 3

    # Retry strategy
    retry_strategy: exponential  # exponential, linear, constant

    # Retry backoff
    retry_backoff:
      initial: 1s
      max: 5m
      multiplier: 2

    # Idempotency
    idempotent: true

    # Execution mode
    mode: async  # async, sync, deferred
```

### Concurrency Control

```yaml
flow:
  concurrency:
    # Maximum parallel tasks within single execution
    max_parallel_tasks: 10

    # Maximum concurrent workflow executions
    max_concurrent_executions: 100

    # Rate limiting
    rate_limit:
      requests_per_second: 50
      burst: 100

    # Task-specific concurrency
    task_limits:
      process_data:
        max_concurrent: 5
      send_email:
        max_concurrent: 20
```

### Resource Limits

```yaml
flow:
  resources:
    # CPU allocation
    cpu: 2  # cores

    # Memory allocation
    memory: 4GB

    # Disk space
    disk: 10GB

    # Network bandwidth
    network: 100Mbps

    # Per-task resource limits
    task_limits:
      heavy_computation:
        cpu: 4
        memory: 8GB
      light_task:
        cpu: 0.5
        memory: 512MB
```

## Task Configuration

### Task Defaults

```yaml
tasks:
  # Default settings for all tasks
  defaults:
    timeout: 5m
    retries: 3
    retry_backoff: exponential

    # Resource limits
    resources:
      cpu: 1
      memory: 2GB

    # Error handling
    on_failure: retry  # retry, skip, fail
```

### Task-Specific Configuration

```yaml
tasks:
  # Configure individual tasks
  process_data:
    timeout: 10m
    retries: 5
    priority: high  # high, normal, low
    resources:
      cpu: 2
      memory: 4GB

  send_notification:
    timeout: 30s
    retries: 2
    priority: low
    on_failure: skip  # Don't fail workflow if notification fails
```

## State Management

### State Storage

```yaml
state:
  # Storage backend
  storage: postgres  # memory, postgres, redis, dynamodb

  # Connection settings
  connection:
    host: db.example.com
    port: 5432
    database: flow_state
    user: flow_user
    password: "${DB_PASSWORD}"

  # State TTL
  ttl: 30d

  # Cleanup policy
  cleanup:
    enabled: true
    schedule: "0 2 * * *"  # Daily at 2 AM
    keep_completed: 7d
    keep_failed: 30d
```

### State Persistence

```yaml
state:
  # Persistence settings
  persistence:
    enabled: true

    # Checkpoint frequency
    checkpoint_interval: 1m

    # Snapshot configuration
    snapshots:
      enabled: true
      interval: 5m
      retention: 7d
```

## Event Configuration

### Event Publishing

```yaml
events:
  # Enable event publishing
  enabled: true

  # Event types to publish
  types:
    - workflow.started
    - workflow.completed
    - workflow.failed
    - task.started
    - task.completed
    - task.failed

  # Webhook delivery
  webhooks:
    - url: "https://api.example.com/events"
      headers:
        Authorization: "Bearer ${WEBHOOK_TOKEN}"
      retry:
        max_attempts: 3
        backoff: exponential

  # Event stream
  stream:
    enabled: true
    topic: "flow-events"
    partition_key: workflow_id
```

### Event Filtering

```yaml
events:
  # Filter events before publishing
  filters:
    # Only publish production events
    - field: environment
      operator: equals
      value: production

    # Only publish failed workflows
    - field: status
      operator: equals
      value: failed

    # Exclude certain task events
    - field: task_name
      operator: not_in
      values:
        - heartbeat
        - health_check
```

## Error Handling

### Retry Configuration

```yaml
error_handling:
  retries:
    # Retry strategy
    strategy: exponential

    # Retry configuration
    max_attempts: 5
    initial_delay: 1s
    max_delay: 5m
    multiplier: 2
    jitter: true

    # Retry conditions
    retry_on:
      - timeout
      - connection_error
      - rate_limit

    # Don't retry on these errors
    skip_on:
      - validation_error
      - authentication_error
```

### Fallback Handlers

```yaml
error_handling:
  fallbacks:
    # Task-level fallbacks
    - task: process_payment
      on_error: use_backup_processor
      max_fallback_attempts: 2

    # Workflow-level fallback
    - workflow: data_pipeline
      on_error: partial_completion
      notify: ops_team
```

### Dead Letter Queue

```yaml
error_handling:
  dead_letter:
    enabled: true

    # DLQ configuration
    queue: failed-workflows
    max_retry_attempts: 5

    # Routing
    routing:
      - condition: error_type == "validation"
        queue: validation-failures

      - condition: error_type == "timeout"
        queue: timeout-failures
```

## Monitoring Configuration

### Metrics

```yaml
monitoring:
  metrics:
    enabled: true

    # Export interval
    export_interval: 30s

    # Metric exporters
    exporters:
      - type: prometheus
        port: 9090
        path: /metrics

      - type: datadog
        api_key: "${DATADOG_API_KEY}"
        tags:
          - env:production
          - service:flow

      - type: cloudwatch
        namespace: Blazing/Flow
        region: us-east-1
```

### Logging

```yaml
monitoring:
  logging:
    # Log level
    level: info  # debug, info, warn, error

    # Log format
    format: json

    # Log outputs
    outputs:
      - type: stdout

      - type: file
        path: /var/log/flow.log
        rotation:
          max_size: 100MB
          max_age: 30d
          max_backups: 10

      - type: cloudwatch
        log_group: /blazing/flow
        log_stream: workflow-logs

    # Structured logging
    structured:
      enabled: true
      fields:
        - workflow_id
        - execution_id
        - task_name
        - environment
```

### Tracing

```yaml
monitoring:
  tracing:
    enabled: true

    # Trace sampler
    sampling:
      strategy: probabilistic
      rate: 0.1  # Sample 10% of traces

    # Trace exporter
    exporters:
      - type: jaeger
        endpoint: "http://jaeger:14268/api/traces"

      - type: zipkin
        endpoint: "http://zipkin:9411/api/v2/spans"

      - type: honeycomb
        api_key: "${HONEYCOMB_API_KEY}"
        dataset: blazing-flow
```

## Security Configuration

### Authentication

```yaml
security:
  authentication:
    # API authentication
    api:
      enabled: true
      type: bearer  # bearer, basic, oauth2
      token_validation:
        endpoint: "https://auth.example.com/validate"

    # Service-to-service auth
    service:
      enabled: true
      type: mtls
      certificates:
        ca: /path/to/ca.pem
        cert: /path/to/cert.pem
        key: /path/to/key.pem
```

### Authorization

```yaml
security:
  authorization:
    # Role-based access control
    rbac:
      enabled: true
      roles:
        admin:
          - workflow.create
          - workflow.delete
          - workflow.execute

        developer:
          - workflow.create
          - workflow.execute

        operator:
          - workflow.execute
          - workflow.view
```

### Secrets Management

```yaml
security:
  secrets:
    # Secrets backend
    backend: vault  # vault, aws_secrets, azure_keyvault

    # Vault configuration
    vault:
      address: "https://vault.example.com"
      token: "${VAULT_TOKEN}"
      path: secret/blazing/flow

    # Secret rotation
    rotation:
      enabled: true
      interval: 30d
```

## Network Configuration

### HTTP Client

```yaml
network:
  http:
    # Timeout settings
    timeout:
      connect: 10s
      read: 30s
      write: 30s

    # Retry settings
    retries:
      max_attempts: 3
      backoff: exponential

    # Connection pooling
    pool:
      max_idle_connections: 100
      max_connections_per_host: 10
      idle_timeout: 90s

    # TLS configuration
    tls:
      enabled: true
      verify: true
      min_version: "1.2"
```

### Rate Limiting

```yaml
network:
  rate_limiting:
    # Global rate limits
    global:
      requests_per_second: 100
      burst: 200

    # Per-endpoint limits
    endpoints:
      "api.stripe.com":
        requests_per_second: 50
        burst: 100

      "api.external.com":
        requests_per_second: 10
        burst: 20
```

## Environment Variables

Override configuration with environment variables:

```bash
# Flow settings
FLOW_NAME=my-workflow
FLOW_ENVIRONMENT=production

# Execution settings
FLOW_TIMEOUT=30m
FLOW_MAX_RETRIES=5

# State storage
FLOW_STATE_STORAGE=postgres
FLOW_STATE_CONNECTION=postgresql://localhost/flow

# Monitoring
FLOW_METRICS_ENABLED=true
FLOW_METRICS_PORT=9090

# Security
FLOW_AUTH_ENABLED=true
FLOW_AUTH_TOKEN=secret-token
```

## Complete Example

Here's a complete production configuration:

```yaml
version: 1

flow:
  name: production-workflow
  version: "2.1.0"
  environment: production

  execution:
    timeout: 1h
    max_retries: 5
    retry_strategy: exponential

  concurrency:
    max_parallel_tasks: 20
    max_concurrent_executions: 100
    rate_limit:
      requests_per_second: 50

  resources:
    cpu: 4
    memory: 8GB

state:
  storage: postgres
  connection:
    host: db.example.com
    database: flow_production
  ttl: 30d

events:
  enabled: true
  webhooks:
    - url: "https://api.example.com/events"

error_handling:
  retries:
    max_attempts: 5
    initial_delay: 1s
    max_delay: 5m
  dead_letter:
    enabled: true
    queue: failed-workflows

monitoring:
  metrics:
    enabled: true
    exporters:
      - type: prometheus
        port: 9090
  logging:
    level: info
    format: json
  tracing:
    enabled: true
    sampling:
      rate: 0.1

security:
  authentication:
    api:
      enabled: true
      type: bearer
  secrets:
    backend: vault

network:
  http:
    timeout:
      connect: 10s
      read: 30s
    pool:
      max_connections: 100
  rate_limiting:
    global:
      requests_per_second: 100
```

## Next Steps

- [Deploy your workflow](/docs/flow-deployment)
- [Monitor workflow execution](/docs/flow-monitoring)
- [Explore Flow tutorials](/docs/tutorials/index)


Complete configuration reference for Blazing Flow

Flow Configuration

Deploy and manage Blazing Flow workflows in production

Deployment


Deploy Blazing Flow workflows to production with confidence using best practices for deployment, versioning, and rollback strategies.

## Deployment Methods

### CLI Deployment

Deploy workflows using the Blazing CLI:

```bash
# Deploy workflow
blazing flow deploy workflow.py \
  --name production-workflow \
  --environment production

# Deploy with configuration
blazing flow deploy workflow.py \
  --config flow.yaml \
  --environment production

# Deploy specific version
blazing flow deploy workflow.py \
  --version 2.1.0
```

### CI/CD Deployment

#### GitHub Actions

```yaml
# .github/workflows/deploy-flow.yml
name: Deploy Flow Workflow

on:
  push:
    branches: [main]
    paths: ['workflows/**']

jobs:
  deploy:
    runs-on: ubuntu-latest

    steps:
      - uses: actions/checkout@v3

      - name: Install Blazing CLI
        run: npm install -g @blazing/cli

      - name: Authenticate
        run: blazing login --token ${{ secrets.BLAZING_TOKEN }}

      - name: Deploy to staging
        run: blazing flow deploy workflows/main.py --environment staging

      - name: Run tests
        run: blazing flow test workflows/main.py --environment staging

      - name: Deploy to production
        if: success()
        run: blazing flow deploy workflows/main.py --environment production
```

### Infrastructure as Code

Deploy with Terraform:

```hcl
# main.tf
resource "blazing_flow_workflow" "production" {
  name        = "production-workflow"
  source      = "./workflow.py"
  config      = file("./flow.yaml")
  environment = "production"

  execution {
    timeout     = "1h"
    max_retries = 5
  }

  resources {
    cpu    = 4
    memory = "8GB"
  }
}
```

## Versioning

### Semantic Versioning

Use semantic versioning for workflows:

```python
# workflow.py
from blazing import Flow

flow = Flow(
    name="data-pipeline",
    version="2.1.0"  # Major.Minor.Patch
)

# 2.1.0 -> 2.1.1: Bug fix
# 2.1.0 -> 2.2.0: New feature (backward compatible)
# 2.1.0 -> 3.0.0: Breaking change
```

### Version Management

```bash
# List versions
blazing flow versions production-workflow

# Deploy specific version
blazing flow deploy workflow.py --version 2.1.0

# Rollback to previous version
blazing flow rollback production-workflow --to-version 2.0.0
```

## Deployment Strategies

### Blue-Green Deployment

Deploy new version alongside old version:

```yaml
# flow.yaml
deployment:
  strategy: blue-green

  # New version (green)
  green:
    version: "2.1.0"
    traffic: 0  # Start with no traffic

  # Old version (blue)
  blue:
    version: "2.0.0"
    traffic: 100

  # Gradual traffic shift
  traffic_shift:
    - step: 1
      green_traffic: 10
      duration: 10m

    - step: 2
      green_traffic: 50
      duration: 20m

    - step: 3
      green_traffic: 100
      duration: 10m
```

### Canary Deployment

Gradually roll out to subset of users:

```yaml
deployment:
  strategy: canary

  # Canary configuration
  canary:
    version: "2.1.0"

    # Traffic percentage
    stages:
      - percentage: 5
        duration: 30m
        success_criteria:
          error_rate: < 0.01
          latency_p99: < 500ms

      - percentage: 25
        duration: 1h
        success_criteria:
          error_rate: < 0.01

      - percentage: 100
        success_criteria:
          error_rate: < 0.01

    # Automatic rollback
    rollback:
      on_failure: true
      threshold:
        error_rate: 0.05  # Rollback if >5% errors
```

## Production Best Practices

### Health Checks

Implement health checks:

```python
from blazing import Flow, health

flow = Flow(name="production-workflow")

@flow.health_check
async def check_health():
    """Health check endpoint"""
    # Check database connectivity
    await db.ping()

    # Check external services
    await external_api.health()

    return {
        "status": "healthy",
        "timestamp": datetime.now().isoformat()
    }
```

### Graceful Shutdown

Handle shutdown gracefully:

```python
from blazing import Flow

flow = Flow(name="production-workflow")

@flow.on_shutdown
async def shutdown_handler():
    """Cleanup on shutdown"""
    # Finish in-flight tasks
    await flow.wait_for_tasks()

    # Close connections
    await db.close()
    await cache.close()

    # Save state
    await save_checkpoint()
```

### Resource Management

Set appropriate resource limits:

```yaml
resources:
  # CPU and memory
  cpu: 4
  memory: 8GB

  # Autoscaling
  autoscaling:
    enabled: true
    min_instances: 2
    max_instances: 10
    target_cpu: 70%
    target_memory: 80%
```

## Monitoring Deployment

### Deployment Metrics

Track deployment health:

```bash
# View deployment status
blazing flow deployment status production-workflow

# Monitor metrics during deployment
blazing flow metrics production-workflow \
  --live \
  --filter deployment

# View deployment logs
blazing flow logs production-workflow \
  --since 10m \
  --follow
```

### Rollback Procedures

Automated rollback on failures:

```yaml
deployment:
  rollback:
    # Automatic rollback
    automatic:
      enabled: true

      # Rollback triggers
      triggers:
        - error_rate > 0.05
        - latency_p99 > 1000ms
        - health_check_failures > 3

    # Manual rollback
    manual:
      retain_versions: 5  # Keep last 5 versions
```

## Multi-Environment Deployment

### Environment Configuration

```yaml
# environments/production.yaml
environment: production

resources:
  cpu: 4
  memory: 8GB

concurrency:
  max_concurrent: 100

state:
  storage: postgres
  connection: postgresql://prod-db/flow

monitoring:
  metrics:
    enabled: true
    exporters: [prometheus, datadog]
```

```yaml
# environments/staging.yaml
environment: staging

resources:
  cpu: 2
  memory: 4GB

concurrency:
  max_concurrent: 20

state:
  storage: postgres
  connection: postgresql://staging-db/flow
```

### Deploy to Multiple Environments

```bash
# Deploy to staging
blazing flow deploy workflow.py \
  --config environments/staging.yaml

# Promote to production
blazing flow promote workflow.py \
  --from staging \
  --to production
```

## Next Steps

- [Configure monitoring](/docs/flow-monitoring)
- [Configure your workflows](/docs/flow-configuration)
- [Explore tutorials](/docs/tutorials/index)


Flow Deployment

Monitor, observe, and troubleshoot Blazing Flow workflows

Monitoring


Monitor Blazing Flow workflows with metrics, logs, traces, and alerts to ensure healthy operation.

## Metrics

### Built-in Metrics

Blazing Flow automatically tracks:

```
# Execution metrics
flow_executions_total{status, workflow}
flow_execution_duration_seconds{workflow}
flow_task_duration_seconds{task, workflow}

# Resource metrics
flow_cpu_usage{workflow}
flow_memory_usage{workflow}
flow_active_executions{workflow}

# Error metrics
flow_errors_total{error_type, workflow}
flow_retries_total{task, workflow}
```

### View Metrics

```bash
# Real-time metrics
blazing flow metrics production-workflow --live

# Export metrics
blazing flow metrics export \
  --format prometheus \
  --output metrics.txt
```

### Custom Metrics

Track custom metrics in your workflow:

```python
from blazing import Flow, metrics

flow = Flow(name="production-workflow")

@flow.task
async def process_data(data):
    # Counter
    metrics.counter("data_processed").inc()

    # Histogram
    metrics.histogram("data_size").observe(len(data))

    # Gauge
    metrics.gauge("queue_size").set(get_queue_size())

    # Timer
    with metrics.timer("processing_duration"):
        result = await process(data)

    return result
```

## Logging

### Structured Logging

```python
from blazing import Flow, logger

flow = Flow(name="production-workflow")

@flow.task
async def process_order(order_id: str):
    # Structured logging
    logger.info("Processing order", extra={
        "order_id": order_id,
        "status": "started"
    })

    try:
        result = await process(order_id)

        logger.info("Order processed", extra={
            "order_id": order_id,
            "status": "completed",
            "result": result
        })

    except Exception as e:
        logger.error("Order processing failed", extra={
            "order_id": order_id,
            "error": str(e)
        }, exc_info=True)
```

### Log Aggregation

Configure log shipping:

```yaml
monitoring:
  logging:
    outputs:
      # CloudWatch
      - type: cloudwatch
        log_group: /blazing/flow
        log_stream: production-workflow

      # Datadog
      - type: datadog
        api_key: "${DATADOG_API_KEY}"

      # Elasticsearch
      - type: elasticsearch
        hosts: ["es.example.com:9200"]
        index: flow-logs
```

## Distributed Tracing

### Enable Tracing

```yaml
monitoring:
  tracing:
    enabled: true
    sampling:
      rate: 0.1  # Sample 10% of traces

    exporters:
      - type: jaeger
        endpoint: "http://jaeger:14268"
```

### Custom Spans

Add custom tracing spans:

```python
from blazing import Flow
from opentelemetry import trace

flow = Flow(name="production-workflow")
tracer = trace.get_tracer(__name__)

@flow.task
async def process_data(data):
    # Create custom span
    with tracer.start_as_current_span("process_data") as span:
        span.set_attribute("data_size", len(data))

        # Nested span
        with tracer.start_as_current_span("validate"):
            await validate_data(data)

        with tracer.start_as_current_span("transform"):
            result = await transform_data(data)

        span.set_attribute("result_size", len(result))
        return result
```

## Alerts

### Configure Alerts

```yaml
monitoring:
  alerts:
    enabled: true

    rules:
      # High error rate
      - name: high_error_rate
        condition: error_rate > 0.05
        duration: 5m
        severity: critical
        message: "Error rate above 5% for 5 minutes"

      # High latency
      - name: high_latency
        condition: p95_latency > 1000ms
        duration: 10m
        severity: warning

      # Low success rate
      - name: low_success_rate
        condition: success_rate < 0.95
        duration: 5m
        severity: warning

    # Notification channels
    notifications:
      - type: pagerduty
        integration_key: "${PAGERDUTY_KEY}"
        severity: [critical]

      - type: slack
        webhook: "${SLACK_WEBHOOK}"
        severity: [critical, warning]

      - type: email
        addresses: [ops@example.com]
        severity: [critical]
```

### Alert on Custom Metrics

```python
from blazing import Flow, metrics, alerts

flow = Flow(name="production-workflow")

@flow.task
async def monitor_queue(queue_name: str):
    queue_size = await get_queue_size(queue_name)

    # Set metric
    metrics.gauge("queue_size").set(queue_size)

    # Alert if queue too large
    if queue_size > 1000:
        alerts.trigger(
            name="large_queue",
            severity="warning",
            message=f"Queue {queue_name} has {queue_size} items"
        )
```

## Dashboards

### Grafana Dashboard

```json
{
  "dashboard": {
    "title": "Blazing Flow - Production Workflow",
    "panels": [
      {
        "title": "Execution Rate",
        "targets": [
          {
            "expr": "rate(flow_executions_total[5m])"
          }
        ]
      },
      {
        "title": "Error Rate",
        "targets": [
          {
            "expr": "rate(flow_errors_total[5m])"
          }
        ]
      },
      {
        "title": "P95 Latency",
        "targets": [
          {
            "expr": "histogram_quantile(0.95, flow_execution_duration_seconds)"
          }
        ]
      }
    ]
  }
}
```

## Performance Profiling

### Enable Profiling

```bash
# Run with profiling
blazing flow run production-workflow \
  --profile \
  --profile-output profile.json

# Analyze profile
blazing flow analyze profile.json
```

### Continuous Profiling

```yaml
monitoring:
  profiling:
    enabled: true
    sample_rate: 0.01  # Profile 1% of executions

    exporters:
      - type: pprof
        endpoint: "http://pprof:8080"
```

## Troubleshooting

### Debug Mode

Enable debug logging:

```bash
blazing flow run production-workflow \
  --log-level debug \
  --trace
```

### Execution Traces

View execution trace:

```bash
# Get execution ID
blazing flow list --recent

# View trace
blazing flow trace <execution-id>

# Export trace
blazing flow trace <execution-id> --export trace.json
```

## Next Steps

- [Review deployment guide](/docs/flow-deployment)
- [Configure your workflows](/docs/flow-configuration)
- [Join our community](https://discord.gg/blazing)


Monitor and observe Blazing Flow workflows

Flow Monitoring

"use strict";
const {Fragment: _Fragment, jsx: _jsx, jsxs: _jsxs} = arguments[0];
function _createMdxContent(props) {
  const _components = {
    a: "a",
    code: "code",
    h2: "h2",
    h3: "h3",
    li: "li",
    ol: "ol",
    p: "p",
    pre: "pre",
    span: "span",
    strong: "strong",
    ul: "ul",
    ...props.components
  }, {PreviewCode} = _components;
  if (!PreviewCode) _missingMdxReference("PreviewCode", true);
  return _jsxs(_Fragment, {
    children: [_jsx(_components.p, {
      children: "Learn how to handle errors, validate inputs, and ensure your batch pipelines are resilient."
    }), "\n", _jsxs(_components.h2, {
      id: "common-error-scenarios",
      children: ["Common Error Scenarios", _jsx(_components.a, {
        "aria-hidden": "true",
        tabIndex: "-1",
        href: "#common-error-scenarios",
        children: _jsx(_components.span, {
          className: "icon icon-link"
        })
      })]
    }), "\n", _jsxs(_components.h3, {
      id: "1-invalid-workflow-name",
      children: ["1. Invalid Workflow Name", _jsx(_components.a, {
        "aria-hidden": "true",
        tabIndex: "-1",
        href: "#1-invalid-workflow-name",
        children: _jsx(_components.span, {
          className: "icon icon-link"
        })
      })]
    }), "\n", _jsx(_components.pre, {
      children: _jsx(_components.code, {
        className: "language-python",
        children: "from blazing import Blazing\n\napp = Blazing()\n\n# Try to use a workflow that doesn't exist\nunit = await app.create_workflow_task(\n    \"nonexistent_workflow_12345\",\n    symbol=\"AAPL\"\n)\n\n# The task is created but fails on execution\ntry:\n    result = await unit.result()\nexcept Exception as e:\n    if \"not registered\" in str(e).lower():\n        print(\"✗ Workflow not found\")\n"
      })
    }), "\n", _jsxs(_components.h3, {
      id: "2-missing-required-parameters",
      children: ["2. Missing Required Parameters", _jsx(_components.a, {
        "aria-hidden": "true",
        tabIndex: "-1",
        href: "#2-missing-required-parameters",
        children: _jsx(_components.span, {
          className: "icon icon-link"
        })
      })]
    }), "\n", _jsx(_components.pre, {
      children: _jsx(_components.code, {
        className: "language-python",
        children: "# Missing 'symbol' parameter\nunit = await app.create_workflow_task(\n    \"fetch_timeseries\",\n    points=100,\n    delay=0\n    # symbol is missing!\n)\n\n# Task created but will fail at execution time\n# Framework validates at runtime, not creation time\n"
      })
    }), "\n", _jsxs(_components.h3, {
      id: "3-invalid-parameter-types",
      children: ["3. Invalid Parameter Types", _jsx(_components.a, {
        "aria-hidden": "true",
        tabIndex: "-1",
        href: "#3-invalid-parameter-types",
        children: _jsx(_components.span, {
          className: "icon icon-link"
        })
      })]
    }), "\n", _jsx(_components.pre, {
      children: _jsx(_components.code, {
        className: "language-python",
        children: "# Invalid type for 'points' (should be int)\ntry:\n    unit = await app.create_workflow_task(\n        \"fetch_timeseries\",\n        symbol=\"AAPL\",\n        points=\"not_a_number\",  # Wrong type!\n        delay=0\n    )\n\n    results = await app.gather([unit])\nexcept (TypeError, ValueError) as e:\n    print(f\"✗ Parameter validation failed: {e}\")\n"
      })
    }), "\n", _jsxs(_components.h2, {
      id: "handling-task-failures",
      children: ["Handling Task Failures", _jsx(_components.a, {
        "aria-hidden": "true",
        tabIndex: "-1",
        href: "#handling-task-failures",
        children: _jsx(_components.span, {
          className: "icon icon-link"
        })
      })]
    }), "\n", _jsxs(_components.h3, {
      id: "try-except-pattern",
      children: ["Try-Except Pattern", _jsx(_components.a, {
        "aria-hidden": "true",
        tabIndex: "-1",
        href: "#try-except-pattern",
        children: _jsx(_components.span, {
          className: "icon icon-link"
        })
      })]
    }), "\n", _jsx(_components.pre, {
      children: _jsx(_components.code, {
        className: "language-python",
        children: "import asyncio\n\napp = Blazing()\n\ntry:\n    unit = await app.create_workflow_task(\n        \"risky_operation\",\n        data=large_dataset\n    )\n\n    # Set timeout to prevent hanging\n    result = await asyncio.wait_for(\n        unit.result(),\n        timeout=30.0\n    )\n\n    print(f\"✓ Task completed: {result}\")\n\nexcept asyncio.TimeoutError:\n    print(\"✗ Task timed out after 30 seconds\")\nexcept Exception as e:\n    print(f\"✗ Task failed: {type(e).__name__}: {e}\")\n"
      })
    }), "\n", _jsxs(_components.h3, {
      id: "graceful-degradation",
      children: ["Graceful Degradation", _jsx(_components.a, {
        "aria-hidden": "true",
        tabIndex: "-1",
        href: "#graceful-degradation",
        children: _jsx(_components.span, {
          className: "icon icon-link"
        })
      })]
    }), "\n", _jsx(_components.pre, {
      children: _jsx(_components.code, {
        className: "language-python",
        children: "async def fetch_with_fallback(symbol: str):\n    \"\"\"Fetch data with fallback to cached value.\"\"\"\n    app = Blazing()\n\n    try:\n        unit = await app.create_workflow_task(\n            \"fetch_timeseries\",\n            symbol=symbol,\n            points=100\n        )\n        result = await unit.result()\n        return result\n\n    except Exception as e:\n        logger.warning(f\"Failed to fetch {symbol}: {e}\")\n        # Return cached or default value\n        return get_cached_data(symbol)\n"
      })
    }), "\n", _jsxs(_components.h2, {
      id: "edge-cases",
      children: ["Edge Cases", _jsx(_components.a, {
        "aria-hidden": "true",
        tabIndex: "-1",
        href: "#edge-cases",
        children: _jsx(_components.span, {
          className: "icon icon-link"
        })
      })]
    }), "\n", _jsxs(_components.h3, {
      id: "empty-task-lists",
      children: ["Empty Task Lists", _jsx(_components.a, {
        "aria-hidden": "true",
        tabIndex: "-1",
        href: "#empty-task-lists",
        children: _jsx(_components.span, {
          className: "icon icon-link"
        })
      })]
    }), "\n", _jsx(_components.pre, {
      children: _jsx(_components.code, {
        className: "language-python",
        children: "# Blazing handles empty lists gracefully\nresults = await app.gather([])\n# Returns: []\n\nawait app.wait([])\n# No-op, doesn't crash\n"
      })
    }), "\n", _jsxs(_components.h3, {
      id: "large-payloads",
      children: ["Large Payloads", _jsx(_components.a, {
        "aria-hidden": "true",
        tabIndex: "-1",
        href: "#large-payloads",
        children: _jsx(_components.span, {
          className: "icon icon-link"
        })
      })]
    }), "\n", _jsx(_components.pre, {
      children: _jsx(_components.code, {
        className: "language-python",
        children: "# Test with very large datasets\nunit = await app.create_workflow_task(\n    \"fetch_timeseries\",\n    symbol=\"HUGE\",\n    points=50000  # 50k data points\n)\n\nresults = await app.gather([unit])\nresult = results[0]\n\nassert result['count'] == 50000\nprint(f\"✓ Large payload handled: {result['count']:,} points\")\n"
      })
    }), "\n", _jsxs(_components.h3, {
      id: "multiple-concurrent-tasks-from-same-app",
      children: ["Multiple Concurrent Tasks from Same App", _jsx(_components.a, {
        "aria-hidden": "true",
        tabIndex: "-1",
        href: "#multiple-concurrent-tasks-from-same-app",
        children: _jsx(_components.span, {
          className: "icon icon-link"
        })
      })]
    }), "\n", _jsx(_components.pre, {
      children: _jsx(_components.code, {
        className: "language-python",
        children: "# Safe: Multiple tasks can share the same app instance\napp = Blazing()\n\nunit1 = await app.create_workflow_task(\n    \"fetch_timeseries\",\n    symbol=\"TEST1\",\n    points=10\n)\nunit2 = await app.create_workflow_task(\n    \"fetch_timeseries\",\n    symbol=\"TEST2\",\n    points=10\n)\n\n# Gather with timeout\ntry:\n    results = await asyncio.wait_for(\n        app.gather([unit1, unit2]),\n        timeout=30\n    )\n    assert results[0]['symbol'] == 'TEST1'\n    assert results[1]['symbol'] == 'TEST2'\nexcept asyncio.TimeoutError:\n    print(\"✗ Timeout: workers may not be processing tasks\")\n"
      })
    }), "\n", _jsxs(_components.h2, {
      id: "resource-management",
      children: ["Resource Management", _jsx(_components.a, {
        "aria-hidden": "true",
        tabIndex: "-1",
        href: "#resource-management",
        children: _jsx(_components.span, {
          className: "icon icon-link"
        })
      })]
    }), "\n", _jsxs(_components.h3, {
      id: "proper-cleanup",
      children: ["Proper Cleanup", _jsx(_components.a, {
        "aria-hidden": "true",
        tabIndex: "-1",
        href: "#proper-cleanup",
        children: _jsx(_components.span, {
          className: "icon icon-link"
        })
      })]
    }), "\n", _jsx(_components.pre, {
      children: _jsx(_components.code, {
        className: "language-python",
        children: "async def process_batch():\n    app = Blazing()\n\n    try:\n        # Your processing logic\n        units = [\n            await app.create_workflow_task(...)\n            for _ in range(100)\n        ]\n        results = await app.gather(units)\n\n        return results\n\n    finally:\n        # Always clean up, even on error\n        await app.close()\n"
      })
    }), "\n", _jsxs(_components.h3, {
      id: "idempotent-close",
      children: ["Idempotent Close", _jsx(_components.a, {
        "aria-hidden": "true",
        tabIndex: "-1",
        href: "#idempotent-close",
        children: _jsx(_components.span, {
          className: "icon icon-link"
        })
      })]
    }), "\n", _jsx(_components.pre, {
      children: _jsx(_components.code, {
        className: "language-python",
        children: "# Multiple close() calls are safe\napp = Blazing()\n\nawait app.close()\nawait app.close()  # Safe - no error\nawait app.close()  # Still safe\n"
      })
    }), "\n", _jsxs(_components.h3, {
      id: "using-after-close",
      children: ["Using After Close", _jsx(_components.a, {
        "aria-hidden": "true",
        tabIndex: "-1",
        href: "#using-after-close",
        children: _jsx(_components.span, {
          className: "icon icon-link"
        })
      })]
    }), "\n", _jsx(_components.pre, {
      children: _jsx(_components.code, {
        className: "language-python",
        children: "app = Blazing()\nawait app.close()\n\n# Attempting to use closed app\ntry:\n    unit = await app.create_workflow_task(\"quick_lookup\", symbol=\"TEST\")\nexcept Exception as e:\n    print(f\"Expected error: {type(e).__name__}\")\n"
      })
    }), "\n", _jsxs(_components.h2, {
      id: "retry-patterns",
      children: ["Retry Patterns", _jsx(_components.a, {
        "aria-hidden": "true",
        tabIndex: "-1",
        href: "#retry-patterns",
        children: _jsx(_components.span, {
          className: "icon icon-link"
        })
      })]
    }), "\n", _jsxs(_components.h3, {
      id: "automatic-retry",
      children: ["Automatic Retry", _jsx(_components.a, {
        "aria-hidden": "true",
        tabIndex: "-1",
        href: "#automatic-retry",
        children: _jsx(_components.span, {
          className: "icon icon-link"
        })
      })]
    }), "\n", _jsx(PreviewCode, {
      tier: "dev-preview",
      lines: "4",
      lang: "python",
      children: _jsx(_components.pre, {
        children: _jsx(_components.code, {
          className: "language-python",
          children: "import asyncio\n\nasync def execute_with_retry(\n    workflow: str,\n    params: dict,\n    max_retries: int = 3\n):\n    \"\"\"Execute task with automatic retry on failure.\"\"\"\n    app = Blazing()\n\n    for attempt in range(max_retries):\n        try:\n            unit = await app.create_workflow_task(workflow, **params)\n            result = await asyncio.wait_for(\n                unit.result(),\n                timeout=30\n            )\n            return result\n\n        except Exception as e:\n            if attempt == max_retries - 1:\n                raise  # Final attempt failed\n\n            wait_time = 2 ** attempt  # Exponential backoff\n            print(f\"Retry {attempt + 1}/{max_retries} after {wait_time}s\")\n            await asyncio.sleep(wait_time)\n\n    await app.close()\n"
        })
      })
    }), "\n", _jsxs(_components.h3, {
      id: "partial-failure-handling",
      children: ["Partial Failure Handling", _jsx(_components.a, {
        "aria-hidden": "true",
        tabIndex: "-1",
        href: "#partial-failure-handling",
        children: _jsx(_components.span, {
          className: "icon icon-link"
        })
      })]
    }), "\n", _jsx(_components.pre, {
      children: _jsx(_components.code, {
        className: "language-python",
        children: "async def process_with_partial_failures(symbols: list[str]):\n    \"\"\"Process all symbols, track failures.\"\"\"\n    app = Blazing()\n\n    tasks = []\n    for symbol in symbols:\n        unit = await app.create_workflow_task(\n            \"fetch_timeseries\",\n            symbol=symbol,\n            points=100\n        )\n        tasks.append((symbol, unit))\n\n    successes = []\n    failures = []\n\n    for symbol, unit in tasks:\n        try:\n            result = await unit.result()\n            successes.append(result)\n        except Exception as e:\n            failures.append((symbol, str(e)))\n\n    print(f\"✓ Completed: {len(successes)}\")\n    print(f\"✗ Failed: {len(failures)}\")\n\n    await app.close()\n    return successes, failures\n"
      })
    }), "\n", _jsxs(_components.h2, {
      id: "best-practices",
      children: ["Best Practices", _jsx(_components.a, {
        "aria-hidden": "true",
        tabIndex: "-1",
        href: "#best-practices",
        children: _jsx(_components.span, {
          className: "icon icon-link"
        })
      })]
    }), "\n", _jsxs(_components.ol, {
      children: ["\n", _jsxs(_components.li, {
        children: [_jsx(_components.strong, {
          children: "Always use timeouts"
        }), " for production workloads"]
      }), "\n", _jsxs(_components.li, {
        children: [_jsx(_components.strong, {
          children: "Handle empty collections"
        }), " gracefully"]
      }), "\n", _jsxs(_components.li, {
        children: [_jsx(_components.strong, {
          children: "Validate parameters"
        }), " before task creation when possible"]
      }), "\n", _jsxs(_components.li, {
        children: [_jsx(_components.strong, {
          children: "Clean up resources"
        }), " in ", _jsx(_components.code, {
          children: "finally"
        }), " blocks"]
      }), "\n", _jsxs(_components.li, {
        children: [_jsx(_components.strong, {
          children: "Log errors"
        }), " for debugging and monitoring"]
      }), "\n", _jsxs(_components.li, {
        children: [_jsx(_components.strong, {
          children: "Implement retries"
        }), " for transient failures"]
      }), "\n", _jsxs(_components.li, {
        children: [_jsx(_components.strong, {
          children: "Track partial failures"
        }), " in batch operations"]
      }), "\n"]
    }), "\n", _jsxs(_components.h2, {
      id: "next-steps",
      children: ["Next Steps", _jsx(_components.a, {
        "aria-hidden": "true",
        tabIndex: "-1",
        href: "#next-steps",
        children: _jsx(_components.span, {
          className: "icon icon-link"
        })
      })]
    }), "\n", _jsxs(_components.ul, {
      children: ["\n", _jsxs(_components.li, {
        children: [_jsx(_components.a, {
          href: "/docs/tutorials/concurrency-patterns",
          children: "Concurrency Patterns"
        }), " - Advanced parallelization"]
      }), "\n", _jsxs(_components.li, {
        children: [_jsx(_components.a, {
          href: "/docs/flow-monitoring",
          children: "Flow Monitoring"
        }), " - Track task health"]
      }), "\n", _jsxs(_components.li, {
        children: [_jsx(_components.a, {
          href: "/docs/tutorials/parallel-execution",
          children: "Parallel Execution"
        }), " - Scale your workflows"]
      }), "\n"]
    }), "\n", _jsxs(_components.h2, {
      id: "source",
      children: ["Source", _jsx(_components.a, {
        "aria-hidden": "true",
        tabIndex: "-1",
        href: "#source",
        children: _jsx(_components.span, {
          className: "icon icon-link"
        })
      })]
    }), "\n", _jsx(_components.p, {
      children: "Based on test suite examples:"
    }), "\n", _jsxs(_components.ul, {
      children: ["\n", _jsxs(_components.li, {
        children: [_jsx(_components.code, {
          children: "tests/test_error_handling.py"
        }), " - Comprehensive error scenarios"]
      }), "\n"]
    })]
  });
}
function MDXContent(props = {}) {
  const {wrapper: MDXLayout} = props.components || ({});
  return MDXLayout ? _jsx(MDXLayout, {
    ...props,
    children: _jsx(_createMdxContent, {
      ...props
    })
  }) : _createMdxContent(props);
}
return {
  default: MDXContent
};
function _missingMdxReference(id, component) {
  throw new Error("Expected " + (component ? "component" : "object") + " `" + id + "` to be defined: you likely forgot to import, pass, or provide it.");
}


The task is created but fails on execution

Task created but will fail at execution time

Framework validates at runtime, not creation time

Invalid type for 'points' (should be int)

Safe: Multiple tasks can share the same app instance