From 2a7c92b16012521144a6e66ff669e6434a93f5e0 Mon Sep 17 00:00:00 2001 From: Matt Corallo Date: Wed, 20 Nov 2024 22:48:18 +0000 Subject: [PATCH] Store TXO values --- src/config.rs | 5 ++++- src/downloader.rs | 11 ++++++++++- src/persistence.rs | 10 +++++++--- src/types.rs | 2 +- src/verifier.rs | 25 +++++++++++++++++++++---- 5 files changed, 43 insertions(+), 10 deletions(-) diff --git a/src/config.rs b/src/config.rs index 54c7b01..2a83d65 100644 --- a/src/config.rs +++ b/src/config.rs @@ -14,7 +14,7 @@ use lightning::util::ser::Readable; use lightning_block_sync::http::HttpEndpoint; use tokio_postgres::Config; -pub(crate) const SCHEMA_VERSION: i32 = 14; +pub(crate) const SCHEMA_VERSION: i32 = 15; pub(crate) const SYMLINK_GRANULARITY_INTERVAL: u32 = 3600 * 3; // three hours pub(crate) const MAX_SNAPSHOT_SCOPE: u32 = 3600 * 24 * 21; // three weeks // generate symlinks based on a 3-hour-granularity @@ -313,6 +313,9 @@ pub(crate) async fn upgrade_db(schema: i32, client: &mut tokio_postgres::Client) tx.execute("UPDATE config SET db_schema = 14 WHERE id = 1", &[]).await.unwrap(); tx.commit().await.unwrap(); } + if schema >= 1 && schema <= 14 { + client.execute("ALTER TABLE channel_announcements ADD COLUMN funding_amount_sats bigint DEFAULT null", &[]).await.unwrap(); + } if schema <= 1 || schema > SCHEMA_VERSION { panic!("Unknown schema in db: {}, we support up to {}", schema, SCHEMA_VERSION); } diff --git a/src/downloader.rs b/src/downloader.rs index 49e3019..696ae43 100644 --- a/src/downloader.rs +++ b/src/downloader.rs @@ -64,7 +64,16 @@ impl GossipRouter where L::Target: Logger { counter.channel_announcements += 1; } - let gossip_message = GossipMessage::ChannelAnnouncement(msg, None); + let mut funding_amount_sats = self.verifier.get_cached_funding_value(msg.contents.short_channel_id); + if funding_amount_sats.is_none() { + tokio::task::block_in_place(move || { tokio::runtime::Handle::current().block_on(async { + funding_amount_sats = self.verifier.retrieve_funding_value(msg.contents.short_channel_id).await.ok(); + })}); + } + let funding_amount_sats = funding_amount_sats + .expect("If we've accepted a ChannelAnnouncement, we must be able to fetch the TXO for it"); + + let gossip_message = GossipMessage::ChannelAnnouncement(msg, funding_amount_sats, None); if let Err(err) = self.sender.try_send(gossip_message) { let gossip_message = match err { TrySendError::Full(msg)|TrySendError::Closed(msg) => msg }; tokio::task::block_in_place(move || { tokio::runtime::Handle::current().block_on(async move { diff --git a/src/persistence.rs b/src/persistence.rs index 0db3091..a7d96c0 100644 --- a/src/persistence.rs +++ b/src/persistence.rs @@ -185,7 +185,7 @@ impl GossipPersister where L::Target: Logger { #[cfg(test)] tasks_spawned.push(_task); }, - GossipMessage::ChannelAnnouncement(announcement, seen_override) => { + GossipMessage::ChannelAnnouncement(announcement, funding_value, seen_override) => { let scid = announcement.contents.short_channel_id as i64; // start with the type prefix, which is already known a priori @@ -198,19 +198,23 @@ impl GossipPersister where L::Target: Logger { .execute("INSERT INTO channel_announcements (\ short_channel_id, \ announcement_signed, \ + funding_amount_sats, \ seen \ - ) VALUES ($1, $2, TO_TIMESTAMP($3)) ON CONFLICT (short_channel_id) DO NOTHING", &[ + ) VALUES ($1, $2, $3, TO_TIMESTAMP($3)) ON CONFLICT (short_channel_id) DO NOTHING", &[ &scid, &announcement_signed, + &(funding_value as i64), &(seen_override.unwrap() as f64) ])).await.unwrap().unwrap(); } else { tokio::time::timeout(POSTGRES_INSERT_TIMEOUT, client .execute("INSERT INTO channel_announcements (\ short_channel_id, \ + funding_amount_sats, \ announcement_signed \ - ) VALUES ($1, $2) ON CONFLICT (short_channel_id) DO NOTHING", &[ + ) VALUES ($1, $2, $3) ON CONFLICT (short_channel_id) DO NOTHING", &[ &scid, + &(funding_value as i64), &announcement_signed ])).await.unwrap().unwrap(); } diff --git a/src/types.rs b/src/types.rs index f38a376..b068448 100644 --- a/src/types.rs +++ b/src/types.rs @@ -16,7 +16,7 @@ pub(crate) type GossipPeerManager = Arc), // the second element is an optional override for the seen value - ChannelAnnouncement(ChannelAnnouncement, Option), + ChannelAnnouncement(ChannelAnnouncement, u64, Option), ChannelUpdate(ChannelUpdate, Option), } diff --git a/src/verifier.rs b/src/verifier.rs index 8256916..a7e3e13 100644 --- a/src/verifier.rs +++ b/src/verifier.rs @@ -1,3 +1,4 @@ +use std::collections::HashMap; use std::io::ErrorKind; use std::ops::Deref; use std::sync::Arc; @@ -23,6 +24,7 @@ pub(crate) struct ChainVerifier where graph: Arc>, outbound_gossiper: Arc>, Arc, L>>, peer_handler: Mutex>>, + scid_value: Arc>>, logger: L } @@ -35,14 +37,24 @@ impl ChainVerifier where L::Target: outbound_gossiper, graph, peer_handler: Mutex::new(None), - logger + scid_value: Arc::new(Mutex::new(HashMap::new())), + logger, } } pub(crate) fn set_ph(&self, peer_handler: GossipPeerManager) { *self.peer_handler.lock().unwrap() = Some(peer_handler); } - async fn retrieve_utxo(client: Arc, short_channel_id: u64, logger: L) -> Result { + pub(crate) fn get_cached_funding_value(&self, scid: u64) -> Option { + self.scid_value.lock().unwrap().get(&scid).map(|v| *v) + } + + pub(crate) async fn retrieve_funding_value(&self, scid: u64) -> Result { + Self::retrieve_cache_txo(Arc::clone(&self.rest_client), Some(Arc::clone(&self.scid_value)), scid, self.logger.clone()) + .await.map(|txo| txo.value.to_sat()) + } + + async fn retrieve_cache_txo(client: Arc, scid_value: Option>>>, short_channel_id: u64, logger: L) -> Result { let block_height = (short_channel_id >> 5 * 8) as u32; // block height is most significant three bytes let transaction_index = ((short_channel_id >> 2 * 8) & 0xffffff) as u32; let output_index = (short_channel_id & 0xffff) as u16; @@ -57,7 +69,11 @@ impl ChainVerifier where L::Target: log_error!(logger, "Could't find output {} in transaction {}", output_index, transaction.compute_txid()); return Err(UtxoLookupError::UnknownTx); } - Ok(transaction.output.swap_remove(output_index as usize)) + let txo = transaction.output.swap_remove(output_index as usize); + if let Some(scid_value) = scid_value { + scid_value.lock().unwrap().insert(short_channel_id, txo.value.to_sat()); + } + Ok(txo) } async fn retrieve_block(client: Arc, block_height: u32, logger: L) -> Result { @@ -99,10 +115,11 @@ impl UtxoLookup for ChainVerifier w let graph_ref = Arc::clone(&self.graph); let client_ref = Arc::clone(&self.rest_client); let gossip_ref = Arc::clone(&self.outbound_gossiper); + let scid_value_cache_ref = Arc::clone(&self.scid_value); let pm_ref = self.peer_handler.lock().unwrap().clone(); let logger_ref = self.logger.clone(); tokio::spawn(async move { - let res = Self::retrieve_utxo(client_ref, short_channel_id, logger_ref).await; + let res = Self::retrieve_cache_txo(client_ref, Some(scid_value_cache_ref), short_channel_id, logger_ref).await; fut.resolve(&*graph_ref, &*gossip_ref, res); if let Some(pm) = pm_ref { pm.process_events(); } }); -- 2.39.5