Merge branch 'mempool' into docker_build

Author: knorrium

contrib/popular-scripts.txt

@@ -2,9 +2,14 @@
 #source "${HOME}/.cargo/env"
 #export PATH="${HOME}/.cargo/bin:${PATH}"
 
+# don't bother making electrs.core files
+ulimit -c
+
+# get credentials from elements.conf directly
 ELEMENTS_RPC_USER=$(grep 'rpcuser=' ${HOME}/elements.conf|cut -d = -f2|head -1)
 ELEMENTS_RPC_PASS=$(grep 'rpcpassword=' ${HOME}/elements.conf|cut -d = -f2|head -1)
 
+# run in loop in case of crash
 until false
 do
     cd "${HOME}/electrs"

electrs-start-liquid

@@ -2,9 +2,14 @@
 #source "${HOME}/.cargo/env"
 #export PATH="${HOME}/.cargo/bin:${PATH}"
 
+# don't bother making electrs.core files
+ulimit -c
+
+# get credentials from elements.conf directly
 ELEMENTS_RPC_USER=$(grep 'rpcuser=' "${HOME}/elements.conf"|cut -d = -f2|head -1)
 ELEMENTS_RPC_PASS=$(grep 'rpcpassword=' "${HOME}/elements.conf"|cut -d = -f2|head -1)
 
+# run in loop in case of crash
 until false
 do
     cd "${HOME}/electrs"

electrs-start-liquidtestnet

@@ -2,9 +2,14 @@
 #source "${HOME}/.cargo/env"
 #export PATH="${HOME}/.cargo/bin:${PATH}"
 
+# don't bother making electrs.core files
+ulimit -c
+
+# get credentials from bitcoin.conf directly
 BITCOIN_RPC_USER=$(grep 'rpcuser=' ${HOME}/bitcoin.conf|cut -d = -f2|head -1)
 BITCOIN_RPC_PASS=$(grep 'rpcpassword=' ${HOME}/bitcoin.conf|cut -d = -f2|head -1)
 
+# run in loop in case of crash
 until false
 do
 	cd "${HOME}/electrs"

electrs-start-mainnet

@@ -2,9 +2,14 @@
 #source "${HOME}/.cargo/env"
 #export PATH="${HOME}/.cargo/bin:${PATH}"
 
+# don't bother making electrs.core files
+ulimit -c
+
+# get credentials from bitcoin.conf directly
 BITCOIN_RPC_USER=$(grep 'rpcuser=' "${HOME}/bitcoin.conf"|cut -d = -f2|head -1)
 BITCOIN_RPC_PASS=$(grep 'rpcpassword=' "${HOME}/bitcoin.conf"|cut -d = -f2|head -1)
 
+# run in loop in case of crash
 until false
 do
 	cd "${HOME}/electrs"

electrs-start-signet

@@ -2,9 +2,14 @@
 #source "${HOME}/.cargo/env"
 #export PATH="${HOME}/.cargo/bin:${PATH}"
 
+# don't bother making electrs.core files
+ulimit -c
+
+# get credentials from bitcoin.conf directly
 BITCOIN_RPC_USER=$(grep 'rpcuser=' ${HOME}/bitcoin.conf|cut -d = -f2|head -1)
 BITCOIN_RPC_PASS=$(grep 'rpcpassword=' ${HOME}/bitcoin.conf|cut -d = -f2|head -1)
 
+# run in loop in case of crash
 until false
 do
 	cd "${HOME}/electrs"

electrs-start-testnet

@@ -1,5 +1,5 @@
 use std::collections::{HashMap, HashSet};
-use std::sync::{Arc, RwLock};
+use std::sync::{Arc, RwLock, RwLockReadGuard};
 
 use bitcoin::hashes::{hex::FromHex, sha256, Hash};
 use elements::confidential::{Asset, Value};
@@ -345,9 +345,14 @@ fn asset_history_row(
     TxHistoryRow { key }
 }
 
