X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=lightning%2Fsrc%2Fchain%2Fchainmonitor.rs;h=cf51dbab72019d9eb87748e0817171c1c0abfb35;hb=efbaa19a054ca1bbe1832c080154b72baf65b480;hp=e7c2b0f18ec001d703eb0249964b90a8a3f1a5ed;hpb=5f96d1334435e74545670a5dff24078edf749d60;p=rust-lightning diff --git a/lightning/src/chain/chainmonitor.rs b/lightning/src/chain/chainmonitor.rs index e7c2b0f1..cf51dbab 100644 --- a/lightning/src/chain/chainmonitor.rs +++ b/lightning/src/chain/chainmonitor.rs @@ -31,7 +31,7 @@ use crate::chain::{ChannelMonitorUpdateStatus, Filter, WatchedOutput}; use crate::chain::chaininterface::{BroadcasterInterface, FeeEstimator}; use crate::chain::channelmonitor::{ChannelMonitor, ChannelMonitorUpdate, Balance, MonitorEvent, TransactionOutputs, LATENCY_GRACE_PERIOD_BLOCKS}; use crate::chain::transaction::{OutPoint, TransactionData}; -use crate::chain::keysinterface::WriteableEcdsaChannelSigner; +use crate::sign::WriteableEcdsaChannelSigner; use crate::events; use crate::events::{Event, EventHandler}; use crate::util::atomic_counter::AtomicCounter; @@ -42,11 +42,12 @@ use crate::ln::channelmanager::ChannelDetails; use crate::prelude::*; use crate::sync::{RwLock, RwLockReadGuard, Mutex, MutexGuard}; +use core::iter::FromIterator; use core::ops::Deref; -use core::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; +use core::sync::atomic::{AtomicUsize, Ordering}; use bitcoin::secp256k1::PublicKey; -#[derive(Clone, Copy, Hash, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, Hash, PartialEq, Eq)] /// A specific update's ID stored in a `MonitorUpdateId`, separated out to make the contents /// entirely opaque. enum UpdateOrigin { @@ -60,7 +61,7 @@ enum UpdateOrigin { } /// An opaque identifier describing a specific [`Persist`] method call. -#[derive(Clone, Copy, Hash, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, Hash, PartialEq, Eq)] pub struct MonitorUpdateId { contents: UpdateOrigin, } @@ -77,23 +78,58 @@ impl MonitorUpdateId { /// `Persist` defines behavior for persisting channel monitors: this could mean /// writing once to disk, and/or uploading to one or more backup services. /// -/// Each method can return three possible values: -/// * If persistence (including any relevant `fsync()` calls) happens immediately, the -/// implementation should return [`ChannelMonitorUpdateStatus::Completed`], indicating normal -/// channel operation should continue. -/// * If persistence happens asynchronously, implementations should first ensure the -/// [`ChannelMonitor`] or [`ChannelMonitorUpdate`] are written durably to disk, and then return -/// [`ChannelMonitorUpdateStatus::InProgress`] while the update continues in the background. -/// Once the update completes, [`ChainMonitor::channel_monitor_updated`] should be called with -/// the corresponding [`MonitorUpdateId`]. +/// Persistence can happen in one of two ways - synchronously completing before the trait method +/// calls return or asynchronously in the background. /// -/// Note that unlike the direct [`chain::Watch`] interface, -/// [`ChainMonitor::channel_monitor_updated`] must be called once for *each* update which occurs. +/// # For those implementing synchronous persistence /// -/// * If persistence fails for some reason, implementations should return -/// [`ChannelMonitorUpdateStatus::PermanentFailure`], in which case the channel will likely be -/// closed without broadcasting the latest state. See -/// [`ChannelMonitorUpdateStatus::PermanentFailure`] for more details. +/// * If persistence completes fully (including any relevant `fsync()` calls), the implementation +/// should return [`ChannelMonitorUpdateStatus::Completed`], indicating normal channel operation +/// should continue. +/// +/// * If persistence fails for some reason, implementations should consider returning +/// [`ChannelMonitorUpdateStatus::InProgress`] and retry all pending persistence operations in +/// the background with [`ChainMonitor::list_pending_monitor_updates`] and +/// [`ChainMonitor::get_monitor`]. +/// +/// Once a full [`ChannelMonitor`] has been persisted, all pending updates for that channel can +/// be marked as complete via [`ChainMonitor::channel_monitor_updated`]. +/// +/// If at some point no further progress can be made towards persisting the pending updates, the +/// node should simply shut down. +/// +/// * If the persistence has failed and cannot be retried further (e.g. because of an outage), +/// [`ChannelMonitorUpdateStatus::UnrecoverableError`] can be used, though this will result in +/// an immediate panic and future operations in LDK generally failing. +/// +/// # For those implementing asynchronous persistence +/// +/// All calls should generally spawn a background task and immediately return +/// [`ChannelMonitorUpdateStatus::InProgress`]. Once the update completes, +/// [`ChainMonitor::channel_monitor_updated`] should be called with the corresponding +/// [`MonitorUpdateId`]. +/// +/// Note that unlike the direct [`chain::Watch`] interface, +/// [`ChainMonitor::channel_monitor_updated`] must be called once for *each* update which occurs. +/// +/// If at some point no further progress can be made towards persisting a pending update, the node +/// should simply shut down. Until then, the background task should either loop indefinitely, or +/// persistence should be regularly retried with [`ChainMonitor::list_pending_monitor_updates`] +/// and [`ChainMonitor::get_monitor`] (note that if a full monitor is persisted all pending +/// monitor updates may be marked completed). +/// +/// # Using remote watchtowers +/// +/// Watchtowers may be updated as a part of an implementation of this trait, utilizing the async +/// update process described above while the watchtower is being updated. The following methods are +/// provided for bulding transactions for a watchtower: +/// [`ChannelMonitor::initial_counterparty_commitment_tx`], +/// [`ChannelMonitor::counterparty_commitment_txs_from_update`], +/// [`ChannelMonitor::sign_to_local_justice_tx`], [`TrustedCommitmentTransaction::revokeable_output_index`], +/// [`TrustedCommitmentTransaction::build_to_local_justice_tx`]. +/// +/// [`TrustedCommitmentTransaction::revokeable_output_index`]: crate::ln::chan_utils::TrustedCommitmentTransaction::revokeable_output_index +/// [`TrustedCommitmentTransaction::build_to_local_justice_tx`]: crate::ln::chan_utils::TrustedCommitmentTransaction::build_to_local_justice_tx pub trait Persist { /// Persist a new channel's data in response to a [`chain::Watch::watch_channel`] call. This is /// called by [`ChannelManager`] for new channels, or may be called directly, e.g. on startup. @@ -168,12 +204,6 @@ struct MonitorHolder { /// the ChannelManager re-adding the same payment entry, before the same block is replayed, /// resulting in a duplicate PaymentSent event. pending_monitor_updates: Mutex>, - /// When the user returns a PermanentFailure error from an update_persisted_channel call during - /// block processing, we inform the ChannelManager that the channel should be closed - /// asynchronously. In order to ensure no further changes happen before the ChannelManager has - /// processed the closure event, we set this to true and return PermanentFailure for any other - /// chain::Watch events. - channel_perm_failed: AtomicBool, /// The last block height at which no [`UpdateOrigin::ChainSync`] monitor updates were present /// in `pending_monitor_updates`. /// If it's been more than [`LATENCY_GRACE_PERIOD_BLOCKS`] since we started waiting on a chain @@ -274,7 +304,34 @@ where C::Target: chain::Filter, where FN: Fn(&ChannelMonitor, &TransactionData) -> Vec { + let err_str = "ChannelMonitor[Update] persistence failed unrecoverably. This indicates we cannot continue normal operation and must shut down."; + let funding_outpoints: HashSet = HashSet::from_iter(self.monitors.read().unwrap().keys().cloned()); + for funding_outpoint in funding_outpoints.iter() { + let monitor_lock = self.monitors.read().unwrap(); + if let Some(monitor_state) = monitor_lock.get(funding_outpoint) { + if self.update_monitor_with_chain_data(header, best_height, txdata, &process, funding_outpoint, &monitor_state).is_err() { + // Take the monitors lock for writing so that we poison it and any future + // operations going forward fail immediately. + core::mem::drop(monitor_state); + core::mem::drop(monitor_lock); + let _poison = self.monitors.write().unwrap(); + log_error!(self.logger, "{}", err_str); + panic!("{}", err_str); + } + } + } + + // do some followup cleanup if any funding outpoints were added in between iterations let monitor_states = self.monitors.write().unwrap(); + for (funding_outpoint, monitor_state) in monitor_states.iter() { + if !funding_outpoints.contains(funding_outpoint) { + if self.update_monitor_with_chain_data(header, best_height, txdata, &process, funding_outpoint, &monitor_state).is_err() { + log_error!(self.logger, "{}", err_str); + panic!("{}", err_str); + } + } + } + if let Some(height) = best_height { // If the best block height is being updated, update highest_chain_height under the // monitors write lock. @@ -284,58 +341,60 @@ where C::Target: chain::Filter, self.highest_chain_height.store(new_height, Ordering::Release); } } + } - for (funding_outpoint, monitor_state) in monitor_states.iter() { - let monitor = &monitor_state.monitor; - let mut txn_outputs; - { - txn_outputs = process(monitor, txdata); - let update_id = MonitorUpdateId { - contents: UpdateOrigin::ChainSync(self.sync_persistence_id.get_increment()), - }; - let mut pending_monitor_updates = monitor_state.pending_monitor_updates.lock().unwrap(); - if let Some(height) = best_height { - if !monitor_state.has_pending_chainsync_updates(&pending_monitor_updates) { - // If there are not ChainSync persists awaiting completion, go ahead and - // set last_chain_persist_height here - we wouldn't want the first - // InProgress to always immediately be considered "overly delayed". - monitor_state.last_chain_persist_height.store(height as usize, Ordering::Release); - } + fn update_monitor_with_chain_data( + &self, header: &BlockHeader, best_height: Option, txdata: &TransactionData, + process: FN, funding_outpoint: &OutPoint, monitor_state: &MonitorHolder + ) -> Result<(), ()> where FN: Fn(&ChannelMonitor, &TransactionData) -> Vec { + let monitor = &monitor_state.monitor; + let mut txn_outputs; + { + txn_outputs = process(monitor, txdata); + let update_id = MonitorUpdateId { + contents: UpdateOrigin::ChainSync(self.sync_persistence_id.get_increment()), + }; + let mut pending_monitor_updates = monitor_state.pending_monitor_updates.lock().unwrap(); + if let Some(height) = best_height { + if !monitor_state.has_pending_chainsync_updates(&pending_monitor_updates) { + // If there are not ChainSync persists awaiting completion, go ahead and + // set last_chain_persist_height here - we wouldn't want the first + // InProgress to always immediately be considered "overly delayed". + monitor_state.last_chain_persist_height.store(height as usize, Ordering::Release); } + } - log_trace!(self.logger, "Syncing Channel Monitor for channel {}", log_funding_info!(monitor)); - match self.persister.update_persisted_channel(*funding_outpoint, None, monitor, update_id) { - ChannelMonitorUpdateStatus::Completed => - log_trace!(self.logger, "Finished syncing Channel Monitor for channel {}", log_funding_info!(monitor)), - ChannelMonitorUpdateStatus::PermanentFailure => { - monitor_state.channel_perm_failed.store(true, Ordering::Release); - self.pending_monitor_events.lock().unwrap().push((*funding_outpoint, vec![MonitorEvent::UpdateFailed(*funding_outpoint)], monitor.get_counterparty_node_id())); - self.event_notifier.notify(); - }, - ChannelMonitorUpdateStatus::InProgress => { - log_debug!(self.logger, "Channel Monitor sync for channel {} in progress, holding events until completion!", log_funding_info!(monitor)); - pending_monitor_updates.push(update_id); - }, - } + log_trace!(self.logger, "Syncing Channel Monitor for channel {}", log_funding_info!(monitor)); + match self.persister.update_persisted_channel(*funding_outpoint, None, monitor, update_id) { + ChannelMonitorUpdateStatus::Completed => + log_trace!(self.logger, "Finished syncing Channel Monitor for channel {}", log_funding_info!(monitor)), + ChannelMonitorUpdateStatus::InProgress => { + log_debug!(self.logger, "Channel Monitor sync for channel {} in progress, holding events until completion!", log_funding_info!(monitor)); + pending_monitor_updates.push(update_id); + }, + ChannelMonitorUpdateStatus::UnrecoverableError => { + return Err(()); + }, } + } - // 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, mut outputs) in txn_outputs.drain(..) { - for (idx, output) in outputs.drain(..) { - // Register any new outputs with the chain source for filtering - let output = WatchedOutput { - block_hash: Some(block_hash), - outpoint: OutPoint { txid, index: idx as u16 }, - script_pubkey: output.script_pubkey, - }; - chain_source.register_output(output) - } + // 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, mut outputs) in txn_outputs.drain(..) { + for (idx, output) in outputs.drain(..) { + // Register any new outputs with the chain source for filtering + let output = WatchedOutput { + block_hash: Some(block_hash), + outpoint: OutPoint { txid, index: idx as u16 }, + script_pubkey: output.script_pubkey, + }; + chain_source.register_output(output) } } } + Ok(()) } /// Creates a new `ChainMonitor` used to watch on-chain activity pertaining to channels. @@ -465,9 +524,8 @@ where C::Target: chain::Filter, // `MonitorEvent`s from the monitor back to the `ChannelManager` until they // complete. let monitor_is_pending_updates = monitor_data.has_pending_offchain_updates(&pending_monitor_updates); - if monitor_is_pending_updates || monitor_data.channel_perm_failed.load(Ordering::Acquire) { - // If there are still monitor updates pending (or an old monitor update - // finished after a later one perm-failed), we cannot yet construct an + if monitor_is_pending_updates { + // If there are still monitor updates pending, we cannot yet construct a // Completed event. return Ok(()); } @@ -502,7 +560,7 @@ where C::Target: chain::Filter, self.event_notifier.notify(); } - #[cfg(any(test, fuzzing, feature = "_test_utils"))] + #[cfg(any(test, feature = "_test_utils"))] pub fn get_and_clear_pending_events(&self) -> Vec { use crate::events::EventsProvider; let events = core::cell::RefCell::new(Vec::new()); @@ -520,12 +578,13 @@ where C::Target: chain::Filter, pub async fn process_pending_events_async Future>( &self, handler: H ) { - let mut pending_events = Vec::new(); - for monitor_state in self.monitors.read().unwrap().values() { - pending_events.append(&mut monitor_state.monitor.get_and_clear_pending_events()); - } - for event in pending_events { - handler(event).await; + // Sadly we can't hold the monitors read lock through an async call. Thus we have to do a + // crazy dance to process a monitor's events then only remove them once we've done so. + let mons_to_process = self.monitors.read().unwrap().keys().cloned().collect::>(); + for funding_txo in mons_to_process { + let mut ev; + super::channelmonitor::process_events_body!( + self.monitors.read().unwrap().get(&funding_txo).map(|m| &m.monitor), ev, handler(ev).await); } } @@ -640,18 +699,12 @@ where C::Target: chain::Filter, L::Target: Logger, P::Target: Persist, { - /// 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. - fn watch_channel(&self, funding_outpoint: OutPoint, monitor: ChannelMonitor) -> ChannelMonitorUpdateStatus { + fn watch_channel(&self, funding_outpoint: OutPoint, monitor: ChannelMonitor) -> Result { let mut monitors = self.monitors.write().unwrap(); 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 ChannelMonitorUpdateStatus::PermanentFailure + return Err(()); }, hash_map::Entry::Vacant(e) => e, }; @@ -664,13 +717,14 @@ where C::Target: chain::Filter, log_info!(self.logger, "Persistence of new ChannelMonitor for channel {} in progress", log_funding_info!(monitor)); pending_monitor_updates.push(update_id); }, - ChannelMonitorUpdateStatus::PermanentFailure => { - log_error!(self.logger, "Persistence of new ChannelMonitor for channel {} failed", log_funding_info!(monitor)); - return persist_res; - }, ChannelMonitorUpdateStatus::Completed => { log_info!(self.logger, "Persistence of new ChannelMonitor for channel {} completed", log_funding_info!(monitor)); - } + }, + ChannelMonitorUpdateStatus::UnrecoverableError => { + let err_str = "ChannelMonitor[Update] persistence failed unrecoverably. This indicates we cannot continue normal operation and must shut down."; + log_error!(self.logger, "{}", err_str); + panic!("{}", err_str); + }, } if let Some(ref chain_source) = self.chain_source { monitor.load_outputs_to_watch(chain_source); @@ -678,28 +732,25 @@ where C::Target: chain::Filter, entry.insert(MonitorHolder { monitor, pending_monitor_updates: Mutex::new(pending_monitor_updates), - channel_perm_failed: AtomicBool::new(false), last_chain_persist_height: AtomicUsize::new(self.highest_chain_height.load(Ordering::Acquire)), }); - persist_res + Ok(persist_res) } - /// Note that we persist the given `ChannelMonitor` update while holding the - /// `ChainMonitor` monitors lock. fn update_channel(&self, funding_txo: OutPoint, update: &ChannelMonitorUpdate) -> ChannelMonitorUpdateStatus { // Update the monitor that watches the channel referred to by the given outpoint. let monitors = self.monitors.read().unwrap(); - match monitors.get(&funding_txo) { + let ret = match monitors.get(&funding_txo) { None => { log_error!(self.logger, "Failed to update channel monitor: no such monitor registered"); // We should never ever trigger this from within ChannelManager. Technically a // user could use this object with some proxying in between which makes this // possible, but in tests and fuzzing, this should be a panic. - #[cfg(any(test, fuzzing))] + #[cfg(debug_assertions)] panic!("ChannelManager generated a channel update for a channel that was not yet registered!"); - #[cfg(not(any(test, fuzzing)))] - ChannelMonitorUpdateStatus::PermanentFailure + #[cfg(not(debug_assertions))] + ChannelMonitorUpdateStatus::InProgress }, Some(monitor_state) => { let monitor = &monitor_state.monitor; @@ -718,23 +769,28 @@ where C::Target: chain::Filter, pending_monitor_updates.push(update_id); log_debug!(self.logger, "Persistence of ChannelMonitorUpdate for channel {} in progress", log_funding_info!(monitor)); }, - ChannelMonitorUpdateStatus::PermanentFailure => { - monitor_state.channel_perm_failed.store(true, Ordering::Release); - log_error!(self.logger, "Persistence of ChannelMonitorUpdate for channel {} failed", log_funding_info!(monitor)); - }, ChannelMonitorUpdateStatus::Completed => { log_debug!(self.logger, "Persistence of ChannelMonitorUpdate for channel {} completed", log_funding_info!(monitor)); }, + ChannelMonitorUpdateStatus::UnrecoverableError => { /* we'll panic in a moment */ }, } if update_res.is_err() { - ChannelMonitorUpdateStatus::PermanentFailure - } else if monitor_state.channel_perm_failed.load(Ordering::Acquire) { - ChannelMonitorUpdateStatus::PermanentFailure + ChannelMonitorUpdateStatus::InProgress } else { persist_res } } + }; + if let ChannelMonitorUpdateStatus::UnrecoverableError = ret { + // Take the monitors lock for writing so that we poison it and any future + // operations going forward fail immediately. + core::mem::drop(monitors); + let _poison = self.monitors.write().unwrap(); + let err_str = "ChannelMonitor[Update] persistence failed unrecoverably. This indicates we cannot continue normal operation and must shut down."; + log_error!(self.logger, "{}", err_str); + panic!("{}", err_str); } + ret } fn release_pending_monitor_events(&self) -> Vec<(OutPoint, Vec, Option)> { @@ -745,19 +801,8 @@ where C::Target: chain::Filter, monitor_state.last_chain_persist_height.load(Ordering::Acquire) + LATENCY_GRACE_PERIOD_BLOCKS as usize > self.highest_chain_height.load(Ordering::Acquire) { - log_info!(self.logger, "A Channel Monitor sync is still in progress, refusing to provide monitor events!"); + log_debug!(self.logger, "A Channel Monitor sync is still in progress, refusing to provide monitor events!"); } else { - if monitor_state.channel_perm_failed.load(Ordering::Acquire) { - // If a `UpdateOrigin::ChainSync` persistence failed with `PermanantFailure`, - // we don't really know if the latest `ChannelMonitor` state is on disk or not. - // We're supposed to hold monitor updates until the latest state is on disk to - // avoid duplicate events, but the user told us persistence is screw-y and may - // not complete. We can't hold events forever because we may learn some payment - // preimage, so instead we just log and hope the user complied with the - // `PermanentFailure` requirements of having at least the local-disk copy - // updated. - log_info!(self.logger, "A Channel Monitor sync returned PermanentFailure. Returning monitor events but duplicate events may appear after reload!"); - } if is_pending_monitor_update { log_error!(self.logger, "A ChannelMonitor sync took longer than {} blocks to complete.", LATENCY_GRACE_PERIOD_BLOCKS); log_error!(self.logger, " To avoid funds-loss, we are allowing monitor updates to be released."); @@ -782,30 +827,13 @@ impl, { - #[cfg(not(anchors))] - /// Processes [`SpendableOutputs`] events produced from each [`ChannelMonitor`] upon maturity. - /// - /// An [`EventHandler`] may safely call back to the provider, though this shouldn't be needed in - /// order to handle these events. - /// - /// [`SpendableOutputs`]: events::Event::SpendableOutputs - fn process_pending_events(&self, handler: H) where H::Target: EventHandler { - let mut pending_events = Vec::new(); - for monitor_state in self.monitors.read().unwrap().values() { - pending_events.append(&mut monitor_state.monitor.get_and_clear_pending_events()); - } - for event in pending_events { - handler.handle_event(event); - } - } - #[cfg(anchors)] /// Processes [`SpendableOutputs`] events produced from each [`ChannelMonitor`] upon maturity. /// /// For channels featuring anchor outputs, this method will also process [`BumpTransaction`] /// events produced from each [`ChannelMonitor`] while there is a balance to claim onchain /// within each channel. As the confirmation of a commitment transaction may be critical to the - /// safety of funds, this method must be invoked frequently, ideally once for every chain tip - /// update (block connected or disconnected). + /// safety of funds, we recommend invoking this every 30 seconds, or lower if running in an + /// environment with spotty connections, like on mobile. /// /// An [`EventHandler`] may safely call back to the provider, though this shouldn't be needed in /// order to handle these events. @@ -813,26 +841,20 @@ impl(&self, handler: H) where H::Target: EventHandler { - let mut pending_events = Vec::new(); for monitor_state in self.monitors.read().unwrap().values() { - pending_events.append(&mut monitor_state.monitor.get_and_clear_pending_events()); - } - for event in pending_events { - handler.handle_event(event); + monitor_state.monitor.process_pending_events(&handler); } } } #[cfg(test)] mod tests { - use bitcoin::{BlockHeader, TxMerkleNode}; - use bitcoin::hashes::Hash; - use crate::{check_added_monitors, check_closed_broadcast, check_closed_event}; - use crate::{expect_payment_sent, expect_payment_claimed, expect_payment_sent_without_paths, expect_payment_path_successful, get_event_msg}; + use crate::check_added_monitors; + use crate::{expect_payment_claimed, expect_payment_path_successful, get_event_msg}; use crate::{get_htlc_update_msgs, get_local_commitment_txn, get_revoke_commit_msgs, get_route_and_payment_hash, unwrap_send_err}; use crate::chain::{ChannelMonitorUpdateStatus, Confirm, Watch}; use crate::chain::channelmonitor::LATENCY_GRACE_PERIOD_BLOCKS; - use crate::events::{Event, ClosureReason, MessageSendEvent, MessageSendEventsProvider}; + use crate::events::{Event, MessageSendEvent, MessageSendEventsProvider}; use crate::ln::channelmanager::{PaymentSendFailure, PaymentId, RecipientOnionFields}; use crate::ln::functional_test_utils::*; use crate::ln::msgs::ChannelMessageHandler; @@ -850,8 +872,8 @@ mod tests { create_announced_chan_between_nodes(&nodes, 0, 1); // Route two payments to be claimed at the same time. - let (payment_preimage_1, payment_hash_1, _) = route_payment(&nodes[0], &[&nodes[1]], 1_000_000); - let (payment_preimage_2, payment_hash_2, _) = route_payment(&nodes[0], &[&nodes[1]], 1_000_000); + let (payment_preimage_1, payment_hash_1, ..) = route_payment(&nodes[0], &[&nodes[1]], 1_000_000); + let (payment_preimage_2, payment_hash_2, ..) = route_payment(&nodes[0], &[&nodes[1]], 1_000_000); chanmon_cfgs[1].persister.offchain_monitor_updates.lock().unwrap().clear(); chanmon_cfgs[1].persister.set_update_ret(ChannelMonitorUpdateStatus::InProgress); @@ -911,7 +933,7 @@ mod tests { let updates = get_htlc_update_msgs!(nodes[1], nodes[0].node.get_our_node_id()); nodes[0].node.handle_update_fulfill_htlc(&nodes[1].node.get_our_node_id(), &updates.update_fulfill_htlcs[0]); - expect_payment_sent_without_paths!(nodes[0], payment_preimage_1); + expect_payment_sent(&nodes[0], payment_preimage_1, None, false, false); nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &updates.commitment_signed); check_added_monitors!(nodes[0], 1); let (as_first_raa, as_first_update) = get_revoke_commit_msgs!(nodes[0], nodes[1].node.get_our_node_id()); @@ -924,7 +946,7 @@ mod tests { let bs_first_raa = get_event_msg!(nodes[1], MessageSendEvent::SendRevokeAndACK, nodes[0].node.get_our_node_id()); nodes[0].node.handle_update_fulfill_htlc(&nodes[1].node.get_our_node_id(), &bs_second_updates.update_fulfill_htlcs[0]); - expect_payment_sent_without_paths!(nodes[0], payment_preimage_2); + expect_payment_sent(&nodes[0], payment_preimage_2, None, false, false); nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &bs_second_updates.commitment_signed); check_added_monitors!(nodes[0], 1); nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &bs_first_raa); @@ -958,7 +980,7 @@ mod tests { let (route, second_payment_hash, _, second_payment_secret) = get_route_and_payment_hash!(nodes[0], nodes[1], 100_000); // First route a payment that we will claim on chain and give the recipient the preimage. - let (payment_preimage, payment_hash, _) = route_payment(&nodes[0], &[&nodes[1]], 1_000_000); + let (payment_preimage, payment_hash, ..) = route_payment(&nodes[0], &[&nodes[1]], 1_000_000); nodes[1].node.claim_funds(payment_preimage); expect_payment_claimed!(nodes[1], payment_hash, 1_000_000); nodes[1].node.get_and_clear_pending_msg_events(); @@ -972,10 +994,7 @@ mod tests { // Connect B's commitment transaction, but only to the ChainMonitor/ChannelMonitor. The // channel is now closed, but the ChannelManager doesn't know that yet. - let new_header = BlockHeader { - version: 2, time: 0, bits: 0, nonce: 0, - prev_blockhash: nodes[0].best_block_info().0, - merkle_root: TxMerkleNode::all_zeros() }; + let new_header = create_dummy_header(nodes[0].best_block_info().0, 0); nodes[0].chain_monitor.chain_monitor.transactions_confirmed(&new_header, &[(0, &remote_txn[0]), (1, &remote_txn[1])], nodes[0].best_block_info().1 + 1); assert!(nodes[0].chain_monitor.release_pending_monitor_events().is_empty()); @@ -987,11 +1006,8 @@ mod tests { chanmon_cfgs[0].persister.set_update_ret(ChannelMonitorUpdateStatus::Completed); unwrap_send_err!(nodes[0].node.send_payment_with_route(&route, second_payment_hash, RecipientOnionFields::secret_only(second_payment_secret), PaymentId(second_payment_hash.0) - ), true, APIError::ChannelUnavailable { ref err }, - assert!(err.contains("ChannelMonitor storage failure"))); - check_added_monitors!(nodes[0], 2); // After the failure we generate a close-channel monitor update - check_closed_broadcast!(nodes[0], true); - check_closed_event!(nodes[0], 1, ClosureReason::ProcessingError { err: "ChannelMonitor storage failure".to_string() }); + ), false, APIError::MonitorUpdateInProgress, {}); + check_added_monitors!(nodes[0], 1); // However, as the ChainMonitor is still waiting for the original persistence to complete, // it won't yet release the MonitorEvents. @@ -999,10 +1015,7 @@ mod tests { if block_timeout { // After three blocks, pending MontiorEvents should be released either way. - let latest_header = BlockHeader { - version: 2, time: 0, bits: 0, nonce: 0, - prev_blockhash: nodes[0].best_block_info().0, - merkle_root: TxMerkleNode::all_zeros() }; + let latest_header = create_dummy_header(nodes[0].best_block_info().0, 0); nodes[0].chain_monitor.chain_monitor.best_block_updated(&latest_header, nodes[0].best_block_info().1 + LATENCY_GRACE_PERIOD_BLOCKS); } else { let persistences = chanmon_cfgs[0].persister.chain_sync_monitor_persistences.lock().unwrap().clone(); @@ -1013,7 +1026,7 @@ mod tests { } } - expect_payment_sent!(nodes[0], payment_preimage); + expect_payment_sent(&nodes[0], payment_preimage, None, true, false); } #[test] @@ -1023,7 +1036,8 @@ mod tests { } #[test] - fn update_during_chainsync_fails_channel() { + #[cfg(feature = "std")] + fn update_during_chainsync_poisons_channel() { 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]); @@ -1031,17 +1045,15 @@ mod tests { create_announced_chan_between_nodes(&nodes, 0, 1); chanmon_cfgs[0].persister.chain_sync_monitor_persistences.lock().unwrap().clear(); - chanmon_cfgs[0].persister.set_update_ret(ChannelMonitorUpdateStatus::PermanentFailure); - - connect_blocks(&nodes[0], 1); - // Before processing events, the ChannelManager will still think the Channel is open and - // there won't be any ChannelMonitorUpdates - assert_eq!(nodes[0].node.list_channels().len(), 1); - check_added_monitors!(nodes[0], 0); - // ... however once we get events once, the channel will close, creating a channel-closed - // ChannelMonitorUpdate. - check_closed_broadcast!(nodes[0], true); - check_closed_event!(nodes[0], 1, ClosureReason::ProcessingError { err: "Failed to persist ChannelMonitor update during chain sync".to_string() }); - check_added_monitors!(nodes[0], 1); + chanmon_cfgs[0].persister.set_update_ret(ChannelMonitorUpdateStatus::UnrecoverableError); + + assert!(std::panic::catch_unwind(|| { + // Returning an UnrecoverableError should always panic immediately + connect_blocks(&nodes[0], 1); + }).is_err()); + assert!(std::panic::catch_unwind(|| { + // ...and also poison our locks causing later use to panic as well + core::mem::drop(nodes); + }).is_err()); } }