feat: Implement SigChain for append-only signed log management

architecture: future ideas for rich crtds
feat: Introduce append-only log file I/O for SignedEntry messages using a new LogRecord protobuf, including read/write utilities and documentation.
2025-12-20 14:29:08 +01:00 · 2025-12-20 14:29:08 +01:00 · 2025-12-20 14:29:08 +01:00 · 2025-12-20 14:29:08 +01:00 · 2025-12-20 14:29:08 +01:00 · 2025-12-20 14:29:03 +01:00
31 changed files with 2584 additions and 4901 deletions
@@ -0,0 +1,17 @@
 # Rust / Cargo
 /target/
 Cargo.lock
 # IDE
 .idea/
 .vscode/
 *.swp
 *.swo
 # OS
 .DS_Store
 Thumbs.db
 # Environment
 .env
 .env.local
@@ -0,0 +1,51 @@
 [workspace]
 resolver = "2"
 members = [
    "lattice-core",
    "lattice-net",
    "lattice-store",
 ]
 [workspace.package]
 version = "0.1.0"
 edition = "2021"
 license = "MIT"
 [workspace.dependencies]
 # Workspace crates
 lattice-core = { path = "lattice-core" }
 lattice-net = { path = "lattice-net" }
 lattice-store = { path = "lattice-store" }
 # Networking (Iroh)
 iroh = "0.95"
 iroh-gossip = "0.95"
 # Cryptography
 ed25519-dalek = { version = "2", features = ["rand_core"] }
 rand = "0.8"
 # Serialization
 prost = "0.13"
 prost-types = "0.13"
 prost-build = "0.13"
 # Async runtime
 tokio = { version = "1", features = ["full"] }
 # Utilities
 thiserror = "2"
 tracing = "0.1"
 bytes = "1"
 dirs = "5"
 blake3 = "1"
 hex = "0.4"
 # Testing
 tokio-test = "0.4"
 [workspace.lints.rust]
 unsafe_code = "forbid"
 [workspace.lints.clippy]
 all = "warn"
@@ -1,3 +0,0 @@
 target/
 blobs/
 identity.key
@@ -1,21 +0,0 @@
 [package]
 name = "lattice-proto"
 version = "0.1.0"
 edition = "2021"
 [dependencies]
 anyhow = "1.0"
 blake3 = "1.5"
 bytes = "1.5"
 clap = { version = "4.5", features = ["derive"] }
 ed25519-dalek = { version = "2.1", features = ["rand_core"] }
 futures = "0.3"
 hex = "0.4"
 iroh = { version = "0.95.1", features = ["discovery-local-network"] }
 iroh-gossip = "0.95.0"
 iroh-tickets = "0.2.0"
 rand = "0.8"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 tokio = { version = "1.36", features = ["full"] }
 tracing-subscriber = { version = "0.3.22", features = ["env-filter"] }
@@ -1,434 +0,0 @@
 use serde::{Serialize, Deserialize};
 use std::collections::HashMap;
 use std::sync::{Arc, Mutex};
 use tokio::sync::{mpsc, Notify};
 use ed25519_dalek::{Signer, Verifier, Signature};
 use iroh::{Endpoint, SecretKey, PublicKey}; 
 use iroh::discovery::mdns::MdnsDiscovery;
 use iroh_gossip::net::Gossip;
 use iroh_tickets::endpoint::EndpointTicket; 
 use iroh_gossip::proto::TopicId;
 use iroh::protocol::Router; 
 use futures::StreamExt;
 use std::io::Write;
 // --- Data Structures ---
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct Entry {
    pub path: String,
    pub content_hash: String,
    pub author: String,
    pub timestamp: u64,
    pub signature: Vec<u8>,
 }
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub enum Message {
    Update(Entry),
    WantBlob(String),
    BlobData(String, Vec<u8>), 
 }
 // --- Logic ---
 type Db = Arc<Mutex<HashMap<String, Entry>>>;
 pub struct Node {
    pub keypair: ed25519_dalek::SigningKey,
    pub db: Db,
    pub blobs_dir: std::path::PathBuf,
    pub gossip_sender: Option<mpsc::Sender<Message>>,
    pub neighbors: Arc<Mutex<std::collections::HashSet<PublicKey>>>,
    pub pending_requests: Arc<Mutex<HashMap<String, Arc<Notify>>>>,
 }
 impl Node {
    pub fn new(blobs_dir: std::path::PathBuf) -> Self {
        let key_path = std::path::Path::new("identity.key");
        let keypair = if key_path.exists() {
             let bytes = std::fs::read(key_path).expect("Failed to read key");
             ed25519_dalek::SigningKey::from_bytes(bytes.as_slice().try_into().unwrap())
        } else {
            let pk = ed25519_dalek::SigningKey::generate(&mut rand::rngs::OsRng);
            std::fs::write(key_path, pk.to_bytes()).expect("Failed to write key");
            pk
        };
        if !blobs_dir.exists() {
            std::fs::create_dir_all(&blobs_dir).expect("Failed to create blobs dir");
        }
        Self {
            keypair,
            db: Arc::new(Mutex::new(HashMap::new())),
            blobs_dir,
            gossip_sender: None,
            neighbors: Arc::new(Mutex::new(std::collections::HashSet::new())),
            pending_requests: Arc::new(Mutex::new(HashMap::new())),
        }
    }
    pub fn load_peers(&self) -> Vec<PublicKey> {
        let peers_file = self.blobs_dir.parent().unwrap().join("peers.txt");
        if !peers_file.exists() {
            return Vec::new();
        }
        let content = std::fs::read_to_string(&peers_file).unwrap_or_default();
        content.lines()
            .filter_map(|line| std::str::FromStr::from_str(line).ok())
            .collect()
    }
    pub fn save_peer(&self, peer: PublicKey) {
         let peers_file = self.blobs_dir.parent().unwrap().join("peers.txt");
         // Avoid duplicates in file trivially by checking if already known in memory?
         // But restart clears memory.
         // Just append. Ideally we read all and check, but performance.
         // Let's just append.
         let mut file = std::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(peers_file)
            .expect("Failed to open peers file");
         writeln!(file, "{}", peer).expect("Failed to write peer");
    }
    pub fn pubkey_hex(&self) -> String {
        hex::encode(self.keypair.verifying_key().to_bytes())
    }
    pub fn sign_entry(&self, path: &str, hash: &str, timestamp: u64) -> Vec<u8> {
        let mut msg = Vec::new();
        msg.extend_from_slice(path.as_bytes());
        msg.extend_from_slice(hash.as_bytes());
        msg.extend_from_slice(&timestamp.to_le_bytes()); 
        self.keypair.sign(&msg).to_bytes().to_vec()
    }
    pub fn verify_entry(entry: &Entry) -> bool {
        let pubkey_bytes = match hex::decode(&entry.author) {
            Ok(b) => b,
            Err(_) => return false,
        };
        let pubkey_arr: [u8; 32] = match pubkey_bytes.as_slice().try_into() {
            Ok(a) => a, 
            Err(_) => return false,
        };
        let pubkey = match ed25519_dalek::VerifyingKey::from_bytes(&pubkey_arr) {
            Ok(pk) => pk,
            Err(_) => return false,
        };
        let mut msg = Vec::new();
        msg.extend_from_slice(entry.path.as_bytes());
        msg.extend_from_slice(entry.content_hash.as_bytes());
        msg.extend_from_slice(&entry.timestamp.to_le_bytes());
        let sig_arr: [u8; 64] = match entry.signature.as_slice().try_into() {
            Ok(a) => a,
            Err(_) => return false,
        };
        let sig = Signature::from_bytes(&sig_arr);
        pubkey.verify(&msg, &sig).is_ok()
    }
    pub fn put_local(&self, path: String, content: String) -> Entry {
        let hash = blake3::hash(content.as_bytes()).to_hex().to_string();
        let blob_path = self.blobs_dir.join(&hash);
        std::fs::write(blob_path, content).expect("Failed to write blob");
        let timestamp = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_micros() as u64;
        let signature = self.sign_entry(&path, &hash, timestamp);
        let entry = Entry {
            path: path.clone(),
            content_hash: hash.clone(),
            author: self.pubkey_hex(),
            timestamp,
            signature,
        };
        println!("[LOCAL] Writing {} -> {}", path, hash);
        let mut db = self.db.lock().unwrap();
        db.insert(path, entry.clone());
        entry
    }
    pub fn process_update(&self, entry: Entry) -> bool {
        if !Self::verify_entry(&entry) {
            println!("[WARN] Invalid Signature for {}", entry.path);
            return false;
        }
        let mut db = self.db.lock().unwrap();
        if let Some(existing) = db.get(&entry.path) {
            if entry.timestamp <= existing.timestamp {
                return false;
            }
        }
        println!("[SYNC] Received Update: {} -> {}", entry.path, entry.content_hash);
        db.insert(entry.path.clone(), entry);
        true
    }
 }
 // --- Main ---
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
    tracing_subscriber::fmt::init(); 
    let blobs_dir = std::path::PathBuf::from("blobs");
    let node = Arc::new(Node::new(blobs_dir));
    println!("Lattice Node Started. Identity: {}", node.pubkey_hex());
    // Convert Node key to Iroh SecretKey
    let secret_key = SecretKey::from_bytes(&node.keypair.to_bytes());
    let node_id = secret_key.public();
    // Configure MDNS
    // Note: If build() returns Result, use ?
    let mdns = MdnsDiscovery::builder()
         .build(node_id)?; 
    // Bind endpoint
    let endpoint = Endpoint::builder()
        .secret_key(secret_key)
        .discovery(mdns)
        .bind()
        .await?;
    println!("Iroh Node ID: {}", endpoint.secret_key().public());
    // Spawn Gossip
    let gossip = Gossip::builder().spawn(endpoint.clone()); 
    // Spawn Router (The Accept Loop handling Gossip ALPN)
    let router = Router::builder(endpoint.clone())
        .accept(iroh_gossip::ALPN, gossip.clone()) 
        .spawn(); // CORRECTED: No await.
    // Create and print Ticket
    let my_addr = endpoint.addr(); // CORRECTED: No ?
    let ticket = EndpointTicket::new(my_addr); 
    println!("My Ticket: {}", ticket);
    let topic_bytes = blake3::hash(b"lattice-test-net").as_bytes().to_owned();
    let topic_id = TopicId::from_bytes(topic_bytes);
    // Join 
    println!("Joining Gossip Topic: lattice-test-net...");
    // Load persisted peers
    let initial_peers = node.load_peers();
    if !initial_peers.is_empty() {
        println!("Loaded {} persisted peers.", initial_peers.len());
        // Populate neighbors set specifically for display if desired, 
        // though NeighborUp will fire when connection is actually established.
    }
    let (sink, mut stream) = gossip.subscribe(topic_id, initial_peers).await?.split();
    println!("Joined Gossip Topic: lattice-test-net");
    // Command Loop
    let (tx_cmd, mut rx_cmd) = mpsc::channel::<String>(100);
    // Stdin Task
    tokio::spawn(async move {
        let stdin = std::io::stdin();
        loop {
            let mut line = String::new();
            if stdin.read_line(&mut line).is_ok() {
                if tx_cmd.send(line).await.is_err() { break; }
            }
        }
    });
    print!("> ");
    std::io::stdout().flush()?;
    // Event Loop
    loop {
        tokio::select! {
             Some(line) = rx_cmd.recv() => {
                 let node = node.clone();
                 let sink = sink.clone();
                 let endpoint = endpoint.clone();
                 let gossip = gossip.clone();
                 tokio::spawn(async move {
                     let parts: Vec<&str> = line.trim().split_whitespace().collect();
                     match parts.as_slice() {
                         ["put", path, content] => {
                             let entry = node.put_local(path.to_string(), content.to_string());
                             let msg = Message::Update(entry);
                             if let Ok(msg_bytes) = serde_json::to_vec(&msg) {
                                 sink.broadcast(msg_bytes.into()).await.ok();
                                 println!("[GOSSIP] Broadcasted update.");
                             }
                         },
                         ["get", path] => {
                            let entry_opt = {
                                let db = node.db.lock().unwrap();
                                db.get(*path).cloned()
                            };
                            if let Some(entry) = entry_opt {
                                 let blob_path = node.blobs_dir.join(&entry.content_hash);
                                 if blob_path.exists() {
                                     if let Ok(content) = std::fs::read_to_string(blob_path) {
                                         println!("Content: {}", content);
                                     }
                                 } else {
                                     println!("[MISSING BLOB] Need hash: {}", entry.content_hash);
                                     let notify = Arc::new(Notify::new());
                                     node.pending_requests.lock().unwrap().insert(entry.content_hash.clone(), notify.clone());
                                     let msg = Message::WantBlob(entry.content_hash.clone());
                                     if let Ok(msg_bytes) = serde_json::to_vec(&msg) {
                                         sink.broadcast(msg_bytes.into()).await.ok();
                                         println!("Requesting blob... waiting...");
                                         // Wait for notification with timeout
                                         match tokio::time::timeout(std::time::Duration::from_secs(5), notify.notified()).await {
                                             Ok(_) => {
                                                 // check again
                                                 let blob_path = node.blobs_dir.join(&entry.content_hash);
                                                 if let Ok(content) = std::fs::read_to_string(blob_path) {
                                                     println!("Content: {}", content);
                                                 } else {
                                                     println!("Error reading received blob.");
                                                 }
                                             },
                                             Err(_) => {
                                                 println!("Timeout waiting for blob.");
                                                 // Clean up
                                                 node.pending_requests.lock().unwrap().remove(&entry.content_hash);
                                             }
                                         }
                                     }
                                 }
                            } else {
                                println!("Path not found.");
                            }
                        },
                         ["list"] => {
                             let db = node.db.lock().unwrap();
                             for (k, v) in db.iter() {
                                 println!("{} -> {} (ts: {})", k, v.content_hash, v.timestamp);
                             }
                         },
                         ["connect", arg] => {
                             use std::str::FromStr;
                             let topic_bytes = blake3::hash(b"lattice-test-net").as_bytes().to_owned();
                             let topic_id = TopicId::from_bytes(topic_bytes);
                             if let Ok(ticket) = EndpointTicket::from_str(arg) {
                                 println!("Connecting via Ticket...");
                                 let addr = iroh::EndpointAddr::from(ticket);
                                 let peer_id = addr.id;
                                 // Connect first (ensure transport)
                                 match endpoint.connect(addr, iroh_gossip::ALPN).await {
                                     Ok(_) => {
                                         println!("Connected! Adding to Gossip...");
                                         match gossip.subscribe(topic_id, vec![peer_id]).await {
                                             Ok(_) => println!("Subscribed peer to gossip topic!"),
                                             Err(e) => println!("Gossip subscribe error: {}", e),
                                         }
                                     },
                                     Err(e) => println!("Connection failed: {}", e),
                                 }
                             } else if let Ok(peer_id) = PublicKey::from_str(arg) {
                                 println!("Connecting via Node ID (Discovery) to: {}", peer_id);
                                 match endpoint.connect(peer_id, iroh_gossip::ALPN).await {
                                     Ok(_) => {
                                         println!("Connected! Adding to Gossip...");
                                         match gossip.subscribe(topic_id, vec![peer_id]).await {
                                             Ok(_) => println!("Subscribed peer to gossip topic!"),
                                             Err(e) => println!("Gossip subscribe error: {}", e),
                                         }
                                     },
                                     Err(e) => println!("Connection failed: {}", e),
                                 }
                             } else {
                                 println!("Invalid ID or Ticket");
                             }
                        },
                         ["peers"] => {
                            let neighbors = node.neighbors.lock().unwrap();
                            println!("Connected Gossip Peers: {}", neighbors.len());
                            for peer in neighbors.iter() {
                                println!("- {}", peer);
                            }
                        },
                        ["quit"] => std::process::exit(0), // Can't break loop from spawn
                        _ => println!("Unknown command. Usage: put <path> <content> | get <path> | list | connect <ticket_or_node_id> | peers"),
                    }
                    print!("> ");
                    std::io::stdout().flush().ok();
                 });
             },
             Some(res) = stream.next() => {
                 let event = res?;
                 match event {
                    iroh_gossip::api::Event::Received(msg) => {
                         if let Ok(message) = serde_json::from_slice::<Message>(&msg.content) {
                             match message {
                                 Message::Update(entry) => { node.process_update(entry); },
                                 Message::WantBlob(hash) => {
                                     let blob_path = node.blobs_dir.join(&hash);
                                     if blob_path.exists() {
                                         if let Ok(content) = std::fs::read_to_string(&blob_path) {
                                             let reply = Message::BlobData(hash, content.into_bytes());
                                             if let Ok(reply_bytes) = serde_json::to_vec(&reply) {
                                                sink.broadcast(reply_bytes.into()).await.ok(); 
                                             }
                                         }
                                     }
                                 },
                                  Message::BlobData(hash, data) => {
                                     let blob_path = node.blobs_dir.join(&hash);
                                      if !blob_path.exists() {
                                         std::fs::write(blob_path, data).expect("Failed to write blob");
                                         println!("[SYNC] Received Blob: {}", hash);
                                         // Notify waiters
                                         if let Some(notify) = node.pending_requests.lock().unwrap().remove(&hash) {
                                             notify.notify_waiters();
                                         }
                                     } 
                                  }
                             }
                         }
                    },
                    iroh_gossip::api::Event::NeighborUp(node_id) => {
                        println!("[GOSSIP] Neighbor Up: {}", node_id);
                        if node.neighbors.lock().unwrap().insert(node_id) {
                            node.save_peer(node_id);
                        }
                    },
                    iroh_gossip::api::Event::NeighborDown(node_id) => {
                        println!("[GOSSIP] Neighbor Down: {}", node_id);
                        node.neighbors.lock().unwrap().remove(&node_id);
                    },
                    _ => {} // Handle Lagged or other future variants
                 }
             }
        }
    }
    router.shutdown().await?;
    Ok(())
 }
