ekkolon · ekkolon · Mar 3, 2026 · Feb 28, 2026 · Mar 3, 2026 · Mar 3, 2026
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -14,6 +14,8 @@ keywords = ["storage", "distributed", "leveldb", "object-store"]
 categories = ["database", "network-programming"]
 
 [workspace.dependencies]
+minikv-core = { path = "minikv-core" }
+
 thiserror = "2"
 tracing = "0.1"
 
@@ -26,8 +28,9 @@ dashmap = "6.1"
 tokio = { version = "1.49", default-features = false }
 reqwest = { version = "0.13", default-features = false }
 bytes = { version = "1.11" }
-
-# Chosen over `leveldb` crate which requires C++ LevelDB via FFI.
+uuid = { version = "1", features = ["v4"] }
+rand = "0.8"
+futures = "0.3"
 rusty-leveldb = "1"
 
 # Testing

diff --git a/minikv-core/src/storage/mod.rs b/minikv-core/src/storage/mod.rs
@@ -42,56 +42,3 @@ pub trait MetadataStore: Send + Sync {
     /// Primarily used by `rebuild_all` to regenerate the database from scratch.
     fn delete_all(&self) -> Result<(), Error>;
 }
-
-/// In-memory `MetadataStore` backed by a `BTreeMap`.
-///
-/// Useful for unit and integration tests that do not require persistent storage.
-#[cfg(debug_assertions)]
-pub mod mem {
-    use super::*;
-    use std::collections::BTreeMap;
-    use std::sync::Mutex;
-
-    #[derive(Default)]
-    pub struct MemStore {
-        inner: Mutex<BTreeMap<Vec<u8>, Vec<u8>>>,
-    }
-
-    impl MetadataStore for MemStore {
-        fn get(&self, key: &[u8]) -> Result<Option<Vec<u8>>, Error> {
-            Ok(self.inner.lock().unwrap().get(key).cloned())
-        }
-
-        fn put(&self, key: &[u8], value: &[u8]) -> Result<(), Error> {
-            self.inner
-                .lock()
-                .unwrap()
-                .insert(key.to_vec(), value.to_vec());
-            Ok(())
-        }
-
-        fn delete(&self, key: &[u8]) -> Result<(), Error> {
-            self.inner.lock().unwrap().remove(key);
-            Ok(())
-        }
-
-        fn scan_prefix(&self, prefix: &[u8]) -> Result<Vec<KeyValuePair>, Error> {
-            let store = self.inner.lock().unwrap();
-            Ok(store
-                .iter()
-                .filter(|(k, _)| k.starts_with(prefix))
-                .map(|(k, v)| (k.clone(), v.clone()))
-                .collect())
-        }
-
-        fn scan_all(&self) -> Result<Vec<KeyValuePair>, Error> {
-            let store = self.inner.lock().unwrap();
-            Ok(store.iter().map(|(k, v)| (k.clone(), v.clone())).collect())
-        }
-
-        fn delete_all(&self) -> Result<(), Error> {
-            self.inner.lock().unwrap().clear();
-            Ok(())
-        }
-    }
-}
diff --git a/minikv-core/tests/rebalance.rs b/minikv-core/tests/rebalance.rs
@@ -8,18 +8,66 @@
 //! An in-memory `MemStore` is used as the metadata backend.
 
 use dashmap::DashMap;
+use minikv_core::Error;
+use minikv_core::KeyValuePair;
+use minikv_core::MetadataStore;
 use minikv_core::locking::KeyLock;
 use minikv_core::rebalance::rebalance_key;
 use minikv_core::replication::build_volume_client;
 use minikv_core::state::AppState;
-use minikv_core::storage::mem::MemStore;
 use minikv_core::volumes::needs_rebalance;
+use std::collections::BTreeMap;
 use std::collections::HashMap;
 use std::sync::Arc;
+use std::sync::Mutex;
 use std::time::Duration;
 use wiremock::matchers::method;
 use wiremock::{Mock, MockServer, ResponseTemplate};
 
