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` has no durable publication waiter yet and is rejected before
20//! mutation. Synchronous timeouts report an uncertain replica outcome after
21//! local commit; `RED_COMMIT_FAIL_ON_TIMEOUT` no longer weakens the policy.
22
23#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
24pub enum CommitPolicy {
25    #[default]
26    Local,
27    RemoteWal,
28    AckN(u32),
29    Quorum,
30}
31
32impl CommitPolicy {
33    pub fn from_env_strict() -> Result<Self, String> {
34        match std::env::var("RED_PRIMARY_COMMIT_POLICY") {
35            Ok(raw) => Self::parse_strict(&raw)
36                .ok_or_else(|| format!("invalid RED_PRIMARY_COMMIT_POLICY: {raw}")),
37            Err(std::env::VarError::NotPresent) => Ok(Self::Local),
38            Err(_) => Err("RED_PRIMARY_COMMIT_POLICY must be valid UTF-8".into()),
39        }
40    }
41
42    pub fn label(self) -> &'static str {
43        match self {
44            Self::Local => "local",
45            Self::RemoteWal => "remote_wal",
46            Self::AckN(_) => "ack_n",
47            Self::Quorum => "quorum",
48        }
49    }
50
51    pub fn detail_label(self) -> String {
52        match self {
53            Self::AckN(n) => format!("ack_n={n}"),
54            _ => self.label().to_string(),
55        }
56    }
57
58    pub fn parse_strict(raw: &str) -> Option<Self> {
59        let lower = raw.trim().to_ascii_lowercase();
60        if lower == "local" {
61            return Some(Self::Local);
62        }
63        if lower == "remote_wal" {
64            return Some(Self::RemoteWal);
65        }
66        if lower == "quorum" {
67            return Some(Self::Quorum);
68        }
69        if let Some(n_str) = lower.strip_prefix("ack_n=") {
70            return n_str.parse::<u32>().ok().map(Self::AckN);
71        }
72        None
73    }
74
75    /// Parse from `RED_PRIMARY_COMMIT_POLICY` env var. Accepts:
76    /// `local` (default), `remote_wal`, `ack_n=N` (decimal),
77    /// `quorum`. Unknown values fall back to `Local` with a warning.
78    pub fn from_env() -> Self {
79        match std::env::var("RED_PRIMARY_COMMIT_POLICY").ok() {
80            Some(raw) => Self::parse(raw.trim()),
81            None => Self::Local,
82        }
83    }
84
85    pub fn parse(raw: &str) -> Self {
86        let lower = raw.to_ascii_lowercase();
87        if lower == "local" || lower.is_empty() {
88            return Self::Local;
89        }
90        if lower == "remote_wal" {
91            return Self::RemoteWal;
92        }
93        if lower == "quorum" {
94            return Self::Quorum;
95        }
96        if let Some(n_str) = lower.strip_prefix("ack_n=") {
97            if let Ok(n) = n_str.parse::<u32>() {
98                return Self::AckN(n);
99            }
100        }
101        tracing::warn!(
102            target: "reddb::replication::commit_policy",
103            value = %raw,
104            "unknown RED_PRIMARY_COMMIT_POLICY; falling back to local"
105        );
106        Self::Local
107    }
108}
109
110#[cfg(test)]
111mod tests {
112    use super::*;
113
114    #[test]
115    fn default_is_local() {
116        assert_eq!(CommitPolicy::default(), CommitPolicy::Local);
117    }
118
119    #[test]
120    fn parse_known_values() {
121        assert_eq!(CommitPolicy::parse("local"), CommitPolicy::Local);
122        assert_eq!(CommitPolicy::parse("LOCAL"), CommitPolicy::Local);
123        assert_eq!(CommitPolicy::parse("remote_wal"), CommitPolicy::RemoteWal);
124        assert_eq!(CommitPolicy::parse("quorum"), CommitPolicy::Quorum);
125        assert_eq!(CommitPolicy::parse("ack_n=3"), CommitPolicy::AckN(3));
126        assert_eq!(CommitPolicy::parse("ack_n=0"), CommitPolicy::AckN(0));
127    }
128
129    #[test]
130    fn parse_unknown_falls_back_to_local() {
131        assert_eq!(CommitPolicy::parse("nonsense"), CommitPolicy::Local);
132        assert_eq!(CommitPolicy::parse("ack_n=abc"), CommitPolicy::Local);
133        assert_eq!(CommitPolicy::parse(""), CommitPolicy::Local);
134    }
135
136    #[test]
137    fn label_round_trips_known_values() {
138        assert_eq!(CommitPolicy::Local.label(), "local");
139        assert_eq!(CommitPolicy::RemoteWal.label(), "remote_wal");
140        assert_eq!(CommitPolicy::AckN(5).label(), "ack_n");
141        assert_eq!(CommitPolicy::Quorum.label(), "quorum");
142    }
143
144    #[test]
145    fn strict_parse_rejects_unknown_values() {
146        assert_eq!(
147            CommitPolicy::parse_strict("ack_n=2"),
148            Some(CommitPolicy::AckN(2))
149        );
150        assert_eq!(CommitPolicy::parse_strict("nonsense"), None);
151        assert_eq!(CommitPolicy::parse_strict(""), None);
152    }
153}
reddb_server/replication/commit_policy.rs

reddb_server/replication/
commit_policy.rs
