//! events. The remote server would make use of [`ChainMonitor`] for block processing and for
//! servicing [`ChannelMonitor`] updates from the client.
-use bitcoin::blockdata::block::BlockHeader;
+use bitcoin::blockdata::block::Header;
use bitcoin::hash_types::{Txid, BlockHash};
use crate::chain;
use crate::chain::chaininterface::{BroadcasterInterface, FeeEstimator};
use crate::chain::channelmonitor::{ChannelMonitor, ChannelMonitorUpdate, Balance, MonitorEvent, TransactionOutputs, LATENCY_GRACE_PERIOD_BLOCKS};
use crate::chain::transaction::{OutPoint, TransactionData};
-use crate::sign::WriteableEcdsaChannelSigner;
+use crate::sign::ecdsa::WriteableEcdsaChannelSigner;
use crate::events;
use crate::events::{Event, EventHandler};
use crate::util::atomic_counter::AtomicCounter;
use core::sync::atomic::{AtomicUsize, Ordering};
use bitcoin::secp256k1::PublicKey;
-#[derive(Clone, Copy, Hash, PartialEq, Eq)]
-/// A specific update's ID stored in a `MonitorUpdateId`, separated out to make the contents
-/// entirely opaque.
-enum UpdateOrigin {
- /// An update that was generated by the `ChannelManager` (via our `chain::Watch`
- /// implementation). This corresponds to an actual [`ChannelMonitorUpdate::update_id`] field
- /// and [`ChannelMonitor::get_latest_update_id`].
- OffChain(u64),
- /// An update that was generated during blockchain processing. The ID here is specific to the
- /// generating [`ChainMonitor`] and does *not* correspond to any on-disk IDs.
- ChainSync(u64),
+mod update_origin {
+ #[derive(Debug, Clone, Copy, Hash, PartialEq, Eq)]
+ /// A specific update's ID stored in a `MonitorUpdateId`, separated out to make the contents
+ /// entirely opaque.
+ pub(crate) enum UpdateOrigin {
+ /// An update that was generated by the `ChannelManager` (via our [`crate::chain::Watch`]
+ /// implementation). This corresponds to an actual [ChannelMonitorUpdate::update_id] field
+ /// and [ChannelMonitor::get_latest_update_id].
+ ///
+ /// [ChannelMonitor::get_latest_update_id]: crate::chain::channelmonitor::ChannelMonitor::get_latest_update_id
+ /// [ChannelMonitorUpdate::update_id]: crate::chain::channelmonitor::ChannelMonitorUpdate::update_id
+ OffChain(u64),
+ /// An update that was generated during blockchain processing. The ID here is specific to the
+ /// generating [ChannelMonitor] and does *not* correspond to any on-disk IDs.
+ ///
+ /// [ChannelMonitor]: crate::chain::channelmonitor::ChannelMonitor
+ ChainSync(u64),
+ }
}
+#[cfg(any(feature = "_test_utils", test))]
+pub(crate) use update_origin::UpdateOrigin;
+#[cfg(not(any(feature = "_test_utils", test)))]
+use update_origin::UpdateOrigin;
+
/// An opaque identifier describing a specific [`Persist`] method call.
-#[derive(Clone, Copy, Hash, PartialEq, Eq)]
+#[derive(Debug, Clone, Copy, Hash, PartialEq, Eq)]
pub struct MonitorUpdateId {
- contents: UpdateOrigin,
+ pub(crate) contents: UpdateOrigin,
}
impl MonitorUpdateId {
/// If at some point no further progress can be made towards persisting the pending updates, the
/// node should simply shut down.
///
-/// * If the persistence has failed and cannot be retried further (e.g. because of some timeout),
+/// * If the persistence has failed and cannot be retried further (e.g. because of an outage),
/// [`ChannelMonitorUpdateStatus::UnrecoverableError`] can be used, though this will result in
/// an immediate panic and future operations in LDK generally failing.
///
/// [`ChainMonitor::channel_monitor_updated`] must be called once for *each* update which occurs.
///
/// If at some point no further progress can be made towards persisting a pending update, the node
-/// should simply shut down.
+/// should simply shut down. Until then, the background task should either loop indefinitely, or
+/// persistence should be regularly retried with [`ChainMonitor::list_pending_monitor_updates`]
+/// and [`ChainMonitor::get_monitor`] (note that if a full monitor is persisted all pending
+/// monitor updates may be marked completed).
///
/// # Using remote watchtowers
///
/// updated monitor itself to disk/backups. See the [`Persist`] trait documentation for more
/// details.
///
- /// During blockchain synchronization operations, this may be called with no
- /// [`ChannelMonitorUpdate`], in which case the full [`ChannelMonitor`] needs to be persisted.
+ /// During blockchain synchronization operations, and in some rare cases, this may be called with
+ /// no [`ChannelMonitorUpdate`], in which case the full [`ChannelMonitor`] needs to be persisted.
/// Note that after the full [`ChannelMonitor`] is persisted any previous
/// [`ChannelMonitorUpdate`]s which were persisted should be discarded - they can no longer be
/// applied to the persisted [`ChannelMonitor`] as they were already applied.
/// updated `txdata`.
///
/// Calls which represent a new blockchain tip height should set `best_height`.
- fn process_chain_data<FN>(&self, header: &BlockHeader, best_height: Option<u32>, txdata: &TransactionData, process: FN)
+ fn process_chain_data<FN>(&self, header: &Header, best_height: Option<u32>, txdata: &TransactionData, process: FN)
where
FN: Fn(&ChannelMonitor<ChannelSigner>, &TransactionData) -> Vec<TransactionOutputs>
{
if self.update_monitor_with_chain_data(header, best_height, txdata, &process, funding_outpoint, &monitor_state).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_state);
core::mem::drop(monitor_lock);
let _poison = self.monitors.write().unwrap();
log_error!(self.logger, "{}", err_str);
}
fn update_monitor_with_chain_data<FN>(
- &self, header: &BlockHeader, best_height: Option<u32>, txdata: &TransactionData,
+ &self, header: &Header, best_height: Option<u32>, txdata: &TransactionData,
process: FN, funding_outpoint: &OutPoint, monitor_state: &MonitorHolder<ChannelSigner>
) -> Result<(), ()> where FN: Fn(&ChannelMonitor<ChannelSigner>, &TransactionData) -> Vec<TransactionOutputs> {
let monitor = &monitor_state.monitor;
/// claims which are awaiting confirmation.
///
/// Includes the balances from each [`ChannelMonitor`] *except* those included in
- /// `ignored_channels`.
+ /// `ignored_channels`, allowing you to filter out balances from channels which are still open
+ /// (and whose balance should likely be pulled from the [`ChannelDetails`]).
///
/// See [`ChannelMonitor::get_claimable_balances`] for more details on the exact criteria for
/// inclusion in the return value.
L::Target: Logger,
P::Target: Persist<ChannelSigner>,
{
- fn filtered_block_connected(&self, header: &BlockHeader, txdata: &TransactionData, height: u32) {
+ fn filtered_block_connected(&self, header: &Header, txdata: &TransactionData, height: u32) {
log_debug!(self.logger, "New best block {} at height {} provided via block_connected", header.block_hash(), height);
self.process_chain_data(header, Some(height), &txdata, |monitor, txdata| {
monitor.block_connected(
});
}
- fn block_disconnected(&self, header: &BlockHeader, height: u32) {
+ fn block_disconnected(&self, header: &Header, height: u32) {
let monitor_states = self.monitors.read().unwrap();
log_debug!(self.logger, "Latest block {} at height {} removed via block_disconnected", header.block_hash(), height);
for monitor_state in monitor_states.values() {
L::Target: Logger,
P::Target: Persist<ChannelSigner>,
{
- fn transactions_confirmed(&self, header: &BlockHeader, txdata: &TransactionData, height: u32) {
+ fn transactions_confirmed(&self, header: &Header, txdata: &TransactionData, height: u32) {
log_debug!(self.logger, "{} provided transactions confirmed at height {} in block {}", txdata.len(), height, header.block_hash());
self.process_chain_data(header, None, txdata, |monitor, txdata| {
monitor.transactions_confirmed(
}
}
- fn best_block_updated(&self, header: &BlockHeader, height: u32) {
+ fn best_block_updated(&self, header: &Header, height: u32) {
log_debug!(self.logger, "New best block {} at height {} provided via best_block_updated", header.block_hash(), height);
self.process_chain_data(header, Some(height), &[], |monitor, txdata| {
// While in practice there shouldn't be any recursive calls when given empty txdata,
});
}
- fn get_relevant_txids(&self) -> Vec<(Txid, Option<BlockHash>)> {
+ fn get_relevant_txids(&self) -> Vec<(Txid, u32, Option<BlockHash>)> {
let mut txids = Vec::new();
let monitor_states = self.monitors.read().unwrap();
for monitor_state in monitor_states.values() {
txids.append(&mut monitor_state.monitor.get_relevant_txids());
}
- txids.sort_unstable();
- txids.dedup();
+ txids.sort_unstable_by(|a, b| a.0.cmp(&b.0).then(b.1.cmp(&a.1)));
+ txids.dedup_by_key(|(txid, _, _)| *txid);
txids
}
}
Some(monitor_state) => {
let monitor = &monitor_state.monitor;
log_trace!(self.logger, "Updating ChannelMonitor for channel {}", log_funding_info!(monitor));
- let update_res = monitor.update_monitor(update, &self.broadcaster, &*self.fee_estimator, &self.logger);
- if update_res.is_err() {
- log_error!(self.logger, "Failed to update ChannelMonitor for channel {}.", log_funding_info!(monitor));
- }
- // Even if updating the monitor returns an error, the monitor's state will
- // still be changed. So, persist the updated monitor despite the error.
+ let update_res = monitor.update_monitor(update, &self.broadcaster, &self.fee_estimator, &self.logger);
+
let update_id = MonitorUpdateId::from_monitor_update(update);
let mut pending_monitor_updates = monitor_state.pending_monitor_updates.lock().unwrap();
- let persist_res = self.persister.update_persisted_channel(funding_txo, Some(update), monitor, update_id);
+ let persist_res = if update_res.is_err() {
+ // Even if updating the monitor returns an error, the monitor's state will
+ // still be changed. Therefore, we should persist the updated monitor despite the error.
+ // We don't want to persist a `monitor_update` which results in a failure to apply later
+ // while reading `channel_monitor` with updates from storage. Instead, we should persist
+ // the entire `channel_monitor` here.
+ log_warn!(self.logger, "Failed to update ChannelMonitor for channel {}. Going ahead and persisting the entire ChannelMonitor", log_funding_info!(monitor));
+ self.persister.update_persisted_channel(funding_txo, None, monitor, update_id)
+ } else {
+ self.persister.update_persisted_channel(funding_txo, Some(update), monitor, update_id)
+ };
match persist_res {
ChannelMonitorUpdateStatus::InProgress => {
pending_monitor_updates.push(update_id);
let mut pending_monitor_events = self.pending_monitor_events.lock().unwrap().split_off(0);
for monitor_state in self.monitors.read().unwrap().values() {
let is_pending_monitor_update = monitor_state.has_pending_chainsync_updates(&monitor_state.pending_monitor_updates.lock().unwrap());
- if is_pending_monitor_update &&
- monitor_state.last_chain_persist_height.load(Ordering::Acquire) + LATENCY_GRACE_PERIOD_BLOCKS as usize
- > self.highest_chain_height.load(Ordering::Acquire)
- {
- log_debug!(self.logger, "A Channel Monitor sync is still in progress, refusing to provide monitor events!");
- } else {
+ if !is_pending_monitor_update || monitor_state.last_chain_persist_height.load(Ordering::Acquire) + LATENCY_GRACE_PERIOD_BLOCKS as usize <= self.highest_chain_height.load(Ordering::Acquire) {
if is_pending_monitor_update {
log_error!(self.logger, "A ChannelMonitor sync took longer than {} blocks to complete.", LATENCY_GRACE_PERIOD_BLOCKS);
log_error!(self.logger, " To avoid funds-loss, we are allowing monitor updates to be released.");
projects / rust-lightning / blobdiff