NodeController

Overview

The NodeController is the core orchestration component in Walrus that glues together metadata (cluster state), bucket storage (Walrus I/O), and RPC surfaces (Kafka facade + internal forwarding). It coordinates all interactions between these systems without parsing Kafka protocol details or managing retention policies. Location: distributed-walrus/src/controller/mod.rs:31

Core Responsibilities

Request Routing

Routes append and read operations to the correct leader nodes based on metadata

Lease Management

Synchronizes write leases with metadata to ensure only leaders can write

Request Forwarding

Forwards requests to remote nodes when the current node is not the leader

Structure

pub struct NodeController {
    pub node_id: NodeId,
    pub bucket: Arc<Storage>,
    pub metadata: Arc<Metadata>,
    pub raft: Arc<OctopiiNode>,
    pub offsets: Arc<RwLock<HashMap<String, u64>>>,
    pub read_cursors: Arc<Mutex<HashMap<String, ReadCursor>>>,
    pub test_fail_forward_read: AtomicBool,
    pub test_fail_monitor: AtomicBool,
    pub test_fail_dir_size: AtomicBool,
}

Fields

Field	Type	Purpose
`node_id`	`NodeId`	Unique identifier for this node in the cluster
`bucket`	`Arc<Storage>`	Handle to the underlying Walrus storage engine
`metadata`	`Arc<Metadata>`	Cluster metadata including topic/segment/leader mappings
`raft`	`Arc<OctopiiNode>`	Raft consensus node for metadata replication
`offsets`	`Arc<RwLock<HashMap<String, u64>>>`	Tracks logical offsets per WAL key
`read_cursors`	`Arc<Mutex<HashMap<String, ReadCursor>>>`	Per-topic read cursors for shared consumption
`test_fail_*`	`AtomicBool`	Fault injection flags for testing

Key Operations

Append Operations

`append_for_topic`

Appends data to a topic. Routes the request to the current leader node. Location: mod.rs:165

pub async fn append_for_topic(&self, topic: &str, data: Vec<u8>) -> Result<()>

Behavior:

Looks up topic state from metadata to find the current leader and segment
If this node is the leader, performs local append via forward_append
Otherwise, forwards the request to the remote leader via forward_append_remote
Returns error if topic doesn’t exist

Only the designated leader for a topic’s current segment can successfully append. Attempting to append on a non-leader will result in forwarding or error.

`forward_append` (Internal)

Handles local append operations with lease verification and retry logic. Location: internal.rs:6

pub(crate) async fn forward_append(&self, wal_key: String, data: Vec<u8>) -> InternalResp

Process:

Updates leases to ensure this node has write permission
Attempts append with retry via append_with_retry (2 attempts)
Records the append in offset tracking
Checks if segment rollover is needed via maybe_rollover
Returns InternalResp::Ok or InternalResp::Error

`append_with_retry`

Retry logic for appends with lease synchronization. Location: mod.rs:540

async fn append_with_retry(&self, wal_key: &str, data: Vec<u8>) -> Result<()>

Makes up to 2 attempts to append, refreshing leases between attempts if needed.

Read Operations

`read_one_for_topic`

Reads a single entry for a topic using the provided cursor, automatically advancing across sealed segments. Location: mod.rs:199

pub async fn read_one_for_topic(
    &self,
    topic: &str,
    cursor: &mut ReadCursor,
) -> Result<Option<Vec<u8>>>

Cursor Management:

Initializes cursor to segment 1 if unset
Advances to next segment when current segment is fully consumed
Uses metadata’s sealed_count to determine when segments are complete
Routes reads to appropriate leader (current or historical segment leader)

Returns:

Ok(Some(data)) - Successfully read an entry
Ok(None) - No data available (reached end of current segment)
Err(_) - Topic doesn’t exist or read error

`read_one_for_topic_shared`

Shared read path that maintains a per-topic cursor across client connections. Location: mod.rs:271

pub async fn read_one_for_topic_shared(&self, topic: &str) -> Result<Option<Vec<u8>>>

