fix flaky snapshot tests (#1020)

libsql-server/src/replication/snapshot.rs

@@ -588,7 +588,7 @@ mod test {
 
     use super::*;
 
-    async fn assert_dir_is_empty(p: &Path) {
+    async fn dir_is_empty(p: &Path) -> bool {
         // there is nothing left in the to_compact directory
         if p.try_exists().unwrap() {
             let mut dir = tokio::fs::read_dir(p).await.unwrap();
@@ -599,7 +599,9 @@ mod test {
                 }
             }
 
-            assert_eq!(count, 0);
+            count == 0
+        } else {
+            true
         }
     }
 
@@ -689,8 +691,7 @@ mod test {
 
         // assert that all indexes are covered
         assert_eq!((start_idx, end_idx), (0, 3));
-
-        assert_dir_is_empty(&to_compact_path).await;
+        assert!(dir_is_empty(&to_compact_path).await);
     }
 
     /// Simulate an empty pending snapshot left by the logger if the logswapping operation was
@@ -715,8 +716,8 @@ mod test {
         tokio::time::sleep(Duration::from_millis(1000)).await;
 
         // emtpy snapshot was discarded
-        assert_dir_is_empty(&tmp.path().join("to_compact")).await;
-        assert_dir_is_empty(&tmp.path().join("snapshots")).await;
+        assert!(dir_is_empty(&tmp.path().join("to_compact")).await);
+        assert!(dir_is_empty(&tmp.path().join("snapshots")).await);
     }
 
     /// In this test, we send a bunch of snapshot to the compactor, and see if it handles it
@@ -784,30 +785,33 @@ mod test {
         .await
         .unwrap();
 
-        // wait a bit for snapshot to be compated
-        tokio::time::sleep(Duration::from_millis(1500)).await;
-
         // no error occured: the loop is still running.
         assert!(!compactor.sender.is_closed());
         assert!(tmp.path().join("snapshots").exists());
-        let mut dir = tokio::fs::read_dir(tmp.path().join("snapshots"))
-            .await
-            .unwrap();
         let mut start_idx = u64::MAX;
         let mut end_idx = u64::MIN;
-        while let Some(entry) = dir.next_entry().await.unwrap() {
-            if entry.file_type().await.unwrap().is_file() {
-                let (_, start, end) =
-                    parse_snapshot_name(entry.file_name().to_str().unwrap()).unwrap();
-                start_idx = start_idx.min(start);
-                end_idx = end_idx.max(end);
+        while end_idx != 9 {
+            let mut dir = tokio::fs::read_dir(tmp.path().join("snapshots"))
+                .await
+                .unwrap();
+            while let Some(entry) = dir.next_entry().await.unwrap() {
+                if entry.file_type().await.unwrap().is_file() {
+                    let name = entry.file_name();
+                    let name = name.to_str().unwrap();
+                    let (_, start, end) = parse_snapshot_name(name).unwrap();
+                    start_idx = start_idx.min(start);
+                    end_idx = end_idx.max(end);
+                }
             }
+            tokio::time::sleep(Duration::from_millis(100)).await;
         }
 
         // assert that all indexes are covered
-        assert_eq!((start_idx, end_idx), (0, 9));
+        assert_eq!(start_idx, 0);
 
-        assert_dir_is_empty(&to_compact_path).await;
+        while !dir_is_empty(&to_compact_path).await {
+            tokio::time::sleep(Duration::from_millis(100)).await;
+        }
     }
 
     #[tokio::test]