use tokio_postgres::Client;
use futures::StreamExt;
-use lightning::{log_gossip, log_info};
+use lightning::{log_debug, log_gossip, log_info};
use lightning::util::logger::Logger;
use crate::config;
/// 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_reference_timestamp: Option<u64>, 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();
log_info!(logger, "Last sync timestamp: {}", last_sync_timestamp);
let last_sync_timestamp_float = last_sync_timestamp as f64;
+ let current_timestamp = snapshot_reference_timestamp.unwrap_or(SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs());
+ log_info!(logger, "Current timestamp: {}", current_timestamp);
+
+ 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);
+
+ // every 5th day at midnight
+ let is_reminder_hour = (current_hour % 24) == 0;
+ 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
let announcement_rows = client.query_raw("SELECT announcement_signed, CAST(EXTRACT('epoch' from seen) AS BIGINT) AS seen FROM channel_announcements WHERE short_channel_id = any($1) ORDER BY short_channel_id ASC", [&channel_ids]).await.unwrap();
log_info!(logger, "Fetched {} update rows of the first update in a new direction", newer_oldest_directional_update_count);
}
- {
+ if include_reminders {
// THIS STEP IS USED TO DETERMINE IF A REMINDER UPDATE SHOULD BE SENT
log_info!(logger, "Annotating channel announcements whose latest channel update in a given direction occurred more than six days ago");
// Steps:
// — Obtain all updates, distinct by (scid, direction), ordered by seen DESC
// — From those updates, select distinct by (scid), ordered by seen ASC (to obtain the older one per direction)
- let reminder_threshold_timestamp = SystemTime::now().checked_sub(config::CHANNEL_REMINDER_AGE).unwrap().duration_since(UNIX_EPOCH).unwrap().as_secs() as f64;
+ let reminder_threshold_timestamp = current_timestamp.checked_sub(config::CHANNEL_REMINDER_AGE.as_secs()).unwrap() as f64;
log_info!(logger, "Fetch first time we saw the current value combination for each direction (prior mutations excepted)");
- let reminder_lookup_threshold_timestamp = SystemTime::now().checked_sub(config::CHANNEL_REMINDER_AGE * 3).unwrap().duration_since(UNIX_EPOCH).unwrap().as_secs() as f64;
+ let reminder_lookup_threshold_timestamp = current_timestamp.checked_sub(config::CHANNEL_REMINDER_AGE.as_secs() * 3).unwrap() as f64;
let params: [&(dyn tokio_postgres::types::ToSql + Sync); 2] = [&channel_ids, &reminder_lookup_threshold_timestamp];
/*