use crate::chaindef::{BlockHeader, ScriptHash, Transaction};
use crate::encode::outpoint_hash;
use crate::query::scripthash_subscriptions::ScripthashSubscriptions;
use crate::signal::{NetworkNotifier, Waiter};
use crate::util::Channel;
use anyhow::{Context, Result};
use bitcoincash::hash_types::Txid;
use futures_util::{stream, StreamExt};
use log::{info, trace, warn};
use std::collections::{HashMap, HashSet};
use std::net::SocketAddr;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::mpsc::error::TrySendError;
use tokio::sync::mpsc::Sender;
use tokio::sync::Mutex;
use crate::chaindef::BlockHash;
use crate::config::Config;
use crate::def::PROTOCOL_VERSION_MAX;
use crate::doslimit::{ConnectionLimits, GlobalLimits};
use crate::metrics::Metrics;
use crate::query::Query;
use crate::metrics;
use crate::rpc::rpcstats::RpcStats;
use super::restconnection::{start_http_server, start_https_server};
use super::tcpcommunication::start_tcp_server;
use super::tcpsslcommunication::start_tcp_ssl_server;
use super::wscommunication::start_ws_server;
use super::wsscommunication::start_wss_server;
use super::{Message, Notification};
#[cfg(bch)]
fn get_output_scripthash(txn: &Transaction, n: Option<usize>) -> Vec<ScriptHash> {
if let Some(out) = n {
vec![ScriptHash::from_script(&txn.output[out].script_pubkey)]
} else {
txn.output
.iter()
.map(|o| ScriptHash::from_script(&o.script_pubkey))
.collect()
}
}
#[cfg(nexa)]
fn get_output_scripthash(txn: &Transaction, n: Option<usize>) -> Vec<ScriptHash> {
if let Some(out) = n {
vec![match txn.output[out].scriptpubkey_without_token() {
Ok(Some(s)) => ScriptHash::from_script(&s),
_ => ScriptHash::from_script(&txn.output[out].script_pubkey),
}]
} else {
txn.output
.iter()
.map(|o| match o.scriptpubkey_without_token() {
Ok(Some(s)) => ScriptHash::from_script(&s),
_ => ScriptHash::from_script(&o.script_pubkey),
})
.collect()
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct ConnectionId(pub(crate) u64);
impl ConnectionId {
pub(crate) fn new() -> Self {
static COUNTER: AtomicU64 = AtomicU64::new(0);
ConnectionId(COUNTER.fetch_add(1, Ordering::Relaxed))
}
}
pub struct Server {
notification: tokio::sync::mpsc::Sender<Notification>,
tcp_server: Option<tokio::task::JoinHandle<()>>,
tcp_ssl_server: Option<tokio::task::JoinHandle<()>>,
wss_server: Option<tokio::task::JoinHandle<()>>,
ws_server: Option<tokio::task::JoinHandle<()>>,
rest_server: Option<tokio::task::JoinHandle<()>>,
https_server: Option<tokio::task::JoinHandle<()>>,
query: Arc<Query>,
subscription_manager: Arc<ScripthashSubscriptions>,
}
impl Server {
async fn start_notifier(
notification: Channel<Notification>,
senders: Arc<Mutex<HashMap<ConnectionId, Sender<Message>>>>,
_subscription_manager: Arc<ScripthashSubscriptions>,
acceptors: Vec<Sender<Option<(TcpStream, SocketAddr)>>>,
) {
crate::thread::spawn_task("notification", async move {
while let Some(msg) = notification.receiver().await.recv().await {
let mut senders = senders.lock().await;
match msg {
Notification::ScriptHashChange(connection_updates) => {
for (conn_id, updates) in connection_updates {
if let Some(sender) = senders.get(&conn_id) {
match sender.try_send(Message::ScriptHashChange(updates)) {
Ok(_) => {}
Err(err) => match err {
TrySendError::Closed(_) => {
trace!("scripthash notify failed: peer disconnected");
}
TrySendError::Full(_) => {
trace!("scripthash notify failed: peer channel full");
}
},
}
}
}
}
Notification::ChainTipChange(hash) => {
senders.retain(|_, sender| {
match sender.try_send(Message::ChainTipChange(hash.clone())) {
Ok(_) => true,
Err(err) => match err {
TrySendError::Closed(_) => {
trace!("chaintip notify failed: peer disconnected");
false
}
TrySendError::Full(_) => {
trace!("chaintip notify failed: peer channel full");
true }
},
}
});
}
Notification::Exit => {
for a in &acceptors {
trace!("peer exit notification");
let _ = a.send(None).await;
}
}
}
}
Ok(())
});
}
async fn start_acceptor(
addr: SocketAddr,
acceptor_type: String,
) -> Channel<Option<(TcpStream, SocketAddr)>> {
let chan = Channel::bounded(100_000);
let acceptor = chan.sender();
crate::thread::spawn_task("acceptor", async move {
let listener = match TcpListener::bind(addr).await {
Ok(l) => l,
Err(e) => {
let errmsg = format!(
"Failed to start {} server - bind({}) failed: {}",
acceptor_type, addr, e
);
Waiter::shutdown(&errmsg);
bail!(errmsg)
}
};
info!(
"Electrum {} RPC server running on {} (protocol {})",
acceptor_type, addr, PROTOCOL_VERSION_MAX
);
loop {
let (stream, addr) = match listener.accept().await {
Ok(conn) => conn,
Err(e) => {
warn!("{} accept failed: {} - will retry", acceptor_type, e);
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
continue;
}
};
match acceptor.send(Some((stream, addr))).await {
Ok(_) => {}
Err(e) => trace!("Failed to send to client {:?}", e),
}
if Waiter::shutdown_check().is_err() {
break;
}
}
Ok(())
});
chan
}
pub async fn start(
config: Arc<Config>,
query: Arc<Query>,
metrics: Arc<Metrics>,
connection_limits: ConnectionLimits,
global_limits: Arc<GlobalLimits>,
network_notifier: Arc<NetworkNotifier>,
) -> Self {
let stats = Arc::new(RpcStats {
latency: metrics.histogram_vec(
"rostrum_rpc_latency",
"RPC latency (seconds)",
&["method"],
metrics::default_duration_buckets(),
),
subscriptions: metrics.gauge_int(prometheus::Opts::new(
"rostrum_scripthash_subscriptions",
"# of scripthash subscriptions for node",
)),
});
stats.subscriptions.set(0);
let blockchain_event_queue = Channel::bounded(1_000_000);
let blockchain_event_sender = blockchain_event_queue.sender();
let tcp_acceptor = if !config.tcp {
info!("TCP endpoint is not enabled in the server configuration");
None
} else {
Some(Self::start_acceptor(config.electrum_rpc_addr, "tcp".to_string()).await)
};
let ws_acceptor = if !config.websocket {
info!("WebSocket endpoint is not enabled in server the configuration.");
None
} else {
Some(Self::start_acceptor(config.electrum_ws_addr, "websocket".to_string()).await)
};
let wss_acceptor = if !config.websocket_ssl {
info!("WebSocket over SSL endpoint is not enabled in server the configuration.");
None
} else {
Some(
Self::start_acceptor(config.electrum_ws_ssl_addr, "websocket_ssl".to_string())
.await,
)
};
let http_acceptor = if !config.http {
info!("HTTP endpoint is not enabled in server the configuration.");
None
} else {
Some(Self::start_acceptor(config.electrum_http_addr, "http".to_string()).await)
};
let https_acceptor = if !config.https {
info!("HTTPS endpoint is not enabled in server the configuration.");
None
} else {
Some(Self::start_acceptor(config.electrum_https_addr, "https".to_string()).await)
};
let ssl_acceptor = if !config.tcp_ssl {
info!("TCP SSL endpoint is not enabled in the server configuration");
None
} else {
Some(Self::start_acceptor(config.electrum_rpc_ssl_addr, "tcp_ssl".to_string()).await)
};
let rpc_queue_senders =
Arc::new(Mutex::new(HashMap::<ConnectionId, Sender<Message>>::new()));
let subscription_manager = Arc::new(ScripthashSubscriptions::new(
query.clone(),
config.statushash_max_elements,
));
{
let mut acceptors: Vec<Sender<Option<(TcpStream, SocketAddr)>>> = Vec::default();
if let Some(a) = tcp_acceptor.as_ref() {
acceptors.push(a.sender());
}
if let Some(a) = ws_acceptor.as_ref() {
acceptors.push(a.sender());
}
if let Some(a) = ssl_acceptor.as_ref() {
acceptors.push(a.sender());
}
Self::start_notifier(
blockchain_event_queue,
rpc_queue_senders.clone(),
subscription_manager.clone(),
acceptors,
)
.await;
}
let rest_server = http_acceptor.map(|acceptor| {
crate::thread::spawn_task(
"http_rpc",
start_http_server(
acceptor,
global_limits.clone(),
config.clone(),
query.clone(),
stats.clone(),
connection_limits,
network_notifier.clone(),
config.rpc_idle_timeout,
subscription_manager.clone(),
),
)
});
let https_server = https_acceptor.map(|acceptor| {
let cert_file = config
.ssl_cert_file
.as_ref()
.context("SSL certificate file not specified")
.unwrap();
let key_file = config
.ssl_key_file
.as_ref()
.context("SSL key file not specified")
.unwrap();
crate::thread::spawn_task(
"https_rpc",
start_https_server(
acceptor,
cert_file.clone(),
key_file.clone(),
global_limits.clone(),
config.clone(),
query.clone(),
stats.clone(),
connection_limits,
network_notifier.clone(),
config.rpc_idle_timeout,
subscription_manager.clone(),
),
)
});
let ws_server = ws_acceptor.map(|acceptor| {
crate::thread::spawn_task(
"ws_rpc",
start_ws_server(
acceptor,
global_limits.clone(),
config.clone(),
query.clone(),
stats.clone(),
connection_limits,
network_notifier.clone(),
rpc_queue_senders.clone(),
subscription_manager.clone(),
),
)
});
let tcp_server = tcp_acceptor.map(|acceptor| {
crate::thread::spawn_task(
"tcp_rpc",
start_tcp_server(
acceptor,
global_limits.clone(),
config.clone(),
query.clone(),
stats.clone(),
connection_limits,
network_notifier.clone(),
rpc_queue_senders.clone(),
subscription_manager.clone(),
),
)
});
let tcp_ssl_server = ssl_acceptor.map(|acceptor| {
let cert_file = config
.ssl_cert_file
.as_ref()
.context("SSL certificate file not specified")
.unwrap();
let key_file = config
.ssl_key_file
.as_ref()
.context("SSL key file not specified")
.unwrap();
crate::thread::spawn_task(
"tcp_ssl_rpc",
start_tcp_ssl_server(
acceptor,
cert_file.clone(),
key_file.clone(),
global_limits.clone(),
config.clone(),
query.clone(),
stats.clone(),
connection_limits,
network_notifier.clone(),
rpc_queue_senders.clone(),
subscription_manager.clone(),
),
)
});
let wss_server = wss_acceptor.map(|acceptor| {
let cert_file = config
.ssl_cert_file
.as_ref()
.context("SSL certificate file not specified")
.unwrap();
let key_file = config
.ssl_key_file
.as_ref()
.context("SSL key file not specified")
.unwrap();
crate::thread::spawn_task(
"wss_rpc",
start_wss_server(
acceptor,
cert_file.clone(),
key_file.clone(),
global_limits.clone(),
config.clone(),
query.clone(),
stats.clone(),
connection_limits,
network_notifier.clone(),
rpc_queue_senders.clone(),
subscription_manager.clone(),
),
)
});
Self {
notification: blockchain_event_sender,
query,
tcp_server,
tcp_ssl_server,
wss_server,
ws_server,
rest_server,
https_server,
subscription_manager,
}
}
async fn get_scripthashes_affected_by_tx(
&self,
txid: &Txid,
blockhash: Option<&BlockHash>,
) -> Result<Vec<ScriptHash>> {
let txn = match self.query.tx().get(txid, blockhash, None, false).await {
Ok(txn) => txn,
Err(e) => {
if e.to_string().contains("genesis") {
return Ok(Vec::default());
}
return Err(e);
}
};
let mut scripthashes = get_output_scripthash(&txn, None);
for txin in txn.input {
if txin.previous_output.is_null() {
continue;
}
let tx = self
.query
.get_tx_funding_prevout(&outpoint_hash(&txin.previous_output))
.await;
if let Ok(Some((tx, index, _))) = tx {
scripthashes.extend(get_output_scripthash(&tx, Some(index as usize)));
}
}
Ok(scripthashes)
}
async fn collect_affected_scripthashes(
&self,
headers_changed: &[BlockHeader],
txs_changed: HashSet<Txid>,
) -> HashSet<ScriptHash> {
let mut txn_done: HashSet<Txid> = HashSet::new();
let mut scripthashes: HashSet<ScriptHash> = HashSet::new();
async fn insert_for_txs(
parent: &Server,
txids: &mut dyn Iterator<Item = Txid>,
blockhash: Option<BlockHash>,
txn_done: &HashSet<Txid>,
) -> (HashSet<Txid>, Vec<ScriptHash>) {
let new: HashSet<_> = txids.collect();
let new: HashSet<_> = new.difference(txn_done).cloned().collect();
let hashes = stream::iter(&new)
.map(|tx| async move {
let effected = parent
.get_scripthashes_affected_by_tx(tx, blockhash.as_ref())
.await;
let hashes = match effected {
Ok(hashes) => hashes,
Err(e) => {
trace!("failed to get effected scripthashes for tx {}: {:?}", tx, e);
vec![]
}
};
stream::iter(hashes.into_iter())
})
.buffer_unordered(10)
.flatten()
.collect::<Vec<ScriptHash>>()
.await;
(new, hashes)
}
for header in headers_changed.iter() {
let blockhash = header.block_hash();
let mut txids = match self.query.getblocktxids(&blockhash).await {
Ok(txids) => txids.into_iter(),
Err(e) => {
warn!("Failed to get blocktxids for {}: {}", blockhash, e);
continue;
}
};
let (new_txs, new_scripthashes) =
insert_for_txs(self, &mut txids, Some(blockhash), &txn_done).await;
txn_done.extend(new_txs.iter());
scripthashes.extend(new_scripthashes.iter());
}
let mut txs_changed_iter = txs_changed.into_iter();
let (new_txs, new_scripthashes) =
insert_for_txs(self, &mut txs_changed_iter, None, &txn_done).await;
txn_done.extend(new_txs.iter());
scripthashes.extend(new_scripthashes.iter());
scripthashes
}
pub async fn notify_scripthash_subscriptions(
&self,
headers_changed: &[BlockHeader],
txs_changed: HashSet<Txid>,
) {
let scripthashes = self
.collect_affected_scripthashes(headers_changed, txs_changed)
.await;
if scripthashes.is_empty() {
return;
}
let connection_updates = match self
.subscription_manager
.notify_scripthashes_changed(scripthashes)
.await
{
Ok(updates) => updates,
Err(e) => {
warn!("Failed to compute subscription updates: {}", e);
return;
}
};
if connection_updates.is_empty() {
return;
}
if let Err(e) = self
.notification
.send(Notification::ScriptHashChange(connection_updates))
.await
{
trace!("Scripthash change notification failed: {}", e);
}
}
pub async fn notify_subscriptions_chaintip(&self, header: BlockHeader) {
if let Err(e) = self
.notification
.send(Notification::ChainTipChange(Box::new(header)))
.await
{
trace!("Failed to notify about chaintip change {}", e);
}
}
pub async fn disconnect_clients(&self) {
trace!("disconncting clients");
let _ = self.notification.send(Notification::Exit).await;
}
}
impl Drop for Server {
fn drop(&mut self) {
trace!("stop accepting new RPCs");
let n = self.notification.clone();
crate::thread::spawn_task("drop_rpc", async move {
let _ = n.send(Notification::Exit).await;
Ok(())
});
if let Some(handle) = self.tcp_server.take() {
info!("sending abort to tcp");
handle.abort();
info!("abort sent");
}
if let Some(handle) = self.ws_server.take() {
handle.abort();
}
if let Some(handle) = self.rest_server.take() {
handle.abort();
}
if let Some(handle) = self.https_server.take() {
handle.abort();
}
if let Some(handle) = self.tcp_ssl_server.take() {
handle.abort();
}
if let Some(handle) = self.wss_server.take() {
handle.abort();
}
trace!("RPC server is stopped");
}
}