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
2. Checks Redis cache for user’s API key (24h TTL)
3. On cache miss, calls billing service
4. Swaps OPENAI_API_KEY environment variable
5. Processes document (all LLM calls billed to user)
6. 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

# 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

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.

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
2. Setup telemetry (traces + metrics)
3. Initialize QueueManager and connect to Redis
4. Initialize lock, idempotency, circuit breaker, API key managers
5. Select processor based on SERVICE_NAME env var
6. Create Worker with all components
7. Start health server on port 8080
8. 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/ directory on GitHub.

🐳 Build Process
# 1. Build base image (Python 3.12 + shared deps)

docker build -f docker/Dockerfile.base \

--build-arg GITHUB_TOKEN=$GITHUB_TOKEN \
-t phoenix-base .

2. Build service images (install processor packages)

docker build -f docker/Dockerfile.router \

--build-arg GITHUB_TOKEN=$GITHUB_TOKEN \
-t phoenix-router .

docker build -f docker/Dockerfile.parser \

--build-arg GITHUB_TOKEN=$GITHUB_TOKEN \
-t phoenix-parser .

docker build -f docker/Dockerfile.grouper \

--build-arg GITHUB_TOKEN=$GITHUB_TOKEN \
-t phoenix-grouper .

Kubernetes Deployment

☸️ Deploy with Helm
# Deploy Router

helm upgrade –install phoenix-router ./charts/phoenix-router \

-f ./environments/staging/phoenix-router.yaml -n staging

Deploy Parser

helm upgrade –install phoenix-parser ./charts/phoenix-parser \

-f ./environments/staging/phoenix-parser.yaml -n staging

Deploy Grouper

helm upgrade –install phoenix-grouper ./charts/phoenix-grouper \

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

{

“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

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

# 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

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:

memory: 4Gi  # Increase from 2Gi

requests:

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

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

# 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

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

# 1. Check Redis cluster status

aws elasticache describe-cache-clusters \

--cache-cluster-id staging-valkey \
--show-cache-node-info

2. ElastiCache handles automatic failover

Monitor the failover progress:

aws elasticache describe-events \

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

# 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

# 1. Manually inject a test message

redis-cli -h $REDIS_HOST XADD phoenix:router:input * \

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

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

"""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 = {

"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”]

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

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

name: phoenix-enricher

minReplicaCount: 2 maxReplicaCount: 10 cooldownPeriod: 180 triggers: - type: redis-streams

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

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

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

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:

context: .
dockerfile: docker/Dockerfile.enricher

environment:

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