X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=src%2Ftests%2Fmod.rs;h=578fa9f7cbe07fc99307d94816b35ba01a3c2758;hb=f46a13944933c9c6c42fbeba66473d6e548d7cda;hp=f26cea8b44f180f3756b632e827c393fe857a57d;hpb=f525858b51ca7c8ad9b855efd92e5e04efa53ed1;p=rapid-gossip-sync-server diff --git a/src/tests/mod.rs b/src/tests/mod.rs index f26cea8..578fa9f 100644 --- a/src/tests/mod.rs +++ b/src/tests/mod.rs @@ -2,20 +2,23 @@ use std::cell::RefCell; use std::sync::Arc; +use std::{fs, thread}; use std::time::{SystemTime, UNIX_EPOCH}; -use bitcoin::{BlockHash, Network}; +use bitcoin::blockdata::constants::ChainHash; +use bitcoin::Network; use bitcoin::secp256k1::ecdsa::Signature; use bitcoin::secp256k1::{Secp256k1, SecretKey}; use bitcoin::hashes::Hash; -use bitcoin::hashes::hex::ToHex; use bitcoin::hashes::sha256d::Hash as Sha256dHash; -use lightning::ln::features::ChannelFeatures; -use lightning::ln::msgs::{ChannelAnnouncement, ChannelUpdate, UnsignedChannelAnnouncement, UnsignedChannelUpdate}; -use lightning::routing::gossip::{NetworkGraph, NodeId}; +use hex_conservative::DisplayHex; +use lightning::ln::features::{ChannelFeatures, NodeFeatures}; +use lightning::ln::msgs::{ChannelAnnouncement, ChannelUpdate, NodeAnnouncement, SocketAddress, UnsignedChannelAnnouncement, UnsignedChannelUpdate, UnsignedNodeAnnouncement}; +use lightning::routing::gossip::{NetworkGraph, NodeAlias, NodeId}; use lightning::util::ser::Writeable; use lightning_rapid_gossip_sync::RapidGossipSync; use crate::{config, serialize_delta}; use crate::persistence::GossipPersister; +use crate::snapshot::Snapshotter; use crate::types::{GossipMessage, tests::TestLogger}; const CLIENT_BACKDATE_INTERVAL: u32 = 3600 * 24 * 7; // client backdates RGS by a week @@ -29,8 +32,8 @@ fn blank_signature() -> Signature { Signature::from_compact(&[0u8; 64]).unwrap() } -fn genesis_hash() -> BlockHash { - bitcoin::blockdata::constants::genesis_block(Network::Bitcoin).block_hash() +fn genesis_hash() -> ChainHash { + ChainHash::using_genesis_block(Network::Bitcoin) } fn current_time() -> u32 { @@ -39,12 +42,40 @@ fn current_time() -> u32 { pub(crate) fn db_test_schema() -> String { DB_TEST_SCHEMA.with(|suffix_reference| { - let mut suffix_option = suffix_reference.borrow(); + let suffix_option = suffix_reference.borrow(); suffix_option.as_ref().unwrap().clone() }) } -fn generate_announcement(short_channel_id: u64) -> ChannelAnnouncement { +fn generate_node_announcement() -> NodeAnnouncement { + let secp_context = Secp256k1::new(); + + let random_private_key = SecretKey::from_slice(&[1; 32]).unwrap(); + let random_public_key = random_private_key.public_key(&secp_context); + let node_id = NodeId::from_pubkey(&random_public_key); + + let announcement = UnsignedNodeAnnouncement { + features: NodeFeatures::empty(), + timestamp: 0, + node_id, + rgb: [0, 128, 255], + alias: NodeAlias([0; 32]), + addresses: vec![], + excess_data: vec![], + excess_address_data: vec![], + }; + + let msg_hash = bitcoin::secp256k1::Message::from_slice(&Sha256dHash::hash(&announcement.encode()[..])[..]).unwrap(); + let signature = secp_context.sign_ecdsa(&msg_hash, &random_private_key); + + NodeAnnouncement { + signature, + contents: announcement, + } +} + + +fn generate_channel_announcement(short_channel_id: u64) -> ChannelAnnouncement { let secp_context = Secp256k1::new(); let random_private_key_1 = SecretKey::from_slice(&[1; 32]).unwrap(); @@ -114,10 +145,13 @@ impl SchemaSanitizer { let unix_time = current_time.duration_since(UNIX_EPOCH).expect("Time went backwards"); let timestamp_seconds = unix_time.as_secs(); let timestamp_nanos = unix_time.as_nanos(); - let preimage = format!("{}", timestamp_nanos); - let suffix = Sha256dHash::hash(preimage.as_bytes()).into_inner().to_hex(); + // sometimes Rust thinks two tests start at the same nanosecond, causing a schema conflict + let thread_id = thread::current().id(); + let preimage = format!("{:?}-{}", thread_id, timestamp_nanos); + println!("test schema preimage: {}", preimage); + let suffix = Sha256dHash::hash(preimage.as_bytes()).encode(); // the schema must start with a letter - let schema = format!("test_{}_{}", timestamp_seconds, suffix); + let schema = format!("test_{}_{}", timestamp_seconds, suffix.as_hex()); *suffix_option = Some(schema); }); @@ -130,12 +164,35 @@ impl Drop for SchemaSanitizer { IS_TEST_SCHEMA_CLEAN.with(|cleanliness_reference| { let is_clean_option = cleanliness_reference.borrow(); if let Some(is_clean) = *is_clean_option { + if std::thread::panicking() { + return; + } assert_eq!(is_clean, true); } }); } } +struct CacheSanitizer {} + +impl CacheSanitizer { + /// The CacheSanitizer instantiation requires that there be a schema sanitizer + fn new(_: &SchemaSanitizer) -> Self { + Self {} + } + + fn cache_path(&self) -> String { + format!("./res/{}/", db_test_schema()) + } +} + +impl Drop for CacheSanitizer { + fn drop(&mut self) { + let cache_path = self.cache_path(); + fs::remove_dir_all(cache_path).unwrap(); + } +} + async fn clean_test_db() { let client = crate::connect_to_db().await; @@ -147,6 +204,22 @@ async fn clean_test_db() { }); } +#[tokio::test] +async fn test_persistence_runtime() { + let _sanitizer = SchemaSanitizer::new(); + let logger = Arc::new(TestLogger::new()); + let network_graph = NetworkGraph::new(Network::Bitcoin, logger.clone()); + let network_graph_arc = Arc::new(network_graph); + let (_persister, _receiver) = GossipPersister::new(network_graph_arc.clone(), logger.clone()); + + tokio::task::spawn_blocking(move || { + drop(_persister); + }).await.unwrap(); + + clean_test_db().await; +} + + #[tokio::test] async fn test_trivial_setup() { let _sanitizer = SchemaSanitizer::new(); @@ -160,7 +233,7 @@ async fn test_trivial_setup() { println!("timestamp: {}", timestamp); { // seed the db - let announcement = generate_announcement(short_channel_id); + let announcement = generate_channel_announcement(short_channel_id); let update_1 = generate_update(short_channel_id, false, timestamp, 0, 0, 0, 5, 0); let update_2 = generate_update(short_channel_id, true, timestamp, 0, 0, 0, 10, 0); @@ -175,7 +248,7 @@ async fn test_trivial_setup() { persister.persist_gossip().await; } - let serialization = serialize_delta(network_graph_arc.clone(), 0, logger.clone()).await; + let serialization = serialize_delta(network_graph_arc.clone(), 0, None, logger.clone()).await; logger.assert_log_contains("rapid_gossip_sync_server", "announcement channel count: 1", 1); clean_test_db().await; @@ -214,4 +287,859 @@ async fn test_trivial_setup() { println!("last update b: {}", last_update_seen_b); assert_eq!(last_update_seen_a, update_result - CLIENT_BACKDATE_INTERVAL); assert_eq!(last_update_seen_b, update_result - CLIENT_BACKDATE_INTERVAL); + + tokio::task::spawn_blocking(move || { + drop(persister); + }).await.unwrap(); +} + +#[tokio::test] +async fn test_node_announcement_persistence() { + let _sanitizer = SchemaSanitizer::new(); + let logger = Arc::new(TestLogger::new()); + let network_graph = NetworkGraph::new(Network::Bitcoin, logger.clone()); + let network_graph_arc = Arc::new(network_graph); + let (mut persister, receiver) = GossipPersister::new(network_graph_arc.clone(), logger.clone()); + + { // seed the db + let mut announcement = generate_node_announcement(); + receiver.send(GossipMessage::NodeAnnouncement(announcement.clone(), None)).await.unwrap(); + receiver.send(GossipMessage::NodeAnnouncement(announcement.clone(), Some(12345))).await.unwrap(); + + { + // modify announcement to contain a bunch of addresses + announcement.contents.addresses.push(SocketAddress::Hostname { + hostname: "google.com".to_string().try_into().unwrap(), + port: 443, + }); + announcement.contents.addresses.push(SocketAddress::TcpIpV4 { addr: [127, 0, 0, 1], port: 9635 }); + announcement.contents.addresses.push(SocketAddress::TcpIpV6 { addr: [1; 16], port: 1337 }); + announcement.contents.addresses.push(SocketAddress::OnionV2([1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12])); + announcement.contents.addresses.push(SocketAddress::OnionV3 { + ed25519_pubkey: [1; 32], + checksum: 2, + version: 3, + port: 4, + }); + } + receiver.send(GossipMessage::NodeAnnouncement(announcement, Some(12345))).await.unwrap(); + + drop(receiver); + persister.persist_gossip().await; + + tokio::task::spawn_blocking(move || { + drop(persister); + }).await.unwrap(); + } + clean_test_db().await; +} + + +/// If a channel has only seen updates in one direction, it should not be announced +#[tokio::test] +async fn test_unidirectional_intermediate_update_consideration() { + let _sanitizer = SchemaSanitizer::new(); + + let logger = Arc::new(TestLogger::new()); + let network_graph = NetworkGraph::new(Network::Bitcoin, logger.clone()); + let network_graph_arc = Arc::new(network_graph); + let (mut persister, receiver) = GossipPersister::new(network_graph_arc.clone(), logger.clone()); + + let short_channel_id = 1; + let timestamp = current_time() - 10; + println!("timestamp: {}", timestamp); + + { // seed the db + let announcement = generate_channel_announcement(short_channel_id); + let update_1 = generate_update(short_channel_id, false, timestamp, 0, 0, 0, 6, 0); + let update_2 = generate_update(short_channel_id, true, timestamp + 1, 0, 0, 0, 3, 0); + let update_3 = generate_update(short_channel_id, true, timestamp + 2, 0, 0, 0, 4, 0); + + network_graph_arc.update_channel_from_announcement_no_lookup(&announcement).unwrap(); + network_graph_arc.update_channel_unsigned(&update_1.contents).unwrap(); + network_graph_arc.update_channel_unsigned(&update_2.contents).unwrap(); + network_graph_arc.update_channel_unsigned(&update_3.contents).unwrap(); + + receiver.send(GossipMessage::ChannelAnnouncement(announcement, Some(timestamp))).await.unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update_1, None)).await.unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update_2, None)).await.unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update_3, None)).await.unwrap(); + drop(receiver); + persister.persist_gossip().await; + } + + let channel_count = network_graph_arc.read_only().channels().len(); + assert_eq!(channel_count, 1); + + let client_graph = NetworkGraph::new(Network::Bitcoin, logger.clone()); + let client_graph_arc = Arc::new(client_graph); + let rgs = RapidGossipSync::new(client_graph_arc.clone(), logger.clone()); + + let serialization = serialize_delta(network_graph_arc.clone(), timestamp + 1, None, logger.clone()).await; + + logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Fetched 1 update rows of the first update in a new direction", 1); + logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Processed 1 reference rows", 1); + logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Processed intermediate rows (2)", 1); + + assert_eq!(serialization.message_count, 3); + assert_eq!(serialization.announcement_count, 1); + assert_eq!(serialization.update_count, 2); + assert_eq!(serialization.update_count_full, 2); + assert_eq!(serialization.update_count_incremental, 0); + + let update_result = rgs.update_network_graph(&serialization.data).unwrap(); + println!("update result: {}", update_result); + // the update result must be a multiple of our snapshot granularity + + let readonly_graph = client_graph_arc.read_only(); + let channels = readonly_graph.channels(); + let client_channel_count = channels.len(); + assert_eq!(client_channel_count, 1); + + tokio::task::spawn_blocking(move || { + drop(persister); + }).await.unwrap(); + + clean_test_db().await; +} + +/// If a channel has only seen updates in one direction, it should not be announced +#[tokio::test] +async fn test_bidirectional_intermediate_update_consideration() { + let _sanitizer = SchemaSanitizer::new(); + + let logger = Arc::new(TestLogger::new()); + let network_graph = NetworkGraph::new(Network::Bitcoin, logger.clone()); + let network_graph_arc = Arc::new(network_graph); + let (mut persister, receiver) = GossipPersister::new(network_graph_arc.clone(), logger.clone()); + + let short_channel_id = 1; + let timestamp = current_time() - 10; + println!("timestamp: {}", timestamp); + + { // seed the db + let announcement = generate_channel_announcement(short_channel_id); + let update_1 = generate_update(short_channel_id, false, timestamp, 0, 0, 0, 5, 0); + let update_2 = generate_update(short_channel_id, false, timestamp + 1, 0, 0, 0, 4, 0); + let update_3 = generate_update(short_channel_id, false, timestamp + 2, 0, 0, 0, 3, 0); + let update_4 = generate_update(short_channel_id, true, timestamp, 0, 0, 0, 3, 0); + + network_graph_arc.update_channel_from_announcement_no_lookup(&announcement).unwrap(); + network_graph_arc.update_channel_unsigned(&update_1.contents).unwrap(); + network_graph_arc.update_channel_unsigned(&update_2.contents).unwrap(); + network_graph_arc.update_channel_unsigned(&update_3.contents).unwrap(); + network_graph_arc.update_channel_unsigned(&update_4.contents).unwrap(); + + receiver.send(GossipMessage::ChannelAnnouncement(announcement, Some(timestamp))).await.unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update_1, None)).await.unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update_2, None)).await.unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update_3, None)).await.unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update_4, None)).await.unwrap(); + drop(receiver); + persister.persist_gossip().await; + } + + let channel_count = network_graph_arc.read_only().channels().len(); + assert_eq!(channel_count, 1); + + let serialization = serialize_delta(network_graph_arc.clone(), timestamp + 1, None, logger.clone()).await; + + logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Fetched 0 update rows of the first update in a new direction", 1); + logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Processed 2 reference rows", 1); + logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Processed intermediate rows (2)", 1); + + assert_eq!(serialization.message_count, 1); + assert_eq!(serialization.announcement_count, 0); + assert_eq!(serialization.update_count, 1); + assert_eq!(serialization.update_count_full, 0); + assert_eq!(serialization.update_count_incremental, 1); + + tokio::task::spawn_blocking(move || { + drop(persister); + }).await.unwrap(); + + clean_test_db().await; +} + +#[tokio::test] +async fn test_channel_reminders() { + let _sanitizer = SchemaSanitizer::new(); + + let logger = Arc::new(TestLogger::new()); + let network_graph = NetworkGraph::new(Network::Bitcoin, logger.clone()); + let network_graph_arc = Arc::new(network_graph); + let (mut persister, receiver) = GossipPersister::new(network_graph_arc.clone(), logger.clone()); + + let timestamp = current_time(); + println!("timestamp: {}", timestamp); + let channel_reminder_delta = config::CHANNEL_REMINDER_AGE.as_secs() as u32; + + { // seed the db + { // unupdated channel + let short_channel_id = 1; + let announcement = generate_channel_announcement(short_channel_id); + let update_1 = generate_update(short_channel_id, false, timestamp - channel_reminder_delta - 1, 0, 0, 0, 5, 0); + let update_2 = generate_update(short_channel_id, true, timestamp - channel_reminder_delta - 1, 0, 0, 0, 3, 0); + + network_graph_arc.update_channel_from_announcement_no_lookup(&announcement).unwrap(); + network_graph_arc.update_channel_unsigned(&update_1.contents).unwrap(); + network_graph_arc.update_channel_unsigned(&update_2.contents).unwrap(); + + receiver.send(GossipMessage::ChannelAnnouncement(announcement, Some(timestamp - channel_reminder_delta - 1))).await.unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update_1, Some(timestamp - channel_reminder_delta - 1))).await.unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update_2, Some(timestamp - channel_reminder_delta - 1))).await.unwrap(); + } + { // unmodified but updated channel + let short_channel_id = 2; + let announcement = generate_channel_announcement(short_channel_id); + let update_1 = generate_update(short_channel_id, false, timestamp - channel_reminder_delta - 10, 0, 0, 0, 5, 0); + // in the false direction, we have one update that's different prior + let update_2 = generate_update(short_channel_id, false, timestamp - channel_reminder_delta - 5, 0, 1, 0, 5, 0); + let update_3 = generate_update(short_channel_id, false, timestamp - channel_reminder_delta - 1, 0, 0, 0, 5, 0); + let update_4 = generate_update(short_channel_id, true, timestamp - channel_reminder_delta - 1, 0, 0, 0, 3, 0); + let update_5 = generate_update(short_channel_id, false, timestamp - channel_reminder_delta + 10, 0, 0, 0, 5, 0); + let update_6 = generate_update(short_channel_id, true, timestamp - channel_reminder_delta + 10, 0, 0, 0, 3, 0); + let update_7 = generate_update(short_channel_id, false, timestamp - channel_reminder_delta + 20, 0, 0, 0, 5, 0); + let update_8 = generate_update(short_channel_id, true, timestamp - channel_reminder_delta + 20, 0, 0, 0, 3, 0); + + network_graph_arc.update_channel_from_announcement_no_lookup(&announcement).unwrap(); + network_graph_arc.update_channel_unsigned(&update_7.contents).unwrap(); + network_graph_arc.update_channel_unsigned(&update_8.contents).unwrap(); + + receiver.send(GossipMessage::ChannelAnnouncement(announcement, Some(timestamp - channel_reminder_delta - 1))).await.unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update_1, Some(timestamp - channel_reminder_delta - 10))).await.unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update_2, Some(timestamp - channel_reminder_delta - 5))).await.unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update_3, Some(timestamp - channel_reminder_delta - 1))).await.unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update_4, Some(timestamp - channel_reminder_delta - 1))).await.unwrap(); + + receiver.send(GossipMessage::ChannelUpdate(update_5, Some(timestamp - channel_reminder_delta + 10))).await.unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update_6, Some(timestamp - channel_reminder_delta + 10))).await.unwrap(); + + receiver.send(GossipMessage::ChannelUpdate(update_7, Some(timestamp - channel_reminder_delta + 20))).await.unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update_8, Some(timestamp - channel_reminder_delta + 20))).await.unwrap(); + } + drop(receiver); + persister.persist_gossip().await; + } + + let channel_count = network_graph_arc.read_only().channels().len(); + assert_eq!(channel_count, 2); + + let serialization = serialize_delta(network_graph_arc.clone(), timestamp - channel_reminder_delta + 15, None, logger.clone()).await; + + logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Fetched 0 update rows of the first update in a new direction", 1); + logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Fetched 4 update rows of the latest update in the less recently updated direction", 1); + logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Processed 2 reference rows", 1); + logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Processed intermediate rows (2)", 1); + + assert_eq!(serialization.message_count, 4); + assert_eq!(serialization.announcement_count, 0); + assert_eq!(serialization.update_count, 4); + assert_eq!(serialization.update_count_full, 0); + assert_eq!(serialization.update_count_incremental, 4); + + tokio::task::spawn_blocking(move || { + drop(persister); + }).await.unwrap(); + + clean_test_db().await; +} + +#[tokio::test] +async fn test_full_snapshot_recency() { + let _sanitizer = SchemaSanitizer::new(); + let logger = Arc::new(TestLogger::new()); + let network_graph = NetworkGraph::new(Network::Bitcoin, logger.clone()); + let network_graph_arc = Arc::new(network_graph); + + let short_channel_id = 1; + let timestamp = current_time(); + println!("timestamp: {}", timestamp); + + { // seed the db + let (mut persister, receiver) = GossipPersister::new(network_graph_arc.clone(), logger.clone()); + let announcement = generate_channel_announcement(short_channel_id); + network_graph_arc.update_channel_from_announcement_no_lookup(&announcement).unwrap(); + receiver.send(GossipMessage::ChannelAnnouncement(announcement, None)).await.unwrap(); + + { // direction false + { // first update + let update = generate_update(short_channel_id, false, timestamp - 1, 0, 0, 0, 0, 38); + network_graph_arc.update_channel_unsigned(&update.contents).unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap(); + } + { // second update + let update = generate_update(short_channel_id, false, timestamp, 0, 0, 0, 0, 39); + network_graph_arc.update_channel_unsigned(&update.contents).unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap(); + } + } + { // direction true + { // first and only update + let update = generate_update(short_channel_id, true, timestamp, 0, 0, 0, 0, 10); + network_graph_arc.update_channel_unsigned(&update.contents).unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap(); + } + } + + drop(receiver); + persister.persist_gossip().await; + + tokio::task::spawn_blocking(move || { + drop(persister); + }).await.unwrap(); + } + + let client_graph = NetworkGraph::new(Network::Bitcoin, logger.clone()); + let client_graph_arc = Arc::new(client_graph); + + { // sync after initial seed + let serialization = serialize_delta(network_graph_arc.clone(), 0, None, logger.clone()).await; + logger.assert_log_contains("rapid_gossip_sync_server", "announcement channel count: 1", 1); + + let channel_count = network_graph_arc.read_only().channels().len(); + + assert_eq!(channel_count, 1); + assert_eq!(serialization.message_count, 3); + assert_eq!(serialization.announcement_count, 1); + assert_eq!(serialization.update_count, 2); + + let rgs = RapidGossipSync::new(client_graph_arc.clone(), logger.clone()); + let update_result = rgs.update_network_graph(&serialization.data).unwrap(); + // the update result must be a multiple of our snapshot granularity + assert_eq!(update_result % config::snapshot_generation_interval(), 0); + assert!(update_result < timestamp); + + let readonly_graph = client_graph_arc.read_only(); + let channels = readonly_graph.channels(); + let client_channel_count = channels.len(); + assert_eq!(client_channel_count, 1); + + let first_channel = channels.get(&short_channel_id).unwrap(); + assert!(&first_channel.announcement_message.is_none()); + // ensure the update in one direction shows the latest fee + assert_eq!(first_channel.one_to_two.as_ref().unwrap().fees.proportional_millionths, 39); + assert_eq!(first_channel.two_to_one.as_ref().unwrap().fees.proportional_millionths, 10); + } + + clean_test_db().await; +} + +#[tokio::test] +async fn test_full_snapshot_recency_with_wrong_seen_order() { + let _sanitizer = SchemaSanitizer::new(); + let logger = Arc::new(TestLogger::new()); + let network_graph = NetworkGraph::new(Network::Bitcoin, logger.clone()); + let network_graph_arc = Arc::new(network_graph); + + let short_channel_id = 1; + let timestamp = current_time(); + println!("timestamp: {}", timestamp); + + { // seed the db + let (mut persister, receiver) = GossipPersister::new(network_graph_arc.clone(), logger.clone()); + let announcement = generate_channel_announcement(short_channel_id); + network_graph_arc.update_channel_from_announcement_no_lookup(&announcement).unwrap(); + receiver.send(GossipMessage::ChannelAnnouncement(announcement, None)).await.unwrap(); + + { // direction false + { // first update, seen latest + let update = generate_update(short_channel_id, false, timestamp - 1, 0, 0, 0, 0, 38); + network_graph_arc.update_channel_unsigned(&update.contents).unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update, Some(timestamp))).await.unwrap(); + } + { // second update, seen first + let update = generate_update(short_channel_id, false, timestamp, 0, 0, 0, 0, 39); + network_graph_arc.update_channel_unsigned(&update.contents).unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update, Some(timestamp - 1))).await.unwrap(); + } + } + { // direction true + { // first and only update + let update = generate_update(short_channel_id, true, timestamp, 0, 0, 0, 0, 10); + network_graph_arc.update_channel_unsigned(&update.contents).unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap(); + } + } + + drop(receiver); + persister.persist_gossip().await; + + tokio::task::spawn_blocking(move || { + drop(persister); + }).await.unwrap(); + } + + let client_graph = NetworkGraph::new(Network::Bitcoin, logger.clone()); + let client_graph_arc = Arc::new(client_graph); + + { // sync after initial seed + let serialization = serialize_delta(network_graph_arc.clone(), 0, None, logger.clone()).await; + logger.assert_log_contains("rapid_gossip_sync_server", "announcement channel count: 1", 1); + + let channel_count = network_graph_arc.read_only().channels().len(); + + assert_eq!(channel_count, 1); + assert_eq!(serialization.message_count, 3); + assert_eq!(serialization.announcement_count, 1); + assert_eq!(serialization.update_count, 2); + + let rgs = RapidGossipSync::new(client_graph_arc.clone(), logger.clone()); + let update_result = rgs.update_network_graph(&serialization.data).unwrap(); + // the update result must be a multiple of our snapshot granularity + assert_eq!(update_result % config::snapshot_generation_interval(), 0); + assert!(update_result < timestamp); + + let readonly_graph = client_graph_arc.read_only(); + let channels = readonly_graph.channels(); + let client_channel_count = channels.len(); + assert_eq!(client_channel_count, 1); + + let first_channel = channels.get(&short_channel_id).unwrap(); + assert!(&first_channel.announcement_message.is_none()); + // ensure the update in one direction shows the latest fee + assert_eq!(first_channel.one_to_two.as_ref().unwrap().fees.proportional_millionths, 39); + assert_eq!(first_channel.two_to_one.as_ref().unwrap().fees.proportional_millionths, 10); + } + + clean_test_db().await; +} + +#[tokio::test] +async fn test_full_snapshot_recency_with_wrong_propagation_order() { + let _sanitizer = SchemaSanitizer::new(); + let logger = Arc::new(TestLogger::new()); + let network_graph = NetworkGraph::new(Network::Bitcoin, logger.clone()); + let network_graph_arc = Arc::new(network_graph); + + let short_channel_id = 1; + let timestamp = current_time(); + println!("timestamp: {}", timestamp); + + { // seed the db + let (mut persister, receiver) = GossipPersister::new(network_graph_arc.clone(), logger.clone()); + let announcement = generate_channel_announcement(short_channel_id); + network_graph_arc.update_channel_from_announcement_no_lookup(&announcement).unwrap(); + receiver.send(GossipMessage::ChannelAnnouncement(announcement, None)).await.unwrap(); + + { // direction false + // apply updates in their timestamp order + let update_1 = generate_update(short_channel_id, false, timestamp - 1, 0, 0, 0, 0, 38); + let update_2 = generate_update(short_channel_id, false, timestamp, 0, 0, 0, 0, 39); + network_graph_arc.update_channel_unsigned(&update_1.contents).unwrap(); + network_graph_arc.update_channel_unsigned(&update_2.contents).unwrap(); + + // propagate updates in their seen order + receiver.send(GossipMessage::ChannelUpdate(update_2, Some(timestamp - 1))).await.unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update_1, Some(timestamp))).await.unwrap(); + } + { // direction true + { // first and only update + let update = generate_update(short_channel_id, true, timestamp, 0, 0, 0, 0, 10); + network_graph_arc.update_channel_unsigned(&update.contents).unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap(); + } + } + + drop(receiver); + persister.persist_gossip().await; + + tokio::task::spawn_blocking(move || { + drop(persister); + }).await.unwrap(); + } + + let client_graph = NetworkGraph::new(Network::Bitcoin, logger.clone()); + let client_graph_arc = Arc::new(client_graph); + + { // sync after initial seed + let serialization = serialize_delta(network_graph_arc.clone(), 0, None, logger.clone()).await; + logger.assert_log_contains("rapid_gossip_sync_server", "announcement channel count: 1", 1); + + let channel_count = network_graph_arc.read_only().channels().len(); + + assert_eq!(channel_count, 1); + assert_eq!(serialization.message_count, 3); + assert_eq!(serialization.announcement_count, 1); + assert_eq!(serialization.update_count, 2); + + let rgs = RapidGossipSync::new(client_graph_arc.clone(), logger.clone()); + let update_result = rgs.update_network_graph(&serialization.data).unwrap(); + // the update result must be a multiple of our snapshot granularity + assert_eq!(update_result % config::snapshot_generation_interval(), 0); + assert!(update_result < timestamp); + + let readonly_graph = client_graph_arc.read_only(); + let channels = readonly_graph.channels(); + let client_channel_count = channels.len(); + assert_eq!(client_channel_count, 1); + + let first_channel = channels.get(&short_channel_id).unwrap(); + assert!(&first_channel.announcement_message.is_none()); + // ensure the update in one direction shows the latest fee + assert_eq!(first_channel.one_to_two.as_ref().unwrap().fees.proportional_millionths, 39); + assert_eq!(first_channel.two_to_one.as_ref().unwrap().fees.proportional_millionths, 10); + } + + clean_test_db().await; +} + +#[tokio::test] +async fn test_full_snapshot_mutiny_scenario() { + let _sanitizer = SchemaSanitizer::new(); + let logger = Arc::new(TestLogger::new()); + let network_graph = NetworkGraph::new(Network::Bitcoin, logger.clone()); + let network_graph_arc = Arc::new(network_graph); + + let short_channel_id = 873706024403271681; + let timestamp = current_time(); + // let oldest_simulation_timestamp = 1693300588; + let latest_simulation_timestamp = 1695909301; + let timestamp_offset = timestamp - latest_simulation_timestamp; + println!("timestamp: {}", timestamp); + + { // seed the db + let (mut persister, receiver) = GossipPersister::new(network_graph_arc.clone(), logger.clone()); + let announcement = generate_channel_announcement(short_channel_id); + network_graph_arc.update_channel_from_announcement_no_lookup(&announcement).unwrap(); + receiver.send(GossipMessage::ChannelAnnouncement(announcement, None)).await.unwrap(); + + { // direction false + { + let update = generate_update(short_channel_id, false, 1693507369 + timestamp_offset, 0, 0, 0, 0, 38); + network_graph_arc.update_channel_unsigned(&update.contents).unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap(); + } + { + let update = generate_update(short_channel_id, false, 1693680390 + timestamp_offset, 0, 0, 0, 0, 38); + network_graph_arc.update_channel_unsigned(&update.contents).unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap(); + } + { + let update = generate_update(short_channel_id, false, 1693749109 + timestamp_offset, 0, 0, 0, 0, 200); + network_graph_arc.update_channel_unsigned(&update.contents).unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap(); + } + { + let update = generate_update(short_channel_id, false, 1693925190 + timestamp_offset, 0, 0, 0, 0, 200); + network_graph_arc.update_channel_unsigned(&update.contents).unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap(); + } + { + let update = generate_update(short_channel_id, false, 1694008323 + timestamp_offset, 0, 0, 0, 0, 209); + network_graph_arc.update_channel_unsigned(&update.contents).unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap(); + } + { + let update = generate_update(short_channel_id, false, 1694219924 + timestamp_offset, 0, 0, 0, 0, 209); + network_graph_arc.update_channel_unsigned(&update.contents).unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap(); + } + { + let update = generate_update(short_channel_id, false, 1694267536 + timestamp_offset, 0, 0, 0, 0, 210); + network_graph_arc.update_channel_unsigned(&update.contents).unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap(); + } + { + let update = generate_update(short_channel_id, false, 1694458808 + timestamp_offset, 0, 0, 0, 0, 210); + network_graph_arc.update_channel_unsigned(&update.contents).unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap(); + } + { + let update = generate_update(short_channel_id, false, 1694526734 + timestamp_offset, 0, 0, 0, 0, 200); + network_graph_arc.update_channel_unsigned(&update.contents).unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap(); + } + { + let update = generate_update(short_channel_id, false, 1694794765 + timestamp_offset, 0, 0, 0, 0, 200); + network_graph_arc.update_channel_unsigned(&update.contents).unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update, Some(1695909301 + 2 * config::SYMLINK_GRANULARITY_INTERVAL + timestamp_offset))).await.unwrap(); + } + { + let update = generate_update(short_channel_id, false, 1695909301 + timestamp_offset, 0, 0, 0, 0, 130); + // network_graph_arc.update_channel_unsigned(&update.contents).unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap(); + } + } + { // direction true + { + let update = generate_update(short_channel_id, true, 1693300588 + timestamp_offset, 0, 0, 0, 0, 10); + network_graph_arc.update_channel_unsigned(&update.contents).unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap(); + } + { + let update = generate_update(short_channel_id, true, 1695003621 + timestamp_offset, 0, 0, 0, 0, 10); + network_graph_arc.update_channel_unsigned(&update.contents).unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap(); + } + } + + drop(receiver); + persister.persist_gossip().await; + + tokio::task::spawn_blocking(move || { + drop(persister); + }).await.unwrap(); + } + + let client_graph = NetworkGraph::new(Network::Bitcoin, logger.clone()); + let client_graph_arc = Arc::new(client_graph); + + { // sync after initial seed + let serialization = serialize_delta(network_graph_arc.clone(), 0, None, logger.clone()).await; + logger.assert_log_contains("rapid_gossip_sync_server", "announcement channel count: 1", 1); + + let channel_count = network_graph_arc.read_only().channels().len(); + + assert_eq!(channel_count, 1); + assert_eq!(serialization.message_count, 3); + assert_eq!(serialization.announcement_count, 1); + assert_eq!(serialization.update_count, 2); + + let rgs = RapidGossipSync::new(client_graph_arc.clone(), logger.clone()); + let update_result = rgs.update_network_graph(&serialization.data).unwrap(); + println!("update result: {}", update_result); + // the update result must be a multiple of our snapshot granularity + assert_eq!(update_result % config::snapshot_generation_interval(), 0); + assert!(update_result < timestamp); + + let timestamp_delta = timestamp - update_result; + println!("timestamp delta: {}", timestamp_delta); + assert!(timestamp_delta < config::snapshot_generation_interval()); + + let readonly_graph = client_graph_arc.read_only(); + let channels = readonly_graph.channels(); + let client_channel_count = channels.len(); + assert_eq!(client_channel_count, 1); + + let first_channel = channels.get(&short_channel_id).unwrap(); + assert!(&first_channel.announcement_message.is_none()); + assert_eq!(first_channel.one_to_two.as_ref().unwrap().fees.proportional_millionths, 130); + assert_eq!(first_channel.two_to_one.as_ref().unwrap().fees.proportional_millionths, 10); + } + + clean_test_db().await; +} + +#[tokio::test] +async fn test_full_snapshot_interlaced_channel_timestamps() { + let _sanitizer = SchemaSanitizer::new(); + let logger = Arc::new(TestLogger::new()); + let network_graph = NetworkGraph::new(Network::Bitcoin, logger.clone()); + let network_graph_arc = Arc::new(network_graph); + + let main_channel_id = 1; + let timestamp = current_time(); + println!("timestamp: {}", timestamp); + + { // seed the db + let (mut persister, receiver) = GossipPersister::new(network_graph_arc.clone(), logger.clone()); + let secondary_channel_id = main_channel_id + 1; + + { // main channel + let announcement = generate_channel_announcement(main_channel_id); + network_graph_arc.update_channel_from_announcement_no_lookup(&announcement).unwrap(); + receiver.send(GossipMessage::ChannelAnnouncement(announcement, None)).await.unwrap(); + } + + { // secondary channel + let announcement = generate_channel_announcement(secondary_channel_id); + network_graph_arc.update_channel_from_announcement_no_lookup(&announcement).unwrap(); + receiver.send(GossipMessage::ChannelAnnouncement(announcement, None)).await.unwrap(); + } + + { // main channel + { // direction false + let update = generate_update(main_channel_id, false, timestamp - 2, 0, 0, 0, 0, 10); + network_graph_arc.update_channel_unsigned(&update.contents).unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap(); + } + { // direction true + let update = generate_update(main_channel_id, true, timestamp - 2, 0, 0, 0, 0, 5); + network_graph_arc.update_channel_unsigned(&update.contents).unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap(); + } + } + + { // in-between channel + { // direction false + let update = generate_update(secondary_channel_id, false, timestamp - 1, 0, 0, 0, 0, 42); + network_graph_arc.update_channel_unsigned(&update.contents).unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap(); + } + { // direction true + let update = generate_update(secondary_channel_id, true, timestamp - 1, 0, 0, 0, 0, 42); + network_graph_arc.update_channel_unsigned(&update.contents).unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap(); + } + } + + { // main channel + { // direction false + let update = generate_update(main_channel_id, false, timestamp, 0, 0, 0, 0, 11); + network_graph_arc.update_channel_unsigned(&update.contents).unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap(); + } + { // direction true + let update = generate_update(main_channel_id, true, timestamp, 0, 0, 0, 0, 6); + network_graph_arc.update_channel_unsigned(&update.contents).unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap(); + } + } + + drop(receiver); + persister.persist_gossip().await; + + tokio::task::spawn_blocking(move || { + drop(persister); + }).await.unwrap(); + } + + let client_graph = NetworkGraph::new(Network::Bitcoin, logger.clone()); + let client_graph_arc = Arc::new(client_graph); + + { // sync after initial seed + let serialization = serialize_delta(network_graph_arc.clone(), 0, None, logger.clone()).await; + logger.assert_log_contains("rapid_gossip_sync_server", "announcement channel count: 2", 1); + + let channel_count = network_graph_arc.read_only().channels().len(); + + assert_eq!(channel_count, 2); + assert_eq!(serialization.message_count, 6); + assert_eq!(serialization.announcement_count, 2); + assert_eq!(serialization.update_count, 4); + + let rgs = RapidGossipSync::new(client_graph_arc.clone(), logger.clone()); + let update_result = rgs.update_network_graph(&serialization.data).unwrap(); + // the update result must be a multiple of our snapshot granularity + assert_eq!(update_result % config::snapshot_generation_interval(), 0); + assert!(update_result < timestamp); + + let readonly_graph = client_graph_arc.read_only(); + let channels = readonly_graph.channels(); + let client_channel_count = channels.len(); + assert_eq!(client_channel_count, 2); + + let first_channel = channels.get(&main_channel_id).unwrap(); + assert!(&first_channel.announcement_message.is_none()); + // ensure the update in one direction shows the latest fee + assert_eq!(first_channel.one_to_two.as_ref().unwrap().fees.proportional_millionths, 11); + assert_eq!(first_channel.two_to_one.as_ref().unwrap().fees.proportional_millionths, 6); + } + + clean_test_db().await; +} + +#[tokio::test] +async fn test_full_snapshot_persistence() { + let schema_sanitizer = SchemaSanitizer::new(); + let logger = Arc::new(TestLogger::new()); + let network_graph = NetworkGraph::new(Network::Bitcoin, logger.clone()); + let network_graph_arc = Arc::new(network_graph); + let snapshotter = Snapshotter::new(network_graph_arc.clone(), logger.clone()); + let cache_sanitizer = CacheSanitizer::new(&schema_sanitizer); + + let short_channel_id = 1; + let timestamp = current_time(); + println!("timestamp: {}", timestamp); + + { // seed the db + let (mut persister, receiver) = GossipPersister::new(network_graph_arc.clone(), logger.clone()); + let announcement = generate_channel_announcement(short_channel_id); + network_graph_arc.update_channel_from_announcement_no_lookup(&announcement).unwrap(); + receiver.send(GossipMessage::ChannelAnnouncement(announcement, None)).await.unwrap(); + + { // direction true + let update = generate_update(short_channel_id, true, timestamp, 0, 0, 0, 0, 10); + network_graph_arc.update_channel_unsigned(&update.contents).unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap(); + } + + { // direction false + let update = generate_update(short_channel_id, false, timestamp - 1, 0, 0, 0, 0, 38); + network_graph_arc.update_channel_unsigned(&update.contents).unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap(); + } + + + drop(receiver); + persister.persist_gossip().await; + + tokio::task::spawn_blocking(move || { + drop(persister); + }).await.unwrap(); + } + + let cache_path = cache_sanitizer.cache_path(); + let symlink_path = format!("{}/symlinks/0.bin", cache_path); + + // generate snapshots + { + snapshotter.generate_snapshots(20, 5, &[5, u64::MAX], &cache_path, Some(10)).await; + + let symlinked_data = fs::read(&symlink_path).unwrap(); + let client_graph = NetworkGraph::new(Network::Bitcoin, logger.clone()); + let client_graph_arc = Arc::new(client_graph); + + let rgs = RapidGossipSync::new(client_graph_arc.clone(), logger.clone()); + let update_result = rgs.update_network_graph(&symlinked_data).unwrap(); + // the update result must be a multiple of our snapshot granularity + assert_eq!(update_result % config::snapshot_generation_interval(), 0); + + let readonly_graph = client_graph_arc.read_only(); + let channels = readonly_graph.channels(); + let client_channel_count = channels.len(); + assert_eq!(client_channel_count, 1); + + let first_channel = channels.get(&short_channel_id).unwrap(); + assert!(&first_channel.announcement_message.is_none()); + // ensure the update in one direction shows the latest fee + assert_eq!(first_channel.one_to_two.as_ref().unwrap().fees.proportional_millionths, 38); + assert_eq!(first_channel.two_to_one.as_ref().unwrap().fees.proportional_millionths, 10); + } + + { // update the db + let (mut persister, receiver) = GossipPersister::new(network_graph_arc.clone(), logger.clone()); + + { // second update + let update = generate_update(short_channel_id, false, timestamp + 30, 0, 0, 0, 0, 39); + network_graph_arc.update_channel_unsigned(&update.contents).unwrap(); + receiver.send(GossipMessage::ChannelUpdate(update, None)).await.unwrap(); + } + + drop(receiver); + persister.persist_gossip().await; + + tokio::task::spawn_blocking(move || { + drop(persister); + }).await.unwrap(); + } + + // regenerate snapshots + { + snapshotter.generate_snapshots(20, 5, &[5, u64::MAX], &cache_path, Some(10)).await; + + let symlinked_data = fs::read(&symlink_path).unwrap(); + let client_graph = NetworkGraph::new(Network::Bitcoin, logger.clone()); + let client_graph_arc = Arc::new(client_graph); + + let rgs = RapidGossipSync::new(client_graph_arc.clone(), logger.clone()); + let update_result = rgs.update_network_graph(&symlinked_data).unwrap(); + // the update result must be a multiple of our snapshot granularity + assert_eq!(update_result % config::snapshot_generation_interval(), 0); + + let readonly_graph = client_graph_arc.read_only(); + let channels = readonly_graph.channels(); + let client_channel_count = channels.len(); + assert_eq!(client_channel_count, 1); + + let first_channel = channels.get(&short_channel_id).unwrap(); + assert!(&first_channel.announcement_message.is_none()); + // ensure the update in one direction shows the latest fee + assert_eq!(first_channel.one_to_two.as_ref().unwrap().fees.proportional_millionths, 39); + assert_eq!(first_channel.two_to_one.as_ref().unwrap().fees.proportional_millionths, 10); + } + + // clean up afterwards + clean_test_db().await; }