feat(relay): Overhaul for production-grade reliability, performance, and observability

This commit introduces a major architectural overhaul of the RMB Relay, focusing on self-healing resilience, high-load performance, and comprehensive monitoring. Core components have been re-engineered to be more robust, efficient, and transparent in a production environment.

Architectural & Reliability Overhaul:

* Federation on Redis Streams: The inter-relay federation mechanism is completely rebuilt on Redis Streams. This provides at-least-once delivery semantics and automatic recovery for failed or pending messages, replacing the previous, less reliable queue.
* Self-Healing Substrate Client: The Substrate client and event listener are now highly resilient to network failures. They feature automatic, lock-free reconnections using `arc-swap`, capped exponential backoff, and a "single-flight" pattern to prevent thundering herd scenarios during cache misses.

Performance & Efficiency:

* Local Peer Circuit-Breaker: Message routing now uses a fast, in-memory check for locally connected peers, bypassing expensive lookups and federation logic to significantly speed up local traffic.
* Binary Serialization: The Redis cache now uses `bincode` instead of `serde_json`, reducing data size and CPU overhead for serialization and deserialization.
* Batched Message ACKs: WebSocket message acknowledgments are now batched, drastically reducing Redis round-trips and improving throughput under heavy load.
* Fail-Fast Tag: A new `fail-fast` message tag provides immediate feedback to senders if a remote destination is known to be offline, avoiding unnecessary queuing and timeouts.

Networking & Protocol Modernization:

* Hyper 1.0 & HTTP/2: The entire networking stack has been upgraded to the modern Hyper 1.0 API. Federation traffic is now explicitly configured for HTTP/2 to leverage its performance benefits.
* Protocol Field Deprecation: The `federation` and `relays` fields in the `Envelope` protobuf message have been formally marked as deprecated, aligning the protocol with the new chain-based discovery mechanism.

Observability:

* Comprehensive Prometheus Metrics: New, detailed Prometheus metrics have been added across the application, tracking the Event Listener's status (reconnects, processed blocks), cache performance (hits, misses), and session evictions.

Author: sameh-farouk

Cargo.lock

@@ -13,30 +13,33 @@ path = "src/bins/rmb-relay.rs"
 
 
 [dependencies]
-anyhow = "1.0.56"
+anyhow = "1"
 async-trait = "0.1"
-base64 = "0.13.0"
-bb8-redis = "0.13"
+base64 = "0.13"
+bb8-redis = "0.16"
+redis = { version = "0.26", features = ["streams", "aio", "tokio-comp"] }
 clap = { version = "4", features = ["derive"] }
 futures = "0.3"
+arc-swap = "1.5"
 hex = { version = "0.4", features = ["alloc"] }
-http = "0.2.7"
+http = "1"
 log = "0.4"
 md5 = "0.7.0"
-nix = "0.24.1"
-serde = { version = "1.0.136", features = ["derive"] }
-serde_json = { version = "1.0.81" }
+nix = "0.24"
+serde = { version = "1", features = ["derive"] }
+serde_json = { version = "1" }
+bincode = "1"
 simple_logger = "2.1"
 thiserror = "1.0.31"
 tokio = { version = "1", features = ["full"] }
 tokio-retry = "0.3"
-uuid = { version = "1.1.0", features = ["v4"] }
+uuid = { version = "1", features = ["v4"] }
 jsonrpsee-core = "0.14.0"
 mime = "0.3"
 mpart-async = "0.6.0"
 url = "2.3.1"
