BM-1484: Bento redis to keydb migration

bento/Cargo.toml

@@ -18,17 +18,23 @@ repository = "https://github.com/risc0/bento/"
 
 [workspace.dependencies]
 anyhow = "1.0.98"
-aws-sdk-s3 = "1.34"                                                       # used for minio for max compatibility
+aws-sdk-s3 = "1.34" # used for minio for max compatibility
 bento-client = { path = "crates/bento-client" }
 bincode = "1.3"
 bonsai-sdk = { version = "1.4.0", features = ["non_blocking"] }
 bytemuck = "1.16"
 clap = { version = "4.5", features = ["derive", "env"] }
 deadpool-redis = "0.15"
 hex = { version = "0.4", default-features = false, features = ["alloc"] }
-redis = { version = "0.25", features = ["tokio-comp"] }
+redis = { version = "0.25", features = [
+  "tokio-comp",
+  "cluster",
+  "connection-manager",
+] }
 risc0-build = { version = "2.3.1" }
-risc0-zkvm = { version = "2.3.1", features = ["unstable"], default-features = false }
+risc0-zkvm = { version = "2.3.1", features = [
+  "unstable",
+], default-features = false }
 sample-guest-common = { path = "crates/sample-guest/common" }
 sample-guest-methods = { path = "crates/sample-guest/methods" }
 serde = { version = "1.0", features = ["derive"] }

bento/crates/workflow/Cargo.toml

@@ -18,8 +18,15 @@ clap = { workspace = true, features = ["env", "derive"] }
 deadpool-redis = { workspace = true }
 hex = { workspace = true }
 nix = { version = "0.29", features = ["fs"] }
-redis = { workspace = true, features = ["tokio-rustls-comp", "tokio-comp"] }
-risc0-zkvm = { workspace = true, default-features = false, features = ["prove"] }
+redis = { workspace = true, features = [
+    "tokio-rustls-comp",
+    "tokio-comp",
+    "cluster",
+    "connection-manager",
+] }
+risc0-zkvm = { workspace = true, default-features = false, features = [
+    "prove",
+] }
 serde = { workspace = true }
 serde_json = { workspace = true }
 signal-hook = "0.3"

bento/crates/workflow/src/redis.rs

@@ -30,6 +30,35 @@ where
     }
 }
 
+/// Batch set multiple keys with expiry using Redis pipelining
+pub async fn batch_set_keys_with_expiry<T>(
+    conn: &mut deadpool_redis::Connection,
+    key_value_pairs: Vec<(String, T)>,
+    ttl: Option<u64>,
+) -> RedisResult<()>
+where
+    T: ToRedisArgs + Send + Sync + 'static,
+{
+    if key_value_pairs.is_empty() {
+        return Ok(());
+    }
+
+    // Use Redis pipelining for bulk operations
+    let mut pipe = redis::pipe();
+
+    for (key, value) in key_value_pairs {
+        if let Some(expiry) = ttl {
+            pipe.set_ex(key, value, expiry);
+        } else {
+            pipe.set(key, value);
+        }
+    }
+
+    // Execute the pipeline
+    pipe.query_async::<_, ()>(conn).await?;
+    Ok(())
+}
+
 /// Scan and delete all keys at a given prefix
 pub async fn scan_and_delete(conn: &mut Connection, prefix: &str) -> RedisResult<()> {
     // Initialize the cursor for SCAN