Update to experimental monero oxide api

Cargo.toml

@@ -121,7 +121,7 @@ futures               = { version = "0.3", default-features = false }
 hex                   = { version = "0.4", default-features = false }
 hex-literal           = { version = "0.4", default-features = false }
 indexmap              = { version = "2", default-features = false }
-monero-serai          = { git = "https://github.com/Cuprate/serai.git", rev = "e6fdef6", default-features = false }
+monero-serai          = { git = "https://github.com/Cuprate/serai.git", rev = "29f904d", default-features = false }
 paste                 = { version = "1", default-features = false }
 pin-project           = { version = "1", default-features = false }
 randomx-rs            = { git = "https://github.com/Cuprate/randomx-rs.git", rev = "0028464", default-features = false }
@@ -144,8 +144,8 @@ tracing-subscriber    = { version = "0.3", default-features = false }
 tracing               = { version = "0.1", default-features = false }
 
 ## workspace.dev-dependencies
-monero-rpc                = { git = "https://github.com/Cuprate/serai.git", rev = "e6fdef6" }
-monero-simple-request-rpc = { git = "https://github.com/Cuprate/serai.git", rev = "e6fdef6" }
+monero-rpc                = { git = "https://github.com/Cuprate/serai.git", rev = "29f904d" }
+monero-simple-request-rpc = { git = "https://github.com/Cuprate/serai.git", rev = "29f904d" }
 tempfile                  = { version = "3" }
 pretty_assertions         = { version = "1" }
 proptest                  = { version = "1" }

binaries/cuprated/Cuprated.toml

@@ -43,9 +43,9 @@ peer_save_period = { secs = 90, nanos = 0 }
 ## The block downloader config.
 [p2p.block_downloader]
 ## The size of the buffer of sequential blocks waiting to be verified and added to the chain (bytes).
-buffer_bytes = 50_000_000
+buffer_bytes = 1_000_000_000
 ## The size of the queue of blocks which are waiting for a parent block to be downloaded (bytes).
-in_progress_queue_bytes = 50_000_000
+in_progress_queue_bytes = 500_000_000
 ## The target size of a batch of blocks (bytes), must not exceed 100MB.
 target_batch_bytes = 10_000_000
 ## The amount of time between checking the pool of connected peers for free peers to download blocks.

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

