Struct Node 

pub struct Node<T>
where
    T: TypeConfig,
{ /* 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

Implementations

impl<T> Node<T>

where T: TypeConfig,

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

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 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 membership_change_notifier(&self) -> Receiver<MembershipSnapshot>

Subscribe to committed membership change notifications.

Returns a watch::Receiver that fires whenever a ConfChange entry commits. The first borrow() returns the current membership state without waiting for a change.

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.

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 'life0: 'async_trait, Node<T>: '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 'life0: 'async_trait, Node<T>: '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 'life0: 'async_trait, Node<T>: '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 'life0: 'async_trait, Node<T>: 'async_trait,

New RPC for leader discovery

impl<T> Debug for Node<T>

where T: TypeConfig,

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

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 'life0: 'async_trait, Node<T>: '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 'life0: 'async_trait, Node<T>: '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 handle_client_scan<'life0, 'async_trait>( &'life0 self, request: Request<ScanRequest>, ) -> Pin<Box<dyn Future<Output = Result<Response<ScanResponse>, Status>> + Send + 'async_trait>>

where 'life0: 'async_trait, Node<T>: 'async_trait,

Scan all keys under a prefix.

Routes through the Raft command channel so the leader serves the scan (linearizable by default). Returns all matching entries plus the applied index at scan time — clients use the revision to filter watch events during reconnection.

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

where 'life0: 'async_trait, Node<T>: '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

fn watch_membership<'life0, 'async_trait>( &'life0 self, _request: Request<WatchMembershipRequest>, ) -> Pin<Box<dyn Future<Output = Result<Response<<Node<T> as RaftClientService>::WatchMembershipStream>, Status>> + Send + 'async_trait>>

where 'life0: 'async_trait, Node<T>: 'async_trait,

Stream committed membership snapshots to the client.

Sends the current snapshot immediately on connect (via mark_changed), then one snapshot per committed ConfChange. Closes with UNAVAILABLE on server shutdown so clients know to reconnect.

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

Server streaming response type for the Watch method.

type WatchMembershipStream = Pin<Box<dyn Stream<Item = Result<MembershipSnapshot, Status>> + Send>>

Server streaming response type for the WatchMembership 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 'life0: 'async_trait, Node<T>: '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 'life0: 'async_trait, Node<T>: '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

fn stream_append_entries<'life0, 'async_trait>( &'life0 self, request: Request<Streaming<AppendEntriesRequest>>, ) -> Pin<Box<dyn Future<Output = Result<Response<<Node<T> as RaftReplicationService>::StreamAppendEntriesStream>, Status>> + Send + 'async_trait>>

where 'life0: 'async_trait, Node<T>: 'async_trait,

Processes a persistent bidirectional AppendEntries stream from the cluster leader.

Decouples request ingestion from response emission:

• recv task: reads batches from the stream, dispatches each as a RaftEvent::AppendEntries (non-blocking between batches)
• forwarder task: drains ordered response handles sequentially; ordering is guaranteed by the Raft single-threaded event loop

type StreamAppendEntriesStream = Pin<Box<dyn Stream<Item = Result<AppendEntriesResponse, Status>> + Send>>

Server streaming response type for the StreamAppendEntries method.

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<<Node<T> as SnapshotService>::StreamSnapshotStream>, Status>> + Send + 'async_trait>>

where 'life0: 'async_trait, Node<T>: '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 'life0: 'async_trait, Node<T>: 'async_trait,

Leader-driven snapshot streaming