Struct ClusterController 

pub struct ClusterController { /* private fields */ }

Facade composing the cluster-control primitives.

Implementations

impl ClusterController

pub fn new( instance_id: NodeId, kv: Arc<dyn ClusterKv>, snapshot: Option<Arc<AssignmentSnapshotStore>>, members_rx: Receiver<Vec<NodeInfo>>, ) -> Self

Wrap the given primitives.

pub fn register_query_handler(&self, handler: Arc<dyn RemoteQueryHandler>)

Register the handler serving cross-node RemoteScan.

pub fn query_client_pool(&self) -> &QueryClientPool

Access the connection pool for remote queries.

pub fn cluster_min_watermark(&self) -> Option<i64>

Latest cluster-wide minimum watermark seen by this instance. None until the leader has published a Commit with a populated min_watermark_ms.

pub fn publish_cluster_min_watermark(&self, wm: i64)

Leader-side monotonic publish. The leader computes the cluster-wide minimum watermark in await_prepare_quorum (its own local watermark folded with every follower’s ack) and must mirror it into the controller atomic so its own operators see the same value that followers pick up via observe_barrier on the matching Commit. Never lowers the published value — event-time progress is monotonic.

pub fn instance_id(&self) -> NodeId

This instance’s ID.

pub fn kv(&self) -> &Arc<dyn ClusterKv>

The cluster gossip KV, exposed so higher layers can advertise/discover per-stream state alongside the control-plane keys.

pub fn current_leader(&self) -> Option<NodeId>

Current leader (lowest id among Active peers plus self).

pub fn is_leader(&self) -> bool

True if this instance is currently the leader.

pub fn set_active(&self, active: bool)

Mark this node’s active status.

pub fn live_instances(&self) -> Vec<NodeId>

Live instance IDs: Active peers plus self.

pub fn note_unresponsive(&self, peers: &[NodeId])

Record peers that failed to ack a capture quorum in time.

pub fn note_responsive(&self, peers: &[NodeId])

Clear peers that acked (they are demonstrably alive).

pub fn is_recently_unresponsive(&self, peer: NodeId) -> bool

Whether peer failed a capture quorum within the TTL window.

pub fn begin_drain(&self)

Mark this node as draining. Idempotent.

pub fn is_draining(&self) -> bool

Whether this node is draining.

pub fn assignable_instances(&self) -> Vec<NodeId>

Node ids eligible to own vnodes: Active peers, plus self unless this node is draining. Mirrors how Self::live_instances folds self in, but filters non-Active peers (see assignable_node_ids) so Joining/Suspected/Draining/Left nodes never receive vnodes.

pub fn set_self_locality(&self, locality: Locality)

Record this node’s own locality. Call once at startup.

pub fn assignable_with_locality(&self) -> Vec<(NodeId, Locality)>

Self::assignable_instances paired with each node’s Locality (peers’ from members_rx, self’s from Self::set_self_locality).

pub fn members_watch(&self) -> Receiver<Vec<NodeInfo>>

Cloneable membership watch. Background tasks subscribe to this to react to join/leave events (changed().await) without polling Self::live_instances on a timer.

pub async fn announce_snapshot_version(&self, version: u64)

Write the current assignment snapshot version to gossip KV.

pub async fn read_snapshot_version(&self) -> Option<u64>

Read the snapshot version from all peers in gossip KV and return the maximum version.

pub async fn start_barrier_server( &self, bind_addr: SocketAddr, advertise_host: Option<String>, ) -> Result<SocketAddr, String>

Start the direct gRPC barrier sync server.

Errors

Propagates BarrierCoordinator::start_server errors.

pub async fn announce_barrier( &self, ann: &BarrierAnnouncement, ) -> Result<(), String>

Leader-side announce.

Errors

Propagates BarrierCoordinator::announce errors.

pub async fn observe_barrier( &self, ) -> Result<Option<BarrierAnnouncement>, String>

Follower-side observe; Ok(None) if no leader is visible.

As a side effect, an Aligned or Commit announcement with a populated min_watermark_ms updates the shared cluster-min-watermark atomic so operators on this instance see the cluster-wide minimum without a separate polling path (Aligned carries it so a resuming pipeline sees fresh event-time progress before the upload-gated Commit).

Errors

Propagates BarrierCoordinator::observe errors.

pub async fn ack_barrier(&self, ack: &BarrierAck) -> Result<(), String>

Follower-side ack.

Errors

Propagates BarrierCoordinator::ack errors.

pub async fn wait_for_barrier<F>( &self, pred: F, timeout: Duration, ) -> Option<BarrierAnnouncement>

where F: FnMut(&BarrierAnnouncement) -> bool,

Wait until Self::observe_barrier yields an announcement matching pred, or timeout expires (→ None). Push-driven off the gRPC announcement watch when available; gossip-KV-only deployments (and KV-only announcements) are covered by a fallback poll — 250ms with the watch, 25ms without.

pub async fn wait_for_quorum( &self, epoch: u64, expected: &[NodeId], deadline: Duration, ) -> QuorumOutcome

Leader-side: poll until quorum or deadline.

pub fn snapshot_store(&self) -> Option<&AssignmentSnapshotStore>

Assignment snapshot store, if configured.

Trait Implementations

impl Debug for ClusterController

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

Formats the value using the given formatter.