cuprated: tx-pool sync

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

@@ -34,7 +34,6 @@ use cuprate_types::{
 use crate::{
     blockchain::manager::commands::{BlockchainManagerCommand, IncomingBlockOk},
     constants::PANIC_CRITICAL_SERVICE_ERROR,
-    signals::REORG_LOCK,
 };
 
 impl super::BlockchainManager {
@@ -400,8 +399,6 @@ impl super::BlockchainManager {
         &mut self,
         top_alt_block: AltBlockInformation,
     ) -> Result<(), anyhow::Error> {
-        let _guard = REORG_LOCK.write().await;
-
         let BlockchainResponse::AltBlocksInChain(mut alt_blocks) = self
             .blockchain_read_handle
             .ready()
@@ -611,6 +608,8 @@ impl super::BlockchainManager {
         self.add_valid_block_to_blockchain_cache(&verified_block)
             .await;
 
+        let block_hash = verified_block.block_hash;
+
         self.blockchain_write_handle
             .ready()
             .await
@@ -620,7 +619,7 @@ impl super::BlockchainManager {
             .expect(PANIC_CRITICAL_SERVICE_ERROR);
 
         self.txpool_manager_handle
-            .new_block(spent_key_images)
+            .new_block(spent_key_images, block_hash)
             .await
             .expect(PANIC_CRITICAL_SERVICE_ERROR);
     }

binaries/cuprated/src/main.rs

@@ -48,7 +48,6 @@ mod killswitch;
 mod logging;
 mod p2p;
 mod rpc;
-mod signals;
 mod statics;
 mod tor;
 mod txpool;

binaries/cuprated/src/rpc/service/blockchain.rs

@@ -266,7 +266,7 @@ pub async fn number_outputs_with_amount(
     blockchain_read: &mut BlockchainReadHandle,
     output_amounts: Vec<u64>,
 ) -> Result<HashMap<u64, usize>, Error> {
-    let BlockchainResponse::NumberOutputsWithAmount(map) = blockchain_read
+    let BlockchainResponse::NumberOutputsWithAmount(map, _) = blockchain_read
         .ready()
         .await?
         .call(BlockchainReadRequest::NumberOutputsWithAmount(

binaries/cuprated/src/txpool/incoming_tx.rs

@@ -40,16 +40,15 @@ use crate::{
     config::TxpoolConfig,
     constants::PANIC_CRITICAL_SERVICE_ERROR,
     p2p::CrossNetworkInternalPeerId,
-    signals::REORG_LOCK,
     txpool::{
-        dandelion::{
-            self, AnonTxService, ConcreteDandelionRouter, DiffuseService, MainDandelionRouter,
-        },
+        dandelion::{self, DiffuseService},
         manager::{start_txpool_manager, TxpoolManagerHandle},
-        relay_rules::{check_tx_relay_rules, RelayRuleError},
+        relay_rules::check_tx_relay_rules,
         txs_being_handled::{TxsBeingHandled, TxsBeingHandledLocally},
+        RelayRuleError,
     },
 };
+use crate::txpool::dandelion::{AnonTxService, MainDandelionRouter};
 
 /// An error that can happen handling an incoming tx.
 #[derive(Debug, thiserror::Error)]
@@ -134,9 +133,14 @@ impl IncomingTxHandler {
             promote_tx,
         );
 
+        let blockchain_read_handle =
+            ConsensusBlockchainReadHandle::new(blockchain_read_handle, BoxError::from);
+
         let txpool_manager = start_txpool_manager(
             txpool_write_handle,
             txpool_read_handle.clone(),
+            blockchain_context_cache.clone(),
+            blockchain_read_handle.clone(),
             promote_rx,
             diffuse_service,
             dandelion_pool_manager.clone(),
@@ -150,10 +154,7 @@ impl IncomingTxHandler {
             dandelion_pool_manager,
             txpool_manager,
             txpool_read_handle,
-            blockchain_read_handle: ConsensusBlockchainReadHandle::new(
-                blockchain_read_handle,
-                BoxError::from,
-            ),
+            blockchain_read_handle,
         }
     }
 }
@@ -196,33 +197,22 @@ async fn handle_incoming_txs(
     mut txpool_manager_handle: TxpoolManagerHandle,
     mut dandelion_pool_manager: DandelionPoolService<DandelionTx, TxId, CrossNetworkInternalPeerId>,
 ) -> Result<(), IncomingTxError> {
-    let _reorg_guard = REORG_LOCK.read().await;
-
     let (txs, stem_pool_txs, txs_being_handled_guard) =
         prepare_incoming_txs(txs, txs_being_handled, &mut txpool_read_handle).await?;
 
-    let context = blockchain_context_cache.blockchain_context();
-
     let txs = start_tx_verification()
         .append_prepped_txs(txs)
         .prepare()
         .map_err(|e| IncomingTxError::Consensus(e.into()))?
-        .full(
-            context.chain_height,
-            context.top_hash,
-            context.current_adjusted_timestamp_for_time_lock(),
-            context.current_hf,
-            blockchain_read_handle,
-            None,
-        )
+        .full(&mut blockchain_context_cache, blockchain_read_handle, None)
         .verify()
         .await
         .map_err(IncomingTxError::Consensus)?;
 
     for tx in txs {
         // TODO: this could be a DoS, if someone spams us with txs that violate these rules?
         // Maybe we should remember these invalid txs for some time to prevent them getting repeatedly sent.
-        if let Err(e) = check_tx_relay_rules(&tx, context) {
+        if let Err(e) = check_tx_relay_rules(&tx, blockchain_context_cache.blockchain_context()) {
             if drop_relay_rule_errors {
                 tracing::debug!(err = %e, tx = hex::encode(tx.tx_hash), "Tx failed relay check, skipping.");
                 continue;