🌐 Implement Connection Pooling for HTTP Clients

🌐 Implement Connection Pooling for HTTP Clients

Issue Description

Performance analysis identified inefficient HTTP client usage patterns where new connections are created for each request, causing significant overhead in network operations. The current implementation lacks connection reuse, leading to poor performance under load.

📍 Problem Areas

Primary Files:

• crates/terraphim_service/src/http_client.rs - Individual client creation per request
• crates/terraphim_service/src/openrouter.rs - LLM API calls without connection reuse
• crates/terraphim_middleware/src/haystack/mcp.rs - MCP server connections
• crates/terraphim_service/src/summarization_worker.rs - Repeated connections for summarization

Specific Issues:

1. No connection reuse: New TCP connections for each HTTP request
2. Missing keep-alive: Connections closed immediately after use
3. No request pipelining: Sequential requests without connection reuse
4. Inefficient TLS handshake: Repeated TLS negotiations for same domains

🎯 Solution Approach

Phase 1: Implement HTTP Client Pool

// Current problematic pattern:
use reqwest::Client;

pub async fn fetch_llm_response(prompt: &str) -> Result<String> {
    let client = Client::new(); // New client every time!
    let response = client
        .post("https://api.openrouter.ai/v1/chat/completions")
        .json(&serde_json::json!({
            "prompt": prompt
        }))
        .send()
        .await?;
    Ok(response.text().await?)
}

// Optimized pattern with connection pool:
use reqwest::Client;
use std::sync::Arc;

pub struct HttpClientPool {
    client: Arc<Client>,
    metrics: ConnectionMetrics,
}

impl HttpClientPool {
    pub fn new() -> Self {
        let client = Client::builder()
            .pool_max_idle_per_host(10)
            .pool_idle_timeout(Duration::from_secs(30))
            .timeout(Duration::from_secs(30))
            .user_agent("terraphim-ai/1.0")
            .build()
            .expect("Failed to create HTTP client");
            
        Self {
            client: Arc::new(client),
            metrics: ConnectionMetrics::new(),
        }
    }
    
    pub async fn get(&self) -> Arc<Client> {
        self.metrics.increment_active_connections();
        Arc::clone(&self.client)
    }
    
    pub fn release(&self) {
        self.metrics.decrement_active_connections();
    }
}

// Usage:
static HTTP_POOL: Lazy<HttpClientPool> = Lazy::new(HttpClientPool::new);

pub async fn fetch_llm_response(prompt: &str) -> Result<String> {
    let client = HTTP_POOL.get().await;
    let response = client
        .post("https://api.openrouter.ai/v1/chat/completions")
        .json(&serde_json::json!({
            "prompt": prompt
        }))
        .send()
        .await?;
    HTTP_POOL.release();
    Ok(response.text().await?)
}

Phase 2: Add Connection Metrics and Health Monitoring

use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};

#[derive(Debug)]
pub struct ConnectionMetrics {
    active_connections: AtomicUsize,
    total_requests: AtomicU64,
    failed_requests: AtomicU64,
    avg_response_time: AtomicU64, // microseconds
}

impl ConnectionMetrics {
    pub fn new() -> Self {
        Self {
            active_connections: AtomicUsize::new(0),
            total_requests: AtomicU64::new(0),
            failed_requests: AtomicU64::new(0),
            avg_response_time: AtomicU64::new(0),
        }
    }
    
    pub fn increment_active_connections(&self) {
        self.active_connections.fetch_add(1, Ordering::Relaxed);
    }
    
    pub fn decrement_active_connections(&self) {
        self.active_connections.fetch_sub(1, Ordering::Relaxed);
    }
    
    pub fn record_request(&self, duration: Duration, success: bool) {
        self.total_requests.fetch_add(1, Ordering::Relaxed);
        if !success {
            self.failed_requests.fetch_add(1, Ordering::Relaxed);
        }
        
        // Update running average
        let duration_micros = duration.as_micros() as u64;
        self.avg_response_time.store(duration_micros, Ordering::Relaxed);
    }
    
