X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=src%2Fmain.rs;h=6480805d92759eed4d90407ada06a3f6eb7f202f;hb=fea709d5a9fba7553c8bbc25236a6640ef52dee5;hp=2012921c334f6a6930693899c11a7549574df601;hpb=8098135a88a92f6ce50747cddea97b13c8296593;p=ldk-sample diff --git a/src/main.rs b/src/main.rs index 2012921..6480805 100644 --- a/src/main.rs +++ b/src/main.rs @@ -16,14 +16,15 @@ use bitcoin_bech32::WitnessProgram; use disk::{INBOUND_PAYMENTS_FNAME, OUTBOUND_PAYMENTS_FNAME}; use lightning::chain::{chainmonitor, ChannelMonitorUpdateStatus}; use lightning::chain::{Filter, Watch}; +use lightning::events::bump_transaction::{BumpTransactionEventHandler, Wallet}; use lightning::events::{Event, PaymentFailureReason, PaymentPurpose}; use lightning::ln::channelmanager::{self, RecentPaymentDetails}; use lightning::ln::channelmanager::{ - ChainParameters, ChannelManagerReadArgs, SimpleArcChannelManager, + ChainParameters, ChannelManagerReadArgs, PaymentId, SimpleArcChannelManager, }; use lightning::ln::msgs::DecodeError; use lightning::ln::peer_handler::{IgnoringMessageHandler, MessageHandler, SimpleArcPeerManager}; -use lightning::ln::{PaymentHash, PaymentPreimage, PaymentSecret}; +use lightning::ln::{ChannelId, PaymentHash, PaymentPreimage, PaymentSecret}; use lightning::onion_message::{DefaultMessageRouter, SimpleArcOnionMessenger}; use lightning::routing::gossip; use lightning::routing::gossip::{NodeId, P2PGossipSync}; @@ -31,7 +32,7 @@ use lightning::routing::router::DefaultRouter; use lightning::routing::scoring::ProbabilisticScoringFeeParameters; use lightning::sign::{EntropySource, InMemorySigner, KeysManager, SpendableOutputDescriptor}; use lightning::util::config::UserConfig; -use lightning::util::persist::KVStorePersister; +use lightning::util::persist::{self, KVStore, MonitorUpdatingPersister}; use lightning::util::ser::{Readable, ReadableArgs, Writeable, Writer}; use lightning::{chain, impl_writeable_tlv_based, impl_writeable_tlv_based_enum}; use lightning_background_processor::{process_events_async, GossipSync}; @@ -40,7 +41,7 @@ use lightning_block_sync::poll; use lightning_block_sync::SpvClient; use lightning_block_sync::UnboundedCache; use lightning_net_tokio::SocketDescriptor; -use lightning_persister::FilesystemPersister; +use lightning_persister::fs_store::FilesystemStore; use rand::{thread_rng, Rng}; use std::collections::hash_map::Entry; use std::collections::HashMap; @@ -52,7 +53,7 @@ use std::io; use std::io::Write; use std::path::Path; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::{Arc, Mutex}; +use std::sync::{Arc, Mutex, RwLock}; use std::time::{Duration, SystemTime}; pub(crate) const PENDING_SPENDABLE_OUTPUT_DIR: &'static str = "pending_spendable_outputs"; @@ -108,11 +109,19 @@ impl_writeable_tlv_based!(PaymentInfo, { (6, amt_msat, required), }); -pub(crate) struct PaymentInfoStorage { +pub(crate) struct InboundPaymentInfoStorage { payments: HashMap, } -impl_writeable_tlv_based!(PaymentInfoStorage, { +impl_writeable_tlv_based!(InboundPaymentInfoStorage, { + (0, payments, required), +}); + +pub(crate) struct OutboundPaymentInfoStorage { + payments: HashMap, +} + +impl_writeable_tlv_based!(OutboundPaymentInfoStorage, { (0, payments, required), }); @@ -122,7 +131,25 @@ type ChainMonitor = chainmonitor::ChainMonitor< Arc, Arc, Arc, - Arc, + Arc< + MonitorUpdatingPersister< + Arc, + Arc, + Arc, + Arc, + >, + >, +>; + +pub(crate) type GossipVerifier = lightning_block_sync::gossip::GossipVerifier< + lightning_block_sync::gossip::TokioSpawner, + Arc, + Arc, + SocketDescriptor, + Arc, + Arc, + IgnoringMessageHandler, + Arc, >; pub(crate) type PeerManager = SimpleArcPeerManager< @@ -130,7 +157,7 @@ pub(crate) type PeerManager = SimpleArcPeerManager< ChainMonitor, BitcoindClient, BitcoindClient, - BitcoindClient, + GossipVerifier, FilesystemLogger, >; @@ -139,13 +166,22 @@ pub(crate) type ChannelManager = pub(crate) type NetworkGraph = gossip::NetworkGraph>; -type OnionMessenger = SimpleArcOnionMessenger; +type OnionMessenger = + SimpleArcOnionMessenger; + +pub(crate) type BumpTxEventHandler = BumpTransactionEventHandler< + Arc, + Arc, Arc>>, + Arc, + Arc, +>; async fn handle_ldk_events( channel_manager: &Arc, bitcoind_client: &BitcoindClient, network_graph: &NetworkGraph, keys_manager: &KeysManager, - inbound_payments: Arc>, - outbound_payments: Arc>, persister: &Arc, + bump_tx_event_handler: &BumpTxEventHandler, + inbound_payments: Arc>, + outbound_payments: Arc>, fs_store: &Arc, network: Network, event: Event, ) { match event { @@ -210,8 +246,7 @@ async fn handle_ldk_events( } => { println!( "\nEVENT: received payment from payment hash {} of {} millisatoshis", - hex_utils::hex_str(&payment_hash.0), - amount_msat, + payment_hash, amount_msat, ); print!("> "); io::stdout().flush().unwrap(); @@ -221,11 +256,17 @@ async fn handle_ldk_events( }; channel_manager.claim_funds(payment_preimage.unwrap()); } - Event::PaymentClaimed { payment_hash, purpose, amount_msat, receiver_node_id: _ } => { + Event::PaymentClaimed { + payment_hash, + purpose, + amount_msat, + receiver_node_id: _, + htlcs: _, + sender_intended_total_msat: _, + } => { println!( "\nEVENT: claimed payment from payment hash {} of {} millisatoshis", - hex_utils::hex_str(&payment_hash.0), - amount_msat, + payment_hash, amount_msat, ); print!("> "); io::stdout().flush().unwrap(); @@ -252,54 +293,94 @@ async fn handle_ldk_events( }); } } - persister.persist(INBOUND_PAYMENTS_FNAME, &*inbound).unwrap(); + fs_store.write("", "", INBOUND_PAYMENTS_FNAME, &inbound.encode()).unwrap(); } - Event::PaymentSent { payment_preimage, payment_hash, fee_paid_msat, .. } => { + Event::PaymentSent { + payment_preimage, payment_hash, fee_paid_msat, payment_id, .. + } => { let mut outbound = outbound_payments.lock().unwrap(); - for (hash, payment) in outbound.payments.iter_mut() { - if *hash == payment_hash { + for (id, payment) in outbound.payments.iter_mut() { + if *id == payment_id.unwrap() { payment.preimage = Some(payment_preimage); payment.status = HTLCStatus::Succeeded; println!( "\nEVENT: successfully sent payment of {} millisatoshis{} from \ - payment hash {:?} with preimage {:?}", + payment hash {} with preimage {}", payment.amt_msat, if let Some(fee) = fee_paid_msat { format!(" (fee {} msat)", fee) } else { "".to_string() }, - hex_utils::hex_str(&payment_hash.0), - hex_utils::hex_str(&payment_preimage.0) + payment_hash, + payment_preimage ); print!("> "); io::stdout().flush().unwrap(); } } - persister.persist(OUTBOUND_PAYMENTS_FNAME, &*outbound).unwrap(); + fs_store.write("", "", OUTBOUND_PAYMENTS_FNAME, &outbound.encode()).unwrap(); } - Event::OpenChannelRequest { .. } => { - // Unreachable, we don't set manually_accept_inbound_channels + Event::OpenChannelRequest { + ref temporary_channel_id, ref counterparty_node_id, .. + } => { + let mut random_bytes = [0u8; 16]; + random_bytes.copy_from_slice(&keys_manager.get_secure_random_bytes()[..16]); + let user_channel_id = u128::from_be_bytes(random_bytes); + let res = channel_manager.accept_inbound_channel( + temporary_channel_id, + counterparty_node_id, + user_channel_id, + ); + + if let Err(e) = res { + print!( + "\nEVENT: Failed to accept inbound channel ({}) from {}: {:?}", + temporary_channel_id, + hex_utils::hex_str(&counterparty_node_id.serialize()), + e, + ); + } else { + print!( + "\nEVENT: Accepted inbound channel ({}) from {}", + temporary_channel_id, + hex_utils::hex_str(&counterparty_node_id.serialize()), + ); + } + print!("> "); + io::stdout().flush().unwrap(); } Event::PaymentPathSuccessful { .. } => {} Event::PaymentPathFailed { .. } => {} Event::ProbeSuccessful { .. } => {} Event::ProbeFailed { .. } => {} - Event::PaymentFailed { payment_hash, reason, .. } => { + Event::PaymentFailed { payment_hash, reason, payment_id, .. } => { print!( - "\nEVENT: Failed to send payment to payment hash {:?}: {:?}", - hex_utils::hex_str(&payment_hash.0), + "\nEVENT: Failed to send payment to payment hash {}: {:?}", + payment_hash, if let Some(r) = reason { r } else { PaymentFailureReason::RetriesExhausted } ); print!("> "); io::stdout().flush().unwrap(); let mut outbound = outbound_payments.lock().unwrap(); - if outbound.payments.contains_key(&payment_hash) { - let payment = outbound.payments.get_mut(&payment_hash).unwrap(); + if outbound.payments.contains_key(&payment_id) { + let payment = outbound.payments.get_mut(&payment_id).unwrap(); payment.status = HTLCStatus::Failed; } - persister.persist(OUTBOUND_PAYMENTS_FNAME, &*outbound).unwrap(); + fs_store.write("", "", OUTBOUND_PAYMENTS_FNAME, &outbound.encode()).unwrap(); + } + Event::InvoiceRequestFailed { payment_id } => { + print!("\nEVENT: Failed to request invoice to send payment with id {}", payment_id); + print!("> "); + io::stdout().flush().unwrap(); + + let mut outbound = outbound_payments.lock().unwrap(); + if outbound.payments.contains_key(&payment_id) { + let payment = outbound.payments.get_mut(&payment_id).unwrap(); + payment.status = HTLCStatus::Failed; + } + fs_store.write("", "", OUTBOUND_PAYMENTS_FNAME, &outbound.encode()).unwrap(); } Event::PaymentForwarded { prev_channel_id, @@ -312,7 +393,7 @@ async fn handle_ldk_events( let nodes = read_only_network_graph.nodes(); let channels = channel_manager.list_channels(); - let node_str = |channel_id: &Option<[u8; 32]>| match channel_id { + let node_str = |channel_id: &Option| match channel_id { None => String::new(), Some(channel_id) => match channels.iter().find(|c| c.channel_id == *channel_id) { None => String::new(), @@ -329,9 +410,9 @@ async fn handle_ldk_events( } }, }; - let channel_str = |channel_id: &Option<[u8; 32]>| { + let channel_str = |channel_id: &Option| { channel_id - .map(|channel_id| format!(" with channel {}", hex_utils::hex_str(&channel_id))) + .map(|channel_id| format!(" with channel {}", channel_id)) .unwrap_or_default() }; let from_prev_str = @@ -373,7 +454,7 @@ async fn handle_ldk_events( forwarding_channel_manager.process_pending_htlc_forwards(); }); } - Event::SpendableOutputs { outputs } => { + Event::SpendableOutputs { outputs, channel_id: _ } => { // SpendableOutputDescriptors, of which outputs is a vec of, are critical to keep track // of! While a `StaticOutput` descriptor is just an output to a static, well-known key, // other descriptors are not currently ever regenerated for you by LDK. Once we return @@ -387,15 +468,13 @@ async fn handle_ldk_events( let key = hex_utils::hex_str(&keys_manager.get_secure_random_bytes()); // Note that if the type here changes our read code needs to change as well. let output: SpendableOutputDescriptor = output; - persister - .persist(&format!("{}/{}", PENDING_SPENDABLE_OUTPUT_DIR, key), &output) - .unwrap(); + fs_store.write(PENDING_SPENDABLE_OUTPUT_DIR, "", &key, &output.encode()).unwrap(); } } Event::ChannelPending { channel_id, counterparty_node_id, .. } => { println!( "\nEVENT: Channel {} with peer {} is pending awaiting funding lock-in!", - hex_utils::hex_str(&channel_id), + channel_id, hex_utils::hex_str(&counterparty_node_id.serialize()), ); print!("> "); @@ -409,16 +488,23 @@ async fn handle_ldk_events( } => { println!( "\nEVENT: Channel {} with peer {} is ready to be used!", - hex_utils::hex_str(channel_id), + channel_id, hex_utils::hex_str(&counterparty_node_id.serialize()), ); print!("> "); io::stdout().flush().unwrap(); } - Event::ChannelClosed { channel_id, reason, user_channel_id: _ } => { + Event::ChannelClosed { + channel_id, + reason, + user_channel_id: _, + counterparty_node_id, + channel_capacity_sats: _, + } => { println!( - "\nEVENT: Channel {} closed due to: {:?}", - hex_utils::hex_str(&channel_id), + "\nEVENT: Channel {} with counterparty {} closed due to: {:?}", + channel_id, + counterparty_node_id.map(|id| format!("{}", id)).unwrap_or("".to_owned()), reason ); print!("> "); @@ -429,7 +515,7 @@ async fn handle_ldk_events( // the funding transaction either confirms, or this event is generated. } Event::HTLCIntercepted { .. } => {} - Event::BumpTransaction(_) => {} + Event::BumpTransaction(event) => bump_tx_event_handler.handle_event(&event), } } @@ -492,19 +578,7 @@ async fn start_ldk() { // broadcaster. let broadcaster = bitcoind_client.clone(); - // Step 4: Initialize Persist - let persister = Arc::new(FilesystemPersister::new(ldk_data_dir.clone())); - - // Step 5: Initialize the ChainMonitor - let chain_monitor: Arc = Arc::new(chainmonitor::ChainMonitor::new( - None, - broadcaster.clone(), - logger.clone(), - fee_estimator.clone(), - persister.clone(), - )); - - // Step 6: Initialize the KeysManager + // Step 4: Initialize the KeysManager // The key seed that we use to derive the node privkey (that corresponds to the node pubkey) and // other secret key material. @@ -532,9 +606,42 @@ async fn start_ldk() { let cur = SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap(); let keys_manager = Arc::new(KeysManager::new(&keys_seed, cur.as_secs(), cur.subsec_nanos())); + let bump_tx_event_handler = Arc::new(BumpTransactionEventHandler::new( + Arc::clone(&broadcaster), + Arc::new(Wallet::new(Arc::clone(&bitcoind_client), Arc::clone(&logger))), + Arc::clone(&keys_manager), + Arc::clone(&logger), + )); + + // Step 5: Initialize Persistence + let fs_store = Arc::new(FilesystemStore::new(ldk_data_dir.clone().into())); + let persister = Arc::new(MonitorUpdatingPersister::new( + Arc::clone(&fs_store), + Arc::clone(&logger), + 1000, + Arc::clone(&keys_manager), + Arc::clone(&keys_manager), + )); + // Alternatively, you can use the `FilesystemStore` as a `Persist` directly, at the cost of + // larger `ChannelMonitor` update writes (but no deletion or cleanup): + //let persister = Arc::clone(&fs_store); + + // Step 6: Initialize the ChainMonitor + let chain_monitor: Arc = Arc::new(chainmonitor::ChainMonitor::new( + None, + Arc::clone(&broadcaster), + Arc::clone(&logger), + Arc::clone(&fee_estimator), + Arc::clone(&persister), + )); + // Step 7: Read ChannelMonitor state from disk - let mut channelmonitors = - persister.read_channelmonitors(keys_manager.clone(), keys_manager.clone()).unwrap(); + let mut channelmonitors = persister + .read_all_channel_monitors_with_updates(&bitcoind_client, &bitcoind_client) + .unwrap(); + // If you are using the `FilesystemStore` as a `Persist` directly, use + // `lightning::util::persist::read_channel_monitors` like this: + //read_channel_monitors(Arc::clone(&persister), Arc::clone(&keys_manager), Arc::clone(&keys_manager)).unwrap(); // Step 8: Poll for the best chain tip, which may be used by the channel manager & spv client let polled_chain_tip = init::validate_best_block_header(bitcoind_client.as_ref()) @@ -547,7 +654,7 @@ async fn start_ldk() { Arc::new(disk::read_network(Path::new(&network_graph_path), args.network, logger.clone())); let scorer_path = format!("{}/scorer", ldk_data_dir.clone()); - let scorer = Arc::new(Mutex::new(disk::read_scorer( + let scorer = Arc::new(RwLock::new(disk::read_scorer( Path::new(&scorer_path), Arc::clone(&network_graph), Arc::clone(&logger), @@ -566,6 +673,8 @@ async fn start_ldk() { // Step 11: Initialize the ChannelManager let mut user_config = UserConfig::default(); user_config.channel_handshake_limits.force_announced_channel_preference = false; + user_config.channel_handshake_config.negotiate_anchors_zero_fee_htlc_tx = true; + user_config.manually_accept_inbound_channels = true; let mut restarting_node = true; let (channel_manager_blockhash, channel_manager) = { if let Ok(mut f) = fs::File::open(format!("{}/manager", ldk_data_dir.clone())) { @@ -654,16 +763,13 @@ async fn start_ldk() { let funding_outpoint = item.2; assert_eq!( chain_monitor.watch_channel(funding_outpoint, channel_monitor), - ChannelMonitorUpdateStatus::Completed + Ok(ChannelMonitorUpdateStatus::Completed) ); } // Step 14: Optional: Initialize the P2PGossipSync - let gossip_sync = Arc::new(P2PGossipSync::new( - Arc::clone(&network_graph), - None::>, - logger.clone(), - )); + let gossip_sync = + Arc::new(P2PGossipSync::new(Arc::clone(&network_graph), None, Arc::clone(&logger))); // Step 15: Initialize the PeerManager let channel_manager: Arc = Arc::new(channel_manager); @@ -672,7 +778,7 @@ async fn start_ldk() { Arc::clone(&keys_manager), Arc::clone(&logger), Arc::new(DefaultMessageRouter {}), - IgnoringMessageHandler {}, + Arc::clone(&channel_manager), IgnoringMessageHandler {}, )); let mut ephemeral_bytes = [0; 32]; @@ -692,6 +798,15 @@ async fn start_ldk() { Arc::clone(&keys_manager), )); + // Install a GossipVerifier in in the P2PGossipSync + let utxo_lookup = GossipVerifier::new( + Arc::clone(&bitcoind_client.bitcoind_rpc_client), + lightning_block_sync::gossip::TokioSpawner, + Arc::clone(&gossip_sync), + Arc::clone(&peer_manager), + ); + gossip_sync.add_utxo_lookup(Some(utxo_lookup)); + // ## Running LDK // Step 16: Initialize networking @@ -734,35 +849,36 @@ async fn start_ldk() { } }); - let inbound_payments = Arc::new(Mutex::new(disk::read_payment_info(Path::new(&format!( - "{}/{}", - ldk_data_dir, INBOUND_PAYMENTS_FNAME - ))))); - let outbound_payments = Arc::new(Mutex::new(disk::read_payment_info(Path::new(&format!( - "{}/{}", - ldk_data_dir, OUTBOUND_PAYMENTS_FNAME - ))))); - let recent_payments_payment_hashes = channel_manager + let inbound_payments = Arc::new(Mutex::new(disk::read_inbound_payment_info(Path::new( + &format!("{}/{}", ldk_data_dir, INBOUND_PAYMENTS_FNAME), + )))); + let outbound_payments = Arc::new(Mutex::new(disk::read_outbound_payment_info(Path::new( + &format!("{}/{}", ldk_data_dir, OUTBOUND_PAYMENTS_FNAME), + )))); + let recent_payments_payment_ids = channel_manager .list_recent_payments() .into_iter() .filter_map(|p| match p { - RecentPaymentDetails::Pending { payment_hash, .. } => Some(payment_hash), - RecentPaymentDetails::Fulfilled { payment_hash } => payment_hash, - RecentPaymentDetails::Abandoned { payment_hash } => Some(payment_hash), + RecentPaymentDetails::Pending { payment_id, .. } => Some(payment_id), + RecentPaymentDetails::Fulfilled { payment_id, .. } => Some(payment_id), + RecentPaymentDetails::Abandoned { payment_id, .. } => Some(payment_id), + RecentPaymentDetails::AwaitingInvoice { payment_id } => Some(payment_id), }) - .collect::>(); - for (payment_hash, payment_info) in outbound_payments + .collect::>(); + for (payment_id, payment_info) in outbound_payments .lock() .unwrap() .payments .iter_mut() .filter(|(_, i)| matches!(i.status, HTLCStatus::Pending)) { - if !recent_payments_payment_hashes.contains(payment_hash) { + if !recent_payments_payment_ids.contains(payment_id) { payment_info.status = HTLCStatus::Failed; } } - persister.persist(OUTBOUND_PAYMENTS_FNAME, &*outbound_payments.lock().unwrap()).unwrap(); + fs_store + .write("", "", OUTBOUND_PAYMENTS_FNAME, &outbound_payments.lock().unwrap().encode()) + .unwrap(); // Step 18: Handle LDK Events let channel_manager_event_listener = Arc::clone(&channel_manager); @@ -771,25 +887,27 @@ async fn start_ldk() { let keys_manager_event_listener = Arc::clone(&keys_manager); let inbound_payments_event_listener = Arc::clone(&inbound_payments); let outbound_payments_event_listener = Arc::clone(&outbound_payments); - let persister_event_listener = Arc::clone(&persister); + let fs_store_event_listener = Arc::clone(&fs_store); let network = args.network; let event_handler = move |event: Event| { let channel_manager_event_listener = Arc::clone(&channel_manager_event_listener); let bitcoind_client_event_listener = Arc::clone(&bitcoind_client_event_listener); let network_graph_event_listener = Arc::clone(&network_graph_event_listener); let keys_manager_event_listener = Arc::clone(&keys_manager_event_listener); + let bump_tx_event_handler = Arc::clone(&bump_tx_event_handler); let inbound_payments_event_listener = Arc::clone(&inbound_payments_event_listener); let outbound_payments_event_listener = Arc::clone(&outbound_payments_event_listener); - let persister_event_listener = Arc::clone(&persister_event_listener); + let fs_store_event_listener = Arc::clone(&fs_store_event_listener); async move { handle_ldk_events( &channel_manager_event_listener, &bitcoind_client_event_listener, &network_graph_event_listener, &keys_manager_event_listener, + &bump_tx_event_handler, inbound_payments_event_listener, outbound_payments_event_listener, - &persister_event_listener, + &fs_store_event_listener, network, event, ) @@ -798,11 +916,11 @@ async fn start_ldk() { }; // Step 19: Persist ChannelManager and NetworkGraph - let persister = Arc::new(FilesystemPersister::new(ldk_data_dir.clone())); + let persister = Arc::new(FilesystemStore::new(ldk_data_dir.clone().into())); // Step 20: Background Processing let (bp_exit, bp_exit_check) = tokio::sync::watch::channel(()); - let background_processor = tokio::spawn(process_events_async( + let mut background_processor = tokio::spawn(process_events_async( Arc::clone(&persister), event_handler, chain_monitor.clone(), @@ -826,7 +944,7 @@ async fn start_ldk() { // Regularly reconnect to channel peers. let connect_cm = Arc::clone(&channel_manager); let connect_pm = Arc::clone(&peer_manager); - let peer_data_path = format!("{}/channel_peer_data", ldk_data_dir.clone()); + let peer_data_path = format!("{}/channel_peer_data", ldk_data_dir); let stop_connect = Arc::clone(&stop_listen_connect); tokio::spawn(async move { let mut interval = tokio::time::interval(Duration::from_secs(1)); @@ -898,29 +1016,68 @@ async fn start_ldk() { )); // Start the CLI. - cli::poll_for_user_input( - Arc::clone(&peer_manager), - Arc::clone(&channel_manager), - Arc::clone(&keys_manager), - Arc::clone(&network_graph), - Arc::clone(&onion_messenger), - inbound_payments, - outbound_payments, - ldk_data_dir, - network, - Arc::clone(&logger), - Arc::clone(&persister), - ) - .await; + let cli_channel_manager = Arc::clone(&channel_manager); + let cli_persister = Arc::clone(&persister); + let cli_logger = Arc::clone(&logger); + let cli_peer_manager = Arc::clone(&peer_manager); + let cli_poll = tokio::task::spawn_blocking(move || { + cli::poll_for_user_input( + cli_peer_manager, + cli_channel_manager, + keys_manager, + network_graph, + onion_messenger, + inbound_payments, + outbound_payments, + ldk_data_dir, + network, + cli_logger, + cli_persister, + ) + }); + + // Exit if either CLI polling exits or the background processor exits (which shouldn't happen + // unless we fail to write to the filesystem). + let mut bg_res = Ok(Ok(())); + tokio::select! { + _ = cli_poll => {}, + bg_exit = &mut background_processor => { + bg_res = bg_exit; + }, + } // Disconnect our peers and stop accepting new connections. This ensures we don't continue // updating our channel data after we've stopped the background processor. stop_listen_connect.store(true, Ordering::Release); peer_manager.disconnect_all_peers(); + if let Err(e) = bg_res { + let persist_res = persister + .write( + persist::CHANNEL_MANAGER_PERSISTENCE_PRIMARY_NAMESPACE, + persist::CHANNEL_MANAGER_PERSISTENCE_SECONDARY_NAMESPACE, + persist::CHANNEL_MANAGER_PERSISTENCE_KEY, + &channel_manager.encode(), + ) + .unwrap(); + use lightning::util::logger::Logger; + lightning::log_error!( + &*logger, + "Last-ditch ChannelManager persistence result: {:?}", + persist_res + ); + panic!( + "ERR: background processing stopped with result {:?}, exiting.\n\ + Last-ditch ChannelManager persistence result {:?}", + e, persist_res + ); + } + // Stop the background processor. - bp_exit.send(()).unwrap(); - background_processor.await.unwrap().unwrap(); + if !bp_exit.is_closed() { + bp_exit.send(()).unwrap(); + background_processor.await.unwrap().unwrap(); + } } #[tokio::main]