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::{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}
64
65// ---------------------------------------------------------------------------
66// SyncedStore — public API
67// ---------------------------------------------------------------------------
68
69/// A synchronized key-value store with device-owned slices.
70///
71/// Each device owns one slice of type `T`. Writes update only the local
72/// slice and broadcast the change. Remote slices are received and cached
73/// automatically.
74///
75/// Created via [`SyncedStore::new`] or [`Node::synced_store`].
76pub struct SyncedStore<T> {
77    pub(crate) inner: Arc<StoreInner<T>>,
78    /// Channel to request outbound broadcasts from the sync task.
79    broadcast_tx: mpsc::UnboundedSender<SyncMessage>,
80    /// Handle to the background sync task.
81    task_handle: tokio::sync::Mutex<Option<JoinHandle<()>>>,
82}
83
84impl<T: Serialize + DeserializeOwned + Clone + Send + Sync + 'static> SyncedStore<T> {
85    /// Create a new SyncedStore and start the background sync task.
86    ///
87    /// The store syncs on namespace `"ss:{store_id}"`. The caller must hold
88    /// an `Arc<Node<N>>` because the background task needs to outlive this
89    /// constructor call.
90    pub fn new<N: NetworkProvider + 'static>(
91        node: Arc<Node<N>>,
92        store_id: &str,
93    ) -> Arc<Self> {
94        let device_id = node.local_info().id;
95        let (event_tx, _) = broadcast::channel(256);
96        let (broadcast_tx, broadcast_rx) = mpsc::unbounded_channel();
97
98        let inner = Arc::new(StoreInner {
99            store_id: store_id.to_string(),
100            device_id,
101            local: RwLock::new(None),
102            remotes: RwLock::new(HashMap::new()),
103            version: AtomicU64::new(0),
104            event_tx,
105        });
106
107        let task_handle = sync::spawn_sync_task(node, inner.clone(), broadcast_rx);
108
109        Arc::new(Self {
110            inner,
111            broadcast_tx,
112            task_handle: tokio::sync::Mutex::new(Some(task_handle)),
113        })
114    }
115
116    // ── Write ───────────────────────────────────────────────────────
117
118    /// Update this device's data. Increments version and broadcasts.
119    pub async fn set(&self, data: T) {
120        let version = self.inner.version.fetch_add(1, Ordering::SeqCst) + 1;
121        let now = SystemTime::now()
122            .duration_since(UNIX_EPOCH)
123            .unwrap_or_default()
124            .as_millis() as u64;
125
126        let slice = Slice {
127            device_id: self.inner.device_id.clone(),
128            data: data.clone(),
129            version,
130            updated_at: now,
131        };
132
133        {
134            let mut local = self.inner.local.write().await;
135            *local = Some(slice);
136        }
137
138        // Request the sync task to broadcast.
139        if let Ok(serialized) = serde_json::to_value(&data) {
140            let msg = SyncMessage::Update {
141                device_id: self.inner.device_id.clone(),
142                data: serialized,
143                version,
144                updated_at: now,
145            };
146            let _ = self.broadcast_tx.send(msg);
147        }
148
149        let _ = self.inner.event_tx.send(StoreEvent::LocalChanged(data));
150    }
151
152    // ── Read ────────────────────────────────────────────────────────
153
154    /// This device's current data, or `None` if `set()` hasn't been called.
155    pub async fn local(&self) -> Option<T> {
156        let local = self.inner.local.read().await;
157        local.as_ref().map(|s| s.data.clone())
158    }
159
160    /// A specific peer's slice, or `None` if we haven't received their data.
161    pub async fn get(&self, device_id: &str) -> Option<Slice<T>> {
162        let remotes = self.inner.remotes.read().await;
163        remotes.get(device_id).cloned()
164    }
165
166    /// All slices (local + remote).
167    pub async fn all(&self) -> HashMap<String, Slice<T>> {
168        let mut result = HashMap::new();
169
170        {
171            let local = self.inner.local.read().await;
172            if let Some(slice) = local.as_ref() {
173                result.insert(slice.device_id.clone(), slice.clone());
174            }
175        }
176
177        {
178            let remotes = self.inner.remotes.read().await;
179            for (id, slice) in remotes.iter() {
180                result.insert(id.clone(), slice.clone());
181            }
182        }
183
184        result
185    }
186
187    /// Device IDs that have data in this store (local + remote).
188    pub async fn device_ids(&self) -> Vec<String> {
189        let mut ids = Vec::new();
190
191        {
192            let local = self.inner.local.read().await;
193            if let Some(slice) = local.as_ref() {
194                ids.push(slice.device_id.clone());
195            }
196        }
197
198        {
199            let remotes = self.inner.remotes.read().await;
200            ids.extend(remotes.keys().cloned());
201        }
202
203        ids
204    }
205
206    /// Subscribe to store change events.
207    pub fn subscribe(&self) -> broadcast::Receiver<StoreEvent<T>> {
208        self.inner.event_tx.subscribe()
209    }
210
211    /// The store identifier.
212    pub fn store_id(&self) -> &str {
213        &self.inner.store_id
214    }
215
216    /// This device's ID.
217    pub fn device_id(&self) -> &str {
218        &self.inner.device_id
219    }
220
221    /// Current local version number.
222    pub fn version(&self) -> u64 {
223        self.inner.version.load(Ordering::SeqCst)
224    }
225
226    // ── Lifecycle ───────────────────────────────────────────────────
227
228    /// Stop the background sync task.
229    pub async fn stop(&self) {
230        let mut handle = self.task_handle.lock().await;
231        if let Some(h) = handle.take() {
232            h.abort();
233            tracing::info!(store = self.inner.store_id.as_str(), "synced_store: stopped");
234        }
235    }
236}
237
238impl<T> Drop for SyncedStore<T> {
239    fn drop(&mut self) {
240        // Best-effort abort on drop.
241        if let Ok(mut handle) = self.task_handle.try_lock() {
242            if let Some(h) = handle.take() {
243                h.abort();
244            }
245        }
246    }
247}
truffle_core/synced_store/mod.rs

truffle_core/synced_store/
mod.rs
