From a15ef28e942f8e1f5669b3b0b3937b15b54e889c Mon Sep 17 00:00:00 2001 From: Anyitechs Date: Thu, 20 Mar 2025 16:26:54 +0100 Subject: [PATCH 1/7] feat(background-processor): add BackgroundProcessorBuilder for optional components --- lightning-background-processor/src/lib.rs | 289 ++++++++++++++++++++++ 1 file changed, 289 insertions(+) diff --git a/lightning-background-processor/src/lib.rs b/lightning-background-processor/src/lib.rs index 46d990bb37e..d578f8aea48 100644 --- a/lightning-background-processor/src/lib.rs +++ b/lightning-background-processor/src/lib.rs @@ -57,6 +57,9 @@ use std::time::Instant; #[cfg(not(feature = "std"))] use alloc::boxed::Box; +#[cfg(feature = "std")] +use std::marker::PhantomData; + /// `BackgroundProcessor` takes care of tasks that (1) need to happen periodically to keep /// Rust-Lightning running properly, and (2) either can or should be run in the background. Its /// responsibilities are: @@ -1046,6 +1049,193 @@ impl BackgroundProcessor { None => Ok(()), } } + + /// Creates a new [`BackgroundProcessorBuilder`] to construct a [`BackgroundProcessor`] with optional components. + pub fn builder< + 'a, + UL: 'static + Deref + Send + Sync, + CF: 'static + Deref + Send + Sync, + T: 'static + Deref + Send + Sync, + F: 'static + Deref + Send + Sync, + G: 'static + Deref> + Send + Sync, + L: 'static + Deref + Send + Sync, + P: '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, + OM: 'static + Deref + Send + Sync, + PGS: 'static + Deref> + Send + Sync, + RGS: 'static + Deref> + Send, + PM: 'static + Deref + Send + Sync, + S: 'static + Deref + Send + Sync, + SC: for<'b> WriteableScore<'b>, + >( + persister: PS, event_handler: EH, chain_monitor: M, channel_manager: CM, + gossip_sync: GossipSync, peer_manager: PM, logger: L, + ) -> BackgroundProcessorBuilder<'a, UL, CF, T, F, G, L, P, EH, PS, M, CM, OM, PGS, RGS, PM, S, SC> + where + UL::Target: 'static + UtxoLookup, + CF::Target: 'static + chain::Filter, + T::Target: 'static + BroadcasterInterface, + F::Target: 'static + FeeEstimator, + L::Target: 'static + Logger, + P::Target: 'static + Persist<::Signer>, + PS::Target: 'static + Persister<'a, CM, L, S>, + CM::Target: AChannelManager + Send + Sync, + OM::Target: AOnionMessenger + Send + Sync, + PM::Target: APeerManager + Send + Sync, + { + BackgroundProcessorBuilder::new( + persister, + event_handler, + chain_monitor, + channel_manager, + gossip_sync, + peer_manager, + logger, + ) + } +} + +/// A builder for constructing a [`BackgroundProcessor`] with optional components. +/// +/// This builder provides a flexible and type-safe way to construct a [`BackgroundProcessor`] +/// with optional components like `onion_messenger` and `scorer`. It helps avoid specifying +/// concrete types for components that aren't being used. +/// +/// Use [`BackgroundProcessor::builder`] to create a new builder instance. +#[cfg(feature = "std")] +pub struct BackgroundProcessorBuilder< + 'a, + UL: 'static + Deref + Send + Sync, + CF: 'static + Deref + Send + Sync, + T: 'static + Deref + Send + Sync, + F: 'static + Deref + Send + Sync, + G: 'static + Deref> + Send + Sync, + L: 'static + Deref + Send + Sync, + P: '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, + OM: 'static + Deref + Send + Sync, + PGS: 'static + Deref> + Send + Sync, + RGS: 'static + Deref> + Send, + PM: 'static + Deref + Send + Sync, + S: 'static + Deref + Send + Sync, + SC: for<'b> WriteableScore<'b>, +> where + UL::Target: 'static + UtxoLookup, + CF::Target: 'static + chain::Filter, + T::Target: 'static + BroadcasterInterface, + F::Target: 'static + FeeEstimator, + L::Target: 'static + Logger, + P::Target: 'static + Persist<::Signer>, + PS::Target: 'static + Persister<'a, CM, L, S>, + CM::Target: AChannelManager + Send + Sync, + OM::Target: AOnionMessenger + Send + Sync, + PM::Target: APeerManager + Send + Sync, +{ + persister: PS, + event_handler: EH, + chain_monitor: M, + channel_manager: CM, + onion_messenger: Option, + gossip_sync: GossipSync, + peer_manager: PM, + logger: L, + scorer: Option, + _phantom: PhantomData<(&'a (), CF, T, F, P)>, +} + +#[cfg(feature = "std")] +impl< + 'a, + UL: 'static + Deref + Send + Sync, + CF: 'static + Deref + Send + Sync, + T: 'static + Deref + Send + Sync, + F: 'static + Deref + Send + Sync, + G: 'static + Deref> + Send + Sync, + L: 'static + Deref + Send + Sync, + P: '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, + OM: 'static + Deref + Send + Sync, + PGS: 'static + Deref> + Send + Sync, + RGS: 'static + Deref> + Send, + PM: 'static + Deref + Send + Sync, + S: 'static + Deref + Send + Sync, + SC: for<'b> WriteableScore<'b>, + > BackgroundProcessorBuilder<'a, UL, CF, T, F, G, L, P, EH, PS, M, CM, OM, PGS, RGS, PM, S, SC> +where + UL::Target: 'static + UtxoLookup, + CF::Target: 'static + chain::Filter, + T::Target: 'static + BroadcasterInterface, + F::Target: 'static + FeeEstimator, + L::Target: 'static + Logger, + P::Target: 'static + Persist<::Signer>, + PS::Target: 'static + Persister<'a, CM, L, S>, + CM::Target: AChannelManager + Send + Sync, + OM::Target: AOnionMessenger + Send + Sync, + PM::Target: APeerManager + Send + Sync, +{ + /// Creates a new builder instance. This is an internal method - use [`BackgroundProcessor::builder`] instead. + pub fn new( + persister: PS, event_handler: EH, chain_monitor: M, channel_manager: CM, + gossip_sync: GossipSync, peer_manager: PM, logger: L, + ) -> Self { + Self { + persister, + event_handler, + chain_monitor, + channel_manager, + onion_messenger: None, + gossip_sync, + peer_manager, + logger, + scorer: None, + _phantom: PhantomData, + } + } + + /// Sets the optional onion messenger component. + pub fn with_onion_messenger(&mut self, onion_messenger: OM) -> &mut Self { + self.onion_messenger = Some(onion_messenger); + self + } + + /// Sets the optional scorer component. + pub fn with_scorer(&mut self, scorer: S) -> &mut Self { + self.scorer = Some(scorer); + self + } + + /// Builds and starts the [`BackgroundProcessor`]. + pub fn build(self) -> BackgroundProcessor { + BackgroundProcessor::start( + self.persister, + self.event_handler, + self.chain_monitor, + self.channel_manager, + self.onion_messenger, + self.gossip_sync, + self.peer_manager, + self.logger, + self.scorer, + ) + } } #[cfg(feature = "std")] @@ -2721,4 +2911,103 @@ mod tests { r1.unwrap().unwrap(); r2.unwrap() } + + #[test] + fn test_background_processor_builder() { + // Test that when a new channel is created, the ChannelManager needs to be re-persisted with + // updates. Also test that when new updates are available, the manager signals that it needs + // re-persistence and is successfully re-persisted. + let (persist_dir, nodes) = create_nodes(2, "test_background_processor_builder"); + + // Go through the channel creation process so that each node has something to persist. Since + // open_channel consumes events, it must complete before starting BackgroundProcessor to + // avoid a race with processing events. + let tx = open_channel!(nodes[0], nodes[1], 100000); + + // Initiate the background processors to watch each node. + let data_dir = nodes[0].kv_store.get_data_dir(); + let persister = Arc::new(Persister::new(data_dir)); + let event_handler = |_: _| Ok(()); + let bg_processor = BackgroundProcessor::builder( + 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(), + ) + .with_onion_messenger(nodes[0].messenger.clone()) + .with_scorer(nodes[0].scorer.clone()) + .build(); + + macro_rules! check_persisted_data { + ($node: expr, $filepath: expr) => { + let mut expected_bytes = Vec::new(); + loop { + expected_bytes.clear(); + match $node.write(&mut expected_bytes) { + Ok(()) => match std::fs::read($filepath) { + Ok(bytes) => { + if bytes == expected_bytes { + break; + } else { + continue; + } + }, + Err(_) => continue, + }, + Err(e) => panic!("Unexpected error: {}", e), + } + } + }; + } + + // Check that the initial data is persisted as expected + let filepath = + get_full_filepath(format!("{}_persister_0", &persist_dir), "manager".to_string()); + check_persisted_data!(nodes[0].node, filepath.clone()); + + loop { + if !nodes[0].node.get_event_or_persist_condvar_value() { + break; + } + } + + // Force-close the channel. + let error_message = "Channel force-closed"; + nodes[0] + .node + .force_close_broadcasting_latest_txn( + &ChannelId::v1_from_funding_outpoint(OutPoint { + txid: tx.compute_txid(), + index: 0, + }), + &nodes[1].node.get_our_node_id(), + error_message.to_string(), + ) + .unwrap(); + + // Check that the force-close updates are persisted + check_persisted_data!(nodes[0].node, manager_path.clone()); + loop { + if !nodes[0].node.get_event_or_persist_condvar_value() { + break; + } + } + + // Check network graph is persisted + let filepath = + get_full_filepath(format!("{}_persister_0", &persist_dir), "network_graph".to_string()); + check_persisted_data!(nodes[0].network_graph, filepath); + + // Check scorer is persisted + let filepath = + get_full_filepath(format!("{}_persister_0", &persist_dir), "scorer".to_string()); + check_persisted_data!(nodes[0].scorer, filepath); + + if !std::thread::panicking() { + bg_processor.stop().unwrap(); + } + } } From 67b41d09faa2da63edfd6411c643eda02ab7f415 Mon Sep 17 00:00:00 2001 From: Anyitechs Date: Tue, 8 Apr 2025 19:38:54 +0100 Subject: [PATCH 2/7] remove the builder() method in BackgroundProcessor --- lightning-background-processor/src/lib.rs | 57 ++--------------------- 1 file changed, 3 insertions(+), 54 deletions(-) diff --git a/lightning-background-processor/src/lib.rs b/lightning-background-processor/src/lib.rs index d578f8aea48..bcd2f7c13d2 100644 --- a/lightning-background-processor/src/lib.rs +++ b/lightning-background-processor/src/lib.rs @@ -1050,55 +1050,6 @@ impl BackgroundProcessor { } } - /// Creates a new [`BackgroundProcessorBuilder`] to construct a [`BackgroundProcessor`] with optional components. - pub fn builder< - 'a, - UL: 'static + Deref + Send + Sync, - CF: 'static + Deref + Send + Sync, - T: 'static + Deref + Send + Sync, - F: 'static + Deref + Send + Sync, - G: 'static + Deref> + Send + Sync, - L: 'static + Deref + Send + Sync, - P: '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, - OM: 'static + Deref + Send + Sync, - PGS: 'static + Deref> + Send + Sync, - RGS: 'static + Deref> + Send, - PM: 'static + Deref + Send + Sync, - S: 'static + Deref + Send + Sync, - SC: for<'b> WriteableScore<'b>, - >( - persister: PS, event_handler: EH, chain_monitor: M, channel_manager: CM, - gossip_sync: GossipSync, peer_manager: PM, logger: L, - ) -> BackgroundProcessorBuilder<'a, UL, CF, T, F, G, L, P, EH, PS, M, CM, OM, PGS, RGS, PM, S, SC> - where - UL::Target: 'static + UtxoLookup, - CF::Target: 'static + chain::Filter, - T::Target: 'static + BroadcasterInterface, - F::Target: 'static + FeeEstimator, - L::Target: 'static + Logger, - P::Target: 'static + Persist<::Signer>, - PS::Target: 'static + Persister<'a, CM, L, S>, - CM::Target: AChannelManager + Send + Sync, - OM::Target: AOnionMessenger + Send + Sync, - PM::Target: APeerManager + Send + Sync, - { - BackgroundProcessorBuilder::new( - persister, - event_handler, - chain_monitor, - channel_manager, - gossip_sync, - peer_manager, - logger, - ) - } } /// A builder for constructing a [`BackgroundProcessor`] with optional components. @@ -1106,8 +1057,6 @@ impl BackgroundProcessor { /// This builder provides a flexible and type-safe way to construct a [`BackgroundProcessor`] /// with optional components like `onion_messenger` and `scorer`. It helps avoid specifying /// concrete types for components that aren't being used. -/// -/// Use [`BackgroundProcessor::builder`] to create a new builder instance. #[cfg(feature = "std")] pub struct BackgroundProcessorBuilder< 'a, @@ -1191,8 +1140,8 @@ where OM::Target: AOnionMessenger + Send + Sync, PM::Target: APeerManager + Send + Sync, { - /// Creates a new builder instance. This is an internal method - use [`BackgroundProcessor::builder`] instead. - pub fn new( + /// Creates a new builder instance. + pub(crate) fn new( persister: PS, event_handler: EH, chain_monitor: M, channel_manager: CM, gossip_sync: GossipSync, peer_manager: PM, logger: L, ) -> Self { @@ -2928,7 +2877,7 @@ mod tests { let data_dir = nodes[0].kv_store.get_data_dir(); let persister = Arc::new(Persister::new(data_dir)); let event_handler = |_: _| Ok(()); - let bg_processor = BackgroundProcessor::builder( + let bg_processor = BackgroundProcessorBuilder::new( persister, event_handler, nodes[0].chain_monitor.clone(), From 2640f34ed2451ddc540f362afe63603da0c041ee Mon Sep 17 00:00:00 2001 From: Anyitechs Date: Wed, 9 Apr 2025 17:51:03 +0100 Subject: [PATCH 3/7] introduce BackgroundProcessorConfig This allows for consistent configuration of sync and async variant of the BackgroundProcessor --- lightning-background-processor/src/lib.rs | 319 ++++++++++++++++++---- 1 file changed, 272 insertions(+), 47 deletions(-) diff --git a/lightning-background-processor/src/lib.rs b/lightning-background-processor/src/lib.rs index bcd2f7c13d2..93fb5ec2596 100644 --- a/lightning-background-processor/src/lib.rs +++ b/lightning-background-processor/src/lib.rs @@ -754,10 +754,27 @@ pub async fn process_events_async< Sleeper: Fn(Duration) -> SleepFuture, FetchTime: Fn() -> Option, >( - persister: PS, event_handler: EventHandler, chain_monitor: M, channel_manager: CM, - onion_messenger: Option, gossip_sync: GossipSync, peer_manager: PM, - logger: L, scorer: Option, sleeper: Sleeper, mobile_interruptable_platform: bool, - fetch_time: FetchTime, + config: BackgroundProcessorConfig< + 'a, + UL, + CF, + T, + F, + G, + L, + P, + EH, + PS, + M, + CM, + OM, + PGS, + RGS, + PM, + S, + SC, + >, + sleeper: Sleeper, mobile_interruptable_platform: bool, fetch_time: FetchTime, ) -> Result<(), lightning::io::Error> where UL::Target: 'static + UtxoLookup, @@ -773,11 +790,11 @@ where { let mut should_break = false; let async_event_handler = |event| { - let network_graph = gossip_sync.network_graph(); - let event_handler = &event_handler; - let scorer = &scorer; - let logger = &logger; - let persister = &persister; + let network_graph = config.gossip_sync.network_graph(); + let event_handler = &config.event_handler; + let scorer = &config.scorer; + let logger = &config.logger; + let persister = &config.persister; let fetch_time = &fetch_time; // We should be able to drop the Box once our MSRV is 1.68 Box::pin(async move { @@ -802,30 +819,30 @@ where }) }; define_run_body!( - persister, - chain_monitor, - chain_monitor.process_pending_events_async(async_event_handler).await, - channel_manager, - channel_manager.get_cm().process_pending_events_async(async_event_handler).await, - onion_messenger, - if let Some(om) = &onion_messenger { + config.persister, + config.chain_monitor, + config.chain_monitor.process_pending_events_async(async_event_handler).await, + config.channel_manager, + config.channel_manager.get_cm().process_pending_events_async(async_event_handler).await, + config.onion_messenger, + if let Some(om) = &config.onion_messenger { om.get_om().process_pending_events_async(async_event_handler).await }, - peer_manager, - gossip_sync, - logger, - scorer, + config.peer_manager, + config.gossip_sync, + config.logger, + config.scorer, should_break, { - let om_fut = if let Some(om) = onion_messenger.as_ref() { + let om_fut = if let Some(om) = config.onion_messenger.as_ref() { let fut = om.get_om().get_update_future(); OptionalSelector { optional_future: Some(fut) } } else { OptionalSelector { optional_future: None } }; let fut = Selector { - a: channel_manager.get_cm().get_event_or_persistence_needed_future(), - b: chain_monitor.get_update_future(), + a: config.channel_manager.get_cm().get_event_or_persistence_needed_future(), + b: config.chain_monitor.get_update_future(), c: om_fut, d: sleeper(if mobile_interruptable_platform { Duration::from_millis(100) @@ -1010,6 +1027,98 @@ impl BackgroundProcessor { Self { stop_thread: stop_thread_clone, thread_handle: Some(handle) } } + /// Creates a new [`BackgroundProcessor`] from a [`BackgroundProcessorConfig`]. + /// This provides a more structured approach to configuration. The processor will start processing events immediately upon creation. + /// + /// This method is functionally equivalent to [`BackgroundProcessor::start`], but takes a configuration + /// object instead of individual parameters. + /// + /// # Example + /// ``` + /// # use lightning_background_processor::*; + /// let config = BackgroundProcessorBuilder::new( + /// persister, + /// event_handler, + /// chain_monitor, + /// channel_manager, + /// gossip_sync, + /// peer_manager, + /// logger + /// ) + /// .with_onion_messenger(messenger) + /// .with_scorer(scorer) + /// .build(); + /// let bg_processor = BackgroundProcessor::from_config(config); + /// ``` + pub fn from_config< + 'a, + UL: 'static + Deref + Send + Sync, + CF: 'static + Deref + Send + Sync, + T: 'static + Deref + Send + Sync, + F: 'static + Deref + Send + Sync, + G: 'static + Deref> + Send + Sync, + L: 'static + Deref + Send + Sync, + P: '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, + OM: 'static + Deref + Send + Sync, + PGS: 'static + Deref> + Send + Sync, + RGS: 'static + Deref> + Send, + PM: 'static + Deref + Send + Sync, + S: 'static + Deref + Send + Sync, + SC: for<'b> WriteableScore<'b>, + >( + config: BackgroundProcessorConfig< + 'a, + UL, + CF, + T, + F, + G, + L, + P, + EH, + PS, + M, + CM, + OM, + PGS, + RGS, + PM, + S, + SC, + >, + ) -> Self + where + UL::Target: 'static + UtxoLookup, + CF::Target: 'static + chain::Filter, + T::Target: 'static + BroadcasterInterface, + F::Target: 'static + FeeEstimator, + L::Target: 'static + Logger, + P::Target: 'static + Persist<::Signer>, + PS::Target: 'static + Persister<'a, CM, L, S>, + CM::Target: AChannelManager + Send + Sync, + OM::Target: AOnionMessenger + Send + Sync, + PM::Target: APeerManager + Send + Sync, + { + Self::start( + config.persister, + config.event_handler, + config.chain_monitor, + config.channel_manager, + config.onion_messenger, + config.gossip_sync, + config.peer_manager, + config.logger, + config.scorer, + ) + } + /// Join `BackgroundProcessor`'s thread, returning any error that occurred while persisting /// [`ChannelManager`]. /// @@ -1049,7 +1158,102 @@ impl BackgroundProcessor { None => Ok(()), } } +} +/// Configuration for synchronous [`BackgroundProcessor`] +#[cfg_attr(feature = "futures", doc = "and asynchronous [`process_events_async`]")] +/// event processing. +/// +/// This configuration holds all components needed for background processing, +/// including required components (like the channel manager and peer manager) and optional +/// components (like the onion messenger and scorer). +/// +/// The configuration can be constructed using [`BackgroundProcessorBuilder`], which provides +/// a convenient builder pattern for setting up both required and optional components. +/// +/// This same configuration can be used for +#[cfg_attr( + not(feature = "futures"), + doc = "creating a [`BackgroundProcessor`] via [`BackgroundProcessor::from_config`]." +)] +#[cfg_attr( + feature = "futures", + doc = "both: +/// * Creating a [`BackgroundProcessor`] via [`BackgroundProcessor::from_config`] +/// * Running the async variant of the background processor via [`process_events_async`]" +)] +/// +/// # Example +/// ``` +/// # use lightning_background_processor::*; +/// let config = BackgroundProcessorBuilder::new( +/// persister, +/// event_handler, +/// chain_monitor, +/// channel_manager, +/// gossip_sync, +/// peer_manager, +/// logger +/// ) +/// .with_onion_messenger(messenger) // Optional +/// .with_scorer(scorer) // Optional +/// .build(); +/// +/// // Use with BackgroundProcessor +/// let processor = BackgroundProcessor::from_config(config); +/// +#[cfg_attr( + feature = "futures", + doc = " +/// // Or use with async processing +/// process_events_async(config, sleeper, mobile_interruptable_platform, fetch_time).await?;" +)] +/// ``` +#[cfg(feature = "std")] +pub struct BackgroundProcessorConfig< + 'a, + UL: 'static + Deref + Send + Sync, + CF: 'static + Deref + Send + Sync, + T: 'static + Deref + Send + Sync, + F: 'static + Deref + Send + Sync, + G: 'static + Deref> + Send + Sync, + L: 'static + Deref + Send + Sync, + P: '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, + OM: 'static + Deref + Send + Sync, + PGS: 'static + Deref> + Send + Sync, + RGS: 'static + Deref> + Send, + PM: 'static + Deref + Send + Sync, + S: 'static + Deref + Send + Sync, + SC: for<'b> WriteableScore<'b>, +> where + UL::Target: 'static + UtxoLookup, + CF::Target: 'static + chain::Filter, + T::Target: 'static + BroadcasterInterface, + F::Target: 'static + FeeEstimator, + L::Target: 'static + Logger, + P::Target: 'static + Persist<::Signer>, + PS::Target: 'static + Persister<'a, CM, L, S>, + CM::Target: AChannelManager + Send + Sync, + OM::Target: AOnionMessenger + Send + Sync, + PM::Target: APeerManager + Send + Sync, +{ + persister: PS, + event_handler: EH, + chain_monitor: M, + channel_manager: CM, + onion_messenger: Option, + gossip_sync: GossipSync, + peer_manager: PM, + logger: L, + scorer: Option, + _phantom: PhantomData<(&'a (), CF, T, F, P)>, } /// A builder for constructing a [`BackgroundProcessor`] with optional components. @@ -1171,19 +1375,23 @@ where self } - /// Builds and starts the [`BackgroundProcessor`]. - pub fn build(self) -> BackgroundProcessor { - BackgroundProcessor::start( - self.persister, - self.event_handler, - self.chain_monitor, - self.channel_manager, - self.onion_messenger, - self.gossip_sync, - self.peer_manager, - self.logger, - self.scorer, - ) + /// Builds and returns a [`BackgroundProcessorConfig`] object. + pub fn build( + self, + ) -> BackgroundProcessorConfig<'a, UL, CF, T, F, G, L, P, EH, PS, M, CM, OM, PGS, RGS, PM, S, SC> + { + BackgroundProcessorConfig { + persister: self.persister, + event_handler: self.event_handler, + chain_monitor: self.chain_monitor, + channel_manager: self.channel_manager, + onion_messenger: self.onion_messenger, + gossip_sync: self.gossip_sync, + peer_manager: self.peer_manager, + logger: self.logger, + scorer: self.scorer, + _phantom: PhantomData, + } } } @@ -2129,16 +2337,21 @@ mod tests { Persister::new(data_dir).with_manager_error(std::io::ErrorKind::Other, "test"), ); - let bp_future = super::process_events_async( + let config = BackgroundProcessorBuilder::new( persister, |_: _| async { Ok(()) }, nodes[0].chain_monitor.clone(), nodes[0].node.clone(), - Some(nodes[0].messenger.clone()), nodes[0].rapid_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(), - Some(nodes[0].scorer.clone()), + ) + .with_onion_messenger(nodes[0].messenger.clone()) + .with_scorer(nodes[0].scorer.clone()) + .build(); + + let bp_future = super::process_events_async( + config, move |dur: Duration| { Box::pin(async move { tokio::time::sleep(dur).await; @@ -2608,17 +2821,22 @@ mod tests { 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(()); - let bp_future = super::process_events_async( + let config = BackgroundProcessorBuilder::new( persister, |_: _| async { Ok(()) }, nodes[0].chain_monitor.clone(), nodes[0].node.clone(), - Some(nodes[0].messenger.clone()), nodes[0].rapid_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(), - Some(nodes[0].scorer.clone()), + ) + .with_onion_messenger(nodes[0].messenger.clone()) + .with_scorer(nodes[0].scorer.clone()) + .build(); + + let (exit_sender, exit_receiver) = tokio::sync::watch::channel(()); + let bp_future = super::process_events_async( + config, move |dur: Duration| { let mut exit_receiver = exit_receiver.clone(); Box::pin(async move { @@ -2821,16 +3039,21 @@ mod tests { let (exit_sender, exit_receiver) = tokio::sync::watch::channel(()); - let bp_future = super::process_events_async( + let config = BackgroundProcessorBuilder::new( persister, event_handler, nodes[0].chain_monitor.clone(), nodes[0].node.clone(), - Some(nodes[0].messenger.clone()), nodes[0].no_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(), - Some(nodes[0].scorer.clone()), + ) + .with_onion_messenger(nodes[0].messenger.clone()) + .with_scorer(nodes[0].scorer.clone()) + .build(); + + let bp_future = super::process_events_async( + config, move |dur: Duration| { let mut exit_receiver = exit_receiver.clone(); Box::pin(async move { @@ -2877,7 +3100,7 @@ mod tests { let data_dir = nodes[0].kv_store.get_data_dir(); let persister = Arc::new(Persister::new(data_dir)); let event_handler = |_: _| Ok(()); - let bg_processor = BackgroundProcessorBuilder::new( + let config = BackgroundProcessorBuilder::new( persister, event_handler, nodes[0].chain_monitor.clone(), @@ -2890,6 +3113,8 @@ mod tests { .with_scorer(nodes[0].scorer.clone()) .build(); + let bg_processor = BackgroundProcessor::from_config(config); + macro_rules! check_persisted_data { ($node: expr, $filepath: expr) => { let mut expected_bytes = Vec::new(); From a9c49a0fabd1e92a9fcc8d0a86172cd7a08654fe Mon Sep 17 00:00:00 2001 From: Anyitechs Date: Mon, 14 Apr 2025 01:13:48 +0100 Subject: [PATCH 4/7] make BackgroundProcessorConfig docs conditional on futures feature From e598990fa0709abbe3c08a7e5d87f9ffe960b2e1 Mon Sep 17 00:00:00 2001 From: Anyitechs Date: Mon, 14 Apr 2025 01:33:32 +0100 Subject: [PATCH 5/7] rename builder to BackgroundProcessorConfigBuilder The builder s and returns a object, so renaming it to seems more suitable. --- lightning-background-processor/src/lib.rs | 23 ++++++++++++----------- 1 file changed, 12 insertions(+), 11 deletions(-) diff --git a/lightning-background-processor/src/lib.rs b/lightning-background-processor/src/lib.rs index 93fb5ec2596..f0729bc9835 100644 --- a/lightning-background-processor/src/lib.rs +++ b/lightning-background-processor/src/lib.rs @@ -1036,7 +1036,7 @@ impl BackgroundProcessor { /// # Example /// ``` /// # use lightning_background_processor::*; - /// let config = BackgroundProcessorBuilder::new( + /// let config = BackgroundProcessorConfigBuilder::new( /// persister, /// event_handler, /// chain_monitor, @@ -1168,7 +1168,7 @@ impl BackgroundProcessor { /// including required components (like the channel manager and peer manager) and optional /// components (like the onion messenger and scorer). /// -/// The configuration can be constructed using [`BackgroundProcessorBuilder`], which provides +/// The configuration can be constructed using [`BackgroundProcessorConfigBuilder`], which provides /// a convenient builder pattern for setting up both required and optional components. /// /// This same configuration can be used for @@ -1186,7 +1186,7 @@ impl BackgroundProcessor { /// # Example /// ``` /// # use lightning_background_processor::*; -/// let config = BackgroundProcessorBuilder::new( +/// let config = BackgroundProcessorConfigBuilder::new( /// persister, /// event_handler, /// chain_monitor, @@ -1256,13 +1256,13 @@ pub struct BackgroundProcessorConfig< _phantom: PhantomData<(&'a (), CF, T, F, P)>, } -/// A builder for constructing a [`BackgroundProcessor`] with optional components. +/// A builder for constructing a [`BackgroundProcessorConfig`] with optional components. /// -/// This builder provides a flexible and type-safe way to construct a [`BackgroundProcessor`] +/// This builder provides a flexible and type-safe way to construct a [`BackgroundProcessorConfig`] /// with optional components like `onion_messenger` and `scorer`. It helps avoid specifying /// concrete types for components that aren't being used. #[cfg(feature = "std")] -pub struct BackgroundProcessorBuilder< +pub struct BackgroundProcessorConfigBuilder< 'a, UL: 'static + Deref + Send + Sync, CF: 'static + Deref + Send + Sync, @@ -1331,7 +1331,8 @@ impl< PM: 'static + Deref + Send + Sync, S: 'static + Deref + Send + Sync, SC: for<'b> WriteableScore<'b>, - > BackgroundProcessorBuilder<'a, UL, CF, T, F, G, L, P, EH, PS, M, CM, OM, PGS, RGS, PM, S, SC> + > + BackgroundProcessorConfigBuilder<'a, UL, CF, T, F, G, L, P, EH, PS, M, CM, OM, PGS, RGS, PM, S, SC> where UL::Target: 'static + UtxoLookup, CF::Target: 'static + chain::Filter, @@ -2337,7 +2338,7 @@ mod tests { Persister::new(data_dir).with_manager_error(std::io::ErrorKind::Other, "test"), ); - let config = BackgroundProcessorBuilder::new( + let config = BackgroundProcessorConfigBuilder::new( persister, |_: _| async { Ok(()) }, nodes[0].chain_monitor.clone(), @@ -2821,7 +2822,7 @@ mod tests { let data_dir = nodes[0].kv_store.get_data_dir(); let persister = Arc::new(Persister::new(data_dir).with_graph_persistence_notifier(sender)); - let config = BackgroundProcessorBuilder::new( + let config = BackgroundProcessorConfigBuilder::new( persister, |_: _| async { Ok(()) }, nodes[0].chain_monitor.clone(), @@ -3039,7 +3040,7 @@ mod tests { let (exit_sender, exit_receiver) = tokio::sync::watch::channel(()); - let config = BackgroundProcessorBuilder::new( + let config = BackgroundProcessorConfigBuilder::new( persister, event_handler, nodes[0].chain_monitor.clone(), @@ -3100,7 +3101,7 @@ mod tests { let data_dir = nodes[0].kv_store.get_data_dir(); let persister = Arc::new(Persister::new(data_dir)); let event_handler = |_: _| Ok(()); - let config = BackgroundProcessorBuilder::new( + let config = BackgroundProcessorConfigBuilder::new( persister, event_handler, nodes[0].chain_monitor.clone(), From a7550fbe4ab7dfad7f5f38487a098fd171e56ba2 Mon Sep 17 00:00:00 2001 From: Anyitechs Date: Mon, 14 Apr 2025 14:45:39 +0100 Subject: [PATCH 6/7] fix(CI): fix CI build errors --- lightning-background-processor/src/lib.rs | 123 ++++++++++++---------- 1 file changed, 70 insertions(+), 53 deletions(-) diff --git a/lightning-background-processor/src/lib.rs b/lightning-background-processor/src/lib.rs index f0729bc9835..b50e8c1fb82 100644 --- a/lightning-background-processor/src/lib.rs +++ b/lightning-background-processor/src/lib.rs @@ -763,7 +763,7 @@ pub async fn process_events_async< G, L, P, - EH, + EventHandler, PS, M, CM, @@ -818,31 +818,41 @@ where event_handler(event).await }) }; + // We should extract these out of config because the macro expects individual arguments + let persister = config.persister; + let chain_monitor = config.chain_monitor; + let channel_manager = config.channel_manager; + let onion_messenger = config.onion_messenger; + let peer_manager = config.peer_manager; + let gossip_sync = config.gossip_sync; + let logger = config.logger; + let scorer = config.scorer; + define_run_body!( - config.persister, - config.chain_monitor, - config.chain_monitor.process_pending_events_async(async_event_handler).await, - config.channel_manager, - config.channel_manager.get_cm().process_pending_events_async(async_event_handler).await, - config.onion_messenger, - if let Some(om) = &config.onion_messenger { + persister, + chain_monitor, + chain_monitor.process_pending_events_async(async_event_handler).await, + channel_manager, + channel_manager.get_cm().process_pending_events_async(async_event_handler).await, + onion_messenger, + if let Some(om) = &onion_messenger { om.get_om().process_pending_events_async(async_event_handler).await }, - config.peer_manager, - config.gossip_sync, - config.logger, - config.scorer, + peer_manager, + gossip_sync, + logger, + scorer, should_break, { - let om_fut = if let Some(om) = config.onion_messenger.as_ref() { + let om_fut = if let Some(om) = onion_messenger.as_ref() { let fut = om.get_om().get_update_future(); OptionalSelector { optional_future: Some(fut) } } else { OptionalSelector { optional_future: None } }; let fut = Selector { - a: config.channel_manager.get_cm().get_event_or_persistence_needed_future(), - b: config.chain_monitor.get_update_future(), + a: channel_manager.get_cm().get_event_or_persistence_needed_future(), + b: chain_monitor.get_update_future(), c: om_fut, d: sleeper(if mobile_interruptable_platform { Duration::from_millis(100) @@ -1036,7 +1046,7 @@ impl BackgroundProcessor { /// # Example /// ``` /// # use lightning_background_processor::*; - /// let config = BackgroundProcessorConfigBuilder::new( + /// let mut builder = BackgroundProcessorConfigBuilder::new( /// persister, /// event_handler, /// chain_monitor, @@ -1044,10 +1054,10 @@ impl BackgroundProcessor { /// gossip_sync, /// peer_manager, /// logger - /// ) - /// .with_onion_messenger(messenger) - /// .with_scorer(scorer) - /// .build(); + /// ); + /// builder.with_onion_messenger(messenger); + /// .with_scorer(scorer); + /// let config = builder.build(); /// let bg_processor = BackgroundProcessor::from_config(config); /// ``` pub fn from_config< @@ -1186,7 +1196,7 @@ impl BackgroundProcessor { /// # Example /// ``` /// # use lightning_background_processor::*; -/// let config = BackgroundProcessorConfigBuilder::new( +/// let mut builder = BackgroundProcessorConfigBuilder::new( /// persister, /// event_handler, /// chain_monitor, @@ -1194,10 +1204,10 @@ impl BackgroundProcessor { /// gossip_sync, /// peer_manager, /// logger -/// ) -/// .with_onion_messenger(messenger) // Optional -/// .with_scorer(scorer) // Optional -/// .build(); +/// ); +/// builder.with_onion_messenger(messenger); // Optional +/// .with_scorer(scorer); // Optional +/// let config = builder.build(); /// /// // Use with BackgroundProcessor /// let processor = BackgroundProcessor::from_config(config); @@ -1209,7 +1219,7 @@ impl BackgroundProcessor { /// process_events_async(config, sleeper, mobile_interruptable_platform, fetch_time).await?;" )] /// ``` -#[cfg(feature = "std")] +#[cfg(any(feature = "std", feature = "futures"))] pub struct BackgroundProcessorConfig< 'a, UL: 'static + Deref + Send + Sync, @@ -1219,7 +1229,8 @@ pub struct BackgroundProcessorConfig< G: 'static + Deref> + Send + Sync, L: 'static + Deref + Send + Sync, P: 'static + Deref + Send + Sync, - EH: 'static + EventHandler + Send, + #[cfg(feature = "std")] EH: 'static + EventHandler + Send, + #[cfg(feature = "futures")] EH: 'static + Fn(Event) -> core::future::Future>, PS: 'static + Deref + Send, M: 'static + Deref::Signer, CF, T, F, L, P>> @@ -1346,7 +1357,7 @@ where PM::Target: APeerManager + Send + Sync, { /// Creates a new builder instance. - pub(crate) fn new( + pub fn new( persister: PS, event_handler: EH, chain_monitor: M, channel_manager: CM, gossip_sync: GossipSync, peer_manager: PM, logger: L, ) -> Self { @@ -1405,7 +1416,9 @@ impl Drop for BackgroundProcessor { #[cfg(all(feature = "std", test))] mod tests { - use super::{BackgroundProcessor, GossipSync, FRESHNESS_TIMER}; + use super::{ + BackgroundProcessor, BackgroundProcessorConfigBuilder, GossipSync, FRESHNESS_TIMER, + }; use bitcoin::constants::{genesis_block, ChainHash}; use bitcoin::hashes::Hash; use bitcoin::locktime::absolute::LockTime; @@ -2338,7 +2351,7 @@ mod tests { Persister::new(data_dir).with_manager_error(std::io::ErrorKind::Other, "test"), ); - let config = BackgroundProcessorConfigBuilder::new( + let mut builder = BackgroundProcessorConfigBuilder::new( persister, |_: _| async { Ok(()) }, nodes[0].chain_monitor.clone(), @@ -2346,10 +2359,11 @@ mod tests { nodes[0].rapid_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(), - ) - .with_onion_messenger(nodes[0].messenger.clone()) - .with_scorer(nodes[0].scorer.clone()) - .build(); + ); + builder + .with_onion_messenger(nodes[0].messenger.clone()) + .with_scorer(nodes[0].scorer.clone()); + let config = builder.build(); let bp_future = super::process_events_async( config, @@ -2822,7 +2836,7 @@ mod tests { let data_dir = nodes[0].kv_store.get_data_dir(); let persister = Arc::new(Persister::new(data_dir).with_graph_persistence_notifier(sender)); - let config = BackgroundProcessorConfigBuilder::new( + let mut builder = BackgroundProcessorConfigBuilder::new( persister, |_: _| async { Ok(()) }, nodes[0].chain_monitor.clone(), @@ -2830,10 +2844,11 @@ mod tests { nodes[0].rapid_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(), - ) - .with_onion_messenger(nodes[0].messenger.clone()) - .with_scorer(nodes[0].scorer.clone()) - .build(); + ); + builder + .with_onion_messenger(nodes[0].messenger.clone()) + .with_scorer(nodes[0].scorer.clone()); + let config = builder.build(); let (exit_sender, exit_receiver) = tokio::sync::watch::channel(()); let bp_future = super::process_events_async( @@ -3040,7 +3055,7 @@ mod tests { let (exit_sender, exit_receiver) = tokio::sync::watch::channel(()); - let config = BackgroundProcessorConfigBuilder::new( + let mut builder = BackgroundProcessorConfigBuilder::new( persister, event_handler, nodes[0].chain_monitor.clone(), @@ -3048,10 +3063,11 @@ mod tests { nodes[0].no_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(), - ) - .with_onion_messenger(nodes[0].messenger.clone()) - .with_scorer(nodes[0].scorer.clone()) - .build(); + ); + builder + .with_onion_messenger(nodes[0].messenger.clone()) + .with_scorer(nodes[0].scorer.clone()); + let config = builder.build(); let bp_future = super::process_events_async( config, @@ -3086,11 +3102,11 @@ mod tests { } #[test] - fn test_background_processor_builder() { + fn test_background_processor_config_builder() { // Test that when a new channel is created, the ChannelManager needs to be re-persisted with // updates. Also test that when new updates are available, the manager signals that it needs // re-persistence and is successfully re-persisted. - let (persist_dir, nodes) = create_nodes(2, "test_background_processor_builder"); + let (persist_dir, nodes) = create_nodes(2, "test_background_processor_config_builder"); // Go through the channel creation process so that each node has something to persist. Since // open_channel consumes events, it must complete before starting BackgroundProcessor to @@ -3101,7 +3117,7 @@ mod tests { let data_dir = nodes[0].kv_store.get_data_dir(); let persister = Arc::new(Persister::new(data_dir)); let event_handler = |_: _| Ok(()); - let config = BackgroundProcessorConfigBuilder::new( + let mut builder = BackgroundProcessorConfigBuilder::new( persister, event_handler, nodes[0].chain_monitor.clone(), @@ -3109,10 +3125,11 @@ mod tests { nodes[0].p2p_gossip_sync(), nodes[0].peer_manager.clone(), nodes[0].logger.clone(), - ) - .with_onion_messenger(nodes[0].messenger.clone()) - .with_scorer(nodes[0].scorer.clone()) - .build(); + ); + builder + .with_onion_messenger(nodes[0].messenger.clone()) + .with_scorer(nodes[0].scorer.clone()); + let config = builder.build(); let bg_processor = BackgroundProcessor::from_config(config); @@ -3164,7 +3181,7 @@ mod tests { .unwrap(); // Check that the force-close updates are persisted - check_persisted_data!(nodes[0].node, manager_path.clone()); + check_persisted_data!(nodes[0].node, filepath.clone()); loop { if !nodes[0].node.get_event_or_persist_condvar_value() { break; @@ -3174,7 +3191,7 @@ mod tests { // Check network graph is persisted let filepath = get_full_filepath(format!("{}_persister_0", &persist_dir), "network_graph".to_string()); - check_persisted_data!(nodes[0].network_graph, filepath); + check_persisted_data!(nodes[0].network_graph, filepath.clone()); // Check scorer is persisted let filepath = From 7e707a9e9bbba3c2f3cb022564695197def78783 Mon Sep 17 00:00:00 2001 From: Anyitechs Date: Wed, 16 Apr 2025 14:14:43 +0100 Subject: [PATCH 7/7] fix ci build errors --- lightning-background-processor/src/lib.rs | 68 +++++------------------ 1 file changed, 14 insertions(+), 54 deletions(-) diff --git a/lightning-background-processor/src/lib.rs b/lightning-background-processor/src/lib.rs index b50e8c1fb82..39da6e95460 100644 --- a/lightning-background-processor/src/lib.rs +++ b/lightning-background-processor/src/lib.rs @@ -44,6 +44,7 @@ use lightning_rapid_gossip_sync::RapidGossipSync; use core::ops::Deref; use core::time::Duration; +use std::marker::PhantomData; #[cfg(feature = "std")] use core::sync::atomic::{AtomicBool, Ordering}; @@ -57,9 +58,6 @@ use std::time::Instant; #[cfg(not(feature = "std"))] use alloc::boxed::Box; -#[cfg(feature = "std")] -use std::marker::PhantomData; - /// `BackgroundProcessor` takes care of tasks that (1) need to happen periodically to keep /// Rust-Lightning running properly, and (2) either can or should be run in the background. Its /// responsibilities are: @@ -754,7 +752,7 @@ pub async fn process_events_async< Sleeper: Fn(Duration) -> SleepFuture, FetchTime: Fn() -> Option, >( - config: BackgroundProcessorConfig< + #[rustfmt::skip] config: BackgroundProcessorConfig< 'a, UL, CF, @@ -1043,23 +1041,6 @@ impl BackgroundProcessor { /// This method is functionally equivalent to [`BackgroundProcessor::start`], but takes a configuration /// object instead of individual parameters. /// - /// # Example - /// ``` - /// # use lightning_background_processor::*; - /// let mut builder = BackgroundProcessorConfigBuilder::new( - /// persister, - /// event_handler, - /// chain_monitor, - /// channel_manager, - /// gossip_sync, - /// peer_manager, - /// logger - /// ); - /// builder.with_onion_messenger(messenger); - /// .with_scorer(scorer); - /// let config = builder.build(); - /// let bg_processor = BackgroundProcessor::from_config(config); - /// ``` pub fn from_config< 'a, UL: 'static + Deref + Send + Sync, @@ -1083,7 +1064,7 @@ impl BackgroundProcessor { S: 'static + Deref + Send + Sync, SC: for<'b> WriteableScore<'b>, >( - config: BackgroundProcessorConfig< + #[rustfmt::skip] config: BackgroundProcessorConfig< 'a, UL, CF, @@ -1193,32 +1174,6 @@ impl BackgroundProcessor { /// * Running the async variant of the background processor via [`process_events_async`]" )] /// -/// # Example -/// ``` -/// # use lightning_background_processor::*; -/// let mut builder = BackgroundProcessorConfigBuilder::new( -/// persister, -/// event_handler, -/// chain_monitor, -/// channel_manager, -/// gossip_sync, -/// peer_manager, -/// logger -/// ); -/// builder.with_onion_messenger(messenger); // Optional -/// .with_scorer(scorer); // Optional -/// let config = builder.build(); -/// -/// // Use with BackgroundProcessor -/// let processor = BackgroundProcessor::from_config(config); -/// -#[cfg_attr( - feature = "futures", - doc = " -/// // Or use with async processing -/// process_events_async(config, sleeper, mobile_interruptable_platform, fetch_time).await?;" -)] -/// ``` #[cfg(any(feature = "std", feature = "futures"))] pub struct BackgroundProcessorConfig< 'a, @@ -1230,7 +1185,8 @@ pub struct BackgroundProcessorConfig< L: 'static + Deref + Send + Sync, P: 'static + Deref + Send + Sync, #[cfg(feature = "std")] EH: 'static + EventHandler + Send, - #[cfg(feature = "futures")] EH: 'static + Fn(Event) -> core::future::Future>, + #[cfg(feature = "futures")] EventHandlerFuture: core::future::Future>, + #[cfg(feature = "futures")] EH: 'static + Fn(Event) -> EventHandlerFuture, PS: 'static + Deref + Send, M: 'static + Deref::Signer, CF, T, F, L, P>> @@ -1272,7 +1228,7 @@ pub struct BackgroundProcessorConfig< /// This builder provides a flexible and type-safe way to construct a [`BackgroundProcessorConfig`] /// with optional components like `onion_messenger` and `scorer`. It helps avoid specifying /// concrete types for components that aren't being used. -#[cfg(feature = "std")] +#[cfg(any(feature = "std", feature = "futures"))] pub struct BackgroundProcessorConfigBuilder< 'a, UL: 'static + Deref + Send + Sync, @@ -1282,7 +1238,9 @@ pub struct BackgroundProcessorConfigBuilder< G: 'static + Deref> + Send + Sync, L: 'static + Deref + Send + Sync, P: 'static + Deref + Send + Sync, - EH: 'static + EventHandler + Send, + #[cfg(feature = "std")] EH: 'static + EventHandler + Send, + #[cfg(feature = "futures")] EventHandlerFuture: core::future::Future>, + #[cfg(feature = "futures")] EH: 'static + Fn(Event) -> EventHandlerFuture, PS: 'static + Deref + Send, M: 'static + Deref::Signer, CF, T, F, L, P>> @@ -1319,7 +1277,7 @@ pub struct BackgroundProcessorConfigBuilder< _phantom: PhantomData<(&'a (), CF, T, F, P)>, } -#[cfg(feature = "std")] +#[cfg(any(feature = "std", feature = "futures"))] impl< 'a, UL: 'static + Deref + Send + Sync, @@ -1329,7 +1287,9 @@ impl< G: 'static + Deref> + Send + Sync, L: 'static + Deref + Send + Sync, P: 'static + Deref + Send + Sync, - EH: 'static + EventHandler + Send, + #[cfg(feature = "std")] EH: 'static + EventHandler + Send, + #[cfg(feature = "futures")] EventHandlerFuture: core::future::Future>, + #[cfg(feature = "futures")] EH: 'static + Fn(Event) -> EventHandlerFuture, PS: 'static + Deref + Send, M: 'static + Deref::Signer, CF, T, F, L, P>> @@ -3196,7 +3156,7 @@ mod tests { // Check scorer is persisted let filepath = get_full_filepath(format!("{}_persister_0", &persist_dir), "scorer".to_string()); - check_persisted_data!(nodes[0].scorer, filepath); + check_persisted_data!(nodes[0].scorer, filepath.clone()); if !std::thread::panicking() { bg_processor.stop().unwrap();