channelmanager: Add retry data to pending_outbound_payments

Author: valentinewallace

lightning/src/ln/channelmanager.rs

@@ -400,6 +400,48 @@ struct PendingInboundPayment {
 	min_value_msat: Option<u64>,
 }
 
+/// XXX
+enum PendingOutboundPayment {
+	Legacy {
+		session_privs: HashSet<[u8; 32]>,
+	},
+	Retryable {
+		session_privs: HashSet<[u8; 32]>,
+		payment_hash: PaymentHash,
+		total_msat: u64,
+		// XXX add payment_secret
+	},
+}
+
+impl PendingOutboundPayment {
+	fn remove(&mut self, session_priv: &[u8; 32]) -> bool {
+		match self {
+			PendingOutboundPayment::Legacy { session_privs } |
+			PendingOutboundPayment::Retryable { session_privs, .. } => {
+				session_privs.remove(session_priv)
+			}
+		}
+	}
+
+	fn insert(&mut self, session_priv: [u8; 32]) -> bool {
+		match self {
+			PendingOutboundPayment::Legacy { session_privs } |
+			PendingOutboundPayment::Retryable { session_privs, .. } => {
+				session_privs.insert(session_priv)
+			}
+		}
+	}
+
+	fn num_parts(&self) -> usize {
+		match self {
+			PendingOutboundPayment::Legacy { session_privs } |
+			PendingOutboundPayment::Retryable { session_privs, .. } => {
+				session_privs.len()
+			}
+		}
+	}
+}
+
 /// SimpleArcChannelManager is useful when you need a ChannelManager with a static lifetime, e.g.
 /// when you're using lightning-net-tokio (since tokio::spawn requires parameters with static
 /// lifetimes). Other times you can afford a reference, which is more efficient, in which case
@@ -486,6 +528,8 @@ pub struct ChannelManager<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref,
 	/// Locked *after* channel_state.
 	pending_inbound_payments: Mutex<HashMap<PaymentHash, PendingInboundPayment>>,
 
+	/// XXX update these
+	///
 	/// The session_priv bytes of outbound payments which are pending resolution.
 	/// The authoritative state of these HTLCs resides either within Channels or ChannelMonitors
 	/// (if the channel has been force-closed), however we track them here to prevent duplicative
@@ -499,7 +543,7 @@ pub struct ChannelManager<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref,
 	/// payments over a single path).
 	///
 	/// Locked *after* channel_state.
-	pending_outbound_payments: Mutex<HashMap<PaymentId, HashSet<[u8; 32]>>>,
+	pending_outbound_payments: Mutex<HashMap<PaymentId, PendingOutboundPayment>>,
 
 	our_network_key: SecretKey,
 	our_network_pubkey: PublicKey,
@@ -1894,8 +1938,12 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
 
 		let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(&self.total_consistency_lock, &self.persistence_notifier);
 		let mut pending_outbounds = self.pending_outbound_payments.lock().unwrap();
