cuprated: add txpool manager (Cuprate#483)

* add txpool manager

* add version check for tx-pool

* add docs

* fix merge

* fix test

* fix merge

* fix cargo hack

* clean up imports

* fix typo

* add small buffer to rebroadcasts

* review fixes

* small clean up

* fmt

* fix merge

* review fixes

Author: Boog900

Cargo.lock

@@ -28,6 +28,7 @@ use crate::{
         types::ConsensusBlockchainReadHandle,
     },
     constants::PANIC_CRITICAL_SERVICE_ERROR,
+    txpool::TxpoolManagerHandle,
 };
 
 mod commands;
@@ -46,7 +47,7 @@ pub async fn init_blockchain_manager(
     clearnet_interface: NetworkInterface<ClearNet>,
     blockchain_write_handle: BlockchainWriteHandle,
     blockchain_read_handle: BlockchainReadHandle,
-    txpool_write_handle: TxpoolWriteHandle,
+    txpool_manager_handle: TxpoolManagerHandle,
     mut blockchain_context_service: BlockchainContextService,
     block_downloader_config: BlockDownloaderConfig,
 ) {
@@ -72,7 +73,7 @@ pub async fn init_blockchain_manager(
             blockchain_read_handle,
             BoxError::from,
         ),
-        txpool_write_handle,
+        txpool_manager_handle,
         blockchain_context_service,
         stop_current_block_downloader,
         broadcast_svc: clearnet_interface.broadcast_svc(),
@@ -93,8 +94,8 @@ pub struct BlockchainManager {
     blockchain_write_handle: BlockchainWriteHandle,
     /// A [`BlockchainReadHandle`].
     blockchain_read_handle: ConsensusBlockchainReadHandle,
-    /// A [`TxpoolWriteHandle`].
-    txpool_write_handle: TxpoolWriteHandle,
+
+    txpool_manager_handle: TxpoolManagerHandle,
     /// The blockchain context cache, this caches the current state of the blockchain to quickly calculate/retrieve
     /// values without needing to go to a [`BlockchainReadHandle`].
     blockchain_context_service: BlockchainContextService,

binaries/cuprated/src/blockchain/manager.rs

@@ -619,11 +619,8 @@ impl super::BlockchainManager {
             .await
             .expect(PANIC_CRITICAL_SERVICE_ERROR);
 
-        self.txpool_write_handle
-            .ready()
-            .await
-            .expect(PANIC_CRITICAL_SERVICE_ERROR)
-            .call(TxpoolWriteRequest::NewBlock { spent_key_images })
+        self.txpool_manager_handle
+            .new_block(spent_key_images)
             .await
             .expect(PANIC_CRITICAL_SERVICE_ERROR);
     }

binaries/cuprated/src/blockchain/manager/handler.rs

@@ -14,9 +14,12 @@ use cuprate_p2p::{block_downloader::BlockBatch, BroadcastSvc};
 use cuprate_p2p_core::handles::HandleBuilder;
 use cuprate_types::{CachedVerificationState, TransactionVerificationData, TxVersion};
 
-use crate::blockchain::{
-    check_add_genesis, manager::BlockchainManager, manager::BlockchainManagerCommand,
-    ConsensusBlockchainReadHandle,
+use crate::{
+    blockchain::{
+        check_add_genesis, manager::BlockchainManager, manager::BlockchainManagerCommand,
+        ConsensusBlockchainReadHandle,
+    },
+    txpool::TxpoolManagerHandle,
 };
 
 async fn mock_manager(data_dir: PathBuf) -> BlockchainManager {
@@ -30,7 +33,7 @@ async fn mock_manager(data_dir: PathBuf) -> BlockchainManager {
     let (mut blockchain_read_handle, mut blockchain_write_handle, _) =
         cuprate_blockchain::service::init(blockchain_config).unwrap();
     let (txpool_read_handle, txpool_write_handle, _) =
-        cuprate_txpool::service::init(txpool_config).unwrap();
+        cuprate_txpool::service::init(&txpool_config).unwrap();
 
     check_add_genesis(
         &mut blockchain_read_handle,
@@ -56,7 +59,7 @@ async fn mock_manager(data_dir: PathBuf) -> BlockchainManager {
     BlockchainManager {
         blockchain_write_handle,
         blockchain_read_handle,
-        txpool_write_handle,
+        txpool_manager_handle: TxpoolManagerHandle::mock(),
         blockchain_context_service,
         stop_current_block_downloader: Arc::new(Default::default()),
         broadcast_svc: BroadcastSvc::mock(),

binaries/cuprated/src/blockchain/manager/tests.rs

@@ -45,7 +45,7 @@ use fs::FileSystemConfig;
 use p2p::P2PConfig;
 use rayon::RayonConfig;
 pub use rpc::RpcConfig;
-use storage::StorageConfig;
+pub use storage::{StorageConfig, TxpoolConfig};
 use tokio::TokioConfig;
 use tor::TorConfig;
 use tracing_config::TracingConfig;

binaries/cuprated/src/config.rs

@@ -62,7 +62,7 @@ config_struct! {
     pub struct BlockchainConfig { }
 
     /// The tx-pool config.
-    #[derive(Debug, Deserialize, Serialize, PartialEq, Eq)]
+    #[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)]
     #[serde(deny_unknown_fields, default)]
     pub struct TxpoolConfig {
         /// The maximum size of the tx-pool.
@@ -71,6 +71,14 @@ config_struct! {
         /// Valid values | >= 0
         /// Examples     | 100_000_000, 50_000_000
         pub max_txpool_byte_size: usize,
+
+        /// The maximum age of transactions in the pool in seconds.
+        /// Transactions will be dropped after this time is reached.
+        ///
+        /// Type         | Number
+        /// Valid values | >= 0
+        /// Examples     | 100_000_000, 50_000_000
+        pub maximum_age_secs: u64,
     }
 }
 
@@ -79,6 +87,7 @@ impl Default for TxpoolConfig {
         Self {
             sync_mode: SyncMode::default(),
             max_txpool_byte_size: 100_000_000,
+            maximum_age_secs: 60 * 60 * 24,
         }
     }
 }

binaries/cuprated/src/config/storage.rs

@@ -92,7 +92,7 @@ fn main() {
         .expect(DATABASE_CORRUPT_MSG);
 
     let (txpool_read_handle, txpool_write_handle, _) =
-        cuprate_txpool::service::init_with_pool(config.txpool_config(), db_thread_pool)
+        cuprate_txpool::service::init_with_pool(&config.txpool_config(), db_thread_pool)
             .inspect_err(|e| error!("Txpool database error: {e}"))
             .expect(DATABASE_CORRUPT_MSG);
 
@@ -137,13 +137,15 @@ fn main() {
 
         // Create the incoming tx handler service.
         let tx_handler = IncomingTxHandler::init(
+            config.storage.txpool.clone(),
             network_interfaces.clearnet_network_interface.clone(),
             network_interfaces.tor_network_interface,
             txpool_write_handle.clone(),
             txpool_read_handle.clone(),
             context_svc.clone(),
             blockchain_read_handle.clone(),
-        );
+        )
+        .await;
 
         // Send tx handler sender to all network zones
         for zone in tx_handler_subscribers {
@@ -157,7 +159,7 @@ fn main() {
             network_interfaces.clearnet_network_interface,
             blockchain_write_handle,
             blockchain_read_handle.clone(),
-            txpool_write_handle.clone(),
+            tx_handler.txpool_manager.clone(),
             context_svc.clone(),
             config.block_downloader_config(),
         )

binaries/cuprated/src/main.rs

@@ -14,6 +14,7 @@ use monero_serai::{block::Block, transaction::Transaction};
 use tokio::sync::{broadcast, oneshot, watch};
 use tokio_stream::wrappers::WatchStream;
 use tower::{Service, ServiceExt};
+use tracing::instrument;
 
 use cuprate_blockchain::service::BlockchainReadHandle;
 use cuprate_consensus::{
@@ -22,10 +23,9 @@ use cuprate_consensus::{
 };
 use cuprate_dandelion_tower::TxState;
 use cuprate_fixed_bytes::ByteArrayVec;
-use cuprate_helper::cast::u64_to_usize;
 use cuprate_helper::{
     asynch::rayon_spawn_async,
-    cast::usize_to_u64,
+    cast::{u64_to_usize, usize_to_u64},
     map::{combine_low_high_bits_to_u128, split_u128_into_low_high_bits},
 };
 use cuprate_p2p::constants::{
@@ -363,6 +363,7 @@ async fn new_fluffy_block<A: NetZoneAddress>(
 }
 
 /// [`ProtocolRequest::NewTransactions`]
+#[instrument(level = "debug", skip_all, fields(txs = request.txs.len(), stem = !request.dandelionpp_fluff))]
 async fn new_transactions<A>(
     peer_information: PeerInformation<A>,
     request: NewTransactions,
@@ -373,16 +374,22 @@ where
     A: NetZoneAddress,
     InternalPeerID<A>: Into<CrossNetworkInternalPeerId>,
 {
+    tracing::debug!("handling new transactions");
+
     let context = blockchain_context_service.blockchain_context();
 
     // If we are more than 2 blocks behind the peer then ignore the txs - we are probably still syncing.
-    if usize_to_u64(context.chain_height + 2)
-        < peer_information
-            .core_sync_data
-            .lock()
-            .unwrap()
-            .current_height
-    {
+    let peer_height = peer_information
+        .core_sync_data
+        .lock()
+        .unwrap()
+        .current_height;
+    if usize_to_u64(context.chain_height + 2) < peer_height {
+        tracing::debug!(
+            our_height = context.chain_height,
+            peer_height,
+            "we are too far behind peer, ignoring txs."
+        );
         return Ok(ProtocolResponse::NA);
     }
 

binaries/cuprated/src/p2p/request_handler.rs

@@ -23,6 +23,7 @@ use cuprate_helper::{
     cast::{u32_to_usize, u64_to_usize, usize_to_u64},
     fmt::HexPrefix,
     map::split_u128_into_low_high_bits,
+    time::current_unix_timestamp,
 };
 use cuprate_hex::{Hex, HexVec};
 use cuprate_p2p_core::{client::handshaker::builder::DummyAddressBook, ClearNet, Network};
@@ -923,13 +924,15 @@ async fn get_transaction_pool_backlog(
     mut state: CupratedRpcHandler,
     _: GetTransactionPoolBacklogRequest,
 ) -> Result<GetTransactionPoolBacklogResponse, Error> {
+    let now = current_unix_timestamp();
+
     let backlog = txpool::backlog(&mut state.txpool_read)
         .await?
         .into_iter()
         .map(|entry| TxBacklogEntry {
-            weight: entry.weight,
+            weight: usize_to_u64(entry.weight),
             fee: entry.fee,
-            time_in_pool: entry.time_in_pool.as_secs(),
+            time_in_pool: now - entry.received_at,
         })
         .collect();
 
@@ -968,7 +971,7 @@ async fn get_miner_data(
         .into_iter()
         .map(|entry| GetMinerDataTxBacklogEntry {
             id: Hex(entry.id),
-            weight: entry.weight,
+            weight: usize_to_u64(entry.weight),
             fee: entry.fee,
         })
         .collect();

binaries/cuprated/src/rpc/handlers/json_rpc.rs

@@ -8,8 +8,10 @@ use cuprate_txpool::service::{TxpoolReadHandle, TxpoolWriteHandle};
 
 mod dandelion;
 mod incoming_tx;
+mod manager;
 mod relay_rules;
 mod txs_being_handled;
 
 pub use incoming_tx::{IncomingTxError, IncomingTxHandler, IncomingTxs};
+pub use manager::TxpoolManagerHandle;
 pub use relay_rules::RelayRuleError;