Implement RGB KVStore Integration

Author: dcorral

lightning/src/chain/channelmonitor.rs

@@ -7107,7 +7107,9 @@ mod tests {
 			SecretKey::from_slice(&[41; 32]).unwrap(),
 			[41; 32],
 			[0; 32],
+			std::path::PathBuf::new(),
 			[0; 32],
+			None,
 		);
 
 		let counterparty_pubkeys = ChannelPublicKeys {
@@ -7370,7 +7372,9 @@ mod tests {
 			SecretKey::from_slice(&[41; 32]).unwrap(),
 			[41; 32],
 			[0; 32],
+			std::path::PathBuf::new(),
 			[0; 32],
+			None,
 		);
 
 		let counterparty_pubkeys = ChannelPublicKeys {

lightning/src/chain/onchaintx.rs

@@ -1304,7 +1304,9 @@ mod tests {
 			SecretKey::from_slice(&[41; 32]).unwrap(),
 			[41; 32],
 			[0; 32],
+			std::path::PathBuf::new(),
 			[0; 32],
+			None,
 		);
 		let counterparty_pubkeys = ChannelPublicKeys {
 			funding_pubkey: PublicKey::from_secret_key(

lightning/src/ln/chan_utils.rs

@@ -11,6 +11,7 @@
 //! largely of interest for those implementing the traits on [`crate::sign`] by hand.
 
 use std::path::PathBuf;
+use std::sync::Arc;
 
 use bitcoin::amount::Amount;
 use bitcoin::constants::WITNESS_SCALE_FACTOR;
@@ -36,6 +37,7 @@ use crate::ln::msgs::DecodeError;
 use crate::rgb_utils::{color_htlc, is_tx_colored};
 use crate::sign::EntropySource;
 use crate::types::payment::{PaymentHash, PaymentPreimage};
+use crate::util::persist::KVStoreSync;
 use crate::util::ser::{Readable, ReadableArgs, RequiredWrapper, Writeable, Writer};
 use crate::util::transaction_utils;
 
@@ -2156,6 +2158,7 @@ impl<'a> TrustedCommitmentTransaction<'a> {
 	pub fn get_htlc_sigs<T: secp256k1::Signing, ES: Deref>(
 		&self, htlc_base_key: &SecretKey, channel_parameters: &DirectedChannelTransactionParameters,
 		entropy_source: &ES, secp_ctx: &Secp256k1<T>, ldk_data_dir: &PathBuf,
+		rgb_kv_store: Option<&Arc<dyn KVStoreSync + Send + Sync>>,
 	) -> Result<Vec<Signature>, ()> where ES::Target: EntropySource {
 		let inner = self.inner;
 		let keys = &inner.keys;
@@ -2167,7 +2170,7 @@ impl<'a> TrustedCommitmentTransaction<'a> {
 			assert!(this_htlc.transaction_output_index.is_some());
 			let mut htlc_tx = build_htlc_transaction(&txid, inner.feerate_per_kw, channel_parameters.contest_delay(), &this_htlc, &self.channel_type_features, &keys.broadcaster_delayed_payment_key, &keys.revocation_key);
 			if inner.is_colored() {
-				if let Err(_e) = color_htlc(&mut htlc_tx, this_htlc, ldk_data_dir) {
+				if let Err(_e) = color_htlc(&mut htlc_tx, this_htlc, ldk_data_dir, rgb_kv_store) {
 					return Err(());
 				}
 			}

lightning/src/ln/channel.rs

@@ -79,8 +79,8 @@ use crate::ln::LN_MAX_MSG_LEN;
 use crate::offers::static_invoice::StaticInvoice;
 use crate::rgb_utils::{
 	color_closing, color_commitment, color_htlc, get_rgb_channel_info_path,
-	get_rgb_channel_info_pending, parse_rgb_channel_info, rename_rgb_files,
-	update_rgb_channel_amount_pending,
+	get_rgb_channel_info_pending, parse_rgb_channel_info,
+	read_rgb_channel_info_kv, rename_rgb_files, update_rgb_channel_amount_pending,
 };
 use crate::routing::gossip::NodeId;
 use crate::sign::ecdsa::EcdsaChannelSigner;
@@ -104,6 +104,8 @@ use crate::prelude::*;
 use crate::sign::type_resolver::ChannelSignerType;
 #[cfg(any(test, fuzzing, debug_assertions))]
 use crate::sync::Mutex;
+use crate::sync::Arc;
+use crate::util::persist::KVStoreSync;
 use core::ops::Deref;
 use core::time::Duration;
 use core::{cmp, fmt, mem};
@@ -3132,6 +3134,9 @@ where
 	pub(super) consignment_endpoint: Option<RgbTransport>,
 
 	pub(crate) ldk_data_dir: PathBuf,
+
+	/// Optional KVStore for RGB data persistence
+	pub(crate) rgb_kv_store: Option<Arc<dyn KVStoreSync + Send + Sync>>,
 }
 
 /// A channel struct implementing this trait can receive an initial counterparty commitment
@@ -3232,7 +3237,7 @@ where
 		let temporary_channel_id = context.channel_id;
 		context.channel_id = channel_id;
 		if context.is_colored() {
-			rename_rgb_files(&context.channel_id, &temporary_channel_id, &context.ldk_data_dir);
+			rename_rgb_files(&context.channel_id, &temporary_channel_id, &context.ldk_data_dir, context.rgb_kv_store.as_ref());
 		}
 
 		assert!(!context.channel_state.is_monitor_update_in_progress()); // We have not had any monitor(s) yet to fail update!
@@ -3401,6 +3406,7 @@ where
 		msg_push_msat: u64,
 		open_channel_fields: msgs::CommonOpenChannelFields,
 		ldk_data_dir: PathBuf,
+		rgb_kv_store: Option<Arc<dyn KVStoreSync + Send + Sync>>,
 	) -> Result<(FundingScope, ChannelContext<SP>), ChannelError>
 		where
 			ES::Target: EntropySource,
@@ -3723,6 +3729,7 @@ where
 
 			consignment_endpoint: open_channel_fields.consignment_endpoint,
 			ldk_data_dir,
+			rgb_kv_store,
 		};
 
 		Ok((funding, channel_context))
@@ -3748,6 +3755,7 @@ where
 		_logger: L,
 		consignment_endpoint: Option<RgbTransport>,
 		ldk_data_dir: PathBuf,
+		rgb_kv_store: Option<Arc<dyn KVStoreSync + Send + Sync>>,
 	) -> Result<(FundingScope, ChannelContext<SP>), APIError>
 		where
 			ES::Target: EntropySource,
@@ -3966,6 +3974,7 @@ where
 
 			consignment_endpoint,
 			ldk_data_dir,
+			rgb_kv_store,
 		};
 
 		Ok((funding, channel_context))
@@ -4344,11 +4353,15 @@ where
 
 	/// Get the channel local RGB amount
 	pub fn get_local_rgb_amount(&self) -> u64 {
-		let info_file_path = get_rgb_channel_info_path(
-			&self.channel_id.0.as_hex().to_string(),
-			&self.ldk_data_dir,
-			false,
-		);
+		let channel_id_str = self.channel_id.0.as_hex().to_string();
+		// Try KVStore first (using injected store)
+		if let Some(ref kv_store) = self.rgb_kv_store {
+			if let Ok(rgb_info) = read_rgb_channel_info_kv(kv_store.as_ref(), &channel_id_str, false) {
+				return rgb_info.local_rgb_amount;
+			}
+		}
+		// Fallback to filesystem
+		let info_file_path = get_rgb_channel_info_path(&channel_id_str, &self.ldk_data_dir, false);
 		if info_file_path.exists() {
 			let rgb_info = parse_rgb_channel_info(&info_file_path);
 			rgb_info.local_rgb_amount
@@ -4359,11 +4372,15 @@ where
 
 	/// Get the channel remote RGB amount
 	pub fn get_remote_rgb_amount(&self) -> u64 {
-		let info_file_path = get_rgb_channel_info_path(
-			&self.channel_id.0.as_hex().to_string(),
-			&self.ldk_data_dir,
-			false,
-		);
+		let channel_id_str = self.channel_id.0.as_hex().to_string();
+		// Try KVStore first (using injected store)
+		if let Some(ref kv_store) = self.rgb_kv_store {
+			if let Ok(rgb_info) = read_rgb_channel_info_kv(kv_store.as_ref(), &channel_id_str, false) {
+				return rgb_info.remote_rgb_amount;
+			}
+		}
+		// Fallback to filesystem
+		let info_file_path = get_rgb_channel_info_path(&channel_id_str, &self.ldk_data_dir, false);
 		if info_file_path.exists() {
 			let rgb_info = parse_rgb_channel_info(&info_file_path);
 			rgb_info.remote_rgb_amount
@@ -5080,7 +5097,7 @@ where
 				&holder_keys.revocation_key,
 			);
 			if self.is_colored() {
-				color_htlc(&mut htlc_tx, htlc, &self.ldk_data_dir)
+				color_htlc(&mut htlc_tx, htlc, &self.ldk_data_dir, self.rgb_kv_store.as_ref())
 					.expect("successful htlc coloring");
 			}
 
@@ -7324,6 +7341,7 @@ where
 				&self.context.channel_id,
 				&mut closing_transaction,
 				&self.context.ldk_data_dir,
+				self.context.rgb_kv_store.as_ref(),
 			)
 			.expect("successful closing TX coloring");
 		}
@@ -8915,6 +8933,7 @@ where
 				rgb_offered_htlc,
 				rgb_received_htlc,
 				&self.context.ldk_data_dir,
+				self.context.rgb_kv_store.as_ref(),
 			);
 		}
 
@@ -11729,7 +11748,7 @@ where
 		let were_node_one = node_id.as_slice() < counterparty_node_id.as_slice();
 
 		let contract_id = if self.context.is_colored() {
-			let (rgb_info, _) = get_rgb_channel_info_pending(&self.context.channel_id, &self.context.ldk_data_dir);
+			let (rgb_info, _) = get_rgb_channel_info_pending(&self.context.channel_id, &self.context.ldk_data_dir, self.context.rgb_kv_store.as_ref());
 			Some(rgb_info.contract_id)
 		} else {
 			None
@@ -12883,7 +12902,7 @@ where
 			}
 		}
 		if self.context.is_colored() && rgb_received_htlc > 0 {
-			update_rgb_channel_amount_pending(&self.context.channel_id, 0, rgb_received_htlc, &self.context.ldk_data_dir);
+			update_rgb_channel_amount_pending(&self.context.channel_id, 0, rgb_received_htlc, &self.context.ldk_data_dir, self.context.rgb_kv_store.as_ref());
 		}
 		if let Some((feerate, update_state)) = self.context.pending_update_fee {
 			if update_state == FeeUpdateState::AwaitingRemoteRevokeToAnnounce {
@@ -13550,6 +13569,7 @@ where
 		fee_estimator: &LowerBoundedFeeEstimator<F>, entropy_source: &ES, signer_provider: &SP, counterparty_node_id: PublicKey, their_features: &InitFeatures,
 		channel_value_satoshis: u64, push_msat: u64, user_id: u128, config: &UserConfig, current_chain_height: u32,
 		outbound_scid_alias: u64, temporary_channel_id: Option<ChannelId>, logger: L, consignment_endpoint: Option<RgbTransport>, ldk_data_dir: PathBuf,
+		rgb_kv_store: Option<Arc<dyn KVStoreSync + Send + Sync>>,
 	) -> Result<OutboundV1Channel<SP>, APIError>
 	where ES::Target: EntropySource,
 	      F::Target: FeeEstimator,
@@ -13589,6 +13609,7 @@ where
 			logger,
 			consignment_endpoint,
 			ldk_data_dir,
+			rgb_kv_store,
 		)?;
 		let unfunded_context = UnfundedChannelContext {
 			unfunded_channel_age_ticks: 0,
@@ -13672,7 +13693,7 @@ where
 		let temporary_channel_id = self.context.channel_id;
 		self.context.channel_id = ChannelId::v1_from_funding_outpoint(funding_txo);
 		if self.context.is_colored() {
-			rename_rgb_files(&self.context.channel_id, &temporary_channel_id, &self.context.ldk_data_dir);
+			rename_rgb_files(&self.context.channel_id, &temporary_channel_id, &self.context.ldk_data_dir, self.context.rgb_kv_store.as_ref());
 		}
 
 		// If the funding transaction is a coinbase transaction, we need to set the minimum depth to 100.
@@ -13925,7 +13946,8 @@ where
 		fee_estimator: &LowerBoundedFeeEstimator<F>, entropy_source: &ES, signer_provider: &SP,
 		counterparty_node_id: PublicKey, our_supported_features: &ChannelTypeFeatures,
 		their_features: &InitFeatures, msg: &msgs::OpenChannel, user_id: u128, config: &UserConfig,
-		current_chain_height: u32, logger: &L, is_0conf: bool, ldk_data_dir: PathBuf
+		current_chain_height: u32, logger: &L, is_0conf: bool, ldk_data_dir: PathBuf,
+		rgb_kv_store: Option<Arc<dyn KVStoreSync + Send + Sync>>,
 	) -> Result<InboundV1Channel<SP>, ChannelError>
 		where ES::Target: EntropySource,
 			  F::Target: FeeEstimator,
@@ -13966,6 +13988,7 @@ where
 			msg.push_msat,
 			msg.common_fields.clone(),
 			ldk_data_dir,
+			rgb_kv_store,
 		)?;
 		let unfunded_context = UnfundedChannelContext {
 			unfunded_channel_age_ticks: 0,
@@ -14166,7 +14189,7 @@ where
 		counterparty_node_id: PublicKey, their_features: &InitFeatures, funding_satoshis: u64,
 		funding_inputs: Vec<FundingTxInput>, user_id: u128, config: &UserConfig,
 		current_chain_height: u32, outbound_scid_alias: u64, funding_confirmation_target: ConfirmationTarget,
-		logger: L, ldk_data_dir: PathBuf,
+		logger: L, ldk_data_dir: PathBuf, rgb_kv_store: Option<Arc<dyn KVStoreSync + Send + Sync>>,
 	) -> Result<Self, APIError>
 	where ES::Target: EntropySource,
 	      F::Target: FeeEstimator,
