feat(bindings): a replica through the binding facade and the browser Database

src/applier.rs

@@ -45,12 +45,20 @@ use serde_json::Value as Json;
 use sha2::{Digest, Sha256};
 
 use crate::error::Result;
-use crate::store::{Datum, Store};
+#[cfg(test)]
+use crate::store::Datum;
+use crate::store::Store;
+#[cfg(test)]
 use crate::types::{Op, Value};
 
 #[path = "applier_digest.rs"]
 mod digest;
+#[path = "applier_record.rs"]
+mod record;
+#[path = "applier_start.rs"]
+mod start;
 pub use digest::Stop;
+pub use record::record_to_datum;
 
 /// The source log a replica follows.
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -397,6 +405,11 @@ impl Applier {
             };
             by_tx.entry(tx).or_default().push(record);
         }
+        if self.progress.is_none()
+            && let Some(reason) = Self::new_replica_refusal(store)?
+        {
+            return Ok(Outcome::Refused { reason });
+        }
         let mut applied = Vec::new();
         for (tx, records) in by_tx {
             if self.progress.as_ref().is_some_and(|p| tx <= p.source_tx) {
@@ -503,94 +516,6 @@ fn apply_one_tx(
     }
 }
 
-/// One feed record as the datum that reproduces it, or why it cannot be
-/// reproduced exactly. The outer `Result` is a store failure.
-///
-/// # Errors
-/// [`Error::Sqlite`] if interning a term fails.
-pub fn record_to_datum(
-    store: &Store,
-    record: &Json,
-    tx_timestamp: &str,
-) -> Result<std::result::Result<Datum, String>> {
-    let (Some(entity), Some(attribute), Some(op)) = (
-        record["entity"].as_str(),
-        record["attribute"].as_str(),
-        record["op"].as_str(),
-    ) else {
-        return Ok(Err("record lacks entity, attribute or op".into()));
-    };
-    let retract = match digest::op_is_retract(op) {
-        Ok(r) => r,
-        Err(reason) => return Ok(Err(reason)),
-    };
-    let value_field = if retract { "old_value" } else { "value" };
-    let value = match value_of(store, &record[value_field])? {
-        Ok(v) => v,
-        Err(reason) => return Ok(Err(reason)),
-    };
-    let valid_from = record["valid_from"]
-        .as_str()
-        .unwrap_or(tx_timestamp)
-        .to_string();
-    Ok(Ok(Datum {
-        entity: store.intern(entity)?,
-        attribute: store.intern(attribute)?,
-        value,
-        valid_from,
-        // A retract names the fact it closes; its own row's valid time
-        // describes the closing, not an interval to replay.
-        valid_to: if retract {
-            None
-        } else {
-            record["valid_to"].as_str().map(String::from)
-        },
-        op: if retract { Op::Retract } else { Op::Assert },
-    }))
-}
-
-fn value_of(store: &Store, json: &Json) -> Result<std::result::Result<Value, String>> {
-    use base64::Engine as _;
-    Ok(Ok(match json {
-        Json::String(s) => Value::Str(s.clone()),
-        Json::Bool(b) => Value::Bool(*b),
-        Json::Number(n) => match n.as_i64() {
-            Some(i) => Value::Int(i),
-            None => match n.as_f64() {
-                Some(f) => Value::Float(f),
-                None => return Ok(Err(format!("unrepresentable number {n}"))),
-            },
-        },
-        Json::Object(o) if o.get("ref").and_then(Json::as_str).is_some() => {
-            Value::Ref(store.intern(o["ref"].as_str().unwrap_or_default())?)
-        }
-        Json::Object(o) if o.contains_key("lang") => Value::Lang {
-            lexical: o["value"].as_str().unwrap_or_default().into(),
-            lang: o["lang"].as_str().unwrap_or_default().into(),
-        },
-        Json::Object(o) if o.contains_key("datatype") => Value::Typed {
-            lexical: o["value"].as_str().unwrap_or_default().into(),
-            datatype: o["datatype"].as_str().unwrap_or_default().into(),
-        },
-        Json::Object(o) if o.get("base64").and_then(Json::as_str).is_some() => {
-            match base64::engine::general_purpose::STANDARD
-                .decode(o["base64"].as_str().unwrap_or_default())
-            {
-                Ok(bytes) => Value::Bytes(bytes),
-                Err(_) => return Ok(Err("byte value is not valid base64".into())),
-            }
-        }
-        // A source from before byte bodies were carried sends only the length.
-        // Writing empty bytes would be a silent loss: refuse instead.
-        Json::Object(o) if o.contains_key("bytes_len") => {
-            return Ok(Err(
-                "byte value sent as its length only; the source predates lossless bytes".into(),
-            ));
-        }
-        other => return Ok(Err(format!("unreplayable value {other}"))),
-    }))
-}
-
 /// Live facts of a store as resolved strings, so two stores with different
 /// term ids compare by meaning; valid time is part of a fact's identity here.
 /// Shared by the applier and feed-replay conformance tests.

src/applier_record.rs

@@ -0,0 +1,97 @@
+//! Feed records decoded into the datums that reproduce them exactly, or
+//! refused when they cannot be.
+
+use serde_json::Value as Json;
+
+use super::digest;
+use crate::error::Result;
+use crate::store::{Datum, Store};
+use crate::types::{Op, Value};
+
+/// One feed record as the datum that reproduces it, or why it cannot be
+/// reproduced exactly. The outer `Result` is a store failure.
+///
+/// # Errors
+/// [`Error::Sqlite`] if interning a term fails.
+pub fn record_to_datum(
+    store: &Store,
+    record: &Json,
+    tx_timestamp: &str,
+) -> Result<std::result::Result<Datum, String>> {
+    let (Some(entity), Some(attribute), Some(op)) = (
+        record["entity"].as_str(),
+        record["attribute"].as_str(),
+        record["op"].as_str(),
+    ) else {
+        return Ok(Err("record lacks entity, attribute or op".into()));
+    };
+    let retract = match digest::op_is_retract(op) {
+        Ok(r) => r,
+        Err(reason) => return Ok(Err(reason)),
+    };
+    let value_field = if retract { "old_value" } else { "value" };
+    let value = match value_of(store, &record[value_field])? {
+        Ok(v) => v,
+        Err(reason) => return Ok(Err(reason)),
+    };
+    let valid_from = record["valid_from"]
+        .as_str()
+        .unwrap_or(tx_timestamp)
+        .to_string();
+    Ok(Ok(Datum {
+        entity: store.intern(entity)?,
+        attribute: store.intern(attribute)?,
+        value,
+        valid_from,
+        // A retract names the fact it closes; its own row's valid time
+        // describes the closing, not an interval to replay.
+        valid_to: if retract {
+            None
+        } else {
+            record["valid_to"].as_str().map(String::from)
+        },
+        op: if retract { Op::Retract } else { Op::Assert },
+    }))
+}
+
+fn value_of(store: &Store, json: &Json) -> Result<std::result::Result<Value, String>> {
+    use base64::Engine as _;
+    Ok(Ok(match json {
+        Json::String(s) => Value::Str(s.clone()),
+        Json::Bool(b) => Value::Bool(*b),
+        Json::Number(n) => match n.as_i64() {
+            Some(i) => Value::Int(i),
+            None => match n.as_f64() {
+                Some(f) => Value::Float(f),
+                None => return Ok(Err(format!("unrepresentable number {n}"))),
+            },
+        },
+        Json::Object(o) if o.get("ref").and_then(Json::as_str).is_some() => {
+            Value::Ref(store.intern(o["ref"].as_str().unwrap_or_default())?)
+        }
+        Json::Object(o) if o.contains_key("lang") => Value::Lang {
+            lexical: o["value"].as_str().unwrap_or_default().into(),
+            lang: o["lang"].as_str().unwrap_or_default().into(),
+        },
+        Json::Object(o) if o.contains_key("datatype") => Value::Typed {
+            lexical: o["value"].as_str().unwrap_or_default().into(),
+            datatype: o["datatype"].as_str().unwrap_or_default().into(),
+        },
+        Json::Object(o) if o.get("base64").and_then(Json::as_str).is_some() => {
+            match base64::engine::general_purpose::STANDARD
+                .decode(o["base64"].as_str().unwrap_or_default())
+            {
+                Ok(bytes) => Value::Bytes(bytes),
+                Err(_) => return Ok(Err("byte value is not valid base64".into())),
+            }
+        }
+        // A source from before byte bodies were carried sends only the length.
+        // Writing empty bytes would be a silent loss: refuse instead.
+        Json::Object(o) if o.contains_key("bytes_len") => {
+            return Ok(Err(
+                "byte value sent as its length only; the source predates lossless bytes".into(),
+            ));
+        }
+        other => return Ok(Err(format!("unreplayable value {other}"))),
+    }))
+}

src/applier_start.rs

@@ -0,0 +1,41 @@
+//! Which stores may become a replica. A replica must equal its source at its
+//! cursor, so it starts with no facts of its own (or from a verified
+//! bootstrap, which refuses a store holding facts on its own).
+
+use super::Applier;
+use crate::error::Result;
+use crate::store::Store;
+
+impl Applier {
+    /// Why `store` cannot become a new replica, or `None` when it can: it
+    /// already follows a source, or it holds no facts (shapes and settings are
+    /// not facts). Local facts with no cursor would be mixed into the replica,
+    /// which would then equal no source while claiming to.
+    ///
+    /// # Errors
+    /// [`Error::Sqlite`] if the store cannot be read.
+    pub fn new_replica_refusal(store: &Store) -> Result<Option<String>> {
+        let follows: bool = store.conn.query_row(
+            "SELECT EXISTS (SELECT 1 FROM sqlite_master \
+             WHERE type = 'table' AND name = 'replica_progress')",
+            [],
+            |r| r.get(0),
+        )?;
+        if follows
+            && store
+                .conn
+                .query_row("SELECT EXISTS (SELECT 1 FROM replica_progress)", [], |r| {
+                    r.get::<_, bool>(0)
+                })?
+        {
+            return Ok(None);
+        }
+        let holds_facts: bool =
+            store
+                .conn
+                .query_row("SELECT EXISTS (SELECT 1 FROM facts)", [], |r| r.get(0))?;
+        Ok(holds_facts.then(|| {
+            "a new replica must start with no facts; this store holds local facts".to_string()
+        }))
+    }
+}

src/applier_tests.rs

@@ -531,3 +531,31 @@ fn outcomes_serialize_tagged_by_kind() {
         json!({"source_tx": 9, "last_frame_age_ms": null, "bounded": false})
     );
 }
+
+#[test]
+fn a_store_holding_local_facts_cannot_start_a_replica() {
+    let src = source();
+    let feed = pages(&src);
+    let mut replica = Store::open_in_memory().unwrap();
+    let local = d(
+        &replica,
+        "urn:local",
+        "urn:p:title",
+        Value::Str("mine".into()),
+        Op::Assert,
+    );
+    replica.transact(&[local], T, None, None).unwrap();
+    let before = live_digest(&replica);
+    let mut applier = Applier::open(&replica).unwrap();
+    let o = applier
+        .apply_frame(&mut replica, "quipu.changes", &feed[0], 1)
+        .unwrap();
+    assert!(matches!(o, Outcome::Refused { .. }), "{o:?}");
+    assert!(applier.progress().is_none());
+    assert_eq!(live_digest(&replica), before);
+    // Once a replica follows a source, its own facts are the source's.
+    let mut empty = Store::open_in_memory().unwrap();
+    let mut follower = Applier::open(&empty).unwrap();
+    replay(&mut empty, &mut follower, &feed[..1]);
+    assert_eq!(Applier::new_replica_refusal(&empty).unwrap(), None);
+}