From 55d4712c9b78f51aab29fed6a38fbe5131869290 Mon Sep 17 00:00:00 2001 From: Bob Bobber Date: Fri, 23 Oct 2026 19:05:15 -0400 Subject: [PATCH 1/4] electrum: answer read-only batch items four at a time answer_batch ran every item one after another, so a 100-address get_history page cost the sum of 100 lookups. Read-only items now go to the blocking pool up to BATCH_CONCURRENCY (4) at a time per connection, each with the negotiated protocol and a copy of the join cache, and land in request order. Slot i is written before read i + 4 starts, so a client that stops reading still stops the work and a connection holds at most four unwritten replies. Any other item (subscribe, unsubscribe, server.version, broadcast and the rest) waits until every earlier slot is written and then runs alone through answer_one, so later items see what it changed. Notifications, non-object items, streaming methods and the 345 cap are unchanged. The single-request path is the same code split into spawn_dispatch and finish_reply, so metering, the API log and last_sent stay per item. A debug-only, opt-in thread-local trace records spawn, serial and write steps so the suite can check the window, the barrier and the order without timing. Co-Authored-By: Claude Opus 5.5 --- COMPAT.md | 2 +- changelog.d/electrum-batch-concurrency.md | 11 + crates/rbitcoin-electrum/src/server.rs | 441 ++++++++++++++----- crates/rbitcoin-electrum/tests/server/mod.rs | 232 ++++++++++ docs/operator/interfaces.md | 1 + 5 files changed, 565 insertions(+), 122 deletions(-) create mode 100644 changelog.d/electrum-batch-concurrency.md diff --git a/COMPAT.md b/COMPAT.md index c863849691..2d69c2b337 100644 --- a/COMPAT.md +++ b/COMPAT.md @@ -136,7 +136,7 @@ Per-method notes, auth, and the shindex matrix live in | Method | Status | Notes | |--------|--------|-------| | server.version / banner / features | done | Banner: libre-relay-class. `server.version[0]` is `rbitcoin-electrs electrumx` — **not electrs**; see below. `server.version` negotiates: omitted → `1.6`; `"1.4"` → `1.4`; `["1.4","1.4.2"]` → `1.4.2`; `["1.4","1.6"]` → `1.6`; `"1.4.2-asof"` (or a range containing it) → as-of dialect. First call wins. `features.protocol_max` is `1.6`; `features.asof_protocol` is `1.4.2-asof`. `features.genesis_hash` is display-order hex (refuse a wrong-chain server before a tweaks scan). `features.tweaks` / `silent_payments` advertise the method; they are not a substitute for the stream. | -| JSON-RPC batch | done | A line holding a JSON array is a JSON-RPC 2.0 batch: one array line back, same order, each slot `{id, result}` or `{id, error}`. Items run in order on the connection, and each slot is written as soon as it is answered. A notification (no `id` member) is skipped: it is not run and gets no slot, and a batch of only notifications writes nothing. A non-object item gets a `-32600` `Invalid Request` slot. `[]` is one `-32600` `empty batch`; more than 345 items is one `-32600` `batch too large`. `blockchain.tweaks.subscribe` and `blockchain.silentpayments.subscribe` stream several lines, so inside a batch they get a `-32600` slot; send them alone. | +| JSON-RPC batch | done | A line holding a JSON array is a JSON-RPC 2.0 batch: one array line back, same order, each slot `{id, result}` or `{id, error}`. Read-only items (scripthash `get_history` / `get_balance` / `listunspent` / `get_mempool`, `transaction.get` / `get_merkle` / `id_from_pos`, `block.header(s)`, `estimatefee`, `outpoint.get_status`, `mempool.get_fee_histogram` / `get_info`) run up to 4 at a time; any other item waits for every earlier slot, then runs alone, so a `subscribe` is seen by the items after it. Slots are written in request order as soon as each is answered, and slot `i` is written before read `i + 4` starts. A notification (no `id` member) is skipped: it is not run and gets no slot, and a batch of only notifications writes nothing. A non-object item gets a `-32600` `Invalid Request` slot. `[]` is one `-32600` `empty batch`; more than 345 items is one `-32600` `batch too large`. `blockchain.tweaks.subscribe` and `blockchain.silentpayments.subscribe` stream several lines, so inside a batch they get a `-32600` slot; send them alone. | | blockchain.tweaks.subscribe | done | **Stream**, not a one-shot result: JSON-RPC `result` is the **first** height only; remaining heights are unsolicited notifications; `{"message":"done"}` ends the **chunk** (60s wall at a wave boundary, or the requested `count` if sooner) — Cake electrs clamps `count` to 1000 instead. Cake Wallet resubscribes from `syncHeight+1` ([cake_wallet#3574](https://github.com/cake-tech/cake_wallet/issues/3574) persist still lags one event). [kiss-bdk](https://github.com/kkdao/kiss-bdk) today treats `done` as the asked range finished ([kiss-bdk#10](https://github.com/kkdao/kiss-bdk/issues/10)). Each height carries tweak + txid + taproot `output_pubkeys` (client scans locally; no block fetch). Naive walk, or `--sp-tweaks` thin index (`len:tweak` only; one `txout` span per wave; the first height shares that span when indexed). Pre-taproot: **one** notify with ≤1024 empty height keys (Cake `fromJson` last key is progress); probe `[0,1,false]` stays `{"0": {}}`. Param `[2]` (Cake `historicalMode`): `false` **cut-through** (omit confirmed-spent P2TR outs / txs); `true` keeps spent outs. Serve-time `--sp-tweaks-dust` (default **1000**) omits P2TR outs with `value <=` the floor (`0` = all; **546** matches Cake electrs `sp_min_dust`). Sparrow Silent Payments uses Frigate `blockchain.silentpayments.subscribe` (server-side scan) — **not** this method. Cake isolate may still hardcode `electrs.cakewallet.com`. | | headers / block headers | done | Tip push on subscribe | | scripthash history / balance / listunspent | done | Unconf when mempool attached; `get_history` optional BCH-style `from_height` / exclusive `to_height` (`-1` = tip + mempool); 1-arg = full history; a `from_height` above 0 scans that block window and is served above `--max-sh-creates` (open history, and a window that starts at height 0, still refuse above the cap); unconfirmed `get_history` rows include `fee` (same as `get_mempool`); `listunspent` mempool height is `0` or `-1` (unconfirmed parent); **subscribe status always full**; `blockchain.scripthash.unsubscribe` returns whether the connection was watching (frees a slot under `--electrum-max-subs`, default 10000 per connection). `listunspent` loads `txid.body` only for unspent creates; one TCP connection reuses the last SH outs+spent join until SH-view **hash** changes. Confirmed methods stamp live tip: a RAM SH head (pending write-behind) joins with durable SH so mempool can drop confirmed txs without a hole. Durable Class B seed waits until after tip announce. `get_history` skips mempool rows already in the confirmed list. `server.features.chain_tip = true`. Trailing **`asof:`** after the official args (`server.features.asof` / `asof_protocol = 1.4.2-asof`): confirmed rows as of that still-live ancestor **at or behind visible SH** (durable + pending), **no** mempool; stamp is the asof block; unknown hash or ahead of visible SH → `asof not on chain`. Prefix keeps it off the future positional-string landmine. Requires negotiated `1.4.2-asof` (first `server.version` only). Electrum `protocol_max` is **1.6**. | diff --git a/changelog.d/electrum-batch-concurrency.md b/changelog.d/electrum-batch-concurrency.md new file mode 100644 index 0000000000..8ef91739a2 --- /dev/null +++ b/changelog.d/electrum-batch-concurrency.md @@ -0,0 +1,11 @@ +Changed + +- **Electrum batches answer read-only items four at a time.** A batch + line ran its items one after another, so a 100-address `get_history` + page cost the sum of 100 lookups. Read-only items (scripthash history, + balance, unspent and mempool, transaction and header reads, fee + estimates) now run up to four at once per connection and still land in + request order. A subscribe, unsubscribe or other stateful item waits for + the reads before it and runs alone. Slot `i` is written before read + `i + 4` starts, so a client that stops reading still stops the work + (#1174). diff --git a/crates/rbitcoin-electrum/src/server.rs b/crates/rbitcoin-electrum/src/server.rs index 2e952dcce6..10abdf4f48 100644 --- a/crates/rbitcoin-electrum/src/server.rs +++ b/crates/rbitcoin-electrum/src/server.rs @@ -13,7 +13,7 @@ use rbitcoin_primitives::{Fk, Height}; use rbitcoin_query::{ChainView, ChainViewKind, HistoryFilter, Query, ShJoinSlot}; use rbitcoin_store::StoreError; use serde_json::{json, Value}; -use std::collections::{HashMap, HashSet}; +use std::collections::{HashMap, HashSet, VecDeque}; use std::net::SocketAddr; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex, OnceLock}; @@ -677,6 +677,58 @@ where Ok(()) } +/// Read-only batch items answered at once on one connection, counted from +/// spawn until the slot is written. Each holds a blocking thread while it +/// runs and its reply until written. The shared blocking pool (`max(cores, +/// 4)` threads) bounds all connections together. On a 4-core node, splitting +/// a 100-item `get_history` page across connections stopped gaining at 4. +#[doc(hidden)] +pub const BATCH_CONCURRENCY: usize = 4; + +/// One scheduling step of a batch, by item index. Recorded only for tests. +#[doc(hidden)] +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum BatchStep { + Spawn(usize), + Serial(usize), + Write(usize), +} + +#[cfg(debug_assertions)] +thread_local! { + static BATCH_TRACE_ON: std::cell::Cell = const { std::cell::Cell::new(false) }; + static BATCH_TRACE: std::cell::RefCell> = + const { std::cell::RefCell::new(Vec::new()) }; +} + +fn batch_mark(_step: BatchStep) { + #[cfg(debug_assertions)] + if BATCH_TRACE_ON.with(|on| on.get()) { + BATCH_TRACE.with(|trace| trace.borrow_mut().push(_step)); + } +} + +/// Start recording batch steps run on this thread, from an empty trace. A +/// debug node schedules every batch through `batch_mark` and nothing reads +/// the trace, so it records only after this call. +#[cfg(debug_assertions)] +pub fn batch_trace_arm() { + BATCH_TRACE.with(|trace| trace.borrow_mut().clear()); + BATCH_TRACE_ON.with(|on| on.set(true)); +} + +/// Stop recording on this thread and drop what was recorded. +#[cfg(debug_assertions)] +pub fn batch_trace_clear() { + BATCH_TRACE_ON.with(|on| on.set(false)); + BATCH_TRACE.with(|trace| trace.borrow_mut().clear()); +} + +#[cfg(debug_assertions)] +pub fn batch_trace_snapshot() -> Vec { + BATCH_TRACE.with(|trace| trace.borrow().clone()) +} + /// The connection's fixed inputs to one request. struct RequestCtx<'a> { peer: SocketAddr, @@ -705,11 +757,128 @@ fn invalid_request(id: Value, message: &str) -> Value { json!({"jsonrpc":"2.0","id": id, "error": {"code": -32600, "message": message}}) } -/// JSON-RPC 2.0 batch, one array line on the wire. Each slot is written as -/// soon as it is answered, so socket backpressure holds the connection to -/// about one reply instead of the whole array. Items run in order because -/// each moves `conn` in and out of the blocking pool. A batch of only -/// notifications writes nothing. +/// Methods a batch may answer out of turn. They read the store, the +/// mempool, or the negotiated protocol and never change connection state. +/// `sh_join` is only a cache of the last scripthash join, so an item that +/// starts from an older copy recomputes; it does not answer differently. +#[doc(hidden)] +pub fn method_reads_only(method: &str) -> bool { + matches!( + method, + "blockchain.scripthash.get_history" + | "blockchain.scripthash.get_balance" + | "blockchain.scripthash.listunspent" + | "blockchain.scripthash.get_mempool" + | "blockchain.transaction.get" + | "blockchain.transaction.get_merkle" + | "blockchain.transaction.id_from_pos" + | "blockchain.block.header" + | "blockchain.block.headers" + | "blockchain.estimatefee" + | "blockchain.outpoint.get_status" + | "mempool.get_fee_histogram" + | "mempool.get_info" + ) +} + +/// One request as asked, and when it was taken up. +struct Asked { + id: Value, + method: String, + params: Value, + t0: Instant, +} + +impl Asked { + fn of(req: &Value) -> Self { + Self { + id: req.get("id").cloned().unwrap_or(Value::Null), + method: req + .get("method") + .and_then(|m| m.as_str()) + .unwrap_or("") + .to_string(), + params: req.get("params").cloned().unwrap_or(json!([])), + t0: Instant::now(), + } + } +} + +/// How a batch takes up one item. +enum BatchItem { + /// A notification (no `id` member, unlike `"id": null`). It is not run: + /// no reply means no slot write to hold back a client that stops + /// reading, and Electrum clients never send notifications. + Skip, + /// Answered without running anything. + Reply(Value), + /// [`method_reads_only`]: may run beside other reads. + Read(Asked), + /// Touches connection state: runs alone, after every earlier slot. + Serial, +} + +fn batch_item(item: &Value) -> BatchItem { + let Some(obj) = item.as_object() else { + return BatchItem::Reply(invalid_request(Value::Null, "Invalid Request")); + }; + let Some(id) = obj.get("id") else { + return BatchItem::Skip; + }; + if request_stream(item).is_some() { + return BatchItem::Reply(invalid_request( + id.clone(), + "streaming method cannot be sent in a batch", + )); + } + let asked = Asked::of(item); + if method_reads_only(&asked.method) { + BatchItem::Read(asked) + } else { + BatchItem::Serial + } +} + +/// The reply array, written slot by slot as one line. +struct BatchOut<'w, W> { + writer: &'w mut W, + idle: Duration, + opened: bool, +} + +impl BatchOut<'_, W> { + async fn slot(&mut self, index: usize, answer: &Value) -> Result<(), std::io::Error> { + batch_mark(BatchStep::Write(index)); + let mut slot = vec![if self.opened { b',' } else { b'[' }]; + slot.extend(serde_json::to_vec(answer).unwrap_or_else(|_| b"{}".to_vec())); + self.opened = true; + write_bytes_within(self.writer, &slot, self.idle).await + } + + async fn close(self) -> Result<(), std::io::Error> { + if !self.opened { + return Ok(()); + } + write_bytes_within(self.writer, b"]\n", self.idle).await + } +} + +type Dispatched = (Result, ElectrumConn, Option); + +/// A read running on the blocking pool, not yet written. +struct InFlight { + index: usize, + asked: Asked, + task: JoinHandle, +} + +/// JSON-RPC 2.0 batch, one array line on the wire, slots in request order. +/// Read-only items run up to [`BATCH_CONCURRENCY`] at once; slot `i` is +/// written before read `i + BATCH_CONCURRENCY` starts, so a client that +/// stops reading stops the work and the connection holds at most that many +/// unwritten replies. Any other item waits until every earlier slot is +/// written, then runs alone, so later items see what it did. A batch of +/// only notifications writes nothing. async fn answer_batch( ctx: &RequestCtx<'_>, conn: &mut ElectrumConn, @@ -729,44 +898,116 @@ async fn answer_batch( ) .await; } - let mut opened = false; - for item in &items { - let Some(answer) = answer_batch_item(ctx, conn, last_sent, item).await? else { - continue; + let mut out = BatchOut { + writer, + idle, + opened: false, + }; + let mut window: VecDeque = VecDeque::with_capacity(BATCH_CONCURRENCY); + for (index, item) in items.iter().enumerate() { + let answer = match batch_item(item) { + BatchItem::Skip => continue, + BatchItem::Read(asked) => { + if window.len() >= BATCH_CONCURRENCY { + if let Some(oldest) = window.pop_front() { + land(ctx, conn, last_sent, &mut out, oldest).await?; + } + } + batch_mark(BatchStep::Spawn(index)); + let task = spawn_dispatch(ctx, read_conn(conn), &asked); + window.push_back(InFlight { index, asked, task }); + continue; + } + BatchItem::Reply(answer) => answer, + BatchItem::Serial => { + while let Some(oldest) = window.pop_front() { + land(ctx, conn, last_sent, &mut out, oldest).await?; + } + batch_mark(BatchStep::Serial(index)); + answer_one(ctx, conn, last_sent, item).await? + } }; - let mut slot = vec![if opened { b',' } else { b'[' }]; - slot.extend(serde_json::to_vec(&answer).unwrap_or_else(|_| b"{}".to_vec())); - opened = true; - write_bytes_within(writer, &slot, idle).await?; + while let Some(oldest) = window.pop_front() { + land(ctx, conn, last_sent, &mut out, oldest).await?; + } + out.slot(index, &answer).await?; } - if opened { - write_bytes_within(writer, b"]\n", idle).await?; + while let Some(oldest) = window.pop_front() { + land(ctx, conn, last_sent, &mut out, oldest).await?; } - Ok(()) + out.close().await } -/// `None` for a notification (no `id` member, unlike `"id": null`). It is -/// not run: no reply means no slot write to hold back a client that stops -/// reading, and Electrum clients never send notifications. -async fn answer_batch_item( +/// The connection state a read needs: the negotiated protocol and a copy of +/// the join cache. Subscription sets stay with the connection. +fn read_conn(conn: &ElectrumConn) -> ElectrumConn { + ElectrumConn { + protocol: conn.protocol.clone(), + header_sub: conn.header_sub, + sh_subs: HashSet::new(), + sh_join: conn.sh_join.clone(), + outpoint_subs: HashSet::new(), + sp_sub: None, + sp_scan_busy: conn.sp_scan_busy, + } +} + +/// Wait for the oldest read and write its slot. Its join cache becomes the +/// connection's, as if the reads had run one after another. +async fn land( ctx: &RequestCtx<'_>, conn: &mut ElectrumConn, last_sent: &mut HashMap<[u8; 32], String>, - item: &Value, -) -> Result, std::io::Error> { - let Some(obj) = item.as_object() else { - return Ok(Some(invalid_request(Value::Null, "Invalid Request"))); - }; - let Some(id) = obj.get("id") else { - return Ok(None); - }; - if request_stream(item).is_some() { - return Ok(Some(invalid_request( - id.clone(), - "streaming method cannot be sent in a batch", - ))); + out: &mut BatchOut<'_, W>, + flight: InFlight, +) -> Result<(), std::io::Error> { + let (result, work, stamped) = flight.task.await.map_err(dispatch_join_error)?; + conn.sh_join = work.sh_join; + let answer = finish_reply(ctx, conn, last_sent, flight.asked, result, stamped); + out.slot(flight.index, &answer).await +} + +fn dispatch_join_error(e: tokio::task::JoinError) -> std::io::Error { + std::io::Error::other(format!("electrum dispatch join: {e}")) +} + +fn spawn_dispatch( + ctx: &RequestCtx<'_>, + work: ElectrumConn, + asked: &Asked, +) -> JoinHandle { + let q = Arc::clone(ctx.query); + let cfg = Arc::clone(ctx.config); + let p = Arc::clone(ctx.params); + let mp = ctx.mempool.clone(); + let method = asked.method.clone(); + let params = asked.params.clone(); + tokio::task::spawn_blocking(move || { + let _g = BlockingRegion::enter(); + dispatch_off_worker(&q, &cfg, &p, mp.as_deref(), &method, ¶ms, work) + }) +} + +/// The blocking half of a request, run with `work` moved in and handed back. +fn dispatch_off_worker( + q: &Query, + cfg: &ElectrumConfig, + p: &ChainParams, + mp: Option<&MempoolHub>, + method: &str, + params: &Value, + mut work: ElectrumConn, +) -> Dispatched { + if !method_stamps_chain_tip(method) { + let r = dispatch_pinned(method, params, q, cfg, p, mp, &mut work, None, false); + return (r, work, None); } - answer_one(ctx, conn, last_sent, item).await.map(Some) + let asof_ok = work.protocol.as_str() == PROTOCOL_ASOF; + let (r, view) = + electrum_at_chain_view(q, method, params, asof_ok, |q, params, view, is_asof| { + dispatch_pinned(method, params, q, cfg, p, mp, &mut work, view, is_asof) + }); + (r, work, view) } /// One request/response method. `Err` is a dispatch task that died, which @@ -777,101 +1018,59 @@ async fn answer_one( last_sent: &mut HashMap<[u8; 32], String>, req: &Value, ) -> Result { - let id = req.get("id").cloned().unwrap_or(Value::Null); - let method = req.get("method").and_then(|m| m.as_str()).unwrap_or(""); - let params_v = req.get("params").cloned().unwrap_or(json!([])); - let t0 = Instant::now(); - let mut stamped: Option = None; - let result = if method_stays_on_worker(method) { - dispatch_with_join( - method, - ¶ms_v, + let asked = Asked::of(req); + if method_stays_on_worker(&asked.method) { + let result = dispatch_with_join( + &asked.method, + &asked.params, ctx.query, ctx.config, ctx.params, ctx.mempool.as_deref(), conn, - ) - } else { - let q = Arc::clone(ctx.query); - let cfg = Arc::clone(ctx.config); - let p = Arc::clone(ctx.params); - let mp = ctx.mempool.clone(); - let method_owned = method.to_string(); - let params_owned = params_v.clone(); - // Move the sub sets in and back (`*conn = work` below): a - // clone per request made a wallet's N-address sync O(N²). - // A JoinError ends the connection, so nothing reuses them. - let mut work = ElectrumConn { - protocol: conn.protocol.clone(), - header_sub: conn.header_sub, - sh_subs: std::mem::take(&mut conn.sh_subs), - sh_join: conn.sh_join.take(), - outpoint_subs: std::mem::take(&mut conn.outpoint_subs), - sp_sub: conn.sp_sub.clone(), - sp_scan_busy: conn.sp_scan_busy, - }; - let stamp = method_stamps_chain_tip(&method_owned); - match tokio::task::spawn_blocking(move || { - let _g = BlockingRegion::enter(); - let asof_ok = work.protocol.as_str() == PROTOCOL_ASOF; - let (r, view) = if stamp { - electrum_at_chain_view( - &q, - &method_owned, - ¶ms_owned, - asof_ok, - |q, params, view, is_asof| { - dispatch_pinned( - &method_owned, - params, - q, - &cfg, - &p, - mp.as_deref(), - &mut work, - view, - is_asof, - ) - }, - ) - } else { - ( - dispatch_pinned( - &method_owned, - ¶ms_owned, - &q, - &cfg, - &p, - mp.as_deref(), - &mut work, - None, - false, - ), - None, - ) - }; - (r, work, view) - }) - .await - { - Ok((r, work, view)) => { - *conn = work; - stamped = view; - r - } - Err(e) => { - return Err(std::io::Error::other(format!( - "electrum dispatch join: {e}" - ))); - } - } + ); + return Ok(finish_reply(ctx, conn, last_sent, asked, result, None)); + } + // Move the sub sets in and back (`*conn = work` below): a clone per + // request made a wallet's N-address sync O(N²). A JoinError ends the + // connection, so nothing reuses them. + let work = ElectrumConn { + protocol: conn.protocol.clone(), + header_sub: conn.header_sub, + sh_subs: std::mem::take(&mut conn.sh_subs), + sh_join: conn.sh_join.take(), + outpoint_subs: std::mem::take(&mut conn.outpoint_subs), + sp_sub: conn.sp_sub.clone(), + sp_scan_busy: conn.sp_scan_busy, }; + let (result, work, stamped) = spawn_dispatch(ctx, work, &asked) + .await + .map_err(dispatch_join_error)?; + *conn = work; + Ok(finish_reply(ctx, conn, last_sent, asked, result, stamped)) +} + +/// Status bookkeeping, metering, the API log line, and the reply object. +fn finish_reply( + ctx: &RequestCtx<'_>, + conn: &ElectrumConn, + last_sent: &mut HashMap<[u8; 32], String>, + asked: Asked, + result: Result, + stamped: Option, +) -> Value { + let Asked { + id, + method, + params: params_v, + t0, + } = asked; + let method = method.as_str(); drop_unsubscribed_status(last_sent, &conn.sh_subs); let wall_ms = t0.elapsed().as_millis() as u64; meter_dispatch_wall(t0.elapsed().as_micros() as u64); let params_s = serde_json::to_string(¶ms_v).unwrap_or_else(|_| "[]".into()); - Ok(match result { + match result { Ok(v) => { if method == "blockchain.scripthash.subscribe" { if let Ok(sh) = param_scripthash(¶ms_v, 0) { @@ -905,7 +1104,7 @@ async fn answer_one( ); json!({"jsonrpc":"2.0","id": id, "error": {"code": 1, "message": e}}) } - }) + } } #[allow(clippy::too_many_arguments)] // call-site args stay unbundled diff --git a/crates/rbitcoin-electrum/tests/server/mod.rs b/crates/rbitcoin-electrum/tests/server/mod.rs index b74e956059..6e3dbc0f8a 100644 --- a/crates/rbitcoin-electrum/tests/server/mod.rs +++ b/crates/rbitcoin-electrum/tests/server/mod.rs @@ -633,6 +633,238 @@ async fn batch_line_answers_every_item_in_one_array() { let _ = std::fs::remove_dir_all(&dir); } +fn get_balance(id: usize) -> Value { + json!({"jsonrpc":"2.0","id":id,"method":"blockchain.scripthash.get_balance", + "params":[format!("{id:064x}")]}) +} + +fn spawned(trace: &[BatchStep]) -> Vec { + trace + .iter() + .filter_map(|s| match s { + BatchStep::Spawn(i) => Some(*i), + _ => None, + }) + .collect() +} + +fn written(trace: &[BatchStep]) -> Vec { + trace + .iter() + .filter_map(|s| match s { + BatchStep::Write(i) => Some(*i), + _ => None, + }) + .collect() +} + +fn at(trace: &[BatchStep], step: BatchStep) -> usize { + trace + .iter() + .position(|s| *s == step) + .unwrap_or_else(|| panic!("{step:?} missing from {trace:?}")) +} + +/// Read-only batch items run up to `BATCH_CONCURRENCY` at a time and land in +/// request order. A subscribe waits for every earlier read to be written, +/// runs alone, and later items see what it did. The connection task runs on +/// this thread (current-thread runtime), so the armed trace sees its steps. +#[allow(clippy::cognitive_complexity)] // one batch, every scheduling rule +#[tokio::test] +async fn batch_reads_overlap_and_land_in_request_order() { + let (dir, q) = tmp_store(); + let params = ChainParams::regtest(); + let (tip_tx, _) = broadcast::channel(4); + let cfg = ElectrumConfig::for_params("127.0.0.1:0".parse().unwrap(), ¶ms); + let handle = run_electrum(cfg, Arc::new(q), params, tip_tx, None) + .await + .expect("listen"); + let mut stream = TcpStream::connect(handle.local_addr).await.unwrap(); + + let watched = format!("{:064x}", 0xabc); + let mut items: Vec = (0..6).map(get_balance).collect(); + items.push(json!({"jsonrpc":"2.0","id":6,"method":"blockchain.scripthash.subscribe","params":[watched]})); + items.extend((7..10).map(get_balance)); + items.push(json!({"jsonrpc":"2.0","id":10,"method":"blockchain.scripthash.unsubscribe","params":[watched]})); + items.push( + json!({"jsonrpc":"2.0","method":"blockchain.scripthash.get_balance","params":[watched]}), + ); + items.push(json!(12)); + + batch_trace_arm(); + let reply = electrum_tcp_batch(&mut stream, &items).await; + let trace = batch_trace_snapshot(); + batch_trace_clear(); + + let ids: Vec = reply + .as_array() + .expect("one array line") + .iter() + .map(|v| v["id"].clone()) + .collect(); + let mut want: Vec = (0..=10).map(|i| json!(i)).collect(); + want.push(Value::Null); + assert_eq!(ids, want, "{reply}"); + assert_eq!( + reply[10]["result"], + json!(true), + "the subscribe ran before the unsubscribe: {reply}" + ); + + assert_eq!(written(&trace), [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 12]); + assert_eq!(spawned(&trace), [0, 1, 2, 3, 4, 5, 7, 8, 9], "{trace:?}"); + let mut open = 0usize; + let mut peak = 0usize; + for step in &trace { + match step { + BatchStep::Spawn(_) => open += 1, + BatchStep::Write(i) if spawned(&trace).contains(i) => open -= 1, + _ => {} + } + peak = peak.max(open); + } + assert_eq!(peak, BATCH_CONCURRENCY, "{trace:?}"); + assert!( + at(&trace, BatchStep::Spawn(1)) < at(&trace, BatchStep::Write(0)), + "reads overlap: {trace:?}" + ); + for barrier in [6, 10] { + assert!( + at(&trace, BatchStep::Write(barrier - 1)) < at(&trace, BatchStep::Serial(barrier)), + "{barrier} waits for earlier reads: {trace:?}" + ); + } + assert!( + at(&trace, BatchStep::Write(6)) < at(&trace, BatchStep::Spawn(7)), + "reads after a subscribe start after it: {trace:?}" + ); + + handle.shutdown().await; + let _ = std::fs::remove_dir_all(&dir); +} + +/// A client that stops reading stops the batch: the server writes slot i +/// before it starts read i + `BATCH_CONCURRENCY`. +#[tokio::test] +async fn stalled_batch_reader_holds_reads_to_the_window() { + use std::pin::Pin; + use std::task::{Context, Poll}; + use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; + + struct OnceRead(Vec); + impl AsyncRead for OnceRead { + fn poll_read( + self: Pin<&mut Self>, + _cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + let this = self.get_mut(); + if this.0.is_empty() { + return Poll::Pending; + } + let n = this.0.len().min(buf.remaining()); + buf.put_slice(&this.0[..n]); + this.0.drain(..n); + Poll::Ready(Ok(())) + } + } + /// Takes the first write, then never another. + struct OneWrite { + took: bool, + stalled: Arc, + } + impl AsyncWrite for OneWrite { + fn poll_write( + self: Pin<&mut Self>, + _cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + let this = self.get_mut(); + if this.took { + this.stalled.store(true, Ordering::SeqCst); + return Poll::Pending; + } + this.took = true; + Poll::Ready(Ok(buf.len())) + } + fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Ok(())) + } + } + struct Duplex(OnceRead, OneWrite); + impl AsyncRead for Duplex { + fn poll_read( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + Pin::new(&mut self.get_mut().0).poll_read(cx, buf) + } + } + impl AsyncWrite for Duplex { + fn poll_write( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + Pin::new(&mut self.get_mut().1).poll_write(cx, buf) + } + fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.get_mut().1).poll_flush(cx) + } + fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.get_mut().1).poll_shutdown(cx) + } + } + + let (dir, q) = tmp_store(); + let params = ChainParams::regtest(); + let cfg = ElectrumConfig::for_params("127.0.0.1:0".parse().unwrap(), ¶ms); + let items: Vec = (0..20).map(get_balance).collect(); + let mut line = serde_json::to_vec(&items).unwrap(); + line.push(b'\n'); + let stalled = Arc::new(AtomicBool::new(false)); + let (_tip_tx, tip_rx) = broadcast::channel(1); + batch_trace_arm(); + let task = tokio::spawn(handle_client( + Duplex( + OnceRead(line), + OneWrite { + took: false, + stalled: Arc::clone(&stalled), + }, + ), + "127.0.0.1:9".parse().unwrap(), + Arc::new(q), + Arc::new(cfg), + Arc::new(params), + tip_rx, + None, + Arc::new(AtomicBool::new(false)), + )); + let deadline = Instant::now() + Duration::from_secs(5); + while !stalled.load(Ordering::SeqCst) { + assert!( + Instant::now() < deadline, + "the second slot was never written" + ); + tokio::time::sleep(Duration::from_millis(5)).await; + } + let trace = batch_trace_snapshot(); + batch_trace_clear(); + task.abort(); + assert_eq!(written(&trace), [0, 1], "{trace:?}"); + assert_eq!( + spawned(&trace), + (0..=BATCH_CONCURRENCY).collect::>(), + "slot 0 is written before read {BATCH_CONCURRENCY} starts, and slot 1 blocks the rest: {trace:?}" + ); + let _ = std::fs::remove_dir_all(&dir); +} + /// A batch reply goes to the socket one slot per write, so a client that /// reads slowly holds the server to about one reply, not the whole array. #[tokio::test] diff --git a/docs/operator/interfaces.md b/docs/operator/interfaces.md index 33bd38c564..75ae060688 100644 --- a/docs/operator/interfaces.md +++ b/docs/operator/interfaces.md @@ -108,6 +108,7 @@ and idle clients fail closed. | Max scripthash subs / conn | 10000 (`--electrum-max-subs`) | Notify fan-out cap. A wallet subscribes every address up to its gap limit; each sub is ~170 B (≈0.45 GB worst case at 256 connections) plus one posting check per block against that block's shared touch set; a reorg restatuses every sub | | Max broadcast hex | ~8 MiB | `transaction.broadcast` hex length | | Max batch items | 345 | Requests in one JSON-RPC batch line; more is one `-32600`. The reply is written one slot at a time, so a slow reader holds back the server, not RAM | +| Batch reads at once | 4 | Read-only batch items in flight per connection (blocking threads and unwritten replies). The blocking pool, `max(cores, 4)` threads, bounds all connections together. Four scripthash joins at once can each reach `--max-sh-creates` | Edge rate-limits, auth, and TLS cipher policy stay on the proxy. See [`SECURITY.md`](../../SECURITY.md). From 1f43db7edbb437e7fc277f3eee77fb4e12d59320 Mon Sep 17 00:00:00 2001 From: Bob Bobber Date: Fri, 23 Oct 2026 21:01:45 -0400 Subject: [PATCH 2/4] query: run heavy scripthash joins one at a time node-wide One unpaged scripthash join near a raised --max-sh-creates was measured at 2.1 GiB peak RSS, and a batch line (or a few connections) could start several at once. An unpaged join with more than --heavy-sh-creates creates (conf heavy_sh_creates=, default 10000, 0 off) now takes the Query's single heavy turn before it walks, so one runs at a time across every Electrum and Esplora client. Lighter joins and paged history do not wait. The create count is the one the --max-sh-creates refusal already reads, so a refused join never takes the turn. The turn is a guard taken on the blocking pool and dropped when the join returns, on success or error. Its holder only reads the store and never waits on a client, so a waiter in the same connection's batch window cannot deadlock it. A debug-only, opt-in probe on Query counts joins by kind and makes each wait for a second of its kind, so the journey checks that heavy joins never overlap while light ones do, from two connections and from one batch. Co-Authored-By: Claude Opus 5.5 --- COMPAT.md | 2 + changelog.d/electrum-batch-concurrency.md | 6 + .../tests/electrum_sh_journey.rs | 102 +++++++++++++- crates/rbitcoin-electrum/tests/server/mod.rs | 2 +- crates/rbitcoin-node/src/cli.rs | 3 +- crates/rbitcoin-node/src/config.rs | 9 ++ crates/rbitcoin-node/src/run.rs | 1 + crates/rbitcoin-node/tests/cli.rs | 6 + crates/rbitcoin-node/tests/config.rs | 7 + crates/rbitcoin-query/src/lib.rs | 26 ++++ crates/rbitcoin-query/src/scripthash.rs | 127 +++++++++++++++++- docs/operator/interfaces.md | 3 +- docs/operator/operations.md | 1 + docs/operator/storage.md | 6 + 14 files changed, 290 insertions(+), 11 deletions(-) diff --git a/COMPAT.md b/COMPAT.md index 2d69c2b337..ad2d544682 100644 --- a/COMPAT.md +++ b/COMPAT.md @@ -48,6 +48,8 @@ electrs-shaped explorers can verify and sync. `--max-sh-creates N` (default **10000**; **0** = unlimited) refuses an unpaged Electrum/Esplora SH joins with more than N creates: Esplora HTTP **503**, Electrum JSON-RPC error `scripthash join exceeds --max-sh-creates`. Stats stay full when under the cap. +`--heavy-sh-creates N` (default **10000**; **0** = off) runs unpaged joins with +more than N creates one at a time across the node; lighter joins stay parallel. `GET …/txs/summary` is a **dialect** route (not Blockstream Esplora `API.md`). Shape and owner row: [Esplora REST surface](#esplora-rest-surface). diff --git a/changelog.d/electrum-batch-concurrency.md b/changelog.d/electrum-batch-concurrency.md index 8ef91739a2..42550c214a 100644 --- a/changelog.d/electrum-batch-concurrency.md +++ b/changelog.d/electrum-batch-concurrency.md @@ -9,3 +9,9 @@ Changed the reads before it and runs alone. Slot `i` is written before read `i + 4` starts, so a client that stops reading still stops the work (#1174). +- **One heavy scripthash join at a time.** An unpaged Electrum or Esplora + join with more than `--heavy-sh-creates` creates (conf + `heavy_sh_creates=`, default 10000, `0` off) now waits until no other + such join runs anywhere on the node. One join near a raised + `--max-sh-creates` can peak above 2 GiB, and a batch or a few + connections could start several at once. Lighter joins stay parallel. diff --git a/crates/rbitcoin-electrum/tests/electrum_sh_journey.rs b/crates/rbitcoin-electrum/tests/electrum_sh_journey.rs index 8fa9d9c43d..f52c3b5340 100644 --- a/crates/rbitcoin-electrum/tests/electrum_sh_journey.rs +++ b/crates/rbitcoin-electrum/tests/electrum_sh_journey.rs @@ -18,7 +18,7 @@ use tokio::io::{AsyncBufReadExt, BufReader}; use tokio::net::TcpStream; use tokio::sync::broadcast; -use crate::server::electrum_tcp_rpc; +use crate::server::{electrum_tcp_batch, electrum_tcp_rpc}; fn tmp_store() -> (rbitcoin_query::testutil::TempDir, Query) { rbitcoin_query::testutil::tiny_query_labeled("electrum-sh") @@ -211,6 +211,105 @@ pub fn casa_cap_then_balance_history_unspent(q: &Query) { assert!(!unspent.as_array().unwrap().is_empty()); } +/// Unpaged joins above `--heavy-sh-creates` run one at a time across every +/// connection, from separate connections and from one batch, while lighter +/// joins overlap. The armed probe makes each join wait inside for a second +/// of its kind, so the peaks only stay at 1 because of the lock. A refused +/// join takes no turn, and a turn is given back after every join. +pub async fn heavy_joins_take_turns(q: Arc) { + let params = ChainParams::regtest(); + let cfg = ElectrumConfig::for_params("127.0.0.1:0".parse().unwrap(), ¶ms); + let (tip_tx, _) = broadcast::channel(4); + let handle = run_electrum(cfg, Arc::clone(&q), params, tip_tx, None) + .await + .expect("listen"); + let heavy = electrum_scripthash_hex(&[0x51]); + let balance = |id: u32, sh: &str| json!({"jsonrpc":"2.0","id":id,"method":"blockchain.scripthash.get_balance","params":[sh]}); + q.set_heavy_sh_creates(1); + q.set_max_sh_creates(2); + let mut refused = TcpStream::connect(handle.local_addr).await.unwrap(); + let err = electrum_tcp_rpc( + &mut refused, + 1, + "blockchain.scripthash.get_balance", + json!([heavy]), + ) + .await; + assert!( + err["error"]["message"] + .as_str() + .unwrap_or("") + .contains("--max-sh-creates"), + "{err}" + ); + q.set_max_sh_creates(0); + q.sh_join_probe.arm(); + + let mut a = TcpStream::connect(handle.local_addr).await.unwrap(); + let mut b = TcpStream::connect(handle.local_addr).await.unwrap(); + let one = |sh: &str| { + let mut line = serde_json::to_string(&balance(1, sh)).unwrap(); + line.push('\n'); + line + }; + use tokio::io::AsyncWriteExt; + a.write_all(one(&heavy).as_bytes()).await.unwrap(); + b.write_all(one(&heavy).as_bytes()).await.unwrap(); + let mut answers = Vec::new(); + for s in [&mut a, &mut b] { + let mut line = String::new(); + BufReader::new(s).read_line(&mut line).await.unwrap(); + answers.push(serde_json::from_str::(&line).unwrap()); + } + assert_eq!(answers[0]["result"], answers[1]["result"], "{answers:?}"); + assert!(answers[0]["result"]["confirmed"].as_u64().unwrap() > 0); + + let light_a = format!("{:064x}", 0x11); + let light_b = format!("{:064x}", 0x22); + let mut c = TcpStream::connect(handle.local_addr).await.unwrap(); + let batch = electrum_tcp_batch( + &mut c, + &[ + balance(1, &heavy), + balance(2, &heavy), + balance(3, &light_a), + balance(4, &light_b), + ], + ) + .await; + let slots = batch.as_array().expect("array"); + assert_eq!(slots.len(), 4, "{batch}"); + assert_eq!(slots[0]["result"], answers[0]["result"], "{batch}"); + assert_eq!(slots[1]["result"], answers[0]["result"], "{batch}"); + assert_eq!(q.sh_join_probe.entered(), (2, 4), "two light, four heavy"); + assert_eq!( + q.sh_join_probe.peaks(), + (2, 1), + "light joins overlap, heavy joins take turns" + ); + + let creates = q + .scripthash_create_count(&rbitcoin_primitives::parse_display_hash32(&heavy).unwrap()) + .unwrap(); + q.set_heavy_sh_creates(creates); + q.sh_join_probe.arm(); + let mut d = TcpStream::connect(handle.local_addr).await.unwrap(); + electrum_tcp_rpc( + &mut d, + 1, + "blockchain.scripthash.get_balance", + json!([heavy]), + ) + .await; + assert_eq!( + q.sh_join_probe.entered(), + (1, 0), + "a join at exactly --heavy-sh-creates is light" + ); + q.set_heavy_sh_creates(rbitcoin_query::DEFAULT_HEAVY_SH_CREATES); + handle.shutdown().await; +} + #[doc(hidden)] pub fn spend_coinbase(cbtxid: bitcoin::Txid, fee: u64) -> Transaction { Transaction { @@ -562,6 +661,7 @@ pub async fn electrum_sh_join() { casa_cap_then_balance_history_unspent(&q); let q = Arc::new(q); + heavy_joins_take_turns(Arc::clone(&q)).await; let hub = MempoolHub::open(dir.join("mempool"), Arc::clone(&q)).unwrap(); hub.set_relay_enabled(true); for (i, cbtxid) in coinbase_txids.iter().enumerate() { diff --git a/crates/rbitcoin-electrum/tests/server/mod.rs b/crates/rbitcoin-electrum/tests/server/mod.rs index 6e3dbc0f8a..0916bdee7b 100644 --- a/crates/rbitcoin-electrum/tests/server/mod.rs +++ b/crates/rbitcoin-electrum/tests/server/mod.rs @@ -460,7 +460,7 @@ fn assert_wallets_read(server: &str) { ); } -async fn electrum_tcp_batch(stream: &mut TcpStream, items: &[Value]) -> Value { +pub(crate) async fn electrum_tcp_batch(stream: &mut TcpStream, items: &[Value]) -> Value { let mut line = serde_json::to_string(items).unwrap(); line.push('\n'); stream.write_all(line.as_bytes()).await.unwrap(); diff --git a/crates/rbitcoin-node/src/cli.rs b/crates/rbitcoin-node/src/cli.rs index 7e619a7114..c1a1e34433 100644 --- a/crates/rbitcoin-node/src/cli.rs +++ b/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] [--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-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\ @@ -376,6 +376,7 @@ Block filters: --block-filter-index (default off) builds BIP158 basic filters. I Kept heights are store/seqsigwit.window/{{height}}.bin plus a RAM cache. Unpruned nodes read seqsigwit.body.\n\ --prune-seqsigwit-ram-threshold-bytes N RAM cap for that cache (default 268435456; 0 keeps nothing in RAM).\n\ --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-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\ diff --git a/crates/rbitcoin-node/src/config.rs b/crates/rbitcoin-node/src/config.rs index c8c16808fb..a3378afdb7 100644 --- a/crates/rbitcoin-node/src/config.rs +++ b/crates/rbitcoin-node/src/config.rs @@ -334,6 +334,9 @@ pub struct NodeConfig { pub sptweaks_dust: u64, /// 0 = unlimited. Electrum + Esplora refuse SH joins above this create count. pub max_sh_creates: u32, + /// 0 = off. Unpaged SH joins above this create count run one at a time + /// across the node. + pub heavy_sh_creates: u32, /// Max `blockchain.scripthash.subscribe` entries per Electrum connection /// (>= 1). Default [`rbitcoin_electrum::DEFAULT_MAX_SCRIPTHASH_SUBS`]. pub electrum_max_subs: usize, @@ -445,6 +448,7 @@ impl Default for NodeConfig { sptweaks: false, sptweaks_dust: rbitcoin_electrum::DEFAULT_TWEAKS_MIN_DUST, 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_block_template: false, metrics: false, @@ -1308,6 +1312,11 @@ impl NodeConfig { .parse() .map_err(|e| NodeError::Config(format!("conf max_sh_creates: {e}")))?; } + "heavy_sh_creates" => { + self.heavy_sh_creates = val + .parse() + .map_err(|e| NodeError::Config(format!("conf heavy_sh_creates: {e}")))?; + } "electrum_max_subs" => { let n: usize = val .parse() diff --git a/crates/rbitcoin-node/src/run.rs b/crates/rbitcoin-node/src/run.rs index 82389e3bca..eb6ffbef72 100644 --- a/crates/rbitcoin-node/src/run.rs +++ b/crates/rbitcoin-node/src/run.rs @@ -1952,6 +1952,7 @@ pub fn apply_startup_index_mode( query.set_sh_index_enabled(config.shindex); query.set_block_filter_index(config.block_filter_index)?; query.set_max_sh_creates(config.max_sh_creates); + query.set_heavy_sh_creates(config.heavy_sh_creates); query.set_seqsigwit_ram_threshold_bytes(config.prune_seqsigwit_ram_threshold_bytes)?; if !config.prune_seqsigwit && query.prune_seqsigwit() { return Err(NodeError::Config( diff --git a/crates/rbitcoin-node/tests/cli.rs b/crates/rbitcoin-node/tests/cli.rs index 037ac943a3..4972d3ee2b 100644 --- a/crates/rbitcoin-node/tests/cli.rs +++ b/crates/rbitcoin-node/tests/cli.rs @@ -542,6 +542,12 @@ fn max_sh_creates_and_esplora_block_template_cli_hyphens() { assert_eq!(n.max_sh_creates, 42); let eq = ready_config(["rbitcoin-node", "--max-sh-creates=9"]); assert_eq!(eq.max_sh_creates, 9); + assert_eq!( + omitted.heavy_sh_creates, + rbitcoin_query::DEFAULT_HEAVY_SH_CREATES + ); + let heavy = ready_config(["rbitcoin-node", "--heavy-sh-creates", "500"]); + assert_eq!(heavy.heavy_sh_creates, 500); let gbt = ready_config(["rbitcoin-node", "--esplora-block-template"]); assert!(gbt.esplora_block_template); let gbt_eq = ready_config(["rbitcoin-node", "--esplora-block-template=1"]); diff --git a/crates/rbitcoin-node/tests/config.rs b/crates/rbitcoin-node/tests/config.rs index 3fb8194cab..13b7d86f52 100644 --- a/crates/rbitcoin-node/tests/config.rs +++ b/crates/rbitcoin-node/tests/config.rs @@ -319,6 +319,13 @@ fn max_sh_creates_and_esplora_block_template_apply_kv() { ); assert_eq!(c.sptweaks_dust, 546); assert_eq!(c.max_sh_creates, 100); + assert_eq!( + c.apply_kv("heavy_sh_creates", "250").unwrap(), + ConfApply::Applied + ); + assert_eq!(c.heavy_sh_creates, 250); + let bad = c.apply_kv("heavy_sh_creates", "nope").unwrap_err(); + assert!(format!("{bad}").contains("heavy_sh_creates"), "{bad}"); let bad = c.apply_kv("max_sh_creates", "nope").unwrap_err(); assert!( format!("{bad}").contains("max_sh_creates"), diff --git a/crates/rbitcoin-query/src/lib.rs b/crates/rbitcoin-query/src/lib.rs index 9bc65f3a54..146c34da3f 100644 --- a/crates/rbitcoin-query/src/lib.rs +++ b/crates/rbitcoin-query/src/lib.rs @@ -143,6 +143,11 @@ pub const SCRIPTHASH_INDEX_DISABLED: &str = "scripthash index disabled"; /// Default unpaged scripthash join cap. `0` stays unlimited. pub const DEFAULT_MAX_SH_CREATES: u32 = 10_000; +/// Default create count above which an unpaged scripthash join waits for +/// every other such join. It matters once `--max-sh-creates` is raised past +/// it. `0` turns the wait off. +pub const DEFAULT_HEAVY_SH_CREATES: u32 = 10_000; + /// Result of [`Query::uring_recover`]. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum UringRecover { @@ -448,6 +453,15 @@ pub struct Query { pub reconstruct_archived: AtomicU64, /// 0 = unlimited. Electrum + Esplora SH joins refuse above this create count. pub max_sh_creates: AtomicU32, + /// 0 = off. An unpaged SH join above this create count holds + /// [`Self::heavy_sh_join`] for its whole walk. + pub heavy_sh_creates: AtomicU32, + /// One heavy unpaged SH join at a time across every Electrum and Esplora + /// client: one near the cap can peak above 2 GiB. + pub heavy_sh_join: Mutex<()>, + /// Unpaged SH join overlap by kind. Tests only. + #[cfg(debug_assertions)] + pub sh_join_probe: crate::scripthash::ShJoinProbe, /// Packed `tx.body` bytes read by [`Self::load_thin_tweaks`]. pub thin_tweak_body_bytes: AtomicU64, /// Max fk `head_insert_many` has published (0 = never). Load polls this @@ -621,6 +635,10 @@ impl Query { height_by_hash: Mutex::new(HeightByHashIndex::default()), reconstruct_archived: AtomicU64::new(0), max_sh_creates: AtomicU32::new(DEFAULT_MAX_SH_CREATES), + heavy_sh_creates: AtomicU32::new(DEFAULT_HEAVY_SH_CREATES), + heavy_sh_join: Mutex::new(()), + #[cfg(debug_assertions)] + sh_join_probe: crate::scripthash::ShJoinProbe::default(), thin_tweak_body_bytes: AtomicU64::new(0), head_drain_fk: AtomicU64::new(head_drained), disconnect_height: AtomicU32::new(0), @@ -1434,6 +1452,14 @@ impl Query { self.max_sh_creates.load(AtomicOrdering::Relaxed) } + pub fn set_heavy_sh_creates(&self, n: u32) { + self.heavy_sh_creates.store(n, AtomicOrdering::Relaxed); + } + + pub fn heavy_sh_creates(&self) -> u32 { + self.heavy_sh_creates.load(AtomicOrdering::Relaxed) + } + pub fn note_reconstruct_archived(&self) { self.reconstruct_archived .fetch_add(1, AtomicOrdering::Relaxed); diff --git a/crates/rbitcoin-query/src/scripthash.rs b/crates/rbitcoin-query/src/scripthash.rs index b1c52c32d5..54cb46e45c 100644 --- a/crates/rbitcoin-query/src/scripthash.rs +++ b/crates/rbitcoin-query/src/scripthash.rs @@ -639,13 +639,13 @@ impl Query { page: Option<&HistoryFilter>, ) -> Result, QueryError> { let paging = page.is_some_and(|f| f.limit.is_some()); - let cap = self.max_sh_creates(); - if cap > 0 && !paging { - let n = self.scripthash_create_count(scripthash)?; - if n > cap { - return Err(StoreError::ShCreatesCap { creates: n, cap }); - } - } + let admit = if paging { + None + } else { + Some(self.unpaged_join_admit(scripthash)?) + }; + #[cfg(debug_assertions)] + let _probe = admit.as_ref().map(|a| self.sh_join_probe.enter(a.heavy)); let t_pages = std::time::Instant::now(); let entries = self.store.scripthash.create_fks(scripthash)?; let pages_us = t_pages.elapsed().as_micros(); @@ -714,9 +714,43 @@ impl Query { spends_us ); } + drop(admit); Ok(out) } + /// Refuse an unpaged join above `--max-sh-creates`, then hand back the + /// heavy-join guard when the create count is above `--heavy-sh-creates`. + /// The guard is a blocking lock taken on the blocking pool. Its holder + /// only reads the store and never waits on a client, so a waiter (even + /// one in the same connection's batch window) always gets it once that + /// join returns, on success or error. + fn unpaged_join_admit(&self, scripthash: &[u8; 32]) -> Result, QueryError> { + let light = JoinAdmit { + heavy: false, + _turn: None, + }; + let cap = self.max_sh_creates(); + let heavy = self.heavy_sh_creates(); + if cap == 0 && heavy == 0 { + return Ok(light); + } + let n = self.scripthash_create_count(scripthash)?; + if cap > 0 && n > cap { + return Err(StoreError::ShCreatesCap { creates: n, cap }); + } + if heavy == 0 || n <= heavy { + return Ok(light); + } + Ok(JoinAdmit { + heavy: true, + _turn: Some( + self.heavy_sh_join + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner), + ), + }) + } + /// A `from_height` above 0 scans that window and never fills the join. /// Any `to_height` also skips the stored join. A limited newest-first /// page never fills it, at every cap. A height-ascending page skips it @@ -2184,3 +2218,82 @@ impl Query { self.chain_stats_from_joined(&recs.joined) } } + +/// An admitted unpaged join. A heavy one holds the node's single heavy turn +/// until this drops. +pub struct JoinAdmit<'a> { + pub heavy: bool, + _turn: Option>, +} + +/// Unpaged scripthash joins running now and their peak, light and heavy +/// apart. Off until [`ShJoinProbe::arm`]. Armed, a join waits inside (up to +/// [`SH_JOIN_PROBE_WAIT`]) for a second join of its kind, so two joins that +/// are allowed to overlap do. +#[cfg(debug_assertions)] +#[derive(Default)] +pub struct ShJoinProbe { + armed: std::sync::atomic::AtomicBool, + running: [AtomicU32; 2], + peak: [AtomicU32; 2], + entered: [AtomicU32; 2], +} + +#[cfg(debug_assertions)] +pub const SH_JOIN_PROBE_WAIT: std::time::Duration = std::time::Duration::from_millis(300); + +#[cfg(debug_assertions)] +pub struct ShJoinProbeGuard<'a> { + probe: &'a ShJoinProbe, + kind: usize, +} + +#[cfg(debug_assertions)] +impl ShJoinProbe { + pub fn arm(&self) { + for k in 0..2 { + self.peak[k].store(0, AtomicOrdering::SeqCst); + self.entered[k].store(0, AtomicOrdering::SeqCst); + } + self.armed.store(true, AtomicOrdering::SeqCst); + } + + /// `(light, heavy)` joins started since [`Self::arm`]. + pub fn entered(&self) -> (u32, u32) { + ( + self.entered[0].load(AtomicOrdering::SeqCst), + self.entered[1].load(AtomicOrdering::SeqCst), + ) + } + + /// `(light, heavy)` peak overlap since [`Self::arm`]. + pub fn peaks(&self) -> (u32, u32) { + ( + self.peak[0].load(AtomicOrdering::SeqCst), + self.peak[1].load(AtomicOrdering::SeqCst), + ) + } + + pub fn enter(&self, heavy: bool) -> Option> { + if !self.armed.load(AtomicOrdering::SeqCst) { + return None; + } + let kind = usize::from(heavy); + self.entered[kind].fetch_add(1, AtomicOrdering::SeqCst); + let now = self.running[kind].fetch_add(1, AtomicOrdering::SeqCst) + 1; + self.peak[kind].fetch_max(now, AtomicOrdering::SeqCst); + let until = std::time::Instant::now() + SH_JOIN_PROBE_WAIT; + while self.peak[kind].load(AtomicOrdering::SeqCst) < 2 && std::time::Instant::now() < until + { + std::thread::sleep(std::time::Duration::from_millis(1)); + } + Some(ShJoinProbeGuard { probe: self, kind }) + } +} + +#[cfg(debug_assertions)] +impl Drop for ShJoinProbeGuard<'_> { + fn drop(&mut self) { + self.probe.running[self.kind].fetch_sub(1, AtomicOrdering::SeqCst); + } +} diff --git a/docs/operator/interfaces.md b/docs/operator/interfaces.md index 75ae060688..e9cd638350 100644 --- a/docs/operator/interfaces.md +++ b/docs/operator/interfaces.md @@ -108,7 +108,8 @@ and idle clients fail closed. | Max scripthash subs / conn | 10000 (`--electrum-max-subs`) | Notify fan-out cap. A wallet subscribes every address up to its gap limit; each sub is ~170 B (≈0.45 GB worst case at 256 connections) plus one posting check per block against that block's shared touch set; a reorg restatuses every sub | | Max broadcast hex | ~8 MiB | `transaction.broadcast` hex length | | Max batch items | 345 | Requests in one JSON-RPC batch line; more is one `-32600`. The reply is written one slot at a time, so a slow reader holds back the server, not RAM | -| Batch reads at once | 4 | Read-only batch items in flight per connection (blocking threads and unwritten replies). The blocking pool, `max(cores, 4)` threads, bounds all connections together. Four scripthash joins at once can each reach `--max-sh-creates` | +| Batch reads at once | 4 | Read-only batch items in flight per connection (blocking threads and unwritten replies). The blocking pool, `max(cores, 4)` threads, bounds all connections together | +| Heavy SH joins at once | 1 (`--heavy-sh-creates`, default 10000) | Node-wide: an unpaged Electrum/Esplora join above N creates waits for any other such join, from any connection or batch. Lighter joins stay parallel. **0** turns it off | Edge rate-limits, auth, and TLS cipher policy stay on the proxy. See [`SECURITY.md`](../../SECURITY.md). diff --git a/docs/operator/operations.md b/docs/operator/operations.md index a6c3407a8f..75042d7bb6 100644 --- a/docs/operator/operations.md +++ b/docs/operator/operations.md @@ -98,6 +98,7 @@ Clean smoke: | `--prune-seqsigwit-ram-threshold-bytes N` | `prune_seqsigwit_ram_threshold_bytes=` | `268435456` (256 MiB). `0` keeps nothing in RAM: every height, including tiny IBD blocks, is read from its file | | `--electrum-max-subs N` | `electrum_max_subs=` | **10000** — `blockchain.scripthash.subscribe` per Electrum connection (>= 1). Wallets subscribe every receive/change address up to their gap limit. | | `--max-sh-creates N` | `max_sh_creates=` | **10000** — unpaged SH join above N is refused (503 / JSON-RPC error). **0** is unlimited. A request that names a page still returns that page. | +| `--heavy-sh-creates N` | `heavy_sh_creates=` | **10000**: an unpaged SH join above N waits until no other such join runs, across every Electrum and Esplora client. **0** turns the wait off. Matters once `--max-sh-creates` is raised above it. | | `--sp-tweaks` | `sp_tweaks=` | **off** — thin BIP-352 tweak index (`sp_tweaks.*`). IBD seals it from the taproot origin when the flag is on from the start. A later enable still materializes after catch-up on `rbtc-idx-wb` (with block filters, one read of Class A). New tips are sealed on the confirm write thread once the watermark has caught the chain. Heights not yet sealed are served by the naive walk. Refused with `--prune-seqsigwit`: tweaks read input keys from scriptSig and witness | | `--sp-tweaks-dust SATS` | `sp_tweaks_dust=` | **1000** — omit served P2TR outs with `value <= SATS` (`0` = serve all; **546** matches Cake electrs) | | `--electrum-listen [ADDR]` | `electrum_listen=` | disabled; omit ADDR → `127.0.0.1:50001`. Address/scripthash methods need `--sh-index` | diff --git a/docs/operator/storage.md b/docs/operator/storage.md index 10b2652364..a2b334e1b5 100644 --- a/docs/operator/storage.md +++ b/docs/operator/storage.md @@ -155,6 +155,12 @@ before Class A expand: Esplora **503** / Electrum JSON-RPC error A history request that names a page (Esplora's 25, or any caller that sets a limit) is still served: the join stops once that page is full. +`--heavy-sh-creates N` (conf `heavy_sh_creates=`, default **10000**, **0** off) +lets one unpaged join with more than N creates run at a time across the +node. One join near a raised cap of 100000 was measured at 2.1 GiB peak RSS, +so several at once can push a small box into swap. A waiting join holds a +blocking-pool thread; lighter joins and paged history do not wait. + Order-of-magnitude costs (mainnet-class SSD; not a warranty): - **During IBD with sh_index=1:** modest extra work (run stream); after IBD, bulk materialize is typically **tens of minutes to a few hours**. From ccdb57a3868ef03fdb1c194e6b6a3961dfa107e5 Mon Sep 17 00:00:00 2001 From: Bob Bobber Date: Sat, 24 Oct 2026 02:49:18 -0400 Subject: [PATCH 3/4] electrum: size the batch window from the blocking pool, time reads alone The per-connection batch window was a fixed 4, the smallest blocking pool the node builds, so one batch could take every blocking thread that block decode, compact reconstruct, header jobs and mempool accept also need. The window is now ElectrumConfig::batch_window, half the pool and at least one (batch_window), set by the node from blocking_pool_size. A 4-thread pool gives 2; 12 gives 6. A batch read's wall time started when the item was classified and stopped when it landed, so it counted the window wait and earlier slot writes. The blocking closure now times the dispatch and returns it, and that is what the API log and the Electrum meter see. A single request still counts from when it is taken up, as before. A read that never uses the join cache no longer puts the pre-window copy back over a join an earlier read built. A scripthash read now takes the cache instead of sharing an Arc clone, so a hit fills it in place without a deep copy; land installs a cache only when the read hands one back. Tests that use the debug-only trace or probe are gated the same way. Co-Authored-By: Claude Opus 5.5 --- COMPAT.md | 2 +- changelog.d/electrum-batch-concurrency.md | 14 +- crates/rbitcoin-electrum/src/lib.rs | 4 +- crates/rbitcoin-electrum/src/server.rs | 150 ++++++++++++++---- .../tests/electrum_sh_journey.rs | 86 +++++++++- crates/rbitcoin-electrum/tests/server/mod.rs | 55 +++++-- crates/rbitcoin-node/src/cli.rs | 4 +- crates/rbitcoin-node/src/run.rs | 1 + docs/operator/interfaces.md | 2 +- 9 files changed, 259 insertions(+), 59 deletions(-) diff --git a/COMPAT.md b/COMPAT.md index ad2d544682..a565556737 100644 --- a/COMPAT.md +++ b/COMPAT.md @@ -138,7 +138,7 @@ Per-method notes, auth, and the shindex matrix live in | Method | Status | Notes | |--------|--------|-------| | server.version / banner / features | done | Banner: libre-relay-class. `server.version[0]` is `rbitcoin-electrs electrumx` — **not electrs**; see below. `server.version` negotiates: omitted → `1.6`; `"1.4"` → `1.4`; `["1.4","1.4.2"]` → `1.4.2`; `["1.4","1.6"]` → `1.6`; `"1.4.2-asof"` (or a range containing it) → as-of dialect. First call wins. `features.protocol_max` is `1.6`; `features.asof_protocol` is `1.4.2-asof`. `features.genesis_hash` is display-order hex (refuse a wrong-chain server before a tweaks scan). `features.tweaks` / `silent_payments` advertise the method; they are not a substitute for the stream. | -| JSON-RPC batch | done | A line holding a JSON array is a JSON-RPC 2.0 batch: one array line back, same order, each slot `{id, result}` or `{id, error}`. Read-only items (scripthash `get_history` / `get_balance` / `listunspent` / `get_mempool`, `transaction.get` / `get_merkle` / `id_from_pos`, `block.header(s)`, `estimatefee`, `outpoint.get_status`, `mempool.get_fee_histogram` / `get_info`) run up to 4 at a time; any other item waits for every earlier slot, then runs alone, so a `subscribe` is seen by the items after it. Slots are written in request order as soon as each is answered, and slot `i` is written before read `i + 4` starts. A notification (no `id` member) is skipped: it is not run and gets no slot, and a batch of only notifications writes nothing. A non-object item gets a `-32600` `Invalid Request` slot. `[]` is one `-32600` `empty batch`; more than 345 items is one `-32600` `batch too large`. `blockchain.tweaks.subscribe` and `blockchain.silentpayments.subscribe` stream several lines, so inside a batch they get a `-32600` slot; send them alone. | +| JSON-RPC batch | done | A line holding a JSON array is a JSON-RPC 2.0 batch: one array line back, same order, each slot `{id, result}` or `{id, error}`. Read-only items (scripthash `get_history` / `get_balance` / `listunspent` / `get_mempool`, `transaction.get` / `get_merkle` / `id_from_pos`, `block.header(s)`, `estimatefee`, `outpoint.get_status`, `mempool.get_fee_histogram` / `get_info`) run up to half the node's blocking pool at a time (at least 1; 2 on a 4-thread pool), and each reports only its own dispatch time; any other item waits for every earlier slot, then runs alone, so a `subscribe` is seen by the items after it. Slots are written in request order as soon as each is answered, and slot `i` is written before read `i + window` starts. A notification (no `id` member) is skipped: it is not run and gets no slot, and a batch of only notifications writes nothing. A non-object item gets a `-32600` `Invalid Request` slot. `[]` is one `-32600` `empty batch`; more than 345 items is one `-32600` `batch too large`. `blockchain.tweaks.subscribe` and `blockchain.silentpayments.subscribe` stream several lines, so inside a batch they get a `-32600` slot; send them alone. | | blockchain.tweaks.subscribe | done | **Stream**, not a one-shot result: JSON-RPC `result` is the **first** height only; remaining heights are unsolicited notifications; `{"message":"done"}` ends the **chunk** (60s wall at a wave boundary, or the requested `count` if sooner) — Cake electrs clamps `count` to 1000 instead. Cake Wallet resubscribes from `syncHeight+1` ([cake_wallet#3574](https://github.com/cake-tech/cake_wallet/issues/3574) persist still lags one event). [kiss-bdk](https://github.com/kkdao/kiss-bdk) today treats `done` as the asked range finished ([kiss-bdk#10](https://github.com/kkdao/kiss-bdk/issues/10)). Each height carries tweak + txid + taproot `output_pubkeys` (client scans locally; no block fetch). Naive walk, or `--sp-tweaks` thin index (`len:tweak` only; one `txout` span per wave; the first height shares that span when indexed). Pre-taproot: **one** notify with ≤1024 empty height keys (Cake `fromJson` last key is progress); probe `[0,1,false]` stays `{"0": {}}`. Param `[2]` (Cake `historicalMode`): `false` **cut-through** (omit confirmed-spent P2TR outs / txs); `true` keeps spent outs. Serve-time `--sp-tweaks-dust` (default **1000**) omits P2TR outs with `value <=` the floor (`0` = all; **546** matches Cake electrs `sp_min_dust`). Sparrow Silent Payments uses Frigate `blockchain.silentpayments.subscribe` (server-side scan) — **not** this method. Cake isolate may still hardcode `electrs.cakewallet.com`. | | headers / block headers | done | Tip push on subscribe | | scripthash history / balance / listunspent | done | Unconf when mempool attached; `get_history` optional BCH-style `from_height` / exclusive `to_height` (`-1` = tip + mempool); 1-arg = full history; a `from_height` above 0 scans that block window and is served above `--max-sh-creates` (open history, and a window that starts at height 0, still refuse above the cap); unconfirmed `get_history` rows include `fee` (same as `get_mempool`); `listunspent` mempool height is `0` or `-1` (unconfirmed parent); **subscribe status always full**; `blockchain.scripthash.unsubscribe` returns whether the connection was watching (frees a slot under `--electrum-max-subs`, default 10000 per connection). `listunspent` loads `txid.body` only for unspent creates; one TCP connection reuses the last SH outs+spent join until SH-view **hash** changes. Confirmed methods stamp live tip: a RAM SH head (pending write-behind) joins with durable SH so mempool can drop confirmed txs without a hole. Durable Class B seed waits until after tip announce. `get_history` skips mempool rows already in the confirmed list. `server.features.chain_tip = true`. Trailing **`asof:`** after the official args (`server.features.asof` / `asof_protocol = 1.4.2-asof`): confirmed rows as of that still-live ancestor **at or behind visible SH** (durable + pending), **no** mempool; stamp is the asof block; unknown hash or ahead of visible SH → `asof not on chain`. Prefix keeps it off the future positional-string landmine. Requires negotiated `1.4.2-asof` (first `server.version` only). Electrum `protocol_max` is **1.6**. | diff --git a/changelog.d/electrum-batch-concurrency.md b/changelog.d/electrum-batch-concurrency.md index 42550c214a..044f0f650c 100644 --- a/changelog.d/electrum-batch-concurrency.md +++ b/changelog.d/electrum-batch-concurrency.md @@ -1,14 +1,16 @@ Changed -- **Electrum batches answer read-only items four at a time.** A batch +- **Electrum batches answer read-only items concurrently.** A batch line ran its items one after another, so a 100-address `get_history` page cost the sum of 100 lookups. Read-only items (scripthash history, balance, unspent and mempool, transaction and header reads, fee - estimates) now run up to four at once per connection and still land in - request order. A subscribe, unsubscribe or other stateful item waits for - the reads before it and runs alone. Slot `i` is written before read - `i + 4` starts, so a client that stops reading still stops the work - (#1174). + estimates) now run up to half the node's blocking pool at once per + connection (2 on a 4-thread pool), so block decode and mempool accept + keep the other half, and still land in request order. Each reports its + own dispatch time to the API log and metrics. A subscribe, unsubscribe + or other stateful item waits for the reads before it and runs alone. + Slot `i` is written before read `i + window` starts, so a client that + stops reading still stops the work (#1174). - **One heavy scripthash join at a time.** An unpaged Electrum or Esplora join with more than `--heavy-sh-creates` creates (conf `heavy_sh_creates=`, default 10000, `0` off) now waits until no other diff --git a/crates/rbitcoin-electrum/src/lib.rs b/crates/rbitcoin-electrum/src/lib.rs index 562bd0315e..65a14c05f0 100644 --- a/crates/rbitcoin-electrum/src/lib.rs +++ b/crates/rbitcoin-electrum/src/lib.rs @@ -22,9 +22,9 @@ pub mod tweaks; pub mod unspent; pub use server::{ - electrum_scripthash_hex, parse_electrum_request_line, perf_totals, run_electrum, + batch_window, electrum_scripthash_hex, parse_electrum_request_line, perf_totals, run_electrum, sample_reset_perf, ElectrumConfig, ElectrumHandle, ServeLimits, TipNotify, - DEFAULT_MAX_SCRIPTHASH_SUBS, + DEFAULT_MAX_SCRIPTHASH_SUBS, MIN_BLOCKING_POOL, }; pub use tweaks::DEFAULT_TWEAKS_MIN_DUST; pub use unspent::{ diff --git a/crates/rbitcoin-electrum/src/server.rs b/crates/rbitcoin-electrum/src/server.rs index 10abdf4f48..c66b8ff9a1 100644 --- a/crates/rbitcoin-electrum/src/server.rs +++ b/crates/rbitcoin-electrum/src/server.rs @@ -144,6 +144,9 @@ pub struct ElectrumConfig { pub tweaks_min_dust: u64, /// Tor v3 hostname + TCP port for `server.features.hosts` (empty until set). pub onion_tcp: Arc>, + /// Read-only batch items one connection may have in flight + /// ([`batch_window`]). The node sets it from its blocking pool. + pub batch_window: usize, } impl ElectrumConfig { @@ -160,6 +163,7 @@ impl ElectrumConfig { max_broadcast_hex: DEFAULT_MAX_BROADCAST_HEX, tweaks_min_dust: crate::tweaks::DEFAULT_TWEAKS_MIN_DUST, onion_tcp: Arc::new(OnceLock::new()), + batch_window: batch_window(MIN_BLOCKING_POOL), } } @@ -677,13 +681,17 @@ where Ok(()) } -/// Read-only batch items answered at once on one connection, counted from -/// spawn until the slot is written. Each holds a blocking thread while it -/// runs and its reply until written. The shared blocking pool (`max(cores, -/// 4)` threads) bounds all connections together. On a 4-core node, splitting -/// a 100-item `get_history` page across connections stopped gaining at 4. -#[doc(hidden)] -pub const BATCH_CONCURRENCY: usize = 4; +/// Smallest blocking pool the node builds (`max(cores, 4)` threads). +pub const MIN_BLOCKING_POOL: usize = 4; + +/// Read-only batch items one connection may have in flight for a blocking +/// pool of `blocking_pool` threads: half of it, at least one. Each item holds +/// a thread while it runs and its reply until written. Block decode, compact +/// reconstruct, header jobs and mempool accept share that pool +/// (`docs/concurrency.md`), so one batch leaves them the other half. +pub fn batch_window(blocking_pool: usize) -> usize { + (blocking_pool / 2).max(1) +} /// One scheduling step of a batch, by item index. Recorded only for tests. #[doc(hidden)] @@ -692,6 +700,8 @@ pub enum BatchStep { Spawn(usize), Serial(usize), Write(usize), + /// Item and the wall time it reports to the API log and the meter. + Timed(usize, Duration), } #[cfg(debug_assertions)] @@ -781,12 +791,11 @@ pub fn method_reads_only(method: &str) -> bool { ) } -/// One request as asked, and when it was taken up. +/// One request as asked. struct Asked { id: Value, method: String, params: Value, - t0: Instant, } impl Asked { @@ -799,7 +808,6 @@ impl Asked { .unwrap_or("") .to_string(), params: req.get("params").cloned().unwrap_or(json!([])), - t0: Instant::now(), } } } @@ -863,7 +871,14 @@ impl BatchOut<'_, W> { } } -type Dispatched = (Result, ElectrumConn, Option); +/// A dispatch result, the worker state it ran with, its chain view, and how +/// long the dispatch itself took on the blocking thread. +type Dispatched = ( + Result, + ElectrumConn, + Option, + Duration, +); /// A read running on the blocking pool, not yet written. struct InFlight { @@ -873,8 +888,8 @@ struct InFlight { } /// JSON-RPC 2.0 batch, one array line on the wire, slots in request order. -/// Read-only items run up to [`BATCH_CONCURRENCY`] at once; slot `i` is -/// written before read `i + BATCH_CONCURRENCY` starts, so a client that +/// Read-only items run up to [`ElectrumConfig::batch_window`] at once; slot +/// `i` is written before read `i + window` starts, so a client that /// stops reading stops the work and the connection holds at most that many /// unwritten replies. Any other item waits until every earlier slot is /// written, then runs alone, so later items see what it did. A batch of @@ -903,18 +918,19 @@ async fn answer_batch( idle, opened: false, }; - let mut window: VecDeque = VecDeque::with_capacity(BATCH_CONCURRENCY); + let limit = ctx.config.batch_window.max(1); + let mut window: VecDeque = VecDeque::with_capacity(limit); for (index, item) in items.iter().enumerate() { let answer = match batch_item(item) { BatchItem::Skip => continue, BatchItem::Read(asked) => { - if window.len() >= BATCH_CONCURRENCY { + if window.len() >= limit { if let Some(oldest) = window.pop_front() { land(ctx, conn, last_sent, &mut out, oldest).await?; } } batch_mark(BatchStep::Spawn(index)); - let task = spawn_dispatch(ctx, read_conn(conn), &asked); + let task = spawn_dispatch(ctx, read_conn(conn, &asked.method), &asked); window.push_back(InFlight { index, asked, task }); continue; } @@ -938,22 +954,41 @@ async fn answer_batch( out.close().await } -/// The connection state a read needs: the negotiated protocol and a copy of -/// the join cache. Subscription sets stay with the connection. -fn read_conn(conn: &ElectrumConn) -> ElectrumConn { +/// Methods that read and fill the connection's scripthash join cache. +fn method_uses_sh_join(method: &str) -> bool { + matches!( + method, + "blockchain.scripthash.get_history" + | "blockchain.scripthash.get_balance" + | "blockchain.scripthash.listunspent" + ) +} + +/// The connection state a read needs: the negotiated protocol, and for a +/// scripthash read the join cache itself, moved out so the read can fill +/// it in place. A scripthash read that starts while another holds the +/// cache gets none and joins afresh. Subscription sets stay with the +/// connection. +fn read_conn(conn: &mut ElectrumConn, method: &str) -> ElectrumConn { ElectrumConn { protocol: conn.protocol.clone(), header_sub: conn.header_sub, sh_subs: HashSet::new(), - sh_join: conn.sh_join.clone(), + sh_join: if method_uses_sh_join(method) { + conn.sh_join.take() + } else { + None + }, outpoint_subs: HashSet::new(), sp_sub: None, sp_scan_busy: conn.sp_scan_busy, } } -/// Wait for the oldest read and write its slot. Its join cache becomes the -/// connection's, as if the reads had run one after another. +/// Wait for the oldest read and write its slot. A join cache the read hands +/// back (the one it took, or one it built) becomes the connection's, so in +/// request order the last scripthash read's join is kept, as when reads run +/// one after another. A read that never held one leaves the cache alone. async fn land( ctx: &RequestCtx<'_>, conn: &mut ElectrumConn, @@ -961,12 +996,40 @@ async fn land( out: &mut BatchOut<'_, W>, flight: InFlight, ) -> Result<(), std::io::Error> { - let (result, work, stamped) = flight.task.await.map_err(dispatch_join_error)?; - conn.sh_join = work.sh_join; - let answer = finish_reply(ctx, conn, last_sent, flight.asked, result, stamped); + let (result, work, stamped, ran) = flight.task.await.map_err(dispatch_join_error)?; + if work.sh_join.is_some() { + conn.sh_join = work.sh_join; + } + batch_mark(BatchStep::Timed(flight.index, ran)); + let answer = finish_reply( + ctx, + conn, + last_sent, + flight.asked, + result, + stamped, + Took::Ran(ran), + ); out.slot(flight.index, &answer).await } +/// The wall time a reply reports. A single request counts from when it was +/// taken up; a batch read counts only its own dispatch, not the window wait +/// or the writes of earlier slots. +enum Took { + Since(Instant), + Ran(Duration), +} + +impl Took { + fn elapsed(&self) -> Duration { + match self { + Took::Since(t0) => t0.elapsed(), + Took::Ran(d) => *d, + } + } +} + fn dispatch_join_error(e: tokio::task::JoinError) -> std::io::Error { std::io::Error::other(format!("electrum dispatch join: {e}")) } @@ -984,7 +1047,10 @@ fn spawn_dispatch( let params = asked.params.clone(); tokio::task::spawn_blocking(move || { let _g = BlockingRegion::enter(); - dispatch_off_worker(&q, &cfg, &p, mp.as_deref(), &method, ¶ms, work) + let t0 = Instant::now(); + let (r, work, view) = + dispatch_off_worker(&q, &cfg, &p, mp.as_deref(), &method, ¶ms, work); + (r, work, view, t0.elapsed()) }) } @@ -997,7 +1063,7 @@ fn dispatch_off_worker( method: &str, params: &Value, mut work: ElectrumConn, -) -> Dispatched { +) -> (Result, ElectrumConn, Option) { if !method_stamps_chain_tip(method) { let r = dispatch_pinned(method, params, q, cfg, p, mp, &mut work, None, false); return (r, work, None); @@ -1019,6 +1085,7 @@ async fn answer_one( req: &Value, ) -> Result { let asked = Asked::of(req); + let t0 = Instant::now(); if method_stays_on_worker(&asked.method) { let result = dispatch_with_join( &asked.method, @@ -1029,7 +1096,15 @@ async fn answer_one( ctx.mempool.as_deref(), conn, ); - return Ok(finish_reply(ctx, conn, last_sent, asked, result, None)); + return Ok(finish_reply( + ctx, + conn, + last_sent, + asked, + result, + None, + Took::Since(t0), + )); } // Move the sub sets in and back (`*conn = work` below): a clone per // request made a wallet's N-address sync O(N²). A JoinError ends the @@ -1043,14 +1118,23 @@ async fn answer_one( sp_sub: conn.sp_sub.clone(), sp_scan_busy: conn.sp_scan_busy, }; - let (result, work, stamped) = spawn_dispatch(ctx, work, &asked) + let (result, work, stamped, _ran) = spawn_dispatch(ctx, work, &asked) .await .map_err(dispatch_join_error)?; *conn = work; - Ok(finish_reply(ctx, conn, last_sent, asked, result, stamped)) + Ok(finish_reply( + ctx, + conn, + last_sent, + asked, + result, + stamped, + Took::Since(t0), + )) } /// Status bookkeeping, metering, the API log line, and the reply object. +#[allow(clippy::too_many_arguments)] // call-site args stay unbundled fn finish_reply( ctx: &RequestCtx<'_>, conn: &ElectrumConn, @@ -1058,17 +1142,17 @@ fn finish_reply( asked: Asked, result: Result, stamped: Option, + took: Took, ) -> Value { let Asked { id, method, params: params_v, - t0, } = asked; let method = method.as_str(); drop_unsubscribed_status(last_sent, &conn.sh_subs); - let wall_ms = t0.elapsed().as_millis() as u64; - meter_dispatch_wall(t0.elapsed().as_micros() as u64); + let wall_ms = took.elapsed().as_millis() as u64; + meter_dispatch_wall(took.elapsed().as_micros() as u64); let params_s = serde_json::to_string(¶ms_v).unwrap_or_else(|_| "[]".into()); match result { Ok(v) => { diff --git a/crates/rbitcoin-electrum/tests/electrum_sh_journey.rs b/crates/rbitcoin-electrum/tests/electrum_sh_journey.rs index f52c3b5340..c6021bc6ff 100644 --- a/crates/rbitcoin-electrum/tests/electrum_sh_journey.rs +++ b/crates/rbitcoin-electrum/tests/electrum_sh_journey.rs @@ -18,7 +18,9 @@ use tokio::io::{AsyncBufReadExt, BufReader}; use tokio::net::TcpStream; use tokio::sync::broadcast; -use crate::server::{electrum_tcp_batch, electrum_tcp_rpc}; +#[cfg(debug_assertions)] +use crate::server::electrum_tcp_batch; +use crate::server::electrum_tcp_rpc; fn tmp_store() -> (rbitcoin_query::testutil::TempDir, Query) { rbitcoin_query::testutil::tiny_query_labeled("electrum-sh") @@ -216,9 +218,11 @@ pub fn casa_cap_then_balance_history_unspent(q: &Query) { /// joins overlap. The armed probe makes each join wait inside for a second /// of its kind, so the peaks only stay at 1 because of the lock. A refused /// join takes no turn, and a turn is given back after every join. +#[cfg(debug_assertions)] pub async fn heavy_joins_take_turns(q: Arc) { let params = ChainParams::regtest(); - let cfg = ElectrumConfig::for_params("127.0.0.1:0".parse().unwrap(), ¶ms); + let mut cfg = ElectrumConfig::for_params("127.0.0.1:0".parse().unwrap(), ¶ms); + cfg.batch_window = 4; let (tip_tx, _) = broadcast::channel(4); let handle = run_electrum(cfg, Arc::clone(&q), params, tip_tx, None) .await @@ -310,6 +314,78 @@ pub async fn heavy_joins_take_turns(q: Arc) { handle.shutdown().await; } +/// A batch read reports its own dispatch time, not the wait behind a slow +/// neighbour, and a read that never used the join cache leaves the one an +/// earlier `get_history` built. The probe holds a lone heavy join for +/// `SH_JOIN_PROBE_WAIT`, which makes the slow neighbour. +#[cfg(debug_assertions)] +pub async fn batch_reads_time_themselves_and_keep_the_join(q: Arc) { + use rbitcoin_query::scripthash::SH_JOIN_PROBE_WAIT; + let params = ChainParams::regtest(); + let mut cfg = ElectrumConfig::for_params("127.0.0.1:0".parse().unwrap(), ¶ms); + cfg.batch_window = 2; + let (tip_tx, _) = broadcast::channel(4); + let handle = run_electrum(cfg, Arc::clone(&q), params, tip_tx, None) + .await + .expect("listen"); + let sh = electrum_scripthash_hex(&[0x51]); + let call = |id: u32, method: &str, params: Value| json!({"jsonrpc":"2.0","id":id,"method":method,"params":params}); + + q.set_heavy_sh_creates(1); + q.sh_join_probe.arm(); + let mut a = TcpStream::connect(handle.local_addr).await.unwrap(); + batch_trace_arm(); + electrum_tcp_batch( + &mut a, + &[ + call(1, "blockchain.scripthash.get_balance", json!([sh])), + call(2, "blockchain.block.header", json!([0])), + ], + ) + .await; + let trace = batch_trace_snapshot(); + batch_trace_clear(); + let took = |i: usize| { + trace + .iter() + .find_map(|s| match s { + BatchStep::Timed(j, d) if *j == i => Some(*d), + _ => None, + }) + .unwrap_or_else(|| panic!("no time for {i}: {trace:?}")) + }; + assert!(took(0) >= SH_JOIN_PROBE_WAIT, "{trace:?}"); + assert!( + took(1) < SH_JOIN_PROBE_WAIT / 2, + "the header read reports its own time, not the join ahead of it: {trace:?}" + ); + + q.set_heavy_sh_creates(rbitcoin_query::DEFAULT_HEAVY_SH_CREATES); + q.sh_join_probe.arm(); + let mut b = TcpStream::connect(handle.local_addr).await.unwrap(); + let missing = format!("{:064x}", 0x77); + electrum_tcp_batch( + &mut b, + &[ + call(1, "blockchain.scripthash.get_history", json!([sh])), + call(2, "blockchain.transaction.get", json!([missing])), + call(3, "blockchain.block.header", json!([0])), + ], + ) + .await; + let again = electrum_tcp_rpc(&mut b, 4, "blockchain.scripthash.get_history", json!([sh])).await; + assert!( + again["result"].as_array().is_some_and(|r| !r.is_empty()), + "{again}" + ); + assert_eq!( + q.sh_join_probe.entered(), + (1, 0), + "the later get_history reuses the batch's join" + ); + handle.shutdown().await; +} + #[doc(hidden)] pub fn spend_coinbase(cbtxid: bitcoin::Txid, fee: u64) -> Transaction { Transaction { @@ -661,7 +737,11 @@ pub async fn electrum_sh_join() { casa_cap_then_balance_history_unspent(&q); let q = Arc::new(q); - heavy_joins_take_turns(Arc::clone(&q)).await; + #[cfg(debug_assertions)] + { + heavy_joins_take_turns(Arc::clone(&q)).await; + batch_reads_time_themselves_and_keep_the_join(Arc::clone(&q)).await; + } let hub = MempoolHub::open(dir.join("mempool"), Arc::clone(&q)).unwrap(); hub.set_relay_enabled(true); for (i, cbtxid) in coinbase_txids.iter().enumerate() { diff --git a/crates/rbitcoin-electrum/tests/server/mod.rs b/crates/rbitcoin-electrum/tests/server/mod.rs index 0916bdee7b..f7a3bebc54 100644 --- a/crates/rbitcoin-electrum/tests/server/mod.rs +++ b/crates/rbitcoin-electrum/tests/server/mod.rs @@ -633,11 +633,13 @@ async fn batch_line_answers_every_item_in_one_array() { let _ = std::fs::remove_dir_all(&dir); } +#[cfg(debug_assertions)] fn get_balance(id: usize) -> Value { json!({"jsonrpc":"2.0","id":id,"method":"blockchain.scripthash.get_balance", "params":[format!("{id:064x}")]}) } +#[cfg(debug_assertions)] fn spawned(trace: &[BatchStep]) -> Vec { trace .iter() @@ -648,6 +650,7 @@ fn spawned(trace: &[BatchStep]) -> Vec { .collect() } +#[cfg(debug_assertions)] fn written(trace: &[BatchStep]) -> Vec { trace .iter() @@ -658,6 +661,7 @@ fn written(trace: &[BatchStep]) -> Vec { .collect() } +#[cfg(debug_assertions)] fn at(trace: &[BatchStep], step: BatchStep) -> usize { trace .iter() @@ -665,17 +669,36 @@ fn at(trace: &[BatchStep], step: BatchStep) -> usize { .unwrap_or_else(|| panic!("{step:?} missing from {trace:?}")) } -/// Read-only batch items run up to `BATCH_CONCURRENCY` at a time and land in -/// request order. A subscribe waits for every earlier read to be written, +#[test] +fn batch_window_is_half_the_blocking_pool() { + assert_eq!(batch_window(1), 1); + assert_eq!(batch_window(MIN_BLOCKING_POOL), 2); + assert_eq!(batch_window(12), 6); + let params = ChainParams::regtest(); + let cfg = ElectrumConfig::for_params("127.0.0.1:0".parse().unwrap(), ¶ms); + assert_eq!(cfg.batch_window, 2, "the smallest pool the node builds"); +} + +/// Read-only batch items run up to the configured window at a time and land +/// in request order. A subscribe waits for every earlier read to be written, /// runs alone, and later items see what it did. The connection task runs on /// this thread (current-thread runtime), so the armed trace sees its steps. -#[allow(clippy::cognitive_complexity)] // one batch, every scheduling rule +#[cfg(debug_assertions)] #[tokio::test] async fn batch_reads_overlap_and_land_in_request_order() { + for window in [1, 2, 4] { + reads_overlap_in_window(window).await; + } +} + +#[cfg(debug_assertions)] +#[allow(clippy::cognitive_complexity)] // one batch, every scheduling rule +async fn reads_overlap_in_window(window: usize) { let (dir, q) = tmp_store(); let params = ChainParams::regtest(); let (tip_tx, _) = broadcast::channel(4); - let cfg = ElectrumConfig::for_params("127.0.0.1:0".parse().unwrap(), ¶ms); + let mut cfg = ElectrumConfig::for_params("127.0.0.1:0".parse().unwrap(), ¶ms); + cfg.batch_window = window; let handle = run_electrum(cfg, Arc::new(q), params, tip_tx, None) .await .expect("listen"); @@ -723,10 +746,11 @@ async fn batch_reads_overlap_and_land_in_request_order() { } peak = peak.max(open); } - assert_eq!(peak, BATCH_CONCURRENCY, "{trace:?}"); - assert!( + assert_eq!(peak, window, "{trace:?}"); + assert_eq!( at(&trace, BatchStep::Spawn(1)) < at(&trace, BatchStep::Write(0)), - "reads overlap: {trace:?}" + window > 1, + "reads overlap only in a window above one: {trace:?}" ); for barrier in [6, 10] { assert!( @@ -744,9 +768,17 @@ async fn batch_reads_overlap_and_land_in_request_order() { } /// A client that stops reading stops the batch: the server writes slot i -/// before it starts read i + `BATCH_CONCURRENCY`. +/// before it starts read i + window. +#[cfg(debug_assertions)] #[tokio::test] async fn stalled_batch_reader_holds_reads_to_the_window() { + for window in [1, 3] { + stalled_reader_in_window(window).await; + } +} + +#[cfg(debug_assertions)] +async fn stalled_reader_in_window(window: usize) { use std::pin::Pin; use std::task::{Context, Poll}; use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; @@ -822,7 +854,8 @@ async fn stalled_batch_reader_holds_reads_to_the_window() { let (dir, q) = tmp_store(); let params = ChainParams::regtest(); - let cfg = ElectrumConfig::for_params("127.0.0.1:0".parse().unwrap(), ¶ms); + let mut cfg = ElectrumConfig::for_params("127.0.0.1:0".parse().unwrap(), ¶ms); + cfg.batch_window = window; let items: Vec = (0..20).map(get_balance).collect(); let mut line = serde_json::to_vec(&items).unwrap(); line.push(b'\n'); @@ -859,8 +892,8 @@ async fn stalled_batch_reader_holds_reads_to_the_window() { assert_eq!(written(&trace), [0, 1], "{trace:?}"); assert_eq!( spawned(&trace), - (0..=BATCH_CONCURRENCY).collect::>(), - "slot 0 is written before read {BATCH_CONCURRENCY} starts, and slot 1 blocks the rest: {trace:?}" + (0..=window).collect::>(), + "slot 0 is written before read {window} starts, and slot 1 blocks the rest: {trace:?}" ); let _ = std::fs::remove_dir_all(&dir); } diff --git a/crates/rbitcoin-node/src/cli.rs b/crates/rbitcoin-node/src/cli.rs index c1a1e34433..fbccc8e494 100644 --- a/crates/rbitcoin-node/src/cli.rs +++ b/crates/rbitcoin-node/src/cli.rs @@ -572,8 +572,8 @@ pub fn cli_apply_err(e: crate::error::NodeError) -> ExitCode { pub fn blocking_pool_size() -> usize { std::thread::available_parallelism() .map(|p| p.get()) - .unwrap_or(4) - .max(4) + .unwrap_or(rbitcoin_electrum::MIN_BLOCKING_POOL) + .max(rbitcoin_electrum::MIN_BLOCKING_POOL) } #[doc(hidden)] diff --git a/crates/rbitcoin-node/src/run.rs b/crates/rbitcoin-node/src/run.rs index eb6ffbef72..a9da2ed2e3 100644 --- a/crates/rbitcoin-node/src/run.rs +++ b/crates/rbitcoin-node/src/run.rs @@ -2309,6 +2309,7 @@ pub async fn start_electrum_if_ready( ecfg.max_scripthash_subs = electrum_max_subs; ecfg.tweaks_min_dust = tweaks_min_dust; ecfg.onion_tcp = Arc::clone(&onion_tcp); + ecfg.batch_window = rbitcoin_electrum::batch_window(crate::cli::blocking_pool_size()); let max_conn = ecfg.limits.max_connections; let max_line = ecfg.limits.max_request_bytes; let idle_secs = ecfg.limits.idle_timeout.as_secs(); diff --git a/docs/operator/interfaces.md b/docs/operator/interfaces.md index e9cd638350..7d9967f821 100644 --- a/docs/operator/interfaces.md +++ b/docs/operator/interfaces.md @@ -108,7 +108,7 @@ and idle clients fail closed. | Max scripthash subs / conn | 10000 (`--electrum-max-subs`) | Notify fan-out cap. A wallet subscribes every address up to its gap limit; each sub is ~170 B (≈0.45 GB worst case at 256 connections) plus one posting check per block against that block's shared touch set; a reorg restatuses every sub | | Max broadcast hex | ~8 MiB | `transaction.broadcast` hex length | | Max batch items | 345 | Requests in one JSON-RPC batch line; more is one `-32600`. The reply is written one slot at a time, so a slow reader holds back the server, not RAM | -| Batch reads at once | 4 | Read-only batch items in flight per connection (blocking threads and unwritten replies). The blocking pool, `max(cores, 4)` threads, bounds all connections together | +| Batch reads at once | half the blocking pool, at least 1 (2 at 4 threads) | Read-only batch items in flight per connection (blocking threads and unwritten replies). The pool is `max(cores, 4)` threads and also runs block decode, compact reconstruct, header jobs and mempool accept, so one batch leaves them the other half | | Heavy SH joins at once | 1 (`--heavy-sh-creates`, default 10000) | Node-wide: an unpaged Electrum/Esplora join above N creates waits for any other such join, from any connection or batch. Lighter joins stay parallel. **0** turns it off | Edge rate-limits, auth, and TLS cipher policy stay on the proxy. See From 34e94554a61ae22ea89f7e2c8ace82da4f105563 Mon Sep 17 00:00:00 2001 From: "rearden-grok[bot]" <317016512+rearden-grok[bot]@users.noreply.github.com> Date: Sat, 24 Oct 2026 12:02:29 -0700 Subject: [PATCH 4/4] nixos: pass --heavy-sh-creates from heavyShCreates Null omits the flag and leaves the node default. A set value, including 0, is passed for both Electrum and Esplora joins. --- changelog.d/electrum-batch-concurrency.md | 9 +++++---- docs/operator/operations.md | 2 +- nix/modules/rbitcoin.nix | 14 ++++++++++++++ nix/tests/nixos-module-eval.nix | 8 ++++++++ 4 files changed, 28 insertions(+), 5 deletions(-) diff --git a/changelog.d/electrum-batch-concurrency.md b/changelog.d/electrum-batch-concurrency.md index 044f0f650c..e263573dba 100644 --- a/changelog.d/electrum-batch-concurrency.md +++ b/changelog.d/electrum-batch-concurrency.md @@ -13,7 +13,8 @@ Changed stops reading still stops the work (#1174). - **One heavy scripthash join at a time.** An unpaged Electrum or Esplora join with more than `--heavy-sh-creates` creates (conf - `heavy_sh_creates=`, default 10000, `0` off) now waits until no other - such join runs anywhere on the node. One join near a raised - `--max-sh-creates` can peak above 2 GiB, and a batch or a few - connections could start several at once. Lighter joins stay parallel. + `heavy_sh_creates=`, NixOS `services.rbitcoin.heavyShCreates`, default + 10000, `0` off) now waits until no other such join runs anywhere on the + node. One join near a raised `--max-sh-creates` can peak above 2 GiB, + and a batch or a few connections could start several at once. Lighter + joins stay parallel. diff --git a/docs/operator/operations.md b/docs/operator/operations.md index 75042d7bb6..54d21c2f4e 100644 --- a/docs/operator/operations.md +++ b/docs/operator/operations.md @@ -98,7 +98,7 @@ Clean smoke: | `--prune-seqsigwit-ram-threshold-bytes N` | `prune_seqsigwit_ram_threshold_bytes=` | `268435456` (256 MiB). `0` keeps nothing in RAM: every height, including tiny IBD blocks, is read from its file | | `--electrum-max-subs N` | `electrum_max_subs=` | **10000** — `blockchain.scripthash.subscribe` per Electrum connection (>= 1). Wallets subscribe every receive/change address up to their gap limit. | | `--max-sh-creates N` | `max_sh_creates=` | **10000** — unpaged SH join above N is refused (503 / JSON-RPC error). **0** is unlimited. A request that names a page still returns that page. | -| `--heavy-sh-creates N` | `heavy_sh_creates=` | **10000**: an unpaged SH join above N waits until no other such join runs, across every Electrum and Esplora client. **0** turns the wait off. Matters once `--max-sh-creates` is raised above it. | +| `--heavy-sh-creates N` | `heavy_sh_creates=` | **10000**: an unpaged SH join above N waits until no other such join runs, across every Electrum and Esplora client. **0** turns the wait off. Matters once `--max-sh-creates` is raised above it. NixOS `services.rbitcoin.heavyShCreates` (null omits the flag) | | `--sp-tweaks` | `sp_tweaks=` | **off** — thin BIP-352 tweak index (`sp_tweaks.*`). IBD seals it from the taproot origin when the flag is on from the start. A later enable still materializes after catch-up on `rbtc-idx-wb` (with block filters, one read of Class A). New tips are sealed on the confirm write thread once the watermark has caught the chain. Heights not yet sealed are served by the naive walk. Refused with `--prune-seqsigwit`: tweaks read input keys from scriptSig and witness | | `--sp-tweaks-dust SATS` | `sp_tweaks_dust=` | **1000** — omit served P2TR outs with `value <= SATS` (`0` = serve all; **546** matches Cake electrs) | | `--electrum-listen [ADDR]` | `electrum_listen=` | disabled; omit ADDR → `127.0.0.1:50001`. Address/scripthash methods need `--sh-index` | diff --git a/nix/modules/rbitcoin.nix b/nix/modules/rbitcoin.nix index e2b5e10684..aabbb0c8cc 100644 --- a/nix/modules/rbitcoin.nix +++ b/nix/modules/rbitcoin.nix @@ -118,6 +118,8 @@ let ++ optional cfg.datadirHdd "--datadir-hdd" ++ optional (cfg.milestone != null) "--milestone" ++ optional (cfg.milestone != null) (toString cfg.milestone) + ++ optional (cfg.heavyShCreates != null) "--heavy-sh-creates" + ++ optional (cfg.heavyShCreates != null) (toString cfg.heavyShCreates) ++ optional (cfg.trustPeer != null) "--trust-peer" ++ optional (cfg.trustPeer != null) cfg.trustPeer ++ lib.optionals cfg.sv2.tp.enable [ @@ -230,6 +232,18 @@ in description = "Build the Class B scripthash index. Electrum and Esplora enable it automatically."; }; + heavyShCreates = mkOption { + type = types.nullOr (types.ints.between 0 4294967295); + default = null; + example = 100000; + description = '' + Pass `--heavy-sh-creates N`. An unpaged Electrum or Esplora scripthash + join with more than N creates waits until no other such join is running + on the node. Null omits the flag and leaves the node default (10000). + 0 turns the wait off. It matters once `--max-sh-creates` is raised past it. + ''; + }; + silentPaymentIndex = mkOption { type = types.bool; default = false; diff --git a/nix/tests/nixos-module-eval.nix b/nix/tests/nixos-module-eval.nix index b08d71bf31..473b2e3b23 100644 --- a/nix/tests/nixos-module-eval.nix +++ b/nix/tests/nixos-module-eval.nix @@ -70,6 +70,7 @@ let }; health.enable = true; metrics = true; + heavyShCreates = 20000; sv2.tp = { enable = true; port = 18447; @@ -104,6 +105,7 @@ let enable = true; package = fakePackage; electrum.maxSubs = 25000; + heavyShCreates = 0; p2p = { listen = false; maxInbound = 0; @@ -267,6 +269,7 @@ assert defaultCfg.tor.control == null; assert defaultCfg.tor.controlCookie == null; assert defaultCfg.electrum.hiddenService == false; assert defaultCfg.electrum.maxSubs == null; +assert defaultCfg.heavyShCreates == null; assert defaultCfg.esplora.hiddenService == false; assert defaultCfg.i2p.sam == null; assert defaultCfg.i2p.acceptIncoming == false; @@ -371,6 +374,11 @@ assert cfg.systemd.tmpfiles.settings."10-rbitcoin"."/run/rbitcoin".d.mode == "07 assert builtins.match ".*--electrum-listen 127.0.0.1:50001.*" execStart != null; assert builtins.match ".*--electrum-max-subs 25000.*" execStart != null; assert builtins.match ".*--electrum-max-subs.*" listenOffExec == null; +assert builtins.match ".*--heavy-sh-creates 20000.*" execStart != null; +assert builtins.match ".*--heavy-sh-creates 0.*" listenOffExec != null; +assert builtins.match ".*--heavy-sh-creates.*" ( + defaultSystem.config.systemd.services.rbitcoin.serviceConfig.ExecStart +) == null; assert builtins.match ".*--esplora-listen 127.0.0.1:3000.*" execStart != null; assert builtins.match ".*--esplora-onion.*" execStart != null; assert builtins.match ".*--sh-index.*" execStart != null;