feat: implement interactive CLI with key-value operations, add development journal, and update roadmap for DAG conflict resolution
This commit is contained in:
@@ -4,6 +4,7 @@ members = [
|
|||||||
"lattice-core",
|
"lattice-core",
|
||||||
"lattice-net",
|
"lattice-net",
|
||||||
"lattice-store",
|
"lattice-store",
|
||||||
|
"lattice-cli",
|
||||||
]
|
]
|
||||||
|
|
||||||
[workspace.package]
|
[workspace.package]
|
||||||
@@ -16,6 +17,10 @@ license = "MIT"
|
|||||||
lattice-core = { path = "lattice-core" }
|
lattice-core = { path = "lattice-core" }
|
||||||
lattice-net = { path = "lattice-net" }
|
lattice-net = { path = "lattice-net" }
|
||||||
lattice-store = { path = "lattice-store" }
|
lattice-store = { path = "lattice-store" }
|
||||||
|
lattice-cli = { path = "lattice-cli" }
|
||||||
|
|
||||||
|
# CLI
|
||||||
|
rustyline = "17"
|
||||||
|
|
||||||
# Networking (Iroh)
|
# Networking (Iroh)
|
||||||
iroh = "0.95"
|
iroh = "0.95"
|
||||||
|
|||||||
+24
-5
@@ -12,7 +12,7 @@
|
|||||||
- [x] Log file I/O (append, read, hash verification)
|
- [x] Log file I/O (append, read, hash verification)
|
||||||
- [x] SigChain (validate entries before appending)
|
- [x] SigChain (validate entries before appending)
|
||||||
- [x] Store (redb) — `kv` + `meta` tables, log replay
|
- [x] Store (redb) — `kv` + `meta` tables, log replay
|
||||||
- [ ] Interactive CLI: `init`, `put`, `get`, `delete`, `status`, `quit`
|
- [x] Interactive CLI: `init`, `put`, `get`, `delete`, `status`, `quit`
|
||||||
|
|
||||||
### Success Criteria
|
### Success Criteria
|
||||||
|
|
||||||
@@ -30,20 +30,39 @@ Current code assumes single store. Changes needed:
|
|||||||
- [ ] Log paths → `stores/{uuid}/logs/{author}.log`
|
- [ ] Log paths → `stores/{uuid}/logs/{author}.log`
|
||||||
- [ ] Add global meta.db for stores table
|
- [ ] Add global meta.db for stores table
|
||||||
- [ ] Proto: SignedEntry/messages need store_id (UUID)
|
- [ ] Proto: SignedEntry/messages need store_id (UUID)
|
||||||
- [ ] Proto: Entry needs `parent_hashes` for DAG (not just `prev_hash`)
|
|
||||||
- [ ] Store: keys/values → `Vec<u8>` (binary, not String)
|
|
||||||
- [ ] Store: KV table value → `Vec<HeadInfo>` for DAG heads
|
|
||||||
- [ ] CLI → `create-store`, `list-stores`, `use <store>`
|
- [ ] CLI → `create-store`, `list-stores`, `use <store>`
|
||||||
|
|
||||||
---
|
---
|
||||||
|
|
||||||
|
## Milestone 1.5: DAG Conflict Resolution ← NEXT
|
||||||
|
|
||||||
|
**Goal:** Upgrade store from simple LWW to DAG-based conflict resolution per architecture.md.
|
||||||
|
|
||||||
|
### Deliverables
|
||||||
|
|
||||||
|
- [ ] Proto: Add `repeated bytes parent_hashes` to Entry (for DAG causality, separate from sigchain `prev_hash`)
|
||||||
|
- [ ] Store: keys → `Vec<u8>` (binary, not String)
|
||||||
|
- [ ] Store: KV table schema → `Vec<u8> → Vec<HeadInfo>` where `HeadInfo = { value, hlc, author, hash }`
|
||||||
|
- [ ] Store: `applied_frontiers` table → `author_id → (seq, hash)` per author
|
||||||
|
- [ ] Store: `apply_entry` → track multiple heads, merge parent tips into new tip
|
||||||
|
- [ ] Store: `get` → deterministic winner from heads (highest HLC, author_id tiebreaker)
|
||||||
|
- [ ] EntryBuilder: `.parent_hashes(...)` method for DAG ancestry
|
||||||
|
|
||||||
|
### Success Criteria
|
||||||
|
|
||||||
|
- Concurrent writes to same key create multiple heads
|
||||||
|
- Reads return deterministic winner
|
||||||
|
- Next write citing both heads merges fork to single tip
|
||||||
|
- All existing tests still pass
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
## Milestone 2: Two-Node Sync
|
## Milestone 2: Two-Node Sync
|
||||||
|
|
||||||
**Goal:** Two nodes can sync their logs over the network.
|
**Goal:** Two nodes can sync their logs over the network.
|
||||||
|
|
||||||
### Deliverables
|
### Deliverables
|
||||||
|
|
||||||
- [ ] Store: add `applied_frontiers` table (sync state per author)
|
|
||||||
- [ ] VectorClock module (diff, merge, missing entries)
|
- [ ] VectorClock module (diff, merge, missing entries)
|
||||||
- [ ] Sync protocol (push missing entries)
|
- [ ] Sync protocol (push missing entries)
|
||||||
- [ ] Iroh integration (peer discovery, connection)
|
- [ ] Iroh integration (peer discovery, connection)
|
||||||
|
|||||||
@@ -0,0 +1,17 @@
|
|||||||
|
[package]
|
||||||
|
name = "lattice-cli"
|
||||||
|
description = "Interactive CLI for Lattice"
|
||||||
|
version.workspace = true
|
||||||
|
edition.workspace = true
|
||||||
|
license.workspace = true
|
||||||
|
|
||||||
|
[[bin]]
|
||||||
|
name = "lattice"
|
||||||
|
path = "src/main.rs"
|
||||||
|
|
||||||
|
[dependencies]
|
||||||
|
lattice-core = { workspace = true }
|
||||||
|
rustyline = { workspace = true }
|
||||||
|
hex = { workspace = true }
|
||||||
|
thiserror = { workspace = true }
|
||||||
|
shlex = "1"
|
||||||
@@ -0,0 +1,152 @@
|
|||||||
|
//! CLI command handlers (presentation layer)
|
||||||
|
|
||||||
|
use crate::node::LatticeNode;
|
||||||
|
use std::time::Instant;
|
||||||
|
|
||||||
|
/// Command handler function type
|
||||||
|
pub type Handler = fn(&mut LatticeNode, &[String]);
|
||||||
|
|
||||||
|
/// Command definition
|
||||||
|
pub struct Command {
|
||||||
|
pub name: &'static str,
|
||||||
|
pub args: &'static str,
|
||||||
|
pub description: &'static str,
|
||||||
|
pub min_args: usize,
|
||||||
|
pub max_args: usize,
|
||||||
|
pub handler: Handler,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Build the command registry
|
||||||
|
pub fn commands() -> Vec<Command> {
|
||||||
|
vec![
|
||||||
|
Command {
|
||||||
|
name: "put",
|
||||||
|
args: "<key> <value>",
|
||||||
|
description: "Store a key-value pair",
|
||||||
|
min_args: 2,
|
||||||
|
max_args: 2,
|
||||||
|
handler: cmd_put,
|
||||||
|
},
|
||||||
|
Command {
|
||||||
|
name: "get",
|
||||||
|
args: "<key>",
|
||||||
|
description: "Retrieve a value by key",
|
||||||
|
min_args: 1,
|
||||||
|
max_args: 1,
|
||||||
|
handler: cmd_get,
|
||||||
|
},
|
||||||
|
Command {
|
||||||
|
name: "delete",
|
||||||
|
args: "<key>",
|
||||||
|
description: "Delete a key",
|
||||||
|
min_args: 1,
|
||||||
|
max_args: 1,
|
||||||
|
handler: cmd_delete,
|
||||||
|
},
|
||||||
|
Command {
|
||||||
|
name: "list",
|
||||||
|
args: "[-v]",
|
||||||
|
description: "List all key-value pairs (-v for verbose)",
|
||||||
|
min_args: 0,
|
||||||
|
max_args: 1,
|
||||||
|
handler: cmd_list,
|
||||||
|
},
|
||||||
|
Command {
|
||||||
|
name: "status",
|
||||||
|
args: "",
|
||||||
|
description: "Show node statistics",
|
||||||
|
min_args: 0,
|
||||||
|
max_args: 0,
|
||||||
|
handler: cmd_status,
|
||||||
|
},
|
||||||
|
Command {
|
||||||
|
name: "help",
|
||||||
|
args: "",
|
||||||
|
description: "Show this help message",
|
||||||
|
min_args: 0,
|
||||||
|
max_args: 0,
|
||||||
|
handler: cmd_help,
|
||||||
|
},
|
||||||
|
]
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Print help from the command registry
|
||||||
|
fn cmd_help(_node: &mut LatticeNode, _args: &[String]) {
|
||||||
|
println!("\nLattice Commands:");
|
||||||
|
for cmd in commands() {
|
||||||
|
if cmd.args.is_empty() {
|
||||||
|
println!(" {:<18} {}", cmd.name, cmd.description);
|
||||||
|
} else {
|
||||||
|
println!(" {} {:<10} {}", cmd.name, cmd.args, cmd.description);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
println!(" quit Exit the CLI");
|
||||||
|
println!("\nTip: Use quotes for values with spaces: put \"my key\" \"hello world\"\n");
|
||||||
|
}
|
||||||
|
|
||||||
|
fn cmd_status(node: &mut LatticeNode, _args: &[String]) {
|
||||||
|
let status = node.status();
|
||||||
|
println!("--- Node Status ---");
|
||||||
|
println!("Node ID: {}", status.node_id);
|
||||||
|
println!("Data Dir: {}", status.data_dir);
|
||||||
|
println!("Log Sequence: {}", status.log_seq);
|
||||||
|
println!("Applied Entries: {}", status.applied_seq);
|
||||||
|
println!("-------------------");
|
||||||
|
}
|
||||||
|
|
||||||
|
fn cmd_put(node: &mut LatticeNode, args: &[String]) {
|
||||||
|
let start = Instant::now();
|
||||||
|
match node.put(&args[0], args[1].as_bytes()) {
|
||||||
|
Ok(seq) => println!("OK (seq: {}, time: {:.2?})", seq, start.elapsed()),
|
||||||
|
Err(e) => eprintln!("Error: {}", e),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn cmd_get(node: &mut LatticeNode, args: &[String]) {
|
||||||
|
let start = Instant::now();
|
||||||
|
match node.get(&args[0]) {
|
||||||
|
Ok(Some(value)) => {
|
||||||
|
println!("{}", format_value(&value));
|
||||||
|
println!("({:.2?})", start.elapsed());
|
||||||
|
}
|
||||||
|
Ok(None) => println!("(nil)"),
|
||||||
|
Err(e) => eprintln!("Error: {}", e),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn cmd_delete(node: &mut LatticeNode, args: &[String]) {
|
||||||
|
let start = Instant::now();
|
||||||
|
match node.delete(&args[0]) {
|
||||||
|
Ok(seq) => println!("OK (seq: {}, time: {:.2?})", seq, start.elapsed()),
|
||||||
|
Err(e) => eprintln!("Error: {}", e),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn cmd_list(node: &mut LatticeNode, args: &[String]) {
|
||||||
|
let verbose = args.first().map(|a| a == "-v").unwrap_or(false);
|
||||||
|
let start = Instant::now();
|
||||||
|
match node.list() {
|
||||||
|
Ok(entries) => {
|
||||||
|
if entries.is_empty() {
|
||||||
|
println!("(empty)");
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
for (key, value) in &entries {
|
||||||
|
if verbose {
|
||||||
|
println!("{} = {} ({} bytes)", key, format_value(value), value.len());
|
||||||
|
} else {
|
||||||
|
println!("{} = {}", key, format_value(value));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
println!("({} keys, {:.2?})", entries.len(), start.elapsed());
|
||||||
|
}
|
||||||
|
Err(e) => eprintln!("Error: {}", e),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn format_value(value: &[u8]) -> String {
|
||||||
|
match std::str::from_utf8(value) {
|
||||||
|
Ok(s) => s.to_string(),
|
||||||
|
Err(_) => format!("0x{}", hex::encode(value)),
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,91 @@
|
|||||||
|
//! Lattice Interactive CLI
|
||||||
|
|
||||||
|
mod node;
|
||||||
|
mod commands;
|
||||||
|
|
||||||
|
use node::LatticeNodeBuilder;
|
||||||
|
use rustyline::error::ReadlineError;
|
||||||
|
use rustyline::DefaultEditor;
|
||||||
|
|
||||||
|
fn main() {
|
||||||
|
println!("Lattice CLI v{}", env!("CARGO_PKG_VERSION"));
|
||||||
|
println!("Type 'help' for commands, 'quit' to exit.\n");
|
||||||
|
|
||||||
|
let mut node = match LatticeNodeBuilder::new().build() {
|
||||||
|
Ok((n, info)) => {
|
||||||
|
println!("Node ID: {}", info.node_id);
|
||||||
|
println!("Data: {}", info.data_path);
|
||||||
|
if info.is_new {
|
||||||
|
println!("Status: New identity created");
|
||||||
|
} else if info.entries_replayed > 0 {
|
||||||
|
println!("Replay: {} log entries applied", info.entries_replayed);
|
||||||
|
}
|
||||||
|
println!();
|
||||||
|
n
|
||||||
|
}
|
||||||
|
Err(e) => {
|
||||||
|
eprintln!("Failed to initialize node: {}", e);
|
||||||
|
eprintln!("Hint: If data is corrupted, remove the data directory and restart.");
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
let mut rl = DefaultEditor::new().expect("Failed to create editor");
|
||||||
|
let cmds = commands::commands();
|
||||||
|
|
||||||
|
loop {
|
||||||
|
match rl.readline("lattice> ") {
|
||||||
|
Ok(line) => {
|
||||||
|
let line = line.trim();
|
||||||
|
if line.is_empty() {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
let _ = rl.add_history_entry(line);
|
||||||
|
|
||||||
|
let args = match shlex::split(line) {
|
||||||
|
Some(a) => a,
|
||||||
|
None => {
|
||||||
|
println!("Error: mismatched quotes");
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
|
||||||
|
let cmd_name = match args.first() {
|
||||||
|
Some(c) => c.as_str(),
|
||||||
|
None => continue,
|
||||||
|
};
|
||||||
|
|
||||||
|
// Handle quit specially
|
||||||
|
if cmd_name == "quit" || cmd_name == "exit" {
|
||||||
|
println!("Goodbye!");
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
|
||||||
|
// Look up command in registry
|
||||||
|
match cmds.iter().find(|c| c.name == cmd_name) {
|
||||||
|
Some(cmd) => {
|
||||||
|
let cmd_args = &args[1..];
|
||||||
|
if cmd_args.len() < cmd.min_args || cmd_args.len() > cmd.max_args {
|
||||||
|
if cmd.min_args == cmd.max_args {
|
||||||
|
println!("Usage: {} {}", cmd.name, cmd.args);
|
||||||
|
} else {
|
||||||
|
println!("Usage: {} {} (got {} args)", cmd.name, cmd.args, cmd_args.len());
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
(cmd.handler)(&mut node, cmd_args);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
None => println!("Unknown command: '{}'. Type 'help' for commands.", cmd_name),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Err(ReadlineError::Interrupted | ReadlineError::Eof) => {
|
||||||
|
println!("Goodbye!");
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
Err(err) => {
|
||||||
|
eprintln!("Error: {:?}", err);
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,269 @@
|
|||||||
|
//! Lattice Node API
|
||||||
|
//!
|
||||||
|
//! A programmatic interface to a local Lattice node.
|
||||||
|
|
||||||
|
use lattice_core::{
|
||||||
|
DataDir, EntryBuilder, Node, SigChain, Store,
|
||||||
|
hlc::HLC,
|
||||||
|
log::LogError,
|
||||||
|
sigchain::SigChainError,
|
||||||
|
store::StoreError,
|
||||||
|
};
|
||||||
|
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] std::io::Error),
|
||||||
|
|
||||||
|
#[error("Store error: {0}")]
|
||||||
|
Store(#[from] StoreError),
|
||||||
|
|
||||||
|
#[error("SigChain error: {0}")]
|
||||||
|
SigChain(#[from] SigChainError),
|
||||||
|
|
||||||
|
#[error("Log error: {0}")]
|
||||||
|
Log(#[from] LogError),
|
||||||
|
|
||||||
|
#[error("Node error: {0}")]
|
||||||
|
Node(#[from] lattice_core::node::NodeError),
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Info returned when building a node
|
||||||
|
pub struct NodeInfo {
|
||||||
|
pub node_id: String,
|
||||||
|
pub data_path: String,
|
||||||
|
pub is_new: bool,
|
||||||
|
pub entries_replayed: u64,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Status information about the node
|
||||||
|
pub struct NodeStatus {
|
||||||
|
pub node_id: String,
|
||||||
|
pub data_dir: String,
|
||||||
|
pub log_seq: u64,
|
||||||
|
pub applied_seq: u64,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Builder for creating a fully initialized LatticeNode
|
||||||
|
pub struct LatticeNodeBuilder {
|
||||||
|
data_dir: DataDir,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl LatticeNodeBuilder {
|
||||||
|
/// Create a builder with the default data directory
|
||||||
|
pub fn new() -> Self {
|
||||||
|
Self {
|
||||||
|
data_dir: DataDir::default(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Build and initialize the node
|
||||||
|
pub fn build(self) -> Result<(LatticeNode, NodeInfo), NodeError> {
|
||||||
|
// Create directories
|
||||||
|
self.data_dir.ensure_dirs()?;
|
||||||
|
|
||||||
|
// Load or create node identity
|
||||||
|
let key_path = self.data_dir.identity_key();
|
||||||
|
let is_new = !key_path.exists();
|
||||||
|
let node = if key_path.exists() {
|
||||||
|
Node::load(&key_path)?
|
||||||
|
} else {
|
||||||
|
let node = Node::generate();
|
||||||
|
node.save(&key_path)?;
|
||||||
|
node
|
||||||
|
};
|
||||||
|
|
||||||
|
let author_id_hex = hex::encode(node.public_key_bytes());
|
||||||
|
|
||||||
|
// Load or create sigchain
|
||||||
|
let log_path = self.data_dir.log_file(&author_id_hex);
|
||||||
|
let sigchain = if log_path.exists() {
|
||||||
|
SigChain::from_log(&log_path, node.public_key_bytes())?
|
||||||
|
} else {
|
||||||
|
SigChain::new(&log_path, node.public_key_bytes())
|
||||||
|
};
|
||||||
|
|
||||||
|
// Open store and replay log
|
||||||
|
let store = Store::open(self.data_dir.state_db())?;
|
||||||
|
let entries_replayed = if log_path.exists() {
|
||||||
|
store.replay_log(&log_path)?
|
||||||
|
} else {
|
||||||
|
0
|
||||||
|
};
|
||||||
|
|
||||||
|
let info = NodeInfo {
|
||||||
|
node_id: hex::encode(node.public_key_bytes()),
|
||||||
|
data_path: self.data_dir.base().display().to_string(),
|
||||||
|
is_new,
|
||||||
|
entries_replayed,
|
||||||
|
};
|
||||||
|
|
||||||
|
Ok((LatticeNode {
|
||||||
|
data_dir: self.data_dir,
|
||||||
|
node,
|
||||||
|
sigchain,
|
||||||
|
store,
|
||||||
|
}, info))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Default for LatticeNodeBuilder {
|
||||||
|
fn default() -> Self {
|
||||||
|
Self::new()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A fully initialized Lattice node
|
||||||
|
///
|
||||||
|
/// Use `LatticeNodeBuilder` to create an instance.
|
||||||
|
pub struct LatticeNode {
|
||||||
|
data_dir: DataDir,
|
||||||
|
node: Node,
|
||||||
|
sigchain: SigChain,
|
||||||
|
store: Store,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl LatticeNode {
|
||||||
|
/// Get the node's public key as hex
|
||||||
|
pub fn node_id(&self) -> String {
|
||||||
|
hex::encode(self.node.public_key_bytes())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Get the path to the data directory
|
||||||
|
pub fn data_path(&self) -> &Path {
|
||||||
|
self.data_dir.base()
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Get the current status of the node
|
||||||
|
pub fn status(&self) -> NodeStatus {
|
||||||
|
NodeStatus {
|
||||||
|
node_id: self.node_id(),
|
||||||
|
data_dir: self.data_dir.base().display().to_string(),
|
||||||
|
log_seq: self.sigchain.len(),
|
||||||
|
applied_seq: self.store.last_seq().unwrap_or(0),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Put a key-value pair
|
||||||
|
pub fn put(&mut self, key: &str, value: &[u8]) -> Result<u64, NodeError> {
|
||||||
|
let entry = EntryBuilder::new(self.sigchain.next_seq(), HLC::now())
|
||||||
|
.prev_hash(self.sigchain.last_hash().to_vec())
|
||||||
|
.put(key, value.to_vec())
|
||||||
|
.sign(&self.node);
|
||||||
|
|
||||||
|
self.commit_entry(entry)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Get a value by key
|
||||||
|
pub fn get(&self, key: &str) -> Result<Option<Vec<u8>>, NodeError> {
|
||||||
|
Ok(self.store.get(key)?)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// List all key-value pairs
|
||||||
|
pub fn list(&self) -> Result<Vec<(String, Vec<u8>)>, NodeError> {
|
||||||
|
Ok(self.store.list_all()?)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Delete a key
|
||||||
|
pub fn delete(&mut self, key: &str) -> Result<u64, NodeError> {
|
||||||
|
let entry = EntryBuilder::new(self.sigchain.next_seq(), HLC::now())
|
||||||
|
.prev_hash(self.sigchain.last_hash().to_vec())
|
||||||
|
.delete(key)
|
||||||
|
.sign(&self.node);
|
||||||
|
|
||||||
|
self.commit_entry(entry)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Commit a signed entry: append to log via sigchain, then apply to store
|
||||||
|
fn commit_entry(&mut self, entry: lattice_core::proto::SignedEntry) -> Result<u64, NodeError> {
|
||||||
|
self.sigchain.append(&entry)?;
|
||||||
|
self.store.apply_entry(&entry)?;
|
||||||
|
|
||||||
|
Ok(self.sigchain.len())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
use std::env::temp_dir;
|
||||||
|
|
||||||
|
fn temp_data_dir(name: &str) -> DataDir {
|
||||||
|
let path = temp_dir().join(format!("lattice_node_test_{}", name));
|
||||||
|
// Clean up from previous runs
|
||||||
|
let _ = std::fs::remove_dir_all(&path);
|
||||||
|
DataDir::new(path)
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_put_survives_restart() {
|
||||||
|
let data_dir = temp_data_dir("restart");
|
||||||
|
|
||||||
|
// First session: put a value
|
||||||
|
{
|
||||||
|
let (mut node, _) = LatticeNodeBuilder { data_dir: data_dir.clone() }
|
||||||
|
.build()
|
||||||
|
.expect("Failed to create node");
|
||||||
|
|
||||||
|
node.put("/test/key", b"hello").expect("put failed");
|
||||||
|
assert_eq!(node.get("/test/key").unwrap(), Some(b"hello".to_vec()));
|
||||||
|
}
|
||||||
|
|
||||||
|
// Second session: value should still be there
|
||||||
|
{
|
||||||
|
let (node, _) = LatticeNodeBuilder { data_dir: data_dir.clone() }
|
||||||
|
.build()
|
||||||
|
.expect("Failed to create node on restart");
|
||||||
|
|
||||||
|
assert_eq!(node.get("/test/key").unwrap(), Some(b"hello".to_vec()));
|
||||||
|
assert_eq!(node.status().log_seq, 1);
|
||||||
|
}
|
||||||
|
|
||||||
|
// Cleanup
|
||||||
|
let _ = std::fs::remove_dir_all(data_dir.base());
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_log_replay_after_db_deletion() {
|
||||||
|
let data_dir = temp_data_dir("replay");
|
||||||
|
|
||||||
|
// First session: put some values
|
||||||
|
{
|
||||||
|
let (mut node, _) = LatticeNodeBuilder { data_dir: data_dir.clone() }
|
||||||
|
.build()
|
||||||
|
.expect("Failed to create node");
|
||||||
|
|
||||||
|
node.put("/key1", b"value1").expect("put failed");
|
||||||
|
node.put("/key2", b"value2").expect("put failed");
|
||||||
|
node.delete("/key1").expect("delete failed");
|
||||||
|
}
|
||||||
|
|
||||||
|
// Delete state.db but keep the log
|
||||||
|
let db_path = data_dir.state_db();
|
||||||
|
std::fs::remove_file(&db_path).expect("Failed to delete state.db");
|
||||||
|
assert!(!db_path.exists(), "state.db should be deleted");
|
||||||
|
|
||||||
|
// Third session: log should be replayed to reconstruct state
|
||||||
|
{
|
||||||
|
let (node, info) = LatticeNodeBuilder { data_dir: data_dir.clone() }
|
||||||
|
.build()
|
||||||
|
.expect("Failed to rebuild node from log");
|
||||||
|
|
||||||
|
// Should have replayed 3 entries
|
||||||
|
assert_eq!(info.entries_replayed, 3);
|
||||||
|
// key1 was deleted
|
||||||
|
assert_eq!(node.get("/key1").unwrap(), None);
|
||||||
|
// key2 should still exist
|
||||||
|
assert_eq!(node.get("/key2").unwrap(), Some(b"value2".to_vec()));
|
||||||
|
// log seq should be 3 (put, put, delete)
|
||||||
|
assert_eq!(node.status().log_seq, 3);
|
||||||
|
}
|
||||||
|
|
||||||
|
// Cleanup
|
||||||
|
let _ = std::fs::remove_dir_all(data_dir.base());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
+77
-71
@@ -66,50 +66,67 @@ impl Store {
|
|||||||
Ok(Self { db })
|
Ok(Self { db })
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Replay a log file and apply all entries to the store
|
/// Replay a log file and apply all entries to the store (batched)
|
||||||
pub fn replay_log(&self, log_path: impl AsRef<Path>) -> Result<u64, StoreError> {
|
pub fn replay_log(&self, log_path: impl AsRef<Path>) -> Result<u64, StoreError> {
|
||||||
let entries = read_entries(log_path)?;
|
let entries = read_entries(log_path)?;
|
||||||
let mut count = 0u64;
|
if entries.is_empty() {
|
||||||
|
return Ok(0);
|
||||||
for signed_entry in entries {
|
|
||||||
self.apply_entry(&signed_entry)?;
|
|
||||||
count += 1;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
Ok(count)
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Apply a single signed entry to the store
|
|
||||||
pub fn apply_entry(&self, signed_entry: &SignedEntry) -> Result<(), StoreError> {
|
|
||||||
let entry = Entry::decode(&signed_entry.entry_bytes[..])?;
|
|
||||||
|
|
||||||
let write_txn = self.db.begin_write()?;
|
let write_txn = self.db.begin_write()?;
|
||||||
{
|
{
|
||||||
let mut kv_table = write_txn.open_table(KV_TABLE)?;
|
let mut kv_table = write_txn.open_table(KV_TABLE)?;
|
||||||
let mut meta_table = write_txn.open_table(META_TABLE)?;
|
let mut meta_table = write_txn.open_table(META_TABLE)?;
|
||||||
|
|
||||||
// Apply operations
|
for signed_entry in &entries {
|
||||||
for op in entry.ops {
|
Self::apply_ops_to_tables(signed_entry, &mut kv_table, &mut meta_table)?;
|
||||||
if let Some(op_type) = op.op_type {
|
}
|
||||||
match op_type {
|
}
|
||||||
operation::OpType::Put(put) => {
|
write_txn.commit()?;
|
||||||
kv_table.insert(put.key.as_str(), put.value.as_slice())?;
|
|
||||||
}
|
Ok(entries.len() as u64)
|
||||||
operation::OpType::Delete(del) => {
|
}
|
||||||
kv_table.remove(del.key.as_str())?;
|
|
||||||
}
|
/// Apply a single signed entry to the store
|
||||||
|
pub fn apply_entry(&self, signed_entry: &SignedEntry) -> Result<(), StoreError> {
|
||||||
|
let write_txn = self.db.begin_write()?;
|
||||||
|
{
|
||||||
|
let mut kv_table = write_txn.open_table(KV_TABLE)?;
|
||||||
|
let mut meta_table = write_txn.open_table(META_TABLE)?;
|
||||||
|
Self::apply_ops_to_tables(signed_entry, &mut kv_table, &mut meta_table)?;
|
||||||
|
}
|
||||||
|
write_txn.commit()?;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Internal: apply operations from a signed entry to tables
|
||||||
|
fn apply_ops_to_tables(
|
||||||
|
signed_entry: &SignedEntry,
|
||||||
|
kv_table: &mut redb::Table<&str, &[u8]>,
|
||||||
|
meta_table: &mut redb::Table<&str, &[u8]>,
|
||||||
|
) -> Result<(), StoreError> {
|
||||||
|
let entry = Entry::decode(&signed_entry.entry_bytes[..])?;
|
||||||
|
|
||||||
|
// Apply operations
|
||||||
|
for op in entry.ops {
|
||||||
|
if let Some(op_type) = op.op_type {
|
||||||
|
match op_type {
|
||||||
|
operation::OpType::Put(put) => {
|
||||||
|
kv_table.insert(put.key.as_str(), put.value.as_slice())?;
|
||||||
|
}
|
||||||
|
operation::OpType::Delete(del) => {
|
||||||
|
kv_table.remove(del.key.as_str())?;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Update meta
|
|
||||||
meta_table.insert(META_LAST_SEQ, &entry.seq.to_le_bytes()[..])?;
|
|
||||||
|
|
||||||
// Compute and store hash of this entry
|
|
||||||
let hash: [u8; 32] = blake3::hash(&signed_entry.entry_bytes).into();
|
|
||||||
meta_table.insert(META_LAST_HASH, &hash[..])?;
|
|
||||||
}
|
}
|
||||||
write_txn.commit()?;
|
|
||||||
|
// Update meta
|
||||||
|
meta_table.insert(META_LAST_SEQ, &entry.seq.to_le_bytes()[..])?;
|
||||||
|
|
||||||
|
// Hash the full SignedEntry (includes signature), not just entry_bytes
|
||||||
|
let hash: [u8; 32] = blake3::hash(&signed_entry.encode_to_vec()).into();
|
||||||
|
meta_table.insert(META_LAST_HASH, &hash[..])?;
|
||||||
|
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
@@ -122,28 +139,17 @@ impl Store {
|
|||||||
Ok(table.get(key)?.map(|v| v.value().to_vec()))
|
Ok(table.get(key)?.map(|v| v.value().to_vec()))
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Put a value (use SigChain.create_entry for proper signing)
|
/// List all key-value pairs
|
||||||
/// This is a low-level method for direct writes
|
pub fn list_all(&self) -> Result<Vec<(String, Vec<u8>)>, StoreError> {
|
||||||
pub fn put(&self, key: &str, value: &[u8]) -> Result<(), StoreError> {
|
let read_txn = self.db.begin_read()?;
|
||||||
let write_txn = self.db.begin_write()?;
|
let table = read_txn.open_table(KV_TABLE)?;
|
||||||
{
|
|
||||||
let mut table = write_txn.open_table(KV_TABLE)?;
|
|
||||||
table.insert(key, value)?;
|
|
||||||
}
|
|
||||||
write_txn.commit()?;
|
|
||||||
Ok(())
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Delete a key
|
let mut result = Vec::new();
|
||||||
pub fn delete(&self, key: &str) -> Result<bool, StoreError> {
|
for entry in table.iter()? {
|
||||||
let write_txn = self.db.begin_write()?;
|
let (key, value) = entry?;
|
||||||
let removed;
|
result.push((key.value().to_string(), value.value().to_vec()));
|
||||||
{
|
|
||||||
let mut table = write_txn.open_table(KV_TABLE)?;
|
|
||||||
removed = table.remove(key)?.is_some();
|
|
||||||
}
|
}
|
||||||
write_txn.commit()?;
|
Ok(result)
|
||||||
Ok(removed)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Get the last applied sequence number
|
/// Get the last applied sequence number
|
||||||
@@ -222,33 +228,33 @@ mod tests {
|
|||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_put_get() {
|
fn test_apply_entry() {
|
||||||
let (db_path, _) = temp_paths("put_get");
|
let (db_path, _) = temp_paths("apply_entry");
|
||||||
std::fs::remove_file(&db_path).ok();
|
std::fs::remove_file(&db_path).ok();
|
||||||
|
|
||||||
let store = Store::open(&db_path).unwrap();
|
let store = Store::open(&db_path).unwrap();
|
||||||
|
let node = Node::generate();
|
||||||
|
let clock = MockClock::new(1000);
|
||||||
|
|
||||||
store.put("key1", b"value1").unwrap();
|
// Create and apply a put entry
|
||||||
store.put("key2", b"value2").unwrap();
|
let entry1 = EntryBuilder::new(1, HLC::now_with_clock(&clock))
|
||||||
|
.prev_hash([0u8; 32].to_vec())
|
||||||
|
.put("/key1", b"value1".to_vec())
|
||||||
|
.sign(&node);
|
||||||
|
store.apply_entry(&entry1).unwrap();
|
||||||
|
|
||||||
assert_eq!(store.get("key1").unwrap(), Some(b"value1".to_vec()));
|
assert_eq!(store.get("/key1").unwrap(), Some(b"value1".to_vec()));
|
||||||
assert_eq!(store.get("key2").unwrap(), Some(b"value2".to_vec()));
|
assert_eq!(store.last_seq().unwrap(), 1);
|
||||||
assert_eq!(store.get("key3").unwrap(), None);
|
|
||||||
|
|
||||||
std::fs::remove_file(&db_path).ok();
|
// Create and apply a delete entry
|
||||||
}
|
let entry2 = EntryBuilder::new(2, HLC::now_with_clock(&clock))
|
||||||
|
.prev_hash([0u8; 32].to_vec()) // simplified for test
|
||||||
|
.delete("/key1")
|
||||||
|
.sign(&node);
|
||||||
|
store.apply_entry(&entry2).unwrap();
|
||||||
|
|
||||||
#[test]
|
assert_eq!(store.get("/key1").unwrap(), None);
|
||||||
fn test_delete() {
|
assert_eq!(store.last_seq().unwrap(), 2);
|
||||||
let (db_path, _) = temp_paths("delete");
|
|
||||||
std::fs::remove_file(&db_path).ok();
|
|
||||||
|
|
||||||
let store = Store::open(&db_path).unwrap();
|
|
||||||
|
|
||||||
store.put("key", b"value").unwrap();
|
|
||||||
assert!(store.delete("key").unwrap());
|
|
||||||
assert_eq!(store.get("key").unwrap(), None);
|
|
||||||
assert!(!store.delete("key").unwrap()); // Already gone
|
|
||||||
|
|
||||||
std::fs::remove_file(&db_path).ok();
|
std::fs::remove_file(&db_path).ok();
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user