gonzalo_core/
ancestry.rs

1//! A [`Store`] decorator that retains each committed body in a content-addressed
2//! [`BlobStore`], keyed by its revision hash, so [`sync`](crate::sync) can fetch
3//! a divergence's common ancestor for a true 3-way merge (ADR 0016).
4
5use async_trait::async_trait;
6
7use crate::{
8    BlobStore, DeleteResult, Identity, KeyPrefix, PutResult, Record, RecordKey, Result, Revision,
9    Store,
10};
11
12/// Wraps a record [`Store`] and an ancestry [`BlobStore`]. On a committed `put`
13/// it also writes the record's `body.bytes()` to the ancestry store; because
14/// `Revision.hash == ContentHash::of(body.bytes())`, each version's body is
15/// later retrievable by its revision hash. `get`/`list` delegate unchanged.
16pub struct AncestryStore<S, B> {
17    inner: S,
18    ancestry: B,
19}
20
21impl<S, B> AncestryStore<S, B> {
22    pub fn new(inner: S, ancestry: B) -> Self {
23        Self { inner, ancestry }
24    }
25
26    /// The ancestry blob store (revision hash → body bytes), to hand to
27    /// [`sync_with_ancestry`](crate::sync::sync_with_ancestry).
28    pub fn ancestry(&self) -> &B {
29        &self.ancestry
30    }
31
32    /// The wrapped record store.
33    pub fn inner(&self) -> &S {
34        &self.inner
35    }
36}
37
38#[async_trait]
39impl<S: Store, B: BlobStore> Store for AncestryStore<S, B> {
40    async fn get(&self, key: &RecordKey) -> Result<Option<Record>> {
41        self.inner.get(key).await
42    }
43
44    async fn put(&self, record: Record, expected: Option<Revision>) -> Result<PutResult> {
45        // Retain the body under its revision hash on a successful commit, so a
46        // later divergence can be merged against this exact version.
47        let body_bytes = record.body.bytes().to_vec();
48        let outcome = self.inner.put(record, expected).await?;
49        if matches!(outcome, PutResult::Committed(_)) {
50            self.ancestry.put_blob(&body_bytes).await?;
51        }
52        Ok(outcome)
53    }
54
55    async fn list(&self, prefix: &KeyPrefix) -> Result<Vec<RecordKey>> {
56        self.inner.list(prefix).await
57    }
58
59    async fn delete_as(
60        &self,
61        key: &RecordKey,
62        expected: Option<Revision>,
63        author: Option<Identity>,
64    ) -> Result<DeleteResult> {
65        // Retained ancestry blobs stay: they may back a later divergence's
66        // 3-way merge (ADR 0016).
67        self.inner.delete_as(key, expected, author).await
68    }
69
70    async fn get_raw(&self, key: &RecordKey) -> Result<Option<Record>> {
71        self.inner.get_raw(key).await
72    }
73
74    async fn list_raw(&self, prefix: &KeyPrefix) -> Result<Vec<RecordKey>> {
75        self.inner.list_raw(prefix).await
76    }
77
78    async fn put_raw(&self, record: Record, expected: Option<Revision>) -> Result<PutResult> {
79        // Replicated bodies are retained exactly like local puts, so a later
80        // divergence can still find this version as its merge base.
81        let body_bytes = record.body.bytes().to_vec();
82        let outcome = self.inner.put_raw(record, expected).await?;
83        if matches!(outcome, PutResult::Committed(_)) {
84            self.ancestry.put_blob(&body_bytes).await?;
85        }
86        Ok(outcome)
87    }
88
89    async fn purge(&self, key: &RecordKey, expected: Revision) -> Result<DeleteResult> {
90        // Retained ancestry bodies stay: they are content-addressed and may
91        // back other records' merges (ADR 0016).
92        self.inner.purge(key, expected).await
93    }
94}
95
96#[cfg(test)]
97pub(crate) mod tests {
98    use super::*;
99    use crate::store::Conflict;
100    use crate::{
101        BlobEntry, Body, ContentHash, CoreError, Identity, Meta, RecordKind, revision::Revision,
102    };
103    use std::collections::BTreeMap;
104    use std::sync::Mutex;
105
106    /// An in-memory `Store` + `BlobStore` double.
107    #[derive(Default)]
108    pub(crate) struct Mem {
109        records: Mutex<BTreeMap<RecordKey, Record>>,
110        blobs: Mutex<BTreeMap<ContentHash, Vec<u8>>>,
111    }
112
113    #[async_trait]
114    impl Store for Mem {
115        async fn get(&self, key: &RecordKey) -> Result<Option<Record>> {
116            Ok(self.records.lock().unwrap().get(key).cloned())
117        }
118        async fn put(&self, record: Record, expected: Option<Revision>) -> Result<PutResult> {
119            let mut g = self.records.lock().unwrap();
120            let current = g.get(&record.key).map(|r| r.revision.clone());
121            if current != expected {
122                if let Some(cur) = g.get(&record.key).cloned() {
123                    return Ok(PutResult::Conflict(Box::new(Conflict {
124                        key: record.key.clone(),
125                        expected,
126                        current: cur,
127                    })));
128                }
129                return Err(CoreError::NotFound(record.key.clone()));
130            }
131            let rev = record.revision.clone();
132            g.insert(record.key.clone(), record);
133            Ok(PutResult::Committed(rev))
134        }
135        async fn list(&self, prefix: &KeyPrefix) -> Result<Vec<RecordKey>> {
136            Ok(self
137                .records
138                .lock()
139                .unwrap()
140                .keys()
141                .filter(|k| prefix.matches(k))
142                .cloned()
143                .collect())
144        }
145        async fn delete_as(
146            &self,
147            key: &RecordKey,
148            expected: Option<Revision>,
149            _author: Option<Identity>,
150        ) -> Result<DeleteResult> {
151            let mut g = self.records.lock().unwrap();
152            match g.get(key) {
153                None => Ok(DeleteResult::Deleted),
154                Some(cur) if expected.is_none() || expected.as_ref() == Some(&cur.revision) => {
155                    g.remove(key);
156                    Ok(DeleteResult::Deleted)
157                }
158                Some(cur) => Ok(DeleteResult::Conflict(Box::new(Conflict {
159                    key: key.clone(),
160                    expected,
161                    current: cur.clone(),
162                }))),
163            }
164        }
165        async fn put_raw(&self, record: Record, expected: Option<Revision>) -> Result<PutResult> {
166            <Self as Store>::put(self, record, expected).await
167        }
168        async fn get_raw(&self, key: &RecordKey) -> Result<Option<Record>> {
169            self.get(key).await
170        }
171        async fn list_raw(&self, prefix: &KeyPrefix) -> Result<Vec<RecordKey>> {
172            self.list(prefix).await
173        }
174        async fn purge(&self, key: &RecordKey, expected: Revision) -> Result<DeleteResult> {
175            self.delete(key, Some(expected)).await
176        }
177    }
178
179    #[async_trait]
180    impl BlobStore for Mem {
181        async fn put_blob(&self, content: &[u8]) -> Result<ContentHash> {
182            let hash = ContentHash::of(content);
183            self.blobs
184                .lock()
185                .unwrap()
186                .insert(hash.clone(), content.to_vec());
187            Ok(hash)
188        }
189        async fn get_blob(&self, hash: &ContentHash) -> Result<Option<Vec<u8>>> {
190            Ok(self.blobs.lock().unwrap().get(hash).cloned())
191        }
192        async fn list_blobs(&self) -> Result<Vec<BlobEntry>> {
193            Ok(self
194                .blobs
195                .lock()
196                .unwrap()
197                .keys()
198                .cloned()
199                .map(|h| BlobEntry::from_system_time(h, std::time::SystemTime::now()))
200                .collect())
201        }
202        async fn delete_blob(&self, hash: &ContentHash) -> Result<()> {
203            self.blobs.lock().unwrap().remove(hash);
204            Ok(())
205        }
206    }
207
208    pub(crate) fn rec(id: &str, kind: RecordKind, payload: &str) -> Record {
209        let body = Body::Inline(payload.as_bytes().to_vec());
210        Record {
211            revision: Revision::initial(body.bytes()),
212            parent: None,
213            body,
214            kind,
215            key: RecordKey::new("ns", "col", id),
216            meta: Meta {
217                author: Identity::new("t"),
218                origin_system: "test".into(),
219                created: 0,
220                updated: 0,
221                labels: BTreeMap::new(),
222            },
223            links: Vec::new(),
224            ancestors: Vec::new(),
225            deleted_at: None,
226            deleted_blob: None,
227            deleted_kind: None,
228        }
229    }
230
231    #[tokio::test]
232    async fn put_retains_body_under_revision_hash_and_delegates() {
233        let store = AncestryStore::new(Mem::default(), Mem::default());
234        let r = rec("k", RecordKind::MemoryTier, r#"{"a":1}"#);
235        let rev = r.revision.clone();
236        assert!(matches!(
237            store.put(r.clone(), None).await.unwrap(),
238            PutResult::Committed(_)
239        ));
240
241        // The body is retrievable from the ancestry store by its revision hash.
242        let retained = store.ancestry().get_blob(&rev.hash).await.unwrap();
243        assert_eq!(retained.as_deref(), Some(r.body.bytes()));
244
245        // get/list delegate to the wrapped record store.
246        assert_eq!(store.get(&r.key).await.unwrap().unwrap().revision, rev);
247        assert_eq!(
248            store.list(&KeyPrefix::default()).await.unwrap(),
249            vec![r.key]
250        );
251    }
252
253    #[tokio::test]
254    async fn conflicting_put_does_not_retain() {
255        let store = AncestryStore::new(Mem::default(), Mem::default());
256        let r = rec("k", RecordKind::MemoryTier, r#"{"a":1}"#);
257        let _ = store.put(r.clone(), None).await.unwrap();
258
259        // A stale write (wrong `expected`) conflicts and must not retain.
260        let other = rec("k", RecordKind::MemoryTier, r#"{"a":2}"#);
261        let other_rev = other.revision.clone();
262        assert!(matches!(
263            store.put(other, None).await.unwrap(),
264            PutResult::Conflict(_)
265        ));
266        assert!(
267            store
268                .ancestry()
269                .get_blob(&other_rev.hash)
270                .await
271                .unwrap()
272                .is_none()
273        );
274    }
275}
gonzalo_core/ancestry.rs

gonzalo_core/
ancestry.rs
