Add transaction gossip listening via a dedicated peer

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<Option<mpsc::Receiver<Transaction>>, ClientError> {
+        let (tx, rx) = tokio::sync::oneshot::channel::<Option<mpsc::Receiver<Transaction>>>();
+        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()

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<Transaction> for Package {
     }
 }
 
+/// Instruct the node to watch for unconfirmed transactions with a certain criteria.
+#[derive(Debug, Clone)]
+pub struct GossipMonitorRequest {
+    scripts: HashSet<ScriptBuf>,
+    txins: HashSet<OutPoint>,
+}
+
+/// Build a new [`GossipMonitorRequest`].
+#[derive(Debug, Clone, Default)]
+pub struct GossipMonitorRequestBuilder {
+    scripts: HashSet<ScriptBuf>,
+    txins: HashSet<OutPoint>,
+}
+
+impl GossipMonitorRequestBuilder {
+    /// Watch for these **output** scripts in an unconfirmed transactions.
+    pub fn from_expected_output_scripts(scripts: impl IntoIterator<Item = ScriptBuf>) -> 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<Item = OutPoint>) -> 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<Item = ScriptBuf>) -> Self {
+        self.scripts.extend(scripts);
+        self
+    }
+
+    /// Add outpoints to the request.
+    pub fn add_outpoints(mut self, txins: impl IntoIterator<Item = OutPoint>) -> 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 {

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<u32, Option<IndexedHeader>>),
     /// Look up the height of a block hash in the chain of most work.
     HeightOfHash(ClientRequest<BlockHash, Option<u32>>),
+    /// Look for unconfirmed transactions.
+    SubscribeGossip(ClientRequest<GossipMonitorRequest, Option<Receiver<Transaction>>>),
+    /// Stop looking for unconfirmed transactions.
+    UnsubscribeGossip,
     /// Send an empty message to see if the node is running.
     NoOp,
 }

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<ScriptBuf>,
+    txins: HashSet<OutPoint>,
+    sender: Sender<Transaction>,
+}
+
+impl GossipMonitor {
+    pub(crate) fn new(
+        scripts: HashSet<ScriptBuf>,
+        txins: HashSet<OutPoint>,
+        sender: Sender<Transaction>,
+    ) -> Self {
+        Self {
+            scripts,
+            txins,
+            sender,
+        }
+    }
+
+    pub(crate) fn extend(
+        &mut self,
+        scripts: impl IntoIterator<Item = ScriptBuf>,
+        txins: impl IntoIterator<Item = OutPoint>,
+    ) {
+        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
+        }
+    }
+}

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<Mutex<GossipMonitor>>),
+}
+
 #[derive(Debug, Clone)]
 struct MessageState {
     general_timeout: Duration,

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<u16>) -> Vec<u8> {
-        let msg = NetworkMessage::Version(make_version(port, &self.network));
+    pub(in crate::network) fn version_message(&mut self, relay_policy: &RelayPolicy) -> Vec<u8> {
+        let msg = NetworkMessage::Version(make_version(relay_policy));
         self.serialize(msg)
     }
 
@@ -81,17 +81,17 @@ fn encrypt_plaintext(encryptor: &mut OutboundCipher, plaintext: Vec<u8>) -> Vec<
     encryptor.encrypt_to_vec(&plaintext, PacketType::Genuine, None)
 }
 
-pub(in crate::network) fn make_version(port: Option<u16>, 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<u16>, network: &Network) ->
         nonce: 1,
         user_agent: format!("/Rust BIP-157:{KYOTO_VERSION}/rust-bitcoin:{RUST_BITCOIN_VERSION}/"),
         start_height: 0,
-        relay: false,
+        relay,
     }
 }