@@ -0,0 +1,159 @@
 # Architecture
 ## Ideas
 - SigChains Ed25519-signed, hash-chained append-only logs per node. Trust via local signature verification.
 - Log-Based State: KV store derived by replaying entries. Watermarks enable safe log pruning + snapshots.
 - Offline-First: Iroh for networking. Vector clocks identify missing entries on reconnect—converges mathematically.
 - Full Replication: All nodes keep all logs until watermark consensus, then prune and snapshot.
 ## Concepts
 - Transitive Pairing. Nodes can introduce new nodes to the mesh.
 ## Stack
 - rust
 - iroh
 - prost protocol buffers
 ### Bootstrap
 - New peers request a full state snapshot from their first connection.
 - The snapshot allows them to skip replaying the entire log history.
 - After bootstrap, the node receives incremental updates via gossip.
 ### Networking
 - Designed for mobile clients that may only sync a few times per day.
 - When peers connect, they exchange vector clocks to identify missing entries.
 - Missing entries are fetched via unicast.
 - MAX_DRIFT should be generous (e.g., hours) to accommodate sleeping devices.
 Networking modes:
 - Active (servers/laptops on power): Frequent gossip broadcasts, proactive sync.
 - Low-power (mobile/battery): Pull-based sync on wake. Query peers instead of relying on push gossip.
 ## Parts
 ### Nodes
 - Identified by their Ed25519 public key.
 - Private key stored locally in `identity.key` (not replicated).
 - Node data stored in KV:
  - `/nodes/{pubkey}/info` = static metadata (name, added_by, added_at)
  - `/nodes/{pubkey}/status` = `active` | `dormant` | `disabled`
  - `/nodes/{pubkey}/role` = `server` | `device` (optional, hints sync priority)
 - Inviting a node = writing entries to `/nodes/{pubkey}/...`.
 - Liveness: Each node tracks `last_seen` locally (from watermark gossip). UI alerts if a peer hasn't been seen for threshold (e.g., 30 days). User decides to mark dormant/disabled.
 - Status effects:
  - `active`: Normal sync participant, blocks watermark until acknowledged.
  - `dormant`: Excluded from watermark consensus, can be reactivated.
  - `disabled`: Permanently removed from mesh.
 - Sync priority: Low-power clients prefer peers marked as `server` or recently active.
 Future:
 - Key rotation: Allow nodes to rotate their keypair. Old key signs a "rotation" entry pointing to new key.
 - Secure storage: Support platform keystores (macOS Keychain, Linux Secret Service, TPM) for private key protection.
 ### Data Model
 - Keys are flat strings using path conventions (e.g., `/nodes/{pubkey}`, `/config/sync/interval`).
 - Prefix queries via string matching (sorted map enables efficient range scans).
 - State is computed by replaying `Put`/`Delete` operations from all authors.
 - Entry ordering: by HLC timestamp, then by author ID as tiebreaker.
 - Conflicts resolved by last-write-wins (using the ordering above).
 ### Timestamps (Hybrid Logical Clocks)
 Timestamps use HLC `<wall_time, counter>` with Causal Clamping:
 - Each entry includes an HLC and a reference to its parent (prev_hash).
 - Standard HLC: `new_hlc = max(local_wall_clock, max_seen_hlc + 1)`.
 - On receive: if `entry.hlc > local_wall_clock + MAX_DRIFT`, clamp to `parent.hlc + 1`.
 - All nodes compute the same clamped time from the parent (deterministic).
 - Genesis entries (no parent) with future timestamps are dropped.
 Pre-flight check (before signing):
 - Compare local_clock to max_peer_hlc (from recent gossip/entries).
 - If `local_clock > max_peer_hlc + MAX_DRIFT`, use `max_peer_hlc + 1` instead.
 - This catches future-clock nodes before they poison the log.
 Authors apply their own entries through the standard receive path to ensure consistent clamping.
 ### Storage
 Each node stores logs as one file per author:
 ```
 data/
 ├── identity.key               # Local node's Ed25519 private key
 ├── logs/
 │   └── {author_id_hex}.log    # Append-only SignedEntry stream per author
 └── state.db                   # redb: KV snapshot + vector clocks + indexes
 ```
 - Logs: Append-only binary files per author, containing serialized `SignedEntry` messages.
 - State DB (redb): Combined KV state, vector clocks, and indexes. Updated as entries are applied.
 #### state.db Tables (redb)
 ```
 Table           Key                     Value                      Purpose
 ─────────────────────────────────────────────────────────────────────────────
 kv              String (path)           Vec<u8>                    Replicated key-value data
 vector_clocks   [u8; 32] (author_id)    (u64 seq, [u8; 32] hash)   Track sync state + chain verification
 entry_index     (author_id, seq)        u64 (offset)               Fast entry lookup by position
 meta            String                  Vec<u8>                    System metadata (own_seq, watermark, etc.)
 ```
 ### Watermarks
 - Nodes gossip their watermarks periodically (throttled).
 - A watermark is a vector clock: how much of each author's log the node has seen.
 - All nodes keep all logs (own + others) for redundancy until watermark consensus.
 - Once all peers have acknowledged entries, they can be pruned and replaced by the snapshot.
 - If a node is offline too long, it re-bootstraps with a fresh snapshot when it reconnects.
 ### Rich CRDTs (Future)
 Instead of a generic scripting language, use specific data types that merge better than LWW.
 Extend value types in redb:
 ```rust
 enum ReplicatedValue {
    LWW(Vec<u8>),          // Standard Last-Write-Wins (current model)
    Counter(i64),          // PN-Counter (Increment/Decrement)
    Set(HashSet<Vec<u8>>), // OR-Set (Observed-Remove Set)
 }
 ```
 **Counter** (for "storage used" etc.):
 - State is `{node_id: value}` map. Merge = sum all nodes. No conflicts possible.
 **OR-Set** (for group membership etc.):
 - Merge = union. Element present if add timestamp > remove timestamp.
 **Op Code Compromise**: Use commutative operations instead of a VM:
 ```protobuf
 message Entry {
  oneof operation {
    PutOp put = 1;
    DeleteOp delete = 2;
    MergeOp merge = 3;
  }
 }
 message MergeOp {
  string key = 1;
  oneof payload {
    int64 counter_delta = 2;
    bytes set_add_member = 3;
    bytes set_remove_member = 4;
  }
 }
 ```
 Recommendation: Use Put/Delete for 90% of data. Add CRDT primitives only when needed (concurrent counters, lists) rather than a scripting language.
