]> git.bitcoin.ninja Git - rapid-gossip-sync-server/blob - src/persistence.rs
Merge pull request #68 from johncantrell97/118
[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::sync::mpsc;
11
12 use crate::config;
13 use crate::types::GossipMessage;
14
15 const POSTGRES_INSERT_TIMEOUT: Duration = Duration::from_secs(15);
16
17 pub(crate) struct GossipPersister<L: Deref> where L::Target: Logger {
18         gossip_persistence_receiver: mpsc::Receiver<GossipMessage>,
19         network_graph: Arc<NetworkGraph<L>>,
20         logger: L
21 }
22
23 impl<L: Deref> GossipPersister<L> where L::Target: Logger {
24         pub fn new(network_graph: Arc<NetworkGraph<L>>, logger: L) -> (Self, mpsc::Sender<GossipMessage>) {
25                 let (gossip_persistence_sender, gossip_persistence_receiver) =
26                         mpsc::channel::<GossipMessage>(100);
27                 (GossipPersister {
28                         gossip_persistence_receiver,
29                         network_graph,
30                         logger
31                 }, gossip_persistence_sender)
32         }
33
34         pub(crate) async fn persist_gossip(&mut self) {
35                 let mut client = crate::connect_to_db().await;
36
37                 {
38                         // initialize the database
39                         let initialization = client
40                                 .execute(config::db_config_table_creation_query(), &[])
41                                 .await;
42                         if let Err(initialization_error) = initialization {
43                                 panic!("db init error: {}", initialization_error);
44                         }
45
46                         let cur_schema = client.query("SELECT db_schema FROM config WHERE id = $1", &[&1]).await.unwrap();
47                         if !cur_schema.is_empty() {
48                                 config::upgrade_db(cur_schema[0].get(0), &mut client).await;
49                         }
50
51                         let preparation = client.execute("set time zone UTC", &[]).await;
52                         if let Err(preparation_error) = preparation {
53                                 panic!("db preparation error: {}", preparation_error);
54                         }
55
56                         let initialization = client
57                                 .execute(
58                                         // TODO: figure out a way to fix the id value without Postgres complaining about
59                                         // its value not being default
60                                         "INSERT INTO config (id, db_schema) VALUES ($1, $2) ON CONFLICT (id) DO NOTHING",
61                                         &[&1, &config::SCHEMA_VERSION]
62                                 ).await;
63                         if let Err(initialization_error) = initialization {
64                                 panic!("db init error: {}", initialization_error);
65                         }
66
67                         let initialization = client
68                                 .execute(config::db_announcement_table_creation_query(), &[])
69                                 .await;
70                         if let Err(initialization_error) = initialization {
71                                 panic!("db init error: {}", initialization_error);
72                         }
73
74                         let initialization = client
75                                 .execute(
76                                         config::db_channel_update_table_creation_query(),
77                                         &[],
78                                 )
79                                 .await;
80                         if let Err(initialization_error) = initialization {
81                                 panic!("db init error: {}", initialization_error);
82                         }
83
84                         let initialization = client
85                                 .batch_execute(config::db_index_creation_query())
86                                 .await;
87                         if let Err(initialization_error) = initialization {
88                                 panic!("db init error: {}", initialization_error);
89                         }
90                 }
91
92                 // print log statement every minute
93                 let mut latest_persistence_log = Instant::now() - Duration::from_secs(60);
94                 let mut i = 0u32;
95                 let mut latest_graph_cache_time = Instant::now();
96                 // TODO: it would be nice to have some sort of timeout here so after 10 seconds of
97                 // inactivity, some sort of message could be broadcast signaling the activation of request
98                 // processing
99                 while let Some(gossip_message) = &self.gossip_persistence_receiver.recv().await {
100                         i += 1; // count the persisted gossip messages
101
102                         if latest_persistence_log.elapsed().as_secs() >= 60 {
103                                 log_info!(self.logger, "Persisting gossip message #{}", i);
104                                 latest_persistence_log = Instant::now();
105                         }
106
107                         // has it been ten minutes? Just cache it
108                         if latest_graph_cache_time.elapsed().as_secs() >= 600 {
109                                 self.persist_network_graph();
110                                 latest_graph_cache_time = Instant::now();
111                         }
112
113                         match &gossip_message {
114                                 GossipMessage::ChannelAnnouncement(announcement, seen_override) => {
115                                         let scid = announcement.contents.short_channel_id as i64;
116
117                                         // start with the type prefix, which is already known a priori
118                                         let mut announcement_signed = Vec::new();
119                                         announcement.write(&mut announcement_signed).unwrap();
120
121                                         if cfg!(test) && seen_override.is_some() {
122                                                 tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client
123                                                         .execute("INSERT INTO channel_announcements (\
124                                                         short_channel_id, \
125                                                         announcement_signed, \
126                                                         seen \
127                                                 ) VALUES ($1, $2, TO_TIMESTAMP($3)) ON CONFLICT (short_channel_id) DO NOTHING", &[
128                                                                 &scid,
129                                                                 &announcement_signed,
130                                                                 &(seen_override.unwrap() as f64)
131                                                         ])).await.unwrap().unwrap();
132                                         } else {
133                                                 tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client
134                                                         .execute("INSERT INTO channel_announcements (\
135                                                         short_channel_id, \
136                                                         announcement_signed \
137                                                 ) VALUES ($1, $2) ON CONFLICT (short_channel_id) DO NOTHING", &[
138                                                                 &scid,
139                                                                 &announcement_signed
140                                                         ])).await.unwrap().unwrap();
141                                         }
142                                 }
143                                 GossipMessage::ChannelUpdate(update, seen_override) => {
144                                         let scid = update.contents.short_channel_id as i64;
145
146                                         let timestamp = update.contents.timestamp as i64;
147
148                                         let direction = (update.contents.flags & 1) == 1;
149                                         let disable = (update.contents.flags & 2) > 0;
150
151                                         let cltv_expiry_delta = update.contents.cltv_expiry_delta as i32;
152                                         let htlc_minimum_msat = update.contents.htlc_minimum_msat as i64;
153                                         let fee_base_msat = update.contents.fee_base_msat as i32;
154                                         let fee_proportional_millionths =
155                                                 update.contents.fee_proportional_millionths as i32;
156                                         let htlc_maximum_msat = update.contents.htlc_maximum_msat as i64;
157
158                                         // start with the type prefix, which is already known a priori
159                                         let mut update_signed = Vec::new();
160                                         update.write(&mut update_signed).unwrap();
161
162                                         let insertion_statement = if cfg!(test) {
163                                                 "INSERT INTO channel_updates (\
164                                                         short_channel_id, \
165                                                         timestamp, \
166                                                         seen, \
167                                                         channel_flags, \
168                                                         direction, \
169                                                         disable, \
170                                                         cltv_expiry_delta, \
171                                                         htlc_minimum_msat, \
172                                                         fee_base_msat, \
173                                                         fee_proportional_millionths, \
174                                                         htlc_maximum_msat, \
175                                                         blob_signed \
176                                                 ) VALUES ($1, $2, TO_TIMESTAMP($3), $4, $5, $6, $7, $8, $9, $10, $11, $12)  ON CONFLICT DO NOTHING"
177                                         } else {
178                                                 "INSERT INTO channel_updates (\
179                                                         short_channel_id, \
180                                                         timestamp, \
181                                                         channel_flags, \
182                                                         direction, \
183                                                         disable, \
184                                                         cltv_expiry_delta, \
185                                                         htlc_minimum_msat, \
186                                                         fee_base_msat, \
187                                                         fee_proportional_millionths, \
188                                                         htlc_maximum_msat, \
189                                                         blob_signed \
190                                                 ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)  ON CONFLICT DO NOTHING"
191                                         };
192
193                                         // this may not be used outside test cfg
194                                         let _seen_timestamp = seen_override.unwrap_or(timestamp as u32) as f64;
195
196                                         tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client
197                                                 .execute(insertion_statement, &[
198                                                         &scid,
199                                                         &timestamp,
200                                                         #[cfg(test)]
201                                                                 &_seen_timestamp,
202                                                         &(update.contents.flags as i16),
203                                                         &direction,
204                                                         &disable,
205                                                         &cltv_expiry_delta,
206                                                         &htlc_minimum_msat,
207                                                         &fee_base_msat,
208                                                         &fee_proportional_millionths,
209                                                         &htlc_maximum_msat,
210                                                         &update_signed
211                                                 ])).await.unwrap().unwrap();
212                                 }
213                         }
214                 }
215         }
216
217         fn persist_network_graph(&self) {
218                 log_info!(self.logger, "Caching network graph…");
219                 let cache_path = config::network_graph_cache_path();
220                 let file = OpenOptions::new()
221                         .create(true)
222                         .write(true)
223                         .truncate(true)
224                         .open(&cache_path)
225                         .unwrap();
226                 self.network_graph.remove_stale_channels_and_tracking();
227                 let mut writer = BufWriter::new(file);
228                 self.network_graph.write(&mut writer).unwrap();
229                 writer.flush().unwrap();
230                 log_info!(self.logger, "Cached network graph!");
231         }
232 }