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

   1 use std::fs::OpenOptions;
   2 use std::io::{BufWriter, Write};
   3 use std::ops::Deref;
   4 use std::sync::Arc;
   5 use std::time::{Duration, Instant};
   6 use lightning::log_info;
   7 use lightning::routing::gossip::NetworkGraph;
   8 use lightning::util::logger::Logger;
   9 use lightning::util::ser::Writeable;
  10 use tokio::runtime::Runtime;
  11 use tokio::sync::{mpsc, Mutex, Semaphore};
  12
  13 use crate::config;
  14 use crate::types::GossipMessage;
  15
  16 const POSTGRES_INSERT_TIMEOUT: Duration = Duration::from_secs(15);
  17 const INSERT_PARALELLISM: usize = 16;
  18
  19 pub(crate) struct GossipPersister<L: Deref> where L::Target: Logger {
  20         gossip_persistence_receiver: mpsc::Receiver<GossipMessage>,
  21         network_graph: Arc<NetworkGraph<L>>,
  22         tokio_runtime: Runtime,
  23         logger: L
  24 }
  25
  26 impl<L: Deref> GossipPersister<L> where L::Target: Logger {
  27         pub fn new(network_graph: Arc<NetworkGraph<L>>, logger: L) -> (Self, mpsc::Sender<GossipMessage>) {
  28                 let (gossip_persistence_sender, gossip_persistence_receiver) =
  29                         mpsc::channel::<GossipMessage>(100);
  30                 let runtime = Runtime::new().unwrap();
  31                 (GossipPersister {
  32                         gossip_persistence_receiver,
  33                         network_graph,
  34                         tokio_runtime: runtime,
  35                         logger
  36                 }, gossip_persistence_sender)
  37         }
  38
  39         pub(crate) async fn persist_gossip(&mut self) {
  40                 { // initialize the database
  41                         // this client instance is only used once
  42                         let mut client = crate::connect_to_db().await;
  43
  44                         let initialization = client
  45                                 .execute(config::db_config_table_creation_query(), &[])
  46                                 .await;
  47                         if let Err(initialization_error) = initialization {
  48                                 panic!("db init error: {}", initialization_error);
  49                         }
  50
  51                         let cur_schema = client.query("SELECT db_schema FROM config WHERE id = $1", &[&1]).await.unwrap();
  52                         if !cur_schema.is_empty() {
  53                                 config::upgrade_db(cur_schema[0].get(0), &mut client).await;
  54                         }
  55
  56                         let preparation = client.execute("set time zone UTC", &[]).await;
  57                         if let Err(preparation_error) = preparation {
  58                                 panic!("db preparation error: {}", preparation_error);
  59                         }
  60
  61                         let initialization = client
  62                                 .execute(
  63                                         // TODO: figure out a way to fix the id value without Postgres complaining about
  64                                         // its value not being default
  65                                         "INSERT INTO config (id, db_schema) VALUES ($1, $2) ON CONFLICT (id) DO NOTHING",
  66                                         &[&1, &config::SCHEMA_VERSION]
  67                                 ).await;
  68                         if let Err(initialization_error) = initialization {
  69                                 panic!("db init error: {}", initialization_error);
  70                         }
  71
  72                         let table_creation_queries = [
  73                                 config::db_announcement_table_creation_query(),
  74                                 config::db_channel_update_table_creation_query(),
  75                                 config::db_channel_update_table_creation_query(),
  76                                 config::db_node_announcement_table_creation_query()
  77                         ];
  78
  79                         for current_table_creation_query in table_creation_queries {
  80                                 let initialization = client
  81                                         .execute(current_table_creation_query, &[])
  82                                         .await;
  83                                 if let Err(initialization_error) = initialization {
  84                                         panic!("db init error: {}", initialization_error);
  85                                 }
  86                         }
  87
  88                         let initialization = client
  89                                 .batch_execute(config::db_index_creation_query())
  90                                 .await;
  91                         if let Err(initialization_error) = initialization {
  92                                 panic!("db init error: {}", initialization_error);
  93                         }
  94                 }
  95
  96                 // print log statement every minute
  97                 let mut latest_persistence_log = Instant::now() - Duration::from_secs(60);
  98                 let mut i = 0u32;
  99                 let mut latest_graph_cache_time = Instant::now();
 100                 let insert_limiter = Arc::new(Semaphore::new(INSERT_PARALELLISM));
 101                 let connections_cache = Arc::new(Mutex::new(Vec::with_capacity(INSERT_PARALELLISM)));
 102                 #[cfg(test)]
 103                 let mut tasks_spawned = Vec::new();
 104                 // TODO: it would be nice to have some sort of timeout here so after 10 seconds of
 105                 // inactivity, some sort of message could be broadcast signaling the activation of request
 106                 // processing
 107                 while let Some(gossip_message) = self.gossip_persistence_receiver.recv().await {
 108                         i += 1; // count the persisted gossip messages
 109
 110                         if latest_persistence_log.elapsed().as_secs() >= 60 {
 111                                 log_info!(self.logger, "Persisting gossip message #{}", i);
 112                                 latest_persistence_log = Instant::now();
 113                         }
 114
 115                         // has it been ten minutes? Just cache it
 116                         if latest_graph_cache_time.elapsed().as_secs() >= 600 {
 117                                 self.persist_network_graph();
 118                                 latest_graph_cache_time = Instant::now();
 119                         }
 120                         insert_limiter.acquire().await.unwrap().forget();
 121
 122                         let limiter_ref = Arc::clone(&insert_limiter);
 123                         let client = {
 124                                 let mut connections_set = connections_cache.lock().await;
 125                                 let client = if connections_set.is_empty() {
 126                                         crate::connect_to_db().await
 127                                 } else {
 128                                         connections_set.pop().unwrap()
 129                                 };
 130                                 client
 131                         };
 132
 133                         let connections_cache_ref = Arc::clone(&connections_cache);
 134                         match gossip_message {
 135                                 GossipMessage::NodeAnnouncement(announcement, seen_override) => {
 136                                         let public_key_hex = announcement.contents.node_id.to_string();
 137
 138                                         let mut announcement_signed = Vec::new();
 139                                         announcement.write(&mut announcement_signed).unwrap();
 140
 141                                         let features = announcement.contents.features.encode();
 142                                         let timestamp = announcement.contents.timestamp as i64;
 143
 144                                         let mut serialized_addresses = Vec::new();
 145                                         announcement.contents.addresses.write(&mut serialized_addresses).unwrap();
 146
 147                                         let _task = self.tokio_runtime.spawn(async move {
 148                                                 if cfg!(test) && seen_override.is_some() {
 149                                                         tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client
 150                                                                 .execute("INSERT INTO node_announcements (\
 151                                                                 public_key, \
 152                                                                 features, \
 153                                                                 socket_addresses, \
 154                                                                 timestamp, \
 155                                                                 announcement_signed, \
 156                                                                 seen \
 157                                                         ) VALUES ($1, $2, $3, $4, $5, TO_TIMESTAMP($6))", &[
 158                                                                         &public_key_hex,
 159                                                                         &features,
 160                                                                         &serialized_addresses,
 161                                                                         &timestamp,
 162                                                                         &announcement_signed,
 163                                                                         &(seen_override.unwrap() as f64)
 164                                                                 ])).await.unwrap().unwrap();
 165                                                 } else {
 166                                                         tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client
 167                                                                 .execute("INSERT INTO node_announcements (\
 168                                                                 public_key, \
 169                                                                 features, \
 170                                                                 socket_addresses, \
 171                                                                 timestamp, \
 172                                                                 announcement_signed \
 173                                                         ) VALUES ($1, $2, $3, $4, $5)", &[
 174                                                                         &public_key_hex,
 175                                                                         &features,
 176                                                                         &serialized_addresses,
 177                                                                         &timestamp,
 178                                                                         &announcement_signed,
 179                                                                 ])).await.unwrap().unwrap();
 180                                                 }
 181                                                 let mut connections_set = connections_cache_ref.lock().await;
 182                                                 connections_set.push(client);
 183                                                 limiter_ref.add_permits(1);
 184                                         });
 185                                         #[cfg(test)]
 186                                         tasks_spawned.push(_task);
 187                                 },
 188                                 GossipMessage::ChannelAnnouncement(announcement, seen_override) => {
 189                                         let scid = announcement.contents.short_channel_id as i64;
 190
 191                                         // start with the type prefix, which is already known a priori
 192                                         let mut announcement_signed = Vec::new();
 193                                         announcement.write(&mut announcement_signed).unwrap();
 194
 195                                         let _task = self.tokio_runtime.spawn(async move {
 196                                                 if cfg!(test) && seen_override.is_some() {
 197                                                         tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client
 198                                                                 .execute("INSERT INTO channel_announcements (\
 199                                                                 short_channel_id, \
 200                                                                 announcement_signed, \
 201                                                                 seen \
 202                                                         ) VALUES ($1, $2, TO_TIMESTAMP($3)) ON CONFLICT (short_channel_id) DO NOTHING", &[
 203                                                                         &scid,
 204                                                                         &announcement_signed,
 205                                                                         &(seen_override.unwrap() as f64)
 206                                                                 ])).await.unwrap().unwrap();
 207                                                 } else {
 208                                                         tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client
 209                                                                 .execute("INSERT INTO channel_announcements (\
 210                                                                 short_channel_id, \
 211                                                                 announcement_signed \
 212                                                         ) VALUES ($1, $2) ON CONFLICT (short_channel_id) DO NOTHING", &[
 213                                                                         &scid,
 214                                                                         &announcement_signed
 215                                                                 ])).await.unwrap().unwrap();
 216                                                 }
 217                                                 let mut connections_set = connections_cache_ref.lock().await;
 218                                                 connections_set.push(client);
 219                                                 limiter_ref.add_permits(1);
 220                                         });
 221                                         #[cfg(test)]
 222                                         tasks_spawned.push(_task);
 223                                 }
 224                                 GossipMessage::ChannelUpdate(update, seen_override) => {
 225                                         let scid = update.contents.short_channel_id as i64;
 226
 227                                         let timestamp = update.contents.timestamp as i64;
 228
 229                                         let direction = (update.contents.flags & 1) == 1;
 230                                         let disable = (update.contents.flags & 2) > 0;
 231
 232                                         let cltv_expiry_delta = update.contents.cltv_expiry_delta as i32;
 233                                         let htlc_minimum_msat = update.contents.htlc_minimum_msat as i64;
 234                                         let fee_base_msat = update.contents.fee_base_msat as i32;
 235                                         let fee_proportional_millionths =
 236                                                 update.contents.fee_proportional_millionths as i32;
 237                                         let htlc_maximum_msat = update.contents.htlc_maximum_msat as i64;
 238
 239                                         // start with the type prefix, which is already known a priori
 240                                         let mut update_signed = Vec::new();
 241                                         update.write(&mut update_signed).unwrap();
 242
 243                                         let insertion_statement = if cfg!(test) {
 244                                                 "INSERT INTO channel_updates (\
 245                                                         short_channel_id, \
 246                                                         timestamp, \
 247                                                         seen, \
 248                                                         channel_flags, \
 249                                                         direction, \
 250                                                         disable, \
 251                                                         cltv_expiry_delta, \
 252                                                         htlc_minimum_msat, \
 253                                                         fee_base_msat, \
 254                                                         fee_proportional_millionths, \
 255                                                         htlc_maximum_msat, \
 256                                                         blob_signed \
 257                                                 ) VALUES ($1, $2, TO_TIMESTAMP($3), $4, $5, $6, $7, $8, $9, $10, $11, $12)  ON CONFLICT DO NOTHING"
 258                                         } else {
 259                                                 "INSERT INTO channel_updates (\
 260                                                         short_channel_id, \
 261                                                         timestamp, \
 262                                                         channel_flags, \
 263                                                         direction, \
 264                                                         disable, \
 265                                                         cltv_expiry_delta, \
 266                                                         htlc_minimum_msat, \
 267                                                         fee_base_msat, \
 268                                                         fee_proportional_millionths, \
 269                                                         htlc_maximum_msat, \
 270                                                         blob_signed \
 271                                                 ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)  ON CONFLICT DO NOTHING"
 272                                         };
 273
 274                                         // this may not be used outside test cfg
 275                                         let _seen_timestamp = seen_override.unwrap_or(timestamp as u32) as f64;
 276
 277                                         let _task = self.tokio_runtime.spawn(async move {
 278                                                 tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client
 279                                                         .execute(insertion_statement, &[
 280                                                                 &scid,
 281                                                                 &timestamp,
 282                                                                 #[cfg(test)]
 283                                                                         &_seen_timestamp,
 284                                                                 &(update.contents.flags as i16),
 285                                                                 &direction,
 286                                                                 &disable,
 287                                                                 &cltv_expiry_delta,
 288                                                                 &htlc_minimum_msat,
 289                                                                 &fee_base_msat,
 290                                                                 &fee_proportional_millionths,
 291                                                                 &htlc_maximum_msat,
 292                                                                 &update_signed
 293                                                         ])).await.unwrap().unwrap();
 294                                                 let mut connections_set = connections_cache_ref.lock().await;
 295                                                 connections_set.push(client);
 296                                                 limiter_ref.add_permits(1);
 297                                         });
 298                                         #[cfg(test)]
 299                                         tasks_spawned.push(_task);
 300                                 }
 301                         }
 302                 }
 303                 #[cfg(test)]
 304                 for task in tasks_spawned {
 305                         task.await.unwrap();
 306                 }
 307         }
 308
 309         fn persist_network_graph(&self) {
 310                 log_info!(self.logger, "Caching network graph…");
 311                 let cache_path = config::network_graph_cache_path();
 312                 let file = OpenOptions::new()
 313                         .create(true)
 314                         .write(true)
 315                         .truncate(true)
 316                         .open(&cache_path)
 317                         .unwrap();
 318                 self.network_graph.remove_stale_channels_and_tracking();
 319                 let mut writer = BufWriter::new(file);
 320                 self.network_graph.write(&mut writer).unwrap();
 321                 writer.flush().unwrap();
 322                 log_info!(self.logger, "Cached network graph!");
 323         }
 324 }