mempool/electrs
mempool/electrs
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases 3 Wiki Activity

Merge branch 'mempool' into mononaut/enable-liquid-asset-registry

Browse Source
This commit is contained in:
mononaut 2026-06-09 20:29:09 +09:00 committed by GitHub
commit 2fa977b5a9
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
10 changed files with 258 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
src/config.rs
View File

@ -69,14 +69,13 @@ pub struct Config {
pub electrum_max_subscriptions: usize,
pub electrum_max_clients: usize,
pub electrum_idle_timeout: u64,
pub electrum_public_hosts: Option<crate::electrum::ServerHosts>,
#[cfg(feature = "liquid")]
pub parent_network: BNetwork,
#[cfg(feature = "liquid")]
pub asset_db_path: Option<PathBuf>,
#[cfg(feature = "electrum-discovery")]
pub electrum_public_hosts: Option<crate::electrum::ServerHosts>,
#[cfg(feature = "electrum-discovery")]
pub electrum_announce: bool,
#[cfg(feature = "electrum-discovery")]
@ -521,6 +520,8 @@ impl Config {
let electrum_public_hosts = m
.value_of("electrum_public_hosts")
.map(|s| serde_json::from_str(s).expect("invalid --electrum-public-hosts"));
#[cfg(not(feature = "electrum-discovery"))]
let electrum_public_hosts: Option<crate::electrum::ServerHosts> = None;
let mut log = stderrlog::new();
log.verbosity(m.occurrences_of("verbosity") as usize);
@ -604,7 +605,6 @@ impl Config {
#[cfg(feature = "liquid")]
asset_db_path,
#[cfg(feature = "electrum-discovery")]
electrum_public_hosts,
#[cfg(feature = "electrum-discovery")]
electrum_announce: m.is_present("electrum_announce"),

35
src/electrum/server.rs
View File

@ -27,7 +27,7 @@ use elements::encode::serialize;
use crate::chain::Txid;
use crate::config::{Config, VERSION_STRING};
use crate::electrum::{get_electrum_height, ProtocolVersion};
use crate::electrum::{get_electrum_height, ProtocolVersion, ServerFeatures};
use crate::errors::*;
use crate::metrics::{Gauge, HistogramOpts, HistogramVec, MetricOpts, Metrics};
use crate::new_index::{Query, Utxo};
@ -41,7 +41,7 @@ const PROTOCOL_VERSION: ProtocolVersion = ProtocolVersion::new(1, 4);
const MAX_HEADERS: usize = 2016;
#[cfg(feature = "electrum-discovery")]
use crate::electrum::{DiscoveryManager, ServerFeatures};
use crate::electrum::DiscoveryManager;
// TODO: Sha256dHash should be a generic hash-container (since script hash is single SHA256)
fn hash_from_value(val: Option<&Value>) -> Result<Sha256dHash> {
@ -128,6 +128,7 @@ struct Connection {
idle_timeout: u64,
last_request_at: Instant,
die_please: Option<Receiver<()>>,
server_features: Arc<ServerFeatures>,
#[cfg(feature = "electrum-discovery")]
discovery: Option<Arc<DiscoveryManager>>,
}
@ -143,6 +144,7 @@ impl Connection {
max_subscriptions: usize,
idle_timeout: u64,
die_please: Receiver<()>,
server_features: Arc<ServerFeatures>,
#[cfg(feature = "electrum-discovery")] discovery: Option<Arc<DiscoveryManager>>,
) -> Connection {
Connection {
@ -158,6 +160,7 @@ impl Connection {
idle_timeout,
last_request_at: Instant::now(),
die_please: Some(die_please),
server_features,
#[cfg(feature = "electrum-discovery")]
discovery,
}
@ -179,13 +182,8 @@ impl Connection {
Ok(json!(self.query.config().electrum_banner.clone()))
}
#[cfg(feature = "electrum-discovery")]
fn server_features(&self) -> Result<Value> {
let discovery = self
.discovery
.as_ref()
.chain_err(|| "discovery is disabled")?;
Ok(json!(discovery.our_features()))
Ok(json!(self.server_features.as_ref()))
}
fn server_donation_address(&self) -> Result<Value> {
@ -490,9 +488,8 @@ impl Connection {
"server.peers.subscribe" => self.server_peers_subscribe(),
"server.ping" => Ok(Value::Null),
"server.version" => self.server_version(),
#[cfg(feature = "electrum-discovery")]
"server.features" => self.server_features(),
#[cfg(feature = "electrum-discovery")]
"server.add_peer" => self.server_add_peer(params),
@ -891,11 +888,10 @@ impl RPC {
let notification = Channel::unbounded();
// Discovery is enabled when electrum-public-hosts is set
#[cfg(feature = "electrum-discovery")]
let discovery = config.electrum_public_hosts.clone().map(|hosts| {
let server_features = {
use crate::chain::genesis_hash;
let features = ServerFeatures {
let hosts = config.electrum_public_hosts.clone().unwrap_or_default();
Arc::new(ServerFeatures {
hosts,
server_version: VERSION_STRING.clone(),
genesis_hash: genesis_hash(config.network_type),
@ -903,10 +899,15 @@ impl RPC {
protocol_max: PROTOCOL_VERSION,
hash_function: "sha256".into(),
pruning: None,
};
})
};
// Discovery is enabled when electrum-public-hosts is set
#[cfg(feature = "electrum-discovery")]
let discovery = config.electrum_public_hosts.as_ref().map(|_hosts| {
let discovery = Arc::new(DiscoveryManager::new(
config.network_type,
features,
server_features.as_ref().clone(),
PROTOCOL_VERSION,
config.electrum_announce,
config.tor_proxy,
@ -977,6 +978,7 @@ impl RPC {
#[cfg(feature = "electrum-discovery")]
let discovery = discovery.clone();
let server_features = Arc::clone(&server_features);
let spawned = spawn_thread("peer", move || {
let addr = stream.addr_string();
@ -990,6 +992,7 @@ impl RPC {
max_subscriptions,
idle_timeout,
peace_receiver,
server_features,
#[cfg(feature = "electrum-discovery")]
discovery,
);

2
src/elements/mod.rs
View File

@ -9,7 +9,7 @@ mod registry;
use asset::get_issuance_entropy;
pub use asset::{lookup_asset, LiquidAsset};
pub use registry::{AssetRegistry, AssetSorting};
pub use registry::{AssetMeta, AssetRegistry, AssetSorting};
#[derive(Serialize, Deserialize, Clone)]
pub struct IssuanceValue {

106
src/elements/registry.rs
View File

@ -14,6 +14,8 @@ use crate::errors::*;
// (in number of hex characters, not bytes)
const DIR_PARTITION_LEN: usize = 2;
const SEARCH_SORT_CANDIDATE_LIMIT: usize = 2000;
pub struct AssetRegistry {
directory: path::PathBuf,
assets_cache: HashMap<AssetId, (SystemTime, AssetMeta)>,
@ -53,6 +55,39 @@ impl AssetRegistry {
)
}
pub fn search(&self, query: &str, limit: usize) -> Vec<AssetEntry<'_>> {
let query = query.trim();
if query.is_empty() || limit == 0 {
return vec![];
}
let (mut results, candidates) = search_by(
self.assets_cache
.iter()
.map(|(asset_id, (_, metadata))| (asset_id, metadata)),
query,
limit,
|metadata| metadata.ticker.as_deref(),
);
if results.len() < limit {
let (name_matches, candidates) =
search_by(candidates, query, limit - results.len(), |metadata| {
Some(&metadata.name)
});
results.extend(name_matches);
if results.len() < limit {
let (domain_matches, _) =
search_by(candidates, query, limit - results.len(), AssetMeta::domain);
results.extend(domain_matches);
}
}
results.truncate(limit);
results
}
pub fn fs_sync(&mut self) -> Result<()> {
for entry in fs::read_dir(&self.directory).chain_err(|| "failed reading asset dir")? {
let entry = entry.chain_err(|| "invalid fh")?;
@ -126,7 +161,7 @@ pub struct AssetMeta {
}
impl AssetMeta {
fn domain(&self) -> Option<&str> {
pub(crate) fn domain(&self) -> Option<&str> {
self.entity["domain"].as_str()
}
}
@ -192,3 +227,72 @@ fn lc_cmp_opt(a: &Option<String>, b: &Option<String>) -> cmp::Ordering {
.map(|a| a.to_lowercase())
.cmp(&b.as_ref().map(|b| b.to_lowercase()))
}
fn search_by<'a, I, F>(
candidates: I,
query: &str,
limit: usize,
field: F,
) -> (Vec<AssetEntry<'a>>, Vec<AssetEntry<'a>>)
where
I: IntoIterator<Item = AssetEntry<'a>>,
F: Fn(&AssetMeta) -> Option<&str>,
{
let mut matches = vec![];
let mut remaining = vec![];
for (asset_id, metadata) in candidates {
let position = field(metadata).and_then(|field| {
// registry fields are ascii, so we don't need full unicode case-folding
ascii_ci_find(field, query).map(|position| (position, field))
});
if let Some((position, field)) = position {
if matches.len() >= SEARCH_SORT_CANDIDATE_LIMIT {
continue;
}
matches.push((position, field, asset_id, metadata));
} else {
remaining.push((asset_id, metadata));
}
}
matches.sort_unstable_by(|a, b| {
a.0.cmp(&b.0)
.then_with(|| ascii_ci_cmp(a.1, b.1))
.then_with(|| a.2.cmp(b.2))
});
(
matches
.into_iter()
.take(limit)
.map(|(_, _, asset_id, metadata)| (asset_id, metadata))
.collect(),
remaining,
)
}
// zero-allocation case-insensitive ASCII substring search
// returns the byte offset of the first match
fn ascii_ci_find(haystack: &str, needle: &str) -> Option<usize> {
let (haystack, needle) = (haystack.as_bytes(), needle.as_bytes());
if needle.is_empty() {
return Some(0);
}
haystack
.windows(needle.len())
.position(|window| window.eq_ignore_ascii_case(needle))
}
// zero-allocation case-insensitive ASCII string comparison
fn ascii_ci_cmp(a: &str, b: &str) -> cmp::Ordering {
let (a, b) = (a.as_bytes(), b.as_bytes());
for i in 0..a.len().min(b.len()) {
match a[i].to_ascii_lowercase().cmp(&b[i].to_ascii_lowercase()) {
cmp::Ordering::Equal => continue,
ord => return ord,
}
}
a.len().cmp(&b.len())
}

7
src/new_index/fetch.rs
View File

@ -290,12 +290,7 @@ fn parse_blocks(blob: Vec<u8>, magic: u32) -> Result<Vec<SizedBlock>> {
cursor.set_position(end);
}
let pool = rayon::ThreadPoolBuilder::new()
.num_threads(0) // CPU-bound
.thread_name(|i| format!("parse-blocks-{}", i))
.build()
.unwrap();
Ok(pool.install(|| {
Ok(super::THREAD_POOL.install(|| {
slices
.into_par_iter()
.map(|(slice, size)| (deserialize(slice).expect("failed to parse Block"), size))

10
src/new_index/mod.rs
View File

@ -5,6 +5,16 @@ pub mod precache;
mod query;
pub mod schema;
use std::sync::LazyLock;
pub(crate) static THREAD_POOL: LazyLock<rayon::ThreadPool> = LazyLock::new(|| {
rayon::ThreadPoolBuilder::new()
.num_threads(0) // 0 = use number of logical CPUs
.thread_name(|i| format!("electrs-worker-{}", i))
.build()
.expect("failed to create global rayon thread pool")
});
pub use self::db::{DBRow, DB};
pub use self::fetch::{BlockEntry, FetchFrom};
pub use self::mempool::Mempool;

34
src/new_index/query.rs
View File

@ -14,7 +14,7 @@ use crate::util::{is_spendable, BlockId, Bytes, TransactionStatus};
#[cfg(feature = "liquid")]
use crate::{
chain::{asset::AssetRegistryLock, AssetId},
elements::{lookup_asset, AssetRegistry, AssetSorting, LiquidAsset},
elements::{lookup_asset, AssetMeta, AssetRegistry, AssetSorting, LiquidAsset},
};
const FEE_ESTIMATES_TTL: u64 = 60; // seconds
@ -271,6 +271,15 @@ impl Query {
)
}
#[cfg(feature = "liquid")]
pub fn lookup_registry_asset(&self, asset_id: &AssetId) -> Result<Option<AssetMeta>> {
let asset_db = self
.asset_db
.as_ref()
.chain_err(|| "asset registry unavailable")?;
Ok(asset_db.read().unwrap().get(asset_id).cloned())
}
#[cfg(feature = "liquid")]
pub fn list_registry_assets(
&self,
@ -298,4 +307,27 @@ impl Query {
.collect::<Result<Vec<_>>>()?;
Ok((total_num, results))
}
#[cfg(feature = "liquid")]
pub fn search_registry_assets<T, F>(
&self,
search: &str,
limit: usize,
mut map: F,
) -> Result<Vec<T>>
where
F: FnMut(&AssetId, &AssetMeta) -> T,
{
let asset_db = self
.asset_db
.as_ref()
.chain_err(|| "asset registry unavailable")?;
Ok(asset_db
.read()
.unwrap()
.search(search, limit)
.into_iter()
.map(|(asset_id, metadata)| map(asset_id, metadata))
.collect())
}
}

19
src/new_index/schema.rs
View File

@ -1447,24 +1447,7 @@ fn lookup_txos(
outpoints: &BTreeSet<OutPoint>,
allow_missing: bool,
) -> HashMap<OutPoint, TxOut> {
let mut loop_count = 10;
let pool = loop {
match rayon::ThreadPoolBuilder::new()
.num_threads(16) // we need to saturate SSD IOPS
.thread_name(|i| format!("lookup-txo-{}", i))
.build()
{
Ok(pool) => break pool,
Err(e) => {
if loop_count == 0 {
panic!("schema::lookup_txos failed to create a ThreadPool: {}", e);
}
std::thread::sleep(std::time::Duration::from_millis(50));
loop_count -= 1;
}
}
};
pool.install(|| {
super::THREAD_POOL.install(|| {
// Should match lookup_txos_sequential
outpoints
.par_iter()

85
src/rest.rs
View File

@ -35,7 +35,7 @@ use hyperlocal::UnixServerExt;
use std::{cmp, fs};
#[cfg(feature = "liquid")]
use {
crate::elements::{peg::PegoutValue, AssetSorting, IssuanceValue},
crate::elements::{peg::PegoutValue, AssetMeta, AssetSorting, IssuanceValue},
elements::{
confidential::{Asset, Nonce, Value},
encode, AssetId,
@ -59,6 +59,12 @@ const MULTI_ADDRESS_LIMIT: usize = 300;
const ASSETS_PER_PAGE: usize = 25;
#[cfg(feature = "liquid")]
const ASSETS_MAX_PER_PAGE: usize = 100;
#[cfg(feature = "liquid")]
const ASSETS_SEARCH_DEFAULT_LIMIT: usize = 15;
#[cfg(feature = "liquid")]
const ASSETS_SEARCH_MAX_LIMIT: usize = 100;
#[cfg(feature = "liquid")]
const ASSETS_SEARCH_MAX_QUERY_LEN: usize = 64;
const TTL_LONG: u32 = 157_784_630; // ttl for static resources (5 years)
const TTL_SHORT: u32 = 10; // ttl for volatie resources
@ -132,6 +138,32 @@ impl BlockValue {
}
}
#[cfg(feature = "liquid")]
#[derive(Serialize)]
struct AssetRegistrySearchResult {
asset_id: AssetId,
name: String,
#[serde(skip_serializing_if = "Option::is_none")]
ticker: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
domain: Option<String>,
}
#[cfg(feature = "liquid")]
impl AssetRegistrySearchResult {
fn new(asset_id: &AssetId, meta: &AssetMeta) -> Self {
let domain = meta.domain().map(String::from);
Self {
asset_id: *asset_id,
name: meta.name.clone(),
ticker: meta.ticker.clone(),
domain,
}
}
}
/// Calculate the difficulty of a BlockHeader
/// using Bitcoin Core code ported to Rust.
///
@ -1774,6 +1806,42 @@ fn handle_request(
json_response(recent, TTL_MEMPOOL_RECENT)
}
(&Method::GET, Some(&"fee-estimates"), None, None, None, None) => {
json_response(query.estimate_fee_map(), TTL_SHORT)
}
#[cfg(feature = "liquid")]
(&Method::GET, Some(&"assets"), Some(&"registry"), Some(&"search"), None, None) => {
let search = query_params.get("q").map(|q| q.trim()).unwrap_or("");
let assets = if search.is_empty() {
vec![]
} else if search.chars().count() > ASSETS_SEARCH_MAX_QUERY_LEN {
return Err(HttpError(
StatusCode::BAD_REQUEST,
"search query too long".to_string(),
));
} else {
let limit = query_params
.get("limit")
.and_then(|n| n.parse::<usize>().ok())
.unwrap_or(ASSETS_SEARCH_DEFAULT_LIMIT)
.min(ASSETS_SEARCH_MAX_LIMIT);
query
.search_registry_assets(search, limit, AssetRegistrySearchResult::new)
.map_err(|e| {
HttpError(StatusCode::SERVICE_UNAVAILABLE, e.description().to_string())
})?
};
Ok(Response::builder()
// Disable caching because we don't currently support caching with query string params
.header("Cache-Control", "no-store")
.header("Content-Type", "application/json")
.body(Body::from(serde_json::to_string(&assets)?))
.unwrap())
}
#[cfg(feature = "liquid")]
(&Method::GET, Some(&"assets"), Some(&"registry"), None, None, None) => {
let start_index: usize = query_params
@ -1800,6 +1868,21 @@ fn handle_request(
.unwrap())
}
#[cfg(feature = "liquid")]
(&Method::GET, Some(&"assets"), Some(&"registry"), Some(asset_str), None, None) => {
let asset_id = AssetId::from_str(asset_str)?;
let registry_entry = query
.lookup_registry_asset(&asset_id)
.map_err(|e| {
HttpError(StatusCode::SERVICE_UNAVAILABLE, e.description().to_string())
})?
.ok_or_else(|| {
HttpError::not_found("Asset id not found in registry".to_string())
})?;
json_response(registry_entry, TTL_SHORT)
}
#[cfg(feature = "liquid")]
(&Method::GET, Some(&"asset"), Some(asset_str), None, None, None) => {
let asset_id = AssetId::from_str(asset_str)?;

2
start
View File

@ -158,7 +158,7 @@ do
ELECTRUM_TXS_LIMIT=500
ELECTRUM_MAX_LINE_SIZE=1048576 # 1 MiB
ELECTRUM_MAX_SUBSCRIPTIONS=100
ELECTRUM_MAX_CLIENTS=10
ELECTRUM_MAX_CLIENTS=1000
MAIN_LOOP_DELAY=500
DAEMON_CONF="${HOME}/${DAEMON}.conf"
HTTP_SOCKET_FILE="${HOME}/socket/esplora-${DAEMON}-${NETWORK}"