Add HTTP layer

Author: ekkolon

Cargo.lock

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

Cargo.toml

@@ -11,3 +11,18 @@ keywords.workspace = true
 categories.workspace = true
 
 [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"] }

minikv/Cargo.toml

@@ -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());
+    }
+}

minikv/src/cli.rs

@@ -0,0 +1,15 @@
+/// Unified error type for minikv.
+#[derive(Debug, thiserror::Error)]
+pub enum Error {
+    /// Standard I/O errors (temp file handling for multipart uploads).
+    #[error("I/O error: {0}")]
+    Io(#[from] std::io::Error),
+
+    /// XML deserialization errors for S3-compatible API bodies.
+    #[error("XML parse error: {0}")]
+    XmlParse(String),
+
+    /// Minikv core library error
+    #[error(transparent)]
+    Core(#[from] minikv_core::Error),
+}