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

   1 use std::fs::OpenOptions;
   2 use std::io::{BufWriter, Write};
   3 use std::sync::Arc;
   4 use std::time::{Duration, 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         gossip_persistence_receiver: mpsc::Receiver<GossipMessage>,
  15         network_graph: Arc<NetworkGraph<TestLogger>>,
  16 }
  17
  18 impl GossipPersister {
  19         pub fn new(network_graph: Arc<NetworkGraph<TestLogger>>) -> (Self, mpsc::Sender<GossipMessage>) {
  20                 let (gossip_persistence_sender, gossip_persistence_receiver) =
  21                         mpsc::channel::<GossipMessage>(100);
  22                 (GossipPersister {
  23                         gossip_persistence_receiver,
  24                         network_graph
  25                 }, gossip_persistence_sender)
  26         }
  27
  28         pub(crate) async fn persist_gossip(&mut self) {
  29                 let connection_config = config::db_connection_config();
  30                 let (mut client, connection) =
  31                         connection_config.connect(NoTls).await.unwrap();
  32
  33                 tokio::spawn(async move {
  34                         if let Err(e) = connection.await {
  35                                 panic!("connection error: {}", e);
  36                         }
  37                 });
  38
  39                 {
  40                         // initialize the database
  41                         let initialization = client
  42                                 .execute(config::db_config_table_creation_query(), &[])
  43                                 .await;
  44                         if let Err(initialization_error) = initialization {
  45                                 panic!("db init error: {}", initialization_error);
  46                         }
  47
  48                         let cur_schema = client.query("SELECT db_schema FROM config WHERE id = $1", &[&1]).await.unwrap();
  49                         if !cur_schema.is_empty() {
  50                                 config::upgrade_db(cur_schema[0].get(0), &mut client).await;
  51                         }
  52
  53                         let initialization = client
  54                                 .execute(
  55                                         // TODO: figure out a way to fix the id value without Postgres complaining about
  56                                         // its value not being default
  57                                         "INSERT INTO config (id, db_schema) VALUES ($1, $2) ON CONFLICT (id) DO NOTHING",
  58                                         &[&1, &config::SCHEMA_VERSION]
  59                                 ).await;
  60                         if let Err(initialization_error) = initialization {
  61                                 panic!("db init error: {}", initialization_error);
  62                         }
  63
  64                         let initialization = client
  65                                 .execute(config::db_announcement_table_creation_query(), &[])
  66                                 .await;
  67                         if let Err(initialization_error) = initialization {
  68                                 panic!("db init error: {}", initialization_error);
  69                         }
  70
  71                         let initialization = client
  72                                 .execute(
  73                                         config::db_channel_update_table_creation_query(),
  74                                         &[],
  75                                 )
  76                                 .await;
  77                         if let Err(initialization_error) = initialization {
  78                                 panic!("db init error: {}", initialization_error);
  79                         }
  80
  81                         let initialization = client
  82                                 .batch_execute(config::db_index_creation_query())
  83                                 .await;
  84                         if let Err(initialization_error) = initialization {
  85                                 panic!("db init error: {}", initialization_error);
  86                         }
  87                 }
  88
  89                 // print log statement every minute
  90                 let mut latest_persistence_log = Instant::now() - Duration::from_secs(60);
  91                 let mut i = 0u32;
  92                 let mut latest_graph_cache_time = Instant::now();
  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 latest_persistence_log.elapsed().as_secs() >= 60 {
 100                                 println!("Persisting gossip message #{}", i);
 101                                 latest_persistence_log = Instant::now();
 102                         }
 103
 104                         // has it been ten minutes? Just cache it
 105                         if latest_graph_cache_time.elapsed().as_secs() >= 600 {
 106                                 self.persist_network_graph();
 107                                 latest_graph_cache_time = Instant::now();
 108                         }
 109
 110                         match &gossip_message {
 111                                 GossipMessage::ChannelAnnouncement(announcement) => {
 112                                         let scid = announcement.contents.short_channel_id;
 113                                         let scid_hex = hex_utils::hex_str(&scid.to_be_bytes());
 114                                         // scid is 8 bytes
 115                                         // block height is the first three bytes
 116                                         // to obtain block height, shift scid right by 5 bytes (40 bits)
 117                                         let block_height = (scid >> 5 * 8) as i32;
 118                                         let chain_hash = announcement.contents.chain_hash.as_ref();
 119                                         let chain_hash_hex = hex_utils::hex_str(chain_hash);
 120
 121                                         // start with the type prefix, which is already known a priori
 122                                         let mut announcement_signed = Vec::new();
 123                                         announcement.write(&mut announcement_signed).unwrap();
 124
 125                                         let result = client
 126                                                 .execute("INSERT INTO channel_announcements (\
 127                                                         short_channel_id, \
 128                                                         block_height, \
 129                                                         chain_hash, \
 130                                                         announcement_signed \
 131                                                 ) VALUES ($1, $2, $3, $4) ON CONFLICT (short_channel_id) DO NOTHING", &[
 132                                                         &scid_hex,
 133                                                         &block_height,
 134                                                         &chain_hash_hex,
 135                                                         &announcement_signed
 136                                                 ]).await;
 137                                         if result.is_err() {
 138                                                 panic!("error: {}", result.err().unwrap());
 139                                         }
 140                                 }
 141                                 GossipMessage::ChannelUpdate(update) => {
 142                                         let scid = update.contents.short_channel_id;
 143                                         let scid_hex = hex_utils::hex_str(&scid.to_be_bytes());
 144
 145                                         let chain_hash = update.contents.chain_hash.as_ref();
 146                                         let chain_hash_hex = hex_utils::hex_str(chain_hash);
 147
 148                                         let timestamp = update.contents.timestamp as i64;
 149
 150                                         let channel_flags = update.contents.flags as i32;
 151                                         let direction = channel_flags & 1;
 152                                         let disable = (channel_flags & 2) > 0;
 153
 154                                         let composite_index = format!("{}:{}:{}", scid_hex, timestamp, direction);
 155
 156                                         let cltv_expiry_delta = update.contents.cltv_expiry_delta as i32;
 157                                         let htlc_minimum_msat = update.contents.htlc_minimum_msat as i64;
 158                                         let fee_base_msat = update.contents.fee_base_msat as i32;
 159                                         let fee_proportional_millionths =
 160                                                 update.contents.fee_proportional_millionths as i32;
 161                                         let htlc_maximum_msat = update.contents.htlc_maximum_msat as i64;
 162
 163                                         // start with the type prefix, which is already known a priori
 164                                         let mut update_signed = Vec::new();
 165                                         update.write(&mut update_signed).unwrap();
 166
 167                                         let result = client
 168                                                 .execute("INSERT INTO channel_updates (\
 169                                                         composite_index, \
 170                                                         chain_hash, \
 171                                                         short_channel_id, \
 172                                                         timestamp, \
 173                                                         channel_flags, \
 174                                                         direction, \
 175                                                         disable, \
 176                                                         cltv_expiry_delta, \
 177                                                         htlc_minimum_msat, \
 178                                                         fee_base_msat, \
 179                                                         fee_proportional_millionths, \
 180                                                         htlc_maximum_msat, \
 181                                                         blob_signed \
 182                                                 ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)  ON CONFLICT (composite_index) DO NOTHING", &[
 183                                                         &composite_index,
 184                                                         &chain_hash_hex,
 185                                                         &scid_hex,
 186                                                         &timestamp,
 187                                                         &channel_flags,
 188                                                         &direction,
 189                                                         &disable,
 190                                                         &cltv_expiry_delta,
 191                                                         &htlc_minimum_msat,
 192                                                         &fee_base_msat,
 193                                                         &fee_proportional_millionths,
 194                                                         &htlc_maximum_msat,
 195                                                         &update_signed
 196                                                 ]).await;
 197                                         if result.is_err() {
 198                                                 panic!("error: {}", result.err().unwrap());
 199                                         }
 200                                 }
 201                         }
 202                 }
 203         }
 204
 205         fn persist_network_graph(&self) {
 206                 println!("Caching network graph…");
 207                 let cache_path = config::network_graph_cache_path();
 208                 let file = OpenOptions::new()
 209                         .create(true)
 210                         .write(true)
 211                         .truncate(true)
 212                         .open(&cache_path)
 213                         .unwrap();
 214                 self.network_graph.remove_stale_channels();
 215                 let mut writer = BufWriter::new(file);
 216                 self.network_graph.write(&mut writer).unwrap();
 217                 writer.flush().unwrap();
 218                 println!("Cached network graph!");
 219         }
 220 }