[WIP] chore: ensure testnet launch fails if build fails

.github/workflows/merge.yml

@@ -97,10 +97,98 @@ jobs:
         run: cargo test --no-run --release
         timeout-minutes: 30
 
+      - name: Run network tests
+        timeout-minutes: 25
+        run: cargo test --release -p safenode -- network
+
       - name: Run protocol tests
         timeout-minutes: 25
         run: cargo test --release -p safenode -- protocol
 
       - name: Run storage tests
         timeout-minutes: 25
         run: cargo test --release -p safenode -- storage
+
+  e2e:
+    if: "!startsWith(github.event.head_commit.message, 'chore(release):')"
+    name: E2E tests
+    runs-on: ${{ matrix.os }}
+    strategy:
+      matrix:
+        os: [ubuntu-latest, windows-latest, macos-latest]
+    steps:
+      - uses: actions/checkout@v2
+
+      - name: Install Rust
+        id: toolchain
+        uses: actions-rs/toolchain@v1
+        with:
+          profile: minimal
+          toolchain: stable
+          override: true
+
+      - uses: Swatinem/rust-cache@v1
+        continue-on-error: true
+        with:
+          cache-on-failure: true
+          sharedKey: ${{github.run_id}}
+
+      - name: install ripgrep ubuntu
+        run: sudo apt-get install ripgrep
+        if: matrix.os == 'ubuntu-latest'
+
+      - name: install ripgrep mac
+        run: brew install ripgrep
+        if: matrix.os == 'macos-latest'
+
+      - name: install ripgrep windows
+        run: choco install ripgrep
+        if: matrix.os == 'windows-latest'
+
+      - name: Build sn bins
+        run: cargo build --release --bins
+        timeout-minutes: 30
+
+      - name: Start a local network
+        run: cargo run --release --bin testnet -- --interval 1 --node-path ./target/release/safenode
+        id: section-startup
+        env:
+          RUST_LOG: "safenode,safe=trace"
+        timeout-minutes: 10
+
+      - name: Start a client to carry out chunk actions
+        run: cargo run --release --bin safe -- --upload-chunks ./README.md --get-chunk confirm_uploaded
+        id: client-chunk-actions
+        env:
+          RUST_LOG: "safenode,safe=trace"
+        timeout-minutes: 2
+
+      - name: Start a client to carry out register actions
+        run: cargo run --release --bin safe -- --create-register myregister --query-register myregister
+        id: client-register-actions
+        env:
+          RUST_LOG: "safenode,safe=trace"
+        timeout-minutes: 2
+
+      - name: Kill all nodes
+        shell: bash
+        timeout-minutes: 1
+        if: failure()
+        continue-on-error: true
+        run: |
+          pkill safenode
+          echo "$(pgrep safenode | wc -l) nodes still running"
+
+      - name: Tar log files
+        shell: bash
+        continue-on-error: true
+        run: find ~/.safe/node/local-test-network -iname '*.log*' | tar -zcvf log_files.tar.gz --files-from -
+        if: failure()
+
+      - name: Upload Node Logs
+        uses: actions/upload-artifact@main
+        with:
+          name: sn_node_logs_e2e_${{matrix.os}}
+          path: log_files.tar.gz
+        if: failure()
+        continue-on-error: true

.gitignore

@@ -17,4 +17,7 @@ peers.json
 
 /.trunk
 
-settings.json
+settings.json
+
+# ignore log files
+*.log

README.md

@@ -1,4 +1,4 @@
-# stableset_net
+# The Safe Network
 
 This is the Safe Network as it was supposed to be, on a kademlia network, enabled by libp2p.
 
@@ -44,7 +44,10 @@ cargo run --release --example registers -- --user bob --reg-nickname myregister
 
 ### TODO
 
-- [ ] Basic messaging to target nodes
 - [ ] Add RPC for simplest node/net interaction (do libp2p CLIs help here?)
