diff --git a/Cargo.lock b/Cargo.lock index e177d97e3e..cdbb702154 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -777,6 +777,7 @@ version = "0.1.1" dependencies = [ "ant-logging", "ant-protocol", + "anyhow", "atomic-write-file", "chrono", "clap", diff --git a/ant-bootstrap/Cargo.toml b/ant-bootstrap/Cargo.toml index b71fecaec0..d83c737e26 100644 --- a/ant-bootstrap/Cargo.toml +++ b/ant-bootstrap/Cargo.toml @@ -36,6 +36,7 @@ wiremock = "0.5" tokio = { version = "1.0", features = ["full", "test-util"] } tracing-subscriber = { version = "0.3", features = ["env-filter"] } tempfile = "3.8.1" +anyhow = "1.0" [target.'cfg(target_arch = "wasm32")'.dependencies] wasmtimer = "0.2.0" diff --git a/ant-bootstrap/tests/cli_integration_tests.rs b/ant-bootstrap/tests/cli_integration_tests.rs index fb0a28e095..711e727ecd 100644 --- a/ant-bootstrap/tests/cli_integration_tests.rs +++ b/ant-bootstrap/tests/cli_integration_tests.rs @@ -10,16 +10,12 @@ use ant_bootstrap::{BootstrapCacheConfig, PeersArgs}; use ant_logging::LogBuilder; use anyhow::Result; use libp2p::Multiaddr; -use std::net::{IpAddr, Ipv4Addr}; -use std::path::PathBuf; use tempfile::TempDir; use wiremock::{ matchers::{method, path}, Mock, MockServer, ResponseTemplate, }; -// Use a private network IP instead of loopback for mDNS to work -const LOCAL_IP: IpAddr = IpAddr::V4(Ipv4Addr::new(192, 168, 1, 23)); async fn setup() -> (TempDir, BootstrapCacheConfig) { let temp_dir = TempDir::new().unwrap(); @@ -70,10 +66,18 @@ async fn test_peer_argument() -> Result<(), Box> { bootstrap_cache_dir: None, }; - let addrs = args.get_addrs(None, None).await?; - - assert_eq!(addrs.len(), 1, "Should have one addr"); - assert_eq!(addrs[0], peer_addr, "Should have the correct address"); + // When local feature is enabled, get_addrs returns empty list for local discovery + #[cfg(not(feature = "local"))] + { + let addrs = args.get_addrs(None, None).await?; + assert_eq!(addrs.len(), 1, "Should have one addr"); + assert_eq!(addrs[0], peer_addr, "Should have the correct address"); + } + #[cfg(feature = "local")] + { + let addrs = args.get_addrs(None, None).await?; + assert_eq!(addrs.len(), 0, "Should have no peers in local mode"); + } Ok(()) } @@ -105,12 +109,21 @@ async fn test_network_contacts_fallback() -> Result<(), Box Result<(), Box> { bootstrap_cache_dir: None, }; - let addrs = args.get_addrs(Some(config), None).await?; - - assert_eq!(addrs.len(), 1, "Should have exactly one test network peer"); - assert_eq!( - addrs[0], peer_addr, - "Should have the correct test network peer" - ); + // When local feature is enabled, get_addrs returns empty list for local discovery + #[cfg(not(feature = "local"))] + { + let addrs = args.get_addrs(Some(config), None).await?; + assert_eq!(addrs.len(), 1, "Should have exactly one test network peer"); + assert_eq!( + addrs[0], peer_addr, + "Should have the correct test network peer" + ); + } + #[cfg(feature = "local")] + { + let addrs = args.get_addrs(Some(config), None).await?; + assert_eq!(addrs.len(), 0, "Should have no peers in local mode"); + } Ok(()) } -#[test] -fn test_cli_add_peer() -> Result<()> { - let temp_dir = TempDir::new()?; - let cache_path = temp_dir.path().join("cache.txt"); - - let peer_addr = format!( - "/ip4/{}/udp/8080/quic-v1/p2p/12D3KooWRBhwfeP2Y4TCx1SM6s9rUoHhR5STiGwxBhgFRcw3UERE", - LOCAL_IP - ); - // ... rest of the test ... - Ok(()) -} -#[test] -fn test_cli_list_peers() -> Result<()> { - let temp_dir = TempDir::new()?; - let cache_path = temp_dir.path().join("cache.txt"); - let peer_addr = format!( - "/ip4/{}/udp/8080/quic-v1/p2p/12D3KooWRBhwfeP2Y4TCx1SM6s9rUoHhR5STiGwxBhgFRcw3UERE\n", - LOCAL_IP - ); - - // ... rest of the test ... - Ok(()) -} - -#[test] -fn test_cli_remove_peer() -> Result<()> { - let temp_dir = TempDir::new()?; - let cache_path = temp_dir.path().join("cache.txt"); - - let peer_addr = format!( - "/ip4/{}/udp/8080/quic-v1/p2p/12D3KooWRBhwfeP2Y4TCx1SM6s9rUoHhR5STiGwxBhgFRcw3UERE", - LOCAL_IP - ); - - // ... rest of the test ... - Ok(()) -} diff --git a/ant-cli/src/utils.rs b/ant-cli/src/utils.rs index 5f031a3c24..f1e2efe52b 100644 --- a/ant-cli/src/utils.rs +++ b/ant-cli/src/utils.rs @@ -7,6 +7,7 @@ // permissions and limitations relating to use of the SAFE Network Software. use autonomi::client::{Amount, ClientEvent, UploadSummary}; +use tracing::error; /// Collects upload summary from the event receiver. /// Send a signal to the returned sender to stop collecting and to return the result via the join handle. @@ -29,6 +30,8 @@ pub fn collect_upload_summary( tokens_spent += upload_summary.tokens_spent; record_count += upload_summary.record_count; } + Some(ClientEvent::PeerDiscovered(_)) => {} + Some(ClientEvent::PeerDisconnected(_)) => {} None => break, } } @@ -43,6 +46,8 @@ pub fn collect_upload_summary( tokens_spent += upload_summary.tokens_spent; record_count += upload_summary.record_count; } + ClientEvent::PeerDiscovered(_) => {} + ClientEvent::PeerDisconnected(_) => {} } } diff --git a/ant-networking/src/lib.rs b/ant-networking/src/lib.rs index 7008bc25ac..15c8b47f1a 100644 --- a/ant-networking/src/lib.rs +++ b/ant-networking/src/lib.rs @@ -9,7 +9,6 @@ #[macro_use] extern crate tracing; -use anyhow::anyhow; mod bootstrap; mod circular_vec; diff --git a/ant-networking/src/record_store.rs b/ant-networking/src/record_store.rs index b4ab4ff6b3..5245244ca1 100644 --- a/ant-networking/src/record_store.rs +++ b/ant-networking/src/record_store.rs @@ -1145,7 +1145,7 @@ mod tests { }; let self_id = PeerId::random(); let (network_event_sender, _) = mpsc::channel(1); - let (swarm_cmd_sender, _) = mpsc::channel(1); + let (swarm_cmd_sender, mut swarm_cmd_receiver) = mpsc::channel(1); let mut store = NodeRecordStore::with_config( self_id, @@ -1172,8 +1172,14 @@ mod tests { .put_verified(record.clone(), RecordType::Chunk) .is_ok()); - // Mark as stored (simulating the CompletedWrite event) - store.mark_as_stored(record.key.clone(), RecordType::Chunk); + // Wait for the async write to complete + if let Some(LocalSwarmCmd::AddLocalRecordAsStored { key, record_type }) = + swarm_cmd_receiver.recv().await + { + store.mark_as_stored(key, record_type); + } else { + panic!("Failed to receive AddLocalRecordAsStored command"); + } // Verify the chunk is stored let stored_record = store.get(&record.key); @@ -1219,12 +1225,12 @@ mod tests { if cfg!(feature = "encrypt-records") { assert!( store_diff.get(&record.key).is_none(), - "Chunk should be gone" + "Record should be removed when encryption enabled" ); } else { assert!( store_diff.get(&record.key).is_some(), - "Chunk shall persists without encryption" + "Record should be stored when encryption disabled" ); } diff --git a/autonomi/src/client/files/fs.rs b/autonomi/src/client/files/fs.rs index d89266dad5..fe1e76c0e9 100644 --- a/autonomi/src/client/files/fs.rs +++ b/autonomi/src/client/files/fs.rs @@ -15,13 +15,13 @@ // permissions and limitations relating to use of the SAFE Network Software. use super::archive::{PrivateArchive, PrivateArchiveAccess}; +use crate::client::files::get_relative_file_path_from_abs_file_and_folder_path; +use crate::client::utils::process_tasks_with_max_concurrency; use crate::client::{ - error::{CostError, GetError, PutError}, data::DataMapChunk, + error::{CostError, GetError, PutError}, Client, }; -use crate::client::files::get_relative_file_path_from_abs_file_and_folder_path; -use crate::client::utils::process_tasks_with_max_concurrency; use ant_evm::EvmWallet; use bytes::Bytes; use std::{path::PathBuf, sync::LazyLock}; @@ -84,24 +84,6 @@ pub enum FileCostError { WalkDir(#[from] walkdir::Error), } -impl From for DownloadError { - fn from(err: GetError) -> Self { - Self::GetError(err) - } -} - -impl From for UploadError { - fn from(err: PutError) -> Self { - Self::PutError(err) - } -} - -impl From for FileCostError { - fn from(err: CostError) -> Self { - Self::Cost(err) - } -} - impl Client { /// Download a private file from network to local file system pub async fn file_download( diff --git a/autonomi/src/client/mod.rs b/autonomi/src/client/mod.rs index 37c97e6c45..8e6c772ae8 100644 --- a/autonomi/src/client/mod.rs +++ b/autonomi/src/client/mod.rs @@ -37,7 +37,7 @@ pub mod wasm; mod rate_limiter; mod utils; -use ant_bootstrap::{BootstrapCacheConfig, BootstrapCacheStore}; +use ant_bootstrap::{BootstrapCacheConfig, BootstrapCacheStore, PeersArgs, ANT_PEERS_ENV}; pub use ant_evm::Amount; use ant_evm::{EvmNetwork, EvmWallet, EvmWalletError}; use ant_networking::{ @@ -144,11 +144,13 @@ impl Default for ClientConfig { impl Client { /// Initialize a new client with default configuration pub async fn init() -> Result { - Self::init_with_config(ClientConfig::default()).await + Self::init_with_config(ClientConfig::default()) + .await + .map_err(Into::into) } /// Initialize the network with the given config - pub async fn init_with_config(config: ClientConfig) -> Result { + pub async fn init_with_config(config: ClientConfig) -> Result { let keypair = Keypair::generate_ed25519(); let mut builder = NetworkBuilder::new(keypair); @@ -157,43 +159,41 @@ impl Client { builder = builder.local(true); } + let (network, _event_receiver, driver) = + builder.build_client().expect("Failed to build network"); + + // Spawn the driver to run in the background + ant_networking::target_arch::spawn(async move { + driver.run().await; + }); + + Ok(Self { + network, + client_event_sender: Arc::new(None), + evm_network: Default::default(), + mode: ClientMode::ReadOnly, + }) + } + /// Initialize a client that bootstraps from a list of peers. /// /// If any of the provided peers is a global address, the client will not be local. - /// - /// ```no_run - /// # use autonomi::Client; - /// # #[tokio::main] - /// # async fn main() -> Result<(), Box> { - /// // Will set `local` to true. - /// let client = Client::init_with_peers(vec!["/ip4/127.0.0.1/udp/1234/quic-v1".parse()?]).await?; - /// # Ok(()) - /// # } - /// ``` - pub async fn init_with_peers(peers: Vec) -> Result { + pub async fn init_with_peers(peers: Vec) -> Result { // Always use local mode for testing - Self::init_with_config(ClientConfig { + Self::init_with_config_and_peers(ClientConfig { local: true, peers: Some(peers), }) .await + .map_err(Into::into) } - /// Initialize the client with the given configuration. + /// Initialize the client with the given configuration and peers. /// /// This will block until [`CLOSE_GROUP_SIZE`] have been added to the routing table. /// /// See [`ClientConfig`]. - /// - /// ```no_run - /// use autonomi::client::Client; - /// # #[tokio::main] - /// # async fn main() -> Result<(), Box> { - /// let client = Client::init_with_config(Default::default()).await?; - /// # Ok(()) - /// # } - /// ``` - pub async fn init_with_config(config: ClientConfig) -> Result { + pub async fn init_with_config_and_peers(config: ClientConfig) -> Result { let (network, event_receiver) = build_client_and_run_swarm(config.local); let peers_args = PeersArgs { @@ -204,7 +204,7 @@ impl Client { let peers = match peers_args.get_addrs(None, None).await { Ok(peers) => peers, - Err(e) => return Err(e.into()), + Err(e) => return Err(ConnectError::Bootstrap(e)), }; let network_clone = network.clone(); @@ -215,14 +215,6 @@ impl Client { error!("Failed to dial addr={addr} with err: {err:?}"); }; } - } - - let (network, _event_receiver, driver) = - builder.build_client().expect("Failed to build network"); - - // Spawn the driver to run in the background - ant_networking::target_arch::spawn(async move { - driver.run().await; }); Ok(Self { @@ -259,15 +251,6 @@ impl Client { }) } - /// Initialize a new client with the given peers - pub async fn init_with_peers(peers: Vec) -> Result { - let config = ClientConfig { - peers: Some(peers), - ..Default::default() - }; - Self::init_with_config(config).await - } - /// Connect to the network. /// /// This will timeout after [`CONNECT_TIMEOUT_SECS`] secs. @@ -533,8 +516,9 @@ fn build_client_and_run_swarm(local: bool) -> (Network, mpsc::Receiver Result<()> { let _log_appender_guard = LogBuilder::init_single_threaded_tokio_test("dir_upload_download", false); - let client = Client::init_local().await?; + let client = Client::init_local(true).await?; let wallet = get_funded_wallet(); let addr = client @@ -81,7 +81,7 @@ fn compute_dir_sha256(dir: &str) -> Result { async fn file_into_vault() -> Result<()> { let _log_appender_guard = LogBuilder::init_single_threaded_tokio_test("file", false); - let client = Client::init_local().await?; + let client = Client::init_local(true).await?; let wallet = get_funded_wallet(); let client_sk = bls::SecretKey::random(); @@ -97,7 +97,7 @@ async fn file_into_vault() -> Result<()> { .await?; // now assert over the stored account packet - let new_client = Client::init_local().await?; + let new_client = Client::init_local(true).await?; let (ap, got_version) = new_client.fetch_and_decrypt_vault(&client_sk).await?; assert_eq!(set_version, got_version); diff --git a/autonomi/tests/linked_list.rs b/autonomi/tests/linked_list.rs index 645aa9673f..3f7a4ed538 100644 --- a/autonomi/tests/linked_list.rs +++ b/autonomi/tests/linked_list.rs @@ -10,7 +10,7 @@ use anyhow::{Context, Result}; use ant_logging::LogBuilder; use ant_networking::find_local_ip; use ant_protocol::storage::LinkedList; -use autonomi::{client::linked_list::TransactionError, Client, ClientConfig}; +use autonomi::{client::linked_list::LinkedListError, Client, ClientConfig}; use test_utils::evm::get_funded_wallet; use bls::SecretKey; use libp2p::Multiaddr; @@ -202,7 +202,7 @@ async fn test_linked_list() -> Result<()> { let res = client.linked_list_put(linked_list2.clone(), &wallet).await; assert!(matches!( res, - Err(TransactionError::TransactionAlreadyExists(address)) + Err(LinkedListError::LinkedListAlreadyExists(address)) if address == linked_list2.address() )); @@ -263,7 +263,7 @@ async fn test_linked_list_with_cost() -> Result<()> { assert!(matches!( res, - Err(TransactionError::TransactionAlreadyExists(address)) + Err(LinkedListError::LinkedListAlreadyExists(address)) if address == linked_list2.address() )); Ok(()) diff --git a/autonomi/tests/register.rs b/autonomi/tests/register.rs index 0709779d5c..f4c3757028 100644 --- a/autonomi/tests/register.rs +++ b/autonomi/tests/register.rs @@ -12,7 +12,7 @@ use ant_logging::LogBuilder; use autonomi::Client; use bytes::Bytes; -use eyre::Result; +use anyhow::Result; use rand::Rng; use std::time::Duration; use test_utils::evm::get_funded_wallet; @@ -22,7 +22,7 @@ use tokio::time::sleep; async fn register() -> Result<()> { let _log_appender_guard = LogBuilder::init_single_threaded_tokio_test("register", false); - let client = Client::init_local().await?; + let client = Client::init_local(true).await?; let wallet = get_funded_wallet(); // Owner key of the register.