diff --git a/binaries/cuprated/src/blockchain/manager.rs b/binaries/cuprated/src/blockchain/manager.rs index 6d014e529..dd07ae823 100644 --- a/binaries/cuprated/src/blockchain/manager.rs +++ b/binaries/cuprated/src/blockchain/manager.rs @@ -81,6 +81,7 @@ pub(crate) async fn init_blockchain_manager( broadcast_svc: clearnet_interface.broadcast_svc(), reorg_lock: Arc::clone(&launch_ctx.reorg_lock), fast_sync_hashes, + node_events: launch_ctx.node_events.clone(), }; launch_ctx @@ -116,6 +117,8 @@ pub struct BlockchainManager { reorg_lock: Arc>, /// Fast-sync hashes for this node's network. fast_sync_hashes: &'static [[u8; 32]], + /// Sender for the node event stream. + node_events: crate::events::NodeEventSender, } impl BlockchainManager { diff --git a/binaries/cuprated/src/blockchain/manager/handler.rs b/binaries/cuprated/src/blockchain/manager/handler.rs index 6a6c06c52..b5139fe8f 100644 --- a/binaries/cuprated/src/blockchain/manager/handler.rs +++ b/binaries/cuprated/src/blockchain/manager/handler.rs @@ -34,6 +34,7 @@ use cuprate_types::{ use crate::{ blockchain::manager::commands::{BlockchainManagerCommand, IncomingBlockOk}, constants::PANIC_CRITICAL_SERVICE_ERROR, + events::NodeEvent, }; impl super::BlockchainManager { @@ -469,14 +470,15 @@ impl super::BlockchainManager { match reorg_res { Ok(()) => { - info!( - top_hash = hex::encode( - self.blockchain_context_service - .blockchain_context() - .top_hash - ), - "Successfully reorged" - ); + let ctx = self.blockchain_context_service.blockchain_context(); + let new_top_hash = ctx.top_hash; + let new_chain_height = ctx.chain_height; + info!(top_hash = hex::encode(new_top_hash), "Successfully reorged"); + self.node_events.send(NodeEvent::Reorg { + split_height, + new_top_hash, + new_chain_height, + }); Ok(()) } Err(e) => { @@ -635,6 +637,9 @@ impl super::BlockchainManager { verified_block: VerifiedBlockInformation, source: BlockSource, ) { + let height = verified_block.height; + let hash = verified_block.block_hash; + // FIXME: this is pretty inefficient, we should probably return the KI map created in the consensus crate. let spent_key_images = verified_block .txs @@ -656,6 +661,10 @@ impl super::BlockchainManager { self.add_valid_block_to_blockchain_database(verified_block) .await; + if matches!(source, BlockSource::Incoming) { + self.node_events.send(NodeEvent::NewBlock { height, hash }); + } + if let Some(block_blob) = block_blob { let chain_height = self .blockchain_context_service diff --git a/binaries/cuprated/src/blockchain/manager/tests.rs b/binaries/cuprated/src/blockchain/manager/tests.rs index 6f6727ec7..c22453b8e 100644 --- a/binaries/cuprated/src/blockchain/manager/tests.rs +++ b/binaries/cuprated/src/blockchain/manager/tests.rs @@ -5,7 +5,7 @@ use monero_oxide::{ ed25519::CompressedPoint, transaction::{Input, Output, Timelock, Transaction, TransactionPrefix}, }; -use tokio::sync::{oneshot, watch}; +use tokio::sync::{broadcast, oneshot, watch}; use tower::BoxError; use cuprate_blockchain::config::Config; @@ -21,10 +21,11 @@ use crate::{ check_add_genesis, manager::BlockchainManager, manager::BlockchainManagerCommand, ConsensusBlockchainReadHandle, }, + events::{NodeEvent, NodeEventListener, NodeEventSender, NODE_EVENT_CHANNEL_CAPACITY}, txpool::TxpoolManagerHandle, }; -async fn mock_manager(data_dir: PathBuf) -> BlockchainManager { +async fn mock_manager(data_dir: PathBuf) -> (BlockchainManager, NodeEventListener) { let config = Config { blob_dir: data_dir.clone(), index_dir: data_dir.clone(), @@ -66,16 +67,23 @@ async fn mock_manager(data_dir: PathBuf) -> BlockchainManager { .await .unwrap(); - BlockchainManager { - blockchain_write_handle, - blockchain_read_handle, - txpool_manager_handle: TxpoolManagerHandle::mock(), - blockchain_context_service, - stop_current_block_downloader: Arc::new(Default::default()), - broadcast_svc: BroadcastSvc::mock(), - reorg_lock: Arc::new(Default::default()), - fast_sync_hashes: &[], - } + let node_events = NodeEventSender::new(); + let listener = node_events.subscribe(); + + ( + BlockchainManager { + blockchain_write_handle, + blockchain_read_handle, + txpool_manager_handle: TxpoolManagerHandle::mock(), + blockchain_context_service, + stop_current_block_downloader: Arc::new(Default::default()), + broadcast_svc: BroadcastSvc::mock(), + node_events, + reorg_lock: Arc::new(Default::default()), + fast_sync_hashes: &[], + }, + listener, + ) } fn generate_block(context: &BlockchainContext) -> Block { @@ -115,10 +123,12 @@ fn generate_block(context: &BlockchainContext) -> Block { async fn simple_reorg() { // create 2 managers let data_dir_1 = tempfile::tempdir().unwrap(); - let mut manager_1 = mock_manager(data_dir_1.path().to_path_buf()).await; + let manager_1_with_events = mock_manager(data_dir_1.path().to_path_buf()).await; + let mut manager_1 = manager_1_with_events.0; let data_dir_2 = tempfile::tempdir().unwrap(); - let mut manager_2 = mock_manager(data_dir_2.path().to_path_buf()).await; + let manager_2_with_events = mock_manager(data_dir_2.path().to_path_buf()).await; + let mut manager_2 = manager_2_with_events.0; // give both managers the same first non-genesis block let block_1 = generate_block(manager_1.blockchain_context_service.blockchain_context()); @@ -228,10 +238,12 @@ async fn simple_reorg_block_batch() { // create 2 managers let data_dir_1 = tempfile::tempdir().unwrap(); - let mut manager_1 = mock_manager(data_dir_1.path().to_path_buf()).await; + let manager_1_with_events = mock_manager(data_dir_1.path().to_path_buf()).await; + let mut manager_1 = manager_1_with_events.0; let data_dir_2 = tempfile::tempdir().unwrap(); - let mut manager_2 = mock_manager(data_dir_2.path().to_path_buf()).await; + let manager_2_with_events = mock_manager(data_dir_2.path().to_path_buf()).await; + let mut manager_2 = manager_2_with_events.0; // give both managers the same first non-genesis block let block_1 = generate_block(manager_1.blockchain_context_service.blockchain_context()); @@ -337,7 +349,8 @@ async fn simple_reorg_block_batch() { #[tokio::test] async fn recover_bad_reorg() { let data_dir_1 = tempfile::tempdir().unwrap(); - let mut manager_1 = mock_manager(data_dir_1.path().to_path_buf()).await; + let manager_1_with_events = mock_manager(data_dir_1.path().to_path_buf()).await; + let mut manager_1 = manager_1_with_events.0; let context_1 = manager_1 .blockchain_context_service @@ -442,3 +455,195 @@ async fn recover_bad_reorg() { manager_1.blockchain_context_service.blockchain_context() ); } + +#[tokio::test] +async fn node_event_delivered_to_prior_subscriber() { + let s = NodeEventSender::new(); + let mut l = s.subscribe(); + + s.send(NodeEvent::NewBlock { + height: 7, + hash: [1_u8; 32], + }); + + assert_eq!( + l.recv().await.unwrap(), + NodeEvent::NewBlock { + height: 7, + hash: [1_u8; 32], + } + ); +} + +#[tokio::test] +async fn node_event_two_subscribers_both_receive() { + let s = NodeEventSender::new(); + let mut l1 = s.subscribe(); + let mut l2 = s.subscribe(); + let event = NodeEvent::NewBlock { + height: 3, + hash: [2_u8; 32], + }; + + s.send(event.clone()); + + assert_eq!(l1.recv().await.unwrap(), event); + assert_eq!(l2.recv().await.unwrap(), event); +} + +#[test] +fn node_event_no_subscriber_is_noop() { + let sender = NodeEventSender::new(); + + sender.send(NodeEvent::NewBlock { + height: 5, + hash: [9_u8; 32], + }); +} + +#[test] +fn node_event_late_subscriber_misses_prior() { + let s = NodeEventSender::new(); + s.send(NodeEvent::NewBlock { + height: 8, + hash: [4_u8; 32], + }); + + let mut listener = s.subscribe(); + + assert_eq!( + listener.try_recv(), + Err(broadcast::error::TryRecvError::Empty) + ); +} + +#[test] +fn node_event_channel_capacity_is_256() { + assert_eq!(NODE_EVENT_CHANNEL_CAPACITY, 256); +} + +#[tokio::test] +async fn new_block_emits_new_block_event() { + let data_dir = tempfile::tempdir().unwrap(); + let manager_with_events = mock_manager(data_dir.path().to_path_buf()).await; + let mut manager = manager_with_events.0; + let mut listener = manager_with_events.1; + + let block = generate_block(manager.blockchain_context_service.blockchain_context()); + let hash = block.hash(); + + manager + .handle_command(BlockchainManagerCommand::AddBlock { + block, + prepped_txs: HashMap::new(), + response_tx: oneshot::channel().0, + }) + .await; + + assert_eq!( + listener.try_recv(), + Ok(NodeEvent::NewBlock { height: 1, hash }) + ); + assert_eq!( + listener.try_recv(), + Err(broadcast::error::TryRecvError::Empty) + ); +} + +#[tokio::test] +async fn reorg_emits_reorg_event() { + let data_dir_1 = tempfile::tempdir().unwrap(); + let manager_1_with_events = mock_manager(data_dir_1.path().to_path_buf()).await; + let mut manager_1 = manager_1_with_events.0; + let mut listener = manager_1_with_events.1; + + let data_dir_2 = tempfile::tempdir().unwrap(); + let manager_2_with_events = mock_manager(data_dir_2.path().to_path_buf()).await; + let mut manager_2 = manager_2_with_events.0; + + let block_1 = generate_block(manager_1.blockchain_context_service.blockchain_context()); + + manager_1 + .handle_command(BlockchainManagerCommand::AddBlock { + block: block_1.clone(), + prepped_txs: HashMap::new(), + response_tx: oneshot::channel().0, + }) + .await; + + manager_2 + .handle_command(BlockchainManagerCommand::AddBlock { + block: block_1, + prepped_txs: HashMap::new(), + response_tx: oneshot::channel().0, + }) + .await; + + let block_2a = generate_block(manager_1.blockchain_context_service.blockchain_context()); + let block_2b = generate_block(manager_2.blockchain_context_service.blockchain_context()); + + manager_1 + .handle_command(BlockchainManagerCommand::AddBlock { + block: block_2a, + prepped_txs: HashMap::new(), + response_tx: oneshot::channel().0, + }) + .await; + + manager_2 + .handle_command(BlockchainManagerCommand::AddBlock { + block: block_2b.clone(), + prepped_txs: HashMap::new(), + response_tx: oneshot::channel().0, + }) + .await; + + manager_1 + .handle_command(BlockchainManagerCommand::AddBlock { + block: block_2b, + prepped_txs: HashMap::new(), + response_tx: oneshot::channel().0, + }) + .await; + + let block_3 = generate_block(manager_2.blockchain_context_service.blockchain_context()); + + // Discard the `NewBlock` events emitted while building the initial main chain (block_1 and + // block_2a were added as live `Incoming` blocks). We only want to observe what the reorg itself + // emits. + while listener.try_recv().is_ok() {} + + manager_1 + .handle_command(BlockchainManagerCommand::AddBlock { + block: block_3, + prepped_txs: HashMap::new(), + response_tx: oneshot::channel().0, + }) + .await; + + let mut events = Vec::new(); + loop { + match listener.try_recv() { + Ok(event) => events.push(event), + Err(broadcast::error::TryRecvError::Empty) => break, + Err(err) => panic!("unexpected broadcast receive error: {err:?}"), + } + } + + let chain_context = manager_1 + .blockchain_context_service + .blockchain_context() + .clone(); + + assert!(events.iter().any(|event| matches!( + event, + NodeEvent::Reorg { + split_height: 2, + new_top_hash: _, + new_chain_height + } if *new_chain_height == chain_context.chain_height + ))); + assert!(!events + .iter() + .any(|event| matches!(event, NodeEvent::NewBlock { .. }))); +} diff --git a/binaries/cuprated/src/events.rs b/binaries/cuprated/src/events.rs new file mode 100644 index 000000000..4d5514bfa --- /dev/null +++ b/binaries/cuprated/src/events.rs @@ -0,0 +1,88 @@ +//! Node event streaming. +//! +//! [`Node::events`](crate::Node::events) returns a [`NodeEventListener`] -- a forward-looking +//! subscription to [`NodeEvent`]s published by node subsystems. + +use tokio::sync::broadcast; + +/// Capacity of the node-event broadcast channel. +/// +/// Slow consumers that fall this far behind receive +/// [`RecvError::Lagged`](broadcast::error::RecvError::Lagged). +pub const NODE_EVENT_CHANNEL_CAPACITY: usize = 256; + +/// An event emitted by a running node. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum NodeEvent { + /// A new block was accepted at the live main-chain tip via the incoming-block path — + /// whether relayed by a peer or submitted locally (e.g. the `submit_block` RPC). + /// + /// Not emitted for blocks applied during initial batch-sync or for blocks re-applied + /// during a reorg. + NewBlock { + /// Height of the new block. + height: usize, + /// Hash of the new block. + hash: [u8; 32], + }, + /// The main chain was reorganized onto a heavier alternative chain. + Reorg { + /// Height of the first block that differs between the old and new chains. + split_height: usize, + /// Hash of the new main-chain tip after the reorg. + new_top_hash: [u8; 32], + /// Main-chain height after the reorg. + new_chain_height: usize, + }, +} + +/// The publishing half of the node-event channel. Cheaply cloneable. +#[derive(Clone)] +pub(crate) struct NodeEventSender { + tx: broadcast::Sender, +} + +impl NodeEventSender { + /// Create a new sender backed by a fresh broadcast channel. + pub(crate) fn new() -> Self { + let (tx, _rx) = broadcast::channel(NODE_EVENT_CHANNEL_CAPACITY); + Self { tx } + } + + /// Publish an event. A send with no active listeners is a no-op (not an error). + pub(crate) fn send(&self, event: NodeEvent) { + let _ = self.tx.send(event); + } + + /// Subscribe a new, forward-looking listener. + pub fn subscribe(&self) -> NodeEventListener { + NodeEventListener { + rx: self.tx.subscribe(), + } + } +} + +/// A subscription to [`NodeEvent`]s. Obtain one from [`Node::events`](crate::Node::events). +#[must_use] +pub struct NodeEventListener { + rx: broadcast::Receiver, +} + +impl NodeEventListener { + /// Await the next event. See [`broadcast::Receiver::recv`]. + pub async fn recv(&mut self) -> Result { + self.rx.recv().await + } + + /// Try to receive the next event without waiting. + pub fn try_recv(&mut self) -> Result { + self.rx.try_recv() + } + + /// Create a new listener that receives events from now on, sharing the same channel. + pub fn resubscribe(&self) -> Self { + Self { + rx: self.rx.resubscribe(), + } + } +} diff --git a/binaries/cuprated/src/lib.rs b/binaries/cuprated/src/lib.rs index 91588c157..5aa6e9040 100644 --- a/binaries/cuprated/src/lib.rs +++ b/binaries/cuprated/src/lib.rs @@ -40,6 +40,7 @@ pub mod logging; pub mod monitor; pub mod version; +mod events; mod p2p; mod rpc; mod tor; @@ -61,11 +62,14 @@ use crate::{ blockchain::{BlockchainInterface, BlockchainManagerHandle, Syncer, SyncerHandle}, config::Config, constants::DATABASE_CORRUPT_MSG, + events::NodeEventSender, monitor::TaskExecutor, tor::initialize_tor_if_enabled, txpool::IncomingTxHandler, }; +pub use events::{NodeEvent, NodeEventListener}; + /// Captures the necessary context for launching the node. /// /// A field belongs here if it is `Clone + Send + Sync`, used by @@ -95,6 +99,9 @@ pub(crate) struct LaunchContext { /// Syncer handle. pub syncer: SyncerHandle, + /// Sender for the node event stream. + pub node_events: NodeEventSender, + /// Task spawning and shutdown coordination. pub task_executor: TaskExecutor, } @@ -119,6 +126,9 @@ pub struct Node { /// Syncer handle. pub syncer: SyncerHandle, + /// Sender for the node event stream. + node_events: NodeEventSender, + /// The configuration this node was launched with. pub config: Arc, @@ -210,6 +220,8 @@ impl Node { blockchain_manager_handle.clone(), ); + let node_events = NodeEventSender::new(); + // Create the launch context. let launch_ctx = LaunchContext { config, @@ -217,6 +229,7 @@ impl Node { blockchain: blockchain_interface, txpool_read: txpool_read_handle.clone(), syncer: syncer_handle, + node_events, task_executor: TaskExecutor::new(), }; @@ -280,6 +293,7 @@ impl Node { blockchain, txpool_read, syncer, + node_events, config, task_executor, .. @@ -291,6 +305,7 @@ impl Node { clearnet: clearnet_interface, tor: if tor_enabled { Some(tor_rx) } else { None }, syncer, + node_events, config, task_executor, }) @@ -301,6 +316,13 @@ impl Node { self.task_executor.trigger_shutdown(); } + /// Subscribe to a forward-looking stream of [`NodeEvent`]s emitted by this node. + /// + /// Each call returns an independent listener. Events published before the call are not replayed. + pub fn events(&self) -> NodeEventListener { + self.node_events.subscribe() + } + /// Wait for shutdown to be triggered, then await all tracked tasks. pub async fn wait_for_shutdown(&self) { self.task_executor.wait_for_shutdown().await;