[Brook] Checked Redis server settings and deleted an owner's streams.

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",

crates/streams/docs/redis.md

@@ -0,0 +1,46 @@
+# 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.
+
+**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
+pressure, losing records and their dedupe state.
+
+**ACL.** Every key of a namespace's streams starts with `<prefix>:{<namespace>:`, 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.

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"

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,
@@ -388,6 +389,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<DeleteOwnerResponse> {
+        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. A producer attempt stays on the chain it began on, and
     /// is refused once a new chain took the Workflow id.
     async fn producer_chain(&self, key: ProducerKey) -> StreamResult<ChainId> {
@@ -1476,6 +1498,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";

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, SingleNodeRoutingInfo},
 };
 use std::{
     collections::HashMap,
@@ -76,6 +77,77 @@ impl Shared {
             client.get_connection_manager_with_config(config).await?,
         ))
     }
+
+    /// The primaries a scan must visit: the server itself, or each primary of a cluster.
+    pub(crate) async fn primaries(&self) -> RedisResult<Vec<Option<Node>>> {
+        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<String>)> {
+        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<Vec<(String, Value)>> {
+        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.

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:*");
+    }
 }