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

reddb_server/replication/
mod.rs
