chore(node): log record_store storing error

sn_client/src/register.rs

@@ -33,7 +33,7 @@ pub struct ClientRegister {
     #[debug(skip)]
     client: Client,
     pub(crate) register: Register,
-    ops: LinkedList<RegisterCmd>, // Cached operations.
+    pub ops: LinkedList<RegisterCmd>, // Cached operations.
 }
 
 impl ClientRegister {

sn_node/src/put_validation.rs

@@ -340,7 +340,7 @@ impl Node {
         // check register and merge if needed
         let updated_register = match self.register_validation(&register, present_locally).await? {
             Some(reg) => {
-                debug!("Register needed to be updated");
+                debug!("Register {pretty_key:?} needed to be updated");
                 reg
             }
             None => {
@@ -358,7 +358,7 @@ impl Node {
         };
         let content_hash = XorName::from_content(&record.value);
 
-        debug!("Storing register {reg_addr:?} as Record locally");
+        info!("Storing register {reg_addr:?} with content of {content_hash:?} as Record locally");
         self.network().put_local_record(record);
 
         self.record_metrics(Marker::ValidRegisterRecordPutFromNetwork(&pretty_key));

sn_node/src/replication.rs

@@ -17,7 +17,7 @@ use sn_protocol::{
     storage::RecordType,
     NetworkAddress, PrettyPrintRecordKey,
 };
-use tokio::task::{spawn, JoinHandle};
+use tokio::task::spawn;
 
 impl Node {
     /// Sends _all_ record keys every interval to all peers within the REPLICATE_RANGE.
@@ -38,7 +38,7 @@ impl Node {
         for (holder, key) in keys_to_fetch {
             let node = self.clone();
             let requester = NetworkAddress::from_peer(self.network().peer_id());
-            let _handle: JoinHandle<Result<()>> = spawn(async move {
+            let _handle = spawn(async move {
                 let pretty_key = PrettyPrintRecordKey::from(&key).into_owned();
                 debug!("Fetching record {pretty_key:?} from node {holder:?}");
                 let req = Request::Query(Query::GetReplicatedRecord {
@@ -76,18 +76,23 @@ impl Node {
                         target_record: None,
                         expected_holders: Default::default(),
                     };
-                    node.network()
-                        .get_record_from_network(key, &get_cfg)
-                        .await?
+                    match node.network().get_record_from_network(key, &get_cfg).await {
+                        Ok(record) => record,
+                        Err(err) => {
+                            error!("During replication fetch of {pretty_key:?}, failed in re-attempt of get from network {err:?}");
+                            return;
+                        }
+                    }
                 };
 
                 debug!(
                     "Got Replication Record {pretty_key:?} from network, validating and storing it"
                 );
-                node.store_replicated_in_record(record).await?;
-                debug!("Completed storing Replication Record {pretty_key:?} from network.");
-
-                Ok(())
+                if let Err(err) = node.store_replicated_in_record(record).await {
+                    error!("During store replication fetched {pretty_key:?}, got error {err:?}");
+                } else {
+                    debug!("Completed storing Replication Record {pretty_key:?} from network.");
+                }
             });
         }
         Ok(())

sn_node/tests/storage_payments.rs

@@ -282,19 +282,57 @@ async fn storage_payment_register_creation_succeeds() -> Result<()> {
         .create_and_pay_for_register(xor_name, &mut wallet_client, true, Permissions::default())
         .await?;
 
+    println!("Newly created register has {} ops", register.read().len());
+
     let retrieved_reg = client.get_register(address).await?;
 
     assert_eq!(register.read(), retrieved_reg.read());
 
     let random_entry = rng.gen::<[u8; 32]>().to_vec();
 
     register.write(&random_entry)?;
+
+    println!(
+        "Register has {} ops after first write",
+        register.read().len()
+    );
+
     register.sync(&mut wallet_client, true, None).await?;
 
     let retrieved_reg = client.get_register(address).await?;
 
     assert_eq!(retrieved_reg.read().iter().next().unwrap().1, random_entry);
 
+    assert_eq!(retrieved_reg.read().len(), 1);
+
+    for index in 1..10 {
+        println!("current index is {index}");
+        let random_entry = rng.gen::<[u8; 32]>().to_vec();
+
+        register.write(&random_entry)?;
+        register.sync(&mut wallet_client, true, None).await?;
+
+        let retrieved_reg = client.get_register(address).await?;
+
+        println!(
+            "current retrieved register entry length is {}",
+            retrieved_reg.read().len()
+        );
+        println!("current expected entry length is {}", register.read().len());
+
+        println!(
+            "current retrieved register ops length is {}",
+            retrieved_reg.ops.len()
+        );
+        println!("current local cached ops length is {}", register.ops.len());
+
+        assert_eq!(retrieved_reg.read().len(), register.read().len());
+
+        assert_eq!(retrieved_reg.read().iter().next().unwrap().1, random_entry);
+
+        std::thread::sleep(std::time::Duration::from_millis(1000));
+    }
+
     Ok(())
 }