Exclude expired transactions when counting block size during block creation

consensus/consensus-types/src/common.rs

@@ -51,6 +51,36 @@ impl fmt::Display for TransactionSummary {
     }
 }
 
+#[derive(Clone, Copy, Debug, Eq, PartialEq, Deserialize, Serialize, Hash, Ord, PartialOrd)]
+pub struct TxnSummaryWithExpiration {
+    pub sender: AccountAddress,
+    pub sequence_number: u64,
+    pub expiration_timestamp_secs: u64,
+    pub hash: HashValue,
+}
+
+impl TxnSummaryWithExpiration {
+    pub fn new(
+        sender: AccountAddress,
+        sequence_number: u64,
+        expiration_timestamp_secs: u64,
+        hash: HashValue,
+    ) -> Self {
+        Self {
+            sender,
+            sequence_number,
+            expiration_timestamp_secs,
+            hash,
+        }
+    }
+}
+
+impl fmt::Display for TxnSummaryWithExpiration {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}:{}", self.sender, self.sequence_number,)
+    }
+}
+
 #[derive(Clone)]
 pub struct TransactionInProgress {
     pub gas_unit_price: u64,

consensus/consensus-types/src/request_response.rs

@@ -4,7 +4,7 @@
 use crate::common::{Payload, PayloadFilter};
 use anyhow::Result;
 use futures::channel::oneshot;
-use std::{fmt, fmt::Formatter};
+use std::{fmt, fmt::Formatter, time::Duration};
 
 pub enum GetPayloadCommand {
     /// Request to pull block to submit to consensus.
@@ -25,6 +25,8 @@ pub enum GetPayloadCommand {
         PayloadFilter,
         // callback to respond to
         oneshot::Sender<Result<GetPayloadResponse>>,
+        // block timestamp
+        Duration,
     ),
 }
 
@@ -40,11 +42,12 @@ impl fmt::Display for GetPayloadCommand {
                 return_non_full,
                 excluded,
                 _,
+                block_timestamp,
             ) => {
                 write!(
                     f,
-                    "GetPayloadRequest [max_txns: {}, max_txns_after_filtering: {}, max_bytes: {}, max_inline_txns: {}, max_inline_bytes:{}, return_non_full: {},  excluded: {}]",
-                    max_txns, max_txns_after_filtering, max_bytes, max_inline_txns, max_inline_bytes, return_non_full, excluded
+                    "GetPayloadRequest [max_txns: {}, max_txns_after_filtering: {}, max_bytes: {}, max_inline_txns: {}, max_inline_bytes:{}, return_non_full: {},  excluded: {}, block_timestamp: {:?}]",
+                    max_txns, max_txns_after_filtering, max_bytes, max_inline_txns, max_inline_bytes, return_non_full, excluded, block_timestamp
                 )
             },
         }

consensus/src/dag/dag_driver.rs

