diff --git a/.github/workflows/build.yaml b/.github/workflows/build.yaml index 47fefaaa..431bf439 100644 --- a/.github/workflows/build.yaml +++ b/.github/workflows/build.yaml @@ -41,9 +41,3 @@ jobs: rustflags: "" - name: Build run: cargo build --release - - name: Build (electrum only) - if: matrix.name == 'linux' && matrix.toolchain == 'stable' - run: cargo build --release --no-default-features --features electrum - - name: Build (esplora only) - if: matrix.name == 'linux' && matrix.toolchain == 'stable' - run: cargo build --release --no-default-features --features esplora diff --git a/.github/workflows/lint.yaml b/.github/workflows/lint.yaml index f12633bc..a7709482 100644 --- a/.github/workflows/lint.yaml +++ b/.github/workflows/lint.yaml @@ -12,7 +12,20 @@ env: jobs: lint: runs-on: ubuntu-latest - timeout-minutes: 10 + timeout-minutes: 20 + strategy: + fail-fast: false + matrix: + features: + - block-sync,electrum + - block-sync,esplora + - block-sync,electrum,esplora + - transaction-sync,electrum + - transaction-sync,esplora + - transaction-sync,electrum,esplora + - block-sync,transaction-sync,electrum + - block-sync,transaction-sync,esplora + - block-sync,transaction-sync,electrum,esplora steps: - uses: actions/checkout@v6 with: @@ -21,9 +34,5 @@ jobs: with: components: clippy rustflags: "" - - name: Lint - run: cargo clippy --all-targets -- -D warnings - - name: Lint (electrum only) - run: cargo clippy --all-targets --no-default-features --features electrum -- -D warnings - - name: Lint (esplora only) - run: cargo clippy --all-targets --no-default-features --features esplora -- -D warnings + - name: Lint (${{ matrix.features }}) + run: cargo clippy --all-targets --no-default-features --features ${{ matrix.features }} -- -D warnings diff --git a/.gitignore b/.gitignore index 5895bd50..cff37a4b 100644 --- a/.gitignore +++ b/.gitignore @@ -1,5 +1,6 @@ # Test files /datacore +/dataesplora /dataindex /dataldk0 /dataldk1 diff --git a/Cargo.lock b/Cargo.lock index 29d8b9c3..fe646016 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1399,9 +1399,9 @@ dependencies = [ [[package]] name = "electrum-client" -version = "0.20.0" +version = "0.24.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8c7b1f8783238bb18e6e137875b0a66f3dffe6c7ea84066e05d033cf180b150f" +checksum = "a5059f13888a90486e7268bbce59b175f5f76b1c55e5b9c568ceaa42d2b8507c" dependencies = [ "bitcoin", "byteorder", @@ -2555,6 +2555,17 @@ dependencies = [ "lightning", ] +[[package]] +name = "lightning-transaction-sync" +version = "0.2.1" +dependencies = [ + "bitcoin", + "electrum-client 0.24.1", + "esplora-client", + "lightning", + "lightning-macros 0.2.1", +] + [[package]] name = "lightning-types" version = "0.3.1" @@ -3770,7 +3781,8 @@ dependencies = [ "clap", "dircmp", "dirs", - "electrum-client 0.20.0", + "electrum-client 0.24.1", + "esplora-client", "futures", "hex-conservative 0.3.2", "http", @@ -3784,6 +3796,7 @@ dependencies = [ "lightning-net-tokio", "lightning-persister", "lightning-rapid-gossip-sync", + "lightning-transaction-sync", "once_cell", "rand 0.8.7", "regex", diff --git a/Cargo.toml b/Cargo.toml index 029f4de9..981a8ccd 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -8,11 +8,34 @@ rust-version = "1.88.0" name = "rgb-lightning-node" [features] -default = ["electrum", "esplora"] +default = [ + "block-sync", + "transaction-sync", + "electrum", + "esplora", +] +# sync LDK from a bitcoind instance, consuming full blocks over JSON-RPC +block-sync = [ + "dep:lightning-block-sync", +] +# sync LDK from the indexer, without requiring a bitcoind instance +transaction-sync = [ + "dep:lightning-transaction-sync", +] # support indexers implementing the electrum protocol -electrum = ["rgb-lib/electrum", "lightning/electrum"] +electrum = [ + "rgb-lib/electrum", + "lightning/electrum", + "lightning-transaction-sync?/electrum", + "dep:electrum-client", +] # support indexers implementing the esplora protocol -esplora = ["rgb-lib/esplora", "lightning/esplora"] +esplora = [ + "rgb-lib/esplora", + "lightning/esplora", + "lightning-transaction-sync?/esplora-blocking", + "dep:esplora-client", +] [dependencies] amplify = { version = "=4.8.1", default-features = false } @@ -29,17 +52,20 @@ chacha20poly1305 = { version = "0.10.1", features = ["stream"] } chrono = { version = "0.4", default-features = false, features = ["clock"] } clap = "4.5.20" dirs = "5.0.1" +electrum-client = { version = "0.24.0", default-features = false, features = ["use-rustls"], optional = true } +esplora-client = { version = "0.12", default-features = false, features = ["blocking-https-rustls"], optional = true } futures = "0.3" hex = { package = "hex-conservative", version = "0.3.0", default-features = false } lightning = { version = "0.2.0", path = "./rust-lightning/lightning", features = ["dnssec"] } lightning-background-processor = { version = "0.2.0", path = "./rust-lightning/lightning-background-processor" } -lightning-block-sync = { version = "0.2.0", features = ["rpc-client", "tokio"] } +lightning-block-sync = { version = "0.2.0", features = ["rpc-client", "tokio"], optional = true } lightning-dns-resolver = { version = "0.3.0", path = "./rust-lightning/lightning-dns-resolver" } lightning-invoice = { version = "0.34.0", features = ["std"], path = "./rust-lightning/lightning-invoice" } lightning-macros = { version = "0.2.0" } lightning-net-tokio = { version = "0.2.0" } lightning-persister = { version = "0.2.0", path = "./rust-lightning/lightning-persister", features = ["tokio"] } lightning-rapid-gossip-sync = { version = "0.2.0", path = "./rust-lightning/lightning-rapid-gossip-sync" } +lightning-transaction-sync = { version = "0.2.0", path = "./rust-lightning/lightning-transaction-sync", optional = true } rand = "0.8.5" regex = { version = "1.11", default-features = false } rgb-lib = { version = "=0.3.0-beta.7", default-features = false } @@ -61,7 +87,7 @@ zip = { version = "2.2.0", default-features = false, features = ["time", "zstd"] [dev-dependencies] dircmp = "0.2.0" -electrum-client = "0.20.0" +electrum-client = "0.24.0" http = "1.4.0" lazy_static = { version = "1.5.0", default-features = false } lightning = { version = "0.2.0", path = "./rust-lightning/lightning", features = ["_rln_test_hooks"] } diff --git a/README.md b/README.md index 0e671df0..06a59f4e 100644 --- a/README.md +++ b/README.md @@ -50,18 +50,34 @@ Support for the indexer protocols is behind cargo features, `electrum` and To support electrum indexers only: ```sh -cargo install --locked --path . --no-default-features --features electrum +cargo install --locked --path . --no-default-features --features electrum,block-sync,transaction-sync ``` To support esplora indexers only: ```sh -cargo install --locked --path . --no-default-features --features esplora +cargo install --locked --path . --no-default-features --features esplora,block-sync,transaction-sync +``` + +### Chain sync support + +Support for the chain sync backends is behind cargo features, `block-sync` and +`transaction-sync` (both enabled by default). At least one of them needs to be +enabled. See [Sync modes](#sync-modes) for what each backend does. + +To support the block-sync backend only: +```sh +cargo install --locked --path . --no-default-features --features block-sync,electrum,esplora +``` + +To support the transaction-sync backend only: +```sh +cargo install --locked --path . --no-default-features --features transaction-sync,electrum,esplora ``` ## Run In order to operate, the node will need: -- a bitcoind node +- a bitcoind node (only for the `BlockSync` [sync mode](#sync-modes)) - an indexer instance (electrum or esplora) Once services are running, daemons can be started. @@ -401,6 +417,23 @@ Example proxy URLs (only when using proxy transport): | Local | `rpc://127.0.0.1:3000/json-rpc` | | Public | `rpcs://proxy.iriswallet.com/0.2/json-rpc` | +## Sync modes + +The node keeps LDK in sync with the chain in one of two ways, selected at unlock +time via the `ldk_chain_sync` field of the `/unlock` payload (see the +`UnlockRequest` schema in `openapi.yaml` for the exact shape): + +- `BlockSync`: consume full blocks from a trusted/local `bitcoind` over JSON-RPC. + The `bitcoind_rpc_*` parameters are provided under this mode's `config`. This + is the more trust-minimized option, since the node does not rely on an indexer + to tell it which transactions are relevant. +- `TransactionSync`: sync through an electrum/esplora indexer, so no `bitcoind` + is needed. The indexer LDK syncs against is given under this mode's `config` + via `indexer_url` and can differ from the one the RGB wallet uses. + +Both modes are available in a stock build. See +[Chain sync support](#chain-sync-support) to build with only one of them. + ## Test Tests for a few scenarios using the regtest network are included. The same diff --git a/compose.yaml b/compose.yaml index 8aba9866..05822a37 100644 --- a/compose.yaml +++ b/compose.yaml @@ -22,6 +22,27 @@ services: - bitcoind ports: - 50001:50001 + # its electrum RPC is only bound inside the container, so it doesn't clash with electrs + esplora: + image: blockstream/esplora:latest + entrypoint: [] + command: > + /srv/explorer/electrs_bitcoin/bin/electrs + --timestamp + --network regtest + --jsonrpc-import + --cookie user:password + --daemon-rpc-addr bitcoind:18443 + --http-addr 0.0.0.0:3002 + --electrum-rpc-addr 0.0.0.0:60401 + --db-dir /data/db + --cors '*' + volumes: + - ./dataesplora:/data + depends_on: + - bitcoind + ports: + - 3002:3002 proxy: image: ghcr.io/rgb-tools/rgb-proxy-server:0.3.0 ports: diff --git a/openapi.yaml b/openapi.yaml index 1b0d3f01..2377076b 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -2228,6 +2228,61 @@ components: example: 89d28bd306aa9bb906fd0ac31092d04c37c919a171b343083167e2a3cdc60578 status: $ref: '#/components/schemas/HTLCStatus' + LdkChainSync: + oneOf: + - $ref: '#/components/schemas/LdkChainSyncBlockSync' + - $ref: '#/components/schemas/LdkChainSyncTransactionSync' + discriminator: + propertyName: mode + mapping: + BlockSync: '#/components/schemas/LdkChainSyncBlockSync' + TransactionSync: '#/components/schemas/LdkChainSyncTransactionSync' + LdkChainSyncBlockSync: + type: object + required: + - mode + - config + properties: + mode: + type: string + enum: [BlockSync] + config: + type: object + required: + - bitcoind_rpc_username + - bitcoind_rpc_password + - bitcoind_rpc_host + - bitcoind_rpc_port + properties: + bitcoind_rpc_username: + type: string + example: user + bitcoind_rpc_password: + type: string + example: password + bitcoind_rpc_host: + type: string + example: localhost + bitcoind_rpc_port: + type: integer + example: 18443 + LdkChainSyncTransactionSync: + type: object + required: + - mode + - config + properties: + mode: + type: string + enum: [TransactionSync] + config: + type: object + required: + - indexer_url + properties: + indexer_url: + type: string + example: 127.0.0.1:50001 ListAssetsRequest: type: object required: @@ -3436,28 +3491,15 @@ components: type: object required: - password - - bitcoind_rpc_username - - bitcoind_rpc_password - - bitcoind_rpc_host - - bitcoind_rpc_port + - ldk_chain_sync - indexer_url - announce_addresses properties: password: type: string example: nodepassword - bitcoind_rpc_username: - type: string - example: user - bitcoind_rpc_password: - type: string - example: password - bitcoind_rpc_host: - type: string - example: localhost - bitcoind_rpc_port: - type: integer - example: 18443 + ldk_chain_sync: + $ref: '#/components/schemas/LdkChainSync' indexer_url: type: string example: 127.0.0.1:50001 diff --git a/regtest.sh b/regtest.sh index 96b623f8..8ba7e60a 100755 --- a/regtest.sh +++ b/regtest.sh @@ -65,12 +65,26 @@ _wait_for_electrs() { done } +_wait_for_esplora() { + # wait for the esplora REST API to become responsive + start_time=$(date +%s) + until curl -sf http://localhost:3002/blocks/tip/height >/dev/null 2>&1; do + current_time=$(date +%s) + if [ $((current_time - start_time)) -gt $TIMEOUT ]; then + echo "Timeout waiting for esplora to start" + $COMPOSE logs esplora + exit 1 + fi + sleep 1 + done +} + _start_services() { _stop_services - mkdir -p data{core,index,ldk0,ldk1,ldk2} + mkdir -p data{core,index,esplora,ldk0,ldk1,ldk2} # see compose.yaml for the exposed ports - EXPOSED_PORTS=(3000 50001) + EXPOSED_PORTS=(3000 3002 50001) for port in "${EXPOSED_PORTS[@]}"; do if _is_port_bound "$port"; then _die "port $port is already bound, services can't be started" @@ -85,11 +99,13 @@ _start_services() { $COMPOSE up -d echo "waiting for electrs to have completed startup" _wait_for_electrs + echo "waiting for esplora to have completed startup" + _wait_for_esplora } _stop_services() { $COMPOSE down -v --remove-orphans - rm -rf data{core,index,ldk0,ldk1,ldk2} + rm -rf data{core,index,esplora,ldk0,ldk1,ldk2} } _mine() { diff --git a/rust-lightning b/rust-lightning index 66884165..e2a0b8e2 160000 --- a/rust-lightning +++ b/rust-lightning @@ -1 +1 @@ -Subproject commit 66884165edfc26f9c1ec80d9a310fd948e49bb1b +Subproject commit e2a0b8e24dc919ccef289f103fd1f8e1974fcc1d diff --git a/src/error.rs b/src/error.rs index dcc75f70..15458941 100644 --- a/src/error.rs +++ b/src/error.rs @@ -4,7 +4,9 @@ use axum::{ response::{IntoResponse, Response}, Json, }; -use rgb_lib::{BitcoinNetwork, Error as RgbLibError}; +#[cfg(feature = "block-sync")] +use rgb_lib::BitcoinNetwork; +use rgb_lib::Error as RgbLibError; use serde::{Deserialize, Serialize}; #[derive(Debug, Deserialize, Serialize)] @@ -74,6 +76,7 @@ pub enum APIError { #[error("Failed to sync BDK: {0}")] FailedBdkSync(String), + #[cfg(feature = "block-sync")] #[error("Failed to connect to bitcoind client: {0}")] FailedBitcoindConnection(String), @@ -263,6 +266,7 @@ pub enum APIError { #[error("Network error: {0}")] Network(String), + #[cfg(feature = "block-sync")] #[error("The network of the given bitcoind ({0}) doesn't match the node's chain ({1})")] NetworkMismatch(String, BitcoinNetwork), @@ -523,7 +527,6 @@ impl APIError { | APIError::ChangingState | APIError::DuplicatePayment(_) | APIError::FailedBdkSync(_) - | APIError::FailedBitcoindConnection(_) | APIError::FailedBroadcast(_) | APIError::FailedPeerConnection | APIError::InsufficientAssets @@ -534,7 +537,6 @@ impl APIError { | APIError::LockedNode | APIError::MaxFeeExceeded(_) | APIError::MinFeeNotMet(_) - | APIError::NetworkMismatch(_, _) | APIError::NoAvailableUtxos | APIError::NoRoute | APIError::NotInitialized @@ -550,6 +552,10 @@ impl APIError { | APIError::UnsupportedInflation(_) | APIError::UnsupportedLayer1(_) | APIError::UnsupportedTransportType => StatusCode::FORBIDDEN, + #[cfg(feature = "block-sync")] + APIError::FailedBitcoindConnection(_) | APIError::NetworkMismatch(_, _) => { + StatusCode::FORBIDDEN + } APIError::Network(_) | APIError::NoValidTransportEndpoint => { StatusCode::SERVICE_UNAVAILABLE } diff --git a/src/ldk.rs b/src/ldk.rs index 497ca883..2546a47f 100644 --- a/src/ldk.rs +++ b/src/ldk.rs @@ -6,10 +6,15 @@ use bitcoin::secp256k1::{All, PublicKey, Secp256k1}; use bitcoin::{io, Amount, Network}; use bitcoin::{BlockHash, TxOut}; use bitcoin_bech32::WitnessProgram; +#[cfg(feature = "block-sync")] +use lightning::chain; +#[cfg(feature = "transaction-sync")] +use lightning::chain::Confirm; use lightning::chain::{chainmonitor, ChannelMonitorUpdateStatus}; use lightning::chain::{BestBlock, Filter}; use lightning::events::bump_transaction::{BumpTransactionEventHandler, Wallet}; use lightning::events::{Event, PaymentFailureReason, PaymentPurpose, ReplayEvent}; +use lightning::impl_writeable_tlv_based; use lightning::ln::channelmanager::{self, PaymentId, RecentPaymentDetails}; use lightning::ln::channelmanager::{ ChainParameters, ChannelManagerReadArgs, SimpleArcChannelManager, @@ -31,6 +36,7 @@ use lightning::routing::gossip; use lightning::routing::gossip::{NodeId, P2PGossipSync}; use lightning::routing::router::DefaultRouter; use lightning::routing::scoring::{ProbabilisticScorer, ProbabilisticScoringFeeParameters}; +use lightning::routing::utxo::UtxoLookup; use lightning::sign::{ EntropySource, InMemorySigner, KeysManager, NodeSigner, OutputSpender, SpendableOutputDescriptor, @@ -45,13 +51,9 @@ use lightning::util::persist::{ }; use lightning::util::ser::{ReadableArgs, Writeable}; use lightning::util::sweep as ldk_sweep; -use lightning::{chain, impl_writeable_tlv_based}; use lightning_background_processor::{process_events_async, GossipSync, NO_LIQUIDITY_MANAGER}; -use lightning_block_sync::gossip::TokioSpawner; -use lightning_block_sync::init; -use lightning_block_sync::poll; -use lightning_block_sync::SpvClient; -use lightning_block_sync::UnboundedCache; +#[cfg(feature = "block-sync")] +use lightning_block_sync::{init, poll, SpvClient, UnboundedCache}; use lightning_dns_resolver::OMDomainResolver; use lightning_invoice::PaymentSecret; use lightning_net_tokio::SocketDescriptor; @@ -96,17 +98,25 @@ use tokio::runtime::Handle; use tokio::sync::watch::Sender; use tokio::task::JoinHandle; -use crate::bitcoind::BitcoindClient; use crate::disk::{ self, FilesystemLogger, CHANNEL_IDS_FNAME, CHANNEL_PEER_DATA, INBOUND_PAYMENTS_FNAME, MAKER_SWAPS_FNAME, OUTBOUND_PAYMENTS_FNAME, OUTPUT_SPENDER_TXES, TAKER_SWAPS_FNAME, }; use crate::error::APIError; +#[cfg(feature = "block-sync")] +use crate::ldk_chain_backend::block_sync::{BitcoindClient, BlockSyncGossipVerifier}; +#[cfg(feature = "transaction-sync")] +use crate::ldk_chain_backend::sync_chain_data; +#[cfg(feature = "transaction-sync")] +use crate::ldk_chain_backend::transaction_sync::{ + IndexerClient, IndexerGossipVerifier, IndexerSyncClient, +}; +use crate::ldk_chain_backend::{ChainBackend, ChainSetup, DynBroadcaster, DynFeeEstimator}; use crate::rgb::{get_rgb_channel_info_optional, RgbLibWalletWrapper}; use crate::rgb_file_transfer::{ PeerChannelGate, RgbFileTransferHandler, REASSEMBLY_SWEEP_INTERVAL, }; -use crate::routes::{HTLCStatus, SwapStatus, UnlockRequest, DUST_LIMIT_MSAT}; +use crate::routes::{HTLCStatus, LdkChainSync, SwapStatus, UnlockRequest, DUST_LIMIT_MSAT}; use crate::swap::SwapData; use crate::utils::{ check_port_is_available, connect_peer_if_necessary, do_connect_peer, get_current_timestamp, @@ -493,8 +503,8 @@ impl UnlockedAppState { pub(crate) type ChainMonitor = chainmonitor::ChainMonitor< InMemorySigner, Arc, - Arc, - Arc, + Arc, + Arc, Arc, Arc< MonitorUpdatingPersister< @@ -502,23 +512,22 @@ pub(crate) type ChainMonitor = chainmonitor::ChainMonitor< Arc, Arc, Arc, - Arc, - Arc, + Arc, + Arc, >, >, Arc, >; -pub(crate) type GossipVerifier = lightning_block_sync::gossip::GossipVerifier< - TokioSpawner, - Arc, - Arc, ->; +// the UTXO lookup is a trait object so it can be backed either by the block-sync or by the +// transaction-sync gossip verifier +pub(crate) type PeerGossipSync = + P2PGossipSync, Arc, Arc>; pub(crate) type PeerManager = LdkPeerManager< SocketDescriptor, Arc, - Arc, Arc, Arc>>, + Arc, Arc, Arc, Arc, @@ -538,7 +547,7 @@ pub(crate) type Router = DefaultRouter< >; pub(crate) type ChannelManager = - SimpleArcChannelManager; + SimpleArcChannelManager; impl PeerChannelGate for ChannelManager { fn channel_count_with(&self, peer: &PublicKey) -> usize { @@ -569,7 +578,7 @@ pub(crate) type OnionMessenger = LdkOnionMessenger< >; pub(crate) type BumpTxEventHandler = BumpTransactionEventHandler< - Arc, + Arc, Arc, Arc>>, Arc, Arc, @@ -586,9 +595,9 @@ pub(crate) struct RgbOutputSpender { } pub(crate) type OutputSweeper = ldk_sweep::OutputSweeper< - Arc, + Arc, Arc, - Arc, + Arc, Arc, Arc, Arc, @@ -1949,37 +1958,6 @@ pub(crate) async fn start_ldk( let network: Network = bitcoin_network.into(); let ldk_peer_listening_port = static_state.ldk_peer_listening_port; - // Initialize our bitcoind client. - let bitcoind_client = match BitcoindClient::new( - unlock_request.bitcoind_rpc_host.clone(), - unlock_request.bitcoind_rpc_port, - unlock_request.bitcoind_rpc_username.clone(), - unlock_request.bitcoind_rpc_password.clone(), - tokio::runtime::Handle::current(), - Arc::clone(&logger), - ) - .await - { - Ok(client) => Arc::new(client), - Err(e) => { - return Err(APIError::FailedBitcoindConnection(e.to_string())); - } - }; - - // Check that the bitcoind we've connected to is running the network we expect - let bitcoind_chain = bitcoind_client.get_blockchain_info().await.chain; - if bitcoind_chain - != match bitcoin_network { - BitcoinNetwork::Mainnet => "main", - BitcoinNetwork::Testnet => "test", - BitcoinNetwork::Testnet4 => "testnet4", - BitcoinNetwork::Regtest => "regtest", - BitcoinNetwork::Signet | BitcoinNetwork::SignetCustom => "signet", - } - { - return Err(APIError::NetworkMismatch(bitcoind_chain, bitcoin_network)); - } - // RGB setup let indexer_url = &unlock_request.indexer_url; let indexer_protocol = check_indexer_url(indexer_url, bitcoin_network)?; @@ -1990,14 +1968,115 @@ pub(crate) async fn start_ldk( let storage_dir_path = app_state.static_state.storage_dir_path.clone(); fs::write(storage_dir_path.join(INDEXER_URL_FNAME), indexer_url).expect("able to write"); - // Initialize the FeeEstimator - // BitcoindClient implements the FeeEstimator trait, so it'll act as our fee estimator. - let fee_estimator = bitcoind_client.clone(); + // Initialize the chain backend for the requested sync mode + let handle = tokio::runtime::Handle::current(); + let ChainSetup { + backend, + fee_estimator, + broadcaster, + chain_filter, + initial_best_block, + } = match &unlock_request.ldk_chain_sync { + #[cfg(feature = "block-sync")] + LdkChainSync::BlockSync { + bitcoind_rpc_username, + bitcoind_rpc_password, + bitcoind_rpc_host, + bitcoind_rpc_port, + } => { + // Initialize our bitcoind client. + let bitcoind_client = match BitcoindClient::new( + bitcoind_rpc_host.clone(), + *bitcoind_rpc_port, + bitcoind_rpc_username.clone(), + bitcoind_rpc_password.clone(), + handle.clone(), + Arc::clone(&logger), + ) + .await + { + Ok(client) => Arc::new(client), + Err(e) => { + return Err(APIError::FailedBitcoindConnection(e.to_string())); + } + }; - // Initialize the BroadcasterInterface - // BitcoindClient implements the BroadcasterInterface trait, so it'll act as our transaction - // broadcaster. - let broadcaster = bitcoind_client.clone(); + // Check that the bitcoind we've connected to is running the network we expect + let bitcoind_chain = bitcoind_client.get_blockchain_info().await.chain; + if bitcoind_chain + != match bitcoin_network { + BitcoinNetwork::Mainnet => "main", + BitcoinNetwork::Testnet => "test", + BitcoinNetwork::Testnet4 => "testnet4", + BitcoinNetwork::Regtest => "regtest", + BitcoinNetwork::Signet | BitcoinNetwork::SignetCustom => "signet", + } + { + return Err(APIError::NetworkMismatch(bitcoind_chain, bitcoin_network)); + } + + // Poll for the best chain tip, used by the channel manager & spv client + let polled_chain_tip = init::validate_best_block_header(bitcoind_client.as_ref()) + .await + .expect("Failed to fetch best block header and best block"); + let initial_best_block = polled_chain_tip.to_best_block(); + + ChainSetup { + fee_estimator: bitcoind_client.clone(), + broadcaster: bitcoind_client.clone(), + backend: ChainBackend::BlockSync { + client: bitcoind_client, + polled_chain_tip, + }, + chain_filter: None, + initial_best_block, + } + } + #[cfg(feature = "transaction-sync")] + LdkChainSync::TransactionSync { + indexer_url: ln_indexer_url, + } => { + // LDK can sync against a different indexer than the RGB wallet, but when the two + // match the URL has already been checked above + let ln_indexer_protocol = if ln_indexer_url == indexer_url { + indexer_protocol.clone() + } else { + check_indexer_url(ln_indexer_url, bitcoin_network)? + }; + let indexer_client = Arc::new( + IndexerClient::new( + ln_indexer_url.to_string(), + ln_indexer_protocol.clone(), + handle.clone(), + Arc::clone(&logger), + ) + .map_err(|e| APIError::InvalidIndexer(e.to_string()))?, + ); + let tx_sync = Arc::new( + IndexerSyncClient::new( + ln_indexer_url.to_string(), + ln_indexer_protocol, + Arc::clone(&logger), + ) + .map_err(|e| APIError::InvalidIndexer(e.to_string()))?, + ); + let initial_best_block = indexer_client + .get_best_block() + .map_err(|e| APIError::InvalidIndexer(e.to_string()))?; + + let chain_filter: Arc = tx_sync.clone(); + ChainSetup { + fee_estimator: indexer_client.clone(), + broadcaster: indexer_client.clone(), + backend: ChainBackend::TransactionSync { + client: indexer_client, + tx_sync, + }, + chain_filter: Some(chain_filter), + initial_best_block, + } + } + }; // Initialize the KeysManager // The key seed that we use to derive the node privkey (that corresponds to the node pubkey) and @@ -2032,13 +2111,15 @@ pub(crate) async fn start_ldk( 1000, Arc::clone(&keys_manager), Arc::clone(&keys_manager), - Arc::clone(&bitcoind_client), - Arc::clone(&bitcoind_client), + Arc::clone(&broadcaster), + Arc::clone(&fee_estimator), )); // Initialize the ChainMonitor + // in transaction-sync mode `chain_filter` is the indexer chain source, so only relevant + // transactions are watched; in block-sync mode it is `None` since full blocks are consumed let chain_monitor: Arc = Arc::new(chainmonitor::ChainMonitor::new( - None, + chain_filter.clone(), Arc::clone(&broadcaster), Arc::clone(&logger), Arc::clone(&fee_estimator), @@ -2050,11 +2131,6 @@ pub(crate) async fn start_ldk( // Read ChannelMonitor state from disk let mut channelmonitors = persister.read_all_channel_monitors_with_updates().unwrap(); - // Poll for the best chain tip, which may be used by the channel manager & spv client - let polled_chain_tip = init::validate_best_block_header(bitcoind_client.as_ref()) - .await - .expect("Failed to fetch best block header and best block"); - // Initialize routing ProbabilisticScorer let network_graph_path = ldk_data_dir.join("network_graph"); let network_graph = Arc::new(disk::read_network( @@ -2093,9 +2169,14 @@ pub(crate) async fn start_ldk( .channel_handshake_config .negotiate_anchors_zero_fee_htlc_tx = true; user_config.manually_accept_inbound_channels = true; - let mut restarting_node = true; + let manager_file = fs::File::open(ldk_data_dir.join("manager")); + // `restarting_node` and `channel_manager_blockhash` are only consumed by the block-sync + // restart path + #[cfg_attr(not(feature = "block-sync"), allow(unused_variables))] + let restarting_node = manager_file.is_ok(); + #[cfg_attr(not(feature = "block-sync"), allow(unused_variables))] let (channel_manager_blockhash, channel_manager) = { - if let Ok(f) = fs::File::open(ldk_data_dir.join("manager")) { + if let Ok(f) = manager_file { let mut channel_monitor_references = Vec::new(); for (_, channel_monitor) in channelmonitors.iter() { channel_monitor_references.push(channel_monitor); @@ -2117,13 +2198,9 @@ pub(crate) async fn start_ldk( <(BlockHash, ChannelManager)>::read(&mut BufReader::new(f), read_args).unwrap() } else { // We're starting a fresh node. - restarting_node = false; - - let polled_best_block = polled_chain_tip.to_best_block(); - let polled_best_block_hash = polled_best_block.block_hash; let chain_params = ChainParameters { network, - best_block: polled_best_block, + best_block: initial_best_block, }; let fresh_channel_manager = channelmanager::ChannelManager::new( fee_estimator.clone(), @@ -2140,7 +2217,7 @@ pub(crate) async fn start_ldk( cur.as_secs() as u32, ldk_data_dir_path.clone(), ); - (polled_best_block_hash, fresh_channel_manager) + (initial_best_block.block_hash, fresh_channel_manager) } }; @@ -2213,6 +2290,8 @@ pub(crate) async fn start_ldk( fs_store: fs_store.clone(), txes, }); + // `sweeper_best_block` is only used by the block-sync restart path. + #[cfg_attr(not(feature = "block-sync"), allow(unused_variables))] let (sweeper_best_block, output_sweeper) = match fs_store.read( OUTPUT_SWEEPER_PERSISTENCE_PRIMARY_NAMESPACE, OUTPUT_SWEEPER_PERSISTENCE_SECONDARY_NAMESPACE, @@ -2249,79 +2328,107 @@ pub(crate) async fn start_ldk( }; // Sync ChannelMonitors, ChannelManager and OutputSweeper to chain tip - let mut chain_listener_channel_monitors = Vec::new(); - let mut cache = UnboundedCache::new(); - let chain_tip = if restarting_node { - let mut chain_listeners = vec![ - ( - channel_manager_blockhash, - &channel_manager as &(dyn chain::Listen + Send + Sync), - ), - ( - sweeper_best_block.block_hash, - &output_sweeper as &(dyn chain::Listen + Send + Sync), - ), - ]; - - for (blockhash, channel_monitor) in channelmonitors.drain(..) { - let outpoint = channel_monitor.get_funding_txo(); - chain_listener_channel_monitors.push(( - blockhash, - ( - channel_monitor, - broadcaster.clone(), - fee_estimator.clone(), - logger.clone(), - ), - outpoint, - )); - } + // block-sync replays blocks from bitcoind before the SPV client takes over, while + // transaction-sync relies on the indexer via the `Confirm` interface + #[cfg(feature = "block-sync")] + let mut block_sync_cache = UnboundedCache::new(); + // with only block-sync built this is always set below, hence the allow + #[cfg(feature = "block-sync")] + #[cfg_attr(not(feature = "transaction-sync"), allow(unused_assignments))] + let mut block_sync_chain_tip: Option = None; + + match &backend { + #[cfg(feature = "block-sync")] + ChainBackend::BlockSync { + client, + polled_chain_tip, + } => { + let mut chain_listener_channel_monitors = Vec::new(); + let chain_tip = if restarting_node { + let mut chain_listeners = vec![ + ( + channel_manager_blockhash, + &channel_manager as &(dyn chain::Listen + Send + Sync), + ), + ( + sweeper_best_block.block_hash, + &output_sweeper as &(dyn chain::Listen + Send + Sync), + ), + ]; + + for (blockhash, channel_monitor) in channelmonitors.drain(..) { + let outpoint = channel_monitor.get_funding_txo(); + chain_listener_channel_monitors.push(( + blockhash, + ( + channel_monitor, + broadcaster.clone(), + fee_estimator.clone(), + logger.clone(), + ), + outpoint, + )); + } - for monitor_listener_info in chain_listener_channel_monitors.iter_mut() { - chain_listeners.push(( - monitor_listener_info.0, - &monitor_listener_info.1 as &(dyn chain::Listen + Send + Sync), - )); - } + for monitor_listener_info in chain_listener_channel_monitors.iter_mut() { + chain_listeners.push(( + monitor_listener_info.0, + &monitor_listener_info.1 as &(dyn chain::Listen + Send + Sync), + )); + } - let mut attempts = 3; - loop { - match init::synchronize_listeners( - bitcoind_client.as_ref(), - network, - &mut cache, - chain_listeners.clone(), - ) - .await - { - Ok(res) => break res, - Err(e) => { - tracing::error!("Error synchronizing chain: {:?}", e); - attempts -= 1; - if attempts == 0 { - return Err(APIError::FailedBitcoindConnection( - e.into_inner().to_string(), - )); + let mut attempts = 3; + loop { + match init::synchronize_listeners( + client.as_ref(), + network, + &mut block_sync_cache, + chain_listeners.clone(), + ) + .await + { + Ok(res) => break res, + Err(e) => { + tracing::error!("Error synchronizing chain: {:?}", e); + attempts -= 1; + if attempts == 0 { + return Err(APIError::FailedBitcoindConnection( + e.into_inner().to_string(), + )); + } + tokio::time::sleep(Duration::from_secs(1)).await; + } } - tokio::time::sleep(Duration::from_secs(1)).await; } + } else { + *polled_chain_tip + }; + block_sync_chain_tip = Some(chain_tip); + + // Give ChannelMonitors to ChainMonitor + for (_, (channel_monitor, _, _, _), _) in chain_listener_channel_monitors { + let channel_id = channel_monitor.channel_id(); + assert_eq!( + chain_monitor.load_existing_monitor(channel_id, channel_monitor), + Ok(ChannelMonitorUpdateStatus::Completed) + ); + } + } + #[cfg(feature = "transaction-sync")] + ChainBackend::TransactionSync { .. } => { + // Give ChannelMonitors to ChainMonitor + for (_, channel_monitor) in channelmonitors.drain(..) { + let channel_id = channel_monitor.channel_id(); + assert_eq!( + chain_monitor.load_existing_monitor(channel_id, channel_monitor), + Ok(ChannelMonitorUpdateStatus::Completed) + ); } } - } else { - polled_chain_tip - }; - - // Give ChannelMonitors to ChainMonitor - for (_, (channel_monitor, _, _, _), _) in chain_listener_channel_monitors { - let channel_id = channel_monitor.channel_id(); - assert_eq!( - chain_monitor.load_existing_monitor(channel_id, channel_monitor), - Ok(ChannelMonitorUpdateStatus::Completed) - ); } // Optional: Initialize the P2PGossipSync - let gossip_sync = Arc::new(P2PGossipSync::new( + let gossip_sync: Arc = Arc::new(P2PGossipSync::new( Arc::clone(&network_graph), None, Arc::clone(&logger), @@ -2383,14 +2490,27 @@ pub(crate) async fn start_ldk( Arc::clone(&keys_manager), )); - // Install a GossipVerifier in in the P2PGossipSync - let utxo_lookup = GossipVerifier::new( - Arc::clone(&bitcoind_client.bitcoind_rpc_client), - TokioSpawner, - Arc::clone(&gossip_sync), - Arc::clone(&peer_manager), - ); - gossip_sync.add_utxo_lookup(Some(Arc::new(utxo_lookup))); + // Install a UTXO lookup in the P2PGossipSync + let peer_manager_wake = Arc::new({ + let peer_manager = Arc::clone(&peer_manager); + move || peer_manager.process_events() + }); + let utxo_lookup: Arc = match &backend { + #[cfg(feature = "block-sync")] + ChainBackend::BlockSync { client, .. } => Arc::new(BlockSyncGossipVerifier::new( + Arc::clone(&client.bitcoind_rpc_client), + Arc::clone(&gossip_sync), + peer_manager_wake, + handle.clone(), + )), + #[cfg(feature = "transaction-sync")] + ChainBackend::TransactionSync { client, .. } => Arc::new(IndexerGossipVerifier::new( + Arc::clone(client), + Arc::clone(&gossip_sync), + peer_manager_wake, + )), + }; + gossip_sync.add_utxo_lookup(Some(utxo_lookup)); // ## Running LDK // Initialize networking @@ -2421,28 +2541,59 @@ pub(crate) async fn start_ldk( // Connect and Disconnect Blocks let output_sweeper: Arc = Arc::new(output_sweeper); - let channel_manager_listener = channel_manager.clone(); - let chain_monitor_listener = chain_monitor.clone(); - let output_sweeper_listener = output_sweeper.clone(); - let bitcoind_block_source = bitcoind_client.clone(); let stop_listen = Arc::clone(&stop_processing); - tokio::spawn(async move { - let chain_poller = poll::ChainPoller::new(bitcoind_block_source.as_ref(), network); - let chain_listener = ( - chain_monitor_listener, - &(channel_manager_listener, output_sweeper_listener), - ); - let mut spv_client = SpvClient::new(chain_tip, chain_poller, &mut cache, &chain_listener); - loop { - if stop_listen.load(Ordering::Acquire) { - return; - } - if let Err(e) = spv_client.poll_best_tip().await { - tracing::error!("Error while polling best tip: {:?}", e); - } - tokio::time::sleep(Duration::from_secs(1)).await; + match backend { + #[cfg(feature = "block-sync")] + ChainBackend::BlockSync { client, .. } => { + let channel_manager_listener = channel_manager.clone(); + let chain_monitor_listener = chain_monitor.clone(); + let output_sweeper_listener = output_sweeper.clone(); + let chain_tip = + block_sync_chain_tip.expect("block-sync chain tip is set while syncing listeners"); + let mut cache = block_sync_cache; + tokio::spawn(async move { + let chain_poller = poll::ChainPoller::new(client.as_ref(), network); + let chain_listener = ( + chain_monitor_listener, + &(channel_manager_listener, output_sweeper_listener), + ); + let mut spv_client = + SpvClient::new(chain_tip, chain_poller, &mut cache, &chain_listener); + loop { + if stop_listen.load(Ordering::Acquire) { + return; + } + if let Err(e) = spv_client.poll_best_tip().await { + tracing::error!("Error while polling best tip: {:?}", e); + } + tokio::time::sleep(Duration::from_secs(1)).await; + } + }); } - }); + #[cfg(feature = "transaction-sync")] + ChainBackend::TransactionSync { tx_sync, .. } => { + let confirmables: Vec> = vec![ + channel_manager.clone(), + chain_monitor.clone(), + output_sweeper.clone(), + ]; + // bring everything up to the current tip before starting to serve + sync_chain_data(tx_sync.clone(), confirmables.clone()) + .await + .map_err(|e| APIError::InvalidIndexer(e.to_string()))?; + tokio::spawn(async move { + loop { + if stop_listen.load(Ordering::Acquire) { + return; + } + if let Err(e) = sync_chain_data(tx_sync.clone(), confirmables.clone()).await { + tracing::error!("Error while syncing via indexer: {:?}", e); + } + tokio::time::sleep(Duration::from_secs(1)).await; + } + }); + } + } let inbound_payments = Arc::new(Mutex::new(disk::read_inbound_payment_info( &ldk_data_dir.join(INBOUND_PAYMENTS_FNAME), diff --git a/src/bitcoind.rs b/src/ldk_chain_backend/block_sync.rs similarity index 62% rename from src/bitcoind.rs rename to src/ldk_chain_backend/block_sync.rs index c29d445d..45134b61 100644 --- a/src/bitcoind.rs +++ b/src/ldk_chain_backend/block_sync.rs @@ -1,24 +1,30 @@ use base64::{engine::general_purpose, Engine as _}; +use bitcoin::block::Block; use bitcoin::blockdata::transaction::Transaction; use bitcoin::consensus::encode; +use bitcoin::constants::ChainHash; use bitcoin::hash_types::BlockHash; +use bitcoin::transaction::{OutPoint, TxOut}; use lightning::chain::chaininterface::{BroadcasterInterface, ConfirmationTarget, FeeEstimator}; use lightning::log_warn; +use lightning::routing::utxo::{UtxoFuture, UtxoLookup, UtxoLookupError, UtxoResult}; use lightning::util::logger::Logger; +use lightning_block_sync::gossip::UtxoSource; use lightning_block_sync::http::HttpEndpoint; use lightning_block_sync::http::JsonResponse; use lightning_block_sync::rpc::RpcClient; use lightning_block_sync::{AsyncBlockSourceResult, BlockData, BlockHeaderData, BlockSource}; -use std::collections::HashMap; +use std::collections::{HashMap, VecDeque}; use std::convert::TryInto; use std::str::FromStr; -use std::sync::atomic::{AtomicU32, Ordering}; -use std::sync::Arc; +use std::sync::atomic::AtomicU32; +use std::sync::{Arc, Mutex}; use std::time::Duration; use crate::disk::FilesystemLogger; -#[cfg(all(test, feature = "electrum"))] -use crate::test::mock_fee; +use crate::ldk::PeerGossipSync; + +use super::{default_fee_buckets, fee_from_bucket, store_fee_estimates, MIN_FEERATE}; pub struct BitcoindClient { pub(crate) bitcoind_rpc_client: Arc, @@ -109,9 +115,6 @@ impl TryInto for JsonResponse { } } -/// The minimum feerate we are allowed to send, as specify by LDK. -const MIN_FEERATE: u32 = 253; - impl BitcoindClient { pub(crate) async fn new( host: String, @@ -135,40 +138,9 @@ impl BitcoindClient { std::io::Error::new(std::io::ErrorKind::PermissionDenied, "failed to make initial call to bitcoind - please check your RPC user/password and access settings") })?; - let mut fees: HashMap = HashMap::new(); - fees.insert( - ConfirmationTarget::MaximumFeeEstimate, - AtomicU32::new(50000), - ); - fees.insert(ConfirmationTarget::UrgentOnChainSweep, AtomicU32::new(5000)); - fees.insert( - ConfirmationTarget::MinAllowedAnchorChannelRemoteFee, - AtomicU32::new(MIN_FEERATE), - ); - fees.insert( - ConfirmationTarget::MinAllowedNonAnchorChannelRemoteFee, - AtomicU32::new(MIN_FEERATE), - ); - fees.insert( - ConfirmationTarget::AnchorChannelFee, - AtomicU32::new(MIN_FEERATE), - ); - fees.insert( - ConfirmationTarget::NonAnchorChannelFee, - AtomicU32::new(2000), - ); - fees.insert( - ConfirmationTarget::ChannelCloseMinimum, - AtomicU32::new(MIN_FEERATE), - ); - fees.insert( - ConfirmationTarget::OutputSpendingFee, - AtomicU32::new(MIN_FEERATE), - ); - let client = Self { bitcoind_rpc_client: Arc::new(bitcoind_rpc_client), - fees: Arc::new(fees), + fees: Arc::new(default_fee_buckets()), handle: handle.clone(), logger, }; @@ -265,30 +237,14 @@ impl BitcoindClient { ) .await; - fees.get(&ConfirmationTarget::MaximumFeeEstimate) - .unwrap() - .store(very_high_prio_estimate, Ordering::Release); - fees.get(&ConfirmationTarget::UrgentOnChainSweep) - .unwrap() - .store(high_prio_estimate, Ordering::Release); - fees.get(&ConfirmationTarget::MinAllowedAnchorChannelRemoteFee) - .unwrap() - .store(mempoolmin_estimate, Ordering::Release); - fees.get(&ConfirmationTarget::MinAllowedNonAnchorChannelRemoteFee) - .unwrap() - .store(background_estimate - 250, Ordering::Release); - fees.get(&ConfirmationTarget::AnchorChannelFee) - .unwrap() - .store(background_estimate, Ordering::Release); - fees.get(&ConfirmationTarget::NonAnchorChannelFee) - .unwrap() - .store(normal_estimate, Ordering::Release); - fees.get(&ConfirmationTarget::ChannelCloseMinimum) - .unwrap() - .store(background_estimate, Ordering::Release); - fees.get(&ConfirmationTarget::OutputSpendingFee) - .unwrap() - .store(background_estimate, Ordering::Release); + store_fee_estimates( + &fees, + background_estimate, + normal_estimate, + high_prio_estimate, + very_high_prio_estimate, + mempoolmin_estimate, + ); tokio::time::sleep(Duration::from_secs(60)).await; } @@ -305,14 +261,7 @@ impl BitcoindClient { impl FeeEstimator for BitcoindClient { fn get_est_sat_per_1000_weight(&self, confirmation_target: ConfirmationTarget) -> u32 { - let fee = self - .fees - .get(&confirmation_target) - .unwrap() - .load(Ordering::Acquire); - #[cfg(all(test, feature = "electrum"))] - let fee = mock_fee(fee); - fee + fee_from_bucket(&self.fees, confirmation_target) } } @@ -356,3 +305,136 @@ impl BroadcasterInterface for BitcoindClient { }); } } + +// `lightning-block-sync`'s own `GossipVerifier` requires the `P2PGossipSync` to be typed with +// `Arc` as its UTXO lookup, which is incompatible with the trait-object lookup that lets a +// single `PeerManager` type serve both sync backends +pub(crate) struct BlockSyncGossipVerifier { + source: Arc, + gossiper: Arc, + peer_manager_wake: Arc, + handle: tokio::runtime::Handle, + block_cache: Arc>>, +} + +const BLOCK_CACHE_SIZE: usize = 5; + +impl BlockSyncGossipVerifier { + pub(crate) fn new( + source: Arc, + gossiper: Arc, + peer_manager_wake: Arc, + handle: tokio::runtime::Handle, + ) -> Self { + Self { + source, + gossiper, + peer_manager_wake, + handle, + block_cache: Arc::new(Mutex::new(VecDeque::with_capacity(BLOCK_CACHE_SIZE))), + } + } + + async fn retrieve_utxo( + source: Arc, + block_cache: Arc>>, + short_channel_id: u64, + ) -> Result { + let block_height = (short_channel_id >> (5 * 8)) as u32; // most significant three bytes + let transaction_index = ((short_channel_id >> (2 * 8)) & 0x00ff_ffff) as u32; + let output_index = (short_channel_id & 0xffff) as u16; + + let (outpoint, output); + + 'tx_found: { + macro_rules! process_block { + ($block: expr) => {{ + if transaction_index as usize >= $block.txdata.len() { + return Err(UtxoLookupError::UnknownTx); + } + let transaction = &$block.txdata[transaction_index as usize]; + if output_index as usize >= transaction.output.len() { + return Err(UtxoLookupError::UnknownTx); + } + outpoint = OutPoint::new(transaction.compute_txid(), output_index.into()); + output = transaction.output[output_index as usize].clone(); + }}; + } + // Serve the funding output from a recently-fetched block when possible, so a burst of + // announcements referencing the same block only fetches it once + { + let recent_blocks = block_cache.lock().unwrap(); + for (height, block) in recent_blocks.iter() { + if *height == block_height { + process_block!(block); + break 'tx_found; + } + } + } + + let (_, tip_height_opt) = source + .get_best_block() + .await + .map_err(|_| UtxoLookupError::UnknownTx)?; + let block_hash = source + .get_block_hash_by_height(block_height) + .await + .map_err(|_| UtxoLookupError::UnknownTx)?; + if let Some(tip_height) = tip_height_opt { + // The BOLT spec requires nodes to wait for six confirmations before + // announcing a channel; give one block of headroom. + if block_height + 5 > tip_height { + return Err(UtxoLookupError::UnknownTx); + } + } + let block = match source + .get_block(&block_hash) + .await + .map_err(|_| UtxoLookupError::UnknownTx)? + { + BlockData::HeaderOnly(_) => return Err(UtxoLookupError::UnknownTx), + BlockData::FullBlock(block) => block, + }; + process_block!(block); + { + let mut recent_blocks = block_cache.lock().unwrap(); + if !recent_blocks + .iter() + .any(|(height, _)| *height == block_height) + { + if recent_blocks.len() >= BLOCK_CACHE_SIZE { + recent_blocks.pop_front(); + } + recent_blocks.push_back((block_height, block)); + } + } + } + + if source + .is_output_unspent(outpoint) + .await + .map_err(|_| UtxoLookupError::UnknownTx)? + { + Ok(output) + } else { + Err(UtxoLookupError::UnknownTx) + } + } +} + +impl UtxoLookup for BlockSyncGossipVerifier { + fn get_utxo(&self, _chain_hash: &ChainHash, short_channel_id: u64) -> UtxoResult { + let res = UtxoFuture::new(); + let fut = res.clone(); + let source = Arc::clone(&self.source); + let gossiper = Arc::clone(&self.gossiper); + let peer_manager_wake = Arc::clone(&self.peer_manager_wake); + let block_cache = Arc::clone(&self.block_cache); + self.handle.spawn(async move { + let lookup = Self::retrieve_utxo(source, block_cache, short_channel_id).await; + fut.resolve(gossiper.network_graph(), &*gossiper, lookup); + peer_manager_wake(); + }); + UtxoResult::Async(res) + } +} diff --git a/src/ldk_chain_backend/mod.rs b/src/ldk_chain_backend/mod.rs new file mode 100644 index 00000000..8995856f --- /dev/null +++ b/src/ldk_chain_backend/mod.rs @@ -0,0 +1,128 @@ +#[cfg(feature = "block-sync")] +pub(crate) mod block_sync; +#[cfg(feature = "transaction-sync")] +pub(crate) mod transaction_sync; + +use std::collections::HashMap; +use std::sync::atomic::{AtomicU32, Ordering}; +use std::sync::Arc; + +use lightning::chain::chaininterface::ConfirmationTarget; +#[cfg(feature = "transaction-sync")] +use lightning::chain::Confirm; +use lightning::chain::{BestBlock, Filter}; + +// the chain backends are used as trait objects so a single set of LDK type aliases works +// regardless of the selected sync mode +pub(crate) type DynFeeEstimator = dyn lightning::chain::chaininterface::FeeEstimator + Send + Sync; +pub(crate) type DynBroadcaster = + dyn lightning::chain::chaininterface::BroadcasterInterface + Send + Sync; + +pub(crate) const MIN_FEERATE: u32 = 253; + +pub(crate) enum ChainBackend { + #[cfg(feature = "block-sync")] + BlockSync { + client: Arc, + polled_chain_tip: lightning_block_sync::poll::ValidatedBlockHeader, + }, + #[cfg(feature = "transaction-sync")] + TransactionSync { + client: Arc, + tx_sync: Arc, + }, +} + +pub(crate) struct ChainSetup { + pub(crate) backend: ChainBackend, + pub(crate) fee_estimator: Arc, + pub(crate) broadcaster: Arc, + pub(crate) chain_filter: Option>, + pub(crate) initial_best_block: BestBlock, +} + +#[cfg(feature = "transaction-sync")] +pub(crate) async fn sync_chain_data( + tx_sync: Arc, + confirmables: Vec>, +) -> Result<(), Box> { + tokio::task::spawn_blocking(move || tx_sync.sync(confirmables)) + .await + .map_err(|e| -> Box { Box::new(e) })? +} + +fn default_fee_buckets() -> HashMap { + let mut fees = HashMap::new(); + fees.insert( + ConfirmationTarget::MaximumFeeEstimate, + AtomicU32::new(50000), + ); + fees.insert(ConfirmationTarget::UrgentOnChainSweep, AtomicU32::new(5000)); + fees.insert( + ConfirmationTarget::MinAllowedAnchorChannelRemoteFee, + AtomicU32::new(MIN_FEERATE), + ); + fees.insert( + ConfirmationTarget::MinAllowedNonAnchorChannelRemoteFee, + AtomicU32::new(MIN_FEERATE), + ); + fees.insert( + ConfirmationTarget::AnchorChannelFee, + AtomicU32::new(MIN_FEERATE), + ); + fees.insert( + ConfirmationTarget::NonAnchorChannelFee, + AtomicU32::new(2000), + ); + fees.insert( + ConfirmationTarget::ChannelCloseMinimum, + AtomicU32::new(MIN_FEERATE), + ); + fees.insert( + ConfirmationTarget::OutputSpendingFee, + AtomicU32::new(MIN_FEERATE), + ); + fees +} + +fn fee_from_bucket( + fees: &HashMap, + confirmation_target: ConfirmationTarget, +) -> u32 { + let fee = fees + .get(&confirmation_target) + .unwrap() + .load(Ordering::Acquire); + #[cfg(all(test, feature = "electrum"))] + let fee = crate::test::mock_fee(fee); + fee +} + +// both backends map their four priority estimates onto the confirmation targets the same way, +// they differ only in the value used for `MinAllowedAnchorChannelRemoteFee` +fn store_fee_estimates( + fees: &HashMap, + background: u32, + normal: u32, + high_prio: u32, + very_high_prio: u32, + min_allowed_anchor: u32, +) { + let set = |target: ConfirmationTarget, value: u32| { + fees.get(&target).unwrap().store(value, Ordering::Release); + }; + set(ConfirmationTarget::MaximumFeeEstimate, very_high_prio); + set(ConfirmationTarget::UrgentOnChainSweep, high_prio); + set( + ConfirmationTarget::MinAllowedAnchorChannelRemoteFee, + min_allowed_anchor, + ); + set( + ConfirmationTarget::MinAllowedNonAnchorChannelRemoteFee, + background.saturating_sub(250), + ); + set(ConfirmationTarget::AnchorChannelFee, background); + set(ConfirmationTarget::NonAnchorChannelFee, normal); + set(ConfirmationTarget::ChannelCloseMinimum, background); + set(ConfirmationTarget::OutputSpendingFee, background); +} diff --git a/src/ldk_chain_backend/transaction_sync.rs b/src/ldk_chain_backend/transaction_sync.rs new file mode 100644 index 00000000..98be1d1c --- /dev/null +++ b/src/ldk_chain_backend/transaction_sync.rs @@ -0,0 +1,584 @@ +use bitcoin::blockdata::transaction::Transaction; +use bitcoin::constants::ChainHash; +use bitcoin::{Script, TxOut, Txid}; +use lightning::chain::chaininterface::{BroadcasterInterface, ConfirmationTarget, FeeEstimator}; +use lightning::chain::{BestBlock, Confirm, Filter, WatchedOutput}; +use lightning::log_warn; +use lightning::routing::utxo::{UtxoFuture, UtxoLookup, UtxoLookupError, UtxoResult}; +use lightning::util::logger::Logger; +use rgb_lib::wallet::rust_only::IndexerProtocol as RgbLibIndexerProtocol; +use std::collections::HashMap; +use std::io; +use std::sync::atomic::AtomicU32; +use std::sync::Arc; +use std::time::Duration; + +#[cfg(feature = "electrum")] +use { + bitcoin::consensus::encode, + electrum_client::{Client as ElectrumClient, ElectrumApi, Param}, + lightning_transaction_sync::ElectrumSyncClient, + std::str::FromStr, +}; + +#[cfg(feature = "esplora")] +use { + esplora_client::blocking::BlockingClient as EsploraBlockingClient, + esplora_client::Builder as EsploraBuilder, lightning_transaction_sync::EsploraSyncClient, + std::collections::BTreeMap, +}; + +use crate::disk::FilesystemLogger; +use crate::ldk::PeerGossipSync; + +use super::{default_fee_buckets, fee_from_bucket, store_fee_estimates, MIN_FEERATE}; + +type Confirmable = Arc; + +enum IndexerBackend { + #[cfg(feature = "electrum")] + Electrum(Arc), + #[cfg(feature = "esplora")] + Esplora(Arc), +} + +pub(crate) struct IndexerClient { + backend: IndexerBackend, + fees: Arc>, + handle: tokio::runtime::Handle, + logger: Arc, +} + +pub(crate) struct IndexerGossipVerifier { + client: Arc, + gossiper: Arc, + peer_manager_wake: Arc, +} + +pub(crate) enum IndexerSyncClient { + #[cfg(feature = "electrum")] + Electrum(ElectrumSyncClient>), + #[cfg(feature = "esplora")] + Esplora(EsploraSyncClient>), +} + +// `check_indexer_url` only ever returns a protocol whose feature is enabled, so this is just a +// safety net for the single-protocol builds +#[cfg(not(all(feature = "electrum", feature = "esplora")))] +fn unsupported_protocol(protocol: RgbLibIndexerProtocol) -> io::Error { + io::Error::other(format!("{protocol} support is not enabled")) +} + +impl IndexerClient { + pub(crate) fn new( + server_url: String, + protocol: RgbLibIndexerProtocol, + handle: tokio::runtime::Handle, + logger: Arc, + ) -> io::Result { + let fees = Arc::new(default_fee_buckets()); + let backend = match protocol { + #[cfg(feature = "electrum")] + RgbLibIndexerProtocol::Electrum => { + let client = Arc::new(ElectrumClient::new(&server_url).map_err(|e| { + io::Error::other(format!("failed to connect to electrum server: {e}")) + })?); + client.server_features().map_err(|e| { + io::Error::other(format!("failed to query electrum server features: {e}")) + })?; + poll_electrum_fee_estimates( + fees.clone(), + client.clone(), + logger.clone(), + handle.clone(), + ); + IndexerBackend::Electrum(client) + } + #[cfg(feature = "esplora")] + RgbLibIndexerProtocol::Esplora => { + let client = Arc::new(EsploraBuilder::new(&server_url).build_blocking()); + client.get_tip_hash().map_err(|e| { + io::Error::other(format!("failed to connect to esplora server: {e}")) + })?; + poll_esplora_fee_estimates( + fees.clone(), + client.clone(), + logger.clone(), + handle.clone(), + ); + IndexerBackend::Esplora(client) + } + #[cfg(not(all(feature = "electrum", feature = "esplora")))] + protocol => return Err(unsupported_protocol(protocol)), + }; + + Ok(Self { + backend, + fees, + handle, + logger, + }) + } + + pub(crate) fn get_best_block(&self) -> io::Result { + match &self.backend { + #[cfg(feature = "electrum")] + IndexerBackend::Electrum(client) => { + let tip = client.block_headers_subscribe().map_err(|e| { + io::Error::other(format!("failed to fetch electrum tip header: {e}")) + })?; + Ok(BestBlock::new(tip.header.block_hash(), tip.height as u32)) + } + #[cfg(feature = "esplora")] + IndexerBackend::Esplora(client) => { + let tip_hash = client.get_tip_hash().map_err(|e| { + io::Error::other(format!("failed to fetch esplora tip hash: {e}")) + })?; + let tip_height = client + .get_block_status(&tip_hash) + .map_err(|e| { + io::Error::other(format!("failed to fetch esplora tip status: {e}")) + })? + .height + .ok_or_else(|| io::Error::other("esplora tip block has no height"))?; + Ok(BestBlock::new(tip_hash, tip_height)) + } + } + } + + fn tip_height(&self) -> io::Result { + Ok(self.get_best_block()?.height) + } + + // whether `txid`'s output at `vout` has not been spent yet. electrum has no way to query an + // outpoint directly, so its unspent set is queried by script and filtered down to the outpoint + fn is_output_unspent(&self, txid: &Txid, vout: usize, txout: &TxOut) -> io::Result { + match &self.backend { + #[cfg(feature = "electrum")] + IndexerBackend::Electrum(client) => { + let unspents = client + .script_list_unspent(&txout.script_pubkey) + .map_err(|e| { + io::Error::other(format!("failed to fetch electrum unspents: {e}")) + })?; + Ok(unspents + .iter() + .any(|unspent| unspent.tx_hash == *txid && unspent.tx_pos == vout)) + } + #[cfg(feature = "esplora")] + IndexerBackend::Esplora(client) => { + // esplora queries the outpoint directly, so the output itself is not needed + let _ = txout; + let status = client.get_output_status(txid, vout as u64).map_err(|e| { + io::Error::other(format!("failed to fetch esplora output status: {e}")) + })?; + // an unknown output is treated as spent, so an announcement is never resolved + // against an output the indexer cannot vouch for + Ok(status.is_some_and(|status| !status.spent)) + } + } + } +} + +impl IndexerGossipVerifier { + pub(crate) fn new( + client: Arc, + gossiper: Arc, + peer_manager_wake: Arc, + ) -> Self { + Self { + client, + gossiper, + peer_manager_wake, + } + } +} + +impl UtxoLookup for IndexerGossipVerifier { + fn get_utxo(&self, _chain_hash: &ChainHash, short_channel_id: u64) -> UtxoResult { + let result = UtxoFuture::new(); + let future = result.clone(); + let client = self.client.clone(); + let gossiper = self.gossiper.clone(); + let peer_manager_wake = self.peer_manager_wake.clone(); + self.client.handle.spawn(async move { + let lookup = tokio::task::spawn_blocking(move || { + let height = (short_channel_id >> 40) as u32; + let tx_index = ((short_channel_id >> 16) & 0x00ff_ffff) as usize; + let vout = (short_channel_id & 0xffff) as usize; + + // like the block-sync gossip verifier, require the funding output to be buried by + // at least six confirmations (with one block of headroom) before resolving it + let tip_height = client + .tip_height() + .map_err(|_| UtxoLookupError::UnknownTx)?; + if height + 5 > tip_height { + return Err(UtxoLookupError::UnknownTx); + } + + let funding = match &client.backend { + #[cfg(feature = "electrum")] + IndexerBackend::Electrum(c) => { + match electrum_txid_from_pos(c, height as usize, tx_index) + .and_then(|txid| Ok((txid, c.transaction_get(&txid)?))) + { + Ok((txid, tx)) => { + tx.output.get(vout).cloned().map(|txout| (txid, txout)) + } + Err(_) => None, + } + } + #[cfg(feature = "esplora")] + IndexerBackend::Esplora(c) => c + .get_block_hash(height) + .and_then(|block_hash| c.get_txid_at_block_index(&block_hash, tx_index)) + .and_then(|txid| match txid { + Some(txid) => c.get_tx_no_opt(&txid).map(|tx| Some((txid, tx))), + None => Ok(None), + }) + .ok() + .flatten() + .and_then(|(txid, tx)| { + tx.output.get(vout).cloned().map(|txout| (txid, txout)) + }), + }; + + let (txid, txout) = funding.ok_or(UtxoLookupError::UnknownTx)?; + + // like the block-sync gossip verifier, only resolve the announcement if the + // funding output is still unspent, so closed channels don't enter the graph + if !client + .is_output_unspent(&txid, vout, &txout) + .map_err(|_| UtxoLookupError::UnknownTx)? + { + return Err(UtxoLookupError::UnknownTx); + } + + Ok(txout) + }) + .await + .unwrap_or(Err(UtxoLookupError::UnknownTx)); + future.resolve(gossiper.network_graph(), &*gossiper, lookup); + peer_manager_wake(); + }); + UtxoResult::Async(result) + } +} + +#[cfg(feature = "electrum")] +fn electrum_txid_from_pos( + client: &ElectrumClient, + height: usize, + tx_pos: usize, +) -> Result { + let value = client.raw_call( + "blockchain.transaction.id_from_pos", + [ + Param::Usize(height), + Param::Usize(tx_pos), + Param::Bool(true), + ], + )?; + let txid = value + .as_str() + .or_else(|| value.get("tx_hash").and_then(serde_json::Value::as_str)) + .or_else(|| value.get("txid").and_then(serde_json::Value::as_str)) + .or_else(|| value.get("tx_id").and_then(serde_json::Value::as_str)) + .map(str::to_owned) + .ok_or_else(|| electrum_client::Error::InvalidResponse(value.clone()))?; + + Txid::from_str(&txid).map_err(|_| electrum_client::Error::InvalidResponse(value)) +} + +impl IndexerSyncClient { + pub(crate) fn new( + server_url: String, + protocol: RgbLibIndexerProtocol, + logger: Arc, + ) -> io::Result { + match protocol { + #[cfg(feature = "electrum")] + RgbLibIndexerProtocol::Electrum => { + let client = ElectrumSyncClient::new(server_url, logger).map_err(|e| { + io::Error::other(format!("failed to initialize electrum sync client: {e}")) + })?; + Ok(Self::Electrum(client)) + } + #[cfg(feature = "esplora")] + RgbLibIndexerProtocol::Esplora => { + Ok(Self::Esplora(EsploraSyncClient::new(server_url, logger))) + } + #[cfg(not(all(feature = "electrum", feature = "esplora")))] + protocol => Err(unsupported_protocol(protocol)), + } + } + + pub(crate) fn sync( + &self, + confirmables: Vec, + ) -> Result<(), Box> { + match self { + #[cfg(feature = "electrum")] + Self::Electrum(client) => client + .sync(confirmables) + .map_err(|e| -> Box { Box::new(e) }), + #[cfg(feature = "esplora")] + Self::Esplora(client) => client + .sync(confirmables) + .map_err(|e| -> Box { Box::new(e) }), + } + } +} + +impl Filter for IndexerSyncClient { + fn register_tx(&self, txid: &Txid, script_pubkey: &Script) { + match self { + #[cfg(feature = "electrum")] + Self::Electrum(client) => client.register_tx(txid, script_pubkey), + #[cfg(feature = "esplora")] + Self::Esplora(client) => client.register_tx(txid, script_pubkey), + } + } + + fn register_output(&self, output: WatchedOutput) { + match self { + #[cfg(feature = "electrum")] + Self::Electrum(client) => client.register_output(output), + #[cfg(feature = "esplora")] + Self::Esplora(client) => client.register_output(output), + } + } +} + +impl FeeEstimator for IndexerClient { + fn get_est_sat_per_1000_weight(&self, confirmation_target: ConfirmationTarget) -> u32 { + fee_from_bucket(&self.fees, confirmation_target) + } +} + +impl BroadcasterInterface for IndexerClient { + fn broadcast_transactions(&self, txs: &[&Transaction]) { + match &self.backend { + #[cfg(feature = "electrum")] + IndexerBackend::Electrum(client) => { + let txs = txs + .iter() + .map(|tx| encode::serialize(*tx)) + .collect::>(); + let client = client.clone(); + let logger = self.logger.clone(); + self.handle.spawn(async move { + let res = tokio::task::spawn_blocking(move || { + let mut last_error = None; + for tx in txs { + if let Err(e) = client.transaction_broadcast_raw(&tx) { + last_error = Some(e.to_string()); + } + } + last_error.map_or(Ok(()), Err) + }) + .await; + + match res { + Ok(Ok(())) => {} + Ok(Err(e)) => { + log_warn!( + logger, + "Warning, failed to broadcast transaction(s) via electrum: {}", + e + ); + } + Err(e) => { + log_warn!( + logger, + "Warning, failed to spawn electrum broadcaster task: {}", + e + ); + } + } + }); + } + #[cfg(feature = "esplora")] + IndexerBackend::Esplora(client) => { + let txs = txs.iter().map(|tx| (*tx).clone()).collect::>(); + let client = client.clone(); + let logger = self.logger.clone(); + self.handle.spawn(async move { + let res = tokio::task::spawn_blocking(move || { + let mut last_error = None; + for tx in txs { + if let Err(e) = client.broadcast(&tx) { + last_error = Some(e.to_string()); + } + } + last_error.map_or(Ok(()), Err) + }) + .await; + + match res { + Ok(Ok(())) => {} + Ok(Err(e)) => { + log_warn!( + logger, + "Warning, failed to broadcast transaction(s) via esplora: {}", + e + ); + } + Err(e) => { + log_warn!( + logger, + "Warning, failed to spawn esplora broadcaster task: {}", + e + ); + } + } + }); + } + } + } +} + +#[cfg(feature = "electrum")] +fn poll_electrum_fee_estimates( + fees: Arc>, + client: Arc, + logger: Arc, + handle: tokio::runtime::Handle, +) { + handle.spawn(async move { + loop { + let res = tokio::task::spawn_blocking({ + let client = client.clone(); + move || { + Ok::<_, electrum_client::Error>(( + client.estimate_fee(144)?, + client.estimate_fee(18)?, + client.estimate_fee(6)?, + client.estimate_fee(2)?, + )) + } + }) + .await; + + match res { + Ok(Ok((background, normal, high_prio, very_high_prio))) => { + let background_estimate = fee_rate_from_btc_per_kb(background, MIN_FEERATE); + let normal_estimate = fee_rate_from_btc_per_kb(normal, 2000); + let high_prio_estimate = fee_rate_from_btc_per_kb(high_prio, 5000); + let very_high_prio_estimate = fee_rate_from_btc_per_kb(very_high_prio, 50000); + + store_fee_estimates( + &fees, + background_estimate, + normal_estimate, + high_prio_estimate, + very_high_prio_estimate, + MIN_FEERATE, + ); + } + Ok(Err(e)) => { + log_warn!(logger, "Error getting fee estimate from electrum: {}", e); + } + Err(e) => { + log_warn!(logger, "Error polling electrum fee estimates: {}", e); + } + } + + tokio::time::sleep(Duration::from_secs(60)).await; + } + }); +} + +#[cfg(feature = "esplora")] +fn poll_esplora_fee_estimates( + fees: Arc>, + client: Arc, + logger: Arc, + handle: tokio::runtime::Handle, +) { + handle.spawn(async move { + loop { + let res = tokio::task::spawn_blocking({ + let client = client.clone(); + move || client.get_fee_estimates() + }) + .await; + + match res { + Ok(Ok(estimate_map)) => { + let estimate_map = + BTreeMap::from_iter(estimate_map.iter().map(|(k, v)| (*k, *v))); + let background_estimate = + estimate_fee_rate_sat_per_kw(&estimate_map, 144, MIN_FEERATE); + let normal_estimate = estimate_fee_rate_sat_per_kw(&estimate_map, 18, 2000); + let high_prio_estimate = estimate_fee_rate_sat_per_kw(&estimate_map, 6, 5000); + let very_high_prio_estimate = + estimate_fee_rate_sat_per_kw(&estimate_map, 2, 50000); + + store_fee_estimates( + &fees, + background_estimate, + normal_estimate, + high_prio_estimate, + very_high_prio_estimate, + MIN_FEERATE, + ); + } + Ok(Err(e)) => { + log_warn!(logger, "Error getting fee estimate from esplora: {}", e) + } + Err(e) => log_warn!(logger, "Error polling esplora fee estimates: {}", e), + } + + tokio::time::sleep(Duration::from_secs(60)).await; + } + }); +} + +#[cfg(feature = "esplora")] +fn estimate_fee_rate_sat_per_kw( + estimate_map: &BTreeMap, + blocks: u16, + default: u32, +) -> u32 { + let Some(sat_per_vb) = interpolate_fee_rate(estimate_map, blocks) else { + return default; + }; + std::cmp::max((sat_per_vb * 250.0).round() as u32, MIN_FEERATE) +} + +#[cfg(feature = "esplora")] +fn interpolate_fee_rate(estimate_map: &BTreeMap, blocks: u16) -> Option { + if blocks == 0 || estimate_map.is_empty() { + return None; + } + + if let Some(estimate) = estimate_map.get(&blocks) { + return Some(*estimate); + } + + let lower_key = estimate_map.range(..blocks).next_back().map(|(k, _)| *k); + let upper_key = estimate_map.range(blocks..).next().map(|(k, _)| *k); + + match (lower_key, upper_key) { + (Some(x1), Some(x2)) if x1 != x2 => { + let y1 = estimate_map[&x1]; + let y2 = estimate_map[&x2]; + Some(y1 + (blocks as f64 - x1 as f64) / (x2 as f64 - x1 as f64) * (y2 - y1)) + } + (Some(x), _) | (_, Some(x)) => estimate_map.get(&x).copied(), + _ => None, + } +} + +#[cfg(feature = "electrum")] +// electrum reports a negative feerate when it has no estimate available +fn fee_rate_from_btc_per_kb(feerate_btc_per_kb: f64, default: u32) -> u32 { + if !feerate_btc_per_kb.is_finite() || feerate_btc_per_kb.is_sign_negative() { + return default; + } + std::cmp::max( + (feerate_btc_per_kb * 100_000_000.0 / 4.0).round() as u32, + MIN_FEERATE, + ) +} diff --git a/src/main.rs b/src/main.rs index 0dfcd975..71f810d3 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,21 +1,27 @@ #[cfg(not(any(feature = "electrum", feature = "esplora")))] compile_error!("at least one of the `electrum` and `esplora` features needs to be enabled"); +#[cfg(not(any(feature = "block-sync", feature = "transaction-sync")))] +compile_error!( + "at least one of the `block-sync` and `transaction-sync` features needs to be enabled" +); + mod args; mod auth; mod backup; -mod bitcoind; mod crypto; mod disk; mod error; mod ldk; +mod ldk_chain_backend; mod rgb; mod rgb_file_transfer; mod routes; mod swap; mod utils; -// the test suite drives a local electrs instance over the electrum protocol +// the test suite calls into `electrum_client` to wait for electrs to catch up with bitcoind, and +// that crate is only pulled in by the `electrum` feature #[cfg(all(test, feature = "electrum"))] mod test; diff --git a/src/routes.rs b/src/routes.rs index 5f1ab5d0..c3e75936 100644 --- a/src/routes.rs +++ b/src/routes.rs @@ -785,6 +785,20 @@ pub(crate) struct KeysendResponse { pub(crate) status: HTLCStatus, } +#[derive(Clone, Debug, Deserialize, Serialize)] +#[serde(tag = "mode", content = "config")] +pub(crate) enum LdkChainSync { + #[cfg(feature = "block-sync")] + BlockSync { + bitcoind_rpc_username: String, + bitcoind_rpc_password: String, + bitcoind_rpc_host: String, + bitcoind_rpc_port: u16, + }, + #[cfg(feature = "transaction-sync")] + TransactionSync { indexer_url: String }, +} + #[derive(Deserialize, Serialize)] pub(crate) struct ListAssetsRequest { pub(crate) filter_asset_schemas: Vec, @@ -1464,10 +1478,7 @@ pub(crate) enum TransportType { #[derive(Deserialize, Serialize)] pub(crate) struct UnlockRequest { pub(crate) password: String, - pub(crate) bitcoind_rpc_username: String, - pub(crate) bitcoind_rpc_password: String, - pub(crate) bitcoind_rpc_host: String, - pub(crate) bitcoind_rpc_port: u16, + pub(crate) ldk_chain_sync: LdkChainSync, pub(crate) indexer_url: String, pub(crate) announce_addresses: Vec, pub(crate) announce_alias: Option, diff --git a/src/test/electrum_opret_confirm.rs b/src/test/electrum_opret_confirm.rs new file mode 100644 index 00000000..e669aac0 --- /dev/null +++ b/src/test/electrum_opret_confirm.rs @@ -0,0 +1,108 @@ +use super::*; + +// Regression test for a `lightning-transaction-sync` bug: its electrum client ignores the +// `script_pubkey` passed to `Filter::register_tx` and instead derives the script whose history it +// queries from the transaction's *first* output. Indexers do not track provably-unspendable +// outputs, so a transaction whose first output is an OP_RETURN -- which is what an RGB `opret` +// commitment in a channel funding transaction looks like -- is never reported as confirmed, and +// the channel never reaches `channel_ready`. + +#[derive(Default)] +struct ConfirmSpy { + confirmed: Mutex>, +} + +impl Confirm for ConfirmSpy { + fn transactions_confirmed(&self, _header: &Header, txdata: &TransactionData, _height: u32) { + let mut confirmed = self.confirmed.lock().unwrap(); + for (_, tx) in txdata { + confirmed.push(tx.compute_txid()); + } + } + fn transaction_unconfirmed(&self, _txid: &Txid) {} + fn best_block_updated(&self, _header: &Header, _height: u32) {} + fn get_relevant_txids(&self) -> Vec<(Txid, u32, Option)> { + vec![] + } +} + +// broadcasts and confirms a transaction whose first output is an OP_RETURN, returning its txid and +// the scriptPubKey of that first output +fn send_opret_tx() -> (Txid, ScriptBuf) { + let address = bitcoind(&["-rpcwallet=miner", "getnewaddress"]); + let outputs = format!( + r#"[{{"data":"{}"}},{{"{address}":0.001}}]"#, + "de".repeat(32) + ); + let funded = bitcoind(&[ + "-rpcwallet=miner", + "walletcreatefundedpsbt", + "[]", + &outputs, + "0", + // bitcoind inserts the change output at a random position by default, which would leave + // the OP_RETURN somewhere other than the first output: pin change last instead + r#"{"fee_rate":5,"changePosition":2}"#, + ]); + let psbt = serde_json::from_str::(&funded).unwrap()["psbt"] + .as_str() + .unwrap() + .to_string(); + let processed = bitcoind(&["-rpcwallet=miner", "walletprocesspsbt", &psbt]); + let processed_psbt = serde_json::from_str::(&processed).unwrap()["psbt"] + .as_str() + .unwrap() + .to_string(); + let finalized = bitcoind(&["-rpcwallet=miner", "finalizepsbt", &processed_psbt]); + let raw = serde_json::from_str::(&finalized).unwrap()["hex"] + .as_str() + .unwrap() + .to_string(); + let txid = + Txid::from_str(&bitcoind(&["-rpcwallet=miner", "sendrawtransaction", &raw])).unwrap(); + bitcoind(&["-rpcwallet=miner", "-generate", "6"]); + + let tx: BitcoinTransaction = encode::deserialize(&hex_str_to_vec(&raw).unwrap()).unwrap(); + let first_script = tx.output.first().unwrap().script_pubkey.clone(); + assert!( + first_script.is_op_return(), + "the first output must be the OP_RETURN for this test to mean anything" + ); + (txid, first_script) +} + +#[serial_test::serial] +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +#[traced_test] +async fn opret_first_output_still_confirms() { + initialize(); + + let (txid, first_script) = send_opret_tx(); + + // the indexer does not track the OP_RETURN, so resolving the transaction through that output + // cannot work -- this is the precondition that makes the bug bite + let probe = electrum_client::Client::new(ELECTRUM_URL_REGTEST).unwrap(); + let history = probe.script_get_history(&first_script).unwrap(); + assert!( + history.is_empty(), + "expected the indexer to have no history for the OP_RETURN output, got {history:?}" + ); + + let logger = Arc::new(FilesystemLogger::new(PathBuf::from( + "tmp/electrum_opret_confirm", + ))); + let sync_client = ElectrumSyncClient::new(ELECTRUM_URL_REGTEST.to_string(), logger).unwrap(); + sync_client.register_tx(&txid, &first_script); + + let spy = Arc::new(ConfirmSpy::default()); + let confirmable: Arc = spy.clone(); + tokio::task::spawn_blocking(move || sync_client.sync(vec![confirmable]).unwrap()) + .await + .unwrap(); + + let confirmed = spy.confirmed.lock().unwrap().clone(); + assert!( + confirmed.contains(&txid), + "transaction with an OP_RETURN first output was never reported as confirmed" + ); +} diff --git a/src/test/mod.rs b/src/test/mod.rs index 95bce44a..fe7df928 100644 --- a/src/test/mod.rs +++ b/src/test/mod.rs @@ -1,15 +1,27 @@ use amplify::s; use biscuit_auth::{builder::date, macros::*, KeyPair}; +#[cfg(all(feature = "transaction-sync", feature = "electrum"))] +use bitcoin::block::Header; +#[cfg(all(feature = "transaction-sync", feature = "electrum"))] +use bitcoin::consensus::encode; use bitcoin::hashes::sha256::Hash as Sha256; use bitcoin::hashes::Hash; use bitcoin::secp256k1::PublicKey; use bitcoin::{Amount, Denomination}; +#[cfg(all(feature = "transaction-sync", feature = "electrum"))] +use bitcoin::{BlockHash, ScriptBuf, Transaction as BitcoinTransaction, Txid}; use chrono::{DateTime, Local, Utc}; use electrum_client::ElectrumApi; use http::response::Builder; use lazy_static::lazy_static; +#[cfg(all(feature = "transaction-sync", feature = "electrum"))] +use lightning::chain::transaction::TransactionData; +#[cfg(all(feature = "transaction-sync", feature = "electrum"))] +use lightning::chain::{Confirm, Filter}; use lightning::ln::channelmanager::DROP_FUNDING_SIGNED_ON_NODE; use lightning_invoice::Bolt11Invoice; +#[cfg(all(feature = "transaction-sync", feature = "electrum"))] +use lightning_transaction_sync::ElectrumSyncClient; use once_cell::sync::Lazy; use reqwest::Response; use rgb_lib::BitcoinNetwork; @@ -20,12 +32,16 @@ use std::net::SocketAddr; use std::path::{Path, PathBuf}; use std::process::{Command, Stdio}; use std::str::FromStr; +#[cfg(all(feature = "esplora", feature = "transaction-sync"))] +use std::sync::atomic::AtomicBool; use std::sync::{atomic::Ordering, Mutex, Once, RwLock}; use time::OffsetDateTime; use tokio::io::AsyncReadExt; use tokio::net::TcpListener; use tracing_test::traced_test; +#[cfg(all(feature = "transaction-sync", feature = "electrum"))] +use crate::disk::FilesystemLogger; use crate::disk::LDK_LOGS_FILE; use crate::error::APIErrorResponse; use crate::ldk::{ @@ -46,11 +62,11 @@ use crate::routes::{ InitResponse, InvoiceStatus, InvoiceStatusRequest, InvoiceStatusResponse, IssueAssetCFARequest, IssueAssetCFAResponse, IssueAssetIFARequest, IssueAssetIFAResponse, IssueAssetNIARequest, IssueAssetNIAResponse, IssueAssetUDARequest, IssueAssetUDAResponse, KeysendRequest, - KeysendResponse, LNInvoiceRequest, LNInvoiceResponse, ListAssetsRequest, ListAssetsResponse, - ListChannelsResponse, ListPaymentsResponse, ListPeersResponse, ListSwapsResponse, - ListTransactionsRequest, ListTransactionsResponse, ListTransfersRequest, ListTransfersResponse, - ListUnspentsRequest, ListUnspentsResponse, MakerExecuteRequest, MakerInitRequest, - MakerInitResponse, NetworkInfoResponse, NodeInfoResponse, OpenChannelRequest, + KeysendResponse, LNInvoiceRequest, LNInvoiceResponse, LdkChainSync, ListAssetsRequest, + ListAssetsResponse, ListChannelsResponse, ListPaymentsResponse, ListPeersResponse, + ListSwapsResponse, ListTransactionsRequest, ListTransactionsResponse, ListTransfersRequest, + ListTransfersResponse, ListUnspentsRequest, ListUnspentsResponse, MakerExecuteRequest, + MakerInitRequest, MakerInitResponse, NetworkInfoResponse, NodeInfoResponse, OpenChannelRequest, OpenChannelResponse, Payment, Peer, PostAssetMediaResponse, ProvideOutOfBandAckRequest, ProvideOutOfBandAckResponse, ProvideOutOfBandConsignmentResponse, Recipient, RefreshRequest, RefreshResponse, RestoreRequest, RevokeTokenRequest, RgbInvoiceRequest, RgbInvoiceResponse, @@ -62,7 +78,9 @@ use crate::utils::{hex_str, hex_str_to_vec, ELECTRUM_URL_REGTEST, LDK_DIR, PROXY use super::*; -const ELECTRUM_URL: &str = "127.0.0.1:50001"; +// only the transaction-sync tests point a node at esplora +#[cfg(all(feature = "esplora", feature = "transaction-sync"))] +const ESPLORA_URL_REGTEST: &str = "http://127.0.0.1:3002"; const NODE1_PEER_PORT: u16 = 9801; const NODE2_PEER_PORT: u16 = 9802; const NODE3_PEER_PORT: u16 = 9803; @@ -128,6 +146,30 @@ impl Drop for ElectrsRestartGuard { } } +// Makes `mine` also wait for esplora to catch up with bitcoind, for the duration of a test that +// syncs a node through it. Scoped to a guard so the rest of the suite, which only queries electrs, +// doesn't pay for an indexer it never reads, and so a panicking test cannot leak the setting. +#[cfg(all(feature = "esplora", feature = "transaction-sync"))] +static WAIT_ESPLORA_SYNC: AtomicBool = AtomicBool::new(false); + +#[cfg(all(feature = "esplora", feature = "transaction-sync"))] +struct EsploraSyncGuard; + +#[cfg(all(feature = "esplora", feature = "transaction-sync"))] +impl EsploraSyncGuard { + fn set() -> Self { + WAIT_ESPLORA_SYNC.store(true, Ordering::SeqCst); + Self + } +} + +#[cfg(all(feature = "esplora", feature = "transaction-sync"))] +impl Drop for EsploraSyncGuard { + fn drop(&mut self) { + WAIT_ESPLORA_SYNC.store(false, Ordering::SeqCst); + } +} + // Sets a test-override static to a node's pubkey and clears it on drop, so a // panicking test cannot leak the override into the next one struct NodeOverrideGuard(&'static Mutex>); @@ -182,6 +224,27 @@ fn bitcoin_cli() -> [String; 7] { ] } +// runs a bitcoin-cli command against the regtest bitcoind, returning its trimmed stdout. wallet +// commands need an explicit `-rpcwallet=` as their first argument +fn bitcoind(args: &[&str]) -> String { + let output = Command::new("docker") + .stdin(Stdio::null()) + .arg("compose") + .args(bitcoin_cli()) + .args(args) + .output() + .expect("failed to call bitcoin-cli"); + assert!( + output.status.success(), + "bitcoin-cli {args:?} failed: {}", + String::from_utf8_lossy(&output.stderr) + ); + String::from_utf8(output.stdout) + .expect("bitcoin-cli output is not valid UTF-8") + .trim() + .to_string() +} + fn check_preimage_matches_hash(payment: &Payment, expected_payment_hash: &str) { let payment_preimage = payment.preimage.as_ref().unwrap(); let payment_preimage_hash = @@ -212,36 +275,11 @@ async fn check_response_is_nok( fn fund_wallet(address: String, sats: u64) { let amt = Amount::from_sat(sats); let btc_str = amt.to_string_in(Denomination::Bitcoin); - let status = Command::new("docker") - .stdin(Stdio::null()) - .stdout(Stdio::null()) - .stderr(Stdio::null()) - .arg("compose") - .args(bitcoin_cli()) - .arg("-rpcwallet=miner") - .arg("sendtoaddress") - .arg(address) - .arg(btc_str) - .status() - .expect("failed to fund wallet"); - assert!(status.success()); + bitcoind(&["-rpcwallet=miner", "sendtoaddress", &address, &btc_str]); } fn get_txout(txid: &str) -> String { - String::from_utf8( - Command::new("docker") - .stdin(Stdio::null()) - .arg("compose") - .args(bitcoin_cli()) - .arg("-rpcwallet=miner") - .arg("gettxout") - .arg(txid) - .arg("0") - .output() - .expect("failed get txout") - .stdout, - ) - .unwrap() + bitcoind(&["-rpcwallet=miner", "gettxout", txid, "0"]) } async fn start_daemon( @@ -322,6 +360,21 @@ async fn start_node( node_test_dir: &str, node_peer_port: u16, keep_node_dir: bool, +) -> (SocketAddr, String) { + start_node_with( + node_test_dir, + node_peer_port, + keep_node_dir, + default_ldk_chain_sync(), + ) + .await +} + +async fn start_node_with( + node_test_dir: &str, + node_peer_port: u16, + keep_node_dir: bool, + ldk_chain_sync: LdkChainSync, ) -> (SocketAddr, String) { println!("starting node with peer port {node_peer_port}"); let node_address = start_daemon(node_test_dir, node_peer_port, None, keep_node_dir).await; @@ -332,7 +385,7 @@ async fn start_node( init(node_address, &password, None).await; } - unlock(node_address, &password).await; + unlock_with(node_address, &password, ldk_chain_sync).await; println!("node on peer port {node_peer_port} started with address {node_address:?}"); (node_address, password) @@ -1974,13 +2027,30 @@ async fn taker(node_address: SocketAddr, swapstring: String) -> EmptyResponse { .unwrap() } -fn unlock_req(password: &str) -> UnlockRequest { - UnlockRequest { - password: password.to_string(), +// the sync mode the suite unlocks its nodes with: block-sync against the local bitcoind when that +// backend is available, falling back to transaction-sync against the local electrs otherwise +fn default_ldk_chain_sync() -> LdkChainSync { + #[cfg(feature = "block-sync")] + return LdkChainSync::BlockSync { bitcoind_rpc_username: s!("user"), bitcoind_rpc_password: s!("password"), bitcoind_rpc_host: s!("localhost"), bitcoind_rpc_port: 18443, + }; + #[cfg(not(feature = "block-sync"))] + return LdkChainSync::TransactionSync { + indexer_url: ELECTRUM_URL_REGTEST.to_string(), + }; +} + +fn unlock_req(password: &str) -> UnlockRequest { + unlock_req_with(password, default_ldk_chain_sync()) +} + +fn unlock_req_with(password: &str, ldk_chain_sync: LdkChainSync) -> UnlockRequest { + UnlockRequest { + password: password.to_string(), + ldk_chain_sync, indexer_url: ELECTRUM_URL_REGTEST.to_string(), announce_addresses: vec![], announce_alias: Some(s!("RLN_alias")), @@ -1988,8 +2058,16 @@ fn unlock_req(password: &str) -> UnlockRequest { } async fn unlock_res(node_address: SocketAddr, password: &str) -> Response { + unlock_res_with(node_address, password, default_ldk_chain_sync()).await +} + +async fn unlock_res_with( + node_address: SocketAddr, + password: &str, + ldk_chain_sync: LdkChainSync, +) -> Response { println!("unlocking node {node_address}"); - let payload = unlock_req(password); + let payload = unlock_req_with(password, ldk_chain_sync); reqwest::Client::new() .post(format!("http://{node_address}/unlock")) .json(&payload) @@ -2000,17 +2078,8 @@ async fn unlock_res(node_address: SocketAddr, password: &str) -> Response { // Output values (in sats) of an on-chain transaction fn tx_output_sats(txid: &str) -> Vec { - let output = Command::new("docker") - .stdin(Stdio::null()) - .arg("compose") - .args(bitcoin_cli()) - .arg("getrawtransaction") - .arg(txid) - .arg("true") - .output() - .expect("able to call getrawtransaction"); - assert!(output.status.success()); - let tx: serde_json::Value = serde_json::from_slice(&output.stdout).expect("valid tx JSON"); + let raw_tx = bitcoind(&["getrawtransaction", txid, "true"]); + let tx: serde_json::Value = serde_json::from_str(&raw_tx).expect("valid tx JSON"); tx["vout"] .as_array() .expect("vout array") @@ -2024,8 +2093,12 @@ fn tx_output_sats(txid: &str) -> Vec { } async fn unlock(node_address: SocketAddr, password: &str) { + unlock_with(node_address, password, default_ldk_chain_sync()).await +} + +async fn unlock_with(node_address: SocketAddr, password: &str, ldk_chain_sync: LdkChainSync) { println!("unlocking node {node_address}"); - let res = unlock_res(node_address, password).await; + let res = unlock_res_with(node_address, password, ldk_chain_sync).await; check_response_is_ok(res) .await .json::() @@ -2149,18 +2222,7 @@ impl Miner { if self.no_mine_count > 0 { return false; } - let status = Command::new("docker") - .stdin(Stdio::null()) - .stdout(Stdio::null()) - .stderr(Stdio::null()) - .arg("compose") - .args(bitcoin_cli()) - .arg("-rpcwallet=miner") - .arg("-generate") - .arg(num_blocks.to_string()) - .status() - .expect("failed to mine"); - assert!(status.success()); + bitcoind(&["-rpcwallet=miner", "-generate", &num_blocks.to_string()]); true } @@ -2201,6 +2263,10 @@ fn mine_n_blocks(resume: bool, num_blocks: u16) { } } wait_electrs_sync(); + #[cfg(all(feature = "esplora", feature = "transaction-sync"))] + if WAIT_ESPLORA_SYNC.load(Ordering::SeqCst) { + wait_esplora_sync(); + } } fn stop_mining() { @@ -2218,29 +2284,35 @@ fn resume_mining() { } fn get_block_count() -> u32 { - let output = Command::new("docker") - .stdin(Stdio::null()) - .stderr(Stdio::null()) - .arg("compose") - .args(bitcoin_cli()) - .arg("getblockcount") - .output() - .expect("failed to call getblockcount"); - assert!(output.status.success()); - let blockcount_str = - std::str::from_utf8(&output.stdout).expect("could not parse blockcount output"); - blockcount_str - .trim() + bitcoind(&["getblockcount"]) .parse::() .expect("could not parse blockcount") } +// the esplora indexer catches up with bitcoind independently of electrs, so a node syncing +// through it needs its own wait after mining +#[cfg(all(feature = "esplora", feature = "transaction-sync"))] +fn wait_esplora_sync() { + let t_0 = OffsetDateTime::now_utc(); + let blockcount = get_block_count(); + let client = esplora_client::Builder::new(ESPLORA_URL_REGTEST).build_blocking(); + loop { + if client.get_height().is_ok_and(|height| height >= blockcount) { + break; + }; + if (OffsetDateTime::now_utc() - t_0).as_seconds_f32() > 30.0 { + panic!("esplora not syncing with bitcoind"); + } + std::thread::sleep(std::time::Duration::from_millis(100)); + } +} + fn wait_electrs_sync() { let t_0 = OffsetDateTime::now_utc(); let blockcount = get_block_count(); loop { std::thread::sleep(std::time::Duration::from_millis(100)); - let synced = electrum_client::Client::new(ELECTRUM_URL) + let synced = electrum_client::Client::new(ELECTRUM_URL_REGTEST) .is_ok_and(|electrum| electrum.block_header(blockcount as usize).is_ok()); if synced { break; @@ -2297,6 +2369,8 @@ mod close_force_standard; mod concurrent_btc_payments; mod concurrent_openchannel; mod drop_funding_signed; +#[cfg(all(feature = "transaction-sync", feature = "electrum"))] +mod electrum_opret_confirm; mod fail_transfers; mod getchannelid; mod htlc_amount_checks; @@ -2337,5 +2411,7 @@ mod swap_roundtrip_multihop_asset_asset; mod swap_roundtrip_multihop_buy; mod swap_roundtrip_multihop_sell; mod swap_roundtrip_sell; +#[cfg(feature = "transaction-sync")] +mod transaction_sync; mod upload_asset_media; mod vanilla_payment_on_rgb_channel; diff --git a/src/test/transaction_sync.rs b/src/test/transaction_sync.rs new file mode 100644 index 00000000..8d68129d --- /dev/null +++ b/src/test/transaction_sync.rs @@ -0,0 +1,155 @@ +use super::*; + +#[cfg(feature = "electrum")] +const TEST_DIR_BASE_ELECTRUM: &str = "tmp/transaction_sync_electrum/"; +#[cfg(feature = "esplora")] +const TEST_DIR_BASE_ESPLORA: &str = "tmp/transaction_sync_esplora/"; + +// send `invoice` from `node_address`, retrying while the payer has not yet found a route: the only +// route to the payee is multihop and is discovered through gossip, whose channel-announcement UTXO +// lookup goes through the indexer +async fn pay_retrying_route(node_address: SocketAddr, invoice: String) -> String { + let t_0 = OffsetDateTime::now_utc(); + loop { + let payload = SendPaymentRequest { + invoice: invoice.clone(), + amt_msat: None, + asset_id: None, + asset_amount: None, + }; + let res = reqwest::Client::new() + .post(format!("http://{node_address}/sendpayment")) + .json(&payload) + .send() + .await + .unwrap(); + if res.status().is_success() { + let resp: SendPaymentResponse = res.json().await.unwrap(); + // TODO: remove unwrap once RGB offers are enabled + return resp.payment_hash.unwrap(); + } + if (OffsetDateTime::now_utc() - t_0).as_seconds_f32() > 60.0 { + panic!("multihop route to the payee never became available"); + } + tokio::time::sleep(std::time::Duration::from_secs(2)).await; + } +} + +// `ln_indexer_url` selects the indexer LDK syncs against, which can differ from the one the RGB +// wallet uses +async fn transaction_sync_roundtrip(test_dir_base: &str, ln_indexer_url: String) { + initialize(); + + let test_dir_node1 = format!("{test_dir_base}node1"); + let test_dir_node2 = format!("{test_dir_base}node2"); + let test_dir_node3 = format!("{test_dir_base}node3"); + + let ldk_chain_sync = || LdkChainSync::TransactionSync { + indexer_url: ln_indexer_url.clone(), + }; + + let (node1_addr, _) = + start_node_with(&test_dir_node1, NODE1_PEER_PORT, false, ldk_chain_sync()).await; + let (node2_addr, _) = + start_node_with(&test_dir_node2, NODE2_PEER_PORT, false, ldk_chain_sync()).await; + let (node3_addr, _) = + start_node_with(&test_dir_node3, NODE3_PEER_PORT, false, ldk_chain_sync()).await; + + fund_and_create_utxos(node1_addr, None).await; + fund_and_create_utxos(node2_addr, None).await; + fund_and_create_utxos(node3_addr, None).await; + + let asset_id = issue_asset_nia(node1_addr).await.asset_id; + + let node1_pubkey = node_info(node1_addr).await.pubkey; + let node2_pubkey = node_info(node2_addr).await.pubkey; + let node3_pubkey = node_info(node3_addr).await.pubkey; + + // give node2 some asset so it can fund the second channel + let recipient_id = rgb_invoice(node2_addr, None, false).await.recipient_id; + send_asset( + node1_addr, + &asset_id, + Assignment::Fungible(400), + recipient_id, + None, + ) + .await; + mine(false); + refresh_transfers(node2_addr).await; + refresh_transfers(node1_addr).await; + assert_eq!(asset_balance_spendable(node1_addr, &asset_id).await, 600); + + // open two announced asset channels forming the node1 -> node2 -> node3 path + let channel_12 = open_channel( + node1_addr, + &node2_pubkey, + Some(NODE2_PEER_PORT), + None, + Some(3500000), + Some(500), + Some(&asset_id), + ) + .await; + let _channel_23 = open_channel( + node2_addr, + &node3_pubkey, + Some(NODE3_PEER_PORT), + None, + Some(3500000), + Some(300), + Some(&asset_id), + ) + .await; + + // multihop RGB payment node1 -> node3, routed through node2: as the far channel is public, + // node1 has no route hint for it and must resolve the route from gossip, verifying + // node2 -> node3's funding output through the indexer + let LNInvoiceResponse { invoice } = + ln_invoice(node3_addr, None, Some(&asset_id), Some(50), 900).await; + let payment_hash = pay_retrying_route(node1_addr, invoice).await; + wait_for_ln_payment(node1_addr, &payment_hash, HTLCStatus::Succeeded).await; + + wait_for_ln_balance(node1_addr, &asset_id, 450).await; + wait_for_ln_balance(node3_addr, &asset_id, 50).await; + + // restart all nodes: they must sync to the chain tip via the indexer and re-establish their + // channels + shutdown(&[node1_addr, node2_addr, node3_addr]).await; + let (node1_addr, _) = + start_node_with(&test_dir_node1, NODE1_PEER_PORT, true, ldk_chain_sync()).await; + let (node2_addr, _) = + start_node_with(&test_dir_node2, NODE2_PEER_PORT, true, ldk_chain_sync()).await; + let (node3_addr, _) = + start_node_with(&test_dir_node3, NODE3_PEER_PORT, true, ldk_chain_sync()).await; + + wait_for_usable_channels(node1_addr, 1).await; + wait_for_usable_channels(node2_addr, 2).await; + wait_for_usable_channels(node3_addr, 1).await; + wait_for_ln_balance(node1_addr, &asset_id, 450).await; + wait_for_ln_balance(node3_addr, &asset_id, 50).await; + + // cooperatively close the node1 -> node2 channel and check the asset returns on-chain to both + // the initiating and the counterparty node + close_channel(node2_addr, &channel_12.channel_id, &node1_pubkey, false).await; + wait_for_balance(node1_addr, &asset_id, 550).await; + wait_for_balance(node2_addr, &asset_id, 150).await; +} + +#[cfg(feature = "electrum")] +#[serial_test::serial] +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +#[traced_test] +async fn transaction_sync_electrum() { + transaction_sync_roundtrip(TEST_DIR_BASE_ELECTRUM, ELECTRUM_URL_REGTEST.to_string()).await; +} + +// point LDK at a dedicated esplora source while the RGB wallet keeps using electrum +#[cfg(feature = "esplora")] +#[serial_test::serial] +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +#[traced_test] +async fn transaction_sync_esplora() { + let _esplora_sync = EsploraSyncGuard::set(); + transaction_sync_roundtrip(TEST_DIR_BASE_ESPLORA, ESPLORA_URL_REGTEST.to_string()).await; +}