    pub fn get_stats(&self) -> ConnectionStats {
        ConnectionStats {
            active_connections: self.active_connections.load(Ordering::Relaxed),
            total_requests: self.total_requests.load(Ordering::Relaxed),
            failed_requests: self.failed_requests.load(Ordering::Relaxed),
            avg_response_time: self.avg_response_time.load(Ordering::Relaxed),
        }
    }
}

Phase 3: Implement Request Batching and Pipelining

use tokio::sync::mpsc;
use futures::stream::{self, StreamExt};

pub struct BatchedRequestHandler {
    client: Arc<Client>,
    request_queue: mpsc::Sender<BatchedRequest>,
    batch_size: usize,
    batch_timeout: Duration,
}

#[derive(Debug)]
pub struct BatchedRequest {
    id: u64,
    url: String,
    payload: serde_json::Value,
    response_tx: mpsc::oneshot::Sender<Result<serde_json::Value>>,
}

impl BatchedRequestHandler {
    pub fn new(client: Arc<Client>) -> Self {
        let (tx, rx) = mpsc::channel(1000);
        let handler = Self {
            client,
            request_queue: tx,
            batch_size: 10,
            batch_timeout: Duration::from_millis(100),
        };
        
        // Start batch processing task
        tokio::spawn(handler.process_batches(rx));
        handler
    }
    
    async fn process_batches(&self, mut rx: mpsc::Receiver<BatchedRequest>) {
        let mut batch = Vec::new();
        let mut interval = tokio::time::interval(self.batch_timeout);
        
        loop {
            tokio::select! {
                request = rx.recv() => {
                    match request {
                        Some(req) => {
                            batch.push(req);
                            if batch.len() >= self.batch_size {
                                self.process_batch(batch.drain(..).collect()).await;
                            }
                        }
                        None => break,
                    }
                }
                _ = interval.tick() => {
                    if !batch.is_empty() {
                        self.process_batch(batch.drain(..).collect()).await;
                    }
                }
            }
        }
    }
    
    async fn process_batch(&self, requests: Vec<BatchedRequest>) {
        // Process requests concurrently with connection reuse
        let futures = requests.into_iter().map(|req| {
            let client = Arc::clone(&self.client);
            async move {
                let start = Instant::now();
                let result = self.execute_single_request(&client, &req).await;
                let duration = start.elapsed();
                
                if let Err(ref e) = result {
                    log::warn!("Request {} failed: {}", req.id, e);
                }
                
                // Send response back
                let _ = req.response_tx.send(result);
                (req.id, duration, result.is_ok())
            }
        });
        
        let results = stream::iter(futures).buffer_unordered(10).collect().await;
        
        // Record metrics
        for (id, duration, success) in results {
            HTTP_POOL.metrics.record_request(duration, success);
        }
    }
}

Phase 4: Add Adaptive Connection Management

pub struct AdaptiveConnectionPool {
    pools: HashMap<String, Arc<ClientPool>>,
    config: PoolConfig,
}

#[derive(Debug, Clone)]
pub struct PoolConfig {
    pub max_connections_per_host: usize,
    pub connection_timeout: Duration,
    pub idle_timeout: Duration,
    pub max_idle_per_host: usize,
}

impl AdaptiveConnectionPool {
    pub fn new(config: PoolConfig) -> Self {
        Self {
            pools: HashMap::new(),
            config,
        }
    }
    
    pub async fn get_client_for_host(&mut self, host: &str) -> Arc<Client> {
        if !self.pools.contains_key(host) {
            let pool = Arc::new(ClientPool::new(host, &self.config).await);
            self.pools.insert(host.to_string(), pool);
        }
        
        self.pools[host].get_client().await
    }
    
    pub async fn cleanup_idle_connections(&self) {
        for pool in self.pools.values() {
            pool.cleanup_idle_connections().await;
        }
    }
}

