Add OutputSweeper and persistence utils

We add an `OutputSweeper` object that will keep track of sweepable
outputs. To this end, we start by adding the general structures and the
required utilities to persist the `SpendableOutputInfo` to our
`KVStore`.

Author: tnull

src/io/mod.rs

@@ -19,6 +19,10 @@ pub(crate) const PEER_INFO_PERSISTENCE_KEY: &str = "peers";
 pub(crate) const PAYMENT_INFO_PERSISTENCE_PRIMARY_NAMESPACE: &str = "payments";
 pub(crate) const PAYMENT_INFO_PERSISTENCE_SECONDARY_NAMESPACE: &str = "";
 
+/// The spendable output information will be persisted under this prefix.
+pub(crate) const SPENDABLE_OUTPUT_INFO_PERSISTENCE_PRIMARY_NAMESPACE: &str = "spendable_outputs";
+pub(crate) const SPENDABLE_OUTPUT_INFO_PERSISTENCE_SECONDARY_NAMESPACE: &str = "";
+
 /// RapidGossipSync's `latest_sync_timestamp` will be persisted under this key.
 pub(crate) const LATEST_RGS_SYNC_TIMESTAMP_PRIMARY_NAMESPACE: &str = "";
 pub(crate) const LATEST_RGS_SYNC_TIMESTAMP_SECONDARY_NAMESPACE: &str = "";

src/io/utils.rs

@@ -3,6 +3,7 @@ use crate::WALLET_KEYS_SEED_LEN;
 
 use crate::logger::log_error;
 use crate::peer_store::PeerStore;
+use crate::sweep::SpendableOutputInfo;
 use crate::{Error, EventQueue, PaymentDetails};
 
 use lightning::routing::gossip::NetworkGraph;
@@ -199,6 +200,36 @@ where
 	Ok(res)
 }
 
+/// Read previously persisted spendable output information from the store.
+pub(crate) fn read_spendable_outputs<K: KVStore + Sync + Send, L: Deref>(
+	kv_store: Arc<K>, logger: L,
+) -> Result<Vec<SpendableOutputInfo>, std::io::Error>
+where
+	L::Target: Logger,
+{
+	let mut res = Vec::new();
+
+	for stored_key in kv_store.list(
+		SPENDABLE_OUTPUT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
+		SPENDABLE_OUTPUT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
+	)? {
+		let mut reader = Cursor::new(kv_store.read(
+			SPENDABLE_OUTPUT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
+			SPENDABLE_OUTPUT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
+			&stored_key,
+		)?);
+		let output = SpendableOutputInfo::read(&mut reader).map_err(|e| {
+			log_error!(logger, "Failed to deserialize SpendableOutputInfo: {}", e);
+			std::io::Error::new(
+				std::io::ErrorKind::InvalidData,
+				"Failed to deserialize SpendableOutputInfo",
+			)
+		})?;
+		res.push(output);
+	}
+	Ok(res)
+}
+
 pub(crate) fn read_latest_rgs_sync_timestamp<K: KVStore + Sync + Send, L: Deref>(
 	kv_store: Arc<K>, logger: L,
 ) -> Result<u32, std::io::Error>

src/lib.rs

@@ -84,6 +84,7 @@ pub mod io;
 mod logger;
 mod payment_store;
 mod peer_store;
+mod sweep;
 #[cfg(test)]
 mod test;
 mod types;

src/sweep.rs

