---
title: "Phoenix Pipeline - Medical Document Processing Infrastructure"
---

Phoenix Pipeline - Medical Document Processing Infrastructure

## Overview

### What is Phoenix Pipeline?

Phoenix Pipeline is an infrastructure for processing medical documents. Built on Redis Streams, it provides per-user billing, exactly-once semantics, and comprehensive observability while maintaining zero modifications** to business logic.

### Why Phoenix?

Phoenix Pipeline was built to solve real production challenges in medical document processing:

## Architecture

- Receives record_id + user_id
- Validates request
- Enqueues to router stream

##### 📨 Streams (Queues)

- phoenix:router:input
- phoenix:parser:input
- phoenix:grouper:input

##### 🔒 Locks

- lock:router:{id}
- lock:parser:{id}
- lock:grouper:{id}

##### ✓ Idempotency

- processed:router:{id}
- processed:parser:{id}
- processed:grouper:{id}

##### 💾 Cache

- pages:{id} (1h TTL)
- api_key:user:{id} (24h)
- fetching_key:{id} (10s)

##### ⚠️ Dead Letter Queues

- phoenix:router:dlq
- phoenix:parser:dlq
- phoenix:grouper:dlq

- Single Redis instance handles all queuing, locking, and caching needs
- Workers scale independently based on queue depth (KEDA autoscaling)
- Three-layer deduplication: consumer groups → distributed locks → idempotency markers
- Zero business logic changes to existing packages (clean wrapper pattern)

### Technology Stack

## Key Features

Existing N1 Healthcare packages (`parser-router`, `parser-sequential`, `enrichment-biomarkers`) are called via their public entry points. No business logic changes required.

- Clean separation between infrastructure and business logic
- Easy to test packages independently
- Simple upgrades without Phoenix modifications

LLM costs are accurately attributed to individual users via LiteLLM virtual keys:

1. Worker receives message with `user_id`
1. Checks Redis cache for user's API key (24h TTL)
1. On cache miss, calls billing service
1. Swaps `OPENAI_API_KEY` environment variable
1. Processes document (all LLM calls billed to user)
1. Restores original key

**Safety:** Batch size = 1 ensures sequential processing makes env var swap safe.

Three layers of deduplication ensure each document is processed exactly once:

#### Layer 1: Consumer Groups (Redis Native)

Redis Streams delivers each message to exactly ONE consumer in the group.

#### Layer 2: Distributed Locks (Defense in Depth)

SETNX lock prevents race conditions if message is re-delivered after crash.

#### Layer 3: Idempotency Markers (Post-Crash Protection)

Prevents reprocessing if pod crashes after completing work but before acknowledging.

Protects against cascading failures when LLM API is down:

When circuit is open, messages remain in queue indefinitely until LLM API recovers.

## Processing Pipeline

### Three-Stage Pipeline

### Stage Details

### Data Flow Between Services

**Router stores pages in Redis:**

**Router sends to parser queue:**

  record_id mr-xxxx-xxxx-xxx \
  user_id xxx-xxx-xxx \
  pages_key pages:mr-xxxx-xxxx-xxx \
  document_name "lab_results.pdf" \
  page_count 5

**Parser retrieves and processes:**

# Parse and extract biomarkers, diagnoses, etc.
DEL pages:mr-xxxx-xxxx-xxx  # Cleanup after use

**Parser sends extraction counts in queue message:**

  record_id mr-xxxx-xxxx-xxx \
  user_id xxx-xxx-xxx \
  biomarker_count 15 \
  diagnosis_count 3 \
  procedures_count 2 \
  genetics_count 0 \
  status completed

**Grouper fetches biomarkers from N1 API:**

The grouper retrieves biomarkers directly from N1 API using the `record_id`, groups them using ML embeddings (MedEmbed-large-v0.1), and enriches with canonical names via LLM.

### Complete Message Lifecycle

```text
# 1. API Backend submits document
```
XADD phoenix:router:input * record_id mr-xxxx-xxxx-xxx user_id xxx-xxx-xxx

