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 swap_db;
43pub mod topology_advertiser;
44pub mod witness;
45
46pub use bookmark::{BookmarkDecodeError, CausalBookmark};
47pub use cascade::{
48    plan_upstream, CascadeRefusal, CascadeRelay, CascadeUpstream, DownstreamSlot, ReplicaClass,
49    UpstreamChoice,
50};
51pub use commit_policy::CommitPolicy;
52pub use commit_waiter::{AwaitOutcome, CommitWaiter};
53pub use control_plane::{
54    ControlPlaneConsensus, ControlPlaneEntry, ControlPlaneEntryKind, ControlPlaneLogIndex,
55    ControlPlanePayload, ControlPlaneRole, MemberId, ProposeRefusal,
56};
57pub use election::{
58    quorum_threshold, randomized_election_timeout, ElectionCoordinator, ElectionOutcome,
59    ElectionRequest, ElectionTransport, FileLastVoteStore, LastVote, LastVoteError, LastVoteStore,
60    Member, MemberKind, MemoryLastVoteStore, RefusalReason, VoteDecision, VoteRequest, Voter,
61    VotingState,
62};
63pub use failover::{
64    FailoverCoordinator, FailoverError, FailoverMode, FailoverNode, FailoverOutcome,
65    FailoverRequest, FailoverTransport, NodeRole, RoleAssignment,
66};
67pub use fence::{
68    term_is_stale, FenceBoundary, FenceVerdict, FileTermStore, MemoryTermStore, StaleTermFenced,
69    StaleTermRejection, StreamHandshake, TermFence, TermStore, TermStoreError,
70};
71pub use flow_control::{Admission, FlowController};
72pub use lease::{LeaseError, LeaseStore, WriterLease};
73pub use quorum::{QuorumConfig, QuorumCoordinator, QuorumError};
74pub use rollback::{
75    DivergentTail, RollbackCoordinator, RollbackError, RollbackEvent, RollbackOutcome,
76    RollbackPlan, RollbackRequest, RollbackTransport, TailRecord,
77};
78pub use swap_db::{RebootstrapInProgress, SwapDb};
79pub use topology_advertiser::{
80    LagConfig, TopologyAdvertiser, TopologyAuthGate, DEFAULT_REPLICA_TIMEOUT_MS,
81    TOPOLOGY_READ_CAPABILITY,
82};
83pub use witness::{RuntimeProfile, WitnessSupervisor};
84
85pub const DEFAULT_REPLICATION_TERM: u64 = reddb_wire::replication::DEFAULT_REPLICATION_TERM;
86pub const DEFAULT_SLOT_RETENTION_MAX_LAG_LSN: u64 = 100_000;
87pub const DEFAULT_SLOT_IDLE_TIMEOUT_MS: u64 = 86_400_000;
88
89/// Role of this RedDB instance in a replication cluster.
90#[derive(Debug, Clone, PartialEq, Eq, Default)]
91pub enum ReplicationRole {
92    /// Standalone instance (default, no replication).
93    #[default]
94    Standalone,
95    /// Primary: accepts reads and writes, streams WAL to replicas.
96    Primary,
97    /// Replica: read-only, receives WAL from primary.
98    Replica {
99        /// gRPC address of the primary (e.g., "http://primary:55055")
100        primary_addr: String,
101    },
102}
103
104/// Configuration for replication.
105#[derive(Debug, Clone)]
106pub struct ReplicationConfig {
107    pub role: ReplicationRole,
108    /// Current replication term/epoch stamped into WAL-derived records.
109    pub term: u64,
110    /// How often replica polls for new WAL records (milliseconds).
111    pub poll_interval_ms: u64,
112    /// Maximum batch size for WAL record transfer.
113    pub max_batch_size: usize,
114    /// Region identifier for this instance (Phase 2.6 multi-region).
115    ///
116    /// Used by the quorum coordinator to spread write acks across
117    /// fault domains: `QuorumConfig::required_regions` forces a commit
118    /// to wait until at least one replica in each listed region has
119    /// acked. Defaults to `"local"` for single-region deployments.
120    pub region: String,
121    /// Quorum configuration (Phase 2.6 multi-region).
122    pub quorum: QuorumConfig,
123    /// Maximum LSN lag a replication slot may pin before the primary
124    /// invalidates it and allows WAL pruning to continue.
125    pub slot_retention_max_lag_lsn: u64,
126    /// Maximum wall-clock idle age for a slot before invalidation.
127    pub slot_idle_timeout_ms: u64,
128    /// Streaming class (issue #838). A [`ReplicaClass::Voting`] node is on the
129    /// durability/election path and always streams directly from the primary;
130    /// a [`ReplicaClass::AsyncReadReplica`] may cascade from an intermediate.
131    /// Defaults to `Voting` — a node only cascades when explicitly declared a
132    /// read-replica.
133    pub replica_class: ReplicaClass,
134    /// Optional intermediate replica to cascade from (issue #838). Honoured
135    /// only for an async read-replica; a voting member refuses it and streams
136    /// directly from the primary. See [`ReplicationConfig::resolved_upstream`].
137    pub cascade_from: Option<CascadeUpstream>,
138}
139
140impl ReplicationConfig {
141    pub fn standalone() -> Self {
142        Self {
143            role: ReplicationRole::Standalone,
144            term: DEFAULT_REPLICATION_TERM,
145            poll_interval_ms: 100,
146            max_batch_size: 1000,
147            region: "local".to_string(),
148            quorum: QuorumConfig::async_commit(),
149            slot_retention_max_lag_lsn: DEFAULT_SLOT_RETENTION_MAX_LAG_LSN,
150            slot_idle_timeout_ms: DEFAULT_SLOT_IDLE_TIMEOUT_MS,
151            replica_class: ReplicaClass::Voting,
152            cascade_from: None,
153        }
154    }
155
156    pub fn primary() -> Self {
157        Self {
158            role: ReplicationRole::Primary,
159            term: DEFAULT_REPLICATION_TERM,
160            poll_interval_ms: 100,
161            max_batch_size: 1000,
162            region: "local".to_string(),
163            quorum: QuorumConfig::async_commit(),
164            slot_retention_max_lag_lsn: DEFAULT_SLOT_RETENTION_MAX_LAG_LSN,
165            slot_idle_timeout_ms: DEFAULT_SLOT_IDLE_TIMEOUT_MS,
166            replica_class: ReplicaClass::Voting,
167            cascade_from: None,
168        }
169    }
170
171    pub fn replica(primary_addr: impl Into<String>) -> Self {
172        Self {
173            role: ReplicationRole::Replica {
174                primary_addr: primary_addr.into(),
175            },
176            term: DEFAULT_REPLICATION_TERM,
177            poll_interval_ms: 100,
178            max_batch_size: 1000,
179            region: "local".to_string(),
180            quorum: QuorumConfig::async_commit(),
181            slot_retention_max_lag_lsn: DEFAULT_SLOT_RETENTION_MAX_LAG_LSN,
182            slot_idle_timeout_ms: DEFAULT_SLOT_IDLE_TIMEOUT_MS,
183            replica_class: ReplicaClass::Voting,
184            cascade_from: None,
185        }
186    }
187
188    /// Attach a quorum configuration (fluent setter).
189    pub fn with_quorum(mut self, quorum: QuorumConfig) -> Self {
190        self.quorum = quorum;
191        self
192    }
193
194    /// Set the region identifier (fluent setter).
195    pub fn with_region(mut self, region: impl Into<String>) -> Self {
196        self.region = region.into();
197        self
198    }
199
200    /// Set the replication term stamped into newly produced records.
201    pub fn with_term(mut self, term: u64) -> Self {
202        self.term = term;
203        self
204    }
205
206    pub fn with_slot_retention_max_lag_lsn(mut self, max_lag_lsn: u64) -> Self {
207        self.slot_retention_max_lag_lsn = max_lag_lsn;
208        self
209    }
210
211    pub fn with_slot_idle_timeout_ms(mut self, timeout_ms: u64) -> Self {
212        self.slot_idle_timeout_ms = timeout_ms;
213        self
214    }
215
216    /// Set the streaming class explicitly (issue #838).
217    pub fn with_replica_class(mut self, class: ReplicaClass) -> Self {
218        self.replica_class = class;
219        self
220    }
221
222    /// Declare this node an async read-replica that cascades from `intermediate`
223    /// (issue #838). Sets [`ReplicaClass::AsyncReadReplica`] and the cascade
224    /// source together — the only combination that actually streams from an
225    /// intermediate. A node left at the default `Voting` class ignores any
226    /// cascade source and streams directly from the primary.
227    pub fn cascading_from(mut self, node_id: impl Into<String>, addr: impl Into<String>) -> Self {
228        self.replica_class = ReplicaClass::AsyncReadReplica;
229        self.cascade_from = Some(CascadeUpstream::new(node_id, addr));
230        self
231    }
232
233    /// Resolve where this node should open its WAL stream, applying the
234    /// cascade policy (issue #838). A voting member always resolves to the
235    /// primary even when a cascade source is configured; the returned
236    /// [`CascadeRefusal`] explains any fallback so it is observable rather
237    /// than silent. `self_node_id` guards against a node cascading from its
238    /// own slot.
239    pub fn resolved_upstream(
240        &self,
241        self_node_id: &str,
242    ) -> (UpstreamChoice, Option<CascadeRefusal>) {
243        plan_upstream(self_node_id, self.replica_class, self.cascade_from.as_ref())
244    }
245}
reddb_server/replication/mod.rs

reddb_server/replication/
mod.rs
