laminar_core/cluster/control/barrier/
mod.rs

1//! Cross-instance barrier protocol. Direct gRPC leader-to-follower calls
2//! under `cluster`, falling back to gossip-KV announce/ack/poll.
3
4#[cfg(feature = "cluster")]
5use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
6use std::sync::Arc;
7use std::time::{Duration, Instant};
8
9use async_trait::async_trait;
10use parking_lot::Mutex;
11use rustc_hash::{FxHashMap, FxHashSet};
12use serde::{Deserialize, Serialize};
13
14use crate::checkpoint::CheckpointAttempt;
15use crate::checkpoint::CheckpointWatermark;
16use crate::cluster::discovery::NodeId;
17#[cfg(feature = "cluster")]
18use crate::cluster::discovery::NodeInfo;
19#[cfg(feature = "cluster")]
20use tokio::sync::watch;
21
22/// KV key for the leader's barrier announcement.
23pub const ANNOUNCEMENT_KEY: &str = "control:barrier";
24
25/// KV key for a follower's barrier ack.
26pub const ACK_KEY: &str = "control:barrier-ack";
27
28/// Gossip KV key used by follower barrier servers to advertise their bound address.
29#[cfg(feature = "cluster")]
30pub const BARRIER_ADDR_KEY: &str = "barrier:addr";
31
32#[cfg(feature = "cluster")]
33const BARRIER_ENDPOINT_VERSION: u8 = 1;
34
35#[cfg(feature = "cluster")]
36const MAX_BARRIER_ENDPOINT_BYTES: usize = 1_024;
37
38#[derive(Default)]
39struct AnnouncementPublicationState {
40    initialized: bool,
41    latest: Option<BarrierAnnouncement>,
42}
43
44/// Cross-instance barrier coordination.
45pub struct BarrierCoordinator {
46    kv: Arc<dyn ClusterKv>,
47    /// Serializes local publication across every runtime mode. The latest admitted value advances
48    /// before cancellable I/O so an ambiguous write result cannot reopen an older phase.
49    publication: tokio::sync::Mutex<AnnouncementPublicationState>,
50    #[cfg(feature = "cluster")]
51    grpc: Arc<parking_lot::Mutex<Option<Arc<GrpcState>>>>,
52    /// First local observation time for each exact Prepare identity. Direct gRPC receipt is the
53    /// preferred clock, while this transport-independent registry gives the gossip fallback the
54    /// same non-refreshing attempt deadline across repeated observations.
55    #[cfg(feature = "cluster")]
56    prepare_observed_at: parking_lot::Mutex<FxHashMap<BarrierIdentity, std::time::Instant>>,
57    /// Highest checkpoint ID whose immutable terminal authority closes any retained Prepare at or
58    /// below it. Recovery may settle an attempt after its mutable transport hint becomes stale.
59    #[cfg(feature = "cluster")]
60    settled_prepare_floor: AtomicU64,
61    #[cfg(feature = "cluster")]
62    leader_election: Arc<parking_lot::Mutex<ActiveLeaderState>>,
63    #[cfg(feature = "cluster")]
64    leader_lease_store: Arc<parking_lot::Mutex<Option<Arc<super::LeaderLeaseStore>>>>,
65    #[cfg(feature = "cluster")]
66    local_leader_proof: Arc<parking_lot::Mutex<Option<LocalLeaderProofProvider>>>,
67    #[cfg(feature = "cluster")]
68    local_process: Arc<std::sync::OnceLock<BarrierProcessIdentity>>,
69    #[cfg(feature = "cluster")]
70    unbound_endpoint_started: parking_lot::Mutex<bool>,
71    #[cfg(feature = "cluster")]
72    process_lease_deadline: Arc<std::sync::OnceLock<Arc<super::LeaseDeadline>>>,
73}
74
75impl std::fmt::Debug for BarrierCoordinator {
76    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
77        f.debug_struct("BarrierCoordinator").finish_non_exhaustive()
78    }
79}
80
81impl Drop for BarrierCoordinator {
82    fn drop(&mut self) {
83        #[cfg(feature = "cluster")]
84        {
85            let grpc_opt = self.grpc.lock().take();
86            if let Some(state) = grpc_opt {
87                abort_grpc_tasks(&state);
88            }
89        }
90    }
91}
92
93mod grpc;
94mod kv;
95mod prepare;
96mod protocol;
97
98#[cfg(feature = "cluster")]
99pub(crate) use grpc::barrier_v1;
100pub use kv::{ClusterKv, InMemoryKv};
101pub use protocol::{BarrierAck, BarrierAckDisposition, BarrierAnnouncement, Phase, QuorumOutcome};
102
103#[cfg(feature = "cluster")]
104use grpc::{
105    abort_grpc_tasks, ack_disposition_from_wire, assignment_fence_to_wire,
106    checkpoint_watermark_from_wire, evict_barrier_client, get_barrier_client,
107    leader_proof_ack_matches, leader_proof_to_wire, send_local_phase_notification,
108    send_phase_notifications, ActiveLeaderState, BarrierClientPool, BarrierClientResolutionError,
109    GrpcBarrierServer, GrpcState, LocalLeaderProofProvider, WireAssignmentFence,
110};
111use prepare::PrepareFanoutBudget;
112#[cfg(feature = "cluster")]
113use prepare::{
114    canonical_expected_roster, clustered_phase_roster, install_prepare_fanout,
115    mark_capture_quorum_reached, preflight_prepare_fanout, prepare_fanout_budget,
116    prepare_fanout_plan, require_aligned_quorum, retire_prepare_fanout, BarrierFlavor,
117    BarrierIdentity, PeerFailure, PendingPrepareWaiter, PrepareAckState, PrepareFanoutBatch,
118    PrepareFanoutState, PrepareWaiterRegistration, MAX_PREPARE_WAITERS_PER_IDENTITY,
119    MAX_RETAINED_BARRIER_IDENTITIES, PREPARE_RPC_TIMEOUT,
120};
121#[cfg(feature = "cluster")]
122use protocol::{
123    decode_barrier_endpoint, encode_barrier_endpoint, merge_direct_announcement,
124    validate_wire_checkpoint_attempt, BarrierProcessIdentity, ExpectedBarrierProcess,
125    PHASE_RPC_TIMEOUT, PREPARE_RETRY_INITIAL_BACKOFF, PREPARE_RETRY_MAX_BACKOFF,
126};
127use protocol::{
128    is_terminal_phase, merge_observed_announcement, same_announcement_identity,
129    validate_ack_attempt, validate_announcement_attempt, validate_publication_order,
130    validate_scanned_announcements,
131};
132
133mod coordinator {
134    pub(super) mod authority;
135    pub(super) mod publication;
136    pub(super) mod quorum;
137}
138
139#[cfg(test)]
140mod tests;
laminar_core/cluster/control/barrier/mod.rs

laminar_core/cluster/control/barrier/
mod.rs
