poimen-memory/docs/M8.2-QUEUE_WORKER_INTEGRATION.md

M8.2 — Queue Worker Integration with DualWriteIndexer

Status: Complete
Architecture: Background task for concurrent dual-write processing
Concurrency: Multiple workers can process queue messages in parallel

---

Overview

The Queue Worker decouples the fast ingest path from the slow dual-write operations (embedding → pgvector + OpenSearch). This improves throughput and reliability:

Before (Synchronous)

IngestWorker
  ├─ Parse document
  ├─ Split into chunks
  ├─ Embed each chunk (slow, sequential)
  ├─ Write to pgvector (slow, I/O)
  ├─ Write to OpenSearch (slow, I/O)
  └─ Return to user [TOTAL: 5-10 seconds]

After (Asynchronous with Queue)

IngestWorker                    QueueWorker (background task)
  ├─ Parse document                ├─ receive_chunks(10, 30s)
  ├─ Split into chunks             ├─ embed_one() for each
  ├─ queue.send_chunk()            ├─ write_pgvector()
  └─ Return immediately (fast)     ├─ write_opensearch()
     [TOTAL: <100ms]               └─ delete/retry cycle

---

Architecture

Data Flow

┌──────────────┐
│ IngestWorker │
├──────────────┤
│ parse doc    │
│ split chunks │
│ queue each   │ ──send_chunk()──>  ┌────────────────┐
│ return 202   │                    │ Gateway Queue  │
└──────────────┘                    │  (api.riotpiao)│
                                    └────────────────┘
                                           ▲  │
                                           │  │
                                receive_chunks(10, 30s)
                                           │  ▼
                                    ┌──────────────────┐
                                    │  QueueWorker     │
                                    ├──────────────────┤
                                    │ for each msg:    │
                                    │  - embed_one()   │
                                    │  - write_pgvec() │
                                    │  - write_os()    │
                                    │  - delete/retry  │
                                    └──────────────────┘

Message Lifecycle

1. QUEUED — Message in queue, waiting for worker pickup
2. RECEIVED — Message checked out (visibility timeout active)
3. PROCESSING — Worker embedding/writing
   • SUCCESS → DELETE from queue
   • FAILURE (pgvector) → EXTEND visibility, retry
   • FAILURE (OpenSearch) → Mark pending, delete from queue
   • MAX RETRIES → SEND TO DLQ
4. PROCESSED or DLQ — Final state

---

Configuration

Environment Variables

# Queue Worker Enable/Disable
ENABLE_QUEUE_WORKER=true                   # Default: true

# Message Processing
QUEUE_BATCH_SIZE=10                        # Max messages per receive (1-10)
QUEUE_VISIBILITY_TIMEOUT=300               # Seconds before retry (5 min)
QUEUE_WAIT_TIME=20                         # Long-poll timeout (0-20s)
QUEUE_MAX_RETRIES=3                        # Retries before DLQ
QUEUE_PROJECT=                             # Optional: process specific project only

# Gateway (if using GatewayQueueAdapter)
GATEWAY_URL=https://api.riotpiao.com
AUTHENTIK_ISSUER=https://authentik.riotpiao.com/application/o/poimen-memory/
AUTHENTIK_CLIENT_ID=poimen-memory
AUTHENTIK_CLIENT_SECRET=<secret>

# Fallback (if GATEWAY_URL not set)
# Uses InMemoryQueueAdapter for development

QueueWorkerConfig struct

pub struct QueueWorkerConfig {
    pub max_messages_per_batch: i32,         // 1-10
    pub visibility_timeout_secs: i32,        // 30-600 recommended
    pub wait_time_secs: i32,                 // 0-20
    pub project: Option<String>,             // Filter by project
    pub max_retries: i32,                    // 2-5 typical
    pub retry_backoff_initial_secs: i32,     // 60 default
    pub empty_poll_interval_secs: u64,       // 5 default
    pub enable_metrics: bool,                // Collect stats
}

---

Usage

Starting the Server (with Queue Worker)

# Kubernetes
kubectl set env deployment/poimen-memory \
  ENABLE_QUEUE_WORKER=true \
  QUEUE_BATCH_SIZE=10 \
  GATEWAY_URL=https://api.riotpiao.com

# Local development
ENABLE_QUEUE_WORKER=true \
QUEUE_BATCH_SIZE=5 \
cargo run --bin mem -- serve --port 9090

Queue Worker is Automatic

The queue worker starts automatically when:

1. ENABLE_QUEUE_WORKER=true (default)
2. HTTP server starts
3. Spawned as background tokio task

No additional code needed:

// http_server.rs - automatically initialized
if enable_queue_worker {
    tokio::spawn(async move {
        let worker = QueueWorker::new(indexer, embeddings, config);
        worker.start().await  // Runs forever (long-polling loop)
    });
}

Monitoring Queue Worker

# Check logs
kubectl logs -f deployment/poimen-memory | grep "Queue worker"

# Expected output
# INFO Queue worker starting: config=QueueWorkerConfig { ... }
# INFO M8.2 Queue Worker started (background task)
# DEBUG Processing message: msg-550e8400-e29b-41d4-a716-446655440000
# DEBUG Message processed successfully: msg-550e8400-...

Metrics

The QueueWorker tracks:

pub struct WorkerMetrics {
    pub messages_received: u64,       // Total received from queue
    pub messages_processed: u64,      // Successfully processed
    pub messages_failed: u64,         // Failed (will retry)
    pub messages_dlq: u64,            // Sent to DLQ (max retries)
    pub total_processing_time_ms: u64, // Cumulative processing time
}

Access metrics:

