Holding cell: if we fail to free an HTLC, fail it backwards

valentinewallace · valentinewallace · commit 3eb9e417428e · 2020-07-29T15:27:04.000-04:00
Plus add a test.
diff --git a/lightning/src/ln/channel.rs b/lightning/src/ln/channel.rs
@@ -2117,7 +2117,7 @@ impl<ChanSigner: ChannelKeys> Channel<ChanSigner> {
 
 	/// Used to fulfill holding_cell_htlcs when we get a remote ack (or implicitly get it by them
 	/// fulfilling or failing the last pending HTLC)
-	fn free_holding_cell_htlcs<L: Deref>(&mut self, logger: &L) -> Result<Option<(msgs::CommitmentUpdate, ChannelMonitorUpdate)>, ChannelError> where L::Target: Logger {
+	fn free_holding_cell_htlcs<L: Deref>(&mut self, logger: &L) -> Result<(Option<(msgs::CommitmentUpdate, ChannelMonitorUpdate)>, Vec<(HTLCSource, PaymentHash)>), ChannelError> where L::Target: Logger {
 		assert_eq!(self.channel_state & ChannelState::MonitorUpdateFailed as u32, 0);
 		if self.holding_cell_htlc_updates.len() != 0 || self.holding_cell_update_fee.is_some() {
 			log_trace!(logger, "Freeing holding cell with {} HTLC updates{}", self.holding_cell_htlc_updates.len(), if self.holding_cell_update_fee.is_some() { " and a fee update" } else { "" });
@@ -2132,6 +2132,7 @@ impl<ChanSigner: ChannelKeys> Channel<ChanSigner> {
 			let mut update_add_htlcs = Vec::with_capacity(htlc_updates.len());
 			let mut update_fulfill_htlcs = Vec::with_capacity(htlc_updates.len());
 			let mut update_fail_htlcs = Vec::with_capacity(htlc_updates.len());
+			let mut htlcs_to_fail = Vec::new();
 			let mut err = None;
 			for htlc_update in htlc_updates.drain(..) {
 				// Note that this *can* fail, though it should be due to rather-rare conditions on
@@ -2150,6 +2151,13 @@ impl<ChanSigner: ChannelKeys> Channel<ChanSigner> {
 									match e {
 										ChannelError::Ignore(ref msg) => {
 											log_info!(logger, "Failed to send HTLC with payment_hash {} due to {}", log_bytes!(payment_hash.0), msg);
+											// If we fail to send here, then this HTLC should
+											// be failed backwards. Failing to send here
+											// indicates that this HTLC may keep being put back
+											// into the holding cell without ever being
+											// successfully forwarded/failed/fulfilled, causing
+											// our counterparty to eventually close on us.
+											htlcs_to_fail.push((source.clone(), *payment_hash));
 										},
 										_ => {
 											log_info!(logger, "Failed to send HTLC with payment_hash {} resulting in a channel closure during holding_cell freeing", log_bytes!(payment_hash.0));
@@ -2188,24 +2196,27 @@ impl<ChanSigner: ChannelKeys> Channel<ChanSigner> {
 						},
 					}
 					if err.is_some() {
-						self.holding_cell_htlc_updates.push(htlc_update);
 						if let Some(ChannelError::Ignore(_)) = err {
 							// If we failed to add the HTLC, but got an Ignore error, we should
 							// still send the new commitment_signed, so reset the err to None.
+							// If we failed to fail or fulfill an HTLC, but got an Ignore error,
+							// it's OK to drop the error because these errors are caused by
+							// the ChannelManager generating duplicate claim/fail events during
+							// block rescan.
 							err = None;
+						} else {
+							self.holding_cell_htlc_updates.push(htlc_update);
 						}
 					}
 				}
 			}
-			//TODO: Need to examine the type of err - if it's a fee issue or similar we may want to
-			//fail it back the route, if it's a temporary issue we can ignore it...
 			match err {
 				None => {
 					if update_add_htlcs.is_empty() && update_fulfill_htlcs.is_empty() && update_fail_htlcs.is_empty() && self.holding_cell_update_fee.is_none() {
-						// This should never actually happen and indicates we got some Errs back
-						// from update_fulfill_htlc/update_fail_htlc, but we handle it anyway in
-						// case there is some strange way to hit duplicate HTLC removes.
-						return Ok(None);
+						// Hitting this case indicates that we got some Errs back from update_fulfill_htlc
+						// or update_fail_htlc.
+						log_warn!(logger, "Attempted to fulfill or fail an HTLC that was already removed");
+						return Ok((None, htlcs_to_fail));
 					}
 					let update_fee = if let Some(feerate) = self.holding_cell_update_fee {
 							self.pending_update_fee = self.holding_cell_update_fee.take();
@@ -2223,19 +2234,19 @@ impl<ChanSigner: ChannelKeys> Channel<ChanSigner> {
 					self.latest_monitor_update_id = monitor_update.update_id;
 					monitor_update.updates.append(&mut additional_update.updates);
 
-					Ok(Some((msgs::CommitmentUpdate {
+					Ok((Some((msgs::CommitmentUpdate {
 						update_add_htlcs,
 						update_fulfill_htlcs,
 						update_fail_htlcs,
 						update_fail_malformed_htlcs: Vec::new(),
 						update_fee: update_fee,
 						commitment_signed,
-					}, monitor_update)))
+					}, monitor_update)), htlcs_to_fail))
 				},
 				Some(e) => Err(e)
 			}
 		} else {
-			Ok(None)
+			Ok((None, Vec::new()))
 		}
 	}
 
@@ -2244,7 +2255,7 @@ impl<ChanSigner: ChannelKeys> Channel<ChanSigner> {
 	/// waiting on this revoke_and_ack. The generation of this new commitment_signed may also fail,
 	/// generating an appropriate error *after* the channel state has been updated based on the
 	/// revoke_and_ack message.
-	pub fn revoke_and_ack<F: Deref, L: Deref>(&mut self, msg: &msgs::RevokeAndACK, fee_estimator: &F, logger: &L) -> Result<(Option<msgs::CommitmentUpdate>, Vec<(PendingHTLCInfo, u64)>, Vec<(HTLCSource, PaymentHash, HTLCFailReason)>, Option<msgs::ClosingSigned>, ChannelMonitorUpdate), ChannelError>
+	pub fn revoke_and_ack<F: Deref, L: Deref>(&mut self, msg: &msgs::RevokeAndACK, fee_estimator: &F, logger: &L) -> Result<(Option<msgs::CommitmentUpdate>, Vec<(PendingHTLCInfo, u64)>, Vec<(HTLCSource, PaymentHash, HTLCFailReason)>, Option<msgs::ClosingSigned>, ChannelMonitorUpdate, Vec<(HTLCSource, PaymentHash)>), ChannelError>
 		where F::Target: FeeEstimator,
 					L::Target: Logger,
 	{
@@ -2419,11 +2430,11 @@ impl<ChanSigner: ChannelKeys> Channel<ChanSigner> {
 			}
 			self.monitor_pending_forwards.append(&mut to_forward_infos);
 			self.monitor_pending_failures.append(&mut revoked_htlcs);
-			return Ok((None, Vec::new(), Vec::new(), None, monitor_update))
+			return Ok((None, Vec::new(), Vec::new(), None, monitor_update, Vec::new()))
 		}
 
 		match self.free_holding_cell_htlcs(logger)? {
-			Some((mut commitment_update, mut additional_update)) => {
+			(Some((mut commitment_update, mut additional_update)), htlcs_to_fail) => {
 				commitment_update.update_fail_htlcs.reserve(update_fail_htlcs.len());
 				for fail_msg in update_fail_htlcs.drain(..) {
 					commitment_update.update_fail_htlcs.push(fail_msg);
@@ -2438,9 +2449,9 @@ impl<ChanSigner: ChannelKeys> Channel<ChanSigner> {
 				self.latest_monitor_update_id = monitor_update.update_id;
 				monitor_update.updates.append(&mut additional_update.updates);
 
-				Ok((Some(commitment_update), to_forward_infos, revoked_htlcs, None, monitor_update))
+				Ok((Some(commitment_update), to_forward_infos, revoked_htlcs, None, monitor_update, htlcs_to_fail))
 			},
-			None => {
+			(None, htlcs_to_fail) => {
 				if require_commitment {
 					let (commitment_signed, mut additional_update) = self.send_commitment_no_status_check(logger)?;
 
@@ -2456,9 +2467,9 @@ impl<ChanSigner: ChannelKeys> Channel<ChanSigner> {
 						update_fail_malformed_htlcs,
 						update_fee: None,
 						commitment_signed
-					}), to_forward_infos, revoked_htlcs, None, monitor_update))
+					}), to_forward_infos, revoked_htlcs, None, monitor_update, htlcs_to_fail))
 				} else {
-					Ok((None, to_forward_infos, revoked_htlcs, self.maybe_propose_first_closing_signed(fee_estimator), monitor_update))
+					Ok((None, to_forward_infos, revoked_htlcs, self.maybe_propose_first_closing_signed(fee_estimator), monitor_update, htlcs_to_fail))
 				}
 			}
 		}
@@ -2727,7 +2738,7 @@ impl<ChanSigner: ChannelKeys> Channel<ChanSigner> {
 
 	/// May panic if some calls other than message-handling calls (which will all Err immediately)
 	/// have been called between remove_uncommitted_htlcs_and_mark_paused and this call.
-	pub fn channel_reestablish<L: Deref>(&mut self, msg: &msgs::ChannelReestablish, logger: &L) -> Result<(Option<msgs::FundingLocked>, Option<msgs::RevokeAndACK>, Option<msgs::CommitmentUpdate>, Option<ChannelMonitorUpdate>, RAACommitmentOrder, Option<msgs::Shutdown>), ChannelError> where L::Target: Logger {
+	pub fn channel_reestablish<L: Deref>(&mut self, msg: &msgs::ChannelReestablish, logger: &L) -> Result<(Option<msgs::FundingLocked>, Option<msgs::RevokeAndACK>, Option<msgs::CommitmentUpdate>, Option<ChannelMonitorUpdate>, Vec<(HTLCSource, PaymentHash)>, RAACommitmentOrder, Option<msgs::Shutdown>), ChannelError> where L::Target: Logger {
 		if self.channel_state & (ChannelState::PeerDisconnected as u32) == 0 {
 			// While BOLT 2 doesn't indicate explicitly we should error this channel here, it
 			// almost certainly indicates we are going to end up out-of-sync in some way, so we
@@ -2778,15 +2789,15 @@ impl<ChanSigner: ChannelKeys> Channel<ChanSigner> {
 					return Err(ChannelError::Close("Peer claimed they saw a revoke_and_ack but we haven't sent funding_locked yet".to_owned()));
 				}
 				// Short circuit the whole handler as there is nothing we can resend them
-				return Ok((None, None, None, None, RAACommitmentOrder::CommitmentFirst, shutdown_msg));
+				return Ok((None, None, None, None, Vec::new(), RAACommitmentOrder::CommitmentFirst, shutdown_msg));
 			}
 
 			// We have OurFundingLocked set!
 			let next_per_commitment_point = self.local_keys.get_per_commitment_point(self.cur_local_commitment_transaction_number, &self.secp_ctx);
 			return Ok((Some(msgs::FundingLocked {
 				channel_id: self.channel_id(),
 				next_per_commitment_point: next_per_commitment_point,
-			}), None, None, None, RAACommitmentOrder::CommitmentFirst, shutdown_msg));
+			}), None, None, None, Vec::new(), RAACommitmentOrder::CommitmentFirst, shutdown_msg));
 		}
 
 		let required_revoke = if msg.next_remote_commitment_number + 1 == INITIAL_COMMITMENT_NUMBER - self.cur_local_commitment_transaction_number {
@@ -2834,11 +2845,11 @@ impl<ChanSigner: ChannelKeys> Channel<ChanSigner> {
 				match self.free_holding_cell_htlcs(logger) {
 					Err(ChannelError::Close(msg)) => return Err(ChannelError::Close(msg)),
 					Err(ChannelError::Ignore(_)) | Err(ChannelError::CloseDelayBroadcast(_)) => panic!("Got non-channel-failing result from free_holding_cell_htlcs"),
-					Ok(Some((commitment_update, monitor_update))) => return Ok((resend_funding_locked, required_revoke, Some(commitment_update), Some(monitor_update), self.resend_order.clone(), shutdown_msg)),
-					Ok(None) => return Ok((resend_funding_locked, required_revoke, None, None, self.resend_order.clone(), shutdown_msg)),
+					Ok((Some((commitment_update, monitor_update)), htlcs_to_fail)) => return Ok((resend_funding_locked, required_revoke, Some(commitment_update), Some(monitor_update), htlcs_to_fail, self.resend_order.clone(), shutdown_msg)),
+					Ok((None, htlcs_to_fail)) => return Ok((resend_funding_locked, required_revoke, None, None, htlcs_to_fail, self.resend_order.clone(), shutdown_msg)),
 				}
 			} else {
-				return Ok((resend_funding_locked, required_revoke, None, None, self.resend_order.clone(), shutdown_msg));
+				return Ok((resend_funding_locked, required_revoke, None, None, Vec::new(), self.resend_order.clone(), shutdown_msg));
 			}
 		} else if msg.next_local_commitment_number == our_next_remote_commitment_number - 1 {
 			if required_revoke.is_some() {
@@ -2849,10 +2860,10 @@ impl<ChanSigner: ChannelKeys> Channel<ChanSigner> {
 
 			if self.channel_state & (ChannelState::MonitorUpdateFailed as u32) != 0 {
 				self.monitor_pending_commitment_signed = true;
-				return Ok((resend_funding_locked, None, None, None, self.resend_order.clone(), shutdown_msg));
+				return Ok((resend_funding_locked, None, None, None, Vec::new(), self.resend_order.clone(), shutdown_msg));
 			}
 
-			return Ok((resend_funding_locked, required_revoke, Some(self.get_last_commitment_update(logger)), None, self.resend_order.clone(), shutdown_msg));
+			return Ok((resend_funding_locked, required_revoke, Some(self.get_last_commitment_update(logger)), None, Vec::new(), self.resend_order.clone(), shutdown_msg));
 		} else {
 			return Err(ChannelError::Close("Peer attempted to reestablish channel with a very old remote commitment transaction".to_owned()));
 		}
diff --git a/lightning/src/ln/channelmanager.rs b/lightning/src/ln/channelmanager.rs
@@ -1822,6 +1822,48 @@ impl<ChanSigner: ChannelKeys, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>
 		} else { false }
 	}
 
+	// Fail a list of HTLCs that were just freed from the holding cell. The HTLCs need to be
+	// failed backwards or, if they were one of our outgoing HTLCs, then their failure needs to
+	// be surfaced to the user.
+	fn fail_holding_cell_htlcs(&self, mut htlcs_to_fail: Vec<(HTLCSource, PaymentHash)>, channel_id: [u8; 32]) -> Result<(), MsgHandleErrInternal> {
+		for (htlc_src, payment_hash) in htlcs_to_fail.drain(..) {
+			match htlc_src {
+				HTLCSource::PreviousHopData(HTLCPreviousHopData { .. }) => {
+					let onion_failure_data =
+						match self.channel_state.lock().unwrap().by_id.entry(channel_id) {
+							hash_map::Entry::Occupied(chan_entry) => {
+								if let Ok(upd) = self.get_channel_update(&chan_entry.get()) {
+									let mut res = Vec::with_capacity(8 + 128);
+									res.extend_from_slice(&byte_utils::be16_to_array(upd.contents.flags));
+									res.extend_from_slice(&upd.encode_with_len()[..]);
+									res
+								} else {
+									return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel", channel_id));
+								}
+							},
+							hash_map::Entry::Vacant(_) => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel", channel_id))
+						};
+						let channel_state = self.channel_state.lock().unwrap();
+						self.fail_htlc_backwards_internal(channel_state,
+							htlc_src, &payment_hash, HTLCFailReason::Reason { failure_code: 0x1000|7, data: onion_failure_data});
+				},
+				HTLCSource::OutboundRoute { .. } => {
+					self.pending_events.lock().unwrap().push(
+						events::Event::PaymentFailed {
+							payment_hash,
+							rejected_by_dest: false,
+#[cfg(test)]
+							error_code: None,
+#[cfg(test)]
+							error_data: None,
+						}
+					)
+				},
+			};
+		}
+		Ok(())
+	}
+
 	/// Fails an HTLC backwards to the sender of it to us.
 	/// Note that while we take a channel_state lock as input, we do *not* assume consistency here.
 	/// There are several callsites that do stupid things like loop over a list of payment_hashes
@@ -2670,7 +2712,7 @@ impl<ChanSigner: ChannelKeys, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>
 	}
 
 	fn internal_revoke_and_ack(&self, their_node_id: &PublicKey, msg: &msgs::RevokeAndACK) -> Result<(), MsgHandleErrInternal> {
-		let (pending_forwards, mut pending_failures, short_channel_id) = {
+		let (pending_forwards, mut pending_failures, short_channel_id, htlcs_to_fail) = {
 			let mut channel_state_lock = self.channel_state.lock().unwrap();
 			let channel_state = &mut *channel_state_lock;
 			match channel_state.by_id.entry(msg.channel_id) {
@@ -2679,7 +2721,7 @@ impl<ChanSigner: ChannelKeys, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>
 						return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!".to_owned(), msg.channel_id));
 					}
 					let was_frozen_for_monitor = chan.get().is_awaiting_monitor_update();
-					let (commitment_update, pending_forwards, pending_failures, closing_signed, monitor_update) =
+					let (commitment_update, pending_forwards, pending_failures, closing_signed, monitor_update, htlcs_to_fail) =
 						try_chan_entry!(self, chan.get_mut().revoke_and_ack(&msg, &self.fee_estimator, &self.logger), channel_state, chan);
 					if let Err(e) = self.monitor.update_monitor(chan.get().get_funding_txo().unwrap(), monitor_update) {
 						if was_frozen_for_monitor {
@@ -2701,14 +2743,15 @@ impl<ChanSigner: ChannelKeys, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>
 							msg,
 						});
 					}
-					(pending_forwards, pending_failures, chan.get().get_short_channel_id().expect("RAA should only work on a short-id-available channel"))
+					(pending_forwards, pending_failures, chan.get().get_short_channel_id().expect("RAA should only work on a short-id-available channel"), htlcs_to_fail)
 				},
 				hash_map::Entry::Vacant(_) => return Err(MsgHandleErrInternal::send_err_msg_no_close("Failed to find corresponding channel".to_owned(), msg.channel_id))
 			}
 		};
 		for failure in pending_failures.drain(..) {
 			self.fail_htlc_backwards_internal(self.channel_state.lock().unwrap(), failure.0, &failure.1, failure.2);
 		}
