1 use std::fs::OpenOptions;
2 use std::io::BufWriter;
4 use std::time::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 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>>>,
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);
25 gossip_persistence_sender,
26 gossip_persistence_receiver,
27 server_sync_completion_sender,
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();
37 tokio::spawn(async move {
38 if let Err(e) = connection.await {
39 panic!("connection error: {}", e);
44 // initialize the database
45 let initialization = client
46 .execute(config::db_config_table_creation_query(), &[])
48 if let Err(initialization_error) = initialization {
49 panic!("db init error: {}", initialization_error);
52 let initialization = client
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]
59 if let Err(initialization_error) = initialization {
60 panic!("db init error: {}", initialization_error);
63 let initialization = client
64 .execute(config::db_announcement_table_creation_query(), &[])
66 if let Err(initialization_error) = initialization {
67 panic!("db init error: {}", initialization_error);
70 let initialization = client
72 config::db_channel_update_table_creation_query(),
76 if let Err(initialization_error) = initialization {
77 panic!("db init error: {}", initialization_error);
80 let initialization = client
81 .batch_execute(config::db_index_creation_query())
83 if let Err(initialization_error) = initialization {
84 panic!("db init error: {}", initialization_error);
88 // print log statement every 10,000 messages
89 let mut persistence_log_threshold = 10000;
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
96 while let Some(gossip_message) = &self.gossip_persistence_receiver.recv().await {
97 i += 1; // count the persisted gossip messages
99 if i == 1 || i % persistence_log_threshold == 0 {
100 println!("Persisting gossip message #{}", i);
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());
110 // initialize graph cache timer
111 latest_graph_cache_time = Some(Instant::now());
114 match &gossip_message {
115 GossipMessage::InitialSyncComplete => {
116 // signal to the server that it may now serve dynamic responses and calculate
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.");
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;
138 println!("Network graph has been cached too recently.");
140 latest_graph_cache_time = Some(Instant::now());
141 self.persist_network_graph();
144 GossipMessage::ChannelAnnouncement(announcement) => {
146 let scid = announcement.contents.short_channel_id;
147 let scid_hex = hex_utils::hex_str(&scid.to_be_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);
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();
160 .execute("INSERT INTO channel_announcements (\
164 announcement_signed \
165 ) VALUES ($1, $2, $3, $4) ON CONFLICT (short_channel_id) DO NOTHING", &[
172 panic!("error: {}", result.err().unwrap());
175 GossipMessage::ChannelUpdate(update) => {
176 let scid = update.contents.short_channel_id;
177 let scid_hex = hex_utils::hex_str(&scid.to_be_bytes());
179 let chain_hash = update.contents.chain_hash.as_ref();
180 let chain_hash_hex = hex_utils::hex_str(chain_hash);
182 let timestamp = update.contents.timestamp as i64;
184 let channel_flags = update.contents.flags as i32;
185 let direction = channel_flags & 1;
186 let disable = (channel_flags & 2) > 0;
188 let composite_index = format!("{}:{}:{}", scid_hex, timestamp, direction);
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;
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();
202 .execute("INSERT INTO channel_updates (\
213 fee_proportional_millionths, \
216 ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13) ON CONFLICT (composite_index) DO NOTHING", &[
227 &fee_proportional_millionths,
232 panic!("error: {}", result.err().unwrap());
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()
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!");