1 use std::fs::OpenOptions;
2 use std::io::{BufWriter, Write};
5 use std::time::{Duration, Instant};
6 use lightning::routing::gossip::NetworkGraph;
7 use lightning::util::logger::Logger;
8 use lightning::util::ser::Writeable;
10 use tokio_postgres::NoTls;
13 use crate::types::GossipMessage;
15 const POSTGRES_INSERT_TIMEOUT: Duration = Duration::from_secs(15);
17 pub(crate) struct GossipPersister<L: Deref> where L::Target: Logger {
18 gossip_persistence_receiver: mpsc::Receiver<GossipMessage>,
19 network_graph: Arc<NetworkGraph<L>>,
22 impl<L: Deref> GossipPersister<L> where L::Target: Logger {
23 pub fn new(network_graph: Arc<NetworkGraph<L>>) -> (Self, mpsc::Sender<GossipMessage>) {
24 let (gossip_persistence_sender, gossip_persistence_receiver) =
25 mpsc::channel::<GossipMessage>(100);
27 gossip_persistence_receiver,
29 }, gossip_persistence_sender)
32 pub(crate) async fn persist_gossip(&mut self) {
33 let connection_config = config::db_connection_config();
34 let (mut 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 cur_schema = client.query("SELECT db_schema FROM config WHERE id = $1", &[&1]).await.unwrap();
53 if !cur_schema.is_empty() {
54 config::upgrade_db(cur_schema[0].get(0), &mut client).await;
57 let initialization = client
59 // TODO: figure out a way to fix the id value without Postgres complaining about
60 // its value not being default
61 "INSERT INTO config (id, db_schema) VALUES ($1, $2) ON CONFLICT (id) DO NOTHING",
62 &[&1, &config::SCHEMA_VERSION]
64 if let Err(initialization_error) = initialization {
65 panic!("db init error: {}", initialization_error);
68 let initialization = client
69 .execute(config::db_announcement_table_creation_query(), &[])
71 if let Err(initialization_error) = initialization {
72 panic!("db init error: {}", initialization_error);
75 let initialization = client
77 config::db_channel_update_table_creation_query(),
81 if let Err(initialization_error) = initialization {
82 panic!("db init error: {}", initialization_error);
85 let initialization = client
86 .batch_execute(config::db_index_creation_query())
88 if let Err(initialization_error) = initialization {
89 panic!("db init error: {}", initialization_error);
93 // print log statement every minute
94 let mut latest_persistence_log = Instant::now() - Duration::from_secs(60);
96 let mut latest_graph_cache_time = Instant::now();
97 // TODO: it would be nice to have some sort of timeout here so after 10 seconds of
98 // inactivity, some sort of message could be broadcast signaling the activation of request
100 while let Some(gossip_message) = &self.gossip_persistence_receiver.recv().await {
101 i += 1; // count the persisted gossip messages
103 if latest_persistence_log.elapsed().as_secs() >= 60 {
104 println!("Persisting gossip message #{}", i);
105 latest_persistence_log = Instant::now();
108 // has it been ten minutes? Just cache it
109 if latest_graph_cache_time.elapsed().as_secs() >= 600 {
110 self.persist_network_graph();
111 latest_graph_cache_time = Instant::now();
114 match &gossip_message {
115 GossipMessage::ChannelAnnouncement(announcement) => {
116 let scid = announcement.contents.short_channel_id as i64;
118 // start with the type prefix, which is already known a priori
119 let mut announcement_signed = Vec::new();
120 announcement.write(&mut announcement_signed).unwrap();
122 tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client
123 .execute("INSERT INTO channel_announcements (\
125 announcement_signed \
126 ) VALUES ($1, $2) ON CONFLICT (short_channel_id) DO NOTHING", &[
129 ])).await.unwrap().unwrap();
131 GossipMessage::ChannelUpdate(update) => {
132 let scid = update.contents.short_channel_id as i64;
134 let timestamp = update.contents.timestamp as i64;
136 let direction = (update.contents.flags & 1) == 1;
137 let disable = (update.contents.flags & 2) > 0;
139 let cltv_expiry_delta = update.contents.cltv_expiry_delta as i32;
140 let htlc_minimum_msat = update.contents.htlc_minimum_msat as i64;
141 let fee_base_msat = update.contents.fee_base_msat as i32;
142 let fee_proportional_millionths =
143 update.contents.fee_proportional_millionths as i32;
144 let htlc_maximum_msat = update.contents.htlc_maximum_msat as i64;
146 // start with the type prefix, which is already known a priori
147 let mut update_signed = Vec::new();
148 update.write(&mut update_signed).unwrap();
150 tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client
151 .execute("INSERT INTO channel_updates (\
160 fee_proportional_millionths, \
163 ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) ON CONFLICT DO NOTHING", &[
166 &(update.contents.flags as i16),
172 &fee_proportional_millionths,
175 ])).await.unwrap().unwrap();
181 fn persist_network_graph(&self) {
182 println!("Caching network graph…");
183 let cache_path = config::network_graph_cache_path();
184 let file = OpenOptions::new()
190 self.network_graph.remove_stale_channels_and_tracking();
191 let mut writer = BufWriter::new(file);
192 self.network_graph.write(&mut writer).unwrap();
193 writer.flush().unwrap();
194 println!("Cached network graph!");