X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=src%2Fmain.rs;h=bfb586adfbe630c3df9ff7fc31b51c67d493ff9c;hb=ed39ac6dd5f0b297fe4bf1468b2c0e6e20ce9f9d;hp=f51528b7b57832b3caf279b2f7fd856b6377de87;hpb=8c621ea7609c9cc785b93862e4afed93b43218b5;p=ldk-sample

diff --git a/src/main.rs b/src/main.rs
index f51528b..bfb586a 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -19,17 +19,16 @@ use lightning::chain;
 use lightning::chain::chaininterface::{BroadcasterInterface, ConfirmationTarget, FeeEstimator};
 use lightning::chain::chainmonitor;
 use lightning::chain::keysinterface::{InMemorySigner, KeysInterface, KeysManager};
-use lightning::chain::Filter;
-use lightning::chain::Watch;
+use lightning::chain::{BestBlock, Filter, Watch};
 use lightning::ln::channelmanager;
 use lightning::ln::channelmanager::{
-	BestBlock, ChainParameters, ChannelManagerReadArgs, SimpleArcChannelManager,
+	ChainParameters, ChannelManagerReadArgs, SimpleArcChannelManager,
 };
 use lightning::ln::peer_handler::{MessageHandler, SimpleArcPeerManager};
 use lightning::ln::{PaymentHash, PaymentPreimage, PaymentSecret};
 use lightning::routing::network_graph::NetGraphMsgHandler;
 use lightning::util::config::UserConfig;
-use lightning::util::events::{Event, EventsProvider};
+use lightning::util::events::{Event, PaymentPurpose};
 use lightning::util::ser::ReadableArgs;
 use lightning_background_processor::BackgroundProcessor;
 use lightning_block_sync::init;
@@ -50,8 +49,6 @@ use std::ops::Deref;
 use std::path::Path;
 use std::sync::{Arc, Mutex};
 use std::time::{Duration, SystemTime};
