1 use std::fs::OpenOptions;
2 use std::io::{BufWriter, Write};
5 use std::time::{Duration, Instant};
6 use lightning::log_info;
7 use lightning::routing::gossip::NetworkGraph;
8 use lightning::util::logger::Logger;
9 use lightning::util::ser::Writeable;
10 use tokio::runtime::Runtime;
11 use tokio::sync::{mpsc, Mutex, Semaphore};
14 use crate::types::GossipMessage;
16 const POSTGRES_INSERT_TIMEOUT: Duration = Duration::from_secs(15);
17 const INSERT_PARALELLISM: usize = 16;
19 pub(crate) struct GossipPersister<L: Deref> where L::Target: Logger {
20 gossip_persistence_receiver: mpsc::Receiver<GossipMessage>,
21 network_graph: Arc<NetworkGraph<L>>,
22 tokio_runtime: Runtime,
26 impl<L: Deref> GossipPersister<L> where L::Target: Logger {
27 pub fn new(network_graph: Arc<NetworkGraph<L>>, logger: L) -> (Self, mpsc::Sender<GossipMessage>) {
28 let (gossip_persistence_sender, gossip_persistence_receiver) =
29 mpsc::channel::<GossipMessage>(100);
30 let runtime = Runtime::new().unwrap();
32 gossip_persistence_receiver,
34 tokio_runtime: runtime,
36 }, gossip_persistence_sender)
39 pub(crate) async fn persist_gossip(&mut self) {
40 { // initialize the database
41 // this client instance is only used once
42 let mut client = crate::connect_to_db().await;
44 let initialization = client
45 .execute(config::db_config_table_creation_query(), &[])
47 if let Err(initialization_error) = initialization {
48 panic!("db init error: {}", initialization_error);
51 let cur_schema = client.query("SELECT db_schema FROM config WHERE id = $1", &[&1]).await.unwrap();
52 if !cur_schema.is_empty() {
53 config::upgrade_db(cur_schema[0].get(0), &mut client).await;
56 let preparation = client.execute("set time zone UTC", &[]).await;
57 if let Err(preparation_error) = preparation {
58 panic!("db preparation error: {}", preparation_error);
61 let initialization = client
63 // TODO: figure out a way to fix the id value without Postgres complaining about
64 // its value not being default
65 "INSERT INTO config (id, db_schema) VALUES ($1, $2) ON CONFLICT (id) DO NOTHING",
66 &[&1, &config::SCHEMA_VERSION]
68 if let Err(initialization_error) = initialization {
69 panic!("db init error: {}", initialization_error);
72 let table_creation_queries = [
73 config::db_announcement_table_creation_query(),
74 config::db_channel_update_table_creation_query(),
75 config::db_channel_update_table_creation_query(),
76 config::db_node_announcement_table_creation_query()
79 for current_table_creation_query in table_creation_queries {
80 let initialization = client
81 .execute(current_table_creation_query, &[])
83 if let Err(initialization_error) = initialization {
84 panic!("db init error: {}", initialization_error);
88 let initialization = client
89 .batch_execute(config::db_index_creation_query())
91 if let Err(initialization_error) = initialization {
92 panic!("db init error: {}", initialization_error);
96 // print log statement every minute
97 let mut latest_persistence_log = Instant::now() - Duration::from_secs(60);
99 let mut latest_graph_cache_time = Instant::now();
100 let insert_limiter = Arc::new(Semaphore::new(INSERT_PARALELLISM));
101 let connections_cache = Arc::new(Mutex::new(Vec::with_capacity(INSERT_PARALELLISM)));
103 let mut tasks_spawned = Vec::new();
104 // TODO: it would be nice to have some sort of timeout here so after 10 seconds of
105 // inactivity, some sort of message could be broadcast signaling the activation of request
107 while let Some(gossip_message) = self.gossip_persistence_receiver.recv().await {
108 i += 1; // count the persisted gossip messages
110 if latest_persistence_log.elapsed().as_secs() >= 60 {
111 log_info!(self.logger, "Persisting gossip message #{}", i);
112 latest_persistence_log = Instant::now();
115 // has it been ten minutes? Just cache it
116 if latest_graph_cache_time.elapsed().as_secs() >= 600 {
117 self.persist_network_graph();
118 latest_graph_cache_time = Instant::now();
120 insert_limiter.acquire().await.unwrap().forget();
122 let limiter_ref = Arc::clone(&insert_limiter);
124 let mut connections_set = connections_cache.lock().await;
125 let client = if connections_set.is_empty() {
126 crate::connect_to_db().await
128 connections_set.pop().unwrap()
133 let connections_cache_ref = Arc::clone(&connections_cache);
134 match gossip_message {
135 GossipMessage::NodeAnnouncement(announcement, seen_override) => {
136 let public_key_hex = announcement.contents.node_id.to_string();
138 let mut announcement_signed = Vec::new();
139 announcement.write(&mut announcement_signed).unwrap();
141 let features = announcement.contents.features.encode();
142 let timestamp = announcement.contents.timestamp as i64;
144 let mut serialized_addresses = Vec::new();
145 announcement.contents.addresses.write(&mut serialized_addresses).unwrap();
147 let _task = self.tokio_runtime.spawn(async move {
148 if cfg!(test) && seen_override.is_some() {
149 tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client
150 .execute("INSERT INTO node_announcements (\
155 announcement_signed, \
157 ) VALUES ($1, $2, $3, $4, $5, TO_TIMESTAMP($6))", &[
160 &serialized_addresses,
162 &announcement_signed,
163 &(seen_override.unwrap() as f64)
164 ])).await.unwrap().unwrap();
166 tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client
167 .execute("INSERT INTO node_announcements (\
172 announcement_signed \
173 ) VALUES ($1, $2, $3, $4, $5)", &[
176 &serialized_addresses,
178 &announcement_signed,
179 ])).await.unwrap().unwrap();
181 let mut connections_set = connections_cache_ref.lock().await;
182 connections_set.push(client);
183 limiter_ref.add_permits(1);
186 tasks_spawned.push(_task);
188 GossipMessage::ChannelAnnouncement(announcement, seen_override) => {
189 let scid = announcement.contents.short_channel_id as i64;
191 // start with the type prefix, which is already known a priori
192 let mut announcement_signed = Vec::new();
193 announcement.write(&mut announcement_signed).unwrap();
195 let _task = self.tokio_runtime.spawn(async move {
196 if cfg!(test) && seen_override.is_some() {
197 tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client
198 .execute("INSERT INTO channel_announcements (\
200 announcement_signed, \
202 ) VALUES ($1, $2, TO_TIMESTAMP($3)) ON CONFLICT (short_channel_id) DO NOTHING", &[
204 &announcement_signed,
205 &(seen_override.unwrap() as f64)
206 ])).await.unwrap().unwrap();
208 tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client
209 .execute("INSERT INTO channel_announcements (\
211 announcement_signed \
212 ) VALUES ($1, $2) ON CONFLICT (short_channel_id) DO NOTHING", &[
215 ])).await.unwrap().unwrap();
217 let mut connections_set = connections_cache_ref.lock().await;
218 connections_set.push(client);
219 limiter_ref.add_permits(1);
222 tasks_spawned.push(_task);
224 GossipMessage::ChannelUpdate(update, seen_override) => {
225 let scid = update.contents.short_channel_id as i64;
227 let timestamp = update.contents.timestamp as i64;
229 let direction = (update.contents.flags & 1) == 1;
230 let disable = (update.contents.flags & 2) > 0;
232 let cltv_expiry_delta = update.contents.cltv_expiry_delta as i32;
233 let htlc_minimum_msat = update.contents.htlc_minimum_msat as i64;
234 let fee_base_msat = update.contents.fee_base_msat as i32;
235 let fee_proportional_millionths =
236 update.contents.fee_proportional_millionths as i32;
237 let htlc_maximum_msat = update.contents.htlc_maximum_msat as i64;
239 // start with the type prefix, which is already known a priori
240 let mut update_signed = Vec::new();
241 update.write(&mut update_signed).unwrap();
243 let insertion_statement = if cfg!(test) {
244 "INSERT INTO channel_updates (\
254 fee_proportional_millionths, \
257 ) VALUES ($1, $2, TO_TIMESTAMP($3), $4, $5, $6, $7, $8, $9, $10, $11, $12) ON CONFLICT DO NOTHING"
259 "INSERT INTO channel_updates (\
268 fee_proportional_millionths, \
271 ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) ON CONFLICT DO NOTHING"
274 // this may not be used outside test cfg
275 let _seen_timestamp = seen_override.unwrap_or(timestamp as u32) as f64;
277 let _task = self.tokio_runtime.spawn(async move {
278 tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client
279 .execute(insertion_statement, &[
284 &(update.contents.flags as i16),
290 &fee_proportional_millionths,
293 ])).await.unwrap().unwrap();
294 let mut connections_set = connections_cache_ref.lock().await;
295 connections_set.push(client);
296 limiter_ref.add_permits(1);
299 tasks_spawned.push(_task);
304 for task in tasks_spawned {
309 fn persist_network_graph(&self) {
310 log_info!(self.logger, "Caching network graph…");
311 let cache_path = config::network_graph_cache_path();
312 let file = OpenOptions::new()
318 self.network_graph.remove_stale_channels_and_tracking();
319 let mut writer = BufWriter::new(file);
320 self.network_graph.write(&mut writer).unwrap();
321 writer.flush().unwrap();
322 log_info!(self.logger, "Cached network graph!");