diff --git a/MIGRATION_GUIDE.md b/MIGRATION_GUIDE.md index 84033586..0ff5d0cd 100644 --- a/MIGRATION_GUIDE.md +++ b/MIGRATION_GUIDE.md @@ -331,3 +331,90 @@ persistence_strategy = "MemFirst" --- **Last Updated:** February 2026 + +--- + +## 🚨 For v0.2.3 Users: API Surface Changes in v0.2.4 (#326) + +### What Changed + +v0.2.4 removes internal implementation details that were accidentally exposed as `pub`. +All removed items were internal — they were never part of the documented public API. + +### Breaking Changes + +#### 1. `EmbeddedClient::node_id()` removed + +This method returned `client_id` (not a node ID), which was semantically incorrect. + +```rust +// Old (v0.2.3) — broken semantics, removed +let id = client.node_id(); + +// New (v0.2.4) — use EmbeddedEngine instead +let id = engine.node_id(); +``` + +#### 2. `GrpcClient` convenience methods now require `ClientApi` trait in scope + +`get_linearizable()`, `get_lease()`, and `get_eventual()` are now only available via +the `ClientApi` trait. The return type is unified to `Option` (was `Option`). + +```rust +// Old (v0.2.3) +let result = client.get_linearizable("key").await?; +let value = result.map(|r| r.value); // extra unwrap needed + +// New (v0.2.4) — add trait import, get Bytes directly +use d_engine_client::ClientApi; +let value = client.get_linearizable("key").await?; // Option +``` + +#### 3. `Node::set_rpc_ready()`, `is_rpc_ready()`, `ready_notifier()` removed from public API + +These were internal lifecycle methods. Use `EmbeddedEngine::wait_ready()` instead. + +```rust +// Old (v0.2.3) +node.set_rpc_ready(true); +let ready = node.is_rpc_ready(); + +// New (v0.2.4) — use the engine-level API +engine.wait_ready(Duration::from_secs(5)).await?; +``` + +#### 4. `Node::node_config` field is no longer public + +```rust +// Old (v0.2.3) +let node_id = node.node_config.cluster.node_id; + +// New (v0.2.4) +let node_id = node.node_id(); +``` + +#### 5. `NodeBuilder::init()` is no longer public + +Use the documented constructors instead. + +```rust +// Old (v0.2.3) +NodeBuilder::init(config, shutdown_rx) + +// New (v0.2.4) +NodeBuilder::new(None, shutdown_rx).node_config(config) +// or +NodeBuilder::from_cluster_config(cluster_config, shutdown_rx) +``` + +#### 6. `QuorumStatus` removed + +This type was defined but never used. Remove any references to it. + +#### 7. `ClientInner` and `ConnectionPool` no longer public + +These are internal connection pool types. Use `Client` and `ClientBuilder` instead. + +--- + +**Last Updated:** April 2026 diff --git a/d-engine-client/src/grpc_client.rs b/d-engine-client/src/grpc_client.rs index f21779e6..91a42745 100644 --- a/d-engine-client/src/grpc_client.rs +++ b/d-engine-client/src/grpc_client.rs @@ -40,29 +40,6 @@ impl GrpcClient { Self { client_inner } } - // Convenience methods for explicit consistency levels - pub async fn get_linearizable( - &self, - key: impl AsRef<[u8]>, - ) -> std::result::Result, ClientApiError> { - self.get_with_policy(key, Some(ReadConsistencyPolicy::LinearizableRead)).await - } - - pub async fn get_lease( - &self, - key: impl AsRef<[u8]>, - ) -> std::result::Result, ClientApiError> { - self.get_with_policy(key, Some(ReadConsistencyPolicy::LeaseRead)).await - } - - pub async fn get_eventual( - &self, - key: impl AsRef<[u8]>, - ) -> std::result::Result, ClientApiError> { - self.get_with_policy(key, Some(ReadConsistencyPolicy::EventualConsistency)) - .await - } - /// Retrieves a single key's value with explicit consistency policy /// /// Allows client to override server's default consistency policy for this specific request. diff --git a/d-engine-client/src/grpc_client_test.rs b/d-engine-client/src/grpc_client_test.rs index 65528be4..6a841a0f 100644 --- a/d-engine-client/src/grpc_client_test.rs +++ b/d-engine-client/src/grpc_client_test.rs @@ -604,7 +604,7 @@ async fn test_get_linearizable_success() { let result = client.get_linearizable(key).await; assert!(result.is_ok()); - assert_eq!(result.unwrap().as_ref().map(|r| &r.value), Some(&value)); + assert_eq!(result.unwrap(), Some(value)); } #[tokio::test] @@ -642,7 +642,7 @@ async fn test_get_lease_success() { let result = client.get_lease(key).await; assert!(result.is_ok()); - assert_eq!(result.unwrap().as_ref().map(|r| &r.value), Some(&value)); + assert_eq!(result.unwrap(), Some(value)); } #[tokio::test] @@ -680,7 +680,7 @@ async fn test_get_eventual_success() { let result = client.get_eventual(key).await; assert!(result.is_ok()); - assert_eq!(result.unwrap().as_ref().map(|r| &r.value), Some(&value)); + assert_eq!(result.unwrap(), Some(value)); } #[tokio::test] @@ -843,7 +843,7 @@ async fn test_get_consistency_methods_failure() { let key = "test_key".to_string().into_bytes(); // Test all convenience methods with network failure - let methods: [(&str, ClientApiResult>); 3] = [ + let methods: [(&str, ClientApiResult>); 3] = [ ( "get_linearizable", client.get_linearizable(key.clone()).await, diff --git a/d-engine-client/src/lib.rs b/d-engine-client/src/lib.rs index a4532870..2ce85e34 100644 --- a/d-engine-client/src/lib.rs +++ b/d-engine-client/src/lib.rs @@ -75,8 +75,7 @@ pub use builder::*; pub use config::*; pub use d_engine_core::client::{ClientApi, ClientApiError, ClientApiResult}; pub use grpc_client::*; -pub use pool::*; -pub use utils::*; +pub(crate) use pool::*; // ==================== Protocol Types (Essential for Public API) ==================== @@ -131,11 +130,11 @@ pub struct Client { } #[derive(Clone)] -pub struct ClientInner { - pool: ConnectionPool, - client_id: u32, - config: ClientConfig, - endpoints: Vec, +pub(crate) struct ClientInner { + pub(crate) pool: ConnectionPool, + pub(crate) client_id: u32, + pub(crate) config: ClientConfig, + pub(crate) endpoints: Vec, } impl std::ops::Deref for Client { diff --git a/d-engine-client/src/utils_test.rs b/d-engine-client/src/utils_test.rs index 546fef07..08ae0add 100644 --- a/d-engine-client/src/utils_test.rs +++ b/d-engine-client/src/utils_test.rs @@ -1,4 +1,4 @@ -use super::*; +use crate::utils::address_str; #[test] fn test_address_str_no_scheme() { diff --git a/d-engine-core/src/errors.rs b/d-engine-core/src/errors.rs index 2e9f22e9..1da23407 100644 --- a/d-engine-core/src/errors.rs +++ b/d-engine-core/src/errors.rs @@ -264,6 +264,7 @@ pub enum StorageError { NotServing(String), } +#[doc(hidden)] #[derive(Debug, thiserror::Error)] pub enum IdAllocationError { /// ID allocation overflow @@ -279,6 +280,7 @@ pub enum IdAllocationError { NoIdsAvailable, } +#[doc(hidden)] #[derive(Debug, thiserror::Error)] pub enum FileError { #[error("Path does not exist: {0}")] @@ -304,6 +306,7 @@ pub enum FileError { } /// Error type for value conversion operations +#[doc(hidden)] #[derive(Debug, thiserror::Error)] pub enum ConvertError { /// Invalid input length error @@ -319,6 +322,7 @@ pub enum ConvertError { ConversionFailure(String), } +#[doc(hidden)] #[derive(Debug, thiserror::Error)] pub enum ReadSendError { #[error("Network timeout")] @@ -328,6 +332,7 @@ pub enum ReadSendError { Connection(#[from] tonic::transport::Error), } +#[doc(hidden)] #[derive(Debug, thiserror::Error)] pub enum WriteSendError { #[error("Not cluster leader")] @@ -374,6 +379,7 @@ pub enum SystemError { } // Serialization is classified separately (across protocol layers and system layers) +#[doc(hidden)] #[derive(Debug, thiserror::Error)] pub enum SerializationError { #[error("Bincode serialization failed: {0}")] @@ -381,6 +387,7 @@ pub enum SerializationError { } /// Wrapper for prost encoding/decoding errors +#[doc(hidden)] #[derive(Debug, thiserror::Error)] pub enum ProstError { #[error("Encoding failed: {0}")] @@ -390,6 +397,7 @@ pub enum ProstError { Decode(#[from] prost::DecodeError), } +#[doc(hidden)] #[derive(Debug, thiserror::Error)] pub enum ElectionError { /// General election process failure @@ -425,6 +433,7 @@ pub enum ElectionError { NoVotingMemberFound { candidate_id: u32 }, } +#[doc(hidden)] #[derive(Debug, thiserror::Error)] pub enum ReplicationError { /// Stale leader detected during AppendEntries RPC @@ -465,6 +474,7 @@ pub enum ReplicationError { } /// Errors that can occur during ReadIndex batching for linearizable reads +#[doc(hidden)] #[derive(Debug, thiserror::Error, Clone)] pub enum ReadIndexError { /// This node is no longer the leader @@ -550,6 +560,7 @@ pub enum MembershipError { ClusterMetadataNotInitialized, } +#[doc(hidden)] #[derive(Debug, thiserror::Error)] pub enum SnapshotError { #[error("Snapshot receiver lagging, dropping chunk")] diff --git a/d-engine-core/src/lib.rs b/d-engine-core/src/lib.rs index 6e9eb347..da3d8d9e 100644 --- a/d-engine-core/src/lib.rs +++ b/d-engine-core/src/lib.rs @@ -77,35 +77,55 @@ pub mod storage; #[cfg(feature = "watch")] pub mod watch; +// ── User-facing public API ────────────────────────────────────────────────── pub use client::*; -pub use commit_handler::*; pub use config::*; -pub use election::*; pub use errors::*; +pub use storage::*; +#[cfg(feature = "watch")] +pub use watch::*; + +// Stable extension points — types developers need when implementing custom +// storage engines, state machines, or transport layers. +pub use membership::Membership; +pub use network::Transport; +pub use purge::PurgeExecutor; +pub use raft::{LeaderInfo, Raft, SignalParams}; +pub use state_machine_handler::{SnapshotPolicy, StateMachineHandler}; +pub use type_config::TypeConfig; + +// ── Internal implementation details (not part of public API) ─────────────── +// These remain accessible to d-engine-server but are hidden from cargo doc. +// Do not depend on these from external crates. +#[doc(hidden)] +pub use commit_handler::*; +#[doc(hidden)] +pub use election::*; +#[doc(hidden)] pub use event::*; +#[doc(hidden)] pub use maybe_clone_oneshot::*; +#[doc(hidden)] pub use membership::*; +#[doc(hidden)] pub use network::*; +#[doc(hidden)] pub use purge::*; -pub use raft::*; +#[doc(hidden)] pub use raft_context::*; -pub use replication::*; -pub use state_machine_handler::*; -pub use storage::*; -#[cfg(feature = "watch")] -pub use watch::*; - -#[cfg(test)] -mod raft_test; - #[doc(hidden)] pub use raft_role::*; -pub(crate) use timer::*; +#[doc(hidden)] +pub use replication::*; +#[doc(hidden)] +pub use state_machine_handler::*; #[doc(hidden)] pub use type_config::*; #[doc(hidden)] pub use utils::*; +pub(crate) use timer::*; + #[cfg(test)] mod maybe_clone_oneshot_test; @@ -161,10 +181,3 @@ pub(crate) fn is_target_log_more_recent( (target_last_log_term > my_last_log_term) || (target_last_log_term == my_last_log_term && target_last_log_index >= my_last_log_index) } - -#[derive(Debug, Clone, Copy)] -pub enum QuorumStatus { - Confirmed, // Confirmed by the majority of nodes - LostQuorum, // Unable to obtain majority - NetworkError, // Network problem (can be retried) -} diff --git a/d-engine-server/src/api/embedded.rs b/d-engine-server/src/api/embedded.rs index e238fe42..f48e8b8d 100644 --- a/d-engine-server/src/api/embedded.rs +++ b/d-engine-server/src/api/embedded.rs @@ -147,6 +147,7 @@ struct Inner { client: Arc, leader_elected_rx: watch::Receiver>, is_stopped: Mutex, + node_id: u32, } /// Embedded d-engine with automatic lifecycle management. @@ -351,6 +352,8 @@ impl EmbeddedEngine { Arc::new(client) }; + let node_id = node.node_id(); + // Spawn node.run() in background let node_handle = tokio::spawn(async move { if let Err(e) = node.run().await { @@ -370,6 +373,7 @@ impl EmbeddedEngine { client, leader_elected_rx, is_stopped: Mutex::new(false), + node_id, }), }) } @@ -503,7 +507,7 @@ impl EmbeddedEngine { .leader_elected_rx .borrow() .as_ref() - .map(|info| info.leader_id == self.inner.client.node_id()) + .map(|info| info.leader_id == self.inner.node_id) .unwrap_or(false) } @@ -546,7 +550,6 @@ impl EmbeddedEngine { Arc::clone(&self.inner.client) } - /// Gracefully stop the embedded engine. /// Stop the embedded d-engine gracefully (idempotent). /// /// This method: @@ -608,12 +611,9 @@ impl EmbeddedEngine { *self.inner.is_stopped.lock().await } - /// Returns the node ID for testing purposes. - /// - /// Useful in integration tests that need to identify which node - /// they're interacting with, especially in multi-node scenarios. + /// Returns the unique identifier for this Raft node. pub fn node_id(&self) -> u32 { - self.inner.client.node_id() + self.inner.node_id } } diff --git a/d-engine-server/src/api/embedded_client.rs b/d-engine-server/src/api/embedded_client.rs index bdda6b7e..4865bbd2 100644 --- a/d-engine-server/src/api/embedded_client.rs +++ b/d-engine-server/src/api/embedded_client.rs @@ -152,6 +152,10 @@ impl EmbeddedClient { } /// Store a key-value pair with strong consistency. + /// + /// # Errors + /// Returns an error if the node is not the leader, the channel is closed, + /// the operation times out, or the state machine returns a server error. pub async fn put( &self, key: impl AsRef<[u8]>, @@ -380,7 +384,11 @@ impl EmbeddedClient { } } - /// Delete a key-value pair. + /// Delete a key-value pair with strong consistency. + /// + /// # Errors + /// Returns an error if the node is not the leader, the channel is closed, + /// the operation times out, or the state machine returns a server error. pub async fn delete( &self, key: impl AsRef<[u8]>, @@ -424,11 +432,6 @@ impl EmbeddedClient { self.timeout } - /// Returns the node ID for testing purposes - pub fn node_id(&self) -> u32 { - self.client_id - } - /// Watch for changes to a specific key /// /// Returns a `WatcherHandle` that yields watch events when the key's value changes. diff --git a/d-engine-server/src/lib.rs b/d-engine-server/src/lib.rs index 0338d456..4508d2a0 100644 --- a/d-engine-server/src/lib.rs +++ b/d-engine-server/src/lib.rs @@ -118,20 +118,21 @@ pub use api::StandaloneEngine; // -------------------- Error Types -------------------- /// Unified error type for all d-engine operations pub use d_engine_core::Error; -/// Hard state (Raft persistent state: term, voted_for, log) -pub use d_engine_core::HardState; /// Log storage trait pub use d_engine_core::LogStore; /// Metadata storage trait pub use d_engine_core::MetaStore; -/// Protocol buffer error type -pub use d_engine_core::ProstError; /// Unified result type (equivalent to Result) pub use d_engine_core::Result; -/// Snapshot operation error type -pub use d_engine_core::SnapshotError; /// Storage-specific error type pub use d_engine_core::StorageError; +// Internal types required by storage implementations — not part of user API +#[doc(hidden)] +pub use d_engine_core::HardState; +#[doc(hidden)] +pub use d_engine_core::ProstError; +#[doc(hidden)] +pub use d_engine_core::SnapshotError; // -------------------- Client API -------------------- pub use api::EmbeddedClient; /// Unified client operations trait (put/get/delete/CAS/watch). @@ -180,9 +181,6 @@ mod membership; mod network; mod utils; -#[doc(hidden)] -pub use d_engine_core::Raft; - // ==================== Test Utilities ==================== /// Test utilities for d-engine-server diff --git a/d-engine-server/src/node/builder.rs b/d-engine-server/src/node/builder.rs index 8100cc0c..5d934d2e 100644 --- a/d-engine-server/src/node/builder.rs +++ b/d-engine-server/src/node/builder.rs @@ -15,7 +15,7 @@ //! ## Example //! ```ignore //! let (shutdown_tx, shutdown_rx) = watch::channel(()); -//! let node = NodeBuilder::init(config, shutdown_rx) +//! let node = NodeBuilder::from_node_config(config, shutdown_rx) //! .storage_engine(custom_storage_engine) // Required component //! .state_machine(custom_state_machine) // Required component //! .start().await?; @@ -169,8 +169,32 @@ where Self::init(node_config, shutdown_signal) } + /// Constructs NodeBuilder from a fully-built [`RaftNodeConfig`]. + /// + /// Use this when you have already assembled a complete `RaftNodeConfig` and + /// want to avoid the implicit `RaftNodeConfig::new()` + `validate()` call + /// that [`new`](NodeBuilder::new) performs internally before applying your + /// config. That detour can panic in environments where no default config + /// file or environment variables are present. + /// + /// # Arguments + /// * `node_config` - Fully assembled and validated node configuration + /// * `shutdown_signal` - Watch channel for graceful shutdown signaling + /// + /// # Usage + /// ```ignore + /// let config: RaftNodeConfig = /* build and validate your config */; + /// let builder = NodeBuilder::from_node_config(config, shutdown_rx); + /// ``` + pub fn from_node_config( + node_config: RaftNodeConfig, + shutdown_signal: watch::Receiver<()>, + ) -> Self { + Self::init(node_config, shutdown_signal) + } + /// Core initialization logic shared by all construction paths - pub fn init( + pub(crate) fn init( node_config: RaftNodeConfig, shutdown_signal: watch::Receiver<()>, ) -> Self { @@ -211,6 +235,7 @@ where mut self, node_config: RaftNodeConfig, ) -> Self { + self.node_id = node_config.cluster.node_id; self.node_config = node_config; self } @@ -599,22 +624,18 @@ where }) } - /// Spawn watch dispatcher as background task. - /// - /// The dispatcher manages all watch streams for the lifetime of the node. /// Sets a custom state machine handler implementation. /// - /// This allows developers to provide their own implementation of the state machine handler - /// which processes committed log entries and applies them to the state machine. + /// Allows providing a custom implementation that processes committed log entries + /// and applies them to the state machine. If not set, a default implementation + /// is used during `build()`. /// /// # Arguments - /// * `handler` - custom state machine handler that must implement the `StateMachineHandler` - /// trait + /// * `handler` - custom handler implementing the `StateMachineHandler` trait /// /// # Notes - /// - The handler must be thread-safe as it will be shared across multiple threads - /// - If not set, a default implementation will be used during `build()` - /// - The handler should properly handle snapshot creation and restoration + /// - The handler must be thread-safe (shared across threads via `Arc`) + /// - The handler must correctly handle snapshot creation and restoration pub fn with_custom_state_machine_handler( mut self, handler: Arc>>, @@ -664,7 +685,7 @@ where /// /// # Example /// ```ignore - /// let node = NodeBuilder::init(config, shutdown_rx) + /// let node = NodeBuilder::from_node_config(config, shutdown_rx) /// .storage_engine(storage) /// .state_machine(state_machine) /// .start().await?; diff --git a/d-engine-server/src/node/builder_test.rs b/d-engine-server/src/node/builder_test.rs index b5249321..03f0e5eb 100644 --- a/d-engine-server/src/node/builder_test.rs +++ b/d-engine-server/src/node/builder_test.rs @@ -265,3 +265,35 @@ fn test_partial_override() { "Cluster config should remain unchanged" ); } + +#[test] +fn test_from_node_config_preserves_caller_config() { + let (_, shutdown_rx) = watch::channel(()); + let mut config = RaftNodeConfig::new().unwrap().validate().unwrap(); + config.cluster.node_id = 99; + + let builder = + NodeBuilder::::from_node_config(config, shutdown_rx); + + // Verify caller's config is used directly — no detour through RaftNodeConfig::new() defaults + assert_eq!( + builder.node_config.cluster.node_id, 99, + "from_node_config must use caller's config without going through RaftNodeConfig::new()" + ); +} + +#[test] +fn test_node_config_setter_syncs_node_id() { + let (_, shutdown_rx) = watch::channel(()); + let mut config = RaftNodeConfig::new().unwrap().validate().unwrap(); + config.cluster.node_id = 7; + + let builder = NodeBuilder::::new(None, shutdown_rx) + .node_config(config); + + // node_config on the builder must reflect the new config after setter + assert_eq!( + builder.node_config.cluster.node_id, 7, + "node_config() setter must sync node_id into both node_config and internal node_id field" + ); +} diff --git a/d-engine-server/src/node/mod.rs b/d-engine-server/src/node/mod.rs index 0a524b36..321f1c7e 100644 --- a/d-engine-server/src/node/mod.rs +++ b/d-engine-server/src/node/mod.rs @@ -94,7 +94,7 @@ where pub(crate) leader_notifier: LeaderNotifier, /// Raft node config - pub node_config: Arc, + pub(crate) node_config: Arc, /// Optional watch registry for watcher registration /// When None, watch functionality is disabled @@ -261,7 +261,7 @@ where /// /// # Usage /// Called internally after RPC server starts and cluster health check passes. - pub fn set_rpc_ready( + pub(crate) fn set_rpc_ready( &self, is_ready: bool, ) { @@ -279,25 +279,10 @@ where /// /// # Note /// This does NOT indicate leader election status. Use `leader_change_notifier()` for that. - pub fn is_rpc_ready(&self) -> bool { + pub(crate) fn is_rpc_ready(&self) -> bool { self.ready.load(Ordering::Acquire) } - /// Returns a receiver for node readiness notifications. - /// - /// Subscribe to this channel to be notified when the node becomes ready - /// to participate in cluster operations (NOT the same as leader election). - /// - /// # Example - /// ```ignore - /// let ready_rx = node.ready_notifier(); - /// ready_rx.wait_for(|&ready| ready).await?; - /// // RPC server is now listening - /// ``` - pub fn ready_notifier(&self) -> watch::Receiver { - self.rpc_ready_tx.subscribe() - } - /// Returns a receiver for leader change notifications. /// /// Subscribe to be notified when: @@ -321,44 +306,6 @@ where self.leader_notifier.subscribe() } - /// Create a Node from a pre-built Raft instance - /// This method is designed to support testing and external builders - pub fn from_raft( - raft: Raft, - shutdown_signal: watch::Receiver<()>, - ) -> Self { - let event_tx = raft.event_sender(); - let node_config = raft.ctx.node_config(); - let membership = raft.ctx.membership(); - let node_id = raft.node_id; - - let (rpc_ready_tx, _rpc_ready_rx) = watch::channel(false); - let leader_notifier = LeaderNotifier::new(); - - // Create dummy cmd_tx (this path is mainly for testing) - let (cmd_tx, _cmd_rx) = mpsc::channel(1024); - - Node { - node_id, - raft_core: Arc::new(Mutex::new(raft)), - membership, - event_tx, - cmd_tx, - ready: AtomicBool::new(false), - rpc_ready_tx, - leader_notifier, - node_config, - #[cfg(feature = "watch")] - watch_registry: None, - #[cfg(feature = "watch")] - _watch_dispatcher_handle: None, - sm_worker_handle: std::sync::Mutex::new(None), - _commit_handler_handle: None, - _lease_cleanup_handle: None, - shutdown_signal, - } - } - /// Returns this node's unique identifier. /// /// Useful for logging, metrics, and integrations that need to identify diff --git a/d-engine-server/src/node/node_test.rs b/d-engine-server/src/node/node_test.rs index 1b7bdb68..fad52430 100644 --- a/d-engine-server/src/node/node_test.rs +++ b/d-engine-server/src/node/node_test.rs @@ -451,25 +451,21 @@ mod leader_change_tests { } #[tokio::test] - async fn test_ready_notifier_independent() { + async fn test_rpc_ready_and_leader_election_are_independent() { let (node, _shutdown_tx) = create_test_node_arc(); - let mut ready_rx = node.ready_notifier(); let mut leader_rx = node.leader_change_notifier(); // Initially both should be false/None - assert!(!(*ready_rx.borrow())); + assert!(!node.is_rpc_ready()); assert_eq!(*leader_rx.borrow(), None); - // Set ready + // Set RPC ready — leader channel must not be affected node.set_rpc_ready(true); - ready_rx.changed().await.expect("Should receive ready change"); - assert!(*ready_rx.borrow()); - - // Leader should still be None + assert!(node.is_rpc_ready()); assert_eq!(*leader_rx.borrow(), None); - // Now set leader + // Elect leader — rpc_ready must remain unchanged let leader_info = LeaderInfo { leader_id: 1, term: 1, @@ -480,9 +476,7 @@ mod leader_change_tests { }); leader_rx.changed().await.expect("Should receive leader change"); assert_eq!(*leader_rx.borrow(), Some(leader_info)); - - // Ready should still be true - assert!(*ready_rx.borrow()); + assert!(node.is_rpc_ready()); } } @@ -507,21 +501,6 @@ mod readiness_tests { assert!(!node.is_rpc_ready()); } - #[tokio::test] - async fn test_ready_notifier_receives_updates() { - let (node, _shutdown_tx) = create_test_node(); - - let mut ready_rx = node.ready_notifier(); - - // Initial state - assert!(!(*ready_rx.borrow())); - - // Set ready - node.set_rpc_ready(true); - ready_rx.changed().await.expect("Should receive update"); - assert!(*ready_rx.borrow()); - } - #[tokio::test] async fn test_concurrent_set_ready() { let (node, _shutdown_tx) = create_test_node_arc(); diff --git a/d-engine-server/tests/cas_operations/distributed_lock_standalone.rs b/d-engine-server/tests/cas_operations/distributed_lock_standalone.rs index 1a350e8a..dea8a281 100644 --- a/d-engine-server/tests/cas_operations/distributed_lock_standalone.rs +++ b/d-engine-server/tests/cas_operations/distributed_lock_standalone.rs @@ -108,7 +108,7 @@ async fn test_distributed_lock_standalone() -> Result<(), ClientApiError> { info!("Test 6: Verify new lock holder"); let new_holder = client.get_linearizable(lock_key).await?; assert_eq!( - new_holder.map(|result| result.value), + new_holder, Some(Bytes::from(b"client_b".to_vec())), "Lock should be held by client_b" ); diff --git a/d-engine-server/tests/cas_operations/snapshot_recovery_standalone.rs b/d-engine-server/tests/cas_operations/snapshot_recovery_standalone.rs index 96408678..12ea87d3 100644 --- a/d-engine-server/tests/cas_operations/snapshot_recovery_standalone.rs +++ b/d-engine-server/tests/cas_operations/snapshot_recovery_standalone.rs @@ -133,7 +133,7 @@ async fn test_snapshot_recovery_standalone() -> Result<(), ClientApiError> { info!("Step 5: Verifying lock persists after recovery"); let recovered_holder = client.get_eventual(lock_key).await?; assert_eq!( - recovered_holder.expect("Lock key should exist").value.as_ref(), + recovered_holder.expect("Lock key should exist").as_ref(), b"owner_before_snapshot", "Lock should persist after snapshot recovery" ); diff --git a/d-engine-server/tests/common/mod.rs b/d-engine-server/tests/common/mod.rs index 9e84e419..ba641d02 100644 --- a/d-engine-server/tests/common/mod.rs +++ b/d-engine-server/tests/common/mod.rs @@ -128,6 +128,16 @@ pub async fn create_node_config( [raft.persistence] strategy = "MemFirst" flush_policy = {{ Batch = {{ threshold = 100, idle_flush_interval_ms = 1 }} }} + + [raft.election] + election_timeout_min = 300 + election_timeout_max = 3000 + + [retry.election] + max_retries = 5 + timeout_ms = 2000 + base_delay_ms = 100 + max_delay_ms = 5000 "# ) } @@ -170,6 +180,16 @@ pub async fn create_node_config_with_role( [raft.persistence] strategy = "MemFirst" flush_policy = {{ Batch = {{ threshold = 1, idle_flush_interval_ms = 1 }} }} + + [raft.election] + election_timeout_min = 300 + election_timeout_max = 3000 + + [retry.election] + max_retries = 5 + timeout_ms = 2000 + base_delay_ms = 100 + max_delay_ms = 5000 "# ) } @@ -284,7 +304,7 @@ pub async fn start_node( let node = build_node(config, graceful_rx, state_machine, storage_engine).await?; let node_clone = node.clone(); - let node_id = node.node_config.cluster.node_id; + let node_id = node.node_id(); let handle = tokio::spawn(async move { run_node(node_id, node_clone).await }); Ok((graceful_tx, handle)) @@ -300,7 +320,7 @@ async fn build_node( ClientApiError, > { // Prepare raft log entries - let mut builder = NodeBuilder::init(config.clone(), graceful_rx); + let mut builder = NodeBuilder::new(None, graceful_rx).node_config(config.clone()); // Use provided storage engine or create a new one with temp directory let storage_engine = if let Some(s) = storage_engine { diff --git a/d-engine-server/tests/failover_and_recovery/leader_failover_standalone.rs b/d-engine-server/tests/failover_and_recovery/leader_failover_standalone.rs index 3de75c0b..b2979d0c 100644 --- a/d-engine-server/tests/failover_and_recovery/leader_failover_standalone.rs +++ b/d-engine-server/tests/failover_and_recovery/leader_failover_standalone.rs @@ -76,7 +76,7 @@ async fn test_3_node_failover() -> Result<(), ClientApiError> { val }; assert_eq!( - result.expect("Key should exist after write").value.as_ref(), + result.expect("Key should exist after write").as_ref(), b"initial-value" ); @@ -125,7 +125,7 @@ async fn test_3_node_failover() -> Result<(), ClientApiError> { } val }; - assert_eq!(old_val.unwrap().value.as_ref(), b"initial-value"); + assert_eq!(old_val.unwrap().as_ref(), b"initial-value"); let new_val = { let mut val = None; @@ -138,7 +138,7 @@ async fn test_3_node_failover() -> Result<(), ClientApiError> { } val }; - assert_eq!(new_val.unwrap().value.as_ref(), b"still-works"); + assert_eq!(new_val.unwrap().as_ref(), b"still-works"); info!("Failover test passed. Cluster operational with 2/3 nodes"); @@ -173,7 +173,7 @@ async fn test_3_node_failover() -> Result<(), ClientApiError> { } val }; - assert_eq!(synced_val.unwrap().value.as_ref(), b"still-works"); + assert_eq!(synced_val.unwrap().as_ref(), b"still-works"); info!("Node 1 synced successfully. Test complete"); diff --git a/examples/service-discovery-standalone/watcher.rs b/examples/service-discovery-standalone/watcher.rs index d2b39a28..1e6484fd 100644 --- a/examples/service-discovery-standalone/watcher.rs +++ b/examples/service-discovery-standalone/watcher.rs @@ -6,6 +6,7 @@ use anyhow::Result; use clap::Parser; use d_engine_client::Client; +use d_engine_client::ClientApi; use d_engine_client::protocol::{WatchEventType, WatchResponse}; use futures::StreamExt; use std::time::Duration; @@ -54,7 +55,7 @@ async fn main() -> Result<()> { .await .map_err(|e| anyhow::anyhow!("Read failed: {e:?}"))?; if let Some(result) = current { - println!(" {} = {}", cli.key, String::from_utf8_lossy(&result.value)); + println!(" {} = {}", cli.key, String::from_utf8_lossy(&result)); } else { println!(" {} = (not found)", cli.key); } diff --git a/examples/three-nodes-standalone/docker/jepsen/vendor/dengine_ctl/src/main.rs b/examples/three-nodes-standalone/docker/jepsen/vendor/dengine_ctl/src/main.rs index 5ccb895c..07b5888a 100644 --- a/examples/three-nodes-standalone/docker/jepsen/vendor/dengine_ctl/src/main.rs +++ b/examples/three-nodes-standalone/docker/jepsen/vendor/dengine_ctl/src/main.rs @@ -2,8 +2,7 @@ use anyhow::Result; use clap::Parser; use clap::Subcommand; use d_engine::client::ClientApiError; -use d_engine::proto::client::ClientResult; - +use d_engine::ClientApi; use d_engine::ClientBuilder; use d_engine::ConvertError; @@ -51,7 +50,6 @@ async fn handle_write( ) -> Result<()> { println!("Writing key-value({}-{}) pair", key, value); client - .put(safe_kv(key), safe_kv(value)) .await .map(|_| println!("Success")) @@ -64,7 +62,6 @@ async fn handle_delete( ) -> Result<()> { println!("Deleting key({})", key); client - .delete(safe_kv(key)) .await .map(|_| println!("Success")) @@ -76,14 +73,15 @@ async fn handle_read( key: u64, linearizable: bool, ) -> Result<()> { - let result = client - - .get(safe_kv(key), linearizable) - .await - .map_err(|e: ClientApiError| anyhow::anyhow!("Read error: {:?}", e))?; + let result = if linearizable { + client.get_linearizable(safe_kv(key)).await + } else { + client.get_eventual(safe_kv(key)).await + } + .map_err(|e: ClientApiError| anyhow::anyhow!("Read error: {:?}", e))?; match result { - Some(ClientResult { key: _, value }) => { + Some(value) => { let value = safe_vk(&value).unwrap(); println!("{:?}", value); }