glide-core: Phase 2 client-side caching - server-assisted invalidation via CLIENT TRACKING

affonsov · affonsov · commit 125fbf99b801 · 2026-05-15T19:25:49.000-07:00
Signed-off-by: affonsov &lt;67347924+affonsov@users.noreply.github.com&gt;
diff --git a/glide-core/redis-rs/redis/src/aio/mod.rs b/glide-core/redis-rs/redis/src/aio/mod.rs
@@ -225,6 +225,37 @@ where
         }
     }
 
+    if connection_info.server_assisted_cache {
+        if connection_info.protocol == ProtocolVersion::RESP2 {
+            return Err(RedisError::from((
+                ErrorKind::InvalidClientConfig,
+                "server_assisted_cache requires RESP3 protocol",
+            )));
+        }
+        match cmd("CLIENT")
+            .arg("TRACKING")
+            .arg("ON")
+            .arg("BCAST")
+            .query_async(con)
+            .await
+        {
+            Ok(Value::Okay) => {}
+            Err(e) => {
+                return Err(RedisError::from((
+                    ErrorKind::ClientError,
+                    "Failed to enable server-assisted client tracking",
+                    e.to_string(),
+                )));
+            }
+            _ => {
+                return Err(RedisError::from((
+                    ErrorKind::ClientError,
+                    "Unexpected response from CLIENT TRACKING ON BCAST",
+                )));
+            }
+        }
+    }
+
     if discover_az {
         update_az_from_info(con).await?;
     }
diff --git a/glide-core/redis-rs/redis/src/aio/multiplexed_connection.rs b/glide-core/redis-rs/redis/src/aio/multiplexed_connection.rs
@@ -113,6 +113,7 @@ pin_project! {
         disconnect_notifier: Option<Box<dyn DisconnectNotifier>>,
         is_stream_closed: Arc<AtomicBool>,
         response_sync_lost: bool,
+        cache: Option<Arc<dyn GlideCache>>,
     }
 
         impl<T> PinnedDrop for PipelineSink<T> {
@@ -140,6 +141,7 @@ where
         push_manager: Arc<ArcSwap<PushManager>>,
         disconnect_notifier: Option<Box<dyn DisconnectNotifier>>,
         is_stream_closed: Arc<AtomicBool>,
+        cache: Option<Arc<dyn GlideCache>>,
     ) -> Self
     where
         T: Sink<SinkItem, Error = RedisError> + Stream<Item = RedisResult<Value>> + 'static,
@@ -152,6 +154,7 @@ where
             disconnect_notifier,
             is_stream_closed,
             response_sync_lost: false,
+            cache,
         }
     }
 
