_ Git - rapid-gossip-sync-server/blob - src/tests/mod.rs

   1 //! Multi-module tests that use database fixtures
   2
   3 use std::cell::RefCell;
   4 use std::sync::Arc;
   5 use std::{fs, thread};
   6 use std::time::{SystemTime, UNIX_EPOCH};
   7 use bitcoin::blockdata::constants::ChainHash;
   8 use bitcoin::Network;
   9 use bitcoin::secp256k1::ecdsa::Signature;
  10 use bitcoin::secp256k1::{Secp256k1, SecretKey};
  11 use bitcoin::hashes::Hash;
  12 use bitcoin::hashes::sha256d::Hash as Sha256dHash;
  13 use hex_conservative::DisplayHex;
  14 use lightning::ln::features::{ChannelFeatures, NodeFeatures};
  15 use lightning::ln::msgs::{ChannelAnnouncement, ChannelUpdate, NodeAnnouncement, SocketAddress, UnsignedChannelAnnouncement, UnsignedChannelUpdate, UnsignedNodeAnnouncement};
  16 use lightning::routing::gossip::{NetworkGraph, NodeAlias, NodeId};
  17 use lightning::util::ser::Writeable;
  18 use lightning_rapid_gossip_sync::RapidGossipSync;
  19 use crate::{calculate_delta, config, serialize_delta};
  20 use crate::persistence::GossipPersister;
  21 use crate::snapshot::Snapshotter;
  22 use crate::types::{GossipMessage, tests::TestLogger};
  23
  24 const CLIENT_BACKDATE_INTERVAL: u32 = 3600 * 24 * 7; // client backdates RGS by a week
  25
  26 thread_local! {
  27         static DB_TEST_SCHEMA: RefCell<Option<String>> = RefCell::new(None);
  28         static IS_TEST_SCHEMA_CLEAN: RefCell<Option<bool>> = RefCell::new(None);
  29 }
  30
  31 fn blank_signature() -> Signature {
  32         Signature::from_compact(&[0u8; 64]).unwrap()
  33 }
  34
  35 fn genesis_hash() -> ChainHash {
  36         ChainHash::using_genesis_block(Network::Bitcoin)
  37 }
  38
  39 fn current_time() -> u32 {
  40         SystemTime::now().duration_since(UNIX_EPOCH).expect("Time must be > 1970").as_secs() as u32
  41 }
  42
  43 pub(crate) fn db_test_schema() -> String {
  44         DB_TEST_SCHEMA.with(|suffix_reference| {
  45                 let suffix_option = suffix_reference.borrow();
  46                 suffix_option.as_ref().unwrap().clone()
  47         })
  48 }
  49
  50 fn generate_node_announcement(private_key: Option<SecretKey>) -> NodeAnnouncement {
  51         let secp_context = Secp256k1::new();
  52
  53         let random_private_key = private_key.unwrap_or(SecretKey::from_slice(&[1; 32]).unwrap());
  54         let random_public_key = random_private_key.public_key(&secp_context);
  55         let node_id = NodeId::from_pubkey(&random_public_key);
  56
  57         let announcement = UnsignedNodeAnnouncement {
  58                 features: NodeFeatures::empty(),
  59                 timestamp: 0,
  60                 node_id,
  61                 rgb: [0, 128, 255],
  62                 alias: NodeAlias([0; 32]),
  63                 addresses: vec![],
  64                 excess_data: vec![],
  65                 excess_address_data: vec![],
  66         };
  67
  68         let msg_hash = bitcoin::secp256k1::Message::from_slice(&Sha256dHash::hash(&announcement.encode()[..])[..]).unwrap();
  69         let signature = secp_context.sign_ecdsa(&msg_hash, &random_private_key);
  70
  71         NodeAnnouncement {
  72                 signature,
  73                 contents: announcement,
  74         }
  75 }
  76
  77
  78 fn generate_channel_announcement(short_channel_id: u64) -> ChannelAnnouncement {
  79         let secp_context = Secp256k1::new();
  80
  81         let random_private_key_1 = SecretKey::from_slice(&[1; 32]).unwrap();
  82         let random_public_key_1 = random_private_key_1.public_key(&secp_context);
  83         let node_id_1 = NodeId::from_pubkey(&random_public_key_1);
  84
  85         let random_private_key_2 = SecretKey::from_slice(&[2; 32]).unwrap();
  86         let random_public_key_2 = random_private_key_2.public_key(&secp_context);
  87         let node_id_2 = NodeId::from_pubkey(&random_public_key_2);
  88
  89         let announcement = UnsignedChannelAnnouncement {
  90                 features: ChannelFeatures::empty(),
  91                 chain_hash: genesis_hash(),
  92                 short_channel_id,
  93                 node_id_1,
  94                 node_id_2,
  95                 bitcoin_key_1: node_id_1,
  96                 bitcoin_key_2: node_id_2,
  97                 excess_data: vec![],
  98         };
  99
 100         let msg_hash = bitcoin::secp256k1::Message::from_slice(&Sha256dHash::hash(&announcement.encode()[..])[..]).unwrap();
 101         let node_signature_1 = secp_context.sign_ecdsa(&msg_hash, &random_private_key_1);
 102         let node_signature_2 = secp_context.sign_ecdsa(&msg_hash, &random_private_key_2);
 103
 104         ChannelAnnouncement {
 105                 node_signature_1,
 106                 node_signature_2,
 107                 bitcoin_signature_1: node_signature_1,
 108                 bitcoin_signature_2: node_signature_2,
 109                 contents: announcement,
 110         }
 111 }
 112
 113 fn generate_update(scid: u64, direction: bool, timestamp: u32, expiry_delta: u16, min_msat: u64, max_msat: u64, base_msat: u32, fee_rate: u32) -> ChannelUpdate {
 114         let flag_mask = if direction { 1 } else { 0 };
 115         ChannelUpdate {
 116                 signature: blank_signature(),
 117                 contents: UnsignedChannelUpdate {
 118                         chain_hash: genesis_hash(),
 119                         short_channel_id: scid,
 120                         timestamp,
 121                         flags: 0 | flag_mask,
 122                         cltv_expiry_delta: expiry_delta,
 123                         htlc_minimum_msat: min_msat,
 124                         htlc_maximum_msat: max_msat,
 125                         fee_base_msat: base_msat,
 126                         fee_proportional_millionths: fee_rate,
 127                         excess_data: vec![],
 128                 },
 129         }
 130 }
 131
 132 struct SchemaSanitizer {}
 133
 134 impl SchemaSanitizer {
 135         fn new() -> Self {
 136                 IS_TEST_SCHEMA_CLEAN.with(|cleanliness_reference| {
 137                         let mut is_clean_option = cleanliness_reference.borrow_mut();
 138                         assert!(is_clean_option.is_none());
 139                         *is_clean_option = Some(false);
 140                 });
 141
 142                 DB_TEST_SCHEMA.with(|suffix_reference| {
 143                         let mut suffix_option = suffix_reference.borrow_mut();
 144                         let current_time = SystemTime::now();
 145                         let unix_time = current_time.duration_since(UNIX_EPOCH).expect("Time went backwards");
 146                         let timestamp_seconds = unix_time.as_secs();
 147                         let timestamp_nanos = unix_time.as_nanos();
 148                         // sometimes Rust thinks two tests start at the same nanosecond, causing a schema conflict
 149                         let thread_id = thread::current().id();
 150                         let preimage = format!("{:?}-{}", thread_id, timestamp_nanos);
 151                         println!("test schema preimage: {}", preimage);
 152                         let suffix = Sha256dHash::hash(preimage.as_bytes()).encode();
 153                         // the schema must start with a letter
 154                         let schema = format!("test_{}_{}", timestamp_seconds, suffix.as_hex());
 155                         *suffix_option = Some(schema);
 156                 });
 157
 158                 return Self {};
 159         }
 160 }
 161
 162 impl Drop for SchemaSanitizer {
 163         fn drop(&mut self) {
 164                 IS_TEST_SCHEMA_CLEAN.with(|cleanliness_reference| {
 165                         let is_clean_option = cleanliness_reference.borrow();
 166                         if let Some(is_clean) = *is_clean_option {
 167                                 if std::thread::panicking() {
 168                                         return;
 169                                 }
 170                                 assert_eq!(is_clean, true);
 171                         }
 172                 });
 173         }
 174 }
 175
 176 struct CacheSanitizer {}
 177
 178 impl CacheSanitizer {
 179         /// The CacheSanitizer instantiation requires that there be a schema sanitizer
 180         fn new(_: &SchemaSanitizer) -> Self {
 181                 Self {}
 182         }
 183
 184         fn cache_path(&self) -> String {
 185                 format!("./res/{}/", db_test_schema())
 186         }
 187 }
 188
 189 impl Drop for CacheSanitizer {
 190         fn drop(&mut self) {
 191                 let cache_path = self.cache_path();
 192                 fs::remove_dir_all(cache_path).unwrap();
 193         }
 194 }
 195
 196
 197 async fn clean_test_db() {
 198         let client = crate::connect_to_db().await;
 199         let schema = db_test_schema();
 200         client.execute(&format!("DROP SCHEMA IF EXISTS {} CASCADE", schema), &[]).await.unwrap();
 201         IS_TEST_SCHEMA_CLEAN.with(|cleanliness_reference| {
 202                 let mut is_clean_option = cleanliness_reference.borrow_mut();
 203                 *is_clean_option = Some(true);
 204         });
 205 }
 206
 207 #[tokio::test]
 208 async fn test_persistence_runtime() {
 209         let _sanitizer = SchemaSanitizer::new();
 210         let logger = Arc::new(TestLogger::new());
 211         let network_graph = NetworkGraph::new(Network::Bitcoin, logger.clone());
 212         let network_graph_arc = Arc::new(network_graph);
 213         let (_persister, _receiver) = GossipPersister::new(network_graph_arc.clone(), logger.clone());
 214
 215         tokio::task::spawn_blocking(move || {
 216                 drop(_persister);
 217         }).await.unwrap();
 218
 219         clean_test_db().await;
 220 }
 221
 222
 223 #[tokio::test]
 224 async fn test_trivial_setup() {
 225         let _sanitizer = SchemaSanitizer::new();
 226         let logger = Arc::new(TestLogger::new());
 227         let network_graph = NetworkGraph::new(Network::Bitcoin, logger.clone());
 228         let network_graph_arc = Arc::new(network_graph);
 229         let (mut persister, receiver) = GossipPersister::new(network_graph_arc.clone(), logger.clone());
 230
 231         let short_channel_id = 1;
 232         let timestamp = current_time() - 10;
 233         println!("timestamp: {}", timestamp);
 234
 235         { // seed the db
 236                 let announcement = generate_channel_announcement(short_channel_id);
 237                 let update_1 = generate_update(short_channel_id, false, timestamp, 0, 0, 0, 5, 0);
 238                 let update_2 = generate_update(short_channel_id, true, timestamp, 0, 0, 0, 10, 0);
 239
 240                 network_graph_arc.update_channel_from_announcement_no_lookup(&announcement).unwrap();
 241                 network_graph_arc.update_channel_unsigned(&update_1.contents).unwrap();
 242                 network_graph_arc.update_channel_unsigned(&update_2.contents).unwrap();
 243
 244                 receiver.send(GossipMessage::ChannelAnnouncement(announcement, None)).await.unwrap();
 245                 receiver.send(GossipMessage::ChannelUpdate(update_1, None)).await.unwrap();
 246                 receiver.send(GossipMessage::ChannelUpdate(update_2, None)).await.unwrap();
 247                 drop(receiver);
 248                 persister.persist_gossip().await;
 249         }
 250
 251         let delta = calculate_delta(network_graph_arc.clone(), 0, None, logger.clone()).await;
 252         let serialization = serialize_delta(&delta, 1, logger.clone());
 253         logger.assert_log_contains("rapid_gossip_sync_server", "announcement channel count: 1", 1);
 254         clean_test_db().await;
 255
 256         let channel_count = network_graph_arc.read_only().channels().len();
 257
 258         assert_eq!(channel_count, 1);
 259         assert_eq!(serialization.message_count, 3);
 260         assert_eq!(serialization.channel_announcement_count, 1);
 261         assert_eq!(serialization.update_count, 2);
 262
 263         let client_graph = NetworkGraph::new(Network::Bitcoin, logger.clone());
 264         let client_graph_arc = Arc::new(client_graph);
 265         let rgs = RapidGossipSync::new(client_graph_arc.clone(), logger.clone());
 266         let update_result = rgs.update_network_graph(&serialization.data).unwrap();
 267         println!("update result: {}", update_result);
 268         // the update result must be a multiple of our snapshot granularity
 269         assert_eq!(update_result % config::snapshot_generation_interval(), 0);
 270         assert!(update_result < timestamp);
 271
 272         let timestamp_delta = timestamp - update_result;
 273         println!("timestamp delta: {}", timestamp_delta);
 274         assert!(timestamp_delta < config::snapshot_generation_interval());
 275
 276         let readonly_graph = client_graph_arc.read_only();
 277         let channels = readonly_graph.channels();
 278         let client_channel_count = channels.len();
 279         assert_eq!(client_channel_count, 1);
 280
 281         let first_channel = channels.get(&short_channel_id).unwrap();
 282         assert!(&first_channel.announcement_message.is_none());
 283         assert_eq!(first_channel.one_to_two.as_ref().unwrap().fees.base_msat, 5);
 284         assert_eq!(first_channel.two_to_one.as_ref().unwrap().fees.base_msat, 10);
 285         let last_update_seen_a = first_channel.one_to_two.as_ref().unwrap().last_update;
 286         let last_update_seen_b = first_channel.two_to_one.as_ref().unwrap().last_update;
 287         println!("last update a: {}", last_update_seen_a);
 288         println!("last update b: {}", last_update_seen_b);
 289         assert_eq!(last_update_seen_a, update_result - CLIENT_BACKDATE_INTERVAL);
 290         assert_eq!(last_update_seen_b, update_result - CLIENT_BACKDATE_INTERVAL);
 291
 292         tokio::task::spawn_blocking(move || {
 293                 drop(persister);
 294         }).await.unwrap();
 295 }
 296
 297 #[tokio::test]
 298 async fn test_node_announcement_persistence() {
 299         let _sanitizer = SchemaSanitizer::new();
 300         let logger = Arc::new(TestLogger::new());
 301         let network_graph = NetworkGraph::new(Network::Bitcoin, logger.clone());
 302         let network_graph_arc = Arc::new(network_graph);
 303         let (mut persister, receiver) = GossipPersister::new(network_graph_arc.clone(), logger.clone());
 304
 305         { // seed the db
 306                 let mut announcement = generate_node_announcement(None);
 307                 receiver.send(GossipMessage::NodeAnnouncement(announcement.clone(), None)).await.unwrap();
 308                 receiver.send(GossipMessage::NodeAnnouncement(announcement.clone(), Some(12345))).await.unwrap();
 309
 310                 {
 311                         // modify announcement to contain a bunch of addresses
 312                         announcement.contents.addresses.push(SocketAddress::Hostname {
 313                                 hostname: "google.com".to_string().try_into().unwrap(),
 314                                 port: 443,
 315                         });
 316                         announcement.contents.addresses.push(SocketAddress::TcpIpV4 { addr: [127, 0, 0, 1], port: 9635 });
 317                         announcement.contents.addresses.push(SocketAddress::TcpIpV6 { addr: [1; 16], port: 1337 });
 318                         announcement.contents.addresses.push(SocketAddress::OnionV2([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]));
 319                         announcement.contents.addresses.push(SocketAddress::OnionV3 {
 320                                 ed25519_pubkey: [1; 32],
 321                                 checksum: 2,
 322                                 version: 3,
 323                                 port: 4,
 324                         });
 325                 }
 326                 receiver.send(GossipMessage::NodeAnnouncement(announcement, Some(12345))).await.unwrap();
 327
 328                 drop(receiver);
 329                 persister.persist_gossip().await;
 330
 331                 tokio::task::spawn_blocking(move || {
 332                         drop(persister);
 333                 }).await.unwrap();
 334         }
 335         clean_test_db().await;
 336 }
 337
 338 #[tokio::test]
 339 async fn test_node_announcement_delta_detection() {
 340         let _sanitizer = SchemaSanitizer::new();
 341         let logger = Arc::new(TestLogger::new());
 342         let network_graph = NetworkGraph::new(Network::Bitcoin, logger.clone());
 343         let network_graph_arc = Arc::new(network_graph);
 344         let (mut persister, receiver) = GossipPersister::new(network_graph_arc.clone(), logger.clone());
 345
 346         let timestamp = current_time() - 10;
 347
 348         { // seed the db
 349                 let mut announcement = generate_node_announcement(None);
 350                 receiver.send(GossipMessage::NodeAnnouncement(announcement.clone(), Some(timestamp - 10))).await.unwrap();
 351                 receiver.send(GossipMessage::NodeAnnouncement(announcement.clone(), Some(timestamp - 8))).await.unwrap();
 352
 353                 {
 354                         let mut current_announcement = generate_node_announcement(Some(SecretKey::from_slice(&[2; 32]).unwrap()));
 355                         current_announcement.contents.features = NodeFeatures::from_be_bytes(vec![23, 48]);
 356                         receiver.send(GossipMessage::NodeAnnouncement(current_announcement, Some(timestamp))).await.unwrap();
 357                 }
 358
 359                 {
 360                         let mut current_announcement = generate_node_announcement(Some(SecretKey::from_slice(&[3; 32]).unwrap()));
 361                         current_announcement.contents.features = NodeFeatures::from_be_bytes(vec![22, 49]);
 362                         receiver.send(GossipMessage::NodeAnnouncement(current_announcement, Some(timestamp))).await.unwrap();
 363                 }
 364
 365                 {
 366                         // modify announcement to contain a bunch of addresses
 367                         announcement.contents.addresses.push(SocketAddress::Hostname {
 368                                 hostname: "google.com".to_string().try_into().unwrap(),
 369                                 port: 443,
 370                         });
 371                         announcement.contents.features = NodeFeatures::from_be_bytes(vec![23, 48]);
 372                         announcement.contents.addresses.push(SocketAddress::TcpIpV4 { addr: [127, 0, 0, 1], port: 9635 });
 373                         announcement.contents.addresses.push(SocketAddress::TcpIpV6 { addr: [1; 16], port: 1337 });
 374                         announcement.contents.addresses.push(SocketAddress::OnionV2([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]));
 375                         announcement.contents.addresses.push(SocketAddress::OnionV3 {
 376                                 ed25519_pubkey: [1; 32],
 377                                 checksum: 2,
 378                                 version: 3,
 379                                 port: 4,
 380                         });
 381                 }
 382                 receiver.send(GossipMessage::NodeAnnouncement(announcement, Some(timestamp))).await.unwrap();
 383
 384                 { // necessary for the node announcements to be considered relevant
 385                         let announcement = generate_channel_announcement(1);
 386                         let update_1 = generate_update(1, false, timestamp, 0, 0, 0, 6, 0);
 387                         let update_2 = generate_update(1, true, timestamp, 0, 0, 0, 6, 0);
 388
 389                         network_graph_arc.update_channel_from_announcement_no_lookup(&announcement).unwrap();
 390                         network_graph_arc.update_channel_unsigned(&update_1.contents).unwrap();
 391                         network_graph_arc.update_channel_unsigned(&update_2.contents).unwrap();
 392
 393                         receiver.send(GossipMessage::ChannelAnnouncement(announcement, Some(timestamp))).await.unwrap();
 394                         receiver.send(GossipMessage::ChannelUpdate(update_1, Some(timestamp))).await.unwrap();
 395                         receiver.send(GossipMessage::ChannelUpdate(update_2, Some(timestamp))).await.unwrap();
 396                 }
 397
 398                 drop(receiver);
 399                 persister.persist_gossip().await;
 400
 401                 tokio::task::spawn_blocking(move || {
 402                         drop(persister);
 403                 }).await.unwrap();
 404         }
 405
 406         let delta = calculate_delta(network_graph_arc.clone(), timestamp - 5, None, logger.clone()).await;
 407         let serialization = serialize_delta(&delta, 2, logger.clone());
 408         clean_test_db().await;
 409
 410         assert_eq!(serialization.message_count, 3);
 411         assert_eq!(serialization.node_announcement_count, 3);
 412         assert_eq!(serialization.node_update_count, 1);
 413         assert_eq!(serialization.node_feature_update_count, 1);
 414         assert_eq!(serialization.node_address_update_count, 1);
 415 }
 416
 417 /// If a channel has only seen updates in one direction, it should not be announced
 418 #[tokio::test]
 419 async fn test_unidirectional_intermediate_update_consideration() {
 420         let _sanitizer = SchemaSanitizer::new();
 421
 422         let logger = Arc::new(TestLogger::new());
 423         let network_graph = NetworkGraph::new(Network::Bitcoin, logger.clone());
 424         let network_graph_arc = Arc::new(network_graph);
 425         let (mut persister, receiver) = GossipPersister::new(network_graph_arc.clone(), logger.clone());
 426
 427         let short_channel_id = 1;
 428         let timestamp = current_time() - 10;
 429         println!("timestamp: {}", timestamp);
 430
 431         { // seed the db
 432                 let announcement = generate_channel_announcement(short_channel_id);
 433                 let update_1 = generate_update(short_channel_id, false, timestamp, 0, 0, 0, 6, 0);
 434                 let update_2 = generate_update(short_channel_id, true, timestamp + 1, 0, 0, 0, 3, 0);
 435                 let update_3 = generate_update(short_channel_id, true, timestamp + 2, 0, 0, 0, 4, 0);
 436
 437                 network_graph_arc.update_channel_from_announcement_no_lookup(&announcement).unwrap();
 438                 network_graph_arc.update_channel_unsigned(&update_1.contents).unwrap();
 439                 network_graph_arc.update_channel_unsigned(&update_2.contents).unwrap();
 440                 network_graph_arc.update_channel_unsigned(&update_3.contents).unwrap();
 441
 442                 receiver.send(GossipMessage::ChannelAnnouncement(announcement, Some(timestamp))).await.unwrap();
 443                 receiver.send(GossipMessage::ChannelUpdate(update_1, None)).await.unwrap();
 444                 receiver.send(GossipMessage::ChannelUpdate(update_2, None)).await.unwrap();
 445                 receiver.send(GossipMessage::ChannelUpdate(update_3, None)).await.unwrap();
 446                 drop(receiver);
 447                 persister.persist_gossip().await;
 448         }
 449
 450         let channel_count = network_graph_arc.read_only().channels().len();
 451         assert_eq!(channel_count, 1);
 452
 453         let client_graph = NetworkGraph::new(Network::Bitcoin, logger.clone());
 454         let client_graph_arc = Arc::new(client_graph);
 455         let rgs = RapidGossipSync::new(client_graph_arc.clone(), logger.clone());
 456
 457         let delta = calculate_delta(network_graph_arc.clone(), timestamp + 1, None, logger.clone()).await;
 458         let serialization = serialize_delta(&delta, 1, logger.clone());
 459
 460         logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Fetched 1 update rows of the first update in a new direction", 1);
 461         logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Processed 1 reference rows", 1);
 462         logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Processed intermediate rows (2)", 1);
 463
 464         assert_eq!(serialization.message_count, 3);
 465         assert_eq!(serialization.channel_announcement_count, 1);
 466         assert_eq!(serialization.update_count, 2);
 467         assert_eq!(serialization.update_count_full, 2);
 468         assert_eq!(serialization.update_count_incremental, 0);
 469
 470         let update_result = rgs.update_network_graph(&serialization.data).unwrap();
 471         println!("update result: {}", update_result);
 472         // the update result must be a multiple of our snapshot granularity
 473
 474         let readonly_graph = client_graph_arc.read_only();
 475         let channels = readonly_graph.channels();
 476         let client_channel_count = channels.len();
 477         assert_eq!(client_channel_count, 1);
 478
 479         tokio::task::spawn_blocking(move || {
 480                 drop(persister);
 481         }).await.unwrap();
 482
 483         clean_test_db().await;
 484 }
 485
 486 /// If a channel has only seen updates in one direction, it should not be announced
 487 #[tokio::test]
 488 async fn test_bidirectional_intermediate_update_consideration() {
 489         let _sanitizer = SchemaSanitizer::new();
 490
 491         let logger = Arc::new(TestLogger::new());
 492         let network_graph = NetworkGraph::new(Network::Bitcoin, logger.clone());
 493         let network_graph_arc = Arc::new(network_graph);
 494         let (mut persister, receiver) = GossipPersister::new(network_graph_arc.clone(), logger.clone());
 495
 496         let short_channel_id = 1;
 497         let timestamp = current_time() - 10;
 498         println!("timestamp: {}", timestamp);
 499
 500         { // seed the db
 501                 let announcement = generate_channel_announcement(short_channel_id);
 502                 let update_1 = generate_update(short_channel_id, false, timestamp, 0, 0, 0, 5, 0);
 503                 let update_2 = generate_update(short_channel_id, false, timestamp + 1, 0, 0, 0, 4, 0);
 504                 let update_3 = generate_update(short_channel_id, false, timestamp + 2, 0, 0, 0, 3, 0);
 505                 let update_4 = generate_update(short_channel_id, true, timestamp, 0, 0, 0, 3, 0);
 506
 507                 network_graph_arc.update_channel_from_announcement_no_lookup(&announcement).unwrap();
 508                 network_graph_arc.update_channel_unsigned(&update_1.contents).unwrap();
 509                 network_graph_arc.update_channel_unsigned(&update_2.contents).unwrap();
 510                 network_graph_arc.update_channel_unsigned(&update_3.contents).unwrap();
 511                 network_graph_arc.update_channel_unsigned(&update_4.contents).unwrap();
 512
 513                 receiver.send(GossipMessage::ChannelAnnouncement(announcement, Some(timestamp))).await.unwrap();
 514                 receiver.send(GossipMessage::ChannelUpdate(update_1, None)).await.unwrap();
 515                 receiver.send(GossipMessage::ChannelUpdate(update_2, None)).await.unwrap();
 516                 receiver.send(GossipMessage::ChannelUpdate(update_3, None)).await.unwrap();
 517                 receiver.send(GossipMessage::ChannelUpdate(update_4, None)).await.unwrap();
 518                 drop(receiver);
 519                 persister.persist_gossip().await;
 520         }
 521
 522         let channel_count = network_graph_arc.read_only().channels().len();
 523         assert_eq!(channel_count, 1);
 524
 525         let delta = calculate_delta(network_graph_arc.clone(), timestamp + 1, None, logger.clone()).await;
 526         let serialization = serialize_delta(&delta, 1, logger.clone());
 527
 528         logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Fetched 0 update rows of the first update in a new direction", 1);
 529         logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Processed 2 reference rows", 1);
 530         logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Processed intermediate rows (2)", 1);
 531
 532         assert_eq!(serialization.message_count, 1);
 533         assert_eq!(serialization.channel_announcement_count, 0);
 534         assert_eq!(serialization.update_count, 1);
 535         assert_eq!(serialization.update_count_full, 0);
 536         assert_eq!(serialization.update_count_incremental, 1);
 537
 538         tokio::task::spawn_blocking(move || {
 539                 drop(persister);
 540         }).await.unwrap();
 541
 542         clean_test_db().await;
 543 }
 544
 545 #[tokio::test]
 546 async fn test_channel_reminders() {
 547         let _sanitizer = SchemaSanitizer::new();
 548
 549         let logger = Arc::new(TestLogger::new());
 550         let network_graph = NetworkGraph::new(Network::Bitcoin, logger.clone());
 551         let network_graph_arc = Arc::new(network_graph);
 552         let (mut persister, receiver) = GossipPersister::new(network_graph_arc.clone(), logger.clone());
 553
 554         let timestamp = current_time();
 555         println!("timestamp: {}", timestamp);
 556         let channel_reminder_delta = config::CHANNEL_REMINDER_AGE.as_secs() as u32;
 557
 558         { // seed the db
 559                 { // unupdated channel
 560                         let short_channel_id = 1;
 561                         let announcement = generate_channel_announcement(short_channel_id);
 562                         let update_1 = generate_update(short_channel_id, false, timestamp - channel_reminder_delta - 1, 0, 0, 0, 5, 0);
 563                         let update_2 = generate_update(short_channel_id, true, timestamp - channel_reminder_delta - 1, 0, 0, 0, 3, 0);
 564
 565                         network_graph_arc.update_channel_from_announcement_no_lookup(&announcement).unwrap();
 566                         network_graph_arc.update_channel_unsigned(&update_1.contents).unwrap();
 567                         network_graph_arc.update_channel_unsigned(&update_2.contents).unwrap();
 568
 569                         receiver.send(GossipMessage::ChannelAnnouncement(announcement, Some(timestamp - channel_reminder_delta - 1))).await.unwrap();
 570                         receiver.send(GossipMessage::ChannelUpdate(update_1, Some(timestamp - channel_reminder_delta - 1))).await.unwrap();
 571                         receiver.send(GossipMessage::ChannelUpdate(update_2, Some(timestamp - channel_reminder_delta - 1))).await.unwrap();
 572                 }
 573                 { // unmodified but updated channel
 574                         let short_channel_id = 2;
 575                         let announcement = generate_channel_announcement(short_channel_id);
 576                         let update_1 = generate_update(short_channel_id, false, timestamp - channel_reminder_delta - 10, 0, 0, 0, 5, 0);
 577                         // in the false direction, we have one update that's different prior
 578                         let update_2 = generate_update(short_channel_id, false, timestamp - channel_reminder_delta - 5, 0, 1, 0, 5, 0);
 579                         let update_3 = generate_update(short_channel_id, false, timestamp - channel_reminder_delta - 1, 0, 0, 0, 5, 0);
 580                         let update_4 = generate_update(short_channel_id, true, timestamp - channel_reminder_delta - 1, 0, 0, 0, 3, 0);
 581                         let update_5 = generate_update(short_channel_id, false, timestamp - channel_reminder_delta + 10, 0, 0, 0, 5, 0);
 582                         let update_6 = generate_update(short_channel_id, true, timestamp - channel_reminder_delta + 10, 0, 0, 0, 3, 0);
 583                         let update_7 = generate_update(short_channel_id, false, timestamp - channel_reminder_delta + 20, 0, 0, 0, 5, 0);
 584                         let update_8 = generate_update(short_channel_id, true, timestamp - channel_reminder_delta + 20, 0, 0, 0, 3, 0);
 585
 586                         network_graph_arc.update_channel_from_announcement_no_lookup(&announcement).unwrap();
 587                         network_graph_arc.update_channel_unsigned(&update_7.contents).unwrap();
 588                         network_graph_arc.update_channel_unsigned(&update_8.contents).unwrap();
 589
 590                         receiver.send(GossipMessage::ChannelAnnouncement(announcement, Some(timestamp - channel_reminder_delta - 1))).await.unwrap();
 591                         receiver.send(GossipMessage::ChannelUpdate(update_1, Some(timestamp - channel_reminder_delta - 10))).await.unwrap();
 592                         receiver.send(GossipMessage::ChannelUpdate(update_2, Some(timestamp - channel_reminder_delta - 5))).await.unwrap();
 593                         receiver.send(GossipMessage::ChannelUpdate(update_3, Some(timestamp - channel_reminder_delta - 1))).await.unwrap();
 594                         receiver.send(GossipMessage::ChannelUpdate(update_4, Some(timestamp - channel_reminder_delta - 1))).await.unwrap();
 595
 596                         receiver.send(GossipMessage::ChannelUpdate(update_5, Some(timestamp - channel_reminder_delta + 10))).await.unwrap();
 597                         receiver.send(GossipMessage::ChannelUpdate(update_6, Some(timestamp - channel_reminder_delta + 10))).await.unwrap();
 598
 599                         receiver.send(GossipMessage::ChannelUpdate(update_7, Some(timestamp - channel_reminder_delta + 20))).await.unwrap();
 600                         receiver.send(GossipMessage::ChannelUpdate(update_8, Some(timestamp - channel_reminder_delta + 20))).await.unwrap();
 601                 }
 602                 drop(receiver);
 603                 persister.persist_gossip().await;
 604         }
 605
 606         let channel_count = network_graph_arc.read_only().channels().len();
 607         assert_eq!(channel_count, 2);
 608
 609         let delta = calculate_delta(network_graph_arc.clone(), timestamp - channel_reminder_delta + 15, None, logger.clone()).await;
 610         let serialization = serialize_delta(&delta, 1, logger.clone());
 611
 612         logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Fetched 0 update rows of the first update in a new direction", 1);
 613         logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Fetched 4 update rows of the latest update in the less recently updated direction", 1);
 614         logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Processed 2 reference rows", 1);
 615         logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Processed intermediate rows (2)", 1);
 616
 617         assert_eq!(serialization.message_count, 4);
 618         assert_eq!(serialization.channel_announcement_count, 0);
 619         assert_eq!(serialization.update_count, 4);
 620         assert_eq!(serialization.update_count_full, 0);
 621         assert_eq!(serialization.update_count_incremental, 4);
 622
 623         tokio::task::spawn_blocking(move || {
 624                 drop(persister);
 625         }).await.unwrap();
 626
 627         clean_test_db().await;
 628 }
 629
 630 #[tokio::test]
 631 async fn test_full_snapshot_recency() {
 632         let _sanitizer = SchemaSanitizer::new();
 633         let logger = Arc::new(TestLogger::new());
 634         let network_graph = NetworkGraph::new(Network::Bitcoin, logger.clone());
 635         let network_graph_arc = Arc::new(network_graph);
 636
 637         let short_channel_id = 1;
 638         let timestamp = current_time();
 639         println!("timestamp: {}", timestamp);
 640
 641         { // seed the db
 642                 let (mut persister, receiver) = GossipPersister::new(network_graph_arc.clone(), logger.clone());
 643                 let announcement = generate_channel_announcement(short_channel_id);
 644                 network_graph_arc.update_channel_from_announcement_no_lookup(&announcement).unwrap();
 645                 receiver.send(GossipMessage::ChannelAnnouncement(announcement, None)).await.unwrap();
 646
 647                 { // direction false
 648                         { // first update
 649                                 let update = generate_update(short_channel_id, false, timestamp - 1, 0, 0, 0, 0, 38);
 650                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 651                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 652                         }
 653                         { // second update
 654                                 let update = generate_update(short_channel_id, false, timestamp, 0, 0, 0, 0, 39);
 655                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 656                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 657                         }
 658                 }
 659                 { // direction true
 660                         { // first and only update
 661                                 let update = generate_update(short_channel_id, true, timestamp, 0, 0, 0, 0, 10);
 662                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 663                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 664                         }
 665                 }
 666
 667                 drop(receiver);
 668                 persister.persist_gossip().await;
 669
 670                 tokio::task::spawn_blocking(move || {
 671                         drop(persister);
 672                 }).await.unwrap();
 673         }
 674
 675         let client_graph = NetworkGraph::new(Network::Bitcoin, logger.clone());
 676         let client_graph_arc = Arc::new(client_graph);
 677
 678         { // sync after initial seed
 679                 let delta = calculate_delta(network_graph_arc.clone(), 0, None, logger.clone()).await;
 680                 let serialization = serialize_delta(&delta, 1, logger.clone());
 681                 logger.assert_log_contains("rapid_gossip_sync_server", "announcement channel count: 1", 1);
 682
 683                 let channel_count = network_graph_arc.read_only().channels().len();
 684
 685                 assert_eq!(channel_count, 1);
 686                 assert_eq!(serialization.message_count, 3);
 687                 assert_eq!(serialization.channel_announcement_count, 1);
 688                 assert_eq!(serialization.update_count, 2);
 689
 690                 let rgs = RapidGossipSync::new(client_graph_arc.clone(), logger.clone());
 691                 let update_result = rgs.update_network_graph(&serialization.data).unwrap();
 692                 // the update result must be a multiple of our snapshot granularity
 693                 assert_eq!(update_result % config::snapshot_generation_interval(), 0);
 694                 assert!(update_result < timestamp);
 695
 696                 let readonly_graph = client_graph_arc.read_only();
 697                 let channels = readonly_graph.channels();
 698                 let client_channel_count = channels.len();
 699                 assert_eq!(client_channel_count, 1);
 700
 701                 let first_channel = channels.get(&short_channel_id).unwrap();
 702                 assert!(&first_channel.announcement_message.is_none());
 703                 // ensure the update in one direction shows the latest fee
 704                 assert_eq!(first_channel.one_to_two.as_ref().unwrap().fees.proportional_millionths, 39);
 705                 assert_eq!(first_channel.two_to_one.as_ref().unwrap().fees.proportional_millionths, 10);
 706         }
 707
 708         clean_test_db().await;
 709 }
 710
 711 #[tokio::test]
 712 async fn test_full_snapshot_recency_with_wrong_seen_order() {
 713         let _sanitizer = SchemaSanitizer::new();
 714         let logger = Arc::new(TestLogger::new());
 715         let network_graph = NetworkGraph::new(Network::Bitcoin, logger.clone());
 716         let network_graph_arc = Arc::new(network_graph);
 717
 718         let short_channel_id = 1;
 719         let timestamp = current_time();
 720         println!("timestamp: {}", timestamp);
 721
 722         { // seed the db
 723                 let (mut persister, receiver) = GossipPersister::new(network_graph_arc.clone(), logger.clone());
 724                 let announcement = generate_channel_announcement(short_channel_id);
 725                 network_graph_arc.update_channel_from_announcement_no_lookup(&announcement).unwrap();
 726                 receiver.send(GossipMessage::ChannelAnnouncement(announcement, None)).await.unwrap();
 727
 728                 { // direction false
 729                         { // first update, seen latest
 730                                 let update = generate_update(short_channel_id, false, timestamp - 1, 0, 0, 0, 0, 38);
 731                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 732                                 receiver.send(GossipMessage::ChannelUpdate(update, Some(timestamp))).await.unwrap();
 733                         }
 734                         { // second update, seen first
 735                                 let update = generate_update(short_channel_id, false, timestamp, 0, 0, 0, 0, 39);
 736                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 737                                 receiver.send(GossipMessage::ChannelUpdate(update, Some(timestamp - 1))).await.unwrap();
 738                         }
 739                 }
 740                 { // direction true
 741                         { // first and only update
 742                                 let update = generate_update(short_channel_id, true, timestamp, 0, 0, 0, 0, 10);
 743                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 744                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 745                         }
 746                 }
 747
 748                 drop(receiver);
 749                 persister.persist_gossip().await;
 750
 751                 tokio::task::spawn_blocking(move || {
 752                         drop(persister);
 753                 }).await.unwrap();
 754         }
 755
 756         let client_graph = NetworkGraph::new(Network::Bitcoin, logger.clone());
 757         let client_graph_arc = Arc::new(client_graph);
 758
 759         { // sync after initial seed
 760                 let delta = calculate_delta(network_graph_arc.clone(), 0, None, logger.clone()).await;
 761                 let serialization = serialize_delta(&delta, 1, logger.clone());
 762                 logger.assert_log_contains("rapid_gossip_sync_server", "announcement channel count: 1", 1);
 763
 764                 let channel_count = network_graph_arc.read_only().channels().len();
 765
 766                 assert_eq!(channel_count, 1);
 767                 assert_eq!(serialization.message_count, 3);
 768                 assert_eq!(serialization.channel_announcement_count, 1);
 769                 assert_eq!(serialization.update_count, 2);
 770
 771                 let rgs = RapidGossipSync::new(client_graph_arc.clone(), logger.clone());
 772                 let update_result = rgs.update_network_graph(&serialization.data).unwrap();
 773                 // the update result must be a multiple of our snapshot granularity
 774                 assert_eq!(update_result % config::snapshot_generation_interval(), 0);
 775                 assert!(update_result < timestamp);
 776
 777                 let readonly_graph = client_graph_arc.read_only();
 778                 let channels = readonly_graph.channels();
 779                 let client_channel_count = channels.len();
 780                 assert_eq!(client_channel_count, 1);
 781
 782                 let first_channel = channels.get(&short_channel_id).unwrap();
 783                 assert!(&first_channel.announcement_message.is_none());
 784                 // ensure the update in one direction shows the latest fee
 785                 assert_eq!(first_channel.one_to_two.as_ref().unwrap().fees.proportional_millionths, 39);
 786                 assert_eq!(first_channel.two_to_one.as_ref().unwrap().fees.proportional_millionths, 10);
 787         }
 788
 789         clean_test_db().await;
 790 }
 791
 792 #[tokio::test]
 793 async fn test_full_snapshot_recency_with_wrong_propagation_order() {
 794         let _sanitizer = SchemaSanitizer::new();
 795         let logger = Arc::new(TestLogger::new());
 796         let network_graph = NetworkGraph::new(Network::Bitcoin, logger.clone());
 797         let network_graph_arc = Arc::new(network_graph);
 798
 799         let short_channel_id = 1;
 800         let timestamp = current_time();
 801         println!("timestamp: {}", timestamp);
 802
 803         { // seed the db
 804                 let (mut persister, receiver) = GossipPersister::new(network_graph_arc.clone(), logger.clone());
 805                 let announcement = generate_channel_announcement(short_channel_id);
 806                 network_graph_arc.update_channel_from_announcement_no_lookup(&announcement).unwrap();
 807                 receiver.send(GossipMessage::ChannelAnnouncement(announcement, None)).await.unwrap();
 808
 809                 { // direction false
 810                         // apply updates in their timestamp order
 811                         let update_1 = generate_update(short_channel_id, false, timestamp - 1, 0, 0, 0, 0, 38);
 812                         let update_2 = generate_update(short_channel_id, false, timestamp, 0, 0, 0, 0, 39);
 813                         network_graph_arc.update_channel_unsigned(&update_1.contents).unwrap();
 814                         network_graph_arc.update_channel_unsigned(&update_2.contents).unwrap();
 815
 816                         // propagate updates in their seen order
 817                         receiver.send(GossipMessage::ChannelUpdate(update_2, Some(timestamp - 1))).await.unwrap();
 818                         receiver.send(GossipMessage::ChannelUpdate(update_1, Some(timestamp))).await.unwrap();
 819                 }
 820                 { // direction true
 821                         { // first and only update
 822                                 let update = generate_update(short_channel_id, true, timestamp, 0, 0, 0, 0, 10);
 823                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 824                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 825                         }
 826                 }
 827
 828                 drop(receiver);
 829                 persister.persist_gossip().await;
 830
 831                 tokio::task::spawn_blocking(move || {
 832                         drop(persister);
 833                 }).await.unwrap();
 834         }
 835
 836         let client_graph = NetworkGraph::new(Network::Bitcoin, logger.clone());
 837         let client_graph_arc = Arc::new(client_graph);
 838
 839         { // sync after initial seed
 840                 let delta = calculate_delta(network_graph_arc.clone(), 0, None, logger.clone()).await;
 841                 let serialization = serialize_delta(&delta, 1, logger.clone());
 842                 logger.assert_log_contains("rapid_gossip_sync_server", "announcement channel count: 1", 1);
 843
 844                 let channel_count = network_graph_arc.read_only().channels().len();
 845
 846                 assert_eq!(channel_count, 1);
 847                 assert_eq!(serialization.message_count, 3);
 848                 assert_eq!(serialization.channel_announcement_count, 1);
 849                 assert_eq!(serialization.update_count, 2);
 850
 851                 let rgs = RapidGossipSync::new(client_graph_arc.clone(), logger.clone());
 852                 let update_result = rgs.update_network_graph(&serialization.data).unwrap();
 853                 // the update result must be a multiple of our snapshot granularity
 854                 assert_eq!(update_result % config::snapshot_generation_interval(), 0);
 855                 assert!(update_result < timestamp);
 856
 857                 let readonly_graph = client_graph_arc.read_only();
 858                 let channels = readonly_graph.channels();
 859                 let client_channel_count = channels.len();
 860                 assert_eq!(client_channel_count, 1);
 861
 862                 let first_channel = channels.get(&short_channel_id).unwrap();
 863                 assert!(&first_channel.announcement_message.is_none());
 864                 // ensure the update in one direction shows the latest fee
 865                 assert_eq!(first_channel.one_to_two.as_ref().unwrap().fees.proportional_millionths, 39);
 866                 assert_eq!(first_channel.two_to_one.as_ref().unwrap().fees.proportional_millionths, 10);
 867         }
 868
 869         clean_test_db().await;
 870 }
 871
 872 #[tokio::test]
 873 async fn test_full_snapshot_mutiny_scenario() {
 874         let _sanitizer = SchemaSanitizer::new();
 875         let logger = Arc::new(TestLogger::new());
 876         let network_graph = NetworkGraph::new(Network::Bitcoin, logger.clone());
 877         let network_graph_arc = Arc::new(network_graph);
 878
 879         let short_channel_id = 873706024403271681;
 880         let timestamp = current_time();
 881         // let oldest_simulation_timestamp = 1693300588;
 882         let latest_simulation_timestamp = 1695909301;
 883         let timestamp_offset = timestamp - latest_simulation_timestamp;
 884         println!("timestamp: {}", timestamp);
 885
 886         { // seed the db
 887                 let (mut persister, receiver) = GossipPersister::new(network_graph_arc.clone(), logger.clone());
 888                 let announcement = generate_channel_announcement(short_channel_id);
 889                 network_graph_arc.update_channel_from_announcement_no_lookup(&announcement).unwrap();
 890                 receiver.send(GossipMessage::ChannelAnnouncement(announcement, None)).await.unwrap();
 891
 892                 { // direction false
 893                         {
 894                                 let update = generate_update(short_channel_id, false, 1693507369 + timestamp_offset, 0, 0, 0, 0, 38);
 895                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 896                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 897                         }
 898                         {
 899                                 let update = generate_update(short_channel_id, false, 1693680390 + timestamp_offset, 0, 0, 0, 0, 38);
 900                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 901                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 902                         }
 903                         {
 904                                 let update = generate_update(short_channel_id, false, 1693749109 + timestamp_offset, 0, 0, 0, 0, 200);
 905                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 906                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 907                         }
 908                         {
 909                                 let update = generate_update(short_channel_id, false, 1693925190 + timestamp_offset, 0, 0, 0, 0, 200);
 910                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 911                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 912                         }
 913                         {
 914                                 let update = generate_update(short_channel_id, false, 1694008323 + timestamp_offset, 0, 0, 0, 0, 209);
 915                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 916                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 917                         }
 918                         {
 919                                 let update = generate_update(short_channel_id, false, 1694219924 + timestamp_offset, 0, 0, 0, 0, 209);
 920                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 921                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 922                         }
 923                         {
 924                                 let update = generate_update(short_channel_id, false, 1694267536 + timestamp_offset, 0, 0, 0, 0, 210);
 925                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 926                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 927                         }
 928                         {
 929                                 let update = generate_update(short_channel_id, false, 1694458808 + timestamp_offset, 0, 0, 0, 0, 210);
 930                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 931                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 932                         }
 933                         {
 934                                 let update = generate_update(short_channel_id, false, 1694526734 + timestamp_offset, 0, 0, 0, 0, 200);
 935                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 936                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 937                         }
 938                         {
 939                                 let update = generate_update(short_channel_id, false, 1694794765 + timestamp_offset, 0, 0, 0, 0, 200);
 940                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 941                                 receiver.send(GossipMessage::ChannelUpdate(update, Some(1695909301 + 2 * config::SYMLINK_GRANULARITY_INTERVAL + timestamp_offset))).await.unwrap();
 942                         }
 943                         {
 944                                 let update = generate_update(short_channel_id, false, 1695909301 + timestamp_offset, 0, 0, 0, 0, 130);
 945                                 // network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 946                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 947                         }
 948                 }
 949                 { // direction true
 950                         {
 951                                 let update = generate_update(short_channel_id, true, 1693300588 + timestamp_offset, 0, 0, 0, 0, 10);
 952                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 953                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 954                         }
 955                         {
 956                                 let update = generate_update(short_channel_id, true, 1695003621 + timestamp_offset, 0, 0, 0, 0, 10);
 957                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 958                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 959                         }
 960                 }
 961
 962                 drop(receiver);
 963                 persister.persist_gossip().await;
 964
 965                 tokio::task::spawn_blocking(move || {
 966                         drop(persister);
 967                 }).await.unwrap();
 968         }
 969
 970         let client_graph = NetworkGraph::new(Network::Bitcoin, logger.clone());
 971         let client_graph_arc = Arc::new(client_graph);
 972
 973         { // sync after initial seed
 974                 let delta = calculate_delta(network_graph_arc.clone(), 0, None, logger.clone()).await;
 975                 let serialization = serialize_delta(&delta, 1, logger.clone());
 976                 logger.assert_log_contains("rapid_gossip_sync_server", "announcement channel count: 1", 1);
 977
 978                 let channel_count = network_graph_arc.read_only().channels().len();
 979
 980                 assert_eq!(channel_count, 1);
 981                 assert_eq!(serialization.message_count, 3);
 982                 assert_eq!(serialization.channel_announcement_count, 1);
 983                 assert_eq!(serialization.update_count, 2);
 984
 985                 let rgs = RapidGossipSync::new(client_graph_arc.clone(), logger.clone());
 986                 let update_result = rgs.update_network_graph(&serialization.data).unwrap();
 987                 println!("update result: {}", update_result);
 988                 // the update result must be a multiple of our snapshot granularity
 989                 assert_eq!(update_result % config::snapshot_generation_interval(), 0);
 990                 assert!(update_result < timestamp);
 991
 992                 let timestamp_delta = timestamp - update_result;
 993                 println!("timestamp delta: {}", timestamp_delta);
 994                 assert!(timestamp_delta < config::snapshot_generation_interval());
 995
 996                 let readonly_graph = client_graph_arc.read_only();
 997                 let channels = readonly_graph.channels();
 998                 let client_channel_count = channels.len();
 999                 assert_eq!(client_channel_count, 1);
1000
1001                 let first_channel = channels.get(&short_channel_id).unwrap();
1002                 assert!(&first_channel.announcement_message.is_none());
1003                 assert_eq!(first_channel.one_to_two.as_ref().unwrap().fees.proportional_millionths, 130);
1004                 assert_eq!(first_channel.two_to_one.as_ref().unwrap().fees.proportional_millionths, 10);
1005         }
1006
1007         clean_test_db().await;
1008 }
1009
1010 #[tokio::test]
1011 async fn test_full_snapshot_interlaced_channel_timestamps() {
1012         let _sanitizer = SchemaSanitizer::new();
1013         let logger = Arc::new(TestLogger::new());
1014         let network_graph = NetworkGraph::new(Network::Bitcoin, logger.clone());
1015         let network_graph_arc = Arc::new(network_graph);
1016
1017         let main_channel_id = 1;
1018         let timestamp = current_time();
1019         println!("timestamp: {}", timestamp);
1020
1021         { // seed the db
1022                 let (mut persister, receiver) = GossipPersister::new(network_graph_arc.clone(), logger.clone());
1023                 let secondary_channel_id = main_channel_id + 1;
1024
1025                 { // main channel
1026                         let announcement = generate_channel_announcement(main_channel_id);
1027                         network_graph_arc.update_channel_from_announcement_no_lookup(&announcement).unwrap();
1028                         receiver.send(GossipMessage::ChannelAnnouncement(announcement, None)).await.unwrap();
1029                 }
1030
1031                 { // secondary channel
1032                         let announcement = generate_channel_announcement(secondary_channel_id);
1033                         network_graph_arc.update_channel_from_announcement_no_lookup(&announcement).unwrap();
1034                         receiver.send(GossipMessage::ChannelAnnouncement(announcement, None)).await.unwrap();
1035                 }
1036
1037                 { // main channel
1038                         { // direction false
1039                                 let update = generate_update(main_channel_id, false, timestamp - 2, 0, 0, 0, 0, 10);
1040                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
1041                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
1042                         }
1043                         { // direction true
1044                                 let update = generate_update(main_channel_id, true, timestamp - 2, 0, 0, 0, 0, 5);
1045                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
1046                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
1047                         }
1048                 }
1049
1050                 { // in-between channel
1051                         { // direction false
1052                                 let update = generate_update(secondary_channel_id, false, timestamp - 1, 0, 0, 0, 0, 42);
1053                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
1054                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
1055                         }
1056                         { // direction true
1057                                 let update = generate_update(secondary_channel_id, true, timestamp - 1, 0, 0, 0, 0, 42);
1058                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
1059                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
1060                         }
1061                 }
1062
1063                 { // main channel
1064                         { // direction false
1065                                 let update = generate_update(main_channel_id, false, timestamp, 0, 0, 0, 0, 11);
1066                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
1067                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
1068                         }
1069                         { // direction true
1070                                 let update = generate_update(main_channel_id, true, timestamp, 0, 0, 0, 0, 6);
1071                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
1072                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
1073                         }
1074                 }
1075
1076                 drop(receiver);
1077                 persister.persist_gossip().await;
1078
1079                 tokio::task::spawn_blocking(move || {
1080                         drop(persister);
1081                 }).await.unwrap();
1082         }
1083
1084         let client_graph = NetworkGraph::new(Network::Bitcoin, logger.clone());
1085         let client_graph_arc = Arc::new(client_graph);
1086
1087         { // sync after initial seed
1088                 let delta = calculate_delta(network_graph_arc.clone(), 0, None, logger.clone()).await;
1089                 let serialization = serialize_delta(&delta, 1, logger.clone());
1090                 logger.assert_log_contains("rapid_gossip_sync_server", "announcement channel count: 2", 1);
1091
1092                 let channel_count = network_graph_arc.read_only().channels().len();
1093
1094                 assert_eq!(channel_count, 2);
1095                 assert_eq!(serialization.message_count, 6);
1096                 assert_eq!(serialization.channel_announcement_count, 2);
1097                 assert_eq!(serialization.update_count, 4);
1098
1099                 let rgs = RapidGossipSync::new(client_graph_arc.clone(), logger.clone());
1100                 let update_result = rgs.update_network_graph(&serialization.data).unwrap();
1101                 // the update result must be a multiple of our snapshot granularity
1102                 assert_eq!(update_result % config::snapshot_generation_interval(), 0);
1103                 assert!(update_result < timestamp);
1104
1105                 let readonly_graph = client_graph_arc.read_only();
1106                 let channels = readonly_graph.channels();
1107                 let client_channel_count = channels.len();
1108                 assert_eq!(client_channel_count, 2);
1109
1110                 let first_channel = channels.get(&main_channel_id).unwrap();
1111                 assert!(&first_channel.announcement_message.is_none());
1112                 // ensure the update in one direction shows the latest fee
1113                 assert_eq!(first_channel.one_to_two.as_ref().unwrap().fees.proportional_millionths, 11);
1114                 assert_eq!(first_channel.two_to_one.as_ref().unwrap().fees.proportional_millionths, 6);
1115         }
1116
1117         clean_test_db().await;
1118 }
1119
1120 #[tokio::test]
1121 async fn test_full_snapshot_persistence() {
1122         let schema_sanitizer = SchemaSanitizer::new();
1123         let logger = Arc::new(TestLogger::new());
1124         let network_graph = NetworkGraph::new(Network::Bitcoin, logger.clone());
1125         let network_graph_arc = Arc::new(network_graph);
1126         let snapshotter = Snapshotter::new(network_graph_arc.clone(), logger.clone());
1127         let cache_sanitizer = CacheSanitizer::new(&schema_sanitizer);
1128
1129         let short_channel_id = 1;
1130         let timestamp = current_time();
1131         println!("timestamp: {}", timestamp);
1132
1133         { // seed the db
1134                 let (mut persister, receiver) = GossipPersister::new(network_graph_arc.clone(), logger.clone());
1135                 let announcement = generate_channel_announcement(short_channel_id);
1136                 network_graph_arc.update_channel_from_announcement_no_lookup(&announcement).unwrap();
1137                 receiver.send(GossipMessage::ChannelAnnouncement(announcement, None)).await.unwrap();
1138
1139                 { // direction true
1140                         let update = generate_update(short_channel_id, true, timestamp, 0, 0, 0, 0, 10);
1141                         network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
1142                         receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
1143                 }
1144
1145                 { // direction false
1146                         let update = generate_update(short_channel_id, false, timestamp - 1, 0, 0, 0, 0, 38);
1147                         network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
1148                         receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
1149                 }
1150
1151
1152                 drop(receiver);
1153                 persister.persist_gossip().await;
1154
1155                 tokio::task::spawn_blocking(move || {
1156                         drop(persister);
1157                 }).await.unwrap();
1158         }
1159
1160         let cache_path = cache_sanitizer.cache_path();
1161         let symlink_path = format!("{}/symlinks/0.bin", cache_path);
1162
1163         // generate snapshots
1164         {
1165                 snapshotter.generate_snapshots(20, 5, &[5, u64::MAX], &cache_path, Some(10)).await;
1166
1167                 let symlinked_data = fs::read(&symlink_path).unwrap();
1168                 let client_graph = NetworkGraph::new(Network::Bitcoin, logger.clone());
1169                 let client_graph_arc = Arc::new(client_graph);
1170
1171                 let rgs = RapidGossipSync::new(client_graph_arc.clone(), logger.clone());
1172                 let update_result = rgs.update_network_graph(&symlinked_data).unwrap();
1173                 // the update result must be a multiple of our snapshot granularity
1174                 assert_eq!(update_result % config::snapshot_generation_interval(), 0);
1175
1176                 let readonly_graph = client_graph_arc.read_only();
1177                 let channels = readonly_graph.channels();
1178                 let client_channel_count = channels.len();
1179                 assert_eq!(client_channel_count, 1);
1180
1181                 let first_channel = channels.get(&short_channel_id).unwrap();
1182                 assert!(&first_channel.announcement_message.is_none());
1183                 // ensure the update in one direction shows the latest fee
1184                 assert_eq!(first_channel.one_to_two.as_ref().unwrap().fees.proportional_millionths, 38);
1185                 assert_eq!(first_channel.two_to_one.as_ref().unwrap().fees.proportional_millionths, 10);
1186         }
1187
1188         { // update the db
1189                 let (mut persister, receiver) = GossipPersister::new(network_graph_arc.clone(), logger.clone());
1190
1191                 { // second update
1192                         let update = generate_update(short_channel_id, false, timestamp + 30, 0, 0, 0, 0, 39);
1193                         network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
1194                         receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
1195                 }
1196
1197                 drop(receiver);
1198                 persister.persist_gossip().await;
1199
1200                 tokio::task::spawn_blocking(move || {
1201                         drop(persister);
1202                 }).await.unwrap();
1203         }
1204
1205         // regenerate snapshots
1206         {
1207                 snapshotter.generate_snapshots(20, 5, &[5, u64::MAX], &cache_path, Some(10)).await;
1208
1209                 let symlinked_data = fs::read(&symlink_path).unwrap();
1210                 let client_graph = NetworkGraph::new(Network::Bitcoin, logger.clone());
1211                 let client_graph_arc = Arc::new(client_graph);
1212
1213                 let rgs = RapidGossipSync::new(client_graph_arc.clone(), logger.clone());
1214                 let update_result = rgs.update_network_graph(&symlinked_data).unwrap();
1215                 // the update result must be a multiple of our snapshot granularity
1216                 assert_eq!(update_result % config::snapshot_generation_interval(), 0);
1217
1218                 let readonly_graph = client_graph_arc.read_only();
1219                 let channels = readonly_graph.channels();
1220                 let client_channel_count = channels.len();
1221                 assert_eq!(client_channel_count, 1);
1222
1223                 let first_channel = channels.get(&short_channel_id).unwrap();
1224                 assert!(&first_channel.announcement_message.is_none());
1225                 // ensure the update in one direction shows the latest fee
1226                 assert_eq!(first_channel.one_to_two.as_ref().unwrap().fees.proportional_millionths, 39);
1227                 assert_eq!(first_channel.two_to_one.as_ref().unwrap().fees.proportional_millionths, 10);
1228         }
1229
1230         // clean up afterwards
1231         clean_test_db().await;
1232 }