From: Valentine Wallace Date: Tue, 15 Sep 2020 00:50:04 +0000 (-0400) Subject: Integrate Persist into ChainMonitor. X-Git-Tag: v0.0.12~10^2~4 X-Git-Url: http://git.bitcoin.ninja/?a=commitdiff_plain;h=9c3f3e76e58586dd0d985ea9f58193e4a757512b;p=rust-lightning Integrate Persist into ChainMonitor. - The ChainMonitor should: Whenever a new channel is added or updated, these updates should be conveyed to the persister and persisted to disk. Even if the update errors while it's being applied, the updated monitor still needs to be persisted. --- diff --git a/fuzz/src/chanmon_consistency.rs b/fuzz/src/chanmon_consistency.rs index 1650e2e25..8202ae3cd 100644 --- a/fuzz/src/chanmon_consistency.rs +++ b/fuzz/src/chanmon_consistency.rs @@ -48,6 +48,7 @@ use lightning::routing::router::{Route, RouteHop}; use utils::test_logger; +use utils::test_persister::TestPersister; use bitcoin::secp256k1::key::{PublicKey,SecretKey}; use bitcoin::secp256k1::Secp256k1; @@ -84,7 +85,7 @@ impl Writer for VecWriter { struct TestChainMonitor { pub logger: Arc, - pub chain_monitor: Arc, Arc, Arc, Arc>>, + pub chain_monitor: Arc, Arc, Arc, Arc, Arc>>, pub update_ret: Mutex>, // If we reload a node with an old copy of ChannelMonitors, the ChannelManager deserialization // logic will automatically force-close our channels for us (as we don't have an up-to-date @@ -95,9 +96,9 @@ struct TestChainMonitor { pub should_update_manager: atomic::AtomicBool, } impl TestChainMonitor { - pub fn new(broadcaster: Arc, logger: Arc, feeest: Arc) -> Self { + pub fn new(broadcaster: Arc, logger: Arc, feeest: Arc, persister: Arc) -> Self { Self { - chain_monitor: Arc::new(chainmonitor::ChainMonitor::new(None, broadcaster, logger.clone(), feeest)), + chain_monitor: Arc::new(chainmonitor::ChainMonitor::new(None, broadcaster, logger.clone(), feeest, persister)), logger, update_ret: Mutex::new(Ok(())), latest_monitors: Mutex::new(HashMap::new()), @@ -127,7 +128,7 @@ impl chain::Watch for TestChainMonitor { }; let mut deserialized_monitor = <(BlockHash, channelmonitor::ChannelMonitor)>:: read(&mut Cursor::new(&map_entry.get().1)).unwrap().1; - deserialized_monitor.update_monitor(update.clone(), &&TestBroadcaster {}, &self.logger).unwrap(); + deserialized_monitor.update_monitor(&update, &&TestBroadcaster {}, &self.logger).unwrap(); let mut ser = VecWriter(Vec::new()); deserialized_monitor.write_for_disk(&mut ser).unwrap(); map_entry.insert((update.update_id, ser.0)); @@ -192,7 +193,7 @@ pub fn do_test(data: &[u8], out: Out) { macro_rules! make_node { ($node_id: expr) => { { let logger: Arc = Arc::new(test_logger::TestLogger::new($node_id.to_string(), out.clone())); - let monitor = Arc::new(TestChainMonitor::new(broadcast.clone(), logger.clone(), fee_est.clone())); + let monitor = Arc::new(TestChainMonitor::new(broadcast.clone(), logger.clone(), fee_est.clone(), Arc::new(TestPersister{}))); let keys_manager = Arc::new(KeyProvider { node_id: $node_id, rand_bytes_id: atomic::AtomicU8::new(0) }); let mut config = UserConfig::default(); @@ -207,7 +208,7 @@ pub fn do_test(data: &[u8], out: Out) { macro_rules! reload_node { ($ser: expr, $node_id: expr, $old_monitors: expr) => { { let logger: Arc = Arc::new(test_logger::TestLogger::new($node_id.to_string(), out.clone())); - let chain_monitor = Arc::new(TestChainMonitor::new(broadcast.clone(), logger.clone(), fee_est.clone())); + let chain_monitor = Arc::new(TestChainMonitor::new(broadcast.clone(), logger.clone(), fee_est.clone(), Arc::new(TestPersister{}))); let keys_manager = Arc::new(KeyProvider { node_id: $node_id, rand_bytes_id: atomic::AtomicU8::new(0) }); let mut config = UserConfig::default(); diff --git a/fuzz/src/full_stack.rs b/fuzz/src/full_stack.rs index 1ed17b9ea..3aeb3d233 100644 --- a/fuzz/src/full_stack.rs +++ b/fuzz/src/full_stack.rs @@ -40,6 +40,7 @@ use lightning::util::logger::Logger; use lightning::util::config::UserConfig; use utils::test_logger; +use utils::test_persister::TestPersister; use bitcoin::secp256k1::key::{PublicKey,SecretKey}; use bitcoin::secp256k1::Secp256k1; @@ -145,13 +146,13 @@ impl<'a> std::hash::Hash for Peer<'a> { type ChannelMan = ChannelManager< EnforcingChannelKeys, - Arc, Arc, Arc, Arc>>, + Arc, Arc, Arc, Arc, Arc>>, Arc, Arc, Arc, Arc>; type PeerMan<'a> = PeerManager, Arc, Arc, Arc>>, Arc>; struct MoneyLossDetector<'a> { manager: Arc, - monitor: Arc, Arc, Arc, Arc>>, + monitor: Arc, Arc, Arc, Arc, Arc>>, handler: PeerMan<'a>, peers: &'a RefCell<[bool; 256]>, @@ -165,7 +166,7 @@ struct MoneyLossDetector<'a> { impl<'a> MoneyLossDetector<'a> { pub fn new(peers: &'a RefCell<[bool; 256]>, manager: Arc, - monitor: Arc, Arc, Arc, Arc>>, + monitor: Arc, Arc, Arc, Arc, Arc>>, handler: PeerMan<'a>) -> Self { MoneyLossDetector { manager, @@ -333,7 +334,7 @@ pub fn do_test(data: &[u8], logger: &Arc) { }; let broadcast = Arc::new(TestBroadcaster{}); - let monitor = Arc::new(chainmonitor::ChainMonitor::new(None, broadcast.clone(), Arc::clone(&logger), fee_est.clone())); + let monitor = Arc::new(chainmonitor::ChainMonitor::new(None, broadcast.clone(), Arc::clone(&logger), fee_est.clone(), Arc::new(TestPersister{}))); let keys_manager = Arc::new(KeyProvider { node_secret: our_network_key.clone(), counter: AtomicU64::new(0) }); let mut config = UserConfig::default(); diff --git a/fuzz/src/utils/mod.rs b/fuzz/src/utils/mod.rs index bb5b00a5b..937eee6b6 100644 --- a/fuzz/src/utils/mod.rs +++ b/fuzz/src/utils/mod.rs @@ -8,3 +8,4 @@ // licenses. pub mod test_logger; +pub mod test_persister; diff --git a/fuzz/src/utils/test_persister.rs b/fuzz/src/utils/test_persister.rs new file mode 100644 index 000000000..0bd60911e --- /dev/null +++ b/fuzz/src/utils/test_persister.rs @@ -0,0 +1,14 @@ +use lightning::chain::channelmonitor; +use lightning::chain::transaction::OutPoint; +use lightning::util::enforcing_trait_impls::EnforcingChannelKeys; + +pub struct TestPersister {} +impl channelmonitor::Persist for TestPersister { + fn persist_new_channel(&self, _funding_txo: OutPoint, _data: &channelmonitor::ChannelMonitor) -> Result<(), channelmonitor::ChannelMonitorUpdateErr> { + Ok(()) + } + + fn update_persisted_channel(&self, _funding_txo: OutPoint, _update: &channelmonitor::ChannelMonitorUpdate, _data: &channelmonitor::ChannelMonitor) -> Result<(), channelmonitor::ChannelMonitorUpdateErr> { + Ok(()) + } +} diff --git a/lightning-net-tokio/src/lib.rs b/lightning-net-tokio/src/lib.rs index e84ee7622..36384380f 100644 --- a/lightning-net-tokio/src/lib.rs +++ b/lightning-net-tokio/src/lib.rs @@ -36,7 +36,8 @@ //! type Logger = dyn lightning::util::logger::Logger; //! type ChainAccess = dyn lightning::chain::Access; //! type ChainFilter = dyn lightning::chain::Filter; -//! type ChainMonitor = lightning::chain::chainmonitor::ChainMonitor, Arc, Arc, Arc>; +//! type DataPersister = dyn lightning::chain::channelmonitor::Persist; +//! type ChainMonitor = lightning::chain::chainmonitor::ChainMonitor, Arc, Arc, Arc, Arc>; //! type ChannelManager = lightning::ln::channelmanager::SimpleArcChannelManager; //! type PeerManager = lightning::ln::peer_handler::SimpleArcPeerManager; //! diff --git a/lightning/src/chain/chainmonitor.rs b/lightning/src/chain/chainmonitor.rs index 6c67f54f1..469837f07 100644 --- a/lightning/src/chain/chainmonitor.rs +++ b/lightning/src/chain/chainmonitor.rs @@ -34,7 +34,8 @@ use bitcoin::blockdata::block::BlockHeader; use chain; use chain::Filter; use chain::chaininterface::{BroadcasterInterface, FeeEstimator}; -use chain::channelmonitor::{ChannelMonitor, ChannelMonitorUpdate, ChannelMonitorUpdateErr, MonitorEvent, MonitorUpdateError}; +use chain::channelmonitor; +use chain::channelmonitor::{ChannelMonitor, ChannelMonitorUpdate, ChannelMonitorUpdateErr, MonitorEvent, Persist}; use chain::transaction::{OutPoint, TransactionData}; use chain::keysinterface::ChannelKeys; use util::logger::Logger; @@ -55,25 +56,28 @@ use std::ops::Deref; /// [`chain::Watch`]: ../trait.Watch.html /// [`ChannelManager`]: ../../ln/channelmanager/struct.ChannelManager.html /// [module-level documentation]: index.html -pub struct ChainMonitor +pub struct ChainMonitor where C::Target: chain::Filter, T::Target: BroadcasterInterface, F::Target: FeeEstimator, L::Target: Logger, + P::Target: channelmonitor::Persist, { /// The monitors pub monitors: Mutex>>, chain_source: Option, broadcaster: T, logger: L, - fee_estimator: F + fee_estimator: F, + persister: P, } -impl ChainMonitor - where C::Target: chain::Filter, - T::Target: BroadcasterInterface, - F::Target: FeeEstimator, - L::Target: Logger, +impl ChainMonitor +where C::Target: chain::Filter, + T::Target: BroadcasterInterface, + F::Target: FeeEstimator, + L::Target: Logger, + P::Target: channelmonitor::Persist, { /// Dispatches to per-channel monitors, which are responsible for updating their on-chain view /// of a channel and reacting accordingly based on transactions in the connected block. See @@ -124,27 +128,47 @@ impl ChainMonit /// transactions relevant to the watched channels. /// /// [`chain::Filter`]: ../trait.Filter.html - pub fn new(chain_source: Option, broadcaster: T, logger: L, feeest: F) -> Self { + pub fn new(chain_source: Option, broadcaster: T, logger: L, feeest: F, persister: P) -> Self { Self { monitors: Mutex::new(HashMap::new()), chain_source, broadcaster, logger, fee_estimator: feeest, + persister, } } +} + +impl chain::Watch for ChainMonitor +where C::Target: chain::Filter, + T::Target: BroadcasterInterface, + F::Target: FeeEstimator, + L::Target: Logger, + P::Target: channelmonitor::Persist, +{ + type Keys = ChanSigner; /// Adds the monitor that watches the channel referred to by the given outpoint. /// /// Calls back to [`chain::Filter`] with the funding transaction and outputs to watch. /// + /// Note that we persist the given `ChannelMonitor` while holding the `ChainMonitor` + /// monitors lock. + /// /// [`chain::Filter`]: ../trait.Filter.html - fn add_monitor(&self, outpoint: OutPoint, monitor: ChannelMonitor) -> Result<(), MonitorUpdateError> { + fn watch_channel(&self, funding_outpoint: OutPoint, monitor: ChannelMonitor) -> Result<(), ChannelMonitorUpdateErr> { let mut monitors = self.monitors.lock().unwrap(); - let entry = match monitors.entry(outpoint) { - hash_map::Entry::Occupied(_) => return Err(MonitorUpdateError("Channel monitor for given outpoint is already present")), + let entry = match monitors.entry(funding_outpoint) { + hash_map::Entry::Occupied(_) => { + log_error!(self.logger, "Failed to add new channel data: channel monitor for given outpoint is already present"); + return Err(ChannelMonitorUpdateErr::PermanentFailure)}, hash_map::Entry::Vacant(e) => e, }; + if let Err(e) = self.persister.persist_new_channel(funding_outpoint, &monitor) { + log_error!(self.logger, "Failed to persist new channel data"); + return Err(e); + } { let funding_txo = monitor.get_funding_txo(); log_trace!(self.logger, "Got new Channel Monitor for channel {}", log_bytes!(funding_txo.0.to_channel_id()[..])); @@ -162,38 +186,34 @@ impl ChainMonit Ok(()) } - /// Updates the monitor that watches the channel referred to by the given outpoint. - fn update_monitor(&self, outpoint: OutPoint, update: ChannelMonitorUpdate) -> Result<(), MonitorUpdateError> { + /// Note that we persist the given `ChannelMonitor` update while holding the + /// `ChainMonitor` monitors lock. + fn update_channel(&self, funding_txo: OutPoint, update: ChannelMonitorUpdate) -> Result<(), ChannelMonitorUpdateErr> { + // Update the monitor that watches the channel referred to by the given outpoint. let mut monitors = self.monitors.lock().unwrap(); - match monitors.get_mut(&outpoint) { + match monitors.get_mut(&funding_txo) { + None => { + log_error!(self.logger, "Failed to update channel monitor: no such monitor registered"); + Err(ChannelMonitorUpdateErr::PermanentFailure) + }, Some(orig_monitor) => { log_trace!(self.logger, "Updating Channel Monitor for channel {}", log_funding_info!(orig_monitor)); - orig_monitor.update_monitor(update, &self.broadcaster, &self.logger) - }, - None => Err(MonitorUpdateError("No such monitor registered")) - } - } -} - -impl chain::Watch for ChainMonitor - where C::Target: chain::Filter, - T::Target: BroadcasterInterface, - F::Target: FeeEstimator, - L::Target: Logger, -{ - type Keys = ChanSigner; - - fn watch_channel(&self, funding_txo: OutPoint, monitor: ChannelMonitor) -> Result<(), ChannelMonitorUpdateErr> { - match self.add_monitor(funding_txo, monitor) { - Ok(_) => Ok(()), - Err(_) => Err(ChannelMonitorUpdateErr::PermanentFailure), - } - } - - fn update_channel(&self, funding_txo: OutPoint, update: ChannelMonitorUpdate) -> Result<(), ChannelMonitorUpdateErr> { - match self.update_monitor(funding_txo, update) { - Ok(_) => Ok(()), - Err(_) => Err(ChannelMonitorUpdateErr::PermanentFailure), + let update_res = orig_monitor.update_monitor(&update, &self.broadcaster, &self.logger); + if let Err(e) = &update_res { + log_error!(self.logger, "Failed to update channel monitor: {:?}", e); + } + // Even if updating the monitor returns an error, the monitor's state will + // still be changed. So, persist the updated monitor despite the error. + let persist_res = self.persister.update_persisted_channel(funding_txo, &update, orig_monitor); + if let Err(ref e) = persist_res { + log_error!(self.logger, "Failed to persist channel monitor update: {:?}", e); + } + if update_res.is_err() { + Err(ChannelMonitorUpdateErr::PermanentFailure) + } else { + persist_res + } + } } } @@ -206,11 +226,12 @@ impl events::EventsProvider for ChainMonitor +impl events::EventsProvider for ChainMonitor where C::Target: chain::Filter, T::Target: BroadcasterInterface, F::Target: FeeEstimator, L::Target: Logger, + P::Target: channelmonitor::Persist, { fn get_and_clear_pending_events(&self) -> Vec { let mut pending_events = Vec::new(); diff --git a/lightning/src/chain/channelmonitor.rs b/lightning/src/chain/channelmonitor.rs index 93a4edb6e..2503a7a76 100644 --- a/lightning/src/chain/channelmonitor.rs +++ b/lightning/src/chain/channelmonitor.rs @@ -95,7 +95,7 @@ impl Readable for ChannelMonitorUpdate { } /// An error enum representing a failure to persist a channel monitor update. -#[derive(Clone)] +#[derive(Clone, Debug)] pub enum ChannelMonitorUpdateErr { /// Used to indicate a temporary failure (eg connection to a watchtower or remote backup of /// our state failed, but is expected to succeed at some point in the future). @@ -159,7 +159,7 @@ pub enum ChannelMonitorUpdateErr { /// inconsistent with the ChannelMonitor being called. eg for ChannelMonitor::update_monitor this /// means you tried to update a monitor for a different channel or the ChannelMonitorUpdate was /// corrupted. -/// Contains a human-readable error message. +/// Contains a developer-readable error message. #[derive(Debug)] pub struct MonitorUpdateError(pub &'static str); @@ -1162,28 +1162,28 @@ impl ChannelMonitor { /// itself. /// /// panics if the given update is not the next update by update_id. - pub fn update_monitor(&mut self, mut updates: ChannelMonitorUpdate, broadcaster: &B, logger: &L) -> Result<(), MonitorUpdateError> + pub fn update_monitor(&mut self, updates: &ChannelMonitorUpdate, broadcaster: &B, logger: &L) -> Result<(), MonitorUpdateError> where B::Target: BroadcasterInterface, L::Target: Logger, { if self.latest_update_id + 1 != updates.update_id { panic!("Attempted to apply ChannelMonitorUpdates out of order, check the update_id before passing an update to update_monitor!"); } - for update in updates.updates.drain(..) { + for update in updates.updates.iter() { match update { ChannelMonitorUpdateStep::LatestHolderCommitmentTXInfo { commitment_tx, htlc_outputs } => { if self.lockdown_from_offchain { panic!(); } - self.provide_latest_holder_commitment_tx_info(commitment_tx, htlc_outputs)? + self.provide_latest_holder_commitment_tx_info(commitment_tx.clone(), htlc_outputs.clone())? }, ChannelMonitorUpdateStep::LatestCounterpartyCommitmentTXInfo { unsigned_commitment_tx, htlc_outputs, commitment_number, their_revocation_point } => - self.provide_latest_counterparty_commitment_tx_info(&unsigned_commitment_tx, htlc_outputs, commitment_number, their_revocation_point, logger), + self.provide_latest_counterparty_commitment_tx_info(&unsigned_commitment_tx, htlc_outputs.clone(), *commitment_number, *their_revocation_point, logger), ChannelMonitorUpdateStep::PaymentPreimage { payment_preimage } => self.provide_payment_preimage(&PaymentHash(Sha256::hash(&payment_preimage.0[..]).into_inner()), &payment_preimage), ChannelMonitorUpdateStep::CommitmentSecret { idx, secret } => - self.provide_secret(idx, secret)?, + self.provide_secret(*idx, *secret)?, ChannelMonitorUpdateStep::ChannelForceClosed { should_broadcast } => { self.lockdown_from_offchain = true; - if should_broadcast { + if *should_broadcast { self.broadcast_latest_holder_commitment_txn(broadcaster, logger); } else { log_error!(logger, "You have a toxic holder commitment transaction avaible in channel monitor, read comment in ChannelMonitor::get_latest_holder_commitment_txn to be informed of manual action to take"); diff --git a/lightning/src/ln/functional_test_utils.rs b/lightning/src/ln/functional_test_utils.rs index 871854ce6..0a7519ea9 100644 --- a/lightning/src/ln/functional_test_utils.rs +++ b/lightning/src/ln/functional_test_utils.rs @@ -94,6 +94,7 @@ pub struct TestChanMonCfg { pub tx_broadcaster: test_utils::TestBroadcaster, pub fee_estimator: test_utils::TestFeeEstimator, pub chain_source: test_utils::TestChainSource, + pub persister: test_utils::TestPersister, pub logger: test_utils::TestLogger, } @@ -197,8 +198,9 @@ impl<'a, 'b, 'c> Drop for Node<'a, 'b, 'c> { }).unwrap(); } + let persister = test_utils::TestPersister{}; let chain_source = test_utils::TestChainSource::new(Network::Testnet); - let chain_monitor = test_utils::TestChainMonitor::new(Some(&chain_source), self.tx_broadcaster.clone(), &self.logger, &feeest); + let chain_monitor = test_utils::TestChainMonitor::new(Some(&chain_source), self.tx_broadcaster.clone(), &self.logger, &feeest, &persister); for deserialized_monitor in deserialized_monitors.drain(..) { if let Err(_) = chain_monitor.watch_channel(deserialized_monitor.get_funding_txo().0, deserialized_monitor) { panic!(); @@ -1115,7 +1117,8 @@ pub fn create_chanmon_cfgs(node_count: usize) -> Vec { let fee_estimator = test_utils::TestFeeEstimator { sat_per_kw: 253 }; let chain_source = test_utils::TestChainSource::new(Network::Testnet); let logger = test_utils::TestLogger::with_id(format!("node {}", i)); - chan_mon_cfgs.push(TestChanMonCfg{ tx_broadcaster, fee_estimator, chain_source, logger }); + let persister = test_utils::TestPersister{}; + chan_mon_cfgs.push(TestChanMonCfg{ tx_broadcaster, fee_estimator, chain_source, logger, persister }); } chan_mon_cfgs @@ -1127,7 +1130,7 @@ pub fn create_node_cfgs<'a>(node_count: usize, chanmon_cfgs: &'a Vec; @@ -4322,7 +4323,8 @@ fn test_no_txn_manager_serialize_deserialize() { logger = test_utils::TestLogger::new(); fee_estimator = test_utils::TestFeeEstimator { sat_per_kw: 253 }; - new_chain_monitor = test_utils::TestChainMonitor::new(Some(nodes[0].chain_source), nodes[0].tx_broadcaster.clone(), &logger, &fee_estimator); + persister = test_utils::TestPersister{}; + new_chain_monitor = test_utils::TestChainMonitor::new(Some(nodes[0].chain_source), nodes[0].tx_broadcaster.clone(), &logger, &fee_estimator, &persister); nodes[0].chain_monitor = &new_chain_monitor; let mut chan_0_monitor_read = &chan_0_monitor_serialized.0[..]; let (_, mut chan_0_monitor) = <(BlockHash, ChannelMonitor)>::read(&mut chan_0_monitor_read).unwrap(); @@ -4380,6 +4382,7 @@ fn test_manager_serialize_deserialize_events() { let node_cfgs = create_node_cfgs(2, &chanmon_cfgs); let node_chanmgrs = create_node_chanmgrs(2, &node_cfgs, &[None, None]); let fee_estimator: test_utils::TestFeeEstimator; + let persister: test_utils::TestPersister; let logger: test_utils::TestLogger; let new_chain_monitor: test_utils::TestChainMonitor; let keys_manager: test_utils::TestKeysInterface; @@ -4429,7 +4432,8 @@ fn test_manager_serialize_deserialize_events() { fee_estimator = test_utils::TestFeeEstimator { sat_per_kw: 253 }; logger = test_utils::TestLogger::new(); - new_chain_monitor = test_utils::TestChainMonitor::new(Some(nodes[0].chain_source), nodes[0].tx_broadcaster.clone(), &logger, &fee_estimator); + persister = test_utils::TestPersister{}; + new_chain_monitor = test_utils::TestChainMonitor::new(Some(nodes[0].chain_source), nodes[0].tx_broadcaster.clone(), &logger, &fee_estimator, &persister); nodes[0].chain_monitor = &new_chain_monitor; let mut chan_0_monitor_read = &chan_0_monitor_serialized.0[..]; let (_, mut chan_0_monitor) = <(BlockHash, ChannelMonitor)>::read(&mut chan_0_monitor_read).unwrap(); @@ -4502,6 +4506,7 @@ fn test_simple_manager_serialize_deserialize() { let node_chanmgrs = create_node_chanmgrs(2, &node_cfgs, &[None, None]); let logger: test_utils::TestLogger; let fee_estimator: test_utils::TestFeeEstimator; + let persister: test_utils::TestPersister; let new_chain_monitor: test_utils::TestChainMonitor; let keys_manager: test_utils::TestKeysInterface; let nodes_0_deserialized: ChannelManager; @@ -4519,7 +4524,8 @@ fn test_simple_manager_serialize_deserialize() { logger = test_utils::TestLogger::new(); fee_estimator = test_utils::TestFeeEstimator { sat_per_kw: 253 }; - new_chain_monitor = test_utils::TestChainMonitor::new(Some(nodes[0].chain_source), nodes[0].tx_broadcaster.clone(), &logger, &fee_estimator); + persister = test_utils::TestPersister{}; + new_chain_monitor = test_utils::TestChainMonitor::new(Some(nodes[0].chain_source), nodes[0].tx_broadcaster.clone(), &logger, &fee_estimator, &persister); nodes[0].chain_monitor = &new_chain_monitor; let mut chan_0_monitor_read = &chan_0_monitor_serialized.0[..]; let (_, mut chan_0_monitor) = <(BlockHash, ChannelMonitor)>::read(&mut chan_0_monitor_read).unwrap(); @@ -4561,6 +4567,7 @@ fn test_manager_serialize_deserialize_inconsistent_monitor() { let node_chanmgrs = create_node_chanmgrs(4, &node_cfgs, &[None, None, None, None]); let logger: test_utils::TestLogger; let fee_estimator: test_utils::TestFeeEstimator; + let persister: test_utils::TestPersister; let new_chain_monitor: test_utils::TestChainMonitor; let keys_manager: test_utils::TestKeysInterface; let nodes_0_deserialized: ChannelManager; @@ -4597,7 +4604,8 @@ fn test_manager_serialize_deserialize_inconsistent_monitor() { logger = test_utils::TestLogger::new(); fee_estimator = test_utils::TestFeeEstimator { sat_per_kw: 253 }; - new_chain_monitor = test_utils::TestChainMonitor::new(Some(nodes[0].chain_source), nodes[0].tx_broadcaster.clone(), &logger, &fee_estimator); + persister = test_utils::TestPersister{}; + new_chain_monitor = test_utils::TestChainMonitor::new(Some(nodes[0].chain_source), nodes[0].tx_broadcaster.clone(), &logger, &fee_estimator, &persister); nodes[0].chain_monitor = &new_chain_monitor; let mut node_0_stale_monitors = Vec::new(); @@ -5742,7 +5750,7 @@ fn test_key_derivation_params() { // We manually create the node configuration to backup the seed. let seed = [42; 32]; let keys_manager = test_utils::TestKeysInterface::new(&seed, Network::Testnet); - let chain_monitor = test_utils::TestChainMonitor::new(Some(&chanmon_cfgs[0].chain_source), &chanmon_cfgs[0].tx_broadcaster, &chanmon_cfgs[0].logger, &chanmon_cfgs[0].fee_estimator); + let chain_monitor = test_utils::TestChainMonitor::new(Some(&chanmon_cfgs[0].chain_source), &chanmon_cfgs[0].tx_broadcaster, &chanmon_cfgs[0].logger, &chanmon_cfgs[0].fee_estimator, &chanmon_cfgs[0].persister); let node = NodeCfg { chain_source: &chanmon_cfgs[0].chain_source, logger: &chanmon_cfgs[0].logger, tx_broadcaster: &chanmon_cfgs[0].tx_broadcaster, fee_estimator: &chanmon_cfgs[0].fee_estimator, chain_monitor, keys_manager, node_seed: seed }; let mut node_cfgs = create_node_cfgs(3, &chanmon_cfgs); node_cfgs.remove(0); @@ -7407,6 +7415,7 @@ fn test_data_loss_protect() { // * we close channel in case of detecting other being fallen behind // * we are able to claim our own outputs thanks to to_remote being static let keys_manager; + let persister; let logger; let fee_estimator; let tx_broadcaster; @@ -7438,7 +7447,8 @@ fn test_data_loss_protect() { tx_broadcaster = test_utils::TestBroadcaster{txn_broadcasted: Mutex::new(Vec::new())}; fee_estimator = test_utils::TestFeeEstimator { sat_per_kw: 253 }; keys_manager = test_utils::TestKeysInterface::new(&nodes[0].node_seed, Network::Testnet); - monitor = test_utils::TestChainMonitor::new(Some(&chain_source), &tx_broadcaster, &logger, &fee_estimator); + persister = test_utils::TestPersister{}; + monitor = test_utils::TestChainMonitor::new(Some(&chain_source), &tx_broadcaster, &logger, &fee_estimator, &persister); node_state_0 = { let mut channel_monitors = HashMap::new(); channel_monitors.insert(OutPoint { txid: chan.3.txid(), index: 0 }, &mut chain_monitor); @@ -8299,6 +8309,7 @@ fn test_update_err_monitor_lockdown() { // Copy ChainMonitor to simulate a watchtower and update block height of node 0 until its ChannelMonitor timeout HTLC onchain let chain_source = test_utils::TestChainSource::new(Network::Testnet); let logger = test_utils::TestLogger::with_id(format!("node {}", 0)); + let persister = test_utils::TestPersister::new(); let watchtower = { let monitors = nodes[0].chain_monitor.chain_monitor.monitors.lock().unwrap(); let monitor = monitors.get(&outpoint).unwrap(); @@ -8307,7 +8318,7 @@ fn test_update_err_monitor_lockdown() { let new_monitor = <(BlockHash, channelmonitor::ChannelMonitor)>::read( &mut ::std::io::Cursor::new(&w.0)).unwrap().1; assert!(new_monitor == *monitor); - let watchtower = test_utils::TestChainMonitor::new(Some(&chain_source), &chanmon_cfgs[0].tx_broadcaster, &logger, &chanmon_cfgs[0].fee_estimator); + let watchtower = test_utils::TestChainMonitor::new(Some(&chain_source), &chanmon_cfgs[0].tx_broadcaster, &logger, &chanmon_cfgs[0].fee_estimator, &persister); assert!(watchtower.watch_channel(outpoint, new_monitor).is_ok()); watchtower }; @@ -8357,6 +8368,7 @@ fn test_concurrent_monitor_claim() { // Copy ChainMonitor to simulate watchtower Alice and update block height her ChannelMonitor timeout HTLC onchain let chain_source = test_utils::TestChainSource::new(Network::Testnet); let logger = test_utils::TestLogger::with_id(format!("node {}", "Alice")); + let persister = test_utils::TestPersister::new(); let watchtower_alice = { let monitors = nodes[0].chain_monitor.chain_monitor.monitors.lock().unwrap(); let monitor = monitors.get(&outpoint).unwrap(); @@ -8365,7 +8377,7 @@ fn test_concurrent_monitor_claim() { let new_monitor = <(BlockHash, channelmonitor::ChannelMonitor)>::read( &mut ::std::io::Cursor::new(&w.0)).unwrap().1; assert!(new_monitor == *monitor); - let watchtower = test_utils::TestChainMonitor::new(Some(&chain_source), &chanmon_cfgs[0].tx_broadcaster, &logger, &chanmon_cfgs[0].fee_estimator); + let watchtower = test_utils::TestChainMonitor::new(Some(&chain_source), &chanmon_cfgs[0].tx_broadcaster, &logger, &chanmon_cfgs[0].fee_estimator, &persister); assert!(watchtower.watch_channel(outpoint, new_monitor).is_ok()); watchtower }; @@ -8382,6 +8394,7 @@ fn test_concurrent_monitor_claim() { // Copy ChainMonitor to simulate watchtower Bob and make it receive a commitment update first. let chain_source = test_utils::TestChainSource::new(Network::Testnet); let logger = test_utils::TestLogger::with_id(format!("node {}", "Bob")); + let persister = test_utils::TestPersister::new(); let watchtower_bob = { let monitors = nodes[0].chain_monitor.chain_monitor.monitors.lock().unwrap(); let monitor = monitors.get(&outpoint).unwrap(); @@ -8390,7 +8403,7 @@ fn test_concurrent_monitor_claim() { let new_monitor = <(BlockHash, channelmonitor::ChannelMonitor)>::read( &mut ::std::io::Cursor::new(&w.0)).unwrap().1; assert!(new_monitor == *monitor); - let watchtower = test_utils::TestChainMonitor::new(Some(&chain_source), &chanmon_cfgs[0].tx_broadcaster, &logger, &chanmon_cfgs[0].fee_estimator); + let watchtower = test_utils::TestChainMonitor::new(Some(&chain_source), &chanmon_cfgs[0].tx_broadcaster, &logger, &chanmon_cfgs[0].fee_estimator, &persister); assert!(watchtower.watch_channel(outpoint, new_monitor).is_ok()); watchtower }; diff --git a/lightning/src/util/test_utils.rs b/lightning/src/util/test_utils.rs index 0370c0e1a..8643ef0f2 100644 --- a/lightning/src/util/test_utils.rs +++ b/lightning/src/util/test_utils.rs @@ -63,18 +63,18 @@ impl chaininterface::FeeEstimator for TestFeeEstimator { pub struct TestChainMonitor<'a> { pub added_monitors: Mutex)>>, pub latest_monitor_update_id: Mutex>, - pub chain_monitor: chainmonitor::ChainMonitor, + pub chain_monitor: chainmonitor::ChainMonitor>, pub update_ret: Mutex>, // If this is set to Some(), after the next return, we'll always return this until update_ret // is changed: pub next_update_ret: Mutex>>, } impl<'a> TestChainMonitor<'a> { - pub fn new(chain_source: Option<&'a TestChainSource>, broadcaster: &'a chaininterface::BroadcasterInterface, logger: &'a TestLogger, fee_estimator: &'a TestFeeEstimator) -> Self { + pub fn new(chain_source: Option<&'a TestChainSource>, broadcaster: &'a chaininterface::BroadcasterInterface, logger: &'a TestLogger, fee_estimator: &'a TestFeeEstimator, persister: &'a channelmonitor::Persist) -> Self { Self { added_monitors: Mutex::new(Vec::new()), latest_monitor_update_id: Mutex::new(HashMap::new()), - chain_monitor: chainmonitor::ChainMonitor::new(chain_source, broadcaster, logger, fee_estimator), + chain_monitor: chainmonitor::ChainMonitor::new(chain_source, broadcaster, logger, fee_estimator, persister), update_ret: Mutex::new(Ok(())), next_update_ret: Mutex::new(None), } @@ -134,6 +134,18 @@ impl<'a> chain::Watch for TestChainMonitor<'a> { } } +pub struct TestPersister {} + +impl channelmonitor::Persist for TestPersister { + fn persist_new_channel(&self, _funding_txo: OutPoint, _data: &channelmonitor::ChannelMonitor) -> Result<(), channelmonitor::ChannelMonitorUpdateErr> { + Ok(()) + } + + fn update_persisted_channel(&self, _funding_txo: OutPoint, _update: &channelmonitor::ChannelMonitorUpdate, _data: &channelmonitor::ChannelMonitor) -> Result<(), channelmonitor::ChannelMonitorUpdateErr> { + Ok(()) + } +} + pub struct TestBroadcaster { pub txn_broadcasted: Mutex>, }