Struct Node 

pub struct Node<T>
where
    T: TypeConfig,
{
    pub node_config: Arc<RaftNodeConfig>,
    /* private fields */
}

Raft consensus node

Represents a single node participating in a Raft cluster. Coordinates protocol execution, storage, and networking.

Created via NodeBuilder.

Running the Node

let node = builder.start()?;
node.run().await?;  // Blocks until shutdown

Fields

node_config: Arc<RaftNodeConfig>

Raft node config

Implementations

impl<T> Node<T>

where T: TypeConfig,

pub async fn run(&self) -> Result<()>

Starts and runs the Raft node’s main execution loop.

Workflow

Strategy-based bootstrap depending on node type:

• Learner: Skip cluster ready check, join cluster after warmup
• Voter: Wait for cluster ready, then warmup connections

Both paths converge to the Raft event processing loop.

Errors

Returns Err if any bootstrap step or Raft execution fails.

Example

let node = Node::new(...);
tokio::spawn(async move {
    node.run().await.expect("Node execution failed");
});

pub fn set_rpc_ready(&self, is_ready: bool)

Marks the node’s RPC server as ready to accept requests.

Parameters

• is_ready: When true, marks RPC server as ready. When false, marks server as temporarily unavailable.

Note

This indicates the RPC server is listening, NOT that leader election is complete. Use leader_change_notifier() to wait for leader election.

Usage

Called internally after RPC server starts and cluster health check passes.

pub fn is_rpc_ready(&self) -> bool

Checks if the node’s RPC server is ready to accept requests.

Returns

true if the RPC server is operational and listening, false otherwise.

Note

This does NOT indicate leader election status. Use leader_change_notifier() for that.

pub fn ready_notifier(&self) -> Receiver<bool>

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

let ready_rx = node.ready_notifier();
ready_rx.wait_for(|&ready| ready).await?;
// RPC server is now listening

pub fn leader_change_notifier(&self) -> Receiver<Option<LeaderInfo>>

Returns a receiver for leader change notifications.

Subscribe to be notified when:

• First leader is elected (initial election)
• Leader changes (re-election)
• No leader exists (during election)

Performance

Event-driven notification, <1ms latency

Example

let mut leader_rx = node.leader_change_notifier();
while leader_rx.changed().await.is_ok() {
    if let Some(info) = leader_rx.borrow().as_ref() {
        println!("Leader: {} (term {})", info.leader_id, info.term);
    }
}

pub fn from_raft(raft: Raft<T>, shutdown_signal: Receiver<()>) -> Self

Create a Node from a pre-built Raft instance This method is designed to support testing and external builders

pub fn node_id(&self) -> u32

Returns this node’s unique identifier.

Useful for logging, metrics, and integrations that need to identify which Raft node is handling operations.

pub fn local_client(&self) -> LocalKvClient

Creates a zero-overhead local KV client for embedded access.

Returns a client that directly communicates with Raft core without gRPC serialization or network traversal.

Performance

• 10-20x faster than gRPC client
• <0.1ms latency per operation

Example

let node = NodeBuilder::new(config).start().await?;
let client = node.local_client();
client.put(b"key", b"value").await?;

Trait Implementations

impl<T> ClusterManagementService for Node<T>

where T: TypeConfig,

