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, P: PlanExecutor> RaftLoop<A, P>

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_shuffle_receiver(self, receiver: Arc<dyn ShuffleReceiver>) -> Self

Attach the cross-node streaming-shuffle receiver (E1, builder chain).

The supplied implementation (backed by nodedb’s ShuffleReceiverRegistry) is called by the ShufflePush transport read-loop to create inboxes, deposit chunks, and record the per-part build barrier.

pub fn with_shuffle_producer(self, producer: Arc<dyn ShuffleProducer>) -> Self

Attach the cross-node shuffle PRODUCER (E4a, builder chain).

The supplied implementation (backed by nodedb’s local streaming executor and receiver registry) is called by the ShuffleProduce transport handler to run the local scan, hash-partition its rows, and fan them out to the part-owners.

pub fn with_shuffle_consumer(self, consumer: Arc<dyn ShuffleConsumer>) -> Self

Attach the cross-node shuffle CONSUMER (E4b, builder chain).

The supplied implementation (backed by nodedb’s shuffle registry + local executor) is called by the ShuffleConsume transport handler to wait for the part’s staged sides to finalize and run the node-local grace join.

pub fn with_shuffle_aggregator( self, aggregator: Arc<dyn ShuffleAggregator>, ) -> Self

Attach the cross-node distributed GROUP BY shuffle CONSUMER (E5b, builder chain).

SINGLE-SIDED aggregate sibling of with_shuffle_consumer. The supplied implementation (backed by nodedb’s shuffle registry + local executor) is called by the ShuffleAggregateConsume transport handler to wait for the part’s single staged side to finalize and run the node-local partial-state merge + finalize.

pub fn with_assign_remote_surrogate( self, assigner: Arc<dyn AssignRemoteSurrogate>, ) -> Self

Attach the routed-surrogate-exchange assigner (F1b, builder chain).

The supplied implementation (backed by nodedb’s SurrogateAssigner) is called by the AssignSurrogate transport handler when this node is the home vShard’s leader: it runs a LOCAL assign for the endpoint key, which yields the authoritative (first-wins, idempotent) surrogate the coordinator must use.

pub fn with_calvin_submit(self, submit: Arc<dyn CalvinSubmit>) -> Self

Attach the routed Calvin-submit hook (Cv1, builder chain).

The supplied implementation (backed by nodedb’s Calvin sequencer inbox and CalvinCompletionRegistry) is called by the SubmitCalvinTxn transport handler when this node is the sequencer-group leader: it submits the carried TxClass to the local inbox and awaits completion, so a cross-shard write routed here from a non-leader coordinator actually commits.

pub fn with_calvin_submit_inbox( self, submit: Arc<dyn CalvinSubmitInbox>, ) -> Self

Attach the routed Calvin-INBOX submit hook (Cv1, builder chain).

OLLP dependent sibling of with_calvin_submit. The supplied implementation (backed by nodedb’s Calvin sequencer inbox and CalvinCompletionRegistry) is called by the SubmitCalvinInbox transport handler when this node is the sequencer-group leader: it submits the carried dependent TxClass to the local inbox and awaits only the assignment, so a cross-shard write routed here from a non-leader coordinator gets its assignment back immediately (the OLLP coordinator drives it to completion itself).

pub fn with_reserve_read(self, hook: Arc<dyn ReserveRead>) -> Self

Attach the routed reserve-read hook (Calvin OLLP, builder chain).

The supplied implementation (backed by nodedb’s Calvin sequencer scheduler) is called by the ReserveRead transport handler when this node is the sequencer-group leader: it decodes the carried LockKey and assign-only reserves the read lock against the local lock table.

pub fn with_release_reservation(self, hook: Arc<dyn ReleaseReservation>) -> Self

Attach the routed release-reservation hook (Calvin OLLP, builder chain).

Ack-only sibling of with_reserve_read. The supplied implementation (backed by nodedb’s Calvin sequencer scheduler) is called by the ReleaseReservation transport handler when this node is the sequencer-group leader: it decodes the carried owner and release reason and releases the reservation.

