Calculate default node features.

src/lib.rs

@@ -218,7 +218,7 @@ async fn serialize_delta<L: Deref + Clone>(network_graph: Arc<NetworkGraph<L>>,
 	log_info!(logger, "update-fetched node count: {}", node_delta_set.len());
 	lookup::filter_delta_set(&mut delta_set, logger.clone());
 	log_info!(logger, "update-filtered channel count: {}", delta_set.len());
-	let serialization_details = serialization::serialize_delta_set(delta_set, last_sync_timestamp);
+	let serialization_details = serialization::serialize_delta_set(delta_set, &node_delta_set, last_sync_timestamp);
 
 	// process announcements
 	// write the number of channel announcements to the output
@@ -279,6 +279,15 @@ async fn serialize_delta<L: Deref + Clone>(network_graph: Arc<NetworkGraph<L>>,
 	let serialized_seen_timestamp = latest_seen_timestamp.saturating_sub(overflow_seconds);
 	serialized_seen_timestamp.write(&mut unversioned_output).unwrap();
 
+	{ // serialize the most common node features
+		let default_feature_count = serialization_details.node_announcement_feature_defaults.len() as u8;
+		default_feature_count.write(&mut unversioned_output).unwrap();
+
+		for current_feature in &serialization_details.node_announcement_feature_defaults {
+			current_feature.write(&mut unversioned_output).unwrap();
+		}
+	}
+
 	let node_id_count = node_ids.len() as u32;
 	node_id_count.write(&mut unversioned_output).unwrap();
 
@@ -290,31 +299,61 @@ async fn serialize_delta<L: Deref + Clone>(network_graph: Arc<NetworkGraph<L>>,
 		let mut current_node_delta_serialization: Vec<u8> = Vec::new();
 		current_node_id.write(&mut current_node_delta_serialization).unwrap();
 
+		let mut current_node_delta_extension: Vec<u8> = Vec::new();
+
 		if let Some(node_delta) = node_delta_set.get(&current_node_id) {
 			let mut has_update = false;
 
-			if node_delta.has_feature_set_changed {
-				// signal the presence of node features
-				current_node_delta_serialization[0] |= 1 << 2;
-				let latest_features = &node_delta.latest_details_after_seen.as_ref().unwrap().features;
-				latest_features.write(&mut current_node_delta_serialization).unwrap();
+			/*
+			Bitmap:
+			7: expect extra data after the pubkey (a u16 for the count, and then that number of bytes)
+			6-4: index of new features among default (0-6). If index is 7 (all 3 bits are set, it's
+			outside the present default range)
+			3: features have changed
+			2: addresses have changed
 
-				has_update = true;
-				node_feature_update_count += 1;
-			}
+			1: used for all keys
+			0: used for odd keys
+			*/
 
 			if node_delta.has_address_set_changed {
 				// signal the presence of node addresses
-				current_node_delta_serialization[0] |= 1 << 3;
+				current_node_delta_serialization[0] |= 1 << 2;
 				let address_set = &node_delta.latest_details_after_seen.as_ref().unwrap().addresses;
 				let addresses: Vec<SocketAddress> = address_set.clone().into_iter().collect();
-				addresses.write(&mut current_node_delta_serialization).unwrap();
+				addresses.write(&mut current_node_delta_extension).unwrap();
 
 				has_update = true;
 				node_address_update_count += 1;
 			}
 
+			if node_delta.has_feature_set_changed {
+				// signal the presence of node features
+
+				current_node_delta_serialization[0] |= 1 << 3;
+				let latest_features = &node_delta.latest_details_after_seen.as_ref().unwrap().features;
+
+				// are these features among the most common ones?
+				if let Some(index) = serialization_details.node_announcement_feature_defaults.iter().position(|f| f == latest_features) {
+					// this feature set is among the 6 defaults
+					current_node_delta_serialization[0] |= (index as u8) << 4;
+				} else {
+					current_node_delta_serialization[0] |= 0b_0111_0000; // 7 << 4
+					latest_features.write(&mut current_node_delta_extension).unwrap();
+				}
+
+				has_update = true;
+				node_feature_update_count += 1;
+			}
+
 			if has_update {
+				let extension_length = current_node_delta_extension.len() as u16;
+				if extension_length > 0 {
+					current_node_delta_serialization[0] |= 0b_1000_0000;
+					extension_length.write(&mut current_node_delta_serialization).unwrap();
+					current_node_delta_serialization.append(&mut current_node_delta_extension);
+				}
+
 				node_update_count += 1;
 			}
 		}

src/serialization.rs

@@ -4,16 +4,18 @@ use std::time::{SystemTime, UNIX_EPOCH};
 
 use bitcoin::Network;
 use bitcoin::blockdata::constants::ChainHash;
+use lightning::ln::features::NodeFeatures;
 use lightning::ln::msgs::{UnsignedChannelAnnouncement, UnsignedChannelUpdate};
 use lightning::util::ser::{BigSize, Writeable};
 use crate::config;
 
-use crate::lookup::{DeltaSet, DirectedUpdateDelta};
+use crate::lookup::{DeltaSet, DirectedUpdateDelta, NodeDeltaSet};
 
 pub(super) struct SerializationSet {
 	pub(super) announcements: Vec<UnsignedChannelAnnouncement>,
 	pub(super) updates: Vec<UpdateSerialization>,
 	pub(super) full_update_defaults: DefaultUpdateValues,
+	pub(super) node_announcement_feature_defaults: Vec<NodeFeatures>,
 	pub(super) latest_seen: u32,
 	pub(super) chain_hash: ChainHash,
 }
@@ -104,11 +106,12 @@ struct FullUpdateValueHistograms {
 	htlc_maximum_msat: HashMap<u64, usize>,
 }
 
-pub(super) fn serialize_delta_set(delta_set: DeltaSet, last_sync_timestamp: u32) -> SerializationSet {
+pub(super) fn serialize_delta_set(channel_delta_set: DeltaSet, node_delta_set: &NodeDeltaSet, last_sync_timestamp: u32) -> SerializationSet {
 	let mut serialization_set = SerializationSet {
 		announcements: vec![],
 		updates: vec![],
 		full_update_defaults: Default::default(),
+		node_announcement_feature_defaults: vec![],
 		chain_hash: ChainHash::using_genesis_block(Network::Bitcoin),
 		latest_seen: 0,
 	};
@@ -134,7 +137,7 @@ pub(super) fn serialize_delta_set(delta_set: DeltaSet, last_sync_timestamp: u32)
 	// if the previous seen update happened more than 6 days ago, the client may have pruned it, and an incremental update wouldn't work
 	let non_incremental_previous_update_threshold_timestamp = SystemTime::now().checked_sub(config::CHANNEL_REMINDER_AGE).unwrap().duration_since(UNIX_EPOCH).unwrap().as_secs() as u32;
 
-	for (scid, channel_delta) in delta_set.into_iter() {
+	for (scid, channel_delta) in channel_delta_set.into_iter() {
 
 		// any announcement chain hash is gonna be the same value. Just set it from the first one.
 		let channel_announcement_delta = channel_delta.announcement.as_ref().unwrap();
@@ -214,6 +217,19 @@ pub(super) fn serialize_delta_set(delta_set: DeltaSet, last_sync_timestamp: u32)
 	};
 
 	serialization_set.full_update_defaults = default_update_values;
+
+	let mut node_feature_histogram: HashMap<NodeFeatures, usize> = Default::default();
+	for (_node_id, node_delta) in node_delta_set.iter() {
+		if node_delta.has_feature_set_changed || node_delta.last_details_before_seen.is_none() {
+			let Some(latest_details) = node_delta.latest_details_after_seen.as_ref() else {
+				continue;
+			};
+			*node_feature_histogram.entry(latest_details.features.clone()).or_insert(0) += 1;
+		}
+	}
+
+	serialization_set.node_announcement_feature_defaults = find_leading_histogram_entries(node_feature_histogram, 7);
+
 	serialization_set
 }
 
@@ -327,3 +343,9 @@ pub(super) fn find_most_common_histogram_entry_with_default<T: Copy>(histogram:
 	// though for htlc maximum msat it could be a u64::max
 	default
 }
+
+pub(super) fn find_leading_histogram_entries(histogram: HashMap<NodeFeatures, usize>, count: usize) -> Vec<NodeFeatures> {
+	let mut entry_counts: Vec<_> = histogram.iter().collect();
+	entry_counts.sort_by(|a, b| b.1.cmp(&a.1));
+	entry_counts.into_iter().take(count).map(|a| a.0.clone()).collect()
+}

src/tests/mod.rs

@@ -47,10 +47,10 @@ pub(crate) fn db_test_schema() -> String {
 	})
 }
 
-fn generate_node_announcement() -> NodeAnnouncement {
+fn generate_node_announcement(private_key: Option<SecretKey>) -> NodeAnnouncement {
 	let secp_context = Secp256k1::new();
 
-	let random_private_key = SecretKey::from_slice(&[1; 32]).unwrap();
+	let random_private_key = private_key.unwrap_or(SecretKey::from_slice(&[1; 32]).unwrap());
 	let random_public_key = random_private_key.public_key(&secp_context);
 	let node_id = NodeId::from_pubkey(&random_public_key);
 
@@ -302,7 +302,7 @@ async fn test_node_announcement_persistence() {
 	let (mut persister, receiver) = GossipPersister::new(network_graph_arc.clone(), logger.clone());
 
 	{ // seed the db
-		let mut announcement = generate_node_announcement();
+		let mut announcement = generate_node_announcement(None);
 		receiver.send(GossipMessage::NodeAnnouncement(announcement.clone(), None)).await.unwrap();
 		receiver.send(GossipMessage::NodeAnnouncement(announcement.clone(), Some(12345))).await.unwrap();
 
@@ -345,16 +345,29 @@ async fn test_node_announcement_delta_detection() {
 	let timestamp = current_time() - 10;
 
 	{ // seed the db
-		let mut announcement = generate_node_announcement();
+		let mut announcement = generate_node_announcement(None);
 		receiver.send(GossipMessage::NodeAnnouncement(announcement.clone(), Some(timestamp - 10))).await.unwrap();
 		receiver.send(GossipMessage::NodeAnnouncement(announcement.clone(), Some(timestamp - 8))).await.unwrap();
 
+		{
+			let mut current_announcement = generate_node_announcement(Some(SecretKey::from_slice(&[2; 32]).unwrap()));
+			current_announcement.contents.features = NodeFeatures::from_be_bytes(vec![23, 48]);
+			receiver.send(GossipMessage::NodeAnnouncement(current_announcement, Some(timestamp))).await.unwrap();
+		}
+
+		{
+			let mut current_announcement = generate_node_announcement(Some(SecretKey::from_slice(&[3; 32]).unwrap()));
+			current_announcement.contents.features = NodeFeatures::from_be_bytes(vec![22, 49]);
+			receiver.send(GossipMessage::NodeAnnouncement(current_announcement, Some(timestamp))).await.unwrap();
+		}
+
 		{
 			// modify announcement to contain a bunch of addresses
 			announcement.contents.addresses.push(SocketAddress::Hostname {
 				hostname: "google.com".to_string().try_into().unwrap(),
 				port: 443,
 			});
+			announcement.contents.features = NodeFeatures::from_be_bytes(vec![23, 48]);
 			announcement.contents.addresses.push(SocketAddress::TcpIpV4 { addr: [127, 0, 0, 1], port: 9635 });
 			announcement.contents.addresses.push(SocketAddress::TcpIpV6 { addr: [1; 16], port: 1337 });
 			announcement.contents.addresses.push(SocketAddress::OnionV2([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]));
@@ -389,14 +402,18 @@ async fn test_node_announcement_delta_detection() {
 		}).await.unwrap();
 	}
 
-	let serialization = serialize_delta(network_graph_arc.clone(), timestamp - 5, logger.clone()).await;
+	let serialization = serialize_delta(network_graph_arc.clone(), timestamp - 5, None, logger.clone()).await;
 	clean_test_db().await;
 
 	assert_eq!(serialization.message_count, 3);
 	assert_eq!(serialization.node_announcement_count, 2);
 	assert_eq!(serialization.node_update_count, 1);
-	assert_eq!(serialization.node_feature_update_count, 0);
+	assert_eq!(serialization.node_feature_update_count, 1);
 	assert_eq!(serialization.node_address_update_count, 1);
+
+	// let encoded = super::rle::encode(&serialization.data);
+	// let decoded = super::rle::decode(&encoded).unwrap();
+	// assert_eq!(serialization.data, decoded);
 }
 
 /// If a channel has only seen updates in one direction, it should not be announced
@@ -541,7 +558,7 @@ async fn test_channel_reminders() {
 	{ // seed the db
 		{ // unupdated channel
 			let short_channel_id = 1;
-			let announcement = generate_announcement(short_channel_id);
+			let announcement = generate_channel_announcement(short_channel_id);
 			let update_1 = generate_update(short_channel_id, false, timestamp - channel_reminder_delta - 1, 0, 0, 0, 5, 0);
 			let update_2 = generate_update(short_channel_id, true, timestamp - channel_reminder_delta - 1, 0, 0, 0, 3, 0);
 
@@ -555,7 +572,7 @@ async fn test_channel_reminders() {
 		}
 		{ // unmodified but updated channel
 			let short_channel_id = 2;
-			let announcement = generate_announcement(short_channel_id);
+			let announcement = generate_channel_announcement(short_channel_id);
 			let update_1 = generate_update(short_channel_id, false, timestamp - channel_reminder_delta - 10, 0, 0, 0, 5, 0);
 			// in the false direction, we have one update that's different prior
 			let update_2 = generate_update(short_channel_id, false, timestamp - channel_reminder_delta - 5, 0, 1, 0, 5, 0);
@@ -597,7 +614,7 @@ async fn test_channel_reminders() {
 	logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Processed intermediate rows (2)", 1);
 
 	assert_eq!(serialization.message_count, 4);
-	assert_eq!(serialization.announcement_count, 0);
+	assert_eq!(serialization.channel_announcement_count, 0);
 	assert_eq!(serialization.update_count, 4);
 	assert_eq!(serialization.update_count_full, 0);
 	assert_eq!(serialization.update_count_incremental, 4);