Skip to content

Peer Storage (Part 3): Identifying Lost Channel States #3897

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.

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

Open
wants to merge 8 commits into
base: main
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
1 change: 1 addition & 0 deletions Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -68,4 +68,5 @@ check-cfg = [
"cfg(splicing)",
"cfg(async_payments)",
"cfg(simple_close)",
"cfg(peer_storage)",
]
2 changes: 2 additions & 0 deletions ci/ci-tests.sh
Original file line number Diff line number Diff line change
Expand Up @@ -158,3 +158,5 @@ RUSTFLAGS="--cfg=async_payments" cargo test --verbose --color always -p lightnin
RUSTFLAGS="--cfg=simple_close" cargo test --verbose --color always -p lightning
[ "$CI_MINIMIZE_DISK_USAGE" != "" ] && cargo clean
RUSTFLAGS="--cfg=lsps1_service" cargo test --verbose --color always -p lightning-liquidity
[ "$CI_MINIMIZE_DISK_USAGE" != "" ] && cargo clean
RUSTFLAGS="--cfg=peer_storage" cargo test --verbose --color always -p lightning
60 changes: 56 additions & 4 deletions lightning/src/chain/chainmonitor.rs
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,8 @@ use bitcoin::hash_types::{BlockHash, Txid};

use crate::chain;
use crate::chain::chaininterface::{BroadcasterInterface, FeeEstimator};
#[cfg(peer_storage)]
use crate::chain::channelmonitor::write_chanmon_internal;
use crate::chain::channelmonitor::{
Balance, ChannelMonitor, ChannelMonitorUpdate, MonitorEvent, TransactionOutputs,
WithChannelMonitor,
Expand All @@ -37,7 +39,7 @@ use crate::chain::{ChannelMonitorUpdateStatus, Filter, WatchedOutput};
use crate::events::{self, Event, EventHandler, ReplayEvent};
use crate::ln::channel_state::ChannelDetails;
use crate::ln::msgs::{self, BaseMessageHandler, Init, MessageSendEvent, SendOnlyMessageHandler};
use crate::ln::our_peer_storage::DecryptedOurPeerStorage;
use crate::ln::our_peer_storage::{DecryptedOurPeerStorage, PeerStorageMonitorHolder};
use crate::ln::types::ChannelId;
use crate::prelude::*;
use crate::sign::ecdsa::EcdsaChannelSigner;
Expand All @@ -47,6 +49,8 @@ use crate::types::features::{InitFeatures, NodeFeatures};
use crate::util::errors::APIError;
use crate::util::logger::{Logger, WithContext};
use crate::util::persist::MonitorName;
#[cfg(peer_storage)]
use crate::util::ser::{VecWriter, Writeable};
use crate::util::wakers::{Future, Notifier};
use bitcoin::secp256k1::PublicKey;
use core::ops::Deref;
Expand Down Expand Up @@ -809,11 +813,57 @@ where
mon.values().map(|monitor| monitor.monitor.get_counterparty_node_id()).collect()
}

#[cfg(peer_storage)]
Copy link
Contributor

Choose a reason for hiding this comment

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

I think you'll also need to introduce the cfg-gate in a few other places to get rid of the warning, e.g., above for all_counterparty_node_ids and on some imports.

