laminar_core/cluster/control/barrier/
kv.rs

1//! Cluster key-value seam and deterministic in-memory implementation.
2
3use super::{async_trait, FxHashMap, Mutex, NodeId};
4
5/// Gossip-KV seam.
6#[async_trait]
7pub trait ClusterKv: Send + Sync + 'static {
8    /// Write `value` to this instance's `key` slot (overwrites).
9    async fn write(&self, key: &str, value: String);
10    /// Write with transport failure reporting when the backend supports it.
11    ///
12    /// Fast gossip implementations may use the default because their write API has no result.
13    /// Durable control implementations override this so recovery never treats a dropped write as
14    /// successful.
15    ///
16    /// # Errors
17    /// Durable implementations return a transport or storage error when the value was not
18    /// accepted by their authority.
19    async fn write_checked(&self, key: &str, value: String) -> Result<(), String> {
20        self.write(key, value).await;
21        Ok(())
22    }
23    /// Read `key` from `who`'s slot.
24    async fn read_from(&self, who: NodeId, key: &str) -> Option<String>;
25    /// Read with transport failure reporting when the backend supports it.
26    ///
27    /// # Errors
28    /// Durable implementations return a transport or storage error. A genuinely absent key is
29    /// `Ok(None)`.
30    async fn read_from_checked(&self, who: NodeId, key: &str) -> Result<Option<String>, String> {
31        Ok(self.read_from(who, key).await)
32    }
33    /// Every visible instance's value for `key`.
34    async fn scan(&self, key: &str) -> Vec<(NodeId, String)>;
35    /// Scan with transport failure reporting when the backend supports it.
36    ///
37    /// # Errors
38    /// Durable implementations fail the whole scan when any visible participant cannot be read.
39    async fn scan_checked(&self, key: &str) -> Result<Vec<(NodeId, String)>, String> {
40        Ok(self.scan(key).await)
41    }
42}
43
44/// In-memory KV for tests.
45#[derive(Debug)]
46pub struct InMemoryKv {
47    local_id: NodeId,
48    state: Mutex<FxHashMap<(NodeId, String), String>>,
49}
50
51impl InMemoryKv {
52    /// Create a new in-memory KV identified as `local_id`.
53    #[must_use]
54    pub fn new(local_id: NodeId) -> Self {
55        Self {
56            local_id,
57            state: Mutex::new(FxHashMap::default()),
58        }
59    }
60
61    /// Seed a remote peer's state for tests.
62    pub fn seed(&self, peer: NodeId, key: &str, value: String) {
63        self.state.lock().insert((peer, key.to_string()), value);
64    }
65}
66
67#[async_trait]
68impl ClusterKv for InMemoryKv {
69    async fn write(&self, key: &str, value: String) {
70        self.state
71            .lock()
72            .insert((self.local_id, key.to_string()), value);
73    }
74
75    async fn read_from(&self, who: NodeId, key: &str) -> Option<String> {
76        self.state.lock().get(&(who, key.to_string())).cloned()
77    }
78
79    async fn scan(&self, key: &str) -> Vec<(NodeId, String)> {
80        self.state
81            .lock()
82            .iter()
83            .filter(|((_, k), _)| k == key)
84            .map(|((n, _), v)| (*n, v.clone()))
85            .collect()
86    }
87}
laminar_core/cluster/control/barrier/kv.rs

laminar_core/cluster/control/barrier/
kv.rs
