Skip to content

Parallelize read_payments #7

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
amackillop merged 3 commits into from
Jan 22, 2026
Merged

Parallelize read_payments #7

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
395bb4b
Use async `KVStore` for `read_X` util methods
tnull Jan 5, 2026
4a0acc2
Parallelize `read_payments`
tnull Jan 5, 2026
80a170f
Add test for payment persistence after node restart
tnull Jan 7, 2026
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 Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -92,6 +92,8 @@ log = { version = "0.4.22", default-features = false, features = ["std"]}

vss-client = { package = "vss-client-ng", version = "0.4" }
prost = { version = "0.11.6", default-features = false}
#bitcoin-payment-instructions = { version = "0.6" }
bitcoin-payment-instructions = { git = "https://github.com/tnull/bitcoin-payment-instructions", branch = "2025-12-ldk-node-base" }

[target.'cfg(windows)'.dependencies]
winapi = { version = "0.3", features = ["winbase"] }
Expand Down
78 changes: 46 additions & 32 deletions src/builder.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,7 @@ use bip39::Mnemonic;
use bitcoin::bip32::{ChildNumber, Xpriv};
use bitcoin::secp256k1::PublicKey;
use bitcoin::{BlockHash, Network};
use bdk_chain::{BlockId, TxUpdate};
use bitcoin_payment_instructions::onion_message_resolver::LDKOnionMessageDNSSECHrnResolver;
use lightning::chain::{chainmonitor, BestBlock, Watch};
use lightning::io::Cursor;
use lightning::ln::channelmanager::{self, ChainParameters, ChannelManagerReadArgs};
Expand Down Expand Up @@ -55,7 +55,9 @@ use crate::fee_estimator::OnchainFeeEstimator;
use crate::gossip::GossipSource;
use crate::io::sqlite_store::SqliteStore;
use crate::io::utils::{
read_external_pathfinding_scores_from_cache, read_node_metrics, write_node_metrics,
read_event_queue, read_external_pathfinding_scores_from_cache, read_network_graph,
read_node_metrics, read_output_sweeper, read_payments, read_peer_info, read_scorer,
write_node_metrics,
};
use crate::io::vss_store::VssStore;
use crate::io::{
Expand Down Expand Up @@ -1233,7 +1235,9 @@ fn build_with_store_internal(
}

// Initialize the status fields.
let node_metrics = match read_node_metrics(Arc::clone(&kv_store), Arc::clone(&logger)) {
let node_metrics = match runtime
.block_on(async { read_node_metrics(Arc::clone(&kv_store), Arc::clone(&logger)).await })
{
Ok(metrics) => Arc::new(RwLock::new(metrics)),
Err(e) => {
if e.kind() == std::io::ErrorKind::NotFound {
Expand All @@ -1247,7 +1251,9 @@ fn build_with_store_internal(
let tx_broadcaster = Arc::new(TransactionBroadcaster::new(Arc::clone(&logger)));
let fee_estimator = Arc::new(OnchainFeeEstimator::new());

let payment_store = match io::utils::read_payments(Arc::clone(&kv_store), Arc::clone(&logger)) {
let payment_store = match runtime
.block_on(async { read_payments(Arc::clone(&kv_store), Arc::clone(&logger)).await })
{
Ok(payments) => Arc::new(PaymentStore::new(
payments,
PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE.to_string(),
Expand Down Expand Up @@ -1474,24 +1480,23 @@ fn build_with_store_internal(
));

// Initialize the network graph, scorer, and router
let network_graph =
match io::utils::read_network_graph(Arc::clone(&kv_store), Arc::clone(&logger)) {
Ok(graph) => Arc::new(graph),
Err(e) => {
if e.kind() == std::io::ErrorKind::NotFound {
Arc::new(Graph::new(config.network.into(), Arc::clone(&logger)))
} else {
log_error!(logger, "Failed to read network graph from store: {}", e);
return Err(BuildError::ReadFailed);
}
},
};
let network_graph = match runtime
.block_on(async { read_network_graph(Arc::clone(&kv_store), Arc::clone(&logger)).await })
{
Ok(graph) => Arc::new(graph),
Err(e) => {
if e.kind() == std::io::ErrorKind::NotFound {
Arc::new(Graph::new(config.network.into(), Arc::clone(&logger)))
} else {
log_error!(logger, "Failed to read network graph from store: {}", e);
return Err(BuildError::ReadFailed);
}
},
};

let local_scorer = match io::utils::read_scorer(
Arc::clone(&kv_store),
Arc::clone(&network_graph),
Arc::clone(&logger),
) {
let local_scorer = match runtime.block_on(async {
read_scorer(Arc::clone(&kv_store), Arc::clone(&network_graph), Arc::clone(&logger)).await
}) {
Ok(scorer) => scorer,
Err(e) => {
if e.kind() == std::io::ErrorKind::NotFound {
Expand All @@ -1507,7 +1512,10 @@ fn build_with_store_internal(
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)) {
match runtime.block_on(async {
read_external_pathfinding_scores_from_cache(Arc::clone(&kv_store), Arc::clone(&logger))
.await
}) {
Ok(external_scores) => {
scorer.lock().unwrap().merge(external_scores, cur_time);
log_trace!(logger, "External scores from cache merged successfully");
Expand Down Expand Up @@ -1709,7 +1717,8 @@ fn build_with_store_internal(
},
};

let event_queue = match io::utils::read_event_queue(Arc::clone(&kv_store), Arc::clone(&logger))
let event_queue = match runtime
.block_on(async { read_event_queue(Arc::clone(&kv_store), Arc::clone(&logger)).await })
{
Ok(event_queue) => Arc::new(event_queue),
Err(e) => {
Expand Down Expand Up @@ -1831,14 +1840,17 @@ fn build_with_store_internal(
let connection_manager =
Arc::new(ConnectionManager::new(Arc::clone(&peer_manager), Arc::clone(&logger)));

let output_sweeper = match io::utils::read_output_sweeper(
Arc::clone(&tx_broadcaster),
Arc::clone(&fee_estimator),
Arc::clone(&chain_source),
Arc::clone(&keys_manager),
Arc::clone(&kv_store),
Arc::clone(&logger),
) {
let output_sweeper = match runtime.block_on(async {
read_output_sweeper(
Arc::clone(&tx_broadcaster),
Arc::clone(&fee_estimator),
Arc::clone(&chain_source),
Arc::clone(&keys_manager),
Arc::clone(&kv_store),
Arc::clone(&logger),
)
.await
}) {
Ok(output_sweeper) => Arc::new(output_sweeper),
Err(e) => {
if e.kind() == std::io::ErrorKind::NotFound {
Expand All @@ -1859,7 +1871,9 @@ fn build_with_store_internal(
},
};

let peer_store = match io::utils::read_peer_info(Arc::clone(&kv_store), Arc::clone(&logger)) {
let peer_store = match runtime
.block_on(async { read_peer_info(Arc::clone(&kv_store), Arc::clone(&logger)).await })
{
Ok(peer_store) => Arc::new(peer_store),
Err(e) => {
if e.kind() == std::io::ErrorKind::NotFound {
Expand Down
3 changes: 2 additions & 1 deletion src/ffi/types.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@ use lightning::offers::invoice::Bolt12Invoice as LdkBolt12Invoice;
pub use lightning::offers::offer::OfferId;
use lightning::offers::offer::{Amount as LdkAmount, Offer as LdkOffer};
use lightning::offers::refund::Refund as LdkRefund;
use lightning::onion_message::dns_resolution::HumanReadableName as LdkHumanReadableName;
pub use lightning::routing::gossip::{NodeAlias, NodeId, RoutingFees};
pub use lightning::routing::router::RouteParametersConfig;
use lightning::util::ser::Writeable;
Expand All @@ -54,7 +55,7 @@ pub use crate::logger::{LogLevel, LogRecord, LogWriter};
pub use crate::payment::store::{
ConfirmationStatus, LSPFeeLimits, PaymentDirection, PaymentKind, PaymentStatus,
};
pub use crate::payment::QrPaymentResult;
pub use crate::payment::UnifiedPaymentResult;
use crate::{hex_utils, SocketAddress, UniffiCustomTypeConverter, UserChannelId};

impl UniffiCustomTypeConverter for PublicKey {
Expand Down
Loading
Loading