Uses a shared cursor stored in read_cursors to enable consumer groups or shared consumption patterns.

`forward_read` (Internal)

Performs local read from storage. Location: mod.rs:304

async fn forward_read(&self, wal_key: &str, _max_entries: usize) -> InternalResp

Returns InternalResp::ReadResult with data and high watermark (tracked entry count).

Topic Management

`ensure_topic`

Creates a topic if it doesn’t already exist, selecting an initial leader. Location: mod.rs:124

pub async fn ensure_topic(&self, topic: &str) -> Result<()>

Leader Selection:

If no nodes exist, uses current node as leader
Otherwise, hashes topic name to deterministically select a node
Proposes MetadataCmd::CreateTopic to Raft
Refreshes leases after creation

`maybe_rollover` (Internal)

Checks if a segment has reached capacity and triggers rollover to a new segment. Location: mod.rs:457

async fn maybe_rollover(&self, topic: &str, segment: u64) -> Result<()>

Process:

Checks if tracked entry count exceeds max_segment_entries() threshold
Retrieves current Raft membership to select next leader
Round-robin leader selection among cluster members
Proposes MetadataCmd::RolloverTopic with sealed segment count

Segment rollover is triggered automatically after appends. The sealed segment count is recorded in metadata for cursor advancement during reads.

Lease Management

`update_leases`

Synchronizes storage leases with current metadata state. Location: mod.rs:91

pub async fn update_leases(&self)

Process:

Queries metadata for all topics where node_id is the leader
Generates set of expected WAL keys (t_<topic>_s_<segment>)
Syncs peer addresses from metadata to Raft layer
Updates bucket storage with new lease set

Called by:

Lease update loop (every 100ms)
After metadata changes
During retry operations

`run_lease_update_loop`

Background task that periodically refreshes leases. Location: mod.rs:284

pub async fn run_lease_update_loop(self: Arc<Self>)

Runs indefinitely with 100ms intervals, calling update_leases().

Request Forwarding

`forward_append_remote`

Forwards append requests to remote leader nodes. Location: mod.rs:321

async fn forward_append_remote(
    &self,
    leader_node: NodeId,
    wal_key: String,
    data: Vec<u8>,
) -> Result<()>

Process:

Resolves leader node address from metadata
Serializes InternalOp::ForwardAppend payload
Sends RPC request via Raft handler with 15s timeout
Deserializes response and returns result

`forward_read_remote`

Forwards read requests to remote nodes. Location: mod.rs:494

async fn forward_read_remote(
    &self,
    leader_node: NodeId,
    wal_key: String,
) -> Result<Vec<Vec<u8>>>

Similar to append forwarding but for read operations, returning the data payload.

Metadata Operations

`propose_metadata`

Proposes metadata changes through Raft consensus. Location: mod.rs:373

pub(crate) async fn propose_metadata(&self, cmd: MetadataCmd) -> Result<()>

Leadership Handling:

If this node is Raft leader, proposes directly with 10s timeout
Otherwise, waits up to 5s to discover current leader
Forwards proposal to remote leader if necessary
Retries with 100ms intervals if leader is unknown

`upsert_node`

Registers or updates a node in cluster metadata. Location: mod.rs:158

pub async fn upsert_node(&self, node_id: NodeId, addr: String) -> Result<()>

Proposes MetadataCmd::UpsertNode to Raft.

`handle_join_cluster`

Handles cluster join requests from new nodes. Location: mod.rs:567

async fn handle_join_cluster(&self, node_id: u64, addr: String) -> InternalResp

Process:

Resolves node address (with DNS lookup fallback)
Adds node as Raft learner
Records node in metadata via upsert_node
Spawns background task to promote learner to voter once caught up (60s timeout, 120 attempts)

RPC Handler

`handle_rpc`

Main RPC dispatch handler for internal operations. Location: mod.rs:44

pub async fn handle_rpc(&self, op: InternalOp) -> InternalResp

Supported Operations:

