[Turbopack] fix race condition in database compaction (#73495)

### What?

* sort new SST files
* assign moved SST sequence number in order instead of concurrently
* add test case

Author: sokra

turbopack/crates/turbo-persistence/src/db.rs

@@ -290,7 +290,7 @@ impl TurboPersistence {
         }
 
         sst_files.retain(|seq| !deleted_files.contains(seq));
-        sst_files.sort();
+        sst_files.sort_unstable();
         let sst_files = sst_files
             .into_iter()
             .map(|seq| self.open_sst(seq))
@@ -407,11 +407,13 @@ impl TurboPersistence {
     /// new files.
     fn commit(
         &self,
-        new_sst_files: Vec<(u32, File)>,
+        mut new_sst_files: Vec<(u32, File)>,
         new_blob_files: Vec<File>,
         mut indicies_to_delete: Vec<usize>,
         mut seq: u32,
     ) -> Result<(), anyhow::Error> {
+        new_sst_files.sort_unstable_by_key(|(seq, _)| *seq);
+
         let mut new_sst_files = new_sst_files
             .into_iter()
             .map(|(seq, file)| {
@@ -433,7 +435,7 @@ impl TurboPersistence {
         {
             let mut inner = self.inner.write();
             inner.current_sequence_number = seq;
-            indicies_to_delete.sort();
+            indicies_to_delete.sort_unstable();
             removed_ssts = remove_indicies(&mut inner.static_sorted_files, &indicies_to_delete);
             inner.static_sorted_files.append(&mut new_sst_files);
         }
@@ -442,7 +444,7 @@ impl TurboPersistence {
             .into_iter()
             .map(|sst| sst.sequence_number())
             .collect::<Vec<_>>();
-        removed_ssts.sort();
+        removed_ssts.sort_unstable();
 
         if !indicies_to_delete.is_empty() {
             // Write *.del file, marking the selected files as to delete
@@ -747,14 +749,21 @@ impl TurboPersistence {
                     })
                     .collect::<Result<Vec<_>>>()?;
 
+                let move_jobs = move_jobs
+                    .into_iter()
+                    .map(|index| {
+                        let seq = sequence_number.fetch_add(1, Ordering::SeqCst) + 1;
+                        (index, seq)
+                    })
+                    .collect::<Vec<_>>();
+
                 // Move SST files
                 let mut new_sst_files = move_jobs
                     .into_par_iter()
                     .with_min_len(1)
-                    .map(|index| {
+                    .map(|(index, seq)| {
                         let index = ssts_with_ranges[index].index;
                         let sst = &static_sorted_files[index];
-                        let seq = sequence_number.fetch_add(1, Ordering::SeqCst) + 1;
                         let src_path = self.path.join(format!("{:08}.sst", sst.sequence_number()));
                         let dst_path = self.path.join(format!("{:08}.sst", seq));
                         if fs::hard_link(&src_path, &dst_path).is_err() {

turbopack/crates/turbo-persistence/src/key.rs

@@ -135,6 +135,12 @@ pub trait StoreKey: KeyBase + Ord {
     fn write_to(&self, buf: &mut Vec<u8>);
 }
 
+impl<const N: usize> StoreKey for [u8; N] {
+    fn write_to(&self, buf: &mut Vec<u8>) {
+        buf.extend_from_slice(&self[..]);
+    }
+}
+
 impl StoreKey for Vec<u8> {
     fn write_to(&self, buf: &mut Vec<u8>) {
         buf.extend_from_slice(self);

turbopack/crates/turbo-persistence/src/tests.rs

@@ -345,3 +345,106 @@ fn full_cycle() -> Result<()> {
     }
     Ok(())
 }
+
+#[test]
+fn persist_changes() -> Result<()> {
+    let tempdir = tempfile::tempdir()?;
+    let path = tempdir.path();
+
+    fn put(b: &WriteBatch<(u8, [u8; 4]), 1>, key: u8, value: u8) -> Result<()> {
+        for i in 0..2000000u32 {
+            b.put(0, (key, i.to_be_bytes()), vec![value].into())?;
+        }
+        Ok(())
+    }
+    fn check(db: &TurboPersistence, key: u8, value: u8) -> Result<()> {
+        for i in 0..200000u32 {
+            let i = i * 10;
+            assert_eq!(
+                db.get(0, &(key, i.to_be_bytes()))?.as_deref(),
+                Some(&[value][..])
+            );
+        }
+        Ok(())
+    }
+
+    {
+        let db = TurboPersistence::open(path.to_path_buf())?;
+        let b = db.write_batch::<_, 1>()?;
+        put(&b, 1, 11)?;
+        put(&b, 2, 21)?;
+        put(&b, 3, 31)?;
+        db.commit_write_batch(b)?;
+
+        check(&db, 1, 11)?;
+        check(&db, 2, 21)?;
+        check(&db, 3, 31)?;
+
+        db.shutdown()?;
+    }
+
+    println!("---");
+    {
+        let db = TurboPersistence::open(path.to_path_buf())?;
+        let b = db.write_batch::<_, 1>()?;
+        put(&b, 1, 12)?;
+        put(&b, 2, 22)?;
+        db.commit_write_batch(b)?;
+
+        check(&db, 1, 12)?;
+        check(&db, 2, 22)?;
+        check(&db, 3, 31)?;
+
+        db.shutdown()?;
+    }
+
+    {
+        let db = TurboPersistence::open(path.to_path_buf())?;
+        let b = db.write_batch::<_, 1>()?;
+        put(&b, 1, 13)?;
+        db.commit_write_batch(b)?;
+
+        check(&db, 1, 13)?;
+        check(&db, 2, 22)?;
+        check(&db, 3, 31)?;
+
+        db.shutdown()?;
+    }
+
+    println!("---");
+    {
+        let db = TurboPersistence::open(path.to_path_buf())?;
+
+        check(&db, 1, 13)?;
+        check(&db, 2, 22)?;
+        check(&db, 3, 31)?;
+
+        db.shutdown()?;
+    }
+
+    println!("---");
+    {
+        let db = TurboPersistence::open(path.to_path_buf())?;
+
+        db.compact(1.0, 3)?;
+
+        check(&db, 1, 13)?;
+        check(&db, 2, 22)?;
+        check(&db, 3, 31)?;
+
+        db.shutdown()?;
+    }
+
+    println!("---");
+    {
+        let db = TurboPersistence::open(path.to_path_buf())?;
+
+        check(&db, 1, 13)?;
+        check(&db, 2, 22)?;
+        check(&db, 3, 31)?;
+
+        db.shutdown()?;
+    }
+
+    Ok(())
+}