mirrors/rust-lightning
1
0
Fork 0
mirror of https://github.com/lightningdevkit/rust-lightning.git synced 2025-02-23 14:50:45 +01:00
Code Issues Projects Releases Packages Wiki Activity

Merge pull request #2035 from TheBlueMatt/2023-02-fix-no-con-discon

Browse source 
Fix (and DRY) the conditionals before calling peer_disconnected
This commit is contained in:
Matt Corallo 2023-02-21 21:28:05 +00:00 • committed by GitHub
commit e954ee8256
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
16 changed files with 250 additions and 222 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
fuzz/src/chanmon_consistency.rs
View file

@ -979,16 +979,16 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out) {
0x0c => { 0x0c => {
if !chan_a_disconnected { if !chan_a_disconnected {
nodes[0].peer_disconnected(&nodes[1].get_our_node_id(), false); nodes[0].peer_disconnected(&nodes[1].get_our_node_id());
nodes[1].peer_disconnected(&nodes[0].get_our_node_id(), false); nodes[1].peer_disconnected(&nodes[0].get_our_node_id());
chan_a_disconnected = true; chan_a_disconnected = true;
drain_msg_events_on_disconnect!(0); drain_msg_events_on_disconnect!(0);
} }
}, },
0x0d => { 0x0d => {
if !chan_b_disconnected { if !chan_b_disconnected {
nodes[1].peer_disconnected(&nodes[2].get_our_node_id(), false); nodes[1].peer_disconnected(&nodes[2].get_our_node_id());
nodes[2].peer_disconnected(&nodes[1].get_our_node_id(), false); nodes[2].peer_disconnected(&nodes[1].get_our_node_id());
chan_b_disconnected = true; chan_b_disconnected = true;
drain_msg_events_on_disconnect!(2); drain_msg_events_on_disconnect!(2);
} }
@ -1040,7 +1040,7 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out) {
0x2c => { 0x2c => {
if !chan_a_disconnected { if !chan_a_disconnected {
nodes[1].peer_disconnected(&nodes[0].get_our_node_id(), false); nodes[1].peer_disconnected(&nodes[0].get_our_node_id());
chan_a_disconnected = true; chan_a_disconnected = true;
drain_msg_events_on_disconnect!(0); drain_msg_events_on_disconnect!(0);
} }
@ -1054,14 +1054,14 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out) {
}, },
0x2d => { 0x2d => {
if !chan_a_disconnected { if !chan_a_disconnected {
nodes[0].peer_disconnected(&nodes[1].get_our_node_id(), false); nodes[0].peer_disconnected(&nodes[1].get_our_node_id());
chan_a_disconnected = true; chan_a_disconnected = true;
nodes[0].get_and_clear_pending_msg_events(); nodes[0].get_and_clear_pending_msg_events();
ab_events.clear(); ab_events.clear();
ba_events.clear(); ba_events.clear();
} }
if !chan_b_disconnected { if !chan_b_disconnected {
nodes[2].peer_disconnected(&nodes[1].get_our_node_id(), false); nodes[2].peer_disconnected(&nodes[1].get_our_node_id());
chan_b_disconnected = true; chan_b_disconnected = true;
nodes[2].get_and_clear_pending_msg_events(); nodes[2].get_and_clear_pending_msg_events();
bc_events.clear(); bc_events.clear();
@ -1073,7 +1073,7 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out) {
}, },
0x2e => { 0x2e => {
if !chan_b_disconnected { if !chan_b_disconnected {
nodes[1].peer_disconnected(&nodes[2].get_our_node_id(), false); nodes[1].peer_disconnected(&nodes[2].get_our_node_id());
chan_b_disconnected = true; chan_b_disconnected = true;
drain_msg_events_on_disconnect!(2); drain_msg_events_on_disconnect!(2);
} }

6
fuzz/src/full_stack.rs
View file

@ -634,11 +634,7 @@ pub fn do_test(data: &[u8], logger: &Arc<dyn Logger>) {
if let Err(e) = channelmanager.funding_transaction_generated(&funding_generation.0, &funding_generation.1, tx.clone()) { if let Err(e) = channelmanager.funding_transaction_generated(&funding_generation.0, &funding_generation.1, tx.clone()) {
// It's possible the channel has been closed in the mean time, but any other // It's possible the channel has been closed in the mean time, but any other
// failure may be a bug. // failure may be a bug.
if let APIError::ChannelUnavailable { err } = e { if let APIError::ChannelUnavailable { .. } = e { } else { panic!(); }
if !err.starts_with("Can't find a peer matching the passed counterparty node_id ") {
assert_eq!(err, "No such channel");
}
} else { panic!(); }
} }
pending_funding_signatures.insert(funding_output, tx); pending_funding_signatures.insert(funding_output, tx);
} }

4
lightning-invoice/src/utils.rs
View file

@ -842,13 +842,13 @@ mod test {
// With only one sufficient-value peer connected we should only get its hint // With only one sufficient-value peer connected we should only get its hint
scid_aliases.remove(&chan_b.0.short_channel_id_alias.unwrap()); scid_aliases.remove(&chan_b.0.short_channel_id_alias.unwrap());
nodes[0].node.peer_disconnected(&nodes[2].node.get_our_node_id(), false); nodes[0].node.peer_disconnected(&nodes[2].node.get_our_node_id());
match_invoice_routes(Some(1_000_000_000), &nodes[0], scid_aliases.clone()); match_invoice_routes(Some(1_000_000_000), &nodes[0], scid_aliases.clone());
// If we don't have any sufficient-value peers connected we should get all hints with // If we don't have any sufficient-value peers connected we should get all hints with
// sufficient value, even though there is a connected insufficient-value peer. // sufficient value, even though there is a connected insufficient-value peer.
scid_aliases.insert(chan_b.0.short_channel_id_alias.unwrap()); scid_aliases.insert(chan_b.0.short_channel_id_alias.unwrap());
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
match_invoice_routes(Some(1_000_000_000), &nodes[0], scid_aliases); match_invoice_routes(Some(1_000_000_000), &nodes[0], scid_aliases);
} }

2
lightning-net-tokio/src/lib.rs
View file

@ -643,7 +643,7 @@ mod tests {
fn handle_update_fee(&self, _their_node_id: &PublicKey, _msg: &UpdateFee) {} fn handle_update_fee(&self, _their_node_id: &PublicKey, _msg: &UpdateFee) {}
fn handle_announcement_signatures(&self, _their_node_id: &PublicKey, _msg: &AnnouncementSignatures) {} fn handle_announcement_signatures(&self, _their_node_id: &PublicKey, _msg: &AnnouncementSignatures) {}
fn handle_channel_update(&self, _their_node_id: &PublicKey, _msg: &ChannelUpdate) {} fn handle_channel_update(&self, _their_node_id: &PublicKey, _msg: &ChannelUpdate) {}
fn peer_disconnected(&self, their_node_id: &PublicKey, _no_connection_possible: bool) { fn peer_disconnected(&self, their_node_id: &PublicKey) {
if *their_node_id == self.expected_pubkey { if *their_node_id == self.expected_pubkey {
self.disconnected_flag.store(true, Ordering::SeqCst); self.disconnected_flag.store(true, Ordering::SeqCst);
self.pubkey_disconnected.clone().try_send(()).unwrap(); self.pubkey_disconnected.clone().try_send(()).unwrap();

52
lightning/src/ln/chanmon_update_fail_tests.rs
View file

@ -181,8 +181,8 @@ fn do_test_simple_monitor_temporary_update_fail(disconnect: bool) {
assert_eq!(nodes[0].node.list_channels().len(), 1); assert_eq!(nodes[0].node.list_channels().len(), 1);
if disconnect { if disconnect {
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
reconnect_nodes(&nodes[0], &nodes[1], (true, true), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (false, false)); reconnect_nodes(&nodes[0], &nodes[1], (true, true), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (false, false));
} }
@ -234,8 +234,8 @@ fn do_test_simple_monitor_temporary_update_fail(disconnect: bool) {
assert_eq!(nodes[0].node.list_channels().len(), 1); assert_eq!(nodes[0].node.list_channels().len(), 1);
if disconnect { if disconnect {
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
reconnect_nodes(&nodes[0], &nodes[1], (false, false), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (false, false)); reconnect_nodes(&nodes[0], &nodes[1], (false, false), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (false, false));
} }
@ -337,8 +337,8 @@ fn do_test_monitor_temporary_update_fail(disconnect_count: usize) {
}; };
if disconnect_count & !disconnect_flags > 0 { if disconnect_count & !disconnect_flags > 0 {
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
} }
// Now fix monitor updating... // Now fix monitor updating...
@ -348,8 +348,8 @@ fn do_test_monitor_temporary_update_fail(disconnect_count: usize) {
check_added_monitors!(nodes[0], 0); check_added_monitors!(nodes[0], 0);
macro_rules! disconnect_reconnect_peers { () => { { macro_rules! disconnect_reconnect_peers { () => { {
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
nodes[0].node.peer_connected(&nodes[1].node.get_our_node_id(), &msgs::Init { features: nodes[1].node.init_features(), remote_network_address: None }).unwrap(); nodes[0].node.peer_connected(&nodes[1].node.get_our_node_id(), &msgs::Init { features: nodes[1].node.init_features(), remote_network_address: None }).unwrap();
let reestablish_1 = get_chan_reestablish_msgs!(nodes[0], nodes[1]); let reestablish_1 = get_chan_reestablish_msgs!(nodes[0], nodes[1]);
@ -1110,8 +1110,8 @@ fn test_monitor_update_fail_reestablish() {
let (payment_preimage, payment_hash, _) = route_payment(&nodes[0], &[&nodes[1], &nodes[2]], 1_000_000); let (payment_preimage, payment_hash, _) = route_payment(&nodes[0], &[&nodes[1], &nodes[2]], 1_000_000);
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
nodes[2].node.claim_funds(payment_preimage); nodes[2].node.claim_funds(payment_preimage);
check_added_monitors!(nodes[2], 1); check_added_monitors!(nodes[2], 1);
@ -1146,8 +1146,8 @@ fn test_monitor_update_fail_reestablish() {
nodes[1].node.get_and_clear_pending_msg_events(); // Free the holding cell nodes[1].node.get_and_clear_pending_msg_events(); // Free the holding cell
check_added_monitors!(nodes[1], 1); check_added_monitors!(nodes[1], 1);
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
nodes[0].node.peer_connected(&nodes[1].node.get_our_node_id(), &msgs::Init { features: nodes[1].node.init_features(), remote_network_address: None }).unwrap(); nodes[0].node.peer_connected(&nodes[1].node.get_our_node_id(), &msgs::Init { features: nodes[1].node.init_features(), remote_network_address: None }).unwrap();
nodes[1].node.peer_connected(&nodes[0].node.get_our_node_id(), &msgs::Init { features: nodes[0].node.init_features(), remote_network_address: None }).unwrap(); nodes[1].node.peer_connected(&nodes[0].node.get_our_node_id(), &msgs::Init { features: nodes[0].node.init_features(), remote_network_address: None }).unwrap();
@ -1315,8 +1315,8 @@ fn claim_while_disconnected_monitor_update_fail() {
// Forward a payment for B to claim // Forward a payment for B to claim
let (payment_preimage_1, payment_hash_1, _) = 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);
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
nodes[1].node.claim_funds(payment_preimage_1); nodes[1].node.claim_funds(payment_preimage_1);
check_added_monitors!(nodes[1], 1); check_added_monitors!(nodes[1], 1);
@ -1451,8 +1451,8 @@ fn monitor_failed_no_reestablish_response() {
// Now disconnect and immediately reconnect, delivering the channel_reestablish while nodes[1] // Now disconnect and immediately reconnect, delivering the channel_reestablish while nodes[1]
// is still failing to update monitors. // is still failing to update monitors.
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
nodes[0].node.peer_connected(&nodes[1].node.get_our_node_id(), &msgs::Init { features: nodes[1].node.init_features(), remote_network_address: None }).unwrap(); nodes[0].node.peer_connected(&nodes[1].node.get_our_node_id(), &msgs::Init { features: nodes[1].node.init_features(), remote_network_address: None }).unwrap();
nodes[1].node.peer_connected(&nodes[0].node.get_our_node_id(), &msgs::Init { features: nodes[0].node.init_features(), remote_network_address: None }).unwrap(); nodes[1].node.peer_connected(&nodes[0].node.get_our_node_id(), &msgs::Init { features: nodes[0].node.init_features(), remote_network_address: None }).unwrap();
@ -1879,8 +1879,8 @@ fn do_during_funding_monitor_fail(confirm_a_first: bool, restore_b_before_conf:
} }
// Make sure nodes[1] isn't stupid enough to re-send the ChannelReady on reconnect // Make sure nodes[1] isn't stupid enough to re-send the ChannelReady on reconnect
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
reconnect_nodes(&nodes[0], &nodes[1], (false, confirm_a_first), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (false, false)); reconnect_nodes(&nodes[0], &nodes[1], (false, confirm_a_first), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (false, false));
assert!(nodes[0].node.get_and_clear_pending_msg_events().is_empty()); assert!(nodes[0].node.get_and_clear_pending_msg_events().is_empty());
assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty()); assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
@ -2047,8 +2047,8 @@ fn test_pending_update_fee_ack_on_reconnect() {
let bs_first_raa = get_event_msg!(nodes[1], MessageSendEvent::SendRevokeAndACK, nodes[0].node.get_our_node_id()); let bs_first_raa = get_event_msg!(nodes[1], MessageSendEvent::SendRevokeAndACK, nodes[0].node.get_our_node_id());
// bs_first_raa is not delivered until it is re-generated after reconnect // bs_first_raa is not delivered until it is re-generated after reconnect
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
nodes[0].node.peer_connected(&nodes[1].node.get_our_node_id(), &msgs::Init { features: nodes[1].node.init_features(), remote_network_address: None }).unwrap(); nodes[0].node.peer_connected(&nodes[1].node.get_our_node_id(), &msgs::Init { features: nodes[1].node.init_features(), remote_network_address: None }).unwrap();
let as_connect_msg = get_chan_reestablish_msgs!(nodes[0], nodes[1]).pop().unwrap(); let as_connect_msg = get_chan_reestablish_msgs!(nodes[0], nodes[1]).pop().unwrap();
@ -2175,8 +2175,8 @@ fn do_update_fee_resend_test(deliver_update: bool, parallel_updates: bool) {
assert!(nodes[0].node.get_and_clear_pending_msg_events().is_empty()); assert!(nodes[0].node.get_and_clear_pending_msg_events().is_empty());
} }
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
nodes[0].node.peer_connected(&nodes[1].node.get_our_node_id(), &msgs::Init { features: nodes[1].node.init_features(), remote_network_address: None }).unwrap(); nodes[0].node.peer_connected(&nodes[1].node.get_our_node_id(), &msgs::Init { features: nodes[1].node.init_features(), remote_network_address: None }).unwrap();
let as_connect_msg = get_chan_reestablish_msgs!(nodes[0], nodes[1]).pop().unwrap(); let as_connect_msg = get_chan_reestablish_msgs!(nodes[0], nodes[1]).pop().unwrap();
@ -2309,9 +2309,9 @@ fn do_channel_holding_cell_serialize(disconnect: bool, reload_a: bool) {
let chan_0_monitor_serialized = get_monitor!(nodes[0], chan_id).encode(); let chan_0_monitor_serialized = get_monitor!(nodes[0], chan_id).encode();
reload_node!(nodes[0], &nodes[0].node.encode(), &[&chan_0_monitor_serialized], persister, new_chain_monitor, nodes_0_deserialized); reload_node!(nodes[0], &nodes[0].node.encode(), &[&chan_0_monitor_serialized], persister, new_chain_monitor, nodes_0_deserialized);
} else { } else {
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
} }
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
// Now reconnect the two // Now reconnect the two
nodes[0].node.peer_connected(&nodes[1].node.get_our_node_id(), &msgs::Init { features: nodes[1].node.init_features(), remote_network_address: None }).unwrap(); nodes[0].node.peer_connected(&nodes[1].node.get_our_node_id(), &msgs::Init { features: nodes[1].node.init_features(), remote_network_address: None }).unwrap();
@ -2499,8 +2499,8 @@ fn do_test_reconnect_dup_htlc_claims(htlc_status: HTLCStatusAtDupClaim, second_f
assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty()); assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
} }
nodes[1].node.peer_disconnected(&nodes[2].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[2].node.get_our_node_id());
nodes[2].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[2].node.peer_disconnected(&nodes[1].node.get_our_node_id());
if second_fails { if second_fails {
reconnect_nodes(&nodes[1], &nodes[2], (false, false), (0, 0), (0, 0), (1, 0), (0, 0), (0, 0), (false, false)); reconnect_nodes(&nodes[1], &nodes[2], (false, false), (0, 0), (0, 0), (1, 0), (0, 0), (0, 0), (false, false));

16
lightning/src/ln/channelmanager.rs
View file

@ -2666,7 +2666,7 @@ where
(chan, funding_msg) (chan, funding_msg)
}, },
Err(_) => { return Err(APIError::ChannelUnavailable { Err(_) => { return Err(APIError::ChannelUnavailable {
err: "Error deriving keys or signing initial commitment transactions - either our RNG or our counterparty's RNG is broken or the Signer refused to sign".to_owned() err: "Signer refused to sign the initial commitment transaction".to_owned()
}) }, }) },
} }
}; };
@ -6245,13 +6245,13 @@ where
let _ = handle_error!(self, self.internal_channel_reestablish(counterparty_node_id, msg), *counterparty_node_id); let _ = handle_error!(self, self.internal_channel_reestablish(counterparty_node_id, msg), *counterparty_node_id);
} }
fn peer_disconnected(&self, counterparty_node_id: &PublicKey, no_connection_possible: bool) { fn peer_disconnected(&self, counterparty_node_id: &PublicKey) {
let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier); let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
let mut failed_channels = Vec::new(); let mut failed_channels = Vec::new();
let mut per_peer_state = self.per_peer_state.write().unwrap(); let mut per_peer_state = self.per_peer_state.write().unwrap();
let remove_peer = { let remove_peer = {
log_debug!(self.logger, "Marking channels with {} disconnected and generating channel_updates. We believe we {} make future connections to this peer.", log_debug!(self.logger, "Marking channels with {} disconnected and generating channel_updates.",
log_pubkey!(counterparty_node_id), if no_connection_possible { "cannot" } else { "can" }); log_pubkey!(counterparty_node_id));
if let Some(peer_state_mutex) = per_peer_state.get(counterparty_node_id) { if let Some(peer_state_mutex) = per_peer_state.get(counterparty_node_id) {
let mut peer_state_lock = peer_state_mutex.lock().unwrap(); let mut peer_state_lock = peer_state_mutex.lock().unwrap();
let peer_state = &mut *peer_state_lock; let peer_state = &mut *peer_state_lock;
@ -6293,7 +6293,7 @@ where
debug_assert!(peer_state.is_connected, "A disconnected peer cannot disconnect"); debug_assert!(peer_state.is_connected, "A disconnected peer cannot disconnect");
peer_state.is_connected = false; peer_state.is_connected = false;
peer_state.ok_to_remove(true) peer_state.ok_to_remove(true)
} else { true } } else { debug_assert!(false, "Unconnected peer disconnected"); true }
}; };
if remove_peer { if remove_peer {
per_peer_state.remove(counterparty_node_id); per_peer_state.remove(counterparty_node_id);
@ -6307,7 +6307,7 @@ where
fn peer_connected(&self, counterparty_node_id: &PublicKey, init_msg: &msgs::Init) -> Result<(), ()> { fn peer_connected(&self, counterparty_node_id: &PublicKey, init_msg: &msgs::Init) -> Result<(), ()> {
if !init_msg.features.supports_static_remote_key() { if !init_msg.features.supports_static_remote_key() {
log_debug!(self.logger, "Peer {} does not support static remote key, disconnecting with no_connection_possible", log_pubkey!(counterparty_node_id)); log_debug!(self.logger, "Peer {} does not support static remote key, disconnecting", log_pubkey!(counterparty_node_id));
return Err(()); return Err(());
} }
@ -8186,8 +8186,8 @@ mod tests {
let chan = create_announced_chan_between_nodes(&nodes, 0, 1); let chan = create_announced_chan_between_nodes(&nodes, 0, 1);
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
nodes[0].node.force_close_broadcasting_latest_txn(&chan.2, &nodes[1].node.get_our_node_id()).unwrap(); nodes[0].node.force_close_broadcasting_latest_txn(&chan.2, &nodes[1].node.get_our_node_id()).unwrap();
check_closed_broadcast!(nodes[0], true); check_closed_broadcast!(nodes[0], true);

58
lightning/src/ln/functional_tests.rs
View file

@ -3524,8 +3524,8 @@ fn test_dup_events_on_peer_disconnect() {
nodes[0].node.handle_update_fulfill_htlc(&nodes[1].node.get_our_node_id(), &claim_msgs.update_fulfill_htlcs[0]); nodes[0].node.handle_update_fulfill_htlc(&nodes[1].node.get_our_node_id(), &claim_msgs.update_fulfill_htlcs[0]);
expect_payment_sent_without_paths!(nodes[0], payment_preimage); expect_payment_sent_without_paths!(nodes[0], payment_preimage);
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
reconnect_nodes(&nodes[0], &nodes[1], (false, false), (0, 0), (1, 0), (0, 0), (0, 0), (0, 0), (false, false)); reconnect_nodes(&nodes[0], &nodes[1], (false, false), (0, 0), (1, 0), (0, 0), (0, 0), (0, 0), (false, false));
expect_payment_path_successful!(nodes[0]); expect_payment_path_successful!(nodes[0]);
@ -3565,8 +3565,8 @@ fn test_peer_disconnected_before_funding_broadcasted() {
// Ensure that the channel is closed with `ClosureReason::DisconnectedPeer` when the peers are // Ensure that the channel is closed with `ClosureReason::DisconnectedPeer` when the peers are
// disconnected before the funding transaction was broadcasted. // disconnected before the funding transaction was broadcasted.
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
check_closed_event!(nodes[0], 1, ClosureReason::DisconnectedPeer); check_closed_event!(nodes[0], 1, ClosureReason::DisconnectedPeer);
check_closed_event!(nodes[1], 1, ClosureReason::DisconnectedPeer); check_closed_event!(nodes[1], 1, ClosureReason::DisconnectedPeer);
@ -3582,8 +3582,8 @@ fn test_simple_peer_disconnect() {
create_announced_chan_between_nodes(&nodes, 0, 1); create_announced_chan_between_nodes(&nodes, 0, 1);
create_announced_chan_between_nodes(&nodes, 1, 2); create_announced_chan_between_nodes(&nodes, 1, 2);
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
reconnect_nodes(&nodes[0], &nodes[1], (true, true), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (false, false)); reconnect_nodes(&nodes[0], &nodes[1], (true, true), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (false, false));
let payment_preimage_1 = route_payment(&nodes[0], &vec!(&nodes[1], &nodes[2])[..], 1000000).0; let payment_preimage_1 = route_payment(&nodes[0], &vec!(&nodes[1], &nodes[2])[..], 1000000).0;
@ -3591,8 +3591,8 @@ fn test_simple_peer_disconnect() {
fail_payment(&nodes[0], &vec!(&nodes[1], &nodes[2]), payment_hash_2); fail_payment(&nodes[0], &vec!(&nodes[1], &nodes[2]), payment_hash_2);
claim_payment(&nodes[0], &vec!(&nodes[1], &nodes[2]), payment_preimage_1); claim_payment(&nodes[0], &vec!(&nodes[1], &nodes[2]), payment_preimage_1);
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
reconnect_nodes(&nodes[0], &nodes[1], (false, false), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (false, false)); reconnect_nodes(&nodes[0], &nodes[1], (false, false), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (false, false));
let (payment_preimage_3, payment_hash_3, _) = route_payment(&nodes[0], &vec!(&nodes[1], &nodes[2])[..], 1000000); let (payment_preimage_3, payment_hash_3, _) = route_payment(&nodes[0], &vec!(&nodes[1], &nodes[2])[..], 1000000);
@ -3600,8 +3600,8 @@ fn test_simple_peer_disconnect() {
let payment_hash_5 = route_payment(&nodes[0], &vec!(&nodes[1], &nodes[2])[..], 1000000).1; let payment_hash_5 = route_payment(&nodes[0], &vec!(&nodes[1], &nodes[2])[..], 1000000).1;
let payment_hash_6 = route_payment(&nodes[0], &vec!(&nodes[1], &nodes[2])[..], 1000000).1; let payment_hash_6 = route_payment(&nodes[0], &vec!(&nodes[1], &nodes[2])[..], 1000000).1;
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
claim_payment_along_route(&nodes[0], &[&[&nodes[1], &nodes[2]]], true, payment_preimage_3); claim_payment_along_route(&nodes[0], &[&[&nodes[1], &nodes[2]]], true, payment_preimage_3);
fail_payment_along_route(&nodes[0], &[&[&nodes[1], &nodes[2]]], true, payment_hash_5); fail_payment_along_route(&nodes[0], &[&[&nodes[1], &nodes[2]]], true, payment_hash_5);
@ -3701,8 +3701,8 @@ fn do_test_drop_messages_peer_disconnect(messages_delivered: u8, simulate_broken
} }
} }
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
if messages_delivered < 3 { if messages_delivered < 3 {
if simulate_broken_lnd { if simulate_broken_lnd {
// lnd has a long-standing bug where they send a channel_ready prior to a // lnd has a long-standing bug where they send a channel_ready prior to a
@ -3751,8 +3751,8 @@ fn do_test_drop_messages_peer_disconnect(messages_delivered: u8, simulate_broken
}; };
} }
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
reconnect_nodes(&nodes[0], &nodes[1], (false, false), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (false, false)); reconnect_nodes(&nodes[0], &nodes[1], (false, false), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (false, false));
nodes[1].node.process_pending_htlc_forwards(); nodes[1].node.process_pending_htlc_forwards();
@ -3834,8 +3834,8 @@ fn do_test_drop_messages_peer_disconnect(messages_delivered: u8, simulate_broken
} }
} }
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
if messages_delivered < 2 { if messages_delivered < 2 {
reconnect_nodes(&nodes[0], &nodes[1], (false, false), (0, 0), (1, 0), (0, 0), (0, 0), (0, 0), (false, false)); reconnect_nodes(&nodes[0], &nodes[1], (false, false), (0, 0), (1, 0), (0, 0), (0, 0), (0, 0), (false, false));
if messages_delivered < 1 { if messages_delivered < 1 {
@ -3861,8 +3861,8 @@ fn do_test_drop_messages_peer_disconnect(messages_delivered: u8, simulate_broken
expect_payment_path_successful!(nodes[0]); expect_payment_path_successful!(nodes[0]);
} }
if messages_delivered <= 5 { if messages_delivered <= 5 {
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
} }
reconnect_nodes(&nodes[0], &nodes[1], (false, false), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (false, false)); reconnect_nodes(&nodes[0], &nodes[1], (false, false), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (false, false));
@ -3976,8 +3976,8 @@ fn test_drop_messages_peer_disconnect_dual_htlc() {
_ => panic!("Unexpected event"), _ => panic!("Unexpected event"),
} }
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
nodes[0].node.peer_connected(&nodes[1].node.get_our_node_id(), &msgs::Init { features: nodes[1].node.init_features(), remote_network_address: None }).unwrap(); nodes[0].node.peer_connected(&nodes[1].node.get_our_node_id(), &msgs::Init { features: nodes[1].node.init_features(), remote_network_address: None }).unwrap();
let reestablish_1 = get_chan_reestablish_msgs!(nodes[0], nodes[1]); let reestablish_1 = get_chan_reestablish_msgs!(nodes[0], nodes[1]);
@ -6292,8 +6292,8 @@ fn test_update_add_htlc_bolt2_receiver_check_repeated_id_ignore() {
nodes[1].node.handle_update_add_htlc(&nodes[0].node.get_our_node_id(), &updates.update_add_htlcs[0]); nodes[1].node.handle_update_add_htlc(&nodes[0].node.get_our_node_id(), &updates.update_add_htlcs[0]);
//Disconnect and Reconnect //Disconnect and Reconnect
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
nodes[0].node.peer_connected(&nodes[1].node.get_our_node_id(), &msgs::Init { features: nodes[1].node.init_features(), remote_network_address: None }).unwrap(); nodes[0].node.peer_connected(&nodes[1].node.get_our_node_id(), &msgs::Init { features: nodes[1].node.init_features(), remote_network_address: None }).unwrap();
let reestablish_1 = get_chan_reestablish_msgs!(nodes[0], nodes[1]); let reestablish_1 = get_chan_reestablish_msgs!(nodes[0], nodes[1]);
assert_eq!(reestablish_1.len(), 1); assert_eq!(reestablish_1.len(), 1);
@ -7032,8 +7032,8 @@ fn test_announce_disable_channels() {
create_announced_chan_between_nodes(&nodes, 0, 1); create_announced_chan_between_nodes(&nodes, 0, 1);
// Disconnect peers // Disconnect peers
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
nodes[0].node.timer_tick_occurred(); // Enabled -> DisabledStaged nodes[0].node.timer_tick_occurred(); // Enabled -> DisabledStaged
nodes[0].node.timer_tick_occurred(); // DisabledStaged -> Disabled nodes[0].node.timer_tick_occurred(); // DisabledStaged -> Disabled
@ -8832,13 +8832,11 @@ fn test_error_chans_closed() {
_ => panic!("Unexpected event"), _ => panic!("Unexpected event"),
} }
// Note that at this point users of a standard PeerHandler will end up calling // Note that at this point users of a standard PeerHandler will end up calling
// peer_disconnected with no_connection_possible set to false, duplicating the // peer_disconnected.
// close-all-channels logic. That's OK, we don't want to end up not force-closing channels for
// users with their own peer handling logic. We duplicate the call here, however.
assert_eq!(nodes[0].node.list_usable_channels().len(), 1); assert_eq!(nodes[0].node.list_usable_channels().len(), 1);
assert!(nodes[0].node.list_usable_channels()[0].channel_id == chan_3.2); assert!(nodes[0].node.list_usable_channels()[0].channel_id == chan_3.2);
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), true); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
assert_eq!(nodes[0].node.list_usable_channels().len(), 1); assert_eq!(nodes[0].node.list_usable_channels().len(), 1);
assert!(nodes[0].node.list_usable_channels()[0].channel_id == chan_3.2); assert!(nodes[0].node.list_usable_channels()[0].channel_id == chan_3.2);
} }
@ -8954,8 +8952,8 @@ fn do_test_tx_confirmed_skipping_blocks_immediate_broadcast(test_height_before_t
create_announced_chan_between_nodes(&nodes, 0, 1); create_announced_chan_between_nodes(&nodes, 0, 1);
let (chan_announce, _, channel_id, _) = create_announced_chan_between_nodes(&nodes, 1, 2); let (chan_announce, _, channel_id, _) = create_announced_chan_between_nodes(&nodes, 1, 2);
let (_, payment_hash, _) = route_payment(&nodes[0], &[&nodes[1], &nodes[2]], 1_000_000); let (_, payment_hash, _) = route_payment(&nodes[0], &[&nodes[1], &nodes[2]], 1_000_000);
nodes[1].node.peer_disconnected(&nodes[2].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[2].node.get_our_node_id());
nodes[2].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[2].node.peer_disconnected(&nodes[1].node.get_our_node_id());
nodes[1].node.force_close_broadcasting_latest_txn(&channel_id, &nodes[2].node.get_our_node_id()).unwrap(); nodes[1].node.force_close_broadcasting_latest_txn(&channel_id, &nodes[2].node.get_our_node_id()).unwrap();
check_closed_broadcast!(nodes[1], true); check_closed_broadcast!(nodes[1], true);

15
lightning/src/ln/msgs.rs
View file

@ -993,14 +993,8 @@ pub trait ChannelMessageHandler : MessageSendEventsProvider {
fn handle_announcement_signatures(&self, their_node_id: &PublicKey, msg: &AnnouncementSignatures); fn handle_announcement_signatures(&self, their_node_id: &PublicKey, msg: &AnnouncementSignatures);
// Connection loss/reestablish: // Connection loss/reestablish:
/// Indicates a connection to the peer failed/an existing connection was lost. If no connection /// Indicates a connection to the peer failed/an existing connection was lost.
/// is believed to be possible in the future (eg they're sending us messages we don't fn peer_disconnected(&self, their_node_id: &PublicKey);
/// understand or indicate they require unknown feature bits), `no_connection_possible` is set
/// and any outstanding channels should be failed.
///
/// Note that in some rare cases this may be called without a corresponding
/// [`Self::peer_connected`].
fn peer_disconnected(&self, their_node_id: &PublicKey, no_connection_possible: bool);
/// Handle a peer reconnecting, possibly generating `channel_reestablish` message(s). /// Handle a peer reconnecting, possibly generating `channel_reestablish` message(s).
/// ///
@ -1115,10 +1109,7 @@ pub trait OnionMessageHandler : OnionMessageProvider {
fn peer_connected(&self, their_node_id: &PublicKey, init: &Init) -> Result<(), ()>; fn peer_connected(&self, their_node_id: &PublicKey, init: &Init) -> Result<(), ()>;
/// Indicates a connection to the peer failed/an existing connection was lost. Allows handlers to /// Indicates a connection to the peer failed/an existing connection was lost. Allows handlers to
/// drop and refuse to forward onion messages to this peer. /// drop and refuse to forward onion messages to this peer.
/// fn peer_disconnected(&self, their_node_id: &PublicKey);
/// Note that in some rare cases this may be called without a corresponding
/// [`Self::peer_connected`].
fn peer_disconnected(&self, their_node_id: &PublicKey, no_connection_possible: bool);
// Handler information: // Handler information:
/// Gets the node feature flags which this handler itself supports. All available handlers are /// Gets the node feature flags which this handler itself supports. All available handlers are

4
lightning/src/ln/onion_route_tests.rs
View file

@ -576,8 +576,8 @@ fn test_onion_failure() {
let short_channel_id = channels[1].0.contents.short_channel_id; let short_channel_id = channels[1].0.contents.short_channel_id;
run_onion_failure_test("channel_disabled", 0, &nodes, &route, &payment_hash, &payment_secret, |_| {}, || { run_onion_failure_test("channel_disabled", 0, &nodes, &route, &payment_hash, &payment_secret, |_| {}, || {
// disconnect event to the channel between nodes[1] ~ nodes[2] // disconnect event to the channel between nodes[1] ~ nodes[2]
nodes[1].node.peer_disconnected(&nodes[2].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[2].node.get_our_node_id());
nodes[2].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[2].node.peer_disconnected(&nodes[1].node.get_our_node_id());
}, true, Some(UPDATE|20), Some(NetworkUpdate::ChannelUpdateMessage{msg: ChannelUpdate::dummy(short_channel_id)}), Some(short_channel_id)); }, true, Some(UPDATE|20), Some(NetworkUpdate::ChannelUpdateMessage{msg: ChannelUpdate::dummy(short_channel_id)}), Some(short_channel_id));
reconnect_nodes(&nodes[1], &nodes[2], (false, false), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (false, false)); reconnect_nodes(&nodes[1], &nodes[2], (false, false), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (false, false));

22
lightning/src/ln/payment_tests.rs
View file

@ -253,8 +253,8 @@ fn no_pending_leak_on_initial_send_failure() {
let (route, payment_hash, _, payment_secret) = get_route_and_payment_hash!(nodes[0], nodes[1], 100_000); let (route, payment_hash, _, payment_secret) = get_route_and_payment_hash!(nodes[0], nodes[1], 100_000);
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
nodes[1].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
unwrap_send_err!(nodes[0].node.send_payment(&route, payment_hash, &Some(payment_secret), PaymentId(payment_hash.0)), unwrap_send_err!(nodes[0].node.send_payment(&route, payment_hash, &Some(payment_secret), PaymentId(payment_hash.0)),
true, APIError::ChannelUnavailable { ref err }, true, APIError::ChannelUnavailable { ref err },
@ -310,8 +310,8 @@ fn do_retry_with_no_persist(confirm_before_reload: bool) {
// We relay the payment to nodes[1] while its disconnected from nodes[2], causing the payment // We relay the payment to nodes[1] while its disconnected from nodes[2], causing the payment
// to be returned immediately to nodes[0], without having nodes[2] fail the inbound payment // to be returned immediately to nodes[0], without having nodes[2] fail the inbound payment
// which would prevent retry. // which would prevent retry.
nodes[1].node.peer_disconnected(&nodes[2].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[2].node.get_our_node_id());
nodes[2].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[2].node.peer_disconnected(&nodes[1].node.get_our_node_id());
nodes[1].node.handle_update_add_htlc(&nodes[0].node.get_our_node_id(), &payment_event.msgs[0]); nodes[1].node.handle_update_add_htlc(&nodes[0].node.get_our_node_id(), &payment_event.msgs[0]);
commitment_signed_dance!(nodes[1], nodes[0], payment_event.commitment_msg, false, true); commitment_signed_dance!(nodes[1], nodes[0], payment_event.commitment_msg, false, true);
@ -340,7 +340,7 @@ fn do_retry_with_no_persist(confirm_before_reload: bool) {
assert_eq!(as_broadcasted_txn.len(), 1); assert_eq!(as_broadcasted_txn.len(), 1);
assert_eq!(as_broadcasted_txn[0], as_commitment_tx); assert_eq!(as_broadcasted_txn[0], as_commitment_tx);
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
nodes[0].node.peer_connected(&nodes[1].node.get_our_node_id(), &msgs::Init { features: nodes[1].node.init_features(), remote_network_address: None }).unwrap(); nodes[0].node.peer_connected(&nodes[1].node.get_our_node_id(), &msgs::Init { features: nodes[1].node.init_features(), remote_network_address: None }).unwrap();
assert!(nodes[0].node.get_and_clear_pending_msg_events().is_empty()); assert!(nodes[0].node.get_and_clear_pending_msg_events().is_empty());
@ -496,7 +496,7 @@ fn do_test_completed_payment_not_retryable_on_reload(use_dust: bool) {
let chan_0_monitor_serialized = get_monitor!(nodes[0], chan_id).encode(); let chan_0_monitor_serialized = get_monitor!(nodes[0], chan_id).encode();
reload_node!(nodes[0], test_default_channel_config(), nodes_0_serialized, &[&chan_0_monitor_serialized], first_persister, first_new_chain_monitor, first_nodes_0_deserialized); reload_node!(nodes[0], test_default_channel_config(), nodes_0_serialized, &[&chan_0_monitor_serialized], first_persister, first_new_chain_monitor, first_nodes_0_deserialized);
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
// On reload, the ChannelManager should realize it is stale compared to the ChannelMonitor and // On reload, the ChannelManager should realize it is stale compared to the ChannelMonitor and
// force-close the channel. // force-close the channel.
@ -591,7 +591,7 @@ fn do_test_completed_payment_not_retryable_on_reload(use_dust: bool) {
assert!(!nodes[0].node.get_and_clear_pending_msg_events().is_empty()); assert!(!nodes[0].node.get_and_clear_pending_msg_events().is_empty());
reload_node!(nodes[0], test_default_channel_config(), nodes_0_serialized, &[&chan_0_monitor_serialized, &chan_1_monitor_serialized], second_persister, second_new_chain_monitor, second_nodes_0_deserialized); reload_node!(nodes[0], test_default_channel_config(), nodes_0_serialized, &[&chan_0_monitor_serialized, &chan_1_monitor_serialized], second_persister, second_new_chain_monitor, second_nodes_0_deserialized);
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
reconnect_nodes(&nodes[0], &nodes[1], (true, true), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (false, false)); reconnect_nodes(&nodes[0], &nodes[1], (true, true), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (false, false));
@ -615,7 +615,7 @@ fn do_test_completed_payment_not_retryable_on_reload(use_dust: bool) {
// Check that after reload we can send the payment again (though we shouldn't, since it was // Check that after reload we can send the payment again (though we shouldn't, since it was
// claimed previously). // claimed previously).
reload_node!(nodes[0], test_default_channel_config(), nodes_0_serialized, &[&chan_0_monitor_serialized, &chan_1_monitor_serialized], third_persister, third_new_chain_monitor, third_nodes_0_deserialized); reload_node!(nodes[0], test_default_channel_config(), nodes_0_serialized, &[&chan_0_monitor_serialized, &chan_1_monitor_serialized], third_persister, third_new_chain_monitor, third_nodes_0_deserialized);
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
reconnect_nodes(&nodes[0], &nodes[1], (false, false), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (false, false)); reconnect_nodes(&nodes[0], &nodes[1], (false, false), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (false, false));
@ -660,8 +660,8 @@ fn do_test_dup_htlc_onchain_fails_on_reload(persist_manager_post_event: bool, co
check_added_monitors!(nodes[0], 1); check_added_monitors!(nodes[0], 1);
check_closed_event!(nodes[0], 1, ClosureReason::HolderForceClosed); check_closed_event!(nodes[0], 1, ClosureReason::HolderForceClosed);
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
// Connect blocks until the CLTV timeout is up so that we get an HTLC-Timeout transaction // Connect blocks until the CLTV timeout is up so that we get an HTLC-Timeout transaction
connect_blocks(&nodes[0], TEST_FINAL_CLTV + LATENCY_GRACE_PERIOD_BLOCKS + 1); connect_blocks(&nodes[0], TEST_FINAL_CLTV + LATENCY_GRACE_PERIOD_BLOCKS + 1);
@ -812,7 +812,7 @@ fn test_fulfill_restart_failure() {
// Now reload nodes[1]... // Now reload nodes[1]...
reload_node!(nodes[1], &chan_manager_serialized, &[&chan_0_monitor_serialized], persister, new_chain_monitor, nodes_1_deserialized); reload_node!(nodes[1], &chan_manager_serialized, &[&chan_0_monitor_serialized], persister, new_chain_monitor, nodes_1_deserialized);
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
reconnect_nodes(&nodes[0], &nodes[1], (false, false), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (false, false)); reconnect_nodes(&nodes[0], &nodes[1], (false, false), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (false, false));
nodes[1].node.fail_htlc_backwards(&payment_hash); nodes[1].node.fail_htlc_backwards(&payment_hash);

216
lightning/src/ln/peer_handler.rs
View file

@ -96,7 +96,7 @@ impl OnionMessageProvider for IgnoringMessageHandler {
impl OnionMessageHandler for IgnoringMessageHandler { impl OnionMessageHandler for IgnoringMessageHandler {
fn handle_onion_message(&self, _their_node_id: &PublicKey, _msg: &msgs::OnionMessage) {} fn handle_onion_message(&self, _their_node_id: &PublicKey, _msg: &msgs::OnionMessage) {}
fn peer_connected(&self, _their_node_id: &PublicKey, _init: &msgs::Init) -> Result<(), ()> { Ok(()) } fn peer_connected(&self, _their_node_id: &PublicKey, _init: &msgs::Init) -> Result<(), ()> { Ok(()) }
fn peer_disconnected(&self, _their_node_id: &PublicKey, _no_connection_possible: bool) {} fn peer_disconnected(&self, _their_node_id: &PublicKey) {}
fn provided_node_features(&self) -> NodeFeatures { NodeFeatures::empty() } fn provided_node_features(&self) -> NodeFeatures { NodeFeatures::empty() }
fn provided_init_features(&self, _their_node_id: &PublicKey) -> InitFeatures { fn provided_init_features(&self, _their_node_id: &PublicKey) -> InitFeatures {
InitFeatures::empty() InitFeatures::empty()
@ -230,7 +230,7 @@ impl ChannelMessageHandler for ErroringMessageHandler {
} }
// msgs::ChannelUpdate does not contain the channel_id field, so we just drop them. // msgs::ChannelUpdate does not contain the channel_id field, so we just drop them.
fn handle_channel_update(&self, _their_node_id: &PublicKey, _msg: &msgs::ChannelUpdate) {} fn handle_channel_update(&self, _their_node_id: &PublicKey, _msg: &msgs::ChannelUpdate) {}
fn peer_disconnected(&self, _their_node_id: &PublicKey, _no_connection_possible: bool) {} fn peer_disconnected(&self, _their_node_id: &PublicKey) {}
fn peer_connected(&self, _their_node_id: &PublicKey, _init: &msgs::Init) -> Result<(), ()> { Ok(()) } fn peer_connected(&self, _their_node_id: &PublicKey, _init: &msgs::Init) -> Result<(), ()> { Ok(()) }
fn handle_error(&self, _their_node_id: &PublicKey, _msg: &msgs::ErrorMessage) {} fn handle_error(&self, _their_node_id: &PublicKey, _msg: &msgs::ErrorMessage) {}
fn provided_node_features(&self) -> NodeFeatures { NodeFeatures::empty() } fn provided_node_features(&self) -> NodeFeatures { NodeFeatures::empty() }
@ -322,16 +322,7 @@ pub trait SocketDescriptor : cmp::Eq + hash::Hash + Clone {
/// generate no further read_event/write_buffer_space_avail/socket_disconnected calls for the /// generate no further read_event/write_buffer_space_avail/socket_disconnected calls for the
/// descriptor. /// descriptor.
#[derive(Clone)] #[derive(Clone)]
pub struct PeerHandleError { pub struct PeerHandleError { }
/// Used to indicate that we probably can't make any future connections to this peer (e.g.
/// because we required features that our peer was missing, or vice versa).
///
/// While LDK's [`ChannelManager`] will not do it automatically, you likely wish to force-close
/// any channels with this peer or check for new versions of LDK.
///
/// [`ChannelManager`]: crate::ln::channelmanager::ChannelManager
pub no_connection_possible: bool,
}
impl fmt::Debug for PeerHandleError { impl fmt::Debug for PeerHandleError {
fn fmt(&self, formatter: &mut fmt::Formatter) -> Result<(), fmt::Error> { fn fmt(&self, formatter: &mut fmt::Formatter) -> Result<(), fmt::Error> {
formatter.write_str("Peer Sent Invalid Data") formatter.write_str("Peer Sent Invalid Data")
@ -400,6 +391,12 @@ struct Peer {
/// We cache a `NodeId` here to avoid serializing peers' keys every time we forward gossip /// We cache a `NodeId` here to avoid serializing peers' keys every time we forward gossip
/// messages in `PeerManager`. Use `Peer::set_their_node_id` to modify this field. /// messages in `PeerManager`. Use `Peer::set_their_node_id` to modify this field.
their_node_id: Option<(PublicKey, NodeId)>, their_node_id: Option<(PublicKey, NodeId)>,
/// The features provided in the peer's [`msgs::Init`] message.
///
/// This is set only after we've processed the [`msgs::Init`] message and called relevant
/// `peer_connected` handler methods. Thus, this field is set *iff* we've finished our
/// handshake and can talk to this peer normally (though use [`Peer::handshake_complete`] to
/// check this.
their_features: Option<InitFeatures>, their_features: Option<InitFeatures>,
their_net_address: Option<NetAddress>, their_net_address: Option<NetAddress>,
@ -431,6 +428,13 @@ struct Peer {
} }
impl Peer { impl Peer {
/// True after we've processed the [`msgs::Init`] message and called relevant `peer_connected`
/// handler methods. Thus, this implies we've finished our handshake and can talk to this peer
/// normally.
fn handshake_complete(&self) -> bool {
self.their_features.is_some()
}
/// Returns true if the channel announcements/updates for the given channel should be /// Returns true if the channel announcements/updates for the given channel should be
/// forwarded to this peer. /// forwarded to this peer.
/// If we are sending our routing table to this peer and we have not yet sent channel /// If we are sending our routing table to this peer and we have not yet sent channel
@ -438,6 +442,7 @@ impl Peer {
/// point and we shouldn't send it yet to avoid sending duplicate updates. If we've already /// point and we shouldn't send it yet to avoid sending duplicate updates. If we've already
/// sent the old versions, we should send the update, and so return true here. /// sent the old versions, we should send the update, and so return true here.
fn should_forward_channel_announcement(&self, channel_id: u64) -> bool { fn should_forward_channel_announcement(&self, channel_id: u64) -> bool {
if !self.handshake_complete() { return false; }
if self.their_features.as_ref().unwrap().supports_gossip_queries() && if self.their_features.as_ref().unwrap().supports_gossip_queries() &&
!self.sent_gossip_timestamp_filter { !self.sent_gossip_timestamp_filter {
return false; return false;
@ -451,6 +456,7 @@ impl Peer {
/// Similar to the above, but for node announcements indexed by node_id. /// Similar to the above, but for node announcements indexed by node_id.
fn should_forward_node_announcement(&self, node_id: NodeId) -> bool { fn should_forward_node_announcement(&self, node_id: NodeId) -> bool {
if !self.handshake_complete() { return false; }
if self.their_features.as_ref().unwrap().supports_gossip_queries() && if self.their_features.as_ref().unwrap().supports_gossip_queries() &&
!self.sent_gossip_timestamp_filter { !self.sent_gossip_timestamp_filter {
return false; return false;
@ -477,19 +483,20 @@ impl Peer {
fn should_buffer_gossip_backfill(&self) -> bool { fn should_buffer_gossip_backfill(&self) -> bool {
self.pending_outbound_buffer.is_empty() && self.gossip_broadcast_buffer.is_empty() self.pending_outbound_buffer.is_empty() && self.gossip_broadcast_buffer.is_empty()
&& self.msgs_sent_since_pong < BUFFER_DRAIN_MSGS_PER_TICK && self.msgs_sent_since_pong < BUFFER_DRAIN_MSGS_PER_TICK
&& self.handshake_complete()
} }
/// Determines if we should push an onion message onto a peer's outbound buffer. This is checked /// Determines if we should push an onion message onto a peer's outbound buffer. This is checked
/// every time the peer's buffer may have been drained. /// every time the peer's buffer may have been drained.
fn should_buffer_onion_message(&self) -> bool { fn should_buffer_onion_message(&self) -> bool {
self.pending_outbound_buffer.is_empty() self.pending_outbound_buffer.is_empty() && self.handshake_complete()
&& self.msgs_sent_since_pong < BUFFER_DRAIN_MSGS_PER_TICK && self.msgs_sent_since_pong < BUFFER_DRAIN_MSGS_PER_TICK
} }
/// Determines if we should push additional gossip broadcast messages onto a peer's outbound /// Determines if we should push additional gossip broadcast messages onto a peer's outbound
/// buffer. This is checked every time the peer's buffer may have been drained. /// buffer. This is checked every time the peer's buffer may have been drained.
fn should_buffer_gossip_broadcast(&self) -> bool { fn should_buffer_gossip_broadcast(&self) -> bool {
self.pending_outbound_buffer.is_empty() self.pending_outbound_buffer.is_empty() && self.handshake_complete()
&& self.msgs_sent_since_pong < BUFFER_DRAIN_MSGS_PER_TICK && self.msgs_sent_since_pong < BUFFER_DRAIN_MSGS_PER_TICK
} }
@ -771,8 +778,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
let peers = self.peers.read().unwrap(); let peers = self.peers.read().unwrap();
peers.values().filter_map(|peer_mutex| { peers.values().filter_map(|peer_mutex| {
let p = peer_mutex.lock().unwrap(); let p = peer_mutex.lock().unwrap();
if !p.channel_encryptor.is_ready_for_encryption() || p.their_features.is_none() || if !p.handshake_complete() {
p.their_node_id.is_none() {
return None; return None;
} }
Some((p.their_node_id.unwrap().0, p.their_net_address.clone())) Some((p.their_node_id.unwrap().0, p.their_net_address.clone()))
@ -1001,7 +1007,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
// This is most likely a simple race condition where the user found that the socket // This is most likely a simple race condition where the user found that the socket
// was writeable, then we told the user to `disconnect_socket()`, then they called // was writeable, then we told the user to `disconnect_socket()`, then they called
// this method. Return an error to make sure we get disconnected. // this method. Return an error to make sure we get disconnected.
return Err(PeerHandleError { no_connection_possible: false }); return Err(PeerHandleError { });
}, },
Some(peer_mutex) => { Some(peer_mutex) => {
let mut peer = peer_mutex.lock().unwrap(); let mut peer = peer_mutex.lock().unwrap();
@ -1034,7 +1040,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
Ok(res) => Ok(res), Ok(res) => Ok(res),
Err(e) => { Err(e) => {
log_trace!(self.logger, "Peer sent invalid data or we decided to disconnect due to a protocol error"); log_trace!(self.logger, "Peer sent invalid data or we decided to disconnect due to a protocol error");
self.disconnect_event_internal(peer_descriptor, e.no_connection_possible); self.disconnect_event_internal(peer_descriptor);
Err(e) Err(e)
} }
} }
@ -1067,7 +1073,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
// This is most likely a simple race condition where the user read some bytes // This is most likely a simple race condition where the user read some bytes
// from the socket, then we told the user to `disconnect_socket()`, then they // from the socket, then we told the user to `disconnect_socket()`, then they
// called this method. Return an error to make sure we get disconnected. // called this method. Return an error to make sure we get disconnected.
return Err(PeerHandleError { no_connection_possible: false }); return Err(PeerHandleError { });
}, },
Some(peer_mutex) => { Some(peer_mutex) => {
let mut read_pos = 0; let mut read_pos = 0;
@ -1081,7 +1087,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
msgs::ErrorAction::DisconnectPeer { msg: _ } => { msgs::ErrorAction::DisconnectPeer { msg: _ } => {
//TODO: Try to push msg //TODO: Try to push msg
log_debug!(self.logger, "Error handling message{}; disconnecting peer with: {}", OptionalFromDebugger(&peer_node_id), e.err); log_debug!(self.logger, "Error handling message{}; disconnecting peer with: {}", OptionalFromDebugger(&peer_node_id), e.err);
return Err(PeerHandleError{ no_connection_possible: false }); return Err(PeerHandleError { });
}, },
msgs::ErrorAction::IgnoreAndLog(level) => { msgs::ErrorAction::IgnoreAndLog(level) => {
log_given_level!(self.logger, level, "Error handling message{}; ignoring: {}", OptionalFromDebugger(&peer_node_id), e.err); log_given_level!(self.logger, level, "Error handling message{}; ignoring: {}", OptionalFromDebugger(&peer_node_id), e.err);
@ -1134,7 +1140,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
hash_map::Entry::Occupied(_) => { hash_map::Entry::Occupied(_) => {
log_trace!(self.logger, "Got second connection with {}, closing", log_pubkey!(peer.their_node_id.unwrap().0)); log_trace!(self.logger, "Got second connection with {}, closing", log_pubkey!(peer.their_node_id.unwrap().0));
peer.their_node_id = None; // Unset so that we don't generate a peer_disconnected event peer.their_node_id = None; // Unset so that we don't generate a peer_disconnected event
return Err(PeerHandleError{ no_connection_possible: false }) return Err(PeerHandleError { })
}, },
hash_map::Entry::Vacant(entry) => { hash_map::Entry::Vacant(entry) => {
log_debug!(self.logger, "Finished noise handshake for connection with {}", log_pubkey!(peer.their_node_id.unwrap().0)); log_debug!(self.logger, "Finished noise handshake for connection with {}", log_pubkey!(peer.their_node_id.unwrap().0));
@ -1191,7 +1197,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
if peer.pending_read_buffer.capacity() > 8192 { peer.pending_read_buffer = Vec::new(); } if peer.pending_read_buffer.capacity() > 8192 { peer.pending_read_buffer = Vec::new(); }
peer.pending_read_buffer.resize(msg_len as usize + 16, 0); peer.pending_read_buffer.resize(msg_len as usize + 16, 0);
if msg_len < 2 { // Need at least the message type tag if msg_len < 2 { // Need at least the message type tag
return Err(PeerHandleError{ no_connection_possible: false }); return Err(PeerHandleError { });
} }
peer.pending_read_is_header = false; peer.pending_read_is_header = false;
} else { } else {
@ -1234,19 +1240,19 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
(msgs::DecodeError::UnknownRequiredFeature, ty) => { (msgs::DecodeError::UnknownRequiredFeature, ty) => {
log_gossip!(self.logger, "Received a message with an unknown required feature flag or TLV, you may want to update!"); log_gossip!(self.logger, "Received a message with an unknown required feature flag or TLV, you may want to update!");
self.enqueue_message(peer, &msgs::WarningMessage { channel_id: [0; 32], data: format!("Received an unknown required feature/TLV in message type {:?}", ty) }); self.enqueue_message(peer, &msgs::WarningMessage { channel_id: [0; 32], data: format!("Received an unknown required feature/TLV in message type {:?}", ty) });
return Err(PeerHandleError { no_connection_possible: false }); return Err(PeerHandleError { });
} }
(msgs::DecodeError::UnknownVersion, _) => return Err(PeerHandleError { no_connection_possible: false }), (msgs::DecodeError::UnknownVersion, _) => return Err(PeerHandleError { }),
(msgs::DecodeError::InvalidValue, _) => { (msgs::DecodeError::InvalidValue, _) => {
log_debug!(self.logger, "Got an invalid value while deserializing message"); log_debug!(self.logger, "Got an invalid value while deserializing message");
return Err(PeerHandleError { no_connection_possible: false }); return Err(PeerHandleError { });
} }
(msgs::DecodeError::ShortRead, _) => { (msgs::DecodeError::ShortRead, _) => {
log_debug!(self.logger, "Deserialization failed due to shortness of message"); log_debug!(self.logger, "Deserialization failed due to shortness of message");
return Err(PeerHandleError { no_connection_possible: false }); return Err(PeerHandleError { });
} }
(msgs::DecodeError::BadLengthDescriptor, _) => return Err(PeerHandleError { no_connection_possible: false }), (msgs::DecodeError::BadLengthDescriptor, _) => return Err(PeerHandleError { }),
(msgs::DecodeError::Io(_), _) => return Err(PeerHandleError { no_connection_possible: false }), (msgs::DecodeError::Io(_), _) => return Err(PeerHandleError { }),
} }
} }
}; };
@ -1298,10 +1304,10 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
if let wire::Message::Init(msg) = message { if let wire::Message::Init(msg) = message {
if msg.features.requires_unknown_bits() { if msg.features.requires_unknown_bits() {
log_debug!(self.logger, "Peer features required unknown version bits"); log_debug!(self.logger, "Peer features required unknown version bits");
return Err(PeerHandleError{ no_connection_possible: true }.into()); return Err(PeerHandleError { }.into());
} }
if peer_lock.their_features.is_some() { if peer_lock.their_features.is_some() {
return Err(PeerHandleError{ no_connection_possible: false }.into()); return Err(PeerHandleError { }.into());
} }
log_info!(self.logger, "Received peer Init message from {}: {}", log_pubkey!(their_node_id), msg.features); log_info!(self.logger, "Received peer Init message from {}: {}", log_pubkey!(their_node_id), msg.features);
@ -1313,22 +1319,22 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
if let Err(()) = self.message_handler.route_handler.peer_connected(&their_node_id, &msg) { if let Err(()) = self.message_handler.route_handler.peer_connected(&their_node_id, &msg) {
log_debug!(self.logger, "Route Handler decided we couldn't communicate with peer {}", log_pubkey!(their_node_id)); log_debug!(self.logger, "Route Handler decided we couldn't communicate with peer {}", log_pubkey!(their_node_id));
return Err(PeerHandleError{ no_connection_possible: true }.into()); return Err(PeerHandleError { }.into());
} }
if let Err(()) = self.message_handler.chan_handler.peer_connected(&their_node_id, &msg) { if let Err(()) = self.message_handler.chan_handler.peer_connected(&their_node_id, &msg) {
log_debug!(self.logger, "Channel Handler decided we couldn't communicate with peer {}", log_pubkey!(their_node_id)); log_debug!(self.logger, "Channel Handler decided we couldn't communicate with peer {}", log_pubkey!(their_node_id));
return Err(PeerHandleError{ no_connection_possible: true }.into()); return Err(PeerHandleError { }.into());
} }
if let Err(()) = self.message_handler.onion_message_handler.peer_connected(&their_node_id, &msg) { if let Err(()) = self.message_handler.onion_message_handler.peer_connected(&their_node_id, &msg) {
log_debug!(self.logger, "Onion Message Handler decided we couldn't communicate with peer {}", log_pubkey!(their_node_id)); log_debug!(self.logger, "Onion Message Handler decided we couldn't communicate with peer {}", log_pubkey!(their_node_id));
return Err(PeerHandleError{ no_connection_possible: true }.into()); return Err(PeerHandleError { }.into());
} }
peer_lock.their_features = Some(msg.features); peer_lock.their_features = Some(msg.features);
return Ok(None); return Ok(None);
} else if peer_lock.their_features.is_none() { } else if peer_lock.their_features.is_none() {
log_debug!(self.logger, "Peer {} sent non-Init first message", log_pubkey!(their_node_id)); log_debug!(self.logger, "Peer {} sent non-Init first message", log_pubkey!(their_node_id));
return Err(PeerHandleError{ no_connection_possible: false }.into()); return Err(PeerHandleError { }.into());
} }
if let wire::Message::GossipTimestampFilter(_msg) = message { if let wire::Message::GossipTimestampFilter(_msg) = message {
@ -1380,7 +1386,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
} }
self.message_handler.chan_handler.handle_error(&their_node_id, &msg); self.message_handler.chan_handler.handle_error(&their_node_id, &msg);
if msg.channel_id == [0; 32] { if msg.channel_id == [0; 32] {
return Err(PeerHandleError{ no_connection_possible: true }.into()); return Err(PeerHandleError { }.into());
} }
}, },
wire::Message::Warning(msg) => { wire::Message::Warning(msg) => {
@ -1510,8 +1516,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
// Unknown messages: // Unknown messages:
wire::Message::Unknown(type_id) if message.is_even() => { wire::Message::Unknown(type_id) if message.is_even() => {
log_debug!(self.logger, "Received unknown even message of type {}, disconnecting peer!", type_id); log_debug!(self.logger, "Received unknown even message of type {}, disconnecting peer!", type_id);
// Fail the channel if message is an even, unknown type as per BOLT #1. return Err(PeerHandleError { }.into());
return Err(PeerHandleError{ no_connection_possible: true }.into());
}, },
wire::Message::Unknown(type_id) => { wire::Message::Unknown(type_id) => {
log_trace!(self.logger, "Received unknown odd message of type {}, ignoring", type_id); log_trace!(self.logger, "Received unknown odd message of type {}, ignoring", type_id);
@ -1531,10 +1536,12 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
for (_, peer_mutex) in peers.iter() { for (_, peer_mutex) in peers.iter() {
let mut peer = peer_mutex.lock().unwrap(); let mut peer = peer_mutex.lock().unwrap();
if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_features.is_none() || if !peer.handshake_complete() ||
!peer.should_forward_channel_announcement(msg.contents.short_channel_id) { !peer.should_forward_channel_announcement(msg.contents.short_channel_id) {
continue continue
} }
debug_assert!(peer.their_node_id.is_some());
debug_assert!(peer.channel_encryptor.is_ready_for_encryption());
if peer.buffer_full_drop_gossip_broadcast() { if peer.buffer_full_drop_gossip_broadcast() {
log_gossip!(self.logger, "Skipping broadcast message to {:?} as its outbound buffer is full", peer.their_node_id); log_gossip!(self.logger, "Skipping broadcast message to {:?} as its outbound buffer is full", peer.their_node_id);
continue; continue;
@ -1556,10 +1563,12 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
for (_, peer_mutex) in peers.iter() { for (_, peer_mutex) in peers.iter() {
let mut peer = peer_mutex.lock().unwrap(); let mut peer = peer_mutex.lock().unwrap();
if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_features.is_none() || if !peer.handshake_complete() ||
!peer.should_forward_node_announcement(msg.contents.node_id) { !peer.should_forward_node_announcement(msg.contents.node_id) {
continue continue
} }
debug_assert!(peer.their_node_id.is_some());
debug_assert!(peer.channel_encryptor.is_ready_for_encryption());
if peer.buffer_full_drop_gossip_broadcast() { if peer.buffer_full_drop_gossip_broadcast() {
log_gossip!(self.logger, "Skipping broadcast message to {:?} as its outbound buffer is full", peer.their_node_id); log_gossip!(self.logger, "Skipping broadcast message to {:?} as its outbound buffer is full", peer.their_node_id);
continue; continue;
@ -1581,10 +1590,12 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
for (_, peer_mutex) in peers.iter() { for (_, peer_mutex) in peers.iter() {
let mut peer = peer_mutex.lock().unwrap(); let mut peer = peer_mutex.lock().unwrap();
if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_features.is_none() || if !peer.handshake_complete() ||
!peer.should_forward_channel_announcement(msg.contents.short_channel_id) { !peer.should_forward_channel_announcement(msg.contents.short_channel_id) {
continue continue
} }
debug_assert!(peer.their_node_id.is_some());
debug_assert!(peer.channel_encryptor.is_ready_for_encryption());
if peer.buffer_full_drop_gossip_broadcast() { if peer.buffer_full_drop_gossip_broadcast() {
log_gossip!(self.logger, "Skipping broadcast message to {:?} as its outbound buffer is full", peer.their_node_id); log_gossip!(self.logger, "Skipping broadcast message to {:?} as its outbound buffer is full", peer.their_node_id);
continue; continue;
@ -1664,7 +1675,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
Some(descriptor) => match peers.get(&descriptor) { Some(descriptor) => match peers.get(&descriptor) {
Some(peer_mutex) => { Some(peer_mutex) => {
let peer_lock = peer_mutex.lock().unwrap(); let peer_lock = peer_mutex.lock().unwrap();
if peer_lock.their_features.is_none() { if !peer_lock.handshake_complete() {
continue; continue;
} }
peer_lock peer_lock
@ -1884,24 +1895,21 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
// thread can be holding the peer lock if we have the global write // thread can be holding the peer lock if we have the global write
// lock). // lock).
if let Some(mut descriptor) = self.node_id_to_descriptor.lock().unwrap().remove(&node_id) { let descriptor_opt = self.node_id_to_descriptor.lock().unwrap().remove(&node_id);
if let Some(mut descriptor) = descriptor_opt {
if let Some(peer_mutex) = peers.remove(&descriptor) { if let Some(peer_mutex) = peers.remove(&descriptor) {
let mut peer = peer_mutex.lock().unwrap();
if let Some(msg) = msg { if let Some(msg) = msg {
log_trace!(self.logger, "Handling DisconnectPeer HandleError event in peer_handler for node {} with message {}", log_trace!(self.logger, "Handling DisconnectPeer HandleError event in peer_handler for node {} with message {}",
log_pubkey!(node_id), log_pubkey!(node_id),
msg.data); msg.data);
let mut peer = peer_mutex.lock().unwrap();
self.enqueue_message(&mut *peer, &msg); self.enqueue_message(&mut *peer, &msg);
// This isn't guaranteed to work, but if there is enough free // This isn't guaranteed to work, but if there is enough free
// room in the send buffer, put the error message there... // room in the send buffer, put the error message there...
self.do_attempt_write_data(&mut descriptor, &mut *peer, false); self.do_attempt_write_data(&mut descriptor, &mut *peer, false);
} else {
log_trace!(self.logger, "Handling DisconnectPeer HandleError event in peer_handler for node {} with no message", log_pubkey!(node_id));
} }
self.do_disconnect(descriptor, &*peer, "DisconnectPeer HandleError");
} }
descriptor.disconnect_socket();
self.message_handler.chan_handler.peer_disconnected(&node_id, false);
self.message_handler.onion_message_handler.peer_disconnected(&node_id, false);
} }
} }
} }
@ -1909,10 +1917,26 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
/// Indicates that the given socket descriptor's connection is now closed. /// Indicates that the given socket descriptor's connection is now closed.
pub fn socket_disconnected(&self, descriptor: &Descriptor) { pub fn socket_disconnected(&self, descriptor: &Descriptor) {
self.disconnect_event_internal(descriptor, false); self.disconnect_event_internal(descriptor);
} }
fn disconnect_event_internal(&self, descriptor: &Descriptor, no_connection_possible: bool) { fn do_disconnect(&self, mut descriptor: Descriptor, peer: &Peer, reason: &'static str) {
if !peer.handshake_complete() {
log_trace!(self.logger, "Disconnecting peer which hasn't completed handshake due to {}", reason);
descriptor.disconnect_socket();
return;
}
debug_assert!(peer.their_node_id.is_some());
if let Some((node_id, _)) = peer.their_node_id {
log_trace!(self.logger, "Disconnecting peer with id {} due to {}", node_id, reason);
self.message_handler.chan_handler.peer_disconnected(&node_id);
self.message_handler.onion_message_handler.peer_disconnected(&node_id);
}
descriptor.disconnect_socket();
}
fn disconnect_event_internal(&self, descriptor: &Descriptor) {
let mut peers = self.peers.write().unwrap(); let mut peers = self.peers.write().unwrap();
let peer_option = peers.remove(descriptor); let peer_option = peers.remove(descriptor);
match peer_option { match peer_option {
@ -1923,13 +1947,13 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
}, },
Some(peer_lock) => { Some(peer_lock) => {
let peer = peer_lock.lock().unwrap(); let peer = peer_lock.lock().unwrap();
if !peer.handshake_complete() { return; }
debug_assert!(peer.their_node_id.is_some());
if let Some((node_id, _)) = peer.their_node_id { if let Some((node_id, _)) = peer.their_node_id {
log_trace!(self.logger, log_trace!(self.logger, "Handling disconnection of peer {}", log_pubkey!(node_id));
"Handling disconnection of peer {}, with {}future connection to the peer possible.",
log_pubkey!(node_id), if no_connection_possible { "no " } else { "" });
self.node_id_to_descriptor.lock().unwrap().remove(&node_id); self.node_id_to_descriptor.lock().unwrap().remove(&node_id);
self.message_handler.chan_handler.peer_disconnected(&node_id, no_connection_possible); self.message_handler.chan_handler.peer_disconnected(&node_id);
self.message_handler.onion_message_handler.peer_disconnected(&node_id, no_connection_possible); self.message_handler.onion_message_handler.peer_disconnected(&node_id);
} }
} }
}; };
@ -1937,21 +1961,17 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
/// Disconnect a peer given its node id. /// Disconnect a peer given its node id.
/// ///
/// Set `no_connection_possible` to true to prevent any further connection with this peer,
/// force-closing any channels we have with it.
///
/// If a peer is connected, this will call [`disconnect_socket`] on the descriptor for the /// If a peer is connected, this will call [`disconnect_socket`] on the descriptor for the
/// peer. Thus, be very careful about reentrancy issues. /// peer. Thus, be very careful about reentrancy issues.
/// ///
/// [`disconnect_socket`]: SocketDescriptor::disconnect_socket /// [`disconnect_socket`]: SocketDescriptor::disconnect_socket
pub fn disconnect_by_node_id(&self, node_id: PublicKey, no_connection_possible: bool) { pub fn disconnect_by_node_id(&self, node_id: PublicKey) {
let mut peers_lock = self.peers.write().unwrap(); let mut peers_lock = self.peers.write().unwrap();
if let Some(mut descriptor) = self.node_id_to_descriptor.lock().unwrap().remove(&node_id) { if let Some(descriptor) = self.node_id_to_descriptor.lock().unwrap().remove(&node_id) {
log_trace!(self.logger, "Disconnecting peer with id {} due to client request", node_id); let peer_opt = peers_lock.remove(&descriptor);
peers_lock.remove(&descriptor); if let Some(peer_mutex) = peer_opt {
self.message_handler.chan_handler.peer_disconnected(&node_id, no_connection_possible); self.do_disconnect(descriptor, &*peer_mutex.lock().unwrap(), "client request");
self.message_handler.onion_message_handler.peer_disconnected(&node_id, no_connection_possible); } else { debug_assert!(false, "node_id_to_descriptor thought we had a peer"); }
descriptor.disconnect_socket();
} }
} }
@ -1962,13 +1982,8 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
let mut peers_lock = self.peers.write().unwrap(); let mut peers_lock = self.peers.write().unwrap();
self.node_id_to_descriptor.lock().unwrap().clear(); self.node_id_to_descriptor.lock().unwrap().clear();
let peers = &mut *peers_lock; let peers = &mut *peers_lock;
for (mut descriptor, peer) in peers.drain() { for (descriptor, peer_mutex) in peers.drain() {
if let Some((node_id, _)) = peer.lock().unwrap().their_node_id { self.do_disconnect(descriptor, &*peer_mutex.lock().unwrap(), "client request to disconnect all peers");
log_trace!(self.logger, "Disconnecting peer with id {} due to client request to disconnect all peers", node_id);
self.message_handler.chan_handler.peer_disconnected(&node_id, false);
self.message_handler.onion_message_handler.peer_disconnected(&node_id, false);
}
descriptor.disconnect_socket();
} }
} }
@ -2009,7 +2024,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
let mut peer = peer_mutex.lock().unwrap(); let mut peer = peer_mutex.lock().unwrap();
if flush_read_disabled { peer.received_channel_announce_since_backlogged = false; } if flush_read_disabled { peer.received_channel_announce_since_backlogged = false; }
if !peer.channel_encryptor.is_ready_for_encryption() || peer.their_node_id.is_none() { if !peer.handshake_complete() {
// The peer needs to complete its handshake before we can exchange messages. We // The peer needs to complete its handshake before we can exchange messages. We
// give peers one timer tick to complete handshake, reusing // give peers one timer tick to complete handshake, reusing
// `awaiting_pong_timer_tick_intervals` to track number of timer ticks taken // `awaiting_pong_timer_tick_intervals` to track number of timer ticks taken
@ -2021,6 +2036,8 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
} }
continue; continue;
} }
debug_assert!(peer.channel_encryptor.is_ready_for_encryption());
debug_assert!(peer.their_node_id.is_some());
loop { // Used as a `goto` to skip writing a Ping message. loop { // Used as a `goto` to skip writing a Ping message.
if peer.awaiting_pong_timer_tick_intervals == -1 { if peer.awaiting_pong_timer_tick_intervals == -1 {
@ -2059,21 +2076,16 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, OM: Deref, L: Deref, CM
if !descriptors_needing_disconnect.is_empty() { if !descriptors_needing_disconnect.is_empty() {
{ {
let mut peers_lock = self.peers.write().unwrap(); let mut peers_lock = self.peers.write().unwrap();
for descriptor in descriptors_needing_disconnect.iter() { for descriptor in descriptors_needing_disconnect {
if let Some(peer) = peers_lock.remove(descriptor) { if let Some(peer_mutex) = peers_lock.remove(&descriptor) {
if let Some((node_id, _)) = peer.lock().unwrap().their_node_id { let peer = peer_mutex.lock().unwrap();
log_trace!(self.logger, "Disconnecting peer with id {} due to ping timeout", node_id); if let Some((node_id, _)) = peer.their_node_id {
self.node_id_to_descriptor.lock().unwrap().remove(&node_id); self.node_id_to_descriptor.lock().unwrap().remove(&node_id);
self.message_handler.chan_handler.peer_disconnected(&node_id, false);
self.message_handler.onion_message_handler.peer_disconnected(&node_id, false);
} }
self.do_disconnect(descriptor, &*peer, "ping timeout");
} }
} }
} }
for mut descriptor in descriptors_needing_disconnect.drain(..) {
descriptor.disconnect_socket();
}
} }
} }
@ -2161,6 +2173,7 @@ fn is_gossip_msg(type_id: u16) -> bool {
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use crate::chain::keysinterface::{NodeSigner, Recipient}; use crate::chain::keysinterface::{NodeSigner, Recipient};
use crate::ln::peer_channel_encryptor::PeerChannelEncryptor;
use crate::ln::peer_handler::{PeerManager, MessageHandler, SocketDescriptor, IgnoringMessageHandler, filter_addresses}; use crate::ln::peer_handler::{PeerManager, MessageHandler, SocketDescriptor, IgnoringMessageHandler, filter_addresses};
use crate::ln::{msgs, wire}; use crate::ln::{msgs, wire};
use crate::ln::msgs::NetAddress; use crate::ln::msgs::NetAddress;
@ -2269,19 +2282,15 @@ mod tests {
// Simple test which builds a network of PeerManager, connects and brings them to NoiseState::Finished and // Simple test which builds a network of PeerManager, connects and brings them to NoiseState::Finished and
// push a DisconnectPeer event to remove the node flagged by id // push a DisconnectPeer event to remove the node flagged by id
let cfgs = create_peermgr_cfgs(2); let cfgs = create_peermgr_cfgs(2);
let chan_handler = test_utils::TestChannelMessageHandler::new(); let peers = create_network(2, &cfgs);
let mut peers = create_network(2, &cfgs);
establish_connection(&peers[0], &peers[1]); establish_connection(&peers[0], &peers[1]);
assert_eq!(peers[0].peers.read().unwrap().len(), 1); assert_eq!(peers[0].peers.read().unwrap().len(), 1);
let their_id = peers[1].node_signer.get_node_id(Recipient::Node).unwrap(); let their_id = peers[1].node_signer.get_node_id(Recipient::Node).unwrap();
cfgs[0].chan_handler.pending_events.lock().unwrap().push(events::MessageSendEvent::HandleError {
chan_handler.pending_events.lock().unwrap().push(events::MessageSendEvent::HandleError {
node_id: their_id, node_id: their_id,
action: msgs::ErrorAction::DisconnectPeer { msg: None }, action: msgs::ErrorAction::DisconnectPeer { msg: None },
}); });
assert_eq!(chan_handler.pending_events.lock().unwrap().len(), 1);
peers[0].message_handler.chan_handler = &chan_handler;
peers[0].process_events(); peers[0].process_events();
assert_eq!(peers[0].peers.read().unwrap().len(), 0); assert_eq!(peers[0].peers.read().unwrap().len(), 0);
@ -2315,6 +2324,35 @@ mod tests {
assert_eq!(peers[1].read_event(&mut fd_b, &a_data).unwrap(), false); assert_eq!(peers[1].read_event(&mut fd_b, &a_data).unwrap(), false);
} }
#[test]
fn test_non_init_first_msg() {
// Simple test of the first message received over a connection being something other than
// Init. This results in an immediate disconnection, which previously included a spurious
// peer_disconnected event handed to event handlers (which would panic in
// `TestChannelMessageHandler` here).
let cfgs = create_peermgr_cfgs(2);
let peers = create_network(2, &cfgs);
let mut fd_dup = FileDescriptor { fd: 3, outbound_data: Arc::new(Mutex::new(Vec::new())) };
let addr_dup = NetAddress::IPv4{addr: [127, 0, 0, 1], port: 1003};
let id_a = cfgs[0].node_signer.get_node_id(Recipient::Node).unwrap();
peers[0].new_inbound_connection(fd_dup.clone(), Some(addr_dup.clone())).unwrap();
let mut dup_encryptor = PeerChannelEncryptor::new_outbound(id_a, SecretKey::from_slice(&[42; 32]).unwrap());
let initial_data = dup_encryptor.get_act_one(&peers[1].secp_ctx);
assert_eq!(peers[0].read_event(&mut fd_dup, &initial_data).unwrap(), false);
peers[0].process_events();
let a_data = fd_dup.outbound_data.lock().unwrap().split_off(0);
let (act_three, _) =
dup_encryptor.process_act_two(&a_data[..], &&cfgs[1].node_signer).unwrap();
assert_eq!(peers[0].read_event(&mut fd_dup, &act_three).unwrap(), false);
let not_init_msg = msgs::Ping { ponglen: 4, byteslen: 0 };
let msg_bytes = dup_encryptor.encrypt_message(&not_init_msg);
assert!(peers[0].read_event(&mut fd_dup, &msg_bytes).is_err());
}
#[test] #[test]
fn test_disconnect_all_peer() { fn test_disconnect_all_peer() {
// Simple test which builds a network of PeerManager, connects and brings them to NoiseState::Finished and // Simple test which builds a network of PeerManager, connects and brings them to NoiseState::Finished and

4
lightning/src/ln/priv_short_conf_tests.rs
View file

@ -90,8 +90,8 @@ fn test_priv_forwarding_rejection() {
// Now disconnect nodes[1] from its peers and restart with accept_forwards_to_priv_channels set // Now disconnect nodes[1] from its peers and restart with accept_forwards_to_priv_channels set
// to true. Sadly there is currently no way to change it at runtime. // to true. Sadly there is currently no way to change it at runtime.
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
nodes[2].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[2].node.peer_disconnected(&nodes[1].node.get_our_node_id());
let nodes_1_serialized = nodes[1].node.encode(); let nodes_1_serialized = nodes[1].node.encode();
let monitor_a_serialized = get_monitor!(nodes[1], chan_id_1).encode(); let monitor_a_serialized = get_monitor!(nodes[1], chan_id_1).encode();

38
lightning/src/ln/reload_tests.rs
View file

@ -44,8 +44,8 @@ fn test_funding_peer_disconnect() {
let mut nodes = create_network(2, &node_cfgs, &node_chanmgrs); let mut nodes = create_network(2, &node_cfgs, &node_chanmgrs);
let tx = create_chan_between_nodes_with_value_init(&nodes[0], &nodes[1], 100000, 10001); let tx = create_chan_between_nodes_with_value_init(&nodes[0], &nodes[1], 100000, 10001);
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
confirm_transaction(&nodes[0], &tx); confirm_transaction(&nodes[0], &tx);
let events_1 = nodes[0].node.get_and_clear_pending_msg_events(); let events_1 = nodes[0].node.get_and_clear_pending_msg_events();
@ -53,8 +53,8 @@ fn test_funding_peer_disconnect() {
reconnect_nodes(&nodes[0], &nodes[1], (false, true), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (false, false)); reconnect_nodes(&nodes[0], &nodes[1], (false, true), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (false, false));
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
confirm_transaction(&nodes[1], &tx); confirm_transaction(&nodes[1], &tx);
let events_2 = nodes[1].node.get_and_clear_pending_msg_events(); let events_2 = nodes[1].node.get_and_clear_pending_msg_events();
@ -169,7 +169,7 @@ fn test_funding_peer_disconnect() {
// Check that after deserialization and reconnection we can still generate an identical // Check that after deserialization and reconnection we can still generate an identical
// channel_announcement from the cached signatures. // channel_announcement from the cached signatures.
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
let chan_0_monitor_serialized = get_monitor!(nodes[0], chan_id).encode(); let chan_0_monitor_serialized = get_monitor!(nodes[0], chan_id).encode();
@ -190,7 +190,7 @@ fn test_no_txn_manager_serialize_deserialize() {
let tx = create_chan_between_nodes_with_value_init(&nodes[0], &nodes[1], 100000, 10001); let tx = create_chan_between_nodes_with_value_init(&nodes[0], &nodes[1], 100000, 10001);
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
let chan_0_monitor_serialized = let chan_0_monitor_serialized =
get_monitor!(nodes[0], OutPoint { txid: tx.txid(), index: 0 }.to_channel_id()).encode(); get_monitor!(nodes[0], OutPoint { txid: tx.txid(), index: 0 }.to_channel_id()).encode();
@ -267,7 +267,7 @@ fn test_manager_serialize_deserialize_events() {
let chan_0_monitor_serialized = get_monitor!(nodes[0], bs_funding_signed.channel_id).encode(); let chan_0_monitor_serialized = get_monitor!(nodes[0], bs_funding_signed.channel_id).encode();
reload_node!(nodes[0], nodes[0].node.encode(), &[&chan_0_monitor_serialized], persister, new_chain_monitor, nodes_0_deserialized); reload_node!(nodes[0], nodes[0].node.encode(), &[&chan_0_monitor_serialized], persister, new_chain_monitor, nodes_0_deserialized);
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
// After deserializing, make sure the funding_transaction is still held by the channel manager // After deserializing, make sure the funding_transaction is still held by the channel manager
let events_4 = nodes[0].node.get_and_clear_pending_events(); let events_4 = nodes[0].node.get_and_clear_pending_events();
@ -313,7 +313,7 @@ fn test_simple_manager_serialize_deserialize() {
let (our_payment_preimage, _, _) = route_payment(&nodes[0], &[&nodes[1]], 1000000); let (our_payment_preimage, _, _) = route_payment(&nodes[0], &[&nodes[1]], 1000000);
let (_, our_payment_hash, _) = route_payment(&nodes[0], &[&nodes[1]], 1000000); let (_, our_payment_hash, _) = route_payment(&nodes[0], &[&nodes[1]], 1000000);
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
let chan_0_monitor_serialized = get_monitor!(nodes[0], chan_id).encode(); let chan_0_monitor_serialized = get_monitor!(nodes[0], chan_id).encode();
reload_node!(nodes[0], nodes[0].node.encode(), &[&chan_0_monitor_serialized], persister, new_chain_monitor, nodes_0_deserialized); reload_node!(nodes[0], nodes[0].node.encode(), &[&chan_0_monitor_serialized], persister, new_chain_monitor, nodes_0_deserialized);
@ -353,9 +353,9 @@ fn test_manager_serialize_deserialize_inconsistent_monitor() {
let nodes_0_serialized = nodes[0].node.encode(); let nodes_0_serialized = nodes[0].node.encode();
route_payment(&nodes[0], &[&nodes[3]], 1000000); route_payment(&nodes[0], &[&nodes[3]], 1000000);
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
nodes[2].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[2].node.peer_disconnected(&nodes[0].node.get_our_node_id());
nodes[3].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[3].node.peer_disconnected(&nodes[0].node.get_our_node_id());
// Now the ChannelMonitor (which is now out-of-sync with ChannelManager for channel w/ // Now the ChannelMonitor (which is now out-of-sync with ChannelManager for channel w/
// nodes[3]) // nodes[3])
@ -488,8 +488,8 @@ fn do_test_data_loss_protect(reconnect_panicing: bool) {
send_payment(&nodes[0], &vec!(&nodes[1])[..], 8000000); send_payment(&nodes[0], &vec!(&nodes[1])[..], 8000000);
send_payment(&nodes[0], &vec!(&nodes[1])[..], 8000000); send_payment(&nodes[0], &vec!(&nodes[1])[..], 8000000);
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
reload_node!(nodes[0], previous_node_state, &[&previous_chain_monitor_state], persister, new_chain_monitor, nodes_0_deserialized); reload_node!(nodes[0], previous_node_state, &[&previous_chain_monitor_state], persister, new_chain_monitor, nodes_0_deserialized);
@ -627,8 +627,8 @@ fn test_forwardable_regen() {
assert!(nodes[1].node.get_and_clear_pending_events().is_empty()); assert!(nodes[1].node.get_and_clear_pending_events().is_empty());
// Now restart nodes[1] and make sure it regenerates a single PendingHTLCsForwardable // Now restart nodes[1] and make sure it regenerates a single PendingHTLCsForwardable
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
nodes[2].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[2].node.peer_disconnected(&nodes[1].node.get_our_node_id());
let chan_0_monitor_serialized = get_monitor!(nodes[1], chan_id_1).encode(); let chan_0_monitor_serialized = get_monitor!(nodes[1], chan_id_1).encode();
let chan_1_monitor_serialized = get_monitor!(nodes[1], chan_id_2).encode(); let chan_1_monitor_serialized = get_monitor!(nodes[1], chan_id_2).encode();
@ -753,8 +753,8 @@ fn do_test_partial_claim_before_restart(persist_both_monitors: bool) {
assert!(get_monitor!(nodes[3], chan_id_persisted).get_stored_preimages().contains_key(&payment_hash)); assert!(get_monitor!(nodes[3], chan_id_persisted).get_stored_preimages().contains_key(&payment_hash));
assert!(get_monitor!(nodes[3], chan_id_not_persisted).get_stored_preimages().contains_key(&payment_hash)); assert!(get_monitor!(nodes[3], chan_id_not_persisted).get_stored_preimages().contains_key(&payment_hash));
nodes[1].node.peer_disconnected(&nodes[3].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[3].node.get_our_node_id());
nodes[2].node.peer_disconnected(&nodes[3].node.get_our_node_id(), false); nodes[2].node.peer_disconnected(&nodes[3].node.get_our_node_id());
// During deserialization, we should have closed one channel and broadcast its latest // During deserialization, we should have closed one channel and broadcast its latest
// commitment transaction. We should also still have the original PaymentClaimable event we // commitment transaction. We should also still have the original PaymentClaimable event we
@ -923,7 +923,7 @@ fn do_forwarded_payment_no_manager_persistence(use_cs_commitment: bool, claim_ht
let bs_commitment_tx = nodes[1].tx_broadcaster.txn_broadcasted.lock().unwrap().split_off(0); let bs_commitment_tx = nodes[1].tx_broadcaster.txn_broadcasted.lock().unwrap().split_off(0);
assert_eq!(bs_commitment_tx.len(), 1); assert_eq!(bs_commitment_tx.len(), 1);
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), true); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
reconnect_nodes(&nodes[0], &nodes[1], (false, false), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (false, false)); reconnect_nodes(&nodes[0], &nodes[1], (false, false), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (false, false));
if use_cs_commitment { if use_cs_commitment {
@ -1038,7 +1038,7 @@ fn removed_payment_no_manager_persistence() {
// Now that the ChannelManager has force-closed the channel which had the HTLC removed, it is // Now that the ChannelManager has force-closed the channel which had the HTLC removed, it is
// now forgotten everywhere. The ChannelManager should have, as a side-effect of reload, // now forgotten everywhere. The ChannelManager should have, as a side-effect of reload,
// learned that the HTLC is gone from the ChannelMonitor and added it to the to-fail-back set. // learned that the HTLC is gone from the ChannelMonitor and added it to the to-fail-back set.
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), true); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
reconnect_nodes(&nodes[0], &nodes[1], (false, false), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (false, false)); reconnect_nodes(&nodes[0], &nodes[1], (false, false), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (false, false));
expect_pending_htlcs_forwardable_and_htlc_handling_failed!(nodes[1], [HTLCDestination::NextHopChannel { node_id: Some(nodes[2].node.get_our_node_id()), channel_id: chan_id_2 }]); expect_pending_htlcs_forwardable_and_htlc_handling_failed!(nodes[1], [HTLCDestination::NextHopChannel { node_id: Some(nodes[2].node.get_our_node_id()), channel_id: chan_id_2 }]);

8
lightning/src/ln/shutdown_tests.rs
View file

@ -243,8 +243,8 @@ fn do_test_shutdown_rebroadcast(recv_count: u8) {
} }
} }
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
nodes[0].node.peer_connected(&nodes[1].node.get_our_node_id(), &msgs::Init { features: nodes[1].node.init_features(), remote_network_address: None }).unwrap(); nodes[0].node.peer_connected(&nodes[1].node.get_our_node_id(), &msgs::Init { features: nodes[1].node.init_features(), remote_network_address: None }).unwrap();
let node_0_reestablish = get_chan_reestablish_msgs!(nodes[0], nodes[1]).pop().unwrap(); let node_0_reestablish = get_chan_reestablish_msgs!(nodes[0], nodes[1]).pop().unwrap();
@ -305,8 +305,8 @@ fn do_test_shutdown_rebroadcast(recv_count: u8) {
assert!(node_0_2nd_closing_signed.is_some()); assert!(node_0_2nd_closing_signed.is_some());
} }
nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id(), false); nodes[0].node.peer_disconnected(&nodes[1].node.get_our_node_id());
nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id(), false); nodes[1].node.peer_disconnected(&nodes[0].node.get_our_node_id());
nodes[1].node.peer_connected(&nodes[0].node.get_our_node_id(), &msgs::Init { features: nodes[0].node.init_features(), remote_network_address: None }).unwrap(); nodes[1].node.peer_connected(&nodes[0].node.get_our_node_id(), &msgs::Init { features: nodes[0].node.init_features(), remote_network_address: None }).unwrap();
let node_1_2nd_reestablish = get_chan_reestablish_msgs!(nodes[1], nodes[0]).pop().unwrap(); let node_1_2nd_reestablish = get_chan_reestablish_msgs!(nodes[1], nodes[0]).pop().unwrap();

2
lightning/src/onion_message/messenger.rs
View file

@ -425,7 +425,7 @@ impl<ES: Deref, NS: Deref, L: Deref, CMH: Deref> OnionMessageHandler for OnionMe
Ok(()) Ok(())
} }
fn peer_disconnected(&self, their_node_id: &PublicKey, _no_connection_possible: bool) { fn peer_disconnected(&self, their_node_id: &PublicKey) {
let mut pending_msgs = self.pending_messages.lock().unwrap(); let mut pending_msgs = self.pending_messages.lock().unwrap();
pending_msgs.remove(their_node_id); pending_msgs.remove(their_node_id);
} }

9
lightning/src/util/test_utils.rs
View file

@ -331,6 +331,7 @@ impl chaininterface::BroadcasterInterface for TestBroadcaster {
pub struct TestChannelMessageHandler { pub struct TestChannelMessageHandler {
pub pending_events: Mutex<Vec<events::MessageSendEvent>>, pub pending_events: Mutex<Vec<events::MessageSendEvent>>,
expected_recv_msgs: Mutex<Option<Vec<wire::Message<()>>>>, expected_recv_msgs: Mutex<Option<Vec<wire::Message<()>>>>,
connected_peers: Mutex<HashSet<PublicKey>>,
} }
impl TestChannelMessageHandler { impl TestChannelMessageHandler {
@ -338,6 +339,7 @@ impl TestChannelMessageHandler {
TestChannelMessageHandler { TestChannelMessageHandler {
pending_events: Mutex::new(Vec::new()), pending_events: Mutex::new(Vec::new()),
expected_recv_msgs: Mutex::new(None), expected_recv_msgs: Mutex::new(None),
connected_peers: Mutex::new(HashSet::new()),
} }
} }
@ -422,8 +424,11 @@ impl msgs::ChannelMessageHandler for TestChannelMessageHandler {
fn handle_channel_reestablish(&self, _their_node_id: &PublicKey, msg: &msgs::ChannelReestablish) { fn handle_channel_reestablish(&self, _their_node_id: &PublicKey, msg: &msgs::ChannelReestablish) {
self.received_msg(wire::Message::ChannelReestablish(msg.clone())); self.received_msg(wire::Message::ChannelReestablish(msg.clone()));
} }
fn peer_disconnected(&self, _their_node_id: &PublicKey, _no_connection_possible: bool) {} fn peer_disconnected(&self, their_node_id: &PublicKey) {
fn peer_connected(&self, _their_node_id: &PublicKey, _msg: &msgs::Init) -> Result<(), ()> { assert!(self.connected_peers.lock().unwrap().remove(their_node_id));
}
fn peer_connected(&self, their_node_id: &PublicKey, _msg: &msgs::Init) -> Result<(), ()> {
assert!(self.connected_peers.lock().unwrap().insert(their_node_id.clone()));
// Don't bother with `received_msg` for Init as its auto-generated and we don't want to // Don't bother with `received_msg` for Init as its auto-generated and we don't want to
// bother re-generating the expected Init message in all tests. // bother re-generating the expected Init message in all tests.
Ok(()) Ok(())