async fn commit_txns(
        &mut self,
        txns: &Self::Payload,
        compute_results: &StateComputeResult,
    ) -> Result<()> {
        let mut rejected_txns = vec![];
        for (txn, status) in txns.iter().zip(compute_results.compute_status().iter()) {
            if let TransactionStatus::Discard(_) = status {
                rejected_txns.push(CommittedTransaction {
                    sender: txn.sender(),
                    sequence_number: txn.sequence_number(),
                });
            }
        }

        if rejected_txns.is_empty() {
            return Ok(());
        }

        let (callback, callback_rcv) = oneshot::channel();
        let req = ConsensusRequest::RejectNotification(rejected_txns, callback);

        // send to shared mempool
        self.consensus_to_mempool_sender.clone().try_send(req)?;

        if let Err(e) = timeout(Duration::from_secs(1), callback_rcv).await {
            Err(format_err!("[consensus] txn manager did not receive ACK for commit notification sent to mempool on time: {:?}", e))
        } else {
            Ok(())
        }
    }
// Consensus notifies mempool of committed transactions that were rejected
Consensus notifies mempool of committed transactions that were rejected
Consensus notifies mempool of committed transactions that were rejected
async fn commit_txns(
        &mut self,
        txns: &Self::Payload,
        compute_results: &StateComputeResult,
    ) -> Result<()> {
        let mut rejected_txns = vec![];
        for (txn, status) in txns.iter().zip(compute_results.compute_status().iter()) {
            if let TransactionStatus::Discard(_) = status {
                rejected_txns.push(CommittedTransaction {
                    sender: txn.sender(),
                    sequence_number: txn.sequence_number(),
                });
            }
        }
        if rejected_txns.is_empty() {
            return Ok(());
        }
        let (callback, callback_rcv) = oneshot::channel();
        let req = ConsensusRequest::RejectNotification(rejected_txns, callback);
        // send to shared mempool
        self.consensus_to_mempool_sender.clone().try_send(req)?;
        if let Err(e) = timeout(Duration::from_secs(1), callback_rcv).await {
            Err(format_err!("[consensus] txn manager did not receive ACK for commit notification sent to mempool on time: {:?}", e))
        } else {
            Ok(())
        }
    }
async fn commit_txns ( & mut self , txns : & Self :: Payload , compute_results : & StateComputeResult , ) -> Result < ( ) > { let mut rejected_txns = vec ! [ ] ; for ( txn , status ) in txns . iter ( ) . zip ( compute_results . compute_status ( ) . iter ( ) ) { if let TransactionStatus :: Discard ( _ ) = status { rejected_txns . push ( CommittedTransaction { sender : txn . sender ( ) , sequence_number : txn . sequence_number ( ) , } ) ; } } if rejected_txns . is_empty ( ) { return Ok ( ( ) ) ; } let ( callback , callback_rcv ) = oneshot :: channel ( ) ; let req = ConsensusRequest :: RejectNotification ( rejected_txns , callback ) ; // send to shared mempool self . consensus_to_mempool_sender . clone ( ) . try_send ( req ) ? ; if let Err ( e ) = timeout ( Duration :: from_secs ( 1 ) , callback_rcv ) . await { Err ( format_err ! ( " " e ) ) } else { Ok ( ( ) ) } }
Consensus notifies mempool of committed transactions that were rejected
Consensus notifies mempool of committed transactions that were rejected
&mut self
txns
&Self::Payload
compute_results
&StateComputeResult
&mut self
txns
&Self::Payload
compute_results
&StateComputeResult