From f80c3f5c96a308c0473880f34c6244500297df70 Mon Sep 17 00:00:00 2001 From: Andy Stark Date: Thu, 29 Oct 2026 16:40:29 +0000 Subject: [PATCH 1/3] DOC-7226 Use multiplexed async connections in the Rust session-store, time-series-dashboard and rate-limiter use cases The session-store and time-series-dashboard demos opened their shared connection with Client::get_async_connection() and held it as redis::aio::Connection. That non-multiplexed connection is gone from Client in redis-rs 1.x, and the leaderboard use case, which pins the same 0.24 crate, already uses get_multiplexed_async_connection() and MultiplexedConnection. Both demos and the session-store page snippet now do the same, and the rate-limiter page's async snippet names MultiplexedConnection instead of the removed type. The change is forward-compatible: both APIs exist in 0.24 and 1.x, so the crate pins are unchanged. Both demos build against redis 0.24.1. The session-store demo was run against Redis 8.10.2 and passed login, session read, increment and logout. The time-series-dashboard demo writes its samples, but its /api/snapshot endpoint returns a parse error that the unmodified demo on main returns identically, so it is not caused by this change. Learned: of the use cases that looked affected by redis-rs 1.x, only session-store and time-series-dashboard used the removed non-multiplexed connection in code; the others already use ConnectionManager or MultiplexedConnection Directive: keep these demos on get_multiplexed_async_connection(), which works on the 0.24 pins and on 1.x; do not reintroduce get_async_connection() Recheck: when the Rust use cases move to redis-rs 1.x, rerun the time-series-dashboard snapshot, which fails to parse f64 from a status reply on 0.24.1 against Redis 8.10.2 Gaps: the crate pins were not bumped, nothing was compiled against 1.x, and the prefetch-cache page still says a bare redis::aio::Connection is not Clone, which names a type that no longer exists in 1.x Ticket: DOC-7226 Co-Authored-By: Claude Sonnet 5.5 --- content/develop/use-cases/rate-limiter/rust/_index.md | 4 ++-- content/develop/use-cases/session-store/rust/_index.md | 2 +- content/develop/use-cases/session-store/rust/demo_server.rs | 6 +++--- .../use-cases/time-series-dashboard/rust/demo_server.rs | 6 +++--- 4 files changed, 9 insertions(+), 9 deletions(-) diff --git a/content/develop/use-cases/rate-limiter/rust/_index.md b/content/develop/use-cases/rate-limiter/rust/_index.md index 58d54fa1bc..856b481f21 100644 --- a/content/develop/use-cases/rate-limiter/rust/_index.md +++ b/content/develop/use-cases/rate-limiter/rust/_index.md @@ -328,11 +328,11 @@ fn handle_request( For async web frameworks like `actix-web` or `axum`, use the async version of the Redis client: ```rust -use redis::aio::Connection; +use redis::aio::MultiplexedConnection; async fn rate_limit_check( limiter: &TokenBucket, - con: &mut Connection, + con: &mut MultiplexedConnection, key: &str, ) -> redis::RedisResult { limiter.allow_async(con, key).await diff --git a/content/develop/use-cases/session-store/rust/_index.md b/content/develop/use-cases/session-store/rust/_index.md index 1e08d9bcc5..191b95e36f 100644 --- a/content/develop/use-cases/session-store/rust/_index.md +++ b/content/develop/use-cases/session-store/rust/_index.md @@ -88,7 +88,7 @@ use std::collections::HashMap; #[tokio::main] async fn main() -> redis::RedisResult<()> { let client = Client::open("redis://localhost:6379/")?; - let mut con = client.get_async_connection().await?; + let mut con = client.get_multiplexed_async_connection().await?; let store = RedisSessionStore::new("session:", 1800)?; diff --git a/content/develop/use-cases/session-store/rust/demo_server.rs b/content/develop/use-cases/session-store/rust/demo_server.rs index 1e6cadad96..83a436a812 100644 --- a/content/develop/use-cases/session-store/rust/demo_server.rs +++ b/content/develop/use-cases/session-store/rust/demo_server.rs @@ -15,7 +15,7 @@ use axum::{ routing::{get, post}, Json, Router, }; -use redis::{aio::Connection, Client}; +use redis::{aio::MultiplexedConnection, Client}; use serde::Serialize; use session_store::RedisSessionStore; use std::{collections::HashMap, env, sync::Arc}; @@ -23,7 +23,7 @@ use tokio::sync::Mutex; #[derive(Clone)] struct AppState { - redis_connection: Arc>, + redis_connection: Arc>, store: RedisSessionStore, } @@ -55,7 +55,7 @@ async fn main() { std::process::exit(1); }); - let connection = client.get_async_connection().await.unwrap_or_else(|e| { + let connection = client.get_multiplexed_async_connection().await.unwrap_or_else(|e| { eprintln!("Failed to connect to Redis at {}: {}", redis_url, e); std::process::exit(1); }); diff --git a/content/develop/use-cases/time-series-dashboard/rust/demo_server.rs b/content/develop/use-cases/time-series-dashboard/rust/demo_server.rs index 48414b58d1..37119e246f 100644 --- a/content/develop/use-cases/time-series-dashboard/rust/demo_server.rs +++ b/content/develop/use-cases/time-series-dashboard/rust/demo_server.rs @@ -13,7 +13,7 @@ use axum::{ routing::get, Json, Router, }; -use redis::{aio::Connection, Client}; +use redis::{aio::MultiplexedConnection, Client}; use sensor_simulator::{sensors, SensorSimulator}; use std::{env, sync::Arc, time::Duration}; use tokio::sync::Mutex; @@ -28,7 +28,7 @@ const GRAPH_PAD_Y: i32 = 18; #[derive(Clone)] struct AppState { - redis_connection: Arc>, + redis_connection: Arc>, store: RedisTimeSeriesStore, } @@ -46,7 +46,7 @@ async fn main() { std::process::exit(1); }); - let connection = client.get_async_connection().await.unwrap_or_else(|error| { + let connection = client.get_multiplexed_async_connection().await.unwrap_or_else(|error| { eprintln!("Failed to connect to Redis at {}: {}", redis_url, error); std::process::exit(1); }); From 51fde7437f876139c27739aec642b676f30f5781 Mon Sep 17 00:00:00 2001 From: Andy Stark Date: Fri, 30 Oct 2026 08:59:07 +0000 Subject: [PATCH 2/3] DOC-7226 Move the Rust use cases to redis-rs 1.8 All 13 Rust use-case demos now depend on redis 1.8 (they pinned 0.24 or 0.27), matching the clients/rust pages, and the use-case pages are updated to match. Compiling cleared most of it: Value::Data, Bulk and Status became BulkString, Array and SimpleString; query_async and invoke_async take one generic; ErrorKind::TypeError became UnexpectedReturnType and ResponseError became Server(ServerErrorKind::ResponseError); borrowed values go through from_redis_value_ref; scan iterators yield results; AsyncCommands needs Send + Sync. Two changes only running the demos could find. hget takes one field in 1.x (the old one silently sent HMGET for several), so multi-field reads use hmget. And 1.x puts a 500 ms response timeout on every connection manager and multiplexed connection, so the job-queue BLMOVE workers and the streaming XREADGROUP BLOCK consumers failed with "timed out" until their managers were built with the response timeout removed. I also fixed the time-series dashboard's as_f64 helper, which already failed on main against a current server: RESP2 time series doubles arrive as simple strings and the helper only accepted integers and bulk strings. Every demo was built and run against Redis 8.10.2 and exercised through its HTTP endpoints, including the two candle-based demos with the cached MiniLM model. The pages now state the Rust 1.88 minimum that redis 1.8 requires, and the prefetch-cache page no longer describes the removed non-multiplexed connection. Bugbot and review have not seen this commit yet. Learned: redis-rs 1.x applies a 500 ms response timeout to every ConnectionManager and multiplexed connection, so blocking commands such as BLMOVE and XREADGROUP BLOCK fail with "timed out" at runtime while the code still compiles Constraint: demos that block on the server (job-queue, streaming) must build their ConnectionManager with ConnectionManagerConfig::new().set_response_timeout(None) Directive: use hmget for several hash fields; hget takes exactly one field in redis-rs 1.x Recheck: the crate is declared as 1.8 and needs Rust 1.88; if the pages' stated Rust minimum or the redis version changes, update all 13 pages together Gaps: nothing was tested against redis 1.0.4 to 1.7, the page snippets were edited to mirror the .rs files but not compiled, and the demos were not run against a cluster or Sentinel Ticket: DOC-7226 Co-Authored-By: Claude Sonnet 5.5 --- .../use-cases/cache-aside/rust/Cargo.toml | 2 +- .../use-cases/cache-aside/rust/_index.md | 8 +- .../use-cases/cache-aside/rust/cache.rs | 8 +- .../use-cases/feature-store/rust/Cargo.toml | 2 +- .../use-cases/feature-store/rust/_index.md | 31 ++- .../feature-store/rust/feature_store.rs | 11 +- .../use-cases/job-queue/rust/Cargo.lock | 227 +++++------------- .../use-cases/job-queue/rust/Cargo.toml | 2 +- .../use-cases/job-queue/rust/_index.md | 15 +- .../use-cases/job-queue/rust/src/main.rs | 7 +- .../use-cases/leaderboard/rust/Cargo.toml | 2 +- .../use-cases/leaderboard/rust/_index.md | 2 +- .../use-cases/leaderboard/rust/leaderboard.rs | 6 +- .../use-cases/memory-layer/rust/Cargo.toml | 2 +- .../memory-layer/rust/src/event_log.rs | 2 +- .../memory-layer/rust/src/long_term_memory.rs | 2 +- .../memory-layer/rust/src/session_store.rs | 2 +- .../use-cases/prefetch-cache/rust/Cargo.toml | 2 +- .../use-cases/prefetch-cache/rust/_index.md | 8 +- .../use-cases/prefetch-cache/rust/cache.rs | 8 +- .../develop/use-cases/pub-sub/rust/Cargo.toml | 2 +- .../develop/use-cases/pub-sub/rust/_index.md | 2 +- .../use-cases/rate-limiter/rust/Cargo.toml | 2 +- .../use-cases/rate-limiter/rust/_index.md | 4 +- .../rate-limiter/rust/token_bucket.rs | 2 +- .../recommendation-engine/rust/Cargo.toml | 2 +- .../recommendation-engine/rust/recommender.rs | 38 ++- .../use-cases/semantic-cache/rust/Cargo.lock | 48 ++-- .../use-cases/semantic-cache/rust/Cargo.toml | 2 +- .../use-cases/semantic-cache/rust/_index.md | 2 +- .../semantic-cache/rust/src/cache.rs | 2 +- .../use-cases/session-store/rust/Cargo.toml | 2 +- .../use-cases/session-store/rust/_index.md | 4 +- .../session-store/rust/session_store.rs | 26 +- .../use-cases/streaming/rust/Cargo.toml | 2 +- .../use-cases/streaming/rust/_index.md | 15 +- .../use-cases/streaming/rust/demo_server.rs | 5 +- .../use-cases/streaming/rust/event_stream.rs | 46 ++-- .../time-series-dashboard/rust/Cargo.toml | 2 +- .../rust/timeseries_store.rs | 29 ++- 40 files changed, 255 insertions(+), 331 deletions(-) diff --git a/content/develop/use-cases/cache-aside/rust/Cargo.toml b/content/develop/use-cases/cache-aside/rust/Cargo.toml index 59b9c42c2f..0d6a9a247f 100644 --- a/content/develop/use-cases/cache-aside/rust/Cargo.toml +++ b/content/develop/use-cases/cache-aside/rust/Cargo.toml @@ -4,7 +4,7 @@ version = "0.1.0" edition = "2021" [dependencies] -redis = { version = "0.24", features = ["tokio-comp", "aio", "connection-manager"] } +redis = { version = "1.8", features = ["tokio-comp", "aio", "connection-manager"] } tokio = { version = "1", features = ["full"] } axum = "0.7" serde = { version = "1.0", features = ["derive"] } diff --git a/content/develop/use-cases/cache-aside/rust/_index.md b/content/develop/use-cases/cache-aside/rust/_index.md index f6208c5b6f..ce7dbe4faf 100644 --- a/content/develop/use-cases/cache-aside/rust/_index.md +++ b/content/develop/use-cases/cache-aside/rust/_index.md @@ -191,7 +191,7 @@ where .release_script .key(&lock_key) .arg(token.as_str()) - .invoke_async::<_, i64>(&mut conn) + .invoke_async::(&mut conn) .await; return result; } @@ -234,10 +234,10 @@ pub async fn update_field(&self, id: &str, field: &str, value: &str) -> RedisRes let cache_key = self.cache_key(id); let mut conn = self.conn.clone(); loop { - redis::cmd("WATCH").arg(&cache_key).query_async::<_, ()>(&mut conn).await?; + redis::cmd("WATCH").arg(&cache_key).query_async::<()>(&mut conn).await?; let exists: i64 = conn.exists(&cache_key).await?; if exists == 0 { - redis::cmd("UNWATCH").query_async::<_, ()>(&mut conn).await?; + redis::cmd("UNWATCH").query_async::<()>(&mut conn).await?; return Ok(false); } let result: Option<((),)> = redis::pipe() @@ -293,7 +293,7 @@ Before running the demo, make sure that: ```toml [dependencies] -redis = { version = "0.24", features = ["tokio-comp", "aio", "connection-manager"] } +redis = { version = "1.8", features = ["tokio-comp", "aio", "connection-manager"] } tokio = { version = "1", features = ["full"] } axum = "0.7" serde = { version = "1.0", features = ["derive"] } diff --git a/content/develop/use-cases/cache-aside/rust/cache.rs b/content/develop/use-cases/cache-aside/rust/cache.rs index 0c1acac282..f78b0c7791 100644 --- a/content/develop/use-cases/cache-aside/rust/cache.rs +++ b/content/develop/use-cases/cache-aside/rust/cache.rs @@ -149,7 +149,7 @@ impl RedisCache { .release_script .key(&lock_key) .arg(token.as_str()) - .invoke_async::<_, i64>(&mut conn) + .invoke_async::(&mut conn) .await; return result; } @@ -190,7 +190,7 @@ impl RedisCache { .ignore() .expire(cache_key, self.cfg.ttl as i64) .ignore() - .query_async::<_, ()>(&mut conn) + .query_async::<()>(&mut conn) .await?; } Ok(record) @@ -212,11 +212,11 @@ impl RedisCache { loop { redis::cmd("WATCH") .arg(&cache_key) - .query_async::<_, ()>(&mut conn) + .query_async::<()>(&mut conn) .await?; let exists: i64 = conn.exists(&cache_key).await?; if exists == 0 { - redis::cmd("UNWATCH").query_async::<_, ()>(&mut conn).await?; + redis::cmd("UNWATCH").query_async::<()>(&mut conn).await?; return Ok(false); } let result: Option<((),)> = redis::pipe() diff --git a/content/develop/use-cases/feature-store/rust/Cargo.toml b/content/develop/use-cases/feature-store/rust/Cargo.toml index b017cfaf14..d21178153e 100644 --- a/content/develop/use-cases/feature-store/rust/Cargo.toml +++ b/content/develop/use-cases/feature-store/rust/Cargo.toml @@ -20,7 +20,7 @@ path = "build_features_bin.rs" # demo uses. HEXPIRE/HTTL are issued via `redis::cmd("HEXPIRE")` # because the typed bindings don't ship on this client line yet — # the wire protocol is the same. -redis = { version = "0.27", features = ["tokio-comp", "aio", "connection-manager"] } +redis = { version = "1.8", features = ["tokio-comp", "aio", "connection-manager"] } tokio = { version = "1", features = ["full"] } axum = "0.7" # axum-extra ships an alternative Form extractor that uses diff --git a/content/develop/use-cases/feature-store/rust/_index.md b/content/develop/use-cases/feature-store/rust/_index.md index d36d6a270c..1ba9051d07 100644 --- a/content/develop/use-cases/feature-store/rust/_index.md +++ b/content/develop/use-cases/feature-store/rust/_index.md @@ -69,10 +69,10 @@ Two crate facts shape the helper: and they all share the same underlying connection. There's no pool to manage. * **The `redis::cmd("HEXPIRE")` builder is how you reach commands not yet - typed on the `AsyncCommands` trait.** Per-field TTL bindings (`hexpire`, - `httl`, `hpersist`) aren't part of the typed surface on redis-rs 0.27, so - the helper issues them with the generic command builder. The wire bytes - are identical to the typed form. + typed on the `AsyncCommands` trait.** The helper issues the per-field TTL + commands (`HEXPIRE`, `HTTL`, `HPERSIST`) with the generic command builder. + redis-rs 1.x also ships typed `hexpire`, `httl`, and `hpersist` methods; + the wire bytes are identical. In this example, the batch features describe a user's longer-term shape (`country_iso`, `risk_segment`, `account_age_days`, `tx_count_7d`, @@ -305,7 +305,7 @@ pub async fn update_streaming( for code in &codes { if *code != 1 { return Err(redis::RedisError::from(( - redis::ErrorKind::ResponseError, + redis::ErrorKind::Server(redis::ServerErrorKind::ResponseError), "HEXPIRE invariant violated", format!("HEXPIRE did not set every field TTL for {key}: {codes:?}"), ))); @@ -340,9 +340,9 @@ debugging and as a freshness signal in the model itself. > **HEXPIRE requires Redis 7.4 or later.** `HEXPIRE` and the field-level > TTL commands (`HTTL`, `HPERSIST`, `HEXPIREAT`, `HPEXPIRE`, `HPEXPIREAT`, > `HPTTL`, `HEXPIRETIME`, `HPEXPIRETIME`) were added in Redis 7.4. The -> demo's `Cargo.toml` pins `redis = "0.27"` and uses -> `redis::cmd("HEXPIRE")` because the typed binding doesn't ship on that -> client line yet — the wire bytes are identical. +> demo's `Cargo.toml` pins `redis = "1.8"` and issues them through +> `redis::cmd("HEXPIRE")`. redis-rs 1.x also has typed bindings for some +> of them (`hexpire`, `httl`, `hpersist`) that send the same wire bytes. ### Inference reads with HMGET @@ -357,7 +357,7 @@ pub async fn get_features( if field_names.is_empty() { return Ok(BTreeMap::new()); } let key = self.key_for(entity_id); let mut conn = self.conn.clone(); - let values: Vec> = conn.hget(&key, field_names).await?; + let values: Vec> = conn.hmget(&key, field_names).await?; let mut out = BTreeMap::new(); for (n, v) in field_names.iter().zip(values.into_iter()) { if let Some(s) = v { out.insert((*n).to_string(), s); } @@ -386,7 +386,7 @@ pub async fn batch_get_features( } let mut pipe = redis::pipe(); for id in entity_ids { - pipe.hget(self.key_for(id), field_names); + pipe.hmget(self.key_for(id), field_names); } let mut conn = self.conn.clone(); let rows: Vec>> = pipe.query_async(&mut conn).await?; @@ -400,7 +400,7 @@ under 1 ms against a local Redis. A Redis Cluster is different: a single `redis::pipe()` is bound to one connection, and a `ConnectionManager` holds one connection to one node. For batch reads on a cluster, use redis-rs's -[`cluster_async`](https://docs.rs/redis/0.27/redis/cluster_async/index.html) +[`cluster_async`](https://docs.rs/redis/latest/redis/cluster_async/index.html) client and either fan out parallel `hget` calls (the cluster client routes each one to the right shard) or, for tighter control, group entity IDs by hash slot and run one pipeline per shard in parallel. A hash tag like @@ -520,10 +520,9 @@ multiplexed socket. Endpoints: * **Redis 7.4 or later.** [`HEXPIRE`](/content/commands/hexpire.md) and [`HTTL`](/content/commands/httl.md) were added in Redis 7.4; the demo relies on per-field TTL for the mixed-staleness story. -* **Rust 1.74 or later.** The demo uses `async fn` in traits, `let-else`, - and other recent ergonomics. Earlier stable Rust may compile after small - tweaks. -* **redis-rs 0.27 or later.** The demo's `Cargo.toml` pins 0.27 with the +* **Rust 1.88 or later.** The `redis` 1.8 crate requires it, and the demo + uses `async fn` in traits, `let-else`, and other recent ergonomics. +* **redis-rs 1.8 or later.** The demo's `Cargo.toml` pins 1.8 with the `tokio-comp`, `aio`, and `connection-manager` features. Per-field TTL commands are issued via `redis::cmd("HEXPIRE")`. @@ -620,7 +619,7 @@ prevents training-serving skew. On a single Redis instance, a pipelined `HMGET` across `N` users is one round trip. A Redis Cluster is different: a single `redis::pipe()` ships through one connection to one node, so on a cluster you need redis-rs's -[`cluster_async`](https://docs.rs/redis/0.27/redis/cluster_async/index.html) +[`cluster_async`](https://docs.rs/redis/latest/redis/cluster_async/index.html) client. Either fan out parallel `hget` calls (the cluster client routes each one to the right shard) or group entity IDs by hash slot and issue one pipeline against each shard in parallel. diff --git a/content/develop/use-cases/feature-store/rust/feature_store.rs b/content/develop/use-cases/feature-store/rust/feature_store.rs index d138725880..99d57bb643 100644 --- a/content/develop/use-cases/feature-store/rust/feature_store.rs +++ b/content/develop/use-cases/feature-store/rust/feature_store.rs @@ -247,7 +247,7 @@ impl FeatureStore { for code in &codes { if *code != 1 { return Err(redis::RedisError::from(( - redis::ErrorKind::ResponseError, + redis::ErrorKind::Server(redis::ServerErrorKind::ResponseError), "HEXPIRE invariant violated", format!( "HEXPIRE did not set every field TTL for {key}: {codes:?}" @@ -278,7 +278,7 @@ impl FeatureStore { } let key = self.key_for(entity_id); let mut conn = self.conn.clone(); - let values: Vec> = conn.hget(&key, field_names).await?; + let values: Vec> = conn.hmget(&key, field_names).await?; let mut out = BTreeMap::new(); for (n, v) in field_names.iter().zip(values.into_iter()) { if let Some(s) = v { @@ -320,7 +320,7 @@ impl FeatureStore { } let mut pipe = redis::pipe(); for id in entity_ids { - pipe.hget(self.key_for(id), field_names); + pipe.hmget(self.key_for(id), field_names); } let mut conn = self.conn.clone(); let rows: Vec>> = pipe.query_async(&mut conn).await?; @@ -377,7 +377,7 @@ impl FeatureStore { // shape coercion here the way redis-py and Lettuce do for // their respective clients. let values: Value = cmd.query_async(&mut conn).await?; - let codes: Vec = Vec::::from_redis_value(&values).unwrap_or_else(|_| { + let codes: Vec = Vec::::from_redis_value_ref(&values).unwrap_or_else(|_| { // Defensive fallback: if the client ever returns nil for // a missing key, treat every field as -2. field_names.iter().map(|_| -2).collect() @@ -404,6 +404,7 @@ impl FeatureStore { let mut ids: Vec = Vec::new(); let prefix_len = self.key_prefix.len(); while let Some(k) = iter.next_item().await { + let k = k?; if k.len() > prefix_len { ids.push(k[prefix_len..].to_string()); } @@ -444,7 +445,7 @@ impl FeatureStore { let mut batch: Vec = Vec::with_capacity(500); let mut iter: AsyncIter = conn.scan_match(&pattern).await?; while let Some(k) = iter.next_item().await { - batch.push(k); + batch.push(k?); if batch.len() >= 500 { let n: i64 = self.conn.clone().del(batch.as_slice()).await?; deleted += n; diff --git a/content/develop/use-cases/job-queue/rust/Cargo.lock b/content/develop/use-cases/job-queue/rust/Cargo.lock index 3f2c6b2cf9..ffd9f97a0b 100644 --- a/content/develop/use-cases/job-queue/rust/Cargo.lock +++ b/content/develop/use-cases/job-queue/rust/Cargo.lock @@ -11,6 +11,23 @@ dependencies = [ "rustversion", ] +[[package]] +name = "arcstr" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "03918c3dbd7701a85c6b9887732e2921175f26c350b4563841d0958c21d57e6d" + +[[package]] +name = "async-lock" +version = "3.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "290f7f2596bd5b78a9fec8088ccd89180d7f9f55b94b0576823bbbdc72ee8311" +dependencies = [ + "event-listener", + "event-listener-strategy", + "pin-project-lite", +] + [[package]] name = "async-trait" version = "0.1.89" @@ -118,9 +135,9 @@ checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" [[package]] name = "combine" -version = "4.6.7" +version = "4.6.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ba5a308b75df32fe02788e748662718f03fde005016435c444eea572398219fd" +checksum = "4356eba6976435498d05a311ee9fc486da3bc81c200d3f0d7dd9703b41913f92" dependencies = [ "bytes", "futures-core", @@ -141,12 +158,6 @@ dependencies = [ "syn", ] -[[package]] -name = "either" -version = "1.15.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "48c757948c5ede0e46177b7add2e67155f70e33c07fea8284df6576da70b3719" - [[package]] name = "errno" version = "0.3.14" @@ -154,7 +165,27 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys", +] + +[[package]] +name = "event-listener" +version = "5.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a23add41df1562121a9393cb065eab5146a1242410f23a644851e90cfd669d2" +dependencies = [ + "parking", + "pin-project-lite", +] + +[[package]] +name = "event-listener-strategy" +version = "0.5.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8be9f3dfaaffdae2972880079a491a1a8bb7cbed0b8dd7a347f668b4150a3b93" +dependencies = [ + "event-listener", + "pin-project-lite", ] [[package]] @@ -172,21 +203,6 @@ dependencies = [ "percent-encoding", ] -[[package]] -name = "futures" -version = "0.3.32" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8b147ee9d1f6d097cef9ce628cd2ee62288d963e16fb287bd9286455b241382d" -dependencies = [ - "futures-channel", - "futures-core", - "futures-executor", - "futures-io", - "futures-sink", - "futures-task", - "futures-util", -] - [[package]] name = "futures-channel" version = "0.3.32" @@ -194,7 +210,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "07bbe89c50d7a535e539b8c17bc0b49bdb77747034daa8087407d655f3f7cc1d" dependencies = [ "futures-core", - "futures-sink", ] [[package]] @@ -203,34 +218,6 @@ version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d" -[[package]] -name = "futures-executor" -version = "0.3.32" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "baf29c38818342a3b26b5b923639e7b1f4a61fc5e76102d4b1981c6dc7a7579d" -dependencies = [ - "futures-core", - "futures-task", - "futures-util", -] - -[[package]] -name = "futures-io" -version = "0.3.32" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cecba35d7ad927e23624b22ad55235f2239cfa44fd10428eecbeba6d6a717718" - -[[package]] -name = "futures-macro" -version = "0.3.32" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b" -dependencies = [ - "proc-macro2", - "quote", - "syn", -] - [[package]] name = "futures-sink" version = "0.3.32" @@ -249,13 +236,9 @@ version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6" dependencies = [ - "futures-channel", "futures-core", - "futures-io", - "futures-macro", "futures-sink", "futures-task", - "memchr", "pin-project-lite", "slab", ] @@ -454,15 +437,6 @@ dependencies = [ "icu_properties", ] -[[package]] -name = "itertools" -version = "0.13.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "413ee7dfc52ee1a4949ceeb7dbc8a33f2d6c088194d9f922fb8318faf1f01186" -dependencies = [ - "either", -] - [[package]] name = "itoa" version = "1.0.18" @@ -534,14 +508,14 @@ checksum = "50b7e5b27aa02a74bac8c3f23f448f8d87ff11f92d3aac1a6ed369ee08cc56c1" dependencies = [ "libc", "wasi", - "windows-sys 0.61.2", + "windows-sys", ] [[package]] name = "num-bigint" -version = "0.4.6" +version = "0.5.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a5e44f723f1133c9deac646763579fdb3ac745e418f2a7af9cd0c431da1f20b9" +checksum = "93e7820bc0a80a0238e650327316f929ba18d5be054b647490a3a6a339f3e7c0" dependencies = [ "num-integer", "num-traits", @@ -571,6 +545,12 @@ version = "1.21.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" +[[package]] +name = "parking" +version = "2.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f38d5652c16fde515bb1ecef450ab0f6a219d619a7274976324d5e377f7dceba" + [[package]] name = "parking_lot" version = "0.12.5" @@ -674,28 +654,30 @@ dependencies = [ [[package]] name = "redis" -version = "0.27.6" +version = "1.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "09d8f99a4090c89cc489a94833c901ead69bfbf3877b4867d5482e321ee875bc" +checksum = "7acd809e80afa754c63f586955699d8990efe1c4b6613183ee5f027a11fbc8b5" dependencies = [ "arc-swap", - "async-trait", + "arcstr", + "async-lock", "backon", "bytes", + "cfg-if", "combine", - "futures", + "futures-channel", "futures-util", - "itertools", "itoa", "num-bigint", "percent-encoding", "pin-project-lite", "ryu", "sha1_smol", - "socket2 0.5.10", + "socket2", "tokio", "tokio-util", "url", + "xxhash-rust", ] [[package]] @@ -819,16 +801,6 @@ version = "1.15.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03" -[[package]] -name = "socket2" -version = "0.5.10" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e22376abed350d73dd1cd119b57ffccad95b4e585a7cda43e286245ce23c0678" -dependencies = [ - "libc", - "windows-sys 0.52.0", -] - [[package]] name = "socket2" version = "0.6.3" @@ -836,7 +808,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3a766e1110788c36f4fa1c2b71b387a7815aa65f88ce0229841826633d93723e" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys", ] [[package]] @@ -895,9 +867,9 @@ dependencies = [ "parking_lot", "pin-project-lite", "signal-hook-registry", - "socket2 0.6.3", + "socket2", "tokio-macros", - "windows-sys 0.61.2", + "windows-sys", ] [[package]] @@ -1008,15 +980,6 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" -[[package]] -name = "windows-sys" -version = "0.52.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d" -dependencies = [ - "windows-targets", -] - [[package]] name = "windows-sys" version = "0.61.2" @@ -1026,76 +989,18 @@ dependencies = [ "windows-link", ] -[[package]] -name = "windows-targets" -version = "0.52.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973" -dependencies = [ - "windows_aarch64_gnullvm", - "windows_aarch64_msvc", - "windows_i686_gnu", - "windows_i686_gnullvm", - "windows_i686_msvc", - "windows_x86_64_gnu", - "windows_x86_64_gnullvm", - "windows_x86_64_msvc", -] - -[[package]] -name = "windows_aarch64_gnullvm" -version = "0.52.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3" - -[[package]] -name = "windows_aarch64_msvc" -version = "0.52.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469" - -[[package]] -name = "windows_i686_gnu" -version = "0.52.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b" - -[[package]] -name = "windows_i686_gnullvm" -version = "0.52.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66" - -[[package]] -name = "windows_i686_msvc" -version = "0.52.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66" - -[[package]] -name = "windows_x86_64_gnu" -version = "0.52.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78" - -[[package]] -name = "windows_x86_64_gnullvm" -version = "0.52.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d" - -[[package]] -name = "windows_x86_64_msvc" -version = "0.52.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" - [[package]] name = "writeable" version = "0.6.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1ffae5123b2d3fc086436f8834ae3ab053a283cfac8fe0a0b8eaae044768a4c4" +[[package]] +name = "xxhash-rust" +version = "0.8.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "550a2b930b62486a393c52d5c3b84bff264b28aa437ed64694d31e93b1757af7" + [[package]] name = "yoke" version = "0.8.2" diff --git a/content/develop/use-cases/job-queue/rust/Cargo.toml b/content/develop/use-cases/job-queue/rust/Cargo.toml index 2d08e6cce9..2631944005 100644 --- a/content/develop/use-cases/job-queue/rust/Cargo.toml +++ b/content/develop/use-cases/job-queue/rust/Cargo.toml @@ -4,7 +4,7 @@ version = "0.1.0" edition = "2021" [dependencies] -redis = { version = "0.27", features = ["tokio-comp", "aio", "connection-manager"] } +redis = { version = "1.8", features = ["tokio-comp", "aio", "connection-manager"] } tokio = { version = "1", features = ["full"] } axum = "0.7" serde = { version = "1.0", features = ["derive"] } diff --git a/content/develop/use-cases/job-queue/rust/_index.md b/content/develop/use-cases/job-queue/rust/_index.md index 82b234b506..b6a9d33c5e 100644 --- a/content/develop/use-cases/job-queue/rust/_index.md +++ b/content/develop/use-cases/job-queue/rust/_index.md @@ -52,7 +52,7 @@ The `RedisJobQueue` struct wraps the queue operations ([source](https://github.com/redis/docs/blob/main/content/develop/use-cases/job-queue/rust/src/job_queue.rs)): ```rust -use redis::aio::ConnectionManager; +use redis::aio::{ConnectionManager, ConnectionManagerConfig}; use redis::Client; use serde_json::json; @@ -61,7 +61,10 @@ use jobqueue_demo::job_queue::{JobQueueOptions, RedisJobQueue}; #[tokio::main] async fn main() -> redis::RedisResult<()> { let client = Client::open("redis://127.0.0.1:6379/")?; - let conn = ConnectionManager::new(client).await?; + // Workers block in BLMOVE for longer than the 500 ms default response timeout + // in redis-rs 1.x, so turn the response timeout off for this connection. + let config = ConnectionManagerConfig::new().set_response_timeout(None); + let conn = ConnectionManager::new_with_config(client, config).await?; let queue = RedisJobQueue::new( conn, JobQueueOptions { @@ -161,7 +164,7 @@ pub async fn enqueue(&self, payload: Value) -> RedisResult { .ignore() .lpush(&self.pending_key, &job_id) .ignore() - .query_async::<_, ()>(&mut conn) + .query_async::<()>(&mut conn) .await?; self.enqueued_total.fetch_add(1, Ordering::Relaxed); @@ -353,14 +356,14 @@ The completed and failed lists are capped via `LTRIM` so they never grow unbound ## Prerequisites * Redis 6.2 or later running locally on the default port (6379). `BLMOVE` requires Redis 6.2+; on older servers, replace the call with `BRPOPLPUSH`. -* Rust 1.75 or later. -* The [`redis`](https://crates.io/crates/redis) crate at 0.27+ (or 0.24+ if you only need `BRPOPLPUSH`-style claims). +* Rust 1.88 or later (the `redis` 1.8 crate requires it). +* The [`redis`](https://crates.io/crates/redis) crate at 1.8 or later. Add the crate dependencies to your `Cargo.toml`: ```toml [dependencies] -redis = { version = "0.27", features = ["tokio-comp", "aio", "connection-manager"] } +redis = { version = "1.8", features = ["tokio-comp", "aio", "connection-manager"] } tokio = { version = "1", features = ["full"] } axum = "0.7" serde = { version = "1.0", features = ["derive"] } diff --git a/content/develop/use-cases/job-queue/rust/src/main.rs b/content/develop/use-cases/job-queue/rust/src/main.rs index ed650322ef..14f5e377a6 100644 --- a/content/develop/use-cases/job-queue/rust/src/main.rs +++ b/content/develop/use-cases/job-queue/rust/src/main.rs @@ -17,7 +17,7 @@ use axum::{ routing::{get, post}, Form, Json, Router, }; -use redis::aio::ConnectionManager; +use redis::aio::{ConnectionManager, ConnectionManagerConfig}; use redis::Client; use serde::Deserialize; use serde_json::{json, Value}; @@ -51,7 +51,10 @@ async fn main() { } }; - let conn = match ConnectionManager::new(client).await { + // Workers block in BLMOVE for longer than the 500 ms default response timeout + // in redis-rs 1.x, so turn the response timeout off for this connection. + let config = ConnectionManagerConfig::new().set_response_timeout(None); + let conn = match ConnectionManager::new_with_config(client, config).await { Ok(c) => c, Err(e) => { eprintln!("Failed to connect to Redis at {}: {}", redis_url, e); diff --git a/content/develop/use-cases/leaderboard/rust/Cargo.toml b/content/develop/use-cases/leaderboard/rust/Cargo.toml index 9b236afb20..8bc9d8b365 100644 --- a/content/develop/use-cases/leaderboard/rust/Cargo.toml +++ b/content/develop/use-cases/leaderboard/rust/Cargo.toml @@ -4,7 +4,7 @@ version = "0.1.0" edition = "2021" [dependencies] -redis = { version = "0.24", features = ["script", "tokio-comp"] } +redis = { version = "1.8", features = ["script", "tokio-comp"] } tokio = { version = "1", features = ["full"] } axum = "0.7" serde = { version = "1.0", features = ["derive"] } diff --git a/content/develop/use-cases/leaderboard/rust/_index.md b/content/develop/use-cases/leaderboard/rust/_index.md index d3367124e7..0c036b383c 100644 --- a/content/develop/use-cases/leaderboard/rust/_index.md +++ b/content/develop/use-cases/leaderboard/rust/_index.md @@ -48,7 +48,7 @@ Add the `redis` crate to your `Cargo.toml`: ```toml [dependencies] -redis = { version = "0.24", features = ["script", "tokio-comp"] } +redis = { version = "1.8", features = ["script", "tokio-comp"] } tokio = { version = "1", features = ["full"] } ``` diff --git a/content/develop/use-cases/leaderboard/rust/leaderboard.rs b/content/develop/use-cases/leaderboard/rust/leaderboard.rs index 7d99a1b033..414452e8b0 100644 --- a/content/develop/use-cases/leaderboard/rust/leaderboard.rs +++ b/content/develop/use-cases/leaderboard/rust/leaderboard.rs @@ -317,7 +317,7 @@ impl AsyncRedisLeaderboard { .arg(self.metadata_key(user_id)) .arg(flatten_metadata(&payload)); } - pipe.query_async::<_, ()>(con).await?; + pipe.query_async::<()>(con).await?; let trimmed_user_ids = self.trim_to_max_entries(con).await?; Ok(self @@ -553,8 +553,8 @@ fn flatten_metadata(metadata: &Metadata) -> Vec { fn parse_score_value(value: Option<&redis::Value>) -> Option { match value { Some(redis::Value::Int(v)) => Some(*v as f64), - Some(redis::Value::Data(v)) => String::from_utf8_lossy(v).parse::().ok(), - Some(redis::Value::Bulk(items)) => items.first().and_then(|value| parse_score_value(Some(value))), + Some(redis::Value::BulkString(v)) => String::from_utf8_lossy(v).parse::().ok(), + Some(redis::Value::Array(items)) => items.first().and_then(|value| parse_score_value(Some(value))), _ => None, } } diff --git a/content/develop/use-cases/memory-layer/rust/Cargo.toml b/content/develop/use-cases/memory-layer/rust/Cargo.toml index f9442800b6..5cd86c2c6a 100644 --- a/content/develop/use-cases/memory-layer/rust/Cargo.toml +++ b/content/develop/use-cases/memory-layer/rust/Cargo.toml @@ -22,7 +22,7 @@ lto = "thin" codegen-units = 1 [dependencies] -redis = { version = "0.27", default-features = false } +redis = { version = "1.8", default-features = false } tiny_http = "0.12" serde = { version = "1", features = ["derive"] } serde_json = "1" diff --git a/content/develop/use-cases/memory-layer/rust/src/event_log.rs b/content/develop/use-cases/memory-layer/rust/src/event_log.rs index 73c89bd5ce..61bc9de4b1 100644 --- a/content/develop/use-cases/memory-layer/rust/src/event_log.rs +++ b/content/develop/use-cases/memory-layer/rust/src/event_log.rs @@ -205,7 +205,7 @@ fn redis_value_to_string(v: &Value) -> Option { Value::Int(n) => Some(n.to_string()), Value::Double(d) => Some(d.to_string()), Value::Nil => None, - _ => String::from_redis_value(v).ok(), + _ => String::from_redis_value_ref(v).ok(), } } diff --git a/content/develop/use-cases/memory-layer/rust/src/long_term_memory.rs b/content/develop/use-cases/memory-layer/rust/src/long_term_memory.rs index 3681ab8698..535fd9ba7d 100644 --- a/content/develop/use-cases/memory-layer/rust/src/long_term_memory.rs +++ b/content/develop/use-cases/memory-layer/rust/src/long_term_memory.rs @@ -596,7 +596,7 @@ fn redis_value_to_string(v: &Value) -> Option { Value::Double(d) => Some(d.to_string()), Value::Boolean(b) => Some(b.to_string()), Value::Nil => None, - _ => String::from_redis_value(v).ok(), + _ => String::from_redis_value_ref(v).ok(), } } diff --git a/content/develop/use-cases/memory-layer/rust/src/session_store.rs b/content/develop/use-cases/memory-layer/rust/src/session_store.rs index 80b245a9a6..95962df73d 100644 --- a/content/develop/use-cases/memory-layer/rust/src/session_store.rs +++ b/content/develop/use-cases/memory-layer/rust/src/session_store.rs @@ -345,6 +345,6 @@ fn redis_value_to_string(v: &Value) -> Option { Value::Double(d) => Some(d.to_string()), Value::Boolean(b) => Some(b.to_string()), Value::Nil => None, - _ => String::from_redis_value(v).ok(), + _ => String::from_redis_value_ref(v).ok(), } } diff --git a/content/develop/use-cases/prefetch-cache/rust/Cargo.toml b/content/develop/use-cases/prefetch-cache/rust/Cargo.toml index 0725eb6f4e..1f0b0335a6 100644 --- a/content/develop/use-cases/prefetch-cache/rust/Cargo.toml +++ b/content/develop/use-cases/prefetch-cache/rust/Cargo.toml @@ -4,7 +4,7 @@ version = "0.1.0" edition = "2021" [dependencies] -redis = { version = "0.24", features = ["tokio-comp", "aio", "connection-manager"] } +redis = { version = "1.8", features = ["tokio-comp", "aio", "connection-manager"] } tokio = { version = "1", features = ["full"] } axum = "0.7" serde = { version = "1.0", features = ["derive"] } diff --git a/content/develop/use-cases/prefetch-cache/rust/_index.md b/content/develop/use-cases/prefetch-cache/rust/_index.md index ffccaf5a96..c3704886d1 100644 --- a/content/develop/use-cases/prefetch-cache/rust/_index.md +++ b/content/develop/use-cases/prefetch-cache/rust/_index.md @@ -116,7 +116,7 @@ pub async fn bulk_load( } if loaded > 0 { let mut conn = self.conn.clone(); - pipe.query_async::<_, ()>(&mut conn).await?; + pipe.query_async::<()>(&mut conn).await?; } self.stats.prefetched.fetch_add(loaded, Ordering::Relaxed); Ok(loaded) @@ -172,7 +172,7 @@ pub async fn apply_change(&self, change: &ChangeEvent) -> RedisResult<()> { .del(&cache_key).ignore() .hset_multiple(&cache_key, &pairs).ignore() .expire(&cache_key, self.cfg.ttl_seconds).ignore() - .query_async::<_, ()>(&mut conn).await?; + .query_async::<()>(&mut conn).await?; } "delete" => { let _: i64 = conn.del(&cache_key).await?; @@ -270,7 +270,7 @@ Before running the demo, make sure that: * Redis is running and accessible. By default, the demo connects to `localhost:6379`. * You have a working [Rust toolchain](https://www.rust-lang.org/tools/install) (`cargo` 1.70+). -* The demo's `Cargo.toml` pins the `redis` crate at 0.24+ with the `tokio-comp`, `aio`, and `connection-manager` features; `cargo build` will pull the rest of the dependency graph. +* The demo's `Cargo.toml` pins the `redis` crate at 1.8 with the `tokio-comp`, `aio`, and `connection-manager` features; `cargo build` will pull the rest of the dependency graph. If your Redis server is running elsewhere, start the demo with `--redis-host` and `--redis-port`. @@ -384,7 +384,7 @@ If multiple applications share a Redis deployment, prefix cache keys with the ap ### Use `ConnectionManager`, not a bare connection -`redis::aio::ConnectionManager` is the recommended async connection type for redis-rs. It is cheap to clone (the demo passes a clone to every async task), reconnects automatically on transient failures, and pipelines commands behind the scenes. A bare `redis::aio::Connection` is not `Clone` and cannot be safely shared across tokio tasks. If you need a true connection pool — for example, to fan out heavy `MULTI`/`EXEC` traffic across multiple sockets — use [`deadpool-redis`](https://crates.io/crates/deadpool-redis) or [`bb8-redis`](https://crates.io/crates/bb8-redis). +`redis::aio::ConnectionManager` is the recommended async connection type for redis-rs. It is cheap to clone (the demo passes a clone to every async task), reconnects automatically on transient failures, and pipelines commands behind the scenes. A bare `redis::aio::MultiplexedConnection` is also cheap to clone, but it does not reconnect: once the underlying socket drops, every clone fails until you open a new connection. If you need a true connection pool — for example, to fan out heavy `MULTI`/`EXEC` traffic across multiple sockets — use [`deadpool-redis`](https://crates.io/crates/deadpool-redis) or [`bb8-redis`](https://crates.io/crates/bb8-redis). ### Inspect cached entries directly in Redis diff --git a/content/develop/use-cases/prefetch-cache/rust/cache.rs b/content/develop/use-cases/prefetch-cache/rust/cache.rs index f6f5a4c72e..d93725d56d 100644 --- a/content/develop/use-cases/prefetch-cache/rust/cache.rs +++ b/content/develop/use-cases/prefetch-cache/rust/cache.rs @@ -119,7 +119,7 @@ impl PrefetchCache { } if loaded > 0 { let mut conn = self.conn.clone(); - pipe.query_async::<_, ()>(&mut conn).await?; + pipe.query_async::<()>(&mut conn).await?; } self.stats.prefetched.fetch_add(loaded, Ordering::Relaxed); Ok(loaded) @@ -191,7 +191,7 @@ impl PrefetchCache { .ignore() .expire(&cache_key, self.cfg.ttl_seconds) .ignore() - .query_async::<_, ()>(&mut conn) + .query_async::<()>(&mut conn) .await?; } "delete" => { @@ -230,7 +230,7 @@ impl PrefetchCache { { let mut iter: AsyncIter = conn.scan_match(&pattern).await?; while let Some(key) = iter.next_item().await { - keys.push(key); + keys.push(key?); } } if keys.is_empty() { @@ -256,7 +256,7 @@ impl PrefetchCache { let mut out: Vec = Vec::new(); let mut iter: AsyncIter = conn.scan_match(&pattern).await?; while let Some(key) = iter.next_item().await { - out.push(self.strip_prefix(&key).to_string()); + out.push(self.strip_prefix(&key?).to_string()); } out.sort(); Ok(out) diff --git a/content/develop/use-cases/pub-sub/rust/Cargo.toml b/content/develop/use-cases/pub-sub/rust/Cargo.toml index b464db35e2..e4e8b9972a 100644 --- a/content/develop/use-cases/pub-sub/rust/Cargo.toml +++ b/content/develop/use-cases/pub-sub/rust/Cargo.toml @@ -4,7 +4,7 @@ version = "0.1.0" edition = "2021" [dependencies] -redis = { version = "0.27", features = ["tokio-comp", "aio", "connection-manager"] } +redis = { version = "1.8", features = ["tokio-comp", "aio", "connection-manager"] } tokio = { version = "1", features = ["full"] } axum = "0.7" serde = { version = "1.0", features = ["derive"] } diff --git a/content/develop/use-cases/pub-sub/rust/_index.md b/content/develop/use-cases/pub-sub/rust/_index.md index 6cb17b869d..c6ad65e6df 100644 --- a/content/develop/use-cases/pub-sub/rust/_index.md +++ b/content/develop/use-cases/pub-sub/rust/_index.md @@ -257,7 +257,7 @@ pub async fn stats(&self) -> Stats { * Redis 6.2 or later running locally on the default port (6379). Earlier versions still work for plain `PUBLISH`/`SUBSCRIBE`; `PUBSUB NUMPAT` is older than that. * A Rust toolchain (rustup + cargo) capable of building the 2021 edition. -* The `redis` crate at version 0.27+ with the `tokio-comp`, `aio`, and `connection-manager` features, plus `tokio`, `axum`, `serde`, `serde_json`, and `futures-util` (declared in `Cargo.toml`). +* The `redis` crate at version 1.8 or later with the `tokio-comp`, `aio`, and `connection-manager` features, plus `tokio`, `axum`, `serde`, `serde_json`, and `futures-util` (declared in `Cargo.toml`). ## Running the demo diff --git a/content/develop/use-cases/rate-limiter/rust/Cargo.toml b/content/develop/use-cases/rate-limiter/rust/Cargo.toml index 8c27cbac7a..41b07eed0c 100644 --- a/content/develop/use-cases/rate-limiter/rust/Cargo.toml +++ b/content/develop/use-cases/rate-limiter/rust/Cargo.toml @@ -4,7 +4,7 @@ version = "0.1.0" edition = "2021" [dependencies] -redis = { version = "0.24", features = ["script"] } +redis = { version = "1.8", features = ["script"] } tokio = { version = "1", features = ["full"] } axum = "0.7" serde = { version = "1.0", features = ["derive"] } diff --git a/content/develop/use-cases/rate-limiter/rust/_index.md b/content/develop/use-cases/rate-limiter/rust/_index.md index 856b481f21..c5b331e089 100644 --- a/content/develop/use-cases/rate-limiter/rust/_index.md +++ b/content/develop/use-cases/rate-limiter/rust/_index.md @@ -121,14 +121,14 @@ Add the `redis` crate to your `Cargo.toml`: ```toml [dependencies] -redis = "0.24" +redis = "1.8" ``` For async support, enable the `tokio-comp` or `async-std-comp` feature: ```toml [dependencies] -redis = { version = "0.24", features = ["tokio-comp"] } +redis = { version = "1.8", features = ["tokio-comp"] } tokio = { version = "1", features = ["full"] } ``` diff --git a/content/develop/use-cases/rate-limiter/rust/token_bucket.rs b/content/develop/use-cases/rate-limiter/rust/token_bucket.rs index 54eca45ec5..3bc17614c3 100644 --- a/content/develop/use-cases/rate-limiter/rust/token_bucket.rs +++ b/content/develop/use-cases/rate-limiter/rust/token_bucket.rs @@ -171,7 +171,7 @@ impl TokenBucket { let remaining = match result.get(1) { Some(redis::Value::Int(v)) => *v as f64, - Some(redis::Value::Data(v)) => { + Some(redis::Value::BulkString(v)) => { String::from_utf8_lossy(v).parse::().unwrap_or(0.0) } _ => 0.0, diff --git a/content/develop/use-cases/recommendation-engine/rust/Cargo.toml b/content/develop/use-cases/recommendation-engine/rust/Cargo.toml index d674a3813a..3654ee2366 100644 --- a/content/develop/use-cases/recommendation-engine/rust/Cargo.toml +++ b/content/develop/use-cases/recommendation-engine/rust/Cargo.toml @@ -4,7 +4,7 @@ version = "0.1.0" edition = "2021" [dependencies] -redis = { version = "0.27", features = ["tokio-comp", "aio", "connection-manager"] } +redis = { version = "1.8", features = ["tokio-comp", "aio", "connection-manager"] } tokio = { version = "1", features = ["full"] } axum = "0.7" serde = { version = "1.0", features = ["derive"] } diff --git a/content/develop/use-cases/recommendation-engine/rust/recommender.rs b/content/develop/use-cases/recommendation-engine/rust/recommender.rs index 1c56165f53..806b734b87 100644 --- a/content/develop/use-cases/recommendation-engine/rust/recommender.rs +++ b/content/develop/use-cases/recommendation-engine/rust/recommender.rs @@ -250,7 +250,7 @@ impl RedisRecommender { for p in products { let key = self.product_key(&p.id); let vec_bytes = decode_embedding(p).map_err(|e| { - RedisError::from((redis::ErrorKind::TypeError, "encode product", e)) + RedisError::from((redis::ErrorKind::UnexpectedReturnType, "encode product", e)) })?; // Build the field list explicitly so we can mix string and // byte values in the same HSET. Strings go through as @@ -412,20 +412,20 @@ impl RedisRecommender { // `HMGET` returns `[Option, Option]` here because the // embedding is binary. let fields: Vec>> = conn - .hget(&product_key, &["embedding", "category"]) + .hmget(&product_key, &["embedding", "category"]) .await?; let clicked_bytes = match fields.get(0).and_then(|v| v.as_ref()) { Some(b) => b.clone(), None => { return Err(RedisError::from(( - redis::ErrorKind::TypeError, + redis::ErrorKind::UnexpectedReturnType, "unknown product", format!("no hash at {}", product_key), ))) } }; let clicked_vec = bytes_to_floats(&clicked_bytes).map_err(|e| { - RedisError::from((redis::ErrorKind::TypeError, "decode embedding", e)) + RedisError::from((redis::ErrorKind::UnexpectedReturnType, "decode embedding", e)) })?; let category = fields .get(1) @@ -439,7 +439,7 @@ impl RedisRecommender { let new_session = match previous { Some(prev_bytes) if !prev_bytes.is_empty() => { let prev_vec = bytes_to_floats(&prev_bytes).map_err(|e| { - RedisError::from((redis::ErrorKind::TypeError, "decode session_vec", e)) + RedisError::from((redis::ErrorKind::UnexpectedReturnType, "decode session_vec", e)) })?; ewma_blend(&prev_vec, &clicked_vec, ewma_alpha) } @@ -494,7 +494,7 @@ impl RedisRecommender { "session_vec" => { let vec = bytes_to_floats(&v).map_err(|e| { RedisError::from(( - redis::ErrorKind::TypeError, + redis::ErrorKind::UnexpectedReturnType, "decode session_vec", e, )) @@ -541,7 +541,7 @@ impl RedisRecommender { /// The HNSW index reflects the change as soon as the `HSET` /// commits, so subsequent `FT.SEARCH` calls see the new vector /// without any index rebuild or serving downtime. Returns - /// `ErrorKind::TypeError` (with detail "unknown product") if the + /// `ErrorKind::UnexpectedReturnType` (with detail "unknown product") if the /// product hash does not exist (HSET would otherwise happily /// create a partial document that the index then picks up). Also /// rejects vectors with the wrong dimensionality so a model swap @@ -553,7 +553,7 @@ impl RedisRecommender { ) -> RedisResult<()> { if new_vector.len() != self.vector_dim { return Err(RedisError::from(( - redis::ErrorKind::TypeError, + redis::ErrorKind::UnexpectedReturnType, "bad vector dim", format!( "new_vector has length {}; index expects {}", @@ -567,7 +567,7 @@ impl RedisRecommender { let exists: i64 = conn.exists(&key).await?; if exists == 0 { return Err(RedisError::from(( - redis::ErrorKind::TypeError, + redis::ErrorKind::UnexpectedReturnType, "unknown product", format!("no hash at {}", key), ))); @@ -815,20 +815,20 @@ fn parse_search_reply(raw: Value, key_prefix: &str) -> RedisResult items, _ => { return Err(RedisError::from(( - redis::ErrorKind::TypeError, + redis::ErrorKind::UnexpectedReturnType, "FT.SEARCH: expected array reply", ))); } }; let mut iter = items.into_iter(); let _total: i64 = match iter.next() { - Some(v) => redis::FromRedisValue::from_redis_value(&v).unwrap_or(0), + Some(v) => redis::FromRedisValue::from_redis_value_ref(&v).unwrap_or(0), None => 0, }; let mut out: Vec = Vec::new(); while let Some(key_v) = iter.next() { let key: String = - redis::FromRedisValue::from_redis_value(&key_v).unwrap_or_default(); + redis::FromRedisValue::from_redis_value_ref(&key_v).unwrap_or_default(); let Some(fields_v) = iter.next() else { break }; let fields = match fields_v { Value::Array(items) => items, @@ -838,9 +838,9 @@ fn parse_search_reply(raw: Value, key_prefix: &str) -> RedisResult Option> { let mut out: HashMap = HashMap::new(); let mut iter = items.iter(); while let (Some(k), Some(value)) = (iter.next(), iter.next()) { - let kk: String = match redis::FromRedisValue::from_redis_value(k) { + let kk: String = match redis::FromRedisValue::from_redis_value_ref(k) { Ok(s) => s, Err(_) => continue, }; @@ -894,18 +894,16 @@ fn flat_array_to_map(v: &Value) -> Option> { } fn value_as_i64(v: &Value) -> Option { - redis::FromRedisValue::from_redis_value(v).ok() + redis::FromRedisValue::from_redis_value_ref(v).ok() } fn value_as_f64(v: &Value) -> Option { // FT.INFO returns size-in-mb as a string in many versions; tolerate // both forms. - let as_f: RedisResult = redis::FromRedisValue::from_redis_value(v); - if let Ok(n) = as_f { + if let Ok(n) = ::from_redis_value_ref(v) { return Some(n); } - let as_s: RedisResult = redis::FromRedisValue::from_redis_value(v); - if let Ok(s) = as_s { + if let Ok(s) = ::from_redis_value_ref(v) { if let Ok(n) = s.parse::() { return Some(n); } diff --git a/content/develop/use-cases/semantic-cache/rust/Cargo.lock b/content/develop/use-cases/semantic-cache/rust/Cargo.lock index 194c45868f..64ceca2801 100644 --- a/content/develop/use-cases/semantic-cache/rust/Cargo.lock +++ b/content/develop/use-cases/semantic-cache/rust/Cargo.lock @@ -33,13 +33,10 @@ dependencies = [ ] [[package]] -name = "arc-swap" -version = "1.9.1" +name = "arcstr" +version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6a3a1fd6f75306b68087b831f025c712524bcb19aad54e557b1129cfa0a2b207" -dependencies = [ - "rustversion", -] +checksum = "03918c3dbd7701a85c6b9887732e2921175f26c350b4563841d0958c21d57e6d" [[package]] name = "ascii" @@ -210,9 +207,9 @@ checksum = "6e4de3bc4ea267985becf712dc6d9eed8b04c953b3fcfb339ebc87acd9804901" [[package]] name = "combine" -version = "4.6.7" +version = "4.6.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ba5a308b75df32fe02788e748662718f03fde005016435c444eea572398219fd" +checksum = "4356eba6976435498d05a311ee9fc486da3bc81c200d3f0d7dd9703b41913f92" dependencies = [ "bytes", "memchr", @@ -1057,15 +1054,6 @@ dependencies = [ "either", ] -[[package]] -name = "itertools" -version = "0.13.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "413ee7dfc52ee1a4949ceeb7dbc8a33f2d6c088194d9f922fb8318faf1f01186" -dependencies = [ - "either", -] - [[package]] name = "itoa" version = "1.0.18" @@ -1681,18 +1669,18 @@ checksum = "03251193000f4bd3b042892be858ee50e8b3719f2b08e5833ac4353724632430" [[package]] name = "redis" -version = "0.27.6" +version = "1.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "09d8f99a4090c89cc489a94833c901ead69bfbf3877b4867d5482e321ee875bc" +checksum = "7acd809e80afa754c63f586955699d8990efe1c4b6613183ee5f027a11fbc8b5" dependencies = [ - "arc-swap", + "arcstr", "combine", - "itertools 0.13.0", "itoa", - "num-bigint", "percent-encoding", "ryu", + "socket2", "url", + "xxhash-rust", ] [[package]] @@ -1966,6 +1954,16 @@ version = "1.15.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03" +[[package]] +name = "socket2" +version = "0.6.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4" +dependencies = [ + "libc", + "windows-sys 0.61.2", +] + [[package]] name = "spm_precompiled" version = "0.1.4" @@ -2730,6 +2728,12 @@ version = "0.6.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1ffae5123b2d3fc086436f8834ae3ab053a283cfac8fe0a0b8eaae044768a4c4" +[[package]] +name = "xxhash-rust" +version = "0.8.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "550a2b930b62486a393c52d5c3b84bff264b28aa437ed64694d31e93b1757af7" + [[package]] name = "yoke" version = "0.7.5" diff --git a/content/develop/use-cases/semantic-cache/rust/Cargo.toml b/content/develop/use-cases/semantic-cache/rust/Cargo.toml index d084be3cd6..86d591c696 100644 --- a/content/develop/use-cases/semantic-cache/rust/Cargo.toml +++ b/content/develop/use-cases/semantic-cache/rust/Cargo.toml @@ -18,7 +18,7 @@ lto = "thin" codegen-units = 1 [dependencies] -redis = { version = "0.27", default-features = false } +redis = { version = "1.8", default-features = false } tiny_http = "0.12" serde = { version = "1", features = ["derive"] } serde_json = "1" diff --git a/content/develop/use-cases/semantic-cache/rust/_index.md b/content/develop/use-cases/semantic-cache/rust/_index.md index 313ab36a8c..0f77943167 100644 --- a/content/develop/use-cases/semantic-cache/rust/_index.md +++ b/content/develop/use-cases/semantic-cache/rust/_index.md @@ -237,7 +237,7 @@ The server holds one `LocalEmbedder`, one `RedisSemanticCache`, and one `MockLlm cd docs/content/develop/use-cases/semantic-cache/rust ``` -2. Make sure a recent Rust toolchain is installed (Rust 1.75 or newer). +2. Make sure a recent Rust toolchain is installed (Rust 1.88 or newer). [`rustup`](https://rustup.rs/) is the usual route. The build pulls in Candle and the Hugging Face tokenizer crates, so the first `cargo build` takes a few minutes. diff --git a/content/develop/use-cases/semantic-cache/rust/src/cache.rs b/content/develop/use-cases/semantic-cache/rust/src/cache.rs index 79af117090..714be823c7 100644 --- a/content/develop/use-cases/semantic-cache/rust/src/cache.rs +++ b/content/develop/use-cases/semantic-cache/rust/src/cache.rs @@ -667,7 +667,7 @@ fn redis_value_to_string(v: &Value) -> Option { // variants we didn't list explicitly (Map, Set, BigNumber, // etc — these shouldn't appear in FT.SEARCH replies but // future protocol additions could). - String::from_redis_value(v).ok() + String::from_redis_value_ref(v).ok() } } } diff --git a/content/develop/use-cases/session-store/rust/Cargo.toml b/content/develop/use-cases/session-store/rust/Cargo.toml index 5287487470..4be9adec62 100644 --- a/content/develop/use-cases/session-store/rust/Cargo.toml +++ b/content/develop/use-cases/session-store/rust/Cargo.toml @@ -4,7 +4,7 @@ version = "0.1.0" edition = "2021" [dependencies] -redis = { version = "0.24", features = ["tokio-comp"] } +redis = { version = "1.8", features = ["tokio-comp"] } tokio = { version = "1", features = ["full"] } axum = "0.7" serde = { version = "1.0", features = ["derive"] } diff --git a/content/develop/use-cases/session-store/rust/_index.md b/content/develop/use-cases/session-store/rust/_index.md index 191b95e36f..683bb43dd0 100644 --- a/content/develop/use-cases/session-store/rust/_index.md +++ b/content/develop/use-cases/session-store/rust/_index.md @@ -176,7 +176,7 @@ pub async fn get_session_async( refresh_ttl: bool, ) -> RedisResult>> where - C: redis::aio::ConnectionLike + Send, + C: redis::aio::ConnectionLike + Send + Sync, { let key = self.session_key(session_id); let session: HashMap = con.hgetall(&key).await?; @@ -205,7 +205,7 @@ Add the crates you need to `Cargo.toml`: ```toml [dependencies] -redis = { version = "0.24", features = ["tokio-comp"] } +redis = { version = "1.8", features = ["tokio-comp"] } tokio = { version = "1", features = ["full"] } axum = "0.7" serde = { version = "1.0", features = ["derive"] } diff --git a/content/develop/use-cases/session-store/rust/session_store.rs b/content/develop/use-cases/session-store/rust/session_store.rs index b713656b30..0a8e8c22c1 100644 --- a/content/develop/use-cases/session-store/rust/session_store.rs +++ b/content/develop/use-cases/session-store/rust/session_store.rs @@ -64,7 +64,7 @@ impl RedisSessionStore { ttl: Option, ) -> RedisResult where - C: redis::aio::ConnectionLike + Send, + C: redis::aio::ConnectionLike + Send + Sync, { let session_id = self.create_session_id(); let key = self.session_key(&session_id); @@ -101,7 +101,7 @@ impl RedisSessionStore { session_id: &str, ) -> RedisResult> where - C: redis::aio::ConnectionLike + Send, + C: redis::aio::ConnectionLike + Send + Sync, { let stored_ttl: Option = con.hget(self.session_key(session_id), "session_ttl").await?; match stored_ttl { @@ -146,7 +146,7 @@ impl RedisSessionStore { refresh_ttl: bool, ) -> RedisResult>> where - C: redis::aio::ConnectionLike + Send, + C: redis::aio::ConnectionLike + Send + Sync, { let key = self.session_key(session_id); let session: HashMap = con.hgetall(&key).await?; @@ -210,7 +210,7 @@ impl RedisSessionStore { data: &HashMap, ) -> RedisResult where - C: redis::aio::ConnectionLike + Send, + C: redis::aio::ConnectionLike + Send + Sync, { let key = self.session_key(session_id); let session: HashMap = con.hgetall(&key).await?; @@ -269,7 +269,7 @@ impl RedisSessionStore { amount: i64, ) -> RedisResult> where - C: redis::aio::ConnectionLike + Send, + C: redis::aio::ConnectionLike + Send + Sync, { let key = self.session_key(session_id); let session: HashMap = con.hgetall(&key).await?; @@ -319,7 +319,7 @@ impl RedisSessionStore { ttl: usize, ) -> RedisResult where - C: redis::aio::ConnectionLike + Send, + C: redis::aio::ConnectionLike + Send + Sync, { let key = self.session_key(session_id); let session: HashMap = con.hgetall(&key).await?; @@ -355,7 +355,7 @@ impl RedisSessionStore { session_id: &str, ) -> RedisResult where - C: redis::aio::ConnectionLike + Send, + C: redis::aio::ConnectionLike + Send + Sync, { let deleted: i64 = con.del(self.session_key(session_id)).await?; Ok(deleted == 1) @@ -378,7 +378,7 @@ impl RedisSessionStore { session_id: &str, ) -> RedisResult where - C: redis::aio::ConnectionLike + Send, + C: redis::aio::ConnectionLike + Send + Sync, { con.ttl(self.session_key(session_id)).await } @@ -389,7 +389,7 @@ impl RedisSessionStore { fn normalize_ttl_value(ttl: usize) -> RedisResult { if ttl < 1 { - Err(RedisError::from((ErrorKind::TypeError, "TTL must be at least 1 second"))) + Err(RedisError::from((ErrorKind::UnexpectedReturnType, "TTL must be at least 1 second"))) } else { Ok(ttl) } @@ -402,11 +402,11 @@ impl RedisSessionStore { fn timestamp(&self) -> RedisResult { let mut ts = OffsetDateTime::now_utc() .replace_nanosecond(0) - .map_err(|_| RedisError::from((ErrorKind::TypeError, "invalid timestamp")))?; + .map_err(|_| RedisError::from((ErrorKind::UnexpectedReturnType, "invalid timestamp")))?; ts = ts.to_offset(time::UtcOffset::UTC); let formatted = ts .format(&Rfc3339) - .map_err(|_| RedisError::from((ErrorKind::TypeError, "invalid timestamp format")))?; + .map_err(|_| RedisError::from((ErrorKind::UnexpectedReturnType, "invalid timestamp format")))?; Ok(formatted.replace('Z', "+00:00")) } @@ -451,8 +451,8 @@ impl RedisSessionStore { fn parse_ttl(session: &HashMap) -> RedisResult { session .get("session_ttl") - .ok_or_else(|| RedisError::from((ErrorKind::TypeError, "missing session_ttl")))? + .ok_or_else(|| RedisError::from((ErrorKind::UnexpectedReturnType, "missing session_ttl")))? .parse::() - .map_err(|_| RedisError::from((ErrorKind::TypeError, "invalid session_ttl"))) + .map_err(|_| RedisError::from((ErrorKind::UnexpectedReturnType, "invalid session_ttl"))) } } diff --git a/content/develop/use-cases/streaming/rust/Cargo.toml b/content/develop/use-cases/streaming/rust/Cargo.toml index 6b27036b79..f81c9bf1ee 100644 --- a/content/develop/use-cases/streaming/rust/Cargo.toml +++ b/content/develop/use-cases/streaming/rust/Cargo.toml @@ -4,7 +4,7 @@ version = "0.1.0" edition = "2021" [dependencies] -redis = { version = "0.24", features = ["tokio-comp", "aio", "connection-manager", "streams"] } +redis = { version = "1.8", features = ["tokio-comp", "aio", "connection-manager", "streams"] } tokio = { version = "1", features = ["full"] } axum = "0.7" serde = { version = "1.0", features = ["derive"] } diff --git a/content/develop/use-cases/streaming/rust/_index.md b/content/develop/use-cases/streaming/rust/_index.md index e532820594..3e22150844 100644 --- a/content/develop/use-cases/streaming/rust/_index.md +++ b/content/develop/use-cases/streaming/rust/_index.md @@ -52,12 +52,15 @@ The `EventStream` struct wraps the stream operations ([source](https://github.com/redis/docs/blob/main/content/develop/use-cases/streaming/rust/event_stream.rs)): ```rust -use redis::aio::ConnectionManager; +use redis::aio::{ConnectionManager, ConnectionManagerConfig}; use redis::Client; use std::collections::HashMap; let client = Client::open("redis://127.0.0.1:6379/")?; -let conn = ConnectionManager::new(client).await?; +// Consumers block in XREADGROUP ... BLOCK for longer than the 500 ms default +// response timeout in redis-rs 1.x, so turn the response timeout off. +let config = ConnectionManagerConfig::new().set_response_timeout(None); +let conn = ConnectionManager::new_with_config(client, config).await?; let stream = EventStream::new( conn, "demo:events:orders", // stream key @@ -252,7 +255,7 @@ pub async fn reap_idle_pel(&self) -> ReapResult { } ``` -The underlying `stream.autoclaim` helper pages through the group's PEL with `XAUTOCLAIM`'s continuation cursor. redis-rs 0.24 has no typed wrapper for `XAUTOCLAIM`, so the helper builds it from raw `redis::cmd("XAUTOCLAIM")` and decodes the three-element Redis 7+ reply by hand: +The underlying `stream.autoclaim` helper pages through the group's PEL with `XAUTOCLAIM`'s continuation cursor. the helper builds it from the generic `redis::cmd("XAUTOCLAIM")` builder (redis-rs 1.x also has a typed `xautoclaim_options`) and decodes the three-element Redis 7+ reply by hand: ```rust pub async fn autoclaim( @@ -372,14 +375,14 @@ The pause and crash levers exist only for the demo. A real consumer is just the * Redis 7.0 or later. `XAUTOCLAIM` was added in Redis 6.2, but its reply gained a third element (the list of deleted IDs) in 7.0; the example relies on that shape. -* Rust 1.70 or later (stable). -* The `redis` crate at version 0.24+, with features `["tokio-comp", "aio", "connection-manager", "streams"]`. The `connection-manager` feature gives `ConnectionManager` for cheap, cloneable, auto-reconnecting connections; `streams` adds the typed reply structs for `XINFO`/`XPENDING`. +* Rust 1.88 or later (the `redis` 1.8 crate requires it). +* The `redis` crate at version 1.8 or later, with features `["tokio-comp", "aio", "connection-manager", "streams"]`. The `connection-manager` feature gives `ConnectionManager` for cheap, cloneable, auto-reconnecting connections; `streams` adds the typed reply structs for `XINFO`/`XPENDING`. The `Cargo.toml` for this demo pins all the runtime crates: ```toml [dependencies] -redis = { version = "0.24", features = ["tokio-comp", "aio", "connection-manager", "streams"] } +redis = { version = "1.8", features = ["tokio-comp", "aio", "connection-manager", "streams"] } tokio = { version = "1", features = ["full"] } axum = "0.7" serde = { version = "1.0", features = ["derive"] } diff --git a/content/develop/use-cases/streaming/rust/demo_server.rs b/content/develop/use-cases/streaming/rust/demo_server.rs index 9dee2c6580..a60d5e8c4f 100644 --- a/content/develop/use-cases/streaming/rust/demo_server.rs +++ b/content/develop/use-cases/streaming/rust/demo_server.rs @@ -345,7 +345,10 @@ async fn main() { let url = format!("redis://{}:{}/", redis_host, redis_port); let client = Client::open(url).expect("failed to create Redis client"); - let conn = ConnectionManager::new(client) + // Consumers block in XREADGROUP ... BLOCK for longer than the 500 ms default + // response timeout in redis-rs 1.x, so turn the response timeout off. + let config = redis::aio::ConnectionManagerConfig::new().set_response_timeout(None); + let conn = ConnectionManager::new_with_config(client, config) .await .expect("failed to connect to Redis"); diff --git a/content/develop/use-cases/streaming/rust/event_stream.rs b/content/develop/use-cases/streaming/rust/event_stream.rs index 77f55ccada..6e076f5ea7 100644 --- a/content/develop/use-cases/streaming/rust/event_stream.rs +++ b/content/develop/use-cases/streaming/rust/event_stream.rs @@ -461,17 +461,17 @@ impl EventStream { Ok(v) => v, Err(_) => return StreamInfo::default(), }; - let map: HashMap = match redis::FromRedisValue::from_redis_value(&raw) { + let map: HashMap = match redis::FromRedisValue::from_redis_value_ref(&raw) { Ok(m) => m, Err(_) => return StreamInfo::default(), }; let length: u64 = map .get("length") - .and_then(|v| redis::FromRedisValue::from_redis_value(v).ok()) + .and_then(|v| redis::FromRedisValue::from_redis_value_ref(v).ok()) .unwrap_or(0); let last_generated_id: Option = map .get("last-generated-id") - .and_then(|v| redis::FromRedisValue::from_redis_value(v).ok()); + .and_then(|v| redis::FromRedisValue::from_redis_value_ref(v).ok()); let first_entry_id = entry_id_from_value(map.get("first-entry")); let last_entry_id = entry_id_from_value(map.get("last-entry")); StreamInfo { @@ -498,7 +498,7 @@ impl EventStream { // a vec of HashMap so we can pull only the keys // we care about (and tolerate version differences). let rows: Vec> = - match redis::FromRedisValue::from_redis_value(&raw) { + match redis::FromRedisValue::from_redis_value_ref(&raw) { Ok(v) => v, Err(_) => return Vec::new(), }; @@ -506,23 +506,23 @@ impl EventStream { .map(|map| GroupInfo { name: map .get("name") - .and_then(|v| redis::FromRedisValue::from_redis_value(v).ok()) + .and_then(|v| redis::FromRedisValue::from_redis_value_ref(v).ok()) .unwrap_or_default(), consumers: map .get("consumers") - .and_then(|v| redis::FromRedisValue::from_redis_value(v).ok()) + .and_then(|v| redis::FromRedisValue::from_redis_value_ref(v).ok()) .unwrap_or(0), pending: map .get("pending") - .and_then(|v| redis::FromRedisValue::from_redis_value(v).ok()) + .and_then(|v| redis::FromRedisValue::from_redis_value_ref(v).ok()) .unwrap_or(0), last_delivered_id: map .get("last-delivered-id") - .and_then(|v| redis::FromRedisValue::from_redis_value(v).ok()) + .and_then(|v| redis::FromRedisValue::from_redis_value_ref(v).ok()) .unwrap_or_default(), lag: map .get("lag") - .and_then(|v| redis::FromRedisValue::from_redis_value(v).ok()), + .and_then(|v| redis::FromRedisValue::from_redis_value_ref(v).ok()), }) .collect() } @@ -541,7 +541,7 @@ impl EventStream { Err(_) => return Vec::new(), }; let rows: Vec> = - match redis::FromRedisValue::from_redis_value(&raw) { + match redis::FromRedisValue::from_redis_value_ref(&raw) { Ok(v) => v, Err(_) => return Vec::new(), }; @@ -549,15 +549,15 @@ impl EventStream { .map(|map| ConsumerInfo { name: map .get("name") - .and_then(|v| redis::FromRedisValue::from_redis_value(v).ok()) + .and_then(|v| redis::FromRedisValue::from_redis_value_ref(v).ok()) .unwrap_or_default(), pending: map .get("pending") - .and_then(|v| redis::FromRedisValue::from_redis_value(v).ok()) + .and_then(|v| redis::FromRedisValue::from_redis_value_ref(v).ok()) .unwrap_or(0), idle_ms: map .get("idle") - .and_then(|v| redis::FromRedisValue::from_redis_value(v).ok()) + .and_then(|v| redis::FromRedisValue::from_redis_value_ref(v).ok()) .unwrap_or(0), }) .collect() @@ -675,7 +675,7 @@ fn stream_ids_to_entries(ids: Vec) -> Vec { fn fields_from_stream_map(map: &HashMap) -> HashMap { let mut out: HashMap = HashMap::with_capacity(map.len()); for (k, v) in map { - let s: String = redis::FromRedisValue::from_redis_value(v).unwrap_or_default(); + let s: String = redis::FromRedisValue::from_redis_value_ref(v).unwrap_or_default(); out.insert(k.clone(), s); } out @@ -688,9 +688,9 @@ fn entry_id_from_value(v: Option<&Value>) -> Option { if matches!(v, Value::Nil) { return None; } - if let Value::Bulk(items) = v { + if let Value::Array(items) = v { if let Some(first) = items.first() { - if let Ok(id) = redis::FromRedisValue::from_redis_value(first) { + if let Ok(id) = redis::FromRedisValue::from_redis_value_ref(first) { return Some(id); } } @@ -702,10 +702,10 @@ fn entry_id_from_value(v: Option<&Value>) -> Option { /// `[ next-cursor, [ [id, [field, value, ...]], ... ], [deleted-id, ...] ]`. fn parse_autoclaim_reply(raw: Value) -> RedisResult<(String, Vec, Vec)> { let parts = match raw { - Value::Bulk(parts) => parts, + Value::Array(parts) => parts, _ => { return Err(RedisError::from(( - redis::ErrorKind::TypeError, + redis::ErrorKind::UnexpectedReturnType, "XAUTOCLAIM: expected array reply", ))) } @@ -713,22 +713,22 @@ fn parse_autoclaim_reply(raw: Value) -> RedisResult<(String, Vec, Vec = match deleted_v { - Some(v) => redis::FromRedisValue::from_redis_value(&v).unwrap_or_default(), + Some(v) => redis::FromRedisValue::from_redis_value_ref(&v).unwrap_or_default(), None => Vec::new(), }; diff --git a/content/develop/use-cases/time-series-dashboard/rust/Cargo.toml b/content/develop/use-cases/time-series-dashboard/rust/Cargo.toml index 88cfb1b401..dec79d11f6 100644 --- a/content/develop/use-cases/time-series-dashboard/rust/Cargo.toml +++ b/content/develop/use-cases/time-series-dashboard/rust/Cargo.toml @@ -6,7 +6,7 @@ edition = "2021" [dependencies] axum = "0.7" rand = "0.8" -redis = { version = "0.24", features = ["tokio-comp"] } +redis = { version = "1.8", features = ["tokio-comp"] } serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" tokio = { version = "1", features = ["full"] } diff --git a/content/develop/use-cases/time-series-dashboard/rust/timeseries_store.rs b/content/develop/use-cases/time-series-dashboard/rust/timeseries_store.rs index 6d57a0f57d..2c44cc5e8b 100644 --- a/content/develop/use-cases/time-series-dashboard/rust/timeseries_store.rs +++ b/content/develop/use-cases/time-series-dashboard/rust/timeseries_store.rs @@ -226,7 +226,7 @@ impl RedisTimeSeriesStore { }; match value { - Value::Bulk(values) if values.len() >= 2 => Ok(Some(LatestValue { + Value::Array(values) if values.len() >= 2 => Ok(Some(LatestValue { timestamp: as_i64(&values[0])?, value: as_f64(&values[1])?, })), @@ -239,19 +239,19 @@ impl RedisTimeSeriesStore { fn now_millis() -> RedisResult { let duration = SystemTime::now() .duration_since(UNIX_EPOCH) - .map_err(|_| RedisError::from((ErrorKind::TypeError, "system clock went backwards")))?; + .map_err(|_| RedisError::from((ErrorKind::UnexpectedReturnType, "system clock went backwards")))?; i64::try_from(duration.as_millis()) - .map_err(|_| RedisError::from((ErrorKind::TypeError, "timestamp overflow"))) + .map_err(|_| RedisError::from((ErrorKind::UnexpectedReturnType, "timestamp overflow"))) } fn parse_points(value: Value) -> RedisResult> { let rows = match value { - Value::Bulk(rows) => rows, + Value::Array(rows) => rows, Value::Nil => return Ok(Vec::new()), other => { return Err(RedisError::from(( - ErrorKind::TypeError, + ErrorKind::UnexpectedReturnType, "unexpected TS.RANGE response", format!("{other:?}"), ))) @@ -260,7 +260,7 @@ fn parse_points(value: Value) -> RedisResult> { let mut points = Vec::with_capacity(rows.len()); for row in rows { - let Value::Bulk(pair) = row else { + let Value::Array(pair) = row else { continue; }; if pair.len() < 2 { @@ -279,11 +279,11 @@ fn parse_points(value: Value) -> RedisResult> { fn as_i64(value: &Value) -> RedisResult { match value { Value::Int(number) => Ok(*number), - Value::Data(bytes) => String::from_utf8_lossy(bytes) + Value::BulkString(bytes) => String::from_utf8_lossy(bytes) .parse::() - .map_err(|_| RedisError::from((ErrorKind::TypeError, "expected integer value"))), + .map_err(|_| RedisError::from((ErrorKind::UnexpectedReturnType, "expected integer value"))), other => Err(RedisError::from(( - ErrorKind::TypeError, + ErrorKind::UnexpectedReturnType, "expected integer value", format!("{other:?}"), ))), @@ -293,11 +293,16 @@ fn as_i64(value: &Value) -> RedisResult { fn as_f64(value: &Value) -> RedisResult { match value { Value::Int(number) => Ok(*number as f64), - Value::Data(bytes) => String::from_utf8_lossy(bytes) + Value::Double(number) => Ok(*number), + // Newer RedisTimeSeries replies send RESP2 doubles as simple strings. + Value::SimpleString(text) => text .parse::() - .map_err(|_| RedisError::from((ErrorKind::TypeError, "expected floating-point value"))), + .map_err(|_| RedisError::from((ErrorKind::UnexpectedReturnType, "expected floating-point value"))), + Value::BulkString(bytes) => String::from_utf8_lossy(bytes) + .parse::() + .map_err(|_| RedisError::from((ErrorKind::UnexpectedReturnType, "expected floating-point value"))), other => Err(RedisError::from(( - ErrorKind::TypeError, + ErrorKind::UnexpectedReturnType, "expected floating-point value", format!("{other:?}"), ))), From 9dad069644d0182d4dfcc474c7d63690cbc37b79 Mon Sep 17 00:00:00 2001 From: Andy Stark Date: Fri, 30 Oct 2026 09:20:36 +0000 Subject: [PATCH 3/3] DOC-7226 Return SCAN errors from the feature-store entity count count_entities looped over the scan iterator with while let Some(_k) and never looked at the item. In redis-rs 1.x next_item yields Option>, so a SCAN error was counted as an entity and never returned. Bugbot caught it. The compiler could not, because a pattern that discards the item type-checks under both versions. The loop now propagates the error, like the other two scan loops in the file. Checked every other next_item loop in the Rust use cases; the only other two (prefetch-cache) already propagate the error. The feature-store demo still reports 200 entities. Learned: after a redis-rs 1.x bump, a loop that discards the item (while let Some(_k) = iter.next_item()) compiles unchanged but silently counts scan errors as items, so grep every next_item loop instead of relying on compiler errors Ticket: DOC-7226 Co-Authored-By: Claude Sonnet 5.5 --- content/develop/use-cases/feature-store/rust/feature_store.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/content/develop/use-cases/feature-store/rust/feature_store.rs b/content/develop/use-cases/feature-store/rust/feature_store.rs index 99d57bb643..50970bfb57 100644 --- a/content/develop/use-cases/feature-store/rust/feature_store.rs +++ b/content/develop/use-cases/feature-store/rust/feature_store.rs @@ -423,7 +423,8 @@ impl FeatureStore { let mut conn = self.conn.clone(); let mut iter: AsyncIter = conn.scan_match(&pattern).await?; let mut n = 0i64; - while let Some(_k) = iter.next_item().await { + while let Some(k) = iter.next_item().await { + k?; n += 1; } Ok(n)