truffle_core/synced_store/
mod.rs

1//! SyncedStore — device-owned state synchronization across the mesh.
2//!
3//! Each device owns exactly one slice of type `T` per store. Writes update
4//! the local slice and broadcast the change. Remote slices are received and
5//! cached automatically via a background sync task.
6//!
7//! # Quick start
8//!
9//! ```ignore
10//! use std::sync::Arc;
11//! use truffle_core::synced_store::SyncedStore;
12//!
13//! let node = Arc::new(node);
14//! let store: Arc<SyncedStore<MyState>> = SyncedStore::new(node, "my-state");
15//!
16//! store.set(MyState { value: 42 }).await;
17//! let my_data = store.local();
18//! let all_data = store.all().await;
19//! ```
20
21pub mod backend;
22pub mod sync;
23pub mod types;
24
25#[cfg(test)]
26mod tests;
27
28pub use backend::{FileBackend, MemoryBackend, StoreBackend};
29pub use types::{Slice, StoreEvent, SyncMessage};
30
31use std::collections::HashMap;
32use std::sync::atomic::{AtomicU64, Ordering};
33use std::sync::Arc;
34use std::time::{SystemTime, UNIX_EPOCH};
35
36use serde::de::DeserializeOwned;
37use serde::Serialize;
38use tokio::sync::{broadcast, mpsc, RwLock};
39use tokio::task::JoinHandle;
40
41use crate::network::NetworkProvider;
42use crate::node::Node;
43
44// ---------------------------------------------------------------------------
45// StoreInner — shared state between public API and sync task
46// ---------------------------------------------------------------------------
47
48/// Internal shared state for a SyncedStore. Accessed by both the public API
49/// and the background sync task.
50pub(crate) struct StoreInner<T> {
51    /// Store identifier (e.g., "sessions", "config").
52    pub(crate) store_id: String,
53    /// This device's stable node ID.
54    pub(crate) device_id: String,
55    /// Local slice (our data).
56    pub(crate) local: RwLock<Option<Slice<T>>>,
57    /// Remote peers' slices, keyed by device_id.
58    pub(crate) remotes: RwLock<HashMap<String, Slice<T>>>,
59    /// Current local version counter.
60    pub(crate) version: AtomicU64,
61    /// Event broadcast channel.
62    pub(crate) event_tx: broadcast::Sender<StoreEvent<T>>,
63    /// Persistence backend.
64    pub(crate) backend: Arc<dyn StoreBackend>,
65}
66
67// ---------------------------------------------------------------------------
68// SyncedStore — public API
69// ---------------------------------------------------------------------------
70
71/// A synchronized key-value store with device-owned slices.
72///
73/// Each device owns one slice of type `T`. Writes update only the local
74/// slice and broadcast the change. Remote slices are received and cached
75/// automatically.
76///
77/// Created via [`SyncedStore::new`] or [`Node::synced_store`].
78pub struct SyncedStore<T> {
79    pub(crate) inner: Arc<StoreInner<T>>,
80    /// Channel to request outbound broadcasts from the sync task.
81    broadcast_tx: mpsc::UnboundedSender<SyncMessage>,
82    /// Handle to the background sync task.
83    task_handle: tokio::sync::Mutex<Option<JoinHandle<()>>>,
84}
85
86impl<T: Serialize + DeserializeOwned + Clone + Send + Sync + 'static> SyncedStore<T> {
87    /// Create a new SyncedStore and start the background sync task.
88    ///
89    /// Uses the default [`MemoryBackend`] (no persistence). For durable
90    /// storage across restarts, use [`new_with_backend`](Self::new_with_backend).
91    ///
92    /// The store syncs on namespace `"ss:{store_id}"`. The caller must hold
93    /// an `Arc<Node<N>>` because the background task needs to outlive this
94    /// constructor call.
95    pub fn new<N: NetworkProvider + 'static>(
96        node: Arc<Node<N>>,
97        store_id: &str,
98    ) -> Arc<Self> {
99        Self::new_with_backend(node, store_id, Arc::new(MemoryBackend))
100    }
101
102    /// Create a new SyncedStore with a custom persistence backend.
103    ///
104    /// On startup, attempts to load the local device's persisted slice. If
105    /// found, the in-memory state and version counter are restored so the
106    /// store picks up where it left off.
107    pub fn new_with_backend<N: NetworkProvider + 'static>(
108        node: Arc<Node<N>>,
109        store_id: &str,
110        backend: Arc<dyn StoreBackend>,
111    ) -> Arc<Self> {
112        let device_id = node.local_info().id;
113        let (event_tx, _) = broadcast::channel(256);
114        let (broadcast_tx, broadcast_rx) = mpsc::unbounded_channel();
115
116        // Attempt to restore persisted local slice.
117        let (local, initial_version) =
118            match backend.load(store_id, &device_id) {
119                Some((data, version)) => {
120                    match serde_json::from_slice::<T>(&data) {
121                        Ok(typed) => {
122                            let now = SystemTime::now()
123                                .duration_since(UNIX_EPOCH)
124                                .unwrap_or_default()
125                                .as_millis() as u64;
126                            let slice = Slice {
127                                device_id: device_id.clone(),
128                                data: typed,
129                                version,
130                                updated_at: now,
131                            };
132                            tracing::info!(
133                                store = store_id,
134                                version = version,
135                                "synced_store: restored local slice from backend"
136                            );
137                            (Some(slice), version)
138                        }
139                        Err(e) => {
140                            tracing::warn!(
141                                store = store_id,
142                                "synced_store: failed to deserialize persisted data, starting fresh: {e}"
143                            );
144                            (None, 0)
145                        }
146                    }
147                }
148                None => (None, 0),
149            };
150
151        let inner = Arc::new(StoreInner {
152            store_id: store_id.to_string(),
153            device_id,
154            local: RwLock::new(local),
155            remotes: RwLock::new(HashMap::new()),
156            version: AtomicU64::new(initial_version),
157            event_tx,
158            backend,
159        });
160
161        let task_handle = sync::spawn_sync_task(node, inner.clone(), broadcast_rx);
162
163        Arc::new(Self {
164            inner,
165            broadcast_tx,
166            task_handle: tokio::sync::Mutex::new(Some(task_handle)),
167        })
168    }
169
170    // ── Write ───────────────────────────────────────────────────────
171
172    /// Update this device's data. Increments version and broadcasts.
173    pub async fn set(&self, data: T) {
174        let version = self.inner.version.fetch_add(1, Ordering::SeqCst) + 1;
175        let now = SystemTime::now()
176            .duration_since(UNIX_EPOCH)
177            .unwrap_or_default()
178            .as_millis() as u64;
179
180        let slice = Slice {
181            device_id: self.inner.device_id.clone(),
182            data: data.clone(),
183            version,
184            updated_at: now,
185        };
186
187        {
188            let mut local = self.inner.local.write().await;
189            *local = Some(slice);
190        }
191
192        // Persist to backend.
193        if let Ok(serialized) = serde_json::to_vec(&data) {
194            self.inner.backend.save(
195                &self.inner.store_id,
196                &self.inner.device_id,
197                &serialized,
198                version,
199            );
200        }
201
202        // Request the sync task to broadcast.
203        if let Ok(serialized) = serde_json::to_value(&data) {
204            let msg = SyncMessage::Update {
205                device_id: self.inner.device_id.clone(),
206                data: serialized,
207                version,
208                updated_at: now,
209            };
210            let _ = self.broadcast_tx.send(msg);
211        }
212
213        let _ = self.inner.event_tx.send(StoreEvent::LocalChanged(data));
214    }
215
216    // ── Read ────────────────────────────────────────────────────────
217
218    /// This device's current data, or `None` if `set()` hasn't been called.
219    pub async fn local(&self) -> Option<T> {
220        let local = self.inner.local.read().await;
221        local.as_ref().map(|s| s.data.clone())
222    }
223
224    /// A specific peer's slice, or `None` if we haven't received their data.
225    pub async fn get(&self, device_id: &str) -> Option<Slice<T>> {
226        let remotes = self.inner.remotes.read().await;
227        remotes.get(device_id).cloned()
228    }
229
230    /// All slices (local + remote).
231    pub async fn all(&self) -> HashMap<String, Slice<T>> {
232        let mut result = HashMap::new();
233
234        {
235            let local = self.inner.local.read().await;
236            if let Some(slice) = local.as_ref() {
237                result.insert(slice.device_id.clone(), slice.clone());
238            }
239        }
240
241        {
242            let remotes = self.inner.remotes.read().await;
243            for (id, slice) in remotes.iter() {
244                result.insert(id.clone(), slice.clone());
245            }
246        }
247
248        result
249    }
250
251    /// Device IDs that have data in this store (local + remote).
252    pub async fn device_ids(&self) -> Vec<String> {
253        let mut ids = Vec::new();
254
255        {
256            let local = self.inner.local.read().await;
257            if let Some(slice) = local.as_ref() {
258                ids.push(slice.device_id.clone());
259            }
260        }
261
262        {
263            let remotes = self.inner.remotes.read().await;
264            ids.extend(remotes.keys().cloned());
265        }
266
267        ids
268    }
269
270    /// Subscribe to store change events.
271    pub fn subscribe(&self) -> broadcast::Receiver<StoreEvent<T>> {
272        self.inner.event_tx.subscribe()
273    }
274
275    /// The store identifier.
276    pub fn store_id(&self) -> &str {
277        &self.inner.store_id
278    }
279
280    /// This device's ID.
281    pub fn device_id(&self) -> &str {
282        &self.inner.device_id
283    }
284
285    /// Current local version number.
286    pub fn version(&self) -> u64 {
287        self.inner.version.load(Ordering::SeqCst)
288    }
289
290    // ── Lifecycle ───────────────────────────────────────────────────
291
292    /// Stop the background sync task.
293    pub async fn stop(&self) {
294        let mut handle = self.task_handle.lock().await;
295        if let Some(h) = handle.take() {
296            h.abort();
297            tracing::info!(store = self.inner.store_id.as_str(), "synced_store: stopped");
298        }
299    }
300}
301
302impl<T> Drop for SyncedStore<T> {
303    fn drop(&mut self) {
304        // Best-effort abort on drop.
305        if let Ok(mut handle) = self.task_handle.try_lock() {
306            if let Some(h) = handle.take() {
307                h.abort();
308            }
309        }
310    }
311}
truffle_core/synced_store/mod.rs

truffle_core/synced_store/
mod.rs
