Merge branch 'blockprod-custom-txs'

blockprod/src/detail/mod.rs

@@ -30,9 +30,9 @@ use common::{
             block_body::BlockBody, signed_block_header::SignedBlockHeader,
             timestamp::BlockTimestamp, BlockCreationError, BlockHeader, BlockReward, ConsensusData,
         },
-        Block, ChainConfig, GenBlock, SignedTransaction,
+        Block, ChainConfig, GenBlock, SignedTransaction, Transaction,
     },
-    primitives::{BlockHeight, Id},
+    primitives::{Amount, BlockHeight, Id, Idable},
     time_getter::TimeGetter,
 };
 use consensus::{
@@ -42,7 +42,7 @@ use consensus::{
 use crypto::random::{make_true_rng, Rng};
 use logging::log;
 use mempool::{
-    tx_accumulator::{DefaultTxAccumulator, TransactionAccumulator},
+    tx_accumulator::{DefaultTxAccumulator, PackingStrategy, TransactionAccumulator},
     MempoolHandle,
 };
 use p2p::P2pHandle;
@@ -180,19 +180,31 @@ impl BlockProduction {
         &self,
         current_tip: Id<GenBlock>,
         min_block_timestamp: BlockTimestamp,
-    ) -> Result<Option<Box<dyn TransactionAccumulator>>, BlockProductionError> {
-        let max_block_size = self.chain_config.max_block_size_from_std_scripts();
+        transactions: Vec<SignedTransaction>,
+        transaction_ids: Vec<Id<Transaction>>,
+        packing_strategy: PackingStrategy,
+    ) -> Result<Box<dyn TransactionAccumulator>, BlockProductionError> {
+        let mut accumulator = Box::new(DefaultTxAccumulator::new(
+            self.chain_config.max_block_size_from_std_scripts(),
+            current_tip,
+            min_block_timestamp,
+        ));
+
+        for transaction in transactions.into_iter() {
+            let transaction_id = transaction.transaction().get_id();
+
+            accumulator
+                .add_tx(transaction, Amount::ZERO.into())
+                .map_err(|err| BlockProductionError::FailedToAddTransaction(transaction_id, err))?
+        }
+
         let returned_accumulator = self
             .mempool_handle
             .call(move |mempool| {
-                mempool.collect_txs(Box::new(DefaultTxAccumulator::new(
-                    max_block_size,
-                    current_tip,
-                    min_block_timestamp,
-                )))
+                mempool.collect_txs(accumulator, transaction_ids, packing_strategy)
             })
-            .await?
-            .map_err(|_| BlockProductionError::MempoolChannelClosed)?;
+            .await??;
+
         Ok(returned_accumulator)
     }
 
@@ -310,15 +322,26 @@ impl BlockProduction {
     pub async fn produce_block(
         &self,
         input_data: GenerateBlockInputData,
-        transactions_source: TransactionsSource,
+        transactions: Vec<SignedTransaction>,
+        transaction_ids: Vec<Id<Transaction>>,
+        packing_strategy: PackingStrategy,
     ) -> Result<(Block, oneshot::Receiver<usize>), BlockProductionError> {
-        self.produce_block_with_custom_id(input_data, transactions_source, None).await
+        self.produce_block_with_custom_id(
+            input_data,
+            transactions,
+            transaction_ids,
+            packing_strategy,
+            None,
+        )
+        .await
     }
 
     async fn produce_block_with_custom_id(
         &self,
         input_data: GenerateBlockInputData,
-        transactions_source: TransactionsSource,
+        transactions: Vec<SignedTransaction>,
+        transaction_ids: Vec<Id<Transaction>>,
+        packing_strategy: PackingStrategy,
         custom_id_maybe: Option<Vec<u8>>,
     ) -> Result<(Block, oneshot::Receiver<usize>), BlockProductionError> {
         if !self.blockprod_config.skip_ibd_check {
@@ -435,36 +458,25 @@ impl BlockProduction {
                 ));
             }
 
-            // TODO: see if we can simplify this
-            let transactions = match transactions_source.clone() {
-                TransactionsSource::Mempool => {
-                    // We conservatively use the minimum timestamp here in order to figure out
-                    // which transactions are valid for the block.
-                    // TODO: Alternatively, we can construct the transaction sequence from the
-                    // scratch every time a different timestamp is attempted. That is more costly
-                    // in terms of computational resources but will allow the node to include more
-                    // transactions since the passing time may release some time locks.
-                    let accumulator = self
-                        .collect_transactions(
-                            current_tip_index.block_id(),
-                            min_constructed_block_timestamp,
-                        )
-                        .await?;
-                    match accumulator {
-                        Some(acc) => acc.transactions().clone(),
-                        None => {
-                            // If the mempool rejects the accumulator (due to tip mismatch, or otherwise), we should start over and try again
-                            log::info!(
-                                "Mempool rejected the transaction accumulator. Restarting the block production attempt."
-                            );
-                            continue;
-                        }
-                    }
-                }
-                TransactionsSource::Provided(txs) => txs,
-            };
+            // We conservatively use the minimum timestamp here in order to figure out
+            // which transactions are valid for the block.
+            // TODO: Alternatively, we can construct the transaction sequence from
+            // scratch every time a different timestamp is attempted. That is more costly
+            // in terms of computational resources but will allow the node to include more
+            // transactions since the passing time may release some time locks.
+            let collected_transactions = self
+                .collect_transactions(
+                    current_tip_index.block_id(),
+                    min_constructed_block_timestamp,
+                    transactions.clone(),
+                    transaction_ids.clone(),
+                    packing_strategy,
+                )
+                .await?
+                .transactions()
+                .clone();
 
-            let block_body = BlockBody::new(block_reward, transactions);
+            let block_body = BlockBody::new(block_reward, collected_transactions);
 
             // A synchronous channel that sends only when the mining/staking is done
             let (ended_sender, ended_receiver) = mpsc::channel::<()>();