rpc: Find utxo (#102)

* Create a simple filter builder

* Index txid

This is useful for fetching utxos for their txid

* Implement query logic

* query script hashes, not scripts

* Move block filters to a dedicated module

* Add a simple kv database for cfilters

Use a kv bucket to store all filters on disk

* feature: allow jsonrpc finding utxos

This commit update the gettxout (and removes findtxout) from our json
rpc that now uses compact block filters to find any given utxo. We also
use the same filters to figure if the TXO is spent or not.

crates/floresta-cli/src/main.rs

@@ -52,7 +52,6 @@ fn get_req(cmd: &Cli) -> (Vec<Box<RawValue>>, String) {
         Methods::GetRoots => "getroots",
         Methods::GetBlock { .. } => "getblock",
         Methods::GetPeerInfo => "getpeerinfo",
-        Methods::FindUtxo { .. } => "findtxout",
         Methods::ListTransactions => "gettransactions",
         Methods::Stop => "stop",
         Methods::AddNode { .. } => "addnode",
@@ -86,9 +85,6 @@ fn get_req(cmd: &Cli) -> (Vec<Box<RawValue>>, String) {
         Methods::GetBlock { hash } => vec![arg(hash)],
         Methods::GetPeerInfo => Vec::new(),
         Methods::ListTransactions => Vec::new(),
-        Methods::FindUtxo { height, txid, vout } => {
-            vec![arg(height), arg(txid), arg(vout)]
-        }
         Methods::Stop => Vec::new(),
         Methods::AddNode { node } => {
             vec![arg(node)]
@@ -137,10 +133,6 @@ pub enum Methods {
     /// Returns the hash of the block associated with height
     #[command(name = "getblockhash")]
     GetBlockHash { height: u32 },
-    /// Returns information about a transaction output, assuming it is cached by our watch
-    /// only wallet
-    #[command(name = "gettxout")]
-    GetTxOut { txid: Txid, vout: u32 },
     /// Returns the proof that one or more transactions were included in a block
     #[command(name = "gettxproof")]
     GetTxProof {
@@ -174,11 +166,11 @@ pub enum Methods {
     /// List all transactions we are watching
     #[command(name = "listtransactions")]
     ListTransactions,
-    /// Finds a TXO by its outpoint and block height. Since we don't have a UTXO set
-    /// the block height is required to find the UTXO. Note that this command doesn't
-    /// check if the UTXO is spent or not.
-    #[command(name = "findutxo")]
-    FindUtxo { height: u32, txid: Txid, vout: u32 },
+    /// Returns the value associated with a UTXO, if it's still not spent.
+    /// This function only works properly if we have the compact block filters
+    /// feature enabled
+    #[command(name = "gettxout")]
+    GetTxOut { txid: Txid, vout: u32 },
     /// Stops the node
     #[command(name = "stop")]
     Stop,

crates/floresta-compact-filters/src/kv_filter_database.rs

@@ -0,0 +1,56 @@
+use std::path::PathBuf;
+
+use bitcoin::util::bip158::BlockFilter;
+use kv::{Bucket, Config, Integer};
+
+use crate::BlockFilterStore;
+
+/// Stores the block filters insinde a kv database
+#[derive(Clone)]
+pub struct KvFilterStore {
+    bucket: Bucket<'static, Integer, Vec<u8>>,
+}
+
+impl KvFilterStore {
+    /// Creates a new [KvFilterStore] that stores it's content in `datadir`.
+    ///
+    /// If the path does't exist it'll be created. This store uses compression by default, if you
+    /// want to make more granular configuration over the underlying Kv database, use `with_config`
+    /// instead.
+    pub fn new(datadir: &PathBuf) -> Self {
+        let store = kv::Store::new(kv::Config {
+            path: datadir.to_owned(),
+            temporary: false,
+            use_compression: false,
+            flush_every_ms: None,
+            cache_capacity: None,
+            segment_size: None,
+        })
+        .expect("Could not open store");
+
+        let bucket = store.bucket(Some("cfilters")).unwrap();
+        KvFilterStore { bucket }
+    }
+    /// Creates a new [KvFilterStore] that stores it's content with a given config
+    pub fn with_config(config: Config) -> Self {
+        let store = kv::Store::new(config).expect("Could not open database");
+        let bucket = store.bucket(Some("cffilters")).unwrap();
+        KvFilterStore { bucket }
+    }
+}
+
+impl BlockFilterStore for KvFilterStore {
+    fn get_filter(&self, block_height: u64) -> Option<bitcoin::util::bip158::BlockFilter> {
+        let value = self
+            .bucket
+            .get(&Integer::from(block_height))
+            .ok()
+            .flatten()?;
+        Some(BlockFilter::new(&value))
+    }
+    fn put_filter(&self, block_height: u64, block_filter: bitcoin::util::bip158::BlockFilter) {
+        self.bucket
+            .set(&Integer::from(block_height), &block_filter.content)
+            .expect("Bucket should be open");
+    }
+}

crates/floresta-compact-filters/src/lib.rs

@@ -22,7 +22,7 @@ use log::error;
 use std::io::Write;
 
 /// A database that stores our compact filters
-pub trait BlockFilterStore {
+pub trait BlockFilterStore: Send + Sync {
     /// Fetches a block filter
     fn get_filter(&self, block_height: u64) -> Option<bip158::BlockFilter>;
     /// Stores a new filter
@@ -140,16 +140,20 @@ impl BlockFilterBackend {
             k1: u64::from_le_bytes(k1),
         }
     }
+    /// Returns a given filter
     pub fn get_filter(&self, block_height: u32) -> Option<bip158::BlockFilter> {
         self.storage.get_filter(block_height as u64)
     }
+
     /// Build and index a given block height
     pub fn filter_block(&self, block: &Block, block_height: u64) -> Result<(), bip158::Error> {
         let mut writer = Vec::new();
         let mut filter = FilterBuilder::new(&mut writer, FILTER_M, FILTER_P, self.k0, self.k1);
+
         if self.index_inputs {
             self.write_inputs(&block.txdata, &mut filter);
         }
+
         if self.index_txids {
             self.write_txids(&block.txdata, &mut filter);
         }
@@ -158,48 +162,52 @@ impl BlockFilterBackend {
         filter.finish()?;
 
         let filter = BlockFilter::new(writer.as_slice());
-        self.storage.put_filter(block_height, filter);
 
+        self.storage.put_filter(block_height, filter);
         Ok(())
     }
+
     /// Maches a set of filters against out current set of filters
     ///
     /// This function will run over each filter inside the range `[start, end]` and sees
     /// if at least one query mathes. It'll return a vector of block heights where it matches.
     /// you should download those blocks and see what if there's anything interesting.
     pub fn match_any(&self, start: u64, end: u64, query: &[QueryType]) -> Option<Vec<u64>> {
-        let mut values = query.iter().map(|filter| filter.as_slice());
-
         let mut blocks = Vec::new();
+        let key = BlockHash::from_inner(self.key);
+        let values = query
+            .iter()
+            .map(|filter| filter.as_slice())
+            .collect::<Vec<_>>();
 
         for i in start..=end {
-            if self
-                .storage
-                .get_filter(i)?
-                .match_any(&BlockHash::from_inner(self.key), &mut values)
-                .ok()?
-            {
-                blocks.push(i);
+            if let Some(result) = self.storage.get_filter(i) {
+                let result = result.match_any(&key, &mut values.iter().copied()).ok()?;
+                if result {
+                    blocks.push(i);
+                }
             }
         }
-
         Some(blocks)
     }
+
     fn write_txids(&self, txs: &Vec<Transaction>, filter: &mut FilterBuilder) {
         for tx in txs {
             filter.put(tx.txid().as_inner());
         }
     }
+
     fn write_inputs(&self, txs: &Vec<Transaction>, filter: &mut FilterBuilder) {
         for tx in txs {
             tx.input.iter().for_each(|input| {
                 let mut ser_input = [0; 36];
                 ser_input[0..32].clone_from_slice(&input.previous_output.txid);
                 ser_input[32..].clone_from_slice(&input.previous_output.vout.to_be_bytes());
                 filter.put(&ser_input);
-            });
+            })
         }
     }
+
     fn write_tx_outs(&self, tx: &Transaction, filter: &mut FilterBuilder) {
         for output in tx.output.iter() {
             let hash = floresta_common::get_spk_hash(&output.script_pubkey);
@@ -220,6 +228,7 @@ impl BlockFilterBackend {
             }
         }
     }
+
     fn write_outputs(&self, txs: &Vec<Transaction>, filter: &mut FilterBuilder) {
         for tx in txs {
             self.write_tx_outs(tx, filter);
@@ -322,6 +331,7 @@ impl FilterBackendBuilder {
 }
 
 /// A serialized output that can be queried against our filter
+#[derive(Debug)]
 pub struct QueriableOutpoint(pub(crate) [u8; 36]);
 
 impl From<OutPoint> for QueriableOutpoint {
@@ -334,6 +344,7 @@ impl From<OutPoint> for QueriableOutpoint {
 }
 
 /// The type of value we are looking for in a filter.
+#[derive(Debug)]
 pub enum QueryType {
     /// We are looking for a specific outpoint being spent
     Input(QueriableOutpoint),
@@ -365,6 +376,10 @@ pub struct MemoryBlockFilterStorage {
     filters: RefCell<Vec<bip158::BlockFilter>>,
 }
 
+#[cfg(test)]
+#[doc(hidden)]
+unsafe impl Sync for MemoryBlockFilterStorage {}
+
 #[doc(hidden)]
 #[cfg(test)]
 impl BlockFilterStore for MemoryBlockFilterStorage {
@@ -399,7 +414,7 @@ impl<'a> KvFiltersStore<'a> {
     }
 }
 
-impl BlockFilterStore for KvFiltersStore<'_> {
+impl<'a> BlockFilterStore for KvFiltersStore<'a> {
     fn get_filter(&self, block_height: u64) -> Option<bip158::BlockFilter> {
         self.bucket
             .get(&block_height.into())

crates/floresta-wire/src/p2p_wire/node.rs

@@ -1096,6 +1096,7 @@ where
         // Use this node state to Initial Block download
         let mut ibd = UtreexoNode(self.0, IBDNode::default());
         try_and_log!(UtreexoNode::<IBDNode, Chain>::run(&mut ibd, kill_signal).await);
+
         // Then take the final state and run the node
         self = UtreexoNode(ibd.0, self.1);
 
@@ -1107,7 +1108,8 @@ where
             .await;
 
         loop {
-            while let Ok(notification) = timeout(Duration::from_secs(1), self.node_rx.recv()).await
+            while let Ok(notification) =
+                timeout(Duration::from_millis(1), self.node_rx.recv()).await
             {
                 try_and_log!(self.handle_notification(notification).await);
             }

florestad/src/json_rpc/res.rs

@@ -18,6 +18,7 @@ pub struct GetBlockchainInfoRes {
     pub progress: f32,
     pub difficulty: u64,
 }
+
 #[derive(Deserialize, Serialize)]
 pub struct RawTxJson {
     pub in_active_chain: bool,
@@ -36,12 +37,14 @@ pub struct RawTxJson {
     pub blocktime: u32,
     pub time: u32,
 }
+
 #[derive(Deserialize, Serialize)]
 pub struct TxOutJson {
     pub value: u64,
     pub n: u32,
     pub script_pub_key: ScriptPubKeyJson,
 }
+
 #[derive(Deserialize, Serialize)]
 pub struct ScriptPubKeyJson {
     pub asm: String,
@@ -51,6 +54,7 @@ pub struct ScriptPubKeyJson {
     pub type_: String,
     pub address: String,
 }
+
 #[derive(Deserialize, Serialize)]
 pub struct TxInJson {
     pub txid: String,
@@ -59,11 +63,13 @@ pub struct TxInJson {
     pub sequence: u32,
     pub witness: Vec<String>,
 }
+
 #[derive(Deserialize, Serialize)]
 pub struct ScriptSigJson {
     pub asm: String,
     pub hex: String,
 }
+
 #[derive(Deserialize, Serialize)]
 pub struct BlockJson {
     pub hash: String,
@@ -85,6 +91,7 @@ pub struct BlockJson {
     pub chainwork: String,
     pub n_tx: usize,
     pub previousblockhash: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
     pub nextblockhash: Option<String>,
 }
 
@@ -96,7 +103,10 @@ pub enum Error {
     Chain,
     InvalidPort,
     InvalidAddress,
+    Node,
+    NoBlockFilters,
 }
+
 impl Display for Error {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         let msg = match self {
@@ -106,10 +116,13 @@ impl Display for Error {
             Error::Chain => "Chain error",
             Error::InvalidPort => "Invalid port",
             Error::InvalidAddress => "Invalid address",
+            Error::Node => "Node returned an error",
+            Error::NoBlockFilters => "You don't have block filters enabled, please start florestad with --cfilters to run this RPC"
         };
         write!(f, "{}", msg)
     }
 }
+
 impl From<Error> for i64 {
     fn from(val: Error) -> Self {
         match val {
@@ -119,9 +132,12 @@ impl From<Error> for i64 {
             Error::InvalidDescriptor => 4,
             Error::InvalidPort => 5,
             Error::InvalidAddress => 6,
+            Error::Node => 7,
+            Error::NoBlockFilters => 8,
         }
     }
 }
+
 impl From<Error> for ErrorCode {
     fn from(val: Error) -> Self {
         let code = val.into();