Expose the historical success probability calculation itself
[rust-lightning] / lightning / src / ln / channel.rs
index 6f78c71163cfeeb2f616669af8e77bb7863bc82b..11f0261d677a3df37b2fa36f94ea142e5ce8f16c 100644 (file)
@@ -25,7 +25,7 @@ use bitcoin::secp256k1;
 use crate::ln::{PaymentPreimage, PaymentHash};
 use crate::ln::features::{ChannelTypeFeatures, InitFeatures};
 use crate::ln::msgs;
-use crate::ln::msgs::{DecodeError, OptionalField, DataLossProtect};
+use crate::ln::msgs::DecodeError;
 use crate::ln::script::{self, ShutdownScript};
 use crate::ln::channelmanager::{self, CounterpartyForwardingInfo, PendingHTLCStatus, HTLCSource, SentHTLCId, HTLCFailureMsg, PendingHTLCInfo, RAACommitmentOrder, BREAKDOWN_TIMEOUT, MIN_CLTV_EXPIRY_DELTA, MAX_LOCAL_BREAKDOWN_TIMEOUT};
 use crate::ln::chan_utils::{CounterpartyCommitmentSecrets, TxCreationKeys, HTLCOutputInCommitment, htlc_success_tx_weight, htlc_timeout_tx_weight, make_funding_redeemscript, ChannelPublicKeys, CommitmentTransaction, HolderCommitmentTransaction, ChannelTransactionParameters, CounterpartyChannelTransactionParameters, MAX_HTLCS, get_commitment_transaction_number_obscure_factor, ClosingTransaction};
@@ -35,7 +35,7 @@ use crate::chain::BestBlock;
 use crate::chain::chaininterface::{FeeEstimator, ConfirmationTarget, LowerBoundedFeeEstimator};
 use crate::chain::channelmonitor::{ChannelMonitor, ChannelMonitorUpdate, ChannelMonitorUpdateStep, LATENCY_GRACE_PERIOD_BLOCKS, CLOSED_CHANNEL_UPDATE_ID};
 use crate::chain::transaction::{OutPoint, TransactionData};
-use crate::chain::keysinterface::{WriteableEcdsaChannelSigner, EntropySource, ChannelSigner, SignerProvider, NodeSigner, Recipient};
+use crate::sign::{WriteableEcdsaChannelSigner, EntropySource, ChannelSigner, SignerProvider, NodeSigner, Recipient};
 use crate::events::ClosureReason;
 use crate::routing::gossip::NodeId;
 use crate::util::ser::{Readable, ReadableArgs, Writeable, Writer, VecWriter};
@@ -479,6 +479,21 @@ pub(crate) const MIN_AFFORDABLE_HTLC_COUNT: usize = 4;
 ///   * `EXPIRE_PREV_CONFIG_TICKS` = convergence_delay / tick_interval
 pub(crate) const EXPIRE_PREV_CONFIG_TICKS: usize = 5;
 
+struct PendingChannelMonitorUpdate {
+       update: ChannelMonitorUpdate,
+       /// In some cases we need to delay letting the [`ChannelMonitorUpdate`] go until after an
+       /// `Event` is processed by the user. This bool indicates the [`ChannelMonitorUpdate`] is
+       /// blocked on some external event and the [`ChannelManager`] will update us when we're ready.
+       ///
+       /// [`ChannelManager`]: super::channelmanager::ChannelManager
+       blocked: bool,
+}
+
+impl_writeable_tlv_based!(PendingChannelMonitorUpdate, {
+       (0, update, required),
+       (2, blocked, required),
+});
+
 // TODO: We should refactor this to be an Inbound/OutboundChannel until initial setup handshaking
 // has been completed, and then turn into a Channel to get compiler-time enforcement of things like
 // calling channel_id() before we're set up or things like get_outbound_funding_signed on an
@@ -744,7 +759,7 @@ pub(super) struct Channel<Signer: ChannelSigner> {
        /// If we then persist the [`channelmanager::ChannelManager`] and crash before the persistence
        /// completes we still need to be able to complete the persistence. Thus, we have to keep a
        /// copy of the [`ChannelMonitorUpdate`] here until it is complete.
-       pending_monitor_updates: Vec<ChannelMonitorUpdate>,
+       pending_monitor_updates: Vec<PendingChannelMonitorUpdate>,
 }
 
 #[cfg(any(test, fuzzing))]
