X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=lightning-background-processor%2Fsrc%2Flib.rs;h=2682e3824b6576d19aaafc8f8d930999ac678a29;hb=989304ed3623a578dc9ad0d9aac0984da0d43584;hp=401c3c19ec2200cf9fd101cbd1fdb2f5b3dcce5b;hpb=63c77a960b3e35a3c28a874b8d321e38831543a0;p=rust-lightning diff --git a/lightning-background-processor/src/lib.rs b/lightning-background-processor/src/lib.rs index 401c3c19..2682e382 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,24 +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 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 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); @@ -648,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)), @@ -780,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) } @@ -859,12 +864,15 @@ mod tests { 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}; 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; @@ -886,9 +894,24 @@ 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>; + 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>>; @@ -899,7 +922,7 @@ 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, @@ -923,9 +946,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), _ => {} } } @@ -936,13 +959,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 { @@ -962,15 +985,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(()) => {}, @@ -983,13 +1016,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) } } @@ -1019,12 +1063,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 ) -> 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() { @@ -1135,12 +1181,12 @@ 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 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, genesis_block.header.time)); @@ -1152,7 +1198,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); } @@ -1247,7 +1293,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())); @@ -1281,7 +1327,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. @@ -1290,7 +1336,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 @@ -1312,7 +1358,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())); @@ -1339,7 +1385,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())); @@ -1359,7 +1405,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( @@ -1385,7 +1431,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())); @@ -1403,7 +1449,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())); @@ -1421,7 +1467,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. @@ -1494,7 +1540,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())); @@ -1566,7 +1612,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 = |_: _| {}; @@ -1585,7 +1631,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(()); @@ -1640,6 +1686,7 @@ 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 }); @@ -1725,7 +1772,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())); @@ -1758,7 +1805,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(());