Use separate schemas for each unit test.

Author: arik-so

src/lib.rs

@@ -54,7 +54,7 @@ const GOSSIP_PREFIX: [u8; 4] = [76, 68, 75, 1];
 
 pub struct RapidSyncProcessor<L: Deref> where L::Target: Logger {
 	network_graph: Arc<NetworkGraph<L>>,
-	logger: L
+	logger: L,
 }
 
 pub struct SerializedResponse {
@@ -86,7 +86,7 @@ impl<L: Deref + Clone + Send + Sync + 'static> RapidSyncProcessor<L> where L::Ta
 		let arc_network_graph = Arc::new(network_graph);
 		Self {
 			network_graph: arc_network_graph,
-			logger
+			logger,
 		}
 	}
 
@@ -145,6 +145,13 @@ pub(crate) async fn connect_to_db() -> Client {
 		}
 	});
 
+	if 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
 }

src/lookup.rs

@@ -84,9 +84,8 @@ pub(super) async fn fetch_channel_announcements<L: Deref>(delta_set: &mut DeltaS
 			.map(|c| c.1.announcement_message.as_ref().unwrap().contents.short_channel_id as i64)
 			.collect::<Vec<_>>()
 	};
-	if cfg!(test) {
-		log_info!(logger, "Channel IDs: {:?}", channel_ids);
-	}
+	#[cfg(test)]
+	log_info!(logger, "Channel IDs: {:?}", channel_ids);
 	log_info!(logger, "Last sync timestamp: {}", last_sync_timestamp);
 	let last_sync_timestamp_float = last_sync_timestamp as f64;
 
@@ -166,7 +165,7 @@ pub(super) async fn fetch_channel_announcements<L: Deref>(delta_set: &mut DeltaS
 			let scid: i64 = current_row.get("short_channel_id");
 			let current_seen_timestamp = current_row.get::<_, i64>("seen") as u32;
 
-			log_trace!(logger, "Channel {} with first update to complete bidirectional data since last sync seen at: {}", scid, current_seen_timestamp);
+			log_gossip!(logger, "Channel {} with first update to complete bidirectional data since last sync seen at: {}", scid, current_seen_timestamp);
 
 			// the newer of the two oldest seen directional updates came after last sync timestamp
 			let current_channel_delta = delta_set.entry(scid as u64).or_insert(ChannelDelta::default());
@@ -209,7 +208,7 @@ pub(super) async fn fetch_channel_announcements<L: Deref>(delta_set: &mut DeltaS
 			let current_row = row_res.unwrap();
 			let scid: i64 = current_row.get("short_channel_id");
 
-			log_trace!(logger, "Channel {} with newest update in less recently updated direction being at least 6 days ago", scid);
+			log_gossip!(logger, "Channel {} with newest update in less recently updated direction being at least 6 days ago", scid);
 
 			// annotate this channel as requiring that reminders be sent to the client
 			let current_channel_delta = delta_set.entry(scid as u64).or_insert(ChannelDelta::default());
@@ -357,14 +356,14 @@ pub(super) async fn fetch_channel_updates<L: Deref>(delta_set: &mut DeltaSet, cl
 					seen: current_seen_timestamp,
 					update: unsigned_channel_update.clone(),
 				});
-				log_trace!(logger, "Channel {} latest update in direction 0: {} (seen: {})", scid, unsigned_channel_update.timestamp, current_seen_timestamp)
+				log_gossip!(logger, "Channel {} latest update in direction 0: {} (seen: {})", scid, unsigned_channel_update.timestamp, current_seen_timestamp)
 			} else if direction && !previously_seen_directions.1 {
 				previously_seen_directions.1 = true;
 				update_delta.latest_update_after_seen = Some(UpdateDelta {
 					seen: current_seen_timestamp,
 					update: unsigned_channel_update.clone(),
 				});
-				log_trace!(logger, "Channel {} latest update in direction 1: {} (seen: {})", scid, unsigned_channel_update.timestamp, current_seen_timestamp)
+				log_gossip!(logger, "Channel {} latest update in direction 1: {} (seen: {})", scid, unsigned_channel_update.timestamp, current_seen_timestamp)
 			}
 		}
 

