Introduce Flow parameter in ChannelManager

Author: shaavan

fuzz/src/chanmon_consistency.rs

@@ -19,7 +19,7 @@
 //! channel being force-closed.
 
 use bitcoin::amount::Amount;
-use bitcoin::constants::genesis_block;
+use bitcoin::constants::{genesis_block, ChainHash};
 use bitcoin::locktime::absolute::LockTime;
 use bitcoin::network::Network;
 use bitcoin::opcodes;
@@ -57,6 +57,7 @@ use lightning::ln::msgs::{
 };
 use lightning::ln::script::ShutdownScript;
 use lightning::ln::types::ChannelId;
+use lightning::offers::flow::OffersMessageFlow;
 use lightning::offers::invoice::UnsignedBolt12Invoice;
 use lightning::onion_message::messenger::{Destination, MessageRouter, OnionMessagePath};
 use lightning::routing::router::{
@@ -84,6 +85,7 @@ use lightning::io::Cursor;
 use lightning::util::dyn_signer::DynSigner;
 use std::cmp::{self, Ordering};
 use std::mem;
+use std::ops::Deref;
 use std::sync::atomic;
 use std::sync::{Arc, Mutex};
 
@@ -113,6 +115,7 @@ impl FeeEstimator for FuzzEstimator {
 	}
 }
 
+#[derive(Clone)]
 struct FuzzRouter {}
 
 impl Router for FuzzRouter {
@@ -653,13 +656,27 @@ pub fn do_test<Out: Output>(data: &[u8], underlying_out: Out, anchors: bool) {
 			let network = Network::Bitcoin;
 			let best_block_timestamp = genesis_block(network).header.time;
 			let params = ChainParameters { network, best_block: BestBlock::from_network(network) };
+			let chain_hash = ChainHash::using_genesis_block(network);
+
+			let flow = OffersMessageFlow::new(
+				chain_hash,
+				params.best_block,
+				keys_manager.get_node_id(Recipient::Node).unwrap(),
+				best_block_timestamp,
+				keys_manager.get_inbound_payment_key(),
+				keys_manager.clone(),
+				&router,
+				&router,
+			);
+
 			(
 				ChannelManager::new(
 					$fee_estimator.clone(),
 					monitor.clone(),
 					broadcast.clone(),
 					&router,
 					&router,
+					flow,
 					Arc::clone(&logger),
 					keys_manager.clone(),
 					keys_manager.clone(),

fuzz/src/full_stack.rs

@@ -15,7 +15,7 @@
 
 use bitcoin::amount::Amount;
 use bitcoin::consensus::encode::deserialize;
-use bitcoin::constants::genesis_block;
+use bitcoin::constants::{genesis_block, ChainHash};
 use bitcoin::locktime::absolute::LockTime;
 use bitcoin::network::Network;
 use bitcoin::opcodes;
@@ -49,6 +49,7 @@ use lightning::ln::peer_handler::{
 };
 use lightning::ln::script::ShutdownScript;
 use lightning::ln::types::ChannelId;
+use lightning::offers::flow::OffersMessageFlow;
 use lightning::offers::invoice::UnsignedBolt12Invoice;
 use lightning::onion_message::messenger::{Destination, MessageRouter, OnionMessagePath};
 use lightning::routing::gossip::{NetworkGraph, P2PGossipSync};
@@ -78,6 +79,7 @@ use bitcoin::secp256k1::{self, Message, PublicKey, Scalar, Secp256k1, SecretKey}
 use lightning::util::dyn_signer::DynSigner;
 use std::cell::RefCell;
 use std::cmp;
+use std::ops::Deref;
 use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
 use std::sync::{Arc, Mutex};
 
@@ -145,6 +147,7 @@ impl FeeEstimator for FuzzEstimator {
 	}
 }
 
+#[derive(Clone)]
 struct FuzzRouter {}
 
 impl Router for FuzzRouter {
@@ -625,12 +628,26 @@ pub fn do_test(mut data: &[u8], logger: &Arc<dyn Logger>) {
 	let network = Network::Bitcoin;
 	let best_block_timestamp = genesis_block(network).header.time;
 	let params = ChainParameters { network, best_block: BestBlock::from_network(network) };
+	let chain_hash = ChainHash::using_genesis_block(network);
+
+	let flow = OffersMessageFlow::new(
+		chain_hash,
+		params.best_block,
+		keys_manager.get_node_id(Recipient::Node).unwrap(),
+		best_block_timestamp,
+		keys_manager.get_inbound_payment_key(),
+		keys_manager.clone(),
+		&router,
+		&router,
+	);
+
 	let channelmanager = Arc::new(ChannelManager::new(
 		fee_est.clone(),
 		monitor.clone(),
 		broadcast.clone(),
 		&router,
 		&router,
+		flow,
 		Arc::clone(&logger),
 		keys_manager.clone(),
 		keys_manager.clone(),

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(),

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(),

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,

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);
 	}