X-Git-Url: http://git.bitcoin.ninja/index.cgi?a=blobdiff_plain;f=lightning%2Fsrc%2Fchain%2Fchainmonitor.rs;h=e6bb9d90778ce46b4cd7ba5f25eafb52daba455d;hb=07d991c82fadf10f59ae01b2b5aef6bc8797fd9a;hp=257f1d0f30bc1db9840b0e2f2da55aa686beef08;hpb=74c9f9b1832fdb2f6e85af12d27cfaab9fe14154;p=rust-lightning diff --git a/lightning/src/chain/chainmonitor.rs b/lightning/src/chain/chainmonitor.rs index 257f1d0f..e6bb9d90 100644 --- a/lightning/src/chain/chainmonitor.rs +++ b/lightning/src/chain/chainmonitor.rs @@ -32,18 +32,19 @@ use crate::chain::chaininterface::{BroadcasterInterface, FeeEstimator}; use crate::chain::channelmonitor::{ChannelMonitor, ChannelMonitorUpdate, Balance, MonitorEvent, TransactionOutputs, WithChannelMonitor}; use crate::chain::transaction::{OutPoint, TransactionData}; use crate::ln::types::ChannelId; -use crate::sign::ecdsa::WriteableEcdsaChannelSigner; +use crate::sign::ecdsa::EcdsaChannelSigner; use crate::events; use crate::events::{Event, EventHandler}; use crate::util::logger::{Logger, WithContext}; use crate::util::errors::APIError; use crate::util::wakers::{Future, Notifier}; -use crate::ln::channelmanager::ChannelDetails; +use crate::ln::channel_state::ChannelDetails; use crate::prelude::*; use crate::sync::{RwLock, RwLockReadGuard, Mutex, MutexGuard}; use core::ops::Deref; use core::sync::atomic::{AtomicUsize, Ordering}; +use bitcoin::hashes::Hash; use bitcoin::secp256k1::PublicKey; /// `Persist` defines behavior for persisting channel monitors: this could mean @@ -101,7 +102,7 @@ use bitcoin::secp256k1::PublicKey; /// /// [`TrustedCommitmentTransaction::revokeable_output_index`]: crate::ln::chan_utils::TrustedCommitmentTransaction::revokeable_output_index /// [`TrustedCommitmentTransaction::build_to_local_justice_tx`]: crate::ln::chan_utils::TrustedCommitmentTransaction::build_to_local_justice_tx -pub trait Persist { +pub trait Persist { /// Persist a new channel's data in response to a [`chain::Watch::watch_channel`] call. This is /// called by [`ChannelManager`] for new channels, or may be called directly, e.g. on startup. /// @@ -148,7 +149,7 @@ pub trait Persist { /// The [`ChannelMonitorUpdate::update_id`] or [`ChannelMonitor::get_latest_update_id`] uniquely /// links this call to [`ChainMonitor::channel_monitor_updated`]. /// For [`Persist::update_persisted_channel`], it is only necessary to call [`ChainMonitor::channel_monitor_updated`] - /// when an [`ChannelMonitorUpdate`] is provided and when you return [`ChannelMonitorUpdateStatus::InProgress`]. + /// when a [`ChannelMonitorUpdate`] is provided and when you return [`ChannelMonitorUpdateStatus::InProgress`]. /// /// See [`Writeable::write`] on [`ChannelMonitor`] for writing out a `ChannelMonitor`, /// [`Writeable::write`] on [`ChannelMonitorUpdate`] for writing out an update, and @@ -163,7 +164,7 @@ pub trait Persist { fn archive_persisted_channel(&self, channel_funding_outpoint: OutPoint); } -struct MonitorHolder { +struct MonitorHolder { monitor: ChannelMonitor, /// The full set of pending monitor updates for this Channel. /// @@ -174,7 +175,7 @@ struct MonitorHolder { pending_monitor_updates: Mutex>, } -impl MonitorHolder { +impl MonitorHolder { fn has_pending_updates(&self, pending_monitor_updates_lock: &MutexGuard>) -> bool { !pending_monitor_updates_lock.is_empty() } @@ -184,12 +185,12 @@ impl MonitorHolder { /// /// Note that this holds a mutex in [`ChainMonitor`] and may block other events until it is /// released. -pub struct LockedChannelMonitor<'a, ChannelSigner: WriteableEcdsaChannelSigner> { +pub struct LockedChannelMonitor<'a, ChannelSigner: EcdsaChannelSigner> { lock: RwLockReadGuard<'a, HashMap>>, funding_txo: OutPoint, } -impl Deref for LockedChannelMonitor<'_, ChannelSigner> { +impl Deref for LockedChannelMonitor<'_, ChannelSigner> { type Target = ChannelMonitor; fn deref(&self) -> &ChannelMonitor { &self.lock.get(&self.funding_txo).expect("Checked at construction").monitor @@ -212,7 +213,7 @@ impl Deref for LockedChannelMonitor< /// [`ChannelManager`]: crate::ln::channelmanager::ChannelManager /// [module-level documentation]: crate::chain::chainmonitor /// [`rebroadcast_pending_claims`]: Self::rebroadcast_pending_claims -pub struct ChainMonitor +pub struct ChainMonitor where C::Target: chain::Filter, T::Target: BroadcasterInterface, F::Target: FeeEstimator, @@ -236,7 +237,7 @@ pub struct ChainMonitor ChainMonitor +impl ChainMonitor where C::Target: chain::Filter, T::Target: BroadcasterInterface, F::Target: FeeEstimator, @@ -260,10 +261,11 @@ where C::Target: chain::Filter, { let err_str = "ChannelMonitor[Update] persistence failed unrecoverably. This indicates we cannot continue normal operation and must shut down."; let funding_outpoints = hash_set_from_iter(self.monitors.read().unwrap().keys().cloned()); + let channel_count = funding_outpoints.len(); for funding_outpoint in funding_outpoints.iter() { let monitor_lock = self.monitors.read().unwrap(); if let Some(monitor_state) = monitor_lock.get(funding_outpoint) { - if self.update_monitor_with_chain_data(header, txdata, &process, funding_outpoint, &monitor_state).is_err() { + if self.update_monitor_with_chain_data(header, best_height, txdata, &process, funding_outpoint, &monitor_state, channel_count).is_err() { // Take the monitors lock for writing so that we poison it and any future // operations going forward fail immediately. core::mem::drop(monitor_lock); @@ -278,7 +280,7 @@ where C::Target: chain::Filter, let monitor_states = self.monitors.write().unwrap(); for (funding_outpoint, monitor_state) in monitor_states.iter() { if !funding_outpoints.contains(funding_outpoint) { - if self.update_monitor_with_chain_data(header, txdata, &process, funding_outpoint, &monitor_state).is_err() { + if self.update_monitor_with_chain_data(header, best_height, txdata, &process, funding_outpoint, &monitor_state, channel_count).is_err() { log_error!(self.logger, "{}", err_str); panic!("{}", err_str); } @@ -297,14 +299,29 @@ where C::Target: chain::Filter, } fn update_monitor_with_chain_data( - &self, header: &Header, txdata: &TransactionData, process: FN, funding_outpoint: &OutPoint, - monitor_state: &MonitorHolder + &self, header: &Header, best_height: Option, txdata: &TransactionData, process: FN, funding_outpoint: &OutPoint, + monitor_state: &MonitorHolder, channel_count: usize, ) -> Result<(), ()> where FN: Fn(&ChannelMonitor, &TransactionData) -> Vec { let monitor = &monitor_state.monitor; - let logger = WithChannelMonitor::from(&self.logger, &monitor); - let mut txn_outputs; - { - txn_outputs = process(monitor, txdata); + let logger = WithChannelMonitor::from(&self.logger, &monitor, None); + + let mut txn_outputs = process(monitor, txdata); + + let get_partition_key = |funding_outpoint: &OutPoint| { + let funding_txid_hash = funding_outpoint.txid.to_raw_hash(); + let funding_txid_hash_bytes = funding_txid_hash.as_byte_array(); + let funding_txid_u32 = u32::from_be_bytes([funding_txid_hash_bytes[0], funding_txid_hash_bytes[1], funding_txid_hash_bytes[2], funding_txid_hash_bytes[3]]); + funding_txid_u32.wrapping_add(best_height.unwrap_or_default()) + }; + + let partition_factor = if channel_count < 15 { + 5 + } else { + 50 // ~ 8hours + }; + + let has_pending_claims = monitor_state.monitor.has_pending_claims(); + if has_pending_claims || get_partition_key(funding_outpoint) % partition_factor == 0 { log_trace!(logger, "Syncing Channel Monitor for channel {}", log_funding_info!(monitor)); match self.persister.update_persisted_channel(*funding_outpoint, None, monitor) { ChannelMonitorUpdateStatus::Completed => @@ -312,11 +329,11 @@ where C::Target: chain::Filter, log_funding_info!(monitor) ), ChannelMonitorUpdateStatus::InProgress => { - log_debug!(logger, "Channel Monitor sync for channel {} in progress.", log_funding_info!(monitor)); - }, + log_trace!(logger, "Channel Monitor sync for channel {} in progress.", log_funding_info!(monitor)); + } ChannelMonitorUpdateStatus::UnrecoverableError => { return Err(()); - }, + } } } @@ -599,7 +616,7 @@ where C::Target: chain::Filter, pub fn archive_fully_resolved_channel_monitors(&self) { let mut have_monitors_to_prune = false; for (_, monitor_holder) in self.monitors.read().unwrap().iter() { - let logger = WithChannelMonitor::from(&self.logger, &monitor_holder.monitor); + let logger = WithChannelMonitor::from(&self.logger, &monitor_holder.monitor, None); if monitor_holder.monitor.is_fully_resolved(&logger) { have_monitors_to_prune = true; } @@ -607,7 +624,7 @@ where C::Target: chain::Filter, if have_monitors_to_prune { let mut monitors = self.monitors.write().unwrap(); monitors.retain(|funding_txo, monitor_holder| { - let logger = WithChannelMonitor::from(&self.logger, &monitor_holder.monitor); + let logger = WithChannelMonitor::from(&self.logger, &monitor_holder.monitor, None); if monitor_holder.monitor.is_fully_resolved(&logger) { log_info!(logger, "Archiving fully resolved ChannelMonitor for funding txo {}", @@ -623,7 +640,7 @@ where C::Target: chain::Filter, } } -impl +impl chain::Listen for ChainMonitor where C::Target: chain::Filter, @@ -652,7 +669,7 @@ where } } -impl +impl chain::Confirm for ChainMonitor where C::Target: chain::Filter, @@ -706,7 +723,7 @@ where } } -impl +impl chain::Watch for ChainMonitor where C::Target: chain::Filter, T::Target: BroadcasterInterface, @@ -715,7 +732,7 @@ where C::Target: chain::Filter, P::Target: Persist, { fn watch_channel(&self, funding_outpoint: OutPoint, monitor: ChannelMonitor) -> Result { - let logger = WithChannelMonitor::from(&self.logger, &monitor); + let logger = WithChannelMonitor::from(&self.logger, &monitor, None); let mut monitors = self.monitors.write().unwrap(); let entry = match monitors.entry(funding_outpoint) { hash_map::Entry::Occupied(_) => { @@ -760,7 +777,7 @@ where C::Target: chain::Filter, let monitors = self.monitors.read().unwrap(); match monitors.get(&funding_txo) { None => { - let logger = WithContext::from(&self.logger, update.counterparty_node_id, Some(channel_id)); + let logger = WithContext::from(&self.logger, update.counterparty_node_id, Some(channel_id), None); log_error!(logger, "Failed to update channel monitor: no such monitor registered"); // We should never ever trigger this from within ChannelManager. Technically a @@ -773,7 +790,7 @@ where C::Target: chain::Filter, }, Some(monitor_state) => { let monitor = &monitor_state.monitor; - let logger = WithChannelMonitor::from(&self.logger, &monitor); + let logger = WithChannelMonitor::from(&self.logger, &monitor, None); log_trace!(logger, "Updating ChannelMonitor to id {} for channel {}", update.update_id, log_funding_info!(monitor)); let update_res = monitor.update_monitor(update, &self.broadcaster, &self.fee_estimator, &self.logger); @@ -841,7 +858,7 @@ where C::Target: chain::Filter, } } -impl events::EventsProvider for ChainMonitor +impl events::EventsProvider for ChainMonitor where C::Target: chain::Filter, T::Target: BroadcasterInterface, F::Target: FeeEstimator, @@ -870,14 +887,17 @@ impl