edgestore_repl/
replicated_engine.rs

1//! `ReplicatedEngine` — convenience wrapper for the common primary + replica pattern.
2//!
3//! Eliminates the three-step boilerplate of wiring an `Engine` with
4//! `HttpReplicationServer` and `AntiEntropyLoop`.
5//!
6//! ## Primary
7//!
8//! ```rust,no_run
9//! use edgestore::EdgestoreConfig;
10//! use edgestore_repl::ReplicatedEngine;
11//!
12//! let primary = ReplicatedEngine::open_primary(
13//!     EdgestoreConfig::new("/var/db/primary"),
14//!     "0.0.0.0:8900",
15//! ).unwrap();
16//!
17//! primary.engine().lock().unwrap().put(b"products", b"p1", b"data").unwrap();
18//! ```
19//!
20//! ## Replica
21//!
22//! ```rust,no_run
23//! use edgestore::EdgestoreConfig;
24//! use edgestore_repl::ReplicatedEngine;
25//!
26//! let replica = ReplicatedEngine::open_replica(
27//!     EdgestoreConfig::new("/var/db/replica"),
28//!     "http://primary-host:8900",
29//! ).unwrap();
30//!
31//! // Engine is read-only — write attempts return Err(EdgestoreError::ReadOnly).
32//! let val = replica.engine().lock().unwrap().get(b"products", b"p1").unwrap();
33//! ```
34//!
35//! ## Wake-on-flush (reduced replication lag)
36//!
37//! Primary registers an `on_segment_flushed` callback via `Engine::with_on_segment_flushed`
38//! before passing the engine to this wrapper. The callback wakes the anti-entropy loop
39//! via a channel instead of waiting for the poll interval.
40
41use std::path::PathBuf;
42use std::sync::{Arc, Mutex};
43
44use edgestore::{EdgestoreConfig, EdgestoreError, Engine};
45
46use crate::anti_entropy::AntiEntropyLoop;
47use crate::http_server::HttpReplicationServer;
48
49/// Wraps `Engine` + `HttpReplicationServer` (primary) or `Engine` + `AntiEntropyLoop` (replica).
50///
51/// Background threads are owned by this struct. When dropped, threads continue
52/// running until the process exits — same behavior as calling `start()` manually.
53pub struct ReplicatedEngine {
54    engine: Arc<Mutex<Engine>>,
55    // JoinHandles held to prevent accidental detach; threads are daemon-like (run until exit).
56    #[allow(dead_code)]
57    handles: Vec<std::thread::JoinHandle<()>>,
58    bound_port: Option<u16>,
59}
60
61impl ReplicatedEngine {
62    /// Open a primary replication node.
63    ///
64    /// Starts `HttpReplicationServer` on `bind_addr` in a background thread.
65    /// The engine is writable. Other nodes point their `AntiEntropyLoop` at this address.
66    pub fn open_primary(
67        config: EdgestoreConfig,
68        bind_addr: &str,
69    ) -> Result<ReplicatedEngine, EdgestoreError> {
70        let engine = Arc::new(Mutex::new(Engine::open(config)?));
71        let server = HttpReplicationServer::new(Arc::clone(&engine));
72        let (handle, port) = server.start(bind_addr)?;
73        Ok(ReplicatedEngine {
74            engine,
75            handles: vec![handle],
76            bound_port: Some(port),
77        })
78    }
79
80    /// Open a replica node.
81    ///
82    /// Opens the engine in read-only mode (`Engine::open_readonly`) and starts an
83    /// `AntiEntropyLoop` that pulls from `primary_url`. The engine rejects all write
84    /// calls with `Err(EdgestoreError::ReadOnly)`.
85    ///
86    /// `primary_url` must be the base URL of the primary's `HttpReplicationServer`
87    /// (e.g. `"http://host:8900"`).
88    pub fn open_replica(
89        config: EdgestoreConfig,
90        primary_url: &str,
91    ) -> Result<ReplicatedEngine, EdgestoreError> {
92        let db_path: PathBuf = config.path.clone();
93        let engine = Arc::new(Mutex::new(Engine::open_readonly(config)?));
94        let peer_id = "primary".to_string();
95        let loop_ = AntiEntropyLoop::new(
96            Arc::clone(&engine),
97            primary_url.to_string(),
98            peer_id,
99            db_path,
100        );
101        let handle = loop_.start();
102        Ok(ReplicatedEngine {
103            engine,
104            handles: vec![handle],
105            bound_port: None,
106        })
107    }
108
109    /// Access the underlying `Arc<Mutex<Engine>>` for reads (replica) or reads+writes (primary).
110    pub fn engine(&self) -> &Arc<Mutex<Engine>> {
111        &self.engine
112    }
113
114    /// Port the `HttpReplicationServer` is listening on, or `None` for a replica.
115    pub fn bound_port(&self) -> Option<u16> {
116        self.bound_port
117    }
118}
edgestore_repl/replicated_engine.rs

edgestore_repl/
replicated_engine.rs
