rabia_engine/lib.rs
1//! # Rabia Engine - SMR Protocol Coordinator
2//!
3//! Implementation of the Rabia consensus protocol for State Machine Replication.
4//!
5//! This crate provides the core engine implementing the Rabia SMR protocol,
6//! ensuring all replicas apply operations in the same order and providing
7//! strong consistency guarantees across distributed nodes. The engine handles
8//! consensus phases, operation ordering, and coordination with state machines.
9//!
10//! ## SMR Protocol Components
11//!
12//! - **RabiaEngine**: The main SMR protocol engine that ensures operation ordering
13//! - **RabiaConfig**: Configuration for the SMR protocol behavior and performance
14//! - **EngineState**: Internal state management for consensus coordination
15//! - **Operation Submission**: Interface for submitting operations to the SMR system
16//!
17//! ## SMR Protocol Usage
18//!
19//! ```rust,no_run
20//! use rabia_engine::{RabiaEngine, RabiaConfig, EngineCommand, CommandRequest};
21//! use rabia_core::{state_machine::{StateMachine, Snapshot}, network::ClusterConfig, NodeId, Command, CommandBatch};
22//! use rabia_persistence::InMemoryPersistence;
23//! use std::collections::HashSet;
24//! use tokio::sync::mpsc;
25//! use bytes::Bytes;
26//!
27//! // Example state machine implementation
28//! #[derive(Clone)]
29//! struct ExampleStateMachine {
30//! counter: i64,
31//! }
32//!
33//! #[async_trait::async_trait]
34//! impl StateMachine for ExampleStateMachine {
35//! type State = i64;
36//!
37//! async fn apply_command(&mut self, _command: &Command) -> rabia_core::Result<Bytes> {
38//! self.counter += 1;
39//! Ok(Bytes::from(format!("Counter: {}", self.counter)))
40//! }
41//!
42//! async fn create_snapshot(&self) -> rabia_core::Result<Snapshot> {
43//! Ok(Snapshot::new(1, self.counter.to_be_bytes().to_vec()))
44//! }
45//!
46//! async fn restore_snapshot(&mut self, snapshot: &Snapshot) -> rabia_core::Result<()> {
47//! let bytes: [u8; 8] = snapshot.data.as_ref().try_into().unwrap_or([0; 8]);
48//! self.counter = i64::from_be_bytes(bytes);
49//! Ok(())
50//! }
51//!
52//! async fn get_state(&self) -> Self::State {
53//! self.counter
54//! }
55//! }
56//!
57//! #[tokio::main]
58//! async fn main() -> Result<(), Box<dyn std::error::Error>> {
59//! let node_id = NodeId::new();
60//! let mut node_ids = HashSet::new();
61//! node_ids.insert(node_id);
62//!
63//! // Protocol configuration
64//! let config = RabiaConfig::default();
65//! let cluster_config = ClusterConfig::new(node_id, node_ids);
66//!
67//! // State machine and persistence
68//! let state_machine = ExampleStateMachine { counter: 0 };
69//! let persistence = InMemoryPersistence::new();
70//!
71//! // Create command channel for engine commands
72//! let (command_tx, command_rx) = mpsc::unbounded_channel();
73//!
74//! // Create Rabia protocol engine with TCP networking
75//! let engine = RabiaEngine::new_with_tcp(
76//! node_id,
77//! config,
78//! cluster_config,
79//! state_machine,
80//! persistence,
81//! command_rx,
82//! ).await?;
83//!
84//! // Start Rabia protocol engine
85//! let handle = tokio::spawn(async move {
86//! engine.run().await
87//! });
88//!
89//! // Submit batch for consensus
90//! let command = Command::new(b"your_command_data".to_vec());
91//! let batch = CommandBatch::new(vec![command]);
92//! let (response_tx, response_rx) = tokio::sync::oneshot::channel();
93//! let request = CommandRequest { batch, response_tx };
94//! command_tx.send(EngineCommand::ProcessBatch(request))?;
95//!
96//! Ok(())
97//! }
98//! ```
99//!
100//! The engine ensures that your state machine's operations are applied in the same
101//! order across all healthy replicas, providing strong consistency for your
102//! distributed application.
103
104pub mod config;
105pub mod engine;
106pub mod leader;
107pub mod network;
108pub mod state;
109
110pub use config::*;
111pub use engine::*;
112pub use leader::*;
113pub use network::*;
114pub use state::*;