1 use std::fs::OpenOptions;
2 use std::io::{BufWriter, Write};
4 use std::time::{Duration, Instant};
5 use lightning::routing::gossip::NetworkGraph;
6 use lightning::util::ser::Writeable;
8 use tokio_postgres::NoTls;
10 use crate::{config, hex_utils, TestLogger};
11 use crate::types::GossipMessage;
13 pub(crate) struct GossipPersister {
14 gossip_persistence_receiver: mpsc::Receiver<GossipMessage>,
15 network_graph: Arc<NetworkGraph<TestLogger>>,
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);
23 gossip_persistence_receiver,
25 }, gossip_persistence_sender)
28 pub(crate) async fn persist_gossip(&mut self) {
29 let connection_config = config::db_connection_config();
30 let (client, connection) =
31 connection_config.connect(NoTls).await.unwrap();
33 tokio::spawn(async move {
34 if let Err(e) = connection.await {
35 panic!("connection error: {}", e);
40 // initialize the database
41 let initialization = client
42 .execute(config::db_config_table_creation_query(), &[])
44 if let Err(initialization_error) = initialization {
45 panic!("db init error: {}", initialization_error);
48 let initialization = client
50 // TODO: figure out a way to fix the id value without Postgres complaining about
51 // its value not being default
52 "INSERT INTO config (id, db_schema) VALUES ($1, $2) ON CONFLICT (id) DO NOTHING",
53 &[&1, &config::SCHEMA_VERSION]
55 if let Err(initialization_error) = initialization {
56 panic!("db init error: {}", initialization_error);
59 let initialization = client
60 .execute(config::db_announcement_table_creation_query(), &[])
62 if let Err(initialization_error) = initialization {
63 panic!("db init error: {}", initialization_error);
66 let initialization = client
68 config::db_channel_update_table_creation_query(),
72 if let Err(initialization_error) = initialization {
73 panic!("db init error: {}", initialization_error);
76 let initialization = client
77 .batch_execute(config::db_index_creation_query())
79 if let Err(initialization_error) = initialization {
80 panic!("db init error: {}", initialization_error);
84 // print log statement every minute
85 let mut latest_persistence_log = Instant::now() - Duration::from_secs(60);
87 let mut latest_graph_cache_time = Instant::now();
88 // TODO: it would be nice to have some sort of timeout here so after 10 seconds of
89 // inactivity, some sort of message could be broadcast signaling the activation of request
91 while let Some(gossip_message) = &self.gossip_persistence_receiver.recv().await {
92 i += 1; // count the persisted gossip messages
94 if latest_persistence_log.elapsed().as_secs() >= 60 {
95 println!("Persisting gossip message #{}", i);
96 latest_persistence_log = Instant::now();
99 // has it been ten minutes? Just cache it
100 if latest_graph_cache_time.elapsed().as_secs() >= 600 {
101 self.persist_network_graph();
102 latest_graph_cache_time = Instant::now();
105 match &gossip_message {
106 GossipMessage::ChannelAnnouncement(announcement) => {
107 let scid = announcement.contents.short_channel_id;
108 let scid_hex = hex_utils::hex_str(&scid.to_be_bytes());
110 // block height is the first three bytes
111 // to obtain block height, shift scid right by 5 bytes (40 bits)
112 let block_height = (scid >> 5 * 8) as i32;
113 let chain_hash = announcement.contents.chain_hash.as_ref();
114 let chain_hash_hex = hex_utils::hex_str(chain_hash);
116 // start with the type prefix, which is already known a priori
117 let mut announcement_signed = Vec::new(); // vec![1, 0];
118 announcement.write(&mut announcement_signed).unwrap();
121 .execute("INSERT INTO channel_announcements (\
125 announcement_signed \
126 ) VALUES ($1, $2, $3, $4) ON CONFLICT (short_channel_id) DO NOTHING", &[
133 panic!("error: {}", result.err().unwrap());
136 GossipMessage::ChannelUpdate(update) => {
137 let scid = update.contents.short_channel_id;
138 let scid_hex = hex_utils::hex_str(&scid.to_be_bytes());
140 let chain_hash = update.contents.chain_hash.as_ref();
141 let chain_hash_hex = hex_utils::hex_str(chain_hash);
143 let timestamp = update.contents.timestamp as i64;
145 let channel_flags = update.contents.flags as i32;
146 let direction = channel_flags & 1;
147 let disable = (channel_flags & 2) > 0;
149 let composite_index = format!("{}:{}:{}", scid_hex, timestamp, direction);
151 let cltv_expiry_delta = update.contents.cltv_expiry_delta as i32;
152 let htlc_minimum_msat = update.contents.htlc_minimum_msat as i64;
153 let fee_base_msat = update.contents.fee_base_msat as i32;
154 let fee_proportional_millionths =
155 update.contents.fee_proportional_millionths as i32;
156 let htlc_maximum_msat = update.contents.htlc_maximum_msat as i64;
158 // start with the type prefix, which is already known a priori
159 let mut update_signed = Vec::new(); // vec![1, 2];
160 update.write(&mut update_signed).unwrap();
163 .execute("INSERT INTO channel_updates (\
174 fee_proportional_millionths, \
177 ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13) ON CONFLICT (composite_index) DO NOTHING", &[
188 &fee_proportional_millionths,
193 panic!("error: {}", result.err().unwrap());
200 fn persist_network_graph(&self) {
201 println!("Caching network graph…");
202 let cache_path = config::network_graph_cache_path();
203 let file = OpenOptions::new()
209 self.network_graph.remove_stale_channels();
210 let mut writer = BufWriter::new(file);
211 self.network_graph.write(&mut writer).unwrap();
212 writer.flush().unwrap();
213 println!("Cached network graph!");