-- [ ] Add in chunking etc
-- [ ] Add in DBCs and validation handling
+
+
+
+### Archive
+
+The elder-membership agreed, section tree backed implementation of the safe network can be found [here](https://github.com/maidsafe/safe_network_archive)

safenode/src/bin/kadclient.rs

@@ -7,28 +7,37 @@
 // permissions and limitations relating to use of the SAFE Network Software.
 
 use safenode::{
-    client::{Client, ClientEvent, Error as ClientError, Files, WalletClient},
+    client::{Client, ClientEvent, Error as ClientError, Files},
     log::init_node_logging,
-    protocol::{address::ChunkAddress, wallet::LocalWallet},
+    protocol::{
+        address::ChunkAddress,
+        wallet::{DepositWallet, LocalWallet, Wallet},
+    },
 };
 
+use sn_dbc::Dbc;
+
 use bytes::Bytes;
 use clap::Parser;
 use dirs_next::home_dir;
 use eyre::Result;
 use std::{fs, path::PathBuf};
-use tracing::info;
+use tracing::{info, warn};
 use walkdir::WalkDir;
 use xor_name::XorName;
 
 #[derive(Parser, Debug)]
 #[clap(name = "safeclient cli")]
 struct Opt {
-    #[clap(long)]
-    client_dir: Option<PathBuf>,
-
     #[clap(long)]
     log_dir: Option<PathBuf>,
+    /// The location of the wallet file.
+    #[clap(long)]
+    wallet_dir: Option<PathBuf>,
+    /// Tries to load a hex encoded `Dbc` from the
+    /// given path and deposit it to the wallet.
+    #[clap(long)]
+    deposit: Option<PathBuf>,
 
     #[clap(long)]
     upload_chunks: Option<PathBuf>,
@@ -53,13 +62,9 @@ async fn main() -> Result<()> {
 
     info!("Instantiating a SAFE client...");
 
-    let client_dir = opt.client_dir.unwrap_or(get_client_dir().await?);
-    let wallet = LocalWallet::load_from(&client_dir).await?;
-
     let secret_key = bls::SecretKey::random();
     let client = Client::new(secret_key)?;
-    let file_api = Files::new(client.clone());
-    let _wallet_client = WalletClient::new(client.clone(), wallet);
+    let file_api: Files = Files::new(client.clone());
 
     let mut client_events_rx = client.events_channel();
     if let Ok(event) = client_events_rx.recv().await {
@@ -70,9 +75,80 @@ async fn main() -> Result<()> {
         }
     }
 
+    wallet(&opt).await?;
+    files(&opt, file_api).await?;
+    registers(&opt, client).await?;
+
+    Ok(())
+}
+
+async fn wallet(opt: &Opt) -> Result<()> {
+    let wallet_dir = opt.wallet_dir.clone().unwrap_or(get_client_dir().await?);
+    let mut wallet = LocalWallet::load_from(&wallet_dir).await?;
+
+    if let Some(deposit_path) = &opt.deposit {
+        let mut deposits = vec![];
+
+        for entry in WalkDir::new(deposit_path).into_iter().flatten() {
+            if entry.file_type().is_file() {
+                let file_name = entry.file_name();
+                info!("Reading deposited tokens from {file_name:?}.");
+                println!("Reading deposited tokens from {file_name:?}.");
+
+                let dbc_data = fs::read_to_string(entry.path())?;
+                let dbc = match Dbc::from_hex(dbc_data.trim()) {
+                    Ok(dbc) => dbc,
+                    Err(_) => {
+                        warn!(
+                            "This file does not appear to have valid hex-encoded DBC data. \
+                            Skipping it."
+                        );
+                        println!(
+                            "This file does not appear to have valid hex-encoded DBC data. \
+                            Skipping it."
+                        );
+                        continue;
+                    }
+                };
+
+                deposits.push(dbc);
+            }
+        }
+
+        let previous_balance = wallet.balance();
+        wallet.deposit(deposits);
+        let new_balance = wallet.balance();
+        let deposited = previous_balance.as_nano() - new_balance.as_nano();
+
+        if deposited > 0 {
+            if let Err(err) = wallet.store().await {
+                warn!("Failed to store deposited amount: {:?}", err);
+                println!("Failed to store deposited amount: {:?}", err);
+            } else {
+                info!("Deposited {:?}.", sn_dbc::Token::from_nano(deposited));
+                println!("Deposited {:?}.", sn_dbc::Token::from_nano(deposited));
+            }
+        } else {
+            info!("Nothing deposited.");
+            println!("Nothing deposited.");
+        }
+    }
+
+    Ok(())
+}
+
+async fn get_client_dir() -> Result<PathBuf> {
+    let mut home_dirs = home_dir().expect("A homedir to exist.");
+    home_dirs.push(".safe");
+    home_dirs.push("client");
+    tokio::fs::create_dir_all(home_dirs.as_path()).await?;
+    Ok(home_dirs)
+}
+
+async fn files(opt: &Opt, file_api: Files) -> Result<()> {
     let mut chunks_to_fetch = Vec::new();
 
-    if let Some(files_path) = opt.upload_chunks {
+    if let Some(files_path) = &opt.upload_chunks {
         for entry in WalkDir::new(files_path).into_iter().flatten() {
             if entry.file_type().is_file() {
                 let file = fs::read(entry.path())?;
@@ -95,7 +171,7 @@ async fn main() -> Result<()> {
         }
     }
 
-    if let Some(input_str) = opt.get_chunk {
+    if let Some(input_str) = &opt.get_chunk {
         println!("String passed in via get_chunk is {input_str}...");
         if input_str.len() == 64 {
             let vec = hex::decode(input_str).expect("Failed to decode xorname!");
@@ -115,7 +191,11 @@ async fn main() -> Result<()> {
         }
     }
 
-    if let Some(reg_nickname) = opt.create_register {
+    Ok(())
+}
+
+async fn registers(opt: &Opt, client: Client) -> Result<()> {
+    if let Some(reg_nickname) = &opt.create_register {
         let xorname = XorName::from_content(reg_nickname.as_bytes());
         let tag = 3006;
         println!("Creating Register with '{reg_nickname}' at xorname: {xorname:x} and tag {tag}");
@@ -130,7 +210,7 @@ async fn main() -> Result<()> {
             ),
         };
 
-        if let Some(entry) = opt.entry {
+        if let Some(entry) = &opt.entry {
             println!("Editing Register '{reg_nickname}' with: {entry}");
             match reg_replica.write(entry.as_bytes()).await {
                 Ok(()) => {}
@@ -169,11 +249,3 @@ async fn main() -> Result<()> {
 
     Ok(())
 }
-
-async fn get_client_dir() -> Result<PathBuf> {
-    let mut home_dirs = home_dir().expect("A homedir to exist.");
-    home_dirs.push(".safe");
-    home_dirs.push("client");
-    tokio::fs::create_dir_all(home_dirs.as_path()).await?;
-    Ok(home_dirs)
-}

safenode/src/bin/kadnode.rs

@@ -8,7 +8,6 @@
 
 use safenode::{
     log::init_node_logging,
-    network::Network,
     node::{Node, NodeEvent},
 };
 
@@ -28,12 +27,25 @@ async fn main() -> Result<()> {
     let _log_appender_guard = init_node_logging(&opt.log_dir)?;
 
     let socket_addr = SocketAddr::new(opt.ip, opt.port);
+    let peers = parse_peer_multiaddreses(&opt.peers)?;
 
     info!("Starting a node...");
-    let node_events_channel = Node::run(socket_addr).await?;
+    let node_events_channel = Node::run(socket_addr, peers).await?;
 
     let mut node_events_rx = node_events_channel.subscribe();
-    if let Ok(event) = node_events_rx.recv().await {
+
+    loop {
+        let event = match node_events_rx.recv().await {
+            Ok(e) => e,
+            Err(tokio::sync::broadcast::error::RecvError::Closed) => {
+                tracing::error!("Node event channel closed!");
+                break;
+            }
+            Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
+                tracing::warn!("Skipped {n} node events!");
+                continue;
+            }
+        };
         match event {
             NodeEvent::ConnectedToNetwork => {
                 info!("Connected to the Network");
@@ -62,18 +74,32 @@ struct Opt {
     /// Defaults to 0.0.0.0, which will bind to all network interfaces.
     #[clap(long, default_value_t = IpAddr::V4(Ipv4Addr::UNSPECIFIED))]
     ip: IpAddr,
+
+    /// Nodes we dial at start to help us get connected to the network. Can be specified multiple times.
+    #[clap(long = "peer")]
+    peers: Vec<Multiaddr>,
 }
 
-// Todo: Implement node bootstrapping to connect to peers from outside the local network
-#[allow(dead_code)]
-async fn bootstrap_node(network_api: &mut Network, addr: Multiaddr) -> Result<()> {
-    let peer_id = match addr.iter().last() {
-        Some(Protocol::P2p(hash)) => PeerId::from_multihash(hash).expect("Valid hash."),
-        _ => return Err(eyre!("Expect peer multiaddr to contain peer ID.")),
-    };
-    network_api
-        .dial(peer_id, addr)
-        .await
-        .expect("Dial to succeed");
-    Ok(())
+/// Parse multiaddresses containing the P2p protocol (`/p2p/<PeerId>`).
+/// Returns an error for the first invalid multiaddress.
+fn parse_peer_multiaddreses(multiaddrs: &[Multiaddr]) -> Result<Vec<(PeerId, Multiaddr)>> {
+    multiaddrs
+        .iter()
+        .map(|multiaddr| {
+            // Take hash from the `/p2p/<hash>` component.
+            let p2p_multihash = multiaddr
+                .iter()
+                .find_map(|p| match p {
+                    Protocol::P2p(hash) => Some(hash),
+                    _ => None,
+                })
+                .ok_or_else(|| eyre!("address does not contain `/p2p/<PeerId>`"))?;
+            // Parse the multihash into the `PeerId`.
+            let peer_id =
+                PeerId::from_multihash(p2p_multihash).map_err(|_| eyre!("invalid p2p PeerId"))?;
+
+            Ok((peer_id, multiaddr.clone()))
+        })
+        // Short circuit on the first error. See rust docs `Result::from_iter`.
+        .collect::<Result<Vec<(PeerId, Multiaddr)>>>()
 }