# 2. Router Worker processes
XREADGROUP GROUP router-group consumer-1 COUNT 1 BLOCK 5000 STREAMS phoenix:router:input >
SETNX lock:router:mr-xxxx-xxxx-xxx consumer-1 EX 600
GET api_key:user:xxx-xxx-xxx  # or fetch from billing service
SET pages:mr-xxxx-xxxx-xxx {pages_json} EX 3600
SET processed:router:mr-xxxx-xxxx-xxx {metadata} EX 86400
XADD phoenix:parser:input * record_id mr-xxxx-xxxx-xxx pages_key pages:mr-xxxx-xxxx-xxx
XACK phoenix:router:input router-group msg-id
DEL lock:router:mr-xxxx-xxxx-xxx

# 3. Parser Worker processes
XREADGROUP GROUP parser-group consumer-2 COUNT 1 BLOCK 5000 STREAMS phoenix:parser:input >
SETNX lock:parser:mr-xxxx-xxxx-xxx consumer-2 EX 600
GET pages:mr-xxxx-xxxx-xxx
# Extract biomarkers, diagnoses, etc.
DEL pages:mr-xxxx-xxxx-xxx
SET processed:parser:mr-xxxx-xxxx-xxx {metadata} EX 86400
XADD phoenix:grouper:input * record_id mr-xxxx-xxxx-xxx biomarker_count 15
XACK phoenix:parser:input parser-group msg-id
DEL lock:parser:mr-xxxx-xxxx-xxx

# 4. Grouper Worker processes
XREADGROUP GROUP grouper-group consumer-3 COUNT 1 BLOCK 5000 STREAMS phoenix:grouper:input >
SETNX lock:grouper:mr-xxxx-xxxx-xxx consumer-3 EX 600
# Fetch biomarkers from N1 API, group, enrich
SET processed:grouper:mr-xxxx-xxxx-xxx {metadata} EX 86400
XACK phoenix:grouper:input grouper-group msg-id
DEL lock:grouper:mr-xxxx-xxxx-xxx

# ✅ Pipeline complete!

## Reliability Patterns

### Stale Message Recovery

**Solution:** Periodic XAUTOCLAIM every 60 seconds

```text
if time.time() - last_cleanup > 60:
claimed = await queue_manager.claim_stale_messages(
stream=input_queue,
group=consumer_group,
consumer=consumer_name,
min_idle_ms=300000  # 5 minutes
)
for msg_id, msg_data in claimed:
await process_message(msg_id, msg_data)
last_cleanup = time.time()

```
Messages idle for more than 5 minutes are automatically claimed and reprocessed. The idempotency layer ensures no double-processing.

### Exponential Backoff Retry

When processing fails due to transient errors:

- Formula: `min(2^attempt * 2, 300)` seconds
- Prevents hammering failing services
- Aging warnings logged at 4+ hours
- Messages stay in queue indefinitely (except 404s)

### Dead Letter Queue (DLQ)

**DLQ Replay:** Failed messages can be manually replayed after fixing root cause.

```text
python scripts/replay_dlq.py redis://... replay router-dlq router-queue 10

```
## Core Components

### Shared Library (`[phoenix_shared](https://github.com/n1healthcare/phoenix-pipeline/tree/main/phoenix_shared)`)

### Worker Framework

Initialization flow:

1. Load configuration from environment
1. Setup telemetry (traces + metrics)
1. Initialize QueueManager and connect to Redis
1. Initialize lock, idempotency, circuit breaker, API key managers
1. Select processor based on `SERVICE_NAME` env var
1. Create Worker with all components
1. Start health server on port 8080
1. Begin processing loop

Main processing loop handles:

- Message reading from input queue (blocks up to 5s)
- Periodic stale message cleanup (every 60s)
- Lock acquisition to prevent concurrent processing
- Idempotency checking to skip already-processed messages
- Circuit-breaker protected processor calls
- Marking messages as processed
- Forwarding to output queue
- Message acknowledgment and lock release

**Error Handling:**

- 404 errors → immediate DLQ (no retry)
- CircuitOpenError → keep in queue (retry indefinitely)
- Other exceptions → exponential backoff retry

### Processor Implementations

## Deployment

### Docker Images

