Struct RaftLoop 

pub struct RaftLoop<A: CommitApplier, P: PlanExecutor = NoopPlanExecutor> { /* private fields */ }

Raft event loop coordinator.

Owns the MultiRaft state (behind Arc<Mutex>) and drives it via periodic ticks. Implements crate::transport::RaftRpcHandler (in super::handle_rpc) so it can be passed directly to NexarTransport::serve for incoming RPC dispatch.

The F: RequestForwarder generic parameter was removed in C-δ.6 when the SQL-string forwarding path was retired. Cross-node SQL routing now goes through gateway.execute / ExecuteRequest (C-β path).

Implementations

impl<A: CommitApplier> RaftLoop<A>

pub fn new( multi_raft: MultiRaft, transport: Arc<NexarTransport>, topology: Arc<RwLock<ClusterTopology>>, applier: A, ) -> Self

impl<A: CommitApplier, P: PlanExecutor> RaftLoop<A, P>

pub fn set_snapshot_quarantine_hook( &mut self, hook: Arc<dyn SnapshotQuarantineHook>, )

Install the snapshot quarantine hook (mutable setter variant).

Prefer with_snapshot_quarantine_hook on the builder chain unless you need to set the hook after construction.

pub fn with_plan_executor<P2: PlanExecutor>( self, executor: Arc<P2>, ) -> RaftLoop<A, P2>

Install a custom plan executor (for cluster mode — C-β path).

pub fn with_group_watchers(self, watchers: Arc<GroupAppliedWatchers>) -> Self

Replace the per-group apply watcher registry.

The host crate calls this with the same Arc it stores on SharedState so proposers and consistent-read paths share one registry with the tick loop’s bump points. Defaults to a fresh empty registry when not set.

pub fn with_snapshot_quarantine_hook( self, hook: Arc<dyn SnapshotQuarantineHook>, ) -> Self

Attach the snapshot quarantine hook (builder chain variant).

The supplied implementation is called by the InstallSnapshotRequest handler to check for, record, and short-circuit quarantined chunks.

pub fn with_data_dir(self, data_dir: PathBuf) -> Self

Set the data directory for partial-snapshot persistence and GC.

When set, the InstallSnapshotRequest handler writes chunks to <data_dir>/recv_snapshots/<group_id>.partial and the GC sweeper removes stale partials on startup. When None (the default, used by unit tests), disk writes are skipped — the receiver operates in-memory only with empty chunk data.

pub fn with_snapshot_chunk_bytes(self, chunk_bytes: u64) -> Self

Override the snapshot chunk byte size (default: 4 MiB).

pub fn with_orphan_partial_max_age_secs(self, secs: u64) -> Self

Override the orphan-partial max age for the GC sweeper (default: 300 s).

pub fn group_watchers(&self) -> Arc<GroupAppliedWatchers>

Shared handle to the per-group apply watcher registry.

pub fn loop_metrics(&self) -> Arc<LoopMetrics>

Shared handle to this loop’s standardized metrics.

pub fn pending_groups(&self) -> usize

Count of Raft groups currently mounted on this node — used to render the raft_tick_loop_pending_groups gauge.

pub fn begin_shutdown(&self)

Signal cooperative shutdown to every detached task spawned inside [super::tick::do_tick].

This is the entry point for test harnesses that want to tear down a RaftLoop without waiting for the external run() shutdown watch channel to propagate. In production the same signal is emitted automatically by run() when its external shutdown receiver fires.

Idempotent: calling this multiple times is a no-op after the first.

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

Subscribe to the boot-time readiness signal.

The returned receiver starts at false and flips to true exactly once, after the first [super::tick::do_tick] completes phase 4 (apply committed entries). Used by the host crate to gate client-facing listener startup until the metadata raft group has produced its first applied entry.

pub fn with_vshard_handler(self, handler: VShardEnvelopeHandler) -> Self

Set a handler for incoming VShardEnvelope messages.

pub fn with_metadata_applier(self, applier: Arc<dyn MetadataApplier>) -> Self

Install the metadata applier used for group-0 commits.

