chore: consistent hash for snapshots

Author: Horusiath

Cargo.lock

@@ -33,6 +33,7 @@ async-trait = "0.1"
 arc-swap = { version = "1.7" }
 uuid = { version = "1.17.0", features = ["v4", "v5", "v7", "serde"] }
 futures-lite = { version = "2.6.0", features = ["futures-io"] }
+blake3 = "1.8"
 
 [patch.crates-io]
 # We're using a specific commit here because rust-rocksdb doesn't publish the latest version that includes the memory alignment fix.

Cargo.toml

@@ -26,6 +26,7 @@ chrono = "0.4.22"
 unicode-segmentation = "1.10.1"
 lazy_static = "1.4.0"
 fastrand = "2.1.0"
+blake3.workspace = true
 
 [target.'cfg(target_arch = "wasm32")'.dependencies]
 web-sys = { version = "0.3" }

collab/Cargo.toml

@@ -5,11 +5,10 @@ use std::panic;
 use std::panic::AssertUnwindSafe;
 
 use arc_swap::ArcSwapOption;
+use blake3::Hasher;
+use serde_json::json;
 use std::sync::Arc;
 use std::vec::IntoIter;
-
-use serde_json::json;
-
 use tokio_stream::wrappers::WatchStream;
 use tracing::trace;
 use yrs::block::{ClientID, Prelim};
@@ -241,6 +240,55 @@ pub fn make_yrs_doc(object_id: &str, skip_gc: bool, client_id: ClientID) -> Doc
 
 pub type CollabVersion = Uuid;
 
+pub trait ConsistentHash {
+  fn hash(&self, h: &mut blake3::Hasher);
+  fn consistent_hash(&self) -> u128 {
+    use blake3::Hasher;
+    let mut h = Hasher::new();
+
+    self.hash(&mut h);
+
+    let mut hash = [0; 16];
+    h.finalize_xof().fill(&mut hash);
+    u128::from_be_bytes(hash)
+  }
+}
+
+impl ConsistentHash for yrs::StateVector {
+  fn hash(&self, h: &mut Hasher) {
+    let mut clients = self.iter().map(|(k, _)| k).collect::<Vec<_>>();
+    clients.sort();
+    for client in clients {
+      let clock = self.get(client);
+      h.update(&client.to_be_bytes());
+      h.update(&clock.to_be_bytes());
+    }
+  }
+}
+
+impl ConsistentHash for yrs::DeleteSet {
+  fn hash(&self, h: &mut Hasher) {
+    let mut clients = self.iter().map(|(c, _)| c).collect::<Vec<_>>();
+    clients.sort();
+    for client in clients {
+      if let Some(range) = self.range(client) {
+        h.update(&client.to_be_bytes());
+        for r in range.iter() {
+          h.update(&r.start.to_be_bytes());
+          h.update(&r.end.to_be_bytes());
+        }
+      }
+    }
+  }
+}
+
+impl ConsistentHash for yrs::Snapshot {
+  fn hash(&self, h: &mut Hasher) {
+    ConsistentHash::hash(&self.state_map, h);
+    ConsistentHash::hash(&self.delete_set, h);
+  }
+}
+
 pub struct CollabOptions {
   pub object_id: Uuid,
   pub data_source: Option<DataSource>,

collab/src/core/collab.rs

@@ -1,7 +1,8 @@
+use collab::core::collab::ConsistentHash;
 use serde_json::json;
 use yrs::types::ToJson;
 use yrs::updates::decoder::Decode;
-use yrs::{Doc, Map, MapPrelim, MapRef, ReadTxn, Transact, Update};
+use yrs::{Doc, Map, MapPrelim, MapRef, ReadTxn, StateVector, Text, Transact, Update};
 
 #[tokio::test]
 async fn state_vec_apply_test() {
@@ -120,6 +121,35 @@ async fn two_way_sync_result_undetermined() {
   assert_eq!(a, b);
 }
 
+#[test]
+fn snapshot_produces_consistent_hash() {
+  let d1 = Doc::with_client_id(0xdeadbeef);
+  let txt1 = d1.get_or_insert_text("text");
+  let mut t1 = d1.transact_mut();
+  txt1.insert(&mut t1, 0, "Hello world!");
+  txt1.remove_range(&mut t1, 4, 5);
+
+  // we need at least 2 client IDs to produce a non-trivial state vector where order could be
+  // possibly different between runs
+  let d2 = Doc::with_client_id(123);
+  let txt2 = d2.get_or_insert_text("text");
+  let mut t2 = d2.transact_mut();
+  t2.apply_update(
+    Update::decode_v1(&t1.encode_state_as_update_v1(&StateVector::default())).unwrap(),
+  )
+  .unwrap();
+  txt2.insert(&mut t2, 0, "Acronym!");
+  txt2.remove_range(&mut t2, 1, 2);
+
+  let snapshot = t2.snapshot();
+
+  // We're going to use consistent hash to uniquely identify the snapshot based on its internal
+  // state. This way we can use it as ID and quickly compare if two snapshots are identical.
+  let hash = snapshot.consistent_hash();
+  let expected: u128 = 42481876838278106308370919647884892234; // produced in previous run
+  assert_eq!(hash, expected);
+}
+
 #[tokio::test]
 async fn two_way_sync_test() {
   let doc_1 = Doc::new();