diff --git a/Cargo.lock b/Cargo.lock index 28e1411..468a305 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -841,6 +841,8 @@ dependencies = [ "hex", "reqwest", "rusty-leveldb", + "serde", + "serde_json", "thiserror 2.0.18", "tokio", "tracing", diff --git a/Cargo.toml b/Cargo.toml index d931739..a77b041 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -17,6 +17,8 @@ categories = ["database", "network-programming"] thiserror = "2" tracing = "0.1" +serde = "1" +serde_json = "1" base64 = "0.22" blake3 = "1.8" hex = "0.4" diff --git a/minikv-core/Cargo.toml b/minikv-core/Cargo.toml index 0679d30..4ece164 100644 --- a/minikv-core/Cargo.toml +++ b/minikv-core/Cargo.toml @@ -13,6 +13,8 @@ categories.workspace = true [dependencies] thiserror.workspace = true tracing.workspace = true +serde = { workspace = true, features = ["derive"] } +serde_json = { workspace = true } base64 = { workspace = true } blake3 = { workspace = true } hex = { workspace = true } @@ -20,6 +22,8 @@ dashmap = { workspace = true } rusty-leveldb = { workspace = true } reqwest = { workspace = true, features = ["rustls", "stream", "json"] } bytes = { workspace = true } +tokio = { workspace = true, features = ["full"] } + [dev-dependencies] tokio = { workspace = true, features = ["full"] } diff --git a/minikv-core/src/error.rs b/minikv-core/src/error.rs index 1f32622..0ebd655 100644 --- a/minikv-core/src/error.rs +++ b/minikv-core/src/error.rs @@ -20,6 +20,13 @@ pub enum Error { /// A remote volume server returned an unexpected HTTP status code. #[error("Remote volume returned {status} for {url}")] RemoteStatus { status: u16, url: String }, + + /// Rebuild encountered a volume that returned malformed autoindex JSON. + #[error("Autoindex JSON parse error for {url}: {source}")] + AutoindexParse { + url: String, + source: serde_json::Error, + }, } impl From for Error { diff --git a/minikv-core/src/lib.rs b/minikv-core/src/lib.rs index a6bcec1..ce2a03a 100644 --- a/minikv-core/src/lib.rs +++ b/minikv-core/src/lib.rs @@ -1,6 +1,7 @@ pub mod error; pub mod hashing; pub mod locking; +pub mod rebuild; pub mod record; pub mod replication; pub mod state; diff --git a/minikv-core/src/rebuild.rs b/minikv-core/src/rebuild.rs new file mode 100644 index 0000000..d4ece79 --- /dev/null +++ b/minikv-core/src/rebuild.rs @@ -0,0 +1,272 @@ +//! Rebuild LevelDB metadata by scanning all configured volume servers. +//! +//! This operation is intended for disaster recovery when the metadata +//! database is lost or corrupted but object files remain intact on the +//! volume servers. +//! +//! The rebuild process is fully deterministic with respect to: +//! - key decoding +//! - volume selection (`key2volume`) +//! - replica ordering +//! +//! It does not recover hash or content-type metadata, as that information +//! exists only in LevelDB and is not stored on volume servers. +//! +//! # Algorithm +//! +//! 1. Delete all existing records from LevelDB (destructive). +//! 2. For each configured volume: +//! - Detect whether subvolumes (`svXX/`) are used. +//! - Traverse the two-level hex directory structure. +//! 3. For each file entry: +//! - Base64-decode the filename to obtain the raw key. +//! - Merge the current volume into the key's record. +//! - Reorder volumes according to `key2volume`, preserving unknown +//! volumes at the end. +//! 4. Write the reconstructed record with: +//! - `deleted = No` +//! - `hash = None` +//! - `content_type = None` +//! +//! # Expected Directory Layout +//! +//! ```text +//! http://vol/ ← either hex dirs or svXX/ +//! http://vol/svXX/ ← optional subvolume +//! http://vol/XX/ ← first-level hex dir +//! http://vol/XX/YY/ ← second-level hex dir +//! http://vol/XX/YY/ ← base64-encoded key filename +//! ``` +//! +//! # Concurrency +//! +//! Leaf directories are processed in parallel, limited by a semaphore +//! of 128 concurrent tasks. +//! +//! # Safety +//! +//! This operation permanently clears the metadata database before scanning. +//! It must not be executed against a healthy database. + +use std::sync::Arc; + +use serde::Deserialize; +use tokio::sync::Semaphore; +use tracing::{error, info, warn}; + +use crate::error::Error; +use crate::record::{Deleted, Record}; +use crate::replication::remote_get; +use crate::state::AppState; +use crate::volumes::key_to_volume; + +/// A single entry in an nginx autoindex JSON response. +#[derive(Debug, Deserialize)] +pub struct AutoindexEntry { + pub name: String, + #[serde(rename = "type")] + pub entry_type: String, + pub mtime: String, +} + +/// Fetch and parse an nginx autoindex JSON listing from `url`. +async fn get_listing(client: &reqwest::Client, url: &str) -> Result, Error> { + let body = remote_get(client, url).await?; + let entries: Vec = + serde_json::from_slice(&body).map_err(|e| Error::AutoindexParse { + url: url.to_string(), + source: e, + })?; + Ok(entries) +} + +/// Return `true` if `entry` looks like a valid 2-char hex directory. +fn is_hex_dir(entry: &AutoindexEntry) -> bool { + entry.entry_type == "directory" + && entry.name.len() == 2 + && entry.name.chars().all(|c| c.is_ascii_hexdigit()) +} + +/// Return `true` if `entry` looks like a subvolume directory (`svXX`). +fn is_subvolume_dir(entry: &AutoindexEntry) -> bool { + entry.entry_type == "directory" + && entry.name.len() == 4 + && entry.name.starts_with("sv") + && entry.name[2..].chars().all(|c| c.is_ascii_hexdigit()) +} + +/// Merge a single object (identified by its base64 filename) into the DB. +/// +/// - Decodes the filename into raw key bytes. +/// - Acquires a per-key lock to prevent concurrent modification. +/// - Computes the ideal replica ordering via `key2volume`. +/// - Merges the current volume into the existing record. +/// - Reorders volumes deterministically. +/// - Writes a reconstructed `Record`. +/// +/// Hard-deleted records are treated as non-existent during rebuild. +/// Hash and content-type metadata cannot be restored. +async fn rebuild_entry(state: &AppState, vol: &str, b64name: &str) -> bool { + // Decode base64 filename → raw key bytes. + use base64::{Engine as _, engine::general_purpose::STANDARD as B64}; + let key = match B64.decode(b64name) { + Ok(k) => k, + Err(e) => { + warn!(?e, b64name, "rebuild: base64 decode error"); + return false; + } + }; + + // Acquire per-key lock (non-blocking; if already held, skip this entry). + let _guard = match state.key_lock.try_lock(&String::from_utf8_lossy(&key)) { + Some(g) => g, + None => { + warn!(key = ?String::from_utf8_lossy(&key), "rebuild: key locked, skipping"); + return false; + } + }; + + // Compute ideal volume ordering. + let kvolumes = key_to_volume(&key, &state.volumes, state.replicas, state.subvolumes); + + // Read existing record (if any) and merge `vol` into it. + let existing = state.get_record(&key).await; + let merged_volumes = if existing.deleted == Deleted::Hard { + vec![vol.to_string()] + } else { + let mut v = existing.volumes.clone(); + if !v.contains(&vol.to_string()) { + v.push(vol.to_string()); + } + v + }; + + // Re-order: prefer kvolumes order, append unknowns at the end. + let mut ordered: Vec = Vec::new(); + for kv in &kvolumes { + if merged_volumes.contains(kv) { + ordered.push(kv.clone()); + } + } + for mv in &merged_volumes { + if !kvolumes.contains(mv) { + ordered.push(mv.clone()); + } + } + + if !state + .put_record( + &key, + Record { + volumes: ordered, + deleted: Deleted::No, + hash: None, + // content_type cannot be recovered during rebuild: MIME metadata + // is stored only in LevelDB, never on the volume servers. + // Objects will need to be re-PUT (or manually patched) to + // restore Content-Type after a full DB rebuild. + content_type: None, + }, + ) + .await + { + error!(key = ?String::from_utf8_lossy(&key), "rebuild: DB put error"); + return false; + } + + true +} + +/// Scan all leaf directories under `base_url` and dispatch rebuild tasks. +async fn scan_volume(state: Arc, vol: String, base_url: String, sem: Arc) { + let listing = match get_listing(&state.http_client, &base_url).await { + Ok(l) => l, + Err(e) => { + warn!(?e, base_url, "rebuild: failed to list volume root"); + return; + } + }; + + let mut handles = Vec::new(); + + for first in listing.iter().filter(|e| is_hex_dir(e)) { + let url1 = format!("{}{}/", base_url, first.name); + let second_listing = match get_listing(&state.http_client, &url1).await { + Ok(l) => l, + Err(e) => { + warn!(?e, url1, "rebuild: failed to list first-level dir"); + continue; + } + }; + + for second in second_listing.iter().filter(|e| is_hex_dir(e)) { + let leaf_url = format!("{}{}/", url1, second.name); + let state = Arc::clone(&state); + let vol = vol.clone(); + let sem = Arc::clone(&sem); + + let handle = tokio::spawn(async move { + let _permit = sem.acquire().await.expect("semaphore closed"); + let files = match get_listing(&state.http_client, &leaf_url).await { + Ok(f) => f, + Err(e) => { + warn!(?e, leaf_url, "rebuild: leaf listing failed"); + return; + } + }; + for file in files.iter().filter(|e| e.entry_type == "file") { + rebuild_entry(&state, &vol, &file.name).await; + } + }); + handles.push(handle); + } + } + + for h in handles { + let _ = h.await; + } +} + +/// Rebuild the entire metadata database from all configured volumes. +/// +/// ## Warning +/// +/// **This is a destructive operation** because it clears the database +/// and reconstructs records solely from object presence on volume servers. +pub async fn rebuild_all(state: Arc) { + info!("starting rebuild on {:?}", state.volumes); + + // Clear all existing records. + if let Err(e) = state.db.delete_all() { + error!(?e, "rebuild: failed to clear DB"); + return; + } + + let sem = Arc::new(Semaphore::new(128)); + + for vol in &state.volumes { + let root_url = format!("http://{vol}/"); + let top_listing = match get_listing(&state.http_client, &root_url).await { + Ok(l) => l, + Err(e) => { + warn!(?e, vol, "rebuild: failed to list volume"); + continue; + } + }; + + // Check if volume uses subvolume directories. + let has_subvolumes = top_listing.iter().any(|e| is_subvolume_dir(e)); + + if has_subvolumes { + for sv in top_listing.iter().filter(|e| is_subvolume_dir(e)) { + let sv_url = format!("{}{}/", root_url, sv.name); + let sv_vol = format!("{vol}/{}", sv.name); + scan_volume(Arc::clone(&state), sv_vol, sv_url, Arc::clone(&sem)).await; + } + } else { + scan_volume(Arc::clone(&state), vol.clone(), root_url, Arc::clone(&sem)).await; + } + } + + info!("rebuild complete"); +} diff --git a/minikv-core/src/replication.rs b/minikv-core/src/replication.rs index 5da9c2d..2c9ec62 100644 --- a/minikv-core/src/replication.rs +++ b/minikv-core/src/replication.rs @@ -168,7 +168,7 @@ mod tests { let client = build_volume_client(); let result = remote_head(&client, &server.uri(), Duration::from_secs(5)).await; - assert!(result.unwrap()); + assert!(!result.unwrap()); } #[tokio::test]