feat(node): notify storage payment on receival

sn_networking/src/event/mod.rs

@@ -143,6 +143,13 @@ pub enum NetworkEvent {
         peer_id: PeerId,
         keys_to_verify: Vec<NetworkAddress>,
     },
+    /// Notification from peer claiming received a storage payment
+    StoragePaymentNotification {
+        spend_addr: NetworkAddress,
+        owner: String,
+        royalty: u64,
+        store_cost: u64,
+    },
 }
 
 /// Terminate node for the following reason
@@ -222,6 +229,17 @@ impl Debug for NetworkEvent {
                     "NetworkEvent::ChunkProofVerification({peer_id:?} {keys_to_verify:?})"
                 )
             }
+            NetworkEvent::StoragePaymentNotification {
+                spend_addr,
+                owner,
+                royalty,
+                store_cost,
+            } => {
+                write!(
+                    f,
+                    "NetworkEvent::StoragePaymentNotification({spend_addr:?} {owner} {store_cost} {royalty})"
+                )
+            }
         }
     }
 }

sn_networking/src/event/request_response.rs

@@ -100,6 +100,28 @@ impl SwarmDriver {
                                 error!("Received a bad_peer notification from {detected_by:?}, targeting {bad_peer:?}, which is not us.");
                             }
                         }
