nils/lattice
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Implement DAG-based conflict resolution with binary keys and multi-head CLI display

Browse Source
This commit is contained in:
nils
2025-12-22 01:56:26 +01:00
parent 346ebccee7
commit 57c2906b10
10 changed files with 950 additions and 240 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/architecture.md
+4 -1
View File
@@ -303,4 +303,7 @@ Read permissions are not enforceable:
Practical model:
- Share store = grant read
- Write access defined in manifest
- Read-only nodes replicate and verify but can't contribute entries
- Read-only nodes replicate and verify but can't contribute entries
Future:
- Capability-based permissions: Explore finer-grained write access (e.g., per-key or per-prefix permissions) via capabilities. Exact mechanism TBD.
docs/roadmap.md
+13 -12
View File
@@ -33,26 +33,27 @@
---
## Milestone 1.5: DAG Conflict Resolution ← NEXT
## Milestone 1.5: DAG Conflict Resolution
**Goal:** Upgrade store from simple LWW to DAG-based conflict resolution per architecture.md.
### Deliverables
- [ ] Proto: Add `repeated bytes parent_hashes` to Entry (for DAG causality, separate from sigchain `prev_hash`)
- [ ] Store: keys → `Vec<u8>` (binary, not String)
- [ ] Store: KV table schema → `Vec<u8> → Vec<HeadInfo>` where `HeadInfo = { value, hlc, author, hash }`
- [ ] Store: `applied_frontiers` table → `author_id → (seq, hash)` per author
- [ ] Store: `apply_entry` → track multiple heads, merge parent tips into new tip
- [ ] Store: `get` → deterministic winner from heads (highest HLC, author_id tiebreaker)
- [ ] EntryBuilder: `.parent_hashes(...)` method for DAG ancestry
- [x] Proto: Add `repeated bytes parent_hashes` to Entry (for DAG causality)
- [x] Proto: Add `HeadInfo` message for multi-head storage
- [x] Store: KV table schema → `Vec<u8> → Vec<HeadInfo>`
- [x] Store: `apply_entry` → track multiple heads, merge parent tips
- [x] Store: `get` → deterministic winner (highest HLC, author tiebreaker)
- [x] Store: `get_heads` → inspect all heads for a key
- [x] EntryBuilder: `.parent_hashes(...)` method for DAG ancestry
- [x] CLI: Show conflict indicator when multiple heads
### Success Criteria
- Concurrent writes to same key create multiple heads
- Reads return deterministic winner
- Next write citing both heads merges fork to single tip
- All existing tests still pass
- [x] Concurrent writes to same key create multiple heads
- [x] Reads return deterministic winner
- [x] Next write citing both heads merges fork to single tip
- [x] All existing tests still pass (71 tests)
---
lattice-cli/src/commands.rs
+60 -12
View File
@@ -67,10 +67,10 @@ pub fn commands() -> Vec<Command> {
},
Command {
name: "get",
args: "<key>",
args: "<key> [-v]",
description: "Retrieve a value by key",
min_args: 1,
max_args: 1,
max_args: 2,
handler: cmd_get,
},
Command {
@@ -240,7 +240,7 @@ fn cmd_put(_node: &LatticeNode, store: Option<&StoreHandle>, args: &[String]) ->
return CommandResult::Ok;
};
let start = Instant::now();
match h.put(&args[0], args[1].as_bytes()) {
match h.put(args[0].as_bytes(), args[1].as_bytes()) {
Ok(seq) => println!("OK (seq: {}, {:.2?})", seq, start.elapsed()),
Err(e) => eprintln!("Error: {}", e),
}
@@ -252,14 +252,48 @@ fn cmd_get(_node: &LatticeNode, store: Option<&StoreHandle>, args: &[String]) ->
println!("No store selected. Use 'init' or 'use <uuid>'");
return CommandResult::Ok;
};
let verbose = args.get(1).map(|a| a == "-v").unwrap_or(false);
let start = Instant::now();
match h.get(&args[0]) {
Ok(Some(v)) => {
println!("{}", format_value(&v));
println!("({:.2?})", start.elapsed());
let key = args[0].as_bytes();
if verbose {
// Show all heads
match h.get_heads(key) {
Ok(heads) if heads.is_empty() => println!("(nil)"),
Ok(heads) => {
for (i, head) in heads.iter().enumerate() {
let winner = if i == 0 { "→" } else { " " };
let tombstone = if head.tombstone { "⊗" } else { "" };
let author_short = hex::encode(&head.author).chars().take(8).collect::<String>();
if head.tombstone {
println!("{} {} (deleted) (hlc:{}, author:{})",
winner, tombstone, head.hlc, author_short);
} else {
println!("{} {} (hlc:{}, author:{})",
winner, format_value(&head.value), head.hlc, author_short);
}
}
if heads.len() > 1 {
println!("⚠ {} heads (conflict)", heads.len());
}
println!("({:.2?})", start.elapsed());
}
Err(e) => eprintln!("Error: {}", e),
}
} else {
match h.get(key) {
Ok(Some(v)) => {
let heads = h.get_heads(key).unwrap_or_default();
if heads.len() > 1 {
println!("{} (⚠ {} heads)", format_value(&v), heads.len());
} else {
println!("{}", format_value(&v));
}
println!("({:.2?})", start.elapsed());
}
Ok(None) => println!("(nil)"),
Err(e) => eprintln!("Error: {}", e),
}
Ok(None) => println!("(nil)"),
Err(e) => eprintln!("Error: {}", e),
}
CommandResult::Ok
}
@@ -270,7 +304,7 @@ fn cmd_delete(_node: &LatticeNode, store: Option<&StoreHandle>, args: &[String])
return CommandResult::Ok;
};
let start = Instant::now();
match h.delete(&args[0]) {
match h.delete(args[0].as_bytes()) {
Ok(seq) => println!("OK (seq: {}, {:.2?})", seq, start.elapsed()),
Err(e) => eprintln!("Error: {}", e),
}
@@ -290,10 +324,24 @@ fn cmd_list(_node: &LatticeNode, store: Option<&StoreHandle>, args: &[String]) -
println!("(empty)");
} else {
for (k, v) in &entries {
let key_str = format_value(k);
if verbose {
println!("{} = {} ({} bytes)", k, format_value(v), v.len());
// Show all heads for this key
let heads = h.get_heads(k).unwrap_or_default();
println!("{}:", key_str);
for (i, head) in heads.iter().enumerate() {
let winner = if i == 0 { "→" } else { " " };
let author_short = hex::encode(&head.author).chars().take(8).collect::<String>();
if head.tombstone {
println!(" {} ⊗ (deleted) (hlc:{}, author:{})",
winner, head.hlc, author_short);
} else {
println!(" {} {} (hlc:{}, author:{})",
winner, format_value(&head.value), head.hlc, author_short);
}
}
} else {
println!("{} = {}", k, format_value(v));
println!("{} = {}", key_str, format_value(v));
}
}
println!("({} keys, {:.2?})", entries.len(), start.elapsed());
lattice-cli/src/node.rs
+30 -14
View File
@@ -189,11 +189,15 @@ pub struct StoreHandle {
impl StoreHandle {
pub fn id(&self) -> Uuid { self.store_id }
pub fn get(&self, key: &str) -> Result<Option<Vec<u8>>, NodeError> {
pub fn get(&self, key: &[u8]) -> Result<Option<Vec<u8>>, NodeError> {
Ok(self.store.get(key)?)
}
pub fn list(&self) -> Result<Vec<(String, Vec<u8>)>, NodeError> {
pub fn get_heads(&self, key: &[u8]) -> Result<Vec<lattice_core::HeadInfo>, NodeError> {
Ok(self.store.get_heads(key)?)
}
pub fn list(&self) -> Result<Vec<(Vec<u8>, Vec<u8>)>, NodeError> {
Ok(self.store.list_all()?)
}
@@ -202,18 +206,29 @@ impl StoreHandle {
}
pub fn applied_seq(&self) -> Result<u64, NodeError> {
Ok(self.store.last_seq()?)
let author = self.node.public_key_bytes();
Ok(self.store.author_state(&author)?
.map(|s| s.seq)
.unwrap_or(0))
}
pub fn put(&self, key: &str, value: &[u8]) -> Result<u64, NodeError> {
self.commit_entry(|b| b.put(key, value.to_vec()))
pub fn put(&self, key: &[u8], value: &[u8]) -> Result<u64, NodeError> {
// Get current heads for this key to cite as parents
let heads = self.store.get_heads(key)?;
let parent_hashes: Vec<Vec<u8>> = heads.iter().map(|h| h.hash.clone()).collect();
self.commit_entry(parent_hashes, |b| b.put(key.to_vec(), value.to_vec()))
}
pub fn delete(&self, key: &str) -> Result<u64, NodeError> {
self.commit_entry(|b| b.delete(key))
pub fn delete(&self, key: &[u8]) -> Result<u64, NodeError> {
// Get current heads for this key to cite as parents
let heads = self.store.get_heads(key)?;
let parent_hashes: Vec<Vec<u8>> = heads.iter().map(|h| h.hash.clone()).collect();
self.commit_entry(parent_hashes, |b| b.delete(key.to_vec()))
}
fn commit_entry<F>(&self, build: F) -> Result<u64, NodeError>
fn commit_entry<F>(&self, parent_hashes: Vec<Vec<u8>>, build: F) -> Result<u64, NodeError>
where
F: FnOnce(EntryBuilder) -> EntryBuilder,
{
@@ -223,7 +238,8 @@ impl StoreHandle {
let builder = EntryBuilder::new(seq, HLC::now())
.store_id(self.store_id.as_bytes().to_vec())
.prev_hash(prev_hash.to_vec());
.prev_hash(prev_hash.to_vec())
.parent_hashes(parent_hashes);
let entry = build(builder).sign(&self.node);
sigchain.append(&entry)?;
@@ -261,8 +277,8 @@ mod tests {
assert!(stores.contains(&store_id));
let (handle, _) = node.open_store(store_id).expect("Failed to open store");
handle.put("/key", b"value").expect("put failed");
assert_eq!(handle.get("/key").unwrap(), Some(b"value".to_vec()));
handle.put(b"/key", b"value").expect("put failed");
assert_eq!(handle.get(b"/key").unwrap(), Some(b"value".to_vec()));
let _ = std::fs::remove_dir_all(data_dir.base());
}
@@ -279,12 +295,12 @@ mod tests {
let store_b = node.create_store().expect("create B");
let (handle_a, _) = node.open_store(store_a).expect("open A");
handle_a.put("/key", b"from A").expect("put A");
handle_a.put(b"/key", b"from A").expect("put A");
let (handle_b, _) = node.open_store(store_b).expect("open B");
assert_eq!(handle_b.get("/key").unwrap(), None);
assert_eq!(handle_b.get(b"/key").unwrap(), None);
assert_eq!(handle_a.get("/key").unwrap(), Some(b"from A".to_vec()));
assert_eq!(handle_a.get(b"/key").unwrap(), Some(b"from A".to_vec()));
let _ = std::fs::remove_dir_all(data_dir.base());
}
lattice-core/src/lib.rs
+1
View File
@@ -41,4 +41,5 @@ pub use signed_entry::{EntryBuilder, sign_entry, verify_signed_entry, hash_signe
pub use log::{append_entry, read_entries, LogReader};
pub use store::Store;
pub use meta_store::MetaStore;
pub use proto::HeadInfo;
pub use uuid::Uuid;
lattice-core/src/proto.rs
+2 -1
View File
@@ -33,6 +33,7 @@ mod tests {
version: 1,
store_id: vec![1u8; 16],
prev_hash: vec![0u8; 32],
parent_hashes: vec![],
seq: 5,
timestamp: Some(Hlc {
wall_time: 1000,
@@ -41,7 +42,7 @@ mod tests {
ops: vec![
Operation {
op_type: Some(operation::OpType::Put(PutOp {
key: "/nodes/abc".to_string(),
key: b"/nodes/abc".to_vec(),
value: b"hello".to_vec(),
})),
},
lattice-core/src/sigchain.rs
+1 -1
View File
@@ -451,7 +451,7 @@ mod tests {
let ops = vec![
Operation {
op_type: Some(operation::OpType::Put(PutOp {
key: "/test".to_string(),
key: b"/test".to_vec(),
value: b"hello".to_vec(),
})),
},
lattice-core/src/signed_entry.rs
+14 -5
View File
@@ -34,6 +34,7 @@ pub struct EntryBuilder {
version: u32,
store_id: Vec<u8>,
prev_hash: Vec<u8>,
parent_hashes: Vec<Vec<u8>>,
seq: u64,
timestamp: HLC,
ops: Vec<Operation>,
@@ -44,8 +45,9 @@ impl EntryBuilder {
pub fn new(seq: u64, timestamp: HLC) -> Self {
Self {
version: 1,
store_id: Vec::new(), // Empty = legacy single-store
prev_hash: vec![0u8; 32], // Genesis or will be set
store_id: Vec::new(),
prev_hash: vec![0u8; 32],
parent_hashes: Vec::new(),
seq,
timestamp,
ops: Vec::new(),
@@ -58,14 +60,20 @@ impl EntryBuilder {
self
}
/// Set the previous entry hash (for chaining)
/// Set the previous entry hash (for sigchain linking)
pub fn prev_hash(mut self, hash: impl Into<Vec<u8>>) -> Self {
self.prev_hash = hash.into();
self
}
/// Set the parent hashes (for DAG ancestry)
pub fn parent_hashes(mut self, hashes: Vec<Vec<u8>>) -> Self {
self.parent_hashes = hashes;
self
}
/// Add a Put operation
pub fn put(mut self, key: impl Into<String>, value: impl Into<Vec<u8>>) -> Self {
pub fn put(mut self, key: impl Into<Vec<u8>>, value: impl Into<Vec<u8>>) -> Self {
self.ops.push(Operation {
op_type: Some(operation::OpType::Put(PutOp {
key: key.into(),
@@ -76,7 +84,7 @@ impl EntryBuilder {
}
/// Add a Delete operation
pub fn delete(mut self, key: impl Into<String>) -> Self {
pub fn delete(mut self, key: impl Into<Vec<u8>>) -> Self {
self.ops.push(Operation {
op_type: Some(operation::OpType::Delete(DeleteOp {
key: key.into(),
@@ -97,6 +105,7 @@ impl EntryBuilder {
version: self.version,
store_id: self.store_id,
prev_hash: self.prev_hash,
parent_hashes: self.parent_hashes,
seq: self.seq,
timestamp: Some(Hlc {
wall_time: self.timestamp.wall_time,
lattice-core/src/store.rs
+797 -190
View File
File diff suppressed because it is too large Load Diff
proto/lattice.proto
+28 -4
View File
@@ -24,14 +24,38 @@ message Entry {
bytes store_id = 6;
// Ordering Metadata
bytes prev_hash = 2; // Link to previous entry (32 bytes)
bytes prev_hash = 2; // Link to previous sigchain entry (32 bytes)
uint64 seq = 3; // Monotonic sequence number
HLC timestamp = 4; // Hybrid Logical Clock
// DAG ancestry: hashes of entries this supersedes (separate from sigchain)
repeated bytes parent_hashes = 7;
// The Batch of Operations
repeated Operation ops = 5;
}
// HeadInfo: a tip/head in the DAG for a key
message HeadInfo {
bytes value = 1; // The value at this head
uint64 hlc = 2; // Combined HLC for ordering (wall_time_ms << 16 | counter)
bytes author = 3; // Author's public key (32 bytes)
bytes hash = 4; // Hash of the SignedEntry that created this head
bool tombstone = 5; // True if this head represents a delete
}
// HeadList: wrapper for storing multiple heads per key in state.db
message HeadList {
repeated HeadInfo heads = 1;
}
// AuthorState: tracks last applied entry per author for replay optimization
message AuthorState {
uint64 seq = 1; // Last applied seq for this author's sigchain
bytes hash = 2; // Hash of last applied entry
uint64 log_offset = 3; // Byte offset in log file for fast resume
}
// Hybrid Logical Clock
message HLC {
uint64 wall_time = 1; // Unix timestamp (ms)
@@ -49,12 +73,12 @@ message Operation {
}
message PutOp {
string key = 1;
bytes value = 2; // Raw bytes allows storing images, JSON, binary, etc.
bytes key = 1;
bytes value = 2;
}
message DeleteOp {
string key = 1;
bytes key = 1;
}
// 4. The Sync Handshake (Vector Clocks)