Try to sweep SpendableOutputs regularly, and in bulk

Rather than trying to sweep `SpendableOutputs` every time we see
one, try to sweep them in bulk once an hour. Then, try to sweep
them repeatedly every hour forever, even after they're claimed.

Fixes #102.

Author: TheBlueMatt

src/main.rs

@@ -15,7 +15,9 @@ use bitcoin::BlockHash;
 use bitcoin_bech32::WitnessProgram;
 use lightning::chain;
 use lightning::chain::chaininterface::{BroadcasterInterface, ConfirmationTarget, FeeEstimator};
-use lightning::chain::keysinterface::{EntropySource, InMemorySigner, KeysManager};
+use lightning::chain::keysinterface::{
+	EntropySource, InMemorySigner, KeysManager, SpendableOutputDescriptor
+};
 use lightning::chain::{chainmonitor, ChannelMonitorUpdateStatus};
 use lightning::chain::{Filter, Watch};
 use lightning::events::{Event, PaymentFailureReason, PaymentPurpose};
@@ -30,7 +32,9 @@ use lightning::routing::gossip;
 use lightning::routing::gossip::{NodeId, P2PGossipSync};
 use lightning::routing::router::DefaultRouter;
 use lightning::util::config::UserConfig;
-use lightning::util::ser::ReadableArgs;
+use lightning::util::logger::Logger;
+use lightning::util::persist::KVStorePersister;
+use lightning::util::ser::{Readable, ReadableArgs, WithoutLength};
 use lightning_background_processor::{process_events_async, GossipSync};
 use lightning_block_sync::init;
 use lightning_block_sync::poll;
@@ -46,8 +50,8 @@ use std::fmt;
 use std::fs;
 use std::fs::File;
 use std::io;
-use std::io::Write;
-use std::path::Path;
+use std::io::{Read, Seek, SeekFrom, Write};
+use std::path::{Path, PathBuf};
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex};
 use std::time::{Duration, SystemTime};
@@ -107,7 +111,7 @@ async fn handle_ldk_events(
 	channel_manager: &Arc<ChannelManager>, bitcoind_client: &BitcoindClient,
 	network_graph: &NetworkGraph, keys_manager: &KeysManager,
 	inbound_payments: &PaymentInfoStorage, outbound_payments: &PaymentInfoStorage,
-	network: Network, event: Event,
+	persister: &Arc<FilesystemPersister>, network: Network, event: Event,
 ) {
 	match event {
 		Event::FundingGenerationReady {
@@ -331,20 +335,21 @@ async fn handle_ldk_events(
 			});
 		}
 		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);
