logicaffeine_system/storage/
mod.rs

1//! Persistent Storage with Journaling
2//!
3//! Provides crash-resilient persistence for CRDTs using an append-only journal
4//! with automatic replay on mount and compaction support.
5//!
6//! # Architecture
7//!
8//! The journal uses a simple binary format:
9//!
10//! ```text
11//! ┌─────────────┬─────────────┬─────────────────┐
12//! │ Length (4B) │ CRC32 (4B)  │ Payload (N B)   │
13//! └─────────────┴─────────────┴─────────────────┘
14//! ```
15//!
16//! Entries are either:
17//! - **Snapshot**: Full state replacement (written during compaction)
18//! - **Delta**: Incremental update (written on each mutation)
19//!
20//! # Recovery
21//!
22//! On mount, the journal is replayed entry-by-entry:
23//! - Snapshots replace the current state
24//! - Deltas are merged via the CRDT's `Merge` trait
25//! - Truncated entries are ignored (WAL semantics)
26//! - Checksum failures return an error
27//!
28//! # Features
29//!
30//! Requires the `persistence` feature.
31//!
32//! # Example
33//!
34//! ```no_run
35//! use logicaffeine_system::storage::Persistent;
36//! use logicaffeine_data::crdt::GCounter;
37//! # use logicaffeine_system::fs::NativeVfs;
38//! # use std::sync::Arc;
39//!
40//! # fn main() {}
41//! # async fn example() -> Result<(), Box<dyn std::error::Error>> {
42//! let vfs: Arc<dyn logicaffeine_system::fs::Vfs + Send + Sync> = Arc::new(NativeVfs::new("/data"));
43//! let counter = Persistent::<GCounter>::mount(vfs, "counter.lsf").await?;
44//!
45//! counter.mutate(|c| c.increment(5)).await?;
46//! counter.compact().await?; // Reduce journal size
47//! # Ok(())
48//! # }
49//! ```
50
51use logicaffeine_data::crdt::Merge;
52use crate::fs::{Vfs, VfsResult, VfsError};
53use serde::{de::DeserializeOwned, Serialize};
54use std::marker::PhantomData;
55use std::sync::Arc;
56use async_lock::Mutex;
57
58/// Operation recorded in the journal.
59#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
60enum JournalOp<T> {
61    /// Full state snapshot (for compaction)
62    Snapshot(T),
63    /// Delta operation (for incremental updates)
64    Delta(T),
65}
66
67/// Journal entry header format:
68/// [4 bytes: length][4 bytes: crc32][N bytes: payload]
69struct JournalHeader;
70
71impl JournalHeader {
72    const SIZE: usize = 8;
73
74    fn encode(payload: &[u8]) -> [u8; Self::SIZE] {
75        let length = payload.len() as u32;
76        let checksum = crc32fast::hash(payload);
77        let mut buf = [0u8; Self::SIZE];
78        buf[0..4].copy_from_slice(&length.to_le_bytes());
79        buf[4..8].copy_from_slice(&checksum.to_le_bytes());
80        buf
81    }
82
83    fn decode(buf: &[u8; Self::SIZE]) -> (u32, u32) {
84        let length = u32::from_le_bytes([buf[0], buf[1], buf[2], buf[3]]);
85        let checksum = u32::from_le_bytes([buf[4], buf[5], buf[6], buf[7]]);
86        (length, checksum)
87    }
88}
89
90/// A persistent CRDT wrapper with journaling (native platform).
91///
92/// Wraps any type implementing `Merge + Serialize + DeserializeOwned`
93/// and provides durable storage with crash recovery.
94///
95/// This is the native version with `Send` bounds required for thread-safe access.
96/// The WASM version relaxes these bounds since JavaScript is single-threaded.
97#[cfg(not(target_arch = "wasm32"))]
98pub struct Persistent<T>
99where
100    T: Merge + Serialize + DeserializeOwned + Clone + Default + Send + 'static,
101{
102    inner: Arc<Mutex<T>>,
103    vfs: Arc<dyn Vfs + Send + Sync>,
104    journal_path: String,
105    entry_count: Arc<Mutex<u64>>,
106    _marker: PhantomData<T>,
107}
108
109/// WASM version without Send bounds (JS is single-threaded).
110#[cfg(target_arch = "wasm32")]
111pub struct Persistent<T>
112where
113    T: Merge + Serialize + DeserializeOwned + Clone + Default + 'static,
114{
115    inner: Arc<Mutex<T>>,
116    vfs: Arc<dyn Vfs>,
117    journal_path: String,
118    entry_count: Arc<Mutex<u64>>,
119    _marker: PhantomData<T>,
120}
121
122/// Helper function to replay journal entries (shared between platforms).
123fn replay_journal<T>(data: &[u8]) -> Result<(T, u64), VfsError>
124where
125    T: Merge + Serialize + DeserializeOwned + Clone + Default,
126{
127    let mut state = T::default();
128    let mut entry_count = 0u64;
129    let mut pos = 0;
130
131    while pos + JournalHeader::SIZE <= data.len() {
132        let header_bytes: [u8; 8] = data[pos..pos + 8].try_into().unwrap();
133        let (length, expected_checksum) = JournalHeader::decode(&header_bytes);
134        pos += JournalHeader::SIZE;
135
136        let payload_end = pos + length as usize;
137        if payload_end > data.len() {
138            // Truncated entry - stop replay (WAL semantics)
139            break;
140        }
141
142        let payload = &data[pos..payload_end];
143        let actual_checksum = crc32fast::hash(payload);
144
145        if actual_checksum != expected_checksum {
146            return Err(VfsError::JournalCorrupted(
147                format!("Entry {} checksum mismatch", entry_count)
148            ));
149        }
150
151        let op: JournalOp<T> = bincode::deserialize(payload)
152            .map_err(|e| VfsError::SerializationError(e.to_string()))?;
153
154        match op {
155            JournalOp::Snapshot(s) => state = s,
156            JournalOp::Delta(d) => state.merge(&d),
157        }
158
159        pos = payload_end;
160        entry_count += 1;
161    }
162
163    Ok((state, entry_count))
164}
165
166/// Native implementation with Send bounds.
167#[cfg(not(target_arch = "wasm32"))]
168impl<T> Persistent<T>
169where
170    T: Merge + Serialize + DeserializeOwned + Clone + Default + Send + 'static,
171{
172    /// Mount a persistent value from a journal file.
173    ///
174    /// If the journal exists, replays all entries to reconstruct state.
175    /// If not, creates a new journal with default state.
176    ///
177        pub async fn mount(vfs: Arc<dyn Vfs + Send + Sync>, path: &str) -> VfsResult<Self> {
178        let (state, entry_count) = if vfs.exists(path).await.unwrap_or(false) {
179            let data = vfs.read(path).await?;
180            replay_journal(&data)?
181        } else {
182            (T::default(), 0)
183        };
184
185        Ok(Self {
186            inner: Arc::new(Mutex::new(state)),
187            vfs,
188            journal_path: path.to_string(),
189            entry_count: Arc::new(Mutex::new(entry_count)),
190            _marker: PhantomData,
191        })
192    }
193
194    /// Get immutable access to the current state.
195    pub async fn get(&self) -> T {
196        self.inner.lock().await.clone()
197    }
198
199    /// Mutate the state and persist the delta.
200    pub async fn mutate<F, R>(&self, f: F) -> VfsResult<R>
201    where
202        F: FnOnce(&mut T) -> R + Send,
203    {
204        let mut guard = self.inner.lock().await;
205        let result = f(&mut *guard);
206
207        let op = JournalOp::Delta(guard.clone());
208        let payload = bincode::serialize(&op)
209            .map_err(|e| VfsError::SerializationError(e.to_string()))?;
210
211        let header = JournalHeader::encode(&payload);
212        let mut entry = Vec::with_capacity(JournalHeader::SIZE + payload.len());
213        entry.extend_from_slice(&header);
214        entry.extend_from_slice(&payload);
215
216        self.vfs.append(&self.journal_path, &entry).await?;
217        *self.entry_count.lock().await += 1;
218
219        Ok(result)
220    }
221
222    /// Compact the journal by writing a snapshot.
223    pub async fn compact(&self) -> VfsResult<()> {
224        let state = self.inner.lock().await.clone();
225
226        let op = JournalOp::<T>::Snapshot(state);
227        let payload = bincode::serialize(&op)
228            .map_err(|e| VfsError::SerializationError(e.to_string()))?;
229
230        let header = JournalHeader::encode(&payload);
231        let mut entry = Vec::with_capacity(JournalHeader::SIZE + payload.len());
232        entry.extend_from_slice(&header);
233        entry.extend_from_slice(&payload);
234
235        self.vfs.write(&self.journal_path, &entry).await?;
236        *self.entry_count.lock().await = 1;
237
238        Ok(())
239    }
240
241    /// Get the number of journal entries.
242    pub async fn entry_count(&self) -> u64 {
243        *self.entry_count.lock().await
244    }
245
246    /// Automatically compact when entry count exceeds threshold.
247    pub async fn maybe_compact(&self, threshold: u64) -> VfsResult<bool> {
248        if self.entry_count().await > threshold {
249            self.compact().await?;
250            Ok(true)
251        } else {
252            Ok(false)
253        }
254    }
255}
256
257/// WASM implementation without Send bounds (single-threaded).
258#[cfg(target_arch = "wasm32")]
259impl<T> Persistent<T>
260where
261    T: Merge + Serialize + DeserializeOwned + Clone + Default + 'static,
262{
263    /// Mount a persistent value from a journal file.
264    pub async fn mount(vfs: Arc<dyn Vfs>, path: &str) -> VfsResult<Self> {
265        let (state, entry_count) = if vfs.exists(path).await.unwrap_or(false) {
266            let data = vfs.read(path).await?;
267            replay_journal(&data)?
268        } else {
269            (T::default(), 0)
270        };
271
272        Ok(Self {
273            inner: Arc::new(Mutex::new(state)),
274            vfs,
275            journal_path: path.to_string(),
276            entry_count: Arc::new(Mutex::new(entry_count)),
277            _marker: PhantomData,
278        })
279    }
280
281    /// Get immutable access to the current state.
282    pub async fn get(&self) -> T {
283        self.inner.lock().await.clone()
284    }
285
286    /// Mutate the state and persist the delta.
287    pub async fn mutate<F, R>(&self, f: F) -> VfsResult<R>
288    where
289        F: FnOnce(&mut T) -> R,
290    {
291        let mut guard = self.inner.lock().await;
292        let result = f(&mut *guard);
293
294        let op = JournalOp::Delta(guard.clone());
295        let payload = bincode::serialize(&op)
296            .map_err(|e| VfsError::SerializationError(e.to_string()))?;
297
298        let header = JournalHeader::encode(&payload);
299        let mut entry = Vec::with_capacity(JournalHeader::SIZE + payload.len());
300        entry.extend_from_slice(&header);
301        entry.extend_from_slice(&payload);
302
303        self.vfs.append(&self.journal_path, &entry).await?;
304        *self.entry_count.lock().await += 1;
305
306        Ok(result)
307    }
308
309    /// Compact the journal by writing a snapshot.
310    pub async fn compact(&self) -> VfsResult<()> {
311        let state = self.inner.lock().await.clone();
312
313        let op = JournalOp::<T>::Snapshot(state);
314        let payload = bincode::serialize(&op)
315            .map_err(|e| VfsError::SerializationError(e.to_string()))?;
316
317        let header = JournalHeader::encode(&payload);
318        let mut entry = Vec::with_capacity(JournalHeader::SIZE + payload.len());
319        entry.extend_from_slice(&header);
320        entry.extend_from_slice(&payload);
321
322        self.vfs.write(&self.journal_path, &entry).await?;
323        *self.entry_count.lock().await = 1;
324
325        Ok(())
326    }
327
328    /// Get the number of journal entries.
329    pub async fn entry_count(&self) -> u64 {
330        *self.entry_count.lock().await
331    }
332
333    /// Automatically compact when entry count exceeds threshold.
334    pub async fn maybe_compact(&self, threshold: u64) -> VfsResult<bool> {
335        if self.entry_count().await > threshold {
336            self.compact().await?;
337            Ok(true)
338        } else {
339            Ok(false)
340        }
341    }
342}
343
344// The storage module directly accepts Arc<dyn Vfs> for flexibility.
345
346#[cfg(test)]
347mod tests {
348    use super::*;
349    use logicaffeine_data::crdt::GCounter;
350    use tempfile::TempDir;
351
352    #[tokio::test]
353    async fn test_persistent_mount_empty() {
354        let temp = TempDir::new().unwrap();
355        let vfs: Arc<dyn Vfs + Send + Sync> = Arc::new(crate::fs::NativeVfs::new(temp.path()));
356
357        let counter = Persistent::<GCounter>::mount(vfs, "counter.journal").await.unwrap();
358
359        assert_eq!(counter.get().await.value(), 0);
360    }
361
362    #[tokio::test]
363    async fn test_persistent_mutate() {
364        let temp = TempDir::new().unwrap();
365        let vfs: Arc<dyn Vfs + Send + Sync> = Arc::new(crate::fs::NativeVfs::new(temp.path()));
366
367        let counter = Persistent::<GCounter>::mount(vfs, "counter.journal").await.unwrap();
368
369        counter.mutate(|c| c.increment(5)).await.unwrap();
370        counter.mutate(|c| c.increment(3)).await.unwrap();
371
372        assert_eq!(counter.get().await.value(), 8);
373    }
374}
logicaffeine_system/storage/mod.rs

logicaffeine_system/storage/
mod.rs