Phoenix uses a multi-stage build with a shared base image. View [docker/](https://github.com/n1healthcare/phoenix-pipeline/tree/main/docker) directory on GitHub.

```text
🐳 Build Process
# 1. Build base image (Python 3.12 + shared deps)
```
docker build -f docker/Dockerfile.base \
```text
--build-arg GITHUB_TOKEN=$GITHUB_TOKEN \
-t phoenix-base .

```
# 2. Build service images (install processor packages)
docker build -f docker/Dockerfile.router \
```text
--build-arg GITHUB_TOKEN=$GITHUB_TOKEN \
-t phoenix-router .

```
docker build -f docker/Dockerfile.parser \
```text
--build-arg GITHUB_TOKEN=$GITHUB_TOKEN \
-t phoenix-parser .

```
docker build -f docker/Dockerfile.grouper \
```text
--build-arg GITHUB_TOKEN=$GITHUB_TOKEN \
-t phoenix-grouper .

```
### Kubernetes Deployment

```text
☸️ Deploy with Helm
# Deploy Router
```
helm upgrade --install phoenix-router ./charts/phoenix-router \
```text
-f ./environments/staging/phoenix-router.yaml -n staging

```
# Deploy Parser
helm upgrade --install phoenix-parser ./charts/phoenix-parser \
```text
-f ./environments/staging/phoenix-parser.yaml -n staging

```
# Deploy Grouper
helm upgrade --install phoenix-grouper ./charts/phoenix-grouper \
```text
-f ./environments/staging/phoenix-grouper.yaml -n staging

```
# Verify deployment
kubectl get pods -n staging -l app.kubernetes.io/part-of=phoenix-pipeline

### Configuration

## Monitoring & Observability

### OpenTelemetry Metrics

### Structured Logging

All logs are JSON-formatted with context:

```text
{
```
  "timestamp": "2024-12-07T10:30:45.123Z",
  "level": "info",
  "event": "processing_complete",
  "service": "parser",
  "record_id": "mr-xxxx-xxxx-xxx",
  "user_id": "xxx-xxx-xxx",
  "duration_seconds": 45.2,
  "biomarker_count": 12,
  "diagnosis_count": 3,
  "trace_id": "abc123def456",
  "span_id": "789ghi"
}

### Queue Monitoring

```text
python scripts/monitor.py redis://your-redis:6379/0

```
# Or manually with Redis CLI
redis-cli XLEN phoenix:router:input
redis-cli XPENDING phoenix:router:input router-group
redis-cli XINFO GROUPS phoenix:router:input
redis-cli LLEN phoenix:router:dlq

## Scaling

### KEDA Autoscaling

Phoenix uses KEDA (Kubernetes Event-Driven Autoscaling) for automatic horizontal scaling based on Redis Streams lag.

### Scaling Behavior

### Manual Scaling

```text
# Scale up immediately
```
kubectl scale deployment/phoenix-router -n staging --replicas=5

# Scale down
kubectl scale deployment/phoenix-router -n staging --replicas=2

# Check current scaling
kubectl get scaledobjects -n staging
kubectl get hpa -n staging

## Troubleshooting

### Common Issues

**Symptoms:** Queue depth increasing, no logs showing processing

**Diagnosis:**

kubectl get pods -n staging -l app=phoenix-router

# Check pod logs for errors
kubectl logs -n staging -l app=phoenix-router --tail=50

# Check if consumer group exists
redis-cli -h $REDIS_HOST XINFO GROUPS phoenix:router:input

**Solutions:**

redis-cli -h $REDIS_HOST XGROUP CREATE phoenix:router:input router-group $ MKSTREAM

# If pods in CrashLoopBackOff
kubectl describe pod -n staging -l app=phoenix-router
# Check events section for reason

# If Redis connection issues
kubectl logs -n staging -l app=phoenix-router | grep "redis"

**Symptoms:** Dead letter queue has accumulating entries

**Diagnosis:**

redis-cli -h $REDIS_HOST LRANGE phoenix:router:dlq 0 10

# Check common error patterns
redis-cli -h $REDIS_HOST LRANGE phoenix:router:dlq 0 -1 | jq -r '.error' | sort | uniq -c | sort -rn

**Solutions:**