pub fn with_snapshot_builder(self, builder: Arc<dyn SnapshotBuilder>) -> Self

Attach the per-group snapshot builder for the SEND path (builder chain).

The supplied implementation (backed by nodedb’s Data Plane snapshot dispatch) is called by the install-snapshot tick step to produce the real serialized engine state for a lagging follower’s group vshards. When not set, the sender falls back to the stub (empty) chunk.

pub fn with_snapshot_applier(self, applier: Arc<dyn SnapshotApplier>) -> Self

Attach the per-group snapshot applier for the RECEIVE path (builder chain).

The supplied implementation (backed by nodedb’s Data Plane restore dispatch) is called by the install-snapshot finalize path to apply a received per-group snapshot to the local state machine after the atomic .partial→.snap rename and before advancing Raft. When not set, the follower advances Raft without restoring engine state.

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 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_replication_factor(self, rf: u32) -> Self

Set the cluster replication factor (target voters per group).

Pass the value loaded from ClusterSettings::replication_factor at startup. The field is immutable after the loop is wrapped in Arc — call this on the builder chain before that wrap.

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.

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 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 node_id(&self) -> u64

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

pub fn replication_factor(&self) -> u32

Target replication factor for this cluster (voters per group).

Loaded once from ClusterSettings at startup and immutable for the loop’s lifetime. Returns 1 when no override was set (single-node default).

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 save_applied_index( &self, group_id: u64, applied_index: u64, ) -> Result<()>

Durably record applied_index as the group’s applied floor.

applied_index MUST name an entry whose state-machine effects are already durable — callers invoke this from the data-plane apply-completion path, after the write funnel’s fsync barrier has returned for that entry. The next boot resumes Raft delivery at applied_index + 1, which is what stops WAL replay and Raft replay both applying it.

Monotonic per group: an index at or below the current floor is a no-op.

pub fn maybe_compact_group( &self, group_id: u64, applied_index: u64, ) -> Result<bool>

Auto-compact a group’s Raft log if its configured threshold has been reached, given the DATA-PLANE applied watermark applied_index.

applied_index MUST be the index the data-plane state machine has durably applied to (NOT raft’s commit index). Callers invoke this from the data-plane apply-completion path so the SnapshotBuilder can never be asked to serialize state past what the engines have actually applied.

Returns Ok(true) when a compaction was performed. No-op (Ok(false)) when the group is absent, the threshold is None, or the retained-entry count is below the threshold.

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>

async fn handle_rpc_streaming( &self, req: ExecuteRequest, sink: impl ChunkSink, ) -> Option<TypedClusterError>

Handle a streaming ExecuteStreamRequest: execute the plan and push each result frame to sink. Returns None on a clean EOF or Some(err) on a terminal failure. The transport writes one RPC_EXECUTE_STREAM_END envelope carrying this outcome after the call returns. See crate::forward::PlanExecutor::execute_plan_streaming.

async fn on_shuffle_request(&self, req: ShufflePushRequest)

Cross-node streaming shuffle (E1) — opening frame of a ShufflePush stream. Lazily creates the receiver inbox for (shuffle_id, part, side) carrying producer_count and num_parts.

async fn on_shuffle_chunk( &self, shuffle_id: u64, part: u32, side: u8, payload: Vec<u8>, ) -> Result<()>

Deposit one shuffle chunk payload into the receiver inbox. Bounded — the implementation blocks while the inbox buffer is full so QUIC flow control back-pressures the producer.

async fn on_shuffle_end( &self, shuffle_id: u64, part: u32, side: u8, error: Option<TypedClusterError>, )

Terminal frame for one producer of a ShufflePush stream: record the End (advancing the per-part build barrier) and capture any terminal error.

async fn on_shuffle_produce( &self, req: ShuffleProduceRequest, ) -> ShuffleProduceResponse

