From abc195054d3fb1c316c6a8ead86985e29f622c17 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 16 Oct 2026 13:59:19 -0700 Subject: [PATCH 1/3] Mapped holds, the change feed and closes with an outcome onto store capabilities. A positions-only read now needs holds or a declared replay horizon, a replay reads up to its recorded end, and a store without close flags keeps a close's outcome in its CLOSE record. --- .../temporal/sdk/streams/v1/message.proto | 7 + crates/protos/src/protos/mod.rs | 2 + crates/streams/docs/redis.md | 4 +- crates/streams/src/close.rs | 23 ++- crates/streams/src/handle.rs | 172 ++++++++++++++++-- crates/streams/src/memory.rs | 57 ++++-- crates/streams/src/reader.rs | 31 +++- crates/streams/src/redis/mod.rs | 34 +++- crates/streams/src/service.rs | 6 +- crates/streams/src/store.rs | 14 +- crates/streams/tests/conformance.rs | 11 +- 11 files changed, 303 insertions(+), 58 deletions(-) diff --git a/crates/protos/protos/local/temporal/sdk/streams/v1/message.proto b/crates/protos/protos/local/temporal/sdk/streams/v1/message.proto index f50347e05..5964322ec 100644 --- a/crates/protos/protos/local/temporal/sdk/streams/v1/message.proto +++ b/crates/protos/protos/local/temporal/sdk/streams/v1/message.proto @@ -5,6 +5,7 @@ package temporal.sdk.streams.v1; option ruby_package = "Temporalio::Internal::Bridge::Api::StreamRecord"; import "temporal/api/common/v1/message.proto"; +import "temporal/api/failure/v1/message.proto"; import "temporal/api/stream/v1/message.proto"; // One entry in a stream. The record is the wire format: stores keep it @@ -44,6 +45,12 @@ message StreamRecord { // A trailing zero means the list goes back to the run's start. A reader // learns from it whether an earlier attempt it delivered completed. repeated int64 completed_tasks = 10; + // On a CLOSE, the result the stream closed with, after the payload codec. + // `producer_id` names who closed it. A CLOSE with no result and no + // failure closed with no outcome. + temporal.api.common.v1.Payload close_result = 11; + // On a CLOSE, the failure the stream closed with. + temporal.api.failure.v1.Failure close_failure = 12; } // What a record means to a reader. diff --git a/crates/protos/src/protos/mod.rs b/crates/protos/src/protos/mod.rs index 679ad6e6f..04fec75e5 100644 --- a/crates/protos/src/protos/mod.rs +++ b/crates/protos/src/protos/mod.rs @@ -3103,6 +3103,8 @@ mod sdk_helpers { link: None, task_started_event_id: 0, completed_tasks: vec![], + close_result: None, + close_failure: None, } } diff --git a/crates/streams/docs/redis.md b/crates/streams/docs/redis.md index aab9b6c69..0767245f0 100644 --- a/crates/streams/docs/redis.md +++ b/crates/streams/docs/redis.md @@ -139,7 +139,9 @@ write, and `+xreadgroup +xautoclaim +xack +xdel +xgroup|create +xlen +xadd +clus kept. So the store declares neither producer state nor a kept last position, and Core takes their fallbacks: readers drop copies and end at the first `CLOSE` record, and a read decides expiry from the record at its cursor. It exists so tests run those fallbacks on a real store. -Listing a chain's topics scans every primary, as a delete does, and needs `+scan`. +Listing a chain's topics scans every primary, as a delete does, and needs `+scan`. With no script +there is no notification log, no floor hold and no archive hold, so plain mode declares neither +holds nor a change feed, and a config that asks for a log or an archive is refused. **Tests.** The Redis tests run when `STREAMS_REDIS_URL` names a standalone server and `STREAMS_REDIS_CLUSTER_URL` a cluster seed. `scripts/redis-test-servers.sh` starts both, the diff --git a/crates/streams/src/close.rs b/crates/streams/src/close.rs index 276af80c8..bc30d9bd5 100644 --- a/crates/streams/src/close.rs +++ b/crates/streams/src/close.rs @@ -7,8 +7,8 @@ use crate::{ StreamResult, StreamStore, close_record, proto::{ - ChainId, StoreAppendRequest, StoreAppendResponse, StoreReadRequest, StreamRecord, - StreamRecordKind, + ChainId, StoreAppendRequest, StoreAppendResponse, StoreReadRequest, StoredClose, + StreamRecord, StreamRecordKind, }, topic_record, }; @@ -108,9 +108,24 @@ async fn chain_topics(store: &dyn StreamStore, chain: &ChainId) -> StreamResult< Ok(topics) } +/// Appends a `CLOSE` record for `topic` that carries `close`'s outcome and names who closed it. +pub(crate) async fn write_close( + store: &S, + chain: &ChainId, + topic: &str, + close: &StoredClose, +) -> StreamResult<()> { + let record = StreamRecord { + close_result: close.result.clone(), + close_failure: close.failure.clone(), + ..close_record(topic, &close.closed_by, 0) + }; + write_one(store, chain, topic, record).await +} + /// Appends one control record. Its sequence is zero, so no reader takes it for a copy. -async fn write_one( - store: &dyn StreamStore, +async fn write_one( + store: &S, chain: &ChainId, topic: &str, record: StreamRecord, diff --git a/crates/streams/src/handle.rs b/crates/streams/src/handle.rs index 1be531bca..5f461094a 100644 --- a/crates/streams/src/handle.rs +++ b/crates/streams/src/handle.rs @@ -16,7 +16,7 @@ use crate::{ StoredClose, StreamAddress, StreamFailureKind, StreamOwnerKind, append_request, read_record, retire_producer_request, }, - reader::{ReadState, read_with}, + reader::{ReadState, read_bounded, read_with}, stored_append_record, stream_hash, }; use prost::Message; @@ -434,7 +434,19 @@ impl Streams { } /// Holds the records after `after` against retention for `holder`, a positions-only reader. + /// A store that can't hold serves the reader only when it declares a replay horizon, and + /// refuses it otherwise, so the read fails at once and not at replay. async fn hold_floor(&self, target: &ReadTarget, after: &str, holder: &str) -> StreamResult<()> { + let capabilities = self.store.capabilities(); + if !capabilities.holds { + if capabilities.replay_horizon.is_some() { + return Ok(()); + } + return Err(StreamError::unsupported( + "a positions-only read needs a store that holds records for a replay or declares \ + a replay horizon, and this store does neither", + )); + } if holder.len() > MAX_FLOOR_HOLDER_BYTES { return Err(StreamError::invalid_argument(format!( "a floor holder is at most {MAX_FLOOR_HOLDER_BYTES} bytes, not {}", @@ -511,23 +523,57 @@ impl Streams { ..Default::default() } } else { - let page_request = ReadRequest { - stream: Some(stream.clone()), - after: request.after.clone(), - max_records: request.stored_records, - ..Default::default() - }; - match read_with(&*self.store, &target, &page_request, &mut state).await { - Ok((page, _, _)) => page, - Err(error) - if matches!( - error.kind, - StreamFailureKind::Expired | StreamFailureKind::NotFound - ) => + // Up to the recorded end, not a record count, since copies and control records the + // read dropped can put the store's page edges elsewhere. + let mut records = vec![]; + let mut after = request.after.clone(); + let mut start_cursor = None; + loop { + let page_request = ReadRequest { + stream: Some(stream.clone()), + after: after.clone(), + max_records: request.stored_records.max(1), + ..Default::default() + }; + let page = match read_bounded( + &*self.store, + &target, + &page_request, + &mut state, + Some(&request.to), + ) + .await { - return Err(lost(error.message)); + Ok((page, _, _)) => page, + Err(error) + if matches!( + error.kind, + StreamFailureKind::Expired | StreamFailureKind::NotFound + ) => + { + return Err(lost(error.message)); + } + Err(error) => return Err(error), + }; + start_cursor.get_or_insert(page.start_cursor); + let moved = page.cursor != after; + records.extend(page.records); + after = page.cursor; + let stored = records + .iter() + .filter(|record| matches!(record.record, Some(read_record::Record::Stored(_)))) + .count(); + if after == request.to || !moved || stored > expected { + break; } - Err(error) => return Err(error), + } + ReadResponse { + cursor: after, + records_digest: crate::reader::records_digest(&records), + records, + chain_first_run_id: request.chain_first_run_id.clone(), + start_cursor: start_cursor.unwrap_or_default(), + ..Default::default() } }; let stored = page @@ -541,6 +587,13 @@ impl Streams { request.after ))); } + if stored > expected { + return Err(mismatch(format!( + "the replayed read holds {stored} records up to {:?}, not the {expected} the read \ + got", + request.to + ))); + } if page.cursor != request.to || page.records_digest != request.records_digest { // A read from the beginning has no position to check, so a trim since the read shows // as other records, which end elsewhere or differ. @@ -2391,6 +2444,93 @@ mod tests { } } + #[tokio::test] + async fn a_replay_reads_to_the_recorded_end_past_dropped_copies() { + let setup = setup_on(without_producer_state(), StreamsOptions::default()); + setup.streams.append(append("p", 1, "1")).await.unwrap(); + // The answer was lost, so the producer sent the batch again. + setup.streams.append(append("p", 1, "1")).await.unwrap(); + setup.streams.append(append("p", 2, "2")).await.unwrap(); + let request = ReadRequest { + stream: Some(stream("out")), + max_records: 3, + ..Default::default() + }; + let page = setup.streams.read(request.clone()).await.unwrap(); + assert_eq!(page.records.len(), 2); + let replayed = setup + .streams + .read_range(replay_of(&request, &page)) + .await + .unwrap(); + assert_eq!(replayed.cursor, page.cursor); + assert_eq!(replayed.records_digest, page.records_digest); + assert_eq!(replayed.records, page.records); + } + + #[tokio::test] + async fn a_positions_only_read_needs_holds_or_a_replay_horizon() { + let no_holds = crate::Capabilities { + holds: false, + ..MemoryStore::new().capabilities() + }; + let request = ReadRequest { + stream: Some(stream("out")), + floor_holder: "reader-run/1".to_string(), + ..Default::default() + }; + let setup = setup_on( + MemoryStore::with_capabilities(no_holds), + StreamsOptions::default(), + ); + setup.streams.append(append("p", 1, "1")).await.unwrap(); + let error = setup.streams.read(request.clone()).await.unwrap_err(); + assert_eq!(error.kind, StreamFailureKind::Unsupported); + let with_horizon = crate::Capabilities { + replay_horizon: Some(Duration::from_secs(30 * 24 * 60 * 60)), + ..no_holds + }; + let setup = setup_on( + MemoryStore::with_capabilities(with_horizon), + StreamsOptions::default(), + ); + setup.streams.append(append("p", 1, "1")).await.unwrap(); + let page = setup.streams.read(request).await.unwrap(); + assert_eq!(page.records.len(), 1); + } + + #[tokio::test] + async fn without_close_flags_a_close_record_carries_its_outcome_to_readers() { + let setup = setup_on(without_producer_state(), quick()); + setup.streams.append(append("p", 1, "1")).await.unwrap(); + let result = Payload { + data: b"done".to_vec(), + ..Default::default() + }; + setup + .streams + .close(CloseRequest { + stream: Some(stream("out")), + result: Some(result.clone()), + ..Default::default() + }) + .await + .unwrap(); + // A second close lands too, and readers keep the first. + setup + .streams + .close(CloseRequest { + stream: Some(stream("out")), + ..Default::default() + }) + .await + .unwrap(); + let (delivered, end) = read_to_end(&setup.streams, Duration::from_secs(5)).await; + assert_eq!(delivered, 1); + assert!(end.done); + assert_eq!(end.close_result, Some(result)); + } + fn attempt(producer: &str, attempt: i64, sequence: i64, value: &str) -> AppendRequest { AppendRequest { producer: Some(Producer::Named(NamedProducer { diff --git a/crates/streams/src/memory.rs b/crates/streams/src/memory.rs index 6a37f52d6..3498eea1a 100644 --- a/crates/streams/src/memory.rs +++ b/crates/streams/src/memory.rs @@ -35,8 +35,8 @@ const FULL: Capabilities = Capabilities { fence: false, }, kept_last: true, - holds: false, - change_feed: false, + holds: true, + change_feed: true, cheap_logs: true, delete_owner: true, topic_listing: true, @@ -321,15 +321,16 @@ impl MemoryStore { } } - /// An empty store with only the capabilities both it and `mask` have, and the limits of - /// `mask`. Without producer state it keeps no dedupe entries and no close flags, without a - /// kept last position its reads never fail as expired or not found, and without a long poll - /// its reads never wait. + /// An empty store with only the capabilities both it and `mask` have, and the limits and + /// the replay horizon of `mask`. Without producer state it keeps no dedupe entries and no + /// close flags, without a kept last position its reads never fail as expired or not found, + /// and without a long poll its reads never wait. pub fn with_capabilities(mask: Capabilities) -> Self { Self { capabilities: Capabilities { max_record_bytes: mask.max_record_bytes, max_batch_bytes: mask.max_batch_bytes, + replay_horizon: mask.replay_horizon, ..FULL.masked(&mask) }, ..Self::default() @@ -358,6 +359,15 @@ impl MemoryStore { topic.base += drop as u64; } + fn holds(&self) -> StreamResult<()> { + if self.capabilities.holds { + return Ok(()); + } + Err(StreamError::unsupported( + "this store can't hold records for a positions-only reader", + )) + } + fn producer_state(&self) -> StreamResult<()> { if self.capabilities.producer_state { return Ok(()); @@ -467,7 +477,10 @@ impl StreamStore for MemoryStore { } fn capabilities(&self) -> Capabilities { - self.capabilities + Capabilities { + change_feed: self.capabilities.change_feed && self.log.is_some(), + ..self.capabilities + } } async fn append(&self, request: StoreAppendRequest) -> StreamResult { @@ -657,6 +670,7 @@ impl StreamStore for MemoryStore { holder: &str, position: &str, ) -> StreamResult<()> { + self.holds()?; let position = parse_position(position)?; let mut chains = self.chains.lock().unwrap(); let holders = chains @@ -678,6 +692,7 @@ impl StreamStore for MemoryStore { /// The memory store has no clock for retention, so a release ends the hold at once. async fn release_floor(&self, chain: &ChainId, topic: &str, holder: &str) -> StreamResult<()> { + self.holds()?; if let Some(holders) = self .chains .lock() @@ -699,6 +714,7 @@ impl StreamStore for MemoryStore { topic: &str, position: &str, ) -> StreamResult<()> { + self.holds()?; if !self.archive_hold { return Err(StreamError::unsupported("this store keeps no archive hold")); } @@ -761,7 +777,10 @@ impl StreamStore for MemoryStore { topic: &str, close: StoredClose, ) -> StreamResult { - self.producer_state()?; + if !self.capabilities.producer_state { + crate::close::write_close(self, chain, topic, &close).await?; + return Ok(close); + } let stored = { let mut chains = self.chains.lock().unwrap(); let topic = chains @@ -941,22 +960,23 @@ mod tests { #[test] fn a_mask_only_takes_capabilities_away() { - assert_eq!(MemoryStore::new().capabilities(), FULL); + // The change feed comes with a notification log. + let no_log = Capabilities { + change_feed: false, + ..FULL + }; + assert_eq!(MemoryStore::new().capabilities(), no_log); + assert_eq!(MemoryStore::with_notification_log().capabilities(), FULL); let masked = log_only().capabilities(); assert!(!masked.producer_state && !masked.kept_last && !masked.long_poll); // Positions stay offsets whatever the mask, and the limits stay the store's. assert!(masked.dense_positions); assert_eq!(masked.max_record_bytes, MAX_RECORD_BYTES); - let all = Capabilities { - holds: true, - change_feed: true, - ..FULL - }; - assert_eq!(MemoryStore::with_capabilities(all).capabilities(), FULL); + assert_eq!(MemoryStore::with_capabilities(FULL).capabilities(), no_log); } #[tokio::test] - async fn without_producer_state_a_repeat_lands_again_and_closes_are_unsupported() { + async fn without_producer_state_a_repeat_lands_again_and_a_close_is_a_record() { let store = log_only(); let first = store.append(append(1, &[b"a"])).await.unwrap(); let again = store.append(append(1, &[b"a"])).await.unwrap(); @@ -964,7 +984,6 @@ mod tests { assert_eq!(store.read(read_after("")).await.unwrap().records.len(), 2); for error in [ store.close_chain(&chain()).await.unwrap_err(), - store.close_topic(&chain(), "out").await.unwrap_err(), store .retire_producer(&chain(), "out", "p", 1) .await @@ -972,6 +991,10 @@ mod tests { ] { assert_eq!(error.kind, StreamFailureKind::Unsupported); } + store.close_topic(&chain(), "out").await.unwrap(); + let read = store.read(read_after("")).await.unwrap(); + assert_eq!(read.records.len(), 3); + assert!(!read.closed); } #[tokio::test] diff --git a/crates/streams/src/reader.rs b/crates/streams/src/reader.rs index 4b14d9960..fcdadd581 100644 --- a/crates/streams/src/reader.rs +++ b/crates/streams/src/reader.rs @@ -285,13 +285,18 @@ pub(crate) async fn read_with( request: &ReadRequest, state: &mut ReadState, ) -> StreamResult<(ReadResponse, bool, Option)> { - if state.closed { - let response = ReadResponse { - cursor: request.after.clone(), - ..Default::default() - }; - return Ok((response, true, None)); - } + read_bounded(store, target, request, state, None).await +} + +/// [read_with] that stops once it reached the cursor `until`, so a replay read takes the records +/// up to the recorded end and its state sees nothing past it. +pub(crate) async fn read_bounded( + store: &dyn StreamStore, + target: &ReadTarget, + request: &ReadRequest, + state: &mut ReadState, + until: Option<&str>, +) -> StreamResult<(ReadResponse, bool, Option)> { let from_end = request.after == END; let position = if from_end { store @@ -383,6 +388,9 @@ pub(crate) async fn read_with( let max_bytes = usize::try_from(request.max_bytes).unwrap_or(usize::MAX); let (mut bytes, mut kept) = (0usize, 0usize); for stored in page.records { + if until.is_some_and(|until| previous == until) { + break; + } // The first record always goes, so a record above the budget still reads. if max_bytes > 0 && kept > 0 && bytes + stored.record.len() > max_bytes { // The records left come first, so the close waits for the next answer. @@ -404,11 +412,20 @@ pub(crate) async fn read_with( }; match StreamRecordKind::try_from(record.kind) { // The cursor stays before the close, so a read that resumes there ends at it too. + // A later call reads the close again, so it answers the outcome too. Ok(StreamRecordKind::Close) if record.topic.is_empty() || record.topic == target.topic => { state.closed = true; closed = true; + // A chain close keeps no outcome. + if !record.topic.is_empty() { + close = Some(StoredClose { + result: record.close_result, + failure: record.close_failure, + closed_by: record.producer_id, + }); + } break; } Ok(StreamRecordKind::Close | StreamRecordKind::Topic) => { diff --git a/crates/streams/src/redis/mod.rs b/crates/streams/src/redis/mod.rs index 10f3f21ea..40845dd85 100644 --- a/crates/streams/src/redis/mod.rs +++ b/crates/streams/src/redis/mod.rs @@ -47,8 +47,8 @@ const CAPABILITIES: Capabilities = Capabilities { fence: false, }, kept_last: true, - holds: false, - change_feed: false, + holds: true, + change_feed: true, cheap_logs: true, delete_owner: true, topic_listing: true, @@ -162,6 +162,12 @@ impl RedisStore { options.retention ))); } + if options.plain && (options.notification_log || options.archive_lease.is_some()) { + return Err(StreamError::refused( + "the Redis store in plain mode runs no script, so it keeps no notification log \ + and no archive hold", + )); + } if options.archive_lease.is_some() && !options.notification_log { return Err(StreamError::refused( "an archive needs the notification log, since a driver archives", @@ -344,6 +350,15 @@ impl RedisStore { }) } + fn holds(&self) -> StreamResult<()> { + if self.plain { + return Err(StreamError::unsupported( + "the Redis store in plain mode can't hold records for a positions-only reader", + )); + } + Ok(()) + } + fn producer_state(&self) -> StreamResult<()> { if self.plain { return Err(StreamError::unsupported( @@ -678,10 +693,15 @@ impl StreamStore for RedisStore { return Capabilities { producer_state: false, kept_last: false, + holds: false, + change_feed: false, ..CAPABILITIES }; } - CAPABILITIES + Capabilities { + change_feed: self.log.is_some(), + ..CAPABILITIES + } } async fn append(&self, request: StoreAppendRequest) -> StreamResult { @@ -905,6 +925,7 @@ impl StreamStore for RedisStore { holder: &str, position: &str, ) -> StreamResult<()> { + self.holds()?; let keys = ChainKeys::new(&self.prefix, chain); // Redis never stores `0-0`, so `0-1` holds every record, and the trim's exclusive range // below it stays valid. Below it, a crafted `0-0` would fail every later append. @@ -929,6 +950,7 @@ impl StreamStore for RedisStore { } async fn release_floor(&self, chain: &ChainId, topic: &str, holder: &str) -> StreamResult<()> { + self.holds()?; let keys = ChainKeys::new(&self.prefix, chain); let mut script = scripts::RELEASE_SCRIPT.key(keys.floors(topic)); // The archive releases once it holds a closed topic to its end, so the topic leaves the @@ -952,6 +974,7 @@ impl StreamStore for RedisStore { topic: &str, position: &str, ) -> StreamResult<()> { + self.holds()?; let Some(lease) = self.archive_lease_ms else { return Err(StreamError::unsupported("this store keeps no archive hold")); }; @@ -1013,7 +1036,10 @@ impl StreamStore for RedisStore { close: StoredClose, ) -> StreamResult { let keys = ChainKeys::new(&self.prefix, chain); - self.producer_state()?; + if self.plain { + crate::close::write_close(self, chain, topic, &close).await?; + return Ok(close); + } // The meta is the topic's tombstone, so the mark and the outcome live as long as the topic // is known. let mut script = scripts::CLOSE_SCRIPT.key(keys.meta(topic)); diff --git a/crates/streams/src/service.rs b/crates/streams/src/service.rs index d6ac91294..00366d51c 100644 --- a/crates/streams/src/service.rs +++ b/crates/streams/src/service.rs @@ -813,7 +813,11 @@ mod tests { assert_eq!(full.name, "memory"); let capabilities = full.capabilities.unwrap(); assert!(capabilities.producer_state && capabilities.kept_last); - assert!(capabilities.dense_positions && !capabilities.holds); + assert!(capabilities.dense_positions && capabilities.holds); + assert!( + !capabilities.change_feed, + "a store with no notification log has no change feed" + ); assert_eq!(capabilities.max_record_bytes, 1536 * 1024); assert_eq!(capabilities.max_batch_bytes, 4 << 20); let masked = StreamService::connect( diff --git a/crates/streams/src/store.rs b/crates/streams/src/store.rs index 48919ec30..7b3d8d2c6 100644 --- a/crates/streams/src/store.rs +++ b/crates/streams/src/store.rs @@ -320,22 +320,28 @@ pub trait StreamStore: Send + Sync { } /// Marks one topic of a chain closed, like [StreamStore::close_chain] for that topic alone. + /// A store without producer state appends a `CLOSE` record instead, which ends every read + /// there but refuses no append. async fn close_topic(&self, chain: &ChainId, topic: &str) -> StreamResult<()> { - let _ = (chain, topic); - unsupported("keep close flags") + crate::close::write_close(self, chain, topic, &StoredClose::default()).await } /// Like [StreamStore::close_topic], with the close's outcome and who closed it. A store that /// keeps the outcome keeps the topic's first close only, and a read of the closed topic /// answers it. Answers the close the store keeps, so a wrapper tells the stream's notifier the - /// same close a reader sees. A store that keeps none answers `close`. + /// same close a reader sees. A store that keeps none answers `close`. Without producer state + /// the `CLOSE` record carries the outcome, and the first record wins for every reader. async fn close_topic_with( &self, chain: &ChainId, topic: &str, close: StoredClose, ) -> StreamResult { - self.close_topic(chain, topic).await?; + if self.capabilities().producer_state { + self.close_topic(chain, topic).await?; + } else { + crate::close::write_close(self, chain, topic, &close).await?; + } Ok(close) } diff --git a/crates/streams/tests/conformance.rs b/crates/streams/tests/conformance.rs index 102d9b9ab..de0e29275 100644 --- a/crates/streams/tests/conformance.rs +++ b/crates/streams/tests/conformance.rs @@ -2532,10 +2532,8 @@ mod cases { unsupported(store.close_chain(&chain).await), !declared.producer_state ); - assert_eq!( - unsupported(store.close_topic(&chain, OUT).await), - !declared.producer_state - ); + // Without close flags a topic close is a `CLOSE` record, which every store can append. + store.close_topic(&chain, OUT).await.unwrap(); assert_eq!( unsupported(store.retire_producer(&chain, OUT, "p", 1).await), !declared.producer_state @@ -2558,6 +2556,11 @@ mod cases { .await; assert_eq!(unsupported(delete.map(drop)), !declared.delete_owner); assert!(declared.max_record_bytes > 0 && declared.max_batch_bytes > 0); + assert_eq!( + unsupported(store.hold_floor(&chain, OTHER, "reader", "").await), + !declared.holds + ); + assert_eq!(store.notification_log().is_some(), declared.change_feed); } /// Without producer state a repeat after a lost answer lands again, and readers drop it. From 118975a0d3e72b61bfcb0df5eada116d085ce849 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 16 Oct 2026 14:02:12 -0700 Subject: [PATCH 2/3] Expected a CLOSE record from a topic close in plain mode. --- crates/streams/src/redis/mod.rs | 17 ++++++++++------- 1 file changed, 10 insertions(+), 7 deletions(-) diff --git a/crates/streams/src/redis/mod.rs b/crates/streams/src/redis/mod.rs index 40845dd85..3c5ffabd6 100644 --- a/crates/streams/src/redis/mod.rs +++ b/crates/streams/src/redis/mod.rs @@ -1334,7 +1334,7 @@ mod tests { } #[tokio::test] - async fn in_plain_mode_a_repeat_lands_again_and_no_close_flag_is_kept() { + async fn in_plain_mode_a_repeat_lands_again_and_a_close_is_a_record() { let Ok(url) = std::env::var("STREAMS_REDIS_URL") else { return; }; @@ -1356,12 +1356,15 @@ mod tests { assert_eq!(read.records.len(), 2); assert!(!read.closed); assert_eq!(store.list_topics(&chain(&run)).await.unwrap(), ["out"]); - for error in [ - store.close_chain(&chain(&run)).await.unwrap_err(), - store.close_topic(&chain(&run), "out").await.unwrap_err(), - ] { - assert_eq!(error.kind, StreamFailureKind::Unsupported); - } + let error = store.close_chain(&chain(&run)).await.unwrap_err(); + assert_eq!(error.kind, StreamFailureKind::Unsupported); + // A topic close is a `CLOSE` record, which lands after the two copies. + store.close_topic(&chain(&run), "out").await.unwrap(); + let read = store + .read(waiting_read(&run, Duration::ZERO)) + .await + .unwrap(); + assert_eq!(read.records.len(), 3); // A read after a position the log never held answers what follows, as a plain log does. let missing = StoreReadRequest { after_position: "1-0".to_string(), From 342c9bc17e33403d2616b3d0118e17fa54e8196c Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Fri, 16 Oct 2026 16:01:18 -0700 Subject: [PATCH 3/3] Checked a replay for a lost range before a count mismatch. A range whose end retention dropped never meets its recorded end, so the read ran on into later records and failed as a mismatch. --- crates/streams/src/handle.rs | 42 +++++++++++++++++++++++++++++------- 1 file changed, 34 insertions(+), 8 deletions(-) diff --git a/crates/streams/src/handle.rs b/crates/streams/src/handle.rs index 5f461094a..cabf2bb55 100644 --- a/crates/streams/src/handle.rs +++ b/crates/streams/src/handle.rs @@ -587,14 +587,10 @@ impl Streams { request.after ))); } - if stored > expected { - return Err(mismatch(format!( - "the replayed read holds {stored} records up to {:?}, not the {expected} the read \ - got", - request.to - ))); - } - if page.cursor != request.to || page.records_digest != request.records_digest { + if stored > expected + || page.cursor != request.to + || page.records_digest != request.records_digest + { // A read from the beginning has no position to check, so a trim since the read shows // as other records, which end elsewhere or differ. if request.after.is_empty() @@ -623,6 +619,13 @@ impl Streams { request.to ))); } + if stored > expected { + return Err(mismatch(format!( + "the replayed read holds {stored} records up to {:?}, not the {expected} the \ + read got", + request.to + ))); + } if page.cursor != request.to { return Err(mismatch(format!( "the replayed read ends at {:?}, not at {:?}", @@ -2468,6 +2471,29 @@ mod tests { assert_eq!(replayed.records, page.records); } + #[tokio::test] + async fn a_replay_whose_end_retention_dropped_is_lost_not_a_mismatch() { + let setup = setup(StreamsOptions::default()); + setup.streams.append(append("p", 1, "1")).await.unwrap(); + setup.streams.append(append("p", 2, "2")).await.unwrap(); + let request = ReadRequest { + stream: Some(stream("out")), + ..Default::default() + }; + let page = setup.streams.read(request.clone()).await.unwrap(); + // A lapsed floor let retention take both records, and the stream went on. + setup.store.truncate(&chain("run-1"), "out", 0); + setup.streams.append(append("p", 3, "3")).await.unwrap(); + setup.streams.append(append("p", 4, "4")).await.unwrap(); + setup.streams.append(append("p", 5, "5")).await.unwrap(); + let error = setup + .streams + .read_range(replay_of(&request, &page)) + .await + .unwrap_err(); + assert_eq!(error.kind, StreamFailureKind::ReplayRangeLost, "{error}"); + } + #[tokio::test] async fn a_positions_only_read_needs_holds_or_a_replay_horizon() { let no_holds = crate::Capabilities {