RECORD_ID="mr-xxxx-xxxx-xxx"
kubectl logs -n staging -l app=phoenix-router --since=24h | grep $RECORD_ID

# Replay DLQ after fixing root cause (see Runbooks section)

**Symptoms:** Logs show "circuit_open_fast_fail", processing stops

**Diagnosis:**

kubectl logs -n staging -l app=phoenix-router | grep "circuit"

# Check LLM service health
curl -s http://litellm-service.staging:4000/health

**Solutions:**

# Circuit breaker will transition OPEN → HALF-OPEN → CLOSED

# Or restart pods to reset circuit breaker state
kubectl rollout restart deployment/phoenix-router -n staging

**Symptoms:** Messages in PENDING state for > 5 minutes

**Diagnosis:**

redis-cli -h $REDIS_HOST XPENDING phoenix:router:input router-group - + 10

# Check if any consumers are stuck
redis-cli -h $REDIS_HOST XINFO CONSUMERS phoenix:router:input router-group

**Solutions:**

# Wait for next cleanup cycle (idle time: 5 minutes)

# Or manually claim and process
redis-cli -h $REDIS_HOST XCLAIM phoenix:router:input router-group manual-consumer 300000 <message-id>

**Symptoms:** Pods getting OOMKilled, high memory usage

**Diagnosis:**

kubectl top pods -n staging -l app=phoenix-grouper

# Check pod events for OOMKilled
kubectl describe pod -n staging -l app=phoenix-grouper | grep -A 5 "OOMKilled"

# Check for memory leaks in logs
kubectl logs -n staging -l app=phoenix-grouper | grep -i memory

**Solutions:**

- **Increase memory limits:** Update Helm values to increase pod memory (default: 2Gi)
- **Reduce batch size:** Ensure BATCH_SIZE=1 (never process multiple records)
- **Check for large models:** ML models loaded in initialize() stay in memory
- **Clear cache data:** Large cached data in Redis should have TTL set

resources:
  limits:
```text
memory: 4Gi  # Increase from 2Gi
```
  requests:
```text
memory: 2Gi

```
**Symptoms:** Processing taking longer than expected, queue backlog growing

**Diagnosis:**

kubectl logs -n staging -l app=phoenix-parser --since=1h | \
  grep "processing_complete" | \
  jq '.duration_seconds'

# Check queue lengths
redis-cli -h $REDIS_HOST XLEN phoenix:parser:input
redis-cli -h $REDIS_HOST XLEN phoenix:grouper:input

# Check if LLM calls are slow
kubectl logs -n staging -l app=phoenix-parser | grep "llm_call"

**Common Causes & Solutions:**

## Operational Runbooks

Step-by-step procedures for common operational tasks.

### 🔄 DLQ Replay

```text
#!/bin/bash
```
# Replay DLQ messages back to the main queue

REDIS_HOST="staging-valkey.zc3ep9.ng.0001.use2.cache.amazonaws.com"
SOURCE_DLQ="phoenix:router:dlq"
TARGET_QUEUE="phoenix:router:input"

# Get DLQ length
DLQ_LEN=$(redis-cli -h $REDIS_HOST LLEN $SOURCE_DLQ)
echo "DLQ has $DLQ_LEN messages"

# Replay each message
for i in $(seq 1 $DLQ_LEN); do
```text
# Pop from DLQ (RPOP from right/oldest first)
MSG=$(redis-cli -h $REDIS_HOST RPOP $SOURCE_DLQ)

if [ -z "$MSG" ]; then
echo "DLQ empty, stopping"
break
fi

# Parse message (assuming JSON format)
RECORD_ID=$(echo $MSG | jq -r '.record_id')
USER_ID=$(echo $MSG | jq -r '.user_id')

echo "Replaying: record_id=$RECORD_ID"

# Add back to main queue
redis-cli -h $REDIS_HOST XADD $TARGET_QUEUE '*' \
record_id "$RECORD_ID" \
user_id "$USER_ID" \
retry_count 0

```
done

echo "Replay complete"

### 🔄 Complete Service Restart

```text
#!/bin/bash
```
# Full restart of Phoenix Pipeline

NAMESPACE="staging"