@@ -0,0 +1,37 @@
 # Test Cases
 ## Timestamp / HLC
 ### Time-traveling node applies own entry
 - Node X has clock at year 2050
 - Node X creates entry, signs, broadcasts
 - All nodes (including X) should clamp to parent.hlc + 1
 - Verify: X's state.db matches other nodes' state.db
 - Failure mode: X applies using 2050, others use 101 → divergence
 ### Clamping with no parent (genesis entry)
 - Node X creates first-ever entry with future timestamp
 - All nodes should DROP the entry (no parent to anchor to)
 - Verify: entry is not applied anywhere
 ### Out-of-order entry arrival
 - Entry B (hlc=91) arrives after Entry A (hlc=100)
 - Both write to same key
 - Verify: A's value wins (LWW with timestamp tracking)
 - Verify: no rollback needed, just comparison on apply
 ### Clock drift detection
 - Node consistently sees its entries clamped
 - Verify: UI alerts user about clock being ahead
 ### Clock in past (Pi without RTC, boots at 1970)
 - Node X has clock at 1970
 - Node X receives entries from peers with HLC around 2024
 - Standard HLC: X uses max(1970, peer_hlc + 1) = peer_hlc + 1
 - Verify: X's entries slot in correctly (no special handling needed)
 ### Pre-flight peer sanity check (future clock)
 - Node X has clock at 2050
 - Before creating entry, X compares local_clock to max_peer_hlc
 - If local_clock > max_peer_hlc + MAX_DRIFT, use max_peer_hlc + 1
 - Verify: X's entry uses sane timestamp, all nodes agree
@@ -0,0 +1,22 @@
 [package]
 name = "lattice-core"
 description = "Core types for Lattice: nodes, sigchains, entries, and vector clocks"
 version.workspace = true
 edition.workspace = true
 license.workspace = true
 [dependencies]
 ed25519-dalek = { workspace = true }
 rand = { workspace = true }
 prost = { workspace = true }
 thiserror = { workspace = true }
 bytes = { workspace = true }
 dirs = { workspace = true }
 blake3 = { workspace = true }
 hex = { workspace = true }
 [build-dependencies]
 prost-build = { workspace = true }
 [dev-dependencies]
 tokio-test = { workspace = true }
@@ -0,0 +1,9 @@
 use std::io::Result;
 fn main() -> Result<()> {
    prost_build::compile_protos(
        &["../proto/lattice.proto"],
        &["../proto/"],
    )?;
    Ok(())
 }
@@ -0,0 +1,62 @@
 //! Clock abstraction for testable time
 //!
 //! Provides a trait for getting the current time, with implementations
 //! for real system time and mock time for testing.
 use std::time::{SystemTime, UNIX_EPOCH};
 /// Trait for getting the current wall clock time in milliseconds
 pub trait Clock: Send + Sync {
    /// Get the current time in milliseconds since Unix epoch
    fn now_ms(&self) -> u64;
 }
 /// Real system clock implementation
 #[derive(Debug, Clone, Copy, Default)]
 pub struct SystemClock;
 impl Clock for SystemClock {
    fn now_ms(&self) -> u64 {
        SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .expect("Time went backwards")
            .as_millis() as u64
    }
 }
 /// Mock clock for testing - returns a fixed time
 #[derive(Debug, Clone, Copy)]
 pub struct MockClock {
    pub time_ms: u64,
 }
 impl MockClock {
    pub fn new(time_ms: u64) -> Self {
        Self { time_ms }
    }
 }
 impl Clock for MockClock {
    fn now_ms(&self) -> u64 {
        self.time_ms
    }
 }
 #[cfg(test)]
 mod tests {
    use super::*;
    #[test]
    fn test_system_clock_returns_reasonable_time() {
        let clock = SystemClock;
        let now = clock.now_ms();
        // Should be after 2025-01-01
        assert!(now > 1_735_689_600_000);
    }
    #[test]
    fn test_mock_clock_returns_fixed_time() {
        let clock = MockClock::new(12345);
        assert_eq!(clock.now_ms(), 12345);
    }
 }
@@ -0,0 +1,111 @@
 //! Data directory management
 //!
 //! Provides platform-specific paths for Lattice data storage:
 //! - `identity.key` — Ed25519 private key
 //! - `logs/` — Append-only log files per author
 //! - `state.db` — KV snapshot and indexes
 use std::path::{Path, PathBuf};
 const APP_NAME: &str = "lattice";
 /// Data directory configuration.
 ///
 /// Handles paths for:
 /// - `identity.key` — node's private key
 /// - `logs/{author_id}.log` — per-author log files
 /// - `state.db` — redb database
 #[derive(Debug, Clone)]
 pub struct DataDir {
    base: PathBuf,
 }
 impl DataDir {
    /// Create a DataDir with a custom base path.
    pub fn new(base: impl Into<PathBuf>) -> Self {
        Self { base: base.into() }
    }
    /// Create a DataDir using the platform-specific data directory.
    ///
    /// - Linux: `~/.local/share/lattice/`
    /// - macOS: `~/Library/Application Support/lattice/`
    /// - Windows: `C:\Users\<user>\AppData\Roaming\lattice\`
    pub fn default_location() -> Option<Self> {
        dirs::data_dir().map(|d| Self::new(d.join(APP_NAME)))
    }
    /// Get the base directory path.
    pub fn base(&self) -> &Path {
        &self.base
    }
    /// Get the path to the identity key file.
    pub fn identity_key(&self) -> PathBuf {
        self.base.join("identity.key")
    }
    /// Get the path to the logs directory.
    pub fn logs_dir(&self) -> PathBuf {
        self.base.join("logs")
    }
    /// Get the path to a specific author's log file.
    pub fn log_file(&self, author_id_hex: &str) -> PathBuf {
        self.logs_dir().join(format!("{}.log", author_id_hex))
    }
    /// Get the path to the state database.
    pub fn state_db(&self) -> PathBuf {
        self.base.join("state.db")
    }
    /// Ensure all required directories exist.
    pub fn ensure_dirs(&self) -> std::io::Result<()> {
        std::fs::create_dir_all(&self.base)?;
        std::fs::create_dir_all(self.logs_dir())?;
        Ok(())
    }
 }
 impl Default for DataDir {
    fn default() -> Self {
        Self::default_location().unwrap_or_else(|| Self::new("./data"))
    }
 }
 #[cfg(test)]
 mod tests {
    use super::*;
    #[test]
    fn test_custom_path() {
        let dd = DataDir::new("/custom/path");
        assert_eq!(dd.base(), Path::new("/custom/path"));
        assert_eq!(dd.identity_key(), PathBuf::from("/custom/path/identity.key"));
        assert_eq!(dd.logs_dir(), PathBuf::from("/custom/path/logs"));
        assert_eq!(dd.state_db(), PathBuf::from("/custom/path/state.db"));
    }
    #[test]
    fn test_log_file_path() {
        let dd = DataDir::new("/data");
        let path = dd.log_file("abc123");
        assert_eq!(path, PathBuf::from("/data/logs/abc123.log"));
    }
    #[test]
    fn test_default_location_exists() {
        // On most systems, default_location should return Some
        let location = DataDir::default_location();
        // Just verify it doesn't panic - actual path varies by platform
        assert!(location.is_some() || true);
    }
    #[test]
    fn test_default_impl() {
        let dd = DataDir::default();
        // Should either be platform default or ./data fallback
        assert!(dd.base().to_str().is_some());
    }
 }
@@ -0,0 +1,9 @@
 //! Log entries (atomic operations)
 /// An atomic, batched operation in the sigchain.
 ///
 /// Entries are the fundamental unit of change in Lattice.
 /// The KV store is a "view" generated by replaying these entries.
 pub struct Entry {
    // TODO: operation data, signature, prev_hash
 }
