Allow passing a timestamp override to snapshot calculation.

diff --git a/src/lib.rs b/src/lib.rs
index 25d13089f4ab8459dccb6099fdefda49f2e05b46..8724ef17b4974fb596022a4af2ff318452b2cb49 100644 (file)
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -14,6 +14,7 @@ use std::fs::File;
 use std::io::BufReader;
 use std::ops::Deref;
 use std::sync::Arc;
+use std::time::SystemTime;
 use bitcoin::blockdata::constants::ChainHash;
 use lightning::log_info;
 
@@ -171,7 +172,7 @@ fn serialize_empty_blob(current_timestamp: u64) -> Vec<u8> {
        blob
 }
 
-async fn serialize_delta<L: Deref + Clone>(network_graph: Arc<NetworkGraph<L>>, last_sync_timestamp: u32, logger: L) -> SerializedResponse where L::Target: Logger {
+async fn serialize_delta<L: Deref + Clone>(network_graph: Arc<NetworkGraph<L>>, last_sync_timestamp: u32, snapshot_calculation_time: Option<SystemTime>, logger: L) -> SerializedResponse where L::Target: Logger {
        let client = connect_to_db().await;
 
        network_graph.remove_stale_channels_and_tracking();
@@ -200,7 +201,7 @@ async fn serialize_delta<L: Deref + Clone>(network_graph: Arc<NetworkGraph<L>>,
        };
 
        let mut delta_set = DeltaSet::new();
-       lookup::fetch_channel_announcements(&mut delta_set, network_graph, &client, last_sync_timestamp, logger.clone()).await;
+       lookup::fetch_channel_announcements(&mut delta_set, network_graph, &client, last_sync_timestamp, snapshot_calculation_time, logger.clone()).await;
        log_info!(logger, "announcement channel count: {}", delta_set.len());
        lookup::fetch_channel_updates(&mut delta_set, &client, last_sync_timestamp, logger.clone()).await;
        log_info!(logger, "update-fetched channel count: {}", delta_set.len());

diff --git a/src/lookup.rs b/src/lookup.rs
index 5504279c63d2198b731856e96ab2f6b01233e444..76b7c2089d80c8b6a8f3484ec46078a16e2e78fb 100644 (file)
--- a/src/lookup.rs
+++ b/src/lookup.rs
@@ -76,7 +76,7 @@ impl Default for DirectedUpdateDelta {
 /// whether they had been seen before.
 /// Also include all announcements for which the first update was announced
 /// after `last_sync_timestamp`
-pub(super) async fn fetch_channel_announcements<L: Deref>(delta_set: &mut DeltaSet, network_graph: Arc<NetworkGraph<L>>, client: &Client, last_sync_timestamp: u32, logger: L) where L::Target: Logger {
+pub(super) async fn fetch_channel_announcements<L: Deref>(delta_set: &mut DeltaSet, network_graph: Arc<NetworkGraph<L>>, client: &Client, last_sync_timestamp: u32, snapshot_calculation_time: Option<SystemTime>, logger: L) where L::Target: Logger {
        log_info!(logger, "Obtaining channel ids from network graph");
        let channel_ids = {
                let read_only_graph = network_graph.read_only();
@@ -92,15 +92,27 @@ pub(super) async fn fetch_channel_announcements<L: Deref>(delta_set: &mut DeltaS
        log_info!(logger, "Last sync timestamp: {}", last_sync_timestamp);
        let last_sync_timestamp_float = last_sync_timestamp as f64;
 
-       let current_time = SystemTime::now();
+       let current_time = snapshot_calculation_time.unwrap_or(SystemTime::now());
        let current_timestamp = current_time.duration_since(UNIX_EPOCH).unwrap().as_secs();
        log_info!(logger, "Current timestamp: {}", current_timestamp);
 
-       let current_day = current_timestamp / (24 * 3600);
-       let snapshot_scope = current_timestamp.saturating_sub(last_sync_timestamp as u64);
-       log_debug!(logger, "Current day index: {}", current_day);
-       log_debug!(logger, "Snapshot scope: {}s", snapshot_scope);
-       let include_reminders = ((current_day % 5) == 0 || snapshot_scope > (40 * 3600));
+       let include_reminders = {
+               let current_hour = current_timestamp / 3600;
+               let current_day = current_timestamp / (24 * 3600);
+
+               log_debug!(logger, "Current day index: {}", current_day);
+               log_debug!(logger, "Current hour: {}", current_hour);
+
+               // anytime between 11pm and 1am
+               let is_reminder_hour = current_hour < 2 || current_hour > 22;
+               let is_reminder_day = (current_day % 5) == 0;
+
+               let snapshot_scope = current_timestamp.saturating_sub(last_sync_timestamp as u64);
+               let is_reminder_scope = snapshot_scope > (40 * 3600);
+               log_debug!(logger, "Snapshot scope: {}s", snapshot_scope);
+
+               (is_reminder_hour && is_reminder_day) || is_reminder_scope
+       };
 
        log_info!(logger, "Obtaining corresponding database entries");
        // get all the channel announcements that are currently in the network graph

diff --git a/src/snapshot.rs b/src/snapshot.rs
index 07427a6c6d94d7f69fd24d5d90bf70898191463f..ed4a0e785d505aae8b443f26044323751e91eba3 100644 (file)
--- a/src/snapshot.rs
+++ b/src/snapshot.rs
@@ -70,7 +70,8 @@ impl<L: Deref + Clone> Snapshotter<L> where L::Target: Logger {
                let relative_symlink_to_snapshot_path = "../snapshots";
 
                // 1. get the current timestamp
-               let snapshot_generation_timestamp = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs();
+               let snapshot_generation_time = SystemTime::now();
+               let snapshot_generation_timestamp = snapshot_generation_time.duration_since(UNIX_EPOCH).unwrap().as_secs();
                let reference_timestamp = Self::round_down_to_nearest_multiple(snapshot_generation_timestamp, snapshot_interval as u64);
                log_info!(self.logger, "Capturing snapshots at {} for: {}", snapshot_generation_timestamp, reference_timestamp);
 
@@ -113,7 +114,7 @@ impl<L: Deref + Clone> Snapshotter<L> where L::Target: Logger {
                        {
                                log_info!(self.logger, "Calculating {}-second snapshot", current_scope);
                                // calculate the snapshot
-                               let snapshot = super::serialize_delta(network_graph_clone, current_last_sync_timestamp.clone() as u32, self.logger.clone()).await;
+                               let snapshot = super::serialize_delta(network_graph_clone, current_last_sync_timestamp.clone() as u32, Some(snapshot_generation_time), self.logger.clone()).await;
 
                                // persist the snapshot and update the symlink
                                let snapshot_filename = format!("snapshot__calculated-at:{}__range:{}-scope__previous-sync:{}.lngossip", reference_timestamp, current_scope, current_last_sync_timestamp);

diff --git a/src/tests/mod.rs b/src/tests/mod.rs
index 33edf62d19e1965d8c41eabd7abc99c714d292a3..e668f0b473409193872a12910720be284e2abe94 100644 (file)
--- a/src/tests/mod.rs
+++ b/src/tests/mod.rs
@@ -220,7 +220,7 @@ async fn test_trivial_setup() {
                persister.persist_gossip().await;
        }
 
-       let serialization = serialize_delta(network_graph_arc.clone(), 0, logger.clone()).await;
+       let serialization = serialize_delta(network_graph_arc.clone(), 0, None, logger.clone()).await;
        logger.assert_log_contains("rapid_gossip_sync_server", "announcement channel count: 1", 1);
        clean_test_db().await;
 
@@ -305,7 +305,7 @@ async fn test_unidirectional_intermediate_update_consideration() {
        let client_graph_arc = Arc::new(client_graph);
        let rgs = RapidGossipSync::new(client_graph_arc.clone(), logger.clone());
 
-       let serialization = serialize_delta(network_graph_arc.clone(), timestamp + 1, logger.clone()).await;
+       let serialization = serialize_delta(network_graph_arc.clone(), timestamp + 1, None, logger.clone()).await;
 
        logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Fetched 1 update rows of the first update in a new direction", 1);
        logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Processed 1 reference rows", 1);
@@ -372,7 +372,7 @@ async fn test_bidirectional_intermediate_update_consideration() {
        let channel_count = network_graph_arc.read_only().channels().len();
        assert_eq!(channel_count, 1);
 
-       let serialization = serialize_delta(network_graph_arc.clone(), timestamp + 1, logger.clone()).await;
+       let serialization = serialize_delta(network_graph_arc.clone(), timestamp + 1, None, logger.clone()).await;
 
        logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Fetched 0 update rows of the first update in a new direction", 1);
        logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Processed 2 reference rows", 1);
@@ -455,7 +455,7 @@ async fn test_channel_reminders() {
        let channel_count = network_graph_arc.read_only().channels().len();
        assert_eq!(channel_count, 2);
 
-       let serialization = serialize_delta(network_graph_arc.clone(), timestamp - channel_reminder_delta + 15, logger.clone()).await;
+       let serialization = serialize_delta(network_graph_arc.clone(), timestamp - channel_reminder_delta + 15, None, logger.clone()).await;
 
        logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Fetched 0 update rows of the first update in a new direction", 1);
        logger.assert_log_contains("rapid_gossip_sync_server::lookup", "Fetched 4 update rows of the latest update in the less recently updated direction", 1);
@@ -524,7 +524,7 @@ async fn test_full_snapshot_recency() {
        let client_graph_arc = Arc::new(client_graph);
 
        { // sync after initial seed
-               let serialization = serialize_delta(network_graph_arc.clone(), 0, logger.clone()).await;
+               let serialization = serialize_delta(network_graph_arc.clone(), 0, None, logger.clone()).await;
                logger.assert_log_contains("rapid_gossip_sync_server", "announcement channel count: 1", 1);
 
                let channel_count = network_graph_arc.read_only().channels().len();
@@ -604,7 +604,7 @@ async fn test_full_snapshot_recency_with_wrong_seen_order() {
        let client_graph_arc = Arc::new(client_graph);
 
        { // sync after initial seed
-               let serialization = serialize_delta(network_graph_arc.clone(), 0, logger.clone()).await;
+               let serialization = serialize_delta(network_graph_arc.clone(), 0, None, logger.clone()).await;
                logger.assert_log_contains("rapid_gossip_sync_server", "announcement channel count: 1", 1);
 
                let channel_count = network_graph_arc.read_only().channels().len();
@@ -683,7 +683,7 @@ async fn test_full_snapshot_recency_with_wrong_propagation_order() {
        let client_graph_arc = Arc::new(client_graph);
 
        { // sync after initial seed
-               let serialization = serialize_delta(network_graph_arc.clone(), 0, logger.clone()).await;
+               let serialization = serialize_delta(network_graph_arc.clone(), 0, None, logger.clone()).await;
                logger.assert_log_contains("rapid_gossip_sync_server", "announcement channel count: 1", 1);
 
                let channel_count = network_graph_arc.read_only().channels().len();
@@ -816,7 +816,7 @@ async fn test_full_snapshot_mutiny_scenario() {
        let client_graph_arc = Arc::new(client_graph);
 
        { // sync after initial seed
-               let serialization = serialize_delta(network_graph_arc.clone(), 0, logger.clone()).await;
+               let serialization = serialize_delta(network_graph_arc.clone(), 0, None, logger.clone()).await;
                logger.assert_log_contains("rapid_gossip_sync_server", "announcement channel count: 1", 1);
 
                let channel_count = network_graph_arc.read_only().channels().len();
@@ -929,7 +929,7 @@ async fn test_full_snapshot_interlaced_channel_timestamps() {
        let client_graph_arc = Arc::new(client_graph);
 
        { // sync after initial seed
-               let serialization = serialize_delta(network_graph_arc.clone(), 0, logger.clone()).await;
+               let serialization = serialize_delta(network_graph_arc.clone(), 0, None, logger.clone()).await;
                logger.assert_log_contains("rapid_gossip_sync_server", "announcement channel count: 2", 1);
 
                let channel_count = network_graph_arc.read_only().channels().len();