fn send_peer_storage(&self, their_node_id: PublicKey) {
// TODO: Serialize `ChannelMonitor`s inside `our_peer_storage`.

#[allow(unused_mut)]
let mut monitors_list: Vec<PeerStorageMonitorHolder> = Vec::new();
let random_bytes = self.entropy_source.get_secure_random_bytes();
let serialised_channels = Vec::new();

const MAX_PEER_STORAGE_SIZE: usize = 65531;
const USIZE_LEN: usize = core::mem::size_of::<usize>();
let mut usize_bytes = [0u8; USIZE_LEN];
usize_bytes.copy_from_slice(&random_bytes[0..USIZE_LEN]);
let random_usize = usize::from_le_bytes(usize_bytes);

let mut curr_size = 0;
let monitors = self.monitors.read().unwrap();
let mut stored_chanmon_idx = alloc::collections::BTreeSet::<usize>::new();
// Used as a fallback reference if the set is empty
let zero = 0;

while curr_size < MAX_PEER_STORAGE_SIZE
&& *stored_chanmon_idx.last().unwrap_or(&zero) < monitors.len()
{
let idx = random_usize % monitors.len();
stored_chanmon_idx.insert(idx + 1);
Comment on lines +834 to +838
Copy link
Contributor

Choose a reason for hiding this comment

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

The loop condition has a potential issue where it might not terminate if the same random index is repeatedly selected. Since stored_chanmon_idx is a BTreeSet, inserting the same value multiple times doesn't increase its size. If random_usize % monitors.len() keeps generating the same index, the condition *stored_chanmon_idx.last().unwrap_or(&zero) < monitors.len() will remain true indefinitely.

Consider modifying the approach to ensure termination, such as:

  1. Using a counter to limit iterations
  2. Tracking already-processed indices in a separate set
  3. Using a deterministic sequence instead of random selection

This would prevent potential infinite loops while still achieving the goal of selecting monitors for peer storage.

Suggested change
while curr_size < MAX_PEER_STORAGE_SIZE
&& *stored_chanmon_idx.last().unwrap_or(&zero) < monitors.len()
{
let idx = random_usize % monitors.len();
stored_chanmon_idx.insert(idx + 1);
while curr_size < MAX_PEER_STORAGE_SIZE
&& *stored_chanmon_idx.last().unwrap_or(&zero) < monitors.len()
&& stored_chanmon_idx.len() < monitors.len()
{
let idx = random_usize % monitors.len();
let inserted = stored_chanmon_idx.insert(idx + 1);
if inserted {
curr_size += 1;
} else {
// If we couldn't insert, try a different index next time
random_usize = random_usize.wrapping_add(1);
}

Spotted by Diamond

Is this helpful? React 👍 or 👎 to let us know.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

@tnull, this is the same concern that I had in my mind as well. Wasn't the previous approach more efficient and secure?

Copy link
Contributor

Choose a reason for hiding this comment

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

How about something along these lines?:

diff --git a/lightning/src/chain/chainmonitor.rs b/lightning/src/chain/chainmonitor.rs
index 4a40ba872..dde117ca9 100644
--- a/lightning/src/chain/chainmonitor.rs
+++ b/lightning/src/chain/chainmonitor.rs
@@ -39,6 +39,7 @@ use crate::chain::{ChannelMonitorUpdateStatus, Filter, WatchedOutput};
 use crate::events::{self, Event, EventHandler, ReplayEvent};
 use crate::ln::channel_state::ChannelDetails;
 use crate::ln::msgs::{self, BaseMessageHandler, Init, MessageSendEvent, SendOnlyMessageHandler};
+#[cfg(peer_storage)]
 use crate::ln::our_peer_storage::{DecryptedOurPeerStorage, PeerStorageMonitorHolder};
 use crate::ln::types::ChannelId;
 use crate::prelude::*;
@@ -53,6 +54,8 @@ use crate::util::persist::MonitorName;
 use crate::util::ser::{VecWriter, Writeable};
 use crate::util::wakers::{Future, Notifier};
 use bitcoin::secp256k1::PublicKey;
+#[cfg(peer_storage)]
+use core::iter::Cycle;
 use core::ops::Deref;
 use core::sync::atomic::{AtomicUsize, Ordering};
 
@@ -808,6 +811,7 @@ where
 
 	/// This function collects the counterparty node IDs from all monitors into a `HashSet`,
 	/// ensuring unique IDs are returned.
+	#[cfg(peer_storage)]
 	fn all_counterparty_node_ids(&self) -> HashSet<PublicKey> {
 		let mon = self.monitors.read().unwrap();
 		mon.values().map(|monitor| monitor.monitor.get_counterparty_node_id()).collect()
@@ -815,51 +819,71 @@ where
 
 	#[cfg(peer_storage)]
 	fn send_peer_storage(&self, their_node_id: PublicKey) {
-		#[allow(unused_mut)]
 		let mut monitors_list: Vec<PeerStorageMonitorHolder> = Vec::new();
 		let random_bytes = self.entropy_source.get_secure_random_bytes();
 
 		const MAX_PEER_STORAGE_SIZE: usize = 65531;
 		const USIZE_LEN: usize = core::mem::size_of::<usize>();
-		let mut usize_bytes = [0u8; USIZE_LEN];
-		usize_bytes.copy_from_slice(&random_bytes[0..USIZE_LEN]);
-		let random_usize = usize::from_le_bytes(usize_bytes);
+		let mut random_bytes_cycle_iter = random_bytes.iter().cycle();
+
+		let mut current_size = 0;
+		let monitors_lock = self.monitors.read().unwrap();
+		let mut channel_ids = monitors_lock.keys().copied().collect();
+
+		fn next_random_id(
+			channel_ids: &mut Vec<ChannelId>,
+			random_bytes_cycle_iter: &mut Cycle<core::slice::Iter<u8>>,
+		) -> Option<ChannelId> {
+			if channel_ids.is_empty() {
+				return None;
+			}
 
-		let mut curr_size = 0;
-		let monitors = self.monitors.read().unwrap();
-		let mut stored_chanmon_idx = alloc::collections::BTreeSet::<usize>::new();
-		// Used as a fallback reference if the set is empty
-		let zero = 0;
+			let random_idx = {
+				let mut usize_bytes = [0u8; USIZE_LEN];
+				usize_bytes.iter_mut().for_each(|b| {
+					*b = *random_bytes_cycle_iter.next().expect("A cycle never ends")
+				});
+				// Take one more to introduce a slight misalignment.
+				random_bytes_cycle_iter.next().expect("A cycle never ends");
+				usize::from_le_bytes(usize_bytes) % channel_ids.len()
+			};
+
+			Some(channel_ids.swap_remove(random_idx))
+		}
 
-		while curr_size < MAX_PEER_STORAGE_SIZE
-			&& *stored_chanmon_idx.last().unwrap_or(&zero) < monitors.len()
+		while let Some(channel_id) = next_random_id(&mut channel_ids, &mut random_bytes_cycle_iter)
 		{
-			let idx = random_usize % monitors.len();
-			stored_chanmon_idx.insert(idx + 1);
-			let (cid, mon) = monitors.iter().skip(idx).next().unwrap();
+			let monitor_holder = if let Some(monitor_holder) = monitors_lock.get(&channel_id) {
+				monitor_holder
+			} else {
+				debug_assert!(
+					false,
+					"Tried to access non-existing monitor, this should never happen"
+				);
+				break;
+			};
 
-			let mut ser_chan = VecWriter(Vec::new());
-			let min_seen_secret = mon.monitor.get_min_seen_secret();
-			let counterparty_node_id = mon.monitor.get_counterparty_node_id();
+			let mut serialized_channel = VecWriter(Vec::new());
+			let min_seen_secret = monitor_holder.monitor.get_min_seen_secret();
+			let counterparty_node_id = monitor_holder.monitor.get_counterparty_node_id();
 			{
-				let chan_mon = mon.monitor.inner.lock().unwrap();
+				let inner_lock = monitor_holder.monitor.inner.lock().unwrap();
 
-				write_chanmon_internal(&chan_mon, true, &mut ser_chan)
+				write_chanmon_internal(&inner_lock, true, &mut serialized_channel)
 					.expect("can not write Channel Monitor for peer storage message");
 			}
 			let peer_storage_monitor = PeerStorageMonitorHolder {
-				channel_id: *cid,
+				channel_id,
 				min_seen_secret,
 				counterparty_node_id,
-				monitor_bytes: ser_chan.0,
+				monitor_bytes: serialized_channel.0,
 			};
 
-			// Adding size of peer_storage_monitor.
-			curr_size += peer_storage_monitor.serialized_length();
-
-			if curr_size > MAX_PEER_STORAGE_SIZE {
-				break;
+			current_size += peer_storage_monitor.serialized_length();
+			if current_size > MAX_PEER_STORAGE_SIZE {
+				continue;
 			}
+
 			monitors_list.push(peer_storage_monitor);
 		}

There might be still some room for improvement, but IMO something like this would be a lot more readable.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

+			if current_size > MAX_PEER_STORAGE_SIZE {
+				continue;
 			}

Why not break here?

let (cid, mon) = monitors.iter().skip(idx).next().unwrap();

let mut ser_chan = VecWriter(Vec::new());
let min_seen_secret = mon.monitor.get_min_seen_secret();
let counterparty_node_id = mon.monitor.get_counterparty_node_id();
{
let chan_mon = mon.monitor.inner.lock().unwrap();
Copy link
Contributor

Choose a reason for hiding this comment

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

Can we put this lock and the write call into a separate scope so we drop the lock ASAP again?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Yes, thanks for pointing this out. Fixed.


write_chanmon_internal(&chan_mon, true, &mut ser_chan)
.expect("can not write Channel Monitor for peer storage message");
}
let peer_storage_monitor = PeerStorageMonitorHolder {
channel_id: *cid,
min_seen_secret,
counterparty_node_id,
monitor_bytes: ser_chan.0,
};

// Adding size of peer_storage_monitor.
curr_size += peer_storage_monitor.serialized_length();

if curr_size > MAX_PEER_STORAGE_SIZE {
break;
}
monitors_list.push(peer_storage_monitor);
}

let serialised_channels = monitors_list.encode();
let our_peer_storage = DecryptedOurPeerStorage::new(serialised_channels);
let cipher = our_peer_storage.encrypt(&self.our_peerstorage_encryption_key, &random_bytes);

Expand Down Expand Up @@ -920,6 +970,7 @@ where
)
});

#[cfg(peer_storage)]
// Send peer storage everytime a new block arrives.
for node_id in self.all_counterparty_node_ids() {
self.send_peer_storage(node_id);
Expand Down Expand Up @@ -1021,6 +1072,7 @@ where
)
});

#[cfg(peer_storage)]
// Send peer storage everytime a new block arrives.
for node_id in self.all_counterparty_node_ids() {
self.send_peer_storage(node_id);
Expand Down
Loading
Loading