@@ -14209,6 +14232,7 @@ where
 			// ok to pass consignment_endpoint as None since this method is unused
 			None,
 			ldk_data_dir,
+			rgb_kv_store,
 		)?;
 		let unfunded_context = UnfundedChannelContext {
 			unfunded_channel_age_ticks: 0,
@@ -14319,7 +14343,7 @@ where
 		holder_node_id: PublicKey, counterparty_node_id: PublicKey, our_supported_features: &ChannelTypeFeatures,
 		their_features: &InitFeatures, msg: &msgs::OpenChannelV2,
 		user_id: u128, config: &UserConfig, current_chain_height: u32, logger: &L,
-		ldk_data_dir: PathBuf,
+		ldk_data_dir: PathBuf, rgb_kv_store: Option<Arc<dyn KVStoreSync + Send + Sync>>,
 	) -> Result<Self, ChannelError>
 		where ES::Target: EntropySource,
 			  F::Target: FeeEstimator,
@@ -14365,6 +14389,7 @@ where
 			0 /* push_msat not used in dual-funding */,
 			msg.common_fields.clone(),
 			ldk_data_dir,
+			rgb_kv_store,
 		)?;
 		let channel_id = ChannelId::v2_from_revocation_basepoints(
 			&funding.get_holder_pubkeys().revocation_basepoint,
@@ -15054,16 +15079,16 @@ where
 	}
 }
 
