X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=lightning%2Fsrc%2Fln%2Fchannelmanager.rs;h=b5e5a43fee828515cd6c3f5570113615354758c3;hb=ba1349982ba28657c9e2d03a5b02c3ecc054b5cc;hp=c8adc721235470dc37bce5a9a2651fc8e74dc8ed;hpb=1d31b0e84e2adeb774f4e626c240337c65cb411c;p=rust-lightning diff --git a/lightning/src/ln/channelmanager.rs b/lightning/src/ln/channelmanager.rs index c8adc721..b5e5a43f 100644 --- a/lightning/src/ln/channelmanager.rs +++ b/lightning/src/ln/channelmanager.rs @@ -308,9 +308,9 @@ impl core::hash::Hash for HTLCSource { } } } -#[cfg(not(feature = "grind_signatures"))] -#[cfg(test)] impl HTLCSource { + #[cfg(not(feature = "grind_signatures"))] + #[cfg(test)] pub fn dummy() -> Self { HTLCSource::OutboundRoute { path: Vec::new(), @@ -320,6 +320,18 @@ impl HTLCSource { payment_secret: None, } } + + #[cfg(debug_assertions)] + /// Checks whether this HTLCSource could possibly match the given HTLC output in a commitment + /// transaction. Useful to ensure different datastructures match up. + pub(crate) fn possibly_matches_output(&self, htlc: &super::chan_utils::HTLCOutputInCommitment) -> bool { + if let HTLCSource::OutboundRoute { first_hop_htlc_msat, .. } = self { + *first_hop_htlc_msat == htlc.amount_msat + } else { + // There's nothing we can check for forwarded HTLCs + true + } + } } struct ReceiveError { @@ -1496,18 +1508,31 @@ macro_rules! send_channel_ready { }} } +macro_rules! emit_channel_pending_event { + ($locked_events: expr, $channel: expr) => { + if $channel.should_emit_channel_pending_event() { + $locked_events.push(events::Event::ChannelPending { + channel_id: $channel.channel_id(), + former_temporary_channel_id: $channel.temporary_channel_id(), + counterparty_node_id: $channel.get_counterparty_node_id(), + user_channel_id: $channel.get_user_id(), + funding_txo: $channel.get_funding_txo().unwrap().into_bitcoin_outpoint(), + }); + $channel.set_channel_pending_event_emitted(); + } + } +} + macro_rules! emit_channel_ready_event { - ($self: expr, $channel: expr) => { + ($locked_events: expr, $channel: expr) => { if $channel.should_emit_channel_ready_event() { - { - let mut pending_events = $self.pending_events.lock().unwrap(); - pending_events.push(events::Event::ChannelReady { - channel_id: $channel.channel_id(), - user_channel_id: $channel.get_user_id(), - counterparty_node_id: $channel.get_counterparty_node_id(), - channel_type: $channel.get_channel_type().clone(), - }); - } + debug_assert!($channel.channel_pending_event_emitted()); + $locked_events.push(events::Event::ChannelReady { + channel_id: $channel.channel_id(), + user_channel_id: $channel.get_user_id(), + counterparty_node_id: $channel.get_counterparty_node_id(), + channel_type: $channel.get_channel_type().clone(), + }); $channel.set_channel_ready_event_emitted(); } } @@ -4184,6 +4209,7 @@ where claim_from_onchain_tx: from_onchain, prev_channel_id, next_channel_id, + outbound_amount_forwarded_msat: forwarded_htlc_value_msat, }}) } else { None } }); @@ -4252,8 +4278,6 @@ where }); } - emit_channel_ready_event!(self, channel); - macro_rules! handle_cs { () => { if let Some(update) = commitment_update { pending_msg_events.push(events::MessageSendEvent::UpdateHTLCs { @@ -4286,6 +4310,12 @@ where self.tx_broadcaster.broadcast_transaction(&tx); } + { + let mut pending_events = self.pending_events.lock().unwrap(); + emit_channel_pending_event!(pending_events, channel); + emit_channel_ready_event!(pending_events, channel); + } + htlc_forwards } @@ -4710,7 +4740,10 @@ where } } - emit_channel_ready_event!(self, chan.get_mut()); + { + let mut pending_events = self.pending_events.lock().unwrap(); + emit_channel_ready_event!(pending_events, chan.get_mut()); + } Ok(()) }, @@ -6035,7 +6068,10 @@ where } } - emit_channel_ready_event!(self, channel); + { + let mut pending_events = self.pending_events.lock().unwrap(); + emit_channel_ready_event!(pending_events, channel); + } if let Some(announcement_sigs) = announcement_sigs { log_trace!(self.logger, "Sending announcement_signatures for channel {}", log_bytes!(channel.channel_id())); @@ -6146,34 +6182,11 @@ where } } - /// Blocks until ChannelManager needs to be persisted or a timeout is reached. It returns a bool - /// indicating whether persistence is necessary. Only one listener on - /// [`await_persistable_update`], [`await_persistable_update_timeout`], or a future returned by - /// [`get_persistable_update_future`] is guaranteed to be woken up. - /// - /// Note that this method is not available with the `no-std` feature. + /// Gets a [`Future`] that completes when this [`ChannelManager`] needs to be persisted. /// - /// [`await_persistable_update`]: Self::await_persistable_update - /// [`await_persistable_update_timeout`]: Self::await_persistable_update_timeout - /// [`get_persistable_update_future`]: Self::get_persistable_update_future - #[cfg(any(test, feature = "std"))] - pub fn await_persistable_update_timeout(&self, max_wait: Duration) -> bool { - self.persistence_notifier.wait_timeout(max_wait) - } - - /// Blocks until ChannelManager needs to be persisted. Only one listener on - /// [`await_persistable_update`], `await_persistable_update_timeout`, or a future returned by - /// [`get_persistable_update_future`] is guaranteed to be woken up. + /// Note that callbacks registered on the [`Future`] MUST NOT call back into this + /// [`ChannelManager`] and should instead register actions to be taken later. /// - /// [`await_persistable_update`]: Self::await_persistable_update - /// [`get_persistable_update_future`]: Self::get_persistable_update_future - pub fn await_persistable_update(&self) { - self.persistence_notifier.wait() - } - - /// Gets a [`Future`] that completes when a persistable update is available. Note that - /// callbacks registered on the [`Future`] MUST NOT call back into this [`ChannelManager`] and - /// should instead register actions to be taken later. pub fn get_persistable_update_future(&self) -> Future { self.persistence_notifier.get_future() } @@ -7354,6 +7367,7 @@ where let mut id_to_peer = HashMap::with_capacity(cmp::min(channel_count as usize, 128)); let mut short_to_chan_info = HashMap::with_capacity(cmp::min(channel_count as usize, 128)); let mut channel_closures = Vec::new(); + let mut pending_background_events = Vec::new(); for _ in 0..channel_count { let mut channel: Channel<::Signer> = Channel::read(reader, ( &args.entropy_source, &args.signer_provider, best_block_height, &provided_channel_type_features(&args.default_config) @@ -7383,9 +7397,11 @@ where log_error!(args.logger, " The channel will be force-closed and the latest commitment transaction from the ChannelMonitor broadcast."); log_error!(args.logger, " The ChannelMonitor for channel {} is at update_id {} but the ChannelManager is at update_id {}.", log_bytes!(channel.channel_id()), monitor.get_latest_update_id(), channel.get_latest_monitor_update_id()); - let (_, mut new_failed_htlcs) = channel.force_shutdown(true); + let (monitor_update, mut new_failed_htlcs) = channel.force_shutdown(true); + if let Some(monitor_update) = monitor_update { + pending_background_events.push(BackgroundEvent::ClosingMonitorUpdate(monitor_update)); + } failed_htlcs.append(&mut new_failed_htlcs); - monitor.broadcast_latest_holder_commitment_txn(&args.tx_broadcaster, &args.logger); channel_closures.push(events::Event::ChannelClosed { channel_id: channel.channel_id(), user_channel_id: channel.get_user_id(), @@ -7450,10 +7466,13 @@ where } } - for (funding_txo, monitor) in args.channel_monitors.iter_mut() { + for (funding_txo, _) in args.channel_monitors.iter() { if !funding_txo_set.contains(funding_txo) { - log_info!(args.logger, "Broadcasting latest holder commitment transaction for closed channel {}", log_bytes!(funding_txo.to_channel_id())); - monitor.broadcast_latest_holder_commitment_txn(&args.tx_broadcaster, &args.logger); + let monitor_update = ChannelMonitorUpdate { + update_id: CLOSED_CHANNEL_UPDATE_ID, + updates: vec![ChannelMonitorUpdateStep::ChannelForceClosed { should_broadcast: true }], + }; + pending_background_events.push(BackgroundEvent::ClosingMonitorUpdate((*funding_txo, monitor_update))); } } @@ -7506,10 +7525,17 @@ where } let background_event_count: u64 = Readable::read(reader)?; - let mut pending_background_events_read: Vec = Vec::with_capacity(cmp::min(background_event_count as usize, MAX_ALLOC_SIZE/mem::size_of::())); for _ in 0..background_event_count { match ::read(reader)? { - 0 => pending_background_events_read.push(BackgroundEvent::ClosingMonitorUpdate((Readable::read(reader)?, Readable::read(reader)?))), + 0 => { + let (funding_txo, monitor_update): (OutPoint, ChannelMonitorUpdate) = (Readable::read(reader)?, Readable::read(reader)?); + if pending_background_events.iter().find(|e| { + let BackgroundEvent::ClosingMonitorUpdate((pending_funding_txo, pending_monitor_update)) = e; + *pending_funding_txo == funding_txo && *pending_monitor_update == monitor_update + }).is_none() { + pending_background_events.push(BackgroundEvent::ClosingMonitorUpdate((funding_txo, monitor_update))); + } + } _ => return Err(DecodeError::InvalidValue), } } @@ -7884,7 +7910,7 @@ where per_peer_state: FairRwLock::new(per_peer_state), pending_events: Mutex::new(pending_events_read), - pending_background_events: Mutex::new(pending_background_events_read), + pending_background_events: Mutex::new(pending_background_events), total_consistency_lock: RwLock::new(()), persistence_notifier: Notifier::new(), @@ -7915,6 +7941,7 @@ mod tests { use bitcoin::hashes::Hash; use bitcoin::hashes::sha256::Hash as Sha256; use bitcoin::secp256k1::{PublicKey, Secp256k1, SecretKey}; + #[cfg(feature = "std")] use core::time::Duration; use core::sync::atomic::Ordering; use crate::events::{Event, HTLCDestination, MessageSendEvent, MessageSendEventsProvider, ClosureReason}; @@ -7940,9 +7967,9 @@ mod tests { // All nodes start with a persistable update pending as `create_network` connects each node // with all other nodes to make most tests simpler. - assert!(nodes[0].node.await_persistable_update_timeout(Duration::from_millis(1))); - assert!(nodes[1].node.await_persistable_update_timeout(Duration::from_millis(1))); - assert!(nodes[2].node.await_persistable_update_timeout(Duration::from_millis(1))); + assert!(nodes[0].node.get_persistable_update_future().poll_is_complete()); + assert!(nodes[1].node.get_persistable_update_future().poll_is_complete()); + assert!(nodes[2].node.get_persistable_update_future().poll_is_complete()); let mut chan = create_announced_chan_between_nodes(&nodes, 0, 1); @@ -7956,19 +7983,19 @@ mod tests { &nodes[0].node.get_our_node_id()).pop().unwrap(); // The first two nodes (which opened a channel) should now require fresh persistence - assert!(nodes[0].node.await_persistable_update_timeout(Duration::from_millis(1))); - assert!(nodes[1].node.await_persistable_update_timeout(Duration::from_millis(1))); + assert!(nodes[0].node.get_persistable_update_future().poll_is_complete()); + assert!(nodes[1].node.get_persistable_update_future().poll_is_complete()); // ... but the last node should not. - assert!(!nodes[2].node.await_persistable_update_timeout(Duration::from_millis(1))); + assert!(!nodes[2].node.get_persistable_update_future().poll_is_complete()); // After persisting the first two nodes they should no longer need fresh persistence. - assert!(!nodes[0].node.await_persistable_update_timeout(Duration::from_millis(1))); - assert!(!nodes[1].node.await_persistable_update_timeout(Duration::from_millis(1))); + assert!(!nodes[0].node.get_persistable_update_future().poll_is_complete()); + assert!(!nodes[1].node.get_persistable_update_future().poll_is_complete()); // Node 3, unrelated to the only channel, shouldn't care if it receives a channel_update // about the channel. nodes[2].node.handle_channel_update(&nodes[1].node.get_our_node_id(), &chan.0); nodes[2].node.handle_channel_update(&nodes[1].node.get_our_node_id(), &chan.1); - assert!(!nodes[2].node.await_persistable_update_timeout(Duration::from_millis(1))); + assert!(!nodes[2].node.get_persistable_update_future().poll_is_complete()); // The nodes which are a party to the channel should also ignore messages from unrelated // parties. @@ -7976,8 +8003,8 @@ mod tests { nodes[0].node.handle_channel_update(&nodes[2].node.get_our_node_id(), &chan.1); nodes[1].node.handle_channel_update(&nodes[2].node.get_our_node_id(), &chan.0); nodes[1].node.handle_channel_update(&nodes[2].node.get_our_node_id(), &chan.1); - assert!(!nodes[0].node.await_persistable_update_timeout(Duration::from_millis(1))); - assert!(!nodes[1].node.await_persistable_update_timeout(Duration::from_millis(1))); + assert!(!nodes[0].node.get_persistable_update_future().poll_is_complete()); + assert!(!nodes[1].node.get_persistable_update_future().poll_is_complete()); // At this point the channel info given by peers should still be the same. assert_eq!(nodes[0].node.list_channels()[0], node_a_chan_info); @@ -7994,8 +8021,8 @@ mod tests { // persisted and that its channel info remains the same. nodes[0].node.handle_channel_update(&nodes[1].node.get_our_node_id(), &as_update); nodes[1].node.handle_channel_update(&nodes[0].node.get_our_node_id(), &bs_update); - assert!(!nodes[0].node.await_persistable_update_timeout(Duration::from_millis(1))); - assert!(!nodes[1].node.await_persistable_update_timeout(Duration::from_millis(1))); + assert!(!nodes[0].node.get_persistable_update_future().poll_is_complete()); + assert!(!nodes[1].node.get_persistable_update_future().poll_is_complete()); assert_eq!(nodes[0].node.list_channels()[0], node_a_chan_info); assert_eq!(nodes[1].node.list_channels()[0], node_b_chan_info); @@ -8003,8 +8030,8 @@ mod tests { // the channel info has updated. nodes[0].node.handle_channel_update(&nodes[1].node.get_our_node_id(), &bs_update); nodes[1].node.handle_channel_update(&nodes[0].node.get_our_node_id(), &as_update); - assert!(nodes[0].node.await_persistable_update_timeout(Duration::from_millis(1))); - assert!(nodes[1].node.await_persistable_update_timeout(Duration::from_millis(1))); + assert!(nodes[0].node.get_persistable_update_future().poll_is_complete()); + assert!(nodes[1].node.get_persistable_update_future().poll_is_complete()); assert_ne!(nodes[0].node.list_channels()[0], node_a_chan_info); assert_ne!(nodes[1].node.list_channels()[0], node_b_chan_info); } @@ -8449,6 +8476,7 @@ mod tests { assert_eq!(nodes_0_lock.len(), 1); assert!(nodes_0_lock.contains_key(channel_id)); } + expect_channel_pending_event(&nodes[1], &nodes[0].node.get_our_node_id()); { // Assert that `nodes[1]`'s `id_to_peer` map is populated with the channel as soon as @@ -8461,6 +8489,7 @@ mod tests { let funding_signed = get_event_msg!(nodes[1], MessageSendEvent::SendFundingSigned, nodes[0].node.get_our_node_id()); nodes[0].node.handle_funding_signed(&nodes[1].node.get_our_node_id(), &funding_signed); check_added_monitors!(nodes[0], 1); + expect_channel_pending_event(&nodes[0], &nodes[1].node.get_our_node_id()); let (channel_ready, _) = create_chan_between_nodes_with_value_confirm(&nodes[0], &nodes[1], &tx); let (announcement, nodes_0_update, nodes_1_update) = create_chan_between_nodes_with_value_b(&nodes[0], &nodes[1], &channel_ready); update_nodes_with_chan_announce(&nodes, 0, 1, &announcement, &nodes_0_update, &nodes_1_update); @@ -8603,10 +8632,13 @@ mod tests { nodes[1].node.handle_funding_created(&nodes[0].node.get_our_node_id(), &funding_created_msg); check_added_monitors!(nodes[1], 1); + expect_channel_pending_event(&nodes[1], &nodes[0].node.get_our_node_id()); + let funding_signed = get_event_msg!(nodes[1], MessageSendEvent::SendFundingSigned, nodes[0].node.get_our_node_id()); nodes[0].node.handle_funding_signed(&nodes[1].node.get_our_node_id(), &funding_signed); check_added_monitors!(nodes[0], 1); + expect_channel_pending_event(&nodes[0], &nodes[1].node.get_our_node_id()); } open_channel_msg.temporary_channel_id = nodes[0].keys_manager.get_secure_random_bytes(); } @@ -8826,7 +8858,7 @@ pub mod bench { use crate::chain::chainmonitor::{ChainMonitor, Persist}; use crate::chain::keysinterface::{EntropySource, KeysManager, InMemorySigner}; use crate::events::{Event, MessageSendEvent, MessageSendEventsProvider}; - use crate::ln::channelmanager::{self, BestBlock, ChainParameters, ChannelManager, PaymentHash, PaymentPreimage, PaymentId}; + use crate::ln::channelmanager::{BestBlock, ChainParameters, ChannelManager, PaymentHash, PaymentPreimage, PaymentId}; use crate::ln::functional_test_utils::*; use crate::ln::msgs::{ChannelMessageHandler, Init}; use crate::routing::gossip::NetworkGraph; @@ -8907,7 +8939,24 @@ pub mod bench { } 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())); + let events_b = node_b.get_and_clear_pending_events(); + assert_eq!(events_b.len(), 1); + match events_b[0] { + Event::ChannelPending{ ref counterparty_node_id, .. } => { + assert_eq!(*counterparty_node_id, node_a.get_our_node_id()); + }, + _ => panic!("Unexpected event"), + } + 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())); + let events_a = node_a.get_and_clear_pending_events(); + assert_eq!(events_a.len(), 1); + match events_a[0] { + Event::ChannelPending{ ref counterparty_node_id, .. } => { + assert_eq!(*counterparty_node_id, node_b.get_our_node_id()); + }, + _ => panic!("Unexpected event"), + } assert_eq!(&tx_broadcaster.txn_broadcasted.lock().unwrap()[..], &[tx.clone()]);