pos mempool: clean pivot-decision sets by committed height, not envelope hash

Pivot-decision transactions are signed over only the PivotBlockDecision payload, not the whole RawTransaction, so a validator can produce many valid, distinct-hash transactions for one vote by varying the unsigned chain_id. Deduplicating to one vote per (decision, sender) stops those duplicates from crashing proposal construction, but it makes commit cleanup leak: the aggregated pivot decision a proposer commits may wrap a different chain_id than the copy a node stored, so removing the committed set by its envelope hash could miss and leave the whole set (and every voter's per-sender slot, cap 128) until the 600s system TTL. A resident lower or conflicting same-height decision, which can never commit, was likewise never cleaned.

Clean by committed height instead: the commit notification carries the committed pivot decision's height, and on commit the store sweeps every pivot-decision set at or below it. Being hash-agnostic, this clears the committed decision regardless of which chain_id envelope was aggregated, and reclaims the obsolete lower / conflicting same-height sets in the same pass.

Key each decision's votes by sender (a map) so per-sender uniqueness is structural, which lets the manual admission dedup scan and the aggregation-time seen_signers guard go away.

Mempool-local: no change to signed bytes, hashing, or on-chain validation, so no consensus or fork risk. Whether chain_id should be brought into the pivot-decision signature is tracked separately.

Covered by a red-green regression test: committing a decision via a chain_id wrapper never stored locally still sweeps its set plus obsolete lower and conflicting same-height sets, while a higher decision survives.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: peilun-conflux

crates/cfxcore/core/src/pos/consensus/state_computer.rs

@@ -12,7 +12,8 @@ use consensus_types::block::Block;
 use diem_crypto::HashValue;
 use diem_logger::prelude::*;
 use diem_types::{
-    ledger_info::LedgerInfoWithSignatures, transaction::Transaction,
+    ledger_info::LedgerInfoWithSignatures,
+    transaction::{Transaction, TransactionPayload},
 };
 use executor_types::{
     BlockExecutor, Error as ExecutionError, StateComputeResult,
@@ -45,18 +46,22 @@ impl ExecutionProxy {
 
     /// Notify mempool of committed transactions so it can prune them.
     async fn notify_mempool(&self, committed_txns: Vec<Transaction>) {
-        let user_txns: Vec<CommittedTransaction> = committed_txns
-            .iter()
-            .filter_map(|txn| match txn {
-                Transaction::UserTransaction(signed_txn) => {
-                    Some(CommittedTransaction {
-                        sender: signed_txn.sender(),
-                        hash: signed_txn.hash(),
-                    })
+        let mut user_txns = Vec::new();
+        let mut committed_pivot_height: Option<u64> = None;
+        for txn in &committed_txns {
+            if let Transaction::UserTransaction(signed_txn) = txn {
+                user_txns.push(CommittedTransaction {
+                    sender: signed_txn.sender(),
+                    hash: signed_txn.hash(),
+                });
+                if let TransactionPayload::PivotDecision(d) =
+                    signed_txn.payload()
+                {
+                    committed_pivot_height =
+                        committed_pivot_height.max(Some(d.height));
                 }
-                _ => None,
-            })
-            .collect();
+            }
+        }
 
         if user_txns.is_empty() {
             return;
@@ -65,6 +70,7 @@ impl ExecutionProxy {
         let (callback, cb_receiver) = oneshot::channel();
         let notification = CommitNotification {
             transactions: user_txns,
+            committed_pivot_height,
             callback,
         };
 

crates/cfxcore/core/src/pos/mempool/core_mempool/mempool.rs

@@ -53,6 +53,11 @@ impl Mempool {
         self.transactions.commit_transaction(hash);
     }
 
+    /// Cleans up pivot-decision votes for a committed decision at `height`.
+    pub(crate) fn commit_pivot_height(&mut self, height: u64) {
+        self.transactions.commit_pivot_height(height);
+    }
+
     /// Used to add a transaction to the Mempool.
     /// Performs basic validation: checks account's sequence number.
     pub(crate) fn add_txn(
@@ -123,7 +128,7 @@ impl Mempool {
             // Admission accepts node_map (non-committee) voters, and
             // `check_voting_power` errors `UnknownAuthor` on them; filter to
             // the committee so one such vote can't fail a valid quorum.
-            let committee_votes: Vec<&(AccountAddress, HashValue)> =
+            let committee_votes: Vec<(AccountAddress, HashValue)> =
                 pivot_decision_set
                     .iter()
                     .filter(|(addr, _)| {
@@ -167,20 +172,17 @@ impl Mempool {
                 self.transactions.get_pivot_decisions(&pivot_decision_hash);
             let senders: Vec<AccountAddress> =
                 validators.get_ordered_account_addresses_iter().collect();
+            // Votes are keyed by sender, so signer indices are already
+            // distinct.
             let mut signatures = vec![];
-            let mut seen_signers = HashSet::new();
             for hash in &txn_hashes {
                 if let Some(txn) = self.transactions.get(hash) {
                     match txn.authenticator() {
                         TransactionAuthenticator::BLS { signature, .. } => {
                             if let Ok(index) =
                                 senders.binary_search(&txn.sender())
                             {
-                                // One slot per validator: guard the index so a
-                                // repeated signer can't abort aggregation.
-                                if seen_signers.insert(index) {
-                                    signatures.push((signature, index));
-                                }
+                                signatures.push((signature, index));
                             }
                         }
                         _ => unreachable!(),

crates/cfxcore/core/src/pos/mempool/core_mempool/transaction_store.rs

@@ -24,16 +24,38 @@ use diem_types::{
     transaction::{SignedTransaction, TransactionPayload},
 };
 use std::{
-    collections::{hash_map::Values, HashMap, HashSet},
+    collections::{hash_map::Values, HashMap},
     time::Duration,
 };
 
+/// Votes for one pivot decision, keyed by sender (one per sender), tagged with
+/// the decision `height` for the commit-time sweep.
+pub(crate) struct PivotDecisionVotes {
+    height: u64,
+    votes: HashMap<AccountAddress, HashValue>,
+}
+
+impl PivotDecisionVotes {
+    fn new(height: u64) -> Self {
+        Self {
+            height,
+            votes: HashMap::new(),
+        }
+    }
+
+    pub(crate) fn iter(
+        &self,
+    ) -> impl Iterator<Item = (AccountAddress, HashValue)> + '_ {
+        self.votes.iter().map(|(&addr, &hash)| (addr, hash))
+    }
+}
+
 /// TransactionStore is in-memory storage for all transactions in mempool.
 pub struct TransactionStore {
     // normal transactions
     transactions: AccountTransactions,
     // pivot decision helper structure
-    pivot_decisions: HashMap<HashValue, HashSet<(AccountAddress, HashValue)>>,
+    pivot_decisions: HashMap<HashValue, PivotDecisionVotes>,
 
     // Evicts txns after `system_transaction_timeout` so stalled commit
     // callbacks cannot clog the mempool indefinitely.
@@ -46,8 +68,7 @@ pub struct TransactionStore {
     capacity_per_sender: usize,
 }
 
-pub type PivotDecisionIter<'a> =
-    Values<'a, HashValue, HashSet<(AccountAddress, HashValue)>>;
+pub type PivotDecisionIter<'a> = Values<'a, HashValue, PivotDecisionVotes>;
 
 impl TransactionStore {
     pub(crate) fn new(config: &MempoolConfig) -> Self {
@@ -84,10 +105,7 @@ impl TransactionStore {
         &self, hash: &HashValue,
     ) -> Vec<HashValue> {
         if let Some(decisions) = self.pivot_decisions.get(hash) {
-            decisions
-                .iter()
-                .map(|(_, tx_hash)| tx_hash.clone())
-                .collect::<_>()
+            decisions.votes.values().copied().collect()
         } else {
             vec![]
         }
@@ -106,17 +124,16 @@ impl TransactionStore {
             return MempoolStatus::new(MempoolStatusCode::Accepted);
         }
 
-        // Only the payload is signed, so a validator could mint distinct-hash
-        // duplicates of one vote by varying `chain_id`; keep one per sender.
         if let TransactionPayload::PivotDecision(pivot_decision) =
             txn.txn.payload()
         {
+            // Only the payload is signed, so a validator could mint
+            // distinct-hash duplicates of one vote by varying `chain_id`; keep
+            // one per sender.
             let already_voted = self
                 .pivot_decisions
                 .get(&pivot_decision.hash())
-                .is_some_and(|set| {
-                    set.iter().any(|(addr, _)| *addr == address)
-                });
+                .is_some_and(|set| set.votes.contains_key(&address));
             if already_voted {
                 return MempoolStatus::new(MempoolStatusCode::Accepted);
             }
@@ -150,15 +167,14 @@ impl TransactionStore {
             txn.txn.payload()
         {
             let pivot_decision_hash = pivot_decision.hash();
-            self.pivot_decisions
+            let entry = self
+                .pivot_decisions
                 .entry(pivot_decision_hash)
-                .or_insert_with(HashSet::new);
-            if let Some(account_decision) =
-                self.pivot_decisions.get_mut(&pivot_decision_hash)
-            {
-                diem_debug!("txpool::insert pivot {:?}", hash);
-                account_decision.insert((address, hash));
-            }
+                .or_insert_with(|| {
+                    PivotDecisionVotes::new(pivot_decision.height)
+                });
+            diem_debug!("txpool::insert pivot {:?}", hash);
+            entry.votes.insert(address, hash);
             self.transactions.insert(hash, txn, true);
         } else {
             self.transactions.insert(hash, txn, false);
@@ -173,26 +189,42 @@ impl TransactionStore {
         MempoolStatus::new(MempoolStatusCode::Accepted)
     }
 
-    /// Handles transaction commit: deletes the transaction and cleans up
-    /// its entries in the timeline and TTL indexes.
+    /// Removes a stored transaction, routing it through the log and index
+    /// accounting — the invariant every `self.transactions` removal must
+    /// uphold. Returns the removed txn for callers that need it (e.g. GC).
+    fn remove_logged(
+        &mut self, hash: &HashValue, log: &mut TxnsLog,
+    ) -> Option<MempoolTransaction> {
+        let txn = self.transactions.remove(hash)?;
+        log.add(txn.get_sender(), txn.get_hash());
+        self.index_remove(&txn);
+        Some(txn)
+    }
+
+    /// Removes a committed transaction by hash.
     pub(crate) fn commit_transaction(&mut self, hash: HashValue) {
         let mut txns_log = TxnsLog::new();
-        if let Some(transaction) = self.transactions.remove(&hash) {
-            txns_log.add(transaction.get_sender(), transaction.get_hash());
-            self.index_remove(&transaction);
-            // handle pivot decision
-            let payload = transaction.txn.into_raw_transaction().into_payload();
-            if let TransactionPayload::PivotDecision(pivot_decision) = payload {
-                let pivot_decision_hash = pivot_decision.hash();
-                if let Some(indices) =
-                    self.pivot_decisions.remove(&pivot_decision_hash)
-                {
-                    for (_, hash) in indices {
-                        if let Some(txn) = self.transactions.remove(&hash) {
-                            txns_log.add(txn.get_sender(), txn.get_hash());
-                            self.index_remove(&txn);
-                        }
-                    }
+        self.remove_logged(&hash, &mut txns_log);
+        diem_debug!(LogSchema::new(LogEntry::CleanCommittedTxn).txns(txns_log));
+    }
+
+    /// Cleans up pivot-decision votes after a decision at `height` commits, by
+    /// sweeping every set at or below it — keyed on height, not `hash`, so it
+    /// survives a `chain_id` wrapper this node never stored, and it also
+    /// reclaims obsolete lower / conflicting same-height sets that can no
+    /// longer commit.
+    pub(crate) fn commit_pivot_height(&mut self, height: u64) {
+        let mut txns_log = TxnsLog::new();
+        let obsolete: Vec<HashValue> = self
+            .pivot_decisions
+            .iter()
+            .filter(|(_, v)| v.height <= height)
+            .map(|(k, _)| *k)
+            .collect();
+        for key in obsolete {
+            if let Some(entry) = self.pivot_decisions.remove(&key) {
+                for (_, tx_hash) in entry.votes {
+                    self.remove_logged(&tx_hash, &mut txns_log);
                 }
             }
         }
@@ -258,22 +290,19 @@ impl TransactionStore {
 
         let mut gc_txns_log = TxnsLog::new();
         for key in gc_txns.iter() {
-            if let Some(txn) = self.transactions.remove(&key.hash) {
+            if let Some(txn) = self.remove_logged(&key.hash, &mut gc_txns_log) {
                 let sender = txn.get_sender();
-                let tx_hash = txn.get_hash();
-                gc_txns_log.add(sender, tx_hash);
-                self.index_remove(&txn);
                 if let TransactionPayload::PivotDecision(pivot_decision) =
                     txn.txn.into_raw_transaction().into_payload()
                 {
                     // Drop only this entry: other validators' live votes
                     // share this set.
                     let pivot_decision_hash = pivot_decision.hash();
-                    if let Some(account_decision) =
+                    if let Some(entry) =
                         self.pivot_decisions.get_mut(&pivot_decision_hash)
                     {
-                        account_decision.remove(&(sender, tx_hash));
-                        if account_decision.is_empty() {
+                        entry.votes.remove(&sender);
+                        if entry.votes.is_empty() {
                             self.pivot_decisions.remove(&pivot_decision_hash);
                         }
                     }
@@ -343,6 +372,13 @@ mod tests {
         )
     }
 
+    fn pivot(height: u64, block_hash_byte: u8) -> PivotBlockDecision {
+        PivotBlockDecision {
+            block_hash: H256::from([block_hash_byte; 32]),
+            height,
+        }
+    }
+
     fn mk_pivot_txn(
         sk: &BLSPrivateKey, sender: AccountAddress,
         decision: &PivotBlockDecision, chain_id: u64,
@@ -435,6 +471,47 @@ mod tests {
         assert_eq!(store.per_sender_count[&sender_b], 1);
     }
 
+    /// Cleanup keys on committed height, not the envelope hash: committing d1
+    /// via a `chain_id` wrapper never stored here still clears d1's set, plus
+    /// every resident obsolete set (lower d0, conflicting same-height d2); a
+    /// higher d_hi survives, and a repeat sweep is a no-op.
+    #[test]
+    fn cleanup_by_committed_height() {
+        let mut store = store_with_cap(8);
+        let (sk0, _, s0) = new_sender();
+        let (sk1, _, s1) = new_sender();
+        let (sk2, _, s2) = new_sender();
+        let (sk_hi, _, s_hi) = new_sender();
+        let d0 = pivot(100, 10);
+        let d1 = pivot(120, 11);
+        let d2 = pivot(120, 12); // same height as d1, conflicting branch
+        let d_hi = pivot(121, 13);
+        for (sk, s, d) in [(&sk0, s0, &d0), (&sk1, s1, &d1), (&sk2, s2, &d2)] {
+            assert_eq!(
+                store.insert(mk_pivot_txn(sk, s, d, 1)).code,
+                MempoolStatusCode::Accepted
+            );
+        }
+        assert_eq!(
+            store.insert(mk_pivot_txn(&sk_hi, s_hi, &d_hi, 1)).code,
+            MempoolStatusCode::Accepted
+        );
+
+        // The network commits d1 via a chain_id=2 wrapper never stored here.
+        let committed_d1 = mk_pivot_txn(&sk1, s1, &d1, 2).get_hash();
+        assert!(store.get(&committed_d1).is_none());
+        store.commit_transaction(committed_d1);
+        store.commit_pivot_height(d1.height);
+        store.commit_pivot_height(d1.height); // repeat = no-op
+
+        for (d, s) in [(&d0, &s0), (&d1, &s1), (&d2, &s2)] {
+            assert!(store.get_pivot_decisions(&d.hash()).is_empty());
+            assert_eq!(store.per_sender_count.get(s), None);
+        }
+        assert_eq!(store.get_pivot_decisions(&d_hi.hash()).len(), 1);
+        assert_eq!(store.per_sender_count[&s_hi], 1);
+    }
+
     #[test]
     fn per_sender_count_insert_and_commit_lifecycle() {
         let mut store = store_with_cap(3);

crates/cfxcore/core/src/pos/mempool/shared_mempool/tasks.rs

@@ -240,7 +240,7 @@ pub(crate) async fn process_committed_transactions(
         LogEvent::Received
     )
     .state_sync_msg(&req));
-    commit_txns(&mempool, req.transactions).await;
+    commit_txns(&mempool, req.transactions, req.committed_pivot_height).await;
     if req.callback.send(Ok(CommitResponse::success())).is_err() {
         diem_error!(LogSchema::event_log(
             LogEntry::StateSyncCommit,
@@ -295,10 +295,14 @@ pub(crate) async fn process_consensus_request(
 
 async fn commit_txns(
     mempool: &Mutex<CoreMempool>, transactions: Vec<CommittedTransaction>,
+    committed_pivot_height: Option<u64>,
 ) {
     let mut pool = mempool.lock();
 
     for transaction in transactions {
         pool.remove_transaction(transaction.hash);
     }
+    if let Some(height) = committed_pivot_height {
+        pool.commit_pivot_height(height);
+    }
 }

crates/cfxcore/core/src/pos/mempool/shared_mempool/types.rs

@@ -153,6 +153,10 @@ pub struct ConsensusResponse {
 /// This notifies mempool to remove committed txns.
 pub struct CommitNotification {
     pub transactions: Vec<CommittedTransaction>,
+    /// Height of the committed pivot decision (max, if the batch spans
+    /// several); the mempool sweeps every pivot-decision set at or below it.
+    /// Kept per notification rather than per txn: it is a whole-commit fact.
+    pub committed_pivot_height: Option<u64>,
     pub callback: oneshot::Sender<Result<CommitResponse>>,
 }