From d8eb4107477d0dabc04318c693f109a5b8960aa7 Mon Sep 17 00:00:00 2001 From: Nelson Dominguez Date: Fri, 27 Feb 2026 21:59:45 +0100 Subject: [PATCH] Add central application state container --- Cargo.lock | 7 + Cargo.toml | 1 + minikv-core/Cargo.toml | 1 + minikv-core/src/lib.rs | 1 + minikv-core/src/state.rs | 273 +++++++++++++++++++++++++++++++++++++++ 5 files changed, 283 insertions(+) create mode 100644 minikv-core/src/state.rs diff --git a/Cargo.lock b/Cargo.lock index e918b00..28e1411 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -486,6 +486,12 @@ version = "0.5.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fc0fef456e4baa96da950455cd02c081ca953b141298e41db3fc7e36b1da849c" +[[package]] +name = "hex" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7f24254aa9a54b5c858eaee2f5bccdb46aaf0e486a595ed5fd8f86ba55232a70" + [[package]] name = "http" version = "1.4.0" @@ -832,6 +838,7 @@ dependencies = [ "blake3", "bytes", "dashmap", + "hex", "reqwest", "rusty-leveldb", "thiserror 2.0.18", diff --git a/Cargo.toml b/Cargo.toml index 1a3baeb..d931739 100644 --- a/Cargo.toml +++ b/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 } diff --git a/minikv-core/Cargo.toml b/minikv-core/Cargo.toml index 3cebbe6..0679d30 100644 --- a/minikv-core/Cargo.toml +++ b/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"] } diff --git a/minikv-core/src/lib.rs b/minikv-core/src/lib.rs index f6cd317..a6bcec1 100644 --- a/minikv-core/src/lib.rs +++ b/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; diff --git a/minikv-core/src/state.rs b/minikv-core/src/state.rs new file mode 100644 index 0000000..bc3a322 --- /dev/null +++ b/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`. +pub struct AppState { + /// LevelDB metadata store, behind a trait for testability. + pub db: Arc, + + /// 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` to avoid a global lock. + pub upload_ids: DashMap, + + // ---- configuration (read-only after startup) ---- + /// Ordered list of all volume server addresses. + pub volumes: Vec, + + /// Optional fallback server for keys not found on any volume. + pub fallback: Option, + + /// 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, + + /// 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, + ) -> 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 + } +}