[Brook] Dropped copies of delivered records in the reader.

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<Vec<Vec<u8>>>,
         lookups: AtomicUsize,
+        /// The last sequence each producer attempt wrote.
+        sequences: Mutex<HashMap<(String, i64), i64>>,
     }
 
     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<String> {
+        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<u64> = 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<u64> = case["delivered"]
+                .as_array()
+                .unwrap()
+                .iter()
+                .map(|index| index.as_u64().unwrap())
+                .collect();
+            assert_eq!(delivered, expected, "{label}");
+            let stale: Vec<u64> = 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<u64> = 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]

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"]
         );
     }