Merge pull request Blockstream#98 from mempool/junderw/batch-rpc

softsimon · web-flow · commit bffba5a48ab0 · 2024-08-26T16:16:49.000+02:00
Adds support for batch operations in electrum RPC
diff --git a/rust-toolchain b/rust-toolchain
@@ -1 +1 @@
-1.70
+1.80
diff --git a/src/electrum/client.rs b/src/electrum/client.rs
@@ -3,7 +3,6 @@ use std::convert::TryFrom;
 
 use bitcoin::hashes::Hash;
 pub use electrum_client::client::Client;
-pub use electrum_client::Error as ElectrumError;
 pub use electrum_client::ServerFeaturesRes;
 
 use crate::chain::BlockHash;
diff --git a/src/electrum/server.rs b/src/electrum/server.rs
@@ -189,7 +189,7 @@ impl Connection {
             .chain_err(|| "discovery is disabled")?;
 
         let features = params
-            .get(0)
+            .first()
             .chain_err(|| "missing features param")?
             .clone();
         let features = serde_json::from_value(features).chain_err(|| "invalid features")?;
@@ -203,7 +203,7 @@ impl Connection {
     }
 
     fn blockchain_block_header(&self, params: &[Value]) -> Result<Value> {
-        let height = usize_from_value(params.get(0), "height")?;
+        let height = usize_from_value(params.first(), "height")?;
         let cp_height = usize_from_value_or(params.get(1), "cp_height", 0)?;
 
         let raw_header_hex: String = self
@@ -226,7 +226,7 @@ impl Connection {
     }
 
     fn blockchain_block_headers(&self, params: &[Value]) -> Result<Value> {
-        let start_height = usize_from_value(params.get(0), "start_height")?;
+        let start_height = usize_from_value(params.first(), "start_height")?;
         let count = MAX_HEADERS.min(usize_from_value(params.get(1), "count")?);
         let cp_height = usize_from_value_or(params.get(2), "cp_height", 0)?;
         let heights: Vec<usize> = (start_height..(start_height + count)).collect();
@@ -261,7 +261,7 @@ impl Connection {
     }
 
     fn blockchain_estimatefee(&self, params: &[Value]) -> Result<Value> {
-        let conf_target = usize_from_value(params.get(0), "blocks_count")?;
+        let conf_target = usize_from_value(params.first(), "blocks_count")?;
         let fee_rate = self
             .query
             .estimate_fee(conf_target as u16)
@@ -277,7 +277,7 @@ impl Connection {
     }
 
     fn blockchain_scripthash_subscribe(&mut self, params: &[Value]) -> Result<Value> {
-        let script_hash = hash_from_value(params.get(0)).chain_err(|| "bad script_hash")?;
+        let script_hash = hash_from_value(params.first()).chain_err(|| "bad script_hash")?;
 
         let history_txids = get_history(&self.query, &script_hash[..], self.txs_limit)?;
         let status_hash = get_status_hash(history_txids, &self.query)
@@ -295,7 +295,7 @@ impl Connection {
 
     #[cfg(not(feature = "liquid"))]
     fn blockchain_scripthash_get_balance(&self, params: &[Value]) -> Result<Value> {
-        let script_hash = hash_from_value(params.get(0)).chain_err(|| "bad script_hash")?;
+        let script_hash = hash_from_value(params.first()).chain_err(|| "bad script_hash")?;
         let (chain_stats, mempool_stats) = self.query.stats(&script_hash[..]);
 
         Ok(json!({
@@ -305,7 +305,7 @@ impl Connection {
     }
 
     fn blockchain_scripthash_get_history(&self, params: &[Value]) -> Result<Value> {
-        let script_hash = hash_from_value(params.get(0)).chain_err(|| "bad script_hash")?;
+        let script_hash = hash_from_value(params.first()).chain_err(|| "bad script_hash")?;
         let history_txids = get_history(&self.query, &script_hash[..], self.txs_limit)?;
 
         Ok(json!(history_txids
@@ -323,7 +323,7 @@ impl Connection {
     }
 
     fn blockchain_scripthash_listunspent(&self, params: &[Value]) -> Result<Value> {
-        let script_hash = hash_from_value(params.get(0)).chain_err(|| "bad script_hash")?;
+        let script_hash = hash_from_value(params.first()).chain_err(|| "bad script_hash")?;
         let utxos = self.query.utxo(&script_hash[..])?;
 
         let to_json = |utxo: Utxo| {
@@ -351,7 +351,7 @@ impl Connection {
     }
 
     fn blockchain_transaction_broadcast(&self, params: &[Value]) -> Result<Value> {
-        let tx = params.get(0).chain_err(|| "missing tx")?;
+        let tx = params.first().chain_err(|| "missing tx")?;
         let tx = tx.as_str().chain_err(|| "non-string tx")?.to_string();
         let txid = self.query.broadcast_raw(&tx)?;
         if let Err(e) = self.chan.sender().try_send(Message::PeriodicUpdate) {
@@ -361,7 +361,7 @@ impl Connection {
     }
 
     fn blockchain_transaction_get(&self, params: &[Value]) -> Result<Value> {
-        let tx_hash = Txid::from(hash_from_value(params.get(0)).chain_err(|| "bad tx_hash")?);
+        let tx_hash = Txid::from(hash_from_value(params.first()).chain_err(|| "bad tx_hash")?);
         let verbose = match params.get(1) {
             Some(value) => value.as_bool().chain_err(|| "non-bool verbose value")?,
             None => false,
@@ -380,7 +380,7 @@ impl Connection {
     }
 
     fn blockchain_transaction_get_merkle(&self, params: &[Value]) -> Result<Value> {
-        let txid = Txid::from(hash_from_value(params.get(0)).chain_err(|| "bad tx_hash")?);
+        let txid = Txid::from(hash_from_value(params.first()).chain_err(|| "bad tx_hash")?);
         let height = usize_from_value(params.get(1), "height")?;
         let blockid = self
             .query
@@ -399,7 +399,7 @@ impl Connection {
     }
 
     fn blockchain_transaction_id_from_pos(&self, params: &[Value]) -> Result<Value> {
-        let height = usize_from_value(params.get(0), "height")?;
+        let height = usize_from_value(params.first(), "height")?;
         let tx_pos = usize_from_value(params.get(1), "tx_pos")?;
         let want_merkle = bool_from_value_or(params.get(2), "merkle", false)?;
 
@@ -513,26 +513,15 @@ impl Connection {
     }
 
     fn handle_replies(&mut self, shutdown: crossbeam_channel::Receiver<()>) -> Result<()> {
-        let empty_params = json!([]);
         loop {
             crossbeam_channel::select! {
                 recv(self.chan.receiver()) -> msg => {
                     let msg = msg.chain_err(|| "channel closed")?;
                     trace!("RPC {:?}", msg);
                     match msg {
                         Message::Request(line) => {
-                            let cmd: Value = from_str(&line).chain_err(|| "invalid JSON format")?;
-                            let reply = match (
-                                cmd.get("method"),
-                                cmd.get("params").unwrap_or(&empty_params),
-                                cmd.get("id"),
-                            ) {
-                                (Some(Value::String(method)), Value::Array(params), Some(id)) => {
-                                    self.handle_command(method, params, id)?
-                                }
-                                _ => bail!("invalid command: {}", cmd),
-                            };
-                            self.send_values(&[reply])?
+                            let result = self.handle_line(&line);
+                            self.send_values(&[result])?
                         }
                         Message::PeriodicUpdate => {
                             let values = self
@@ -554,6 +543,48 @@ impl Connection {
         }
     }
 
+    #[inline]
+    fn handle_line(&mut self, line: &String) -> Value {
+        if let Ok(json_value) = from_str(line) {
+            match json_value {
+                Value::Array(mut arr) => {
+                    for cmd in &mut arr {
+                        // Replace each cmd with its response in-memory.
+                        *cmd = self.handle_value(cmd);
+                    }
+                    Value::Array(arr)
+                }
+                cmd => self.handle_value(&cmd),
+            }
+        } else {
+            // serde_json was unable to parse
+            invalid_json_rpc(line)
+        }
+    }
+
+    #[inline]
+    fn handle_value(&mut self, value: &Value) -> Value {
+        match (
+            value.get("method"),
+            value.get("params").unwrap_or(&json!([])),
+            value.get("id"),
+        ) {
+            (Some(Value::String(method)), Value::Array(params), Some(id)) => self
+                .handle_command(method, params, id)
+                .unwrap_or_else(|err| {
+                    json!({
+                        "error": {
+                            "code": 1,
+                            "message": format!("{method} RPC error: {err}")
+                        },
+                        "id": id,
+                        "jsonrpc": "2.0"
+                    })
+                }),
+            _ => invalid_json_rpc(value),
+        }
+    }
+
     fn handle_requests(
         mut reader: BufReader<ConnectionStream>,
         tx: crossbeam_channel::Sender<Message>,
@@ -629,6 +660,18 @@ impl Connection {
     }
 }
 
+#[inline]
+fn invalid_json_rpc(input: impl core::fmt::Display) -> Value {
+    json!({
+        "error": {
+            "code": -32600,
+            "message": format!("invalid request: {input}")
+        },
+        "id": null,
+        "jsonrpc": "2.0"
+    })
+}
+
 fn get_history(
     query: &Query,
     scripthash: &[u8],
diff --git a/src/elements/asset.rs b/src/elements/asset.rs
@@ -204,10 +204,7 @@ pub fn index_mempool_tx_assets(
 ) {
     let (history, issuances) = index_tx_assets(tx, network, parent_network);
     for (asset_id, info) in history {
-        asset_history
-            .entry(asset_id)
-            .or_insert_with(Vec::new)
-            .push(info);
+        asset_history.entry(asset_id).or_default().push(info);
     }
     for (asset_id, issuance) in issuances {
         asset_issuance.insert(asset_id, issuance);
@@ -386,7 +383,7 @@ pub fn lookup_asset(
     Ok(if let Some(row) = row {
         let reissuance_token = parse_asset_id(&row.reissuance_token);
 
-        let meta = meta.map(Clone::clone).or_else(|| match registry {
+        let meta = meta.cloned().or_else(|| match registry {
             Some(AssetRegistryLock::RwLock(rwlock)) => {
                 rwlock.read().unwrap().get(asset_id).cloned()
             }
diff --git a/src/new_index/mempool.rs b/src/new_index/mempool.rs
@@ -407,7 +407,7 @@ impl Mempool {
     }
 
     pub fn add_by_txid(&mut self, daemon: &Daemon, txid: &Txid) -> Result<()> {
-        if self.txstore.get(txid).is_none() {
+        if !self.txstore.contains_key(txid) {
             if let Ok(tx) = daemon.getmempooltx(txid) {
                 if self.add(vec![tx]) == 0 {
                     return Err(format!(
@@ -524,10 +524,7 @@ impl Mempool {
 
             // Index funding/spending history entries and spend edges
             for (scripthash, entry) in funding.chain(spending) {
-                self.history
-                    .entry(scripthash)
-                    .or_insert_with(Vec::new)
-                    .push(entry);
+                self.history.entry(scripthash).or_default().push(entry);
             }
             for (i, txi) in tx.input.iter().enumerate() {
                 self.edges.insert(txi.previous_output, (txid, i as u32));
diff --git a/src/new_index/schema.rs b/src/new_index/schema.rs
@@ -1654,7 +1654,7 @@ impl TxHistoryRow {
     }
 
     fn prefix_end(code: u8, hash: &[u8]) -> Bytes {
-        bincode_util::serialize_big(&(code, full_hash(hash), std::u32::MAX)).unwrap()
+        bincode_util::serialize_big(&(code, full_hash(hash), u32::MAX)).unwrap()
     }
 
     fn prefix_height(code: u8, hash: &[u8], height: u32) -> Bytes {
diff --git a/src/util/block.rs b/src/util/block.rs
@@ -238,7 +238,7 @@ impl HeaderList {
         // Use the timestamp as the mtp of the genesis block.
         // Matches bitcoind's behaviour: bitcoin-cli getblock `bitcoin-cli getblockhash 0` | jq '.time == .mediantime'
         if height == 0 {
-            self.headers.get(0).unwrap().header.time
+            self.headers.first().unwrap().header.time
         } else if height > self.len() - 1 {
             0
         } else {
diff --git a/src/util/transaction.rs b/src/util/transaction.rs
@@ -340,18 +340,12 @@ pub(super) mod sigops {
         let last_witness = witness.last();
         match (witness_version, witness_program.len()) {
             (0, 20) => 1,
-            (0, 32) => {
-                if let Some(n) = last_witness
-                    .map(|sl| sl.iter().map(|v| Ok(*v)))
-                    .map(script::Script::from_byte_iter)
-                    // I only return Ok 2 lines up, so there is no way to error
-                    .map(|s| count_sigops(&s.unwrap(), true))
-                {
-                    n
-                } else {
-                    0
-                }
-            }
+            (0, 32) => last_witness
+                .map(|sl| sl.iter().map(|v| Ok(*v)))
+                .map(script::Script::from_byte_iter)
+                // I only return Ok 2 lines up, so there is no way to error
+                .map(|s| count_sigops(&s.unwrap(), true))
+                .unwrap_or_default(),
             _ => 0,
         }
     }

Original file line number	Diff line number	Diff line change
`@@ -1654,7 +1654,7 @@ impl TxHistoryRow {`
`1654`	`1654`	`}`
`1655`	`1655`
`1656`	`1656`	`fn prefix_end(code: u8, hash: &[u8]) -> Bytes {`
`1657`		`- bincode_util::serialize_big(&(code, full_hash(hash), std::u32::MAX)).unwrap()`
	`1657`	`+ bincode_util::serialize_big(&(code, full_hash(hash), u32::MAX)).unwrap()`
`1658`	`1658`	`}`
`1659`	`1659`
`1660`	`1660`	`fn prefix_height(code: u8, hash: &[u8], height: u32) -> Bytes {`
Original file line number	Diff line number	Diff line change
`@@ -238,7 +238,7 @@ impl HeaderList {`
`238`	`238`	`// Use the timestamp as the mtp of the genesis block.`
`239`	`239`	// Matches bitcoind's behaviour: bitcoin-cli getblock `bitcoin-cli getblockhash 0` \| jq '.time == .mediantime'
`240`	`240`	`if height == 0 {`
`241`		`- self.headers.get(0).unwrap().header.time`
	`241`	`+ self.headers.first().unwrap().header.time`
`242`	`242`	`} else if height > self.len() - 1 {`
`243`	`243`	`0`
`244`	`244`	`} else {`