feat: refactor federation to use Redis Streams for reliable message delivery

Author: sameh-farouk

Cargo.lock

@@ -16,7 +16,8 @@ path = "src/bins/rmb-relay.rs"
 anyhow = "1.0.56"
 async-trait = "0.1"
 base64 = "0.13.0"
-bb8-redis = "0.13"
+bb8-redis = "0.16"
+redis = { version = "0.26", features = ["streams", "aio", "tokio-comp"] }
 clap = { version = "4", features = ["derive"] }
 futures = "0.3"
 hex = { version = "0.4", features = ["alloc"] }

Cargo.toml

@@ -53,7 +53,7 @@ where
             .arg(&self.prefix)
             .arg(key.to_string())
             .arg(obj)
-            .query_async::<_, i64>(&mut *conn)
+            .query_async::<i64>(&mut *conn)
             .await?;
         if added == 1 {
             RMB_CACHE_ENTRIES.inc();

src/cache/redis.rs

@@ -5,7 +5,6 @@ use crate::types::{self, Address, AddressExt, EnvelopeExt};
 use crate::types::{Backlog, Envelope};
 use anyhow::{Context, Result};
 use async_trait::async_trait;
-use bb8_redis::redis;
 pub use redis::*;
 pub use redis_storage::RedisStorage;
 use serde::{Deserialize, Serialize};
@@ -170,7 +169,7 @@ impl redis::ToRedisArgs for JsonOutgoingRequest {
 
 impl redis::FromRedisValue for JsonOutgoingRequest {
     fn from_redis_value(v: &redis::Value) -> redis::RedisResult<Self> {
-        if let redis::Value::Data(data) = v {
+        if let redis::Value::BulkString(data) = v {
             serde_json::from_slice(data).map_err(|e| {
                 redis::RedisError::from((
                     redis::ErrorKind::TypeError,
@@ -247,7 +246,7 @@ impl redis::ToRedisArgs for JsonIncomingRequest {
 
 impl redis::FromRedisValue for JsonIncomingRequest {
     fn from_redis_value(v: &redis::Value) -> redis::RedisResult<Self> {
-        if let redis::Value::Data(data) = v {
+        if let redis::Value::BulkString(data) = v {
             serde_json::from_slice(data).map_err(|e| {
                 redis::RedisError::from((
                     redis::ErrorKind::TypeError,
@@ -318,7 +317,7 @@ impl redis::ToRedisArgs for JsonOutgoingResponse {
 
 impl redis::FromRedisValue for JsonOutgoingResponse {
     fn from_redis_value(v: &redis::Value) -> redis::RedisResult<Self> {
-        if let redis::Value::Data(data) = v {
+        if let redis::Value::BulkString(data) = v {
             serde_json::from_slice(data).map_err(|e| {
                 redis::RedisError::from((
                     redis::ErrorKind::TypeError,
@@ -433,7 +432,7 @@ impl redis::ToRedisArgs for JsonIncomingResponse {
 
 impl redis::FromRedisValue for JsonIncomingResponse {
     fn from_redis_value(v: &redis::Value) -> redis::RedisResult<Self> {
-        if let redis::Value::Data(data) = v {
+        if let redis::Value::BulkString(data) = v {
             serde_json::from_slice(data).map_err(|e| {
                 redis::RedisError::from((
                     redis::ErrorKind::TypeError,

src/peer/storage/mod.rs

@@ -152,7 +152,7 @@ impl Storage for RedisStorage {
         }
         let mut conn = self.get_connection().await?;
         let key = BacklogKey(&backlog.uid);
-        let _: () = conn.set_ex(&key, backlog, backlog.ttl as usize).await?;
+        let _: () = conn.set_ex(&key, backlog, backlog.ttl).await?;
 
         Ok(())
     }
@@ -189,7 +189,7 @@ impl Storage for RedisStorage {
         let resp_queue = Queue::Response.as_ref();
         let queues = (req_queue, resp_queue);
 
-        let (queue, value): (String, Value) = conn.brpop(queues, 0).await?;
+        let (queue, value): (String, Value) = conn.brpop(queues, 0.0).await?;
 
         let msg: JsonMessage = if queue == req_queue {
             JsonOutgoingRequest::from_redis_value(&value)