From 539680dbba0e2670cb17d6987b628ebb3cdd2b88 Mon Sep 17 00:00:00 2001 From: Nelson Dominguez Date: Fri, 27 Feb 2026 13:54:52 +0100 Subject: [PATCH] Add blake3 object key hashing functions --- Cargo.lock | 141 ++++++++++++++++++++++++++ Cargo.toml | 3 + minikv-core/Cargo.toml | 3 + minikv-core/src/hashing.rs | 81 +++++++++++++++ minikv-core/src/lib.rs | 3 +- minikv-core/src/volumes.rs | 186 +++++++++++++++++++++++++++++++++++ minikv-core/tests/hashing.rs | 146 +++++++++++++++++++++++++++ 7 files changed, 562 insertions(+), 1 deletion(-) create mode 100644 minikv-core/src/hashing.rs create mode 100644 minikv-core/src/volumes.rs create mode 100644 minikv-core/tests/hashing.rs diff --git a/Cargo.lock b/Cargo.lock index 1bdf145..be6e9bc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,6 +2,81 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "arrayref" +version = "0.3.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "76a2e8124351fda1ef8aaaa3bbd7ebbcb486bbcd4225aca0aa0d84bb2db8fecb" + +[[package]] +name = "arrayvec" +version = "0.7.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7c02d123df017efcdfbd739ef81735b36c5ba83ec3c59c80a9d7ecc718f92e50" + +[[package]] +name = "base64" +version = "0.22.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" + +[[package]] +name = "blake3" +version = "1.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2468ef7d57b3fb7e16b576e8377cdbde2320c60e1491e961d11da40fc4f02a2d" +dependencies = [ + "arrayref", + "arrayvec", + "cc", + "cfg-if", + "constant_time_eq", + "cpufeatures", +] + +[[package]] +name = "cc" +version = "1.2.56" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aebf35691d1bfb0ac386a69bac2fde4dd276fb618cf8bf4f5318fe285e821bb2" +dependencies = [ + "find-msvc-tools", + "shlex", +] + +[[package]] +name = "cfg-if" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" + +[[package]] +name = "constant_time_eq" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3d52eff69cd5e647efe296129160853a42795992097e8af39800e1060caeea9b" + +[[package]] +name = "cpufeatures" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "59ed5838eebb26a2bb2e58f6d5b5316989ae9d08bab10e0e6d103e656d1b0280" +dependencies = [ + "libc", +] + +[[package]] +name = "find-msvc-tools" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" + +[[package]] +name = "libc" +version = "0.2.182" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6800badb6cb2082ffd7b6a67e6125bb39f18782f793520caee8cb8846be06112" + [[package]] name = "minikv" version = "0.1.0" @@ -9,3 +84,69 @@ version = "0.1.0" [[package]] name = "minikv-core" version = "0.1.0" +dependencies = [ + "base64", + "blake3", + "thiserror", +] + +[[package]] +name = "proc-macro2" +version = "1.0.106" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8fd00f0bb2e90d81d1044c2b32617f68fcb9fa3bb7640c23e9c748e53fb30934" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quote" +version = "1.0.44" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "21b2ebcf727b7760c461f091f9f0f539b77b8e87f2fd88131e7f1b433b3cece4" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "shlex" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64" + +[[package]] +name = "syn" +version = "2.0.117" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e665b8803e7b1d2a727f4023456bbbbe74da67099c585258af0ad9c5013b9b99" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "thiserror" +version = "2.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4288b5bcbc7920c07a1149a35cf9590a2aa808e0bc1eafaade0b80947865fbc4" +dependencies = [ + "thiserror-impl", +] + +[[package]] +name = "thiserror-impl" +version = "2.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ebc4ee7f67670e9b64d05fa4253e753e016c6c95ff35b89b7941d6b856dec1d5" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "unicode-ident" +version = "1.0.24" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" diff --git a/Cargo.toml b/Cargo.toml index c18341e..acc9709 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -14,3 +14,6 @@ keywords = ["storage", "distributed", "leveldb", "object-store"] categories = ["database", "network-programming"] [workspace.dependencies] +thiserror = "2" +base64 = "0.22" +blake3 = "1.8" diff --git a/minikv-core/Cargo.toml b/minikv-core/Cargo.toml index 20143b1..df7c839 100644 --- a/minikv-core/Cargo.toml +++ b/minikv-core/Cargo.toml @@ -11,3 +11,6 @@ keywords.workspace = true categories.workspace = true [dependencies] +thiserror.workspace = true +base64 = { workspace = true } +blake3 = { workspace = true } diff --git a/minikv-core/src/hashing.rs b/minikv-core/src/hashing.rs new file mode 100644 index 0000000..4f50b87 --- /dev/null +++ b/minikv-core/src/hashing.rs @@ -0,0 +1,81 @@ +use base64::{Engine as _, engine::general_purpose::STANDARD as B64_STANDARD}; + +/// Convert a raw key into a stable, deterministic storage path for volume servers. +/// +/// Algorithm: +/// 1. Compute BLAKE3-256 hash of `key`. +/// 2. Use the first two bytes as two hex directory components (fanout = 256²). +/// 3. Base64-encode the raw key as the filename. +/// +/// Format: `/%02x/%02x/` +/// +/// # Stability +/// Changing this function after data is stored is a breaking change.\ +/// Stored keys map to different paths if the algorithm is modified, requiring a full rebalance. +#[must_use] +pub fn key_to_path(key: &[u8]) -> String { + let hash = blake3::hash(key); + let b = hash.as_bytes(); + let b64 = B64_STANDARD.encode(key); + format!("/{:02x}/{:02x}/{}", b[0], b[1], b64) +} + +/// Compute a per-volume BLAKE3 score for a key. +/// +/// Combines the key bytes and the volume name to produce a 32-byte hash suitable +/// for consistent shard placement. +#[must_use] +pub fn volume_score(key: &[u8], volume: &str) -> [u8; 32] { + let mut hasher = blake3::Hasher::new(); + hasher.update(key); + hasher.update(volume.as_bytes()); + *hasher.finalize().as_bytes() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn key_to_path_stability_hello() { + let path = key_to_path(b"hello"); + assert_eq!(path, "/ea/8f/aGVsbG8="); + } + + #[test] + fn key_to_path_stability_helloworld() { + let path = key_to_path(b"helloworld"); + let parts: Vec<&str> = path.splitn(4, '/').collect(); + assert_eq!(parts.len(), 4); + assert_eq!(parts[1].len(), 2); + assert_eq!(parts[2].len(), 2); + let decoded = B64_STANDARD.decode(parts[3]).unwrap(); + assert_eq!(decoded, b"helloworld"); + } + + #[test] + fn key_to_path_snapshot_helloworld() { + let path = key_to_path(b"helloworld"); + let hash = blake3::hash(b"helloworld"); + let b = hash.as_bytes(); + let expected = format!("/{:02x}/{:02x}/aGVsbG93b3JsZA==", b[0], b[1]); + assert_eq!(path, expected); + } + + #[test] + fn key_to_path_empty_key() { + let path = key_to_path(b""); + let parts: Vec<&str> = path.splitn(4, '/').collect(); + assert_eq!(parts.len(), 4); + } + + #[test] + fn key_to_path_binary_key() { + let key: Vec = (0u8..=255).collect(); + let path = key_to_path(&key); + let parts: Vec<&str> = path.splitn(4, '/').collect(); + assert_eq!(parts.len(), 4); + let decoded = B64_STANDARD.decode(parts[3]).unwrap(); + assert_eq!(decoded, key); + } +} diff --git a/minikv-core/src/lib.rs b/minikv-core/src/lib.rs index 8b13789..4f50687 100644 --- a/minikv-core/src/lib.rs +++ b/minikv-core/src/lib.rs @@ -1 +1,2 @@ - +pub mod hashing; +pub mod volumes; diff --git a/minikv-core/src/volumes.rs b/minikv-core/src/volumes.rs new file mode 100644 index 0000000..a8de95c --- /dev/null +++ b/minikv-core/src/volumes.rs @@ -0,0 +1,186 @@ +//! Volume selection. Deterministic mapping from key to replica volumes. +//! +//! # Stability +//! The output of `key_to_volume` must be **identical** for the same `(key, volumes, +//! replicas, subvolumes)` tuple across all crate versions, platforms, and time. +//! Any change will break stored data layout and requires a full data migration. +//! +//! Algorithm: +//! 1. For each volume: `score = BLAKE3(key || volume_bytes)`. +//! 2. Stable descending sort volumes by score (byte-wise, big-endian). +//! 3. Take the top `replicas` entries. +//! 4. For each entry, compute a subvolume index when `subvolumes > 1`: +//! ```text +//! sv_idx = u32_be(score[28..32]) % subvolumes +//! ``` +//! 5. Append `/sv{idx:02X}` if subvolumes > 1, otherwise return just the volume string. + +use crate::hashing::volume_score; + +/// Internal structure to store a volume with its score. +#[derive(Debug)] +struct ScoredVolume { + score: [u8; 32], + volume: String, +} + +/// Deterministically select `replicas` volume addresses for a given key. +/// +/// - `volumes`: full list of available volume servers. +/// - `replicas`: number of replicas to select (≤ `volumes.len()`). +/// - `subvolumes`: number of sub-volume shards per machine (1 disables subvolumes). +/// +/// Returns a `Vec` of length `replicas`. +/// Each entry is either `"hostname:port"` (`subvolumes == 1`) or `"hostname:port/svXX"` (`subvolumes > 1`). +#[must_use] +pub fn key_to_volume( + key: &[u8], + volumes: &[String], + replicas: usize, + subvolumes: usize, +) -> Vec { + debug_assert!(replicas <= volumes.len(), "replicas must be ≤ volume count"); + debug_assert!(subvolumes >= 1, "subvolumes must be ≥ 1"); + + let mut scored: Vec = volumes + .iter() + .map(|v| ScoredVolume { + score: volume_score(key, v), + volume: v.clone(), + }) + .collect(); + + // Stable descending sort by score (big-endian byte-wise) + scored.sort_by(|a, b| b.score.cmp(&a.score)); + + scored + .into_iter() + .take(replicas) + .map(|sv| { + if subvolumes == 1 { + sv.volume + } else { + let last4 = &sv.score[28..32]; + let svhash = u64::from(last4[0]) << 24 + | u64::from(last4[1]) << 16 + | u64::from(last4[2]) << 8 + | u64::from(last4[3]); + let sv_idx = svhash % subvolumes as u64; + format!("{}/sv{:02X}", sv.volume, sv_idx) + } + }) + .collect() +} + +/// Determine whether a key needs rebalancing. +/// +/// Returns `true` if the currently recorded volumes differ from the ideal +/// volumes computed by `key_to_volume`. +#[must_use] +pub fn needs_rebalance(current: &[String], ideal: &[String]) -> bool { + current != ideal +} + +#[cfg(test)] +mod tests { + use super::*; + + fn vols() -> Vec { + vec!["larry".into(), "moe".into(), "curly".into()] + } + + /// Stability tests for `key_to_volume`. + #[test] + fn key_to_volume_returns_correct_count() { + let volumes = vols(); + for replicas in 1..=3 { + let result = key_to_volume(b"hello", &volumes, replicas, 1); + assert_eq!(result.len(), replicas); + } + } + + #[test] + fn key_to_volume_no_duplicates() { + let volumes = vols(); + let result = key_to_volume(b"hello", &volumes, 3, 1); + let mut seen = std::collections::HashSet::new(); + for v in &result { + assert!(seen.insert(v.clone()), "duplicate volume: {v}"); + } + } + + #[test] + fn key_to_volume_only_known_volumes() { + let volumes = vols(); + let result = key_to_volume(b"hello", &volumes, 3, 1); + for v in &result { + let base = v.split('/').next().unwrap(); + assert!(volumes.contains(&base.to_string()), "unknown volume: {v}"); + } + } + + #[test] + fn key_to_volume_deterministic() { + let volumes = vols(); + let r1 = key_to_volume(b"hello", &volumes, 2, 1); + let r2 = key_to_volume(b"hello", &volumes, 2, 1); + assert_eq!(r1, r2); + } + + #[test] + fn key_to_volume_stable_snapshot_hello_replicas1_sv1() { + let volumes = vols(); + let result = key_to_volume(b"hello", &volumes, 1, 1); + let top = &result[0]; + + let mut scored: Vec<_> = volumes + .iter() + .map(|v| (volume_score(b"hello", v), v.clone())) + .collect(); + scored.sort_by(|a, b| b.0.cmp(&a.0)); + + assert_eq!(top, &scored[0].1); + } + + #[test] + fn key_to_volume_subvolumes_format() { + let volumes = vols(); + let result = key_to_volume(b"hello", &volumes, 1, 10); + let v = &result[0]; + assert!(v.contains("/sv"), "missing /sv: {v}"); + let suffix = v.split("/sv").last().unwrap(); + assert_eq!(suffix.len(), 2); + assert!(suffix.chars().all(|c| c.is_ascii_hexdigit())); + } + + #[test] + fn key_to_volume_subvolumes_range() { + let volumes = vols(); + for key in [b"hello".as_ref(), b"world", b"foo", b"bar", b"baz"] { + let result = key_to_volume(key, &volumes, 1, 4); + let suffix = result[0].split("/sv").last().unwrap(); + let idx = u64::from_str_radix(suffix, 16).unwrap(); + assert!(idx < 4, "sv index {idx} out of range for svcount=4"); + } + } + + #[test] + fn needs_rebalance_equal_is_false() { + let v = vec!["a".into(), "b".into()]; + assert!(!needs_rebalance(&v, &v)); + } + + #[test] + fn needs_rebalance_different_is_true() { + let a = vec!["a".into(), "b".into()]; + let b = vec!["a".into(), "c".into()]; + assert!(needs_rebalance(&a, &b)); + } + + #[test] + fn needs_rebalance_different_length_is_true() { + let a = vec!["a".into(), "b".into()]; + let b = vec!["a".into()]; + assert!(needs_rebalance(&a, &b)); + } +} diff --git a/minikv-core/tests/hashing.rs b/minikv-core/tests/hashing.rs new file mode 100644 index 0000000..05fc769 --- /dev/null +++ b/minikv-core/tests/hashing.rs @@ -0,0 +1,146 @@ +//! Hashing stability tests. +//! +//! These tests pin the exact outputs of `key_to_path` and the scoring function +//! used by `key_to_volume`. Any test failure indicates that stored data could +//! be misrouted. Such changes must be reverted or require a full data migration. + +use base64::Engine; +use minikv_core::hashing::{key_to_path, volume_score}; +use minikv_core::volumes::key_to_volume; + +// ── key_to_path stability ───────────────────────────────────────────────────── + +#[test] +fn key_to_path_hello_exact() { + // Exact BLAKE3 hash prefix and Base64 filename for "hello". + let path = key_to_path(b"hello"); + assert_eq!(path, "/ea/8f/aGVsbG8="); +} + +#[test] +fn key_to_path_helloworld_format() { + let path = key_to_path(b"helloworld"); + // Verify structure: /<2hex>/<2hex>/ + let parts: Vec<&str> = path.splitn(4, '/').collect(); + assert_eq!(parts.len(), 4, "path must contain exactly 3 slashes"); + assert_eq!(parts[0], "", "path must start with '/'"); + assert_eq!( + parts[1].len(), + 2, + "first directory component must be 2 hex chars" + ); + assert_eq!( + parts[2].len(), + 2, + "second directory component must be 2 hex chars" + ); + let decoded = base64::engine::general_purpose::STANDARD + .decode(parts[3]) + .expect("filename must be valid Base64"); + assert_eq!(decoded, b"helloworld"); +} + +#[test] +fn key_to_path_helloworld_prefix_snapshot() { + let path = key_to_path(b"helloworld"); + let hash = blake3::hash(b"helloworld"); + let b = hash.as_bytes(); + assert!( + path.starts_with(&format!("/{:02x}/{:02x}/", b[0], b[1])), + "prefix changed: {path}" + ); +} + +#[test] +fn key_to_path_empty_key() { + let path = key_to_path(b""); + // Empty input produces a valid deterministic path. + let parts: Vec<&str> = path.splitn(4, '/').collect(); + assert_eq!(parts.len(), 4); +} + +#[test] +fn key_to_path_binary_key_roundtrip() { + let key: Vec = (0u8..=255u8).collect(); + let path = key_to_path(&key); + let b64_part = path.splitn(4, '/').nth(3).unwrap(); + let decoded = base64::engine::general_purpose::STANDARD + .decode(b64_part) + .unwrap(); + assert_eq!(decoded, key); +} + +// ── key_to_volume stability ─────────────────────────────────────────────────── + +#[test] +fn key_to_volume_hello_deterministic() { + let vols: Vec = vec!["larry".into(), "moe".into(), "curly".into()]; + let a = key_to_volume(b"hello", &vols, 1, 1); + let b = key_to_volume(b"hello", &vols, 1, 1); + assert_eq!(a, b); +} + +#[test] +fn key_to_volume_returns_correct_count() { + let vols: Vec = vec!["larry".into(), "moe".into(), "curly".into()]; + assert_eq!(key_to_volume(b"hello", &vols, 1, 1).len(), 1); + assert_eq!(key_to_volume(b"hello", &vols, 2, 1).len(), 2); + assert_eq!(key_to_volume(b"hello", &vols, 3, 1).len(), 3); +} + +#[test] +fn key_to_volume_no_duplicate_volumes() { + let vols: Vec = vec!["larry".into(), "moe".into(), "curly".into()]; + let result = key_to_volume(b"hello", &vols, 3, 1); + let unique: std::collections::HashSet<_> = result.iter().collect(); + assert_eq!(unique.len(), 3); +} + +#[test] +fn key_to_volume_winner_is_highest_scorer() { + let vols: Vec = vec!["larry".into(), "moe".into(), "curly".into()]; + let result = key_to_volume(b"hello", &vols, 1, 1); + + let mut scored: Vec<_> = vols + .iter() + .map(|v| (volume_score(b"hello", v), v.clone())) + .collect(); + scored.sort_by(|a, b| b.0.cmp(&a.0)); + + assert_eq!(result[0], scored[0].1); +} + +#[test] +fn key_to_volume_subvolume_suffix_format() { + let vols: Vec = vec!["larry".into(), "moe".into(), "curly".into()]; + let result = key_to_volume(b"hello", &vols, 1, 10); + let v = &result[0]; + assert!(v.contains("/sv"), "missing /sv suffix: {v}"); + let sv_part = v.split("/sv").last().unwrap(); + assert_eq!(sv_part.len(), 2, "sv suffix must be 2 hex chars"); + assert!( + sv_part.chars().all(|c| c.is_ascii_hexdigit()), + "sv suffix must be hex" + ); +} + +#[test] +fn key_to_volume_subvolume_in_range() { + let vols: Vec = vec!["larry".into(), "moe".into(), "curly".into()]; + for key in [b"hello".as_ref(), b"world", b"foo", b"aaaa", b"zzzz"] { + let result = key_to_volume(key, &vols, 1, 8); + let sv_str = result[0].split("/sv").last().unwrap(); + let idx = u64::from_str_radix(sv_str, 16).unwrap(); + assert!(idx < 8, "sv index {idx} out of range for svcount=8"); + } +} + +#[test] +fn key_to_volume_single_subvolume_no_suffix() { + let vols: Vec = vec!["larry".into(), "moe".into()]; + let result = key_to_volume(b"hello", &vols, 1, 1); + assert!( + !result[0].contains("/sv"), + "should not have /sv when svcount=1" + ); +}