use lightning::chain::chaininterface::{BroadcasterInterface, FeeEstimator};
use lightning::chain::keysinterface::{Sign, KeysInterface};
use lightning::ln::channelmanager::ChannelManager;
+use lightning::ln::msgs::{ChannelMessageHandler, RoutingMessageHandler};
+use lightning::ln::peer_handler::{PeerManager, SocketDescriptor};
use lightning::util::logger::Logger;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
/// [`ChannelManager`]: lightning::ln::channelmanager::ChannelManager
/// [`ChannelManager::write`]: lightning::ln::channelmanager::ChannelManager#impl-Writeable
/// [`FilesystemPersister::persist_manager`]: lightning_persister::FilesystemPersister::persist_manager
- pub fn start<PM, Signer, M, T, K, F, L>(persist_manager: PM, manager: Arc<ChannelManager<Signer, Arc<M>, Arc<T>, Arc<K>, Arc<F>, Arc<L>>>, logger: Arc<L>) -> Self
- where Signer: 'static + Sign,
- M: 'static + chain::Watch<Signer>,
- T: 'static + BroadcasterInterface,
- K: 'static + KeysInterface<Signer=Signer>,
- F: 'static + FeeEstimator,
- L: 'static + Logger,
- PM: 'static + Send + Fn(&ChannelManager<Signer, Arc<M>, Arc<T>, Arc<K>, Arc<F>, Arc<L>>) -> Result<(), std::io::Error>,
+ pub fn start<PM, Signer, M, T, K, F, L, Descriptor: 'static + SocketDescriptor + Send, CM, RM>(
+ persist_channel_manager: PM,
+ channel_manager: Arc<ChannelManager<Signer, Arc<M>, Arc<T>, Arc<K>, Arc<F>, Arc<L>>>,
+ peer_manager: Arc<PeerManager<Descriptor, Arc<CM>, Arc<RM>, Arc<L>>>, logger: Arc<L>,
+ ) -> Self
+ where
+ Signer: 'static + Sign,
+ M: 'static + chain::Watch<Signer>,
+ T: 'static + BroadcasterInterface,
+ K: 'static + KeysInterface<Signer = Signer>,
+ F: 'static + FeeEstimator,
+ L: 'static + Logger,
+ CM: 'static + ChannelMessageHandler,
+ RM: 'static + RoutingMessageHandler,
+ PM: 'static
+ + Send
+ + Fn(
+ &ChannelManager<Signer, Arc<M>, Arc<T>, Arc<K>, Arc<F>, Arc<L>>,
+ ) -> Result<(), std::io::Error>,
{
let stop_thread = Arc::new(AtomicBool::new(false));
let stop_thread_clone = stop_thread.clone();
let handle = thread::spawn(move || -> Result<(), std::io::Error> {
let mut current_time = Instant::now();
loop {
- let updates_available = manager.await_persistable_update_timeout(Duration::from_millis(100));
+ peer_manager.process_events();
+ let updates_available =
+ channel_manager.await_persistable_update_timeout(Duration::from_millis(100));
if updates_available {
- persist_manager(&*manager)?;
+ persist_channel_manager(&*channel_manager)?;
}
// Exit the loop if the background processor was requested to stop.
if stop_thread.load(Ordering::Acquire) == true {
log_trace!(logger, "Terminating background processor.");
- return Ok(())
+ return Ok(());
}
if current_time.elapsed().as_secs() > CHAN_FRESHNESS_TIMER {
log_trace!(logger, "Calling manager's timer_chan_freshness_every_min");
- manager.timer_chan_freshness_every_min();
+ channel_manager.timer_chan_freshness_every_min();
current_time = Instant::now();
}
}
});
- Self {
- stop_thread: stop_thread_clone,
- thread_handle: handle,
- }
+ Self { stop_thread: stop_thread_clone, thread_handle: handle }
}
/// Stop `BackgroundProcessor`'s thread.
use lightning::ln::channelmanager::{ChainParameters, ChannelManager, SimpleArcChannelManager};
use lightning::ln::features::InitFeatures;
use lightning::ln::msgs::ChannelMessageHandler;
+ use lightning::ln::peer_handler::{PeerManager, MessageHandler, SocketDescriptor};
use lightning::util::config::UserConfig;
use lightning::util::events::{Event, EventsProvider, MessageSendEventsProvider, MessageSendEvent};
use lightning::util::logger::Logger;
use std::time::Duration;
use super::BackgroundProcessor;
+ #[derive(Clone, Eq, Hash, PartialEq)]
+ struct TestDescriptor{}
+ impl SocketDescriptor for TestDescriptor {
+ fn send_data(&mut self, _data: &[u8], _resume_read: bool) -> usize {
+ 0
+ }
+
+ fn disconnect_socket(&mut self) {}
+ }
+
type ChainMonitor = chainmonitor::ChainMonitor<InMemorySigner, Arc<test_utils::TestChainSource>, Arc<test_utils::TestBroadcaster>, Arc<test_utils::TestFeeEstimator>, Arc<test_utils::TestLogger>, Arc<FilesystemPersister>>;
struct Node {
node: Arc<SimpleArcChannelManager<ChainMonitor, test_utils::TestBroadcaster, test_utils::TestFeeEstimator, test_utils::TestLogger>>,
+ peer_manager: Arc<PeerManager<TestDescriptor, Arc<test_utils::TestChannelMessageHandler>, Arc<test_utils::TestRoutingMessageHandler>, Arc<test_utils::TestLogger>>>,
persister: Arc<FilesystemPersister>,
logger: Arc<test_utils::TestLogger>,
}
latest_height: 0,
};
let manager = Arc::new(ChannelManager::new(fee_estimator.clone(), chain_monitor.clone(), tx_broadcaster, logger.clone(), keys_manager.clone(), UserConfig::default(), params));
- let node = Node { node: manager, persister, logger };
+ let msg_handler = MessageHandler { chan_handler: Arc::new(test_utils::TestChannelMessageHandler::new()), route_handler: Arc::new(test_utils::TestRoutingMessageHandler::new() )};
+ let peer_manager = Arc::new(PeerManager::new(msg_handler, keys_manager.get_node_secret(), &seed, logger.clone()));
+ let node = Node { node: manager, peer_manager, persister, logger };
nodes.push(node);
}
nodes
// Initiate the background processors to watch each node.
let data_dir = nodes[0].persister.get_data_dir();
let callback = move |node: &ChannelManager<InMemorySigner, Arc<ChainMonitor>, Arc<test_utils::TestBroadcaster>, Arc<KeysManager>, Arc<test_utils::TestFeeEstimator>, Arc<test_utils::TestLogger>>| FilesystemPersister::persist_manager(data_dir.clone(), node);
- let bg_processor = BackgroundProcessor::start(callback, nodes[0].node.clone(), nodes[0].logger.clone());
+ let bg_processor = BackgroundProcessor::start(callback, nodes[0].node.clone(), nodes[0].peer_manager.clone(), nodes[0].logger.clone());
// Go through the channel creation process until each node should have something persisted.
let tx = open_channel!(nodes[0], nodes[1], 100000);
let nodes = create_nodes(1, "test_chan_freshness_called".to_string());
let data_dir = nodes[0].persister.get_data_dir();
let callback = move |node: &ChannelManager<InMemorySigner, Arc<ChainMonitor>, Arc<test_utils::TestBroadcaster>, Arc<KeysManager>, Arc<test_utils::TestFeeEstimator>, Arc<test_utils::TestLogger>>| FilesystemPersister::persist_manager(data_dir.clone(), node);
- let bg_processor = BackgroundProcessor::start(callback, nodes[0].node.clone(), nodes[0].logger.clone());
+ let bg_processor = BackgroundProcessor::start(callback, nodes[0].node.clone(), nodes[0].peer_manager.clone(), nodes[0].logger.clone());
loop {
let log_entries = nodes[0].logger.lines.lock().unwrap();
let desired_log = "Calling manager's timer_chan_freshness_every_min".to_string();
}
let nodes = create_nodes(2, "test_persist_error".to_string());
- let bg_processor = BackgroundProcessor::start(persist_manager, nodes[0].node.clone(), nodes[0].logger.clone());
+ let bg_processor = BackgroundProcessor::start(persist_manager, nodes[0].node.clone(), nodes[0].peer_manager.clone(), nodes[0].logger.clone());
open_channel!(nodes[0], nodes[1], 100000);
let _ = bg_processor.thread_handle.join().unwrap().expect_err("Errored persisting manager: test");
_ if err.starts_with("Cannot send value that would put our balance under counterparty-announced channel reserve value") => {},
_ if err.starts_with("Cannot send value that would overdraw remaining funds.") => {},
_ if err.starts_with("Cannot send value that would not leave enough to pay for fees.") => {},
- _ => panic!(err),
+ _ => panic!("{}", err),
}
},
APIError::MonitorUpdateFailed => {
msgs::DecodeError::InvalidValue => return,
msgs::DecodeError::BadLengthDescriptor => return,
msgs::DecodeError::ShortRead => panic!("We picked the length..."),
- msgs::DecodeError::Io(e) => panic!(format!("{:?}", e)),
+ msgs::DecodeError::Io(e) => panic!("{:?}", e),
}
}
}}
}
/// Result type for `BlockSource` requests.
-type BlockSourceResult<T> = Result<T, BlockSourceError>;
+pub type BlockSourceResult<T> = Result<T, BlockSourceError>;
// TODO: Replace with BlockSourceResult once `async` trait functions are supported. For details,
// see: https://areweasyncyet.rs.
/// Result type for asynchronous `BlockSource` requests.
-type AsyncBlockSourceResult<'a, T> = Pin<Box<dyn Future<Output = BlockSourceResult<T>> + 'a + Send>>;
+pub type AsyncBlockSourceResult<'a, T> = Pin<Box<dyn Future<Output = BlockSourceResult<T>> + 'a + Send>>;
/// Error type for `BlockSource` requests.
///
Utilities to manage Rust-Lightning channel data persistence and retrieval.
"""
+[features]
+unstable = ["lightning/unstable"]
+
[dependencies]
bitcoin = "0.26"
lightning = { version = "0.0.13", path = "../lightning" }
#![deny(broken_intra_doc_links)]
#![deny(missing_docs)]
+#![cfg_attr(all(test, feature = "unstable"), feature(test))]
+#[cfg(all(test, feature = "unstable"))] extern crate test;
+
mod util;
extern crate lightning;
added_monitors.clear();
}
}
+
+#[cfg(all(test, feature = "unstable"))]
+pub mod bench {
+ use test::Bencher;
+
+ #[bench]
+ fn bench_sends(bench: &mut Bencher) {
+ let persister_a = super::FilesystemPersister::new("bench_filesystem_persister_a".to_string());
+ let persister_b = super::FilesystemPersister::new("bench_filesystem_persister_b".to_string());
+ lightning::ln::channelmanager::bench::bench_two_sends(bench, persister_a, persister_b);
+ }
+}
use bitcoin::blockdata::block::{Block, BlockHeader};
use chain;
-use chain::Filter;
+use chain::{Filter, WatchedOutput};
use chain::chaininterface::{BroadcasterInterface, FeeEstimator};
use chain::channelmonitor;
use chain::channelmonitor::{ChannelMonitor, ChannelMonitorUpdate, ChannelMonitorUpdateErr, MonitorEvent, Persist};
/// descendants of such transactions. It is not necessary to re-fetch the block to obtain
/// updated `txdata`.
pub fn block_connected(&self, header: &BlockHeader, txdata: &TransactionData, height: u32) {
+ let mut dependent_txdata = Vec::new();
let monitors = self.monitors.read().unwrap();
for monitor in monitors.values() {
let mut txn_outputs = monitor.block_connected(header, txdata, height, &*self.broadcaster, &*self.fee_estimator, &*self.logger);
+ // Register any new outputs with the chain source for filtering, storing any dependent
+ // transactions from within the block that previously had not been included in txdata.
if let Some(ref chain_source) = self.chain_source {
+ let block_hash = header.block_hash();
for (txid, outputs) in txn_outputs.drain(..) {
for (idx, output) in outputs.iter() {
- chain_source.register_output(&OutPoint { txid, index: *idx as u16 }, &output.script_pubkey);
+ // Register any new outputs with the chain source for filtering and recurse
+ // if it indicates that there are dependent transactions within the block
+ // that had not been previously included in txdata.
+ let output = WatchedOutput {
+ block_hash: Some(block_hash),
+ outpoint: OutPoint { txid, index: *idx as u16 },
+ script_pubkey: output.script_pubkey.clone(),
+ };
+ if let Some(tx) = chain_source.register_output(output) {
+ dependent_txdata.push(tx);
+ }
}
}
}
}
+
+ // Recursively call for any dependent transactions that were identified by the chain source.
+ if !dependent_txdata.is_empty() {
+ dependent_txdata.sort_unstable_by_key(|(index, _tx)| *index);
+ dependent_txdata.dedup_by_key(|(index, _tx)| *index);
+ let txdata: Vec<_> = dependent_txdata.iter().map(|(index, tx)| (*index, tx)).collect();
+ self.block_connected(header, &txdata, height);
+ }
}
/// Dispatches to per-channel monitors, which are responsible for updating their on-chain view
pending_events
}
}
+
+#[cfg(test)]
+mod tests {
+ use ::{check_added_monitors, get_local_commitment_txn};
+ use ln::features::InitFeatures;
+ use ln::functional_test_utils::*;
+ use util::events::EventsProvider;
+ use util::events::MessageSendEventsProvider;
+ use util::test_utils::{OnRegisterOutput, TxOutReference};
+
+ /// Tests that in-block dependent transactions are processed by `block_connected` when not
+ /// included in `txdata` but returned by [`chain::Filter::register_output`]. For instance,
+ /// a (non-anchor) commitment transaction's HTLC output may be spent in the same block as the
+ /// commitment transaction itself. An Electrum client may filter the commitment transaction but
+ /// needs to return the HTLC transaction so it can be processed.
+ #[test]
+ fn connect_block_checks_dependent_transactions() {
+ let chanmon_cfgs = create_chanmon_cfgs(2);
+ let node_cfgs = create_node_cfgs(2, &chanmon_cfgs);
+ let node_chanmgrs = create_node_chanmgrs(2, &node_cfgs, &[None, None]);
+ let nodes = create_network(2, &node_cfgs, &node_chanmgrs);
+ let channel = create_announced_chan_between_nodes(
+ &nodes, 0, 1, InitFeatures::known(), InitFeatures::known());
+
+ // Send a payment, saving nodes[0]'s revoked commitment and HTLC-Timeout transactions.
+ let (commitment_tx, htlc_tx) = {
+ let payment_preimage = route_payment(&nodes[0], &vec!(&nodes[1])[..], 5_000_000).0;
+ let mut txn = get_local_commitment_txn!(nodes[0], channel.2);
+ claim_payment(&nodes[0], &vec!(&nodes[1])[..], payment_preimage, 5_000_000);
+
+ assert_eq!(txn.len(), 2);
+ (txn.remove(0), txn.remove(0))
+ };
+
+ // Set expectations on nodes[1]'s chain source to return dependent transactions.
+ let htlc_output = TxOutReference(commitment_tx.clone(), 0);
+ let to_local_output = TxOutReference(commitment_tx.clone(), 1);
+ let htlc_timeout_output = TxOutReference(htlc_tx.clone(), 0);
+ nodes[1].chain_source
+ .expect(OnRegisterOutput { with: htlc_output, returns: Some((1, htlc_tx)) })
+ .expect(OnRegisterOutput { with: to_local_output, returns: None })
+ .expect(OnRegisterOutput { with: htlc_timeout_output, returns: None });
+
+ // Notify nodes[1] that nodes[0]'s revoked commitment transaction was mined. The chain
+ // source should return the dependent HTLC transaction when the HTLC output is registered.
+ mine_transaction(&nodes[1], &commitment_tx);
+
+ // Clean up so uninteresting assertions don't fail.
+ check_added_monitors!(nodes[1], 1);
+ nodes[1].node.get_and_clear_pending_msg_events();
+ nodes[1].node.get_and_clear_pending_events();
+ }
+}
use ln::channelmanager::{HTLCSource, PaymentPreimage, PaymentHash};
use ln::onchaintx::{OnchainTxHandler, InputDescriptors};
use chain;
+use chain::WatchedOutput;
use chain::chaininterface::{BroadcasterInterface, FeeEstimator};
use chain::transaction::{OutPoint, TransactionData};
use chain::keysinterface::{SpendableOutputDescriptor, StaticPaymentOutputDescriptor, DelayedPaymentOutputDescriptor, Sign, KeysInterface};
for (txid, outputs) in lock.get_outputs_to_watch().iter() {
for (index, script_pubkey) in outputs.iter() {
assert!(*index <= u16::max_value() as u32);
- filter.register_output(&OutPoint { txid: *txid, index: *index as u16 }, script_pubkey);
+ filter.register_output(WatchedOutput {
+ block_hash: None,
+ outpoint: OutPoint { txid: *txid, index: *index as u16 },
+ script_pubkey: script_pubkey.clone(),
+ });
}
}
}
use bitcoin::blockdata::block::{Block, BlockHeader};
use bitcoin::blockdata::script::Script;
-use bitcoin::blockdata::transaction::TxOut;
+use bitcoin::blockdata::transaction::{Transaction, TxOut};
use bitcoin::hash_types::{BlockHash, Txid};
use chain::channelmonitor::{ChannelMonitor, ChannelMonitorUpdate, ChannelMonitorUpdateErr, MonitorEvent};
/// a spending condition.
fn register_tx(&self, txid: &Txid, script_pubkey: &Script);
- /// Registers interest in spends of a transaction output identified by `outpoint` having
- /// `script_pubkey` as the spending condition.
- fn register_output(&self, outpoint: &OutPoint, script_pubkey: &Script);
+ /// Registers interest in spends of a transaction output.
+ ///
+ /// Optionally, when `output.block_hash` is set, should return any transaction spending the
+ /// output that is found in the corresponding block along with its index.
+ ///
+ /// This return value is useful for Electrum clients in order to supply in-block descendant
+ /// transactions which otherwise were not included. This is not necessary for other clients if
+ /// such descendant transactions were already included (e.g., when a BIP 157 client provides the
+ /// full block).
+ fn register_output(&self, output: WatchedOutput) -> Option<(usize, Transaction)>;
+}
+
+/// A transaction output watched by a [`ChannelMonitor`] for spends on-chain.
+///
+/// Used to convey to a [`Filter`] such an output with a given spending condition. Any transaction
+/// spending the output must be given to [`ChannelMonitor::block_connected`] either directly or via
+/// the return value of [`Filter::register_output`].
+///
+/// If `block_hash` is `Some`, this indicates the output was created in the corresponding block and
+/// may have been spent there. See [`Filter::register_output`] for details.
+///
+/// [`ChannelMonitor`]: channelmonitor::ChannelMonitor
+/// [`ChannelMonitor::block_connected`]: channelmonitor::ChannelMonitor::block_connected
+pub struct WatchedOutput {
+ /// First block where the transaction output may have been spent.
+ pub block_hash: Option<BlockHash>,
+
+ /// Outpoint identifying the transaction output.
+ pub outpoint: OutPoint,
+
+ /// Spending condition of the transaction output.
+ pub script_pubkey: Script,
}
impl<T: Listen> Listen for std::ops::Deref<Target = T> {
#![allow(bare_trait_objects)]
#![allow(ellipsis_inclusive_range_patterns)]
-#![cfg_attr(all(test, feature = "unstable"), feature(test))]
-#[cfg(all(test, feature = "unstable"))] extern crate test;
+#![cfg_attr(all(any(test, feature = "_test_utils"), feature = "unstable"), feature(test))]
+#[cfg(all(any(test, feature = "_test_utils"), feature = "unstable"))] extern crate test;
extern crate bitcoin;
#[cfg(any(test, feature = "_test_utils"))] extern crate hex;
#[cfg(not(any(test, feature = "_test_utils")))]
channel_state: Mutex<ChannelHolder<Signer>>,
our_network_key: SecretKey,
+ our_network_pubkey: PublicKey,
/// Used to track the last value sent in a node_announcement "timestamp" field. We ensure this
/// value increases strictly since we don't assume access to a time source.
latest_block_height: AtomicUsize::new(params.latest_height),
last_block_hash: RwLock::new(params.latest_hash),
- secp_ctx,
channel_state: Mutex::new(ChannelHolder{
by_id: HashMap::new(),
pending_msg_events: Vec::new(),
}),
our_network_key: keys_manager.get_node_secret(),
+ our_network_pubkey: PublicKey::from_secret_key(&secp_ctx, &keys_manager.get_node_secret()),
+ secp_ctx,
last_node_announcement_serial: AtomicUsize::new(0),
/// Gets the node_id held by this ChannelManager
pub fn get_our_node_id(&self) -> PublicKey {
- PublicKey::from_secret_key(&self.secp_ctx, &self.our_network_key)
+ self.our_network_pubkey.clone()
}
/// Restores a single, given channel to normal operation after a
latest_block_height: AtomicUsize::new(latest_block_height as usize),
last_block_hash: RwLock::new(last_block_hash),
- secp_ctx,
channel_state: Mutex::new(ChannelHolder {
by_id,
pending_msg_events: Vec::new(),
}),
our_network_key: args.keys_manager.get_node_secret(),
+ our_network_pubkey: PublicKey::from_secret_key(&secp_ctx, &args.keys_manager.get_node_secret()),
+ secp_ctx,
last_node_announcement_serial: AtomicUsize::new(last_node_announcement_serial as usize),
}
}
}
+
+#[cfg(all(any(test, feature = "_test_utils"), feature = "unstable"))]
+pub mod bench {
+ use chain::Listen;
+ use chain::chainmonitor::ChainMonitor;
+ use chain::channelmonitor::Persist;
+ use chain::keysinterface::{KeysManager, InMemorySigner};
+ use chain::transaction::OutPoint;
+ use ln::channelmanager::{ChainParameters, ChannelManager, PaymentHash, PaymentPreimage};
+ use ln::features::InitFeatures;
+ use ln::functional_test_utils::*;
+ use ln::msgs::ChannelMessageHandler;
+ use routing::network_graph::NetworkGraph;
+ use routing::router::get_route;
+ use util::test_utils;
+ use util::config::UserConfig;
+ use util::events::{Event, EventsProvider, MessageSendEvent, MessageSendEventsProvider};
+
+ use bitcoin::hashes::Hash;
+ use bitcoin::hashes::sha256::Hash as Sha256;
+ use bitcoin::{Block, BlockHeader, Transaction, TxOut};
+
+ use std::sync::Mutex;
+
+ use test::Bencher;
+
+ struct NodeHolder<'a, P: Persist<InMemorySigner>> {
+ node: &'a ChannelManager<InMemorySigner,
+ &'a ChainMonitor<InMemorySigner, &'a test_utils::TestChainSource,
+ &'a test_utils::TestBroadcaster, &'a test_utils::TestFeeEstimator,
+ &'a test_utils::TestLogger, &'a P>,
+ &'a test_utils::TestBroadcaster, &'a KeysManager,
+ &'a test_utils::TestFeeEstimator, &'a test_utils::TestLogger>
+ }
+
+ #[cfg(test)]
+ #[bench]
+ fn bench_sends(bench: &mut Bencher) {
+ bench_two_sends(bench, test_utils::TestPersister::new(), test_utils::TestPersister::new());
+ }
+
+ pub fn bench_two_sends<P: Persist<InMemorySigner>>(bench: &mut Bencher, persister_a: P, persister_b: P) {
+ // Do a simple benchmark of sending a payment back and forth between two nodes.
+ // Note that this is unrealistic as each payment send will require at least two fsync
+ // calls per node.
+ let network = bitcoin::Network::Testnet;
+ let genesis_hash = bitcoin::blockdata::constants::genesis_block(network).header.block_hash();
+
+ let tx_broadcaster = test_utils::TestBroadcaster{txn_broadcasted: Mutex::new(Vec::new())};
+ let fee_estimator = test_utils::TestFeeEstimator { sat_per_kw: 253 };
+
+ let mut config: UserConfig = Default::default();
+ config.own_channel_config.minimum_depth = 1;
+
+ let logger_a = test_utils::TestLogger::with_id("node a".to_owned());
+ let chain_monitor_a = ChainMonitor::new(None, &tx_broadcaster, &logger_a, &fee_estimator, &persister_a);
+ let seed_a = [1u8; 32];
+ let keys_manager_a = KeysManager::new(&seed_a, 42, 42);
+ let node_a = ChannelManager::new(&fee_estimator, &chain_monitor_a, &tx_broadcaster, &logger_a, &keys_manager_a, config.clone(), ChainParameters {
+ network,
+ latest_hash: genesis_hash,
+ latest_height: 0,
+ });
+ let node_a_holder = NodeHolder { node: &node_a };
+
+ let logger_b = test_utils::TestLogger::with_id("node a".to_owned());
+ let chain_monitor_b = ChainMonitor::new(None, &tx_broadcaster, &logger_a, &fee_estimator, &persister_b);
+ let seed_b = [2u8; 32];
+ let keys_manager_b = KeysManager::new(&seed_b, 42, 42);
+ let node_b = ChannelManager::new(&fee_estimator, &chain_monitor_b, &tx_broadcaster, &logger_b, &keys_manager_b, config.clone(), ChainParameters {
+ network,
+ latest_hash: genesis_hash,
+ latest_height: 0,
+ });
+ let node_b_holder = NodeHolder { node: &node_b };
+
+ node_a.create_channel(node_b.get_our_node_id(), 8_000_000, 100_000_000, 42, None).unwrap();
+ node_b.handle_open_channel(&node_a.get_our_node_id(), InitFeatures::known(), &get_event_msg!(node_a_holder, MessageSendEvent::SendOpenChannel, node_b.get_our_node_id()));
+ node_a.handle_accept_channel(&node_b.get_our_node_id(), InitFeatures::known(), &get_event_msg!(node_b_holder, MessageSendEvent::SendAcceptChannel, node_a.get_our_node_id()));
+
+ let tx;
+ if let Event::FundingGenerationReady { temporary_channel_id, output_script, .. } = get_event!(node_a_holder, Event::FundingGenerationReady) {
+ tx = Transaction { version: 2, lock_time: 0, input: Vec::new(), output: vec![TxOut {
+ value: 8_000_000, script_pubkey: output_script,
+ }]};
+ let funding_outpoint = OutPoint { txid: tx.txid(), index: 0 };
+ node_a.funding_transaction_generated(&temporary_channel_id, funding_outpoint);
+ } else { panic!(); }
+
+ node_b.handle_funding_created(&node_a.get_our_node_id(), &get_event_msg!(node_a_holder, MessageSendEvent::SendFundingCreated, node_b.get_our_node_id()));
+ node_a.handle_funding_signed(&node_b.get_our_node_id(), &get_event_msg!(node_b_holder, MessageSendEvent::SendFundingSigned, node_a.get_our_node_id()));
+
+ get_event!(node_a_holder, Event::FundingBroadcastSafe);
+
+ let block = Block {
+ header: BlockHeader { version: 0x20000000, prev_blockhash: genesis_hash, merkle_root: Default::default(), time: 42, bits: 42, nonce: 42 },
+ txdata: vec![tx],
+ };
+ Listen::block_connected(&node_a, &block, 1);
+ Listen::block_connected(&node_b, &block, 1);
+
+ node_a.handle_funding_locked(&node_b.get_our_node_id(), &get_event_msg!(node_b_holder, MessageSendEvent::SendFundingLocked, node_a.get_our_node_id()));
+ node_b.handle_funding_locked(&node_a.get_our_node_id(), &get_event_msg!(node_a_holder, MessageSendEvent::SendFundingLocked, node_b.get_our_node_id()));
+
+ let dummy_graph = NetworkGraph::new(genesis_hash);
+
+ macro_rules! send_payment {
+ ($node_a: expr, $node_b: expr) => {
+ let usable_channels = $node_a.list_usable_channels();
+ let route = get_route(&$node_a.get_our_node_id(), &dummy_graph, &$node_b.get_our_node_id(), None, Some(&usable_channels.iter().map(|r| r).collect::<Vec<_>>()), &[], 10_000, TEST_FINAL_CLTV, &logger_a).unwrap();
+
+ let payment_preimage = PaymentPreimage([0; 32]);
+ let payment_hash = PaymentHash(Sha256::hash(&payment_preimage.0[..]).into_inner());
+
+ $node_a.send_payment(&route, payment_hash, &None).unwrap();
+ let payment_event = SendEvent::from_event($node_a.get_and_clear_pending_msg_events().pop().unwrap());
+ $node_b.handle_update_add_htlc(&$node_a.get_our_node_id(), &payment_event.msgs[0]);
+ $node_b.handle_commitment_signed(&$node_a.get_our_node_id(), &payment_event.commitment_msg);
+ let (raa, cs) = get_revoke_commit_msgs!(NodeHolder { node: &$node_b }, $node_a.get_our_node_id());
+ $node_a.handle_revoke_and_ack(&$node_b.get_our_node_id(), &raa);
+ $node_a.handle_commitment_signed(&$node_b.get_our_node_id(), &cs);
+ $node_b.handle_revoke_and_ack(&$node_a.get_our_node_id(), &get_event_msg!(NodeHolder { node: &$node_a }, MessageSendEvent::SendRevokeAndACK, $node_b.get_our_node_id()));
+
+ expect_pending_htlcs_forwardable!(NodeHolder { node: &$node_b });
+ expect_payment_received!(NodeHolder { node: &$node_b }, payment_hash, 10_000);
+ assert!($node_b.claim_funds(payment_preimage, &None, 10_000));
+
+ match $node_b.get_and_clear_pending_msg_events().pop().unwrap() {
+ MessageSendEvent::UpdateHTLCs { node_id, updates } => {
+ assert_eq!(node_id, $node_a.get_our_node_id());
+ $node_a.handle_update_fulfill_htlc(&$node_b.get_our_node_id(), &updates.update_fulfill_htlcs[0]);
+ $node_a.handle_commitment_signed(&$node_b.get_our_node_id(), &updates.commitment_signed);
+ },
+ _ => panic!("Failed to generate claim event"),
+ }
+
+ let (raa, cs) = get_revoke_commit_msgs!(NodeHolder { node: &$node_a }, $node_b.get_our_node_id());
+ $node_b.handle_revoke_and_ack(&$node_a.get_our_node_id(), &raa);
+ $node_b.handle_commitment_signed(&$node_a.get_our_node_id(), &cs);
+ $node_a.handle_revoke_and_ack(&$node_b.get_our_node_id(), &get_event_msg!(NodeHolder { node: &$node_b }, MessageSendEvent::SendRevokeAndACK, $node_a.get_our_node_id()));
+
+ expect_payment_sent!(NodeHolder { node: &$node_a }, payment_preimage);
+ }
+ }
+
+ bench.iter(|| {
+ send_payment!(node_a, node_b);
+ send_payment!(node_b, node_a);
+ });
+ }
+}
}
}
+/// Get a specific event from the pending events queue.
+#[macro_export]
+macro_rules! get_event {
+ ($node: expr, $event_type: path) => {
+ {
+ let mut events = $node.node.get_and_clear_pending_events();
+ assert_eq!(events.len(), 1);
+ let ev = events.pop().unwrap();
+ match ev {
+ $event_type { .. } => {
+ ev
+ },
+ _ => panic!("Unexpected event"),
+ }
+ }
+ }
+}
+
#[cfg(test)]
macro_rules! get_htlc_update_msgs {
($node: expr, $node_id: expr) => {
}
}
-#[cfg(test)]
+/// Returns any local commitment transactions for the channel.
+#[macro_export]
macro_rules! get_local_commitment_txn {
($node: expr, $channel_id: expr) => {
{
}}
}
-#[cfg(test)]
+#[cfg(any(test, feature = "unstable"))]
macro_rules! expect_payment_received {
($node: expr, $expected_payment_hash: expr, $expected_recv_value: expr) => {
let events = $node.node.get_and_clear_pending_events();
//! you want to learn things about the network topology (eg get a route for sending a payment),
//! call into your NetGraphMsgHandler.
+#[cfg(any(test, feature = "_test_utils"))]
+#[macro_use]
+pub mod functional_test_utils;
+
pub mod channelmanager;
pub mod msgs;
pub mod peer_handler;
// without the node parameter being mut. This is incorrect, and thus newer rustcs will complain
// about an unnecessary mut. Thus, we silence the unused_mut warning in two test modules below.
-#[cfg(any(test, feature = "_test_utils"))]
-#[macro_use]
-pub mod functional_test_utils;
#[cfg(test)]
#[allow(unused_mut)]
mod functional_tests;
// licenses.
use chain;
+use chain::WatchedOutput;
use chain::chaininterface;
use chain::chaininterface::ConfirmationTarget;
use chain::chainmonitor;
use std::sync::{Mutex, Arc};
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::{cmp, mem};
-use std::collections::{HashMap, HashSet};
+use std::collections::{HashMap, HashSet, VecDeque};
use chain::keysinterface::InMemorySigner;
pub struct TestVecWriter(pub Vec<u8>);
*self.update_ret.lock().unwrap() = ret;
}
}
-impl channelmonitor::Persist<EnforcingSigner> for TestPersister {
- fn persist_new_channel(&self, _funding_txo: OutPoint, _data: &channelmonitor::ChannelMonitor<EnforcingSigner>) -> Result<(), channelmonitor::ChannelMonitorUpdateErr> {
+impl<Signer: keysinterface::Sign> channelmonitor::Persist<Signer> for TestPersister {
+ fn persist_new_channel(&self, _funding_txo: OutPoint, _data: &channelmonitor::ChannelMonitor<Signer>) -> Result<(), channelmonitor::ChannelMonitorUpdateErr> {
self.update_ret.lock().unwrap().clone()
}
- fn update_persisted_channel(&self, _funding_txo: OutPoint, _update: &channelmonitor::ChannelMonitorUpdate, _data: &channelmonitor::ChannelMonitor<EnforcingSigner>) -> Result<(), channelmonitor::ChannelMonitorUpdateErr> {
+ fn update_persisted_channel(&self, _funding_txo: OutPoint, _update: &channelmonitor::ChannelMonitorUpdate, _data: &channelmonitor::ChannelMonitor<Signer>) -> Result<(), channelmonitor::ChannelMonitorUpdateErr> {
self.update_ret.lock().unwrap().clone()
}
}
pub utxo_ret: Mutex<Result<TxOut, chain::AccessError>>,
pub watched_txn: Mutex<HashSet<(Txid, Script)>>,
pub watched_outputs: Mutex<HashSet<(OutPoint, Script)>>,
+ expectations: Mutex<Option<VecDeque<OnRegisterOutput>>>,
}
impl TestChainSource {
utxo_ret: Mutex::new(Ok(TxOut { value: u64::max_value(), script_pubkey })),
watched_txn: Mutex::new(HashSet::new()),
watched_outputs: Mutex::new(HashSet::new()),
+ expectations: Mutex::new(None),
}
}
+
+ /// Sets an expectation that [`chain::Filter::register_output`] is called.
+ pub fn expect(&self, expectation: OnRegisterOutput) -> &Self {
+ self.expectations.lock().unwrap()
+ .get_or_insert_with(|| VecDeque::new())
+ .push_back(expectation);
+ self
+ }
}
impl chain::Access for TestChainSource {
self.watched_txn.lock().unwrap().insert((*txid, script_pubkey.clone()));
}
- fn register_output(&self, outpoint: &OutPoint, script_pubkey: &Script) {
- self.watched_outputs.lock().unwrap().insert((*outpoint, script_pubkey.clone()));
+ fn register_output(&self, output: WatchedOutput) -> Option<(usize, Transaction)> {
+ let dependent_tx = match &mut *self.expectations.lock().unwrap() {
+ None => None,
+ Some(expectations) => match expectations.pop_front() {
+ None => {
+ panic!("Unexpected register_output: {:?}",
+ (output.outpoint, output.script_pubkey));
+ },
+ Some(expectation) => {
+ assert_eq!(output.outpoint, expectation.outpoint());
+ assert_eq!(&output.script_pubkey, expectation.script_pubkey());
+ expectation.returns
+ },
+ },
+ };
+
+ self.watched_outputs.lock().unwrap().insert((output.outpoint, output.script_pubkey));
+ dependent_tx
+ }
+}
+
+impl Drop for TestChainSource {
+ fn drop(&mut self) {
+ if std::thread::panicking() {
+ return;
+ }
+
+ if let Some(expectations) = &*self.expectations.lock().unwrap() {
+ if !expectations.is_empty() {
+ panic!("Unsatisfied expectations: {:?}", expectations);
+ }
+ }
+ }
+}
+
+/// An expectation that [`chain::Filter::register_output`] was called with a transaction output and
+/// returns an optional dependent transaction that spends the output in the same block.
+pub struct OnRegisterOutput {
+ /// The transaction output to register.
+ pub with: TxOutReference,
+
+ /// A dependent transaction spending the output along with its position in the block.
+ pub returns: Option<(usize, Transaction)>,
+}
+
+/// A transaction output as identified by an index into a transaction's output list.
+pub struct TxOutReference(pub Transaction, pub usize);
+
+impl OnRegisterOutput {
+ fn outpoint(&self) -> OutPoint {
+ let txid = self.with.0.txid();
+ let index = self.with.1 as u16;
+ OutPoint { txid, index }
+ }
+
+ fn script_pubkey(&self) -> &Script {
+ let index = self.with.1;
+ &self.with.0.output[index].script_pubkey
+ }
+}
+
+impl std::fmt::Debug for OnRegisterOutput {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ f.debug_struct("OnRegisterOutput")
+ .field("outpoint", &self.outpoint())
+ .field("script_pubkey", self.script_pubkey())
+ .finish()
}
}
projects / rust-lightning / commitdiff