diff --git a/src/daemon.rs b/src/daemon.rs index cefb4c673..af6e12f30 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -5,7 +5,7 @@ use std::io::{BufRead, BufReader, Lines, Write}; use std::net::{SocketAddr, TcpStream}; use std::path::PathBuf; use std::str::FromStr; -use std::sync::{Arc, Mutex}; +use std::sync::{Arc, Condvar, Mutex}; use std::time::{Duration, Instant}; use std::{env, fs, io}; @@ -45,6 +45,24 @@ lazy_static! { static ref DAEMON_CONN_RECYCLE_COOLDOWN: Duration = Duration::from_secs( env::var("DAEMON_CONN_RECYCLE_COOLDOWN").map_or(30, |s| s.parse().unwrap()) ); + // Maximum number of daemon RPCs made on behalf of API clients (transaction broadcast + // and package submission) that may be in flight at once. These endpoints are reachable + // anonymously, so this is what bounds how many threads a slow or wedged daemon can + // park. Kept well below bitcoind's own rpcthreads/rpcworkqueue so client traffic + // cannot starve the indexer of daemon capacity. + static ref DAEMON_PROXY_MAX_CONCURRENCY: usize = + env::var("DAEMON_PROXY_MAX_CONCURRENCY").map_or(8, |s| s.parse().unwrap()); + // Read/write timeout for client-proxied RPCs. Deliberately far shorter than + // DAEMON_READ_TIMEOUT: an API client must get an answer (or a 504) in bounded time, + // whereas the indexer can afford to wait out a long-running daemon call. + static ref DAEMON_PROXY_RPC_TIMEOUT: Duration = Duration::from_secs( + env::var("DAEMON_PROXY_RPC_TIMEOUT").map_or(30, |s| s.parse().unwrap()) + ); + // How long a client-proxied RPC waits for a free slot before giving up with + // `DaemonBusy`, rather than queueing behind an unbounded backlog. + static ref DAEMON_PROXY_QUEUE_TIMEOUT: Duration = Duration::from_secs( + env::var("DAEMON_PROXY_QUEUE_TIMEOUT").map_or(5, |s| s.parse().unwrap()) + ); } const MAX_ATTEMPTS: u32 = 5; @@ -533,6 +551,68 @@ impl Connection { } } +/// A counting semaphore for blocking (non-async) callers, used to cap how many daemon RPCs +/// may be in flight on behalf of API clients at any one time. +/// +/// Client-triggered RPCs are anonymous and unmetered, so without a cap a caller can open as +/// many concurrent daemon calls as it can open sockets. Each of those calls occupies a +/// thread for as long as the daemon takes to answer, which is what turns a slow daemon into +/// a full API outage. Bounding them means a slow daemon degrades the endpoints that need it +/// and leaves every other endpoint untouched. +struct BlockingSemaphore { + /// Number of permits still available. + available: Mutex, + released: Condvar, + capacity: usize, +} + +impl BlockingSemaphore { + fn new(capacity: usize) -> Self { + // A zero capacity would deadlock every caller, so treat it as "one at a time". + let capacity = capacity.max(1); + BlockingSemaphore { + available: Mutex::new(capacity), + released: Condvar::new(), + capacity, + } + } + + /// Take a permit, waiting at most `wait_timeout` for one to be released. Returns `None` + /// if none became available in time, so the caller can fail fast instead of queueing + /// behind an unbounded backlog of requests to an unresponsive daemon. + fn acquire(&self, wait_timeout: Duration) -> Option> { + let deadline = Instant::now() + wait_timeout; + let mut available = self.available.lock().unwrap(); + loop { + if *available > 0 { + *available -= 1; + return Some(SemaphorePermit { semaphore: self }); + } + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + return None; + } + available = self.released.wait_timeout(available, remaining).unwrap().0; + } + } + + fn capacity(&self) -> usize { + self.capacity + } +} + +/// Returns its permit to the [`BlockingSemaphore`] on drop. +struct SemaphorePermit<'a> { + semaphore: &'a BlockingSemaphore, +} + +impl Drop for SemaphorePermit<'_> { + fn drop(&mut self) { + *self.semaphore.available.lock().unwrap() += 1; + self.semaphore.released.notify_one(); + } +} + struct Counter { value: Mutex, } @@ -562,10 +642,15 @@ pub struct Daemon { rpc_threads: Arc, + // Caps concurrent RPCs issued on behalf of API clients (see `request_proxied`). + // Shared across reconnects so the cap stays global to the process. + proxy_limit: Arc, + // monitoring latency: HistogramVec, size: HistogramVec, conn_recycle: CounterVec, + proxy_rpc: CounterVec, } impl Daemon { @@ -604,6 +689,7 @@ impl Daemon { .build() .unwrap(), ), + proxy_limit: Arc::new(BlockingSemaphore::new(*DAEMON_PROXY_MAX_CONCURRENCY)), latency: metrics.histogram_vec( HistogramOpts::new("daemon_rpc", "Bitcoind RPC latency (in seconds)"), &["method"], @@ -612,6 +698,13 @@ impl Daemon { HistogramOpts::new("daemon_bytes", "Bitcoind RPC size (in bytes)"), &["method", "dir"], ), + proxy_rpc: metrics.counter_vec( + MetricOpts::new( + "daemon_rpc_proxied", + "Daemon RPCs made on behalf of API clients (by result)", + ), + &["result"], + ), conn_recycle: metrics.counter_vec( MetricOpts::new( "daemon_rpc_conn_recycled", @@ -662,9 +755,11 @@ impl Daemon { message_id: Counter::new(), signal: self.signal.clone(), rpc_threads: self.rpc_threads.clone(), + proxy_limit: Arc::clone(&self.proxy_limit), latency: self.latency.clone(), size: self.size.clone(), conn_recycle: self.conn_recycle.clone(), + proxy_rpc: self.proxy_rpc.clone(), }) } @@ -808,6 +903,60 @@ impl Daemon { parse_jsonrpc_reply(reply, method, id) } + /// Perform one RPC on behalf of an API client, bounded in both concurrency and time. + /// + /// The REST and Electrum endpoints that reach the daemon are anonymous and unmetered, + /// so they must not use `request`: that path serializes on the process-wide + /// `Mutex` (letting one slow client request stall the indexer and every + /// other client) and retries transport failures forever (letting one client request + /// occupy a thread indefinitely). Instead each call gets its own short-lived + /// connection with a short I/O timeout, and `proxy_limit` caps how many may be in + /// flight at once. Failures are reported to the client rather than retried. + #[trace(method = %method)] + fn request_proxied(&self, method: &str, params: Value) -> Result { + let _permit = self + .proxy_limit + .acquire(*DAEMON_PROXY_QUEUE_TIMEOUT) + .ok_or_else(|| { + self.proxy_rpc.with_label_values(&["busy"]).inc(); + Error::from(ErrorKind::DaemonBusy(format!( + "all {} client RPC slots are in use, gave up waiting after {:?}", + self.proxy_limit.capacity(), + *DAEMON_PROXY_QUEUE_TIMEOUT + ))) + })?; + + match self.request_once(method, params, *DAEMON_PROXY_RPC_TIMEOUT) { + Ok(result) => { + self.proxy_rpc.with_label_values(&["ok"]).inc(); + Ok(result) + } + Err(err) => { + // Report transport-level failures (which include hitting + // DAEMON_PROXY_RPC_TIMEOUT) distinctly from daemon-level rejections, so + // callers can answer "the daemon didn't respond" with a gateway error + // instead of blaming the client's request. + let unavailable = match err.kind() { + ErrorKind::Connection(msg) => Some(format!("{} failed: {}", method, msg)), + _ => None, + }; + match unavailable { + Some(msg) => { + // The concise message is what the client sees, so log the full + // chain (which carries the underlying io error) before dropping it. + warn!("client daemon RPC failed: {}", err.display_chain()); + self.proxy_rpc.with_label_values(&["unavailable"]).inc(); + Err(ErrorKind::DaemonUnavailable(msg).into()) + } + None => { + self.proxy_rpc.with_label_values(&["error"]).inc(); + Err(err) + } + } + } + } + } + #[trace] fn retry_reconnect(&self) -> Daemon { // XXX add a max reconnection attempts limit? @@ -1020,9 +1169,14 @@ impl Daemon { self.broadcast_raw(&serialize_hex(tx)) } + /// Broadcast a raw transaction on behalf of an API client. + /// + /// Uses the bounded client RPC path (`request_proxied`) rather than the shared + /// singleton connection: this is reachable anonymously over both the REST and Electrum + /// interfaces, so it must not be able to stall the indexer or other clients. #[trace] pub fn broadcast_raw(&self, txhex: &str) -> Result { - let txid = self.request("sendrawtransaction", json!([txhex]))?; + let txid = self.request_proxied("sendrawtransaction", json!([txhex]))?; Ok( Txid::from_str(txid.as_str().chain_err(|| "non-string txid")?) .chain_err(|| "failed to parse txid")?, @@ -1043,7 +1197,8 @@ impl Daemon { (None, Some(burn)) => json!([txhex, null, format!("{:.8}", burn)]), (None, None) => json!([txhex]), }; - let result = self.request("submitpackage", params)?; + // Anonymously reachable, so bounded like broadcast_raw() above. + let result = self.request_proxied("submitpackage", params)?; serde_json::from_value::(result) .chain_err(|| "invalid submitpackage reply") } @@ -1165,13 +1320,16 @@ impl Daemon { #[cfg(test)] mod tests { - use super::{parse_jsonrpc_reply, recycle_due, ConnectionConfig, CookieGetter}; + use super::{ + parse_jsonrpc_reply, recycle_due, BlockingSemaphore, ConnectionConfig, CookieGetter, + }; use crate::errors::{Error, ErrorKind, Result}; use crate::signal::Waiter; use serde_json::json; use std::net::TcpListener; use std::sync::Arc; - use std::time::Duration; + use std::thread; + use std::time::{Duration, Instant}; const COOLDOWN: Duration = Duration::from_secs(30); const MAX_AGE: Option = Some(Duration::from_secs(60)); @@ -1232,6 +1390,107 @@ mod tests { assert!(recycle_due(secs(600), MAX_AGE, Some(secs(31)), COOLDOWN)); } + #[test] + fn one_shot_connection_recv_gives_up_at_the_endpoint_timeout() { + // A daemon that accepts the connection and then never answers is exactly the + // failure mode behind XF-05: without a short client-facing timeout the caller + // blocks for DAEMON_READ_TIMEOUT (10 minutes by default). Hold the accepted socket + // open for the duration so the read blocks rather than seeing EOF. + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let blackhole = thread::spawn(move || { + let (socket, _) = listener.accept().unwrap(); + thread::sleep(secs(5)); + drop(socket); + }); + + let config = ConnectionConfig { + addr, + fallback: None, + cookie_getter: Arc::new(StaticCookie), + signal: Waiter::start(crossbeam_channel::never()), + max_age: None, + }; + + let mut connection = config.connect_once(secs(1)).unwrap(); + connection + .send(&json!({"method": "getblockcount"}).to_string()) + .unwrap(); + + let started = Instant::now(); + let err = connection.recv().unwrap_err(); + let elapsed = started.elapsed(); + + assert!( + matches!(err.kind(), ErrorKind::Connection(_)), + "expected a connection error, got {:?}", + err + ); + assert!( + elapsed >= secs(1) && elapsed < secs(4), + "recv should give up at the endpoint timeout, took {:?}", + elapsed + ); + + blackhole.join().unwrap(); + } + + #[test] + fn semaphore_hands_out_capacity_permits() { + let semaphore = BlockingSemaphore::new(2); + let first = semaphore.acquire(secs(0)); + let second = semaphore.acquire(secs(0)); + assert!(first.is_some()); + assert!(second.is_some()); + // At capacity: the third caller is refused rather than queued indefinitely. + assert!(semaphore.acquire(secs(0)).is_none()); + } + + #[test] + fn semaphore_permit_is_returned_on_drop() { + let semaphore = BlockingSemaphore::new(1); + { + let _permit = semaphore.acquire(secs(0)).unwrap(); + assert!(semaphore.acquire(secs(0)).is_none()); + } + assert!(semaphore.acquire(secs(0)).is_some()); + } + + #[test] + fn semaphore_zero_capacity_is_treated_as_one() { + // A zero cap would wedge every client request, so it is clamped rather than honored. + let semaphore = BlockingSemaphore::new(0); + assert_eq!(semaphore.capacity(), 1); + assert!(semaphore.acquire(secs(0)).is_some()); + } + + #[test] + fn semaphore_gives_up_after_wait_timeout() { + let semaphore = BlockingSemaphore::new(1); + let _permit = semaphore.acquire(secs(0)).unwrap(); + + // This is what stops client requests from piling up behind a wedged daemon: the + // caller waits a bounded time and is then refused. + let started = Instant::now(); + assert!(semaphore.acquire(Duration::from_millis(150)).is_none()); + assert!(started.elapsed() >= Duration::from_millis(150)); + } + + #[test] + fn semaphore_wakes_a_waiter_when_a_permit_is_released() { + let semaphore = Arc::new(BlockingSemaphore::new(1)); + let permit = semaphore.acquire(secs(0)).unwrap(); + + let waiter = { + let semaphore = Arc::clone(&semaphore); + thread::spawn(move || semaphore.acquire(secs(5)).is_some()) + }; + + thread::sleep(Duration::from_millis(50)); + drop(permit); + assert!(waiter.join().unwrap()); + } + #[test] fn warmup_error_parses_as_rpc_error() { let reply = json!({ diff --git a/src/electrum/server.rs b/src/electrum/server.rs index fe537f813..5451a51f9 100644 --- a/src/electrum/server.rs +++ b/src/electrum/server.rs @@ -107,7 +107,11 @@ fn jsonrpc_code(e: &Error) -> JsonRpcV2Error { match e.kind() { ErrorKind::InvalidParams(_) => JsonRpcV2Error::InvalidParams, ErrorKind::TooPopular | ErrorKind::TooManyUtxos => JsonRpcV2Error::BadRequest, - ErrorKind::RpcError(..) => JsonRpcV2Error::DaemonError, + // The daemon could not be reached (or we refused to queue for it) for a request + // made on the client's behalf. This is a downstream failure, not a client error. + ErrorKind::RpcError(..) | ErrorKind::DaemonBusy(_) | ErrorKind::DaemonUnavailable(_) => { + JsonRpcV2Error::DaemonError + } _ => JsonRpcV2Error::InternalError, } } diff --git a/src/errors.rs b/src/errors.rs index fadc6f9ca..c80e85bba 100644 --- a/src/errors.rs +++ b/src/errors.rs @@ -34,6 +34,22 @@ error_chain! { display("{}", msg) } + // Raised when a request made on behalf of an API client cannot get one of the + // bounded client RPC slots within its wait budget. The daemon itself may be + // perfectly healthy - we are simply refusing to queue any deeper. + DaemonBusy(msg: String) { + description("Daemon RPC concurrency limit reached") + display("Daemon is busy: {}", msg) + } + + // Raised when a request made on behalf of an API client fails at the transport + // level (connect failure, or a read that exceeded the client-facing timeout). + // Unlike internal callers, client requests are not retried indefinitely. + DaemonUnavailable(msg: String) { + description("Daemon RPC unavailable") + display("Daemon is unavailable: {}", msg) + } + #[cfg(feature = "electrum-discovery")] ElectrumClient(e: electrum_client::Error) { description("Electrum client error") diff --git a/src/rest.rs b/src/rest.rs index 888eb9bf3..5493a0fc0 100644 --- a/src/rest.rs +++ b/src/rest.rs @@ -533,7 +533,14 @@ fn spawn_conn( let resp_result = match body_result { Ok(Ok(collected)) => { - handle_request(method, uri, collected.to_bytes(), &query, &config).await + handle_request( + method, + uri, + collected.to_bytes(), + query, + Arc::clone(&config), + ) + .await } // Inner Err by http_body_util::Limited, either a LengthLimitError or an error by the underlying body stream Ok(Err(e)) if e.is::() => Err(HttpError( @@ -671,8 +678,67 @@ impl Handle { } } +/// Whether `uri` addresses the block template endpoint, the one route handled on the async +/// runtime rather than on the blocking pool (see `handle_request`). Matched exactly the way +/// the router below matches it, so the two cannot drift apart. +fn is_block_template_request(method: &Method, uri: &hyper::Uri) -> bool { + let mut path = uri.path().split('/').skip(1); + *method == Method::GET && path.next() == Some("block-template") && path.next().is_none() +} + +/// Dispatch a request, keeping blocking work off the async worker threads. +/// +/// Almost every handler is synchronous: it reads RocksDB, and some of them (transaction +/// broadcast, package submission, and any lookup in `--lightmode`) make a blocking JSON-RPC +/// call to the daemon. Running those directly on a Tokio worker lets a slow or unresponsive +/// daemon park every worker the runtime has, at which point even fully in-memory endpoints +/// such as `GET /blocks/tip/height` stop being served. Moving them to the blocking pool +/// keeps the runtime free to answer everything else. +/// +/// The block template endpoint is the exception: it is genuinely asynchronous (concurrent +/// callers share one in-flight daemon fetch) and already does its own blocking work on the +/// blocking pool, so it stays on the runtime. #[trace] async fn handle_request( + method: Method, + uri: hyper::Uri, + body: Bytes, + query: Arc, + config: Arc, +) -> Result>, HttpError> { + if is_block_template_request(&method, &uri) { + return handle_block_template_request(&query, &config).await; + } + + let path = uri.path().to_string(); + tokio::task::spawn_blocking(move || handle_blocking_request(method, uri, body, &query, &config)) + .await + .unwrap_or_else(|err| { + // The handler panicked or was cancelled; hyper still needs a response. + warn!("request handler for path='{}' failed: {}", path, err); + Err(HttpError( + StatusCode::INTERNAL_SERVER_ERROR, + "Internal server error".to_string(), + )) + }) +} + +async fn handle_block_template_request( + query: &Query, + config: &Config, +) -> Result>, HttpError> { + if !config.enable_mining_rest { + return Err(HttpError::forbidden( + "mining REST endpoints are disabled".to_string(), + )); + } + getblocktemplate_response(query.getblocktemplate().await) +} + +/// The synchronous body of the router. Always invoked from the blocking pool by +/// `handle_request`, never directly from an async worker thread. +#[trace] +fn handle_blocking_request( method: Method, uri: hyper::Uri, body: Bytes, @@ -1150,15 +1216,8 @@ async fn handle_request( json_response(query.estimate_fee_map(), TTL_SHORT) } - (&Method::GET, Some(&"block-template"), None, None, None, None) => { - if !config.enable_mining_rest { - return Err(HttpError::forbidden( - "mining REST endpoints are disabled".to_string(), - )); - } - getblocktemplate_response(query.getblocktemplate().await) - } - + // NOTE: `GET /block-template` is intercepted by `handle_request` before reaching + // here, because it is the only asynchronous handler. See `is_block_template_request`. #[cfg(feature = "liquid")] (&Method::GET, Some(&"assets"), Some(&"registry"), None, None, None) => { let start_index: usize = query_params @@ -1502,6 +1561,26 @@ impl From for HttpError { impl From for HttpError { fn from(e: errors::Error) -> Self { warn!("errors::Error: {:?}", e); + // Downstream daemon failures are ours, not the client's: answering them with the + // default 400 would tell callers (and caches, and load balancers) that a perfectly + // valid request was malformed. + match e.kind() { + // We refused to queue any deeper for the daemon. Retrying later may well work. + errors::ErrorKind::DaemonBusy(_) => { + return HttpError(StatusCode::SERVICE_UNAVAILABLE, e.to_string()) + } + // The daemon could not be reached, or did not answer within the client-facing + // timeout (see DAEMON_PROXY_RPC_TIMEOUT). + errors::ErrorKind::DaemonUnavailable(_) => { + return HttpError(StatusCode::GATEWAY_TIMEOUT, e.to_string()) + } + // -28 is bitcoind's "still warming up". Client-proxied requests are no longer + // retried until it finishes, so report it as a transient server-side condition. + errors::ErrorKind::RpcError(-28, _, _) => { + return HttpError(StatusCode::SERVICE_UNAVAILABLE, e.to_string()) + } + _ => (), + } match e.description().to_string().as_ref() { "getblock RPC error: {\"code\":-5,\"message\":\"Block not found\"}" => { HttpError::not_found("Block not found".to_string()) @@ -1542,12 +1621,62 @@ impl From for HttpError { #[cfg(test)] mod tests { - use crate::{errors, rest::HttpError}; + use crate::rest::{is_block_template_request, HttpError}; + use crate::{errors, errors::ErrorKind}; use http_body_util::BodyExt; - use hyper::StatusCode; + use hyper::{Method, StatusCode}; use serde_json::Value; use std::collections::HashMap; + #[test] + fn block_template_is_the_only_route_kept_on_the_async_runtime() { + let is_async = |method: Method, uri: &str| { + is_block_template_request(&method, &uri.parse::().unwrap()) + }; + + assert!(is_async(Method::GET, "/block-template")); + assert!(is_async(Method::GET, "/block-template?ignored=1")); + + // Everything else must fall through to the blocking pool, including near-misses + // that the router itself would not match as the block template route. + assert!(!is_async(Method::GET, "/block-template/")); + assert!(!is_async(Method::GET, "/block-template/extra")); + assert!(!is_async(Method::POST, "/block-template")); + assert!(!is_async(Method::GET, "/blocks/tip/height")); + assert!(!is_async(Method::POST, "/tx")); + } + + #[test] + fn daemon_failures_map_to_gateway_statuses() { + // A client-proxied daemon failure is a downstream problem, so it must not be + // reported as a 400 (which would blame - and let caches memoize - a valid request). + let busy = HttpError::from(errors::Error::from(ErrorKind::DaemonBusy( + "all 8 client RPC slots are in use".to_string(), + ))); + assert_eq!(busy.0, StatusCode::SERVICE_UNAVAILABLE); + + let unavailable = HttpError::from(errors::Error::from(ErrorKind::DaemonUnavailable( + "sendrawtransaction failed".to_string(), + ))); + assert_eq!(unavailable.0, StatusCode::GATEWAY_TIMEOUT); + + // bitcoind still warming up: transient, and no longer retried for client requests. + let warming_up = HttpError::from(errors::Error::from(ErrorKind::RpcError( + -28, + "Loading block index...".to_string(), + "sendrawtransaction".to_string(), + ))); + assert_eq!(warming_up.0, StatusCode::SERVICE_UNAVAILABLE); + + // Unrelated daemon errors keep their existing 400 mapping. + let rejected = HttpError::from(errors::Error::from(ErrorKind::RpcError( + -26, + "min relay fee not met".to_string(), + "sendrawtransaction".to_string(), + ))); + assert_eq!(rejected.0, StatusCode::BAD_REQUEST); + } + #[test] fn test_parse_query_param() { let mut query_params = HashMap::new();