# 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 ```bash # 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= # Fallback (if GATEWAY_URL not set) # Uses InMemoryQueueAdapter for development ``` ### QueueWorkerConfig struct ```rust 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, // 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) ```bash # 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: ```rust // 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 ```bash # 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: ```rust 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: ```rust 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: ```bash # In development: # Check queue adapter's failed_messages state # In production: # Query OpenSearch DLQ index for analysis ``` --- ## Performance Tuning ### Throughput Optimization ```bash # 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 ```bash # 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: ```bash # 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 ```bash 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 ```bash 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): ```bash # 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**: ```bash # 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**: ```bash 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**: ```bash # 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**: ```bash # 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](./M8.2-DUAL_WRITE_INDEXER.md) - [Gateway Queue Adapter](./M8.2-GATEWAY_QUEUE_ADAPTER.md) - [Queue Adapter Trait](../crates/mem-cli/src/queue_adapter.rs) - SQS Concepts: https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/