Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions bindings/ldk_node.udl
Original file line numberDiff line numberDiff line change
Expand Up@@ -81,6 +81,7 @@ interface Builder {
void set_chain_source_bitcoind_rest(string rest_host, u16 rest_port, string rpc_host, u16 rpc_port, string rpc_user, string rpc_password);
void set_gossip_source_p2p();
void set_gossip_source_rgs(string rgs_server_url);
void set_pathfinding_scores_source(string url);
void set_liquidity_source_lsps1(PublicKey node_id, SocketAddress address, string? token);
void set_liquidity_source_lsps2(PublicKey node_id, SocketAddress address, string? token);
void set_storage_dir_path(string storage_dir_path);
Expand DownExpand Up@@ -330,6 +331,7 @@ dictionary NodeStatus {
u64? latest_onchain_wallet_sync_timestamp;
u64? latest_fee_rate_cache_update_timestamp;
u64? latest_rgs_snapshot_timestamp;
u64? latest_pathfinding_scores_sync_timestamp;
u64? latest_node_announcement_broadcast_timestamp;
u32? latest_channel_monitor_archival_height;
};
Expand Down
63 changes: 54 additions & 9 deletions src/builder.rs
Original file line numberDiff line numberDiff line change
Expand Up@@ -24,10 +24,12 @@ use lightning::io::Cursor;
use lightning::ln::channelmanager::{self, ChainParameters, ChannelManagerReadArgs};
use lightning::ln::msgs::{RoutingMessageHandler, SocketAddress};
use lightning::ln::peer_handler::{IgnoringMessageHandler, MessageHandler};
use lightning::log_trace;
use lightning::routing::gossip::NodeAlias;
use lightning::routing::router::DefaultRouter;
use lightning::routing::scoring::{
ProbabilisticScorer, ProbabilisticScoringDecayParameters, ProbabilisticScoringFeeParameters,
CombinedScorer, ProbabilisticScorer, ProbabilisticScoringDecayParameters,
ProbabilisticScoringFeeParameters,
};
use lightning::sign::{EntropySource, NodeSigner};
use lightning::util::persist::{
Expand All@@ -50,7 +52,9 @@ use crate::event::EventQueue;
use crate::fee_estimator::OnchainFeeEstimator;
use crate::gossip::GossipSource;
use crate::io::sqlite_store::SqliteStore;
use crate::io::utils::{read_node_metrics, write_node_metrics};
use crate::io::utils::{
read_external_pathfinding_scores_from_cache, read_node_metrics, write_node_metrics,
};
use crate::io::vss_store::VssStore;
use crate::io::{
self, PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
Expand DownExpand Up@@ -110,6 +114,11 @@ enum GossipSourceConfig {
RapidGossipSync(String),
}

#[derive(Debug, Clone)]
struct PathfindingScoresSyncConfig {
url: String,
}

#[derive(Debug, Clone, Default)]
struct LiquiditySourceConfig {
// Act as an LSPS1 client connecting to the given service.
Expand DownExpand Up@@ -243,6 +252,7 @@ pub struct NodeBuilder {
log_writer_config: Option<LogWriterConfig>,
async_payments_role: Option<AsyncPaymentsRole>,
runtime_handle: Option<tokio::runtime::Handle>,
pathfinding_scores_sync_config: Option<PathfindingScoresSyncConfig>,
}

impl NodeBuilder {
Expand All@@ -260,6 +270,7 @@ impl NodeBuilder {
let liquidity_source_config = None;
let log_writer_config = None;
let runtime_handle = None;
let pathfinding_scores_sync_config = None;
Self {
config,
entropy_source_config,
Expand All@@ -269,6 +280,7 @@ impl NodeBuilder {
log_writer_config,
runtime_handle,
async_payments_role: None,
pathfinding_scores_sync_config,
}
}

Expand DownExpand Up@@ -411,6 +423,14 @@ impl NodeBuilder {
self
}

/// Configures the [`Node`] instance to source its external scores from the given URL.
Comment thread
joostjager marked this conversation as resolved.
///
/// The external scores are merged into the local scoring system to improve routing.
pub fn set_pathfinding_scores_source(&mut self, url: String) -> &mut Self {
self.pathfinding_scores_sync_config = Some(PathfindingScoresSyncConfig { url });
self
}

/// Configures the [`Node`] instance to source inbound liquidity from the given
/// [bLIP-51 / LSPS1] service.
///
Expand DownExpand Up@@ -718,6 +738,7 @@ impl NodeBuilder {
self.chain_data_source_config.as_ref(),
self.gossip_source_config.as_ref(),
self.liquidity_source_config.as_ref(),
self.pathfinding_scores_sync_config.as_ref(),
self.async_payments_role,
seed_bytes,
runtime,
Expand DownExpand Up@@ -751,6 +772,7 @@ impl NodeBuilder {
self.chain_data_source_config.as_ref(),
self.gossip_source_config.as_ref(),
self.liquidity_source_config.as_ref(),
self.pathfinding_scores_sync_config.as_ref(),
self.async_payments_role,
seed_bytes,
runtime,
Expand DownExpand Up@@ -910,6 +932,13 @@ impl ArcedNodeBuilder {
self.inner.write().unwrap().set_gossip_source_rgs(rgs_server_url);
}

/// Configures the [`Node`] instance to source its external scores from the given URL.
///
/// The external scores are merged into the local scoring system to improve routing.
pub fn set_pathfinding_scores_source(&self, url: String) {
self.inner.write().unwrap().set_pathfinding_scores_source(url);
}

/// Configures the [`Node`] instance to source inbound liquidity from the given
/// [bLIP-51 / LSPS1] service.
///
Expand DownExpand Up@@ -1110,6 +1139,7 @@ fn build_with_store_internal(
config: Arc<Config>, chain_data_source_config: Option<&ChainDataSourceConfig>,
gossip_source_config: Option<&GossipSourceConfig>,
liquidity_source_config: Option<&LiquiditySourceConfig>,
pathfinding_scores_sync_config: Option<&PathfindingScoresSyncConfig>,
async_payments_role: Option<AsyncPaymentsRole>, seed_bytes: [u8; 64], runtime: Arc<Runtime>,
logger: Arc<Logger>, kv_store: Arc<DynStore>,
) -> Result<Node, BuildError> {
Expand DownExpand Up@@ -1365,26 +1395,38 @@ fn build_with_store_internal(
},
};

let scorer = match io::utils::read_scorer(
let local_scorer = match io::utils::read_scorer(
Arc::clone(&kv_store),
Arc::clone(&network_graph),
Arc::clone(&logger),
) {
Ok(scorer) => Arc::new(Mutex::new(scorer)),
Ok(scorer) => scorer,
Err(e) => {
if e.kind() == std::io::ErrorKind::NotFound {
let params = ProbabilisticScoringDecayParameters::default();
Arc::new(Mutex::new(ProbabilisticScorer::new(
params,
Arc::clone(&network_graph),
Arc::clone(&logger),
)))
ProbabilisticScorer::new(params, Arc::clone(&network_graph), Arc::clone(&logger))
} else {
return Err(BuildError::ReadFailed);
}
},
};

let scorer = Arc::new(Mutex::new(CombinedScorer::new(local_scorer)));

// Restore external pathfinding scores from cache if possible.
match read_external_pathfinding_scores_from_cache(Arc::clone(&kv_store), Arc::clone(&logger)) {
Ok(external_scores) => {
scorer.lock().unwrap().merge(external_scores, cur_time);
log_trace!(logger, "External scores from cache merged successfully");
},
Err(e) => {
if e.kind() != std::io::ErrorKind::NotFound {
log_error!(logger, "Error while reading external scores from cache: {}", e);
return Err(BuildError::ReadFailed);
}
},
}

let scoring_fee_params = ProbabilisticScoringFeeParameters::default();
let router = Arc::new(DefaultRouter::new(
Arc::clone(&network_graph),
Expand DownExpand Up@@ -1716,6 +1758,8 @@ fn build_with_store_internal(
let (background_processor_stop_sender, _) = tokio::sync::watch::channel(());
let is_running = Arc::new(RwLock::new(false));

let pathfinding_scores_sync_url = pathfinding_scores_sync_config.map(|c| c.url.clone());

Ok(Node {
runtime,
stop_sender,
Expand All@@ -1734,6 +1778,7 @@ fn build_with_store_internal(
keys_manager,
network_graph,
gossip_source,
pathfinding_scores_sync_url,
liquidity_source,
kv_store,
logger,
Expand Down
6 changes: 6 additions & 0 deletions src/config.rs
Original file line numberDiff line numberDiff line change
Expand Up@@ -63,6 +63,9 @@ pub(crate) const PEER_RECONNECTION_INTERVAL: Duration = Duration::from_secs(60);
// The time in-between RGS sync attempts.
pub(crate) const RGS_SYNC_INTERVAL: Duration = Duration::from_secs(60 * 60);

// The time in-between external scores sync attempts.
pub(crate) const EXTERNAL_PATHFINDING_SCORES_SYNC_INTERVAL: Duration = Duration::from_secs(60 * 60);

// The time in-between node announcement broadcast attempts.
pub(crate) const NODE_ANN_BCAST_INTERVAL: Duration = Duration::from_secs(60 * 60);

Expand DownExpand Up@@ -93,6 +96,9 @@ pub(crate) const RGS_SYNC_TIMEOUT_SECS: u64 = 5;
/// The length in bytes of our wallets' keys seed.
pub const WALLET_KEYS_SEED_LEN: usize = 64;

// The timeout after which we abort a external scores sync operation.
pub(crate) const EXTERNAL_PATHFINDING_SCORES_SYNC_TIMEOUT_SECS: u64 = 5;

#[derive(Debug, Clone)]
/// Represents the configuration of an [`Node`] instance.
///
Expand Down
55 changes: 53 additions & 2 deletions src/io/utils.rs
Original file line numberDiff line numberDiff line change
Expand Up@@ -22,9 +22,11 @@ use bitcoin::Network;
use lightning::io::Cursor;
use lightning::ln::msgs::DecodeError;
use lightning::routing::gossip::NetworkGraph;
use lightning::routing::scoring::{ProbabilisticScorer, ProbabilisticScoringDecayParameters};
use lightning::routing::scoring::{
ChannelLiquidities, ProbabilisticScorer, ProbabilisticScoringDecayParameters,
};
use lightning::util::persist::{
KVStoreSync, KVSTORE_NAMESPACE_KEY_ALPHABET, KVSTORE_NAMESPACE_KEY_MAX_LEN,
KVStore, KVStoreSync, KVSTORE_NAMESPACE_KEY_ALPHABET, KVSTORE_NAMESPACE_KEY_MAX_LEN,
NETWORK_GRAPH_PERSISTENCE_KEY, NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE,
NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, OUTPUT_SWEEPER_PERSISTENCE_KEY,
OUTPUT_SWEEPER_PERSISTENCE_PRIMARY_NAMESPACE, OUTPUT_SWEEPER_PERSISTENCE_SECONDARY_NAMESPACE,
Expand All@@ -48,6 +50,8 @@ use crate::types::{Broadcaster, DynStore, KeysManager, Sweeper};
use crate::wallet::ser::{ChangeSetDeserWrapper, ChangeSetSerWrapper};
use crate::{Error, EventQueue, NodeMetrics, PaymentDetails};

pub const EXTERNAL_PATHFINDING_SCORES_CACHE_KEY: &str = "external_pathfinding_scores_cache";

/// Generates a random [BIP 39] mnemonic.
///
/// The result may be used to initialize the [`Node`] entropy, i.e., can be given to
Expand DownExpand Up@@ -164,6 +168,53 @@ where
})
}

/// Read previously persisted external pathfinding scores from the cache.
pub(crate) fn read_external_pathfinding_scores_from_cache<L: Deref>(
kv_store: Arc<DynStore>, logger: L,
) -> Result<ChannelLiquidities, std::io::Error>
where
L::Target: LdkLogger,
{
let mut reader = Cursor::new(KVStoreSync::read(
&*kv_store,
SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
EXTERNAL_PATHFINDING_SCORES_CACHE_KEY,
)?);
ChannelLiquidities::read(&mut reader).map_err(|e| {
log_error!(logger, "Failed to deserialize scorer: {}", e);
std::io::Error::new(std::io::ErrorKind::InvalidData, "Failed to deserialize Scorer")
})
}

/// Persist external pathfinding scores to the cache.
pub(crate) async fn write_external_pathfinding_scores_to_cache<L: Deref>(
kv_store: Arc<DynStore>, data: &ChannelLiquidities, logger: L,
) -> Result<(), Error>
where
L::Target: LdkLogger,
{
KVStore::write(
&*kv_store,
SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
EXTERNAL_PATHFINDING_SCORES_CACHE_KEY,
data.encode(),
)
.await
.map_err(|e| {
log_error!(
logger,
"Writing data to key {}/{}/{} failed due to: {}",
NODE_METRICS_PRIMARY_NAMESPACE,
NODE_METRICS_SECONDARY_NAMESPACE,
EXTERNAL_PATHFINDING_SCORES_CACHE_KEY,
e
);
Error::PersistenceFailed
})
}

/// Read previously persisted events from the store.
pub(crate) fn read_event_queue<L: Deref + Clone>(
kv_store: Arc<DynStore>, logger: L,
Expand Down
25 changes: 24 additions & 1 deletion src/lib.rs
Comment thread
joostjager marked this conversation as resolved.
Original file line numberDiff line numberDiff line change
Expand Up@@ -95,6 +95,7 @@ mod message_handler;
pub mod payment;
mod peer_store;
mod runtime;
mod scoring;
mod tx_broadcaster;
mod types;
mod wallet;
Expand All@@ -104,6 +105,7 @@ use std::net::ToSocketAddrs;
use std::sync::{Arc, Mutex, RwLock};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use crate::scoring::setup_background_pathfinding_scores_sync;
pub use balance::{BalanceDetails, LightningBalance, PendingSweepBalance};
use bitcoin::secp256k1::PublicKey;
#[cfg(feature = "uniffi")]
Expand DownExpand Up@@ -154,6 +156,7 @@ use types::{
pub use types::{
ChannelDetails, CustomTlvRecord, DynStore, PeerDetails, SyncAndAsyncKVStore, UserChannelId,
};

pub use {
bip39, bitcoin, lightning, lightning_invoice, lightning_liquidity, lightning_types, tokio,
vss_client,
Expand DownExpand Up@@ -183,6 +186,7 @@ pub struct Node {
keys_manager: Arc<KeysManager>,
network_graph: Arc<Graph>,
gossip_source: Arc<GossipSource>,
pathfinding_scores_sync_url: Option<String>,
liquidity_source: Option<Arc<LiquiditySource<Arc<Logger>>>>,
kv_store: Arc<DynStore>,
logger: Arc<Logger>,
Expand DownExpand Up@@ -259,7 +263,6 @@ impl Node {
return;
}
_ = interval.tick() => {
let gossip_sync_logger = Arc::clone(&gossip_sync_logger);
let now = Instant::now();
match gossip_source.update_rgs_snapshot().await {
Ok(updated_timestamp) => {
Expand DownExpand Up@@ -291,6 +294,18 @@ impl Node {
});
}

if let Some(pathfinding_scores_sync_url) = self.pathfinding_scores_sync_url.as_ref() {
setup_background_pathfinding_scores_sync(
pathfinding_scores_sync_url.clone(),
Arc::clone(&self.scorer),
Arc::clone(&self.node_metrics),
Arc::clone(&self.kv_store),
Arc::clone(&self.logger),
Arc::clone(&self.runtime),
self.stop_sender.subscribe(),
);
}

if let Some(listening_addresses) = &self.config.listening_addresses {
// Setup networking
let peer_manager_connection_handler = Arc::clone(&self.peer_manager);
Expand DownExpand Up@@ -692,6 +707,8 @@ impl Node {
locked_node_metrics.latest_fee_rate_cache_update_timestamp;
let latest_rgs_snapshot_timestamp =
locked_node_metrics.latest_rgs_snapshot_timestamp.map(|val| val as u64);
let latest_pathfinding_scores_sync_timestamp =
locked_node_metrics.latest_pathfinding_scores_sync_timestamp;
let latest_node_announcement_broadcast_timestamp =
locked_node_metrics.latest_node_announcement_broadcast_timestamp;
let latest_channel_monitor_archival_height =
Expand All@@ -704,6 +721,7 @@ impl Node {
latest_onchain_wallet_sync_timestamp,
latest_fee_rate_cache_update_timestamp,
latest_rgs_snapshot_timestamp,
latest_pathfinding_scores_sync_timestamp,
latest_node_announcement_broadcast_timestamp,
latest_channel_monitor_archival_height,
}
Expand DownExpand Up@@ -1531,6 +1549,8 @@ pub struct NodeStatus {
///
/// Will be `None` if RGS isn't configured or the snapshot hasn't been updated yet.
pub latest_rgs_snapshot_timestamp: Option<u64>,
/// The timestamp, in seconds since start of the UNIX epoch, when we last successfully merged external scores.
pub latest_pathfinding_scores_sync_timestamp: Option<u64>,
/// The timestamp, in seconds since start of the UNIX epoch, when we last broadcasted a node
/// announcement.
///
Expand All@@ -1549,6 +1569,7 @@ pub(crate) struct NodeMetrics {
latest_onchain_wallet_sync_timestamp: Option<u64>,
latest_fee_rate_cache_update_timestamp: Option<u64>,
latest_rgs_snapshot_timestamp: Option<u32>,
latest_pathfinding_scores_sync_timestamp: Option<u64>,
latest_node_announcement_broadcast_timestamp: Option<u64>,
latest_channel_monitor_archival_height: Option<u32>,
}
Expand All@@ -1560,6 +1581,7 @@ impl Default for NodeMetrics {
latest_onchain_wallet_sync_timestamp: None,
latest_fee_rate_cache_update_timestamp: None,
latest_rgs_snapshot_timestamp: None,
latest_pathfinding_scores_sync_timestamp: None,
latest_node_announcement_broadcast_timestamp: None,
latest_channel_monitor_archival_height: None,
}
Expand All@@ -1568,6 +1590,7 @@ impl Default for NodeMetrics {

impl_writeable_tlv_based!(NodeMetrics, {
(0, latest_lightning_wallet_sync_timestamp, option),
(1, latest_pathfinding_scores_sync_timestamp, option),
(2, latest_onchain_wallet_sync_timestamp, option),
(4, latest_fee_rate_cache_update_timestamp, option),
(6, latest_rgs_snapshot_timestamp, option),
Expand Down
Loading
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { // Add copy buttons to all
 blocks
(function() {
function addCopyButtons() {
document.querySelectorAll('pre code').forEach(function(codeBlock) {
if (codeBlock.parentElement.hasAttribute('data-copy-added')) return;
codeBlock.parentElement.setAttribute('data-copy-added', 'true');
var btn = document.createElement('button');
btn.textContent = 'Copy';
btn.style.cssText = 'position:absolute;top:4px;right:4px;padding:2px 8px;font-size:11px;background:#4ecdc4;border:none;border-radius:4px;color:#1a1a2e;cursor:pointer;opacity:0.7;transition:opacity 0.2s;';
btn.onmouseover = function() { this.style.opacity = '1'; };
btn.onmouseout = function() { this.style.opacity = '0.7'; };
btn.onclick = function() {
navigator.clipboard.writeText(codeBlock.textContent).then(function() {
btn.textContent = 'Copied!';
setTimeout(function() { btn.textContent = 'Copy'; }, 1500);
});
};
codeBlock.parentElement.style.position = 'relative';
codeBlock.parentElement.appendChild(btn);
});
}
addCopyButtons();
// Re-run on dynamic content
var observer = new MutationObserver(addCopyButtons);
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Add Copy Buttons to Code Blocks]', __e); }
})();
(function(){
try {
var __m = "github.com";
var __re = new RegExp('^' + "github\\.com" + '
Periodical external pathfinding scores merge by joostjager · Pull Request #449 · lightningdevkit/ldk-node · GitHub
Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions bindings/ldk_node.udl
Original file line numberDiff line numberDiff line change
Expand Up@@ -81,6 +81,7 @@ interface Builder {
void set_chain_source_bitcoind_rest(string rest_host, u16 rest_port, string rpc_host, u16 rpc_port, string rpc_user, string rpc_password);
void set_gossip_source_p2p();
void set_gossip_source_rgs(string rgs_server_url);
void set_pathfinding_scores_source(string url);
void set_liquidity_source_lsps1(PublicKey node_id, SocketAddress address, string? token);
void set_liquidity_source_lsps2(PublicKey node_id, SocketAddress address, string? token);
void set_storage_dir_path(string storage_dir_path);
Expand DownExpand Up@@ -330,6 +331,7 @@ dictionary NodeStatus {
u64? latest_onchain_wallet_sync_timestamp;
u64? latest_fee_rate_cache_update_timestamp;
u64? latest_rgs_snapshot_timestamp;
u64? latest_pathfinding_scores_sync_timestamp;
u64? latest_node_announcement_broadcast_timestamp;
u32? latest_channel_monitor_archival_height;
};
Expand Down
63 changes: 54 additions & 9 deletions src/builder.rs
Original file line numberDiff line numberDiff line change
Expand Up@@ -24,10 +24,12 @@ use lightning::io::Cursor;
use lightning::ln::channelmanager::{self, ChainParameters, ChannelManagerReadArgs};
use lightning::ln::msgs::{RoutingMessageHandler, SocketAddress};
use lightning::ln::peer_handler::{IgnoringMessageHandler, MessageHandler};
use lightning::log_trace;
use lightning::routing::gossip::NodeAlias;
use lightning::routing::router::DefaultRouter;
use lightning::routing::scoring::{
ProbabilisticScorer, ProbabilisticScoringDecayParameters, ProbabilisticScoringFeeParameters,
CombinedScorer, ProbabilisticScorer, ProbabilisticScoringDecayParameters,
ProbabilisticScoringFeeParameters,
};
use lightning::sign::{EntropySource, NodeSigner};
use lightning::util::persist::{
Expand All@@ -50,7 +52,9 @@ use crate::event::EventQueue;
use crate::fee_estimator::OnchainFeeEstimator;
use crate::gossip::GossipSource;
use crate::io::sqlite_store::SqliteStore;
use crate::io::utils::{read_node_metrics, write_node_metrics};
use crate::io::utils::{
read_external_pathfinding_scores_from_cache, read_node_metrics, write_node_metrics,
};
use crate::io::vss_store::VssStore;
use crate::io::{
self, PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
Expand DownExpand Up@@ -110,6 +114,11 @@ enum GossipSourceConfig {
RapidGossipSync(String),
}

#[derive(Debug, Clone)]
struct PathfindingScoresSyncConfig {
url: String,
}

#[derive(Debug, Clone, Default)]
struct LiquiditySourceConfig {
// Act as an LSPS1 client connecting to the given service.
Expand DownExpand Up@@ -243,6 +252,7 @@ pub struct NodeBuilder {
log_writer_config: Option<LogWriterConfig>,
async_payments_role: Option<AsyncPaymentsRole>,
runtime_handle: Option<tokio::runtime::Handle>,
pathfinding_scores_sync_config: Option<PathfindingScoresSyncConfig>,
}

impl NodeBuilder {
Expand All@@ -260,6 +270,7 @@ impl NodeBuilder {
let liquidity_source_config = None;
let log_writer_config = None;
let runtime_handle = None;
let pathfinding_scores_sync_config = None;
Self {
config,
entropy_source_config,
Expand All@@ -269,6 +280,7 @@ impl NodeBuilder {
log_writer_config,
runtime_handle,
async_payments_role: None,
pathfinding_scores_sync_config,
}
}

Expand DownExpand Up@@ -411,6 +423,14 @@ impl NodeBuilder {
self
}

/// Configures the [`Node`] instance to source its external scores from the given URL.
Comment thread
joostjager marked this conversation as resolved.
///
/// The external scores are merged into the local scoring system to improve routing.
pub fn set_pathfinding_scores_source(&mut self, url: String) -> &mut Self {
self.pathfinding_scores_sync_config = Some(PathfindingScoresSyncConfig { url });
self
}

/// Configures the [`Node`] instance to source inbound liquidity from the given
/// [bLIP-51 / LSPS1] service.
///
Expand DownExpand Up@@ -718,6 +738,7 @@ impl NodeBuilder {
self.chain_data_source_config.as_ref(),
self.gossip_source_config.as_ref(),
self.liquidity_source_config.as_ref(),
self.pathfinding_scores_sync_config.as_ref(),
self.async_payments_role,
seed_bytes,
runtime,
Expand DownExpand Up@@ -751,6 +772,7 @@ impl NodeBuilder {
self.chain_data_source_config.as_ref(),
self.gossip_source_config.as_ref(),
self.liquidity_source_config.as_ref(),
self.pathfinding_scores_sync_config.as_ref(),
self.async_payments_role,
seed_bytes,
runtime,
Expand DownExpand Up@@ -910,6 +932,13 @@ impl ArcedNodeBuilder {
self.inner.write().unwrap().set_gossip_source_rgs(rgs_server_url);
}

/// Configures the [`Node`] instance to source its external scores from the given URL.
///
/// The external scores are merged into the local scoring system to improve routing.
pub fn set_pathfinding_scores_source(&self, url: String) {
self.inner.write().unwrap().set_pathfinding_scores_source(url);
}

/// Configures the [`Node`] instance to source inbound liquidity from the given
/// [bLIP-51 / LSPS1] service.
///
Expand DownExpand Up@@ -1110,6 +1139,7 @@ fn build_with_store_internal(
config: Arc<Config>, chain_data_source_config: Option<&ChainDataSourceConfig>,
gossip_source_config: Option<&GossipSourceConfig>,
liquidity_source_config: Option<&LiquiditySourceConfig>,
pathfinding_scores_sync_config: Option<&PathfindingScoresSyncConfig>,
async_payments_role: Option<AsyncPaymentsRole>, seed_bytes: [u8; 64], runtime: Arc<Runtime>,
logger: Arc<Logger>, kv_store: Arc<DynStore>,
) -> Result<Node, BuildError> {
Expand DownExpand Up@@ -1365,26 +1395,38 @@ fn build_with_store_internal(
},
};

let scorer = match io::utils::read_scorer(
let local_scorer = match io::utils::read_scorer(
Arc::clone(&kv_store),
Arc::clone(&network_graph),
Arc::clone(&logger),
) {
Ok(scorer) => Arc::new(Mutex::new(scorer)),
Ok(scorer) => scorer,
Err(e) => {
if e.kind() == std::io::ErrorKind::NotFound {
let params = ProbabilisticScoringDecayParameters::default();
Arc::new(Mutex::new(ProbabilisticScorer::new(
params,
Arc::clone(&network_graph),
Arc::clone(&logger),
)))
ProbabilisticScorer::new(params, Arc::clone(&network_graph), Arc::clone(&logger))
} else {
return Err(BuildError::ReadFailed);
}
},
};

let scorer = Arc::new(Mutex::new(CombinedScorer::new(local_scorer)));

// Restore external pathfinding scores from cache if possible.
match read_external_pathfinding_scores_from_cache(Arc::clone(&kv_store), Arc::clone(&logger)) {
Ok(external_scores) => {
scorer.lock().unwrap().merge(external_scores, cur_time);
log_trace!(logger, "External scores from cache merged successfully");
},
Err(e) => {
if e.kind() != std::io::ErrorKind::NotFound {
log_error!(logger, "Error while reading external scores from cache: {}", e);
return Err(BuildError::ReadFailed);
}
},
}

let scoring_fee_params = ProbabilisticScoringFeeParameters::default();
let router = Arc::new(DefaultRouter::new(
Arc::clone(&network_graph),
Expand DownExpand Up@@ -1716,6 +1758,8 @@ fn build_with_store_internal(
let (background_processor_stop_sender, _) = tokio::sync::watch::channel(());
let is_running = Arc::new(RwLock::new(false));

let pathfinding_scores_sync_url = pathfinding_scores_sync_config.map(|c| c.url.clone());

Ok(Node {
runtime,
stop_sender,
Expand All@@ -1734,6 +1778,7 @@ fn build_with_store_internal(
keys_manager,
network_graph,
gossip_source,
pathfinding_scores_sync_url,
liquidity_source,
kv_store,
logger,
Expand Down
6 changes: 6 additions & 0 deletions src/config.rs
Original file line numberDiff line numberDiff line change
Expand Up@@ -63,6 +63,9 @@ pub(crate) const PEER_RECONNECTION_INTERVAL: Duration = Duration::from_secs(60);
// The time in-between RGS sync attempts.
pub(crate) const RGS_SYNC_INTERVAL: Duration = Duration::from_secs(60 * 60);

// The time in-between external scores sync attempts.
pub(crate) const EXTERNAL_PATHFINDING_SCORES_SYNC_INTERVAL: Duration = Duration::from_secs(60 * 60);

// The time in-between node announcement broadcast attempts.
pub(crate) const NODE_ANN_BCAST_INTERVAL: Duration = Duration::from_secs(60 * 60);

Expand DownExpand Up@@ -93,6 +96,9 @@ pub(crate) const RGS_SYNC_TIMEOUT_SECS: u64 = 5;
/// The length in bytes of our wallets' keys seed.
pub const WALLET_KEYS_SEED_LEN: usize = 64;

// The timeout after which we abort a external scores sync operation.
pub(crate) const EXTERNAL_PATHFINDING_SCORES_SYNC_TIMEOUT_SECS: u64 = 5;

#[derive(Debug, Clone)]
/// Represents the configuration of an [`Node`] instance.
///
Expand Down
55 changes: 53 additions & 2 deletions src/io/utils.rs
Original file line numberDiff line numberDiff line change
Expand Up@@ -22,9 +22,11 @@ use bitcoin::Network;
use lightning::io::Cursor;
use lightning::ln::msgs::DecodeError;
use lightning::routing::gossip::NetworkGraph;
use lightning::routing::scoring::{ProbabilisticScorer, ProbabilisticScoringDecayParameters};
use lightning::routing::scoring::{
ChannelLiquidities, ProbabilisticScorer, ProbabilisticScoringDecayParameters,
};
use lightning::util::persist::{
KVStoreSync, KVSTORE_NAMESPACE_KEY_ALPHABET, KVSTORE_NAMESPACE_KEY_MAX_LEN,
KVStore, KVStoreSync, KVSTORE_NAMESPACE_KEY_ALPHABET, KVSTORE_NAMESPACE_KEY_MAX_LEN,
NETWORK_GRAPH_PERSISTENCE_KEY, NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE,
NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, OUTPUT_SWEEPER_PERSISTENCE_KEY,
OUTPUT_SWEEPER_PERSISTENCE_PRIMARY_NAMESPACE, OUTPUT_SWEEPER_PERSISTENCE_SECONDARY_NAMESPACE,
Expand All@@ -48,6 +50,8 @@ use crate::types::{Broadcaster, DynStore, KeysManager, Sweeper};
use crate::wallet::ser::{ChangeSetDeserWrapper, ChangeSetSerWrapper};
use crate::{Error, EventQueue, NodeMetrics, PaymentDetails};

pub const EXTERNAL_PATHFINDING_SCORES_CACHE_KEY: &str = "external_pathfinding_scores_cache";

/// Generates a random [BIP 39] mnemonic.
///
/// The result may be used to initialize the [`Node`] entropy, i.e., can be given to
Expand DownExpand Up@@ -164,6 +168,53 @@ where
})
}

/// Read previously persisted external pathfinding scores from the cache.
pub(crate) fn read_external_pathfinding_scores_from_cache<L: Deref>(
kv_store: Arc<DynStore>, logger: L,
) -> Result<ChannelLiquidities, std::io::Error>
where
L::Target: LdkLogger,
{
let mut reader = Cursor::new(KVStoreSync::read(
&*kv_store,
SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
EXTERNAL_PATHFINDING_SCORES_CACHE_KEY,
)?);
ChannelLiquidities::read(&mut reader).map_err(|e| {
log_error!(logger, "Failed to deserialize scorer: {}", e);
std::io::Error::new(std::io::ErrorKind::InvalidData, "Failed to deserialize Scorer")
})
}

/// Persist external pathfinding scores to the cache.
pub(crate) async fn write_external_pathfinding_scores_to_cache<L: Deref>(
kv_store: Arc<DynStore>, data: &ChannelLiquidities, logger: L,
) -> Result<(), Error>
where
L::Target: LdkLogger,
{
KVStore::write(
&*kv_store,
SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
EXTERNAL_PATHFINDING_SCORES_CACHE_KEY,
data.encode(),
)
.await
.map_err(|e| {
log_error!(
logger,
"Writing data to key {}/{}/{} failed due to: {}",
NODE_METRICS_PRIMARY_NAMESPACE,
NODE_METRICS_SECONDARY_NAMESPACE,
EXTERNAL_PATHFINDING_SCORES_CACHE_KEY,
e
);
Error::PersistenceFailed
})
}

/// Read previously persisted events from the store.
pub(crate) fn read_event_queue<L: Deref + Clone>(
kv_store: Arc<DynStore>, logger: L,
Expand Down
25 changes: 24 additions & 1 deletion src/lib.rs
Comment thread
joostjager marked this conversation as resolved.
Original file line numberDiff line numberDiff line change
Expand Up@@ -95,6 +95,7 @@ mod message_handler;
pub mod payment;
mod peer_store;
mod runtime;
mod scoring;
mod tx_broadcaster;
mod types;
mod wallet;
Expand All@@ -104,6 +105,7 @@ use std::net::ToSocketAddrs;
use std::sync::{Arc, Mutex, RwLock};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use crate::scoring::setup_background_pathfinding_scores_sync;
pub use balance::{BalanceDetails, LightningBalance, PendingSweepBalance};
use bitcoin::secp256k1::PublicKey;
#[cfg(feature = "uniffi")]
Expand DownExpand Up@@ -154,6 +156,7 @@ use types::{
pub use types::{
ChannelDetails, CustomTlvRecord, DynStore, PeerDetails, SyncAndAsyncKVStore, UserChannelId,
};

pub use {
bip39, bitcoin, lightning, lightning_invoice, lightning_liquidity, lightning_types, tokio,
vss_client,
Expand DownExpand Up@@ -183,6 +186,7 @@ pub struct Node {
keys_manager: Arc<KeysManager>,
network_graph: Arc<Graph>,
gossip_source: Arc<GossipSource>,
pathfinding_scores_sync_url: Option<String>,
liquidity_source: Option<Arc<LiquiditySource<Arc<Logger>>>>,
kv_store: Arc<DynStore>,
logger: Arc<Logger>,
Expand DownExpand Up@@ -259,7 +263,6 @@ impl Node {
return;
}
_ = interval.tick() => {
let gossip_sync_logger = Arc::clone(&gossip_sync_logger);
let now = Instant::now();
match gossip_source.update_rgs_snapshot().await {
Ok(updated_timestamp) => {
Expand DownExpand Up@@ -291,6 +294,18 @@ impl Node {
});
}

if let Some(pathfinding_scores_sync_url) = self.pathfinding_scores_sync_url.as_ref() {
setup_background_pathfinding_scores_sync(
pathfinding_scores_sync_url.clone(),
Arc::clone(&self.scorer),
Arc::clone(&self.node_metrics),
Arc::clone(&self.kv_store),
Arc::clone(&self.logger),
Arc::clone(&self.runtime),
self.stop_sender.subscribe(),
);
}

if let Some(listening_addresses) = &self.config.listening_addresses {
// Setup networking
let peer_manager_connection_handler = Arc::clone(&self.peer_manager);
Expand DownExpand Up@@ -692,6 +707,8 @@ impl Node {
locked_node_metrics.latest_fee_rate_cache_update_timestamp;
let latest_rgs_snapshot_timestamp =
locked_node_metrics.latest_rgs_snapshot_timestamp.map(|val| val as u64);
let latest_pathfinding_scores_sync_timestamp =
locked_node_metrics.latest_pathfinding_scores_sync_timestamp;
let latest_node_announcement_broadcast_timestamp =
locked_node_metrics.latest_node_announcement_broadcast_timestamp;
let latest_channel_monitor_archival_height =
Expand All@@ -704,6 +721,7 @@ impl Node {
latest_onchain_wallet_sync_timestamp,
latest_fee_rate_cache_update_timestamp,
latest_rgs_snapshot_timestamp,
latest_pathfinding_scores_sync_timestamp,
latest_node_announcement_broadcast_timestamp,
latest_channel_monitor_archival_height,
}
Expand DownExpand Up@@ -1531,6 +1549,8 @@ pub struct NodeStatus {
///
/// Will be `None` if RGS isn't configured or the snapshot hasn't been updated yet.
pub latest_rgs_snapshot_timestamp: Option<u64>,
/// The timestamp, in seconds since start of the UNIX epoch, when we last successfully merged external scores.
pub latest_pathfinding_scores_sync_timestamp: Option<u64>,
/// The timestamp, in seconds since start of the UNIX epoch, when we last broadcasted a node
/// announcement.
///
Expand All@@ -1549,6 +1569,7 @@ pub(crate) struct NodeMetrics {
latest_onchain_wallet_sync_timestamp: Option<u64>,
latest_fee_rate_cache_update_timestamp: Option<u64>,
latest_rgs_snapshot_timestamp: Option<u32>,
latest_pathfinding_scores_sync_timestamp: Option<u64>,
latest_node_announcement_broadcast_timestamp: Option<u64>,
latest_channel_monitor_archival_height: Option<u32>,
}
Expand All@@ -1560,6 +1581,7 @@ impl Default for NodeMetrics {
latest_onchain_wallet_sync_timestamp: None,
latest_fee_rate_cache_update_timestamp: None,
latest_rgs_snapshot_timestamp: None,
latest_pathfinding_scores_sync_timestamp: None,
latest_node_announcement_broadcast_timestamp: None,
latest_channel_monitor_archival_height: None,
}
Expand All@@ -1568,6 +1590,7 @@ impl Default for NodeMetrics {

impl_writeable_tlv_based!(NodeMetrics, {
(0, latest_lightning_wallet_sync_timestamp, option),
(1, latest_pathfinding_scores_sync_timestamp, option),
(2, latest_onchain_wallet_sync_timestamp, option),
(4, latest_fee_rate_cache_update_timestamp, option),
(6, latest_rgs_snapshot_timestamp, option),
Expand Down
Loading
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { // Force GitHub README to respect dark mode (function() { var style = document.createElement('style'); style.textContent = ' .markdown-body { color-scheme: dark light; } .markdown-body pre { background: #161b22 !important; } .markdown-body code { background: rgba(110, 118, 129, 0.4) !important; } .markdown-body table th, .markdown-body table td { border-color: #30363d !important; } .markdown-body img { background: #0d1117; } .markdown-body blockquote { border-left-color: #8b949e; } .markdown-body hr { border-color: #30363d; } '; document.head.appendChild(style); })(); } } catch(__e) { console.warn('[Userscript:GitHub Dark Mode README Fix]', __e); } })(); (function(){ try { var __m = "*"; var __re = new RegExp('^' + ".*" + ' Periodical external pathfinding scores merge by joostjager · Pull Request #449 · lightningdevkit/ldk-node · GitHub
Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions bindings/ldk_node.udl
Original file line numberDiff line numberDiff line change
Expand Up@@ -81,6 +81,7 @@ interface Builder {
void set_chain_source_bitcoind_rest(string rest_host, u16 rest_port, string rpc_host, u16 rpc_port, string rpc_user, string rpc_password);
void set_gossip_source_p2p();
void set_gossip_source_rgs(string rgs_server_url);
void set_pathfinding_scores_source(string url);
void set_liquidity_source_lsps1(PublicKey node_id, SocketAddress address, string? token);
void set_liquidity_source_lsps2(PublicKey node_id, SocketAddress address, string? token);
void set_storage_dir_path(string storage_dir_path);
Expand DownExpand Up@@ -330,6 +331,7 @@ dictionary NodeStatus {
u64? latest_onchain_wallet_sync_timestamp;
u64? latest_fee_rate_cache_update_timestamp;
u64? latest_rgs_snapshot_timestamp;
u64? latest_pathfinding_scores_sync_timestamp;
u64? latest_node_announcement_broadcast_timestamp;
u32? latest_channel_monitor_archival_height;
};
Expand Down
63 changes: 54 additions & 9 deletions src/builder.rs
Original file line numberDiff line numberDiff line change
Expand Up@@ -24,10 +24,12 @@ use lightning::io::Cursor;
use lightning::ln::channelmanager::{self, ChainParameters, ChannelManagerReadArgs};
use lightning::ln::msgs::{RoutingMessageHandler, SocketAddress};
use lightning::ln::peer_handler::{IgnoringMessageHandler, MessageHandler};
use lightning::log_trace;
use lightning::routing::gossip::NodeAlias;
use lightning::routing::router::DefaultRouter;
use lightning::routing::scoring::{
ProbabilisticScorer, ProbabilisticScoringDecayParameters, ProbabilisticScoringFeeParameters,
CombinedScorer, ProbabilisticScorer, ProbabilisticScoringDecayParameters,
ProbabilisticScoringFeeParameters,
};
use lightning::sign::{EntropySource, NodeSigner};
use lightning::util::persist::{
Expand All@@ -50,7 +52,9 @@ use crate::event::EventQueue;
use crate::fee_estimator::OnchainFeeEstimator;
use crate::gossip::GossipSource;
use crate::io::sqlite_store::SqliteStore;
use crate::io::utils::{read_node_metrics, write_node_metrics};
use crate::io::utils::{
read_external_pathfinding_scores_from_cache, read_node_metrics, write_node_metrics,
};
use crate::io::vss_store::VssStore;
use crate::io::{
self, PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
Expand DownExpand Up@@ -110,6 +114,11 @@ enum GossipSourceConfig {
RapidGossipSync(String),
}

#[derive(Debug, Clone)]
struct PathfindingScoresSyncConfig {
url: String,
}

#[derive(Debug, Clone, Default)]
struct LiquiditySourceConfig {
// Act as an LSPS1 client connecting to the given service.
Expand DownExpand Up@@ -243,6 +252,7 @@ pub struct NodeBuilder {
log_writer_config: Option<LogWriterConfig>,
async_payments_role: Option<AsyncPaymentsRole>,
runtime_handle: Option<tokio::runtime::Handle>,
pathfinding_scores_sync_config: Option<PathfindingScoresSyncConfig>,
}

impl NodeBuilder {
Expand All@@ -260,6 +270,7 @@ impl NodeBuilder {
let liquidity_source_config = None;
let log_writer_config = None;
let runtime_handle = None;
let pathfinding_scores_sync_config = None;
Self {
config,
entropy_source_config,
Expand All@@ -269,6 +280,7 @@ impl NodeBuilder {
log_writer_config,
runtime_handle,
async_payments_role: None,
pathfinding_scores_sync_config,
}
}

Expand DownExpand Up@@ -411,6 +423,14 @@ impl NodeBuilder {
self
}

/// Configures the [`Node`] instance to source its external scores from the given URL.
Comment thread
joostjager marked this conversation as resolved.
///
/// The external scores are merged into the local scoring system to improve routing.
pub fn set_pathfinding_scores_source(&mut self, url: String) -> &mut Self {
self.pathfinding_scores_sync_config = Some(PathfindingScoresSyncConfig { url });
self
}

/// Configures the [`Node`] instance to source inbound liquidity from the given
/// [bLIP-51 / LSPS1] service.
///
Expand DownExpand Up@@ -718,6 +738,7 @@ impl NodeBuilder {
self.chain_data_source_config.as_ref(),
self.gossip_source_config.as_ref(),
self.liquidity_source_config.as_ref(),
self.pathfinding_scores_sync_config.as_ref(),
self.async_payments_role,
seed_bytes,
runtime,
Expand DownExpand Up@@ -751,6 +772,7 @@ impl NodeBuilder {
self.chain_data_source_config.as_ref(),
self.gossip_source_config.as_ref(),
self.liquidity_source_config.as_ref(),
self.pathfinding_scores_sync_config.as_ref(),
self.async_payments_role,
seed_bytes,
runtime,
Expand DownExpand Up@@ -910,6 +932,13 @@ impl ArcedNodeBuilder {
self.inner.write().unwrap().set_gossip_source_rgs(rgs_server_url);
}

/// Configures the [`Node`] instance to source its external scores from the given URL.
///
/// The external scores are merged into the local scoring system to improve routing.
pub fn set_pathfinding_scores_source(&self, url: String) {
self.inner.write().unwrap().set_pathfinding_scores_source(url);
}

/// Configures the [`Node`] instance to source inbound liquidity from the given
/// [bLIP-51 / LSPS1] service.
///
Expand DownExpand Up@@ -1110,6 +1139,7 @@ fn build_with_store_internal(
config: Arc<Config>, chain_data_source_config: Option<&ChainDataSourceConfig>,
gossip_source_config: Option<&GossipSourceConfig>,
liquidity_source_config: Option<&LiquiditySourceConfig>,
pathfinding_scores_sync_config: Option<&PathfindingScoresSyncConfig>,
async_payments_role: Option<AsyncPaymentsRole>, seed_bytes: [u8; 64], runtime: Arc<Runtime>,
logger: Arc<Logger>, kv_store: Arc<DynStore>,
) -> Result<Node, BuildError> {
Expand DownExpand Up@@ -1365,26 +1395,38 @@ fn build_with_store_internal(
},
};

let scorer = match io::utils::read_scorer(
let local_scorer = match io::utils::read_scorer(
Arc::clone(&kv_store),
Arc::clone(&network_graph),
Arc::clone(&logger),
) {
Ok(scorer) => Arc::new(Mutex::new(scorer)),
Ok(scorer) => scorer,
Err(e) => {
if e.kind() == std::io::ErrorKind::NotFound {
let params = ProbabilisticScoringDecayParameters::default();
Arc::new(Mutex::new(ProbabilisticScorer::new(
params,
Arc::clone(&network_graph),
Arc::clone(&logger),
)))
ProbabilisticScorer::new(params, Arc::clone(&network_graph), Arc::clone(&logger))
} else {
return Err(BuildError::ReadFailed);
}
},
};

let scorer = Arc::new(Mutex::new(CombinedScorer::new(local_scorer)));

// Restore external pathfinding scores from cache if possible.
match read_external_pathfinding_scores_from_cache(Arc::clone(&kv_store), Arc::clone(&logger)) {
Ok(external_scores) => {
scorer.lock().unwrap().merge(external_scores, cur_time);
log_trace!(logger, "External scores from cache merged successfully");
},
Err(e) => {
if e.kind() != std::io::ErrorKind::NotFound {
log_error!(logger, "Error while reading external scores from cache: {}", e);
return Err(BuildError::ReadFailed);
}
},
}

let scoring_fee_params = ProbabilisticScoringFeeParameters::default();
let router = Arc::new(DefaultRouter::new(
Arc::clone(&network_graph),
Expand DownExpand Up@@ -1716,6 +1758,8 @@ fn build_with_store_internal(
let (background_processor_stop_sender, _) = tokio::sync::watch::channel(());
let is_running = Arc::new(RwLock::new(false));

let pathfinding_scores_sync_url = pathfinding_scores_sync_config.map(|c| c.url.clone());

Ok(Node {
runtime,
stop_sender,
Expand All@@ -1734,6 +1778,7 @@ fn build_with_store_internal(
keys_manager,
network_graph,
gossip_source,
pathfinding_scores_sync_url,
liquidity_source,
kv_store,
logger,
Expand Down
6 changes: 6 additions & 0 deletions src/config.rs
Original file line numberDiff line numberDiff line change
Expand Up@@ -63,6 +63,9 @@ pub(crate) const PEER_RECONNECTION_INTERVAL: Duration = Duration::from_secs(60);
// The time in-between RGS sync attempts.
pub(crate) const RGS_SYNC_INTERVAL: Duration = Duration::from_secs(60 * 60);

// The time in-between external scores sync attempts.
pub(crate) const EXTERNAL_PATHFINDING_SCORES_SYNC_INTERVAL: Duration = Duration::from_secs(60 * 60);

// The time in-between node announcement broadcast attempts.
pub(crate) const NODE_ANN_BCAST_INTERVAL: Duration = Duration::from_secs(60 * 60);

Expand DownExpand Up@@ -93,6 +96,9 @@ pub(crate) const RGS_SYNC_TIMEOUT_SECS: u64 = 5;
/// The length in bytes of our wallets' keys seed.
pub const WALLET_KEYS_SEED_LEN: usize = 64;

// The timeout after which we abort a external scores sync operation.
pub(crate) const EXTERNAL_PATHFINDING_SCORES_SYNC_TIMEOUT_SECS: u64 = 5;

#[derive(Debug, Clone)]
/// Represents the configuration of an [`Node`] instance.
///
Expand Down
55 changes: 53 additions & 2 deletions src/io/utils.rs
Original file line numberDiff line numberDiff line change
Expand Up@@ -22,9 +22,11 @@ use bitcoin::Network;
use lightning::io::Cursor;
use lightning::ln::msgs::DecodeError;
use lightning::routing::gossip::NetworkGraph;
use lightning::routing::scoring::{ProbabilisticScorer, ProbabilisticScoringDecayParameters};
use lightning::routing::scoring::{
ChannelLiquidities, ProbabilisticScorer, ProbabilisticScoringDecayParameters,
};
use lightning::util::persist::{
KVStoreSync, KVSTORE_NAMESPACE_KEY_ALPHABET, KVSTORE_NAMESPACE_KEY_MAX_LEN,
KVStore, KVStoreSync, KVSTORE_NAMESPACE_KEY_ALPHABET, KVSTORE_NAMESPACE_KEY_MAX_LEN,
NETWORK_GRAPH_PERSISTENCE_KEY, NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE,
NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, OUTPUT_SWEEPER_PERSISTENCE_KEY,
OUTPUT_SWEEPER_PERSISTENCE_PRIMARY_NAMESPACE, OUTPUT_SWEEPER_PERSISTENCE_SECONDARY_NAMESPACE,
Expand All@@ -48,6 +50,8 @@ use crate::types::{Broadcaster, DynStore, KeysManager, Sweeper};
use crate::wallet::ser::{ChangeSetDeserWrapper, ChangeSetSerWrapper};
use crate::{Error, EventQueue, NodeMetrics, PaymentDetails};

pub const EXTERNAL_PATHFINDING_SCORES_CACHE_KEY: &str = "external_pathfinding_scores_cache";

/// Generates a random [BIP 39] mnemonic.
///
/// The result may be used to initialize the [`Node`] entropy, i.e., can be given to
Expand DownExpand Up@@ -164,6 +168,53 @@ where
})
}

/// Read previously persisted external pathfinding scores from the cache.
pub(crate) fn read_external_pathfinding_scores_from_cache<L: Deref>(
kv_store: Arc<DynStore>, logger: L,
) -> Result<ChannelLiquidities, std::io::Error>
where
L::Target: LdkLogger,
{
let mut reader = Cursor::new(KVStoreSync::read(
&*kv_store,
SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
EXTERNAL_PATHFINDING_SCORES_CACHE_KEY,
)?);
ChannelLiquidities::read(&mut reader).map_err(|e| {
log_error!(logger, "Failed to deserialize scorer: {}", e);
std::io::Error::new(std::io::ErrorKind::InvalidData, "Failed to deserialize Scorer")
})
}

/// Persist external pathfinding scores to the cache.
pub(crate) async fn write_external_pathfinding_scores_to_cache<L: Deref>(
kv_store: Arc<DynStore>, data: &ChannelLiquidities, logger: L,
) -> Result<(), Error>
where
L::Target: LdkLogger,
{
KVStore::write(
&*kv_store,
SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
EXTERNAL_PATHFINDING_SCORES_CACHE_KEY,
data.encode(),
)
.await
.map_err(|e| {
log_error!(
logger,
"Writing data to key {}/{}/{} failed due to: {}",
NODE_METRICS_PRIMARY_NAMESPACE,
NODE_METRICS_SECONDARY_NAMESPACE,
EXTERNAL_PATHFINDING_SCORES_CACHE_KEY,
e
);
Error::PersistenceFailed
})
}

/// Read previously persisted events from the store.
pub(crate) fn read_event_queue<L: Deref + Clone>(
kv_store: Arc<DynStore>, logger: L,
Expand Down
25 changes: 24 additions & 1 deletion src/lib.rs
Comment thread
joostjager marked this conversation as resolved.
Original file line numberDiff line numberDiff line change
Expand Up@@ -95,6 +95,7 @@ mod message_handler;
pub mod payment;
mod peer_store;
mod runtime;
mod scoring;
mod tx_broadcaster;
mod types;
mod wallet;
Expand All@@ -104,6 +105,7 @@ use std::net::ToSocketAddrs;
use std::sync::{Arc, Mutex, RwLock};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use crate::scoring::setup_background_pathfinding_scores_sync;
pub use balance::{BalanceDetails, LightningBalance, PendingSweepBalance};
use bitcoin::secp256k1::PublicKey;
#[cfg(feature = "uniffi")]
Expand DownExpand Up@@ -154,6 +156,7 @@ use types::{
pub use types::{
ChannelDetails, CustomTlvRecord, DynStore, PeerDetails, SyncAndAsyncKVStore, UserChannelId,
};

pub use {
bip39, bitcoin, lightning, lightning_invoice, lightning_liquidity, lightning_types, tokio,
vss_client,
Expand DownExpand Up@@ -183,6 +186,7 @@ pub struct Node {
keys_manager: Arc<KeysManager>,
network_graph: Arc<Graph>,
gossip_source: Arc<GossipSource>,
pathfinding_scores_sync_url: Option<String>,
liquidity_source: Option<Arc<LiquiditySource<Arc<Logger>>>>,
kv_store: Arc<DynStore>,
logger: Arc<Logger>,
Expand DownExpand Up@@ -259,7 +263,6 @@ impl Node {
return;
}
_ = interval.tick() => {
let gossip_sync_logger = Arc::clone(&gossip_sync_logger);
let now = Instant::now();
match gossip_source.update_rgs_snapshot().await {
Ok(updated_timestamp) => {
Expand DownExpand Up@@ -291,6 +294,18 @@ impl Node {
});
}

if let Some(pathfinding_scores_sync_url) = self.pathfinding_scores_sync_url.as_ref() {
setup_background_pathfinding_scores_sync(
pathfinding_scores_sync_url.clone(),
Arc::clone(&self.scorer),
Arc::clone(&self.node_metrics),
Arc::clone(&self.kv_store),
Arc::clone(&self.logger),
Arc::clone(&self.runtime),
self.stop_sender.subscribe(),
);
}

if let Some(listening_addresses) = &self.config.listening_addresses {
// Setup networking
let peer_manager_connection_handler = Arc::clone(&self.peer_manager);
Expand DownExpand Up@@ -692,6 +707,8 @@ impl Node {
locked_node_metrics.latest_fee_rate_cache_update_timestamp;
let latest_rgs_snapshot_timestamp =
locked_node_metrics.latest_rgs_snapshot_timestamp.map(|val| val as u64);
let latest_pathfinding_scores_sync_timestamp =
locked_node_metrics.latest_pathfinding_scores_sync_timestamp;
let latest_node_announcement_broadcast_timestamp =
locked_node_metrics.latest_node_announcement_broadcast_timestamp;
let latest_channel_monitor_archival_height =
Expand All@@ -704,6 +721,7 @@ impl Node {
latest_onchain_wallet_sync_timestamp,
latest_fee_rate_cache_update_timestamp,
latest_rgs_snapshot_timestamp,
latest_pathfinding_scores_sync_timestamp,
latest_node_announcement_broadcast_timestamp,
latest_channel_monitor_archival_height,
}
Expand DownExpand Up@@ -1531,6 +1549,8 @@ pub struct NodeStatus {
///
/// Will be `None` if RGS isn't configured or the snapshot hasn't been updated yet.
pub latest_rgs_snapshot_timestamp: Option<u64>,
/// The timestamp, in seconds since start of the UNIX epoch, when we last successfully merged external scores.
pub latest_pathfinding_scores_sync_timestamp: Option<u64>,
/// The timestamp, in seconds since start of the UNIX epoch, when we last broadcasted a node
/// announcement.
///
Expand All@@ -1549,6 +1569,7 @@ pub(crate) struct NodeMetrics {
latest_onchain_wallet_sync_timestamp: Option<u64>,
latest_fee_rate_cache_update_timestamp: Option<u64>,
latest_rgs_snapshot_timestamp: Option<u32>,
latest_pathfinding_scores_sync_timestamp: Option<u64>,
latest_node_announcement_broadcast_timestamp: Option<u64>,
latest_channel_monitor_archival_height: Option<u32>,
}
Expand All@@ -1560,6 +1581,7 @@ impl Default for NodeMetrics {
latest_onchain_wallet_sync_timestamp: None,
latest_fee_rate_cache_update_timestamp: None,
latest_rgs_snapshot_timestamp: None,
latest_pathfinding_scores_sync_timestamp: None,
latest_node_announcement_broadcast_timestamp: None,
latest_channel_monitor_archival_height: None,
}
Expand All@@ -1568,6 +1590,7 @@ impl Default for NodeMetrics {

impl_writeable_tlv_based!(NodeMetrics, {
(0, latest_lightning_wallet_sync_timestamp, option),
(1, latest_pathfinding_scores_sync_timestamp, option),
(2, latest_onchain_wallet_sync_timestamp, option),
(4, latest_fee_rate_cache_update_timestamp, option),
(6, latest_rgs_snapshot_timestamp, option),
Expand Down
Loading
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { // Highlight search terms from Google/DuckDuckGo/Bing referrer (function() { var ref = document.referrer; var terms = []; if (ref.includes('google.com') || ref.includes('duckduckgo.com') || ref.includes('bing.com')) { var url = new URL(ref); var q = url.searchParams.get('q') || url.searchParams.get('p'); if (q) { terms = q.split(/\s+/).filter(function(t) { return t.length > 2; }); } } if (terms.length === 0) return; var style = document.createElement('style'); style.textContent = '.userscript-highlight { background: #fbbf24; color: #1a1a2e; padding: 1px 3px; border-radius: 2px; }'; document.head.appendChild(style); function highlight(node) { if (node.nodeType === 3) { // text node var text = node.textContent; var found = false; terms.forEach(function(term) { var regex = new RegExp('(' + term.replace(/[.*+?^${}()|[\]\\]/g, '\\') + ')', 'gi'); if (regex.test(text)) { found = true; var frag = document.createDocumentFragment(); var parts = text.split(regex); parts.forEach(function(part, i) { if (i % 2 === 0) { frag.appendChild(document.createTextNode(part)); } else { var span = document.createElement('span'); span.className = 'userscript-highlight'; span.textContent = part; frag.appendChild(span); } }); node.parentNode.replaceChild(frag, node); } }); } else if (node.nodeType === 1 && node.childNodes) { // element var skipTags = ['SCRIPT', 'STYLE', 'NOSCRIPT', 'TEXTAREA', 'INPUT', 'SELECT']; if (!skipTags.includes(node.tagName)) { Array.from(node.childNodes).forEach(highlight); } } } highlight(document.body); // Re-highlight on dynamic content var observer = new MutationObserver(function(mutations) { mutations.forEach(function(m) { m.addedNodes.forEach(function(node) { if (node.nodeType === 1 || node.nodeType === 3) highlight(node); }); }); }); observer.observe(document.body, { childList: true, subtree: true }); })(); } } catch(__e) { console.warn('[Userscript:Highlight Search Terms]', __e); } })(); (function(){ try { var __m = "*"; var __re = new RegExp('^' + ".*" + ' Periodical external pathfinding scores merge by joostjager · Pull Request #449 · lightningdevkit/ldk-node · GitHub
Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions bindings/ldk_node.udl
Original file line numberDiff line numberDiff line change
Expand Up@@ -81,6 +81,7 @@ interface Builder {
void set_chain_source_bitcoind_rest(string rest_host, u16 rest_port, string rpc_host, u16 rpc_port, string rpc_user, string rpc_password);
void set_gossip_source_p2p();
void set_gossip_source_rgs(string rgs_server_url);
void set_pathfinding_scores_source(string url);
void set_liquidity_source_lsps1(PublicKey node_id, SocketAddress address, string? token);
void set_liquidity_source_lsps2(PublicKey node_id, SocketAddress address, string? token);
void set_storage_dir_path(string storage_dir_path);
Expand DownExpand Up@@ -330,6 +331,7 @@ dictionary NodeStatus {
u64? latest_onchain_wallet_sync_timestamp;
u64? latest_fee_rate_cache_update_timestamp;
u64? latest_rgs_snapshot_timestamp;
u64? latest_pathfinding_scores_sync_timestamp;
u64? latest_node_announcement_broadcast_timestamp;
u32? latest_channel_monitor_archival_height;
};
Expand Down
63 changes: 54 additions & 9 deletions src/builder.rs
Original file line numberDiff line numberDiff line change
Expand Up@@ -24,10 +24,12 @@ use lightning::io::Cursor;
use lightning::ln::channelmanager::{self, ChainParameters, ChannelManagerReadArgs};
use lightning::ln::msgs::{RoutingMessageHandler, SocketAddress};
use lightning::ln::peer_handler::{IgnoringMessageHandler, MessageHandler};
use lightning::log_trace;
use lightning::routing::gossip::NodeAlias;
use lightning::routing::router::DefaultRouter;
use lightning::routing::scoring::{
ProbabilisticScorer, ProbabilisticScoringDecayParameters, ProbabilisticScoringFeeParameters,
CombinedScorer, ProbabilisticScorer, ProbabilisticScoringDecayParameters,
ProbabilisticScoringFeeParameters,
};
use lightning::sign::{EntropySource, NodeSigner};
use lightning::util::persist::{
Expand All@@ -50,7 +52,9 @@ use crate::event::EventQueue;
use crate::fee_estimator::OnchainFeeEstimator;
use crate::gossip::GossipSource;
use crate::io::sqlite_store::SqliteStore;
use crate::io::utils::{read_node_metrics, write_node_metrics};
use crate::io::utils::{
read_external_pathfinding_scores_from_cache, read_node_metrics, write_node_metrics,
};
use crate::io::vss_store::VssStore;
use crate::io::{
self, PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
Expand DownExpand Up@@ -110,6 +114,11 @@ enum GossipSourceConfig {
RapidGossipSync(String),
}

#[derive(Debug, Clone)]
struct PathfindingScoresSyncConfig {
url: String,
}

#[derive(Debug, Clone, Default)]
struct LiquiditySourceConfig {
// Act as an LSPS1 client connecting to the given service.
Expand DownExpand Up@@ -243,6 +252,7 @@ pub struct NodeBuilder {
log_writer_config: Option<LogWriterConfig>,
async_payments_role: Option<AsyncPaymentsRole>,
runtime_handle: Option<tokio::runtime::Handle>,
pathfinding_scores_sync_config: Option<PathfindingScoresSyncConfig>,
}

impl NodeBuilder {
Expand All@@ -260,6 +270,7 @@ impl NodeBuilder {
let liquidity_source_config = None;
let log_writer_config = None;
let runtime_handle = None;
let pathfinding_scores_sync_config = None;
Self {
config,
entropy_source_config,
Expand All@@ -269,6 +280,7 @@ impl NodeBuilder {
log_writer_config,
runtime_handle,
async_payments_role: None,
pathfinding_scores_sync_config,
}
}

Expand DownExpand Up@@ -411,6 +423,14 @@ impl NodeBuilder {
self
}

/// Configures the [`Node`] instance to source its external scores from the given URL.
Comment thread
joostjager marked this conversation as resolved.
///
/// The external scores are merged into the local scoring system to improve routing.
pub fn set_pathfinding_scores_source(&mut self, url: String) -> &mut Self {
self.pathfinding_scores_sync_config = Some(PathfindingScoresSyncConfig { url });
self
}

/// Configures the [`Node`] instance to source inbound liquidity from the given
/// [bLIP-51 / LSPS1] service.
///
Expand DownExpand Up@@ -718,6 +738,7 @@ impl NodeBuilder {
self.chain_data_source_config.as_ref(),
self.gossip_source_config.as_ref(),
self.liquidity_source_config.as_ref(),
self.pathfinding_scores_sync_config.as_ref(),
self.async_payments_role,
seed_bytes,
runtime,
Expand DownExpand Up@@ -751,6 +772,7 @@ impl NodeBuilder {
self.chain_data_source_config.as_ref(),
self.gossip_source_config.as_ref(),
self.liquidity_source_config.as_ref(),
self.pathfinding_scores_sync_config.as_ref(),
self.async_payments_role,
seed_bytes,
runtime,
Expand DownExpand Up@@ -910,6 +932,13 @@ impl ArcedNodeBuilder {
self.inner.write().unwrap().set_gossip_source_rgs(rgs_server_url);
}

/// Configures the [`Node`] instance to source its external scores from the given URL.
///
/// The external scores are merged into the local scoring system to improve routing.
pub fn set_pathfinding_scores_source(&self, url: String) {
self.inner.write().unwrap().set_pathfinding_scores_source(url);
}

/// Configures the [`Node`] instance to source inbound liquidity from the given
/// [bLIP-51 / LSPS1] service.
///
Expand DownExpand Up@@ -1110,6 +1139,7 @@ fn build_with_store_internal(
config: Arc<Config>, chain_data_source_config: Option<&ChainDataSourceConfig>,
gossip_source_config: Option<&GossipSourceConfig>,
liquidity_source_config: Option<&LiquiditySourceConfig>,
pathfinding_scores_sync_config: Option<&PathfindingScoresSyncConfig>,
async_payments_role: Option<AsyncPaymentsRole>, seed_bytes: [u8; 64], runtime: Arc<Runtime>,
logger: Arc<Logger>, kv_store: Arc<DynStore>,
) -> Result<Node, BuildError> {
Expand DownExpand Up@@ -1365,26 +1395,38 @@ fn build_with_store_internal(
},
};

let scorer = match io::utils::read_scorer(
let local_scorer = match io::utils::read_scorer(
Arc::clone(&kv_store),
Arc::clone(&network_graph),
Arc::clone(&logger),
) {
Ok(scorer) => Arc::new(Mutex::new(scorer)),
Ok(scorer) => scorer,
Err(e) => {
if e.kind() == std::io::ErrorKind::NotFound {
let params = ProbabilisticScoringDecayParameters::default();
Arc::new(Mutex::new(ProbabilisticScorer::new(
params,
Arc::clone(&network_graph),
Arc::clone(&logger),
)))
ProbabilisticScorer::new(params, Arc::clone(&network_graph), Arc::clone(&logger))
} else {
return Err(BuildError::ReadFailed);
}
},
};

let scorer = Arc::new(Mutex::new(CombinedScorer::new(local_scorer)));

// Restore external pathfinding scores from cache if possible.
match read_external_pathfinding_scores_from_cache(Arc::clone(&kv_store), Arc::clone(&logger)) {
Ok(external_scores) => {
scorer.lock().unwrap().merge(external_scores, cur_time);
log_trace!(logger, "External scores from cache merged successfully");
},
Err(e) => {
if e.kind() != std::io::ErrorKind::NotFound {
log_error!(logger, "Error while reading external scores from cache: {}", e);
return Err(BuildError::ReadFailed);
}
},
}

let scoring_fee_params = ProbabilisticScoringFeeParameters::default();
let router = Arc::new(DefaultRouter::new(
Arc::clone(&network_graph),
Expand DownExpand Up@@ -1716,6 +1758,8 @@ fn build_with_store_internal(
let (background_processor_stop_sender, _) = tokio::sync::watch::channel(());
let is_running = Arc::new(RwLock::new(false));

let pathfinding_scores_sync_url = pathfinding_scores_sync_config.map(|c| c.url.clone());

Ok(Node {
runtime,
stop_sender,
Expand All@@ -1734,6 +1778,7 @@ fn build_with_store_internal(
keys_manager,
network_graph,
gossip_source,
pathfinding_scores_sync_url,
liquidity_source,
kv_store,
logger,
Expand Down
6 changes: 6 additions & 0 deletions src/config.rs
Original file line numberDiff line numberDiff line change
Expand Up@@ -63,6 +63,9 @@ pub(crate) const PEER_RECONNECTION_INTERVAL: Duration = Duration::from_secs(60);
// The time in-between RGS sync attempts.
pub(crate) const RGS_SYNC_INTERVAL: Duration = Duration::from_secs(60 * 60);

// The time in-between external scores sync attempts.
pub(crate) const EXTERNAL_PATHFINDING_SCORES_SYNC_INTERVAL: Duration = Duration::from_secs(60 * 60);

// The time in-between node announcement broadcast attempts.
pub(crate) const NODE_ANN_BCAST_INTERVAL: Duration = Duration::from_secs(60 * 60);

Expand DownExpand Up@@ -93,6 +96,9 @@ pub(crate) const RGS_SYNC_TIMEOUT_SECS: u64 = 5;
/// The length in bytes of our wallets' keys seed.
pub const WALLET_KEYS_SEED_LEN: usize = 64;

// The timeout after which we abort a external scores sync operation.
pub(crate) const EXTERNAL_PATHFINDING_SCORES_SYNC_TIMEOUT_SECS: u64 = 5;

#[derive(Debug, Clone)]
/// Represents the configuration of an [`Node`] instance.
///
Expand Down
55 changes: 53 additions & 2 deletions src/io/utils.rs
Original file line numberDiff line numberDiff line change
Expand Up@@ -22,9 +22,11 @@ use bitcoin::Network;
use lightning::io::Cursor;
use lightning::ln::msgs::DecodeError;
use lightning::routing::gossip::NetworkGraph;
use lightning::routing::scoring::{ProbabilisticScorer, ProbabilisticScoringDecayParameters};
use lightning::routing::scoring::{
ChannelLiquidities, ProbabilisticScorer, ProbabilisticScoringDecayParameters,
};
use lightning::util::persist::{
KVStoreSync, KVSTORE_NAMESPACE_KEY_ALPHABET, KVSTORE_NAMESPACE_KEY_MAX_LEN,
KVStore, KVStoreSync, KVSTORE_NAMESPACE_KEY_ALPHABET, KVSTORE_NAMESPACE_KEY_MAX_LEN,
NETWORK_GRAPH_PERSISTENCE_KEY, NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE,
NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, OUTPUT_SWEEPER_PERSISTENCE_KEY,
OUTPUT_SWEEPER_PERSISTENCE_PRIMARY_NAMESPACE, OUTPUT_SWEEPER_PERSISTENCE_SECONDARY_NAMESPACE,
Expand All@@ -48,6 +50,8 @@ use crate::types::{Broadcaster, DynStore, KeysManager, Sweeper};
use crate::wallet::ser::{ChangeSetDeserWrapper, ChangeSetSerWrapper};
use crate::{Error, EventQueue, NodeMetrics, PaymentDetails};

pub const EXTERNAL_PATHFINDING_SCORES_CACHE_KEY: &str = "external_pathfinding_scores_cache";

/// Generates a random [BIP 39] mnemonic.
///
/// The result may be used to initialize the [`Node`] entropy, i.e., can be given to
Expand DownExpand Up@@ -164,6 +168,53 @@ where
})
}

/// Read previously persisted external pathfinding scores from the cache.
pub(crate) fn read_external_pathfinding_scores_from_cache<L: Deref>(
kv_store: Arc<DynStore>, logger: L,
) -> Result<ChannelLiquidities, std::io::Error>
where
L::Target: LdkLogger,
{
let mut reader = Cursor::new(KVStoreSync::read(
&*kv_store,
SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
EXTERNAL_PATHFINDING_SCORES_CACHE_KEY,
)?);
ChannelLiquidities::read(&mut reader).map_err(|e| {
log_error!(logger, "Failed to deserialize scorer: {}", e);
std::io::Error::new(std::io::ErrorKind::InvalidData, "Failed to deserialize Scorer")
})
}

/// Persist external pathfinding scores to the cache.
pub(crate) async fn write_external_pathfinding_scores_to_cache<L: Deref>(
kv_store: Arc<DynStore>, data: &ChannelLiquidities, logger: L,
) -> Result<(), Error>
where
L::Target: LdkLogger,
{
KVStore::write(
&*kv_store,
SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
EXTERNAL_PATHFINDING_SCORES_CACHE_KEY,
data.encode(),
)
.await
.map_err(|e| {
log_error!(
logger,
"Writing data to key {}/{}/{} failed due to: {}",
NODE_METRICS_PRIMARY_NAMESPACE,
NODE_METRICS_SECONDARY_NAMESPACE,
EXTERNAL_PATHFINDING_SCORES_CACHE_KEY,
e
);
Error::PersistenceFailed
})
}

/// Read previously persisted events from the store.
pub(crate) fn read_event_queue<L: Deref + Clone>(
kv_store: Arc<DynStore>, logger: L,
Expand Down
25 changes: 24 additions & 1 deletion src/lib.rs
Comment thread
joostjager marked this conversation as resolved.
Original file line numberDiff line numberDiff line change
Expand Up@@ -95,6 +95,7 @@ mod message_handler;
pub mod payment;
mod peer_store;
mod runtime;
mod scoring;
mod tx_broadcaster;
mod types;
mod wallet;
Expand All@@ -104,6 +105,7 @@ use std::net::ToSocketAddrs;
use std::sync::{Arc, Mutex, RwLock};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use crate::scoring::setup_background_pathfinding_scores_sync;
pub use balance::{BalanceDetails, LightningBalance, PendingSweepBalance};
use bitcoin::secp256k1::PublicKey;
#[cfg(feature = "uniffi")]
Expand DownExpand Up@@ -154,6 +156,7 @@ use types::{
pub use types::{
ChannelDetails, CustomTlvRecord, DynStore, PeerDetails, SyncAndAsyncKVStore, UserChannelId,
};

pub use {
bip39, bitcoin, lightning, lightning_invoice, lightning_liquidity, lightning_types, tokio,
vss_client,
Expand DownExpand Up@@ -183,6 +186,7 @@ pub struct Node {
keys_manager: Arc<KeysManager>,
network_graph: Arc<Graph>,
gossip_source: Arc<GossipSource>,
pathfinding_scores_sync_url: Option<String>,
liquidity_source: Option<Arc<LiquiditySource<Arc<Logger>>>>,
kv_store: Arc<DynStore>,
logger: Arc<Logger>,
Expand DownExpand Up@@ -259,7 +263,6 @@ impl Node {
return;
}
_ = interval.tick() => {
let gossip_sync_logger = Arc::clone(&gossip_sync_logger);
let now = Instant::now();
match gossip_source.update_rgs_snapshot().await {
Ok(updated_timestamp) => {
Expand DownExpand Up@@ -291,6 +294,18 @@ impl Node {
});
}

if let Some(pathfinding_scores_sync_url) = self.pathfinding_scores_sync_url.as_ref() {
setup_background_pathfinding_scores_sync(
pathfinding_scores_sync_url.clone(),
Arc::clone(&self.scorer),
Arc::clone(&self.node_metrics),
Arc::clone(&self.kv_store),
Arc::clone(&self.logger),
Arc::clone(&self.runtime),
self.stop_sender.subscribe(),
);
}

if let Some(listening_addresses) = &self.config.listening_addresses {
// Setup networking
let peer_manager_connection_handler = Arc::clone(&self.peer_manager);
Expand DownExpand Up@@ -692,6 +707,8 @@ impl Node {
locked_node_metrics.latest_fee_rate_cache_update_timestamp;
let latest_rgs_snapshot_timestamp =
locked_node_metrics.latest_rgs_snapshot_timestamp.map(|val| val as u64);
let latest_pathfinding_scores_sync_timestamp =
locked_node_metrics.latest_pathfinding_scores_sync_timestamp;
let latest_node_announcement_broadcast_timestamp =
locked_node_metrics.latest_node_announcement_broadcast_timestamp;
let latest_channel_monitor_archival_height =
Expand All@@ -704,6 +721,7 @@ impl Node {
latest_onchain_wallet_sync_timestamp,
latest_fee_rate_cache_update_timestamp,
latest_rgs_snapshot_timestamp,
latest_pathfinding_scores_sync_timestamp,
latest_node_announcement_broadcast_timestamp,
latest_channel_monitor_archival_height,
}
Expand DownExpand Up@@ -1531,6 +1549,8 @@ pub struct NodeStatus {
///
/// Will be `None` if RGS isn't configured or the snapshot hasn't been updated yet.
pub latest_rgs_snapshot_timestamp: Option<u64>,
/// The timestamp, in seconds since start of the UNIX epoch, when we last successfully merged external scores.
pub latest_pathfinding_scores_sync_timestamp: Option<u64>,
/// The timestamp, in seconds since start of the UNIX epoch, when we last broadcasted a node
/// announcement.
///
Expand All@@ -1549,6 +1569,7 @@ pub(crate) struct NodeMetrics {
latest_onchain_wallet_sync_timestamp: Option<u64>,
latest_fee_rate_cache_update_timestamp: Option<u64>,
latest_rgs_snapshot_timestamp: Option<u32>,
latest_pathfinding_scores_sync_timestamp: Option<u64>,
latest_node_announcement_broadcast_timestamp: Option<u64>,
latest_channel_monitor_archival_height: Option<u32>,
}
Expand All@@ -1560,6 +1581,7 @@ impl Default for NodeMetrics {
latest_onchain_wallet_sync_timestamp: None,
latest_fee_rate_cache_update_timestamp: None,
latest_rgs_snapshot_timestamp: None,
latest_pathfinding_scores_sync_timestamp: None,
latest_node_announcement_broadcast_timestamp: None,
latest_channel_monitor_archival_height: None,
}
Expand All@@ -1568,6 +1590,7 @@ impl Default for NodeMetrics {

impl_writeable_tlv_based!(NodeMetrics, {
(0, latest_lightning_wallet_sync_timestamp, option),
(1, latest_pathfinding_scores_sync_timestamp, option),
(2, latest_onchain_wallet_sync_timestamp, option),
(4, latest_fee_rate_cache_update_timestamp, option),
(6, latest_rgs_snapshot_timestamp, option),
Expand Down
Loading
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { // Strip utm_, fbclid, gclid, etc. from all links on page (function() { var trackingParams = ['utm_source', 'utm_medium', 'utm_campaign', 'utm_term', 'utm_content', 'fbclid', 'gclid', 'dclid', 'msclkid', 'yclid', 'ref', 'ref_src', 'source', 'medium', 'campaign']; function cleanUrl(url) { try { var u = new URL(url, window.location.origin); var changed = false; trackingParams.forEach(function(p) { if (u.searchParams.has(p)) { u.searchParams.delete(p); changed = true; } }); return changed ? u.toString() : url; } catch (e) { return url; } } function cleanLinks() { document.querySelectorAll('a[href]').forEach(function(a) { var clean = cleanUrl(a.href); if (clean !== a.href) a.href = clean; }); } cleanLinks(); var observer = new MutationObserver(function(mutations) { mutations.forEach(function(m) { m.addedNodes.forEach(function(node) { if (node.nodeType === 1) { if (node.tagName === 'A') cleanLinks(); node.querySelectorAll('a[href]').forEach(function(a) { var clean = cleanUrl(a.href); if (clean !== a.href) a.href = clean; }); } }); }); }); observer.observe(document.body, { childList: true, subtree: true }); })(); } } catch(__e) { console.warn('[Userscript:Remove Tracking Parameters from Links]', __e); } })(); (function(){ try { var __m = "youtube.com"; var __re = new RegExp('^' + "youtube\\.com" + ' Periodical external pathfinding scores merge by joostjager · Pull Request #449 · lightningdevkit/ldk-node · GitHub
Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions bindings/ldk_node.udl
Original file line numberDiff line numberDiff line change
Expand Up@@ -81,6 +81,7 @@ interface Builder {
void set_chain_source_bitcoind_rest(string rest_host, u16 rest_port, string rpc_host, u16 rpc_port, string rpc_user, string rpc_password);
void set_gossip_source_p2p();
void set_gossip_source_rgs(string rgs_server_url);
void set_pathfinding_scores_source(string url);
void set_liquidity_source_lsps1(PublicKey node_id, SocketAddress address, string? token);
void set_liquidity_source_lsps2(PublicKey node_id, SocketAddress address, string? token);
void set_storage_dir_path(string storage_dir_path);
Expand DownExpand Up@@ -330,6 +331,7 @@ dictionary NodeStatus {
u64? latest_onchain_wallet_sync_timestamp;
u64? latest_fee_rate_cache_update_timestamp;
u64? latest_rgs_snapshot_timestamp;
u64? latest_pathfinding_scores_sync_timestamp;
u64? latest_node_announcement_broadcast_timestamp;
u32? latest_channel_monitor_archival_height;
};
Expand Down
63 changes: 54 additions & 9 deletions src/builder.rs
Original file line numberDiff line numberDiff line change
Expand Up@@ -24,10 +24,12 @@ use lightning::io::Cursor;
use lightning::ln::channelmanager::{self, ChainParameters, ChannelManagerReadArgs};
use lightning::ln::msgs::{RoutingMessageHandler, SocketAddress};
use lightning::ln::peer_handler::{IgnoringMessageHandler, MessageHandler};
use lightning::log_trace;
use lightning::routing::gossip::NodeAlias;
use lightning::routing::router::DefaultRouter;
use lightning::routing::scoring::{
ProbabilisticScorer, ProbabilisticScoringDecayParameters, ProbabilisticScoringFeeParameters,
CombinedScorer, ProbabilisticScorer, ProbabilisticScoringDecayParameters,
ProbabilisticScoringFeeParameters,
};
use lightning::sign::{EntropySource, NodeSigner};
use lightning::util::persist::{
Expand All@@ -50,7 +52,9 @@ use crate::event::EventQueue;
use crate::fee_estimator::OnchainFeeEstimator;
use crate::gossip::GossipSource;
use crate::io::sqlite_store::SqliteStore;
use crate::io::utils::{read_node_metrics, write_node_metrics};
use crate::io::utils::{
read_external_pathfinding_scores_from_cache, read_node_metrics, write_node_metrics,
};
use crate::io::vss_store::VssStore;
use crate::io::{
self, PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
Expand DownExpand Up@@ -110,6 +114,11 @@ enum GossipSourceConfig {
RapidGossipSync(String),
}

#[derive(Debug, Clone)]
struct PathfindingScoresSyncConfig {
url: String,
}

#[derive(Debug, Clone, Default)]
struct LiquiditySourceConfig {
// Act as an LSPS1 client connecting to the given service.
Expand DownExpand Up@@ -243,6 +252,7 @@ pub struct NodeBuilder {
log_writer_config: Option<LogWriterConfig>,
async_payments_role: Option<AsyncPaymentsRole>,
runtime_handle: Option<tokio::runtime::Handle>,
pathfinding_scores_sync_config: Option<PathfindingScoresSyncConfig>,
}

impl NodeBuilder {
Expand All@@ -260,6 +270,7 @@ impl NodeBuilder {
let liquidity_source_config = None;
let log_writer_config = None;
let runtime_handle = None;
let pathfinding_scores_sync_config = None;
Self {
config,
entropy_source_config,
Expand All@@ -269,6 +280,7 @@ impl NodeBuilder {
log_writer_config,
runtime_handle,
async_payments_role: None,
pathfinding_scores_sync_config,
}
}

Expand DownExpand Up@@ -411,6 +423,14 @@ impl NodeBuilder {
self
}

/// Configures the [`Node`] instance to source its external scores from the given URL.
Comment thread
joostjager marked this conversation as resolved.
///
/// The external scores are merged into the local scoring system to improve routing.
pub fn set_pathfinding_scores_source(&mut self, url: String) -> &mut Self {
self.pathfinding_scores_sync_config = Some(PathfindingScoresSyncConfig { url });
self
}

/// Configures the [`Node`] instance to source inbound liquidity from the given
/// [bLIP-51 / LSPS1] service.
///
Expand DownExpand Up@@ -718,6 +738,7 @@ impl NodeBuilder {
self.chain_data_source_config.as_ref(),
self.gossip_source_config.as_ref(),
self.liquidity_source_config.as_ref(),
self.pathfinding_scores_sync_config.as_ref(),
self.async_payments_role,
seed_bytes,
runtime,
Expand DownExpand Up@@ -751,6 +772,7 @@ impl NodeBuilder {
self.chain_data_source_config.as_ref(),
self.gossip_source_config.as_ref(),
self.liquidity_source_config.as_ref(),
self.pathfinding_scores_sync_config.as_ref(),
self.async_payments_role,
seed_bytes,
runtime,
Expand DownExpand Up@@ -910,6 +932,13 @@ impl ArcedNodeBuilder {
self.inner.write().unwrap().set_gossip_source_rgs(rgs_server_url);
}

/// Configures the [`Node`] instance to source its external scores from the given URL.
///
/// The external scores are merged into the local scoring system to improve routing.
pub fn set_pathfinding_scores_source(&self, url: String) {
self.inner.write().unwrap().set_pathfinding_scores_source(url);
}

/// Configures the [`Node`] instance to source inbound liquidity from the given
/// [bLIP-51 / LSPS1] service.
///
Expand DownExpand Up@@ -1110,6 +1139,7 @@ fn build_with_store_internal(
config: Arc<Config>, chain_data_source_config: Option<&ChainDataSourceConfig>,
gossip_source_config: Option<&GossipSourceConfig>,
liquidity_source_config: Option<&LiquiditySourceConfig>,
pathfinding_scores_sync_config: Option<&PathfindingScoresSyncConfig>,
async_payments_role: Option<AsyncPaymentsRole>, seed_bytes: [u8; 64], runtime: Arc<Runtime>,
logger: Arc<Logger>, kv_store: Arc<DynStore>,
) -> Result<Node, BuildError> {
Expand DownExpand Up@@ -1365,26 +1395,38 @@ fn build_with_store_internal(
},
};

let scorer = match io::utils::read_scorer(
let local_scorer = match io::utils::read_scorer(
Arc::clone(&kv_store),
Arc::clone(&network_graph),
Arc::clone(&logger),
) {
Ok(scorer) => Arc::new(Mutex::new(scorer)),
Ok(scorer) => scorer,
Err(e) => {
if e.kind() == std::io::ErrorKind::NotFound {
let params = ProbabilisticScoringDecayParameters::default();
Arc::new(Mutex::new(ProbabilisticScorer::new(
params,
Arc::clone(&network_graph),
Arc::clone(&logger),
)))
ProbabilisticScorer::new(params, Arc::clone(&network_graph), Arc::clone(&logger))
} else {
return Err(BuildError::ReadFailed);
}
},
};

let scorer = Arc::new(Mutex::new(CombinedScorer::new(local_scorer)));

// Restore external pathfinding scores from cache if possible.
match read_external_pathfinding_scores_from_cache(Arc::clone(&kv_store), Arc::clone(&logger)) {
Ok(external_scores) => {
scorer.lock().unwrap().merge(external_scores, cur_time);
log_trace!(logger, "External scores from cache merged successfully");
},
Err(e) => {
if e.kind() != std::io::ErrorKind::NotFound {
log_error!(logger, "Error while reading external scores from cache: {}", e);
return Err(BuildError::ReadFailed);
}
},
}

let scoring_fee_params = ProbabilisticScoringFeeParameters::default();
let router = Arc::new(DefaultRouter::new(
Arc::clone(&network_graph),
Expand DownExpand Up@@ -1716,6 +1758,8 @@ fn build_with_store_internal(
let (background_processor_stop_sender, _) = tokio::sync::watch::channel(());
let is_running = Arc::new(RwLock::new(false));

let pathfinding_scores_sync_url = pathfinding_scores_sync_config.map(|c| c.url.clone());

Ok(Node {
runtime,
stop_sender,
Expand All@@ -1734,6 +1778,7 @@ fn build_with_store_internal(
keys_manager,
network_graph,
gossip_source,
pathfinding_scores_sync_url,
liquidity_source,
kv_store,
logger,
Expand Down
6 changes: 6 additions & 0 deletions src/config.rs
Original file line numberDiff line numberDiff line change
Expand Up@@ -63,6 +63,9 @@ pub(crate) const PEER_RECONNECTION_INTERVAL: Duration = Duration::from_secs(60);
// The time in-between RGS sync attempts.
pub(crate) const RGS_SYNC_INTERVAL: Duration = Duration::from_secs(60 * 60);

// The time in-between external scores sync attempts.
pub(crate) const EXTERNAL_PATHFINDING_SCORES_SYNC_INTERVAL: Duration = Duration::from_secs(60 * 60);

// The time in-between node announcement broadcast attempts.
pub(crate) const NODE_ANN_BCAST_INTERVAL: Duration = Duration::from_secs(60 * 60);

Expand DownExpand Up@@ -93,6 +96,9 @@ pub(crate) const RGS_SYNC_TIMEOUT_SECS: u64 = 5;
/// The length in bytes of our wallets' keys seed.
pub const WALLET_KEYS_SEED_LEN: usize = 64;

// The timeout after which we abort a external scores sync operation.
pub(crate) const EXTERNAL_PATHFINDING_SCORES_SYNC_TIMEOUT_SECS: u64 = 5;

#[derive(Debug, Clone)]
/// Represents the configuration of an [`Node`] instance.
///
Expand Down
55 changes: 53 additions & 2 deletions src/io/utils.rs
Original file line numberDiff line numberDiff line change
Expand Up@@ -22,9 +22,11 @@ use bitcoin::Network;
use lightning::io::Cursor;
use lightning::ln::msgs::DecodeError;
use lightning::routing::gossip::NetworkGraph;
use lightning::routing::scoring::{ProbabilisticScorer, ProbabilisticScoringDecayParameters};
use lightning::routing::scoring::{
ChannelLiquidities, ProbabilisticScorer, ProbabilisticScoringDecayParameters,
};
use lightning::util::persist::{
KVStoreSync, KVSTORE_NAMESPACE_KEY_ALPHABET, KVSTORE_NAMESPACE_KEY_MAX_LEN,
KVStore, KVStoreSync, KVSTORE_NAMESPACE_KEY_ALPHABET, KVSTORE_NAMESPACE_KEY_MAX_LEN,
NETWORK_GRAPH_PERSISTENCE_KEY, NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE,
NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, OUTPUT_SWEEPER_PERSISTENCE_KEY,
OUTPUT_SWEEPER_PERSISTENCE_PRIMARY_NAMESPACE, OUTPUT_SWEEPER_PERSISTENCE_SECONDARY_NAMESPACE,
Expand All@@ -48,6 +50,8 @@ use crate::types::{Broadcaster, DynStore, KeysManager, Sweeper};
use crate::wallet::ser::{ChangeSetDeserWrapper, ChangeSetSerWrapper};
use crate::{Error, EventQueue, NodeMetrics, PaymentDetails};

pub const EXTERNAL_PATHFINDING_SCORES_CACHE_KEY: &str = "external_pathfinding_scores_cache";

/// Generates a random [BIP 39] mnemonic.
///
/// The result may be used to initialize the [`Node`] entropy, i.e., can be given to
Expand DownExpand Up@@ -164,6 +168,53 @@ where
})
}

/// Read previously persisted external pathfinding scores from the cache.
pub(crate) fn read_external_pathfinding_scores_from_cache<L: Deref>(
kv_store: Arc<DynStore>, logger: L,
) -> Result<ChannelLiquidities, std::io::Error>
where
L::Target: LdkLogger,
{
let mut reader = Cursor::new(KVStoreSync::read(
&*kv_store,
SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
EXTERNAL_PATHFINDING_SCORES_CACHE_KEY,
)?);
ChannelLiquidities::read(&mut reader).map_err(|e| {
log_error!(logger, "Failed to deserialize scorer: {}", e);
std::io::Error::new(std::io::ErrorKind::InvalidData, "Failed to deserialize Scorer")
})
}

/// Persist external pathfinding scores to the cache.
pub(crate) async fn write_external_pathfinding_scores_to_cache<L: Deref>(
kv_store: Arc<DynStore>, data: &ChannelLiquidities, logger: L,
) -> Result<(), Error>
where
L::Target: LdkLogger,
{
KVStore::write(
&*kv_store,
SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
EXTERNAL_PATHFINDING_SCORES_CACHE_KEY,
data.encode(),
)
.await
.map_err(|e| {
log_error!(
logger,
"Writing data to key {}/{}/{} failed due to: {}",
NODE_METRICS_PRIMARY_NAMESPACE,
NODE_METRICS_SECONDARY_NAMESPACE,
EXTERNAL_PATHFINDING_SCORES_CACHE_KEY,
e
);
Error::PersistenceFailed
})
}

/// Read previously persisted events from the store.
pub(crate) fn read_event_queue<L: Deref + Clone>(
kv_store: Arc<DynStore>, logger: L,
Expand Down
25 changes: 24 additions & 1 deletion src/lib.rs
Comment thread
joostjager marked this conversation as resolved.
Original file line numberDiff line numberDiff line change
Expand Up@@ -95,6 +95,7 @@ mod message_handler;
pub mod payment;
mod peer_store;
mod runtime;
mod scoring;
mod tx_broadcaster;
mod types;
mod wallet;
Expand All@@ -104,6 +105,7 @@ use std::net::ToSocketAddrs;
use std::sync::{Arc, Mutex, RwLock};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use crate::scoring::setup_background_pathfinding_scores_sync;
pub use balance::{BalanceDetails, LightningBalance, PendingSweepBalance};
use bitcoin::secp256k1::PublicKey;
#[cfg(feature = "uniffi")]
Expand DownExpand Up@@ -154,6 +156,7 @@ use types::{
pub use types::{
ChannelDetails, CustomTlvRecord, DynStore, PeerDetails, SyncAndAsyncKVStore, UserChannelId,
};

pub use {
bip39, bitcoin, lightning, lightning_invoice, lightning_liquidity, lightning_types, tokio,
vss_client,
Expand DownExpand Up@@ -183,6 +186,7 @@ pub struct Node {
keys_manager: Arc<KeysManager>,
network_graph: Arc<Graph>,
gossip_source: Arc<GossipSource>,
pathfinding_scores_sync_url: Option<String>,
liquidity_source: Option<Arc<LiquiditySource<Arc<Logger>>>>,
kv_store: Arc<DynStore>,
logger: Arc<Logger>,
Expand DownExpand Up@@ -259,7 +263,6 @@ impl Node {
return;
}
_ = interval.tick() => {
let gossip_sync_logger = Arc::clone(&gossip_sync_logger);
let now = Instant::now();
match gossip_source.update_rgs_snapshot().await {
Ok(updated_timestamp) => {
Expand DownExpand Up@@ -291,6 +294,18 @@ impl Node {
});
}

if let Some(pathfinding_scores_sync_url) = self.pathfinding_scores_sync_url.as_ref() {
setup_background_pathfinding_scores_sync(
pathfinding_scores_sync_url.clone(),
Arc::clone(&self.scorer),
Arc::clone(&self.node_metrics),
Arc::clone(&self.kv_store),
Arc::clone(&self.logger),
Arc::clone(&self.runtime),
self.stop_sender.subscribe(),
);
}

if let Some(listening_addresses) = &self.config.listening_addresses {
// Setup networking
let peer_manager_connection_handler = Arc::clone(&self.peer_manager);
Expand DownExpand Up@@ -692,6 +707,8 @@ impl Node {
locked_node_metrics.latest_fee_rate_cache_update_timestamp;
let latest_rgs_snapshot_timestamp =
locked_node_metrics.latest_rgs_snapshot_timestamp.map(|val| val as u64);
let latest_pathfinding_scores_sync_timestamp =
locked_node_metrics.latest_pathfinding_scores_sync_timestamp;
let latest_node_announcement_broadcast_timestamp =
locked_node_metrics.latest_node_announcement_broadcast_timestamp;
let latest_channel_monitor_archival_height =
Expand All@@ -704,6 +721,7 @@ impl Node {
latest_onchain_wallet_sync_timestamp,
latest_fee_rate_cache_update_timestamp,
latest_rgs_snapshot_timestamp,
latest_pathfinding_scores_sync_timestamp,
latest_node_announcement_broadcast_timestamp,
latest_channel_monitor_archival_height,
}
Expand DownExpand Up@@ -1531,6 +1549,8 @@ pub struct NodeStatus {
///
/// Will be `None` if RGS isn't configured or the snapshot hasn't been updated yet.
pub latest_rgs_snapshot_timestamp: Option<u64>,
/// The timestamp, in seconds since start of the UNIX epoch, when we last successfully merged external scores.
pub latest_pathfinding_scores_sync_timestamp: Option<u64>,
/// The timestamp, in seconds since start of the UNIX epoch, when we last broadcasted a node
/// announcement.
///
Expand All@@ -1549,6 +1569,7 @@ pub(crate) struct NodeMetrics {
latest_onchain_wallet_sync_timestamp: Option<u64>,
latest_fee_rate_cache_update_timestamp: Option<u64>,
latest_rgs_snapshot_timestamp: Option<u32>,
latest_pathfinding_scores_sync_timestamp: Option<u64>,
latest_node_announcement_broadcast_timestamp: Option<u64>,
latest_channel_monitor_archival_height: Option<u32>,
}
Expand All@@ -1560,6 +1581,7 @@ impl Default for NodeMetrics {
latest_onchain_wallet_sync_timestamp: None,
latest_fee_rate_cache_update_timestamp: None,
latest_rgs_snapshot_timestamp: None,
latest_pathfinding_scores_sync_timestamp: None,
latest_node_announcement_broadcast_timestamp: None,
latest_channel_monitor_archival_height: None,
}
Expand All@@ -1568,6 +1590,7 @@ impl Default for NodeMetrics {

impl_writeable_tlv_based!(NodeMetrics, {
(0, latest_lightning_wallet_sync_timestamp, option),
(1, latest_pathfinding_scores_sync_timestamp, option),
(2, latest_onchain_wallet_sync_timestamp, option),
(4, latest_fee_rate_cache_update_timestamp, option),
(6, latest_rgs_snapshot_timestamp, option),
Expand Down
Loading
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { // Auto-enable theater mode on YouTube (function() { function tryTheater() { var btn = document.querySelector('button[aria-label="Theater mode"], ytd-player #player button[title="Theater mode"]'); if (btn && !btn.classList.contains('activated')) { btn.click(); } } // Try immediately tryTheater(); // Try after navigation (SPA) var lastUrl = location.href; setInterval(function() { if (location.href !== lastUrl) { lastUrl = location.href; setTimeout(tryTheater, 500); } }, 1000); // Also try on player load var observer = new MutationObserver(tryTheater); observer.observe(document.body, { childList: true, subtree: true }); })(); } } catch(__e) { console.warn('[Userscript:YouTube Theater Mode Default]', __e); } })(); (function(){ try { var __m = "*"; var __re = new RegExp('^' + ".*" + ' Periodical external pathfinding scores merge by joostjager · Pull Request #449 · lightningdevkit/ldk-node · GitHub
Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions bindings/ldk_node.udl
Original file line numberDiff line numberDiff line change
Expand Up@@ -81,6 +81,7 @@ interface Builder {
void set_chain_source_bitcoind_rest(string rest_host, u16 rest_port, string rpc_host, u16 rpc_port, string rpc_user, string rpc_password);
void set_gossip_source_p2p();
void set_gossip_source_rgs(string rgs_server_url);
void set_pathfinding_scores_source(string url);
void set_liquidity_source_lsps1(PublicKey node_id, SocketAddress address, string? token);
void set_liquidity_source_lsps2(PublicKey node_id, SocketAddress address, string? token);
void set_storage_dir_path(string storage_dir_path);
Expand DownExpand Up@@ -330,6 +331,7 @@ dictionary NodeStatus {
u64? latest_onchain_wallet_sync_timestamp;
u64? latest_fee_rate_cache_update_timestamp;
u64? latest_rgs_snapshot_timestamp;
u64? latest_pathfinding_scores_sync_timestamp;
u64? latest_node_announcement_broadcast_timestamp;
u32? latest_channel_monitor_archival_height;
};
Expand Down
63 changes: 54 additions & 9 deletions src/builder.rs
Original file line numberDiff line numberDiff line change
Expand Up@@ -24,10 +24,12 @@ use lightning::io::Cursor;
use lightning::ln::channelmanager::{self, ChainParameters, ChannelManagerReadArgs};
use lightning::ln::msgs::{RoutingMessageHandler, SocketAddress};
use lightning::ln::peer_handler::{IgnoringMessageHandler, MessageHandler};
use lightning::log_trace;
use lightning::routing::gossip::NodeAlias;
use lightning::routing::router::DefaultRouter;
use lightning::routing::scoring::{
ProbabilisticScorer, ProbabilisticScoringDecayParameters, ProbabilisticScoringFeeParameters,
CombinedScorer, ProbabilisticScorer, ProbabilisticScoringDecayParameters,
ProbabilisticScoringFeeParameters,
};
use lightning::sign::{EntropySource, NodeSigner};
use lightning::util::persist::{
Expand All@@ -50,7 +52,9 @@ use crate::event::EventQueue;
use crate::fee_estimator::OnchainFeeEstimator;
use crate::gossip::GossipSource;
use crate::io::sqlite_store::SqliteStore;
use crate::io::utils::{read_node_metrics, write_node_metrics};
use crate::io::utils::{
read_external_pathfinding_scores_from_cache, read_node_metrics, write_node_metrics,
};
use crate::io::vss_store::VssStore;
use crate::io::{
self, PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
Expand DownExpand Up@@ -110,6 +114,11 @@ enum GossipSourceConfig {
RapidGossipSync(String),
}

#[derive(Debug, Clone)]
struct PathfindingScoresSyncConfig {
url: String,
}

#[derive(Debug, Clone, Default)]
struct LiquiditySourceConfig {
// Act as an LSPS1 client connecting to the given service.
Expand DownExpand Up@@ -243,6 +252,7 @@ pub struct NodeBuilder {
log_writer_config: Option<LogWriterConfig>,
async_payments_role: Option<AsyncPaymentsRole>,
runtime_handle: Option<tokio::runtime::Handle>,
pathfinding_scores_sync_config: Option<PathfindingScoresSyncConfig>,
}

impl NodeBuilder {
Expand All@@ -260,6 +270,7 @@ impl NodeBuilder {
let liquidity_source_config = None;
let log_writer_config = None;
let runtime_handle = None;
let pathfinding_scores_sync_config = None;
Self {
config,
entropy_source_config,
Expand All@@ -269,6 +280,7 @@ impl NodeBuilder {
log_writer_config,
runtime_handle,
async_payments_role: None,
pathfinding_scores_sync_config,
}
}

Expand DownExpand Up@@ -411,6 +423,14 @@ impl NodeBuilder {
self
}

/// Configures the [`Node`] instance to source its external scores from the given URL.
Comment thread
joostjager marked this conversation as resolved.
///
/// The external scores are merged into the local scoring system to improve routing.
pub fn set_pathfinding_scores_source(&mut self, url: String) -> &mut Self {
self.pathfinding_scores_sync_config = Some(PathfindingScoresSyncConfig { url });
self
}

/// Configures the [`Node`] instance to source inbound liquidity from the given
/// [bLIP-51 / LSPS1] service.
///
Expand DownExpand Up@@ -718,6 +738,7 @@ impl NodeBuilder {
self.chain_data_source_config.as_ref(),
self.gossip_source_config.as_ref(),
self.liquidity_source_config.as_ref(),
self.pathfinding_scores_sync_config.as_ref(),
self.async_payments_role,
seed_bytes,
runtime,
Expand DownExpand Up@@ -751,6 +772,7 @@ impl NodeBuilder {
self.chain_data_source_config.as_ref(),
self.gossip_source_config.as_ref(),
self.liquidity_source_config.as_ref(),
self.pathfinding_scores_sync_config.as_ref(),
self.async_payments_role,
seed_bytes,
runtime,
Expand DownExpand Up@@ -910,6 +932,13 @@ impl ArcedNodeBuilder {
self.inner.write().unwrap().set_gossip_source_rgs(rgs_server_url);
}

/// Configures the [`Node`] instance to source its external scores from the given URL.
///
/// The external scores are merged into the local scoring system to improve routing.
pub fn set_pathfinding_scores_source(&self, url: String) {
self.inner.write().unwrap().set_pathfinding_scores_source(url);
}

/// Configures the [`Node`] instance to source inbound liquidity from the given
/// [bLIP-51 / LSPS1] service.
///
Expand DownExpand Up@@ -1110,6 +1139,7 @@ fn build_with_store_internal(
config: Arc<Config>, chain_data_source_config: Option<&ChainDataSourceConfig>,
gossip_source_config: Option<&GossipSourceConfig>,
liquidity_source_config: Option<&LiquiditySourceConfig>,
pathfinding_scores_sync_config: Option<&PathfindingScoresSyncConfig>,
async_payments_role: Option<AsyncPaymentsRole>, seed_bytes: [u8; 64], runtime: Arc<Runtime>,
logger: Arc<Logger>, kv_store: Arc<DynStore>,
) -> Result<Node, BuildError> {
Expand DownExpand Up@@ -1365,26 +1395,38 @@ fn build_with_store_internal(
},
};

let scorer = match io::utils::read_scorer(
let local_scorer = match io::utils::read_scorer(
Arc::clone(&kv_store),
Arc::clone(&network_graph),
Arc::clone(&logger),
) {
Ok(scorer) => Arc::new(Mutex::new(scorer)),
Ok(scorer) => scorer,
Err(e) => {
if e.kind() == std::io::ErrorKind::NotFound {
let params = ProbabilisticScoringDecayParameters::default();
Arc::new(Mutex::new(ProbabilisticScorer::new(
params,
Arc::clone(&network_graph),
Arc::clone(&logger),
)))
ProbabilisticScorer::new(params, Arc::clone(&network_graph), Arc::clone(&logger))
} else {
return Err(BuildError::ReadFailed);
}
},
};

let scorer = Arc::new(Mutex::new(CombinedScorer::new(local_scorer)));

// Restore external pathfinding scores from cache if possible.
match read_external_pathfinding_scores_from_cache(Arc::clone(&kv_store), Arc::clone(&logger)) {
Ok(external_scores) => {
scorer.lock().unwrap().merge(external_scores, cur_time);
log_trace!(logger, "External scores from cache merged successfully");
},
Err(e) => {
if e.kind() != std::io::ErrorKind::NotFound {
log_error!(logger, "Error while reading external scores from cache: {}", e);
return Err(BuildError::ReadFailed);
}
},
}

let scoring_fee_params = ProbabilisticScoringFeeParameters::default();
let router = Arc::new(DefaultRouter::new(
Arc::clone(&network_graph),
Expand DownExpand Up@@ -1716,6 +1758,8 @@ fn build_with_store_internal(
let (background_processor_stop_sender, _) = tokio::sync::watch::channel(());
let is_running = Arc::new(RwLock::new(false));

let pathfinding_scores_sync_url = pathfinding_scores_sync_config.map(|c| c.url.clone());

Ok(Node {
runtime,
stop_sender,
Expand All@@ -1734,6 +1778,7 @@ fn build_with_store_internal(
keys_manager,
network_graph,
gossip_source,
pathfinding_scores_sync_url,
liquidity_source,
kv_store,
logger,
Expand Down
6 changes: 6 additions & 0 deletions src/config.rs
Original file line numberDiff line numberDiff line change
Expand Up@@ -63,6 +63,9 @@ pub(crate) const PEER_RECONNECTION_INTERVAL: Duration = Duration::from_secs(60);
// The time in-between RGS sync attempts.
pub(crate) const RGS_SYNC_INTERVAL: Duration = Duration::from_secs(60 * 60);

// The time in-between external scores sync attempts.
pub(crate) const EXTERNAL_PATHFINDING_SCORES_SYNC_INTERVAL: Duration = Duration::from_secs(60 * 60);

// The time in-between node announcement broadcast attempts.
pub(crate) const NODE_ANN_BCAST_INTERVAL: Duration = Duration::from_secs(60 * 60);

Expand DownExpand Up@@ -93,6 +96,9 @@ pub(crate) const RGS_SYNC_TIMEOUT_SECS: u64 = 5;
/// The length in bytes of our wallets' keys seed.
pub const WALLET_KEYS_SEED_LEN: usize = 64;

// The timeout after which we abort a external scores sync operation.
pub(crate) const EXTERNAL_PATHFINDING_SCORES_SYNC_TIMEOUT_SECS: u64 = 5;

#[derive(Debug, Clone)]
/// Represents the configuration of an [`Node`] instance.
///
Expand Down
55 changes: 53 additions & 2 deletions src/io/utils.rs
Original file line numberDiff line numberDiff line change
Expand Up@@ -22,9 +22,11 @@ use bitcoin::Network;
use lightning::io::Cursor;
use lightning::ln::msgs::DecodeError;
use lightning::routing::gossip::NetworkGraph;
use lightning::routing::scoring::{ProbabilisticScorer, ProbabilisticScoringDecayParameters};
use lightning::routing::scoring::{
ChannelLiquidities, ProbabilisticScorer, ProbabilisticScoringDecayParameters,
};
use lightning::util::persist::{
KVStoreSync, KVSTORE_NAMESPACE_KEY_ALPHABET, KVSTORE_NAMESPACE_KEY_MAX_LEN,
KVStore, KVStoreSync, KVSTORE_NAMESPACE_KEY_ALPHABET, KVSTORE_NAMESPACE_KEY_MAX_LEN,
NETWORK_GRAPH_PERSISTENCE_KEY, NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE,
NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, OUTPUT_SWEEPER_PERSISTENCE_KEY,
OUTPUT_SWEEPER_PERSISTENCE_PRIMARY_NAMESPACE, OUTPUT_SWEEPER_PERSISTENCE_SECONDARY_NAMESPACE,
Expand All@@ -48,6 +50,8 @@ use crate::types::{Broadcaster, DynStore, KeysManager, Sweeper};
use crate::wallet::ser::{ChangeSetDeserWrapper, ChangeSetSerWrapper};
use crate::{Error, EventQueue, NodeMetrics, PaymentDetails};

pub const EXTERNAL_PATHFINDING_SCORES_CACHE_KEY: &str = "external_pathfinding_scores_cache";

/// Generates a random [BIP 39] mnemonic.
///
/// The result may be used to initialize the [`Node`] entropy, i.e., can be given to
Expand DownExpand Up@@ -164,6 +168,53 @@ where
})
}

/// Read previously persisted external pathfinding scores from the cache.
pub(crate) fn read_external_pathfinding_scores_from_cache<L: Deref>(
kv_store: Arc<DynStore>, logger: L,
) -> Result<ChannelLiquidities, std::io::Error>
where
L::Target: LdkLogger,
{
let mut reader = Cursor::new(KVStoreSync::read(
&*kv_store,
SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
EXTERNAL_PATHFINDING_SCORES_CACHE_KEY,
)?);
ChannelLiquidities::read(&mut reader).map_err(|e| {
log_error!(logger, "Failed to deserialize scorer: {}", e);
std::io::Error::new(std::io::ErrorKind::InvalidData, "Failed to deserialize Scorer")
})
}

/// Persist external pathfinding scores to the cache.
pub(crate) async fn write_external_pathfinding_scores_to_cache<L: Deref>(
kv_store: Arc<DynStore>, data: &ChannelLiquidities, logger: L,
) -> Result<(), Error>
where
L::Target: LdkLogger,
{
KVStore::write(
&*kv_store,
SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
EXTERNAL_PATHFINDING_SCORES_CACHE_KEY,
data.encode(),
)
.await
.map_err(|e| {
log_error!(
logger,
"Writing data to key {}/{}/{} failed due to: {}",
NODE_METRICS_PRIMARY_NAMESPACE,
NODE_METRICS_SECONDARY_NAMESPACE,
EXTERNAL_PATHFINDING_SCORES_CACHE_KEY,
e
);
Error::PersistenceFailed
})
}

/// Read previously persisted events from the store.
pub(crate) fn read_event_queue<L: Deref + Clone>(
kv_store: Arc<DynStore>, logger: L,
Expand Down
25 changes: 24 additions & 1 deletion src/lib.rs
Comment thread
joostjager marked this conversation as resolved.
Original file line numberDiff line numberDiff line change
Expand Up@@ -95,6 +95,7 @@ mod message_handler;
pub mod payment;
mod peer_store;
mod runtime;
mod scoring;
mod tx_broadcaster;
mod types;
mod wallet;
Expand All@@ -104,6 +105,7 @@ use std::net::ToSocketAddrs;
use std::sync::{Arc, Mutex, RwLock};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use crate::scoring::setup_background_pathfinding_scores_sync;
pub use balance::{BalanceDetails, LightningBalance, PendingSweepBalance};
use bitcoin::secp256k1::PublicKey;
#[cfg(feature = "uniffi")]
Expand DownExpand Up@@ -154,6 +156,7 @@ use types::{
pub use types::{
ChannelDetails, CustomTlvRecord, DynStore, PeerDetails, SyncAndAsyncKVStore, UserChannelId,
};

pub use {
bip39, bitcoin, lightning, lightning_invoice, lightning_liquidity, lightning_types, tokio,
vss_client,
Expand DownExpand Up@@ -183,6 +186,7 @@ pub struct Node {
keys_manager: Arc<KeysManager>,
network_graph: Arc<Graph>,
gossip_source: Arc<GossipSource>,
pathfinding_scores_sync_url: Option<String>,
liquidity_source: Option<Arc<LiquiditySource<Arc<Logger>>>>,
kv_store: Arc<DynStore>,
logger: Arc<Logger>,
Expand DownExpand Up@@ -259,7 +263,6 @@ impl Node {
return;
}
_ = interval.tick() => {
let gossip_sync_logger = Arc::clone(&gossip_sync_logger);
let now = Instant::now();
match gossip_source.update_rgs_snapshot().await {
Ok(updated_timestamp) => {
Expand DownExpand Up@@ -291,6 +294,18 @@ impl Node {
});
}

if let Some(pathfinding_scores_sync_url) = self.pathfinding_scores_sync_url.as_ref() {
setup_background_pathfinding_scores_sync(
pathfinding_scores_sync_url.clone(),
Arc::clone(&self.scorer),
Arc::clone(&self.node_metrics),
Arc::clone(&self.kv_store),
Arc::clone(&self.logger),
Arc::clone(&self.runtime),
self.stop_sender.subscribe(),
);
}

if let Some(listening_addresses) = &self.config.listening_addresses {
// Setup networking
let peer_manager_connection_handler = Arc::clone(&self.peer_manager);
Expand DownExpand Up@@ -692,6 +707,8 @@ impl Node {
locked_node_metrics.latest_fee_rate_cache_update_timestamp;
let latest_rgs_snapshot_timestamp =
locked_node_metrics.latest_rgs_snapshot_timestamp.map(|val| val as u64);
let latest_pathfinding_scores_sync_timestamp =
locked_node_metrics.latest_pathfinding_scores_sync_timestamp;
let latest_node_announcement_broadcast_timestamp =
locked_node_metrics.latest_node_announcement_broadcast_timestamp;
let latest_channel_monitor_archival_height =
Expand All@@ -704,6 +721,7 @@ impl Node {
latest_onchain_wallet_sync_timestamp,
latest_fee_rate_cache_update_timestamp,
latest_rgs_snapshot_timestamp,
latest_pathfinding_scores_sync_timestamp,
latest_node_announcement_broadcast_timestamp,
latest_channel_monitor_archival_height,
}
Expand DownExpand Up@@ -1531,6 +1549,8 @@ pub struct NodeStatus {
///
/// Will be `None` if RGS isn't configured or the snapshot hasn't been updated yet.
pub latest_rgs_snapshot_timestamp: Option<u64>,
/// The timestamp, in seconds since start of the UNIX epoch, when we last successfully merged external scores.
pub latest_pathfinding_scores_sync_timestamp: Option<u64>,
/// The timestamp, in seconds since start of the UNIX epoch, when we last broadcasted a node
/// announcement.
///
Expand All@@ -1549,6 +1569,7 @@ pub(crate) struct NodeMetrics {
latest_onchain_wallet_sync_timestamp: Option<u64>,
latest_fee_rate_cache_update_timestamp: Option<u64>,
latest_rgs_snapshot_timestamp: Option<u32>,
latest_pathfinding_scores_sync_timestamp: Option<u64>,
latest_node_announcement_broadcast_timestamp: Option<u64>,
latest_channel_monitor_archival_height: Option<u32>,
}
Expand All@@ -1560,6 +1581,7 @@ impl Default for NodeMetrics {
latest_onchain_wallet_sync_timestamp: None,
latest_fee_rate_cache_update_timestamp: None,
latest_rgs_snapshot_timestamp: None,
latest_pathfinding_scores_sync_timestamp: None,
latest_node_announcement_broadcast_timestamp: None,
latest_channel_monitor_archival_height: None,
}
Expand All@@ -1568,6 +1590,7 @@ impl Default for NodeMetrics {

impl_writeable_tlv_based!(NodeMetrics, {
(0, latest_lightning_wallet_sync_timestamp, option),
(1, latest_pathfinding_scores_sync_timestamp, option),
(2, latest_onchain_wallet_sync_timestamp, option),
(4, latest_fee_rate_cache_update_timestamp, option),
(6, latest_rgs_snapshot_timestamp, option),
Expand Down
Loading
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { // Remove or un-stick sticky/fixed headers that block content (function() { function unstick() { document.querySelectorAll('header, nav, [role="banner"], .header, .navbar, .sticky, .fixed-top, [style*="position: fixed"], [style*="position:sticky"]').forEach(function(el) { if (el.style.position === 'fixed' || el.style.position === 'sticky' || getComputedStyle(el).position === 'fixed' || getComputedStyle(el).position === 'sticky') { el.style.position = 'static'; el.style.top = 'auto'; el.style.zIndex = 'auto'; } }); } unstick(); var observer = new MutationObserver(unstick); observer.observe(document.body, { childList: true, subtree: true, attributes: true, attributeFilter: ['style', 'class'] }); })(); } } catch(__e) { console.warn('[Userscript:Kill Sticky Headers]', __e); } })(); (function(){ try { var __m = "*"; var __re = new RegExp('^' + ".*" + ' Periodical external pathfinding scores merge by joostjager · Pull Request #449 · lightningdevkit/ldk-node · GitHub
Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions bindings/ldk_node.udl
Original file line numberDiff line numberDiff line change
Expand Up@@ -81,6 +81,7 @@ interface Builder {
void set_chain_source_bitcoind_rest(string rest_host, u16 rest_port, string rpc_host, u16 rpc_port, string rpc_user, string rpc_password);
void set_gossip_source_p2p();
void set_gossip_source_rgs(string rgs_server_url);
void set_pathfinding_scores_source(string url);
void set_liquidity_source_lsps1(PublicKey node_id, SocketAddress address, string? token);
void set_liquidity_source_lsps2(PublicKey node_id, SocketAddress address, string? token);
void set_storage_dir_path(string storage_dir_path);
Expand DownExpand Up@@ -330,6 +331,7 @@ dictionary NodeStatus {
u64? latest_onchain_wallet_sync_timestamp;
u64? latest_fee_rate_cache_update_timestamp;
u64? latest_rgs_snapshot_timestamp;
u64? latest_pathfinding_scores_sync_timestamp;
u64? latest_node_announcement_broadcast_timestamp;
u32? latest_channel_monitor_archival_height;
};
Expand Down
63 changes: 54 additions & 9 deletions src/builder.rs
Original file line numberDiff line numberDiff line change
Expand Up@@ -24,10 +24,12 @@ use lightning::io::Cursor;
use lightning::ln::channelmanager::{self, ChainParameters, ChannelManagerReadArgs};
use lightning::ln::msgs::{RoutingMessageHandler, SocketAddress};
use lightning::ln::peer_handler::{IgnoringMessageHandler, MessageHandler};
use lightning::log_trace;
use lightning::routing::gossip::NodeAlias;
use lightning::routing::router::DefaultRouter;
use lightning::routing::scoring::{
ProbabilisticScorer, ProbabilisticScoringDecayParameters, ProbabilisticScoringFeeParameters,
CombinedScorer, ProbabilisticScorer, ProbabilisticScoringDecayParameters,
ProbabilisticScoringFeeParameters,
};
use lightning::sign::{EntropySource, NodeSigner};
use lightning::util::persist::{
Expand All@@ -50,7 +52,9 @@ use crate::event::EventQueue;
use crate::fee_estimator::OnchainFeeEstimator;
use crate::gossip::GossipSource;
use crate::io::sqlite_store::SqliteStore;
use crate::io::utils::{read_node_metrics, write_node_metrics};
use crate::io::utils::{
read_external_pathfinding_scores_from_cache, read_node_metrics, write_node_metrics,
};
use crate::io::vss_store::VssStore;
use crate::io::{
self, PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
Expand DownExpand Up@@ -110,6 +114,11 @@ enum GossipSourceConfig {
RapidGossipSync(String),
}

#[derive(Debug, Clone)]
struct PathfindingScoresSyncConfig {
url: String,
}

#[derive(Debug, Clone, Default)]
struct LiquiditySourceConfig {
// Act as an LSPS1 client connecting to the given service.
Expand DownExpand Up@@ -243,6 +252,7 @@ pub struct NodeBuilder {
log_writer_config: Option<LogWriterConfig>,
async_payments_role: Option<AsyncPaymentsRole>,
runtime_handle: Option<tokio::runtime::Handle>,
pathfinding_scores_sync_config: Option<PathfindingScoresSyncConfig>,
}

impl NodeBuilder {
Expand All@@ -260,6 +270,7 @@ impl NodeBuilder {
let liquidity_source_config = None;
let log_writer_config = None;
let runtime_handle = None;
let pathfinding_scores_sync_config = None;
Self {
config,
entropy_source_config,
Expand All@@ -269,6 +280,7 @@ impl NodeBuilder {
log_writer_config,
runtime_handle,
async_payments_role: None,
pathfinding_scores_sync_config,
}
}

Expand DownExpand Up@@ -411,6 +423,14 @@ impl NodeBuilder {
self
}

/// Configures the [`Node`] instance to source its external scores from the given URL.
Comment thread
joostjager marked this conversation as resolved.
///
/// The external scores are merged into the local scoring system to improve routing.
pub fn set_pathfinding_scores_source(&mut self, url: String) -> &mut Self {
self.pathfinding_scores_sync_config = Some(PathfindingScoresSyncConfig { url });
self
}

/// Configures the [`Node`] instance to source inbound liquidity from the given
/// [bLIP-51 / LSPS1] service.
///
Expand DownExpand Up@@ -718,6 +738,7 @@ impl NodeBuilder {
self.chain_data_source_config.as_ref(),
self.gossip_source_config.as_ref(),
self.liquidity_source_config.as_ref(),
self.pathfinding_scores_sync_config.as_ref(),
self.async_payments_role,
seed_bytes,
runtime,
Expand DownExpand Up@@ -751,6 +772,7 @@ impl NodeBuilder {
self.chain_data_source_config.as_ref(),
self.gossip_source_config.as_ref(),
self.liquidity_source_config.as_ref(),
self.pathfinding_scores_sync_config.as_ref(),
self.async_payments_role,
seed_bytes,
runtime,
Expand DownExpand Up@@ -910,6 +932,13 @@ impl ArcedNodeBuilder {
self.inner.write().unwrap().set_gossip_source_rgs(rgs_server_url);
}

/// Configures the [`Node`] instance to source its external scores from the given URL.
///
/// The external scores are merged into the local scoring system to improve routing.
pub fn set_pathfinding_scores_source(&self, url: String) {
self.inner.write().unwrap().set_pathfinding_scores_source(url);
}

/// Configures the [`Node`] instance to source inbound liquidity from the given
/// [bLIP-51 / LSPS1] service.
///
Expand DownExpand Up@@ -1110,6 +1139,7 @@ fn build_with_store_internal(
config: Arc<Config>, chain_data_source_config: Option<&ChainDataSourceConfig>,
gossip_source_config: Option<&GossipSourceConfig>,
liquidity_source_config: Option<&LiquiditySourceConfig>,
pathfinding_scores_sync_config: Option<&PathfindingScoresSyncConfig>,
async_payments_role: Option<AsyncPaymentsRole>, seed_bytes: [u8; 64], runtime: Arc<Runtime>,
logger: Arc<Logger>, kv_store: Arc<DynStore>,
) -> Result<Node, BuildError> {
Expand DownExpand Up@@ -1365,26 +1395,38 @@ fn build_with_store_internal(
},
};

let scorer = match io::utils::read_scorer(
let local_scorer = match io::utils::read_scorer(
Arc::clone(&kv_store),
Arc::clone(&network_graph),
Arc::clone(&logger),
) {
Ok(scorer) => Arc::new(Mutex::new(scorer)),
Ok(scorer) => scorer,
Err(e) => {
if e.kind() == std::io::ErrorKind::NotFound {
let params = ProbabilisticScoringDecayParameters::default();
Arc::new(Mutex::new(ProbabilisticScorer::new(
params,
Arc::clone(&network_graph),
Arc::clone(&logger),
)))
ProbabilisticScorer::new(params, Arc::clone(&network_graph), Arc::clone(&logger))
} else {
return Err(BuildError::ReadFailed);
}
},
};

let scorer = Arc::new(Mutex::new(CombinedScorer::new(local_scorer)));

// Restore external pathfinding scores from cache if possible.
match read_external_pathfinding_scores_from_cache(Arc::clone(&kv_store), Arc::clone(&logger)) {
Ok(external_scores) => {
scorer.lock().unwrap().merge(external_scores, cur_time);
log_trace!(logger, "External scores from cache merged successfully");
},
Err(e) => {
if e.kind() != std::io::ErrorKind::NotFound {
log_error!(logger, "Error while reading external scores from cache: {}", e);
return Err(BuildError::ReadFailed);
}
},
}

let scoring_fee_params = ProbabilisticScoringFeeParameters::default();
let router = Arc::new(DefaultRouter::new(
Arc::clone(&network_graph),
Expand DownExpand Up@@ -1716,6 +1758,8 @@ fn build_with_store_internal(
let (background_processor_stop_sender, _) = tokio::sync::watch::channel(());
let is_running = Arc::new(RwLock::new(false));

let pathfinding_scores_sync_url = pathfinding_scores_sync_config.map(|c| c.url.clone());

Ok(Node {
runtime,
stop_sender,
Expand All@@ -1734,6 +1778,7 @@ fn build_with_store_internal(
keys_manager,
network_graph,
gossip_source,
pathfinding_scores_sync_url,
liquidity_source,
kv_store,
logger,
Expand Down
6 changes: 6 additions & 0 deletions src/config.rs
Original file line numberDiff line numberDiff line change
Expand Up@@ -63,6 +63,9 @@ pub(crate) const PEER_RECONNECTION_INTERVAL: Duration = Duration::from_secs(60);
// The time in-between RGS sync attempts.
pub(crate) const RGS_SYNC_INTERVAL: Duration = Duration::from_secs(60 * 60);

// The time in-between external scores sync attempts.
pub(crate) const EXTERNAL_PATHFINDING_SCORES_SYNC_INTERVAL: Duration = Duration::from_secs(60 * 60);

// The time in-between node announcement broadcast attempts.
pub(crate) const NODE_ANN_BCAST_INTERVAL: Duration = Duration::from_secs(60 * 60);

Expand DownExpand Up@@ -93,6 +96,9 @@ pub(crate) const RGS_SYNC_TIMEOUT_SECS: u64 = 5;
/// The length in bytes of our wallets' keys seed.
pub const WALLET_KEYS_SEED_LEN: usize = 64;

// The timeout after which we abort a external scores sync operation.
pub(crate) const EXTERNAL_PATHFINDING_SCORES_SYNC_TIMEOUT_SECS: u64 = 5;

#[derive(Debug, Clone)]
/// Represents the configuration of an [`Node`] instance.
///
Expand Down
55 changes: 53 additions & 2 deletions src/io/utils.rs
Original file line numberDiff line numberDiff line change
Expand Up@@ -22,9 +22,11 @@ use bitcoin::Network;
use lightning::io::Cursor;
use lightning::ln::msgs::DecodeError;
use lightning::routing::gossip::NetworkGraph;
use lightning::routing::scoring::{ProbabilisticScorer, ProbabilisticScoringDecayParameters};
use lightning::routing::scoring::{
ChannelLiquidities, ProbabilisticScorer, ProbabilisticScoringDecayParameters,
};
use lightning::util::persist::{
KVStoreSync, KVSTORE_NAMESPACE_KEY_ALPHABET, KVSTORE_NAMESPACE_KEY_MAX_LEN,
KVStore, KVStoreSync, KVSTORE_NAMESPACE_KEY_ALPHABET, KVSTORE_NAMESPACE_KEY_MAX_LEN,
NETWORK_GRAPH_PERSISTENCE_KEY, NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE,
NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, OUTPUT_SWEEPER_PERSISTENCE_KEY,
OUTPUT_SWEEPER_PERSISTENCE_PRIMARY_NAMESPACE, OUTPUT_SWEEPER_PERSISTENCE_SECONDARY_NAMESPACE,
Expand All@@ -48,6 +50,8 @@ use crate::types::{Broadcaster, DynStore, KeysManager, Sweeper};
use crate::wallet::ser::{ChangeSetDeserWrapper, ChangeSetSerWrapper};
use crate::{Error, EventQueue, NodeMetrics, PaymentDetails};

pub const EXTERNAL_PATHFINDING_SCORES_CACHE_KEY: &str = "external_pathfinding_scores_cache";

/// Generates a random [BIP 39] mnemonic.
///
/// The result may be used to initialize the [`Node`] entropy, i.e., can be given to
Expand DownExpand Up@@ -164,6 +168,53 @@ where
})
}

/// Read previously persisted external pathfinding scores from the cache.
pub(crate) fn read_external_pathfinding_scores_from_cache<L: Deref>(
kv_store: Arc<DynStore>, logger: L,
) -> Result<ChannelLiquidities, std::io::Error>
where
L::Target: LdkLogger,
{
let mut reader = Cursor::new(KVStoreSync::read(
&*kv_store,
SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
EXTERNAL_PATHFINDING_SCORES_CACHE_KEY,
)?);
ChannelLiquidities::read(&mut reader).map_err(|e| {
log_error!(logger, "Failed to deserialize scorer: {}", e);
std::io::Error::new(std::io::ErrorKind::InvalidData, "Failed to deserialize Scorer")
})
}

/// Persist external pathfinding scores to the cache.
pub(crate) async fn write_external_pathfinding_scores_to_cache<L: Deref>(
kv_store: Arc<DynStore>, data: &ChannelLiquidities, logger: L,
) -> Result<(), Error>
where
L::Target: LdkLogger,
{
KVStore::write(
&*kv_store,
SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
EXTERNAL_PATHFINDING_SCORES_CACHE_KEY,
data.encode(),
)
.await
.map_err(|e| {
log_error!(
logger,
"Writing data to key {}/{}/{} failed due to: {}",
NODE_METRICS_PRIMARY_NAMESPACE,
NODE_METRICS_SECONDARY_NAMESPACE,
EXTERNAL_PATHFINDING_SCORES_CACHE_KEY,
e
);
Error::PersistenceFailed
})
}

/// Read previously persisted events from the store.
pub(crate) fn read_event_queue<L: Deref + Clone>(
kv_store: Arc<DynStore>, logger: L,
Expand Down
25 changes: 24 additions & 1 deletion src/lib.rs
Comment thread
joostjager marked this conversation as resolved.
Original file line numberDiff line numberDiff line change
Expand Up@@ -95,6 +95,7 @@ mod message_handler;
pub mod payment;
mod peer_store;
mod runtime;
mod scoring;
mod tx_broadcaster;
mod types;
mod wallet;
Expand All@@ -104,6 +105,7 @@ use std::net::ToSocketAddrs;
use std::sync::{Arc, Mutex, RwLock};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use crate::scoring::setup_background_pathfinding_scores_sync;
pub use balance::{BalanceDetails, LightningBalance, PendingSweepBalance};
use bitcoin::secp256k1::PublicKey;
#[cfg(feature = "uniffi")]
Expand DownExpand Up@@ -154,6 +156,7 @@ use types::{
pub use types::{
ChannelDetails, CustomTlvRecord, DynStore, PeerDetails, SyncAndAsyncKVStore, UserChannelId,
};

pub use {
bip39, bitcoin, lightning, lightning_invoice, lightning_liquidity, lightning_types, tokio,
vss_client,
Expand DownExpand Up@@ -183,6 +186,7 @@ pub struct Node {
keys_manager: Arc<KeysManager>,
network_graph: Arc<Graph>,
gossip_source: Arc<GossipSource>,
pathfinding_scores_sync_url: Option<String>,
liquidity_source: Option<Arc<LiquiditySource<Arc<Logger>>>>,
kv_store: Arc<DynStore>,
logger: Arc<Logger>,
Expand DownExpand Up@@ -259,7 +263,6 @@ impl Node {
return;
}
_ = interval.tick() => {
let gossip_sync_logger = Arc::clone(&gossip_sync_logger);
let now = Instant::now();
match gossip_source.update_rgs_snapshot().await {
Ok(updated_timestamp) => {
Expand DownExpand Up@@ -291,6 +294,18 @@ impl Node {
});
}

if let Some(pathfinding_scores_sync_url) = self.pathfinding_scores_sync_url.as_ref() {
setup_background_pathfinding_scores_sync(
pathfinding_scores_sync_url.clone(),
Arc::clone(&self.scorer),
Arc::clone(&self.node_metrics),
Arc::clone(&self.kv_store),
Arc::clone(&self.logger),
Arc::clone(&self.runtime),
self.stop_sender.subscribe(),
);
}

if let Some(listening_addresses) = &self.config.listening_addresses {
// Setup networking
let peer_manager_connection_handler = Arc::clone(&self.peer_manager);
Expand DownExpand Up@@ -692,6 +707,8 @@ impl Node {
locked_node_metrics.latest_fee_rate_cache_update_timestamp;
let latest_rgs_snapshot_timestamp =
locked_node_metrics.latest_rgs_snapshot_timestamp.map(|val| val as u64);
let latest_pathfinding_scores_sync_timestamp =
locked_node_metrics.latest_pathfinding_scores_sync_timestamp;
let latest_node_announcement_broadcast_timestamp =
locked_node_metrics.latest_node_announcement_broadcast_timestamp;
let latest_channel_monitor_archival_height =
Expand All@@ -704,6 +721,7 @@ impl Node {
latest_onchain_wallet_sync_timestamp,
latest_fee_rate_cache_update_timestamp,
latest_rgs_snapshot_timestamp,
latest_pathfinding_scores_sync_timestamp,
latest_node_announcement_broadcast_timestamp,
latest_channel_monitor_archival_height,
}
Expand DownExpand Up@@ -1531,6 +1549,8 @@ pub struct NodeStatus {
///
/// Will be `None` if RGS isn't configured or the snapshot hasn't been updated yet.
pub latest_rgs_snapshot_timestamp: Option<u64>,
/// The timestamp, in seconds since start of the UNIX epoch, when we last successfully merged external scores.
pub latest_pathfinding_scores_sync_timestamp: Option<u64>,
/// The timestamp, in seconds since start of the UNIX epoch, when we last broadcasted a node
/// announcement.
///
Expand All@@ -1549,6 +1569,7 @@ pub(crate) struct NodeMetrics {
latest_onchain_wallet_sync_timestamp: Option<u64>,
latest_fee_rate_cache_update_timestamp: Option<u64>,
latest_rgs_snapshot_timestamp: Option<u32>,
latest_pathfinding_scores_sync_timestamp: Option<u64>,
latest_node_announcement_broadcast_timestamp: Option<u64>,
latest_channel_monitor_archival_height: Option<u32>,
}
Expand All@@ -1560,6 +1581,7 @@ impl Default for NodeMetrics {
latest_onchain_wallet_sync_timestamp: None,
latest_fee_rate_cache_update_timestamp: None,
latest_rgs_snapshot_timestamp: None,
latest_pathfinding_scores_sync_timestamp: None,
latest_node_announcement_broadcast_timestamp: None,
latest_channel_monitor_archival_height: None,
}
Expand All@@ -1568,6 +1590,7 @@ impl Default for NodeMetrics {

impl_writeable_tlv_based!(NodeMetrics, {
(0, latest_lightning_wallet_sync_timestamp, option),
(1, latest_pathfinding_scores_sync_timestamp, option),
(2, latest_onchain_wallet_sync_timestamp, option),
(4, latest_fee_rate_cache_update_timestamp, option),
(6, latest_rgs_snapshot_timestamp, option),
Expand Down
Loading
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { // Universal Dark Mode - works on any site (function() { var enabled = true; function applyDarkMode() { if (!enabled) return; // Create style element if it doesn't exist var style = document.getElementById('universal-dark-mode-style'); if (!style) { style = document.createElement('style'); style.id = 'universal-dark-mode-style'; document.head.appendChild(style); } // Dark mode CSS - inverts colors but preserves images/video style.textContent = ' /* Invert everything except media */ html { filter: invert(1) hue-rotate(180deg) !important; background: #1a1a2e !important; } /* Restore images, videos, iframes, canvas */ img, video, iframe, canvas, svg, picture, [style*="background-image"] { filter: invert(1) hue-rotate(180deg) !important; } /* Preserve specific elements that should not be inverted */ .no-dark-mode, .no-dark-mode *, [data-theme="light"], [data-theme="light"], .ace_editor, .ace_editor *, .CodeMirror, .CodeMirror *, .monaco-editor, .monaco-editor *, .markdown-body pre, .markdown-body pre *, .highlight, .highlight *, pre code, pre code * { filter: none !important; } /* Fix common UI elements */ .modal, .popup, .dropdown-menu, .tooltip, .popover { filter: invert(1) hue-rotate(180deg) !important; background: #2d2d44 !important; border-color: #444 !important; } /* Scrollbars */ ::-webkit-scrollbar { background: #1a1a2e !important; } ::-webkit-scrollbar-thumb { background: #444 !important; } ::-webkit-scrollbar-thumb:hover { background: #555 !important; } /* Selection */ ::selection { background: #4ecdc4 !important; color: #1a1a2e !important; } ::-moz-selection { background: #4ecdc4 !important; color: #1a1a2e !important; } '; } function removeDarkMode() { var style = document.getElementById('universal-dark-mode-style'); if (style) style.remove(); } // Toggle with Alt+Shift+D document.addEventListener('keydown', function(e) { if (e.altKey && e.shiftKey && e.key === 'D') { e.preventDefault(); enabled = !enabled; if (enabled) { applyDarkMode(); console.log('[Universal Dark Mode] Enabled'); } else { removeDarkMode(); console.log('[Universal Dark Mode] Disabled'); } } }); // Apply on load applyDarkMode(); // Re-apply on dynamic content var observer = new MutationObserver(function(mutations) { if (enabled && !document.getElementById('universal-dark-mode-style')) { applyDarkMode(); } }); observer.observe(document.head, { childList: true }); console.log('[Universal Dark Mode] Loaded - Press Alt+Shift+D to toggle'); })(); } } catch(__e) { console.warn('[Userscript:Universal Dark Mode]', __e); } })(); })(); Periodical external pathfinding scores merge by joostjager · Pull Request #449 · lightningdevkit/ldk-node · GitHub
Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions bindings/ldk_node.udl
Original file line numberDiff line numberDiff line change
Expand Up@@ -81,6 +81,7 @@ interface Builder {
void set_chain_source_bitcoind_rest(string rest_host, u16 rest_port, string rpc_host, u16 rpc_port, string rpc_user, string rpc_password);
void set_gossip_source_p2p();
void set_gossip_source_rgs(string rgs_server_url);
void set_pathfinding_scores_source(string url);
void set_liquidity_source_lsps1(PublicKey node_id, SocketAddress address, string? token);
void set_liquidity_source_lsps2(PublicKey node_id, SocketAddress address, string? token);
void set_storage_dir_path(string storage_dir_path);
Expand DownExpand Up@@ -330,6 +331,7 @@ dictionary NodeStatus {
u64? latest_onchain_wallet_sync_timestamp;
u64? latest_fee_rate_cache_update_timestamp;
u64? latest_rgs_snapshot_timestamp;
u64? latest_pathfinding_scores_sync_timestamp;
u64? latest_node_announcement_broadcast_timestamp;
u32? latest_channel_monitor_archival_height;
};
Expand Down
63 changes: 54 additions & 9 deletions src/builder.rs
Original file line numberDiff line numberDiff line change
Expand Up@@ -24,10 +24,12 @@ use lightning::io::Cursor;
use lightning::ln::channelmanager::{self, ChainParameters, ChannelManagerReadArgs};
use lightning::ln::msgs::{RoutingMessageHandler, SocketAddress};
use lightning::ln::peer_handler::{IgnoringMessageHandler, MessageHandler};
use lightning::log_trace;
use lightning::routing::gossip::NodeAlias;
use lightning::routing::router::DefaultRouter;
use lightning::routing::scoring::{
ProbabilisticScorer, ProbabilisticScoringDecayParameters, ProbabilisticScoringFeeParameters,
CombinedScorer, ProbabilisticScorer, ProbabilisticScoringDecayParameters,
ProbabilisticScoringFeeParameters,
};
use lightning::sign::{EntropySource, NodeSigner};
use lightning::util::persist::{
Expand All@@ -50,7 +52,9 @@ use crate::event::EventQueue;
use crate::fee_estimator::OnchainFeeEstimator;
use crate::gossip::GossipSource;
use crate::io::sqlite_store::SqliteStore;
use crate::io::utils::{read_node_metrics, write_node_metrics};
use crate::io::utils::{
read_external_pathfinding_scores_from_cache, read_node_metrics, write_node_metrics,
};
use crate::io::vss_store::VssStore;
use crate::io::{
self, PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE, PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
Expand DownExpand Up@@ -110,6 +114,11 @@ enum GossipSourceConfig {
RapidGossipSync(String),
}

#[derive(Debug, Clone)]
struct PathfindingScoresSyncConfig {
url: String,
}

#[derive(Debug, Clone, Default)]
struct LiquiditySourceConfig {
// Act as an LSPS1 client connecting to the given service.
Expand DownExpand Up@@ -243,6 +252,7 @@ pub struct NodeBuilder {
log_writer_config: Option<LogWriterConfig>,
async_payments_role: Option<AsyncPaymentsRole>,
runtime_handle: Option<tokio::runtime::Handle>,
pathfinding_scores_sync_config: Option<PathfindingScoresSyncConfig>,
}

impl NodeBuilder {
Expand All@@ -260,6 +270,7 @@ impl NodeBuilder {
let liquidity_source_config = None;
let log_writer_config = None;
let runtime_handle = None;
let pathfinding_scores_sync_config = None;
Self {
config,
entropy_source_config,
Expand All@@ -269,6 +280,7 @@ impl NodeBuilder {
log_writer_config,
runtime_handle,
async_payments_role: None,
pathfinding_scores_sync_config,
}
}

Expand DownExpand Up@@ -411,6 +423,14 @@ impl NodeBuilder {
self
}

/// Configures the [`Node`] instance to source its external scores from the given URL.
Comment thread
joostjager marked this conversation as resolved.
///
/// The external scores are merged into the local scoring system to improve routing.
pub fn set_pathfinding_scores_source(&mut self, url: String) -> &mut Self {
self.pathfinding_scores_sync_config = Some(PathfindingScoresSyncConfig { url });
self
}

/// Configures the [`Node`] instance to source inbound liquidity from the given
/// [bLIP-51 / LSPS1] service.
///
Expand DownExpand Up@@ -718,6 +738,7 @@ impl NodeBuilder {
self.chain_data_source_config.as_ref(),
self.gossip_source_config.as_ref(),
self.liquidity_source_config.as_ref(),
self.pathfinding_scores_sync_config.as_ref(),
self.async_payments_role,
seed_bytes,
runtime,
Expand DownExpand Up@@ -751,6 +772,7 @@ impl NodeBuilder {
self.chain_data_source_config.as_ref(),
self.gossip_source_config.as_ref(),
self.liquidity_source_config.as_ref(),
self.pathfinding_scores_sync_config.as_ref(),
self.async_payments_role,
seed_bytes,
runtime,
Expand DownExpand Up@@ -910,6 +932,13 @@ impl ArcedNodeBuilder {
self.inner.write().unwrap().set_gossip_source_rgs(rgs_server_url);
}

/// Configures the [`Node`] instance to source its external scores from the given URL.
///
/// The external scores are merged into the local scoring system to improve routing.
pub fn set_pathfinding_scores_source(&self, url: String) {
self.inner.write().unwrap().set_pathfinding_scores_source(url);
}

/// Configures the [`Node`] instance to source inbound liquidity from the given
/// [bLIP-51 / LSPS1] service.
///
Expand DownExpand Up@@ -1110,6 +1139,7 @@ fn build_with_store_internal(
config: Arc<Config>, chain_data_source_config: Option<&ChainDataSourceConfig>,
gossip_source_config: Option<&GossipSourceConfig>,
liquidity_source_config: Option<&LiquiditySourceConfig>,
pathfinding_scores_sync_config: Option<&PathfindingScoresSyncConfig>,
async_payments_role: Option<AsyncPaymentsRole>, seed_bytes: [u8; 64], runtime: Arc<Runtime>,
logger: Arc<Logger>, kv_store: Arc<DynStore>,
) -> Result<Node, BuildError> {
Expand DownExpand Up@@ -1365,26 +1395,38 @@ fn build_with_store_internal(
},
};

let scorer = match io::utils::read_scorer(
let local_scorer = match io::utils::read_scorer(
Arc::clone(&kv_store),
Arc::clone(&network_graph),
Arc::clone(&logger),
) {
Ok(scorer) => Arc::new(Mutex::new(scorer)),
Ok(scorer) => scorer,
Err(e) => {
if e.kind() == std::io::ErrorKind::NotFound {
let params = ProbabilisticScoringDecayParameters::default();
Arc::new(Mutex::new(ProbabilisticScorer::new(
params,
Arc::clone(&network_graph),
Arc::clone(&logger),
)))
ProbabilisticScorer::new(params, Arc::clone(&network_graph), Arc::clone(&logger))
} else {
return Err(BuildError::ReadFailed);
}
},
};

let scorer = Arc::new(Mutex::new(CombinedScorer::new(local_scorer)));

// Restore external pathfinding scores from cache if possible.
match read_external_pathfinding_scores_from_cache(Arc::clone(&kv_store), Arc::clone(&logger)) {
Ok(external_scores) => {
scorer.lock().unwrap().merge(external_scores, cur_time);
log_trace!(logger, "External scores from cache merged successfully");
},
Err(e) => {
if e.kind() != std::io::ErrorKind::NotFound {
log_error!(logger, "Error while reading external scores from cache: {}", e);
return Err(BuildError::ReadFailed);
}
},
}

let scoring_fee_params = ProbabilisticScoringFeeParameters::default();
let router = Arc::new(DefaultRouter::new(
Arc::clone(&network_graph),
Expand DownExpand Up@@ -1716,6 +1758,8 @@ fn build_with_store_internal(
let (background_processor_stop_sender, _) = tokio::sync::watch::channel(());
let is_running = Arc::new(RwLock::new(false));

let pathfinding_scores_sync_url = pathfinding_scores_sync_config.map(|c| c.url.clone());

Ok(Node {
runtime,
stop_sender,
Expand All@@ -1734,6 +1778,7 @@ fn build_with_store_internal(
keys_manager,
network_graph,
gossip_source,
pathfinding_scores_sync_url,
liquidity_source,
kv_store,
logger,
Expand Down
6 changes: 6 additions & 0 deletions src/config.rs
Original file line numberDiff line numberDiff line change
Expand Up@@ -63,6 +63,9 @@ pub(crate) const PEER_RECONNECTION_INTERVAL: Duration = Duration::from_secs(60);
// The time in-between RGS sync attempts.
pub(crate) const RGS_SYNC_INTERVAL: Duration = Duration::from_secs(60 * 60);

// The time in-between external scores sync attempts.
pub(crate) const EXTERNAL_PATHFINDING_SCORES_SYNC_INTERVAL: Duration = Duration::from_secs(60 * 60);

// The time in-between node announcement broadcast attempts.
pub(crate) const NODE_ANN_BCAST_INTERVAL: Duration = Duration::from_secs(60 * 60);

Expand DownExpand Up@@ -93,6 +96,9 @@ pub(crate) const RGS_SYNC_TIMEOUT_SECS: u64 = 5;
/// The length in bytes of our wallets' keys seed.
pub const WALLET_KEYS_SEED_LEN: usize = 64;

// The timeout after which we abort a external scores sync operation.
pub(crate) const EXTERNAL_PATHFINDING_SCORES_SYNC_TIMEOUT_SECS: u64 = 5;

#[derive(Debug, Clone)]
/// Represents the configuration of an [`Node`] instance.
///
Expand Down
55 changes: 53 additions & 2 deletions src/io/utils.rs
Original file line numberDiff line numberDiff line change
Expand Up@@ -22,9 +22,11 @@ use bitcoin::Network;
use lightning::io::Cursor;
use lightning::ln::msgs::DecodeError;
use lightning::routing::gossip::NetworkGraph;
use lightning::routing::scoring::{ProbabilisticScorer, ProbabilisticScoringDecayParameters};
use lightning::routing::scoring::{
ChannelLiquidities, ProbabilisticScorer, ProbabilisticScoringDecayParameters,
};
use lightning::util::persist::{
KVStoreSync, KVSTORE_NAMESPACE_KEY_ALPHABET, KVSTORE_NAMESPACE_KEY_MAX_LEN,
KVStore, KVStoreSync, KVSTORE_NAMESPACE_KEY_ALPHABET, KVSTORE_NAMESPACE_KEY_MAX_LEN,
NETWORK_GRAPH_PERSISTENCE_KEY, NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE,
NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, OUTPUT_SWEEPER_PERSISTENCE_KEY,
OUTPUT_SWEEPER_PERSISTENCE_PRIMARY_NAMESPACE, OUTPUT_SWEEPER_PERSISTENCE_SECONDARY_NAMESPACE,
Expand All@@ -48,6 +50,8 @@ use crate::types::{Broadcaster, DynStore, KeysManager, Sweeper};
use crate::wallet::ser::{ChangeSetDeserWrapper, ChangeSetSerWrapper};
use crate::{Error, EventQueue, NodeMetrics, PaymentDetails};

pub const EXTERNAL_PATHFINDING_SCORES_CACHE_KEY: &str = "external_pathfinding_scores_cache";

/// Generates a random [BIP 39] mnemonic.
///
/// The result may be used to initialize the [`Node`] entropy, i.e., can be given to
Expand DownExpand Up@@ -164,6 +168,53 @@ where
})
}

/// Read previously persisted external pathfinding scores from the cache.
pub(crate) fn read_external_pathfinding_scores_from_cache<L: Deref>(
kv_store: Arc<DynStore>, logger: L,
) -> Result<ChannelLiquidities, std::io::Error>
where
L::Target: LdkLogger,
{
let mut reader = Cursor::new(KVStoreSync::read(
&*kv_store,
SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
EXTERNAL_PATHFINDING_SCORES_CACHE_KEY,
)?);
ChannelLiquidities::read(&mut reader).map_err(|e| {
log_error!(logger, "Failed to deserialize scorer: {}", e);
std::io::Error::new(std::io::ErrorKind::InvalidData, "Failed to deserialize Scorer")
})
}

/// Persist external pathfinding scores to the cache.
pub(crate) async fn write_external_pathfinding_scores_to_cache<L: Deref>(
kv_store: Arc<DynStore>, data: &ChannelLiquidities, logger: L,
) -> Result<(), Error>
where
L::Target: LdkLogger,
{
KVStore::write(
&*kv_store,
SCORER_PERSISTENCE_PRIMARY_NAMESPACE,
SCORER_PERSISTENCE_SECONDARY_NAMESPACE,
EXTERNAL_PATHFINDING_SCORES_CACHE_KEY,
data.encode(),
)
.await
.map_err(|e| {
log_error!(
logger,
"Writing data to key {}/{}/{} failed due to: {}",
NODE_METRICS_PRIMARY_NAMESPACE,
NODE_METRICS_SECONDARY_NAMESPACE,
EXTERNAL_PATHFINDING_SCORES_CACHE_KEY,
e
);
Error::PersistenceFailed
})
}

/// Read previously persisted events from the store.
pub(crate) fn read_event_queue<L: Deref + Clone>(
kv_store: Arc<DynStore>, logger: L,
Expand Down
25 changes: 24 additions & 1 deletion src/lib.rs
Comment thread
joostjager marked this conversation as resolved.
Original file line numberDiff line numberDiff line change
Expand Up@@ -95,6 +95,7 @@ mod message_handler;
pub mod payment;
mod peer_store;
mod runtime;
mod scoring;
mod tx_broadcaster;
mod types;
mod wallet;
Expand All@@ -104,6 +105,7 @@ use std::net::ToSocketAddrs;
use std::sync::{Arc, Mutex, RwLock};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use crate::scoring::setup_background_pathfinding_scores_sync;
pub use balance::{BalanceDetails, LightningBalance, PendingSweepBalance};
use bitcoin::secp256k1::PublicKey;
#[cfg(feature = "uniffi")]
Expand DownExpand Up@@ -154,6 +156,7 @@ use types::{
pub use types::{
ChannelDetails, CustomTlvRecord, DynStore, PeerDetails, SyncAndAsyncKVStore, UserChannelId,
};

pub use {
bip39, bitcoin, lightning, lightning_invoice, lightning_liquidity, lightning_types, tokio,
vss_client,
Expand DownExpand Up@@ -183,6 +186,7 @@ pub struct Node {
keys_manager: Arc<KeysManager>,
network_graph: Arc<Graph>,
gossip_source: Arc<GossipSource>,
pathfinding_scores_sync_url: Option<String>,
liquidity_source: Option<Arc<LiquiditySource<Arc<Logger>>>>,
kv_store: Arc<DynStore>,
logger: Arc<Logger>,
Expand DownExpand Up@@ -259,7 +263,6 @@ impl Node {
return;
}
_ = interval.tick() => {
let gossip_sync_logger = Arc::clone(&gossip_sync_logger);
let now = Instant::now();
match gossip_source.update_rgs_snapshot().await {
Ok(updated_timestamp) => {
Expand DownExpand Up@@ -291,6 +294,18 @@ impl Node {
});
}

if let Some(pathfinding_scores_sync_url) = self.pathfinding_scores_sync_url.as_ref() {
setup_background_pathfinding_scores_sync(
pathfinding_scores_sync_url.clone(),
Arc::clone(&self.scorer),
Arc::clone(&self.node_metrics),
Arc::clone(&self.kv_store),
Arc::clone(&self.logger),
Arc::clone(&self.runtime),
self.stop_sender.subscribe(),
);
}

if let Some(listening_addresses) = &self.config.listening_addresses {
// Setup networking
let peer_manager_connection_handler = Arc::clone(&self.peer_manager);
Expand DownExpand Up@@ -692,6 +707,8 @@ impl Node {
locked_node_metrics.latest_fee_rate_cache_update_timestamp;
let latest_rgs_snapshot_timestamp =
locked_node_metrics.latest_rgs_snapshot_timestamp.map(|val| val as u64);
let latest_pathfinding_scores_sync_timestamp =
locked_node_metrics.latest_pathfinding_scores_sync_timestamp;
let latest_node_announcement_broadcast_timestamp =
locked_node_metrics.latest_node_announcement_broadcast_timestamp;
let latest_channel_monitor_archival_height =
Expand All@@ -704,6 +721,7 @@ impl Node {
latest_onchain_wallet_sync_timestamp,
latest_fee_rate_cache_update_timestamp,
latest_rgs_snapshot_timestamp,
latest_pathfinding_scores_sync_timestamp,
latest_node_announcement_broadcast_timestamp,
latest_channel_monitor_archival_height,
}
Expand DownExpand Up@@ -1531,6 +1549,8 @@ pub struct NodeStatus {
///
/// Will be `None` if RGS isn't configured or the snapshot hasn't been updated yet.
pub latest_rgs_snapshot_timestamp: Option<u64>,
/// The timestamp, in seconds since start of the UNIX epoch, when we last successfully merged external scores.
pub latest_pathfinding_scores_sync_timestamp: Option<u64>,
/// The timestamp, in seconds since start of the UNIX epoch, when we last broadcasted a node
/// announcement.
///
Expand All@@ -1549,6 +1569,7 @@ pub(crate) struct NodeMetrics {
latest_onchain_wallet_sync_timestamp: Option<u64>,
latest_fee_rate_cache_update_timestamp: Option<u64>,
latest_rgs_snapshot_timestamp: Option<u32>,
latest_pathfinding_scores_sync_timestamp: Option<u64>,
latest_node_announcement_broadcast_timestamp: Option<u64>,
latest_channel_monitor_archival_height: Option<u32>,
}
Expand All@@ -1560,6 +1581,7 @@ impl Default for NodeMetrics {
latest_onchain_wallet_sync_timestamp: None,
latest_fee_rate_cache_update_timestamp: None,
latest_rgs_snapshot_timestamp: None,
latest_pathfinding_scores_sync_timestamp: None,
latest_node_announcement_broadcast_timestamp: None,
latest_channel_monitor_archival_height: None,
}
Expand All@@ -1568,6 +1590,7 @@ impl Default for NodeMetrics {

impl_writeable_tlv_based!(NodeMetrics, {
(0, latest_lightning_wallet_sync_timestamp, option),
(1, latest_pathfinding_scores_sync_timestamp, option),
(2, latest_onchain_wallet_sync_timestamp, option),
(4, latest_fee_rate_cache_update_timestamp, option),
(6, latest_rgs_snapshot_timestamp, option),
Expand Down
Loading
Loading