let metrics = worker.metrics().await;
println!("Processed: {}", metrics.messages_processed);
println!("Failed: {}", metrics.messages_failed);
println!("Avg time/msg: {}ms", 
    metrics.total_processing_time_ms / metrics.messages_processed.max(1));

---

Error Handling

Retry Logic

1. pgvector write fails → Extend visibility (300s), retry
2. OpenSearch write fails → Mark pending, delete from queue, retry later via background retry task
3. Max retries exceeded → Send to DLQ, alert operators

DLQ (Dead-Letter Queue)

Messages are sent to DLQ when:

• receive_count >= max_retries (default: 3)
• pgvector consistently fails (data issues)
• Invalid message format

DLQ messages can be examined via:

# In development:
# Check queue adapter's failed_messages state

# In production:
# Query OpenSearch DLQ index for analysis

---

Performance Tuning

Throughput Optimization

# For high-volume workloads
QUEUE_BATCH_SIZE=10          # Max messages per poll
QUEUE_VISIBILITY_TIMEOUT=300 # 5 min timeout
QUEUE_WAIT_TIME=20           # Full 20s long-poll

# Result: ~100 msgs/sec (depends on embedding latency)

Latency Optimization

# For low-latency requirements
QUEUE_BATCH_SIZE=1           # Process one at a time
QUEUE_VISIBILITY_TIMEOUT=60  # 1 min timeout
QUEUE_WAIT_TIME=1            # Short poll

# Result: Faster feedback, lower throughput

Resource Constraints

If embedding service is slow:

# Run multiple worker replicas
kubectl scale deployment/poimen-memory --replicas=3

# Each replica runs its own QueueWorker
# Total concurrency = 3 × QUEUE_BATCH_SIZE = 30 messages

---

Testing

Unit Tests

cargo test --lib queue_worker

Tests cover:

• Config validation
• Message roundtrip (send → receive → delete)
• Batch operations (multiple messages)
• DLQ transitions
• Attributes preservation
• Stats tracking

Integration Tests

cargo test --test it_queue_worker_integration

Tests verify:

• Full pipeline (IngestWorker → Queue → DualWriteIndexer)
• Message lifecycle states
• Error handling and retries
• Concurrent processing

Local Development

Use in-memory adapter (no GATEWAY_URL):

# Development server
ENABLE_QUEUE_WORKER=true \
QUEUE_BATCH_SIZE=3 \
cargo run --bin mem -- serve --port 9090

# Queue worker logs
# ...INFO M8.2 Queue Worker started
# ...DEBUG Received 0 messages from queue (max_messages=3)
# ...INFO Queue empty, waiting 5s before retry

# Test ingestion
curl -X POST http://localhost:9090/memory/ingest \
  -H "apikey: test-key" \
  -H "Content-Type: application/json" \
  -d '{"project":"test", "source":"cli", "ingest_id":"123", "records":[{"text":"hello"}]}'

# Watch worker process it

---

Deployment Checklist

• ENABLE_QUEUE_WORKER=true set in K8s env
• GATEWAY_URL and Authentik credentials configured (if using gateway)
• Queue topic/queue created in message broker (if applicable)
• OpenSearch cluster healthy (for dual-write)
• Embedding service accessible and responsive
• Replica count ≥ 1 (recommended: 2-3 for HA)
• Logs monitored for "Queue worker error"
• Health checks passing (/health)
• DLQ monitoring set up (alert on high DLQ count)

---

Troubleshooting

Queue Worker Not Starting

Symptom: No "Queue worker starting" in logs

Check:

# Verify env var
kubectl get deployment poimen-memory -o json | \
  jq '.spec.template.spec.containers[0].env' | grep ENABLE_QUEUE_WORKER

# Verify logs
kubectl logs deployment/poimen-memory | grep -i "queue worker"

Fix:

kubectl set env deployment/poimen-memory ENABLE_QUEUE_WORKER=true
kubectl rollout restart deployment/poimen-memory

Messages Stuck in Queue

Symptom: Queue not emptying, messages keep retrying

Check:

# Check embedding service
curl http://embedding-service:8000/health

# Check OpenSearch
curl http://opensearch:9200/_cluster/health

# Check pgvector
psql -h memory-db -U app memory -c "SELECT count(*) FROM chunks;"

Fix:

• Restart embedding service if slow/hung
• Check OpenSearch cluster health
• Increase visibility timeout: QUEUE_VISIBILITY_TIMEOUT=600

Too Many DLQ Messages

Symptom: High rate of messages in DLQ

Check:

# Inspect DLQ messages
# (implementation-specific)

# Check message format
# Ensure ChunkInput JSON is valid

Fix:

• Verify ingest source is producing valid JSON
• Check for data corruption in ingest pipeline
• Increase retries: QUEUE_MAX_RETRIES=5

---

Architecture Notes

Why Async Queue?

1. Decoupling: Ingest doesn't wait for embedding + write
2. Scaling: Single ingest API handles many more requests
3. Resilience: OpenSearch failure doesn't block ingest
4. Throughput: Embeddings computed in parallel

Why Long-Polling?

Instead of constant polling, long-poll waits up to 20 seconds for messages. This:

• Reduces CPU usage (no tight loop)
• Reduces network overhead
• Achieves near-real-time processing
• Matches SQS/Kafka semantics

Why Visibility Timeout?

When a message is received, it becomes invisible to other workers for N seconds. This prevents:

• Duplicate processing (if one worker crashes)
• Race conditions (two workers on same message)
• Lost messages (message stays in queue until ack'd)

---

References

• M8.2 Dual-Write Indexer
• Gateway Queue Adapter
• Queue Adapter Trait
• SQS Concepts: https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/