+		if let Err(e) = self.fail_holding_cell_htlcs(htlcs_to_fail, msg.channel_id) { return Err(e) };
 		self.forward_htlcs(&mut [(short_channel_id, pending_forwards)]);
 
 		Ok(())
@@ -2792,7 +2835,7 @@ impl<ChanSigner: ChannelKeys, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>
 				if chan.get().get_their_node_id() != *their_node_id {
 					return Err(MsgHandleErrInternal::send_err_msg_no_close("Got a message for a channel from the wrong node!".to_owned(), msg.channel_id));
 				}
-				let (funding_locked, revoke_and_ack, commitment_update, monitor_update_opt, mut order, shutdown) =
+				let (funding_locked, revoke_and_ack, commitment_update, monitor_update_opt, htlcs_to_fail, mut order, shutdown) =
 					try_chan_entry!(self, chan.get_mut().channel_reestablish(msg, &self.logger), channel_state, chan);
 				if let Some(monitor_update) = monitor_update_opt {
 					if let Err(e) = self.monitor.update_monitor(chan.get().get_funding_txo().unwrap(), monitor_update) {
@@ -2809,6 +2852,7 @@ impl<ChanSigner: ChannelKeys, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>
 						//TODO: Resend the funding_locked if needed once we get the monitor running again
 					}
 				}
+				if let Err(e) = self.fail_holding_cell_htlcs(htlcs_to_fail, msg.channel_id) { return Err(e) };
 				if let Some(msg) = funding_locked {
 					channel_state.pending_msg_events.push(events::MessageSendEvent::SendFundingLocked {
 						node_id: their_node_id.clone(),
diff --git a/lightning/src/ln/functional_tests.rs b/lightning/src/ln/functional_tests.rs
