From dcf34b97f3934dc70eb923485910792f1fe59ba0 Mon Sep 17 00:00:00 2001 From: Anirudh Prasad Date: Thu, 30 Jul 2026 20:11:32 +0530 Subject: [PATCH] fix: reject BatchProofQueue ingress collisions and fix commit accounting --- .../src/quorum_store/batch_proof_queue.rs | 46 ++++- consensus/src/quorum_store/counters.rs | 15 ++ .../tests/batch_proof_queue_test.rs | 179 ++++++++++++++++++ 3 files changed, 231 insertions(+), 9 deletions(-) diff --git a/consensus/src/quorum_store/batch_proof_queue.rs b/consensus/src/quorum_store/batch_proof_queue.rs index 01a16cef86c..a0fa6b9e75c 100644 --- a/consensus/src/quorum_store/batch_proof_queue.rs +++ b/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(); diff --git a/consensus/src/quorum_store/counters.rs b/consensus/src/quorum_store/counters.rs index 3299fc7e233..f5069b2399d 100644 --- a/consensus/src/quorum_store/counters.rs +++ b/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> = 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 = 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 = Lazy::new(|| { register_int_counter!( diff --git a/consensus/src/quorum_store/tests/batch_proof_queue_test.rs b/consensus/src/quorum_store/tests/batch_proof_queue_test.rs index 04fd143a0c9..15e6f79a25f 100644 --- a/consensus/src/quorum_store/tests/batch_proof_queue_test.rs +++ b/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); +}