X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=src%2Fpersistence.rs;h=04c6b9a0c4076cab725b90b5a4ec769718d183a0;hb=49858cac9e83132572ea59b786922ab4a135f48a;hp=51c242b7c8e94fab862fc305e8558e23cde719f7;hpb=fd6ee2405e4bebb5be32e2d7a7494ada10b49397;p=rapid-gossip-sync-server diff --git a/src/persistence.rs b/src/persistence.rs index 51c242b..04c6b9a 100644 --- a/src/persistence.rs +++ b/src/persistence.rs @@ -1,43 +1,46 @@ use std::fs::OpenOptions; use std::io::{BufWriter, Write}; +use std::ops::Deref; use std::sync::Arc; use std::time::{Duration, Instant}; +use lightning::log_info; use lightning::routing::gossip::NetworkGraph; +use lightning::util::logger::Logger; use lightning::util::ser::Writeable; -use tokio::sync::mpsc; -use tokio_postgres::NoTls; +use tokio::runtime::Runtime; +use tokio::sync::{mpsc, Mutex, Semaphore}; -use crate::{config, TestLogger}; +use crate::config; use crate::types::GossipMessage; -pub(crate) struct GossipPersister { +const POSTGRES_INSERT_TIMEOUT: Duration = Duration::from_secs(15); +const INSERT_PARALELLISM: usize = 16; + +pub(crate) struct GossipPersister where L::Target: Logger { gossip_persistence_receiver: mpsc::Receiver, - network_graph: Arc>, + network_graph: Arc>, + tokio_runtime: Runtime, + logger: L } -impl GossipPersister { - pub fn new(network_graph: Arc>) -> (Self, mpsc::Sender) { +impl GossipPersister where L::Target: Logger { + pub fn new(network_graph: Arc>, logger: L) -> (Self, mpsc::Sender) { let (gossip_persistence_sender, gossip_persistence_receiver) = mpsc::channel::(100); + let runtime = Runtime::new().unwrap(); (GossipPersister { gossip_persistence_receiver, - network_graph + network_graph, + tokio_runtime: runtime, + logger }, gossip_persistence_sender) } pub(crate) async fn persist_gossip(&mut self) { - let connection_config = config::db_connection_config(); - let (mut client, connection) = - connection_config.connect(NoTls).await.unwrap(); - - tokio::spawn(async move { - if let Err(e) = connection.await { - panic!("connection error: {}", e); - } - }); + { // initialize the database + // this client instance is only used once + let mut client = crate::connect_to_db().await; - { - // initialize the database let initialization = client .execute(config::db_config_table_creation_query(), &[]) .await; @@ -50,6 +53,11 @@ impl GossipPersister { config::upgrade_db(cur_schema[0].get(0), &mut client).await; } + let preparation = client.execute("set time zone UTC", &[]).await; + if let Err(preparation_error) = preparation { + panic!("db preparation error: {}", preparation_error); + } + let initialization = client .execute( // TODO: figure out a way to fix the id value without Postgres complaining about @@ -61,21 +69,20 @@ impl GossipPersister { panic!("db init error: {}", initialization_error); } - let initialization = client - .execute(config::db_announcement_table_creation_query(), &[]) - .await; - if let Err(initialization_error) = initialization { - panic!("db init error: {}", initialization_error); - } + let table_creation_queries = [ + config::db_announcement_table_creation_query(), + config::db_channel_update_table_creation_query(), + config::db_channel_update_table_creation_query(), + config::db_node_announcement_table_creation_query() + ]; - let initialization = client - .execute( - config::db_channel_update_table_creation_query(), - &[], - ) - .await; - if let Err(initialization_error) = initialization { - panic!("db init error: {}", initialization_error); + for current_table_creation_query in table_creation_queries { + let initialization = client + .execute(current_table_creation_query, &[]) + .await; + if let Err(initialization_error) = initialization { + panic!("db init error: {}", initialization_error); + } } let initialization = client @@ -90,14 +97,18 @@ impl GossipPersister { let mut latest_persistence_log = Instant::now() - Duration::from_secs(60); let mut i = 0u32; let mut latest_graph_cache_time = Instant::now(); + let insert_limiter = Arc::new(Semaphore::new(INSERT_PARALELLISM)); + let connections_cache = Arc::new(Mutex::new(Vec::with_capacity(INSERT_PARALELLISM))); + #[cfg(test)] + let mut tasks_spawned = Vec::new(); // TODO: it would be nice to have some sort of timeout here so after 10 seconds of // inactivity, some sort of message could be broadcast signaling the activation of request // processing - while let Some(gossip_message) = &self.gossip_persistence_receiver.recv().await { + while let Some(gossip_message) = self.gossip_persistence_receiver.recv().await { i += 1; // count the persisted gossip messages if latest_persistence_log.elapsed().as_secs() >= 60 { - println!("Persisting gossip message #{}", i); + log_info!(self.logger, "Persisting gossip message #{}", i); latest_persistence_log = Instant::now(); } @@ -106,28 +117,111 @@ impl GossipPersister { self.persist_network_graph(); latest_graph_cache_time = Instant::now(); } + insert_limiter.acquire().await.unwrap().forget(); + + let limiter_ref = Arc::clone(&insert_limiter); + let client = { + let mut connections_set = connections_cache.lock().await; + let client = if connections_set.is_empty() { + crate::connect_to_db().await + } else { + connections_set.pop().unwrap() + }; + client + }; + + let connections_cache_ref = Arc::clone(&connections_cache); + match gossip_message { + GossipMessage::NodeAnnouncement(announcement, seen_override) => { + let public_key_hex = announcement.contents.node_id.to_string(); + + let mut announcement_signed = Vec::new(); + announcement.write(&mut announcement_signed).unwrap(); + + let features = announcement.contents.features.encode(); + let timestamp = announcement.contents.timestamp as i64; - match &gossip_message { - GossipMessage::ChannelAnnouncement(announcement) => { + let mut serialized_addresses = Vec::new(); + announcement.contents.addresses.write(&mut serialized_addresses).unwrap(); + + let _task = self.tokio_runtime.spawn(async move { + if cfg!(test) && seen_override.is_some() { + tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client + .execute("INSERT INTO node_announcements (\ + public_key, \ + features, \ + socket_addresses, \ + timestamp, \ + announcement_signed, \ + seen \ + ) VALUES ($1, $2, $3, $4, $5, TO_TIMESTAMP($6))", &[ + &public_key_hex, + &features, + &serialized_addresses, + ×tamp, + &announcement_signed, + &(seen_override.unwrap() as f64) + ])).await.unwrap().unwrap(); + } else { + tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client + .execute("INSERT INTO node_announcements (\ + public_key, \ + features, \ + socket_addresses, \ + timestamp, \ + announcement_signed \ + ) VALUES ($1, $2, $3, $4, $5)", &[ + &public_key_hex, + &features, + &serialized_addresses, + ×tamp, + &announcement_signed, + ])).await.unwrap().unwrap(); + } + let mut connections_set = connections_cache_ref.lock().await; + connections_set.push(client); + limiter_ref.add_permits(1); + }); + #[cfg(test)] + tasks_spawned.push(_task); + }, + GossipMessage::ChannelAnnouncement(announcement, seen_override) => { let scid = announcement.contents.short_channel_id as i64; // start with the type prefix, which is already known a priori let mut announcement_signed = Vec::new(); announcement.write(&mut announcement_signed).unwrap(); - let result = client - .execute("INSERT INTO channel_announcements (\ - short_channel_id, \ - announcement_signed \ - ) VALUES ($1, $2) ON CONFLICT (short_channel_id) DO NOTHING", &[ - &scid, - &announcement_signed - ]).await; - if result.is_err() { - panic!("error: {}", result.err().unwrap()); - } + let _task = self.tokio_runtime.spawn(async move { + if cfg!(test) && seen_override.is_some() { + tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client + .execute("INSERT INTO channel_announcements (\ + short_channel_id, \ + announcement_signed, \ + seen \ + ) VALUES ($1, $2, TO_TIMESTAMP($3)) ON CONFLICT (short_channel_id) DO NOTHING", &[ + &scid, + &announcement_signed, + &(seen_override.unwrap() as f64) + ])).await.unwrap().unwrap(); + } else { + tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client + .execute("INSERT INTO channel_announcements (\ + short_channel_id, \ + announcement_signed \ + ) VALUES ($1, $2) ON CONFLICT (short_channel_id) DO NOTHING", &[ + &scid, + &announcement_signed + ])).await.unwrap().unwrap(); + } + let mut connections_set = connections_cache_ref.lock().await; + connections_set.push(client); + limiter_ref.add_permits(1); + }); + #[cfg(test)] + tasks_spawned.push(_task); } - GossipMessage::ChannelUpdate(update) => { + GossipMessage::ChannelUpdate(update, seen_override) => { let scid = update.contents.short_channel_id as i64; let timestamp = update.contents.timestamp as i64; @@ -146,10 +240,11 @@ impl GossipPersister { let mut update_signed = Vec::new(); update.write(&mut update_signed).unwrap(); - let result = client - .execute("INSERT INTO channel_updates (\ + let insertion_statement = if cfg!(test) { + "INSERT INTO channel_updates (\ short_channel_id, \ timestamp, \ + seen, \ channel_flags, \ direction, \ disable, \ @@ -159,29 +254,60 @@ impl GossipPersister { fee_proportional_millionths, \ htlc_maximum_msat, \ blob_signed \ - ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) ON CONFLICT DO NOTHING", &[ - &scid, - ×tamp, - &(update.contents.flags as i16), - &direction, - &disable, - &cltv_expiry_delta, - &htlc_minimum_msat, - &fee_base_msat, - &fee_proportional_millionths, - &htlc_maximum_msat, - &update_signed - ]).await; - if result.is_err() { - panic!("error: {}", result.err().unwrap()); - } + ) VALUES ($1, $2, TO_TIMESTAMP($3), $4, $5, $6, $7, $8, $9, $10, $11, $12) ON CONFLICT DO NOTHING" + } else { + "INSERT INTO channel_updates (\ + short_channel_id, \ + timestamp, \ + channel_flags, \ + direction, \ + disable, \ + cltv_expiry_delta, \ + htlc_minimum_msat, \ + fee_base_msat, \ + fee_proportional_millionths, \ + htlc_maximum_msat, \ + blob_signed \ + ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) ON CONFLICT DO NOTHING" + }; + + // this may not be used outside test cfg + let _seen_timestamp = seen_override.unwrap_or(timestamp as u32) as f64; + + let _task = self.tokio_runtime.spawn(async move { + tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client + .execute(insertion_statement, &[ + &scid, + ×tamp, + #[cfg(test)] + &_seen_timestamp, + &(update.contents.flags as i16), + &direction, + &disable, + &cltv_expiry_delta, + &htlc_minimum_msat, + &fee_base_msat, + &fee_proportional_millionths, + &htlc_maximum_msat, + &update_signed + ])).await.unwrap().unwrap(); + let mut connections_set = connections_cache_ref.lock().await; + connections_set.push(client); + limiter_ref.add_permits(1); + }); + #[cfg(test)] + tasks_spawned.push(_task); } } } + #[cfg(test)] + for task in tasks_spawned { + task.await.unwrap(); + } } fn persist_network_graph(&self) { - println!("Caching network graph…"); + log_info!(self.logger, "Caching network graph…"); let cache_path = config::network_graph_cache_path(); let file = OpenOptions::new() .create(true) @@ -189,10 +315,10 @@ impl GossipPersister { .truncate(true) .open(&cache_path) .unwrap(); - self.network_graph.remove_stale_channels(); + self.network_graph.remove_stale_channels_and_tracking(); let mut writer = BufWriter::new(file); self.network_graph.write(&mut writer).unwrap(); writer.flush().unwrap(); - println!("Cached network graph!"); + log_info!(self.logger, "Cached network graph!"); } }