X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;ds=sidebyside;f=src%2Fconfig.rs;h=b0e3a7ce7722dafb236001df3e6bb46a1f6a60f7;hb=refs%2Fheads%2F2023-07-further-opt;hp=5369e8d04c89bf15eda6e1a4f76eb25a319c3521;hpb=63c51a6ed448fdc1c4d3656ebc8cce1d1e2e1d13;p=rapid-gossip-sync-server

diff --git a/src/config.rs b/src/config.rs
index 5369e8d..b0e3a7c 100644
--- a/src/config.rs
+++ b/src/config.rs
@@ -1,22 +1,47 @@
+use crate::hex_utils;
+
 use std::convert::TryInto;
 use std::env;
-use std::net::SocketAddr;
 use std::io::Cursor;
+use std::net::{SocketAddr, ToSocketAddrs};
+use std::time::Duration;
+
+use bitcoin::Network;
+use bitcoin::hashes::hex::FromHex;
 use bitcoin::secp256k1::PublicKey;
+use futures::stream::{FuturesUnordered, StreamExt};
 use lightning::ln::msgs::ChannelAnnouncement;
 use lightning::util::ser::Readable;
 use lightning_block_sync::http::HttpEndpoint;
 use tokio_postgres::Config;
-use crate::hex_utils;
 
-use futures::stream::{FuturesUnordered, StreamExt};
-
-pub(crate) const SCHEMA_VERSION: i32 = 6;
+pub(crate) const SCHEMA_VERSION: i32 = 11;
 pub(crate) const SNAPSHOT_CALCULATION_INTERVAL: u32 = 3600 * 24; // every 24 hours, in seconds
+/// If the last update in either direction was more than six days ago, we send a reminder
+/// That reminder may be either in the form of a channel announcement, or in the form of empty
+/// updates in both directions.
+pub(crate) const CHANNEL_REMINDER_AGE: Duration = Duration::from_secs(6 * 24 * 60 * 60);
 pub(crate) const DOWNLOAD_NEW_GOSSIP: bool = true;
 
-pub(crate) fn network_graph_cache_path() -> &'static str {
-	"./res/network_graph.bin"
+pub(crate) fn network() -> Network {
+	let network = env::var("RAPID_GOSSIP_SYNC_SERVER_NETWORK").unwrap_or("bitcoin".to_string()).to_lowercase();
+	match network.as_str() {
+		"mainnet" => Network::Bitcoin,
+		"bitcoin" => Network::Bitcoin,
+		"testnet" => Network::Testnet,
+		"signet" => Network::Signet,
+		"regtest" => Network::Regtest,
+		_ => panic!("Invalid network"),
+	}
+}
+
+pub(crate) fn network_graph_cache_path() -> String {
+	format!("{}/network_graph.bin", cache_path())
+}
+
+pub(crate) fn cache_path() -> String {
+	let path = env::var("RAPID_GOSSIP_SYNC_SERVER_CACHES_PATH").unwrap_or("./res".to_string()).to_lowercase();
+	path
 }
 
 pub(crate) fn db_connection_config() -> Config {
@@ -60,33 +85,30 @@ pub(crate) fn db_announcement_table_creation_query() -> &'static str {
 }
 
 pub(crate) fn db_channel_update_table_creation_query() -> &'static str {
-	// We'll run out of room in composite index at block 8,388,608 or in the year 2286
 	"CREATE TABLE IF NOT EXISTS channel_updates (
 		id SERIAL PRIMARY KEY,
-		composite_index character(29) UNIQUE,
 		short_channel_id bigint NOT NULL,
-		timestamp bigint,
-		channel_flags smallint,
+		timestamp bigint NOT NULL,
+		channel_flags smallint NOT NULL,
 		direction boolean NOT NULL,
-		disable boolean,
-		cltv_expiry_delta integer,
-		htlc_minimum_msat bigint,
-		fee_base_msat integer,
-		fee_proportional_millionths integer,
-		htlc_maximum_msat bigint,
-		blob_signed BYTEA,
+		disable boolean NOT NULL,
+		cltv_expiry_delta integer NOT NULL,
+		htlc_minimum_msat bigint NOT NULL,
+		fee_base_msat integer NOT NULL,
+		fee_proportional_millionths integer NOT NULL,
+		htlc_maximum_msat bigint NOT NULL,
+		blob_signed BYTEA NOT NULL,
 		seen timestamp NOT NULL DEFAULT NOW()
 	)"
 }
 
 pub(crate) fn db_index_creation_query() -> &'static str {
 	"
