rust volume: optional redb insert_before bulk load (#11205)

* rust volume: quick-repair redb on durable checkpoints

set_quick_repair(true) on the durable checkpoint transaction so an
OOM-killed volume server opens without a full-file repair scan.

* rust volume: optional redb insert_before bulk load

Behind redb-experimental-cursor (default off). Production binary
stays on sorted insert(). CI unit tests run both feature settings.

* rust volume: exercise insert_before across leaf splits

Replace the 5-key cfg clone with a 4000-key reverse-order rebuild
so CursorMut::insert_before hits page splits. CI runs the feature
only on storage::needle_map unit tests.

.github/workflows/rust-volume-server-tests.yml

@@ -62,6 +62,9 @@ jobs:
       - name: Run Rust unit tests
         run: cd seaweed-volume && cargo test
 
+      - name: Run Rust unit tests (redb experimental cursor)
+        run: cd seaweed-volume && cargo test --features redb-experimental-cursor --lib storage::needle_map
+
   rust-integration-tests:
     name: Rust Integration Tests
     runs-on: ubuntu-22.04

seaweed-volume/Cargo.toml

@@ -16,6 +16,9 @@ path = "src/main.rs"
 # Disable with --no-default-features for 4-byte offsets (32GB max volume size).
 default = ["5bytes"]
 5bytes = []
+# Unstable redb cursor bulk-load for full_rebuild. Off in production.
+# Pulls redb's experimental_cursor (and therefore experimental-api-5).
+redb-experimental-cursor = ["redb/experimental_cursor"]
 
 [dependencies]
 # Async runtime

seaweed-volume/src/storage/needle_map.rs

@@ -13,6 +13,9 @@ use std::io::{self, Read, Seek, Write};
 use std::path::Path;
 use std::sync::atomic::{AtomicI64, AtomicU64, Ordering};
 
+#[cfg(feature = "redb-experimental-cursor")]
+use std::ops::Bound;
+
 mod compact_map;
 pub mod file_pool;
 mod idx_metric;
@@ -489,13 +492,16 @@ impl RedbNeedleMap {
     /// much of the .idx the table reflects, so a reload replays only the
     /// tail appended after it. When `sync_idx` is true the .idx is fsynced
     /// first (see [`checkpoint`](Self::checkpoint)).
+    /// Durable commit uses `set_quick_repair(true)` so the next open after a
+    /// crash does not scan the whole file.
     fn begin_checkpoint(&self, sync_idx: bool) -> io::Result<redb::WriteTransaction> {
         if sync_idx {
             self.sync()?;
         }
-        let txn = self.db.begin_write().map_err(|e| {
+        let mut txn = self.db.begin_write().map_err(|e| {
             io::Error::new(io::ErrorKind::Other, format!("redb begin_write: {}", e))
         })?;
+        txn.set_quick_repair(true);
         if self.idx_file.is_some() {
             let mut meta = txn.open_table(META_TABLE).map_err(|e| {
                 io::Error::new(io::ErrorKind::Other, format!("redb open meta: {}", e))
@@ -567,14 +573,18 @@ impl RedbNeedleMap {
         let meta = txn
             .open_table(META_TABLE)
             .map_err(|e| io::Error::new(io::ErrorKind::Other, format!("redb open meta: {}", e)))?;
-        match meta.get(META_IDX_SIZE) {
+        // experimental-api-5 drops inherent ReadOnlyTable::get ('static guard).
+        // ReadableTable::get guard borrows `meta`; bind the match so the
+        // temporary Result is dropped before `meta`.
+        let result = match meta.get(META_IDX_SIZE) {
             Ok(Some(guard)) => Ok(Some(guard.value())),
             Ok(None) => Ok(None),
             Err(e) => Err(io::Error::new(
                 io::ErrorKind::Other,
                 format!("redb get meta: {}", e),
             )),
-        }
+        };
+        result
     }
 
     /// Load from an .idx file, reusing an existing .rdb if it is consistent.
@@ -764,6 +774,8 @@ impl RedbNeedleMap {
                     })?;
                 }
 
+                #[cfg(not(feature = "redb-experimental-cursor"))]
+                {
                     for (key, nv) in &entries {
                         let key_u64: u64 = (*key).into();
                         let packed = pack_needle_value(nv);
@@ -772,6 +784,31 @@ impl RedbNeedleMap {
                         })?;
                     }
                 }
+                #[cfg(feature = "redb-experimental-cursor")]
+                {
+                    let mut cursor = table
+                        .upper_bound_mut(Bound::<u64>::Unbounded)
+                        .map_err(|e| {
+                            io::Error::new(
+                                io::ErrorKind::Other,
+                                format!("redb upper_bound_mut: {}", e),
+                            )
+                        })?;
+                    for (key, nv) in &entries {
+                        let key_u64: u64 = (*key).into();
+                        let packed = pack_needle_value(nv);
+                        cursor.insert_before(key_u64, packed.as_slice()).map_err(|e| {
+                            io::Error::new(
+                                io::ErrorKind::Other,
+                                format!("redb insert_before: {}", e),
+                            )
+                        })?;
+                    }
+                    cursor.close().map_err(|e| {
+                        io::Error::new(io::ErrorKind::Other, format!("redb cursor close: {}", e))
+                    })?;
+                }
+            }
             txn.commit()
                 .map_err(|e| io::Error::new(io::ErrorKind::Other, format!("redb commit: {}", e)))?;
 
@@ -864,14 +901,18 @@ impl RedbNeedleMap {
         let table = txn
             .open_table(NEEDLE_TABLE)
             .map_err(|e| io::Error::new(io::ErrorKind::Other, format!("redb open_table: {}", e)))?;
-        match table.get(key_u64) {
+        // experimental-api-5 drops inherent ReadOnlyTable::get ('static guard).
+        // ReadableTable::get guard borrows `table`; bind the match so the
+        // temporary Result is dropped before `table`.
+        let result = match table.get(key_u64) {
             Ok(Some(guard)) => Ok(packed_to_needle_value(guard.value())),
             Ok(None) => Ok(None),
             Err(e) => Err(io::Error::new(
                 io::ErrorKind::Other,
                 format!("redb get: {}", e),
             )),
-        }
+        };
+        result
     }
 
     /// Mark a needle as deleted. Appends tombstone to .idx file, negates size in redb.
@@ -1612,6 +1653,67 @@ mod tests {
         );
     }
 
+    #[cfg(feature = "redb-experimental-cursor")]
+    #[test]
+    fn test_redb_full_rebuild_insert_before_reads_back_thousands_of_shuffled_keys() {
+        // Enough keys to split leaves. insert_before vs table.insert only
+        // diverges across page boundaries (pending-insert buffer / close).
+        const N: u64 = 4000;
+        let mut idx_data = Vec::new();
+        for i in (1..=N).rev() {
+            idx::write_index_entry(
+                &mut idx_data,
+                NeedleId(i),
+                Offset::from_actual_offset((8 * i) as i64),
+                Size(i as i32),
+            )
+            .unwrap();
+        }
+        idx::write_index_entry(
+            &mut idx_data,
+            NeedleId(1),
+            Offset::from_actual_offset(200),
+            Size(200),
+        )
+        .unwrap();
+        idx::write_index_entry(
+            &mut idx_data,
+            NeedleId(2),
+            Offset::default(),
+            TOMBSTONE_FILE_SIZE,
+        )
+        .unwrap();
+
+        let dir = tempfile::tempdir().unwrap();
+        let db_path = dir.path().join("test.rdb");
+        let mut cursor = Cursor::new(idx_data);
+        let nm = RedbNeedleMap::load_from_idx(
+            db_path.to_str().unwrap(),
+            &mut cursor,
+            Version::current(),
+            redb_test_cache(),
+        )
+        .unwrap();
+
+        let v1 = nm.get(NeedleId(1)).unwrap().unwrap();
+        assert_eq!(v1.size, Size(200));
+        assert_eq!(v1.offset, Offset::from_actual_offset(200));
+        assert!(nm.get(NeedleId(2)).unwrap().is_none());
+        let v_n = nm.get(NeedleId(N)).unwrap().unwrap();
+        assert_eq!(v_n.size, Size(N as i32));
+        let v3 = nm.get(NeedleId(3)).unwrap().unwrap();
+        assert_eq!(v3.size, Size(3));
+        assert_eq!(v3.offset, Offset::from_actual_offset(24));
+
+        let mut live = 0u64;
+        nm.ascending_visit(|_, _| {
+            live += 1;
+            Ok(())
+        })
+        .unwrap();
+        assert_eq!(live, N - 1);
+    }
+
     #[test]
     fn test_redb_needle_map_put_get() {
         let dir = tempfile::tempdir().unwrap();