Merge pull request #427 from AdExNetwork/fix-redis-test-pool-broken-pipe

Fix redis test pool broken pipe

Author: elpiel

sentry/Cargo.toml

@@ -36,7 +36,7 @@ migrant_lib = { version = "^0.32", features = ["d-postgres"] }
 # Logger
 slog = { version = "^2.2.3", features = ["max_level_trace"] }
 # Serde
-serde = { version = "^1.0", features = ['derive'] }
+serde = { version = "^1.0", features = ["derive"] }
 serde_json = "^1.0"
 serde_urlencoded = "^0.7"
 # Other

sentry/src/db.rs

@@ -402,7 +402,7 @@ pub mod tests_postgres {
 pub mod redis_pool {
 
     use dashmap::DashMap;
-    use deadpool::managed::{Manager as ManagerTrait, RecycleError, RecycleResult};
+    use deadpool::managed::{Manager as ManagerTrait, RecycleResult};
     use thiserror::Error;
 
     use crate::db::redis_connection;
@@ -420,6 +420,7 @@ pub mod redis_pool {
     #[derive(Clone)]
     pub struct Database {
         available: bool,
+        index: u8,
         pub connection: MultiplexedConnection,
     }
 
@@ -475,6 +476,8 @@ pub mod redis_pool {
     pub enum Error {
         #[error("A redis error occurred")]
         Redis(#[from] RedisError),
+        #[error("Creation of new database connection failed")]
+        CreationFailed,
     }
 
     #[async_trait]
@@ -483,45 +486,58 @@ pub mod redis_pool {
         type Error = Error;
 
         async fn create(&self) -> Result<Self::Type, Self::Error> {
-            loop {
-                for mut record in self.connections.iter_mut() {
-                    let database = record.value_mut().as_mut();
-
-                    match database {
-                        Some(database) if database.available => {
-                            database.available = false;
-                            return Ok(database.clone());
-                        }
-                        // if Some but not available, skip it
-                        Some(_) => continue,
-                        None => {
-                            let mut redis_conn =
-                                redis_connection(&format!("{}{}", Self::URL, record.key())).await?;
-
-                            // run `FLUSHDB` to clean any leftovers of previous tests
-                            // even from different test runs as there might be leftovers
-                            Self::flush_db(&mut redis_conn).await?;
-
-                            let database = Database {
-                                available: false,
-                                connection: redis_conn,
-                            };
-
-                            *record.value_mut() = Some(database.clone());
-
-                            return Ok(database);
-                        }
+            for mut record in self.connections.iter_mut() {
+                let database = record.value_mut().as_mut();
+
+                match database {
+                    Some(database) if database.available => {
+                        database.available = false;
+                        return Ok(database.clone());
+                    }
+                    // if Some but not available, skip it
+                    Some(database) if !database.available => continue,
+                    // if there is no connection or it's available
+                    // always create a new redis connection because of a known issue in redis
+                    // see https://github.com/mitsuhiko/redis-rs/issues/325
+                    _ => {
+                        let mut redis_conn =
+                            redis_connection(&format!("{}{}", Self::URL, record.key()))
+                                .await
+                                .expect("Should connect");
+
+                        // run `FLUSHDB` to clean any leftovers of previous tests
+                        // even from different test runs as there might be leftovers
+                        // flush never fails as an operation
+                        Self::flush_db(&mut redis_conn).await.expect("Should flush");
+
+                        let database = Database {
+                            available: false,
+                            index: *record.key(),
+                            connection: redis_conn,
+                        };
+
+                        *record.value_mut() = Some(database.clone());
+
+                        return Ok(database);
                     }
                 }
             }
+
+            Err(Error::CreationFailed)
         }
 
         async fn recycle(&self, database: &mut Database) -> RecycleResult<Self::Error> {
-            // run `FLUSHDB` to clean any leftovers of previous tests
-            Self::flush_db(&mut database.connection)
+            // always make a new connection because of know redis crate issue
+            // see https://github.com/mitsuhiko/redis-rs/issues/325
+            let connection = redis_connection(&format!("{}{}", Self::URL, database.index))
                 .await
-                .map_err(RecycleError::Backend)?;
+                .expect("Should connect");
+            // make the database available
             database.available = true;
+            database.connection = connection;
+            Self::flush_db(&mut database.connection)
+                .await
+                .expect("Should flush");
 
             Ok(())
         }

sentry/src/routes/campaign.rs

@@ -792,7 +792,7 @@ mod test {
                 spender: create.creator,
                 channel: create.channel.clone(),
                 deposit: Deposit {
-                    // a deposit equal to double the Campaign Budget
+                    // a deposit 4 times larger than the Campaign Budget
                     total: UnifiedNum::from(200_000_000_000),
                     still_on_create2: UnifiedNum::from(0),
                 },