Replication syncing

Author: Grisha Kruglov

Cargo.toml

@@ -16,8 +16,9 @@ version = "0.6.1"
 build = "build/version.rs"
 
 [features]
-default = ["bundled_sqlite3"]
+default = ["bundled_sqlite3", "syncable"]
 bundled_sqlite3 = ["rusqlite/bundled"]
+syncable = ["mentat_tolstoy"]
 
 [workspace]
 members = ["tools/cli"]
@@ -78,6 +79,7 @@ path = "tx-parser"
 
 [dependencies.mentat_tolstoy]
 path = "tolstoy"
+optional = true
 
 [profile.release]
 debug = true

src/conn.rs

@@ -59,10 +59,6 @@ use mentat_tx::entities::TempId;
 
 use mentat_tx_parser;
 
-use mentat_tolstoy::Syncer;
-
-use uuid::Uuid;
-
 use entity_builder::{
     InProgressBuilder,
 };
@@ -129,8 +125,8 @@ pub struct Conn {
 /// A convenience wrapper around a single SQLite connection and a Conn. This is suitable
 /// for applications that don't require complex connection management.
 pub struct Store {
+    pub sqlite: rusqlite::Connection,
     conn: Conn,
-    sqlite: rusqlite::Connection,
 }
 
 impl Store {
@@ -172,10 +168,6 @@ pub trait Queryable {
         where E: Into<Entid>;
 }
 
-pub trait Syncable {
-    fn sync(&mut self, server_uri: &String, user_uuid: &String) -> Result<()>;
-}
-
 /// Represents an in-progress, not yet committed, set of changes to the store.
 /// Call `commit` to commit your changes, or `rollback` to discard them.
 /// A transaction is held open until you do so.
@@ -493,13 +485,6 @@ pub enum CacheAction {
     Deregister,
 }
 
-impl Syncable for Store {
-    fn sync(&mut self, server_uri: &String, user_uuid: &String) -> Result<()> {
-        let uuid = Uuid::parse_str(&user_uuid)?;
-        Ok(Syncer::flow(&mut self.sqlite, server_uri, &uuid)?)
-    }
-}
-
 impl Conn {
     // Intentionally not public.
     fn new(partition_map: PartitionMap, schema: Schema) -> Conn {

src/lib.rs

@@ -29,10 +29,12 @@ extern crate mentat_query_parser;
 extern crate mentat_query_projector;
 extern crate mentat_query_translator;
 extern crate mentat_sql;
-extern crate mentat_tolstoy;
 extern crate mentat_tx;
 extern crate mentat_tx_parser;
 
+#[cfg(feature = "syncable")]
+extern crate mentat_tolstoy;
+
 pub use mentat_core::{
     Attribute,
     Entid,
@@ -95,6 +97,13 @@ pub mod conn;
 pub mod query;
 pub mod entity_builder;
 
+#[cfg(feature = "syncable")]
+pub mod sync;
+
+pub fn get_name() -> String {
+    return String::from("mentat");
+}
+
 pub use query::{
     IntoResult,
     PlainSymbol,
@@ -115,7 +124,6 @@ pub use conn::{
     InProgress,
     Metadata,
     Queryable,
-    Syncable,
     Store,
 };
 

src/sync.rs

@@ -0,0 +1,81 @@
+// Copyright 2016 Mozilla
+//
+// Licensed under the Apache License, Version 2.0 (the "License"); you may not use
+// this file except in compliance with the License. You may obtain a copy of the
+// License at http://www.apache.org/licenses/LICENSE-2.0
+// Unless required by applicable law or agreed to in writing, software distributed
+// under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR
+// CONDITIONS OF ANY KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations under the License.
+
+use uuid::Uuid;
+
+use conn::Store;
+use errors::Result;
+
+use mentat_core::KnownEntid;
+
+use entity_builder::BuildTerms;
+
+use mentat_tolstoy::{
+    Syncer,
+    SyncMetadataClient,
+    TxMapper,
+};
+use mentat_tolstoy::metadata::HeadTrackable;
+
+pub trait Syncable {
+    fn sync(&mut self, server_uri: &String, user_uuid: &String) -> Result<()>;
+}
+
+impl Syncable for Store {
+    // TODO refactor this out of there, somehow!
+    fn sync(&mut self, server_uri: &String, user_uuid: &String) -> Result<()> {
+        let uuid = Uuid::parse_str(&user_uuid)?;
+        let mut tx_list = None;
+        {
+            let mut db_tx = self.sqlite.transaction()?;
+            if let Some(list) = Syncer::flow(&mut db_tx, server_uri, &uuid)? {
+                tx_list = Some(list);
+            }
+
+            // TODO this should be done _after_ any parts below are transacted!!
+            // Commits any changes Syncer made (tu mappings during an upload!)
+            db_tx.commit()?;
+        }
+        let mut last_tx_entid = None;
+        let mut last_tx_uuid = None;
+
+        match tx_list {
+            Some(list) => {
+                for tx in list {
+                    let in_progress = self.begin_transaction()?;
+                    let mut builder = in_progress.builder();
+                    for part in tx.parts {
+                        if part.added {
+                            builder.add(KnownEntid(part.e), KnownEntid(part.a), part.v.clone())?;
+                        } else {
+                            builder.retract(KnownEntid(part.e), KnownEntid(part.a), part.v.clone())?;
+                        }
+                    }
+                    let report = builder.commit()?;
+                    last_tx_entid = Some(report.tx_id);
+                    last_tx_uuid = Some(tx.tx.clone());
+                }
+                ()
+            },
+            None => ()
+        }
+
+        if let Some(uuid) = last_tx_uuid {
+            if let Some(entid) = last_tx_entid {
+                let mut db_tx = self.sqlite.transaction()?;
+                TxMapper::set_tx_uuid(&mut db_tx, entid, &uuid)?;
+                SyncMetadataClient::set_remote_head(&mut db_tx, &uuid)?;
+                db_tx.commit()?;
+            }
+        }
+        
+        Ok(())
+    }
+}

tests/tolstoy.rs

@@ -97,12 +97,12 @@ fn test_reader() {
     let mut conn = Conn::connect(&mut c).expect("Couldn't open DB.");
     {
         let db_tx = c.transaction().expect("db tx");
-        // Don't inspect the bootstrap transaction, but we'd like to see it's there.
+        // Ensure that the first (bootstrap) transaction is skipped over.
         let mut receiver = TxCountingReceiver::new();
         assert_eq!(false, receiver.is_done);
         Processor::process(&db_tx, None, &mut receiver).expect("processor");
         assert_eq!(true, receiver.is_done);
-        assert_eq!(1, receiver.tx_count);
+        assert_eq!(0, receiver.tx_count);
     }
 
     let ids = conn.transact(&mut c, r#"[
@@ -112,7 +112,7 @@ fn test_reader() {
     ]"#).expect("successful transaction").tempids;
     let numba_entity_id = ids.get("s").unwrap();
 
-    let mut bootstrap_tx = None;
+    let first_tx;
     {
         let db_tx = c.transaction().expect("db tx");
         // Expect to see one more transaction of four parts (one for tx datom itself).
@@ -121,10 +121,10 @@ fn test_reader() {
 
         println!("{:#?}", receiver);
 
-        assert_eq!(2, receiver.txes.keys().count());
-        assert_tx_datoms_count(&receiver, 1, 4);
+        assert_eq!(1, receiver.txes.keys().count());
+        assert_tx_datoms_count(&receiver, 0, 4);
 
-        bootstrap_tx = Some(*receiver.txes.keys().nth(0).expect("bootstrap tx"));
+        first_tx = Some(*receiver.txes.keys().nth(0).expect("first tx"));
     }
 
     let ids = conn.transact(&mut c, r#"[
@@ -138,14 +138,14 @@ fn test_reader() {
         // Expect to see a single two part transaction
         let mut receiver = TestingReceiver::new();
 
-        // Note that we're asking for the bootstrap tx to be skipped by the processor.
-        Processor::process(&db_tx, bootstrap_tx, &mut receiver).expect("processor");
+        // Note that we're asking for the first transacted tx to be skipped by the processor.
+        Processor::process(&db_tx, first_tx, &mut receiver).expect("processor");
 
-        assert_eq!(2, receiver.txes.keys().count());
-        assert_tx_datoms_count(&receiver, 1, 2);
+        assert_eq!(1, receiver.txes.keys().count());
+        assert_tx_datoms_count(&receiver, 0, 2);
 
         // Inspect the transaction part.
-        let tx_id = receiver.txes.keys().nth(1).expect("tx");
+        let tx_id = receiver.txes.keys().nth(0).expect("tx");
         let datoms = receiver.txes.get(tx_id).expect("datoms");
         let part = &datoms[0];
 

tolstoy/src/lib.rs

@@ -38,7 +38,9 @@ pub mod tx_processor;
 pub mod errors;
 pub mod syncer;
 pub mod tx_mapper;
+pub use tx_mapper::TxMapper;
 pub use syncer::Syncer;
+pub use metadata::SyncMetadataClient;
 pub use errors::{
     Error,
     ErrorKind,

tolstoy/src/metadata.rs

@@ -54,16 +54,16 @@ mod tests {
 
     #[test]
     fn test_get_remote_head_default() {
-        let mut conn = schema::tests::setup_conn();
-        let tx = conn.transaction().expect("db tx");
+        let mut conn = schema::tests::setup_conn_bare();
+        let tx = schema::tests::setup_tx(&mut conn);
         assert_eq!(Uuid::nil(), SyncMetadataClient::remote_head(&tx).expect("fetch succeeded"));
     }
 
     #[test]
     fn test_set_and_get_remote_head() {
-        let mut conn = schema::tests::setup_conn();
+        let mut conn = schema::tests::setup_conn_bare();
+        let tx = schema::tests::setup_tx(&mut conn);
         let uuid = Uuid::new_v4();
-        let tx = conn.transaction().expect("db tx");
         SyncMetadataClient::set_remote_head(&tx, &uuid).expect("update succeeded");
         assert_eq!(uuid, SyncMetadataClient::remote_head(&tx).expect("fetch succeeded"));
     }

tolstoy/src/schema.rs

@@ -24,23 +24,21 @@ lazy_static! {
     };
 }
 
-pub fn ensure_current_version(conn: &mut rusqlite::Connection) -> Result<()> {
-    let tx = conn.transaction()?;
-
+pub fn ensure_current_version(tx: &mut rusqlite::Transaction) -> Result<()> {
     for statement in (&SCHEMA_STATEMENTS).iter() {
         tx.execute(statement, &[])?;
     }
 
     tx.execute("INSERT OR IGNORE INTO tolstoy_metadata (key, value) VALUES (?, zeroblob(16))", &[&REMOTE_HEAD_KEY])?;
-    tx.commit().map_err(|e| e.into())
+    Ok(())
 }
 
 #[cfg(test)]
 pub mod tests {
     use super::*;
     use uuid::Uuid;
 
-    fn setup_conn_bare() -> rusqlite::Connection {
+    pub fn setup_conn_bare() -> rusqlite::Connection {
         let conn = rusqlite::Connection::open_in_memory().unwrap();
 
         conn.execute_batch("
@@ -54,19 +52,24 @@ pub mod tests {
         conn
     }
 
-    pub fn setup_conn() -> rusqlite::Connection {
-        let mut conn = setup_conn_bare();
-        ensure_current_version(&mut conn).expect("connection setup");
-        conn
+    pub fn setup_tx_bare<'a>(conn: &'a mut rusqlite::Connection) -> rusqlite::Transaction<'a> {
+        conn.transaction().expect("tx")
+    }
+
+    pub fn setup_tx<'a>(conn: &'a mut rusqlite::Connection) -> rusqlite::Transaction<'a> {
+        let mut tx = conn.transaction().expect("tx");
+        ensure_current_version(&mut tx).expect("connection setup");
+        tx
     }
 
     #[test]
     fn test_empty() {
         let mut conn = setup_conn_bare();
+        let mut tx = setup_tx_bare(&mut conn);
 
-        assert!(ensure_current_version(&mut conn).is_ok());
+        assert!(ensure_current_version(&mut tx).is_ok());
 
-        let mut stmt = conn.prepare("SELECT key FROM tolstoy_metadata WHERE value = zeroblob(16)").unwrap();
+        let mut stmt = tx.prepare("SELECT key FROM tolstoy_metadata WHERE value = zeroblob(16)").unwrap();
         let mut keys_iter = stmt.query_map(&[], |r| r.get(0)).expect("query works");
 
         let first: Result<String> = keys_iter.next().unwrap().map_err(|e| e.into());
@@ -82,27 +85,23 @@ pub mod tests {
     #[test]
     fn test_non_empty() {
         let mut conn = setup_conn_bare();
+        let mut tx = setup_tx_bare(&mut conn);
 
-        assert!(ensure_current_version(&mut conn).is_ok());
+        assert!(ensure_current_version(&mut tx).is_ok());
 
         let test_uuid = Uuid::new_v4();
         {
-            let tx = conn.transaction().unwrap();
             let uuid_bytes = test_uuid.as_bytes().to_vec();
             match tx.execute("UPDATE tolstoy_metadata SET value = ? WHERE key = ?", &[&uuid_bytes, &REMOTE_HEAD_KEY]) {
                 Err(e) => panic!("Error running an update: {}", e),
                 _ => ()
             }
-            match tx.commit() {
-                Err(e) => panic!("Error committing an update: {}", e),
-                _ => ()
-            }
         }
 
-        assert!(ensure_current_version(&mut conn).is_ok());
+        assert!(ensure_current_version(&mut tx).is_ok());
 
         // Check that running ensure_current_version on an initialized conn doesn't change anything.
-        let mut stmt = conn.prepare("SELECT value FROM tolstoy_metadata").unwrap();
+        let mut stmt = tx.prepare("SELECT value FROM tolstoy_metadata").unwrap();
         let mut values_iter = stmt.query_map(&[], |r| {
             let raw_uuid: Vec<u8> = r.get(0);
             Uuid::from_bytes(raw_uuid.as_slice()).unwrap()