From 48eb0b554b698fe5af2d44f34e674f5d119d3b05 Mon Sep 17 00:00:00 2001 From: Edward Houston Date: Mon, 17 Aug 2026 12:50:19 +0200 Subject: [PATCH] Bound anonymous REST daemon-proxy requests The REST server dispatched every request through an async fn on the Tokio runtime, but almost every handler is synchronous. The broadcast and package submission handlers make a blocking JSON-RPC call to the daemon, serialized on a single process-wide Mutex with a 600 second read timeout. One concurrent request per CPU core against a slow or paused daemon therefore parked every worker thread, and all REST endpoints stopped being served, including fully in-memory ones such as GET /blocks/tip/height. Move the synchronous router onto the Tokio blocking pool, keeping only GET /block-template (the one genuinely async handler) on the runtime. Stop routing client-triggered daemon RPC through the shared singleton connection. sendrawtransaction and submitpackage now run on their own short-lived connection with a 30 second I/O timeout, capped at 8 concurrent calls, so one slow client request can no longer stall indexing or other clients, and failures are reported rather than retried forever. Report the resulting failures as gateway errors instead of the previous 400: 503 when the concurrency cap is reached or bitcoind is warming up, 504 when the daemon does not answer in time. Tunable via DAEMON_PROXY_MAX_CONCURRENCY, DAEMON_PROXY_RPC_TIMEOUT and DAEMON_PROXY_QUEUE_TIMEOUT. New daemon_rpc_proxied metric counts these calls by result. --- src/daemon.rs | 269 ++++++++++++++++++++++++++++++++++++++++- src/electrum/server.rs | 6 +- src/errors.rs | 16 +++ src/rest.rs | 153 +++++++++++++++++++++-- 4 files changed, 426 insertions(+), 18 deletions(-) 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();