esplora: one request limit for the whole listener, and --esplora-max-inflight

changelog.d/esplora-global-conn-limit.md

@@ -0,0 +1,18 @@
+Fixed
+
+- **The Esplora request limit bounds the listener, not each route.**
+  axum's `Router::layer` wraps every route on its own, so
+  `ConcurrencyLimitLayer` gave each of the listener's routes its own 256
+  permits: in practice no limit. The limit is now one semaphore shared by
+  every route and the fallback. Requests over it wait for a slot; they are
+  not refused. The request timeout now covers that wait too, so a queue of
+  slow requests ahead cannot hold a later one past one timeout (**408**).
+  The `/blindbit/v1/` routes share this limit too. A BlindBit request waits
+  for its own permit first, so one queued there holds no listener slot.
+
+Added
+
+- **`--esplora-max-inflight N`** (conf `esplora_max_inflight=`, NixOS
+  `services.rbitcoin.esplora.maxInflight`) sets that limit. The default stays
+  256. A busy public listener may want more. Values above tokio's
+  semaphore limit are refused at startup.

crates/rbitcoin-electrum/src/lib.rs

@@ -24,7 +24,7 @@ pub mod unspent;
 pub use server::{
     batch_window, electrum_scripthash_hex, parse_electrum_request_line, perf_totals, run_electrum,
     sample_reset_perf, ElectrumConfig, ElectrumHandle, ServeLimits, TipNotify,
-    DEFAULT_MAX_SCRIPTHASH_SUBS, MIN_BLOCKING_POOL,
+    DEFAULT_MAX_CONNECTIONS, DEFAULT_MAX_SCRIPTHASH_SUBS, MIN_BLOCKING_POOL,
 };
 pub use tweaks::DEFAULT_TWEAKS_MIN_DUST;
 pub use unspent::{

crates/rbitcoin-esplora/src/server.rs

@@ -29,7 +29,7 @@ use std::time::{Duration, Instant};
 use tokio::net::TcpListener;
 use tokio::sync::Semaphore;
 use tokio::task::JoinHandle;
-use tower::limit::ConcurrencyLimitLayer;
+use tower::limit::GlobalConcurrencyLimitLayer;
 use tower_http::limit::RequestBodyLimitLayer;
 use tower_http::timeout::TimeoutLayer;
 
@@ -936,35 +936,34 @@ pub async fn run_esplora(
     } else {
         rest
     };
-    // Outer → inner: concurrency → body → timeout → meter → chain-view stamp.
+    // Outer → inner: timeout → concurrency → body → meter → chain-view stamp.
+    // The timeout goes on last, outside the limit, so the wait for a slot
+    // counts against it.
+    let conn_limit = GlobalConcurrencyLimitLayer::new(max_conn);
     let listener_layers = |r: Router<AppState>| {
         r.layer(middleware::from_fn_with_state(
             state.clone(),
             stamp_chain_view_mw,
         ))
         .layer(middleware::from_fn(meter_rest))
-        .layer(TimeoutLayer::with_status_code(
-            StatusCode::REQUEST_TIMEOUT,
-            timeout,
-        ))
         .layer(RequestBodyLimitLayer::new(max_body))
-        .layer(ConcurrencyLimitLayer::new(max_conn))
+        // `Router::layer` wraps each route on its own; one global limit, cloned,
+        // shares its semaphore, so `max_conn` bounds the listener.
+        .layer(conn_limit.clone())
     };
-    let rest = listener_layers(rest.fallback(fallback_404));
-    // BlindBit takes its own permit first, under its own timeout, so a scan
-    // queued on it holds no listener slot. Merged after the listener layers,
-    // so they wrap these routes once.
+    let request_timeout = || TimeoutLayer::with_status_code(StatusCode::REQUEST_TIMEOUT, timeout);
+    let rest = listener_layers(rest.fallback(fallback_404)).layer(request_timeout());
+    // BlindBit takes its own permit before a listener slot, so a scan queued
+    // on it holds no slot. One timeout covers both waits. Merged after the
+    // listener layers, so they wrap these routes once.
     let rest = if state.query.sptweaks_enabled() {
         let sem = Arc::new(Semaphore::new(config.blindbit_max_inflight.max(1)));
         let blindbit = listener_layers(crate::blindbit::blindbit_routes())
             .layer(middleware::from_fn_with_state(
                 sem,
                 crate::blindbit::blindbit_permit_mw,
             ))
-            .layer(TimeoutLayer::with_status_code(
-                StatusCode::REQUEST_TIMEOUT,
-                timeout,
-            ));
+            .layer(request_timeout());
         rest.merge(blindbit)
     } else {
         rest

crates/rbitcoin-esplora/tests/blindbit_journey.rs

@@ -456,16 +456,35 @@ async fn blindbit_v1_routes_share_one_inflight_limit() {
         }
     }
     assert_eq!(queued.len(), 3, "a full BlindBit limit makes the next asks wait");
-    // While the listener limit is per route, this route's slots are its own
-    // and this passes either way. It checks the permit order once the
-    // listener limit spans every route.
+    // The listener's last slot is free only if the queued asks hold none.
     let (st, _) = tokio::time::timeout(
         Duration::from_millis(1000),
         http_get(addr, "/blocks/tip/height"),
     )
     .await
     .expect("asks queued on BlindBit hold no listener slot");
     assert_eq!(st, 200);
+    // A stalled POST /tx takes that last slot. The held BlindBit bodies hold
+    // the rest, so a GET waits: the two routers share one limit.
+    let mut spare = TcpStream::connect(addr).await.unwrap();
+    let req = "POST /tx HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Length: 64\r\n\r\n";
+    spare.write_all(req.as_bytes()).await.unwrap();
+    let mut waiting = None;
+    for _ in 0..20 {
+        let get = tokio::spawn(async move { http_get(addr, "/blocks/tip/height").await });
+        tokio::time::sleep(Duration::from_millis(300)).await;
+        if !get.is_finished() {
+            waiting = Some(get);
+            break;
+        }
+    }
+    let waiting = waiting.expect("BlindBit and the other routes share the listener limit");
+    drop(spare);
+    let (st, _) = tokio::time::timeout(Duration::from_secs(10), waiting)
+        .await
+        .expect("the freed slot serves the waiting GET")
+        .unwrap();
+    assert_eq!(st, 200);
     drop(held.pop());
     let (st, body) = tokio::time::timeout(Duration::from_secs(10), queued.remove(0))
         .await

crates/rbitcoin-esplora/tests/esplora_http_journey.rs

@@ -809,3 +809,96 @@ async fn esplora_tcp_wallet_mempool_and_replace() {
     handle.shutdown().await;
     let _ = std::fs::remove_dir_all(&dir);
 }
+
+/// `max_connections` bounds the listener, not each route. Requests held on
+/// one route fill it; a request on another route waits, and runs once a
+/// slot frees.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn esplora_max_connections_spans_every_route() {
+    const MAX: usize = 4;
+    let (_dir, q) = temp_query("esplora-max-conn");
+    let (h0, cb0) = coinbase(0, Fk::NULL, None);
+    q.connect_block(Height(0), &h0, &[cb0]).unwrap();
+    let mut cfg = EsploraConfig::with_network("127.0.0.1:0".parse().unwrap(), Network::Regtest);
+    cfg.limits.max_connections = MAX;
+    let handle = run_esplora(cfg, Arc::new(q), None)
+        .await
+        .expect("listen");
+    let addr = handle.local_addr;
+
+    // A POST /tx whose body never arrives holds its slot in the extractor.
+    let mut held = Vec::new();
+    for _ in 0..MAX {
+        let mut stream = TcpStream::connect(addr).await.unwrap();
+        let req = "POST /tx HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Length: 64\r\n\r\n";
+        stream.write_all(req.as_bytes()).await.unwrap();
+        held.push(stream);
+    }
+    // One GET on another route must stay unanswered while the slots are
+    // full. Retry while the held requests may not all be admitted yet.
+    let mut tip = None;
+    for _ in 0..20 {
+        let get = tokio::spawn(async move { http_get(addr, "/blocks/tip/height").await });
+        tokio::time::sleep(Duration::from_millis(500)).await;
+        if !get.is_finished() {
+            tip = Some(get);
+            break;
+        }
+    }
+    let tip = tip.expect("a full listener makes a request on another route wait");
+    // Freeing one slot serves that same request.
+    drop(held.pop());
+    let (st, body) = tokio::time::timeout(Duration::from_secs(10), tip)
+        .await
+        .expect("a freed slot serves the waiting request")
+        .unwrap();
+    assert_eq!((st, body.as_str()), (200, "0"));
+    handle.shutdown().await;
+}
+
+/// The request timeout covers the wait for a slot as well as the work, so a
+/// queue of slow requests ahead cannot hold a later one past one timeout.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn esplora_request_timeout_covers_the_wait_for_a_slot() {
+    let (_dir, q) = temp_query("esplora-queue-timeout");
+    let (h0, cb0) = coinbase(0, Fk::NULL, None);
+    q.connect_block(Height(0), &h0, &[cb0]).unwrap();
+    let mut cfg = EsploraConfig::with_network("127.0.0.1:0".parse().unwrap(), Network::Regtest);
+    cfg.limits.max_connections = 1;
+    cfg.limits.idle_timeout = Duration::from_secs(1);
+    let handle = run_esplora(cfg, Arc::new(q), None)
+        .await
+        .expect("listen");
+    let addr = handle.local_addr;
+
+    // Five POST /tx whose bodies never arrive queue for the one slot; each
+    // holds it until the timeout once admitted.
+    let mut held = Vec::new();
+    for _ in 0..5 {
+        let mut stream = TcpStream::connect(addr).await.unwrap();
+        let req = "POST /tx HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Length: 64\r\n\r\n";
+        stream.write_all(req.as_bytes()).await.unwrap();
+        held.push(stream);
+        tokio::time::sleep(Duration::from_millis(20)).await;
+    }
+    let sent = Instant::now();
+    let (st, _) = tokio::time::timeout(
+        Duration::from_secs(10),
+        http_get(addr, "/blocks/tip/height"),
+    )
+    .await
+    .expect("no answer within 10s");
+    let took = sent.elapsed();
+    // The held requests' deadlines started before this one's, so their 408s
+    // can free the slot just in time: either answer is within one timeout.
+    assert!(
+        st == 200 || st == 408,
+        "the wait ends in an answer or the request timeout, got {st}"
+    );
+    assert!(
+        took < Duration::from_millis(2500),
+        "a request waited {took:?} behind the queue, past one 1s timeout"
+    );
+    drop(held);
+    handle.shutdown().await;
+}