pub struct ClientPool {
    clients: Arc<Vec<Arc<Client>>>,
    next_client: AtomicUsize,
    host: String,
}

impl ClientPool {
    async fn new(host: &str, config: &PoolConfig) -> Self {
        let mut clients = Vec::new();
        
        for _ in 0..config.max_connections_per_host {
            let client = Client::builder()
                .pool_max_idle_per_host(config.max_idle_per_host)
                .pool_idle_timeout(config.idle_timeout)
                .timeout(config.connection_timeout)
                .build()
                .expect("Failed to create client");
            clients.push(Arc::new(client));
        }
        
        Self {
            clients: Arc::new(clients),
            next_client: AtomicUsize::new(0),
            host: host.to_string(),
        }
    }
    
    pub async fn get_client(&self) -> Arc<Client> {
        let index = self.next_client.fetch_add(1, Ordering::Relaxed) % self.clients.len();
        Arc::clone(&self.clients[index])
    }
}

📋 Implementation Tasks

1. Implement basic connection pool:

   • Replace individual Client::new() calls with pool
   • Add connection reuse for same domains
   • Implement proper connection lifetime management

2. Add metrics and monitoring:

   • Track active connections and request rates
   • Monitor response times and failure rates
   • Add health checks for connection quality

3. Implement request batching:

   • Add batch processing for multiple requests
   • Implement request pipelining where supported
   • Add configurable batch sizes and timeouts

4. Add adaptive management:

   • Implement per-host connection pools
   • Add dynamic pool sizing based on load
   • Implement connection cleanup for idle resources

5. Optimize for different use cases:

   • Special handling for LLM API calls
   • Optimizations for MCP server connections
   • Different strategies for different request patterns

📊 Expected Performance Improvements

• Connection overhead: 80-90% reduction in TCP/TLS handshake time
• Request latency: 40-60% improvement for repeated requests to same host
• Throughput: 2-3x better performance under high load
• Resource usage: 50-70% reduction in file descriptors and memory

🧪 Testing Strategy

Unit Tests

• Test connection pool lifecycle management
• Verify connection reuse for same hosts
• Test metrics collection accuracy

Integration Tests

• Load testing with concurrent requests
• Connection failure and recovery testing
• Performance benchmarking against baseline

Benchmarks

• Connection establishment time measurements
• Throughput tests with different pool sizes
• Memory usage monitoring under load

🔧 Configuration Options

// Add to configuration:
pub struct HttpPoolConfig {
    pub max_connections_per_host: usize,
    pub connection_timeout_seconds: u64,
    pub idle_timeout_seconds: u64,
    pub max_idle_per_host: usize,
    pub enable_batching: bool,
    pub batch_size: usize,
    pub batch_timeout_ms: u64,
}

impl Default for HttpPoolConfig {
    fn default() -> Self {
        Self {
            max_connections_per_host: 10,
            connection_timeout_seconds: 30,
            idle_timeout_seconds: 60,
            max_idle_per_host: 5,
            enable_batching: true,
            batch_size: 10,
            batch_timeout_ms: 100,
        }
    }
}

📚 Resources

• Reqwest Connection Pooling
• HTTP/2 Connection Reuse
• Connection Pooling Best Practices

🏁 Definition of Done

• Connection pooling is implemented across all HTTP clients
• Performance benchmarks show 40%+ improvement
• Connection metrics are collected and monitored
• All existing tests pass
• New integration tests are added
• Documentation is updated
• Code review is approved

Related to: #193 (Performance Optimization Epic)
Estimated Effort: 4-5 days
Priority: Medium (Network efficiency)

Start by tracing the Client::new() usage in crates/terraphim_service/src/http_client.rs, then compare the request paths in openrouter.rs, haystack/mcp.rs, and summarization_worker.rs. Done means the listed request paths reuse connections and the proposed lifecycle, metrics, batching, adaptive management, and unit or integration checks are addressed.