[Quorum Store] check digest match, and log sender if mismatch (#9867)

consensus/src/network.rs

@@ -363,14 +363,15 @@ impl QuorumStoreSender for NetworkSender {
         recipient: Author,
         timeout: Duration,
     ) -> anyhow::Result<Batch> {
+        let request_digest = request.digest();
         let msg = ConsensusMsg::BatchRequestMsg(Box::new(request));
         let response = self
             .consensus_network_client
             .send_rpc(recipient, msg, timeout)
             .await?;
         match response {
             ConsensusMsg::BatchResponse(batch) => {
-                batch.verify()?;
+                batch.verify_with_digest(request_digest)?;
                 Ok(*batch)
             },
             _ => Err(anyhow!("Invalid batch response")),

consensus/src/quorum_store/batch_requester.rs

@@ -157,12 +157,10 @@ impl<T: QuorumStoreSender + Sync + 'static> BatchRequester<T> {
                         Some(response) = futures.next() => {
                             if let Ok(batch) = response {
                                 counters::RECEIVED_BATCH_RESPONSE_COUNT.inc();
-                                if batch.verify().is_ok() {
-                                    let digest = *batch.digest();
-                                    let payload = batch.into_transactions();
-                                    request_state.serve_request(digest, Some(payload));
-                                    return;
-                                }
+                                let digest = *batch.digest();
+                                let payload = batch.into_transactions();
+                                request_state.serve_request(digest, Some(payload));
+                                return;
                             }
                         },
                     }

consensus/src/quorum_store/tests/types_test.rs

@@ -6,8 +6,9 @@ use crate::quorum_store::{
     types::{Batch, BatchPayload, BatchRequest},
 };
 use aptos_consensus_types::proof_of_store::BatchId;
-use aptos_crypto::hash::CryptoHash;
+use aptos_crypto::{hash::CryptoHash, HashValue};
 use aptos_types::account_address::AccountAddress;
+use claims::{assert_err, assert_ok};
 
 #[test]
 fn test_batch() {
@@ -32,6 +33,10 @@ fn test_batch() {
         0,
     );
 
-    assert!(batch.verify().is_ok());
+    assert_ok!(batch.verify());
+    assert_ok!(batch.verify_with_digest(digest));
+    // verify should fail if the digest does not match.
+    assert_err!(batch.verify_with_digest(HashValue::random()));
+
     assert_eq!(batch.into_transactions(), signed_txns);
 }

consensus/src/quorum_store/types.rs

@@ -180,6 +180,16 @@ impl Batch {
         Ok(())
     }
 
+    /// Verify the batch, and that it matches the requested digest
+    pub fn verify_with_digest(&self, requested_digest: HashValue) -> anyhow::Result<()> {
+        ensure!(
+            requested_digest == *self.digest(),
+            "Response digest doesn't match the request"
+        );
+        self.verify()?;
+        Ok(())
+    }
+
     pub fn into_transactions(self) -> Vec<SignedTransaction> {
         self.payload.txns
     }