feat: count a Stream's Retention from its close

crates/streams/src/memory.rs

@@ -41,7 +41,7 @@ struct Keys {
     /// `None` when the log key doesn't exist.
     log: Option<BTreeMap<(u64, u64), Fields>>,
     log_expires: i64,
-    /// The newest id of the log, which Redis keeps after a trim and drops with the expired key.
+    /// The newest id of the log, which Redis drops with the expired key.
     last_id: (u64, u64),
     /// The `producers` sorted set, each `<kind>|<producer id>` with the ms of its last append.
     producers: HashMap<String, i64>,
@@ -234,16 +234,15 @@ impl Keys {
         self.set("exp", exp.to_string());
         self.meta_expires = exp.saturating_add(tombstone);
         if self.log.is_some() {
-            // A read hides every Record one Retention after the last write, so the log goes then.
-            self.log_expires = exp.min(now.saturating_add(retention));
+            self.log_expires = exp;
         }
         if !self.producers.is_empty() {
             self.producers_expires = exp;
         }
     }
 
     /// Removes the `p|` field of the Producers whose last append is older than the Retention, as
-    /// `prune` in `lib.lua`. Their Records are older too, so they hold none.
+    /// `prune` in `lib.lua`.
     fn prune(&mut self, now: i64) {
         let retention = self.number("ret");
         let min = now - retention;
@@ -272,31 +271,14 @@ impl Keys {
         now.max(self.newest_id().0 as i64)
     }
 
-    fn trim(&mut self, now: i64) {
-        if self.log.is_none() {
-            // The expired log took every Record, so a read from an older offset must fail.
-            if let Some(newest) = self.text("newest").and_then(parse_id) {
-                self.raise_trimmed(newest);
-            }
+    /// Raises the trimmed offset to `newest` when the log is gone, as `mark_log_gone` in `lib.lua`.
+    /// The log took every Record with it, so a read from an older offset must fail.
+    fn mark_log_gone(&mut self) {
+        if self.log.is_some() {
             return;
         }
-        let retention = self.number("ret");
-        let min = now - retention;
-        if min <= 0 {
-            return;
-        }
-        let Some(log) = self.log.as_ref() else {
-            return;
-        };
-        let old: Vec<_> = log.range(..(min as u64, 0)).map(|(id, _)| *id).collect();
-        let Some(newest_old) = old.last().copied() else {
-            return;
-        };
-        self.raise_trimmed(newest_old);
-        if let Some(log) = self.log.as_mut() {
-            for id in old {
-                log.remove(&id);
-            }
+        if let Some(newest) = self.text("newest").and_then(parse_id) {
+            self.raise_trimmed(newest);
         }
     }
 
@@ -445,6 +427,16 @@ impl MemoryStore {
             keys.log.as_ref().map_or(0, BTreeMap::len)
         })
     }
+
+    /// Deletes the log, as `DEL`, for tests.
+    #[doc(hidden)]
+    pub fn test_drop_log(&self, namespace: &str, stream_id: &str) {
+        self.with(namespace, stream_id, |keys, _| {
+            keys.log = None;
+            keys.log_expires = 0;
+            keys.last_id = (0, 0);
+        });
+    }
 }
 
 fn offsets(first: &str, last: &str) -> StreamResult<Option<(i64, i64)>> {
@@ -575,7 +567,7 @@ impl Store for MemoryStore {
             let records = batch.records.len() as u64;
             keys.check_ids_fit(now, records)?;
 
-            keys.trim(now);
+            keys.mark_log_gone();
             keys.prune(now);
             let at = keys.next_ms(now);
             let (mut first, mut last) = (String::new(), String::new());

crates/streams/src/service.rs

@@ -448,8 +448,8 @@ impl StreamService {
         from: i64,
         (max, max_bytes): (usize, usize),
     ) -> StreamResult<Option<ReadRecordsResponse>> {
-        // A read from the start skips the Records that the Retention hides. A read from an offset
-        // fails on them, since the Reader asked for those Records.
+        // A read from the start skips the Records that the log lost. A read from an offset fails on
+        // them, since the Reader asked for those Records.
         let retained = state.retained_from();
         let mut start = if from > 0 { from } else { retained };
         let mut records = vec![];
@@ -466,7 +466,7 @@ impl StreamService {
             let room = (max_bytes != usize::MAX).then(|| max_bytes.saturating_sub(bytes));
             let chunk = chunk_len(max - records.len(), room, pulled);
             let (chunk_state, entries) = self.store.read(namespace, id, start, chunk).await?;
-            // A trim after the state read leaves a gap before these entries.
+            // Records that the log lost after the state read leave a gap before these entries.
             check_expiry(&chunk_state, if from > 0 { start } else { 0 })?;
             end = (chunk_state.closed, chunk_state.newest);
             close_status = chunk_state.close_status;
@@ -562,7 +562,7 @@ fn decode(bytes: &[u8], offset: i64) -> StreamResult<StreamRecord> {
     Ok(record)
 }
 
-/// Fails a read whose Records the Retention or the expiry removed, so that the Reader always sees
+/// Fails a read whose Records the log lost or the expiry removed, so that the Reader always sees
 /// the gap. A read from the start of the Stream has no gap.
 ///
 /// A Tombstone fails every read. An open Stream that expired gets no more Records, so the Reader

crates/streams/src/store.rs

@@ -125,7 +125,7 @@ pub struct StreamState {
     pub close_status: i32,
     /// The offset of the newest entry written, zero if none.
     pub newest: i64,
-    /// The offset of the newest entry that Retention dropped, zero if none.
+    /// The newest offset when a write found the log gone, zero if none.
     pub trimmed: i64,
     /// The Retention.
     pub retention: Duration,
@@ -141,15 +141,10 @@ impl StreamState {
         self.exists && self.expires_ms <= self.now_ms
     }
 
-    /// The offset below which Retention dropped or hides the Records, as of the read.
+    /// The offset below which the log lost the Records. The Retention counts from the close, so
+    /// it hides no Record of a Stream that hasn't expired.
     pub fn retained_from(&self) -> i64 {
-        let cut = self.now_ms - self.retention.as_millis() as i64;
-        let hidden = if cut > 0 {
-            cut << crate::offset::SEQ_BITS
-        } else {
-            0
-        };
-        hidden.max(self.trimmed + 1)
+        self.trimmed + 1
     }
 }
 
@@ -230,9 +225,9 @@ fn unsupported(capability: &str) -> StreamError {
 /// The calls of a Store. Each one is atomic, as one Lua script is on Redis.
 ///
 /// The required calls make the log contract. They are a create key, an atomic batch append, an
-/// ordered read with the Stream's state, stable offsets, prefix Retention and a wait for the next
-/// entry. An append applies the rules of the capabilities that the Store declares. The optional
-/// calls fail by default.
+/// ordered read with the Stream's state, stable offsets, a Retention that counts from the close and
+/// a wait for the next entry. An append applies the rules of the capabilities that the Store
+/// declares. The optional calls fail by default.
 #[async_trait::async_trait]
 pub trait Store: Send + Sync + 'static {
     /// What the Store does atomically beyond the log. It has no default, so a Store that wraps

crates/streams/src/testing/mod.rs

@@ -33,6 +33,8 @@ pub trait RawKeys: Send + Sync {
     async fn add_raw_entry(&self, namespace: &str, stream: &str, id: &str);
     /// The number of log entries.
     async fn log_len(&self, namespace: &str, stream: &str) -> usize;
+    /// Deletes the log, as an eviction would.
+    async fn drop_log(&self, namespace: &str, stream: &str);
 }
 
 /// A backend that the suites run on.
@@ -92,6 +94,10 @@ impl RawKeys for MemoryStore {
     async fn log_len(&self, namespace: &str, stream: &str) -> usize {
         self.test_log_len(namespace, stream)
     }
+
+    async fn drop_log(&self, namespace: &str, stream: &str) {
+        self.test_drop_log(namespace, stream)
+    }
 }
 
 /// One case of a suite.

crates/streams/src/testing/service.rs

@@ -693,55 +693,33 @@ async fn a_read_of_an_expired_closed_stream_fails_instead_of_waiting() {
     }
 }
 
-/// Creates the case's Stream with a 300 ms Retention. While it is open, its meta stays and only
-/// its log expires.
-async fn create_short(case: &Case) {
-    case.create(millis(300)).await.unwrap();
-}
-
-/// A read from an offset that an idle open Stream's expired log took fails as expired at once,
-/// before the next write and after it. A read from the start skips the gap.
-async fn a_read_behind_an_expired_log_fails_before_and_after_the_next_write() {
+/// An open Stream keeps its Records past its Retention, which counts from the close. So a read
+/// gets every Record and the close item until one Retention after the close, and then the Stream
+/// is expired.
+async fn the_retention_counts_from_the_close() {
     for case in cases_for(Needs::SHORT_TIMES).await {
-        create_short(&case).await;
+        case.create(millis(300)).await.unwrap();
         let first = case
             .append(client(1), 1, vec![record("a"), record("b")])
             .await
             .unwrap();
         tokio::time::sleep(Duration::from_millis(400)).await;
-        let behind = ReadRecordsRequest {
+        let after_first = ReadRecordsRequest {
             from_offset: first.first_offset.clone(),
-            wait: millis(1500),
             ..Default::default()
         };
-        let started = Instant::now();
-        let error = case.read(behind.clone()).await.unwrap_err();
-        assert_eq!(reason(&error), "stream_expired");
-        assert!(started.elapsed() < Duration::from_secs(1));
+        let read = case.read(after_first.clone()).await.unwrap();
+        assert_eq!(texts(&read), ["b"]);
         case.append(client(1), 2, vec![record("c")]).await.unwrap();
-        let error = case.read(behind).await.unwrap_err();
-        assert_eq!(reason(&error), "stream_expired");
+        case.close(client(1)).await.unwrap();
         let read = case.read(ReadRecordsRequest::default()).await.unwrap();
-        assert_eq!(texts(&read), ["c"]);
-    }
-}
-
-/// A close after an idle open Stream's log expired keeps the gap, so a read from an offset that the
-/// log took fails as expired.
-async fn a_close_after_the_log_expired_keeps_the_gap() {
-    for case in cases_for(Needs::SHORT_TIMES).await {
-        create_short(&case).await;
-        let first = case
-            .append(client(1), 1, vec![record("a"), record("b")])
-            .await
-            .unwrap();
+        assert_eq!(texts(&read), ["a", "b", "c", "<close>"]);
+        let describe = case.describe().await.unwrap();
+        assert_eq!(describe.status, StreamStatus::Closed as i32);
         tokio::time::sleep(Duration::from_millis(400)).await;
-        case.close(client(1)).await.unwrap();
-        let behind = ReadRecordsRequest {
-            from_offset: first.first_offset.clone(),
-            ..Default::default()
-        };
-        let error = case.read(behind).await.unwrap_err();
+        let describe = case.describe().await.unwrap();
+        assert_eq!(describe.status, StreamStatus::Expired as i32);
+        let error = case.read(after_first).await.unwrap_err();
         assert_eq!(reason(&error), "stream_expired");
     }
 }
@@ -949,8 +927,7 @@ pub const CASES: &[super::Case] = &[
     case!(a_closed_read_ends_with_a_close_item),
     case!(an_unknown_method_is_unimplemented),
     case!(a_read_of_an_expired_closed_stream_fails_instead_of_waiting),
-    case!(a_read_behind_an_expired_log_fails_before_and_after_the_next_write),
-    case!(a_close_after_the_log_expired_keeps_the_gap),
+    case!(the_retention_counts_from_the_close),
     case!(a_create_answer_names_the_backend_and_the_store_never_reads_it),
     case!(a_read_of_a_closed_stream_with_no_item_left_answers_at_once),
     case!(a_waiting_read_wakes_with_done_on_a_close),

crates/streams/src/testing/store.rs

@@ -297,19 +297,17 @@ async fn an_expired_stream_refuses_writes_and_its_create() {
     }
 }
 
-/// An open Stream's meta keeps the safety TTL, and its log keeps the Retention, after which a read
-/// hides every Record. Its close gives the meta the Retention too.
+/// An open Stream's meta and log keep the safety TTL, so the log keeps its Records past the
+/// Retention. Its close gives both the Retention, and the meta the Tombstone after it.
 async fn the_ttl_is_the_safety_ttl_until_the_close() {
     for target in stores_for(Needs::RAW_KEYS).await {
         target.create().await;
         append(&target, &client(1), 1, &["a"]).await.unwrap();
         let week = Duration::from_secs(7 * 86_400).as_millis() as i64;
+        let safety = Duration::from_secs(80 * 86_400).as_millis() as i64;
         let (ttl, meta) = target.ttls().await;
-        assert!(0 < ttl && ttl <= week, "{ttl}");
-        assert!(
-            meta > Duration::from_secs(80 * 86_400).as_millis() as i64,
-            "{meta}"
-        );
+        assert!(ttl > safety, "{ttl}");
+        assert!(meta > safety, "{meta}");
         target.close(&client(1)).await.unwrap();
         let (ttl, meta) = target.ttls().await;
         assert!(0 < ttl && ttl <= week, "{ttl}");
@@ -320,52 +318,41 @@ async fn the_ttl_is_the_safety_ttl_until_the_close() {
     }
 }
 
-/// A read hides Records past the Retention, and the next append trims them and keeps the
-/// newest trimmed offset.
-async fn an_append_trims_by_retention_and_keeps_the_watermark() {
+/// An open Stream keeps every Record past its Retention, while it is idle and after its next
+/// append. No append trims a Record and no read hides one, since the Retention counts from the
+/// close.
+async fn an_open_stream_keeps_its_records_past_its_retention() {
     for target in stores_for(Needs::RAW_KEYS.and(Needs::SHORT_TIMES)).await {
+        let retention = Duration::from_millis(300);
         target
             .create_with(&Create {
-                retention: Some(Duration::from_millis(600)),
+                retention: Some(retention),
                 ..Default::default()
             })
             .await
             .unwrap();
         let old = append(&target, &client(1), 1, &["old"]).await.unwrap();
         tokio::time::sleep(Duration::from_millis(400)).await;
-        // This append keeps the log past the old Record's Retention.
-        append(&target, &client(1), 2, &["mid"]).await.unwrap();
-        tokio::time::sleep(Duration::from_millis(350)).await;
-        // The read hides the old Record before any append trims it.
-        let (state, entries) = target
-            .store
-            .read(&target.namespace, &target.stream, 0, 10)
-            .await
-            .unwrap();
-        assert!(state.retained_from() > old.offsets.unwrap().1);
-        assert_eq!(entries.len(), 2);
-        append(&target, &client(1), 3, &["new"]).await.unwrap();
+        assert_eq!(target.texts().await, ["old"]);
+        append(&target, &client(1), 2, &["new"]).await.unwrap();
         let (state, entries) = target
             .store
             .read(&target.namespace, &target.stream, 0, 10)
             .await
             .unwrap();
-        assert_eq!(state.trimmed, old.offsets.unwrap().1);
-        assert_eq!(records_of(&entries), ["mid", "new"]);
+        assert_eq!(state.trimmed, 0);
+        assert!(state.retained_from() <= old.offsets.unwrap().0);
+        assert_eq!(records_of(&entries), ["old", "new"]);
+        let (ttl, _) = target.ttls().await;
+        assert!(ttl > retention.as_millis() as i64, "{ttl}");
     }
 }
 
-/// The log expires one Retention after the last write. The next write raises the trimmed offset
-/// to the meta's newest one, and its offsets go above it, also with the clock behind it.
-async fn offsets_go_on_after_the_log_expires() {
-    for target in stores_for(Needs::RAW_KEYS.and(Needs::SHORT_TIMES)).await {
-        target
-            .create_with(&Create {
-                retention: Some(Duration::from_millis(200)),
-                ..Default::default()
-            })
-            .await
-            .unwrap();
+/// A write that finds the log gone, as after an eviction, raises the trimmed offset to the meta's
+/// newest one. Its offsets go above it, also with the clock behind it.
+async fn offsets_go_on_after_the_log_is_gone() {
+    for target in stores_for(Needs::RAW_KEYS).await {
+        target.create().await;
         append(&target, &client(1), 1, &["old"]).await.unwrap();
         // A newest id one minute ahead, as after the clock steps back.
         let now = std::time::SystemTime::now()
@@ -374,7 +361,7 @@ async fn offsets_go_on_after_the_log_expires() {
             .as_millis();
         let ahead = format!("{}-3", now + 60_000);
         target.set_meta("newest", &ahead).await;
-        tokio::time::sleep(Duration::from_millis(300)).await;
+        target.drop_log().await;
         assert_eq!(target.ttls().await.0, -2);
         let new = append(&target, &client(1), 2, &["new"]).await.unwrap();
         let ahead = offset_of_entry_id(&ahead).unwrap();
@@ -557,8 +544,8 @@ async fn a_large_sequence_dedupes_and_one_past_2_53_is_refused() {
     }
 }
 
-/// An append removes the `p|` field of a Producer whose last append is older than the Retention,
-/// since its Records are gone too. A lower epoch of that Producer then writes.
+/// An append removes the `p|` field of a Producer whose last append is older than the Retention.
+/// A lower epoch of that Producer then writes, and the open Stream keeps the older Records.
 async fn an_append_forgets_the_producers_that_the_retention_outlived() {
     for target in stores_for(Needs::RAW_KEYS.and(Needs::SHORT_TIMES)).await {
         target
@@ -576,7 +563,7 @@ async fn an_append_forgets_the_producers_that_the_retention_outlived() {
         assert_eq!(target.meta(fields::producer(kind, &old.id)).await, None);
         assert!(target.meta(fields::producer(kind, &new.id)).await.is_some());
         append(&target, &client(1), 1, &["late"]).await.unwrap();
-        assert_eq!(target.texts().await, ["new", "late"]);
+        assert_eq!(target.texts().await, ["old", "new", "late"]);
     }
 }
 
@@ -699,8 +686,8 @@ pub const CASES: &[super::Case] = &[
     case!(a_newer_format_refuses_every_write),
     case!(an_expired_stream_refuses_writes_and_its_create),
     case!(the_ttl_is_the_safety_ttl_until_the_close),
-    case!(an_append_trims_by_retention_and_keeps_the_watermark),
-    case!(offsets_go_on_after_the_log_expires),
+    case!(an_open_stream_keeps_its_records_past_its_retention),
+    case!(offsets_go_on_after_the_log_is_gone),
     case!(a_blocked_read_wakes_on_an_append_and_ends_at_its_deadline),
     case!(a_close_writes_no_entry_and_ends_a_wait),
     case!(a_batch_gets_consecutive_offsets_that_end_at_its_last_offset),