diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index 33f95676a..75ed49c8d 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -92,6 +92,11 @@ relevant information. * Experimental: `StreamStoreConfig.archive` works with the PostgreSQL store. A topic's first record makes the archive's hold as a `floors` row, and the janitor drops only what the archive holds while the driver keeps up. The archive is off unless the config sets it. +* Experimental: the Redis append no longer costs more the longer a topic's retained log is. Its + trim finds the newest record to drop with `XRANGE` forward from the oldest, where the SDKs' + script read the whole log with `XREVRANGE`. At 100,000 retained records an append took about + 4 ms before and takes about the same as on a short log now. The append script's hashes are now + `188afac5`, `eeed648a` with the notification log, and `337494c0` with the archive. * 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 fd7ca09ac..0130e2b13 100644 --- a/crates/streams/docs/redis.md +++ b/crates/streams/docs/redis.md @@ -83,9 +83,13 @@ and land a second time. Readers lose nothing, since a store compares a batch wit attempt's field only and readers find a newer attempt from the records. **Retention.** Every script that writes a log trims it with `XTRIM MINID` to the retention and -slides the expiry of the log and its meta, so a stream dies a retention after its last write -and nobody cleans up when the Workflow closes. The meta outlives its log by a 30-day grace, as -the tombstone that tells a log that expired from one that never existed. +slides the expiry of the log and its meta, so a stream dies a retention after its last write and +nobody cleans up when the Workflow closes. The trim reads the oldest record with `XRANGE`, and only +when it is below the floor finds the newest record it drops with `XREVRANGE` back from the floor. +Both cost O(log n). `XREVRANGE` alone reads the whole retained log when no record is older than the +floor, and a forward read copies every dropped record into the script, which blocks Redis on a large +trim. The `+xrange` rule already covers the new call. The meta outlives its log by a 30-day grace, +as the tombstone that tells a log that expired from one that never existed. **Activity owners.** A standalone Activity run is a chain of one run. Its keys have the hash tag `{:!a::}`, and every other part of the layout is the same. Each diff --git a/crates/streams/lua/keep.lua b/crates/streams/lua/keep.lua index f22280c3b..c493ca811 100644 --- a/crates/streams/lua/keep.lua +++ b/crates/streams/lua/keep.lua @@ -28,8 +28,11 @@ local function trim(log, meta, retention, floors) local floor = floor_ms .. '-0' local life floor, life = held(floors, floor, retention) - local doomed = redis.call('XREVRANGE', log, '(' .. floor, '-', 'COUNT', 1) - if #doomed > 0 then + -- Two reads that cost O(log n) each. Reading back from the floor walks the whole log + -- when no record is older, so it runs only when the oldest record is. + local oldest = redis.call('XRANGE', log, '-', '+', 'COUNT', 1) + if #oldest > 0 and id_below(oldest[1][1], floor) then + local doomed = redis.call('XREVRANGE', log, '(' .. floor, '-', 'COUNT', 1) redis.call('XTRIM', log, 'MINID', floor) redis.call('HSET', meta, 'trimmed', doomed[1][1]) end diff --git a/crates/streams/src/redis/scripts.rs b/crates/streams/src/redis/scripts.rs index c2b9353e9..fbcd307aa 100644 --- a/crates/streams/src/redis/scripts.rs +++ b/crates/streams/src/redis/scripts.rs @@ -2,9 +2,11 @@ //! //! The append script is the one the SDKs sent, with one more check, the topic's close flag, so a //! batch can't land after the topic's close returned (DD-16), and its trim keeps what -//! positions-only readers hold. Redis doesn't roll a script back, so it runs every step that can -//! fail before its first record lands. `keep.lua` is the prelude of every script that writes a -//! log, and `floor.lua` of every script that reads or writes the holds. +//! positions-only readers hold. Its trim reads back from the floor only when the oldest record is +//! below it, since `XREVRANGE` from the floor reads the whole retained log when no record is older, +//! and a forward read copies every doomed record into Lua. Redis doesn't roll a script back, so it +//! runs every step that can fail before its first record lands. `keep.lua` is the prelude of every +//! script that writes a log, and `floor.lua` of every script that reads or writes the holds. use redis::Script; use std::sync::LazyLock; @@ -130,6 +132,25 @@ if topics then end "; + /// How the trim finds the newest record to drop, and what it replaces. + const TRIM_SCAN: (&str, &str) = ( + " -- Two reads that cost O(log n) each. Reading back from the floor walks the whole log + -- when no record is older, so it runs only when the oldest record is. + local oldest = redis.call('XRANGE', log, '-', '+', 'COUNT', 1) + if #oldest > 0 and id_below(oldest[1][1], floor) then + local doomed = redis.call('XREVRANGE', log, '(' .. floor, '-', 'COUNT', 1) + redis.call('XTRIM', log, 'MINID', floor) + redis.call('HSET', meta, 'trimmed', doomed[1][1]) + end +", + " local doomed = redis.call('XREVRANGE', log, '(' .. floor, '-', 'COUNT', 1) + if #doomed > 0 then + redis.call('XTRIM', log, 'MINID', floor) + redis.call('HSET', meta, 'trimmed', doomed[1][1]) + end +", + ); + /// What the floor adds to the append script, each with what it replaces. const FLOOR_EDITS: [(&str, &str); 8] = [ ( @@ -174,7 +195,7 @@ end assert_eq!(APPEND_LOGGED.matches("XADD', KEYS[5]").count(), 1); assert_eq!( APPEND_LOGGED_SCRIPT.get_hash(), - "e20b57d34060cd9498c9189e7cf1c2a7ae01190c" + "49820889a3c7680122a0b43ed0946c4b4cb0286e" ); } @@ -193,7 +214,7 @@ end assert_eq!(was, *APPEND_LOGGED); assert_eq!( APPEND_ARCHIVED_SCRIPT.get_hash(), - "bb8e042129e393c24f89efa5b81339d87090e8c1" + "98302a233c87f8a10303d4ca17b418cb51fe2799" ); } @@ -201,12 +222,15 @@ end fn the_append_script_is_the_one_before_the_floor_with_the_floor_edits_only() { assert_eq!( APPEND_SCRIPT.get_hash(), - "792ae093a9c09ea8da9078a724eced11ddf5935d" + "b818ae457ec5777a3a1d321e17215037cfdf5c6e" ); let mut before_floor = APPEND .strip_prefix(include_str!("../../lua/floor.lua")) .unwrap() .to_string(); + let (scan, was) = TRIM_SCAN; + assert_eq!(before_floor.matches(scan).count(), 1); + before_floor = before_floor.replacen(scan, was, 1); assert_eq!(before_floor.matches(TOPIC_SET).count(), 1); before_floor = before_floor.replacen(TOPIC_SET, "", 1); for (added, was) in FLOOR_EDITS { diff --git a/crates/streams/tests/conformance.rs b/crates/streams/tests/conformance.rs index bc73496b9..e37b838d6 100644 --- a/crates/streams/tests/conformance.rs +++ b/crates/streams/tests/conformance.rs @@ -461,6 +461,19 @@ mod on_redis { ) } + fn now_ms() -> u128 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_millis() + } + + /// A stream id as its two numbers, so ids compare in log order. + fn stream_id(id: &str) -> (u64, u64) { + let (ms, seq) = id.split_once('-').unwrap_or((id, "0")); + (ms.parse().unwrap(), seq.parse().unwrap()) + } + pub(crate) struct Redis { pub(crate) store: Arc, pub(crate) raw: Raw, @@ -767,6 +780,19 @@ mod on_redis { let _: i64 = raw.query(redis::cmd("DEL").arg(key)).await; } } + + async fn lapse_archive_hold(&self, chain: &ChainId, topic: &str) { + let floors = format!("{}:floors", log_key(&self.prefix, chain, topic)); + let _: i64 = self + .raw + .clone() + .query( + redis::cmd("HDEL") + .arg(floors) + .arg(temporalio_streams::ARCHIVE_HOLDER), + ) + .await; + } } async fn redis_with(variable: &str, cluster: bool, notification_log: bool) -> Option { @@ -804,17 +830,38 @@ mod on_redis { } async fn drop_oldest(&self, chain: &ChainId, topic: &str, keep: usize) { - // Trims as the append script does, watermark included. + // Trims as the append script does, watermark included, and keeps what a live hold + // holds, as `keep.lua` does. let mut raw = self.raw.clone(); let log = log_key(&self.prefix, chain, topic); let entries: Vec<(String, Value)> = raw .query(redis::cmd("XRANGE").arg(&log).arg("-").arg("+")) .await; - let Some(doomed) = entries.len().checked_sub(keep).filter(|&n| n > 0) else { + let floors: Vec = raw + .query(redis::cmd("HVALS").arg(format!("{log}:floors"))) + .await; + let held = floors + .iter() + .filter_map(|floor| { + let (id, until) = floor.split_once('|')?; + (until.parse::().ok()? > now_ms()).then(|| stream_id(id)) + }) + .min(); + let below_hold = entries + .iter() + .take_while(|(id, _)| held.is_none_or(|held| stream_id(id) < held)) + .count(); + let doomed = entries.len().saturating_sub(keep).min(below_hold); + if doomed == 0 { return; - }; + } let _: i64 = raw - .query(redis::cmd("XTRIM").arg(&log).arg("MAXLEN").arg(keep)) + .query( + redis::cmd("XTRIM") + .arg(&log) + .arg("MAXLEN") + .arg(entries.len() - doomed), + ) .await; raw.hset(&format!("{log}:meta"), "trimmed", &entries[doomed - 1].0) .await; @@ -2855,8 +2902,8 @@ macro_rules! archive_conformance { }; } -/// The archive cases. A hot store whose test trim ignores holds skips -/// `the_trim_keeps_what_the_archive_does_not_hold`. +/// The archive cases. `the_trim_keeps_what_the_archive_does_not_hold` runs on its own, since it +/// needs a test trim that keeps what a hold holds. macro_rules! archive_cases { ($case:expr) => { archive_conformance!($case; @@ -3338,8 +3385,15 @@ mod archive_standalone_redis { "STREAMS_REDIS_URL", false ))); - // The archive hold doesn't change the store contract. - store_cases!(super::tiered(super::on_redis::redis_with_archive( + archive_conformance!(super::tiered(super::on_redis::redis_with_archive( + "STREAMS_REDIS_URL", + false + )); + the_trim_keeps_what_the_archive_does_not_hold, + ); + // The test trim keeps what the archive's hold holds, so the store cases that trim run with no + // hold, as on memory and PostgreSQL. + store_cases!(super::tiered(super::on_redis::redis_with_log( "STREAMS_REDIS_URL", false ))); @@ -3367,7 +3421,13 @@ mod archive_cluster_redis { "STREAMS_REDIS_CLUSTER_URL", true ))); - store_cases!(super::tiered(super::on_redis::redis_with_archive( + archive_conformance!(super::tiered(super::on_redis::redis_with_archive( + "STREAMS_REDIS_CLUSTER_URL", + true + )); + the_trim_keeps_what_the_archive_does_not_hold, + ); + store_cases!(super::tiered(super::on_redis::redis_with_log( "STREAMS_REDIS_CLUSTER_URL", true ))); diff --git a/crates/streams/tests/redis_store.rs b/crates/streams/tests/redis_store.rs index a83c6ffad..6e8652656 100644 --- a/crates/streams/tests/redis_store.rs +++ b/crates/streams/tests/redis_store.rs @@ -2142,6 +2142,52 @@ async fn a_readers_short_hold_never_ends_the_archives_hold() { assert_eq!(sequences(&mut setup, &chain).await, [1, 2, 3]); } +/// The median time of `n` single-record appends to `chain`'s topic, from sequence `first`. +async fn median_append(store: &RedisStore, chain: &ChainId, first: i64, n: i64) -> Duration { + let mut times = vec![]; + for sequence in first..first + n { + let started = tokio::time::Instant::now(); + store + .append(append(chain, "p", sequence, 1, &sequence.to_be_bytes())) + .await + .unwrap(); + times.push(started.elapsed()); + } + times.sort(); + times[times.len() / 2] +} + +#[tokio::test] +async fn an_append_costs_the_same_however_long_the_retained_log_is() { + let Some(setup) = setup().await else { return }; + let (short, long) = (chain(), chain()); + let mut sequence = 1; + // 100,000 entries, all within the retention, so the trim drops none of them. + while sequence <= 100_000 { + setup + .store + .append(append(&long, "p", sequence, 1000, &sequence.to_be_bytes())) + .await + .unwrap(); + sequence += 1000; + } + setup + .store + .append(append(&short, "p", 1, 1, b"1")) + .await + .unwrap(); + let on_short = median_append(&setup.store, &short, 2, 30).await; + let on_long = median_append(&setup.store, &long, sequence, 30).await; + // Finding the newest record to trim once scanned the whole log, about 4 ms at this length. + assert!( + on_long < Duration::from_millis(1), + "{on_long:?} on the long log, {on_short:?} on the short one" + ); + assert!( + on_long < on_short * 3 + Duration::from_micros(200), + "{on_long:?} on the long log, {on_short:?} on the short one" + ); +} #[tokio::test] async fn a_retention_past_the_epoch_is_refused_at_connect() { // The check runs before any connection, so no server is needed. @@ -2433,3 +2479,49 @@ async fn a_deleted_topics_archive_mark_is_dropped() { let score: Option = setup.raw.zscore(&pending, &log).await.unwrap(); assert_eq!(score, None); } + +#[tokio::test] +async fn a_large_trim_takes_no_longer_than_a_small_one() { + let Some(url) = url() else { return }; + let setup = setup_with(&url, &url, &unique("large-trim"), |options| { + options.retention = Duration::from_millis(500); + options.floor_lease = Some(Duration::from_secs(60)); + }) + .await; + let chain = chain(); + // A hold keeps 50,000 records of 1 KiB past the retention, then ends. + setup + .store + .hold_floor(&chain, "events", "reader-run/1", "") + .await + .unwrap(); + let body = vec![7u8; 1024]; + let mut sequence = 1; + while sequence <= 50_000 { + let mut batch = append(&chain, "p", sequence, 1000, &sequence.to_be_bytes()); + batch.records = vec![body.clone(); 1000]; + setup.store.append(batch).await.unwrap(); + sequence += 1000; + } + setup + .store + .release_floor(&chain, "events", "reader-run/1") + .await + .unwrap(); + let mut raw = setup.raw.clone(); + let floors = format!("{}:floors", log_key(&setup.prefix, &chain, "events")); + let _: () = raw.del(&floors).await.unwrap(); + tokio::time::sleep(Duration::from_millis(600)).await; + let started = tokio::time::Instant::now(); + setup + .store + .append(append(&chain, "p", sequence, 1, b"last")) + .await + .unwrap(); + let took = started.elapsed(); + let log = log_key(&setup.prefix, &chain, "events"); + let length: usize = raw.xlen(&log).await.unwrap(); + assert_eq!(length, 1, "the trim dropped every held record"); + // Copying 50,000 doomed records into Lua took about 80 ms. + assert!(took < Duration::from_millis(30), "{took:?}"); +}