crates/rbitcoin-node/src/cli.rs

@@ -324,7 +324,7 @@ pub fn operator_usage() -> String {
     [--i2p-sam [HOST:PORT]] [--i2p-accept-incoming] \\\n\
     [--electrum-listen ADDR] [--esplora-listen ADDR] [--esplora-onion[=0|1]] [--health-listen [ADDR]] [--metrics] \\\n\
     [--sv2-tp-listen ADDR] [--sv2-tp-authority-sec KEY] [--sv2-tp-authority-sec-file PATH] [--sv2-tp-cert-validity SECS] [--sv2-tp-stale-grace SECS] [--sv2-tp-fee-delta SATS] [--sv2-tp-template-interval SECS] \\\n\
-    [--sh-index] [--block-filter-index] [--prune-seqsigwit] [--prune-seqsigwit-ram-threshold-bytes N] [--datadir-hdd] [--sp-tweaks] [--sp-tweaks-dust SATS] [--max-sh-creates N] [--heavy-sh-creates N] [--electrum-max-subs N] [--esplora-block-template] \\\n\
+    [--sh-index] [--block-filter-index] [--prune-seqsigwit] [--prune-seqsigwit-ram-threshold-bytes N] [--datadir-hdd] [--sp-tweaks] [--sp-tweaks-dust SATS] [--max-sh-creates N] [--heavy-sh-creates N] [--electrum-max-subs N] [--esplora-max-inflight N] [--esplora-block-template] \\\n\
     [--rpc] [--rpc-listen [ADDR]] [--rest] [--rpc-socket PATH] [--rpc-token-file PATH] [--rpc-cookie-file PATH] [--rpc-cookie-perms owner|group|all] [--rpc-work-queue N] \\\n\
     [--milestone HEIGHT] \\\n\
     [--max-outbound N] [--max-inbound N] \\\n\
@@ -378,6 +378,7 @@ Block filters: --block-filter-index (default off) builds BIP158 basic filters. I
   --max-sh-creates N refuses an unpaged Electrum/Esplora join with more than N creates (default 10000; 0 = unlimited). A paged history request is still served.\n\
   --heavy-sh-creates N runs unpaged Electrum/Esplora joins with more than N creates one at a time across the node (default 10000; 0 = no limit). Lighter joins stay parallel.\n\
   --electrum-max-subs N caps blockchain.scripthash.subscribe per Electrum connection (default 10000; >= 1). Wallets subscribe every address up to their gap limit.\n\
+  --esplora-max-inflight N caps Esplora requests in flight across the whole listener (default 256; >= 1). Further requests wait for a slot; the wait counts against the request timeout (408).\n\
   --esplora-block-template enables GET /block-template (GBT template JSON; default off).\n\
   --esplora-onion (default on) ADD_ONION for --esplora-listen when --tor-control is set.\n\
 Silent payments: --sp-tweaks (default off) writes/serves the thin BIP-352 tweak index.\n\

crates/rbitcoin-node/src/config.rs

@@ -340,6 +340,9 @@ pub struct NodeConfig {
     /// Max `blockchain.scripthash.subscribe` entries per Electrum connection
     /// (>= 1). Default [`rbitcoin_electrum::DEFAULT_MAX_SCRIPTHASH_SUBS`].
     pub electrum_max_subs: usize,
+    /// Max Esplora requests in flight across the whole listener (>= 1).
+    /// Default [`rbitcoin_electrum::DEFAULT_MAX_CONNECTIONS`].
+    pub esplora_max_inflight: usize,
     /// Opt-in Esplora `GET /block-template` (GBT template JSON). Default off.
     pub esplora_block_template: bool,
     /// Prometheus `GET /metrics` on the health listener. Default off.
@@ -450,6 +453,7 @@ impl Default for NodeConfig {
             max_sh_creates: rbitcoin_query::DEFAULT_MAX_SH_CREATES,
             heavy_sh_creates: rbitcoin_query::DEFAULT_HEAVY_SH_CREATES,
             electrum_max_subs: rbitcoin_electrum::DEFAULT_MAX_SCRIPTHASH_SUBS,
+            esplora_max_inflight: rbitcoin_electrum::DEFAULT_MAX_CONNECTIONS,
             esplora_block_template: false,
             metrics: false,
             esplora_onion: true,
@@ -1328,6 +1332,19 @@ impl NodeConfig {
                 }
                 self.electrum_max_subs = n;
             }
+            "esplora_max_inflight" => {
+                let n: usize = val
+                    .parse()
+                    .map_err(|e| NodeError::Config(format!("conf esplora_max_inflight: {e}")))?;
+                // `Semaphore::new` panics above `MAX_PERMITS`.
+                if n == 0 || n > tokio::sync::Semaphore::MAX_PERMITS {
+                    return Err(NodeError::Config(format!(
+                        "conf esplora_max_inflight must be >= 1 and <= {}",
+                        tokio::sync::Semaphore::MAX_PERMITS
+                    )));
+                }
+                self.esplora_max_inflight = n;
+            }
             "esplora_block_template" => {
                 self.esplora_block_template = parse_conf_bool(val)
                     .map_err(|e| NodeError::Config(format!("conf esplora_block_template: {e}")))?;

crates/rbitcoin-node/src/run.rs

@@ -975,6 +975,7 @@ pub async fn run_p2p(config: NodeConfig) -> Result<(), NodeError> {
         config
             .esplora_block_template
             .then_some(config.mempool.blocksonly),
+        config.esplora_max_inflight,
         &shutdown,
         Arc::clone(&node.hub),
         &mempool,
@@ -1227,6 +1228,7 @@ pub async fn run_p2p(config: NodeConfig) -> Result<(), NodeError> {
                         config
                             .esplora_block_template
                             .then_some(config.mempool.blocksonly),
+                        config.esplora_max_inflight,
                         &shutdown,
                         Arc::clone(&node.hub),
                         &mempool,
@@ -2336,12 +2338,15 @@ pub async fn start_electrum_if_ready(
     }
 }
 
+#[allow(clippy::too_many_arguments)] // call-site args stay unbundled
 pub async fn start_esplora_if_ready(
     sh_tip_ready: bool,
     listen: Option<EsploraListen>,
     network: Network,
     // `Some(blocksonly)` turns on `/block-template`. `None` leaves it off.
     block_template: Option<bool>,
+    // Requests in flight across the listener (`--esplora-max-inflight`).
+    max_inflight: usize,
     shutdown: &Shutdown,
     hub: Arc<ChainHub>,
     mempool: &std::sync::Arc<MempoolHub>,
@@ -2355,6 +2360,7 @@ pub async fn start_esplora_if_ready(
     let q = hub.query.clone();
     let btc_net = rbitcoin_rpc::methods::mine::rpc_btc_network(network);
     let mut ecfg = EsploraConfig::with_listen(listen, btc_net);
+    ecfg.limits.max_connections = max_inflight;
     if let Some(blocksonly) = block_template {
         let q = Arc::clone(&hub.query);
         let mp = Arc::clone(mempool);

crates/rbitcoin-node/tests/cli.rs

@@ -504,6 +504,41 @@ fn operator_conf_and_argv() {
     );
 }
 
+#[test]
+fn esplora_max_inflight_cli_conf_and_bounds() {
+    assert_eq!(
+        ready_config(["rbitcoin-node"]).esplora_max_inflight,
+        rbitcoin_electrum::DEFAULT_MAX_CONNECTIONS
+    );
+    assert_eq!(
+        ready_config(["rbitcoin-node", "--esplora-max-inflight", "1024"]).esplora_max_inflight,
+        1024
+    );
+    assert_eq!(
+        ready_config(["rbitcoin-node", "--esplora-max-inflight=8"]).esplora_max_inflight,
+        8
+    );
+    let mut c = NodeConfig::default();
+    let zero = c.apply_kv("esplora_max_inflight", "0").unwrap_err();
+    assert!(format!("{zero}").contains(">= 1"), "{zero}");
+    let bad = c.apply_kv("esplora_max_inflight", "nope").unwrap_err();
+    assert!(format!("{bad}").contains("esplora_max_inflight"), "{bad}");
+    let max = tokio::sync::Semaphore::MAX_PERMITS;
+    let over = c
+        .apply_kv("esplora_max_inflight", &(max + 1).to_string())
+        .unwrap_err();
+    assert!(format!("{over}").contains("<="), "{over}");
+    let huge = c
+        .apply_kv("esplora_max_inflight", "9999999999999999999")
+        .unwrap_err();
+    assert!(format!("{huge}").contains("esplora_max_inflight"), "{huge}");
+    assert_eq!(
+        c.esplora_max_inflight,
+        rbitcoin_electrum::DEFAULT_MAX_CONNECTIONS,
+        "a refused value leaves the default"
+    );
+}
+
 #[test]
 fn electrum_max_subs_cli_conf_and_bounds() {
     assert_eq!(