Introduce Flow parameter in ChannelManager

shaavan · shaavan · commit 3d6f0683c9d0 · 2025-04-24T19:52:40.000+05:30
diff --git a/lightning-background-processor/src/lib.rs b/lightning-background-processor/src/lib.rs
@@ -1081,11 +1081,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;
@@ -1557,6 +1560,21 @@ mod tests {
 				network_graph.clone(),
 				Arc::clone(&keys_manager),
 			));
+			let best_block = BestBlock::from_network(network);
+			let params = ChainParameters { network, best_block };
+			let chain_hash = ChainHash::using_genesis_block(params.network);
+
+			let flow = OffersMessageFlow::new(
+				chain_hash,
+				params.best_block,
+				keys_manager.get_node_id(Recipient::Node).unwrap(),
+				genesis_block.header.time,
+				keys_manager.get_inbound_payment_key(),
+				keys_manager.clone(),
+				msg_router.clone(),
+				router.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()));
@@ -1569,14 +1587,14 @@ mod tests {
 				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(),
 				msg_router.clone(),
+				flow,
 				logger.clone(),
 				keys_manager.clone(),
 				keys_manager.clone(),
diff --git a/lightning-liquidity/tests/common/mod.rs b/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;
@@ -421,6 +422,22 @@ pub(crate) fn create_liquidity_node(
 	));
 	let msg_router =
 		Arc::new(DefaultMessageRouter::new(Arc::clone(&network_graph), Arc::clone(&keys_manager)));
+
+	let best_block = BestBlock::from_network(network);
+	let chain_params = ChainParameters { network, best_block };
+	let chain_hash = ChainHash::using_genesis_block(chain_params.network);
+
+	let flow = OffersMessageFlow::new(
+		chain_hash,
+		chain_params.best_block,
+		keys_manager.get_node_id(Recipient::Node).unwrap(),
+		genesis_block.header.time,
+		keys_manager.get_inbound_payment_key(),
+		keys_manager.clone(),
+		msg_router.clone(),
+		router.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()));
@@ -431,14 +448,13 @@ pub(crate) fn create_liquidity_node(
 		fee_estimator.clone(),
 		kv_store.clone(),
 	));