fn update_cluster_conf<'life0, 'async_trait>( &'life0 self, request: Request<ClusterConfChangeRequest>, ) -> Pin<Box<dyn Future<Output = Result<Response<ClusterConfUpdateResponse>, Status>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Handles cluster membership changes (joint consensus)

Raft Protocol Logic

• Implements cluster configuration changes (Section 6)
• Validates new configuration against current cluster state
• Ensures safety during membership transitions

fn get_cluster_metadata<'life0, 'async_trait>( &'life0 self, request: Request<MetadataRequest>, ) -> Pin<Box<dyn Future<Output = Result<Response<ClusterMembership>, Status>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Returns current cluster membership and state metadata

Usage

• Administrative API for cluster inspection
• Provides snapshot of current configuration

fn join_cluster<'life0, 'async_trait>( &'life0 self, request: Request<JoinRequest>, ) -> Pin<Box<dyn Future<Output = Result<Response<JoinResponse>, Status>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Request to join the cluster as a new learner node

fn discover_leader<'life0, 'async_trait>( &'life0 self, request: Request<LeaderDiscoveryRequest>, ) -> Pin<Box<dyn Future<Output = Result<Response<LeaderDiscoveryResponse>, Status>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

New RPC for leader discovery

impl<T> Debug for Node<T>

where T: TypeConfig,

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<T> RaftClientService for Node<T>

where T: TypeConfig,

fn handle_client_write<'life0, 'async_trait>( &'life0 self, request: Request<ClientWriteRequest>, ) -> Pin<Box<dyn Future<Output = Result<Response<ClientResponse>, Status>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Processes client write requests requiring consensus

Raft Protocol Logic

• Entry point for client proposals (Section 7)
• Validates requests before appending to leader’s log
• Ensures linearizable writes through log replication

fn handle_client_read<'life0, 'async_trait>( &'life0 self, request: Request<ClientReadRequest>, ) -> Pin<Box<dyn Future<Output = Result<Response<ClientResponse>, Status>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Handles client read requests with linearizability guarantees

Raft Protocol Logic

• Implements lease-based leader reads (Section 6.4)
• Verifies leadership before serving reads
• Ensures read-after-write consistency

fn watch<'life0, 'async_trait>( &'life0 self, request: Request<WatchRequest>, ) -> Pin<Box<dyn Future<Output = Result<Response<Self::WatchStream>, Status>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Watch for changes to a specific key

Returns a stream of events (PUT/DELETE) for the specified key. The stream will continue until the client disconnects or the server shuts down.

Arguments

• request - Contains the key to watch

Returns

A stream of WatchResponse messages containing PUT/DELETE events

Errors

Returns Status::UNAVAILABLE if Watch is disabled in configuration

type WatchStream = Pin<Box<dyn Stream<Item = Result<WatchResponse, Status>> + Send>>

Server streaming response type for the Watch method.

impl<T> RaftElectionService for Node<T>

where T: TypeConfig,

fn request_vote<'life0, 'async_trait>( &'life0 self, request: Request<VoteRequest>, ) -> Pin<Box<dyn Future<Output = Result<Response<VoteResponse>, Status>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Handles RequestVote RPC calls from candidate nodes during leader elections

Raft Protocol Logic

• Part of leader election mechanism (Section 5.2)
• Validates candidate’s term and log completeness
• Grants vote if candidate’s log is at least as up-to-date as local log

impl<T> RaftReplicationService for Node<T>

where T: TypeConfig,

fn append_entries<'life0, 'async_trait>( &'life0 self, request: Request<AppendEntriesRequest>, ) -> Pin<Box<dyn Future<Output = Result<Response<AppendEntriesResponse>, Status>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Processes AppendEntries RPC calls from cluster leader

Raft Protocol Logic

• Heartbeat mechanism (Section 5.2)
• Log replication entry point (Section 5.3)
• Term comparison logic:
  • If incoming term > current term: revert to follower state
  • Reset election timeout on valid leader communication

impl<T> SnapshotService for Node<T>

where T: TypeConfig,

type StreamSnapshotStream = Streaming<SnapshotChunk>

Server streaming response type for the StreamSnapshot method.

fn stream_snapshot<'life0, 'async_trait>( &'life0 self, request: Request<Streaming<SnapshotAck>>, ) -> Pin<Box<dyn Future<Output = Result<Response<Self::StreamSnapshotStream>, Status>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Learner-driven snapshot streaming

fn install_snapshot<'life0, 'async_trait>( &'life0 self, request: Request<Streaming<SnapshotChunk>>, ) -> Pin<Box<dyn Future<Output = Result<Response<SnapshotResponse>, Status>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Leader-driven snapshot streaming

fn purge_log<'life0, 'async_trait>( &'life0 self, request: Request<PurgeLogRequest>, ) -> Pin<Box<dyn Future<Output = Result<Response<PurgeLogResponse>, Status>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Log compaction request