diff --git a/docs/roadmap.md b/docs/roadmap.md index 3f6e25b..4a3431f 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -68,16 +68,16 @@ - [x] StoreHandle wraps channel sender, keeps current API - [x] Validate: CLI works as before with actor -**Phase 2: Async Runtime** -- [ ] Add tokio runtime (`#[tokio::main]`) -- [ ] Migrate `std::sync::mpsc` → `tokio::sync::mpsc` -- [ ] Async CLI using `tokio::io::stdin()` or `rustyline` async +**Phase 2: Async Runtime** ✓ +- [x] Add tokio runtime (`#[tokio::main]`) +- [x] Migrate `std::sync::mpsc` → `tokio::sync::mpsc` +- [x] Async CLI using `block_in_place` for sync handlers ### Success Criteria - [x] CLI still works as before - [x] Store operations serialized (no data races) -- [ ] Ready for concurrent network tasks +- [x] Ready for concurrent network tasks --- diff --git a/lattice-cli/Cargo.toml b/lattice-cli/Cargo.toml index 8020d9d..58bf50a 100644 --- a/lattice-cli/Cargo.toml +++ b/lattice-cli/Cargo.toml @@ -14,4 +14,5 @@ lattice-core = { workspace = true } rustyline = { workspace = true } hex = { workspace = true } thiserror = { workspace = true } +tokio = { workspace = true } shlex = "1" diff --git a/lattice-cli/src/commands.rs b/lattice-cli/src/commands.rs index 3cb3550..69ad722 100644 --- a/lattice-cli/src/commands.rs +++ b/lattice-cli/src/commands.rs @@ -12,6 +12,13 @@ pub enum CommandResult { SwitchTo(StoreHandle), } +/// Helper to call async code from sync command handlers +fn block_async(f: F) -> F::Output { + tokio::task::block_in_place(|| { + tokio::runtime::Handle::current().block_on(f) + }) +} + pub type Handler = fn(&LatticeNode, Option<&StoreHandle>, &[String]) -> CommandResult; pub struct Command { @@ -232,8 +239,8 @@ fn cmd_status(node: &LatticeNode, store: Option<&StoreHandle>, _args: &[String]) } if let Some(h) = store { println!("Store: {}", h.id()); - println!("Log Seq: {}", h.log_seq()); - println!("Applied: {}", h.applied_seq().unwrap_or(0)); + println!("Log Seq: {}", block_async(h.log_seq())); + println!("Applied: {}", block_async(h.applied_seq()).unwrap_or(0)); } else { println!("Store: (none)"); } @@ -248,7 +255,7 @@ fn cmd_put(_node: &LatticeNode, store: Option<&StoreHandle>, args: &[String]) -> return CommandResult::Ok; }; let start = Instant::now(); - match h.put(args[0].as_bytes(), args[1].as_bytes()) { + 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), } @@ -266,7 +273,7 @@ fn cmd_get(_node: &LatticeNode, store: Option<&StoreHandle>, args: &[String]) -> if verbose { // Show all heads - match h.get_heads(key) { + match block_async(h.get_heads(key)) { Ok(heads) if heads.is_empty() => println!("(nil)"), Ok(heads) => { for (i, head) in heads.iter().enumerate() { @@ -289,9 +296,9 @@ fn cmd_get(_node: &LatticeNode, store: Option<&StoreHandle>, args: &[String]) -> Err(e) => eprintln!("Error: {}", e), } } else { - match h.get(key) { + match block_async(h.get(key)) { Ok(Some(v)) => { - let heads = h.get_heads(key).unwrap_or_default(); + let heads = block_async(h.get_heads(key)).unwrap_or_default(); if heads.len() > 1 { println!("{} (⚠ {} heads)", format_value(&v), heads.len()); } else { @@ -312,7 +319,7 @@ fn cmd_delete(_node: &LatticeNode, store: Option<&StoreHandle>, args: &[String]) return CommandResult::Ok; }; let start = Instant::now(); - match h.delete(args[0].as_bytes()) { + match block_async(h.delete(args[0].as_bytes())) { Ok(seq) => println!("OK (seq: {}, {:.2?})", seq, start.elapsed()), Err(e) => eprintln!("Error: {}", e), } @@ -326,7 +333,7 @@ fn cmd_list(_node: &LatticeNode, store: Option<&StoreHandle>, args: &[String]) - }; let verbose = args.first().map(|a| a == "-v").unwrap_or(false); let start = Instant::now(); - match h.list() { + match block_async(h.list()) { Ok(entries) => { if entries.is_empty() { println!("(empty)"); @@ -335,7 +342,7 @@ fn cmd_list(_node: &LatticeNode, store: Option<&StoreHandle>, args: &[String]) - let key_str = format_value(k); if verbose { // Show all heads for this key - let heads = h.get_heads(k).unwrap_or_default(); + 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 { " " }; @@ -391,7 +398,7 @@ fn cmd_author_state(node: &LatticeNode, store: Option<&StoreHandle>, args: &[Str } }; - match store.author_state(&author_bytes) { + match block_async(store.author_state(&author_bytes)) { Ok(Some(state)) => { println!("Author: {}", hex::encode(&author_bytes)); println!(" seq: {}", state.seq); diff --git a/lattice-cli/src/main.rs b/lattice-cli/src/main.rs index 48589df..ea7c151 100644 --- a/lattice-cli/src/main.rs +++ b/lattice-cli/src/main.rs @@ -9,7 +9,8 @@ use node::{LatticeNodeBuilder, StoreHandle}; use rustyline::error::ReadlineError; use rustyline::DefaultEditor; -fn main() { +#[tokio::main] +async fn main() { println!("Lattice CLI v{}", env!("CARGO_PKG_VERSION")); println!("Type 'help' for commands, 'quit' to exit.\n"); diff --git a/lattice-cli/src/node.rs b/lattice-cli/src/node.rs index fcec953..8d5c77d 100644 --- a/lattice-cli/src/node.rs +++ b/lattice-cli/src/node.rs @@ -188,95 +188,97 @@ impl LatticeNode { /// A handle to a specific store - wraps channel to actor thread pub struct StoreHandle { store_id: Uuid, - tx: std::sync::mpsc::Sender, + tx: tokio::sync::mpsc::Sender, actor_handle: Option>, } impl StoreHandle { pub fn id(&self) -> Uuid { self.store_id } - pub fn get(&self, key: &[u8]) -> Result>, NodeError> { + pub async fn get(&self, key: &[u8]) -> Result>, NodeError> { use crate::store_actor::StoreCmd; - let (resp_tx, resp_rx) = std::sync::mpsc::channel(); - self.tx.send(StoreCmd::Get { key: key.to_vec(), resp: resp_tx }) + let (resp_tx, resp_rx) = tokio::sync::oneshot::channel(); + self.tx.send(StoreCmd::Get { key: key.to_vec(), resp: resp_tx }).await .map_err(|_| NodeError::ChannelClosed)?; - resp_rx.recv() + resp_rx.await .map_err(|_| NodeError::ChannelClosed)? .map_err(NodeError::Store) } - pub fn get_heads(&self, key: &[u8]) -> Result, NodeError> { + pub async fn get_heads(&self, key: &[u8]) -> Result, NodeError> { use crate::store_actor::StoreCmd; - let (resp_tx, resp_rx) = std::sync::mpsc::channel(); - self.tx.send(StoreCmd::GetHeads { key: key.to_vec(), resp: resp_tx }) + let (resp_tx, resp_rx) = tokio::sync::oneshot::channel(); + self.tx.send(StoreCmd::GetHeads { key: key.to_vec(), resp: resp_tx }).await .map_err(|_| NodeError::ChannelClosed)?; - resp_rx.recv() + resp_rx.await .map_err(|_| NodeError::ChannelClosed)? .map_err(NodeError::Store) } - pub fn list(&self) -> Result, Vec)>, NodeError> { + pub async fn list(&self) -> Result, Vec)>, NodeError> { use crate::store_actor::StoreCmd; - let (resp_tx, resp_rx) = std::sync::mpsc::channel(); - self.tx.send(StoreCmd::List { resp: resp_tx }) + let (resp_tx, resp_rx) = tokio::sync::oneshot::channel(); + self.tx.send(StoreCmd::List { resp: resp_tx }).await .map_err(|_| NodeError::ChannelClosed)?; - resp_rx.recv() + resp_rx.await .map_err(|_| NodeError::ChannelClosed)? .map_err(NodeError::Store) } - pub fn log_seq(&self) -> u64 { + pub async fn log_seq(&self) -> u64 { use crate::store_actor::StoreCmd; - let (resp_tx, resp_rx) = std::sync::mpsc::channel(); - let _ = self.tx.send(StoreCmd::LogSeq { resp: resp_tx }); - resp_rx.recv().unwrap_or(0) + let (resp_tx, resp_rx) = tokio::sync::oneshot::channel(); + let _ = self.tx.send(StoreCmd::LogSeq { resp: resp_tx }).await; + resp_rx.await.unwrap_or(0) } - pub fn applied_seq(&self) -> Result { + pub async fn applied_seq(&self) -> Result { use crate::store_actor::StoreCmd; - let (resp_tx, resp_rx) = std::sync::mpsc::channel(); - self.tx.send(StoreCmd::AppliedSeq { resp: resp_tx }) + let (resp_tx, resp_rx) = tokio::sync::oneshot::channel(); + self.tx.send(StoreCmd::AppliedSeq { resp: resp_tx }).await .map_err(|_| NodeError::ChannelClosed)?; - resp_rx.recv() + resp_rx.await .map_err(|_| NodeError::ChannelClosed)? .map_err(NodeError::Store) } - pub fn author_state(&self, author: &[u8; 32]) -> Result, NodeError> { + pub async fn author_state(&self, author: &[u8; 32]) -> Result, NodeError> { use crate::store_actor::StoreCmd; - let (resp_tx, resp_rx) = std::sync::mpsc::channel(); - self.tx.send(StoreCmd::AuthorState { author: *author, resp: resp_tx }) + let (resp_tx, resp_rx) = tokio::sync::oneshot::channel(); + self.tx.send(StoreCmd::AuthorState { author: *author, resp: resp_tx }).await .map_err(|_| NodeError::ChannelClosed)?; - resp_rx.recv() + resp_rx.await .map_err(|_| NodeError::ChannelClosed)? .map_err(NodeError::Store) } - pub fn put(&self, key: &[u8], value: &[u8]) -> Result { + pub async fn put(&self, key: &[u8], value: &[u8]) -> Result { use crate::store_actor::StoreCmd; - let (resp_tx, resp_rx) = std::sync::mpsc::channel(); - self.tx.send(StoreCmd::Put { key: key.to_vec(), value: value.to_vec(), resp: resp_tx }) + let (resp_tx, resp_rx) = tokio::sync::oneshot::channel(); + self.tx.send(StoreCmd::Put { key: key.to_vec(), value: value.to_vec(), resp: resp_tx }).await .map_err(|_| NodeError::ChannelClosed)?; - resp_rx.recv() + resp_rx.await .map_err(|_| NodeError::ChannelClosed)? .map_err(|e| NodeError::Actor(e.to_string())) } - pub fn delete(&self, key: &[u8]) -> Result { + pub async fn delete(&self, key: &[u8]) -> Result { use crate::store_actor::StoreCmd; - let (resp_tx, resp_rx) = std::sync::mpsc::channel(); - self.tx.send(StoreCmd::Delete { key: key.to_vec(), resp: resp_tx }) + let (resp_tx, resp_rx) = tokio::sync::oneshot::channel(); + self.tx.send(StoreCmd::Delete { key: key.to_vec(), resp: resp_tx }).await .map_err(|_| NodeError::ChannelClosed)?; - resp_rx.recv() + resp_rx.await .map_err(|_| NodeError::ChannelClosed)? .map_err(|e| NodeError::Actor(e.to_string())) } + } impl Drop for StoreHandle { fn drop(&mut self) { - // Send shutdown command and wait for actor to finish - let _ = self.tx.send(crate::store_actor::StoreCmd::Shutdown); + // Send shutdown command (non-blocking) and wait for actor to finish + // Use try_send to avoid panic in async context + let _ = self.tx.try_send(crate::store_actor::StoreCmd::Shutdown); if let Some(handle) = self.actor_handle.take() { let _ = handle.join(); } @@ -294,8 +296,8 @@ mod tests { DataDir::new(path) } - #[test] - fn test_create_and_open_store() { + #[tokio::test] + async fn test_create_and_open_store() { let data_dir = temp_data_dir("meta_store"); let node = LatticeNodeBuilder { data_dir: data_dir.clone() } @@ -311,14 +313,14 @@ mod tests { assert!(stores.contains(&store_id)); let (handle, _) = node.open_store(store_id).expect("Failed to open store"); - handle.put(b"/key", b"value").expect("put failed"); - assert_eq!(handle.get(b"/key").unwrap(), Some(b"value".to_vec())); + handle.put(b"/key", b"value").await.expect("put failed"); + assert_eq!(handle.get(b"/key").await.unwrap(), Some(b"value".to_vec())); let _ = std::fs::remove_dir_all(data_dir.base()); } - #[test] - fn test_store_isolation() { + #[tokio::test] + async fn test_store_isolation() { let data_dir = temp_data_dir("meta_isolation"); let node = LatticeNodeBuilder { data_dir: data_dir.clone() } @@ -329,12 +331,12 @@ mod tests { let store_b = node.create_store().expect("create B"); let (handle_a, _) = node.open_store(store_a).expect("open A"); - handle_a.put(b"/key", b"from A").expect("put A"); + handle_a.put(b"/key", b"from A").await.expect("put A"); let (handle_b, _) = node.open_store(store_b).expect("open B"); - assert_eq!(handle_b.get(b"/key").unwrap(), None); + assert_eq!(handle_b.get(b"/key").await.unwrap(), None); - assert_eq!(handle_a.get(b"/key").unwrap(), Some(b"from A".to_vec())); + assert_eq!(handle_a.get(b"/key").await.unwrap(), Some(b"from A".to_vec())); let _ = std::fs::remove_dir_all(data_dir.base()); } @@ -398,8 +400,8 @@ mod tests { let _ = std::fs::remove_dir_all(data_dir.base()); } - #[test] - fn test_idempotent_put_and_delete() { + #[tokio::test] + async fn test_idempotent_put_and_delete() { let data_dir = temp_data_dir("idempotent"); let node = LatticeNodeBuilder { data_dir: data_dir.clone() } @@ -409,22 +411,22 @@ mod tests { let (store, _) = node.open_store(store_id).expect("open store"); // Put twice with same value - second should be idempotent - let seq1 = store.put(b"/key", b"value").expect("put 1"); + let seq1 = store.put(b"/key", b"value").await.expect("put 1"); assert_eq!(seq1, 1); - let seq2 = store.put(b"/key", b"value").expect("put 2"); + let seq2 = store.put(b"/key", b"value").await.expect("put 2"); assert_eq!(seq2, 1, "Second put with same value should be idempotent (no new entry)"); - assert_eq!(store.log_seq(), 1, "Log should have 1 entry, not 2"); + assert_eq!(store.log_seq().await, 1, "Log should have 1 entry, not 2"); // Delete twice - second should be idempotent - let seq3 = store.delete(b"/key").expect("delete 1"); + let seq3 = store.delete(b"/key").await.expect("delete 1"); assert_eq!(seq3, 2); - let seq4 = store.delete(b"/key").expect("delete 2"); + let seq4 = store.delete(b"/key").await.expect("delete 2"); assert_eq!(seq4, 2, "Second delete should be idempotent (no new entry)"); - assert_eq!(store.log_seq(), 2, "Log should have 2 entries, not 3"); + assert_eq!(store.log_seq().await, 2, "Log should have 2 entries, not 3"); let _ = std::fs::remove_dir_all(data_dir.base()); } diff --git a/lattice-cli/src/store_actor.rs b/lattice-cli/src/store_actor.rs index 12a3191..2bcc43d 100644 --- a/lattice-cli/src/store_actor.rs +++ b/lattice-cli/src/store_actor.rs @@ -7,40 +7,40 @@ use lattice_core::{ sigchain::SigChainError, store::StoreError, }; -use std::sync::mpsc::{self, Receiver, Sender}; +use tokio::sync::{mpsc, oneshot}; use std::thread::{self, JoinHandle}; /// Commands sent to the store actor pub enum StoreCmd { Get { key: Vec, - resp: std::sync::mpsc::Sender>, StoreError>>, + resp: oneshot::Sender>, StoreError>>, }, GetHeads { key: Vec, - resp: std::sync::mpsc::Sender, StoreError>>, + resp: oneshot::Sender, StoreError>>, }, List { - resp: std::sync::mpsc::Sender, Vec)>, StoreError>>, + resp: oneshot::Sender, Vec)>, StoreError>>, }, Put { key: Vec, value: Vec, - resp: std::sync::mpsc::Sender>, + resp: oneshot::Sender>, }, Delete { key: Vec, - resp: std::sync::mpsc::Sender>, + resp: oneshot::Sender>, }, LogSeq { - resp: std::sync::mpsc::Sender, + resp: oneshot::Sender, }, AppliedSeq { - resp: std::sync::mpsc::Sender>, + resp: oneshot::Sender>, }, AuthorState { author: [u8; 32], - resp: std::sync::mpsc::Sender, StoreError>>, + resp: oneshot::Sender, StoreError>>, }, Shutdown, } @@ -80,7 +80,7 @@ pub struct StoreActor { store: Store, sigchain: SigChain, node: Node, - rx: Receiver, + rx: mpsc::Receiver, } impl StoreActor { @@ -90,7 +90,7 @@ impl StoreActor { store: Store, sigchain: SigChain, node: Node, - rx: Receiver, + rx: mpsc::Receiver, ) -> Self { Self { store_id, @@ -102,8 +102,9 @@ impl StoreActor { } /// Run the actor loop - processes commands until Shutdown received + /// Uses blocking_recv since redb is sync and we run in spawn_blocking pub fn run(mut self) { - while let Ok(cmd) = self.rx.recv() { + while let Some(cmd) = self.rx.blocking_recv() { match cmd { StoreCmd::Get { key, resp } => { let _ = resp.send(self.store.get(&key)); @@ -186,13 +187,14 @@ impl StoreActor { } /// Spawn a store actor in a new thread, returns (sender, join_handle) +/// Uses std::thread since redb is blocking pub fn spawn_store_actor( store_id: Uuid, store: Store, sigchain: SigChain, node: Node, -) -> (Sender, JoinHandle<()>) { - let (tx, rx) = mpsc::channel(); +) -> (mpsc::Sender, JoinHandle<()>) { + let (tx, rx) = mpsc::channel(32); let actor = StoreActor::new(store_id, store, sigchain, node, rx); let handle = thread::spawn(move || actor.run()); (tx, handle)