@@ -166,9 +166,12 @@ impl super::BlockchainManager {
     /// This function will panic if any internal service returns an unexpected error that we cannot
     /// recover from or if the incoming batch contains no blocks.
     async fn handle_incoming_block_batch_main_chain(&mut self, batch: BlockBatch) {
-        let Ok(prepped_blocks) =
-            batch_prepare_main_chain_blocks(batch.blocks, &mut self.blockchain_context_service)
-                .await
+        let Ok((prepped_blocks, mut output_cache)) = batch_prepare_main_chain_blocks(
+            batch.blocks,
+            &mut self.blockchain_context_service,
+            self.blockchain_read_handle.clone(),
+        )
+        .await
         else {
             batch.peer_handle.ban_peer(LONG_BAN);
             self.stop_current_block_downloader.notify_one();
@@ -181,6 +184,7 @@ impl super::BlockchainManager {
                 txs,
                 &mut self.blockchain_context_service,
                 self.blockchain_read_handle.clone(),
+                Some(&mut output_cache),
             )
             .await
             else {
@@ -267,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
@@ -404,6 +424,7 @@ impl super::BlockchainManager {
                 prepped_txs,
                 &mut self.blockchain_context_service,
                 self.blockchain_read_handle.clone(),
+                None,
             )
             .await?;
 
@@ -431,7 +452,7 @@ impl super::BlockchainManager {
             .iter()
             .flat_map(|tx| {
                 tx.tx.prefix().inputs.iter().map(|input| match input {
-                    Input::ToKey { key_image, .. } => key_image.compress().0,
+                    Input::ToKey { key_image, .. } => key_image.0,
                     Input::Gen(_) => unreachable!(),
                 })
             })
@@ -474,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/config/p2p.rs

@@ -47,8 +47,8 @@ impl From<BlockDownloaderConfig> for cuprate_p2p::block_downloader::BlockDownloa
 impl Default for BlockDownloaderConfig {
     fn default() -> Self {
         Self {
-            buffer_bytes: 50_000_000,
-            in_progress_queue_bytes: 50_000_000,
+            buffer_bytes: 1_000_000_000,
+            in_progress_queue_bytes: 500_000_000,
             check_client_pool_interval: Duration::from_secs(30),
             target_batch_bytes: 10_000_000,
         }

binaries/cuprated/src/main.rs

@@ -16,8 +16,8 @@
     reason = "TODO: remove after v1.0.0"
 )]
 
-use std::mem;
-use std::sync::Arc;
+use std::{mem, sync::Arc};
+
 use tokio::sync::mpsc;
 use tower::{Service, ServiceExt};
 use tracing::level_filters::LevelFilter;
@@ -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,

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

@@ -6,13 +6,15 @@ use std::{
 };
 
 use anyhow::Error;
+use indexmap::{IndexMap, IndexSet};
 use monero_serai::block::Block;
 use tower::{Service, ServiceExt};
 
 use cuprate_blockchain::{service::BlockchainReadHandle, types::AltChainInfo};
 use cuprate_helper::cast::{u64_to_usize, usize_to_u64};
 use cuprate_types::{
     blockchain::{BlockchainReadRequest, BlockchainResponse},
+    output_cache::OutputCache,
     Chain, ChainInfo, CoinbaseTxSum, ExtendedBlockHeader, HardFork, MinerData,
     OutputHistogramEntry, OutputHistogramInput, OutputOnChain,
 };
@@ -185,8 +187,8 @@ pub(crate) async fn generated_coins(
 /// [`BlockchainReadRequest::Outputs`]
 pub(crate) async fn outputs(
     blockchain_read: &mut BlockchainReadHandle,
-    outputs: HashMap<u64, HashSet<u64>>,
-) -> Result<HashMap<u64, HashMap<u64, OutputOnChain>>, Error> {
+    outputs: IndexMap<u64, IndexSet<u64>>,
+) -> Result<OutputCache, Error> {
     let BlockchainResponse::Outputs(outputs) = blockchain_read
         .ready()
         .await?

binaries/cuprated/src/txpool/incoming_tx.rs

@@ -172,6 +172,7 @@ async fn handle_incoming_txs(
             context.current_adjusted_timestamp_for_time_lock(),
             context.current_hf,
             blockchain_read_handle,
+            None,
         )
         .verify()
         .await

books/architecture/src/resources/cap/intro.md

@@ -8,7 +8,7 @@ There are "synchronization primitives" that help with this, common ones being:
 - [Channels](https://en.wikipedia.org/wiki/Channel_(programming))
 - [Atomics](https://en.wikipedia.org/wiki/Linearizability#Primitive_atomic_instructions)
 
-These tools are relatively easy to use in isolation, but trickier to do so when considering the entire system. It is not uncommon for _the_ bottleneck to be the [poor orchastration](https://en.wikipedia.org/wiki/Starvation_(computer_science)) of these primitives.
+These tools are relatively easy to use in isolation, but trickier to do so when considering the entire system. It is not uncommon for _the_ bottleneck to be the [poor orchestration](https://en.wikipedia.org/wiki/Starvation_(computer_science)) of these primitives.
 
 ## Analogy
 A common analogy for a parallel system is an intersection.

consensus/Cargo.toml

@@ -23,6 +23,7 @@ monero-serai = { workspace = true, features = ["std"] }
 rayon = { workspace = true }
 thread_local = { workspace = true }
 
+indexmap = { workspace = true, features = ["std"] }
 hex = { workspace = true }
 rand = { workspace = true }