[Jetty] Found the newest trimmed Redis record without a scan.

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

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
 `{<namespace>:!a:<activity id>:<run id>}`, and every other part of the layout is the same. Each

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

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,20 +214,23 @@ end
         assert_eq!(was, *APPEND_LOGGED);
         assert_eq!(
             APPEND_ARCHIVED_SCRIPT.get_hash(),
-            "bb8e042129e393c24f89efa5b81339d87090e8c1"
+            "98302a233c87f8a10303d4ca17b418cb51fe2799"
         );
     }
 
     #[test]
     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 {

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<RedisStore>,
         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<Redis> {
@@ -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<String> = 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::<u128>().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
     )));

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<f64> = 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:?}");
+}