-	CREATE INDEX IF NOT EXISTS channels_seen ON channel_announcements(seen);
-	CREATE INDEX IF NOT EXISTS channel_updates_scid ON channel_updates(short_channel_id);
-	CREATE INDEX IF NOT EXISTS channel_updates_direction ON channel_updates (short_channel_id, direction);
-	CREATE INDEX IF NOT EXISTS channel_updates_seen ON channel_updates(seen);
-	CREATE INDEX IF NOT EXISTS channel_updates_scid_seen ON channel_updates(short_channel_id, seen);
-	CREATE INDEX IF NOT EXISTS channel_updates_scid_dir_seen ON channel_updates(short_channel_id ASC, direction ASC, seen DESC);
+	CREATE INDEX IF NOT EXISTS channel_updates_seen_with_id_direction_blob ON channel_updates(seen) INCLUDE (id, direction, blob_signed);
+	CREATE INDEX IF NOT EXISTS channel_updates_seen_scid ON channel_updates(seen, short_channel_id);
+	CREATE INDEX IF NOT EXISTS channel_updates_scid_dir_seen_asc ON channel_updates(short_channel_id, direction, seen);
+	CREATE INDEX IF NOT EXISTS channel_updates_scid_dir_seen_desc_with_id ON channel_updates(short_channel_id ASC, direction ASC, seen DESC) INCLUDE (id);
+	CREATE UNIQUE INDEX IF NOT EXISTS channel_updates_key ON channel_updates (short_channel_id, direction, timestamp);
 	"
 }
 
@@ -127,8 +149,6 @@ pub(crate) async fn upgrade_db(schema: i32, client: &mut tokio_postgres::Client)
 			}
 			while let Some(_) = updates.next().await { }
 		}
-		tx.execute("CREATE INDEX channel_updates_scid ON channel_updates(short_channel_id)", &[]).await.unwrap();
-		tx.execute("CREATE INDEX channel_updates_direction ON channel_updates (short_channel_id, direction)", &[]).await.unwrap();
 		tx.execute("ALTER TABLE channel_updates ALTER short_channel_id DROP DEFAULT", &[]).await.unwrap();
 		tx.execute("ALTER TABLE channel_updates ALTER short_channel_id SET NOT NULL", &[]).await.unwrap();
 		tx.execute("ALTER TABLE channel_updates ALTER direction DROP DEFAULT", &[]).await.unwrap();
@@ -175,24 +195,112 @@ pub(crate) async fn upgrade_db(schema: i32, client: &mut tokio_postgres::Client)
 		tx.execute("UPDATE config SET db_schema = 6 WHERE id = 1", &[]).await.unwrap();
 		tx.commit().await.unwrap();
 	}