@@ -194,6 +197,26 @@ where
         if let Ok(res) = &result {
             if let Value::Push { kind, data: _data } = res {
                 self_.push_manager.load().try_send_raw(res);
+                if kind == &PushKind::Invalidate {
+                    if let Some(cache) = self_.cache {
+                        match _data.first() {
+                            Some(Value::Array(keys)) => {
+                                for key in keys {
+                                    if let Value::BulkString(k) = key {
+                                        cache.invalidate(k);
+                                    } else if let Value::VerbatimString { text, .. } = key {
+                                        cache.invalidate(text.as_bytes());
+                                    }
+                                }
+                            }
+                            Some(Value::Nil) => {
+                                cache.flush_all();
+                            }
+                            None => { /* malformed push, ignore */ }
+                            _ => {}
+                        }
+                    }
+                }
                 if !kind.has_reply() {
                     return;
                 }
@@ -503,6 +526,7 @@ where
     fn new<T>(
         sink_stream: T,
         disconnect_notifier: Option<Box<dyn DisconnectNotifier>>,
+        cache: Option<Arc<dyn GlideCache>>,
     ) -> (Self, impl Future<Output = ()>)
     where
         T: Sink<SinkItem, Error = RedisError> + Stream<Item = RedisResult<Value>> + 'static,
@@ -521,6 +545,7 @@ where
             push_manager.clone(),
             disconnect_notifier,
             is_stream_closed.clone(),
+            cache,
         );
         let f = stream::poll_fn(move |cx| receiver.poll_recv(cx))
             .map(Ok)
@@ -702,8 +727,11 @@ impl MultiplexedConnection {
         let codec = ValueCodec::default()
             .framed(stream)
             .and_then(|msg| async move { msg });
-        let (mut pipeline, driver) =
-            Pipeline::new(codec, glide_connection_options.disconnect_notifier);
+        let (mut pipeline, driver) = Pipeline::new(
+            codec,
+            glide_connection_options.disconnect_notifier,
+            connection_info.redis.cache.clone(),
+        );
         let driver = Box::pin(driver);
         let pm = PushManager::new(
             glide_connection_options.push_sender,
@@ -1160,7 +1188,7 @@ mod tests {
         };
 
         // Create pipeline but don't drive it, the channel will fill and send() will block
-        let (mut pipeline, driver) = Pipeline::new(stalling_sink, None);
+        let (mut pipeline, driver) = Pipeline::new(stalling_sink, None, None);
         std::mem::forget(driver);
 
         // Fill the 50-slot pipeline channel
@@ -1299,7 +1327,7 @@ mod tests {
             waker: None,
         };
 
-        let (pipeline, driver) = Pipeline::new(stream, None);
+        let (pipeline, driver) = Pipeline::new(stream, None, None);
         let driver_handle = tokio::spawn(driver);
 
         // Send first command — this should go through fine
diff --git a/glide-core/redis-rs/redis/src/cache/glide_cache.rs b/glide-core/redis-rs/redis/src/cache/glide_cache.rs
@@ -456,6 +456,9 @@ pub trait GlideCache: Send + Sync + Debug {
     /// * `key` - The key to invalidate
     fn invalidate(&self, key: &[u8]);
 
+    /// Removes all entries from the cache
+    fn flush_all(&self);
+
     // ==================== Metrics ====================
 
     /// Returns current cache metrics (hits, misses, etc.)
@@ -637,6 +640,20 @@ impl<S: EvictionStrategy + 'static> GlideCache for GlideCacheImpl<S> {
         }
     }
 
+    fn flush_all(&self) {
+        let mut store = self.store.write().unwrap();
+        while let Some(entry) = store.evict_one() {
+            self.core.uncharge(entry.size);
+        }
+        if let Some(stats) = self.core.stats() {
+            stats.record_invalidation();
+        }
+        debug!(
+            "cache_flush_all - [{}] Flushed all entries",
+            store.policy_name()
+        );
+    }
+
     fn entry_count(&self) -> u64 {
         self.store.read().unwrap().len() as u64
     }
diff --git a/glide-core/redis-rs/redis/src/cluster.rs b/glide-core/redis-rs/redis/src/cluster.rs
@@ -1028,6 +1028,7 @@ pub(crate) fn get_connection_info(
             protocol: cluster_params.protocol,
             db: cluster_params.database_id,
             cache: cluster_params.cache,
+            server_assisted_cache: cluster_params.server_assisted_cache,
         },
     })
 }
diff --git a/glide-core/redis-rs/redis/src/cluster_client.rs b/glide-core/redis-rs/redis/src/cluster_client.rs
@@ -50,6 +50,7 @@ struct BuilderParams {
     database_id: i64,
     tcp_nodelay: bool,
     cache: Option<Arc<dyn GlideCache>>,
+    server_assisted_cache: bool,
     address_resolver: Option<Arc<dyn AddressResolver>>,
 }
 
@@ -155,6 +156,7 @@ pub struct ClusterParams {
     pub(crate) database_id: i64,
     pub(crate) tcp_nodelay: bool,
     pub(crate) cache: Option<Arc<dyn GlideCache>>,
+    pub(crate) server_assisted_cache: bool,
     /// Optional callback for resolving addresses before connection.
     pub(crate) address_resolver: Option<Arc<dyn AddressResolver>>,
 }
@@ -190,6 +192,7 @@ impl ClusterParams {
             database_id: value.database_id,
             tcp_nodelay: value.tcp_nodelay,
             cache: value.cache,
+            server_assisted_cache: value.server_assisted_cache,
             address_resolver: value.address_resolver,
         })
     }
@@ -222,6 +225,7 @@ impl ClusterParams {
             database_id: 0,
             tcp_nodelay: false,
             cache: None,
+            server_assisted_cache: false,
             address_resolver: None,
         }
     }
@@ -593,6 +597,12 @@ impl ClusterClientBuilder {
         self
     }
 
+    /// Sets whether server-assisted client-side caching (CLIENT TRACKING) is enabled.
+    pub fn server_assisted_cache(mut self, enabled: bool) -> ClusterClientBuilder {
+        self.builder_params.server_assisted_cache = enabled;
+        self
+    }
+
     /// Use `build()`.
     #[deprecated(since = "0.22.0", note = "Use build()")]
     pub fn open(self) -> RedisResult<ClusterClient> {
diff --git a/glide-core/redis-rs/redis/src/connection.rs b/glide-core/redis-rs/redis/src/connection.rs
@@ -237,6 +237,8 @@ pub struct RedisConnectionInfo {
     pub lib_name: Option<String>,
     /// Optionally a cache used for client-side caching
     pub cache: Option<Arc<dyn GlideCache>>,
+    /// Whether to enable server-assisted client tracking (CLIENT TRACKING ON BCAST)
+    pub server_assisted_cache: bool,
 }
 
 impl FromStr for ConnectionInfo {
@@ -396,6 +398,7 @@ fn url_to_tcp_connection_info(url: url::Url) -> RedisResult<ConnectionInfo> {
             client_name: None,
             lib_name: None,
             cache: None,
+            server_assisted_cache: false,
         },
     })
 }
