X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=lightning-background-processor%2Fsrc%2Flib.rs;h=1e6d447c8cdaf8d8ec9b103ff8869da00d38ed09;hb=f0ecc3ec73dcdb9303b1bd5ac687a361decce2dd;hp=248f79dd59ea5060a0239f566c308cd882120984;hpb=99985c620962fd35163ee8b585c3a7edb6491db7;p=rust-lightning diff --git a/lightning-background-processor/src/lib.rs b/lightning-background-processor/src/lib.rs index 248f79dd..1e6d447c 100644 --- a/lightning-background-processor/src/lib.rs +++ b/lightning-background-processor/src/lib.rs @@ -34,7 +34,7 @@ use lightning::ln::peer_handler::APeerManager; use lightning::routing::gossip::{NetworkGraph, P2PGossipSync}; use lightning::routing::utxo::UtxoLookup; use lightning::routing::router::Router; -use lightning::routing::scoring::{Score, WriteableScore}; +use lightning::routing::scoring::{ScoreUpdate, WriteableScore}; use lightning::util::logger::Logger; use lightning::util::persist::Persister; #[cfg(feature = "std")] @@ -241,23 +241,27 @@ fn handle_network_graph_update( 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), .. } => { + let mut score = scorer.write_lock(); score.payment_path_failed(path, *scid); }, Event::PaymentPathFailed { ref path, payment_failed_permanently: true, .. } => { // Reached if the destination explicitly failed it back. We treat this as a successful probe // because the payment made it all the way to the destination with sufficient liquidity. + let mut score = scorer.write_lock(); score.probe_successful(path); }, Event::PaymentPathSuccessful { path, .. } => { + let mut score = scorer.write_lock(); score.payment_path_successful(path); }, Event::ProbeSuccessful { path, .. } => { + let mut score = scorer.write_lock(); score.probe_successful(path); }, Event::ProbeFailed { path, short_channel_id: Some(scid), .. } => { + let mut score = scorer.write_lock(); score.probe_failed(path, *scid); }, _ => return false, @@ -311,7 +315,7 @@ macro_rules! define_run_body { // see `await_start`'s use below. let mut await_start = None; if $check_slow_await { await_start = Some($get_timer(1)); } - let updates_available = $await; + $await; let await_slow = if $check_slow_await { $timer_elapsed(&mut await_start.unwrap(), 1) } else { false }; // Exit the loop if the background processor was requested to stop. @@ -320,7 +324,7 @@ macro_rules! define_run_body { break; } - if updates_available { + if $channel_manager.get_and_clear_needs_persistence() { log_trace!($logger, "Persisting ChannelManager..."); $persister.persist_manager(&*$channel_manager)?; log_trace!($logger, "Done persisting ChannelManager."); @@ -496,9 +500,16 @@ use core::task; /// For example, in order to process background events in a [Tokio](https://tokio.rs/) task, you /// could setup `process_events_async` like this: /// ``` -/// # struct MyPersister {} -/// # impl lightning::util::persist::KVStorePersister for MyPersister { -/// # fn persist(&self, key: &str, object: &W) -> lightning::io::Result<()> { Ok(()) } +/// # use lightning::io; +/// # use std::sync::{Arc, RwLock}; +/// # use std::sync::atomic::{AtomicBool, Ordering}; +/// # use lightning_background_processor::{process_events_async, GossipSync}; +/// # struct MyStore {} +/// # impl lightning::util::persist::KVStore for MyStore { +/// # fn read(&self, primary_namespace: &str, secondary_namespace: &str, key: &str) -> io::Result> { Ok(Vec::new()) } +/// # fn write(&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: &[u8]) -> io::Result<()> { Ok(()) } +/// # fn remove(&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool) -> io::Result<()> { Ok(()) } +/// # fn list(&self, primary_namespace: &str, secondary_namespace: &str) -> io::Result> { Ok(Vec::new()) } /// # } /// # struct MyEventHandler {} /// # impl MyEventHandler { @@ -510,23 +521,20 @@ use core::task; /// # fn send_data(&mut self, _data: &[u8], _resume_read: bool) -> usize { 0 } /// # fn disconnect_socket(&mut self) {} /// # } -/// # use std::sync::{Arc, Mutex}; -/// # use std::sync::atomic::{AtomicBool, Ordering}; -/// # 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::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 MyChainMonitor = lightning::chain::chainmonitor::ChainMonitor, Arc, Arc, Arc, Arc>; +/// # type MyPeerManager = lightning::ln::peer_handler::SimpleArcPeerManager, MyLogger>; /// # type MyNetworkGraph = lightning::routing::gossip::NetworkGraph>; /// # type MyGossipSync = lightning::routing::gossip::P2PGossipSync, Arc, Arc>; /// # type MyChannelManager = lightning::ln::channelmanager::SimpleArcChannelManager; -/// # type MyScorer = Mutex, Arc>>; +/// # type MyScorer = RwLock, Arc>>; /// -/// # async fn setup_background_processing(my_persister: Arc, my_event_handler: Arc, my_chain_monitor: Arc, my_channel_manager: Arc, my_gossip_sync: Arc, my_logger: Arc, my_scorer: Arc, my_peer_manager: Arc) { +/// # async fn setup_background_processing(my_persister: Arc, my_event_handler: Arc, my_chain_monitor: Arc, my_channel_manager: Arc, my_gossip_sync: Arc, my_logger: Arc, my_scorer: Arc, my_peer_manager: Arc) { /// let background_persister = Arc::clone(&my_persister); /// let background_event_handler = Arc::clone(&my_event_handler); /// let background_chain_mon = Arc::clone(&my_chain_monitor); @@ -591,7 +599,7 @@ pub async fn process_events_async< EventHandlerFuture: core::future::Future, EventHandler: Fn(Event) -> EventHandlerFuture, PS: 'static + Deref + Send, - M: 'static + Deref::Signer, CF, T, F, L, P>> + Send + Sync, + M: 'static + Deref::EcdsaSigner, CF, T, F, L, P>> + Send + Sync, CM: 'static + Deref> + Send + Sync, PGS: 'static + Deref> + Send + Sync, RGS: 'static + Deref> + Send, @@ -609,7 +617,7 @@ pub async fn process_events_async< where UL::Target: 'static + UtxoLookup, CF::Target: 'static + chain::Filter, - CW::Target: 'static + chain::Watch<::Signer>, + CW::Target: 'static + chain::Watch<::EcdsaSigner>, T::Target: 'static + BroadcasterInterface, ES::Target: 'static + EntropySource, NS::Target: 'static + NodeSigner, @@ -617,7 +625,7 @@ where F::Target: 'static + FeeEstimator, R::Target: 'static + Router, L::Target: 'static + Logger, - P::Target: 'static + Persist<::Signer>, + P::Target: 'static + Persist<::EcdsaSigner>, PS::Target: 'static + Persister<'a, CW, T, ES, NS, SP, F, R, L, SC>, { let mut should_break = false; @@ -647,16 +655,14 @@ where channel_manager, channel_manager.process_pending_events_async(async_event_handler).await, gossip_sync, peer_manager, logger, scorer, should_break, { let fut = Selector { - a: channel_manager.get_persistable_update_future(), + a: channel_manager.get_event_or_persistence_needed_future(), b: chain_monitor.get_update_future(), c: sleeper(if mobile_interruptable_platform { Duration::from_millis(100) } else { Duration::from_secs(FASTEST_TIMER) }), }; match fut.await { - SelectorOutput::A => true, - SelectorOutput::B => false, + SelectorOutput::A|SelectorOutput::B => {}, SelectorOutput::C(exit) => { should_break = exit; - false } } }, |t| sleeper(Duration::from_secs(t)), @@ -732,7 +738,7 @@ impl BackgroundProcessor { P: 'static + Deref + Send + Sync, EH: 'static + EventHandler + Send, PS: 'static + Deref + Send, - M: 'static + Deref::Signer, CF, T, F, L, P>> + Send + Sync, + M: 'static + Deref::EcdsaSigner, CF, T, F, L, P>> + Send + Sync, CM: 'static + Deref> + Send + Sync, PGS: 'static + Deref> + Send + Sync, RGS: 'static + Deref> + Send, @@ -747,7 +753,7 @@ impl BackgroundProcessor { where UL::Target: 'static + UtxoLookup, CF::Target: 'static + chain::Filter, - CW::Target: 'static + chain::Watch<::Signer>, + CW::Target: 'static + chain::Watch<::EcdsaSigner>, T::Target: 'static + BroadcasterInterface, ES::Target: 'static + EntropySource, NS::Target: 'static + NodeSigner, @@ -755,7 +761,7 @@ impl BackgroundProcessor { F::Target: 'static + FeeEstimator, R::Target: 'static + Router, L::Target: 'static + Logger, - P::Target: 'static + Persist<::Signer>, + P::Target: 'static + Persist<::EcdsaSigner>, PS::Target: 'static + Persister<'a, CW, T, ES, NS, SP, F, R, L, SC>, { let stop_thread = Arc::new(AtomicBool::new(false)); @@ -779,10 +785,10 @@ impl BackgroundProcessor { define_run_body!(persister, chain_monitor, chain_monitor.process_pending_events(&event_handler), channel_manager, channel_manager.process_pending_events(&event_handler), gossip_sync, peer_manager, logger, scorer, stop_thread.load(Ordering::Acquire), - Sleeper::from_two_futures( - channel_manager.get_persistable_update_future(), + { Sleeper::from_two_futures( + channel_manager.get_event_or_persistence_needed_future(), chain_monitor.get_update_future() - ).wait_timeout(Duration::from_millis(100)), + ).wait_timeout(Duration::from_millis(100)); }, |_| Instant::now(), |time: &Instant, dur| time.elapsed().as_secs() > dur, false) }); Self { stop_thread: stop_thread_clone, thread_handle: Some(handle) } @@ -839,7 +845,7 @@ impl Drop for BackgroundProcessor { #[cfg(all(feature = "std", test))] mod tests { use bitcoin::blockdata::constants::{genesis_block, ChainHash}; - use bitcoin::blockdata::locktime::PackedLockTime; + use bitcoin::blockdata::locktime::absolute::LockTime; use bitcoin::blockdata::transaction::{Transaction, TxOut}; use bitcoin::network::constants::Network; use bitcoin::secp256k1::{SecretKey, PublicKey, Secp256k1}; @@ -857,13 +863,16 @@ mod tests { use lightning::ln::msgs::{ChannelMessageHandler, Init}; use lightning::ln::peer_handler::{PeerManager, MessageHandler, SocketDescriptor, IgnoringMessageHandler}; use lightning::routing::gossip::{NetworkGraph, NodeId, P2PGossipSync}; - use lightning::routing::router::{DefaultRouter, Path, RouteHop}; - use lightning::routing::scoring::{ChannelUsage, Score}; + use lightning::routing::scoring::{ChannelUsage, ScoreUpdate, ScoreLookUp, LockableScore}; + use lightning::routing::router::{DefaultRouter, Path, RouteHop, CandidateRouteHop}; use lightning::util::config::UserConfig; use lightning::util::ser::Writeable; use lightning::util::test_utils; - use lightning::util::persist::KVStorePersister; - use lightning_persister::FilesystemPersister; + use lightning::util::persist::{KVStore, + CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE, CHANNEL_MANAGER_PERSISTENCE_KEY, + NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE, NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE, NETWORK_GRAPH_PERSISTENCE_KEY, + SCORER_PERSISTENCE_PRIMARY_NAMESPACE, SCORER_PERSISTENCE_SECONDARY_NAMESPACE, SCORER_PERSISTENCE_KEY}; + use lightning_persister::fs_store::FilesystemStore; use std::collections::VecDeque; use std::{fs, env}; use std::path::PathBuf; @@ -885,9 +894,29 @@ mod tests { fn disconnect_socket(&mut self) {} } - type ChannelManager = channelmanager::ChannelManager, Arc, Arc, Arc, Arc, Arc, Arc>>, Arc, Arc>, (), TestScorer>>, Arc>; - - type ChainMonitor = chainmonitor::ChainMonitor, Arc, Arc, Arc, Arc>; + #[cfg(c_bindings)] + type LockingWrapper = lightning::routing::scoring::MultiThreadedLockableScore; + #[cfg(not(c_bindings))] + type LockingWrapper = Mutex; + + type ChannelManager = + channelmanager::ChannelManager< + Arc, + Arc, + Arc, + Arc, + Arc, + Arc, + Arc>>, + Arc, + Arc>, + (), + TestScorer> + >, + Arc>; + + type ChainMonitor = chainmonitor::ChainMonitor, Arc, Arc, Arc, Arc>; type PGS = Arc>>, Arc, Arc>>; type RGS = Arc>>, Arc>>; @@ -898,12 +927,12 @@ mod tests { rapid_gossip_sync: RGS, peer_manager: Arc, Arc, IgnoringMessageHandler, Arc, IgnoringMessageHandler, Arc>>, chain_monitor: Arc, - persister: Arc, + kv_store: Arc, tx_broadcaster: Arc, network_graph: Arc>>, logger: Arc, best_block: BestBlock, - scorer: Arc>, + scorer: Arc>, } impl Node { @@ -922,9 +951,9 @@ mod tests { impl Drop for Node { fn drop(&mut self) { - let data_dir = self.persister.get_data_dir(); + let data_dir = self.kv_store.get_data_dir(); match fs::remove_dir_all(data_dir.clone()) { - Err(e) => println!("Failed to remove test persister directory {}: {}", data_dir, e), + Err(e) => println!("Failed to remove test store directory {}: {}", data_dir.display(), e), _ => {} } } @@ -935,13 +964,13 @@ mod tests { graph_persistence_notifier: Option>, manager_error: Option<(std::io::ErrorKind, &'static str)>, scorer_error: Option<(std::io::ErrorKind, &'static str)>, - filesystem_persister: FilesystemPersister, + kv_store: FilesystemStore, } impl Persister { - fn new(data_dir: String) -> Self { - let filesystem_persister = FilesystemPersister::new(data_dir); - Self { graph_error: None, graph_persistence_notifier: None, manager_error: None, scorer_error: None, filesystem_persister } + fn new(data_dir: PathBuf) -> Self { + let kv_store = FilesystemStore::new(data_dir); + Self { graph_error: None, graph_persistence_notifier: None, manager_error: None, scorer_error: None, kv_store } } fn with_graph_error(self, error: std::io::ErrorKind, message: &'static str) -> Self { @@ -961,15 +990,25 @@ mod tests { } } - impl KVStorePersister for Persister { - fn persist(&self, key: &str, object: &W) -> std::io::Result<()> { - if key == "manager" { + impl KVStore for Persister { + fn read(&self, primary_namespace: &str, secondary_namespace: &str, key: &str) -> lightning::io::Result> { + self.kv_store.read(primary_namespace, secondary_namespace, key) + } + + fn write(&self, primary_namespace: &str, secondary_namespace: &str, key: &str, buf: &[u8]) -> lightning::io::Result<()> { + if primary_namespace == CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE && + secondary_namespace == CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE && + key == CHANNEL_MANAGER_PERSISTENCE_KEY + { if let Some((error, message)) = self.manager_error { return Err(std::io::Error::new(error, message)) } } - if key == "network_graph" { + if primary_namespace == NETWORK_GRAPH_PERSISTENCE_PRIMARY_NAMESPACE && + secondary_namespace == NETWORK_GRAPH_PERSISTENCE_SECONDARY_NAMESPACE && + key == NETWORK_GRAPH_PERSISTENCE_KEY + { if let Some(sender) = &self.graph_persistence_notifier { match sender.send(()) { Ok(()) => {}, @@ -982,13 +1021,24 @@ mod tests { } } - if key == "scorer" { + if primary_namespace == SCORER_PERSISTENCE_PRIMARY_NAMESPACE && + secondary_namespace == SCORER_PERSISTENCE_SECONDARY_NAMESPACE && + key == SCORER_PERSISTENCE_KEY + { if let Some((error, message)) = self.scorer_error { return Err(std::io::Error::new(error, message)) } } - self.filesystem_persister.persist(key, object) + self.kv_store.write(primary_namespace, secondary_namespace, key, buf) + } + + fn remove(&self, primary_namespace: &str, secondary_namespace: &str, key: &str, lazy: bool) -> lightning::io::Result<()> { + self.kv_store.remove(primary_namespace, secondary_namespace, key, lazy) + } + + fn list(&self, primary_namespace: &str, secondary_namespace: &str) -> lightning::io::Result> { + self.kv_store.list(primary_namespace, secondary_namespace) } } @@ -1018,12 +1068,14 @@ mod tests { fn write(&self, _: &mut W) -> Result<(), lightning::io::Error> { Ok(()) } } - impl Score for TestScorer { + impl ScoreLookUp for TestScorer { type ScoreParams = (); fn channel_penalty_msat( - &self, _short_channel_id: u64, _source: &NodeId, _target: &NodeId, _usage: ChannelUsage, _score_params: &Self::ScoreParams + &self, _candidate: &CandidateRouteHop, _usage: ChannelUsage, _score_params: &Self::ScoreParams ) -> u64 { unimplemented!(); } + } + impl ScoreUpdate for TestScorer { fn payment_path_failed(&mut self, actual_path: &Path, actual_short_channel_id: u64) { if let Some(expectations) = &mut self.event_expectations { match expectations.pop_front().unwrap() { @@ -1101,6 +1153,9 @@ mod tests { } } + #[cfg(c_bindings)] + impl lightning::routing::scoring::Score for TestScorer {} + impl Drop for TestScorer { fn drop(&mut self) { if std::thread::panicking() { @@ -1132,17 +1187,17 @@ mod tests { let logger = Arc::new(test_utils::TestLogger::with_id(format!("node {}", i))); let genesis_block = genesis_block(network); let network_graph = Arc::new(NetworkGraph::new(network, logger.clone())); - let scorer = Arc::new(Mutex::new(TestScorer::new())); + let scorer = Arc::new(LockingWrapper::new(TestScorer::new())); let seed = [i as u8; 32]; - let router = Arc::new(DefaultRouter::new(network_graph.clone(), logger.clone(), seed, scorer.clone(), ())); + let router = Arc::new(DefaultRouter::new(network_graph.clone(), logger.clone(), seed, scorer.clone(), Default::default())); let chain_source = Arc::new(test_utils::TestChainSource::new(Network::Bitcoin)); - let persister = Arc::new(FilesystemPersister::new(format!("{}_persister_{}", &persist_dir, i))); + let kv_store = Arc::new(FilesystemStore::new(format!("{}_persister_{}", &persist_dir, i).into())); 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 chain_monitor = Arc::new(chainmonitor::ChainMonitor::new(Some(chain_source.clone()), tx_broadcaster.clone(), logger.clone(), fee_estimator.clone(), kv_store.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 { @@ -1151,7 +1206,7 @@ mod tests { 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 }; + let node = Node { node: manager, p2p_gossip_sync, rapid_gossip_sync, peer_manager, chain_monitor, kv_store, tx_broadcaster, network_graph, logger, best_block, scorer }; nodes.push(node); } @@ -1186,7 +1241,7 @@ mod tests { macro_rules! begin_open_channel { ($node_a: expr, $node_b: expr, $channel_value: expr) => {{ - $node_a.node.create_channel($node_b.node.get_our_node_id(), $channel_value, 100, 42, None).unwrap(); + $node_a.node.create_channel($node_b.node.get_our_node_id(), $channel_value, 100, 42, None, None).unwrap(); $node_b.node.handle_open_channel(&$node_a.node.get_our_node_id(), &get_event_msg!($node_a, MessageSendEvent::SendOpenChannel, $node_b.node.get_our_node_id())); $node_a.node.handle_accept_channel(&$node_b.node.get_our_node_id(), &get_event_msg!($node_b, MessageSendEvent::SendAcceptChannel, $node_a.node.get_our_node_id())); }} @@ -1199,7 +1254,7 @@ mod tests { assert_eq!(channel_value_satoshis, $channel_value); assert_eq!(user_channel_id, 42); - let tx = Transaction { version: 1 as i32, lock_time: PackedLockTime(0), input: Vec::new(), output: vec![TxOut { + let tx = Transaction { version: 1 as i32, lock_time: LockTime::ZERO, input: Vec::new(), output: vec![TxOut { value: channel_value_satoshis, script_pubkey: output_script.clone(), }]}; (temporary_channel_id, tx) @@ -1246,7 +1301,7 @@ mod tests { let tx = open_channel!(nodes[0], nodes[1], 100000); // Initiate the background processors to watch each node. - let data_dir = nodes[0].persister.get_data_dir(); + let data_dir = nodes[0].kv_store.get_data_dir(); let persister = Arc::new(Persister::new(data_dir)); let event_handler = |_: _| {}; let bg_processor = BackgroundProcessor::start(persister, event_handler, nodes[0].chain_monitor.clone(), nodes[0].node.clone(), nodes[0].p2p_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(), Some(nodes[0].scorer.clone())); @@ -1280,7 +1335,7 @@ mod tests { check_persisted_data!(nodes[0].node, filepath.clone()); loop { - if !nodes[0].node.get_persistence_condvar_value() { break } + if !nodes[0].node.get_event_or_persist_condvar_value() { break } } // Force-close the channel. @@ -1289,7 +1344,7 @@ mod tests { // Check that the force-close updates are persisted. check_persisted_data!(nodes[0].node, filepath.clone()); loop { - if !nodes[0].node.get_persistence_condvar_value() { break } + if !nodes[0].node.get_event_or_persist_condvar_value() { break } } // Check network graph is persisted @@ -1311,7 +1366,7 @@ mod tests { // `ChainMonitor::rebroadcast_pending_claims` is called every `REBROADCAST_TIMER`, and // `PeerManager::timer_tick_occurred` every `PING_TIMER`. let (_, nodes) = create_nodes(1, "test_timer_tick_called"); - let data_dir = nodes[0].persister.get_data_dir(); + let data_dir = nodes[0].kv_store.get_data_dir(); let persister = Arc::new(Persister::new(data_dir)); let event_handler = |_: _| {}; let bg_processor = BackgroundProcessor::start(persister, event_handler, nodes[0].chain_monitor.clone(), nodes[0].node.clone(), nodes[0].no_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(), Some(nodes[0].scorer.clone())); @@ -1320,9 +1375,9 @@ mod tests { let desired_log_1 = "Calling ChannelManager's timer_tick_occurred".to_string(); let desired_log_2 = "Calling PeerManager's timer_tick_occurred".to_string(); let desired_log_3 = "Rebroadcasting monitor's pending claims".to_string(); - if log_entries.get(&("lightning_background_processor".to_string(), desired_log_1)).is_some() && - log_entries.get(&("lightning_background_processor".to_string(), desired_log_2)).is_some() && - log_entries.get(&("lightning_background_processor".to_string(), desired_log_3)).is_some() { + if log_entries.get(&("lightning_background_processor", desired_log_1)).is_some() && + log_entries.get(&("lightning_background_processor", desired_log_2)).is_some() && + log_entries.get(&("lightning_background_processor", desired_log_3)).is_some() { break } } @@ -1338,7 +1393,7 @@ mod tests { let (_, nodes) = create_nodes(2, "test_persist_error"); open_channel!(nodes[0], nodes[1], 100000); - let data_dir = nodes[0].persister.get_data_dir(); + let data_dir = nodes[0].kv_store.get_data_dir(); let persister = Arc::new(Persister::new(data_dir).with_manager_error(std::io::ErrorKind::Other, "test")); let event_handler = |_: _| {}; let bg_processor = BackgroundProcessor::start(persister, event_handler, nodes[0].chain_monitor.clone(), nodes[0].node.clone(), nodes[0].no_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(), Some(nodes[0].scorer.clone())); @@ -1358,7 +1413,7 @@ mod tests { let (_, nodes) = create_nodes(2, "test_persist_error_sync"); open_channel!(nodes[0], nodes[1], 100000); - let data_dir = nodes[0].persister.get_data_dir(); + let data_dir = nodes[0].kv_store.get_data_dir(); let persister = Arc::new(Persister::new(data_dir).with_manager_error(std::io::ErrorKind::Other, "test")); let bp_future = super::process_events_async( @@ -1384,7 +1439,7 @@ mod tests { fn test_network_graph_persist_error() { // Test that if we encounter an error during network graph persistence, an error gets returned. let (_, nodes) = create_nodes(2, "test_persist_network_graph_error"); - let data_dir = nodes[0].persister.get_data_dir(); + let data_dir = nodes[0].kv_store.get_data_dir(); let persister = Arc::new(Persister::new(data_dir).with_graph_error(std::io::ErrorKind::Other, "test")); let event_handler = |_: _| {}; let bg_processor = BackgroundProcessor::start(persister, event_handler, nodes[0].chain_monitor.clone(), nodes[0].node.clone(), nodes[0].p2p_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(), Some(nodes[0].scorer.clone())); @@ -1402,7 +1457,7 @@ mod tests { fn test_scorer_persist_error() { // Test that if we encounter an error during scorer persistence, an error gets returned. let (_, nodes) = create_nodes(2, "test_persist_scorer_error"); - let data_dir = nodes[0].persister.get_data_dir(); + let data_dir = nodes[0].kv_store.get_data_dir(); let persister = Arc::new(Persister::new(data_dir).with_scorer_error(std::io::ErrorKind::Other, "test")); let event_handler = |_: _| {}; let bg_processor = BackgroundProcessor::start(persister, event_handler, nodes[0].chain_monitor.clone(), nodes[0].node.clone(), nodes[0].no_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(), Some(nodes[0].scorer.clone())); @@ -1420,7 +1475,7 @@ mod tests { fn test_background_event_handling() { let (_, mut nodes) = create_nodes(2, "test_background_event_handling"); let channel_value = 100000; - let data_dir = nodes[0].persister.get_data_dir(); + let data_dir = nodes[0].kv_store.get_data_dir(); let persister = Arc::new(Persister::new(data_dir.clone())); // Set up a background event handler for FundingGenerationReady events. @@ -1493,7 +1548,7 @@ mod tests { #[test] fn test_scorer_persistence() { let (_, nodes) = create_nodes(2, "test_scorer_persistence"); - let data_dir = nodes[0].persister.get_data_dir(); + let data_dir = nodes[0].kv_store.get_data_dir(); let persister = Arc::new(Persister::new(data_dir)); let event_handler = |_: _| {}; let bg_processor = BackgroundProcessor::start(persister, event_handler, nodes[0].chain_monitor.clone(), nodes[0].node.clone(), nodes[0].no_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(), Some(nodes[0].scorer.clone())); @@ -1501,7 +1556,7 @@ mod tests { loop { let log_entries = nodes[0].logger.lines.lock().unwrap(); let expected_log = "Persisting scorer".to_string(); - if log_entries.get(&("lightning_background_processor".to_string(), expected_log)).is_some() { + if log_entries.get(&("lightning_background_processor", expected_log)).is_some() { break } } @@ -1525,7 +1580,7 @@ mod tests { $sleep; let log_entries = $nodes[0].logger.lines.lock().unwrap(); let loop_counter = "Calling ChannelManager's timer_tick_occurred".to_string(); - if *log_entries.get(&("lightning_background_processor".to_string(), loop_counter)) + if *log_entries.get(&("lightning_background_processor", loop_counter)) .unwrap_or(&0) > 1 { // Wait until the loop has gone around at least twice. @@ -1565,7 +1620,7 @@ mod tests { let (sender, receiver) = std::sync::mpsc::sync_channel(1); let (_, nodes) = create_nodes(2, "test_not_pruning_network_graph_until_graph_sync_completion"); - let data_dir = nodes[0].persister.get_data_dir(); + let data_dir = nodes[0].kv_store.get_data_dir(); let persister = Arc::new(Persister::new(data_dir).with_graph_persistence_notifier(sender)); let event_handler = |_: _| {}; @@ -1584,7 +1639,7 @@ mod tests { let (sender, receiver) = std::sync::mpsc::sync_channel(1); let (_, nodes) = create_nodes(2, "test_not_pruning_network_graph_until_graph_sync_completion_async"); - let data_dir = nodes[0].persister.get_data_dir(); + let data_dir = nodes[0].kv_store.get_data_dir(); let persister = Arc::new(Persister::new(data_dir).with_graph_persistence_notifier(sender)); let (exit_sender, exit_receiver) = tokio::sync::watch::channel(()); @@ -1639,9 +1694,10 @@ mod tests { channel_features: ChannelFeatures::empty(), fee_msat: 0, cltv_expiry_delta: MIN_CLTV_EXPIRY_DELTA as u32, + maybe_announced_channel: true, }], blinded_tail: None }; - $nodes[0].scorer.lock().unwrap().expect(TestResult::PaymentFailure { path: path.clone(), short_channel_id: scored_scid }); + $nodes[0].scorer.write_lock().expect(TestResult::PaymentFailure { path: path.clone(), short_channel_id: scored_scid }); $nodes[0].node.push_pending_event(Event::PaymentPathFailed { payment_id: None, payment_hash: PaymentHash([42; 32]), @@ -1658,7 +1714,7 @@ mod tests { // Ensure we'll score payments that were explicitly failed back by the destination as // ProbeSuccess. - $nodes[0].scorer.lock().unwrap().expect(TestResult::ProbeSuccess { path: path.clone() }); + $nodes[0].scorer.write_lock().expect(TestResult::ProbeSuccess { path: path.clone() }); $nodes[0].node.push_pending_event(Event::PaymentPathFailed { payment_id: None, payment_hash: PaymentHash([42; 32]), @@ -1673,7 +1729,7 @@ mod tests { _ => panic!("Unexpected event"), } - $nodes[0].scorer.lock().unwrap().expect(TestResult::PaymentSuccess { path: path.clone() }); + $nodes[0].scorer.write_lock().expect(TestResult::PaymentSuccess { path: path.clone() }); $nodes[0].node.push_pending_event(Event::PaymentPathSuccessful { payment_id: PaymentId([42; 32]), payment_hash: None, @@ -1685,7 +1741,7 @@ mod tests { _ => panic!("Unexpected event"), } - $nodes[0].scorer.lock().unwrap().expect(TestResult::ProbeSuccess { path: path.clone() }); + $nodes[0].scorer.write_lock().expect(TestResult::ProbeSuccess { path: path.clone() }); $nodes[0].node.push_pending_event(Event::ProbeSuccessful { payment_id: PaymentId([42; 32]), payment_hash: PaymentHash([42; 32]), @@ -1697,7 +1753,7 @@ mod tests { _ => panic!("Unexpected event"), } - $nodes[0].scorer.lock().unwrap().expect(TestResult::ProbeFailure { path: path.clone() }); + $nodes[0].scorer.write_lock().expect(TestResult::ProbeFailure { path: path.clone() }); $nodes[0].node.push_pending_event(Event::ProbeFailed { payment_id: PaymentId([42; 32]), payment_hash: PaymentHash([42; 32]), @@ -1724,7 +1780,7 @@ mod tests { }; let (_, nodes) = create_nodes(1, "test_payment_path_scoring"); - let data_dir = nodes[0].persister.get_data_dir(); + let data_dir = nodes[0].kv_store.get_data_dir(); let persister = Arc::new(Persister::new(data_dir)); let bg_processor = BackgroundProcessor::start(persister, event_handler, nodes[0].chain_monitor.clone(), nodes[0].node.clone(), nodes[0].no_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(), Some(nodes[0].scorer.clone())); @@ -1736,7 +1792,7 @@ mod tests { 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); + assert_eq!(*log_entries.get(&("lightning_background_processor", expected_log)).unwrap(), 5); } #[tokio::test] @@ -1757,7 +1813,7 @@ mod tests { }; let (_, nodes) = create_nodes(1, "test_payment_path_scoring_async"); - let data_dir = nodes[0].persister.get_data_dir(); + let data_dir = nodes[0].kv_store.get_data_dir(); let persister = Arc::new(Persister::new(data_dir)); let (exit_sender, exit_receiver) = tokio::sync::watch::channel(()); @@ -1782,7 +1838,7 @@ mod tests { 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); + assert_eq!(*log_entries.get(&("lightning_background_processor", expected_log)).unwrap(), 5); }); let (r1, r2) = tokio::join!(t1, t2);