+	if schema >= 1 && schema <= 6 {
+		let tx = client.transaction().await.unwrap();
+		tx.execute("ALTER TABLE channel_updates DROP COLUMN composite_index", &[]).await.unwrap();
+		tx.execute("ALTER TABLE channel_updates ALTER timestamp SET NOT NULL", &[]).await.unwrap();
+		tx.execute("ALTER TABLE channel_updates ALTER channel_flags SET NOT NULL", &[]).await.unwrap();
+		tx.execute("ALTER TABLE channel_updates ALTER disable SET NOT NULL", &[]).await.unwrap();
+		tx.execute("ALTER TABLE channel_updates ALTER cltv_expiry_delta SET NOT NULL", &[]).await.unwrap();
+		tx.execute("ALTER TABLE channel_updates ALTER htlc_minimum_msat SET NOT NULL", &[]).await.unwrap();
+		tx.execute("ALTER TABLE channel_updates ALTER fee_base_msat SET NOT NULL", &[]).await.unwrap();
+		tx.execute("ALTER TABLE channel_updates ALTER fee_proportional_millionths SET NOT NULL", &[]).await.unwrap();
+		tx.execute("ALTER TABLE channel_updates ALTER htlc_maximum_msat SET NOT NULL", &[]).await.unwrap();
+		tx.execute("ALTER TABLE channel_updates ALTER blob_signed SET NOT NULL", &[]).await.unwrap();
+		tx.execute("CREATE UNIQUE INDEX channel_updates_key ON channel_updates (short_channel_id, direction, timestamp)", &[]).await.unwrap();
+		tx.execute("UPDATE config SET db_schema = 7 WHERE id = 1", &[]).await.unwrap();
+		tx.commit().await.unwrap();
+	}
+	if schema >= 1 && schema <= 7 {
+		let tx = client.transaction().await.unwrap();
+		tx.execute("DROP INDEX IF EXISTS channels_seen", &[]).await.unwrap();
+		tx.execute("DROP INDEX IF EXISTS channel_updates_scid", &[]).await.unwrap();
+		tx.execute("DROP INDEX IF EXISTS channel_updates_direction", &[]).await.unwrap();
+		tx.execute("DROP INDEX IF EXISTS channel_updates_seen", &[]).await.unwrap();
+		tx.execute("DROP INDEX IF EXISTS channel_updates_scid_seen", &[]).await.unwrap();
+		tx.execute("DROP INDEX IF EXISTS channel_updates_scid_dir_seen", &[]).await.unwrap();
+		tx.execute("UPDATE config SET db_schema = 8 WHERE id = 1", &[]).await.unwrap();
+		tx.commit().await.unwrap();
+	}
+	if schema >= 1 && schema <= 8 {
+		let tx = client.transaction().await.unwrap();
+		tx.execute("DROP INDEX IF EXISTS channel_updates_seen", &[]).await.unwrap();
+		tx.execute("DROP INDEX IF EXISTS channel_updates_scid_seen", &[]).await.unwrap();
+		tx.execute("UPDATE config SET db_schema = 9 WHERE id = 1", &[]).await.unwrap();
+		tx.commit().await.unwrap();
+	}
+	if schema >= 1 && schema <= 9 {
+		let tx = client.transaction().await.unwrap();
+		tx.execute("DROP INDEX IF EXISTS channel_updates_scid_dir_seen", &[]).await.unwrap();
+		tx.execute("UPDATE config SET db_schema = 10 WHERE id = 1", &[]).await.unwrap();
+		tx.commit().await.unwrap();
+	}
+	if schema >= 1 && schema <= 10 {
+		let tx = client.transaction().await.unwrap();
+		tx.execute("DROP INDEX IF EXISTS channel_updates_id_with_scid_dir_blob", &[]).await.unwrap();
+		tx.execute("UPDATE config SET db_schema = 11 WHERE id = 1", &[]).await.unwrap();
+		tx.commit().await.unwrap();
+	}
 	if schema <= 1 || schema > SCHEMA_VERSION {
 		panic!("Unknown schema in db: {}, we support up to {}", schema, SCHEMA_VERSION);
 	}
+	// PostgreSQL (at least v13, but likely later versions as well) handles insert-only tables
+	// *very* poorly. After some number of inserts, it refuses to rely on indexes, assuming them to
+	// be possibly-stale, until a VACUUM happens. Thus, we set the vacuum factor really low here,
+	// pushing PostgreSQL to vacuum often.
+	// See https://www.cybertec-postgresql.com/en/postgresql-autovacuum-insert-only-tables/
+	let _ = client.execute("ALTER TABLE channel_updates SET ( autovacuum_vacuum_insert_scale_factor = 0.005 );", &[]).await;
+	let _ = client.execute("ALTER TABLE channel_announcements SET ( autovacuum_vacuum_insert_scale_factor = 0.005 );", &[]).await;
 }
 