+// In-memory `MetadataStore` backed by a `BTreeMap`.
+#[derive(Default)]
+pub struct MemStore {
+    inner: Mutex<BTreeMap<Vec<u8>, Vec<u8>>>,
+}
+
+impl MetadataStore for MemStore {
+    fn get(&self, key: &[u8]) -> Result<Option<Vec<u8>>, Error> {
+        Ok(self.inner.lock().unwrap().get(key).cloned())
+    }
+
+    fn put(&self, key: &[u8], value: &[u8]) -> Result<(), Error> {
+        self.inner
+            .lock()
+            .unwrap()
+            .insert(key.to_vec(), value.to_vec());
+        Ok(())
+    }
+
+    fn delete(&self, key: &[u8]) -> Result<(), Error> {
+        self.inner.lock().unwrap().remove(key);
+        Ok(())
+    }
+
+    fn scan_prefix(&self, prefix: &[u8]) -> Result<Vec<KeyValuePair>, Error> {
+        let store = self.inner.lock().unwrap();
+        Ok(store
+            .iter()
+            .filter(|(k, _)| k.starts_with(prefix))
+            .map(|(k, v)| (k.clone(), v.clone()))
+            .collect())
+    }
+
+    fn scan_all(&self) -> Result<Vec<KeyValuePair>, Error> {
+        let store = self.inner.lock().unwrap();
+        Ok(store.iter().map(|(k, v)| (k.clone(), v.clone())).collect())
+    }
+
+    fn delete_all(&self) -> Result<(), Error> {
+        self.inner.lock().unwrap().clear();
+        Ok(())
+    }
+}
+
 /// Build a test `AppState` backed by `MemStore` and pointing at `volumes`.
 fn make_state(volumes: Vec<String>) -> Arc<AppState> {
     let replicas = volumes.len().min(2);

diff --git a/minikv/Cargo.toml b/minikv/Cargo.toml
@@ -10,4 +10,28 @@ authors.workspace = true
 keywords.workspace = true
 categories.workspace = true
 
+[lib]
+name = "minikv_server"
+path = "src/lib.rs"
+
 [dependencies]
+minikv-core = { workspace = true }
+tracing = { workspace = true }
+dashmap = { workspace = true }
+thiserror = { workspace = true }
+bytes = { workspace = true }
+serde = { workspace = true, features = ["derive"] }
+serde_json = { workspace = true }
+futures = { workspace = true }
+rand = { workspace = true }
+uuid = { workspace = true }
+tokio = { workspace = true, features = ["full"] }
+clap = { version = "4", features = ["derive", "env"] }
+tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] }
+axum = { version = "0.7", features = [] }
+quick-xml = { version = "0.31", features = ["serialize"] }
+
+[dev-dependencies]
+tower-util = { version = "0.3" }
+minikv-core = { workspace = true }
+wiremock = "0.6"
diff --git a/minikv/src/cli.rs b/minikv/src/cli.rs
@@ -0,0 +1,210 @@
+/// CLI definition for `minikv`.
+///
+/// # Subcommands
+/// - `server`              run the HTTP metadata coordinator
+/// - `rebuild`             reconstruct LevelDB from volume server autoindex
+/// - `rebalance`           move all keys to their ideal volume set
+/// - `print-nginx-config`  emit the nginx volume server config to stdout
+use std::path::PathBuf;
+use std::time::Duration;
+
+use clap::{Args, Parser, Subcommand};
+
+#[derive(Parser, Debug)]
+#[command(
+    name = "minikv",
+    version,
+    about = "Distributed object storage coordinator",
+    long_about = None,
+)]
+pub struct Cli {
+    #[command(subcommand)]
+    pub command: Command,
+}
+
+#[derive(Subcommand, Debug)]
+pub enum Command {
+    /// Run the HTTP metadata coordinator server.
+    Server(ServerArgs),
+
+    /// Rebuild LevelDB metadata from volume server autoindex listings.
+    ///
+    /// WARNING: This clears the entire DB before scanning.
+    Rebuild(CommonArgs),
+
+    /// Rebalance all keys to their ideal volume set.
+    Rebalance(CommonArgs),
+
+    /// Print the nginx volume server configuration to stdout.
+    PrintNginxConfig,
+}
+
+/// Arguments shared by server, rebuild, and rebalance.
+#[derive(Args, Debug, Clone)]
+pub struct CommonArgs {
+    /// Path to the LevelDB database directory.
+    #[arg(long, env = "MINIKV_DB", required = true)]
+    pub db: PathBuf,
+
+    /// Comma-separated list of volume server addresses (host:port).
+    #[arg(long, env = "MINIKV_VOLUMES", required = true, value_delimiter = ',')]
+    pub volumes: Vec<String>,
+
+    /// Number of replicas to maintain per object.
+    #[arg(long, env = "MINIKV_REPLICAS", default_value = "3")]
+    pub replicas: usize,
+
+    /// Number of sub-volume shards per volume server.
+    /// Use 1 to disable sub-volume path components.
+    #[arg(long, env = "MINIKV_SUBVOLUMES", default_value = "10")]
+    pub subvolumes: usize,
+
+    /// Timeout for HEAD probes to volume servers (e.g. "1s", "500ms").
+    #[arg(long, env = "MINIKV_VOLTIMEOUT", default_value = "1s", value_parser = parse_duration)]
+    pub voltimeout: Duration,
+}
+
+/// Additional arguments for the `server` subcommand.
+#[derive(Args, Debug)]
+pub struct ServerArgs {
+    #[command(flatten)]
+    pub common: CommonArgs,
+
+    /// Port to listen on.
+    #[arg(long, env = "MINIKV_PORT", default_value = "3000")]
+    pub port: u16,
+
+    /// Optional fallback server for keys missing from all volumes.
+    #[arg(long, env = "MINIKV_FALLBACK")]
+    pub fallback: Option<String>,
+
+    /// Public-facing addresses for each volume server, used in Location
+    /// redirect headers returned to clients.
+    ///
+    /// Must be the same length as --volumes and in the same order.
+    /// Each entry is the externally-reachable host:port for the corresponding
+    /// --volumes entry.
+    ///
+    /// Example:
+    ///   --volumes=volume1:8080,volume2:8080,volume3:8080
+    ///   --public-volumes=localhost:8001,localhost:8002,localhost:8003
+    ///
+    /// When omitted, --volumes addresses are used as-is in Location headers
+    /// (correct for bare-metal deployments where internal == external).
+    #[arg(
+        long,
+        env = "MINIKV_PUBLIC_VOLUMES",
+        value_delimiter = ',',
+        requires = "volumes"
+    )]
+    pub public_volumes: Option<Vec<String>>,
+
+    /// Require UNLINK (soft-delete) before a hard DELETE is allowed.
+    #[arg(long, env = "MINIKV_PROTECT", default_value = "false")]
+    pub protect: bool,
+
+    /// Compute and store a BLAKE3 checksum for every uploaded object.
+    #[arg(long, env = "MINIKV_CHECKSUM", default_value = "true")]
+    pub checksum: bool,
+
+    /// Enable X-Accel-Redirect mode for GET/HEAD responses.
+    ///
+    /// When enabled, the coordinator returns `X-Accel-Redirect: /accel/<host>/<path>`
+    /// instead of `302 Location`. A frontend nginx must be configured with:
+    ///
+    ///   proxy_pass http://coordinator:3000;
+    ///   location ~ ^/accel/([^/]+)/(.+)$ {
+    ///       internal;
+    ///       proxy_pass http://$1/$2;
+    ///   }
+    ///
+    /// This allows nginx to stream the object body while the coordinator controls
+    /// all response headers, including Content-Type from stored object metadata.
+    ///
+    /// When disabled (default), GET/HEAD returns a standard 302 redirect.
+    #[arg(long, env = "MINIKV_ACCEL_REDIRECT", default_value = "false")]
+    pub accel_redirect: bool,
+
+    /// Enable verbose structured logging.
+    #[arg(short, long, env = "MINIKV_VERBOSE", default_value = "false")]
+    pub verbose: bool,
+}
+
+/// Parse a human-friendly duration string such as "1s" or "500ms".
+fn parse_duration(s: &str) -> Result<Duration, String> {
+    if let Some(ms) = s.strip_suffix("ms") {
+        ms.parse::<u64>()
+            .map(Duration::from_millis)
+            .map_err(|e| e.to_string())
+    } else if let Some(secs) = s.strip_suffix('s') {
+        secs.parse::<u64>()
+            .map(Duration::from_secs)
+            .map_err(|e| e.to_string())
+    } else {
+        // Fall back: treat as integer milliseconds.
+        s.parse::<u64>()
+            .map(Duration::from_millis)
+            .map_err(|_| format!("invalid duration '{s}'. Use '1s' or '500ms'"))
+    }
+}
+
+/// Validate common arguments and panic with a clear message on bad input.
+pub fn validate_common(args: &CommonArgs) {
+    if args.volumes.is_empty() {
+        eprintln!("error: --volumes must contain at least one volume server");
+        std::process::exit(1);
+    }
+    if args.replicas == 0 {
+        eprintln!("error: --replicas must be ≥ 1");
+        std::process::exit(1);
+    }
+    if args.volumes.len() < args.replicas {
+        eprintln!(
+            "error: need at least as many volumes ({}) as replicas ({})",
+            args.volumes.len(),
+            args.replicas
+        );
+        std::process::exit(1);
+    }
+    if args.subvolumes == 0 {
+        eprintln!("error: --subvolumes must be ≥ 1");
+        std::process::exit(1);
+    }
+}
+
+/// Validate server-only arguments.
+#[allow(unused)]
+pub fn validate_server(args: &ServerArgs) {
+    validate_common(&args.common);
+    if let Some(ref pv) = args.public_volumes
+        && pv.len() != args.common.volumes.len()
+    {
+        eprintln!(
+            "error: --public-volumes has {} entries but --volumes has {}; they must match 1:1",
+            pv.len(),
+            args.common.volumes.len()
+        );
+        std::process::exit(1);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parse_duration_seconds() {
+        assert_eq!(parse_duration("1s").unwrap(), Duration::from_secs(1));
+        assert_eq!(parse_duration("30s").unwrap(), Duration::from_secs(30));
+    }
+
+    #[test]
+    fn parse_duration_millis() {
+        assert_eq!(parse_duration("500ms").unwrap(), Duration::from_millis(500));
+    }
+
+    #[test]
+    fn parse_duration_invalid() {
+        assert!(parse_duration("1h").is_err());
+    }
+}