From a8273e9dc7bbcd4bf82fd671ddba42a409a22072 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 13 Oct 2026 12:30:11 -0700 Subject: [PATCH 1/2] Checked the Redis version and eviction policy on each primary. --- crates/streams/Cargo.toml | 1 + crates/streams/docs/redis.md | 42 +++++ crates/streams/scripts/redis-test-servers.sh | 51 ++++++ crates/streams/src/redis/conn.rs | 30 ++++ crates/streams/src/redis/mod.rs | 159 +++++++++++++++-- crates/streams/tests/conformance.rs | 139 +++++++++++++++ crates/streams/tests/redis_store.rs | 173 +++++++++++++++++++ 7 files changed, 577 insertions(+), 18 deletions(-) create mode 100644 crates/streams/docs/redis.md create mode 100755 crates/streams/scripts/redis-test-servers.sh diff --git a/crates/streams/Cargo.toml b/crates/streams/Cargo.toml index 2a01324de4..dfbeba218a 100644 --- a/crates/streams/Cargo.toml +++ b/crates/streams/Cargo.toml @@ -40,6 +40,7 @@ default-features = false [dev-dependencies] serde_json = { workspace = true } +tracing-subscriber = { version = "0.3", default-features = false, features = ["fmt"] } tokio = { version = "1.47", features = [ "io-util", "macros", diff --git a/crates/streams/docs/redis.md b/crates/streams/docs/redis.md new file mode 100644 index 0000000000..6427b7dffb --- /dev/null +++ b/crates/streams/docs/redis.md @@ -0,0 +1,42 @@ +# The Redis stream store + +The store on the application's Redis. + +Records live in Redis and never pass through Temporal. The key layout, the stored values and +the Lua scripts are the ones the SDKs used before the store moved into Core, apart from the +append's topic check, so a stream one wrote stays readable and writable by the other. + +**Appends.** One script writes a whole batch, so a reader never sees part of one. The topic's +meta hash keeps one field per producer attempt: the newest batch's first sequence, its record +count, its first and last entry ids, and lang's digest of it in hex. The script refuses a +new batch once the chain or the topic is closed, in the same step as the write, so no batch +lands after a close returned. Only that topic check differs from the SDKs' script. + +**Retention.** Every script that writes a log trims it with `XTRIM MINID` to the retention and +slides the expiry of the log and its meta, so a stream dies a retention after its last write +and nobody cleans up when the Workflow closes. The meta outlives its log by a 30-day grace, as +the tombstone that tells a log that expired from one that never existed. + +**Server.** Redis 7.0 or later is required, and the store refuses an older server when it +connects. At the same moment it reads `maxmemory-policy` on every primary and logs a warning +for anything but `noeviction`, since an evicting Redis drops whole stream keys under memory +pressure, losing records and their dedupe state. + +**ACL.** Every key of a namespace's streams starts with `:{:`, each part +percent-encoded. A Redis ACL user for one namespace's applications, with the rules the +conformance suite runs as: + +```text +ACL SETUSER streams-app on >secret resetkeys ~temporal-streams:{my-ns:* resetchannels -@all +evalsha +eval +script|load +multi +exec +xadd +xread +xrevrange +xrange +xtrim +xlen +hget +hset +hgetall +hincrby +hdel +hscan +rpush +lrange +exists +del +unlink +pexpire +pttl +time +info +config|get +ping +hello +client|setinfo +cluster|slots +``` + +A cluster client reads the topology with `CLUSTER SLOTS` first. An ACL user lives on one node, +so on a cluster the user must exist on every node. Deleting an owner's streams also needs +`+scan`. `+config|get` only reads +`maxmemory-policy`, and the store goes on when a server refuses it. + +**Tests.** The Redis tests run when `STREAMS_REDIS_URL` names a standalone server and +`STREAMS_REDIS_CLUSTER_URL` a cluster seed. `scripts/redis-test-servers.sh` starts both. The +conformance suite runs on each, and once more as a user with exactly the ACL above. The +cross-implementation test also needs `STREAMS_BROOK_PY`, a Python SDK checkout whose provider +predates the move. diff --git a/crates/streams/scripts/redis-test-servers.sh b/crates/streams/scripts/redis-test-servers.sh new file mode 100755 index 0000000000..a5d27198e5 --- /dev/null +++ b/crates/streams/scripts/redis-test-servers.sh @@ -0,0 +1,51 @@ +#!/usr/bin/env bash +# Starts the Redis servers the stream store tests run against: one standalone server, and a +# cluster of three primaries. Uses a local redis-server when there is one, and docker on a Linux +# host otherwise, since a cluster's nodes must reach each other on the addresses they announce. +# +# STREAMS_REDIS_URL=redis://localhost:6384/0 +# STREAMS_REDIS_CLUSTER_URL=redis://localhost:7101/0 +set -euo pipefail + +STANDALONE=${STREAMS_STANDALONE_PORT:-6384} +CLUSTER=(7101 7102 7103) +DATA=${STREAMS_REDIS_DATA:-${TMPDIR:-/tmp}/streams-redis} +mkdir -p "$DATA" + +start() { + local port=$1 + shift + if redis-cli -p "$port" ping >/dev/null 2>&1; then + return + fi + if command -v redis-server >/dev/null; then + mkdir -p "$DATA/$port" + redis-server --port "$port" --save '' --appendonly no --daemonize yes \ + --dir "$DATA/$port" --logfile "$DATA/$port.log" "$@" + else + docker run -d --rm --name "streams-redis-$port" --network host redis:7 \ + redis-server --port "$port" --save '' --appendonly no "$@" >/dev/null + fi +} + +start "$STANDALONE" +for port in "${CLUSTER[@]}"; do + start "$port" --cluster-enabled yes --cluster-config-file nodes.conf +done +for _ in $(seq 50); do + redis-cli -p "${CLUSTER[2]}" ping >/dev/null 2>&1 && break + sleep 0.1 +done +if ! redis-cli -p "${CLUSTER[0]}" cluster info | grep -q cluster_state:ok; then + nodes=() + for port in "${CLUSTER[@]}"; do + nodes+=("127.0.0.1:$port") + done + redis-cli --cluster create "${nodes[@]}" --cluster-replicas 0 --cluster-yes >/dev/null + for _ in $(seq 50); do + redis-cli -p "${CLUSTER[0]}" cluster info | grep -q cluster_state:ok && break + sleep 0.1 + done +fi +echo "STREAMS_REDIS_URL=redis://localhost:$STANDALONE/0" +echo "STREAMS_REDIS_CLUSTER_URL=redis://localhost:${CLUSTER[0]}/0" diff --git a/crates/streams/src/redis/conn.rs b/crates/streams/src/redis/conn.rs index e206adac7e..824c2821b6 100644 --- a/crates/streams/src/redis/conn.rs +++ b/crates/streams/src/redis/conn.rs @@ -12,6 +12,7 @@ use redis::{ aio::{ConnectionLike, ConnectionManager, ConnectionManagerConfig, MultiplexedConnection}, cluster::ClusterClient, cluster_async::ClusterConnection, + cluster_routing::{MultipleNodeRoutingInfo, RoutingInfo}, }; use std::{ collections::HashMap, @@ -76,6 +77,35 @@ impl Shared { client.get_connection_manager_with_config(config).await?, )) } + + /// Runs `cmd` on the server, or on every primary of a cluster, and answers each reply with + /// the node that sent it. Each primary holds its own share of the keys and its own settings. + pub(crate) async fn on_each_primary(&self, cmd: Cmd) -> RedisResult> { + match self.clone() { + Shared::Single(mut conn) => { + Ok(vec![(String::new(), cmd.query_async(&mut conn).await?)]) + } + Shared::Cluster(mut conn) => { + let routing = RoutingInfo::MultiNode((MultipleNodeRoutingInfo::AllMasters, None)); + match conn.route_command(cmd, routing).await? { + Value::Map(replies) => Ok(replies + .into_iter() + .map(|(node, reply)| { + let node = match node { + Value::BulkString(bytes) => { + String::from_utf8_lossy(&bytes).into_owned() + } + Value::SimpleString(text) => text, + other => format!("{other:?}"), + }; + (node, reply) + }) + .collect()), + other => Ok(vec![(String::new(), other)]), + } + } + } + } } /// The node a key lives on, for a store that talks to one. diff --git a/crates/streams/src/redis/mod.rs b/crates/streams/src/redis/mod.rs index e294e7a8db..3a8e515a7e 100644 --- a/crates/streams/src/redis/mod.rs +++ b/crates/streams/src/redis/mod.rs @@ -1,21 +1,4 @@ -//! The store on the application's Redis. -//! -//! Records live in Redis and never pass through Temporal. The key layout, the stored values and -//! the Lua scripts are the ones the SDKs used before the store moved into Core, so a stream one -//! wrote stays readable and writable by the other. -//! -//! **Appends.** One script writes a whole batch, so a reader never sees part of one. The topic's -//! meta hash keeps one field per producer attempt: the newest batch's first sequence, its record -//! count, its first and last entry ids, and lang's digest of it in hex. The script refuses a -//! new batch once the chain or the topic is closed, in the same step as the write, so no batch -//! lands after a close returned. Only that topic check differs from the SDKs' script. -//! -//! **Retention.** Every script that writes a log trims it with `XTRIM MINID` to the retention and -//! slides the expiry of the log and its meta, so a stream dies a retention after its last write -//! and nobody cleans up when the Workflow closes. The meta outlives its log by a 30-day grace, as -//! the tombstone that tells a log that expired from one that never existed. -//! -//! Redis 7.0 or later is required. +#![doc = include_str!("../../docs/redis.md")] mod conn; mod errors; @@ -114,6 +97,7 @@ impl RedisStore { let shared = Shared::connect(&options.urls, options.cluster, options.response_timeout) .await .mapped(Call::Read)?; + check_server(&shared).await?; let blocking = BlockingPool::new( seed, options.blocking_reads_per_node, @@ -167,6 +151,100 @@ impl RedisStore { } } +/// Refuses a server older than Redis 7.0, and warns about one that may evict stream keys. +async fn check_server(shared: &Shared) -> StreamResult<()> { + let mut info = redis::cmd("INFO"); + info.arg("server"); + for (node, reply) in shared.on_each_primary(info).await.mapped(Call::Read)? { + let text: String = redis::from_redis_value(reply).map_err(|error| { + StreamError::storage(format!("Redis answered INFO with no text: {error}")) + })?; + check_version(&node, &text)?; + } + let mut policy = redis::cmd("CONFIG"); + policy.arg("GET").arg("maxmemory-policy"); + match shared.on_each_primary(policy).await { + Ok(replies) => { + let policies: Vec<_> = replies + .into_iter() + .map(|(node, reply)| (node, eviction_policy(reply))) + .collect(); + if let Some(warning) = eviction_warning(&policies) { + tracing::warn!("{warning}"); + } + } + // Managed services often refuse CONFIG. The policy is theirs to document then. + Err(error) => tracing::debug!("Could not read maxmemory-policy: {error}"), + } + Ok(()) +} + +fn check_version(node: &str, info: &str) -> StreamResult<()> { + let version = info + .lines() + .find_map(|line| line.trim().strip_prefix("redis_version:")) + .unwrap_or("0"); + let major: u32 = version + .split('.') + .next() + .and_then(|major| major.parse().ok()) + .unwrap_or(0); + if major < 7 { + let at = if node.is_empty() { + String::new() + } else { + format!(" at {node}") + }; + return Err(StreamError::unsupported(format!( + "the Redis store needs Redis 7.0 or later, but the server{at} reports redis_version \ + {version}" + ))); + } + Ok(()) +} + +fn eviction_policy(reply: Value) -> Option { + let pairs: Vec<(String, String)> = match reply { + Value::Map(pairs) => pairs + .into_iter() + .filter_map(|(name, value)| { + Some(( + redis::from_redis_value(name).ok()?, + redis::from_redis_value(value).ok()?, + )) + }) + .collect(), + other => { + let flat: Vec = redis::from_redis_value(other).ok()?; + flat.chunks(2) + .filter_map(|pair| Some((pair.first()?.clone(), pair.get(1)?.clone()))) + .collect() + } + }; + pairs + .into_iter() + .find(|(name, _)| name == "maxmemory-policy") + .map(|(_, value)| value) +} + +fn eviction_warning(policies: &[(String, Option)]) -> Option { + let (node, policy) = policies.iter().find_map(|(node, policy)| { + policy + .as_deref() + .filter(|policy| !policy.is_empty() && *policy != "noeviction") + .map(|policy| (node, policy)) + })?; + let at = if node.is_empty() { + String::new() + } else { + format!(" at {node}") + }; + Some(format!( + "Redis{at} has maxmemory-policy {policy:?}. Under memory pressure Redis may evict whole \ + stream keys, losing records and their dedupe state. Use \"noeviction\" for streams." + )) +} + /// A Redis stream id as its two numbers, so ids compare in log order. fn entry(id: &str) -> StreamResult<(u64, u64)> { let (milliseconds, sequence) = id.split_once('-').unwrap_or((id, "0")); @@ -501,6 +579,7 @@ impl StreamStore for RedisStore { #[cfg(test)] mod tests { use super::*; + use crate::proto::StreamFailureKind; #[test] fn stream_ids_compare_in_log_order() { @@ -512,6 +591,50 @@ mod tests { } } + #[test] + fn a_server_older_than_redis_7_is_refused() { + for (info, ok) in [ + ("# Server\r\nredis_version:7.0.0\r\n", true), + ("redis_version:8.10.1\r\nredis_mode:standalone", true), + ("redis_version:6.2.14\r\n", false), + ("no version here", false), + ] { + let checked = check_version("", info); + assert_eq!(checked.is_ok(), ok, "{info:?}"); + if let Err(error) = checked { + assert_eq!(error.kind, StreamFailureKind::Unsupported); + assert!(error.message.contains("Redis 7.0"), "{error}"); + } + } + let error = check_version("10.0.0.2:7102", "redis_version:6.0.0").unwrap_err(); + assert!(error.message.contains("10.0.0.2:7102"), "{error}"); + } + + #[test] + fn only_a_policy_that_may_evict_is_warned_about() { + let primary = |policy: Option<&str>| (String::new(), policy.map(str::to_string)); + assert_eq!(eviction_warning(&[primary(Some("noeviction"))]), None); + // A server that refused CONFIG, or answered nothing, says nothing either way. + assert_eq!(eviction_warning(&[primary(None)]), None); + let warning = eviction_warning(&[ + primary(Some("noeviction")), + ("10.0.0.3:7103".to_string(), Some("allkeys-lru".to_string())), + ]) + .unwrap(); + assert!(warning.contains("allkeys-lru"), "{warning}"); + assert!(warning.contains("10.0.0.3:7103"), "{warning}"); + } + + #[test] + fn the_policy_reads_from_either_reply_shape() { + let bulk = |text: &str| Value::BulkString(text.as_bytes().to_vec()); + let array = Value::Array(vec![bulk("maxmemory-policy"), bulk("allkeys-lru")]); + assert_eq!(eviction_policy(array).as_deref(), Some("allkeys-lru")); + let map = Value::Map(vec![(bulk("maxmemory-policy"), bulk("noeviction"))]); + assert_eq!(eviction_policy(map).as_deref(), Some("noeviction")); + assert_eq!(eviction_policy(Value::Nil), None); + } + #[test] fn the_digest_is_stored_as_lowercase_hex() { assert_eq!(hex(&[0xab, 0x01]), "ab01"); diff --git a/crates/streams/tests/conformance.rs b/crates/streams/tests/conformance.rs index 65bc08585f..6a99317678 100644 --- a/crates/streams/tests/conformance.rs +++ b/crates/streams/tests/conformance.rs @@ -138,6 +138,127 @@ mod on_redis { pub(crate) store: Arc, pub(crate) raw: Raw, pub(crate) prefix: String, + _user: Option, + } + + /// The rules of the documented `ACL SETUSER` line, after the user name. + pub(crate) fn documented_rules() -> Vec { + let page = include_str!("../docs/redis.md"); + let lines: Vec<_> = page + .lines() + .filter_map(|line| line.strip_prefix("ACL SETUSER ")) + .collect(); + assert_eq!( + lines.len(), + 1, + "the Redis page should give one ACL SETUSER line" + ); + lines[0] + .split_whitespace() + .skip(1) + .map(str::to_string) + .collect() + } + + const DOCUMENTED_KEYS: &str = "~temporal-streams:{my-ns:*"; + + /// A Redis user with exactly the documented rules, for one prefix and namespace. The page's + /// key pattern and password are swapped for the test's own, and nothing else changes. + pub(crate) struct DocumentedUser { + admin_urls: Vec, + name: String, + pub(crate) url: String, + } + + impl DocumentedUser { + pub(crate) async fn create( + url: &str, + prefix: &str, + namespace: &str, + extra: &[&str], + ) -> Self { + let rules = documented_rules(); + assert!( + rules.iter().any(|rule| rule == DOCUMENTED_KEYS), + "{rules:?}" + ); + assert!(rules.iter().any(|rule| rule == ">secret"), "{rules:?}"); + let name = unique("streams-acl"); + let password = unique("pw"); + let keys = format!("~{}:{{{}:*", part(prefix), part(namespace)); + let mut setuser = redis::cmd("ACL"); + setuser.arg("SETUSER").arg(&name); + for rule in &rules { + match rule.as_str() { + DOCUMENTED_KEYS => setuser.arg(&keys), + ">secret" => setuser.arg(format!(">{password}")), + rule => setuser.arg(rule), + }; + } + for rule in extra { + setuser.arg(*rule); + } + // An ACL user lives on one node, so a cluster needs it on every node. + let admins = nodes(url).await; + for admin in &admins { + let mut admin = Raw::connect(admin, false).await; + let _: () = admin.query(&setuser).await; + } + let address = url.trim_start_matches("redis://"); + Self { + admin_urls: admins, + url: format!("redis://{name}:{password}@{address}"), + name, + } + } + } + + impl Drop for DocumentedUser { + fn drop(&mut self) { + for url in &self.admin_urls { + if let Ok(mut admin) = + redis::Client::open(url.as_str()).and_then(|client| client.get_connection()) + { + let _: redis::RedisResult<()> = redis::cmd("ACL") + .arg("DELUSER") + .arg(&self.name) + .query(&mut admin); + } + } + } + } + + /// Every node of the server at `url`: the server itself, or each node of a cluster. + async fn nodes(url: &str) -> Vec { + let mut admin = Raw::connect(url, false).await; + let info: String = admin.query(redis::cmd("INFO").arg("cluster")).await; + if !info.contains("cluster_enabled:1") { + return vec![url.to_string()]; + } + let nodes: String = admin.query(redis::cmd("CLUSTER").arg("NODES")).await; + nodes + .lines() + .filter_map(|line| line.split_whitespace().nth(1)) + .map(|address| format!("redis://{}/0", address.split('@').next().unwrap())) + .collect() + } + + pub(crate) async fn redis_as_documented_user(variable: &str, cluster: bool) -> Option { + let Ok(url) = std::env::var(variable) else { + eprintln!("set {variable} to run the Redis store cases"); + return None; + }; + let prefix = unique("conformance-acl"); + let user = DocumentedUser::create(&url, &prefix, "default", &[]).await; + let mut options = RedisStoreOptions::new(user.url.clone()); + options.key_prefix = prefix.clone(); + options.cluster = cluster; + Some(Redis { + store: Arc::new(RedisStore::connect(options).await.unwrap()), + raw: Raw::connect(&url, cluster).await, + prefix, + _user: Some(user), + }) } pub(crate) async fn redis(variable: &str, cluster: bool) -> Option { @@ -154,6 +275,7 @@ mod on_redis { store: Arc::new(RedisStore::connect(options).await.unwrap()), raw: Raw::connect(&url, cluster).await, prefix, + _user: None, }) } @@ -1132,3 +1254,20 @@ mod standalone_redis { mod cluster_redis { redis_cases!(super::on_redis::redis("STREAMS_REDIS_CLUSTER_URL", true)); } + +#[cfg(feature = "redis")] +mod documented_acl_redis { + redis_cases!(super::on_redis::redis_as_documented_user( + "STREAMS_REDIS_URL", + false + )); +} + +#[cfg(feature = "redis")] +mod documented_acl_cluster_redis { + // A cluster client asks the topology first, so the documented rules must allow it. + redis_cases!(super::on_redis::redis_as_documented_user( + "STREAMS_REDIS_CLUSTER_URL", + true + )); +} diff --git a/crates/streams/tests/redis_store.rs b/crates/streams/tests/redis_store.rs index ad61bd3652..de5601239e 100644 --- a/crates/streams/tests/redis_store.rs +++ b/crates/streams/tests/redis_store.rs @@ -1051,3 +1051,176 @@ async fn rule_20_4_a_malformed_entry_fails_with_its_cursor() { .unwrap(); assert_eq!(after.records.len(), 1); } + +/// Tests that change the server's eviction policy run one at a time. +static POLICY: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); + +/// What the store logs while `connect` runs. +async fn logs_of_connect(options: RedisStoreOptions) -> String { + #[derive(Clone, Default)] + struct Logs(Arc>>); + impl std::io::Write for Logs { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + self.0.lock().unwrap().extend_from_slice(bytes); + Ok(bytes.len()) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + let logs = Logs::default(); + let writer = logs.clone(); + let subscriber = tracing_subscriber::fmt() + .with_writer(move || writer.clone()) + .with_ansi(false) + .with_max_level(tracing::Level::WARN) + .finish(); + let _guard = tracing::subscriber::set_default(subscriber); + RedisStore::connect(options).await.unwrap(); + String::from_utf8(logs.0.lock().unwrap().clone()).unwrap() +} + +async fn set_policy(url: &str, policy: &str) { + let mut admin = redis::Client::open(url) + .unwrap() + .get_multiplexed_async_connection() + .await + .unwrap(); + let _: () = redis::cmd("CONFIG") + .arg("SET") + .arg("maxmemory-policy") + .arg(policy) + .query_async(&mut admin) + .await + .unwrap(); +} + +#[tokio::test] +async fn a_noeviction_server_is_not_warned_about() { + let Some(url) = url() else { + return; + }; + let _serial = POLICY.lock().await; + set_policy(&url, "noeviction").await; + let logs = logs_of_connect(RedisStoreOptions::new(url)).await; + assert!(!logs.contains("maxmemory-policy"), "{logs}"); +} + +#[tokio::test] +async fn an_evicting_server_is_warned_about() { + let Some(url) = url() else { + return; + }; + let _serial = POLICY.lock().await; + set_policy(&url, "allkeys-lru").await; + let logs = logs_of_connect(RedisStoreOptions::new(url.clone())).await; + set_policy(&url, "noeviction").await; + assert!(logs.contains("allkeys-lru"), "{logs}"); + assert_eq!(logs.matches("maxmemory-policy").count(), 1, "{logs}"); +} + +#[tokio::test] +async fn a_server_that_refuses_config_is_not_warned_about() { + let Some(url) = url() else { + return; + }; + let _serial = POLICY.lock().await; + set_policy(&url, "allkeys-lru").await; + // Managed services often refuse CONFIG, so the store goes on without the check. + let name = unique("no-config"); + let mut admin = redis::Client::open(url.as_str()) + .unwrap() + .get_multiplexed_async_connection() + .await + .unwrap(); + let _: () = redis::cmd("ACL") + .arg("SETUSER") + .arg(&name) + .arg("on") + .arg(">pw") + .arg("allkeys") + .arg("+@all") + .arg("-config") + .query_async(&mut admin) + .await + .unwrap(); + let address = url.trim_start_matches("redis://"); + let logs = logs_of_connect(RedisStoreOptions::new(format!( + "redis://{name}:pw@{address}" + ))) + .await; + let _: () = redis::cmd("ACL") + .arg("DELUSER") + .arg(&name) + .query_async(&mut admin) + .await + .unwrap(); + set_policy(&url, "noeviction").await; + assert!(!logs.contains("maxmemory-policy"), "{logs}"); +} + +#[tokio::test] +async fn an_evicting_primary_of_a_cluster_is_warned_about() { + let Ok(url) = std::env::var("STREAMS_REDIS_CLUSTER_URL") else { + eprintln!("set STREAMS_REDIS_CLUSTER_URL to run the Redis cluster tests"); + return; + }; + let _serial = POLICY.lock().await; + // One primary of three, so a check of only the seed would miss it. + let mut nodes = redis::Client::open(url.as_str()) + .unwrap() + .get_multiplexed_async_connection() + .await + .unwrap(); + let shards: String = redis::cmd("CLUSTER") + .arg("NODES") + .query_async(&mut nodes) + .await + .unwrap(); + let seed = url + .trim_start_matches("redis://") + .split('/') + .next() + .unwrap(); + let other = shards + .lines() + .filter(|line| line.contains("master")) + .filter_map(|line| line.split_whitespace().nth(1)) + .map(|address| address.split('@').next().unwrap().to_string()) + .find(|address| !seed.ends_with(address.rsplit(':').next().unwrap())) + .unwrap(); + let other_url = format!("redis://{other}/0"); + set_policy(&other_url, "allkeys-lru").await; + let mut options = RedisStoreOptions::new(url); + options.cluster = true; + let logs = logs_of_connect(options).await; + set_policy(&other_url, "noeviction").await; + assert!(logs.contains("allkeys-lru"), "{logs}"); + assert!(logs.contains(other.rsplit(':').next().unwrap()), "{logs}"); +} + +#[tokio::test] +async fn the_documented_acl_is_the_python_sdks() { + // One rule set for every SDK, so an application's ACL keeps working across the move. + let Ok(checkout) = std::env::var("STREAMS_BROOK_PY") else { + eprintln!("set STREAMS_BROOK_PY to compare the documented ACL with the Python SDK's"); + return; + }; + let rules = |page: &str| -> Vec { + page.lines() + .find_map(|line| line.strip_prefix("ACL SETUSER ")) + .unwrap() + .split_whitespace() + .map(str::to_string) + .collect() + }; + let python = std::fs::read_to_string( + std::path::Path::new(&checkout).join("temporalio/contrib/streams/docs/redis-guarantees.md"), + ) + .unwrap(); + // A cluster client asks the topology with `CLUSTER SLOTS`, which the Python SDK's page left + // out. It is the one rule Core adds. + let mut expected = rules(&python); + expected.push("+cluster|slots".to_string()); + assert_eq!(rules(include_str!("../docs/redis.md")), expected); +} From 460cdfb12821454de15b3fdb0170dbb5ab81bb4d Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 13 Oct 2026 12:31:11 -0700 Subject: [PATCH 2/2] Deleted an owner's Redis streams across every primary. --- crates/streams/docs/redis.md | 4 + crates/streams/src/handle.rs | 67 ++++++++- crates/streams/src/redis/conn.rs | 44 +++++- crates/streams/src/redis/keys.rs | 31 ++++ crates/streams/src/redis/mod.rs | 45 ++++-- crates/streams/tests/conformance.rs | 34 ++++- crates/streams/tests/redis_store.rs | 215 ++++++++++++++++++++++++++++ 7 files changed, 423 insertions(+), 17 deletions(-) diff --git a/crates/streams/docs/redis.md b/crates/streams/docs/redis.md index 6427b7dffb..44e2128292 100644 --- a/crates/streams/docs/redis.md +++ b/crates/streams/docs/redis.md @@ -17,6 +17,10 @@ slides the expiry of the log and its meta, so a stream dies a retention after it and nobody cleans up when the Workflow closes. The meta outlives its log by a 30-day grace, as the tombstone that tells a log that expired from one that never existed. +**Deletes.** Deleting an owner's streams scans every primary for the Workflow id's keys. `SCAN` +makes no promise about keys that move to another node while it runs, so a delete during a +cluster reshard can miss some. Run it again once the reshard is done. + **Server.** Redis 7.0 or later is required, and the store refuses an older server when it connects. At the same moment it reads `maxmemory-policy` on every primary and logs a warning for anything but `noeviction`, since an evicting Redis drops whole stream keys under memory diff --git a/crates/streams/src/handle.rs b/crates/streams/src/handle.rs index a9fa335ece..0a4c75d4db 100644 --- a/crates/streams/src/handle.rs +++ b/crates/streams/src/handle.rs @@ -8,9 +8,10 @@ use crate::{ check_append_digest, mint_cursor, owner::{OwnerClient, OwnerDescription, OwnerError}, proto::{ - AppendRequest, AppendResponse, ChainId, CloseRequest, CloseResponse, LatestRequest, - LatestResponse, ReadRequest, ReadResponse, StoreAppendRequest, StoreLatestRequest, - StreamAddress, StreamOwnerKind, append_request::Producer, + AppendRequest, AppendResponse, ChainId, CloseRequest, CloseResponse, DeleteOwnerRequest, + DeleteOwnerResponse, LatestRequest, LatestResponse, ReadRequest, ReadResponse, + StoreAppendRequest, StoreLatestRequest, StreamAddress, StreamOwnerKind, + append_request::Producer, }, reader::{ReadState, read_with}, stored_append_record, stream_hash, @@ -371,6 +372,27 @@ impl Streams { Ok(CloseResponse {}) } + /// Deletes every stream of one owner, across all its chains. It asks Temporal nothing, so it + /// works after the owner is gone. Delete only after the chain closed and its producers + /// stopped, since a late append writes a fresh stream with no dedupe state and no close mark. + pub async fn delete_owner( + &self, + request: DeleteOwnerRequest, + ) -> StreamResult { + if request.owner_kind != StreamOwnerKind::Workflow as i32 { + return Err(StreamError::unsupported(format!( + "this release keeps streams of Workflows only, not owner kind {}", + request.owner_kind + ))); + } + if request.namespace.is_empty() || request.workflow_id.is_empty() { + return Err(StreamError::invalid_argument( + "a delete needs the namespace and the owner's Workflow id", + )); + } + self.store.delete_owner(request).await + } + /// The chain a producer writes on, asking about the owner when the producer is new or its /// last check is older than the recheck interval. async fn producer_chain(&self, key: ProducerKey) -> StreamResult { @@ -1329,6 +1351,45 @@ mod tests { assert_eq!(setup.owners.describes(), before); } + #[tokio::test] + async fn an_owner_delete_asks_temporal_nothing() { + let setup = setup(StreamsOptions::default()); + setup.streams.append(append("p", 1, "1")).await.unwrap(); + let before = setup.owners.describes(); + let request = DeleteOwnerRequest { + namespace: "ns".to_string(), + owner_kind: StreamOwnerKind::Workflow as i32, + workflow_id: "wf".to_string(), + }; + assert!( + setup + .streams + .delete_owner(request.clone()) + .await + .unwrap() + .deleted + > 0 + ); + assert_eq!(setup.owners.describes(), before); + for (field, expected) in [ + ("namespace", StreamFailureKind::InvalidArgument), + ("workflow_id", StreamFailureKind::InvalidArgument), + ("owner_kind", StreamFailureKind::Unsupported), + ] { + let mut bad = request.clone(); + match field { + "namespace" => bad.namespace.clear(), + "workflow_id" => bad.workflow_id.clear(), + _ => bad.owner_kind = StreamOwnerKind::Unspecified as i32, + } + assert_eq!( + kind(setup.streams.delete_owner(bad).await), + expected, + "{field}" + ); + } + } + /// The digest the Python SDK's Redis provider took over `[{"n": 1}, {"n": 2}]` from producer /// `p`, attempt 1, sequence 1. const PYTHON_DIGEST: &str = "0499768af1856bc6fd19cf6cf90e28e971099f1680e8347d70bdb4b65273f30f"; diff --git a/crates/streams/src/redis/conn.rs b/crates/streams/src/redis/conn.rs index 824c2821b6..29e6f88f2b 100644 --- a/crates/streams/src/redis/conn.rs +++ b/crates/streams/src/redis/conn.rs @@ -12,7 +12,7 @@ use redis::{ aio::{ConnectionLike, ConnectionManager, ConnectionManagerConfig, MultiplexedConnection}, cluster::ClusterClient, cluster_async::ClusterConnection, - cluster_routing::{MultipleNodeRoutingInfo, RoutingInfo}, + cluster_routing::{MultipleNodeRoutingInfo, RoutingInfo, SingleNodeRoutingInfo}, }; use std::{ collections::HashMap, @@ -78,6 +78,48 @@ impl Shared { )) } + /// The primaries a scan must visit: the server itself, or each primary of a cluster. + pub(crate) async fn primaries(&self) -> RedisResult>> { + if let Shared::Single(_) = self { + return Ok(vec![None]); + } + Ok(self + .on_each_primary(redis::cmd("PING")) + .await? + .into_iter() + .map(|(node, _)| { + node.rsplit_once(':') + .and_then(|(host, port)| Some((host.to_string(), port.parse().ok()?))) + }) + .collect()) + } + + /// One `SCAN` step on one primary, answering the next cursor and the keys found. + pub(crate) async fn scan( + &self, + node: Option<&Node>, + cursor: &str, + pattern: &str, + ) -> RedisResult<(String, Vec)> { + let mut scan = redis::cmd("SCAN"); + scan.arg(cursor) + .arg("MATCH") + .arg(pattern) + .arg("COUNT") + .arg(1000); + match (self.clone(), node) { + (Shared::Cluster(mut conn), Some((host, port))) => { + let routing = RoutingInfo::SingleNode(SingleNodeRoutingInfo::ByAddress { + host: host.clone(), + port: *port, + }); + redis::from_redis_value(conn.route_command(scan, routing).await?) + .map_err(RedisError::from) + } + (mut shared, _) => scan.query_async(&mut shared).await, + } + } + /// Runs `cmd` on the server, or on every primary of a cluster, and answers each reply with /// the node that sent it. Each primary holds its own share of the keys and its own settings. pub(crate) async fn on_each_primary(&self, cmd: Cmd) -> RedisResult> { diff --git a/crates/streams/src/redis/keys.rs b/crates/streams/src/redis/keys.rs index 9254cb74f0..82da443913 100644 --- a/crates/streams/src/redis/keys.rs +++ b/crates/streams/src/redis/keys.rs @@ -68,6 +68,16 @@ pub(crate) fn session_field(producer_id: &str, attempt: i64) -> String { format!("hw:{}:{producer_id}:{attempt}", producer_id.len()) } +/// The pattern that matches every key of one Workflow id, across all its chains. +pub(crate) fn owner_pattern(prefix: &str, namespace: &str, workflow_id: &str) -> String { + format!( + "{}:{{{}:{}:*", + part(prefix), + part(namespace), + part(workflow_id) + ) +} + #[cfg(test)] mod tests { use super::*; @@ -110,4 +120,25 @@ mod tests { ); } } + + #[test] + fn an_owner_pattern_covers_every_chain_of_one_workflow_only() { + let chain = ChainId { + namespace: "ns".to_string(), + workflow_id: "wf".to_string(), + first_run_id: "run-1".to_string(), + }; + let pattern = owner_pattern("p", "ns", "wf"); + assert_eq!(pattern, "p:{ns:wf:*"); + let matches = |key: &str| key.starts_with(pattern.trim_end_matches('*')); + assert!(matches(&ChainKeys::new("p", &chain).chain())); + // The `:` after the id keeps `wf` from matching the chains of `wf2`. + let other = ChainId { + workflow_id: "wf2".to_string(), + ..chain + }; + assert!(!matches(&ChainKeys::new("p", &other).chain())); + // Each part is encoded as the keys encode it, so a glob character matches only itself. + assert_eq!(owner_pattern("p", "ns", "a*b"), "p:{ns:a%2Ab:*"); + } } diff --git a/crates/streams/src/redis/mod.rs b/crates/streams/src/redis/mod.rs index 3a8e515a7e..4bea9b3b5b 100644 --- a/crates/streams/src/redis/mod.rs +++ b/crates/streams/src/redis/mod.rs @@ -10,11 +10,12 @@ use crate::{ proto::{ ChainId, DeleteOwnerRequest, DeleteOwnerResponse, StoreAppendRequest, StoreAppendResponse, StoreLatestRequest, StoreLatestResponse, StoreReadRequest, StoreReadResponse, StoredRecord, + StreamOwnerKind, }, }; use conn::{BlockingPool, Shared}; use errors::{Call, Mapped}; -use keys::{ChainKeys, session_field}; +use keys::{ChainKeys, owner_pattern, session_field}; use redis::{Cmd, Value, streams::StreamReadReply}; use std::{collections::HashMap, time::Duration}; use tokio::time::Instant; @@ -23,6 +24,7 @@ const NAME: &str = "redis"; const RECORD_FIELD: &str = "r"; const TOMBSTONE_GRACE: Duration = Duration::from_secs(30 * 24 * 60 * 60); const CLOSED_FIELD: &str = "closed"; +const DELETE_BATCH: usize = 500; /// How to reach Redis and how long streams live there. #[derive(Debug, Clone, PartialEq, Eq)] @@ -566,13 +568,40 @@ impl StreamStore for RedisStore { .mapped(Call::Write) } - async fn delete_owner( - &self, - _request: DeleteOwnerRequest, - ) -> StreamResult { - Err(StreamError::unsupported( - "the Redis store does not delete owners yet", - )) + async fn delete_owner(&self, request: DeleteOwnerRequest) -> StreamResult { + if request.owner_kind != StreamOwnerKind::Workflow as i32 { + return Err(StreamError::unsupported(format!( + "this release keeps streams of Workflows only, not owner kind {}", + request.owner_kind + ))); + } + // A Workflow id's chains hash to different slots, so every primary is scanned. Some keys + // may be gone when a later step fails, which is why a failure leaves the outcome unknown. + let pattern = owner_pattern(&self.prefix, &request.namespace, &request.workflow_id); + let mut deleted = 0; + for node in self.shared.primaries().await.mapped(Call::Write)? { + let mut cursor = "0".to_string(); + loop { + let (next, keys) = self + .shared + .scan(node.as_ref(), &cursor, &pattern) + .await + .mapped(Call::Write)?; + for batch in keys.chunks(DELETE_BATCH) { + let removed: u64 = redis::cmd("UNLINK") + .arg(batch) + .query_async(&mut self.shared.clone()) + .await + .mapped(Call::Write)?; + deleted += removed; + } + if next == "0" { + break; + } + cursor = next; + } + } + Ok(DeleteOwnerResponse { deleted }) } } diff --git a/crates/streams/tests/conformance.rs b/crates/streams/tests/conformance.rs index 6a99317678..391f533e21 100644 --- a/crates/streams/tests/conformance.rs +++ b/crates/streams/tests/conformance.rs @@ -243,13 +243,17 @@ mod on_redis { .collect() } - pub(crate) async fn redis_as_documented_user(variable: &str, cluster: bool) -> Option { + pub(crate) async fn redis_as_documented_user( + variable: &str, + cluster: bool, + extra: &[&str], + ) -> Option { let Ok(url) = std::env::var(variable) else { eprintln!("set {variable} to run the Redis store cases"); return None; }; let prefix = unique("conformance-acl"); - let user = DocumentedUser::create(&url, &prefix, "default", &[]).await; + let user = DocumentedUser::create(&url, &prefix, "default", extra).await; let mut options = RedisStoreOptions::new(user.url.clone()); options.key_prefix = prefix.clone(); options.cluster = cluster; @@ -1204,7 +1208,7 @@ mod memory { ); } -/// Every case but owner deletes, which come with that feature. +/// Every case but owner deletes, which need a rule of their own under the documented ACL. #[cfg(feature = "redis")] macro_rules! redis_cases { ($case:expr) => { @@ -1248,19 +1252,31 @@ macro_rules! redis_cases { #[cfg(feature = "redis")] mod standalone_redis { redis_cases!(super::on_redis::redis("STREAMS_REDIS_URL", false)); + conformance!(super::on_redis::redis("STREAMS_REDIS_URL", false); + deleting_an_owner_drops_every_chain_of_it, + ); } #[cfg(feature = "redis")] mod cluster_redis { redis_cases!(super::on_redis::redis("STREAMS_REDIS_CLUSTER_URL", true)); + // A Workflow id's chains hash to different slots, so the delete reaches every primary. + conformance!(super::on_redis::redis("STREAMS_REDIS_CLUSTER_URL", true); + deleting_an_owner_drops_every_chain_of_it, + ); } #[cfg(feature = "redis")] mod documented_acl_redis { redis_cases!(super::on_redis::redis_as_documented_user( "STREAMS_REDIS_URL", - false + false, + &[] )); + // The page names the one rule a delete needs beyond the others. + conformance!(super::on_redis::redis_as_documented_user("STREAMS_REDIS_URL", false, &["+scan"]); + deleting_an_owner_drops_every_chain_of_it, + ); } #[cfg(feature = "redis")] @@ -1268,6 +1284,14 @@ mod documented_acl_cluster_redis { // A cluster client asks the topology first, so the documented rules must allow it. redis_cases!(super::on_redis::redis_as_documented_user( "STREAMS_REDIS_CLUSTER_URL", - true + true, + &[] )); + conformance!(super::on_redis::redis_as_documented_user( + "STREAMS_REDIS_CLUSTER_URL", + true, + &["+scan"] + ); + deleting_an_owner_drops_every_chain_of_it, + ); } diff --git a/crates/streams/tests/redis_store.rs b/crates/streams/tests/redis_store.rs index de5601239e..a8011bff60 100644 --- a/crates/streams/tests/redis_store.rs +++ b/crates/streams/tests/redis_store.rs @@ -1224,3 +1224,218 @@ async fn the_documented_acl_is_the_python_sdks() { expected.push("+cluster|slots".to_string()); assert_eq!(rules(include_str!("../docs/redis.md")), expected); } + +fn delete_request(workflow_id: &str) -> temporalio_streams::proto::DeleteOwnerRequest { + temporalio_streams::proto::DeleteOwnerRequest { + namespace: "ns".to_string(), + owner_kind: temporalio_streams::proto::StreamOwnerKind::Workflow as i32, + workflow_id: workflow_id.to_string(), + } +} + +#[tokio::test] +async fn an_admin_deletes_one_workflows_stream_keys() { + let Some(mut setup) = setup().await else { + return; + }; + let chain = chain(); + let next = ChainId { + first_run_id: "run-2".to_string(), + ..chain.clone() + }; + let bystander = self::chain(); + // A log, its meta and a close flag on one chain, and a log on the next chain of the same + // Workflow id. + setup + .store + .append(append(&chain, "p", 1, 1, b"\x01")) + .await + .unwrap(); + setup.store.close_chain(&chain).await.unwrap(); + setup + .store + .append(append(&next, "p", 1, 1, b"\x01")) + .await + .unwrap(); + setup + .store + .append(append(&bystander, "p", 1, 1, b"\x01")) + .await + .unwrap(); + let pattern = format!("{}:{{ns:{}:*", setup.prefix, part(&chain.workflow_id)); + let before: Vec = redis::cmd("KEYS") + .arg(&pattern) + .query_async(&mut setup.raw) + .await + .unwrap(); + assert_eq!(before.len(), 5, "{before:?}"); + let deleted = setup + .store + .delete_owner(delete_request(&chain.workflow_id)) + .await + .unwrap(); + assert_eq!(deleted.deleted, 5); + let after: Vec = redis::cmd("KEYS") + .arg(&pattern) + .query_async(&mut setup.raw) + .await + .unwrap(); + assert!(after.is_empty(), "{after:?}"); + let left: usize = setup + .raw + .xlen(log_key(&setup.prefix, &bystander, "events")) + .await + .unwrap(); + assert_eq!(left, 1); + let again = setup + .store + .delete_owner(delete_request(&chain.workflow_id)) + .await + .unwrap(); + assert_eq!(again.deleted, 0); +} + +#[tokio::test] +async fn a_store_error_on_delete_arrives_as_a_stream_error() { + let Some(url) = url() else { + return; + }; + let faults = Faults::start(&url).await; + let setup = setup_at(&faults.url(), &url).await; + let chain = chain(); + setup + .store + .append(append(&chain, "p", 1, 1, b"\x01")) + .await + .unwrap(); + faults.next(LOSE_REQUEST); + // Some keys may already be gone when a later step fails. + let error = setup + .store + .delete_owner(delete_request(&chain.workflow_id)) + .await + .unwrap_err(); + assert_eq!(error.kind, StreamFailureKind::OutcomeUnknown, "{error}"); +} + +#[tokio::test] +async fn a_delete_without_scan_is_refused_under_the_documented_acl() { + let Some(url) = url() else { + return; + }; + let prefix = unique("acl-delete"); + let rules = |page: &str| -> Vec { + page.lines() + .find_map(|line| line.strip_prefix("ACL SETUSER ")) + .unwrap() + .split_whitespace() + .skip(1) + .map(str::to_string) + .collect() + }; + let name = unique("streams-acl"); + let mut admin = redis::Client::open(url.as_str()) + .unwrap() + .get_multiplexed_async_connection() + .await + .unwrap(); + let mut setuser = redis::cmd("ACL"); + setuser.arg("SETUSER").arg(&name); + for rule in rules(include_str!("../docs/redis.md")) { + match rule.as_str() { + "~temporal-streams:{my-ns:*" => setuser.arg(format!("~{}:{{ns:*", part(&prefix))), + ">secret" => setuser.arg(">pw"), + rule => setuser.arg(rule), + }; + } + let _: () = setuser.query_async(&mut admin).await.unwrap(); + let address = url.trim_start_matches("redis://"); + let mut options = RedisStoreOptions::new(format!("redis://{name}:pw@{address}")); + options.key_prefix = prefix; + let store = RedisStore::connect(options).await.unwrap(); + let error = store.delete_owner(delete_request("wf")).await.unwrap_err(); + let _: () = redis::cmd("ACL") + .arg("DELUSER") + .arg(&name) + .query_async(&mut admin) + .await + .unwrap(); + assert_eq!(error.kind, StreamFailureKind::Refused, "{error}"); + assert!(error.message.contains("NOPERM"), "{error}"); +} + +#[tokio::test] +async fn a_delete_reaches_every_primary_of_a_cluster() { + let Ok(url) = std::env::var("STREAMS_REDIS_CLUSTER_URL") else { + eprintln!("set STREAMS_REDIS_CLUSTER_URL to run the Redis cluster tests"); + return; + }; + let prefix = unique("cluster-delete"); + let mut options = RedisStoreOptions::new(url.clone()); + options.cluster = true; + options.key_prefix = prefix.clone(); + let store = RedisStore::connect(options).await.unwrap(); + let workflow_id = unique("wf"); + for run in 0..30 { + let chain = ChainId { + namespace: "ns".to_string(), + workflow_id: workflow_id.clone(), + first_run_id: format!("run-{run}"), + }; + store + .append(append(&chain, "p", 1, 1, b"\x01")) + .await + .unwrap(); + } + let mut seed = redis::Client::open(url.as_str()) + .unwrap() + .get_multiplexed_async_connection() + .await + .unwrap(); + let nodes: String = redis::cmd("CLUSTER") + .arg("NODES") + .query_async(&mut seed) + .await + .unwrap(); + let primaries: Vec = nodes + .lines() + .filter(|line| line.contains("master")) + .filter_map(|line| line.split_whitespace().nth(1)) + .map(|address| address.split('@').next().unwrap().to_string()) + .collect(); + let pattern = format!("{}:{{ns:{}:*", part(&prefix), part(&workflow_id)); + let mut held = Vec::new(); + for primary in &primaries { + let mut node = redis::Client::open(format!("redis://{primary}/0")) + .unwrap() + .get_multiplexed_async_connection() + .await + .unwrap(); + let keys: Vec = redis::cmd("KEYS") + .arg(&pattern) + .query_async(&mut node) + .await + .unwrap(); + held.push((node, keys.len())); + } + assert!( + held.iter().all(|(_, count)| *count > 0), + "the chains should land on every primary" + ); + let deleted = store + .delete_owner(delete_request(&workflow_id)) + .await + .unwrap(); + assert_eq!( + deleted.deleted as usize, + held.iter().map(|(_, count)| count).sum::() + ); + for (mut node, _) in held { + let keys: Vec = redis::cmd("KEYS") + .arg(&pattern) + .query_async(&mut node) + .await + .unwrap(); + assert!(keys.is_empty(), "{keys:?}"); + } +}