diff options
Diffstat (limited to 'aero-collections')
| -rw-r--r-- | aero-collections/src/davdag.rs | 174 |
1 files changed, 113 insertions, 61 deletions
diff --git a/aero-collections/src/davdag.rs b/aero-collections/src/davdag.rs index 59dcc7b..63a76a8 100644 --- a/aero-collections/src/davdag.rs +++ b/aero-collections/src/davdag.rs @@ -4,14 +4,18 @@ use im::{OrdMap, OrdSet, ordset}; use aero_bayou::*; -use crate::unique_ident::UniqueIdent; +use crate::unique_ident::{gen_ident, UniqueIdent}; /// Parents are only persisted in the event log, /// not in the checkpoints. -pub type Parents = Vec<UniqueIdent>; +pub type Token = UniqueIdent; +pub type Parents = Vec<Token>; +pub type SyncDesc = (Parents, Token); + +pub type BlobId = UniqueIdent; pub type Etag = String; pub type FileName = String; -pub type IndexEntry = (FileName, Etag); +pub type IndexEntry = (BlobId, FileName, Etag); #[derive(Clone, Default)] pub struct DavDag { @@ -22,8 +26,6 @@ pub struct DavDag { pub idx_by_filename: OrdMap<FileName, UniqueIdent>, /// Partial synchronization graph - /// parent -> direct children - pub successors: OrdMap<UniqueIdent, OrdSet<UniqueIdent>>, pub ancestors: OrdMap<UniqueIdent, OrdSet<UniqueIdent>>, /// All nodes @@ -37,33 +39,46 @@ pub struct DavDag { #[derive(Clone, Serialize, Deserialize, Debug)] pub enum DavDagOp { /// Merge is a virtual operation run when multiple heads are discovered - Merge(Parents, UniqueIdent), + Merge(SyncDesc), /// Add an item to the collection - Put(Parents, UniqueIdent, IndexEntry), + Put(SyncDesc, IndexEntry), /// Delete an item from the collection - Delete(Parents, UniqueIdent), + Delete(SyncDesc, BlobId), +} +impl DavDagOp { + pub fn token(&self) -> Token { + match self { + Self::Merge((_, t)) => *t, + Self::Put((_, t), _) => *t, + Self::Delete((_, t), _) => *t, + } + } } impl DavDag { - pub fn op_merge(&self, ident: UniqueIdent) -> DavDagOp { - DavDagOp::Merge(self.heads_vec(), ident) + pub fn op_merge(&self) -> DavDagOp { + DavDagOp::Merge(self.sync_desc()) } - pub fn op_put(&self, ident: UniqueIdent, entry: IndexEntry) -> DavDagOp { - DavDagOp::Put(self.heads_vec(), ident, entry) + pub fn op_put(&self, entry: IndexEntry) -> DavDagOp { + DavDagOp::Put(self.sync_desc(), entry) } - pub fn op_delete(&self, ident: UniqueIdent) -> DavDagOp { - DavDagOp::Delete(self.heads_vec(), ident) + pub fn op_delete(&self, ident: BlobId) -> DavDagOp { + DavDagOp::Delete(self.sync_desc(), ident) } // HELPER functions - /// All HEAD events - pub fn heads_vec(&self) -> Vec<UniqueIdent> { - self.heads.clone().into_iter().collect() + pub fn heads_vec(&self) -> Vec<Token> { + self.heads.clone().into_iter().collect() + } + + /// A sync descriptor + pub fn sync_desc(&self) -> SyncDesc { + (self.heads_vec(), gen_ident()) } /// Resolve a sync token @@ -71,18 +86,21 @@ impl DavDag { let already_known = self.all_ancestors(known); // We can't capture all missing events if we are not connected - // to all sinks of the graph, ie. if we don't already know all the sinks. + // to all sinks of the graph, + // ie. if we don't already know all the sinks, + // ie. if we are missing so much history that + // the event log has been transformed into a checkpoint if !self.origins.is_subset(already_known.clone()) { bail!("Not enough history to produce a correct diff, a full resync is needed"); } - // Missing items are all existing graph items from which - // we removed all items known by the given node. - // In other words, all values in all_nodes that are not in already_known. + // Missing items are *all existing graph items* from which + // we removed *all items known by the given node*. + // In other words, all values in `all_nodes` that are not in `already_known`. Ok(self.all_nodes.clone().relative_complement(already_known)) } - /// Find all ancestors of a given + /// Find all ancestors of a given node fn all_ancestors(&self, known: UniqueIdent) -> OrdSet<UniqueIdent> { let mut all_known: OrdSet<UniqueIdent> = OrdSet::new(); let mut to_collect = vec![known]; @@ -111,21 +129,23 @@ impl DavDag { // INTERNAL functions /// Register a WebDAV item (put, copy, move) - fn register(&mut self, ident: UniqueIdent, entry: IndexEntry) { + fn register(&mut self, entry: IndexEntry) { + let (blob_id, filename, _etag) = entry.clone(); + // Insert item in the source of trust - self.table.insert(ident, entry.clone()); + self.table.insert(blob_id, entry); // Update the cache - let (filename, _etag) = entry; - self.idx_by_filename.insert(filename, ident); + self.idx_by_filename.insert(filename, blob_id); } /// Unregister a WebDAV item (delete, move) fn unregister(&mut self, ident: &UniqueIdent) { // Query the source of truth to get the information we // need to clean the indexes - let (filename, _etag) = match self.table.get(ident) { + let (_blob_id, filename, _etag) = match self.table.get(ident) { Some(v) => v, + // Element does not exist, return early None => return, }; self.idx_by_filename.remove(filename); @@ -134,29 +154,9 @@ impl DavDag { self.table.remove(ident); } - // @FIXME: maybe in case of error we could simply disable the sync graph - // and ask the client to rely on manual sync. For now, we are skipping the event - // which is midly satisfying. - /// When an event is processed, update the synchronization DAG - fn sync_dag(&mut self, child: &UniqueIdent, parents: &[UniqueIdent]) -> bool { - // --- Update SUCCESSORS - // All parents must exist in successors otherwise we can't accept item: - // do the check + update successors - let mut try_successors = self.successors.clone(); - for par in parents.iter() { - match try_successors.get_mut(par) { - None => { - tracing::warn!("Unable to push a Dav DAG sync op into the graph, an event is missing, it's a bug"); - return false - }, - Some(v) => v.insert(*child), - }; - } - self.successors = try_successors; - - // This event is also a future successor - self.successors.insert(*child, ordset![]); + fn sync_dag(&mut self, sync_desc: &SyncDesc) -> bool { + let (parents, child) = sync_desc; // --- Update ANCESTORS // We register ancestors as it is required for the sync algorithm @@ -177,6 +177,9 @@ impl DavDag { // This event becomes a new HEAD in turn self.heads.insert(*child); + + // --- Update ALL NODES + self.all_nodes.insert(*child); true } @@ -189,18 +192,18 @@ impl BayouState for DavDag { let mut new = self.clone(); match op { - DavDagOp::Put(parents, ident, entry) => { - if new.sync_dag(ident, parents.as_slice()) { - new.register(*ident, entry.clone()); + DavDagOp::Put(sync_desc, entry) => { + if new.sync_dag(sync_desc) { + new.register(entry.clone()); } }, - DavDagOp::Delete(parents, ident) => { - if new.sync_dag(ident, parents.as_slice()) { - new.unregister(ident); + DavDagOp::Delete(sync_desc, blob_id) => { + if new.sync_dag(sync_desc) { + new.unregister(blob_id); } }, - DavDagOp::Merge(parents, ident) => { - new.sync_dag(ident, parents.as_slice()); + DavDagOp::Merge(sync_desc) => { + new.sync_dag(sync_desc); } } @@ -211,7 +214,7 @@ impl BayouState for DavDag { // CUSTOM SERIALIZATION & DESERIALIZATION #[derive(Serialize, Deserialize)] struct DavDagSerializedRepr { - items: Vec<(UniqueIdent, IndexEntry)>, + items: Vec<IndexEntry>, heads: Vec<UniqueIdent>, } @@ -224,11 +227,10 @@ impl<'de> Deserialize<'de> for DavDag { let mut davdag = DavDag::default(); // Build the table + index - val.items.into_iter().for_each(|(ident, entry)| davdag.register(ident, entry)); + val.items.into_iter().for_each(|entry| davdag.register(entry)); // Initialize the synchronization DAG with its roots val.heads.into_iter().for_each(|ident| { - davdag.successors.insert(ident, ordset![]); davdag.heads.insert(ident); davdag.origins.insert(ident); davdag.all_nodes.insert(ident); @@ -244,7 +246,7 @@ impl Serialize for DavDag { S: Serializer, { // Indexes are rebuilt on the fly, we serialize only the core database - let items = self.table.iter().map(|(ident, entry)| (*ident, entry.clone())).collect(); + let items = self.table.iter().map(|(_, entry)| entry.clone()).collect(); // We keep only the head entries from the sync graph, // these entries will be used to initialize it back when deserializing @@ -255,3 +257,53 @@ impl Serialize for DavDag { val.serialize(serializer) } } + +// ---- TESTS ---- + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn base() { + let mut state = DavDag::default(); + + // Add item 1 + { + let m = UniqueIdent([0x01; 24]); + let ev = state.op_put((m, "cal.ics".into(), "321-321".into())); + state = state.apply(&ev); + + assert_eq!(state.table.len(), 1); + assert_eq!(state.resolve(ev.token()).unwrap().len(), 0); + } + + // Add 2 concurrent items + let (t1, t2) = { + let blob1 = UniqueIdent([0x02; 24]); + let ev1 = state.op_put((blob1, "cal2.ics".into(), "321-321".into())); + + let blob2 = UniqueIdent([0x01; 24]); + let ev2 = state.op_delete(blob2); + + state = state.apply(&ev1); + state = state.apply(&ev2); + + assert_eq!(state.table.len(), 1); + assert_eq!(state.resolve(ev1.token()).unwrap(), ordset![ev2.token()]); + + (ev1.token(), ev2.token()) + }; + + // Add later a new item + { + let blob3 = UniqueIdent([0x03; 24]); + let ev = state.op_put((blob3, "cal3.ics".into(), "321-321".into())); + + state = state.apply(&ev); + assert_eq!(state.table.len(), 2); + assert_eq!(state.resolve(ev.token()).unwrap().len(), 0); + assert_eq!(state.resolve(t1).unwrap(), ordset![t2, ev.token()]); + } + } +} |