diff --git a/lightning-background-processor/src/lib.rs b/lightning-background-processor/src/lib.rs index 31e519b9f..38a4640da 100644 --- a/lightning-background-processor/src/lib.rs +++ b/lightning-background-processor/src/lib.rs @@ -362,21 +362,51 @@ type DynMessageRouter = lightning::onion_message::messenger::DefaultMessageRoute &'static (dyn EntropySource + Send + Sync), >; +/// Placeholder [`lightning::util::persist::KVStoreSync`] used only to name the `Dyn*` alias types +/// below, which are never instantiated. +#[cfg(not(c_bindings))] +#[doc(hidden)] +pub struct UnusedKvStore; + +#[cfg(not(c_bindings))] +impl lightning::util::persist::KVStoreSync for UnusedKvStore { + fn read( + &self, _primary_namespace: &str, _secondary_namespace: &str, _key: &str, + ) -> Result, lightning::io::Error> { + unreachable!() + } + fn write( + &self, _primary_namespace: &str, _secondary_namespace: &str, _key: &str, _buf: Vec, + ) -> Result<(), lightning::io::Error> { + unreachable!() + } + fn remove( + &self, _primary_namespace: &str, _secondary_namespace: &str, _key: &str, _lazy: bool, + ) -> Result<(), lightning::io::Error> { + unreachable!() + } + fn list( + &self, _primary_namespace: &str, _secondary_namespace: &str, + ) -> Result, lightning::io::Error> { + unreachable!() + } +} + #[cfg(all(not(c_bindings), not(taproot)))] -type DynSignerProvider = dyn lightning::sign::SignerProvider +type DynSignerProvider = dyn lightning::sign::SignerProvider> + Send + Sync; #[cfg(all(not(c_bindings), taproot))] type DynSignerProvider = (dyn lightning::sign::SignerProvider< - EcdsaSigner = lightning::sign::InMemorySigner, - TaprootSigner = lightning::sign::InMemorySigner, + EcdsaSigner = lightning::sign::InMemorySigner, + TaprootSigner = lightning::sign::InMemorySigner, > + Send + Sync); #[cfg(not(c_bindings))] type DynChannelManager = lightning::ln::channelmanager::ChannelManager< - &'static (dyn chain::Watch + Send + Sync), + &'static (dyn chain::Watch> + Send + Sync), &'static (dyn BroadcasterInterface + Send + Sync), &'static (dyn EntropySource + Send + Sync), &'static (dyn lightning::sign::NodeSigner + Send + Sync), @@ -385,6 +415,7 @@ type DynChannelManager = lightning::ln::channelmanager::ChannelManager< &'static DynRouter, &'static DynMessageRouter, &'static (dyn Logger + Send + Sync), + UnusedKvStore, >; /// When initializing a background processor without an onion messenger, this can be used to avoid diff --git a/lightning/Cargo.toml b/lightning/Cargo.toml index e281c5b58..8a6b8263c 100644 --- a/lightning/Cargo.toml +++ b/lightning/Cargo.toml @@ -52,6 +52,7 @@ libm = { version = "0.2", default-features = false } inventory = { version = "0.3", optional = true } # RGB and related +bincode = "1.3" futures = "0.3" rgb-lib = { version = "=0.3.0-beta.7", features = [ "electrum", @@ -60,7 +61,6 @@ rgb-lib = { version = "=0.3.0-beta.7", features = [ serde = { version = "^1.0", features = [ "derive", ] } -serde_json = "^1.0" tokio = { version = "1.14.1", features = [ "macros", "rt-multi-thread", diff --git a/lightning/src/ln/chan_utils.rs b/lightning/src/ln/chan_utils.rs index 3e337e1aa..33ad5b30c 100644 --- a/lightning/src/ln/chan_utils.rs +++ b/lightning/src/ln/chan_utils.rs @@ -36,6 +36,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; @@ -2153,9 +2154,10 @@ impl<'a> TrustedCommitmentTransaction<'a> { /// /// This function is only valid in the holder commitment context, it always uses EcdsaSighashType::All. #[rustfmt::skip] - pub fn get_htlc_sigs( + pub fn get_htlc_sigs( &self, htlc_base_key: &SecretKey, channel_parameters: &DirectedChannelTransactionParameters, entropy_source: &ES, secp_ctx: &Secp256k1, ldk_data_dir: &PathBuf, + rgb_kv_store: &K, ) -> Result, ()> where ES::Target: EntropySource { let inner = self.inner; let keys = &inner.keys; @@ -2167,7 +2169,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(()); } } diff --git a/lightning/src/ln/channel.rs b/lightning/src/ln/channel.rs index 79ab65373..c922244bd 100644 --- a/lightning/src/ln/channel.rs +++ b/lightning/src/ln/channel.rs @@ -78,9 +78,9 @@ use crate::ln::types::ChannelId; 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, is_asset_known, parse_rgb_channel_info, rename_rgb_files, - set_counterparty_knows_asset, update_rgb_channel_amount_pending, + color_closing, color_commitment, color_htlc, get_rgb_channel_info_pending, is_asset_known, + set_counterparty_knows_asset, update_rgb_channel_amount_pending, update_rgb_channel_id, + RgbKvStoreExt, }; use crate::routing::gossip::NodeId; use crate::sign::ecdsa::EcdsaChannelSigner; @@ -102,8 +102,10 @@ use alloc::collections::{btree_map, BTreeMap}; use crate::io; use crate::prelude::*; use crate::sign::type_resolver::ChannelSignerType; +use crate::sync::Arc; #[cfg(any(test, fuzzing, debug_assertions))] use crate::sync::Mutex; +use crate::util::persist::KVStoreSync; use core::ops::Deref; use core::time::Duration; use core::{cmp, fmt, mem}; @@ -998,8 +1000,8 @@ impl<'a, 'b, L: Deref> WithChannelContext<'a, L> where L::Target: Logger, { - pub(super) fn from( - logger: &'a L, context: &'b ChannelContext, payment_hash: Option, + pub(super) fn from( + logger: &'a L, context: &'b ChannelContext, payment_hash: Option, ) -> Self where S::Target: SignerProvider, @@ -1420,32 +1422,32 @@ impl_writeable_tlv_based!(PendingChannelMonitorUpdate, { /// A payment channel with a counterparty throughout its life-cycle, encapsulating negotiation and /// funding phases. -pub(super) struct Channel +pub(super) struct Channel where SP::Target: SignerProvider, { - phase: ChannelPhase, + phase: ChannelPhase, } /// The `ChannelPhase` enum describes the current phase in life of a lightning channel with each of /// its variants containing an appropriate channel struct. -enum ChannelPhase +enum ChannelPhase where SP::Target: SignerProvider, { Undefined, - UnfundedOutboundV1(OutboundV1Channel), - UnfundedInboundV1(InboundV1Channel), - UnfundedV2(PendingV2Channel), - Funded(FundedChannel), + UnfundedOutboundV1(OutboundV1Channel), + UnfundedInboundV1(InboundV1Channel), + UnfundedV2(PendingV2Channel), + Funded(FundedChannel), } -impl Channel +impl Channel where SP::Target: SignerProvider, ::EcdsaSigner: ChannelSigner, { - pub fn context(&self) -> &ChannelContext { + pub fn context(&self) -> &ChannelContext { match &self.phase { ChannelPhase::Undefined => unreachable!(), ChannelPhase::Funded(chan) => &chan.context, @@ -1455,7 +1457,7 @@ where } } - pub fn context_mut(&mut self) -> &mut ChannelContext { + pub fn context_mut(&mut self) -> &mut ChannelContext { match &mut self.phase { ChannelPhase::Undefined => unreachable!(), ChannelPhase::Funded(chan) => &mut chan.context, @@ -1486,7 +1488,7 @@ where } } - pub fn funding_and_context_mut(&mut self) -> (&FundingScope, &mut ChannelContext) { + pub fn funding_and_context_mut(&mut self) -> (&FundingScope, &mut ChannelContext) { match &mut self.phase { ChannelPhase::Undefined => unreachable!(), ChannelPhase::Funded(chan) => (&chan.funding, &mut chan.context), @@ -1513,7 +1515,7 @@ where matches!(self.phase, ChannelPhase::Funded(_)) } - pub fn as_funded(&self) -> Option<&FundedChannel> { + pub fn as_funded(&self) -> Option<&FundedChannel> { if let ChannelPhase::Funded(channel) = &self.phase { Some(channel) } else { @@ -1521,7 +1523,7 @@ where } } - pub fn as_funded_mut(&mut self) -> Option<&mut FundedChannel> { + pub fn as_funded_mut(&mut self) -> Option<&mut FundedChannel> { if let ChannelPhase::Funded(channel) = &mut self.phase { Some(channel) } else { @@ -1529,7 +1531,7 @@ where } } - pub fn as_unfunded_outbound_v1_mut(&mut self) -> Option<&mut OutboundV1Channel> { + pub fn as_unfunded_outbound_v1_mut(&mut self) -> Option<&mut OutboundV1Channel> { if let ChannelPhase::UnfundedOutboundV1(channel) = &mut self.phase { Some(channel) } else { @@ -1561,7 +1563,7 @@ where } } - pub fn into_unfunded_outbound_v1(self) -> Result, Self> { + pub fn into_unfunded_outbound_v1(self) -> Result, Self> { if let ChannelPhase::UnfundedOutboundV1(channel) = self.phase { Ok(channel) } else { @@ -1569,7 +1571,7 @@ where } } - pub fn into_unfunded_inbound_v1(self) -> Result, Self> { + pub fn into_unfunded_inbound_v1(self) -> Result, Self> { if let ChannelPhase::UnfundedInboundV1(channel) = self.phase { Ok(channel) } else { @@ -1577,7 +1579,7 @@ where } } - pub fn as_unfunded_v2(&self) -> Option<&PendingV2Channel> { + pub fn as_unfunded_v2(&self) -> Option<&PendingV2Channel> { if let ChannelPhase::UnfundedV2(channel) = &self.phase { Some(channel) } else { @@ -1989,7 +1991,7 @@ where #[rustfmt::skip] pub fn funding_signed( &mut self, msg: &msgs::FundingSigned, best_block: BestBlock, signer_provider: &SP, logger: &L - ) -> Result<(&mut FundedChannel, ChannelMonitor<::EcdsaSigner>), ChannelError> + ) -> Result<(&mut FundedChannel, ChannelMonitor<::EcdsaSigner>), ChannelError> where L::Target: Logger { @@ -2244,42 +2246,46 @@ where } } -impl From> for Channel +impl From> + for Channel where SP::Target: SignerProvider, ::EcdsaSigner: ChannelSigner, { - fn from(channel: OutboundV1Channel) -> Self { + fn from(channel: OutboundV1Channel) -> Self { Channel { phase: ChannelPhase::UnfundedOutboundV1(channel) } } } -impl From> for Channel +impl From> + for Channel where SP::Target: SignerProvider, ::EcdsaSigner: ChannelSigner, { - fn from(channel: InboundV1Channel) -> Self { + fn from(channel: InboundV1Channel) -> Self { Channel { phase: ChannelPhase::UnfundedInboundV1(channel) } } } -impl From> for Channel +impl From> + for Channel where SP::Target: SignerProvider, ::EcdsaSigner: ChannelSigner, { - fn from(channel: PendingV2Channel) -> Self { + fn from(channel: PendingV2Channel) -> Self { Channel { phase: ChannelPhase::UnfundedV2(channel) } } } -impl From> for Channel +impl From> + for Channel where SP::Target: SignerProvider, ::EcdsaSigner: ChannelSigner, { - fn from(channel: FundedChannel) -> Self { + fn from(channel: FundedChannel) -> Self { Channel { phase: ChannelPhase::Funded(channel) } } } @@ -2551,10 +2557,10 @@ impl FundingScope { } /// Constructs a `FundingScope` for splicing a channel. - fn for_splice( - prev_funding: &Self, context: &ChannelContext, our_funding_contribution: SignedAmount, - their_funding_contribution: SignedAmount, counterparty_funding_pubkey: PublicKey, - our_new_holder_keys: ChannelPublicKeys, + fn for_splice( + prev_funding: &Self, context: &ChannelContext, + our_funding_contribution: SignedAmount, their_funding_contribution: SignedAmount, + counterparty_funding_pubkey: PublicKey, our_new_holder_keys: ChannelPublicKeys, ) -> Self where SP::Target: SignerProvider, @@ -2760,8 +2766,8 @@ impl FundingNegotiation { } impl PendingFunding { - fn check_get_splice_locked( - &mut self, context: &ChannelContext, confirmed_funding_index: usize, height: u32, + fn check_get_splice_locked( + &mut self, context: &ChannelContext, confirmed_funding_index: usize, height: u32, ) -> Option where SP::Target: SignerProvider, @@ -2868,7 +2874,7 @@ impl<'a> From<&'a Transaction> for ConfirmedTransaction<'a> { } /// Contains everything about the channel including state, and various flags. -pub(crate) struct ChannelContext +pub(crate) struct ChannelContext where SP::Target: SignerProvider, { @@ -3155,17 +3161,20 @@ where pub(super) is_colored: bool, pub(crate) ldk_data_dir: PathBuf, + + /// KVStore for RGB data persistence + pub(crate) rgb_kv_store: Arc, } /// A channel struct implementing this trait can receive an initial counterparty commitment /// transaction signature. -trait InitialRemoteCommitmentReceiver +trait InitialRemoteCommitmentReceiver where SP::Target: SignerProvider, { - fn context(&self) -> &ChannelContext; + fn context(&self) -> &ChannelContext; - fn context_mut(&mut self) -> &mut ChannelContext; + fn context_mut(&mut self) -> &mut ChannelContext; fn funding(&self) -> &FundingScope; @@ -3255,7 +3264,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); + update_rgb_channel_id(&context.channel_id, &temporary_channel_id, 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! @@ -3301,15 +3310,16 @@ where fn is_v2_established(&self) -> bool; } -impl InitialRemoteCommitmentReceiver for OutboundV1Channel +impl InitialRemoteCommitmentReceiver + for OutboundV1Channel where SP::Target: SignerProvider, { - fn context(&self) -> &ChannelContext { + fn context(&self) -> &ChannelContext { &self.context } - fn context_mut(&mut self) -> &mut ChannelContext { + fn context_mut(&mut self) -> &mut ChannelContext { &mut self.context } @@ -3330,15 +3340,16 @@ where } } -impl InitialRemoteCommitmentReceiver for InboundV1Channel +impl InitialRemoteCommitmentReceiver + for InboundV1Channel where SP::Target: SignerProvider, { - fn context(&self) -> &ChannelContext { + fn context(&self) -> &ChannelContext { &self.context } - fn context_mut(&mut self) -> &mut ChannelContext { + fn context_mut(&mut self) -> &mut ChannelContext { &mut self.context } @@ -3359,15 +3370,16 @@ where } } -impl InitialRemoteCommitmentReceiver for FundedChannel +impl InitialRemoteCommitmentReceiver + for FundedChannel where SP::Target: SignerProvider, { - fn context(&self) -> &ChannelContext { + fn context(&self) -> &ChannelContext { &self.context } - fn context_mut(&mut self) -> &mut ChannelContext { + fn context_mut(&mut self) -> &mut ChannelContext { &mut self.context } @@ -3400,7 +3412,7 @@ where } } -impl ChannelContext +impl ChannelContext where SP::Target: SignerProvider, { @@ -3425,7 +3437,8 @@ where open_channel_fields: msgs::CommonOpenChannelFields, rgb_asset: Option<(ContractId, Option)>, ldk_data_dir: PathBuf, - ) -> Result<(FundingScope, ChannelContext), ChannelError> + rgb_kv_store: Arc, + ) -> Result<(FundingScope, ChannelContext), ChannelError> where ES::Target: EntropySource, F::Target: FeeEstimator, @@ -3467,7 +3480,7 @@ where if open_channel_fields.htlc_minimum_msat >= full_channel_value_msat { return Err(ChannelError::close(format!("Minimum htlc value ({}) was larger than full channel value ({})", open_channel_fields.htlc_minimum_msat, full_channel_value_msat))); } - FundedChannel::::check_remote_fee(&channel_type, fee_estimator, open_channel_fields.commitment_feerate_sat_per_1000_weight, None, &&logger)?; + FundedChannel::::check_remote_fee(&channel_type, fee_estimator, open_channel_fields.commitment_feerate_sat_per_1000_weight, None, &&logger)?; let max_counterparty_selected_contest_delay = u16::min(config.channel_handshake_limits.their_to_self_delay, MAX_LOCAL_BREAKDOWN_TIMEOUT); if open_channel_fields.to_self_delay > max_counterparty_selected_contest_delay { @@ -3748,6 +3761,7 @@ where is_colored: funding.is_colored(), ldk_data_dir, + rgb_kv_store, }; Ok((funding, channel_context)) @@ -3773,7 +3787,8 @@ where _logger: L, rgb_asset: Option<(ContractId, Option)>, ldk_data_dir: PathBuf, - ) -> Result<(FundingScope, ChannelContext), APIError> + rgb_kv_store: Arc, + ) -> Result<(FundingScope, ChannelContext), APIError> where ES::Target: EntropySource, F::Target: FeeEstimator, @@ -3992,6 +4007,7 @@ where is_colored: funding.is_colored(), ldk_data_dir, + rgb_kv_store, }; Ok((funding, channel_context)) @@ -4370,13 +4386,8 @@ 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, - ); - if info_file_path.exists() { - let rgb_info = parse_rgb_channel_info(&info_file_path); + let channel_id_str = self.channel_id.0.as_hex().to_string(); + if let Ok(rgb_info) = self.rgb_kv_store.read_rgb_channel_info(&channel_id_str, false) { rgb_info.local_rgb_amount } else { 0 @@ -4385,13 +4396,8 @@ 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, - ); - if info_file_path.exists() { - let rgb_info = parse_rgb_channel_info(&info_file_path); + let channel_id_str = self.channel_id.0.as_hex().to_string(); + if let Ok(rgb_info) = self.rgb_kv_store.read_rgb_channel_info(&channel_id_str, false) { rgb_info.remote_rgb_amount } else { 0 @@ -5106,7 +5112,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"); } @@ -6745,8 +6751,12 @@ pub(super) struct FundingNegotiationContext { impl FundingNegotiationContext { /// Prepare and start interactive transaction negotiation. /// If error occurs, it is caused by our side, not the counterparty. - fn into_interactive_tx_constructor( - mut self, context: &ChannelContext, funding: &FundingScope, signer_provider: &SP, + fn into_interactive_tx_constructor< + SP: Deref, + ES: Deref, + KV: KVStoreSync + Send + Sync + 'static, + >( + mut self, context: &ChannelContext, funding: &FundingScope, signer_provider: &SP, entropy_source: &ES, holder_node_id: PublicKey, ) -> Result where @@ -6855,12 +6865,12 @@ impl FundingNegotiationContext { // Holder designates channel data owned for the benefit of the user client. // Counterparty designates channel data owned by the another channel participant entity. -pub(super) struct FundedChannel +pub(super) struct FundedChannel where SP::Target: SignerProvider, { pub funding: FundingScope, - pub context: ChannelContext, + pub context: ChannelContext, holder_commitment_point: HolderCommitmentPoint, /// Information about any pending splice candidates, including RBF attempts. @@ -7038,12 +7048,12 @@ pub struct SpliceFundingPromotion { pub discarded_funding: Vec, } -impl FundedChannel +impl FundedChannel where SP::Target: SignerProvider, ::EcdsaSigner: EcdsaChannelSigner, { - pub fn context(&self) -> &ChannelContext { + pub fn context(&self) -> &ChannelContext { &self.context } @@ -7350,6 +7360,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"); } @@ -8949,7 +8960,7 @@ where &self.context.channel_id, rgb_offered_htlc, rgb_received_htlc, - &self.context.ldk_data_dir, + self.context.rgb_kv_store.as_ref(), ); } @@ -9657,7 +9668,7 @@ where core::iter::once(&self.funding) .chain(self.pending_funding().iter()) - .try_for_each(|funding| FundedChannel::::check_remote_fee(funding.get_channel_type(), fee_estimator, msg.feerate_per_kw, Some(self.context.feerate_per_kw), logger))?; + .try_for_each(|funding| FundedChannel::::check_remote_fee(funding.get_channel_type(), fee_estimator, msg.feerate_per_kw, Some(self.context.feerate_per_kw), logger))?; self.context.pending_update_fee = Some((msg.feerate_per_kw, FeeUpdateState::RemoteAnnounced)); self.context.update_time_counter += 1; @@ -11763,7 +11774,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.rgb_kv_store.as_ref()); Some(rgb_info.contract_id) } else { None @@ -12917,7 +12928,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.rgb_kv_store.as_ref()); } if let Some((feerate, update_state)) = self.context.pending_update_fee { if update_state == FeeUpdateState::AwaitingRemoteRevokeToAnnounce { @@ -13557,12 +13568,12 @@ where } /// A not-yet-funded outbound (from holder) channel using V1 channel establishment. -pub(super) struct OutboundV1Channel +pub(super) struct OutboundV1Channel where SP::Target: SignerProvider, { pub funding: FundingScope, - pub context: ChannelContext, + pub context: ChannelContext, pub unfunded_context: UnfundedChannelContext, /// We tried to send an `open_channel` message but our commitment point wasn't ready. /// This flag tells us we need to send it when we are retried once the @@ -13570,7 +13581,7 @@ where pub signer_pending_open_channel: bool, } -impl OutboundV1Channel +impl OutboundV1Channel where SP::Target: SignerProvider, { @@ -13584,7 +13595,8 @@ where fee_estimator: &LowerBoundedFeeEstimator, 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, logger: L, rgb_asset: Option<(ContractId, Option)>, ldk_data_dir: PathBuf, - ) -> Result, APIError> + rgb_kv_store: Arc, + ) -> Result, APIError> where ES::Target: EntropySource, F::Target: FeeEstimator, L::Target: Logger, @@ -13623,6 +13635,7 @@ where logger, rgb_asset, ldk_data_dir, + rgb_kv_store, )?; let unfunded_context = UnfundedChannelContext { unfunded_channel_age_ticks: 0, @@ -13706,7 +13719,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); + update_rgb_channel_id(&self.context.channel_id, &temporary_channel_id, self.context.rgb_kv_store.as_ref()); } // If the funding transaction is a coinbase transaction, we need to set the minimum depth to 100. @@ -13821,7 +13834,10 @@ where )?; if self.funding.is_colored() && msg.known_asset { - set_counterparty_knows_asset(&self.context.channel_id, &self.context.ldk_data_dir); + set_counterparty_knows_asset( + &self.context.channel_id, + self.context.rgb_kv_store.as_ref(), + ); } Ok(()) @@ -13832,7 +13848,7 @@ where #[rustfmt::skip] pub fn funding_signed( mut self, msg: &msgs::FundingSigned, best_block: BestBlock, signer_provider: &SP, logger: &L - ) -> Result<(FundedChannel, ChannelMonitor<::EcdsaSigner>), (OutboundV1Channel, ChannelError)> + ) -> Result<(FundedChannel, ChannelMonitor<::EcdsaSigner>), (OutboundV1Channel, ChannelError)> where L::Target: Logger { @@ -13913,12 +13929,12 @@ where } /// A not-yet-funded inbound (from counterparty) channel using V1 channel establishment. -pub(super) struct InboundV1Channel +pub(super) struct InboundV1Channel where SP::Target: SignerProvider, { pub funding: FundingScope, - pub context: ChannelContext, + pub context: ChannelContext, pub unfunded_context: UnfundedChannelContext, pub signer_pending_accept_channel: bool, } @@ -13954,7 +13970,7 @@ pub(super) fn channel_type_from_open_channel( Ok(channel_type.clone()) } -impl InboundV1Channel +impl InboundV1Channel where SP::Target: SignerProvider, { @@ -13965,8 +13981,9 @@ where fee_estimator: &LowerBoundedFeeEstimator, 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 - ) -> Result, ChannelError> + current_chain_height: u32, logger: &L, is_0conf: bool, ldk_data_dir: PathBuf, + rgb_kv_store: Arc, + ) -> Result, ChannelError> where ES::Target: EntropySource, F::Target: FeeEstimator, L::Target: Logger, @@ -14007,6 +14024,7 @@ where msg.common_fields.clone(), msg.rgb_asset, ldk_data_dir, + rgb_kv_store, )?; let unfunded_context = UnfundedChannelContext { unfunded_channel_age_ticks: 0, @@ -14063,7 +14081,11 @@ where let keys = self.funding.get_holder_pubkeys(); let known_asset = match self.funding.contract_id() { - Some(contract_id) => is_asset_known(contract_id, &self.context.ldk_data_dir), + Some(contract_id) => is_asset_known( + contract_id, + &self.context.ldk_data_dir, + self.context.rgb_kv_store.as_ref(), + ), None => false, }; @@ -14112,7 +14134,7 @@ where #[rustfmt::skip] pub fn funding_created( mut self, msg: &msgs::FundingCreated, best_block: BestBlock, signer_provider: &SP, logger: &L - ) -> Result<(FundedChannel, Option, ChannelMonitor<::EcdsaSigner>), (Self, ChannelError)> + ) -> Result<(FundedChannel, Option, ChannelMonitor<::EcdsaSigner>), (Self, ChannelError)> where L::Target: Logger { @@ -14190,19 +14212,19 @@ where } // A not-yet-funded channel using V2 channel establishment. -pub(super) struct PendingV2Channel +pub(super) struct PendingV2Channel where SP::Target: SignerProvider, { pub funding: FundingScope, - pub context: ChannelContext, + pub context: ChannelContext, pub unfunded_context: UnfundedChannelContext, pub funding_negotiation_context: FundingNegotiationContext, /// The current interactive transaction construction session under negotiation. pub interactive_tx_constructor: Option, } -impl PendingV2Channel +impl PendingV2Channel where SP::Target: SignerProvider, { @@ -14213,7 +14235,7 @@ where counterparty_node_id: PublicKey, their_features: &InitFeatures, funding_satoshis: u64, funding_inputs: Vec, 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: Arc, ) -> Result where ES::Target: EntropySource, F::Target: FeeEstimator, @@ -14256,6 +14278,7 @@ where // ok to pass rgb_asset as None since this method is unused None, ldk_data_dir, + rgb_kv_store, )?; let unfunded_context = UnfundedChannelContext { unfunded_channel_age_ticks: 0, @@ -14365,7 +14388,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: Arc, ) -> Result where ES::Target: EntropySource, F::Target: FeeEstimator, @@ -14412,6 +14435,7 @@ where msg.common_fields.clone(), None, ldk_data_dir, + rgb_kv_store, )?; let channel_id = ChannelId::v2_from_revocation_basepoints( &funding.get_holder_pubkeys().revocation_basepoint, @@ -14629,7 +14653,7 @@ impl Readable for AnnouncementSigsState { } } -impl Writeable for FundedChannel +impl Writeable for FundedChannel where SP::Target: SignerProvider, { @@ -15103,16 +15127,17 @@ where } } -impl<'a, 'b, 'c, ES: Deref, SP: Deref> - ReadableArgs<(&'a ES, &'b SP, &'c ChannelTypeFeatures, PathBuf)> for FundedChannel +impl<'a, 'b, 'c, ES: Deref, SP: Deref, KV: KVStoreSync + Send + Sync + 'static> + ReadableArgs<(&'a ES, &'b SP, &'c ChannelTypeFeatures, PathBuf, Arc)> for FundedChannel where ES::Target: EntropySource, SP::Target: SignerProvider, { fn read( - reader: &mut R, args: (&'a ES, &'b SP, &'c ChannelTypeFeatures, PathBuf), + reader: &mut R, args: (&'a ES, &'b SP, &'c ChannelTypeFeatures, PathBuf, Arc), ) -> Result { - 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); @@ -15908,6 +15933,7 @@ where interactive_tx_signing_session, is_colored: rgb_asset.is_some(), ldk_data_dir, + rgb_kv_store, }, holder_commitment_point, pending_splice, diff --git a/lightning/src/ln/channel_state.rs b/lightning/src/ln/channel_state.rs index c68a46acf..a0e703433 100644 --- a/lightning/src/ln/channel_state.rs +++ b/lightning/src/ln/channel_state.rs @@ -21,6 +21,7 @@ use crate::sign::SignerProvider; use crate::types::features::{ChannelTypeFeatures, InitFeatures}; use crate::types::payment::PaymentHash; use crate::util::config::ChannelConfig; +use crate::util::persist::KVStoreSync; use core::ops::Deref; @@ -530,8 +531,8 @@ impl ChannelDetails { } } - pub(super) fn from_channel( - channel: &Channel, best_block_height: u32, latest_features: InitFeatures, + pub(super) fn from_channel( + channel: &Channel, best_block_height: u32, latest_features: InitFeatures, fee_estimator: &LowerBoundedFeeEstimator, ) -> Self where diff --git a/lightning/src/ln/channelmanager.rs b/lightning/src/ln/channelmanager.rs index 6a9a2cc32..849bc36ed 100644 --- a/lightning/src/ln/channelmanager.rs +++ b/lightning/src/ln/channelmanager.rs @@ -119,10 +119,7 @@ use crate::onion_message::messenger::{ MessageRouter, MessageSendInstructions, Responder, ResponseInstruction, }; use crate::onion_message::offers::{OffersMessage, OffersMessageHandler}; -use crate::rgb_utils::{ - get_rgb_channel_info, get_rgb_payment_info_path, handle_funding, is_channel_rgb, - parse_rgb_payment_info, -}; +use crate::rgb_utils::{handle_funding, is_channel_rgb, RgbKvStoreExt}; use crate::routing::router::{ BlindedTail, FixedRouter, InFlightHtlcs, Path, Payee, PaymentParameters, Route, RouteParameters, RouteParametersConfig, Router, @@ -139,6 +136,7 @@ use crate::types::string::UntrustedString; use crate::util::config::{ChannelConfig, ChannelConfigOverrides, ChannelConfigUpdate, UserConfig}; use crate::util::errors::APIError; use crate::util::logger::{Level, Logger, WithContext}; +use crate::util::persist::KVStoreSync; use crate::util::scid_utils::fake_scid; use crate::util::ser::{ BigSize, FixedLengthReader, LengthReadable, MaybeReadable, Readable, ReadableArgs, VecWriter, @@ -803,7 +801,7 @@ mod fuzzy_channelmanager { use super::*; /// Tracks the inbound corresponding to an outbound HTLC - #[allow(clippy::derive_hash_xor_eq)] // Our Hash is faithful to the data, we just don't have SecretKey::hash + #[allow(clippy::derived_hash_with_manual_eq)] // Our Hash is faithful to the data, we just don't have SecretKey::hash #[derive(Clone, Debug, PartialEq, Eq)] pub enum HTLCSource { PreviousHopData(HTLCPreviousHopData), @@ -847,7 +845,7 @@ pub use self::fuzzy_channelmanager::*; #[cfg(not(fuzzing))] pub(crate) use self::fuzzy_channelmanager::*; -#[allow(clippy::derive_hash_xor_eq)] // Our Hash is faithful to the data, we just don't have SecretKey::hash +#[allow(clippy::derived_hash_with_manual_eq)] // Our Hash is faithful to the data, we just don't have SecretKey::hash impl core::hash::Hash for HTLCSource { fn hash(&self, hasher: &mut H) { match self { @@ -1590,14 +1588,14 @@ impl Readable for Option { } /// State we hold per-peer. -pub(super) struct PeerState +pub(super) struct PeerState where SP::Target: SignerProvider, { /// `channel_id` -> `Channel` /// /// Holds all channels where the peer is the counterparty. - pub(super) channel_by_id: HashMap>, + pub(super) channel_by_id: HashMap>, /// `temporary_channel_id` -> `InboundChannelRequest`. /// /// When manual channel acceptance is enabled, this holds all unaccepted inbound channels where @@ -1668,7 +1666,7 @@ where peer_storage: Vec, } -impl PeerState +impl PeerState where SP::Target: SignerProvider, { @@ -1684,7 +1682,7 @@ where return false; } } - let chan_is_funded_or_outbound = |(_, channel): (_, &Channel)| { + let chan_is_funded_or_outbound = |(_, channel): (_, &Channel)| { channel.is_funded() || channel.funding().is_outbound() }; !self.channel_by_id.iter().any(chan_is_funded_or_outbound) @@ -1766,25 +1764,26 @@ struct PendingInboundPayment { /// /// This is not exported to bindings users as type aliases aren't supported in most languages. #[cfg(not(c_bindings))] -pub type SimpleArcChannelManager = ChannelManager< +pub type SimpleArcChannelManager = ChannelManager< Arc, Arc, - Arc, - Arc, - Arc, + Arc>, + Arc>, + Arc>, Arc, Arc< DefaultRouter< Arc>>, Arc, - Arc, + Arc>, Arc>>, Arc>>>, ProbabilisticScoringFeeParameters, ProbabilisticScorer>>, Arc>, >, >, - Arc>>, Arc, Arc>>, + Arc>>, Arc, Arc>>>, Arc, + KV, >; /// [`SimpleRefChannelManager`] is a type alias for a ChannelManager reference, and is the reference @@ -1799,24 +1798,26 @@ pub type SimpleArcChannelManager = ChannelManager< /// /// This is not exported to bindings users as type aliases aren't supported in most languages. #[cfg(not(c_bindings))] -pub type SimpleRefChannelManager<'a, 'b, 'c, 'd, 'e, 'f, 'g, 'h, 'i, M, T, F, L> = ChannelManager< - &'a M, - &'b T, - &'c KeysManager, - &'c KeysManager, - &'c KeysManager, - &'d F, - &'e DefaultRouter< - &'f NetworkGraph<&'g L>, +pub type SimpleRefChannelManager<'a, 'b, 'c, 'd, 'e, 'f, 'g, 'h, 'i, M, T, F, L, KV> = + ChannelManager< + &'a M, + &'b T, + &'c KeysManager, + &'c KeysManager, + &'c KeysManager, + &'d F, + &'e DefaultRouter< + &'f NetworkGraph<&'g L>, + &'g L, + &'c KeysManager, + &'h RwLock, &'g L>>, + ProbabilisticScoringFeeParameters, + ProbabilisticScorer<&'f NetworkGraph<&'g L>, &'g L>, + >, + &'i DefaultMessageRouter<&'f NetworkGraph<&'g L>, &'g L, &'c KeysManager>, &'g L, - &'c KeysManager, - &'h RwLock, &'g L>>, - ProbabilisticScoringFeeParameters, - ProbabilisticScorer<&'f NetworkGraph<&'g L>, &'g L>, - >, - &'i DefaultMessageRouter<&'f NetworkGraph<&'g L>, &'g L, &'c KeysManager>, - &'g L, ->; + KV, + >; /// A trivial trait which describes any [`ChannelManager`]. /// @@ -1861,6 +1862,8 @@ pub trait AChannelManager { type Logger: Logger + ?Sized; /// A type that may be dereferenced to [`Self::Logger`]. type L: Deref; + /// A type implementing [`KVStoreSync`]. + type KVStore: KVStoreSync + Send + Sync + 'static; /// Returns a reference to the actual [`ChannelManager`] object. fn get_cm( &self, @@ -1874,6 +1877,7 @@ pub trait AChannelManager { Self::R, Self::MR, Self::L, + Self::KVStore, >; } @@ -1887,7 +1891,8 @@ impl< R: Deref, MR: Deref, L: Deref, - > AChannelManager for ChannelManager + KV: KVStoreSync + Send + Sync + 'static, + > AChannelManager for ChannelManager where M::Target: chain::Watch<::EcdsaSigner>, T::Target: BroadcasterInterface, @@ -1918,7 +1923,8 @@ where type MR = MR; type Logger = L::Target; type L = L; - fn get_cm(&self) -> &ChannelManager { + type KVStore = KV; + fn get_cm(&self) -> &ChannelManager { self } } @@ -2712,6 +2718,7 @@ pub struct ChannelManager< R: Deref, MR: Deref, L: Deref, + KV: KVStoreSync + Send + Sync + 'static, > where M::Target: chain::Watch<::EcdsaSigner>, T::Target: BroadcasterInterface, @@ -2755,7 +2762,7 @@ pub struct ChannelManager< /// See `PendingOutboundPayment` documentation for more info. /// /// See `ChannelManager` struct-level documentation for lock order requirements. - pending_outbound_payments: OutboundPayments, + pending_outbound_payments: OutboundPayments, /// SCID/SCID Alias -> forward infos. Key of 0 means payments received. /// @@ -2860,9 +2867,9 @@ pub struct ChannelManager< /// /// See `ChannelManager` struct-level documentation for lock order requirements. #[cfg(not(any(test, feature = "_test_utils")))] - per_peer_state: FairRwLock>>>, + per_peer_state: FairRwLock>>>, #[cfg(any(test, feature = "_test_utils"))] - pub(super) per_peer_state: FairRwLock>>>, + pub(super) per_peer_state: FairRwLock>>>, /// We only support using one of [`ChannelMonitorUpdateStatus::InProgress`] and /// [`ChannelMonitorUpdateStatus::Completed`] without restarting. Because the API does not @@ -2963,6 +2970,9 @@ pub struct ChannelManager< logger: L, ldk_data_dir: PathBuf, + + /// KVStore for RGB data persistence + rgb_kv_store: Arc, } /// Chain-related parameters used to construct a new `ChannelManager`. @@ -3420,7 +3430,7 @@ macro_rules! convert_channel_err { $self.get_channel_update_for_broadcast(&$funded_channel).ok(), ) }; - let mut locked_close = |shutdown_res_mut: &mut ShutdownResult, funded_channel: &mut FundedChannel<_>| { + let mut locked_close = |shutdown_res_mut: &mut ShutdownResult, funded_channel: &mut FundedChannel<_, _>| { locked_close_channel!($self, $peer_state, funded_channel, shutdown_res_mut, FUNDED); }; let (close, mut err) = @@ -3437,7 +3447,7 @@ macro_rules! convert_channel_err { $self.get_channel_update_for_broadcast(&$funded_channel).ok(), ) }; - let mut locked_close = |shutdown_res_mut: &mut ShutdownResult, funded_channel: &mut FundedChannel<_>| { + let mut locked_close = |shutdown_res_mut: &mut ShutdownResult, funded_channel: &mut FundedChannel<_, _>| { locked_close_channel!($self, $peer_state, funded_channel, shutdown_res_mut, FUNDED); }; convert_channel_err!($self, $peer_state, $err, $funded_channel, do_close, locked_close, chan_id, _internal) @@ -3445,7 +3455,7 @@ macro_rules! convert_channel_err { ($self: ident, $peer_state: expr, $err: expr, $channel: expr, UNFUNDED_CHANNEL) => { { let chan_id = $channel.context().channel_id(); let mut do_close = |reason| { ($channel.force_shutdown(reason), None) }; - let locked_close = |_, chan: &mut Channel<_>| { locked_close_channel!($self, chan.context(), UNFUNDED); }; + let locked_close = |_, chan: &mut Channel<_, _>| { locked_close_channel!($self, chan.context(), UNFUNDED); }; convert_channel_err!($self, $peer_state, $err, $channel, do_close, locked_close, chan_id, _internal) } }; ($self: ident, $peer_state: expr, $err: expr, $channel: expr) => { @@ -3953,7 +3963,8 @@ impl< R: Deref, MR: Deref, L: Deref, - > ChannelManager + KV: KVStoreSync + Send + Sync + 'static, + > ChannelManager where M::Target: chain::Watch<::EcdsaSigner>, T::Target: BroadcasterInterface, @@ -3986,7 +3997,8 @@ where pub fn new( fee_est: F, chain_monitor: M, tx_broadcaster: T, router: R, message_router: MR, logger: L, entropy_source: ES, node_signer: NS, signer_provider: SP, config: UserConfig, - params: ChainParameters, current_timestamp: u32, ldk_data_dir: PathBuf + params: ChainParameters, current_timestamp: u32, ldk_data_dir: PathBuf, + rgb_kv_store: Arc, ) -> Self where L: Clone, @@ -4015,7 +4027,7 @@ where best_block: RwLock::new(params.best_block), outbound_scid_aliases: Mutex::new(new_hash_set()), - pending_outbound_payments: OutboundPayments::new(new_hash_map(), ldk_data_dir.clone()), + pending_outbound_payments: OutboundPayments::new(new_hash_map(), Arc::clone(&rgb_kv_store)), forward_htlcs: Mutex::new(new_hash_map()), decode_update_add_htlcs: Mutex::new(new_hash_map()), claimable_payments: Mutex::new(ClaimablePayments { claimable_payments: new_hash_map(), pending_claiming_payments: new_hash_map() }), @@ -4062,6 +4074,7 @@ where testing_dnssec_proof_offer_resolution_override: Mutex::new(new_hash_map()), ldk_data_dir, + rgb_kv_store, } } @@ -4180,7 +4193,8 @@ where }; match OutboundV1Channel::new(&self.fee_estimator, &self.entropy_source, &self.signer_provider, their_network_key, their_features, channel_value_satoshis, push_msat, user_channel_id, config, - self.best_block.read().unwrap().height, outbound_scid_alias, temporary_channel_id, &*self.logger, rgb_asset, self.ldk_data_dir.clone()) + self.best_block.read().unwrap().height, outbound_scid_alias, temporary_channel_id, &*self.logger, rgb_asset, self.ldk_data_dir.clone(), + Arc::clone(&self.rgb_kv_store)) { Ok(res) => res, Err(e) => { @@ -4214,7 +4228,7 @@ where } fn list_funded_channels_with_filter< - Fn: FnMut(&(&InitFeatures, &ChannelId, &Channel)) -> bool, + Fn: FnMut(&(&InitFeatures, &ChannelId, &Channel)) -> bool, >( &self, mut f: Fn, ) -> Vec { @@ -4952,7 +4966,7 @@ where #[rustfmt::skip] fn can_forward_htlc_to_outgoing_channel( - &self, chan: &mut FundedChannel, msg: &msgs::UpdateAddHTLC, next_packet: &NextPacketDetails + &self, chan: &mut FundedChannel, msg: &msgs::UpdateAddHTLC, next_packet: &NextPacketDetails ) -> Result<(), LocalHTLCFailureReason> { if !chan.context.should_announce() && !self.config.read().unwrap().accept_forwards_to_priv_channels @@ -4995,7 +5009,7 @@ where /// Executes a callback `C` that returns some value `X` on the channel found with the given /// `scid`. `None` is returned when the channel is not found. - fn do_funded_channel_callback) -> X>( + fn do_funded_channel_callback) -> X>( &self, scid: u64, callback: C, ) -> Option { let (counterparty_node_id, channel_id) = @@ -5026,7 +5040,7 @@ where return Err(LocalHTLCFailureReason::InvalidTrampolineForward); } }; - match self.do_funded_channel_callback(outgoing_scid, |chan: &mut FundedChannel| { + match self.do_funded_channel_callback(outgoing_scid, |chan: &mut FundedChannel| { self.can_forward_htlc_to_outgoing_channel(chan, msg, next_packet_details) }) { Some(Ok(())) => {}, @@ -5171,7 +5185,7 @@ where /// [`channel_update`]: msgs::ChannelUpdate /// [`internal_closing_signed`]: Self::internal_closing_signed fn get_channel_update_for_broadcast( - &self, chan: &FundedChannel, + &self, chan: &FundedChannel, ) -> Result { if !chan.context.should_announce() { return Err(LightningError { @@ -5206,7 +5220,7 @@ where /// [`channel_update`]: msgs::ChannelUpdate /// [`internal_closing_signed`]: Self::internal_closing_signed #[rustfmt::skip] - fn get_channel_update_for_unicast(&self, chan: &FundedChannel) -> Result { + fn get_channel_update_for_unicast(&self, chan: &FundedChannel) -> Result { let logger = WithChannelContext::from(&self.logger, &chan.context, None); log_trace!(logger, "Attempting to generate channel update for channel {}", chan.context.channel_id()); let short_channel_id = match chan.funding.get_short_channel_id().or(chan.context.latest_inbound_scid_alias()) { @@ -5284,18 +5298,11 @@ where // The top-level caller should hold the total_consistency_lock read lock. debug_assert!(self.total_consistency_lock.try_write().is_err()); - let rgb_payment_info_hash_path_outbound = - get_rgb_payment_info_path(payment_hash, &self.ldk_data_dir, false); - let needs_rgb_modification = if rgb_payment_info_hash_path_outbound.exists() { - let info = parse_rgb_payment_info(&rgb_payment_info_hash_path_outbound); - if !info.swap_payment { - Some(info) - } else { - None - } - } else { - None - }; + let needs_rgb_modification = + match self.rgb_kv_store.read_rgb_payment_info(payment_hash, false) { + Ok(info) if !info.swap_payment => Some(info), + _ => None, + }; let modified_path; let path = if let Some(rgb_payment_info) = needs_rgb_modification { modified_path = { @@ -6166,7 +6173,7 @@ where /// Handles the generation of a funding transaction, optionally (for tests) with a function /// which checks the correctness of the funding transaction given the associated channel. #[rustfmt::skip] - fn funding_transaction_generated_intern) -> Result>( + fn funding_transaction_generated_intern) -> Result>( &self, temporary_channel_id: ChannelId, counterparty_node_id: PublicKey, funding_transaction: Transaction, is_batch_funding: bool, mut find_funding_output: FundingOutput, is_manual_broadcast: bool, ) -> Result<(), APIError> { @@ -6697,7 +6704,7 @@ where } fn broadcast_interactive_funding( - &self, channel: &mut FundedChannel, funding_tx: &Transaction, logger: &L, + &self, channel: &mut FundedChannel, funding_tx: &Transaction, logger: &L, ) { let logger = WithChannelContext::from(logger, channel.context(), None); log_info!( @@ -7013,7 +7020,7 @@ where should_persist = true; let incoming_channel_details_opt = self.do_funded_channel_callback( incoming_scid_alias, - |chan: &mut FundedChannel| { + |chan: &mut FundedChannel| { let counterparty_node_id = chan.context.get_counterparty_node_id(); let channel_id = chan.context.channel_id(); let funding_txo = chan.funding.get_funding_txo().unwrap(); @@ -7097,7 +7104,7 @@ where // Process the HTLC on the incoming channel. match self.do_funded_channel_callback( incoming_scid_alias, - |chan: &mut FundedChannel| { + |chan: &mut FundedChannel| { let logger = WithChannelContext::from( &self.logger, &chan.context, @@ -7601,15 +7608,19 @@ where .contains(&outgoing_amt_msat); if is_in_range && chan.context.is_usable() { if let Some((cid, outgoing_amount_rgb)) = outgoing_rgb_payment { - if !is_channel_rgb(&chan.context.channel_id, &self.ldk_data_dir) - { + if !is_channel_rgb( + &chan.context.channel_id, + self.rgb_kv_store.as_ref(), + ) { return None; } - let (rgb_chan_info, _) = get_rgb_channel_info( - &chan.context.channel_id.0.as_hex().to_string(), - &self.ldk_data_dir, - false, - ); + let rgb_chan_info = self + .rgb_kv_store + .read_rgb_channel_info( + &chan.context.channel_id.0.as_hex().to_string(), + false, + ) + .expect("channel info must exist in KVStore"); if rgb_chan_info.contract_id == *cid && rgb_chan_info.local_rgb_amount >= *outgoing_amount_rgb { @@ -8226,7 +8237,7 @@ where } #[rustfmt::skip] - fn update_channel_fee(&self, chan_id: &ChannelId, chan: &mut FundedChannel, new_feerate: u32) -> NotifyOption { + fn update_channel_fee(&self, chan_id: &ChannelId, chan: &mut FundedChannel, new_feerate: u32) -> NotifyOption { if !chan.funding.is_outbound() { return NotifyOption::SkipPersistNoEvents; } let logger = WithChannelContext::from(&self.logger, &chan.context, None); @@ -9693,7 +9704,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/ /// update completion. #[rustfmt::skip] fn handle_channel_resumption(&self, pending_msg_events: &mut Vec, - channel: &mut FundedChannel, raa: Option, + channel: &mut FundedChannel, raa: Option, commitment_update: Option, commitment_order: RAACommitmentOrder, pending_forwards: Vec<(PendingHTLCInfo, u64)>, pending_update_adds: Vec, funding_broadcastable: Option, @@ -10071,7 +10082,8 @@ This indicates a bug inside LDK. Please report this error at https://github.com/ InboundV1Channel::new( &self.fee_estimator, &self.entropy_source, &self.signer_provider, *counterparty_node_id, &self.channel_type_features(), &peer_state.latest_features, &open_channel_msg, - user_channel_id, &config, best_block_height, &self.logger, accept_0conf, self.ldk_data_dir.clone() + user_channel_id, &config, best_block_height, &self.logger, accept_0conf, self.ldk_data_dir.clone(), + Arc::clone(&self.rgb_kv_store), ).map_err(|err| MsgHandleErrInternal::from_chan_no_close(err, *temporary_channel_id) ).map(|mut channel| { let logger = WithChannelContext::from(&self.logger, &channel.context, None); @@ -10093,6 +10105,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/ user_channel_id, &config, best_block_height, &self.logger, self.ldk_data_dir.clone(), + Arc::clone(&self.rgb_kv_store), ).map_err(|e| { let channel_id = open_channel_msg.common_fields.temporary_channel_id; MsgHandleErrInternal::from_chan_no_close(e, channel_id) @@ -10187,7 +10200,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/ /// non-0-conf channels we have with the peer. fn peers_without_funded_channels(&self, maybe_count_peer: Filter) -> usize where - Filter: Fn(&PeerState) -> bool, + Filter: Fn(&PeerState) -> bool, { let mut peers_without_funded_channels = 0; let best_block_height = self.best_block.read().unwrap().height; @@ -10209,7 +10222,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/ #[rustfmt::skip] fn unfunded_channel_count( - peer: &PeerState, best_block_height: u32 + peer: &PeerState, best_block_height: u32 ) -> usize { let mut num_unfunded_channels = 0; for (_, chan) in peer.channel_by_id.iter() { @@ -10362,7 +10375,8 @@ This indicates a bug inside LDK. Please report this error at https://github.com/ let mut channel = InboundV1Channel::new( &self.fee_estimator, &self.entropy_source, &self.signer_provider, *counterparty_node_id, &self.channel_type_features(), &peer_state.latest_features, msg, user_channel_id, - &self.config.read().unwrap(), best_block_height, &self.logger, /*is_0conf=*/false, self.ldk_data_dir.clone() + &self.config.read().unwrap(), best_block_height, &self.logger, /*is_0conf=*/false, self.ldk_data_dir.clone(), + Arc::clone(&self.rgb_kv_store), ).map_err(|e| MsgHandleErrInternal::from_chan_no_close(e, msg.common_fields.temporary_channel_id))?; let logger = WithChannelContext::from(&self.logger, &channel.context, None); let message_send_event = channel.accept_inbound_channel(&&logger).map(|msg| { @@ -10380,6 +10394,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/ &peer_state.latest_features, msg, user_channel_id, &self.config.read().unwrap(), best_block_height, &self.logger, self.ldk_data_dir.clone(), + Arc::clone(&self.rgb_kv_store), ).map_err(|e| MsgHandleErrInternal::from_chan_no_close(e, msg.common_fields.temporary_channel_id))?; let message_send_event = MessageSendEvent::SendAcceptChannelV2 { node_id: *counterparty_node_id, @@ -10464,7 +10479,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/ Some(Ok(inbound_chan)) => { let logger = WithChannelContext::from(&self.logger, &inbound_chan.context, None); if inbound_chan.funding.is_colored() { - match handle_funding(&msg.temporary_channel_id, msg.funding_txid.to_string(), &self.ldk_data_dir, inbound_chan.funding.push_asset_amount()) { + match handle_funding(&msg.temporary_channel_id, msg.funding_txid.to_string(), &self.ldk_data_dir, inbound_chan.funding.push_asset_amount(), self.rgb_kv_store.as_ref()) { Ok(()) => (), Err(e) => { // at this point the channel initiator already transitioned its channel to the funded channel ID @@ -10728,7 +10743,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/ fn internal_tx_msg< HandleTxMsgFn: Fn( - &mut Channel, + &mut Channel, ) -> Result)>, >( &self, counterparty_node_id: &PublicKey, channel_id: ChannelId, @@ -10782,33 +10797,41 @@ This indicates a bug inside LDK. Please report this error at https://github.com/ fn internal_tx_add_input( &self, counterparty_node_id: PublicKey, msg: &msgs::TxAddInput, ) -> Result { - self.internal_tx_msg(&counterparty_node_id, msg.channel_id, |channel: &mut Channel| { - channel.tx_add_input(msg, &self.logger) - }) + self.internal_tx_msg( + &counterparty_node_id, + msg.channel_id, + |channel: &mut Channel| channel.tx_add_input(msg, &self.logger), + ) } fn internal_tx_add_output( &self, counterparty_node_id: PublicKey, msg: &msgs::TxAddOutput, ) -> Result { - self.internal_tx_msg(&counterparty_node_id, msg.channel_id, |channel: &mut Channel| { - channel.tx_add_output(msg, &self.logger) - }) + self.internal_tx_msg( + &counterparty_node_id, + msg.channel_id, + |channel: &mut Channel| channel.tx_add_output(msg, &self.logger), + ) } fn internal_tx_remove_input( &self, counterparty_node_id: PublicKey, msg: &msgs::TxRemoveInput, ) -> Result { - self.internal_tx_msg(&counterparty_node_id, msg.channel_id, |channel: &mut Channel| { - channel.tx_remove_input(msg, &self.logger) - }) + self.internal_tx_msg( + &counterparty_node_id, + msg.channel_id, + |channel: &mut Channel| channel.tx_remove_input(msg, &self.logger), + ) } fn internal_tx_remove_output( &self, counterparty_node_id: PublicKey, msg: &msgs::TxRemoveOutput, ) -> Result { - self.internal_tx_msg(&counterparty_node_id, msg.channel_id, |channel: &mut Channel| { - channel.tx_remove_output(msg, &self.logger) - }) + self.internal_tx_msg( + &counterparty_node_id, + msg.channel_id, + |channel: &mut Channel| channel.tx_remove_output(msg, &self.logger), + ) } #[rustfmt::skip] @@ -12379,7 +12402,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/ for (_cp_id, peer_state_mutex) in per_peer_state.iter() { 'chan_loop: loop { let mut peer_state_lock = peer_state_mutex.lock().unwrap(); - let peer_state: &mut PeerState<_> = &mut *peer_state_lock; + let peer_state: &mut PeerState<_, _> = &mut *peer_state_lock; for (channel_id, chan) in peer_state.channel_by_id.iter_mut().filter_map(|(chan_id, chan)| { chan.as_funded_mut().map(|chan| (chan_id, chan)) @@ -12439,7 +12462,7 @@ This indicates a bug inside LDK. Please report this error at https://github.com/ let _persistence_guard = PersistenceNotifierGuard::notify_on_drop(self); // Returns whether we should remove this channel as it's just been closed. - let unblock_chan = |chan: &mut Channel, pending_msg_events: &mut Vec| -> Option { + let unblock_chan = |chan: &mut Channel, pending_msg_events: &mut Vec| -> Option { let channel_id = chan.context().channel_id(); let outbound_scid_alias = chan.context().outbound_scid_alias(); let logger = WithChannelContext::from(&self.logger, &chan.context(), None); @@ -13081,7 +13104,8 @@ impl< R: Deref, MR: Deref, L: Deref, - > ChannelManager + KV: KVStoreSync + Send + Sync + 'static, + > ChannelManager where M::Target: chain::Watch<::EcdsaSigner>, T::Target: BroadcasterInterface, @@ -13950,7 +13974,8 @@ impl< R: Deref, MR: Deref, L: Deref, - > BaseMessageHandler for ChannelManager + KV: KVStoreSync + Send + Sync + 'static, + > BaseMessageHandler for ChannelManager where M::Target: chain::Watch<::EcdsaSigner>, T::Target: BroadcasterInterface, @@ -14291,7 +14316,8 @@ impl< R: Deref, MR: Deref, L: Deref, - > EventsProvider for ChannelManager + KV: KVStoreSync + Send + Sync + 'static, + > EventsProvider for ChannelManager where M::Target: chain::Watch<::EcdsaSigner>, T::Target: BroadcasterInterface, @@ -14326,7 +14352,8 @@ impl< R: Deref, MR: Deref, L: Deref, - > chain::Listen for ChannelManager + KV: KVStoreSync + Send + Sync + 'static, + > chain::Listen for ChannelManager where M::Target: chain::Watch<::EcdsaSigner>, T::Target: BroadcasterInterface, @@ -14387,7 +14414,8 @@ impl< R: Deref, MR: Deref, L: Deref, - > chain::Confirm for ChannelManager + KV: KVStoreSync + Send + Sync + 'static, + > chain::Confirm for ChannelManager where M::Target: chain::Watch<::EcdsaSigner>, T::Target: BroadcasterInterface, @@ -14417,7 +14445,7 @@ where let last_best_block_height = self.best_block.read().unwrap().height; if height < last_best_block_height { let timestamp = self.highest_seen_timestamp.load(Ordering::Acquire); - let do_update = |channel: &mut FundedChannel| { + let do_update = |channel: &mut FundedChannel| { channel.best_block_updated( last_best_block_height, Some(timestamp as u32), @@ -14560,7 +14588,8 @@ impl< R: Deref, MR: Deref, L: Deref, - > ChannelManager + KV: KVStoreSync + Send + Sync + 'static, + > ChannelManager where M::Target: chain::Watch<::EcdsaSigner>, T::Target: BroadcasterInterface, @@ -14576,7 +14605,7 @@ where /// un/confirmed, etc) on each channel, handling any resulting errors or messages generated by /// the function. #[rustfmt::skip] - fn do_chain_event) -> Result<(Option, Vec<(HTLCSource, PaymentHash)>, Option), ClosureReason>> + fn do_chain_event) -> Result<(Option, Vec<(HTLCSource, PaymentHash)>, Option), ClosureReason>> (&self, height_opt: Option, f: FN) { // Note that we MUST NOT end up calling methods on self.chain_monitor here - we're called // during initialization prior to the chain_monitor being fully configured in some cases. @@ -14888,7 +14917,8 @@ impl< R: Deref, MR: Deref, L: Deref, - > ChannelMessageHandler for ChannelManager + KV: KVStoreSync + Send + Sync + 'static, + > ChannelMessageHandler for ChannelManager where M::Target: chain::Watch<::EcdsaSigner>, T::Target: BroadcasterInterface, @@ -15463,7 +15493,8 @@ impl< R: Deref, MR: Deref, L: Deref, - > OffersMessageHandler for ChannelManager + KV: KVStoreSync + Send + Sync + 'static, + > OffersMessageHandler for ChannelManager where M::Target: chain::Watch<::EcdsaSigner>, T::Target: BroadcasterInterface, @@ -15633,7 +15664,8 @@ impl< R: Deref, MR: Deref, L: Deref, - > AsyncPaymentsMessageHandler for ChannelManager + KV: KVStoreSync + Send + Sync + 'static, + > AsyncPaymentsMessageHandler for ChannelManager where M::Target: chain::Watch<::EcdsaSigner>, T::Target: BroadcasterInterface, @@ -15835,7 +15867,8 @@ impl< R: Deref, MR: Deref, L: Deref, - > DNSResolverMessageHandler for ChannelManager + KV: KVStoreSync + Send + Sync + 'static, + > DNSResolverMessageHandler for ChannelManager where M::Target: chain::Watch<::EcdsaSigner>, T::Target: BroadcasterInterface, @@ -15903,7 +15936,8 @@ impl< R: Deref, MR: Deref, L: Deref, - > NodeIdLookUp for ChannelManager + KV: KVStoreSync + Send + Sync + 'static, + > NodeIdLookUp for ChannelManager where M::Target: chain::Watch<::EcdsaSigner>, T::Target: BroadcasterInterface, @@ -16421,7 +16455,8 @@ impl< R: Deref, MR: Deref, L: Deref, - > Writeable for ChannelManager + KV: KVStoreSync + Send + Sync + 'static, + > Writeable for ChannelManager where M::Target: chain::Watch<::EcdsaSigner>, T::Target: BroadcasterInterface, @@ -16788,6 +16823,7 @@ pub struct ChannelManagerReadArgs< R: Deref, MR: Deref, L: Deref + Clone, + KV: KVStoreSync + Send + Sync + 'static, > where M::Target: chain::Watch<::EcdsaSigner>, T::Target: BroadcasterInterface, @@ -16858,6 +16894,9 @@ pub struct ChannelManagerReadArgs< /// LDK data directory pub ldk_data_dir: PathBuf, + + /// KVStore for RGB data persistence + pub rgb_kv_store: Arc, } impl< @@ -16871,7 +16910,8 @@ impl< R: Deref, MR: Deref, L: Deref + Clone, - > ChannelManagerReadArgs<'a, M, T, ES, NS, SP, F, R, MR, L> + KV: KVStoreSync + Send + Sync + 'static, + > ChannelManagerReadArgs<'a, M, T, ES, NS, SP, F, R, MR, L, KV> where M::Target: chain::Watch<::EcdsaSigner>, T::Target: BroadcasterInterface, @@ -16891,7 +16931,7 @@ where chain_monitor: M, tx_broadcaster: T, router: R, message_router: MR, logger: L, config: UserConfig, mut channel_monitors: Vec<&'a ChannelMonitor<::EcdsaSigner>>, - ldk_data_dir: PathBuf, + ldk_data_dir: PathBuf, rgb_kv_store: Arc, ) -> Self { Self { entropy_source, @@ -16908,6 +16948,7 @@ where channel_monitors.drain(..).map(|monitor| (monitor.channel_id(), monitor)), ), ldk_data_dir, + rgb_kv_store, } } } @@ -16925,8 +16966,9 @@ impl< R: Deref, MR: Deref, L: Deref + Clone, - > ReadableArgs> - for (BlockHash, Arc>) + KV: KVStoreSync + Send + Sync + 'static, + > ReadableArgs> + for (BlockHash, Arc>) where M::Target: chain::Watch<::EcdsaSigner>, T::Target: BroadcasterInterface, @@ -16939,10 +16981,10 @@ where L::Target: Logger, { fn read( - reader: &mut Reader, args: ChannelManagerReadArgs<'a, M, T, ES, NS, SP, F, R, MR, L>, + reader: &mut Reader, args: ChannelManagerReadArgs<'a, M, T, ES, NS, SP, F, R, MR, L, KV>, ) -> Result { let (blockhash, chan_manager) = - <(BlockHash, ChannelManager)>::read(reader, args)?; + <(BlockHash, ChannelManager)>::read(reader, args)?; Ok((blockhash, Arc::new(chan_manager))) } } @@ -16958,8 +17000,9 @@ impl< R: Deref, MR: Deref, L: Deref + Clone, - > ReadableArgs> - for (BlockHash, ChannelManager) + KV: KVStoreSync + Send + Sync + 'static, + > ReadableArgs> + for (BlockHash, ChannelManager) where M::Target: chain::Watch<::EcdsaSigner>, T::Target: BroadcasterInterface, @@ -16972,7 +17015,8 @@ where L::Target: Logger, { fn read( - reader: &mut Reader, mut args: ChannelManagerReadArgs<'a, M, T, ES, NS, SP, F, R, MR, L>, + reader: &mut Reader, + mut args: ChannelManagerReadArgs<'a, M, T, ES, NS, SP, F, R, MR, L, KV>, ) -> Result { let _ver = read_ver_prefix!(reader, SERIALIZATION_VERSION); @@ -16998,19 +17042,20 @@ where let mut channel_id_set = hash_set_with_capacity(cmp::min(channel_count as usize, 128)); let mut per_peer_state = hash_map_with_capacity(cmp::min( channel_count as usize, - MAX_ALLOC_SIZE / mem::size_of::<(PublicKey, Mutex>)>(), + MAX_ALLOC_SIZE / mem::size_of::<(PublicKey, Mutex>)>(), )); let mut short_to_chan_info = hash_map_with_capacity(cmp::min(channel_count as usize, 128)); let mut channel_closures = VecDeque::new(); let mut close_background_events = Vec::new(); for _ in 0..channel_count { - let mut channel: FundedChannel = FundedChannel::read( + let mut channel: FundedChannel = FundedChannel::read( reader, ( &args.entropy_source, &args.signer_provider, &provided_channel_type_features(&args.config), args.ldk_data_dir.clone(), + Arc::clone(&args.rgb_kv_store), ), )?; let logger = WithChannelContext::from(&args.logger, &channel.context, None); @@ -17440,8 +17485,10 @@ where } pending_outbound_payments = Some(outbounds); } - let pending_outbounds = - OutboundPayments::new(pending_outbound_payments.unwrap(), args.ldk_data_dir.clone()); + let pending_outbounds = OutboundPayments::new( + pending_outbound_payments.unwrap(), + Arc::clone(&args.rgb_kv_store), + ); for (peer_pubkey, peer_storage) in peer_storage_dir { if let Some(peer_state) = per_peer_state.get_mut(&peer_pubkey) { @@ -18384,6 +18431,7 @@ where testing_dnssec_proof_offer_resolution_override: Mutex::new(new_hash_map()), ldk_data_dir: args.ldk_data_dir, + rgb_kv_store: args.rgb_kv_store, }; let mut processed_claims: HashSet> = new_hash_set(); diff --git a/lightning/src/ln/outbound_payment.rs b/lightning/src/ln/outbound_payment.rs index 3a4fd4ff9..c48c6ff10 100644 --- a/lightning/src/ln/outbound_payment.rs +++ b/lightning/src/ln/outbound_payment.rs @@ -26,9 +26,7 @@ use crate::offers::invoice::{Bolt12Invoice, DerivedSigningPubkey, InvoiceBuilder use crate::offers::invoice_request::InvoiceRequest; use crate::offers::nonce::Nonce; use crate::offers::static_invoice::StaticInvoice; -use crate::rgb_utils::{ - filter_first_hops, get_rgb_payment_info_path, is_payment_rgb, parse_rgb_payment_info, -}; +use crate::rgb_utils::RgbKvStoreExt; use crate::routing::router::{ BlindedTail, InFlightHtlcs, Path, PaymentParameters, Route, RouteParameters, RouteParametersConfig, Router, @@ -47,10 +45,9 @@ use core::ops::Deref; use core::sync::atomic::{AtomicBool, Ordering}; use core::time::Duration; -use std::path::PathBuf; - use crate::prelude::*; -use crate::sync::Mutex; +use crate::sync::{Arc, Mutex}; +use crate::util::persist::KVStoreSync; /// The number of ticks of [`ChannelManager::timer_tick_occurred`] until we time-out the idempotency /// of payments by [`PaymentId`]. See [`OutboundPayments::remove_stale_payments`]. @@ -842,17 +839,17 @@ pub(super) struct SendAlongPathArgs<'a> { pub hold_htlc_at_next_hop: bool, } -pub(super) struct OutboundPayments { +pub(super) struct OutboundPayments { pub(super) pending_outbound_payments: Mutex>, awaiting_invoice: AtomicBool, retry_lock: Mutex<()>, - ldk_data_dir: PathBuf, + rgb_kv_store: Arc, } -impl OutboundPayments { +impl OutboundPayments { pub(super) fn new( pending_outbound_payments: HashMap, - ldk_data_dir: PathBuf, + rgb_kv_store: Arc, ) -> Self { let has_invoice_requests = pending_outbound_payments.values().any(|payment| { matches!( @@ -867,12 +864,12 @@ impl OutboundPayments { pending_outbound_payments: Mutex::new(pending_outbound_payments), awaiting_invoice: AtomicBool::new(has_invoice_requests), retry_lock: Mutex::new(()), - ldk_data_dir, + rgb_kv_store, } } } -impl OutboundPayments { +impl OutboundPayments { #[rustfmt::skip] pub(super) fn send_payment( &self, payment_hash: PaymentHash, recipient_onion: RecipientOnionFields, payment_id: PaymentId, @@ -1049,9 +1046,11 @@ impl OutboundPayments { } let mut filtered_first_hops = first_hops.into_iter().collect::>(); - let rgb_payment = is_payment_rgb(&self.ldk_data_dir, &payment_hash).then(|| { - filter_first_hops(&self.ldk_data_dir, &payment_hash, &mut filtered_first_hops) - }); + let rgb_payment = if self.rgb_kv_store.is_payment_rgb(&payment_hash) { + Some(self.rgb_kv_store.filter_first_hops(&payment_hash, &mut filtered_first_hops)) + } else { + None + }; let mut route_params = RouteParameters::from_payment_params_and_value( PaymentParameters::from_bolt12_invoice(&invoice) .with_user_config_ignoring_fee_limit(params_config), invoice.amount_msats(), @@ -1410,14 +1409,11 @@ impl OutboundPayments { .. } = pmt { - let rgb_payment_info_path = - get_rgb_payment_info_path(payment_hash, &self.ldk_data_dir, false); - let rgb_payment = if rgb_payment_info_path.exists() { - let rgb_payment_info = parse_rgb_payment_info(&rgb_payment_info_path); - Some((rgb_payment_info.contract_id, rgb_payment_info.amount)) - } else { - None - }; + let rgb_payment = + match self.rgb_kv_store.read_rgb_payment_info(payment_hash, false) { + Ok(info) => Some((info.contract_id, info.amount)), + Err(_) => None, + }; if pending_amt_msat < total_msat { retry_id_route_params = Some(( *payment_hash, @@ -1569,9 +1565,9 @@ impl OutboundPayments { SP: Fn(SendAlongPathArgs) -> Result<(), APIError>, { let mut filtered_first_hops = first_hops.into_iter().collect::>(); - is_payment_rgb(&self.ldk_data_dir, &payment_hash).then(|| { - filter_first_hops(&self.ldk_data_dir, &payment_hash, &mut filtered_first_hops) - }); + if self.rgb_kv_store.is_payment_rgb(&payment_hash) { + self.rgb_kv_store.filter_first_hops(&payment_hash, &mut filtered_first_hops); + } let route = self.find_initial_route( payment_id, payment_hash, &recipient_onion, keysend_preimage, None, &mut route_params, router, &filtered_first_hops, &inflight_htlcs, node_signer, best_block_height, logger, @@ -1626,9 +1622,9 @@ impl OutboundPayments { } let mut filtered_first_hops = first_hops.into_iter().collect::>(); - is_payment_rgb(&self.ldk_data_dir, &payment_hash).then(|| { - filter_first_hops(&self.ldk_data_dir, &payment_hash, &mut filtered_first_hops) - }); + if self.rgb_kv_store.is_payment_rgb(&payment_hash) { + self.rgb_kv_store.filter_first_hops(&payment_hash, &mut filtered_first_hops); + } let mut route = match router.find_route_with_id( &node_signer.get_node_id(Recipient::Node).unwrap(), &route_params, diff --git a/lightning/src/ln/peer_handler.rs b/lightning/src/ln/peer_handler.rs index 8ae548b3b..8f8c41c1a 100644 --- a/lightning/src/ln/peer_handler.rs +++ b/lightning/src/ln/peer_handler.rs @@ -913,15 +913,25 @@ impl Peer { /// /// This is not exported to bindings users as type aliases aren't supported in most languages. #[cfg(not(c_bindings))] -pub type SimpleArcPeerManager = PeerManager< +pub type SimpleArcPeerManager = PeerManager< SD, - Arc>, + Arc>, Arc>>, C, Arc>>, - Arc>, + Arc>, Arc, IgnoringMessageHandler, - Arc, - Arc, Arc, Arc, Arc, Arc, Arc>>, + Arc>, + Arc< + ChainMonitor< + InMemorySigner, + Arc, + Arc, + Arc, + Arc, + Arc, + Arc>, + >, + >, >; /// SimpleRefPeerManager is a type alias for a PeerManager reference, and is the reference @@ -935,16 +945,16 @@ pub type SimpleArcPeerManager = PeerManager< #[cfg(not(c_bindings))] #[rustfmt::skip] pub type SimpleRefPeerManager< - 'a, 'b, 'c, 'd, 'e, 'f, 'logger, 'h, 'i, 'j, 'graph, 'k, 'mr, SD, M, T, F, C, L + 'a, 'b, 'c, 'd, 'e, 'f, 'logger, 'h, 'i, 'j, 'graph, 'k, 'mr, SD, M, T, F, C, L, KV > = PeerManager< SD, - &'j SimpleRefChannelManager<'a, 'b, 'c, 'd, 'e, 'graph, 'logger, 'i, 'mr, M, T, F, L>, + &'j SimpleRefChannelManager<'a, 'b, 'c, 'd, 'e, 'graph, 'logger, 'i, 'mr, M, T, F, L, KV>, &'f P2PGossipSync<&'graph NetworkGraph<&'logger L>, C, &'logger L>, - &'h SimpleRefOnionMessenger<'a, 'b, 'c, 'd, 'e, 'graph, 'logger, 'i, 'j, 'k, M, T, F, L>, + &'h SimpleRefOnionMessenger<'a, 'b, 'c, 'd, 'e, 'graph, 'logger, 'i, 'j, 'k, M, T, F, L, KV>, &'logger L, IgnoringMessageHandler, - &'c KeysManager, - &'j ChainMonitor<&'a M, C, &'b T, &'c F, &'logger L, &'c KeysManager, &'c KeysManager>, + &'c KeysManager, + &'j ChainMonitor<&'a M, C, &'b T, &'c F, &'logger L, &'c KeysManager, &'c KeysManager>, >; /// A generic trait which is implemented for all [`PeerManager`]s. This makes bounding functions or diff --git a/lightning/src/onion_message/messenger.rs b/lightning/src/onion_message/messenger.rs index 9a2c06bb7..32411f5dc 100644 --- a/lightning/src/onion_message/messenger.rs +++ b/lightning/src/onion_message/messenger.rs @@ -2381,15 +2381,15 @@ where /// [`SimpleArcPeerManager`]: crate::ln::peer_handler::SimpleArcPeerManager #[cfg(not(c_bindings))] #[cfg(feature = "dnssec")] -pub type SimpleArcOnionMessenger = OnionMessenger< - Arc, - Arc, +pub type SimpleArcOnionMessenger = OnionMessenger< + Arc>, + Arc>, Arc, - Arc>, - Arc>>, Arc, Arc>>, - Arc>, - Arc>, - Arc>, + Arc>, + Arc>>, Arc, Arc>>>, + Arc>, + Arc>, + Arc>, IgnoringMessageHandler, >; @@ -2402,14 +2402,14 @@ pub type SimpleArcOnionMessenger = OnionMessenger< /// [`SimpleArcPeerManager`]: crate::ln::peer_handler::SimpleArcPeerManager #[cfg(not(c_bindings))] #[cfg(not(feature = "dnssec"))] -pub type SimpleArcOnionMessenger = OnionMessenger< - Arc, - Arc, +pub type SimpleArcOnionMessenger = OnionMessenger< + Arc>, + Arc>, Arc, - Arc>, - Arc>>, Arc, Arc>>, - Arc>, - Arc>, + Arc>, + Arc>>, Arc, Arc>>>, + Arc>, + Arc>, IgnoringMessageHandler, IgnoringMessageHandler, >; @@ -2423,16 +2423,16 @@ pub type SimpleArcOnionMessenger = OnionMessenger< /// [`SimpleRefPeerManager`]: crate::ln::peer_handler::SimpleRefPeerManager #[cfg(not(c_bindings))] #[cfg(feature = "dnssec")] -pub type SimpleRefOnionMessenger<'a, 'b, 'c, 'd, 'e, 'f, 'g, 'h, 'i, 'j, M, T, F, L> = +pub type SimpleRefOnionMessenger<'a, 'b, 'c, 'd, 'e, 'f, 'g, 'h, 'i, 'j, M, T, F, L, KV> = OnionMessenger< - &'a KeysManager, - &'a KeysManager, + &'a KeysManager, + &'a KeysManager, &'b L, - &'j SimpleRefChannelManager<'a, 'b, 'c, 'd, 'e, 'f, 'g, 'h, 'i, M, T, F, L>, - &'i DefaultMessageRouter<&'g NetworkGraph<&'b L>, &'b L, &'a KeysManager>, - &'j SimpleRefChannelManager<'a, 'b, 'c, 'd, 'e, 'f, 'g, 'h, 'i, M, T, F, L>, - &'j SimpleRefChannelManager<'a, 'b, 'c, 'd, 'e, 'f, 'g, 'h, 'i, M, T, F, L>, - &'j SimpleRefChannelManager<'a, 'b, 'c, 'd, 'e, 'f, 'g, 'h, 'i, M, T, F, L>, + &'j SimpleRefChannelManager<'a, 'b, 'c, 'd, 'e, 'f, 'g, 'h, 'i, M, T, F, L, KV>, + &'i DefaultMessageRouter<&'g NetworkGraph<&'b L>, &'b L, &'a KeysManager>, + &'j SimpleRefChannelManager<'a, 'b, 'c, 'd, 'e, 'f, 'g, 'h, 'i, M, T, F, L, KV>, + &'j SimpleRefChannelManager<'a, 'b, 'c, 'd, 'e, 'f, 'g, 'h, 'i, M, T, F, L, KV>, + &'j SimpleRefChannelManager<'a, 'b, 'c, 'd, 'e, 'f, 'g, 'h, 'i, M, T, F, L, KV>, IgnoringMessageHandler, >; @@ -2445,15 +2445,15 @@ pub type SimpleRefOnionMessenger<'a, 'b, 'c, 'd, 'e, 'f, 'g, 'h, 'i, 'j, M, T, F /// [`SimpleRefPeerManager`]: crate::ln::peer_handler::SimpleRefPeerManager #[cfg(not(c_bindings))] #[cfg(not(feature = "dnssec"))] -pub type SimpleRefOnionMessenger<'a, 'b, 'c, 'd, 'e, 'f, 'g, 'h, 'i, 'j, M, T, F, L> = +pub type SimpleRefOnionMessenger<'a, 'b, 'c, 'd, 'e, 'f, 'g, 'h, 'i, 'j, M, T, F, L, KV> = OnionMessenger< - &'a KeysManager, - &'a KeysManager, + &'a KeysManager, + &'a KeysManager, &'b L, - &'j SimpleRefChannelManager<'a, 'b, 'c, 'd, 'e, 'f, 'g, 'h, 'i, M, T, F, L>, - &'i DefaultMessageRouter<&'g NetworkGraph<&'b L>, &'b L, &'a KeysManager>, - &'j SimpleRefChannelManager<'a, 'b, 'c, 'd, 'e, 'f, 'g, 'h, 'i, M, T, F, L>, - &'j SimpleRefChannelManager<'a, 'b, 'c, 'd, 'e, 'f, 'g, 'h, 'i, M, T, F, L>, + &'j SimpleRefChannelManager<'a, 'b, 'c, 'd, 'e, 'f, 'g, 'h, 'i, M, T, F, L, KV>, + &'i DefaultMessageRouter<&'g NetworkGraph<&'b L>, &'b L, &'a KeysManager>, + &'j SimpleRefChannelManager<'a, 'b, 'c, 'd, 'e, 'f, 'g, 'h, 'i, M, T, F, L, KV>, + &'j SimpleRefChannelManager<'a, 'b, 'c, 'd, 'e, 'f, 'g, 'h, 'i, M, T, F, L, KV>, IgnoringMessageHandler, IgnoringMessageHandler, >; diff --git a/lightning/src/rgb_utils/mod.rs b/lightning/src/rgb_utils/mod.rs index 2973bc762..0ca260d70 100644 --- a/lightning/src/rgb_utils/mod.rs +++ b/lightning/src/rgb_utils/mod.rs @@ -10,6 +10,7 @@ use crate::ln::types::ChannelId; use crate::sign::SignerProvider; use crate::types::features::ChannelTypeFeatures; use crate::types::payment::PaymentHash; +use crate::util::persist::{KVStoreSync, KvOp}; use bitcoin::blockdata::transaction::Transaction; use bitcoin::hashes::{sha256, Hash}; @@ -27,24 +28,46 @@ use rgb_lib::{ WitnessOrd, }; use serde::{Deserialize, Serialize}; -use tokio::runtime::Handle; +use crate::io; use core::ops::Deref; use std::collections::{HashMap, HashSet}; use std::fs; use std::path::{Path, PathBuf}; use std::str::FromStr; -/// Static blinding costant (will be removed in the future) +/// Static blinding constant (will be removed in the future) pub const STATIC_BLINDING: u64 = 777; -/// Name of the file containing the electrum URL +/// KVStore key for the bitcoin network +pub const BITCOIN_NETWORK_FNAME: &str = "bitcoin_network"; +/// KVStore key for the electrum URL pub const INDEXER_URL_FNAME: &str = "indexer_url"; -/// Name of the file containing the master fingerprint of the wallet +/// KVStore key for the wallet fingerprint +pub const WALLET_FINGERPRINT_FNAME: &str = "wallet_fingerprint"; +/// KVStore key for the account-level xPub of the vanilla-side of the wallet +pub const WALLET_ACCOUNT_XPUB_VANILLA_FNAME: &str = "wallet_account_xpub_vanilla"; +/// KVStore key for the account-level xPub of the colored-side of the wallet +pub const WALLET_ACCOUNT_XPUB_COLORED_FNAME: &str = "wallet_account_xpub_colored"; +/// KVStore key for the master fingerprint of the wallet pub const WALLET_MASTER_FINGERPRINT_FNAME: &str = "wallet_master_fingerprint"; -const INBOUND_EXT: &str = "inbound"; -const OUTBOUND_EXT: &str = "outbound"; const VANILLA_SYNC_LOOKBACK: u32 = 20; +// kv_store namespace constants for RGB data persistence +/// Primary namespace for all RGB data +pub const RGB_PRIMARY_NS: &str = "rgb"; +/// Secondary namespace for channel info +pub const RGB_CHANNEL_INFO_NS: &str = "channel_info"; +/// Secondary namespace for pending channel info +pub const RGB_CHANNEL_INFO_PENDING_NS: &str = "channel_info_pending"; +/// Secondary namespace for inbound payment info +pub const RGB_PAYMENT_INFO_INBOUND_NS: &str = "payment_info_inbound"; +/// Secondary namespace for outbound payment info +pub const RGB_PAYMENT_INFO_OUTBOUND_NS: &str = "payment_info_outbound"; +/// Secondary namespace for transfer info +pub const RGB_TRANSFER_INFO_NS: &str = "transfer_info"; +/// Secondary namespace for wallet config values +pub const RGB_WALLET_CONFIG_NS: &str = "wallet_config"; + /// RGB channel info #[derive(Debug, Clone, Deserialize, Serialize)] pub struct RgbInfo { @@ -58,7 +81,6 @@ pub struct RgbInfo { /// Channel RGB remote amount pub remote_rgb_amount: u64, /// Batch transfer index from rgb-lib (set after rgb_send_begin) - #[serde(default, skip_serializing_if = "Option::is_none")] pub batch_transfer_idx: Option, /// Whether the channel acceptor told us (in `accept_channel`) that it already knows the asset #[serde(default)] @@ -114,55 +136,49 @@ mod contract_id_serde { } } -fn _get_file_in_parent(ldk_data_dir: &Path, fname: &str) -> PathBuf { - ldk_data_dir.parent().unwrap().join(fname) -} - -fn _read_file_in_parent(ldk_data_dir: &Path, fname: &str) -> String { - fs::read_to_string(_get_file_in_parent(ldk_data_dir, fname)).unwrap() -} - -fn _get_master_fingerprint(ldk_data_dir: &Path) -> String { - _read_file_in_parent(ldk_data_dir, WALLET_MASTER_FINGERPRINT_FNAME) +fn _get_master_fingerprint(kv_store: &K) -> String { + kv_store + .read_config(WALLET_MASTER_FINGERPRINT_FNAME) + .expect("wallet_master_fingerprint must be in KVStore") } -fn _get_indexer_url(ldk_data_dir: &Path) -> String { - _read_file_in_parent(ldk_data_dir, INDEXER_URL_FNAME) +fn _get_indexer_url(kv_store: &K) -> String { + kv_store.read_config(INDEXER_URL_FNAME).expect("indexer_url must be in KVStore") } fn _load_rgb_wallet(data_dir: String, master_fingerprint: String) -> Wallet { Wallet::load(&data_dir, &master_fingerprint, None).expect("valid rgb-lib wallet") } -fn _get_wallet_data(ldk_data_dir: &Path) -> (String, String) { +fn _get_wallet_data( + ldk_data_dir: &Path, kv_store: &K, +) -> (String, String) { let data_dir = ldk_data_dir.parent().unwrap().to_string_lossy().to_string(); - let master_fingerprint = _get_master_fingerprint(ldk_data_dir); + let master_fingerprint = _get_master_fingerprint(kv_store); (data_dir, master_fingerprint) } -async fn _get_rgb_wallet(ldk_data_dir: &Path) -> Wallet { - let (data_dir, master_fingerprint) = _get_wallet_data(ldk_data_dir); - tokio::task::spawn_blocking(move || _load_rgb_wallet(data_dir, master_fingerprint)) - .await - .unwrap() +fn _get_rgb_wallet(ldk_data_dir: &Path, kv_store: &K) -> Wallet { + let (data_dir, master_fingerprint) = _get_wallet_data(ldk_data_dir, kv_store); + tokio::task::block_in_place(move || _load_rgb_wallet(data_dir, master_fingerprint)) } -pub(crate) fn is_asset_known(contract_id: ContractId, ldk_data_dir: &Path) -> bool { - let handle = Handle::current(); - let _ = handle.enter(); - let wallet = futures::executor::block_on(_get_rgb_wallet(ldk_data_dir)); +pub(crate) fn is_asset_known( + contract_id: ContractId, ldk_data_dir: &Path, kv_store: &K, +) -> bool { + let wallet = _get_rgb_wallet(ldk_data_dir, kv_store); wallet.is_asset_known(contract_id).unwrap_or(false) } -async fn _accept_transfer( - ldk_data_dir: &Path, funding_txid: String, +fn _accept_transfer( + ldk_data_dir: &Path, funding_txid: String, kv_store: &K, ) -> Result<(RgbTransfer, Vec, HashSet, PathBuf), RgbLibError> { let funding_vout = 1; - let (data_dir, master_fingerprint) = _get_wallet_data(ldk_data_dir); - let indexer_url = _get_indexer_url(ldk_data_dir); + let (data_dir, master_fingerprint) = _get_wallet_data(ldk_data_dir, kv_store); + let indexer_url = _get_indexer_url(kv_store); // the consignment is received from the channel counterparty over the p2p link and written to disk let consignment_path = ldk_data_dir.join(format!("consignment_{funding_txid}")); - tokio::task::spawn_blocking(move || { + tokio::task::block_in_place(move || { let mut wallet = _load_rgb_wallet(data_dir, master_fingerprint); let online = wallet.go_online(OnlineOptions { indexer_url, @@ -178,20 +194,6 @@ async fn _accept_transfer( )?; Ok((consignment, assignments, media_digests, wallet.get_media_dir())) }) - .await - .unwrap() -} - -/// Read TransferInfo file -pub fn read_rgb_transfer_info(path: &Path) -> TransferInfo { - let serialized_info = fs::read_to_string(path).expect("able to read transfer info file"); - serde_json::from_str(&serialized_info).expect("valid transfer info") -} - -/// Write TransferInfo file -pub fn write_rgb_transfer_info(path: &PathBuf, info: &TransferInfo) { - let serialized_info = serde_json::to_string(&info).expect("valid transfer info"); - fs::write(path, serialized_info).expect("able to write transfer info file") } fn _counterparty_output_index( @@ -217,8 +219,8 @@ pub fn is_tx_colored(tx: &Transaction) -> bool { } /// Color commitment transaction -pub(crate) fn color_commitment( - channel_context: &ChannelContext, funding_scope: &FundingScope, +pub(crate) fn color_commitment( + channel_context: &ChannelContext, funding_scope: &FundingScope, commitment_transaction: &mut CommitmentTransaction, counterparty: bool, ) -> Result<(), ChannelError> where @@ -226,10 +228,11 @@ where { let channel_id = &channel_context.channel_id; let ldk_data_dir = channel_context.ldk_data_dir.as_path(); + let kv_store = channel_context.rgb_kv_store.as_ref(); let commitment_tx = commitment_transaction.clone().built.transaction; - let (rgb_info, _) = get_rgb_channel_info_pending(channel_id, ldk_data_dir); + let rgb_info = get_rgb_channel_info_pending(channel_id, kv_store); let contract_id = rgb_info.contract_id; let chan_id = channel_id.0.as_hex(); @@ -250,48 +253,66 @@ where let htlc_payment_hash = htlc.payment_hash.0.as_hex().to_string(); let htlc_proxy_id = format!("{chan_id}{htlc_payment_hash}"); - let mut rgb_payment_info_proxy_id_path = ldk_data_dir.join(htlc_proxy_id); - let rgb_payment_info_path = ldk_data_dir.join(htlc_payment_hash); - let mut rgb_payment_info_path = rgb_payment_info_path.clone(); - if inbound { - rgb_payment_info_proxy_id_path.set_extension(INBOUND_EXT); - rgb_payment_info_path.set_extension(INBOUND_EXT); - } else { - rgb_payment_info_proxy_id_path.set_extension(OUTBOUND_EXT); - rgb_payment_info_path.set_extension(OUTBOUND_EXT); - } - let rgb_payment_info_tmp_path = _append_pending_extension(&rgb_payment_info_path); + let pending_key = format!("{htlc_payment_hash}_pending"); + let namespace = + if inbound { RGB_PAYMENT_INFO_INBOUND_NS } else { RGB_PAYMENT_INFO_OUTBOUND_NS }; - if rgb_payment_info_tmp_path.exists() { - let mut rgb_payment_info = parse_rgb_payment_info(&rgb_payment_info_tmp_path); + if let Ok(data) = kv_store.read(RGB_PRIMARY_NS, namespace, &pending_key) { + let mut rgb_payment_info: RgbPaymentInfo = + bincode::deserialize(&data).expect("valid data"); rgb_payment_info.local_rgb_amount = rgb_info.local_rgb_amount; rgb_payment_info.remote_rgb_amount = rgb_info.remote_rgb_amount; - let serialized_info = - serde_json::to_string(&rgb_payment_info).expect("valid rgb payment info"); - fs::write(&rgb_payment_info_proxy_id_path, serialized_info) - .expect("able to write rgb payment info file"); - fs::remove_file(rgb_payment_info_tmp_path).expect("able to remove file"); + let data = bincode::serialize(&rgb_payment_info).expect("valid rgb payment info"); + kv_store + .execute_batch( + RGB_PRIMARY_NS, + vec![ + KvOp::Write { + secondary_namespace: namespace.to_string(), + key: htlc_proxy_id.clone(), + value: data, + }, + KvOp::Remove { + secondary_namespace: namespace.to_string(), + key: pending_key.clone(), + }, + ], + ) + .expect("able to promote pending payment info"); } - let rgb_payment_info = if rgb_payment_info_proxy_id_path.exists() { - parse_rgb_payment_info(&rgb_payment_info_proxy_id_path) - } else { - let rgb_payment_info = RgbPaymentInfo { - contract_id, - amount: htlc_amount_rgb, - local_rgb_amount: rgb_info.local_rgb_amount, - remote_rgb_amount: rgb_info.remote_rgb_amount, - swap_payment: true, - inbound, + let rgb_payment_info = + if let Ok(data) = kv_store.read(RGB_PRIMARY_NS, namespace, &htlc_proxy_id) { + bincode::deserialize(&data).expect("valid data") + } else { + let rgb_payment_info = RgbPaymentInfo { + contract_id, + amount: htlc_amount_rgb, + local_rgb_amount: rgb_info.local_rgb_amount, + remote_rgb_amount: rgb_info.remote_rgb_amount, + swap_payment: true, + inbound, + }; + let data = bincode::serialize(&rgb_payment_info).expect("valid rgb payment info"); + kv_store + .execute_batch( + RGB_PRIMARY_NS, + vec![ + KvOp::Write { + secondary_namespace: namespace.to_string(), + key: htlc_proxy_id.clone(), + value: data.clone(), + }, + KvOp::Write { + secondary_namespace: namespace.to_string(), + key: htlc_payment_hash.clone(), + value: data, + }, + ], + ) + .expect("able to write rgb payment info"); + rgb_payment_info }; - let serialized_info = - serde_json::to_string(&rgb_payment_info).expect("valid rgb payment info"); - fs::write(rgb_payment_info_proxy_id_path, serialized_info.clone()) - .expect("able to write rgb payment info file"); - fs::write(rgb_payment_info_path, serialized_info) - .expect("able to write rgb payment info file"); - rgb_payment_info - }; if inbound { rgb_received_htlc += rgb_payment_info.amount @@ -344,9 +365,7 @@ where }; let psbt = Psbt::from_unsigned_tx(commitment_tx.clone()).unwrap(); let mut psbt = RgbLibPsbt::from_str(&psbt.to_string()).unwrap(); - let handle = Handle::current(); - let _ = handle.enter(); - let wallet = futures::executor::block_on(_get_rgb_wallet(ldk_data_dir)); + let wallet = _get_rgb_wallet(ldk_data_dir, kv_store); let (fascia, _) = wallet.color_psbt(&mut psbt, coloring_info).unwrap(); let psbt = Psbt::from_str(&psbt.to_string()).unwrap(); let modified_tx = match psbt.extract_tx() { @@ -360,17 +379,15 @@ where wallet.consume_fascia(fascia.clone(), Some(WitnessOrd::Ignored)).unwrap(); - // save RGB transfer data to disk let transfer_info = TransferInfo { contract_id, output_map }; - let transfer_info_path = ldk_data_dir.join(format!("{txid}_transfer_info")); - write_rgb_transfer_info(&transfer_info_path, &transfer_info); + kv_store.write_rgb_transfer_info(&txid.to_string(), &transfer_info); Ok(()) } /// Color HTLC transaction -pub(crate) fn color_htlc( - htlc_tx: &mut Transaction, htlc: &HTLCOutputInCommitment, ldk_data_dir: &Path, +pub(crate) fn color_htlc( + htlc_tx: &mut Transaction, htlc: &HTLCOutputInCommitment, ldk_data_dir: &Path, kv_store: &K, ) -> Result<(), ChannelError> { if htlc.rgb_payment.is_none_or(|(_, a)| a == 0) { return Ok(()); @@ -380,8 +397,7 @@ pub(crate) fn color_htlc( let consignment_htlc_outpoint = htlc_tx.input.first().unwrap().previous_output; let commitment_txid = consignment_htlc_outpoint.txid.to_string(); - let transfer_info_path = ldk_data_dir.join(format!("{commitment_txid}_transfer_info")); - let transfer_info = read_rgb_transfer_info(&transfer_info_path); + let transfer_info = kv_store.read_rgb_transfer_info(&commitment_txid); let contract_id = transfer_info.contract_id; let output_map = HashMap::from([(0, htlc_amount_rgb)]); @@ -396,9 +412,7 @@ pub(crate) fn color_htlc( }; let psbt = Psbt::from_unsigned_tx(htlc_tx.clone()).unwrap(); let mut psbt = RgbLibPsbt::from_str(&psbt.to_string()).unwrap(); - let handle = Handle::current(); - let _ = handle.enter(); - let wallet = futures::executor::block_on(_get_rgb_wallet(ldk_data_dir)); + let wallet = _get_rgb_wallet(ldk_data_dir, kv_store); let (fascia, _) = wallet.color_psbt(&mut psbt, coloring_info).unwrap(); let psbt = Psbt::from_str(&psbt.to_string()).unwrap(); let modified_tx = match psbt.extract_tx() { @@ -410,21 +424,20 @@ pub(crate) fn color_htlc( wallet.consume_fascia(fascia.clone(), Some(WitnessOrd::Ignored)).unwrap(); - // save RGB transfer data to disk let transfer_info = TransferInfo { contract_id, output_map }; - let transfer_info_path = ldk_data_dir.join(format!("{txid}_transfer_info")); - write_rgb_transfer_info(&transfer_info_path, &transfer_info); + kv_store.write_rgb_transfer_info(&txid.to_string(), &transfer_info); Ok(()) } /// Color closing transaction -pub(crate) fn color_closing( +pub(crate) fn color_closing( channel_id: &ChannelId, closing_transaction: &mut ClosingTransaction, ldk_data_dir: &Path, + kv_store: &K, ) -> Result<(), ChannelError> { let closing_tx = closing_transaction.clone().built; - let (rgb_info, _) = get_rgb_channel_info_pending(channel_id, ldk_data_dir); + let rgb_info = get_rgb_channel_info_pending(channel_id, kv_store); let contract_id = rgb_info.contract_id; let holder_vout_amount = rgb_info.local_rgb_amount; @@ -461,9 +474,7 @@ pub(crate) fn color_closing( }; let psbt = Psbt::from_unsigned_tx(closing_tx.clone()).unwrap(); let mut psbt = RgbLibPsbt::from_str(&psbt.to_string()).unwrap(); - let handle = Handle::current(); - let _ = handle.enter(); - let wallet = futures::executor::block_on(_get_rgb_wallet(ldk_data_dir)); + let wallet = _get_rgb_wallet(ldk_data_dir, kv_store); let (fascia, _) = wallet.color_psbt(&mut psbt, coloring_info).unwrap(); let psbt = Psbt::from_str(&psbt.to_string()).unwrap(); let modified_tx = match psbt.extract_tx() { @@ -477,85 +488,40 @@ pub(crate) fn color_closing( wallet.consume_fascia(fascia.clone(), Some(WitnessOrd::Ignored)).unwrap(); - // save RGB transfer data to disk let transfer_info = TransferInfo { contract_id, output_map }; - let transfer_info_path = ldk_data_dir.join(format!("{txid}_transfer_info")); - write_rgb_transfer_info(&transfer_info_path, &transfer_info); + kv_store.write_rgb_transfer_info(&txid.to_string(), &transfer_info); Ok(()) } -/// Get RgbPaymentInfo file path -pub fn get_rgb_payment_info_path( - payment_hash: &PaymentHash, ldk_data_dir: &Path, inbound: bool, -) -> PathBuf { - let mut path = ldk_data_dir.join(payment_hash.0.as_hex().to_string()); - path.set_extension(if inbound { INBOUND_EXT } else { OUTBOUND_EXT }); - path +/// Get RgbInfo from KVStore +pub(crate) fn get_rgb_channel_info( + channel_id: &str, pending: bool, kv_store: &K, +) -> RgbInfo { + kv_store.read_rgb_channel_info(channel_id, pending).expect("channel info must exist in KVStore") } -/// Parse RgbPaymentInfo -pub fn parse_rgb_payment_info(rgb_payment_info_path: &PathBuf) -> RgbPaymentInfo { - let serialized_info = - fs::read_to_string(rgb_payment_info_path).expect("valid rgb payment info"); - serde_json::from_str(&serialized_info).expect("valid rgb info file") +/// Get pending RgbInfo from KVStore +pub fn get_rgb_channel_info_pending( + channel_id: &ChannelId, kv_store: &K, +) -> RgbInfo { + get_rgb_channel_info(&channel_id.0.as_hex().to_string(), true, kv_store) } -/// Get RgbInfo file path -pub fn get_rgb_channel_info_path(channel_id: &str, ldk_data_dir: &Path, pending: bool) -> PathBuf { - let mut info_file_path = ldk_data_dir.join(channel_id); - if pending { - info_file_path.set_extension("pending"); - } - info_file_path +/// Whether the channel has RGB data in KVStore +pub fn is_channel_rgb(channel_id: &ChannelId, kv_store: &K) -> bool { + let channel_id_str = channel_id.0.as_hex().to_string(); + kv_store.read_rgb_channel_info(&channel_id_str, false).is_ok() } -/// Get RgbInfo file -pub(crate) fn get_rgb_channel_info( - channel_id: &str, ldk_data_dir: &Path, pending: bool, -) -> (RgbInfo, PathBuf) { - let info_file_path = get_rgb_channel_info_path(channel_id, ldk_data_dir, pending); - let info = parse_rgb_channel_info(&info_file_path); - (info, info_file_path) -} - -/// Get pending RgbInfo file -pub fn get_rgb_channel_info_pending( - channel_id: &ChannelId, ldk_data_dir: &Path, -) -> (RgbInfo, PathBuf) { - get_rgb_channel_info(&channel_id.0.as_hex().to_string(), ldk_data_dir, true) -} - -/// Parse RgbInfo -pub fn parse_rgb_channel_info(rgb_channel_info_path: &PathBuf) -> RgbInfo { - let serialized_info = fs::read_to_string(rgb_channel_info_path).expect("valid rgb info file"); - serde_json::from_str(&serialized_info).expect("valid rgb info file") -} - -/// Whether the channel data for a channel exist -pub fn is_channel_rgb(channel_id: &ChannelId, ldk_data_dir: &Path) -> bool { - get_rgb_channel_info_path(&channel_id.0.as_hex().to_string(), ldk_data_dir, false).exists() -} - -/// Write RgbInfo file -pub fn write_rgb_channel_info(path: &PathBuf, rgb_info: &RgbInfo) { - let serialized_info = serde_json::to_string(&rgb_info).expect("valid rgb info"); - fs::write(path, serialized_info).expect("able to write") -} - -fn _append_pending_extension(path: &Path) -> PathBuf { - let mut new_path = path.to_path_buf(); - new_path.set_extension(format!("{}_pending", new_path.extension().unwrap().to_string_lossy())); - new_path -} - -/// Write RGB payment info to file -pub fn write_rgb_payment_info_file( - ldk_data_dir: &Path, payment_hash: &PaymentHash, contract_id: ContractId, amount_rgb: u64, - swap_payment: bool, inbound: bool, +/// Write RGB payment info to database +/// +/// Not atomic against concurrent writers of the same payment hash: callers must be externally +/// serialized per key (LDK serializes via `peer_state`, RLN via sequential event handling). +pub fn write_rgb_payment_info( + payment_hash: &PaymentHash, contract_id: ContractId, amount_rgb: u64, swap_payment: bool, + inbound: bool, kv_store: &K, ) { - let rgb_payment_info_path = get_rgb_payment_info_path(payment_hash, ldk_data_dir, inbound); - let rgb_payment_info_tmp_path = _append_pending_extension(&rgb_payment_info_path); let rgb_payment_info = RgbPaymentInfo { contract_id, amount: amount_rgb, @@ -564,30 +530,69 @@ pub fn write_rgb_payment_info_file( swap_payment, inbound, }; - let serialized_info = serde_json::to_string(&rgb_payment_info).expect("valid rgb payment info"); - std::fs::write(rgb_payment_info_path, serialized_info.clone()) - .expect("able to write rgb payment info file"); - std::fs::write(rgb_payment_info_tmp_path, serialized_info) - .expect("able to write rgb payment info tmp file"); + let payment_hash_hex = payment_hash.0.as_hex().to_string(); + let pending_key = format!("{payment_hash_hex}_pending"); + let namespace = + if inbound { RGB_PAYMENT_INFO_INBOUND_NS } else { RGB_PAYMENT_INFO_OUTBOUND_NS }; + let data = bincode::serialize(&rgb_payment_info).expect("valid rgb payment info"); + kv_store + .execute_batch( + RGB_PRIMARY_NS, + vec![ + KvOp::Write { + secondary_namespace: namespace.to_string(), + key: payment_hash_hex, + value: data.clone(), + }, + KvOp::Write { + secondary_namespace: namespace.to_string(), + key: pending_key, + value: data, + }, + ], + ) + .expect("able to write rgb payment info"); } -/// Rename RGB files from temporary to final channel ID -pub(crate) fn rename_rgb_files( - channel_id: &ChannelId, temporary_channel_id: &ChannelId, ldk_data_dir: &Path, +/// update RGB data from temporary to final channel ID in KVStore +pub(crate) fn update_rgb_channel_id( + channel_id: &ChannelId, temporary_channel_id: &ChannelId, kv_store: &K, ) { + if channel_id == temporary_channel_id { + return; + } let temp_chan_id = temporary_channel_id.0.as_hex().to_string(); let chan_id = channel_id.0.as_hex().to_string(); - fs::rename( - get_rgb_channel_info_path(&temp_chan_id, ldk_data_dir, false), - get_rgb_channel_info_path(&chan_id, ldk_data_dir, false), - ) - .expect("rename ok"); - fs::rename( - get_rgb_channel_info_path(&temp_chan_id, ldk_data_dir, true), - get_rgb_channel_info_path(&chan_id, ldk_data_dir, true), - ) - .expect("rename ok"); + let rgb_info = kv_store.read_rgb_channel_info(&temp_chan_id, false).expect("rename ok"); + let rgb_info_pending = kv_store.read_rgb_channel_info(&temp_chan_id, true).expect("rename ok"); + let data = bincode::serialize(&rgb_info).expect("valid rgb channel info"); + let data_pending = bincode::serialize(&rgb_info_pending).expect("valid rgb channel info"); + kv_store + .execute_batch( + RGB_PRIMARY_NS, + vec![ + KvOp::Write { + secondary_namespace: RGB_CHANNEL_INFO_NS.to_string(), + key: chan_id.clone(), + value: data, + }, + KvOp::Remove { + secondary_namespace: RGB_CHANNEL_INFO_NS.to_string(), + key: temp_chan_id.clone(), + }, + KvOp::Write { + secondary_namespace: RGB_CHANNEL_INFO_PENDING_NS.to_string(), + key: chan_id.clone(), + value: data_pending, + }, + KvOp::Remove { + secondary_namespace: RGB_CHANNEL_INFO_PENDING_NS.to_string(), + key: temp_chan_id.clone(), + }, + ], + ) + .expect("rename ok"); } /// Directory holding the media received for a funding, before the contract has vouched for it. @@ -596,14 +601,11 @@ pub fn get_media_staging_dir(ldk_data_dir: &Path, funding_txid: &str) -> PathBuf } /// Handle funding on the receiver side -pub(crate) fn handle_funding( +pub(crate) fn handle_funding( temporary_channel_id: &ChannelId, funding_txid: String, ldk_data_dir: &Path, - push_asset_amount: Option, + push_asset_amount: Option, kv_store: &K, ) -> Result<(), ChannelError> { - let handle = Handle::current(); - let _ = handle.enter(); - let accept_res = - futures::executor::block_on(_accept_transfer(ldk_data_dir, funding_txid.clone())); + let accept_res = _accept_transfer(ldk_data_dir, funding_txid.clone(), kv_store); let (consignment, remote_rgb_assignments, media_digests, media_dir) = match accept_res { Ok(res) => res, Err(RgbLibError::InvalidConsignment) => { @@ -674,37 +676,47 @@ pub(crate) fn handle_funding( counterparty_knows_asset: false, }; let temporary_channel_id_str = temporary_channel_id.0.as_hex().to_string(); - write_rgb_channel_info( - &get_rgb_channel_info_path(&temporary_channel_id_str, ldk_data_dir, true), - &rgb_info, - ); - write_rgb_channel_info( - &get_rgb_channel_info_path(&temporary_channel_id_str, ldk_data_dir, false), - &rgb_info, - ); + + let data = bincode::serialize(&rgb_info).expect("valid rgb channel info"); + kv_store + .execute_batch( + RGB_PRIMARY_NS, + vec![ + KvOp::Write { + secondary_namespace: RGB_CHANNEL_INFO_PENDING_NS.to_string(), + key: temporary_channel_id_str.clone(), + value: data.clone(), + }, + KvOp::Write { + secondary_namespace: RGB_CHANNEL_INFO_NS.to_string(), + key: temporary_channel_id_str.clone(), + value: data, + }, + ], + ) + .expect("KVStore write failed"); Ok(()) } -pub(crate) fn set_counterparty_knows_asset(channel_id: &ChannelId, ldk_data_dir: &Path) { +pub(crate) fn set_counterparty_knows_asset(channel_id: &ChannelId, kv_store: &dyn KVStoreSync) { let channel_id = channel_id.0.as_hex().to_string(); for pending in [true, false] { - let info_file_path = get_rgb_channel_info_path(&channel_id, ldk_data_dir, pending); - if !info_file_path.exists() { - continue; + if let Ok(mut rgb_info) = kv_store.read_rgb_channel_info(&channel_id, pending) { + rgb_info.counterparty_knows_asset = true; + kv_store.write_rgb_channel_info(&channel_id, &rgb_info, pending); } - let mut rgb_info = parse_rgb_channel_info(&info_file_path); - rgb_info.counterparty_knows_asset = true; - write_rgb_channel_info(&info_file_path, &rgb_info); } } -/// Update RGB channel amount -pub fn update_rgb_channel_amount( - channel_id: &str, rgb_offered_htlc: u64, rgb_received_htlc: u64, ldk_data_dir: &Path, - pending: bool, +/// Update RGB channel amount in KVStore +/// +/// Read-modify-write, not atomic at the store level: callers must be externally serialized per +/// channel (LDK serializes via `peer_state`, RLN via sequential event handling). +pub fn update_rgb_channel_amount( + channel_id: &str, rgb_offered_htlc: u64, rgb_received_htlc: u64, pending: bool, kv_store: &K, ) { - let (mut rgb_info, info_file_path) = get_rgb_channel_info(channel_id, ldk_data_dir, pending); + let mut rgb_info = get_rgb_channel_info(channel_id, pending, kv_store); if rgb_offered_htlc > rgb_received_htlc { let spent = rgb_offered_htlc - rgb_received_htlc; @@ -716,45 +728,149 @@ pub fn update_rgb_channel_amount( rgb_info.remote_rgb_amount -= received; } - write_rgb_channel_info(&info_file_path, &rgb_info) + kv_store.write_rgb_channel_info(channel_id, &rgb_info, pending); } /// Update pending RGB channel amount -pub(crate) fn update_rgb_channel_amount_pending( - channel_id: &ChannelId, rgb_offered_htlc: u64, rgb_received_htlc: u64, ldk_data_dir: &Path, +pub(crate) fn update_rgb_channel_amount_pending( + channel_id: &ChannelId, rgb_offered_htlc: u64, rgb_received_htlc: u64, kv_store: &K, ) { update_rgb_channel_amount( &channel_id.0.as_hex().to_string(), rgb_offered_htlc, rgb_received_htlc, - ldk_data_dir, true, + kv_store, ) } -/// Whether the payment is colored -pub(crate) fn is_payment_rgb(ldk_data_dir: &Path, payment_hash: &PaymentHash) -> bool { - get_rgb_payment_info_path(payment_hash, ldk_data_dir, false).exists() - || get_rgb_payment_info_path(payment_hash, ldk_data_dir, true).exists() -} +/// extension trait for RGB-specific KVStore operations +pub trait RgbKvStoreExt { + /// read transfer info from KVStore + fn read_rgb_transfer_info(&self, txid: &str) -> TransferInfo; + /// write transfer info to KVStore + fn write_rgb_transfer_info(&self, txid: &str, info: &TransferInfo); + /// read channel info from KVStore + fn read_rgb_channel_info(&self, channel_id: &str, pending: bool) -> Result; + /// write channel info to KVStore + fn write_rgb_channel_info(&self, channel_id: &str, rgb_info: &RgbInfo, pending: bool); + /// read payment info from KVStore + fn read_rgb_payment_info( + &self, payment_hash: &PaymentHash, inbound: bool, + ) -> Result; + /// write payment info to KVStore + fn write_rgb_payment_info(&self, payment_hash: &PaymentHash, info: &RgbPaymentInfo); + /// remove channel info from KVStore + fn remove_rgb_channel_info(&self, channel_id: &str, pending: bool) -> Result<(), io::Error>; + /// move channel info from one key to another (read + write + remove) + fn update_rgb_channel_info( + &self, old_channel_id: &str, new_channel_id: &str, pending: bool, + ) -> Result<(), io::Error>; + /// whether the payment is colored + fn is_payment_rgb(&self, payment_hash: &PaymentHash) -> bool; + /// filter first hops to only include channels with sufficient RGB assets + fn filter_first_hops( + &self, payment_hash: &PaymentHash, first_hops: &mut Vec, + ) -> (ContractId, u64); + /// read a wallet config value from KVStore + fn read_config(&self, key: &str) -> Result; +} + +impl RgbKvStoreExt for K { + fn read_rgb_transfer_info(&self, txid: &str) -> TransferInfo { + let data = + self.read(RGB_PRIMARY_NS, RGB_TRANSFER_INFO_NS, txid).expect("KVStore read failed"); + bincode::deserialize(&data).expect("valid transfer info") + } -/// Detect the contract ID of the payment and then filter hops based on contract ID and amount -pub(crate) fn filter_first_hops( - ldk_data_dir: &Path, payment_hash: &PaymentHash, first_hops: &mut Vec, -) -> (ContractId, u64) { - let rgb_payment_info_path = get_rgb_payment_info_path(payment_hash, ldk_data_dir, false); - let rgb_payment_info = parse_rgb_payment_info(&rgb_payment_info_path); - let contract_id = rgb_payment_info.contract_id; - let rgb_amount = rgb_payment_info.amount; - first_hops.retain(|h| { - let info_file_path = ldk_data_dir.join(h.channel_id.0.as_hex().to_string()); - if !info_file_path.exists() { - return false; - } - let serialized_info = fs::read_to_string(info_file_path).expect("valid rgb info file"); - let rgb_info: RgbInfo = - serde_json::from_str(&serialized_info).expect("valid rgb info file"); - rgb_info.contract_id == contract_id && rgb_info.local_rgb_amount >= rgb_amount - }); - (contract_id, rgb_amount) + fn write_rgb_transfer_info(&self, txid: &str, info: &TransferInfo) { + let data = bincode::serialize(info).expect("valid transfer info"); + self.write(RGB_PRIMARY_NS, RGB_TRANSFER_INFO_NS, txid, data).expect("KVStore write failed"); + } + + fn read_rgb_channel_info(&self, channel_id: &str, pending: bool) -> Result { + let namespace = if pending { RGB_CHANNEL_INFO_PENDING_NS } else { RGB_CHANNEL_INFO_NS }; + let data = self.read(RGB_PRIMARY_NS, namespace, channel_id)?; + bincode::deserialize(&data).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e)) + } + + fn write_rgb_channel_info(&self, channel_id: &str, rgb_info: &RgbInfo, pending: bool) { + let namespace = if pending { RGB_CHANNEL_INFO_PENDING_NS } else { RGB_CHANNEL_INFO_NS }; + let data = bincode::serialize(rgb_info).expect("valid rgb channel info"); + self.write(RGB_PRIMARY_NS, namespace, channel_id, data).expect("KVStore write failed"); + } + + fn read_rgb_payment_info( + &self, payment_hash: &PaymentHash, inbound: bool, + ) -> Result { + let namespace = + if inbound { RGB_PAYMENT_INFO_INBOUND_NS } else { RGB_PAYMENT_INFO_OUTBOUND_NS }; + let key = payment_hash.0.as_hex().to_string(); + let data = self.read(RGB_PRIMARY_NS, namespace, &key)?; + bincode::deserialize(&data).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e)) + } + + fn write_rgb_payment_info(&self, payment_hash: &PaymentHash, info: &RgbPaymentInfo) { + let namespace = + if info.inbound { RGB_PAYMENT_INFO_INBOUND_NS } else { RGB_PAYMENT_INFO_OUTBOUND_NS }; + let key = payment_hash.0.as_hex().to_string(); + let data = bincode::serialize(info).expect("valid rgb payment info"); + self.write(RGB_PRIMARY_NS, namespace, &key, data).expect("KVStore write failed"); + } + + fn remove_rgb_channel_info(&self, channel_id: &str, pending: bool) -> Result<(), io::Error> { + let namespace = if pending { RGB_CHANNEL_INFO_PENDING_NS } else { RGB_CHANNEL_INFO_NS }; + self.remove(RGB_PRIMARY_NS, namespace, channel_id, false) + } + + fn update_rgb_channel_info( + &self, old_channel_id: &str, new_channel_id: &str, pending: bool, + ) -> Result<(), io::Error> { + let rgb_info = self.read_rgb_channel_info(old_channel_id, pending)?; + let namespace = if pending { RGB_CHANNEL_INFO_PENDING_NS } else { RGB_CHANNEL_INFO_NS }; + let data = bincode::serialize(&rgb_info).expect("valid rgb channel info"); + self.execute_batch( + RGB_PRIMARY_NS, + vec![ + KvOp::Write { + secondary_namespace: namespace.to_string(), + key: new_channel_id.to_string(), + value: data, + }, + KvOp::Remove { + secondary_namespace: namespace.to_string(), + key: old_channel_id.to_string(), + }, + ], + ) + } + + fn is_payment_rgb(&self, payment_hash: &PaymentHash) -> bool { + self.read_rgb_payment_info(payment_hash, false).is_ok() + || self.read_rgb_payment_info(payment_hash, true).is_ok() + } + + fn filter_first_hops( + &self, payment_hash: &PaymentHash, first_hops: &mut Vec, + ) -> (ContractId, u64) { + let rgb_payment_info = + self.read_rgb_payment_info(payment_hash, false).expect("payment info must exist"); + let contract_id = rgb_payment_info.contract_id; + let rgb_amount = rgb_payment_info.amount; + first_hops.retain(|h| { + let channel_id_str = h.channel_id.0.as_hex().to_string(); + match self.read_rgb_channel_info(&channel_id_str, false) { + Ok(rgb_info) => { + rgb_info.contract_id == contract_id && rgb_info.local_rgb_amount >= rgb_amount + }, + Err(_) => false, + } + }); + (contract_id, rgb_amount) + } + + fn read_config(&self, key: &str) -> Result { + let data = self.read(RGB_PRIMARY_NS, RGB_WALLET_CONFIG_NS, key)?; + String::from_utf8(data).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e)) + } } diff --git a/lightning/src/sign/mod.rs b/lightning/src/sign/mod.rs index 58eaeccac..92772c35a 100644 --- a/lightning/src/sign/mod.rs +++ b/lightning/src/sign/mod.rs @@ -38,6 +38,7 @@ use bitcoin::{secp256k1, Psbt, Sequence, Txid, WPubkeyHash, Witness}; use lightning_invoice::RawBolt11Invoice; use std::path::PathBuf; +use std::sync::Arc; use crate::chain::transaction::OutPoint; use crate::crypto::utils::{hkdf_extract_expand_twice, sign, sign_with_aux_rand}; @@ -62,6 +63,7 @@ use crate::rgb_utils::color_htlc; use crate::types::features::ChannelTypeFeatures; use crate::types::payment::PaymentPreimage; use crate::util::async_poll::AsyncResult; +use crate::util::persist::KVStoreSync; use crate::util::ser::{ReadableArgs, Writeable}; use crate::util::transaction_utils; @@ -1022,8 +1024,8 @@ pub trait OutputSpender { #[cfg(taproot)] #[doc(hidden)] #[deprecated(note = "Remove once taproot cfg is removed")] -pub type DynSignerProvider = - dyn SignerProvider; +pub type DynSignerProvider = + dyn SignerProvider, TaprootSigner = InMemorySigner>; /// A dynamic [`SignerProvider`] temporarily needed for doc tests. /// @@ -1031,7 +1033,7 @@ pub type DynSignerProvider = #[cfg(not(taproot))] #[doc(hidden)] #[deprecated(note = "Remove once taproot cfg is removed")] -pub type DynSignerProvider = dyn SignerProvider; +pub type DynSignerProvider = dyn SignerProvider>; /// A trait that can return signer instances for individual channels. pub trait SignerProvider { @@ -1192,7 +1194,7 @@ pub fn compute_funding_key_tweak( /// /// This implementation performs no policy checks and is insufficient by itself as /// a secure external signer. -pub struct InMemorySigner { +pub struct InMemorySigner { /// Holder secret key in the 2-of-2 multisig script of a channel. This key also backs the /// holder's anchor output in a commitment transaction, if one is present. funding_key: sealed::MaybeTweakedSecretKey, @@ -1218,9 +1220,11 @@ pub struct InMemorySigner { entropy_source: RandomBytes, /// The LDK data directory ldk_data_dir: PathBuf, + /// KVStore for RGB data persistence + rgb_kv_store: Arc, } -impl PartialEq for InMemorySigner { +impl PartialEq for InMemorySigner { fn eq(&self, other: &Self) -> bool { self.funding_key == other.funding_key && self.revocation_base_key == other.revocation_base_key @@ -1234,7 +1238,7 @@ impl PartialEq for InMemorySigner { } } -impl Clone for InMemorySigner { +impl Clone for InMemorySigner { fn clone(&self) -> Self { Self { funding_key: self.funding_key.clone(), @@ -1248,18 +1252,20 @@ impl Clone for InMemorySigner { channel_keys_id: self.channel_keys_id, entropy_source: RandomBytes::new(self.get_secure_random_bytes()), ldk_data_dir: self.ldk_data_dir.clone(), + rgb_kv_store: Arc::clone(&self.rgb_kv_store), } } } -impl InMemorySigner { +impl InMemorySigner { #[cfg(any(feature = "_test_utils", test))] pub fn new( funding_key: SecretKey, revocation_base_key: SecretKey, payment_key_v1: SecretKey, payment_key_v2: SecretKey, v2_remote_key_derivation: bool, delayed_payment_base_key: SecretKey, htlc_base_key: SecretKey, commitment_seed: [u8; 32], channel_keys_id: [u8; 32], ldk_data_dir: PathBuf, rand_bytes_unique_start: [u8; 32], - ) -> InMemorySigner { + rgb_kv_store: Arc, + ) -> InMemorySigner { InMemorySigner { funding_key: sealed::MaybeTweakedSecretKey::from(funding_key), revocation_base_key, @@ -1272,6 +1278,7 @@ impl InMemorySigner { channel_keys_id, entropy_source: RandomBytes::new(rand_bytes_unique_start), ldk_data_dir, + rgb_kv_store, } } @@ -1281,7 +1288,8 @@ impl InMemorySigner { payment_key_v2: SecretKey, v2_remote_key_derivation: bool, delayed_payment_base_key: SecretKey, htlc_base_key: SecretKey, commitment_seed: [u8; 32], channel_keys_id: [u8; 32], ldk_data_dir: PathBuf, rand_bytes_unique_start: [u8; 32], - ) -> InMemorySigner { + rgb_kv_store: Arc, + ) -> InMemorySigner { InMemorySigner { funding_key: sealed::MaybeTweakedSecretKey::from(funding_key), revocation_base_key, @@ -1294,6 +1302,7 @@ impl InMemorySigner { channel_keys_id, entropy_source: RandomBytes::new(rand_bytes_unique_start), ldk_data_dir, + rgb_kv_store, } } @@ -1463,13 +1472,13 @@ impl InMemorySigner { } } -impl EntropySource for InMemorySigner { +impl EntropySource for InMemorySigner { fn get_secure_random_bytes(&self) -> [u8; 32] { self.entropy_source.get_secure_random_bytes() } } -impl ChannelSigner for InMemorySigner { +impl ChannelSigner for InMemorySigner { fn get_per_commitment_point( &self, idx: u64, secp_ctx: &Secp256k1, ) -> Result { @@ -1527,7 +1536,7 @@ impl ChannelSigner for InMemorySigner { const MISSING_PARAMS_ERR: &'static str = "ChannelTransactionParameters must be populated before signing operations"; -impl EcdsaChannelSigner for InMemorySigner { +impl EcdsaChannelSigner for InMemorySigner { fn sign_counterparty_commitment( &self, channel_parameters: &ChannelTransactionParameters, commitment_tx: &CommitmentTransaction, _inbound_htlc_preimages: Vec, @@ -1568,7 +1577,9 @@ impl EcdsaChannelSigner for InMemorySigner { &keys.revocation_key, ); if commitment_tx.is_colored() { - if let Err(_e) = color_htlc(&mut htlc_tx, htlc, &self.ldk_data_dir) { + if let Err(_e) = + color_htlc(&mut htlc_tx, htlc, &self.ldk_data_dir, self.rgb_kv_store.as_ref()) + { return Err(()); } } @@ -1897,7 +1908,7 @@ impl EcdsaChannelSigner for InMemorySigner { #[cfg(taproot)] #[allow(unused)] -impl TaprootChannelSigner for InMemorySigner { +impl TaprootChannelSigner for InMemorySigner { fn generate_local_nonce_pair( &self, commitment_number: u64, secp_ctx: &Secp256k1, ) -> PublicNonce { @@ -1967,7 +1978,7 @@ impl TaprootChannelSigner for InMemorySigner { /// /// Note that switching between this struct and [`PhantomKeysManager`] will invalidate any /// previously issued invoices and attempts to pay previous invoices will fail. -pub struct KeysManager { +pub struct KeysManager { secp_ctx: Secp256k1, node_secret: SecretKey, node_id: PublicKey, @@ -1993,9 +2004,10 @@ pub struct KeysManager { starting_time_secs: u64, starting_time_nanos: u32, ldk_data_dir: PathBuf, + rgb_kv_store: Arc, } -impl KeysManager { +impl KeysManager { /// Constructs a [`KeysManager`] from a 32-byte seed. If the seed is in some way biased (e.g., /// your CSRNG is busted) this may panic (but more importantly, you will possibly lose funds). /// `starting_time` isn't strictly required to actually be a time, but it must absolutely, @@ -2020,7 +2032,7 @@ impl KeysManager { /// [`ChannelMonitor`]: crate::chain::channelmonitor::ChannelMonitor pub fn new( seed: &[u8; 32], starting_time_secs: u64, starting_time_nanos: u32, - v2_remote_key_derivation: bool, ldk_data_dir: PathBuf, + v2_remote_key_derivation: bool, ldk_data_dir: PathBuf, rgb_kv_store: Arc, ) -> Self { // Constants for key derivation path indices used in this function. const NODE_SECRET_INDEX: ChildNumber = ChildNumber::Hardened { index: 0 }; @@ -2115,6 +2127,7 @@ impl KeysManager { starting_time_secs, starting_time_nanos, ldk_data_dir, + rgb_kv_store, }; let secp_seed = res.get_secure_random_bytes(); res.secp_ctx.seeded_randomize(&secp_seed); @@ -2175,7 +2188,7 @@ impl KeysManager { } /// Derive an old [`EcdsaChannelSigner`] containing per-channel secrets based on a key derivation parameters. - pub fn derive_channel_keys(&self, params: &[u8; 32]) -> InMemorySigner { + pub fn derive_channel_keys(&self, params: &[u8; 32]) -> InMemorySigner { let chan_id = u64::from_be_bytes(params[0..8].try_into().unwrap()); let mut unique_start = Sha256::engine(); unique_start.input(params); @@ -2234,6 +2247,7 @@ impl KeysManager { params.clone(), self.ldk_data_dir.clone(), prng_seed, + Arc::clone(&self.rgb_kv_store), ) } @@ -2248,7 +2262,7 @@ impl KeysManager { pub fn sign_spendable_outputs_psbt( &self, descriptors: &[&SpendableOutputDescriptor], mut psbt: Psbt, secp_ctx: &Secp256k1, ) -> Result { - let mut keys_cache: Option<(InMemorySigner, [u8; 32])> = None; + let mut keys_cache: Option<(InMemorySigner, [u8; 32])> = None; for outp in descriptors { let get_input_idx = |outpoint: &OutPoint| { psbt.unsigned_tx @@ -2357,13 +2371,13 @@ impl KeysManager { } } -impl EntropySource for KeysManager { +impl EntropySource for KeysManager { fn get_secure_random_bytes(&self) -> [u8; 32] { self.entropy_source.get_secure_random_bytes() } } -impl NodeSigner for KeysManager { +impl NodeSigner for KeysManager { fn get_node_id(&self, recipient: Recipient) -> Result { match recipient { Recipient::Node => Ok(self.node_id.clone()), @@ -2426,7 +2440,7 @@ impl NodeSigner for KeysManager { } } -impl OutputSpender for KeysManager { +impl OutputSpender for KeysManager { /// Creates a [`Transaction`] which spends the given descriptors to the given outputs, plus an /// output to the given change destination (if sufficient change value remains). /// @@ -2465,10 +2479,10 @@ impl OutputSpender for KeysManager { } } -impl SignerProvider for KeysManager { - type EcdsaSigner = InMemorySigner; +impl SignerProvider for KeysManager { + type EcdsaSigner = InMemorySigner; #[cfg(taproot)] - type TaprootSigner = InMemorySigner; + type TaprootSigner = InMemorySigner; fn generate_channel_keys_id(&self, _inbound: bool, user_channel_id: u128) -> [u8; 32] { let child_idx = self.channel_child_index.fetch_add(1, Ordering::AcqRel); @@ -2520,23 +2534,23 @@ impl SignerProvider for KeysManager { // /// Switching between this struct and [`KeysManager`] will invalidate any previously issued /// invoices and attempts to pay previous invoices will fail. -pub struct PhantomKeysManager { +pub struct PhantomKeysManager { #[cfg(test)] - pub(crate) inner: KeysManager, + pub(crate) inner: KeysManager, #[cfg(not(test))] - inner: KeysManager, + inner: KeysManager, inbound_payment_key: ExpandedKey, phantom_secret: SecretKey, phantom_node_id: PublicKey, } -impl EntropySource for PhantomKeysManager { +impl EntropySource for PhantomKeysManager { fn get_secure_random_bytes(&self) -> [u8; 32] { self.inner.get_secure_random_bytes() } } -impl NodeSigner for PhantomKeysManager { +impl NodeSigner for PhantomKeysManager { fn get_node_id(&self, recipient: Recipient) -> Result { match recipient { Recipient::Node => self.inner.get_node_id(Recipient::Node), @@ -2595,7 +2609,7 @@ impl NodeSigner for PhantomKeysManager { } } -impl OutputSpender for PhantomKeysManager { +impl OutputSpender for PhantomKeysManager { /// See [`OutputSpender::spend_spendable_outputs`] and [`KeysManager::spend_spendable_outputs`] /// for documentation on this method. fn spend_spendable_outputs( @@ -2614,10 +2628,10 @@ impl OutputSpender for PhantomKeysManager { } } -impl SignerProvider for PhantomKeysManager { - type EcdsaSigner = InMemorySigner; +impl SignerProvider for PhantomKeysManager { + type EcdsaSigner = InMemorySigner; #[cfg(taproot)] - type TaprootSigner = InMemorySigner; + type TaprootSigner = InMemorySigner; fn generate_channel_keys_id(&self, inbound: bool, user_channel_id: u128) -> [u8; 32] { self.inner.generate_channel_keys_id(inbound, user_channel_id) @@ -2636,7 +2650,7 @@ impl SignerProvider for PhantomKeysManager { } } -impl PhantomKeysManager { +impl PhantomKeysManager { /// Constructs a [`PhantomKeysManager`] given a 32-byte seed and an additional `cross_node_seed` /// that is shared across all nodes that intend to participate in [phantom node payments] /// together. @@ -2651,6 +2665,7 @@ impl PhantomKeysManager { pub fn new( seed: &[u8; 32], starting_time_secs: u64, starting_time_nanos: u32, cross_node_seed: &[u8; 32], v2_remote_key_derivation: bool, ldk_data_dir: PathBuf, + rgb_kv_store: Arc, ) -> Self { let inner = KeysManager::new( seed, @@ -2658,6 +2673,7 @@ impl PhantomKeysManager { starting_time_nanos, v2_remote_key_derivation, ldk_data_dir, + rgb_kv_store, ); let (inbound_key, phantom_key) = hkdf_extract_expand_twice( b"LDK Inbound and Phantom Payment Key Expansion", @@ -2674,7 +2690,7 @@ impl PhantomKeysManager { } /// See [`KeysManager::derive_channel_keys`] for documentation on this method. - pub fn derive_channel_keys(&self, params: &[u8; 32]) -> InMemorySigner { + pub fn derive_channel_keys(&self, params: &[u8; 32]) -> InMemorySigner { self.inner.derive_channel_keys(params) } diff --git a/lightning/src/util/persist.rs b/lightning/src/util/persist.rs index d00e29e68..d67623719 100644 --- a/lightning/src/util/persist.rs +++ b/lightning/src/util/persist.rs @@ -190,6 +190,44 @@ pub trait KVStoreSync { fn list( &self, primary_namespace: &str, secondary_namespace: &str, ) -> Result, io::Error>; + + /// Executes the given operations under `primary_namespace`. + /// + /// The default implementation applies them sequentially; implementations backed by a + /// transactional store should override this to apply the whole batch atomically. + fn execute_batch(&self, primary_namespace: &str, ops: Vec) -> Result<(), io::Error> { + for op in ops { + match op { + KvOp::Write { secondary_namespace, key, value } => { + self.write(primary_namespace, &secondary_namespace, &key, value)? + }, + KvOp::Remove { secondary_namespace, key } => { + self.remove(primary_namespace, &secondary_namespace, &key, false)? + }, + } + } + Ok(()) + } +} + +/// A single operation for [`KVStoreSync::execute_batch`]. +pub enum KvOp { + /// Persists `value` under `secondary_namespace`/`key`. + Write { + /// The secondary namespace to write under. + secondary_namespace: String, + /// The key to write. + key: String, + /// The value to persist. + value: Vec, + }, + /// Removes any data stored under `secondary_namespace`/`key`. + Remove { + /// The secondary namespace to remove from. + secondary_namespace: String, + /// The key to remove. + key: String, + }, } /// A wrapper around a [`KVStoreSync`] that implements the [`KVStore`] trait. It is not necessary to use this type