@@ -430,6 +433,7 @@ fn url_to_unix_connection_info(url: url::Url) -> RedisResult<ConnectionInfo> {
             client_name: None,
             lib_name: None,
             cache: None,
+            server_assisted_cache: false,
         },
     })
 }
@@ -1878,6 +1882,7 @@ mod tests {
                         client_name: None,
                         lib_name: None,
                         cache: None,
+                        server_assisted_cache: false,
                     },
                 },
             ),
diff --git a/glide-core/src/client/mod.rs b/glide-core/src/client/mod.rs
@@ -203,6 +203,12 @@ pub async fn get_valkey_connection_info(
             )
         });
 
+    let server_assisted_cache = connection_request
+        .client_side_cache
+        .as_ref()
+        .map(|c| c.server_assisted)
+        .unwrap_or(false);
+
     match &connection_request.authentication_info {
         Some(info) => {
             // If we have IAM configuration and a token manager, use the IAM token as password
@@ -222,6 +228,7 @@ pub async fn get_valkey_connection_info(
                     client_name,
                     lib_name,
                     cache,
+                    server_assisted_cache,
                 }
             } else {
                 // Regular password-based authentication
@@ -233,6 +240,7 @@ pub async fn get_valkey_connection_info(
                     client_name,
                     lib_name,
                     cache,
+                    server_assisted_cache,
                 }
             }
         }
@@ -242,6 +250,7 @@ pub async fn get_valkey_connection_info(
             client_name,
             lib_name,
             cache,
+            server_assisted_cache,
             ..Default::default()
         },
     }
@@ -1797,6 +1806,7 @@ async fn create_cluster_client(
     builder = builder.use_protocol(request.protocol.unwrap_or_default());
     builder = builder.database_id(valkey_connection_info.db);
     builder = builder.cache(valkey_connection_info.cache);
+    builder = builder.server_assisted_cache(valkey_connection_info.server_assisted_cache);
     if let Some(client_name) = valkey_connection_info.client_name {
         builder = builder.client_name(client_name);
     }
diff --git a/glide-core/src/client/types.rs b/glide-core/src/client/types.rs
@@ -72,6 +72,7 @@ pub struct ClientSideCache {
     pub entry_ttl_ms: u64,
     pub eviction_policy: Option<EvictionPolicy>,
     pub enable_metrics: bool,
+    pub server_assisted: bool,
 }
 
 /// Authentication information for connecting to Redis/Valkey servers
@@ -367,6 +368,7 @@ impl From<protobuf::ConnectionRequest> for ConnectionRequest {
                         protobuf::EvictionPolicy::LFU => EvictionPolicy::Lfu,
                     }),
                 enable_metrics: proto_cache.enable_metrics,
+                server_assisted: proto_cache.server_assisted,
             });
 
         // Convert protobuf compression config to internal compression config
diff --git a/glide-core/src/protobuf/connection_request.proto b/glide-core/src/protobuf/connection_request.proto
@@ -97,6 +97,7 @@ message ClientSideCache {
     uint64 entry_ttl_ms = 3; // 0 = no expiration
     optional EvictionPolicy eviction_policy = 4;
     bool enable_metrics = 5;
+    bool server_assisted = 6;
 }
 
 enum EvictionPolicy {
diff --git a/glide-core/tests/test_cache.rs b/glide-core/tests/test_cache.rs

Original file line number	Diff line number	Diff line change
`@@ -1028,6 +1028,7 @@ pub(crate) fn get_connection_info(`
`1028`	`1028`	`protocol: cluster_params.protocol,`
`1029`	`1029`	`db: cluster_params.database_id,`
`1030`	`1030`	`cache: cluster_params.cache,`
	`1031`	`+ server_assisted_cache: cluster_params.server_assisted_cache,`
`1031`	`1032`	`},`
`1032`	`1033`	`})`
`1033`	`1034`	`}`
Original file line number	Diff line number	Diff line change
`@@ -97,6 +97,7 @@ message ClientSideCache {`
`97`	`97`	`uint64 entry_ttl_ms = 3; // 0 = no expiration`
`98`	`98`	`optional EvictionPolicy eviction_policy = 4;`
`99`	`99`	`bool enable_metrics = 5;`
	`100`	`+ bool server_assisted = 6;`
`100`	`101`	`}`
`101`	`102`
`102`	`103`	`enum EvictionPolicy {`