ForwardAppend - Handles forwarded append requests
ForwardRead - Handles forwarded read requests
ForwardMetadata - Handles forwarded metadata proposals
JoinCluster - Handles cluster join requests
TestControl - Fault injection for testing

Helper Functions

`wal_key`

Generates a WAL key from topic name and segment number. Location: types.rs:3

pub fn wal_key(topic: &str, segment: u64) -> String

Format: t_<topic>_s_<segment> (e.g., t_events_s_1)

`parse_wal_key`

Parses a WAL key back into topic and segment. Location: types.rs:8

pub fn parse_wal_key(wal_key: &str) -> Option<(String, u64)>

Example: t_events_s_1 → ("events", 1)

Read Cursor

Location: types.rs:19

#[derive(Default, Clone, Debug)]
pub struct ReadCursor {
    pub segment: u64,
    pub delivered_in_segment: u64,
}

Tracks read position within a topic:

segment - Current segment being read
delivered_in_segment - Number of entries delivered from this segment

Lifecycle

Initialization - Controller is created with node ID, storage, metadata, and Raft instances
Lease Loop - Background task continuously syncs leases with metadata
Request Processing - Handles Kafka API calls and internal RPC operations
Leader Changes - Automatically adapts to leadership changes via metadata updates
Segment Rollover - Automatically creates new segments when capacity reached

Thread Safety

All shared state is protected by:

RwLock for offsets (read-heavy, write-occasional)
Mutex for read cursors (exclusive access needed)
Atomic bools for test flags (lock-free)
Arc for shared ownership across async tasks

Error Handling

NotLeaderForPartition

Returned when attempting to append to a WAL key without a valid lease. Clients should retry against the correct leader.

Raft Propose Timeout

Metadata proposals timeout after 10 seconds. Indicates Raft consensus issues or cluster unavailability.

Unknown Topic

Returned when operating on a topic that hasn’t been created. Use ensure_topic first.

Forward Failures

Remote forwarding can fail due to network issues, leader unavailability, or lease mismatches. Includes 15s RPC timeout.

Testing Hooks

The controller includes fault injection capabilities for testing:

test_fail_forward_read - Forces read forwarding to fail
test_fail_monitor - Forces monitoring operations to fail
test_fail_dir_size - Forces directory size checks to fail

These are controlled via TestControl RPC operations.

Metadata - Cluster state management
Bucket Storage - Lease-based storage operations

Library API

CLI Commands

Cluster API

Documentation Index

​Overview

​Core Responsibilities

Request Routing

Lease Management

Request Forwarding

​Structure

​Fields

​Key Operations

​Append Operations

​append_for_topic

​forward_append (Internal)

​append_with_retry

​Read Operations

​read_one_for_topic

​read_one_for_topic_shared

​forward_read (Internal)

​Topic Management

​ensure_topic

​maybe_rollover (Internal)

​Lease Management

​update_leases

​run_lease_update_loop

​Request Forwarding

​forward_append_remote

​forward_read_remote

​Metadata Operations

​propose_metadata

​upsert_node

​handle_join_cluster

​RPC Handler

​handle_rpc

​Helper Functions

​wal_key

​parse_wal_key

​Read Cursor

​Lifecycle

​Thread Safety

​Error Handling

​Testing Hooks

​Related Components

Build docs developers (and LLMs) love

Overview

Core Responsibilities

Structure

Fields

Key Operations

Append Operations

`append_for_topic`

`forward_append` (Internal)

`append_with_retry`

Read Operations

`read_one_for_topic`

`read_one_for_topic_shared`

`forward_read` (Internal)

Topic Management

`ensure_topic`

`maybe_rollover` (Internal)

Lease Management

`update_leases`

`run_lease_update_loop`

Request Forwarding

`forward_append_remote`

`forward_read_remote`

Metadata Operations

`propose_metadata`

`upsert_node`

`handle_join_cluster`

RPC Handler

`handle_rpc`

Helper Functions

`wal_key`

`parse_wal_key`

Read Cursor

Lifecycle

Thread Safety

Error Handling

Testing Hooks

Related Components