-tokio-tungstenite = { version = "0.20", features = ["native-tls"] }
-futures-util = "0.3.25"
+tokio-tungstenite = { version = "0.21", features = ["native-tls"] }
+futures-util = "0.3"
 jwt = "0.16"
 subxt = { version = "0.28.0", features = ["substrate-compat"] }
 codec = { package = "parity-scale-codec", version = "3.0.0", default-features = false, features = [
@@ -53,13 +56,16 @@ openssl = { version = "0.10", features = ["vendored"] }
 workers = { git = "https://github.com/threefoldtech/tokio-worker-pool", branch = "main" }
 protobuf = { version = "3.2.0", features = ["with-bytes"] }
 
-hyper = { version = "0.14", features = ["full"] }
-hyper-tungstenite = "0.11"
+hyper = { version = "1", features = ["full"] }
+hyper-util = { version = "0.1", features = ["server", "http1", "http2", "tokio"] }
+http-body-util = "0.1"
+hyper-tungstenite = "0.14"
 lazy_static = "1.4.0"
-prometheus = { version = "0.13.3", features = ["process"] }
+prometheus = { version = "0.13", features = ["process"] }
 
 tfchain-client = { git = "https://github.com/threefoldtech/tfchain.git", version = "0.2.0", branch = "development" }
-reqwest = "0.11"
+reqwest = "0.12"
+bytes = "1"
 
 # for e2e
 bip39 = { version = "2.0.0", default-features = false }

Cargo.toml

@@ -62,8 +62,8 @@ message Envelope {
   optional string schema = 10;
 
   // a federation url (domain)
-  // this filed ignored by relays but left for back compatibilty with old peers
-  optional string federation = 11;
+  // deprecated: this filed ignored by relays but left for back compatibilty with old peers
+  optional string federation = 11 [deprecated = true];
 
   // pyload of the message is interpreted differently based
   // on the message filed
@@ -72,5 +72,6 @@ message Envelope {
     bytes cipher = 14;
   }
 
-  repeated string relays = 17;
+  // deprecated: relays discovery is now driven by chain/cache; this field will be removed
+  repeated string relays = 17 [deprecated = true];
 }

proto/types.proto

@@ -3,6 +3,7 @@ use std::time::Duration;
 
 use anyhow::{Context, Result};
 use clap::{builder::ArgAction, Parser};
+use prometheus::default_registry;
 use rmb::cache::RedisCache;
 use rmb::events;
 use rmb::redis;
@@ -11,7 +12,7 @@ use rmb::relay::{
     limiter::{FixedWindowOptions, Limiters},
 };
 use rmb::twin::SubstrateTwinDB;
-use tokio::sync::oneshot;
+use tokio::task::JoinHandle;
 
 /// A peer requires only which rely to connect to, and
 /// which identity (mnemonics)
@@ -65,6 +66,10 @@ struct Args {
     /// failures that occurred outside this specified period will be disregarded.
     #[clap(short = 'p', long, default_value_t = 3600)]
     ranker_period: u64,
+
+    /// maximum number of blocks to catch up without flushing cache (on reconnect)
+    #[clap(long, default_value_t = 600)]
+    catchup_threshold: u64,
 }
 
 fn set_limits() -> Result<()> {
@@ -92,7 +97,7 @@ fn set_limits() -> Result<()> {
     Ok(())
 }
 
-async fn app(args: Args, tx: oneshot::Sender<()>) -> Result<()> {
+async fn app(args: Args) -> Result<JoinHandle<()>> {
     if args.workers == 0 {
         anyhow::bail!("number of workers cannot be zero");
     }
@@ -122,12 +127,22 @@ async fn app(args: Args, tx: oneshot::Sender<()>) -> Result<()> {
     // push to the queue that depends on how fast messages are sent but we can assume an extra 10%
     // of number of workers is needed
 
-    // a wiggle is 10% of number of workers with min of 1
-    let wiggle = std::cmp::max((args.workers * 10) / 100, 1);
-    let pool_size = args.workers + wiggle;
+    // a wiggle is 10% of number of workers with min of 8
+    let wiggle = std::cmp::max((args.workers * 10) / 100, 8);
+    // federation workers scale with wiggle (existing behavior)
     let fed_size = wiggle * 2;
+    // blocking connections are from switch workers and federation consumers
+    let blocking = args.workers + fed_size;
+    // explicit headroom for ops (XADD/XACK/etc.) to avoid starvation
+    let ops_headroom = std::cmp::max(blocking / 4, 16);
+    let pool_size = blocking + ops_headroom;
 
-    log::info!("redis pool size: {}", pool_size);
+    log::info!(
+        "redis pool size: {} (blocking: {}, ops_headroom: {})",
+        pool_size,
+        blocking,
+        ops_headroom
+    );
     log::info!("switch workers: {}", args.workers);
     log::info!("federation workers: {}", fed_size);
     log::info!(
@@ -166,6 +181,7 @@ async fn app(args: Args, tx: oneshot::Sender<()>) -> Result<()> {
         )
     };
     let ranker = relay::ranker::RelayRanker::new(Duration::from_secs(args.ranker_period));
+    rmb::cache::register_cache_metrics(default_registry());
     let r = relay::Relay::new(
         args.domains.iter().cloned().collect(),
         twins,
@@ -176,68 +192,55 @@ async fn app(args: Args, tx: oneshot::Sender<()>) -> Result<()> {
     )
     .context("failed to initialize relay")?;
 
-    let mut l = events::Listener::new(args.substrate, redis_cache).await?;
-    tokio::spawn(async move {
-        let max_retries = 9; // max wait is 2^9 = 512 seconds ( 5 minutes )
-        let mut attempt = 0;
-        let mut backoff = Duration::from_secs(1);
-        let mut got_hit = false;
-
-        loop {
-            match l
-                .listen(&mut got_hit)
-                .await
-                .context("failed to listen to chain events")
-            {
-                Ok(_) => break,
-                Err(e) => {
-                    if got_hit {
-                        log::warn!("Listener got a hit, but failed to listen to chain events before no attempts will be reset");
-                        got_hit = false;
-                        attempt = 0;
-                        backoff = Duration::from_secs(1);
-                    }
-                    attempt += 1;
-                    if attempt > max_retries {
-                        log::error!("Listener failed after {} attempts: {:?}", attempt - 1, e);
-                        let _ = tx.send(());
-                        break;
-                    }
-                    log::warn!(
-                        "Listener failed on attempt {}: {:?}. Retrying in {:?}...",
-                        attempt,
-                        e,
-                        backoff
-                    );
-                    tokio::time::sleep(backoff).await;
-                    backoff *= 2;
+    let mut l = events::EventListenerOptions::new()
+        .with_catchup_threshold(args.catchup_threshold)
+        .build(args.substrate, redis_cache)
+        .await?;
+    let listener_handler = tokio::spawn(async move {
+        // The listener self-heals and retries internally; this task should run indefinitely.
+        if let Err(e) = l.listen().await.context("failed to listen to chain events") {
+            log::error!("Listener exited with error: {:?}", e);
+        }
+    });
+
+    let relay_handler = tokio::spawn(async move {
+        r.start(&args.listen).await.unwrap();
+    });
+
+    let main_handler = tokio::spawn(async move {
+        tokio::select! {
+            _ = relay_handler => {
+                log::info!("Relay is closing successfully.");
+            }
+            result = listener_handler => {
+                match result {
+                    Ok(_) => log::warn!("Listener task finished unexpectedly."),
+                    Err(e) => log::error!("Listener panicked: {:?}", e),
                 }
             }
         }
     });
 
-    r.start(&args.listen).await.unwrap();
-    Ok(())
+    Ok(main_handler)
 }
 
 #[tokio::main]
 async fn main() {
     let args = Args::parse();
-    let (tx, rx) = oneshot::channel();
-    let app_handle = tokio::spawn(async move {
-        if let Err(e) = app(args, tx).await {
+    let app_handle = match app(args).await {
+        Ok(handles) => handles,
+        Err(e) => {
             eprintln!("{:#}", e);
             std::process::exit(1);
         }
-    });
+    };
 
     tokio::select! {
         _ = app_handle => {
             log::info!("Application is closing successfully.");
         }
-        _ = rx => {
-            log::error!("Listener shutdown signal received. Exiting application.");
-            std::process::exit(1);
+        _ = tokio::signal::ctrl_c() => {
+            log::info!("Ctrl-C received. Shutting down...");
         }
     }
 }

src/bins/rmb-relay.rs

@@ -3,7 +3,9 @@ use std::sync::Arc;
 use std::time::Duration;
 use tokio::sync::RwLock;
 
-use super::Cache;
+#[cfg(feature = "tracker")]
+use super::RMB_CACHE_HITS_TOTAL;
+use super::{Cache, RMB_CACHE_ENTRIES, RMB_CACHE_FLUSHES_TOTAL, RMB_CACHE_MISSES_TOTAL};
 use anyhow::Result;
 use ttl_cache::TtlCache;
 
@@ -38,21 +40,35 @@ where
 {
     async fn set<K: ToString + Send + Sync>(&self, key: K, obj: T) -> Result<()> {
         let mut mem = self.mem.write().await;
-        mem.insert(key.to_string(), obj, self.ttl);
+        let prev = mem.insert(key.to_string(), obj, self.ttl);
+        if prev.is_none() {
+            RMB_CACHE_ENTRIES.inc();
+        }
 
         Ok(())
     }
 
     async fn get<K: ToString + Send + Sync>(&self, key: K) -> Result<Option<T>> {
         let mem = self.mem.read().await;
         match mem.get(&key.to_string()) {
-            None => Ok(None),
-            Some(v) => Ok(Some(v.clone())),
+            None => {
+                RMB_CACHE_MISSES_TOTAL.inc();
+                Ok(None)
+            }
+            Some(v) => {
+                #[cfg(feature = "tracker")]
+                {
+                    RMB_CACHE_HITS_TOTAL.inc();
+                }
+                Ok(Some(v.clone()))
+            }
         }
     }
     async fn flush(&self) -> Result<()> {
         let mut mem = self.mem.write().await;
         mem.clear();
+        RMB_CACHE_ENTRIES.set(0);
+        RMB_CACHE_FLUSHES_TOTAL.inc();
 
         Ok(())
     }

src/cache/memory.rs

@@ -9,6 +9,45 @@ use std::{
     marker::{Send, Sync},
 };
 
+use lazy_static::lazy_static;
+use prometheus::{IntCounter, IntGauge, Registry};
+
+#[cfg(feature = "tracker")]
+lazy_static! {
+    static ref RMB_CACHE_HITS_TOTAL: IntCounter = IntCounter::new(
+        "rmb_cache_hits_total",
+        "Number of cache hits when fetching twins",
+    )
+    .unwrap();
+}
+
+lazy_static! {
+    static ref RMB_CACHE_MISSES_TOTAL: IntCounter = IntCounter::new(
+        "rmb_cache_misses_total",
+        "Number of cache misses when fetching twins (RPC fallback likely)",
+    )
+    .unwrap();
+    static ref RMB_CACHE_FLUSHES_TOTAL: IntCounter = IntCounter::new(
+        "rmb_cache_flushes_total",
+        "Number of times the twin cache has been flushed",
+    )
+    .unwrap();
+    static ref RMB_CACHE_ENTRIES: IntGauge = IntGauge::new(
+        "rmb_cache_entries",
+        "Current number of entries in the twin cache",
+    )
+    .unwrap();
+}
+
+pub fn register_cache_metrics(registry: &Registry) {
+    // Best-effort register into provided registry; ignore duplicate registration errors.
+    #[cfg(feature = "tracker")]
+    let _ = registry.register(Box::new(RMB_CACHE_HITS_TOTAL.clone()));
+    let _ = registry.register(Box::new(RMB_CACHE_MISSES_TOTAL.clone()));
+    let _ = registry.register(Box::new(RMB_CACHE_FLUSHES_TOTAL.clone()));
+    let _ = registry.register(Box::new(RMB_CACHE_ENTRIES.clone()));
+}
+
 pub trait Cache<T>: Send + Sync + 'static {
     fn set<S: ToString + Send + Sync>(
         &self,