use bitcoin::hash_types::{BlockHash, WPubkeyHash};
use lightning::chain;
-use lightning::chain::{BestBlock, chainmonitor, channelmonitor, Confirm, Watch};
-use lightning::chain::channelmonitor::{ChannelMonitor, ChannelMonitorUpdateErr, MonitorEvent};
+use lightning::chain::{BestBlock, ChannelMonitorUpdateErr, chainmonitor, channelmonitor, Confirm, Watch};
+use lightning::chain::channelmonitor::{ChannelMonitor, MonitorEvent};
use lightning::chain::transaction::OutPoint;
use lightning::chain::chaininterface::{BroadcasterInterface, ConfirmationTarget, FeeEstimator};
use lightning::chain::keysinterface::{KeysInterface, InMemorySigner};
struct TestChainMonitor {
pub logger: Arc<dyn Logger>,
pub keys: Arc<KeyProvider>,
+ pub persister: Arc<TestPersister>,
pub chain_monitor: Arc<chainmonitor::ChainMonitor<EnforcingSigner, Arc<dyn chain::Filter>, Arc<TestBroadcaster>, Arc<FuzzEstimator>, Arc<dyn Logger>, Arc<TestPersister>>>,
- pub update_ret: Mutex<Result<(), channelmonitor::ChannelMonitorUpdateErr>>,
// If we reload a node with an old copy of ChannelMonitors, the ChannelManager deserialization
// logic will automatically force-close our channels for us (as we don't have an up-to-date
// monitor implying we are not able to punish misbehaving counterparties). Because this test
impl TestChainMonitor {
pub fn new(broadcaster: Arc<TestBroadcaster>, logger: Arc<dyn Logger>, feeest: Arc<FuzzEstimator>, persister: Arc<TestPersister>, keys: Arc<KeyProvider>) -> Self {
Self {
- chain_monitor: Arc::new(chainmonitor::ChainMonitor::new(None, broadcaster, logger.clone(), feeest, persister)),
+ chain_monitor: Arc::new(chainmonitor::ChainMonitor::new(None, broadcaster, logger.clone(), feeest, Arc::clone(&persister))),
logger,
keys,
- update_ret: Mutex::new(Ok(())),
+ persister,
latest_monitors: Mutex::new(HashMap::new()),
should_update_manager: atomic::AtomicBool::new(false),
}
}
}
impl chain::Watch<EnforcingSigner> for TestChainMonitor {
- fn watch_channel(&self, funding_txo: OutPoint, monitor: channelmonitor::ChannelMonitor<EnforcingSigner>) -> Result<(), channelmonitor::ChannelMonitorUpdateErr> {
+ fn watch_channel(&self, funding_txo: OutPoint, monitor: channelmonitor::ChannelMonitor<EnforcingSigner>) -> Result<(), chain::ChannelMonitorUpdateErr> {
let mut ser = VecWriter(Vec::new());
monitor.write(&mut ser).unwrap();
if let Some(_) = self.latest_monitors.lock().unwrap().insert(funding_txo, (monitor.get_latest_update_id(), ser.0)) {
panic!("Already had monitor pre-watch_channel");
}
self.should_update_manager.store(true, atomic::Ordering::Relaxed);
- assert!(self.chain_monitor.watch_channel(funding_txo, monitor).is_ok());
- self.update_ret.lock().unwrap().clone()
+ self.chain_monitor.watch_channel(funding_txo, monitor)
}
- fn update_channel(&self, funding_txo: OutPoint, update: channelmonitor::ChannelMonitorUpdate) -> Result<(), channelmonitor::ChannelMonitorUpdateErr> {
+ fn update_channel(&self, funding_txo: OutPoint, update: channelmonitor::ChannelMonitorUpdate) -> Result<(), chain::ChannelMonitorUpdateErr> {
let mut map_lock = self.latest_monitors.lock().unwrap();
let mut map_entry = match map_lock.entry(funding_txo) {
hash_map::Entry::Occupied(entry) => entry,
deserialized_monitor.write(&mut ser).unwrap();
map_entry.insert((update.update_id, ser.0));
self.should_update_manager.store(true, atomic::Ordering::Relaxed);
- assert!(self.chain_monitor.update_channel(funding_txo, update).is_ok());
- self.update_ret.lock().unwrap().clone()
+ self.chain_monitor.update_channel(funding_txo, update)
}
fn release_pending_monitor_events(&self) -> Vec<MonitorEvent> {
($node_id: expr, $fee_estimator: expr) => { {
let logger: Arc<dyn Logger> = Arc::new(test_logger::TestLogger::new($node_id.to_string(), out.clone()));
let keys_manager = Arc::new(KeyProvider { node_id: $node_id, rand_bytes_id: atomic::AtomicU32::new(0), enforcement_states: Mutex::new(HashMap::new()) });
- let monitor = Arc::new(TestChainMonitor::new(broadcast.clone(), logger.clone(), $fee_estimator.clone(), Arc::new(TestPersister{}), Arc::clone(&keys_manager)));
+ let monitor = Arc::new(TestChainMonitor::new(broadcast.clone(), logger.clone(), $fee_estimator.clone(),
+ Arc::new(TestPersister { update_ret: Mutex::new(Ok(())) }), Arc::clone(&keys_manager)));
let mut config = UserConfig::default();
config.channel_options.forwarding_fee_proportional_millionths = 0;
($ser: expr, $node_id: expr, $old_monitors: expr, $keys_manager: expr, $fee_estimator: expr) => { {
let keys_manager = Arc::clone(& $keys_manager);
let logger: Arc<dyn Logger> = Arc::new(test_logger::TestLogger::new($node_id.to_string(), out.clone()));
- let chain_monitor = Arc::new(TestChainMonitor::new(broadcast.clone(), logger.clone(), $fee_estimator.clone(), Arc::new(TestPersister{}), Arc::clone(& $keys_manager)));
+ let chain_monitor = Arc::new(TestChainMonitor::new(broadcast.clone(), logger.clone(), $fee_estimator.clone(),
+ Arc::new(TestPersister { update_ret: Mutex::new(Ok(())) }), Arc::clone(& $keys_manager)));
let mut config = UserConfig::default();
config.channel_options.forwarding_fee_proportional_millionths = 0;
// bit-twiddling mutations to have similar effects. This is probably overkill, but no
// harm in doing so.
- 0x00 => *monitor_a.update_ret.lock().unwrap() = Err(ChannelMonitorUpdateErr::TemporaryFailure),
- 0x01 => *monitor_b.update_ret.lock().unwrap() = Err(ChannelMonitorUpdateErr::TemporaryFailure),
- 0x02 => *monitor_c.update_ret.lock().unwrap() = Err(ChannelMonitorUpdateErr::TemporaryFailure),
- 0x04 => *monitor_a.update_ret.lock().unwrap() = Ok(()),
- 0x05 => *monitor_b.update_ret.lock().unwrap() = Ok(()),
- 0x06 => *monitor_c.update_ret.lock().unwrap() = Ok(()),
+ 0x00 => *monitor_a.persister.update_ret.lock().unwrap() = Err(ChannelMonitorUpdateErr::TemporaryFailure),
+ 0x01 => *monitor_b.persister.update_ret.lock().unwrap() = Err(ChannelMonitorUpdateErr::TemporaryFailure),
+ 0x02 => *monitor_c.persister.update_ret.lock().unwrap() = Err(ChannelMonitorUpdateErr::TemporaryFailure),
+ 0x04 => *monitor_a.persister.update_ret.lock().unwrap() = Ok(()),
+ 0x05 => *monitor_b.persister.update_ret.lock().unwrap() = Ok(()),
+ 0x06 => *monitor_c.persister.update_ret.lock().unwrap() = Ok(()),
0x08 => {
if let Some((id, _)) = monitor_a.latest_monitors.lock().unwrap().get(&chan_1_funding) {
// after we resolve all pending events.
// First make sure there are no pending monitor updates, resetting the error state
// and calling channel_monitor_updated for each monitor.
- *monitor_a.update_ret.lock().unwrap() = Ok(());
- *monitor_b.update_ret.lock().unwrap() = Ok(());
- *monitor_c.update_ret.lock().unwrap() = Ok(());
+ *monitor_a.persister.update_ret.lock().unwrap() = Ok(());
+ *monitor_b.persister.update_ret.lock().unwrap() = Ok(());
+ *monitor_c.persister.update_ret.lock().unwrap() = Ok(());
if let Some((id, _)) = monitor_a.latest_monitors.lock().unwrap().get(&chan_1_funding) {
nodes[0].channel_monitor_updated(&chan_1_funding, *id);
};
let broadcast = Arc::new(TestBroadcaster{ txn_broadcasted: Mutex::new(Vec::new()) });
- let monitor = Arc::new(chainmonitor::ChainMonitor::new(None, broadcast.clone(), Arc::clone(&logger), fee_est.clone(), Arc::new(TestPersister{})));
+ let monitor = Arc::new(chainmonitor::ChainMonitor::new(None, broadcast.clone(), Arc::clone(&logger), fee_est.clone(),
+ Arc::new(TestPersister { update_ret: Mutex::new(Ok(())) })));
let keys_manager = Arc::new(KeyProvider { node_secret: our_network_key.clone(), counter: AtomicU64::new(0) });
let mut config = UserConfig::default();
-use lightning::chain::channelmonitor;
+use lightning::chain;
+use lightning::chain::{chainmonitor, channelmonitor};
use lightning::chain::transaction::OutPoint;
use lightning::util::enforcing_trait_impls::EnforcingSigner;
-pub struct TestPersister {}
-impl channelmonitor::Persist<EnforcingSigner> for TestPersister {
- fn persist_new_channel(&self, _funding_txo: OutPoint, _data: &channelmonitor::ChannelMonitor<EnforcingSigner>) -> Result<(), channelmonitor::ChannelMonitorUpdateErr> {
- Ok(())
+use std::sync::Mutex;
+
+pub struct TestPersister {
+ pub update_ret: Mutex<Result<(), chain::ChannelMonitorUpdateErr>>,
+}
+impl chainmonitor::Persist<EnforcingSigner> for TestPersister {
+ fn persist_new_channel(&self, _funding_txo: OutPoint, _data: &channelmonitor::ChannelMonitor<EnforcingSigner>) -> Result<(), chain::ChannelMonitorUpdateErr> {
+ self.update_ret.lock().unwrap().clone()
}
- fn update_persisted_channel(&self, _funding_txo: OutPoint, _update: &channelmonitor::ChannelMonitorUpdate, _data: &channelmonitor::ChannelMonitor<EnforcingSigner>) -> Result<(), channelmonitor::ChannelMonitorUpdateErr> {
- Ok(())
+ fn update_persisted_channel(&self, _funding_txo: OutPoint, _update: &channelmonitor::ChannelMonitorUpdate, _data: &channelmonitor::ChannelMonitor<EnforcingSigner>) -> Result<(), chain::ChannelMonitorUpdateErr> {
+ self.update_ret.lock().unwrap().clone()
}
}
use lightning::chain;
use lightning::chain::chaininterface::{BroadcasterInterface, FeeEstimator};
-use lightning::chain::chainmonitor::ChainMonitor;
-use lightning::chain::channelmonitor;
+use lightning::chain::chainmonitor::{ChainMonitor, Persist};
use lightning::chain::keysinterface::{Sign, KeysInterface};
use lightning::ln::channelmanager::ChannelManager;
use lightning::ln::msgs::{ChannelMessageHandler, RoutingMessageHandler};
K::Target: 'static + KeysInterface<Signer = Signer>,
F::Target: 'static + FeeEstimator,
L::Target: 'static + Logger,
- P::Target: 'static + channelmonitor::Persist<Signer>,
+ P::Target: 'static + Persist<Signer>,
CMH::Target: 'static + ChannelMessageHandler,
RMH::Target: 'static + RoutingMessageHandler,
UMH::Target: 'static + CustomMessageHandler,
///
/// use lightning::chain;
/// use lightning::chain::Watch;
+/// use lightning::chain::chainmonitor;
/// use lightning::chain::chainmonitor::ChainMonitor;
-/// use lightning::chain::channelmonitor;
/// use lightning::chain::channelmonitor::ChannelMonitor;
/// use lightning::chain::chaininterface::BroadcasterInterface;
/// use lightning::chain::chaininterface::FeeEstimator;
/// F: FeeEstimator,
/// L: Logger,
/// C: chain::Filter,
-/// P: channelmonitor::Persist<S>,
+/// P: chainmonitor::Persist<S>,
/// >(
/// block_source: &mut B,
/// chain_monitor: &ChainMonitor<S, &C, &T, &F, &L, &P>,
//! type Logger = dyn lightning::util::logger::Logger + Send + Sync;
//! type ChainAccess = dyn lightning::chain::Access + Send + Sync;
//! type ChainFilter = dyn lightning::chain::Filter + Send + Sync;
-//! type DataPersister = dyn lightning::chain::channelmonitor::Persist<lightning::chain::keysinterface::InMemorySigner> + Send + Sync;
+//! type DataPersister = dyn lightning::chain::chainmonitor::Persist<lightning::chain::keysinterface::InMemorySigner> + Send + Sync;
//! type ChainMonitor = lightning::chain::chainmonitor::ChainMonitor<lightning::chain::keysinterface::InMemorySigner, Arc<ChainFilter>, Arc<TxBroadcaster>, Arc<FeeEstimator>, Arc<Logger>, Arc<DataPersister>>;
//! type ChannelManager = Arc<lightning::ln::channelmanager::SimpleArcChannelManager<ChainMonitor, TxBroadcaster, FeeEstimator, Logger>>;
//! type PeerManager = Arc<lightning::ln::peer_handler::SimpleArcPeerManager<lightning_net_tokio::SocketDescriptor, ChainMonitor, TxBroadcaster, FeeEstimator, ChainAccess, Logger>>;
use crate::util::DiskWriteable;
use lightning::chain;
use lightning::chain::chaininterface::{BroadcasterInterface, FeeEstimator};
-use lightning::chain::channelmonitor::{ChannelMonitor, ChannelMonitorUpdate, ChannelMonitorUpdateErr};
-use lightning::chain::channelmonitor;
+use lightning::chain::channelmonitor::{ChannelMonitor, ChannelMonitorUpdate};
+use lightning::chain::chainmonitor;
use lightning::chain::keysinterface::{Sign, KeysInterface};
use lightning::chain::transaction::OutPoint;
use lightning::ln::channelmanager::ChannelManager;
}
}
-impl<ChannelSigner: Sign> channelmonitor::Persist<ChannelSigner> for FilesystemPersister {
- fn persist_new_channel(&self, funding_txo: OutPoint, monitor: &ChannelMonitor<ChannelSigner>) -> Result<(), ChannelMonitorUpdateErr> {
+impl<ChannelSigner: Sign> chainmonitor::Persist<ChannelSigner> for FilesystemPersister {
+ fn persist_new_channel(&self, funding_txo: OutPoint, monitor: &ChannelMonitor<ChannelSigner>) -> Result<(), chain::ChannelMonitorUpdateErr> {
let filename = format!("{}_{}", funding_txo.txid.to_hex(), funding_txo.index);
util::write_to_file(self.path_to_monitor_data(), filename, monitor)
- .map_err(|_| ChannelMonitorUpdateErr::PermanentFailure)
+ .map_err(|_| chain::ChannelMonitorUpdateErr::PermanentFailure)
}
- fn update_persisted_channel(&self, funding_txo: OutPoint, _update: &ChannelMonitorUpdate, monitor: &ChannelMonitor<ChannelSigner>) -> Result<(), ChannelMonitorUpdateErr> {
+ fn update_persisted_channel(&self, funding_txo: OutPoint, _update: &ChannelMonitorUpdate, monitor: &ChannelMonitor<ChannelSigner>) -> Result<(), chain::ChannelMonitorUpdateErr> {
let filename = format!("{}_{}", funding_txo.txid.to_hex(), funding_txo.index);
util::write_to_file(self.path_to_monitor_data(), filename, monitor)
- .map_err(|_| ChannelMonitorUpdateErr::PermanentFailure)
+ .map_err(|_| chain::ChannelMonitorUpdateErr::PermanentFailure)
}
}
use bitcoin::blockdata::block::{Block, BlockHeader};
use bitcoin::hashes::hex::FromHex;
use bitcoin::Txid;
- use lightning::chain::channelmonitor::{Persist, ChannelMonitorUpdateErr};
+ use lightning::chain::ChannelMonitorUpdateErr;
+ use lightning::chain::chainmonitor::Persist;
use lightning::chain::transaction::OutPoint;
use lightning::{check_closed_broadcast, check_closed_event, check_added_monitors};
use lightning::ln::features::InitFeatures;
use bitcoin::hash_types::Txid;
use chain;
-use chain::{Filter, WatchedOutput};
+use chain::{ChannelMonitorUpdateErr, Filter, WatchedOutput};
use chain::chaininterface::{BroadcasterInterface, FeeEstimator};
-use chain::channelmonitor;
-use chain::channelmonitor::{ChannelMonitor, ChannelMonitorUpdate, ChannelMonitorUpdateErr, Balance, MonitorEvent, Persist, TransactionOutputs};
+use chain::channelmonitor::{ChannelMonitor, ChannelMonitorUpdate, Balance, MonitorEvent, TransactionOutputs};
use chain::transaction::{OutPoint, TransactionData};
use chain::keysinterface::Sign;
use util::logger::Logger;
use ln::channelmanager::ChannelDetails;
use prelude::*;
-use sync::RwLock;
+use sync::{RwLock, RwLockReadGuard};
use core::ops::Deref;
+/// `Persist` defines behavior for persisting channel monitors: this could mean
+/// writing once to disk, and/or uploading to one or more backup services.
+///
+/// Note that for every new monitor, you **must** persist the new `ChannelMonitor`
+/// to disk/backups. And, on every update, you **must** persist either the
+/// `ChannelMonitorUpdate` or the updated monitor itself. Otherwise, there is risk
+/// of situations such as revoking a transaction, then crashing before this
+/// revocation can be persisted, then unintentionally broadcasting a revoked
+/// transaction and losing money. This is a risk because previous channel states
+/// are toxic, so it's important that whatever channel state is persisted is
+/// kept up-to-date.
+pub trait Persist<ChannelSigner: Sign> {
+ /// Persist a new channel's data. The data can be stored any way you want, but
+ /// the identifier provided by Rust-Lightning is the channel's outpoint (and
+ /// it is up to you to maintain a correct mapping between the outpoint and the
+ /// stored channel data). Note that you **must** persist every new monitor to
+ /// disk. See the `Persist` trait documentation for more details.
+ ///
+ /// See [`Writeable::write`] on [`ChannelMonitor`] for writing out a `ChannelMonitor`
+ /// and [`ChannelMonitorUpdateErr`] for requirements when returning errors.
+ ///
+ /// [`Writeable::write`]: crate::util::ser::Writeable::write
+ fn persist_new_channel(&self, id: OutPoint, data: &ChannelMonitor<ChannelSigner>) -> Result<(), ChannelMonitorUpdateErr>;
+
+ /// Update one channel's data. The provided `ChannelMonitor` has already
+ /// applied the given update.
+ ///
+ /// Note that on every update, you **must** persist either the
+ /// `ChannelMonitorUpdate` or the updated monitor itself to disk/backups. See
+ /// the `Persist` trait documentation for more details.
+ ///
+ /// If an implementer chooses to persist the updates only, they need to make
+ /// sure that all the updates are applied to the `ChannelMonitors` *before*
+ /// the set of channel monitors is given to the `ChannelManager`
+ /// deserialization routine. See [`ChannelMonitor::update_monitor`] for
+ /// applying a monitor update to a monitor. If full `ChannelMonitors` are
+ /// persisted, then there is no need to persist individual updates.
+ ///
+ /// Note that there could be a performance tradeoff between persisting complete
+ /// channel monitors on every update vs. persisting only updates and applying
+ /// them in batches. The size of each monitor grows `O(number of state updates)`
+ /// whereas updates are small and `O(1)`.
+ ///
+ /// See [`Writeable::write`] on [`ChannelMonitor`] for writing out a `ChannelMonitor`,
+ /// [`Writeable::write`] on [`ChannelMonitorUpdate`] for writing out an update, and
+ /// [`ChannelMonitorUpdateErr`] for requirements when returning errors.
+ ///
+ /// [`Writeable::write`]: crate::util::ser::Writeable::write
+ fn update_persisted_channel(&self, id: OutPoint, update: &ChannelMonitorUpdate, data: &ChannelMonitor<ChannelSigner>) -> Result<(), ChannelMonitorUpdateErr>;
+}
+
+struct MonitorHolder<ChannelSigner: Sign> {
+ monitor: ChannelMonitor<ChannelSigner>,
+}
+
+/// A read-only reference to a current ChannelMonitor.
+///
+/// Note that this holds a mutex in [`ChainMonitor`] and may block other events until it is
+/// released.
+pub struct LockedChannelMonitor<'a, ChannelSigner: Sign> {
+ lock: RwLockReadGuard<'a, HashMap<OutPoint, MonitorHolder<ChannelSigner>>>,
+ funding_txo: OutPoint,
+}
+
+impl<ChannelSigner: Sign> Deref for LockedChannelMonitor<'_, ChannelSigner> {
+ type Target = ChannelMonitor<ChannelSigner>;
+ fn deref(&self) -> &ChannelMonitor<ChannelSigner> {
+ &self.lock.get(&self.funding_txo).expect("Checked at construction").monitor
+ }
+}
+
/// An implementation of [`chain::Watch`] for monitoring channels.
///
/// Connected and disconnected blocks must be provided to `ChainMonitor` as documented by
T::Target: BroadcasterInterface,
F::Target: FeeEstimator,
L::Target: Logger,
- P::Target: channelmonitor::Persist<ChannelSigner>,
+ P::Target: Persist<ChannelSigner>,
{
- /// The monitors
- pub monitors: RwLock<HashMap<OutPoint, ChannelMonitor<ChannelSigner>>>,
+ monitors: RwLock<HashMap<OutPoint, MonitorHolder<ChannelSigner>>>,
chain_source: Option<C>,
broadcaster: T,
logger: L,
T::Target: BroadcasterInterface,
F::Target: FeeEstimator,
L::Target: Logger,
- P::Target: channelmonitor::Persist<ChannelSigner>,
+ P::Target: Persist<ChannelSigner>,
{
/// Dispatches to per-channel monitors, which are responsible for updating their on-chain view
/// of a channel and reacting accordingly based on transactions in the given chain data. See
FN: Fn(&ChannelMonitor<ChannelSigner>, &TransactionData) -> Vec<TransactionOutputs>
{
let mut dependent_txdata = Vec::new();
- let monitors = self.monitors.read().unwrap();
- for monitor in monitors.values() {
- let mut txn_outputs = process(monitor, txdata);
+ let monitor_states = self.monitors.read().unwrap();
+ for monitor_state in monitor_states.values() {
+ let mut txn_outputs = process(&monitor_state.monitor, txdata);
// Register any new outputs with the chain source for filtering, storing any dependent
// transactions from within the block that previously had not been included in txdata.
/// inclusion in the return value.
pub fn get_claimable_balances(&self, ignored_channels: &[&ChannelDetails]) -> Vec<Balance> {
let mut ret = Vec::new();
- let monitors = self.monitors.read().unwrap();
- for (_, monitor) in monitors.iter().filter(|(funding_outpoint, _)| {
+ let monitor_states = self.monitors.read().unwrap();
+ for (_, monitor_state) in monitor_states.iter().filter(|(funding_outpoint, _)| {
for chan in ignored_channels {
if chan.funding_txo.as_ref() == Some(funding_outpoint) {
return false;
}
true
}) {
- ret.append(&mut monitor.get_claimable_balances());
+ ret.append(&mut monitor_state.monitor.get_claimable_balances());
}
ret
}
+ /// Gets the [`LockedChannelMonitor`] for a given funding outpoint, returning an `Err` if no
+ /// such [`ChannelMonitor`] is currently being monitored for.
+ ///
+ /// Note that the result holds a mutex over our monitor set, and should not be held
+ /// indefinitely.
+ pub fn get_monitor(&self, funding_txo: OutPoint) -> Result<LockedChannelMonitor<'_, ChannelSigner>, ()> {
+ let lock = self.monitors.read().unwrap();
+ if lock.get(&funding_txo).is_some() {
+ Ok(LockedChannelMonitor { lock, funding_txo })
+ } else {
+ Err(())
+ }
+ }
+
+ /// Lists the funding outpoint of each [`ChannelMonitor`] being monitored.
+ ///
+ /// Note that [`ChannelMonitor`]s are not removed when a channel is closed as they are always
+ /// monitoring for on-chain state resolutions.
+ pub fn list_monitors(&self) -> Vec<OutPoint> {
+ self.monitors.read().unwrap().keys().map(|outpoint| *outpoint).collect()
+ }
+
+ #[cfg(test)]
+ pub fn remove_monitor(&self, funding_txo: &OutPoint) -> ChannelMonitor<ChannelSigner> {
+ self.monitors.write().unwrap().remove(funding_txo).unwrap().monitor
+ }
+
#[cfg(any(test, feature = "fuzztarget", feature = "_test_utils"))]
pub fn get_and_clear_pending_events(&self) -> Vec<events::Event> {
use util::events::EventsProvider;
T::Target: BroadcasterInterface,
F::Target: FeeEstimator,
L::Target: Logger,
- P::Target: channelmonitor::Persist<ChannelSigner>,
+ P::Target: Persist<ChannelSigner>,
{
fn block_connected(&self, block: &Block, height: u32) {
let header = &block.header;
}
fn block_disconnected(&self, header: &BlockHeader, height: u32) {
- let monitors = self.monitors.read().unwrap();
+ 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 in monitors.values() {
- monitor.block_disconnected(
+ for monitor_state in monitor_states.values() {
+ monitor_state.monitor.block_disconnected(
header, height, &*self.broadcaster, &*self.fee_estimator, &*self.logger);
}
}
T::Target: BroadcasterInterface,
F::Target: FeeEstimator,
L::Target: Logger,
- P::Target: channelmonitor::Persist<ChannelSigner>,
+ P::Target: Persist<ChannelSigner>,
{
fn transactions_confirmed(&self, header: &BlockHeader, txdata: &TransactionData, height: u32) {
log_debug!(self.logger, "{} provided transactions confirmed at height {} in block {}", txdata.len(), height, header.block_hash());
fn transaction_unconfirmed(&self, txid: &Txid) {
log_debug!(self.logger, "Transaction {} reorganized out of chain", txid);
- let monitors = self.monitors.read().unwrap();
- for monitor in monitors.values() {
- monitor.transaction_unconfirmed(txid, &*self.broadcaster, &*self.fee_estimator, &*self.logger);
+ let monitor_states = self.monitors.read().unwrap();
+ for monitor_state in monitor_states.values() {
+ monitor_state.monitor.transaction_unconfirmed(txid, &*self.broadcaster, &*self.fee_estimator, &*self.logger);
}
}
fn get_relevant_txids(&self) -> Vec<Txid> {
let mut txids = Vec::new();
- let monitors = self.monitors.read().unwrap();
- for monitor in monitors.values() {
- txids.append(&mut monitor.get_relevant_txids());
+ 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();
T::Target: BroadcasterInterface,
F::Target: FeeEstimator,
L::Target: Logger,
- P::Target: channelmonitor::Persist<ChannelSigner>,
+ P::Target: Persist<ChannelSigner>,
{
/// Adds the monitor that watches the channel referred to by the given outpoint.
///
return Err(ChannelMonitorUpdateErr::PermanentFailure)},
hash_map::Entry::Vacant(e) => e,
};
- if let Err(e) = self.persister.persist_new_channel(funding_outpoint, &monitor) {
- log_error!(self.logger, "Failed to persist new channel data");
- return Err(e);
+ let persist_res = self.persister.persist_new_channel(funding_outpoint, &monitor);
+ if persist_res.is_err() {
+ log_error!(self.logger, "Failed to persist new channel data: {:?}", persist_res);
+ }
+ if persist_res == Err(ChannelMonitorUpdateErr::PermanentFailure) {
+ return persist_res;
}
{
let funding_txo = monitor.get_funding_txo();
monitor.load_outputs_to_watch(chain_source);
}
}
- entry.insert(monitor);
- Ok(())
+ entry.insert(MonitorHolder { monitor });
+ persist_res
}
/// Note that we persist the given `ChannelMonitor` update while holding the
#[cfg(not(any(test, feature = "fuzztarget")))]
Err(ChannelMonitorUpdateErr::PermanentFailure)
},
- Some(monitor) => {
+ Some(monitor_state) => {
+ let monitor = &monitor_state.monitor;
log_trace!(self.logger, "Updating Channel Monitor for channel {}", log_funding_info!(monitor));
let update_res = monitor.update_monitor(&update, &self.broadcaster, &self.fee_estimator, &self.logger);
if let Err(e) = &update_res {
fn release_pending_monitor_events(&self) -> Vec<MonitorEvent> {
let mut pending_monitor_events = Vec::new();
- for monitor in self.monitors.read().unwrap().values() {
- pending_monitor_events.append(&mut monitor.get_and_clear_pending_monitor_events());
+ for monitor_state in self.monitors.read().unwrap().values() {
+ pending_monitor_events.append(&mut monitor_state.monitor.get_and_clear_pending_monitor_events());
}
pending_monitor_events
}
T::Target: BroadcasterInterface,
F::Target: FeeEstimator,
L::Target: Logger,
- P::Target: channelmonitor::Persist<ChannelSigner>,
+ P::Target: Persist<ChannelSigner>,
{
/// Processes [`SpendableOutputs`] events produced from each [`ChannelMonitor`] upon maturity.
///
/// [`SpendableOutputs`]: events::Event::SpendableOutputs
fn process_pending_events<H: Deref>(&self, handler: H) where H::Target: EventHandler {
let mut pending_events = Vec::new();
- for monitor in self.monitors.read().unwrap().values() {
- pending_events.append(&mut monitor.get_and_clear_pending_events());
+ for monitor_state in self.monitors.read().unwrap().values() {
+ pending_events.append(&mut monitor_state.monitor.get_and_clear_pending_events());
}
for event in pending_events.drain(..) {
handler.handle_event(&event);
}
}
-/// An error enum representing a failure to persist a channel monitor update.
-#[derive(Clone, Copy, Debug, PartialEq)]
-pub enum ChannelMonitorUpdateErr {
- /// Used to indicate a temporary failure (eg connection to a watchtower or remote backup of
- /// our state failed, but is expected to succeed at some point in the future).
- ///
- /// Such a failure will "freeze" a channel, preventing us from revoking old states or
- /// submitting new commitment transactions to the counterparty. Once the update(s) which failed
- /// have been successfully applied, ChannelManager::channel_monitor_updated can be used to
- /// restore the channel to an operational state.
- ///
- /// Note that a given ChannelManager will *never* re-generate a given ChannelMonitorUpdate. If
- /// you return a TemporaryFailure you must ensure that it is written to disk safely before
- /// writing out the latest ChannelManager state.
- ///
- /// Even when a channel has been "frozen" updates to the ChannelMonitor can continue to occur
- /// (eg if an inbound HTLC which we forwarded was claimed upstream resulting in us attempting
- /// to claim it on this channel) and those updates must be applied wherever they can be. At
- /// least one such updated ChannelMonitor must be persisted otherwise PermanentFailure should
- /// be returned to get things on-chain ASAP using only the in-memory copy. Obviously updates to
- /// the channel which would invalidate previous ChannelMonitors are not made when a channel has
- /// been "frozen".
- ///
- /// Note that even if updates made after TemporaryFailure succeed you must still call
- /// channel_monitor_updated to ensure you have the latest monitor and re-enable normal channel
- /// operation.
- ///
- /// Note that the update being processed here will not be replayed for you when you call
- /// ChannelManager::channel_monitor_updated, so you must store the update itself along
- /// with the persisted ChannelMonitor on your own local disk prior to returning a
- /// TemporaryFailure. You may, of course, employ a journaling approach, storing only the
- /// ChannelMonitorUpdate on disk without updating the monitor itself, replaying the journal at
- /// reload-time.
- ///
- /// For deployments where a copy of ChannelMonitors and other local state are backed up in a
- /// remote location (with local copies persisted immediately), it is anticipated that all
- /// updates will return TemporaryFailure until the remote copies could be updated.
- TemporaryFailure,
- /// Used to indicate no further channel monitor updates will be allowed (eg we've moved on to a
- /// different watchtower and cannot update with all watchtowers that were previously informed
- /// of this channel).
- ///
- /// At reception of this error, ChannelManager will force-close the channel and return at
- /// least a final ChannelMonitorUpdate::ChannelForceClosed which must be delivered to at
- /// least one ChannelMonitor copy. Revocation secret MUST NOT be released and offchain channel
- /// update must be rejected.
- ///
- /// This failure may also signal a failure to update the local persisted copy of one of
- /// the channel monitor instance.
- ///
- /// Note that even when you fail a holder commitment transaction update, you must store the
- /// update to ensure you can claim from it in case of a duplicate copy of this ChannelMonitor
- /// broadcasts it (e.g distributed channel-monitor deployment)
- ///
- /// In case of distributed watchtowers deployment, the new version must be written to disk, as
- /// state may have been stored but rejected due to a block forcing a commitment broadcast. This
- /// storage is used to claim outputs of rejected state confirmed onchain by another watchtower,
- /// lagging behind on block processing.
- PermanentFailure,
-}
-
/// General Err type for ChannelMonitor actions. Generally, this implies that the data provided is
/// inconsistent with the ChannelMonitor being called. eg for ChannelMonitor::update_monitor this
/// means you tried to update a monitor for a different channel or the ChannelMonitorUpdate was
}
}
-/// `Persist` defines behavior for persisting channel monitors: this could mean
-/// writing once to disk, and/or uploading to one or more backup services.
-///
-/// Note that for every new monitor, you **must** persist the new `ChannelMonitor`
-/// to disk/backups. And, on every update, you **must** persist either the
-/// `ChannelMonitorUpdate` or the updated monitor itself. Otherwise, there is risk
-/// of situations such as revoking a transaction, then crashing before this
-/// revocation can be persisted, then unintentionally broadcasting a revoked
-/// transaction and losing money. This is a risk because previous channel states
-/// are toxic, so it's important that whatever channel state is persisted is
-/// kept up-to-date.
-pub trait Persist<ChannelSigner: Sign> {
- /// Persist a new channel's data. The data can be stored any way you want, but
- /// the identifier provided by Rust-Lightning is the channel's outpoint (and
- /// it is up to you to maintain a correct mapping between the outpoint and the
- /// stored channel data). Note that you **must** persist every new monitor to
- /// disk. See the `Persist` trait documentation for more details.
- ///
- /// See [`ChannelMonitor::write`] for writing out a `ChannelMonitor`,
- /// and [`ChannelMonitorUpdateErr`] for requirements when returning errors.
- fn persist_new_channel(&self, id: OutPoint, data: &ChannelMonitor<ChannelSigner>) -> Result<(), ChannelMonitorUpdateErr>;
-
- /// Update one channel's data. The provided `ChannelMonitor` has already
- /// applied the given update.
- ///
- /// Note that on every update, you **must** persist either the
- /// `ChannelMonitorUpdate` or the updated monitor itself to disk/backups. See
- /// the `Persist` trait documentation for more details.
- ///
- /// If an implementer chooses to persist the updates only, they need to make
- /// sure that all the updates are applied to the `ChannelMonitors` *before*
- /// the set of channel monitors is given to the `ChannelManager`
- /// deserialization routine. See [`ChannelMonitor::update_monitor`] for
- /// applying a monitor update to a monitor. If full `ChannelMonitors` are
- /// persisted, then there is no need to persist individual updates.
- ///
- /// Note that there could be a performance tradeoff between persisting complete
- /// channel monitors on every update vs. persisting only updates and applying
- /// them in batches. The size of each monitor grows `O(number of state updates)`
- /// whereas updates are small and `O(1)`.
- ///
- /// See [`ChannelMonitor::write`] for writing out a `ChannelMonitor`,
- /// [`ChannelMonitorUpdate::write`] for writing out an update, and
- /// [`ChannelMonitorUpdateErr`] for requirements when returning errors.
- fn update_persisted_channel(&self, id: OutPoint, update: &ChannelMonitorUpdate, data: &ChannelMonitor<ChannelSigner>) -> Result<(), ChannelMonitorUpdateErr>;
-}
-
impl<Signer: Sign, T: Deref, F: Deref, L: Deref> chain::Listen for (ChannelMonitor<Signer>, T, F, L)
where
T::Target: BroadcasterInterface,
use bitcoin::hash_types::{BlockHash, Txid};
use bitcoin::network::constants::Network;
-use chain::channelmonitor::{ChannelMonitor, ChannelMonitorUpdate, ChannelMonitorUpdateErr, MonitorEvent};
+use chain::channelmonitor::{ChannelMonitor, ChannelMonitorUpdate, MonitorEvent};
use chain::keysinterface::Sign;
use chain::transaction::{OutPoint, TransactionData};
fn get_relevant_txids(&self) -> Vec<Txid>;
}
+/// An error enum representing a failure to persist a channel monitor update.
+#[derive(Clone, Copy, Debug, PartialEq)]
+pub enum ChannelMonitorUpdateErr {
+ /// Used to indicate a temporary failure (eg connection to a watchtower or remote backup of
+ /// our state failed, but is expected to succeed at some point in the future).
+ ///
+ /// Such a failure will "freeze" a channel, preventing us from revoking old states or
+ /// submitting new commitment transactions to the counterparty. Once the update(s) which failed
+ /// have been successfully applied, ChannelManager::channel_monitor_updated can be used to
+ /// restore the channel to an operational state.
+ ///
+ /// Note that a given ChannelManager will *never* re-generate a given ChannelMonitorUpdate. If
+ /// you return a TemporaryFailure you must ensure that it is written to disk safely before
+ /// writing out the latest ChannelManager state.
+ ///
+ /// Even when a channel has been "frozen" updates to the ChannelMonitor can continue to occur
+ /// (eg if an inbound HTLC which we forwarded was claimed upstream resulting in us attempting
+ /// to claim it on this channel) and those updates must be applied wherever they can be. At
+ /// least one such updated ChannelMonitor must be persisted otherwise PermanentFailure should
+ /// be returned to get things on-chain ASAP using only the in-memory copy. Obviously updates to
+ /// the channel which would invalidate previous ChannelMonitors are not made when a channel has
+ /// been "frozen".
+ ///
+ /// Note that even if updates made after TemporaryFailure succeed you must still call
+ /// channel_monitor_updated to ensure you have the latest monitor and re-enable normal channel
+ /// operation.
+ ///
+ /// Note that the update being processed here will not be replayed for you when you call
+ /// ChannelManager::channel_monitor_updated, so you must store the update itself along
+ /// with the persisted ChannelMonitor on your own local disk prior to returning a
+ /// TemporaryFailure. You may, of course, employ a journaling approach, storing only the
+ /// ChannelMonitorUpdate on disk without updating the monitor itself, replaying the journal at
+ /// reload-time.
+ ///
+ /// For deployments where a copy of ChannelMonitors and other local state are backed up in a
+ /// remote location (with local copies persisted immediately), it is anticipated that all
+ /// updates will return TemporaryFailure until the remote copies could be updated.
+ TemporaryFailure,
+ /// Used to indicate no further channel monitor updates will be allowed (eg we've moved on to a
+ /// different watchtower and cannot update with all watchtowers that were previously informed
+ /// of this channel).
+ ///
+ /// At reception of this error, ChannelManager will force-close the channel and return at
+ /// least a final ChannelMonitorUpdate::ChannelForceClosed which must be delivered to at
+ /// least one ChannelMonitor copy. Revocation secret MUST NOT be released and offchain channel
+ /// update must be rejected.
+ ///
+ /// This failure may also signal a failure to update the local persisted copy of one of
+ /// the channel monitor instance.
+ ///
+ /// Note that even when you fail a holder commitment transaction update, you must store the
+ /// update to ensure you can claim from it in case of a duplicate copy of this ChannelMonitor
+ /// broadcasts it (e.g distributed channel-monitor deployment)
+ ///
+ /// In case of distributed watchtowers deployment, the new version must be written to disk, as
+ /// state may have been stored but rejected due to a block forcing a commitment broadcast. This
+ /// storage is used to claim outputs of rejected state confirmed onchain by another watchtower,
+ /// lagging behind on block processing.
+ PermanentFailure,
+}
+
/// The `Watch` trait defines behavior for watching on-chain activity pertaining to channels as
/// blocks are connected and disconnected.
///
/// funds in the channel. See [`ChannelMonitorUpdateErr`] for more details about how to handle
/// multiple instances.
///
-/// [`ChannelMonitor`]: channelmonitor::ChannelMonitor
-/// [`ChannelMonitorUpdateErr`]: channelmonitor::ChannelMonitorUpdateErr
-/// [`PermanentFailure`]: channelmonitor::ChannelMonitorUpdateErr::PermanentFailure
+/// [`PermanentFailure`]: ChannelMonitorUpdateErr::PermanentFailure
pub trait Watch<ChannelSigner: Sign> {
/// Watches a channel identified by `funding_txo` using `monitor`.
///
/// [`ChannelMonitorUpdateErr`] for invariants around returning an error.
///
/// [`update_monitor`]: channelmonitor::ChannelMonitor::update_monitor
- /// [`ChannelMonitorUpdateErr`]: channelmonitor::ChannelMonitorUpdateErr
fn update_channel(&self, funding_txo: OutPoint, update: ChannelMonitorUpdate) -> Result<(), ChannelMonitorUpdateErr>;
/// Returns any monitor events since the last call. Subsequent calls must only return new
/// processed later. Then, in order to block until the data has been processed, any [`Watch`]
/// invocation that has called the `Filter` must return [`TemporaryFailure`].
///
-/// [`TemporaryFailure`]: channelmonitor::ChannelMonitorUpdateErr::TemporaryFailure
+/// [`TemporaryFailure`]: ChannelMonitorUpdateErr::TemporaryFailure
/// [BIP 157]: https://github.com/bitcoin/bips/blob/master/bip-0157.mediawiki
/// [BIP 158]: https://github.com/bitcoin/bips/blob/master/bip-0158.mediawiki
pub trait Filter {
use bitcoin::blockdata::constants::genesis_block;
use bitcoin::hash_types::BlockHash;
use bitcoin::network::constants::Network;
-use chain::channelmonitor::{ChannelMonitor, ChannelMonitorUpdateErr};
+use chain::channelmonitor::ChannelMonitor;
use chain::transaction::OutPoint;
-use chain::Listen;
-use chain::Watch;
+use chain::{ChannelMonitorUpdateErr, Listen, Watch};
use ln::{PaymentPreimage, PaymentHash};
use ln::channelmanager::{ChannelManager, ChannelManagerReadArgs, RAACommitmentOrder, PaymentSendFailure};
use ln::features::InitFeatures;
use prelude::*;
use sync::{Arc, Mutex};
-// If persister_fail is true, we have the persister return a PermanentFailure
-// instead of the higher-level ChainMonitor.
-fn do_test_simple_monitor_permanent_update_fail(persister_fail: bool) {
+#[test]
+fn test_simple_monitor_permanent_update_fail() {
// Test that we handle a simple permanent monitor update failure
- let mut chanmon_cfgs = create_chanmon_cfgs(2);
+ let chanmon_cfgs = create_chanmon_cfgs(2);
let node_cfgs = create_node_cfgs(2, &chanmon_cfgs);
let node_chanmgrs = create_node_chanmgrs(2, &node_cfgs, &[None, None]);
let mut nodes = create_network(2, &node_cfgs, &node_chanmgrs);
create_announced_chan_between_nodes(&nodes, 0, 1, InitFeatures::known(), InitFeatures::known());
let (route, payment_hash_1, _, payment_secret_1) = get_route_and_payment_hash!(&nodes[0], nodes[1], 1000000);
-
- match persister_fail {
- true => chanmon_cfgs[0].persister.set_update_ret(Err(ChannelMonitorUpdateErr::PermanentFailure)),
- false => *nodes[0].chain_monitor.update_ret.lock().unwrap() = Some(Err(ChannelMonitorUpdateErr::PermanentFailure))
- }
+ chanmon_cfgs[0].persister.set_update_ret(Err(ChannelMonitorUpdateErr::PermanentFailure));
unwrap_send_err!(nodes[0].node.send_payment(&route, payment_hash_1, &Some(payment_secret_1)), true, APIError::ChannelUnavailable {..}, {});
check_added_monitors!(nodes[0], 2);
blocks: Arc::new(Mutex::new(vec![(genesis_block(Network::Testnet).header, 200); 200])),
};
let chain_mon = {
- let monitors = nodes[0].chain_monitor.chain_monitor.monitors.read().unwrap();
- let monitor = monitors.get(&outpoint).unwrap();
+ let monitor = nodes[0].chain_monitor.chain_monitor.get_monitor(outpoint).unwrap();
let mut w = test_utils::TestVecWriter(Vec::new());
monitor.write(&mut w).unwrap();
let new_monitor = <(BlockHash, ChannelMonitor<EnforcingSigner>)>::read(
assert_eq!(events.len(), 1);
}
-#[test]
-fn test_simple_monitor_permanent_update_fail() {
- do_test_simple_monitor_permanent_update_fail(false);
-
- // Test behavior when the persister returns a PermanentFailure.
- do_test_simple_monitor_permanent_update_fail(true);
-}
-
-// If persister_fail is true, we have the persister return a TemporaryFailure instead of the
-// higher-level ChainMonitor.
-fn do_test_simple_monitor_temporary_update_fail(disconnect: bool, persister_fail: bool) {
+fn do_test_simple_monitor_temporary_update_fail(disconnect: bool) {
// Test that we can recover from a simple temporary monitor update failure optionally with
// a disconnect in between
- let mut chanmon_cfgs = create_chanmon_cfgs(2);
+ let chanmon_cfgs = create_chanmon_cfgs(2);
let node_cfgs = create_node_cfgs(2, &chanmon_cfgs);
let node_chanmgrs = create_node_chanmgrs(2, &node_cfgs, &[None, None]);
let mut nodes = create_network(2, &node_cfgs, &node_chanmgrs);
let (route, payment_hash_1, payment_preimage_1, payment_secret_1) = get_route_and_payment_hash!(&nodes[0], nodes[1], 1000000);
- match persister_fail {
- true => chanmon_cfgs[0].persister.set_update_ret(Err(ChannelMonitorUpdateErr::TemporaryFailure)),
- false => *nodes[0].chain_monitor.update_ret.lock().unwrap() = Some(Err(ChannelMonitorUpdateErr::TemporaryFailure))
- }
+ chanmon_cfgs[0].persister.set_update_ret(Err(ChannelMonitorUpdateErr::TemporaryFailure));
{
unwrap_send_err!(nodes[0].node.send_payment(&route, payment_hash_1, &Some(payment_secret_1)), false, APIError::MonitorUpdateFailed, {});
reconnect_nodes(&nodes[0], &nodes[1], (true, true), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (false, false));
}
- match persister_fail {
- true => chanmon_cfgs[0].persister.set_update_ret(Ok(())),
- false => *nodes[0].chain_monitor.update_ret.lock().unwrap() = Some(Ok(()))
- }
+ chanmon_cfgs[0].persister.set_update_ret(Ok(()));
let (outpoint, latest_update) = nodes[0].chain_monitor.latest_monitor_update_id.lock().unwrap().get(&channel_id).unwrap().clone();
nodes[0].node.channel_monitor_updated(&outpoint, latest_update);
check_added_monitors!(nodes[0], 0);
// Now set it to failed again...
let (route, payment_hash_2, _, payment_secret_2) = get_route_and_payment_hash!(&nodes[0], nodes[1], 1000000);
{
- match persister_fail {
- true => chanmon_cfgs[0].persister.set_update_ret(Err(ChannelMonitorUpdateErr::TemporaryFailure)),
- false => *nodes[0].chain_monitor.update_ret.lock().unwrap() = Some(Err(ChannelMonitorUpdateErr::TemporaryFailure))
- }
+ chanmon_cfgs[0].persister.set_update_ret(Err(ChannelMonitorUpdateErr::TemporaryFailure));
unwrap_send_err!(nodes[0].node.send_payment(&route, payment_hash_2, &Some(payment_secret_2)), false, APIError::MonitorUpdateFailed, {});
check_added_monitors!(nodes[0], 1);
}
#[test]
fn test_simple_monitor_temporary_update_fail() {
- do_test_simple_monitor_temporary_update_fail(false, false);
- do_test_simple_monitor_temporary_update_fail(true, false);
-
- // Test behavior when the persister returns a TemporaryFailure.
- do_test_simple_monitor_temporary_update_fail(false, true);
- do_test_simple_monitor_temporary_update_fail(true, true);
+ do_test_simple_monitor_temporary_update_fail(false);
+ do_test_simple_monitor_temporary_update_fail(true);
}
fn do_test_monitor_temporary_update_fail(disconnect_count: usize) {
// Now try to send a second payment which will fail to send
let (route, payment_hash_2, payment_preimage_2, payment_secret_2) = get_route_and_payment_hash!(nodes[0], nodes[1], 1000000);
{
- *nodes[0].chain_monitor.update_ret.lock().unwrap() = Some(Err(ChannelMonitorUpdateErr::TemporaryFailure));
+ chanmon_cfgs[0].persister.set_update_ret(Err(ChannelMonitorUpdateErr::TemporaryFailure));
unwrap_send_err!(nodes[0].node.send_payment(&route, payment_hash_2, &Some(payment_secret_2)), false, APIError::MonitorUpdateFailed, {});
check_added_monitors!(nodes[0], 1);
}
}
// Now fix monitor updating...
- *nodes[0].chain_monitor.update_ret.lock().unwrap() = Some(Ok(()));
+ chanmon_cfgs[0].persister.set_update_ret(Ok(()));
let (outpoint, latest_update) = nodes[0].chain_monitor.latest_monitor_update_id.lock().unwrap().get(&channel_id).unwrap().clone();
nodes[0].node.channel_monitor_updated(&outpoint, latest_update);
check_added_monitors!(nodes[0], 0);
let send_event = SendEvent::from_event(nodes[0].node.get_and_clear_pending_msg_events().remove(0));
nodes[1].node.handle_update_add_htlc(&nodes[0].node.get_our_node_id(), &send_event.msgs[0]);
- *nodes[1].chain_monitor.update_ret.lock().unwrap() = Some(Err(ChannelMonitorUpdateErr::TemporaryFailure));
+ chanmon_cfgs[1].persister.set_update_ret(Err(ChannelMonitorUpdateErr::TemporaryFailure));
nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), &send_event.commitment_msg);
assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
nodes[1].logger.assert_log("lightning::ln::channelmanager".to_string(), "Failed to update ChannelMonitor".to_string(), 1);
check_added_monitors!(nodes[1], 1);
assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
- *nodes[1].chain_monitor.update_ret.lock().unwrap() = Some(Ok(()));
+ chanmon_cfgs[1].persister.set_update_ret(Ok(()));
let (outpoint, latest_update) = nodes[1].chain_monitor.latest_monitor_update_id.lock().unwrap().get(&channel_id).unwrap().clone();
nodes[1].node.channel_monitor_updated(&outpoint, latest_update);
check_added_monitors!(nodes[1], 0);
assert!(updates.update_fee.is_none());
assert_eq!(*node_id, nodes[0].node.get_our_node_id());
- *nodes[0].chain_monitor.update_ret.lock().unwrap() = Some(Err(ChannelMonitorUpdateErr::TemporaryFailure));
+ chanmon_cfgs[0].persister.set_update_ret(Err(ChannelMonitorUpdateErr::TemporaryFailure));
nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &updates.commitment_signed);
assert!(nodes[0].node.get_and_clear_pending_msg_events().is_empty());
nodes[0].logger.assert_log("lightning::ln::channelmanager".to_string(), "Failed to update ChannelMonitor".to_string(), 1);
_ => panic!("Unexpected event"),
}
- *nodes[0].chain_monitor.update_ret.lock().unwrap() = Some(Ok(()));
+ chanmon_cfgs[0].persister.set_update_ret(Ok(()));
let (outpoint, latest_update) = nodes[0].chain_monitor.latest_monitor_update_id.lock().unwrap().get(&channel_id).unwrap().clone();
nodes[0].node.channel_monitor_updated(&outpoint, latest_update);
check_added_monitors!(nodes[0], 0);
nodes[1].node.handle_update_add_htlc(&nodes[0].node.get_our_node_id(), &send_event.msgs[0]);
let bs_raa = commitment_signed_dance!(nodes[1], nodes[0], send_event.commitment_msg, false, true, false, true);
- *nodes[1].chain_monitor.update_ret.lock().unwrap() = Some(Err(ChannelMonitorUpdateErr::TemporaryFailure));
+ chanmon_cfgs[1].persister.set_update_ret(Err(ChannelMonitorUpdateErr::TemporaryFailure));
nodes[1].node.handle_revoke_and_ack(&nodes[0].node.get_our_node_id(), &bs_raa);
assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
nodes[1].logger.assert_log("lightning::ln::channelmanager".to_string(), "Failed to update ChannelMonitor".to_string(), 1);
assert!(nodes[1].node.get_and_clear_pending_events().is_empty());
check_added_monitors!(nodes[1], 1);
- *nodes[1].chain_monitor.update_ret.lock().unwrap() = Some(Ok(()));
+ chanmon_cfgs[1].persister.set_update_ret(Ok(()));
let (outpoint, latest_update) = nodes[1].chain_monitor.latest_monitor_update_id.lock().unwrap().get(&channel_id).unwrap().clone();
nodes[1].node.channel_monitor_updated(&outpoint, latest_update);
assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
check_added_monitors!(nodes[1], 1);
let bs_raa = get_event_msg!(nodes[1], MessageSendEvent::SendRevokeAndACK, nodes[0].node.get_our_node_id());
- *nodes[0].chain_monitor.update_ret.lock().unwrap() = Some(Err(ChannelMonitorUpdateErr::TemporaryFailure));
+ chanmon_cfgs[0].persister.set_update_ret(Err(ChannelMonitorUpdateErr::TemporaryFailure));
nodes[0].node.handle_update_add_htlc(&nodes[1].node.get_our_node_id(), &send_event_2.msgs[0]);
nodes[0].node.handle_commitment_signed(&nodes[1].node.get_our_node_id(), &send_event_2.commitment_msg);
assert!(nodes[0].node.get_and_clear_pending_msg_events().is_empty());
nodes[0].logger.assert_log("lightning::ln::channelmanager".to_string(), "Previous monitor update failure prevented responses to RAA".to_string(), 1);
check_added_monitors!(nodes[0], 1);
- *nodes[0].chain_monitor.update_ret.lock().unwrap() = Some(Ok(()));
+ chanmon_cfgs[0].persister.set_update_ret(Ok(()));
let (outpoint, latest_update) = nodes[0].chain_monitor.latest_monitor_update_id.lock().unwrap().get(&channel_id).unwrap().clone();
nodes[0].node.channel_monitor_updated(&outpoint, latest_update);
check_added_monitors!(nodes[0], 0);
assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
// Now fail monitor updating.
- *nodes[1].chain_monitor.update_ret.lock().unwrap() = Some(Err(ChannelMonitorUpdateErr::TemporaryFailure));
+ chanmon_cfgs[1].persister.set_update_ret(Err(ChannelMonitorUpdateErr::TemporaryFailure));
nodes[1].node.handle_revoke_and_ack(&nodes[2].node.get_our_node_id(), &bs_revoke_and_ack);
assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
nodes[1].logger.assert_log("lightning::ln::channelmanager".to_string(), "Failed to update ChannelMonitor".to_string(), 1);
check_added_monitors!(nodes[0], 1);
}
- *nodes[1].chain_monitor.update_ret.lock().unwrap() = Some(Ok(())); // We succeed in updating the monitor for the first channel
+ chanmon_cfgs[1].persister.set_update_ret(Ok(())); // We succeed in updating the monitor for the first channel
send_event = SendEvent::from_event(nodes[0].node.get_and_clear_pending_msg_events().remove(0));
nodes[1].node.handle_update_add_htlc(&nodes[0].node.get_our_node_id(), &send_event.msgs[0]);
commitment_signed_dance!(nodes[1], nodes[0], send_event.commitment_msg, false, true);
// Restore monitor updating, ensuring we immediately get a fail-back update and a
// update_add update.
- *nodes[1].chain_monitor.update_ret.lock().unwrap() = Some(Ok(()));
+ chanmon_cfgs[1].persister.set_update_ret(Ok(()));
let (outpoint, latest_update) = nodes[1].chain_monitor.latest_monitor_update_id.lock().unwrap().get(&chan_2.2).unwrap().clone();
nodes[1].node.channel_monitor_updated(&outpoint, latest_update);
check_added_monitors!(nodes[1], 0);
assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
commitment_signed_dance!(nodes[1], nodes[2], updates.commitment_signed, false);
- *nodes[1].chain_monitor.update_ret.lock().unwrap() = Some(Err(ChannelMonitorUpdateErr::TemporaryFailure));
+ chanmon_cfgs[1].persister.set_update_ret(Err(ChannelMonitorUpdateErr::TemporaryFailure));
nodes[0].node.peer_connected(&nodes[1].node.get_our_node_id(), &msgs::Init { features: InitFeatures::empty() });
nodes[1].node.peer_connected(&nodes[0].node.get_our_node_id(), &msgs::Init { features: InitFeatures::empty() });
get_event_msg!(nodes[1], MessageSendEvent::SendChannelUpdate, nodes[0].node.get_our_node_id())
.contents.flags & 2, 0); // The "disabled" bit should be unset as we just reconnected
- *nodes[1].chain_monitor.update_ret.lock().unwrap() = Some(Ok(()));
+ chanmon_cfgs[1].persister.set_update_ret(Ok(()));
let (outpoint, latest_update) = nodes[1].chain_monitor.latest_monitor_update_id.lock().unwrap().get(&chan_1.2).unwrap().clone();
nodes[1].node.channel_monitor_updated(&outpoint, latest_update);
check_added_monitors!(nodes[1], 0);
// Now we have a CS queued up which adds a new HTLC (which will need a RAA/CS response from
// nodes[1]) followed by an RAA. Fail the monitor updating prior to the CS, deliver the RAA,
// then restore channel monitor updates.
- *nodes[1].chain_monitor.update_ret.lock().unwrap() = Some(Err(ChannelMonitorUpdateErr::TemporaryFailure));
+ chanmon_cfgs[1].persister.set_update_ret(Err(ChannelMonitorUpdateErr::TemporaryFailure));
nodes[1].node.handle_update_add_htlc(&nodes[0].node.get_our_node_id(), &payment_event.msgs[0]);
nodes[1].node.handle_commitment_signed(&nodes[0].node.get_our_node_id(), &payment_event.commitment_msg);
assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
nodes[1].logger.assert_log("lightning::ln::channelmanager".to_string(), "Previous monitor update failure prevented responses to RAA".to_string(), 1);
check_added_monitors!(nodes[1], 1);
- *nodes[1].chain_monitor.update_ret.lock().unwrap() = Some(Ok(()));
+ chanmon_cfgs[1].persister.set_update_ret(Ok(()));
let (outpoint, latest_update) = nodes[1].chain_monitor.latest_monitor_update_id.lock().unwrap().get(&channel_id).unwrap().clone();
nodes[1].node.channel_monitor_updated(&outpoint, latest_update);
// nodes[1] should be AwaitingRAA here!
// Now deliver a's reestablish, freeing the claim from the holding cell, but fail the monitor
// update.
- *nodes[1].chain_monitor.update_ret.lock().unwrap() = Some(Err(ChannelMonitorUpdateErr::TemporaryFailure));
+ chanmon_cfgs[1].persister.set_update_ret(Err(ChannelMonitorUpdateErr::TemporaryFailure));
nodes[1].node.handle_channel_reestablish(&nodes[0].node.get_our_node_id(), &as_reconnect);
let _bs_channel_update = get_event_msg!(nodes[1], MessageSendEvent::SendChannelUpdate, nodes[0].node.get_our_node_id());
// Now un-fail the monitor, which will result in B sending its original commitment update,
// receiving the commitment update from A, and the resulting commitment dances.
- *nodes[1].chain_monitor.update_ret.lock().unwrap() = Some(Ok(()));
+ chanmon_cfgs[1].persister.set_update_ret(Ok(()));
let (outpoint, latest_update) = nodes[1].chain_monitor.latest_monitor_update_id.lock().unwrap().get(&channel_id).unwrap().clone();
nodes[1].node.channel_monitor_updated(&outpoint, latest_update);
check_added_monitors!(nodes[1], 0);
check_added_monitors!(nodes[0], 1);
}
- *nodes[1].chain_monitor.update_ret.lock().unwrap() = Some(Err(ChannelMonitorUpdateErr::TemporaryFailure));
+ chanmon_cfgs[1].persister.set_update_ret(Err(ChannelMonitorUpdateErr::TemporaryFailure));
let mut events = nodes[0].node.get_and_clear_pending_msg_events();
assert_eq!(events.len(), 1);
let payment_event = SendEvent::from_event(events.pop().unwrap());
nodes[0].node.handle_channel_reestablish(&nodes[1].node.get_our_node_id(), &bs_reconnect);
let _as_channel_update = get_event_msg!(nodes[0], MessageSendEvent::SendChannelUpdate, nodes[1].node.get_our_node_id());
- *nodes[1].chain_monitor.update_ret.lock().unwrap() = Some(Ok(()));
+ chanmon_cfgs[1].persister.set_update_ret(Ok(()));
let (outpoint, latest_update) = nodes[1].chain_monitor.latest_monitor_update_id.lock().unwrap().get(&channel_id).unwrap().clone();
nodes[1].node.channel_monitor_updated(&outpoint, latest_update);
check_added_monitors!(nodes[1], 0);
let payment_event = SendEvent::from_event(events.pop().unwrap());
assert_eq!(payment_event.node_id, nodes[1].node.get_our_node_id());
- *nodes[1].chain_monitor.update_ret.lock().unwrap() = Some(Err(ChannelMonitorUpdateErr::TemporaryFailure));
+ chanmon_cfgs[1].persister.set_update_ret(Err(ChannelMonitorUpdateErr::TemporaryFailure));
// Deliver the final RAA for the first payment, which does not require a response. RAAs
// generally require a commitment_signed, so the fact that we're expecting an opposite response
assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
nodes[1].logger.assert_log("lightning::ln::channelmanager".to_string(), "Previous monitor update failure prevented generation of RAA".to_string(), 1);
- *nodes[1].chain_monitor.update_ret.lock().unwrap() = Some(Ok(()));
+ chanmon_cfgs[1].persister.set_update_ret(Ok(()));
let (outpoint, latest_update) = nodes[1].chain_monitor.latest_monitor_update_id.lock().unwrap().get(&channel_id).unwrap().clone();
nodes[1].node.channel_monitor_updated(&outpoint, latest_update);
check_added_monitors!(nodes[1], 0);
let (payment_preimage_1, _, _) = route_payment(&nodes[0], &[&nodes[1]], 1000000);
- *nodes[1].chain_monitor.update_ret.lock().unwrap() = Some(Err(ChannelMonitorUpdateErr::TemporaryFailure));
+ chanmon_cfgs[1].persister.set_update_ret(Err(ChannelMonitorUpdateErr::TemporaryFailure));
assert!(nodes[1].node.claim_funds(payment_preimage_1));
nodes[1].logger.assert_log("lightning::ln::channelmanager".to_string(), "Temporary failure claiming HTLC, treating as success: Failed to update ChannelMonitor".to_string(), 1);
check_added_monitors!(nodes[1], 1);
// Successfully update the monitor on the 1<->2 channel, but the 0<->1 channel should still be
// paused, so forward shouldn't succeed until we call channel_monitor_updated().
- *nodes[1].chain_monitor.update_ret.lock().unwrap() = Some(Ok(()));
+ chanmon_cfgs[1].persister.set_update_ret(Ok(()));
let mut events = nodes[2].node.get_and_clear_pending_msg_events();
assert_eq!(events.len(), 1);
nodes[1].node.handle_update_add_htlc(&nodes[2].node.get_our_node_id(), &payment_event.msgs[0]);
commitment_signed_dance!(nodes[1], nodes[2], payment_event.commitment_msg, false);
- *nodes[1].chain_monitor.update_ret.lock().unwrap() = Some(Err(ChannelMonitorUpdateErr::TemporaryFailure));
+ chanmon_cfgs[1].persister.set_update_ret(Err(ChannelMonitorUpdateErr::TemporaryFailure));
expect_pending_htlcs_forwardable!(nodes[1]);
check_added_monitors!(nodes[1], 1);
assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
nodes[1].logger.assert_log("lightning::ln::channelmanager".to_string(), "Failed to update ChannelMonitor".to_string(), 1);
- *nodes[1].chain_monitor.update_ret.lock().unwrap() = Some(Ok(()));
+ chanmon_cfgs[1].persister.set_update_ret(Ok(()));
let (outpoint, latest_update) = nodes[1].chain_monitor.latest_monitor_update_id.lock().unwrap().get(&chan_1.2).unwrap().clone();
nodes[1].node.channel_monitor_updated(&outpoint, latest_update);
check_added_monitors!(nodes[1], 0);
nodes[1].node.handle_update_add_htlc(&nodes[0].node.get_our_node_id(), &payment_event.msgs[0]);
let as_raa = commitment_signed_dance!(nodes[1], nodes[0], payment_event.commitment_msg, false, true, false, true);
- *nodes[1].chain_monitor.update_ret.lock().unwrap() = Some(Err(ChannelMonitorUpdateErr::TemporaryFailure));
+ chanmon_cfgs[1].persister.set_update_ret(Err(ChannelMonitorUpdateErr::TemporaryFailure));
assert!(nodes[1].node.claim_funds(payment_preimage_1));
check_added_monitors!(nodes[1], 1);
let events = nodes[1].node.get_and_clear_pending_msg_events();
assert_eq!(events.len(), 0);
nodes[1].logger.assert_log("lightning::ln::channelmanager".to_string(), "Temporary failure claiming HTLC, treating as success: Failed to update ChannelMonitor".to_string(), 1);
- *nodes[1].chain_monitor.update_ret.lock().unwrap() = Some(Ok(()));
+ chanmon_cfgs[1].persister.set_update_ret(Ok(()));
let (outpoint, latest_update) = nodes[1].chain_monitor.latest_monitor_update_id.lock().unwrap().get(&channel_id).unwrap().clone();
nodes[1].node.channel_monitor_updated(&outpoint, latest_update);
check_added_monitors!(nodes[1], 0);
nodes[0].node.funding_transaction_generated(&temporary_channel_id, funding_tx.clone()).unwrap();
check_added_monitors!(nodes[0], 0);
- *nodes[1].chain_monitor.update_ret.lock().unwrap() = Some(Err(ChannelMonitorUpdateErr::TemporaryFailure));
+ chanmon_cfgs[1].persister.set_update_ret(Err(ChannelMonitorUpdateErr::TemporaryFailure));
let funding_created_msg = get_event_msg!(nodes[0], MessageSendEvent::SendFundingCreated, nodes[1].node.get_our_node_id());
let channel_id = OutPoint { txid: funding_created_msg.funding_txid, index: funding_created_msg.funding_output_index }.to_channel_id();
nodes[1].node.handle_funding_created(&nodes[0].node.get_our_node_id(), &funding_created_msg);
check_added_monitors!(nodes[1], 1);
- *nodes[0].chain_monitor.update_ret.lock().unwrap() = Some(Err(ChannelMonitorUpdateErr::TemporaryFailure));
+ chanmon_cfgs[0].persister.set_update_ret(Err(ChannelMonitorUpdateErr::TemporaryFailure));
nodes[0].node.handle_funding_signed(&nodes[1].node.get_our_node_id(), &get_event_msg!(nodes[1], MessageSendEvent::SendFundingSigned, nodes[0].node.get_our_node_id()));
assert!(nodes[0].node.get_and_clear_pending_msg_events().is_empty());
nodes[0].logger.assert_log("lightning::ln::channelmanager".to_string(), "Failed to update ChannelMonitor".to_string(), 1);
check_added_monitors!(nodes[0], 1);
assert!(nodes[0].node.get_and_clear_pending_events().is_empty());
- *nodes[0].chain_monitor.update_ret.lock().unwrap() = Some(Ok(()));
+ chanmon_cfgs[0].persister.set_update_ret(Ok(()));
let (outpoint, latest_update) = nodes[0].chain_monitor.latest_monitor_update_id.lock().unwrap().get(&channel_id).unwrap().clone();
nodes[0].node.channel_monitor_updated(&outpoint, latest_update);
check_added_monitors!(nodes[0], 0);
assert!(nodes[1].node.get_and_clear_pending_events().is_empty());
}
- *nodes[1].chain_monitor.update_ret.lock().unwrap() = Some(Ok(()));
+ chanmon_cfgs[1].persister.set_update_ret(Ok(()));
let (outpoint, latest_update) = nodes[1].chain_monitor.latest_monitor_update_id.lock().unwrap().get(&channel_id).unwrap().clone();
nodes[1].node.channel_monitor_updated(&outpoint, latest_update);
check_added_monitors!(nodes[1], 0);
// Set it so that the first monitor update (for the path 0 -> 1 -> 3) succeeds, but the second
// (for the path 0 -> 2 -> 3) fails.
- *nodes[0].chain_monitor.update_ret.lock().unwrap() = Some(Ok(()));
- *nodes[0].chain_monitor.next_update_ret.lock().unwrap() = Some(Err(ChannelMonitorUpdateErr::TemporaryFailure));
+ chanmon_cfgs[0].persister.set_update_ret(Ok(()));
+ chanmon_cfgs[0].persister.set_next_update_ret(Some(Err(ChannelMonitorUpdateErr::TemporaryFailure)));
// Now check that we get the right return value, indicating that the first path succeeded but
// the second got a MonitorUpdateFailed err. This implies PaymentSendFailure::PartialFailure as
if let Err(APIError::MonitorUpdateFailed) = results[1] {} else { panic!(); }
} else { panic!(); }
check_added_monitors!(nodes[0], 2);
- *nodes[0].chain_monitor.update_ret.lock().unwrap() = Some(Ok(()));
+ chanmon_cfgs[0].persister.set_update_ret(Ok(()));
// Pass the first HTLC of the payment along to nodes[3].
let mut events = nodes[0].node.get_and_clear_pending_msg_events();
nodes[0].node.send_payment(&route, payment_hash_2, &Some(payment_secret_2)).unwrap();
check_added_monitors!(nodes[0], 0);
- *nodes[0].chain_monitor.update_ret.lock().unwrap() = Some(Err(ChannelMonitorUpdateErr::TemporaryFailure));
+ chanmon_cfgs[0].persister.set_update_ret(Err(ChannelMonitorUpdateErr::TemporaryFailure));
assert!(nodes[0].node.claim_funds(payment_preimage_0));
check_added_monitors!(nodes[0], 1);
if reload_a {
let nodes_0_serialized = nodes[0].node.encode();
let mut chan_0_monitor_serialized = test_utils::TestVecWriter(Vec::new());
- nodes[0].chain_monitor.chain_monitor.monitors.read().unwrap().iter().next().unwrap().1.write(&mut chan_0_monitor_serialized).unwrap();
+ get_monitor!(nodes[0], chan_id).write(&mut chan_0_monitor_serialized).unwrap();
persister = test_utils::TestPersister::new();
let keys_manager = &chanmon_cfgs[0].keys_manager;
// If we finish updating the monitor, we should free the holding cell right away (this did
// not occur prior to #756).
- *nodes[0].chain_monitor.update_ret.lock().unwrap() = None;
+ chanmon_cfgs[0].persister.set_update_ret(Ok(()));
let (funding_txo, mon_id) = nodes[0].chain_monitor.latest_monitor_update_id.lock().unwrap().get(&chan_id).unwrap().clone();
nodes[0].node.channel_monitor_updated(&funding_txo, mon_id);
let (_, _, channel_id, funding_tx) = create_announced_chan_between_nodes(&nodes, 0, 1, InitFeatures::known(), InitFeatures::known());
- *nodes[0].chain_monitor.update_ret.lock().unwrap() = Some(Err(ChannelMonitorUpdateErr::TemporaryFailure));
- *nodes[1].chain_monitor.update_ret.lock().unwrap() = Some(Err(ChannelMonitorUpdateErr::TemporaryFailure));
+ chanmon_cfgs[0].persister.set_update_ret(Err(ChannelMonitorUpdateErr::TemporaryFailure));
+ chanmon_cfgs[1].persister.set_update_ret(Err(ChannelMonitorUpdateErr::TemporaryFailure));
nodes[0].node.close_channel(&channel_id).unwrap();
nodes[1].node.handle_shutdown(&nodes[0].node.get_our_node_id(), &InitFeatures::known(), &get_event_msg!(nodes[0], MessageSendEvent::SendShutdown, nodes[1].node.get_our_node_id()));
assert!(nodes[0].node.get_and_clear_pending_msg_events().is_empty());
- *nodes[0].chain_monitor.update_ret.lock().unwrap() = None;
- *nodes[1].chain_monitor.update_ret.lock().unwrap() = None;
+ chanmon_cfgs[0].persister.set_update_ret(Ok(()));
+ chanmon_cfgs[1].persister.set_update_ret(Ok(()));
let (outpoint, latest_update) = nodes[0].chain_monitor.latest_monitor_update_id.lock().unwrap().get(&channel_id).unwrap().clone();
nodes[0].node.channel_monitor_updated(&outpoint, latest_update);
assert!(nodes[1].node.get_and_clear_pending_msg_events().is_empty());
- *nodes[1].chain_monitor.update_ret.lock().unwrap() = None;
+ chanmon_cfgs[1].persister.set_update_ret(Ok(()));
let (outpoint, latest_update) = nodes[1].chain_monitor.latest_monitor_update_id.lock().unwrap().get(&channel_id).unwrap().clone();
nodes[1].node.channel_monitor_updated(&outpoint, latest_update);
let mut nodes = create_network(2, &node_cfgs, &node_chanmgrs);
let channel_id = create_announced_chan_between_nodes(&nodes, 0, 1, InitFeatures::known(), InitFeatures::known()).2;
- *nodes[0].chain_monitor.update_ret.lock().unwrap() = Some(Err(ChannelMonitorUpdateErr::PermanentFailure));
+ chanmon_cfgs[0].persister.set_update_ret(Err(ChannelMonitorUpdateErr::PermanentFailure));
assert!(nodes[0].node.close_channel(&channel_id).is_ok());
check_closed_broadcast!(nodes[0], true);
let mut nodes = create_network(2, &node_cfgs, &node_chanmgrs);
let channel_id = create_announced_chan_between_nodes(&nodes, 0, 1, InitFeatures::known(), InitFeatures::known()).2;
- *nodes[1].chain_monitor.update_ret.lock().unwrap() = Some(Err(ChannelMonitorUpdateErr::PermanentFailure));
+ chanmon_cfgs[1].persister.set_update_ret(Err(ChannelMonitorUpdateErr::PermanentFailure));
assert!(nodes[0].node.close_channel(&channel_id).is_ok());
let shutdown = get_event_msg!(nodes[0], MessageSendEvent::SendShutdown, nodes[1].node.get_our_node_id());
let (payment_preimage_1, _, _) = route_payment(&nodes[0], &[&nodes[1]], 1000000);
let (payment_preimage_2, _, _) = route_payment(&nodes[0], &[&nodes[1]], 1000000);
- *nodes[1].chain_monitor.update_ret.lock().unwrap() = Some(Err(ChannelMonitorUpdateErr::TemporaryFailure));
+ chanmon_cfgs[1].persister.set_update_ret(Err(ChannelMonitorUpdateErr::TemporaryFailure));
// `claim_funds` results in a ChannelMonitorUpdate.
assert!(nodes[1].node.claim_funds(payment_preimage_1));
check_added_monitors!(nodes[1], 1);
let (funding_tx, latest_update_1) = nodes[1].chain_monitor.latest_monitor_update_id.lock().unwrap().get(&channel_id).unwrap().clone();
- *nodes[1].chain_monitor.update_ret.lock().unwrap() = Some(Err(ChannelMonitorUpdateErr::TemporaryFailure));
+ chanmon_cfgs[1].persister.set_update_ret(Err(ChannelMonitorUpdateErr::TemporaryFailure));
// Previously, this would've panicked due to a double-call to `Channel::monitor_update_failed`,
// which had some asserts that prevented it from being called twice.
assert!(nodes[1].node.claim_funds(payment_preimage_2));
check_added_monitors!(nodes[1], 1);
- *nodes[1].chain_monitor.update_ret.lock().unwrap() = Some(Ok(()));
+ chanmon_cfgs[1].persister.set_update_ret(Ok(()));
let (_, latest_update_2) = nodes[1].chain_monitor.latest_monitor_update_id.lock().unwrap().get(&channel_id).unwrap().clone();
nodes[1].node.channel_monitor_updated(&funding_tx, latest_update_1);
use bitcoin::secp256k1;
use chain;
-use chain::{Confirm, Watch, BestBlock};
+use chain::{Confirm, ChannelMonitorUpdateErr, Watch, BestBlock};
use chain::chaininterface::{BroadcasterInterface, ConfirmationTarget, FeeEstimator};
-use chain::channelmonitor::{ChannelMonitor, ChannelMonitorUpdate, ChannelMonitorUpdateStep, ChannelMonitorUpdateErr, HTLC_FAIL_BACK_BUFFER, CLTV_CLAIM_BUFFER, LATENCY_GRACE_PERIOD_BLOCKS, ANTI_REORG_DELAY, MonitorEvent, CLOSED_CHANNEL_UPDATE_ID};
+use chain::channelmonitor::{ChannelMonitor, ChannelMonitorUpdate, ChannelMonitorUpdateStep, HTLC_FAIL_BACK_BUFFER, CLTV_CLAIM_BUFFER, LATENCY_GRACE_PERIOD_BLOCKS, ANTI_REORG_DELAY, MonitorEvent, CLOSED_CHANNEL_UPDATE_ID};
use chain::transaction::{OutPoint, TransactionData};
// Since this struct is returned in `list_channels` methods, expose it here in case users want to
// construct one themselves.
#[cfg(all(any(test, feature = "_test_utils"), feature = "unstable"))]
pub mod bench {
use chain::Listen;
- use chain::chainmonitor::ChainMonitor;
- use chain::channelmonitor::Persist;
+ use chain::chainmonitor::{ChainMonitor, Persist};
use chain::keysinterface::{KeysManager, InMemorySigner};
use ln::channelmanager::{BestBlock, ChainParameters, ChannelManager, PaymentHash, PaymentPreimage};
use ln::features::{InitFeatures, InvoiceFeatures};
let feeest = test_utils::TestFeeEstimator { sat_per_kw: Mutex::new(253) };
let mut deserialized_monitors = Vec::new();
{
- let old_monitors = self.chain_monitor.chain_monitor.monitors.read().unwrap();
- for (_, old_monitor) in old_monitors.iter() {
+ for outpoint in self.chain_monitor.chain_monitor.list_monitors() {
let mut w = test_utils::TestVecWriter(Vec::new());
- old_monitor.write(&mut w).unwrap();
+ self.chain_monitor.chain_monitor.get_monitor(outpoint).unwrap().write(&mut w).unwrap();
let (_, deserialized_monitor) = <(BlockHash, ChannelMonitor<EnforcingSigner>)>::read(
&mut io::Cursor::new(&w.0), self.keys_manager).unwrap();
deserialized_monitors.push(deserialized_monitor);
}
}
-/// Returns any local commitment transactions for the channel.
+/// Returns a channel monitor given a channel id, making some naive assumptions
#[macro_export]
-macro_rules! get_local_commitment_txn {
+macro_rules! get_monitor {
($node: expr, $channel_id: expr) => {
{
- let monitors = $node.chain_monitor.chain_monitor.monitors.read().unwrap();
- let mut commitment_txn = None;
- for (funding_txo, monitor) in monitors.iter() {
- if funding_txo.to_channel_id() == $channel_id {
- commitment_txn = Some(monitor.unsafe_get_latest_holder_commitment_txn(&$node.logger));
+ use bitcoin::hashes::Hash;
+ let mut monitor = None;
+ // Assume funding vout is either 0 or 1 blindly
+ for index in 0..2 {
+ if let Ok(mon) = $node.chain_monitor.chain_monitor.get_monitor(
+ $crate::chain::transaction::OutPoint {
+ txid: bitcoin::Txid::from_slice(&$channel_id[..]).unwrap(), index
+ })
+ {
+ monitor = Some(mon);
break;
}
}
- commitment_txn.unwrap()
+ monitor.unwrap()
+ }
+ }
+}
+
+/// Returns any local commitment transactions for the channel.
+#[macro_export]
+macro_rules! get_local_commitment_txn {
+ ($node: expr, $channel_id: expr) => {
+ {
+ $crate::get_monitor!($node, $channel_id).unsafe_get_latest_holder_commitment_txn(&$node.logger)
}
}
}
use util::ser::{Writeable, ReadableArgs};
use util::config::UserConfig;
-use bitcoin::hash_types::{Txid, BlockHash};
+use bitcoin::hash_types::BlockHash;
use bitcoin::blockdata::block::{Block, BlockHeader};
use bitcoin::blockdata::script::Builder;
use bitcoin::blockdata::opcodes;
// Drop the ChannelMonitor for the previous channel to avoid it broadcasting transactions and
// confusing us in the following tests.
- let chan_3_mon = nodes[3].chain_monitor.chain_monitor.monitors.write().unwrap().remove(&OutPoint { txid: chan_3.3.txid(), index: 0 }).unwrap();
+ let chan_3_mon = nodes[3].chain_monitor.chain_monitor.remove_monitor(&OutPoint { txid: chan_3.3.txid(), index: 0 });
// One pending HTLC to time out:
let payment_preimage_2 = route_payment(&nodes[3], &vec!(&nodes[4])[..], 3000000).0;
assert_eq!(nodes[3].node.list_channels().len(), 0);
assert_eq!(nodes[4].node.list_channels().len(), 0);
- nodes[3].chain_monitor.chain_monitor.monitors.write().unwrap().insert(OutPoint { txid: chan_3.3.txid(), index: 0 }, chan_3_mon);
+ nodes[3].chain_monitor.chain_monitor.watch_channel(OutPoint { txid: chan_3.3.txid(), index: 0 }, chan_3_mon).unwrap();
check_closed_event!(nodes[3], 1, ClosureReason::CommitmentTxConfirmed);
check_closed_event!(nodes[4], 1, ClosureReason::CommitmentTxConfirmed);
}
// Now check that if we add the preimage to ChannelMonitor it broadcasts our HTLC-Success..
{
- let mut monitors = nodes[2].chain_monitor.chain_monitor.monitors.read().unwrap();
- monitors.get(&OutPoint{ txid: Txid::from_slice(&payment_event.commitment_msg.channel_id[..]).unwrap(), index: 0 }).unwrap()
+ get_monitor!(nodes[2], payment_event.commitment_msg.channel_id)
.provide_payment_preimage(&our_payment_hash, &our_payment_preimage, &node_cfgs[2].tx_broadcaster, &node_cfgs[2].fee_estimator, &node_cfgs[2].logger);
}
mine_transaction(&nodes[2], &tx);
confirm_transaction(&nodes[0], &tx);
let events_1 = nodes[0].node.get_and_clear_pending_msg_events();
+ let chan_id;
assert_eq!(events_1.len(), 1);
match events_1[0] {
- MessageSendEvent::SendFundingLocked { ref node_id, msg: _ } => {
+ MessageSendEvent::SendFundingLocked { ref node_id, ref msg } => {
assert_eq!(*node_id, nodes[1].node.get_our_node_id());
+ chan_id = msg.channel_id;
},
_ => panic!("Unexpected event"),
}
let nodes_0_serialized = nodes[0].node.encode();
let mut chan_0_monitor_serialized = test_utils::TestVecWriter(Vec::new());
- nodes[0].chain_monitor.chain_monitor.monitors.read().unwrap().iter().next().unwrap().1.write(&mut chan_0_monitor_serialized).unwrap();
+ get_monitor!(nodes[0], chan_id).write(&mut chan_0_monitor_serialized).unwrap();
persister = test_utils::TestPersister::new();
let keys_manager = &chanmon_cfgs[0].keys_manager;
let nodes_0_serialized = nodes[0].node.encode();
let mut chan_0_monitor_serialized = test_utils::TestVecWriter(Vec::new());
- nodes[0].chain_monitor.chain_monitor.monitors.read().unwrap().iter().next().unwrap().1.write(&mut chan_0_monitor_serialized).unwrap();
+ get_monitor!(nodes[0], OutPoint { txid: tx.txid(), index: 0 }.to_channel_id())
+ .write(&mut chan_0_monitor_serialized).unwrap();
logger = test_utils::TestLogger::new();
fee_estimator = test_utils::TestFeeEstimator { sat_per_kw: Mutex::new(253) };
let nodes_0_deserialized: ChannelManager<EnforcingSigner, &test_utils::TestChainMonitor, &test_utils::TestBroadcaster, &test_utils::TestKeysInterface, &test_utils::TestFeeEstimator, &test_utils::TestLogger>;
let mut nodes = create_network(2, &node_cfgs, &node_chanmgrs);
- create_announced_chan_between_nodes(&nodes, 0, 1, InitFeatures::known(), InitFeatures::known());
+ let chan_id = create_announced_chan_between_nodes(&nodes, 0, 1, InitFeatures::known(), InitFeatures::known()).2;
// Route a payment, but force-close the channel before the HTLC fulfill message arrives at
// nodes[0].
// fairly normal behavior as ChannelMonitor(s) are often not re-serialized when on-chain events
// happen, unlike ChannelManager which tends to be re-serialized after any relevant event(s).
let mut chan_0_monitor_serialized = test_utils::TestVecWriter(Vec::new());
- nodes[0].chain_monitor.chain_monitor.monitors.read().unwrap().iter().next().unwrap().1.write(&mut chan_0_monitor_serialized).unwrap();
+ get_monitor!(nodes[0], chan_id).write(&mut chan_0_monitor_serialized).unwrap();
header.prev_blockhash = nodes[0].best_block_hash();
let claim_block = Block { header, txdata: claim_txn};
added_monitors.clear();
}
- node_a.node.handle_funding_signed(&node_b.node.get_our_node_id(), &get_event_msg!(node_b, MessageSendEvent::SendFundingSigned, node_a.node.get_our_node_id()));
+ let bs_funding_signed = get_event_msg!(node_b, MessageSendEvent::SendFundingSigned, node_a.node.get_our_node_id());
+ node_a.node.handle_funding_signed(&node_b.node.get_our_node_id(), &bs_funding_signed);
{
let mut added_monitors = node_a.chain_monitor.added_monitors.lock().unwrap();
assert_eq!(added_monitors.len(), 1);
// Start the de/seriailization process mid-channel creation to check that the channel manager will hold onto events that are serialized
let nodes_0_serialized = nodes[0].node.encode();
let mut chan_0_monitor_serialized = test_utils::TestVecWriter(Vec::new());
- nodes[0].chain_monitor.chain_monitor.monitors.read().unwrap().iter().next().unwrap().1.write(&mut chan_0_monitor_serialized).unwrap();
+ get_monitor!(nodes[0], bs_funding_signed.channel_id).write(&mut chan_0_monitor_serialized).unwrap();
fee_estimator = test_utils::TestFeeEstimator { sat_per_kw: Mutex::new(253) };
logger = test_utils::TestLogger::new();
let new_chain_monitor: test_utils::TestChainMonitor;
let nodes_0_deserialized: ChannelManager<EnforcingSigner, &test_utils::TestChainMonitor, &test_utils::TestBroadcaster, &test_utils::TestKeysInterface, &test_utils::TestFeeEstimator, &test_utils::TestLogger>;
let mut nodes = create_network(2, &node_cfgs, &node_chanmgrs);
- create_announced_chan_between_nodes(&nodes, 0, 1, InitFeatures::known(), InitFeatures::known());
+ let chan_id = create_announced_chan_between_nodes(&nodes, 0, 1, InitFeatures::known(), InitFeatures::known()).2;
let (our_payment_preimage, _, _) = route_payment(&nodes[0], &[&nodes[1]], 1000000);
let (_, our_payment_hash, _) = route_payment(&nodes[0], &[&nodes[1]], 1000000);
let nodes_0_serialized = nodes[0].node.encode();
let mut chan_0_monitor_serialized = test_utils::TestVecWriter(Vec::new());
- nodes[0].chain_monitor.chain_monitor.monitors.read().unwrap().iter().next().unwrap().1.write(&mut chan_0_monitor_serialized).unwrap();
+ get_monitor!(nodes[0], chan_id).write(&mut chan_0_monitor_serialized).unwrap();
logger = test_utils::TestLogger::new();
fee_estimator = test_utils::TestFeeEstimator { sat_per_kw: Mutex::new(253) };
let new_chain_monitor: test_utils::TestChainMonitor;
let nodes_0_deserialized: ChannelManager<EnforcingSigner, &test_utils::TestChainMonitor, &test_utils::TestBroadcaster, &test_utils::TestKeysInterface, &test_utils::TestFeeEstimator, &test_utils::TestLogger>;
let mut nodes = create_network(4, &node_cfgs, &node_chanmgrs);
- create_announced_chan_between_nodes(&nodes, 0, 1, InitFeatures::known(), InitFeatures::known());
- create_announced_chan_between_nodes(&nodes, 2, 0, InitFeatures::known(), InitFeatures::known());
+ let chan_id_1 = create_announced_chan_between_nodes(&nodes, 0, 1, InitFeatures::known(), InitFeatures::known()).2;
+ let chan_id_2 = create_announced_chan_between_nodes(&nodes, 2, 0, InitFeatures::known(), InitFeatures::known()).2;
let (_, _, channel_id, funding_tx) = create_announced_chan_between_nodes(&nodes, 0, 3, InitFeatures::known(), InitFeatures::known());
let mut node_0_stale_monitors_serialized = Vec::new();
- for monitor in nodes[0].chain_monitor.chain_monitor.monitors.read().unwrap().iter() {
+ for chan_id_iter in &[chan_id_1, chan_id_2, channel_id] {
let mut writer = test_utils::TestVecWriter(Vec::new());
- monitor.1.write(&mut writer).unwrap();
+ get_monitor!(nodes[0], chan_id_iter).write(&mut writer).unwrap();
node_0_stale_monitors_serialized.push(writer.0);
}
// Now the ChannelMonitor (which is now out-of-sync with ChannelManager for channel w/
// nodes[3])
let mut node_0_monitors_serialized = Vec::new();
- for monitor in nodes[0].chain_monitor.chain_monitor.monitors.read().unwrap().iter() {
+ for chan_id_iter in &[chan_id_1, chan_id_2, channel_id] {
let mut writer = test_utils::TestVecWriter(Vec::new());
- monitor.1.write(&mut writer).unwrap();
+ get_monitor!(nodes[0], chan_id_iter).write(&mut writer).unwrap();
node_0_monitors_serialized.push(writer.0);
}
// Cache node A state before any channel update
let previous_node_state = nodes[0].node.encode();
let mut previous_chain_monitor_state = test_utils::TestVecWriter(Vec::new());
- nodes[0].chain_monitor.chain_monitor.monitors.read().unwrap().iter().next().unwrap().1.write(&mut previous_chain_monitor_state).unwrap();
+ get_monitor!(nodes[0], chan.2).write(&mut previous_chain_monitor_state).unwrap();
send_payment(&nodes[0], &vec!(&nodes[1])[..], 8000000);
send_payment(&nodes[0], &vec!(&nodes[1])[..], 8000000);
let nodes_1_deserialized: ChannelManager<EnforcingSigner, &test_utils::TestChainMonitor, &test_utils::TestBroadcaster, &test_utils::TestKeysInterface, &test_utils::TestFeeEstimator, &test_utils::TestLogger>;
let mut nodes = create_network(3, &node_cfgs, &node_chanmgrs);
- create_announced_chan_between_nodes_with_value(&nodes, 0, 1, 1_000_000, 500_000_000, InitFeatures::known(), InitFeatures::known());
+ let chan_id_1 = create_announced_chan_between_nodes_with_value(&nodes, 0, 1, 1_000_000, 500_000_000, InitFeatures::known(), InitFeatures::known()).2;
// Note that the create_*_chan functions in utils requires announcement_signatures, which we do
// not send for private channels.
nodes[2].node.handle_funding_created(&nodes[1].node.get_our_node_id(), &get_event_msg!(nodes[1], MessageSendEvent::SendFundingCreated, nodes[2].node.get_our_node_id()));
check_added_monitors!(nodes[2], 1);
- nodes[1].node.handle_funding_signed(&nodes[2].node.get_our_node_id(), &get_event_msg!(nodes[2], MessageSendEvent::SendFundingSigned, nodes[1].node.get_our_node_id()));
+ let cs_funding_signed = get_event_msg!(nodes[2], MessageSendEvent::SendFundingSigned, nodes[1].node.get_our_node_id());
+ nodes[1].node.handle_funding_signed(&nodes[2].node.get_our_node_id(), &cs_funding_signed);
check_added_monitors!(nodes[1], 1);
let conf_height = core::cmp::max(nodes[1].best_block_info().1 + 1, nodes[2].best_block_info().1 + 1);
let nodes_1_serialized = nodes[1].node.encode();
let mut monitor_a_serialized = test_utils::TestVecWriter(Vec::new());
let mut monitor_b_serialized = test_utils::TestVecWriter(Vec::new());
- {
- let mons = nodes[1].chain_monitor.chain_monitor.monitors.read().unwrap();
- let mut mon_iter = mons.iter();
- mon_iter.next().unwrap().1.write(&mut monitor_a_serialized).unwrap();
- mon_iter.next().unwrap().1.write(&mut monitor_b_serialized).unwrap();
- }
+ get_monitor!(nodes[1], chan_id_1).write(&mut monitor_a_serialized).unwrap();
+ get_monitor!(nodes[1], cs_funding_signed.channel_id).write(&mut monitor_b_serialized).unwrap();
persister = test_utils::TestPersister::new();
let keys_manager = &chanmon_cfgs[1].keys_manager;
connect_block(&nodes[0], &Block { header: header_130, txdata: penalty_txn });
connect_blocks(&nodes[0], ANTI_REORG_DELAY - 1);
{
- let monitors = nodes[0].chain_monitor.chain_monitor.monitors.read().unwrap();
- if let Some(monitor) = monitors.get(&OutPoint { txid: chan.3.txid(), index: 0 }) {
- assert!(monitor.inner.lock().unwrap().onchain_tx_handler.pending_claim_requests.is_empty());
- assert!(monitor.inner.lock().unwrap().onchain_tx_handler.claimable_outpoints.is_empty());
- }
+ let monitor = nodes[0].chain_monitor.chain_monitor.get_monitor(OutPoint { txid: chan.3.txid(), index: 0 }).unwrap();
+ assert!(monitor.inner.lock().unwrap().onchain_tx_handler.pending_claim_requests.is_empty());
+ assert!(monitor.inner.lock().unwrap().onchain_tx_handler.claimable_outpoints.is_empty());
}
}
let logger = test_utils::TestLogger::with_id(format!("node {}", 0));
let persister = test_utils::TestPersister::new();
let watchtower = {
- let monitors = nodes[0].chain_monitor.chain_monitor.monitors.read().unwrap();
- let monitor = monitors.get(&outpoint).unwrap();
+ let monitor = nodes[0].chain_monitor.chain_monitor.get_monitor(outpoint).unwrap();
let mut w = test_utils::TestVecWriter(Vec::new());
monitor.write(&mut w).unwrap();
let new_monitor = <(BlockHash, channelmonitor::ChannelMonitor<EnforcingSigner>)>::read(
let logger = test_utils::TestLogger::with_id(format!("node {}", "Alice"));
let persister = test_utils::TestPersister::new();
let watchtower_alice = {
- let monitors = nodes[0].chain_monitor.chain_monitor.monitors.read().unwrap();
- let monitor = monitors.get(&outpoint).unwrap();
+ let monitor = nodes[0].chain_monitor.chain_monitor.get_monitor(outpoint).unwrap();
let mut w = test_utils::TestVecWriter(Vec::new());
monitor.write(&mut w).unwrap();
let new_monitor = <(BlockHash, channelmonitor::ChannelMonitor<EnforcingSigner>)>::read(
let logger = test_utils::TestLogger::with_id(format!("node {}", "Bob"));
let persister = test_utils::TestPersister::new();
let watchtower_bob = {
- let monitors = nodes[0].chain_monitor.chain_monitor.monitors.read().unwrap();
- let monitor = monitors.get(&outpoint).unwrap();
+ let monitor = nodes[0].chain_monitor.chain_monitor.get_monitor(outpoint).unwrap();
let mut w = test_utils::TestVecWriter(Vec::new());
monitor.write(&mut w).unwrap();
let new_monitor = <(BlockHash, channelmonitor::ChannelMonitor<EnforcingSigner>)>::read(
let new_chain_monitor: test_utils::TestChainMonitor;
let nodes_1_deserialized: ChannelManager<EnforcingSigner, &test_utils::TestChainMonitor, &test_utils::TestBroadcaster, &test_utils::TestKeysInterface, &test_utils::TestFeeEstimator, &test_utils::TestLogger>;
let mut nodes = create_network(3, &node_cfgs, &node_chanmgrs);
- create_announced_chan_between_nodes(&nodes, 0, 1, InitFeatures::known(), InitFeatures::known());
- create_announced_chan_between_nodes(&nodes, 1, 2, InitFeatures::known(), InitFeatures::known());
+ let chan_id_1 = create_announced_chan_between_nodes(&nodes, 0, 1, InitFeatures::known(), InitFeatures::known()).2;
+ let chan_id_2 = create_announced_chan_between_nodes(&nodes, 1, 2, InitFeatures::known(), InitFeatures::known()).2;
// First send a payment to nodes[1]
let (route, payment_hash, payment_preimage, payment_secret) = get_route_and_payment_hash!(nodes[0], nodes[1], 100_000);
let nodes_1_serialized = nodes[1].node.encode();
let mut chan_0_monitor_serialized = test_utils::TestVecWriter(Vec::new());
let mut chan_1_monitor_serialized = test_utils::TestVecWriter(Vec::new());
- {
- let monitors = nodes[1].chain_monitor.chain_monitor.monitors.read().unwrap();
- let mut monitor_iter = monitors.iter();
- monitor_iter.next().unwrap().1.write(&mut chan_0_monitor_serialized).unwrap();
- monitor_iter.next().unwrap().1.write(&mut chan_1_monitor_serialized).unwrap();
- }
+ get_monitor!(nodes[1], chan_id_1).write(&mut chan_0_monitor_serialized).unwrap();
+ get_monitor!(nodes[1], chan_id_2).write(&mut chan_1_monitor_serialized).unwrap();
persister = test_utils::TestPersister::new();
let keys_manager = &chanmon_cfgs[1].keys_manager;
assert_eq!(vec![Balance::ClaimableOnChannelClose {
claimable_amount_satoshis: 1_000_000 - 1_000 - chan_feerate * channel::COMMITMENT_TX_BASE_WEIGHT / 1000
}],
- nodes[0].chain_monitor.chain_monitor.monitors.read().unwrap().get(&funding_outpoint).unwrap().get_claimable_balances());
+ nodes[0].chain_monitor.chain_monitor.get_monitor(funding_outpoint).unwrap().get_claimable_balances());
assert_eq!(vec![Balance::ClaimableOnChannelClose { claimable_amount_satoshis: 1_000, }],
- nodes[1].chain_monitor.chain_monitor.monitors.read().unwrap().get(&funding_outpoint).unwrap().get_claimable_balances());
+ nodes[1].chain_monitor.chain_monitor.get_monitor(funding_outpoint).unwrap().get_claimable_balances());
nodes[0].node.close_channel(&chan_id).unwrap();
let node_0_shutdown = get_event_msg!(nodes[0], MessageSendEvent::SendShutdown, nodes[1].node.get_our_node_id());
claimable_amount_satoshis: 1_000_000 - 1_000 - chan_feerate * channel::COMMITMENT_TX_BASE_WEIGHT / 1000,
confirmation_height: nodes[0].best_block_info().1 + ANTI_REORG_DELAY - 1,
}],
- nodes[0].chain_monitor.chain_monitor.monitors.read().unwrap().get(&funding_outpoint).unwrap().get_claimable_balances());
+ nodes[0].chain_monitor.chain_monitor.get_monitor(funding_outpoint).unwrap().get_claimable_balances());
assert_eq!(vec![Balance::ClaimableAwaitingConfirmations {
claimable_amount_satoshis: 1000,
confirmation_height: nodes[1].best_block_info().1 + ANTI_REORG_DELAY - 1,
}],
- nodes[1].chain_monitor.chain_monitor.monitors.read().unwrap().get(&funding_outpoint).unwrap().get_claimable_balances());
+ nodes[1].chain_monitor.chain_monitor.get_monitor(funding_outpoint).unwrap().get_claimable_balances());
connect_blocks(&nodes[0], ANTI_REORG_DELAY - 1);
connect_blocks(&nodes[1], ANTI_REORG_DELAY - 1);
assert_eq!(Vec::<Balance>::new(),
- nodes[0].chain_monitor.chain_monitor.monitors.read().unwrap().get(&funding_outpoint).unwrap().get_claimable_balances());
+ nodes[0].chain_monitor.chain_monitor.get_monitor(funding_outpoint).unwrap().get_claimable_balances());
assert_eq!(Vec::<Balance>::new(),
- nodes[1].chain_monitor.chain_monitor.monitors.read().unwrap().get(&funding_outpoint).unwrap().get_claimable_balances());
+ nodes[1].chain_monitor.chain_monitor.get_monitor(funding_outpoint).unwrap().get_claimable_balances());
let mut node_a_spendable = nodes[0].chain_monitor.chain_monitor.get_and_clear_pending_events();
assert_eq!(node_a_spendable.len(), 1);
claimable_amount_satoshis: 4_000,
claimable_height: htlc_cltv_timeout,
}]),
- sorted_vec(nodes[0].chain_monitor.chain_monitor.monitors.read().unwrap().get(&funding_outpoint).unwrap().get_claimable_balances()));
+ sorted_vec(nodes[0].chain_monitor.chain_monitor.get_monitor(funding_outpoint).unwrap().get_claimable_balances()));
assert_eq!(vec![Balance::ClaimableOnChannelClose {
claimable_amount_satoshis: 1_000,
}],
- nodes[1].chain_monitor.chain_monitor.monitors.read().unwrap().get(&funding_outpoint).unwrap().get_claimable_balances());
+ nodes[1].chain_monitor.chain_monitor.get_monitor(funding_outpoint).unwrap().get_claimable_balances());
nodes[1].node.claim_funds(payment_preimage);
check_added_monitors!(nodes[1], 1);
});
}
assert_eq!(sorted_vec(a_expected_balances),
- sorted_vec(nodes[0].chain_monitor.chain_monitor.monitors.read().unwrap().get(&funding_outpoint).unwrap().get_claimable_balances()));
+ sorted_vec(nodes[0].chain_monitor.chain_monitor.get_monitor(funding_outpoint).unwrap().get_claimable_balances()));
assert_eq!(vec![Balance::ClaimableOnChannelClose {
claimable_amount_satoshis: 1_000 + 3_000 + 4_000,
}],
- nodes[1].chain_monitor.chain_monitor.monitors.read().unwrap().get(&funding_outpoint).unwrap().get_claimable_balances());
+ nodes[1].chain_monitor.chain_monitor.get_monitor(funding_outpoint).unwrap().get_claimable_balances());
// Broadcast the closing transaction (which has both pending HTLCs in it) and get B's
// broadcasted HTLC claim transaction with preimage.
claimable_amount_satoshis: 4_000,
claimable_height: htlc_cltv_timeout,
}]),
- sorted_vec(nodes[0].chain_monitor.chain_monitor.monitors.read().unwrap().get(&funding_outpoint).unwrap().get_claimable_balances()));
+ sorted_vec(nodes[0].chain_monitor.chain_monitor.get_monitor(funding_outpoint).unwrap().get_claimable_balances()));
// The main non-HTLC balance is just awaiting confirmations, but the claimable height is the
// CSV delay, not ANTI_REORG_DELAY.
assert_eq!(sorted_vec(vec![Balance::ClaimableAwaitingConfirmations {
claimable_amount_satoshis: 4_000,
timeout_height: htlc_cltv_timeout,
}]),
- sorted_vec(nodes[1].chain_monitor.chain_monitor.monitors.read().unwrap().get(&funding_outpoint).unwrap().get_claimable_balances()));
+ sorted_vec(nodes[1].chain_monitor.chain_monitor.get_monitor(funding_outpoint).unwrap().get_claimable_balances()));
connect_blocks(&nodes[0], ANTI_REORG_DELAY - 1);
expect_payment_failed!(nodes[0], dust_payment_hash, true);
claimable_amount_satoshis: 4_000,
claimable_height: htlc_cltv_timeout,
}]),
- sorted_vec(nodes[0].chain_monitor.chain_monitor.monitors.read().unwrap().get(&funding_outpoint).unwrap().get_claimable_balances()));
+ sorted_vec(nodes[0].chain_monitor.chain_monitor.get_monitor(funding_outpoint).unwrap().get_claimable_balances()));
assert_eq!(sorted_vec(vec![Balance::ClaimableAwaitingConfirmations {
claimable_amount_satoshis: 1_000,
confirmation_height: node_b_commitment_claimable,
claimable_amount_satoshis: 4_000,
timeout_height: htlc_cltv_timeout,
}]),
- sorted_vec(nodes[1].chain_monitor.chain_monitor.monitors.read().unwrap().get(&funding_outpoint).unwrap().get_claimable_balances()));
+ sorted_vec(nodes[1].chain_monitor.chain_monitor.get_monitor(funding_outpoint).unwrap().get_claimable_balances()));
let mut node_a_spendable = nodes[0].chain_monitor.chain_monitor.get_and_clear_pending_events();
assert_eq!(node_a_spendable.len(), 1);
claimable_amount_satoshis: 4_000,
claimable_height: htlc_cltv_timeout,
}]),
- sorted_vec(nodes[0].chain_monitor.chain_monitor.monitors.read().unwrap().get(&funding_outpoint).unwrap().get_claimable_balances()));
+ sorted_vec(nodes[0].chain_monitor.chain_monitor.get_monitor(funding_outpoint).unwrap().get_claimable_balances()));
connect_blocks(&nodes[0], ANTI_REORG_DELAY - 1);
assert_eq!(vec![Balance::MaybeClaimableHTLCAwaitingTimeout {
claimable_amount_satoshis: 4_000,
claimable_height: htlc_cltv_timeout,
}],
- nodes[0].chain_monitor.chain_monitor.monitors.read().unwrap().get(&funding_outpoint).unwrap().get_claimable_balances());
+ nodes[0].chain_monitor.chain_monitor.get_monitor(funding_outpoint).unwrap().get_claimable_balances());
// When the HTLC timeout output is spendable in the next block, A should broadcast it
connect_blocks(&nodes[0], htlc_cltv_timeout - nodes[0].best_block_info().1 - 1);
claimable_amount_satoshis: 4_000,
confirmation_height: nodes[0].best_block_info().1 + ANTI_REORG_DELAY - 1,
}],
- nodes[0].chain_monitor.chain_monitor.monitors.read().unwrap().get(&funding_outpoint).unwrap().get_claimable_balances());
+ nodes[0].chain_monitor.chain_monitor.get_monitor(funding_outpoint).unwrap().get_claimable_balances());
// After ANTI_REORG_DELAY, A will generate a SpendableOutputs event and drop the claimable
// balance entry.
connect_blocks(&nodes[0], ANTI_REORG_DELAY - 1);
assert_eq!(Vec::<Balance>::new(),
- nodes[0].chain_monitor.chain_monitor.monitors.read().unwrap().get(&funding_outpoint).unwrap().get_claimable_balances());
+ nodes[0].chain_monitor.chain_monitor.get_monitor(funding_outpoint).unwrap().get_claimable_balances());
expect_payment_failed!(nodes[0], timeout_payment_hash, true);
let mut node_a_spendable = nodes[0].chain_monitor.chain_monitor.get_and_clear_pending_events();
claimable_amount_satoshis: 4_000,
timeout_height: htlc_cltv_timeout,
}]),
- sorted_vec(nodes[1].chain_monitor.chain_monitor.monitors.read().unwrap().get(&funding_outpoint).unwrap().get_claimable_balances()));
+ sorted_vec(nodes[1].chain_monitor.chain_monitor.get_monitor(funding_outpoint).unwrap().get_claimable_balances()));
// After reaching the commitment output CSV, we'll get a SpendableOutputs event for it and have
// only the HTLCs claimable on node B.
claimable_amount_satoshis: 4_000,
timeout_height: htlc_cltv_timeout,
}]),
- sorted_vec(nodes[1].chain_monitor.chain_monitor.monitors.read().unwrap().get(&funding_outpoint).unwrap().get_claimable_balances()));
+ sorted_vec(nodes[1].chain_monitor.chain_monitor.get_monitor(funding_outpoint).unwrap().get_claimable_balances()));
// After reaching the claimed HTLC output CSV, we'll get a SpendableOutptus event for it and
// have only one HTLC output left spendable.
claimable_amount_satoshis: 4_000,
timeout_height: htlc_cltv_timeout,
}],
- nodes[1].chain_monitor.chain_monitor.monitors.read().unwrap().get(&funding_outpoint).unwrap().get_claimable_balances());
+ nodes[1].chain_monitor.chain_monitor.get_monitor(funding_outpoint).unwrap().get_claimable_balances());
// Finally, mine the HTLC timeout transaction that A broadcasted (even though B should be able
// to claim this HTLC with the preimage it knows!). It will remain listed as a claimable HTLC
claimable_amount_satoshis: 4_000,
timeout_height: htlc_cltv_timeout,
}],
- nodes[1].chain_monitor.chain_monitor.monitors.read().unwrap().get(&funding_outpoint).unwrap().get_claimable_balances());
+ nodes[1].chain_monitor.chain_monitor.get_monitor(funding_outpoint).unwrap().get_claimable_balances());
connect_blocks(&nodes[1], ANTI_REORG_DELAY - 1);
assert_eq!(Vec::<Balance>::new(),
- nodes[1].chain_monitor.chain_monitor.monitors.read().unwrap().get(&funding_outpoint).unwrap().get_claimable_balances());
+ nodes[1].chain_monitor.chain_monitor.get_monitor(funding_outpoint).unwrap().get_claimable_balances());
}
#[test]
// it when we go to deserialize, and then use the ChannelManager.
let nodes_0_serialized = nodes[0].node.encode();
let mut chan_0_monitor_serialized = test_utils::TestVecWriter(Vec::new());
- nodes[0].chain_monitor.chain_monitor.monitors.read().unwrap().iter().next().unwrap().1.write(&mut chan_0_monitor_serialized).unwrap();
+ get_monitor!(nodes[0], chan.2).write(&mut chan_0_monitor_serialized).unwrap();
persister = test_utils::TestPersister::new();
let keys_manager = &chanmon_cfgs[0].keys_manager;
pub struct TestChainMonitor<'a> {
pub added_monitors: Mutex<Vec<(OutPoint, channelmonitor::ChannelMonitor<EnforcingSigner>)>>,
pub latest_monitor_update_id: Mutex<HashMap<[u8; 32], (OutPoint, u64)>>,
- pub chain_monitor: chainmonitor::ChainMonitor<EnforcingSigner, &'a TestChainSource, &'a chaininterface::BroadcasterInterface, &'a TestFeeEstimator, &'a TestLogger, &'a channelmonitor::Persist<EnforcingSigner>>,
+ pub chain_monitor: chainmonitor::ChainMonitor<EnforcingSigner, &'a TestChainSource, &'a chaininterface::BroadcasterInterface, &'a TestFeeEstimator, &'a TestLogger, &'a chainmonitor::Persist<EnforcingSigner>>,
pub keys_manager: &'a TestKeysInterface,
- pub update_ret: Mutex<Option<Result<(), channelmonitor::ChannelMonitorUpdateErr>>>,
- /// If this is set to Some(), after the next return, we'll always return this until update_ret
- /// is changed:
- pub next_update_ret: Mutex<Option<Result<(), channelmonitor::ChannelMonitorUpdateErr>>>,
/// If this is set to Some(), the next update_channel call (not watch_channel) must be a
/// ChannelForceClosed event for the given channel_id with should_broadcast set to the given
/// boolean.
pub expect_channel_force_closed: Mutex<Option<([u8; 32], bool)>>,
}
impl<'a> TestChainMonitor<'a> {
- pub fn new(chain_source: Option<&'a TestChainSource>, broadcaster: &'a chaininterface::BroadcasterInterface, logger: &'a TestLogger, fee_estimator: &'a TestFeeEstimator, persister: &'a channelmonitor::Persist<EnforcingSigner>, keys_manager: &'a TestKeysInterface) -> Self {
+ pub fn new(chain_source: Option<&'a TestChainSource>, broadcaster: &'a chaininterface::BroadcasterInterface, logger: &'a TestLogger, fee_estimator: &'a TestFeeEstimator, persister: &'a chainmonitor::Persist<EnforcingSigner>, keys_manager: &'a TestKeysInterface) -> Self {
Self {
added_monitors: Mutex::new(Vec::new()),
latest_monitor_update_id: Mutex::new(HashMap::new()),
chain_monitor: chainmonitor::ChainMonitor::new(chain_source, broadcaster, logger, fee_estimator, persister),
keys_manager,
- update_ret: Mutex::new(None),
- next_update_ret: Mutex::new(None),
expect_channel_force_closed: Mutex::new(None),
}
}
}
impl<'a> chain::Watch<EnforcingSigner> for TestChainMonitor<'a> {
- fn watch_channel(&self, funding_txo: OutPoint, monitor: channelmonitor::ChannelMonitor<EnforcingSigner>) -> Result<(), channelmonitor::ChannelMonitorUpdateErr> {
+ fn watch_channel(&self, funding_txo: OutPoint, monitor: channelmonitor::ChannelMonitor<EnforcingSigner>) -> Result<(), chain::ChannelMonitorUpdateErr> {
// At every point where we get a monitor update, we should be able to send a useful monitor
// to a watchtower and disk...
let mut w = TestVecWriter(Vec::new());
assert!(new_monitor == monitor);
self.latest_monitor_update_id.lock().unwrap().insert(funding_txo.to_channel_id(), (funding_txo, monitor.get_latest_update_id()));
self.added_monitors.lock().unwrap().push((funding_txo, monitor));
- let watch_res = self.chain_monitor.watch_channel(funding_txo, new_monitor);
-
- let ret = self.update_ret.lock().unwrap().clone();
- if let Some(next_ret) = self.next_update_ret.lock().unwrap().take() {
- *self.update_ret.lock().unwrap() = Some(next_ret);
- }
- if ret.is_some() {
- assert!(watch_res.is_ok());
- return ret.unwrap();
- }
- watch_res
+ self.chain_monitor.watch_channel(funding_txo, new_monitor)
}
- fn update_channel(&self, funding_txo: OutPoint, update: channelmonitor::ChannelMonitorUpdate) -> Result<(), channelmonitor::ChannelMonitorUpdateErr> {
+ fn update_channel(&self, funding_txo: OutPoint, update: channelmonitor::ChannelMonitorUpdate) -> Result<(), chain::ChannelMonitorUpdateErr> {
// Every monitor update should survive roundtrip
let mut w = TestVecWriter(Vec::new());
update.write(&mut w).unwrap();
let update_res = self.chain_monitor.update_channel(funding_txo, update);
// At every point where we get a monitor update, we should be able to send a useful monitor
// to a watchtower and disk...
- let monitors = self.chain_monitor.monitors.read().unwrap();
- let monitor = monitors.get(&funding_txo).unwrap();
+ let monitor = self.chain_monitor.get_monitor(funding_txo).unwrap();
w.0.clear();
monitor.write(&mut w).unwrap();
let new_monitor = <(BlockHash, channelmonitor::ChannelMonitor<EnforcingSigner>)>::read(
&mut io::Cursor::new(&w.0), self.keys_manager).unwrap().1;
assert!(new_monitor == *monitor);
self.added_monitors.lock().unwrap().push((funding_txo, new_monitor));
-
- let ret = self.update_ret.lock().unwrap().clone();
- if let Some(next_ret) = self.next_update_ret.lock().unwrap().take() {
- *self.update_ret.lock().unwrap() = Some(next_ret);
- }
- if ret.is_some() {
- assert!(update_res.is_ok());
- return ret.unwrap();
- }
update_res
}
}
pub struct TestPersister {
- pub update_ret: Mutex<Result<(), channelmonitor::ChannelMonitorUpdateErr>>
+ pub update_ret: Mutex<Result<(), chain::ChannelMonitorUpdateErr>>,
+ /// If this is set to Some(), after the next return, we'll always return this until update_ret
+ /// is changed:
+ pub next_update_ret: Mutex<Option<Result<(), chain::ChannelMonitorUpdateErr>>>,
+
}
impl TestPersister {
pub fn new() -> Self {
Self {
- update_ret: Mutex::new(Ok(()))
+ update_ret: Mutex::new(Ok(())),
+ next_update_ret: Mutex::new(None),
}
}
- pub fn set_update_ret(&self, ret: Result<(), channelmonitor::ChannelMonitorUpdateErr>) {
+ pub fn set_update_ret(&self, ret: Result<(), chain::ChannelMonitorUpdateErr>) {
*self.update_ret.lock().unwrap() = ret;
}
+
+ pub fn set_next_update_ret(&self, next_ret: Option<Result<(), chain::ChannelMonitorUpdateErr>>) {
+ *self.next_update_ret.lock().unwrap() = next_ret;
+ }
}
-impl<Signer: keysinterface::Sign> channelmonitor::Persist<Signer> for TestPersister {
- fn persist_new_channel(&self, _funding_txo: OutPoint, _data: &channelmonitor::ChannelMonitor<Signer>) -> Result<(), channelmonitor::ChannelMonitorUpdateErr> {
- self.update_ret.lock().unwrap().clone()
+impl<Signer: keysinterface::Sign> chainmonitor::Persist<Signer> for TestPersister {
+ fn persist_new_channel(&self, _funding_txo: OutPoint, _data: &channelmonitor::ChannelMonitor<Signer>) -> Result<(), chain::ChannelMonitorUpdateErr> {
+ let ret = self.update_ret.lock().unwrap().clone();
+ if let Some(next_ret) = self.next_update_ret.lock().unwrap().take() {
+ *self.update_ret.lock().unwrap() = next_ret;
+ }
+ ret
}
- fn update_persisted_channel(&self, _funding_txo: OutPoint, _update: &channelmonitor::ChannelMonitorUpdate, _data: &channelmonitor::ChannelMonitor<Signer>) -> Result<(), channelmonitor::ChannelMonitorUpdateErr> {
- self.update_ret.lock().unwrap().clone()
+ fn update_persisted_channel(&self, _funding_txo: OutPoint, _update: &channelmonitor::ChannelMonitorUpdate, _data: &channelmonitor::ChannelMonitor<Signer>) -> Result<(), chain::ChannelMonitorUpdateErr> {
+ let ret = self.update_ret.lock().unwrap().clone();
+ if let Some(next_ret) = self.next_update_ret.lock().unwrap().take() {
+ *self.update_ret.lock().unwrap() = next_ret;
+ }
+ ret
}
}
projects / rust-lightning / commitdiff