Allow custom logger types.

[rapid-gossip-sync-server] / src / persistence.rs

diff --git a/src/persistence.rs b/src/persistence.rs
index 8fbf1664621cf80da643b79acbd0beb035ed9f45..22abf0216d8575692a1e3e7f9b7d4ce66844df66 100644 (file)
--- a/src/persistence.rs
+++ b/src/persistence.rs
@@ -3,20 +3,23 @@ use std::io::{BufWriter, Write};
 use std::sync::Arc;
 use std::time::{Duration, Instant};
 use lightning::routing::gossip::NetworkGraph;
+use lightning::util::logger::Logger;
 use lightning::util::ser::Writeable;
 use tokio::sync::mpsc;
 use tokio_postgres::NoTls;
 
-use crate::{config, hex_utils, TestLogger};
+use crate::config;
 use crate::types::GossipMessage;
 
-pub(crate) struct GossipPersister {
+const POSTGRES_INSERT_TIMEOUT: Duration = Duration::from_secs(15);
+
+pub(crate) struct GossipPersister<L: Logger> {
        gossip_persistence_receiver: mpsc::Receiver<GossipMessage>,
-       network_graph: Arc<NetworkGraph<TestLogger>>,
+       network_graph: Arc<NetworkGraph<Arc<L>>>,
 }
 
-impl GossipPersister {
-       pub fn new(network_graph: Arc<NetworkGraph<TestLogger>>) -> (Self, mpsc::Sender<GossipMessage>) {
+impl<L: Logger> GossipPersister<L> {
+       pub fn new(network_graph: Arc<NetworkGraph<Arc<L>>>) -> (Self, mpsc::Sender<GossipMessage>) {
                let (gossip_persistence_sender, gossip_persistence_receiver) =
                        mpsc::channel::<GossipMessage>(100);
                (GossipPersister {
@@ -109,42 +112,28 @@ impl GossipPersister {
 
                        match &gossip_message {
                                GossipMessage::ChannelAnnouncement(announcement) => {
-                                       let scid = announcement.contents.short_channel_id;
-                                       let scid_hex = hex_utils::hex_str(&scid.to_be_bytes());
-                                       // scid is 8 bytes
-                                       // block height is the first three bytes
-                                       // to obtain block height, shift scid right by 5 bytes (40 bits)
-                                       let block_height = (scid >> 5 * 8) as i32;
+                                       let scid = announcement.contents.short_channel_id as i64;
 
                                        // start with the type prefix, which is already known a priori
                                        let mut announcement_signed = Vec::new();
                                        announcement.write(&mut announcement_signed).unwrap();
 
-                                       let result = client
+                                       tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client
                                                .execute("INSERT INTO channel_announcements (\
                                                        short_channel_id, \
-                                                       block_height, \
                                                        announcement_signed \
-                                               ) VALUES ($1, $2, $3) ON CONFLICT (short_channel_id) DO NOTHING", &[
-                                                       &scid_hex,
-                                                       &block_height,
+                                               ) VALUES ($1, $2) ON CONFLICT (short_channel_id) DO NOTHING", &[
+                                                       &scid,
                                                        &announcement_signed
-                                               ]).await;
-                                       if result.is_err() {
-                                               panic!("error: {}", result.err().unwrap());
-                                       }
+                                               ])).await.unwrap().unwrap();
                                }
                                GossipMessage::ChannelUpdate(update) => {
-                                       let scid = update.contents.short_channel_id;
-                                       let scid_hex = hex_utils::hex_str(&scid.to_be_bytes());
+                                       let scid = update.contents.short_channel_id as i64;
 
                                        let timestamp = update.contents.timestamp as i64;
 
-                                       let channel_flags = update.contents.flags as i32;
-                                       let direction = channel_flags & 1;
-                                       let disable = (channel_flags & 2) > 0;
-
-                                       let composite_index = format!("{}:{}:{}", scid_hex, timestamp, direction);
+                                       let direction = (update.contents.flags & 1) == 1;
+                                       let disable = (update.contents.flags & 2) > 0;
 
                                        let cltv_expiry_delta = update.contents.cltv_expiry_delta as i32;
                                        let htlc_minimum_msat = update.contents.htlc_minimum_msat as i64;
@@ -157,9 +146,8 @@ impl GossipPersister {
                                        let mut update_signed = Vec::new();
                                        update.write(&mut update_signed).unwrap();
 
-                                       let result = client
+                                       tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client
                                                .execute("INSERT INTO channel_updates (\
-                                                       composite_index, \
                                                        short_channel_id, \
                                                        timestamp, \
                                                        channel_flags, \
@@ -171,11 +159,10 @@ impl GossipPersister {
                                                        fee_proportional_millionths, \
                                                        htlc_maximum_msat, \
                                                        blob_signed \
-                                               ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12)  ON CONFLICT (composite_index) DO NOTHING", &[
-                                                       &composite_index,
-                                                       &scid_hex,
+                                               ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)  ON CONFLICT DO NOTHING", &[
+                                                       &scid,
                                                        &timestamp,
-                                                       &channel_flags,
+                                                       &(update.contents.flags as i16),
                                                        &direction,
                                                        &disable,
                                                        &cltv_expiry_delta,
@@ -184,10 +171,7 @@ impl GossipPersister {
                                                        &fee_proportional_millionths,
                                                        &htlc_maximum_msat,
                                                        &update_signed
-                                               ]).await;
-                                       if result.is_err() {
-                                               panic!("error: {}", result.err().unwrap());
-                                       }
+                                               ])).await.unwrap().unwrap();
                                }
                        }
                }
@@ -202,7 +186,7 @@ impl GossipPersister {
                        .truncate(true)
                        .open(&cache_path)
                        .unwrap();
-               self.network_graph.remove_stale_channels();
+               self.network_graph.remove_stale_channels_and_tracking();
                let mut writer = BufWriter::new(file);
                self.network_graph.write(&mut writer).unwrap();
                writer.flush().unwrap();