-	let best_block = BestBlock::from_network(network);
-	let chain_params = ChainParameters { network, best_block };
 	let channel_manager = Arc::new(ChannelManager::new(
 		fee_estimator.clone(),
 		chain_monitor.clone(),
 		tx_broadcaster.clone(),
 		router.clone(),
 		msg_router.clone(),
+		flow,
 		logger.clone(),
 		keys_manager.clone(),
 		keys_manager.clone(),
diff --git a/lightning/src/ln/channelmanager.rs b/lightning/src/ln/channelmanager.rs
@@ -49,6 +49,7 @@ use crate::events::{self, Event, EventHandler, EventsProvider, InboundChannelFun
 // construct one themselves.
 use crate::ln::inbound_payment;
 use crate::ln::types::ChannelId;
+use crate::offers::flow::OffersMessageFlow;
 use crate::types::payment::{PaymentHash, PaymentPreimage, PaymentSecret};
 use crate::ln::channel::{self, Channel, ChannelError, ChannelUpdateStatus, FundedChannel, ShutdownResult, UpdateFulfillCommitFetch, OutboundV1Channel, ReconnectionMsg, InboundV1Channel, WithChannelContext};
 use crate::ln::channel::PendingV2Channel;
@@ -1718,6 +1719,7 @@ where
 /// #     tx_broadcaster: &dyn lightning::chain::chaininterface::BroadcasterInterface,
 /// #     router: &lightning::routing::router::DefaultRouter<&NetworkGraph<&'a L>, &'a L, &ES, &S, SP, SL>,
 /// #     message_router: &lightning::onion_message::messenger::DefaultMessageRouter<&NetworkGraph<&'a L>, &'a L, &ES>,
+/// 	  flow: lightning::offers::flow::OffersMessageFlow<&ES,&lightning::onion_message::messenger::DefaultMessageRouter<&NetworkGraph<&'a L>, &'a L, &ES>, &lightning::routing::router::DefaultRouter<&NetworkGraph<&'a L>, &'a L, &ES, &S, SP, SL>>,
 /// #     logger: &L,
 /// #     entropy_source: &ES,
 /// #     node_signer: &dyn lightning::sign::NodeSigner,
@@ -1733,7 +1735,7 @@ where
 /// };
 /// let default_config = UserConfig::default();
 /// let channel_manager = ChannelManager::new(
-///     fee_estimator, chain_monitor, tx_broadcaster, router, message_router, logger,
+///     fee_estimator, chain_monitor, tx_broadcaster, router, message_router, flow, logger,
 ///     entropy_source, node_signer, signer_provider, default_config.clone(), params, current_timestamp,
 /// );
 ///
@@ -2445,6 +2447,11 @@ where
 	router: R,
 	message_router: MR,
 
+	#[cfg(test)]
+	pub(super) flow: OffersMessageFlow<ES, MR, R>,
+	#[cfg(not(test))]
+	flow: OffersMessageFlow<ES, MR, R>,
+
 	/// See `ChannelManager` struct-level documentation for lock order requirements.
 	#[cfg(any(test, feature = "_test_utils"))]
 	pub(super) best_block: RwLock<BestBlock>,
@@ -3540,8 +3547,8 @@ where
 	/// [`block_disconnected`]: chain::Listen::block_disconnected
 	/// [`params.best_block.block_hash`]: chain::BestBlock::block_hash
 	pub fn new(
-		fee_est: F, chain_monitor: M, tx_broadcaster: T, router: R, message_router: MR, logger: L,
-		entropy_source: ES, node_signer: NS, signer_provider: SP, config: UserConfig,
+		fee_est: F, chain_monitor: M, tx_broadcaster: T, router: R, message_router: MR, flow: OffersMessageFlow<ES, MR, R>,
+		logger: L, entropy_source: ES, node_signer: NS, signer_provider: SP, config: UserConfig,
 		params: ChainParameters, current_timestamp: u32,
 	) -> Self {
 		let mut secp_ctx = Secp256k1::new();
@@ -3555,6 +3562,7 @@ where
 			tx_broadcaster,
 			router,
 			message_router,
+			flow,
 
 			best_block: RwLock::new(params.best_block),
 
@@ -11499,6 +11507,7 @@ where
 
 		self.transactions_confirmed(header, txdata, height);
 		self.best_block_updated(header, height);
+		self.flow.best_block_updated(header);
 	}
 
 	fn block_disconnected(&self, header: &Header, height: u32) {
@@ -13602,7 +13611,7 @@ impl Readable for VecDeque<(Event, Option<EventCompletionAction>)> {
 /// which you've already broadcasted the transaction.
 ///
 /// [`ChainMonitor`]: crate::chain::chainmonitor::ChainMonitor
-pub struct ChannelManagerReadArgs<'a, M: Deref, T: Deref, ES: Deref, NS: Deref, SP: Deref, F: Deref, R: Deref, MR: Deref, L: Deref>
+pub struct ChannelManagerReadArgs<'a, M: Deref, T: Deref, ES: Deref + Clone, NS: Deref, SP: Deref, F: Deref, R: Deref + Clone, MR: Deref + Clone, L: Deref>
 where
 	M::Target: chain::Watch<<SP::Target as SignerProvider>::EcdsaSigner>,
 	T::Target: BroadcasterInterface,
@@ -13648,6 +13657,7 @@ where
 	/// The [`MessageRouter`] used for constructing [`BlindedMessagePath`]s for [`Offer`]s,
 	/// [`Refund`]s, and any reply paths.
 	pub message_router: MR,
+
 	/// The Logger for use in the ChannelManager and which may be used to log information during
 	/// deserialization.
 	pub logger: L,
@@ -13669,7 +13679,7 @@ where
 	pub channel_monitors: HashMap<ChannelId, &'a ChannelMonitor<<SP::Target as SignerProvider>::EcdsaSigner>>,
 }
 
-impl<'a, M: Deref, T: Deref, ES: Deref, NS: Deref, SP: Deref, F: Deref, R: Deref, MR: Deref, L: Deref>
+impl<'a, M: Deref, T: Deref, ES: Deref + Clone, NS: Deref, SP: Deref, F: Deref, R: Deref + Clone, MR: Deref + Clone, L: Deref>
 		ChannelManagerReadArgs<'a, M, T, ES, NS, SP, F, R, MR, L>
 where
 	M::Target: chain::Watch<<SP::Target as SignerProvider>::EcdsaSigner>,
@@ -13703,7 +13713,7 @@ where
 
 // Implement ReadableArgs for an Arc'd ChannelManager to make it a bit easier to work with the
 // SipmleArcChannelManager type:
-impl<'a, M: Deref, T: Deref, ES: Deref, NS: Deref, SP: Deref, F: Deref, R: Deref, MR: Deref, L: Deref>
+impl<'a, M: Deref, T: Deref, ES: Deref + Clone, NS: Deref, SP: Deref, F: Deref, R: Deref + Clone, MR: Deref + Clone, L: Deref>
 	ReadableArgs<ChannelManagerReadArgs<'a, M, T, ES, NS, SP, F, R, MR, L>> for (BlockHash, Arc<ChannelManager<M, T, ES, NS, SP, F, R, MR, L>>)
 where
 	M::Target: chain::Watch<<SP::Target as SignerProvider>::EcdsaSigner>,
@@ -13722,7 +13732,7 @@ where
 	}
 }
 
-impl<'a, M: Deref, T: Deref, ES: Deref, NS: Deref, SP: Deref, F: Deref, R: Deref, MR: Deref, L: Deref>
+impl<'a, M: Deref, T: Deref, ES: Deref + Clone, NS: Deref, SP: Deref, F: Deref, R: Deref + Clone, MR: Deref + Clone, L: Deref>
 	ReadableArgs<ChannelManagerReadArgs<'a, M, T, ES, NS, SP, F, R, MR, L>> for (BlockHash, ChannelManager<M, T, ES, NS, SP, F, R, MR, L>)
 where
 	M::Target: chain::Watch<<SP::Target as SignerProvider>::EcdsaSigner>,
@@ -14704,15 +14714,19 @@ where
 			}
 		}
 