+			// 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
+			// from this method, the descriptor will be gone, and you may lose track of some funds.
+			//
+			// Here we simply persist them to disk, with a background task running which will try
+			// to spend them regularly (possibly duplicatively/RBF'ing them). These can just be
+			// treated as normal funds where possible - they are only spendable by us and there is
+			// no rush to claim them.
+			for output in outputs {
+				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_outputs/{}", key), &output).unwrap();
+			}
 		}
 		Event::ChannelPending { channel_id, counterparty_node_id, .. } => {
 			println!(
@@ -693,6 +698,7 @@ 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 network = args.network;
 	let event_handler = move |event: Event| {
 		let channel_manager_event_listener = Arc::clone(&channel_manager_event_listener);
@@ -701,6 +707,7 @@ async fn start_ldk() {
 		let keys_manager_event_listener = Arc::clone(&keys_manager_event_listener);
 		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);
 		async move {
 			handle_ldk_events(
 				&channel_manager_event_listener,
@@ -709,6 +716,7 @@ async fn start_ldk() {
 				&keys_manager_event_listener,
 				&inbound_payments_event_listener,
 				&outbound_payments_event_listener,
+				&persister_event_listener,
 				network,
 				event,
 			)
@@ -722,7 +730,7 @@ async fn start_ldk() {
 	// Step 20: Background Processing
 	let (bp_exit, bp_exit_check) = tokio::sync::watch::channel(());
 	let background_processor = tokio::spawn(process_events_async(
-		persister,
+		Arc::clone(&persister),
 		event_handler,
 		chain_monitor.clone(),
 		channel_manager.clone(),
@@ -800,6 +808,90 @@ async fn start_ldk() {
 		}
 	});
 
+	// Regularly claim outputs which are exclusively spendable by us and send them to Bitcoin Core.
+	// Note that if you more tightly integrate your wallet with LDK you may not need to do this -
+	// these outputs can just be treated as normal outputs during coin selection.
+	let pending_spendables_dir = format!("{}/pending_spendable_outputs", ldk_data_dir);
+	let processing_spendables_dir = format!("{}/processing_spendable_outputs", ldk_data_dir);
+	let spendables_dir = format!("{}/spendable_outputs", ldk_data_dir);
+	let spending_keys_manager = Arc::clone(&keys_manager);
+	let spending_logger = Arc::clone(&logger);
+	tokio::spawn(async move {
+		let mut interval = tokio::time::interval(Duration::from_secs(3600));
+		loop {
+			interval.tick().await;
+			if let Ok(dir_iter) = fs::read_dir(&pending_spendables_dir) {
+				// Move any spendable descriptors from pending folder so that we don't have any
+				// races with new files being added.
+				for file_res in dir_iter {
+					let file = file_res.unwrap();
+					// Only move a file if its a 32-byte-hex'd filename, otherwise it might be a
+					// temporary file.
+					if file.file_name().len() == 64 {
+						fs::create_dir_all(&processing_spendables_dir).unwrap();
+						let mut holding_path = PathBuf::new();
+						holding_path.push(&processing_spendables_dir);
+						holding_path.push(&file.file_name());
+						fs::rename(file.path(), holding_path).unwrap();
+					}
+				}
+				// Now concatenate all the pending files we moved into one file in the
+				// `spendable_outputs` directory and drop the processing directory.
+				let mut outputs = Vec::new();
+				if let Ok(processing_iter) = fs::read_dir(&processing_spendables_dir) {
+					for file_res in processing_iter {
+						outputs.append(&mut fs::read(file_res.unwrap().path()).unwrap());
+					}
+				}
+				if !outputs.is_empty() {
+					let key = hex_utils::hex_str(&spending_keys_manager.get_secure_random_bytes());
+					persister
+						.persist(&format!("spendable_outputs/{}", key), &WithoutLength(&outputs))
+						.unwrap();
+					fs::remove_dir_all(&processing_spendables_dir).unwrap();
+				}
+			}
+			// Iterate over all the sets of spendable outputs in `spendables_dir` and try to claim
+			// them.
+			if let Ok(dir_iter) = fs::read_dir(&spendables_dir) {
+				for file_res in dir_iter {
+					let mut outputs: Vec<SpendableOutputDescriptor> = Vec::new();
+					let mut file = File::open(file_res.unwrap().path()).unwrap();
+					loop {
+						// Check if there are any bytes left to read, and if so read a descriptor.
+						match file.read_exact(&mut [0; 1]) {
+							Ok(_) => {
+								file.seek(SeekFrom::Current(-1)).unwrap();
+							},
+							Err(e) if e.kind() == io::ErrorKind::UnexpectedEof => break,
+							Err(e) => Err(e).unwrap(),
+						}
+						outputs.push(Readable::read(&mut file).unwrap());
+					}
+					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);
+					if let Ok(spending_tx) = spending_keys_manager.spend_spendable_outputs(
+						output_descriptors,
+						Vec::new(),
+						destination_address.script_pubkey(),
+						tx_feerate,
+						&Secp256k1::new(),
+					) {
+						// Note that, most likely, we've already sweeped this set of outputs
+						// and they're already confirmed on-chain, so this broadcast will fail.
+						bitcoind_client.broadcast_transaction(&spending_tx);
+					} else {
+						lightning::log_error!(
+							spending_logger,
+							"Failed to sweep spendable outputs! This may indicate the outputs are dust. Will try again in an hour.");
+					}
+				}
+			}
+		}
+	});
+
 	// Start the CLI.
 	cli::poll_for_user_input(
 		Arc::clone(&peer_manager),
@@ -809,7 +901,7 @@ async fn start_ldk() {
 		Arc::clone(&onion_messenger),
 		inbound_payments,
 		outbound_payments,
-		ldk_data_dir.clone(),
+		ldk_data_dir,
 		network,
 		Arc::clone(&logger),
 	)