Integrate OutputSweeper

src/builder.rs

@@ -7,6 +7,7 @@ use crate::io::{KVStore, CHANNEL_MANAGER_PERSISTENCE_KEY, CHANNEL_MANAGER_PERSIS
 use crate::logger::{log_error, FilesystemLogger, Logger};
 use crate::payment_store::PaymentStore;
 use crate::peer_store::PeerStore;
+use crate::sweep::OutputSweeper;
 use crate::types::{
 	ChainMonitor, ChannelManager, FakeMessageRouter, GossipSync, KeysManager, NetAddress,
 	NetworkGraph, OnionMessenger, PeerManager,
@@ -716,6 +717,23 @@ fn build_with_store_internal<K: KVStore + Sync + Send + 'static>(
 		}
 	};
 
+	let best_block = channel_manager.current_best_block();
+	let output_sweeper =
+		match io::utils::read_spendable_outputs(Arc::clone(&kv_store), Arc::clone(&logger)) {
+			Ok(outputs) => Arc::new(OutputSweeper::new(
+				outputs,
+				Arc::clone(&wallet),
+				Arc::clone(&keys_manager),
+				Arc::clone(&kv_store),
+				best_block,
+				Some(Arc::clone(&tx_sync)),
+				Arc::clone(&logger),
+			)),
+			Err(_) => {
+				return Err(BuildError::ReadFailed);
+			}
+		};
+
 	let (stop_sender, stop_receiver) = tokio::sync::watch::channel(());
 
 	Ok(Node {
@@ -728,6 +746,7 @@ fn build_with_store_internal<K: KVStore + Sync + Send + 'static>(
 		event_queue,
 		channel_manager,
 		chain_monitor,
+		output_sweeper,
 		peer_manager,
 		keys_manager,
 		network_graph,

src/event.rs

@@ -9,6 +9,7 @@ use crate::payment_store::{
 
 use crate::io::{KVStore, EVENT_QUEUE_PERSISTENCE_KEY, EVENT_QUEUE_PERSISTENCE_NAMESPACE};
 use crate::logger::{log_debug, log_error, log_info, Logger};
+use crate::types::Sweeper;
 
 use lightning::chain::chaininterface::{BroadcasterInterface, ConfirmationTarget, FeeEstimator};
 use lightning::events::Event as LdkEvent;
@@ -229,6 +230,7 @@ where
 	network_graph: Arc<NetworkGraph>,
 	keys_manager: Arc<KeysManager>,
 	payment_store: Arc<PaymentStore<K, L>>,
+	output_sweeper: Arc<Sweeper<K>>,
 	runtime: Arc<RwLock<Option<tokio::runtime::Runtime>>>,
 	logger: L,
 	config: Arc<Config>,
@@ -242,7 +244,8 @@ where
 		wallet: Arc<Wallet<bdk::database::SqliteDatabase, L>>, event_queue: Arc<EventQueue<K, L>>,
 		channel_manager: Arc<ChannelManager<K>>, network_graph: Arc<NetworkGraph>,
 		keys_manager: Arc<KeysManager>, payment_store: Arc<PaymentStore<K, L>>,
-		runtime: Arc<RwLock<Option<tokio::runtime::Runtime>>>, logger: L, config: Arc<Config>,
+		output_sweeper: Arc<Sweeper<K>>, runtime: Arc<RwLock<Option<tokio::runtime::Runtime>>>,
+		logger: L, config: Arc<Config>,
 	) -> Self {
 		Self {
 			event_queue,
@@ -251,6 +254,7 @@ where
 			network_graph,
 			keys_manager,
 			payment_store,
+			output_sweeper,
 			logger,
 			runtime,
 			config,
@@ -553,36 +557,10 @@ where
 				}
 			}
 			LdkEvent::SpendableOutputs { outputs } => {
-				// TODO: We should eventually remember the outputs and supply them to the wallet's coin selection, once BDK allows us to do so.
-				let destination_address = self.wallet.get_new_address().unwrap_or_else(|e| {
-					log_error!(self.logger, "Failed to get destination address: {}", e);
-					panic!("Failed to get destination address");
-				});
-
-				let output_descriptors = &outputs.iter().collect::<Vec<_>>();
-				let tx_feerate =
-					self.wallet.get_est_sat_per_1000_weight(ConfirmationTarget::Normal);
-
-				// We set nLockTime to the current height to discourage fee sniping.
-				let cur_height = self.channel_manager.current_best_block().height();
-				let locktime: PackedLockTime =
-					LockTime::from_height(cur_height).map_or(PackedLockTime::ZERO, |l| l.into());
-				let res = self.keys_manager.spend_spendable_outputs(
-					output_descriptors,
-					Vec::new(),
-					destination_address.script_pubkey(),
-					tx_feerate,
-					Some(locktime),
-					&Secp256k1::new(),
-				);
-
-				match res {
-					Ok(Some(spending_tx)) => self.wallet.broadcast_transactions(&[&spending_tx]),
-					Ok(None) => {
-						log_debug!(self.logger, "Omitted spending static outputs: {:?}", outputs);
-					}
-					Err(err) => {
-						log_error!(self.logger, "Error spending outputs: {:?}", err);
+				match self.output_sweeper.add_outputs(outputs) {
+					Ok(()) => {}
+					Err(e) => {
+						log_error!(self.logger, "Error spending outputs: {:?}", e);
 					}
 				}
 			}

src/lib.rs

@@ -120,7 +120,9 @@ use io::KVStore;
 use payment_store::PaymentStore;
 pub use payment_store::{PaymentDetails, PaymentDirection, PaymentStatus};
 use peer_store::{PeerInfo, PeerStore};
-use types::{ChainMonitor, ChannelManager, KeysManager, NetworkGraph, PeerManager, Router, Scorer};
+use types::{
+	ChainMonitor, ChannelManager, KeysManager, NetworkGraph, PeerManager, Router, Scorer, Sweeper,
+};
 pub use types::{ChannelDetails, ChannelId, PeerDetails, UserChannelId};
 use wallet::Wallet;
 
@@ -280,6 +282,7 @@ pub struct Node<K: KVStore + Sync + Send + 'static> {
 	event_queue: Arc<EventQueue<K, Arc<FilesystemLogger>>>,
 	channel_manager: Arc<ChannelManager<K>>,
 	chain_monitor: Arc<ChainMonitor<K>>,
+	output_sweeper: Arc<Sweeper<K>>,
 	peer_manager: Arc<PeerManager<K>>,
 	keys_manager: Arc<KeysManager>,
 	network_graph: Arc<NetworkGraph>,
@@ -403,6 +406,7 @@ impl<K: KVStore + Sync + Send + 'static> Node<K> {
 		let tx_sync = Arc::clone(&self.tx_sync);
 		let sync_cman = Arc::clone(&self.channel_manager);
 		let sync_cmon = Arc::clone(&self.chain_monitor);
+		let sync_sweeper = Arc::clone(&self.output_sweeper);
 		let sync_logger = Arc::clone(&self.logger);
 		let mut stop_sync = self.stop_receiver.clone();
 		let wallet_sync_interval_secs =
@@ -420,6 +424,7 @@ impl<K: KVStore + Sync + Send + 'static> Node<K> {
 						let confirmables = vec![
 							&*sync_cman as &(dyn Confirm + Sync + Send),
 							&*sync_cmon as &(dyn Confirm + Sync + Send),
+							&*sync_sweeper as &(dyn Confirm + Sync + Send),
 						];
 						let now = Instant::now();
 						match tx_sync.sync(confirmables).await {
@@ -644,6 +649,7 @@ impl<K: KVStore + Sync + Send + 'static> Node<K> {
 			Arc::clone(&self.network_graph),
 			Arc::clone(&self.keys_manager),
 			Arc::clone(&self.payment_store),
+			Arc::clone(&self.output_sweeper),
 			Arc::clone(&self.runtime),
 			Arc::clone(&self.logger),
 			Arc::clone(&self.config),
@@ -974,10 +980,12 @@ impl<K: KVStore + Sync + Send + 'static> Node<K> {
 		let tx_sync = Arc::clone(&self.tx_sync);
 		let sync_cman = Arc::clone(&self.channel_manager);
 		let sync_cmon = Arc::clone(&self.chain_monitor);
+		let sync_sweeper = Arc::clone(&self.output_sweeper);
 		let sync_logger = Arc::clone(&self.logger);
 		let confirmables = vec![
 			&*sync_cman as &(dyn Confirm + Sync + Send),
 			&*sync_cmon as &(dyn Confirm + Sync + Send),
+			&*sync_sweeper as &(dyn Confirm + Sync + Send),
 		];
 
 		tokio::task::block_in_place(move || {

src/types.rs

@@ -1,4 +1,5 @@
 use crate::logger::FilesystemLogger;
+use crate::sweep::OutputSweeper;
 use crate::wallet::{Wallet, WalletKeysManager};
 
 use lightning::chain::chainmonitor;
@@ -107,6 +108,9 @@ impl lightning::onion_message::MessageRouter for FakeMessageRouter {
 	}
 }
 
+pub(crate) type Sweeper<K> =
+	OutputSweeper<K, Arc<EsploraSyncClient<Arc<FilesystemLogger>>>, Arc<FilesystemLogger>>;
+
 /// The global identifier of a channel.
 ///
 /// Note that this will start out to be a temporary ID until channel funding negotiation is