@@ -1322,7 +1337,7 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
 
                let counterparty_shutdown_scriptpubkey = if their_features.supports_upfront_shutdown_script() {
                        match &msg.shutdown_scriptpubkey {
-                               &OptionalField::Present(ref script) => {
+                               &Some(ref script) => {
                                        // Peer is signaling upfront_shutdown and has opt-out with a 0-length script. We don't enforce anything
                                        if script.len() == 0 {
                                                None
@@ -1334,7 +1349,7 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
                                        }
                                },
                                // Peer is signaling upfront shutdown but don't opt-out with correct mechanism (a.k.a 0-length script). Peer looks buggy, we fail the channel
-                               &OptionalField::Absent => {
+                               &None => {
                                        return Err(ChannelError::Close("Peer is signaling upfront_shutdown but we don't get any script. Use 0-length script to opt-out".to_owned()));
                                }
                        }
@@ -1995,28 +2010,52 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
        }
 
        pub fn get_update_fulfill_htlc_and_commit<L: Deref>(&mut self, htlc_id: u64, payment_preimage: PaymentPreimage, logger: &L) -> UpdateFulfillCommitFetch where L::Target: Logger {
+               let release_cs_monitor = self.pending_monitor_updates.iter().all(|upd| !upd.blocked);
                match self.get_update_fulfill_htlc(htlc_id, payment_preimage, logger) {
-                       UpdateFulfillFetch::NewClaim { mut monitor_update, htlc_value_msat, msg: Some(_) } => {
-                               let mut additional_update = self.build_commitment_no_status_check(logger);
-                               // build_commitment_no_status_check may bump latest_monitor_id but we want them to be
-                               // strictly increasing by one, so decrement it here.
-                               self.latest_monitor_update_id = monitor_update.update_id;
-                               monitor_update.updates.append(&mut additional_update.updates);
-                               self.monitor_updating_paused(false, true, false, Vec::new(), Vec::new(), Vec::new());
-                               self.pending_monitor_updates.push(monitor_update);
+                       UpdateFulfillFetch::NewClaim { mut monitor_update, htlc_value_msat, msg } => {
+                               // Even if we aren't supposed to let new monitor updates with commitment state
+                               // updates run, we still need to push the preimage ChannelMonitorUpdateStep no
+                               // matter what. Sadly, to push a new monitor update which flies before others
+                               // already queued, we have to insert it into the pending queue and update the
+                               // update_ids of all the following monitors.
+                               let unblocked_update_pos = if release_cs_monitor && msg.is_some() {
+                                       let mut additional_update = self.build_commitment_no_status_check(logger);
+                                       // build_commitment_no_status_check may bump latest_monitor_id but we want them
+                                       // to be strictly increasing by one, so decrement it here.
+                                       self.latest_monitor_update_id = monitor_update.update_id;
+                                       monitor_update.updates.append(&mut additional_update.updates);
+                                       self.pending_monitor_updates.push(PendingChannelMonitorUpdate {
+                                               update: monitor_update, blocked: false,
+                                       });
+                                       self.pending_monitor_updates.len() - 1
+                               } else {
+                                       let insert_pos = self.pending_monitor_updates.iter().position(|upd| upd.blocked)
+                                               .unwrap_or(self.pending_monitor_updates.len());
+                                       let new_mon_id = self.pending_monitor_updates.get(insert_pos)
+                                               .map(|upd| upd.update.update_id).unwrap_or(monitor_update.update_id);
+                                       monitor_update.update_id = new_mon_id;
+                                       self.pending_monitor_updates.insert(insert_pos, PendingChannelMonitorUpdate {
+                                               update: monitor_update, blocked: false,
+                                       });
+                                       for held_update in self.pending_monitor_updates.iter_mut().skip(insert_pos + 1) {
+                                               held_update.update.update_id += 1;
+                                       }
+                                       if msg.is_some() {
+                                               debug_assert!(false, "If there is a pending blocked monitor we should have MonitorUpdateInProgress set");
+                                               let update = self.build_commitment_no_status_check(logger);
+                                               self.pending_monitor_updates.push(PendingChannelMonitorUpdate {
+                                                       update, blocked: true,
+                                               });
+                                       }
+                                       insert_pos
+                               };
+                               self.monitor_updating_paused(false, msg.is_some(), false, Vec::new(), Vec::new(), Vec::new());
                                UpdateFulfillCommitFetch::NewClaim {
-                                       monitor_update: self.pending_monitor_updates.last().unwrap(),
+                                       monitor_update: &self.pending_monitor_updates.get(unblocked_update_pos)
+                                               .expect("We just pushed the monitor update").update,
                                        htlc_value_msat,
                                }
                        },
-                       UpdateFulfillFetch::NewClaim { monitor_update, htlc_value_msat, msg: None } => {
-                               self.monitor_updating_paused(false, false, false, Vec::new(), Vec::new(), Vec::new());
-                               self.pending_monitor_updates.push(monitor_update);
-                               UpdateFulfillCommitFetch::NewClaim {
-                                       monitor_update: self.pending_monitor_updates.last().unwrap(),
-                                       htlc_value_msat,
-                               }
-                       }
                        UpdateFulfillFetch::DuplicateClaim {} => UpdateFulfillCommitFetch::DuplicateClaim {},
                }
        }
@@ -2207,7 +2246,7 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
 
                let counterparty_shutdown_scriptpubkey = if their_features.supports_upfront_shutdown_script() {
                        match &msg.shutdown_scriptpubkey {
-                               &OptionalField::Present(ref script) => {
+                               &Some(ref script) => {
                                        // Peer is signaling upfront_shutdown and has opt-out with a 0-length script. We don't enforce anything
                                        if script.len() == 0 {
                                                None
@@ -2219,7 +2258,7 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
                                        }
                                },
                                // Peer is signaling upfront shutdown but don't opt-out with correct mechanism (a.k.a 0-length script). Peer looks buggy, we fail the channel
-                               &OptionalField::Absent => {
+                               &None => {
                                        return Err(ChannelError::Close("Peer is signaling upfront_shutdown but we don't get any script. Use 0-length script to opt-out".to_owned()));
                                }
                        }
@@ -3084,7 +3123,7 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
                Ok(())
        }
 
-       pub fn commitment_signed<L: Deref>(&mut self, msg: &msgs::CommitmentSigned, logger: &L) -> Result<&ChannelMonitorUpdate, ChannelError>
+       pub fn commitment_signed<L: Deref>(&mut self, msg: &msgs::CommitmentSigned, logger: &L) -> Result<Option<&ChannelMonitorUpdate>, ChannelError>
                where L::Target: Logger
        {
                if (self.channel_state & (ChannelState::ChannelReady as u32)) != (ChannelState::ChannelReady as u32) {
@@ -3284,8 +3323,7 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
                        }
                        log_debug!(logger, "Received valid commitment_signed from peer in channel {}, updated HTLC state but awaiting a monitor update resolution to reply.",
                                log_bytes!(self.channel_id));
-                       self.pending_monitor_updates.push(monitor_update);
-                       return Ok(self.pending_monitor_updates.last().unwrap());
+                       return Ok(self.push_ret_blockable_mon_update(monitor_update));
                }
 
                let need_commitment_signed = if need_commitment && (self.channel_state & (ChannelState::AwaitingRemoteRevoke as u32)) == 0 {
@@ -3302,9 +3340,8 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
 
                log_debug!(logger, "Received valid commitment_signed from peer in channel {}, updating HTLC state and responding with{} a revoke_and_ack.",
                        log_bytes!(self.channel_id()), if need_commitment_signed { " our own commitment_signed and" } else { "" });
-               self.pending_monitor_updates.push(monitor_update);
                self.monitor_updating_paused(true, need_commitment_signed, false, Vec::new(), Vec::new(), Vec::new());
-               return Ok(self.pending_monitor_updates.last().unwrap());
+               return Ok(self.push_ret_blockable_mon_update(monitor_update));
        }
 
        /// Public version of the below, checking relevant preconditions first.
@@ -3419,8 +3456,7 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
                                update_add_htlcs.len(), update_fulfill_htlcs.len(), update_fail_htlcs.len());
 
                        self.monitor_updating_paused(false, true, false, Vec::new(), Vec::new(), Vec::new());
-                       self.pending_monitor_updates.push(monitor_update);
-                       (Some(self.pending_monitor_updates.last().unwrap()), htlcs_to_fail)
+                       (self.push_ret_blockable_mon_update(monitor_update), htlcs_to_fail)
                } else {
                        (None, Vec::new())
                }
@@ -3431,7 +3467,7 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
        /// waiting on this revoke_and_ack. The generation of this new commitment_signed may also fail,
        /// generating an appropriate error *after* the channel state has been updated based on the
        /// revoke_and_ack message.
-       pub fn revoke_and_ack<L: Deref>(&mut self, msg: &msgs::RevokeAndACK, logger: &L) -> Result<(Vec<(HTLCSource, PaymentHash)>, &ChannelMonitorUpdate), ChannelError>
+       pub fn revoke_and_ack<L: Deref>(&mut self, msg: &msgs::RevokeAndACK, logger: &L) -> Result<(Vec<(HTLCSource, PaymentHash)>, Option<&ChannelMonitorUpdate>), ChannelError>
                where L::Target: Logger,
        {
                if (self.channel_state & (ChannelState::ChannelReady as u32)) != (ChannelState::ChannelReady as u32) {
@@ -3628,21 +3664,19 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
                        self.monitor_pending_failures.append(&mut revoked_htlcs);
                        self.monitor_pending_finalized_fulfills.append(&mut finalized_claimed_htlcs);
                        log_debug!(logger, "Received a valid revoke_and_ack for channel {} but awaiting a monitor update resolution to reply.", log_bytes!(self.channel_id()));
-                       self.pending_monitor_updates.push(monitor_update);
-                       return Ok((Vec::new(), self.pending_monitor_updates.last().unwrap()));
+                       return Ok((Vec::new(), self.push_ret_blockable_mon_update(monitor_update)));
                }
 
                match self.free_holding_cell_htlcs(logger) {
                        (Some(_), htlcs_to_fail) => {
-                               let mut additional_update = self.pending_monitor_updates.pop().unwrap();
+                               let mut additional_update = self.pending_monitor_updates.pop().unwrap().update;
                                // free_holding_cell_htlcs may bump latest_monitor_id multiple times but we want them to be
                                // strictly increasing by one, so decrement it here.
                                self.latest_monitor_update_id = monitor_update.update_id;
                                monitor_update.updates.append(&mut additional_update.updates);
 
                                self.monitor_updating_paused(false, true, false, to_forward_infos, revoked_htlcs, finalized_claimed_htlcs);
-                               self.pending_monitor_updates.push(monitor_update);
-                               Ok((htlcs_to_fail, self.pending_monitor_updates.last().unwrap()))
+                               Ok((htlcs_to_fail, self.push_ret_blockable_mon_update(monitor_update)))
                        },
                        (None, htlcs_to_fail) => {
                                if require_commitment {
@@ -3656,13 +3690,11 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
                                        log_debug!(logger, "Received a valid revoke_and_ack for channel {}. Responding with a commitment update with {} HTLCs failed.",
                                                log_bytes!(self.channel_id()), update_fail_htlcs.len() + update_fail_malformed_htlcs.len());
                                        self.monitor_updating_paused(false, true, false, to_forward_infos, revoked_htlcs, finalized_claimed_htlcs);
-                                       self.pending_monitor_updates.push(monitor_update);
-                                       Ok((htlcs_to_fail, self.pending_monitor_updates.last().unwrap()))
+                                       Ok((htlcs_to_fail, self.push_ret_blockable_mon_update(monitor_update)))
                                } else {
                                        log_debug!(logger, "Received a valid revoke_and_ack for channel {} with no reply necessary.", log_bytes!(self.channel_id()));
                                        self.monitor_updating_paused(false, false, false, to_forward_infos, revoked_htlcs, finalized_claimed_htlcs);
-                                       self.pending_monitor_updates.push(monitor_update);
-                                       Ok((htlcs_to_fail, self.pending_monitor_updates.last().unwrap()))
+                                       Ok((htlcs_to_fail, self.push_ret_blockable_mon_update(monitor_update)))
                                }
                        }
                }
@@ -3851,7 +3883,12 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
        {
                assert_eq!(self.channel_state & ChannelState::MonitorUpdateInProgress as u32, ChannelState::MonitorUpdateInProgress as u32);
                self.channel_state &= !(ChannelState::MonitorUpdateInProgress as u32);
-               self.pending_monitor_updates.clear();
+               let mut found_blocked = false;
+               self.pending_monitor_updates.retain(|upd| {
+                       if found_blocked { debug_assert!(upd.blocked, "No mons may be unblocked after a blocked one"); }
+                       if upd.blocked { found_blocked = true; }
+                       upd.blocked
+               });
 
                // If we're past (or at) the FundingSent stage on an outbound channel, try to
                // (re-)broadcast the funding transaction as we may have declined to broadcast it when we
@@ -4055,36 +4092,31 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
 
                if msg.next_local_commitment_number >= INITIAL_COMMITMENT_NUMBER || msg.next_remote_commitment_number >= INITIAL_COMMITMENT_NUMBER ||
                        msg.next_local_commitment_number == 0 {
-                       return Err(ChannelError::Close("Peer sent a garbage channel_reestablish".to_owned()));
+                       return Err(ChannelError::Close("Peer sent a garbage channel_reestablish (usually an lnd node with lost state asking us to force-close for them)".to_owned()));
                }
 
                if msg.next_remote_commitment_number > 0 {
-                       match msg.data_loss_protect {
-                               OptionalField::Present(ref data_loss) => {
-                                       let expected_point = self.holder_signer.get_per_commitment_point(INITIAL_COMMITMENT_NUMBER - msg.next_remote_commitment_number + 1, &self.secp_ctx);
-                                       let given_secret = SecretKey::from_slice(&data_loss.your_last_per_commitment_secret)
-                                               .map_err(|_| ChannelError::Close("Peer sent a garbage channel_reestablish with unparseable secret key".to_owned()))?;
-                                       if expected_point != PublicKey::from_secret_key(&self.secp_ctx, &given_secret) {
-                                               return Err(ChannelError::Close("Peer sent a garbage channel_reestablish with secret key not matching the commitment height provided".to_owned()));
+                       let expected_point = self.holder_signer.get_per_commitment_point(INITIAL_COMMITMENT_NUMBER - msg.next_remote_commitment_number + 1, &self.secp_ctx);
+                       let given_secret = SecretKey::from_slice(&msg.your_last_per_commitment_secret)
+                               .map_err(|_| ChannelError::Close("Peer sent a garbage channel_reestablish with unparseable secret key".to_owned()))?;
+                       if expected_point != PublicKey::from_secret_key(&self.secp_ctx, &given_secret) {
+                               return Err(ChannelError::Close("Peer sent a garbage channel_reestablish with secret key not matching the commitment height provided".to_owned()));
+                       }
+                       if msg.next_remote_commitment_number > INITIAL_COMMITMENT_NUMBER - self.cur_holder_commitment_transaction_number {
+                               macro_rules! log_and_panic {
+                                       ($err_msg: expr) => {
+                                               log_error!(logger, $err_msg, log_bytes!(self.channel_id), log_pubkey!(self.counterparty_node_id));
+                                               panic!($err_msg, log_bytes!(self.channel_id), log_pubkey!(self.counterparty_node_id));
                                        }
-                                       if msg.next_remote_commitment_number > INITIAL_COMMITMENT_NUMBER - self.cur_holder_commitment_transaction_number {
-                                               macro_rules! log_and_panic {
-                                                       ($err_msg: expr) => {
-                                                               log_error!(logger, $err_msg, log_bytes!(self.channel_id), log_pubkey!(self.counterparty_node_id));
-                                                               panic!($err_msg, log_bytes!(self.channel_id), log_pubkey!(self.counterparty_node_id));
-                                                       }
-                                               }
-                                               log_and_panic!("We have fallen behind - we have received proof that if we broadcast our counterparty is going to claim all our funds.\n\
-                                                       This implies you have restarted with lost ChannelMonitor and ChannelManager state, the first of which is a violation of the LDK chain::Watch requirements.\n\
-                                                       More specifically, this means you have a bug in your implementation that can cause loss of funds, or you are running with an old backup, which is unsafe.\n\
-                                                       If you have restored from an old backup and wish to force-close channels and return to operation, you should start up, call\n\
-                                                       ChannelManager::force_close_without_broadcasting_txn on channel {} with counterparty {} or\n\
-                                                       ChannelManager::force_close_all_channels_without_broadcasting_txn, then reconnect to peer(s).\n\
-                                                       Note that due to a long-standing bug in lnd you may have to reach out to peers running lnd-based nodes to ask them to manually force-close channels\n\
-                                                       See https://github.com/lightningdevkit/rust-lightning/issues/1565 for more info.");
-                                       }
-                               },
-                               OptionalField::Absent => {}
+                               }
+                               log_and_panic!("We have fallen behind - we have received proof that if we broadcast our counterparty is going to claim all our funds.\n\
+                                       This implies you have restarted with lost ChannelMonitor and ChannelManager state, the first of which is a violation of the LDK chain::Watch requirements.\n\
+                                       More specifically, this means you have a bug in your implementation that can cause loss of funds, or you are running with an old backup, which is unsafe.\n\
+                                       If you have restored from an old backup and wish to force-close channels and return to operation, you should start up, call\n\
+                                       ChannelManager::force_close_without_broadcasting_txn on channel {} with counterparty {} or\n\
+                                       ChannelManager::force_close_all_channels_without_broadcasting_txn, then reconnect to peer(s).\n\
+                                       Note that due to a long-standing bug in lnd you may have to reach out to peers running lnd-based nodes to ask them to manually force-close channels\n\
+                                       See https://github.com/lightningdevkit/rust-lightning/issues/1565 for more info.");
                        }
                }
 
@@ -4397,8 +4429,9 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
                                }],
                        };
                        self.monitor_updating_paused(false, false, false, Vec::new(), Vec::new(), Vec::new());
-                       self.pending_monitor_updates.push(monitor_update);
-                       Some(self.pending_monitor_updates.last().unwrap())
+                       if self.push_blockable_mon_update(monitor_update) {
+                               self.pending_monitor_updates.last().map(|upd| &upd.update)
+                       } else { None }
                } else { None };
                let shutdown = if send_shutdown {
                        Some(msgs::Shutdown {
@@ -4970,8 +5003,49 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
                (self.channel_state & ChannelState::MonitorUpdateInProgress as u32) != 0
        }
 
-       pub fn get_next_monitor_update(&self) -> Option<&ChannelMonitorUpdate> {
-               self.pending_monitor_updates.first()
+       pub fn get_latest_complete_monitor_update_id(&self) -> u64 {
+               if self.pending_monitor_updates.is_empty() { return self.get_latest_monitor_update_id(); }
+               self.pending_monitor_updates[0].update.update_id - 1
+       }
+
+       /// Returns the next blocked monitor update, if one exists, and a bool which indicates a
+       /// further blocked monitor update exists after the next.
+       pub fn unblock_next_blocked_monitor_update(&mut self) -> Option<(&ChannelMonitorUpdate, bool)> {
+               for i in 0..self.pending_monitor_updates.len() {
+                       if self.pending_monitor_updates[i].blocked {
+                               self.pending_monitor_updates[i].blocked = false;
+                               return Some((&self.pending_monitor_updates[i].update,
+                                       self.pending_monitor_updates.len() > i + 1));
+                       }
+               }
+               None
+       }
+
+       /// Pushes a new monitor update into our monitor update queue, returning whether it should be
+       /// immediately given to the user for persisting or if it should be held as blocked.
+       fn push_blockable_mon_update(&mut self, update: ChannelMonitorUpdate) -> bool {
+               let release_monitor = self.pending_monitor_updates.iter().all(|upd| !upd.blocked);
+               self.pending_monitor_updates.push(PendingChannelMonitorUpdate {
+                       update, blocked: !release_monitor
+               });
+               release_monitor
+       }
+
+       /// Pushes a new monitor update into our monitor update queue, returning a reference to it if
+       /// it should be immediately given to the user for persisting or `None` if it should be held as
+       /// blocked.
+       fn push_ret_blockable_mon_update(&mut self, update: ChannelMonitorUpdate)
+       -> Option<&ChannelMonitorUpdate> {
+               let release_monitor = self.push_blockable_mon_update(update);
+               if release_monitor { self.pending_monitor_updates.last().map(|upd| &upd.update) } else { None }
+       }
+
+       pub fn no_monitor_updates_pending(&self) -> bool {
+               self.pending_monitor_updates.is_empty()
+       }
+
+       pub fn complete_one_mon_update(&mut self, update_id: u64) {
+               self.pending_monitor_updates.retain(|upd| upd.update.update_id != update_id);
        }
 
        /// Returns true if funding_created was sent/received.
@@ -5342,7 +5416,7 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
                        htlc_basepoint: keys.htlc_basepoint,
                        first_per_commitment_point,
                        channel_flags: if self.config.announced_channel {1} else {0},
-                       shutdown_scriptpubkey: OptionalField::Present(match &self.shutdown_scriptpubkey {
+                       shutdown_scriptpubkey: Some(match &self.shutdown_scriptpubkey {
                                Some(script) => script.clone().into_inner(),
                                None => Builder::new().into_script(),
                        }),
@@ -5408,7 +5482,7 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
                        delayed_payment_basepoint: keys.delayed_payment_basepoint,
                        htlc_basepoint: keys.htlc_basepoint,
                        first_per_commitment_point,
-                       shutdown_scriptpubkey: OptionalField::Present(match &self.shutdown_scriptpubkey {
+                       shutdown_scriptpubkey: Some(match &self.shutdown_scriptpubkey {
                                Some(script) => script.clone().into_inner(),
                                None => Builder::new().into_script(),
                        }),
@@ -5670,19 +5744,13 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
                // valid, and valid in fuzzing mode's arbitrary validity criteria:
                let mut pk = [2; 33]; pk[1] = 0xff;
                let dummy_pubkey = PublicKey::from_slice(&pk).unwrap();
-               let data_loss_protect = if self.cur_counterparty_commitment_transaction_number + 1 < INITIAL_COMMITMENT_NUMBER {
+               let remote_last_secret = if self.cur_counterparty_commitment_transaction_number + 1 < INITIAL_COMMITMENT_NUMBER {
                        let remote_last_secret = self.commitment_secrets.get_secret(self.cur_counterparty_commitment_transaction_number + 2).unwrap();
                        log_trace!(logger, "Enough info to generate a Data Loss Protect with per_commitment_secret {} for channel {}", log_bytes!(remote_last_secret), log_bytes!(self.channel_id()));
-                       OptionalField::Present(DataLossProtect {
-                               your_last_per_commitment_secret: remote_last_secret,
-                               my_current_per_commitment_point: dummy_pubkey
-                       })
+                       remote_last_secret
                } else {
                        log_info!(logger, "Sending a data_loss_protect with no previous remote per_commitment_secret for channel {}", log_bytes!(self.channel_id()));
-                       OptionalField::Present(DataLossProtect {
-                               your_last_per_commitment_secret: [0;32],
-                               my_current_per_commitment_point: dummy_pubkey,
-                       })
+                       [0;32]
                };
                msgs::ChannelReestablish {
                        channel_id: self.channel_id(),
@@ -5704,7 +5772,12 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
                        // dropped this channel on disconnect as it hasn't yet reached FundingSent so we can't
                        // overflow here.
                        next_remote_commitment_number: INITIAL_COMMITMENT_NUMBER - self.cur_counterparty_commitment_transaction_number - 1,
-                       data_loss_protect,
+                       your_last_per_commitment_secret: remote_last_secret,
+                       my_current_per_commitment_point: dummy_pubkey,
+                       // TODO(dual_funding): If we've sent `commtiment_signed` for an interactive transaction
+                       // construction but have not received `tx_signatures` we MUST set `next_funding_txid` to the
+                       // txid of that interactive transaction, else we MUST NOT set it.
+                       next_funding_txid: None,
                }
        }
 
@@ -6019,8 +6092,7 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
                        Some(_) => {
                                let monitor_update = self.build_commitment_no_status_check(logger);
                                self.monitor_updating_paused(false, true, false, Vec::new(), Vec::new(), Vec::new());
-                               self.pending_monitor_updates.push(monitor_update);
-                               Ok(Some(self.pending_monitor_updates.last().unwrap()))
+                               Ok(self.push_ret_blockable_mon_update(monitor_update))
                        },
                        None => Ok(None)
                }
@@ -6050,7 +6122,7 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
        /// May jump to the channel being fully shutdown (see [`Self::is_shutdown`]) in which case no
        /// [`ChannelMonitorUpdate`] will be returned).
        pub fn get_shutdown<SP: Deref>(&mut self, signer_provider: &SP, their_features: &InitFeatures,
-               target_feerate_sats_per_kw: Option<u32>)
+               target_feerate_sats_per_kw: Option<u32>, override_shutdown_script: Option<ShutdownScript>)
        -> Result<(msgs::Shutdown, Option<&ChannelMonitorUpdate>, Vec<(HTLCSource, PaymentHash)>), APIError>
        where SP::Target: SignerProvider {
                for htlc in self.pending_outbound_htlcs.iter() {
@@ -6066,6 +6138,9 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
                                return Err(APIError::ChannelUnavailable{err: "Shutdown already in progress by remote".to_owned()});
                        }
                }
+               if self.shutdown_scriptpubkey.is_some() && override_shutdown_script.is_some() {
+                       return Err(APIError::APIMisuseError{err: "Cannot override shutdown script for a channel with one already set".to_owned()});
+               }
                assert_eq!(self.channel_state & ChannelState::ShutdownComplete as u32, 0);
                if self.channel_state & (ChannelState::PeerDisconnected as u32 | ChannelState::MonitorUpdateInProgress as u32) != 0 {
                        return Err(APIError::ChannelUnavailable{err: "Cannot begin shutdown while peer is disconnected or we're waiting on a monitor update, maybe force-close instead?".to_owned()});
@@ -6081,9 +6156,16 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
                let update_shutdown_script = match self.shutdown_scriptpubkey {
                        Some(_) => false,
                        None if !chan_closed => {
-                               let shutdown_scriptpubkey = match signer_provider.get_shutdown_scriptpubkey() {
-                                       Ok(scriptpubkey) => scriptpubkey,
-                                       Err(_) => return Err(APIError::ChannelUnavailable { err: "Failed to get shutdown scriptpubkey".to_owned() }),
+                               // use override shutdown script if provided
+                               let shutdown_scriptpubkey = match override_shutdown_script {
+                                       Some(script) => script,
+                                       None => {
+                                               // otherwise, use the shutdown scriptpubkey provided by the signer
+                                               match signer_provider.get_shutdown_scriptpubkey() {
+                                                       Ok(scriptpubkey) => scriptpubkey,
+                                                       Err(_) => return Err(APIError::ChannelUnavailable{err: "Failed to get shutdown scriptpubkey".to_owned()}),
+                                               }
+                                       },
                                };
                                if !shutdown_scriptpubkey.is_compatible(their_features) {
                                        return Err(APIError::IncompatibleShutdownScript { script: shutdown_scriptpubkey.clone() });
@@ -6112,8 +6194,9 @@ impl<Signer: WriteableEcdsaChannelSigner> Channel<Signer> {
                                }],
                        };
                        self.monitor_updating_paused(false, false, false, Vec::new(), Vec::new(), Vec::new());
-                       self.pending_monitor_updates.push(monitor_update);
-                       Some(self.pending_monitor_updates.last().unwrap())
+                       if self.push_blockable_mon_update(monitor_update) {
+                               self.pending_monitor_updates.last().map(|upd| &upd.update)
+                       } else { None }
                } else { None };
                let shutdown = msgs::Shutdown {
                        channel_id: self.channel_id,
@@ -6550,6 +6633,7 @@ impl<Signer: WriteableEcdsaChannelSigner> Writeable for Channel<Signer> {
                        (28, holder_max_accepted_htlcs, option),
                        (29, self.temporary_channel_id, option),
                        (31, channel_pending_event_emitted, option),
+                       (33, self.pending_monitor_updates, vec_type),
                });
 
                Ok(())
@@ -6826,6 +6910,8 @@ impl<'a, 'b, 'c, ES: Deref, SP: Deref> ReadableArgs<(&'a ES, &'b SP, u32, &'c Ch
                let mut temporary_channel_id: Option<[u8; 32]> = None;
                let mut holder_max_accepted_htlcs: Option<u16> = None;
 
+               let mut pending_monitor_updates = Some(Vec::new());
+
                read_tlv_fields!(reader, {
                        (0, announcement_sigs, option),
                        (1, minimum_depth, option),
@@ -6848,6 +6934,7 @@ impl<'a, 'b, 'c, ES: Deref, SP: Deref> ReadableArgs<(&'a ES, &'b SP, u32, &'c Ch
                        (28, holder_max_accepted_htlcs, option),
                        (29, temporary_channel_id, option),
                        (31, channel_pending_event_emitted, option),
+                       (33, pending_monitor_updates, vec_type),
                });
 
                let (channel_keys_id, holder_signer) = if let Some(channel_keys_id) = channel_keys_id {
@@ -7017,7 +7104,7 @@ impl<'a, 'b, 'c, ES: Deref, SP: Deref> ReadableArgs<(&'a ES, &'b SP, u32, &'c Ch
                        channel_type: channel_type.unwrap(),
                        channel_keys_id,
 
-                       pending_monitor_updates: Vec::new(),
+                       pending_monitor_updates: pending_monitor_updates.unwrap(),
                })
        }
 }
@@ -7038,13 +7125,13 @@ mod tests {
        use crate::ln::channel::{Channel, InboundHTLCOutput, OutboundHTLCOutput, InboundHTLCState, OutboundHTLCState, HTLCCandidate, HTLCInitiator};
        use crate::ln::channel::{MAX_FUNDING_SATOSHIS_NO_WUMBO, TOTAL_BITCOIN_SUPPLY_SATOSHIS, MIN_THEIR_CHAN_RESERVE_SATOSHIS};
        use crate::ln::features::ChannelTypeFeatures;
-       use crate::ln::msgs::{ChannelUpdate, DataLossProtect, DecodeError, OptionalField, UnsignedChannelUpdate, MAX_VALUE_MSAT};
+       use crate::ln::msgs::{ChannelUpdate, DecodeError, UnsignedChannelUpdate, MAX_VALUE_MSAT};
        use crate::ln::script::ShutdownScript;
        use crate::ln::chan_utils;
        use crate::ln::chan_utils::{htlc_success_tx_weight, htlc_timeout_tx_weight};
        use crate::chain::BestBlock;
        use crate::chain::chaininterface::{FeeEstimator, LowerBoundedFeeEstimator, ConfirmationTarget};
-       use crate::chain::keysinterface::{ChannelSigner, InMemorySigner, EntropySource, SignerProvider};
+       use crate::sign::{ChannelSigner, InMemorySigner, EntropySource, SignerProvider};
        use crate::chain::transaction::OutPoint;
        use crate::routing::router::Path;
        use crate::util::config::UserConfig;
@@ -7339,12 +7426,7 @@ mod tests {
                let msg = node_b_chan.get_channel_reestablish(&&logger);
                assert_eq!(msg.next_local_commitment_number, 1); // now called next_commitment_number
                assert_eq!(msg.next_remote_commitment_number, 0); // now called next_revocation_number
-               match msg.data_loss_protect {
-                       OptionalField::Present(DataLossProtect { your_last_per_commitment_secret, .. }) => {
-                               assert_eq!(your_last_per_commitment_secret, [0; 32]);
-                       },
-                       _ => panic!()
-               }
+               assert_eq!(msg.your_last_per_commitment_secret, [0; 32]);
 
                // Check that the commitment point in Node A's channel_reestablish message
                // is sane.
@@ -7352,12 +7434,7 @@ mod tests {
                let msg = node_a_chan.get_channel_reestablish(&&logger);
                assert_eq!(msg.next_local_commitment_number, 1); // now called next_commitment_number
                assert_eq!(msg.next_remote_commitment_number, 0); // now called next_revocation_number
-               match msg.data_loss_protect {
-                       OptionalField::Present(DataLossProtect { your_last_per_commitment_secret, .. }) => {
-                               assert_eq!(your_last_per_commitment_secret, [0; 32]);
-                       },
-                       _ => panic!()
-               }
+               assert_eq!(msg.your_last_per_commitment_secret, [0; 32]);
        }
 
        #[test]
@@ -7548,7 +7625,7 @@ mod tests {
                use bitcoin::hashes::hex::FromHex;
                use bitcoin::hash_types::Txid;
                use bitcoin::secp256k1::Message;
-               use crate::chain::keysinterface::EcdsaChannelSigner;
+               use crate::sign::EcdsaChannelSigner;
                use crate::ln::PaymentPreimage;
                use crate::ln::channel::{HTLCOutputInCommitment ,TxCreationKeys};
                use crate::ln::chan_utils::{ChannelPublicKeys, HolderCommitmentTransaction, CounterpartyChannelTransactionParameters};