]> git.bitcoin.ninja Git - rapid-gossip-sync-server/blob - src/persistence.rs
Fix unused `Result` in tests
[rapid-gossip-sync-server] / src / persistence.rs
1 use std::fs::OpenOptions;
2 use std::io::{BufWriter, Write};
3 use std::ops::Deref;
4 use std::sync::Arc;
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};
12
13 use crate::config;
14 use crate::types::GossipMessage;
15
16 const POSTGRES_INSERT_TIMEOUT: Duration = Duration::from_secs(15);
17 const INSERT_PARALELLISM: usize = 16;
18
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,
23         logger: L
24 }
25
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();
31                 (GossipPersister {
32                         gossip_persistence_receiver,
33                         network_graph,
34                         tokio_runtime: runtime,
35                         logger
36                 }, gossip_persistence_sender)
37         }
38
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;
43
44                         let initialization = client
45                                 .execute(config::db_config_table_creation_query(), &[])
46                                 .await;
47                         if let Err(initialization_error) = initialization {
48                                 panic!("db init error: {}", initialization_error);
49                         }
50
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;
54                         }
55
56                         let preparation = client.execute("set time zone UTC", &[]).await;
57                         if let Err(preparation_error) = preparation {
58                                 panic!("db preparation error: {}", preparation_error);
59                         }
60
61                         let initialization = client
62                                 .execute(
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]
67                                 ).await;
68                         if let Err(initialization_error) = initialization {
69                                 panic!("db init error: {}", initialization_error);
70                         }
71
72                         let initialization = client
73                                 .execute(config::db_announcement_table_creation_query(), &[])
74                                 .await;
75                         if let Err(initialization_error) = initialization {
76                                 panic!("db init error: {}", initialization_error);
77                         }
78
79                         let initialization = client
80                                 .execute(
81                                         config::db_channel_update_table_creation_query(),
82                                         &[],
83                                 )
84                                 .await;
85                         if let Err(initialization_error) = initialization {
86                                 panic!("db init error: {}", initialization_error);
87                         }
88
89                         let initialization = client
90                                 .batch_execute(config::db_index_creation_query())
91                                 .await;
92                         if let Err(initialization_error) = initialization {
93                                 panic!("db init error: {}", initialization_error);
94                         }
95                 }
96
97                 // print log statement every minute
98                 let mut latest_persistence_log = Instant::now() - Duration::from_secs(60);
99                 let mut i = 0u32;
100                 let mut latest_graph_cache_time = Instant::now();
101                 let insert_limiter = Arc::new(Semaphore::new(INSERT_PARALELLISM));
102                 let connections_cache = Arc::new(Mutex::new(Vec::with_capacity(INSERT_PARALELLISM)));
103                 #[cfg(test)]
104                 let mut tasks_spawned = Vec::new();
105                 // TODO: it would be nice to have some sort of timeout here so after 10 seconds of
106                 // inactivity, some sort of message could be broadcast signaling the activation of request
107                 // processing
108                 while let Some(gossip_message) = self.gossip_persistence_receiver.recv().await {
109                         i += 1; // count the persisted gossip messages
110
111                         if latest_persistence_log.elapsed().as_secs() >= 60 {
112                                 log_info!(self.logger, "Persisting gossip message #{}", i);
113                                 latest_persistence_log = Instant::now();
114                         }
115
116                         // has it been ten minutes? Just cache it
117                         if latest_graph_cache_time.elapsed().as_secs() >= 600 {
118                                 self.persist_network_graph();
119                                 latest_graph_cache_time = Instant::now();
120                         }
121                         insert_limiter.acquire().await.unwrap().forget();
122
123                         let limiter_ref = Arc::clone(&insert_limiter);
124                         let client = {
125                                 let mut connections_set = connections_cache.lock().await;
126                                 let client = if connections_set.is_empty() {
127                                         crate::connect_to_db().await
128                                 } else {
129                                         connections_set.pop().unwrap()
130                                 };
131                                 client
132                         };
133
134                         let connections_cache_ref = Arc::clone(&connections_cache);
135                         match gossip_message {
136                                 GossipMessage::ChannelAnnouncement(announcement, seen_override) => {
137                                         let scid = announcement.contents.short_channel_id as i64;
138
139                                         // start with the type prefix, which is already known a priori
140                                         let mut announcement_signed = Vec::new();
141                                         announcement.write(&mut announcement_signed).unwrap();
142
143                                         let _task = self.tokio_runtime.spawn(async move {
144                                                 if cfg!(test) && seen_override.is_some() {
145                                                         tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client
146                                                                 .execute("INSERT INTO channel_announcements (\
147                                                                 short_channel_id, \
148                                                                 announcement_signed, \
149                                                                 seen \
150                                                         ) VALUES ($1, $2, TO_TIMESTAMP($3)) ON CONFLICT (short_channel_id) DO NOTHING", &[
151                                                                         &scid,
152                                                                         &announcement_signed,
153                                                                         &(seen_override.unwrap() as f64)
154                                                                 ])).await.unwrap().unwrap();
155                                                 } else {
156                                                         tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client
157                                                                 .execute("INSERT INTO channel_announcements (\
158                                                                 short_channel_id, \
159                                                                 announcement_signed \
160                                                         ) VALUES ($1, $2) ON CONFLICT (short_channel_id) DO NOTHING", &[
161                                                                         &scid,
162                                                                         &announcement_signed
163                                                                 ])).await.unwrap().unwrap();
164                                                 }
165                                                 let mut connections_set = connections_cache_ref.lock().await;
166                                                 connections_set.push(client);
167                                                 limiter_ref.add_permits(1);
168                                         });
169                                         #[cfg(test)]
170                                         tasks_spawned.push(_task);
171                                 }
172                                 GossipMessage::ChannelUpdate(update, seen_override) => {
173                                         let scid = update.contents.short_channel_id as i64;
174
175                                         let timestamp = update.contents.timestamp as i64;
176
177                                         let direction = (update.contents.flags & 1) == 1;
178                                         let disable = (update.contents.flags & 2) > 0;
179
180                                         let cltv_expiry_delta = update.contents.cltv_expiry_delta as i32;
181                                         let htlc_minimum_msat = update.contents.htlc_minimum_msat as i64;
182                                         let fee_base_msat = update.contents.fee_base_msat as i32;
183                                         let fee_proportional_millionths =
184                                                 update.contents.fee_proportional_millionths as i32;
185                                         let htlc_maximum_msat = update.contents.htlc_maximum_msat as i64;
186
187                                         // start with the type prefix, which is already known a priori
188                                         let mut update_signed = Vec::new();
189                                         update.write(&mut update_signed).unwrap();
190
191                                         let insertion_statement = if cfg!(test) {
192                                                 "INSERT INTO channel_updates (\
193                                                         short_channel_id, \
194                                                         timestamp, \
195                                                         seen, \
196                                                         channel_flags, \
197                                                         direction, \
198                                                         disable, \
199                                                         cltv_expiry_delta, \
200                                                         htlc_minimum_msat, \
201                                                         fee_base_msat, \
202                                                         fee_proportional_millionths, \
203                                                         htlc_maximum_msat, \
204                                                         blob_signed \
205                                                 ) VALUES ($1, $2, TO_TIMESTAMP($3), $4, $5, $6, $7, $8, $9, $10, $11, $12)  ON CONFLICT DO NOTHING"
206                                         } else {
207                                                 "INSERT INTO channel_updates (\
208                                                         short_channel_id, \
209                                                         timestamp, \
210                                                         channel_flags, \
211                                                         direction, \
212                                                         disable, \
213                                                         cltv_expiry_delta, \
214                                                         htlc_minimum_msat, \
215                                                         fee_base_msat, \
216                                                         fee_proportional_millionths, \
217                                                         htlc_maximum_msat, \
218                                                         blob_signed \
219                                                 ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)  ON CONFLICT DO NOTHING"
220                                         };
221
222                                         // this may not be used outside test cfg
223                                         let _seen_timestamp = seen_override.unwrap_or(timestamp as u32) as f64;
224
225                                         let _task = self.tokio_runtime.spawn(async move {
226                                                 tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client
227                                                         .execute(insertion_statement, &[
228                                                                 &scid,
229                                                                 &timestamp,
230                                                                 #[cfg(test)]
231                                                                         &_seen_timestamp,
232                                                                 &(update.contents.flags as i16),
233                                                                 &direction,
234                                                                 &disable,
235                                                                 &cltv_expiry_delta,
236                                                                 &htlc_minimum_msat,
237                                                                 &fee_base_msat,
238                                                                 &fee_proportional_millionths,
239                                                                 &htlc_maximum_msat,
240                                                                 &update_signed
241                                                         ])).await.unwrap().unwrap();
242                                                 let mut connections_set = connections_cache_ref.lock().await;
243                                                 connections_set.push(client);
244                                                 limiter_ref.add_permits(1);
245                                         });
246                                         #[cfg(test)]
247                                         tasks_spawned.push(_task);
248                                 }
249                         }
250                 }
251                 #[cfg(test)]
252                 for task in tasks_spawned {
253                         task.await.unwrap();
254                 }
255         }
256
257         fn persist_network_graph(&self) {
258                 log_info!(self.logger, "Caching network graph…");
259                 let cache_path = config::network_graph_cache_path();
260                 let file = OpenOptions::new()
261                         .create(true)
262                         .write(true)
263                         .truncate(true)
264                         .open(&cache_path)
265                         .unwrap();
266                 self.network_graph.remove_stale_channels_and_tracking();
267                 let mut writer = BufWriter::new(file);
268                 self.network_graph.write(&mut writer).unwrap();
269                 writer.flush().unwrap();
270                 log_info!(self.logger, "Cached network graph!");
271         }
272 }