Skip to content

Forward over substitute channels #1578

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
ViktorT-11 wants to merge 21 commits into from
Closed

Forward over substitute channels #1578

Changes from 1 commit
Commits
Show all changes
21 commits
Select commit Hold shift + click to select a range
9864a26
Add `counterparty_node_id` to `short_to_id` map
ViktorT-11 May 17, 2022
0780fb9
Rename `short_to_id` map to `short_to_chan_info`
ViktorT-11 Jun 9, 2022
7833589
Add id_to_peer map
ViktorT-11 Jun 3, 2022
b0975eb
Add `ChannelManager:id_to_peer` map coverage test
ViktorT-11 Jun 6, 2022
d3667d9
Add `counterparty_node` to test macros
ViktorT-11 May 27, 2022
6c62533
Store channels per-peer
ViktorT-11 May 25, 2022
519fcb4
f - Store per peer: aquire locks inside get_channel_ref
ViktorT-11 Jun 9, 2022
c37b984
f - Make per_peer_state a FairRwLock
ViktorT-11 Jun 6, 2022
f7b51d2
f - Remove unreacable branches that can be reached
ViktorT-11 Jun 6, 2022
2c112d7
f - Store channels per-peer: Update per_peer_state docs
ViktorT-11 Jun 7, 2022
7b594ee
Remove unnecessary `per_peer_state` branch
ViktorT-11 Jun 10, 2022
7e9ac7b
Avoid retaking locks
ViktorT-11 Jun 6, 2022
6c4b80d
Update failure to query `Channel` error messages
ViktorT-11 Jun 6, 2022
4991456
Add duplicate temporary_channel_id for 2 peers test
ViktorT-11 May 31, 2022
ae665ce
Add handle unkown peer test
ViktorT-11 Jun 13, 2022
ef4107a
Enable htlc forwarding over substitute channels
ViktorT-11 Jun 20, 2022
1d44945
Test htlc forwarding over substitute channels
ViktorT-11 Jun 20, 2022
1dec1f9
f - Move creation of CommitmentUpdate structs
ViktorT-11 Jun 27, 2022
6b9a261
-f DRY up htlc_msg macros
ViktorT-11 Jun 27, 2022
13dbfe3
f - cleanup channel selection order
ViktorT-11 Jun 29, 2022
92a067c
f - cleanup tests
ViktorT-11 Jun 29, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
154 changes: 79 additions & 75 deletions lightning/src/ln/channelmanager.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3005,6 +3005,36 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
let channel_state = &mut *channel_state_lock;
let per_peer_state = self.per_peer_state.read().unwrap();

let mut htlcs_msgs_by_id: HashMap<[u8; 32], (Vec<msgs::UpdateAddHTLC>, Vec<msgs::UpdateFailHTLC>, PublicKey)> = HashMap::new();

macro_rules! add_channel_key {
($channel_id: expr, $counterparty_node_id: expr) => {{
if !htlcs_msgs_by_id.contains_key(&$channel_id){
htlcs_msgs_by_id.insert($channel_id, (Vec::new(), Vec::new(), $counterparty_node_id));
}
}}
}

macro_rules! add_update_add_htlc {
($add_htlc_msg: expr, $channel_id: expr, $counterparty_node_id: expr) => {{
add_channel_key!($channel_id, $counterparty_node_id);
if let hash_map::Entry::Occupied(mut entry) = htlcs_msgs_by_id.entry($channel_id) {
let msgs_entry = entry.get_mut();
msgs_entry.0.push($add_htlc_msg);
}
}}
}

macro_rules! add_update_fail_htlc {
($fail_htlc_msg: expr, $channel_id: expr, $counterparty_node_id: expr) => {{
add_channel_key!($channel_id, $counterparty_node_id);
if let hash_map::Entry::Occupied(mut entry) = htlcs_msgs_by_id.entry($channel_id) {
let msgs_entry = entry.get_mut();
msgs_entry.1.push($fail_htlc_msg);
}
}}
}

