From ed5b342f0cd43cca9224627ac548d57778b7d143 Mon Sep 17 00:00:00 2001 From: Eliah Rusin Date: Mon, 7 Sep 2026 23:15:47 +0300 Subject: [PATCH] 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. --- .../workflows/rust-volume-server-tests.yml | 3 + seaweed-volume/Cargo.toml | 3 + seaweed-volume/src/storage/needle_map.rs | 122 ++++++++++++++++-- 3 files changed, 118 insertions(+), 10 deletions(-) diff --git a/.github/workflows/rust-volume-server-tests.yml b/.github/workflows/rust-volume-server-tests.yml index ba8e09a39..388ff8523 100644 --- a/.github/workflows/rust-volume-server-tests.yml +++ b/.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 diff --git a/seaweed-volume/Cargo.toml b/seaweed-volume/Cargo.toml index 7a7ec1fdf..01f97e1d7 100644 --- a/seaweed-volume/Cargo.toml +++ b/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 diff --git a/seaweed-volume/src/storage/needle_map.rs b/seaweed-volume/src/storage/needle_map.rs index f2774a623..f08e896c2 100644 --- a/seaweed-volume/src/storage/needle_map.rs +++ b/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 { 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,11 +774,38 @@ impl RedbNeedleMap { })?; } - for (key, nv) in &entries { - let key_u64: u64 = (*key).into(); - let packed = pack_needle_value(nv); - table.insert(key_u64, packed.as_slice()).map_err(|e| { - io::Error::new(io::ErrorKind::Other, format!("redb insert: {}", e)) + #[cfg(not(feature = "redb-experimental-cursor"))] + { + for (key, nv) in &entries { + let key_u64: u64 = (*key).into(); + let packed = pack_needle_value(nv); + table.insert(key_u64, packed.as_slice()).map_err(|e| { + io::Error::new(io::ErrorKind::Other, format!("redb insert: {}", e)) + })?; + } + } + #[cfg(feature = "redb-experimental-cursor")] + { + let mut cursor = table + .upper_bound_mut(Bound::::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)) })?; } } @@ -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();