cuprated: Blockchain Manager

binaries/cuprated/src/blockchain.rs

@@ -19,6 +19,7 @@ use cuprate_types::{
 
 use crate::constants::PANIC_CRITICAL_SERVICE_ERROR;
 
+mod chain_service;
 pub mod interface;
 mod manager;
 mod syncer;

binaries/cuprated/src/blockchain/chain_service.rs

@@ -0,0 +1,72 @@
+use std::task::{Context, Poll};
+
+use futures::{future::BoxFuture, FutureExt, TryFutureExt};
+use tower::Service;
+
+use cuprate_blockchain::service::BlockchainReadHandle;
+use cuprate_p2p::block_downloader::{ChainSvcRequest, ChainSvcResponse};
+use cuprate_types::blockchain::{BlockchainReadRequest, BlockchainResponse};
+
+/// That service that allows retrieving the chain state to give to the P2P crates, so we can figure out
+/// what blocks we need.
+///
+/// This has a more minimal interface than [`BlockchainReadRequest`] to make using the p2p crates easier.
+#[derive(Clone)]
+pub struct ChainService(pub BlockchainReadHandle);
+
+impl Service<ChainSvcRequest> for ChainService {
+    type Response = ChainSvcResponse;
+    type Error = tower::BoxError;
+    type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;
+
+    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        self.0.poll_ready(cx).map_err(Into::into)
+    }
+
+    fn call(&mut self, req: ChainSvcRequest) -> Self::Future {
+        let map_res = |res: BlockchainResponse| match res {
+            BlockchainResponse::CompactChainHistory {
+                block_ids,
+                cumulative_difficulty,
+            } => ChainSvcResponse::CompactHistory {
+                block_ids,
+                cumulative_difficulty,
+            },
+            BlockchainResponse::FindFirstUnknown(res) => ChainSvcResponse::FindFirstUnknown(res),
+            _ => unreachable!(),
+        };
+
+        match req {
+            ChainSvcRequest::CompactHistory => self
+                .0
+                .call(BlockchainReadRequest::CompactChainHistory)
+                .map_ok(map_res)
+                .map_err(Into::into)
+                .boxed(),
+            ChainSvcRequest::FindFirstUnknown(req) => self
+                .0
+                .call(BlockchainReadRequest::FindFirstUnknown(req))
+                .map_ok(map_res)
+                .map_err(Into::into)
+                .boxed(),
+            ChainSvcRequest::CumulativeDifficulty => self
+                .0
+                .call(BlockchainReadRequest::CompactChainHistory)
+                .map_ok(|res| {
+                    // TODO create a custom request instead of hijacking this one.
+                    // TODO: use the context cache.
+                    let BlockchainResponse::CompactChainHistory {
+                        cumulative_difficulty,
+                        ..
+                    } = res
+                    else {
+                        unreachable!()
+                    };
+
+                    ChainSvcResponse::CumulativeDifficulty(cumulative_difficulty)
+                })
+                .map_err(Into::into)
+                .boxed(),
+        }
+    }
+}

binaries/cuprated/src/blockchain/interface.rs

@@ -1,10 +1,10 @@
-//! The blockchain manger interface.
+//! The blockchain manager interface.
 //!
 //! This module contains all the functions to mutate the blockchain's state in any way, through the
-//! blockchain manger.
+//! blockchain manager.
 use std::{
     collections::{HashMap, HashSet},
-    sync::{Mutex, OnceLock},
+    sync::{LazyLock, Mutex, OnceLock},
 };
 
 use monero_serai::{block::Block, transaction::Transaction};
@@ -21,18 +21,15 @@ use cuprate_types::{
 };
 
 use crate::{
-    blockchain::manager::BlockchainManagerCommand, constants::PANIC_CRITICAL_SERVICE_ERROR,
+    blockchain::manager::{BlockchainManagerCommand, IncomingBlockOk},
+    constants::PANIC_CRITICAL_SERVICE_ERROR,
 };
 
-/// The channel used to send [`BlockchainManagerCommand`]s to the blockchain manger.
-pub static COMMAND_TX: OnceLock<mpsc::Sender<BlockchainManagerCommand>> = OnceLock::new();
-
-/// A [`HashSet`] of block hashes that the blockchain manager is currently handling.
+/// The channel used to send [`BlockchainManagerCommand`]s to the blockchain manager.
 ///
-/// This is used over something like a dashmap as we expect a lot of collisions in a short amount of
-/// time for new blocks so we would lose the benefit of sharded locks. A dashmap is made up of `RwLocks`
-/// which are also more expensive than `Mutex`s.
-pub static BLOCKS_BEING_HANDLED: OnceLock<Mutex<HashSet<[u8; 32]>>> = OnceLock::new();
+/// This channel is initialized in [`init_blockchain_manager`](super::manager::init_blockchain_manager), the functions
+/// in this file document what happens if this is not initialized when they are called.
+pub(super) static COMMAND_TX: OnceLock<mpsc::Sender<BlockchainManagerCommand>> = OnceLock::new();
 
 /// An error that can be returned from [`handle_incoming_block`].
 #[derive(Debug, thiserror::Error)]