for (short_chan_id, mut pending_forwards) in channel_state.forward_htlcs.drain() {
if short_chan_id != 0 {
let (counterparty_node_id, forward_chan_id) = match channel_state.short_to_chan_info.get(&short_chan_id) {
Expand Down Expand Up @@ -3082,35 +3112,6 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
let mut peer_state_lock = peer_state_mutex.lock().unwrap();
let peer_state = &mut *peer_state_lock;
if peer_state.channel_by_id.contains_key(&forward_chan_id) {
let mut htlcs_msgs_by_id: HashMap<[u8; 32], (Vec<msgs::UpdateAddHTLC>, Vec<msgs::UpdateFailHTLC>)> = HashMap::new();

macro_rules! add_channel_key {
($channel_id: expr) => {{
if !htlcs_msgs_by_id.contains_key(&$channel_id){
htlcs_msgs_by_id.insert($channel_id, (Vec::new(), Vec::new()));
}
}}
}

macro_rules! add_update_add_htlc {
($add_htlc_msg: expr, $channel_id: expr) => {{
add_channel_key!($channel_id);
if let hash_map::Entry::Occupied(mut entry) = htlcs_msgs_by_id.entry($channel_id) {
let msgs_entry = entry.get_mut();
msgs_entry.0.push($add_htlc_msg);
}
}}
}

macro_rules! add_update_fail_htlc {
($fail_htlc_msg: expr, $channel_id: expr) => {{
add_channel_key!($channel_id);
if let hash_map::Entry::Occupied(mut entry) = htlcs_msgs_by_id.entry($channel_id) {
let msgs_entry = entry.get_mut();
msgs_entry.1.push($fail_htlc_msg);
}
}}
}
for forward_info in pending_forwards.drain(..) {
match forward_info {
HTLCForwardInfo::AddHTLC { prev_short_channel_id, prev_htlc_id, forward_info: PendingHTLCInfo {
Expand Down Expand Up @@ -3153,7 +3154,7 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
match update_add {
Some(msg) => {
log_info!(self.logger, "Will forward HTLC with payment_hash {}, over channel {}", log_bytes!(payment_hash.0), log_bytes!(chan_id));
add_update_add_htlc!(msg, chan_id);
add_update_add_htlc!(msg, chan_id, counterparty_node_id);
},
None => {
// Nothing to do here...we're waiting on a remote
Expand Down Expand Up @@ -3196,7 +3197,7 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
// the chain and sending the HTLC-Timeout is their problem.
continue;
},
Ok(Some(msg)) => { add_update_fail_htlc!(msg, forward_chan_id); },
Ok(Some(msg)) => { add_update_fail_htlc!(msg, forward_chan_id, counterparty_node_id); },
Ok(None) => {
// Nothing to do here...we're waiting on a remote
// revoke_and_ack before we can update the commitment
Expand All @@ -3211,50 +3212,6 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
},
}
}

for (chan_id, (add_htlc_msgs, fail_htlc_msgs)) in htlcs_msgs_by_id.into_iter() {
if let hash_map::Entry::Occupied(mut chan) = peer_state.channel_by_id.entry(chan_id) {
let (commitment_msg, monitor_update) = match chan.get_mut().send_commitment(&self.logger) {
Ok(res) => res,
Err(e) => {
// We surely failed send_commitment due to bad keys, in that case
// close channel and then send error message to peer.
let counterparty_node_id = chan.get().get_counterparty_node_id();
let err: Result<(), _> = match e {
ChannelError::Ignore(_) | ChannelError::Warn(_) => {
panic!("Stated return value requirements in send_commitment() were not met");
}
ChannelError::Close(msg) => {
log_trace!(self.logger, "Closing channel {} due to Close-required error: {}", log_bytes!(chan.key()[..]), msg);
let mut channel = remove_channel!(self, channel_state, chan);
// ChannelClosed event is generated by handle_error for us.
Err(MsgHandleErrInternal::from_finish_shutdown(msg, channel.channel_id(), channel.get_user_id(), channel.force_shutdown(true), self.get_channel_update_for_broadcast(&channel).ok()))
},
ChannelError::CloseDelayBroadcast(_) => { panic!("Wait is only generated on receipt of channel_reestablish, which is handled by try_chan_entry, we don't bother to support it here"); }
};
handle_errors.push((counterparty_node_id, err));
continue;
}
};
if let Err(e) = self.chain_monitor.update_channel(chan.get().get_funding_txo().unwrap(), monitor_update) {
handle_errors.push((chan.get().get_counterparty_node_id(), handle_monitor_err!(self, e, channel_state, chan, RAACommitmentOrder::CommitmentFirst, false, true)));
continue;
}
log_debug!(self.logger, "Forwarding HTLCs resulted in a commitment update with {} HTLCs added and {} HTLCs failed for channel {}",
add_htlc_msgs.len(), fail_htlc_msgs.len(), log_bytes!(chan.get().channel_id()));
channel_state.pending_msg_events.push(events::MessageSendEvent::UpdateHTLCs {
node_id: chan.get().get_counterparty_node_id(),
updates: msgs::CommitmentUpdate {
update_add_htlcs: add_htlc_msgs,
update_fulfill_htlcs: Vec::new(),
update_fail_htlcs: fail_htlc_msgs,
update_fail_malformed_htlcs: Vec::new(),
update_fee: None,
commitment_signed: commitment_msg,
},
});
}
}
} else {
let err = Err(MsgHandleErrInternal::send_err_msg_no_close(format!("No such channel for the counterparty_node_id {}, as indicated by the short_to_id map", counterparty_node_id), forward_chan_id));
handle_errors.push((counterparty_node_id, err));
Expand Down Expand Up @@ -3438,6 +3395,53 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
}
}
}
for (chan_id, (add_htlc_msgs, fail_htlc_msgs, counterparty_node_id)) in htlcs_msgs_by_id.into_iter() {

@ViktorT-11 ViktorT-11 Jun 28, 2022 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Am I correct that we'd like to move this out of the channel_state.forward_htlcs loop, to avoid that we else may create multiple CommitmentUpdate instances for the same channels, despite that we'll then need to drop and reacquire the PeerState mutex for that peer?

This will happen in cases where channel_state.forward_htlcs contains one htlc that is successfully forwarded over a specified channel (a), and another htlc that is forwarded over that same channel through substitution.

if let Some(peer_state_mutex) = per_peer_state.get(&counterparty_node_id) {
let mut peer_state_lock = peer_state_mutex.lock().unwrap();
let peer_state = &mut *peer_state_lock;
if let hash_map::Entry::Occupied(mut chan) = peer_state.channel_by_id.entry(chan_id) {
let (commitment_msg, monitor_update) = match chan.get_mut().send_commitment(&self.logger) {
Ok(res) => res,
Err(e) => {
// We surely failed send_commitment due to bad keys, in that case
// close channel and then send error message to peer.
let counterparty_node_id = chan.get().get_counterparty_node_id();
let err: Result<(), _> = match e {
ChannelError::Ignore(_) | ChannelError::Warn(_) => {
panic!("Stated return value requirements in send_commitment() were not met");
}
ChannelError::Close(msg) => {
log_trace!(self.logger, "Closing channel {} due to Close-required error: {}", log_bytes!(chan.key()[..]), msg);
let mut channel = remove_channel!(self, channel_state, chan);
// ChannelClosed event is generated by handle_error for us.
Err(MsgHandleErrInternal::from_finish_shutdown(msg, channel.channel_id(), channel.get_user_id(), channel.force_shutdown(true), self.get_channel_update_for_broadcast(&channel).ok()))
},
ChannelError::CloseDelayBroadcast(_) => { panic!("Wait is only generated on receipt of channel_reestablish, which is handled by try_chan_entry, we don't bother to support it here"); }
};
handle_errors.push((counterparty_node_id, err));
continue;
}
};
if let Err(e) = self.chain_monitor.update_channel(chan.get().get_funding_txo().unwrap(), monitor_update) {
handle_errors.push((chan.get().get_counterparty_node_id(), handle_monitor_err!(self, e, channel_state, chan, RAACommitmentOrder::CommitmentFirst, false, true)));
continue;
}
log_debug!(self.logger, "Forwarding HTLCs resulted in a commitment update with {} HTLCs added and {} HTLCs failed for channel {}",
add_htlc_msgs.len(), fail_htlc_msgs.len(), log_bytes!(chan.get().channel_id()));
channel_state.pending_msg_events.push(events::MessageSendEvent::UpdateHTLCs {
node_id: chan.get().get_counterparty_node_id(),
updates: msgs::CommitmentUpdate {
update_add_htlcs: add_htlc_msgs,
update_fulfill_htlcs: Vec::new(),
update_fail_htlcs: fail_htlc_msgs,
update_fail_malformed_htlcs: Vec::new(),
update_fee: None,
commitment_signed: commitment_msg,
},
});
}
}
}
}

for (htlc_source, payment_hash, failure_reason) in failed_forwards.drain(..) {
Expand Down