reddb_server/replication/
mod.rs

1//! Replication Module
2//!
3//! Implements single-primary, multi-replica replication via WAL streaming.
4//!
5//! # Architecture
6//!
7//! - Primary: accepts writes and streams WAL records to replicas
8//! - Replica: read-only, connects to primary for WAL streaming
9//! - Initial sync via snapshot transfer, then incremental WAL
10//!
11//! # Usage
12//!
13//! ```ignore
14//! // Primary
15//! let options = RedDBOptions::persistent("./primary-data")
16//!     .with_replication(ReplicationConfig::primary());
17//!
18//! // Replica
19//! let options = RedDBOptions::persistent("./replica-data")
20//!     .with_replication(ReplicationConfig::replica("http://primary:55055"));
21//! ```
22
23pub mod bookmark;
24pub mod cascade;
25pub mod cdc;
26pub mod commit_policy;
27pub mod commit_waiter;
28pub mod control_plane;
29pub mod dst;
30pub mod election;
31pub mod failover;
32pub mod fence;
33pub mod flow_control;
34pub mod lease;
35pub mod logical;
36pub(crate) mod metadata;
37pub mod primary;
38pub mod quorum;
39pub mod reconnect;
40pub mod replica;
41pub(crate) mod request_auth;
42pub mod rollback;
43pub mod scheduler;
44pub mod signal_plane;
45pub mod swap_db;
46pub mod topology_advertiser;
47pub mod witness;
48
49pub use bookmark::{BookmarkDecodeError, CausalBookmark};
50pub use cascade::{
51    plan_upstream, CascadeRefusal, CascadeRelay, CascadeUpstream, DownstreamSlot, ReplicaClass,
52    UpstreamChoice,
53};
54pub use commit_policy::CommitPolicy;
55pub use commit_waiter::{AwaitOutcome, CommitWaiter};
56pub use control_plane::{
57    ControlPlaneConsensus, ControlPlaneEntry, ControlPlaneEntryKind, ControlPlaneLogIndex,
58    ControlPlanePayload, ControlPlaneRole, MemberId, ProposeRefusal,
59};
60pub use election::{
61    quorum_threshold, randomized_election_timeout, ElectionCoordinator, ElectionOutcome,
62    ElectionRequest, ElectionTransport, FileLastVoteStore, LastVote, LastVoteError, LastVoteStore,
63    Member, MemberKind, MemoryLastVoteStore, RefusalReason, VoteDecision, VoteRequest, Voter,
64    VotingState,
65};
66pub use failover::{
67    FailoverCoordinator, FailoverError, FailoverMode, FailoverNode, FailoverOutcome,
68    FailoverRequest, FailoverTransport, NodeRole, RoleAssignment,
69};
70pub use fence::{
71    term_is_stale, FenceBoundary, FenceVerdict, FileTermStore, MemoryTermStore, StaleTermFenced,
72    StaleTermRejection, StreamHandshake, TermFence, TermStore, TermStoreError,
73};
74pub use flow_control::{Admission, FlowController};
75pub use lease::{LeaseError, LeaseStore, WriterLease};
76pub use quorum::{QuorumConfig, QuorumCoordinator, QuorumError};
77pub use rollback::{
78    DivergentTail, RollbackCoordinator, RollbackError, RollbackEvent, RollbackOutcome,
79    RollbackPlan, RollbackRequest, RollbackTransport, TailRecord,
80};
81pub use signal_plane::{
82    CatalogVersionHint, IntraClusterSignalBus, LivenessObservation, LivenessStatus, LoadBucket,
83    LoadMetricSample, MemberHealthInput, ReceivedSignal, SharedSignalTransport, SignalFrame,
84    SignalPlane, SignalPlaneLimits, SignalPlaneMessage, SignalPlaneMetrics, SignalPlaneSchedule,
85    SignalTransportError, SimulatedSignalPlane, TopologyHint, TransportSignalPlane,
86};
87pub use swap_db::{RebootstrapInProgress, SwapDb};
88pub use topology_advertiser::{
89    LagConfig, TopologyAdvertiser, TopologyAuthGate, DEFAULT_REPLICA_TIMEOUT_MS,
90    TOPOLOGY_READ_CAPABILITY,
91};
92pub use witness::{RuntimeProfile, WitnessSupervisor};
93
94pub const DEFAULT_REPLICATION_TERM: u64 = reddb_wire::replication::DEFAULT_REPLICATION_TERM;
95pub const DEFAULT_SLOT_RETENTION_MAX_LAG_LSN: u64 = 100_000;
96pub const DEFAULT_SLOT_IDLE_TIMEOUT_MS: u64 = 86_400_000;
97pub const SHIPPED_FAILOVER_PROFILES: [FailoverProfile; 3] = [
98    FailoverProfile::CONSERVATIVE,
99    FailoverProfile::BALANCED,
100    FailoverProfile::AGGRESSIVE,
101];
102
103/// Named failover timing posture.
104#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
105pub enum FailoverProfileName {
106    Conservative,
107    Balanced,
108    Aggressive,
109    Custom,
110}
111
112impl FailoverProfileName {
113    pub const fn as_str(self) -> &'static str {
114        match self {
115            Self::Conservative => "conservative",
116            Self::Balanced => "balanced",
117            Self::Aggressive => "aggressive",
118            Self::Custom => "custom",
119        }
120    }
121}
122
123/// Tuned failover constants over the existing lease, health, and grace
124/// mechanisms.
125///
126/// Shipped profiles document the intended trade-off:
127/// - `conservative`: long lease and high health threshold; favors avoiding
128///   false promotion over fast recovery.
129/// - `balanced`: default posture; keeps a moderate lease and health threshold
130///   for ordinary multi-node deployments.
131/// - `aggressive`: short lease and lower health threshold; favors fast recovery
132///   when the operator accepts a narrower timing margin.
133#[derive(Debug, Clone, Copy, PartialEq, Eq)]
134pub struct FailoverProfile {
135    pub name: FailoverProfileName,
136    pub lease_window_ms: u64,
137    pub member_health_score_threshold: u8,
138    pub promotion_grace_ms: u64,
139    pub max_clock_drift_ms: u64,
140}
141
142impl FailoverProfile {
143    pub const CONSERVATIVE: Self = Self {
144        name: FailoverProfileName::Conservative,
145        lease_window_ms: 60_000,
146        member_health_score_threshold: 90,
147        promotion_grace_ms: 30_000,
148        max_clock_drift_ms: 5_000,
149    };
150
151    pub const BALANCED: Self = Self {
152        name: FailoverProfileName::Balanced,
153        lease_window_ms: 30_000,
154        member_health_score_threshold: 75,
155        promotion_grace_ms: 10_000,
156        max_clock_drift_ms: 5_000,
157    };
158
159    pub const AGGRESSIVE: Self = Self {
160        name: FailoverProfileName::Aggressive,
161        lease_window_ms: 15_000,
162        member_health_score_threshold: 60,
163        promotion_grace_ms: 5_000,
164        max_clock_drift_ms: 2_000,
165    };
166
167    pub const fn shipped() -> &'static [Self] {
168        &SHIPPED_FAILOVER_PROFILES
169    }
170
171    pub const fn name_str(self) -> &'static str {
172        self.name.as_str()
173    }
174
175    pub fn validate(self) -> Result<Self, String> {
176        if self.lease_window_ms == 0 {
177            return Err("failover profile lease_window_ms must be positive".to_string());
178        }
179        if self.member_health_score_threshold > 100 {
180            return Err(
181                "failover profile member_health_score_threshold must be <= 100".to_string(),
182            );
183        }
184        if self.promotion_grace_ms >= self.lease_window_ms {
185            return Err(
186                "failover profile promotion_grace_ms must be smaller than lease_window_ms"
187                    .to_string(),
188            );
189        }
190        let safety_margin_ms = self.lease_window_ms - self.promotion_grace_ms;
191        if safety_margin_ms < self.max_clock_drift_ms {
192            return Err(format!(
193                "failover profile lease safety margin {safety_margin_ms}ms is below configured max clock drift {}ms",
194                self.max_clock_drift_ms
195            ));
196        }
197        Ok(self)
198    }
199
200    pub const fn lease_safety_margin_ms(self) -> u64 {
201        self.lease_window_ms - self.promotion_grace_ms
202    }
203}
204
205#[cfg(test)]
206mod tests {
207    use super::*;
208
209    #[test]
210    fn shipped_failover_profiles_satisfy_lease_safety_validation() {
211        for profile in FailoverProfile::shipped() {
212            profile.validate().expect(profile.name_str());
213            assert!(
214                profile.lease_safety_margin_ms() >= profile.max_clock_drift_ms,
215                "{} profile violates lease clock-discipline margin",
216                profile.name_str()
217            );
218        }
219    }
220
221    #[test]
222    fn failover_profile_rejects_margin_below_configured_clock_drift() {
223        let invalid = FailoverProfile {
224            name: FailoverProfileName::Custom,
225            lease_window_ms: 1_000,
226            member_health_score_threshold: 80,
227            promotion_grace_ms: 900,
228            max_clock_drift_ms: 200,
229        };
230
231        let err = invalid.validate().expect_err("profile must be rejected");
232
233        assert!(err.contains("lease safety margin"), "{err}");
234        assert!(err.contains("max clock drift"), "{err}");
235    }
236
237    #[test]
238    fn runtime_open_rejects_invalid_failover_profile() {
239        let invalid = FailoverProfile {
240            name: FailoverProfileName::Custom,
241            lease_window_ms: 1_000,
242            member_health_score_threshold: 80,
243            promotion_grace_ms: 900,
244            max_clock_drift_ms: 200,
245        };
246        let options = crate::RedDBOptions::in_memory()
247            .with_replication(ReplicationConfig::primary().with_failover_profile(invalid));
248
249        let err = match crate::RedDBRuntime::with_options(options) {
250            Ok(_) => panic!("boot must reject config"),
251            Err(err) => err,
252        };
253
254        assert!(err.to_string().contains("invalid config"), "{err}");
255        assert!(err.to_string().contains("lease safety margin"), "{err}");
256    }
257
258    #[test]
259    fn changing_failover_profile_updates_active_profile() {
260        let mut config = ReplicationConfig::primary();
261
262        config
263            .change_failover_profile(FailoverProfile::AGGRESSIVE, "test")
264            .expect("valid profile applies");
265
266        assert_eq!(
267            config.failover_profile.name,
268            FailoverProfileName::Aggressive
269        );
270    }
271}
272
273/// Role of this RedDB instance in a replication cluster.
274#[derive(Debug, Clone, PartialEq, Eq, Default)]
275pub enum ReplicationRole {
276    /// Standalone instance (default, no replication).
277    #[default]
278    Standalone,
279    /// Primary: accepts reads and writes, streams WAL to replicas.
280    Primary,
281    /// Replica: read-only, receives WAL from primary.
282    Replica {
283        /// gRPC address of the primary (e.g., "http://primary:55055")
284        primary_addr: String,
285    },
286}
287
288/// Configuration for replication.
289#[derive(Debug, Clone)]
290pub struct ReplicationConfig {
291    pub role: ReplicationRole,
292    /// Current replication term/epoch stamped into WAL-derived records.
293    pub term: u64,
294    /// How often replica polls for new WAL records (milliseconds).
295    pub poll_interval_ms: u64,
296    /// Maximum batch size for WAL record transfer.
297    pub max_batch_size: usize,
298    /// Region identifier for this instance (Phase 2.6 multi-region).
299    ///
300    /// Used by the quorum coordinator to spread write acks across
301    /// fault domains: `QuorumConfig::required_regions` forces a commit
302    /// to wait until at least one replica in each listed region has
303    /// acked. Defaults to `"local"` for single-region deployments.
304    pub region: String,
305    /// Quorum configuration (Phase 2.6 multi-region).
306    pub quorum: QuorumConfig,
307    /// Maximum LSN lag a replication slot may pin before the primary
308    /// invalidates it and allows WAL pruning to continue.
309    pub slot_retention_max_lag_lsn: u64,
310    /// Maximum wall-clock idle age for a slot before invalidation.
311    pub slot_idle_timeout_ms: u64,
312    /// Streaming class (issue #838). A [`ReplicaClass::Voting`] node is on the
313    /// durability/election path and always streams directly from the primary;
314    /// a [`ReplicaClass::AsyncReadReplica`] may cascade from an intermediate.
315    /// Defaults to `Voting` — a node only cascades when explicitly declared a
316    /// read-replica.
317    pub replica_class: ReplicaClass,
318    /// Optional intermediate replica to cascade from (issue #838). Honoured
319    /// only for an async read-replica; a voting member refuses it and streams
320    /// directly from the primary. See [`ReplicationConfig::resolved_upstream`].
321    pub cascade_from: Option<CascadeUpstream>,
322    /// Active named failover profile. The profile is a bundle of tuned
323    /// lease/health/grace constants; it does not add new failover machinery.
324    pub failover_profile: FailoverProfile,
325}
326
327impl ReplicationConfig {
328    pub fn standalone() -> Self {
329        Self {
330            role: ReplicationRole::Standalone,
331            term: DEFAULT_REPLICATION_TERM,
332            poll_interval_ms: 100,
333            max_batch_size: 1000,
334            region: "local".to_string(),
335            quorum: QuorumConfig::async_commit(),
336            slot_retention_max_lag_lsn: DEFAULT_SLOT_RETENTION_MAX_LAG_LSN,
337            slot_idle_timeout_ms: DEFAULT_SLOT_IDLE_TIMEOUT_MS,
338            replica_class: ReplicaClass::Voting,
339            cascade_from: None,
340            failover_profile: FailoverProfile::BALANCED,
341        }
342    }
343
344    pub fn primary() -> Self {
345        Self {
346            role: ReplicationRole::Primary,
347            term: DEFAULT_REPLICATION_TERM,
348            poll_interval_ms: 100,
349            max_batch_size: 1000,
350            region: "local".to_string(),
351            quorum: QuorumConfig::async_commit(),
352            slot_retention_max_lag_lsn: DEFAULT_SLOT_RETENTION_MAX_LAG_LSN,
353            slot_idle_timeout_ms: DEFAULT_SLOT_IDLE_TIMEOUT_MS,
354            replica_class: ReplicaClass::Voting,
355            cascade_from: None,
356            failover_profile: FailoverProfile::BALANCED,
357        }
358    }
359
360    pub fn replica(primary_addr: impl Into<String>) -> Self {
361        Self {
362            role: ReplicationRole::Replica {
363                primary_addr: primary_addr.into(),
364            },
365            term: DEFAULT_REPLICATION_TERM,
366            poll_interval_ms: 100,
367            max_batch_size: 1000,
368            region: "local".to_string(),
369            quorum: QuorumConfig::async_commit(),
370            slot_retention_max_lag_lsn: DEFAULT_SLOT_RETENTION_MAX_LAG_LSN,
371            slot_idle_timeout_ms: DEFAULT_SLOT_IDLE_TIMEOUT_MS,
372            replica_class: ReplicaClass::Voting,
373            cascade_from: None,
374            failover_profile: FailoverProfile::BALANCED,
375        }
376    }
377
378    /// Attach a quorum configuration (fluent setter).
379    pub fn with_quorum(mut self, quorum: QuorumConfig) -> Self {
380        self.quorum = quorum;
381        self
382    }
383
384    /// Set the region identifier (fluent setter).
385    pub fn with_region(mut self, region: impl Into<String>) -> Self {
386        self.region = region.into();
387        self
388    }
389
390    /// Set the replication term stamped into newly produced records.
391    pub fn with_term(mut self, term: u64) -> Self {
392        self.term = term;
393        self
394    }
395
396    pub fn with_slot_retention_max_lag_lsn(mut self, max_lag_lsn: u64) -> Self {
397        self.slot_retention_max_lag_lsn = max_lag_lsn;
398        self
399    }
400
401    pub fn with_slot_idle_timeout_ms(mut self, timeout_ms: u64) -> Self {
402        self.slot_idle_timeout_ms = timeout_ms;
403        self
404    }
405
406    pub fn with_failover_profile(mut self, profile: FailoverProfile) -> Self {
407        self.failover_profile = profile;
408        self
409    }
410
411    pub fn validate_failover_profile(&self) -> Result<(), String> {
412        self.failover_profile.validate().map(|_| ())
413    }
414
415    pub fn change_failover_profile(
416        &mut self,
417        profile: FailoverProfile,
418        changed_by: impl Into<String>,
419    ) -> Result<(), String> {
420        let profile = profile.validate()?;
421        let old_profile = self.failover_profile.name_str().to_string();
422        self.failover_profile = profile;
423        crate::telemetry::operator_event::OperatorEvent::FailoverProfileChanged {
424            old_profile,
425            new_profile: profile.name_str().to_string(),
426            changed_by: changed_by.into(),
427        }
428        .emit_global();
429        Ok(())
430    }
431
432    /// Set the streaming class explicitly (issue #838).
433    pub fn with_replica_class(mut self, class: ReplicaClass) -> Self {
434        self.replica_class = class;
435        self
436    }
437
438    /// Declare this node an async read-replica that cascades from `intermediate`
439    /// (issue #838). Sets [`ReplicaClass::AsyncReadReplica`] and the cascade
440    /// source together — the only combination that actually streams from an
441    /// intermediate. A node left at the default `Voting` class ignores any
442    /// cascade source and streams directly from the primary.
443    pub fn cascading_from(mut self, node_id: impl Into<String>, addr: impl Into<String>) -> Self {
444        self.replica_class = ReplicaClass::AsyncReadReplica;
445        self.cascade_from = Some(CascadeUpstream::new(node_id, addr));
446        self
447    }
448
449    /// Resolve where this node should open its WAL stream, applying the
450    /// cascade policy (issue #838). A voting member always resolves to the
451    /// primary even when a cascade source is configured; the returned
452    /// [`CascadeRefusal`] explains any fallback so it is observable rather
453    /// than silent. `self_node_id` guards against a node cascading from its
454    /// own slot.
455    pub fn resolved_upstream(
456        &self,
457        self_node_id: &str,
458    ) -> (UpstreamChoice, Option<CascadeRefusal>) {
459        plan_upstream(self_node_id, self.replica_class, self.cascade_from.as_ref())
460    }
461}
reddb_server/replication/mod.rs

reddb_server/replication/
mod.rs
