diff --git a/Cargo.lock b/Cargo.lock index 9150cee..f61ee3c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -51,12 +51,35 @@ version = "0.13.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9e1b586273c5702936fe7b7d6896644d8be71e6314cfe09d3167c95f712589e8" +[[package]] +name = "base64" +version = "0.22.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" + [[package]] name = "bech32" version = "0.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d965446196e3b7decd44aa7ee49e31d630118f90ef12f97900f262eb915c951d" +[[package]] +name = "bindex" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1278fa172b7d90655ab8df4ab21b3956f6f0929cf1e3d97b6671188ef7c7e998" +dependencies = [ + "bitcoin", + "bitcoin_slices", + "log", + "rayon", + "rust-rocksdb", + "serde", + "serde_json", + "thiserror 2.0.16", + "ureq", +] + [[package]] name = "bindgen" version = "0.69.5" @@ -145,31 +168,6 @@ checksum = "82b80fcc031ed36f91c31639a4d97acb1487985fdee34565651b2f52f4346859" dependencies = [ "bitcoin", "bitcoin_hashes", - "sha2", -] - -[[package]] -name = "bitcoincore-rpc" -version = "0.19.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "aedd23ae0fd321affb4bbbc36126c6f49a32818dc6b979395d24da8c9d4e80ee" -dependencies = [ - "bitcoincore-rpc-json", - "jsonrpc", - "log", - "serde", - "serde_json", -] - -[[package]] -name = "bitcoincore-rpc-json" -version = "0.19.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d8909583c5fab98508e80ef73e5592a651c954993dc6b7739963257d19f0e71a" -dependencies = [ - "bitcoin", - "serde", - "serde_json", ] [[package]] @@ -178,15 +176,6 @@ version = "2.9.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1b8e56985ec62d17e9c1001dc89c88ecd7dc08e47eba5ec7c29c7b5eeecde967" -[[package]] -name = "block-buffer" -version = "0.10.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71" -dependencies = [ - "generic-array", -] - [[package]] name = "block2" version = "0.6.2" @@ -196,6 +185,12 @@ dependencies = [ "objc2", ] +[[package]] +name = "bytes" +version = "1.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b35204fbdc0b3f4446b89fc1ac2cf84a8a68971995d0bf2e925ec7cd960f9cb3" + [[package]] name = "bzip2-sys" version = "0.1.13+1.0.8" @@ -291,15 +286,6 @@ dependencies = [ "void", ] -[[package]] -name = "cpufeatures" -version = "0.2.17" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "59ed5838eebb26a2bb2e58f6d5b5316989ae9d08bab10e0e6d103e656d1b0280" -dependencies = [ - "libc", -] - [[package]] name = "crossbeam-channel" version = "0.5.15" @@ -334,16 +320,6 @@ version = "0.8.21" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d0a5c400df2834b80a4c3327b3aad3a4c4cd4de0629063962b03235697506a28" -[[package]] -name = "crypto-common" -version = "0.1.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1bfb12502f3fc46cca1bb51ac28df9d618d813cdc3d2f25b9fe775a34af26bb3" -dependencies = [ - "generic-array", - "typenum", -] - [[package]] name = "ctrlc" version = "3.5.2" @@ -355,16 +331,6 @@ dependencies = [ "windows-sys 0.61.0", ] -[[package]] -name = "digest" -version = "0.10.7" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" -dependencies = [ - "block-buffer", - "crypto-common", -] - [[package]] name = "dirs-next" version = "2.0.0" @@ -409,10 +375,8 @@ name = "electrs" version = "0.11.1" dependencies = [ "anyhow", - "bitcoin", + "bindex", "bitcoin-test-data", - "bitcoin_slices", - "bitcoincore-rpc", "configure_me", "configure_me_codegen", "crossbeam-channel", @@ -420,11 +384,10 @@ dependencies = [ "dirs-next", "env_logger", "hex_lit", + "jsonrpc", "log", - "parking_lot", "prometheus", "rayon", - "rust-rocksdb", "serde", "serde_derive", "serde_json", @@ -474,16 +437,6 @@ version = "1.0.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1" -[[package]] -name = "generic-array" -version = "0.14.7" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a" -dependencies = [ - "typenum", - "version_check", -] - [[package]] name = "getrandom" version = "0.2.16" @@ -540,6 +493,22 @@ version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3011d1213f159867b13cfd6ac92d2cd5f1345762c63be3554e84092d85a50bbd" +[[package]] +name = "http" +version = "1.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e3ba2a386d7f85a81f119ad7498ebe444d2e22c2af0b86b069416ace48b3311a" +dependencies = [ + "bytes", + "itoa", +] + +[[package]] +name = "httparse" +version = "1.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" + [[package]] name = "httpdate" version = "1.0.3" @@ -594,8 +563,7 @@ version = "0.18.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3662a38d341d77efecb73caf01420cfa5aa63c0253fd7bc05289ef9f6616e1bf" dependencies = [ - "base64", - "minreq", + "base64 0.13.1", "serde", "serde_json", ] @@ -679,17 +647,6 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a" -[[package]] -name = "minreq" -version = "2.13.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f0d2aaba477837b46ec1289588180fabfccf0c3b1d1a0c6b1866240cd6cd5ce9" -dependencies = [ - "log", - "serde", - "serde_json", -] - [[package]] name = "nix" version = "0.31.1" @@ -771,21 +728,18 @@ version = "1.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5bddc33f680b79eaf1e2e56da792c3c2236f86985bbc3a886e8ddee17ae4d3a4" +[[package]] +name = "percent-encoding" +version = "2.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220" + [[package]] name = "pkg-config" version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7edddbd0b52d732b21ad9a5fab5c704c14cd949e5e9a1ec5929a24fded1b904c" -[[package]] -name = "ppv-lite86" -version = "0.2.21" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "85eae3c4ed2f50dcfe72643da4befc30deadb458a9b590d720cde2f2b1e97da9" -dependencies = [ - "zerocopy", -] - [[package]] name = "proc-macro2" version = "1.0.95" @@ -869,36 +823,6 @@ version = "5.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "74765f6d916ee2faa39bc8e68e4f3ed8949b48cccdac59983d287a7cb71ce9c5" -[[package]] -name = "rand" -version = "0.8.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5ca0ecfa931c29007047d1bc58e623ab12e5590e8c7cc53200d5202b69266d8a" -dependencies = [ - "libc", - "rand_chacha", - "rand_core", -] - -[[package]] -name = "rand_chacha" -version = "0.3.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e6c10a63a0fa32252be49d21e7709d4d4baf8d231c2dbce1eaa8141b9b127d88" -dependencies = [ - "ppv-lite86", - "rand_core", -] - -[[package]] -name = "rand_core" -version = "0.6.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c" -dependencies = [ - "getrandom 0.2.16", -] - [[package]] name = "rayon" version = "1.12.0" @@ -1038,7 +962,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9465315bc9d4566e1724f0fffcbcc446268cb522e60f9a27bcded6b19c108113" dependencies = [ "bitcoin_hashes", - "rand", "secp256k1-sys", "serde", ] @@ -1095,17 +1018,6 @@ dependencies = [ "zmij", ] -[[package]] -name = "sha2" -version = "0.10.9" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" -dependencies = [ - "cfg-if", - "cpufeatures", - "digest", -] - [[package]] name = "shlex" version = "1.3.0" @@ -1231,12 +1143,6 @@ dependencies = [ "serde", ] -[[package]] -name = "typenum" -version = "1.18.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1dccffe3ce07af9386bfd29e80c0ab1a8205a2fc34e4bcd40364df902cfa8f3f" - [[package]] name = "unicode-ident" version = "1.0.18" @@ -1249,18 +1155,43 @@ version = "1.12.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f6ccf251212114b54433ec949fd6a7841275f9ada20dddd2f29e9ceea4501493" +[[package]] +name = "ureq" +version = "3.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d39cb1dbab692d82a977c0392ffac19e188bd9186a9f32806f0aaa859d75585a" +dependencies = [ + "base64 0.22.1", + "log", + "percent-encoding", + "ureq-proto", + "utf-8", +] + +[[package]] +name = "ureq-proto" +version = "0.5.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d81f9efa9df032be5934a46a068815a10a042b494b6a58cb0a1a97bb5467ed6f" +dependencies = [ + "base64 0.22.1", + "http", + "httparse", + "log", +] + +[[package]] +name = "utf-8" +version = "0.7.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09cc8ee72d2a9becf2f2febe0205bbed8fc6615b7cb429ad062dc7b7ddd036a9" + [[package]] name = "vcpkg" version = "0.2.15" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "accd4ea62f7bb7a82fe23066fb0957d48ef677f6eeb8215f372f52e48bb32426" -[[package]] -name = "version_check" -version = "0.9.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" - [[package]] name = "void" version = "1.0.2" @@ -1410,26 +1341,6 @@ dependencies = [ "bitflags", ] -[[package]] -name = "zerocopy" -version = "0.8.25" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a1702d9583232ddb9174e01bb7c15a2ab8fb1bc6f227aa1233858c351a3ba0cb" -dependencies = [ - "zerocopy-derive", -] - -[[package]] -name = "zerocopy-derive" -version = "0.8.25" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "28a6e20d751156648aa063f3800b706ee209a32c0b4d9f24be3d980b01be55ef" -dependencies = [ - "proc-macro2", - "quote", - "syn", -] - [[package]] name = "zmij" version = "1.0.2" diff --git a/Cargo.toml b/Cargo.toml index 112d3bd..a5d1a8c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -23,15 +23,13 @@ spec = "internal/config_specification.toml" [dependencies] anyhow = "1.0" -bitcoin = { version = "0.32.9", features = ["serde", "rand-std"] } -bitcoin_slices = { version = "0.11.0", features = ["bitcoin", "sha2"] } -bitcoincore-rpc = { version = "0.19.0" } +bindex = { version = "0.1.1", default-features = false, features = ["bitcoin_serde"] } configure_me = "0.4" crossbeam-channel = "0.5" dirs-next = "2.0" env_logger = "0.10" +jsonrpc = { version = "0.18.0", default-features = false, features = ["simple_http"]} log = "0.4" -parking_lot = "0.12" prometheus = { version = "0.14", optional = true } rayon = "1.12" serde = "1.0.184" @@ -45,14 +43,6 @@ ctrlc = "=3.5.2" [target.'cfg(not(windows))'.dependencies] signal-hook = "0.4" -[dependencies.rust-rocksdb] -version = "0.36" - -default-features = false -# ZSTD is used for data compression -# Snappy is only for checking old DB -features = ["zstd", "snappy"] - [build-dependencies] configure_me_codegen = { version = "0.4.8", default-features = false } @@ -60,9 +50,3 @@ configure_me_codegen = { version = "0.4.8", default-features = false } bitcoin-test-data = "0.2.0" hex_lit = "0.1.1" tempfile = "3.25" - -[profile.release] -lto = true - -[lints.clippy] -uninlined_format_args = "allow" # TODO: https://github.com/romanz/electrs/issues/1199 diff --git a/Dockerfile.ci b/Dockerfile.ci index 9106ac1..1c4d53d 100644 --- a/Dockerfile.ci +++ b/Dockerfile.ci @@ -4,7 +4,7 @@ FROM debian:trixie-slim AS base RUN apt-get update -qqy -RUN apt-get install -qqy librocksdb-dev wget +RUN apt-get install -qqy librocksdb-dev libsqlite3-dev libevent-dev libboost-dev wget ### Electrum Rust Server ### FROM base AS electrs-build @@ -17,10 +17,11 @@ ENV ROCKSDB_INCLUDE_DIR=/usr/include ENV ROCKSDB_LIB_DIR=/usr/lib RUN cargo install --locked --path . -### Bitcoin Core ### +### Bitcoin Core (custom build for /blockpart/ REST endpoint) ### FROM base AS bitcoin-build # Download WORKDIR /build/bitcoin + RUN wget -q https://bitcoincore.org/bin/bitcoin-core-31.0/test.rc2/bitcoin-31.0rc2-x86_64-linux-gnu.tar.gz RUN tar xvf bitcoin-31.0rc2-x86_64-linux-gnu.tar.gz RUN mv -v bitcoin-31.0rc2/bin/bitcoind . diff --git a/doc/config_example.toml b/doc/config_example.toml index 1272ddb..59cf429 100644 --- a/doc/config_example.toml +++ b/doc/config_example.toml @@ -14,9 +14,6 @@ cookie_file = "/var/run/bitcoin-mainnet/cookie" # The listening RPC address of bitcoind, port is usually 8332 daemon_rpc_addr = "127.0.0.1:8332" -# The listening P2P address of bitcoind, port is usually 8333 -daemon_p2p_addr = "127.0.0.1:8333" - # Directory where the index should be stored. It should have at least 70GB of free space. db_dir = "/some/fast/storage/with/big/size" diff --git a/doc/usage.md b/doc/usage.md index e6da8c7..abbdd72 100644 --- a/doc/usage.md +++ b/doc/usage.md @@ -21,7 +21,7 @@ $ du -ch ~/.bitcoin/blocks/blk*.dat | tail -n1 336G total $ ./target/release/electrs --network bitcoin --db-dir ./db --daemon-dir /home/user/.bitcoin -Starting electrs 0.10.0 on x86_64 linux with Config { network: Bitcoin, db_path: "./db/bitcoin", daemon_dir: "/home/user/.bitcoin", daemon_auth: CookieFile("/home/user/.bitcoin/.cookie"), daemon_rpc_addr: 127.0.0.1:8332, daemon_p2p_addr: 127.0.0.1:8333, electrum_rpc_addr: 127.0.0.1:50001, monitoring_addr: 127.0.0.1:4224, wait_duration: 10s, jsonrpc_timeout: 15s, index_batch_size: 10, index_lookup_limit: None, reindex_last_blocks: 0, auto_reindex: true, ignore_mempool: false, sync_once: false, skip_block_download_wait: false, disable_electrum_rpc: false, server_banner: "Welcome to electrs 0.10.0 (Electrum Rust Server)!", magic: f9beb4d9, args: [] } +Starting electrs 0.10.0 on x86_64 linux with Config { network: Bitcoin, db_path: "./db/bitcoin", daemon_dir: "/home/user/.bitcoin", daemon_auth: CookieFile("/home/user/.bitcoin/.cookie"), daemon_rpc_addr: 127.0.0.1:8332, electrum_rpc_addr: 127.0.0.1:50001, monitoring_addr: 127.0.0.1:4224, wait_duration: 10s, jsonrpc_timeout: 15s, index_lookup_limit: None, reindex_last_blocks: 0, auto_reindex: true, ignore_mempool: false, sync_once: false, skip_block_download_wait: false, disable_electrum_rpc: false, server_banner: "Welcome to electrs 0.10.0 (Electrum Rust Server)!", magic: f9beb4d9, args: [] } [2023-08-16T19:17:11.193Z INFO electrs::metrics::metrics_impl] serving Prometheus metrics on 127.0.0.1:4224 [2023-08-16T19:17:11.193Z INFO electrs::server] serving Electrum RPC on 127.0.0.1:50001 [2023-08-16T19:17:12.355Z INFO electrs::db] "./db/bitcoin": 0 SST files, 0 GB, 0 Grows diff --git a/examples/tx_collisions.rs b/examples/tx_collisions.rs deleted file mode 100644 index 0b7fd58..0000000 --- a/examples/tx_collisions.rs +++ /dev/null @@ -1,31 +0,0 @@ -use anyhow::{Context, Result}; -use rust_rocksdb::{ColumnFamilyDescriptor, IteratorMode, Options, DB}; - -fn main() -> Result<()> { - let path = std::env::args().nth(1).context("missing DB path")?; - let cf_names = DB::list_cf(&Options::default(), &path)?; - let cfs: Vec<_> = cf_names - .iter() - .map(|name| ColumnFamilyDescriptor::new(name, Options::default())) - .collect(); - let db = DB::open_cf_descriptors(&Options::default(), &path, cfs)?; - let cf = db.cf_handle("txid").context("missing column family")?; - - let mut state: Option<(u64, u32)> = None; - for row in db.iterator_cf(cf, IteratorMode::Start) { - let (curr, _value) = row?; - let curr_prefix = u64::from_le_bytes(curr[..8].try_into()?); - let curr_height = u32::from_le_bytes(curr[8..].try_into()?); - - if let Some((prev_prefix, prev_height)) = state { - if prev_prefix == curr_prefix { - eprintln!( - "prefix={:x} heights: {} {}", - curr_prefix, prev_height, curr_height - ); - }; - } - state = Some((curr_prefix, curr_height)); - } - Ok(()) -} diff --git a/internal/config_specification.toml b/internal/config_specification.toml index f4ae65e..f445f41 100644 --- a/internal/config_specification.toml +++ b/internal/config_specification.toml @@ -63,7 +63,7 @@ doc = "JSONRPC authentication cookie file (default: ~/.bitcoin/.cookie)" [[param]] name = "network" type = "crate::config::BitcoinNetwork" -convert_into = "::bitcoin::Network" +convert_into = "crate::bitcoin::Network" doc = "Select Bitcoin network type ('bitcoin', 'testnet', 'testnet4', 'regtest' or 'signet')" default = "Default::default()" @@ -76,10 +76,6 @@ doc = "Electrum server JSONRPC 'addr:port' to listen on (default: '127.0.0.1:500 name = "daemon_rpc_addr" type = "crate::config::ResolvAddr" doc = "Bitcoin daemon JSONRPC 'addr:port' to connect (default: 127.0.0.1:8332 for mainnet, 127.0.0.1:18332 for testnet, 127.0.0.1:18443 for regtest and 127.0.0.1:18554 for signet)" -[[param]] -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 = "monitoring_addr" @@ -98,12 +94,6 @@ type = "u64" doc = "Duration to wait until bitcoind JSON-RPC timeouts (must be greater than wait_duration_secs)." default = "15" -[[param]] -name = "index_batch_size" -type = "usize" -doc = "Number of blocks to get in a single p2p protocol request from bitcoind" -default = "10" - [[switch]] name = "ignore_mempool" doc = "Don't sync mempool - queries will show only confirmed transactions." @@ -146,8 +136,3 @@ default = "concat!(\"Welcome to electrs \", env!(\"CARGO_PKG_VERSION\"), \" (Ele name = "log_filters" type = "String" doc = "Logging filters, overriding `RUST_LOG` environment variable (see https://docs.rs/env_logger/ for details)" - -[[param]] -name = "magic" -type = "String" -doc = "network magic for custom network in hex format, as found in Bitcoin Core logs" diff --git a/server.sh b/server.sh index c66a4c9..a4d8301 100755 --- a/server.sh +++ b/server.sh @@ -8,8 +8,8 @@ cargo build --all --features "metrics_process" --release NETWORK=$1 shift -DB=${DB-./db} -export RUST_LOG=${RUST_LOG-electrs=INFO} +DB=${DB-./_db} +export RUST_LOG=${RUST_LOG-INFO} target/release/electrs --network $NETWORK --db-dir $DB --daemon-dir $HOME/.bitcoin $* # use SIGINT to quit diff --git a/src/cache.rs b/src/cache.rs deleted file mode 100644 index b09a916..0000000 --- a/src/cache.rs +++ /dev/null @@ -1,43 +0,0 @@ -use bitcoin::Txid; -use parking_lot::RwLock; - -use std::collections::HashMap; -use std::sync::Arc; - -use crate::metrics::{self, Histogram, Metrics}; - -pub(crate) struct Cache { - txs: Arc>>>, - - // stats - txs_size: Histogram, -} - -impl Cache { - pub fn new(metrics: &Metrics) -> Self { - Cache { - txs: Default::default(), - txs_size: metrics.histogram_vec( - "cache_txs_size", - "Cached transactions' size (in bytes)", - "type", - metrics::default_size_buckets(), - ), - } - } - - pub fn add_tx(&self, txid: Txid, f: impl FnOnce() -> Box<[u8]>) { - self.txs.write().entry(txid).or_insert_with(|| { - let tx = f(); - self.txs_size.observe("serialized", tx.len() as f64); - tx - }); - } - - pub fn get_tx(&self, txid: &Txid, f: F) -> Option - where - F: FnOnce(&[u8]) -> T, - { - self.txs.read().get(txid).map(|tx_bytes| f(tx_bytes)) - } -} diff --git a/src/chain.rs b/src/chain.rs deleted file mode 100644 index a55e50d..0000000 --- a/src/chain.rs +++ /dev/null @@ -1,261 +0,0 @@ -use std::collections::HashMap; - -use bitcoin::blockdata::block::Header as BlockHeader; -use bitcoin::{BlockHash, Network}; - -/// A new header found, to be added to the chain at specific height -pub(crate) struct NewHeader { - header: BlockHeader, - hash: BlockHash, - height: usize, -} - -impl NewHeader { - pub(crate) fn from((header, height): (BlockHeader, usize)) -> Self { - Self { - header, - hash: header.block_hash(), - height, - } - } - - pub(crate) fn height(&self) -> usize { - self.height - } - - pub(crate) fn hash(&self) -> BlockHash { - self.hash - } -} - -/// Current blockchain headers' list -pub struct Chain { - headers: Vec<(BlockHash, BlockHeader)>, - heights: HashMap, -} - -impl Chain { - // create an empty chain - pub fn new(network: Network) -> Self { - let genesis = bitcoin::blockdata::constants::genesis_block(network); - let genesis_hash = genesis.block_hash(); - Self { - headers: vec![(genesis_hash, genesis.header)], - heights: std::iter::once((genesis_hash, 0)).collect(), // genesis header @ zero height - } - } - - pub(crate) fn drop_last_headers(&mut self, n: usize) { - if n == 0 { - return; - } - let new_height = self.height().saturating_sub(n); - self.update(vec![NewHeader::from(( - self.headers[new_height].1, - new_height, - ))]); - } - - /// Load the chain from a collection of headers, up to the given tip - pub(crate) fn load(&mut self, headers: impl Iterator, tip: BlockHash) { - let genesis_hash = self.headers[0].0; - - let header_map: HashMap = - headers.map(|h| (h.block_hash(), h)).collect(); - let mut blockhash = tip; - let mut new_headers: Vec<&BlockHeader> = Vec::with_capacity(header_map.len()); - while blockhash != genesis_hash { - let header = match header_map.get(&blockhash) { - Some(header) => header, - None => panic!("missing header {} while loading from DB", blockhash), - }; - blockhash = header.prev_blockhash; - new_headers.push(header); - } - info!("loading {} headers, tip={}", new_headers.len(), tip); - let new_headers = new_headers.into_iter().rev().copied(); // order by height - self.update(new_headers.zip(1..).map(NewHeader::from).collect()) - } - - /// Get the block hash at specified height (if exists) - pub(crate) fn get_block_hash(&self, height: usize) -> Option { - self.headers.get(height).map(|(hash, _header)| *hash) - } - - /// Get the block header at specified height (if exists) - pub(crate) fn get_block_header(&self, height: usize) -> Option<&BlockHeader> { - self.headers.get(height).map(|(_hash, header)| header) - } - - /// Get the block height given the specified hash (if exists) - pub(crate) fn get_block_height(&self, blockhash: &BlockHash) -> Option { - self.heights.get(blockhash).copied() - } - - /// Update the chain with a list of new headers (possibly a reorg) - pub(crate) fn update(&mut self, headers: Vec) { - if let Some(first_height) = headers.first().map(|h| h.height) { - for (hash, _header) in self.headers.drain(first_height..) { - assert!(self.heights.remove(&hash).is_some()); - } - for (h, height) in headers.into_iter().zip(first_height..) { - assert_eq!(h.height, height); - assert_eq!(h.hash, h.header.block_hash()); - assert!(self.heights.insert(h.hash, h.height).is_none()); - self.headers.push((h.hash, h.header)); - } - info!( - "chain updated: tip={}, height={}", - self.headers.last().unwrap().0, - self.headers.len() - 1 - ); - } - } - - /// Best block hash - pub(crate) fn tip(&self) -> BlockHash { - self.headers.last().expect("empty chain").0 - } - - /// Number of blocks (excluding genesis block) - pub(crate) fn height(&self) -> usize { - self.headers.len() - 1 - } - - /// List of block hashes for efficient fork detection and block/header sync - /// see https://en.bitcoin.it/wiki/Protocol_documentation#getblocks - pub(crate) fn locator(&self) -> Vec { - let mut result = vec![]; - let mut index = self.headers.len() - 1; - let mut step = 1; - loop { - if result.len() >= 10 { - step *= 2; - } - result.push(self.headers[index].0); - if index == 0 { - break; - } - index = index.saturating_sub(step); - } - result - } -} - -#[cfg(test)] -mod tests { - use super::{Chain, NewHeader}; - use bitcoin::blockdata::block::Header as BlockHeader; - use bitcoin::consensus::deserialize; - use bitcoin::Network::Regtest; - use hex_lit::hex; - - #[test] - fn test_genesis() { - let regtest = Chain::new(Regtest); - assert_eq!(regtest.height(), 0); - assert_eq!( - regtest.tip(), - "0f9188f13cb7b2c71f2a335e3a4fc328bf5beb436012afca590b1a11466e2206" - .parse() - .unwrap() - ); - } - - #[test] - fn test_updates() { - let byte_headers = [ -hex!("0000002006226e46111a0b59caaf126043eb5bbf28c34f3a5e332a1fc7b2b73cf188910f1d14d3c7ff12d6adf494ebbcfba69baa915a066358b68a2b8c37126f74de396b1d61cc60ffff7f2000000000"), -hex!("00000020d700ae5d3c705702e0a5d9ababd22ded079f8a63b880b1866321d6bfcb028c3fc816efcf0e84ccafa1dda26be337f58d41b438170c357cda33a68af5550590bc1e61cc60ffff7f2004000000"), -hex!("00000020d13731bc59bc0989e06a5e7cab9843a4e17ad65c7ca47cd77f50dfd24f1f55793f7f342526aca9adb6ce8f33d8a07662c97d29d83b9e18117fb3eceecb2ab99b1e61cc60ffff7f2001000000"), -hex!("00000020a603def3e1255cadfb6df072946327c58b344f9bfb133e8e3e280d1c2d55b31c731a68f70219472864a7cb010cd53dc7e0f67e57f7d08b97e5e092b0c3942ad51f61cc60ffff7f2001000000"), -hex!("0000002041dd202b3b2edcdd3c8582117376347d48ff79ff97c95e5ac814820462012e785142dc360975b982ca43eecd14b4ba6f019041819d4fc5936255d7a2c45a96651f61cc60ffff7f2000000000"), -hex!("0000002072e297a2d6b633c44f3c9b1a340d06f3ce4e6bcd79ebd4c4ff1c249a77e1e37c59c7be1ca0964452e1735c0d2740f0d98a11445a6140c36b55770b5c0bcf801f1f61cc60ffff7f2000000000"), -hex!("000000200c9eb5889a8e924d1c4e8e79a716514579e41114ef37d72295df8869d6718e4ac5840f28de43ff25c7b9200aaf7873b20587c92827eaa61943484ca828bdd2e11f61cc60ffff7f2000000000"), -hex!("000000205873f322b333933e656b07881bb399dae61a6c0fa74188b5fb0e3dd71c9e2442f9e2f433f54466900407cf6a9f676913dd54aad977f7b05afcd6dcd81e98ee752061cc60ffff7f2004000000"), -hex!("00000020fd1120713506267f1dba2e1856ca1d4490077d261cde8d3e182677880df0d856bf94cfa5e189c85462813751ab4059643759ed319a81e0617113758f8adf67bc2061cc60ffff7f2000000000"), -hex!("000000200030d7f9c11ef35b89a0eefb9a5e449909339b5e7854d99804ea8d6a49bf900a0304d2e55fe0b6415949cff9bca0f88c0717884a5e5797509f89f856af93624a2061cc60ffff7f2002000000"), - ]; - let headers: Vec = byte_headers - .iter() - .map(|byte_header| deserialize(byte_header).unwrap()) - .collect(); - - for chunk_size in 1..headers.len() { - let mut regtest = Chain::new(Regtest); - let mut height = 0; - let mut tip = regtest.tip(); - for chunk in headers.chunks(chunk_size) { - let mut update = vec![]; - for header in chunk { - height += 1; - tip = header.block_hash(); - update.push(NewHeader::from((*header, height))) - } - regtest.update(update); - assert_eq!(regtest.tip(), tip); - assert_eq!(regtest.height(), height); - } - assert_eq!(regtest.tip(), headers.last().unwrap().block_hash()); - assert_eq!(regtest.height(), headers.len()); - } - - // test loading from a list of headers and tip - let mut regtest = Chain::new(Regtest); - regtest.load( - headers.iter().copied(), - headers.last().unwrap().block_hash(), - ); - assert_eq!(regtest.height(), headers.len()); - - // test getters - for (header, height) in headers.iter().zip(1usize..) { - assert_eq!(regtest.get_block_header(height), Some(header)); - assert_eq!(regtest.get_block_hash(height), Some(header.block_hash())); - assert_eq!(regtest.get_block_height(&header.block_hash()), Some(height)); - } - - // test chain shortening - for i in (0..=headers.len()).rev() { - let hash = regtest.get_block_hash(i).unwrap(); - assert_eq!(regtest.get_block_height(&hash), Some(i)); - assert_eq!(regtest.height(), i); - assert_eq!(regtest.tip(), hash); - regtest.drop_last_headers(1); - } - assert_eq!(regtest.height(), 0); - assert_eq!( - regtest.tip(), - "0f9188f13cb7b2c71f2a335e3a4fc328bf5beb436012afca590b1a11466e2206" - .parse() - .unwrap() - ); - - regtest.drop_last_headers(1); - assert_eq!(regtest.height(), 0); - assert_eq!( - regtest.tip(), - "0f9188f13cb7b2c71f2a335e3a4fc328bf5beb436012afca590b1a11466e2206" - .parse() - .unwrap() - ); - - // test reorg - let mut regtest = Chain::new(Regtest); - regtest.load( - headers.iter().copied(), - headers.last().unwrap().block_hash(), - ); - let height = regtest.height(); - - let new_header: BlockHeader = deserialize(&hex!("000000200030d7f9c11ef35b89a0eefb9a5e449909339b5e7854d99804ea8d6a49bf900a0304d2e55fe0b6415949cff9bca0f88c0717884a5e5797509f89f856af93624a7a6bcc60ffff7f2000000000")).unwrap(); - regtest.update(vec![NewHeader::from((new_header, height))]); - assert_eq!(regtest.height(), height); - assert_eq!( - regtest.tip(), - "0e16637fe0700a7c52e9a6eaa58bd6ac7202652103be8f778680c66f51ad2e9b" - .parse() - .unwrap() - ); - } -} diff --git a/src/config.rs b/src/config.rs index b844dda..d147408 100644 --- a/src/config.rs +++ b/src/config.rs @@ -1,6 +1,5 @@ -use bitcoin::p2p::Magic; -use bitcoin::Network; -use bitcoincore_rpc::Auth; +use crate::bitcoin::Network; +use crate::types::Auth; use dirs_next::home_dir; use std::ffi::{OsStr, OsString}; @@ -17,7 +16,7 @@ pub const ELECTRS_VERSION: &str = env!("CARGO_PKG_VERSION"); const DEFAULT_SERVER_ADDRESS: [u8; 4] = [127, 0, 0, 1]; // by default, serve on IPv4 localhost mod internal { - #![allow(unused_attributes, unused_imports)] + #![allow(unused_attributes, unused_imports, clippy::enum_variant_names)] include!(concat!(env!("OUT_DIR"), "/configure_me_config.rs")); } @@ -122,21 +121,20 @@ impl From for Network { } /// Parsed and post-processed configuration +#[allow(dead_code)] #[derive(Debug)] pub struct Config { // See below for the documentation of each field: pub network: Network, - pub db_path: PathBuf, + pub db_dir: PathBuf, pub db_log_dir: Option, pub db_parallelism: u8, pub daemon_auth: SensitiveAuth, pub daemon_rpc_addr: SocketAddr, - pub daemon_p2p_addr: SocketAddr, pub electrum_rpc_addr: SocketAddr, pub monitoring_addr: SocketAddr, pub wait_duration: Duration, pub jsonrpc_timeout: Duration, - pub index_batch_size: usize, pub index_lookup_limit: Option, pub reindex_last_blocks: usize, pub auto_reindex: bool, @@ -145,7 +143,6 @@ pub struct Config { pub skip_block_download_wait: bool, pub disable_electrum_rpc: bool, pub server_banner: String, - pub magic: Magic, } pub struct SensitiveAuth(pub Auth); @@ -198,16 +195,6 @@ impl Config { internal::prelude::Config::including_optional_config_files(default_config_files()) .unwrap_or_exit(); - let db_subdir = match config.network { - Network::Bitcoin => "bitcoin", - Network::Testnet => "testnet", - Network::Testnet4 => "testnet4", - Network::Regtest => "regtest", - Network::Signet => "signet", - }; - - config.db_dir.push(db_subdir); - let default_daemon_rpc_port = match config.network { Network::Bitcoin => 8332, Network::Testnet => 18332, @@ -215,13 +202,6 @@ impl Config { Network::Regtest => 18443, Network::Signet => 38332, }; - let default_daemon_p2p_port = match config.network { - Network::Bitcoin => 8333, - Network::Testnet => 18333, - Network::Testnet4 => 48333, - Network::Regtest => 18444, - Network::Signet => 38333, - }; let default_electrum_port = match config.network { Network::Bitcoin => 50001, Network::Testnet => 60001, @@ -237,25 +217,10 @@ impl Config { Network::Signet => 34224, }; - let magic = match config.magic { - Some(magic_hex) => magic_hex.parse().unwrap_or_else(|error| { - eprintln!( - "Error: magic '{}' is not a valid hex string: {}", - magic_hex, error - ); - std::process::exit(1); - }), - None => config.network.magic(), - }; - let daemon_rpc_addr: SocketAddr = config.daemon_rpc_addr.map_or( (DEFAULT_SERVER_ADDRESS, default_daemon_rpc_port).into(), ResolvAddr::resolve_or_exit, ); - let daemon_p2p_addr: SocketAddr = config.daemon_p2p_addr.map_or( - (DEFAULT_SERVER_ADDRESS, default_daemon_p2p_port).into(), - ResolvAddr::resolve_or_exit, - ); let electrum_rpc_addr: SocketAddr = config.electrum_rpc_addr.map_or( (DEFAULT_SERVER_ADDRESS, default_electrum_port).into(), ResolvAddr::resolve_or_exit, @@ -339,17 +304,15 @@ impl Config { let config = Config { network: config.network, - db_path: config.db_dir, + db_dir: config.db_dir, db_log_dir: config.db_log_dir, db_parallelism: config.db_parallelism, daemon_auth, daemon_rpc_addr, - daemon_p2p_addr, electrum_rpc_addr, monitoring_addr, wait_duration: Duration::from_secs(config.wait_duration_secs), jsonrpc_timeout: Duration::from_secs(config.jsonrpc_timeout_secs), - index_batch_size: config.index_batch_size, index_lookup_limit, reindex_last_blocks: config.reindex_last_blocks, auto_reindex: config.auto_reindex, @@ -358,7 +321,6 @@ impl Config { skip_block_download_wait: config.skip_block_download_wait, disable_electrum_rpc: config.disable_electrum_rpc, server_banner: config.server_banner, - magic, }; eprintln!( "Starting electrs {} on {} {} with {:?}", @@ -382,9 +344,6 @@ mod tests { #[test] fn test_auth_debug() { - let auth = Auth::None; - assert_eq!(format!("{:?}", SensitiveAuth(auth)), "None"); - let auth = Auth::CookieFile(Path::new("/foo/bar/.cookie").to_path_buf()); assert_eq!( format!("{:?}", SensitiveAuth(auth)), diff --git a/src/daemon.rs b/src/daemon.rs index 40df169..f31337e 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -1,11 +1,10 @@ use anyhow::{Context, Result}; -use bitcoin::consensus::encode::serialize_hex; -use bitcoin::{consensus::deserialize, hashes::hex::FromHex}; -use bitcoin::{Amount, BlockHash, Transaction, Txid}; -use bitcoincore_rpc::{json, jsonrpc, Auth, Client, RpcApi}; -use crossbeam_channel::Receiver; -use parking_lot::Mutex; +use crate::bitcoin::{ + self, consensus::deserialize, consensus::encode::serialize_hex, hashes::hex::FromHex, Amount, + BlockHash, Transaction, Txid, +}; +use crate::types::Auth; use serde::Serialize; use serde_json::{json, value::RawValue, Value}; @@ -13,52 +12,62 @@ use std::fs::File; use std::io::Read; use std::path::Path; -use crate::{ - chain::{Chain, NewHeader}, - config::Config, - metrics::Metrics, - p2p::Connection, - signals::ExitFlag, - types::SerBlock, -}; +use crate::{config::Config, signals::ExitFlag}; enum PollResult { Done(Result<()>), Retry, } -fn rpc_poll(client: &mut Client, skip_block_download_wait: bool) -> PollResult { - match client.get_blockchain_info() { - Ok(info) => { - if skip_block_download_wait { - // bitcoind RPC is available, don't wait for block download to finish - return PollResult::Done(Ok(())); - } - let left_blocks = info.headers - info.blocks; - if info.initial_block_download || left_blocks > 0 { - info!( - "waiting for {} blocks to download{}", - left_blocks, - if info.initial_block_download { - " (IBD)" - } else { - "" - } - ); - return PollResult::Retry; - } - PollResult::Done(Ok(())) - } - Err(err) => { - if let Some(e) = extract_bitcoind_error(&err) { - if e.code == -28 { - debug!("waiting for RPC warmup: {}", e.message); - return PollResult::Retry; - } - } - PollResult::Done(Err(err).context("daemon not available")) - } - } +#[derive(Deserialize)] +pub struct GetMempoolEntryResultFees { + /// Transaction fee in BTC + #[serde(with = "bitcoin::amount::serde::as_btc")] + pub base: Amount, +} + +#[derive(Deserialize)] +pub struct GetMempoolEntryResult { + /// Virtual transaction size as defined in BIP 141. This is different from actual serialized + /// size for witness transactions as witness data is discounted. + #[serde(alias = "size")] + pub vsize: u64, + /// Fee information + pub fees: GetMempoolEntryResultFees, + /// Unconfirmed transactions used as inputs for this transaction + pub depends: Vec, +} + +#[derive(Deserialize)] +pub struct GetBlockchainInfoResult { + pub blocks: u64, + pub headers: u64, + pub initialblockdownload: bool, + pub pruned: bool, +} + +#[derive(Deserialize)] +pub struct EstimateSmartFeeResult { + /// Estimate fee rate in BTC/kB. + #[serde( + default, + rename = "feerate", + skip_serializing_if = "Option::is_none", + with = "bitcoin::amount::serde::as_btc::opt" + )] + pub fee_rate: Option, +} + +#[derive(Deserialize)] +pub struct GetNetworkInfoResult { + #[serde(rename = "relayfee", with = "bitcoin::amount::serde::as_btc")] + pub relay_fee: Amount, +} + +#[derive(Deserialize)] +pub struct GetMempoolInfoResult { + /// True if the mempool is fully loaded + pub loaded: Option, } fn read_cookie(path: &Path) -> Result<(String, String)> { @@ -79,7 +88,7 @@ fn read_cookie(path: &Path) -> Result<(String, String)> { Ok((parts[0].to_owned(), parts[1].to_owned())) } -fn rpc_connect(config: &Config) -> Result { +fn jsonrpc_client(config: &Config) -> Result { let rpc_url = format!("http://{}", config.daemon_rpc_addr); // Allow RPC calls to take longer before timing out. // See https://github.com/romanz/electrs/issues/495 for more details. @@ -87,36 +96,30 @@ fn rpc_connect(config: &Config) -> Result { .url(&rpc_url)? .timeout(config.jsonrpc_timeout); let builder = match config.daemon_auth.get_auth() { - Auth::None => builder, Auth::UserPass(user, pass) => builder.auth(user, Some(pass)), Auth::CookieFile(path) => { let (user, pass) = read_cookie(&path)?; builder.auth(user, Some(pass)) } }; - Ok(Client::from_jsonrpc(jsonrpc::Client::with_transport( - builder.build(), - ))) + Ok(jsonrpc::Client::with_transport(builder.build())) } pub struct Daemon { - p2p: Mutex, - rpc: Client, + client: jsonrpc::Client, } impl Daemon { - pub(crate) fn connect( - config: &Config, - exit_flag: &ExitFlag, - metrics: &Metrics, - ) -> Result { - let mut rpc = rpc_connect(config)?; + pub(crate) fn connect(config: &Config, exit_flag: &ExitFlag) -> Result { + let daemon = Daemon { + client: jsonrpc_client(config)?, + }; loop { exit_flag .poll() .context("bitcoin RPC polling interrupted")?; - match rpc_poll(&mut rpc, config.skip_block_download_wait) { + match daemon.rpc_poll(config.skip_block_download_wait) { PollResult::Done(result) => { result.context("bitcoind RPC polling failed")?; break; // on success, finish polling @@ -127,33 +130,85 @@ impl Daemon { } } - let network_info = rpc.get_network_info()?; - 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 info = rpc.get_blockchain_info()?; - if info.pruned { - bail!("electrs requires non-pruned bitcoind node"); - } + Ok(daemon) + } - let p2p = Mutex::new(Connection::connect( - config.daemon_p2p_addr, - metrics, - config.magic, - )?); - Ok(Self { p2p, rpc }) + fn call serde::de::Deserialize<'a>>( + &self, + cmd: &str, + args: &[serde_json::Value], + ) -> Result { + let raw = serde_json::value::to_raw_value(args)?; + let req = self.client.build_request(cmd, Some(&*raw)); + let resp = self.client.send_request(req)?; + resp.result() + } + + fn batch_request(&self, name: &str, items: &[T]) -> Result>> + where + T: Serialize, + { + debug!("calling {} on {} items", name, items.len()); + let args: Vec> = items + .iter() + .map(|item| jsonrpc::try_arg([item]).context("failed to serialize into JSON")) + .collect::>>()?; + let reqs: Vec = args + .iter() + .map(|arg| self.client.build_request(name, Some(arg))) + .collect(); + match self.client.send_batch(&reqs) { + Ok(values) => { + assert_eq!(items.len(), values.len()); + Ok(values) + } + Err(err) => bail!("batch {} request failed: {}", name, err), + } + } + + fn rpc_poll(&self, skip_block_download_wait: bool) -> PollResult { + match self.call::("getblockchaininfo", &[]) { + Ok(info) => { + if info.pruned { + return PollResult::Done(Err(anyhow!( + "electrs requires non-pruned bitcoind node" + ))); + } + if skip_block_download_wait { + // bitcoind RPC is available, don't wait for block download to finish + return PollResult::Done(Ok(())); + } + let left_blocks = info.headers - info.blocks; + if info.initialblockdownload || left_blocks > 0 { + info!( + "waiting for {} blocks to download{}", + left_blocks, + if info.initialblockdownload { + " (IBD)" + } else { + "" + } + ); + return PollResult::Retry; + } + PollResult::Done(Ok(())) + } + Err(err) => { + if let Some(e) = extract_bitcoind_error(&err) { + if e.code == -28 { + debug!("waiting for RPC warmup: {}", e.message); + return PollResult::Retry; + } + } + PollResult::Done(Err(err).context("daemon not available")) + } + } } pub(crate) fn estimate_fee(&self, nblocks: u16) -> Result> { - let res = self.rpc.estimate_smart_fee(nblocks, None); - if let Err(bitcoincore_rpc::Error::JsonRpc(jsonrpc::Error::Rpc(RpcError { - code: -32603, - .. - }))) = res - { + let res = + self.call::("estimatesmartfee", &[json!(nblocks), Value::Null]); + if let Err(jsonrpc::Error::Rpc(jsonrpc::error::RpcError { code: -32603, .. })) = res { return Ok(None); // don't fail when fee estimation is disabled (e.g. with `-blocksonly=1`) } Ok(res.context("failed to estimate fee")?.fee_rate) @@ -161,91 +216,64 @@ impl Daemon { pub(crate) fn get_relay_fee(&self) -> Result { Ok(self - .rpc - .get_network_info() + .call::("getnetworkinfo", &[]) .context("failed to get relay fee")? .relay_fee) } pub(crate) fn broadcast(&self, tx: &Transaction) -> Result { - self.rpc - .send_raw_transaction(tx) + self.call("sendrawtransaction", &[json!(serialize_hex(tx))]) .context("failed to broadcast transaction") } pub(crate) fn submitpackage(&self, txs: &[Transaction]) -> Result { let package: Vec = txs.iter().map(serialize_hex).collect(); - self.rpc - .call("submitpackage", &[json!(package)]) + self.call("submitpackage", &[json!(package)]) .context("failed to submitpackage package") } - pub(crate) fn get_transaction_info( - &self, - txid: &Txid, - blockhash: Option, - ) -> Result { - // No need to parse the resulting JSON, just return it as-is to the client. - self.rpc - .call( - "getrawtransaction", - &[json!(txid), json!(true), json!(blockhash)], - ) - .context("failed to get transaction info") - } - - pub(crate) fn get_transaction_hex( - &self, - txid: &Txid, - blockhash: Option, - ) -> Result { - use bitcoin::consensus::serde::{hex::Lower, Hex, With}; - - let tx = self.get_transaction(txid, blockhash)?; - #[derive(serde::Serialize)] - #[serde(transparent)] - struct TxAsHex(#[serde(with = "With::>")] Transaction); - serde_json::to_value(TxAsHex(tx)).map_err(Into::into) - } - pub(crate) fn get_transaction( &self, txid: &Txid, blockhash: Option, - ) -> Result { - self.rpc - .get_raw_transaction(txid, blockhash.as_ref()) - .context("failed to get transaction") + verbose: bool, + ) -> Result { + self.call( + "getrawtransaction", + &[json!(txid), json!(verbose), json!(blockhash)], + ) + .context("failed to get transaction") } pub(crate) fn get_block_txids(&self, blockhash: BlockHash) -> Result> { - Ok(self - .rpc - .get_block_info(&blockhash) - .context("failed to get block txids")? - .tx) + #[derive(Clone, PartialEq, Debug, Deserialize, Serialize)] + struct GetBlockResult { + pub tx: Vec, + } + let res: GetBlockResult = self + .call("getblock", &[json!(blockhash), json!(1)]) + .context("failed to get block txids")?; + Ok(res.tx) } - pub(crate) fn get_mempool_info(&self) -> Result { - self.rpc - .get_mempool_info() + pub(crate) fn get_mempool_info(&self) -> Result { + self.call("getmempoolinfo", &[]) .context("failed to get mempool info") } pub(crate) fn get_mempool_txids(&self) -> Result> { - self.rpc - .get_raw_mempool() + self.call("getrawmempool", &[]) .context("failed to get mempool txids") } pub(crate) fn get_mempool_entries( &self, txids: &[Txid], - ) -> Result>> { - let results = batch_request(self.rpc.get_jsonrpc_client(), "getmempoolentry", txids)?; + ) -> Result>> { + let results = self.batch_request("getmempoolentry", txids)?; Ok(results .into_iter() - .map(|r| match r?.result::() { + .map(|r| match r?.result::() { Ok(entry) => Some(entry), Err(err) => { debug!("failed to get mempool entry: {}", err); // probably due to RBF @@ -259,7 +287,7 @@ impl Daemon { &self, txids: &[Txid], ) -> Result>> { - let results = batch_request(self.rpc.get_jsonrpc_client(), "getrawtransaction", txids)?; + let results = self.batch_request("getrawtransaction", txids)?; Ok(results .into_iter() .map(|r| -> Option { @@ -287,58 +315,11 @@ impl Daemon { }) .collect()) } - - pub(crate) fn get_new_headers(&self, chain: &Chain) -> Result> { - self.p2p.lock().get_new_headers(chain) - } - - pub(crate) fn for_blocks(&self, blockhashes: B, func: F) -> Result<()> - where - B: IntoIterator, - F: FnMut(BlockHash, SerBlock), - { - self.p2p.lock().for_blocks(blockhashes, func) - } - - pub(crate) fn new_block_notification(&self) -> Receiver<()> { - self.p2p.lock().new_block_notification() - } } -pub(crate) type RpcError = bitcoincore_rpc::jsonrpc::error::RpcError; - -pub(crate) fn extract_bitcoind_error(err: &bitcoincore_rpc::Error) -> Option<&RpcError> { - use bitcoincore_rpc::{ - jsonrpc::error::Error::Rpc as ServerError, Error::JsonRpc as JsonRpcError, - }; +pub(crate) fn extract_bitcoind_error(err: &jsonrpc::Error) -> Option<&jsonrpc::error::RpcError> { match err { - JsonRpcError(ServerError(e)) => Some(e), + jsonrpc::error::Error::Rpc(e) => Some(e), _ => None, } } - -fn batch_request( - client: &jsonrpc::Client, - name: &str, - items: &[T], -) -> Result>> -where - T: Serialize, -{ - debug!("calling {} on {} items", name, items.len()); - let args: Vec> = items - .iter() - .map(|item| jsonrpc::try_arg([item]).context("failed to serialize into JSON")) - .collect::>>()?; - let reqs: Vec = args - .iter() - .map(|arg| client.build_request(name, Some(arg))) - .collect(); - match client.send_batch(&reqs) { - Ok(values) => { - assert_eq!(items.len(), values.len()); - Ok(values) - } - Err(err) => bail!("batch {} request failed: {}", name, err), - } -} diff --git a/src/db.rs b/src/db.rs deleted file mode 100644 index c6b156b..0000000 --- a/src/db.rs +++ /dev/null @@ -1,554 +0,0 @@ -use anyhow::{Context, Result}; -use rust_rocksdb as rocksdb; - -use std::path::Path; -use std::sync::atomic::{AtomicBool, Ordering}; - -use crate::types::{HashPrefix, SerializedHashPrefixRow, SerializedHeaderRow}; - -#[derive(Default)] -pub(crate) struct WriteBatch { - pub(crate) tip_row: [u8; 32], - pub(crate) header_rows: Vec, - pub(crate) funding_rows: Vec, - pub(crate) spending_rows: Vec, - pub(crate) txid_rows: Vec, -} - -impl WriteBatch { - pub(crate) fn sort(&mut self) { - self.header_rows.sort_unstable(); - self.funding_rows.sort_unstable(); - self.spending_rows.sort_unstable(); - self.txid_rows.sort_unstable(); - } -} - -/// RocksDB wrapper for index storage -pub struct DBStore { - db: rocksdb::DB, - bulk_import: AtomicBool, -} - -const CONFIG_CF: &str = "config"; -const HEADERS_CF: &str = "headers"; -const TXID_CF: &str = "txid"; -const FUNDING_CF: &str = "funding"; -const SPENDING_CF: &str = "spending"; - -const COLUMN_FAMILIES: &[&str] = &[CONFIG_CF, HEADERS_CF, TXID_CF, FUNDING_CF, SPENDING_CF]; - -const CONFIG_KEY: &str = "C"; -const TIP_KEY: &[u8] = b"T"; - -// Taken from https://github.com/facebook/rocksdb/blob/master/include/rocksdb/db.h#L654-L689 -const DB_PROPERTIES: &[&str] = &[ - "rocksdb.num-immutable-mem-table", - "rocksdb.mem-table-flush-pending", - "rocksdb.compaction-pending", - "rocksdb.background-errors", - "rocksdb.cur-size-active-mem-table", - "rocksdb.cur-size-all-mem-tables", - "rocksdb.size-all-mem-tables", - "rocksdb.num-entries-active-mem-table", - "rocksdb.num-entries-imm-mem-tables", - "rocksdb.num-deletes-active-mem-table", - "rocksdb.num-deletes-imm-mem-tables", - "rocksdb.estimate-num-keys", - "rocksdb.estimate-table-readers-mem", - "rocksdb.is-file-deletions-enabled", - "rocksdb.num-snapshots", - "rocksdb.oldest-snapshot-time", - "rocksdb.num-live-versions", - "rocksdb.current-super-version-number", - "rocksdb.estimate-live-data-size", - "rocksdb.min-log-number-to-keep", - "rocksdb.min-obsolete-sst-number-to-keep", - "rocksdb.total-sst-files-size", - "rocksdb.live-sst-files-size", - "rocksdb.base-level", - "rocksdb.estimate-pending-compaction-bytes", - "rocksdb.num-running-compactions", - "rocksdb.num-running-flushes", - "rocksdb.actual-delayed-write-rate", - "rocksdb.is-write-stopped", - "rocksdb.estimate-oldest-key-time", - "rocksdb.block-cache-capacity", - "rocksdb.block-cache-usage", - "rocksdb.block-cache-pinned-usage", -]; - -#[derive(Debug, Deserialize, Serialize)] -struct Config { - compacted: bool, - format: u64, -} - -const CURRENT_FORMAT: u64 = 0; - -impl Default for Config { - fn default() -> Self { - Config { - compacted: false, - format: CURRENT_FORMAT, - } - } -} - -fn default_opts(parallelism: u8) -> rocksdb::Options { - let mut block_opts = rocksdb::BlockBasedOptions::default(); - block_opts.set_checksum_type(rocksdb::ChecksumType::CRC32c); - - let mut opts = rocksdb::Options::default(); - opts.increase_parallelism(parallelism.into()); - opts.set_max_subcompactions(parallelism.into()); - - opts.set_keep_log_file_num(10); - opts.set_max_open_files(16); - opts.set_compaction_style(rocksdb::DBCompactionStyle::Level); - opts.set_compression_type(rocksdb::DBCompressionType::Zstd); - opts.set_target_file_size_base(256 << 20); - opts.set_write_buffer_size(256 << 20); - opts.set_disable_auto_compactions(true); // for initial bulk load - opts.set_advise_random_on_open(false); // bulk load uses sequential I/O - opts.set_prefix_extractor(rocksdb::SliceTransform::create_fixed_prefix(8)); - opts.set_block_based_table_factory(&block_opts); - opts -} - -impl DBStore { - fn create_cf_descriptors(parallelism: u8) -> Vec { - COLUMN_FAMILIES - .iter() - .map(|&name| rocksdb::ColumnFamilyDescriptor::new(name, default_opts(parallelism))) - .collect() - } - - fn open_internal(path: &Path, log_dir: Option<&Path>, parallelism: u8) -> Result { - let mut db_opts = default_opts(parallelism); - db_opts.create_if_missing(true); - db_opts.create_missing_column_families(true); - if let Some(d) = log_dir { - db_opts.set_db_log_dir(d); - } - - let db = rocksdb::DB::open_cf_descriptors( - &db_opts, - path, - Self::create_cf_descriptors(parallelism), - ) - .with_context(|| format!("failed to open DB: {}", path.display()))?; - let live_files = db.live_files()?; - info!( - "{:?}: {} SST files, {} GB, {} Grows", - path, - live_files.len(), - live_files.iter().map(|f| f.size).sum::() as f64 / 1e9, - live_files.iter().map(|f| f.num_entries).sum::() as f64 / 1e9 - ); - let store = DBStore { - db, - bulk_import: AtomicBool::new(true), - }; - Ok(store) - } - - fn is_legacy_format(&self) -> bool { - // In legacy DB format, all data was stored in a single (default) column family. - self.db - .iterator(rocksdb::IteratorMode::Start) - .next() - .is_some() - } - - /// Opens a new RocksDB at the specified location. - pub fn open( - path: &Path, - log_dir: Option<&Path>, - auto_reindex: bool, - parallelism: u8, - ) -> Result { - let mut store = Self::open_internal(path, log_dir, parallelism)?; - let config = store.get_config(); - debug!("DB {:?}", config); - let mut config = config.unwrap_or_default(); // use default config when DB is empty - - let reindex_cause = if store.is_legacy_format() { - Some("legacy format".to_owned()) - } else if config.format != CURRENT_FORMAT { - Some(format!( - "unsupported format {} != {}", - config.format, CURRENT_FORMAT - )) - } else { - None - }; - if let Some(cause) = reindex_cause { - if !auto_reindex { - bail!("re-index required due to {}", cause); - } - warn!( - "Database needs to be re-indexed due to {}, going to delete {}", - cause, - path.display() - ); - // close DB before deletion - drop(store); - rocksdb::DB::destroy(&default_opts(parallelism), path).with_context(|| { - format!( - "re-index required but the old database ({}) can not be deleted", - path.display() - ) - })?; - store = Self::open_internal(path, log_dir, parallelism)?; - config = Config::default(); // re-init config after dropping DB - } - if config.compacted { - store.start_compactions(); - } - store.set_config(config); - Ok(store) - } - - fn config_cf(&self) -> &rocksdb::ColumnFamily { - self.db.cf_handle(CONFIG_CF).expect("missing CONFIG_CF") - } - - fn funding_cf(&self) -> &rocksdb::ColumnFamily { - self.db.cf_handle(FUNDING_CF).expect("missing FUNDING_CF") - } - - fn spending_cf(&self) -> &rocksdb::ColumnFamily { - self.db.cf_handle(SPENDING_CF).expect("missing SPENDING_CF") - } - - fn txid_cf(&self) -> &rocksdb::ColumnFamily { - self.db.cf_handle(TXID_CF).expect("missing TXID_CF") - } - - fn headers_cf(&self) -> &rocksdb::ColumnFamily { - self.db.cf_handle(HEADERS_CF).expect("missing HEADERS_CF") - } - - pub(crate) fn iter_funding( - &self, - prefix: HashPrefix, - ) -> impl Iterator + '_ { - self.iter_prefix_cf(self.funding_cf(), prefix) - } - - pub(crate) fn iter_spending( - &self, - prefix: HashPrefix, - ) -> impl Iterator + '_ { - self.iter_prefix_cf(self.spending_cf(), prefix) - } - - pub(crate) fn iter_txid( - &self, - prefix: HashPrefix, - ) -> impl Iterator + '_ { - self.iter_prefix_cf(self.txid_cf(), prefix) - } - - fn iter_cf( - &self, - cf: &rocksdb::ColumnFamily, - readopts: rocksdb::ReadOptions, - prefix: Option, - ) -> impl Iterator + '_ { - DBIterator::new(self.db.raw_iterator_cf_opt(cf, readopts), prefix) - } - - fn iter_prefix_cf( - &self, - cf: &rocksdb::ColumnFamily, - prefix: HashPrefix, - ) -> impl Iterator + '_ { - let mut opts = rocksdb::ReadOptions::default(); - opts.set_prefix_same_as_start(true); // requires .set_prefix_extractor() above. - self.iter_cf(cf, opts, Some(prefix)) - } - - pub(crate) fn iter_headers(&self) -> impl Iterator + '_ { - let mut opts = rocksdb::ReadOptions::default(); - opts.fill_cache(false); - self.iter_cf(self.headers_cf(), opts, None) - } - - pub(crate) fn get_tip(&self) -> Option> { - self.db - .get_cf(self.headers_cf(), TIP_KEY) - .expect("get_tip failed") - } - - pub(crate) fn write(&self, batch: &WriteBatch) { - let mut db_batch = rocksdb::WriteBatch::default(); - let funding_cf = self.funding_cf(); - for key in &batch.funding_rows { - db_batch.put_cf(funding_cf, key, b""); - } - let spending_cf = self.spending_cf(); - for key in &batch.spending_rows { - db_batch.put_cf(spending_cf, key, b""); - } - let txid_cf = self.txid_cf(); - for key in &batch.txid_rows { - db_batch.put_cf(txid_cf, key, b""); - } - let headers_cf = self.headers_cf(); - for key in &batch.header_rows { - db_batch.put_cf(headers_cf, key, b""); - } - db_batch.put_cf(headers_cf, TIP_KEY, batch.tip_row); - - let mut opts = rocksdb::WriteOptions::new(); - let bulk_import = self.bulk_import.load(Ordering::Relaxed); - opts.set_sync(!bulk_import); - opts.disable_wal(bulk_import); - self.db.write_opt(db_batch, &opts).unwrap(); - } - - pub(crate) fn flush(&self) { - debug!("flushing DB column families"); - let mut config = self.get_config().unwrap_or_default(); - for name in COLUMN_FAMILIES { - let cf = self.db.cf_handle(name).expect("missing CF"); - self.db.flush_cf(cf).expect("CF flush failed"); - } - if !config.compacted { - for name in COLUMN_FAMILIES { - info!("starting {} compaction", name); - let cf = self.db.cf_handle(name).expect("missing CF"); - self.db.compact_range_cf(cf, None::<&[u8]>, None::<&[u8]>); - } - config.compacted = true; - self.set_config(config); - info!("finished full compaction"); - self.start_compactions(); - } - if log_enabled!(log::Level::Trace) { - let stats = self - .db - .property_value("rocksdb.dbstats") - .expect("failed to get property") - .expect("missing property"); - trace!("RocksDB stats: {}", stats); - } - } - - pub(crate) fn get_properties( - &self, - ) -> impl Iterator + '_ { - COLUMN_FAMILIES.iter().flat_map(move |cf_name| { - let cf = self.db.cf_handle(cf_name).expect("missing CF"); - DB_PROPERTIES.iter().filter_map(move |property_name| { - let value = self - .db - .property_int_value_cf(cf, *property_name) - .expect("failed to get property"); - Some((*cf_name, *property_name, value?)) - }) - }) - } - - fn start_compactions(&self) { - self.bulk_import.store(false, Ordering::Relaxed); - for name in COLUMN_FAMILIES { - let cf = self.db.cf_handle(name).expect("missing CF"); - self.db - .set_options_cf(cf, &[("disable_auto_compactions", "false")]) - .expect("failed to start auto-compactions"); - } - debug!("auto-compactions enabled"); - } - - fn set_config(&self, config: Config) { - let mut opts = rocksdb::WriteOptions::default(); - opts.set_sync(true); - opts.disable_wal(false); - let value = serde_json::to_vec(&config).expect("failed to serialize config"); - self.db - .put_cf_opt(self.config_cf(), CONFIG_KEY, value, &opts) - .expect("DB::put failed"); - } - - fn get_config(&self) -> Option { - self.db - .get_cf(self.config_cf(), CONFIG_KEY) - .expect("DB::get failed") - .map(|value| serde_json::from_slice(&value).expect("failed to deserialize Config")) - } -} - -struct DBIterator<'a, const N: usize> { - raw: rocksdb::DBRawIterator<'a>, - prefix: Option, - done: bool, -} - -impl<'a, const N: usize> DBIterator<'a, N> { - fn new(mut raw: rocksdb::DBRawIterator<'a>, prefix: Option) -> Self { - match prefix { - Some(key) => raw.seek(key), - None => raw.seek_to_first(), - }; - Self { - raw, - prefix, - done: false, - } - } -} - -impl Iterator for DBIterator<'_, N> { - type Item = [u8; N]; - - fn next(&mut self) -> Option { - while !self.done { - let key = match self.raw.key() { - Some(key) => key, - None => { - self.raw.status().expect("DB scan failed"); - break; // end of scan - } - }; - let prefix_match = match self.prefix { - Some(key_prefix) => key.starts_with(&key_prefix), - None => true, - }; - if !prefix_match { - break; // prefix mismatch - } - let result: Option<[u8; N]> = key.try_into().ok(); - self.raw.next(); - match result { - Some(value) => return Some(value), - None => continue, // skip keys with size != N - } - } - self.done = true; - None - } -} - -impl Drop for DBStore { - fn drop(&mut self) { - info!("closing DB at {}", self.db.path().display()); - } -} - -#[cfg(test)] -mod tests { - use super::{rocksdb, DBStore, WriteBatch, CURRENT_FORMAT}; - use std::ffi::{OsStr, OsString}; - use std::path::Path; - - #[test] - fn test_reindex_new_format() { - let dir = tempfile::tempdir().unwrap(); - { - let store = DBStore::open(dir.path(), None, false, 1).unwrap(); - let mut config = store.get_config().unwrap(); - config.format += 1; - store.set_config(config); - }; - assert_eq!( - DBStore::open(dir.path(), None, false, 1) - .err() - .unwrap() - .to_string(), - format!( - "re-index required due to unsupported format {} != {}", - CURRENT_FORMAT + 1, - CURRENT_FORMAT - ) - ); - { - let store = DBStore::open(dir.path(), None, true, 1).unwrap(); - store.flush(); - let config = store.get_config().unwrap(); - assert_eq!(config.format, CURRENT_FORMAT); - assert!(!store.is_legacy_format()); - } - } - - #[test] - fn test_reindex_legacy_format() { - let dir = tempfile::tempdir().unwrap(); - { - let mut db_opts = rocksdb::Options::default(); - db_opts.create_if_missing(true); - let db = rocksdb::DB::open(&db_opts, dir.path()).unwrap(); - db.put(b"F", b"").unwrap(); // insert legacy DB compaction marker (in 'default' column family) - }; - assert_eq!( - DBStore::open(dir.path(), None, false, 1) - .err() - .unwrap() - .to_string(), - format!("re-index required due to legacy format",) - ); - { - let store = DBStore::open(dir.path(), None, true, 1).unwrap(); - store.flush(); - let config = store.get_config().unwrap(); - assert_eq!(config.format, CURRENT_FORMAT); - } - } - - #[test] - fn test_db_prefix_scan() { - let dir = tempfile::tempdir().unwrap(); - let store = DBStore::open(dir.path(), None, true, 1).unwrap(); - - let items = [ - *b"ab ", - *b"abcdefgh ", - *b"abcdefghj ", - *b"abcdefghjk ", - *b"abcdefghxyz ", - *b"abcdefgi ", - *b"b ", - *b"c ", - ]; - - store.write(&WriteBatch { - txid_rows: items.to_vec(), - ..Default::default() - }); - - let rows = store.iter_txid(*b"abcdefgh"); - assert_eq!(rows.collect::>(), items[1..5]); - } - - #[test] - fn test_db_log_in_same_dir() { - let dir1 = tempfile::tempdir().unwrap(); - let _store = DBStore::open(dir1.path(), None, true, 1).unwrap(); - - // LOG file is created in dir1 - let dir_files = list_log_files(dir1.path()); - assert_eq!(dir_files, vec![OsStr::new("LOG")]); - - let dir2 = tempfile::tempdir().unwrap(); - let dir3 = tempfile::tempdir().unwrap(); - let _store = DBStore::open(dir2.path(), Some(dir3.path()), true, 1).unwrap(); - - // *_LOG file is not created in dir2, but in dir3 - let dir_files = list_log_files(dir2.path()); - assert_eq!(dir_files, Vec::::new()); - - let dir_files = list_log_files(dir3.path()); - assert_eq!(dir_files.len(), 1); - assert!(dir_files[0].to_str().unwrap().ends_with("_LOG")); - } - - fn list_log_files(path: &Path) -> Vec { - path.read_dir() - .unwrap() - .map(|e| e.unwrap().file_name()) - .filter(|e| e.to_str().unwrap().contains("LOG")) - .collect() - } -} diff --git a/src/electrum.rs b/src/electrum.rs index e6ef9c6..c03a4f5 100644 --- a/src/electrum.rs +++ b/src/electrum.rs @@ -1,11 +1,11 @@ -use anyhow::{bail, Context, Result}; -use bitcoin::{ +use crate::bitcoin::{ consensus::{deserialize, encode::serialize_hex}, hashes::hex::FromHex, hex::DisplayHex, BlockHash, Transaction, Txid, }; -use crossbeam_channel::Receiver; +use anyhow::{bail, Context, Result}; +use bindex::ScriptHash; use rayon::prelude::*; use serde_derive::Deserialize; use serde_json::{self, json, Value}; @@ -17,15 +17,13 @@ use std::net::SocketAddr; use std::str::FromStr; use crate::{ - cache::Cache, config::{Config, ELECTRS_VERSION}, - daemon::{self, extract_bitcoind_error, Daemon}, + daemon::{extract_bitcoind_error, Daemon}, merkle::Proof, metrics::{self, Histogram, Metrics}, signals::Signal, status::ScriptHashStatus, tracker::Tracker, - types::ScriptHash, }; const PROTOCOL_VERSION: &str = "1.4"; @@ -114,7 +112,7 @@ enum RpcError { Standard(StandardError), // Electrum-specific errors BadRequest(anyhow::Error), - DaemonError(daemon::RpcError), + DaemonError(jsonrpc::error::RpcError), UnavailableIndex, } @@ -146,7 +144,6 @@ impl RpcError { /// Electrum RPC handler pub struct Rpc { tracker: Tracker, - cache: Cache, rpc_duration: Histogram, daemon: Daemon, signal: Signal, @@ -166,11 +163,9 @@ impl Rpc { let tracker = Tracker::new(config, metrics)?; let signal = Signal::new(); - let daemon = Daemon::connect(config, signal.exit_flag(), tracker.metrics())?; - let cache = Cache::new(tracker.metrics()); + let daemon = Daemon::connect(config, signal.exit_flag())?; Ok(Self { tracker, - cache, rpc_duration, daemon, signal, @@ -183,24 +178,17 @@ impl Rpc { &self.signal } - pub fn new_block_notification(&self) -> Receiver<()> { - self.daemon.new_block_notification() - } - pub fn sync(&mut self) -> Result { self.tracker.sync(&self.daemon, self.signal.exit_flag()) } pub fn update_client(&self, client: &mut Client) -> Result> { - let chain = self.tracker.chain(); + let headers = self.tracker.headers(); let mut notifications = client .scripthashes .par_iter_mut() .filter_map(|(scripthash, status)| -> Option> { - match self - .tracker - .update_scripthash_status(status, &self.daemon, &self.cache) - { + match self.tracker.update_scripthash_status(status) { Ok(true) => Some(Ok(notification( "blockchain.scripthash.subscribe", &[json!(scripthash), json!(status.statushash())], @@ -213,11 +201,11 @@ impl Rpc { .context("failed to update status")?; if let Some(old_tip) = client.tip { - let new_tip = self.tracker.chain().tip(); - if old_tip != new_tip { - client.tip = Some(new_tip); - let height = chain.height(); - let header = chain.get_block_header(height).unwrap(); + let new_tip = headers.tip().unwrap(); + if old_tip != new_tip.hash() { + client.tip = Some(new_tip.hash()); + let height = headers.tip_height().unwrap(); + let header = new_tip.header(); notifications.push(notification( "blockchain.headers.subscribe", &[json!({"hex": serialize_hex(&header), "height": height})], @@ -228,34 +216,36 @@ impl Rpc { } fn headers_subscribe(&self, client: &mut Client) -> Result { - let chain = self.tracker.chain(); - client.tip = Some(chain.tip()); - let height = chain.height(); - let header = chain.get_block_header(height).unwrap(); - Ok(json!({"hex": serialize_hex(header), "height": height})) + let headers = self.tracker.headers(); + let height = headers.tip_height().unwrap(); + let header = headers.tip().unwrap(); + client.tip = Some(header.hash()); + Ok(json!({"hex": serialize_hex(header.header()), "height": height})) } fn block_header(&self, (height,): (usize,)) -> Result { - let chain = self.tracker.chain(); - let header = match chain.get_block_header(height) { + let header = match self.tracker.headers().iter_headers().nth(height) { None => bail!("no header at {}", height), Some(header) => header, }; - Ok(json!(serialize_hex(header))) + Ok(json!(serialize_hex(header.header()))) } fn block_headers(&self, (start_height, count): (usize, usize)) -> Result { - let chain = self.tracker.chain(); + let headers = self.tracker.headers(); let max_count = 2016usize; // return only the available block headers let end_height = std::cmp::min( - chain.height() + 1, + headers.tip_height().map_or(0, |h| h + 1), start_height + std::cmp::min(count, max_count), ); let heights = start_height..end_height; let count = heights.len(); - let hex_headers = - heights.filter_map(|height| chain.get_block_header(height).map(serialize_hex)); + let hex_headers = headers + .iter_headers() + .skip(start_height) + .take(count) + .map(|h| serialize_hex(h.header())); Ok(json!({"count": count, "hex": String::from_iter(hex_headers), "max": max_count})) } @@ -377,8 +367,7 @@ impl Rpc { fn new_status(&self, scripthash: ScriptHash) -> Result { let mut status = ScriptHashStatus::new(scripthash); - self.tracker - .update_scripthash_status(&mut status, &self.daemon, &self.cache)?; + self.tracker.update_scripthash_status(&mut status)?; Ok(status) } @@ -422,35 +411,26 @@ impl Rpc { if verbose { let blockhash = self .tracker - .lookup_transaction(&self.daemon, txid)? + .lookup_transaction(txid)? .map(|(blockhash, _tx)| blockhash); - return self.daemon.get_transaction_info(&txid, blockhash); + return self.daemon.get_transaction(&txid, blockhash, verbose); } - // if the scripthash was subscribed, tx should be cached - if let Some(tx_hex) = self - .cache - .get_tx(&txid, |tx_bytes| tx_bytes.to_lower_hex_string()) - { - return Ok(json!(tx_hex)); - } - debug!("tx cache miss: txid={}", txid); // use internal index to load confirmed transaction if let Some(tx_hex) = self .tracker - .lookup_transaction(&self.daemon, txid)? + .lookup_transaction(txid)? .map(|(_blockhash, tx)| tx.to_lower_hex_string()) { return Ok(json!(tx_hex)); } // load unconfirmed transaction via RPC - Ok(json!(self.daemon.get_transaction_hex(&txid, None)?)) + Ok(json!(self.daemon.get_transaction(&txid, None, verbose)?)) } fn transaction_get_merkle(&self, (txid, height): &(Txid, usize)) -> Result { - let chain = self.tracker.chain(); - let blockhash = match chain.get_block_hash(*height) { + let blockhash = match self.tracker.headers().iter_headers().nth(*height) { None => bail!("missing block at {}", height), - Some(blockhash) => blockhash, + Some(header) => header.hash(), }; let txids = self.daemon.get_block_txids(blockhash)?; match txids.iter().position(|current_txid| *current_txid == *txid) { @@ -470,10 +450,9 @@ impl Rpc { &self, (height, tx_pos, merkle): (usize, usize, bool), ) -> Result { - let chain = self.tracker.chain(); - let blockhash = match chain.get_block_hash(height) { + let blockhash = match self.tracker.headers().iter_headers().nth(height) { None => bail!("missing block at {}", height), - Some(blockhash) => blockhash, + Some(header) => header.hash(), }; let txids = self.daemon.get_block_txids(blockhash)?; if tx_pos >= txids.len() { @@ -507,7 +486,7 @@ impl Rpc { fn features(&self) -> Result { Ok(json!({ - "genesis_hash": self.tracker.chain().get_block_hash(0), + "genesis_hash": self.tracker.headers().genesis().unwrap().hash(), "hosts": { self.addr.ip().to_string(): { "tcp_port": self.addr.port() @@ -718,7 +697,7 @@ impl Call { Err(err) => { warn!("RPC {} failed: {:#}", self.method, err); match err - .downcast_ref::() + .downcast_ref::() .and_then(extract_bitcoind_error) { Some(e) => error_msg(&self.id, RpcError::DaemonError(e.clone())), diff --git a/src/index.rs b/src/index.rs deleted file mode 100644 index 5c30d9c..0000000 --- a/src/index.rs +++ /dev/null @@ -1,323 +0,0 @@ -use anyhow::{Context, Result}; -use bitcoin::consensus::{deserialize, Decodable, Encodable}; -use bitcoin::hashes::Hash; -use bitcoin::{BlockHash, OutPoint, Txid}; -use bitcoin_slices::{bsl, Visit, Visitor}; -use std::ops::ControlFlow; -use std::thread; - -use crate::{ - chain::{Chain, NewHeader}, - daemon::Daemon, - db::{DBStore, WriteBatch}, - metrics::{self, Gauge, Histogram, Metrics}, - signals::ExitFlag, - types::{ - bsl_txid, HashPrefixRow, HeaderRow, ScriptHash, ScriptHashRow, SerBlock, SpendingPrefixRow, - TxidRow, - }, -}; - -#[derive(Clone)] -struct Stats { - update_duration: Histogram, - update_size: Histogram, - height: Gauge, - db_properties: Gauge, -} - -impl Stats { - fn new(metrics: &Metrics) -> Self { - Self { - update_duration: metrics.histogram_vec( - "index_update_duration", - "Index update duration (in seconds)", - "step", - metrics::default_duration_buckets(), - ), - update_size: metrics.histogram_vec( - "index_update_size", - "Index update size (in bytes)", - "step", - metrics::default_size_buckets(), - ), - height: metrics.gauge("index_height", "Indexed block height", "type"), - db_properties: metrics.gauge("index_db_properties", "Index DB properties", "name"), - } - } - - fn observe_duration(&self, label: &str, f: impl FnOnce() -> T) -> T { - self.update_duration.observe_duration(label, f) - } - - fn observe_size(&self, label: &str, rows: &[[u8; N]]) { - self.update_size.observe(label, (rows.len() * N) as f64); - } - - fn observe_batch(&self, batch: &WriteBatch) { - self.observe_size("write_funding_rows", &batch.funding_rows); - self.observe_size("write_spending_rows", &batch.spending_rows); - self.observe_size("write_txid_rows", &batch.txid_rows); - self.observe_size("write_header_rows", &batch.header_rows); - debug!( - "writing {} funding and {} spending rows from {} transactions, {} blocks", - batch.funding_rows.len(), - batch.spending_rows.len(), - batch.txid_rows.len(), - batch.header_rows.len() - ); - } - - fn observe_chain(&self, chain: &Chain) { - self.height.set("tip", chain.height() as f64); - } - - fn observe_db(&self, store: &DBStore) { - for (cf, name, value) in store.get_properties() { - self.db_properties - .set(&format!("{}:{}", name, cf), value as f64); - } - } -} - -/// Confirmed transactions' address index -pub struct Index { - store: DBStore, - batch_size: usize, - lookup_limit: Option, - chain: Chain, - stats: Stats, - is_ready: bool, - flush_needed: bool, -} - -impl Index { - pub(crate) fn load( - store: DBStore, - mut chain: Chain, - metrics: &Metrics, - batch_size: usize, - lookup_limit: Option, - reindex_last_blocks: usize, - ) -> Result { - if let Some(row) = store.get_tip() { - let tip = deserialize(&row).expect("invalid tip"); - let headers = store - .iter_headers() - .map(|row| HeaderRow::from_db_row(row).header); - chain.load(headers, tip); - chain.drop_last_headers(reindex_last_blocks); - }; - let stats = Stats::new(metrics); - stats.observe_chain(&chain); - stats.observe_db(&store); - Ok(Index { - store, - batch_size, - lookup_limit, - chain, - stats, - is_ready: false, - flush_needed: false, - }) - } - - pub(crate) fn chain(&self) -> &Chain { - &self.chain - } - - pub(crate) fn limit_result(&self, entries: impl Iterator) -> Result> { - let mut entries = entries.fuse(); - let result: Vec = match self.lookup_limit { - Some(lookup_limit) => entries.by_ref().take(lookup_limit).collect(), - None => entries.by_ref().collect(), - }; - if entries.next().is_some() { - bail!(">{} index entries, query may take too long", result.len()) - } - Ok(result) - } - - pub(crate) fn filter_by_txid(&self, txid: Txid) -> impl Iterator + '_ { - self.store - .iter_txid(TxidRow::scan_prefix(txid)) - .map(|row| HashPrefixRow::from_db_row(row).height()) - .filter_map(move |height| self.chain.get_block_hash(height)) - } - - pub(crate) fn filter_by_funding( - &self, - scripthash: ScriptHash, - ) -> impl Iterator + '_ { - self.store - .iter_funding(ScriptHashRow::scan_prefix(scripthash)) - .map(|row| HashPrefixRow::from_db_row(row).height()) - .filter_map(move |height| self.chain.get_block_hash(height)) - } - - pub(crate) fn filter_by_spending( - &self, - outpoint: OutPoint, - ) -> impl Iterator + '_ { - self.store - .iter_spending(SpendingPrefixRow::scan_prefix(outpoint)) - .map(|row| HashPrefixRow::from_db_row(row).height()) - .filter_map(move |height| self.chain.get_block_hash(height)) - } - - // Return `Ok(true)` when the chain is fully synced and the index is compacted. - pub(crate) fn sync(&mut self, daemon: &Daemon, exit_flag: &ExitFlag) -> Result { - let new_headers = self - .stats - .observe_duration("headers", || daemon.get_new_headers(&self.chain))?; - match (new_headers.first(), new_headers.last()) { - (Some(first), Some(last)) => { - let count = new_headers.len(); - info!( - "indexing {} blocks: [{}..{}]", - count, - first.height(), - last.height() - ); - } - _ => { - if self.flush_needed { - self.store.flush(); // full compaction is performed on the first flush call - self.flush_needed = false; - } - self.is_ready = true; - return Ok(true); // no more blocks to index (done for now) - } - } - - thread::scope(|scope| -> Result<()> { - let (tx, rx) = crossbeam_channel::bounded(1); - - let chunks = new_headers.chunks(self.batch_size); - let index = &self; // to be moved into reader thread - let reader = thread::Builder::new() - .name("index_build".into()) - .spawn_scoped(scope, move || -> Result<()> { - for chunk in chunks { - exit_flag.poll().with_context(|| { - format!( - "indexing interrupted at height: {}", - chunk.first().unwrap().height() - ) - })?; - let batch = index.index_blocks(daemon, chunk)?; - tx.send(batch).context("writer disconnected")?; - } - Ok(()) // `tx` is dropped, to stop the iteration on `rx` - }) - .expect("spawn failed"); - - let index = &self; // to be moved into writer thread - let writer = thread::Builder::new() - .name("index_write".into()) - .spawn_scoped(scope, move || { - let stats = &index.stats; - for mut batch in rx { - stats.observe_duration("sort", || batch.sort()); // pre-sort to optimize DB writes - stats.observe_batch(&batch); - stats.observe_duration("write", || index.store.write(&batch)); - stats.observe_db(&index.store); - } - }) - .expect("spawn failed"); - - reader.join().expect("reader thread panic")?; - writer.join().expect("writer thread panic"); - Ok(()) - })?; - self.chain.update(new_headers); - self.stats.observe_chain(&self.chain); - self.flush_needed = true; - Ok(false) // sync is not done - } - - fn index_blocks(&self, daemon: &Daemon, chunk: &[NewHeader]) -> Result { - let blockhashes: Vec = chunk.iter().map(|h| h.hash()).collect(); - let mut heights = chunk.iter().map(|h| h.height()); - - let mut batch = WriteBatch::default(); - - daemon.for_blocks(blockhashes, |blockhash, block| { - let height = heights.next().expect("unexpected block"); - self.stats.observe_duration("block", || { - index_single_block(blockhash, block, height, &mut batch); - }); - self.stats.height.set("tip", height as f64); - })?; - let heights: Vec<_> = heights.collect(); - assert!( - heights.is_empty(), - "some blocks were not indexed: {:?}", - heights - ); - Ok(batch) - } - - pub(crate) fn is_ready(&self) -> bool { - self.is_ready - } -} - -fn index_single_block( - block_hash: BlockHash, - block: SerBlock, - height: usize, - batch: &mut WriteBatch, -) { - struct IndexBlockVisitor<'a> { - batch: &'a mut WriteBatch, - height: usize, - } - - impl Visitor for IndexBlockVisitor<'_> { - fn visit_transaction(&mut self, tx: &bsl::Transaction) -> ControlFlow<()> { - let txid = bsl_txid(tx); - self.batch - .txid_rows - .push(TxidRow::row(txid, self.height).to_db_row()); - ControlFlow::Continue(()) - } - - fn visit_tx_out(&mut self, _vout: usize, tx_out: &bsl::TxOut) -> ControlFlow<()> { - let script = bitcoin::Script::from_bytes(tx_out.script_pubkey()); - // skip indexing unspendable outputs - if !script.is_op_return() { - let row = ScriptHashRow::row(ScriptHash::new(script), self.height); - self.batch.funding_rows.push(row.to_db_row()); - } - ControlFlow::Continue(()) - } - - fn visit_tx_in(&mut self, _vin: usize, tx_in: &bsl::TxIn) -> ControlFlow<()> { - let prevout: OutPoint = tx_in.prevout().into(); - // skip indexing coinbase transactions' input - if !prevout.is_null() { - let row = SpendingPrefixRow::row(prevout, self.height); - self.batch.spending_rows.push(row.to_db_row()); - } - ControlFlow::Continue(()) - } - - fn visit_block_header(&mut self, header: &bsl::BlockHeader) -> ControlFlow<()> { - let header = bitcoin::block::Header::consensus_decode(&mut header.as_ref()) - .expect("block header was already validated"); - self.batch - .header_rows - .push(HeaderRow::new(header).to_db_row()); - ControlFlow::Continue(()) - } - } - - let mut index_block = IndexBlockVisitor { batch, height }; - bsl::Block::visit(&block, &mut index_block).expect("core returned invalid block"); - - let len = block_hash - .consensus_encode(&mut (&mut batch.tip_row as &mut [u8])) - .expect("in-memory writers don't error"); - debug_assert_eq!(len, BlockHash::LEN); -} diff --git a/src/lib.rs b/src/lib.rs index 1458f06..9b8fad8 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -7,17 +7,12 @@ extern crate log; #[macro_use] extern crate serde_derive; -mod cache; -mod chain; mod config; mod daemon; -mod db; mod electrum; -mod index; mod mempool; mod merkle; mod metrics; -mod p2p; mod server; mod signals; mod status; @@ -26,3 +21,6 @@ mod tracker; mod types; pub use server::run; + +use bindex::bitcoin; +use bindex::bitcoin_slices; diff --git a/src/mempool.rs b/src/mempool.rs index 04457c9..f797a14 100644 --- a/src/mempool.rs +++ b/src/mempool.rs @@ -5,15 +5,15 @@ use std::convert::TryFrom; use std::iter::FromIterator; use std::ops::Bound; -use bitcoin::hashes::Hash; -use bitcoin::{Amount, OutPoint, Transaction, Txid}; +use crate::bitcoin::hashes::Hash; +use crate::bitcoin::{Amount, OutPoint, Transaction, Txid}; +use bindex::ScriptHash; use serde::ser::{Serialize, SerializeSeq, Serializer}; use crate::{ daemon::Daemon, metrics::{Gauge, Metrics}, signals::ExitFlag, - types::ScriptHash, }; pub(crate) struct Entry { @@ -368,7 +368,7 @@ impl Serialize for FeeHistogram { #[cfg(test)] mod tests { use super::FeeHistogram; - use bitcoin::Amount; + use crate::bitcoin::Amount; use serde_json::json; #[test] diff --git a/src/merkle.rs b/src/merkle.rs index 8294280..e34e8eb 100644 --- a/src/merkle.rs +++ b/src/merkle.rs @@ -1,4 +1,4 @@ -use bitcoin::{hash_types::TxMerkleNode, hashes::Hash, Txid}; +use crate::bitcoin::{hash_types::TxMerkleNode, hashes::Hash, Txid}; pub(crate) struct Proof { proof: Vec, @@ -54,7 +54,7 @@ impl Proof { #[cfg(test)] mod tests { - use bitcoin::{consensus::encode::deserialize, Block, Txid}; + use crate::bitcoin::{consensus::encode::deserialize, Block, Txid}; use std::path::Path; use super::Proof; diff --git a/src/p2p.rs b/src/p2p.rs deleted file mode 100644 index 0787f36..0000000 --- a/src/p2p.rs +++ /dev/null @@ -1,406 +0,0 @@ -use anyhow::{Context, Result}; -use bitcoin::blockdata::block::Header as BlockHeader; -use bitcoin::consensus::Encodable; -use bitcoin::{ - consensus::{ - encode::{self, ReadExt, VarInt}, - Decodable, - }, - hashes::Hash, - io, - p2p::{ - self, address, - message::{self, CommandString, NetworkMessage}, - message_blockdata::{GetHeadersMessage, Inventory}, - message_network, Magic, - }, - secp256k1::{self, rand::Rng}, - Block, BlockHash, -}; -use bitcoin_slices::{bsl, Parse}; -use crossbeam_channel::{bounded, select, Receiver, Sender}; - -use std::io::Write; -use std::net::{IpAddr, Ipv4Addr, SocketAddr, TcpStream}; -use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; - -use crate::types::SerBlock; -use crate::{ - chain::{Chain, NewHeader}, - config::ELECTRS_VERSION, - metrics::{default_duration_buckets, default_size_buckets, Histogram, Metrics}, -}; - -enum Request { - GetNewHeaders(GetHeadersMessage), - GetBlocks(Vec), -} - -impl Request { - fn get_new_headers(chain: &Chain) -> Request { - Request::GetNewHeaders(GetHeadersMessage::new( - chain.locator(), - BlockHash::all_zeros(), - )) - } - - fn get_blocks(blockhashes: &[BlockHash]) -> Request { - Request::GetBlocks( - blockhashes - .iter() - .map(|blockhash| Inventory::WitnessBlock(*blockhash)) - .collect(), - ) - } -} - -pub(crate) struct Connection { - req_send: Sender, - blocks_recv: Receiver, - headers_recv: Receiver>, - new_block_recv: Receiver<()>, - - blocks_duration: Histogram, -} - -impl Connection { - /// Get new block headers (supporting reorgs). - /// https://en.bitcoin.it/wiki/Protocol_documentation#getheaders - /// Defined as `&mut self` to prevent concurrent invocations (https://github.com/romanz/electrs/pull/526#issuecomment-934685515). - pub(crate) fn get_new_headers(&mut self, chain: &Chain) -> Result> { - self.req_send.send(Request::get_new_headers(chain))?; - let headers = self - .headers_recv - .recv() - .context("failed to get new headers")?; - - debug!("got {} new headers", headers.len()); - let prev_blockhash = match headers.first() { - None => return Ok(vec![]), - Some(first) => first.prev_blockhash, - }; - let new_heights = match chain.get_block_height(&prev_blockhash) { - Some(last_height) => (last_height + 1).., - None => bail!("missing prev_blockhash: {}", prev_blockhash), - }; - Ok(headers - .into_iter() - .zip(new_heights) - .map(NewHeader::from) - .collect()) - } - - /// Request and process the specified blocks (in the specified order). - /// See https://en.bitcoin.it/wiki/Protocol_documentation#getblocks for details. - /// Defined as `&mut self` to prevent concurrent invocations (https://github.com/romanz/electrs/pull/526#issuecomment-934685515). - pub(crate) fn for_blocks(&mut self, blockhashes: B, mut func: F) -> Result<()> - where - B: IntoIterator, - F: FnMut(BlockHash, SerBlock), - { - self.blocks_duration.observe_duration("total", || { - let blockhashes: Vec = blockhashes.into_iter().collect(); - if blockhashes.is_empty() { - return Ok(()); - } - self.blocks_duration.observe_duration("request", || { - debug!("loading {} blocks", blockhashes.len()); - self.req_send.send(Request::get_blocks(&blockhashes)) - })?; - - for hash in blockhashes { - let block = self.blocks_duration.observe_duration("response", || { - let block = self - .blocks_recv - .recv() - .with_context(|| format!("failed to get block {}", hash))?; - let header = bsl::BlockHeader::parse(&block[..]) - .expect("core returned invalid blockheader") - .parsed_owned(); - ensure!( - &header.block_hash_sha2()[..] == hash.as_byte_array(), - "got unexpected block" - ); - Ok(block) - })?; - self.blocks_duration - .observe_duration("process", || func(hash, block)); - } - Ok(()) - }) - } - - /// Note: only a single receiver will get the notification (https://github.com/romanz/electrs/pull/526#issuecomment-934687415). - pub(crate) fn new_block_notification(&self) -> Receiver<()> { - self.new_block_recv.clone() - } - - pub(crate) fn connect(address: SocketAddr, metrics: &Metrics, magic: Magic) -> Result { - let recv_conn = TcpStream::connect(address) - .with_context(|| format!("p2p failed to connect: {:?}", address))?; - let mut send_conn = recv_conn - .try_clone() - .context("failed to clone connection")?; - - let (tx_send, tx_recv) = bounded::(1); - let (rx_send, rx_recv) = bounded::(1); - - let send_duration = metrics.histogram_vec( - "p2p_send_duration", - "Time spent sending p2p messages (in seconds)", - "step", - default_duration_buckets(), - ); - let recv_duration = metrics.histogram_vec( - "p2p_recv_duration", - "Time spent receiving p2p messages (in seconds)", - "step", - default_duration_buckets(), - ); - let parse_duration = metrics.histogram_vec( - "p2p_parse_duration", - "Time spent parsing p2p messages (in seconds)", - "step", - default_duration_buckets(), - ); - let recv_size = metrics.histogram_vec( - "p2p_recv_size", - "Size of p2p messages read (in bytes)", - "message", - default_size_buckets(), - ); - let blocks_duration = metrics.histogram_vec( - "p2p_blocks_duration", - "Time spent getting blocks via p2p protocol (in seconds)", - "step", - default_duration_buckets(), - ); - - let mut buffer = vec![]; - crate::thread::spawn("p2p_send", move || loop { - use std::net::Shutdown; - let msg = match send_duration.observe_duration("wait", || tx_recv.recv()) { - Ok(msg) => msg, - Err(_) => { - // p2p_loop is closed, so tx_send is disconnected - debug!("closing p2p_send thread: no more messages to send"); - // close the stream reader (p2p_recv thread may block on it) - if let Err(e) = send_conn.shutdown(Shutdown::Read) { - warn!("failed to shutdown p2p connection: {}", e) - } - return Ok(()); - } - }; - send_duration.observe_duration("send", || { - trace!("send: {:?}", msg); - let raw_msg = message::RawNetworkMessage::new(magic, msg); - buffer.clear(); - raw_msg - .consensus_encode(&mut buffer) - .expect("in-memory writers don't error"); - send_conn - .write_all(buffer.as_slice()) - .context("p2p failed to send") - })?; - }); - - let mut stream_reader = std::io::BufReader::new(recv_conn); - crate::thread::spawn("p2p_recv", move || loop { - let start = Instant::now(); - let raw_msg = RawNetworkMessage::consensus_decode(&mut stream_reader); - { - let duration = duration_to_seconds(start.elapsed()); - let label = format!( - "recv_{}", - raw_msg - .as_ref() - .map(|msg| msg.cmd.as_ref()) - .unwrap_or("err") - ); - recv_duration.observe(&label, duration); - } - let raw_msg = match raw_msg { - Ok(raw_msg) => { - recv_size.observe(raw_msg.cmd.as_ref(), raw_msg.raw.len() as f64); - if raw_msg.magic != magic { - bail!("unexpected magic {} (instead of {})", raw_msg.magic, magic) - } - raw_msg - } - Err(encode::Error::Io(e)) if e.kind() == io::ErrorKind::UnexpectedEof => { - debug!("closing p2p_recv thread: connection closed"); - return Ok(()); - } - Err(e) => bail!("failed to recv a message from peer: {}", e), - }; - - recv_duration.observe_duration("wait", || rx_send.send(raw_msg))?; - }); - - let (req_send, req_recv) = bounded::(1); - let (blocks_send, blocks_recv) = bounded::(10); - let (headers_send, headers_recv) = bounded::>(1); - let (new_block_send, new_block_recv) = bounded::<()>(0); - let (init_send, init_recv) = bounded::<()>(0); - - tx_send.send(build_version_message())?; - - crate::thread::spawn("p2p_loop", move || loop { - select! { - recv(rx_recv) -> result => { - let raw_msg = match result { - Ok(raw_msg) => raw_msg, - Err(_) => { // p2p_recv is closed, so rx_send is disconnected - debug!("closing p2p_loop thread: peer has disconnected"); - return Ok(()); // new_block_send is dropped, causing the server to exit - } - }; - - let label = format!("parse_{}", raw_msg.cmd.as_ref()); - let msg = match parse_duration.observe_duration(&label, || raw_msg.parse()) { - Ok(msg) => msg, - Err(err) => bail!("failed to parse {err}"), - }; - trace!("recv: {:?}", msg); - - match msg { - ParsedNetworkMessage::Version(version) => { - debug!("peer version: {:?}", version); - tx_send.send(NetworkMessage::Verack)?; - } - ParsedNetworkMessage::Inv(inventory) => { - debug!("peer inventory: {:?}", inventory); - if inventory.iter().any(|inv| matches!(inv, Inventory::Block(_))) { - let _ = new_block_send.try_send(()); // best-effort notification - } - - }, - ParsedNetworkMessage::Ping(nonce) => { - tx_send.send(NetworkMessage::Pong(nonce))?; // connection keep-alive - } - ParsedNetworkMessage::Verack => { - init_send.send(())?; // peer acknowledged our version - } - ParsedNetworkMessage::Block(block) => blocks_send.send(block)?, - ParsedNetworkMessage::Headers(headers) => headers_send.send(headers)?, - ParsedNetworkMessage::Ignored => (), - } - } - recv(req_recv) -> result => { - let req = match result { - Ok(req) => req, - Err(_) => { // self is dropped, so req_send is disconnected - debug!("closing p2p_loop thread: no more requests to handle"); - return Ok(()); - } - }; - let msg = match req { - Request::GetNewHeaders(msg) => NetworkMessage::GetHeaders(msg), - Request::GetBlocks(inv) => NetworkMessage::GetData(inv), - }; - tx_send.send(msg)?; - } - } - }); - - init_recv.recv()?; // wait until `verack` is received - - Ok(Connection { - req_send, - blocks_recv, - headers_recv, - new_block_recv, - blocks_duration, - }) - } -} - -fn build_version_message() -> NetworkMessage { - let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), 0); - let timestamp = SystemTime::now() - .duration_since(UNIX_EPOCH) - .expect("Time error") - .as_secs() as i64; - - let services = p2p::ServiceFlags::NONE; - - NetworkMessage::Version(message_network::VersionMessage { - version: p2p::PROTOCOL_VERSION, - services, - timestamp, - receiver: address::Address::new(&addr, services), - sender: address::Address::new(&addr, services), - nonce: secp256k1::rand::thread_rng().gen(), - user_agent: format!("/electrs:{}/", ELECTRS_VERSION), - start_height: 0, - relay: false, - }) -} - -struct RawNetworkMessage { - magic: Magic, - cmd: CommandString, - raw: Vec, -} - -impl RawNetworkMessage { - fn parse(self) -> Result { - let mut raw: &[u8] = &self.raw; - let payload = match self.cmd.as_ref() { - "version" => ParsedNetworkMessage::Version(Decodable::consensus_decode(&mut raw)?), - "verack" => ParsedNetworkMessage::Verack, - "inv" => ParsedNetworkMessage::Inv(Decodable::consensus_decode(&mut raw)?), - "block" => ParsedNetworkMessage::Block(self.raw), - "headers" => { - let len = VarInt::consensus_decode(&mut raw)?.0; - let mut headers = Vec::with_capacity(len as usize); - for _ in 0..len { - headers.push(Block::consensus_decode(&mut raw)?.header); - } - ParsedNetworkMessage::Headers(headers) - } - "ping" => ParsedNetworkMessage::Ping(Decodable::consensus_decode(&mut raw)?), - "pong" => ParsedNetworkMessage::Ignored, // unused - "addr" => ParsedNetworkMessage::Ignored, // unused - "alert" => ParsedNetworkMessage::Ignored, // https://bitcoin.org/en/alert/2016-11-01-alert-retirement - _ => bail!( - "unsupported message: command={}, payload={:?}", - self.cmd, - self.raw - ), - }; - Ok(payload) - } -} - -#[derive(Debug)] -enum ParsedNetworkMessage { - Version(message_network::VersionMessage), - Verack, - Inv(Vec), - Ping(u64), - Headers(Vec), - Block(SerBlock), - Ignored, -} - -impl Decodable for RawNetworkMessage { - fn consensus_decode(d: &mut D) -> Result { - let magic = Decodable::consensus_decode(d)?; - let cmd = Decodable::consensus_decode(d)?; - - let len = u32::consensus_decode(d)?; - let _checksum = <[u8; 4]>::consensus_decode(d)?; // assume data is correct - let mut raw = vec![0u8; len as usize]; - d.read_slice(&mut raw)?; - - Ok(RawNetworkMessage { magic, cmd, raw }) - } -} - -/// `duration_to_seconds` converts Duration to seconds. -#[inline] -pub fn duration_to_seconds(d: Duration) -> f64 { - let nanos = f64::from(d.subsec_nanos()) / 1e9; - d.as_secs() as f64 + nanos -} diff --git a/src/server.rs b/src/server.rs index f50662f..221fb5a 100644 --- a/src/server.rs +++ b/src/server.rs @@ -85,7 +85,6 @@ fn serve() -> Result<()> { ); let mut rpc = Rpc::new(&config, metrics)?; - let new_block_rx = rpc.new_block_notification(); let mut peers = HashMap::::new(); loop { // initial sync and compaction may take a few hours @@ -107,14 +106,6 @@ fn serve() -> Result<()> { result.context("signal channel disconnected")?; rpc.signal().exit_flag().poll().context("RPC server interrupted")?; }, - // Handle new blocks' notifications - recv(new_block_rx) -> result => match result { - Ok(_) => (), // sync and update - Err(_) => { - info!("disconnected from bitcoind"); - return Ok(()); - } - }, // Handle Electrum RPC requests recv(server_rx) -> event => { let first = once(event.context("server disconnected")?); diff --git a/src/status.rs b/src/status.rs index 29b1651..fe16acf 100644 --- a/src/status.rs +++ b/src/status.rs @@ -1,27 +1,16 @@ -use anyhow::Result; -use bitcoin::{ - consensus::serialize, +use crate::bitcoin::{ + consensus::Decodable, hashes::{sha256, Hash, HashEngine}, Amount, BlockHash, OutPoint, SignedAmount, Transaction, Txid, }; -use bitcoin_slices::{bsl, Visit, Visitor}; -use rayon::prelude::*; +use anyhow::Result; +use bindex::{IndexedChain, ScriptHash}; use serde::ser::{Serialize, Serializer}; +use std::collections::{BTreeMap, HashMap, HashSet}; use std::convert::TryFrom; -use std::{ - collections::{BTreeMap, HashMap, HashSet}, - ops::ControlFlow, -}; -use crate::{ - cache::Cache, - chain::Chain, - daemon::Daemon, - index::Index, - mempool::Mempool, - types::{bsl_txid, ScriptHash, SerBlock, StatusHash}, -}; +use crate::{mempool::Mempool, types::StatusHash}; /// Given a scripthash, store relevant inputs and outputs of a specific transaction struct TxEntry { @@ -98,7 +87,7 @@ pub(crate) struct HistoryEntry { height: Height, #[serde( skip_serializing_if = "Option::is_none", - with = "bitcoin::amount::serde::as_sat::opt" + with = "crate::bitcoin::amount::serde::as_sat::opt" )] fee: Option, } @@ -130,11 +119,20 @@ impl HistoryEntry { } } +/// Make sure blocks are sorted by ascending height. +#[derive(Debug, PartialEq, Eq, PartialOrd, Ord)] +struct BlockKey(usize, BlockHash); + +impl From> for BlockKey { + fn from(location: bindex::Location<'_>) -> Self { + Self(location.block_height(), location.block_hash()) + } +} + /// ScriptHash subscription status pub struct ScriptHashStatus { scripthash: ScriptHash, // specific scripthash to be queried - tip: BlockHash, // used for skipping confirmed entries' sync - confirmed: HashMap>, // confirmed entries, partitioned per block (may contain stale blocks) + confirmed: BTreeMap>, // confirmed entries (chronologically ordered), partitioned per block mempool: Vec, // unconfirmed entries history: Vec, // computed from confirmed and mempool entries statushash: Option, // computed from history @@ -144,9 +142,9 @@ pub struct ScriptHashStatus { /// https://electrum-protocol.readthedocs.io/en/latest/protocol-methods.html#blockchain-scripthash-get-balance #[derive(Default, Eq, PartialEq, Serialize)] pub(crate) struct Balance { - #[serde(with = "bitcoin::amount::serde::as_sat", rename = "confirmed")] + #[serde(with = "crate::bitcoin::amount::serde::as_sat", rename = "confirmed")] confirmed_balance: Amount, - #[serde(with = "bitcoin::amount::serde::as_sat", rename = "unconfirmed")] + #[serde(with = "crate::bitcoin::amount::serde::as_sat", rename = "unconfirmed")] mempool_delta: SignedAmount, } @@ -157,7 +155,7 @@ pub(crate) struct UnspentEntry { height: usize, // 0 = mempool entry tx_hash: Txid, tx_pos: u32, - #[serde(with = "bitcoin::amount::serde::as_sat")] + #[serde(with = "crate::bitcoin::amount::serde::as_sat")] value: Amount, } @@ -169,25 +167,23 @@ struct Unspent { } impl Unspent { - fn build(status: &ScriptHashStatus, chain: &Chain) -> Self { + fn build(status: &ScriptHashStatus) -> Self { let mut unspent = Unspent::default(); // First, add all relevant entries' funding outputs to the outpoints' map status - .confirmed_height_entries(chain) + .confirmed_height_entries() .for_each(|(height, entries)| entries.iter().for_each(|e| unspent.insert(e, height))); // Then, remove spent outpoints from the map - status - .confirmed_entries(chain) - .for_each(|e| unspent.remove(e)); + status.confirmed_entries().for_each(|e| unspent.remove(e)); unspent.balance.confirmed_balance = unspent.balance(); // Now, do the same over the mempool (first add funding outputs, and then remove spent ones) status.mempool.iter().for_each(|e| unspent.insert(e, 0)); // mempool height = 0 status.mempool.iter().for_each(|e| unspent.remove(e)); - unspent.balance.mempool_delta = unspent.balance().to_signed().unwrap() - - unspent.balance.confirmed_balance.to_signed().unwrap(); - + let total_balance = unspent.balance().to_signed().unwrap(); + let confirmed_balance = unspent.balance.confirmed_balance.to_signed().unwrap(); + unspent.balance.mempool_delta = total_balance - confirmed_balance; unspent } @@ -232,8 +228,7 @@ impl ScriptHashStatus { pub fn new(scripthash: ScriptHash) -> Self { Self { scripthash, - tip: BlockHash::all_zeros(), - confirmed: HashMap::new(), + confirmed: BTreeMap::new(), mempool: Vec::new(), history: Vec::new(), statushash: None, @@ -242,41 +237,33 @@ impl ScriptHashStatus { /// Iterate through confirmed TxEntries with their corresponding block heights. /// Skip entries from stale blocks. - fn confirmed_height_entries<'a>( - &'a self, - chain: &'a Chain, - ) -> impl Iterator + 'a { + fn confirmed_height_entries<'a>(&'a self) -> impl Iterator + 'a { self.confirmed .iter() - .filter_map(move |(blockhash, entries)| { - chain - .get_block_height(blockhash) - .map(|height| (height, &entries[..])) - }) + .map(move |(blockid, entries)| (blockid.0, &entries[..])) } /// Iterate through confirmed TxEntries. - /// Skip entries from stale blocks. - fn confirmed_entries<'a>(&'a self, chain: &'a Chain) -> impl Iterator + 'a { - self.confirmed_height_entries(chain) + fn confirmed_entries<'a>(&'a self) -> impl Iterator + 'a { + self.confirmed_height_entries() .flat_map(|(_height, entries)| entries) } /// Collect all funded and confirmed outpoints (as a set). - fn confirmed_outpoints(&self, chain: &Chain) -> HashSet { - self.confirmed_entries(chain) + fn confirmed_outpoints(&self) -> HashSet { + self.confirmed_entries() .flat_map(TxEntry::funding_outpoints) .collect() } /// Collect unspent transaction entries - pub(crate) fn get_unspent(&self, chain: &Chain) -> Vec { - Unspent::build(self, chain).into_entries() + pub(crate) fn get_unspent(&self) -> Vec { + Unspent::build(self).into_entries() } /// Collect unspent transaction balance - pub(crate) fn get_balance(&self, chain: &Chain) -> Balance { - Unspent::build(self, chain).balance + pub(crate) fn get_balance(&self) -> Balance { + Unspent::build(self).balance } /// Collect transaction history entries @@ -285,8 +272,8 @@ impl ScriptHashStatus { } /// Collect all confirmed history entries (in block order). - fn get_confirmed_history(&self, chain: &Chain) -> Vec { - self.confirmed_height_entries(chain) + fn get_confirmed_history(&self) -> Vec { + self.confirmed_height_entries() .collect::>() .into_iter() .flat_map(|(height, entries)| { @@ -311,88 +298,59 @@ impl ScriptHashStatus { .collect() } - /// Apply `func` only on the new blocks (to be fetched via p2p interface). - fn for_new_blocks(&self, blockhashes: B, daemon: &Daemon, func: F) -> Result<()> - where - B: IntoIterator, - F: FnMut(BlockHash, SerBlock), - { - daemon.for_blocks( - blockhashes - .into_iter() - .filter(|blockhash| !self.confirmed.contains_key(blockhash)), - func, - ) - } - /// Get funding and spending entries from new blocks. /// Also cache relevant transactions and their merkle proofs. - fn sync_confirmed( - &self, - index: &Index, - daemon: &Daemon, - cache: &Cache, - outpoints: &mut HashSet, - ) -> Result>> { - // Will be updated during the following block scans - let mut result = HashMap::>::new(); - - let funding_blockhashes = index.limit_result(index.filter_by_funding(self.scripthash))?; - self.for_new_blocks(funding_blockhashes, daemon, |blockhash, block| { - let block_entries = result.entry(blockhash).or_default(); // the block may already exist - - // extract relevant funding transactions - for filtered_outputs in filter_block_txs_outputs(block, self.scripthash) { - cache.add_tx(filtered_outputs.txid, move || filtered_outputs.tx_bytes); - // store funded outpoints (to check for spending later) - outpoints.extend(make_outpoints( - filtered_outputs.txid, - &filtered_outputs.result, - )); - block_entries - .entry(filtered_outputs.pos) // the transaction may already exist - .or_insert_with(|| TxEntry::new(filtered_outputs.txid)) - .outputs = filtered_outputs.result; + fn sync_confirmed(&mut self, index: &IndexedChain) -> Result> { + let headers = index.headers(); + let mut latest_header = None; + // Drop entries from stale blocks + while let Some(entry) = self.confirmed.last_entry() { + let BlockKey(height, hash) = entry.key(); + match headers.get_header(*hash, *height) { + Ok(header) => { + latest_header = Some(header); + break; + } + Err(err) => { + warn!("drop reorged block: {}", err); + entry.remove(); + continue; + } } - })?; - let spending_blockhashes: HashSet = outpoints - .par_iter() // use rayon for concurrent index lookups - .flat_map_iter(|outpoint| index.filter_by_spending(*outpoint)) - .collect(); - self.for_new_blocks(spending_blockhashes, daemon, |blockhash, block| { - let block_entries = result.entry(blockhash).or_default(); // the block may already exist + } + // Recompute all funded outpoints + let mut outpoints = self.confirmed_outpoints(); + // Process transactions in chronological order + for location in index.locations_by_scripthash(&self.scripthash, latest_header)? { + let tx_bytes = index.get_tx_bytes(&location)?; + let tx = Transaction::consensus_decode_from_finite_reader(&mut &tx_bytes[..])?; - // extract relevant spending transactions - for filtered_inputs in filter_block_txs_inputs(&block, outpoints) { - cache.add_tx(filtered_inputs.txid, move || filtered_inputs.tx_bytes); - block_entries - .entry(filtered_inputs.pos) // the transaction may already exist - .or_insert_with(|| TxEntry::new(filtered_inputs.txid)) - .spent = filtered_inputs.result; + // Check if this transaction has relevant inputs/outputs: + let spent = filter_inputs(&tx, &outpoints); + let outputs = filter_outputs(&tx, self.scripthash); + if spent.is_empty() && outputs.is_empty() { + continue; } - })?; - Ok(result - .into_iter() - .map(|(blockhash, entries_map)| { - let sorted_entries: Vec = entries_map - .into_iter() - .collect::>() // sort transactions by their position in a block - .into_values() // drop position within block - .collect(); - (blockhash, sorted_entries) - }) - .collect()) + // Build new TxEntry and add new outpoints (for next transactions) + let mut tx_entry = TxEntry::new(tx.compute_txid()); + tx_entry.spent = spent; + tx_entry.outputs = outputs; + outpoints.extend(tx_entry.funding_outpoints()); + + // Add new per-block entry (if needed) + self.confirmed + .entry(BlockKey::from(location)) + .or_default() + .push(tx_entry); + } + + Ok(outpoints) } /// Get funding and spending entries from current mempool. /// Also cache relevant transactions. - fn sync_mempool( - &self, - mempool: &Mempool, - cache: &Cache, - outpoints: &mut HashSet, - ) -> Vec { + fn sync_mempool(&self, mempool: &Mempool, mut outpoints: HashSet) -> Vec { let mut result = HashMap::::new(); // extract relevant funding transactions for entry in mempool.filter_by_funding(&self.scripthash) { @@ -404,40 +362,25 @@ impl ScriptHashStatus { .entry(entry.txid) // the transaction may already exist .or_insert_with(|| TxEntry::new(entry.txid)) .outputs = funding_outputs; - cache.add_tx(entry.txid, || serialize(&entry.tx).into_boxed_slice()); } for entry in outpoints .iter() .flat_map(|outpoint| mempool.filter_by_spending(outpoint)) { - let spent_outpoints = filter_inputs(&entry.tx, outpoints); + let spent_outpoints = filter_inputs(&entry.tx, &outpoints); assert!(!spent_outpoints.is_empty()); result .entry(entry.txid) // the transaction may already exist .or_insert_with(|| TxEntry::new(entry.txid)) .spent = spent_outpoints; - cache.add_tx(entry.txid, || serialize(&entry.tx).into_boxed_slice()); } result.into_values().collect() } - /// Sync with currently confirmed txs and mempool, downloading non-cached transactions via p2p protocol. + /// Sync with currently confirmed txs and mempool, downloading non-cached transactions via REST API. /// After a successful sync, scripthash status is updated. - pub(crate) fn sync( - &mut self, - index: &Index, - mempool: &Mempool, - daemon: &Daemon, - cache: &Cache, - ) -> Result<()> { - let mut outpoints: HashSet = self.confirmed_outpoints(index.chain()); - - let new_tip = index.chain().tip(); - if self.tip != new_tip { - let update = self.sync_confirmed(index, daemon, cache, &mut outpoints)?; - self.confirmed.extend(update); // add new blocks to the map - self.tip = new_tip; - } + pub(crate) fn sync(&mut self, index: &IndexedChain, mempool: &Mempool) -> Result<()> { + let outpoints = self.sync_confirmed(index)?; if !self.confirmed.is_empty() { debug!( "{} transactions from {} blocks", @@ -445,14 +388,13 @@ impl ScriptHashStatus { self.confirmed.len() ); } - self.mempool = self.sync_mempool(mempool, cache, &mut outpoints); + self.mempool = self.sync_mempool(mempool, outpoints); if !self.mempool.is_empty() { debug!("{} mempool transactions", self.mempool.len()); } // update history entries and status hash self.history.clear(); - self.history - .extend(self.get_confirmed_history(index.chain())); + self.history.extend(self.get_confirmed_history()); self.history.extend(self.get_mempool_history(mempool)); self.statushash = compute_status_hash(&self.history); @@ -471,32 +413,24 @@ fn make_outpoints(txid: Txid, outputs: &[TxOutput]) -> impl Iterator Vec { let outputs = tx.output.iter().zip(0u32..); outputs - .filter_map(move |(txo, vout)| { - if ScriptHash::new(&txo.script_pubkey) == scripthash { - Some(TxOutput { - index: vout, - value: txo.value, - }) - } else { - None - } + .filter(|&(txo, _vout)| ScriptHash::new(&txo.script_pubkey) == scripthash) + .map(|(txo, vout)| TxOutput { + index: vout, + value: txo.value, }) .collect() } +/// Collect inputs that spend one of the specified outpoints. fn filter_inputs(tx: &Transaction, outpoints: &HashSet) -> Vec { - tx.input - .iter() - .filter_map(|txi| { - if outpoints.contains(&txi.previous_output) { - Some(txi.previous_output) - } else { - None - } - }) + let inputs = tx.input.iter(); + inputs + .filter(|&txi| outpoints.contains(&txi.previous_output)) + .map(|txi| txi.previous_output) .collect() } @@ -512,114 +446,10 @@ fn compute_status_hash(history: &[HistoryEntry]) -> Option { Some(StatusHash::from_engine(engine)) } -struct FilteredTx { - tx_bytes: Box<[u8]>, - txid: Txid, - pos: usize, - result: Vec, -} - -fn filter_block_txs_outputs(block: SerBlock, scripthash: ScriptHash) -> Vec> { - struct FindOutputs { - scripthash: ScriptHash, - result: Vec>, - buffer: Vec, - pos: usize, - } - impl Visitor for FindOutputs { - // Called after all TxOuts are visited - fn visit_transaction(&mut self, tx: &bsl::Transaction) -> ControlFlow<()> { - if !self.buffer.is_empty() { - self.result.push(FilteredTx:: { - tx_bytes: tx.as_ref().into(), - txid: bsl_txid(tx), - pos: self.pos, - result: std::mem::take(&mut self.buffer), // clear buffer for next tx - }); - } - self.pos += 1; - ControlFlow::Continue(()) - } - // Keep only relevant outputs - fn visit_tx_out(&mut self, vout: usize, tx_out: &bsl::TxOut) -> ControlFlow<()> { - let current = ScriptHash::hash(tx_out.script_pubkey()); - if current == self.scripthash { - self.buffer.push(TxOutput { - index: vout as u32, - value: Amount::from_sat(tx_out.value()), - }) - } - ControlFlow::Continue(()) - } - } - let mut find_outputs = FindOutputs { - scripthash, - result: vec![], - buffer: vec![], - pos: 0, - }; - - bsl::Block::visit(&block, &mut find_outputs).expect("core returned invalid block"); - - find_outputs.result -} - -fn filter_block_txs_inputs( - block: &SerBlock, - outpoints: &HashSet, -) -> Vec> { - struct FindInputs<'a> { - outpoints: &'a HashSet, - result: Vec>, - buffer: Vec, - pos: usize, - } - - impl Visitor for FindInputs<'_> { - // Called after all TxIns are visited - fn visit_transaction(&mut self, tx: &bsl::Transaction) -> ControlFlow<()> { - if !self.buffer.is_empty() { - self.result.push(FilteredTx:: { - tx_bytes: tx.as_ref().into(), - txid: bsl_txid(tx), - pos: self.pos, - result: std::mem::take(&mut self.buffer), // clear buffer for next tx - }); - } - self.pos += 1; - ControlFlow::Continue(()) - } - // Keep only relevant outpoints - fn visit_tx_in(&mut self, _vin: usize, tx_in: &bsl::TxIn) -> ControlFlow<()> { - let current: OutPoint = tx_in.prevout().into(); - if self.outpoints.contains(¤t) { - self.buffer.push(current); - } - ControlFlow::Continue(()) - } - } - - let mut find_inputs = FindInputs { - outpoints, - result: vec![], - buffer: vec![], - pos: 0, - }; - - bsl::Block::visit(block, &mut find_inputs).expect("core returned invalid block"); - - find_inputs.result -} - #[cfg(test)] mod tests { - use std::{collections::HashSet, str::FromStr}; - - use crate::types::ScriptHash; - use super::HistoryEntry; - use bitcoin::{Address, Amount}; - use bitcoin_test_data::blocks::mainnet_702861; + use crate::bitcoin::Amount; use serde_json::json; #[test] @@ -644,42 +474,4 @@ mod tests { json!({"tx_hash": "5b75086dafeede555fc8f9a810d8b10df57c46f9f176ccc3dd8d2fa20edd685b", "height": 0, "fee": 123}) ); } - - #[test] - fn test_find_outputs() { - let block = mainnet_702861().to_vec(); - - let addr = Address::from_str("1A9MXXG26vZVySrNNytQK1N8bX42ZuJ6Ax") - .unwrap() - .assume_checked(); - let scripthash = ScriptHash::new(&addr.script_pubkey()); - - let result = &super::filter_block_txs_outputs(block, scripthash)[0]; - assert_eq!( - result.txid.to_string(), - "7bcdcb44422da5a99daad47d6ba1c3d6f2e48f961a75e42c4fa75029d4b0ef49" - ); - assert_eq!(result.pos, 8); - assert_eq!(result.result[0].index, 0); - assert_eq!(result.result[0].value.to_sat(), 709503); - } - - #[test] - fn test_find_inputs() { - let block = mainnet_702861().to_vec(); - let outpoint = bitcoin::OutPoint::from_str( - "cc135e792b37a9c4ffd784f696b1e38bd1197f8e67ae1f96c9f13e4618b91866:3", - ) - .unwrap(); - let mut outpoints = HashSet::new(); - outpoints.insert(outpoint); - - let result = &super::filter_block_txs_inputs(&block, &outpoints)[0]; - assert_eq!( - result.txid.to_string(), - "7bcdcb44422da5a99daad47d6ba1c3d6f2e48f961a75e42c4fa75029d4b0ef49" - ); - assert_eq!(result.pos, 8); - assert_eq!(result.result[0], outpoint); - } } diff --git a/src/tracker.rs b/src/tracker.rs index 2abcf9d..97d43a8 100644 --- a/src/tracker.rs +++ b/src/tracker.rs @@ -1,78 +1,51 @@ -use std::ops::ControlFlow; - +use crate::bitcoin::{BlockHash, Txid}; +use crate::bitcoin_slices::{bsl, EmptyVisitor, Visit}; use anyhow::{Context, Result}; -use bitcoin::{BlockHash, Txid}; -use bitcoin_slices::{bsl, Error::VisitBreak, Visit, Visitor}; +use bindex::IndexedChain; use crate::{ - cache::Cache, - chain::Chain, config::Config, daemon::Daemon, - db::DBStore, - index::Index, mempool::{FeeHistogram, Mempool}, metrics::Metrics, signals::ExitFlag, status::{Balance, ScriptHashStatus, UnspentEntry}, - types::bsl_txid, }; /// Electrum protocol subscriptions' tracker pub struct Tracker { - index: Index, + index: IndexedChain, mempool: Mempool, - metrics: Metrics, ignore_mempool: bool, } -pub(crate) enum Error { - NotReady, -} - impl Tracker { pub fn new(config: &Config, metrics: Metrics) -> Result { - let store = DBStore::open( - &config.db_path, - config.db_log_dir.as_deref(), - config.auto_reindex, - config.db_parallelism, - )?; - let chain = Chain::new(config.network); + let index = + IndexedChain::open(&config.db_dir, config.network).context("failed to open index")?; Ok(Self { - index: Index::load( - store, - chain, - &metrics, - config.index_batch_size, - config.index_lookup_limit, - config.reindex_last_blocks, - ) - .context("failed to open index")?, + index, mempool: Mempool::new(&metrics), - metrics, ignore_mempool: config.ignore_mempool, }) } - pub(crate) fn chain(&self) -> &Chain { - self.index.chain() + pub(crate) fn headers(&self) -> &bindex::Headers { + self.index.headers() } pub(crate) fn fees_histogram(&self) -> &FeeHistogram { self.mempool.fees_histogram() } - pub(crate) fn metrics(&self) -> &Metrics { - &self.metrics - } - pub(crate) fn get_unspent(&self, status: &ScriptHashStatus) -> Vec { - status.get_unspent(self.index.chain()) + status.get_unspent() } pub(crate) fn sync(&mut self, daemon: &Daemon, exit_flag: &ExitFlag) -> Result { - let done = self.index.sync(daemon, exit_flag)?; + exit_flag.poll()?; + let stats = self.index.sync(1000)?; + let done = stats.indexed_blocks == 0; if done && !self.ignore_mempool { self.mempool.sync(daemon, exit_flag); // TODO: double check tip - and retry on diff @@ -80,67 +53,36 @@ impl Tracker { Ok(done) } - pub(crate) fn status(&self) -> Result<(), Error> { - if self.index.is_ready() { - return Ok(()); - } - Err(Error::NotReady) + pub(crate) fn status(&self) -> Result<()> { + Ok(()) } - pub(crate) fn update_scripthash_status( - &self, - status: &mut ScriptHashStatus, - daemon: &Daemon, - cache: &Cache, - ) -> Result { + pub(crate) fn update_scripthash_status(&self, status: &mut ScriptHashStatus) -> Result { let prev_statushash = status.statushash(); - status.sync(&self.index, &self.mempool, daemon, cache)?; + status.sync(&self.index, &self.mempool)?; Ok(prev_statushash != status.statushash()) } pub(crate) fn get_balance(&self, status: &ScriptHashStatus) -> Balance { - status.get_balance(self.chain()) + status.get_balance() } - pub(crate) fn lookup_transaction( - &self, - daemon: &Daemon, - txid: Txid, - ) -> Result)>> { + pub(crate) fn lookup_transaction(&self, txid: Txid) -> Result)>> { // Note: there are two blocks with coinbase transactions having same txid (see BIP-30) - let blockhashes = self.index.filter_by_txid(txid); - let mut result = None; - daemon.for_blocks(blockhashes, |blockhash, block| { - if result.is_some() { - return; // keep first matching transaction + for loc in self.index.locations_by_txid(&txid)? { + let tx_bytes = self.index.get_tx_bytes(&loc)?; + if txid == compute_txid(&tx_bytes)? { + return Ok(Some((loc.block_hash(), tx_bytes.into_boxed_slice()))); } - let mut visitor = FindTransaction::new(txid); - result = match bsl::Block::visit(&block, &mut visitor) { - Ok(_) | Err(VisitBreak) => visitor.found.map(|tx| (blockhash, tx)), - Err(e) => panic!("core returned invalid block: {:?}", e), - }; - })?; - Ok(result) - } -} - -pub struct FindTransaction { - txid: bitcoin::Txid, - found: Option>, // no need to deserialize -} - -impl FindTransaction { - pub fn new(txid: bitcoin::Txid) -> Self { - Self { txid, found: None } - } -} -impl Visitor for FindTransaction { - fn visit_transaction(&mut self, tx: &bsl::Transaction) -> ControlFlow<()> { - if self.txid == bsl_txid(tx) { - self.found = Some(tx.as_ref().into()); - ControlFlow::Break(()) - } else { - ControlFlow::Continue(()) } + Ok(None) } } + +fn compute_txid(tx_bytes: &[u8]) -> Result { + let mut visit = EmptyVisitor {}; + let res = bsl::Transaction::visit(tx_bytes, &mut visit) + .map_err(|err| anyhow!("invalid transaction: {:?}", err))?; + ensure!(res.remaining().is_empty(), "non-empty remaining bytes"); + Ok(Txid::from_raw_hash(res.parsed().txid())) +} diff --git a/src/types.rs b/src/types.rs index 9bd4be6..54a3ae7 100644 --- a/src/types.rs +++ b/src/types.rs @@ -1,292 +1,13 @@ -use anyhow::Result; - -use std::convert::TryFrom; - -use bitcoin::blockdata::block::Header as BlockHeader; -use bitcoin::{ - consensus::encode::{deserialize, Decodable, Encodable}, - hashes::{hash_newtype, sha256, Hash}, - io, OutPoint, Script, Txid, -}; -use bitcoin_slices::bsl; - -macro_rules! impl_consensus_encoding { - ($thing:ident, $($field:ident),+) => ( - impl Encodable for $thing { - #[inline] - fn consensus_encode( - &self, - s: &mut S, - ) -> Result { - let mut len = 0; - $(len += self.$field.consensus_encode(s)?;)+ - Ok(len) - } - } - - impl Decodable for $thing { - #[inline] - fn consensus_decode( - d: &mut D, - ) -> Result<$thing, bitcoin::consensus::encode::Error> { - Ok($thing { - $($field: Decodable::consensus_decode(d)?),+ - }) - } - } - ); -} - -pub const HASH_PREFIX_LEN: usize = 8; -const HEIGHT_SIZE: usize = 4; - -pub(crate) type HashPrefix = [u8; HASH_PREFIX_LEN]; -pub(crate) type SerializedHashPrefixRow = [u8; HASH_PREFIX_ROW_SIZE]; -type Height = u32; -pub(crate) type SerBlock = Vec; - -#[derive(Debug, Serialize, Deserialize, PartialEq)] -pub(crate) struct HashPrefixRow { - prefix: HashPrefix, - height: Height, // transaction confirmed height -} - -pub const HASH_PREFIX_ROW_SIZE: usize = HASH_PREFIX_LEN + HEIGHT_SIZE; - -impl HashPrefixRow { - pub(crate) fn to_db_row(&self) -> SerializedHashPrefixRow { - let mut row = [0; HASH_PREFIX_ROW_SIZE]; - let len = self - .consensus_encode(&mut (&mut row as &mut [u8])) - .expect("in-memory writers don't error"); - debug_assert_eq!(len, HASH_PREFIX_ROW_SIZE); - row - } - - pub(crate) fn from_db_row(row: SerializedHashPrefixRow) -> Self { - deserialize(&row).expect("bad HashPrefixRow") - } - - pub fn height(&self) -> usize { - usize::try_from(self.height).expect("invalid height") - } -} - -impl_consensus_encoding!(HashPrefixRow, prefix, height); - -hash_newtype! { - /// https://electrum-protocol.readthedocs.io/en/latest/protocol-basics.html#script-hashes - #[hash_newtype(backward)] - pub struct ScriptHash(sha256::Hash); -} - -impl ScriptHash { - pub fn new(script: &Script) -> Self { - ScriptHash::hash(script.as_bytes()) - } - - fn prefix(&self) -> HashPrefix { - let mut prefix = HashPrefix::default(); - prefix.copy_from_slice(&self.0[..HASH_PREFIX_LEN]); - prefix - } -} - -pub(crate) struct ScriptHashRow; - -impl ScriptHashRow { - pub(crate) fn scan_prefix(scripthash: ScriptHash) -> HashPrefix { - scripthash.0[..HASH_PREFIX_LEN].try_into().unwrap() - } - - pub(crate) fn row(scripthash: ScriptHash, height: usize) -> HashPrefixRow { - HashPrefixRow { - prefix: scripthash.prefix(), - height: Height::try_from(height).expect("invalid height"), - } - } -} - -// *************************************************************************** +use crate::bitcoin::hashes::{hash_newtype, sha256}; hash_newtype! { /// https://electrum-protocol.readthedocs.io/en/latest/protocol-basics.html#status pub struct StatusHash(sha256::Hash); } -// *************************************************************************** - -fn spending_prefix(prev: OutPoint) -> HashPrefix { - let txid_prefix = HashPrefix::try_from(&prev.txid[..HASH_PREFIX_LEN]).unwrap(); - let value = u64::from_be_bytes(txid_prefix); - let value = value.wrapping_add(prev.vout.into()); - value.to_be_bytes() -} - -pub(crate) struct SpendingPrefixRow; - -impl SpendingPrefixRow { - pub(crate) fn scan_prefix(outpoint: OutPoint) -> HashPrefix { - spending_prefix(outpoint) - } - - pub(crate) fn row(outpoint: OutPoint, height: usize) -> HashPrefixRow { - HashPrefixRow { - prefix: spending_prefix(outpoint), - height: Height::try_from(height).expect("invalid height"), - } - } -} - -// *************************************************************************** - -fn txid_prefix(txid: &Txid) -> HashPrefix { - let mut prefix = [0u8; HASH_PREFIX_LEN]; - prefix.copy_from_slice(&txid[..HASH_PREFIX_LEN]); - prefix -} - -pub(crate) struct TxidRow; - -impl TxidRow { - pub(crate) fn scan_prefix(txid: Txid) -> HashPrefix { - txid_prefix(&txid) - } - - pub(crate) fn row(txid: Txid, height: usize) -> HashPrefixRow { - HashPrefixRow { - prefix: txid_prefix(&txid), - height: Height::try_from(height).expect("invalid height"), - } - } -} - -// *************************************************************************** - -pub(crate) type SerializedHeaderRow = [u8; HEADER_ROW_SIZE]; - -#[derive(Debug, Serialize, Deserialize)] -pub(crate) struct HeaderRow { - pub(crate) header: BlockHeader, -} - -pub const HEADER_ROW_SIZE: usize = 80; - -impl_consensus_encoding!(HeaderRow, header); - -impl HeaderRow { - pub(crate) fn new(header: BlockHeader) -> Self { - Self { header } - } - - pub(crate) fn to_db_row(&self) -> SerializedHeaderRow { - let mut row = [0; HEADER_ROW_SIZE]; - let len = self - .consensus_encode(&mut (&mut row as &mut [u8])) - .expect("in-memory writers don't error"); - debug_assert_eq!(len, HEADER_ROW_SIZE); - row - } - - pub(crate) fn from_db_row(row: SerializedHeaderRow) -> Self { - deserialize(&row).expect("bad HeaderRow") - } -} - -pub(crate) fn bsl_txid(tx: &bsl::Transaction) -> Txid { - bitcoin::Txid::from_slice(tx.txid_sha2().as_slice()).expect("invalid txid") -} - -#[cfg(test)] -mod tests { - use crate::types::{spending_prefix, HashPrefixRow, ScriptHash, ScriptHashRow, TxidRow}; - use bitcoin::{Address, OutPoint, Txid}; - use hex_lit::hex; - use serde_json::{from_str, json}; - - use std::str::FromStr; - - #[test] - fn test_scripthash_serde() { - let hex = "\"4b3d912c1523ece4615e91bf0d27381ca72169dbf6b1c2ffcc9f92381d4984a3\""; - let scripthash: ScriptHash = from_str(hex).unwrap(); - assert_eq!(format!("\"{}\"", scripthash), hex); - assert_eq!(json!(scripthash).to_string(), hex); - } - - #[test] - fn test_scripthash_row() { - let hex = "\"4b3d912c1523ece4615e91bf0d27381ca72169dbf6b1c2ffcc9f92381d4984a3\""; - let scripthash: ScriptHash = from_str(hex).unwrap(); - let row1 = ScriptHashRow::row(scripthash, 123456); - let db_row = row1.to_db_row(); - assert_eq!(db_row, hex!("a384491d38929fcc40e20100")); - let row2 = HashPrefixRow::from_db_row(db_row); - assert_eq!(row1, row2); - } - - #[test] - fn test_scripthash() { - let addr = Address::from_str("1KVNjD3AAnQ3gTMqoTKcWFeqSFujq9gTBT") - .unwrap() - .assume_checked(); - let scripthash = ScriptHash::new(&addr.script_pubkey()); - assert_eq!( - scripthash, - "00dfb264221d07712a144bda338e89237d1abd2db4086057573895ea2659766a" - .parse() - .unwrap() - ); - } - - #[test] - fn test_txid1_prefix() { - // duplicate txids from BIP-30 - let hex = "d5d27987d2a3dfc724e359870c6644b40e497bdc0589a033220fe15429d88599"; - let txid = Txid::from_str(hex).unwrap(); - - let row1 = TxidRow::row(txid, 91812); - let row2 = TxidRow::row(txid, 91842); - - assert_eq!(row1.to_db_row(), hex!("9985d82954e10f22a4660100")); - assert_eq!(row2.to_db_row(), hex!("9985d82954e10f22c2660100")); - } - - #[test] - fn test_txid2_prefix() { - // duplicate txids from BIP-30 - let hex = "e3bf3d07d4b0375638d5f1db5255fe07ba2c4cb067cd81b84ee974b6585fb468"; - let txid = Txid::from_str(hex).unwrap(); - - let row1 = TxidRow::row(txid, 91722); - let row2 = TxidRow::row(txid, 91880); - - // low-endian encoding => rows should be sorted according to block height - assert_eq!(row1.to_db_row(), hex!("68b45f58b674e94e4a660100")); - assert_eq!(row2.to_db_row(), hex!("68b45f58b674e94ee8660100")); - } - - #[test] - fn test_spending_prefix() { - let txid = "000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f" - .parse() - .unwrap(); - - assert_eq!( - spending_prefix(OutPoint { txid, vout: 0 }), - [31, 30, 29, 28, 27, 26, 25, 24] - ); - assert_eq!( - spending_prefix(OutPoint { txid, vout: 10 }), - [31, 30, 29, 28, 27, 26, 25, 34] - ); - assert_eq!( - spending_prefix(OutPoint { txid, vout: 255 }), - [31, 30, 29, 28, 27, 26, 26, 23] - ); - assert_eq!( - spending_prefix(OutPoint { txid, vout: 256 }), - [31, 30, 29, 28, 27, 26, 26, 24] - ); - } +/// The different authentication methods for the client. +#[derive(Clone, Debug, Hash, Eq, PartialEq, Ord, PartialOrd)] +pub enum Auth { + UserPass(String, String), + CookieFile(std::path::PathBuf), } diff --git a/tests/run.sh b/tests/run.sh index 1e2c03c..3a75f89 100755 --- a/tests/run.sh +++ b/tests/run.sh @@ -34,7 +34,7 @@ tail_log() { } echo "Starting $(bitcoind -version | head -n1)..." -bitcoind -regtest -datadir=data/bitcoin -printtoconsole=0 & +bitcoind -rest -regtest -datadir=data/bitcoin -printtoconsole=0 & BITCOIND_PID=$! $BTC -rpcwait getblockcount > /dev/null