1 use std::fs::OpenOptions;
2 use std::io::{BufWriter, Write};
6 use std::time::{Duration, Instant};
7 use lightning::log_info;
8 use lightning::routing::gossip::NetworkGraph;
9 use lightning::util::logger::Logger;
10 use lightning::util::ser::Writeable;
11 use tokio::runtime::Runtime;
12 use tokio::sync::{mpsc, Mutex, Semaphore};
15 use crate::types::GossipMessage;
17 const POSTGRES_INSERT_TIMEOUT: Duration = Duration::from_secs(15);
18 const INSERT_PARALELLISM: usize = 16;
20 pub(crate) struct GossipPersister<L: Deref> where L::Target: Logger {
21 gossip_persistence_receiver: mpsc::Receiver<GossipMessage>,
22 network_graph: Arc<NetworkGraph<L>>,
23 tokio_runtime: Runtime,
27 impl<L: Deref> GossipPersister<L> where L::Target: Logger {
28 pub fn new(network_graph: Arc<NetworkGraph<L>>, logger: L) -> (Self, mpsc::Sender<GossipMessage>) {
29 let (gossip_persistence_sender, gossip_persistence_receiver) =
30 mpsc::channel::<GossipMessage>(100);
31 let runtime = Runtime::new().unwrap();
33 gossip_persistence_receiver,
35 tokio_runtime: runtime,
37 }, gossip_persistence_sender)
40 pub(crate) async fn persist_gossip(&mut self) {
41 let mut client = crate::connect_to_db().await;
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 preparation = client.execute("set time zone UTC", &[]).await;
58 if let Err(preparation_error) = preparation {
59 panic!("db preparation error: {}", preparation_error);
62 let initialization = client
64 // TODO: figure out a way to fix the id value without Postgres complaining about
65 // its value not being default
66 "INSERT INTO config (id, db_schema) VALUES ($1, $2) ON CONFLICT (id) DO NOTHING",
67 &[&1, &config::SCHEMA_VERSION]
69 if let Err(initialization_error) = initialization {
70 panic!("db init error: {}", initialization_error);
73 let initialization = client
74 .execute(config::db_announcement_table_creation_query(), &[])
76 if let Err(initialization_error) = initialization {
77 panic!("db init error: {}", initialization_error);
80 let initialization = client
82 config::db_channel_update_table_creation_query(),
86 if let Err(initialization_error) = initialization {
87 panic!("db init error: {}", initialization_error);
90 let initialization = client
91 .batch_execute(config::db_index_creation_query())
93 if let Err(initialization_error) = initialization {
94 panic!("db init error: {}", initialization_error);
98 // print log statement every minute
99 let mut latest_persistence_log = Instant::now() - Duration::from_secs(60);
101 let mut latest_graph_cache_time = Instant::now();
102 let insert_limiter = Arc::new(Semaphore::new(INSERT_PARALELLISM));
103 let connections_cache = Arc::new(Mutex::new(Vec::with_capacity(INSERT_PARALELLISM)));
105 let mut tasks_spawned = Vec::new();
106 // TODO: it would be nice to have some sort of timeout here so after 10 seconds of
107 // inactivity, some sort of message could be broadcast signaling the activation of request
109 while let Some(gossip_message) = self.gossip_persistence_receiver.recv().await {
110 i += 1; // count the persisted gossip messages
112 if latest_persistence_log.elapsed().as_secs() >= 60 {
113 log_info!(self.logger, "Persisting gossip message #{}", i);
114 latest_persistence_log = Instant::now();
117 // has it been ten minutes? Just cache it
118 if latest_graph_cache_time.elapsed().as_secs() >= 600 {
119 self.persist_network_graph();
120 latest_graph_cache_time = Instant::now();
122 insert_limiter.acquire().await.unwrap().forget();
124 let limiter_ref = Arc::clone(&insert_limiter);
125 let connections_cache_ref = Arc::clone(&connections_cache);
126 match gossip_message {
127 GossipMessage::ChannelAnnouncement(announcement, seen_override) => {
128 let scid = announcement.contents.short_channel_id as i64;
130 // start with the type prefix, which is already known a priori
131 let mut announcement_signed = Vec::new();
132 announcement.write(&mut announcement_signed).unwrap();
134 let _task = self.tokio_runtime.spawn(async move {
137 let mut connections_set = connections_cache_ref.lock().await;
138 if connections_set.is_empty() {
139 mem::drop(connections_set);
140 client = crate::connect_to_db().await;
142 client = connections_set.pop().unwrap();
145 if cfg!(test) && seen_override.is_some() {
146 tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client
147 .execute("INSERT INTO channel_announcements (\
149 announcement_signed, \
151 ) VALUES ($1, $2, TO_TIMESTAMP($3)) ON CONFLICT (short_channel_id) DO NOTHING", &[
153 &announcement_signed,
154 &(seen_override.unwrap() as f64)
155 ])).await.unwrap().unwrap();
157 tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client
158 .execute("INSERT INTO channel_announcements (\
160 announcement_signed \
161 ) VALUES ($1, $2) ON CONFLICT (short_channel_id) DO NOTHING", &[
164 ])).await.unwrap().unwrap();
166 let mut connections_set = connections_cache_ref.lock().await;
167 connections_set.push(client);
168 limiter_ref.add_permits(1);
171 tasks_spawned.push(_task);
173 GossipMessage::ChannelUpdate(update, seen_override) => {
174 let scid = update.contents.short_channel_id as i64;
176 let timestamp = update.contents.timestamp as i64;
178 let direction = (update.contents.flags & 1) == 1;
179 let disable = (update.contents.flags & 2) > 0;
181 let cltv_expiry_delta = update.contents.cltv_expiry_delta as i32;
182 let htlc_minimum_msat = update.contents.htlc_minimum_msat as i64;
183 let fee_base_msat = update.contents.fee_base_msat as i32;
184 let fee_proportional_millionths =
185 update.contents.fee_proportional_millionths as i32;
186 let htlc_maximum_msat = update.contents.htlc_maximum_msat as i64;
188 // start with the type prefix, which is already known a priori
189 let mut update_signed = Vec::new();
190 update.write(&mut update_signed).unwrap();
192 let insertion_statement = if cfg!(test) {
193 "INSERT INTO channel_updates (\
203 fee_proportional_millionths, \
206 ) VALUES ($1, $2, TO_TIMESTAMP($3), $4, $5, $6, $7, $8, $9, $10, $11, $12) ON CONFLICT DO NOTHING"
208 "INSERT INTO channel_updates (\
217 fee_proportional_millionths, \
220 ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) ON CONFLICT DO NOTHING"
223 // this may not be used outside test cfg
224 let _seen_timestamp = seen_override.unwrap_or(timestamp as u32) as f64;
226 let _task = self.tokio_runtime.spawn(async move {
229 let mut connections_set = connections_cache_ref.lock().await;
230 if connections_set.is_empty() {
231 mem::drop(connections_set);
232 client = crate::connect_to_db().await;
234 client = connections_set.pop().unwrap();
237 tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client
238 .execute(insertion_statement, &[
243 &(update.contents.flags as i16),
249 &fee_proportional_millionths,
252 ])).await.unwrap().unwrap();
253 let mut connections_set = connections_cache_ref.lock().await;
254 connections_set.push(client);
255 limiter_ref.add_permits(1);
258 tasks_spawned.push(_task);
263 for task in tasks_spawned {
268 fn persist_network_graph(&self) {
269 log_info!(self.logger, "Caching network graph…");
270 let cache_path = config::network_graph_cache_path();
271 let file = OpenOptions::new()
277 self.network_graph.remove_stale_channels_and_tracking();
278 let mut writer = BufWriter::new(file);
279 self.network_graph.write(&mut writer).unwrap();
280 writer.flush().unwrap();
281 log_info!(self.logger, "Cached network graph!");