diff --git a/crates/ember-core/src/engine.rs b/crates/ember-core/src/engine.rs index 801d1a00..200f92af 100644 --- a/crates/ember-core/src/engine.rs +++ b/crates/ember-core/src/engine.rs @@ -4,8 +4,6 @@ //! of the key. Each shard is an independent tokio task — no locks on //! the hot path. -use std::hash::{Hash, Hasher}; - use crate::dropper::DropHandle; use crate::error::ShardError; use crate::keyspace::ShardConfig; @@ -197,14 +195,25 @@ impl Engine { /// Pure function: maps a key to a shard index. /// -/// Uses ahash (AHash) for fast, non-cryptographic hashing. ~3x faster -/// than SipHash for short keys. Deterministic within a single process — -/// that's all we need for local sharding. Shard routing is trusted -/// internal logic so DoS-resistant hashing is unnecessary here. +/// Uses FNV-1a hashing for deterministic shard routing across restarts. +/// This is critical for AOF/snapshot recovery — keys must hash to the +/// same shard on every startup, otherwise recovered data lands in the +/// wrong shard. +/// +/// FNV-1a is simple, fast for short keys, and completely deterministic +/// (no per-process randomization). Shard routing is trusted internal +/// logic so DoS-resistant hashing is unnecessary here. fn shard_index(key: &str, shard_count: usize) -> usize { - let mut hasher = ahash::AHasher::default(); - key.hash(&mut hasher); - (hasher.finish() as usize) % shard_count + // FNV-1a 64-bit + const FNV_OFFSET: u64 = 0xcbf29ce484222325; + const FNV_PRIME: u64 = 0x100000001b3; + + let mut hash = FNV_OFFSET; + for byte in key.as_bytes() { + hash ^= *byte as u64; + hash = hash.wrapping_mul(FNV_PRIME); + } + (hash as usize) % shard_count } #[cfg(test)] diff --git a/tests/integration/Cargo.toml b/tests/integration/Cargo.toml index 8507a90c..7b286e85 100644 --- a/tests/integration/Cargo.toml +++ b/tests/integration/Cargo.toml @@ -14,8 +14,9 @@ harness = true tokio = { workspace = true, features = ["rt-multi-thread", "macros", "net", "io-util", "time"] } bytes = { workspace = true } ember-protocol = { workspace = true } +ember-server = { path = "../../crates/ember-server", default-features = false } prost-reflect = { workspace = true } tempfile = "3" [features] -protobuf = [] +protobuf = ["ember-server/protobuf"] diff --git a/tests/integration/src/basic_operations.rs b/tests/integration/src/basic_operations.rs index 1e488ae4..ce56d83e 100644 --- a/tests/integration/src/basic_operations.rs +++ b/tests/integration/src/basic_operations.rs @@ -2,7 +2,7 @@ use ember_protocol::Frame; -use crate::helpers::TestServer; +use crate::helpers::{ServerOptions, TestServer}; #[tokio::test] async fn ping_pong() { @@ -206,7 +206,11 @@ async fn type_command() { #[tokio::test] async fn rename() { - let server = TestServer::start(); + // use a single shard so RENAME doesn't hit cross-shard errors + let server = TestServer::start_with(ServerOptions { + shards: Some(1), + ..Default::default() + }); let mut c = server.connect().await; c.ok(&["SET", "old", "value"]).await; diff --git a/tests/integration/src/helpers.rs b/tests/integration/src/helpers.rs index ef8343c5..b0b1581f 100644 --- a/tests/integration/src/helpers.rs +++ b/tests/integration/src/helpers.rs @@ -33,6 +33,8 @@ pub struct ServerOptions { pub cluster_bootstrap: bool, /// Enable protobuf value storage. pub protobuf: bool, + /// Number of shards (defaults to 2 for test coverage). + pub shards: Option, /// Use concurrent (DashMap) mode instead of sharded channels. pub concurrent: bool, } @@ -54,7 +56,7 @@ impl TestServer { let mut cmd = Command::new(&binary); cmd.arg("--port").arg(port.to_string()); cmd.arg("--host").arg("127.0.0.1"); - cmd.arg("--shards").arg("2"); + cmd.arg("--shards").arg(opts.shards.unwrap_or(2).to_string()); // suppress tracing output in tests cmd.env("RUST_LOG", "error");