Implement raw transaction store

Add txid->rawtx store, using 't' as the database prefix,
and a query.txstore_get() method to fetch txs from it.

Used for processing GET /tx/:txid requests, as well as for fetching the
prevout txs of inputs. Not used by electrs internally elsewhere or for the
Electrum server, which still fetch via the block height index and bitcoind.

Information about the confirmation status of transactions was removed from
REST API responses, since it is volatile and cannot be determined from the
rawtx index alone. This will be reintroduced as a separate endpoint.

(cherry picked from commit f4badf9)

src/index.rs

@@ -152,6 +152,46 @@ impl TxRow {
     }
 }
 
+pub struct RawTxRow {
+    pub key: TxKey,
+    pub rawtx: Bytes,
+}
+
+impl RawTxRow {
+    pub fn new(txid: &Sha256dHash, rawtx: Bytes) -> RawTxRow {
+        RawTxRow {
+            key: TxKey {
+                code: b't',
+                txid: full_hash(&txid[..]),
+            },
+            rawtx: rawtx,
+        }
+    }
+
+    pub fn filter_prefix(txid_prefix: &HashPrefix) -> Bytes {
+        [b"t", &txid_prefix[..]].concat()
+    }
+
+    pub fn filter_full(txid: &Sha256dHash) -> Bytes {
+        [b"t", &txid[..]].concat()
+    }
+
+    pub fn to_row(&self) -> Row {
+        Row {
+            key: bincode::serialize(&self.key).unwrap(),
+            value: bincode::serialize(&self.rawtx).unwrap(),
+        }
+    }
+
+    pub fn from_row(row: &Row) -> RawTxRow {
+        RawTxRow {
+            key: bincode::deserialize(&row.key).expect("failed to parse TxKey for RawTx"),
+            rawtx: bincode::deserialize(&row.value).expect("failed to parse rawtx"),
+        }
+    }
+}
+
+
 #[derive(Serialize, Deserialize)]
 struct BlockKey {
     code: u8,
@@ -180,6 +220,7 @@ pub fn index_transaction(txn: &Transaction, height: usize, rows: &mut Vec<Row>)
     }
     // Persist transaction ID and confirmed height
     rows.push(TxRow::new(&txid, height as u32).to_row());
