refactor sync to new model

taskchampion/src/replica.rs

@@ -145,8 +145,8 @@ impl Replica {
     }
 
     /// Synchronize this replica against the given server.
-    pub fn sync(&mut self, username: &str, server: &mut dyn Server) -> Fallible<()> {
+    pub fn sync(&mut self, server: &mut dyn Server) -> Fallible<()> {
-        self.taskdb.sync(username, server)
+        self.taskdb.sync(server)
     }
 
     /// Perform "garbage collection" on this replica.  In particular, this renumbers the working

taskchampion/src/server/mod.rs

@@ -4,4 +4,4 @@ pub(crate) mod test;
 mod signing;
 mod types;
 
-pub use types::{Blob, Server, VersionAdd};
+pub use types::*;

taskchampion/src/server/test.rs

@@ -1,84 +1,78 @@
-use crate::server::{Blob, Server, VersionAdd};
+use crate::server::{
+    AddVersionResult, GetVersionResult, HistorySegment, Server, VersionId, NO_VERSION_ID,
+};
 use failure::Fallible;
 use std::collections::HashMap;
+use uuid::Uuid;
 
-pub(crate) struct TestServer {
+struct Version {
-    users: HashMap<String, User>,
+    version_id: VersionId,
+    parent_version_id: VersionId,
+    history_segment: HistorySegment,
 }
 
-struct User {
+pub(crate) struct TestServer {
-    // versions, indexed at v-1
+    latest_version_id: VersionId,
-    versions: Vec<Blob>,
+    // NOTE: indexed by parent_version_id!
-    snapshots: HashMap<u64, Blob>,
+    versions: HashMap<VersionId, Version>,
 }
 
 impl TestServer {
+    /// A test server has no notion of clients, signatures, encryption, etc.
     pub fn new() -> TestServer {
         TestServer {
-            users: HashMap::new(),
+            latest_version_id: NO_VERSION_ID,
+            versions: HashMap::new(),
         }
     }
-
-    fn get_user_mut(&mut self, username: &str) -> &mut User {
-        self.users
-            .entry(username.to_string())
-            .or_insert_with(User::new)
-    }
 }
 
 impl Server for TestServer {
-    /// Get a vector of all versions after `since_version`
-    fn get_versions(&self, username: &str, since_version: u64) -> Fallible<Vec<Blob>> {
-        if let Some(user) = self.users.get(username) {
-            user.get_versions(since_version)
-        } else {
-            Ok(vec![])
-        }
-    }
-
     /// Add a new version.  If the given version number is incorrect, this responds with the
     /// appropriate version and expects the caller to try again.
-    fn add_version(&mut self, username: &str, version: u64, blob: Blob) -> Fallible<VersionAdd> {
+    fn add_version(
-        self.get_user_mut(username).add_version(version, blob)
+        &mut self,
-    }
+        parent_version_id: VersionId,
+        history_segment: HistorySegment,
+    ) -> Fallible<AddVersionResult> {
+        // no client lookup
+        // no signature validation
 
-    fn add_snapshot(&mut self, username: &str, version: u64, blob: Blob) -> Fallible<()> {
+        // check the parent_version_id for linearity
-        self.get_user_mut(username).add_snapshot(version, blob)
+        if self.latest_version_id != NO_VERSION_ID {
+            if parent_version_id != self.latest_version_id {
+                return Ok(AddVersionResult::ExpectedParentVersion(
+                    self.latest_version_id,
+                ));
             }
         }
 
-impl User {
+        // invent a new ID for this version
-    fn new() -> User {
+        let version_id = Uuid::new_v4();
-        User {
+
-            versions: vec![],
+        self.versions.insert(
-            snapshots: HashMap::new(),
+            parent_version_id,
-        }
+            Version {
+                version_id,
+                parent_version_id,
+                history_segment,
+            },
+        );
+        self.latest_version_id = version_id;
+
+        Ok(AddVersionResult::Ok(version_id))
     }
 
-    fn get_versions(&self, since_version: u64) -> Fallible<Vec<Blob>> {
+    /// Get a vector of all versions after `since_version`
-        let last_version = self.versions.len();
+    fn get_child_version(&self, parent_version_id: VersionId) -> Fallible<GetVersionResult> {
-        if last_version == since_version as usize {
+        if let Some(version) = self.versions.get(&parent_version_id) {
-            return Ok(vec![]);
+            Ok(GetVersionResult::Version {
-        }
+                version_id: version.version_id,
-        Ok(self.versions[since_version as usize..last_version]
+                parent_version_id: version.parent_version_id,
-            .iter()
+                history_segment: version.history_segment.clone(),
-            .map(|r| r.clone())
+            })
-            .collect::<Vec<Blob>>())
+        } else {
-    }
+            Ok(GetVersionResult::NoSuchVersion)
-
+        }
-    fn add_version(&mut self, version: u64, blob: Blob) -> Fallible<VersionAdd> {
-        // of by one here: client wants to send version 1 first
-        let expected_version = self.versions.len() as u64 + 1;
-        if version != expected_version {
-            return Ok(VersionAdd::ExpectedVersion(expected_version));
-        }
-        self.versions.push(blob);
-
-        Ok(VersionAdd::Ok)
-    }
-
-    fn add_snapshot(&mut self, version: u64, blob: Blob) -> Fallible<()> {
-        self.snapshots.insert(version, blob);
-        Ok(())
     }
 }

taskchampion/src/server/types.rs

@@ -1,26 +1,46 @@
 use failure::Fallible;
+use uuid::Uuid;
 
-/// A Blob is a hunk of encoded data that is sent to the server.  The server does not interpret
+/// Versions are referred to with sha2 hashes.
-/// this data at all.
+pub type VersionId = Uuid;
-pub type Blob = Vec<u8>;
+
+/// The distinguished value for "no version"
+pub const NO_VERSION_ID: VersionId = Uuid::nil();
+
+/// A segment in the history of this task database, in the form of a sequence of operations.  This
+/// data is pre-encoded, and from the protocol level appears as a sequence of bytes.
+pub type HistorySegment = Vec<u8>;
 
 /// VersionAdd is the response type from [`crate:server::Server::add_version`].
-pub enum VersionAdd {
+pub enum AddVersionResult {
-    /// OK, version added
+    /// OK, version added with the given ID
-    Ok,
+    Ok(VersionId),
-    /// Rejected, must be based on the the given version
+    /// Rejected; expected a version with the given parent version
-    ExpectedVersion(u64),
+    ExpectedParentVersion(VersionId),
+}
+
+/// A version as downloaded from the server
+pub enum GetVersionResult {
+    /// No such version exists
+    NoSuchVersion,
+
+    /// The requested version
+    Version {
+        version_id: VersionId,
+        parent_version_id: VersionId,
+        history_segment: HistorySegment,
+    },
 }
 
 /// A value implementing this trait can act as a server against which a replica can sync.
 pub trait Server {
-    /// Get a vector of all versions after `since_version`
+    /// Add a new version.
-    fn get_versions(&self, username: &str, since_version: u64) -> Fallible<Vec<Blob>>;
+    fn add_version(
+        &mut self,
+        parent_version_id: VersionId,
+        history_segment: HistorySegment,
+    ) -> Fallible<AddVersionResult>;
 
-    /// Add a new version.  If the given version number is incorrect, this responds with the
+    /// Get the version with the given parent VersionId
-    /// appropriate version and expects the caller to try again.
+    fn get_child_version(&self, parent_version_id: VersionId) -> Fallible<GetVersionResult>;
-    fn add_version(&mut self, username: &str, version: u64, blob: Blob) -> Fallible<VersionAdd>;
-
-    /// TODO: undefined yet
-    fn add_snapshot(&mut self, username: &str, version: u64, blob: Blob) -> Fallible<()>;
 }

taskchampion/src/taskdb.rs

@@ -1,5 +1,5 @@
 use crate::errors::Error;
-use crate::server::{Server, VersionAdd};
+use crate::server::{AddVersionResult, GetVersionResult, Server};
 use crate::taskstorage::{Operation, TaskMap, TaskStorage, TaskStorageTxn};
 use failure::Fallible;
 use serde::{Deserialize, Serialize};
@@ -13,7 +13,6 @@ pub struct TaskDB {
 
 #[derive(Serialize, Deserialize, Debug)]
 struct Version {
-    version: u64,
     operations: Vec<Operation>,
 }
 
@@ -165,21 +164,34 @@ impl TaskDB {
     }
 
     /// Sync to the given server, pulling remote changes and pushing local changes.
-    pub fn sync(&mut self, username: &str, server: &mut dyn Server) -> Fallible<()> {
+    pub fn sync(&mut self, server: &mut dyn Server) -> Fallible<()> {
         let mut txn = self.storage.txn()?;
 
         // retry synchronizing until the server accepts our version (this allows for races between
         // replicas trying to sync to the same server)
         loop {
-            // first pull changes and "rebase" on top of them
+            let mut base_version_id = txn.base_version()?;
-            let new_versions = server.get_versions(username, txn.base_version()?)?;
-            for version_blob in new_versions {
-                let version_str = str::from_utf8(&version_blob).unwrap();
-                let version: Version = serde_json::from_str(version_str).unwrap();
-                assert_eq!(version.version, txn.base_version()? + 1);
-                println!("applying version {:?} from server", version.version);
 
+            // first pull changes and "rebase" on top of them
+            loop {
+                if let GetVersionResult::Version {
+                    version_id,
+                    history_segment,
+                    ..
+                } = server.get_child_version(base_version_id)?
+                {
+                    let version_str = str::from_utf8(&history_segment).unwrap();
+                    let version: Version = serde_json::from_str(version_str).unwrap();
+                    println!("applying version {:?} from server", version_id);
+
+                    // apply this verison and update base_version in storage
                     TaskDB::apply_version(txn.as_mut(), version)?;
+                    txn.set_base_version(version_id)?;
+                    base_version_id = version_id;
+                } else {
+                    // at the moment, no more child versions, so we can try adding our own
+                    break;
+                }
             }
 
             let operations: Vec<Operation> = txn.operations()?.to_vec();
@@ -189,19 +201,24 @@ impl TaskDB {
             }
 
             // now make a version of our local changes and push those
-            let new_version = Version {
+            let new_version = Version { operations };
-                version: txn.base_version()? + 1,
+            let history_segment = serde_json::to_string(&new_version).unwrap().into();
-                operations,
+            println!("sending new version to server");
-            };
+            match server.add_version(base_version_id, history_segment)? {
-            let new_version_str = serde_json::to_string(&new_version).unwrap();
+                AddVersionResult::Ok(new_version_id) => {
-            println!("sending version {:?} to server", new_version.version);
+                    println!("version {:?} received by server", new_version_id);
-            if let VersionAdd::Ok =
+                    txn.set_base_version(new_version_id)?;
-                server.add_version(username, new_version.version, new_version_str.into())?
-            {
-                txn.set_base_version(new_version.version)?;
                     txn.set_operations(vec![])?;
                     break;
                 }
+                AddVersionResult::ExpectedParentVersion(parent_version_id) => {
+                    println!(
+                        "new version rejected; must be based on {:?}",
+                        parent_version_id
+                    );
+                    // ..continue the outer loop
+                }
+            }
         }
 
         txn.commit()?;
@@ -256,7 +273,6 @@ impl TaskDB {
             }
             local_operations = new_local_ops;
         }
-        txn.set_base_version(version.version)?;
         txn.set_operations(local_operations)?;
         Ok(())
     }
@@ -518,10 +534,10 @@ mod tests {
         let mut server = TestServer::new();
 
         let mut db1 = newdb();
-        db1.sync("me", &mut server).unwrap();
+        db1.sync(&mut server).unwrap();
 
         let mut db2 = newdb();
-        db2.sync("me", &mut server).unwrap();
+        db2.sync(&mut server).unwrap();
 
         // make some changes in parallel to db1 and db2..
         let uuid1 = Uuid::new_v4();
@@ -545,9 +561,9 @@ mod tests {
         .unwrap();
 
         // and synchronize those around
-        db1.sync("me", &mut server).unwrap();
+        db1.sync(&mut server).unwrap();
-        db2.sync("me", &mut server).unwrap();
+        db2.sync(&mut server).unwrap();
-        db1.sync("me", &mut server).unwrap();
+        db1.sync(&mut server).unwrap();
         assert_eq!(db1.sorted_tasks(), db2.sorted_tasks());
 
         // now make updates to the same task on both sides
@@ -567,9 +583,9 @@ mod tests {
         .unwrap();
 
         // and synchronize those around
-        db1.sync("me", &mut server).unwrap();
+        db1.sync(&mut server).unwrap();
-        db2.sync("me", &mut server).unwrap();
+        db2.sync(&mut server).unwrap();
-        db1.sync("me", &mut server).unwrap();
+        db1.sync(&mut server).unwrap();
         assert_eq!(db1.sorted_tasks(), db2.sorted_tasks());
     }
 
@@ -578,10 +594,10 @@ mod tests {
         let mut server = TestServer::new();
 
         let mut db1 = newdb();
-        db1.sync("me", &mut server).unwrap();
+        db1.sync(&mut server).unwrap();
 
         let mut db2 = newdb();
-        db2.sync("me", &mut server).unwrap();
+        db2.sync(&mut server).unwrap();
 
         // create and update a task..
         let uuid = Uuid::new_v4();
@@ -595,9 +611,9 @@ mod tests {
         .unwrap();
 
         // and synchronize those around
-        db1.sync("me", &mut server).unwrap();
+        db1.sync(&mut server).unwrap();
-        db2.sync("me", &mut server).unwrap();
+        db2.sync(&mut server).unwrap();
-        db1.sync("me", &mut server).unwrap();
+        db1.sync(&mut server).unwrap();
         assert_eq!(db1.sorted_tasks(), db2.sorted_tasks());
 
         // delete and re-create the task on db1
@@ -620,9 +636,9 @@ mod tests {
         })
         .unwrap();
 
-        db1.sync("me", &mut server).unwrap();
+        db1.sync(&mut server).unwrap();
-        db2.sync("me", &mut server).unwrap();
+        db2.sync(&mut server).unwrap();
-        db1.sync("me", &mut server).unwrap();
+        db1.sync(&mut server).unwrap();
         assert_eq!(db1.sorted_tasks(), db2.sorted_tasks());
     }
 
@@ -678,7 +694,7 @@ mod tests {
                             println!("  {:?} (ignored)", e);
                         }
                     },
-                    Action::Sync => db.sync("me", &mut server).unwrap(),
+                    Action::Sync => db.sync(&mut server).unwrap(),
                 }
             }
 

taskchampion/src/taskstorage/inmemory.rs

@@ -1,6 +1,8 @@
 #![allow(clippy::new_without_default)]
 
-use crate::taskstorage::{Operation, TaskMap, TaskStorage, TaskStorageTxn};
+use crate::taskstorage::{
+    Operation, TaskMap, TaskStorage, TaskStorageTxn, VersionId, DEFAULT_BASE_VERSION,
+};
 use failure::Fallible;
 use std::collections::hash_map::Entry;
 use std::collections::HashMap;
@@ -9,7 +11,7 @@ use uuid::Uuid;
 #[derive(PartialEq, Debug, Clone)]
 struct Data {
     tasks: HashMap<Uuid, TaskMap>,
-    base_version: u64,
+    base_version: VersionId,
     operations: Vec<Operation>,
     working_set: Vec<Option<Uuid>>,
 }
@@ -79,11 +81,11 @@ impl<'t> TaskStorageTxn for Txn<'t> {
         Ok(self.data_ref().tasks.keys().copied().collect())
     }
 
-    fn base_version(&mut self) -> Fallible<u64> {
+    fn base_version(&mut self) -> Fallible<VersionId> {
-        Ok(self.data_ref().base_version)
+        Ok(self.data_ref().base_version.clone())
     }
 
-    fn set_base_version(&mut self, version: u64) -> Fallible<()> {
+    fn set_base_version(&mut self, version: VersionId) -> Fallible<()> {
         self.mut_data_ref().base_version = version;
         Ok(())
     }
@@ -138,7 +140,7 @@ impl InMemoryStorage {
         InMemoryStorage {
             data: Data {
                 tasks: HashMap::new(),
-                base_version: 0,
+                base_version: DEFAULT_BASE_VERSION.into(),
                 operations: vec![],
                 working_set: vec![None],
             },

taskchampion/src/taskstorage/kv.rs

@@ -1,4 +1,6 @@
-use crate::taskstorage::{Operation, TaskMap, TaskStorage, TaskStorageTxn};
+use crate::taskstorage::{
+    Operation, TaskMap, TaskStorage, TaskStorageTxn, VersionId, DEFAULT_BASE_VERSION,
+};
 use failure::Fallible;
 use kv::msgpack::Msgpack;
 use kv::{Bucket, Config, Error, Integer, Serde, Store, ValueBuf};
@@ -48,6 +50,7 @@ pub struct KVStorage<'t> {
     store: Store,
     tasks_bucket: Bucket<'t, Key, ValueBuf<Msgpack<TaskMap>>>,
     numbers_bucket: Bucket<'t, Integer, ValueBuf<Msgpack<u64>>>,
+    uuids_bucket: Bucket<'t, Integer, ValueBuf<Msgpack<Uuid>>>,
     operations_bucket: Bucket<'t, Integer, ValueBuf<Msgpack<Operation>>>,
     working_set_bucket: Bucket<'t, Integer, ValueBuf<Msgpack<Uuid>>>,
 }
@@ -61,6 +64,7 @@ impl<'t> KVStorage<'t> {
         let mut config = Config::default(directory);
         config.bucket("tasks", None);
         config.bucket("numbers", None);
+        config.bucket("uuids", None);
         config.bucket("operations", None);
         config.bucket("working_set", None);
         let store = Store::new(config)?;
@@ -71,6 +75,9 @@ impl<'t> KVStorage<'t> {
         // this bucket contains various u64s, indexed by constants above
         let numbers_bucket = store.int_bucket::<ValueBuf<Msgpack<u64>>>(Some("numbers"))?;
 
+        // this bucket contains various Uuids, indexed by constants above
+        let uuids_bucket = store.int_bucket::<ValueBuf<Msgpack<Uuid>>>(Some("uuids"))?;
+
         // this bucket contains operations, numbered consecutively; the NEXT_OPERATION number gives
         // the index of the next operation to insert
         let operations_bucket =
@@ -85,6 +92,7 @@ impl<'t> KVStorage<'t> {
             store,
             tasks_bucket,
             numbers_bucket,
+            uuids_bucket,
             operations_bucket,
             working_set_bucket,
         })
@@ -122,6 +130,9 @@ impl<'t> Txn<'t> {
     fn numbers_bucket(&self) -> &'t Bucket<'t, Integer, ValueBuf<Msgpack<u64>>> {
         &self.storage.numbers_bucket
     }
+    fn uuids_bucket(&self) -> &'t Bucket<'t, Integer, ValueBuf<Msgpack<Uuid>>> {
+        &self.storage.uuids_bucket
+    }
     fn operations_bucket(&self) -> &'t Bucket<'t, Integer, ValueBuf<Msgpack<Operation>>> {
         &self.storage.operations_bucket
     }
@@ -193,26 +204,26 @@ impl<'t> TaskStorageTxn for Txn<'t> {
             .collect())
     }
 
-    fn base_version(&mut self) -> Fallible<u64> {
+    fn base_version(&mut self) -> Fallible<VersionId> {
-        let bucket = self.numbers_bucket();
+        let bucket = self.uuids_bucket();
         let base_version = match self.kvtxn().get(bucket, BASE_VERSION.into()) {
             Ok(buf) => buf,
-            Err(Error::NotFound) => return Ok(0),
+            Err(Error::NotFound) => return Ok(DEFAULT_BASE_VERSION.into()),
             Err(e) => return Err(e.into()),
         }
         .inner()?
         .to_serde();
-        Ok(base_version)
+        Ok(base_version as VersionId)
     }
 
-    fn set_base_version(&mut self, version: u64) -> Fallible<()> {
+    fn set_base_version(&mut self, version: VersionId) -> Fallible<()> {
-        let numbers_bucket = self.numbers_bucket();
+        let uuids_bucket = self.uuids_bucket();
         let kvtxn = self.kvtxn();
 
         kvtxn.set(
-            numbers_bucket,
+            uuids_bucket,
             BASE_VERSION.into(),
-            Msgpack::to_value_buf(version)?,
+            Msgpack::to_value_buf(version as Uuid)?,
         )?;
         Ok(())
     }
@@ -528,7 +539,7 @@ mod test {
         let mut storage = KVStorage::new(&tmp_dir.path())?;
         {
             let mut txn = storage.txn()?;
-            assert_eq!(txn.base_version()?, 0);
+            assert_eq!(txn.base_version()?, DEFAULT_BASE_VERSION);
         }
         Ok(())
     }
@@ -537,14 +548,15 @@ mod test {
     fn test_base_version_setting() -> Fallible<()> {
         let tmp_dir = TempDir::new("test")?;
         let mut storage = KVStorage::new(&tmp_dir.path())?;
+        let u = Uuid::new_v4();
         {
             let mut txn = storage.txn()?;
-            txn.set_base_version(3)?;
+            txn.set_base_version(u)?;
             txn.commit()?;
         }
         {
             let mut txn = storage.txn()?;
-            assert_eq!(txn.base_version()?, 3);
+            assert_eq!(txn.base_version()?, u);
         }
         Ok(())
     }

taskchampion/src/taskstorage/mod.rs

@@ -23,6 +23,12 @@ fn taskmap_with(mut properties: Vec<(String, String)>) -> TaskMap {
     rv
 }
 
+/// The type of VersionIds
+pub use crate::server::VersionId;
+
+/// The default for base_version.
+pub(crate) const DEFAULT_BASE_VERSION: Uuid = crate::server::NO_VERSION_ID;
+
 /// A TaskStorage transaction, in which storage operations are performed.
 ///
 /// # Concurrency
@@ -58,10 +64,10 @@ pub trait TaskStorageTxn {
     fn all_task_uuids(&mut self) -> Fallible<Vec<Uuid>>;
 
     /// Get the current base_version for this storage -- the last version synced from the server.
-    fn base_version(&mut self) -> Fallible<u64>;
+    fn base_version(&mut self) -> Fallible<VersionId>;
 
     /// Set the current base_version for this storage.
-    fn set_base_version(&mut self, version: u64) -> Fallible<()>;
+    fn set_base_version(&mut self, version: VersionId) -> Fallible<()>;
 
     /// Get the current set of outstanding operations (operations that have not been sync'd to the
     /// server yet)