sapphire_framework_sync/
store.rs

1//! Replica store (redb): one table of path states and one metadata record.
2
3use std::path::Path;
4
5use redb::{Database, ReadableDatabase, ReadableTable, TableDefinition};
6use serde::{Deserialize, Serialize};
7
8use crate::error::{Error, RedbExt, Result};
9use crate::hlc::Hlc;
10use crate::id::ReplicaId;
11use crate::state::PathState;
12use crate::vv::VersionVector;
13
14/// On-disk format of the replica store. Bump for any change in stored data *or* in
15/// merge behaviour (spec §5.4).
16pub const FORMAT_VERSION: u32 = 1;
17
18const PATHS: TableDefinition<&str, &[u8]> = TableDefinition::new("paths");
19const META: TableDefinition<&str, &[u8]> = TableDefinition::new("meta");
20const META_KEY: &str = "meta";
21
22/// Replica-wide metadata.
23#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
24pub struct Meta {
25    pub format_version: u32,
26    pub replica_id: ReplicaId,
27    /// Counter of the last dot this replica assigned.
28    pub counter: u64,
29    pub hlc: Hlc,
30    /// Every dot whose effects this replica's states include.
31    pub vv: VersionVector,
32    /// Workspace root this store belongs to.
33    pub root: String,
34}
35
36/// Persistent replica state.
37pub struct ReplicaStore {
38    db: Database,
39}
40
41#[derive(Deserialize)]
42struct FormatProbe {
43    format_version: u32,
44}
45
46impl ReplicaStore {
47    pub fn open(path: &Path, root: &str) -> Result<Self> {
48        Self::open_with_id(path, root, None)
49    }
50
51    /// Open or create the store. `id` is used only when the store is created.
52    pub fn open_with_id(path: &Path, root: &str, id: Option<ReplicaId>) -> Result<Self> {
53        if let Some(parent) = path.parent() {
54            std::fs::create_dir_all(parent)?;
55        }
56        let db = Database::create(path).db()?;
57        let wtx = db.begin_write().db()?;
58        {
59            wtx.open_table(PATHS).db()?;
60            let mut meta_table = wtx.open_table(META).db()?;
61            let existing = meta_table.get(META_KEY).db()?.map(|g| g.value().to_vec());
62            match existing {
63                None => {
64                    let meta = Meta {
65                        format_version: FORMAT_VERSION,
66                        replica_id: id.unwrap_or_default(),
67                        counter: 0,
68                        hlc: Hlc::default(),
69                        vv: VersionVector::new(),
70                        root: root.to_owned(),
71                    };
72                    meta_table
73                        .insert(META_KEY, serde_json::to_vec(&meta)?.as_slice())
74                        .db()?;
75                }
76                Some(bytes) => {
77                    let probe: FormatProbe = serde_json::from_slice(&bytes)?;
78                    if probe.format_version > FORMAT_VERSION {
79                        return Err(Error::FormatTooNew {
80                            found: probe.format_version,
81                            supported: FORMAT_VERSION,
82                        });
83                    }
84                    let meta: Meta = serde_json::from_slice(&bytes)?;
85                    if meta.root != root {
86                        return Err(Error::RootMismatch {
87                            stored: meta.root,
88                            requested: root.to_owned(),
89                        });
90                    }
91                }
92            }
93        }
94        wtx.commit().db()?;
95        Ok(Self { db })
96    }
97
98    pub fn meta(&self) -> Result<Meta> {
99        let rtx = self.db.begin_read().db()?;
100        let table = rtx.open_table(META).db()?;
101        let guard = table
102            .get(META_KEY)
103            .db()?
104            .ok_or_else(|| Error::Corrupt("missing meta record".into()))?;
105        Ok(serde_json::from_slice(guard.value())?)
106    }
107
108    pub fn get(&self, path: &str) -> Result<Option<PathState>> {
109        let rtx = self.db.begin_read().db()?;
110        let table = rtx.open_table(PATHS).db()?;
111        match table.get(path).db()? {
112            Some(guard) => Ok(Some(serde_json::from_slice(guard.value())?)),
113            None => Ok(None),
114        }
115    }
116
117    pub fn all(&self) -> Result<Vec<(String, PathState)>> {
118        let rtx = self.db.begin_read().db()?;
119        let table = rtx.open_table(PATHS).db()?;
120        let mut out = Vec::new();
121        for item in table.iter().db()? {
122            let (k, v) = item.db()?;
123            out.push((k.value().to_owned(), serde_json::from_slice(v.value())?));
124        }
125        Ok(out)
126    }
127
128    pub fn any_materialized(&self) -> Result<bool> {
129        Ok(self.all()?.iter().any(|(_, s)| s.disk.hash.is_some()))
130    }
131
132    /// Write `meta` and `states` in one transaction.
133    pub fn commit(&self, meta: &Meta, states: &[(String, PathState)]) -> Result<()> {
134        let wtx = self.db.begin_write().db()?;
135        {
136            let mut meta_table = wtx.open_table(META).db()?;
137            meta_table
138                .insert(META_KEY, serde_json::to_vec(meta)?.as_slice())
139                .db()?;
140            let mut paths = wtx.open_table(PATHS).db()?;
141            for (path, state) in states {
142                paths
143                    .insert(path.as_str(), serde_json::to_vec(state)?.as_slice())
144                    .db()?;
145            }
146        }
147        wtx.commit().db()?;
148        Ok(())
149    }
150}
151
152#[cfg(test)]
153mod tests {
154    use super::*;
155    use crate::entry::{Content, Entry};
156    use crate::hash::ContentHash;
157    use crate::state::DiskState;
158    use crate::vv::Dot;
159    use grain_id::GrainId;
160
161    fn state(id: ReplicaId, body: &str, on_disk: bool) -> PathState {
162        let e = Entry {
163            path: "a.txt".into(),
164            content: Content::File {
165                hash: ContentHash::of_bytes(body.as_bytes()),
166                len: 1,
167            },
168            hlc: Hlc::default(),
169            dot: Dot {
170                replica: id,
171                counter: 1,
172            },
173            context: VersionVector::new(),
174            author: GrainId::NIL,
175        };
176        let mut seen = VersionVector::new();
177        seen.add_dot(&e.dot);
178        let disk = DiskState {
179            hash: on_disk.then(|| ContentHash::of_bytes(body.as_bytes())),
180            ..DiskState::default()
181        };
182        PathState {
183            versions: vec![e],
184            seen,
185            disk,
186        }
187    }
188
189    #[test]
190    fn creates_meta_once_and_persists_commits() {
191        let dir = tempfile::tempdir().unwrap();
192        let path = dir.path().join("sync.redb");
193        let (id, s) = {
194            let store = ReplicaStore::open(&path, "/root").unwrap();
195            let mut meta = store.meta().unwrap();
196            assert_eq!(meta.format_version, FORMAT_VERSION);
197            assert_eq!(meta.counter, 0);
198            meta.counter = 7;
199            let s = state(meta.replica_id, "x", false);
200            store.commit(&meta, &[("a.txt".into(), s.clone())]).unwrap();
201            (meta.replica_id, s)
202        };
203        let store = ReplicaStore::open(&path, "/root").unwrap();
204        let meta = store.meta().unwrap();
205        assert_eq!((meta.replica_id, meta.counter), (id, 7));
206        assert_eq!(store.get("a.txt").unwrap(), Some(s.clone()));
207        assert_eq!(store.get("b.txt").unwrap(), None);
208        assert_eq!(store.all().unwrap(), vec![("a.txt".to_string(), s)]);
209        assert!(!store.any_materialized().unwrap());
210    }
211
212    #[test]
213    fn any_materialized_sees_files_on_disk() {
214        let dir = tempfile::tempdir().unwrap();
215        let store = ReplicaStore::open(&dir.path().join("s.redb"), "/r").unwrap();
216        let meta = store.meta().unwrap();
217        store
218            .commit(
219                &meta,
220                &[("a.txt".into(), state(meta.replica_id, "x", true))],
221            )
222            .unwrap();
223        assert!(store.any_materialized().unwrap());
224    }
225
226    #[test]
227    fn refuses_another_root() {
228        let dir = tempfile::tempdir().unwrap();
229        let path = dir.path().join("s.redb");
230        drop(ReplicaStore::open(&path, "/one").unwrap());
231        assert!(matches!(
232            ReplicaStore::open(&path, "/two"),
233            Err(Error::RootMismatch { .. })
234        ));
235    }
236
237    #[test]
238    fn refuses_a_newer_format() {
239        let dir = tempfile::tempdir().unwrap();
240        let path = dir.path().join("s.redb");
241        {
242            let store = ReplicaStore::open(&path, "/r").unwrap();
243            let mut meta = store.meta().unwrap();
244            meta.format_version = FORMAT_VERSION + 1;
245            store.commit(&meta, &[]).unwrap();
246        }
247        assert!(matches!(
248            ReplicaStore::open(&path, "/r"),
249            Err(Error::FormatTooNew { found, .. }) if found == FORMAT_VERSION + 1
250        ));
251    }
252
253    #[test]
254    fn open_with_id_uses_the_id_only_on_creation() {
255        let dir = tempfile::tempdir().unwrap();
256        let path = dir.path().join("s.redb");
257        let fixed = ReplicaId(uuid::Uuid::from_u128(5));
258        let store = ReplicaStore::open_with_id(&path, "/r", Some(fixed)).unwrap();
259        assert_eq!(store.meta().unwrap().replica_id, fixed);
260        drop(store);
261        let other = ReplicaId(uuid::Uuid::from_u128(6));
262        let store = ReplicaStore::open_with_id(&path, "/r", Some(other)).unwrap();
263        assert_eq!(store.meta().unwrap().replica_id, fixed);
264    }
265}
sapphire_framework_sync/store.rs

sapphire_framework_sync/
store.rs
