feat: impl reqpool (#447)

* feat: impl reqpool

* feat(reqpool): mock pool

* feat(reqpool): remove Pool trait

* feat(reqpool): remove memory pool

* test(reqpool): add case for multiple redis pools

* chore: allow unused_imports

* feat(reqpool): update Cargo.toml

* chore(reqpool): impl Display for Status

* feat(reqpool): remove feature "enable-mock"

reqpool/Cargo.toml

@@ -0,0 +1,28 @@
+[package]
+name = "raiko-reqpool"
+version = "0.1.0"
+authors = ["Taiko Labs"]
+edition = "2021"
+
+[dependencies]
+raiko-lib = { workspace = true }
+raiko-core = { workspace = true }
+raiko-redis-derive = { workspace = true }
+chrono = { workspace = true, features = ["serde"] }
+serde = { workspace = true }
+serde_json = { workspace = true }
+serde_with = { workspace = true }
+tracing = { workspace = true }
+tokio = { workspace = true }
+async-trait = { workspace = true }
+redis = { workspace = true }
+backoff = { workspace = true }
+derive-getters = { workspace = true }
+proc-macro2 = { workspace = true }
+quote = { workspace = true }
+syn = { workspace = true }
+alloy-primitives = { workspace = true }
+lazy_static = { workspace = true }
+
+[features]
+test-utils = []

reqpool/src/config.rs

@@ -0,0 +1,10 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+/// The configuration for the redis-backend request pool
+pub struct RedisPoolConfig {
+    /// The URL of the Redis database, e.g. "redis://localhost:6379"
+    pub redis_url: String,
+    /// The TTL of the Redis database
+    pub redis_ttl: u64,
+}

reqpool/src/lib.rs

@@ -0,0 +1,16 @@
+mod config;
+mod macros;
+mod mock;
+mod redis_pool;
+mod request;
+mod utils;
+
+// Re-export
+pub use config::RedisPoolConfig;
+pub use mock::{mock_redis_pool, MockRedisConnection};
+pub use redis_pool::Pool;
+pub use request::{
+    AggregationRequestEntity, AggregationRequestKey, RequestEntity, RequestKey,
+    SingleProofRequestEntity, SingleProofRequestKey, Status, StatusWithContext,
+};
+pub use utils::proof_key_to_hack_request_key;

reqpool/src/macros.rs

@@ -0,0 +1,44 @@
+/// This macro implements the Display trait for a type by using serde_json's pretty printing.
+/// If the type cannot be serialized to JSON, it falls back to using Debug formatting.
+///
+/// # Example
+///
+/// ```rust
+/// use serde::{Serialize, Deserialize};
+///
+/// #[derive(Debug, Serialize, Deserialize)]
+/// struct Person {
+///     name: String,
+///     age: u32
+/// }
+///
+/// impl_display_using_json_pretty!(Person);
+///
+/// let person = Person {
+///     name: "John".to_string(),
+///     age: 30
+/// };
+///
+/// // Will print:
+/// // {
+/// //   "name": "John",
+/// //   "age": 30
+/// // }
+/// println!("{}", person);
+/// ```
+///
+/// The type must implement serde's Serialize trait for JSON serialization to work.
+/// If serialization fails, it will fall back to using the Debug implementation.
+#[macro_export]
+macro_rules! impl_display_using_json_pretty {
+    ($type:ty) => {
+        impl std::fmt::Display for $type {
+            fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+                match serde_json::to_string_pretty(self) {
+                    Ok(s) => write!(f, "{}", s),
+                    Err(_) => write!(f, "{:?}", self),
+                }
+            }
+        }
+    };
+}

reqpool/src/mock.rs