-		let sessions = pending_outbounds.entry(payment_id).or_insert(HashSet::new());
-		assert!(sessions.insert(session_priv_bytes));
+		let payment = pending_outbounds.entry(payment_id).or_insert(PendingOutboundPayment::Retryable {
+			session_privs: HashSet::new(),
+			payment_hash: *payment_hash,
+			total_msat: total_value,
+		});
+		payment.insert(session_priv_bytes);
 
 		let err: Result<(), _> = loop {
 			let mut channel_lock = self.channel_state.lock().unwrap();
@@ -2883,23 +2931,23 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
 					let mut session_priv_bytes = [0; 32];
 					session_priv_bytes.copy_from_slice(&session_priv[..]);
 					let mut outbounds = self.pending_outbound_payments.lock().unwrap();
-					if let hash_map::Entry::Occupied(mut sessions) = outbounds.entry(payment_id) {
-						if sessions.get_mut().remove(&session_priv_bytes) {
+					if let hash_map::Entry::Occupied(mut payment) = outbounds.entry(payment_id) {
+						if payment.get_mut().remove(&session_priv_bytes) {
 							self.pending_events.lock().unwrap().push(
 								events::Event::PaymentPathFailed {
 									payment_hash,
 									rejected_by_dest: false,
 									network_update: None,
-									all_paths_failed: sessions.get().len() == 0,
+									all_paths_failed: payment.get().num_parts() == 0,
 									path: path.clone(),
 									#[cfg(test)]
 									error_code: None,
 									#[cfg(test)]
 									error_data: None,
 								}
 							);
-							if sessions.get().len() == 0 {
-								sessions.remove();
+							if payment.get().num_parts() == 0 {
+								payment.remove();
 							}
 						}
 					} else {
@@ -2936,7 +2984,7 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> ChannelMana
 						log_trace!(self.logger, "Received duplicative fail for HTLC with payment_hash {}", log_bytes!(payment_hash.0));
 						return;
 					}
-					if sessions.get().len() == 0 {
+					if sessions.get().num_parts() == 0 {
 						all_paths_failed = true;
 						sessions.remove();
 					}
@@ -5161,6 +5209,17 @@ impl_writeable_tlv_based!(PendingInboundPayment, {
 	(8, min_value_msat, required),
 });
 
+impl_writeable_tlv_based_enum!(PendingOutboundPayment,
+	(0, Legacy) => {
+		(0, session_privs, required),
+	},
+	(2, Retryable) => {
+		(0, session_privs, required),
+		(2, payment_hash, required),
+		(4, total_msat, required),
+	},
+;);
+
 impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> Writeable for ChannelManager<Signer, M, T, K, F, L>
 	where M::Target: chain::Watch<Signer>,
         T::Target: BroadcasterInterface,
@@ -5251,18 +5310,30 @@ impl<Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref> Writeable f
 		let pending_outbound_payments = self.pending_outbound_payments.lock().unwrap();
 		// For backwards compat, write the session privs and their total length.
 		let mut num_pending_outbounds_compat: u64 = 0;
-		for (_, outbounds) in pending_outbound_payments.iter() {
-			num_pending_outbounds_compat += outbounds.len() as u64;
+		for (_, outbound) in pending_outbound_payments.iter() {
+			match outbound {
+				PendingOutboundPayment::Legacy { session_privs } |
+				PendingOutboundPayment::Retryable { session_privs, .. } => {
+					num_pending_outbounds_compat += session_privs.len() as u64;
+				}
+			}
 		}
 		num_pending_outbounds_compat.write(writer)?;
-		for (_, outbounds) in pending_outbound_payments.iter() {
-			for outbound in outbounds.iter() {
-				outbound.write(writer)?;
+		for (_, outbound) in pending_outbound_payments.iter() {
+			match outbound {
+				PendingOutboundPayment::Legacy { session_privs } |
+				PendingOutboundPayment::Retryable { session_privs, .. } => {
+					for session_priv in session_privs.iter() {
+						session_priv.write(writer)?;
+					}
+				}
 			}
 		}
 
+		let pending_outbound_payments_compat_2: HashMap<PaymentId, HashSet<[u8; 32]>> = HashMap::new();
 		write_tlv_fields!(writer, {
-			(1, pending_outbound_payments, required),
+			(1, pending_outbound_payments_compat_2, required),
+			(3, pending_outbound_payments, required),
 		});
 
 		Ok(())
@@ -5522,21 +5593,32 @@ impl<'a, Signer: Sign, M: Deref, T: Deref, K: Deref, F: Deref, L: Deref>
 		}
 
 		let pending_outbound_payments_count_compat: u64 = Readable::read(reader)?;
-		let mut pending_outbound_payments_compat: HashMap<PaymentId, HashSet<[u8; 32]>> =
+		let mut pending_outbound_payments_compat: HashMap<PaymentId, PendingOutboundPayment> =
 			HashMap::with_capacity(cmp::min(pending_outbound_payments_count_compat as usize, MAX_ALLOC_SIZE/32));
 		for _ in 0..pending_outbound_payments_count_compat {
 			let session_priv = Readable::read(reader)?;
-			if pending_outbound_payments_compat.insert(PaymentId(session_priv), [session_priv].iter().cloned().collect()).is_some() {
+			let payment = PendingOutboundPayment::Legacy {
+				session_privs: [session_priv].iter().cloned().collect()
+			};
+			if pending_outbound_payments_compat.insert(PaymentId(session_priv), payment).is_some() {
 				return Err(DecodeError::InvalidValue)
 			};
 		}
 
+		let mut pending_outbound_payments_compat_2: Option<HashMap<PaymentId, HashSet<[u8; 32]>>> = None;
 		let mut pending_outbound_payments = None;
 		read_tlv_fields!(reader, {
-			(1, pending_outbound_payments, option),
+			(1, pending_outbound_payments_compat_2, option),
+			(3, pending_outbound_payments, option),
 		});
-		if pending_outbound_payments.is_none() {
+		if pending_outbound_payments.is_none() && pending_outbound_payments_compat_2.is_none() {
 			pending_outbound_payments = Some(pending_outbound_payments_compat);
+		} else if pending_outbound_payments.is_none() {
+			let mut outbounds = HashMap::new();
+			for (id, session_privs) in pending_outbound_payments_compat_2.unwrap().drain() {
+				outbounds.insert(id, PendingOutboundPayment::Legacy { session_privs });
+			}
+			pending_outbound_payments = Some(outbounds);
 		}
 
 		let mut secp_ctx = Secp256k1::new();