reddb_server/replication/
commit_policy.rs

1//! Primary commit policies (PLAN.md Phase 11.4).
2//!
3//! `Local` — commit returns after the local WAL is durable (default;
4//! current behaviour). No replica involvement at commit time.
5//!
6//! `RemoteWal` — commit returns after the WAL segment containing the
7//! transaction has been archived to the remote backend. Bounds
8//! durability to "survives a single-node loss as long as the remote
9//! is reachable".
10//!
11//! `AckN(n)` — commit returns after `n` replicas have ack'd the
12//! transaction's LSN via `ack_replica_lsn`. `n=0` is equivalent to
13//! `Local`. The primary blocks the commit response until the count
14//! is met or `RED_REPLICATION_ACK_TIMEOUT_MS` elapses.
15//!
16//! `Quorum` — commit returns after the configured `QuorumConfig`
17//! reaches its durable commit watermark.
18//!
19//! `RemoteWal` is parsed for observability but does not block the
20//! write path yet. The default remains `Local`.
21
22#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
23pub enum CommitPolicy {
24    #[default]
25    Local,
26    RemoteWal,
27    AckN(u32),
28    Quorum,
29}
30
31impl CommitPolicy {
32    pub fn label(self) -> &'static str {
33        match self {
34            Self::Local => "local",
35            Self::RemoteWal => "remote_wal",
36            Self::AckN(_) => "ack_n",
37            Self::Quorum => "quorum",
38        }
39    }
40
41    pub fn detail_label(self) -> String {
42        match self {
43            Self::AckN(n) => format!("ack_n={n}"),
44            _ => self.label().to_string(),
45        }
46    }
47
48    pub fn parse_strict(raw: &str) -> Option<Self> {
49        let lower = raw.trim().to_ascii_lowercase();
50        if lower == "local" {
51            return Some(Self::Local);
52        }
53        if lower == "remote_wal" {
54            return Some(Self::RemoteWal);
55        }
56        if lower == "quorum" {
57            return Some(Self::Quorum);
58        }
59        if let Some(n_str) = lower.strip_prefix("ack_n=") {
60            return n_str.parse::<u32>().ok().map(Self::AckN);
61        }
62        None
63    }
64
65    /// Parse from `RED_PRIMARY_COMMIT_POLICY` env var. Accepts:
66    /// `local` (default), `remote_wal`, `ack_n=N` (decimal),
67    /// `quorum`. Unknown values fall back to `Local` with a warning.
68    pub fn from_env() -> Self {
69        match std::env::var("RED_PRIMARY_COMMIT_POLICY").ok() {
70            Some(raw) => Self::parse(raw.trim()),
71            None => Self::Local,
72        }
73    }
74
75    pub fn parse(raw: &str) -> Self {
76        let lower = raw.to_ascii_lowercase();
77        if lower == "local" || lower.is_empty() {
78            return Self::Local;
79        }
80        if lower == "remote_wal" {
81            return Self::RemoteWal;
82        }
83        if lower == "quorum" {
84            return Self::Quorum;
85        }
86        if let Some(n_str) = lower.strip_prefix("ack_n=") {
87            if let Ok(n) = n_str.parse::<u32>() {
88                return Self::AckN(n);
89            }
90        }
91        tracing::warn!(
92            target: "reddb::replication::commit_policy",
93            value = %raw,
94            "unknown RED_PRIMARY_COMMIT_POLICY; falling back to local"
95        );
96        Self::Local
97    }
98}
99
100#[cfg(test)]
101mod tests {
102    use super::*;
103
104    #[test]
105    fn default_is_local() {
106        assert_eq!(CommitPolicy::default(), CommitPolicy::Local);
107    }
108
109    #[test]
110    fn parse_known_values() {
111        assert_eq!(CommitPolicy::parse("local"), CommitPolicy::Local);
112        assert_eq!(CommitPolicy::parse("LOCAL"), CommitPolicy::Local);
113        assert_eq!(CommitPolicy::parse("remote_wal"), CommitPolicy::RemoteWal);
114        assert_eq!(CommitPolicy::parse("quorum"), CommitPolicy::Quorum);
115        assert_eq!(CommitPolicy::parse("ack_n=3"), CommitPolicy::AckN(3));
116        assert_eq!(CommitPolicy::parse("ack_n=0"), CommitPolicy::AckN(0));
117    }
118
119    #[test]
120    fn parse_unknown_falls_back_to_local() {
121        assert_eq!(CommitPolicy::parse("nonsense"), CommitPolicy::Local);
122        assert_eq!(CommitPolicy::parse("ack_n=abc"), CommitPolicy::Local);
123        assert_eq!(CommitPolicy::parse(""), CommitPolicy::Local);
124    }
125
126    #[test]
127    fn label_round_trips_known_values() {
128        assert_eq!(CommitPolicy::Local.label(), "local");
129        assert_eq!(CommitPolicy::RemoteWal.label(), "remote_wal");
130        assert_eq!(CommitPolicy::AckN(5).label(), "ack_n");
131        assert_eq!(CommitPolicy::Quorum.label(), "quorum");
132    }
133
134    #[test]
135    fn strict_parse_rejects_unknown_values() {
136        assert_eq!(
137            CommitPolicy::parse_strict("ack_n=2"),
138            Some(CommitPolicy::AckN(2))
139        );
140        assert_eq!(CommitPolicy::parse_strict("nonsense"), None);
141        assert_eq!(CommitPolicy::parse_strict(""), None);
142    }
143}
reddb_server/replication/commit_policy.rs

reddb_server/replication/
commit_policy.rs
