---
title: "Phoenix Infrastructure - Developer Guide"
---

# Phoenix Infrastructure

## What is Phoenix?

**Phoenix is infrastructure, not a pipeline.** It's a framework that wraps your processing libraries with Redis Streams queuing, exponential backoff retries, distributed locks, and structured observability.

Using Phoenix infrastructure, we built a medical document processing pipeline with three stages: Router → Parser → Grouper. This is just ONE example of what you can build with Phoenix.

### What Phoenix Provides (Infrastructure)

- **Queue Technology:** Redis Streams (Valkey 8)
- **Exactly-Once Processing:** Three-layer deduplication (even with pod crashes)
- **Resilience:** Circuit breaker, exponential backoff, infinite retries
- **Per-User Billing:** API key swapping for accurate cost attribution
- **Autoscaling:** Queue-based horizontal scaling
- **Observability:** Structured logging, metrics, tracing

### What We Built Using Phoenix

A medical document processing pipeline with three stages:

- **Router:** Downloads PDFs, uses MinerU VLM, routes pages by type
- **Parser:** Extracts biomarkers, diagnoses, procedures, genetics
- **Grouper:** Groups and enriches biomarkers

**Key Distinction:** Phoenix is reusable infrastructure. You can use Phoenix to build ANY processing pipeline by plugging in your own libraries.

## Our Document Processing Pipeline

Built using Phoenix infrastructure:

```
API Backend
    ↓
Redis / Valkey Queues:
    - phoenix:router:input
    - phoenix:parser:input
    - phoenix:grouper:input
    - Dead Letter Queue
    - Locks & Markers
    ↓
Router Workers (2-10 pods)
    → Parser Workers (2-10 pods)
    → Grouper Workers (2-6 pods)
```

## Our Pipeline: Three Processing Stages

Each stage is a library wrapped by Phoenix infrastructure:

| Stage | Package | Function | Duration |
|-------|---------|----------|----------|
| **Router** | parser-router | Downloads PDFs, uses MinerU VLM, routes by content type | 30-60s |
| **Parser** | parser-sequential | Extracts biomarkers, diagnoses, procedures, genetics | 60-120s |
| **Grouper** | enrichment-biomarkers | Groups biomarkers, enriches with canonical names | 2-5min |
## Component Architecture

How your library integrates with Phoenix infrastructure:

### Phoenix Shared Library

- `queue.py` - Redis Streams
- `locks.py` - Distributed locks
- `idempotency.py` - Duplicate prevention
- `circuit_breaker.py` - Failure isolation
- `api_key_manager.py` - Per-user billing

### Phoenix Worker Framework

- `worker.py` - Message loop
- `base.py` - BaseProcessor

### Your Code

- `genetics_processor.py` - ~50 lines wrapper
- `genetics-parser` - Your library

**Separation of Concerns:** Phoenix provides the shared library and worker framework (all infrastructure). You write a thin wrapper (~50 lines) that calls your processing library. Your library has zero infrastructure code.

## Exactly-Once Processing: Three-Layer Deduplication

Ensures messages are processed exactly once, even with pod crashes.

### The Three Layers

1. **Redis Consumer Groups** - Delivers to ONE consumer
2. **Distributed Lock** - `SETNX lock:service:record_id`
3. **Idempotency Check** - `GETEX processed:service:record_id`

### Flow

1. Message arrives
2. Redis Consumer Groups delivers to ONE consumer
3. Distributed Lock: SETNX lock:service:record_id
   - If no: Skip - Another pod processing
   - If yes: Continue
4. Idempotency Check: GETEX processed:service:record_id
   - If already done: Skip processing, just ACK
   - If not: Process message
5. Mark as processed: SET processed:service:record_id

### Crash Scenario Examples

**Scenario 1: Crash Before Processing**
1. Pod A reads message via XREADGROUP
2. Pod A acquires lock
3. Pod A crashes
4. Lock expires (30s)
5. Pod B acquires lock
6. No idempotency marker → Process normally

**Scenario 2: Crash After Processing**
1. Pod A processes message successfully
2. Pod A sets idempotency marker
3. Pod A crashes before XACK
4. Pod B claims stale message (5min)
5. Pod B checks: marker exists!
6. Pod B skips processing, just ACKs

**Key Insight:** Idempotency markers use Time-to-Idle pattern (GETEX refreshes TTL). This ensures markers stay valid during processing but expire if truly abandoned.

## Circuit Breaker

Prevents cascade failures when LLM APIs are down. Messages stay in queue instead of flooding a failing service.

### Three States

- **CLOSED** - Normal operation, calls go through
- **OPEN** - 5 failures detected, fail fast immediately
- **HALF_OPEN** - Testing recovery with 3 test calls

