Add inital implementation of persisted payment store

Author: tnull

src/event.rs

@@ -1,6 +1,6 @@
 use crate::{
-	hex_utils, ChannelManager, Config, Error, KeysManager, NetworkGraph, PaymentInfo,
-	PaymentInfoStorage, PaymentStatus, Wallet,
+	hex_utils, ChannelManager, Config, Error, KeysManager, NetworkGraph, PaymentDirection,
+	PaymentInfo, PaymentInfoStorage, PaymentStatus, Wallet,
 };
 
 use crate::logger::{log_error, log_info, Logger};
@@ -18,7 +18,7 @@ use lightning::util::ser::{Readable, ReadableArgs, Writeable, Writer};
 
 use bitcoin::secp256k1::Secp256k1;
 use rand::{thread_rng, Rng};
-use std::collections::{hash_map, VecDeque};
+use std::collections::VecDeque;
 use std::ops::Deref;
 use std::sync::{Arc, Condvar, Mutex};
 use std::time::Duration;
@@ -191,8 +191,7 @@ where
 	channel_manager: Arc<ChannelManager>,
 	network_graph: Arc<NetworkGraph>,
 	keys_manager: Arc<KeysManager>,
-	inbound_payments: Arc<PaymentInfoStorage>,
-	outbound_payments: Arc<PaymentInfoStorage>,
+	payment_store: Arc<PaymentInfoStorage<K>>,
 	tokio_runtime: Arc<tokio::runtime::Runtime>,
 	logger: L,
 	_config: Arc<Config>,
@@ -206,18 +205,16 @@ where
 	pub fn new(
 		wallet: Arc<Wallet<bdk::database::SqliteDatabase>>, event_queue: Arc<EventQueue<K>>,
 		channel_manager: Arc<ChannelManager>, network_graph: Arc<NetworkGraph>,
-		keys_manager: Arc<KeysManager>, inbound_payments: Arc<PaymentInfoStorage>,
-		outbound_payments: Arc<PaymentInfoStorage>, tokio_runtime: Arc<tokio::runtime::Runtime>,
-		logger: L, _config: Arc<Config>,
+		keys_manager: Arc<KeysManager>, payment_store: Arc<PaymentInfoStorage<K>>,
+		tokio_runtime: Arc<tokio::runtime::Runtime>, logger: L, _config: Arc<Config>,
 	) -> Self {
 		Self {
 			event_queue,
 			wallet,
 			channel_manager,
 			network_graph,
 			keys_manager,
-			inbound_payments,
-			outbound_payments,
+			payment_store,
 			logger,
 			tokio_runtime,
 			_config,
@@ -298,9 +295,25 @@ where
 				via_channel_id: _,
 				via_user_channel_id: _,
 			} => {
+				if let Some(info) = self.payment_store.get(&payment_hash) {
+					if info.status == PaymentStatus::Succeeded {
+						log_info!(
+							self.logger,
+							"Refused duplicate inbound payment from payment hash {} of {}msat",
+							hex_utils::to_string(&payment_hash.0),
+							amount_msat,
+						);
+						self.channel_manager.fail_htlc_backwards(&payment_hash);
+						self.payment_store
+							.set_status(&payment_hash, PaymentStatus::Failed)
+							.expect("Failed to access payment store");
+						return;
+					}
+				}
+
 				log_info!(
 					self.logger,
-					"Received payment from payment hash {} of {} msats",
+					"Received payment from payment hash {} of {}msat",
 					hex_utils::to_string(&payment_hash.0),
 					amount_msat,
 				);
@@ -326,7 +339,9 @@ where
 						hex_utils::to_string(&payment_hash.0),
 					);
 					self.channel_manager.fail_htlc_backwards(&payment_hash);
-					self.inbound_payments.lock().unwrap().remove(&payment_hash);
+					self.payment_store
+						.set_status(&payment_hash, PaymentStatus::Failed)
+						.expect("Failed to access payment store");
 				}
 			}
 			LdkEvent::PaymentClaimed {
@@ -337,7 +352,7 @@ where
 			} => {
 				log_info!(
 					self.logger,
-					"Claimed payment from payment hash {} of {} msats.",
+					"Claimed payment from payment hash {} of {}msat.",
 					hex_utils::to_string(&payment_hash.0),
 					amount_msat,
 				);
@@ -347,49 +362,50 @@ where
 					}
 					PaymentPurpose::SpontaneousPayment(preimage) => (Some(preimage), None),
 				};
