Apply feedback

Author: Jake-Shadle

crates/corro-agent/src/api/peer/mod.rs

@@ -751,7 +751,7 @@ fn send_change_chunks<I: Iterator<Item = rusqlite::Result<Change>>>(
             Some(Ok((changes, seqs))) => {
                 let start = Instant::now();
 
-                if changes.is_empty() && *seqs.start() == CrsqlSeq(0) && *seqs.end() == last_seq {
+                if changes.is_empty() && seqs.start() == CrsqlSeq(0) && seqs.end() == last_seq {
                     warn!(%actor_id, %version, "got an empty changes we should've had");
                     return Ok(());
                 } else {
@@ -760,7 +760,7 @@ fn send_change_chunks<I: Iterator<Item = rusqlite::Result<Change>>>(
                         changeset: Changeset::Full {
                             version,
                             changes,
-                            seqs: seqs.into(),
+                            seqs,
                             last_seq,
                             ts,
                         },

crates/corro-types/src/agent.rs

@@ -1140,7 +1140,7 @@ impl VersionsSnapshot {
             }
             let details = json!({"count": count, "range": range});
             assert_always!(count == 1, "ineffective deletion of gaps in-db", &details);
-            for version in CrsqlDbVersionRange::from(range.clone()) {
+            for version in CrsqlDbVersionRange::from(&range) {
                 self.partials.remove(&version);
             }
             self.needed.remove(range);

crates/corro-types/src/broadcast.rs

@@ -170,7 +170,7 @@ impl Changeset {
             // todo: this returns dummy version because empty set has an array of versions.
             // probably shouldn't be doing this
             Changeset::EmptySet { .. } => CrsqlDbVersionRange::empty(),
-            Changeset::Full { version, .. } => (*version..=*version).into(),
+            Changeset::Full { version, .. } => CrsqlDbVersionRange::single(*version),
         }
     }
 
@@ -549,7 +549,7 @@ pub async fn broadcast_changes(
                                     changeset: Changeset::Full {
                                         version: db_version,
                                         changes,
-                                        seqs: seqs.into(),
+                                        seqs,
                                         last_seq,
                                         ts,
                                     },

crates/corro-types/src/change.rs

@@ -1,9 +1,9 @@
-use std::{iter::Peekable, ops::RangeInclusive};
+use std::iter::Peekable;
 
 use antithesis_sdk::assert_always;
 pub use corro_api_types::SqliteValue;
 use corro_api_types::{ColumnName, TableName};
-use corro_base_types::CrsqlDbVersion;
+use corro_base_types::{CrsqlDbVersion, CrsqlSeqRange};
 use rusqlite::{Connection, Row};
 use serde::{Deserialize, Serialize};
 use serde_json::json;
@@ -104,7 +104,7 @@ impl<I> Iterator for ChunkedChanges<I>
 where
     I: Iterator<Item = rusqlite::Result<Change>>,
 {
-    type Item = Result<(Vec<Change>, RangeInclusive<CrsqlSeq>), rusqlite::Error>;
+    type Item = Result<(Vec<Change>, CrsqlSeqRange), rusqlite::Error>;
 
     fn next(&mut self) -> Option<Self::Item> {
         // previously marked as done because the Rows iterator returned None
@@ -153,7 +153,7 @@ where
 
                         return Some(Ok((
                             self.changes.drain(..).collect(),
-                            start_seq..=self.last_pushed_seq,
+                            CrsqlSeqRange::new(start_seq, self.last_pushed_seq),
                         )));
                     }
                 }
@@ -171,8 +171,8 @@ where
 
         // return buffered changes
         Some(Ok((
-            self.changes.clone(),                // no need to drain here like before
-            self.last_start_seq..=self.last_seq, // even if empty, this is all we have still applied
+            self.changes.clone(), // no need to drain here like before
+            CrsqlSeqRange::new(self.last_start_seq, self.last_seq), // even if empty, this is all we have still applied
         )))
     }
 }
@@ -262,14 +262,14 @@ pub fn insert_local_changes(
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::base::dbsri;
+    use crate::base::dbsr;
 
     #[test]
     fn test_change_chunker() {
         // empty interator
         let mut chunker = ChunkedChanges::new(vec![].into_iter(), CrsqlSeq(0), CrsqlSeq(100), 50);
 
-        assert_eq!(chunker.next(), Some(Ok((vec![], dbsri!(0, 100)))));
+        assert_eq!(chunker.next(), Some(Ok((vec![], dbsr!(0, 100)))));
         assert_eq!(chunker.next(), None);
 
         let changes: Vec<Change> = (0..100)
@@ -296,12 +296,12 @@ mod tests {
             chunker.next(),
             Some(Ok((
                 vec![changes[0].clone(), changes[1].clone()],
-                dbsri!(0, 1)
+                dbsr!(0, 1)
             )))
         );
         assert_eq!(
             chunker.next(),
-            Some(Ok((vec![changes[2].clone()], dbsri!(2, 100))))
+            Some(Ok((vec![changes[2].clone()], dbsr!(2, 100))))
         );
         assert_eq!(chunker.next(), None);
 
@@ -314,7 +314,7 @@ mod tests {
 
         assert_eq!(
             chunker.next(),
-            Some(Ok((vec![changes[0].clone()], dbsri!(0, 0))))
+            Some(Ok((vec![changes[0].clone()], dbsr!(0, 0))))
         );
         assert_eq!(chunker.next(), None);
 
@@ -330,7 +330,7 @@ mod tests {
             chunker.next(),
             Some(Ok((
                 vec![changes[0].clone(), changes[2].clone()],
-                dbsri!(0, 100)
+                dbsr!(0, 100)
             )))
         );
 
@@ -359,7 +359,7 @@ mod tests {
                     changes[7].clone(),
                     changes[8].clone()
                 ],
-                dbsri!(0, 100)
+                dbsr!(0, 100)
             )))
         );
 
@@ -383,15 +383,15 @@ mod tests {
             chunker.next(),
             Some(Ok((
                 vec![changes[2].clone(), changes[4].clone(),],
-                dbsri!(0, 4)
+                dbsr!(0, 4)
             )))
         );
 
         assert_eq!(
             chunker.next(),
             Some(Ok((
                 vec![changes[7].clone(), changes[8].clone(),],
-                dbsri!(5, 10)
+                dbsr!(5, 10)
             )))
         );
 

crates/corro-types/src/sync.rs

@@ -167,7 +167,7 @@ impl SyncStateV1 {
                         let start = cmp::max(range.start(), overlap.start());
                         let end = cmp::min(range.end(), overlap.end());
                         needs.entry(*actor_id).or_default().push(SyncNeedV1::Full {
-                            versions: (*start..=*end).into(),
+                            versions: CrsqlDbVersionRange::new(*start, *end),
                         })
                     }
                 }

rust-toolchain.toml

@@ -0,0 +1,2 @@
+[toolchain]
+channel = "stable"