Add central application state container

Cargo.toml

@@ -19,6 +19,7 @@ tracing = "0.1"
 
 base64 = "0.22"
 blake3 = "1.8"
+hex = "0.4"
 dashmap = "6.1"
 tokio = { version = "1.49", default-features = false }
 reqwest = { version = "0.13", default-features = false }

minikv-core/Cargo.toml

@@ -15,6 +15,7 @@ thiserror.workspace = true
 tracing.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"] }

minikv-core/src/lib.rs

@@ -3,6 +3,7 @@ pub mod hashing;
 pub mod locking;
 pub mod record;
 pub mod replication;
+pub mod state;
 pub mod storage;
 pub mod volumes;
 

minikv-core/src/state.rs

@@ -0,0 +1,273 @@
+//! Central application state shared across all request handlers.
+//!
+//! `AppState` holds all mutable and immutable state needed for request
+//! processing, including the LevelDB metadata store, per-key locks, multipart
+//! upload registry, volume configuration, and the shared HTTP client.
+//!
+//! All fields follow strict concurrency rules: immutable fields remain read-only,
+//! while mutable fields are protected by fine-grained locks (`KeyLock`, `DashMap`,
+//! or `tokio::sync::Mutex` for the DB). LevelDB operations that block are wrapped
+//! in `spawn_blocking` when used in async contexts to avoid holding async locks
+//! during blocking I/O.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::Duration;
+
+use dashmap::DashMap;
+use tracing::{debug, warn};
+
+use crate::locking::KeyLock;
+use crate::record::{Deleted, Record};
+use crate::storage::MetadataStore;
+
+/// Central application state, shared via `Arc<AppState>`.
+pub struct AppState {
+    /// LevelDB metadata store, behind a trait for testability.
+    pub db: Arc<dyn MetadataStore>,
+
+    /// Per-key write lock map. This is used to prevent concurrent PUT/DELETE on the same key.
+    pub key_lock: KeyLock,
+
+    /// In-flight multipart upload IDs.
+    /// `DashMap` is used instead of `Mutex<HashMap>` to avoid a global lock.
+    pub upload_ids: DashMap<String, ()>,
+
+    // ---- configuration (read-only after startup) ----
+    /// Ordered list of all volume server addresses.
+    pub volumes: Vec<String>,
+
+    /// Optional fallback server for keys not found on any volume.
+    pub fallback: Option<String>,
+
+    /// Number of replicas to write per PUT.
+    pub replicas: usize,
+
+    /// Number of subvolume shards per volume server (1 = no subvolumes).
+    pub subvolumes: usize,
+
+    /// If `true`, DELETE is only allowed after an UNLINK (soft-delete first).
+    pub protect: bool,
+
+    /// If `true`, compute and store a BLAKE3 checksum for each object body.
+    pub checksum: bool,
+
+    /// Timeout for HEAD requests to volume servers during GET redirect.
+    pub vol_timeout: Duration,
+
+    /// Shared HTTP client for all volume server communication.
+    pub http_client: reqwest::Client,
+
+    /// Maps internal volume address (e.g. `"volume1:8080"`) to its
+    /// public-facing address (e.g. `"localhost:8001"`).
+    ///
+    /// Built once at startup from `--volumes` + `--public-volumes`.
+    /// Empty when `--public-volumes` is not set — Location headers will
+    /// then use internal addresses unchanged (correct for bare-metal).
+    pub vol_rewrite: HashMap<String, String>,
+
+    /// When `true`, GET/HEAD returns `X-Accel-Redirect` instead of `302`.
+    ///
+    /// Requires a frontend nginx configured with `proxy_pass` to the
+    /// coordinator and an `internal` proxy location for `/accel/`.
+    /// The frontend nginx then fetches the object from the volume server
+    /// directly, serving the body with the coordinator\'s response headers
+    /// (including `Content-Type` from stored metadata).
+    ///
+    /// When `false` (default), GET/HEAD returns a standard `302 Found`
+    /// redirect to the volume server URL.
+    pub accel_redirect: bool,
+}
+
+impl AppState {
+    /// Read a record from LevelDB.
+    ///
+    /// Returns `Record::not_found()` (deleted == Hard) when the key is absent.
+    pub async fn get_record(&self, key: &[u8]) -> Record {
+        match self.db.get(key) {
+            Ok(Some(bytes)) => match Record::decode(&bytes) {
+                Ok(rec) => rec,
+                Err(e) => {
+                    warn!(?e, key = ?String::from_utf8_lossy(key), "corrupt record in DB");
+                    Record::not_found()
+                }
+            },
+            Ok(None) => Record::not_found(),
+            Err(e) => {
+                warn!(?e, "LevelDB get error");
+                Record::not_found()
+            }
+        }
+    }
+
+    /// Write a record to LevelDB.
+    ///
+    /// Returns `false` on any error (matching Go's bool-return pattern),
+    /// and logs the error. Callers translate `false` => HTTP 500.
+    pub async fn put_record(&self, key: &[u8], rec: Record) -> bool {
+        match rec.encode() {
+            Err(e) => {
+                warn!(?e, "attempted to encode invalid record");
+                false
+            }
+            Ok(bytes) => match self.db.put(key, &bytes) {
+                Ok(()) => {
+                    debug!(key = ?String::from_utf8_lossy(key), "record written");
+                    true
+                }
+                Err(e) => {
+                    warn!(?e, "LevelDB put error");
+                    false
+                }
+            },
+        }
+    }
+
+    /// Hard-delete a key from LevelDB entirely.
+    pub async fn delete_record(&self, key: &[u8]) -> bool {
+        match self.db.delete(key) {
+            Ok(()) => true,
+            Err(e) => {
+                warn!(?e, "LevelDB delete error");
+                false
+            }
+        }
+    }
+
+    /// Write an object to all configured replicas and update LevelDB.
+    ///
+    /// Steps:
+    /// 1. Compute target volumes via `key_to_volume`.
+    /// 2. Mark key as SOFT-deleted in DB (partially written sentinel).
+    /// 3. Write body bytes to each replica.
+    /// 4. Optionally compute BLAKE3 hash of the body.
+    /// 5. Mark key as fully present (NO deletion) with optional hash.
+    ///
+    /// Returns the HTTP status code to send to the client (201 or 500).
+    pub async fn write_to_replicas(
+        &self,
+        key: &[u8],
+        body: bytes::Bytes,
+        content_type: Option<String>,
+    ) -> u16 {
+        use crate::hashing::key_to_path;
+        use crate::replication::remote_put;
+        use crate::volumes::key_to_volume;
+
+        let kvolumes = key_to_volume(key, &self.volumes, self.replicas, self.subvolumes);
+
+        // Step 1: mark as in-progress (SOFT) so a crash doesn't leave
+        // a record pointing at volumes that were never written.
+        if !self
+            .put_record(
+                key,
+                Record {
+                    volumes: kvolumes.clone(),
+                    deleted: Deleted::Soft,
+                    hash: None,
+                    content_type: None,
+                },
+            )
+            .await
+        {
+            return 500;
+        }
+
+        // Step 2: write to each replica.
+        let kp = key_to_path(key);
+        for volume in &kvolumes {
+            let url = format!("http://{volume}{kp}");
+            if let Err(e) = remote_put(&self.http_client, &url, body.clone()).await {
+                warn!(?e, url, "replica write failed");
+                return 500;
+            }
+        }
+
+        // Step 3: optionally compute content hash.
+        let hash = if self.checksum {
+            let digest = blake3::hash(&body);
+            Some(hex::encode(digest.as_bytes()))
+        } else {
+            None
+        };
+
+        // Step 4: mark as fully present.
+        if !self
+            .put_record(
+                key,
+                Record {
+                    volumes: kvolumes,
+                    deleted: Deleted::No,
+                    hash,
+                    content_type,
+                },
+            )
+            .await
+        {
+            return 500;
+        }
+
+        201
+    }
+
+    /// Delete an object (soft or hard depending on `unlink`).
+    ///
+    /// Returns the HTTP status code: 204, 403, 404, or 500.
+    pub async fn delete(&self, key: &[u8], unlink: bool) -> u16 {
+        use crate::hashing::key_to_path;
+        use crate::replication::remote_delete;
+
+        let rec = self.get_record(key).await;
+
+        if rec.deleted == Deleted::Hard {
+            return 404;
+        }
+        if unlink && rec.deleted == Deleted::Soft {
+            return 404;
+        }
+        if !unlink && self.protect && rec.deleted == Deleted::No {
+            return 403;
+        }
+
+        // Mark as soft-deleted first.
+        if !self
+            .put_record(
+                key,
+                Record {
+                    volumes: rec.volumes.clone(),
+                    deleted: Deleted::Soft,
+                    hash: rec.hash.clone(),
+                    content_type: rec.content_type.clone(),
+                },
+            )
+            .await
+        {
+            return 500;
+        }
+
+        if unlink {
+            return 204;
+        }
+
+        // Hard delete: remove from volume servers, then from DB.
+        let kp = key_to_path(key);
+        let mut delete_error = false;
+        for volume in &rec.volumes {
+            let url = format!("http://{volume}{kp}");
+            if let Err(e) = remote_delete(&self.http_client, &url).await {
+                warn!(?e, url, "remote delete failed: possible orphan file");
+                delete_error = true;
+            }
+        }
+
+        if delete_error {
+            return 500;
+        }
+
+        if !self.delete_record(key).await {
+            return 500;
+        }
+
+        204
+    }
+}