Implement multi-transaction TxRaw lookup using MultiGet

- Change lookup_txns to use MultiGet

- Use lookup_txns for block transactions and reconstruction too
  (GET /block/:hash/txs and GET /block/:hash/raw)

(This was already possible with the V1 schema, but related to and
builds upon the other V2 changes.)

Plus some related changes:

- Remove expensive sanity check assertion in lookup_txn (involved txid
  computation and wasn't really necessary)

- Add test for raw block reconstruction

Author: shesek

src/new_index/schema.rs

@@ -494,6 +494,28 @@ impl ChainQuery {
         }
     }
 
+    pub fn get_block_txs(
+        &self,
+        hash: &BlockHash,
+        start_index: usize,
+        limit: usize,
+    ) -> Result<Vec<Transaction>> {
+        let txids = self.get_block_txids(hash).chain_err(|| "block not found")?;
+        ensure!(start_index < txids.len(), "start index out of range");
+
+        let txids_with_blockhash = txids
+            .into_iter()
+            .skip(start_index)
+            .take(limit)
+            .map(|txid| (txid, *hash))
+            .collect::<Vec<_>>();
+
+        self.lookup_txns(&txids_with_blockhash)
+
+        // XXX use getblock in lightmode? a single RPC call, but would fetch all txs to get one page
+        // self.daemon.getblock(hash)?.txdata.into_iter().skip(start_index).take(limit).collect()
+    }
+
     pub fn get_block_meta(&self, hash: &BlockHash) -> Option<BlockMeta> {
         let _timer = self.start_timer("get_block_meta");
 
@@ -519,17 +541,19 @@ impl ChainQuery {
             let entry = self.header_by_hash(hash)?;
             let meta = self.get_block_meta(hash)?;
             let txids = self.get_block_txids(hash)?;
+            let txids_with_blockhash: Vec<_> =
+                txids.into_iter().map(|txid| (txid, *hash)).collect();
+            let raw_txs = self.lookup_raw_txns(&txids_with_blockhash).ok()?; // TODO avoid hiding all errors as None, return a Result
 
             // Reconstruct the raw block using the header and txids,
             // as <raw header><tx count varint><raw txs>
             let mut raw = Vec::with_capacity(meta.size as usize);
 
             raw.append(&mut serialize(entry.header()));
-            raw.append(&mut serialize(&VarInt(txids.len() as u64)));
+            raw.append(&mut serialize(&VarInt(raw_txs.len() as u64)));
 
-            for txid in txids {
-                // we don't need to provide the blockhash because we know we're not in light mode
-                raw.append(&mut self.lookup_raw_txn(&txid, None)?);
+            for mut raw_tx in raw_txs {
+                raw.append(&mut raw_tx);
             }
 
             Some(raw)
@@ -589,7 +613,7 @@ impl ChainQuery {
     ) -> Vec<(Transaction, BlockId)> {
         let _timer_scan = self.start_timer("history");
         let headers = self.store.indexed_headers.read().unwrap();
-        let txs_conf = self
+        let history_iter = self
             .history_iter_scan_reverse(code, hash)
             .map(TxHistoryRow::from_row)
             .map(|row| (row.get_txid(), row.key.confirmed_height as usize))
@@ -605,16 +629,22 @@ impl ChainQuery {
                 None => 0,
             })
             // skip over entries that point to non-existing heights (may happen during reorg handling)
-            .filter_map(|(txid, height)| Some((txid, headers.header_by_height(height)?.into())))
-            .take(limit)
-            .collect::<Vec<(Txid, BlockId)>>();
+            .filter_map(|(txid, height)| Some((txid, headers.header_by_height(height)?)))
+            .take(limit);
+
+        let mut txids_with_blockhash = Vec::with_capacity(limit);
+        let mut blockids = Vec::with_capacity(limit);
+        for (txid, header) in history_iter {
+            txids_with_blockhash.push((txid, *header.hash()));
+            blockids.push(BlockId::from(header));
+        }
         drop(headers);
 
-        self.lookup_txns(&txs_conf)
+        self.lookup_txns(&txids_with_blockhash)
             .expect("failed looking up txs in history index")
             .into_iter()
-            .zip(txs_conf)
-            .map(|(tx, (_, blockid))| (tx, blockid))
+            .zip(blockids)
+            .map(|(tx, blockid)| (tx, blockid))
             .collect()
     }
 
@@ -926,26 +956,40 @@ impl ChainQuery {
             .clone()
     }
 
-    // TODO: can we pass txids as a "generic iterable"?
-    // TODO: should also use a custom ThreadPoolBuilder?
-    pub fn lookup_txns(&self, txids: &[(Txid, BlockId)]) -> Result<Vec<Transaction>> {
+    pub fn lookup_txns(&self, txids: &[(Txid, BlockHash)]) -> Result<Vec<Transaction>> {
         let _timer = self.start_timer("lookup_txns");
-        txids
-            .par_iter()
-            .map(|(txid, blockid)| {
-                self.lookup_txn(txid, Some(&blockid.hash))
-                    .chain_err(|| "missing tx")
-            })
-            .collect::<Result<Vec<Transaction>>>()
+        Ok(self
+            .lookup_raw_txns(txids)?
+            .into_iter()
+            .map(|rawtx| deserialize(&rawtx).expect("failed to parse Transaction"))
+            .collect())
     }
 
     pub fn lookup_txn(&self, txid: &Txid, blockhash: Option<&BlockHash>) -> Option<Transaction> {
         let _timer = self.start_timer("lookup_txn");
-        self.lookup_raw_txn(txid, blockhash).map(|rawtx| {
-            let txn: Transaction = deserialize(&rawtx).expect("failed to parse Transaction");
-            assert_eq!(*txid, txn.compute_txid());
-            txn
-        })
+        let rawtx = self.lookup_raw_txn(txid, blockhash)?;
+        Some(deserialize(&rawtx).expect("failed to parse Transaction"))
+    }
+
+    pub fn lookup_raw_txns(&self, txids: &[(Txid, BlockHash)]) -> Result<Vec<Bytes>> {
+        let _timer = self.start_timer("lookup_raw_txns");
+        if self.light_mode {
+            txids
+                .par_iter()
+                .map(|(txid, blockhash)| {
+                    self.lookup_raw_txn(txid, Some(blockhash))
+                        .chain_err(|| "missing tx")
+                })
+                .collect()
+        } else {
+            let keys = txids.iter().map(|(txid, _)| TxRow::key(&txid[..]));
+            self.store
+                .txstore_db
+                .multi_get(keys)
+                .into_iter()
+                .map(|val| val.unwrap().chain_err(|| "missing tx"))
+                .collect()
+        }
     }
 
     pub fn lookup_raw_txn(&self, txid: &Txid, blockhash: Option<&BlockHash>) -> Option<Bytes> {

src/rest.rs

@@ -718,41 +718,28 @@ fn handle_request(
         }
         (&Method::GET, Some(&"block"), Some(hash), Some(&"txs"), start_index, None) => {
             let hash = BlockHash::from_str(hash)?;
-            let txids = query
-                .chain()
-                .get_block_txids(&hash)
-                .ok_or_else(|| HttpError::not_found("Block not found".to_string()))?;
-
             let start_index = start_index
                 .map_or(0u32, |el| el.parse().unwrap_or(0))
                 .max(0u32) as usize;
-            if start_index >= txids.len() {
-                bail!(HttpError::not_found("start index out of range".to_string()));
-            } else if start_index % CHAIN_TXS_PER_PAGE != 0 {
-                bail!(HttpError::from(format!(
-                    "start index must be a multipication of {}",
-                    CHAIN_TXS_PER_PAGE
-                )));
-            }
 
-            // blockid_by_hash() only returns the BlockId for non-orphaned blocks,
-            // or None for orphaned
-            let confirmed_blockid = query.chain().blockid_by_hash(&hash);
+            ensure!(
+                start_index % CHAIN_TXS_PER_PAGE == 0,
+                "start index must be a multipication of {}",
+                CHAIN_TXS_PER_PAGE
+            );
+
+            // The BlockId would not be available for stale blocks
+            let blockid = query.chain().blockid_by_hash(&hash);
 
-            let txs = txids
-                .iter()
-                .skip(start_index)
-                .take(CHAIN_TXS_PER_PAGE)
-                .map(|txid| {
-                    query
-                        .lookup_txn(&txid)
-                        .map(|tx| (tx, confirmed_blockid.clone()))
-                        .ok_or_else(|| "missing tx".to_string())
-                })
-                .collect::<Result<Vec<(Transaction, Option<BlockId>)>, _>>()?;
+            let txs = query
+                .chain()
+                .get_block_txs(&hash, start_index, CHAIN_TXS_PER_PAGE)?
+                .into_iter()
+                .map(|tx| (tx, blockid))
+                .collect();
 
-            // XXX orphraned blocks alway get TTL_SHORT
-            let ttl = ttl_by_depth(confirmed_blockid.map(|b| b.height), query);
+            // XXX stale blocks alway get TTL_SHORT
+            let ttl = ttl_by_depth(blockid.map(|b| b.height), query);
 
             json_response(prepare_txs(txs, query, config), ttl)
         }

tests/rest.rs

@@ -1,3 +1,4 @@
+use bitcoin::hex::FromHex;
 use bitcoind::bitcoincore_rpc::RpcApi;
 use serde_json::Value;
 use std::collections::HashSet;
@@ -15,17 +16,9 @@ use common::Result;
 fn test_rest() -> Result<()> {
     let (rest_handle, rest_addr, mut tester) = common::init_rest_tester().unwrap();
 
-    let get_json = |path: &str| -> Result<Value> {
-        Ok(ureq::get(&format!("http://{}{}", rest_addr, path))
-            .call()?
-            .into_json::<Value>()?)
-    };
-
-    let get_plain = |path: &str| -> Result<String> {
-        Ok(ureq::get(&format!("http://{}{}", rest_addr, path))
-            .call()?
-            .into_string()?)
-    };
+    let get = |path: &str| ureq::get(&format!("http://{}{}", rest_addr, path)).call();
+    let get_json = |path: &str| -> Result<Value> { Ok(get(path)?.into_json::<Value>()?) };
+    let get_plain = |path: &str| -> Result<String> { Ok(get(path)?.into_string()?) };
 
     // Send transaction and confirm it
     let addr1 = tester.newaddress()?;
@@ -141,6 +134,14 @@ fn test_rest() -> Result<()> {
     );
     assert_eq!(res["tx_count"].as_u64(), Some(2));
 
+    // Test GET /block/:hash/raw
+    let mut res = get(&format!("/block/{}/raw", blockhash))?.into_reader();
+    let mut rest_rawblock = Vec::new();
+    res.read_to_end(&mut rest_rawblock).unwrap();
+    let node_hexblock = // uses low-level call() to support Elements
+        tester.call::<String>("getblock", &[blockhash.to_string().into(), 0.into()])?;
+    assert_eq!(rest_rawblock, Vec::from_hex(&node_hexblock).unwrap());
+
     // Test GET /block/:hash/txs
     let res = get_json(&format!("/block/{}/txs", blockhash))?;
     let block_txs = res.as_array().expect("list of txs");