forge_orchestration/
storage.rs

1//! Storage backends for Forge state management
2//!
3//! ## Table of Contents
4//! - **StateStore**: Trait for state storage backends
5//! - **MemoryStore**: In-memory store (default)
6//! - **FileStore**: File-based persistent storage
7//!
8//! Optional backends (require feature flags):
9//! - **RocksDbStore**: Local fast storage using RocksDB
10//! - **EtcdStore**: Distributed storage using etcd
11
12use crate::error::{ForgeError, Result};
13use async_trait::async_trait;
14use serde::{de::DeserializeOwned, Serialize};
15use std::collections::HashMap;
16use std::path::{Path, PathBuf};
17use std::sync::Arc;
18use tokio::sync::RwLock;
19use tracing::{debug, info};
20
21/// Trait for state storage backends
22#[async_trait]
23pub trait StateStore: Send + Sync {
24    /// Get a value by key
25    async fn get(&self, key: &str) -> Result<Option<Vec<u8>>>;
26
27    /// Set a value
28    async fn set(&self, key: &str, value: Vec<u8>) -> Result<()>;
29
30    /// Delete a key
31    async fn delete(&self, key: &str) -> Result<()>;
32
33    /// List keys with a prefix
34    async fn list_prefix(&self, prefix: &str) -> Result<Vec<String>>;
35
36    /// Store name for logging
37    fn name(&self) -> &str;
38}
39
40/// Extension functions for StateStore (not in trait to keep it object-safe)
41
42/// Check if a key exists in the store
43pub async fn store_exists(store: &dyn StateStore, key: &str) -> Result<bool> {
44    Ok(store.get(key).await?.is_some())
45}
46
47/// Get and deserialize JSON from the store
48pub async fn store_get_json<T: DeserializeOwned>(store: &dyn StateStore, key: &str) -> Result<Option<T>> {
49    match store.get(key).await? {
50        Some(bytes) => {
51            let value = serde_json::from_slice(&bytes)?;
52            Ok(Some(value))
53        }
54        None => Ok(None),
55    }
56}
57
58/// Serialize and set JSON in the store
59pub async fn store_set_json<T: Serialize>(store: &dyn StateStore, key: &str, value: &T) -> Result<()> {
60    let bytes = serde_json::to_vec(value)?;
61    store.set(key, bytes).await
62}
63
64/// In-memory store for testing
65#[derive(Debug, Default)]
66pub struct MemoryStore {
67    data: RwLock<HashMap<String, Vec<u8>>>,
68}
69
70impl MemoryStore {
71    /// Create a new memory store
72    pub fn new() -> Self {
73        Self {
74            data: RwLock::new(HashMap::new()),
75        }
76    }
77}
78
79#[async_trait]
80impl StateStore for MemoryStore {
81    async fn get(&self, key: &str) -> Result<Option<Vec<u8>>> {
82        let data = self.data.read().await;
83        Ok(data.get(key).cloned())
84    }
85
86    async fn set(&self, key: &str, value: Vec<u8>) -> Result<()> {
87        let mut data = self.data.write().await;
88        data.insert(key.to_string(), value);
89        Ok(())
90    }
91
92    async fn delete(&self, key: &str) -> Result<()> {
93        let mut data = self.data.write().await;
94        data.remove(key);
95        Ok(())
96    }
97
98    async fn list_prefix(&self, prefix: &str) -> Result<Vec<String>> {
99        let data = self.data.read().await;
100        Ok(data
101            .keys()
102            .filter(|k| k.starts_with(prefix))
103            .cloned()
104            .collect())
105    }
106
107    fn name(&self) -> &str {
108        "memory"
109    }
110}
111
112/// File-based persistent storage
113///
114/// Simple JSON file storage for development and small deployments.
115pub struct FileStore {
116    path: PathBuf,
117    data: RwLock<HashMap<String, Vec<u8>>>,
118}
119
120impl FileStore {
121    /// Open or create a file store
122    pub fn open(path: impl AsRef<Path>) -> Result<Self> {
123        let path = path.as_ref().to_path_buf();
124
125        let data = if path.exists() {
126            let contents = std::fs::read_to_string(&path)
127                .map_err(|e| ForgeError::storage(format!("Failed to read store: {}", e)))?;
128            serde_json::from_str(&contents).unwrap_or_default()
129        } else {
130            HashMap::new()
131        };
132
133        info!(path = %path.display(), "File store opened");
134
135        Ok(Self {
136            path,
137            data: RwLock::new(data),
138        })
139    }
140
141    /// Persist data to disk
142    pub async fn flush(&self) -> Result<()> {
143        let data = self.data.read().await;
144        let contents = serde_json::to_string_pretty(&*data)?;
145
146        if let Some(parent) = self.path.parent() {
147            std::fs::create_dir_all(parent)
148                .map_err(|e| ForgeError::storage(format!("Failed to create dir: {}", e)))?;
149        }
150
151        std::fs::write(&self.path, contents)
152            .map_err(|e| ForgeError::storage(format!("Failed to write store: {}", e)))?;
153
154        debug!(path = %self.path.display(), "File store flushed");
155        Ok(())
156    }
157}
158
159#[async_trait]
160impl StateStore for FileStore {
161    async fn get(&self, key: &str) -> Result<Option<Vec<u8>>> {
162        let data = self.data.read().await;
163        Ok(data.get(key).cloned())
164    }
165
166    async fn set(&self, key: &str, value: Vec<u8>) -> Result<()> {
167        let mut data = self.data.write().await;
168        data.insert(key.to_string(), value);
169        Ok(())
170    }
171
172    async fn delete(&self, key: &str) -> Result<()> {
173        let mut data = self.data.write().await;
174        data.remove(key);
175        Ok(())
176    }
177
178    async fn list_prefix(&self, prefix: &str) -> Result<Vec<String>> {
179        let data = self.data.read().await;
180        Ok(data
181            .keys()
182            .filter(|k| k.starts_with(prefix))
183            .cloned()
184            .collect())
185    }
186
187    fn name(&self) -> &str {
188        "file"
189    }
190}
191
192/// Consensus-backed state store (requires the `raft` feature).
193///
194/// `storage.rs` is a single file rather than a directory module, so the
195/// submodule is declared with an explicit `#[path]` to `storage/raft.rs`.
196#[cfg(feature = "raft")]
197#[path = "storage/raft.rs"]
198pub mod raft;
199
200/// Re-export the Raft-backed store for ergonomic access (requires the `raft` feature).
201///
202/// Bootstrap with `RaftStateStore::bootstrap_single_node(id).await?` and pass the
203/// `Arc` to `ForgeBuilder::with_store(...)`.
204#[cfg(feature = "raft")]
205pub use raft::RaftStateStore;
206
207/// Type alias for boxed store
208pub type BoxedStateStore = Arc<dyn StateStore>;
209
210/// Create a memory store
211pub fn memory_store() -> BoxedStateStore {
212    Arc::new(MemoryStore::new()) as BoxedStateStore
213}
214
215/// Key prefixes for different data types
216pub mod keys {
217    /// Job key prefix
218    pub const JOBS: &str = "forge/jobs";
219    /// Shard key prefix
220    pub const SHARDS: &str = "forge/shards";
221    /// Expert key prefix
222    pub const EXPERTS: &str = "forge/experts";
223    /// Node key prefix
224    pub const NODES: &str = "forge/nodes";
225    /// Config key prefix
226    pub const CONFIG: &str = "forge/config";
227    /// Metrics key prefix
228    pub const METRICS: &str = "forge/metrics";
229    /// Assignments key (reconciler actual-state map)
230    pub const ASSIGNMENTS: &str = "forge/assignments";
231
232    /// Build a job key
233    pub fn job(id: &str) -> String {
234        format!("{}/{}", JOBS, id)
235    }
236
237    /// Build a shard key
238    pub fn shard(id: u64) -> String {
239        format!("{}/{}", SHARDS, id)
240    }
241
242    /// Build an expert key
243    pub fn expert(index: usize) -> String {
244        format!("{}/{}", EXPERTS, index)
245    }
246
247    /// Build a node key
248    pub fn node(id: &str) -> String {
249        format!("{}/{}", NODES, id)
250    }
251}
252
253#[cfg(test)]
254mod tests {
255    use super::*;
256
257    #[tokio::test]
258    async fn test_memory_store_basic() {
259        let store = MemoryStore::new();
260
261        // Set and get
262        store.set("key1", b"value1".to_vec()).await.unwrap();
263        let value = store.get("key1").await.unwrap();
264        assert_eq!(value, Some(b"value1".to_vec()));
265
266        // Delete
267        store.delete("key1").await.unwrap();
268        let value = store.get("key1").await.unwrap();
269        assert!(value.is_none());
270    }
271
272    #[tokio::test]
273    async fn test_memory_store_prefix() {
274        let store = MemoryStore::new();
275
276        store.set("prefix/a", b"1".to_vec()).await.unwrap();
277        store.set("prefix/b", b"2".to_vec()).await.unwrap();
278        store.set("other/c", b"3".to_vec()).await.unwrap();
279
280        let keys = store.list_prefix("prefix/").await.unwrap();
281        assert_eq!(keys.len(), 2);
282        assert!(keys.contains(&"prefix/a".to_string()));
283        assert!(keys.contains(&"prefix/b".to_string()));
284    }
285
286    #[tokio::test]
287    async fn test_memory_store_json() {
288        let store = MemoryStore::new();
289
290        #[derive(Debug, PartialEq, Serialize, serde::Deserialize)]
291        struct TestData {
292            name: String,
293            value: i32,
294        }
295
296        let data = TestData {
297            name: "test".to_string(),
298            value: 42,
299        };
300
301        store_set_json(&store, "json_key", &data).await.unwrap();
302        let loaded: Option<TestData> = store_get_json(&store, "json_key").await.unwrap();
303        assert_eq!(loaded, Some(data));
304    }
305
306    #[test]
307    fn test_key_builders() {
308        assert_eq!(keys::job("my-job"), "forge/jobs/my-job");
309        assert_eq!(keys::shard(42), "forge/shards/42");
310        assert_eq!(keys::expert(0), "forge/experts/0");
311    }
312}
forge_orchestration/storage.rs

forge_orchestration/
storage.rs