@@ -0,0 +1,325 @@
 //! Hybrid Logical Clock (HLC) implementation
 //!
 //! HLCs combine wall clock time with a logical counter to provide
 //! causally consistent ordering even with clock drift.
 use std::cmp::Ordering;
 use crate::clock::{Clock, SystemClock};
 /// Default maximum drift allowed before clamping (1 hour in ms)
 pub const DEFAULT_MAX_DRIFT_MS: u64 = 60 * 60 * 1000;
 /// Hybrid Logical Clock
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub struct HLC {
    /// Wall clock time in milliseconds since Unix epoch
    pub wall_time: u64,
    /// Logical counter for ordering events at same wall_time
    pub counter: u32,
 }
 impl HLC {
    /// Create a new HLC with the given wall_time and counter
    pub fn new(wall_time: u64, counter: u32) -> Self {
        Self { wall_time, counter }
    }
    /// Create an HLC from the current system time
    pub fn now() -> Self {
        Self::now_with_clock(&SystemClock)
    }
    /// Create an HLC from the given clock (for testing)
    pub fn now_with_clock(clock: &impl Clock) -> Self {
        Self {
            wall_time: clock.now_ms(),
            counter: 0,
        }
    }
    /// Update this clock upon receiving a message with the given HLC.
    /// Uses the system clock.
    pub fn update(&self, received: &HLC) -> HLC {
        self.update_with_clock(received, &SystemClock)
    }
    /// Update this clock with an explicit clock source (for testing)
    pub fn update_with_clock(&self, received: &HLC, clock: &impl Clock) -> HLC {
        let local_wall_time = clock.now_ms();
        if local_wall_time > self.wall_time && local_wall_time > received.wall_time {
            // Local wall clock is ahead of everything, use it
            HLC::new(local_wall_time, 0)
        } else if self.wall_time > received.wall_time {
            // Our last HLC is ahead, increment counter
            HLC::new(self.wall_time, self.counter + 1)
        } else if received.wall_time > self.wall_time {
            // Received HLC is ahead, use it and increment
            HLC::new(received.wall_time, received.counter + 1)
        } else {
            // Same wall_time, take max counter and increment
            HLC::new(self.wall_time, self.counter.max(received.counter) + 1)
        }
    }
    /// Clamp a potentially-future HLC to be at most parent + 1.
    /// Returns the clamped HLC if the original exceeds max_drift,
    /// otherwise returns the original.
    pub fn clamp_future(&self, parent: &HLC, local_wall_time: u64, max_drift_ms: u64) -> HLC {
        if self.wall_time > local_wall_time + max_drift_ms {
            // Clamp to parent + 1 (deterministic across all nodes)
            HLC::new(parent.wall_time, parent.counter + 1)
        } else {
            *self
        }
    }
    /// Clamp with clock source (convenience method)
    pub fn clamp_future_with_clock(&self, parent: &HLC, clock: &impl Clock, max_drift_ms: u64) -> HLC {
        self.clamp_future(parent, clock.now_ms(), max_drift_ms)
    }
    /// Check if this HLC exceeds the given wall time by more than max_drift
    pub fn is_future(&self, local_wall_time: u64, max_drift_ms: u64) -> bool {
        self.wall_time > local_wall_time + max_drift_ms
    }
    /// Increment this HLC for a new local event
    pub fn tick(&self) -> HLC {
        self.tick_with_clock(&SystemClock)
    }
    /// Increment with explicit clock (for testing)
    pub fn tick_with_clock(&self, clock: &impl Clock) -> HLC {
        let now = clock.now_ms();
        if now > self.wall_time {
            HLC::new(now, 0)
        } else {
            HLC::new(self.wall_time, self.counter + 1)
        }
    }
 }
 impl Ord for HLC {
    fn cmp(&self, other: &Self) -> Ordering {
        match self.wall_time.cmp(&other.wall_time) {
            Ordering::Equal => self.counter.cmp(&other.counter),
            other => other,
        }
    }
 }
 impl PartialOrd for HLC {
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        Some(self.cmp(other))
    }
 }
 impl Default for HLC {
    fn default() -> Self {
        Self::now()
    }
 }
 #[cfg(test)]
 mod tests {
    use super::*;
    use crate::clock::MockClock;
    #[test]
    fn test_hlc_ordering() {
        let a = HLC::new(100, 0);
        let b = HLC::new(100, 1);
        let c = HLC::new(101, 0);
        assert!(a < b);
        assert!(b < c);
        assert!(a < c);
    }
    #[test]
    fn test_hlc_update_received_ahead() {
        let local = HLC::new(100, 5);
        let received = HLC::new(200, 3);
        let clock = MockClock::new(50); // Wall clock behind both
        let updated = local.update_with_clock(&received, &clock);
        assert!(updated > received);
        assert_eq!(updated.wall_time, 200);
        assert_eq!(updated.counter, 4);
    }
    #[test]
    fn test_hlc_update_local_ahead() {
        let local = HLC::new(200, 5);
        let received = HLC::new(100, 3);
        let clock = MockClock::new(50); // Wall clock behind both
        let updated = local.update_with_clock(&received, &clock);
        assert!(updated > local);
        assert_eq!(updated.wall_time, 200);
        assert_eq!(updated.counter, 6);
    }
    #[test]
    fn test_hlc_update_wall_clock_ahead() {
        let local = HLC::new(100, 5);
        let received = HLC::new(150, 3);
        let clock = MockClock::new(500); // Wall clock ahead of both
        let updated = local.update_with_clock(&received, &clock);
        assert_eq!(updated.wall_time, 500);
        assert_eq!(updated.counter, 0);
    }
    #[test]
    fn test_hlc_clamp_future() {
        let future = HLC::new(2050_000_000_000, 0);
        let parent = HLC::new(100, 5);
        let clock = MockClock::new(1000);
        let clamped = future.clamp_future_with_clock(&parent, &clock, DEFAULT_MAX_DRIFT_MS);
        assert_eq!(clamped.wall_time, 100);
        assert_eq!(clamped.counter, 6);
    }
    #[test]
    fn test_hlc_clamp_within_drift() {
        let normal = HLC::new(1000, 3);
        let parent = HLC::new(100, 5);
        let clock = MockClock::new(900);
        let clamped = normal.clamp_future_with_clock(&parent, &clock, DEFAULT_MAX_DRIFT_MS);
        assert_eq!(clamped, normal);
    }
    #[test]
    fn test_clock_in_past_uses_received() {
        // Simulates Pi booting with clock at 1970
        let old_clock = HLC::new(0, 0);
        let received = HLC::new(1_700_000_000_000, 5);
        let clock = MockClock::new(0); // Clock also at 1970
        let updated = old_clock.update_with_clock(&received, &clock);
        assert_eq!(updated.wall_time, 1_700_000_000_000);
        assert_eq!(updated.counter, 6);
    }
    #[test]
    fn test_tick_with_mock_clock() {
        let hlc = HLC::new(100, 5);
        // Clock behind: counter increments
        let clock = MockClock::new(50);
        let ticked = hlc.tick_with_clock(&clock);
        assert_eq!(ticked.wall_time, 100);
        assert_eq!(ticked.counter, 6);
        // Clock ahead: use new wall time
        let clock = MockClock::new(200);
        let ticked = hlc.tick_with_clock(&clock);
        assert_eq!(ticked.wall_time, 200);
        assert_eq!(ticked.counter, 0);
    }
    #[test]
    fn test_now_with_mock_clock() {
        let clock = MockClock::new(12345);
        let hlc = HLC::now_with_clock(&clock);
        assert_eq!(hlc.wall_time, 12345);
        assert_eq!(hlc.counter, 0);
    }
    #[test]
    fn test_hlc_update_same_wall_time_collision() {
        // Both local and received have the same wall_time (collision branch)
        let local = HLC::new(100, 5);
        let received = HLC::new(100, 8);
        let clock = MockClock::new(50); // Wall clock behind both
        let updated = local.update_with_clock(&received, &clock);
        // Should take max(5, 8) + 1 = 9
        assert_eq!(updated.wall_time, 100);
        assert_eq!(updated.counter, 9);
    }
    #[test]
    fn test_hlc_update_same_wall_time_local_counter_higher() {
        // Same wall_time, but local has higher counter
        let local = HLC::new(100, 10);
        let received = HLC::new(100, 3);
        let clock = MockClock::new(50);
        let updated = local.update_with_clock(&received, &clock);
        // Should take max(10, 3) + 1 = 11
        assert_eq!(updated.wall_time, 100);
        assert_eq!(updated.counter, 11);
    }
    #[test]
    fn test_is_future() {
        let hlc = HLC::new(1000, 0);
        // Within drift: not future
        assert!(!hlc.is_future(500, 600));
        // Exactly at drift boundary: not future (>= vs >)
        assert!(!hlc.is_future(500, 500));
        // Beyond drift: is future
        assert!(hlc.is_future(500, 400));
        // Way in the future
        let future = HLC::new(2050_000_000_000, 0);
        assert!(future.is_future(1_700_000_000_000, DEFAULT_MAX_DRIFT_MS));
    }
    #[test]
    fn test_system_clock_smoke() {
        // Ensure SystemClock compiles and returns reasonable values
        let hlc = HLC::now();
        // Should be after 2025-01-01 (1735689600000 ms)
        assert!(hlc.wall_time > 1_735_689_600_000);
        assert_eq!(hlc.counter, 0);
    }
    #[test]
    fn test_default_uses_system_clock() {
        let hlc = HLC::default();
        // Should be after 2025-01-01
        assert!(hlc.wall_time > 1_735_689_600_000);
        assert_eq!(hlc.counter, 0);
    }
    #[test]
    fn test_update_with_system_clock_smoke() {
        let local = HLC::new(100, 5);
        let received = HLC::new(200, 3);
        // This should use the real system clock internally
        let updated = local.update(&received);
        // Updated should be greater than both
        assert!(updated > local);
        assert!(updated > received);
    }
    #[test]
    fn test_tick_with_system_clock_smoke() {
        let hlc = HLC::new(100, 5);
        let ticked = hlc.tick();
        // Should be greater than original
        assert!(ticked > hlc);
    }
 }
@@ -0,0 +1,38 @@
 //! Lattice Core
 //!
 //! Core types for the Lattice distributed mesh:
 //! - **Node**: Identity with Ed25519 keypair
 //! - **SigChain**: Append-only cryptographically signed log
 //! - **Entry**: Atomic operations in the log
 //! - **VectorClock**: Causality tracking for reconciliation
 //! - **HLC**: Hybrid Logical Clock for ordering
 //! - **Clock**: Time abstraction for testability
 //! - **Proto**: Generated protobuf types from lattice.proto
 //! - **DataDir**: Platform-specific data directory paths
 //! - **SignedEntry**: Entry creation, signing, and verification
 //! - **Log**: Append-only log file I/O
 pub mod node;
 pub mod sigchain;
 pub mod entry;
 pub mod vector_clock;
 pub mod hlc;
 pub mod clock;
 pub mod proto;
 pub mod data_dir;
 pub mod signed_entry;
 pub mod log;
 // Constants
 /// Maximum size of a serialized SignedEntry (16 MB)
 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 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};
