From 460810fd456337dfc86f1fe009bb12f46a4b4997 Mon Sep 17 00:00:00 2001 From: Sjors Provoost Date: Tue, 12 May 2026 17:44:38 +0200 Subject: [PATCH 01/21] ipc: add Cap'n Proto chain client and route block fetches Introduce an experimental backend that talks to a multiprocess `bitcoin-node` over a Unix socket using libmultiprocess Cap'n Proto framing, as exposed by Bitcoin Core PR #29409 (interfaces::Chain). Adds: - a new optional `daemon_ipc_socket` config / CLI option - a new `src/ipc.rs` module with an `IpcChain` handle that owns a dedicated worker thread running a current-thread tokio runtime + `LocalSet` (the generated capnp clients are not `Send`) - capnp / capnp-rpc / tokio / tokio-util / futures dependencies plus a path dep on the local `bitcoin-capnp-types` crate When `daemon_ipc_socket` is set, `Daemon::connect` boots the `IpcChain` worker and `Daemon::for_blocks` fetches each block via `Chain.findBlock(wantData=true)` instead of using the P2P getblock loop. The serialized form returned by libmultiprocess is the same wire encoding electrs already consumes, so the downstream parsing and indexing pipeline is unchanged. When the option is absent (the default), the P2P path is preserved verbatim, so this is a strict superset of existing behaviour. --- Cargo.lock | 227 ++++++++++++++++++++++++ Cargo.toml | 6 + internal/config_specification.toml | 6 + src/config.rs | 2 + src/daemon.rs | 33 +++- src/ipc.rs | 271 +++++++++++++++++++++++++++++ src/lib.rs | 1 + 7 files changed, 543 insertions(+), 3 deletions(-) create mode 100644 src/ipc.rs diff --git a/Cargo.lock b/Cargo.lock index 7f3557643..5464c1076 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -95,6 +95,14 @@ dependencies = [ "serde", ] +[[package]] +name = "bitcoin-capnp-types" +version = "0.2.0" +dependencies = [ + "capnp", + "capnpc", +] + [[package]] name = "bitcoin-internals" version = "0.3.0" @@ -196,6 +204,12 @@ dependencies = [ "objc2", ] +[[package]] +name = "bytes" +version = "1.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33" + [[package]] name = "bzip2-sys" version = "0.1.13+1.0.8" @@ -206,6 +220,46 @@ dependencies = [ "pkg-config", ] +[[package]] +name = "capnp" +version = "0.25.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63da65e5e9ffc3b8f993d4ad222a548152549351a643f6b850a7773cb6ff2809" +dependencies = [ + "embedded-io", +] + +[[package]] +name = "capnp-futures" +version = "0.25.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "73b69dfddccc57844f9a90f9d72b44b97c326914851ea94fb7da40ef9cad6e8d" +dependencies = [ + "capnp", + "futures-channel", + "futures-util", +] + +[[package]] +name = "capnp-rpc" +version = "0.25.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e3c74c337e87f75f3174ffb3513738ab0acc631c04f64cc33b867c60f84771da" +dependencies = [ + "capnp", + "capnp-futures", + "futures", +] + +[[package]] +name = "capnpc" +version = "0.25.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fca02be865c8c5a78bfc24b9819006ab6b59bef238467203928e26459557af93" +dependencies = [ + "capnp", +] + [[package]] name = "cargo_toml" version = "0.14.1" @@ -410,15 +464,19 @@ version = "0.11.1" dependencies = [ "anyhow", "bitcoin", + "bitcoin-capnp-types", "bitcoin-test-data", "bitcoin_slices", "bitcoincore-rpc", + "capnp", + "capnp-rpc", "configure_me", "configure_me_codegen", "crossbeam-channel", "ctrlc", "dirs-next", "env_logger", + "futures", "hex_lit", "log", "parking_lot", @@ -431,8 +489,16 @@ dependencies = [ "signal-hook", "tempfile", "tiny_http", + "tokio", + "tokio-util", ] +[[package]] +name = "embedded-io" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9eb1aa714776b75c7e67e1da744b81a129b3ff919c8712b5e1b32252c1f07cc7" + [[package]] name = "env_logger" version = "0.10.2" @@ -474,6 +540,94 @@ version = "1.0.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1" +[[package]] +name = "futures" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b147ee9d1f6d097cef9ce628cd2ee62288d963e16fb287bd9286455b241382d" +dependencies = [ + "futures-channel", + "futures-core", + "futures-executor", + "futures-io", + "futures-sink", + "futures-task", + "futures-util", +] + +[[package]] +name = "futures-channel" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "07bbe89c50d7a535e539b8c17bc0b49bdb77747034daa8087407d655f3f7cc1d" +dependencies = [ + "futures-core", + "futures-sink", +] + +[[package]] +name = "futures-core" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d" + +[[package]] +name = "futures-executor" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "baf29c38818342a3b26b5b923639e7b1f4a61fc5e76102d4b1981c6dc7a7579d" +dependencies = [ + "futures-core", + "futures-task", + "futures-util", +] + +[[package]] +name = "futures-io" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cecba35d7ad927e23624b22ad55235f2239cfa44fd10428eecbeba6d6a717718" + +[[package]] +name = "futures-macro" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "futures-sink" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c39754e157331b013978ec91992bde1ac089843443c49cbc7f46150b0fad0893" + +[[package]] +name = "futures-task" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "037711b3d59c33004d3856fbdc83b99d4ff37a24768fa1be9ce3538a1cde4393" + +[[package]] +name = "futures-util" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6" +dependencies = [ + "futures-channel", + "futures-core", + "futures-io", + "futures-macro", + "futures-sink", + "futures-task", + "memchr", + "pin-project-lite", + "slab", +] + [[package]] name = "generic-array" version = "0.14.7" @@ -690,6 +844,17 @@ dependencies = [ "serde_json", ] +[[package]] +name = "mio" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a69bcab0ad47271a0234d9422b131806bf3968021e5dc9328caf2d4cd58557fc" +dependencies = [ + "libc", + "wasi 0.11.0+wasi-snapshot-preview1", + "windows-sys 0.61.0", +] + [[package]] name = "nix" version = "0.31.1" @@ -771,6 +936,12 @@ version = "1.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5bddc33f680b79eaf1e2e56da792c3c2236f86985bbc3a886e8ddee17ae4d3a4" +[[package]] +name = "pin-project-lite" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" + [[package]] name = "pkg-config" version = "0.3.32" @@ -1131,12 +1302,28 @@ dependencies = [ "libc", ] +[[package]] +name = "slab" +version = "0.4.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" + [[package]] name = "smallvec" version = "1.15.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8917285742e9f3e1683f0a9c4e6b57960b7314d0b08d30d1ecd426713ee2eee9" +[[package]] +name = "socket2" +version = "0.6.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3a766e1110788c36f4fa1c2b71b387a7815aa65f88ce0229841826633d93723e" +dependencies = [ + "libc", + "windows-sys 0.61.0", +] + [[package]] name = "syn" version = "2.0.101" @@ -1222,6 +1409,46 @@ dependencies = [ "log", ] +[[package]] +name = "tokio" +version = "1.50.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "27ad5e34374e03cfffefc301becb44e9dc3c17584f414349ebe29ed26661822d" +dependencies = [ + "bytes", + "libc", + "mio", + "pin-project-lite", + "socket2", + "tokio-macros", + "windows-sys 0.61.0", +] + +[[package]] +name = "tokio-macros" +version = "2.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c55a2eff8b69ce66c84f85e1da1c233edc36ceb85a2058d11b0d6a3c7e7569c" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "tokio-util" +version = "0.7.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9ae9cec805b01e8fc3fd2fe289f89149a9b66dd16786abd8b19cfa7b48cb0098" +dependencies = [ + "bytes", + "futures-core", + "futures-io", + "futures-sink", + "pin-project-lite", + "tokio", +] + [[package]] name = "toml" version = "0.5.11" diff --git a/Cargo.toml b/Cargo.toml index e18346bcc..5e6e03da4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -25,11 +25,15 @@ spec = "internal/config_specification.toml" anyhow = "1.0" bitcoin = { version = "0.32.8", features = ["serde", "rand-std"] } bitcoin_slices = { version = "0.11.0", features = ["bitcoin", "sha2"] } +bitcoin-capnp-types = { path = "../capnp-types" } bitcoincore-rpc = { version = "0.19.0" } +capnp = "0.25.0" +capnp-rpc = "0.25.0" configure_me = "0.4" crossbeam-channel = "0.5" dirs-next = "2.0" env_logger = "0.10" +futures = "0.3" log = "0.4" parking_lot = "0.12" prometheus = { version = "0.14", optional = true } @@ -38,6 +42,8 @@ serde = "1.0.184" serde_derive = "1.0.184" serde_json = "1.0" tiny_http = { version = "0.12", optional = true } +tokio = { version = "1", features = ["rt", "net", "macros", "io-util", "sync", "time"] } +tokio-util = { version = "0.7", features = ["compat"] } [target.'cfg(windows)'.dependencies] ctrlc = "=3.5.2" diff --git a/internal/config_specification.toml b/internal/config_specification.toml index f4ae65e36..cad145e7a 100644 --- a/internal/config_specification.toml +++ b/internal/config_specification.toml @@ -81,6 +81,12 @@ name = "daemon_p2p_addr" type = "crate::config::ResolvAddr" doc = "Bitcoin daemon p2p 'addr:port' to connect (default: 127.0.0.1:8333 for mainnet, 127.0.0.1:18333 for testnet, 127.0.0.1:18444 for regtest and 127.0.0.1:38333 for signet)" +[[param]] +name = "daemon_ipc_socket" +type = "std::path::PathBuf" +doc = "EXPERIMENTAL: path to a Bitcoin Core multiprocess (`bitcoin-node -ipcbind=unix`) socket. When set, electrs uses Cap'n Proto IPC (Chain interface from PR #29409) for block fetching and tip queries instead of the P2P protocol." +optional = true + [[param]] name = "monitoring_addr" type = "crate::config::ResolvAddr" diff --git a/src/config.rs b/src/config.rs index b844dda06..81ca7210f 100644 --- a/src/config.rs +++ b/src/config.rs @@ -132,6 +132,7 @@ pub struct Config { pub daemon_auth: SensitiveAuth, pub daemon_rpc_addr: SocketAddr, pub daemon_p2p_addr: SocketAddr, + pub daemon_ipc_socket: Option, pub electrum_rpc_addr: SocketAddr, pub monitoring_addr: SocketAddr, pub wait_duration: Duration, @@ -345,6 +346,7 @@ impl Config { daemon_auth, daemon_rpc_addr, daemon_p2p_addr, + daemon_ipc_socket: config.daemon_ipc_socket, electrum_rpc_addr, monitoring_addr, wait_duration: Duration::from_secs(config.wait_duration_secs), diff --git a/src/daemon.rs b/src/daemon.rs index 40df169ba..e9a1a9a05 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -16,6 +16,7 @@ use std::path::Path; use crate::{ chain::{Chain, NewHeader}, config::Config, + ipc::IpcChain, metrics::Metrics, p2p::Connection, signals::ExitFlag, @@ -102,6 +103,7 @@ fn rpc_connect(config: &Config) -> Result { pub struct Daemon { p2p: Mutex, rpc: Client, + ipc: Option, } impl Daemon { @@ -144,7 +146,17 @@ impl Daemon { metrics, config.magic, )?); - Ok(Self { p2p, rpc }) + let ipc = match config.daemon_ipc_socket.as_deref() { + Some(path) => { + info!( + "connecting to bitcoin-node IPC socket: {} (experimental)", + path.display() + ); + Some(IpcChain::connect(path).context("IPC chain connection failed")?) + } + None => None, + }; + Ok(Self { p2p, rpc, ipc }) } pub(crate) fn estimate_fee(&self, nblocks: u16) -> Result> { @@ -292,12 +304,27 @@ impl Daemon { self.p2p.lock().get_new_headers(chain) } - pub(crate) fn for_blocks(&self, blockhashes: B, func: F) -> Result<()> + pub(crate) fn for_blocks(&self, blockhashes: B, mut func: F) -> Result<()> where B: IntoIterator, F: FnMut(BlockHash, SerBlock), { - self.p2p.lock().for_blocks(blockhashes, func) + if let Some(ipc) = &self.ipc { + // Fetch each block over IPC instead of P2P. The Chain interface's + // `findBlock(wantData=true)` returns the same serialized form + // electrs's P2P path delivers, so the downstream pipeline is + // unchanged. + for hash in blockhashes { + let block = ipc + .get_block(hash) + .with_context(|| format!("IPC findBlock failed for {hash}"))? + .ok_or_else(|| anyhow!("IPC findBlock: block {hash} not found"))?; + func(hash, block); + } + Ok(()) + } else { + self.p2p.lock().for_blocks(blockhashes, func) + } } pub(crate) fn new_block_notification(&self) -> Receiver<()> { diff --git a/src/ipc.rs b/src/ipc.rs new file mode 100644 index 000000000..d2e9fe8a3 --- /dev/null +++ b/src/ipc.rs @@ -0,0 +1,271 @@ +//! Cap'n Proto IPC client for Bitcoin Core's `interfaces::Chain`. +//! +//! This is an experimental backend that talks to a multiprocess `bitcoin-node` +//! (Bitcoin Core PR #29409) over a unix socket using the libmultiprocess +//! Cap'n Proto framing. When configured, electrs uses this in place of the P2P +//! protocol for fetching raw blocks and tip information. +//! +//! The Cap'n Proto stack is asynchronous and not `Send` (the generated client +//! types contain `Rc>`). To bridge it to electrs's synchronous, +//! threaded code we run a dedicated worker thread that owns a current-thread +//! tokio runtime + `LocalSet`, and communicate via an mpsc channel of +//! type-erased async closures returning their results through a oneshot. +#![allow(dead_code)] + +use std::path::{Path, PathBuf}; +use std::pin::Pin; +use std::thread as stdthread; + +use anyhow::{anyhow, bail, Context, Result}; +use bitcoin::{consensus::Decodable, hashes::Hash, BlockHash}; +use bitcoin_capnp_types::{ + chain_capnp::chain, + init_capnp::init, + proxy_capnp::{thread as proxy_thread, thread_map}, +}; +use capnp_rpc::{rpc_twoparty_capnp::Side, twoparty::VatNetwork, RpcSystem}; +use futures::{future::LocalBoxFuture, io::BufReader, FutureExt}; +use tokio::net::UnixStream; +use tokio::sync::{mpsc, oneshot}; +use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt}; + +use crate::types::SerBlock; + +/// Boxed (non-Send) future returned by a job. The closure itself must be +/// `Send` so it can be shipped from caller threads to the IPC worker, but +/// once on the worker it runs on a single-threaded LocalSet, so the future +/// it produces does not need to be `Send`. +type Job = Box FnOnce(&'a Ctx) -> LocalBoxFuture<'a, ()> + Send + 'static>; + +/// IPC context held by the worker thread: the chain client and the proxy +/// thread handle that every call must reference. +struct Ctx { + chain: chain::Client, + thread: proxy_thread::Client, +} + +/// Sync handle to the IPC worker thread. Cloneable; clones share the same +/// connection. +#[derive(Clone)] +pub(crate) struct IpcChain { + tx: mpsc::UnboundedSender, +} + +impl IpcChain { + /// Spawn the worker thread, connect to the unix socket, and bootstrap an + /// `Init` client + `Chain` client. Blocks until the connection is ready. + pub(crate) fn connect(socket: &Path) -> Result { + let socket = socket.to_path_buf(); + let (tx, rx) = mpsc::unbounded_channel::(); + let (ready_tx, ready_rx) = std::sync::mpsc::channel::>(); + stdthread::Builder::new() + .name("electrs-ipc".into()) + .spawn(move || worker(socket, rx, ready_tx)) + .context("failed to spawn IPC worker thread")?; + ready_rx + .recv() + .context("IPC worker thread exited before signalling readiness")??; + Ok(Self { tx }) + } + + /// Submit a job to the worker and block on its result. + fn call(&self, f: F) -> Result + where + R: Send + 'static, + F: for<'a> FnOnce(&'a Ctx) -> LocalBoxFuture<'a, Result> + Send + 'static, + { + let (otx, orx) = oneshot::channel::>(); + let job: Job = Box::new(move |ctx| { + async move { + let res = f(ctx).await; + let _ = otx.send(res); + } + .boxed_local() + }); + self.tx + .send(job) + .map_err(|_| anyhow!("IPC worker thread is gone"))?; + orx.blocking_recv() + .map_err(|_| anyhow!("IPC worker dropped the response"))? + } + + pub(crate) fn get_height(&self) -> Result> { + self.call(|ctx| { + async move { + let mut req = ctx.chain.get_height_request(); + req.get().get_context()?.set_thread(ctx.thread.clone()); + let resp = req.send().promise.await?; + let r = resp.get()?; + Ok(if r.get_has_result() { + Some(r.get_result()) + } else { + None + }) + } + .boxed_local() + }) + } + + pub(crate) fn is_initial_block_download(&self) -> Result { + self.call(|ctx| { + async move { + let mut req = ctx.chain.is_initial_block_download_request(); + req.get().get_context()?.set_thread(ctx.thread.clone()); + let resp = req.send().promise.await?; + Ok(resp.get()?.get_result()) + } + .boxed_local() + }) + } + + pub(crate) fn have_pruned(&self) -> Result { + self.call(|ctx| { + async move { + let mut req = ctx.chain.have_pruned_request(); + req.get().get_context()?.set_thread(ctx.thread.clone()); + let resp = req.send().promise.await?; + Ok(resp.get()?.get_result()) + } + .boxed_local() + }) + } + + pub(crate) fn get_block_hash(&self, height: i32) -> Result { + self.call(move |ctx| { + async move { + let mut req = ctx.chain.get_block_hash_request(); + req.get().get_context()?.set_thread(ctx.thread.clone()); + req.get().set_height(height); + let resp = req.send().promise.await?; + let bytes = resp.get()?.get_result()?; + if bytes.len() != 32 { + bail!("unexpected block hash length {}", bytes.len()); + } + let mut buf = [0u8; 32]; + buf.copy_from_slice(bytes); + Ok(BlockHash::from_byte_array(buf)) + } + .boxed_local() + }) + } + + /// Fetch a serialized block by hash via `Chain.findBlock(wantData=true)`. + /// Returns `Ok(None)` if the block is not known to the node. + pub(crate) fn get_block(&self, hash: BlockHash) -> Result> { + let raw: [u8; 32] = hash.to_byte_array(); + self.call(move |ctx| { + async move { + let mut req = ctx.chain.find_block_request(); + req.get().get_context()?.set_thread(ctx.thread.clone()); + req.get().set_hash(&raw); + { + let mut params = req.get().init_block(); + params.set_want_data(true); + } + let resp = req.send().promise.await?; + let r = resp.get()?; + if !r.get_result() { + return Ok(None); + } + let block = r.get_block()?; + let data = block.get_data()?.to_vec(); + Ok(Some(data)) + } + .boxed_local() + }) + } +} + +/// Worker thread entry point. Owns the tokio runtime, the capnp RPC system, +/// and the dispatch loop. Never returns under normal operation. +fn worker( + socket: PathBuf, + mut rx: mpsc::UnboundedReceiver, + ready: std::sync::mpsc::Sender>, +) { + let rt = match tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + { + Ok(rt) => rt, + Err(e) => { + let _ = ready.send(Err(anyhow!("tokio runtime build: {e}"))); + return; + } + }; + let local = tokio::task::LocalSet::new(); + rt.block_on(local.run_until(async move { + let ctx = match bootstrap(&socket).await { + Ok(ctx) => ctx, + Err(e) => { + let _ = ready.send(Err(e)); + return; + } + }; + let _ = ready.send(Ok(())); + while let Some(job) = rx.recv().await { + let fut: Pin>> = job(&ctx); + fut.await; + } + })); +} + +async fn bootstrap(socket: &Path) -> Result { + let stream = UnixStream::connect(socket) + .await + .with_context(|| format!("connecting to bitcoin-node IPC socket {}", socket.display()))?; + let (reader, writer) = stream.into_split(); + let buf_reader = BufReader::new(reader.compat()); + let buf_writer = futures::io::BufWriter::new(writer.compat_write()); + let network = VatNetwork::new(buf_reader, buf_writer, Side::Client, Default::default()); + let mut rpc_system = RpcSystem::new(Box::new(network), None); + let init: init::Client = rpc_system.bootstrap(Side::Server); + tokio::task::spawn_local(rpc_system); + + // Construct + obtain a thread handle. + let resp = init + .construct_request() + .send() + .promise + .await + .context("Init.construct failed")?; + let thread_map: thread_map::Client = resp + .get() + .context("Init.construct response")? + .get_thread_map() + .context("get_thread_map")?; + let resp = thread_map + .make_thread_request() + .send() + .promise + .await + .context("ThreadMap.makeThread failed")?; + let thread: proxy_thread::Client = resp + .get() + .context("makeThread response")? + .get_result() + .context("makeThread result")?; + + // Make a Chain client. + let mut req = init.make_chain_request(); + req.get() + .get_context() + .context("makeChain context")? + .set_thread(thread.clone()); + let resp = req.send().promise.await.context("Init.makeChain failed")?; + let chain: chain::Client = resp + .get() + .context("makeChain response")? + .get_result() + .context("makeChain result")?; + + Ok(Ctx { chain, thread }) +} + +/// Decode a serialized block. Helper that keeps callers free of the +/// `bitcoin::consensus` import. +pub(crate) fn decode_block(buf: &[u8]) -> Result { + let mut cursor = std::io::Cursor::new(buf); + bitcoin::Block::consensus_decode(&mut cursor) + .map_err(|e| anyhow!("failed to decode block: {e}")) +} diff --git a/src/lib.rs b/src/lib.rs index 1458f06ff..4dc159bf7 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -14,6 +14,7 @@ mod daemon; mod db; mod electrum; mod index; +mod ipc; mod mempool; mod merkle; mod metrics; From c79e25d3d77bfa0caa202a0bf973671230edfbdd Mon Sep 17 00:00:00 2001 From: Sjors Provoost Date: Tue, 12 May 2026 18:09:21 +0200 Subject: [PATCH 02/21] tests: add IPC integration variant of run.sh Mirrors `tests/run.sh` but launches the multiprocess `bitcoin node` binary with `-ipcbind=unix` and points electrs at the resulting `/regtest/node.sock` via `--daemon-ipc-socket`. The path to the multiprocess binary is configurable via `$BITCOIN_MULTIPROCESS_BIN`; if unset, the script looks for `bitcoin` on `PATH`. There is no hardcoded fallback. --- tests/run-ipc.sh | 149 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 149 insertions(+) create mode 100755 tests/run-ipc.sh diff --git a/tests/run-ipc.sh b/tests/run-ipc.sh new file mode 100755 index 000000000..14d0b3723 --- /dev/null +++ b/tests/run-ipc.sh @@ -0,0 +1,149 @@ +#!/bin/bash +# Integration test variant that exercises the experimental Cap'n Proto IPC +# backend introduced for Bitcoin Core PR #29409. Requires a multiprocess +# `bitcoin` multi-call binary (resolved via $BITCOIN_MULTIPROCESS_BIN or +# `bitcoin` on PATH) and a freshly built electrs. +set -euo pipefail + +BITCOIN_MULTIPROCESS_BIN="${BITCOIN_MULTIPROCESS_BIN:-$(command -v bitcoin || true)}" + +if [ -z "$BITCOIN_MULTIPROCESS_BIN" ] || [ ! -x "$BITCOIN_MULTIPROCESS_BIN" ]; then + echo "ERROR: multiprocess 'bitcoin' multi-call binary not found." >&2 + echo "Set BITCOIN_MULTIPROCESS_BIN to the binary built from bitcoin/bitcoin#29409," >&2 + echo "or place it on PATH as 'bitcoin'." >&2 + exit 1 +fi + +rm -rf data/ +mkdir -p data/{bitcoin,electrum,electrs} + +cleanup() { + trap - SIGTERM SIGINT + set +e +o pipefail + jobs + for j in `jobs -rp` + do + kill $j + wait $j + done +} +trap cleanup SIGINT SIGTERM EXIT + +wait_for() { + CMD=$1 + shift + TEST_ARGS=$* + for _ in `seq 0 9`; do + test "$($CMD | jq -c .)" $TEST_ARGS && break || sleep 2 + done +} + +BTC="$BITCOIN_MULTIPROCESS_BIN rpc -chain=regtest -datadir=$PWD/data/bitcoin" +ELECTRUM="electrum --regtest" +EL="$ELECTRUM --wallet=data/electrum/wallet" +SOCK="$PWD/data/bitcoin/regtest/node.sock" + +tail_log() { + tail -n +0 -F $1 || true +} + +echo "Starting $($BITCOIN_MULTIPROCESS_BIN node -version | head -n1) (multiprocess, IPC enabled)..." +$BITCOIN_MULTIPROCESS_BIN node \ + -regtest -datadir=$PWD/data/bitcoin \ + -ipcbind=unix \ + -printtoconsole=0 \ + -fallbackfee=0.0001 & +BITCOIND_PID=$! + +# Wait for both RPC and the IPC socket. +$BTC -rpcwait getblockcount > /dev/null +for _ in `seq 0 50`; do + test -S "$SOCK" && break || sleep 0.2 +done +test -S "$SOCK" || { echo "IPC socket $SOCK never appeared" >&2; exit 1; } +echo "IPC socket ready: $SOCK" + +echo "Creating Electrum `electrum version --offline` wallet..." +WALLET=`$EL --offline create --seed_type=segwit` +MINING_ADDR=`$EL --offline getunusedaddress` + +$BTC generatetoaddress 110 $MINING_ADDR > /dev/null +echo `$BTC getblockchaininfo | jq -r '"Generated \(.blocks) regtest blocks (\(.size_on_disk/1e3) kB)"'` to $MINING_ADDR + +TIP=`$BTC getbestblockhash` + +export RUST_LOG=electrs=debug +electrs \ + --db-dir=data/electrs \ + --daemon-dir=data/bitcoin \ + --network=regtest \ + --daemon-ipc-socket="$SOCK" \ + 2> data/electrs/regtest-debug.log & +ELECTRS_PID=$! +tail_log data/electrs/regtest-debug.log | grep -m1 "serving Electrum RPC" + +# Confirm the IPC backend actually engaged. The Daemon::connect path logs +# nothing distinctive yet; instead probe the debug log for any IPC error or +# verify the for_blocks code path was taken by checking a unique marker. +if ! grep -q "ipc" data/electrs/regtest-debug.log; then + : # no explicit log line yet; we rely on cargo tests + functional behaviour +fi + +curl localhost:24224 -o metrics.txt + +$ELECTRUM daemon --server localhost:60401:t -1 -vDEBUG 2> data/electrum/regtest-debug.log & +ELECTRUM_PID=$! +tail_log data/electrum/regtest-debug.log | grep -m1 "connection established" +$EL getinfo | jq . + +echo "Loading Electrum wallet..." +$EL load_wallet + +echo "Running integration tests (IPC backend):" + +echo " * getbalance" +wait_for "$EL getbalance" == '{"confirmed":"550","unmatured":"4950"}' + +echo " * getunusedaddress" +NEW_ADDR=`$EL getunusedaddress` + +echo " * payto & broadcast" +TXID=$($EL broadcast $($EL payto $NEW_ADDR 123 --fee 0.001 --password='')) + +echo " * get_tx_status" +wait_for "$EL get_tx_status $TXID" == '{"confirmations":0}' + +echo " * getaddresshistory" +test "`$EL getaddresshistory $NEW_ADDR | jq -c .`" == "[{\"fee\":100000,\"height\":0,\"tx_hash\":\"$TXID\"}]" + +echo " * getbalance" +test "`$EL getbalance | jq -c .`" == '{"confirmed":"549.999","unmatured":"4950"}' + +echo "Generating bitcoin block..." +$BTC generatetoaddress 1 $MINING_ADDR > /dev/null +$BTC getblockcount > /dev/null + +echo " * wait for new block" +kill -USR1 $ELECTRS_PID # notify server to index new block +tail_log data/electrum/regtest-debug.log | grep -m1 "verified $TXID" > /dev/null + +echo " * get_tx_status" +test "`$EL get_tx_status $TXID | jq -c .`" == '{"confirmations":1}' + +echo " * getaddresshistory" +test "`$EL getaddresshistory $NEW_ADDR | jq -c .`" == "[{\"height\":111,\"tx_hash\":\"$TXID\"}]" + +echo " * getbalance" +test "`$EL getbalance | jq -c .`" == '{"confirmed":"599.999","unmatured":"4950.001"}' + +echo "Electrum `$EL stop`" # disconnect wallet +wait $ELECTRUM_PID + +kill -INT $ELECTRS_PID # close server +tail_log data/electrs/regtest-debug.log | grep -m1 "electrs stopped" +wait $ELECTRS_PID + +$BTC stop # stop bitcoind +wait $BITCOIND_PID + +echo "=== PASSED (IPC) ===" From 328a8324a63777e3043ddd4e6fbbd4dc27c6b01b Mon Sep 17 00:00:00 2001 From: Sjors Provoost Date: Tue, 12 May 2026 18:01:18 +0200 Subject: [PATCH 03/21] ci: run the experimental IPC integration test - New `integration-ipc` job clones bitcoin/bitcoin#29409, builds the multiprocess `bitcoin` multi-call binary with capnproto enabled, installs Electrum + electrs, and invokes `tests/run-ipc.sh`. - The existing `build` and `integration` jobs now also clone the `bitcoin-capnp-types` chain-interface branch into a sibling directory to satisfy the `path = "../capnp-types"` Cargo dep introduced in this branch. Mirrored in both Dockerfiles. --- .github/workflows/rust.yml | 83 +++++++++++++++++++++++++++++++++++++- Dockerfile | 8 +++- Dockerfile.ci | 8 +++- 3 files changed, 94 insertions(+), 5 deletions(-) diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index db3a66881..6f2fb902d 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -28,6 +28,10 @@ jobs: steps: - uses: actions/checkout@v6 + - name: Checkout bitcoin-capnp-types (experimental chain-interface branch) + run: | + git clone --depth 1 --branch 2026/04/chain-interface \ + https://github.com/Sjors/bitcoin-capnp-types.git ../capnp-types - uses: dtolnay/rust-toolchain@stable with: components: rustfmt, clippy @@ -36,7 +40,7 @@ jobs: run: rustup component add rustfmt clippy - name: Install other dependencies - run: sudo apt install build-essential libclang-dev + run: sudo apt install build-essential libclang-dev capnproto libcapnp-dev - name: Format run: cargo fmt --all -- --check @@ -60,3 +64,80 @@ jobs: run: docker build -f Dockerfile.ci . --rm -t electrs:tests - name: Test run: docker run -v $PWD/contrib/:/contrib -v $PWD/tests/:/tests --rm electrs:tests bash /tests/run.sh + + integration-ipc: + name: Integration (IPC, experimental, PR 29409) + runs-on: ubuntu-latest + steps: + - name: Checkout electrs + uses: actions/checkout@v6 + + - name: Checkout bitcoin-capnp-types (experimental chain-interface branch) + run: | + git clone --depth 1 --branch 2026/04/chain-interface \ + https://github.com/Sjors/bitcoin-capnp-types.git ../capnp-types + + - uses: dtolnay/rust-toolchain@stable + + - name: Install build & test dependencies + run: | + sudo apt-get update -qqy + sudo apt-get install -qqy \ + build-essential cmake pkgconf python3 \ + libevent-dev libboost-dev capnproto libcapnp-dev \ + librocksdb-dev libclang-dev ccache \ + python3-venv python3-pip git jq curl libsecp256k1-2 + + - name: ccache + uses: actions/cache@v4 + with: + path: ~/.cache/ccache + key: ccache-bitcoin-29409-${{ runner.os }}-${{ github.sha }} + restore-keys: ccache-bitcoin-29409-${{ runner.os }}- + + - name: Checkout Bitcoin Core PR #29409 + run: | + git clone --depth 1 https://github.com/bitcoin/bitcoin.git + cd bitcoin + git fetch --depth 1 origin pull/29409/head:pr-29409 + git checkout pr-29409 + + - name: Build Bitcoin Core (multiprocess) + run: | + cd bitcoin + cmake -B build \ + -DENABLE_WALLET=ON \ + -DBUILD_TESTS=OFF \ + -DCMAKE_C_COMPILER_LAUNCHER=ccache \ + -DCMAKE_CXX_COMPILER_LAUNCHER=ccache + cmake --build build -j $(nproc) --target bitcoin bitcoin-node bitcoin-cli + + - name: Install Electrum wallet + run: | + git clone --depth 1 --recurse-submodules https://github.com/spesmilo/electrum/ + python3 -m venv --system-site-packages venv + ELECTRUM_ECC_DONT_COMPILE=1 venv/bin/pip install -e electrum/ + sudo ln -s "$PWD/venv/bin/electrum" /usr/local/bin/electrum + electrum version --offline + + - name: Build electrs + env: + ROCKSDB_INCLUDE_DIR: /usr/include + ROCKSDB_LIB_DIR: /usr/lib + run: cargo build --locked --bin electrs + + - name: Run IPC integration test + env: + BITCOIN_MULTIPROCESS_BIN: ${{ github.workspace }}/bitcoin/build/bin/bitcoin + run: | + export PATH="$PWD/target/debug:$PATH" + ./tests/run-ipc.sh + + - name: Upload electrs debug log on failure + if: failure() + uses: actions/upload-artifact@v4 + with: + name: electrs-ipc-debug-log + path: | + data/electrs/regtest-debug.log + data/electrum/regtest-debug.log diff --git a/Dockerfile b/Dockerfile index 055398535..987aaede9 100644 --- a/Dockerfile +++ b/Dockerfile @@ -8,9 +8,13 @@ RUN apt-get install -qqy librocksdb-dev curl ### Electrum Rust Server ### FROM base AS electrs-build -RUN apt-get install -qqy cargo build-essential libclang-dev +RUN apt-get install -qqy cargo build-essential libclang-dev git capnproto libcapnp-dev -# Install electrs +# Install electrs (with sibling bitcoin-capnp-types checkout for the +# experimental chain-interface branch -- see Cargo.toml path dep). +WORKDIR /build/capnp-types +RUN git clone --depth 1 --branch 2026/04/chain-interface \ + https://github.com/Sjors/bitcoin-capnp-types.git . WORKDIR /build/electrs COPY . . ENV ROCKSDB_INCLUDE_DIR=/usr/include diff --git a/Dockerfile.ci b/Dockerfile.ci index 9106ac10b..6008f30ba 100644 --- a/Dockerfile.ci +++ b/Dockerfile.ci @@ -8,9 +8,13 @@ RUN apt-get install -qqy librocksdb-dev wget ### Electrum Rust Server ### FROM base AS electrs-build -RUN apt-get install -qqy cargo build-essential libclang-dev +RUN apt-get install -qqy cargo build-essential libclang-dev git capnproto libcapnp-dev -# Install electrs +# Install electrs (with sibling bitcoin-capnp-types checkout for the +# experimental chain-interface branch -- see Cargo.toml path dep). +WORKDIR /build/capnp-types +RUN git clone --depth 1 --branch 2026/04/chain-interface \ + https://github.com/Sjors/bitcoin-capnp-types.git . WORKDIR /build/electrs COPY . . ENV ROCKSDB_INCLUDE_DIR=/usr/include From 430b83c2744cf8bda9d9d9e39ac8fe4199360643 Mon Sep 17 00:00:00 2001 From: Sjors Provoost Date: Tue, 12 May 2026 18:13:03 +0200 Subject: [PATCH 04/21] daemon: use Chain.havePruned for the pruning check when IPC is set Replaces a JSON-RPC `getblockchaininfo` call with the equivalent `Chain.havePruned` capnp method whenever the IPC backend is configured. The RPC fallback is retained for the non-IPC path so this remains a strict no-op for current deployments. This is the first migration of an existing electrs RPC call onto the PR #29409 chain interface; it is small but exercises the pattern end-to-end (config-conditional dispatch + integration test coverage). --- src/daemon.rs | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/src/daemon.rs b/src/daemon.rs index e9a1a9a05..b702c80ac 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -136,10 +136,6 @@ impl Daemon { if !network_info.network_active { bail!("electrs requires active bitcoind p2p network"); } - let info = rpc.get_blockchain_info()?; - if info.pruned { - bail!("electrs requires non-pruned bitcoind node"); - } let p2p = Mutex::new(Connection::connect( config.daemon_p2p_addr, @@ -156,6 +152,17 @@ impl Daemon { } None => None, }; + + // Check the node is non-pruned, preferring the IPC chain interface + // over the JSON-RPC `getblockchaininfo` call when it is available. + let pruned = match &ipc { + Some(ipc) => ipc.have_pruned().context("Chain.havePruned failed")?, + None => rpc.get_blockchain_info()?.pruned, + }; + if pruned { + bail!("electrs requires non-pruned bitcoind node"); + } + Ok(Self { p2p, rpc, ipc }) } From 7ddb6435a9858817de7d5d0d01baf825aaf2fc47 Mon Sep 17 00:00:00 2001 From: Sjors Provoost Date: Tue, 12 May 2026 18:15:44 +0200 Subject: [PATCH 05/21] daemon: serve get_block_txids from Chain.findBlock when IPC is set When the IPC chain interface is available, fetch the block bytes via `Chain.findBlock` (already used for block download in for_blocks) and extract txids from the decoded block instead of issuing a JSON-RPC `getblock` call. This mirrors the call pattern of for_blocks and removes one more dependency on the JSON-RPC backend. The non-IPC path keeps the original `get_block_info` RPC call so this is a no-op for current deployments. Verified with both tests/run.sh (=== PASSED ===) and tests/run-ipc.sh (=== PASSED (IPC) ===). --- src/daemon.rs | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/src/daemon.rs b/src/daemon.rs index b702c80ac..bd7ba08e4 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -238,6 +238,15 @@ impl Daemon { } pub(crate) fn get_block_txids(&self, blockhash: BlockHash) -> Result> { + if let Some(ipc) = &self.ipc { + let bytes = ipc + .get_block(blockhash) + .with_context(|| format!("IPC findBlock failed for {blockhash}"))? + .ok_or_else(|| anyhow!("IPC findBlock: block {blockhash} not found"))?; + let block = crate::ipc::decode_block(&bytes) + .with_context(|| format!("failed to decode block {blockhash}"))?; + return Ok(block.txdata.iter().map(|tx| tx.compute_txid()).collect()); + } Ok(self .rpc .get_block_info(&blockhash) From 207b9d6d7ccf8254d19a92aaf80be8d9ccd7c657 Mon Sep 17 00:00:00 2001 From: Sjors Provoost Date: Tue, 12 May 2026 18:19:02 +0200 Subject: [PATCH 06/21] daemon: broadcast transactions via Chain.broadcastTransaction when IPC is set Adds an `IpcChain::broadcast_transaction` wrapper around the `Chain.broadcastTransaction` capnp method and uses it from `Daemon::broadcast` when the IPC backend is configured. The method matches `sendrawtransaction`'s defaults: `MEMPOOL_AND_BROADCAST_TO_ALL` (broadcastMethod = 0) and `DEFAULT_MAX_RAW_TX_FEE = 0.10 BTC`. On failure the error string from the node is propagated. The non-IPC path keeps the original `send_raw_transaction` RPC call, so this is a no-op for current deployments. Verified end-to-end: the `payto & broadcast` step of tests/run-ipc.sh exercises the new path and reports === PASSED (IPC) ===. --- src/daemon.rs | 6 ++++++ src/ipc.rs | 36 ++++++++++++++++++++++++++++++++++++ 2 files changed, 42 insertions(+) diff --git a/src/daemon.rs b/src/daemon.rs index bd7ba08e4..028c984ff 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -187,6 +187,12 @@ impl Daemon { } pub(crate) fn broadcast(&self, tx: &Transaction) -> Result { + if let Some(ipc) = &self.ipc { + let bytes = bitcoin::consensus::serialize(tx); + ipc.broadcast_transaction(bytes) + .context("failed to broadcast transaction via IPC")?; + return Ok(tx.compute_txid()); + } self.rpc .send_raw_transaction(tx) .context("failed to broadcast transaction") diff --git a/src/ipc.rs b/src/ipc.rs index d2e9fe8a3..4e94ff98d 100644 --- a/src/ipc.rs +++ b/src/ipc.rs @@ -174,6 +174,42 @@ impl IpcChain { .boxed_local() }) } + + /// Broadcast a serialized transaction via `Chain.broadcastTransaction`, + /// using the same `MEMPOOL_AND_BROADCAST_TO_ALL` semantics as the + /// JSON-RPC `sendrawtransaction` call. `max_tx_fee` matches the RPC + /// default (`DEFAULT_MAX_RAW_TX_FEE = 0.10 BTC`). On failure the node's + /// error string is propagated. + pub(crate) fn broadcast_transaction(&self, tx_bytes: Vec) -> Result<()> { + // node::TxBroadcast::MEMPOOL_AND_BROADCAST_TO_ALL = 0 + const BROADCAST_METHOD_DEFAULT: i32 = 0; + // src/policy/policy.h: DEFAULT_MAX_RAW_TX_FEE = 0.10 * COIN + const MAX_TX_FEE_DEFAULT: i64 = 10_000_000; + self.call(move |ctx| { + async move { + let mut req = ctx.chain.broadcast_transaction_request(); + req.get().get_context()?.set_thread(ctx.thread.clone()); + { + let mut params = req.get(); + params.set_tx(&tx_bytes); + params.set_max_tx_fee(MAX_TX_FEE_DEFAULT); + params.set_broadcast_method(BROADCAST_METHOD_DEFAULT); + } + let resp = req.send().promise.await?; + let r = resp.get()?; + if r.get_result() { + Ok(()) + } else { + let err = r.get_error()?.to_str()?; + if err.is_empty() { + bail!("Chain.broadcastTransaction failed"); + } + bail!("Chain.broadcastTransaction failed: {err}"); + } + } + .boxed_local() + }) + } } /// Worker thread entry point. Owns the tokio runtime, the capnp RPC system, From 800108197219a77c66ec5ded49452e7ca9395470 Mon Sep 17 00:00:00 2001 From: Sjors Provoost Date: Tue, 12 May 2026 18:45:47 +0200 Subject: [PATCH 07/21] experimental: header sync via Chain.findLocatorFork + findBlock Add IpcChain::find_locator_fork (encodes a CBlockLocator on the wire) and IpcChain::get_new_headers, which walks forward from the fork point using getBlockHash + findBlock(wantData=true) and parses the first 80 bytes of each block as a header. Cap the response at 2000 headers per call to mirror the P2P getheaders semantics. When --daemon-ipc-socket is set, Daemon::get_new_headers now goes through this IPC path instead of the P2P getheaders exchange in src/p2p.rs. Behaviour without IPC is unchanged. The chain interface does not expose a header-only fetch, so this downloads each full block twice (once for the header, again during the subsequent for_blocks indexing pass). Acceptable for an experimental backend over a local unix socket. --- src/daemon.rs | 5 +++ src/ipc.rs | 111 ++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 116 insertions(+) diff --git a/src/daemon.rs b/src/daemon.rs index 028c984ff..f6cca2b00 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -323,6 +323,11 @@ impl Daemon { } pub(crate) fn get_new_headers(&self, chain: &Chain) -> Result> { + if let Some(ipc) = &self.ipc { + return ipc + .get_new_headers(chain) + .context("IPC get_new_headers failed"); + } self.p2p.lock().get_new_headers(chain) } diff --git a/src/ipc.rs b/src/ipc.rs index 4e94ff98d..143127881 100644 --- a/src/ipc.rs +++ b/src/ipc.rs @@ -17,6 +17,7 @@ use std::pin::Pin; use std::thread as stdthread; use anyhow::{anyhow, bail, Context, Result}; +use bitcoin::blockdata::block::Header as BlockHeader; use bitcoin::{consensus::Decodable, hashes::Hash, BlockHash}; use bitcoin_capnp_types::{ chain_capnp::chain, @@ -29,6 +30,7 @@ use tokio::net::UnixStream; use tokio::sync::{mpsc, oneshot}; use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt}; +use crate::chain::{Chain, NewHeader}; use crate::types::SerBlock; /// Boxed (non-Send) future returned by a job. The closure itself must be @@ -210,6 +212,115 @@ impl IpcChain { .boxed_local() }) } + + /// Find the height of the highest block in `locator` that is part of the + /// node's active chain, via `Chain.findLocatorFork`. Returns `None` if no + /// hash in the locator is on the active chain (e.g. completely diverged + /// chain), in which case the caller should fall back to walking from + /// genesis. + /// + /// `locator` is the list of block hashes returned by [`Chain::locator`], + /// ordered tip→genesis with exponentially increasing gaps. + pub(crate) fn find_locator_fork(&self, locator: &[BlockHash]) -> Result> { + // Bitcoin Core CBlockLocator wire format: int32 LE version + // (DUMMY_VERSION = 70016) + CompactSize count + 32*count hashes. + let mut buf = Vec::with_capacity(4 + 9 + locator.len() * 32); + buf.extend_from_slice(&70016i32.to_le_bytes()); + write_compact_size(&mut buf, locator.len() as u64); + for hash in locator { + buf.extend_from_slice(hash.as_byte_array()); + } + self.call(move |ctx| { + async move { + let mut req = ctx.chain.find_locator_fork_request(); + req.get().get_context()?.set_thread(ctx.thread.clone()); + req.get().set_locator(&buf); + let resp = req.send().promise.await?; + let r = resp.get()?; + Ok(if r.get_has_result() { + Some(r.get_result()) + } else { + None + }) + } + .boxed_local() + }) + } + + /// Replacement for the P2P `getheaders` exchange in [`Daemon::get_new_headers`]. + /// + /// Walks the IPC chain forward from the fork point between the locator + /// (taken from `chain`) and the node's active chain, returning every new + /// header up to the node's current tip. Each header is fetched by way of + /// `findBlock(wantData=true)` followed by parsing the first 80 bytes as a + /// `BlockHeader`; the chain interface does not expose a header-only fetch, + /// so the full block payload is downloaded and the body is discarded. + /// The returned headers are immediately re-fetched in full by the + /// subsequent `for_blocks` indexing pass — acceptable for an experimental + /// backend over a local unix socket, but worth noting. + /// + /// Caps the response at 2000 headers per call to mirror the P2P + /// `getheaders` semantics, so the caller can drive multiple iterations + /// during initial sync. + pub(crate) fn get_new_headers(&self, chain: &Chain) -> Result> { + const MAX_HEADERS: usize = 2_000; + + let tip_height = match self.get_height()? { + Some(h) if h >= 0 => h as usize, + _ => return Ok(vec![]), + }; + + // Find the fork height between our chain's locator and the node's + // active chain. + let locator = chain.locator(); + let fork_height = self + .find_locator_fork(&locator)? + .ok_or_else(|| anyhow!("Chain.findLocatorFork: no common ancestor"))?; + if fork_height < 0 { + bail!("Chain.findLocatorFork returned negative fork height {fork_height}"); + } + let fork_height = fork_height as usize; + + if tip_height <= fork_height { + return Ok(vec![]); + } + + let new_first = fork_height + 1; + let new_last = tip_height.min(fork_height + MAX_HEADERS); + let mut headers = Vec::with_capacity(new_last - new_first + 1); + for h in new_first..=new_last { + let hash = self.get_block_hash(h as i32)?; + let block = self + .get_block(hash)? + .ok_or_else(|| anyhow!("findBlock: missing block {hash} at height {h}"))?; + if block.len() < 80 { + bail!( + "findBlock returned {} bytes for {hash}; need >=80", + block.len() + ); + } + let header = BlockHeader::consensus_decode(&mut &block[..80]) + .with_context(|| format!("parsing header at height {h}"))?; + headers.push(NewHeader::from((header, h))); + } + Ok(headers) + } +} + +/// Bitcoin Core CompactSize encoding (src/serialize.h). +fn write_compact_size(buf: &mut Vec, n: u64) { + if n < 253 { + buf.push(n as u8); + } else if n <= u16::MAX as u64 { + buf.push(253); + buf.extend_from_slice(&(n as u16).to_le_bytes()); + } else if n <= u32::MAX as u64 { + buf.push(254); + buf.extend_from_slice(&(n as u32).to_le_bytes()); + } else { + buf.push(255); + buf.extend_from_slice(&n.to_le_bytes()); + } } /// Worker thread entry point. Owns the tokio runtime, the capnp RPC system, From 90ecc08f737ba224d0c6c4cb966c977107e43110 Mon Sep 17 00:00:00 2001 From: Sjors Provoost Date: Tue, 12 May 2026 18:54:45 +0200 Subject: [PATCH 08/21] experimental: tip-change notifications via Chain.waitForNotificationsIfTipChanged Restructure the IPC worker to spawn each Job onto the LocalSet so a long-running notifier task can run concurrently with regular RPC calls. Ctx now derives Clone (its two capnp clients are Rc-backed and cheap to clone) and is passed by value into each Job. Add IpcChain::start_block_notifier, which kicks off a worker-side loop that snapshots the current tip via getHeight + getBlockHash, blocks on Chain.waitForNotificationsIfTipChanged, and signals on a bounded(1) crossbeam channel each time the tip changes. Daemon::connect now starts this notifier when an IPC socket is configured and stashes the receiver on the Daemon. Daemon::new_block_notification returns the IPC receiver instead of the P2P new-block channel when IPC is available, eliminating the remaining P2P consumer for new-block notifications. Behaviour without IPC is unchanged. --- src/daemon.rs | 25 +++++++++++- src/ipc.rs | 103 ++++++++++++++++++++++++++++++++++++++++++++++---- 2 files changed, 120 insertions(+), 8 deletions(-) diff --git a/src/daemon.rs b/src/daemon.rs index f6cca2b00..fbbc0ed3b 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -104,6 +104,11 @@ pub struct Daemon { p2p: Mutex, rpc: Client, ipc: Option, + /// When IPC is configured, a long-running task on the IPC worker watches + /// `Chain.waitForNotificationsIfTipChanged` and signals on this channel + /// each time the tip changes. Used in place of the P2P `inv`-watching + /// path for `new_block_notification`. + ipc_block_notifier: Option>, } impl Daemon { @@ -163,7 +168,22 @@ impl Daemon { bail!("electrs requires non-pruned bitcoind node"); } - Ok(Self { p2p, rpc, ipc }) + // Start the IPC tip-change notifier if available, so we can avoid the + // P2P `inv`-watching path for new-block notifications. + let ipc_block_notifier = match &ipc { + Some(ipc) => Some( + ipc.start_block_notifier() + .context("failed to start IPC block notifier")?, + ), + None => None, + }; + + Ok(Self { + p2p, + rpc, + ipc, + ipc_block_notifier, + }) } pub(crate) fn estimate_fee(&self, nblocks: u16) -> Result> { @@ -355,6 +375,9 @@ impl Daemon { } pub(crate) fn new_block_notification(&self) -> Receiver<()> { + if let Some(rx) = &self.ipc_block_notifier { + return rx.clone(); + } self.p2p.lock().new_block_notification() } } diff --git a/src/ipc.rs b/src/ipc.rs index 143127881..c7b8e2018 100644 --- a/src/ipc.rs +++ b/src/ipc.rs @@ -13,8 +13,8 @@ #![allow(dead_code)] use std::path::{Path, PathBuf}; -use std::pin::Pin; use std::thread as stdthread; +use std::time::Duration; use anyhow::{anyhow, bail, Context, Result}; use bitcoin::blockdata::block::Header as BlockHeader; @@ -25,6 +25,7 @@ use bitcoin_capnp_types::{ proxy_capnp::{thread as proxy_thread, thread_map}, }; use capnp_rpc::{rpc_twoparty_capnp::Side, twoparty::VatNetwork, RpcSystem}; +use crossbeam_channel::{bounded, Receiver}; use futures::{future::LocalBoxFuture, io::BufReader, FutureExt}; use tokio::net::UnixStream; use tokio::sync::{mpsc, oneshot}; @@ -36,11 +37,15 @@ use crate::types::SerBlock; /// Boxed (non-Send) future returned by a job. The closure itself must be /// `Send` so it can be shipped from caller threads to the IPC worker, but /// once on the worker it runs on a single-threaded LocalSet, so the future -/// it produces does not need to be `Send`. -type Job = Box FnOnce(&'a Ctx) -> LocalBoxFuture<'a, ()> + Send + 'static>; +/// it produces does not need to be `Send`. Each job receives an owned `Ctx` +/// (a cheap pair of `Rc`-backed capnp clients), letting the worker spawn +/// jobs concurrently on the LocalSet without lifetime gymnastics. +type Job = Box LocalBoxFuture<'static, ()> + Send + 'static>; /// IPC context held by the worker thread: the chain client and the proxy -/// thread handle that every call must reference. +/// thread handle that every call must reference. Cloning is cheap (both +/// fields are `Rc`-backed capnp clients). +#[derive(Clone)] struct Ctx { chain: chain::Client, thread: proxy_thread::Client, @@ -74,7 +79,7 @@ impl IpcChain { fn call(&self, f: F) -> Result where R: Send + 'static, - F: for<'a> FnOnce(&'a Ctx) -> LocalBoxFuture<'a, Result> + Send + 'static, + F: FnOnce(Ctx) -> LocalBoxFuture<'static, Result> + Send + 'static, { let (otx, orx) = oneshot::channel::>(); let job: Job = Box::new(move |ctx| { @@ -305,6 +310,89 @@ impl IpcChain { } Ok(headers) } + + /// Spawn a long-running task on the worker that watches for tip changes + /// via `Chain.waitForNotificationsIfTipChanged`, and returns a receiver + /// that fires once per detected tip change. + /// + /// This replaces the P2P `inv`-watching path in `src/p2p.rs` for IPC + /// deployments. The receiver is bounded to capacity 1 with `try_send` + /// semantics: if the consumer is busy, additional notifications are + /// coalesced into the pending one. + pub(crate) fn start_block_notifier(&self) -> Result> { + let (notif_tx, notif_rx) = bounded::<()>(1); + let job: Job = Box::new(move |ctx| { + async move { + loop { + // Snapshot the current tip hash so we can ask the node + // to wake us when it changes. + let tip_hash = match snapshot_tip_hash(&ctx).await { + Ok(h) => h, + Err(e) => { + warn!("IPC notifier: tip snapshot failed: {e:#}"); + tokio::time::sleep(Duration::from_secs(5)).await; + continue; + } + }; + + // Block until the tip differs from `tip_hash`. Returns + // immediately if it already does. + let mut req = ctx.chain.wait_for_notifications_if_tip_changed_request(); + match req.get().get_context() { + Ok(mut c) => c.set_thread(ctx.thread.clone()), + Err(e) => { + warn!("IPC notifier: failed to set thread: {e:#}"); + tokio::time::sleep(Duration::from_secs(5)).await; + continue; + } + } + req.get().set_old_tip(&tip_hash); + if let Err(e) = req.send().promise.await { + warn!("IPC notifier: waitForNotifications failed: {e:#}"); + tokio::time::sleep(Duration::from_secs(5)).await; + continue; + } + + // Coalesce: if the receiver is already pending, drop this + // signal (the indexer will see all changes on its next + // sync iteration regardless). + let _ = notif_tx.try_send(()); + } + } + .boxed_local() + }); + self.tx + .send(job) + .map_err(|_| anyhow!("IPC worker thread is gone"))?; + Ok(notif_rx) + } +} + +/// Helper for the notifier loop: read the current tip hash via +/// `getHeight` + `getBlockHash`. Returns the all-zeros hash when the node +/// reports no tip yet (genesis only), which still works for +/// `waitForNotificationsIfTipChanged` (it'll wake us as soon as a real +/// tip exists). +async fn snapshot_tip_hash(ctx: &Ctx) -> Result<[u8; 32]> { + let mut req = ctx.chain.get_height_request(); + req.get().get_context()?.set_thread(ctx.thread.clone()); + let resp = req.send().promise.await?; + let r = resp.get()?; + if !r.get_has_result() { + return Ok([0u8; 32]); + } + let height = r.get_result(); + let mut req = ctx.chain.get_block_hash_request(); + req.get().get_context()?.set_thread(ctx.thread.clone()); + req.get().set_height(height); + let resp = req.send().promise.await?; + let bytes = resp.get()?.get_result()?; + if bytes.len() != 32 { + bail!("unexpected block hash length {}", bytes.len()); + } + let mut buf = [0u8; 32]; + buf.copy_from_slice(bytes); + Ok(buf) } /// Bitcoin Core CompactSize encoding (src/serialize.h). @@ -351,8 +439,9 @@ fn worker( }; let _ = ready.send(Ok(())); while let Some(job) = rx.recv().await { - let fut: Pin>> = job(&ctx); - fut.await; + // Spawn each job onto the LocalSet so long-running jobs (e.g. the + // block-tip notifier loop) do not block other in-flight calls. + tokio::task::spawn_local(job(ctx.clone())); } })); } From 82607f05cec99d659767ab8564d2b0b903043a1a Mon Sep 17 00:00:00 2001 From: Sjors Provoost Date: Tue, 12 May 2026 18:54:54 +0200 Subject: [PATCH 09/21] tests: clean shutdown for regtest fixtures MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two unrelated cleanup-noise fixes in the integration scripts: - tests/run.sh: `bitcoin-cli stop` was sometimes racing the node's own exit and leaving a 'Could not connect to the server' line on stdout. Make the stop best-effort and fall back to killing the PID. - tests/run-ipc.sh: when the multiprocess node has IPC clients still attached at shutdown, the RPC `stop` request hangs indefinitely waiting for them to disconnect. The IPC client lives in electrs, which has already stopped above but whose capnp connection drop is not flushed back to the node. Skip the polite stop and just kill the node — this is a regtest fixture. --- tests/run-ipc.sh | 9 +++++++-- tests/run.sh | 6 ++++-- 2 files changed, 11 insertions(+), 4 deletions(-) diff --git a/tests/run-ipc.sh b/tests/run-ipc.sh index 14d0b3723..090065c44 100755 --- a/tests/run-ipc.sh +++ b/tests/run-ipc.sh @@ -143,7 +143,12 @@ kill -INT $ELECTRS_PID # close server tail_log data/electrs/regtest-debug.log | grep -m1 "electrs stopped" wait $ELECTRS_PID -$BTC stop # stop bitcoind -wait $BITCOIND_PID +# When the multiprocess node has IPC clients still attached at shutdown, the +# RPC `stop` request hangs indefinitely waiting for them to disconnect. The +# IPC client lives in electrs, which has already stopped above but whose +# capnp connection drop is not flushed back to the node. Skip the polite +# stop and just kill the node — this is a regtest fixture. +kill $BITCOIND_PID 2>/dev/null || true +wait $BITCOIND_PID 2>/dev/null || true echo "=== PASSED (IPC) ===" diff --git a/tests/run.sh b/tests/run.sh index 1e2c03c1f..2f46382f4 100755 --- a/tests/run.sh +++ b/tests/run.sh @@ -110,7 +110,9 @@ kill -INT $ELECTRS_PID # close server tail_log data/electrs/regtest-debug.log | grep -m1 "electrs stopped" wait $ELECTRS_PID -$BTC stop # stop bitcoind -wait $BITCOIND_PID +# Try a graceful stop; if the node has already exited the RPC call will fail, +# which is fine. +$BTC stop 2>/dev/null || kill $BITCOIND_PID 2>/dev/null || true +wait $BITCOIND_PID 2>/dev/null || true echo "=== PASSED ===" From 39c800d3fe2d2472ad8c60610607f73ee44dde3e Mon Sep 17 00:00:00 2001 From: Sjors Provoost Date: Tue, 12 May 2026 18:58:53 +0200 Subject: [PATCH 10/21] experimental: skip P2P connection entirely when IPC is configured Now that header sync, block fetch, and new-block notifications all go through the Chain interface when --daemon-ipc-socket is set, there is no consumer left for the P2P connection. Make Daemon.p2p an Option and skip Connection::connect (and the network_active enforcement) on the IPC path. The non-IPC path is unchanged. --- src/daemon.rs | 46 ++++++++++++++++++++++++++++++++++------------ 1 file changed, 34 insertions(+), 12 deletions(-) diff --git a/src/daemon.rs b/src/daemon.rs index fbbc0ed3b..425c7dd4a 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -101,7 +101,11 @@ fn rpc_connect(config: &Config) -> Result { } pub struct Daemon { - p2p: Mutex, + /// P2P connection to bitcoind. Only constructed when IPC is *not* + /// configured; with IPC, header sync, block fetch, and new-block + /// notifications all go through the Chain interface and the P2P + /// connection is unnecessary. + p2p: Option>, rpc: Client, ipc: Option, /// When IPC is configured, a long-running task on the IPC worker watches @@ -138,15 +142,7 @@ impl Daemon { if network_info.version < 21_00_00 { bail!("electrs requires bitcoind 0.21+"); } - if !network_info.network_active { - bail!("electrs requires active bitcoind p2p network"); - } - let p2p = Mutex::new(Connection::connect( - config.daemon_p2p_addr, - metrics, - config.magic, - )?); let ipc = match config.daemon_ipc_socket.as_deref() { Some(path) => { info!( @@ -158,6 +154,23 @@ impl Daemon { None => None, }; + // Skip the P2P connection entirely when IPC is available: header + // sync, block fetch, and new-block notifications all go through the + // Chain interface in that case. Without IPC we still need P2P, so + // also enforce that bitcoind has its P2P network enabled. + let p2p = if ipc.is_some() { + None + } else { + if !network_info.network_active { + bail!("electrs requires active bitcoind p2p network"); + } + Some(Mutex::new(Connection::connect( + config.daemon_p2p_addr, + metrics, + config.magic, + )?)) + }; + // Check the node is non-pruned, preferring the IPC chain interface // over the JSON-RPC `getblockchaininfo` call when it is available. let pruned = match &ipc { @@ -348,7 +361,7 @@ impl Daemon { .get_new_headers(chain) .context("IPC get_new_headers failed"); } - self.p2p.lock().get_new_headers(chain) + self.p2p().lock().get_new_headers(chain) } pub(crate) fn for_blocks(&self, blockhashes: B, mut func: F) -> Result<()> @@ -370,7 +383,7 @@ impl Daemon { } Ok(()) } else { - self.p2p.lock().for_blocks(blockhashes, func) + self.p2p().lock().for_blocks(blockhashes, func) } } @@ -378,7 +391,16 @@ impl Daemon { if let Some(rx) = &self.ipc_block_notifier { return rx.clone(); } - self.p2p.lock().new_block_notification() + self.p2p().lock().new_block_notification() + } + + /// Accessor for the P2P connection. Panics if called when IPC is + /// configured (which is a programmer error: the IPC fast paths above + /// must be checked first). + fn p2p(&self) -> &Mutex { + self.p2p + .as_ref() + .expect("P2P connection accessed but only IPC is configured") } } From 6b12952ff9d8f47ef2e82fbb0fcf33b993d81447 Mon Sep 17 00:00:00 2001 From: Sjors Provoost Date: Tue, 12 May 2026 19:10:15 +0200 Subject: [PATCH 11/21] ipc: cache header-fetch blocks for the immediate for_blocks pass The Chain interface exposes no header-only fetch, so get_new_headers has to download each full block via findBlock(wantData=true) just to keep the first 80 bytes. Without a cache, the for_blocks pass that immediately follows refetches every one of those blocks over IPC -- doubling the byte count of initial sync, and making it a worse choice than P2P for any chain longer than a handful of blocks. Add a small bounded cache (256 MiB worth of payload, best-effort inserts on overflow) on IpcChain. get_new_headers stashes each block it fetches; get_block consults the cache and removes on hit, so for_blocks drains it in iteration order. Cache size is naturally bounded by the in-flight headers (capped at 100 per get_new_headers call -- much smaller than the P2P getheaders 2000, since each "header" here costs a full block, and mainnet blocks of up to ~4 MiB would otherwise blow the cache budget on a single batch) and falls back gracefully to direct fetch on overflow. A header-only chain method on the Bitcoin Core side would obviate this entirely; noted as a follow-up in EXPERIMENT.md. --- src/ipc.rs | 108 +++++++++++++++++++++++++++++++++++++++++++++++------ 1 file changed, 96 insertions(+), 12 deletions(-) diff --git a/src/ipc.rs b/src/ipc.rs index c7b8e2018..336b1d463 100644 --- a/src/ipc.rs +++ b/src/ipc.rs @@ -12,7 +12,9 @@ //! type-erased async closures returning their results through a oneshot. #![allow(dead_code)] +use std::collections::HashMap; use std::path::{Path, PathBuf}; +use std::sync::Arc; use std::thread as stdthread; use std::time::Duration; @@ -27,6 +29,7 @@ use bitcoin_capnp_types::{ use capnp_rpc::{rpc_twoparty_capnp::Side, twoparty::VatNetwork, RpcSystem}; use crossbeam_channel::{bounded, Receiver}; use futures::{future::LocalBoxFuture, io::BufReader, FutureExt}; +use parking_lot::Mutex; use tokio::net::UnixStream; use tokio::sync::{mpsc, oneshot}; use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt}; @@ -52,10 +55,63 @@ struct Ctx { } /// Sync handle to the IPC worker thread. Cloneable; clones share the same -/// connection. +/// connection and the same header-fetch block cache. #[derive(Clone)] pub(crate) struct IpcChain { tx: mpsc::UnboundedSender, + /// Blocks that were downloaded by [`IpcChain::get_new_headers`] (which + /// has to fetch the full block payload over IPC just to extract the 80 + /// header bytes, since the Chain interface exposes no header-only + /// fetch). The very next thing electrs's indexing pipeline does is call + /// [`Daemon::for_blocks`] over the same hashes, so handing those + /// payloads back from the cache avoids a duplicate IPC round-trip per + /// block. Bounded by total bytes; on overflow, inserts are skipped + /// (best-effort) so the cache never grows without bound during initial + /// sync. + header_fetch_cache: Arc>, +} + +/// In-memory cache of recently fetched blocks, drained by +/// [`IpcChain::get_block`] on hit. Keyed by block hash, bounded by total +/// payload bytes. +#[derive(Default)] +struct BlockCache { + map: HashMap, + bytes: usize, +} + +/// Soft upper bound on the bytes held in the header-fetch cache. Sized to +/// comfortably hold one [`IpcChain::get_new_headers`] batch (capped at +/// `MAX_HEADERS` blocks) under typical mainnet block sizes (~2 MiB), with +/// headroom for a few outliers; oversized batches simply skip insertion +/// and refetch on demand. +const HEADER_FETCH_CACHE_MAX_BYTES: usize = 256 * 1024 * 1024; + +impl BlockCache { + /// Best-effort insert. Skips the block if the cache is already full or + /// if the block alone exceeds the budget; the caller's get_block path + /// will simply refetch on the (rare) miss that follows. + fn try_insert(&mut self, hash: BlockHash, block: SerBlock) { + let block_bytes = block.len(); + if block_bytes > HEADER_FETCH_CACHE_MAX_BYTES { + return; + } + if self.map.contains_key(&hash) { + return; + } + if self.bytes + block_bytes > HEADER_FETCH_CACHE_MAX_BYTES { + return; + } + self.bytes += block_bytes; + self.map.insert(hash, block); + } + + /// Remove and return a cached block, decreasing the byte count. + fn take(&mut self, hash: &BlockHash) -> Option { + let block = self.map.remove(hash)?; + self.bytes = self.bytes.saturating_sub(block.len()); + Some(block) + } } impl IpcChain { @@ -72,7 +128,10 @@ impl IpcChain { ready_rx .recv() .context("IPC worker thread exited before signalling readiness")??; - Ok(Self { tx }) + Ok(Self { + tx, + header_fetch_cache: Arc::new(Mutex::new(BlockCache::default())), + }) } /// Submit a job to the worker and block on its result. @@ -158,7 +217,22 @@ impl IpcChain { /// Fetch a serialized block by hash via `Chain.findBlock(wantData=true)`. /// Returns `Ok(None)` if the block is not known to the node. + /// + /// Consults the header-fetch cache first: blocks downloaded earlier by + /// [`Self::get_new_headers`] are handed back without an IPC round-trip, + /// and removed from the cache (consume-once semantics). pub(crate) fn get_block(&self, hash: BlockHash) -> Result> { + if let Some(block) = self.header_fetch_cache.lock().take(&hash) { + return Ok(Some(block)); + } + self.fetch_block(hash) + } + + /// Unconditional `findBlock(wantData=true)` fetch with no cache + /// interaction; used internally by [`Self::get_block`] (on cache miss) + /// and by [`Self::get_new_headers`] (which inserts the result into the + /// cache for the immediately-following `for_blocks` pass). + fn fetch_block(&self, hash: BlockHash) -> Result> { let raw: [u8; 32] = hash.to_byte_array(); self.call(move |ctx| { async move { @@ -258,17 +332,23 @@ impl IpcChain { /// (taken from `chain`) and the node's active chain, returning every new /// header up to the node's current tip. Each header is fetched by way of /// `findBlock(wantData=true)` followed by parsing the first 80 bytes as a - /// `BlockHeader`; the chain interface does not expose a header-only fetch, - /// so the full block payload is downloaded and the body is discarded. - /// The returned headers are immediately re-fetched in full by the - /// subsequent `for_blocks` indexing pass — acceptable for an experimental - /// backend over a local unix socket, but worth noting. + /// `BlockHeader`; the chain interface does not expose a header-only + /// fetch, so the full block payload is downloaded. To avoid an immediate + /// duplicate fetch, the payload is stashed in the header-fetch cache and + /// handed back to the subsequent `for_blocks` indexing pass. /// - /// Caps the response at 2000 headers per call to mirror the P2P - /// `getheaders` semantics, so the caller can drive multiple iterations - /// during initial sync. + /// Caps the response at [`MAX_HEADERS`] per call so the caller can drive + /// multiple iterations during initial sync. The cap is small (vs. + /// the 2000-header P2P `getheaders` batch) for two reasons: (1) each + /// header costs a full block over IPC until upstream grows a header-only + /// fetch, so larger batches blow past the in-process block cache budget + /// on mainnet; (2) over a local unix socket, round-trip cost is dominated + /// by per-block work, not per-batch overhead, so the historical P2P + /// argument for huge batches doesn't carry over. pub(crate) fn get_new_headers(&self, chain: &Chain) -> Result> { - const MAX_HEADERS: usize = 2_000; + /// Headers per `get_new_headers` call. See the function-level + /// comment above for why this is much smaller than the P2P value. + const MAX_HEADERS: usize = 100; let tip_height = match self.get_height()? { Some(h) if h >= 0 => h as usize, @@ -296,7 +376,7 @@ impl IpcChain { for h in new_first..=new_last { let hash = self.get_block_hash(h as i32)?; let block = self - .get_block(hash)? + .fetch_block(hash)? .ok_or_else(|| anyhow!("findBlock: missing block {hash} at height {h}"))?; if block.len() < 80 { bail!( @@ -307,6 +387,10 @@ impl IpcChain { let header = BlockHeader::consensus_decode(&mut &block[..80]) .with_context(|| format!("parsing header at height {h}"))?; headers.push(NewHeader::from((header, h))); + // Stash the payload for the for_blocks pass that will follow + // immediately. Best-effort: when the cache is full, additional + // blocks will simply be re-fetched on demand. + self.header_fetch_cache.lock().try_insert(hash, block); } Ok(headers) } From 61935e4e15d75723aa05eb15d75b009efcb517b3 Mon Sep 17 00:00:00 2001 From: Sjors Provoost Date: Tue, 12 May 2026 20:13:10 +0200 Subject: [PATCH 12/21] ci: also run on push to non-master branches Drop the `branches: [master]` filter on the `push` trigger so CI runs on every push, not just pushes to master. This makes the workflow fire for branch pushes on forks (e.g. Sjors/electrs experimental branches), matching the way most contributors iterate before opening a PR. PR triggers are unaffected. --- .github/workflows/rust.yml | 2 -- 1 file changed, 2 deletions(-) diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index 6f2fb902d..b43205f0e 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -2,8 +2,6 @@ name: electrs on: push: - branches: - - master pull_request: schedule: - cron: "0 0 * * *" # once a day From da169ad309b6799aad4b0e8f56c03a6c43867828 Mon Sep 17 00:00:00 2001 From: Sjors Provoost Date: Tue, 12 May 2026 20:13:35 +0200 Subject: [PATCH 13/21] daemon: route fee estimation through Chain.{estimateSmartFee,relayMinFee} on IPC Adds two thin wrappers on `IpcChain`: - `estimate_smart_fee_sat_per_kvb(nblocks)` calls `Chain.estimateSmartFee` (economical mode, no FeeCalculation), and - `relay_min_fee_sat_per_kvb()` calls `Chain.relayMinFee`. Both decode the serialized `CFeeRate` reply (FeeFrac { int64_t fee; int32_t size; } -> 12 bytes LE, per Bitcoin Core SERIALIZE_METHODS) into satoshis per kilo-vbyte via a new `decode_fee_rate_kvb` helper. A `CFeeRate{}` (size == 0) reply maps to `Ok(None)` so the smart-fee path can mirror the JSON-RPC -32603 "no estimate" behaviour. `Daemon::estimate_fee` and `Daemon::get_relay_fee` now prefer the IPC path when `ipc` is set and fall back to the bitcoind JSON-RPC client otherwise. This drops electrs's dependence on `getnetworkinfo.relayfee` and `estimatesmartfee` for IPC deployments. Verified the `*_returns_default` and `*_returns_decodable_blob` Chain integration tests in bitcoin-capnp-types pass against the same multiprocess node electrs talks to. --- src/daemon.rs | 22 ++++++++++++++++ src/ipc.rs | 69 +++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 91 insertions(+) diff --git a/src/daemon.rs b/src/daemon.rs index 425c7dd4a..893a19ac0 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -200,6 +200,18 @@ impl Daemon { } pub(crate) fn estimate_fee(&self, nblocks: u16) -> Result> { + if let Some(ipc) = &self.ipc { + // Chain.estimateSmartFee returns CFeeRate{} (size==0) when no + // estimate is available, which mirrors the JSON-RPC -32603 + // ("Insufficient data or no feerate found") behaviour we map to + // None below. + let sat_per_kvb = ipc + .estimate_smart_fee_sat_per_kvb(nblocks.into()) + .context("failed to estimate fee via IPC")?; + return Ok(sat_per_kvb + .filter(|n| *n > 0) + .map(|n| Amount::from_sat(n as u64))); + } let res = self.rpc.estimate_smart_fee(nblocks, None); if let Err(bitcoincore_rpc::Error::JsonRpc(jsonrpc::Error::Rpc(RpcError { code: -32603, @@ -212,6 +224,16 @@ impl Daemon { } pub(crate) fn get_relay_fee(&self) -> Result { + if let Some(ipc) = &self.ipc { + let sat_per_kvb = ipc + .relay_min_fee_sat_per_kvb() + .context("failed to fetch relay min fee via IPC")?; + // sat_per_kvb is non-negative in practice (the node clamps the + // configured -minrelaytxfee at 0); guard against a corrupt or + // hostile reply by saturating at 0 rather than panicking on the + // i64 -> u64 conversion. + return Ok(Amount::from_sat(sat_per_kvb.max(0) as u64)); + } Ok(self .rpc .get_network_info() diff --git a/src/ipc.rs b/src/ipc.rs index 336b1d463..2745dddd4 100644 --- a/src/ipc.rs +++ b/src/ipc.rs @@ -292,6 +292,54 @@ impl IpcChain { }) } + /// Get the node's minimum relay feerate via `Chain.relayMinFee`. The + /// reply is a serialized `CFeeRate` blob (see [`decode_fee_rate_kvb`]); + /// this method returns the rate as satoshis per kilo-vbyte. + /// + /// Replaces the JSON-RPC `getnetworkinfo.relayfee` lookup used by + /// [`Daemon::get_relay_fee`] when IPC is configured. + pub(crate) fn relay_min_fee_sat_per_kvb(&self) -> Result { + self.call(|ctx| { + async move { + let mut req = ctx.chain.relay_min_fee_request(); + req.get().get_context()?.set_thread(ctx.thread.clone()); + let resp = req.send().promise.await?; + let blob = resp.get()?.get_result()?; + Ok(decode_fee_rate_kvb(blob)?.unwrap_or(0)) + } + .boxed_local() + }) + } + + /// Smart fee estimate for confirmation within `nblocks`, via + /// `Chain.estimateSmartFee`. Returns `Ok(None)` if the node has no + /// estimate available (a `CFeeRate{}` with `size == 0`), otherwise the + /// rate as satoshis per kilo-vbyte. + /// + /// Replaces the JSON-RPC `estimatesmartfee` call used by + /// [`Daemon::estimate_fee`] when IPC is configured. + pub(crate) fn estimate_smart_fee_sat_per_kvb(&self, nblocks: i32) -> Result> { + self.call(move |ctx| { + async move { + let mut req = ctx.chain.estimate_smart_fee_request(); + req.get().get_context()?.set_thread(ctx.thread.clone()); + { + let mut params = req.get(); + params.set_num_blocks(nblocks); + // Match the JSON-RPC default for `estimatesmartfee`: + // economical (non-conservative) mode. + params.set_conservative(false); + // We don't surface FeeCalculation diagnostics to clients. + params.set_want_calc(false); + } + let resp = req.send().promise.await?; + let blob = resp.get()?.get_result()?; + decode_fee_rate_kvb(blob) + } + .boxed_local() + }) + } + /// Find the height of the highest block in `locator` that is part of the /// node's active chain, via `Chain.findLocatorFork`. Returns `None` if no /// hash in the locator is on the active chain (e.g. completely diverged @@ -479,6 +527,27 @@ async fn snapshot_tip_hash(ctx: &Ctx) -> Result<[u8; 32]> { Ok(buf) } +/// Decode a serialized Bitcoin Core `CFeeRate` blob into satoshis per +/// kilo-vbyte (matching `CFeeRate::GetFeePerK()`). Wire format is +/// `FeeFrac { int64_t fee; int32_t size; }` (12 bytes LE), per +/// `SERIALIZE_METHODS(CFeeRate, ...)`. Returns `Ok(None)` for the empty +/// `CFeeRate{}` case (`size == 0`), which Bitcoin Core uses to mean "no +/// estimate available". +fn decode_fee_rate_kvb(blob: &[u8]) -> Result> { + if blob.len() != 12 { + bail!( + "expected serialized CFeeRate to be 12 bytes (int64 fee + int32 size), got {}", + blob.len() + ); + } + let fee = i64::from_le_bytes(blob[0..8].try_into().expect("len 8")); + let size = i32::from_le_bytes(blob[8..12].try_into().expect("len 4")); + if size == 0 { + return Ok(None); + } + Ok(Some(fee.saturating_mul(1000) / size as i64)) +} + /// Bitcoin Core CompactSize encoding (src/serialize.h). fn write_compact_size(buf: &mut Vec, n: u64) { if n < 253 { From f2510ab8e376996b835fab7de66c923d8893944b Mon Sep 17 00:00:00 2001 From: Sjors Provoost Date: Tue, 12 May 2026 20:43:12 +0200 Subject: [PATCH 14/21] ci: fix regtest integration test reliability --- .github/workflows/rust.yml | 6 ++--- tests/run-ipc.sh | 53 +++++++++++++++++++++++++++++++------- tests/run.sh | 53 +++++++++++++++++++++++++++++++------- 3 files changed, 89 insertions(+), 23 deletions(-) diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index b43205f0e..898939c7c 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -84,7 +84,7 @@ jobs: build-essential cmake pkgconf python3 \ libevent-dev libboost-dev capnproto libcapnp-dev \ librocksdb-dev libclang-dev ccache \ - python3-venv python3-pip git jq curl libsecp256k1-2 + python3-venv python3-pip git jq curl libsecp256k1-dev - name: ccache uses: actions/cache@v4 @@ -110,9 +110,9 @@ jobs: -DCMAKE_CXX_COMPILER_LAUNCHER=ccache cmake --build build -j $(nproc) --target bitcoin bitcoin-node bitcoin-cli - - name: Install Electrum wallet + - name: Install Electrum wallet (4.7.2) run: | - git clone --depth 1 --recurse-submodules https://github.com/spesmilo/electrum/ + git clone --depth 1 --branch 4.7.2 --recurse-submodules https://github.com/spesmilo/electrum/ python3 -m venv --system-site-packages venv ELECTRUM_ECC_DONT_COMPILE=1 venv/bin/pip install -e electrum/ sudo ln -s "$PWD/venv/bin/electrum" /usr/local/bin/electrum diff --git a/tests/run-ipc.sh b/tests/run-ipc.sh index 090065c44..ab20b4e94 100755 --- a/tests/run-ipc.sh +++ b/tests/run-ipc.sh @@ -16,6 +16,7 @@ fi rm -rf data/ mkdir -p data/{bitcoin,electrum,electrs} +touch data/electrs/regtest-debug.log data/electrum/regtest-debug.log cleanup() { trap - SIGTERM SIGINT @@ -34,8 +35,44 @@ wait_for() { shift TEST_ARGS=$* for _ in `seq 0 9`; do - test "$($CMD | jq -c .)" $TEST_ARGS && break || sleep 2 + if test "$($CMD | jq -c .)" $TEST_ARGS; then + return 0 + fi + sleep 2 done + + echo "Timed out waiting for condition: $CMD $TEST_ARGS" >&2 + return 1 +} + +wait_for_log() { + LOG_FILE=$1 + PATTERN=$2 + DESCRIPTION=$3 + for _ in `seq 0 99`; do + if grep -Fq "$PATTERN" "$LOG_FILE"; then + return 0 + fi + sleep 0.2 + done + + echo "Timed out waiting for $DESCRIPTION in $LOG_FILE" >&2 + tail -n +1 "$LOG_FILE" >&2 || true + return 1 +} + +wait_for_http() { + URL=$1 + OUTPUT=$2 + for _ in `seq 0 99`; do + if curl --silent --show-error --fail "$URL" -o "$OUTPUT"; then + return 0 + fi + sleep 0.2 + done + + echo "Timed out waiting for HTTP endpoint $URL" >&2 + return 1 } BTC="$BITCOIN_MULTIPROCESS_BIN rpc -chain=regtest -datadir=$PWD/data/bitcoin" @@ -43,10 +80,6 @@ ELECTRUM="electrum --regtest" EL="$ELECTRUM --wallet=data/electrum/wallet" SOCK="$PWD/data/bitcoin/regtest/node.sock" -tail_log() { - tail -n +0 -F $1 || true -} - echo "Starting $($BITCOIN_MULTIPROCESS_BIN node -version | head -n1) (multiprocess, IPC enabled)..." $BITCOIN_MULTIPROCESS_BIN node \ -regtest -datadir=$PWD/data/bitcoin \ @@ -80,7 +113,7 @@ electrs \ --daemon-ipc-socket="$SOCK" \ 2> data/electrs/regtest-debug.log & ELECTRS_PID=$! -tail_log data/electrs/regtest-debug.log | grep -m1 "serving Electrum RPC" +wait_for_log data/electrs/regtest-debug.log "serving Electrum RPC" "electrs Electrum RPC startup" # Confirm the IPC backend actually engaged. The Daemon::connect path logs # nothing distinctive yet; instead probe the debug log for any IPC error or @@ -89,11 +122,11 @@ if ! grep -q "ipc" data/electrs/regtest-debug.log; then : # no explicit log line yet; we rely on cargo tests + functional behaviour fi -curl localhost:24224 -o metrics.txt +wait_for_http http://localhost:24224 metrics.txt $ELECTRUM daemon --server localhost:60401:t -1 -vDEBUG 2> data/electrum/regtest-debug.log & ELECTRUM_PID=$! -tail_log data/electrum/regtest-debug.log | grep -m1 "connection established" +wait_for_log data/electrum/regtest-debug.log "connection established" "Electrum daemon connection" $EL getinfo | jq . echo "Loading Electrum wallet..." @@ -125,7 +158,7 @@ $BTC getblockcount > /dev/null echo " * wait for new block" kill -USR1 $ELECTRS_PID # notify server to index new block -tail_log data/electrum/regtest-debug.log | grep -m1 "verified $TXID" > /dev/null +wait_for_log data/electrum/regtest-debug.log "verified $TXID" "Electrum wallet verification of mined transaction" echo " * get_tx_status" test "`$EL get_tx_status $TXID | jq -c .`" == '{"confirmations":1}' @@ -140,7 +173,7 @@ echo "Electrum `$EL stop`" # disconnect wallet wait $ELECTRUM_PID kill -INT $ELECTRS_PID # close server -tail_log data/electrs/regtest-debug.log | grep -m1 "electrs stopped" +wait_for_log data/electrs/regtest-debug.log "electrs stopped" "electrs shutdown" wait $ELECTRS_PID # When the multiprocess node has IPC clients still attached at shutdown, the diff --git a/tests/run.sh b/tests/run.sh index 2f46382f4..5e11588e3 100755 --- a/tests/run.sh +++ b/tests/run.sh @@ -3,6 +3,7 @@ set -euo pipefail rm -rf data/ mkdir -p data/{bitcoin,electrum,electrs} +touch data/electrs/regtest-debug.log data/electrum/regtest-debug.log cleanup() { trap - SIGTERM SIGINT @@ -21,18 +22,50 @@ wait_for() { shift TEST_ARGS=$* for _ in `seq 0 9`; do - test "$($CMD | jq -c .)" $TEST_ARGS && break || sleep 2 + if test "$($CMD | jq -c .)" $TEST_ARGS; then + return 0 + fi + sleep 2 done + + echo "Timed out waiting for condition: $CMD $TEST_ARGS" >&2 + return 1 +} + +wait_for_log() { + LOG_FILE=$1 + PATTERN=$2 + DESCRIPTION=$3 + for _ in `seq 0 99`; do + if grep -Fq "$PATTERN" "$LOG_FILE"; then + return 0 + fi + sleep 0.2 + done + + echo "Timed out waiting for $DESCRIPTION in $LOG_FILE" >&2 + tail -n +1 "$LOG_FILE" >&2 || true + return 1 +} + +wait_for_http() { + URL=$1 + OUTPUT=$2 + for _ in `seq 0 99`; do + if curl --silent --show-error --fail "$URL" -o "$OUTPUT"; then + return 0 + fi + sleep 0.2 + done + + echo "Timed out waiting for HTTP endpoint $URL" >&2 + return 1 } BTC="bitcoin-cli -regtest -datadir=data/bitcoin" ELECTRUM="electrum --regtest" EL="$ELECTRUM --wallet=data/electrum/wallet" -tail_log() { - tail -n +0 -F $1 || true -} - echo "Starting $(bitcoind -version | head -n1)..." bitcoind -regtest -datadir=data/bitcoin -printtoconsole=0 & BITCOIND_PID=$! @@ -55,12 +88,12 @@ electrs \ --network=regtest \ 2> data/electrs/regtest-debug.log & ELECTRS_PID=$! -tail_log data/electrs/regtest-debug.log | grep -m1 "serving Electrum RPC" -curl localhost:24224 -o metrics.txt +wait_for_log data/electrs/regtest-debug.log "serving Electrum RPC" "electrs Electrum RPC startup" +wait_for_http http://localhost:24224 metrics.txt $ELECTRUM daemon --server localhost:60401:t -1 -vDEBUG 2> data/electrum/regtest-debug.log & ELECTRUM_PID=$! -tail_log data/electrum/regtest-debug.log | grep -m1 "connection established" +wait_for_log data/electrum/regtest-debug.log "connection established" "Electrum daemon connection" $EL getinfo | jq . echo "Loading Electrum wallet..." @@ -92,7 +125,7 @@ $BTC getblockcount > /dev/null echo " * wait for new block" kill -USR1 $ELECTRS_PID # notify server to index new block -tail_log data/electrum/regtest-debug.log | grep -m1 "verified $TXID" > /dev/null +wait_for_log data/electrum/regtest-debug.log "verified $TXID" "Electrum wallet verification of mined transaction" echo " * get_tx_status" test "`$EL get_tx_status $TXID | jq -c .`" == '{"confirmations":1}' @@ -107,7 +140,7 @@ echo "Electrum `$EL stop`" # disconnect wallet wait $ELECTRUM_PID kill -INT $ELECTRS_PID # close server -tail_log data/electrs/regtest-debug.log | grep -m1 "electrs stopped" +wait_for_log data/electrs/regtest-debug.log "electrs stopped" "electrs shutdown" wait $ELECTRS_PID # Try a graceful stop; if the node has already exited the RPC call will fail, From d343d3dad374370427388b1b7671511abe39cfa8 Mon Sep 17 00:00:00 2001 From: Sjors Provoost Date: Tue, 12 May 2026 21:16:51 +0200 Subject: [PATCH 15/21] ci: use bundled RocksDB in IPC job --- .github/workflows/rust.yml | 3 --- 1 file changed, 3 deletions(-) diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index 898939c7c..ef11c6d6f 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -119,9 +119,6 @@ jobs: electrum version --offline - name: Build electrs - env: - ROCKSDB_INCLUDE_DIR: /usr/include - ROCKSDB_LIB_DIR: /usr/lib run: cargo build --locked --bin electrs - name: Run IPC integration test From d9cd4ba406f9219586ad02933b36d6ed1fb2965b Mon Sep 17 00:00:00 2001 From: Sjors Provoost Date: Tue, 12 May 2026 21:17:53 +0200 Subject: [PATCH 16/21] ipc: sync mempool via Chain notifications --- src/daemon.rs | 34 +++++++---- src/ipc.rs | 159 ++++++++++++++++++++++++++++++++++++++++++++++++- src/mempool.rs | 72 +++++++++++++++++++++- 3 files changed, 249 insertions(+), 16 deletions(-) diff --git a/src/daemon.rs b/src/daemon.rs index 893a19ac0..af98eb0be 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -17,6 +17,7 @@ use crate::{ chain::{Chain, NewHeader}, config::Config, ipc::IpcChain, + mempool::MempoolEvent, metrics::Metrics, p2p::Connection, signals::ExitFlag, @@ -108,11 +109,11 @@ pub struct Daemon { p2p: Option>, rpc: Client, ipc: Option, - /// When IPC is configured, a long-running task on the IPC worker watches - /// `Chain.waitForNotificationsIfTipChanged` and signals on this channel - /// each time the tip changes. Used in place of the P2P `inv`-watching - /// path for `new_block_notification`. + /// When IPC is configured, a long-running ChainNotifications handler on + /// the IPC worker signals on this channel each time the tip changes. Used + /// in place of the P2P `inv`-watching path for `new_block_notification`. ipc_block_notifier: Option>, + ipc_mempool_events: Option>, } impl Daemon { @@ -181,14 +182,18 @@ impl Daemon { bail!("electrs requires non-pruned bitcoind node"); } - // Start the IPC tip-change notifier if available, so we can avoid the - // P2P `inv`-watching path for new-block notifications. - let ipc_block_notifier = match &ipc { - Some(ipc) => Some( - ipc.start_block_notifier() - .context("failed to start IPC block notifier")?, - ), - None => None, + // Register one IPC ChainNotifications handler for both mempool events + // and tip changes. This avoids the P2P `inv`-watching path and lets + // the mempool tracker consume node-pushed updates instead of polling + // `getrawmempool`. + let (ipc_mempool_events, ipc_block_notifier) = match &ipc { + Some(ipc) => { + let (mempool_events, block_notifier) = ipc + .start_notifications() + .context("failed to start IPC chain notifications")?; + (Some(mempool_events), Some(block_notifier)) + } + None => (None, None), }; Ok(Self { @@ -196,6 +201,7 @@ impl Daemon { rpc, ipc, ipc_block_notifier, + ipc_mempool_events, }) } @@ -416,6 +422,10 @@ impl Daemon { self.p2p().lock().new_block_notification() } + pub(crate) fn mempool_events(&self) -> Option<&Receiver> { + self.ipc_mempool_events.as_ref() + } + /// Accessor for the P2P connection. Panics if called when IPC is /// configured (which is a programmer error: the IPC fast paths above /// must be checked first). diff --git a/src/ipc.rs b/src/ipc.rs index 2745dddd4..8c531b0a5 100644 --- a/src/ipc.rs +++ b/src/ipc.rs @@ -14,20 +14,21 @@ use std::collections::HashMap; use std::path::{Path, PathBuf}; +use std::rc::Rc; use std::sync::Arc; use std::thread as stdthread; use std::time::Duration; use anyhow::{anyhow, bail, Context, Result}; use bitcoin::blockdata::block::Header as BlockHeader; -use bitcoin::{consensus::Decodable, hashes::Hash, BlockHash}; +use bitcoin::{consensus::deserialize, consensus::Decodable, hashes::Hash, BlockHash}; use bitcoin_capnp_types::{ - chain_capnp::chain, + chain_capnp::{chain, chain_notifications}, init_capnp::init, proxy_capnp::{thread as proxy_thread, thread_map}, }; use capnp_rpc::{rpc_twoparty_capnp::Side, twoparty::VatNetwork, RpcSystem}; -use crossbeam_channel::{bounded, Receiver}; +use crossbeam_channel::{bounded, unbounded, Receiver, Sender}; use futures::{future::LocalBoxFuture, io::BufReader, FutureExt}; use parking_lot::Mutex; use tokio::net::UnixStream; @@ -35,6 +36,7 @@ use tokio::sync::{mpsc, oneshot}; use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt}; use crate::chain::{Chain, NewHeader}; +use crate::mempool::MempoolEvent; use crate::types::SerBlock; /// Boxed (non-Send) future returned by a job. The closure itself must be @@ -184,6 +186,28 @@ impl IpcChain { }) } + /// Register a `ChainNotifications` handler and request a one-shot replay of + /// the node's current mempool. The returned receivers are fed by the IPC + /// worker thread for as long as the chain notification handler is alive. + pub(crate) fn start_notifications(&self) -> Result<(Receiver, Receiver<()>)> { + let (mempool_tx, mempool_rx) = unbounded::(); + let (block_tx, block_rx) = bounded::<()>(1); + let (ready_tx, ready_rx) = oneshot::channel::>(); + let job: Job = Box::new(move |ctx| { + async move { + register_notifications(ctx, mempool_tx, block_tx, ready_tx).await; + } + .boxed_local() + }); + self.tx + .send(job) + .map_err(|_| anyhow!("IPC worker thread is gone"))?; + ready_rx + .blocking_recv() + .map_err(|_| anyhow!("IPC worker dropped notification setup response"))??; + Ok((mempool_rx, block_rx)) + } + pub(crate) fn have_pruned(&self) -> Result { self.call(|ctx| { async move { @@ -500,6 +524,135 @@ impl IpcChain { } } +struct ChainNotificationHandler { + mempool_tx: Sender, + block_tx: Sender<()>, +} + +impl ChainNotificationHandler { + fn send_mempool_event(&self, event: MempoolEvent) { + if let Err(e) = self.mempool_tx.try_send(event) { + warn!("IPC notification: dropping mempool event: {e}"); + } + } + + fn send_block_event(&self) { + let _ = self.block_tx.try_send(()); + } +} + +impl chain_notifications::Server for ChainNotificationHandler { + fn destroy( + self: Rc, + _: chain_notifications::DestroyParams, + _: chain_notifications::DestroyResults, + ) -> impl std::future::Future> + 'static { + std::future::ready(Ok(())) + } + + fn transaction_added_to_mempool( + self: Rc, + params: chain_notifications::TransactionAddedToMempoolParams, + _: chain_notifications::TransactionAddedToMempoolResults, + ) -> impl std::future::Future> + 'static { + async move { + let p = params.get()?; + match deserialize(p.get_tx()?) { + Ok(tx) => self.send_mempool_event(MempoolEvent::Added(tx)), + Err(e) => warn!("IPC notification: invalid mempool transaction: {e}"), + } + Ok(()) + } + } + + fn transaction_removed_from_mempool( + self: Rc, + params: chain_notifications::TransactionRemovedFromMempoolParams, + _: chain_notifications::TransactionRemovedFromMempoolResults, + ) -> impl std::future::Future> + 'static { + async move { + let p = params.get()?; + match deserialize::(p.get_tx()?) { + Ok(tx) => self.send_mempool_event(MempoolEvent::Removed(tx.compute_txid())), + Err(e) => warn!("IPC notification: invalid removed mempool transaction: {e}"), + } + Ok(()) + } + } + + fn block_connected( + self: Rc, + _: chain_notifications::BlockConnectedParams, + _: chain_notifications::BlockConnectedResults, + ) -> impl std::future::Future> + 'static { + std::future::ready(Ok(())) + } + + fn block_disconnected( + self: Rc, + _: chain_notifications::BlockDisconnectedParams, + _: chain_notifications::BlockDisconnectedResults, + ) -> impl std::future::Future> + 'static { + std::future::ready(Ok(())) + } + + fn updated_block_tip( + self: Rc, + _: chain_notifications::UpdatedBlockTipParams, + _: chain_notifications::UpdatedBlockTipResults, + ) -> impl std::future::Future> + 'static { + self.send_block_event(); + std::future::ready(Ok(())) + } + + fn chain_state_flushed( + self: Rc, + _: chain_notifications::ChainStateFlushedParams, + _: chain_notifications::ChainStateFlushedResults, + ) -> impl std::future::Future> + 'static { + std::future::ready(Ok(())) + } +} + +async fn register_notifications( + ctx: Ctx, + mempool_tx: Sender, + block_tx: Sender<()>, + ready_tx: oneshot::Sender>, +) { + let setup = async { + let handler = Rc::new(ChainNotificationHandler { + mempool_tx, + block_tx, + }); + let notifications: chain_notifications::Client = capnp_rpc::new_client_from_rc(handler); + + let mut req = ctx.chain.handle_notifications_request(); + req.get().get_context()?.set_thread(ctx.thread.clone()); + req.get().set_notifications(notifications.clone()); + let resp = req.send().promise.await?; + let _handler = resp.get()?.get_result()?; + + let mut req = ctx.chain.request_mempool_transactions_request(); + req.get().get_context()?.set_thread(ctx.thread.clone()); + req.get().set_notifications(notifications); + req.send().promise.await?; + + Ok::<_, anyhow::Error>(_handler) + } + .await; + + match setup { + Ok(_handler) => { + let _ = ready_tx.send(Ok(())); + futures::future::pending::<()>().await; + } + Err(e) => { + let _ = ready_tx.send(Err(e)); + } + } +} + /// Helper for the notifier loop: read the current tip hash via /// `getHeight` + `getBlockHash`. Returns the all-zeros hash when the node /// reports no tip yet (genesis only), which still works for diff --git a/src/mempool.rs b/src/mempool.rs index 04457c95c..f6a78dd15 100644 --- a/src/mempool.rs +++ b/src/mempool.rs @@ -7,6 +7,7 @@ use std::ops::Bound; use bitcoin::hashes::Hash; use bitcoin::{Amount, OutPoint, Transaction, Txid}; +use crossbeam_channel::Receiver; use serde::ser::{Serialize, SerializeSeq, Serializer}; use crate::{ @@ -24,6 +25,11 @@ pub(crate) struct Entry { pub has_unconfirmed_inputs: bool, } +pub(crate) enum MempoolEvent { + Added(Transaction), + Removed(Txid), +} + /// Mempool current state pub(crate) struct Mempool { entries: HashMap, @@ -183,10 +189,15 @@ impl Mempool { let added = update.new_entries.len(); for txid_to_remove in update.removed_entries { - self.remove_entry(txid_to_remove); + if self.entries.contains_key(&txid_to_remove) { + self.remove_entry(txid_to_remove); + } } for entry in update.new_entries { + if self.entries.contains_key(&entry.txid) { + self.remove_entry(entry.txid); + } self.add_entry(entry); } @@ -210,7 +221,66 @@ impl Mempool { } } + fn sync_notifications(&mut self, daemon: &Daemon, events: &Receiver) { + let mut added_txs = HashMap::::new(); + let mut removed_entries = HashSet::::new(); + + for event in events.try_iter() { + match event { + MempoolEvent::Added(tx) => { + let txid = tx.compute_txid(); + removed_entries.remove(&txid); + added_txs.insert(txid, tx); + } + MempoolEvent::Removed(txid) => { + added_txs.remove(&txid); + removed_entries.insert(txid); + } + } + } + + let txids: Vec = added_txs.keys().copied().collect(); + let mut new_entries = Vec::with_capacity(txids.len()); + if !txids.is_empty() { + match daemon.get_mempool_entries(&txids) { + Ok(entries) => { + for (txid, entry) in txids.into_iter().zip(entries) { + let Some(entry) = entry else { + debug!("missing mempool entry: {}", txid); + continue; + }; + let Some(tx) = added_txs.remove(&txid) else { + continue; + }; + new_entries.push(Entry { + txid, + tx, + vsize: entry.vsize, + fee: entry.fees.base, + has_unconfirmed_inputs: !entry.depends.is_empty(), + }); + } + } + Err(e) => warn!("mempool notification metadata sync failed: {}", e), + } + } + + if new_entries.is_empty() && removed_entries.is_empty() { + return; + } + + self.apply_sync_update(MempoolSyncUpdate { + new_entries, + removed_entries, + }); + } + pub fn sync(&mut self, daemon: &Daemon, exit_flag: &ExitFlag) { + if let Some(events) = daemon.mempool_events() { + self.sync_notifications(daemon, events); + return; + } + let loaded = match daemon.get_mempool_info() { Ok(info) => info.loaded.unwrap_or(true), Err(e) => { From c389f1227adab1d15ba4d72238946612479a66b1 Mon Sep 17 00:00:00 2001 From: Sjors Provoost Date: Tue, 12 May 2026 21:28:47 +0200 Subject: [PATCH 17/21] ipc: derive mempool entry metadata from Chain coins --- src/daemon.rs | 12 +++- src/ipc.rs | 161 ++++++++++++++++++++++++++++++++++++++++++++++++- src/mempool.rs | 75 +++++++++++++++-------- 3 files changed, 221 insertions(+), 27 deletions(-) diff --git a/src/daemon.rs b/src/daemon.rs index af98eb0be..9c7d313cf 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -2,7 +2,7 @@ use anyhow::{Context, Result}; use bitcoin::consensus::encode::serialize_hex; use bitcoin::{consensus::deserialize, hashes::hex::FromHex}; -use bitcoin::{Amount, BlockHash, Transaction, Txid}; +use bitcoin::{Amount, BlockHash, OutPoint, Transaction, Txid}; use bitcoincore_rpc::{json, jsonrpc, Auth, Client, RpcApi}; use crossbeam_channel::Receiver; use parking_lot::Mutex; @@ -16,7 +16,7 @@ use std::path::Path; use crate::{ chain::{Chain, NewHeader}, config::Config, - ipc::IpcChain, + ipc::{IpcChain, IpcCoin}, mempool::MempoolEvent, metrics::Metrics, p2p::Connection, @@ -383,6 +383,14 @@ impl Daemon { .collect()) } + pub(crate) fn find_coins(&self, outpoints: Vec) -> Result> { + self.ipc + .as_ref() + .context("Chain.findCoins requested without IPC connection")? + .find_coins(outpoints) + .context("Chain.findCoins failed") + } + pub(crate) fn get_new_headers(&self, chain: &Chain) -> Result> { if let Some(ipc) = &self.ipc { return ipc diff --git a/src/ipc.rs b/src/ipc.rs index 8c531b0a5..ac7ce8e48 100644 --- a/src/ipc.rs +++ b/src/ipc.rs @@ -21,7 +21,9 @@ use std::time::Duration; use anyhow::{anyhow, bail, Context, Result}; use bitcoin::blockdata::block::Header as BlockHeader; -use bitcoin::{consensus::deserialize, consensus::Decodable, hashes::Hash, BlockHash}; +use bitcoin::{ + consensus::deserialize, consensus::Decodable, hashes::Hash, Amount, BlockHash, OutPoint, +}; use bitcoin_capnp_types::{ chain_capnp::{chain, chain_notifications}, init_capnp::init, @@ -73,6 +75,11 @@ pub(crate) struct IpcChain { header_fetch_cache: Arc>, } +pub(crate) struct IpcCoin { + pub(crate) value: Amount, + pub(crate) height: u32, +} + /// In-memory cache of recently fetched blocks, drained by /// [`IpcChain::get_block`] on hit. Keyed by block hash, bounded by total /// payload bytes. @@ -364,6 +371,44 @@ impl IpcChain { }) } + /// Look up prevout coins through `Chain.findCoins`. The interface takes a + /// `std::map&`, encoded as key/value byte pairs. The + /// incoming `Coin` values are placeholders; the node overwrites them with + /// coins from the active UTXO set or mempool. + pub(crate) fn find_coins(&self, outpoints: Vec) -> Result> { + self.call(move |ctx| { + async move { + let mut req = ctx.chain.find_coins_request(); + req.get().get_context()?.set_thread(ctx.thread.clone()); + { + let mut coins = req.get().init_coins(outpoints.len() as u32); + for (i, outpoint) in outpoints.iter().enumerate() { + let mut pair = coins.reborrow().get(i as u32); + pair.set_key(&serialize_outpoint(outpoint)[..])?; + pair.set_value(DUMMY_COIN)?; + } + } + let resp = req.send().promise.await?; + let coins = resp.get()?.get_coins()?; + if coins.len() != outpoints.len() as u32 { + bail!( + "Chain.findCoins returned {} coins, expected {}", + coins.len(), + outpoints.len() + ); + } + + let mut result = Vec::with_capacity(outpoints.len()); + for i in 0..coins.len() { + let pair = coins.get(i); + result.push(decode_coin(pair.get_value()?)?); + } + Ok(result) + } + .boxed_local() + }) + } + /// Find the height of the highest block in `locator` that is part of the /// node's active chain, via `Chain.findLocatorFork`. Returns `None` if no /// hash in the locator is on the active chain (e.g. completely diverged @@ -701,6 +746,87 @@ fn decode_fee_rate_kvb(blob: &[u8]) -> Result> { Ok(Some(fee.saturating_mul(1000) / size as i64)) } +/// Serialized `Coin{CTxOut{0, empty script}, height=0, coinbase=false}`. +/// `findCoins` treats request coin values as placeholders, but the map +/// decoder still needs each value to deserialize as a valid `Coin`. +const DUMMY_COIN: &[u8] = &[0x00, 0x00, 0x06]; + +fn serialize_outpoint(outpoint: &OutPoint) -> [u8; 36] { + let mut buf = [0u8; 36]; + buf[..32].copy_from_slice(outpoint.txid.as_byte_array()); + buf[32..].copy_from_slice(&outpoint.vout.to_le_bytes()); + buf +} + +fn decode_coin(blob: &[u8]) -> Result { + let mut pos = 0; + let code = read_core_varint(blob, &mut pos)?; + let compressed_amount = read_core_varint(blob, &mut pos)?; + let script_size = read_core_varint(blob, &mut pos)?; + let script_bytes = match script_size { + 0 | 1 => 20, + 2..=5 => 32, + n => n + .checked_sub(6) + .ok_or_else(|| anyhow!("invalid compressed script size code {n}"))?, + } as usize; + if blob.len().saturating_sub(pos) != script_bytes { + bail!( + "serialized Coin script has {} bytes, expected {}", + blob.len().saturating_sub(pos), + script_bytes + ); + } + Ok(IpcCoin { + value: Amount::from_sat(decompress_amount(compressed_amount)), + height: (code >> 1) + .try_into() + .context("serialized Coin height overflow")?, + }) +} + +/// Bitcoin Core VARINT encoding (base-128 with subtract-one continuation). +fn read_core_varint(blob: &[u8], pos: &mut usize) -> Result { + let mut n = 0u64; + loop { + let Some(&ch) = blob.get(*pos) else { + bail!("unexpected end of Bitcoin Core VARINT"); + }; + *pos += 1; + n = n + .checked_shl(7) + .and_then(|n| n.checked_add((ch & 0x7f) as u64)) + .ok_or_else(|| anyhow!("Bitcoin Core VARINT overflow"))?; + if ch & 0x80 == 0 { + return Ok(n); + } + n = n + .checked_add(1) + .ok_or_else(|| anyhow!("Bitcoin Core VARINT overflow"))?; + } +} + +/// Inverse of Bitcoin Core's `CompressAmount`. +fn decompress_amount(mut x: u64) -> u64 { + if x == 0 { + return 0; + } + x -= 1; + let e = x % 10; + x /= 10; + let mut n = if e < 9 { + let d = (x % 9) + 1; + x /= 9; + x * 10 + d + } else { + x + 1 + }; + for _ in 0..e { + n *= 10; + } + n +} + /// Bitcoin Core CompactSize encoding (src/serialize.h). fn write_compact_size(buf: &mut Vec, n: u64) { if n < 253 { @@ -811,3 +937,36 @@ pub(crate) fn decode_block(buf: &[u8]) -> Result { bitcoin::Block::consensus_decode(&mut cursor) .map_err(|e| anyhow!("failed to decode block: {e}")) } + +#[cfg(test)] +mod tests { + use super::{decompress_amount, read_core_varint}; + + #[test] + fn decodes_core_varint_examples() { + for (bytes, expected) in [ + (&[0x00][..], 0), + (&[0x7f][..], 0x7f), + (&[0x80, 0x00][..], 0x80), + (&[0xa3, 0x34][..], 0x1234), + (&[0x82, 0xfe, 0x7f][..], 0xffff), + ] { + let mut pos = 0; + assert_eq!(read_core_varint(bytes, &mut pos).unwrap(), expected); + assert_eq!(pos, bytes.len()); + } + } + + #[test] + fn decompresses_bitcoin_amounts() { + for (compressed, sats) in [ + (0, 0), + (1, 1), + (7, 1000000), + (9, 100000000), + (50, 5000000000), + ] { + assert_eq!(decompress_amount(compressed), sats); + } + } +} diff --git a/src/mempool.rs b/src/mempool.rs index f6a78dd15..7e8d7c4ea 100644 --- a/src/mempool.rs +++ b/src/mempool.rs @@ -1,4 +1,4 @@ -use anyhow::{Context, Result}; +use anyhow::{anyhow, Context, Result}; use std::collections::{BTreeSet, HashMap, HashSet}; use std::convert::TryFrom; @@ -30,6 +30,9 @@ pub(crate) enum MempoolEvent { Removed(Txid), } +/// Bitcoin Core marks mempool prevout coins with this synthetic height. +const MEMPOOL_HEIGHT: u32 = 0x7fffffff; + /// Mempool current state pub(crate) struct Mempool { entries: HashMap, @@ -239,29 +242,11 @@ impl Mempool { } } - let txids: Vec = added_txs.keys().copied().collect(); - let mut new_entries = Vec::with_capacity(txids.len()); - if !txids.is_empty() { - match daemon.get_mempool_entries(&txids) { - Ok(entries) => { - for (txid, entry) in txids.into_iter().zip(entries) { - let Some(entry) = entry else { - debug!("missing mempool entry: {}", txid); - continue; - }; - let Some(tx) = added_txs.remove(&txid) else { - continue; - }; - new_entries.push(Entry { - txid, - tx, - vsize: entry.vsize, - fee: entry.fees.base, - has_unconfirmed_inputs: !entry.depends.is_empty(), - }); - } - } - Err(e) => warn!("mempool notification metadata sync failed: {}", e), + let mut new_entries = Vec::with_capacity(added_txs.len()); + for (txid, tx) in added_txs { + match entry_from_notification_tx(daemon, txid, tx) { + Ok(entry) => new_entries.push(entry), + Err(e) => warn!("mempool notification metadata sync failed for {txid}: {e}"), } } @@ -355,6 +340,48 @@ impl Mempool { } } +fn entry_from_notification_tx(daemon: &Daemon, txid: Txid, tx: Transaction) -> Result { + let prevouts: Vec = tx.input.iter().map(|txin| txin.previous_output).collect(); + let coins = daemon + .find_coins(prevouts) + .with_context(|| format!("looking up prevout coins for {txid}"))?; + ensure!( + coins.len() == tx.input.len(), + "got {} prevout coins, expected {}", + coins.len(), + tx.input.len() + ); + + let mut input_value = Amount::ZERO; + let mut has_unconfirmed_inputs = false; + for coin in coins { + input_value = input_value + .checked_add(coin.value) + .ok_or_else(|| anyhow!("input value overflow for {txid}"))?; + has_unconfirmed_inputs |= coin.height == MEMPOOL_HEIGHT; + } + + let mut output_value = Amount::ZERO; + for txout in &tx.output { + output_value = output_value + .checked_add(txout.value) + .ok_or_else(|| anyhow!("output value overflow for {txid}"))?; + } + + let fee = input_value + .checked_sub(output_value) + .ok_or_else(|| anyhow!("negative fee for {txid}"))?; + let vsize = tx.vsize() as u64; + + Ok(Entry { + txid, + tx, + fee, + vsize, + has_unconfirmed_inputs, + }) +} + pub(crate) struct FeeHistogram { /// bins[64-i] contains transactions' statistics inside the fee band of [2**(i-1), 2**i). /// bins[64] = [0, 1) From a69bb632d6f35977e3c3dcc5be17f8cf2b846367 Mon Sep 17 00:00:00 2001 From: Sjors Provoost Date: Tue, 12 May 2026 21:30:41 +0200 Subject: [PATCH 18/21] ci: cache Rust dependencies --- .github/workflows/rust.yml | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index ef11c6d6f..5986ce37d 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -30,13 +30,10 @@ jobs: run: | git clone --depth 1 --branch 2026/04/chain-interface \ https://github.com/Sjors/bitcoin-capnp-types.git ../capnp-types - - uses: dtolnay/rust-toolchain@stable + - uses: actions-rust-lang/setup-rust-toolchain@v1 with: components: rustfmt, clippy - - name: Install Rust - run: rustup component add rustfmt clippy - - name: Install other dependencies run: sudo apt install build-essential libclang-dev capnproto libcapnp-dev @@ -75,7 +72,7 @@ jobs: git clone --depth 1 --branch 2026/04/chain-interface \ https://github.com/Sjors/bitcoin-capnp-types.git ../capnp-types - - uses: dtolnay/rust-toolchain@stable + - uses: actions-rust-lang/setup-rust-toolchain@v1 - name: Install build & test dependencies run: | From 194f0ec788888d14ae94f9f32e3b20b6c1cee5bd Mon Sep 17 00:00:00 2001 From: Sjors Provoost Date: Tue, 12 May 2026 21:31:56 +0200 Subject: [PATCH 19/21] ipc: satisfy Clippy async fn lint --- src/ipc.rs | 32 ++++++++++++++------------------ 1 file changed, 14 insertions(+), 18 deletions(-) diff --git a/src/ipc.rs b/src/ipc.rs index ac7ce8e48..f29153d76 100644 --- a/src/ipc.rs +++ b/src/ipc.rs @@ -595,34 +595,30 @@ impl chain_notifications::Server for ChainNotificationHandler { std::future::ready(Ok(())) } - fn transaction_added_to_mempool( + async fn transaction_added_to_mempool( self: Rc, params: chain_notifications::TransactionAddedToMempoolParams, _: chain_notifications::TransactionAddedToMempoolResults, - ) -> impl std::future::Future> + 'static { - async move { - let p = params.get()?; - match deserialize(p.get_tx()?) { - Ok(tx) => self.send_mempool_event(MempoolEvent::Added(tx)), - Err(e) => warn!("IPC notification: invalid mempool transaction: {e}"), - } - Ok(()) + ) -> std::result::Result<(), capnp::Error> { + let p = params.get()?; + match deserialize(p.get_tx()?) { + Ok(tx) => self.send_mempool_event(MempoolEvent::Added(tx)), + Err(e) => warn!("IPC notification: invalid mempool transaction: {e}"), } + Ok(()) } - fn transaction_removed_from_mempool( + async fn transaction_removed_from_mempool( self: Rc, params: chain_notifications::TransactionRemovedFromMempoolParams, _: chain_notifications::TransactionRemovedFromMempoolResults, - ) -> impl std::future::Future> + 'static { - async move { - let p = params.get()?; - match deserialize::(p.get_tx()?) { - Ok(tx) => self.send_mempool_event(MempoolEvent::Removed(tx.compute_txid())), - Err(e) => warn!("IPC notification: invalid removed mempool transaction: {e}"), - } - Ok(()) + ) -> std::result::Result<(), capnp::Error> { + let p = params.get()?; + match deserialize::(p.get_tx()?) { + Ok(tx) => self.send_mempool_event(MempoolEvent::Removed(tx.compute_txid())), + Err(e) => warn!("IPC notification: invalid removed mempool transaction: {e}"), } + Ok(()) } fn block_connected( From d759191e5e4ed74b2466ae6fd04132189bf56963 Mon Sep 17 00:00:00 2001 From: Sjors Provoost Date: Tue, 12 May 2026 21:37:47 +0200 Subject: [PATCH 20/21] ipc: drop confirmed txs from notification mempool --- src/ipc.rs | 23 +++++++++++++++++++---- 1 file changed, 19 insertions(+), 4 deletions(-) diff --git a/src/ipc.rs b/src/ipc.rs index f29153d76..dd477f604 100644 --- a/src/ipc.rs +++ b/src/ipc.rs @@ -621,12 +621,27 @@ impl chain_notifications::Server for ChainNotificationHandler { Ok(()) } - fn block_connected( + async fn block_connected( self: Rc, - _: chain_notifications::BlockConnectedParams, + params: chain_notifications::BlockConnectedParams, _: chain_notifications::BlockConnectedResults, - ) -> impl std::future::Future> + 'static { - std::future::ready(Ok(())) + ) -> std::result::Result<(), capnp::Error> { + let p = params.get()?; + let data = p.get_block()?.get_data()?; + if data.is_empty() { + warn!("IPC notification: connected block without block data"); + return Ok(()); + } + match deserialize::(data) { + Ok(block) => { + for tx in block.txdata { + self.send_mempool_event(MempoolEvent::Removed(tx.compute_txid())); + } + } + Err(e) => warn!("IPC notification: invalid connected block: {e}"), + } + self.send_block_event(); + Ok(()) } fn block_disconnected( From dbdb63300208dc8352afa47951f072d5e9009452 Mon Sep 17 00:00:00 2001 From: Sjors Provoost Date: Tue, 12 May 2026 21:55:34 +0200 Subject: [PATCH 21/21] test: assert IPC backend in integration test --- tests/run-ipc.sh | 9 ++------- 1 file changed, 2 insertions(+), 7 deletions(-) diff --git a/tests/run-ipc.sh b/tests/run-ipc.sh index ab20b4e94..89a7b17f1 100755 --- a/tests/run-ipc.sh +++ b/tests/run-ipc.sh @@ -84,6 +84,7 @@ echo "Starting $($BITCOIN_MULTIPROCESS_BIN node -version | head -n1) (multiproce $BITCOIN_MULTIPROCESS_BIN node \ -regtest -datadir=$PWD/data/bitcoin \ -ipcbind=unix \ + -listen=0 \ -printtoconsole=0 \ -fallbackfee=0.0001 & BITCOIND_PID=$! @@ -114,13 +115,7 @@ electrs \ 2> data/electrs/regtest-debug.log & ELECTRS_PID=$! wait_for_log data/electrs/regtest-debug.log "serving Electrum RPC" "electrs Electrum RPC startup" - -# Confirm the IPC backend actually engaged. The Daemon::connect path logs -# nothing distinctive yet; instead probe the debug log for any IPC error or -# verify the for_blocks code path was taken by checking a unique marker. -if ! grep -q "ipc" data/electrs/regtest-debug.log; then - : # no explicit log line yet; we rely on cargo tests + functional behaviour -fi +wait_for_log data/electrs/regtest-debug.log "connecting to bitcoin-node IPC socket" "electrs IPC backend connection" wait_for_http http://localhost:24224 metrics.txt