+                        Request::Cmd(sn_protocol::messages::Cmd::StoragePaymentReceived {
+                            spend_addr,
+                            owner,
+                            royalty,
+                            store_cost,
+                        }) => {
+                            let response = Response::Cmd(
+                                sn_protocol::messages::CmdResponse::StoragePaymentReceived(Ok(())),
+                            );
+                            self.swarm
+                                .behaviour_mut()
+                                .request_response
+                                .send_response(channel, response)
+                                .map_err(|_| NetworkError::InternalMsgChannelDropped)?;
+
+                            self.send_event(NetworkEvent::StoragePaymentNotification {
+                                spend_addr,
+                                owner,
+                                royalty,
+                                store_cost,
+                            })
+                        }
                         Request::Query(query) => {
                             self.send_event(NetworkEvent::QueryRequestReceived {
                                 query,

sn_networking/src/spends.rs

@@ -54,4 +54,49 @@ impl Network {
 
         Ok(())
     }
+
+    /// This function verifies a received notification of an owner's claim of storage payment
+    /// It fetches the correspondent spend, then carry out verification.
+    /// Once verified, the owner and amount will be reported to Node for further record.
+    pub async fn handle_storage_payment_notification(&self, spend_addr: NetworkAddress,
+                                owner: String,
+                                royalty: u64,
+                                store_cost: u64) {
+        let spend_address = if let Some(spend_address) = spend_addr.as_spend_address() {
+            spend_address
+        } else {
+            error!("When verify storage payment notification, cannot parse SpendAddress from {spend_addr:?}");
+            return;
+        };
+
+        let spend = match self.get_spend(spend_address).await {
+            Ok(spend) => spend,
+            Err(err) => {
+                error!("When verify storage payment notification, cannot get spend {spend_address:?}");
+                return;
+            }
+        };
+
+        // TODO: using proper CASH_NOTE_PURPOSE consts once PR 1653 merged
+        let royalty_keyword = "CASH_NOTE_REASON_FOR_NETWORK_ROYALTIES".to_string();
+        let change_keyword = "CASH_NOTE_REASON_FOR_CHANGE".to_string();
+
+        // 1, The spend's outputs shall have equal number of royalty and store_cost payments
+        // 2, The claimed payment shall be within the spend's outputs
+        let num_of_royalties = spend.spent_tx().outputs.iter().filter(|o| o.purpose == royalty_keyword.to_string()).count();
+        let num_of_change = spend.spent_tx().outputs.iter().filter(|o| o.purpose == change_keyword).count();
+        let num_of_store_cost = spend.spent_tx().outputs.len() - num_of_royalties - num_of_change;
+
+        let payments_match = num_of_store_cost == num_of_royalties;
+
+        let find_royalty = spend.spent_tx().outputs.iter().any(|o| o.purpose == royalty_keyword && o.amount.as_nano() == royalty);
+        let find_store_cost = spend.spent_tx().outputs.iter().any(|o| o.purpose == owner && o.amount.as_nano() == store_cost);
+
+        if payments_match && find_royalty && find_store_cost {
+                    self.events_channel
+                       .broadcast(NodeEvent::StoragePayments{owner, royalty, store_cost});
+        } else {
+            error!("Claimed storage payment of ({owner} {royalty} {store_cost}) cann't be verified by the spend {spend:?}");
+        }
+    }
 }

sn_node/src/bin/safenode/main.rs

@@ -335,6 +335,11 @@ You can check your reward balance by running:
 
 fn monitor_node_events(mut node_events_rx: NodeEventsReceiver, ctrl_tx: mpsc::Sender<NodeCtrl>) {
     let _handle = tokio::spawn(async move {
+        let mut spends_statistics: BTreeMap<String, Vec<u64>> = Default::default();
+
+        // TODO: use the proper CASH_NOTE_PURPOSE const
+        let royalty_reason = "ROYALTY".to_string();
+
         loop {
             match node_events_rx.recv().await {
                 Ok(NodeEvent::ConnectedToNetwork) => Marker::NodeConnectedToNetwork.log(),
@@ -366,6 +371,21 @@ fn monitor_node_events(mut node_events_rx: NodeEventsReceiver, ctrl_tx: mpsc::Se
                         break;
                     }
                 }
+                Ok(NodeEvent::StoragePayments{owner, royalty, store_cost}) => {
+                    {
+                        let holders = spends_statistics.entry(royalty_reason.clone()).or_default();
+                        holders.push(royalty);
+                    }
+                    {
+                        let holders = spends_statistics.entry(owner.clone()).or_default();
+                        holders = holders.push(store_cost);
+                    }
+                    let statistics = spends_statistics.iter().map(|(owner, payments)| {
+                        let total_amount: u64 = payments.iter().sum();
+                        (owner.clone(), payments.len(), total_amount)
+                    }).collect_vec();
+                    info!("Current storage statistics is: {statistics:?}");
+                }
                 Ok(event) => {
                     /* we ignore other events */
                     trace!("Currently ignored node event {event:?}");

sn_node/src/event.rs

@@ -68,6 +68,12 @@ pub enum NodeEvent {
     ChannelClosed,
     /// Terminates the node
     TerminateNode(String),
+    /// StoragePayment notification
+    StoragePayments {
+        owner: String,
+        royalty: u64,
+        store_cost: u64,
+    },
 }
 
 impl NodeEvent {

sn_node/src/node.rs

@@ -485,6 +485,23 @@ impl Node {
                     network.record_node_issues(peer_id, NodeIssue::FailedChunkProofCheck);
                 });
             }
+            NetworkEvent::StoragePaymentNotification{
+                                spend_addr,
+                                owner,
+                                royalty,
+                                store_cost,
+                            } => {
+                event_header = "StoragePaymentNotification";
+                let network = self.network.clone();
+                // Note: this log will be checked in CI
+                //       any change to the keyword `failed to fetch` shall incur
+                //       correspondent CI script change as well.
+                info!("Received StoragePaymentNotification, notifying owner {owner:?} received \
+                    {store_cost} tokens for store_cost and {royalty} tokens for royalty, with spend {spend_addr:?}");
+                let _handle = spawn(async move {
+                        network.handle_storage_payment_notification(spend_addr, owner, royalty, store_cost).await;
+                });
+            }
         }
 
         trace!(

sn_protocol/src/lib.rs

@@ -146,6 +146,14 @@ impl NetworkAddress {
         }
     }
 
+    /// Try to return the represented `SpendAddress`.
+    pub fn as_spend_address(&self) -> Option<SpendAddress> {
+        match self {
+            NetworkAddress::SpendAddress(spend_address) => Some(*spend_address),
+            _ => None,
+        }
+    }
+
     /// Return the convertable `RecordKey`.
     pub fn to_record_key(&self) -> RecordKey {
         match self {

sn_protocol/src/messages/cmd.rs

@@ -41,6 +41,14 @@ pub enum Cmd {
         bad_peer: NetworkAddress,
         bad_behaviour: String,
     },
+    /// Notify the peer the send received a storage payment
+    StoragePaymentReceived {
+        // Address of the spend that holding the outputs of the storage payment
+        spend_addr: NetworkAddress,
+        owner: String,
+        royalty: u64,
+        store_cost: u64,
+    },
 }
 
 impl std::fmt::Debug for Cmd {
@@ -69,6 +77,18 @@ impl std::fmt::Debug for Cmd {
                 .field("bad_peer", bad_peer)
                 .field("bad_behaviour", bad_behaviour)
                 .finish(),
+            Cmd::StoragePaymentReceived {
+                spend_addr,
+                owner,
+                royalty,
+                store_cost,
+            } => f
+                .debug_struct("Cmd::StoragePaymentReceived")
+                .field("spend_addr", spend_addr)
+                .field("owner", owner)
+                .field("royalty", royalty)
+                .field("store_cost", store_cost)
+                .finish(),
         }
     }
 }

sn_protocol/src/messages/response.rs

@@ -116,6 +116,11 @@ pub enum CmdResponse {
     //
     /// Response to the considered as bad notification
     PeerConsideredAsBad(Result<()>),
+    //
+    // ===== StoragePaymentReceived =====
+    //
+    /// Response to the notification of received a storage payment
+    StoragePaymentReceived(Result<()>),
 }
 
 /// The Ok variant of a CmdResponse