Merge branch 'main' into hack

Author: Boog900

Cargo.lock

@@ -82,6 +82,7 @@ cuprate-rpc-interface       = { path = "rpc/interface"          ,default-feature
 anyhow                = { version = "1.0.89", default-features = false }
 async-trait           = { version = "0.1.82", default-features = false }
 bitflags              = { version = "2.6.0", default-features = false }
+blake3                = { version = "1", default-features = false }
 borsh                 = { version = "1.5.1", default-features = false }
 bytemuck              = { version = "1.18.0", default-features = false }
 bytes                 = { version = "1.7.2", default-features = false }

Cargo.toml

@@ -21,7 +21,7 @@ cuprate-levin            = { workspace = true }
 cuprate-wire             = { workspace = true }
 cuprate-p2p              = { workspace = true }
 cuprate-p2p-core         = { workspace = true }
-cuprate-dandelion-tower  = { workspace = true }
+cuprate-dandelion-tower  = { workspace = true, features = ["txpool"] }
 cuprate-async-buffer     = { workspace = true }
 cuprate-address-book     = { workspace = true }
 cuprate-blockchain       = { workspace = true, features = ["service"] }

binaries/cuprated/Cargo.toml

@@ -25,7 +25,7 @@ mod manager;
 mod syncer;
 mod types;
 
-use types::{
+pub use types::{
     ConcreteBlockVerifierService, ConcreteTxVerifierService, ConsensusBlockchainReadHandle,
 };
 

binaries/cuprated/src/blockchain.rs

@@ -8,17 +8,16 @@ use std::{
 };
 
 use monero_serai::{block::Block, transaction::Transaction};
-use rayon::prelude::*;
 use tokio::sync::{mpsc, oneshot};
 use tower::{Service, ServiceExt};
 
 use cuprate_blockchain::service::BlockchainReadHandle;
 use cuprate_consensus::transactions::new_tx_verification_data;
-use cuprate_helper::cast::usize_to_u64;
-use cuprate_types::{
-    blockchain::{BlockchainReadRequest, BlockchainResponse},
-    Chain,
+use cuprate_txpool::service::{
+    interface::{TxpoolReadRequest, TxpoolReadResponse},
+    TxpoolReadHandle,
 };
+use cuprate_types::blockchain::{BlockchainReadRequest, BlockchainResponse};
 
 use crate::{
     blockchain::manager::{BlockchainManagerCommand, IncomingBlockOk},
@@ -38,7 +37,7 @@ pub enum IncomingBlockError {
     ///
     /// The inner values are the block hash and the indexes of the missing txs in the block.
     #[error("Unknown transactions in block.")]
-    UnknownTransactions([u8; 32], Vec<u64>),
+    UnknownTransactions([u8; 32], Vec<usize>),
     /// We are missing the block's parent.
     #[error("The block has an unknown parent.")]
     Orphan,
@@ -59,8 +58,9 @@ pub enum IncomingBlockError {
 ///  - the block's parent is unknown
 pub async fn handle_incoming_block(
     block: Block,
-    given_txs: Vec<Transaction>,
+    mut given_txs: HashMap<[u8; 32], Transaction>,
     blockchain_read_handle: &mut BlockchainReadHandle,
+    txpool_read_handle: &mut TxpoolReadHandle,
 ) -> Result<IncomingBlockOk, IncomingBlockError> {
     /// A [`HashSet`] of block hashes that the blockchain manager is currently handling.
     ///
@@ -72,7 +72,12 @@ pub async fn handle_incoming_block(
     /// which are also more expensive than `Mutex`s.
     static BLOCKS_BEING_HANDLED: LazyLock<Mutex<HashSet<[u8; 32]>>> =
         LazyLock::new(|| Mutex::new(HashSet::new()));
-    // FIXME: we should look in the tx-pool for txs when that is ready.
+
+    if given_txs.len() > block.transactions.len() {
+        return Err(IncomingBlockError::InvalidBlock(anyhow::anyhow!(
+            "Too many transactions given for block"
+        )));
+    }
 
     if !block_exists(block.header.previous, blockchain_read_handle)
         .await
@@ -90,23 +95,36 @@ pub async fn handle_incoming_block(
         return Ok(IncomingBlockOk::AlreadyHave);
     }
 
-    // TODO: remove this when we have a working tx-pool.
-    if given_txs.len() != block.transactions.len() {
-        return Err(IncomingBlockError::UnknownTransactions(
-            block_hash,
-            (0..usize_to_u64(block.transactions.len())).collect(),
-        ));
-    }
+    let TxpoolReadResponse::TxsForBlock { mut txs, missing } = txpool_read_handle
+        .ready()
+        .await
+        .expect(PANIC_CRITICAL_SERVICE_ERROR)
+        .call(TxpoolReadRequest::TxsForBlock(block.transactions.clone()))
+        .await
+        .expect(PANIC_CRITICAL_SERVICE_ERROR)
+    else {
+        unreachable!()
+    };
 
-    // TODO: check we actually got given the right txs.
-    let prepped_txs = given_txs
-        .into_par_iter()
-        .map(|tx| {
-            let tx = new_tx_verification_data(tx)?;
-            Ok((tx.tx_hash, tx))
-        })
-        .collect::<Result<_, anyhow::Error>>()
-        .map_err(IncomingBlockError::InvalidBlock)?;
+    if !missing.is_empty() {
+        let needed_hashes = missing.iter().map(|index| block.transactions[*index]);
+
+        for needed_hash in needed_hashes {
+            let Some(tx) = given_txs.remove(&needed_hash) else {
+                // We return back the indexes of all txs missing from our pool, not taking into account the txs
+                // that were given with the block, as these txs will be dropped. It is not worth it to try to add
+                // these txs to the pool as this will only happen with a misbehaving peer or if the txpool reaches
+                // the size limit.
+                return Err(IncomingBlockError::UnknownTransactions(block_hash, missing));
+            };
+
+            txs.insert(
+                needed_hash,
+                new_tx_verification_data(tx)
+                    .map_err(|e| IncomingBlockError::InvalidBlock(e.into()))?,
+            );
+        }
+    }
 
     let Some(incoming_block_tx) = COMMAND_TX.get() else {
         // We could still be starting up the blockchain manager.
@@ -119,28 +137,37 @@ pub async fn handle_incoming_block(
         return Ok(IncomingBlockOk::AlreadyHave);
     }
 
-    // From this point on we MUST not early return without removing the block hash from `BLOCKS_BEING_HANDLED`.
+    // We must remove the block hash from `BLOCKS_BEING_HANDLED`.
+    let _guard = {
+        struct RemoveFromBlocksBeingHandled {
+            block_hash: [u8; 32],
+        }
+        impl Drop for RemoveFromBlocksBeingHandled {
+            fn drop(&mut self) {
+                BLOCKS_BEING_HANDLED
+                    .lock()
+                    .unwrap()
+                    .remove(&self.block_hash);
+            }
+        }
+        RemoveFromBlocksBeingHandled { block_hash }
+    };
 
     let (response_tx, response_rx) = oneshot::channel();
 
     incoming_block_tx
         .send(BlockchainManagerCommand::AddBlock {
             block,
-            prepped_txs,
+            prepped_txs: txs,
             response_tx,
         })
         .await
         .expect("TODO: don't actually panic here, an err means we are shutting down");
 
-    let res = response_rx
+    response_rx
         .await
         .expect("The blockchain manager will always respond")
-        .map_err(IncomingBlockError::InvalidBlock);
-
-    // Remove the block hash from the blocks being handled.
-    BLOCKS_BEING_HANDLED.lock().unwrap().remove(&block_hash);
-
-    res
+        .map_err(IncomingBlockError::InvalidBlock)
 }
 
 /// Check if we have a block with the given hash.

binaries/cuprated/src/blockchain/interface.rs

@@ -18,6 +18,7 @@ use cuprate_p2p::{
     BroadcastSvc, NetworkInterface,
 };
 use cuprate_p2p_core::ClearNet;
+use cuprate_txpool::service::TxpoolWriteHandle;
 use cuprate_types::{
     blockchain::{BlockchainReadRequest, BlockchainResponse},
     Chain, TransactionVerificationData,
@@ -46,6 +47,7 @@ pub async fn init_blockchain_manager(
     clearnet_interface: NetworkInterface<ClearNet>,
     blockchain_write_handle: BlockchainWriteHandle,
     blockchain_read_handle: BlockchainReadHandle,
+    txpool_write_handle: TxpoolWriteHandle,
     mut blockchain_context_service: BlockChainContextService,
     block_verifier_service: ConcreteBlockVerifierService,
     block_downloader_config: BlockDownloaderConfig,
@@ -80,6 +82,7 @@ pub async fn init_blockchain_manager(
     let manager = BlockchainManager {
         blockchain_write_handle,
         blockchain_read_handle,
+        txpool_write_handle,
         blockchain_context_service,
         cached_blockchain_context: blockchain_context.unchecked_blockchain_context().clone(),
         block_verifier_service,
@@ -102,6 +105,8 @@ pub struct BlockchainManager {
     blockchain_write_handle: BlockchainWriteHandle,
     /// A [`BlockchainReadHandle`].
     blockchain_read_handle: BlockchainReadHandle,
+    /// A [`TxpoolWriteHandle`].
+    txpool_write_handle: TxpoolWriteHandle,
     // TODO: Improve the API of the cache service.
     // TODO: rename the cache service -> `BlockchainContextService`.
     /// The blockchain context cache, this caches the current state of the blockchain to quickly calculate/retrieve

binaries/cuprated/src/blockchain/manager.rs

@@ -1,7 +1,10 @@
 //! The blockchain manager handler functions.
 use bytes::Bytes;
 use futures::{TryFutureExt, TryStreamExt};
-use monero_serai::{block::Block, transaction::Transaction};
+use monero_serai::{
+    block::Block,
+    transaction::{Input, Transaction},
+};
 use rayon::prelude::*;
 use std::ops::ControlFlow;
 use std::{collections::HashMap, sync::Arc};
@@ -17,16 +20,14 @@ use cuprate_consensus::{
 use cuprate_consensus_context::NewBlockData;
 use cuprate_helper::cast::usize_to_u64;
 use cuprate_p2p::{block_downloader::BlockBatch, constants::LONG_BAN, BroadcastRequest};
+use cuprate_txpool::service::interface::TxpoolWriteRequest;
 use cuprate_types::{
     blockchain::{BlockchainReadRequest, BlockchainResponse, BlockchainWriteRequest},
     AltBlockInformation, HardFork, TransactionVerificationData, VerifiedBlockInformation,
 };
 
-use crate::blockchain::manager::commands::IncomingBlockOk;
 use crate::{
-    blockchain::{
-        manager::commands::BlockchainManagerCommand, types::ConsensusBlockchainReadHandle,
-    },
+    blockchain::manager::commands::{BlockchainManagerCommand, IncomingBlockOk},
     constants::PANIC_CRITICAL_SERVICE_ERROR,
     signals::REORG_LOCK,
 };
@@ -434,6 +435,18 @@ impl super::BlockchainManager {
         &mut self,
         verified_block: VerifiedBlockInformation,
     ) {
+        // FIXME: this is pretty inefficient, we should probably return the KI map created in the consensus crate.
+        let spent_key_images = verified_block
+            .txs
+            .iter()
+            .flat_map(|tx| {
+                tx.tx.prefix().inputs.iter().map(|input| match input {
+                    Input::ToKey { key_image, .. } => key_image.compress().0,
+                    Input::Gen(_) => unreachable!(),
+                })
+            })
+            .collect::<Vec<[u8; 32]>>();
+
         self.blockchain_context_service
             .ready()
             .await
@@ -472,6 +485,14 @@ impl super::BlockchainManager {
         };
 
         self.cached_blockchain_context = blockchain_context.unchecked_blockchain_context().clone();
+
+        self.txpool_write_handle
+            .ready()
+            .await
+            .expect(PANIC_CRITICAL_SERVICE_ERROR)
+            .call(TxpoolWriteRequest::NewBlock { spent_key_images })
+            .await
+            .expect(PANIC_CRITICAL_SERVICE_ERROR);
     }
 }
 

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

@@ -1,11 +1,10 @@
 // FIXME: This whole module is not great and should be rewritten when the PeerSet is made.
-use std::{pin::pin, sync::Arc, time::Duration};
+use std::{sync::Arc, time::Duration};
 
 use futures::StreamExt;
-use tokio::time::interval;
 use tokio::{
     sync::{mpsc, Notify},
-    time::sleep,
+    time::interval,
 };
 use tower::{Service, ServiceExt};
 use tracing::instrument;

binaries/cuprated/src/blockchain/syncer.rs

@@ -1,13 +1,7 @@
-use std::task::{Context, Poll};
-
-use futures::future::BoxFuture;
-use futures::{FutureExt, TryFutureExt};
-use tower::{util::MapErr, Service};
+use tower::util::MapErr;
 
 use cuprate_blockchain::{cuprate_database::RuntimeError, service::BlockchainReadHandle};
 use cuprate_consensus::{BlockChainContextService, BlockVerifierService, TxVerifierService};
-use cuprate_p2p::block_downloader::{ChainSvcRequest, ChainSvcResponse};
-use cuprate_types::blockchain::{BlockchainReadRequest, BlockchainResponse};
 
 /// The [`BlockVerifierService`] with all generic types defined.
 pub type ConcreteBlockVerifierService = BlockVerifierService<

binaries/cuprated/src/blockchain/types.rs

@@ -2,4 +2,7 @@
 //!
 //! Will handle initiating the P2P and contains a protocol request handler.
 
+mod network_address;
 pub mod request_handler;
+
+pub use network_address::CrossNetworkInternalPeerId;