_ 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::{BlockHash, Network};
   8 use bitcoin::secp256k1::ecdsa::Signature;
   9 use bitcoin::secp256k1::{Secp256k1, SecretKey};
  10 use bitcoin::hashes::Hash;
  11 use bitcoin::hashes::hex::ToHex;
  12 use bitcoin::hashes::sha256d::Hash as Sha256dHash;
  13 use lightning::ln::features::ChannelFeatures;
  14 use lightning::ln::msgs::{ChannelAnnouncement, ChannelUpdate, UnsignedChannelAnnouncement, UnsignedChannelUpdate};
  15 use lightning::routing::gossip::{NetworkGraph, NodeId};
  16 use lightning::util::ser::Writeable;
  17 use lightning_rapid_gossip_sync::RapidGossipSync;
  18 use crate::{config, serialize_delta};
  19 use crate::persistence::GossipPersister;
  20 use crate::snapshot::Snapshotter;
  21 use crate::types::{GossipMessage, tests::TestLogger};
  22
  23 const CLIENT_BACKDATE_INTERVAL: u32 = 3600 * 24 * 7; // client backdates RGS by a week
  24
  25 thread_local! {
  26         static DB_TEST_SCHEMA: RefCell<Option<String>> = RefCell::new(None);
  27         static IS_TEST_SCHEMA_CLEAN: RefCell<Option<bool>> = RefCell::new(None);
  28 }
  29
  30 fn blank_signature() -> Signature {
  31         Signature::from_compact(&[0u8; 64]).unwrap()
  32 }
  33
  34 fn genesis_hash() -> BlockHash {
  35         bitcoin::blockdata::constants::genesis_block(Network::Bitcoin).block_hash()
  36 }
  37
  38 fn current_time() -> u32 {
  39         SystemTime::now().duration_since(UNIX_EPOCH).expect("Time must be > 1970").as_secs() as u32
  40 }
  41
  42 pub(crate) fn db_test_schema() -> String {
  43         DB_TEST_SCHEMA.with(|suffix_reference| {
  44                 let suffix_option = suffix_reference.borrow();
  45                 suffix_option.as_ref().unwrap().clone()
  46         })
  47 }
  48
  49 fn generate_announcement(short_channel_id: u64) -> ChannelAnnouncement {
  50         let secp_context = Secp256k1::new();
  51
  52         let random_private_key_1 = SecretKey::from_slice(&[1; 32]).unwrap();
  53         let random_public_key_1 = random_private_key_1.public_key(&secp_context);
  54         let node_id_1 = NodeId::from_pubkey(&random_public_key_1);
  55
  56         let random_private_key_2 = SecretKey::from_slice(&[2; 32]).unwrap();
  57         let random_public_key_2 = random_private_key_2.public_key(&secp_context);
  58         let node_id_2 = NodeId::from_pubkey(&random_public_key_2);
  59
  60         let announcement = UnsignedChannelAnnouncement {
  61                 features: ChannelFeatures::empty(),
  62                 chain_hash: genesis_hash(),
  63                 short_channel_id,
  64                 node_id_1,
  65                 node_id_2,
  66                 bitcoin_key_1: node_id_1,
  67                 bitcoin_key_2: node_id_2,
  68                 excess_data: vec![],
  69         };
  70
  71         let msg_hash = bitcoin::secp256k1::Message::from_slice(&Sha256dHash::hash(&announcement.encode()[..])[..]).unwrap();
  72         let node_signature_1 = secp_context.sign_ecdsa(&msg_hash, &random_private_key_1);
  73         let node_signature_2 = secp_context.sign_ecdsa(&msg_hash, &random_private_key_2);
  74
  75         ChannelAnnouncement {
  76                 node_signature_1,
  77                 node_signature_2,
  78                 bitcoin_signature_1: node_signature_1,
  79                 bitcoin_signature_2: node_signature_2,
  80                 contents: announcement,
  81         }
  82 }
  83
  84 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 {
  85         let flag_mask = if direction { 1 } else { 0 };
  86         ChannelUpdate {
  87                 signature: blank_signature(),
  88                 contents: UnsignedChannelUpdate {
  89                         chain_hash: genesis_hash(),
  90                         short_channel_id: scid,
  91                         timestamp,
  92                         flags: 0 | flag_mask,
  93                         cltv_expiry_delta: expiry_delta,
  94                         htlc_minimum_msat: min_msat,
  95                         htlc_maximum_msat: max_msat,
  96                         fee_base_msat: base_msat,
  97                         fee_proportional_millionths: fee_rate,
  98                         excess_data: vec![],
  99                 },
 100         }
 101 }
 102
 103 struct SchemaSanitizer {}
 104
 105 impl SchemaSanitizer {
 106         fn new() -> Self {
 107                 IS_TEST_SCHEMA_CLEAN.with(|cleanliness_reference| {
 108                         let mut is_clean_option = cleanliness_reference.borrow_mut();
 109                         assert!(is_clean_option.is_none());
 110                         *is_clean_option = Some(false);
 111                 });
 112
 113                 DB_TEST_SCHEMA.with(|suffix_reference| {
 114                         let mut suffix_option = suffix_reference.borrow_mut();
 115                         let current_time = SystemTime::now();
 116                         let unix_time = current_time.duration_since(UNIX_EPOCH).expect("Time went backwards");
 117                         let timestamp_seconds = unix_time.as_secs();
 118                         let timestamp_nanos = unix_time.as_nanos();
 119                         // sometimes Rust thinks two tests start at the same nanosecond, causing a schema conflict
 120                         let thread_id = thread::current().id();
 121                         let preimage = format!("{:?}-{}", thread_id, timestamp_nanos);
 122                         println!("test schema preimage: {}", preimage);
 123                         let suffix = Sha256dHash::hash(preimage.as_bytes()).into_inner().to_hex();
 124                         // the schema must start with a letter
 125                         let schema = format!("test_{}_{}", timestamp_seconds, suffix);
 126                         *suffix_option = Some(schema);
 127                 });
 128
 129                 return Self {};
 130         }
 131 }
 132
 133 impl Drop for SchemaSanitizer {
 134         fn drop(&mut self) {
 135                 IS_TEST_SCHEMA_CLEAN.with(|cleanliness_reference| {
 136                         let is_clean_option = cleanliness_reference.borrow();
 137                         if let Some(is_clean) = *is_clean_option {
 138                                 assert_eq!(is_clean, true);
 139                         }
 140                 });
 141         }
 142 }
 143
 144 struct CacheSanitizer {}
 145
 146 impl CacheSanitizer {
 147         /// The CacheSanitizer instantiation requires that there be a schema sanitizer
 148         fn new(_: &SchemaSanitizer) -> Self {
 149                 Self {}
 150         }
 151
 152         fn cache_path(&self) -> String {
 153                 format!("./res/{}/", db_test_schema())
 154         }
 155 }
 156
 157 impl Drop for CacheSanitizer {
 158         fn drop(&mut self) {
 159                 let cache_path = self.cache_path();
 160                 fs::remove_dir_all(cache_path).unwrap();
 161         }
 162 }
 163
 164
 165 async fn clean_test_db() {
 166         let client = crate::connect_to_db().await;
 167         let schema = db_test_schema();
 168         client.execute(&format!("DROP SCHEMA IF EXISTS {} CASCADE", schema), &[]).await.unwrap();
 169         IS_TEST_SCHEMA_CLEAN.with(|cleanliness_reference| {
 170                 let mut is_clean_option = cleanliness_reference.borrow_mut();
 171                 *is_clean_option = Some(true);
 172         });
 173 }
 174
 175 #[tokio::test]
 176 async fn test_trivial_setup() {
 177         let _sanitizer = SchemaSanitizer::new();
 178         let logger = Arc::new(TestLogger::new());
 179         let network_graph = NetworkGraph::new(Network::Bitcoin, logger.clone());
 180         let network_graph_arc = Arc::new(network_graph);
 181         let (mut persister, receiver) = GossipPersister::new(network_graph_arc.clone(), logger.clone());
 182
 183         let short_channel_id = 1;
 184         let timestamp = current_time() - 10;
 185         println!("timestamp: {}", timestamp);
 186
 187         { // seed the db
 188                 let announcement = generate_announcement(short_channel_id);
 189                 let update_1 = generate_update(short_channel_id, false, timestamp, 0, 0, 0, 5, 0);
 190                 let update_2 = generate_update(short_channel_id, true, timestamp, 0, 0, 0, 10, 0);
 191
 192                 network_graph_arc.update_channel_from_announcement_no_lookup(&announcement).unwrap();
 193                 network_graph_arc.update_channel_unsigned(&update_1.contents).unwrap();
 194                 network_graph_arc.update_channel_unsigned(&update_2.contents).unwrap();
 195
 196                 receiver.send(GossipMessage::ChannelAnnouncement(announcement, None)).await.unwrap();
 197                 receiver.send(GossipMessage::ChannelUpdate(update_1, None)).await.unwrap();
 198                 receiver.send(GossipMessage::ChannelUpdate(update_2, None)).await.unwrap();
 199                 drop(receiver);
 200                 persister.persist_gossip().await;
 201         }
 202
 203         let serialization = serialize_delta(network_graph_arc.clone(), 0, logger.clone()).await;
 204         logger.assert_log_contains("rapid_gossip_sync_server", "announcement channel count: 1", 1);
 205         clean_test_db().await;
 206
 207         let channel_count = network_graph_arc.read_only().channels().len();
 208
 209         assert_eq!(channel_count, 1);
 210         assert_eq!(serialization.message_count, 3);
 211         assert_eq!(serialization.announcement_count, 1);
 212         assert_eq!(serialization.update_count, 2);
 213
 214         let client_graph = NetworkGraph::new(Network::Bitcoin, logger.clone());
 215         let client_graph_arc = Arc::new(client_graph);
 216         let rgs = RapidGossipSync::new(client_graph_arc.clone(), logger.clone());
 217         let update_result = rgs.update_network_graph(&serialization.data).unwrap();
 218         println!("update result: {}", update_result);
 219         // the update result must be a multiple of our snapshot granularity
 220         assert_eq!(update_result % config::snapshot_generation_interval(), 0);
 221         assert!(update_result < timestamp);
 222
 223         let timestamp_delta = timestamp - update_result;
 224         println!("timestamp delta: {}", timestamp_delta);
 225         assert!(timestamp_delta < config::snapshot_generation_interval());
 226
 227         let readonly_graph = client_graph_arc.read_only();
 228         let channels = readonly_graph.channels();
 229         let client_channel_count = channels.len();
 230         assert_eq!(client_channel_count, 1);
 231
 232         let first_channel = channels.get(&short_channel_id).unwrap();
 233         assert!(&first_channel.announcement_message.is_none());
 234         assert_eq!(first_channel.one_to_two.as_ref().unwrap().fees.base_msat, 5);
 235         assert_eq!(first_channel.two_to_one.as_ref().unwrap().fees.base_msat, 10);
 236         let last_update_seen_a = first_channel.one_to_two.as_ref().unwrap().last_update;
 237         let last_update_seen_b = first_channel.two_to_one.as_ref().unwrap().last_update;
 238         println!("last update a: {}", last_update_seen_a);
 239         println!("last update b: {}", last_update_seen_b);
 240         assert_eq!(last_update_seen_a, update_result - CLIENT_BACKDATE_INTERVAL);
 241         assert_eq!(last_update_seen_b, update_result - CLIENT_BACKDATE_INTERVAL);
 242 }
 243
 244 #[tokio::test]
 245 async fn test_full_snapshot_recency() {
 246         let _sanitizer = SchemaSanitizer::new();
 247         let logger = Arc::new(TestLogger::new());
 248         let network_graph = NetworkGraph::new(Network::Bitcoin, logger.clone());
 249         let network_graph_arc = Arc::new(network_graph);
 250
 251         let short_channel_id = 1;
 252         let timestamp = current_time();
 253         println!("timestamp: {}", timestamp);
 254
 255         { // seed the db
 256                 let (mut persister, receiver) = GossipPersister::new(network_graph_arc.clone(), logger.clone());
 257                 let announcement = generate_announcement(short_channel_id);
 258                 network_graph_arc.update_channel_from_announcement_no_lookup(&announcement).unwrap();
 259                 receiver.send(GossipMessage::ChannelAnnouncement(announcement, None)).await.unwrap();
 260
 261                 { // direction false
 262                         { // first update
 263                                 let update = generate_update(short_channel_id, false, timestamp - 1, 0, 0, 0, 0, 38);
 264                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 265                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 266                         }
 267                         { // second update
 268                                 let update = generate_update(short_channel_id, false, timestamp, 0, 0, 0, 0, 39);
 269                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 270                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 271                         }
 272                 }
 273                 { // direction true
 274                         { // first and only update
 275                                 let update = generate_update(short_channel_id, true, timestamp, 0, 0, 0, 0, 10);
 276                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 277                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 278                         }
 279                 }
 280
 281                 drop(receiver);
 282                 persister.persist_gossip().await;
 283         }
 284
 285         let client_graph = NetworkGraph::new(Network::Bitcoin, logger.clone());
 286         let client_graph_arc = Arc::new(client_graph);
 287
 288         { // sync after initial seed
 289                 let serialization = serialize_delta(network_graph_arc.clone(), 0, logger.clone()).await;
 290                 logger.assert_log_contains("rapid_gossip_sync_server", "announcement channel count: 1", 1);
 291
 292                 let channel_count = network_graph_arc.read_only().channels().len();
 293
 294                 assert_eq!(channel_count, 1);
 295                 assert_eq!(serialization.message_count, 3);
 296                 assert_eq!(serialization.announcement_count, 1);
 297                 assert_eq!(serialization.update_count, 2);
 298
 299                 let rgs = RapidGossipSync::new(client_graph_arc.clone(), logger.clone());
 300                 let update_result = rgs.update_network_graph(&serialization.data).unwrap();
 301                 // the update result must be a multiple of our snapshot granularity
 302                 assert_eq!(update_result % config::snapshot_generation_interval(), 0);
 303                 assert!(update_result < timestamp);
 304
 305                 let readonly_graph = client_graph_arc.read_only();
 306                 let channels = readonly_graph.channels();
 307                 let client_channel_count = channels.len();
 308                 assert_eq!(client_channel_count, 1);
 309
 310                 let first_channel = channels.get(&short_channel_id).unwrap();
 311                 assert!(&first_channel.announcement_message.is_none());
 312                 // ensure the update in one direction shows the latest fee
 313                 assert_eq!(first_channel.one_to_two.as_ref().unwrap().fees.proportional_millionths, 39);
 314                 assert_eq!(first_channel.two_to_one.as_ref().unwrap().fees.proportional_millionths, 10);
 315         }
 316
 317         clean_test_db().await;
 318 }
 319
 320 #[tokio::test]
 321 async fn test_full_snapshot_recency_with_wrong_seen_order() {
 322         let _sanitizer = SchemaSanitizer::new();
 323         let logger = Arc::new(TestLogger::new());
 324         let network_graph = NetworkGraph::new(Network::Bitcoin, logger.clone());
 325         let network_graph_arc = Arc::new(network_graph);
 326
 327         let short_channel_id = 1;
 328         let timestamp = current_time();
 329         println!("timestamp: {}", timestamp);
 330
 331         { // seed the db
 332                 let (mut persister, receiver) = GossipPersister::new(network_graph_arc.clone(), logger.clone());
 333                 let announcement = generate_announcement(short_channel_id);
 334                 network_graph_arc.update_channel_from_announcement_no_lookup(&announcement).unwrap();
 335                 receiver.send(GossipMessage::ChannelAnnouncement(announcement, None)).await.unwrap();
 336
 337                 { // direction false
 338                         { // first update, seen latest
 339                                 let update = generate_update(short_channel_id, false, timestamp - 1, 0, 0, 0, 0, 38);
 340                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 341                                 receiver.send(GossipMessage::ChannelUpdate(update, Some(timestamp))).await.unwrap();
 342                         }
 343                         { // second update, seen first
 344                                 let update = generate_update(short_channel_id, false, timestamp, 0, 0, 0, 0, 39);
 345                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 346                                 receiver.send(GossipMessage::ChannelUpdate(update, Some(timestamp - 1))).await.unwrap();
 347                         }
 348                 }
 349                 { // direction true
 350                         { // first and only update
 351                                 let update = generate_update(short_channel_id, true, timestamp, 0, 0, 0, 0, 10);
 352                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 353                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 354                         }
 355                 }
 356
 357                 drop(receiver);
 358                 persister.persist_gossip().await;
 359         }
 360
 361         let client_graph = NetworkGraph::new(Network::Bitcoin, logger.clone());
 362         let client_graph_arc = Arc::new(client_graph);
 363
 364         { // sync after initial seed
 365                 let serialization = serialize_delta(network_graph_arc.clone(), 0, logger.clone()).await;
 366                 logger.assert_log_contains("rapid_gossip_sync_server", "announcement channel count: 1", 1);
 367
 368                 let channel_count = network_graph_arc.read_only().channels().len();
 369
 370                 assert_eq!(channel_count, 1);
 371                 assert_eq!(serialization.message_count, 3);
 372                 assert_eq!(serialization.announcement_count, 1);
 373                 assert_eq!(serialization.update_count, 2);
 374
 375                 let rgs = RapidGossipSync::new(client_graph_arc.clone(), logger.clone());
 376                 let update_result = rgs.update_network_graph(&serialization.data).unwrap();
 377                 // the update result must be a multiple of our snapshot granularity
 378                 assert_eq!(update_result % config::snapshot_generation_interval(), 0);
 379                 assert!(update_result < timestamp);
 380
 381                 let readonly_graph = client_graph_arc.read_only();
 382                 let channels = readonly_graph.channels();
 383                 let client_channel_count = channels.len();
 384                 assert_eq!(client_channel_count, 1);
 385
 386                 let first_channel = channels.get(&short_channel_id).unwrap();
 387                 assert!(&first_channel.announcement_message.is_none());
 388                 // ensure the update in one direction shows the latest fee
 389                 assert_eq!(first_channel.one_to_two.as_ref().unwrap().fees.proportional_millionths, 39);
 390                 assert_eq!(first_channel.two_to_one.as_ref().unwrap().fees.proportional_millionths, 10);
 391         }
 392
 393         clean_test_db().await;
 394 }
 395
 396 #[tokio::test]
 397 async fn test_full_snapshot_recency_with_wrong_propagation_order() {
 398         let _sanitizer = SchemaSanitizer::new();
 399         let logger = Arc::new(TestLogger::new());
 400         let network_graph = NetworkGraph::new(Network::Bitcoin, logger.clone());
 401         let network_graph_arc = Arc::new(network_graph);
 402
 403         let short_channel_id = 1;
 404         let timestamp = current_time();
 405         println!("timestamp: {}", timestamp);
 406
 407         { // seed the db
 408                 let (mut persister, receiver) = GossipPersister::new(network_graph_arc.clone(), logger.clone());
 409                 let announcement = generate_announcement(short_channel_id);
 410                 network_graph_arc.update_channel_from_announcement_no_lookup(&announcement).unwrap();
 411                 receiver.send(GossipMessage::ChannelAnnouncement(announcement, None)).await.unwrap();
 412
 413                 { // direction false
 414                         // apply updates in their timestamp order
 415                         let update_1 = generate_update(short_channel_id, false, timestamp - 1, 0, 0, 0, 0, 38);
 416                         let update_2 = generate_update(short_channel_id, false, timestamp, 0, 0, 0, 0, 39);
 417                         network_graph_arc.update_channel_unsigned(&update_1.contents).unwrap();
 418                         network_graph_arc.update_channel_unsigned(&update_2.contents).unwrap();
 419
 420                         // propagate updates in their seen order
 421                         receiver.send(GossipMessage::ChannelUpdate(update_2, Some(timestamp - 1))).await.unwrap();
 422                         receiver.send(GossipMessage::ChannelUpdate(update_1, Some(timestamp))).await.unwrap();
 423                 }
 424                 { // direction true
 425                         { // first and only update
 426                                 let update = generate_update(short_channel_id, true, timestamp, 0, 0, 0, 0, 10);
 427                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 428                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 429                         }
 430                 }
 431
 432                 drop(receiver);
 433                 persister.persist_gossip().await;
 434         }
 435
 436         let client_graph = NetworkGraph::new(Network::Bitcoin, logger.clone());
 437         let client_graph_arc = Arc::new(client_graph);
 438
 439         { // sync after initial seed
 440                 let serialization = serialize_delta(network_graph_arc.clone(), 0, logger.clone()).await;
 441                 logger.assert_log_contains("rapid_gossip_sync_server", "announcement channel count: 1", 1);
 442
 443                 let channel_count = network_graph_arc.read_only().channels().len();
 444
 445                 assert_eq!(channel_count, 1);
 446                 assert_eq!(serialization.message_count, 3);
 447                 assert_eq!(serialization.announcement_count, 1);
 448                 assert_eq!(serialization.update_count, 2);
 449
 450                 let rgs = RapidGossipSync::new(client_graph_arc.clone(), logger.clone());
 451                 let update_result = rgs.update_network_graph(&serialization.data).unwrap();
 452                 // the update result must be a multiple of our snapshot granularity
 453                 assert_eq!(update_result % config::snapshot_generation_interval(), 0);
 454                 assert!(update_result < timestamp);
 455
 456                 let readonly_graph = client_graph_arc.read_only();
 457                 let channels = readonly_graph.channels();
 458                 let client_channel_count = channels.len();
 459                 assert_eq!(client_channel_count, 1);
 460
 461                 let first_channel = channels.get(&short_channel_id).unwrap();
 462                 assert!(&first_channel.announcement_message.is_none());
 463                 // ensure the update in one direction shows the latest fee
 464                 assert_eq!(first_channel.one_to_two.as_ref().unwrap().fees.proportional_millionths, 39);
 465                 assert_eq!(first_channel.two_to_one.as_ref().unwrap().fees.proportional_millionths, 10);
 466         }
 467
 468         clean_test_db().await;
 469 }
 470
 471 #[tokio::test]
 472 async fn test_full_snapshot_mutiny_scenario() {
 473         let _sanitizer = SchemaSanitizer::new();
 474         let logger = Arc::new(TestLogger::new());
 475         let network_graph = NetworkGraph::new(Network::Bitcoin, logger.clone());
 476         let network_graph_arc = Arc::new(network_graph);
 477
 478         let short_channel_id = 873706024403271681;
 479         let timestamp = current_time();
 480         // let oldest_simulation_timestamp = 1693300588;
 481         let latest_simulation_timestamp = 1695909301;
 482         let timestamp_offset = timestamp - latest_simulation_timestamp;
 483         println!("timestamp: {}", timestamp);
 484
 485         { // seed the db
 486                 let (mut persister, receiver) = GossipPersister::new(network_graph_arc.clone(), logger.clone());
 487                 let announcement = generate_announcement(short_channel_id);
 488                 network_graph_arc.update_channel_from_announcement_no_lookup(&announcement).unwrap();
 489                 receiver.send(GossipMessage::ChannelAnnouncement(announcement, None)).await.unwrap();
 490
 491                 { // direction false
 492                         {
 493                                 let update = generate_update(short_channel_id, false, 1693507369 + timestamp_offset, 0, 0, 0, 0, 38);
 494                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 495                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 496                         }
 497                         {
 498                                 let update = generate_update(short_channel_id, false, 1693680390 + timestamp_offset, 0, 0, 0, 0, 38);
 499                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 500                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 501                         }
 502                         {
 503                                 let update = generate_update(short_channel_id, false, 1693749109 + timestamp_offset, 0, 0, 0, 0, 200);
 504                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 505                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 506                         }
 507                         {
 508                                 let update = generate_update(short_channel_id, false, 1693925190 + timestamp_offset, 0, 0, 0, 0, 200);
 509                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 510                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 511                         }
 512                         {
 513                                 let update = generate_update(short_channel_id, false, 1694008323 + timestamp_offset, 0, 0, 0, 0, 209);
 514                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 515                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 516                         }
 517                         {
 518                                 let update = generate_update(short_channel_id, false, 1694219924 + timestamp_offset, 0, 0, 0, 0, 209);
 519                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 520                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 521                         }
 522                         {
 523                                 let update = generate_update(short_channel_id, false, 1694267536 + timestamp_offset, 0, 0, 0, 0, 210);
 524                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 525                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 526                         }
 527                         {
 528                                 let update = generate_update(short_channel_id, false, 1694458808 + timestamp_offset, 0, 0, 0, 0, 210);
 529                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 530                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 531                         }
 532                         {
 533                                 let update = generate_update(short_channel_id, false, 1694526734 + timestamp_offset, 0, 0, 0, 0, 200);
 534                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 535                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 536                         }
 537                         {
 538                                 let update = generate_update(short_channel_id, false, 1694794765 + timestamp_offset, 0, 0, 0, 0, 200);
 539                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 540                                 receiver.send(GossipMessage::ChannelUpdate(update, Some(1695909301 + 2 * config::SYMLINK_GRANULARITY_INTERVAL + timestamp_offset))).await.unwrap();
 541                         }
 542                         {
 543                                 let update = generate_update(short_channel_id, false, 1695909301 + timestamp_offset, 0, 0, 0, 0, 130);
 544                                 // network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 545                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 546                         }
 547                 }
 548                 { // direction true
 549                         {
 550                                 let update = generate_update(short_channel_id, true, 1693300588 + timestamp_offset, 0, 0, 0, 0, 10);
 551                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 552                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 553                         }
 554                         {
 555                                 let update = generate_update(short_channel_id, true, 1695003621 + timestamp_offset, 0, 0, 0, 0, 10);
 556                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 557                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 558                         }
 559                 }
 560
 561                 drop(receiver);
 562                 persister.persist_gossip().await;
 563         }
 564
 565         let client_graph = NetworkGraph::new(Network::Bitcoin, logger.clone());
 566         let client_graph_arc = Arc::new(client_graph);
 567
 568         { // sync after initial seed
 569                 let serialization = serialize_delta(network_graph_arc.clone(), 0, logger.clone()).await;
 570                 logger.assert_log_contains("rapid_gossip_sync_server", "announcement channel count: 1", 1);
 571
 572                 let channel_count = network_graph_arc.read_only().channels().len();
 573
 574                 assert_eq!(channel_count, 1);
 575                 assert_eq!(serialization.message_count, 3);
 576                 assert_eq!(serialization.announcement_count, 1);
 577                 assert_eq!(serialization.update_count, 2);
 578
 579                 let rgs = RapidGossipSync::new(client_graph_arc.clone(), logger.clone());
 580                 let update_result = rgs.update_network_graph(&serialization.data).unwrap();
 581                 println!("update result: {}", update_result);
 582                 // the update result must be a multiple of our snapshot granularity
 583                 assert_eq!(update_result % config::snapshot_generation_interval(), 0);
 584                 assert!(update_result < timestamp);
 585
 586                 let timestamp_delta = timestamp - update_result;
 587                 println!("timestamp delta: {}", timestamp_delta);
 588                 assert!(timestamp_delta < config::snapshot_generation_interval());
 589
 590                 let readonly_graph = client_graph_arc.read_only();
 591                 let channels = readonly_graph.channels();
 592                 let client_channel_count = channels.len();
 593                 assert_eq!(client_channel_count, 1);
 594
 595                 let first_channel = channels.get(&short_channel_id).unwrap();
 596                 assert!(&first_channel.announcement_message.is_none());
 597                 assert_eq!(first_channel.one_to_two.as_ref().unwrap().fees.proportional_millionths, 130);
 598                 assert_eq!(first_channel.two_to_one.as_ref().unwrap().fees.proportional_millionths, 10);
 599         }
 600
 601         clean_test_db().await;
 602 }
 603
 604 #[tokio::test]
 605 async fn test_full_snapshot_interlaced_channel_timestamps() {
 606         let _sanitizer = SchemaSanitizer::new();
 607         let logger = Arc::new(TestLogger::new());
 608         let network_graph = NetworkGraph::new(Network::Bitcoin, logger.clone());
 609         let network_graph_arc = Arc::new(network_graph);
 610
 611         let main_channel_id = 1;
 612         let timestamp = current_time();
 613         println!("timestamp: {}", timestamp);
 614
 615         { // seed the db
 616                 let (mut persister, receiver) = GossipPersister::new(network_graph_arc.clone(), logger.clone());
 617                 let secondary_channel_id = main_channel_id + 1;
 618
 619                 { // main channel
 620                         let announcement = generate_announcement(main_channel_id);
 621                         network_graph_arc.update_channel_from_announcement_no_lookup(&announcement).unwrap();
 622                         receiver.send(GossipMessage::ChannelAnnouncement(announcement, None)).await.unwrap();
 623                 }
 624
 625                 { // secondary channel
 626                         let announcement = generate_announcement(secondary_channel_id);
 627                         network_graph_arc.update_channel_from_announcement_no_lookup(&announcement).unwrap();
 628                         receiver.send(GossipMessage::ChannelAnnouncement(announcement, None)).await.unwrap();
 629                 }
 630
 631                 { // main channel
 632                         { // direction false
 633                                 let update = generate_update(main_channel_id, false, timestamp - 2, 0, 0, 0, 0, 10);
 634                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 635                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 636                         }
 637                         { // direction true
 638                                 let update = generate_update(main_channel_id, true, timestamp - 2, 0, 0, 0, 0, 5);
 639                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 640                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 641                         }
 642                 }
 643
 644                 { // in-between channel
 645                         { // direction false
 646                                 let update = generate_update(secondary_channel_id, false, timestamp - 1, 0, 0, 0, 0, 42);
 647                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 648                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 649                         }
 650                         { // direction true
 651                                 let update = generate_update(secondary_channel_id, true, timestamp - 1, 0, 0, 0, 0, 42);
 652                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 653                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 654                         }
 655                 }
 656
 657                 { // main channel
 658                         { // direction false
 659                                 let update = generate_update(main_channel_id, false, timestamp, 0, 0, 0, 0, 11);
 660                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 661                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 662                         }
 663                         { // direction true
 664                                 let update = generate_update(main_channel_id, true, timestamp, 0, 0, 0, 0, 6);
 665                                 network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 666                                 receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 667                         }
 668                 }
 669
 670                 drop(receiver);
 671                 persister.persist_gossip().await;
 672         }
 673
 674         let client_graph = NetworkGraph::new(Network::Bitcoin, logger.clone());
 675         let client_graph_arc = Arc::new(client_graph);
 676
 677         { // sync after initial seed
 678                 let serialization = serialize_delta(network_graph_arc.clone(), 0, logger.clone()).await;
 679                 logger.assert_log_contains("rapid_gossip_sync_server", "announcement channel count: 2", 1);
 680
 681                 let channel_count = network_graph_arc.read_only().channels().len();
 682
 683                 assert_eq!(channel_count, 2);
 684                 assert_eq!(serialization.message_count, 6);
 685                 assert_eq!(serialization.announcement_count, 2);
 686                 assert_eq!(serialization.update_count, 4);
 687
 688                 let rgs = RapidGossipSync::new(client_graph_arc.clone(), logger.clone());
 689                 let update_result = rgs.update_network_graph(&serialization.data).unwrap();
 690                 // the update result must be a multiple of our snapshot granularity
 691                 assert_eq!(update_result % config::snapshot_generation_interval(), 0);
 692                 assert!(update_result < timestamp);
 693
 694                 let readonly_graph = client_graph_arc.read_only();
 695                 let channels = readonly_graph.channels();
 696                 let client_channel_count = channels.len();
 697                 assert_eq!(client_channel_count, 2);
 698
 699                 let first_channel = channels.get(&main_channel_id).unwrap();
 700                 assert!(&first_channel.announcement_message.is_none());
 701                 // ensure the update in one direction shows the latest fee
 702                 assert_eq!(first_channel.one_to_two.as_ref().unwrap().fees.proportional_millionths, 11);
 703                 assert_eq!(first_channel.two_to_one.as_ref().unwrap().fees.proportional_millionths, 6);
 704         }
 705
 706         clean_test_db().await;
 707 }
 708
 709 #[tokio::test]
 710 async fn test_full_snapshot_persistence() {
 711         let schema_sanitizer = SchemaSanitizer::new();
 712         let logger = Arc::new(TestLogger::new());
 713         let network_graph = NetworkGraph::new(Network::Bitcoin, logger.clone());
 714         let network_graph_arc = Arc::new(network_graph);
 715         let snapshotter = Snapshotter::new(network_graph_arc.clone(), logger.clone());
 716         let cache_sanitizer = CacheSanitizer::new(&schema_sanitizer);
 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_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 true
 729                         let update = generate_update(short_channel_id, true, timestamp, 0, 0, 0, 0, 10);
 730                         network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 731                         receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 732                 }
 733
 734                 { // direction false
 735                         let update = generate_update(short_channel_id, false, timestamp - 1, 0, 0, 0, 0, 38);
 736                         network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 737                         receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 738                 }
 739
 740
 741                 drop(receiver);
 742                 persister.persist_gossip().await;
 743         }
 744
 745         let cache_path = cache_sanitizer.cache_path();
 746         let symlink_path = format!("{}/symlinks/0.bin", cache_path);
 747
 748         // generate snapshots
 749         {
 750                 snapshotter.generate_snapshots(20, 5, &[5, u64::MAX], &cache_path, Some(10)).await;
 751
 752                 let symlinked_data = fs::read(&symlink_path).unwrap();
 753                 let client_graph = NetworkGraph::new(Network::Bitcoin, logger.clone());
 754                 let client_graph_arc = Arc::new(client_graph);
 755
 756                 let rgs = RapidGossipSync::new(client_graph_arc.clone(), logger.clone());
 757                 let update_result = rgs.update_network_graph(&symlinked_data).unwrap();
 758                 // the update result must be a multiple of our snapshot granularity
 759                 assert_eq!(update_result % config::snapshot_generation_interval(), 0);
 760
 761                 let readonly_graph = client_graph_arc.read_only();
 762                 let channels = readonly_graph.channels();
 763                 let client_channel_count = channels.len();
 764                 assert_eq!(client_channel_count, 1);
 765
 766                 let first_channel = channels.get(&short_channel_id).unwrap();
 767                 assert!(&first_channel.announcement_message.is_none());
 768                 // ensure the update in one direction shows the latest fee
 769                 assert_eq!(first_channel.one_to_two.as_ref().unwrap().fees.proportional_millionths, 38);
 770                 assert_eq!(first_channel.two_to_one.as_ref().unwrap().fees.proportional_millionths, 10);
 771         }
 772
 773         { // update the db
 774                 let (mut persister, receiver) = GossipPersister::new(network_graph_arc.clone(), logger.clone());
 775
 776                 { // second update
 777                         let update = generate_update(short_channel_id, false, timestamp + 30, 0, 0, 0, 0, 39);
 778                         network_graph_arc.update_channel_unsigned(&update.contents).unwrap();
 779                         receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap();
 780                 }
 781
 782                 drop(receiver);
 783                 persister.persist_gossip().await;
 784         }
 785
 786         // regenerate snapshots
 787         {
 788                 snapshotter.generate_snapshots(20, 5, &[5, u64::MAX], &cache_path, Some(10)).await;
 789
 790                 let symlinked_data = fs::read(&symlink_path).unwrap();
 791                 let client_graph = NetworkGraph::new(Network::Bitcoin, logger.clone());
 792                 let client_graph_arc = Arc::new(client_graph);
 793
 794                 let rgs = RapidGossipSync::new(client_graph_arc.clone(), logger.clone());
 795                 let update_result = rgs.update_network_graph(&symlinked_data).unwrap();
 796                 // the update result must be a multiple of our snapshot granularity
 797                 assert_eq!(update_result % config::snapshot_generation_interval(), 0);
 798
 799                 let readonly_graph = client_graph_arc.read_only();
 800                 let channels = readonly_graph.channels();
 801                 let client_channel_count = channels.len();
 802                 assert_eq!(client_channel_count, 1);
 803
 804                 let first_channel = channels.get(&short_channel_id).unwrap();
 805                 assert!(&first_channel.announcement_message.is_none());
 806                 // ensure the update in one direction shows the latest fee
 807                 assert_eq!(first_channel.one_to_two.as_ref().unwrap().fees.proportional_millionths, 39);
 808                 assert_eq!(first_channel.two_to_one.as_ref().unwrap().fees.proportional_millionths, 10);
 809         }
 810
 811         // clean up afterwards
 812         clean_test_db().await;
 813 }