diff --git a/crates/streams/src/memory.rs b/crates/streams/src/memory.rs index 2d2676530..9b2b3f2d1 100644 --- a/crates/streams/src/memory.rs +++ b/crates/streams/src/memory.rs @@ -41,7 +41,7 @@ struct Keys { /// `None` when the log key doesn't exist. log: Option>, 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 `|` with the ms of its last append. producers: HashMap, @@ -234,8 +234,7 @@ 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; @@ -243,7 +242,7 @@ impl Keys { } /// 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> { @@ -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()); diff --git a/crates/streams/src/service.rs b/crates/streams/src/service.rs index cb1b85458..12d49f141 100644 --- a/crates/streams/src/service.rs +++ b/crates/streams/src/service.rs @@ -448,8 +448,8 @@ impl StreamService { from: i64, (max, max_bytes): (usize, usize), ) -> StreamResult> { - // 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 { 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 diff --git a/crates/streams/src/store.rs b/crates/streams/src/store.rs index 7a0d44961..ccc543be8 100644 --- a/crates/streams/src/store.rs +++ b/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 diff --git a/crates/streams/src/testing/mod.rs b/crates/streams/src/testing/mod.rs index 94b3b5c85..917b3a769 100644 --- a/crates/streams/src/testing/mod.rs +++ b/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. diff --git a/crates/streams/src/testing/service.rs b/crates/streams/src/testing/service.rs index 4cb24998d..76d97c293 100644 --- a/crates/streams/src/testing/service.rs +++ b/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", ""]); + 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), diff --git a/crates/streams/src/testing/store.rs b/crates/streams/src/testing/store.rs index 27fde6818..4924bf598 100644 --- a/crates/streams/src/testing/store.rs +++ b/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), diff --git a/crates/streams/src/testing/support.rs b/crates/streams/src/testing/support.rs index 8cf34d010..6faea071e 100644 --- a/crates/streams/src/testing/support.rs +++ b/crates/streams/src/testing/support.rs @@ -248,6 +248,11 @@ impl Target { self.raw.log_len(&self.namespace, &self.stream).await } + /// Deletes the log, as an eviction would. + pub(crate) async fn drop_log(&self) { + self.raw.drop_log(&self.namespace, &self.stream).await + } + /// Every entry of the log. pub(crate) async fn entries(&self) -> Vec { let (_, entries) = self