fix(sdk): reconnect on dropped socket via ConnectionManager

sdk/rust/Cargo.toml

@@ -10,7 +10,7 @@ keywords = ["redis", "vector-search", "graph", "database", "async"]
 categories = ["database", "asynchronous", "network-programming"]
 
 [dependencies]
-redis = { version = "1.1", features = ["tokio-comp", "aio"] }
+redis = { version = "1.1", features = ["tokio-comp", "aio", "connection-manager"] }
 tokio = { version = "1", features = ["rt", "net", "macros"] }
 thiserror = "2"
 bytes = "1"

sdk/rust/src/cache.rs

@@ -9,7 +9,7 @@ use crate::util::parse_cache_search_results;
 ///
 /// Obtain via [`MoonClient::cache`](crate::MoonClient::cache).
 pub struct CacheClient {
-    pub(crate) conn: redis::aio::MultiplexedConnection,
+    pub(crate) conn: redis::aio::ConnectionManager,
 }
 
 impl CacheClient {

sdk/rust/src/client.rs

@@ -30,17 +30,27 @@ macro_rules! cmd {
 /// use moondb::MoonClient;
 ///
 /// #[tokio::main]
-/// async fn main() -> moon::Result<()> {
+/// async fn main() -> moondb::Result<()> {
 ///     let mut client = MoonClient::connect("redis://127.0.0.1:6399").await?;
 ///     client.set("hello", "world").await?;
 ///     let v: String = client.get("hello").await?;
 ///     println!("{v}");
 ///     Ok(())
 /// }
 /// ```
+/// The connection is a [`redis::aio::ConnectionManager`], NOT a bare
+/// `MultiplexedConnection`: the manager transparently RECONNECTS (with
+/// backoff) when the underlying socket drops — a server restart, an
+/// idle-timeout reap, a transient network blip. A `MultiplexedConnection`
+/// opens one socket at connect time and never rebuilds it, so after any drop
+/// every later command fails with `broken pipe` forever until the whole
+/// client is reconstructed. That wedged Lunaris in production after a live
+/// Moon daemon flip; see `sdk/rust/tests/reconnect.rs`. `ConnectionManager`
+/// is `Clone` and shares its reconnect state across clones via `Arc`, so a
+/// heal on any handle repairs every sub-client.
 #[derive(Clone)]
 pub struct MoonClient {
-    pub(crate) conn: redis::aio::MultiplexedConnection,
+    pub(crate) conn: redis::aio::ConnectionManager,
 }
 
 impl MoonClient {
@@ -52,7 +62,7 @@ impl MoonClient {
     /// - `rediss://127.0.0.1:6399` — TLS (requires `tls-rustls` or `tls-native-tls` feature)
     pub async fn connect(url: impl redis::IntoConnectionInfo) -> Result<Self> {
         let client = redis::Client::open(url)?;
-        let conn = client.get_multiplexed_async_connection().await?;
+        let conn = client.get_connection_manager().await?;
         Ok(Self { conn })
     }
 
@@ -65,16 +75,21 @@ impl MoonClient {
         response_timeout: std::time::Duration,
     ) -> Result<Self> {
         let client = redis::Client::open(url)?;
-        let config = redis::AsyncConnectionConfig::new().set_response_timeout(Some(response_timeout));
-        let conn = client
-            .get_multiplexed_async_connection_with_config(&config)
-            .await?;
+        // `ConnectionManagerConfig` carries the same per-command response
+        // timeout as the old `AsyncConnectionConfig` path, so the io-failsafe
+        // contract (every command is response-bounded) is preserved — while
+        // ALSO gaining transparent reconnect on a dropped socket.
+        let config =
+            redis::aio::ConnectionManagerConfig::new().set_response_timeout(Some(response_timeout));
+        let conn = client.get_connection_manager_with_config(config).await?;
         Ok(Self { conn })
     }
 
-    /// Access the raw underlying multiplexed connection for operations not covered
-    /// by the typed API.
-    pub fn inner_mut(&mut self) -> &mut redis::aio::MultiplexedConnection {
+    /// Access the raw underlying reconnecting connection for operations not
+    /// covered by the typed API. `ConnectionManager` implements `ConnectionLike`,
+    /// so `redis::cmd(..).query_async(client.inner_mut())` works exactly as it
+    /// did against the old `MultiplexedConnection`.
+    pub fn inner_mut(&mut self) -> &mut redis::aio::ConnectionManager {
         &mut self.conn
     }
 

sdk/rust/src/graph.rs

@@ -11,7 +11,7 @@ use crate::util::{parse_query_result, value_to_i64};
 /// use moondb::MoonClient;
 ///
 /// #[tokio::main]
-/// async fn main() -> moon::Result<()> {
+/// async fn main() -> moondb::Result<()> {
 ///     let client = MoonClient::connect("redis://127.0.0.1:6399").await?;
 ///     let mut g = client.graph();
 ///     g.create("social").await?;
@@ -26,7 +26,7 @@ use crate::util::{parse_query_result, value_to_i64};
 /// }
 /// ```
 pub struct GraphClient {
-    pub(crate) conn: redis::aio::MultiplexedConnection,
+    pub(crate) conn: redis::aio::ConnectionManager,
 }
 
 impl GraphClient {

sdk/rust/src/mq.rs

@@ -9,7 +9,7 @@ use crate::util::value_to_string;
 ///
 /// Obtain via [`MoonClient::mq`](crate::MoonClient::mq).
 pub struct MqClient {
-    pub(crate) conn: redis::aio::MultiplexedConnection,
+    pub(crate) conn: redis::aio::ConnectionManager,
 }
 
 impl MqClient {

sdk/rust/src/session.rs

@@ -9,7 +9,7 @@ use crate::util::parse_search_results;
 ///
 /// Obtain via [`MoonClient::session`](crate::MoonClient::session).
 pub struct SessionClient {
-    pub(crate) conn: redis::aio::MultiplexedConnection,
+    pub(crate) conn: redis::aio::ConnectionManager,
 }
 
 impl SessionClient {

sdk/rust/src/temporal.rs

@@ -7,7 +7,7 @@ use crate::error::Result;
 ///
 /// Obtain via [`MoonClient::temporal`](crate::MoonClient::temporal).
 pub struct TemporalClient {
-    pub(crate) conn: redis::aio::MultiplexedConnection,
+    pub(crate) conn: redis::aio::ConnectionManager,
 }
 
 impl TemporalClient {

sdk/rust/src/text.rs

@@ -101,7 +101,7 @@ fn encode_filter(cmd: &mut redis::Cmd, filter: &HybridFilter, top_level: bool) {
 ///
 /// Obtain via [`MoonClient::text`](crate::MoonClient::text).
 pub struct TextClient {
-    pub(crate) conn: redis::aio::MultiplexedConnection,
+    pub(crate) conn: redis::aio::ConnectionManager,
 }
 
 impl TextClient {
@@ -174,9 +174,7 @@ impl TextClient {
         // Client-side filter validation (same depth/leaf limits as the server parser).
         if let Some(f) = filter {
             let mut validator = FilterValidator::new();
-            validator
-                .validate(f, 0)
-                .map_err(MoonError::invalid_arg)?;
+            validator.validate(f, 0).map_err(MoonError::invalid_arg)?;
         }
 
         let blob = encode_vector(query_vec);
@@ -222,7 +220,9 @@ impl TextClient {
     ///
     /// # Example
     /// ```no_run
+    /// # use moondb::MoonClient;
     /// use moondb::types::Reducer;
+    /// # async fn demo(client: &mut MoonClient) -> moondb::Result<()> {
     /// let rows = client.text().aggregate(
     ///     "docs",
     ///     "*",
@@ -231,6 +231,8 @@ impl TextClient {
     ///     Some(("count", false)),
     ///     Some(100),
     /// ).await?;
+    /// # Ok(())
+    /// # }
     /// ```
     pub async fn aggregate(
         &mut self,

sdk/rust/src/vector.rs

@@ -11,7 +11,7 @@ use crate::util::{parse_cache_search_results, parse_index_info, parse_search_res
 /// use moondb::{MoonClient, types::{VectorIndexOptions, DistanceMetric}};
 ///
 /// #[tokio::main]
-/// async fn main() -> moon::Result<()> {
+/// async fn main() -> moondb::Result<()> {
 ///     let client = MoonClient::connect("redis://127.0.0.1:6399").await?;
 ///     let mut v = client.vector();
 ///     v.create_index("idx", VectorIndexOptions::new(384, DistanceMetric::Cosine)).await?;
@@ -20,7 +20,7 @@ use crate::util::{parse_cache_search_results, parse_index_info, parse_search_res
 /// }
 /// ```
 pub struct VectorClient {
-    pub(crate) conn: redis::aio::MultiplexedConnection,
+    pub(crate) conn: redis::aio::ConnectionManager,
 }
 
 impl VectorClient {

sdk/rust/src/workspace.rs

@@ -11,7 +11,7 @@ use std::collections::HashMap;
 ///
 /// Obtain via [`MoonClient::workspace`](crate::MoonClient::workspace).
 pub struct WorkspaceClient {
-    pub(crate) conn: redis::aio::MultiplexedConnection,
+    pub(crate) conn: redis::aio::ConnectionManager,
 }
 
 impl WorkspaceClient {

sdk/rust/tests/integration.rs

@@ -331,7 +331,7 @@ async fn test_graph_explain() {
 fn vector_encode_decode_roundtrip() {
     let original = vec![0.1_f32, -0.5, 1.23456, f32::MIN_POSITIVE];
     let encoded = encode_vector(&original);
-    let decoded = moon::decode_vector(&encoded).unwrap();
+    let decoded = moondb::decode_vector(&encoded).unwrap();
     for (a, b) in original.iter().zip(decoded.iter()) {
         assert!((a - b).abs() < 1e-7, "mismatch: {a} vs {b}");
     }
@@ -340,5 +340,5 @@ fn vector_encode_decode_roundtrip() {
 #[test]
 fn vector_decode_invalid_length() {
     let bad = vec![0u8, 1, 2]; // 3 bytes — not a multiple of 4
-    assert!(moon::decode_vector(&bad).is_err());
+    assert!(moondb::decode_vector(&bad).is_err());
 }