Consensus: Fix reorgs (#384)

* add specific method for context

* add new statemachine for tx verification

* fix consensus crates build

* working builds

* fix CI

* add docs

* fix CI

* fix docs

* fix clippy

* cleanup

* add docs to `blockchain_context`

* fix doc tests

* add output cache

* new monero-serai

* todo

* todo

* Revert "new monero-serai"

This reverts commit fe3f6ac.

* use indexmap to request outputs

* clean up

* fix typos

* fix CI

* fix cargo hack

* fix reorgs

* check if a block is already present before adding it to the alt block cache

* fmt

Author: Boog900

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

@@ -271,12 +271,28 @@ impl super::BlockchainManager {
         block: Block,
         prepared_txs: HashMap<[u8; 32], TransactionVerificationData>,
     ) -> Result<AddAltBlock, anyhow::Error> {
+        // Check if a block already exists.
+        let BlockchainResponse::FindBlock(chain) = self
+            .blockchain_read_handle
+            .ready()
+            .await
+            .expect(PANIC_CRITICAL_SERVICE_ERROR)
+            .call(BlockchainReadRequest::FindBlock(block.hash()))
+            .await
+            .expect(PANIC_CRITICAL_SERVICE_ERROR)
+        else {
+            unreachable!();
+        };
+
+        if chain.is_some() {
+            // The block could also be in the main-chain here under some circumstances.
+            return Ok(AddAltBlock::Cached);
+        }
+
         let alt_block_info =
             sanity_check_alt_block(block, prepared_txs, self.blockchain_context_service.clone())
                 .await?;
 
-        // TODO: check in consensus crate if alt block with this hash already exists.
-
         // If this alt chain has more cumulative difficulty, reorg.
         if alt_block_info.cumulative_difficulty
             > self
@@ -479,7 +495,7 @@ impl super::BlockchainManager {
 
 /// The result from successfully adding an alt-block.
 enum AddAltBlock {
-    /// The alt-block was cached.
+    /// The alt-block was cached or was already present in the DB.
     Cached,
     /// The chain was reorged.
     Reorged,

binaries/cuprated/src/blockchain/syncer.rs

@@ -3,7 +3,7 @@ use std::{sync::Arc, time::Duration};
 
 use futures::StreamExt;
 use tokio::{
-    sync::{mpsc, Notify},
+    sync::{mpsc, oneshot, Notify},
     time::interval,
 };
 use tower::{Service, ServiceExt};

binaries/cuprated/src/main.rs

@@ -27,6 +27,7 @@ use cuprate_consensus_context::{
     BlockChainContextRequest, BlockChainContextResponse, BlockchainContextService,
 };
 use cuprate_helper::time::secs_to_hms;
+use cuprate_types::blockchain::BlockchainWriteRequest;
 
 use crate::{
     config::Config, constants::PANIC_CRITICAL_SERVICE_ERROR, logging::CupratedTracingFilter,

consensus/context/src/alt_chains.rs

@@ -83,12 +83,8 @@ impl AltChainMap {
     }
 
     /// Add an alt chain cache to the map.
-    pub(crate) fn add_alt_cache(
-        &mut self,
-        prev_id: [u8; 32],
-        alt_cache: Box<AltChainContextCache>,
-    ) {
-        self.alt_cache_map.insert(prev_id, alt_cache);
+    pub(crate) fn add_alt_cache(&mut self, alt_cache: Box<AltChainContextCache>) {
+        self.alt_cache_map.insert(alt_cache.top_hash, alt_cache);
     }
 
     /// Attempts to take an [`AltChainContextCache`] from the map, returning [`None`] if no cache is
@@ -119,7 +115,7 @@ impl AltChainMap {
             weight_cache: None,
             difficulty_cache: None,
             cached_rx_vm: None,
-            chain_height: top_height,
+            chain_height: top_height + 1,
             top_hash: prev_id,
             chain_id: None,
             parent_chain,

consensus/context/src/lib.rs

@@ -307,8 +307,6 @@ pub enum BlockChainContextRequest {
     /// This variant is private and is not callable from outside this crate, the block verifier service will
     /// handle returning the alt cache to the context service.
     AddAltChainContextCache {
-        /// The previous block field in a [`BlockHeader`](monero_serai::block::BlockHeader).
-        prev_id: [u8; 32],
         /// The cache.
         cache: Box<AltChainContextCache>,
         /// An internal token to prevent external crates calling this request.

consensus/context/src/task.rs

@@ -303,12 +303,8 @@ impl<D: Database + Clone + Send + 'static> ContextTask<D> {
                     .get_alt_vm(height, chain, &mut self.database)
                     .await?,
             ),
-            BlockChainContextRequest::AddAltChainContextCache {
-                prev_id,
-                cache,
-                _token,
-            } => {
-                self.alt_chain_cache_map.add_alt_cache(prev_id, cache);
+            BlockChainContextRequest::AddAltChainContextCache { cache, _token } => {
+                self.alt_chain_cache_map.add_alt_cache(cache);
                 BlockChainContextResponse::Ok
             }
             BlockChainContextRequest::HardForkInfo(_)

consensus/src/block/alt_block.rs

@@ -178,7 +178,6 @@ where
     // Add this alt cache back to the context service.
     context_svc
         .oneshot(BlockChainContextRequest::AddAltChainContextCache {
-            prev_id: block_info.block.header.previous,
             cache: alt_context_cache,
             _token: AltChainRequestToken,
         })