Use the generation time, not last seen time, for update timestamps

When we write an update, we have to include a timestamp that
represents both the the time that the client will use to fetch the
next diff and the timestamp used as a reference point when adding
updates to the network graph (the client actually uses the included
timestamp minus two weeks).

Because of the second use, updates were being generated with the
highest timestmap of all included updates, rounded down to the
nearest update-interval multiple (because of the first use).

In practice, because we expect to see many updates in any hour
window, this meant we were really calculating the generation
reference timestamp in a very indirect way. Here we simply change
to using the reference timestamp directly.

This updates regtest results when generating an initial sync
against an empty graph to include a real timestamp, rather than 0,
though this shouldn't impact clients' correctness.

Author: TheBlueMatt

src/lib.rs

@@ -203,7 +203,9 @@ async fn calculate_delta<L: Deref + Clone>(network_graph: Arc<NetworkGraph<L>>,
 	serialization::serialize_delta_set(delta_set, node_delta_set, last_sync_timestamp)
 }
 
-fn serialize_delta<L: Deref + Clone>(serialization_details: &SerializationSet, serialization_version: u8, logger: L) -> SerializedResponse where L::Target: Logger {
+fn serialize_delta<L: Deref + Clone>(
+	serialization_details: &SerializationSet, serialization_time: u32, serialization_version: u8, logger: L,
+) -> SerializedResponse where L::Target: Logger {
 	let mut output: Vec<u8> = vec![];
 	let snapshot_interval = config::snapshot_generation_interval();
 
@@ -277,11 +279,9 @@ fn serialize_delta<L: Deref + Clone>(serialization_details: &SerializationSet, s
 
 	// always write the chain hash
 	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 % snapshot_interval;
-	let serialized_seen_timestamp = latest_seen_timestamp.saturating_sub(overflow_seconds);
-	serialized_seen_timestamp.write(&mut prefixed_output).unwrap();
+	// always write the time we're generating this object
+	assert_eq!(serialization_time % snapshot_interval, 0);
+	serialization_time.write(&mut prefixed_output).unwrap();
 
 	if serialization_version >= 2 { // serialize the most common node features
 		for mutated_node_id in serialization_details.node_mutations.keys() {
@@ -389,7 +389,7 @@ fn serialize_delta<L: Deref + Clone>(serialization_details: &SerializationSet, s
 	prefixed_output.append(&mut output);
 
 	log_info!(logger, "duplicated node ids: {}", duplicate_node_ids);
-	log_info!(logger, "latest seen timestamp: {:?}", serialization_details.latest_seen);
+	log_info!(logger, "generated at: {}", serialization_time);
 
 	SerializedResponse {
 		data: prefixed_output,

src/serialization.rs

@@ -1,4 +1,3 @@
-use std::cmp::max;
 use std::collections::HashMap;
 use std::time::{SystemTime, UNIX_EPOCH};
 
@@ -16,7 +15,6 @@ pub(super) struct SerializationSet {
 	pub(super) full_update_defaults: DefaultUpdateValues,
 	pub(super) node_announcement_feature_defaults: Vec<NodeFeatures>,
 	pub(super) node_mutations: NodeDeltaSet,
-	pub(super) latest_seen: u32,
 	pub(super) chain_hash: ChainHash,
 }
 
@@ -130,7 +128,6 @@ pub(super) fn serialize_delta_set(channel_delta_set: DeltaSet, node_delta_set: N
 		node_announcement_feature_defaults: vec![],
 		node_mutations: Default::default(),
 		chain_hash: ChainHash::using_genesis_block(config::network()),
-		latest_seen: 0,
 	};
 
 	let mut full_update_histograms = FullUpdateValueHistograms {
@@ -165,7 +162,6 @@ pub(super) fn serialize_delta_set(channel_delta_set: DeltaSet, node_delta_set: N
 		};
 		let send_announcement = is_new_announcement || is_newly_included_announcement;
 		if send_announcement {
-			serialization_set.latest_seen = max(serialization_set.latest_seen, current_announcement_seen);
 			serialization_set.announcements.push(channel_delta.announcement.unwrap().announcement);
 		}
 
@@ -178,10 +174,6 @@ pub(super) fn serialize_delta_set(channel_delta_set: DeltaSet, node_delta_set: N
 					let latest_update = latest_update_delta.update;
 					assert_eq!(latest_update.short_channel_id, scid, "Update in DB had wrong SCID column");
 
-					// the returned seen timestamp should be the latest of all the returned
-					// announcements and latest updates
-					serialization_set.latest_seen = max(serialization_set.latest_seen, latest_update_delta.seen);
-
 					if let Some(update_delta) = updates.last_update_before_seen {
 						let mutated_properties = updates.mutated_properties;
 						if send_announcement || mutated_properties.len() == 5 || update_delta.seen <= non_incremental_previous_update_threshold_timestamp {

src/snapshot.rs

@@ -73,6 +73,7 @@ impl<L: Deref + Clone> Snapshotter<L> where L::Target: Logger {
 		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);
+		let serialized_timestamp: u32 = reference_timestamp.try_into().unwrap();
 		log_info!(self.logger, "Capturing snapshots at {} for: {}", snapshot_generation_timestamp, reference_timestamp);
 
 		// 2. sleep until the next round interval
@@ -121,8 +122,8 @@ impl<L: Deref + Clone> Snapshotter<L> where L::Target: Logger {
 				log_info!(self.logger, "Calculating {}-second snapshot", current_scope);
 				// calculate the snapshot
 				let delta = super::calculate_delta(network_graph_clone.clone(), current_last_sync_timestamp.clone() as u32, Some(reference_timestamp), self.logger.clone()).await;
-				let snapshot_v1 = super::serialize_delta(&delta, 1, self.logger.clone());
-				let snapshot_v2 = super::serialize_delta(&delta, 2, self.logger.clone());
+				let snapshot_v1 = super::serialize_delta(&delta, serialized_timestamp, 1, self.logger.clone());
+				let snapshot_v2 = super::serialize_delta(&delta, serialized_timestamp, 2, self.logger.clone());
 
 				// 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);

src/tests/mod.rs

@@ -263,7 +263,8 @@ async fn test_trivial_setup() {
 	}
 
 	let delta = calculate_delta(network_graph_arc.clone(), 0, None, logger.clone()).await;
-	let serialization = serialize_delta(&delta, 1, logger.clone());
+	let delta_time = timestamp % config::snapshot_generation_interval();
+	let serialization = serialize_delta(&delta, delta_time, 1, logger.clone());
 	logger.assert_log_contains("rapid_gossip_sync_server", "announcement channel count: 1", 1);
 	clean_test_db().await;
 
@@ -280,8 +281,7 @@ async fn test_trivial_setup() {
 	let update_result = rgs.update_network_graph(&serialization.data).unwrap();
 	println!("update result: {}", update_result);
 	// the update result must be a multiple of our snapshot granularity
-	assert_eq!(update_result % config::snapshot_generation_interval(), 0);
-	assert!(update_result < timestamp);
+	assert_eq!(update_result, delta_time);
 
 	let timestamp_delta = timestamp - update_result;
 	println!("timestamp delta: {}", timestamp_delta);
@@ -428,7 +428,7 @@ async fn test_node_announcement_delta_detection() {
 	}
 
 	let delta = calculate_delta(network_graph_arc.clone(), timestamp - 5, None, logger.clone()).await;
-	let serialization = serialize_delta(&delta, 2, logger.clone());
+	let serialization = serialize_delta(&delta, timestamp, 2, logger.clone());
 	clean_test_db().await;
 
 	assert_eq!(serialization.message_count, 3);
@@ -479,7 +479,7 @@ async fn test_unidirectional_intermediate_update_consideration() {
 	let rgs = RapidGossipSync::new(client_graph_arc.clone(), logger.clone());
 
 	let delta = calculate_delta(network_graph_arc.clone(), timestamp + 1, None, logger.clone()).await;
-	let serialization = serialize_delta(&delta, 1, logger.clone());
+	let serialization = serialize_delta(&delta, current_time(), 1, logger.clone());
 
 	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);
@@ -547,7 +547,7 @@ async fn test_bidirectional_intermediate_update_consideration() {
 	assert_eq!(channel_count, 1);
 
 	let delta = calculate_delta(network_graph_arc.clone(), timestamp + 1, None, logger.clone()).await;
-	let serialization = serialize_delta(&delta, 1, logger.clone());
+	let serialization = serialize_delta(&delta, current_time(), 1, logger.clone());
 
 	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);
@@ -631,7 +631,7 @@ async fn test_channel_reminders() {
 	assert_eq!(channel_count, 2);
 
 	let delta = calculate_delta(network_graph_arc.clone(), timestamp - channel_reminder_delta + 15, None, logger.clone()).await;
-	let serialization = serialize_delta(&delta, 1, logger.clone());
+	let serialization = serialize_delta(&delta, current_time(), 1, logger.clone());
 
 	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);
@@ -701,7 +701,8 @@ async fn test_full_snapshot_recency() {
 
 	{ // sync after initial seed
 		let delta = calculate_delta(network_graph_arc.clone(), 0, None, logger.clone()).await;
-		let serialization = serialize_delta(&delta, 1, logger.clone());
+		let delta_time = current_time() % config::snapshot_generation_interval();
+		let serialization = serialize_delta(&delta, current_time(), 1, logger.clone());
 		logger.assert_log_contains("rapid_gossip_sync_server", "announcement channel count: 1", 1);
 
 		let channel_count = network_graph_arc.read_only().channels().len();
@@ -714,8 +715,7 @@ async fn test_full_snapshot_recency() {
 		let rgs = RapidGossipSync::new(client_graph_arc.clone(), logger.clone());
 		let update_result = rgs.update_network_graph(&serialization.data).unwrap();
 		// the update result must be a multiple of our snapshot granularity
-		assert_eq!(update_result % config::snapshot_generation_interval(), 0);
-		assert!(update_result < timestamp);
+		assert_eq!(update_result, delta_time);
 
 		let readonly_graph = client_graph_arc.read_only();
 		let channels = readonly_graph.channels();
@@ -782,7 +782,8 @@ async fn test_full_snapshot_recency_with_wrong_seen_order() {
 
 	{ // sync after initial seed
 		let delta = calculate_delta(network_graph_arc.clone(), 0, None, logger.clone()).await;
-		let serialization = serialize_delta(&delta, 1, logger.clone());
+		let delta_time = current_time() % config::snapshot_generation_interval();
+		let serialization = serialize_delta(&delta, delta_time, 1, logger.clone());
 		logger.assert_log_contains("rapid_gossip_sync_server", "announcement channel count: 1", 1);
 
 		let channel_count = network_graph_arc.read_only().channels().len();
@@ -795,8 +796,7 @@ async fn test_full_snapshot_recency_with_wrong_seen_order() {
 		let rgs = RapidGossipSync::new(client_graph_arc.clone(), logger.clone());
 		let update_result = rgs.update_network_graph(&serialization.data).unwrap();
 		// the update result must be a multiple of our snapshot granularity
-		assert_eq!(update_result % config::snapshot_generation_interval(), 0);
-		assert!(update_result < timestamp);
+		assert_eq!(update_result, delta_time);
 
 		let readonly_graph = client_graph_arc.read_only();
 		let channels = readonly_graph.channels();
@@ -862,7 +862,8 @@ async fn test_full_snapshot_recency_with_wrong_propagation_order() {
 
 	{ // sync after initial seed
 		let delta = calculate_delta(network_graph_arc.clone(), 0, None, logger.clone()).await;
-		let serialization = serialize_delta(&delta, 1, logger.clone());
+		let delta_time = current_time() % config::snapshot_generation_interval();
+		let serialization = serialize_delta(&delta, delta_time, 1, logger.clone());
 		logger.assert_log_contains("rapid_gossip_sync_server", "announcement channel count: 1", 1);
 
 		let channel_count = network_graph_arc.read_only().channels().len();
@@ -875,8 +876,7 @@ async fn test_full_snapshot_recency_with_wrong_propagation_order() {
 		let rgs = RapidGossipSync::new(client_graph_arc.clone(), logger.clone());
 		let update_result = rgs.update_network_graph(&serialization.data).unwrap();
 		// the update result must be a multiple of our snapshot granularity
-		assert_eq!(update_result % config::snapshot_generation_interval(), 0);
-		assert!(update_result < timestamp);
+		assert_eq!(update_result, delta_time);
 
 		let readonly_graph = client_graph_arc.read_only();
 		let channels = readonly_graph.channels();
@@ -996,7 +996,8 @@ async fn test_full_snapshot_mutiny_scenario() {
 
 	{ // sync after initial seed
 		let delta = calculate_delta(network_graph_arc.clone(), 0, None, logger.clone()).await;
-		let serialization = serialize_delta(&delta, 1, logger.clone());
+		let delta_time = current_time() % config::snapshot_generation_interval();
+		let serialization = serialize_delta(&delta, delta_time, 1, logger.clone());
 		logger.assert_log_contains("rapid_gossip_sync_server", "announcement channel count: 1", 1);
 
 		let channel_count = network_graph_arc.read_only().channels().len();
@@ -1010,8 +1011,7 @@ async fn test_full_snapshot_mutiny_scenario() {
 		let update_result = rgs.update_network_graph(&serialization.data).unwrap();
 		println!("update result: {}", update_result);
 		// the update result must be a multiple of our snapshot granularity
-		assert_eq!(update_result % config::snapshot_generation_interval(), 0);
-		assert!(update_result < timestamp);
+		assert_eq!(update_result, delta_time);
 
 		let timestamp_delta = timestamp - update_result;
 		println!("timestamp delta: {}", timestamp_delta);
@@ -1110,7 +1110,8 @@ async fn test_full_snapshot_interlaced_channel_timestamps() {
 
 	{ // sync after initial seed
 		let delta = calculate_delta(network_graph_arc.clone(), 0, None, logger.clone()).await;
-		let serialization = serialize_delta(&delta, 1, logger.clone());
+		let delta_time = current_time() % config::snapshot_generation_interval();
+		let serialization = serialize_delta(&delta, delta_time, 1, logger.clone());
 		logger.assert_log_contains("rapid_gossip_sync_server", "announcement channel count: 2", 1);
 
 		let channel_count = network_graph_arc.read_only().channels().len();
@@ -1123,8 +1124,7 @@ async fn test_full_snapshot_interlaced_channel_timestamps() {
 		let rgs = RapidGossipSync::new(client_graph_arc.clone(), logger.clone());
 		let update_result = rgs.update_network_graph(&serialization.data).unwrap();
 		// the update result must be a multiple of our snapshot granularity
-		assert_eq!(update_result % config::snapshot_generation_interval(), 0);
-		assert!(update_result < timestamp);
+		assert_eq!(update_result, delta_time);
 
 		let readonly_graph = client_graph_arc.read_only();
 		let channels = readonly_graph.channels();