+    rows.push(RawTxRow::new(&txid, serialize(txn).unwrap()).to_row()); // @TODO avoid re-serialization
 }
 
 pub fn index_block(block: &Block, height: usize) -> Vec<Row> {

src/query.rs

@@ -7,11 +7,11 @@ use std::collections::HashMap;
 use std::sync::{Arc, RwLock};
 
 use app::App;
-use index::{compute_script_hash, TxInRow, TxOutRow, TxRow};
+use index::{compute_script_hash, TxInRow, TxOutRow, TxRow, RawTxRow};
 use mempool::Tracker;
 use metrics::Metrics;
 use serde_json::Value;
-use store::ReadStore;
+use store::{ReadStore,Row};
 use util::{FullHash, HashPrefix, HeaderEntry};
 use lru_cache::LruCache;
 
@@ -122,6 +122,12 @@ fn merklize(left: Sha256dHash, right: Sha256dHash) -> Sha256dHash {
 }
 
 
+fn rawtxrow_by_txid(store: &ReadStore, txid: &Sha256dHash) -> Option<RawTxRow> {
+    let key = RawTxRow::filter_full(&txid);
+    let value = store.get(&key)?;
+    Some(RawTxRow::from_row(&Row { key, value }))
+}
+
 fn txrows_by_prefix(store: &ReadStore, txid_prefix: &HashPrefix) -> Vec<TxRow> {
     store
         .scan(&TxRow::filter_prefix(&txid_prefix))
@@ -318,7 +324,19 @@ impl Query {
         self.app.daemon().gettransaction(tx_hash)
     }
 
+    // Read from txstore, fallback to reading from bitcoind rpc
+    pub fn txstore_get(&self, txid: &Sha256dHash) -> Result<Transaction> {
+        match rawtxrow_by_txid(self.app.read_store(), txid) {
+            Some(row) => Ok(deserialize(&row.rawtx).chain_err(|| "cannot parse tx")?),
+            None => {
+                debug!("missing tx {} in txstore, asking node", txid);
+                self.app.daemon().gettransaction(txid)
+            }
+        }
+    }
+
     // Public API for transaction retrieval (for Electrum RPC)
+    // Fetched from bitcoind, includes tx confirmation information (number of confirmations and block hash)
     pub fn get_transaction(&self, tx_hash: &Sha256dHash, verbose: bool) -> Result<Value> {
         self.app
             .daemon()

src/rest.rs

@@ -1,5 +1,5 @@
 use bitcoin::util::hash::{Sha256dHash,HexError};
-use bitcoin::network::serialize::{serialize,deserialize};
+use bitcoin::network::serialize::serialize;
 use bitcoin::{Script,network,BitcoinHash};
 use config::Config;
 use elements::{Block,TxIn,TxOut,OutPoint,Transaction};
@@ -13,7 +13,7 @@ use lru_cache::LruCache;
 use query::Query;
 use serde_json;
 use serde::Serialize;
-use std::collections::HashMap;
+use std::collections::{HashMap,BTreeMap};
 use std::error::Error;
 use std::num::ParseIntError;
 use std::thread;
@@ -75,9 +75,7 @@ struct TransactionValue {
     txid: Sha256dHash,
     vin: Vec<TxInValue>,
     vout: Vec<TxOutValue>,
-    confirmations: Option<u32>,
-    hex: Option<String>,
-    block_hash: Option<String>,
+    hex: String,
     size: u32,
     weight: u32,
 }
@@ -92,10 +90,8 @@ impl From<Transaction> for TransactionValue {
             txid: tx.txid(),
             vin,
             vout,
-            confirmations: None,
-            hex: None,
-            block_hash: None,
             size: bytes.len() as u32,
+            hex: hex::encode(bytes),
             weight: tx.get_weight() as u32,
         }
     }
@@ -188,30 +184,46 @@ impl From<TxOut> for TxOutValue {
     }
 }
 
-// @XXX we should ideally set the scriptpubkey_address inside TxOutValue::from(), but it
-// cannot easily access the Network, so for now, we attach it later with mutation instead
 
-fn attach_tx_data(tx: &mut TransactionValue, network: &Network, query: &Arc<Query>) {
-    for mut vin in tx.vin.iter_mut() {
-        if vin.is_coinbase { continue; }
-
-        let prevtx = get_tx(&query, &vin.outpoint.txid).unwrap();
-        let mut prevout = prevtx.vout[vin.outpoint.vout as usize].clone();
-        prevout.scriptpubkey_address = script_to_address(&prevout.scriptpubkey_hex, &network);
-        vin.prevout = Some(prevout);
-    }
-
-    for mut vout in tx.vout.iter_mut() {
-        vout.scriptpubkey_address = script_to_address(&vout.scriptpubkey_hex, &network);
-    }
+fn attach_tx_data(tx: TransactionValue, network: &Network, query: &Arc<Query>) -> TransactionValue {
+    let mut txs = vec![tx];
+    attach_txs_data(&mut txs, network, query);
+    txs.remove(0)
 }
 
 fn attach_txs_data(txs: &mut Vec<TransactionValue>, network: &Network, query: &Arc<Query>) {
+    // a map of prev txids/vouts to lookup, with a reference to the "next in" that spends them
+    let mut lookups: BTreeMap<Sha256dHash, Vec<(u32, &mut TxInValue)>> = BTreeMap::new();
+    // using BTreeMap ensures the txid keys are in order. querying the db with keys in order leverage memory
+    // locality from empirical test up to 2 or 3 times faster
+
     for mut tx in txs.iter_mut() {
-        attach_tx_data(&mut tx, &network, &query);
+        // collect lookups
+        for mut vin in tx.vin.iter_mut() {
+            if !vin.is_coinbase {
+                lookups.entry(vin.outpoint.txid).or_insert(vec![]).push((vin.outpoint.vout, vin));
+            }
+        }
+        // attach encoded address (should ideally happen in TxOutValue::from(), but it cannot
+        // easily access the network)
+        for mut vout in tx.vout.iter_mut() {
+            vout.scriptpubkey_address = script_to_address(&vout.scriptpubkey_hex, &network);
+        }
     }
+
+    // fetch prevtxs and attach prevouts to nextins
+    for (prev_txid, prev_vouts) in lookups {
+        let prevtx = query.txstore_get(&prev_txid).unwrap();
+        for (prev_out_idx, ref mut nextin) in prev_vouts {
+            let mut prevout = TxOutValue::from(prevtx.output[prev_out_idx as usize].clone());
+            prevout.scriptpubkey_address = script_to_address(&prevout.scriptpubkey_hex, &network);
+            nextin.prevout = Some(prevout);
+        }
+    }
+
 }
 
+
 pub fn run_server(config: &Config, query: Arc<Query>) {
     let addr = ([127, 0, 0, 1], 3000).into();  // TODO take from config
     info!("REST server running on {}", addr);
@@ -286,22 +298,25 @@ fn handle_request(req: Request<Body>, query: &Arc<Query>, cache: &Arc<Mutex<LruC
         (&Method::GET, Some(&"block"), Some(hash), Some(&"with-txs")) => {
             let hash = Sha256dHash::from_hex(hash)?;
             let block = query.get_block_with_cache(&hash, &cache)?;
-            let block_hash = block.header.bitcoin_hash().be_hex_string();
-            let header_entry : HeaderEntry = query.get_best_header()?;
-            let confirmations = header_entry.height() as u32 - block.header.height + 1;
+
+            // @TODO optimization: skip deserializing transactions outside of range
+            let start = (query_params.get("start_index")
+                .map_or(0u32, |el| el.parse().unwrap_or(0))
+                .max(0u32) as usize).min(block.txdata.len());
+            let limit = query_params.get("limit")
+                .map_or(10u32,|el| el.parse().unwrap_or(10u32) )
+                .min(50u32) as usize;
+            let end = (start+limit).min(block.txdata.len());
+            let block = Block { header: block.header, txdata: block.txdata[start..end].to_vec() };
             let mut value = BlockAndTxsValue::from(block);
-            value.block_summary.confirmations = Some(confirmations);
-            for tx_value in value.txs.iter_mut() {
-                tx_value.confirmations = Some(confirmations);
-                tx_value.block_hash = Some(block_hash.clone());
-            }
-            attach_txs_data(&mut value.txs, &network, &query);
+            attach_txs_data(&mut value.txs, network, query);
             json_response(value)
         },
         (&Method::GET, Some(&"tx"), Some(hash), None) => {
             let hash = Sha256dHash::from_hex(hash)?;
-            let mut value = get_tx(&query, &hash)?;
-            attach_tx_data(&mut value, &network, &query);
+            let transaction = query.txstore_get(&hash)?;
+            let mut value = TransactionValue::from(transaction);
+            let value = attach_tx_data(value, network, query);
             json_response(value)
         },
         _ => {
@@ -342,32 +357,6 @@ fn redirect(status: StatusCode, path: String) -> Response<Body> {
         .unwrap()
 }
 
-fn get_tx(query: &Arc<Query>, hash: &Sha256dHash) -> Result<TransactionValue, StringError> {
-    let tx_value = query.get_transaction(&hash,true)?;
-    let tx_hex = tx_value
-        .get("hex").ok_or(StringError("hex not in tx json".to_string()))?
-        .as_str().ok_or(StringError("hex not a string".to_string()))?;
-
-    let confirmations = match tx_value.get("confirmations") {
-        Some(confs) => Some(confs.as_u64().ok_or(StringError("confirmations not a u64".to_string()))? as u32),
-        None => None
-    };
-
-    let blockhash = match tx_value.get("blockhash") {
-        Some(hash) => Some(hash.as_str().ok_or(StringError("blockhash not a string".to_string()))?.to_string()),
-        None => None
-    };
-
-    let tx : Transaction = deserialize(&hex::decode(tx_hex)? )?;
-
-    let mut value = TransactionValue::from(tx);
-    value.confirmations = confirmations;
-    value.hex = Some(tx_hex.to_string());
-    value.block_hash = blockhash;
-
-    Ok(value)
-}
-
 fn last_blocks(query: &Arc<Query>, limit: u32, block_cache : &Mutex<LruCache<Sha256dHash,Block>>) -> Result<Response<Body>,StringError> {
     let header_entry : HeaderEntry = query.get_best_header()?;
     blocks(query,&header_entry,limit, block_cache)