The host crate (nodedb) calls this with a production applier that wraps an in-memory MetadataCache and additionally persists to redb / broadcasts catalog change events. The default NoopMetadataApplier is kept only for tests that don’t care.

pub fn with_tick_interval(self, interval: Duration) -> Self

pub fn with_catalog(self, catalog: Arc<ClusterCatalog>) -> Self

Attach a cluster catalog — used by the join flow to persist the updated topology + routing after a conf-change commits.

pub fn node_id(&self) -> u64

This node’s id (exposed for handlers and tests).

pub async fn run(&self, shutdown: Receiver<bool>)

Run the event loop until shutdown.

This drives Raft elections, heartbeats, and message dispatch. Call NexarTransport::serve separately with Arc<Self> as the handler.

When the externally-supplied shutdown receiver fires, the loop also propagates the signal to the internal cooperative-shutdown channel so every detached task spawned inside do_tick exits promptly and drops its Arc<Mutex<MultiRaft>> clone.

pub fn multi_raft_handle(&self) -> Arc<Mutex<MultiRaft>>

Returns the inner multi-raft handle. Exposed for tests and for the host crate’s metadata proposer so it can hold a second reference to the same underlying mutex without pulling the whole raft loop into the caller’s lifetime.

pub fn group_statuses(&self) -> Vec<GroupStatus>

Snapshot all Raft group states for observability (SHOW RAFT GROUPS).

impl<A: CommitApplier, P: PlanExecutor> RaftLoop<A, P>

pub fn propose(&self, vshard_id: u32, data: Vec<u8>) -> Result<(u64, u64)>

Propose a command to the Raft group owning the given vShard.

Returns (group_id, log_index) on success.

pub fn propose_to_metadata_group(&self, data: Vec<u8>) -> Result<u64>

Propose a command directly to the metadata Raft group (group 0).

Used by the host crate’s metadata proposer and by integration tests that exercise the replicated-catalog path without a pgwire client. Fails with ClusterError::GroupNotFound if group 0 does not exist on this node, and with ClusterError::Raft(NotLeader) if this node is not the current leader of group 0.

pub async fn propose_to_metadata_group_via_leader( &self, data: Vec<u8>, ) -> Result<u64>

Propose to the metadata Raft group, transparently forwarding to the current leader if this node is not it.

Tries a local propose first. On ClusterError::Raft(NotLeader { leader_hint }), looks up the hinted leader’s address in cluster topology and sends a crate::rpc_codec::MetadataProposeRequest over QUIC. The receiving leader applies the proposal locally and returns the log index.

On NotLeader { leader_hint: None } (election in progress, no observed leader yet) the call returns the original NotLeader error so the caller can decide whether to retry. We deliberately do not implement a wait-and-retry loop here because the caller (the host-side proposer) may have a shorter deadline than any reasonable retry budget.

The leader-side path through this function is identical to the bare propose_to_metadata_group — the only extra cost is an is_leader_locally check before the local propose.

pub async fn propose_via_data_leader( &self, vshard_id: u32, data: Vec<u8>, ) -> Result<(u64, u64)>

Propose a command to the data Raft group owning the given vShard, transparently forwarding to the group leader if this node is not it.

Tries a local propose first. On NotLeader { leader_hint: Some(id) }, looks up the hinted leader’s address in the cluster topology and sends a DataProposeRequest over QUIC. The receiving leader applies the proposal locally and returns (group_id, log_index).

On NotLeader { leader_hint: None } (election in progress) the call returns the original NotLeader error so the caller can retry.

pub fn propose_conf_change( &self, group_id: u64, change: &ConfChange, ) -> Result<(u64, u64)>

Propose a configuration change to a Raft group.

Returns (group_id, log_index) on success.

Trait Implementations

impl<A, P> GroupStatusProvider for RaftLoop<A, P>

where A: CommitApplier, P: PlanExecutor,

fn group_statuses(&self) -> Vec<GroupStatus>

Current status of every Raft group hosted on this node.

impl<A: CommitApplier, P: PlanExecutor> RaftRpcHandler for RaftLoop<A, P>

async fn handle_rpc(&self, rpc: RaftRpc) -> Result<RaftRpc>