Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion content/develop/use-cases/cache-aside/rust/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -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"] }
Expand Down
8 changes: 4 additions & 4 deletions content/develop/use-cases/cache-aside/rust/_index.md
Original file line number Diff line number Diff line change
Expand Up @@ -191,7 +191,7 @@ where
.release_script
.key(&lock_key)
.arg(token.as_str())
.invoke_async::<_, i64>(&mut conn)
.invoke_async::<i64>(&mut conn)
.await;
return result;
}
Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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"] }
Expand Down
8 changes: 4 additions & 4 deletions content/develop/use-cases/cache-aside/rust/cache.rs
Original file line number Diff line number Diff line change
Expand Up @@ -149,7 +149,7 @@ impl RedisCache {
.release_script
.key(&lock_key)
.arg(token.as_str())
.invoke_async::<_, i64>(&mut conn)
.invoke_async::<i64>(&mut conn)
.await;
return result;
}
Expand Down Expand Up @@ -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)
Expand All @@ -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()
Expand Down
2 changes: 1 addition & 1 deletion content/develop/use-cases/feature-store/rust/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
31 changes: 15 additions & 16 deletions content/develop/use-cases/feature-store/rust/_index.md
Original file line number Diff line number Diff line change
Expand Up @@ -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`,
Expand Down Expand Up @@ -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:?}"),
)));
Expand Down Expand Up @@ -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

Expand All @@ -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<Option<String>> = conn.hget(&key, field_names).await?;
let values: Vec<Option<String>> = 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); }
Expand Down Expand Up @@ -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<Vec<Option<String>>> = pipe.query_async(&mut conn).await?;
Expand All @@ -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
Expand Down Expand Up @@ -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")`.

Expand Down Expand Up @@ -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.
Expand Down
14 changes: 8 additions & 6 deletions content/develop/use-cases/feature-store/rust/feature_store.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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:?}"
Expand Down Expand Up @@ -278,7 +278,7 @@ impl FeatureStore {
}
let key = self.key_for(entity_id);
let mut conn = self.conn.clone();
let values: Vec<Option<String>> = conn.hget(&key, field_names).await?;
let values: Vec<Option<String>> = 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 {
Expand Down Expand Up @@ -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<Vec<Option<String>>> = pipe.query_async(&mut conn).await?;
Expand Down Expand Up @@ -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<i64> = Vec::<i64>::from_redis_value(&values).unwrap_or_else(|_| {
let codes: Vec<i64> = Vec::<i64>::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()
Expand All @@ -404,6 +404,7 @@ impl FeatureStore {
let mut ids: Vec<String> = Vec::new();
let prefix_len = self.key_prefix.len();
while let Some(k) = iter.next_item().await {
let k = k?;

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

SCAN count ignores iterator errors

Medium Severity

count_entities still treats every next_item() yield as a key. In redis-rs 1.8 that method returns Option<RedisResult<T>>, so a SCAN error is counted as an entity and never returned. The same file already unwraps the Result in list_entity_ids and reset.

Fix in Cursor Fix in Web

Reviewed by Cursor Bugbot for commit 51fde74. Configure here.

if k.len() > prefix_len {
ids.push(k[prefix_len..].to_string());
}
Expand All @@ -422,7 +423,8 @@ impl FeatureStore {
let mut conn = self.conn.clone();
let mut iter: AsyncIter<String> = 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)
Expand All @@ -444,7 +446,7 @@ impl FeatureStore {
let mut batch: Vec<String> = Vec::with_capacity(500);
let mut iter: AsyncIter<String> = 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;
Expand Down
Loading
Loading