[Jetty] Mapped holds, the change feed and close outcomes onto store capabilities.

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.

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,
                 }
             }
 

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

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<S: StreamStore + ?Sized>(
+    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<S: StreamStore + ?Sized>(
+    store: &S,
     chain: &ChainId,
     topic: &str,
     record: StreamRecord,

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,7 +587,10 @@ impl Streams {
                 request.after
             )));
         }
-        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()
@@ -570,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 {:?}",
@@ -2391,6 +2447,116 @@ 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_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 {
+            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 {