@@ -0,0 +1,534 @@
 //! Log file I/O for append-only entry storage
 //!
 //! Each author has a log file containing length-delimited LogRecord messages.
 //! LogRecord = { hash: [u8; 32], entry_bytes: SignedEntry }
 use crate::proto::{LogRecord, SignedEntry};
 use crate::MAX_ENTRY_SIZE;
 use prost::Message;
 use std::fs::{File, OpenOptions};
 use std::io::{self, BufReader, BufWriter, Read, Write};
 use std::path::Path;
 use thiserror::Error;
 /// Errors that can occur during log operations
 #[derive(Error, Debug)]
 pub enum LogError {
    #[error("IO error: {0}")]
    Io(#[from] io::Error),
    #[error("Proto decode error: {0}")]
    Decode(#[from] prost::DecodeError),
    #[error("Entry too large: {0} bytes (max {MAX_ENTRY_SIZE})")]
    EntryTooLarge(usize),
    #[error("Unexpected EOF while reading entry")]
    UnexpectedEof,
    #[error("Hash mismatch: stored hash does not match computed hash")]
    HashMismatch,
 }
 /// Append a SignedEntry to a log file as a LogRecord
 pub fn append_entry(path: impl AsRef<Path>, entry: &SignedEntry) -> Result<u64, LogError> {
    let path = path.as_ref();
    let file = OpenOptions::new()
        .create(true)
        .append(true)
        .open(path)?;
    let mut writer = BufWriter::new(file);
    // Serialize the SignedEntry
    let entry_bytes = entry.encode_to_vec();
    if entry_bytes.len() > MAX_ENTRY_SIZE {
        return Err(LogError::EntryTooLarge(entry_bytes.len()));
    }
    // Compute hash
    let hash: [u8; 32] = blake3::hash(&entry_bytes).into();
    // Create LogRecord
    let record = LogRecord {
        hash: hash.to_vec(),
        entry_bytes,
    };
    // Write length-delimited LogRecord
    let mut buf = Vec::new();
    record.encode_length_delimited(&mut buf)
        .map_err(|e| io::Error::new(io::ErrorKind::Other, e))?;
    writer.write_all(&buf)?;
    writer.flush()?;
    // Ensure data is physically written to disk
    writer.get_ref().sync_all()?;
    // Return new file size
    let metadata = std::fs::metadata(path)?;
    Ok(metadata.len())
 }
 /// Read all SignedEntry messages from a log file (with hash verification)
 pub fn read_entries(path: impl AsRef<Path>) -> Result<Vec<SignedEntry>, LogError> {
    read_entries_after(path, None)
 }
 /// Read all entries that come AFTER the given hash.
 /// If `last_hash` is None, reads all entries.
 pub fn read_entries_after(path: impl AsRef<Path>, last_hash: Option<[u8; 32]>) -> Result<Vec<SignedEntry>, LogError> {
    let reader = match LogReader::open(&path) {
        Ok(r) => r,
        Err(LogError::Io(e)) if e.kind() == io::ErrorKind::NotFound => return Ok(Vec::new()),
        Err(e) => return Err(e),
    };
    let mut entries = Vec::new();
    let mut found_start = last_hash.is_none();
    for result in reader {
        let (hash, entry) = result?;
        if found_start {
            entries.push(entry);
        } else if let Some(target) = last_hash {
            if hash == target {
                found_start = true;
            }
        }
    }
    Ok(entries)
 }
 /// Iterator over entries in a log file
 /// Returns (hash, SignedEntry) pairs
 pub struct LogReader {
    reader: BufReader<File>,
 }
 impl LogReader {
    /// Open a log file for reading
    pub fn open(path: impl AsRef<Path>) -> Result<Self, LogError> {
        let file = File::open(path)?;
        Ok(Self {
            reader: BufReader::new(file),
        })
    }
 }
 impl Iterator for LogReader {
    type Item = Result<([u8; 32], SignedEntry), LogError>;
    fn next(&mut self) -> Option<Self::Item> {
        match read_one_record(&mut self.reader) {
            Ok(Some(pair)) => Some(Ok(pair)),
            Ok(None) => None,
            Err(e) => Some(Err(e)),
        }
    }
 }
 /// Read a single LogRecord, returning (hash, SignedEntry)
 fn read_one_record<R: Read>(reader: &mut R) -> Result<Option<([u8; 32], SignedEntry)>, LogError> {
    // Read length-delimited bytes
    let record_bytes = match read_length_delimited_bytes(reader) {
        Ok(bytes) => bytes,
        Err(LogError::Io(e)) if e.kind() == io::ErrorKind::UnexpectedEof => return Ok(None),
        Err(e) => return Err(e),
    };
    // Decode LogRecord
    let record = LogRecord::decode(&record_bytes[..])?;
    // Verify hash
    let computed_hash: [u8; 32] = blake3::hash(&record.entry_bytes).into();
    let stored_hash: [u8; 32] = record.hash.try_into()
        .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "invalid hash length"))?;
    if computed_hash != stored_hash {
        return Err(LogError::HashMismatch);
    }
    // Decode SignedEntry
    let entry = SignedEntry::decode(&record.entry_bytes[..])?;
    Ok(Some((stored_hash, entry)))
 }
 /// Read length-delimited bytes from a reader
 fn read_length_delimited_bytes<R: Read>(reader: &mut R) -> Result<Vec<u8>, LogError> {
    // Read varint length prefix
    let mut prefix_buf = Vec::with_capacity(10);
    let mut byte = [0u8; 1];
    loop {
        match reader.read_exact(&mut byte) {
            Ok(()) => {}
            Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => return Err(e.into()),
            Err(e) => return Err(e.into()),
        }
        prefix_buf.push(byte[0]);
        if byte[0] & 0x80 == 0 {
            break;
        }
        if prefix_buf.len() > 10 {
            return Err(io::Error::new(io::ErrorKind::InvalidData, "varint too long").into());
        }
    }
    // Decode the length
    let len = prost::decode_length_delimiter(&prefix_buf[..])
        .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
    if len > MAX_ENTRY_SIZE {
        return Err(LogError::EntryTooLarge(len));
    }
    // Read the data
    let mut data_buf = vec![0u8; len];
    reader.read_exact(&mut data_buf).map_err(|e| {
        if e.kind() == io::ErrorKind::UnexpectedEof {
            LogError::UnexpectedEof
        } else {
            e.into()
        }
    })?;
    Ok(data_buf)
 }
 #[cfg(test)]
 mod tests {
    use super::*;
    use crate::clock::MockClock;
    use crate::hlc::HLC;
    use crate::node::Node;
    use crate::signed_entry::EntryBuilder;
    use std::env::temp_dir;
    fn temp_log_path(name: &str) -> std::path::PathBuf {
        temp_dir().join(format!("lattice_test_{}.log", name))
    }
    /// Compute hash the same way append_entry does
    fn compute_entry_hash(entry: &SignedEntry) -> [u8; 32] {
        let entry_bytes = entry.encode_to_vec();
        blake3::hash(&entry_bytes).into()
    }
    #[test]
    fn test_append_and_read_single() {
        let path = temp_log_path("single_v6");
        std::fs::remove_file(&path).ok();
        let node = Node::generate();
        let clock = MockClock::new(1000);
        let hlc = HLC::now_with_clock(&clock);
        let entry = EntryBuilder::new(1, hlc)
            .put("/test/key", b"value".to_vec())
            .sign(&node);
        append_entry(&path, &entry).unwrap();
        let entries = read_entries(&path).unwrap();
        assert_eq!(entries.len(), 1);
        assert_eq!(entries[0].entry_bytes, entry.entry_bytes);
        std::fs::remove_file(&path).ok();
    }
    #[test]
    fn test_append_multiple() {
        let path = temp_log_path("multiple_v6");
        std::fs::remove_file(&path).ok();
        let node = Node::generate();
        let clock = MockClock::new(1000);
        for i in 1..=5 {
            let entry = EntryBuilder::new(i, HLC::now_with_clock(&clock))
                .put(format!("/key/{}", i), format!("value{}", i).into_bytes())
                .sign(&node);
            append_entry(&path, &entry).unwrap();
        }
        let entries = read_entries(&path).unwrap();
        assert_eq!(entries.len(), 5);
        std::fs::remove_file(&path).ok();
    }
    #[test]
    fn test_read_entries_after() {
        let path = temp_log_path("after_v6");
        std::fs::remove_file(&path).ok();
        let node = Node::generate();
        let clock = MockClock::new(1000);
        let mut entries = Vec::new();
        let mut hashes = Vec::new();
        for i in 1..=5 {
            let entry = EntryBuilder::new(i, HLC::now_with_clock(&clock))
                .put(format!("/key/{}", i), format!("value{}", i).into_bytes())
                .sign(&node);
            hashes.push(compute_entry_hash(&entry));
            append_entry(&path, &entry).unwrap();
            entries.push(entry);
        }
        // Read entries after hash[1] (second entry) -> should get entries 3, 4, 5
        let result = read_entries_after(&path, Some(hashes[1])).unwrap();
        assert_eq!(result.len(), 3);
        assert_eq!(result[0].entry_bytes, entries[2].entry_bytes);
        // Read all entries (no hash)
        let all = read_entries_after(&path, None).unwrap();
        assert_eq!(all.len(), 5);
        std::fs::remove_file(&path).ok();
    }
    #[test]
    fn test_hash_not_found_returns_empty() {
        let path = temp_log_path("not_found_v6");
        std::fs::remove_file(&path).ok();
        let node = Node::generate();
        let clock = MockClock::new(1000);
        let entry = EntryBuilder::new(1, HLC::now_with_clock(&clock))
            .put("/key", b"value".to_vec())
            .sign(&node);
        append_entry(&path, &entry).unwrap();
        let fake_hash = [0u8; 32];
        let result = read_entries_after(&path, Some(fake_hash)).unwrap();
        assert_eq!(result.len(), 0);
        std::fs::remove_file(&path).ok();
    }
    #[test]
    fn test_log_reader_returns_hash() {
        let path = temp_log_path("reader_hash_v6");
        std::fs::remove_file(&path).ok();
        let node = Node::generate();
        let clock = MockClock::new(1000);
        let entry = EntryBuilder::new(1, HLC::now_with_clock(&clock))
            .put("/key", b"value".to_vec())
            .sign(&node);
        let expected_hash = compute_entry_hash(&entry);
        append_entry(&path, &entry).unwrap();
        let mut reader = LogReader::open(&path).unwrap();
        let (hash, _) = reader.next().unwrap().unwrap();
        assert_eq!(hash, expected_hash);
        std::fs::remove_file(&path).ok();
    }
    #[test]
    fn test_read_empty_file() {
        let path = temp_log_path("empty_v6");
        std::fs::remove_file(&path).ok();
        File::create(&path).unwrap();
        let entries = read_entries(&path).unwrap();
        assert_eq!(entries.len(), 0);
        std::fs::remove_file(&path).ok();
    }
    #[test]
    fn test_read_nonexistent() {
        let path = temp_log_path("nonexistent_v6");
        std::fs::remove_file(&path).ok();
        let entries = read_entries(&path).unwrap();
        assert_eq!(entries.len(), 0);
    }
    // --- Negative Tests ---
    #[test]
    fn test_corrupted_entry_detected() {
        use std::io::Seek;
        let path = temp_log_path("corrupted_v6");
        std::fs::remove_file(&path).ok();
        let node = Node::generate();
        let clock = MockClock::new(1000);
        let entry = EntryBuilder::new(1, HLC::now_with_clock(&clock))
            .put("/key", b"original".to_vec())
            .sign(&node);
        append_entry(&path, &entry).unwrap();
        // Corrupt the file: change a byte in the middle
        let mut file = OpenOptions::new().write(true).open(&path).unwrap();
        file.seek(io::SeekFrom::Start(20)).unwrap();
        file.write_all(&[0xFF]).unwrap();
        drop(file);
        let result = read_entries(&path);
        match result {
            Err(LogError::HashMismatch) => (),
            Err(LogError::Decode(_)) => (),
            Err(e) => panic!("Expected HashMismatch or Decode error, got: {:?}", e),
            Ok(_) => panic!("Corrupted entry was accepted!"),
        }
        std::fs::remove_file(&path).ok();
    }
    #[test]
    fn test_truncated_file() {
        let path = temp_log_path("truncated_v6");
        std::fs::remove_file(&path).ok();
        let node = Node::generate();
        let clock = MockClock::new(1000);
        let entry = EntryBuilder::new(1, HLC::now_with_clock(&clock))
            .put("/key", b"data".to_vec())
            .sign(&node);
        append_entry(&path, &entry).unwrap();
        // Truncate file by 1 byte
        let file = OpenOptions::new().write(true).open(&path).unwrap();
        let len = file.metadata().unwrap().len();
        file.set_len(len - 1).unwrap();
        drop(file);
        let result = read_entries(&path);
        match result {
            Err(LogError::UnexpectedEof) => (),
            Err(LogError::Io(e)) if e.kind() == io::ErrorKind::UnexpectedEof => (),
            Err(LogError::Decode(_)) => (), // Also acceptable
            res => panic!("Expected UnexpectedEof, got: {:?}", res),
        }
        std::fs::remove_file(&path).ok();
    }
    #[test]
    fn test_append_too_large() {
        let path = temp_log_path("too_large_v6");
        std::fs::remove_file(&path).ok();
        let node = Node::generate();
        let clock = MockClock::new(1000);
        // Create payload larger than MAX_ENTRY_SIZE
        let huge_payload = vec![0u8; crate::MAX_ENTRY_SIZE + 100];
        let entry = EntryBuilder::new(1, HLC::now_with_clock(&clock))
            .put("/huge", huge_payload)
            .sign(&node);
        let result = append_entry(&path, &entry);
        match result {
            Err(LogError::EntryTooLarge(size)) => assert!(size > crate::MAX_ENTRY_SIZE),
            _ => panic!("Expected EntryTooLarge error"),
        }
        std::fs::remove_file(&path).ok();
    }
    #[test]
    fn test_read_after_last_element() {
        let path = temp_log_path("boundary_last_v6");
        std::fs::remove_file(&path).ok();
        let node = Node::generate();
        let clock = MockClock::new(1000);
        let entry = EntryBuilder::new(1, HLC::now_with_clock(&clock))
            .put("/key", b"val".to_vec())
            .sign(&node);
        append_entry(&path, &entry).unwrap();
        let hash = compute_entry_hash(&entry);
        // Ask for everything AFTER the only entry
        let result = read_entries_after(&path, Some(hash)).unwrap();
        // Result must be empty
        assert_eq!(result.len(), 0);
        std::fs::remove_file(&path).ok();
    }
    #[test]
    fn test_read_entry_exceeding_limit() {
        let path = temp_log_path("huge_read_v6");
        std::fs::remove_file(&path).ok();
        // Write a length prefix claiming the entry is > MAX_ENTRY_SIZE
        let mut file = File::create(&path).unwrap();
        let too_big = (crate::MAX_ENTRY_SIZE + 1) as usize;
        let mut buf = Vec::new();
        prost::encode_length_delimiter(too_big, &mut buf).unwrap();
        file.write_all(&buf).unwrap();
        // Write some dummy bytes (reader should reject before reading these)
        file.write_all(&[0u8; 10]).unwrap();
        drop(file);
        let result = read_entries(&path);
        match result {
            Err(LogError::EntryTooLarge(size)) => assert_eq!(size, too_big),
            _ => panic!("Expected EntryTooLarge before allocating RAM"),
        }
        std::fs::remove_file(&path).ok();
    }
    #[test]
    fn test_corruption_in_middle_of_stream() {
        let path = temp_log_path("corruption_middle_v6");
        std::fs::remove_file(&path).ok();
        let node = Node::generate();
        let clock = MockClock::new(1000);
        // Write 3 entries
        for i in 0..3 {
            let entry = EntryBuilder::new(i + 1, HLC::now_with_clock(&clock))
                .put(format!("/key/{}", i), b"val".to_vec())
                .sign(&node);
            append_entry(&path, &entry).unwrap();
        }
        // Corrupt a byte in the middle of the file
        let mut file_bytes = std::fs::read(&path).unwrap();
        let mid_idx = file_bytes.len() / 2;
        file_bytes[mid_idx] = !file_bytes[mid_idx]; // Bitflip
        std::fs::write(&path, file_bytes).unwrap();
        // Reading should fail at some point
        let result = read_entries(&path);
        // Must fail with HashMismatch or DecodeError  
        assert!(result.is_err());
        std::fs::remove_file(&path).ok();
    }
 }
