diff --git a/src/client.rs b/src/client.rs index 1e277b5..b18b70e 100644 --- a/src/client.rs +++ b/src/client.rs @@ -1,6 +1,6 @@ use bitcoin::p2p::address::AddrV2; use bitcoin::p2p::ServiceFlags; -use bitcoin::{Amount, Wtxid}; +use bitcoin::{Amount, Transaction, Wtxid}; use bitcoin::{BlockHash, FeeRate}; use tokio::sync::mpsc; use tokio::sync::mpsc::UnboundedSender; @@ -9,7 +9,7 @@ use tokio::sync::oneshot; use crate::chain::block_subsidy; use crate::chain::IndexedHeader; use crate::messages::ClientRequest; -use crate::{Event, HashCheckpoint, Info, Package, TrustedPeer, Warning}; +use crate::{Event, GossipMonitorRequest, HashCheckpoint, Info, Package, TrustedPeer, Warning}; use super::{error::ClientError, messages::ClientMessage}; use super::{error::FetchBlockError, IndexedBlock}; @@ -262,6 +262,62 @@ impl Requester { rx.await.map_err(|_| ClientError::RecvError) } + /// Subscribe to transaction gossip that involves scripts or outpoints of interest. This + /// feature is considered **experimental**. + /// + /// # Behavior + /// + /// When called for the first time, this will create a _new_ peer-to-peer connection for the + /// sole purpose of finding transactions that involve scripts and outpoints provided in the + /// request. When called subsequent times, the scripts or outpoints provided will be added to + /// the existing set. + /// + /// # Properties + /// + /// - No loss of privacy: all transactions are annouced by `inv` messages, so the node will + /// receive all transaction gossip and does not need to request special data that would + /// otherwise reveal information. + /// - Unreliable: remote nodes are not obligated to forward unconfirmed transactions. + /// Transactions may exist involving the scripts and outpoints, but the peer may not forward + /// them. + /// - Untrustworthy: unconfirmed transactions may not be treated as final! It does not take any + /// proof of work to generate transactions. No claims on transaction validity are possible + /// until the transaction has been included in a block of sufficient work. + /// + /// ## When this feature makes sense to use + /// + /// - You are expecting to see a payment from a trusted counterparty, for instance an exchange, + /// broker, friend, or colleague. + /// + /// ## When this feature does not make sense + /// + /// - You are expecting to see a payment from an untrusted counterparty. + /// + /// # Returns + /// + /// On first subscription, this function an [`mpsc::Receiver`] that will forward any transaction + /// matching the scripts and outpoints provided. Note, duplicate transactions may be sent. On + /// subsequent subscriptions, this method returns `None`, as there is an exisiting + /// [`mpsc::Receiver`]. + pub async fn subscribe_to_gossip( + &self, + req: GossipMonitorRequest, + ) -> Result>, ClientError> { + let (tx, rx) = tokio::sync::oneshot::channel::>>(); + let request = ClientRequest::new(req, tx); + self.ntx + .send(ClientMessage::SubscribeGossip(request)) + .map_err(|_| ClientError::SendError)?; + rx.await.map_err(|_| ClientError::RecvError) + } + + /// Unsubscribe from all transaction gossip. + pub async fn unsubscribe_from_gossip(&self) -> Result<(), ClientError> { + self.ntx + .send(ClientMessage::UnsubscribeGossip) + .map_err(|_| ClientError::SendError) + } + /// Check if the node is running. pub fn is_running(&self) -> bool { self.ntx.send(ClientMessage::NoOp).is_ok() diff --git a/src/lib.rs b/src/lib.rs index 40dacbf..3c50c9a 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -63,7 +63,10 @@ pub mod node; use bitcoin::OutPoint; use chain::Filter; -use std::net::{IpAddr, Ipv4Addr, SocketAddr}; +use std::{ + collections::HashSet, + net::{IpAddr, Ipv4Addr, SocketAddr}, +}; // Re-exports #[doc(inline)] @@ -568,6 +571,62 @@ impl From for Package { } } +/// Instruct the node to watch for unconfirmed transactions with a certain criteria. +#[derive(Debug, Clone)] +pub struct GossipMonitorRequest { + scripts: HashSet, + txins: HashSet, +} + +/// Build a new [`GossipMonitorRequest`]. +#[derive(Debug, Clone, Default)] +pub struct GossipMonitorRequestBuilder { + scripts: HashSet, + txins: HashSet, +} + +impl GossipMonitorRequestBuilder { + /// Watch for these **output** scripts in an unconfirmed transactions. + pub fn from_expected_output_scripts(scripts: impl IntoIterator) -> Self { + let mut script_set = HashSet::new(); + script_set.extend(scripts); + Self { + scripts: script_set, + txins: HashSet::new(), + } + } + + /// Watch for these [`OutPoint`] used as an input in unconfirmed transactions. + pub fn from_expected_inputs(txins: impl IntoIterator) -> Self { + let mut op_set = HashSet::new(); + op_set.extend(txins); + Self { + scripts: HashSet::new(), + txins: op_set, + } + } + + /// Add scripts to the request. + pub fn add_scripts(mut self, scripts: impl IntoIterator) -> Self { + self.scripts.extend(scripts); + self + } + + /// Add outpoints to the request. + pub fn add_outpoints(mut self, txins: impl IntoIterator) -> Self { + self.txins.extend(txins); + self + } + + /// Complete the configuration for the request. + pub fn build(self) -> GossipMonitorRequest { + GossipMonitorRequest { + scripts: self.scripts, + txins: self.txins, + } + } +} + macro_rules! impl_sourceless_error { ($e:ident) => { impl std::error::Error for $e { diff --git a/src/messages.rs b/src/messages.rs index af9e291..250b7a0 100644 --- a/src/messages.rs +++ b/src/messages.rs @@ -3,11 +3,13 @@ use std::ops::Div; use bitcoin::p2p::address::AddrV2; use bitcoin::p2p::ServiceFlags; +use bitcoin::Transaction; use bitcoin::{block::Header, p2p::message_network::RejectReason, BlockHash, FeeRate, Wtxid}; +use tokio::sync::mpsc::Receiver; use crate::chain::{BlockHeaderChanges, IndexedHeader}; use crate::{chain::checkpoints::HashCheckpoint, IndexedBlock, TrustedPeer}; -use crate::{IndexedFilter, Package}; +use crate::{GossipMonitorRequest, IndexedFilter, Package}; use super::error::FetchBlockError; @@ -155,6 +157,10 @@ pub(crate) enum ClientMessage { GetHeader(ClientRequest>), /// Look up the height of a block hash in the chain of most work. HeightOfHash(ClientRequest>), + /// Look for unconfirmed transactions. + SubscribeGossip(ClientRequest>>), + /// Stop looking for unconfirmed transactions. + UnsubscribeGossip, /// Send an empty message to see if the node is running. NoOp, } diff --git a/src/network/gossip.rs b/src/network/gossip.rs new file mode 100644 index 0000000..3219855 --- /dev/null +++ b/src/network/gossip.rs @@ -0,0 +1,64 @@ +use std::collections::HashSet; + +use bitcoin::{OutPoint, ScriptBuf, Transaction}; +use tokio::sync::mpsc::Sender; + +#[derive(Debug, Clone, Copy)] +pub(crate) enum TransactionCheck { + NoMatch, + SentToClient, + Failed, +} + +#[derive(Debug)] +pub(crate) struct GossipMonitor { + scripts: HashSet, + txins: HashSet, + sender: Sender, +} + +impl GossipMonitor { + pub(crate) fn new( + scripts: HashSet, + txins: HashSet, + sender: Sender, + ) -> Self { + Self { + scripts, + txins, + sender, + } + } + + pub(crate) fn extend( + &mut self, + scripts: impl IntoIterator, + txins: impl IntoIterator, + ) { + self.scripts.extend(scripts); + self.txins.extend(txins); + } + + fn matches(&self, tx: &Transaction) -> bool { + tx.input + .iter() + .any(|txin| self.txins.contains(&txin.previous_output)) + || tx + .output + .iter() + .any(|txout| self.scripts.contains(&txout.script_pubkey)) + } + + pub(crate) fn check_tx(&self, tx: Transaction) -> TransactionCheck { + if self.matches(&tx) { + let sent = self.sender.try_send(tx).is_ok(); + if sent { + TransactionCheck::SentToClient + } else { + TransactionCheck::Failed + } + } else { + TransactionCheck::NoMatch + } + } +} diff --git a/src/network/mod.rs b/src/network/mod.rs index 068d272..31f05e0 100644 --- a/src/network/mod.rs +++ b/src/network/mod.rs @@ -3,6 +3,7 @@ use std::{ fs::{self, File}, net::IpAddr, path::PathBuf, + sync::Arc, time::Duration, }; @@ -24,14 +25,15 @@ use bitcoin::{ Block, BlockHash, FeeRate, Wtxid, }; use socks::{create_socks5, SocksConnection}; -use tokio::{net::TcpStream, time::Instant}; +use tokio::{net::TcpStream, sync::Mutex, time::Instant}; use error::PeerError; -use crate::Socks5Proxy; +use crate::{network::gossip::GossipMonitor, Socks5Proxy}; pub(crate) mod dns; pub(crate) mod error; +pub(crate) mod gossip; pub(crate) mod inbound; pub(crate) mod outbound; pub(crate) mod peer; @@ -193,6 +195,12 @@ impl ConnectionType { } } +#[derive(Debug, Clone)] +pub(crate) enum RelayPolicy { + BlocksOnly, + Transactions(Arc>), +} + #[derive(Debug, Clone)] struct MessageState { general_timeout: Duration, diff --git a/src/network/outbound.rs b/src/network/outbound.rs index 49bb586..1014848 100644 --- a/src/network/outbound.rs +++ b/src/network/outbound.rs @@ -15,7 +15,7 @@ use bitcoin::{ BlockHash, Network, Transaction, Wtxid, }; -use crate::{default_port_from_network, BlockType}; +use crate::{network::RelayPolicy, BlockType}; use super::{KYOTO_VERSION, PROTOCOL_VERSION, RUST_BITCOIN_VERSION}; @@ -45,8 +45,8 @@ impl MessageGenerator { } } - pub(in crate::network) fn version_message(&mut self, port: Option) -> Vec { - let msg = NetworkMessage::Version(make_version(port, &self.network)); + pub(in crate::network) fn version_message(&mut self, relay_policy: &RelayPolicy) -> Vec { + let msg = NetworkMessage::Version(make_version(relay_policy)); self.serialize(msg) } @@ -81,17 +81,17 @@ fn encrypt_plaintext(encryptor: &mut OutboundCipher, plaintext: Vec) -> Vec< encryptor.encrypt_to_vec(&plaintext, PacketType::Genuine, None) } -pub(in crate::network) fn make_version(port: Option, network: &Network) -> VersionMessage { +pub(in crate::network) fn make_version(relay_policy: &RelayPolicy) -> VersionMessage { let now = SystemTime::now() .duration_since(UNIX_EPOCH) .expect("time went backwards") .as_secs(); - let default_port = default_port_from_network(network); - let ip = SocketAddr::new( - IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), - port.unwrap_or(default_port), - ); + let ip = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 8333); let from_and_recv = Address::new(&ip, ServiceFlags::NONE); + let relay = match relay_policy { + RelayPolicy::Transactions(_) => true, + RelayPolicy::BlocksOnly => false, + }; VersionMessage { version: PROTOCOL_VERSION, services: ServiceFlags::NONE, @@ -101,6 +101,6 @@ pub(in crate::network) fn make_version(port: Option, network: &Network) -> nonce: 1, user_agent: format!("/Rust BIP-157:{KYOTO_VERSION}/rust-bitcoin:{RUST_BITCOIN_VERSION}/"), start_height: 0, - relay: false, + relay, } } diff --git a/src/network/peer.rs b/src/network/peer.rs index 61826db..eeb44b5 100644 --- a/src/network/peer.rs +++ b/src/network/peer.rs @@ -4,6 +4,7 @@ use std::{sync::Arc, time::Duration}; use addrman::Record; use bip324::futures::{Protocol, ProtocolReader}; use bip324::{OutboundCipher, Role}; +use bitcoin::BlockHash; use bitcoin::{ p2p::{message::NetworkMessage, message_blockdata::Inventory, ServiceFlags}, Network, @@ -19,6 +20,8 @@ use tokio::{ time::{Instant, MissedTickBehavior}, }; +use crate::network::gossip::TransactionCheck; +use crate::network::RelayPolicy; use crate::{broadcaster::BroadcastQueue, messages::Warning, BlockType, Dialog, Info}; use super::{ @@ -45,6 +48,7 @@ pub(crate) struct Peer { timeout_config: PeerTimeoutConfig, message_state: MessageState, tx_queue: Arc>, + relay_policy: RelayPolicy, } impl Peer { @@ -54,6 +58,7 @@ impl Peer { source: Record, network: Network, block_type: BlockType, + relay_policy: RelayPolicy, main_thread_sender: Sender, main_thread_recv: Receiver, dialog: Arc, @@ -68,6 +73,7 @@ impl Peer { main_thread_recv, network, block_type, + relay_policy, dialog, db, timeout_config, @@ -114,7 +120,7 @@ impl Peer { (outbound_messages, reader) }; - let message = outbound_messages.version_message(None); + let message = outbound_messages.version_message(&self.relay_policy); self.write_bytes(&mut writer, message).await?; self.message_state.start_version_handshake(); let read_handle = tokio::spawn(async move { peer_reader.read_from_remote().await }); @@ -259,15 +265,65 @@ impl Peer { .await?; Ok(()) } - ReaderMessage::NewBlocks(block_hashes) => { - self.main_thread_sender - .send(PeerThreadMessage { - nonce: self.nonce, - message: PeerMessage::NewBlocks(block_hashes), - }) - .await?; - Ok(()) - } + ReaderMessage::Inventory(hashes) => match self.relay_policy { + RelayPolicy::BlocksOnly => { + let blocks: Vec = hashes + .into_iter() + .filter_map(|inv| match inv { + Inventory::Block(hash) + | Inventory::CompactBlock(hash) + | Inventory::WitnessBlock(hash) => Some(hash), + _ => None, + }) + .collect(); + if blocks.is_empty() { + return Ok(()); + } + self.main_thread_sender + .send(PeerThreadMessage { + nonce: self.nonce, + message: PeerMessage::NewBlocks(blocks), + }) + .await?; + Ok(()) + } + RelayPolicy::Transactions(_) => { + let requests: Vec = hashes + .into_iter() + .filter(|inv| { + matches!( + inv, + Inventory::Transaction(_) + | Inventory::WitnessTransaction(_) + | Inventory::WTx(_) + ) + }) + .collect(); + if requests.is_empty() { + return Ok(()); + } + let msg = message_generator.serialize(NetworkMessage::GetData(requests)); + self.write_bytes(writer, msg).await?; + Ok(()) + } + }, + ReaderMessage::Transaction(tx) => match &self.relay_policy { + RelayPolicy::Transactions(watch) => { + let watch_guard = watch.lock().await; + let tx_check = watch_guard.check_tx(tx); + match tx_check { + TransactionCheck::SentToClient => { + crate::debug!("Found unconfirmed transaction in mempool") + } + TransactionCheck::Failed => { + self.dialog.send_warning(Warning::ChannelDropped); + } + TransactionCheck::NoMatch => (), + } + Ok(()) + } + RelayPolicy::BlocksOnly => Ok(()), + }, ReaderMessage::GetData(requests) => { let mut tx_queue = self.tx_queue.lock().await; for inv in requests { diff --git a/src/network/peer_map.rs b/src/network/peer_map.rs index 359648f..7fdb811 100644 --- a/src/network/peer_map.rs +++ b/src/network/peer_map.rs @@ -1,5 +1,5 @@ use std::{ - collections::HashMap, + collections::{HashMap, HashSet}, fmt::Debug, net::{IpAddr, Ipv4Addr}, sync::Arc, @@ -9,12 +9,12 @@ use addrman::Record; use bitcoin::{ key::rand, p2p::{address::AddrV2, ServiceFlags}, - FeeRate, Network, + FeeRate, Network, OutPoint, ScriptBuf, Transaction, }; use rand::{rngs::StdRng, seq::IteratorRandom, SeedableRng}; use tokio::{ sync::{ - mpsc::{self, Sender}, + mpsc::{self, Receiver, Sender}, Mutex, }, task::JoinHandle, @@ -23,7 +23,10 @@ use tokio::{ use crate::{ broadcaster::BroadcastQueue, default_port_from_network, - network::{dns::bootstrap_dns, error::PeerError, peer::Peer, PeerId, PeerTimeoutConfig}, + network::{ + dns::bootstrap_dns, error::PeerError, gossip::GossipMonitor, peer::Peer, PeerId, + PeerTimeoutConfig, RelayPolicy, + }, BlockType, Dialog, TrustedPeer, TrustedPeerInner, }; @@ -43,6 +46,24 @@ pub(crate) struct ManagedPeer { handle: JoinHandle>, } +/// A peer that is used solely for watching incoming transaction gossip. +#[derive(Debug)] +struct GossipPeer { + id: PeerId, + ptx: Sender, + handle: JoinHandle>, +} + +#[derive(Debug, Default)] +enum GossipSubscriptionStatus { + Subscribed { + monitor: Arc>, + active_peer: Option, + }, + #[default] + Unsubscribed, +} + // The `PeerMap` manages connections with peers, adds and bans peers, and manages the peer database #[derive(Debug)] pub(crate) struct PeerMap { @@ -53,6 +74,7 @@ pub(crate) struct PeerMap { block_type: BlockType, mtx: Sender, map: HashMap, + gossip: GossipSubscriptionStatus, db: Arc>, connector: ConnectionType, whitelist: Whitelist, @@ -80,6 +102,7 @@ impl PeerMap { block_type, mtx, map: HashMap::new(), + gossip: GossipSubscriptionStatus::Unsubscribed, db: Arc::new(Mutex::new(AddressBook::new())), connector: connection_type, whitelist, @@ -89,7 +112,7 @@ impl PeerMap { } // Remove any finished connections - pub async fn clean(&mut self) { + pub fn clean(&mut self) { self.map.retain(|_, peer| !peer.handle.is_finished()); } @@ -106,8 +129,46 @@ impl PeerMap { self.whitelist.push(peer); } - // Send out a TCP connection to a new peer and begin tracking the task pub async fn dispatch(&mut self, loaded_peer: Record) -> Result<(), PeerError> { + let (id, peer) = self + .spawn_peer(loaded_peer, RelayPolicy::BlocksOnly) + .await?; + self.map.insert(id, peer); + Ok(()) + } + + pub async fn dispatch_gossip(&mut self, loaded_peer: Record) -> Result<(), PeerError> { + match core::mem::take(&mut self.gossip) { + GossipSubscriptionStatus::Subscribed { + monitor, + active_peer: _, + } => { + let relay_policy = RelayPolicy::Transactions(Arc::clone(&monitor)); + // Assign first in case the connection fails. The user still wants to be subscribed. + self.gossip = GossipSubscriptionStatus::Subscribed { + monitor: Arc::clone(&monitor), + active_peer: None, + }; + let (id, ManagedPeer { ptx, handle, .. }) = + self.spawn_peer(loaded_peer, relay_policy).await?; + // The connection loop is live. If it fails, it will be overridden by this function + // again. + self.gossip = GossipSubscriptionStatus::Subscribed { + monitor, + active_peer: Some(GossipPeer { id, ptx, handle }), + }; + Ok(()) + } + GossipSubscriptionStatus::Unsubscribed => Ok(()), + } + } + + // Send out a TCP connection to a new peer and begin tracking the task + async fn spawn_peer( + &mut self, + loaded_peer: Record, + relay_policy: RelayPolicy, + ) -> Result<(PeerId, ManagedPeer), PeerError> { let (ptx, prx) = mpsc::channel::(32); let (addr, port) = loaded_peer.network_addr(); if !self.connector.can_connect(&addr) { @@ -122,6 +183,7 @@ impl PeerMap { loaded_peer.clone(), self.network, self.block_type, + relay_policy, self.mtx.clone(), prx, Arc::clone(&self.dialog), @@ -143,7 +205,7 @@ impl PeerMap { }; let is_proxy = self.connector.is_proxy(); let handle = tokio::spawn(async move { peer.run(connection, is_proxy).await }); - self.map.insert( + Ok(( self.current_id, ManagedPeer { record: loaded_peer, @@ -151,8 +213,7 @@ impl PeerMap { ptx, handle, }, - ); - Ok(()) + )) } // Set the minimum fee rate this peer will accept @@ -189,6 +250,20 @@ impl PeerMap { pub async fn send_message(&self, nonce: PeerId, message: MainThreadMessage) { if let Some(peer) = self.map.get(&nonce) { let _ = peer.ptx.send(message).await; + return; + } + match &self.gossip { + GossipSubscriptionStatus::Subscribed { + monitor: _, + active_peer, + } => { + if let Some(peer) = active_peer { + if peer.id == nonce { + let _ = peer.ptx.send(message).await; + } + } + } + GossipSubscriptionStatus::Unsubscribed => (), } } @@ -309,4 +384,60 @@ impl PeerMap { db.ban(&peer.record); } } + + pub async fn subscribe_to_gossip( + &mut self, + scripts: HashSet, + txins: HashSet, + ) -> Option> { + match core::mem::take(&mut self.gossip) { + GossipSubscriptionStatus::Unsubscribed => { + let (tx, rx) = mpsc::channel(64); + self.gossip = GossipSubscriptionStatus::Subscribed { + monitor: Arc::new(Mutex::new(GossipMonitor::new(scripts, txins, tx))), + active_peer: None, + }; + Some(rx) + } + GossipSubscriptionStatus::Subscribed { + monitor, + active_peer, + } => { + monitor.lock().await.extend(scripts, txins); + self.gossip = GossipSubscriptionStatus::Subscribed { + monitor, + active_peer, + }; + None + } + } + } + + pub async fn unsubscribe_from_gossip(&mut self) { + match core::mem::take(&mut self.gossip) { + GossipSubscriptionStatus::Subscribed { + monitor: _, + active_peer, + } => { + if let Some(peer) = active_peer { + let _ = peer.ptx.send(MainThreadMessage::Disconnect).await; + } + self.gossip = GossipSubscriptionStatus::Unsubscribed; + } + GossipSubscriptionStatus::Unsubscribed => (), + } + } + + // Should a new connection be established to monitor transaction gossip. + pub fn needs_gossip_listener(&self) -> bool { + match &self.gossip { + GossipSubscriptionStatus::Subscribed { + monitor: _, + active_peer, + } => active_peer + .as_ref() + .is_none_or(|peer| peer.handle.is_finished()), + GossipSubscriptionStatus::Unsubscribed => false, + } + } } diff --git a/src/network/reader.rs b/src/network/reader.rs index 207b602..b7b968e 100644 --- a/src/network/reader.rs +++ b/src/network/reader.rs @@ -9,7 +9,7 @@ use bitcoin::{ message_network::VersionMessage, ServiceFlags, }, - Block, BlockHash, + Block, Transaction, }; use bitcoin::{FeeRate, Wtxid}; use tokio::io::AsyncRead; @@ -59,26 +59,17 @@ impl Reader { if inventory.len() > MAX_INV { return Some(ReaderMessage::Disconnect); } - let blocks: Vec = inventory - .into_iter() - .filter_map(|inv| match inv { - Inventory::Block(hash) - | Inventory::CompactBlock(hash) - | Inventory::WitnessBlock(hash) => Some(hash), - _ => None, - }) - .collect(); - if blocks.is_empty() { + if inventory.is_empty() { return None; } - Some(ReaderMessage::NewBlocks(blocks)) + Some(ReaderMessage::Inventory(inventory)) } NetworkMessage::GetData(inventory) => Some(ReaderMessage::GetData(inventory)), NetworkMessage::NotFound(_) => None, NetworkMessage::GetBlocks(_) => None, NetworkMessage::GetHeaders(_) => None, NetworkMessage::MemPool => None, - NetworkMessage::Tx(_) => None, + NetworkMessage::Tx(tx) => Some(ReaderMessage::Transaction(tx)), NetworkMessage::Block(block) => Some(ReaderMessage::Block(block)), NetworkMessage::Headers(headers) => { if headers.len() > MAX_HEADERS { @@ -158,7 +149,8 @@ pub(in crate::network) enum ReaderMessage { FilterHeaders(CFHeaders), Filter(CFilter), Block(Block), - NewBlocks(Vec), + Inventory(Vec), + Transaction(Transaction), Reject(RejectPayload), Disconnect, Verack, @@ -184,40 +176,3 @@ impl ReaderMessage { } } } - -#[cfg(test)] -mod tests { - use super::*; - - fn test_reader() -> Reader { - let (tx, _rx) = tokio::sync::mpsc::channel(1); - Reader::new( - MessageParser::V1(tokio::io::empty(), bitcoin::Network::Regtest), - tx, - ) - } - - #[test] - fn inv_parsing_surfaces_only_block_hashes() { - let reader = test_reader(); - let block = BlockHash::from_byte_array([1; 32]); - let witness_block = BlockHash::from_byte_array([2; 32]); - let txid = bitcoin::Txid::from_byte_array([3; 32]); - // Mixed inventory: only block hashes surface, in order. - let parsed = reader.parse_message(NetworkMessage::Inv(vec![ - Inventory::Transaction(txid), - Inventory::Block(block), - Inventory::WitnessBlock(witness_block), - ])); - assert!( - matches!(parsed, Some(ReaderMessage::NewBlocks(hashes)) if hashes == vec![block, witness_block]) - ); - // Transaction-only inventory remains ignored. - let parsed = reader.parse_message(NetworkMessage::Inv(vec![Inventory::Transaction(txid)])); - assert!(parsed.is_none()); - // Oversized inventory still disconnects. - let oversized = vec![Inventory::Block(block); MAX_INV + 1]; - let parsed = reader.parse_message(NetworkMessage::Inv(oversized)); - assert!(matches!(parsed, Some(ReaderMessage::Disconnect))); - } -} diff --git a/src/node.rs b/src/node.rs index 63b4450..608f7c8 100644 --- a/src/node.rs +++ b/src/node.rs @@ -283,6 +283,16 @@ impl Node { self.dialog.send_warning(Warning::ChannelDropped); }; } + ClientMessage::SubscribeGossip(request) => { + let (req, oneshot) = request.into_values(); + let recv = self.peer_map.subscribe_to_gossip(req.scripts, req.txins).await; + if oneshot.send(recv).is_err() { + self.dialog.send_warning(Warning::ChannelDropped); + }; + } + ClientMessage::UnsubscribeGossip => { + self.peer_map.unsubscribe_from_gossip().await; + } ClientMessage::NoOp => (), } } @@ -294,7 +304,7 @@ impl Node { // Connect to a new peer if we are not connected to enough async fn dispatch(&mut self) -> Result<(), NodeError> { - self.peer_map.clean().await; + self.peer_map.clean(); let live = self.peer_map.live(); let required = self.next_required_peers(); // Find more peers when lower than the desired threshold. @@ -312,6 +322,13 @@ impl Node { self.dialog.send_warning(Warning::CouldNotConnect); } } + if self.peer_map.needs_gossip_listener() { + if let Some(peer) = self.peer_map.next_peer().await { + if self.peer_map.dispatch_gossip(peer).await.is_err() { + self.dialog.send_warning(Warning::CouldNotConnect); + } + } + } Ok(()) } diff --git a/tests/core.rs b/tests/core.rs index 0b627c0..22771aa 100644 --- a/tests/core.rs +++ b/tests/core.rs @@ -8,7 +8,8 @@ use bip157::{ chain::{checkpoints::HashCheckpoint, BlockHeaderChanges, ChainState}, client::Client, node::Node, - Address, BlockHash, Event, Info, ServiceFlags, Transaction, TrustedPeer, Warning, + Address, BlockHash, Event, GossipMonitorRequestBuilder, Info, ServiceFlags, Transaction, + TrustedPeer, Warning, }; use bitcoin::{ absolute, @@ -666,6 +667,79 @@ async fn tx_can_broadcast() { .unwrap(); } +#[tokio::test] +async fn gossip_subscription() { + setup_debug_output(); + let (bitcoind, socket_addr) = start_bitcoind(true).unwrap(); + let rpc = &bitcoind.client; + let miner = rpc.new_address().unwrap(); + mine_blocks(rpc, &miner, 110, 5).await; + + let host = (IpAddr::V4(*socket_addr.ip()), Some(socket_addr.port())); + let builder = bip157::builder::Builder::new(bitcoin::Network::Regtest) + .chain_state(ChainState::Checkpoint(HashCheckpoint::from_genesis( + bitcoin::Network::Regtest, + ))) + .add_peer(host) + .add_peer(host); + let (node, client) = builder.build(); + tokio::task::spawn(async move { node.run().await }); + let Client { + requester, + info_rx, + warn_rx, + event_rx: mut channel, + } = client; + tokio::task::spawn(async move { print_logs(info_rx, warn_rx).await }); + + let best = best_hash(rpc); + sync_assert(&best, &mut channel).await; + + let mut rng = StdRng::seed_from_u64(20002); + let secret = SecretKey::new(&mut rng); + let secp = Secp256k1::new(); + let keypair = Keypair::from_secret_key(&secp, &secret); + let (internal_key, _) = keypair.x_only_public_key(); + let watch_address = Address::p2tr(&secp, internal_key, None, KnownHrp::Regtest); + let scripts = vec![watch_address.script_pubkey()]; + + let mut receiver = requester + .subscribe_to_gossip( + GossipMonitorRequestBuilder::from_expected_output_scripts(scripts).build(), + ) + .await + .unwrap() + .unwrap(); + assert_eq!(receiver.max_capacity(), 64); + + tokio::time::sleep(Duration::from_secs(5)).await; + + let repeat = requester + .subscribe_to_gossip( + GossipMonitorRequestBuilder::from_expected_output_scripts(Vec::new()).build(), + ) + .await + .unwrap(); + assert!( + repeat.is_none(), + "subscribing while active should extend in place and return None" + ); + + let sent = rpc + .send_to_address(&watch_address, Amount::from_sat(50_000)) + .unwrap(); + let expected_txid = sent.txid().unwrap(); + + let received = tokio::time::timeout(Duration::from_secs(30), receiver.recv()) + .await + .expect("timed out waiting for gossip transaction") + .expect("gossip channel closed unexpectedly"); + assert_eq!(received.compute_txid(), expected_txid); + + requester.shutdown().unwrap(); + rpc.stop().unwrap(); +} + #[tokio::test] async fn whitelist_only_sync() { setup_debug_output();