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

```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=<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<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)

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