src/serialization.rs

@@ -5,12 +5,11 @@ use std::ops::Deref;
 
 use bitcoin::BlockHash;
 use lightning::ln::msgs::{UnsignedChannelAnnouncement, UnsignedChannelUpdate};
-use lightning::log_trace;
+use lightning::log_gossip;
 use lightning::util::logger::Logger;
 use lightning::util::ser::{BigSize, Writeable};
 use crate::config;
 
-use crate::config;
 use crate::lookup::{DeltaSet, DirectedUpdateDelta};
 
 pub(super) struct SerializationSet {
@@ -143,14 +142,12 @@ pub(super) fn serialize_delta_set<L: Deref>(delta_set: DeltaSet, last_sync_times
 		let current_announcement_seen = channel_announcement_delta.seen;
 		let is_new_announcement = current_announcement_seen >= last_sync_timestamp;
 		let is_newly_included_announcement = if let Some(first_update_seen) = channel_delta.first_bidirectional_updates_seen {
-			#[cfg(test)]
-			log_trace!(logger, "Channel {} first bidirectional update seen: {}", scid, first_update_seen);
+			log_gossip!(logger, "Channel {} first bidirectional update seen: {}", scid, first_update_seen);
 			first_update_seen >= last_sync_timestamp
 		} else {
 			false
 		};
-		#[cfg(test)]
-		log_trace!(logger, "Channel {} announcement seen at {} (new: {}, newly included: {})", scid, current_announcement_seen, is_new_announcement, is_newly_included_announcement);
+		log_gossip!(logger, "Channel {} announcement seen at {} (new: {}, newly included: {})", scid, current_announcement_seen, is_new_announcement, is_newly_included_announcement);
 		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);

src/tests/mod.rs

@@ -2,12 +2,14 @@
 #![allow(unused_imports)]
 //! Multi-module tests that use database fixtures
 
+use std::cell::{RefCell, RefMut};
 use std::sync::Arc;
 use std::time::{SystemTime, UNIX_EPOCH};
 use bitcoin::{BlockHash, Network};
 use bitcoin::secp256k1::ecdsa::Signature;
 use bitcoin::secp256k1::{Secp256k1, SecretKey};
 use bitcoin::hashes::Hash;
+use bitcoin::hashes::hex::ToHex;
 use bitcoin::hashes::sha256d::Hash as Sha256dHash;
 use lightning::ln::features::ChannelFeatures;
 use lightning::ln::msgs::{ChannelAnnouncement, ChannelUpdate, UnsignedChannelAnnouncement, UnsignedChannelUpdate};
@@ -21,6 +23,10 @@ use crate::types::{GossipMessage, tests::TestLogger};
 
 const CLIENT_BACKDATE_INTERVAL: u32 = 3600 * 24 * 7; // client backdates RGS by a week
 
+thread_local! {
+	static DB_TEST_SCHEMA:RefCell<Option<String>> = RefCell::new(None);
+}
+
 fn blank_signature() -> Signature {
 	Signature::from_compact(&[0u8; 64]).unwrap()
 }
@@ -33,6 +39,29 @@ fn current_time() -> u32 {
 	SystemTime::now().duration_since(UNIX_EPOCH).expect("Time must be > 1970").as_secs() as u32
 }
 
+pub(crate) fn db_test_schema() -> String {
+	DB_TEST_SCHEMA.with(|suffix_reference| {
+		let mut suffix_option = suffix_reference.borrow_mut();
+		match suffix_option.as_ref() {
+			None => {
+				let current_time = SystemTime::now();
+				let unix_time = current_time.duration_since(UNIX_EPOCH).expect("Time went backwards");
+				let timestamp_seconds = unix_time.as_secs();
+				let timestamp_nanos = unix_time.as_nanos();
+				let preimage = format!("{}", timestamp_nanos);
+				let suffix = Sha256dHash::hash(preimage.as_bytes()).into_inner().to_hex();
+				// the schema must start with a letter
+				let schema = format!("test_{}_{}", timestamp_seconds, suffix);
+				suffix_option.replace(schema.clone());
+				schema
+			}
+			Some(suffix) => {
+				suffix.clone()
+			}
+		}
+	})
+}
+
 fn generate_announcement(short_channel_id: u64) -> ChannelAnnouncement {
 	let secp_context = Secp256k1::new();
 
@@ -88,25 +117,13 @@ fn generate_update(scid: u64, direction: bool, timestamp: u32, expiry_delta: u16
 }
 
 async fn clean_test_db() {
-	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);
-		}
-	});
-
-	client.query("TRUNCATE TABLE channel_announcements RESTART IDENTITY CASCADE", &[]).await.unwrap();
-	client.query("TRUNCATE TABLE channel_updates RESTART IDENTITY CASCADE", &[]).await.unwrap();
-	client.query("TRUNCATE TABLE config RESTART IDENTITY CASCADE", &[]).await.unwrap();
+	let client = crate::connect_to_db().await;
+	let schema = db_test_schema();
+	client.execute(&format!("DROP SCHEMA IF EXISTS {} CASCADE", schema), &[]).await.unwrap();
 }
 
 #[tokio::test]
 async fn test_trivial_setup() {
-	// start off with a clean slate
-	clean_test_db().await;
-
 	let logger = Arc::new(TestLogger::new());
 	let network_graph = NetworkGraph::new(Network::Bitcoin, logger.clone());
 	let network_graph_arc = Arc::new(network_graph);
@@ -149,12 +166,12 @@ 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_CALCULATION_INTERVAL, 0);
+	assert_eq!(update_result % config::snapshot_generation_interval(), 0);
 	assert!(update_result < timestamp);
 
 	let timestamp_delta = timestamp - update_result;
 	println!("timestamp delta: {}", timestamp_delta);
