mocra_cluster/
control.rs

1//! Control plane API: strongly consistent KV + distributed locks on top of the state machine.
2//!
3//! [`LocalControlPlane`] is the **single-node** implementation (commands are applied directly to
4//! the local state machine, with no consensus). In a distributed deployment it is replaced by
5//! `RaftControlPlane` — commands are first replicated to a majority through Raft and only then
6//! applied (a follow-up increment).
7//!
8//! This trait's shape mirrors the main crate's `CoordinationBackend`; the adapter between the two
9//! lives on the `mocra` side, so the embedded control plane can be used for coordination.
10
11use std::path::Path;
12use std::sync::Arc;
13use std::time::{SystemTime, UNIX_EPOCH};
14
15use async_trait::async_trait;
16
17use crate::cmd::{Cmd, CmdResult};
18use crate::state_machine::{StateMachine, StateMachineError};
19
20/// Control plane errors: state machine errors plus consensus / configuration errors.
21#[derive(Debug, thiserror::Error)]
22pub enum ControlError {
23    #[error(transparent)]
24    StateMachine(#[from] StateMachineError),
25    #[error("raft: {0}")]
26    Raft(String),
27    #[error("config: {0}")]
28    Config(String),
29}
30
31/// Control plane: strongly consistent KV + distributed locks (+ membership/ownership later).
32#[async_trait]
33pub trait ControlPlane: Send + Sync {
34    async fn set(&self, key: &[u8], value: &[u8]) -> Result<(), ControlError>;
35    async fn get(&self, key: &[u8]) -> Result<Option<Vec<u8>>, ControlError>;
36    async fn delete(&self, key: &[u8]) -> Result<(), ControlError>;
37    /// Compare-and-swap; returns whether it succeeded.
38    async fn cas(
39        &self,
40        key: &[u8],
41        expect: Option<&[u8]>,
42        value: &[u8],
43    ) -> Result<bool, ControlError>;
44    /// Acquire a lock; returns a fencing token on success.
45    async fn acquire_lock(
46        &self,
47        key: &str,
48        holder: &str,
49        ttl_ms: u64,
50    ) -> Result<Option<u64>, ControlError>;
51    async fn renew_lock(&self, key: &str, holder: &str, ttl_ms: u64) -> Result<bool, ControlError>;
52    async fn release_lock(&self, key: &str, holder: &str) -> Result<(), ControlError>;
53}
54
55fn now_ms() -> u64 {
56    SystemTime::now()
57        .duration_since(UNIX_EPOCH)
58        .map(|d| d.as_millis() as u64)
59        .unwrap_or(0)
60}
61
62/// Single-node control plane: commands are applied directly to the local redb state machine
63/// (no consensus).
64#[derive(Clone)]
65pub struct LocalControlPlane {
66    sm: Arc<StateMachine>,
67}
68
69impl LocalControlPlane {
70    pub fn new(sm: Arc<StateMachine>) -> Self {
71        Self { sm }
72    }
73
74    /// Open a redb-backed single-node control plane.
75    pub fn open(path: impl AsRef<Path>) -> Result<Self, ControlError> {
76        Ok(Self {
77            sm: Arc::new(StateMachine::open(path)?),
78        })
79    }
80}
81
82#[async_trait]
83impl ControlPlane for LocalControlPlane {
84    async fn set(&self, key: &[u8], value: &[u8]) -> Result<(), ControlError> {
85        self.sm.apply(&Cmd::Set {
86            key: key.to_vec(),
87            value: value.to_vec(),
88        })?;
89        Ok(())
90    }
91
92    async fn get(&self, key: &[u8]) -> Result<Option<Vec<u8>>, ControlError> {
93        Ok(self.sm.get(key)?)
94    }
95
96    async fn delete(&self, key: &[u8]) -> Result<(), ControlError> {
97        self.sm.apply(&Cmd::Delete { key: key.to_vec() })?;
98        Ok(())
99    }
100
101    async fn cas(
102        &self,
103        key: &[u8],
104        expect: Option<&[u8]>,
105        value: &[u8],
106    ) -> Result<bool, ControlError> {
107        match self.sm.apply(&Cmd::Cas {
108            key: key.to_vec(),
109            expect: expect.map(|e| e.to_vec()),
110            value: value.to_vec(),
111        })? {
112            CmdResult::Bool(b) => Ok(b),
113            _ => Ok(false),
114        }
115    }
116
117    async fn acquire_lock(
118        &self,
119        key: &str,
120        holder: &str,
121        ttl_ms: u64,
122    ) -> Result<Option<u64>, ControlError> {
123        match self.sm.apply(&Cmd::AcquireLock {
124            key: key.to_string(),
125            holder: holder.to_string(),
126            now_ms: now_ms(),
127            ttl_ms,
128        })? {
129            CmdResult::Fencing(t) => Ok(t),
130            _ => Ok(None),
131        }
132    }
133
134    async fn renew_lock(&self, key: &str, holder: &str, ttl_ms: u64) -> Result<bool, ControlError> {
135        match self.sm.apply(&Cmd::RenewLock {
136            key: key.to_string(),
137            holder: holder.to_string(),
138            now_ms: now_ms(),
139            ttl_ms,
140        })? {
141            CmdResult::Bool(b) => Ok(b),
142            _ => Ok(false),
143        }
144    }
145
146    async fn release_lock(&self, key: &str, holder: &str) -> Result<(), ControlError> {
147        self.sm.apply(&Cmd::ReleaseLock {
148            key: key.to_string(),
149            holder: holder.to_string(),
150        })?;
151        Ok(())
152    }
153}
mocra_cluster/control.rs

mocra_cluster/
control.rs