-				let mut payments = self.inbound_payments.lock().unwrap();
-				match payments.entry(payment_hash) {
-					hash_map::Entry::Occupied(mut e) => {
-						let payment = e.get_mut();
-						payment.status = PaymentStatus::Succeeded;
-						payment.preimage = payment_preimage;
-						payment.secret = payment_secret;
-						payment.amount_msat = Some(amount_msat);
-					}
-					hash_map::Entry::Vacant(e) => {
-						e.insert(PaymentInfo {
+
+				let payment_info =
+					if let Some(mut payment_info) = self.payment_store.get(&payment_hash) {
+						payment_info.status = PaymentStatus::Succeeded;
+						payment_info.preimage = payment_preimage;
+						payment_info.secret = payment_secret;
+						payment_info.amount_msat = Some(amount_msat);
+						payment_info
+					} else {
+						PaymentInfo {
 							preimage: payment_preimage,
+							payment_hash,
 							secret: payment_secret,
-							status: PaymentStatus::Succeeded,
 							amount_msat: Some(amount_msat),
-						});
-					}
-				}
+							direction: PaymentDirection::Inbound,
+							status: PaymentStatus::Succeeded,
+						}
+					};
+
+				self.payment_store.insert(payment_info).expect("Failed to access payment store");
 				self.event_queue
 					.add_event(Event::PaymentReceived { payment_hash, amount_msat })
 					.expect("Failed to push to event queue");
 			}
 			LdkEvent::PaymentSent { payment_preimage, payment_hash, fee_paid_msat, .. } => {
-				let mut payments = self.outbound_payments.lock().unwrap();
-				for (hash, payment) in payments.iter_mut() {
-					if *hash == payment_hash {
-						payment.preimage = Some(payment_preimage);
-						payment.status = PaymentStatus::Succeeded;
-						log_info!(
-							self.logger,
-							"Successfully sent payment of {} msats{} from \
-								 payment hash {:?} with preimage {:?}",
-							payment.amount_msat.unwrap(),
-							if let Some(fee) = fee_paid_msat {
-								format!(" (fee {} msats)", fee)
-							} else {
-								"".to_string()
-							},
-							hex_utils::to_string(&payment_hash.0),
-							hex_utils::to_string(&payment_preimage.0)
-						);
-						break;
-					}
+				if let Some(mut payment_info) = self.payment_store.get(&payment_hash) {
+					payment_info.preimage = Some(payment_preimage);
+					payment_info.status = PaymentStatus::Succeeded;
+					self.payment_store
+						.insert(payment_info.clone())
+						.expect("Failed to access payment store");
+					log_info!(
+						self.logger,
+						"Successfully sent payment of {}msat{} from \
+						payment hash {:?} with preimage {:?}",
+						payment_info.amount_msat.unwrap(),
+						if let Some(fee) = fee_paid_msat {
+							format!(" (fee {} msat)", fee)
+						} else {
+							"".to_string()
+						},
+						hex_utils::to_string(&payment_hash.0),
+						hex_utils::to_string(&payment_preimage.0)
+					);
 				}
 				self.event_queue
 					.add_event(Event::PaymentSuccessful { payment_hash })
@@ -402,12 +418,9 @@ where
 					hex_utils::to_string(&payment_hash.0)
 				);
 
-				let mut payments = self.outbound_payments.lock().unwrap();
-				if payments.contains_key(&payment_hash) {
-					let payment = payments.get_mut(&payment_hash).unwrap();
-					assert_eq!(payment.status, PaymentStatus::Pending);
-					payment.status = PaymentStatus::Failed;
-				}
+				self.payment_store
+					.set_status(&payment_hash, PaymentStatus::Failed)
+					.expect("Failed to access payment store");
 				self.event_queue
 					.add_event(Event::PaymentFailed { payment_hash })
 					.expect("Failed to push to event queue");
@@ -493,15 +506,15 @@ where
 				if claim_from_onchain_tx {
 					log_info!(
 						self.logger,
-						"Forwarded payment{}{}, earning {} msats in fees from claiming onchain.",
+						"Forwarded payment{}{}, earning {}msat in fees from claiming onchain.",
 						from_prev_str,
 						to_next_str,
 						fee_earned,
 					);
 				} else {
 					log_info!(
 						self.logger,
-						"Forwarded payment{}{}, earning {} msats in fees.",
+						"Forwarded payment{}{}, earning {}msat in fees.",
 						from_prev_str,
 						to_next_str,
 						fee_earned,

src/io_utils.rs

@@ -1,7 +1,8 @@
+use crate::payment_store::{PaymentInfo, PAYMENT_INFO_PERSISTENCE_PREFIX};
 use crate::{Config, FilesystemLogger, NetworkGraph, Scorer, WALLET_KEYS_SEED_LEN};
 
 use lightning::routing::scoring::{ProbabilisticScorer, ProbabilisticScoringParameters};
-use lightning::util::ser::ReadableArgs;
+use lightning::util::ser::{Readable, ReadableArgs};
 
 use rand::{thread_rng, RngCore};
 
@@ -60,3 +61,25 @@ pub(crate) fn read_scorer(
 	}
 	ProbabilisticScorer::new(params, network_graph, logger)
 }
+
+pub(crate) fn read_payment_info(config: &Config) -> Vec<PaymentInfo> {
+	let ldk_data_dir = format!("{}/ldk", config.storage_dir_path);
+	let payment_store_path = format!("{}/{}", ldk_data_dir, PAYMENT_INFO_PERSISTENCE_PREFIX);
+	let mut payments = Vec::new();
+
+	if let Ok(res) = fs::read_dir(payment_store_path) {
+		for entry in res {
+			if let Ok(entry) = entry {
+				if entry.path().is_file() {
+					if let Ok(mut f) = fs::File::open(entry.path()) {
+						if let Ok(payment_info) = PaymentInfo::read(&mut f) {
+							payments.push(payment_info);
+						}
+					}
+				}
+			}
+		}
+	}
+
+	payments
+}