use bitcoin::blockdata::block::{Block, BlockHeader};
use bitcoin::hash_types::Txid;
use chain;
use chain::{Filter, WatchedOutput};
use chain::chaininterface::{BroadcasterInterface, FeeEstimator};
use chain::channelmonitor;
use chain::channelmonitor::{ChannelMonitor, ChannelMonitorUpdate, ChannelMonitorUpdateErr, MonitorEvent, Persist, TransactionOutputs};
use chain::transaction::{OutPoint, TransactionData};
use chain::keysinterface::Sign;
use util::logger::Logger;
use util::events;
use util::events::Event;
use std::collections::{HashMap, hash_map};
use std::sync::RwLock;
use std::ops::Deref;
pub struct ChainMonitor<ChannelSigner: Sign, C: Deref, T: Deref, F: Deref, L: Deref, P: Deref>
where C::Target: chain::Filter,
T::Target: BroadcasterInterface,
F::Target: FeeEstimator,
L::Target: Logger,
P::Target: channelmonitor::Persist<ChannelSigner>,
{
pub monitors: RwLock<HashMap<OutPoint, ChannelMonitor<ChannelSigner>>>,
chain_source: Option<C>,
broadcaster: T,
logger: L,
fee_estimator: F,
persister: P,
}
impl<ChannelSigner: Sign, C: Deref, T: Deref, F: Deref, L: Deref, P: Deref> ChainMonitor<ChannelSigner, C, T, F, L, P>
where C::Target: chain::Filter,
T::Target: BroadcasterInterface,
F::Target: FeeEstimator,
L::Target: Logger,
P::Target: channelmonitor::Persist<ChannelSigner>,
{
fn process_chain_data<FN>(&self, header: &BlockHeader, txdata: &TransactionData, process: FN)
where
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);
if let Some(ref chain_source) = self.chain_source {
let block_hash = header.block_hash();
for (txid, mut outputs) in txn_outputs.drain(..) {
for (idx, output) in outputs.drain(..) {
let output = WatchedOutput {
block_hash: Some(block_hash),
outpoint: OutPoint { txid, index: idx as u16 },
script_pubkey: output.script_pubkey,
};
if let Some(tx) = chain_source.register_output(output) {
dependent_txdata.push(tx);
}
}
}
}
}
if !dependent_txdata.is_empty() {
dependent_txdata.sort_unstable_by_key(|(index, _tx)| *index);
dependent_txdata.dedup_by_key(|(index, _tx)| *index);
let txdata: Vec<_> = dependent_txdata.iter().map(|(index, tx)| (*index, tx)).collect();
self.process_chain_data(header, &txdata, process);
}
}
pub fn new(chain_source: Option<C>, broadcaster: T, logger: L, feeest: F, persister: P) -> Self {
Self {
monitors: RwLock::new(HashMap::new()),
chain_source,
broadcaster,
logger,
fee_estimator: feeest,
persister,
}
}
}
impl<ChannelSigner: Sign, C: Deref, T: Deref, F: Deref, L: Deref, P: Deref>
chain::Listen for ChainMonitor<ChannelSigner, C, T, F, L, P>
where
C::Target: chain::Filter,
T::Target: BroadcasterInterface,
F::Target: FeeEstimator,
L::Target: Logger,
P::Target: channelmonitor::Persist<ChannelSigner>,
{
fn block_connected(&self, block: &Block, height: u32) {
let header = &block.header;
let txdata: Vec<_> = block.txdata.iter().enumerate().collect();
self.process_chain_data(header, &txdata, |monitor, txdata| {
monitor.block_connected(
header, txdata, height, &*self.broadcaster, &*self.fee_estimator, &*self.logger)
});
}
fn block_disconnected(&self, header: &BlockHeader, height: u32) {
let monitors = self.monitors.read().unwrap();
for monitor in monitors.values() {
monitor.block_disconnected(
header, height, &*self.broadcaster, &*self.fee_estimator, &*self.logger);
}
}
}
impl<ChannelSigner: Sign, C: Deref, T: Deref, F: Deref, L: Deref, P: Deref>
chain::Confirm for ChainMonitor<ChannelSigner, C, T, F, L, P>
where
C::Target: chain::Filter,
T::Target: BroadcasterInterface,
F::Target: FeeEstimator,
L::Target: Logger,
P::Target: channelmonitor::Persist<ChannelSigner>,
{
fn transactions_confirmed(&self, header: &BlockHeader, txdata: &TransactionData, height: u32) {
self.process_chain_data(header, txdata, |monitor, txdata| {
monitor.transactions_confirmed(
header, txdata, height, &*self.broadcaster, &*self.fee_estimator, &*self.logger)
});
}
fn transaction_unconfirmed(&self, txid: &Txid) {
let monitors = self.monitors.read().unwrap();
for monitor in monitors.values() {
monitor.transaction_unconfirmed(txid, &*self.broadcaster, &*self.fee_estimator, &*self.logger);
}
}
fn best_block_updated(&self, header: &BlockHeader, height: u32) {
self.process_chain_data(header, &[], |monitor, txdata| {
debug_assert!(txdata.is_empty());
monitor.best_block_updated(
header, height, &*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());
}
txids.sort_unstable();
txids.dedup();
txids
}
}
impl<ChannelSigner: Sign, C: Deref , T: Deref , F: Deref , L: Deref , P: Deref >
chain::Watch<ChannelSigner> for ChainMonitor<ChannelSigner, C, T, F, L, P>
where C::Target: chain::Filter,
T::Target: BroadcasterInterface,
F::Target: FeeEstimator,
L::Target: Logger,
P::Target: channelmonitor::Persist<ChannelSigner>,
{
fn watch_channel(&self, funding_outpoint: OutPoint, monitor: ChannelMonitor<ChannelSigner>) -> Result<(), ChannelMonitorUpdateErr> {
let mut monitors = self.monitors.write().unwrap();
let entry = match monitors.entry(funding_outpoint) {
hash_map::Entry::Occupied(_) => {
log_error!(self.logger, "Failed to add new channel data: channel monitor for given outpoint is already present");
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 funding_txo = monitor.get_funding_txo();
log_trace!(self.logger, "Got new Channel Monitor for channel {}", log_bytes!(funding_txo.0.to_channel_id()[..]));
if let Some(ref chain_source) = self.chain_source {
monitor.load_outputs_to_watch(chain_source);
}
}
entry.insert(monitor);
Ok(())
}
fn update_channel(&self, funding_txo: OutPoint, update: ChannelMonitorUpdate) -> Result<(), ChannelMonitorUpdateErr> {
let monitors = self.monitors.read().unwrap();
match monitors.get(&funding_txo) {
None => {
log_error!(self.logger, "Failed to update channel monitor: no such monitor registered");
#[cfg(any(test, feature = "fuzztarget"))]
panic!("ChannelManager generated a channel update for a channel that was not yet registered!");
#[cfg(not(any(test, feature = "fuzztarget")))]
Err(ChannelMonitorUpdateErr::PermanentFailure)
},
Some(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 {
log_error!(self.logger, "Failed to update channel monitor: {:?}", e);
}
let persist_res = self.persister.update_persisted_channel(funding_txo, &update, monitor);
if let Err(ref e) = persist_res {
log_error!(self.logger, "Failed to persist channel monitor update: {:?}", e);
}
if update_res.is_err() {
Err(ChannelMonitorUpdateErr::PermanentFailure)
} else {
persist_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());
}
pending_monitor_events
}
}
impl<ChannelSigner: Sign, C: Deref, T: Deref, F: Deref, L: Deref, P: Deref> events::EventsProvider for ChainMonitor<ChannelSigner, C, T, F, L, P>
where C::Target: chain::Filter,
T::Target: BroadcasterInterface,
F::Target: FeeEstimator,
L::Target: Logger,
P::Target: channelmonitor::Persist<ChannelSigner>,
{
fn get_and_clear_pending_events(&self) -> Vec<Event> {
let mut pending_events = Vec::new();
for monitor in self.monitors.read().unwrap().values() {
pending_events.append(&mut monitor.get_and_clear_pending_events());
}
pending_events
}
}
#[cfg(test)]
mod tests {
use ::{check_added_monitors, get_local_commitment_txn};
use ln::features::InitFeatures;
use ln::functional_test_utils::*;
use util::events::EventsProvider;
use util::events::MessageSendEventsProvider;
use util::test_utils::{OnRegisterOutput, TxOutReference};
#[test]
fn connect_block_checks_dependent_transactions() {
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 nodes = create_network(2, &node_cfgs, &node_chanmgrs);
let channel = create_announced_chan_between_nodes(
&nodes, 0, 1, InitFeatures::known(), InitFeatures::known());
let (commitment_tx, htlc_tx) = {
let payment_preimage = route_payment(&nodes[0], &vec!(&nodes[1])[..], 5_000_000).0;
let mut txn = get_local_commitment_txn!(nodes[0], channel.2);
claim_payment(&nodes[0], &vec!(&nodes[1])[..], payment_preimage);
assert_eq!(txn.len(), 2);
(txn.remove(0), txn.remove(0))
};
let htlc_output = TxOutReference(commitment_tx.clone(), 0);
let to_local_output = TxOutReference(commitment_tx.clone(), 1);
let htlc_timeout_output = TxOutReference(htlc_tx.clone(), 0);
nodes[1].chain_source
.expect(OnRegisterOutput { with: htlc_output, returns: Some((1, htlc_tx)) })
.expect(OnRegisterOutput { with: to_local_output, returns: None })
.expect(OnRegisterOutput { with: htlc_timeout_output, returns: None });
mine_transaction(&nodes[1], &commitment_tx);
check_added_monitors!(nodes[1], 1);
nodes[1].node.get_and_clear_pending_msg_events();
nodes[1].node.get_and_clear_pending_events();
}
}