-impl<'a, 'b, 'c, ES: Deref, SP: Deref> ReadableArgs<(&'a ES, &'b SP, &'c ChannelTypeFeatures, PathBuf)>
+impl<'a, 'b, 'c, ES: Deref, SP: Deref> ReadableArgs<(&'a ES, &'b SP, &'c ChannelTypeFeatures, PathBuf, Option<Arc<dyn KVStoreSync + Send + Sync>>)>
 	for FundedChannel<SP>
 where
 	ES::Target: EntropySource,
 	SP::Target: SignerProvider,
 {
 	fn read<R: io::Read>(
-		reader: &mut R, args: (&'a ES, &'b SP, &'c ChannelTypeFeatures, PathBuf),
+		reader: &mut R, args: (&'a ES, &'b SP, &'c ChannelTypeFeatures, PathBuf, Option<Arc<dyn KVStoreSync + Send + Sync>>),
 	) -> Result<Self, DecodeError> {
-		let (entropy_source, signer_provider, our_supported_features, ldk_data_dir) = args;
+		let (entropy_source, signer_provider, our_supported_features, ldk_data_dir, rgb_kv_store) = args;
 		let ver = read_ver_prefix!(reader, SERIALIZATION_VERSION);
 		if ver <= 2 {
 			return Err(DecodeError::UnknownVersion);
@@ -15859,6 +15884,7 @@ where
 
 				consignment_endpoint,
 				ldk_data_dir,
+				rgb_kv_store,
 			},
 			holder_commitment_point,
 			pending_splice,
@@ -16770,7 +16796,9 @@ mod tests {
 			// These aren't set in the test vectors:
 			[0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff],
 			[0; 32],
+			std::path::PathBuf::new(),
 			[0; 32],
+			None,
 		);
 
 		let holder_pubkeys = signer.pubkeys(&secp_ctx);