Support rebuild of leveldb metadata (#8)

* Fix incorrect assertion in replication test

* Support rebuild of leveldb metadata

Author: ekkolon

Cargo.lock

@@ -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"

Cargo.toml

@@ -13,13 +13,17 @@ 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 }
 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"] }

minikv-core/Cargo.toml

@@ -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<rusty_leveldb::Status> for Error {

minikv-core/src/error.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;

minikv-core/src/lib.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/<b64>    ← 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<Vec<AutoindexEntry>, Error> {
+    let body = remote_get(client, url).await?;
+    let entries: Vec<AutoindexEntry> =
+        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<String> = 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<AppState>, vol: String, base_url: String, sem: Arc<Semaphore>) {
+    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<AppState>) {
+    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");
+}

minikv-core/src/rebuild.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]