@@ -0,0 +1,215 @@
 //! Node identity and cryptographic keys
 //!
 //! Each node has an Ed25519 keypair:
 //! - Private key: stored locally in `identity.key` (never replicated)
 //! - Public key: serves as the node's identity (32 bytes)
 use ed25519_dalek::{Signature, Signer, SigningKey, Verifier, VerifyingKey};
 use rand::rngs::OsRng;
 use std::fs;
 use std::io::{self, Read, Write};
 use std::path::Path;
 use thiserror::Error;
 /// Errors that can occur during node operations
 #[derive(Error, Debug)]
 pub enum NodeError {
    #[error("IO error: {0}")]
    Io(#[from] io::Error),
    #[error("Invalid key length: expected 32 bytes, got {0}")]
    InvalidKeyLength(usize),
    #[error("Invalid signature")]
    InvalidSignature,
 }
 /// A node in the Lattice mesh.
 ///
 /// Each node has an Ed25519 keypair used for signing sigchain entries
 /// and establishing trust within the network.
 pub struct Node {
    signing_key: SigningKey,
 }
 impl Node {
    /// Generate a new node with a random keypair.
    pub fn generate() -> Self {
        let signing_key = SigningKey::generate(&mut OsRng);
        Self { signing_key }
    }
    /// Create a node from an existing signing key.
    pub fn from_signing_key(signing_key: SigningKey) -> Self {
        Self { signing_key }
    }
    /// Load a node's identity from a key file, or generate and save if it doesn't exist.
    pub fn load_or_generate(path: impl AsRef<Path>) -> Result<Self, NodeError> {
        let path = path.as_ref();
        if path.exists() {
            Self::load(path)
        } else {
            let node = Self::generate();
            node.save(path)?;
            Ok(node)
        }
    }
    /// Load a node's identity from a key file.
    pub fn load(path: impl AsRef<Path>) -> Result<Self, NodeError> {
        let mut file = fs::File::open(path)?;
        let mut bytes = Vec::new();
        file.read_to_end(&mut bytes)?;
        if bytes.len() != 32 {
            return Err(NodeError::InvalidKeyLength(bytes.len()));
        }
        let key_bytes: [u8; 32] = bytes.try_into().unwrap();
        let signing_key = SigningKey::from_bytes(&key_bytes);
        Ok(Self { signing_key })
    }
    /// Save the node's private key to a file.
    pub fn save(&self, path: impl AsRef<Path>) -> Result<(), NodeError> {
        let path = path.as_ref();
        // Create parent directories if they don't exist
        if let Some(parent) = path.parent() {
            fs::create_dir_all(parent)?;
        }
        let mut file = fs::File::create(path)?;
        file.write_all(self.signing_key.as_bytes())?;
        Ok(())
    }
    /// Get the node's public key (identity).
    pub fn public_key(&self) -> VerifyingKey {
        self.signing_key.verifying_key()
    }
    /// Get the node's public key as bytes (32 bytes).
    pub fn public_key_bytes(&self) -> [u8; 32] {
        self.signing_key.verifying_key().to_bytes()
    }
    /// Get the signing key for creating signatures.
    pub fn signing_key(&self) -> &SigningKey {
        &self.signing_key
    }
    /// Sign a message.
    pub fn sign(&self, message: &[u8]) -> Signature {
        self.signing_key.sign(message)
    }
    /// Verify a signature against this node's public key.
    pub fn verify(&self, message: &[u8], signature: &Signature) -> Result<(), NodeError> {
        self.public_key()
            .verify(message, signature)
            .map_err(|_| NodeError::InvalidSignature)
    }
    /// Verify a signature using a raw public key.
    pub fn verify_with_key(
        public_key: &VerifyingKey,
        message: &[u8],
        signature: &Signature,
    ) -> Result<(), NodeError> {
        public_key
            .verify(message, signature)
            .map_err(|_| NodeError::InvalidSignature)
    }
 }
 #[cfg(test)]
 mod tests {
    use super::*;
    use std::env::temp_dir;
    #[test]
    fn test_generate() {
        let node = Node::generate();
        let pk = node.public_key_bytes();
        assert_eq!(pk.len(), 32);
    }
    #[test]
    fn test_sign_and_verify() {
        let node = Node::generate();
        let message = b"hello lattice";
        let signature = node.sign(message);
        assert!(node.verify(message, &signature).is_ok());
    }
    #[test]
    fn test_verify_wrong_message() {
        let node = Node::generate();
        let signature = node.sign(b"original");
        assert!(node.verify(b"tampered", &signature).is_err());
    }
    #[test]
    fn test_verify_with_different_key() {
        let node1 = Node::generate();
        let node2 = Node::generate();
        let signature = node1.sign(b"message");
        assert!(node2.verify(b"message", &signature).is_err());
    }
    #[test]
    fn test_save_and_load() {
        let temp_path = temp_dir().join("lattice_test_identity.key");
        // Generate and save
        let node1 = Node::generate();
        let pk1 = node1.public_key_bytes();
        node1.save(&temp_path).unwrap();
        // Load and verify same key
        let node2 = Node::load(&temp_path).unwrap();
        let pk2 = node2.public_key_bytes();
        assert_eq!(pk1, pk2);
        // Cleanup
        fs::remove_file(&temp_path).ok();
    }
    #[test]
    fn test_load_or_generate() {
        let temp_path = temp_dir().join("lattice_test_identity2.key");
        // Remove if exists
        fs::remove_file(&temp_path).ok();
        // First call: generates
        let node1 = Node::load_or_generate(&temp_path).unwrap();
        let pk1 = node1.public_key_bytes();
        // Second call: loads existing
        let node2 = Node::load_or_generate(&temp_path).unwrap();
        let pk2 = node2.public_key_bytes();
        assert_eq!(pk1, pk2);
        // Cleanup
        fs::remove_file(&temp_path).ok();
    }
    #[test]
    fn test_verify_with_key_static() {
        let node = Node::generate();
        let pk = node.public_key();
        let message = b"test message";
        let signature = node.sign(message);
        assert!(Node::verify_with_key(&pk, message, &signature).is_ok());
    }
 }
@@ -0,0 +1,76 @@
 //! Generated protobuf types for Lattice
 //!
 //! This module re-exports types generated from `proto/lattice.proto`
 // Include the generated code from prost-build
 include!(concat!(env!("OUT_DIR"), "/lattice.rs"));
 #[cfg(test)]
 mod tests {
    use super::*;
    #[test]
    fn test_hlc_roundtrip() {
        let hlc = Hlc {
            wall_time: 1234567890,
            counter: 42,
        };
        // Encode
        let mut buf = Vec::new();
        prost::Message::encode(&hlc, &mut buf).unwrap();
        // Decode
        let decoded: Hlc = prost::Message::decode(&buf[..]).unwrap();
        assert_eq!(decoded.wall_time, 1234567890);
        assert_eq!(decoded.counter, 42);
    }
    #[test]
    fn test_entry_with_ops() {
        let entry = Entry {
            version: 1,
            prev_hash: vec![0u8; 32],
            seq: 5,
            timestamp: Some(Hlc {
                wall_time: 1000,
                counter: 0,
            }),
            ops: vec![
                Operation {
                    op_type: Some(operation::OpType::Put(PutOp {
                        key: "/nodes/abc".to_string(),
                        value: b"hello".to_vec(),
                    })),
                },
            ],
        };
        // Encode
        let mut buf = Vec::new();
        prost::Message::encode(&entry, &mut buf).unwrap();
        // Decode
        let decoded: Entry = prost::Message::decode(&buf[..]).unwrap();
        assert_eq!(decoded.version, 1);
        assert_eq!(decoded.seq, 5);
        assert_eq!(decoded.ops.len(), 1);
    }
    #[test]
    fn test_signed_entry() {
        let signed = SignedEntry {
            entry_bytes: vec![1, 2, 3, 4],
            signature: vec![0u8; 64],
            author_id: vec![0u8; 32],
        };
        let mut buf = Vec::new();
        prost::Message::encode(&signed, &mut buf).unwrap();
        let decoded: SignedEntry = prost::Message::decode(&buf[..]).unwrap();
        assert_eq!(decoded.entry_bytes, vec![1, 2, 3, 4]);
    }
 }
@@ -0,0 +1,430 @@
 //! Cryptographic SigChain (append-only signed log)
 //!
 //! A SigChain manages a single author's append-only log. It validates entries
 //! before appending (correct seq, prev_hash, valid signature) and persists to disk.
 use crate::log::{append_entry, read_entries, LogError};
 use crate::node::Node;
 use crate::proto::{Entry, SignedEntry};
 use crate::signed_entry::{hash_signed_entry, verify_signed_entry};
 use prost::Message;
 use std::path::{Path, PathBuf};
 use thiserror::Error;
 /// Errors that can occur during sigchain operations
 #[derive(Error, Debug)]
 pub enum SigChainError {
    #[error("Log error: {0}")]
    Log(#[from] LogError),
    #[error("Invalid signature")]
    InvalidSignature,
    #[error("Wrong author: expected {expected}, got {got}")]
    WrongAuthor { expected: String, got: String },
    #[error("Invalid sequence: expected {expected}, got {got}")]
    InvalidSequence { expected: u64, got: u64 },
    #[error("Invalid prev_hash: expected {expected}, got {got}")]
    InvalidPrevHash { expected: String, got: String },
    #[error("Decode error: {0}")]
    Decode(#[from] prost::DecodeError),
 }
 /// An append-only log where each entry is cryptographically signed
 /// and hash-linked to the previous entry.
 pub struct SigChain {
    /// Path to the log file
    log_path: PathBuf,
    /// Author's public key (32 bytes)
    author_id: [u8; 32],
    /// Next expected sequence number
    next_seq: u64,
    /// Hash of the last entry (zeroes if empty)
    last_hash: [u8; 32],
 }
 impl SigChain {
    /// Create a new empty sigchain for an author
    pub fn new(log_path: impl AsRef<Path>, author_id: [u8; 32]) -> Self {
        Self {
            log_path: log_path.as_ref().to_path_buf(),
            author_id,
            next_seq: 1,
            last_hash: [0u8; 32],
        }
    }
    /// Load a sigchain from an existing log file
    pub fn from_log(log_path: impl AsRef<Path>, author_id: [u8; 32]) -> Result<Self, SigChainError> {
        let log_path = log_path.as_ref().to_path_buf();
        let entries = read_entries(&log_path)?;
        let mut chain = Self::new(&log_path, author_id);
        for signed_entry in entries {
            // Verify signature
            verify_signed_entry(&signed_entry)
                .map_err(|_| SigChainError::InvalidSignature)?;
            // Validate author (author_id is in SignedEntry)
            let entry_author: [u8; 32] = signed_entry.author_id.clone().try_into()
                .unwrap_or([0u8; 32]);
            if entry_author != author_id {
                return Err(SigChainError::WrongAuthor {
                    expected: hex::encode(author_id),
                    got: hex::encode(&entry_author),
                });
            }
            // Decode Entry
            let entry = Entry::decode(&signed_entry.entry_bytes[..])?;
            // Validate sequence
            if entry.seq != chain.next_seq {
                return Err(SigChainError::InvalidSequence {
                    expected: chain.next_seq,
                    got: entry.seq,
                });
            }
            // Validate prev_hash
            let expected_prev: [u8; 32] = chain.last_hash;
            let got_prev: [u8; 32] = entry.prev_hash.try_into()
                .unwrap_or([0u8; 32]);
            if got_prev != expected_prev {
                return Err(SigChainError::InvalidPrevHash {
                    expected: hex::encode(expected_prev),
                    got: hex::encode(got_prev),
                });
            }
            // Update state
            chain.last_hash = hash_signed_entry(&signed_entry);
            chain.next_seq += 1;
        }
        Ok(chain)
    }
    /// Get the author's public key
    pub fn author_id(&self) -> &[u8; 32] {
        &self.author_id
    }
    /// Get the next expected sequence number
    pub fn next_seq(&self) -> u64 {
        self.next_seq
    }
    /// Get the hash of the last entry
    pub fn last_hash(&self) -> &[u8; 32] {
        &self.last_hash
    }
    /// Get the current length of the chain
    pub fn len(&self) -> u64 {
        self.next_seq - 1
    }
    /// Check if the chain is empty
    pub fn is_empty(&self) -> bool {
        self.next_seq == 1
    }
    /// Validate a signed entry without appending
    pub fn validate(&self, signed_entry: &SignedEntry) -> Result<(), SigChainError> {
        // Verify signature
        verify_signed_entry(signed_entry)
            .map_err(|_| SigChainError::InvalidSignature)?;
        // Validate author (author_id is in SignedEntry)
        let author: [u8; 32] = signed_entry.author_id.clone().try_into()
            .unwrap_or([0u8; 32]);
        if author != self.author_id {
            return Err(SigChainError::WrongAuthor {
                expected: hex::encode(self.author_id),
                got: hex::encode(author),
            });
        }
        // Decode entry
        let entry = Entry::decode(&signed_entry.entry_bytes[..])?;
        // Validate sequence
        if entry.seq != self.next_seq {
            return Err(SigChainError::InvalidSequence {
                expected: self.next_seq,
                got: entry.seq,
            });
        }
        // Validate prev_hash
        let prev: [u8; 32] = entry.prev_hash.try_into()
            .unwrap_or([0u8; 32]);
        if prev != self.last_hash {
            return Err(SigChainError::InvalidPrevHash {
                expected: hex::encode(self.last_hash),
                got: hex::encode(prev),
            });
        }
        Ok(())
    }
    /// Append a signed entry to the chain (validates first)
    pub fn append(&mut self, signed_entry: &SignedEntry) -> Result<(), SigChainError> {
        // Validate
        self.validate(signed_entry)?;
        // Write to log
        append_entry(&self.log_path, signed_entry)?;
        // Update state
        self.last_hash = hash_signed_entry(signed_entry);
        self.next_seq += 1;
        Ok(())
    }
    /// Create and append a new entry using the node's key
    pub fn create_entry(&mut self, node: &Node, ops: Vec<crate::proto::Operation>) -> Result<SignedEntry, SigChainError> {
        use crate::clock::SystemClock;
        use crate::hlc::HLC;
        use crate::signed_entry::EntryBuilder;
        let hlc = HLC::now_with_clock(&SystemClock);
        let mut builder = EntryBuilder::new(self.next_seq, hlc)
            .prev_hash(self.last_hash.to_vec());
        // Add operations
        for op in ops {
            builder = builder.operation(op);
        }
        let signed = builder.sign(node);
        self.append(&signed)?;
        Ok(signed)
    }
 }
 #[cfg(test)]
 mod tests {
    use super::*;
    use crate::clock::MockClock;
    use crate::hlc::HLC;
    use crate::node::Node;
    use crate::proto::{operation, Operation, PutOp};
    use crate::signed_entry::EntryBuilder;
    use std::env::temp_dir;
    fn temp_log_path(name: &str) -> PathBuf {
        temp_dir().join(format!("lattice_sigchain_test_{}.log", name))
    }
    #[test]
    fn test_new_sigchain() {
        let path = temp_log_path("new");
        let author = [1u8; 32];
        let chain = SigChain::new(&path, author);
        assert_eq!(chain.author_id(), &author);
        assert_eq!(chain.next_seq(), 1);
        assert_eq!(chain.last_hash(), &[0u8; 32]);
        assert!(chain.is_empty());
        assert_eq!(chain.len(), 0);
    }
    #[test]
    fn test_append_entry() {
        let path = temp_log_path("append");
        std::fs::remove_file(&path).ok();
        let node = Node::generate();
        let author = node.public_key_bytes();
        let mut chain = SigChain::new(&path, author);
        let clock = MockClock::new(1000);
        let entry = EntryBuilder::new(1, HLC::now_with_clock(&clock))
            .prev_hash([0u8; 32].to_vec())
            .put("/key", b"value".to_vec())
            .sign(&node);
        chain.append(&entry).unwrap();
        assert_eq!(chain.next_seq(), 2);
        assert_eq!(chain.len(), 1);
        assert!(!chain.is_empty());
        std::fs::remove_file(&path).ok();
    }
    #[test]
    fn test_append_multiple() {
        let path = temp_log_path("multiple");
        std::fs::remove_file(&path).ok();
        let node = Node::generate();
        let author = node.public_key_bytes();
        let mut chain = SigChain::new(&path, author);
        let clock = MockClock::new(1000);
        for i in 1..=3 {
            let entry = EntryBuilder::new(i, HLC::now_with_clock(&clock))
                .prev_hash(chain.last_hash.to_vec())
                .put(format!("/key/{}", i), format!("value{}", i).into_bytes())
                .sign(&node);
            chain.append(&entry).unwrap();
        }
        assert_eq!(chain.len(), 3);
        assert_eq!(chain.next_seq(), 4);
        std::fs::remove_file(&path).ok();
    }
    #[test]
    fn test_from_log() {
        let path = temp_log_path("from_log");
        std::fs::remove_file(&path).ok();
        let node = Node::generate();
        let author = node.public_key_bytes();
        let clock = MockClock::new(1000);
        // Write some entries
        {
            let mut chain = SigChain::new(&path, author);
            for i in 1..=3 {
                let entry = EntryBuilder::new(i, HLC::now_with_clock(&clock))
                    .prev_hash(chain.last_hash.to_vec())
                    .put("/key", b"val".to_vec())
                    .sign(&node);
                chain.append(&entry).unwrap();
            }
        }
        // Reload from log
        let chain = SigChain::from_log(&path, author).unwrap();
        assert_eq!(chain.len(), 3);
        assert_eq!(chain.next_seq(), 4);
        std::fs::remove_file(&path).ok();
    }
    #[test]
    fn test_reject_wrong_sequence() {
        let path = temp_log_path("wrong_seq");
        std::fs::remove_file(&path).ok();
        let node = Node::generate();
        let author = node.public_key_bytes();
        let mut chain = SigChain::new(&path, author);
        let clock = MockClock::new(1000);
        // Try to append with wrong seq (2 instead of 1)
        let entry = EntryBuilder::new(2, HLC::now_with_clock(&clock))
            .prev_hash([0u8; 32].to_vec())
            .put("/key", b"val".to_vec())
            .sign(&node);
        let result = chain.append(&entry);
        assert!(matches!(result, Err(SigChainError::InvalidSequence { .. })));
        std::fs::remove_file(&path).ok();
    }
    #[test]
    fn test_reject_wrong_prev_hash() {
        let path = temp_log_path("wrong_prev");
        std::fs::remove_file(&path).ok();
        let node = Node::generate();
        let author = node.public_key_bytes();
        let mut chain = SigChain::new(&path, author);
        let clock = MockClock::new(1000);
        // First entry
        let entry1 = EntryBuilder::new(1, HLC::now_with_clock(&clock))
            .prev_hash([0u8; 32].to_vec())
            .put("/key", b"v1".to_vec())
            .sign(&node);
        chain.append(&entry1).unwrap();
        // Second entry with wrong prev_hash
        let entry2 = EntryBuilder::new(2, HLC::now_with_clock(&clock))
            .prev_hash([99u8; 32].to_vec()) // Wrong!
            .put("/key", b"v2".to_vec())
            .sign(&node);
        let result = chain.append(&entry2);
        assert!(matches!(result, Err(SigChainError::InvalidPrevHash { .. })));
        std::fs::remove_file(&path).ok();
    }
    #[test]
    fn test_reject_wrong_author() {
        let path = temp_log_path("wrong_author");
        std::fs::remove_file(&path).ok();
        let node = Node::generate();
        let other_author = [99u8; 32]; // Different author
        let mut chain = SigChain::new(&path, other_author);
        let clock = MockClock::new(1000);
        // Entry signed by node but chain expects other_author
        let entry = EntryBuilder::new(1, HLC::now_with_clock(&clock))
            .prev_hash([0u8; 32].to_vec())
            .put("/key", b"val".to_vec())
            .sign(&node);
        let result = chain.append(&entry);
        assert!(matches!(result, Err(SigChainError::WrongAuthor { .. })));
        std::fs::remove_file(&path).ok();
    }
    #[test]
    fn test_create_entry() {
        let path = temp_log_path("create");
        std::fs::remove_file(&path).ok();
        let node = Node::generate();
        let author = node.public_key_bytes();
        let mut chain = SigChain::new(&path, author);
        let ops = vec![
            Operation {
                op_type: Some(operation::OpType::Put(PutOp {
                    key: "/test".to_string(),
                    value: b"hello".to_vec(),
                })),
            },
        ];
        let signed = chain.create_entry(&node, ops).unwrap();
        assert_eq!(chain.len(), 1);
        // Verify it was written
        let entries = read_entries(&path).unwrap();
        assert_eq!(entries.len(), 1);
        assert_eq!(entries[0].entry_bytes, signed.entry_bytes);
        std::fs::remove_file(&path).ok();
    }
 }
@@ -0,0 +1,266 @@
 //! Signed entry creation and verification
 //!
 //! Provides utilities for:
 //! - Building Entry messages with operations
 //! - Signing entries to create SignedEntry
 //! - Verifying signatures
 //! - Computing entry hashes for prev_hash linking
 use crate::hlc::HLC;
 use crate::node::{Node, NodeError};
 use crate::proto::{Entry, Hlc, Operation, PutOp, DeleteOp, SignedEntry, operation};
 use ed25519_dalek::{Signature, VerifyingKey};
 use prost::Message;
 use thiserror::Error;
 /// Errors that can occur during entry operations
 #[derive(Error, Debug)]
 pub enum EntryError {
    #[error("Signature verification failed: {0}")]
    Signature(#[from] NodeError),
    #[error("Proto decode error: {0}")]
    Decode(#[from] prost::DecodeError),
    #[error("Invalid signature length: expected 64 bytes, got {0}")]
    InvalidSignatureLength(usize),
    #[error("Invalid public key length: expected 32 bytes, got {0}")]
    InvalidPublicKeyLength(usize),
 }
 /// Builder for creating Entry messages
 pub struct EntryBuilder {
    version: u32,
    prev_hash: Vec<u8>,
    seq: u64,
    timestamp: HLC,
    ops: Vec<Operation>,
 }
 impl EntryBuilder {
    /// Create a new entry builder with the given sequence number and timestamp
    pub fn new(seq: u64, timestamp: HLC) -> Self {
        Self {
            version: 1,
            prev_hash: vec![0u8; 32], // Genesis or will be set
            seq,
            timestamp,
            ops: Vec::new(),
        }
    }
    /// Set the previous entry hash (for chaining)
    pub fn prev_hash(mut self, hash: impl Into<Vec<u8>>) -> Self {
        self.prev_hash = hash.into();
        self
    }
    /// Add a Put operation
    pub fn put(mut self, key: impl Into<String>, value: impl Into<Vec<u8>>) -> Self {
        self.ops.push(Operation {
            op_type: Some(operation::OpType::Put(PutOp {
                key: key.into(),
                value: value.into(),
            })),
        });
        self
    }
    /// Add a Delete operation
    pub fn delete(mut self, key: impl Into<String>) -> Self {
        self.ops.push(Operation {
            op_type: Some(operation::OpType::Delete(DeleteOp {
                key: key.into(),
            })),
        });
        self
    }
    /// Add a raw operation
    pub fn operation(mut self, op: Operation) -> Self {
        self.ops.push(op);
        self
    }
    /// Build the Entry proto message
    pub fn build(self) -> Entry {
        Entry {
            version: self.version,
            prev_hash: self.prev_hash,
            seq: self.seq,
            timestamp: Some(Hlc {
                wall_time: self.timestamp.wall_time,
                counter: self.timestamp.counter,
            }),
            ops: self.ops,
        }
    }
    /// Build and sign the entry, returning a SignedEntry
    pub fn sign(self, node: &Node) -> SignedEntry {
        let entry = self.build();
        sign_entry(&entry, node)
    }
 }
 /// Sign an Entry to create a SignedEntry
 pub fn sign_entry(entry: &Entry, node: &Node) -> SignedEntry {
    let entry_bytes = entry.encode_to_vec();
    let signature = node.sign(&entry_bytes);
    SignedEntry {
        entry_bytes,
        signature: signature.to_bytes().to_vec(),
        author_id: node.public_key_bytes().to_vec(),
    }
 }
 /// Verify a SignedEntry's signature
 pub fn verify_signed_entry(signed: &SignedEntry) -> Result<Entry, EntryError> {
    // Parse public key
    if signed.author_id.len() != 32 {
        return Err(EntryError::InvalidPublicKeyLength(signed.author_id.len()));
    }
    let pk_bytes: [u8; 32] = signed.author_id.clone().try_into().unwrap();
    let public_key = VerifyingKey::from_bytes(&pk_bytes)
        .map_err(|_| NodeError::InvalidSignature)?;
    // Parse signature
    if signed.signature.len() != 64 {
        return Err(EntryError::InvalidSignatureLength(signed.signature.len()));
    }
    let sig_bytes: [u8; 64] = signed.signature.clone().try_into().unwrap();
    let signature = Signature::from_bytes(&sig_bytes);
    // Verify
    Node::verify_with_key(&public_key, &signed.entry_bytes, &signature)?;
    // Decode entry
    let entry = Entry::decode(&signed.entry_bytes[..])?;
    Ok(entry)
 }
 /// Compute the BLAKE3 hash of a SignedEntry (for prev_hash linking)
 pub fn hash_signed_entry(signed: &SignedEntry) -> [u8; 32] {
    let bytes = signed.encode_to_vec();
    blake3::hash(&bytes).into()
 }
 /// Compute the BLAKE3 hash of entry_bytes (alternative for lighter hashing)
 pub fn hash_entry_bytes(entry_bytes: &[u8]) -> [u8; 32] {
    blake3::hash(entry_bytes).into()
 }
 #[cfg(test)]
 mod tests {
    use super::*;
    use crate::clock::MockClock;
    #[test]
    fn test_entry_builder() {
        let clock = MockClock::new(1000);
        let hlc = HLC::now_with_clock(&clock);
        let entry = EntryBuilder::new(1, hlc)
            .put("/test/key", b"value".to_vec())
            .delete("/test/old")
            .build();
        assert_eq!(entry.version, 1);
        assert_eq!(entry.seq, 1);
        assert_eq!(entry.ops.len(), 2);
    }
    #[test]
    fn test_sign_and_verify() {
        let node = Node::generate();
        let clock = MockClock::new(1000);
        let hlc = HLC::now_with_clock(&clock);
        let signed = EntryBuilder::new(1, hlc)
            .put("/nodes/abc", b"test".to_vec())
            .sign(&node);
        assert_eq!(signed.author_id.len(), 32);
        assert_eq!(signed.signature.len(), 64);
        // Verify
        let entry = verify_signed_entry(&signed).unwrap();
        assert_eq!(entry.seq, 1);
        assert_eq!(entry.ops.len(), 1);
    }
    #[test]
    fn test_verify_tampered_fails() {
        let node = Node::generate();
        let clock = MockClock::new(1000);
        let hlc = HLC::now_with_clock(&clock);
        let mut signed = EntryBuilder::new(1, hlc)
            .put("/key", b"value".to_vec())
            .sign(&node);
        // Tamper with entry bytes
        signed.entry_bytes[0] ^= 0xFF;
        assert!(verify_signed_entry(&signed).is_err());
    }
    #[test]
    fn test_verify_wrong_key_fails() {
        let node1 = Node::generate();
        let node2 = Node::generate();
        let clock = MockClock::new(1000);
        let hlc = HLC::now_with_clock(&clock);
        let mut signed = EntryBuilder::new(1, hlc)
            .put("/key", b"value".to_vec())
            .sign(&node1);
        // Replace author with different key
        signed.author_id = node2.public_key_bytes().to_vec();
        assert!(verify_signed_entry(&signed).is_err());
    }
    #[test]
    fn test_hash_signed_entry() {
        let node = Node::generate();
        let clock = MockClock::new(1000);
        let hlc = HLC::now_with_clock(&clock);
        let signed = EntryBuilder::new(1, hlc)
            .put("/key", b"value".to_vec())
            .sign(&node);
        let hash = hash_signed_entry(&signed);
        assert_eq!(hash.len(), 32);
        // Same entry should produce same hash
        let hash2 = hash_signed_entry(&signed);
        assert_eq!(hash, hash2);
    }
    #[test]
    fn test_prev_hash_chaining() {
        let node = Node::generate();
        let clock = MockClock::new(1000);
        // First entry
        let entry1 = EntryBuilder::new(1, HLC::now_with_clock(&clock))
            .put("/key", b"v1".to_vec())
            .sign(&node);
        let hash1 = hash_signed_entry(&entry1);
        // Second entry links to first
        let entry2 = EntryBuilder::new(2, HLC::now_with_clock(&clock))
            .prev_hash(hash1)
            .put("/key", b"v2".to_vec())
            .sign(&node);
        let decoded = verify_signed_entry(&entry2).unwrap();
        assert_eq!(decoded.prev_hash, hash1.to_vec());
    }
 }
@@ -0,0 +1,41 @@
 //! 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()
    }
 }
@@ -0,0 +1,18 @@
 [package]
 name = "lattice-net"
 description = "Networking layer for Lattice using Iroh"
 version.workspace = true
 edition.workspace = true
 license.workspace = true
 [dependencies]
 lattice-core = { workspace = true }
 iroh = { workspace = true }
 iroh-gossip = { workspace = true }
 tokio = { workspace = true }
 thiserror = { workspace = true }
 tracing = { workspace = true }
 bytes = { workspace = true }
 [dev-dependencies]
 tokio-test = { workspace = true }
@@ -0,0 +1,3 @@
 //! Gossip protocol for broadcasting changes
 // TODO: Implement gossip using iroh-gossip
@@ -0,0 +1,8 @@
 //! Lattice Networking
 //!
 //! Networking layer using Iroh:
 //! - **Gossip**: Broadcasting changes across the mesh
 //! - **Unicast**: Point-to-point communication for reconciliation
 pub mod gossip;
 pub mod unicast;
@@ -0,0 +1,3 @@
 //! Unicast communication for direct peer-to-peer messaging
 // TODO: Implement unicast using iroh
@@ -0,0 +1,15 @@
 [package]
 name = "lattice-store"
 description = "Log-based KV store with snapshots and watermarks"
 version.workspace = true
 edition.workspace = true
 license.workspace = true
 [dependencies]
 lattice-core = { workspace = true }
 tokio = { workspace = true }
 thiserror = { workspace = true }
 tracing = { workspace = true }
 [dev-dependencies]
 tokio-test = { workspace = true }
@@ -0,0 +1,12 @@
 //! Lattice Store
 //!
 //! Log-based Key-Value store:
 //! - State is a "view" generated by replaying entries
 //! - Watermarks for agreeing on safe compaction points
 //! - Snapshots for replacing old logs
 pub mod store;
 pub mod snapshot;
 pub mod watermark;
 pub use store::Store;
@@ -0,0 +1,6 @@
 //! Snapshots for log compaction
 /// A verified snapshot that can replace old log entries.
 pub struct Snapshot {
    // TODO: snapshot data, watermark, verification
 }
@@ -0,0 +1,35 @@
 //! The main KV store
 use std::collections::HashMap;
 /// A log-based Key-Value store.
 ///
 /// The store is a dynamic "view" generated by replaying entries.
 pub struct Store {
    data: HashMap<Vec<u8>, Vec<u8>>,
 }
 impl Store {
    /// Create a new empty store.
    pub fn new() -> Self {
        Self {
            data: HashMap::new(),
        }
    }
    /// Get a value by key.
    pub fn get(&self, key: &[u8]) -> Option<&[u8]> {
        self.data.get(key).map(|v| v.as_slice())
    }
    /// Set a value (this would normally go through the log).
    pub fn set(&mut self, key: Vec<u8>, value: Vec<u8>) {
        self.data.insert(key, value);
    }
 }
 impl Default for Store {
    fn default() -> Self {
        Self::new()
    }
 }
@@ -0,0 +1,9 @@
 //! Watermarks for agreeing on safe compaction points
 /// A watermark representing a safe cut-off point for log compaction.
 ///
 /// Nodes use watermarks to agree on which entries can be safely
 /// deleted and replaced with snapshots.
 pub struct Watermark {
    // TODO: watermark data
 }
@@ -0,0 +1,73 @@
 syntax = "proto3";
 package lattice;
 // 1. The Wrapper (What flies over the wire)
 message SignedEntry {
  // The serialized bytes of the 'Entry' message.
  // We keep this as raw bytes so the signature verification is stable.
  bytes entry_bytes = 1;
  // Ed25519 Signature of 'entry_bytes'
  bytes signature = 2;
  // Public Key of the author (32 bytes)
  bytes author_id = 3;
 }
 // 2. The Log Entry (The Atomic Unit)
 message Entry {
  // Versioning allows us to change the format radically later if needed
  uint32 version = 1;
  // Ordering Metadata
  bytes prev_hash = 2;   // Link to previous entry (32 bytes)
  uint64 seq = 3;        // Monotonic sequence number
  HLC timestamp = 4;     // Hybrid Logical Clock
  // The Batch of Operations
  repeated Operation ops = 5;
 }
 // Hybrid Logical Clock
 message HLC {
  uint64 wall_time = 1;  // Unix timestamp (ms)
  uint32 counter = 2;    // Logical counter for same-ms ordering
 }
 // 3. The Operation (The Change)
 message Operation {
  // "oneof" is how Protobuf handles Rust Enums
  oneof op_type {
    PutOp put = 1;
    DeleteOp delete = 2;
    // Future: MergeOp merge = 3;
  }
 }
 message PutOp {
  string key = 1;
  bytes value = 2; // Raw bytes allows storing images, JSON, binary, etc.
 }
 message DeleteOp {
  string key = 1;
 }
 // 4. The Sync Handshake (Vector Clocks)
 message SyncState {
  repeated Frontier frontiers = 1;
  HLC sender_hlc = 2;  // Sender's current clock (for peer time awareness)
 }
 message Frontier {
  bytes author_id = 1;  // Ed25519 public key (32 bytes)
  uint64 max_seq = 2;   // Highest sequence number seen from this author
  bytes last_hash = 3;  // Hash of the last entry (for chain verification)
 }
 // 5. Log File Record (wrapper for storage)
 message LogRecord {
  bytes hash = 1;         // BLAKE3 hash of entry_bytes (32 bytes)
  bytes entry_bytes = 2;  // Serialized SignedEntry
 }
Author	SHA1	Message	Date
nils	a699c0b1a6	feat: Implement SigChain for append-only signed log management	2025-12-20 14:29:08 +01:00
nils	b16f9a0b39	architecture: future ideas for rich crtds	2025-12-20 14:29:08 +01:00
nils	b64afa063f	feat: Introduce append-only log file I/O for SignedEntry messages using a new LogRecord protobuf, including read/write utilities and documentation.	2025-12-20 14:29:08 +01:00
nils	f153cba267	feat: Introduce signed, verifiable entries with protobuf definitions, cryptographic signing, and hashing, updating the data model and architecture.	2025-12-20 14:29:08 +01:00
nils	2f48483d0e	feat: Add `Clock` abstraction for testable time and implement Hybrid Logical Clock (HLC).	2025-12-20 14:29:08 +01:00
nils	e1405bb910	feat: Introduce Hybrid Logical Clocks in protobuf, update architecture documentation, and add HLC test cases.	2025-12-20 14:29:03 +01:00
		`@@ -0,0 +1,3 @@`
							`//! Gossip protocol for broadcasting changes`

							`// TODO: Implement gossip using iroh-gossip`
		`@@ -0,0 +1,3 @@`
							`//! Unicast communication for direct peer-to-peer messaging`

							`// TODO: Implement unicast using iroh`