connection-pool-fix

src/app_state.rs

@@ -6,11 +6,139 @@ use aws_config::BehaviorVersion;
 use aws_sdk_s3::{Client as S3Client, config::Credentials};
 use log::warn;
 use redis::{AsyncCommands, Client as RedisClient, aio::MultiplexedConnection};
-use std::{env, time::Duration};
+use std::{env, sync::Arc, time::Duration};
+use tokio::sync::Mutex;
+
+/// Redis Connection Pool
+#[derive(Clone)]
+pub struct RedisConnectionPool {
+    client: RedisClient,
+    max_connections: usize,
+    connections: Arc<Mutex<Vec<MultiplexedConnection>>>,
+    timeout: Duration,
+}
+
+impl RedisConnectionPool {
+    /// Создает новый connection pool
+    pub async fn new(
+        redis_url: String,
+        max_connections: usize,
+        timeout: Duration,
+    ) -> Result<Self, redis::RedisError> {
+        let client = RedisClient::open(redis_url)?;
+        let connections = Arc::new(Mutex::new(Vec::new()));
+
+        let pool = Self {
+            client,
+            max_connections,
+            connections,
+            timeout,
+        };
+
+        // Предварительно создаем несколько соединений
+        pool.warm_up_connections(3).await?;
+
+        Ok(pool)
+    }
+
+    /// Предварительно создает соединения для пула
+    async fn warm_up_connections(&self, count: usize) -> Result<(), redis::RedisError> {
+        let mut connections = self.connections.lock().await;
+
+        for _ in 0..count.min(self.max_connections) {
+            match tokio::time::timeout(self.timeout, self.client.get_multiplexed_async_connection())
+                .await
+            {
+                Ok(Ok(conn)) => {
+                    connections.push(conn);
+                }
+                Ok(Err(e)) => {
+                    warn!("Failed to create Redis connection during warm-up: {}", e);
+                    return Err(e);
+                }
+                Err(_) => {
+                    warn!("Timeout creating Redis connection during warm-up");
+                    return Err(redis::RedisError::from((
+                        redis::ErrorKind::IoError,
+                        "Connection timeout",
+                    )));
+                }
+            }
+        }
+
+        warn!(
+            "✅ Redis connection pool warmed up with {} connections",
+            connections.len()
+        );
+        Ok(())
+    }
+
+    /// Получает соединение из пула
+    pub async fn get_connection(&self) -> Result<MultiplexedConnection, redis::RedisError> {
+        let mut connections = self.connections.lock().await;
+
+        // Пытаемся взять существующее соединение
+        if let Some(conn) = connections.pop() {
+            return Ok(conn);
+        }
+
+        // Если соединений нет, создаем новое
+        drop(connections); // Освобождаем мьютекс
+
+        match tokio::time::timeout(self.timeout, self.client.get_multiplexed_async_connection())
+            .await
+        {
+            Ok(Ok(conn)) => Ok(conn),
+            Ok(Err(e)) => {
+                warn!("Failed to create new Redis connection: {}", e);
+                Err(e)
+            }
+            Err(_) => {
+                warn!("Timeout creating new Redis connection");
+                Err(redis::RedisError::from((
+                    redis::ErrorKind::IoError,
+                    "Connection timeout",
+                )))
+            }
+        }
+    }
+
+    /// Возвращает соединение обратно в пул
+    pub async fn return_connection(&self, conn: MultiplexedConnection) {
+        let mut connections = self.connections.lock().await;
+
+        if connections.len() < self.max_connections {
+            connections.push(conn);
+        }
+        // Если пул полный, соединение просто закрывается
+    }
+
+    /// Проверяет здоровье пула
+    pub async fn health_check(&self) -> bool {
+        match self.get_connection().await {
+            Ok(mut conn) => {
+                match tokio::time::timeout(Duration::from_secs(2), conn.ping::<String>()).await {
+                    Ok(Ok(_)) => {
+                        self.return_connection(conn).await;
+                        true
+                    }
+                    _ => false,
+                }
+            }
+            Err(_) => false,
+        }
+    }
+
+    /// Получает статистику пула
+    pub async fn get_stats(&self) -> (usize, usize) {
+        let connections = self.connections.lock().await;
+        (connections.len(), self.max_connections)
+    }
+}
 
 #[derive(Clone)]
 pub struct AppState {
-    pub redis: Option<MultiplexedConnection>,
+    pub redis_pool: Option<RedisConnectionPool>,
     pub storj_client: S3Client,
     pub aws_client: S3Client,
     pub bucket: String,
@@ -27,6 +155,43 @@ impl AppState {
         Self::new_with_config(security_config).await
     }
 
+    /// Получает соединение из Redis connection pool
+    pub async fn get_redis_connection(&self) -> Result<MultiplexedConnection, redis::RedisError> {
+        if let Some(ref pool) = self.redis_pool {
+            pool.get_connection().await
+        } else {
+            Err(redis::RedisError::from((
+                redis::ErrorKind::IoError,
+                "Redis pool not available",
+            )))
+        }
+    }
+
+    /// Возвращает соединение обратно в пул
+    pub async fn return_redis_connection(&self, conn: MultiplexedConnection) {
+        if let Some(ref pool) = self.redis_pool {
+            pool.return_connection(conn).await;
+        }
+    }
+
+    /// Проверяет здоровье Redis connection pool
+    pub async fn redis_health_check(&self) -> bool {
+        if let Some(ref pool) = self.redis_pool {
+            pool.health_check().await
+        } else {
+            false
+        }
+    }
+
+    /// Получает статистику Redis connection pool
+    pub async fn redis_pool_stats(&self) -> Option<(usize, usize)> {
+        if let Some(ref pool) = self.redis_pool {
+            Some(pool.get_stats().await)
+        } else {
+            None
+        }
+    }
+
     /// Инициализация с кастомной конфигурацией безопасности.
     pub async fn new_with_config(security_config: SecurityConfig) -> Self {
         log::warn!("🚀 Starting AppState initialization...");
@@ -89,64 +254,33 @@ impl AppState {
         security_config: SecurityConfig,
         redis_url: String,
     ) -> Self {
-        let redis_client = match RedisClient::open(redis_url) {
-            Ok(client) => {
-                log::warn!("✅ Redis client created successfully");
-                client
-            }
-            Err(e) => {
-                log::error!("❌ Failed to create Redis client: {}", e);
-                panic!("Redis client creation failed: {}", e);
-            }
-        };
-
         // Устанавливаем таймаут для Redis операций с graceful fallback
         log::warn!(
             "🔄 Attempting Redis connection with timeout: {}s",
             security_config.request_timeout_seconds
         );
 
-        let redis_connection = match tokio::time::timeout(
+        // Создаем Redis connection pool
+        let redis_pool = match RedisConnectionPool::new(
+            redis_url.clone(),
+            20, // max_connections согласно руководству
             Duration::from_secs(security_config.request_timeout_seconds),
-            redis_client.get_multiplexed_async_connection(),
         )
         .await
         {
-            Ok(Ok(mut conn)) => {
-                log::warn!("✅ Redis connection established");
-
-                // Тестируем подключение простой командой
-                match tokio::time::timeout(Duration::from_secs(2), conn.ping::<String>()).await {
-                    Ok(Ok(result)) => {
-                        log::warn!("✅ Redis PING successful: {}", result);
-                        Some(conn)
-                    }
-                    Ok(Err(e)) => {
-                        log::warn!("⚠️ Redis PING failed: {}", e);
-                        None
-                    }
-                    Err(_) => {
-                        log::warn!("⚠️ Redis PING timeout");
-                        None
-                    }
-                }
+            Ok(pool) => {
+                log::warn!("✅ Redis connection pool created successfully");
+                Some(pool)
             }
-            Ok(Err(e)) => {
-                log::warn!("⚠️ Redis connection failed: {}", e);
-                log::warn!("  Error type: {:?}", e.kind());
-                log::warn!("⚠️ Running in fallback mode without Redis (quotas disabled)");
-                None
-            }
-            Err(_) => {
-                log::warn!(
-                    "⚠️ Redis connection timeout after {} seconds",
-                    security_config.request_timeout_seconds
-                );
+            Err(e) => {
+                log::warn!("⚠️ Redis connection pool creation failed: {}", e);
                 log::warn!("⚠️ Running in fallback mode without Redis (quotas disabled)");
                 None
             }
         };
 
+        // Одиночное соединение больше не нужно - используем только connection pool
+
         // Получаем конфигурацию для S3 (Storj)
         let s3_access_key = env::var("STORJ_ACCESS_KEY").expect("STORJ_ACCESS_KEY must be set");
         let s3_secret_key = env::var("STORJ_SECRET_KEY").expect("STORJ_SECRET_KEY must be set");
@@ -209,7 +343,7 @@ impl AppState {
         let aws_client = S3Client::new(&aws_config);
 
         let app_state = AppState {
-            redis: redis_connection,
+            redis_pool,
             storj_client,
             aws_client,
             bucket,
@@ -226,10 +360,13 @@ impl AppState {
     pub async fn cache_filelist(&self) {
         warn!("caching AWS filelist...");
 
-        // Проверяем доступность Redis
-        let Some(mut redis) = self.redis.clone() else {
-            warn!("⚠️ Redis not available, skipping filelist caching");
-            return;
+        // Получаем соединение из пула
+        let mut redis = match self.get_redis_connection().await {
+            Ok(conn) => conn,
+            Err(_) => {
+                warn!("⚠️ Redis pool not available, skipping filelist caching");
+                return;
+            }
         };
 
         // Запрашиваем список файлов из Storj S3
@@ -249,31 +386,43 @@ impl AppState {
         }
 
         warn!("cached {} files", filelist.len());
+
+        // Возвращаем соединение в пул
+        self.return_redis_connection(redis).await;
     }
 
     /// Получает путь из ключа (имени файла) в Redis с таймаутом.
     pub async fn get_path(&self, filename: &str) -> Result<Option<String>, actix_web::Error> {
-        let Some(mut redis) = self.redis.clone() else {
-            warn!("⚠️ Redis not available, returning None for path lookup");
-            return Ok(None);
+        let mut redis = match self.get_redis_connection().await {
+            Ok(conn) => conn,
+            Err(_) => {
+                warn!("⚠️ Redis pool not available, returning None for path lookup");
+                return Ok(None);
+            }
         };
 
-        let new_path: Option<String> =
+        let result: Option<String> =
             tokio::time::timeout(self.request_timeout, redis.hget(PATH_MAPPING_KEY, filename))
                 .await
                 .map_err(|_| ErrorInternalServerError("Redis operation timeout"))?
                 .map_err(|_| ErrorInternalServerError("Failed to get path mapping from Redis"))?;
 
-        Ok(new_path)
+        // Возвращаем соединение в пул
+        self.return_redis_connection(redis).await;
+
+        Ok(result)
     }
 
     pub async fn set_path(&self, filename: &str, filepath: &str) {
-        let Some(mut redis) = self.redis.clone() else {
-            warn!(
-                "⚠️ Redis not available, skipping path caching for {}",
-                filename
-            );
-            return;
+        let mut redis = match self.get_redis_connection().await {
+            Ok(conn) => conn,
+            Err(_) => {
+                warn!(
+                    "⚠️ Redis pool not available, skipping path caching for {}",
+                    filename
+                );
+                return;
+            }
         };
 
         if let Err(e) = tokio::time::timeout(
@@ -284,16 +433,22 @@ impl AppState {
         {
             warn!("⚠️ Redis operation failed for {}: {}", filename, e);
         }
+
+        // Возвращаем соединение в пул
+        self.return_redis_connection(redis).await;
     }
 
     /// создает или получает текущее значение квоты пользователя с таймаутом
     pub async fn get_or_create_quota(&self, user_id: &str) -> Result<u64, actix_web::Error> {
-        let Some(mut redis) = self.redis.clone() else {
-            warn!(
-                "⚠️ Redis not available, returning default quota for user {}",
-                user_id
-            );
-            return Ok(0); // Возвращаем 0 как fallback
+        let mut redis = match self.get_redis_connection().await {
+            Ok(conn) => conn,
+            Err(_) => {
+                warn!(
+                    "⚠️ Redis pool not available, returning default quota for user {}",
+                    user_id
+                );
+                return Ok(0); // Возвращаем 0 как fallback
+            }
         };
         let quota_key = format!("quota:{}", user_id);
 
@@ -313,8 +468,12 @@ impl AppState {
             .map_err(|_| ErrorInternalServerError("Redis timeout setting user quota"))?
             .map_err(|_| ErrorInternalServerError("Failed to set initial user quota in Redis"))?;
 
+            // Возвращаем соединение в пул
+            self.return_redis_connection(redis).await;
             Ok(0) // Возвращаем 0 как начальную квоту
         } else {
+            // Возвращаем соединение в пул
+            self.return_redis_connection(redis).await;
             Ok(quota)
         }
     }
@@ -325,12 +484,15 @@ impl AppState {
         user_id: &str,
         bytes: u64,
     ) -> Result<u64, actix_web::Error> {
-        let Some(mut redis) = self.redis.clone() else {
-            warn!(
-                "⚠️ Redis not available, skipping quota increment for user {}",
-                user_id
-            );
-            return Ok(0); // Возвращаем 0 как fallback
+        let mut redis = match self.get_redis_connection().await {
+            Ok(conn) => conn,
+            Err(_) => {
+                warn!(
+                    "⚠️ Redis pool not available, skipping quota increment for user {}",
+                    user_id
+                );
+                return Ok(0); // Возвращаем 0 как fallback
+            }
         };
         let quota_key = format!("quota:{}", user_id);
 
@@ -354,6 +516,9 @@ impl AppState {
             .await
             .map_err(|_| ErrorInternalServerError("Redis timeout setting initial user quota"))?
             .map_err(|_| ErrorInternalServerError("Failed to set initial user quota in Redis"))?;
+
+            // Возвращаем соединение в пул
+            self.return_redis_connection(redis).await;
             return Ok(bytes);
         }
 
@@ -366,6 +531,8 @@ impl AppState {
         .map_err(|_| ErrorInternalServerError("Redis timeout incrementing user quota"))?
         .map_err(|_| ErrorInternalServerError("Failed to increment user quota in Redis"))?;
 
+        // Возвращаем соединение в пул
+        self.return_redis_connection(redis).await;
         Ok(new_quota)
     }