fix: reject BatchProofQueue ingress collisions and fix commit accounting

consensus/src/quorum_store/batch_proof_queue.rs

@@ -188,6 +188,20 @@ impl BatchProofQueue {
             return;
         }
 
+        // Reject a proof whose metadata collides with an already-queued entry for the same
+        // (author, batch_id) key. A colliding proof would leave item.info and item.proof
+        // inconsistent, enabling an underflow in the commit-accounting path.
+        if let Some(existing_item) = self.items.get(&batch_key) {
+            if existing_item.info != *proof.info() {
+                warn!(
+                    "insert_proof: rejecting PoS with colliding batch key, author: {}",
+                    proof.info().author().short_str()
+                );
+                counters::inc_rejected_pos_count(counters::POS_COLLISION_LABEL);
+                return;
+            }
+        }
+
         let author = proof.author();
         let bucket = proof.gas_bucket_start();
         let num_txns = proof.num_txns();
@@ -266,14 +280,23 @@ impl BatchProofQueue {
             let batch_sort_key = BatchSortKey::from_info(&batch_info);
             let batch_key = BatchKey::from_info(&batch_info);
 
-            // If the batch is either committed or the txn summary already exists, skip
-            // inserting this batch.
-            if self
-                .items
-                .get(&batch_key)
-                .is_some_and(|item| item.is_committed() || item.txn_summaries.is_some())
-            {
-                continue;
+            // Collision check must come before the committed/duplicate check so that
+            // collisions on already-committed or already-summarized slots are still
+            // counted in metrics.
+            if let Some(existing_item) = self.items.get(&batch_key) {
+                if existing_item.info != batch_info {
+                    warn!(
+                        "insert_batches: rejecting batch summary with colliding batch key, author: {}",
+                        batch_info.author().short_str()
+                    );
+                    counters::inc_rejected_batch_count(counters::BATCH_COLLISION_LABEL);
+                    continue;
+                }
+                // If the batch is either committed or the txn summary already exists, skip
+                // inserting this batch.
+                if existing_item.is_committed() || existing_item.txn_summaries.is_some() {
+                    continue;
+                }
             }
 
             self.author_to_batches
@@ -857,7 +880,12 @@ impl BatchProofQueue {
                         proof.gas_bucket_start(),
                         insertion_time.elapsed().as_secs_f64(),
                     );
-                    self.dec_remaining_proofs(&batch.author(), batch.num_txns());
+                    // Use the stored item's info rather than the caller-supplied batch arg so
+                    // that the decrement always matches the increment done at proof-insertion
+                    // time, even if a colliding BatchInfo reaches this path.
+                    let stored_num_txns = item.info.num_txns();
+                    let stored_author = item.info.author();
+                    self.dec_remaining_proofs(&stored_author, stored_num_txns);
                     counters::GARBAGE_COLLECTED_IN_PROOF_QUEUE_COUNTER
                         .with_label_values(&["committed_proof"])
                         .inc();

consensus/src/quorum_store/counters.rs

@@ -22,6 +22,8 @@ pub const CALLBACK_SUCCESS_LABEL: &str = "callback_success";
 
 pub const POS_EXPIRED_LABEL: &str = "expired";
 pub const POS_DUPLICATE_LABEL: &str = "duplicate";
+pub const POS_COLLISION_LABEL: &str = "collision";
+pub const BATCH_COLLISION_LABEL: &str = "batch_collision";
 
 static TRANSACTION_COUNT_BUCKETS: Lazy<Vec<f64>> = Lazy::new(|| {
     exponential_buckets(
@@ -628,6 +630,19 @@ pub fn inc_rejected_pos_count(reason: &str) {
     REJECTED_POS_COUNT.with_label_values(&[reason]).inc();
 }
 
+static REJECTED_BATCH_COUNT: Lazy<IntCounterVec> = Lazy::new(|| {
+    register_int_counter_vec!(
+        "quorum_store_rejected_batch_count",
+        "Count of the rejected batch summaries since last restart, grouped by reason.",
+        &["reason"]
+    )
+    .unwrap()
+});
+
+pub fn inc_rejected_batch_count(reason: &str) {
+    REJECTED_BATCH_COUNT.with_label_values(&[reason]).inc();
+}
+
 /// Count of the received batches since last restart.
 pub static RECEIVED_REMOTE_BATCH_COUNT: Lazy<IntCounter> = Lazy::new(|| {
     register_int_counter!(

consensus/src/quorum_store/tests/batch_proof_queue_test.rs

@@ -826,3 +826,182 @@ async fn test_proof_queue_pull_full_utilization() {
     proof_queue.handle_updated_block_timestamp(10);
     assert!(proof_queue.is_empty());
 }
+
+/// Build a BatchInfo with a deterministic (non-random) digest so two calls with the same
+/// `num_txns` produce the same BatchInfo, and calls with different `num_txns` produce
+/// different ones — giving us a proper collision: same (author, batch_id), different metadata.
+///
+/// Expiration is set to `u64::MAX` so that `gc_expired_batch_summaries_without_proofs` never
+/// evicts the entry during the test.
+fn batch_info_with_num_txns(
+    author: PeerId,
+    batch_id: BatchId,
+    gas_bucket_start: u64,
+    num_txns: u64,
+) -> BatchInfo {
+    let digest = HashValue::sha3_256_of(&num_txns.to_le_bytes());
+    BatchInfo::new(
+        author,
+        batch_id,
+        0,
+        u64::MAX, // far-future expiration — never GC'd by wall-clock
+        digest,
+        num_txns,
+        num_txns,
+        gas_bucket_start,
+    )
+}
+
+/// Regression test: a batch summary is inserted first, then a proof arrives for the same
+/// (author, batch_id) key but with different metadata.  The proof must be rejected.
+#[tokio::test]
+async fn test_proof_queue_rejects_batch_key_collision() {
+    let my_peer_id = PeerId::random();
+    let batch_store = batch_store_for_test(5 * 1024 * 1024);
+    let mut proof_queue = BatchProofQueue::new(my_peer_id, batch_store, 1);
+
+    let author = PeerId::random();
+    let batch_id = BatchId::new_for_test(0);
+
+    // Insert a batch summary with num_txns = 5.
+    let summary_info = batch_info_with_num_txns(author, batch_id, 100, 5);
+    proof_queue.insert_batches(vec![(summary_info, vec![])]);
+
+    // A proof for the same (author, batch_id) but num_txns = 100 must be rejected.
+    let colliding_proof = ProofOfStore::new(
+        batch_info_with_num_txns(author, batch_id, 100, 100),
+        AggregateSignature::empty(),
+    );
+    proof_queue.insert_proof(colliding_proof);
+
+    // The collision was rejected: no proof is tracked.
+    let (remaining_txns, remaining_proofs) = proof_queue.remaining_txns_and_proofs();
+    assert_eq!(remaining_proofs, 0, "colliding proof must not be counted");
+    assert_eq!(remaining_txns, 0);
+
+    // The batch summary is still present.
+    assert_eq!(proof_queue.batch_summaries_len(), 1);
+}
+
+/// Regression test: a proof arrives first, then a second proof for the same (author, batch_id)
+/// key arrives with different metadata.  The second proof must be rejected.
+#[tokio::test]
+async fn test_proof_queue_rejects_batch_key_collision_proof_first() {
+    let my_peer_id = PeerId::random();
+    let batch_store = batch_store_for_test(5 * 1024 * 1024);
+    let mut proof_queue = BatchProofQueue::new(my_peer_id, batch_store, 1);
+
+    let author = PeerId::random();
+    let batch_id = BatchId::new_for_test(1);
+
+    // Insert the legitimate proof (num_txns = 5).
+    let proof_v1 = ProofOfStore::new(
+        batch_info_with_num_txns(author, batch_id, 100, 5),
+        AggregateSignature::empty(),
+    );
+    proof_queue.insert_proof(proof_v1);
+
+    let (txns_after_first, proofs_after_first) = proof_queue.remaining_txns_and_proofs();
+    assert_eq!(proofs_after_first, 1);
+    assert_eq!(txns_after_first, 5);
+
+    // A second proof with the same (author, batch_id) but num_txns = 100 must be rejected.
+    let proof_v2 = ProofOfStore::new(
+        batch_info_with_num_txns(author, batch_id, 100, 100),
+        AggregateSignature::empty(),
+    );
+    proof_queue.insert_proof(proof_v2);
+
+    // Accounting must be unchanged — the colliding proof was dropped.
+    let (remaining_txns, remaining_proofs) = proof_queue.remaining_txns_and_proofs();
+    assert_eq!(remaining_proofs, 1, "only the original proof must be counted");
+    assert_eq!(remaining_txns, 5, "txn count must reflect original proof only");
+}
+
+/// Regression test: a proof is inserted first, then a batch summary for the same
+/// (author, batch_id) key arrives with different metadata.  The summary must be rejected.
+#[tokio::test]
+async fn test_proof_queue_rejects_batch_summary_collision() {
+    let my_peer_id = PeerId::random();
+    let batch_store = batch_store_for_test(5 * 1024 * 1024);
+    let mut proof_queue = BatchProofQueue::new(my_peer_id, batch_store, 1);
+
+    let author = PeerId::random();
+    let batch_id = BatchId::new_for_test(2);
+
+    // Insert the legitimate proof (num_txns = 5).
+    let proof = ProofOfStore::new(
+        batch_info_with_num_txns(author, batch_id, 100, 5),
+        AggregateSignature::empty(),
+    );
+    proof_queue.insert_proof(proof);
+
+    // A batch summary with the same key but num_txns = 100 must be rejected.
+    let colliding_summary = batch_info_with_num_txns(author, batch_id, 100, 100);
+    proof_queue.insert_batches(vec![(colliding_summary, vec![])]);
+
+    // Proof accounting must be unchanged.
+    let (remaining_txns, remaining_proofs) = proof_queue.remaining_txns_and_proofs();
+    assert_eq!(remaining_proofs, 1);
+    assert_eq!(remaining_txns, 5, "txn count must not be corrupted by colliding summary");
+
+    // No batch summary must have been recorded.
+    assert_eq!(proof_queue.batch_summaries_len(), 0);
+}
+
+/// Regression test: mark_committed must use the stored item's num_txns for accounting,
+/// not the num_txns from the caller-supplied BatchInfo.  A colliding BatchInfo with a
+/// larger num_txns previously caused an integer underflow / panic.
+#[tokio::test]
+async fn test_mark_committed_with_colliding_info_uses_stored_num_txns() {
+    let my_peer_id = PeerId::random();
+    let batch_store = batch_store_for_test(5 * 1024 * 1024);
+    let mut proof_queue = BatchProofQueue::new(my_peer_id, batch_store, 1);
+
+    let author = PeerId::random();
+    let batch_id = BatchId::new_for_test(3);
+
+    // Insert a proof with num_txns = 5; remaining counter becomes 5.
+    let proof = ProofOfStore::new(
+        batch_info_with_num_txns(author, batch_id, 100, 5),
+        AggregateSignature::empty(),
+    );
+    proof_queue.insert_proof(proof);
+
+    let (txns_before, proofs_before) = proof_queue.remaining_txns_and_proofs();
+    assert_eq!(proofs_before, 1);
+    assert_eq!(txns_before, 5);
+
+    // Call mark_committed with a BatchInfo for the same key but num_txns = 1_000_000.
+    // Before the fix this would underflow (5 - 1_000_000) and panic in debug mode.
+    let colliding_commit_info =
+        batch_info_with_num_txns(author, batch_id, 100, 1_000_000);
+    proof_queue.mark_committed(vec![colliding_commit_info]);
+
+    // The counter must have been decremented by the stored 5, not the caller's 1_000_000.
+    let (remaining_txns, remaining_proofs) = proof_queue.remaining_txns_and_proofs();
+    assert_eq!(remaining_proofs, 0);
+    assert_eq!(remaining_txns, 0, "accounting must use stored num_txns, not caller's");
+}
+
+/// Exact retransmits (same key, same metadata) must still be accepted as a no-op.
+#[tokio::test]
+async fn test_proof_queue_accepts_identical_resend() {
+    let my_peer_id = PeerId::random();
+    let batch_store = batch_store_for_test(5 * 1024 * 1024);
+    let mut proof_queue = BatchProofQueue::new(my_peer_id, batch_store, 1);
+
+    let author = PeerId::random();
+    let batch_id = BatchId::new_for_test(4);
+    let info = batch_info_with_num_txns(author, batch_id, 100, 7);
+
+    let proof1 = ProofOfStore::new(info.clone(), AggregateSignature::empty());
+    let proof2 = ProofOfStore::new(info, AggregateSignature::empty());
+
+    proof_queue.insert_proof(proof1);
+    proof_queue.insert_proof(proof2); // identical resend — must not panic or double-count
+
+    let (remaining_txns, remaining_proofs) = proof_queue.remaining_txns_and_proofs();
+    assert_eq!(remaining_proofs, 1, "duplicate proof must not be double-counted");
+    assert_eq!(remaining_txns, 7);
+}