Add comment noting that ChannelMonitor events are serialized
[rust-lightning] / lightning / src / ln / channelmonitor.rs
index 373f811e9ae4ea8ec77f104edff12fef1c0bca08..1bc8c76b49e82765f73e550c41e17c3243a027fc 100644 (file)
@@ -37,7 +37,7 @@ use chain::chaininterface::{ChainListener, ChainWatchInterface, BroadcasterInter
 use chain::transaction::OutPoint;
 use chain::keysinterface::{SpendableOutputDescriptor, ChannelKeys};
 use util::logger::Logger;
-use util::ser::{ReadableArgs, Readable, Writer, Writeable, U48};
+use util::ser::{ReadableArgs, Readable, MaybeReadable, Writer, Writeable, U48};
 use util::{byte_utils, events};
 
 use std::collections::{HashMap, hash_map, HashSet};
@@ -131,9 +131,9 @@ pub enum ChannelMonitorUpdateErr {
 }
 
 /// General Err type for ChannelMonitor actions. Generally, this implies that the data provided is
-/// inconsistent with the ChannelMonitor being called. eg for ChannelMonitor::insert_combine this
-/// means you tried to merge two monitors for different channels or for a channel which was
-/// restored from a backup and then generated new commitment updates.
+/// 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
+/// corrupted.
 /// Contains a human-readable error message.
 #[derive(Debug)]
 pub struct MonitorUpdateError(pub &'static str);
@@ -150,7 +150,7 @@ impl_writeable!(HTLCUpdate, 0, { payment_hash, payment_preimage, source });
 
 /// Simple trait indicating ability to track a set of ChannelMonitors and multiplex events between
 /// them. Generally should be implemented by keeping a local SimpleManyChannelMonitor and passing
-/// events to it, while also taking any add_update_monitor events and passing them to some remote
+/// events to it, while also taking any add/update_monitor events and passing them to some remote
 /// server(s).
 ///
 /// Note that any updates to a channel's monitor *must* be applied to each instance of the
@@ -164,7 +164,7 @@ impl_writeable!(HTLCUpdate, 0, { payment_hash, payment_preimage, source });
 /// BlockNotifier and call the BlockNotifier's `block_(dis)connected` methods, which will notify
 /// all registered listeners in one go.
 pub trait ManyChannelMonitor<ChanSigner: ChannelKeys>: Send + Sync {
-       /// Adds or updates a monitor for the given `funding_txo`.
+       /// Adds a monitor for the given `funding_txo`.
        ///
        /// Implementer must also ensure that the funding_txo txid *and* outpoint are registered with
        /// any relevant ChainWatchInterfaces such that the provided monitor receives block_connected
@@ -176,7 +176,7 @@ pub trait ManyChannelMonitor<ChanSigner: ChannelKeys>: Send + Sync {
        ///
        /// Any spends of outputs which should have been registered which aren't passed to
        /// ChannelMonitors via block_connected may result in FUNDS LOSS.
-       fn add_update_monitor(&self, funding_txo: OutPoint, monitor: ChannelMonitor<ChanSigner>) -> Result<(), ChannelMonitorUpdateErr>;
+       fn add_monitor(&self, funding_txo: OutPoint, monitor: ChannelMonitor<ChanSigner>) -> Result<(), ChannelMonitorUpdateErr>;
 
        /// Updates a monitor for the given `funding_txo`.
        ///
@@ -212,33 +212,31 @@ pub trait ManyChannelMonitor<ChanSigner: ChannelKeys>: Send + Sync {
 ///
 /// If you're using this for local monitoring of your own channels, you probably want to use
 /// `OutPoint` as the key, which will give you a ManyChannelMonitor implementation.
-pub struct SimpleManyChannelMonitor<Key, ChanSigner: ChannelKeys, T: Deref> where T::Target: BroadcasterInterface {
+pub struct SimpleManyChannelMonitor<Key, ChanSigner: ChannelKeys, T: Deref, F: Deref>
+       where T::Target: BroadcasterInterface,
+        F::Target: FeeEstimator
+{
        #[cfg(test)] // Used in ChannelManager tests to manipulate channels directly
        pub monitors: Mutex<HashMap<Key, ChannelMonitor<ChanSigner>>>,
        #[cfg(not(test))]
        monitors: Mutex<HashMap<Key, ChannelMonitor<ChanSigner>>>,
        chain_monitor: Arc<ChainWatchInterface>,
        broadcaster: T,
-       pending_events: Mutex<Vec<events::Event>>,
        logger: Arc<Logger>,
-       fee_estimator: Arc<FeeEstimator>
+       fee_estimator: F
 }
 
-impl<'a, Key : Send + cmp::Eq + hash::Hash, ChanSigner: ChannelKeys, T: Deref + Sync + Send> ChainListener for SimpleManyChannelMonitor<Key, ChanSigner, T>
-       where T::Target: BroadcasterInterface
+impl<'a, Key : Send + cmp::Eq + hash::Hash, ChanSigner: ChannelKeys, T: Deref + Sync + Send, F: Deref + Sync + Send>
+       ChainListener for SimpleManyChannelMonitor<Key, ChanSigner, T, F>
+       where T::Target: BroadcasterInterface,
+             F::Target: FeeEstimator
 {
        fn block_connected(&self, header: &BlockHeader, height: u32, txn_matched: &[&Transaction], _indexes_of_txn_matched: &[u32]) {
                let block_hash = header.bitcoin_hash();
-               let mut new_events: Vec<events::Event> = Vec::with_capacity(0);
                {
                        let mut monitors = self.monitors.lock().unwrap();
                        for monitor in monitors.values_mut() {
-                               let (txn_outputs, spendable_outputs) = monitor.block_connected(txn_matched, height, &block_hash, &*self.broadcaster, &*self.fee_estimator);
-                               if spendable_outputs.len() > 0 {
-                                       new_events.push(events::Event::SpendableOutputs {
-                                               outputs: spendable_outputs,
-                                       });
-                               }
+                               let txn_outputs = monitor.block_connected(txn_matched, height, &block_hash, &*self.broadcaster, &*self.fee_estimator);
 
                                for (ref txid, ref outputs) in txn_outputs {
                                        for (idx, output) in outputs.iter().enumerate() {
@@ -247,8 +245,6 @@ impl<'a, Key : Send + cmp::Eq + hash::Hash, ChanSigner: ChannelKeys, T: Deref +
                                }
                        }
                }
-               let mut pending_events = self.pending_events.lock().unwrap();
-               pending_events.append(&mut new_events);
        }
 
        fn block_disconnected(&self, header: &BlockHeader, disconnected_height: u32) {
@@ -260,17 +256,17 @@ impl<'a, Key : Send + cmp::Eq + hash::Hash, ChanSigner: ChannelKeys, T: Deref +
        }
 }
 
-impl<Key : Send + cmp::Eq + hash::Hash + 'static, ChanSigner: ChannelKeys, T: Deref> SimpleManyChannelMonitor<Key, ChanSigner, T>
-       where T::Target: BroadcasterInterface
+impl<Key : Send + cmp::Eq + hash::Hash + 'static, ChanSigner: ChannelKeys, T: Deref, F: Deref> SimpleManyChannelMonitor<Key, ChanSigner, T, F>
+       where T::Target: BroadcasterInterface,
+             F::Target: FeeEstimator
 {
        /// Creates a new object which can be used to monitor several channels given the chain
        /// interface with which to register to receive notifications.
-       pub fn new(chain_monitor: Arc<ChainWatchInterface>, broadcaster: T, logger: Arc<Logger>, feeest: Arc<FeeEstimator>) -> SimpleManyChannelMonitor<Key, ChanSigner, T> {
+       pub fn new(chain_monitor: Arc<ChainWatchInterface>, broadcaster: T, logger: Arc<Logger>, feeest: F) -> SimpleManyChannelMonitor<Key, ChanSigner, T, F> {
                let res = SimpleManyChannelMonitor {
                        monitors: Mutex::new(HashMap::new()),
                        chain_monitor,
                        broadcaster,
-                       pending_events: Mutex::new(Vec::new()),
                        logger,
                        fee_estimator: feeest,
                };
@@ -279,14 +275,11 @@ impl<Key : Send + cmp::Eq + hash::Hash + 'static, ChanSigner: ChannelKeys, T: De
        }
 
        /// Adds or updates the monitor which monitors the channel referred to by the given key.
-       pub fn add_update_monitor_by_key(&self, key: Key, monitor: ChannelMonitor<ChanSigner>) -> Result<(), MonitorUpdateError> {
+       pub fn add_monitor_by_key(&self, key: Key, monitor: ChannelMonitor<ChanSigner>) -> Result<(), MonitorUpdateError> {
                let mut monitors = self.monitors.lock().unwrap();
-               match monitors.get_mut(&key) {
-                       Some(orig_monitor) => {
-                               log_trace!(self, "Updating Channel Monitor for channel {}", log_funding_info!(monitor.key_storage));
-                               return orig_monitor.insert_combine(monitor);
-                       },
-                       None => {}
+               let entry = match monitors.entry(key) {
+                       hash_map::Entry::Occupied(_) => return Err(MonitorUpdateError("Channel monitor for given key is already present")),
+                       hash_map::Entry::Vacant(e) => e,
                };
                match monitor.key_storage {
                        Storage::Local { ref funding_info, .. } => {
@@ -310,7 +303,7 @@ impl<Key : Send + cmp::Eq + hash::Hash + 'static, ChanSigner: ChannelKeys, T: De
                                self.chain_monitor.install_watch_outpoint((*txid, idx as u32), script);
                        }
                }
-               monitors.insert(key, monitor);
+               entry.insert(monitor);
                Ok(())
        }
 
@@ -327,11 +320,12 @@ impl<Key : Send + cmp::Eq + hash::Hash + 'static, ChanSigner: ChannelKeys, T: De
        }
 }
 
-impl<ChanSigner: ChannelKeys, T: Deref + Sync + Send> ManyChannelMonitor<ChanSigner> for SimpleManyChannelMonitor<OutPoint, ChanSigner, T>
-       where T::Target: BroadcasterInterface
+impl<ChanSigner: ChannelKeys, T: Deref + Sync + Send, F: Deref + Sync + Send> ManyChannelMonitor<ChanSigner> for SimpleManyChannelMonitor<OutPoint, ChanSigner, T, F>
+       where T::Target: BroadcasterInterface,
+             F::Target: FeeEstimator
 {
-       fn add_update_monitor(&self, funding_txo: OutPoint, monitor: ChannelMonitor<ChanSigner>) -> Result<(), ChannelMonitorUpdateErr> {
-               match self.add_update_monitor_by_key(funding_txo, monitor) {
+       fn add_monitor(&self, funding_txo: OutPoint, monitor: ChannelMonitor<ChanSigner>) -> Result<(), ChannelMonitorUpdateErr> {
+               match self.add_monitor_by_key(funding_txo, monitor) {
                        Ok(_) => Ok(()),
                        Err(_) => Err(ChannelMonitorUpdateErr::PermanentFailure),
                }
@@ -353,14 +347,16 @@ impl<ChanSigner: ChannelKeys, T: Deref + Sync + Send> ManyChannelMonitor<ChanSig
        }
 }
 
-impl<Key : Send + cmp::Eq + hash::Hash, ChanSigner: ChannelKeys, T: Deref> events::EventsProvider for SimpleManyChannelMonitor<Key, ChanSigner, T>
-       where T::Target: BroadcasterInterface
+impl<Key : Send + cmp::Eq + hash::Hash, ChanSigner: ChannelKeys, T: Deref, F: Deref> events::EventsProvider for SimpleManyChannelMonitor<Key, ChanSigner, T, F>
+       where T::Target: BroadcasterInterface,
+             F::Target: FeeEstimator
 {
        fn get_and_clear_pending_events(&self) -> Vec<events::Event> {
-               let mut pending_events = self.pending_events.lock().unwrap();
-               let mut ret = Vec::new();
-               mem::swap(&mut ret, &mut *pending_events);
-               ret
+               let mut pending_events = Vec::new();
+               for chan in self.monitors.lock().unwrap().values_mut() {
+                       pending_events.append(&mut chan.get_and_clear_pending_events());
+               }
+               pending_events
        }
 }
 
@@ -393,7 +389,6 @@ pub(crate) const LATENCY_GRACE_PERIOD_BLOCKS: u32 = 3;
 /// keeping bumping another claim tx to solve the outpoint.
 pub(crate) const ANTI_REORG_DELAY: u32 = 6;
 
-#[derive(Clone)]
 enum Storage<ChanSigner: ChannelKeys> {
        Local {
                keys: ChanSigner,
@@ -788,7 +783,11 @@ impl<R: ::std::io::Read> Readable<R> for ChannelMonitorUpdateStep {
 ///
 /// You MUST ensure that no ChannelMonitors for a given channel anywhere contain out-of-date
 /// information and are actively monitoring the chain.
-#[derive(Clone)]
+///
+/// Pending Events or updated HTLCs which have not yet been read out by
+/// get_and_clear_pending_htlcs_updated or get_and_clear_pending_events are serialized to disk and
+/// reloaded at deserialize-time. Thus, you must ensure that, when handling events, all events
+/// gotten are fully handled before re-serializing the new state.
 pub struct ChannelMonitor<ChanSigner: ChannelKeys> {
        latest_update_id: u64,
        commitment_transaction_number_obscure_factor: u64,
@@ -832,6 +831,7 @@ pub struct ChannelMonitor<ChanSigner: ChannelKeys> {
        payment_preimages: HashMap<PaymentHash, PaymentPreimage>,
 
        pending_htlcs_updated: Vec<HTLCUpdate>,
+       pending_events: Vec<events::Event>,
 
        destination_script: Script,
        // Thanks to data loss protection, we may be able to claim our non-htlc funds
@@ -879,7 +879,7 @@ pub struct ChannelMonitor<ChanSigner: ChannelKeys> {
 
        // We simply modify last_block_hash in Channel's block_connected so that serialization is
        // consistent but hopefully the users' copy handles block_connected in a consistent way.
-       // (we do *not*, however, update them in insert_combine to ensure any local user copies keep
+       // (we do *not*, however, update them in update_monitor to ensure any local user copies keep
        // their last_block_hash from its state and not based on updated copies that didn't run through
        // the full block_connected).
        pub(crate) last_block_hash: Sha256dHash,
@@ -945,6 +945,7 @@ impl<ChanSigner: ChannelKeys> PartialEq for ChannelMonitor<ChanSigner> {
                        self.current_local_signed_commitment_tx != other.current_local_signed_commitment_tx ||
                        self.payment_preimages != other.payment_preimages ||
                        self.pending_htlcs_updated != other.pending_htlcs_updated ||
+                       self.pending_events.len() != other.pending_events.len() || // We trust events to round-trip properly
                        self.destination_script != other.destination_script ||
                        self.to_remote_rescue != other.to_remote_rescue ||
                        self.pending_claim_requests != other.pending_claim_requests ||
@@ -1132,6 +1133,11 @@ impl<ChanSigner: ChannelKeys + Writeable> ChannelMonitor<ChanSigner> {
                        data.write(writer)?;
                }
 
+               writer.write_all(&byte_utils::be64_to_array(self.pending_events.len() as u64))?;
+               for event in self.pending_events.iter() {
+                       event.write(writer)?;
+               }
+
                self.last_block_hash.write(writer)?;
                self.destination_script.write(writer)?;
                if let Some((ref to_remote_script, ref local_key)) = self.to_remote_rescue {
@@ -1215,31 +1221,43 @@ impl<ChanSigner: ChannelKeys + Writeable> ChannelMonitor<ChanSigner> {
 }
 
 impl<ChanSigner: ChannelKeys> ChannelMonitor<ChanSigner> {
-       pub(super) fn new(keys: ChanSigner, funding_key: &SecretKey, revocation_base_key: &SecretKey, delayed_payment_base_key: &SecretKey, htlc_base_key: &SecretKey, payment_base_key: &SecretKey, shutdown_pubkey: &PublicKey, our_to_self_delay: u16, destination_script: Script, logger: Arc<Logger>) -> ChannelMonitor<ChanSigner> {
+       pub(super) fn new(keys: ChanSigner, shutdown_pubkey: &PublicKey,
+                       our_to_self_delay: u16, destination_script: &Script, funding_info: (OutPoint, Script),
+                       their_htlc_base_key: &PublicKey, their_delayed_payment_base_key: &PublicKey,
+                       their_to_self_delay: u16, funding_redeemscript: Script, channel_value_satoshis: u64,
+                       commitment_transaction_number_obscure_factor: u64,
+                       logger: Arc<Logger>) -> ChannelMonitor<ChanSigner> {
+
+               assert!(commitment_transaction_number_obscure_factor <= (1 << 48));
+               let funding_key = keys.funding_key().clone();
+               let revocation_base_key = keys.revocation_base_key().clone();
+               let htlc_base_key = keys.htlc_base_key().clone();
+               let delayed_payment_base_key = keys.delayed_payment_base_key().clone();
+               let payment_base_key = keys.payment_base_key().clone();
                ChannelMonitor {
                        latest_update_id: 0,
-                       commitment_transaction_number_obscure_factor: 0,
+                       commitment_transaction_number_obscure_factor,
 
                        key_storage: Storage::Local {
                                keys,
-                               funding_key: funding_key.clone(),
-                               revocation_base_key: revocation_base_key.clone(),
-                               htlc_base_key: htlc_base_key.clone(),
-                               delayed_payment_base_key: delayed_payment_base_key.clone(),
-                               payment_base_key: payment_base_key.clone(),
+                               funding_key,
+                               revocation_base_key,
+                               htlc_base_key,
+                               delayed_payment_base_key,
+                               payment_base_key,
                                shutdown_pubkey: shutdown_pubkey.clone(),
-                               funding_info: None,
+                               funding_info: Some(funding_info),
                                current_remote_commitment_txid: None,
                                prev_remote_commitment_txid: None,
                        },
-                       their_htlc_base_key: None,
-                       their_delayed_payment_base_key: None,
-                       funding_redeemscript: None,
-                       channel_value_satoshis: None,
+                       their_htlc_base_key: Some(their_htlc_base_key.clone()),
+                       their_delayed_payment_base_key: Some(their_delayed_payment_base_key.clone()),
+                       funding_redeemscript: Some(funding_redeemscript),
+                       channel_value_satoshis: Some(channel_value_satoshis),
                        their_cur_revocation_points: None,
 
                        our_to_self_delay: our_to_self_delay,
-                       their_to_self_delay: None,
+                       their_to_self_delay: Some(their_to_self_delay),
 
                        commitment_secrets: CounterpartyCommitmentSecrets::new(),
                        remote_claimable_outpoints: HashMap::new(),
@@ -1252,8 +1270,9 @@ impl<ChanSigner: ChannelKeys> ChannelMonitor<ChanSigner> {
 
                        payment_preimages: HashMap::new(),
                        pending_htlcs_updated: Vec::new(),
+                       pending_events: Vec::new(),
 
-                       destination_script: destination_script,
+                       destination_script: destination_script.clone(),
                        to_remote_rescue: None,
 
                        pending_claim_requests: HashMap::new(),
@@ -1318,7 +1337,6 @@ impl<ChanSigner: ChannelKeys> ChannelMonitor<ChanSigner> {
 
                // Prune HTLCs from the previous remote commitment tx so we don't generate failure/fulfill
                // events for now-revoked/fulfilled HTLCs.
-               // TODO: We should probably consider whether we're really getting the next secret here.
                if let Storage::Local { ref mut prev_remote_commitment_txid, .. } = self.key_storage {
                        if let Some(txid) = prev_remote_commitment_txid.take() {
                                for &mut (_, ref mut source) in self.remote_claimable_outpoints.get_mut(&txid).unwrap() {
@@ -1497,97 +1515,6 @@ impl<ChanSigner: ChannelKeys> ChannelMonitor<ChanSigner> {
                Ok(())
        }
 
-       /// Combines this ChannelMonitor with the information contained in the other ChannelMonitor.
-       /// After a successful call this ChannelMonitor is up-to-date and is safe to use to monitor the
-       /// chain for new blocks/transactions.
-       pub fn insert_combine(&mut self, mut other: ChannelMonitor<ChanSigner>) -> Result<(), MonitorUpdateError> {
-               match self.key_storage {
-                       Storage::Local { ref funding_info, .. } => {
-                               if funding_info.is_none() { return Err(MonitorUpdateError("Try to combine a Local monitor without funding_info")); }
-                               let our_funding_info = funding_info;
-                               if let Storage::Local { ref funding_info, .. } = other.key_storage {
-                                       if funding_info.is_none() { return Err(MonitorUpdateError("Try to combine a Local monitor without funding_info")); }
-                                       // We should be able to compare the entire funding_txo, but in fuzztarget it's trivially
-                                       // easy to collide the funding_txo hash and have a different scriptPubKey.
-                                       if funding_info.as_ref().unwrap().0 != our_funding_info.as_ref().unwrap().0 {
-                                               return Err(MonitorUpdateError("Funding transaction outputs are not identical!"));
-                                       }
-                               } else {
-                                       return Err(MonitorUpdateError("Try to combine a Local monitor with a Watchtower one !"));
-                               }
-                       },
-                       Storage::Watchtower { .. } => {
-                               if let Storage::Watchtower { .. } = other.key_storage {
-                                       unimplemented!();
-                               } else {
-                                       return Err(MonitorUpdateError("Try to combine a Watchtower monitor with a Local one !"));
-                               }
-                       },
-               }
-               let other_min_secret = other.get_min_seen_secret();
-               let our_min_secret = self.get_min_seen_secret();
-               if our_min_secret > other_min_secret {
-                       self.provide_secret(other_min_secret, other.get_secret(other_min_secret).unwrap())?;
-               }
-               if let Some(ref local_tx) = self.current_local_signed_commitment_tx {
-                       if let Some(ref other_local_tx) = other.current_local_signed_commitment_tx {
-                               let our_commitment_number = 0xffffffffffff - ((((local_tx.tx.without_valid_witness().input[0].sequence as u64 & 0xffffff) << 3*8) | (local_tx.tx.without_valid_witness().lock_time as u64 & 0xffffff)) ^ self.commitment_transaction_number_obscure_factor);
-                               let other_commitment_number = 0xffffffffffff - ((((other_local_tx.tx.without_valid_witness().input[0].sequence as u64 & 0xffffff) << 3*8) | (other_local_tx.tx.without_valid_witness().lock_time as u64 & 0xffffff)) ^ other.commitment_transaction_number_obscure_factor);
-                               if our_commitment_number >= other_commitment_number {
-                                       self.key_storage = other.key_storage;
-                               }
-                       }
-               }
-               // TODO: We should use current_remote_commitment_number and the commitment number out of
-               // local transactions to decide how to merge
-               if our_min_secret >= other_min_secret {
-                       self.their_cur_revocation_points = other.their_cur_revocation_points;
-                       for (txid, htlcs) in other.remote_claimable_outpoints.drain() {
-                               self.remote_claimable_outpoints.insert(txid, htlcs);
-                       }
-                       if let Some(local_tx) = other.prev_local_signed_commitment_tx {
-                               self.prev_local_signed_commitment_tx = Some(local_tx);
-                       }
-                       if let Some(local_tx) = other.current_local_signed_commitment_tx {
-                               self.current_local_signed_commitment_tx = Some(local_tx);
-                       }
-                       self.payment_preimages = other.payment_preimages;
-                       self.to_remote_rescue = other.to_remote_rescue;
-               }
-
-               self.current_remote_commitment_number = cmp::min(self.current_remote_commitment_number, other.current_remote_commitment_number);
-               Ok(())
-       }
-
-       /// Allows this monitor to scan only for transactions which are applicable. Note that this is
-       /// optional, without it this monitor cannot be used in an SPV client, but you may wish to
-       /// avoid this on a monitor you wish to send to a watchtower as it provides slightly better
-       /// privacy.
-       /// It's the responsibility of the caller to register outpoint and script with passing the former
-       /// value as key to add_update_monitor.
-       pub(super) fn set_funding_info(&mut self, new_funding_info: (OutPoint, Script)) {
-               match self.key_storage {
-                       Storage::Local { ref mut funding_info, .. } => {
-                               *funding_info = Some(new_funding_info);
-                       },
-                       Storage::Watchtower { .. } => {
-                               panic!("Channel somehow ended up with its internal ChannelMonitor being in Watchtower mode?");
-                       }
-               }
-       }
-
-       /// We log these base keys at channel opening to being able to rebuild redeemscript in case of leaked revoked commit tx
-       /// Panics if commitment_transaction_number_obscure_factor doesn't fit in 48 bits
-       pub(super) fn set_basic_channel_info(&mut self, their_htlc_base_key: &PublicKey, their_delayed_payment_base_key: &PublicKey, their_to_self_delay: u16, funding_redeemscript: Script, channel_value_satoshis: u64, commitment_transaction_number_obscure_factor: u64) {
-               self.their_htlc_base_key = Some(their_htlc_base_key.clone());
-               self.their_delayed_payment_base_key = Some(their_delayed_payment_base_key.clone());
-               self.their_to_self_delay = Some(their_to_self_delay);
-               self.funding_redeemscript = Some(funding_redeemscript);
-               self.channel_value_satoshis = Some(channel_value_satoshis);
-               assert!(commitment_transaction_number_obscure_factor < (1 << 48));
-               self.commitment_transaction_number_obscure_factor = commitment_transaction_number_obscure_factor;
-       }
-
        /// Gets the update_id from the latest ChannelMonitorUpdate which was applied to this
        /// ChannelMonitor.
        pub fn get_latest_update_id(&self) -> u64 {
@@ -1637,6 +1564,18 @@ impl<ChanSigner: ChannelKeys> ChannelMonitor<ChanSigner> {
                ret
        }
 
+       /// Gets the list of pending events which were generated by previous actions, clearing the list
+       /// in the process.
+       ///
+       /// This is called by ManyChannelMonitor::get_and_clear_pending_events() and is equivalent to
+       /// EventsProvider::get_and_clear_pending_events() except that it requires &mut self as we do
+       /// no internal locking in ChannelMonitors.
+       pub fn get_and_clear_pending_events(&mut self) -> Vec<events::Event> {
+               let mut ret = Vec::new();
+               mem::swap(&mut ret, &mut self.pending_events);
+               ret
+       }
+
        /// Can only fail if idx is < get_min_seen_secret
        pub(super) fn get_secret(&self, idx: u64) -> Option<[u8; 32]> {
                self.commitment_secrets.get_secret(idx)
@@ -1662,7 +1601,9 @@ impl<ChanSigner: ChannelKeys> ChannelMonitor<ChanSigner> {
        /// HTLC-Success/HTLC-Timeout transactions.
        /// Return updates for HTLC pending in the channel and failed automatically by the broadcast of
        /// revoked remote commitment tx
-       fn check_spend_remote_transaction(&mut self, tx: &Transaction, height: u32, fee_estimator: &FeeEstimator) -> (Vec<Transaction>, (Sha256dHash, Vec<TxOut>), Vec<SpendableOutputDescriptor>) {
+       fn check_spend_remote_transaction<F: Deref>(&mut self, tx: &Transaction, height: u32, fee_estimator: F) -> (Vec<Transaction>, (Sha256dHash, Vec<TxOut>), Vec<SpendableOutputDescriptor>)
+               where F::Target: FeeEstimator
+       {
                // Most secp and related errors trying to create keys means we have no hope of constructing
                // a spend transaction...so we return no transactions to broadcast
                let mut txn_to_broadcast = Vec::new();
@@ -2237,7 +2178,9 @@ impl<ChanSigner: ChannelKeys> ChannelMonitor<ChanSigner> {
        }
 
        /// Attempts to claim a remote HTLC-Success/HTLC-Timeout's outputs using the revocation key
-       fn check_spend_remote_htlc(&mut self, tx: &Transaction, commitment_number: u64, height: u32, fee_estimator: &FeeEstimator) -> (Option<Transaction>, Option<SpendableOutputDescriptor>) {
+       fn check_spend_remote_htlc<F: Deref>(&mut self, tx: &Transaction, commitment_number: u64, height: u32, fee_estimator: F) -> (Option<Transaction>, Option<SpendableOutputDescriptor>)
+               where F::Target: FeeEstimator
+       {
                //TODO: send back new outputs to guarantee pending_claim_request consistency
                if tx.input.len() != 1 || tx.output.len() != 1 {
                        return (None, None)
@@ -2607,8 +2550,9 @@ impl<ChanSigner: ChannelKeys> ChannelMonitor<ChanSigner> {
        /// Eventually this should be pub and, roughly, implement ChainListener, however this requires
        /// &mut self, as well as returns new spendable outputs and outpoints to watch for spending of
        /// on-chain.
-       fn block_connected<B: Deref>(&mut self, txn_matched: &[&Transaction], height: u32, block_hash: &Sha256dHash, broadcaster: B, fee_estimator: &FeeEstimator)-> (Vec<(Sha256dHash, Vec<TxOut>)>, Vec<SpendableOutputDescriptor>)
-               where B::Target: BroadcasterInterface
+       fn block_connected<B: Deref, F: Deref>(&mut self, txn_matched: &[&Transaction], height: u32, block_hash: &Sha256dHash, broadcaster: B, fee_estimator: F)-> Vec<(Sha256dHash, Vec<TxOut>)>
+               where B::Target: BroadcasterInterface,
+                     F::Target: FeeEstimator
        {
                for tx in txn_matched {
                        let mut output_val = 0;
@@ -2641,7 +2585,7 @@ impl<ChanSigner: ChannelKeys> ChannelMonitor<ChanSigner> {
                                };
                                if funding_txo.is_none() || (prevout.txid == funding_txo.as_ref().unwrap().0.txid && prevout.vout == funding_txo.as_ref().unwrap().0.index as u32) {
                                        if (tx.input[0].sequence >> 8*3) as u8 == 0x80 && (tx.lock_time >> 8*3) as u8 == 0x20 {
-                                               let (remote_txn, new_outputs, mut spendable_output) = self.check_spend_remote_transaction(&tx, height, fee_estimator);
+                                               let (remote_txn, new_outputs, mut spendable_output) = self.check_spend_remote_transaction(&tx, height, &*fee_estimator);
                                                txn = remote_txn;
                                                spendable_outputs.append(&mut spendable_output);
                                                if !new_outputs.1.is_empty() {
@@ -2663,7 +2607,7 @@ impl<ChanSigner: ChannelKeys> ChannelMonitor<ChanSigner> {
                                        }
                                } else {
                                        if let Some(&(commitment_number, _)) = self.remote_commitment_txn_on_chain.get(&prevout.txid) {
-                                               let (tx, spendable_output) = self.check_spend_remote_htlc(&tx, commitment_number, height, fee_estimator);
+                                               let (tx, spendable_output) = self.check_spend_remote_htlc(&tx, commitment_number, height, &*fee_estimator);
                                                if let Some(tx) = tx {
                                                        txn.push(tx);
                                                }
@@ -2823,7 +2767,7 @@ impl<ChanSigner: ChannelKeys> ChannelMonitor<ChanSigner> {
                for first_claim_txid in bump_candidates.iter() {
                        if let Some((new_timer, new_feerate)) = {
                                if let Some(claim_material) = self.pending_claim_requests.get(first_claim_txid) {
-                                       if let Some((new_timer, new_feerate, bump_tx)) = self.bump_claim_tx(height, &claim_material, fee_estimator) {
+                                       if let Some((new_timer, new_feerate, bump_tx)) = self.bump_claim_tx(height, &claim_material, &*fee_estimator) {
                                                broadcaster.broadcast_transaction(&bump_tx);
                                                Some((new_timer, new_feerate))
                                        } else { None }
@@ -2839,11 +2783,19 @@ impl<ChanSigner: ChannelKeys> ChannelMonitor<ChanSigner> {
                for &(ref txid, ref output_scripts) in watch_outputs.iter() {
                        self.outputs_to_watch.insert(txid.clone(), output_scripts.iter().map(|o| o.script_pubkey.clone()).collect());
                }
-               (watch_outputs, spendable_outputs)
+
+               if spendable_outputs.len() > 0 {
+                       self.pending_events.push(events::Event::SpendableOutputs {
+                               outputs: spendable_outputs,
+                       });
+               }
+
+               watch_outputs
        }
 
-       fn block_disconnected<B: Deref>(&mut self, height: u32, block_hash: &Sha256dHash, broadcaster: B, fee_estimator: &FeeEstimator)
-               where B::Target: BroadcasterInterface
+       fn block_disconnected<B: Deref, F: Deref>(&mut self, height: u32, block_hash: &Sha256dHash, broadcaster: B, fee_estimator: F)
+               where B::Target: BroadcasterInterface,
+                     F::Target: FeeEstimator
        {
                log_trace!(self, "Block {} at height {} disconnected", block_hash, height);
                let mut bump_candidates = HashMap::new();
@@ -2869,7 +2821,7 @@ impl<ChanSigner: ChannelKeys> ChannelMonitor<ChanSigner> {
                        }
                }
                for (_, claim_material) in bump_candidates.iter_mut() {
-                       if let Some((new_timer, new_feerate, bump_tx)) = self.bump_claim_tx(height, &claim_material, fee_estimator) {
+                       if let Some((new_timer, new_feerate, bump_tx)) = self.bump_claim_tx(height, &claim_material, &*fee_estimator) {
                                claim_material.height_timer = new_timer;
                                claim_material.feerate_previous = new_feerate;
                                broadcaster.broadcast_transaction(&bump_tx);
@@ -3099,7 +3051,9 @@ impl<ChanSigner: ChannelKeys> ChannelMonitor<ChanSigner> {
 
        /// Lightning security model (i.e being able to redeem/timeout HTLC or penalize coutnerparty onchain) lays on the assumption of claim transactions getting confirmed before timelock expiration
        /// (CSV or CLTV following cases). In case of high-fee spikes, claim tx may stuck in the mempool, so you need to bump its feerate quickly using Replace-By-Fee or Child-Pay-For-Parent.
-       fn bump_claim_tx(&self, height: u32, cached_claim_datas: &ClaimTxBumpMaterial, fee_estimator: &FeeEstimator) -> Option<(u32, u64, Transaction)> {
+       fn bump_claim_tx<F: Deref>(&self, height: u32, cached_claim_datas: &ClaimTxBumpMaterial, fee_estimator: F) -> Option<(u32, u64, Transaction)>
+               where F::Target: FeeEstimator
+       {
                if cached_claim_datas.per_input_material.len() == 0 { return None } // But don't prune pending claiming request yet, we may have to resurrect HTLCs
                let mut inputs = Vec::new();
                for outp in cached_claim_datas.per_input_material.keys() {
@@ -3438,6 +3392,14 @@ impl<R: ::std::io::Read, ChanSigner: ChannelKeys + Readable<R>> ReadableArgs<R,
                        pending_htlcs_updated.push(Readable::read(reader)?);
                }
 
+               let pending_events_len: u64 = Readable::read(reader)?;
+               let mut pending_events = Vec::with_capacity(cmp::min(pending_events_len as usize, MAX_ALLOC_SIZE / mem::size_of::<events::Event>()));
+               for _ in 0..pending_events_len {
+                       if let Some(event) = MaybeReadable::read(reader)? {
+                               pending_events.push(event);
+                       }
+               }
+
                let last_block_hash: Sha256dHash = Readable::read(reader)?;
                let destination_script = Readable::read(reader)?;
                let to_remote_rescue = match <u8 as Readable<R>>::read(reader)? {
@@ -3540,6 +3502,7 @@ impl<R: ::std::io::Read, ChanSigner: ChannelKeys + Readable<R>> ReadableArgs<R,
 
                        payment_preimages,
                        pending_htlcs_updated,
+                       pending_events,
 
                        destination_script,
                        to_remote_rescue,
@@ -3571,6 +3534,7 @@ mod tests {
        use bitcoin_hashes::sha256d::Hash as Sha256dHash;
        use bitcoin_hashes::hex::FromHex;
        use hex;
+       use chain::transaction::OutPoint;
        use ln::channelmanager::{PaymentPreimage, PaymentHash};
        use ln::channelmonitor::{ChannelMonitor, InputDescriptors};
        use ln::chan_utils;
@@ -3663,7 +3627,13 @@ mod tests {
 
                // Prune with one old state and a local commitment tx holding a few overlaps with the
                // old state.
-               let mut monitor = ChannelMonitor::new(keys, &SecretKey::from_slice(&[41; 32]).unwrap(), &SecretKey::from_slice(&[42; 32]).unwrap(), &SecretKey::from_slice(&[43; 32]).unwrap(), &SecretKey::from_slice(&[44; 32]).unwrap(), &SecretKey::from_slice(&[44; 32]).unwrap(), &PublicKey::from_secret_key(&secp_ctx, &SecretKey::from_slice(&[45; 32]).unwrap()), 0, Script::new(), logger.clone());
+               let mut monitor = ChannelMonitor::new(keys,
+                       &PublicKey::from_secret_key(&secp_ctx, &SecretKey::from_slice(&[42; 32]).unwrap()), 0, &Script::new(),
+                       (OutPoint { txid: Sha256dHash::from_slice(&[43; 32]).unwrap(), index: 0 }, Script::new()),
+                       &PublicKey::from_secret_key(&secp_ctx, &SecretKey::from_slice(&[44; 32]).unwrap()),
+                       &PublicKey::from_secret_key(&secp_ctx, &SecretKey::from_slice(&[45; 32]).unwrap()),
+                       0, Script::new(), 46, 0, logger.clone());
+
                monitor.their_to_self_delay = Some(10);
 
                monitor.provide_latest_local_commitment_tx_info(LocalCommitmentTransaction::dummy(), dummy_keys!(), 0, preimages_to_local_htlcs!(preimages[0..10])).unwrap();