echo "=== Stopping Workers ==="
# Scale down to 0 to drain gracefully
kubectl scale deployment/phoenix-router -n $NAMESPACE --replicas=0
kubectl scale deployment/phoenix-parser -n $NAMESPACE --replicas=0
kubectl scale deployment/phoenix-grouper -n $NAMESPACE --replicas=0

echo "Waiting for pods to terminate..."
kubectl wait --for=delete pod -l app=phoenix-router -n $NAMESPACE --timeout=120s
kubectl wait --for=delete pod -l app=phoenix-parser -n $NAMESPACE --timeout=120s
kubectl wait --for=delete pod -l app=phoenix-grouper -n $NAMESPACE --timeout=120s

echo "=== Clearing Stale Locks (Optional) ==="
# Optional: Clear any stuck locks
# redis-cli -h $REDIS_HOST KEYS "lock:*" | xargs -r redis-cli -h $REDIS_HOST DEL

echo "=== Starting Workers ==="
kubectl scale deployment/phoenix-router -n $NAMESPACE --replicas=2
kubectl scale deployment/phoenix-parser -n $NAMESPACE --replicas=2
kubectl scale deployment/phoenix-grouper -n $NAMESPACE --replicas=2

echo "Waiting for pods to be ready..."
kubectl wait --for=condition=ready pod -l app=phoenix-router -n $NAMESPACE --timeout=120s
kubectl wait --for=condition=ready pod -l app=phoenix-parser -n $NAMESPACE --timeout=120s
kubectl wait --for=condition=ready pod -l app=phoenix-grouper -n $NAMESPACE --timeout=120s

echo "=== Restart Complete ==="
kubectl get pods -n $NAMESPACE -l app.kubernetes.io/part-of=phoenix-pipeline

### ⚠️ Emergency Redis Failover

```text
# 1. Check Redis cluster status
```
aws elasticache describe-cache-clusters \
```text
--cache-cluster-id staging-valkey \
--show-cache-node-info

```
# 2. ElastiCache handles automatic failover
# Monitor the failover progress:
aws elasticache describe-events \
```text
--source-type cache-cluster \
--source-identifier staging-valkey

```
# 3. Workers will automatically reconnect
# Monitor for connection recovery:
kubectl logs -n staging -l app=phoenix-router | grep -i redis

### 📊 Manual Backlog Processing

```text
# 1. Check current backlog
```
redis-cli -h $REDIS_HOST XLEN phoenix:router:input
redis-cli -h $REDIS_HOST XLEN phoenix:parser:input
redis-cli -h $REDIS_HOST XLEN phoenix:grouper:input

# 2. Scale up aggressively
kubectl scale deployment/phoenix-router -n staging --replicas=10
kubectl scale deployment/phoenix-parser -n staging --replicas=10
kubectl scale deployment/phoenix-grouper -n staging --replicas=10

# 3. Monitor processing rate
watch -n 5 'kubectl logs -n staging -l app=phoenix-router --since=1m | grep "processing_complete" | wc -l'

# 4. Scale back down when backlog cleared
kubectl scale deployment/phoenix-router -n staging --replicas=2
kubectl scale deployment/phoenix-parser -n staging --replicas=2
kubectl scale deployment/phoenix-grouper -n staging --replicas=2

### 🧪 Test Single Record Processing

```text
# 1. Manually inject a test message
```
redis-cli -h $REDIS_HOST XADD phoenix:router:input * \
```text
record_id "mr-test-1234-test" \
user_id "test-usr-001"

```
# 2. Watch logs for this specific record
kubectl logs -n staging -l app=phoenix-router -f | grep "mr-test-1234-test"

# 3. Check if record was processed successfully
redis-cli -h $REDIS_HOST GET processed:router:mr-test-1234-test
redis-cli -h $REDIS_HOST GET processed:parser:mr-test-1234-test
redis-cli -h $REDIS_HOST GET processed:grouper:mr-test-1234-test

# 4. Check if it ended up in DLQ
redis-cli -h $REDIS_HOST LRANGE phoenix:router:dlq 0 -1 | grep "mr-test-1234-test"

### 🔍 Incident Response Quick Reference

## Redis Key Reference

### Common Redis Commands