### Configuration

- Failure threshold: 5
- Reset timeout: 60 seconds
- Half-open test calls: 3

### State Transitions

```
CLOSED → (5 failures) → OPEN → (60s timeout) → HALF_OPEN
                                              ↓
                                    (3 successes) → CLOSED
                                    (any failure) → OPEN
```

**Important:** When circuit breaker opens, messages are NOT sent to DLQ. They stay in queue and retry after circuit closes.

## Retry Logic: Exponential Backoff

**Formula:** `backoff = min(2^attempt × 2 seconds, 300 seconds max)`

### Attempt Schedule

- Attempt 1: 4 seconds
- Attempt 2: 8 seconds
- Attempt 3: 16 seconds
- Attempt 4: 32 seconds
- Attempt 5+: 300s (5 min max)
- Dead Letter Queue: Only 404s → DLQ; Everything else retries forever

### Error Handling Patterns

| Error Type | Behavior | Why |
|------------|----------|-----|
| `N1APIRecordNotFoundError` | Send to DLQ immediately | Record doesn't exist, no point retrying |
| `CircuitBreakerOpenError` | Keep in queue, NO retry delay | Wait for circuit to close (60s) |
| Timeout, Network errors | Retry with exponential backoff | Transient errors, likely to recover |
| All other exceptions | Retry with exponential backoff | Unknown errors, attempt recovery |
## Redis Streams: Key Operations

| Operation | Function | Example |
|-----------|----------|---------|
| `XADD` | Add message to queue | XADD phoenix:router:input * data |
| `XREADGROUP` | Read message as consumer | XREADGROUP GROUP router-group worker COUNT 1 |
| `XACK` | Acknowledge processed message | XACK phoenix:router:input router-group msg-id |
| `XAUTOCLAIM` | Reclaim stale messages (5 min idle) | XAUTOCLAIM phoenix:router:input router-group worker 300000 |
| `XPENDING` | Check pending/unacked messages | XPENDING phoenix:router:input router-group |
## Per-User LLM Billing

API key swapping for accurate cost attribution per user.

### Sequence

1. Worker: GETEX api_key:user:user-456
   - Cache Hit: Returns sk-user-456-xxx
   - Cache Miss:
     - Worker: POST /keys/generate
     - Billing: Returns sk-user-456-xxx
     - Worker: SET api_key:user:user-456 EX 86400
2. Worker: os.environ["OPENAI_API_KEY"] = user_key
3. Worker: Process with user key → Results
4. Worker: Restore original key

**Safe by Design:** Workers process one message at a time (batch_size=1), so no concurrency issues with key swapping.

## Building a Service: Step-by-Step

Phoenix separates business logic from infrastructure. You build a library with zero infrastructure code. Phoenix wraps it with queuing, locks, retries, and billing.

### Step 1: Build Your Library

Create a Python package with this interface:

```python
# genetics_parser/parser.py
from typing import Dict, List, Any
import os

class GeneticsParser:
    async def run(
        self,
        record_id: str,
        user_id: str,
        pages: List[Dict[str, Any]]
    ) -> Dict[str, Any]:
        """
        Process genetics data from pages.

        Args:
            record_id: Document identifier
            user_id: User identifier
            pages: List of page data:
                [{
                    "page_idx": 1,
                    "text_markdown": "...",
                    "categories": ["genetics"],
                    "pdf_base64": "..."
                }]

        Returns:
            Processing results: {
                "variants": [...],
                "genes": [...]
            }
        """
        # Your library uses OPENAI_API_KEY from environment
        # Phoenix swaps this per-user automatically
        api_key = os.environ.get("OPENAI_API_KEY")

        # Your processing logic here
        variants = self.extract_variants(pages)
        genes = self.extract_genes(pages)

        return {
            "variants": variants,
            "genes": genes
        }
```

**Requirements:**
- Expose `async def run(record_id, user_id, pages) → Dict`
- Read `OPENAI_API_KEY` from environment (Phoenix swaps per-user)
- Be stateless - no shared state between calls
- No Redis, no queues, no infrastructure code

### Step 2: Create Phoenix Wrapper

Thin processor that imports your library (~50 lines):

