Improve logging on persistence failure

Author: tnull

Cargo.toml

@@ -58,6 +58,7 @@ libc = "0.2"
 electrsd = { version = "0.22.0", features = ["legacy", "esplora_a33e97e1", "bitcoind_23_0"] }
 electrum-client = "0.12.0"
 proptest = "1.0.0"
+regex = "1.5.6"
 
 [profile.release]
 panic = "abort"

src/event.rs

@@ -82,30 +82,33 @@ impl_writeable_tlv_based_enum!(Event,
 	};
 );
 
-pub struct EventQueue<K: Deref>
+pub struct EventQueue<K: Deref, L: Deref>
 where
 	K::Target: KVStore,
+	L::Target: Logger,
 {
 	queue: Mutex<VecDeque<Event>>,
 	notifier: Condvar,
 	kv_store: K,
+	logger: L,
 }
 
-impl<K: Deref> EventQueue<K>
+impl<K: Deref, L: Deref> EventQueue<K, L>
 where
 	K::Target: KVStore,
+	L::Target: Logger,
 {
-	pub(crate) fn new(kv_store: K) -> Self {
+	pub(crate) fn new(kv_store: K, logger: L) -> Self {
 		let queue: Mutex<VecDeque<Event>> = Mutex::new(VecDeque::new());
 		let notifier = Condvar::new();
-		Self { queue, notifier, kv_store }
+		Self { queue, notifier, kv_store, logger }
 	}
 
 	pub(crate) fn add_event(&self, event: Event) -> Result<(), Error> {
 		{
 			let mut locked_queue = self.queue.lock().unwrap();
 			locked_queue.push_back(event);
-			self.persist_queue(&locked_queue)?;
+			self.write_queue_and_commit(&locked_queue)?;
 		}
 
 		self.notifier.notify_one();
@@ -122,37 +125,64 @@ where
 		{
 			let mut locked_queue = self.queue.lock().unwrap();
 			locked_queue.pop_front();
-			self.persist_queue(&locked_queue)?;
+			self.write_queue_and_commit(&locked_queue)?;
 		}
 		self.notifier.notify_one();
 		Ok(())
 	}
 
-	fn persist_queue(&self, locked_queue: &VecDeque<Event>) -> Result<(), Error> {
+	fn write_queue_and_commit(&self, locked_queue: &VecDeque<Event>) -> Result<(), Error> {
 		let mut writer = self
 			.kv_store
 			.write(EVENT_QUEUE_PERSISTENCE_NAMESPACE, EVENT_QUEUE_PERSISTENCE_KEY)
-			.map_err(|_| Error::PersistenceFailed)?;
-		EventQueueSerWrapper(locked_queue)
-			.write(&mut writer)
-			.map_err(|_| Error::PersistenceFailed)?;
-		writer.commit().map_err(|_| Error::PersistenceFailed)?;
+			.map_err(|e| {
+				log_error!(
+					self.logger,
+					"Getting writer for key {}/{} failed due to: {}",
+					EVENT_QUEUE_PERSISTENCE_NAMESPACE,
+					EVENT_QUEUE_PERSISTENCE_KEY,
+					e
+				);
+				Error::PersistenceFailed
+			})?;
+		EventQueueSerWrapper(locked_queue).write(&mut writer).map_err(|e| {
+			log_error!(
+				self.logger,
+				"Writing event queue data to key {}/{} failed due to: {}",
+				EVENT_QUEUE_PERSISTENCE_NAMESPACE,
+				EVENT_QUEUE_PERSISTENCE_KEY,
+				e
+			);
+			Error::PersistenceFailed
+		})?;
+		writer.commit().map_err(|e| {
+			log_error!(
+				self.logger,
+				"Committing event queue data to key {}/{} failed due to: {}",
+				EVENT_QUEUE_PERSISTENCE_NAMESPACE,
+				EVENT_QUEUE_PERSISTENCE_KEY,
+				e
+			);
+			Error::PersistenceFailed
+		})?;
 		Ok(())
 	}
 }
 
-impl<K: Deref> ReadableArgs<K> for EventQueue<K>
+impl<K: Deref, L: Deref> ReadableArgs<(K, L)> for EventQueue<K, L>
 where
 	K::Target: KVStore,
+	L::Target: Logger,
 {
 	#[inline]
 	fn read<R: lightning::io::Read>(
-		reader: &mut R, kv_store: K,
+		reader: &mut R, args: (K, L),
 	) -> Result<Self, lightning::ln::msgs::DecodeError> {
+		let (kv_store, logger) = args;
 		let read_queue: EventQueueDeserWrapper = Readable::read(reader)?;
 		let queue: Mutex<VecDeque<Event>> = Mutex::new(read_queue.0);
 		let notifier = Condvar::new();
-		Ok(Self { queue, notifier, kv_store })
+		Ok(Self { queue, notifier, kv_store, logger })
 	}
 }
 
@@ -189,11 +219,11 @@ where
 	L::Target: Logger,
 {
 	wallet: Arc<Wallet<bdk::database::SqliteDatabase>>,
-	event_queue: Arc<EventQueue<K>>,
+	event_queue: Arc<EventQueue<K, L>>,
 	channel_manager: Arc<ChannelManager>,
 	network_graph: Arc<NetworkGraph>,
 	keys_manager: Arc<KeysManager>,
-	payment_store: Arc<PaymentInfoStorage<K>>,
+	payment_store: Arc<PaymentInfoStorage<K, L>>,
 	tokio_runtime: Arc<tokio::runtime::Runtime>,
 	logger: L,
 	_config: Arc<Config>,
@@ -205,9 +235,9 @@ where
 	L::Target: Logger,
 {
 	pub fn new(
-		wallet: Arc<Wallet<bdk::database::SqliteDatabase>>, event_queue: Arc<EventQueue<K>>,
+		wallet: Arc<Wallet<bdk::database::SqliteDatabase>>, event_queue: Arc<EventQueue<K, L>>,
 		channel_manager: Arc<ChannelManager>, network_graph: Arc<NetworkGraph>,
-		keys_manager: Arc<KeysManager>, payment_store: Arc<PaymentInfoStorage<K>>,
+		keys_manager: Arc<KeysManager>, payment_store: Arc<PaymentInfoStorage<K, L>>,
 		tokio_runtime: Arc<tokio::runtime::Runtime>, logger: L, _config: Arc<Config>,
 	) -> Self {
 		Self {
@@ -555,12 +585,13 @@ where
 #[cfg(test)]
 mod tests {
 	use super::*;
-	use crate::test::utils::TestStore;
+	use crate::test::utils::{TestLogger, TestStore};
 
 	#[test]
 	fn event_queue_persistence() {
 		let store = Arc::new(TestStore::new());
-		let event_queue = EventQueue::new(Arc::clone(&store));
+		let logger = Arc::new(TestLogger::new());
+		let event_queue = EventQueue::new(Arc::clone(&store), Arc::clone(&logger));
 
 		let expected_event = Event::ChannelReady { channel_id: [23u8; 32], user_channel_id: 2323 };
 		event_queue.add_event(expected_event.clone()).unwrap();
@@ -577,7 +608,7 @@ mod tests {
 			.get_persisted_bytes(EVENT_QUEUE_PERSISTENCE_NAMESPACE, EVENT_QUEUE_PERSISTENCE_KEY)
 			.unwrap();
 		let deser_event_queue =
-			EventQueue::read(&mut &persisted_bytes[..], Arc::clone(&store)).unwrap();
+			EventQueue::read(&mut &persisted_bytes[..], (Arc::clone(&store), logger)).unwrap();
 		assert_eq!(deser_event_queue.next_event(), expected_event);
 		assert!(!store.get_and_clear_did_persist());
 

src/io/utils.rs

@@ -124,25 +124,31 @@ where
 }
 
 /// Read previously persisted events from the store.
-pub(crate) fn read_event_queue<K: Deref>(kv_store: K) -> Result<EventQueue<K>, std::io::Error>
+pub(crate) fn read_event_queue<K: Deref, L: Deref>(
+	kv_store: K, logger: L,
+) -> Result<EventQueue<K, L>, std::io::Error>
 where
 	K::Target: KVStore,
+	L::Target: Logger,
 {
 	let mut reader =
 		kv_store.read(EVENT_QUEUE_PERSISTENCE_NAMESPACE, EVENT_QUEUE_PERSISTENCE_KEY)?;
-	let event_queue = EventQueue::read(&mut reader, kv_store).map_err(|_| {
+	let event_queue = EventQueue::read(&mut reader, (kv_store, logger)).map_err(|_| {
 		std::io::Error::new(std::io::ErrorKind::InvalidData, "Failed to deserialize EventQueue")
 	})?;
 	Ok(event_queue)
 }
 
 /// Read previously persisted peer info from the store.
-pub(crate) fn read_peer_info<K: Deref>(kv_store: K) -> Result<PeerInfoStorage<K>, std::io::Error>
+pub(crate) fn read_peer_info<K: Deref, L: Deref>(
+	kv_store: K, logger: L,
+) -> Result<PeerInfoStorage<K, L>, std::io::Error>
 where
 	K::Target: KVStore,
+	L::Target: Logger,
 {
 	let mut reader = kv_store.read(PEER_INFO_PERSISTENCE_NAMESPACE, PEER_INFO_PERSISTENCE_KEY)?;
-	let peer_info = PeerInfoStorage::read(&mut reader, kv_store).map_err(|_| {
+	let peer_info = PeerInfoStorage::read(&mut reader, (kv_store, logger)).map_err(|_| {
 		std::io::Error::new(
 			std::io::ErrorKind::InvalidData,
 			"Failed to deserialize PeerInfoStorage",

src/lib.rs

@@ -503,32 +503,36 @@ impl Builder {
 
 		// Init payment info storage
 		let payment_store = match io::utils::read_payment_info(Arc::clone(&kv_store)) {
-			Ok(payments) => {
-				Arc::new(PaymentInfoStorage::from_payments(payments, Arc::clone(&kv_store)))
-			}
+			Ok(payments) => Arc::new(PaymentInfoStorage::from_payments(
+				payments,
+				Arc::clone(&kv_store),
+				Arc::clone(&logger),
+			)),
 			Err(e) => {
 				log_error!(logger, "Failed to read payment information: {}", e.to_string());
 				panic!("Failed to read payment information: {}", e.to_string());
 			}
 		};
 
-		let event_queue = match io::utils::read_event_queue(Arc::clone(&kv_store)) {
-			Ok(event_queue) => Arc::new(event_queue),
-			Err(e) => {
-				if e.kind() == std::io::ErrorKind::NotFound {
-					Arc::new(EventQueue::new(Arc::clone(&kv_store)))
-				} else {
-					log_error!(logger, "Failed to read event queue: {}", e.to_string());
-					panic!("Failed to read event queue: {}", e.to_string());
+		let event_queue =
+			match io::utils::read_event_queue(Arc::clone(&kv_store), Arc::clone(&logger)) {
+				Ok(event_queue) => Arc::new(event_queue),
+				Err(e) => {
+					if e.kind() == std::io::ErrorKind::NotFound {
+						Arc::new(EventQueue::new(Arc::clone(&kv_store), Arc::clone(&logger)))
+					} else {
+						log_error!(logger, "Failed to read event queue: {}", e.to_string());
+						panic!("Failed to read event queue: {}", e.to_string());
+					}
 				}
-			}
-		};
+			};
 
-		let peer_store = match io::utils::read_peer_info(Arc::clone(&kv_store)) {
+		let peer_store = match io::utils::read_peer_info(Arc::clone(&kv_store), Arc::clone(&logger))
+		{
 			Ok(peer_store) => Arc::new(peer_store),
 			Err(e) => {
 				if e.kind() == std::io::ErrorKind::NotFound {
-					Arc::new(PeerInfoStorage::new(Arc::clone(&kv_store)))
+					Arc::new(PeerInfoStorage::new(Arc::clone(&kv_store), Arc::clone(&logger)))
 				} else {
 					log_error!(logger, "Failed to read peer store: {}", e.to_string());
 					panic!("Failed to read peer store: {}", e.to_string());
@@ -576,7 +580,7 @@ pub struct Node {
 	config: Arc<Config>,
 	wallet: Arc<Wallet<bdk::database::SqliteDatabase>>,
 	tx_sync: Arc<EsploraSyncClient<Arc<FilesystemLogger>>>,
-	event_queue: Arc<EventQueue<Arc<FilesystemStore>>>,
+	event_queue: Arc<EventQueue<Arc<FilesystemStore>, Arc<FilesystemLogger>>>,
 	channel_manager: Arc<ChannelManager>,
 	chain_monitor: Arc<ChainMonitor>,
 	peer_manager: Arc<PeerManager>,
@@ -586,8 +590,8 @@ pub struct Node {
 	kv_store: Arc<FilesystemStore>,
 	logger: Arc<FilesystemLogger>,
 	scorer: Arc<Mutex<Scorer>>,
-	peer_store: Arc<PeerInfoStorage<Arc<FilesystemStore>>>,
-	payment_store: Arc<PaymentInfoStorage<Arc<FilesystemStore>>>,
+	peer_store: Arc<PeerInfoStorage<Arc<FilesystemStore>, Arc<FilesystemLogger>>>,
+	payment_store: Arc<PaymentInfoStorage<Arc<FilesystemStore>, Arc<FilesystemLogger>>>,
 }
 
 impl Node {