-use tokio::sync::mpsc;
-use tokio::sync::mpsc::Receiver;
 
 pub(crate) enum HTLCStatus {
 	Pending,
@@ -101,166 +98,157 @@ pub(crate) type ChannelManager =
 	SimpleArcChannelManager<ChainMonitor, BitcoindClient, BitcoindClient, FilesystemLogger>;
 
 async fn handle_ldk_events(
-	channel_manager: Arc<ChannelManager>, chain_monitor: Arc<ChainMonitor>,
-	bitcoind_client: Arc<BitcoindClient>, keys_manager: Arc<KeysManager>,
-	inbound_payments: PaymentInfoStorage, outbound_payments: PaymentInfoStorage, network: Network,
-	mut event_receiver: Receiver<()>,
+	channel_manager: Arc<ChannelManager>, bitcoind_client: Arc<BitcoindClient>,
+	keys_manager: Arc<KeysManager>, inbound_payments: PaymentInfoStorage,
+	outbound_payments: PaymentInfoStorage, network: Network, event: Event,
 ) {
-	loop {
-		let received = event_receiver.recv();
-		if received.await.is_none() {
-			println!("LDK Event channel closed!");
-			return;
+	match event {
+		Event::FundingGenerationReady {
+			temporary_channel_id,
+			channel_value_satoshis,
+			output_script,
+			..
+		} => {
+			// Construct the raw transaction with one output, that is paid the amount of the
+			// channel.
+			let addr = WitnessProgram::from_scriptpubkey(
+				&output_script[..],
+				match network {
+					Network::Bitcoin => bitcoin_bech32::constants::Network::Bitcoin,
+					Network::Testnet => bitcoin_bech32::constants::Network::Testnet,
+					Network::Regtest => bitcoin_bech32::constants::Network::Regtest,
+					Network::Signet => panic!("Signet unsupported"),
+				},
+			)
+			.expect("Lightning funding tx should always be to a SegWit output")
+			.to_address();
+			let mut outputs = vec![HashMap::with_capacity(1)];
+			outputs[0].insert(addr, channel_value_satoshis as f64 / 100_000_000.0);
+			let raw_tx = bitcoind_client.create_raw_transaction(outputs).await;
+
+			// Have your wallet put the inputs into the transaction such that the output is
+			// satisfied.
+			let funded_tx = bitcoind_client.fund_raw_transaction(raw_tx).await;
+			let change_output_position = funded_tx.changepos;
+			assert!(change_output_position == 0 || change_output_position == 1);
+
+			// Sign the final funding transaction and broadcast it.
+			let signed_tx = bitcoind_client.sign_raw_transaction_with_wallet(funded_tx.hex).await;
+			assert_eq!(signed_tx.complete, true);
+			let final_tx: Transaction =
+				encode::deserialize(&hex_utils::to_vec(&signed_tx.hex).unwrap()).unwrap();
+			// Give the funding transaction back to LDK for opening the channel.
+			if channel_manager
+				.funding_transaction_generated(&temporary_channel_id, final_tx)
+				.is_err()
+			{
+				println!(
+					"\nERROR: Channel went away before we could fund it. The peer disconnected or refused the channel.");
+				print!("> ");
+				io::stdout().flush().unwrap();
+			}
 		}
-		let loop_channel_manager = channel_manager.clone();
-		let mut events = channel_manager.get_and_clear_pending_events();
-		events.append(&mut chain_monitor.get_and_clear_pending_events());
-		for event in events {
-			match event {
-				Event::FundingGenerationReady {
-					temporary_channel_id,
-					channel_value_satoshis,
-					output_script,
-					..
-				} => {
-					// Construct the raw transaction with one output, that is paid the amount of the
-					// channel.
-					let addr = WitnessProgram::from_scriptpubkey(
-						&output_script[..],
-						match network {
-							Network::Bitcoin => bitcoin_bech32::constants::Network::Bitcoin,
-							Network::Testnet => bitcoin_bech32::constants::Network::Testnet,
-							Network::Regtest => bitcoin_bech32::constants::Network::Regtest,
-							Network::Signet => panic!("Signet unsupported"),
-						},
-					)
-					.expect("Lightning funding tx should always be to a SegWit output")
-					.to_address();
-					let mut outputs = vec![HashMap::with_capacity(1)];
-					outputs[0].insert(addr, channel_value_satoshis as f64 / 100_000_000.0);
-					let raw_tx = bitcoind_client.create_raw_transaction(outputs).await;
-
-					// Have your wallet put the inputs into the transaction such that the output is
-					// satisfied.
-					let funded_tx = bitcoind_client.fund_raw_transaction(raw_tx).await;
-					let change_output_position = funded_tx.changepos;
-					assert!(change_output_position == 0 || change_output_position == 1);
-
-					// Sign the final funding transaction and broadcast it.
-					let signed_tx =
-						bitcoind_client.sign_raw_transaction_with_wallet(funded_tx.hex).await;
-					assert_eq!(signed_tx.complete, true);
-					let final_tx: Transaction =
-						encode::deserialize(&hex_utils::to_vec(&signed_tx.hex).unwrap()).unwrap();
-					// Give the funding transaction back to LDK for opening the channel.
-					loop_channel_manager
-						.funding_transaction_generated(&temporary_channel_id, final_tx)
-						.unwrap();
+		Event::PaymentReceived { payment_hash, purpose, amt, .. } => {
+			let mut payments = inbound_payments.lock().unwrap();
+			let (payment_preimage, payment_secret) = match purpose {
+				PaymentPurpose::InvoicePayment { payment_preimage, payment_secret, .. } => {
+					(payment_preimage, Some(payment_secret))
 				}
-				Event::PaymentReceived {
-					payment_hash,
-					payment_preimage,
-					payment_secret,
-					amt,
-					..
-				} => {
-					let mut payments = inbound_payments.lock().unwrap();
-					let status = match loop_channel_manager.claim_funds(payment_preimage.unwrap()) {
-						true => {
-							println!(
-								"\nEVENT: received payment from payment hash {} of {} millisatoshis",
-								hex_utils::hex_str(&payment_hash.0),
-								amt
-							);
-							print!("> ");
-							io::stdout().flush().unwrap();
-							HTLCStatus::Succeeded
-						}
-						_ => HTLCStatus::Failed,
-					};
-					match payments.entry(payment_hash) {
-						Entry::Occupied(mut e) => {
-							let payment = e.get_mut();
-							payment.status = status;
-							payment.preimage = Some(payment_preimage.unwrap());
-							payment.secret = Some(payment_secret);
-						}
-						Entry::Vacant(e) => {
-							e.insert(PaymentInfo {
-								preimage: Some(payment_preimage.unwrap()),
-								secret: Some(payment_secret),
-								status,
-								amt_msat: MillisatAmount(Some(amt)),
-							});
-						}
-					}
-				}
-				Event::PaymentSent { payment_preimage } => {
-					let hashed = PaymentHash(Sha256::hash(&payment_preimage.0).into_inner());
-					let mut payments = outbound_payments.lock().unwrap();
-					for (payment_hash, payment) in payments.iter_mut() {
-						if *payment_hash == hashed {
-							payment.preimage = Some(payment_preimage);
-							payment.status = HTLCStatus::Succeeded;
-							println!(
-								"\nEVENT: successfully sent payment of {} millisatoshis from \
-                                         payment hash {:?} with preimage {:?}",
-								payment.amt_msat,
-								hex_utils::hex_str(&payment_hash.0),
-								hex_utils::hex_str(&payment_preimage.0)
-							);
-							print!("> ");
-							io::stdout().flush().unwrap();
-						}
-					}
-				}
-				Event::PaymentFailed { payment_hash, rejected_by_dest } => {
-					print!(
-						"\nEVENT: Failed to send payment to payment hash {:?}: ",
-						hex_utils::hex_str(&payment_hash.0)
+				PaymentPurpose::SpontaneousPayment(preimage) => (Some(preimage), None),
+			};
+			let status = match channel_manager.claim_funds(payment_preimage.unwrap()) {
+				true => {
+					println!(
+						"\nEVENT: received payment from payment hash {} of {} millisatoshis",
+						hex_utils::hex_str(&payment_hash.0),
+						amt
 					);
-					if rejected_by_dest {
-						println!("re-attempting the payment will not succeed");
-					} else {
-						println!("payment may be retried");
-					}
 					print!("> ");
 					io::stdout().flush().unwrap();
-
-					let mut payments = outbound_payments.lock().unwrap();
-					if payments.contains_key(&payment_hash) {
-						let payment = payments.get_mut(&payment_hash).unwrap();
-						payment.status = HTLCStatus::Failed;
-					}
+					HTLCStatus::Succeeded
 				}
-				Event::PendingHTLCsForwardable { time_forwardable } => {
-					let forwarding_channel_manager = loop_channel_manager.clone();
-					tokio::spawn(async move {
-						let min = time_forwardable.as_millis() as u64;
-						let millis_to_sleep = thread_rng().gen_range(min, min * 5) as u64;
-						tokio::time::sleep(Duration::from_millis(millis_to_sleep)).await;
-						forwarding_channel_manager.process_pending_htlc_forwards();
+				_ => HTLCStatus::Failed,
+			};
+			match payments.entry(payment_hash) {
+				Entry::Occupied(mut e) => {
+					let payment = e.get_mut();
+					payment.status = status;
+					payment.preimage = payment_preimage;
+					payment.secret = payment_secret;
+				}
+				Entry::Vacant(e) => {
+					e.insert(PaymentInfo {
+						preimage: payment_preimage,
+						secret: payment_secret,
+						status,
+						amt_msat: MillisatAmount(Some(amt)),
 					});
 				}
-				Event::SpendableOutputs { outputs } => {
-					let destination_address = bitcoind_client.get_new_address().await;
-					let output_descriptors = &outputs.iter().map(|a| a).collect::<Vec<_>>();
-					let tx_feerate =
-						bitcoind_client.get_est_sat_per_1000_weight(ConfirmationTarget::Normal);
-					let spending_tx = keys_manager
-						.spend_spendable_outputs(
-							output_descriptors,
-							Vec::new(),
-							destination_address.script_pubkey(),
-							tx_feerate,
-							&Secp256k1::new(),
-						)
-						.unwrap();
-					bitcoind_client.broadcast_transaction(&spending_tx);
+			}
+		}
+		Event::PaymentSent { payment_preimage } => {
+			let hashed = PaymentHash(Sha256::hash(&payment_preimage.0).into_inner());
+			let mut payments = outbound_payments.lock().unwrap();
+			for (payment_hash, payment) in payments.iter_mut() {
+				if *payment_hash == hashed {
+					payment.preimage = Some(payment_preimage);
+					payment.status = HTLCStatus::Succeeded;
+					println!(
+						"\nEVENT: successfully sent payment of {} millisatoshis from \
+								 payment hash {:?} with preimage {:?}",
+						payment.amt_msat,
+						hex_utils::hex_str(&payment_hash.0),
+						hex_utils::hex_str(&payment_preimage.0)
+					);
+					print!("> ");
+					io::stdout().flush().unwrap();
 				}
 			}
 		}
-		tokio::time::sleep(Duration::from_secs(1)).await;
+		Event::PaymentFailed { payment_hash, rejected_by_dest } => {
+			print!(
+				"\nEVENT: Failed to send payment to payment hash {:?}: ",
+				hex_utils::hex_str(&payment_hash.0)
+			);
+			if rejected_by_dest {
+				println!("re-attempting the payment will not succeed");
+			} else {
+				println!("payment may be retried");
+			}
+			print!("> ");
+			io::stdout().flush().unwrap();
+
+			let mut payments = outbound_payments.lock().unwrap();
+			if payments.contains_key(&payment_hash) {
+				let payment = payments.get_mut(&payment_hash).unwrap();
+				payment.status = HTLCStatus::Failed;
+			}
+		}
+		Event::PendingHTLCsForwardable { time_forwardable } => {
+			let forwarding_channel_manager = channel_manager.clone();
+			tokio::spawn(async move {
+				let min = time_forwardable.as_millis() as u64;
+				let millis_to_sleep = thread_rng().gen_range(min, min * 5) as u64;
+				tokio::time::sleep(Duration::from_millis(millis_to_sleep)).await;
+				forwarding_channel_manager.process_pending_htlc_forwards();
+			});
+		}
+		Event::SpendableOutputs { outputs } => {
+			let destination_address = bitcoind_client.get_new_address().await;
+			let output_descriptors = &outputs.iter().map(|a| a).collect::<Vec<_>>();
+			let tx_feerate =
+				bitcoind_client.get_est_sat_per_1000_weight(ConfirmationTarget::Normal);
+			let spending_tx = keys_manager
+				.spend_spendable_outputs(
+					output_descriptors,
+					Vec::new(),
+					destination_address.script_pubkey(),
+					tx_feerate,
+					&Secp256k1::new(),
+				)
+				.unwrap();
+			bitcoind_client.broadcast_transaction(&spending_tx);
+		}
 	}
 }
 
@@ -365,7 +353,8 @@ async fn start_ldk() {
 	let mut channelmonitors = persister.read_channelmonitors(keys_manager.clone()).unwrap();
 
 	// Step 8: Initialize the ChannelManager
-	let user_config = UserConfig::default();
+	let mut user_config = UserConfig::default();
+	user_config.peer_channel_config_limits.force_announced_channel_preference = false;
 	let mut restarting_node = true;
 	let (channel_manager_blockhash, mut channel_manager) = {
 		if let Ok(mut f) = fs::File::open(format!("{}/manager", ldk_data_dir.clone())) {
@@ -451,13 +440,33 @@ async fn start_ldk() {
 	}
 
 	// Step 11: Optional: Initialize the NetGraphMsgHandler
-	// XXX persist routing data
 	let genesis = genesis_block(args.network).header.block_hash();
-	let router = Arc::new(NetGraphMsgHandler::new(
-		genesis,
+	let network_graph_path = format!("{}/network_graph", ldk_data_dir.clone());
+	let network_graph = disk::read_network(Path::new(&network_graph_path), genesis);
+	let router = Arc::new(NetGraphMsgHandler::from_net_graph(
 		None::<Arc<dyn chain::Access + Send + Sync>>,
 		logger.clone(),
+		network_graph,
 	));
+	let router_persist = Arc::clone(&router);
+	tokio::spawn(async move {
+		let mut interval = tokio::time::interval(Duration::from_secs(600));
+		loop {
+			interval.tick().await;
+			if disk::persist_network(
+				Path::new(&network_graph_path),
+				&*router_persist.network_graph.read().unwrap(),
+			)
+			.is_err()
+			{
+				// Persistence errors here are non-fatal as we can just fetch the routing graph
+				// again later, but they may indicate a disk error which could be fatal elsewhere.
+				eprintln!(
+					"Warning: Failed to persist network graph, check your disk and permissions"
+				);
+			}
+		}
+	});
 
 	// Step 12: Initialize the PeerManager
 	let channel_manager: Arc<ChannelManager> = Arc::new(channel_manager);
@@ -475,11 +484,7 @@ async fn start_ldk() {
 	// ## Running LDK
 	// Step 13: Initialize networking
 
-	// We poll for events in handle_ldk_events(..) rather than waiting for them over the
-	// mpsc::channel, so we can leave the event receiver as unused.
-	let (event_ntfn_sender, event_ntfn_receiver) = mpsc::channel(2);
 	let peer_manager_connection_handler = peer_manager.clone();
-	let event_notifier = event_ntfn_sender.clone();
 	let listening_port = args.ldk_peer_listening_port;
 	tokio::spawn(async move {
 		let listener = tokio::net::TcpListener::bind(format!("0.0.0.0:{}", listening_port))
@@ -487,12 +492,10 @@ async fn start_ldk() {
 			.expect("Failed to bind to listen port - is something else already listening on it?");
 		loop {
 			let peer_mgr = peer_manager_connection_handler.clone();
-			let notifier = event_notifier.clone();
 			let tcp_stream = listener.accept().await.unwrap().0;
 			tokio::spawn(async move {
 				lightning_net_tokio::setup_inbound(
 					peer_mgr.clone(),
-					notifier.clone(),
 					tcp_stream.into_std().unwrap(),
 				)
 				.await;
@@ -521,9 +524,8 @@ async fn start_ldk() {
 		}
 	});
 
-	// Step 15: Initialize LDK Event Handling
+	// Step 15: Handle LDK Events
 	let channel_manager_event_listener = channel_manager.clone();
-	let chain_monitor_event_listener = chain_monitor.clone();
 	let keys_manager_listener = keys_manager.clone();
 	// TODO: persist payment info to disk
 	let inbound_payments: PaymentInfoStorage = Arc::new(Mutex::new(HashMap::new()));
@@ -532,26 +534,27 @@ async fn start_ldk() {
 	let outbound_pmts_for_events = outbound_payments.clone();
 	let network = args.network;
 	let bitcoind_rpc = bitcoind_client.clone();
-	tokio::spawn(async move {
-		handle_ldk_events(
-			channel_manager_event_listener,
-			chain_monitor_event_listener,
-			bitcoind_rpc,
-			keys_manager_listener,
-			inbound_pmts_for_events,
-			outbound_pmts_for_events,
+	let handle = tokio::runtime::Handle::current();
+	let event_handler = move |event| {
+		handle.block_on(handle_ldk_events(
+			channel_manager_event_listener.clone(),
+			bitcoind_rpc.clone(),
+			keys_manager_listener.clone(),
+			inbound_pmts_for_events.clone(),
+			outbound_pmts_for_events.clone(),
 			network,
-			event_ntfn_receiver,
-		)
-		.await;
-	});
-
-	// Step 16 & 17: Persist ChannelManager & Background Processing
+			event,
+		))
+	};
+	// Step 16: Persist ChannelManager
 	let data_dir = ldk_data_dir.clone();
 	let persist_channel_manager_callback =
 		move |node: &ChannelManager| FilesystemPersister::persist_manager(data_dir.clone(), &*node);
+	// Step 17: Background Processing
 	BackgroundProcessor::start(
 		persist_channel_manager_callback,
+		event_handler,
+		chain_monitor.clone(),
 		channel_manager.clone(),
 		peer_manager.clone(),
 		logger.clone(),
@@ -563,13 +566,10 @@ async fn start_ldk() {
 		Ok(mut info) => {
 			for (pubkey, peer_addr) in info.drain() {
 				for chan_info in channel_manager.list_channels() {
-					if pubkey == chan_info.remote_network_id {
-						let _ = cli::connect_peer_if_necessary(
-							pubkey,
-							peer_addr,
-							peer_manager.clone(),
-							event_ntfn_sender.clone(),
-						);
+					if pubkey == chan_info.counterparty.node_id {
+						let _ =
+							cli::connect_peer_if_necessary(pubkey, peer_addr, peer_manager.clone())
+								.await;
 					}
 				}
 			}
@@ -577,6 +577,26 @@ async fn start_ldk() {
 		Err(e) => println!("ERROR: errored reading channel peer info from disk: {:?}", e),
 	}
 
+	// Regularly broadcast our node_announcement. This is only required (or possible) if we have
+	// some public channels, and is only useful if we have public listen address(es) to announce.
+	// In a production environment, this should occur only after the announcement of new channels
+	// to avoid churn in the global network graph.
+	let chan_manager = Arc::clone(&channel_manager);
+	let network = args.network;
+	if !args.ldk_announced_listen_addr.is_empty() {
+		tokio::spawn(async move {
+			let mut interval = tokio::time::interval(Duration::from_secs(60));
+			loop {
+				interval.tick().await;
+				chan_manager.broadcast_node_announcement(
+					[0; 3],
+					args.ldk_announced_node_name,
+					args.ldk_announced_listen_addr.clone(),
+				);
+			}
+		});
+	}
+
 	// Start the CLI.
 	cli::poll_for_user_input(
 		peer_manager.clone(),
@@ -585,10 +605,9 @@ async fn start_ldk() {
 		router.clone(),
 		inbound_payments,
 		outbound_payments,
-		event_ntfn_sender,
 		ldk_data_dir.clone(),
 		logger.clone(),
-		args.network,
+		network,
 	)
 	.await;
 }