tell the txpool about incoming blocks

binaries/cuprated/src/blockchain/manager.rs

@@ -36,6 +36,7 @@ mod commands;
 mod handler;
 
 pub use commands::{BlockchainManagerCommand, IncomingBlockOk};
+use cuprate_txpool::service::TxpoolWriteHandle;
 
 /// Initialize the blockchain manager.
 ///
@@ -45,6 +46,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,
@@ -79,6 +81,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,
@@ -101,6 +104,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/handler.rs

@@ -1,6 +1,7 @@
 //! The blockchain manager handler functions.
 use bytes::Bytes;
 use futures::{TryFutureExt, TryStreamExt};
+use monero_serai::transaction::Input;
 use monero_serai::{block::Block, transaction::Transaction};
 use rayon::prelude::*;
 use std::ops::ControlFlow;
@@ -17,6 +18,7 @@ use cuprate_consensus::{
 };
 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,
@@ -434,6 +436,19 @@ 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]>>();
+        let blockchain_height = verified_block.height + 1;
+
         self.blockchain_context_service
             .ready()
             .await
@@ -472,6 +487,17 @@ 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,
+                blockchain_height,
+            })
+            .await
+            .expect(PANIC_CRITICAL_SERVICE_ERROR);
     }
 }
 

binaries/cuprated/src/txpool/incoming_tx.rs

@@ -271,9 +271,8 @@ async fn handle_valid_tx(
         return;
     };
 
-    // TODO: check blockchain for double spends to prevent a race condition.
+    // TODO: There is a race condition possible if a tx and block come in at the same time <https://github.com/Cuprate/cuprate/issues/314>.
 
-    // TODO: fill this in properly.
     let incoming_tx = incoming_tx
         .with_routing_state(state)
         .with_state_in_db(None)

storage/txpool/src/service/interface.rs

@@ -1,9 +1,12 @@
 //! Tx-pool [`service`](super) interface.
 //!
 //! This module contains `cuprate_txpool`'s [`tower::Service`] request and response enums.
+use std::{
+    collections::{HashMap, HashSet},
+    sync::Arc,
+};
+
 use cuprate_types::TransactionVerificationData;
-use std::collections::HashMap;
-use std::{collections::HashSet, sync::Arc};
 
 use crate::types::{KeyImage, TransactionBlobHash, TransactionHash};
 
@@ -38,8 +41,11 @@ pub enum TxpoolReadResponse {
         /// The tx hashes of the blob hashes that were known but were in the stem pool.
         stem_pool_hashes: Vec<TransactionHash>,
     },
+    /// The response for [`TxpoolReadRequest::TxsForBlock`].
     TxsForBlock {
+        /// The txs we had in the txpool.
         txs: HashMap<[u8; 32], TransactionVerificationData>,
+        /// The indexes of the missing txs.
         missing: Vec<usize>,
     },
 }
@@ -68,9 +74,11 @@ pub enum TxpoolWriteRequest {
     ///
     /// Returns [`TxpoolWriteResponse::Ok`].
     Promote(TransactionHash),
-
+    /// Tell the tx-pool about a new block.
     NewBlock {
+        /// The new blockchain height.
         blockchain_height: usize,
+        /// The spent key images in the new block.
         spent_key_images: Vec<KeyImage>,
     },
 }