Cross-node shuffle PRODUCER trigger (E4a). Execute the local scan fragment carried by req, hash-partition each output row on req.keys, and fan the rows out to the per-part owners as ShufflePush streams (looping back for parts this node owns). Returns a ShuffleProduceResponse whose error is None on a clean produce or Some(err) if the scan failed (every part has already been Ended with the same error), and whose read_version_lsn carries the max per-collection read version the scan observed. The transport writes exactly one ShuffleProduceResponse carrying this outcome back to the coordinator.

async fn on_shuffle_consume( &self, req: ShuffleConsumeRequest, ) -> ShuffleConsumeResponse

Cross-node shuffle CONSUMER trigger (E4b). Complete the part carried by req: wait for both staged sides of (shuffle_id, part) to finalize, run the node-local grace-hash join over them, and return the joined rows. The transport writes exactly one ShuffleConsumeResponse carrying the rows (or a typed error) back to the coordinator. Never hangs — the finalize wait is bounded by req.deadline_remaining_ms.

async fn on_shuffle_aggregate( &self, req: ShuffleAggregateConsumeRequest, ) -> ShuffleAggregateConsumeResponse

Cross-node distributed GROUP BY shuffle CONSUMER trigger (E5b). The single-sided aggregate sibling of on_shuffle_consume. Complete the part carried by req: wait for the part’s single staged producer side (side 0) of (shuffle_id, part) to finalize, merge + finalize the partial GroupStates, and return the aggregate rows. The transport writes exactly one ShuffleAggregateConsumeResponse carrying the rows (or a typed error) back to the coordinator. Never hangs — the finalize wait is bounded by req.deadline_remaining_ms.

async fn on_assign_surrogate( &self, req: AssignSurrogateRequest, ) -> AssignSurrogateResponse

Routed-surrogate-exchange (F1b). This node is the LEADER of the home vShard for the (collection, pk) endpoint key carried by req: assign-or-return the AUTHORITATIVE global surrogate (a LOCAL assign on the leader yields the authoritative, first-wins, idempotent value) and return it. The transport writes exactly one AssignSurrogateResponse carrying the surrogate (or a typed error) back to the coordinator.

async fn on_submit_calvin_txn( &self, req: SubmitCalvinTxnRequest, ) -> SubmitCalvinTxnResponse

Routed Calvin-submit (Cv1). This node is the SEQUENCER-GROUP leader: decode the TxClass carried by req, submit it to the local Calvin sequencer inbox, and await assignment + completion. The transport writes exactly one SubmitCalvinTxnResponse carrying success (or a typed error) back to the coordinator. The await is bounded by req.deadline_remaining_ms.

async fn on_submit_calvin_inbox( &self, req: SubmitCalvinInboxRequest, ) -> SubmitCalvinInboxResponse

Routed Calvin-INBOX submit (Cv1). The OLLP dependent sibling of on_submit_calvin_txn. This node is the SEQUENCER-GROUP leader: decode the TxClass carried by req, submit it to the local Calvin sequencer inbox, and await only the ASSIGNMENT (NOT completion). The transport writes exactly one SubmitCalvinInboxResponse carrying the assignment (or a typed error) back to the coordinator. The await is bounded by req.deadline_remaining_ms.

async fn on_reserve_read(&self, req: ReserveReadRequest) -> ReserveReadResponse

Routed reserve-read (Calvin OLLP). This node is the SEQUENCER-GROUP leader: decode the LockKey carried by req and assign-only reserve the read lock. The transport writes exactly one ReserveReadResponse carrying the minted owner (or a typed error) back to the coordinator. The reserve is bounded by req.deadline_remaining_ms.

async fn on_release_reservation( &self, req: ReleaseReservationRequest, ) -> ReleaseReservationResponse

Routed release-reservation (Calvin OLLP). The ack-only sibling of on_reserve_read. This node is the SEQUENCER-GROUP leader: decode the owner and release reason carried by req and release the reservation. The transport writes exactly one ReleaseReservationResponse carrying success (or a typed error) back to the coordinator. The release is bounded by req.deadline_remaining_ms.

async fn on_timeout_now(&self, req: TimeoutNowRequest)

TimeoutNow (leadership transfer). One-way — the receiver immediately starts an election for the matching group. No reply is written.