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

Redis Consumer Groups - Delivers to ONE consumer
Distributed Lock - SETNX lock:service:record_id
Idempotency Check - GETEX processed:service:record_id

Flow

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

Crash Scenario Examples

Scenario 1: Crash Before Processing

Pod A reads message via XREADGROUP
Pod A acquires lock
Pod A crashes
Lock expires (30s)
Pod B acquires lock
No idempotency marker → Process normally

Scenario 2: Crash After Processing

Pod A processes message successfully
Pod A sets idempotency marker
Pod A crashes before XACK
Pod B claims stale message (5min)
Pod B checks: marker exists!
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

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
Worker: os.environ[“OPENAI_API_KEY”] = user_key
Worker: Process with user key → Results
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:

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):

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:

# 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

redis-cli -h redis.staging.n1

2. Check Queue Status

# 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:

# 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

# 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

#!/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

XREADGROUP → Phoenix
SETNX lock:router:rec-123 → Phoenix
GETEX processed:router:rec-123 → Phoenix
GET api_key:user:user-456 → Phoenix
os.environ[‘OPENAI_API_KEY’] = user_key → Phoenix
DocumentRoutingPipeline.run() → YOUR LIBRARY
SET pages:rec-123 → Phoenix
SET processed:router:rec-123 → Phoenix
XADD phoenix:parser:input → Phoenix
XACK & DEL lock → Phoenix

Parser Worker

XREADGROUP → Phoenix
SETNX lock:parser:rec-123 → Phoenix
GETEX processed:parser:rec-123 → Phoenix
GET pages:rec-123 → Phoenix
N1DataProcessPipeline.run() → YOUR LIBRARY
SET processed:parser:rec-123 → Phoenix
XADD phoenix:grouper:input → Phoenix
DEL pages:rec-123 → Phoenix
XACK & DEL lock → Phoenix

Grouper Worker

XREADGROUP → Phoenix
SETNX lock:grouper:rec-123 → Phoenix
enrichment_main() → YOUR LIBRARY
SET processed:grouper:rec-123 → Phoenix
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