```text
🔧 Redis Operations
# Queue operations
```
XLEN phoenix:router:input                # Queue depth
XPENDING phoenix:router:input router-group  # Pending messages
XINFO GROUPS phoenix:router:input        # Consumer group info
XINFO CONSUMERS phoenix:router:input router-group  # Consumer info

# Lock operations
GET lock:router:mr-xxxx-xxxx-xxx                  # Check lock owner
TTL lock:router:mr-xxxx-xxxx-xxx                  # Check lock TTL

# Idempotency operations
GET processed:router:mr-xxxx-xxxx-xxx             # Check if processed
GETEX processed:router:mr-xxxx-xxxx-xxx EX 86400  # Get and refresh TTL

# DLQ operations
LLEN phoenix:router:dlq                  # DLQ size
LRANGE phoenix:router:dlq 0 10           # View DLQ entries

# Cache operations
GET api_key:user:xxx-xxx-xxx                # Check cached API key
GET pages:mr-xxxx-xxxx-xxx                        # Check cached pages

## Technologies & Tools

### Technology Tags

### Design Decisions

- **Simpler CI/CD:** One build pipeline, three image tags
- **Shared Dependencies:** All workers use the same phoenix_shared library
- **Consistent Versioning:** All services deploy together
- **Configuration-Driven:** SERVICE_NAME env var determines behavior

- **Grouper Processing Time:** Can take up to 60 seconds
- **Safety Margin:** 2x processing time for unexpected delays
- **Automatic Cleanup:** Locks expire if pod crashes
- **No Manual Intervention:** Dead locks self-heal

## 🚀 Adding New Services to Phoenix

### Developer Guide

Phoenix Pipeline is designed to be easily extensible. Adding a new processing service takes less than an hour and requires **zero changes to the core infrastructure**. Follow this step-by-step guide to add your own service.

#### Quick Start: What You'll Create

In this guide, you'll add a new service called `enricher` that processes documents between the parser and grouper stages. The same pattern works for any new service!

from .base import BaseProcessor
import structlog

logger = structlog.get_logger()

class EnricherProcessor(BaseProcessor):
```json
"""Enriches medical documents with additional metadata"""

async def initialize(self) -> None:
"""Initialize any resources (models, connections, etc.)"""
logger.info("enricher_initialized")

async def process(
self,
record_id: str,
user_id: str,
data: Dict
) -> Optional[Dict]:
"""Process a document and return data for next stage"""

logger.info(
"enricher_processing",
record_id=record_id,
user_id=user_id
)

# Your business logic here
# Call your enrichment package/API
enrichment_result = await enrich_document(record_id, user_id)

# Return data to pass to next stage
return {
"enrichment_count": enrichment_result.count,
"status": "completed"
}

async def cleanup(self) -> None:
"""Clean up resources on shutdown"""
logger.info("enricher_cleanup")

```
from processors.enricher import EnricherProcessor

# Add to processor registry
PROCESSORS = {
```text
"router": RouterProcessor,
"parser": ParserProcessor,
"enricher": EnricherProcessor,  # ← ADD THIS
"grouper": GrouperProcessor,
```
}

# Install your enrichment package
ARG GITHUB_TOKEN
RUN pip install git+https://${GITHUB_TOKEN}@github.com/n1healthcare/enrichment-package.git

# Copy Phoenix worker code
COPY worker/src /app/worker/src
WORKDIR /app/worker/src

# Set service name
ENV SERVICE_NAME=enricher

# Health check
HEALTHCHECK --interval=30s --timeout=3s --retries=3 \
  CMD curl -f http://localhost:8080/health || exit 1

CMD ["python", "-u", "main.py"]

```text
--build-arg GITHUB_TOKEN=$GITHUB_TOKEN \
-t phoenix-enricher:latest .

```
XGROUP CREATE phoenix:enricher:input enricher-group 0 MKSTREAM

# Create dead letter queue
LPUSH phoenix:enricher:dlq ""
LPOP phoenix:enricher:dlq

