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

   1 use std::fs::OpenOptions;
   2 use std::io::BufWriter;
   3 use std::sync::Arc;
   4 use std::time::Instant;
   5 use lightning::routing::gossip::NetworkGraph;
   6 use lightning::util::ser::Writeable;
   7 use tokio::sync::mpsc;
   8 use tokio_postgres::NoTls;
   9
  10 use crate::{config, hex_utils, TestLogger};
  11 use crate::types::GossipMessage;
  12
  13 pub(crate) struct GossipPersister {
  14         pub(crate) gossip_persistence_sender: mpsc::Sender<GossipMessage>,
  15         gossip_persistence_receiver: mpsc::Receiver<GossipMessage>,
  16         server_sync_completion_sender: mpsc::Sender<()>,
  17         network_graph: Arc<NetworkGraph<Arc<TestLogger>>>,
  18 }
  19
  20 impl GossipPersister {
  21         pub fn new(server_sync_completion_sender: mpsc::Sender<()>, network_graph: Arc<NetworkGraph<Arc<TestLogger>>>) -> Self {
  22                 let (gossip_persistence_sender, gossip_persistence_receiver) =
  23                         mpsc::channel::<GossipMessage>(100);
  24                 GossipPersister {
  25                         gossip_persistence_sender,
  26                         gossip_persistence_receiver,
  27                         server_sync_completion_sender,
  28                         network_graph
  29                 }
  30         }
  31
  32         pub(crate) async fn persist_gossip(&mut self) {
  33                 let connection_config = config::db_connection_config();
  34                 let (client, connection) =
  35                         connection_config.connect(NoTls).await.unwrap();
  36
  37                 tokio::spawn(async move {
  38                         if let Err(e) = connection.await {
  39                                 panic!("connection error: {}", e);
  40                         }
  41                 });
  42
  43                 {
  44                         // initialize the database
  45                         let initialization = client
  46                                 .execute(config::db_config_table_creation_query(), &[])
  47                                 .await;
  48                         if let Err(initialization_error) = initialization {
  49                                 panic!("db init error: {}", initialization_error);
  50                         }
  51
  52                         let initialization = client
  53                                 .execute(
  54                                         // TODO: figure out a way to fix the id value without Postgres complaining about
  55                                         // its value not being default
  56                                         "INSERT INTO config (id, db_schema) VALUES ($1, $2) ON CONFLICT (id) DO NOTHING",
  57                                         &[&1, &config::SCHEMA_VERSION]
  58                                 ).await;
  59                         if let Err(initialization_error) = initialization {
  60                                 panic!("db init error: {}", initialization_error);
  61                         }
  62
  63                         let initialization = client
  64                                 .execute(config::db_announcement_table_creation_query(), &[])
  65                                 .await;
  66                         if let Err(initialization_error) = initialization {
  67                                 panic!("db init error: {}", initialization_error);
  68                         }
  69
  70                         let initialization = client
  71                                 .execute(
  72                                         config::db_channel_update_table_creation_query(),
  73                                         &[],
  74                                 )
  75                                 .await;
  76                         if let Err(initialization_error) = initialization {
  77                                 panic!("db init error: {}", initialization_error);
  78                         }
  79
  80                         let initialization = client
  81                                 .batch_execute(config::db_index_creation_query())
  82                                 .await;
  83                         if let Err(initialization_error) = initialization {
  84                                 panic!("db init error: {}", initialization_error);
  85                         }
  86                 }
  87
  88                 // print log statement every 10,000 messages
  89                 let mut persistence_log_threshold = 10000;
  90                 let mut i = 0u32;
  91                 let mut server_sync_completion_sent = false;
  92                 let mut latest_graph_cache_time: Option<Instant> = None;
  93                 // TODO: it would be nice to have some sort of timeout here so after 10 seconds of
  94                 // inactivity, some sort of message could be broadcast signaling the activation of request
  95                 // processing
  96                 while let Some(gossip_message) = &self.gossip_persistence_receiver.recv().await {
  97                         i += 1; // count the persisted gossip messages
  98
  99                         if i == 1 || i % persistence_log_threshold == 0 {
 100                                 println!("Persisting gossip message #{}", i);
 101                         }
 102
 103                         if let Some(last_cache_time) = latest_graph_cache_time {
 104                                 // has it been ten minutes? Just cache it
 105                                 if last_cache_time.elapsed().as_secs() >= 600 {
 106                                         self.persist_network_graph();
 107                                         latest_graph_cache_time = Some(Instant::now());
 108                                 }
 109                         } else {
 110                                 // initialize graph cache timer
 111                                 latest_graph_cache_time = Some(Instant::now());
 112                         }
 113
 114                         match &gossip_message {
 115                                 GossipMessage::InitialSyncComplete => {
 116                                         // signal to the server that it may now serve dynamic responses and calculate
 117                                         // snapshots
 118                                         // we take this detour through the persister to ensure that all previous
 119                                         // messages have already been persisted to the database
 120                                         println!("Persister caught up with gossip!");
 121                                         i -= 1; // this wasn't an actual gossip message that needed persisting
 122                                         persistence_log_threshold = 50;
 123                                         if !server_sync_completion_sent {
 124                                                 server_sync_completion_sent = true;
 125                                                 self.server_sync_completion_sender.send(()).await.unwrap();
 126                                                 println!("Server has been notified of persistence completion.");
 127                                         }
 128
 129                                         // now, cache the persisted network graph
 130                                         // also persist the network graph here
 131                                         let mut too_soon = false;
 132                                         if let Some(latest_graph_cache_time) = latest_graph_cache_time {
 133                                                 let time_since_last_cached = latest_graph_cache_time.elapsed().as_secs();
 134                                                 // don't cache more frequently than every 2 minutes
 135                                                 too_soon = time_since_last_cached < 120;
 136                                         }
 137                                         if too_soon {
 138                                                 println!("Network graph has been cached too recently.");
 139                                         }else {
 140                                                 latest_graph_cache_time = Some(Instant::now());
 141                                                 self.persist_network_graph();
 142                                         }
 143                                 }
 144                                 GossipMessage::ChannelAnnouncement(announcement) => {
 145
 146                                         let scid = announcement.contents.short_channel_id;
 147                                         let scid_hex = hex_utils::hex_str(&scid.to_be_bytes());
 148                                         // scid is 8 bytes
 149                                         // block height is the first three bytes
 150                                         // to obtain block height, shift scid right by 5 bytes (40 bits)
 151                                         let block_height = (scid >> 5 * 8) as i32;
 152                                         let chain_hash = announcement.contents.chain_hash.as_ref();
 153                                         let chain_hash_hex = hex_utils::hex_str(chain_hash);
 154
 155                                         // start with the type prefix, which is already known a priori
 156                                         let mut announcement_signed = Vec::new(); // vec![1, 0];
 157                                         announcement.write(&mut announcement_signed).unwrap();
 158
 159                                         let result = client
 160                                                 .execute("INSERT INTO channel_announcements (\
 161                                                         short_channel_id, \
 162                                                         block_height, \
 163                                                         chain_hash, \
 164                                                         announcement_signed \
 165                                                 ) VALUES ($1, $2, $3, $4) ON CONFLICT (short_channel_id) DO NOTHING", &[
 166                                                         &scid_hex,
 167                                                         &block_height,
 168                                                         &chain_hash_hex,
 169                                                         &announcement_signed
 170                                                 ]).await;
 171                                         if result.is_err() {
 172                                                 panic!("error: {}", result.err().unwrap());
 173                                         }
 174                                 }
 175                                 GossipMessage::ChannelUpdate(update) => {
 176                                         let scid = update.contents.short_channel_id;
 177                                         let scid_hex = hex_utils::hex_str(&scid.to_be_bytes());
 178
 179                                         let chain_hash = update.contents.chain_hash.as_ref();
 180                                         let chain_hash_hex = hex_utils::hex_str(chain_hash);
 181
 182                                         let timestamp = update.contents.timestamp as i64;
 183
 184                                         let channel_flags = update.contents.flags as i32;
 185                                         let direction = channel_flags & 1;
 186                                         let disable = (channel_flags & 2) > 0;
 187
 188                                         let composite_index = format!("{}:{}:{}", scid_hex, timestamp, direction);
 189
 190                                         let cltv_expiry_delta = update.contents.cltv_expiry_delta as i32;
 191                                         let htlc_minimum_msat = update.contents.htlc_minimum_msat as i64;
 192                                         let fee_base_msat = update.contents.fee_base_msat as i32;
 193                                         let fee_proportional_millionths =
 194                                                 update.contents.fee_proportional_millionths as i32;
 195                                         let htlc_maximum_msat = update.contents.htlc_maximum_msat as i64;
 196
 197                                         // start with the type prefix, which is already known a priori
 198                                         let mut update_signed = Vec::new(); // vec![1, 2];
 199                                         update.write(&mut update_signed).unwrap();
 200
 201                                         let result = client
 202                                                 .execute("INSERT INTO channel_updates (\
 203                                                         composite_index, \
 204                                                         chain_hash, \
 205                                                         short_channel_id, \
 206                                                         timestamp, \
 207                                                         channel_flags, \
 208                                                         direction, \
 209                                                         disable, \
 210                                                         cltv_expiry_delta, \
 211                                                         htlc_minimum_msat, \
 212                                                         fee_base_msat, \
 213                                                         fee_proportional_millionths, \
 214                                                         htlc_maximum_msat, \
 215                                                         blob_signed \
 216                                                 ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)  ON CONFLICT (composite_index) DO NOTHING", &[
 217                                                         &composite_index,
 218                                                         &chain_hash_hex,
 219                                                         &scid_hex,
 220                                                         &timestamp,
 221                                                         &channel_flags,
 222                                                         &direction,
 223                                                         &disable,
 224                                                         &cltv_expiry_delta,
 225                                                         &htlc_minimum_msat,
 226                                                         &fee_base_msat,
 227                                                         &fee_proportional_millionths,
 228                                                         &htlc_maximum_msat,
 229                                                         &update_signed
 230                                                 ]).await;
 231                                         if result.is_err() {
 232                                                 panic!("error: {}", result.err().unwrap());
 233                                         }
 234                                 }
 235                         }
 236                 }
 237         }
 238
 239         fn persist_network_graph(&self) {
 240                 println!("Caching network graph…");
 241                 let cache_path = config::network_graph_cache_path();
 242                 let file = OpenOptions::new()
 243                         .create(true)
 244                         .write(true)
 245                         .truncate(true)
 246                         .open(&cache_path)
 247                         .unwrap();
 248                 self.network_graph.remove_stale_channels();
 249                 let mut writer = BufWriter::new(file);
 250                 self.network_graph.write(&mut writer).unwrap();
 251                 println!("Cached network graph!");
 252         }
 253 }