```python
# worker/src/processors/genetics.py
import json
from typing import Dict, Any, Optional
from .base import BaseProcessor
from genetics_parser import GeneticsParser

class GeneticsProcessor(BaseProcessor):
    """Phoenix wrapper for genetics-parser library."""

    def __init__(self, service_name, redis_client, api_key_manager):
        super().__init__(service_name, api_key_manager)
        self.redis = redis_client

    async def process(
        self,
        record_id: str,
        user_id: str,
        data: Dict[str, Any]
    ) -> Optional[Dict[str, Any]]:
        """
        Phoenix calls this for each message.
        Infrastructure (locks, retries, billing) handled by BaseProcessor.
        """

        # Get pages from Redis (stored by Router)
        pages_key = data.get("pages_key")
        pages_json = await self.redis.get(pages_key)
        pages = json.loads(pages_json)

        # Call your library
        parser = GeneticsParser()
        result = await parser.run(record_id, user_id, pages)

        self.logger.info(
            "genetics_processing_complete",
            record_id=record_id,
            variant_count=len(result.get("variants", []))
        )

        return result
```

**What BaseProcessor Provides:** Logging, API key management through `self.api_key_manager`, structured logger. Worker framework handles: locks, retries, circuit breaker, idempotency, queue operations.

### Step 3: Queue Setup

Create Redis consumer group for your service:

```bash
# Connect to Redis
redis-cli -h redis.staging.n1

# Create queue and consumer group
XGROUP CREATE phoenix:genetics:input genetics-group $ MKSTREAM

# Verify
XINFO GROUPS phoenix:genetics:input
```

### Step 4: Configuration

Add Dockerfile and docker-compose configuration. See `docs/ADDING_SERVICES.md` for complete templates.

| Step | What to Do | Effort |
|------|-----------|--------|
| **1. Build Library** | Create Python package with `run()` method | Your timeline |
| **2. Create Wrapper** | Thin Phoenix processor importing & calling library | ~50 lines code |
| **3. Queue Setup** | Create Redis consumer group | 1 command |
| **4. Configure** | Dockerfile, docker-compose, environment variables | ~30 minutes |
## Debugging Your Service

### 1. Connect to Redis

```bash
redis-cli -h redis.staging.n1
```

### 2. Check Queue Status

```bash
# Check queue depth
XLEN phoenix:genetics:input

# Check pending (unacknowledged) messages
XPENDING phoenix:genetics:input genetics-group

# Check consumer groups
XINFO GROUPS phoenix:genetics:input
XINFO CONSUMERS phoenix:genetics:input genetics-group

# View recent messages
XRANGE phoenix:genetics:input - + COUNT 10
```

### 3. Inspect Infrastructure Keys

These Redis keys control Phoenix's deduplication and locking:

```bash
# Check if message is locked (active processing)
GET lock:genetics:rec-123
# Returns: pod-name or (nil)

# Check if already processed (idempotency marker)
GET processed:genetics:rec-123
# Returns: {"timestamp":1701936000,"service":"genetics"} or (nil)

# Check user's API key (billing)
GET api_key:user:user-456
# Returns: sk-user-456-xxx or (nil)

# View DLQ entries
LRANGE phoenix:genetics:dlq 0 10
# Returns: List of failed messages
```

### 4. Troubleshooting Decision Tree

```
Message not processing?
    ├─ XLEN shows queue empty?
    │   └─ Check if messages being added upstream
    └─ XPENDING shows stuck?
        ├─ GET lock:service:rec-id
        │   ├─ Lock exists?
        │   │   └─ Check if pod crashed; Wait 30s for lock expiry
        │   └─ No lock?
        │       └─ GET processed:service:rec-id
        │           ├─ Marker exists?
        │           │   └─ Already processed! Wait for XAUTOCLAIM to ACK and remove
        │           └─ No marker?
        │               └─ Check logs for processing errors
        └─ No pending?
            └─ Messages processing normally - check logs
```

### 5. Common Issues & Solutions

| Symptom | Diagnosis | Solution |
|---------|-----------|----------|
| Messages stuck in PENDING | `GET lock:service:rec-123` shows lock exists | Pod crashed. Wait 30s for lock expiry, or 5min for XAUTOCLAIM |
| DLQ has entries | `LRANGE phoenix:genetics:dlq 0 10` | Check error field, fix root cause. Only 404s should be in DLQ |
| Logs show "circuit_open" | Circuit breaker protecting LLM API | Wait 60s for auto-recovery. Messages stay in queue, no DLQ |
| Message reprocessing | `GET processed:service:rec-123` shows (nil) | Idempotency marker missing. Check if pod crashed before marking |
| No messages processing | `XINFO CONSUMERS` shows no consumers | Workers not running. Check pod status |
### 6. Manual Recovery Operations

