Paying to each holder in each group

.github/workflows/benchmark-prs.yml

@@ -88,8 +88,8 @@ jobs:
       - name: Check client memory usage
         shell: bash
         run: |
-          client_peak_mem_limit_mb="2000" # mb
-          client_avg_mem_limit_mb="700" # mb
+          client_peak_mem_limit_mb="2500" # mb
+          client_avg_mem_limit_mb="1500" # mb
 
           peak_mem_usage=$(
             rg '"memory_used_mb":[^,]*' $CLIENT_DATA_PATH/logs/safe.* -o --no-line-number --no-filename |

.github/workflows/memcheck.yml

@@ -104,12 +104,12 @@ jobs:
         timeout-minutes: 25
 
       - name: Chunks data integrity during nodes churn
-        run: cargo test --release -p sn_node --test data_with_churn -- --nocapture 
+        run: cargo test --release -p sn_node --test data_with_churn -- --nocapture
         env:
           TEST_DURATION_MINS: 15
           TEST_TOTAL_CHURN_CYCLES: 15
           SN_LOG: "all"
-        timeout-minutes: 20
+        timeout-minutes: 30
 
       - name: Verify restart of nodes using rg
         shell: bash
@@ -151,7 +151,7 @@ jobs:
           received_list_count=$(rg "Replicate list received from" $NODE_DATA_PATH -c --stats | \
             rg "(\d+) matches" | rg "\d+" -o)
           echo "Received $received_list_count replication lists"
-          fetching_attempt_count=$(rg "Fetching replication" $NODE_DATA_PATH -c --stats | \
+          fetching_attempt_count=$(rg "FetchingKeysForReplication" $NODE_DATA_PATH -c --stats | \
             rg "(\d+) matches" | rg "\d+" -o)
           echo "Carried out $fetching_attempt_count fetching attempts"
           replication_attempt_count=$(rg "Replicating chunk" $NODE_DATA_PATH -c --stats | \
@@ -206,7 +206,7 @@ jobs:
         # The memory usage here will be significantly higher here than in the benchmark test,
         # where we don't have a bootstrap node.
         run: |
-          node_peak_mem_limit_mb="160" # mb
+          node_peak_mem_limit_mb="200" # mb
 
           peak_mem_usage=$(
             rg '"memory_used_mb":[^,]*' $NODE_DATA_PATH/*/logs/* -o --no-line-number --no-filename | 
@@ -225,8 +225,8 @@ jobs:
       - name: Check client memory usage
         shell: bash
         run: |
-          client_peak_mem_limit_mb="2000" # mb
-          client_avg_mem_limit_mb="700" # mb
+          client_peak_mem_limit_mb="2500" # mb
+          client_avg_mem_limit_mb="1500" # mb
 
           peak_mem_usage=$(
             rg '"memory_used_mb":[^,]*' $CLIENT_DATA_PATH/logs/safe.* -o --no-line-number --no-filename | 

.github/workflows/merge.yml

@@ -262,14 +262,14 @@ jobs:
           fi
 
       - name: execute the dbc spend tests
-        run: cargo test --release --features="local-discovery" --test sequential_transfers -- --nocapture
+        run: cargo test --release -p sn_node --features="local-discovery" --test sequential_transfers -- --nocapture
         env:
           SN_LOG: "all"
           CARGO_TARGET_DIR: "./transfer-target"
         timeout-minutes: 25
 
       - name: execute the storage payment tests
-        run: cargo test --release --features="local-discovery" --test storage_payments -- --nocapture --test-threads=1
+        run: cargo test --release -p sn_node --features="local-discovery" --test storage_payments -- --nocapture --test-threads=1
         env:
           SN_LOG: "all"
           CARGO_TARGET_DIR: "./transfer-target"

sn_cli/Cargo.toml

@@ -41,7 +41,7 @@ sn_transfers = { path = "../sn_transfers", version = "0.10.28" }
 sn_logging = { path = "../sn_logging", version = "0.2.4" }
 sn_peers_acquisition= { path="../sn_peers_acquisition", version = "0.1.4" }
 sn_protocol = { path = "../sn_protocol", version = "0.5.3" }
-tokio = { version = "1.17.0", features = ["fs", "io-util", "macros", "parking_lot", "rt", "sync", "time"] }
+tokio = { version = "1.32.0", features = ["fs", "io-util", "macros", "parking_lot", "rt", "sync", "time"] }
 tracing = { version = "~0.1.26" }
 tracing-core = "0.1.30"
 url = "2.4.0"

sn_cli/src/subcommands/files.rs

@@ -13,7 +13,7 @@ use clap::Parser;
 use color_eyre::Result;
 use sn_client::{Client, Files};
 use sn_protocol::storage::{Chunk, ChunkAddress};
-use sn_transfers::wallet::PaymentProofsMap;
+use sn_transfers::client_transfers::ContentPaymentsMap;
 
 use std::{
     fs,
@@ -96,7 +96,7 @@ async fn upload_files(
     let file_names_path = root_dir.join("uploaded_files");
 
     // Payment shall always be verified.
-    let (chunks_to_upload, payment_proofs) =
+    let (chunks_to_upload, mut content_payments_map) =
         chunk_and_pay_for_storage(&client, root_dir, &files_path, true).await?;
 
     let mut chunks_to_fetch = Vec::new();
@@ -114,8 +114,14 @@ async fn upload_files(
             chunks.len()
         );
 
-        if let Err(error) =
-            upload_chunks(&file_api, &file_name, chunks, &payment_proofs, verify_store).await
+        if let Err(error) = upload_chunks(
+            &file_api,
+            &file_name,
+            chunks,
+            &mut content_payments_map,
+            verify_store,
+        )
+        .await
         {
             println!("Failed to store all chunks of file '{file_name}' to all nodes in the close group: {error}")
         } else {
@@ -140,7 +146,7 @@ async fn upload_chunks(
     file_api: &Files,
     file_name: &str,
     chunks_paths: Vec<(XorName, PathBuf)>,
-    payment_proofs: &PaymentProofsMap,
+    content_payments_map: &mut ContentPaymentsMap,
     verify_store: bool,
 ) -> Result<()> {
     let chunks_reader = chunks_paths
@@ -159,7 +165,7 @@ async fn upload_chunks(
         });
 
     file_api
-        .upload_chunks_in_batches(chunks_reader, payment_proofs, verify_store)
+        .upload_chunks_in_batches(chunks_reader, content_payments_map, verify_store)
         .await?;
     Ok(())
 }

sn_cli/src/subcommands/wallet.rs

@@ -8,7 +8,11 @@
 
 use sn_client::{Client, Files, WalletClient};
 use sn_dbc::Token;
-use sn_transfers::wallet::{parse_public_address, LocalWallet, PaymentProofsMap};
+use sn_protocol::storage::ChunkAddress;
+use sn_transfers::{
+    client_transfers::ContentPaymentsMap,
+    wallet::{parse_public_address, LocalWallet},
+};
 
 use bytes::Bytes;
 use clap::Parser;
@@ -236,13 +240,15 @@ pub(super) async fn chunk_and_pay_for_storage(
     root_dir: &Path,
     files_path: &Path,
     verify_store: bool,
-) -> Result<(BTreeMap<XorName, ChunkedFile>, PaymentProofsMap)> {
+) -> Result<(BTreeMap<XorName, ChunkedFile>, ContentPaymentsMap)> {
+    trace!("Starting to chunk_and_pay_for_storage");
     let wallet = LocalWallet::load_from(root_dir)
         .await
         .wrap_err("Unable to read wallet file in {path:?}")
         .suggestion(
             "If you have an old wallet file, it may no longer be compatible. Try removing it",
         )?;
+
     let mut wallet_client = WalletClient::new(client.clone(), wallet);
     let file_api: Files = Files::new(client.clone());
 
@@ -302,27 +308,25 @@ pub(super) async fn chunk_and_pay_for_storage(
         chunked_files.len()
     );
 
-    let (proofs, cost) = wallet_client
+    let (content_payments_map, cost) = wallet_client
         .pay_for_storage(
             chunked_files
                 .values()
                 .flat_map(|chunked_file| &chunked_file.chunks)
-                .map(|(name, _)| name),
+                .map(|(name, _)| {
+                    sn_protocol::NetworkAddress::ChunkAddress(ChunkAddress::new(*name))
+                }),
             verify_store,
         )
         .await?;
 
-    if let Some(cost) = cost {
-        let total_cost = proofs.len() as u64 * cost.as_nano();
-        println!(
-            "Successfully made payment of {total_cost} for {} records. (At a cost per record of {cost:?}.)",
-            proofs.len(),
-        );
-    } else {
-        println!("No payment needed for {} records.", proofs.len(),);
-    }
+    println!(
+        "Successfully made payment of {cost} for {} records. (At a cost per record of {cost:?}.)",
+        content_payments_map.len(),
+    );
 
     let wallet = wallet_client.into_wallet();
+
     if let Err(err) = wallet.store().await {
         println!("Failed to store wallet: {err:?}");
     } else {
@@ -333,5 +337,5 @@ pub(super) async fn chunk_and_pay_for_storage(
     }
 
     println!("Successfully paid for storage and generated the proofs. They can now be sent to the storage nodes when uploading paid chunks.");
-    Ok((chunked_files, proofs))
+    Ok((chunked_files, content_payments_map))
 }

sn_client/Cargo.toml

@@ -34,6 +34,6 @@ sn_registers = { path = "../sn_registers", version = "0.2.3" }
 sn_transfers = { path = "../sn_transfers", version = "0.10.28" }
 thiserror = "1.0.23"
 tiny-keccak = "~2.0.2"
-tokio = { version = "1.17.0", features = ["fs", "io-util", "macros", "parking_lot", "rt", "sync", "time"] }
+tokio = { version = "1.32.0", features = ["fs", "io-util", "macros", "parking_lot", "rt", "sync", "time"] }
 tracing = { version = "~0.1.26" }
 xor_name = "5.0.0"

sn_client/src/api.rs

@@ -14,11 +14,10 @@ use super::{
 use bls::{PublicKey, SecretKey, Signature};
 use indicatif::ProgressBar;
 use libp2p::{kad::Record, Multiaddr};
-use sn_dbc::{DbcId, SignedSpend, Token};
+use sn_dbc::{Dbc, DbcId, PublicAddress, SignedSpend, Token};
 use sn_networking::{multiaddr_is_global, NetworkEvent, SwarmDriver, CLOSE_GROUP_SIZE};
 use sn_protocol::{
     error::Error as ProtocolError,
-    messages::PaymentProof,
     storage::{
         try_deserialize_record, try_serialize_record, Chunk, ChunkAddress, ChunkWithPayment,
         DbcAddress, RecordHeader, RecordKind, RegisterAddress,
@@ -62,7 +61,6 @@ impl Client {
             signer,
             peers_added: 0,
             progress: Some(Self::setup_connection_progress()),
-            network_store_cost: 0,
         };
 
         // subscribe to our events channel first, so we don't have intermittent
@@ -145,6 +143,7 @@ impl Client {
                 }
             }
         }
+
         // The above loop breaks if `ConnectedToNetwork` is received, but we might need the
         // receiver to still be active for us to not get any error if any other event is sent
         let mut client_events_rx = client.events_channel();
@@ -273,7 +272,7 @@ impl Client {
     pub(super) async fn store_chunk(
         &self,
         chunk: Chunk,
-        payment: PaymentProof,
+        payment: Vec<Dbc>,
         verify_store: bool,
     ) -> Result<()> {
         info!("Store chunk: {:?}", chunk.address());
@@ -316,7 +315,6 @@ impl Client {
         let dbc_addr = DbcAddress::from_dbc_id(&dbc_id);
 
         trace!("Sending spend {dbc_id:?} to the network via put_record, with addr of {dbc_addr:?}");
-
         let key = NetworkAddress::from_dbc_address(dbc_addr).to_record_key();
         let record = Record {
             key,
@@ -401,33 +399,15 @@ impl Client {
     }
 
     /// Get the store cost at a given address
-    /// Replaces current network_store_cost with the new one, unless average is set to true
-    pub async fn get_store_cost_at_address(
-        &mut self,
-        address: NetworkAddress,
-        only_update_cost_if_higher: bool,
-    ) -> Result<Token> {
-        trace!("Getting store cost at {address:?}, will update only if higher cost?: {only_update_cost_if_higher:?}");
-
-        // if we're averaging over many samples across the network, any cost will do
-        let any_cost_will_do = only_update_cost_if_higher;
-
-        let cost = self
-            .network
-            .get_store_cost_from_network(address.clone(), any_cost_will_do)
-            .await?
-            .as_nano();
-
-        if cost > self.network_store_cost {
-            self.network_store_cost = cost;
-        }
-
-        if !only_update_cost_if_higher {
-            self.network_store_cost = cost;
-        }
-
-        trace!("Set store cost: {}", self.network_store_cost);
+    pub async fn get_store_costs_at_address(
+        &self,
+        address: &NetworkAddress,
+    ) -> Result<Vec<(PublicAddress, Token)>> {
+        trace!("Getting store cost at {address:?}");
 
-        Ok(Token::from_nano(cost))
+        Ok(self
+            .network
+            .get_store_costs_from_network(address.clone())
+            .await?)
     }
 }