feat: Implement new synchronization state management and deterministic head sorting, replacing the old vector clock module.
This commit is contained in:
@@ -4,7 +4,7 @@
|
||||
//! - **Node**: Identity with Ed25519 keypair
|
||||
//! - **SigChain**: Append-only cryptographically signed log
|
||||
//! - **Entry**: Atomic operations in the log
|
||||
//! - **VectorClock**: Causality tracking for reconciliation
|
||||
//! - **SyncState**: Per-author sequence tracking for reconciliation
|
||||
//! - **HLC**: Hybrid Logical Clock for ordering
|
||||
//! - **Clock**: Time abstraction for testability
|
||||
//! - **Proto**: Generated protobuf types from lattice.proto
|
||||
@@ -16,7 +16,7 @@
|
||||
pub mod node;
|
||||
pub mod sigchain;
|
||||
pub mod entry;
|
||||
pub mod vector_clock;
|
||||
pub mod sync_state;
|
||||
pub mod hlc;
|
||||
pub mod clock;
|
||||
pub mod proto;
|
||||
@@ -33,12 +33,12 @@ pub const MAX_ENTRY_SIZE: usize = 16 * 1024 * 1024;
|
||||
pub use node::Node;
|
||||
pub use sigchain::SigChain;
|
||||
pub use entry::Entry;
|
||||
pub use vector_clock::VectorClock;
|
||||
pub use sync_state::{SyncState, AuthorInfo, MissingRange};
|
||||
pub use hlc::HLC;
|
||||
pub use clock::{Clock, SystemClock, MockClock};
|
||||
pub use data_dir::DataDir;
|
||||
pub use signed_entry::{EntryBuilder, sign_entry, verify_signed_entry, hash_signed_entry};
|
||||
pub use log::{append_entry, read_entries, LogReader};
|
||||
pub use log::{append_entry, read_entries, read_entries_after, LogReader};
|
||||
pub use store::Store;
|
||||
pub use meta_store::MetaStore;
|
||||
pub use proto::HeadInfo;
|
||||
|
||||
+415
-10
@@ -206,25 +206,40 @@ impl Store {
|
||||
}
|
||||
}
|
||||
|
||||
/// Get all heads for a key (for conflict inspection)
|
||||
/// Get all heads for a key (for conflict inspection).
|
||||
/// Heads are sorted deterministically: highest HLC first, ties broken by author.
|
||||
pub fn get_heads(&self, key: &[u8]) -> Result<Vec<HeadInfo>, StoreError> {
|
||||
let read_txn = self.db.begin_read()?;
|
||||
let table = read_txn.open_table(KV_TABLE)?;
|
||||
|
||||
match table.get(key)? {
|
||||
Some(v) => Ok(HeadList::decode(v.value())?.heads),
|
||||
Some(v) => {
|
||||
let mut heads = HeadList::decode(v.value())?.heads;
|
||||
// Sort by winner criteria: highest HLC first, then highest author (deterministic)
|
||||
heads.sort_by(|a, b| {
|
||||
b.hlc.cmp(&a.hlc)
|
||||
.then_with(|| b.author.cmp(&a.author))
|
||||
});
|
||||
Ok(heads)
|
||||
}
|
||||
None => Ok(Vec::new()),
|
||||
}
|
||||
}
|
||||
|
||||
/// Pick deterministic winner from heads: highest HLC, then highest author bytes
|
||||
/// Pick deterministic winner from heads: highest HLC, then highest author bytes.
|
||||
/// Heads should already be sorted by get_heads(), so winner is first.
|
||||
fn pick_winner(heads: &[HeadInfo]) -> Option<&HeadInfo> {
|
||||
heads.iter().max_by(|a, b| {
|
||||
match a.hlc.cmp(&b.hlc) {
|
||||
std::cmp::Ordering::Equal => a.author.cmp(&b.author),
|
||||
ord => ord,
|
||||
}
|
||||
})
|
||||
// If heads are already sorted (via get_heads), first is winner
|
||||
// If not sorted, compute winner via max
|
||||
if heads.is_empty() {
|
||||
None
|
||||
} else {
|
||||
// Use max_by for correctness even on unsorted input
|
||||
heads.iter().max_by(|a, b| {
|
||||
a.hlc.cmp(&b.hlc)
|
||||
.then_with(|| a.author.cmp(&b.author))
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
/// List all key-value pairs (winner values only)
|
||||
@@ -270,6 +285,34 @@ impl Store {
|
||||
None => Ok(None),
|
||||
}
|
||||
}
|
||||
|
||||
/// Get sync state for all authors (for reconciliation).
|
||||
///
|
||||
/// Returns a SyncState with each author's highest seen sequence number and hash.
|
||||
pub fn sync_state(&self) -> Result<crate::sync_state::SyncState, StoreError> {
|
||||
use crate::sync_state::SyncState;
|
||||
|
||||
let read_txn = self.db.begin_read()?;
|
||||
let table = read_txn.open_table(AUTHOR_TABLE)?;
|
||||
|
||||
let mut state = SyncState::new();
|
||||
for entry in table.iter()? {
|
||||
let (key, value) = entry?;
|
||||
if key.value().len() == 32 {
|
||||
if let Ok(author_state) = AuthorState::decode(value.value()) {
|
||||
let mut author = [0u8; 32];
|
||||
author.copy_from_slice(key.value());
|
||||
let mut hash = [0u8; 32];
|
||||
if author_state.hash.len() == 32 {
|
||||
hash.copy_from_slice(&author_state.hash);
|
||||
}
|
||||
state.set(author, author_state.seq, hash);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Ok(state)
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
@@ -282,7 +325,8 @@ mod tests {
|
||||
use std::env::temp_dir;
|
||||
|
||||
fn temp_db_path(name: &str) -> std::path::PathBuf {
|
||||
temp_dir().join(format!("lattice_dag_store_test_{}.db", name))
|
||||
let tid = std::thread::current().id();
|
||||
temp_dir().join(format!("lattice_dag_store_test_{}_{:?}.db", name, tid))
|
||||
}
|
||||
|
||||
const TEST_STORE: [u8; 16] = [1u8; 16];
|
||||
@@ -1064,4 +1108,365 @@ mod tests {
|
||||
}];
|
||||
assert!(!Store::needs_delete(&heads));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_sync_state_diff_and_apply() {
|
||||
// Test that two stores can compute diff and sync entries
|
||||
let path_a = temp_db_path("sync_a");
|
||||
let path_b = temp_db_path("sync_b");
|
||||
let log_path_a = temp_db_path("sync_a_log");
|
||||
let _ = std::fs::remove_file(&path_a);
|
||||
let _ = std::fs::remove_file(&path_b);
|
||||
let _ = std::fs::remove_file(&log_path_a);
|
||||
|
||||
// Node A writes some entries
|
||||
let store_a = Store::open(&path_a).unwrap();
|
||||
let node_a = Node::generate();
|
||||
|
||||
// Write 3 entries on node A
|
||||
for i in 1u64..=3 {
|
||||
let clock = MockClock::new(1000 + i * 100);
|
||||
let entry = EntryBuilder::new(i, HLC::now_with_clock(&clock))
|
||||
.store_id(TEST_STORE.to_vec())
|
||||
.prev_hash([0u8; 32].to_vec())
|
||||
.put(format!("/key{}", i), format!("value{}", i).into_bytes())
|
||||
.sign(&node_a);
|
||||
store_a.apply_entry(&entry).unwrap();
|
||||
crate::log::append_entry(&log_path_a, &entry).unwrap();
|
||||
}
|
||||
|
||||
// Node B is empty
|
||||
let store_b = Store::open(&path_b).unwrap();
|
||||
|
||||
// Get sync states
|
||||
let sync_a = store_a.sync_state().unwrap();
|
||||
let sync_b = store_b.sync_state().unwrap();
|
||||
|
||||
// Compute diff: B needs entries from A
|
||||
let missing = sync_b.diff(&sync_a);
|
||||
|
||||
// Should need entries for author A
|
||||
assert_eq!(missing.len(), 1);
|
||||
assert_eq!(missing[0].author, node_a.public_key_bytes());
|
||||
assert_eq!(missing[0].from_seq, 0); // B has nothing
|
||||
assert_eq!(missing[0].to_seq, 3); // A has 3 entries
|
||||
|
||||
// Fetch entries from A's log (using from_hash = 0 means read all)
|
||||
let entries = crate::log::read_entries_after(
|
||||
&log_path_a,
|
||||
if missing[0].from_hash == [0u8; 32] { None } else { Some(missing[0].from_hash) }
|
||||
).unwrap();
|
||||
assert_eq!(entries.len(), 3);
|
||||
|
||||
// Apply entries to B
|
||||
for entry in &entries {
|
||||
store_b.apply_entry(entry).unwrap();
|
||||
}
|
||||
|
||||
// Verify B has same KV state as A
|
||||
assert_eq!(store_b.get(b"/key1").unwrap(), Some(b"value1".to_vec()));
|
||||
assert_eq!(store_b.get(b"/key2").unwrap(), Some(b"value2".to_vec()));
|
||||
assert_eq!(store_b.get(b"/key3").unwrap(), Some(b"value3".to_vec()));
|
||||
|
||||
// Verify sync states now match
|
||||
let sync_a_after = store_a.sync_state().unwrap();
|
||||
let sync_b_after = store_b.sync_state().unwrap();
|
||||
assert!(sync_b_after.diff(&sync_a_after).is_empty());
|
||||
|
||||
let _ = std::fs::remove_file(path_a);
|
||||
let _ = std::fs::remove_file(path_b);
|
||||
let _ = std::fs::remove_file(log_path_a);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_bidirectional_sync() {
|
||||
// Test that two stores can sync in both directions
|
||||
let path_a = temp_db_path("bidir_a");
|
||||
let path_b = temp_db_path("bidir_b");
|
||||
let log_path_a = temp_db_path("bidir_log_a");
|
||||
let log_path_b = temp_db_path("bidir_log_b");
|
||||
let _ = std::fs::remove_file(&path_a);
|
||||
let _ = std::fs::remove_file(&path_b);
|
||||
let _ = std::fs::remove_file(&log_path_a);
|
||||
let _ = std::fs::remove_file(&log_path_b);
|
||||
|
||||
let store_a = Store::open(&path_a).unwrap();
|
||||
let store_b = Store::open(&path_b).unwrap();
|
||||
let node_a = Node::generate();
|
||||
let node_b = Node::generate();
|
||||
|
||||
// Node A writes entries
|
||||
for i in 1u64..=2 {
|
||||
let clock = MockClock::new(1000 + i * 100);
|
||||
let entry = EntryBuilder::new(i, HLC::now_with_clock(&clock))
|
||||
.store_id(TEST_STORE.to_vec())
|
||||
.prev_hash([0u8; 32].to_vec())
|
||||
.put(format!("/a{}", i), format!("from_a{}", i).into_bytes())
|
||||
.sign(&node_a);
|
||||
store_a.apply_entry(&entry).unwrap();
|
||||
crate::log::append_entry(&log_path_a, &entry).unwrap();
|
||||
}
|
||||
|
||||
// Node B writes different entries
|
||||
for i in 1u64..=2 {
|
||||
let clock = MockClock::new(2000 + i * 100);
|
||||
let entry = EntryBuilder::new(i, HLC::now_with_clock(&clock))
|
||||
.store_id(TEST_STORE.to_vec())
|
||||
.prev_hash([0u8; 32].to_vec())
|
||||
.put(format!("/b{}", i), format!("from_b{}", i).into_bytes())
|
||||
.sign(&node_b);
|
||||
store_b.apply_entry(&entry).unwrap();
|
||||
crate::log::append_entry(&log_path_b, &entry).unwrap();
|
||||
}
|
||||
|
||||
// Get sync states
|
||||
let sync_a = store_a.sync_state().unwrap();
|
||||
let sync_b = store_b.sync_state().unwrap();
|
||||
|
||||
// A needs B's entries
|
||||
let a_needs = sync_a.diff(&sync_b);
|
||||
assert_eq!(a_needs.len(), 1);
|
||||
assert_eq!(a_needs[0].author, node_b.public_key_bytes());
|
||||
|
||||
// B needs A's entries
|
||||
let b_needs = sync_b.diff(&sync_a);
|
||||
assert_eq!(b_needs.len(), 1);
|
||||
assert_eq!(b_needs[0].author, node_a.public_key_bytes());
|
||||
|
||||
// Sync A → B
|
||||
let entries_a = crate::log::read_entries(&log_path_a).unwrap();
|
||||
for entry in &entries_a {
|
||||
store_b.apply_entry(entry).unwrap();
|
||||
}
|
||||
|
||||
// Sync B → A
|
||||
let entries_b = crate::log::read_entries(&log_path_b).unwrap();
|
||||
for entry in &entries_b {
|
||||
store_a.apply_entry(entry).unwrap();
|
||||
}
|
||||
|
||||
// Both should now have all 4 keys
|
||||
assert_eq!(store_a.get(b"/a1").unwrap(), Some(b"from_a1".to_vec()));
|
||||
assert_eq!(store_a.get(b"/b1").unwrap(), Some(b"from_b1".to_vec()));
|
||||
assert_eq!(store_b.get(b"/a1").unwrap(), Some(b"from_a1".to_vec()));
|
||||
assert_eq!(store_b.get(b"/b1").unwrap(), Some(b"from_b1".to_vec()));
|
||||
|
||||
// Sync states should match
|
||||
let sync_a_after = store_a.sync_state().unwrap();
|
||||
let sync_b_after = store_b.sync_state().unwrap();
|
||||
assert!(sync_a_after.diff(&sync_b_after).is_empty());
|
||||
assert!(sync_b_after.diff(&sync_a_after).is_empty());
|
||||
|
||||
let _ = std::fs::remove_file(path_a);
|
||||
let _ = std::fs::remove_file(path_b);
|
||||
let _ = std::fs::remove_file(log_path_a);
|
||||
let _ = std::fs::remove_file(log_path_b);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_three_way_sync() {
|
||||
// Test that three stores can all sync with each other
|
||||
let path_a = temp_db_path("three_a");
|
||||
let path_b = temp_db_path("three_b");
|
||||
let path_c = temp_db_path("three_c");
|
||||
let log_path_a = temp_db_path("three_log_a");
|
||||
let log_path_b = temp_db_path("three_log_b");
|
||||
let log_path_c = temp_db_path("three_log_c");
|
||||
for p in [&path_a, &path_b, &path_c, &log_path_a, &log_path_b, &log_path_c] {
|
||||
let _ = std::fs::remove_file(p);
|
||||
}
|
||||
|
||||
let store_a = Store::open(&path_a).unwrap();
|
||||
let store_b = Store::open(&path_b).unwrap();
|
||||
let store_c = Store::open(&path_c).unwrap();
|
||||
let node_a = Node::generate();
|
||||
let node_b = Node::generate();
|
||||
let node_c = Node::generate();
|
||||
|
||||
// Each node writes one entry
|
||||
let entry_a = EntryBuilder::new(1, HLC::now_with_clock(&MockClock::new(1000)))
|
||||
.store_id(TEST_STORE.to_vec())
|
||||
.prev_hash([0u8; 32].to_vec())
|
||||
.put("/key_a", b"from_a".to_vec())
|
||||
.sign(&node_a);
|
||||
store_a.apply_entry(&entry_a).unwrap();
|
||||
crate::log::append_entry(&log_path_a, &entry_a).unwrap();
|
||||
|
||||
let entry_b = EntryBuilder::new(1, HLC::now_with_clock(&MockClock::new(2000)))
|
||||
.store_id(TEST_STORE.to_vec())
|
||||
.prev_hash([0u8; 32].to_vec())
|
||||
.put("/key_b", b"from_b".to_vec())
|
||||
.sign(&node_b);
|
||||
store_b.apply_entry(&entry_b).unwrap();
|
||||
crate::log::append_entry(&log_path_b, &entry_b).unwrap();
|
||||
|
||||
let entry_c = EntryBuilder::new(1, HLC::now_with_clock(&MockClock::new(3000)))
|
||||
.store_id(TEST_STORE.to_vec())
|
||||
.prev_hash([0u8; 32].to_vec())
|
||||
.put("/key_c", b"from_c".to_vec())
|
||||
.sign(&node_c);
|
||||
store_c.apply_entry(&entry_c).unwrap();
|
||||
crate::log::append_entry(&log_path_c, &entry_c).unwrap();
|
||||
|
||||
// Sync A ↔ B
|
||||
for entry in crate::log::read_entries(&log_path_a).unwrap() {
|
||||
store_b.apply_entry(&entry).unwrap();
|
||||
}
|
||||
for entry in crate::log::read_entries(&log_path_b).unwrap() {
|
||||
store_a.apply_entry(&entry).unwrap();
|
||||
}
|
||||
|
||||
// Sync B ↔ C
|
||||
for entry in crate::log::read_entries(&log_path_b).unwrap() {
|
||||
store_c.apply_entry(&entry).unwrap();
|
||||
}
|
||||
for entry in crate::log::read_entries(&log_path_c).unwrap() {
|
||||
store_b.apply_entry(&entry).unwrap();
|
||||
}
|
||||
|
||||
// Sync A ↔ C (A should get C's entry, C should get A's entry)
|
||||
for entry in crate::log::read_entries(&log_path_a).unwrap() {
|
||||
store_c.apply_entry(&entry).unwrap();
|
||||
}
|
||||
for entry in crate::log::read_entries(&log_path_c).unwrap() {
|
||||
store_a.apply_entry(&entry).unwrap();
|
||||
}
|
||||
|
||||
// All three stores should have all three keys
|
||||
for store in [&store_a, &store_b, &store_c] {
|
||||
assert_eq!(store.get(b"/key_a").unwrap(), Some(b"from_a".to_vec()));
|
||||
assert_eq!(store.get(b"/key_b").unwrap(), Some(b"from_b".to_vec()));
|
||||
assert_eq!(store.get(b"/key_c").unwrap(), Some(b"from_c".to_vec()));
|
||||
}
|
||||
|
||||
// All sync states should match
|
||||
let sync_a = store_a.sync_state().unwrap();
|
||||
let sync_b = store_b.sync_state().unwrap();
|
||||
let sync_c = store_c.sync_state().unwrap();
|
||||
assert!(sync_a.diff(&sync_b).is_empty());
|
||||
assert!(sync_b.diff(&sync_c).is_empty());
|
||||
assert!(sync_c.diff(&sync_a).is_empty());
|
||||
|
||||
for p in [path_a, path_b, path_c, log_path_a, log_path_b, log_path_c] {
|
||||
let _ = std::fs::remove_file(p);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_conflict_deterministic_resolution() {
|
||||
// Test that two nodes writing the same key resolve deterministically
|
||||
let path_a = temp_db_path("conflict_a");
|
||||
let path_b = temp_db_path("conflict_b");
|
||||
let log_path_a = temp_db_path("conflict_log_a");
|
||||
let log_path_b = temp_db_path("conflict_log_b");
|
||||
for p in [&path_a, &path_b, &log_path_a, &log_path_b] {
|
||||
let _ = std::fs::remove_file(p);
|
||||
}
|
||||
|
||||
let store_a = Store::open(&path_a).unwrap();
|
||||
let store_b = Store::open(&path_b).unwrap();
|
||||
let node_a = Node::generate();
|
||||
let node_b = Node::generate();
|
||||
|
||||
// Both nodes write to the SAME key with different values
|
||||
// Use same HLC to force conflict (tie-break on author)
|
||||
let entry_a = EntryBuilder::new(1, HLC::now_with_clock(&MockClock::new(1000)))
|
||||
.store_id(TEST_STORE.to_vec())
|
||||
.prev_hash([0u8; 32].to_vec())
|
||||
.put("/shared_key", b"value_from_a".to_vec())
|
||||
.sign(&node_a);
|
||||
store_a.apply_entry(&entry_a).unwrap();
|
||||
crate::log::append_entry(&log_path_a, &entry_a).unwrap();
|
||||
|
||||
let entry_b = EntryBuilder::new(1, HLC::now_with_clock(&MockClock::new(1000))) // Same HLC!
|
||||
.store_id(TEST_STORE.to_vec())
|
||||
.prev_hash([0u8; 32].to_vec())
|
||||
.put("/shared_key", b"value_from_b".to_vec())
|
||||
.sign(&node_b);
|
||||
store_b.apply_entry(&entry_b).unwrap();
|
||||
crate::log::append_entry(&log_path_b, &entry_b).unwrap();
|
||||
|
||||
// Before sync: A has A's value, B has B's value
|
||||
assert_eq!(store_a.get(b"/shared_key").unwrap(), Some(b"value_from_a".to_vec()));
|
||||
assert_eq!(store_b.get(b"/shared_key").unwrap(), Some(b"value_from_b".to_vec()));
|
||||
|
||||
// Sync A → B and B → A
|
||||
for entry in crate::log::read_entries(&log_path_a).unwrap() {
|
||||
store_b.apply_entry(&entry).unwrap();
|
||||
}
|
||||
for entry in crate::log::read_entries(&log_path_b).unwrap() {
|
||||
store_a.apply_entry(&entry).unwrap();
|
||||
}
|
||||
|
||||
// After sync: both should have SAME value (deterministic winner)
|
||||
let value_a = store_a.get(b"/shared_key").unwrap();
|
||||
let value_b = store_b.get(b"/shared_key").unwrap();
|
||||
assert_eq!(value_a, value_b, "Conflict should resolve deterministically");
|
||||
|
||||
// Both should have 2 heads for this key (conflict)
|
||||
let heads_a = store_a.get_heads(b"/shared_key").unwrap();
|
||||
let heads_b = store_b.get_heads(b"/shared_key").unwrap();
|
||||
assert_eq!(heads_a.len(), 2, "Should have 2 heads (conflict)");
|
||||
assert_eq!(heads_b.len(), 2, "Should have 2 heads (conflict)");
|
||||
|
||||
// Both stores have the same heads in same order (deterministic)
|
||||
assert_eq!(heads_a[0].value, heads_b[0].value, "Winner should be same");
|
||||
assert_eq!(heads_a[0].author, heads_b[0].author, "Winner author should be same");
|
||||
|
||||
// Verify tie-breaker: winner is the one with higher author bytes (deterministic)
|
||||
// Since HLC is the same, the author with lexicographically higher bytes wins
|
||||
let winner_author = &heads_a[0].author;
|
||||
let loser_author = &heads_a[1].author;
|
||||
assert!(winner_author > loser_author, "Winner should have higher author bytes");
|
||||
|
||||
for p in [path_a, path_b, log_path_a, log_path_b] {
|
||||
let _ = std::fs::remove_file(p);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_hlc_tiebreak_explicit() {
|
||||
// Explicit test: equal HLC, winner determined by node ID (author bytes)
|
||||
let path = temp_db_path("tiebreak");
|
||||
let _ = std::fs::remove_file(&path);
|
||||
|
||||
let store = Store::open(&path).unwrap();
|
||||
let node_low = Node::generate();
|
||||
let node_high = Node::generate();
|
||||
|
||||
// Determine which node has "higher" author bytes
|
||||
let (high_node, low_node) = if node_high.public_key_bytes() > node_low.public_key_bytes() {
|
||||
(&node_high, &node_low)
|
||||
} else {
|
||||
(&node_low, &node_high)
|
||||
};
|
||||
|
||||
// Both entries have SAME HLC
|
||||
let clock = MockClock::new(5000);
|
||||
|
||||
let entry_low = EntryBuilder::new(1, HLC::now_with_clock(&clock))
|
||||
.store_id(TEST_STORE.to_vec())
|
||||
.prev_hash([0u8; 32].to_vec())
|
||||
.put("/tiebreak_key", b"from_low".to_vec())
|
||||
.sign(low_node);
|
||||
store.apply_entry(&entry_low).unwrap();
|
||||
|
||||
let entry_high = EntryBuilder::new(1, HLC::now_with_clock(&clock))
|
||||
.store_id(TEST_STORE.to_vec())
|
||||
.prev_hash([0u8; 32].to_vec())
|
||||
.put("/tiebreak_key", b"from_high".to_vec())
|
||||
.sign(high_node);
|
||||
store.apply_entry(&entry_high).unwrap();
|
||||
|
||||
// Winner should be the one with higher author bytes
|
||||
let value = store.get(b"/tiebreak_key").unwrap();
|
||||
assert_eq!(value, Some(b"from_high".to_vec()), "Higher author bytes should win");
|
||||
|
||||
let heads = store.get_heads(b"/tiebreak_key").unwrap();
|
||||
assert_eq!(heads.len(), 2);
|
||||
assert_eq!(heads[0].value, b"from_high".to_vec(), "heads[0] should be winner");
|
||||
assert_eq!(heads[0].author, high_node.public_key_bytes().to_vec());
|
||||
|
||||
let _ = std::fs::remove_file(path);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,179 @@
|
||||
//! Sync state for causality tracking and reconciliation
|
||||
|
||||
use std::collections::HashMap;
|
||||
|
||||
/// Author ID type (32-byte Ed25519 public key)
|
||||
pub type Author = [u8; 32];
|
||||
|
||||
/// Per-author sync information (seq + hash for resume).
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct AuthorInfo {
|
||||
pub seq: u64,
|
||||
pub hash: [u8; 32],
|
||||
}
|
||||
|
||||
/// Sync state tracking per-author sequence numbers and hashes.
|
||||
///
|
||||
/// Used during reconciliation to identify missing entries between peers.
|
||||
/// Each author's highest seen sequence number and hash is tracked.
|
||||
#[derive(Debug, Clone, Default)]
|
||||
pub struct SyncState {
|
||||
authors: HashMap<Author, AuthorInfo>,
|
||||
}
|
||||
|
||||
/// Describes entries needed from a peer for a specific author.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct MissingRange {
|
||||
pub author: Author,
|
||||
pub from_seq: u64, // exclusive - we have up to this
|
||||
pub from_hash: [u8; 32], // hash to resume reading after
|
||||
pub to_seq: u64, // inclusive - peer has up to this
|
||||
}
|
||||
|
||||
impl SyncState {
|
||||
/// Create a new empty sync state.
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
authors: HashMap::new(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Get the info for an author (returns None if not present).
|
||||
pub fn get(&self, author: &Author) -> Option<&AuthorInfo> {
|
||||
self.authors.get(author)
|
||||
}
|
||||
|
||||
/// Get the sequence number for an author (returns 0 if not present).
|
||||
pub fn seq(&self, author: &Author) -> u64 {
|
||||
self.authors.get(author).map(|i| i.seq).unwrap_or(0)
|
||||
}
|
||||
|
||||
/// Set the info for an author.
|
||||
pub fn set(&mut self, author: Author, seq: u64, hash: [u8; 32]) {
|
||||
self.authors.insert(author, AuthorInfo { seq, hash });
|
||||
}
|
||||
|
||||
/// Get all authors and their info.
|
||||
pub fn authors(&self) -> &HashMap<Author, AuthorInfo> {
|
||||
&self.authors
|
||||
}
|
||||
|
||||
/// Compute what entries we're missing compared to a peer's state.
|
||||
///
|
||||
/// Returns ranges of entries we need from the peer.
|
||||
/// Each range includes the hash to resume reading after.
|
||||
pub fn diff(&self, peer: &SyncState) -> Vec<MissingRange> {
|
||||
let mut missing = Vec::new();
|
||||
|
||||
for (author, peer_info) in peer.authors() {
|
||||
let my_seq = self.seq(author);
|
||||
if peer_info.seq > my_seq {
|
||||
// We need entries from my_seq+1 to peer_info.seq
|
||||
// Use our hash (or zero if we have nothing) as resume point
|
||||
let from_hash = self.get(author)
|
||||
.map(|i| i.hash)
|
||||
.unwrap_or([0u8; 32]);
|
||||
|
||||
missing.push(MissingRange {
|
||||
author: *author,
|
||||
from_seq: my_seq,
|
||||
from_hash,
|
||||
to_seq: peer_info.seq,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
missing
|
||||
}
|
||||
|
||||
/// Merge another sync state into this one (take max seq per author).
|
||||
pub fn merge(&mut self, other: &SyncState) {
|
||||
for (author, info) in other.authors() {
|
||||
let my_seq = self.seq(author);
|
||||
if info.seq > my_seq {
|
||||
self.set(*author, info.seq, info.hash);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn test_diff_empty() {
|
||||
let a = SyncState::new();
|
||||
let b = SyncState::new();
|
||||
assert!(a.diff(&b).is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_diff_peer_ahead() {
|
||||
let mut a = SyncState::new();
|
||||
let mut b = SyncState::new();
|
||||
|
||||
let author = [1u8; 32];
|
||||
let hash_a = [0xAA; 32];
|
||||
let hash_b = [0xBB; 32];
|
||||
|
||||
a.set(author, 5, hash_a);
|
||||
b.set(author, 10, hash_b);
|
||||
|
||||
let missing = a.diff(&b);
|
||||
assert_eq!(missing.len(), 1);
|
||||
assert_eq!(missing[0].author, author);
|
||||
assert_eq!(missing[0].from_seq, 5);
|
||||
assert_eq!(missing[0].from_hash, hash_a); // Resume after our hash
|
||||
assert_eq!(missing[0].to_seq, 10);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_diff_i_am_ahead() {
|
||||
let mut a = SyncState::new();
|
||||
let mut b = SyncState::new();
|
||||
|
||||
let author = [1u8; 32];
|
||||
a.set(author, 10, [0xAA; 32]);
|
||||
b.set(author, 5, [0xBB; 32]);
|
||||
|
||||
// I'm ahead, so I don't need anything from peer
|
||||
let missing = a.diff(&b);
|
||||
assert!(missing.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_diff_new_author() {
|
||||
let a = SyncState::new();
|
||||
let mut b = SyncState::new();
|
||||
|
||||
let author = [2u8; 32];
|
||||
b.set(author, 3, [0xBB; 32]);
|
||||
|
||||
// Peer has author I don't have
|
||||
let missing = a.diff(&b);
|
||||
assert_eq!(missing.len(), 1);
|
||||
assert_eq!(missing[0].from_seq, 0);
|
||||
assert_eq!(missing[0].from_hash, [0u8; 32]); // Zero hash = read from start
|
||||
assert_eq!(missing[0].to_seq, 3);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_merge() {
|
||||
let mut a = SyncState::new();
|
||||
let mut b = SyncState::new();
|
||||
|
||||
let author1 = [1u8; 32];
|
||||
let author2 = [2u8; 32];
|
||||
|
||||
a.set(author1, 10, [0xA1; 32]);
|
||||
a.set(author2, 5, [0xA2; 32]);
|
||||
|
||||
b.set(author1, 5, [0xB1; 32]); // a is ahead
|
||||
b.set(author2, 8, [0xB2; 32]); // b is ahead
|
||||
|
||||
a.merge(&b);
|
||||
assert_eq!(a.seq(&author1), 10); // kept a's value
|
||||
assert_eq!(a.seq(&author2), 8); // took b's value
|
||||
}
|
||||
}
|
||||
@@ -1,41 +0,0 @@
|
||||
//! Vector clocks for causality tracking
|
||||
|
||||
use std::collections::HashMap;
|
||||
|
||||
/// A vector clock for tracking "how much" of each node's log has been seen.
|
||||
///
|
||||
/// Used during reconciliation to identify missing entries between peers.
|
||||
pub struct VectorClock {
|
||||
clocks: HashMap<[u8; 32], u64>,
|
||||
}
|
||||
|
||||
impl VectorClock {
|
||||
/// Create a new empty vector clock.
|
||||
pub fn new() -> Self {
|
||||
Self {
|
||||
clocks: HashMap::new(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Get the clock value for a node (returns 0 if not present).
|
||||
pub fn get(&self, node_id: &[u8; 32]) -> u64 {
|
||||
self.clocks.get(node_id).copied().unwrap_or(0)
|
||||
}
|
||||
|
||||
/// Set the clock value for a node.
|
||||
pub fn set(&mut self, node_id: [u8; 32], value: u64) {
|
||||
self.clocks.insert(node_id, value);
|
||||
}
|
||||
|
||||
/// Increment the clock for a node.
|
||||
pub fn increment(&mut self, node_id: [u8; 32]) {
|
||||
let current = self.get(&node_id);
|
||||
self.set(node_id, current + 1);
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for VectorClock {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user