SERVICE_NAME=enricher
INPUT_QUEUE=phoenix:enricher:input
OUTPUT_QUEUE=phoenix:grouper:input
CONSUMER_GROUP=enricher-group
DLQ_NAME=phoenix:enricher:dlq
LOCK_TIMEOUT=600
IDEMPOTENCY_TTL=86400
PROCESSING_TIMEOUT=600

```text
-f ./environments/staging/phoenix-enricher.yaml \
-n staging

```
# Verify deployment
kubectl get pods -n staging -l service=phoenix-enricher
kubectl logs -n staging -l service=phoenix-enricher --tail=50

kind: ScaledObject
metadata:
  name: phoenix-enricher-scaler
spec:
  scaleTargetRef:
```text
name: phoenix-enricher
```
  minReplicaCount: 2
  maxReplicaCount: 10
  cooldownPeriod: 180
  triggers:
    - type: redis-streams
```text
metadata:
addressFromEnv: REDIS_URL
stream: phoenix:enricher:input
consumerGroup: enricher-group
lagCount: "10"  # Scale when lag > 10

```
- Automatic horizontal scaling
- Circuit breaker protection
- Distributed locking
- Idempotency guarantees
- Full observability (metrics, logs, traces)
- Dead letter queue for failures

### Post-Deployment Checklist

### Common Patterns & Tips

### ✅ Best Practices: What You SHOULD Do

### ❌ Anti-Patterns: What You Should NOT Do

### ⚡ Performance Guidelines

- **Keep memory usage under 2GB per pod** - Allows more pods per node
- **Don't load large models in `initialize()`** unless you need them for every request
- **Use generators/streams for large files** - Don't load entire files into memory
- **Clean up after processing** - Delete temp files, close connections, clear large variables
- **Monitor pod memory in production** - Set Kubernetes memory limits appropriately

async def process_large_file(file_path):
```text
async with aiofiles.open(file_path, 'r') as f:
async for line in f:
await process_line(line)

```
async def process_large_file(file_path):
```text
with open(file_path, 'r') as f:
data = f.read()  # ❌ Loads entire file
for line in data.split('\n'):
process_line(line)

```
async def process(self, record_id, user_id, data):
```json
try:
result = await external_api.process(record_id)
except httpx.HTTPStatusError as e:
if e.response.status_code == 404:
# Permanent failure - move to DLQ immediately
raise N1APIRecordNotFoundError(
f"Record not found: {record_id}"
)
# Other errors - let them bubble for retry
raise

return result

try:
result = await external_api.process(record_id)
except Exception as e:
# ❌ Swallowing exception prevents retry
logger.error("processing_failed", error=str(e))
return {"status": "failed"}  # ❌ Looks like success!

return result

```
### 🔒 Security Considerations

### 🧪 Testing Your Service

#### Local Testing Workflow

enricher:
  build:
```text
context: .
dockerfile: docker/Dockerfile.enricher
```
  environment:
```text
SERVICE_NAME: enricher
INPUT_QUEUE: phoenix:enricher:input
OUTPUT_QUEUE: phoenix:grouper:input
```
  depends_on:
    - redis

# 2. Start services
docker-compose up -d redis enricher

# 3. Setup queues
docker-compose exec redis redis-cli XGROUP CREATE phoenix:enricher:input enricher-group 0 MKSTREAM

# 4. Send test message
docker-compose exec redis redis-cli XADD phoenix:enricher:input * \
  record_id mr-test-1234-test \
  user_id test-usr-001

# 5. Watch logs
docker-compose logs -f enricher

# 6. Check if message was processed
docker-compose exec redis redis-cli GET processed:enricher:mr-test-1234-test

## Summary

### Phoenix Pipeline at a Glance

Phoenix Pipeline is a medical document processing system built with Redis Streams, distributed locks, circuit breakers, and comprehensive observability.

### Key Strengths

### Resources

- Repository:** `phoenix-pipeline/`
- **Architecture Guide:** `docs/ARCHITECTURE.md` (1,761 lines)
- **Operations Manual:** `docs/OPERATIONS.md` (777 lines)
- **Adding Services:** `docs/ADDING_SERVICES.md` (2,135 lines)

Phoenix Pipeline Documentation - N1 Healthcare

Medical Document Processing Infrastructure

Built with Redis Streams, Python, Kubernetes, and KEDA