diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index cd75732..9d3f702 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -89,3 +89,12 @@ jobs: env: STREAMS_REQUIRE: redis STREAMS_REDIS_CLUSTER_ADMIN: "1" + + optional-restore: + name: The optional layers' restore commits copy 4aebbd4a + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + with: + fetch-depth: 0 + - run: scripts/check-optional-restore.py diff --git a/crates/streams/docs/managed-redis.md b/crates/streams/docs/managed-redis.md index feff80f..6e765a3 100644 --- a/crates/streams/docs/managed-redis.md +++ b/crates/streams/docs/managed-redis.md @@ -40,18 +40,23 @@ MemoryDB is the AWS production target for the Redis store. MemoryDB takes users through its API, not `ACL SETUSER`. An access string has no password, no `reset` rules and no `config` rule (https://docs.aws.amazon.com/memorydb/latest/devguide/clusters.acls.html). This is the access -string for one namespace's applications, with owner deletes. The +string for one namespace's applications, with the notification log on and owner deletes. The conformance suite runs as a user with exactly these rules on local Redis. ```text -on ~temporal-streams:{my-ns:* -@all +evalsha +eval +script|load +multi +exec +xadd +xread +xrevrange +xrange +xtrim +xlen +hget +hset +hgetall +hlen +hincrby +hdel +hscan +sadd +smembers +rpush +lrange +exists +del +unlink +scan +pexpire +pttl +time +info +ping +hello +client|setinfo +cluster|slots +on ~temporal-streams:{my-ns:* ~temporal-streams:{*}:n ~temporal-streams:{*}:bell -@all +evalsha +eval +script|load +multi +exec +xadd +xread +xrevrange +xrange +xtrim +xlen +hget +hset +hgetall +hlen +hincrby +hdel +hscan +sadd +smembers +rpush +lrange +exists +del +unlink +scan +pexpire +pttl +time +info +ping +hello +client|setinfo +cluster|slots ``` -It differs from the `ACL SETUSER` rules in [redis.md](redis.md) in three ways. +It differs from the `ACL SETUSER` rules in [redis.md](redis.md) in four ways. - It has no `>secret`. The password goes in the user's authentication mode. - It has no `resetkeys` and no `resetchannels`. A new user has no keys and no channels. - It has no `+config|get`. The store goes on without the eviction check. +- The doorbell keys use `~`, not `%W~`. We do not know if MemoryDB takes key selectors with + permissions. `~` gives read access too, which the application does not need. + +Without the notification log, drop the two `{*}` key patterns. A driver also needs +`+xreadgroup +xautoclaim +xack +xdel +xgroup|create` on the `{*}` keys. Create the user and its ACL with the AWS CLI. Then give the ACL to the cluster. @@ -72,8 +77,8 @@ aws memorydb create-acl --acl-name streams --user-names streams-app - **Serverless.** It supports `EVAL`, `EVALSHA`, `SCRIPT LOAD`, the stream and consumer group commands the store uses, `HSCAN`, `SCAN`, `UNLINK` and `TIME`. `CLUSTER SLOTS` answers one shard with every slot - (https://docs.aws.amazon.com/AmazonElastiCache/latest/dg/SupportedCommands.html). We did not - run the suite on it. + (https://docs.aws.amazon.com/AmazonElastiCache/latest/dg/SupportedCommands.html). The driver + then sweeps every slot on one node. We did not run the suite on it. - **Access.** `CONFIG` and `ACL SETUSER` are restricted. ElastiCache users take an access string through the API, so use the MemoryDB access string above. diff --git a/crates/streams/docs/redis.md b/crates/streams/docs/redis.md index 48e2e31..cd390ef 100644 --- a/crates/streams/docs/redis.md +++ b/crates/streams/docs/redis.md @@ -43,6 +43,24 @@ doesn't read the floors hash, so its appends can trim records that a positions-o topic that a positions-only reader reads must take its appends only through this crate'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, also ones a driver took and has not acked. So a driver that falls more +than 10,000 appends behind on a slot loses the oldest entries, and a quiet stream that shares the +slot with a busy one can lose its only entry. Its readers then wake on their recheck. A repeat of +the newest batch adds none. The script answers the stream's length too, and when it is 1, the store +adds the slot to doorbell `d = s % doorbells`, at +`:{}: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 @@ -94,8 +112,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. **Plain mode.** `RedisStoreOptions::plain` runs the store with no Lua. An append is a `MULTI` of `XADD`s that trim by age, a read is `XREAD` alone, and no meta and no close flag is @@ -103,21 +124,25 @@ kept. So the store declares neither producer state nor a kept last position, and their fallbacks: readers drop copies and end at the first `CLOSE` record, and a read decides expiry from the record at its cursor. It exists so tests run those fallbacks on a real store. Listing a chain's topics scans every primary, as a delete does, and needs `+scan`. With no script -there is no floor hold, so plain mode declares no holds. +there is no notification log and no floor hold, so plain mode declares neither holds nor a +change feed, and a config that asks for a log is refused. **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 -cluster with one replica per primary. The conformance suite runs on each, once more as a user -with exactly the ACL above, and once in plain mode with the cases of the fallbacks. -`tests/fallbacks.rs` adds faults on plain Redis too. The cross-implementation test also needs -`STREAMS_BROOK_PY`, a Python SDK checkout with its own Redis provider. When `STREAMS_REQUIRE` -names `redis`, a Redis test with no URL set fails and does not skip. +cluster with one replica per primary. `STREAMS_REDIS_CLUSTER_ADMIN=1` lets the driver tests move +a slot and fail a primary over, so set it only for a cluster of the tests' own. The +conformance suite runs on each, once more as a user with exactly the ACL above, and once in +plain mode with the cases of the fallbacks. `tests/fallbacks.rs` adds faults on plain Redis +too. The cross-implementation test also needs `STREAMS_BROOK_PY`, a Python SDK checkout with its +own Redis provider. When `STREAMS_REQUIRE` names `redis`, a Redis test with no URL set fails and +does not skip. **Managed Redis.** [managed-redis.md](managed-redis.md) covers Amazon MemoryDB and ElastiCache, with the access string that those services take in place of `ACL SETUSER`. **Kvrocks lane.** `STREAMS_KVROCKS_URL` names an Apache Kvrocks server, which keeps its data on disk. The script starts one when it finds a `kvrocks` binary or Docker. The conformance suite runs -on it standalone, with the store cases and owner deletes. This lane is a test only. We make no -support claim for Kvrocks. Kvrocks has no ACL users and runs no Redis Cluster, so those lanes do -not run. +on it standalone, with the store cases, owner deletes and the notification log. This lane is a +test only. We make no support claim for Kvrocks. Its doorbell case does not run, since Kvrocks +answers an empty `XREADGROUP` where Redis fails with `NOGROUP`, so a driver never makes the +doorbell's group. Kvrocks has no ACL users and runs no Redis Cluster, so those lanes do not run. diff --git a/crates/streams/protos/temporal/sdk/core/streams/streams.proto b/crates/streams/protos/temporal/sdk/core/streams/streams.proto index b46220e..189f5a8 100644 --- a/crates/streams/protos/temporal/sdk/core/streams/streams.proto +++ b/crates/streams/protos/temporal/sdk/core/streams/streams.proto @@ -54,6 +54,10 @@ service StreamService { // unsupported when the config turns no notifier client on. rpc DescribeStreamNotifier(temporal.api.workflowservice.v1.DescribeStreamNotifierRequest) returns (temporal.api.workflowservice.v1.DescribeStreamNotifierResponse); + // Runs a stream driver over the store's notification log until the caller cancels the call. + // It answers only when the server takes no stream notifications, or when the store keeps no + // log. Each driver of one store needs its own consumer name. + rpc RunStreamDriver(RunStreamDriverRequest) returns (RunStreamDriverResponse); // Reads again the exact range an earlier read answered, for a positions-only Workflow reader on // replay, and checks it against that read's digest. Waits for nothing and checks no owner. rpc ReadRange(ReadRangeRequest) returns (ReadResponse); @@ -104,6 +108,10 @@ message StreamStoreConfig { // owner does. After it the read fails as NOT_FOUND. Unset waits 60 seconds, and zero fails at // once. google.protobuf.Duration owner_start_grace = 6; + // Each append writes a notification entry in the same store step, and producers send no + // notifications. A `StreamDriver` reads the entries and notifies the server, so readers wake + // only while one runs. Off by default. A store without a notification log refuses it. + bool notification_log = 7; } // Streams in a Redis or Redis Cluster that the service connects to itself. @@ -127,6 +135,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; // Runs with no Lua, so the store keeps no producer state and no last position, and the service // takes their fallbacks. For tests of the fallbacks on a real store. bool plain = 10; @@ -615,6 +626,21 @@ message WaitForNotificationResponse { temporal.api.failure.v1.Failure close_failure = 6; } +message RunStreamDriverRequest { + // The driver's name in the log's consumer group, unique among the store's drivers. + string consumer = 1; + // How often the driver looks at every slot that holds entries. Unset uses 30 seconds. + google.protobuf.Duration sweep_interval = 2; + // How long another driver's unacked entries wait before this one takes them. Unset uses 60 + // seconds. + google.protobuf.Duration claim_idle = 3; + // The most entries one read takes, and so the most streams one notify call names. Zero uses + // 100. + uint32 batch = 4; +} + +message RunStreamDriverResponse {} + message DeleteOwnerRequest { string namespace = 1; StreamOwnerKind owner_kind = 2; diff --git a/crates/streams/scripts/redis-test-servers.sh b/crates/streams/scripts/redis-test-servers.sh index 1d244c6..ed7a326 100755 --- a/crates/streams/scripts/redis-test-servers.sh +++ b/crates/streams/scripts/redis-test-servers.sh @@ -1,6 +1,6 @@ #!/usr/bin/env bash # Starts the Redis servers the stream store tests run against: one standalone server, and a -# cluster of three primaries with one replica each. +# cluster of three primaries with one replica each, so the driver tests can fail a primary over. # 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_CLUSTER_BASE moves # the six cluster ports. diff --git a/crates/streams/src/driver.rs b/crates/streams/src/driver.rs new file mode 100644 index 0000000..f2d1e34 --- /dev/null +++ b/crates/streams/src/driver.rs @@ -0,0 +1,912 @@ +//! The stream driver: reads a store's notification log and tells the server's stream notifiers. +//! +//! A driver blocks on each doorbell, drains each slot a doorbell names, and sweeps every slot +//! that holds entries now and then, since a ring can be lost. It folds a drain's entries to the +//! newest counter of each stream, sends them in one call per namespace, and acks them only once +//! the call landed. A failed call leaves them pending, so this driver or another one sends them +//! again. Delivery is at least once, and a notifier ignores a counter it has seen. A call the +//! server refuses goes again one stream at a time, so one bad entry holds back no other stream. +//! The log keeps a capped backlog. When drivers fall behind by more than the cap, the oldest +//! entries go before any driver takes them, and those readers wake at the server's next Recheck. + +use crate::{ + NotificationEntry, NotificationLog, NotifierClient, NotifyError, StreamError, + StreamNotification, StreamResult, StreamStore, chain_owner, mint_cursor, progress_counter, + proto::ChainId, stream_hash, +}; +use std::{ + collections::{BTreeMap, HashMap}, + sync::{ + Arc, Mutex, + atomic::{AtomicBool, Ordering}, + }, + time::Duration, +}; +use tokio::task::JoinSet; + +/// How a driver reads the log. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct StreamDriverOptions { + /// The driver's name in the log's consumer group. Each driver needs its own, and the default + /// is the host name and the process id. + pub consumer: String, + /// How often the driver looks at every slot that holds entries. + pub sweep_interval: Duration, + /// How long another driver's unacked entries wait before this one takes them. + pub claim_idle: Duration, + /// How long one doorbell read blocks. + pub doorbell_wait: Duration, + /// The most entries one take reads, and so the most streams one notify call names. + pub batch: usize, +} + +impl Default for StreamDriverOptions { + fn default() -> Self { + Self { + consumer: format!( + "{}-{}", + gethostname::gethostname().to_string_lossy(), + std::process::id() + ), + sweep_interval: Duration::from_secs(30), + claim_idle: Duration::from_secs(60), + doorbell_wait: Duration::from_secs(5), + batch: 100, + } + } +} + +/// Waits before a slot whose notify call failed is drained again. +const RETRY_WAIT: Duration = Duration::from_secs(1); + +/// How many times one entry's call may fail while the server takes other streams' calls before +/// the entry is dropped. +const MAX_ENTRY_FAILURES: u32 = 10; + +/// How many entries' failures a driver counts at most. +const MAX_TRACKED_FAILURES: usize = 10_000; + +/// What one drain did with its entries. +#[derive(Default)] +struct Sent { + /// The entries to ack, since they landed or never will. + acked: Vec, + /// Some entries are left for the next drain. + left_pending: bool, + /// A namespace answered that the server takes no notifications. + unsupported: bool, +} + +impl Sent { + fn ack(&mut self, ids: impl IntoIterator) { + self.acked.extend(ids); + } +} + +/// A driver over one store's notification log. +pub struct StreamDriver { + store: Arc, + log: Arc, + client: Arc, + options: StreamDriverOptions, + /// The slots a drain runs on now, so the doorbells and the sweep don't drain one twice. Set + /// when another task wanted to drain the slot meanwhile, so the drain looks once more. + draining: Mutex>, + /// Whether any call landed, so a server with no notifiers at all is told from one namespace + /// without them. + delivered: AtomicBool, + /// How often each entry's call failed while the server took other calls. + failures: Mutex>, +} + +impl StreamDriver { + /// A driver for `store`, which must keep a notification log, sending through `client`. + pub fn new( + store: Arc, + client: Arc, + options: StreamDriverOptions, + ) -> StreamResult { + let log = store.notification_log().ok_or_else(|| { + StreamError::refused(format!( + "the {} store keeps no notification log, so it needs no driver", + store.name() + )) + })?; + if options.consumer.is_empty() { + return Err(StreamError::invalid_argument( + "a driver needs a consumer name", + )); + } + Ok(Self { + store, + log, + client, + options, + draining: Mutex::default(), + delivered: AtomicBool::new(false), + failures: Mutex::default(), + }) + } + + /// Runs until the returned future is dropped. It only ends early when the server can't take + /// notifications at all: every namespace it sent for answered so, and no call ever landed. + pub async fn run(self: Arc) -> StreamResult<()> { + let mut tasks = JoinSet::new(); + for doorbell in 0..self.log.doorbells() { + let driver = self.clone(); + tasks.spawn(async move { driver.listen(doorbell).await }); + } + let driver = self.clone(); + tasks.spawn(async move { driver.sweep().await }); + // Dropping the set aborts every task, so a caller stops the driver by dropping this. + match tasks.join_next().await { + Some(Ok(result)) => result, + Some(Err(error)) => Err(StreamError::storage(format!( + "a stream driver task stopped: {error}" + ))), + None => Ok(()), + } + } + + async fn listen(&self, doorbell: u32) -> StreamResult<()> { + loop { + let slots = match self + .log + .wait_doorbell(doorbell, &self.options.consumer, self.options.doorbell_wait) + .await + { + Ok(slots) => slots, + Err(error) => { + tracing::warn!(doorbell, %error, "A stream driver could not read a doorbell"); + tokio::time::sleep(RETRY_WAIT).await; + continue; + } + }; + for slot in slots { + self.drain(slot).await?; + } + } + } + + async fn sweep(&self) -> StreamResult<()> { + loop { + match self.log.busy().await { + Ok(slots) => { + for slot in slots { + self.drain(slot).await?; + } + } + Err(error) => { + tracing::warn!(%error, "A stream driver could not sweep the notification log"); + } + } + tokio::time::sleep(self.options.sweep_interval).await; + } + } + + /// Sends everything `slot` holds, one batch at a time, unless another task drains it now. + /// Fails only when the server can't take notifications at all. + async fn drain(&self, slot: u32) -> StreamResult<()> { + { + let mut draining = self.draining.lock().unwrap(); + if let Some(again) = draining.get_mut(&slot) { + *again = true; + return Ok(()); + } + draining.insert(slot, false); + } + loop { + let drained = self.drain_slot(slot).await; + let mut draining = self.draining.lock().unwrap(); + if drained.is_ok() && draining.get(&slot) == Some(&true) { + draining.insert(slot, false); + continue; + } + draining.remove(&slot); + return drained; + } + } + + async fn drain_slot(&self, slot: u32) -> StreamResult<()> { + loop { + let entries = match self + .log + .take( + slot, + &self.options.consumer, + self.options.claim_idle, + self.options.batch, + ) + .await + { + Ok(entries) if entries.is_empty() => return Ok(()), + Ok(entries) => entries, + Err(error) => { + tracing::warn!(slot, %error, "A stream driver could not read a slot"); + return Ok(()); + } + }; + // A server call names one namespace, and one namespace's refusal must not hold back + // the others in the slot. + let mut by_namespace: BTreeMap> = BTreeMap::new(); + for entry in entries { + by_namespace + .entry(entry.chain.namespace.clone()) + .or_default() + .push(entry); + } + let mut sent = Sent::default(); + for (namespace, entries) in by_namespace { + self.send_namespace(slot, &namespace, entries, &mut sent) + .await; + } + if !sent.acked.is_empty() { + if let Err(error) = self.log.ack(slot, &sent.acked).await { + // The entries come again, and the notifier drops the repeat. + tracing::warn!(slot, %error, "A stream driver could not ack its entries"); + return Ok(()); + } + let mut failures = self.failures.lock().unwrap(); + for id in &sent.acked { + failures.remove(id); + } + } + if sent.unsupported && !self.delivered.load(Ordering::Relaxed) { + return Err(StreamError::unsupported( + "the server takes no stream notifications", + )); + } + if sent.left_pending { + // Left pending, so the next drain sends them again. + tokio::time::sleep(RETRY_WAIT).await; + return Ok(()); + } + } + } + + /// Sends one namespace's entries in one call. The server answers each stream, and a stream it + /// refused or failed is settled on its own. When the server refuses the whole call, each + /// stream goes on its own, so one bad entry holds back no other stream. + async fn send_namespace( + &self, + slot: u32, + namespace: &str, + entries: Vec, + sent: &mut Sent, + ) { + let (streams, unsendable): (Vec<_>, Vec<_>) = self + .fold(entries) + .into_iter() + .partition(|(notification, _)| notification.counter > 0); + sent.ack(unsendable.into_iter().flat_map(|(_, ids)| ids)); + if streams.is_empty() { + return; + } + let notifications = streams.iter().map(|(n, _)| n.clone()).collect(); + match self.client.notify_streams_each(notifications).await { + Ok(results) => { + // A server that took some entries is up, so the ones it failed count toward + // their drop. A server that failed them all may be busy, so they wait. + let reached = results.iter().any(Result::is_ok); + if reached { + self.delivered.store(true, Ordering::Relaxed); + } + for (stream, result) in streams.into_iter().zip(results) { + match result { + Ok(()) => sent.ack(stream.1), + Err(error) => self.settle_one(slot, error, vec![stream], sent, reached), + } + } + } + Err(NotifyError::Unsupported(error)) => { + // A server with no notifiers answers this for every namespace. Only one namespace + // answering it, the others still get theirs. + tracing::warn!( + slot, + namespace, + error, + "A namespace takes no stream notifications" + ); + sent.unsupported = true; + sent.ack(streams.into_iter().flat_map(|(_, ids)| ids)); + } + Err(error) if streams.len() == 1 => { + self.settle_one(slot, error, streams, sent, false); + } + Err(_) => { + let mut reached = false; + let mut failed = vec![]; + for (notification, ids) in streams { + match self.client.notify_streams(vec![notification.clone()]).await { + Ok(()) => { + self.delivered.store(true, Ordering::Relaxed); + reached = true; + sent.ack(ids); + } + Err(error) => failed.push((error, notification, ids)), + } + } + for (error, notification, ids) in failed { + self.settle_one(slot, error, vec![(notification, ids)], sent, reached); + } + } + } + } + + /// What becomes of one stream's entries whose call failed. A refusal won't change on a retry, + /// so it is logged and acked. A failure is tried again, but only [MAX_ENTRY_FAILURES] times + /// while the server takes other streams' calls, since then the server is up and this entry + /// is the one it can't take. + fn settle_one( + &self, + slot: u32, + error: NotifyError, + streams: Vec<(StreamNotification, Vec)>, + sent: &mut Sent, + server_reached: bool, + ) { + for (notification, ids) in streams { + let workflow_id = notification.chain.workflow_id.as_str(); + let topic = notification.topic.as_str(); + match &error { + NotifyError::Unsupported(message) => { + tracing::warn!(slot, workflow_id, topic, message, "Dropped a notification"); + sent.unsupported = true; + sent.ack(ids); + } + NotifyError::Refused(message) => { + tracing::warn!( + slot, + workflow_id, + topic, + message, + "The server refused a stream notification, so it is dropped" + ); + sent.ack(ids); + } + NotifyError::Failed(message) => { + let failures = if server_reached { + self.failed_again(&ids) + } else { + 0 + }; + if failures >= MAX_ENTRY_FAILURES { + tracing::warn!( + slot, + workflow_id, + topic, + message, + "A stream notification failed {failures} times while others landed, \ + so it is dropped" + ); + sent.ack(ids); + } else { + tracing::warn!(slot, workflow_id, topic, message, "A notify call failed"); + sent.left_pending = true; + } + } + } + } + } + + /// Counts one more failure for each entry, and answers the highest count. + fn failed_again(&self, ids: &[String]) -> u32 { + let mut failures = self.failures.lock().unwrap(); + // Entries another driver acked since are never seen again, so the map is bounded. + if failures.len() > MAX_TRACKED_FAILURES { + failures.clear(); + } + ids.iter() + .map(|id| { + let count = failures.entry(id.clone()).or_default(); + *count += 1; + *count + }) + .max() + .unwrap_or(0) + } + + /// One notification per stream, at the newest counter the entries hold for it, with the ids + /// of every entry it stands for. + fn fold(&self, entries: Vec) -> Vec<(StreamNotification, Vec)> { + let mut newest: BTreeMap<(ChainKey, String), (i64, String, Vec)> = BTreeMap::new(); + for entry in entries { + let Some(counter) = progress_counter(&entry.position) else { + tracing::warn!( + workflow_id = %entry.chain.workflow_id, + activity_id = %entry.chain.activity_id, + topic = %entry.topic, + position = %entry.position, + "No notifier counter comes from this position, so it isn't notified" + ); + // No call will take it, so it is acked with the stream's other entries, if any. + newest + .entry((ChainKey::of(&entry.chain), entry.topic)) + .or_insert((0, String::new(), vec![])) + .2 + .push(entry.id); + continue; + }; + let key = (ChainKey::of(&entry.chain), entry.topic); + let held = newest.entry(key).or_insert((0, String::new(), vec![])); + held.2.push(entry.id); + if held.0 < counter { + (held.0, held.1) = (counter, entry.position); + } + } + newest + .into_iter() + .map(|((chain, topic), (counter, position, ids))| { + let chain = chain.into_chain(); + let (kind, id) = chain_owner(&chain); + let hash = stream_hash(&chain.namespace, kind, &id, &topic); + let notification = StreamNotification { + position: mint_cursor(self.store.name(), &hash, &position), + chain, + topic, + counter, + close: None, + }; + (notification, ids) + }) + .collect() + } +} + +/// A chain as a map key: its namespace, Workflow id, first run, owner kind and Activity id. +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] +struct ChainKey(String, String, String, i32, String); + +impl ChainKey { + fn of(chain: &ChainId) -> Self { + Self( + chain.namespace.clone(), + chain.workflow_id.clone(), + chain.first_run_id.clone(), + chain.owner_kind, + chain.activity_id.clone(), + ) + } + + fn into_chain(self) -> ChainId { + ChainId { + namespace: self.0, + workflow_id: self.1, + first_run_id: self.2, + owner_kind: self.3, + activity_id: self.4, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{MemoryStore, proto::StoreAppendRequest}; + use std::sync::atomic::{AtomicBool, Ordering}; + + /// Records each notify call, and fails them while `fail` is set. + #[derive(Default)] + struct Recorder { + calls: Mutex>>, + fail: AtomicBool, + unsupported: AtomicBool, + /// A call that names this Workflow id is refused whole, as the server does. + refuse_workflow: Mutex>, + /// A call that names this Workflow id fails whole. + fail_workflow: Mutex>, + /// A call for this namespace answers that it takes no notifications. + unsupported_namespace: Mutex>, + /// The server answers an entry for this Workflow id as refused, and takes the others. + refuse_entry: Mutex>, + /// The server answers an entry for this Workflow id as failed, and takes the others. + fail_entry: Mutex>, + } + + #[async_trait::async_trait] + impl NotifierClient for Recorder { + async fn notify(&self, _: StreamNotification) -> Result<(), NotifyError> { + unreachable!("a driver sends batches") + } + + async fn notify_streams( + &self, + notifications: Vec, + ) -> Result<(), NotifyError> { + if self.unsupported.load(Ordering::SeqCst) { + return Err(NotifyError::Unsupported("no notifier".to_string())); + } + if self.fail.load(Ordering::SeqCst) { + return Err(NotifyError::Failed("the server is down".to_string())); + } + let names = |field: &Mutex>, of: fn(&StreamNotification) -> &str| { + let field = field.lock().unwrap(); + notifications + .iter() + .any(|n| field.as_deref() == Some(of(n))) + }; + if names(&self.unsupported_namespace, |n| &n.chain.namespace) { + return Err(NotifyError::Unsupported("off here".to_string())); + } + if names(&self.refuse_workflow, |n| &n.chain.workflow_id) { + return Err(NotifyError::Refused("no such owner".to_string())); + } + if names(&self.fail_workflow, |n| &n.chain.workflow_id) { + return Err(NotifyError::Failed("internal".to_string())); + } + self.calls.lock().unwrap().push(notifications); + Ok(()) + } + + async fn notify_streams_each( + &self, + notifications: Vec, + ) -> Result>, NotifyError> { + let refused = self.refuse_entry.lock().unwrap().clone(); + let failed = self.fail_entry.lock().unwrap().clone(); + let results: Vec<_> = notifications + .iter() + .map(|n| match Some(n.chain.workflow_id.as_str()) { + id if id == refused.as_deref() => { + Err(NotifyError::Refused("no such owner".to_string())) + } + id if id == failed.as_deref() => Err(NotifyError::Failed("busy".to_string())), + _ => Ok(()), + }) + .collect(); + let landed = notifications + .into_iter() + .zip(&results) + .filter(|(_, result)| result.is_ok()) + .map(|(notification, _)| notification) + .collect(); + self.notify_streams(landed).await?; + Ok(results) + } + } + + impl Recorder { + /// Each stream's notifications, as `(workflow id, topic, counter)`, in call order. + fn sent(&self) -> Vec<(String, String, i64)> { + self.calls + .lock() + .unwrap() + .iter() + .flatten() + .map(|n| (n.chain.workflow_id.clone(), n.topic.clone(), n.counter)) + .collect() + } + } + + fn chain_in(namespace: &str, workflow_id: &str) -> ChainId { + ChainId { + namespace: namespace.to_string(), + workflow_id: workflow_id.to_string(), + first_run_id: "run-1".to_string(), + ..Default::default() + } + } + + async fn append(store: &MemoryStore, workflow_id: &str, topic: &str, sequence: i64) { + append_in(store, "ns", workflow_id, topic, sequence).await; + } + + async fn append_in( + store: &MemoryStore, + namespace: &str, + workflow_id: &str, + topic: &str, + sequence: i64, + ) { + store + .append(StoreAppendRequest { + chain: Some(chain_in(namespace, workflow_id)), + topic: topic.to_string(), + producer_id: "p".to_string(), + attempt: 1, + sequence, + digest: vec![sequence as u8], + records: vec![b"record".to_vec()], + }) + .await + .unwrap(); + } + + fn driver(store: &Arc, client: &Arc, consumer: &str) -> StreamDriver { + driver_claiming(store, client, consumer, Duration::from_secs(60)) + } + + fn driver_claiming( + store: &Arc, + client: &Arc, + consumer: &str, + claim_idle: Duration, + ) -> StreamDriver { + StreamDriver::new( + store.clone(), + client.clone(), + StreamDriverOptions { + consumer: consumer.to_string(), + claim_idle, + ..Default::default() + }, + ) + .unwrap() + } + + #[tokio::test] + async fn a_drain_sends_each_streams_newest_counter_once_then_acks() { + let store = Arc::new(MemoryStore::with_notification_log()); + let client = Arc::new(Recorder::default()); + for sequence in 1..=3 { + append(&store, "a", "out", sequence).await; + } + append(&store, "b", "out", 1).await; + let driver = driver(&store, &client, "d1"); + driver.drain(0).await.unwrap(); + // Memory positions are offsets, and the counter is the offset plus one. + assert_eq!( + client.sent(), + [ + ("a".to_string(), "out".to_string(), 3), + ("b".to_string(), "out".to_string(), 1) + ] + ); + assert_eq!(client.calls.lock().unwrap().len(), 1, "one call per batch"); + let position = client.calls.lock().unwrap()[0][0].position.clone(); + assert!(position.starts_with("memory:"), "a cursor"); + assert!( + store + .notification_log() + .unwrap() + .busy() + .await + .unwrap() + .is_empty() + ); + } + + #[tokio::test] + async fn an_activity_chains_entries_notify_the_activitys_stream() { + let store = Arc::new(MemoryStore::with_notification_log()); + let client = Arc::new(Recorder::default()); + let activity = ChainId { + namespace: "ns".to_string(), + first_run_id: "run-1".to_string(), + owner_kind: crate::proto::StreamOwnerKind::Activity as i32, + activity_id: "a".to_string(), + ..Default::default() + }; + // A Workflow with the same id and run writes another stream. + append(&store, "a", "out", 1).await; + let landed = store + .append(StoreAppendRequest { + chain: Some(activity.clone()), + topic: "out".to_string(), + producer_id: "p".to_string(), + attempt: 1, + sequence: 1, + digest: vec![1], + records: vec![b"record".to_vec(), b"record".to_vec()], + }) + .await + .unwrap(); + driver(&store, &client, "d1").drain(0).await.unwrap(); + let calls = client.calls.lock().unwrap().clone(); + assert_eq!(calls[0].len(), 2, "two streams, {calls:?}"); + let sent = calls[0].iter().find(|n| n.chain == activity).unwrap(); + assert_eq!(sent.counter, 2); + assert_eq!( + sent.position, + mint_cursor( + "memory", + &stream_hash("ns", crate::ACTIVITY_OWNER_KIND, "a", "out"), + &landed.last_position + ) + ); + } + + #[tokio::test] + async fn a_failed_call_leaves_the_entries_for_the_next_drain() { + let store = Arc::new(MemoryStore::with_notification_log()); + let client = Arc::new(Recorder::default()); + append(&store, "a", "out", 1).await; + let driver = driver(&store, &client, "d1"); + client.fail.store(true, Ordering::SeqCst); + driver.drain(0).await.unwrap(); + assert_eq!(store.notification_log().unwrap().busy().await.unwrap(), [0]); + client.fail.store(false, Ordering::SeqCst); + driver.drain(0).await.unwrap(); + assert_eq!(client.sent(), [("a".to_string(), "out".to_string(), 1)]); + } + + #[tokio::test] + async fn a_stopped_drivers_entries_move_to_another_driver() { + let store = Arc::new(MemoryStore::with_notification_log()); + let client = Arc::new(Recorder::default()); + append(&store, "a", "out", 1).await; + client.fail.store(true, Ordering::SeqCst); + driver(&store, &client, "stopped").drain(0).await.unwrap(); + client.fail.store(false, Ordering::SeqCst); + driver(&store, &client, "other").drain(0).await.unwrap(); + assert!( + client.sent().is_empty(), + "the first driver's hold is still fresh" + ); + driver_claiming(&store, &client, "other", Duration::from_millis(10)) + .drain(0) + .await + .unwrap(); + assert_eq!(client.sent(), [("a".to_string(), "out".to_string(), 1)]); + } + + #[tokio::test] + async fn a_running_driver_wakes_on_the_doorbell() { + let store = Arc::new(MemoryStore::with_notification_log()); + let client = Arc::new(Recorder::default()); + let running = tokio::spawn(Arc::new(driver(&store, &client, "d1")).run()); + append(&store, "a", "out", 1).await; + tokio::time::timeout(Duration::from_secs(5), async { + while client.sent().is_empty() { + tokio::time::sleep(Duration::from_millis(5)).await; + } + }) + .await + .expect("the doorbell woke the driver"); + running.abort(); + } + + #[tokio::test] + async fn a_server_without_notifiers_stops_the_driver() { + let store = Arc::new(MemoryStore::with_notification_log()); + let client = Arc::new(Recorder::default()); + client.unsupported.store(true, Ordering::SeqCst); + append(&store, "a", "out", 1).await; + let error = tokio::time::timeout( + Duration::from_secs(5), + Arc::new(driver(&store, &client, "d1")).run(), + ) + .await + .unwrap() + .unwrap_err(); + assert_eq!(error.kind, crate::proto::StreamFailureKind::Unsupported); + } + + async fn log_is_empty(store: &MemoryStore) -> bool { + store + .notification_log() + .unwrap() + .busy() + .await + .unwrap() + .is_empty() + } + + #[tokio::test] + async fn one_refused_entry_holds_back_no_other_stream() { + let store = Arc::new(MemoryStore::with_notification_log()); + let client = Arc::new(Recorder::default()); + *client.refuse_workflow.lock().unwrap() = Some("bad".to_string()); + append(&store, "bad", "out", 1).await; + append(&store, "good", "out", 1).await; + driver(&store, &client, "d1").drain(0).await.unwrap(); + assert_eq!(client.sent(), [("good".to_string(), "out".to_string(), 1)]); + assert!(log_is_empty(&store).await, "the refused entry is acked too"); + } + + #[tokio::test] + async fn an_entry_the_answer_refuses_is_acked_with_no_second_call() { + let store = Arc::new(MemoryStore::with_notification_log()); + let client = Arc::new(Recorder::default()); + *client.refuse_entry.lock().unwrap() = Some("bad".to_string()); + append(&store, "bad", "out", 1).await; + append(&store, "good", "out", 1).await; + driver(&store, &client, "d1").drain(0).await.unwrap(); + assert_eq!(client.calls.lock().unwrap().len(), 1); + assert_eq!(client.sent(), [("good".to_string(), "out".to_string(), 1)]); + assert!(log_is_empty(&store).await, "the refused entry is acked too"); + } + + #[tokio::test(start_paused = true)] + async fn an_entry_the_answer_fails_stays_while_no_other_lands() { + let store = Arc::new(MemoryStore::with_notification_log()); + let client = Arc::new(Recorder::default()); + *client.fail_entry.lock().unwrap() = Some("bad".to_string()); + append(&store, "bad", "out", 1).await; + let driver = driver(&store, &client, "d1"); + for _ in 0..=MAX_ENTRY_FAILURES { + driver.drain(0).await.unwrap(); + } + assert!(!log_is_empty(&store).await); + *client.fail_entry.lock().unwrap() = None; + driver.drain(0).await.unwrap(); + assert_eq!(client.sent(), [("bad".to_string(), "out".to_string(), 1)]); + assert!(log_is_empty(&store).await); + } + + #[tokio::test(start_paused = true)] + async fn an_entry_the_answer_keeps_failing_while_others_land_is_dropped() { + let store = Arc::new(MemoryStore::with_notification_log()); + let client = Arc::new(Recorder::default()); + *client.fail_entry.lock().unwrap() = Some("bad".to_string()); + append(&store, "bad", "out", 1).await; + let driver = driver(&store, &client, "d1"); + for sequence in 1..=MAX_ENTRY_FAILURES as i64 { + assert!(!log_is_empty(&store).await, "kept after {sequence} drains"); + append(&store, "good", "out", sequence).await; + driver.drain(0).await.unwrap(); + } + assert!(log_is_empty(&store).await); + assert!(client.sent().iter().all(|(id, _, _)| id == "good")); + } + + #[tokio::test] + async fn a_namespace_without_notifiers_stops_neither_the_driver_nor_the_others() { + let store = Arc::new(MemoryStore::with_notification_log()); + let client = Arc::new(Recorder::default()); + *client.unsupported_namespace.lock().unwrap() = Some("off".to_string()); + append_in(&store, "off", "a", "out", 1).await; + append_in(&store, "on", "b", "out", 1).await; + driver(&store, &client, "d1").drain(0).await.unwrap(); + assert_eq!(client.sent(), [("b".to_string(), "out".to_string(), 1)]); + assert!(log_is_empty(&store).await); + } + + #[tokio::test(start_paused = true)] + async fn an_entry_that_keeps_failing_while_others_land_is_dropped() { + let store = Arc::new(MemoryStore::with_notification_log()); + let client = Arc::new(Recorder::default()); + *client.fail_workflow.lock().unwrap() = Some("bad".to_string()); + append(&store, "bad", "out", 1).await; + let driver = driver(&store, &client, "d1"); + for sequence in 1..=MAX_ENTRY_FAILURES as i64 { + assert!(!log_is_empty(&store).await, "kept after {sequence} drains"); + append(&store, "good", "out", sequence).await; + driver.drain(0).await.unwrap(); + } + assert!(log_is_empty(&store).await); + assert_eq!( + client.sent().last(), + Some(&( + "good".to_string(), + "out".to_string(), + MAX_ENTRY_FAILURES as i64 + )) + ); + } + + #[tokio::test(start_paused = true)] + async fn a_failing_server_drops_nothing() { + let store = Arc::new(MemoryStore::with_notification_log()); + let client = Arc::new(Recorder::default()); + append(&store, "a", "out", 1).await; + append(&store, "b", "out", 1).await; + client.fail.store(true, Ordering::SeqCst); + let driver = driver(&store, &client, "d1"); + for _ in 0..=MAX_ENTRY_FAILURES { + driver.drain(0).await.unwrap(); + } + assert!(!log_is_empty(&store).await); + } + + #[test] + fn the_default_consumer_names_the_host_and_the_process() { + let consumer = StreamDriverOptions::default().consumer; + assert!( + consumer.ends_with(&format!("-{}", std::process::id())), + "{consumer}" + ); + } + + #[test] + fn a_store_without_a_log_needs_no_driver() { + let error = StreamDriver::new( + Arc::new(MemoryStore::new()), + Arc::new(Recorder::default()), + StreamDriverOptions::default(), + ) + .err() + .unwrap(); + assert_eq!(error.kind, crate::proto::StreamFailureKind::Refused); + } +} diff --git a/crates/streams/src/lib.rs b/crates/streams/src/lib.rs index 293485b..cf57c87 100644 --- a/crates/streams/src/lib.rs +++ b/crates/streams/src/lib.rs @@ -13,6 +13,7 @@ mod close; mod connection; mod conventions; mod cursor; +mod driver; mod error; mod follow; mod handle; @@ -39,7 +40,7 @@ mod wake; mod workflow; #[cfg(feature = "redis")] -pub use self::redis::{RedisStore, RedisStoreOptions}; +pub use self::redis::{DEFAULT_NOTIFY_DOORBELLS, RedisStore, RedisStoreOptions}; pub use close::{CHAIN_LOG, append_batch, close_chain, close_topic}; pub use connection::{connect_stream_service, execute_sync_nexus_operation}; pub use conventions::{ @@ -47,6 +48,7 @@ pub use conventions::{ check_topic, content_hash_text, }; pub use cursor::{BEGINNING, END, cursor_position, mint_cursor, stream_hash}; +pub use driver::{StreamDriver, StreamDriverOptions}; pub use error::{StreamError, StreamResult}; pub use handle::{OWNER_RECHECK, OWNER_START_GRACE, Streams, StreamsOptions}; pub use memory::MemoryStore; diff --git a/crates/streams/src/memory.rs b/crates/streams/src/memory.rs index b42bd26..93fd128 100644 --- a/crates/streams/src/memory.rs +++ b/crates/streams/src/memory.rs @@ -8,8 +8,9 @@ //! [MemoryStore::with_capabilities] masks capabilities off, so the fallbacks run in tests. use crate::{ - ACTIVITY_OWNER_KIND, Capabilities, MAX_BATCH_BYTES, MAX_RECORD_BYTES, StreamError, - StreamResult, StreamStore, WORKFLOW_ACTIVITY_OWNER_KIND, WORKFLOW_OWNER_KIND, chain_owner, + ACTIVITY_OWNER_KIND, Capabilities, MAX_BATCH_BYTES, MAX_RECORD_BYTES, NotificationEntry, + NotificationLog, StreamError, StreamResult, StreamStore, WORKFLOW_ACTIVITY_OWNER_KIND, + WORKFLOW_OWNER_KIND, chain_owner, proto::{ ChainId, DeleteOwnerRequest, DeleteOwnerResponse, StoreAppendRequest, StoreAppendResponse, StoreLatestRequest, StoreLatestResponse, StoreReadRequest, StoreReadResponse, StoredClose, @@ -19,7 +20,7 @@ use crate::{ }; use std::{ collections::{BTreeSet, HashMap, VecDeque}, - sync::Mutex, + sync::{Arc, Mutex}, time::Duration, }; use tokio::{sync::watch, time::Instant}; @@ -31,7 +32,7 @@ const FULL: Capabilities = Capabilities { producer_state: true, kept_last: true, holds: true, - change_feed: false, + change_feed: true, cheap_logs: true, delete_owner: true, topic_listing: true, @@ -113,6 +114,7 @@ pub struct MemoryStore { chains: Mutex>, /// Bumped on every write, so a waiting read checks again. written: watch::Sender, + log: Option>, capabilities: Capabilities, } @@ -121,17 +123,145 @@ impl Default for MemoryStore { Self { chains: Mutex::default(), written: watch::Sender::new(0), + log: None, capabilities: FULL, } } } +/// A notification log in memory: one partition and one doorbell. +#[derive(Debug)] +struct MemoryLog { + state: Mutex, + /// Bumped on every entry, so a waiting driver looks again. + rung: watch::Sender, +} + +#[derive(Debug, Default)] +struct LogState { + next_id: u64, + items: VecDeque, +} + +#[derive(Debug)] +struct LogItem { + entry: NotificationEntry, + /// The consumer that took the entry and when, until an ack drops it. + taken: Option<(String, Instant)>, +} + +impl MemoryLog { + fn push(&self, chain: &ChainId, topic: &str, position: String) { + let mut state = self.state.lock().unwrap(); + state.next_id += 1; + let id = state.next_id.to_string(); + state.items.push_back(LogItem { + entry: NotificationEntry { + id, + chain: chain.clone(), + topic: topic.to_string(), + position, + }, + taken: None, + }); + drop(state); + self.rung.send_modify(|rung| *rung += 1); + } + + fn holds_entries(&self) -> bool { + !self.state.lock().unwrap().items.is_empty() + } +} + +#[async_trait::async_trait] +impl NotificationLog for MemoryLog { + fn partitions(&self) -> u32 { + 1 + } + + fn doorbells(&self) -> u32 { + 1 + } + + async fn wait_doorbell( + &self, + _doorbell: u32, + _consumer: &str, + block: Duration, + ) -> StreamResult> { + let mut rung = self.rung.subscribe(); + if !self.holds_entries() { + let _ = tokio::time::timeout(block, rung.changed()).await; + } + Ok(if self.holds_entries() { + vec![0] + } else { + vec![] + }) + } + + async fn take( + &self, + _partition: u32, + consumer: &str, + claim_idle: Duration, + max: usize, + ) -> StreamResult> { + let now = Instant::now(); + let mut state = self.state.lock().unwrap(); + let mut taken = vec![]; + for item in state.items.iter_mut() { + if taken.len() >= max { + break; + } + let free = match &item.taken { + None => true, + Some((owner, _)) if owner == consumer => true, + Some((_, at)) => now.duration_since(*at) >= claim_idle, + }; + if free { + item.taken = Some((consumer.to_string(), now)); + taken.push(item.entry.clone()); + } + } + Ok(taken) + } + + async fn ack(&self, _partition: u32, ids: &[String]) -> StreamResult<()> { + self.state + .lock() + .unwrap() + .items + .retain(|item| !ids.contains(&item.entry.id)); + Ok(()) + } + + async fn busy(&self) -> StreamResult> { + Ok(if self.holds_entries() { + vec![0] + } else { + vec![] + }) + } +} + impl MemoryStore { /// An empty store. pub fn new() -> Self { Self::default() } + /// An empty store whose appends write a notification log in the same step. + pub fn with_notification_log() -> Self { + Self { + log: Some(Arc::new(MemoryLog { + state: Mutex::default(), + rung: watch::Sender::new(0), + })), + ..Self::default() + } + } + /// Forgets a topic and its holds, as a store does when the topic's last key expired. pub fn forget(&self, chain: &ChainId, topic: &str) { if let Some(chain) = self.chains.lock().unwrap().get_mut(&chain_key(chain)) { @@ -285,8 +415,15 @@ impl StreamStore for MemoryStore { NAME } + fn notification_log(&self) -> Option> { + self.log.clone().map(|log| log as Arc) + } + fn capabilities(&self) -> Capabilities { - self.capabilities + Capabilities { + change_feed: self.capabilities.change_feed && self.log.is_some(), + ..self.capabilities + } } async fn append(&self, request: StoreAppendRequest) -> StreamResult { @@ -363,6 +500,10 @@ impl StreamStore for MemoryStore { }, ); } + // Under the store's lock, so the entry lands in the same step as the records. + if let (Some(log), Some(chain)) = (&self.log, &request.chain) { + log.push(chain, &request.topic, last.to_string()); + } drop(chains); self.wrote(); Ok(StoreAppendResponse { @@ -700,13 +841,19 @@ mod tests { #[test] fn a_mask_only_takes_capabilities_away() { - assert_eq!(MemoryStore::new().capabilities(), FULL); + // The change feed comes with a notification log. + let no_log = Capabilities { + change_feed: false, + ..FULL + }; + assert_eq!(MemoryStore::new().capabilities(), no_log); + assert_eq!(MemoryStore::with_notification_log().capabilities(), FULL); let masked = log_only().capabilities(); assert!(!masked.producer_state && !masked.kept_last && !masked.long_poll); // Positions stay offsets whatever the mask, and the limits stay the store's. assert!(masked.dense_positions); assert_eq!(masked.max_record_bytes, MAX_RECORD_BYTES); - assert_eq!(MemoryStore::with_capabilities(FULL).capabilities(), FULL); + assert_eq!(MemoryStore::with_capabilities(FULL).capabilities(), no_log); } #[tokio::test] diff --git a/crates/streams/src/redis/conn.rs b/crates/streams/src/redis/conn.rs index b58be09..e001f86 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> { diff --git a/crates/streams/src/redis/keys.rs b/crates/streams/src/redis/keys.rs index 7163e2b..b95d8cd 100644 --- a/crates/streams/src/redis/keys.rs +++ b/crates/streams/src/redis/keys.rs @@ -24,9 +24,21 @@ //! :{:!wa:::}:topics //! ``` //! +//! 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 +//! :{}:a slot s's topics with records to archive, with an archive +//! ``` +//! //! 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, StreamOwnerKind}; pub(crate) use crate::record::part; @@ -103,6 +115,28 @@ impl ChainKeys { } unpart(part) } + + /// 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)) +} + +/// Slot `slot`'s set of log keys with records to archive, scored by their newest write. +pub(crate) fn archive_pending(prefix: &str, slot: u16) -> String { + format!("{}:{{{}}}:a", 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 @@ -284,7 +318,7 @@ mod tests { } #[test] - fn a_log_key_names_its_chain_and_topic() { + 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 { @@ -292,6 +326,10 @@ mod tests { 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!(slot(¬ifications), slot(&keys.topics())); assert_eq!( chain_of_log(prefix, &keys.log(topic)), Some((chain, topic.to_string())) @@ -421,4 +459,12 @@ mod tests { "p:{ns:!wa:a%2Ab:*" ); } + + #[test] + fn the_doorbells_spread_over_the_slot_range() { + let slots: Vec = (0..16) + .map(|d| super::super::conn::slot(&doorbell("p", d, 16))) + .collect(); + assert_eq!(slots, (0..16).map(|d| d * 1024).collect::>()); + } } diff --git a/crates/streams/src/redis/log.rs b/crates/streams/src/redis/log.rs new file mode 100644 index 0000000..37b8b11 --- /dev/null +++ b/crates/streams/src/redis/log.rs @@ -0,0 +1,446 @@ +//! 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::{archive_pending, chain_of_log, doorbell, notifications}, + slots::SLOTS, +}; +use crate::{ArchiveMark, 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, + /// Whether the store marks topics for an archive, so the slots' sets are worth reading. + archive: bool, +} + +impl RedisLog { + pub(crate) fn new( + shared: Shared, + blocking: BlockingPool, + prefix: String, + doorbells: u32, + archive: bool, + ) -> Self { + Self { + shared, + blocking, + prefix, + doorbells: doorbells.clamp(1, SLOTS as u32), + archive, + } + } + + /// 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, + 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, self.doorbells); + 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) + } + + async fn archive_marks(&self, max: usize) -> StreamResult> { + if !self.archive || max == 0 { + return Ok(vec![]); + } + let mut marks = vec![]; + for (node, slots) in self.primaries().await? { + let mut pipeline = redis::pipe(); + for &slot in &slots { + pipeline + .cmd("ZRANGE") + .arg(archive_pending(&self.prefix, slot)) + .arg(0) + .arg(max - 1) + .arg("WITHSCORES"); + } + let sets: Vec> = self + .shared + .query_pipeline_on(node.as_ref(), pipeline, slots.len()) + .await + .mapped(Call::Read)?; + // The scores come from the scripts' `TIME`, so the age is taken on the same clock. + let mut time = redis::pipe(); + time.cmd("TIME"); + let [(seconds, micros)]: [(u64, u64); 1] = self + .shared + .query_pipeline_on(node.as_ref(), time, 1) + .await + .mapped(Call::Read)?; + let now_ms = seconds * 1000 + micros / 1000; + for (log, written_ms) in sets.into_iter().flatten() { + match chain_of_log(&self.prefix, &log) { + Some((chain, topic)) => marks.push(ArchiveMark { + chain, + topic, + written_ms: written_ms as u64, + idle_ms: now_ms.saturating_sub(written_ms as u64), + }), + None => tracing::warn!(log, "A slot's set to archive names no log key"), + } + } + } + Ok(marks) + } +} + +/// 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 1d32dcf..e3dcdb4 100644 --- a/crates/streams/src/redis/mod.rs +++ b/crates/streams/src/redis/mod.rs @@ -3,10 +3,13 @@ mod conn; mod errors; mod keys; +mod log; mod scripts; +mod slots; use crate::{ - Capabilities, MAX_BATCH_BYTES, MAX_RECORD_BYTES, StreamError, StreamResult, StreamStore, + Capabilities, MAX_BATCH_BYTES, MAX_RECORD_BYTES, NotificationLog, StreamError, StreamResult, + StreamStore, proto::{ ChainId, DeleteOwnerRequest, DeleteOwnerResponse, StoreAppendRequest, StoreAppendResponse, StoreLatestRequest, StoreLatestResponse, StoreReadRequest, StoreReadResponse, StoredClose, @@ -22,7 +25,7 @@ use prost::Message; use redis::{Cmd, Script, Value, streams::StreamReadReply}; use std::{ collections::HashMap, - sync::LazyLock, + sync::{Arc, LazyLock}, time::{Duration, SystemTime, UNIX_EPOCH}, }; use tokio::time::Instant; @@ -39,7 +42,7 @@ const CAPABILITIES: Capabilities = Capabilities { producer_state: true, kept_last: true, holds: true, - change_feed: false, + change_feed: true, cheap_logs: true, delete_owner: true, topic_listing: true, @@ -72,6 +75,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, /// Runs with no Lua: no dedupe entries, no close flags and no topic meta. The service then /// takes the fallbacks of producer state and the kept last position, as on a store that has /// only a log. For tests of those fallbacks on a real store. @@ -90,11 +97,16 @@ 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, plain: false, } } } +/// 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, @@ -104,6 +116,7 @@ pub struct RedisStore { grace_ms: u64, floor_lease_ms: u64, floor_release_grace_ms: u64, + log: Option>, plain: bool, } @@ -136,6 +149,11 @@ impl RedisStore { options.retention ))); } + if options.plain && options.notification_log { + return Err(StreamError::refused( + "the Redis store in plain mode runs no script, so it keeps no notification log", + )); + } // The scripts trim below `now - retention` as a stream id, which can't be negative. let now_ms = SystemTime::now() .duration_since(UNIX_EPOCH) @@ -157,7 +175,27 @@ 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, + false, + ))) + } else { + None + }; Ok(Self { + log, shared, blocking, prefix: options.key_prefix, @@ -605,6 +643,10 @@ impl StreamStore for RedisStore { NAME } + fn notification_log(&self) -> Option> { + self.log.clone().map(|log| log as Arc) + } + fn capabilities(&self) -> Capabilities { if self.plain { return Capabilities { @@ -615,7 +657,10 @@ impl StreamStore for RedisStore { ..CAPABILITIES }; } - CAPABILITIES + Capabilities { + change_feed: self.log.is_some(), + ..CAPABILITIES + } } async fn append(&self, request: StoreAppendRequest) -> StreamResult { @@ -629,12 +674,18 @@ impl StreamStore for RedisStore { return self.append_plain(&request).await; } 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)); + if self.log.is_some() { + script.key(keys::notifications(&self.prefix, keys.slot())); + } if request.chain.as_ref().is_some_and(ChainKeys::keeps_topics) { script.key(keys.topics()); } @@ -657,11 +708,17 @@ impl StreamStore for RedisStore { } appended => appended?, }; - let [first_position, last_position] = answer.as_slice() else { + 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, diff --git a/crates/streams/src/redis/scripts.rs b/crates/streams/src/redis/scripts.rs index 032abed..7d86ffd 100644 --- a/crates/streams/src/redis/scripts.rs +++ b/crates/streams/src/redis/scripts.rs @@ -19,6 +19,33 @@ pub(crate) const APPEND: &str = concat!( pub(crate) static APPEND_SCRIPT: LazyLock