Reconcile incoming bootstrap transaction on first sync.

Author: Grisha Kruglov

src/sync.rs

@@ -18,7 +18,6 @@ use conn::{
 };
 
 use errors::{
-    MentatError,
     Result,
 };
 
@@ -46,8 +45,8 @@ use mentat_tolstoy::{
     SyncResult,
     Tx,
     TxMapper,
+    TolstoyError,
 };
-use mentat_tolstoy::metadata::HeadTrackable;
 
 pub trait Syncable {
     fn sync(&mut self, server_uri: &String, user_uuid: &String) -> Result<()>;
@@ -124,8 +123,17 @@ impl Conn {
             SyncResult::EmptyServer => (),
             SyncResult::NoChanges => (),
             SyncResult::ServerFastForward => (),
-            SyncResult::Merge => bail!(MentatError::NotYetImplemented("Can't sync against diverged local.".into())),
+            SyncResult::Merge => bail!(TolstoyError::NotYetImplemented(
+                format!("Can't sync against diverged local.")
+            )),
             SyncResult::LocalFastForward(txs) => fast_forward_local(&mut in_progress, txs)?,
+            SyncResult::BadServerState => bail!(TolstoyError::NotYetImplemented(
+                format!("Bad server state.")
+            )),
+            SyncResult::AdoptedRemoteOnFirstSync => (),
+            SyncResult::IncompatibleBootstrapSchema => bail!(TolstoyError::NotYetImplemented(
+                format!("IncompatibleBootstrapSchema.")
+            )),
         }
 
         in_progress.commit()

tests/tolstoy.rs

@@ -99,12 +99,12 @@ fn test_reader() {
     let mut conn = Conn::connect(&mut c).expect("Couldn't open DB.");
     {
         let db_tx = c.transaction().expect("db tx");
-        // Ensure that the first (bootstrap) transaction is skipped over.
+        // Ensure that we see a bootstrap transaction.
         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!(0, receiver.tx_count);
+        assert_eq!(1, receiver.tx_count);
     }
 
     let ids = conn.transact(&mut c, r#"[
@@ -114,6 +114,11 @@ fn test_reader() {
     ]"#).expect("successful transaction").tempids;
     let numba_entity_id = ids.get("s").unwrap();
 
+    let ids = conn.transact(&mut c, r#"[
+        [:db/add "b" :foo/numba 123]
+    ]"#).expect("successful transaction").tempids;
+    let _asserted_e = ids.get("b").unwrap();
+
     let first_tx;
     {
         let db_tx = c.transaction().expect("db tx");
@@ -123,10 +128,11 @@ fn test_reader() {
 
         println!("{:#?}", receiver);
 
-        assert_eq!(1, receiver.txes.keys().count());
-        assert_tx_datoms_count(&receiver, 0, 4);
+        // Three transactions: bootstrap, vocab, assertion.
+        assert_eq!(3, receiver.txes.keys().count());
+        assert_tx_datoms_count(&receiver, 2, 2);
 
-        first_tx = Some(*receiver.txes.keys().nth(0).expect("first tx"));
+        first_tx = Some(*receiver.txes.keys().nth(1).expect("first non-bootstrap tx"));
     }
 
     let ids = conn.transact(&mut c, r#"[
@@ -143,11 +149,13 @@ fn test_reader() {
         // 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!(1, receiver.txes.keys().count());
-        assert_tx_datoms_count(&receiver, 0, 2);
+        // Vocab, assertion.
+        assert_eq!(2, receiver.txes.keys().count());
+        // Assertion datoms.
+        assert_tx_datoms_count(&receiver, 1, 2);
 
-        // Inspect the transaction part.
-        let tx_id = receiver.txes.keys().nth(0).expect("tx");
+        // Inspect the assertion.
+        let tx_id = receiver.txes.keys().nth(1).expect("tx");
         let datoms = receiver.txes.get(tx_id).expect("datoms");
         let part = datoms.iter().find(|&part| &part.e == asserted_e).expect("to find asserted datom");
 

tolstoy/src/bootstrap.rs

@@ -0,0 +1,62 @@
+// Copyright 2018 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 mentat_core::{
+    Keyword,
+};
+
+use mentat_db::{
+    CORE_SCHEMA_VERSION,
+};
+
+use errors::{
+    Result,
+    TolstoyError,
+};
+use parts::{
+    PartsHelper,
+};
+use types::{
+    Tx,
+};
+
+pub struct BootstrapHelper<'a> {
+    parts: PartsHelper<'a>
+}
+
+impl<'a> BootstrapHelper<'a> {
+    pub fn new(assumed_bootstrap_tx: &Tx) -> BootstrapHelper {
+        BootstrapHelper {
+            parts: PartsHelper::new(&assumed_bootstrap_tx.parts),
+        }
+    }
+
+    // TODO we could also iterate through our own bootstrap schema definition and check that everything matches
+    // "version" is used here as a proxy for doing that work
+    pub fn is_compatible(&self) -> Result<bool> {
+        Ok(self.core_schema_version()? == CORE_SCHEMA_VERSION as i64)
+    }
+
+    pub fn core_schema_version(&self) -> Result<i64> {
+        match self.parts.ea_lookup(
+            Keyword::namespaced("db.schema", "core"),
+            Keyword::namespaced("db.schema", "version"),
+        ) {
+            Some(v) => {
+                // TODO v is just a type tag and a Copy value, we shouldn't need to clone.
+                match v.clone().into_long() {
+                    Some(v) => Ok(v),
+                    None => bail!(TolstoyError::BadServerState("incorrect type for core schema version".to_string()))
+                }
+            },
+            None => bail!(TolstoyError::BadServerState("missing core schema version".to_string()))
+        }
+    }
+}

tolstoy/src/errors.rs

@@ -26,6 +26,10 @@ pub enum TolstoyError {
     #[fail(display = "Received bad response from the server: {}", _0)]
     BadServerResponse(String),
 
+    // TODO expand this into concrete error types
+    #[fail(display = "Received bad server state: {}", _0)]
+    BadServerState(String),
+
     #[fail(display = "encountered more than one metadata value for key: {}", _0)]
     DuplicateMetadata(String),
 
@@ -38,6 +42,9 @@ pub enum TolstoyError {
     #[fail(display = "encountered unexpected state: {}", _0)]
     UnexpectedState(String),
 
+    #[fail(display = "incompatible remote bootstrap schema. local: {}, remote: {}", _0, _1)]
+    IncompatibleBootstrapSchema(i64, i64),
+
     #[fail(display = "not yet implemented: {}", _0)]
     NotYetImplemented(String),
 }

tolstoy/src/lib.rs

@@ -32,25 +32,28 @@ extern crate rusqlite;
 extern crate uuid;
 
 #[macro_use] pub mod errors;
-mod remote_client;
+pub use errors::{
+    TolstoyError,
+};
 
-pub mod schema;
+pub mod bootstrap;
 pub mod metadata;
-pub mod tx_processor;
-pub mod syncer;
-pub mod tx_mapper;
-pub use tx_mapper::{
-    TxMapper,
+pub use metadata::{
+    SyncMetadataClient,
 };
+mod parts;
+mod remote_client;
+pub mod schema;
+pub mod syncer;
 pub use syncer::{
     Syncer,
     SyncResult,
 };
-pub use metadata::SyncMetadataClient;
-pub use errors::{
-    TolstoyError,
-    Result,
+pub mod tx_mapper;
+pub use tx_mapper::{
+    TxMapper,
 };
+pub mod tx_processor;
 pub mod types;
 pub use types::{
     Tx,

tolstoy/src/metadata.rs

@@ -19,15 +19,10 @@ use errors::{
     Result,
 };
 
-pub trait HeadTrackable {
-    fn remote_head(tx: &rusqlite::Transaction) -> Result<Uuid>;
-    fn set_remote_head(tx: &rusqlite::Transaction, uuid: &Uuid) -> Result<()>;
-}
-
 pub struct SyncMetadataClient {}
 
-impl HeadTrackable for SyncMetadataClient {
-    fn remote_head(tx: &rusqlite::Transaction) -> Result<Uuid> {
+impl SyncMetadataClient {
+    pub fn remote_head(tx: &rusqlite::Transaction) -> Result<Uuid> {
         tx.query_row(
             "SELECT value FROM tolstoy_metadata WHERE key = ?",
             &[&schema::REMOTE_HEAD_KEY], |r| {
@@ -37,7 +32,7 @@ impl HeadTrackable for SyncMetadataClient {
         )?.map_err(|e| e.into())
     }
 
-    fn set_remote_head(tx: &rusqlite::Transaction, uuid: &Uuid) -> Result<()> {
+    pub fn set_remote_head(tx: &rusqlite::Transaction, uuid: &Uuid) -> Result<()> {
         let uuid_bytes = uuid.as_bytes().to_vec();
         let updated = tx.execute("UPDATE tolstoy_metadata SET value = ? WHERE key = ?",
             &[&uuid_bytes, &schema::REMOTE_HEAD_KEY])?;

tolstoy/src/parts.rs

@@ -0,0 +1,63 @@
+// Copyright 2018 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 mentat_core::{
+    Entid,
+    TypedValue,
+    Keyword,
+};
+
+use types::TxPart;
+
+pub struct PartsHelper<'a> {
+    parts: &'a Vec<TxPart>,
+}
+
+impl<'a> PartsHelper<'a> {
+    pub fn new(parts: &'a Vec<TxPart>) -> PartsHelper {
+        PartsHelper {
+            parts: parts,
+        }
+    }
+
+    // TODO these are obviously quite inefficient
+    pub fn e_lookup(&self, e: Keyword) -> Option<Entid> {
+        // This wraps Keyword (e) in ValueRc (aliased Arc), which is rather expensive.
+        let kw_e = TypedValue::Keyword(e.into());
+
+        for part in self.parts {
+            if kw_e == part.v {
+                return Some(part.e);
+            }
+        }
+
+        None
+    }
+
+    pub fn ea_lookup(&self, e: Keyword, a: Keyword) -> Option<&TypedValue> {
+        let e_e = self.e_lookup(e);
+        let a_e = self.e_lookup(a);
+
+        if e_e.is_none() || a_e.is_none() {
+            return None;
+        }
+
+        let e_e = e_e.unwrap();
+        let a_e = a_e.unwrap();
+
+        for part in self.parts {
+            if part.e == e_e && part.a == a_e {
+                return Some(&part.v);
+            }
+        }
+
+        None
+    }
+}

tolstoy/src/remote_client.rs

@@ -111,7 +111,6 @@ impl RemoteClient {
         let uri = uri.parse()?;
 
         d(&format!("parsed uri {:?}", uri));
-
         let work = client.get(uri).and_then(|res| {
             println!("Response: {}", res.status());
 

tolstoy/src/schema.rs

@@ -11,15 +11,15 @@
 use rusqlite;
 use errors::Result;
 
-pub static REMOTE_HEAD_KEY: &str = r#"remote_head"#;
+pub static REMOTE_HEAD_KEY: &str = r"remote_head";
 
 lazy_static! {
     /// SQL statements to be executed, in order, to create the Tolstoy SQL schema (version 1).
     #[cfg_attr(rustfmt, rustfmt_skip)]
     static ref SCHEMA_STATEMENTS: Vec<&'static str> = { vec![
-        r#"CREATE TABLE IF NOT EXISTS tolstoy_tu (tx INTEGER PRIMARY KEY, uuid BLOB NOT NULL UNIQUE) WITHOUT ROWID"#,
-        r#"CREATE TABLE IF NOT EXISTS tolstoy_metadata (key BLOB NOT NULL UNIQUE, value BLOB NOT NULL)"#,
-        r#"CREATE INDEX IF NOT EXISTS idx_tolstoy_tu_ut ON tolstoy_tu (uuid, tx)"#,
+        "CREATE TABLE IF NOT EXISTS tolstoy_tu (tx INTEGER PRIMARY KEY, uuid BLOB NOT NULL UNIQUE) WITHOUT ROWID",
+        "CREATE TABLE IF NOT EXISTS tolstoy_metadata (key BLOB NOT NULL UNIQUE, value BLOB NOT NULL)",
+        "CREATE INDEX IF NOT EXISTS idx_tolstoy_tu_ut ON tolstoy_tu (uuid, tx)",
         ]
     };
 }