From 96053eafe81893552a5cb5c2996b6faabcc0f85e Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Wed, 14 Oct 2026 16:41:49 -0700 Subject: [PATCH 1/3] Wrote a notification entry per Redis append in the chain's slot. The log, the meta and the entry share a Cluster slot, so one script writes them and a reshard moves them together. A doorbell wakes a driver, and a sweep finds a lost ring. --- .../temporal/sdk/core/streams/streams.proto | 3 + crates/sdk-core/CHANGELOG.md | 5 + crates/streams/docs/redis.md | 18 +- crates/streams/src/lib.rs | 2 +- crates/streams/src/redis/conn.rs | 23 +- crates/streams/src/redis/keys.rs | 85 + crates/streams/src/redis/log.rs | 394 + crates/streams/src/redis/mod.rs | 66 +- crates/streams/src/redis/scripts.rs | 34 + crates/streams/src/redis/slots.rs | 67 + crates/streams/src/service.rs | 51 +- .../streams/testdata/redis/notify_slots.json | 16386 ++++++++++++++++ crates/streams/tests/conformance.rs | 75 +- crates/streams/tests/redis_store.rs | 63 + 14 files changed, 17243 insertions(+), 29 deletions(-) create mode 100644 crates/streams/src/redis/log.rs create mode 100644 crates/streams/src/redis/slots.rs create mode 100644 crates/streams/testdata/redis/notify_slots.json diff --git a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto index efa809cc82..67a2bd802f 100644 --- a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto +++ b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto @@ -100,6 +100,9 @@ message RedisStoreConfig { google.protobuf.Duration floor_lease = 7; // How long a hold lasts after its reader released it. Unset keeps one day. google.protobuf.Duration floor_release_grace = 8; + // With `StreamStoreConfig.notification_log`, how many doorbell streams the log rings, each one + // a driver blocks on. Zero uses 16. + uint32 notify_doorbells = 9; } // Streams in this process's memory, for tests and local development. Nothing survives the diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index 2417ecd4a2..c68427473c 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -53,6 +53,11 @@ relevant information. memory store keeps one, and a store without one refuses the setting. `NotifierClient` gains `notify_streams`, which Core's client sends as one `NotifyStreams` call per namespace, and the C bridge takes `NotifyStreams` as a raw call. +* Experimental: the Redis store keeps a notification log with `notification_log`. The append + script (hash `9fdc33e7`) also adds an entry to the stream of the chain's Cluster slot, + `:{}:n`, and Core rings one of `notify_doorbells` (16 when zero) doorbell + streams when that stream takes its first entry. `testdata/redis/notify_slots.json` pins the slot + tags for other SDKs. Drivers read as the consumer group `streams-driver`. * Experimental: the new `FlushNotifications` stream call waits up to 10 seconds for the notifications still out. A process calls it before it stops. * Experimental: a server without the stream notifier answers the first notifier call with diff --git a/crates/streams/docs/redis.md b/crates/streams/docs/redis.md index fb108954d7..8852afc473 100644 --- a/crates/streams/docs/redis.md +++ b/crates/streams/docs/redis.md @@ -41,6 +41,19 @@ doesn't read the floors hash, so its appends can trim records that a positions-o A topic that a positions-only reader reads must take its appends only through Core's store. Other providers on such a topic are unsupported. +**Notification log.** With `notification_log` on, the append script also adds an entry to the +notification stream of the chain's Cluster slot, `:{}:n`, where the tag is the one +`testdata/redis/notify_slots.json` pins for that slot. So the log, the meta and the entry share a +slot, one script writes them, and a reshard moves them together. A standalone server uses the +same keys. An entry holds `k`, the log key, and `p`, the append's last entry id. The stream keeps +about the newest 10,000 entries. A repeat of the newest batch adds none. The script answers the +stream's length too, and when it is 1, Core adds the slot to doorbell `:{}:bell`, apart from the script, so a lost ring is possible. Drivers read the slot +streams and the doorbells as the consumer group `streams-driver` (`XREADGROUP`, `XAUTOCLAIM`, +`XACK`, `XDEL`, `XGROUP CREATE`), and sweep with `XLEN` over each primary's slots from `CLUSTER +SLOTS`. The slot streams have no expiry, and a driver empties them. The logged append script has +its own hash. The plain one is unchanged, and an SDK's provider writes no entries. + **Retired producers.** A field stays until a write's trim finds it older than the retention, so a topic written by many short-lived producers holds one field per producer attempt of the last retention window, about 235 bytes each. `RetireProducer` deletes one attempt's field with @@ -73,7 +86,10 @@ ACL SETUSER streams-app on >secret resetkeys ~temporal-streams:{my-ns:* resetcha 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 +`+scan`. With the notification log on, an application also needs `~temporal-streams:{*}:n +%W~temporal-streams:{*}:bell`, since its appends write the slot streams. Those entries name +other namespaces' streams and positions, never records. A driver needs those keys with read and +write, and `+xreadgroup +xautoclaim +xack +xdel +xgroup|create +xlen +xadd +cluster|slots`. `+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 diff --git a/crates/streams/src/lib.rs b/crates/streams/src/lib.rs index 97455aaacf..b440c9a991 100644 --- a/crates/streams/src/lib.rs +++ b/crates/streams/src/lib.rs @@ -24,7 +24,7 @@ mod store; mod subscribe; #[cfg(feature = "redis")] -pub use self::redis::{RedisStore, RedisStoreOptions}; +pub use self::redis::{DEFAULT_NOTIFY_DOORBELLS, RedisStore, RedisStoreOptions}; pub use cursor::{BEGINNING, END, cursor_position, mint_cursor, stream_hash}; pub use error::{StreamError, StreamResult}; pub use handle::{OWNER_RECHECK, OWNER_START_GRACE, Streams, StreamsOptions}; diff --git a/crates/streams/src/redis/conn.rs b/crates/streams/src/redis/conn.rs index 29e6f88f2b..640f7cd953 100644 --- a/crates/streams/src/redis/conn.rs +++ b/crates/streams/src/redis/conn.rs @@ -120,6 +120,27 @@ impl Shared { } } + /// Runs `pipeline`, which holds `count` commands that all answer, on `node`, or on the server + /// when the store talks to one. + pub(crate) async fn query_pipeline_on( + &self, + node: Option<&Node>, + pipeline: Pipeline, + count: usize, + ) -> RedisResult { + match (self.clone(), node) { + (Shared::Cluster(mut conn), Some((host, port))) => { + let routing = SingleNodeRoutingInfo::ByAddress { + host: host.clone(), + port: *port, + }; + let replies = conn.route_pipeline(pipeline, 0, count, routing).await?; + redis::from_redis_value(Value::Array(replies)).map_err(RedisError::from) + } + (mut shared, _) => pipeline.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> { @@ -151,7 +172,7 @@ impl Shared { } /// The node a key lives on, for a store that talks to one. -type Node = (String, u16); +pub(crate) type Node = (String, u16); struct NodePool { info: ConnectionInfo, diff --git a/crates/streams/src/redis/keys.rs b/crates/streams/src/redis/keys.rs index eed9f65b82..5928da8488 100644 --- a/crates/streams/src/redis/keys.rs +++ b/crates/streams/src/redis/keys.rs @@ -11,9 +11,19 @@ //! :{::}:chain the close flag //! ``` //! +//! The notification log lives apart from the chains, one stream per Cluster slot and a few +//! doorbells, each under the hash tag of a slot (see `slots.rs`). A chain's appends go to the +//! stream of the chain's own slot, so the append script writes both: +//! +//! ```text +//! :{}:n slot s's notification entries +//! :{}:bell doorbell d, which slots s with s % doorbells == d ring +//! ``` +//! //! Each part, the prefix included, is percent-encoded, so a `:` or a brace in an id cannot make //! two streams share a key. +use super::{conn::slot, slots}; use crate::proto::ChainId; /// Percent-encodes every byte but the unreserved ones, as Python's `quote(text, safe="")` does. @@ -64,6 +74,56 @@ impl ChainKeys { pub(crate) fn floors(&self, topic: &str) -> String { format!("{}:floors", self.log(topic)) } + + /// The Cluster slot every key of the chain lands in. + pub(crate) fn slot(&self) -> u16 { + slot(&self.base) + } +} + +/// Slot `slot`'s notification stream. +pub(crate) fn notifications(prefix: &str, slot: u16) -> String { + format!("{}:{{{}}}:n", part(prefix), slots::tag(slot)) +} + +/// Doorbell `doorbell`'s stream. +pub(crate) fn doorbell(prefix: &str, doorbell: u32) -> String { + format!("{}:{{{}}}:bell", part(prefix), slots::tag(doorbell as u16)) +} + +/// The chain and topic of a log key that a notification entry names, or `None` when the key is +/// not a log key under `prefix`. +pub(crate) fn chain_of_log(prefix: &str, key: &str) -> Option<(ChainId, String)> { + let rest = key.strip_prefix(&format!("{}:{{", part(prefix)))?; + let (tag, topic) = rest.split_once("}:t:")?; + let mut parts = tag.split(':'); + let (namespace, workflow_id, first_run_id) = (parts.next()?, parts.next()?, parts.next()?); + if parts.next().is_some() { + return None; + } + Some(( + ChainId { + namespace: unpart(namespace)?, + workflow_id: unpart(workflow_id)?, + first_run_id: unpart(first_run_id)?, + }, + unpart(topic)?, + )) +} + +/// Undoes [part]. +fn unpart(text: &str) -> Option { + let mut bytes = Vec::with_capacity(text.len()); + let mut input = text.bytes(); + while let Some(byte) = input.next() { + if byte == b'%' { + let hex = [input.next()?, input.next()?]; + bytes.push(u8::from_str_radix(std::str::from_utf8(&hex).ok()?, 16).ok()?); + } else { + bytes.push(byte); + } + } + String::from_utf8(bytes).ok() } /// The meta field that holds one producer attempt's newest batch. @@ -115,6 +175,31 @@ mod tests { } } + #[test] + fn a_chains_notification_stream_shares_its_slot() { + let cases: Vec = + serde_json::from_str(include_str!("../../testdata/redis/keys.json")).unwrap(); + for case in &cases { + let chain = ChainId { + namespace: text(case, "namespace").to_string(), + workflow_id: text(case, "workflow_id").to_string(), + first_run_id: text(case, "first_run_id").to_string(), + }; + let prefix = text(case, "prefix"); + let keys = ChainKeys::new(prefix, &chain); + let topic = text(case, "topic"); + let notifications = notifications(prefix, keys.slot()); + assert_eq!(slot(¬ifications), slot(&keys.log(topic))); + assert_eq!(slot(¬ifications), slot(&keys.meta(topic))); + assert_eq!( + chain_of_log(prefix, &keys.log(topic)), + Some((chain, topic.to_string())) + ); + } + assert_eq!(chain_of_log("p", "q:{ns:wf:run}:t:out"), None); + assert_eq!(chain_of_log("p", "p:{ns:wf}:t:out"), None); + } + #[test] fn session_fields_match_the_python_sdk() { let cases: Vec = diff --git a/crates/streams/src/redis/log.rs b/crates/streams/src/redis/log.rs new file mode 100644 index 0000000000..e5061786c4 --- /dev/null +++ b/crates/streams/src/redis/log.rs @@ -0,0 +1,394 @@ +//! The notification log on Redis: one stream per Cluster slot, read by drivers as one consumer +//! group, and a few doorbell streams that say which slot took its first entry. + +use super::{ + conn::{BlockingPool, Node, Shared}, + errors::{Call, Mapped}, + keys::{chain_of_log, doorbell, notifications}, + slots::SLOTS, +}; +use crate::{NotificationEntry, NotificationLog, StreamError, StreamResult}; +use redis::{RedisError, Value}; +use std::{collections::BTreeSet, time::Duration}; +use tokio::time::Instant; + +/// The consumer group every driver reads as. +pub(crate) const GROUP: &str = "streams-driver"; +/// The entry fields: the log key the append wrote, and the append's last position. +const KEY_FIELD: &str = "k"; +const POSITION_FIELD: &str = "p"; +/// The doorbell entry field that names the slot. +const SLOT_FIELD: &str = "s"; +/// A doorbell only ever needs its newest rings. +const DOORBELL_LENGTH: u64 = 1000; + +/// The notification log of one [super::RedisStore]. +pub(crate) struct RedisLog { + shared: Shared, + blocking: BlockingPool, + prefix: String, + doorbells: u32, +} + +impl RedisLog { + pub(crate) fn new( + shared: Shared, + blocking: BlockingPool, + prefix: String, + doorbells: u32, + ) -> Self { + Self { + shared, + blocking, + prefix, + doorbells: doorbells.clamp(1, SLOTS as u32), + } + } + + /// Rings the doorbell of `slot`, after an append made its entry the slot's only one. It runs + /// apart from the append, so a ring can be lost. The sweep finds the entry then. + pub(crate) async fn ring(&self, slot: u16) { + let rung: StreamResult = redis::cmd("XADD") + .arg(doorbell(&self.prefix, u32::from(slot) % self.doorbells)) + .arg("MAXLEN") + .arg("~") + .arg(DOORBELL_LENGTH) + .arg("*") + .arg(SLOT_FIELD) + .arg(slot) + .query_async(&mut self.shared.clone()) + .await + .mapped(Call::Write); + if let Err(error) = rung { + tracing::debug!(slot, %error, "Could not ring a notification doorbell"); + } + } + + /// Makes the driver group on `key`. Answers `false` when the key doesn't exist and + /// `create_key` is off, since a slot's stream that never took an entry needs no group. + async fn create_group(&self, key: &str, create_key: bool) -> StreamResult { + // From the start, so entries and rings from before the first driver count. + let mut create = redis::cmd("XGROUP"); + create.arg("CREATE").arg(key).arg(GROUP).arg("0"); + if create_key { + create.arg("MKSTREAM"); + } + match create.query_async::<()>(&mut self.shared.clone()).await { + Ok(()) => Ok(true), + Err(error) if error.code() == Some("BUSYGROUP") => Ok(true), + Err(error) + if !create_key && error.to_string().contains("requires the key to exist") => + { + Ok(false) + } + Err(error) => Err(error).mapped(Call::Write), + } + } + + /// Reads `key` for `consumer` from `from`: `0` for the consumer's own unacked entries, `>` + /// for new ones. Creates the group once when it is missing. + async fn read_group( + &self, + key: &str, + consumer: &str, + from: &str, + count: usize, + ) -> StreamResult)>> { + for first in [true, false] { + let read: Result = redis::cmd("XREADGROUP") + .arg("GROUP") + .arg(GROUP) + .arg(consumer) + .arg("COUNT") + .arg(count.max(1)) + .arg("STREAMS") + .arg(key) + .arg(from) + .query_async(&mut self.shared.clone()) + .await; + match read { + Ok(reply) => return Ok(stream_entries(reply)), + Err(error) if first && error.code() == Some("NOGROUP") => { + if !self.create_group(key, false).await? { + return Ok(vec![]); + } + } + Err(error) => return Err(error).mapped(Call::Read), + } + } + Ok(vec![]) + } + + /// The slots each primary serves: the whole range for one server. + async fn primaries(&self) -> StreamResult, Vec)>> { + if let Shared::Single(_) = self.shared { + return Ok(vec![(None, (0..SLOTS as u16).collect())]); + } + let reply: Value = redis::cmd("CLUSTER") + .arg("SLOTS") + .query_async(&mut self.shared.clone()) + .await + .mapped(Call::Read)?; + let mut primaries: Vec<(Option, Vec)> = vec![]; + let Value::Array(ranges) = reply else { + return Err(StreamError::storage(format!( + "Redis answered CLUSTER SLOTS with {reply:?}" + ))); + }; + for range in ranges { + let Value::Array(range) = range else { continue }; + let (Some(Value::Int(start)), Some(Value::Int(end)), Some(Value::Array(node))) = + (range.first(), range.get(1), range.get(2)) + else { + continue; + }; + let host = match node.first() { + Some(Value::BulkString(host)) => String::from_utf8_lossy(host).into_owned(), + Some(Value::SimpleString(host)) => host.clone(), + _ => continue, + }; + let Some(Value::Int(port)) = node.get(1) else { + continue; + }; + let node = Some((host, *port as u16)); + let slots = (*start as u16..=*end as u16).collect::>(); + match primaries.iter_mut().find(|(known, _)| *known == node) { + Some((_, known)) => known.extend(slots), + None => primaries.push((node, slots)), + } + } + Ok(primaries) + } +} + +#[async_trait::async_trait] +impl NotificationLog for RedisLog { + fn partitions(&self) -> u32 { + SLOTS as u32 + } + + fn doorbells(&self) -> u32 { + self.doorbells + } + + async fn wait_doorbell( + &self, + doorbell_index: u32, + consumer: &str, + block: Duration, + ) -> StreamResult> { + let key = doorbell(&self.prefix, doorbell_index); + let deadline = Instant::now() + block; + let mut read = redis::cmd("XREADGROUP"); + read.arg("GROUP") + .arg(GROUP) + .arg(consumer) + .arg("COUNT") + .arg(DOORBELL_LENGTH) + .arg("BLOCK") + .arg(u64::try_from(block.as_millis()).unwrap_or(u64::MAX).max(1)) + .arg("STREAMS") + .arg(&key) + .arg(">"); + let mut reply = None; + for first in [true, false] { + match self.blocking.query(&key, &read, deadline).await { + Ok(answer) => { + reply = answer; + break; + } + Err(error) if first && error.code() == Some("NOGROUP") => { + self.create_group(&key, true).await?; + } + Err(error) => return Err(error).mapped(Call::Read), + } + } + let rings = reply.map(stream_entries).unwrap_or_default(); + if rings.is_empty() { + return Ok(vec![]); + } + // A ring is a hint, so it goes as soon as it is read. + let ids: Vec<&str> = rings.iter().map(|(id, _)| id.as_str()).collect(); + let mut pipeline = redis::pipe(); + pipeline + .cmd("XACK") + .arg(&key) + .arg(GROUP) + .arg(&ids) + .ignore() + .cmd("XDEL") + .arg(&key) + .arg(&ids) + .ignore(); + pipeline + .query_async::<()>(&mut self.shared.clone()) + .await + .mapped(Call::Write)?; + let slots: BTreeSet = rings + .iter() + .filter_map(|(_, fields)| text_field(fields, SLOT_FIELD)?.parse().ok()) + .collect(); + Ok(slots.into_iter().collect()) + } + + async fn take( + &self, + partition: u32, + consumer: &str, + claim_idle: Duration, + max: usize, + ) -> StreamResult> { + let key = notifications(&self.prefix, partition as u16); + // Claimed entries join this consumer's own unacked ones, which the read from `0` gives. + let claim: Result = redis::cmd("XAUTOCLAIM") + .arg(&key) + .arg(GROUP) + .arg(consumer) + .arg(u64::try_from(claim_idle.as_millis()).unwrap_or(u64::MAX)) + .arg("0-0") + .arg("COUNT") + .arg(max.max(1)) + .arg("JUSTID") + .query_async(&mut self.shared.clone()) + .await; + match claim { + Ok(_) => {} + Err(error) if error.code() == Some("NOGROUP") => { + if !self.create_group(&key, false).await? { + return Ok(vec![]); + } + } + Err(error) => return Err(error).mapped(Call::Write), + } + let mut entries = self.read_group(&key, consumer, "0", max).await?; + if entries.len() < max { + entries.extend( + self.read_group(&key, consumer, ">", max - entries.len()) + .await?, + ); + } + let mut taken = vec![]; + let mut broken = vec![]; + for (id, fields) in entries { + let parsed = text_field(&fields, KEY_FIELD) + .and_then(|key| chain_of_log(&self.prefix, &key)) + .zip(text_field(&fields, POSITION_FIELD)); + match parsed { + Some(((chain, topic), position)) => taken.push(NotificationEntry { + id, + chain, + topic, + position, + }), + None => broken.push(id), + } + } + // An entry that names no stream can never be delivered. + if !broken.is_empty() { + self.ack(partition, &broken).await?; + } + Ok(taken) + } + + async fn ack(&self, partition: u32, ids: &[String]) -> StreamResult<()> { + if ids.is_empty() { + return Ok(()); + } + let key = notifications(&self.prefix, partition as u16); + let mut pipeline = redis::pipe(); + pipeline + .cmd("XACK") + .arg(&key) + .arg(GROUP) + .arg(ids) + .ignore() + .cmd("XDEL") + .arg(&key) + .arg(ids) + .ignore(); + pipeline + .query_async::<()>(&mut self.shared.clone()) + .await + .mapped(Call::Write) + } + + async fn busy(&self) -> StreamResult> { + let mut busy = vec![]; + for (node, slots) in self.primaries().await? { + let mut pipeline = redis::pipe(); + for &slot in &slots { + pipeline.cmd("XLEN").arg(notifications(&self.prefix, slot)); + } + let lengths: Vec = self + .shared + .query_pipeline_on(node.as_ref(), pipeline, slots.len()) + .await + .mapped(Call::Read)?; + busy.extend( + slots + .iter() + .zip(lengths) + .filter(|(_, length)| *length > 0) + .map(|(slot, _)| u32::from(*slot)), + ); + } + Ok(busy) + } +} + +/// The entries of an `XREADGROUP` reply on one stream, as their ids and fields. An entry an +/// `XDEL` dropped while it was pending has no fields. +fn stream_entries(reply: Value) -> Vec<(String, Vec<(String, Value)>)> { + let streams = match reply { + Value::Array(streams) => streams + .into_iter() + .filter_map(|stream| match stream { + Value::Array(mut pair) if pair.len() == 2 => pair.pop(), + _ => None, + }) + .collect::>(), + Value::Map(streams) => streams.into_iter().map(|(_, entries)| entries).collect(), + _ => vec![], + }; + streams + .into_iter() + .flat_map(|entries| match entries { + Value::Array(entries) => entries, + _ => vec![], + }) + .filter_map(|entry| { + let Value::Array(mut parts) = entry else { + return None; + }; + let fields = parts.pop()?; + let id = text(parts.pop()?)?; + let fields = match fields { + Value::Array(flat) => flat + .chunks(2) + .filter_map(|pair| Some((text(pair.first()?.clone())?, pair.get(1)?.clone()))) + .collect(), + Value::Map(pairs) => pairs + .into_iter() + .filter_map(|(name, value)| Some((text(name)?, value))) + .collect(), + _ => vec![], + }; + Some((id, fields)) + }) + .collect() +} + +fn text(value: Value) -> Option { + match value { + Value::BulkString(bytes) => String::from_utf8(bytes).ok(), + Value::SimpleString(text) => Some(text), + _ => None, + } +} + +fn text_field(fields: &[(String, Value)], name: &str) -> Option { + fields + .iter() + .find(|(field, _)| field == name) + .and_then(|(_, value)| text(value.clone())) +} diff --git a/crates/streams/src/redis/mod.rs b/crates/streams/src/redis/mod.rs index 981e988da7..ca6b6ffcc6 100644 --- a/crates/streams/src/redis/mod.rs +++ b/crates/streams/src/redis/mod.rs @@ -3,10 +3,12 @@ mod conn; mod errors; mod keys; +mod log; mod scripts; +mod slots; use crate::{ - StreamError, StreamResult, StreamStore, + NotificationLog, StreamError, StreamResult, StreamStore, proto::{ ChainId, DeleteOwnerRequest, DeleteOwnerResponse, StoreAppendRequest, StoreAppendResponse, StoreLatestRequest, StoreLatestResponse, StoreReadRequest, StoreReadResponse, StoredClose, @@ -18,7 +20,7 @@ use errors::{Call, Mapped}; use keys::{ChainKeys, owner_pattern, session_field}; use prost::Message; use redis::{Cmd, Value, streams::StreamReadReply}; -use std::{collections::HashMap, time::Duration}; +use std::{collections::HashMap, sync::Arc, time::Duration}; use tokio::time::Instant; const NAME: &str = "redis"; @@ -51,6 +53,10 @@ pub struct RedisStoreOptions { pub floor_lease: Option, /// How long a hold lasts after its reader released it. pub floor_release_grace: Duration, + /// Each append also writes an entry to its slot's notification stream, which a driver reads. + pub notification_log: bool, + /// How many doorbells the notification log rings, each a stream that a driver blocks on. + pub notify_doorbells: u32, } impl RedisStoreOptions { @@ -65,10 +71,15 @@ impl RedisStoreOptions { response_timeout: Duration::from_secs(5), floor_lease: None, floor_release_grace: Duration::from_secs(24 * 60 * 60), + notification_log: false, + notify_doorbells: DEFAULT_NOTIFY_DOORBELLS, } } } +/// How many doorbells a notification log rings when the options don't say. +pub const DEFAULT_NOTIFY_DOORBELLS: u32 = 16; + /// A [StreamStore] on Redis. pub struct RedisStore { shared: Shared, @@ -78,6 +89,7 @@ pub struct RedisStore { grace_ms: u64, floor_lease_ms: u64, floor_release_grace_ms: u64, + log: Option>, } impl std::fmt::Debug for RedisStore { @@ -119,7 +131,26 @@ impl RedisStore { options.response_timeout, ) .mapped(Call::Read)?; + let log = if options.notification_log { + // Doorbell reads block, so they get connections of their own. + let doorbell_reads = BlockingPool::new( + seed, + options.notify_doorbells.max(1) as usize, + options.cluster, + options.response_timeout, + ) + .mapped(Call::Read)?; + Some(Arc::new(log::RedisLog::new( + shared.clone(), + doorbell_reads, + options.key_prefix.clone(), + options.notify_doorbells, + ))) + } else { + None + }; Ok(Self { + log, shared, blocking, prefix: options.key_prefix, @@ -446,6 +477,10 @@ impl StreamStore for RedisStore { NAME } + fn notification_log(&self) -> Option> { + self.log.clone().map(|log| log as Arc) + } + async fn append(&self, request: StoreAppendRequest) -> StreamResult { if request.records.is_empty() { return Err(StreamError::refused("an append needs at least one record")); @@ -454,12 +489,19 @@ impl StreamStore for RedisStore { return Err(StreamError::refused("an append needs the batch's digest")); } let keys = self.keys(request.chain.as_ref())?; - let mut script = scripts::APPEND_SCRIPT.prepare_invoke(); + let mut script = match self.log { + Some(_) => scripts::APPEND_LOGGED_SCRIPT.prepare_invoke(), + None => scripts::APPEND_SCRIPT.prepare_invoke(), + }; script .key(keys.log(&request.topic)) .key(keys.meta(&request.topic)) .key(keys.chain()) - .key(keys.floors(&request.topic)) + .key(keys.floors(&request.topic)); + if self.log.is_some() { + script.key(keys::notifications(&self.prefix, keys.slot())); + } + script .arg(self.retention_ms) .arg(self.grace_ms) .arg(session_field(&request.producer_id, request.attempt)) @@ -468,16 +510,28 @@ impl StreamStore for RedisStore { for record in &request.records { script.arg(record.as_slice()); } - let appended: StreamResult<(String, String)> = script + let appended: StreamResult> = script .invoke_async(&mut self.shared.clone()) .await .mapped(Call::Append); - let (first_position, last_position) = match appended { + let answer = match appended { Err(error) if error.kind == StreamFailureKind::Closed => { return Err(self.with_closed_by(error, &keys, &request.topic).await); } appended => appended?, }; + let [first_position, last_position, rest @ ..] = answer.as_slice() else { + return Err(StreamError::storage(format!( + "Redis answered the append with {answer:?}" + ))); + }; + // The entry made a slot's stream non-empty, so a driver may be asleep on its doorbell. + if let (Some(log), [length]) = (&self.log, rest) + && length == "1" + { + log.ring(keys.slot()).await; + } + let (first_position, last_position) = (first_position.clone(), last_position.clone()); Ok(StoreAppendResponse { first_position, last_position, diff --git a/crates/streams/src/redis/scripts.rs b/crates/streams/src/redis/scripts.rs index 51bb43ecf3..09cade9d37 100644 --- a/crates/streams/src/redis/scripts.rs +++ b/crates/streams/src/redis/scripts.rs @@ -15,6 +15,27 @@ pub(crate) const APPEND: &str = concat!( ); pub(crate) static APPEND_SCRIPT: LazyLock