//! events. The remote server would make use of [`ChainMonitor`] for block processing and for
//! servicing [`ChannelMonitor`] updates from the client.
-use bitcoin::blockdata::block::{Block, BlockHeader};
+use bitcoin::blockdata::block::BlockHeader;
use bitcoin::hash_types::Txid;
use chain;
use sync::{RwLock, RwLockReadGuard, Mutex, MutexGuard};
use core::ops::Deref;
use core::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+use bitcoin::secp256k1::PublicKey;
#[derive(Clone, Copy, Hash, PartialEq, Eq)]
/// A specific update's ID stored in a `MonitorUpdateId`, separated out to make the contents
persister: P,
/// "User-provided" (ie persistence-completion/-failed) [`MonitorEvent`]s. These came directly
/// from the user and not from a [`ChannelMonitor`].
- pending_monitor_events: Mutex<Vec<MonitorEvent>>,
+ pending_monitor_events: Mutex<Vec<(OutPoint, Vec<MonitorEvent>, Option<PublicKey>)>>,
/// The best block height seen, used as a proxy for the passage of time.
highest_chain_height: AtomicUsize,
}
where
FN: Fn(&ChannelMonitor<ChannelSigner>, &TransactionData) -> Vec<TransactionOutputs>
{
- let mut dependent_txdata = Vec::new();
- {
- let monitor_states = self.monitors.write().unwrap();
- if let Some(height) = best_height {
- // If the best block height is being updated, update highest_chain_height under the
- // monitors write lock.
- let old_height = self.highest_chain_height.load(Ordering::Acquire);
- let new_height = height as usize;
- if new_height > old_height {
- self.highest_chain_height.store(new_height, Ordering::Release);
- }
+ let monitor_states = self.monitors.write().unwrap();
+ if let Some(height) = best_height {
+ // If the best block height is being updated, update highest_chain_height under the
+ // monitors write lock.
+ let old_height = self.highest_chain_height.load(Ordering::Acquire);
+ let new_height = height as usize;
+ if new_height > old_height {
+ 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
- // TemporaryFailure to always immediately be considered "overly delayed".
- monitor_state.last_chain_persist_height.store(height as usize, 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
+ // TemporaryFailure 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) {
- Ok(()) =>
- log_trace!(self.logger, "Finished syncing Channel Monitor for channel {}", log_funding_info!(monitor)),
- Err(ChannelMonitorUpdateErr::PermanentFailure) => {
- monitor_state.channel_perm_failed.store(true, Ordering::Release);
- self.pending_monitor_events.lock().unwrap().push(MonitorEvent::UpdateFailed(*funding_outpoint));
- },
- Err(ChannelMonitorUpdateErr::TemporaryFailure) => {
- 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) {
+ Ok(()) =>
+ log_trace!(self.logger, "Finished syncing Channel Monitor for channel {}", log_funding_info!(monitor)),
+ Err(ChannelMonitorUpdateErr::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()));
+ },
+ Err(ChannelMonitorUpdateErr::TemporaryFailure) => {
+ 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);
+ },
}
+ }
- // 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 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,
- };
- if let Some(tx) = chain_source.register_output(output) {
- dependent_txdata.push(tx);
- }
- }
+ // 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)
}
}
}
}
-
- // 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.process_chain_data(header, None, &txdata, process); // We skip the best height the second go-around
- }
}
/// Creates a new `ChainMonitor` used to watch on-chain activity pertaining to channels.
// UpdateCompleted event.
return Ok(());
}
- self.pending_monitor_events.lock().unwrap().push(MonitorEvent::UpdateCompleted {
+ self.pending_monitor_events.lock().unwrap().push((funding_txo, vec![MonitorEvent::UpdateCompleted {
funding_txo,
monitor_update_id: monitor_data.monitor.get_latest_update_id(),
- });
+ }], monitor_data.monitor.get_counterparty_node_id()));
},
MonitorUpdateId { contents: UpdateOrigin::ChainSync(_) } => {
if !monitor_data.has_pending_chainsync_updates(&pending_monitor_updates) {
/// This wrapper avoids having to update some of our tests for now as they assume the direct
/// chain::Watch API wherein we mark a monitor fully-updated by just calling
/// channel_monitor_updated once with the highest ID.
- #[cfg(any(test, feature = "fuzztarget"))]
+ #[cfg(any(test, fuzzing))]
pub fn force_channel_monitor_updated(&self, funding_txo: OutPoint, monitor_update_id: u64) {
- self.pending_monitor_events.lock().unwrap().push(MonitorEvent::UpdateCompleted {
+ let monitors = self.monitors.read().unwrap();
+ let counterparty_node_id = monitors.get(&funding_txo).and_then(|m| m.monitor.get_counterparty_node_id());
+ self.pending_monitor_events.lock().unwrap().push((funding_txo, vec![MonitorEvent::UpdateCompleted {
funding_txo,
monitor_update_id,
- });
+ }], counterparty_node_id));
}
- #[cfg(any(test, feature = "fuzztarget", feature = "_test_utils"))]
+ #[cfg(any(test, fuzzing, feature = "_test_utils"))]
pub fn get_and_clear_pending_events(&self) -> Vec<events::Event> {
use util::events::EventsProvider;
let events = core::cell::RefCell::new(Vec::new());
L::Target: Logger,
P::Target: Persist<ChannelSigner>,
{
- fn block_connected(&self, block: &Block, height: u32) {
- let header = &block.header;
- let txdata: Vec<_> = block.txdata.iter().enumerate().collect();
+ fn filtered_block_connected(&self, header: &BlockHeader, txdata: &TransactionData, height: u32) {
log_debug!(self.logger, "New best block {} at height {} provided via block_connected", header.block_hash(), height);
self.process_chain_data(header, Some(height), &txdata, |monitor, txdata| {
monitor.block_connected(
// 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, feature = "fuzztarget"))]
+ #[cfg(any(test, fuzzing))]
panic!("ChannelManager generated a channel update for a channel that was not yet registered!");
- #[cfg(not(any(test, feature = "fuzztarget")))]
+ #[cfg(not(any(test, fuzzing)))]
Err(ChannelMonitorUpdateErr::PermanentFailure)
},
Some(monitor_state) => {
let monitor = &monitor_state.monitor;
log_trace!(self.logger, "Updating ChannelMonitor for channel {}", log_funding_info!(monitor));
- let update_res = monitor.update_monitor(&update, &self.broadcaster, &self.fee_estimator, &self.logger);
+ let update_res = monitor.update_monitor(&update, &self.broadcaster, &*self.fee_estimator, &self.logger);
if update_res.is_err() {
log_error!(self.logger, "Failed to update ChannelMonitor for channel {}.", log_funding_info!(monitor));
}
}
}
- fn release_pending_monitor_events(&self) -> Vec<MonitorEvent> {
+ fn release_pending_monitor_events(&self) -> Vec<(OutPoint, Vec<MonitorEvent>, Option<PublicKey>)> {
let mut pending_monitor_events = self.pending_monitor_events.lock().unwrap().split_off(0);
for monitor_state in self.monitors.read().unwrap().values() {
let is_pending_monitor_update = monitor_state.has_pending_chainsync_updates(&monitor_state.pending_monitor_updates.lock().unwrap());
log_error!(self.logger, " To avoid funds-loss, we are allowing monitor updates to be released.");
log_error!(self.logger, " This may cause duplicate payment events to be generated.");
}
- pending_monitor_events.append(&mut monitor_state.monitor.get_and_clear_pending_monitor_events());
+ let monitor_events = monitor_state.monitor.get_and_clear_pending_monitor_events();
+ if monitor_events.len() > 0 {
+ let monitor_outpoint = monitor_state.monitor.get_funding_txo().0;
+ let counterparty_node_id = monitor_state.monitor.get_counterparty_node_id();
+ pending_monitor_events.push((monitor_outpoint, monitor_events, counterparty_node_id));
+ }
}
}
pending_monitor_events
#[cfg(test)]
mod tests {
- use ::{check_added_monitors, get_local_commitment_txn};
- use ln::features::InitFeatures;
+ use bitcoin::{BlockHeader, TxMerkleNode};
+ use bitcoin::hashes::Hash;
+ use ::{check_added_monitors, check_closed_broadcast, check_closed_event};
+ use ::{expect_payment_sent, expect_payment_claimed, expect_payment_sent_without_paths, expect_payment_path_successful, get_event_msg};
+ use ::{get_htlc_update_msgs, get_local_commitment_txn, get_revoke_commit_msgs, get_route_and_payment_hash, unwrap_send_err};
+ use chain::{ChannelMonitorUpdateErr, Confirm, Watch};
+ use chain::channelmonitor::LATENCY_GRACE_PERIOD_BLOCKS;
+ use ln::channelmanager::{self, PaymentSendFailure};
use ln::functional_test_utils::*;
- 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.
+ use ln::msgs::ChannelMessageHandler;
+ use util::errors::APIError;
+ use util::events::{ClosureReason, MessageSendEvent, MessageSendEventsProvider};
+
#[test]
- fn connect_block_checks_dependent_transactions() {
+ fn test_async_ooo_offchain_updates() {
+ // Test that if we have multiple offchain updates being persisted and they complete
+ // out-of-order, the ChainMonitor waits until all have completed before informing the
+ // ChannelManager.
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());
+ create_announced_chan_between_nodes(&nodes, 0, 1, channelmanager::provided_init_features(), channelmanager::provided_init_features());
- // 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);
+ // 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);
- assert_eq!(txn.len(), 2);
- (txn.remove(0), txn.remove(0))
- };
+ chanmon_cfgs[1].persister.offchain_monitor_updates.lock().unwrap().clear();
+ chanmon_cfgs[1].persister.set_update_ret(Err(ChannelMonitorUpdateErr::TemporaryFailure));
- // 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 });
+ nodes[1].node.claim_funds(payment_preimage_1);
+ check_added_monitors!(nodes[1], 1);
+ expect_payment_claimed!(nodes[1], payment_hash_1, 1_000_000);
+ nodes[1].node.claim_funds(payment_preimage_2);
+ check_added_monitors!(nodes[1], 1);
+ expect_payment_claimed!(nodes[1], payment_hash_2, 1_000_000);
+
+ chanmon_cfgs[1].persister.set_update_ret(Ok(()));
+
+ let persistences = chanmon_cfgs[1].persister.offchain_monitor_updates.lock().unwrap().clone();
+ assert_eq!(persistences.len(), 1);
+ let (funding_txo, updates) = persistences.iter().next().unwrap();
+ assert_eq!(updates.len(), 2);
+
+ // Note that updates is a HashMap so the ordering here is actually random. This shouldn't
+ // fail either way but if it fails intermittently it's depending on the ordering of updates.
+ let mut update_iter = updates.iter();
+ nodes[1].chain_monitor.chain_monitor.channel_monitor_updated(*funding_txo, update_iter.next().unwrap().clone()).unwrap();
+ assert!(nodes[1].chain_monitor.release_pending_monitor_events().is_empty());
+ assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
+ nodes[1].chain_monitor.chain_monitor.channel_monitor_updated(*funding_txo, update_iter.next().unwrap().clone()).unwrap();
+
+ // Now manually walk the commitment signed dance - because we claimed two payments
+ // back-to-back it doesn't fit into the neat walk commitment_signed_dance does.
+
+ 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);
+ 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());
+
+ nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), &as_first_raa);
+ check_added_monitors!(nodes[1], 1);
+ let bs_second_updates = get_htlc_update_msgs!(nodes[1], nodes[0].node.get_our_node_id());
+ nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), &as_first_update);
+ check_added_monitors!(nodes[1], 1);
+ 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);
+ 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);
+ expect_payment_path_successful!(nodes[0]);
+ check_added_monitors!(nodes[0], 1);
+ let (as_second_raa, as_second_update) = get_revoke_commit_msgs!(nodes[0], nodes[1].node.get_our_node_id());
+
+ nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), &as_second_raa);
+ check_added_monitors!(nodes[1], 1);
+ nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), &as_second_update);
+ check_added_monitors!(nodes[1], 1);
+ let bs_second_raa = get_event_msg!(nodes[1], MessageSendEvent::SendRevokeAndACK, nodes[0].node.get_our_node_id());
- // 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);
+ nodes[0].node.handle_revoke_and_ack(&nodes[1].node.get_our_node_id(), &bs_second_raa);
+ expect_payment_path_successful!(nodes[0]);
+ check_added_monitors!(nodes[0], 1);
+ }
- // Clean up so uninteresting assertions don't fail.
- check_added_monitors!(nodes[1], 1);
+ fn do_chainsync_pauses_events(block_timeout: bool) {
+ // When a chainsync monitor update occurs, any MonitorUpdates should be held before being
+ // passed upstream to a `ChannelManager` via `Watch::release_pending_monitor_events`. This
+ // tests that behavior, as well as some ways it might go wrong.
+ 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, channelmanager::provided_init_features(), channelmanager::provided_init_features());
+
+ // Get a route for later and rebalance the channel somewhat
+ send_payment(&nodes[0], &[&nodes[1]], 10_000_000);
+ 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);
+ 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();
- nodes[1].node.get_and_clear_pending_events();
+ check_added_monitors!(nodes[1], 1);
+ let remote_txn = get_local_commitment_txn!(nodes[1], channel.2);
+ assert_eq!(remote_txn.len(), 2);
+
+ // Temp-fail the block connection which will hold the channel-closed event
+ chanmon_cfgs[0].persister.chain_sync_monitor_persistences.lock().unwrap().clear();
+ chanmon_cfgs[0].persister.set_update_ret(Err(ChannelMonitorUpdateErr::TemporaryFailure));
+
+ // 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() };
+ 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());
+ nodes[0].chain_monitor.chain_monitor.best_block_updated(&new_header, nodes[0].best_block_info().1 + 1);
+ assert!(nodes[0].chain_monitor.release_pending_monitor_events().is_empty());
+
+ // If the ChannelManager tries to update the channel, however, the ChainMonitor will pass
+ // the update through to the ChannelMonitor which will refuse it (as the channel is closed).
+ chanmon_cfgs[0].persister.set_update_ret(Ok(()));
+ unwrap_send_err!(nodes[0].node.send_payment(&route, second_payment_hash, &Some(second_payment_secret)),
+ 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() });
+
+ // However, as the ChainMonitor is still waiting for the original persistence to complete,
+ // it won't yet release the MonitorEvents.
+ assert!(nodes[0].chain_monitor.release_pending_monitor_events().is_empty());
+
+ 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() };
+ 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();
+ for (funding_outpoint, update_ids) in persistences {
+ for update_id in update_ids {
+ nodes[0].chain_monitor.chain_monitor.channel_monitor_updated(funding_outpoint, update_id).unwrap();
+ }
+ }
+ }
+
+ expect_payment_sent!(nodes[0], payment_preimage);
+ }
+
+ #[test]
+ fn chainsync_pauses_events() {
+ do_chainsync_pauses_events(false);
+ do_chainsync_pauses_events(true);
+ }
+
+ #[test]
+ fn update_during_chainsync_fails_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]);
+ let nodes = create_network(2, &node_cfgs, &node_chanmgrs);
+ create_announced_chan_between_nodes(&nodes, 0, 1, channelmanager::provided_init_features(), channelmanager::provided_init_features());
+
+ chanmon_cfgs[0].persister.chain_sync_monitor_persistences.lock().unwrap().clear();
+ chanmon_cfgs[0].persister.set_update_ret(Err(ChannelMonitorUpdateErr::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);
}
}