```bash
# Manually claim a stuck message (if needed)
XCLAIM phoenix:genetics:input genetics-group manual-consumer 300000 <message-id>

# View message content
XRANGE phoenix:genetics:input <message-id> <message-id>

# Check all pending for a consumer
XPENDING phoenix:genetics:input genetics-group - + 10 consumer-name

# Clear lock manually (emergency only)
DEL lock:genetics:rec-123

# Clear idempotency marker (emergency only - causes reprocessing!)
DEL processed:genetics:rec-123

# View DLQ and replay after fixing
LRANGE phoenix:genetics:dlq 0 -1
# ... fix root cause ...
# Replay is manual - parse JSON and XADD back to input queue
```

### 7. Monitoring Script

```bash
#!/bin/bash
REDIS_HOST="redis.staging.n1"
SERVICE="genetics"

echo "=== $SERVICE Queue Status ==="

QUEUE="phoenix:${SERVICE}:input"
GROUP="${SERVICE}-group"
DLQ="phoenix:${SERVICE}:dlq"

DEPTH=$(redis-cli -h $REDIS_HOST XLEN $QUEUE)
PENDING=$(redis-cli -h $REDIS_HOST XPENDING $QUEUE $GROUP | head -1)
DLQ_SIZE=$(redis-cli -h $REDIS_HOST LLEN $DLQ)

echo "Queue Depth:  $DEPTH"
echo "Pending:      $PENDING"
echo "DLQ Size:     $DLQ_SIZE"

# Check for stuck messages (pending > 5 min)
if [ "$PENDING" != "0" ]; then
    echo ""
    echo "Checking for stuck messages..."
    redis-cli -h $REDIS_HOST XPENDING $QUEUE $GROUP - + 10
fi
```

## Complete Message Lifecycle

Our document processing pipeline showing Phoenix infrastructure vs your library code:

### Router Worker
1. XREADGROUP → *Phoenix*
2. SETNX lock:router:rec-123 → *Phoenix*
3. GETEX processed:router:rec-123 → *Phoenix*
4. GET api_key:user:user-456 → *Phoenix*
5. os.environ['OPENAI_API_KEY'] = user_key → *Phoenix*
6. DocumentRoutingPipeline.run() → **YOUR LIBRARY**
7. SET pages:rec-123 → *Phoenix*
8. SET processed:router:rec-123 → *Phoenix*
9. XADD phoenix:parser:input → *Phoenix*
10. XACK & DEL lock → *Phoenix*

### Parser Worker
1. XREADGROUP → *Phoenix*
2. SETNX lock:parser:rec-123 → *Phoenix*
3. GETEX processed:parser:rec-123 → *Phoenix*
4. GET pages:rec-123 → *Phoenix*
5. N1DataProcessPipeline.run() → **YOUR LIBRARY**
6. SET processed:parser:rec-123 → *Phoenix*
7. XADD phoenix:grouper:input → *Phoenix*
8. DEL pages:rec-123 → *Phoenix*
9. XACK & DEL lock → *Phoenix*

### Grouper Worker
1. XREADGROUP → *Phoenix*
2. SETNX lock:grouper:rec-123 → *Phoenix*
3. enrichment_main() → **YOUR LIBRARY**
4. SET processed:grouper:rec-123 → *Phoenix*
5. XACK & DEL lock → *Phoenix*

**Key Distinction:** Phoenix handles ALL infrastructure steps (locks, idempotency, queuing, billing). Your library is called ONCE per worker for the actual processing logic. Everything else is Phoenix.

## Key Takeaways

### Reliability
Three-layer deduplication ensures exactly-once processing even with pod crashes. Idempotency markers prevent reprocessing.

### Resilience
Circuit breaker protects LLM APIs. Exponential backoff retries. Messages stay in queue when circuit opens.

### Simple Integration
Build your library with `run()` method. Phoenix wrapper is ~50 lines. No infrastructure code in your library.

### Debuggable
Inspect locks, idempotency markers, DLQ entries directly in Redis. Clear troubleshooting paths.

**Core Design:** Separation of concerns. Your library has zero infrastructure code (no queues, no Redis, no retries). Phoenix handles Redis Streams, distributed locks, idempotency markers, circuit breakers, and per-user billing. Test your libraries independently.

## Future Phoenix Features

Planned enhancements to Phoenix infrastructure:

### Progress Tracking
Phoenix infrastructure: Real-time progress updates per record using Redis pub/sub

### Enhanced Observability
Phoenix infrastructure: Grafana dashboards for queue depths, processing rates, errors

### Response Caching
Phoenix infrastructure: Cache LLM responses to reduce costs

### Example Pipelines to Build

- **Specialized Parsers** - Split into Simple, Visual, Complex parsers based on content
- **Imaging Pipeline** - CT, MRI, X-ray specialized processors with DICOM support
- **Genetics Pipeline** - Dedicated genetics enrichment with variant databases