Introduce Flow utilities and OffersMessageFlow implementation

lightning-background-processor/src/lib.rs

@@ -1083,11 +1083,14 @@ mod tests {
 		IgnoringMessageHandler, MessageHandler, PeerManager, SocketDescriptor,
 	};
 	use lightning::ln::types::ChannelId;
+	use lightning::offers::flow::OffersMessageFlow;
 	use lightning::onion_message::messenger::{DefaultMessageRouter, OnionMessenger};
 	use lightning::routing::gossip::{NetworkGraph, P2PGossipSync};
 	use lightning::routing::router::{CandidateRouteHop, DefaultRouter, Path, RouteHop};
 	use lightning::routing::scoring::{ChannelUsage, LockableScore, ScoreLookUp, ScoreUpdate};
-	use lightning::sign::{ChangeDestinationSource, InMemorySigner, KeysManager};
+	use lightning::sign::{
+		ChangeDestinationSource, InMemorySigner, KeysManager, NodeSigner, Recipient,
+	};
 	use lightning::types::features::{ChannelFeatures, NodeFeatures};
 	use lightning::types::payment::PaymentHash;
 	use lightning::util::config::UserConfig;
@@ -1152,6 +1155,11 @@ mod tests {
 				Arc<KeysManager>,
 			>,
 		>,
+		Arc<
+			OffersMessageFlow<
+				Arc<KeysManager>,
+			>,
+		>,
 		Arc<test_utils::TestLogger>,
 	>;
 
@@ -1559,6 +1567,13 @@ mod tests {
 				network_graph.clone(),
 				Arc::clone(&keys_manager),
 			));
+			let flow = Arc::new(OffersMessageFlow::new(
+				network,
+				keys_manager.get_node_id(Recipient::Node).unwrap(),
+				genesis_block.header.time,
+				keys_manager.get_inbound_payment_key(),
+				keys_manager.clone(),
+			));
 			let chain_source = Arc::new(test_utils::TestChainSource::new(Network::Bitcoin));
 			let kv_store =
 				Arc::new(FilesystemStore::new(format!("{}_persister_{}", &persist_dir, i).into()));
@@ -1579,6 +1594,7 @@ mod tests {
 				tx_broadcaster.clone(),
 				router.clone(),
 				msg_router.clone(),
+				flow.clone(),
 				logger.clone(),
 				keys_manager.clone(),
 				keys_manager.clone(),

lightning-block-sync/src/init.rs

@@ -50,6 +50,7 @@ where
 /// use lightning::chain::chaininterface::BroadcasterInterface;
 /// use lightning::chain::chaininterface::FeeEstimator;
 /// use lightning::ln::channelmanager::{ChannelManager, ChannelManagerReadArgs};
+/// use lightning::offers::flow::Flow;
 /// use lightning::onion_message::messenger::MessageRouter;
 /// use lightning::routing::router::Router;
 /// use lightning::sign;
@@ -71,6 +72,7 @@ where
 /// 	F: FeeEstimator,
 /// 	R: Router,
 /// 	MR: MessageRouter,
+/// 	FW: Flow,
 /// 	L: Logger,
 /// 	C: chain::Filter,
 /// 	P: chainmonitor::Persist<SP::EcdsaSigner>,
@@ -85,6 +87,7 @@ where
 /// 	fee_estimator: &F,
 /// 	router: &R,
 /// 	message_router: &MR,
+/// 	flow: &FW,
 /// 	logger: &L,
 /// 	persister: &P,
 /// ) {
@@ -105,11 +108,12 @@ where
 /// 			tx_broadcaster,
 /// 			router,
 /// 			message_router,
+/// 			flow,
 /// 			logger,
 /// 			config,
 /// 			vec![&mut monitor],
 /// 		);
-/// 		<(BlockHash, ChannelManager<&ChainMonitor<SP::EcdsaSigner, &C, &T, &F, &L, &P>, &T, &ES, &NS, &SP, &F, &R, &MR, &L>)>::read(
+/// 		<(BlockHash, ChannelManager<&ChainMonitor<SP::EcdsaSigner, &C, &T, &F, &L, &P>, &T, &ES, &NS, &SP, &F, &R, &MR, &FW, &L>)>::read(
 /// 			&mut Cursor::new(&serialized_manager), read_args).unwrap()
 /// 	};
 ///

lightning-liquidity/tests/common/mod.rs

@@ -5,7 +5,8 @@
 #![allow(unused_macros)]
 
 use lightning::chain::Filter;
-use lightning::sign::EntropySource;
+use lightning::offers::flow::OffersMessageFlow;
+use lightning::sign::{EntropySource, NodeSigner, Recipient};
 
 use bitcoin::blockdata::constants::{genesis_block, ChainHash};
 use bitcoin::blockdata::transaction::Transaction;
@@ -91,6 +92,11 @@ type ChannelManager = channelmanager::ChannelManager<
 			Arc<KeysManager>,
 		>,
 	>,
+	Arc<
+		OffersMessageFlow<
+			Arc<KeysManager>,
+		>,
+	>,
 	Arc<test_utils::TestLogger>,
 >;
 
@@ -421,6 +427,13 @@ pub(crate) fn create_liquidity_node(
 	));
 	let msg_router =
 		Arc::new(DefaultMessageRouter::new(Arc::clone(&network_graph), Arc::clone(&keys_manager)));