+		let best_block = BestBlock::new(best_block_hash, best_block_height);
+		let flow = OffersMessageFlow::new(chain_hash, best_block, our_network_pubkey, highest_seen_timestamp, expanded_inbound_key, args.entropy_source.clone(), args.message_router.clone(), args.router.clone());
+
 		let channel_manager = ChannelManager {
 			chain_hash,
 			fee_estimator: bounded_fee_estimator,
 			chain_monitor: args.chain_monitor,
 			tx_broadcaster: args.tx_broadcaster,
 			router: args.router,
 			message_router: args.message_router,
+			flow,
 
-			best_block: RwLock::new(BestBlock::new(best_block_hash, best_block_height)),
+			best_block: RwLock::new(best_block),
 
 			inbound_payment_key: expanded_inbound_key,
 			pending_outbound_payments: pending_outbounds,
diff --git a/lightning/src/ln/functional_test_utils.rs b/lightning/src/ln/functional_test_utils.rs
@@ -16,6 +16,7 @@ use crate::chain::transaction::OutPoint;
 use crate::events::{ClaimedHTLC, ClosureReason, Event, HTLCDestination, PaidBolt12Invoice, PathFailure, PaymentFailureReason, PaymentPurpose};
 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;
@@ -27,7 +28,7 @@ use crate::onion_message::messenger::OnionMessenger;
 use crate::ln::onion_utils::LocalHTLCFailureReason;
 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};
 use crate::util::logger::Logger;
 use crate::util::scid_utils;
@@ -37,6 +38,7 @@ use crate::util::test_utils;
 use crate::util::test_utils::{TestChainMonitor, TestScorer, TestKeysInterface};
 use crate::util::ser::{ReadableArgs, Writeable};
 
+use bitcoin::constants::ChainHash;
 use bitcoin::{Weight, WPubkeyHash};
 use bitcoin::amount::Amount;
 use bitcoin::block::{Block, Header, Version as BlockVersion};
@@ -3383,7 +3385,17 @@ 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 chain_hash = ChainHash::using_genesis_block(params.network);
+		let our_network_pubkey = cfgs[i].keys_manager.get_node_id(Recipient::Node).unwrap();
+		let expanded_inbound_key = cfgs[i].keys_manager.get_inbound_payment_key();
+
+		let flow = OffersMessageFlow::new(
+			chain_hash, params.best_block, our_network_pubkey,
+			genesis_block.header.time, expanded_inbound_key,
+			cfgs[i].keys_manager, &cfgs[i].message_router, &cfgs[i].router
+		);
+
+		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);
 	}
diff --git a/lightning/src/offers/flow.rs b/lightning/src/offers/flow.rs
@@ -162,7 +162,7 @@ where
 		now
 	}
 
-	fn best_block_updated(&self, header: &Header) {
+	pub(crate) fn best_block_updated(&self, header: &Header) {
 		macro_rules! max_time {
 			($timestamp: expr) => {
 				loop {

Original file line number	Diff line number	Diff line change
`@@ -162,7 +162,7 @@ where`
`162`	`162`	`now`
`163`	`163`	`}`
`164`	`164`
`165`		`- fn best_block_updated(&self, header: &Header) {`
	`165`	`+ pub(crate) fn best_block_updated(&self, header: &Header) {`
`166`	`166`	`macro_rules! max_time {`
`167`	`167`	`($timestamp: expr) => {`
`168`	`168`	`loop {`