hightower_kv/
replication.rs

1use std::sync::Arc;
2
3use crate::command::Command;
4use crate::engine::{KvEngine, SnapshotEngine};
5use crate::error::{Error, Result};
6use crate::state::{ApplyOutcome, KvState};
7
8/// Trait for submitting commands to a replication system
9pub trait CommandSubmitter: Send + Sync {
10    /// Submits a single command and returns the outcome
11    fn submit(&self, command: Command) -> Result<ApplyOutcome>;
12
13    /// Submits a batch of commands and returns the outcomes
14    fn submit_batch<I>(&self, commands: I) -> Result<Vec<ApplyOutcome>>
15    where
16        I: IntoIterator<Item = Command>;
17}
18
19/// Trait for providing snapshots of the replicated state
20pub trait SnapshotProvider: Send + Sync {
21    /// Returns a snapshot of the current state
22    fn snapshot_state(&self) -> Result<KvState>;
23    /// Returns the latest version number
24    fn latest_version(&self) -> Result<u64>;
25}
26
27/// Combined trait for replication handles that can submit commands and provide snapshots
28pub trait ReplicationHandle: CommandSubmitter + SnapshotProvider {}
29
30impl<T> ReplicationHandle for T where T: CommandSubmitter + SnapshotProvider {}
31
32/// Null implementation of replication that returns unimplemented errors
33#[derive(Debug)]
34pub struct NullReplication;
35
36impl CommandSubmitter for NullReplication {
37    fn submit(&self, _command: Command) -> Result<ApplyOutcome> {
38        Err(Error::Unimplemented("replication::submit"))
39    }
40
41    fn submit_batch<I>(&self, _commands: I) -> Result<Vec<ApplyOutcome>>
42    where
43        I: IntoIterator<Item = Command>,
44    {
45        Err(Error::Unimplemented("replication::submit_batch"))
46    }
47}
48
49impl SnapshotProvider for NullReplication {
50    fn snapshot_state(&self) -> Result<KvState> {
51        Err(Error::Unimplemented("replication::latest_snapshot"))
52    }
53
54    fn latest_version(&self) -> Result<u64> {
55        Err(Error::Unimplemented("replication::latest_version"))
56    }
57}
58
59/// Local replication implementation that forwards to a single engine
60#[derive(Debug, Clone)]
61pub struct LocalReplication<E>
62where
63    E: KvEngine + SnapshotEngine,
64{
65    engine: Arc<E>,
66}
67
68impl<E> LocalReplication<E>
69where
70    E: KvEngine + SnapshotEngine,
71{
72    /// Creates a new local replication handle for the given engine
73    pub fn new(engine: Arc<E>) -> Self {
74        Self { engine }
75    }
76}
77
78impl<E> CommandSubmitter for LocalReplication<E>
79where
80    E: KvEngine + SnapshotEngine,
81{
82    fn submit(&self, command: Command) -> Result<ApplyOutcome> {
83        self.engine.submit(command)
84    }
85
86    fn submit_batch<I>(&self, commands: I) -> Result<Vec<ApplyOutcome>>
87    where
88        I: IntoIterator<Item = Command>,
89    {
90        self.engine.submit_batch(commands)
91    }
92}
93
94impl<E> SnapshotProvider for LocalReplication<E>
95where
96    E: KvEngine + SnapshotEngine,
97{
98    fn snapshot_state(&self) -> Result<KvState> {
99        Ok(self.engine.snapshot_state())
100    }
101
102    fn latest_version(&self) -> Result<u64> {
103        Ok(self.engine.latest_version())
104    }
105}
106
107#[cfg(test)]
108mod tests {
109    use super::*;
110    use crate::config::StoreConfig;
111    use crate::engine::{SingleNodeEngine, SnapshotEngine};
112    use tempfile::tempdir;
113
114    #[test]
115    fn submit_returns_unimplemented() {
116        let repl = NullReplication;
117        let err = repl
118            .submit(Command::Delete {
119                key: b"k".to_vec(),
120                version: 0,
121                timestamp: 0,
122            })
123            .unwrap_err();
124        assert!(matches!(err, Error::Unimplemented("replication::submit")));
125    }
126
127    #[test]
128    fn local_replication_forwards_calls() {
129        let temp = tempdir().unwrap();
130        let mut cfg = StoreConfig::default();
131        cfg.data_dir = temp.path().join("repl").to_string_lossy().into_owned();
132        let engine = Arc::new(SingleNodeEngine::with_config(cfg).unwrap());
133        let repl = LocalReplication::new(Arc::clone(&engine));
134
135        repl.submit(Command::Set {
136            key: b"k".to_vec(),
137            value: b"v".to_vec(),
138            version: engine.latest_version() + 1,
139            timestamp: 1,
140        })
141        .unwrap();
142
143        let snapshot = repl.snapshot_state().unwrap();
144        assert_eq!(snapshot.get(b"k"), Some(&b"v"[..]));
145    }
146}
hightower_kv/replication.rs

hightower_kv/
replication.rs
