X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=lightning-background-processor%2Fsrc%2Flib.rs;h=401c3c19ec2200cf9fd101cbd1fdb2f5b3dcce5b;hb=82e0df5e4dcc41c2d32ded6ff433727b15570ea4;hp=bc42c6eb68be4967479289b637b9722986a03d1e;hpb=3a643df99797ee2dd5cc19a6f9d090212b1c7963;p=rust-lightning diff --git a/lightning-background-processor/src/lib.rs b/lightning-background-processor/src/lib.rs index bc42c6eb..401c3c19 100644 --- a/lightning-background-processor/src/lib.rs +++ b/lightning-background-processor/src/lib.rs @@ -25,13 +25,12 @@ extern crate lightning_rapid_gossip_sync; use lightning::chain; use lightning::chain::chaininterface::{BroadcasterInterface, FeeEstimator}; use lightning::chain::chainmonitor::{ChainMonitor, Persist}; -use lightning::chain::keysinterface::{EntropySource, NodeSigner, SignerProvider}; +use lightning::sign::{EntropySource, NodeSigner, SignerProvider}; use lightning::events::{Event, PathFailure}; #[cfg(feature = "std")] use lightning::events::{EventHandler, EventsProvider}; use lightning::ln::channelmanager::ChannelManager; -use lightning::ln::msgs::{ChannelMessageHandler, OnionMessageHandler, RoutingMessageHandler}; -use lightning::ln::peer_handler::{CustomMessageHandler, PeerManager, SocketDescriptor}; +use lightning::ln::peer_handler::APeerManager; use lightning::routing::gossip::{NetworkGraph, P2PGossipSync}; use lightning::routing::utxo::UtxoLookup; use lightning::routing::router::Router; @@ -81,6 +80,8 @@ use alloc::vec::Vec; /// /// [`ChannelMonitor`]: lightning::chain::channelmonitor::ChannelMonitor /// [`Event`]: lightning::events::Event +/// [`PeerManager::timer_tick_occurred`]: lightning::ln::peer_handler::PeerManager::timer_tick_occurred +/// [`PeerManager::process_events`]: lightning::ln::peer_handler::PeerManager::process_events #[cfg(feature = "std")] #[must_use = "BackgroundProcessor will immediately stop on drop. It should be stored until shutdown."] pub struct BackgroundProcessor { @@ -107,7 +108,7 @@ const PING_TIMER: u64 = 1; const NETWORK_PRUNE_TIMER: u64 = 60 * 60; #[cfg(not(test))] -const SCORER_PERSIST_TIMER: u64 = 30; +const SCORER_PERSIST_TIMER: u64 = 60 * 60; #[cfg(test)] const SCORER_PERSIST_TIMER: u64 = 1; @@ -235,9 +236,11 @@ fn handle_network_graph_update( } } +/// Updates scorer based on event and returns whether an update occurred so we can decide whether +/// to persist. fn update_scorer<'a, S: 'static + Deref + Send + Sync, SC: 'a + WriteableScore<'a>>( scorer: &'a S, event: &Event -) { +) -> bool { let mut score = scorer.lock(); match event { Event::PaymentPathFailed { ref path, short_channel_id: Some(scid), .. } => { @@ -257,8 +260,9 @@ fn update_scorer<'a, S: 'static + Deref + Send + Sync, SC: 'a + Wri Event::ProbeFailed { path, short_channel_id: Some(scid), .. } => { score.probe_failed(path, *scid); }, - _ => {}, + _ => return false, } + true } macro_rules! define_run_body { @@ -295,7 +299,7 @@ macro_rules! define_run_body { // ChannelManager, we want to minimize methods blocking on a ChannelManager // generally, and as a fallback place such blocking only immediately before // persistence. - $peer_manager.process_events(); + $peer_manager.as_ref().process_events(); // Exit the loop if the background processor was requested to stop. if $loop_exit_check { @@ -340,20 +344,26 @@ macro_rules! define_run_body { // more than a handful of seconds to complete, and shouldn't disconnect all our // peers. log_trace!($logger, "100ms sleep took more than a second, disconnecting peers."); - $peer_manager.disconnect_all_peers(); + $peer_manager.as_ref().disconnect_all_peers(); last_ping_call = $get_timer(PING_TIMER); } else if $timer_elapsed(&mut last_ping_call, PING_TIMER) { log_trace!($logger, "Calling PeerManager's timer_tick_occurred"); - $peer_manager.timer_tick_occurred(); + $peer_manager.as_ref().timer_tick_occurred(); last_ping_call = $get_timer(PING_TIMER); } // Note that we want to run a graph prune once not long after startup before // falling back to our usual hourly prunes. This avoids short-lived clients never // pruning their network graph. We run once 60 seconds after startup before - // continuing our normal cadence. + // continuing our normal cadence. For RGS, since 60 seconds is likely too long, + // we prune after an initial sync completes. let prune_timer = if have_pruned { NETWORK_PRUNE_TIMER } else { FIRST_NETWORK_PRUNE_TIMER }; - if $timer_elapsed(&mut last_prune_call, prune_timer) { + let prune_timer_elapsed = $timer_elapsed(&mut last_prune_call, prune_timer); + let should_prune = match $gossip_sync { + GossipSync::Rapid(_) => !have_pruned || prune_timer_elapsed, + _ => prune_timer_elapsed, + }; + if should_prune { // The network graph must not be pruned while rapid sync completion is pending if let Some(network_graph) = $gossip_sync.prunable_network_graph() { #[cfg(feature = "std")] { @@ -505,12 +515,13 @@ use core::task; /// # use lightning_background_processor::{process_events_async, GossipSync}; /// # type MyBroadcaster = dyn lightning::chain::chaininterface::BroadcasterInterface + Send + Sync; /// # type MyFeeEstimator = dyn lightning::chain::chaininterface::FeeEstimator + Send + Sync; -/// # type MyNodeSigner = dyn lightning::chain::keysinterface::NodeSigner + Send + Sync; +/// # type MyNodeSigner = dyn lightning::sign::NodeSigner + Send + Sync; /// # type MyUtxoLookup = dyn lightning::routing::utxo::UtxoLookup + Send + Sync; /// # type MyFilter = dyn lightning::chain::Filter + Send + Sync; /// # type MyLogger = dyn lightning::util::logger::Logger + Send + Sync; -/// # type MyChainMonitor = lightning::chain::chainmonitor::ChainMonitor, Arc, Arc, Arc, Arc>; -/// # type MyPeerManager = lightning::ln::peer_handler::SimpleArcPeerManager; +/// # type MyMessageRouter = dyn lightning::onion_message::MessageRouter + Send + Sync; +/// # type MyChainMonitor = lightning::chain::chainmonitor::ChainMonitor, Arc, Arc, Arc, Arc>; +/// # type MyPeerManager = lightning::ln::peer_handler::SimpleArcPeerManager; /// # type MyNetworkGraph = lightning::routing::gossip::NetworkGraph>; /// # type MyGossipSync = lightning::routing::gossip::P2PGossipSync, Arc, Arc>; /// # type MyChannelManager = lightning::ln::channelmanager::SimpleArcChannelManager; @@ -578,10 +589,6 @@ pub async fn process_events_async< G: 'static + Deref> + Send + Sync, L: 'static + Deref + Send + Sync, P: 'static + Deref + Send + Sync, - Descriptor: 'static + SocketDescriptor + Send + Sync, - CMH: 'static + Deref + Send + Sync, - RMH: 'static + Deref + Send + Sync, - OMH: 'static + Deref + Send + Sync, EventHandlerFuture: core::future::Future, EventHandler: Fn(Event) -> EventHandlerFuture, PS: 'static + Deref + Send, @@ -589,8 +596,8 @@ pub async fn process_events_async< CM: 'static + Deref> + Send + Sync, PGS: 'static + Deref> + Send + Sync, RGS: 'static + Deref> + Send, - UMH: 'static + Deref + Send + Sync, - PM: 'static + Deref> + Send + Sync, + APM: APeerManager + Send + Sync, + PM: 'static + Deref + Send + Sync, S: 'static + Deref + Send + Sync, SC: for<'b> WriteableScore<'b>, SleepFuture: core::future::Future + core::marker::Unpin, @@ -612,10 +619,6 @@ where R::Target: 'static + Router, L::Target: 'static + Logger, P::Target: 'static + Persist<::Signer>, - CMH::Target: 'static + ChannelMessageHandler, - OMH::Target: 'static + OnionMessageHandler, - RMH::Target: 'static + RoutingMessageHandler, - UMH::Target: 'static + CustomMessageHandler, PS::Target: 'static + Persister<'a, CW, T, ES, NS, SP, F, R, L, SC>, { let mut should_break = false; @@ -623,12 +626,19 @@ where let network_graph = gossip_sync.network_graph(); let event_handler = &event_handler; let scorer = &scorer; + let logger = &logger; + let persister = &persister; async move { if let Some(network_graph) = network_graph { handle_network_graph_update(network_graph, &event) } if let Some(ref scorer) = scorer { - update_scorer(scorer, &event); + if update_scorer(scorer, &event) { + log_trace!(logger, "Persisting scorer after update"); + if let Err(e) = persister.persist_scorer(&scorer) { + log_error!(logger, "Error: Failed to persist scorer, check your disk and permissions {}", e) + } + } } event_handler(event).await; } @@ -721,18 +731,14 @@ impl BackgroundProcessor { G: 'static + Deref> + Send + Sync, L: 'static + Deref + Send + Sync, P: 'static + Deref + Send + Sync, - Descriptor: 'static + SocketDescriptor + Send + Sync, - CMH: 'static + Deref + Send + Sync, - OMH: 'static + Deref + Send + Sync, - RMH: 'static + Deref + Send + Sync, EH: 'static + EventHandler + Send, PS: 'static + Deref + Send, M: 'static + Deref::Signer, CF, T, F, L, P>> + Send + Sync, CM: 'static + Deref> + Send + Sync, PGS: 'static + Deref> + Send + Sync, RGS: 'static + Deref> + Send, - UMH: 'static + Deref + Send + Sync, - PM: 'static + Deref> + Send + Sync, + APM: APeerManager + Send + Sync, + PM: 'static + Deref + Send + Sync, S: 'static + Deref + Send + Sync, SC: for <'b> WriteableScore<'b>, >( @@ -751,10 +757,6 @@ impl BackgroundProcessor { R::Target: 'static + Router, L::Target: 'static + Logger, P::Target: 'static + Persist<::Signer>, - CMH::Target: 'static + ChannelMessageHandler, - OMH::Target: 'static + OnionMessageHandler, - RMH::Target: 'static + RoutingMessageHandler, - UMH::Target: 'static + CustomMessageHandler, PS::Target: 'static + Persister<'a, CW, T, ES, NS, SP, F, R, L, SC>, { let stop_thread = Arc::new(AtomicBool::new(false)); @@ -766,7 +768,12 @@ impl BackgroundProcessor { handle_network_graph_update(network_graph, &event) } if let Some(ref scorer) = scorer { - update_scorer(scorer, &event); + if update_scorer(scorer, &event) { + log_trace!(logger, "Persisting scorer after update"); + if let Err(e) = persister.persist_scorer(&scorer) { + log_error!(logger, "Error: Failed to persist scorer, check your disk and permissions {}", e) + } + } } event_handler.handle_event(event); }; @@ -832,15 +839,14 @@ impl Drop for BackgroundProcessor { #[cfg(all(feature = "std", test))] mod tests { - use bitcoin::blockdata::block::BlockHeader; - use bitcoin::blockdata::constants::genesis_block; + use bitcoin::blockdata::constants::{genesis_block, ChainHash}; use bitcoin::blockdata::locktime::PackedLockTime; use bitcoin::blockdata::transaction::{Transaction, TxOut}; use bitcoin::network::constants::Network; use bitcoin::secp256k1::{SecretKey, PublicKey, Secp256k1}; use lightning::chain::{BestBlock, Confirm, chainmonitor}; use lightning::chain::channelmonitor::ANTI_REORG_DELAY; - use lightning::chain::keysinterface::{InMemorySigner, KeysManager}; + use lightning::sign::{InMemorySigner, KeysManager}; use lightning::chain::transaction::OutPoint; use lightning::events::{Event, PathFailure, MessageSendEventsProvider, MessageSendEvent}; use lightning::{get_event_msg, get_event}; @@ -848,6 +854,7 @@ mod tests { use lightning::ln::channelmanager; use lightning::ln::channelmanager::{BREAKDOWN_TIMEOUT, ChainParameters, MIN_CLTV_EXPIRY_DELTA, PaymentId}; use lightning::ln::features::{ChannelFeatures, NodeFeatures}; + use lightning::ln::functional_test_utils::*; use lightning::ln::msgs::{ChannelMessageHandler, Init}; use lightning::ln::peer_handler::{PeerManager, MessageHandler, SocketDescriptor, IgnoringMessageHandler}; use lightning::routing::gossip::{NetworkGraph, NodeId, P2PGossipSync}; @@ -864,8 +871,6 @@ mod tests { use std::sync::{Arc, Mutex}; use std::sync::mpsc::SyncSender; use std::time::Duration; - use bitcoin::hashes::Hash; - use bitcoin::TxMerkleNode; use lightning_rapid_gossip_sync::RapidGossipSync; use super::{BackgroundProcessor, GossipSync, FRESHNESS_TIMER}; @@ -881,7 +886,7 @@ mod tests { fn disconnect_socket(&mut self) {} } - type ChannelManager = channelmanager::ChannelManager, Arc, Arc, Arc, Arc, Arc, Arc>>, Arc, Arc>>>, Arc>; + type ChannelManager = channelmanager::ChannelManager, Arc, Arc, Arc, Arc, Arc, Arc>>, Arc, Arc>, (), TestScorer>>, Arc>; type ChainMonitor = chainmonitor::ChainMonitor, Arc, Arc, Arc, Arc>; @@ -1015,8 +1020,9 @@ mod tests { } impl Score for TestScorer { + type ScoreParams = (); fn channel_penalty_msat( - &self, _short_channel_id: u64, _source: &NodeId, _target: &NodeId, _usage: ChannelUsage + &self, _short_channel_id: u64, _source: &NodeId, _target: &NodeId, _usage: ChannelUsage, _score_params: &Self::ScoreParams ) -> u64 { unimplemented!(); } fn payment_path_failed(&mut self, actual_path: &Path, actual_short_channel_id: u64) { @@ -1119,7 +1125,7 @@ mod tests { fn create_nodes(num_nodes: usize, persist_dir: &str) -> (String, Vec) { let persist_temp_path = env::temp_dir().join(persist_dir); let persist_dir = persist_temp_path.to_string_lossy().to_string(); - let network = Network::Testnet; + let network = Network::Bitcoin; let mut nodes = Vec::new(); for i in 0..num_nodes { let tx_broadcaster = Arc::new(test_utils::TestBroadcaster::new(network)); @@ -1129,27 +1135,35 @@ mod tests { let network_graph = Arc::new(NetworkGraph::new(network, logger.clone())); let scorer = Arc::new(Mutex::new(TestScorer::new())); let seed = [i as u8; 32]; - let router = Arc::new(DefaultRouter::new(network_graph.clone(), logger.clone(), seed, scorer.clone())); - let chain_source = Arc::new(test_utils::TestChainSource::new(Network::Testnet)); + let router = Arc::new(DefaultRouter::new(network_graph.clone(), logger.clone(), seed, scorer.clone(), ())); + let chain_source = Arc::new(test_utils::TestChainSource::new(Network::Bitcoin)); let persister = Arc::new(FilesystemPersister::new(format!("{}_persister_{}", &persist_dir, i))); let now = Duration::from_secs(genesis_block.header.time as u64); let keys_manager = Arc::new(KeysManager::new(&seed, now.as_secs(), now.subsec_nanos())); let chain_monitor = Arc::new(chainmonitor::ChainMonitor::new(Some(chain_source.clone()), tx_broadcaster.clone(), logger.clone(), fee_estimator.clone(), persister.clone())); let best_block = BestBlock::from_network(network); let params = ChainParameters { network, best_block }; - let manager = Arc::new(ChannelManager::new(fee_estimator.clone(), chain_monitor.clone(), tx_broadcaster.clone(), router.clone(), logger.clone(), keys_manager.clone(), keys_manager.clone(), keys_manager.clone(), UserConfig::default(), params)); + let manager = Arc::new(ChannelManager::new(fee_estimator.clone(), chain_monitor.clone(), tx_broadcaster.clone(), router.clone(), logger.clone(), keys_manager.clone(), keys_manager.clone(), keys_manager.clone(), UserConfig::default(), params, genesis_block.header.time)); let p2p_gossip_sync = Arc::new(P2PGossipSync::new(network_graph.clone(), Some(chain_source.clone()), logger.clone())); let rapid_gossip_sync = Arc::new(RapidGossipSync::new(network_graph.clone(), logger.clone())); - let msg_handler = MessageHandler { chan_handler: Arc::new(test_utils::TestChannelMessageHandler::new()), route_handler: Arc::new(test_utils::TestRoutingMessageHandler::new()), onion_message_handler: IgnoringMessageHandler{}}; - let peer_manager = Arc::new(PeerManager::new(msg_handler, 0, &seed, logger.clone(), IgnoringMessageHandler{}, keys_manager.clone())); + let msg_handler = MessageHandler { + chan_handler: Arc::new(test_utils::TestChannelMessageHandler::new(ChainHash::using_genesis_block(Network::Testnet))), + route_handler: Arc::new(test_utils::TestRoutingMessageHandler::new()), + onion_message_handler: IgnoringMessageHandler{}, custom_message_handler: IgnoringMessageHandler{} + }; + let peer_manager = Arc::new(PeerManager::new(msg_handler, 0, &seed, logger.clone(), keys_manager.clone())); let node = Node { node: manager, p2p_gossip_sync, rapid_gossip_sync, peer_manager, chain_monitor, persister, tx_broadcaster, network_graph, logger, best_block, scorer }; nodes.push(node); } for i in 0..num_nodes { for j in (i+1)..num_nodes { - nodes[i].node.peer_connected(&nodes[j].node.get_our_node_id(), &Init { features: nodes[j].node.init_features(), remote_network_address: None }, true).unwrap(); - nodes[j].node.peer_connected(&nodes[i].node.get_our_node_id(), &Init { features: nodes[i].node.init_features(), remote_network_address: None }, false).unwrap(); + nodes[i].node.peer_connected(&nodes[j].node.get_our_node_id(), &Init { + features: nodes[j].node.init_features(), networks: None, remote_network_address: None + }, true).unwrap(); + nodes[j].node.peer_connected(&nodes[i].node.get_our_node_id(), &Init { + features: nodes[i].node.init_features(), networks: None, remote_network_address: None + }, false).unwrap(); } } @@ -1200,7 +1214,7 @@ mod tests { for i in 1..=depth { let prev_blockhash = node.best_block.block_hash(); let height = node.best_block.height() + 1; - let header = BlockHeader { version: 0x20000000, prev_blockhash, merkle_root: TxMerkleNode::all_zeros(), time: height, bits: 42, nonce: 42 }; + let header = create_dummy_header(prev_blockhash, height); let txdata = vec![(0, tx)]; node.best_block = BestBlock::new(header.block_hash(), height); match i { @@ -1720,6 +1734,10 @@ mod tests { if !std::thread::panicking() { bg_processor.stop().unwrap(); } + + let log_entries = nodes[0].logger.lines.lock().unwrap(); + let expected_log = "Persisting scorer after update".to_string(); + assert_eq!(*log_entries.get(&("lightning_background_processor".to_string(), expected_log)).unwrap(), 5); } #[tokio::test] @@ -1762,6 +1780,10 @@ mod tests { let t2 = tokio::spawn(async move { do_test_payment_path_scoring!(nodes, receiver.recv().await); exit_sender.send(()).unwrap(); + + let log_entries = nodes[0].logger.lines.lock().unwrap(); + let expected_log = "Persisting scorer after update".to_string(); + assert_eq!(*log_entries.get(&("lightning_background_processor".to_string(), expected_log)).unwrap(), 5); }); let (r1, r2) = tokio::join!(t1, t2);