From 52547ffec30ec747406262d68124f24f69185892 Mon Sep 17 00:00:00 2001 From: Mohammad Dashti Date: Tue, 13 Oct 2026 23:58:01 -0700 Subject: [PATCH 1/7] Let a positions-only reader replay its reads from the store. A read answers a digest of its records, its chain and its resolved start, and ReadRange reads that exact range again on replay and checks the digest. A reader's floor holds what it read against retention under a lease that each read renews, and ReleaseFloor ends it after a grace. --- .../temporal/sdk/core/streams/streams.proto | 56 +++ crates/sdk-core/CHANGELOG.md | 8 + crates/streams/docs/redis.md | 13 + crates/streams/lua/append.lua | 2 +- crates/streams/lua/floor.lua | 28 ++ crates/streams/lua/hold.lua | 21 ++ crates/streams/lua/keep.lua | 8 +- crates/streams/lua/release.lua | 10 + crates/streams/src/handle.rs | 335 +++++++++++++++++- crates/streams/src/memory.rs | 59 ++- crates/streams/src/notify.rs | 14 + crates/streams/src/reader.rs | 46 ++- crates/streams/src/redis/keys.rs | 6 + crates/streams/src/redis/mod.rs | 59 +++ crates/streams/src/redis/scripts.rs | 64 +++- crates/streams/src/service.rs | 80 ++++- crates/streams/src/store.rs | 27 ++ crates/streams/tests/conformance.rs | 41 +++ crates/streams/tests/redis_store.rs | 233 ++++++++++++ 19 files changed, 1085 insertions(+), 25 deletions(-) create mode 100644 crates/streams/lua/floor.lua create mode 100644 crates/streams/lua/hold.lua create mode 100644 crates/streams/lua/release.lua diff --git a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto index c1fbdd9dac..c6675bf3a3 100644 --- a/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto +++ b/crates/protos/protos/local/temporal/sdk/core/streams/streams.proto @@ -43,6 +43,12 @@ service StreamService { // wait limit (`streamnotifier.pollMaxWait`). The reader then reads again with `arm`. After the // wait limit, that read finds records whose notification was lost, or arms again. rpc WaitForNotification(WaitForNotificationRequest) returns (WaitForNotificationResponse); + // Reads again the exact range an earlier read answered, for a positions-only Workflow reader on + // replay, and checks it against that read's digest. Waits for nothing and checks no owner. + rpc ReadRange(ReadRangeRequest) returns (ReadResponse); + // Ends a positions-only reader's hold on a stream's records after a grace, so retention takes + // them again. + rpc ReleaseFloor(ReleaseFloorRequest) returns (ReleaseFloorResponse); } // Which store holds a process's streams, and how to reach it. Lang builds one store from it and @@ -83,6 +89,11 @@ message RedisStoreConfig { uint32 blocking_reads_per_node = 5; // How long a short call waits for its reply. Unset uses the default. google.protobuf.Duration response_timeout = 6; + // How long a positions-only reader holds a stream's records after its last read. Unset uses + // `retention`. + google.protobuf.Duration floor_lease = 7; + // How long a hold lasts after its reader released it. Unset keeps one day. + google.protobuf.Duration floor_release_grace = 8; } // Streams in this process's memory, for tests and local development. Nothing survives the @@ -183,6 +194,10 @@ message ReadRequest { // finds nothing, the store arms the stream in the same step, so the next append notifies the // notifier. A reader that doesn't wait on the notifier leaves it unset. bool arm = 6; + // For a positions-only Workflow reader: names the reader, as its run id and its subscription + // id. The store holds the records after this read's start against retention for that holder, + // until `ReleaseFloor` and its grace, or until the holder's lease passes with no read. + string floor_holder = 7; } // A later attempt of a producer started writing. Readers synthesize it, and no store holds it. @@ -231,8 +246,43 @@ message ReadResponse { temporal.api.failure.v1.Failure close_failure = 8; // Set with `done` to the `closed_by` of the topic's first close. string closed_by = 9; + // SHA-256 over the answer's records, as `ReadRange` checks it: per record its cursor, its stale + // flag, and for a stored record its kind, producer, attempt, sequence and plaintext content + // hash, or for a supersession its producer and both attempts. It leaves out the encoded body. + bytes records_digest = 10; + // The first run of the chain the read read. `ReadRange` takes it, so replay needs no describe. + string chain_first_run_id = 11; + // The cursor this answer reads strictly after, with `$end` resolved. With `cursor`, it is the + // range `ReadRange` reads again. + string start_cursor = 12; +} + +message ReadRangeRequest { + // The stream, as the read named it. + StreamAddress stream = 1; + // The read's `chain_first_run_id`. + string chain_first_run_id = 2; + // The read's `start_cursor` and its answer's `cursor`. + string after = 3; + string to = 4; + // How many stored records the answer held, not counting supersessions. + uint32 stored_records = 5; + // The read's input `state`. + bytes state = 6; + // The answer's `records_digest`. + bytes records_digest = 7; + // As in `ReadRequest`, so a replay holds the records for its own run. + string floor_holder = 8; +} + +message ReleaseFloorRequest { + StreamAddress stream = 1; + string chain_first_run_id = 2; + string floor_holder = 3; } +message ReleaseFloorResponse {} + message LatestRequest { StreamAddress stream = 1; } @@ -383,6 +433,12 @@ enum StreamFailureKind { // the same call fails the same way. A lang built before this kind maps it, as any kind it // doesn't know, to a storage failure. STREAM_FAILURE_KIND_INVALID_ARGUMENT = 12; + // A replay read found other records than the read it repeats: other content, another count + // or another end. Lang reports it as nondeterminism. + STREAM_FAILURE_KIND_REPLAY_MISMATCH = 13; + // A replay read found that the store no longer holds the range the read it repeats got + // (REC-01), such as records that retention trimmed or a deleted log. + STREAM_FAILURE_KIND_REPLAY_RANGE_LOST = 14; } message StreamFailure { diff --git a/crates/sdk-core/CHANGELOG.md b/crates/sdk-core/CHANGELOG.md index 55086a2e1a..61851aec3e 100644 --- a/crates/sdk-core/CHANGELOG.md +++ b/crates/sdk-core/CHANGELOG.md @@ -93,6 +93,14 @@ relevant information. append refusal names the close's `closed_by` in `StreamFailure.closed_by`. The Redis store keeps it in the topic's meta as `close`, set only once by the close script. `StreamStore::close_topic_with` and `close_stream` take a `StoredClose`. +* Experimental: a positions-only Workflow reader can record what it read in place of the records. + `ReadResponse` gains `records_digest`, `chain_first_run_id` and `start_cursor`, and the new + `ReadRange` stream call reads that exact range again on replay and checks the digest. Other + records fail as `REPLAY_MISMATCH`, and records the store no longer holds as + `REPLAY_RANGE_LOST` (REC-01). `ReadRequest.floor_holder` holds the records a reader read + against retention, under a lease that each read renews, and the new `ReleaseFloor` call ends + the hold after a grace. The Redis store keeps holds in a `floors` hash per topic, and its append + script trims around them. `RedisStoreConfig` gains `floor_lease` and `floor_release_grace`. * Experimental: the stream service's `RetireProducer` call drops one producer attempt's dedupe state on a stream, which the Redis store keeps as one field of the topic's meta hash until retention. A producer that writes once per short-lived attempt calls it after the attempt's diff --git a/crates/streams/docs/redis.md b/crates/streams/docs/redis.md index 9adf86a0e9..55da976f76 100644 --- a/crates/streams/docs/redis.md +++ b/crates/streams/docs/redis.md @@ -36,6 +36,19 @@ reaches the end sees the outcome without a wake. An append that the close refuse beyond the documented ones, since the script uses `HGET`, `HSET` and `PEXPIRE` only. An SDK's provider ignores it. +**Floors.** A positions-only Workflow reader records which range it read, not the records, and +reads the range again on replay. So the store holds those records past retention while the +reader's run can replay. A hash `:t::floors` keeps one field per reader, its run id +and subscription id, with `|`: the lowest position the reader read after, and +the end of its lease. A read or a replay read renews the lease, which lasts the retention by +default, and extends the log's and the meta's expiry to it. `ReleaseFloor` moves the end to now +plus a release grace, one day by default. The append script's trim reads the hash first. It drops +the ended holds, trims to the lower of the retention floor and the lowest held position, and keeps +the log until the latest lease ends. A hold from the beginning holds `0-1`, so it keeps the whole +topic. The scripts use `HGET`, `HSET`, `HGETALL`, `HDEL`, `PTTL` and `PEXPIRE`, which the +documented ACL allows. Without holds the hash doesn't exist and the trim is as before. The append +script's hash changes again. + **Retired producers.** A field stays until a write's trim finds it older than the retention, so a topic written by many short-lived producers holds one field per producer attempt of the last retention window, about 235 bytes each. `RetireProducer` deletes one attempt's field with diff --git a/crates/streams/lua/append.lua b/crates/streams/lua/append.lua index 69153abf75..3dcf7e0bcc 100644 --- a/crates/streams/lua/append.lua +++ b/crates/streams/lua/append.lua @@ -34,5 +34,5 @@ for i = 6, #ARGV do end redis.call('HSET', KEYS[2], field, ARGV[4] .. '|' .. (#ARGV - 5) .. '|' .. first .. '|' .. last .. '|' .. digest_arg) -keep(KEYS[1], KEYS[2], #ARGV - 5, last, retention, grace) +keep(KEYS[1], KEYS[2], #ARGV - 5, last, retention, grace, KEYS[4]) return {first, last, redis.call('HGET', KEYS[2], 'arm') or '0'} diff --git a/crates/streams/lua/floor.lua b/crates/streams/lua/floor.lua new file mode 100644 index 0000000000..d2f762c787 --- /dev/null +++ b/crates/streams/lua/floor.lua @@ -0,0 +1,28 @@ +local function clock_ms() + local time = redis.call('TIME') + return tonumber(time[1]) * 1000 + math.floor(tonumber(time[2]) / 1000) +end +local function id_below(a, b) + local am, as = string.match(a, '^(%d+)%-(%d+)$') + local bm, bs = string.match(b, '^(%d+)%-(%d+)$') + am, as, bm, bs = tonumber(am), tonumber(as), tonumber(bm), tonumber(bs) + return am < bm or (am == bm and as < bs) +end +local function held(floors, floor, retention) + local now = clock_ms() + local life = retention + local entries = redis.call('HGETALL', floors) + for i = 1, #entries, 2 do + local id, until_ms = string.match(entries[i + 1], '^(%d+%-%d+)|(%d+)$') + until_ms = tonumber(until_ms) + if not until_ms or until_ms <= now then + redis.call('HDEL', floors, entries[i]) + else + if id_below(id, floor) then + floor = id + end + life = math.max(life, until_ms - now) + end + end + return floor, life +end diff --git a/crates/streams/lua/hold.lua b/crates/streams/lua/hold.lua new file mode 100644 index 0000000000..b37e2929bf --- /dev/null +++ b/crates/streams/lua/hold.lua @@ -0,0 +1,21 @@ +local now = clock_ms() +local lease, grace = tonumber(ARGV[3]), tonumber(ARGV[4]) +local id = ARGV[2] +local held_value = redis.call('HGET', KEYS[1], ARGV[1]) +if held_value then + local old = string.match(held_value, '^(%d+%-%d+)|') + if old and id_below(old, id) then + id = old + end +end +redis.call('HSET', KEYS[1], ARGV[1], id .. '|' .. (now + lease)) +local function extend(key, ms, create) + local ttl = redis.call('PTTL', key) + if (ttl >= 0 or (create and ttl == -1)) and ttl < ms then + redis.call('PEXPIRE', key, ms) + end +end +extend(KEYS[1], lease, true) +extend(KEYS[2], lease, false) +extend(KEYS[3], lease + grace, false) +return 1 diff --git a/crates/streams/lua/keep.lua b/crates/streams/lua/keep.lua index 24dfc6c052..6e83456876 100644 --- a/crates/streams/lua/keep.lua +++ b/crates/streams/lua/keep.lua @@ -23,17 +23,19 @@ local function sweep_sessions(meta, floor_ms) end redis.call('HSET', meta, 'hwscan', found[1]) end -local function keep(log, meta, added, last, retention, grace) +local function keep(log, meta, added, last, retention, grace, floors) local floor_ms = now_ms() - retention 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 redis.call('XTRIM', log, 'MINID', floor) redis.call('HSET', meta, 'trimmed', doomed[1][1]) end - redis.call('PEXPIRE', log, retention) + redis.call('PEXPIRE', log, life) redis.call('HINCRBY', meta, 'added', added) redis.call('HSET', meta, 'last', last) - redis.call('PEXPIRE', meta, retention + grace) + redis.call('PEXPIRE', meta, life + grace) sweep_sessions(meta, floor_ms) end diff --git a/crates/streams/lua/release.lua b/crates/streams/lua/release.lua new file mode 100644 index 0000000000..3841c125a6 --- /dev/null +++ b/crates/streams/lua/release.lua @@ -0,0 +1,10 @@ +local held_value = redis.call('HGET', KEYS[1], ARGV[1]) +if not held_value then + return 0 +end +local id, until_ms = string.match(held_value, '^(%d+%-%d+)|(%d+)$') +local release_ms = clock_ms() + tonumber(ARGV[2]) +if id and tonumber(until_ms) > release_ms then + redis.call('HSET', KEYS[1], ARGV[1], id .. '|' .. release_ms) +end +return 1 diff --git a/crates/streams/src/handle.rs b/crates/streams/src/handle.rs index 31b6f29708..de0f31f39f 100644 --- a/crates/streams/src/handle.rs +++ b/crates/streams/src/handle.rs @@ -4,15 +4,16 @@ //! where the contract says so, and mints and checks cursors. Stores only move records. use crate::{ - MAX_CLOSE_BYTES, ReadTarget, StreamError, StreamResult, StreamStore, WORKFLOW_OWNER_KIND, - activity_producer_id, check_append_digest, mint_cursor, + END, MAX_CLOSE_BYTES, ReadTarget, StreamError, StreamResult, StreamStore, WORKFLOW_OWNER_KIND, + activity_producer_id, check_append_digest, cursor_position, mint_cursor, owner::{OwnerClient, OwnerDescription, OwnerError}, proto::{ ActivityProducer, AppendRequest, AppendResponse, ChainId, CloseRequest, CloseResponse, DeleteOwnerRequest, DeleteOwnerResponse, LatestRequest, LatestResponse, NamedProducer, - ReadRequest, ReadResponse, RetireProducerRequest, RetireProducerResponse, - StoreAppendRequest, StoreLatestRequest, StoredClose, StreamAddress, StreamOwnerKind, - append_request, retire_producer_request, + ReadRangeRequest, ReadRequest, ReadResponse, ReleaseFloorRequest, ReleaseFloorResponse, + RetireProducerRequest, RetireProducerResponse, StoreAppendRequest, StoreLatestRequest, + StoredClose, StreamAddress, StreamFailureKind, StreamOwnerKind, append_request, + read_record, retire_producer_request, }, reader::{ReadState, read_with}, stored_append_record, stream_hash, @@ -26,6 +27,9 @@ use std::{ use temporalio_common::protos::temporal::api::enums::v1::WorkflowExecutionStatus; use tokio::time::Instant; +/// The longest `floor_holder` a read takes. +const MAX_FLOOR_HOLDER_BYTES: usize = 512; + /// The longest `closed_by` a close takes. const MAX_CLOSED_BY_BYTES: usize = 256; @@ -251,6 +255,10 @@ impl Streams { topic: stream.topic.clone(), stream_hash: self.hash(stream), }; + if !request.floor_holder.is_empty() { + self.hold_floor(&target, &request.after, &request.floor_holder) + .await?; + } let wait = request .wait .and_then(|wait| Duration::try_from(wait).ok()) @@ -326,6 +334,162 @@ impl Streams { }) } + /// Holds the records after `after` against retention for `holder`, a positions-only reader. + async fn hold_floor(&self, target: &ReadTarget, after: &str, holder: &str) -> StreamResult<()> { + 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 {}", + holder.len() + ))); + } + let position = if after == END { + self.store + .latest(StoreLatestRequest { + chain: Some(target.chain.clone()), + topic: target.topic.clone(), + }) + .await? + .position + } else { + cursor_position(after, self.store.name(), &target.stream_hash)? + .unwrap_or_default() + .to_string() + }; + self.store + .hold_floor(&target.chain, &target.topic, holder, &position) + .await + } + + /// Reads again the exact range of an earlier answer and checks it against that answer's + /// digest (RR-2). It runs the read that the answer came from, with the same input state, so + /// it gives the same supersessions and stale marks, and it waits for nothing and checks no + /// owner. Records with other content fail as a mismatch. Records the store no longer holds + /// fail as lost (REC-01). + pub async fn read_range(&self, request: ReadRangeRequest) -> StreamResult { + let stream = address(request.stream.as_ref())?; + if request.chain_first_run_id.is_empty() { + return Err(StreamError::invalid_argument( + "a replay read needs the chain its read read", + )); + } + if request.after == END { + return Err(StreamError::invalid_argument( + "a replay read starts at the read's resolved start_cursor, not at $end", + )); + } + let mut state = ReadState::decode_from(&request.state)?; + if state.first_run_id.is_empty() { + state.first_run_id = request.chain_first_run_id.clone(); + } + let target = ReadTarget { + chain: chain_of(stream, &request.chain_first_run_id), + topic: stream.topic.clone(), + stream_hash: self.hash(stream), + }; + if !request.floor_holder.is_empty() { + self.hold_floor(&target, &request.after, &request.floor_holder) + .await?; + } + let expected = request.stored_records as usize; + let lost = |message: String| { + StreamError::new( + StreamFailureKind::ReplayRangeLost, + format!("REC-01: {message}"), + ) + }; + let mismatch = + |message: String| StreamError::new(StreamFailureKind::ReplayMismatch, message); + let page = if expected == 0 { + ReadResponse { + cursor: request.after.clone(), + records_digest: crate::reader::records_digest(&[]), + chain_first_run_id: request.chain_first_run_id.clone(), + start_cursor: request.after.clone(), + ..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 + ) => + { + return Err(lost(error.message)); + } + Err(error) => return Err(error), + } + }; + let stored = page + .records + .iter() + .filter(|record| matches!(record.record, Some(read_record::Record::Stored(_)))) + .count(); + if stored < expected { + return Err(lost(format!( + "the store holds {stored} of the {expected} records after {:?} that the read got", + request.after + ))); + } + if page.cursor != request.to { + return Err(mismatch(format!( + "the replayed read ends at {:?}, not at {:?}", + page.cursor, request.to + ))); + } + if 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. + if request.after.is_empty() + && self + .store + .dropped_after(&target.chain, &target.topic, "") + .await? + .is_some() + { + return Err(lost( + "retention trimmed the start of the stream, which the read began at" + .to_string(), + )); + } + return Err(mismatch( + "the replayed records differ from the records the read got".to_string(), + )); + } + Ok(ReadResponse { + state: state.encode_to_vec(), + ..page + }) + } + + /// Ends a positions-only reader's hold after the store's release grace. + pub async fn release_floor( + &self, + request: ReleaseFloorRequest, + ) -> StreamResult { + let stream = address(request.stream.as_ref())?; + if request.chain_first_run_id.is_empty() || request.floor_holder.is_empty() { + return Err(StreamError::invalid_argument( + "a release needs the chain and the floor holder", + )); + } + self.store + .release_floor( + &chain_of(stream, &request.chain_first_run_id), + &stream.topic, + &request.floor_holder, + ) + .await?; + Ok(ReleaseFloorResponse {}) + } + /// Whether the owner a read follows ended. Following the chain, a run that continued as /// new is not the end, and a new chain on the Workflow id is. async fn owner_ended(&self, stream: &StreamAddress, chain: &ChainId) -> StreamResult { @@ -1397,6 +1561,167 @@ mod tests { setup.streams.append(append("p", 1, "1")).await.unwrap(); } + /// The replay read that repeats the live `read` of `request` that answered `page`. + fn replay_of(request: &ReadRequest, page: &ReadResponse) -> ReadRangeRequest { + ReadRangeRequest { + stream: Some(stream("out")), + chain_first_run_id: page.chain_first_run_id.clone(), + after: page.start_cursor.clone(), + to: page.cursor.clone(), + stored_records: page + .records + .iter() + .filter(|r| matches!(r.record, Some(read_record::Record::Stored(_)))) + .count() as u32, + state: request.state.clone(), + records_digest: page.records_digest.clone(), + floor_holder: String::new(), + } + } + + fn attempt(producer: &str, attempt: i64, sequence: i64, value: &str) -> AppendRequest { + AppendRequest { + producer: Some(Producer::Named(NamedProducer { + producer_id: producer.to_string(), + attempt, + })), + ..append(producer, sequence, value) + } + } + + /// A read after the stream's first record, with its replay request. + async fn first_page(setup: &Setup, holder: &str) -> (ReadRequest, ReadResponse) { + let first = setup.streams.append(append("p", 1, "1")).await.unwrap(); + setup.streams.append(append("p", 2, "2")).await.unwrap(); + // A newer attempt makes the read synthesize a supersession. + setup.streams.append(attempt("p", 2, 1, "3")).await.unwrap(); + let request = ReadRequest { + floor_holder: holder.to_string(), + ..read_after(&first.last_cursor, &[], Duration::ZERO) + }; + let page = setup.streams.read(request.clone()).await.unwrap(); + (request, page) + } + + #[tokio::test] + async fn a_replay_read_gives_the_records_the_read_got() { + let setup = setup(quick()); + let (request, page) = first_page(&setup, "").await; + assert!( + page.records + .iter() + .any(|r| matches!(r.record, Some(read_record::Record::Superseded(_)))), + "{page:?}" + ); + assert_eq!(page.chain_first_run_id, "run-1"); + assert_eq!(page.start_cursor, request.after); + assert_eq!(page.records_digest.len(), 32); + // More records after the range change nothing in it. + setup.streams.append(attempt("p", 2, 2, "4")).await.unwrap(); + let replayed = setup + .streams + .read_range(replay_of(&request, &page)) + .await + .unwrap(); + assert_eq!(replayed.records, page.records); + assert_eq!(replayed.cursor, page.cursor); + assert_eq!(replayed.records_digest, page.records_digest); + } + + #[tokio::test] + async fn a_replay_read_needs_no_owner() { + let setup = setup(quick()); + let (request, page) = first_page(&setup, "").await; + // History dropped the owner, as for a Replayer with only the store. + setup.owners.end("wf", WorkflowExecutionStatus::Completed); + let before = setup.owners.describes(); + setup + .streams + .read_range(replay_of(&request, &page)) + .await + .unwrap(); + assert_eq!(setup.owners.describes(), before); + } + + #[tokio::test] + async fn a_replay_read_with_other_records_is_a_mismatch() { + let setup = setup(quick()); + let (request, page) = first_page(&setup, "").await; + let mut other_digest = replay_of(&request, &page); + other_digest.records_digest[0] ^= 1; + let mut other_end = replay_of(&request, &page); + other_end.stored_records -= 1; + for replay in [other_digest, other_end] { + let error = setup.streams.read_range(replay).await.unwrap_err(); + assert_eq!(error.kind, StreamFailureKind::ReplayMismatch, "{error}"); + } + } + + #[tokio::test] + async fn a_replay_read_of_records_retention_dropped_is_rec_01() { + let setup = setup(quick()); + let (request, page) = first_page(&setup, "").await; + setup.store.truncate(&chain("run-1"), "out", 1); + let error = setup + .streams + .read_range(replay_of(&request, &page)) + .await + .unwrap_err(); + assert_eq!(error.kind, StreamFailureKind::ReplayRangeLost, "{error}"); + assert!(error.message.starts_with("REC-01"), "{error}"); + } + + #[tokio::test] + async fn a_floor_keeps_what_a_positions_only_reader_read_until_it_is_released() { + let setup = setup(quick()); + let (request, page) = first_page(&setup, "reader-run/1").await; + setup.store.truncate(&chain("run-1"), "out", 1); + let replayed = setup + .streams + .read_range(ReadRangeRequest { + floor_holder: "reader-run/1".to_string(), + ..replay_of(&request, &page) + }) + .await + .unwrap(); + assert_eq!(replayed.records, page.records, "the floor kept the range"); + setup + .streams + .release_floor(ReleaseFloorRequest { + stream: Some(stream("out")), + chain_first_run_id: "run-1".to_string(), + floor_holder: "reader-run/1".to_string(), + }) + .await + .unwrap(); + setup.store.truncate(&chain("run-1"), "out", 1); + 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_replay_read_needs_its_chain_and_a_resolved_start() { + let setup = setup(quick()); + let (request, page) = first_page(&setup, "").await; + for replay in [ + ReadRangeRequest { + chain_first_run_id: String::new(), + ..replay_of(&request, &page) + }, + ReadRangeRequest { + after: crate::END.to_string(), + ..replay_of(&request, &page) + }, + ] { + let error = setup.streams.read_range(replay).await.unwrap_err(); + assert_eq!(error.kind, StreamFailureKind::InvalidArgument, "{error}"); + } + } + #[tokio::test] async fn only_the_pass_that_waits_out_the_read_arms() { // Owner checks every 20 ms cut the wait into passes. One arm for the whole wait keeps a diff --git a/crates/streams/src/memory.rs b/crates/streams/src/memory.rs index c08f1a31e5..68fb7b62c1 100644 --- a/crates/streams/src/memory.rs +++ b/crates/streams/src/memory.rs @@ -77,6 +77,9 @@ struct Chain { /// Kept apart from the topics, so an arm on a topic with no records leaves it unknown to /// reads and to the topic list. arms: HashMap, + /// Per topic, the position each positions-only reader holds the records after. `None` holds + /// the whole topic. + floors: HashMap>>, } /// A topic's arm: the generation a reader set, zero once cleared, and the last one handed out. @@ -126,15 +129,23 @@ impl MemoryStore { } /// Drops all but the newest `keep` records of a topic, standing in for a store's retention. + /// It keeps the records a positions-only reader holds, from the position it holds on. pub fn truncate(&self, chain: &ChainId, topic: &str, keep: usize) { let mut chains = self.chains.lock().unwrap(); - let Some(topic) = chains - .get_mut(&chain_key(chain)) - .and_then(|chain| chain.topics.get_mut(topic)) - else { + let Some(chain) = chains.get_mut(&chain_key(chain)) else { + return; + }; + let held = chain + .floors + .get(topic) + .and_then(|holders| holders.values().map(|position| position.unwrap_or(0)).min()); + let Some(topic) = chain.topics.get_mut(topic) else { return; }; - let drop = topic.records.len().saturating_sub(keep); + let mut drop = topic.records.len().saturating_sub(keep); + if let Some(held) = held { + drop = drop.min(held.saturating_sub(topic.base) as usize); + } topic.records.drain(..drop); topic.base += drop as u64; } @@ -400,6 +411,44 @@ impl StreamStore for MemoryStore { .map(|offset| offset.to_string())) } + async fn hold_floor( + &self, + chain: &ChainId, + topic: &str, + holder: &str, + position: &str, + ) -> StreamResult<()> { + let position = parse_position(position)?; + let mut chains = self.chains.lock().unwrap(); + let held = chains + .entry(chain_key(chain)) + .or_default() + .floors + .entry(topic.to_string()) + .or_default() + .entry(holder.to_string()) + .or_insert(position); + *held = match (*held, position) { + (Some(held), Some(position)) => Some(held.min(position)), + _ => None, + }; + Ok(()) + } + + /// 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<()> { + if let Some(holders) = self + .chains + .lock() + .unwrap() + .get_mut(&chain_key(chain)) + .and_then(|chain| chain.floors.get_mut(topic)) + { + holders.remove(holder); + } + Ok(()) + } + async fn close_chain(&self, chain: &ChainId) -> StreamResult<()> { self.chains .lock() diff --git a/crates/streams/src/notify.rs b/crates/streams/src/notify.rs index ef884545be..70af58dbc6 100644 --- a/crates/streams/src/notify.rs +++ b/crates/streams/src/notify.rs @@ -917,6 +917,20 @@ impl StreamStore for NotifyingStore { self.inner.delete_owner(request).await } + async fn hold_floor( + &self, + chain: &ChainId, + topic: &str, + holder: &str, + position: &str, + ) -> StreamResult<()> { + self.inner.hold_floor(chain, topic, holder, position).await + } + + async fn release_floor(&self, chain: &ChainId, topic: &str, holder: &str) -> StreamResult<()> { + self.inner.release_floor(chain, topic, holder).await + } + async fn close_stream( &self, chain: &ChainId, diff --git a/crates/streams/src/reader.rs b/crates/streams/src/reader.rs index 70685ec86c..97c3f8eb05 100644 --- a/crates/streams/src/reader.rs +++ b/crates/streams/src/reader.rs @@ -231,7 +231,7 @@ pub(crate) async fn read_with( .await?; let (closed, armed, close) = (page.closed, page.armed, page.close); let mut records = Vec::with_capacity(page.records.len()); - let mut previous = start; + let mut previous = start.clone(); for stored in page.records { let cursor = mint_cursor(store.name(), &target.stream_hash, &stored.position); let record = match parse(&stored.record) { @@ -270,7 +270,6 @@ pub(crate) async fn read_with( } Ok(( ReadResponse { - records, cursor: previous, state: Vec::new(), done: false, @@ -282,6 +281,10 @@ pub(crate) async fn read_with( } else { 0 }, + records_digest: records_digest(&records), + records, + chain_first_run_id: target.chain.first_run_id.clone(), + start_cursor: start.clone(), ..Default::default() }, closed, @@ -289,6 +292,45 @@ pub(crate) async fn read_with( )) } +/// The digest of a read's answer that a replay read checks (RR-1). It covers what a reader acts +/// on and leaves out the encoded body, so a codec change keeps it. Each field has a length +/// prefix, so two answers can't give the same bytes. +pub(crate) fn records_digest(records: &[ReadRecord]) -> Vec { + use sha2::{Digest, Sha256}; + let mut digest = Sha256::new(); + let mut field = |bytes: &[u8]| { + digest.update((bytes.len() as u64).to_be_bytes()); + digest.update(bytes); + }; + for record in records { + field(record.cursor.as_bytes()); + field(&[u8::from(record.stale)]); + match &record.record { + Some(read_record::Record::Stored(stored)) => { + field(b"stored"); + field(&stored.kind.to_be_bytes()); + field(stored.producer_id.as_bytes()); + field(&stored.attempt.to_be_bytes()); + field(&stored.sequence.to_be_bytes()); + field( + stored + .metadata + .get(temporalio_common::streams::CONTENT_HASH_KEY) + .map_or(&[][..], |hash| hash.data.as_slice()), + ); + } + Some(read_record::Record::Superseded(superseded)) => { + field(b"superseded"); + field(superseded.producer_id.as_bytes()); + field(&superseded.previous_attempt.to_be_bytes()); + field(&superseded.attempt.to_be_bytes()); + } + None => field(b"none"), + } + } + digest.finalize().to_vec() +} + /// A stored record as a reader delivers it. Fails on bytes that are no record, and on a kind no /// writer stores, such as the `SUPERSEDED` that only readers synthesize. /// diff --git a/crates/streams/src/redis/keys.rs b/crates/streams/src/redis/keys.rs index 82da443913..eed9f65b82 100644 --- a/crates/streams/src/redis/keys.rs +++ b/crates/streams/src/redis/keys.rs @@ -7,6 +7,7 @@ //! ```text //! :{::}:t: the log //! :{::}:t::meta its meta +//! :{::}:t::floors the readers that hold it //! :{::}:chain the close flag //! ``` //! @@ -58,6 +59,11 @@ impl ChainKeys { pub(crate) fn chain(&self) -> String { format!("{}:chain", self.base) } + + /// The positions-only readers that hold a topic's records, so a replay finds them again. + pub(crate) fn floors(&self, topic: &str) -> String { + format!("{}:floors", self.log(topic)) + } } /// The meta field that holds one producer attempt's newest batch. diff --git a/crates/streams/src/redis/mod.rs b/crates/streams/src/redis/mod.rs index ffef0ba29d..e305f78ff5 100644 --- a/crates/streams/src/redis/mod.rs +++ b/crates/streams/src/redis/mod.rs @@ -46,6 +46,11 @@ pub struct RedisStoreOptions { pub blocking_reads_per_node: usize, /// How long a short call waits for its reply. pub response_timeout: Duration, + /// How long a positions-only reader's hold lasts after its last read. `None` uses the + /// retention. + pub floor_lease: Option, + /// How long a hold lasts after its reader released it. + pub floor_release_grace: Duration, } impl RedisStoreOptions { @@ -58,6 +63,8 @@ impl RedisStoreOptions { retention: Duration::from_secs(7 * 24 * 60 * 60), blocking_reads_per_node: 256, response_timeout: Duration::from_secs(5), + floor_lease: None, + floor_release_grace: Duration::from_secs(24 * 60 * 60), } } } @@ -69,6 +76,8 @@ pub struct RedisStore { prefix: String, retention_ms: u64, grace_ms: u64, + floor_lease_ms: u64, + floor_release_grace_ms: u64, } impl std::fmt::Debug for RedisStore { @@ -116,6 +125,10 @@ impl RedisStore { prefix: options.key_prefix, retention_ms, grace_ms: TOMBSTONE_GRACE.as_millis() as u64, + floor_lease_ms: options + .floor_lease + .map_or(retention_ms, |lease| millis(lease).max(1)), + floor_release_grace_ms: millis(options.floor_release_grace), }) } @@ -441,6 +454,10 @@ fn wait_of(request: &StoreReadRequest) -> Duration { }) } +fn millis(duration: Duration) -> u64 { + u64::try_from(duration.as_millis()).unwrap_or(u64::MAX) +} + /// A topic's arm as a script answered it, zero when none. fn generation(arm: &str) -> StreamResult { arm.parse() @@ -470,6 +487,7 @@ impl StreamStore for RedisStore { .key(keys.log(&request.topic)) .key(keys.meta(&request.topic)) .key(keys.chain()) + .key(keys.floors(&request.topic)) .arg(self.retention_ms) .arg(self.grace_ms) .arg(session_field(&request.producer_id, request.attempt)) @@ -638,6 +656,47 @@ impl StreamStore for RedisStore { Ok(Some(trimmed)) } + async fn hold_floor( + &self, + chain: &ChainId, + topic: &str, + holder: &str, + position: &str, + ) -> StreamResult<()> { + 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. + let position = if position.is_empty() { + "0-1" + } else { + entry(position)?; + position + }; + scripts::HOLD_SCRIPT + .key(keys.floors(topic)) + .key(keys.log(topic)) + .key(keys.meta(topic)) + .arg(holder) + .arg(position) + .arg(self.floor_lease_ms) + .arg(self.grace_ms) + .invoke_async::(&mut self.shared.clone()) + .await + .mapped(Call::Write) + .map(drop) + } + + async fn release_floor(&self, chain: &ChainId, topic: &str, holder: &str) -> StreamResult<()> { + scripts::RELEASE_SCRIPT + .key(ChainKeys::new(&self.prefix, chain).floors(topic)) + .arg(holder) + .arg(self.floor_release_grace_ms) + .invoke_async::(&mut self.shared.clone()) + .await + .mapped(Call::Write) + .map(drop) + } + async fn close_chain(&self, chain: &ChainId) -> StreamResult<()> { let keys = ChainKeys::new(&self.prefix, chain); redis::pipe() diff --git a/crates/streams/src/redis/scripts.rs b/crates/streams/src/redis/scripts.rs index e98172cd1e..10e71a7ff2 100644 --- a/crates/streams/src/redis/scripts.rs +++ b/crates/streams/src/redis/scripts.rs @@ -2,13 +2,15 @@ //! //! The append script is D1b's with one more check, the topic's close flag, so a batch can't land //! after the topic's close returned (DD-96). It also reports the topic's arm, so a producer -//! notifies only while a reader sleeps (DD-118). `keep.lua` is the prelude of every script that -//! writes a log. +//! notifies only while a reader sleeps (DD-118), and its trim keeps what positions-only readers +//! hold (RR-4). `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; pub(crate) const APPEND: &str = concat!( + include_str!("../../lua/floor.lua"), include_str!("../../lua/keep.lua"), include_str!("../../lua/append.lua") ); @@ -22,6 +24,18 @@ pub(crate) static DISARM_SCRIPT: LazyLock