@@ -0,0 +1,155 @@
+use crate::hex_utils;
+use crate::io::{
+	SPENDABLE_OUTPUT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
+	SPENDABLE_OUTPUT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
+};
+use crate::logger::{log_debug, log_error, Logger};
+use crate::wallet::Wallet;
+use crate::{Error, KeysManager};
+
+use lightning::chain::chaininterface::{BroadcasterInterface, ConfirmationTarget, FeeEstimator};
+use lightning::chain::BestBlock;
+use lightning::impl_writeable_tlv_based;
+use lightning::sign::{EntropySource, SpendableOutputDescriptor};
+use lightning::util::persist::KVStore;
+use lightning::util::ser::Writeable;
+
+use bitcoin::secp256k1::Secp256k1;
+use bitcoin::{BlockHash, LockTime, PackedLockTime, Transaction};
+
+use std::ops::Deref;
+use std::sync::{Arc, Mutex};
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub(crate) struct SpendableOutputInfo {
+	id: [u8; 32],
+	descriptor: SpendableOutputDescriptor,
+	spending_tx: Option<Transaction>,
+	broadcast_height: Option<u32>,
+	confirmed_in_block: Option<(u32, BlockHash)>,
+}
+
+impl_writeable_tlv_based!(SpendableOutputInfo, {
+	(0, id, required),
+	(2, descriptor, required),
+	(4, spending_tx, option),
+	(6, broadcast_height, option),
+	(8, confirmed_in_block, option),
+});
+
+pub(crate) struct OutputSweeper<K: KVStore + Sync + Send, L: Deref>
+where
+	L::Target: Logger,
+{
+	outputs: Mutex<Vec<SpendableOutputInfo>>,
+	wallet: Arc<Wallet<bdk::database::SqliteDatabase, L>>,
+	keys_manager: Arc<KeysManager>,
+	kv_store: Arc<K>,
+	best_block: Mutex<BestBlock>,
+	logger: L,
+}
+
+impl<K: KVStore + Sync + Send, L: Deref> OutputSweeper<K, L>
+where
+	L::Target: Logger,
+{
+	pub(crate) fn new(
+		outputs: Vec<SpendableOutputInfo>, wallet: Arc<Wallet<bdk::database::SqliteDatabase, L>>,
+		keys_manager: Arc<KeysManager>, kv_store: Arc<K>, best_block: BestBlock, logger: L,
+	) -> Self {
+		let outputs = Mutex::new(outputs);
+		let best_block = Mutex::new(best_block);
+		Self { outputs, wallet, keys_manager, kv_store, best_block, logger }
+	}
+
+	pub(crate) fn add_outputs(&self, output_descriptors: Vec<SpendableOutputDescriptor>) {
+		let mut locked_outputs = self.outputs.lock().unwrap();
+
+		let cur_height = self.best_block.lock().unwrap().height();
+
+		let (spending_tx, broadcast_height) =
+			match self.get_spending_tx(&output_descriptors, cur_height) {
+				Ok(Some(spending_tx)) => {
+					self.wallet.broadcast_transactions(&[&spending_tx]);
+					(Some(spending_tx), Some(cur_height))
+				}
+				Ok(None) => {
+					log_debug!(
+						self.logger,
+						"Omitted spending static outputs: {:?}",
+						output_descriptors
+					);
+					(None, None)
+				}
+				Err(e) => {
+					log_error!(self.logger, "Error spending outputs: {:?}", e);
+					(None, None)
+				}
+			};
+
+		for descriptor in output_descriptors {
+			let id = self.keys_manager.get_secure_random_bytes();
+			let output_info = SpendableOutputInfo {
+				id,
+				descriptor,
+				spending_tx: spending_tx.clone(),
+				broadcast_height,
+				confirmed_in_block: None,
+			};
+
+			locked_outputs.push(output_info.clone());
+			match self.persist_info(&output_info) {
+				Ok(()) => {}
+				Err(e) => {
+					log_error!(self.logger, "Error persisting spendable output info: {:?}", e)
+				}
+			}
+		}
+	}
+
+	fn get_spending_tx(
+		&self, output_descriptors: &Vec<SpendableOutputDescriptor>, cur_height: u32,
+	) -> Result<Option<Transaction>, ()> {
+		let tx_feerate =
+			self.wallet.get_est_sat_per_1000_weight(ConfirmationTarget::NonAnchorChannelFee);
+
+		let destination_address = self.wallet.get_new_address().map_err(|e| {
+			log_error!(self.logger, "Failed to get destination address from wallet: {}", e);
+		})?;
+
+		let locktime: PackedLockTime =
+			LockTime::from_height(cur_height).map_or(PackedLockTime::ZERO, |l| l.into());
+
+		self.keys_manager.spend_spendable_outputs(
+			&output_descriptors.iter().collect::<Vec<_>>(),
+			Vec::new(),
+			destination_address.script_pubkey(),
+			tx_feerate,
+			Some(locktime),
+			&Secp256k1::new(),
+		)
+	}
+
+	fn persist_info(&self, output: &SpendableOutputInfo) -> Result<(), Error> {
+		let key = hex_utils::to_string(&output.id);
+		let data = output.encode();
+		self.kv_store
+			.write(
+				SPENDABLE_OUTPUT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
+				SPENDABLE_OUTPUT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
+				&key,
+				&data,
+			)
+			.map_err(|e| {
+				log_error!(
+					self.logger,
+					"Write for key {}/{}/{} failed due to: {}",
+					SPENDABLE_OUTPUT_INFO_PERSISTENCE_PRIMARY_NAMESPACE,
+					SPENDABLE_OUTPUT_INFO_PERSISTENCE_SECONDARY_NAMESPACE,
+					key,
+					e
+				);
+				Error::PersistenceFailed
+			})
+	}
+}