@@ -0,0 +1,153 @@
+use crate::{Pool, RedisPoolConfig};
+use lazy_static::lazy_static;
+use redis::{RedisError, RedisResult};
+use serde::Serialize;
+use serde_json::{json, Value};
+use std::{
+    collections::HashMap,
+    sync::{Arc, Mutex},
+};
+
+type SingleStorage = Arc<Mutex<HashMap<Value, Value>>>;
+type GlobalStorage = Mutex<HashMap<String, SingleStorage>>;
+
+lazy_static! {
+    // #{redis_url => single_storage}
+    //
+    // We use redis_url to distinguish different redis database for tests, to prevent
+    // data race problem when running multiple tests.
+    static ref GLOBAL_STORAGE: GlobalStorage = Mutex::new(HashMap::new());
+}
+
+pub struct MockRedisConnection {
+    storage: SingleStorage,
+}
+
+impl MockRedisConnection {
+    pub fn new(redis_url: String) -> Self {
+        let mut global = GLOBAL_STORAGE.lock().unwrap();
+        Self {
+            storage: global
+                .entry(redis_url)
+                .or_insert_with(|| Arc::new(Mutex::new(HashMap::new())))
+                .clone(),
+        }
+    }
+
+    pub fn set_ex<K: Serialize, V: Serialize>(
+        &mut self,
+        key: K,
+        val: V,
+        _ttl: u64,
+    ) -> RedisResult<()> {
+        let mut lock = self.storage.lock().unwrap();
+        lock.insert(json!(key), json!(val));
+        Ok(())
+    }
+
+    pub fn get<K: Serialize, V: serde::de::DeserializeOwned>(&mut self, key: &K) -> RedisResult<V> {
+        let lock = self.storage.lock().unwrap();
+        match lock.get(&json!(key)) {
+            None => Err(RedisError::from((redis::ErrorKind::TypeError, "not found"))),
+            Some(v) => serde_json::from_value(v.clone()).map_err(|e| {
+                RedisError::from((
+                    redis::ErrorKind::TypeError,
+                    "deserialization error",
+                    e.to_string(),
+                ))
+            }),
+        }
+    }
+
+    pub fn del<K: Serialize>(&mut self, key: K) -> RedisResult<usize> {
+        let mut lock = self.storage.lock().unwrap();
+        if lock.remove(&json!(key)).is_none() {
+            Ok(0)
+        } else {
+            Ok(1)
+        }
+    }
+
+    pub fn keys<K: serde::de::DeserializeOwned>(&mut self, key: &str) -> RedisResult<Vec<K>> {
+        assert_eq!(key, "*", "mock redis only supports '*'");
+
+        let lock = self.storage.lock().unwrap();
+        Ok(lock
+            .keys()
+            .map(|k| serde_json::from_value(k.clone()).unwrap())
+            .collect())
+    }
+}
+
+/// Return the mock redis pool with the given id.
+///
+/// This is used for testing. Please use the test case name as the id to prevent data race.
+pub fn mock_redis_pool<S: ToString>(id: S) -> Pool {
+    let config = RedisPoolConfig {
+        redis_ttl: 111,
+        redis_url: format!("redis://{}:6379", id.to_string()),
+    };
+    Pool::open(config).unwrap()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use redis::RedisResult;
+
+    #[test]
+    fn test_mock_redis_pool() {
+        let mut pool = mock_redis_pool("test_mock_redis_pool");
+        let mut conn = pool.conn().expect("mock conn");
+
+        let key = "hello".to_string();
+        let val = "world".to_string();
+        conn.set_ex(key.clone(), val.clone(), 111)
+            .expect("mock set_ex");
+
+        let actual: RedisResult<String> = conn.get(&key);
+        assert_eq!(actual, Ok(val));
+
+        let _ = conn.del(&key);
+        let actual: RedisResult<String> = conn.get(&key);
+        assert!(actual.is_err());
+    }
+
+    #[test]
+    fn test_mock_multiple_redis_pool() {
+        let mut pool1 = mock_redis_pool("test_mock_multiple_redis_pool_1");
+        let mut pool2 = mock_redis_pool("test_mock_multiple_redis_pool_2");
+        let mut conn1 = pool1.conn().expect("mock conn");
+        let mut conn2 = pool2.conn().expect("mock conn");
+
+        let key = "hello".to_string();
+        let world = "world".to_string();
+
+        {
+            conn1
+                .set_ex(key.clone(), world.clone(), 111)
+                .expect("mock set_ex");
+            let actual: RedisResult<String> = conn1.get(&key);
+            assert_eq!(actual, Ok(world.clone()));
+        }
+
+        {
+            let actual: RedisResult<String> = conn2.get(&key);
+            assert!(actual.is_err());
+        }
+
+        {
+            let meme = "meme".to_string();
+            conn2
+                .set_ex(key.clone(), meme.clone(), 111)
+                .expect("mock set_ex");
+            let actual: RedisResult<String> = conn2.get(&key);
+            assert_eq!(actual, Ok(meme));
+        }
+
+        {
+            let actual: RedisResult<String> = conn1.get(&key);
+            assert_eq!(actual, Ok(world));
+        }
+    }
+}