handle an extra case where we can collapse ranges properly. further r…

…estrict the number of rows sqlite needs to visit

crates/corro-agent/src/agent/tests.rs

@@ -999,6 +999,7 @@ fn test_store_empty_changeset() -> eyre::Result<()> {
         [actor_id],
     )?;
 
+    // store an empty version 1..=2 when 1 was considered non-empty
     {
         let tx = conn.transaction()?;
         assert_eq!(
@@ -1035,6 +1036,7 @@ fn test_store_empty_changeset() -> eyre::Result<()> {
         [actor_id],
     )?;
 
+    // insert empty version 5..=7 that does not overlap with anything else
     {
         let tx = conn.transaction()?;
         assert_eq!(
@@ -1082,6 +1084,7 @@ fn test_store_empty_changeset() -> eyre::Result<()> {
         }
     );
 
+    // insert empty changes 3..=6 which touches non-empty version 3 and empty versions 5..=7
     {
         let tx = conn.transaction()?;
         assert_eq!(
@@ -1120,6 +1123,7 @@ fn test_store_empty_changeset() -> eyre::Result<()> {
         [actor_id],
     )?;
 
+    // insert changes that hae the same start as already emptied rows, but go up higher
     {
         let tx = conn.transaction()?;
         assert_eq!(
@@ -1162,6 +1166,8 @@ fn test_store_empty_changeset() -> eyre::Result<()> {
         }
     );
 
+    // insert changes that hae the same start as already emptied rows, but go up higher
+
     {
         let tx = conn.transaction()?;
         assert_eq!(
@@ -1209,6 +1215,8 @@ fn test_store_empty_changeset() -> eyre::Result<()> {
         [actor_id],
     )?;
 
+    // insert empties that don't touch any other rows
+
     {
         let tx = conn.transaction()?;
         assert_eq!(
@@ -1268,6 +1276,8 @@ fn test_store_empty_changeset() -> eyre::Result<()> {
         }
     );
 
+
+    // empties multiple non-empty versions (12 and 13) and touches already emptied version (14)
     {
         let tx = conn.transaction()?;
         assert_eq!(
@@ -1310,6 +1320,7 @@ fn test_store_empty_changeset() -> eyre::Result<()> {
         [actor_id],
     )?;
 
+    // empties a version in between 2 ranges of empties
     {
         let tx = conn.transaction()?;
         assert_eq!(
@@ -1343,5 +1354,39 @@ fn test_store_empty_changeset() -> eyre::Result<()> {
         }
     );
 
+    // empties versions overlapping the end of a previous range
+    {
+        let tx = conn.transaction()?;
+        assert_eq!(
+            store_empty_changeset(&tx, actor_id, Version(15)..=Version(23))?,
+            1
+        );
+        tx.commit()?;
+    }
+
+    let rows = conn
+        .prepare("SELECT actor_id, start_version, end_version FROM __corro_bookkeeping")?
+        .query_map([], |row| {
+            Ok(CorroBook {
+                actor_id: row.get(0)?,
+                start_version: row.get(1)?,
+                end_version: row.get(2)?,
+            })
+        })
+        .and_then(|rows| rows.collect::<rusqlite::Result<Vec<_>>>())?;
+
+    println!("rows: {rows:?}");
+
+    assert_eq!(rows.len(), 1);
+
+    assert_eq!(
+        rows[0],
+        CorroBook {
+            actor_id,
+            start_version: Version(1),
+            end_version: Some(Version(23))
+        }
+    );
+
     Ok(())
 }

crates/corro-agent/src/agent/util.rs

@@ -789,6 +789,7 @@ pub fn store_empty_changeset(
         DELETE FROM __corro_bookkeeping
             WHERE
                 actor_id = :actor_id AND
+                -- try to find the previous range
                 start_version >= COALESCE((
                     SELECT start_version
                         FROM __corro_bookkeeping
@@ -799,20 +800,34 @@ pub fn store_empty_changeset(
                         LIMIT 1
                 ), 1)
                 AND
+                -- try to find the next range
+                start_version <= COALESCE((
+                    SELECT start_version
+                        FROM __corro_bookkeeping
+                        WHERE
+                            actor_id = :actor_id AND
+                            start_version > :end
+                        ORDER BY start_version ASC
+                        LIMIT 1
+                ), :end + 1)
+                AND
                 (
-                    -- start_version is between start and end of range AND no end_version
+                    -- [:start]---[start_version]---[:end]
                     ( start_version BETWEEN :start AND :end AND end_version IS NULL ) OR
 
-                    -- start_version and end_version are within the range
+                    -- [start_version]---[:start]---[:end]---[end_version]
                     ( start_version >= :start AND end_version <= :end ) OR
 
-                    -- range being inserted is partially contained within another
+                    -- [:start]---[start_version]---[:end]---[end_version]
                     ( start_version <= :end AND end_version >= :end ) OR
 
-                    -- start_version = end + 1 (to collapse ranges)
+                    -- [start_version]---[:start]---[end_version]---[:end]
+                    ( start_version <= :start AND end_version <= :end ) OR
+
+                    -- ---[:end][start_version]
                     ( start_version = :end + 1 AND end_version IS NOT NULL ) OR
 
-                    -- end_version = start - 1 (to collapse ranges)
+                    -- [end_version][:start]---
                     ( end_version = :start - 1 )
                 )
             RETURNING start_version, end_version",