From 6ecc371247258f4a3c32241bda8873c5ba70fc77 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Thu, 15 Oct 2026 20:00:05 -0700 Subject: [PATCH 1/5] Found the newest trimmed Redis record without a scan. The trim's XREVRANGE from the floor read the whole retained log when no record was older than the floor, so an append cost more the longer the log was. XRANGE forward from the oldest reads only what the trim drops. --- crates/sdk-core/CHANGELOG.md | 5 +++ crates/streams/docs/redis.md | 6 +++- crates/streams/lua/keep.lua | 14 +++++++-- crates/streams/src/redis/scripts.rs | 39 +++++++++++++++++++++--- crates/streams/tests/redis_store.rs | 47 +++++++++++++++++++++++++++++ 5 files changed, 102 insertions(+), 9 deletions(-) diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index 33f95676a0..75ed49c8dd 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 ad0cad99f7..8b7e36a32f 100644 --- a/crates/streams/docs/redis.md +++ b/crates/streams/docs/redis.md @@ -79,7 +79,11 @@ 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 +and nobody cleans up when the Workflow closes. The trim finds the newest record it drops with +`XRANGE` forward from the oldest, 1,000 at a time, so it reads only what it drops. The SDKs' +script used `XREVRANGE` back from the floor, which read the whole retained log when no record was +older than the floor, so an append cost more the longer the log was. 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 diff --git a/crates/streams/lua/keep.lua b/crates/streams/lua/keep.lua index 6e83456876..476a2ce8c0 100644 --- a/crates/streams/lua/keep.lua +++ b/crates/streams/lua/keep.lua @@ -28,10 +28,18 @@ local function keep(log, meta, added, last, retention, grace, 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 + local doomed + local from = '-' + repeat + local older = redis.call('XRANGE', log, from, '(' .. floor, 'COUNT', 1000) + if #older > 0 then + doomed = older[#older][1] + from = '(' .. doomed + end + until #older < 1000 + if doomed then redis.call('XTRIM', log, 'MINID', floor) - redis.call('HSET', meta, 'trimmed', doomed[1][1]) + redis.call('HSET', meta, 'trimmed', doomed) end redis.call('PEXPIRE', log, life) redis.call('HINCRBY', meta, 'added', added) diff --git a/crates/streams/src/redis/scripts.rs b/crates/streams/src/redis/scripts.rs index b5e9e9c8b0..3452c101b8 100644 --- a/crates/streams/src/redis/scripts.rs +++ b/crates/streams/src/redis/scripts.rs @@ -2,7 +2,9 @@ //! //! 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. `keep.lua` is the prelude of every script that writes a log, and +//! positions-only readers hold. Its trim also finds the newest record to drop by reading forward +//! from the oldest, since `XREVRANGE` from the floor read the whole retained log when no record +//! was older than the floor. `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; @@ -130,6 +132,30 @@ if topics then end "; + /// How the trim finds the newest record to drop, and what it replaces. + const TRIM_SCAN: (&str, &str) = ( + " local doomed + local from = '-' + repeat + local older = redis.call('XRANGE', log, from, '(' .. floor, 'COUNT', 1000) + if #older > 0 then + doomed = older[#older][1] + from = '(' .. doomed + end + until #older < 1000 + if doomed then + redis.call('XTRIM', log, 'MINID', floor) + redis.call('HSET', meta, 'trimmed', doomed) + 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); 5] = [ ( @@ -165,7 +191,7 @@ end assert_eq!(APPEND_LOGGED.matches("XADD', KEYS[5]").count(), 1); assert_eq!( APPEND_LOGGED_SCRIPT.get_hash(), - "418b9d100738aa50bc1ee5c1b7e61e62c8f9c434" + "eeed648a51793f36d1ed5f89c735dc8032d5fe7f" ); } @@ -184,20 +210,23 @@ end assert_eq!(was, *APPEND_LOGGED); assert_eq!( APPEND_ARCHIVED_SCRIPT.get_hash(), - "b9976d6b00aba4a108bca0afcbd125367e567aee" + "337494c09860e9e0bd9dbdf4a6db81a57719bd4c" ); } #[test] - fn the_append_script_matches_the_sdks_with_the_topic_check_floor_and_topic_set_only() { + fn the_append_script_matches_the_sdks_with_the_topic_check_floor_topic_set_and_scan_only() { assert_eq!( APPEND_SCRIPT.get_hash(), - "eec20388c2d5655389344177b25a640e1ba3f625" + "188afac5fc0de85f8a3f1ae86129a8e19b861cec" ); 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/redis_store.rs b/crates/streams/tests/redis_store.rs index 76c2e5a5e0..25bea3a440 100644 --- a/crates/streams/tests/redis_store.rs +++ b/crates/streams/tests/redis_store.rs @@ -2141,3 +2141,50 @@ async fn a_readers_short_hold_never_ends_the_archives_hold() { // The reader's lease ended, and the records the archive has not taken stay. 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" + ); +} From 7f2e9e2ca58b59bf410d9707150189f08023bf24 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Thu, 15 Oct 2026 20:58:59 -0700 Subject: [PATCH 2/5] Kept held records in the Redis conformance trim. The test trim now keeps what a live hold holds, as keep.lua does. Redis runs the archive trim case, and the tiered store cases run with no hold on every store. --- crates/streams/tests/conformance.rs | 65 +++++++++++++++++++++++++---- 1 file changed, 56 insertions(+), 9 deletions(-) diff --git a/crates/streams/tests/conformance.rs b/crates/streams/tests/conformance.rs index 08bbe00059..4eb61d6063 100644 --- a/crates/streams/tests/conformance.rs +++ b/crates/streams/tests/conformance.rs @@ -438,6 +438,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, @@ -737,17 +750,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; @@ -2717,8 +2751,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; @@ -3198,8 +3232,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 ))); @@ -3215,7 +3256,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 ))); From cc2044c30b223943d029c096016ec10e35a75cd7 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 16 Oct 2026 06:08:40 -0700 Subject: [PATCH 3/5] Read back from the trim floor only when a record is below it. The forward scan copied every doomed record into Lua, and a large trim blocked Redis for hundreds of milliseconds. --- crates/streams/docs/redis.md | 14 ++++----- crates/streams/lua/keep.lua | 17 ++++------- crates/streams/src/redis/scripts.rs | 33 +++++++++------------ crates/streams/tests/redis_store.rs | 46 +++++++++++++++++++++++++++++ 4 files changed, 73 insertions(+), 37 deletions(-) diff --git a/crates/streams/docs/redis.md b/crates/streams/docs/redis.md index bc1dc25ca4..0130e2b138 100644 --- a/crates/streams/docs/redis.md +++ b/crates/streams/docs/redis.md @@ -83,13 +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 trim finds the newest record it drops with -`XRANGE` forward from the oldest, 1,000 at a time, so it reads only what it drops. The SDKs' -script used `XREVRANGE` back from the floor, which read the whole retained log when no record was -older than the floor, so an append cost more the longer the log was. 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. +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 2be93400a8..e8aa12afec 100644 --- a/crates/streams/lua/keep.lua +++ b/crates/streams/lua/keep.lua @@ -28,18 +28,13 @@ local function trim(log, meta, retention, floors) local floor = floor_ms .. '-0' local life floor, life = held(floors, floor, retention) - local doomed - local from = '-' - repeat - local older = redis.call('XRANGE', log, from, '(' .. floor, 'COUNT', 1000) - if #older > 0 then - doomed = older[#older][1] - from = '(' .. doomed - end - until #older < 1000 - if doomed 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) + redis.call('HSET', meta, 'trimmed', doomed[1][1]) end return floor_ms, life end diff --git a/crates/streams/src/redis/scripts.rs b/crates/streams/src/redis/scripts.rs index 4b1468277e..665ec8c49c 100644 --- a/crates/streams/src/redis/scripts.rs +++ b/crates/streams/src/redis/scripts.rs @@ -2,11 +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. Its trim also finds the newest record to drop by reading forward -//! from the oldest, since `XREVRANGE` from the floor read the whole retained log when no record -//! was older than the floor. 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; @@ -134,18 +134,13 @@ end /// How the trim finds the newest record to drop, and what it replaces. const TRIM_SCAN: (&str, &str) = ( - " local doomed - local from = '-' - repeat - local older = redis.call('XRANGE', log, from, '(' .. floor, 'COUNT', 1000) - if #older > 0 then - doomed = older[#older][1] - from = '(' .. doomed - end - until #older < 1000 - if doomed 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) + redis.call('HSET', meta, 'trimmed', doomed[1][1]) end ", " local doomed = redis.call('XREVRANGE', log, '(' .. floor, '-', 'COUNT', 1) @@ -200,7 +195,7 @@ end assert_eq!(APPEND_LOGGED.matches("XADD', KEYS[5]").count(), 1); assert_eq!( APPEND_LOGGED_SCRIPT.get_hash(), - "8882324049920202484e88e6607768395cd93bd0" + "5ea5fe3e6a0a6812bf50a57906031cddb79c7fd2" ); } @@ -219,7 +214,7 @@ end assert_eq!(was, *APPEND_LOGGED); assert_eq!( APPEND_ARCHIVED_SCRIPT.get_hash(), - "84b8cfa23327bc39fc91a0201f307fd28fd4fac3" + "d1cba6fb10c613d027bab7de7e06c39782dc8db4" ); } @@ -227,7 +222,7 @@ end fn the_append_script_is_the_one_before_the_floor_with_the_floor_edits_only() { assert_eq!( APPEND_SCRIPT.get_hash(), - "861f910d9da400eaa7dc1a17b25a9881a23a4244" + "7f3b95cfa4e62a5421c0ac696a527b210da72e99" ); let mut before_floor = APPEND .strip_prefix(include_str!("../../lua/floor.lua")) diff --git a/crates/streams/tests/redis_store.rs b/crates/streams/tests/redis_store.rs index 720997e47b..6e8652656e 100644 --- a/crates/streams/tests/redis_store.rs +++ b/crates/streams/tests/redis_store.rs @@ -2479,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:?}"); +} From 618b628032cfa2825f32e90911d7f1c0d69752cc Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 16 Oct 2026 06:10:02 -0700 Subject: [PATCH 4/5] Lapsed the Redis archive hold at once in the lost-records case. The Redis case helper keeps live holds, and a release keeps the hold for its grace. --- crates/streams/tests/conformance.rs | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/crates/streams/tests/conformance.rs b/crates/streams/tests/conformance.rs index 0fd591998d..e37b838d65 100644 --- a/crates/streams/tests/conformance.rs +++ b/crates/streams/tests/conformance.rs @@ -780,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 { From 3e147ada392697de808c6fe8c02013fd784764df Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 16 Oct 2026 06:55:07 -0700 Subject: [PATCH 5/5] Wrapped the trim comment so its test copy fits 100 characters. --- crates/streams/lua/keep.lua | 4 ++-- crates/streams/src/redis/scripts.rs | 10 +++++----- 2 files changed, 7 insertions(+), 7 deletions(-) diff --git a/crates/streams/lua/keep.lua b/crates/streams/lua/keep.lua index e8aa12afec..c493ca811f 100644 --- a/crates/streams/lua/keep.lua +++ b/crates/streams/lua/keep.lua @@ -28,8 +28,8 @@ local function trim(log, meta, retention, floors) local floor = floor_ms .. '-0' local life floor, life = held(floors, floor, retention) - -- 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. + -- 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) diff --git a/crates/streams/src/redis/scripts.rs b/crates/streams/src/redis/scripts.rs index 665ec8c49c..fbcd307aa8 100644 --- a/crates/streams/src/redis/scripts.rs +++ b/crates/streams/src/redis/scripts.rs @@ -134,8 +134,8 @@ 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. + " -- 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) @@ -195,7 +195,7 @@ end assert_eq!(APPEND_LOGGED.matches("XADD', KEYS[5]").count(), 1); assert_eq!( APPEND_LOGGED_SCRIPT.get_hash(), - "5ea5fe3e6a0a6812bf50a57906031cddb79c7fd2" + "49820889a3c7680122a0b43ed0946c4b4cb0286e" ); } @@ -214,7 +214,7 @@ end assert_eq!(was, *APPEND_LOGGED); assert_eq!( APPEND_ARCHIVED_SCRIPT.get_hash(), - "d1cba6fb10c613d027bab7de7e06c39782dc8db4" + "98302a233c87f8a10303d4ca17b418cb51fe2799" ); } @@ -222,7 +222,7 @@ end fn the_append_script_is_the_one_before_the_floor_with_the_floor_edits_only() { assert_eq!( APPEND_SCRIPT.get_hash(), - "7f3b95cfa4e62a5421c0ac696a527b210da72e99" + "b818ae457ec5777a3a1d321e17215037cfdf5c6e" ); let mut before_floor = APPEND .strip_prefix(include_str!("../../lua/floor.lua"))