@@ -52,10 +49,7 @@ pub enum IncomingBlockError {
 
 /// Try to add a new block to the blockchain.
 ///
-/// This returns a [`bool`] indicating if the block was added to the main-chain ([`true`]) or an alt-chain
-/// ([`false`]).
-///
-/// If we already knew about this block or the blockchain manger is not setup yet `Ok(false)` is returned.
+/// On success returns [`IncomingBlockOk`].
 ///
 /// # Errors
 ///
@@ -67,7 +61,17 @@ pub async fn handle_incoming_block(
     block: Block,
     given_txs: Vec<Transaction>,
     blockchain_read_handle: &mut BlockchainReadHandle,
-) -> Result<bool, IncomingBlockError> {
+) -> Result<IncomingBlockOk, IncomingBlockError> {
+    /// A [`HashSet`] of block hashes that the blockchain manager is currently handling.
+    ///
+    /// This lock prevents sending the same block to the blockchain manager from multiple connections
+    /// before one of them actually gets added to the chain, allowing peers to do other things.
+    ///
+    /// This is used over something like a dashmap as we expect a lot of collisions in a short amount of
+    /// time for new blocks, so we would lose the benefit of sharded locks. A dashmap is made up of `RwLocks`
+    /// 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 !block_exists(block.header.previous, blockchain_read_handle)
@@ -83,7 +87,7 @@ pub async fn handle_incoming_block(
         .await
         .expect(PANIC_CRITICAL_SERVICE_ERROR)
     {
-        return Ok(false);
+        return Ok(IncomingBlockOk::AlreadyHave);
     }
 
     // TODO: remove this when we have a working tx-pool.
@@ -105,20 +109,14 @@ pub async fn handle_incoming_block(
         .map_err(IncomingBlockError::InvalidBlock)?;
 
     let Some(incoming_block_tx) = COMMAND_TX.get() else {
-        // We could still be starting up the blockchain manger, so just return this as there is nothing
-        // else we can do.
-        return Ok(false);
+        // We could still be starting up the blockchain manager.
+        return Ok(IncomingBlockOk::NotReady);
     };
 
     // Add the blocks hash to the blocks being handled.
-    if !BLOCKS_BEING_HANDLED
-        .get_or_init(|| Mutex::new(HashSet::new()))
-        .lock()
-        .unwrap()
-        .insert(block_hash)
-    {
+    if !BLOCKS_BEING_HANDLED.lock().unwrap().insert(block_hash) {
         // If another place is already adding this block then we can stop.
-        return Ok(false);
+        return Ok(IncomingBlockOk::AlreadyHave);
     }
 
     // From this point on we MUST not early return without removing the block hash from `BLOCKS_BEING_HANDLED`.
@@ -140,12 +138,7 @@ pub async fn handle_incoming_block(
         .map_err(IncomingBlockError::InvalidBlock);
 
     // Remove the block hash from the blocks being handled.
-    BLOCKS_BEING_HANDLED
-        .get()
-        .unwrap()
-        .lock()
-        .unwrap()
-        .remove(&block_hash);
+    BLOCKS_BEING_HANDLED.lock().unwrap().remove(&block_hash);
 
     res
 }
@@ -161,7 +154,7 @@ async fn block_exists(
         .call(BlockchainReadRequest::FindBlock(block_hash))
         .await?
     else {
-        panic!("Invalid blockchain response!");
+        unreachable!();
     };
 
     Ok(chain.is_some())

binaries/cuprated/src/blockchain/manager.rs

@@ -24,9 +24,9 @@ use cuprate_types::{
 
 use crate::{
     blockchain::{
+        chain_service::ChainService,
         interface::COMMAND_TX,
         syncer,
-        types::ChainService,
         types::{ConcreteBlockVerifierService, ConsensusBlockchainReadHandle},
     },
     constants::PANIC_CRITICAL_SERVICE_ERROR,
@@ -35,13 +35,13 @@ use crate::{
 mod commands;
 mod handler;
 
-pub use commands::BlockchainManagerCommand;
+pub use commands::{BlockchainManagerCommand, IncomingBlockOk};
 
-/// Initialize the blockchain manger.
+/// Initialize the blockchain manager.
 ///
 /// This function sets up the [`BlockchainManager`] and the [`syncer`] so that the functions in [`interface`](super::interface)
 /// can be called.
-pub async fn init_blockchain_manger(
+pub async fn init_blockchain_manager(
     clearnet_interface: NetworkInterface<ClearNet>,
     blockchain_write_handle: BlockchainWriteHandle,
     blockchain_read_handle: BlockchainReadHandle,
@@ -73,10 +73,10 @@ pub async fn init_blockchain_manger(
         .await
         .expect(PANIC_CRITICAL_SERVICE_ERROR)
     else {
-        panic!("Blockchain context service returned wrong response!");
+        unreachable!()
     };
 
-    let manger = BlockchainManager {
+    let manager = BlockchainManager {
         blockchain_write_handle,
         blockchain_read_handle,
         blockchain_context_service,
@@ -86,7 +86,7 @@ pub async fn init_blockchain_manger(
         broadcast_svc: clearnet_interface.broadcast_svc(),
     };
 
-    tokio::spawn(manger.run(batch_rx, command_rx));
+    tokio::spawn(manager.run(batch_rx, command_rx));
 }
 
 /// The blockchain manager.

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

@@ -1,4 +1,4 @@
-//! This module contains the commands for th blockchain manager.
+//! This module contains the commands for the blockchain manager.
 use std::collections::HashMap;
 
 use monero_serai::block::Block;
@@ -15,6 +15,18 @@ pub enum BlockchainManagerCommand {
         /// All the transactions defined in [`Block::transactions`].
         prepped_txs: HashMap<[u8; 32], TransactionVerificationData>,
         /// The channel to send the response down.
-        response_tx: oneshot::Sender<Result<bool, anyhow::Error>>,
+        response_tx: oneshot::Sender<Result<IncomingBlockOk, anyhow::Error>>,
     },
 }
+
+/// The [`Ok`] response for an incoming block.
+pub enum IncomingBlockOk {
+    /// The block was added to the main-chain.
+    AddedToMainChain,
+    /// The blockchain manager is not ready yet.
+    NotReady,
+    /// The block was added to an alt-chain.
+    AddedToAltChain,
+    /// We already have the block.
+    AlreadyHave,
+}