diff --git a/crates/streams/src/reader.rs b/crates/streams/src/reader.rs index ee00b6cb2..ff93d39f3 100644 --- a/crates/streams/src/reader.rs +++ b/crates/streams/src/reader.rs @@ -116,6 +116,10 @@ struct Tracked { /// producer. #[prost(int64, tag = "3")] started: i64, + /// The highest sequence of the newest attempt this read delivered. A record of that attempt + /// at or below it is a copy of one already delivered. + #[prost(int64, tag = "4")] + sequence: i64, } /// What one record tells the read about its producer's attempts. @@ -131,6 +135,9 @@ enum Observed { /// The newest attempt the read delivered. newest: i64, }, + /// A copy of a record of the newest attempt that the read already delivered. A store without + /// producer state lands one when a producer repeats a batch whose answer it lost. + Copy, } impl ReadState { @@ -205,11 +212,18 @@ impl ReadState { return Observed::Behind { newest: seen }; } if record.attempt == seen { + // A producer sends one batch at a time and repeats it until it has an answer, so a + // copy always comes after the record it copies. + if record.sequence > 0 && record.sequence <= tracked.sequence { + return Observed::Copy; + } + tracked.sequence = tracked.sequence.max(record.sequence); return Observed::Known; } let started = tracked.started; tracked.attempt = record.attempt; tracked.started = record.task_started_event_id; + tracked.sequence = record.sequence; self.forget_beyond(MAX_TRACKED_PRODUCERS); let superseded = seen > 0 && supersedes(seen, started, record); Observed::Newer(superseded.then(|| Supersession { @@ -234,6 +248,11 @@ impl ReadState { /// most [MAX_TRACKED_PRODUCERS] producers in at most [MAX_STATE_BYTES], and the next attempt of one /// it forgot reads without `SUPERSEDED`. /// +/// A record of a producer's newest attempt at or below the highest sequence the read delivered of +/// it is a copy, and the read drops it, on every store. A record with no sequence is never a +/// copy. A read that starts from a cursor without state knows only the cursor record's sequence, +/// so it can deliver a copy whose first copy came before the cursor. +/// /// A record that does not parse fails the call as `RECORD` with its cursor, so the caller resumes /// past it on purpose. Skipping it would lose data no one hears of. The records before it in the /// same page are returned first, and the next call fails. @@ -329,6 +348,10 @@ pub(crate) async fn read_with( Err(_) => break, }; let stale = match state.observe(&record) { + Observed::Copy => { + previous = cursor; + continue; + } Observed::Known => false, Observed::Newer(superseded) => { if let Some(superseded) = superseded { @@ -415,6 +438,8 @@ mod tests { struct Log { entries: Mutex>>, lookups: AtomicUsize, + /// The last sequence each producer attempt wrote. + sequences: Mutex>, } impl Log { @@ -424,8 +449,24 @@ mod tests { entries.len().to_string() } + /// Writes the attempt's next record. fn write(&self, producer: &str, attempt: i64, data: &[u8]) -> String { - self.push(data_record(producer, attempt, data).encode_to_vec()) + let mut sequences = self.sequences.lock().unwrap(); + let sequence = sequences + .entry((producer.to_string(), attempt)) + .or_default(); + *sequence += 1; + let sequence = *sequence; + drop(sequences); + self.write_at(producer, attempt, sequence, data) + } + + fn write_at(&self, producer: &str, attempt: i64, sequence: i64, data: &[u8]) -> String { + let record = StreamRecord { + sequence, + ..data_record(producer, attempt, data) + }; + self.push(record.encode_to_vec()) } } @@ -569,8 +610,165 @@ mod tests { const DATA: i32 = StreamRecordKind::Data as i32; const SUPERSEDED: i32 = 3; + /// Observes a record with no sequence, which is never a copy. fn observe(state: &mut ReadState, producer: &str, attempt: i64) -> Observed { - state.observe(&data_record(producer, attempt, b"")) + state.observe(&StreamRecord { + sequence: 0, + ..data_record(producer, attempt, b"") + }) + } + + fn observe_at(state: &mut ReadState, producer: &str, attempt: i64, sequence: i64) -> Observed { + state.observe(&StreamRecord { + sequence, + ..data_record(producer, attempt, b"") + }) + } + + #[test] + fn a_record_at_or_below_the_highest_sequence_of_its_attempt_is_a_copy() { + let mut state = ReadState::default(); + assert_eq!(observe_at(&mut state, "p", 1, 1), Observed::Newer(None)); + assert_eq!(observe_at(&mut state, "p", 1, 2), Observed::Known); + assert_eq!(observe_at(&mut state, "p", 1, 2), Observed::Copy); + assert_eq!(observe_at(&mut state, "p", 1, 1), Observed::Copy); + assert_eq!(observe_at(&mut state, "p", 1, 0), Observed::Known); + // A gap is no copy, and the highest sequence moves past it. + assert_eq!(observe_at(&mut state, "p", 1, 5), Observed::Known); + assert_eq!(observe_at(&mut state, "p", 1, 4), Observed::Copy); + // A new attempt counts from its own records. + assert!(matches!( + observe_at(&mut state, "p", 2, 1), + Observed::Newer(Some(_)) + )); + assert_eq!(observe_at(&mut state, "p", 2, 1), Observed::Copy); + // An older attempt's copy is still delivered, as stale. + assert_eq!( + observe_at(&mut state, "p", 1, 1), + Observed::Behind { newest: 2 } + ); + assert_eq!(observe_at(&mut state, "q", 1, 1), Observed::Newer(None)); + } + + #[tokio::test] + async fn a_copy_is_dropped_and_the_cursor_moves_past_it() { + let log = Log::default(); + log.write_at("p", 1, 1, b"1"); + log.write_at("q", 1, 1, b"2"); + // `p` lost the answer to its batch and sent it again. + log.write_at("p", 1, 1, b"1"); + log.write_at("p", 1, 2, b"3"); + let all = read(&log, BEGINNING, &[]).await.unwrap(); + assert_eq!(values(&all), ["1", "2", "3"]); + let first = read_page( + &log, + &target(), + &ReadRequest { + after: BEGINNING.to_string(), + max_records: 3, + ..Default::default() + }, + ) + .await + .unwrap(); + assert_eq!(values(&first), ["1", "2"]); + assert_eq!(first.cursor, cursor("3")); + let next = read(&log, &first.cursor, &first.state).await.unwrap(); + assert_eq!(values(&next), ["3"]); + } + + #[tokio::test] + async fn a_read_from_a_cursor_knows_the_sequence_of_the_cursor_record() { + let log = Log::default(); + log.write_at("p", 1, 1, b"1"); + log.write_at("p", 1, 2, b"2"); + log.write_at("p", 1, 2, b"2"); + log.write_at("p", 1, 1, b"1"); + let resumed = read(&log, &cursor("2"), &[]).await.unwrap(); + assert!(resumed.records.is_empty()); + assert_eq!(resumed.cursor, cursor("4")); + // A cursor before the first copy delivers its batch once. + let earlier = read(&log, &cursor("1"), &[]).await.unwrap(); + assert_eq!(values(&earlier), ["2"]); + } + + fn values(response: &ReadResponse) -> Vec { + response + .records + .iter() + .filter_map(|record| match record.record.as_ref().unwrap() { + read_record::Record::Stored(stored) => { + Some(String::from_utf8(stored.body.clone().unwrap().data).unwrap()) + } + read_record::Record::Superseded(_) => None, + }) + .collect() + } + + /// The reader rules every SDK checks itself against. + #[tokio::test] + async fn the_reader_rules_match_the_shared_vectors() { + let vectors: serde_json::Value = + serde_json::from_str(include_str!("../testdata/reader_rules.json")).unwrap(); + for case in vectors["cases"].as_array().unwrap() { + let label = case["label"].as_str().unwrap(); + let log = Log::default(); + for (index, record) in case["records"].as_array().unwrap().iter().enumerate() { + let stored = StreamRecord { + topic: record["topic"].as_str().unwrap_or("out").to_string(), + kind: record["kind"].as_i64().unwrap_or(1) as i32, + producer_id: record["producer"].as_str().unwrap().to_string(), + attempt: record["attempt"].as_i64().unwrap(), + sequence: record["sequence"].as_i64().unwrap(), + body: Some(Payload { + data: index.to_string().into_bytes(), + ..Default::default() + }), + ..Default::default() + }; + log.push(stored.encode_to_vec()); + } + let after = match case["resume_after"].as_u64() { + Some(index) => cursor(&(index + 1).to_string()), + None => BEGINNING.to_string(), + }; + let page = read(&log, &after, &[]).await.unwrap(); + let delivered: Vec = page + .records + .iter() + .filter_map(|record| match record.record.as_ref().unwrap() { + read_record::Record::Stored(stored) => { + let index = String::from_utf8(stored.body.clone().unwrap().data); + Some(index.unwrap().parse().unwrap()) + } + read_record::Record::Superseded(_) => None, + }) + .collect(); + let expected: Vec = case["delivered"] + .as_array() + .unwrap() + .iter() + .map(|index| index.as_u64().unwrap()) + .collect(); + assert_eq!(delivered, expected, "{label}"); + let stale: Vec = page + .records + .iter() + .filter(|record| record.stale) + .map(|record| match record.record.as_ref().unwrap() { + read_record::Record::Stored(stored) => { + let index = String::from_utf8(stored.body.clone().unwrap().data); + index.unwrap().parse().unwrap() + } + read_record::Record::Superseded(_) => unreachable!(), + }) + .collect(); + let expected_stale: Vec = case["stale"] + .as_array() + .map(|indexes| indexes.iter().map(|i| i.as_u64().unwrap()).collect()) + .unwrap_or_default(); + assert_eq!(stale, expected_stale, "{label}"); + } } #[test] diff --git a/crates/streams/testdata/reader_rules.json b/crates/streams/testdata/reader_rules.json new file mode 100644 index 000000000..ff47cf47c --- /dev/null +++ b/crates/streams/testdata/reader_rules.json @@ -0,0 +1,83 @@ +{ + "comment": "The rules a reader applies on every store. Each case lists the stored records of one topic in log order, and the indexes of the records a read from the start delivers (or a read that resumes after the record at index resume_after, with no state). kind defaults to 1 (DATA) and topic to \"out\".", + "cases": [ + { + "label": "a copy after a lost answer is dropped", + "records": [ + {"producer": "p", "attempt": 1, "sequence": 1}, + {"producer": "q", "attempt": 1, "sequence": 1}, + {"producer": "p", "attempt": 1, "sequence": 1}, + {"producer": "p", "attempt": 1, "sequence": 2} + ], + "delivered": [0, 1, 3] + }, + { + "label": "a copy of a whole batch is dropped", + "records": [ + {"producer": "p", "attempt": 1, "sequence": 1}, + {"producer": "p", "attempt": 1, "sequence": 2}, + {"producer": "p", "attempt": 1, "sequence": 1}, + {"producer": "p", "attempt": 1, "sequence": 2}, + {"producer": "p", "attempt": 1, "sequence": 3} + ], + "delivered": [0, 1, 4] + }, + { + "label": "a record with no sequence is never a copy", + "records": [ + {"producer": "p", "attempt": 1, "sequence": 0}, + {"producer": "p", "attempt": 1, "sequence": 0} + ], + "delivered": [0, 1] + }, + { + "label": "a gap in the sequence is no copy", + "records": [ + {"producer": "p", "attempt": 1, "sequence": 1}, + {"producer": "p", "attempt": 1, "sequence": 4}, + {"producer": "p", "attempt": 1, "sequence": 3} + ], + "delivered": [0, 1] + }, + { + "label": "a copy of an older attempt is delivered as stale", + "records": [ + {"producer": "p", "attempt": 1, "sequence": 1}, + {"producer": "p", "attempt": 2, "sequence": 1}, + {"producer": "p", "attempt": 1, "sequence": 1} + ], + "delivered": [0, 1, 2], + "stale": [2] + }, + { + "label": "producers count their sequences apart", + "records": [ + {"producer": "p", "attempt": 1, "sequence": 1}, + {"producer": "q", "attempt": 1, "sequence": 1}, + {"producer": "q", "attempt": 1, "sequence": 1} + ], + "delivered": [0, 1] + }, + { + "label": "a read from a cursor knows the cursor record's sequence", + "records": [ + {"producer": "p", "attempt": 1, "sequence": 1}, + {"producer": "p", "attempt": 1, "sequence": 2}, + {"producer": "p", "attempt": 1, "sequence": 2}, + {"producer": "p", "attempt": 1, "sequence": 3} + ], + "resume_after": 1, + "delivered": [3] + }, + { + "label": "a read from a cursor can deliver a copy whose first copy came before it", + "records": [ + {"producer": "p", "attempt": 1, "sequence": 1}, + {"producer": "q", "attempt": 1, "sequence": 1}, + {"producer": "p", "attempt": 1, "sequence": 1} + ], + "resume_after": 1, + "delivered": [2] + } + ] +} diff --git a/crates/streams/tests/conformance.rs b/crates/streams/tests/conformance.rs index cd54a18c2..681240dec 100644 --- a/crates/streams/tests/conformance.rs +++ b/crates/streams/tests/conformance.rs @@ -1150,14 +1150,14 @@ mod cases { .await; assert_eq!(repeat, kept); // The retired attempt no longer does, which is why only an attempt whose last append - // returned may be retired. + // returned may be retired. Readers still drop the second copy. let again = Producer::new(case, &chain, OUT, "p", 1) .append(&["1"]) .await; assert_ne!(again.first_position, first.first_position); assert_eq!( values(&read(case, &chain, OUT, BEGINNING).await.unwrap()), - ["1", "2", "1"] + ["1", "2"] ); }