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>(node: Arc<Node<N>>, store_id: &str) -> Arc<Self> {
96        Self::new_with_backend(node, store_id, Arc::new(MemoryBackend))
97    }
98
99    /// Create a new SyncedStore with a custom persistence backend.
100    ///
101    /// On startup, attempts to load the local device's persisted slice. If
102    /// found, the in-memory state and version counter are restored so the
103    /// store picks up where it left off.
104    pub fn new_with_backend<N: NetworkProvider + 'static>(
105        node: Arc<Node<N>>,
106        store_id: &str,
107        backend: Arc<dyn StoreBackend>,
108    ) -> Arc<Self> {
109        // RFC 017 §8: synced store slices are keyed by the stable
110        // `device_id` (ULID), which is propagated between peers via the
111        // WebSocket hello handshake.
112        let device_id = node.local_info().device_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) = match backend.load(store_id, &device_id) {
118            Some((data, version)) => match serde_json::from_slice::<T>(&data) {
119                Ok(typed) => {
120                    let now = SystemTime::now()
121                        .duration_since(UNIX_EPOCH)
122                        .unwrap_or_default()
123                        .as_millis() as u64;
124                    let slice = Slice {
125                        device_id: device_id.clone(),
126                        data: typed,
127                        version,
128                        updated_at: now,
129                    };
130                    tracing::info!(
131                        store = store_id,
132                        version = version,
133                        "synced_store: restored local slice from backend"
134                    );
135                    (Some(slice), version)
136                }
137                Err(e) => {
138                    tracing::warn!(
139                        store = store_id,
140                        "synced_store: failed to deserialize persisted data, starting fresh: {e}"
141                    );
142                    (None, 0)
143                }
144            },
145            None => (None, 0),
146        };
147
148        let inner = Arc::new(StoreInner {
149            store_id: store_id.to_string(),
150            device_id,
151            local: RwLock::new(local),
152            remotes: RwLock::new(HashMap::new()),
153            version: AtomicU64::new(initial_version),
154            event_tx,
155            backend,
156        });
157
158        let task_handle = sync::spawn_sync_task(node, inner.clone(), broadcast_rx);
159
160        Arc::new(Self {
161            inner,
162            broadcast_tx,
163            task_handle: tokio::sync::Mutex::new(Some(task_handle)),
164        })
165    }
166
167    // ── Write ───────────────────────────────────────────────────────
168
169    /// Update this device's data. Increments version and broadcasts.
170    pub async fn set(&self, data: T) {
171        let version = self.inner.version.fetch_add(1, Ordering::SeqCst) + 1;
172        let now = SystemTime::now()
173            .duration_since(UNIX_EPOCH)
174            .unwrap_or_default()
175            .as_millis() as u64;
176
177        let slice = Slice {
178            device_id: self.inner.device_id.clone(),
179            data: data.clone(),
180            version,
181            updated_at: now,
182        };
183
184        {
185            let mut local = self.inner.local.write().await;
186            *local = Some(slice);
187        }
188
189        // Persist to backend.
190        if let Ok(serialized) = serde_json::to_vec(&data) {
191            self.inner.backend.save(
192                &self.inner.store_id,
193                &self.inner.device_id,
194                &serialized,
195                version,
196            );
197        }
198
199        // Request the sync task to broadcast.
200        if let Ok(serialized) = serde_json::to_value(&data) {
201            let msg = SyncMessage::Update {
202                device_id: self.inner.device_id.clone(),
203                data: serialized,
204                version,
205                updated_at: now,
206            };
207            let _ = self.broadcast_tx.send(msg);
208        }
209
210        let _ = self.inner.event_tx.send(StoreEvent::LocalChanged(data));
211    }
212
213    // ── Read ────────────────────────────────────────────────────────
214
215    /// This device's current data, or `None` if `set()` hasn't been called.
216    pub async fn local(&self) -> Option<T> {
217        let local = self.inner.local.read().await;
218        local.as_ref().map(|s| s.data.clone())
219    }
220
221    /// A specific peer's slice, or `None` if we haven't received their data.
222    pub async fn get(&self, device_id: &str) -> Option<Slice<T>> {
223        let remotes = self.inner.remotes.read().await;
224        remotes.get(device_id).cloned()
225    }
226
227    /// All slices (local + remote).
228    pub async fn all(&self) -> HashMap<String, Slice<T>> {
229        let mut result = HashMap::new();
230
231        {
232            let local = self.inner.local.read().await;
233            if let Some(slice) = local.as_ref() {
234                result.insert(slice.device_id.clone(), slice.clone());
235            }
236        }
237
238        {
239            let remotes = self.inner.remotes.read().await;
240            for (id, slice) in remotes.iter() {
241                result.insert(id.clone(), slice.clone());
242            }
243        }
244
245        result
246    }
247
248    /// Device IDs that have data in this store (local + remote).
249    pub async fn device_ids(&self) -> Vec<String> {
250        let mut ids = Vec::new();
251
252        {
253            let local = self.inner.local.read().await;
254            if let Some(slice) = local.as_ref() {
255                ids.push(slice.device_id.clone());
256            }
257        }
258
259        {
260            let remotes = self.inner.remotes.read().await;
261            ids.extend(remotes.keys().cloned());
262        }
263
264        ids
265    }
266
267    /// Subscribe to store change events.
268    pub fn subscribe(&self) -> broadcast::Receiver<StoreEvent<T>> {
269        self.inner.event_tx.subscribe()
270    }
271
272    /// The store identifier.
273    pub fn store_id(&self) -> &str {
274        &self.inner.store_id
275    }
276
277    /// This device's ID.
278    pub fn device_id(&self) -> &str {
279        &self.inner.device_id
280    }
281
282    /// Current local version number.
283    pub fn version(&self) -> u64 {
284        self.inner.version.load(Ordering::SeqCst)
285    }
286
287    // ── Lifecycle ───────────────────────────────────────────────────
288
289    /// Stop the background sync task.
290    pub async fn stop(&self) {
291        let mut handle = self.task_handle.lock().await;
292        if let Some(h) = handle.take() {
293            h.abort();
294            tracing::info!(
295                store = self.inner.store_id.as_str(),
296                "synced_store: stopped"
297            );
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
