use lightning::util::logger::Logger;
use lightning::util::ser::{ReadableArgs, Writeable};
use tokio::sync::mpsc;
+use tokio_postgres::{Client, NoTls};
+use crate::config::SYMLINK_GRANULARITY_INTERVAL;
use crate::lookup::DeltaSet;
use crate::persistence::GossipPersister;
pub mod types;
+#[cfg(test)]
+mod tests;
+
/// The purpose of this prefix is to identify the serialization format, should other rapid gossip
/// sync formats arise in the future.
///
}
pub async fn start_sync(&self) {
+ log_info!(self.logger, "Starting Rapid Gossip Sync Server");
+ log_info!(self.logger, "Snapshot interval: {} seconds", config::snapshot_generation_interval());
+
// means to indicate sync completion status within this module
let (sync_completion_sender, mut sync_completion_receiver) = mpsc::channel::<()>(1);
}
}
+pub(crate) async fn connect_to_db() -> Client {
+ let connection_config = config::db_connection_config();
+ let (client, connection) = connection_config.connect(NoTls).await.unwrap();
+
+ tokio::spawn(async move {
+ if let Err(e) = connection.await {
+ panic!("connection error: {}", e);
+ }
+ });
+
+ #[cfg(test)]
+ {
+ let schema_name = tests::db_test_schema();
+ let schema_creation_command = format!("CREATE SCHEMA IF NOT EXISTS {}", schema_name);
+ client.execute(&schema_creation_command, &[]).await.unwrap();
+ client.execute(&format!("SET search_path TO {}", schema_name), &[]).await.unwrap();
+ }
+
+ client.execute("set time zone UTC", &[]).await.unwrap();
+ client
+}
+
/// This method generates a no-op blob that can be used as a delta where none exists.
///
/// The primary purpose of this method is the scenario of a client retrieving and processing a
let chain_hash = genesis_block.block_hash();
chain_hash.write(&mut blob).unwrap();
- let blob_timestamp = Snapshotter::<Arc<RGSSLogger>>::round_down_to_nearest_multiple(current_timestamp, config::SNAPSHOT_CALCULATION_INTERVAL as u64) as u32;
+ let blob_timestamp = Snapshotter::<Arc<RGSSLogger>>::round_down_to_nearest_multiple(current_timestamp, SYMLINK_GRANULARITY_INTERVAL as u64) as u32;
blob_timestamp.write(&mut blob).unwrap();
0u32.write(&mut blob).unwrap(); // node count
}
async fn serialize_delta<L: Deref + Clone>(network_graph: Arc<NetworkGraph<L>>, last_sync_timestamp: u32, logger: L) -> SerializedResponse where L::Target: Logger {
- let (client, connection) = lookup::connect_to_db().await;
+ let client = connect_to_db().await;
network_graph.remove_stale_channels_and_tracking();
- tokio::spawn(async move {
- if let Err(e) = connection.await {
- panic!("connection error: {}", e);
- }
- });
-
let mut output: Vec<u8> = vec![];
+ let snapshot_interval = config::snapshot_generation_interval();
// set a flag if the chain hash is prepended
// chain hash only necessary if either channel announcements or non-incremental updates are present
serialization_details.chain_hash.write(&mut prefixed_output).unwrap();
// always write the latest seen timestamp
let latest_seen_timestamp = serialization_details.latest_seen;
- let overflow_seconds = latest_seen_timestamp % config::SNAPSHOT_CALCULATION_INTERVAL;
+ let overflow_seconds = latest_seen_timestamp % snapshot_interval;
let serialized_seen_timestamp = latest_seen_timestamp.saturating_sub(overflow_seconds);
serialized_seen_timestamp.write(&mut prefixed_output).unwrap();