-/// EDIT ME
 pub(crate) fn ln_peers() -> Vec<(PublicKey, SocketAddr)> {
-	vec![
-		// Bitfinex
-		// (hex_utils::to_compressed_pubkey("033d8656219478701227199cbd6f670335c8d408a92ae88b962c49d4dc0e83e025").unwrap(), "34.65.85.39:9735".parse().unwrap()),
+	const WALLET_OF_SATOSHI: &str = "035e4ff418fc8b5554c5d9eea66396c227bd429a3251c8cbc711002ba215bfc226@170.75.163.209:9735";
+	let list = env::var("LN_PEERS").unwrap_or(WALLET_OF_SATOSHI.to_string());
+	let mut peers = Vec::new();
+	for peer_info in list.split(',') {
+		peers.push(resolve_peer_info(peer_info).expect("Invalid peer info in LN_PEERS"));
+	}
+	peers
+}
+
+fn resolve_peer_info(peer_info: &str) -> Result<(PublicKey, SocketAddr), &str> {
+	let mut peer_info = peer_info.splitn(2, '@');
 
-		// Matt Corallo
-		// (hex_utils::to_compressed_pubkey("03db10aa09ff04d3568b0621750794063df401e6853c79a21a83e1a3f3b5bfb0c8").unwrap(), "69.59.18.80:9735".parse().unwrap())
+	let pubkey = peer_info.next().ok_or("Invalid peer info. Should be formatted as: `pubkey@host:port`")?;
+	let pubkey = Vec::from_hex(pubkey).map_err(|_| "Invalid node pubkey")?;
+	let pubkey = PublicKey::from_slice(&pubkey).map_err(|_| "Invalid node pubkey")?;
 
-		// River Financial
-		// (hex_utils::to_compressed_pubkey("03037dc08e9ac63b82581f79b662a4d0ceca8a8ca162b1af3551595b8f2d97b70a").unwrap(), "104.196.249.140:9735".parse().unwrap())
+	let socket_address = peer_info.next().ok_or("Invalid peer info. Should be formatted as: `pubkey@host:port`")?;
+	let socket_address = socket_address
+		.to_socket_addrs()
+		.map_err(|_| "Cannot resolve node address")?
+		.next()
+		.ok_or("Cannot resolve node address")?;
+
+	Ok((pubkey, socket_address))
+}
 
-		// Wallet of Satoshi | 035e4ff418fc8b5554c5d9eea66396c227bd429a3251c8cbc711002ba215bfc226@170.75.163.209:9735
-		(hex_utils::to_compressed_pubkey("035e4ff418fc8b5554c5d9eea66396c227bd429a3251c8cbc711002ba215bfc226").unwrap(), "170.75.163.209:9735".parse().unwrap())
-	]
+#[cfg(test)]
+mod tests {
+	use super::resolve_peer_info;
+	use bitcoin::hashes::hex::ToHex;
+
+	#[test]
+	fn test_resolve_peer_info() {
+		let wallet_of_satoshi = "035e4ff418fc8b5554c5d9eea66396c227bd429a3251c8cbc711002ba215bfc226@170.75.163.209:9735";
+		let (pubkey, socket_address) = resolve_peer_info(wallet_of_satoshi).unwrap();
+		assert_eq!(pubkey.serialize().to_hex(), "035e4ff418fc8b5554c5d9eea66396c227bd429a3251c8cbc711002ba215bfc226");
+		assert_eq!(socket_address.to_string(), "170.75.163.209:9735");
+
+		let ipv6 = "033d8656219478701227199cbd6f670335c8d408a92ae88b962c49d4dc0e83e025@[2001:db8::1]:80";
+		let (pubkey, socket_address) = resolve_peer_info(ipv6).unwrap();
+		assert_eq!(pubkey.serialize().to_hex(), "033d8656219478701227199cbd6f670335c8d408a92ae88b962c49d4dc0e83e025");
+		assert_eq!(socket_address.to_string(), "[2001:db8::1]:80");
+
+		let localhost = "033d8656219478701227199cbd6f670335c8d408a92ae88b962c49d4dc0e83e025@localhost:9735";
+		let (pubkey, socket_address) = resolve_peer_info(localhost).unwrap();
+		assert_eq!(pubkey.serialize().to_hex(), "033d8656219478701227199cbd6f670335c8d408a92ae88b962c49d4dc0e83e025");
+		let socket_address = socket_address.to_string();
+		assert!(socket_address == "127.0.0.1:9735" || socket_address == "[::1]:9735");
+	}
 }