@@ -268,6 +268,7 @@ impl DagDriver {
                 false,
                 0,
                 0.0,
+                self.time_service.now_unix_time(),
             )
             .await
         {

consensus/src/liveness/proposal_generator.rs

@@ -376,6 +376,7 @@ impl ProposalGenerator {
                     pending_ordering,
                     pending_blocks.len(),
                     max_fill_fraction,
+                    timestamp,
                 )
                 .await
                 .context("Fail to retrieve payload")?;

consensus/src/payload_client/mixed.rs

@@ -78,6 +78,7 @@ impl PayloadClient for MixedPayloadClient {
         pending_ordering: bool,
         pending_uncommitted_blocks: usize,
         recent_max_fill_fraction: f32,
+        block_timestamp: Duration,
     ) -> anyhow::Result<(Vec<ValidatorTransaction>, Payload), QuorumStoreError> {
         // Pull validator txns first.
         let validator_txn_pull_timer = Instant::now();
@@ -124,6 +125,7 @@ impl PayloadClient for MixedPayloadClient {
                 pending_ordering,
                 pending_uncommitted_blocks,
                 recent_max_fill_fraction,
+                block_timestamp,
             )
             .await?;
 
@@ -165,6 +167,7 @@ async fn mixed_payload_client_should_prioritize_validator_txns() {
             false,
             0,
             0.,
+            aptos_infallible::duration_since_epoch(),
         )
         .await
         .unwrap()
@@ -189,6 +192,7 @@ async fn mixed_payload_client_should_prioritize_validator_txns() {
             false,
             0,
             0.,
+            aptos_infallible::duration_since_epoch(),
         )
         .await
         .unwrap()
@@ -213,6 +217,7 @@ async fn mixed_payload_client_should_prioritize_validator_txns() {
             false,
             0,
             0.,
+            aptos_infallible::duration_since_epoch(),
         )
         .await
         .unwrap()
@@ -237,6 +242,7 @@ async fn mixed_payload_client_should_prioritize_validator_txns() {
             false,
             0,
             0.,
+            aptos_infallible::duration_since_epoch(),
         )
         .await
         .unwrap()
@@ -279,6 +285,7 @@ async fn mixed_payload_client_should_respect_validator_txn_feature_flag() {
             false,
             0,
             0.,
+            aptos_infallible::duration_since_epoch(),
         )
         .await
         .unwrap()

consensus/src/payload_client/mod.rs

@@ -28,6 +28,7 @@ pub trait PayloadClient: Send + Sync {
         pending_ordering: bool,
         pending_uncommitted_blocks: usize,
         recent_max_fill_fraction: f32,
+        block_timestamp: Duration,
     ) -> anyhow::Result<(Vec<ValidatorTransaction>, Payload), QuorumStoreError>;
 
     fn trace_payloads(&self) {}

consensus/src/payload_client/user/mod.rs

@@ -27,6 +27,7 @@ pub trait UserPayloadClient: Send + Sync {
         pending_ordering: bool,
         pending_uncommitted_blocks: usize,
         recent_max_fill_fraction: f32,
+        block_timestamp: Duration,
     ) -> anyhow::Result<Payload, QuorumStoreError>;
 }
 
@@ -59,6 +60,7 @@ impl UserPayloadClient for DummyClient {
         _pending_ordering: bool,
         _pending_uncommitted_blocks: usize,
         _recent_max_fill_fraction: f32,
+        _block_timestamp: Duration,
     ) -> anyhow::Result<Payload, QuorumStoreError> {
         let timer = Instant::now();
         let mut nxt_txn_idx = 0;

consensus/src/payload_client/user/quorum_store_client.rs

@@ -52,6 +52,7 @@ impl QuorumStoreClient {
         max_inline_bytes: u64,
         return_non_full: bool,
         exclude_payloads: PayloadFilter,
+        block_timestamp: Duration,
     ) -> anyhow::Result<Payload, QuorumStoreError> {
         let (callback, callback_rcv) = oneshot::channel();
         let req = GetPayloadCommand::GetPayloadRequest(
@@ -63,6 +64,7 @@ impl QuorumStoreClient {
             return_non_full,
             exclude_payloads.clone(),
             callback,
+            block_timestamp,
         );
         // send to shared mempool
         self.consensus_to_quorum_store_sender
@@ -99,6 +101,7 @@ impl UserPayloadClient for QuorumStoreClient {
         pending_ordering: bool,
         pending_uncommitted_blocks: usize,
         recent_max_fill_fraction: f32,
+        block_timestamp: Duration,
     ) -> anyhow::Result<Payload, QuorumStoreError> {
         let return_non_full = recent_max_fill_fraction
             < self.wait_for_full_blocks_above_recent_fill_threshold
@@ -126,6 +129,7 @@ impl UserPayloadClient for QuorumStoreClient {
                     max_inline_bytes,
                     return_non_full || return_empty || done,
                     exclude.clone(),
+                    block_timestamp,
                 )
                 .await?;
             if payload.is_empty() && !return_empty && !done {

consensus/src/quorum_store/direct_mempool_quorum_store.rs

@@ -146,6 +146,7 @@ impl DirectMempoolQuorumStore {
                 return_non_full,
                 payload_filter,
                 callback,
+                _block_timestamp,
             ) => {
                 self.handle_block_request(
                     max_txns_after_filtering,

consensus/src/quorum_store/proof_manager.rs

@@ -11,7 +11,7 @@ use crate::{
     },
 };
 use aptos_consensus_types::{
-    common::{Payload, PayloadFilter, ProofWithData, TransactionSummary},
+    common::{Payload, PayloadFilter, ProofWithData, TxnSummaryWithExpiration},
     proof_of_store::{BatchInfo, ProofOfStore, ProofOfStoreMsg},
     request_response::{GetPayloadCommand, GetPayloadResponse},
 };
@@ -29,7 +29,7 @@ use std::{
 #[derive(Debug)]
 pub enum ProofManagerCommand {
     ReceiveProofs(ProofOfStoreMsg),
-    ReceiveBatches(Vec<(BatchInfo, Vec<TransactionSummary>)>),
+    ReceiveBatches(Vec<(BatchInfo, Vec<TxnSummaryWithExpiration>)>),
     CommitNotification(u64, Vec<BatchInfo>),
     Shutdown(tokio::sync::oneshot::Sender<()>),
 }
@@ -168,7 +168,7 @@ impl ProofManager {
 
     pub(crate) fn receive_batches(
         &mut self,
-        batch_summaries: Vec<(BatchInfo, Vec<TransactionSummary>)>,
+        batch_summaries: Vec<(BatchInfo, Vec<TxnSummaryWithExpiration>)>,
     ) {
         if self.allow_batches_without_pos_in_proposal {
             let batches = batch_summaries
@@ -212,6 +212,7 @@ impl ProofManager {
                 return_non_full,
                 filter,
                 callback,
+                block_timestamp,
             ) => {
                 let excluded_batches: HashSet<_> = match filter {
                     PayloadFilter::Empty => HashSet::new(),
@@ -228,6 +229,7 @@ impl ProofManager {
                         max_txns_after_filtering,
                         max_bytes,
                         return_non_full,
+                        block_timestamp,
                     );
 
                 counters::NUM_BATCHES_WITHOUT_PROOF_OF_STORE.observe(self.batch_queue.len() as f64);

consensus/src/quorum_store/tests/direct_mempool_quorum_store_test.rs

@@ -38,6 +38,7 @@ async fn test_block_request_no_txns() {
             true,
             PayloadFilter::DirectMempool(vec![]),
             consensus_callback,
+            aptos_infallible::duration_since_epoch(),
         ))
         .unwrap();
 

consensus/src/quorum_store/tests/proof_manager_test.rs

@@ -62,6 +62,7 @@ async fn get_proposal(
         true,
         PayloadFilter::InQuorumStore(filter_set),
         callback_tx,
+        aptos_infallible::duration_since_epoch(),
     );
     proof_manager.handle_proposal_request(req);
     let GetPayloadResponse::GetPayloadResponse(payload) = callback_rx.await.unwrap().unwrap();