-	assert!(timestamp_delta < config::SNAPSHOT_CALCULATION_INTERVAL);
+	assert!(timestamp_delta < config::snapshot_generation_interval());
 
 	let readonly_graph = client_graph_arc.read_only();
 	let channels = readonly_graph.channels();
@@ -200,7 +217,7 @@ async fn test_unidirectional_intermediate_update_consideration() {
 		network_graph_arc.update_channel_unsigned(&update_3.contents).unwrap();
 		network_graph_arc.update_channel_unsigned(&update_4.contents).unwrap();
 
-		receiver.send(GossipMessage::ChannelAnnouncement(announcement)).await.unwrap();
+		receiver.send(GossipMessage::ChannelAnnouncement(announcement, Some(current_timestamp))).await.unwrap();
 		receiver.send(GossipMessage::ChannelUpdate(update_1)).await.unwrap();
 		receiver.send(GossipMessage::ChannelUpdate(update_2)).await.unwrap();
 		receiver.send(GossipMessage::ChannelUpdate(update_3)).await.unwrap();
@@ -247,7 +264,7 @@ async fn test_unidirectional_intermediate_update_consideration() {
 	println!("last sync timestamp: {}", last_sync_timestamp);
 	println!("next timestamp: {}", next_timestamp);
 	// the update result must be a multiple of our snapshot granularity
-	assert_eq!(next_timestamp % config::SNAPSHOT_CALCULATION_INTERVAL, 0);
+	assert_eq!(next_timestamp % config::snapshot_generation_interval(), 0);
 
 	let readonly_graph = client_graph_arc.read_only();
 	let channels = readonly_graph.channels();