[Jetty] Wrote a Redis notification entry per append in the chain slot.

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

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,
+  `<prefix>:{<slot tag>}: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

crates/streams/docs/redis.md

@@ -43,6 +43,21 @@ 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, `<prefix>:{<tag>}: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 `d = s % doorbells`, at
+`<prefix>:{<tag of slot d * 16384 / doorbells>}:bell`, apart from the script, so a lost ring is
+possible. The doorbells spread over the slot range, so a cluster's primaries share them, and an
+application and its drivers must agree on the number of doorbells. 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
@@ -75,8 +90,11 @@ 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
-`maxmemory-policy`, and the store goes on when a server refuses it.
+`+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
 `STREAMS_REDIS_CLUSTER_URL` a cluster seed. `scripts/redis-test-servers.sh` starts both. The

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};

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<T: redis::FromRedisValue>(
+        &self,
+        node: Option<&Node>,
+        pipeline: Pipeline,
+        count: usize,
+    ) -> RedisResult<T> {
+        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<Vec<(String, Value)>> {
@@ -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,

crates/streams/src/redis/keys.rs

@@ -11,9 +11,20 @@
 //! <prefix>:{<namespace>:<workflow id>:<first run id>}: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
+//! <prefix>:{<tag of slot s>}:n      slot s's notification entries
+//! <prefix>:{<tag of slot d * 16384 / doorbells>}: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 +75,58 @@ 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, of `doorbells`. The doorbells spread over the slot range, so
+/// a cluster's primaries share them.
+pub(crate) fn doorbell(prefix: &str, doorbell: u32, doorbells: u32) -> String {
+    let slot = u64::from(doorbell) * slots::SLOTS as u64 / u64::from(doorbells.max(1));
+    format!("{}:{{{}}}:bell", part(prefix), slots::tag(slot 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<String> {
+    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 +178,31 @@ mod tests {
         }
     }
 
+    #[test]
+    fn a_chains_notification_stream_shares_its_slot() {
+        let cases: Vec<Value> =
+            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(&notifications), slot(&keys.log(topic)));
+            assert_eq!(slot(&notifications), 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<Value> =
@@ -147,4 +235,12 @@ mod tests {
         // 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:*");
     }
+
+    #[test]
+    fn the_doorbells_spread_over_the_slot_range() {
+        let slots: Vec<u16> = (0..16)
+            .map(|d| super::super::conn::slot(&doorbell("p", d, 16)))
+            .collect();
+        assert_eq!(slots, (0..16).map(|d| d * 1024).collect::<Vec<u16>>());
+    }
 }