+pub enum AssetRegistryLock<'a> {
+    RwLock(&'a Arc<RwLock<AssetRegistry>>),
+    RwLockReadGuard(&'a RwLockReadGuard<'a, AssetRegistry>),
+}
+
 pub fn lookup_asset(
     query: &Query,
-    registry: Option<&Arc<RwLock<AssetRegistry>>>,
+    registry: Option<AssetRegistryLock>,
     asset_id: &AssetId,
     meta: Option<&AssetMeta>, // may optionally be provided if already known
 ) -> Result<Option<LiquidAsset>> {
@@ -362,7 +367,8 @@ pub fn lookup_asset(
     }
 
     let history_db = query.chain().store().history_db();
-    let mempool_issuances = &query.mempool().asset_issuance;
+    let mempool = query.mempool();
+    let mempool_issuances = &mempool.asset_issuance;
 
     let chain_row = history_db
         .get(&[b"i", &asset_id.into_inner()[..]].concat())
@@ -375,11 +381,14 @@ pub fn lookup_asset(
     Ok(if let Some(row) = row {
         let reissuance_token = parse_asset_id(&row.reissuance_token);
 
-        let registry = registry.map(|r| r.read().unwrap());
-        let meta = meta
-            .or_else(|| registry.as_ref().and_then(|r| r.get(asset_id)))
-            .cloned();
-        let stats = issued_asset_stats(query, asset_id, &reissuance_token);
+        let meta = meta.map(Clone::clone).or_else(|| match registry {
+            Some(AssetRegistryLock::RwLock(rwlock)) => {
+                rwlock.read().unwrap().get(asset_id).cloned()
+            }
+            Some(AssetRegistryLock::RwLockReadGuard(guard)) => guard.get(asset_id).cloned(),
+            None => None,
+        });
+        let stats = issued_asset_stats(query.chain(), &mempool, asset_id, &reissuance_token);
         let status = query.get_tx_status(&deserialize(&row.issuance_txid).unwrap());
 
         let asset = IssuedAsset::new(asset_id, row, stats, meta, status);
@@ -459,21 +468,20 @@ fn pegged_asset_stats(query: &Query, asset_id: &AssetId) -> (PeggedAssetStats, P
 
 // Get stats for issued assets
 fn issued_asset_stats(
-    query: &Query,
+    chain: &ChainQuery,
+    mempool: &Mempool,
     asset_id: &AssetId,
     reissuance_token: &AssetId,
 ) -> (IssuedAssetStats, IssuedAssetStats) {
     let afn = apply_issued_asset_stats;
 
-    let chain = query.chain();
     let mut chain_stats = chain_asset_stats(chain, asset_id, afn);
     chain_stats.burned_reissuance_tokens =
         chain_asset_stats(chain, reissuance_token, afn).burned_amount;
 
-    let mempool = query.mempool();
-    let mut mempool_stats = mempool_asset_stats(&mempool, asset_id, afn);
+    let mut mempool_stats = mempool_asset_stats(mempool, asset_id, afn);
     mempool_stats.burned_reissuance_tokens =
-        mempool_asset_stats(&mempool, reissuance_token, afn).burned_amount;
+        mempool_asset_stats(mempool, reissuance_token, afn).burned_amount;
 
     (chain_stats, mempool_stats)
 }

src/elements/asset.rs

@@ -6,8 +6,9 @@ use bitcoin::consensus::encode::serialize;
 #[cfg(feature = "liquid")]
 use elements::{encode::serialize, AssetId};
 
-use std::collections::{BTreeSet, HashMap, HashSet};
+use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
 use std::iter::FromIterator;
+use std::ops::Bound::{Excluded, Unbounded};
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 
@@ -29,7 +30,7 @@ use crate::elements::asset;
 pub struct Mempool {
     chain: Arc<ChainQuery>,
     config: Arc<Config>,
-    txstore: HashMap<Txid, Transaction>,
+    txstore: BTreeMap<Txid, Transaction>,
     feeinfo: HashMap<Txid, TxFeeInfo>,
     history: HashMap<FullHash, Vec<TxHistoryInfo>>, // ScriptHash -> {history_entries}
     edges: HashMap<OutPoint, (Txid, u32)>,          // OutPoint -> (spending_txid, spending_vin)
@@ -62,7 +63,7 @@ impl Mempool {
     pub fn new(chain: Arc<ChainQuery>, metrics: &Metrics, config: Arc<Config>) -> Self {
         Mempool {
             chain,
-            txstore: HashMap::new(),
+            txstore: BTreeMap::new(),
             feeinfo: HashMap::new(),
             history: HashMap::new(),
             edges: HashMap::new(),
@@ -289,6 +290,31 @@ impl Mempool {
         self.txstore.keys().collect()
     }
 
+    // Get all txs in the mempool
+    pub fn txs(&self) -> Vec<Transaction> {
+        let _timer = self.latency.with_label_values(&["txs"]).start_timer();
+        self.txstore.values().cloned().collect()
+    }
+
+    // Get n txs after the given txid in the mempool
+    pub fn txs_page(&self, n: usize, start: Option<Txid>) -> Vec<Transaction> {
+        let _timer = self.latency.with_label_values(&["txs"]).start_timer();
+        let mut page = Vec::with_capacity(n);
+        let start_bound = match start {
+            Some(txid) => Excluded(txid),
+            None => Unbounded,
+        };
+
+        self.txstore
+            .range((start_bound, Unbounded))
+            .take(n)
+            .for_each(|(_, value)| {
+                page.push(value.clone());
+            });
+
+        page
+    }
+
     // Get an overview of the most recent transactions
     pub fn recent_txs_overview(&self) -> Vec<&TxOverview> {
         // We don't bother ever deleting elements from the recent list.

src/new_index/mempool.rs

@@ -13,7 +13,7 @@ use crate::util::{is_spendable, BlockId, Bytes, TransactionStatus};
 
 #[cfg(feature = "liquid")]
 use crate::{
-    chain::AssetId,
+    chain::{asset::AssetRegistryLock, AssetId},
     elements::{lookup_asset, AssetRegistry, AssetSorting, LiquidAsset},
 };
 
@@ -232,7 +232,12 @@ impl Query {
 
     #[cfg(feature = "liquid")]
     pub fn lookup_asset(&self, asset_id: &AssetId) -> Result<Option<LiquidAsset>> {
-        lookup_asset(self, self.asset_db.as_ref(), asset_id, None)
+        lookup_asset(
+            self,
+            self.asset_db.as_ref().map(AssetRegistryLock::RwLock),
+            asset_id,
+            None,
+        )
     }
 
     #[cfg(feature = "liquid")]
@@ -251,8 +256,13 @@ impl Query {
         let results = results
             .into_iter()
             .map(|(asset_id, metadata)| {
-                lookup_asset(self, None, asset_id, Some(metadata))?
-                    .chain_err(|| "missing registered asset")
+                lookup_asset(
+                    self,
+                    Some(AssetRegistryLock::RwLockReadGuard(&asset_db)),
+                    asset_id,
+                    Some(metadata),
+                )?
+                .chain_err(|| "missing registered asset")
             })
             .collect::<Result<Vec<_>>>()?;
         Ok((total_num, results))

src/new_index/query.rs

@@ -549,7 +549,7 @@ async fn run_server(config: Arc<Config>, query: Arc<Query>, rx: oneshot::Receive
                             Response::builder()
                                 .status(err.0)
                                 .header("Content-Type", "text/plain")
-                                .header("Server", &**VERSION_STRING)
+                                .header("X-Powered-By", &**VERSION_STRING)
                                 .body(Body::from(err.1))
                                 .unwrap()
                         });
@@ -720,7 +720,7 @@ fn handle_request(
                 .status(StatusCode::OK)
                 .header("Content-Type", "application/octet-stream")
                 .header("Cache-Control", format!("public, max-age={:}", TTL_LONG))
-                .header("Server", &**VERSION_STRING)
+                .header("X-Powered-By", &**VERSION_STRING)
                 .body(Body::from(raw))
                 .unwrap())
         }
@@ -991,7 +991,7 @@ fn handle_request(
                 .status(StatusCode::OK)
                 .header("Content-Type", content_type)
                 .header("Cache-Control", format!("public, max-age={:}", ttl))
-                .header("Server", &**VERSION_STRING)
+                .header("X-Powered-By", &**VERSION_STRING)
                 .body(body)
                 .unwrap())
         }
@@ -1086,6 +1086,27 @@ fn handle_request(
         (&Method::GET, Some(&"mempool"), Some(&"txids"), None, None, None) => {
             json_response(query.mempool().txids(), TTL_SHORT)
         }
+        (&Method::GET, Some(&"mempool"), Some(&"txs"), Some(&"all"), None, None) => {
+            let txs = query
+                .mempool()
+                .txs()
+                .into_iter()
+                .map(|tx| (tx, None))
+                .collect();
+
+            json_maybe_error_response(prepare_txs(txs, query, config), TTL_SHORT)
+        }
+        (&Method::GET, Some(&"mempool"), Some(&"txs"), last_seen_txid, None, None) => {
+            let last_seen_txid = last_seen_txid.and_then(|txid| Txid::from_hex(txid).ok());
+            let txs = query
+                .mempool()
+                .txs_page(10_000, last_seen_txid)
+                .into_iter()
+                .map(|tx| (tx, None))
+                .collect();
+
+            json_maybe_error_response(prepare_txs(txs, query, config), TTL_SHORT)
+        }
         (&Method::GET, Some(&"mempool"), Some(&"recent"), None, None, None) => {
             let mempool = query.mempool();
             let recent = mempool.recent_txs_overview();
@@ -1117,7 +1138,7 @@ fn handle_request(
                 // Disable caching because we don't currently support caching with query string params
                 .header("Cache-Control", "no-store")
                 .header("Content-Type", "application/json")
-                .header("Server", &**VERSION_STRING)
+                .header("X-Powered-By", &**VERSION_STRING)
                 .header("X-Total-Results", total_num.to_string())
                 .body(Body::from(serde_json::to_string(&assets)?))
                 .unwrap())
@@ -1233,7 +1254,7 @@ where
         .status(status)
         .header("Content-Type", "text/plain")
         .header("Cache-Control", format!("public, max-age={:}", ttl))
-        .header("Server", &**VERSION_STRING)
+        .header("X-Powered-By", &**VERSION_STRING)
         .body(message.into())
         .unwrap())
 }
@@ -1243,7 +1264,7 @@ fn json_response<T: Serialize>(value: T, ttl: u32) -> Result<Response<Body>, Htt
     Ok(Response::builder()
         .header("Content-Type", "application/json")
         .header("Cache-Control", format!("public, max-age={:}", ttl))
-        .header("Server", &**VERSION_STRING)
+        .header("X-Powered-By", &**VERSION_STRING)
         .body(Body::from(value))
         .unwrap())
 }
@@ -1255,7 +1276,7 @@ fn json_maybe_error_response<T: Serialize>(
     let response = Response::builder()
         .header("Content-Type", "application/json")
         .header("Cache-Control", format!("public, max-age={:}", ttl))
-        .header("Server", &**VERSION_STRING);
+        .header("X-Powered-By", &**VERSION_STRING);
     Ok(match value {
         Ok(v) => response
             .body(Body::from(serde_json::to_string(&v)?))