Merge pull request #2954 from TheBlueMatt/2024-03-test-ci-beta-fail
[rust-lightning] / lightning-rapid-gossip-sync / src / processing.rs
index 4b6de04c6556a5ef302d4dced78f1b833f5e5380..b3fae0ffd8bea8e19822c5a2cf9bc19d3fe33939 100644 (file)
@@ -2,7 +2,7 @@ use core::cmp::max;
 use core::ops::Deref;
 use core::sync::atomic::Ordering;
 
-use bitcoin::BlockHash;
+use bitcoin::blockdata::constants::ChainHash;
 use bitcoin::secp256k1::PublicKey;
 
 use lightning::ln::msgs::{
@@ -10,16 +10,16 @@ use lightning::ln::msgs::{
 };
 use lightning::routing::gossip::NetworkGraph;
 use lightning::util::logger::Logger;
+use lightning::{log_debug, log_warn, log_trace, log_given_level, log_gossip};
 use lightning::util::ser::{BigSize, Readable};
 use lightning::io;
 
-use crate::error::GraphSyncError;
-use crate::RapidGossipSync;
+use crate::{GraphSyncError, RapidGossipSync};
 
 #[cfg(all(feature = "std", not(test)))]
 use std::time::{SystemTime, UNIX_EPOCH};
 
-#[cfg(not(feature = "std"))]
+#[cfg(all(not(feature = "std"), not(test)))]
 use alloc::{vec::Vec, borrow::ToOwned};
 
 /// The purpose of this prefix is to identify the serialization format, should other rapid gossip
@@ -37,13 +37,14 @@ const MAX_INITIAL_NODE_ID_VECTOR_CAPACITY: u32 = 50_000;
 const STALE_RGS_UPDATE_AGE_LIMIT_SECS: u64 = 60 * 60 * 24 * 14;
 
 impl<NG: Deref<Target=NetworkGraph<L>>, L: Deref> RapidGossipSync<NG, L> where L::Target: Logger {
+       #[cfg(feature = "std")]
        pub(crate) fn update_network_graph_from_byte_stream<R: io::Read>(
                &self,
                read_cursor: &mut R,
        ) -> Result<u32, GraphSyncError> {
-               #[allow(unused_mut)]
+               #[allow(unused_mut, unused_assignments)]
                let mut current_time_unix = None;
-               #[cfg(all(feature = "std", not(test)))]
+               #[cfg(not(test))]
                {
                        // Note that many tests rely on being able to set arbitrarily old timestamps, thus we
                        // disable this check during tests!
@@ -57,6 +58,7 @@ impl<NG: Deref<Target=NetworkGraph<L>>, L: Deref> RapidGossipSync<NG, L> where L
                mut read_cursor: &mut R,
                current_time_unix: Option<u64>
        ) -> Result<u32, GraphSyncError> {
+               log_trace!(self.logger, "Processing RGS data...");
                let mut prefix = [0u8; 4];
                read_cursor.read_exact(&mut prefix)?;
 
@@ -64,7 +66,17 @@ impl<NG: Deref<Target=NetworkGraph<L>>, L: Deref> RapidGossipSync<NG, L> where L
                        return Err(DecodeError::UnknownVersion.into());
                }
 
-               let chain_hash: BlockHash = Readable::read(read_cursor)?;
+               let chain_hash: ChainHash = Readable::read(read_cursor)?;
+               let ng_chain_hash = self.network_graph.get_chain_hash();
+               if chain_hash != ng_chain_hash {
+                       return Err(
+                               LightningError {
+                                       err: "Rapid Gossip Sync data's chain hash does not match the network graph's".to_owned(),
+                                       action: ErrorAction::IgnoreError,
+                               }.into()
+                       );
+               }
+
                let latest_seen_timestamp: u32 = Readable::read(read_cursor)?;
 
                if let Some(time) = current_time_unix {
@@ -109,6 +121,9 @@ impl<NG: Deref<Target=NetworkGraph<L>>, L: Deref> RapidGossipSync<NG, L> where L
                        let node_id_1 = node_ids[node_id_1_index.0 as usize];
                        let node_id_2 = node_ids[node_id_2_index.0 as usize];
 
+                       log_gossip!(self.logger, "Adding channel {} from RGS announcement at {}",
+                               short_channel_id, latest_seen_timestamp);
+
                        let announcement_result = network_graph.add_channel_from_partial_announcement(
                                short_channel_id,
                                backdated_timestamp as u64,
@@ -120,6 +135,7 @@ impl<NG: Deref<Target=NetworkGraph<L>>, L: Deref> RapidGossipSync<NG, L> where L
                                if let ErrorAction::IgnoreDuplicateGossip = lightning_error.action {
                                        // everything is fine, just a duplicate channel announcement
                                } else {
+                                       log_warn!(self.logger, "Failed to process channel announcement: {:?}", lightning_error);
                                        return Err(lightning_error.into());
                                }
                        }
@@ -128,6 +144,8 @@ impl<NG: Deref<Target=NetworkGraph<L>>, L: Deref> RapidGossipSync<NG, L> where L
                previous_scid = 0; // updates start at a new scid
 
                let update_count: u32 = Readable::read(read_cursor)?;
+               log_debug!(self.logger, "Processing RGS update from {} with {} nodes, {} channel announcements and {} channel updates.",
+                       latest_seen_timestamp, node_id_count, announcement_count, update_count);
                if update_count == 0 {
                        return Ok(latest_seen_timestamp);
                }
@@ -169,24 +187,19 @@ impl<NG: Deref<Target=NetworkGraph<L>>, L: Deref> RapidGossipSync<NG, L> where L
                        if (channel_flags & 0b_1000_0000) != 0 {
                                // incremental update, field flags will indicate mutated values
                                let read_only_network_graph = network_graph.read_only();
-                               if let Some(channel) = read_only_network_graph
-                                       .channels()
-                                       .get(&short_channel_id) {
-
-                                       let directional_info = channel
-                                               .get_directional_info(channel_flags)
-                                               .ok_or(LightningError {
-                                                       err: "Couldn't find previous directional data for update".to_owned(),
-                                                       action: ErrorAction::IgnoreError,
-                                               })?;
-
+                               if let Some(directional_info) =
+                                       read_only_network_graph.channels().get(&short_channel_id)
+                                       .and_then(|channel| channel.get_directional_info(channel_flags))
+                               {
                                        synthetic_update.cltv_expiry_delta = directional_info.cltv_expiry_delta;
                                        synthetic_update.htlc_minimum_msat = directional_info.htlc_minimum_msat;
                                        synthetic_update.htlc_maximum_msat = directional_info.htlc_maximum_msat;
                                        synthetic_update.fee_base_msat = directional_info.fees.base_msat;
                                        synthetic_update.fee_proportional_millionths = directional_info.fees.proportional_millionths;
-
                                } else {
+                                       log_trace!(self.logger,
+                                               "Skipping application of channel update for chan {} with flags {} as original data is missing.",
+                                               short_channel_id, channel_flags);
                                        skip_update_for_unknown_channel = true;
                                }
                        };
@@ -220,17 +233,27 @@ impl<NG: Deref<Target=NetworkGraph<L>>, L: Deref> RapidGossipSync<NG, L> where L
                                continue;
                        }
 
+                       log_gossip!(self.logger, "Updating channel {} with flags {} from RGS announcement at {}",
+                               short_channel_id, channel_flags, latest_seen_timestamp);
                        match network_graph.update_channel_unsigned(&synthetic_update) {
                                Ok(_) => {},
                                Err(LightningError { action: ErrorAction::IgnoreDuplicateGossip, .. }) => {},
-                               Err(LightningError { action: ErrorAction::IgnoreAndLog(_), .. }) => {},
+                               Err(LightningError { action: ErrorAction::IgnoreAndLog(level), err }) => {
+                                       log_given_level!(self.logger, level, "Failed to apply channel update: {:?}", err);
+                               },
                                Err(LightningError { action: ErrorAction::IgnoreError, .. }) => {},
                                Err(e) => return Err(e.into()),
                        }
                }
 
                self.network_graph.set_last_rapid_gossip_sync_timestamp(latest_seen_timestamp);
+
+               if let Some(time) = current_time_unix {
+                       self.network_graph.remove_stale_channels_and_tracking_with_time(time)
+               }
+
                self.is_initial_sync_complete.store(true, Ordering::Release);
+               log_trace!(self.logger, "Done processing RGS data from {}", latest_seen_timestamp);
                Ok(latest_seen_timestamp)
        }
 }
@@ -239,13 +262,14 @@ impl<NG: Deref<Target=NetworkGraph<L>>, L: Deref> RapidGossipSync<NG, L> where L
 mod tests {
        use bitcoin::Network;
 
+       #[cfg(feature = "std")]
        use lightning::ln::msgs::DecodeError;
+
        use lightning::routing::gossip::NetworkGraph;
        use lightning::util::test_utils::TestLogger;
 
-       use crate::error::GraphSyncError;
        use crate::processing::STALE_RGS_UPDATE_AGE_LIMIT_SECS;
-       use crate::RapidGossipSync;
+       use crate::{GraphSyncError, RapidGossipSync};
 
        const VALID_RGS_BINARY: [u8; 300] = [
                76, 68, 75, 1, 111, 226, 140, 10, 182, 241, 179, 114, 193, 166, 162, 70, 174, 99, 247,
@@ -267,6 +291,7 @@ mod tests {
        const VALID_BINARY_TIMESTAMP: u64 = 1642291930;
 
        #[test]
+       #[cfg(feature = "std")]
        fn network_graph_fails_to_update_from_clipped_input() {
                let logger = TestLogger::new();
                let network_graph = NetworkGraph::new(Network::Bitcoin, &logger);
@@ -287,7 +312,7 @@ mod tests {
                        0, 0, 0, 0, 0, 1, 0, 0, 0, 0, 58, 85, 116, 216, 255, 2, 68, 226, 0, 6, 11, 0, 1, 24, 0,
                        0, 3, 232, 0, 0, 0,
                ];
-               let rapid_sync = RapidGossipSync::new(&network_graph);
+               let rapid_sync = RapidGossipSync::new(&network_graph, &logger);
                let update_result = rapid_sync.update_network_graph(&example_input[..]);
                assert!(update_result.is_err());
                if let Err(GraphSyncError::DecodeError(DecodeError::ShortRead)) = update_result {
@@ -298,6 +323,7 @@ mod tests {
        }
 
        #[test]
+       #[cfg(feature = "std")]
        fn incremental_only_update_ignores_missing_channel() {
                let incremental_update_input = vec![
                        76, 68, 75, 1, 111, 226, 140, 10, 182, 241, 179, 114, 193, 166, 162, 70, 174, 99, 247,
@@ -312,12 +338,13 @@ mod tests {
 
                assert_eq!(network_graph.read_only().channels().len(), 0);
 
-               let rapid_sync = RapidGossipSync::new(&network_graph);
+               let rapid_sync = RapidGossipSync::new(&network_graph, &logger);
                let update_result = rapid_sync.update_network_graph(&incremental_update_input[..]);
                assert!(update_result.is_ok());
        }
 
        #[test]
+       #[cfg(feature = "std")]
        fn incremental_only_update_fails_without_prior_updates() {
                let announced_update_input = vec![
                        76, 68, 75, 1, 111, 226, 140, 10, 182, 241, 179, 114, 193, 166, 162, 70, 174, 99, 247,
@@ -340,20 +367,12 @@ mod tests {
 
                assert_eq!(network_graph.read_only().channels().len(), 0);
 
-               let rapid_sync = RapidGossipSync::new(&network_graph);
-               let update_result = rapid_sync.update_network_graph(&announced_update_input[..]);
-               assert!(update_result.is_err());
-               if let Err(GraphSyncError::LightningError(lightning_error)) = update_result {
-                       assert_eq!(
-                               lightning_error.err,
-                               "Couldn't find previous directional data for update"
-                       );
-               } else {
-                       panic!("Unexpected update result: {:?}", update_result)
-               }
+               let rapid_sync = RapidGossipSync::new(&network_graph, &logger);
+               rapid_sync.update_network_graph(&announced_update_input[..]).unwrap();
        }
 
        #[test]
+       #[cfg(feature = "std")]
        fn incremental_only_update_fails_without_prior_same_direction_updates() {
                let initialization_input = vec![
                        76, 68, 75, 1, 111, 226, 140, 10, 182, 241, 179, 114, 193, 166, 162, 70, 174, 99, 247,
@@ -376,7 +395,7 @@ mod tests {
 
                assert_eq!(network_graph.read_only().channels().len(), 0);
 
-               let rapid_sync = RapidGossipSync::new(&network_graph);
+               let rapid_sync = RapidGossipSync::new(&network_graph, &logger);
                let initialization_result = rapid_sync.update_network_graph(&initialization_input[..]);
                if initialization_result.is_err() {
                        panic!(
@@ -405,19 +424,11 @@ mod tests {
                        0, 0, 0, 0, 0, 0, 0, 0, 0, 255, 8, 153, 192, 0, 2, 27, 0, 0, 136, 0, 0, 0, 221, 255, 2,
                        68, 226, 0, 6, 11, 0, 1, 128,
                ];
-               let update_result = rapid_sync.update_network_graph(&opposite_direction_incremental_update_input[..]);
-               assert!(update_result.is_err());
-               if let Err(GraphSyncError::LightningError(lightning_error)) = update_result {
-                       assert_eq!(
-                               lightning_error.err,
-                               "Couldn't find previous directional data for update"
-                       );
-               } else {
-                       panic!("Unexpected update result: {:?}", update_result)
-               }
+               rapid_sync.update_network_graph(&opposite_direction_incremental_update_input[..]).unwrap();
        }
 
        #[test]
+       #[cfg(feature = "std")]
        fn incremental_update_succeeds_with_prior_announcements_and_full_updates() {
                let initialization_input = vec![
                        76, 68, 75, 1, 111, 226, 140, 10, 182, 241, 179, 114, 193, 166, 162, 70, 174, 99, 247,
@@ -442,7 +453,7 @@ mod tests {
 
                assert_eq!(network_graph.read_only().channels().len(), 0);
 
-               let rapid_sync = RapidGossipSync::new(&network_graph);
+               let rapid_sync = RapidGossipSync::new(&network_graph, &logger);
                let initialization_result = rapid_sync.update_network_graph(&initialization_input[..]);
                assert!(initialization_result.is_ok());
 
@@ -477,6 +488,7 @@ mod tests {
        }
 
        #[test]
+       #[cfg(feature = "std")]
        fn update_succeeds_when_duplicate_gossip_is_applied() {
                let initialization_input = vec![
                        76, 68, 75, 1, 111, 226, 140, 10, 182, 241, 179, 114, 193, 166, 162, 70, 174, 99, 247,
@@ -501,7 +513,7 @@ mod tests {
 
                assert_eq!(network_graph.read_only().channels().len(), 0);
 
-               let rapid_sync = RapidGossipSync::new(&network_graph);
+               let rapid_sync = RapidGossipSync::new(&network_graph, &logger);
                let initialization_result = rapid_sync.update_network_graph(&initialization_input[..]);
                assert!(initialization_result.is_ok());
 
@@ -520,13 +532,14 @@ mod tests {
        }
 
        #[test]
+       #[cfg(feature = "std")]
        fn full_update_succeeds() {
                let logger = TestLogger::new();
                let network_graph = NetworkGraph::new(Network::Bitcoin, &logger);
 
                assert_eq!(network_graph.read_only().channels().len(), 0);
 
-               let rapid_sync = RapidGossipSync::new(&network_graph);
+               let rapid_sync = RapidGossipSync::new(&network_graph, &logger);
                let update_result = rapid_sync.update_network_graph(&VALID_RGS_BINARY);
                if update_result.is_err() {
                        panic!("Unexpected update result: {:?}", update_result)
@@ -557,13 +570,41 @@ mod tests {
 
                assert_eq!(network_graph.read_only().channels().len(), 0);
 
-               let rapid_sync = RapidGossipSync::new(&network_graph);
+               let rapid_sync = RapidGossipSync::new(&network_graph, &logger);
                // this is mostly for checking uint underflow issues before the fuzzer does
                let update_result = rapid_sync.update_network_graph_no_std(&VALID_RGS_BINARY, Some(0));
                assert!(update_result.is_ok());
                assert_eq!(network_graph.read_only().channels().len(), 2);
        }
 
+       #[test]
+       fn prunes_after_update() {
+               // this is the timestamp encoded in the binary data of valid_input below
+               let logger = TestLogger::new();
+
+               let latest_nonpruning_time = VALID_BINARY_TIMESTAMP + 60 * 60 * 24 * 7;
+
+               {
+                       let network_graph = NetworkGraph::new(Network::Bitcoin, &logger);
+                       assert_eq!(network_graph.read_only().channels().len(), 0);
+
+                       let rapid_sync = RapidGossipSync::new(&network_graph, &logger);
+                       let update_result = rapid_sync.update_network_graph_no_std(&VALID_RGS_BINARY, Some(latest_nonpruning_time));
+                       assert!(update_result.is_ok());
+                       assert_eq!(network_graph.read_only().channels().len(), 2);
+               }
+
+               {
+                       let network_graph = NetworkGraph::new(Network::Bitcoin, &logger);
+                       assert_eq!(network_graph.read_only().channels().len(), 0);
+
+                       let rapid_sync = RapidGossipSync::new(&network_graph, &logger);
+                       let update_result = rapid_sync.update_network_graph_no_std(&VALID_RGS_BINARY, Some(latest_nonpruning_time + 1));
+                       assert!(update_result.is_ok());
+                       assert_eq!(network_graph.read_only().channels().len(), 0);
+               }
+       }
+
        #[test]
        fn timestamp_edge_cases_are_handled_correctly() {
                // this is the timestamp encoded in the binary data of valid_input below
@@ -576,17 +617,17 @@ mod tests {
                        let network_graph = NetworkGraph::new(Network::Bitcoin, &logger);
                        assert_eq!(network_graph.read_only().channels().len(), 0);
 
-                       let rapid_sync = RapidGossipSync::new(&network_graph);
+                       let rapid_sync = RapidGossipSync::new(&network_graph, &logger);
                        let update_result = rapid_sync.update_network_graph_no_std(&VALID_RGS_BINARY, Some(latest_succeeding_time));
                        assert!(update_result.is_ok());
-                       assert_eq!(network_graph.read_only().channels().len(), 2);
+                       assert_eq!(network_graph.read_only().channels().len(), 0);
                }
 
                {
                        let network_graph = NetworkGraph::new(Network::Bitcoin, &logger);
                        assert_eq!(network_graph.read_only().channels().len(), 0);
 
-                       let rapid_sync = RapidGossipSync::new(&network_graph);
+                       let rapid_sync = RapidGossipSync::new(&network_graph, &logger);
                        let update_result = rapid_sync.update_network_graph_no_std(&VALID_RGS_BINARY, Some(earliest_failing_time));
                        assert!(update_result.is_err());
                        if let Err(GraphSyncError::LightningError(lightning_error)) = update_result {
@@ -601,6 +642,7 @@ mod tests {
        }
 
        #[test]
+       #[cfg(feature = "std")]
        pub fn update_fails_with_unknown_version() {
                let unknown_version_input = vec![
                        76, 68, 75, 2, 111, 226, 140, 10, 182, 241, 179, 114, 193, 166, 162, 70, 174, 99, 247,
@@ -622,7 +664,7 @@ mod tests {
 
                let logger = TestLogger::new();
                let network_graph = NetworkGraph::new(Network::Bitcoin, &logger);
-               let rapid_sync = RapidGossipSync::new(&network_graph);
+               let rapid_sync = RapidGossipSync::new(&network_graph, &logger);
                let update_result = rapid_sync.update_network_graph(&unknown_version_input[..]);
 
                assert!(update_result.is_err());
@@ -633,4 +675,22 @@ mod tests {
                        panic!("Unexpected update result: {:?}", update_result)
                }
        }
+
+       #[test]
+       fn fails_early_on_chain_hash_mismatch() {
+               let logger = TestLogger::new();
+               // Set to testnet so that the VALID_RGS_BINARY chain hash of mainnet does not match.
+               let network_graph = NetworkGraph::new(Network::Testnet, &logger);
+
+               assert_eq!(network_graph.read_only().channels().len(), 0);
+
+               let rapid_sync = RapidGossipSync::new(&network_graph, &logger);
+               let update_result = rapid_sync.update_network_graph_no_std(&VALID_RGS_BINARY, Some(0));
+               assert!(update_result.is_err());
+               if let Err(GraphSyncError::LightningError(err)) = update_result {
+                       assert_eq!(err.err, "Rapid Gossip Sync data's chain hash does not match the network graph's");
+               } else {
+                       panic!("Unexpected update result: {:?}", update_result)
+               }
+       }
 }