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

feat: Refactor CLI commands into dedicated modules, update roadmap, and simplify sync function signature.

Browse Source
This commit is contained in:
nils
2025-12-22 23:06:36 +01:00
parent 57ecbffaed
commit 76810f8d8e
9 changed files with 798 additions and 782 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lattice-cli/src/commands.rs
+41 -745
View File
@@ -1,24 +1,21 @@
//! CLI command handlers
use lattice_core::{Node, StoreHandle};
use lattice_core::{Uuid, PeerStatus};
use lattice_net::LatticeEndpoint;
use chrono::DateTime;
use std::time::Instant;
/// Result of a command that may switch stores
/// Result of a command that may switch stores or exit
pub enum CommandResult {
/// No store change
Ok,
/// Switch to this store
SwitchTo(StoreHandle),
/// Exit the CLI
Quit,
}
/// Helper to call async code from sync command handlers
fn block_async<F: std::future::Future>(f: F) -> F::Output {
tokio::task::block_in_place(|| {
tokio::runtime::Handle::current().block_on(f)
})
pub fn block_async<F: std::future::Future>(f: F) -> F::Output {
tokio::task::block_in_place(|| tokio::runtime::Handle::current().block_on(f))
}
pub type Handler = fn(&Node, Option<&StoreHandle>, Option<&LatticeEndpoint>, &[String]) -> CommandResult;
@@ -26,758 +23,57 @@ pub type Handler = fn(&Node, Option<&StoreHandle>, Option<&LatticeEndpoint>, &[S
pub struct Command {
pub name: &'static str,
pub args: &'static str,
pub description: &'static str,
pub desc: &'static str,
pub group: &'static str,
pub min_args: usize,
pub max_args: usize,
pub handler: Handler,
}
/// Get all available commands
pub fn commands() -> Vec<Command> {
vec![
Command {
name: "init",
args: "",
description: "Initialize node with root store",
min_args: 0,
max_args: 0,
handler: cmd_init,
},
Command {
name: "create-store",
args: "",
description: "Create a new store",
min_args: 0,
max_args: 0,
handler: cmd_create_store,
},
Command {
name: "use",
args: "<uuid>",
description: "Switch to a store",
min_args: 1,
max_args: 1,
handler: cmd_use_store,
},
Command {
name: "list-stores",
args: "",
description: "List all stores",
min_args: 0,
max_args: 0,
handler: cmd_list_stores,
},
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> [-v]",
description: "Retrieve a value by key",
min_args: 1,
max_args: 2,
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/store info",
min_args: 0,
max_args: 0,
handler: cmd_status,
},
Command {
name: "author-state",
args: "[author-hex]",
description: "Show author state (default: self)",
min_args: 0,
max_args: 1,
handler: cmd_author_state,
},
Command {
name: "invite",
args: "<pubkey-hex>",
description: "Invite a peer node (writes to root store)",
min_args: 1,
max_args: 1,
handler: cmd_invite,
},
Command {
name: "peers",
args: "",
description: "List known peers from root store",
min_args: 0,
max_args: 0,
handler: cmd_peers,
},
Command {
name: "remove",
args: "<pubkey-hex>",
description: "Remove a peer (set status to removed)",
min_args: 1,
max_args: 1,
handler: cmd_remove,
},
Command {
name: "join",
args: "<nodeid-hex>",
description: "Join a mesh by connecting to a peer (requires no local store)",
min_args: 1,
max_args: 1,
handler: cmd_join,
},
Command {
name: "sync",
args: "[nodeid]",
description: "Sync entries with a peer (or all peers if none specified)",
min_args: 0,
max_args: 1,
handler: cmd_sync,
},
Command {
name: "help",
args: "",
description: "Show this help message",
min_args: 0,
max_args: 0,
handler: cmd_help,
},
]
}
// --- Store management ---
fn cmd_init(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult {
match block_async(node.init()) {
Ok((store_id, handle)) => {
println!("Initialized with root store: {}", store_id);
println!("Node info stored in /nodes/{}/*", hex::encode(node.node_id()));
CommandResult::SwitchTo(handle)
}
Err(e) => {
eprintln!("Error: {}", e);
CommandResult::Ok
}
}
}
fn cmd_create_store(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult {
match node.create_store() {
Ok(store_id) => {
println!("Created store: {}", store_id);
match node.open_store(store_id) {
Ok((handle, _)) => {
println!("Switched to new store");
CommandResult::SwitchTo(handle)
}
Err(e) => {
eprintln!("Warning: {}", e);
CommandResult::Ok
}
}
}
Err(e) => {
eprintln!("Error: {}", e);
CommandResult::Ok
}
}
}
fn cmd_use_store(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult {
let store_id = match Uuid::parse_str(&args[0]) {
Ok(id) => id,
Err(_) => {
eprintln!("Error: invalid UUID '{}'", args[0]);
return CommandResult::Ok;
}
};
let mut cmds = Vec::new();
let start = Instant::now();
match node.open_store(store_id) {
Ok((handle, info)) => {
if info.entries_replayed > 0 {
println!("Replayed {} entries ({:.2?})", info.entries_replayed, start.elapsed());
} else {
println!("Switched to store {}", store_id);
}
CommandResult::SwitchTo(handle)
}
Err(e) => {
eprintln!("Error: {}", e);
CommandResult::Ok
}
}
}
fn cmd_list_stores(node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult {
let stores = match node.list_stores() {
Ok(s) => s,
Err(e) => {
eprintln!("Error: {}", e);
return CommandResult::Ok;
}
};
let current_id = store.map(|s| s.id());
// General CLI commands
cmds.push(Command {
name: "help", args: "", desc: "Show this help",
group: "general", min_args: 0, max_args: 0, handler: cmd_help as Handler
});
cmds.push(Command {
name: "quit", args: "", desc: "Exit",
group: "general", min_args: 0, max_args: 0, handler: cmd_quit as Handler
});
if stores.is_empty() {
println!("No stores. Use 'init' or 'create-store'.");
} else {
for store_id in stores {
let marker = if Some(store_id) == current_id { " *" } else { "" };
println!("{}{}", store_id, marker);
}
}
CommandResult::Ok
// Node commands (operations on the node)
cmds.extend(crate::node_commands::node_commands());
// Store commands (raw KV operations)
cmds.extend(crate::store_commands::store_commands());
cmds
}
// --- Info ---
fn cmd_help(_node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult {
println!("\nCommands:");
for cmd in commands() {
if cmd.args.is_empty() {
println!(" {:<16} {}", cmd.name, cmd.description);
} else {
println!(" {} {:<8} {}", cmd.name, cmd.args, cmd.description);
let cmds = commands();
let mut last_group = "";
for cmd in &cmds {
if cmd.group != last_group {
println!();
println!("[{}]", cmd.group);
last_group = cmd.group;
}
let usage = if cmd.args.is_empty() {
cmd.name.to_string()
} else {
format!("{} {}", cmd.name, cmd.args)
};
println!(" {:18} {}", usage, cmd.desc);
}
println!(" quit Exit");
println!();
CommandResult::Ok
}
fn cmd_status(node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult {
println!("Node ID: {}", hex::encode(node.node_id()));
println!("Data: {}", node.data_path().display());
match node.root_store() {
Ok(Some(id)) => println!("Root: {}", id),
Ok(None) => println!("Root: (not set)"),
Err(_) => println!("Root: (error)"),
}
if let Some(h) = store {
println!("Store: {}", h.id());
println!("Log Seq: {}", block_async(h.log_seq()));
println!("Applied: {}", block_async(h.applied_seq()).unwrap_or(0));
// Show sync state summary
if let Ok(sync_state) = block_async(h.sync_state()) {
let authors = sync_state.authors();
let total_entries: u64 = authors.values().map(|a| a.seq).sum();
let num_authors = authors.len();
println!("Authors: {} ({} total entries)", num_authors, total_entries);
// Show per-author details
for (author, info) in authors {
println!(" {}...: seq={}, heads={}",
hex::encode(&author[..6]),
info.seq,
info.heads.len());
}
}
// Show log directory size
let logs_dir = node.data_path().join("stores").join(h.id().to_string()).join("logs");
if logs_dir.exists() {
let mut total_size = 0u64;
let mut file_count = 0;
if let Ok(entries) = std::fs::read_dir(&logs_dir) {
for entry in entries.flatten() {
if let Ok(meta) = entry.metadata() {
if meta.is_file() {
total_size += meta.len();
file_count += 1;
}
}
}
}
println!("Logs: {} files, {} bytes", file_count, total_size);
}
} else {
println!("Store: (none)");
}
CommandResult::Ok
}
// --- KV ---
fn cmd_put(_node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult {
let Some(h) = store else {
println!("No store selected. Use 'init' or 'use <uuid>'");
return CommandResult::Ok;
};
let start = Instant::now();
match block_async(h.put(args[0].as_bytes(), args[1].as_bytes())) {
Ok(seq) => println!("OK (seq: {}, {:.2?})", seq, start.elapsed()),
Err(e) => eprintln!("Error: {}", e),
}
CommandResult::Ok
}
fn cmd_get(_node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult {
let Some(h) = store else {
println!("No store selected. Use 'init' or 'use <uuid>'");
return CommandResult::Ok;
};
let verbose = args.get(1).map(|a| a == "-v").unwrap_or(false);
let start = Instant::now();
let key = args[0].as_bytes();
if verbose {
// Show all heads
match block_async(h.get_heads(key)) {
Ok(heads) if heads.is_empty() => println!("(nil)"),
Ok(heads) => {
for (i, head) in heads.iter().enumerate() {
let winner = if i == 0 { "→" } else { " " };
let tombstone = if head.tombstone { "⊗" } else { "" };
let author_short = hex::encode(&head.author).chars().take(8).collect::<String>();
if head.tombstone {
println!("{} {} (deleted) (hlc:{}, author:{})",
winner, tombstone, head.hlc, author_short);
} else {
println!("{} {} (hlc:{}, author:{})",
winner, format_value(&head.value), head.hlc, author_short);
}
}
if heads.len() > 1 {
println!("⚠ {} heads (conflict)", heads.len());
}
println!("({:.2?})", start.elapsed());
}
Err(e) => eprintln!("Error: {}", e),
}
} else {
match block_async(h.get(key)) {
Ok(Some(v)) => {
let heads = block_async(h.get_heads(key)).unwrap_or_default();
if heads.len() > 1 {
println!("{} (⚠ {} heads)", format_value(&v), heads.len());
} else {
println!("{}", format_value(&v));
}
println!("({:.2?})", start.elapsed());
}
Ok(None) => println!("(nil)"),
Err(e) => eprintln!("Error: {}", e),
}
}
CommandResult::Ok
}
fn cmd_delete(_node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult {
let Some(h) = store else {
println!("No store selected. Use 'init' or 'use <uuid>'");
return CommandResult::Ok;
};
let start = Instant::now();
match block_async(h.delete(args[0].as_bytes())) {
Ok(seq) => println!("OK (seq: {}, {:.2?})", seq, start.elapsed()),
Err(e) => eprintln!("Error: {}", e),
}
CommandResult::Ok
}
fn cmd_list(_node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult {
let Some(h) = store else {
println!("No store selected. Use 'init' or 'use <uuid>'");
return CommandResult::Ok;
};
let verbose = args.first().map(|a| a == "-v").unwrap_or(false);
let start = Instant::now();
match block_async(h.list()) {
Ok(entries) => {
if entries.is_empty() {
println!("(empty)");
} else {
for (k, v) in &entries {
let key_str = format_value(k);
if verbose {
// Show all heads for this key
let heads = block_async(h.get_heads(k)).unwrap_or_default();
println!("{}:", key_str);
for (i, head) in heads.iter().enumerate() {
let winner = if i == 0 { "→" } else { " " };
let author_short = hex::encode(&head.author).chars().take(8).collect::<String>();
if head.tombstone {
println!(" {} ⊗ (deleted) (hlc:{}, author:{})",
winner, head.hlc, author_short);
} else {
println!(" {} {} (hlc:{}, author:{})",
winner, format_value(&head.value), head.hlc, author_short);
}
}
} else {
// Check for multiple heads
let heads = block_async(h.get_heads(k)).unwrap_or_default();
if heads.len() > 1 {
println!("{} = {} (⚠ {} heads)", key_str, format_value(v), heads.len());
} else {
println!("{} = {}", key_str, format_value(v));
}
}
}
println!("({} keys, {:.2?})", entries.len(), start.elapsed());
}
}
Err(e) => eprintln!("Error: {}", e),
}
CommandResult::Ok
}
fn format_value(v: &[u8]) -> String {
std::str::from_utf8(v).map(String::from).unwrap_or_else(|_| format!("0x{}", hex::encode(v)))
}
fn cmd_author_state(node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult {
let store = match store {
Some(s) => s,
None => {
eprintln!("Error: no store selected");
return CommandResult::Ok;
}
};
// Get author: from arg or default to self
let author_bytes: [u8; 32] = if args.is_empty() {
node.node_id()
} else {
let hex_str = args[0].trim_start_matches("0x");
match hex::decode(hex_str) {
Ok(bytes) if bytes.len() == 32 => bytes.try_into().unwrap(),
Ok(bytes) => {
eprintln!("Error: author must be 32 bytes, got {}", bytes.len());
return CommandResult::Ok;
}
Err(e) => {
eprintln!("Error: invalid hex: {}", e);
return CommandResult::Ok;
}
}
};
match block_async(store.author_state(&author_bytes)) {
Ok(Some(state)) => {
println!("Author: {}", hex::encode(&author_bytes));
println!(" seq: {}", state.seq);
println!(" hash: {}", hex::encode(&state.hash));
println!(" log_offset: {}", state.log_offset);
}
Ok(None) => {
println!("No state for author: {}", hex::encode(&author_bytes));
}
Err(e) => eprintln!("Error: {}", e),
}
CommandResult::Ok
}
// --- Peer management ---
fn cmd_invite(node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult {
let store = match store {
Some(s) => s,
None => {
eprintln!("Not in a store. Run 'use' or 'init' first.");
return CommandResult::Ok;
}
};
let pubkey_hex = &args[0];
let _peer_pubkey = match hex::decode(pubkey_hex) {
Ok(bytes) if bytes.len() == 32 => bytes,
_ => {
eprintln!("Invalid pubkey: expected 64 hex chars (32 bytes)");
return CommandResult::Ok;
}
};
// Write peer info as separate keys
let inviter_hex = hex::encode(node.node_id());
let added_at = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
let added_by_key = format!("/nodes/{}/added_by", pubkey_hex);
match block_async(store.put(added_by_key.as_bytes(), inviter_hex.as_bytes())) {
Ok(_) => {}
Err(e) => {
eprintln!("Error writing added_by: {}", e);
return CommandResult::Ok;
}
}
let added_at_key = format!("/nodes/{}/added_at", pubkey_hex);
match block_async(store.put(added_at_key.as_bytes(), added_at.to_string().as_bytes())) {
Ok(_) => {}
Err(e) => {
eprintln!("Error writing added_at: {}", e);
return CommandResult::Ok;
}
}
// Write /nodes/{pubkey}/status = invited (becomes active after sync)
let status_key = format!("/nodes/{}/status", pubkey_hex);
match block_async(store.put(status_key.as_bytes(), PeerStatus::Invited.as_str().as_bytes())) {
Ok(_) => {}
Err(e) => {
eprintln!("Error writing status: {}", e);
return CommandResult::Ok;
}
}
println!("Invited peer: {}", pubkey_hex);
println!(" /nodes/{}/added_by", pubkey_hex);
println!(" /nodes/{}/added_at", pubkey_hex);
println!(" /nodes/{}/status = {} (will become active after sync)", pubkey_hex, PeerStatus::Invited.as_str());
CommandResult::Ok
}
fn cmd_peers(_node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult {
let store = match store {
Some(s) => s,
None => {
eprintln!("Not in a store. Run 'use' or 'init' first.");
return CommandResult::Ok;
}
};
// List all keys under /nodes/
let all = match block_async(store.list()) {
Ok(v) => v,
Err(e) => {
eprintln!("Error listing: {}", e);
return CommandResult::Ok;
}
};
// Collect unique pubkeys with status
let mut peers: std::collections::HashMap<String, PeerStatus> = std::collections::HashMap::new();
for (key, value) in &all {
let key_str = String::from_utf8_lossy(key);
if key_str.ends_with("/status") {
if let Some(pubkey) = key_str.strip_prefix("/nodes/").and_then(|s| s.strip_suffix("/status")) {
let status_str = String::from_utf8_lossy(value);
if let Some(status) = PeerStatus::from_str(&status_str) {
peers.insert(pubkey.to_string(), status);
}
}
}
}
if peers.is_empty() {
println!("No peers found.");
} else {
// Group peers by status
let mut by_status: std::collections::HashMap<PeerStatus, Vec<(String, String, String)>> =
std::collections::HashMap::new();
for (pubkey, status) in &peers {
// Try to get name and added_at from separate keys
let name_key = format!("/nodes/{}/name", pubkey);
let added_at_key = format!("/nodes/{}/added_at", pubkey);
let name = match block_async(store.get(name_key.as_bytes())) {
Ok(Some(bytes)) => String::from_utf8_lossy(&bytes).to_string(),
_ => String::new(),
};
let added = match block_async(store.get(added_at_key.as_bytes())) {
Ok(Some(bytes)) => {
if let Ok(ts) = String::from_utf8_lossy(&bytes).parse::<i64>() {
DateTime::from_timestamp(ts, 0)
.map(|dt| dt.format("%Y-%m-%d").to_string())
.unwrap_or_default()
} else {
String::new()
}
}
_ => String::new(),
};
by_status.entry(*status)
.or_default()
.push((pubkey.clone(), name, added));
}
// Print grouped by status in order: active, invited, removed
let status_order = [PeerStatus::Active, PeerStatus::Invited, PeerStatus::Removed];
for status in &status_order {
if let Some(peer_list) = by_status.get(status) {
println!("\n[{}] ({}):", status.as_str(), peer_list.len());
let mut sorted = peer_list.clone();
sorted.sort_by(|a, b| a.0.cmp(&b.0));
for (pubkey, name, added) in &sorted {
let info_str = match (name.is_empty(), added.is_empty()) {
(false, false) => format!(" {} ({})", name, added),
(false, true) => format!(" {}", name),
(true, false) => format!(" ({})", added),
_ => String::new(),
};
println!(" {}{}", pubkey, info_str);
}
}
}
}
CommandResult::Ok
}
fn cmd_remove(node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult {
let store = match store {
Some(s) => s,
None => {
eprintln!("Not in a store. Run 'use' or 'init' first.");
return CommandResult::Ok;
}
};
let pubkey_hex = &args[0];
// Validate pubkey format (should be 64 hex chars)
if pubkey_hex.len() != 64 || !pubkey_hex.chars().all(|c| c.is_ascii_hexdigit()) {
eprintln!("Invalid pubkey: expected 64 hex characters");
return CommandResult::Ok;
}
// Prevent self-removal
let my_pubkey = hex::encode(node.node_id());
if pubkey_hex == &my_pubkey {
eprintln!("Cannot remove yourself.");
return CommandResult::Ok;
}
// Check if peer exists
let status_key = format!("/nodes/{}/status", pubkey_hex);
match block_async(store.get(status_key.as_bytes())) {
Ok(Some(status)) => {
if status == PeerStatus::Removed.as_str().as_bytes() {
println!("Peer {} is already removed.", &pubkey_hex[..10]);
return CommandResult::Ok;
}
}
Ok(None) => {
eprintln!("Peer {} not found.", &pubkey_hex[..10]);
return CommandResult::Ok;
}
Err(e) => {
eprintln!("Error checking peer: {}", e);
return CommandResult::Ok;
}
}
// Set status to removed
match block_async(store.put(status_key.as_bytes(), PeerStatus::Removed.as_str().as_bytes())) {
Ok(_) => println!("Removed peer: {}...", &pubkey_hex[..10]),
Err(e) => eprintln!("Error removing peer: {}", e),
}
CommandResult::Ok
}
fn cmd_join(node: &Node, store: Option<&StoreHandle>, endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult {
let endpoint = match endpoint {
Some(ep) => ep,
None => {
eprintln!("Iroh endpoint not started.");
return CommandResult::Ok;
}
};
if store.is_some() {
eprintln!("Already initialized. Use 'sync' to sync with peers.");
return CommandResult::Ok;
}
let peer_id = match lattice_net::parse_node_id(&args[0]) {
Ok(id) => id,
Err(e) => {
eprintln!("Invalid node ID: {}", e);
return CommandResult::Ok;
}
};
println!("Joining mesh via {}...", peer_id.fmt_short());
match block_async(lattice_net::join_mesh(node, endpoint, peer_id)) {
Ok(handle) => {
println!("Joined mesh! Use 'sync' command to sync entries.");
CommandResult::SwitchTo(handle)
}
Err(e) => {
eprintln!("Join failed: {}", e);
CommandResult::Ok
}
}
}
fn cmd_sync(node: &Node, store: Option<&StoreHandle>, endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult {
let endpoint = match endpoint {
Some(ep) => ep,
None => {
eprintln!("Iroh endpoint not started.");
return CommandResult::Ok;
}
};
let store = match store {
Some(s) => s,
None => {
eprintln!("No store open. Use 'init' or 'join' first.");
return CommandResult::Ok;
}
};
if args.is_empty() {
// Sync with all active peers
match block_async(lattice_net::sync_all(node, endpoint, store)) {
Ok(results) => {
if results.is_empty() {
println!("No peers to sync with.");
} else {
let total: u64 = results.iter().map(|r| r.entries_applied).sum();
println!("\nSync complete! Applied {} entries from {} peer(s).", total, results.len());
}
}
Err(e) => eprintln!("Sync failed: {}", e),
}
} else {
// Sync with specific peer
let peer_id = match lattice_net::parse_node_id(&args[0]) {
Ok(id) => id,
Err(e) => {
eprintln!("Invalid node ID: {}", e);
return CommandResult::Ok;
}
};
println!("Syncing with {}...", peer_id.fmt_short());
match block_async(lattice_net::sync_with_peer(node, endpoint, store, peer_id)) {
Ok(result) => {
println!("Sync complete! Applied {} entries (peer sent {})",
result.entries_applied, result.entries_sent_by_peer);
}
Err(e) => eprintln!("Sync failed: {}", e),
}
}
CommandResult::Ok
fn cmd_quit(_node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult {
println!("Goodbye!");
CommandResult::Quit
}
lattice-cli/src/main.rs
+10 -10
View File
@@ -1,6 +1,8 @@
//! Lattice Interactive CLI
mod commands;
mod node_commands;
mod store_commands;
use lattice_net::spawn_accept_loop;
use commands::CommandResult;
@@ -52,15 +54,17 @@ async fn main() {
}
let mut current_store: Option<StoreHandle> = match node.open_root_store() {
Ok(Some((h, open_info))) => {
Ok(Some(open_info)) => {
if open_info.entries_replayed > 0 {
println!("Root: {} (replayed {})", open_info.store_id, open_info.entries_replayed);
} else {
println!("Root: {}", open_info.store_id);
}
// Update shared store for accept loop
*shared_store.write().await = Some(h.clone());
Some(h)
let h = node.root_store().as_ref().cloned();
if let Some(ref handle) = h {
*shared_store.write().await = Some(handle.clone());
}
h
}
Ok(None) => {
println!("Status: Not initialized (use 'init')");
@@ -97,13 +101,8 @@ async fn main() {
};
let cmd_name = args.first().map(|s| s.as_str()).unwrap_or("");
if cmd_name == "quit" || cmd_name == "exit" {
println!("Goodbye!");
break;
}
match cmds.iter().find(|c| c.name == cmd_name) {
match cmds.iter().find(|c| c.name == cmd_name || (cmd_name == "exit" && c.name == "quit")) {
Some(cmd) => {
let cmd_args = &args[1..];
if cmd_args.len() < cmd.min_args || cmd_args.len() > cmd.max_args {
@@ -116,6 +115,7 @@ async fn main() {
*shared_store.write().await = Some(h.clone());
current_store = Some(h);
}
CommandResult::Quit => break,
}
}
}
lattice-cli/src/node_commands.rs
+313
View File
@@ -0,0 +1,313 @@
//! Node commands - operations on the node (mesh, peers, status)
use crate::commands::{block_async, Command, CommandResult, Handler};
use lattice_core::{Node, StoreHandle, PeerStatus, Uuid};
use lattice_net::LatticeEndpoint;
use chrono::DateTime;
use std::time::Instant;
pub fn node_commands() -> Vec<Command> {
vec![
// Store management
Command { name: "init", args: "", desc: "Initialize root store", group: "node", min_args: 0, max_args: 0, handler: cmd_init as Handler },
Command { name: "create-store", args: "", desc: "Create a new store", group: "node", min_args: 0, max_args: 0, handler: cmd_create_store as Handler },
Command { name: "use", args: "<uuid>", desc: "Switch to a store", group: "node", min_args: 1, max_args: 1, handler: cmd_use_store as Handler },
Command { name: "list-stores", args: "", desc: "List all stores", group: "node", min_args: 0, max_args: 0, handler: cmd_list_stores as Handler },
Command { name: "node-status", args: "", desc: "Show node info", group: "node", min_args: 0, max_args: 0, handler: cmd_node_status as Handler },
// Peer management
Command { name: "invite", args: "<pubkey>", desc: "Invite a peer", group: "peers", min_args: 1, max_args: 1, handler: cmd_invite as Handler },
Command { name: "peers", args: "", desc: "List all peers", group: "peers", min_args: 0, max_args: 0, handler: cmd_peers as Handler },
Command { name: "remove", args: "<pubkey>", desc: "Remove a peer", group: "peers", min_args: 1, max_args: 1, handler: cmd_remove as Handler },
// Networking
Command { name: "join", args: "<node_id>", desc: "Join an existing mesh", group: "network", min_args: 1, max_args: 1, handler: cmd_join as Handler },
Command { name: "sync", args: "[node_id]", desc: "Sync with peers", group: "network", min_args: 0, max_args: 1, handler: cmd_sync as Handler },
]
}
// --- Store management ---
fn cmd_init(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult {
match block_async(node.init()) {
Ok(store_id) => {
println!("Initialized with root store: {}", store_id);
println!("Node info stored in /nodes/{}/*", hex::encode(node.node_id()));
match node.root_store().as_ref() {
Some(h) => CommandResult::SwitchTo(h.clone()),
None => CommandResult::Ok,
}
}
Err(e) => {
eprintln!("Error: {}", e);
CommandResult::Ok
}
}
}
fn cmd_create_store(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult {
match node.create_store() {
Ok(store_id) => {
println!("Created store: {}", store_id);
match node.open_store(store_id) {
Ok((handle, _)) => {
println!("Switched to new store");
CommandResult::SwitchTo(handle)
}
Err(e) => {
eprintln!("Warning: {}", e);
CommandResult::Ok
}
}
}
Err(e) => {
eprintln!("Error: {}", e);
CommandResult::Ok
}
}
}
fn cmd_use_store(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult {
let store_id = match Uuid::parse_str(&args[0]) {
Ok(id) => id,
Err(_) => {
eprintln!("Error: invalid UUID '{}'", args[0]);
return CommandResult::Ok;
}
};
let start = Instant::now();
match node.open_store(store_id) {
Ok((handle, info)) => {
if info.entries_replayed > 0 {
println!("Replayed {} entries ({:.2?})", info.entries_replayed, start.elapsed());
} else {
println!("Switched to store {}", store_id);
}
CommandResult::SwitchTo(handle)
}
Err(e) => {
eprintln!("Error: {}", e);
CommandResult::Ok
}
}
}
fn cmd_list_stores(node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult {
let stores = match node.list_stores() {
Ok(s) => s,
Err(e) => {
eprintln!("Error: {}", e);
return CommandResult::Ok;
}
};
let current_id = store.map(|s| s.id());
if stores.is_empty() {
println!("No stores. Use 'init' or 'create-store'.");
} else {
for store_id in stores {
let marker = if Some(store_id) == current_id { " *" } else { "" };
println!("{}{}", store_id, marker);
}
}
CommandResult::Ok
}
// --- Info ---
fn cmd_node_status(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult {
println!("Node ID: {}", hex::encode(node.node_id()));
if let Some(name) = node.name() {
println!("Name: {}", name);
}
println!("Data: {}", node.data_path().display());
match node.root_store_id() {
Ok(Some(id)) => println!("Root: {}", id),
Ok(None) => println!("Root: (not set)"),
Err(_) => println!("Root: (error)"),
}
// Count peers using node.list_peers()
if let Ok(peers) = block_async(node.list_peers()) {
let active = peers.iter().filter(|p| p.status == PeerStatus::Active).count();
let invited = peers.iter().filter(|p| p.status == PeerStatus::Invited).count();
println!("Peers: {} active, {} invited", active, invited);
}
CommandResult::Ok
}
// --- Peer management ---
fn cmd_invite(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult {
let pubkey_hex = &args[0];
let pubkey: [u8; 32] = match hex::decode(pubkey_hex) {
Ok(bytes) if bytes.len() == 32 => bytes.try_into().unwrap(),
_ => {
eprintln!("Invalid pubkey: expected 64 hex chars (32 bytes)");
return CommandResult::Ok;
}
};
match block_async(node.invite_peer(&pubkey)) {
Ok(()) => {
println!("Invited peer: {}", pubkey_hex);
println!(" Status: {} (will become active after sync)", PeerStatus::Invited.as_str());
}
Err(e) => eprintln!("Error: {}", e),
}
CommandResult::Ok
}
fn cmd_peers(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult {
let peers = match block_async(node.list_peers()) {
Ok(p) => p,
Err(e) => {
eprintln!("Error: {}", e);
return CommandResult::Ok;
}
};
if peers.is_empty() {
println!("No peers found.");
return CommandResult::Ok;
}
// Group peers by status
let mut by_status: std::collections::HashMap<PeerStatus, Vec<&lattice_core::PeerInfo>> =
std::collections::HashMap::new();
for peer in &peers {
by_status.entry(peer.status).or_default().push(peer);
}
// Print grouped by status in order: active, invited, removed
let status_order = [PeerStatus::Active, PeerStatus::Invited, PeerStatus::Removed];
for status in &status_order {
if let Some(peer_list) = by_status.get(status) {
println!("\n[{}] ({}):", status.as_str(), peer_list.len());
let mut sorted: Vec<_> = peer_list.iter().collect();
sorted.sort_by(|a, b| a.pubkey.cmp(&b.pubkey));
for peer in sorted {
let added_str = peer.added_at
.and_then(|ts| DateTime::from_timestamp(ts as i64, 0))
.map(|dt| dt.format("%Y-%m-%d").to_string())
.unwrap_or_default();
let info_str = match (peer.name.as_ref(), added_str.is_empty()) {
(Some(name), false) => format!(" {} ({})", name, added_str),
(Some(name), true) => format!(" {}", name),
(None, false) => format!(" ({})", added_str),
(None, true) => String::new(),
};
println!(" {}{}", peer.pubkey, info_str);
}
}
}
CommandResult::Ok
}
fn cmd_remove(node: &Node, _store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult {
let pubkey_hex = &args[0];
let pubkey: [u8; 32] = match hex::decode(pubkey_hex) {
Ok(bytes) if bytes.len() == 32 => bytes.try_into().unwrap(),
_ => {
eprintln!("Invalid pubkey: expected 64 hex characters");
return CommandResult::Ok;
}
};
match block_async(node.remove_peer(&pubkey)) {
Ok(()) => println!("Removed peer: {}...", &pubkey_hex[..10]),
Err(e) => eprintln!("Error: {}", e),
}
CommandResult::Ok
}
// --- Networking ---
fn cmd_join(node: &Node, store: Option<&StoreHandle>, endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult {
let endpoint = match endpoint {
Some(ep) => ep,
None => {
eprintln!("Iroh endpoint not started.");
return CommandResult::Ok;
}
};
if store.is_some() {
eprintln!("Already initialized. Use 'sync' to sync with peers.");
return CommandResult::Ok;
}
let peer_id = match lattice_net::parse_node_id(&args[0]) {
Ok(id) => id,
Err(e) => {
eprintln!("Invalid node ID: {}", e);
return CommandResult::Ok;
}
};
println!("Joining mesh via {}...", peer_id.fmt_short());
match block_async(lattice_net::join_mesh(node, endpoint, peer_id)) {
Ok(handle) => {
println!("Joined mesh! Use 'sync' command to sync entries.");
CommandResult::SwitchTo(handle)
}
Err(e) => {
eprintln!("Join failed: {}", e);
CommandResult::Ok
}
}
}
fn cmd_sync(node: &Node, store: Option<&StoreHandle>, endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult {
let endpoint = match endpoint {
Some(ep) => ep,
None => {
eprintln!("Iroh endpoint not started.");
return CommandResult::Ok;
}
};
let store = match store {
Some(s) => s,
None => {
eprintln!("No store open. Use 'init' or 'join' first.");
return CommandResult::Ok;
}
};
if args.is_empty() {
// Sync with all active peers
match block_async(lattice_net::sync_all(node, endpoint, store)) {
Ok(results) => {
if results.is_empty() {
println!("No peers to sync with.");
} else {
let total: u64 = results.iter().map(|r| r.entries_applied).sum();
println!("\nSync complete! Applied {} entries from {} peer(s).", total, results.len());
}
}
Err(e) => eprintln!("Sync failed: {}", e),
}
} else {
// Sync with specific peer
let peer_id = match lattice_net::parse_node_id(&args[0]) {
Ok(id) => id,
Err(e) => {
eprintln!("Invalid node ID: {}", e);
return CommandResult::Ok;
}
};
println!("Syncing with {}...", peer_id.fmt_short());
match block_async(lattice_net::sync_with_peer(endpoint, store, peer_id)) {
Ok(result) => {
println!("Sync complete! Applied {} entries (peer sent {})",
result.entries_applied, result.entries_sent_by_peer);
}
Err(e) => eprintln!("Sync failed: {}", e),
}
}
CommandResult::Ok
}
lattice-cli/src/store_commands.rs
+209
View File
@@ -0,0 +1,209 @@
//! Store commands - direct KV operations
use crate::commands::{block_async, Command, CommandResult, Handler};
use lattice_core::{Node, StoreHandle};
use lattice_net::LatticeEndpoint;
use std::time::Instant;
pub fn store_commands() -> Vec<Command> {
vec![
Command { name: "store-status", args: "", desc: "Show store info", group: "store", min_args: 0, max_args: 0, handler: cmd_store_status as Handler },
Command { name: "put", args: "<key> <value>", desc: "Store a key-value pair", group: "store", min_args: 2, max_args: 2, handler: cmd_put as Handler },
Command { name: "get", args: "<key> [-v]", desc: "Get value for key", group: "store", min_args: 1, max_args: 2, handler: cmd_get as Handler },
Command { name: "delete", args: "<key>", desc: "Delete a key", group: "store", min_args: 1, max_args: 1, handler: cmd_delete as Handler },
Command { name: "list", args: "[-v]", desc: "List all keys", group: "store", min_args: 0, max_args: 1, handler: cmd_list as Handler },
Command { name: "author-state", args: "[pubkey]", desc: "Show author sync state", group: "store", min_args: 0, max_args: 1, handler: cmd_author_state as Handler },
]
}
fn cmd_store_status(_node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, _args: &[String]) -> CommandResult {
let Some(h) = store else {
println!("No store selected. Use 'init' or 'use <uuid>'");
return CommandResult::Ok;
};
println!("Store ID: {}", h.id());
println!("Log Seq: {}", block_async(h.log_seq()));
println!("Applied: {}", block_async(h.applied_seq()).unwrap_or(0));
let all = block_async(h.list()).unwrap_or_default();
println!("Keys: {}", all.len());
// Show log directory size
let (file_count, total_size) = block_async(h.log_stats());
if file_count > 0 {
println!("Logs: {} files, {} bytes", file_count, total_size);
}
CommandResult::Ok
}
fn cmd_put(_node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult {
let Some(h) = store else {
println!("No store selected. Use 'init' or 'use <uuid>'");
return CommandResult::Ok;
};
let start = Instant::now();
match block_async(h.put(args[0].as_bytes(), args[1].as_bytes())) {
Ok(seq) => println!("OK (seq: {}, {:.2?})", seq, start.elapsed()),
Err(e) => eprintln!("Error: {}", e),
}
CommandResult::Ok
}
fn cmd_get(_node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult {
let Some(h) = store else {
println!("No store selected. Use 'init' or 'use <uuid>'");
return CommandResult::Ok;
};
let verbose = args.get(1).map(|a| a == "-v").unwrap_or(false);
let start = Instant::now();
let key = args[0].as_bytes();
if verbose {
// Show all heads
match block_async(h.get_heads(key)) {
Ok(heads) if heads.is_empty() => println!("(nil)"),
Ok(heads) => {
for (i, head) in heads.iter().enumerate() {
let winner = if i == 0 { "→" } else { " " };
let tombstone = if head.tombstone { "⊗" } else { "" };
let author_short = hex::encode(&head.author).chars().take(8).collect::<String>();
if head.tombstone {
println!("{} {} (deleted) (hlc:{}, author:{})",
winner, tombstone, head.hlc, author_short);
} else {
println!("{} {} (hlc:{}, author:{})",
winner, format_value(&head.value), head.hlc, author_short);
}
}
if heads.len() > 1 {
println!("⚠ {} heads (conflict)", heads.len());
}
println!("({:.2?})", start.elapsed());
}
Err(e) => eprintln!("Error: {}", e),
}
} else {
match block_async(h.get(key)) {
Ok(Some(v)) => {
let heads = block_async(h.get_heads(key)).unwrap_or_default();
if heads.len() > 1 {
println!("{} (⚠ {} heads)", format_value(&v), heads.len());
} else {
println!("{}", format_value(&v));
}
println!("({:.2?})", start.elapsed());
}
Ok(None) => println!("(nil)"),
Err(e) => eprintln!("Error: {}", e),
}
}
CommandResult::Ok
}
fn cmd_delete(_node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult {
let Some(h) = store else {
println!("No store selected. Use 'init' or 'use <uuid>'");
return CommandResult::Ok;
};
let start = Instant::now();
match block_async(h.delete(args[0].as_bytes())) {
Ok(seq) => println!("OK (seq: {}, {:.2?})", seq, start.elapsed()),
Err(e) => eprintln!("Error: {}", e),
}
CommandResult::Ok
}
fn cmd_list(_node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult {
let Some(h) = store else {
println!("No store selected. Use 'init' or 'use <uuid>'");
return CommandResult::Ok;
};
let verbose = args.first().map(|a| a == "-v").unwrap_or(false);
let start = Instant::now();
match block_async(h.list()) {
Ok(entries) => {
if entries.is_empty() {
println!("(empty)");
} else {
for (k, v) in &entries {
let key_str = format_value(k);
if verbose {
// Show all heads for this key
let heads = block_async(h.get_heads(k)).unwrap_or_default();
println!("{}:", key_str);
for (i, head) in heads.iter().enumerate() {
let winner = if i == 0 { "→" } else { " " };
let author_short = hex::encode(&head.author).chars().take(8).collect::<String>();
if head.tombstone {
println!(" {} ⊗ (deleted) (hlc:{}, author:{})",
winner, head.hlc, author_short);
} else {
println!(" {} {} (hlc:{}, author:{})",
winner, format_value(&head.value), head.hlc, author_short);
}
}
} else {
// Check for multiple heads
let heads = block_async(h.get_heads(k)).unwrap_or_default();
if heads.len() > 1 {
println!("{} = {} (⚠ {} heads)", key_str, format_value(v), heads.len());
} else {
println!("{} = {}", key_str, format_value(v));
}
}
}
println!("({} keys, {:.2?})", entries.len(), start.elapsed());
}
}
Err(e) => eprintln!("Error: {}", e),
}
CommandResult::Ok
}
fn cmd_author_state(node: &Node, store: Option<&StoreHandle>, _endpoint: Option<&LatticeEndpoint>, args: &[String]) -> CommandResult {
let store = match store {
Some(s) => s,
None => {
eprintln!("Error: no store selected");
return CommandResult::Ok;
}
};
// Get author: from arg or default to self
let author_bytes: [u8; 32] = if args.is_empty() {
node.node_id()
} else {
let hex_str = args[0].trim_start_matches("0x");
match hex::decode(hex_str) {
Ok(bytes) if bytes.len() == 32 => bytes.try_into().unwrap(),
Ok(bytes) => {
eprintln!("Error: author must be 32 bytes, got {}", bytes.len());
return CommandResult::Ok;
}
Err(e) => {
eprintln!("Error: invalid hex: {}", e);
return CommandResult::Ok;
}
}
};
match block_async(store.author_state(&author_bytes)) {
Ok(Some(state)) => {
println!("Author: {}", hex::encode(&author_bytes));
println!(" seq: {}", state.seq);
println!(" hash: {}", hex::encode(&state.hash));
println!(" log_offset: {}", state.log_offset);
}
Ok(None) => {
println!("No state for author: {}", hex::encode(&author_bytes));
}
Err(e) => eprintln!("Error: {}", e),
}
CommandResult::Ok
}
fn format_value(v: &[u8]) -> String {
std::str::from_utf8(v).map(String::from).unwrap_or_else(|_| format!("0x{}", hex::encode(v)))
}
lattice-core/src/lib.rs
+1 -1
View File
@@ -35,7 +35,7 @@ pub mod store_actor;
pub const MAX_ENTRY_SIZE: usize = 16 * 1024 * 1024;
pub use node_identity::{NodeIdentity, PeerStatus};
pub use node::{Node, NodeBuilder, NodeInfo, StoreInfo, StoreHandle, NodeError};
pub use node::{Node, NodeBuilder, NodeInfo, StoreInfo, StoreHandle, NodeError, PeerInfo};
pub use sigchain::{SigChain, SigChainManager};
pub use entry::Entry;
pub use sync_state::{SyncState, AuthorInfo, MissingRange};
lattice-core/src/node.rs
+196 -23
View File
@@ -54,6 +54,15 @@ pub struct StoreInfo {
pub entries_replayed: u64,
}
/// Information about a peer in the mesh
pub struct PeerInfo {
pub pubkey: String,
pub name: Option<String>,
pub added_at: Option<u64>,
pub added_by: Option<String>,
pub status: PeerStatus,
}
pub struct NodeBuilder {
pub data_dir: DataDir,
}
@@ -90,6 +99,7 @@ impl NodeBuilder {
data_dir: self.data_dir,
node: Rc::new(node),
meta,
root_store: std::cell::RefCell::new(None),
})
}
}
@@ -103,6 +113,7 @@ pub struct Node {
data_dir: DataDir,
node: Rc<NodeIdentity>,
meta: MetaStore,
root_store: std::cell::RefCell<Option<StoreHandle>>,
}
impl Node {
@@ -133,13 +144,13 @@ impl Node {
}
/// Set the node's display name.
/// Updates meta.db and if a store handle is provided, also updates /nodes/{pubkey}/name
pub async fn set_name(&self, name: &str, store: Option<&StoreHandle>) -> Result<(), NodeError> {
/// Updates meta.db and if root store is open, also updates /nodes/{pubkey}/name
pub async fn set_name(&self, name: &str) -> Result<(), NodeError> {
// Update meta.db
self.meta.set_name(name)?;
// If store provided, update there too
if let Some(handle) = store {
// If root store is open, update there too
if let Some(handle) = self.root_store.borrow().as_ref() {
let pubkey_hex = hex::encode(self.node.public_key_bytes());
let name_key = format!("/nodes/{}/name", pubkey_hex);
handle.put(name_key.as_bytes(), name.as_bytes()).await?;
@@ -149,20 +160,31 @@ impl Node {
}
/// Get the root store ID
pub fn root_store(&self) -> Result<Option<Uuid>, NodeError> {
pub fn root_store_id(&self) -> Result<Option<Uuid>, NodeError> {
Ok(self.meta.root_store()?)
}
/// Open the root store if set
pub fn open_root_store(&self) -> Result<Option<(StoreHandle, StoreInfo)>, NodeError> {
/// Get reference to the cached root store handle (if open)
pub fn root_store(&self) -> std::cell::Ref<'_, Option<StoreHandle>> {
self.root_store.borrow()
}
/// Open the root store if set. Node owns the handle internally.
/// Returns StoreInfo on success, or None if no root store is set.
pub fn open_root_store(&self) -> Result<Option<StoreInfo>, NodeError> {
match self.meta.root_store()? {
Some(id) => Ok(Some(self.open_store(id)?)),
Some(id) => {
let (handle, info) = self.open_store(id)?;
*self.root_store.borrow_mut() = Some(handle);
Ok(Some(info))
}
None => Ok(None),
}
}
/// Initialize the node with a root store (fails if already initialized).
/// Writes the node's pubkey to `/nodes/{pubkey}/info` in the root store.
pub async fn init(&self) -> Result<(Uuid, StoreHandle), NodeError> {
/// Node owns the store handle internally. Access via root_store().
pub async fn init(&self) -> Result<Uuid, NodeError> {
if self.meta.root_store()?.is_some() {
return Err(NodeError::AlreadyInitialized);
}
@@ -190,7 +212,140 @@ impl Node {
let status_key = format!("/nodes/{}/status", pubkey_hex);
handle.put(status_key.as_bytes(), PeerStatus::Active.as_str().as_bytes()).await?;
Ok((store_id, handle))
// Store the handle - node owns it
*self.root_store.borrow_mut() = Some(handle);
Ok(store_id)
}
// --- Peer Management ---
/// Invite a peer to the mesh. Writes their info with status = invited.
pub async fn invite_peer(&self, pubkey: &[u8; 32]) -> Result<(), NodeError> {
let store = self.root_store.borrow();
let store = store.as_ref()
.ok_or_else(|| NodeError::Actor("No root store open".to_string()))?;
let pubkey_hex = hex::encode(pubkey);
let my_pubkey_hex = hex::encode(self.node.public_key_bytes());
let added_at = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs())
.unwrap_or(0);
// Write added_by
let added_by_key = format!("/nodes/{}/added_by", pubkey_hex);
store.put(added_by_key.as_bytes(), my_pubkey_hex.as_bytes()).await?;
// Write added_at
let added_at_key = format!("/nodes/{}/added_at", pubkey_hex);
store.put(added_at_key.as_bytes(), added_at.to_string().as_bytes()).await?;
// Write status = invited
let status_key = format!("/nodes/{}/status", pubkey_hex);
store.put(status_key.as_bytes(), PeerStatus::Invited.as_str().as_bytes()).await?;
Ok(())
}
/// List all peers in the mesh with their info
pub async fn list_peers(&self) -> Result<Vec<PeerInfo>, NodeError> {
let store = self.root_store.borrow();
let store = store.as_ref()
.ok_or_else(|| NodeError::Actor("No root store open".to_string()))?;
let all = store.list().await?;
// Collect unique pubkeys with status
let mut peers_map: std::collections::HashMap<String, PeerStatus> = std::collections::HashMap::new();
for (key, value) in &all {
let key_str = String::from_utf8_lossy(key);
if key_str.ends_with("/status") {
if let Some(pubkey) = key_str.strip_prefix("/nodes/").and_then(|s| s.strip_suffix("/status")) {
let status_str = String::from_utf8_lossy(value);
if let Some(status) = PeerStatus::from_str(&status_str) {
peers_map.insert(pubkey.to_string(), status);
}
}
}
}
// Build PeerInfo for each peer
let mut peers = Vec::new();
for (pubkey, status) in peers_map {
let name_key = format!("/nodes/{}/name", pubkey);
let added_at_key = format!("/nodes/{}/added_at", pubkey);
let added_by_key = format!("/nodes/{}/added_by", pubkey);
let name = store.get(name_key.as_bytes()).await?
.map(|b| String::from_utf8_lossy(&b).to_string());
let added_at = store.get(added_at_key.as_bytes()).await?
.and_then(|b| String::from_utf8_lossy(&b).parse().ok());
let added_by = store.get(added_by_key.as_bytes()).await?
.map(|b| String::from_utf8_lossy(&b).to_string());
peers.push(PeerInfo {
pubkey,
name,
added_at,
added_by,
status,
});
}
Ok(peers)
}
/// Remove a peer from the mesh (sets status to removed)
pub async fn remove_peer(&self, pubkey: &[u8; 32]) -> Result<(), NodeError> {
let store = self.root_store.borrow();
let store = store.as_ref()
.ok_or_else(|| NodeError::Actor("No root store open".to_string()))?;
let pubkey_hex = hex::encode(pubkey);
// Prevent self-removal
if pubkey == &self.node.public_key_bytes() {
return Err(NodeError::Actor("Cannot remove yourself".to_string()));
}
// Check if peer exists
let status_key = format!("/nodes/{}/status", pubkey_hex);
match store.get(status_key.as_bytes()).await? {
Some(status) if status == PeerStatus::Removed.as_str().as_bytes() => {
return Err(NodeError::Actor("Peer already removed".to_string()));
}
None => {
return Err(NodeError::Actor("Peer not found".to_string()));
}
_ => {}
}
// Set status to removed
store.put(status_key.as_bytes(), PeerStatus::Removed.as_str().as_bytes()).await?;
Ok(())
}
/// Get a peer's status
pub async fn get_peer_status(&self, pubkey: &[u8; 32]) -> Result<Option<PeerStatus>, NodeError> {
let store = self.root_store.borrow();
let store = store.as_ref()
.ok_or_else(|| NodeError::Actor("No root store open".to_string()))?;
let pubkey_hex = hex::encode(pubkey);
let status_key = format!("/nodes/{}/status", pubkey_hex);
match store.get(status_key.as_bytes()).await? {
Some(bytes) => {
let status_str = String::from_utf8_lossy(&bytes);
Ok(PeerStatus::from_str(&status_str))
}
None => Ok(None),
}
}
pub fn list_stores(&self) -> Result<Vec<Uuid>, NodeError> {
@@ -336,6 +491,14 @@ impl StoreHandle {
.map_err(NodeError::Store)
}
/// Get log directory statistics (file count, total bytes)
pub async fn log_stats(&self) -> (usize, u64) {
use StoreCmd;
let (resp_tx, resp_rx) = tokio::sync::oneshot::channel();
let _ = self.tx.send(StoreCmd::LogStats { resp: resp_tx }).await;
resp_rx.await.unwrap_or((0, 0))
}
pub async fn sync_state(&self) -> Result<crate::sync_state::SyncState, NodeError> {
use StoreCmd;
let (resp_tx, resp_rx) = tokio::sync::oneshot::channel();
@@ -464,11 +627,11 @@ mod tests {
.expect("create node");
// Initially no root store
assert!(node.root_store().unwrap().is_none());
assert!(node.root_store().is_none());
// Init creates root store
let (root_id, _handle) = node.init().await.expect("init failed");
assert_eq!(node.root_store().unwrap(), Some(root_id));
let root_id = node.init().await.expect("init failed");
assert_eq!(node.root_store_id().unwrap(), Some(root_id));
let _ = std::fs::remove_dir_all(data_dir.base());
}
@@ -503,7 +666,7 @@ mod tests {
let node = NodeBuilder { data_dir: data_dir.clone() }
.build()
.expect("create node");
let (root_id, _) = node.init().await.expect("init");
let root_id = node.init().await.expect("init");
drop(node); // End first session
// Second session: root_store should persist
@@ -511,7 +674,7 @@ mod tests {
.build()
.expect("reload node");
assert_eq!(node.root_store().unwrap(), Some(root_id));
assert_eq!(node.root_store_id().unwrap(), Some(root_id));
let _ = std::fs::remove_dir_all(data_dir.base());
}
@@ -523,7 +686,9 @@ mod tests {
let node = NodeBuilder { data_dir: data_dir.clone() }
.build()
.expect("create node");
let (_, store) = node.init().await.expect("init");
node.init().await.expect("init");
let store = node.root_store();
let store = store.as_ref().unwrap();
// Get baseline seq after init
let baseline = store.log_seq().await;
@@ -562,24 +727,32 @@ mod tests {
let initial_name = node.name().unwrap();
// Init creates root store
let (_, store) = node.init().await.expect("init");
node.init().await.expect("init");
// Verify initial name is in store
let pubkey_hex = hex::encode(node.node_id());
let name_key = format!("/nodes/{}/name", pubkey_hex);
let stored_name = store.get(name_key.as_bytes()).await.unwrap();
assert_eq!(stored_name, Some(initial_name.as_bytes().to_vec()));
{
let store = node.root_store();
let store = store.as_ref().unwrap();
let stored_name = store.get(name_key.as_bytes()).await.unwrap();
assert_eq!(stored_name, Some(initial_name.as_bytes().to_vec()));
}
// Change name
let new_name = "my-custom-name";
node.set_name(new_name, Some(&store)).await.expect("set_name");
node.set_name(new_name).await.expect("set_name");
// Verify meta.db updated
assert_eq!(node.name(), Some(new_name.to_string()));
// Verify store updated
let stored_name = store.get(name_key.as_bytes()).await.unwrap();
assert_eq!(stored_name, Some(new_name.as_bytes().to_vec()));
{
let store = node.root_store();
let store = store.as_ref().unwrap();
let stored_name = store.get(name_key.as_bytes()).await.unwrap();
assert_eq!(stored_name, Some(new_name.as_bytes().to_vec()));
}
let _ = std::fs::remove_dir_all(data_dir.base());
}
lattice-core/src/sigchain.rs
+20
View File
@@ -315,6 +315,26 @@ impl SigChainManager {
pub fn logs_dir(&self) -> &Path {
&self.logs_dir
}
/// Get log directory statistics (file count, total bytes)
pub fn log_stats(&self) -> (usize, u64) {
if !self.logs_dir.exists() {
return (0, 0);
}
let mut total_size = 0u64;
let mut file_count = 0;
if let Ok(entries) = std::fs::read_dir(&self.logs_dir) {
for entry in entries.flatten() {
if let Ok(meta) = entry.metadata() {
if meta.is_file() {
total_size += meta.len();
file_count += 1;
}
}
}
}
(file_count, total_size)
}
}
#[cfg(test)]
lattice-core/src/store_actor.rs
+6
View File
@@ -58,6 +58,9 @@ pub enum StoreCmd {
entry: SignedEntry,
resp: oneshot::Sender<Result<(), StoreError>>,
},
LogStats {
resp: oneshot::Sender<(usize, u64)>,
},
Shutdown,
}
@@ -185,6 +188,9 @@ impl StoreActor {
let result = self.store.apply_entry(&entry);
let _ = resp.send(result);
}
StoreCmd::LogStats { resp } => {
let _ = resp.send(self.chain_manager.log_stats());
}
StoreCmd::Shutdown => {
break;
}
lattice-net/src/mesh/sync.rs
+2 -3
View File
@@ -60,7 +60,7 @@ pub async fn join_mesh(
// Immediately sync with the peer to get initial data
println!("[Join] Syncing with peer to get initial data...");
match sync_with_peer(node, endpoint, &handle, peer_id).await {
match sync_with_peer(endpoint, &handle, peer_id).await {
Ok(result) => {
println!("[Join] Initial sync complete: applied {} entries", result.entries_applied);
}
@@ -87,7 +87,6 @@ pub async fn join_mesh(
/// Sync with a specific peer (bidirectional).
/// Both sides exchange states and send missing entries to each other.
pub async fn sync_with_peer(
node: &Node,
endpoint: &LatticeEndpoint,
store: &StoreHandle,
peer_id: iroh::PublicKey,
@@ -192,7 +191,7 @@ pub async fn sync_all(
// Sync with each peer
let mut results = Vec::new();
for peer_id in peer_ids {
match sync_with_peer(node, endpoint, store, peer_id).await {
match sync_with_peer(endpoint, store, peer_id).await {
Ok(result) => results.push(result),
Err(e) => {
// Log error but continue with other peers