+	let flow = Arc::new(OffersMessageFlow::new(
+		network,
+		keys_manager.get_node_id(Recipient::Node).unwrap(),
+		genesis_block.header.time,
+		keys_manager.get_inbound_payment_key(),
+		keys_manager.clone(),
+	));
 	let chain_source = Arc::new(test_utils::TestChainSource::new(Network::Bitcoin));
 	let kv_store =
 		Arc::new(FilesystemStore::new(format!("{}_persister_{}", &persist_dir, i).into()));
@@ -439,6 +452,7 @@ pub(crate) fn create_liquidity_node(
 		tx_broadcaster.clone(),
 		router.clone(),
 		msg_router.clone(),
+		flow.clone(),
 		logger.clone(),
 		keys_manager.clone(),
 		keys_manager.clone(),

lightning/src/ln/functional_test_utils.rs

@@ -16,6 +16,7 @@ use crate::chain::transaction::OutPoint;
 use crate::events::{ClaimedHTLC, ClosureReason, Event, HTLCDestination, MessageSendEvent, MessageSendEventsProvider, PathFailure, PaymentPurpose, PaymentFailureReason};
 use crate::events::bump_transaction::{BumpTransactionEvent, BumpTransactionEventHandler, Wallet, WalletSource};
 use crate::ln::types::ChannelId;
+use crate::offers::flow::OffersMessageFlow;
 use crate::types::payment::{PaymentPreimage, PaymentHash, PaymentSecret};
 use crate::ln::channelmanager::{AChannelManager, ChainParameters, ChannelManager, ChannelManagerReadArgs, RAACommitmentOrder, RecipientOnionFields, PaymentId, MIN_CLTV_EXPIRY_DELTA};
 use crate::types::features::InitFeatures;
@@ -26,7 +27,7 @@ use crate::ln::peer_handler::IgnoringMessageHandler;
 use crate::onion_message::messenger::OnionMessenger;
 use crate::routing::gossip::{P2PGossipSync, NetworkGraph, NetworkUpdate};
 use crate::routing::router::{self, PaymentParameters, Route, RouteParameters};
-use crate::sign::{EntropySource, RandomBytes};
+use crate::sign::{EntropySource, NodeSigner, RandomBytes, Recipient};
 use crate::util::config::{MaxDustHTLCExposure, UserConfig};
 #[cfg(test)]
 use crate::util::logger::Logger;
@@ -408,6 +409,7 @@ type TestChannelManager<'node_cfg, 'chan_mon_cfg> = ChannelManager<
 	&'chan_mon_cfg test_utils::TestFeeEstimator,
 	&'node_cfg test_utils::TestRouter<'chan_mon_cfg>,
 	&'node_cfg test_utils::TestMessageRouter<'chan_mon_cfg>,
+	Arc<TestOffersMessageFlow<'node_cfg, 'chan_mon_cfg>>,
 	&'chan_mon_cfg test_utils::TestLogger,
 >;
 
@@ -437,6 +439,10 @@ type TestOnionMessenger<'chan_man, 'node_cfg, 'chan_mon_cfg> = OnionMessenger<
 	IgnoringMessageHandler,
 >;
 
+pub type TestOffersMessageFlow<'node_cfg, 'chan_mon_cfg> = OffersMessageFlow<
+	&'node_cfg test_utils::TestKeysInterface,
+>;
+
 /// For use with [`OnionMessenger`] otherwise `test_restored_packages_retry` will fail. This is
 /// because that test uses older serialized data produced by calling [`EntropySource`] in a specific
 /// manner. Using the same [`EntropySource`] with [`OnionMessenger`] would introduce another call,
@@ -454,6 +460,7 @@ pub struct Node<'chan_man, 'node_cfg: 'chan_man, 'chan_mon_cfg: 'node_cfg> {
 	pub fee_estimator: &'chan_mon_cfg test_utils::TestFeeEstimator,
 	pub router: &'node_cfg test_utils::TestRouter<'chan_mon_cfg>,
 	pub message_router: &'node_cfg test_utils::TestMessageRouter<'chan_mon_cfg>,
+	pub flow: Arc<TestOffersMessageFlow<'node_cfg, 'chan_mon_cfg,>>,
 	pub chain_monitor: &'node_cfg test_utils::TestChainMonitor<'chan_mon_cfg>,
 	pub keys_manager: &'chan_mon_cfg test_utils::TestKeysInterface,
 	pub node: &'chan_man TestChannelManager<'node_cfg, 'chan_mon_cfg>,
@@ -610,6 +617,7 @@ pub trait NodeHolder {
 		<Self::CM as AChannelManager>::F,
 		<Self::CM as AChannelManager>::R,
 		<Self::CM as AChannelManager>::MR,
+		<Self::CM as AChannelManager>::FW,
 		<Self::CM as AChannelManager>::L>;
 	fn chain_monitor(&self) -> Option<&test_utils::TestChainMonitor>;
 }
@@ -624,6 +632,7 @@ impl<H: NodeHolder> NodeHolder for &H {
 		<Self::CM as AChannelManager>::F,
 		<Self::CM as AChannelManager>::R,
 		<Self::CM as AChannelManager>::MR,
+		<Self::CM as AChannelManager>::FW,
 		<Self::CM as AChannelManager>::L> { (*self).node() }
 	fn chain_monitor(&self) -> Option<&test_utils::TestChainMonitor> { (*self).chain_monitor() }
 }
@@ -718,14 +727,15 @@ impl<'a, 'b, 'c> Drop for Node<'a, 'b, 'c> {
 				let scorer = RwLock::new(test_utils::TestScorer::new());
 				let mut w = test_utils::TestVecWriter(Vec::new());
 				self.node.write(&mut w).unwrap();
-				<(BlockHash, ChannelManager<&test_utils::TestChainMonitor, &test_utils::TestBroadcaster, &test_utils::TestKeysInterface, &test_utils::TestKeysInterface, &test_utils::TestKeysInterface, &test_utils::TestFeeEstimator, &test_utils::TestRouter, &test_utils::TestMessageRouter, &test_utils::TestLogger>)>::read(&mut io::Cursor::new(w.0), ChannelManagerReadArgs {
+				<(BlockHash, ChannelManager<&test_utils::TestChainMonitor, &test_utils::TestBroadcaster, &test_utils::TestKeysInterface, &test_utils::TestKeysInterface, &test_utils::TestKeysInterface, &test_utils::TestFeeEstimator, &test_utils::TestRouter, &test_utils::TestMessageRouter, Arc<TestOffersMessageFlow>, &test_utils::TestLogger>)>::read(&mut io::Cursor::new(w.0), ChannelManagerReadArgs {
 					default_config: *self.node.get_current_default_configuration(),
 					entropy_source: self.keys_manager,
 					node_signer: self.keys_manager,
 					signer_provider: self.keys_manager,
 					fee_estimator: &test_utils::TestFeeEstimator::new(253),
 					router: &test_utils::TestRouter::new(Arc::clone(&network_graph), &self.logger, &scorer),
 					message_router: &test_utils::TestMessageRouter::new(network_graph, self.keys_manager),
+					flow: self.flow.clone(),
 					chain_monitor: self.chain_monitor,
 					tx_broadcaster: &broadcaster,
 					logger: &self.logger,
@@ -1154,6 +1164,7 @@ pub fn _reload_node<'a, 'b, 'c>(node: &'a Node<'a, 'b, 'c>, default_config: User
 			fee_estimator: node.fee_estimator,
 			router: node.router,
 			message_router: node.message_router,
+			flow: node.flow.clone(),
 			chain_monitor: node.chain_monitor,
 			tx_broadcaster: node.tx_broadcaster,
 			logger: node.logger,
@@ -3332,7 +3343,7 @@ pub fn test_default_channel_config() -> UserConfig {
 	default_config
 }
 
-pub fn create_node_chanmgrs<'a, 'b>(node_count: usize, cfgs: &'a Vec<NodeCfg<'b>>, node_config: &[Option<UserConfig>]) -> Vec<ChannelManager<&'a TestChainMonitor<'b>, &'b test_utils::TestBroadcaster, &'a test_utils::TestKeysInterface, &'a test_utils::TestKeysInterface, &'a test_utils::TestKeysInterface, &'b test_utils::TestFeeEstimator, &'a test_utils::TestRouter<'b>, &'a test_utils::TestMessageRouter<'b>, &'b test_utils::TestLogger>> {
+pub fn create_node_chanmgrs<'a, 'b>(node_count: usize, cfgs: &'a Vec<NodeCfg<'b>>, node_config: &[Option<UserConfig>]) -> Vec<ChannelManager<&'a TestChainMonitor<'b>, &'b test_utils::TestBroadcaster, &'a test_utils::TestKeysInterface, &'a test_utils::TestKeysInterface, &'a test_utils::TestKeysInterface, &'b test_utils::TestFeeEstimator, &'a test_utils::TestRouter<'b>, &'a test_utils::TestMessageRouter<'b>, Arc<TestOffersMessageFlow<'a, 'b>>, &'b test_utils::TestLogger>> {
 	let mut chanmgrs = Vec::new();
 	for i in 0..node_count {
 		let network = Network::Testnet;
@@ -3341,20 +3352,24 @@ pub fn create_node_chanmgrs<'a, 'b>(node_count: usize, cfgs: &'a Vec<NodeCfg<'b>
 			network,
 			best_block: BestBlock::from_network(network),
 		};
-		let node = ChannelManager::new(cfgs[i].fee_estimator, &cfgs[i].chain_monitor, cfgs[i].tx_broadcaster, &cfgs[i].router, &cfgs[i].message_router, cfgs[i].logger, cfgs[i].keys_manager,
+		let flow = Arc::new(TestOffersMessageFlow::new(network, cfgs[i].keys_manager.get_node_id(Recipient::Node).unwrap(), genesis_block.header.time, cfgs[i].keys_manager.get_inbound_payment_key(), cfgs[i].keys_manager));
+		let node = ChannelManager::new(cfgs[i].fee_estimator, &cfgs[i].chain_monitor, cfgs[i].tx_broadcaster, &cfgs[i].router, &cfgs[i].message_router, flow, cfgs[i].logger, cfgs[i].keys_manager,
 			cfgs[i].keys_manager, cfgs[i].keys_manager, if node_config[i].is_some() { node_config[i].clone().unwrap() } else { test_default_channel_config() }, params, genesis_block.header.time);
 		chanmgrs.push(node);
 	}
 
 	chanmgrs
 }
 
-pub fn create_network<'a, 'b: 'a, 'c: 'b>(node_count: usize, cfgs: &'b Vec<NodeCfg<'c>>, chan_mgrs: &'a Vec<ChannelManager<&'b TestChainMonitor<'c>, &'c test_utils::TestBroadcaster, &'b test_utils::TestKeysInterface, &'b test_utils::TestKeysInterface, &'b test_utils::TestKeysInterface, &'c test_utils::TestFeeEstimator, &'c test_utils::TestRouter, &'c test_utils::TestMessageRouter, &'c test_utils::TestLogger>>) -> Vec<Node<'a, 'b, 'c>> {
+pub fn create_network<'a, 'b: 'a, 'c: 'b>(node_count: usize, cfgs: &'b Vec<NodeCfg<'c>>, chan_mgrs: &'a Vec<ChannelManager<&'b TestChainMonitor<'c>, &'c test_utils::TestBroadcaster, &'b test_utils::TestKeysInterface, &'b test_utils::TestKeysInterface, &'b test_utils::TestKeysInterface, &'c test_utils::TestFeeEstimator, &'c test_utils::TestRouter, &'c test_utils::TestMessageRouter, Arc<TestOffersMessageFlow<'b, 'c>>, &'c test_utils::TestLogger>>) -> Vec<Node<'a, 'b, 'c>> {
 	let mut nodes = Vec::new();
 	let chan_count = Rc::new(RefCell::new(0));
 	let payment_count = Rc::new(RefCell::new(0));
 	let connect_style = Rc::new(RefCell::new(ConnectStyle::random_style()));
 
+	let network = Network::Testnet;
+	let genesis_block = bitcoin::constants::genesis_block(network);
+
 	for i in 0..node_count {
 		let dedicated_entropy = DedicatedEntropy(RandomBytes::new([i as u8; 32]));
 		#[cfg(feature = "dnssec")]
@@ -3369,12 +3384,16 @@ pub fn create_network<'a, 'b: 'a, 'c: 'b>(node_count: usize, cfgs: &'b Vec<NodeC
 			&cfgs[i].message_router, &chan_mgrs[i], &chan_mgrs[i], IgnoringMessageHandler {},
 			IgnoringMessageHandler {},
 		);
+
+		let flow = Arc::new(TestOffersMessageFlow::new(network, cfgs[i].keys_manager.get_node_id(Recipient::Node).unwrap(), genesis_block.header.time, cfgs[i].keys_manager.get_inbound_payment_key(), &cfgs[i].keys_manager));
+
 		let gossip_sync = P2PGossipSync::new(cfgs[i].network_graph.as_ref(), None, cfgs[i].logger);
 		let wallet_source = Arc::new(test_utils::TestWalletSource::new(SecretKey::from_slice(&[i as u8 + 1; 32]).unwrap()));
 		nodes.push(Node{
 			chain_source: cfgs[i].chain_source, tx_broadcaster: cfgs[i].tx_broadcaster,
 			fee_estimator: cfgs[i].fee_estimator, router: &cfgs[i].router,
-			message_router: &cfgs[i].message_router, chain_monitor: &cfgs[i].chain_monitor,
+			message_router: &cfgs[i].message_router, flow,
+			chain_monitor: &cfgs[i].chain_monitor,
 			keys_manager: &cfgs[i].keys_manager, node: &chan_mgrs[i],
 			network_graph: cfgs[i].network_graph.as_ref(), gossip_sync,
 			node_seed: cfgs[i].node_seed, onion_messenger, network_chan_count: chan_count.clone(),

lightning/src/ln/inbound_payment.rs

@@ -194,7 +194,7 @@ pub fn create_from_hash(keys: &ExpandedKey, min_value_msat: Option<u64>, payment
 }
 
 #[cfg(async_payments)]
-pub(super) fn create_for_spontaneous_payment(
+pub(crate) fn create_for_spontaneous_payment(
 	keys: &ExpandedKey, min_value_msat: Option<u64>, invoice_expiry_delta_secs: u32,
 	current_time: u64, min_final_cltv_expiry_delta: Option<u16>
 ) -> Result<PaymentSecret, ()> {
@@ -213,7 +213,7 @@ pub(super) fn create_for_spontaneous_payment(
 	Ok(construct_payment_secret(&iv_bytes, &metadata_bytes, &keys.metadata_key))
 }
 
-pub(super) fn calculate_absolute_expiry(highest_seen_timestamp: u64, invoice_expiry_delta_secs: u32) -> u64 {
+pub(crate) fn calculate_absolute_expiry(highest_seen_timestamp: u64, invoice_expiry_delta_secs: u32) -> u64 {
 	// We assume that highest_seen_timestamp is pretty close to the current time - it's updated when
 	// we receive a new block with the maximum time we've seen in a header. It should never be more
 	// than two hours in the future.  Thus, we add two hours here as a buffer to ensure we

lightning/src/ln/invoice_utils.rs

@@ -3,6 +3,7 @@
 use lightning_invoice::{Bolt11Invoice, CreationError, Currency, InvoiceBuilder, SignOrCreationError};
 use lightning_invoice::{Description, Bolt11InvoiceDescription, Sha256};
 
+use crate::offers::flow::Flow;
 use crate::prelude::*;
 
 use bitcoin::hashes::Hash;
@@ -328,8 +329,8 @@ fn rotate_through_iterators<T, I: Iterator<Item = T>>(mut vecs: Vec<I>) -> impl
 /// confirmations during routing.
 ///
 /// [`MIN_FINAL_CLTV_EXPIRY_DETLA`]: crate::ln::channelmanager::MIN_FINAL_CLTV_EXPIRY_DELTA
-pub fn create_invoice_from_channelmanager<M: Deref, T: Deref, ES: Deref, NS: Deref, SP: Deref, F: Deref, R: Deref, MR: Deref, L: Deref>(
-	channelmanager: &ChannelManager<M, T, ES, NS, SP, F, R, MR, L>, amt_msat: Option<u64>,
+pub fn create_invoice_from_channelmanager<M: Deref, T: Deref, ES: Deref, NS: Deref, SP: Deref, F: Deref, R: Deref, MR: Deref, FW: Deref, L: Deref>(
+	channelmanager: &ChannelManager<M, T, ES, NS, SP, F, R, MR, FW, L>, amt_msat: Option<u64>,
 	description: String, invoice_expiry_delta_secs: u32, min_final_cltv_expiry_delta: Option<u16>,
 ) -> Result<Bolt11Invoice, SignOrCreationError<()>>
 where
@@ -341,6 +342,7 @@ where
 	F::Target: FeeEstimator,
 	R::Target: Router,
 	MR::Target: MessageRouter,
+	FW::Target: Flow,
 	L::Target: Logger,
 {
 	let description = Description::new(description).map_err(SignOrCreationError::CreationError)?;
@@ -371,8 +373,8 @@ where
 /// confirmations during routing.
 ///
 /// [`MIN_FINAL_CLTV_EXPIRY_DETLA`]: crate::ln::channelmanager::MIN_FINAL_CLTV_EXPIRY_DELTA
-pub fn create_invoice_from_channelmanager_with_description_hash<M: Deref, T: Deref, ES: Deref, NS: Deref, SP: Deref, F: Deref, R: Deref, MR: Deref, L: Deref>(
-	channelmanager: &ChannelManager<M, T, ES, NS, SP, F, R, MR, L>, amt_msat: Option<u64>,
+pub fn create_invoice_from_channelmanager_with_description_hash<M: Deref, T: Deref, ES: Deref, NS: Deref, SP: Deref, F: Deref, R: Deref, MR: Deref, FW: Deref, L: Deref>(
+	channelmanager: &ChannelManager<M, T, ES, NS, SP, F, R, MR, FW, L>, amt_msat: Option<u64>,
 	description_hash: Sha256, invoice_expiry_delta_secs: u32,
 	min_final_cltv_expiry_delta: Option<u16>,
 ) -> Result<Bolt11Invoice, SignOrCreationError<()>>
@@ -385,6 +387,7 @@ where
 	F::Target: FeeEstimator,
 	R::Target: Router,
 	MR::Target: MessageRouter,
+	FW::Target: Flow,
 	L::Target: Logger,
 {
 	let params = Bolt11InvoiceParameters {
@@ -405,8 +408,8 @@ where
 /// This may be useful if you're building an on-chain swap or involving another protocol where
 /// the payment hash is also involved outside the scope of lightning and want to set the
 /// description hash.
-pub fn create_invoice_from_channelmanager_with_description_hash_and_payment_hash<M: Deref, T: Deref, ES: Deref, NS: Deref, SP: Deref, F: Deref, R: Deref, MR: Deref, L: Deref>(
-	channelmanager: &ChannelManager<M, T, ES, NS, SP, F, R, MR, L>, amt_msat: Option<u64>,
+pub fn create_invoice_from_channelmanager_with_description_hash_and_payment_hash<M: Deref, T: Deref, ES: Deref, NS: Deref, SP: Deref, F: Deref, R: Deref, MR: Deref, FW: Deref, L: Deref>(
+	channelmanager: &ChannelManager<M, T, ES, NS, SP, F, R, MR, FW, L>, amt_msat: Option<u64>,
 	description_hash: Sha256, invoice_expiry_delta_secs: u32, payment_hash: PaymentHash,
 	min_final_cltv_expiry_delta: Option<u16>,
 ) -> Result<Bolt11Invoice, SignOrCreationError<()>>
@@ -419,6 +422,7 @@ where
 	F::Target: FeeEstimator,
 	R::Target: Router,
 	MR::Target: MessageRouter,
+	FW::Target: Flow,
 	L::Target: Logger,
 {
 	let params = Bolt11InvoiceParameters {
@@ -437,8 +441,8 @@ where
 /// This version allows for providing a custom [`PaymentHash`] for the invoice.
 /// This may be useful if you're building an on-chain swap or involving another protocol where
 /// the payment hash is also involved outside the scope of lightning.
-pub fn create_invoice_from_channelmanager_with_payment_hash<M: Deref, T: Deref, ES: Deref, NS: Deref, SP: Deref, F: Deref, R: Deref, MR: Deref, L: Deref>(
-	channelmanager: &ChannelManager<M, T, ES, NS, SP, F, R, MR, L>, amt_msat: Option<u64>,
+pub fn create_invoice_from_channelmanager_with_payment_hash<M: Deref, T: Deref, ES: Deref, NS: Deref, SP: Deref, F: Deref, R: Deref, MR: Deref, FW: Deref, L: Deref>(
+	channelmanager: &ChannelManager<M, T, ES, NS, SP, F, R, MR, FW, L>, amt_msat: Option<u64>,
 	description: String, invoice_expiry_delta_secs: u32, payment_hash: PaymentHash,
 	min_final_cltv_expiry_delta: Option<u16>,
 ) -> Result<Bolt11Invoice, SignOrCreationError<()>>
@@ -451,6 +455,7 @@ where
 	F::Target: FeeEstimator,
 	R::Target: Router,
 	MR::Target: MessageRouter,
+	FW::Target: Flow,
 	L::Target: Logger,
 {
 	let description = Description::new(description).map_err(SignOrCreationError::CreationError)?;

lightning/src/ln/offers_tests.rs

@@ -1402,7 +1402,7 @@ fn fails_authentication_when_handling_invoice_request() {
 	expect_recent_payment!(david, RecentPaymentDetails::AwaitingInvoice, payment_id);
 
 	connect_peers(david, alice);
-	match &mut david.node.pending_offers_messages.lock().unwrap().first_mut().unwrap().1 {
+	match &mut david.node.flow.pending_offers_messages.lock().unwrap().first_mut().unwrap().1 {
 		MessageSendInstructions::WithSpecifiedReplyPath { destination, .. } =>
 			*destination = Destination::Node(alice_id),
 		_ => panic!(),
@@ -1427,7 +1427,7 @@ fn fails_authentication_when_handling_invoice_request() {
 		.unwrap();
 	expect_recent_payment!(david, RecentPaymentDetails::AwaitingInvoice, payment_id);
 
-	match &mut david.node.pending_offers_messages.lock().unwrap().first_mut().unwrap().1 {
+	match &mut david.node.flow.pending_offers_messages.lock().unwrap().first_mut().unwrap().1 {
 		MessageSendInstructions::WithSpecifiedReplyPath { destination, .. } =>
 			*destination = Destination::BlindedPath(invalid_path),
 		_ => panic!(),
@@ -1507,7 +1507,7 @@ fn fails_authentication_when_handling_invoice_for_offer() {
 
 	// Don't send the invoice request, but grab its reply path to use with a different request.
 	let invalid_reply_path = {
-		let mut pending_offers_messages = david.node.pending_offers_messages.lock().unwrap();
+		let mut pending_offers_messages = david.node.flow.pending_offers_messages.lock().unwrap();
 		let pending_invoice_request = pending_offers_messages.pop().unwrap();
 		pending_offers_messages.clear();
 		match pending_invoice_request.1 {
@@ -1524,7 +1524,7 @@ fn fails_authentication_when_handling_invoice_for_offer() {
 	// Swap out the reply path to force authentication to fail when handling the invoice since it
 	// will be sent over the wrong blinded path.
 	{
-		let mut pending_offers_messages = david.node.pending_offers_messages.lock().unwrap();
+		let mut pending_offers_messages = david.node.flow.pending_offers_messages.lock().unwrap();
 		let mut pending_invoice_request = pending_offers_messages.first_mut().unwrap();
 		match &mut pending_invoice_request.1 {
 			MessageSendInstructions::WithSpecifiedReplyPath { reply_path, .. } =>
@@ -1611,7 +1611,7 @@ fn fails_authentication_when_handling_invoice_for_refund() {
 	let expected_invoice = alice.node.request_refund_payment(&refund).unwrap();
 
 	connect_peers(david, alice);
-	match &mut alice.node.pending_offers_messages.lock().unwrap().first_mut().unwrap().1 {
+	match &mut alice.node.flow.pending_offers_messages.lock().unwrap().first_mut().unwrap().1 {
 		MessageSendInstructions::WithSpecifiedReplyPath { destination, .. } =>
 			*destination = Destination::Node(david_id),
 		_ => panic!(),
@@ -1642,7 +1642,7 @@ fn fails_authentication_when_handling_invoice_for_refund() {
 
 	let expected_invoice = alice.node.request_refund_payment(&refund).unwrap();
 
-	match &mut alice.node.pending_offers_messages.lock().unwrap().first_mut().unwrap().1 {
+	match &mut alice.node.flow.pending_offers_messages.lock().unwrap().first_mut().unwrap().1 {
 		MessageSendInstructions::WithSpecifiedReplyPath { destination, .. } =>
 			*destination = Destination::BlindedPath(invalid_path),
 		_ => panic!(),
@@ -2233,7 +2233,7 @@ fn fails_paying_invoice_with_unknown_required_features() {
 		destination: Destination::BlindedPath(reply_path),
 	};
 	let message = OffersMessage::Invoice(invoice);
-	alice.node.pending_offers_messages.lock().unwrap().push((message, instructions));
+	alice.node.flow.pending_offers_messages.lock().unwrap().push((message, instructions));
 
 	let onion_message = alice.onion_messenger.next_onion_message_for_peer(charlie_id).unwrap();
 	charlie.onion_messenger.handle_onion_message(alice_id, &onion_message);

lightning/src/ln/reload_tests.rs

@@ -9,6 +9,8 @@
 
 //! Functional tests which test for correct behavior across node restarts.
 
+use std::sync::Arc;
+
 use crate::chain::{ChannelMonitorUpdateStatus, Watch};
 use crate::chain::chaininterface::LowerBoundedFeeEstimator;
 use crate::chain::channelmonitor::{ChannelMonitor, ChannelMonitorUpdateStep};
@@ -421,14 +423,15 @@ fn test_manager_serialize_deserialize_inconsistent_monitor() {
 
 	let mut nodes_0_read = &nodes_0_serialized[..];
 	if let Err(msgs::DecodeError::DangerousValue) =
-		<(BlockHash, ChannelManager<&test_utils::TestChainMonitor, &test_utils::TestBroadcaster, &test_utils::TestKeysInterface, &test_utils::TestKeysInterface, &test_utils::TestKeysInterface, &test_utils::TestFeeEstimator, &test_utils::TestRouter, &test_utils::TestMessageRouter, &test_utils::TestLogger>)>::read(&mut nodes_0_read, ChannelManagerReadArgs {
+		<(BlockHash, ChannelManager<&test_utils::TestChainMonitor, &test_utils::TestBroadcaster, &test_utils::TestKeysInterface, &test_utils::TestKeysInterface, &test_utils::TestKeysInterface, &test_utils::TestFeeEstimator, &test_utils::TestRouter, &test_utils::TestMessageRouter, Arc<TestOffersMessageFlow>, &test_utils::TestLogger>)>::read(&mut nodes_0_read, ChannelManagerReadArgs {
 		default_config: UserConfig::default(),
 		entropy_source: keys_manager,
 		node_signer: keys_manager,
 		signer_provider: keys_manager,
 		fee_estimator: &fee_estimator,
 		router: &nodes[0].router,
 		message_router: &nodes[0].message_router,
+		flow: nodes[0].flow.clone(),
 		chain_monitor: nodes[0].chain_monitor,
 		tx_broadcaster: nodes[0].tx_broadcaster,
 		logger: &logger,
@@ -439,14 +442,15 @@ fn test_manager_serialize_deserialize_inconsistent_monitor() {
 
 	let mut nodes_0_read = &nodes_0_serialized[..];
 	let (_, nodes_0_deserialized_tmp) =
-		<(BlockHash, ChannelManager<&test_utils::TestChainMonitor, &test_utils::TestBroadcaster, &test_utils::TestKeysInterface, &test_utils::TestKeysInterface, &test_utils::TestKeysInterface, &test_utils::TestFeeEstimator, &test_utils::TestRouter, &test_utils::TestMessageRouter, &test_utils::TestLogger>)>::read(&mut nodes_0_read, ChannelManagerReadArgs {
+		<(BlockHash, ChannelManager<&test_utils::TestChainMonitor, &test_utils::TestBroadcaster, &test_utils::TestKeysInterface, &test_utils::TestKeysInterface, &test_utils::TestKeysInterface, &test_utils::TestFeeEstimator, &test_utils::TestRouter, &test_utils::TestMessageRouter, Arc<TestOffersMessageFlow>, &test_utils::TestLogger>)>::read(&mut nodes_0_read, ChannelManagerReadArgs {
 		default_config: UserConfig::default(),
 		entropy_source: keys_manager,
 		node_signer: keys_manager,
 		signer_provider: keys_manager,
 		fee_estimator: &fee_estimator,
 		router: nodes[0].router,
 		message_router: &nodes[0].message_router,
+		flow: nodes[0].flow.clone(),
 		chain_monitor: nodes[0].chain_monitor,
 		tx_broadcaster: nodes[0].tx_broadcaster,
 		logger: &logger,

lightning/src/offers/mod.rs

@@ -14,6 +14,7 @@
 
 #[macro_use]
 pub mod offer;
+pub mod flow;
 
 pub mod invoice;
 pub mod invoice_error;