From a368093803dc66ae19a00a6dbbdb6650656b2379 Mon Sep 17 00:00:00 2001 From: Valentine Wallace Date: Mon, 11 Jan 2021 18:03:32 -0500 Subject: [PATCH] Add BackgroundProcessor for ChannelManager persistence and other Other includes calling timer_chan_freshness_every_minute() and in the future, possibly persisting channel graph data. This struct is suitable for things that need to happen periodically and can happen in the background. --- Cargo.toml | 1 + background-processor/Cargo.toml | 15 ++ background-processor/src/lib.rs | 294 ++++++++++++++++++++++++++++++++ lightning-persister/src/lib.rs | 41 ++++- lightning-persister/src/util.rs | 2 +- 5 files changed, 350 insertions(+), 3 deletions(-) create mode 100644 background-processor/Cargo.toml create mode 100644 background-processor/src/lib.rs diff --git a/Cargo.toml b/Cargo.toml index 96f4b1d17..1ffd75a87 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -5,6 +5,7 @@ members = [ "lightning-block-sync", "lightning-net-tokio", "lightning-persister", + "background-processor", ] # Our tests do actual crypo and lots of work, the tradeoff for -O1 is well worth it. diff --git a/background-processor/Cargo.toml b/background-processor/Cargo.toml new file mode 100644 index 000000000..71fbbbff8 --- /dev/null +++ b/background-processor/Cargo.toml @@ -0,0 +1,15 @@ +[package] +name = "background-processor" +version = "0.1.0" +authors = ["Valentine Wallace "] +edition = "2018" + +# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html + +[dependencies] +bitcoin = "0.24" +lightning = { version = "0.0.12", path = "../lightning", features = ["allow_wallclock_use"] } +lightning-persister = { version = "0.0.1", path = "../lightning-persister" } + +[dev-dependencies] +lightning = { version = "0.0.12", path = "../lightning", features = ["_test_utils"] } diff --git a/background-processor/src/lib.rs b/background-processor/src/lib.rs new file mode 100644 index 000000000..63b9ef311 --- /dev/null +++ b/background-processor/src/lib.rs @@ -0,0 +1,294 @@ +#[macro_use] extern crate lightning; + +use lightning::chain; +use lightning::chain::chaininterface::{BroadcasterInterface, FeeEstimator}; +use lightning::chain::keysinterface::{ChannelKeys, KeysInterface}; +use lightning::ln::channelmanager::ChannelManager; +use lightning::util::logger::Logger; +use lightning::util::ser::Writeable; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::thread; +use std::thread::JoinHandle; +use std::time::{Duration, Instant}; + +/// BackgroundProcessor takes care of tasks that (1) need to happen periodically to keep +/// Rust-Lightning running properly, and (2) either can or should be run in the background. Its +/// responsibilities are: +/// * Monitoring whether the ChannelManager needs to be re-persisted to disk, and if so, +/// writing it to disk/backups by invoking the callback given to it at startup. +/// ChannelManager persistence should be done in the background. +/// * Calling `ChannelManager::timer_chan_freshness_every_min()` every minute (can be done in the +/// background). +/// +/// Note that if ChannelManager persistence fails and the persisted manager becomes out-of-date, +/// then there is a risk of channels force-closing on startup when the manager realizes it's +/// outdated. However, as long as `ChannelMonitor` backups are sound, no funds besides those used +/// for unilateral chain closure fees are at risk. +pub struct BackgroundProcessor { + stop_thread: Arc, + /// May be used to retrieve and handle the error if `BackgroundProcessor`'s thread + /// exits due to an error while persisting. + pub thread_handle: JoinHandle>, +} + +#[cfg(not(test))] +const CHAN_FRESHNESS_TIMER: u64 = 60; +#[cfg(test)] +const CHAN_FRESHNESS_TIMER: u64 = 1; + +impl BackgroundProcessor { + /// Start a background thread that takes care of responsibilities enumerated in the top-level + /// documentation. + /// + /// If `persist_manager` returns an error, then this thread will return said error (and `start()` + /// will need to be called again to restart the `BackgroundProcessor`). Users should wait on + /// [`thread_handle`]'s `join()` method to be able to tell if and when an error is returned, or + /// implement `persist_manager` such that an error is never returned to the `BackgroundProcessor` + /// + /// `persist_manager` is responsible for writing out the `ChannelManager` to disk, and/or uploading + /// to one or more backup services. See [`ChannelManager::write`] for writing out a `ChannelManager`. + /// See [`FilesystemPersister::persist_manager`] for Rust-Lightning's provided implementation. + /// + /// [`thread_handle`]: struct.BackgroundProcessor.html#structfield.thread_handle + /// [`ChannelManager::write`]: ../lightning/ln/channelmanager/struct.ChannelManager.html#method.write + /// [`FilesystemPersister::persist_manager`]: ../lightning_persister/struct.FilesystemPersister.html#impl + pub fn start(persist_manager: PM, manager: Arc, Arc, Arc, Arc, Arc>>, logger: Arc) -> Self + where ChanSigner: 'static + ChannelKeys + Writeable, + M: 'static + chain::Watch, + T: 'static + BroadcasterInterface, + K: 'static + KeysInterface, + F: 'static + FeeEstimator, + L: 'static + Logger, + PM: 'static + Send + Fn(&ChannelManager, Arc, Arc, Arc, Arc>) -> Result<(), std::io::Error>, + { + let stop_thread = Arc::new(AtomicBool::new(false)); + let stop_thread_clone = stop_thread.clone(); + let handle = thread::spawn(move || -> Result<(), std::io::Error> { + let mut current_time = Instant::now(); + loop { + let updates_available = manager.wait_timeout(Duration::from_millis(100)); + if updates_available { + persist_manager(&*manager)?; + } + // Exit the loop if the background processor was requested to stop. + if stop_thread.load(Ordering::Acquire) == true { + log_trace!(logger, "Terminating background processor."); + return Ok(()) + } + if current_time.elapsed().as_secs() > CHAN_FRESHNESS_TIMER { + log_trace!(logger, "Calling manager's timer_chan_freshness_every_min"); + manager.timer_chan_freshness_every_min(); + current_time = Instant::now(); + } + } + }); + Self { + stop_thread: stop_thread_clone, + thread_handle: handle, + } + } + + /// Stop `BackgroundProcessor`'s thread. + pub fn stop(self) -> Result<(), std::io::Error> { + self.stop_thread.store(true, Ordering::Release); + self.thread_handle.join().unwrap() + } +} + +#[cfg(test)] +mod tests { + use bitcoin::blockdata::constants::genesis_block; + use bitcoin::blockdata::transaction::{Transaction, TxOut}; + use bitcoin::network::constants::Network; + use lightning::chain; + use lightning::chain::chaininterface::{BroadcasterInterface, FeeEstimator}; + use lightning::chain::chainmonitor; + use lightning::chain::keysinterface::{ChannelKeys, InMemoryChannelKeys, KeysInterface, KeysManager}; + use lightning::chain::transaction::OutPoint; + use lightning::get_event_msg; + use lightning::ln::channelmanager::{ChannelManager, SimpleArcChannelManager}; + use lightning::ln::features::InitFeatures; + use lightning::ln::msgs::ChannelMessageHandler; + use lightning::util::config::UserConfig; + use lightning::util::events::{Event, EventsProvider, MessageSendEventsProvider, MessageSendEvent}; + use lightning::util::logger::Logger; + use lightning::util::ser::Writeable; + use lightning::util::test_utils; + use lightning_persister::FilesystemPersister; + use std::fs; + use std::path::PathBuf; + use std::sync::{Arc, Mutex}; + use std::time::Duration; + use super::BackgroundProcessor; + + type ChainMonitor = chainmonitor::ChainMonitor, Arc, Arc, Arc, Arc>; + + struct Node { + node: SimpleArcChannelManager, + persister: Arc, + logger: Arc, + } + + impl Drop for Node { + fn drop(&mut self) { + let data_dir = self.persister.get_data_dir(); + match fs::remove_dir_all(data_dir.clone()) { + Err(e) => println!("Failed to remove test persister directory {}: {}", data_dir, e), + _ => {} + } + } + } + + fn get_full_filepath(filepath: String, filename: String) -> String { + let mut path = PathBuf::from(filepath); + path.push(filename); + path.to_str().unwrap().to_string() + } + + fn create_nodes(num_nodes: usize, persist_dir: String) -> Vec { + let mut nodes = Vec::new(); + for i in 0..num_nodes { + let tx_broadcaster = Arc::new(test_utils::TestBroadcaster{txn_broadcasted: Mutex::new(Vec::new())}); + let fee_estimator = Arc::new(test_utils::TestFeeEstimator { sat_per_kw: 253 }); + let chain_source = Arc::new(test_utils::TestChainSource::new(Network::Testnet)); + let logger = Arc::new(test_utils::TestLogger::with_id(format!("node {}", i))); + let persister = Arc::new(FilesystemPersister::new(format!("{}_persister_{}", persist_dir, i))); + let seed = [i as u8; 32]; + let network = Network::Testnet; + let now = Duration::from_secs(genesis_block(network).header.time as u64); + let keys_manager = Arc::new(KeysManager::new(&seed, now.as_secs(), now.subsec_nanos())); + let chain_monitor = Arc::new(chainmonitor::ChainMonitor::new(Some(chain_source.clone()), tx_broadcaster.clone(), logger.clone(), fee_estimator.clone(), persister.clone())); + let manager = Arc::new(ChannelManager::new(Network::Testnet, fee_estimator.clone(), chain_monitor.clone(), tx_broadcaster, logger.clone(), keys_manager.clone(), UserConfig::default(), i)); + let node = Node { node: manager, persister, logger }; + nodes.push(node); + } + nodes + } + + macro_rules! open_channel { + ($node_a: expr, $node_b: expr, $channel_value: expr) => {{ + $node_a.node.create_channel($node_b.node.get_our_node_id(), $channel_value, 100, 42, None).unwrap(); + $node_b.node.handle_open_channel(&$node_a.node.get_our_node_id(), InitFeatures::known(), &get_event_msg!($node_a, MessageSendEvent::SendOpenChannel, $node_b.node.get_our_node_id())); + $node_a.node.handle_accept_channel(&$node_b.node.get_our_node_id(), InitFeatures::known(), &get_event_msg!($node_b, MessageSendEvent::SendAcceptChannel, $node_a.node.get_our_node_id())); + let events = $node_a.node.get_and_clear_pending_events(); + assert_eq!(events.len(), 1); + let (temporary_channel_id, tx, funding_output) = match events[0] { + Event::FundingGenerationReady { ref temporary_channel_id, ref channel_value_satoshis, ref output_script, user_channel_id } => { + assert_eq!(*channel_value_satoshis, $channel_value); + assert_eq!(user_channel_id, 42); + + let tx = Transaction { version: 1 as i32, lock_time: 0, input: Vec::new(), output: vec![TxOut { + value: *channel_value_satoshis, script_pubkey: output_script.clone(), + }]}; + let funding_outpoint = OutPoint { txid: tx.txid(), index: 0 }; + (*temporary_channel_id, tx, funding_outpoint) + }, + _ => panic!("Unexpected event"), + }; + + $node_a.node.funding_transaction_generated(&temporary_channel_id, funding_output); + $node_b.node.handle_funding_created(&$node_a.node.get_our_node_id(), &get_event_msg!($node_a, MessageSendEvent::SendFundingCreated, $node_b.node.get_our_node_id())); + $node_a.node.handle_funding_signed(&$node_b.node.get_our_node_id(), &get_event_msg!($node_b, MessageSendEvent::SendFundingSigned, $node_a.node.get_our_node_id())); + tx + }} + } + + #[test] + fn test_background_processor() { + // Test that when a new channel is created, the ChannelManager needs to be re-persisted with + // updates. Also test that when new updates are available, the manager signals that it needs + // re-persistence and is successfully re-persisted. + let nodes = create_nodes(2, "test_background_processor".to_string()); + + // Initiate the background processors to watch each node. + let data_dir = nodes[0].persister.get_data_dir(); + let callback = move |node: &ChannelManager, Arc, Arc, Arc, Arc>| FilesystemPersister::persist_manager(data_dir.clone(), node); + let bg_processor = BackgroundProcessor::start(callback, nodes[0].node.clone(), nodes[0].logger.clone()); + + // Go through the channel creation process until each node should have something persisted. + let tx = open_channel!(nodes[0], nodes[1], 100000); + + macro_rules! check_persisted_data { + ($node: expr, $filepath: expr, $expected_bytes: expr) => { + match $node.write(&mut $expected_bytes) { + Ok(()) => { + loop { + match std::fs::read($filepath) { + Ok(bytes) => { + if bytes == $expected_bytes { + break + } else { + continue + } + }, + Err(_) => continue + } + } + }, + Err(e) => panic!("Unexpected error: {}", e) + } + } + } + + // Check that the initial channel manager data is persisted as expected. + let filepath = get_full_filepath("test_background_processor_persister_0".to_string(), "manager".to_string()); + let mut expected_bytes = Vec::new(); + check_persisted_data!(nodes[0].node, filepath.clone(), expected_bytes); + loop { + if !nodes[0].node.get_persistence_condvar_value() { break } + } + + // Force-close the channel. + nodes[0].node.force_close_channel(&OutPoint { txid: tx.txid(), index: 0 }.to_channel_id()).unwrap(); + + // Check that the force-close updates are persisted. + let mut expected_bytes = Vec::new(); + check_persisted_data!(nodes[0].node, filepath.clone(), expected_bytes); + loop { + if !nodes[0].node.get_persistence_condvar_value() { break } + } + + assert!(bg_processor.stop().is_ok()); + } + + #[test] + fn test_chan_freshness_called() { + // Test that ChannelManager's `timer_chan_freshness_every_min` is called every + // `CHAN_FRESHNESS_TIMER`. + let nodes = create_nodes(1, "test_chan_freshness_called".to_string()); + let data_dir = nodes[0].persister.get_data_dir(); + let callback = move |node: &ChannelManager, Arc, Arc, Arc, Arc>| FilesystemPersister::persist_manager(data_dir.clone(), node); + let bg_processor = BackgroundProcessor::start(callback, nodes[0].node.clone(), nodes[0].logger.clone()); + loop { + let log_entries = nodes[0].logger.lines.lock().unwrap(); + let desired_log = "Calling manager's timer_chan_freshness_every_min".to_string(); + if log_entries.get(&("background_processor".to_string(), desired_log)).is_some() { + break + } + } + + assert!(bg_processor.stop().is_ok()); + } + + #[test] + fn test_persist_error() { + // Test that if we encounter an error during manager persistence, the thread panics. + fn persist_manager(_data: &ChannelManager, Arc, Arc, Arc, Arc>) -> Result<(), std::io::Error> + where ChanSigner: 'static + ChannelKeys + Writeable, + M: 'static + chain::Watch, + T: 'static + BroadcasterInterface, + K: 'static + KeysInterface, + F: 'static + FeeEstimator, + L: 'static + Logger, + { + Err(std::io::Error::new(std::io::ErrorKind::Other, "test")) + } + + let nodes = create_nodes(2, "test_persist_error".to_string()); + let bg_processor = BackgroundProcessor::start(persist_manager, nodes[0].node.clone(), nodes[0].logger.clone()); + open_channel!(nodes[0], nodes[1], 100000); + + let _ = bg_processor.thread_handle.join().unwrap().expect_err("Errored persisting manager: test"); + } +} diff --git a/lightning-persister/src/lib.rs b/lightning-persister/src/lib.rs index da5cba95e..0257eb50e 100644 --- a/lightning-persister/src/lib.rs +++ b/lightning-persister/src/lib.rs @@ -6,17 +6,21 @@ extern crate libc; use bitcoin::hashes::hex::ToHex; use crate::util::DiskWriteable; +use lightning::chain; +use lightning::chain::chaininterface::{BroadcasterInterface, FeeEstimator}; use lightning::chain::channelmonitor::{ChannelMonitor, ChannelMonitorUpdate, ChannelMonitorUpdateErr}; use lightning::chain::channelmonitor; -use lightning::chain::keysinterface::ChannelKeys; +use lightning::chain::keysinterface::{ChannelKeys, KeysInterface}; use lightning::chain::transaction::OutPoint; +use lightning::ln::channelmanager::ChannelManager; +use lightning::util::logger::Logger; use lightning::util::ser::Writeable; use std::fs; use std::io::Error; +use std::sync::Arc; #[cfg(test)] use { - lightning::chain::keysinterface::KeysInterface, lightning::util::ser::ReadableArgs, bitcoin::{BlockHash, Txid}, bitcoin::hashes::hex::FromHex, @@ -46,6 +50,19 @@ impl DiskWriteable for ChannelMonitor { } } +impl DiskWriteable for ChannelManager, Arc, Arc, Arc, Arc> +where ChanSigner: ChannelKeys + Writeable, + M: chain::Watch, + T: BroadcasterInterface, + K: KeysInterface, + F: FeeEstimator, + L: Logger, +{ + fn write_to_file(&self, writer: &mut fs::File) -> Result<(), std::io::Error> { + self.write(writer) + } +} + impl FilesystemPersister { /// Initialize a new FilesystemPersister and set the path to the individual channels' /// files. @@ -55,6 +72,26 @@ impl FilesystemPersister { } } + pub fn get_data_dir(&self) -> String { + self.path_to_channel_data.clone() + } + + /// Writes the provided `ChannelManager` to the path provided at `FilesystemPersister` + /// initialization, within a file called "manager". + pub fn persist_manager( + data_dir: String, + manager: &ChannelManager, Arc, Arc, Arc, Arc> + ) -> Result<(), std::io::Error> + where ChanSigner: ChannelKeys + Writeable, + M: chain::Watch, + T: BroadcasterInterface, + K: KeysInterface, + F: FeeEstimator, + L: Logger + { + util::write_to_file(data_dir, "manager".to_string(), manager) + } + #[cfg(test)] fn load_channel_data(&self, keys: &Keys) -> Result>, ChannelMonitorUpdateErr> { diff --git a/lightning-persister/src/util.rs b/lightning-persister/src/util.rs index b7aa61d7a..daacb00f0 100644 --- a/lightning-persister/src/util.rs +++ b/lightning-persister/src/util.rs @@ -17,7 +17,7 @@ pub(crate) trait DiskWriteable { fn write_to_file(&self, writer: &mut fs::File) -> Result<(), std::io::Error>; } -pub fn get_full_filepath(filepath: String, filename: String) -> String { +pub(crate) fn get_full_filepath(filepath: String, filename: String) -> String { let mut path = PathBuf::from(filepath); path.push(filename); path.to_str().unwrap().to_string() -- 2.39.5