Merge pull request #2730 from benthecarman/invoice-utils
[rust-lightning] / lightning / src / ln / functional_test_utils.rs
index db43c37fd5eb5b8d1dd299c22f836bf017db9f42..4dadd054ce219bef867e09f9ac01e91e4613665b 100644 (file)
@@ -15,9 +15,9 @@ use crate::sign::EntropySource;
 use crate::chain::channelmonitor::ChannelMonitor;
 use crate::chain::transaction::OutPoint;
 use crate::events::{ClaimedHTLC, ClosureReason, Event, HTLCDestination, MessageSendEvent, MessageSendEventsProvider, PathFailure, PaymentPurpose, PaymentFailureReason};
-use crate::events::bump_transaction::{BumpTransactionEventHandler, Wallet, WalletSource};
+use crate::events::bump_transaction::{BumpTransactionEvent, BumpTransactionEventHandler, Wallet, WalletSource};
 use crate::ln::{ChannelId, PaymentPreimage, PaymentHash, PaymentSecret};
-use crate::ln::channelmanager::{self, AChannelManager, ChainParameters, ChannelManager, ChannelManagerReadArgs, RAACommitmentOrder, PaymentSendFailure, RecipientOnionFields, PaymentId, MIN_CLTV_EXPIRY_DELTA};
+use crate::ln::channelmanager::{AChannelManager, ChainParameters, ChannelManager, ChannelManagerReadArgs, RAACommitmentOrder, PaymentSendFailure, RecipientOnionFields, PaymentId, MIN_CLTV_EXPIRY_DELTA};
 use crate::routing::gossip::{P2PGossipSync, NetworkGraph, NetworkUpdate};
 use crate::routing::router::{self, PaymentParameters, Route, RouteParameters};
 use crate::ln::features::InitFeatures;
@@ -30,6 +30,8 @@ use crate::util::test_utils::{panicking, TestChainMonitor, TestScorer, TestKeysI
 use crate::util::errors::APIError;
 use crate::util::config::{UserConfig, MaxDustHTLCExposure};
 use crate::util::ser::{ReadableArgs, Writeable};
+#[cfg(test)]
+use crate::util::logger::Logger;
 
 use bitcoin::blockdata::block::{Block, BlockHeader};
 use bitcoin::blockdata::transaction::{Transaction, TxOut};
@@ -73,6 +75,20 @@ pub fn mine_transactions<'a, 'b, 'c, 'd>(node: &'a Node<'b, 'c, 'd>, txn: &[&Tra
        let height = node.best_block_info().1 + 1;
        confirm_transactions_at(node, txn, height);
 }
+/// Mine a single block containing the given transaction without extra consistency checks which may
+/// impact ChannelManager state.
+pub fn mine_transaction_without_consistency_checks<'a, 'b, 'c, 'd>(node: &'a Node<'b, 'c, 'd>, tx: &Transaction) {
+       let height = node.best_block_info().1 + 1;
+       let mut block = Block {
+               header: BlockHeader { version: 0x20000000, prev_blockhash: node.best_block_hash(), merkle_root: TxMerkleNode::all_zeros(), time: height, bits: 42, nonce: 42 },
+               txdata: Vec::new(),
+       };
+       for _ in 0..*node.network_chan_count.borrow() { // Make sure we don't end up with channels at the same short id by offsetting by chan_count
+               block.txdata.push(Transaction { version: 0, lock_time: PackedLockTime::ZERO, input: Vec::new(), output: Vec::new() });
+       }
+       block.txdata.push((*tx).clone());
+       do_connect_block_without_consistency_checks(node, block, false);
+}
 /// Mine the given transaction at the given height, mining blocks as required to build to that
 /// height
 ///
@@ -211,16 +227,16 @@ pub fn connect_blocks<'a, 'b, 'c, 'd>(node: &'a Node<'b, 'c, 'd>, depth: u32) ->
        assert!(depth >= 1);
        for i in 1..depth {
                let prev_blockhash = block.header.block_hash();
-               do_connect_block(node, block, skip_intermediaries);
+               do_connect_block_with_consistency_checks(node, block, skip_intermediaries);
                block = create_dummy_block(prev_blockhash, height + i, Vec::new());
        }
        let hash = block.header.block_hash();
-       do_connect_block(node, block, false);
+       do_connect_block_with_consistency_checks(node, block, false);
        hash
 }
 
 pub fn connect_block<'a, 'b, 'c, 'd>(node: &'a Node<'b, 'c, 'd>, block: &Block) {
-       do_connect_block(node, block.clone(), false);
+       do_connect_block_with_consistency_checks(node, block.clone(), false);
 }
 
 fn call_claimable_balances<'a, 'b, 'c, 'd>(node: &'a Node<'b, 'c, 'd>) {
@@ -230,8 +246,14 @@ fn call_claimable_balances<'a, 'b, 'c, 'd>(node: &'a Node<'b, 'c, 'd>) {
        }
 }
 
-fn do_connect_block<'a, 'b, 'c, 'd>(node: &'a Node<'b, 'c, 'd>, block: Block, skip_intermediaries: bool) {
+fn do_connect_block_with_consistency_checks<'a, 'b, 'c, 'd>(node: &'a Node<'b, 'c, 'd>, block: Block, skip_intermediaries: bool) {
        call_claimable_balances(node);
+       do_connect_block_without_consistency_checks(node, block, skip_intermediaries);
+       call_claimable_balances(node);
+       node.node.test_process_background_events();
+}
+
+fn do_connect_block_without_consistency_checks<'a, 'b, 'c, 'd>(node: &'a Node<'b, 'c, 'd>, block: Block, skip_intermediaries: bool) {
        let height = node.best_block_info().1 + 1;
        #[cfg(feature = "std")] {
                eprintln!("Connecting block using Block Connection Style: {:?}", *node.connect_style.borrow());
@@ -286,8 +308,6 @@ fn do_connect_block<'a, 'b, 'c, 'd>(node: &'a Node<'b, 'c, 'd>, block: Block, sk
                        }
                }
        }
-       call_claimable_balances(node);
-       node.node.test_process_background_events();
 
        for tx in &block.txdata {
                for input in &tx.input {
@@ -404,6 +424,10 @@ pub struct Node<'chan_man, 'node_cfg: 'chan_man, 'chan_mon_cfg: 'node_cfg> {
                &'chan_mon_cfg test_utils::TestLogger,
        >,
 }
+#[cfg(feature = "std")]
+impl<'a, 'b, 'c> std::panic::UnwindSafe for Node<'a, 'b, 'c> {}
+#[cfg(feature = "std")]
+impl<'a, 'b, 'c> std::panic::RefUnwindSafe for Node<'a, 'b, 'c> {}
 impl<'a, 'b, 'c> Node<'a, 'b, 'c> {
        pub fn best_block_hash(&self) -> BlockHash {
                self.blocks.lock().unwrap().last().unwrap().0.block_hash()
@@ -414,6 +438,25 @@ impl<'a, 'b, 'c> Node<'a, 'b, 'c> {
        pub fn get_block_header(&self, height: u32) -> BlockHeader {
                self.blocks.lock().unwrap()[height as usize].0.header
        }
+       /// Changes the channel signer's availability for the specified peer and channel.
+       ///
+       /// When `available` is set to `true`, the channel signer will behave normally. When set to
+       /// `false`, the channel signer will act like an off-line remote signer and will return `Err` for
+       /// several of the signing methods. Currently, only `get_per_commitment_point` and
+       /// `release_commitment_secret` are affected by this setting.
+       #[cfg(test)]
+       pub fn set_channel_signer_available(&self, peer_id: &PublicKey, chan_id: &ChannelId, available: bool) {
+               let per_peer_state = self.node.per_peer_state.read().unwrap();
+               let chan_lock = per_peer_state.get(peer_id).unwrap().lock().unwrap();
+               let signer = (|| {
+                       match chan_lock.channel_by_id.get(chan_id) {
+                               Some(phase) => phase.context().get_signer(),
+                               None => panic!("Couldn't find a channel with id {}", chan_id),
+                       }
+               })();
+               log_debug!(self.logger, "Setting channel signer for {} as available={}", chan_id, available);
+               signer.as_ecdsa().unwrap().set_available(available);
+       }
 }
 
 /// If we need an unsafe pointer to a `Node` (ie to reference it in a thread
@@ -560,7 +603,7 @@ impl<'a, 'b, 'c> Drop for Node<'a, 'b, 'c> {
                        let chain_source = test_utils::TestChainSource::new(Network::Testnet);
                        let chain_monitor = test_utils::TestChainMonitor::new(Some(&chain_source), &broadcaster, &self.logger, &feeest, &persister, &self.keys_manager);
                        for deserialized_monitor in deserialized_monitors.drain(..) {
-                               if chain_monitor.watch_channel(deserialized_monitor.get_funding_txo().0, deserialized_monitor) != ChannelMonitorUpdateStatus::Completed {
+                               if chain_monitor.watch_channel(deserialized_monitor.get_funding_txo().0, deserialized_monitor) != Ok(ChannelMonitorUpdateStatus::Completed) {
                                        panic!();
                                }
                        }
@@ -643,6 +686,12 @@ pub fn get_err_msg(node: &Node, recipient: &PublicKey) -> msgs::ErrorMessage {
                        assert_eq!(node_id, recipient);
                        (*msg).clone()
                },
+               MessageSendEvent::HandleError {
+                       action: msgs::ErrorAction::DisconnectPeer { ref msg }, ref node_id
+               } => {
+                       assert_eq!(node_id, recipient);
+                       msg.as_ref().unwrap().clone()
+               },
                _ => panic!("Unexpected event"),
        }
 }
@@ -765,6 +814,18 @@ pub fn remove_first_msg_event_to_node(msg_node_id: &PublicKey, msg_events: &mut
                MessageSendEvent::SendOpenChannelV2 { node_id, .. } => {
                        node_id == msg_node_id
                },
+               MessageSendEvent::SendStfu { node_id, .. } => {
+                       node_id == msg_node_id
+               },
+               MessageSendEvent::SendSplice { node_id, .. } => {
+                       node_id == msg_node_id
+               },
+               MessageSendEvent::SendSpliceAck { node_id, .. } => {
+                       node_id == msg_node_id
+               },
+               MessageSendEvent::SendSpliceLocked { node_id, .. } => {
+                       node_id == msg_node_id
+               },
                MessageSendEvent::SendTxAddInput { node_id, .. } => {
                        node_id == msg_node_id
                },
@@ -896,7 +957,8 @@ macro_rules! unwrap_send_err {
 pub fn check_added_monitors<CM: AChannelManager, H: NodeHolder<CM=CM>>(node: &H, count: usize) {
        if let Some(chain_monitor) = node.chain_monitor() {
                let mut added_monitors = chain_monitor.added_monitors.lock().unwrap();
-               assert_eq!(added_monitors.len(), count);
+               let n = added_monitors.len();
+               assert_eq!(n, count, "expected {} monitors to be added, not {}", count, n);
                added_monitors.clear();
        }
 }
@@ -959,7 +1021,7 @@ pub fn _reload_node<'a, 'b, 'c>(node: &'a Node<'a, 'b, 'c>, default_config: User
 
        for monitor in monitors_read.drain(..) {
                assert_eq!(node.chain_monitor.watch_channel(monitor.get_funding_txo().0, monitor),
-                       ChannelMonitorUpdateStatus::Completed);
+                       Ok(ChannelMonitorUpdateStatus::Completed));
                check_added_monitors!(node, 1);
        }
 
@@ -1076,7 +1138,7 @@ pub fn open_zero_conf_channel<'a, 'b, 'c, 'd>(initiator: &'a Node<'b, 'c, 'd>, r
        let initiator_channels = initiator.node.list_usable_channels().len();
        let receiver_channels = receiver.node.list_usable_channels().len();
 
-       initiator.node.create_channel(receiver.node.get_our_node_id(), 100_000, 10_001, 42, initiator_config).unwrap();
+       initiator.node.create_channel(receiver.node.get_our_node_id(), 100_000, 10_001, 42, None, initiator_config).unwrap();
        let open_channel = get_event_msg!(initiator, MessageSendEvent::SendOpenChannel, receiver.node.get_our_node_id());
 
        receiver.node.handle_open_channel(&initiator.node.get_our_node_id(), &open_channel);
@@ -1142,7 +1204,7 @@ pub fn open_zero_conf_channel<'a, 'b, 'c, 'd>(initiator: &'a Node<'b, 'c, 'd>, r
 }
 
 pub fn create_chan_between_nodes_with_value_init<'a, 'b, 'c>(node_a: &Node<'a, 'b, 'c>, node_b: &Node<'a, 'b, 'c>, channel_value: u64, push_msat: u64) -> Transaction {
-       let create_chan_id = node_a.node.create_channel(node_b.node.get_our_node_id(), channel_value, push_msat, 42, None).unwrap();
+       let create_chan_id = node_a.node.create_channel(node_b.node.get_our_node_id(), channel_value, push_msat, 42, None, None).unwrap();
        let open_channel_msg = get_event_msg!(node_a, MessageSendEvent::SendOpenChannel, node_b.node.get_our_node_id());
        assert_eq!(open_channel_msg.temporary_channel_id, create_chan_id);
        assert_eq!(node_a.node.list_channels().iter().find(|channel| channel.channel_id == create_chan_id).unwrap().user_channel_id, 42);
@@ -1259,7 +1321,7 @@ pub fn create_announced_chan_between_nodes_with_value<'a, 'b, 'c: 'd, 'd>(nodes:
 pub fn create_unannounced_chan_between_nodes_with_value<'a, 'b, 'c, 'd>(nodes: &'a Vec<Node<'b, 'c, 'd>>, a: usize, b: usize, channel_value: u64, push_msat: u64) -> (msgs::ChannelReady, Transaction) {
        let mut no_announce_cfg = test_default_channel_config();
        no_announce_cfg.channel_handshake_config.announced_channel = false;
-       nodes[a].node.create_channel(nodes[b].node.get_our_node_id(), channel_value, push_msat, 42, Some(no_announce_cfg)).unwrap();
+       nodes[a].node.create_channel(nodes[b].node.get_our_node_id(), channel_value, push_msat, 42, None, Some(no_announce_cfg)).unwrap();
        let open_channel = get_event_msg!(nodes[a], MessageSendEvent::SendOpenChannel, nodes[b].node.get_our_node_id());
        nodes[b].node.handle_open_channel(&nodes[a].node.get_our_node_id(), &open_channel);
        let accept_channel = get_event_msg!(nodes[b], MessageSendEvent::SendAcceptChannel, nodes[a].node.get_our_node_id());
@@ -1424,10 +1486,15 @@ pub fn check_closed_broadcast(node: &Node, num_channels: usize, with_error_msg:
                                assert_eq!(msg.contents.flags & 2, 2);
                                None
                        },
-                       MessageSendEvent::HandleError { action: msgs::ErrorAction::SendErrorMessage { ref msg }, node_id: _ } => {
+                       MessageSendEvent::HandleError { action: msgs::ErrorAction::SendErrorMessage { msg }, node_id: _ } => {
                                assert!(with_error_msg);
                                // TODO: Check node_id
-                               Some(msg.clone())
+                               Some(msg)
+                       },
+                       MessageSendEvent::HandleError { action: msgs::ErrorAction::DisconnectPeer { msg }, node_id: _ } => {
+                               assert!(with_error_msg);
+                               // TODO: Check node_id
+                               Some(msg.unwrap())
                        },
                        _ => panic!("Unexpected event"),
                }
@@ -1445,27 +1512,61 @@ macro_rules! check_closed_broadcast {
        }
 }
 
+#[derive(Default)]
+pub struct ExpectedCloseEvent {
+       pub channel_capacity_sats: Option<u64>,
+       pub channel_id: Option<ChannelId>,
+       pub counterparty_node_id: Option<PublicKey>,
+       pub discard_funding: bool,
+       pub reason: Option<ClosureReason>,
+}
+
+/// Check that multiple channel closing events have been issued.
+pub fn check_closed_events(node: &Node, expected_close_events: &[ExpectedCloseEvent]) {
+       let closed_events_count = expected_close_events.len();
+       let discard_events_count = expected_close_events.iter().filter(|e| e.discard_funding).count();
+       let events = node.node.get_and_clear_pending_events();
+       assert_eq!(events.len(), closed_events_count + discard_events_count, "{:?}", events);
+       for expected_event in expected_close_events {
+               assert!(events.iter().any(|e| matches!(
+                       e,
+                       Event::ChannelClosed {
+                               channel_id,
+                               reason,
+                               counterparty_node_id,
+                               channel_capacity_sats,
+                               ..
+                       } if (
+                               expected_event.channel_id.map(|expected| *channel_id == expected).unwrap_or(true) &&
+                               expected_event.reason.as_ref().map(|expected| reason == expected).unwrap_or(true) &&
+                               expected_event.counterparty_node_id.map(|expected| *counterparty_node_id == Some(expected)).unwrap_or(true) &&
+                               expected_event.channel_capacity_sats.map(|expected| *channel_capacity_sats == Some(expected)).unwrap_or(true)
+                       )
+               )));
+       }
+       assert_eq!(events.iter().filter(|e| matches!(
+               e,
+               Event::DiscardFunding { .. },
+       )).count(), discard_events_count);
+}
+
 /// Check that a channel's closing channel events has been issued
 pub fn check_closed_event(node: &Node, events_count: usize, expected_reason: ClosureReason, is_check_discard_funding: bool,
        expected_counterparty_node_ids: &[PublicKey], expected_channel_capacity: u64) {
-       let events = node.node.get_and_clear_pending_events();
-       assert_eq!(events.len(), events_count, "{:?}", events);
-       let mut issues_discard_funding = false;
-       for (idx, event) in events.into_iter().enumerate() {
-               match event {
-                       Event::ChannelClosed { ref reason, counterparty_node_id, 
-                               channel_capacity_sats, .. } => {
-                               assert_eq!(*reason, expected_reason);
-                               assert_eq!(counterparty_node_id.unwrap(), expected_counterparty_node_ids[idx]);
-                               assert_eq!(channel_capacity_sats.unwrap(), expected_channel_capacity);
-                       },
-                       Event::DiscardFunding { .. } => {
-                               issues_discard_funding = true;
-                       }
-                       _ => panic!("Unexpected event"),
-               }
-       }
-       assert_eq!(is_check_discard_funding, issues_discard_funding);
+       let expected_events_count = if is_check_discard_funding {
+               2 * expected_counterparty_node_ids.len()
+       } else {
+               expected_counterparty_node_ids.len()
+       };
+       assert_eq!(events_count, expected_events_count);
+       let expected_close_events = expected_counterparty_node_ids.iter().map(|node_id| ExpectedCloseEvent {
+               channel_capacity_sats: Some(expected_channel_capacity),
+               channel_id: None,
+               counterparty_node_id: Some(*node_id),
+               discard_funding: is_check_discard_funding,
+               reason: Some(expected_reason.clone()),
+       }).collect::<Vec<_>>();
+       check_closed_events(node, expected_close_events.as_slice());
 }
 
 /// Check that a channel's closing channel events has been issued
@@ -1477,11 +1578,26 @@ macro_rules! check_closed_event {
                check_closed_event!($node, $events, $reason, false, $counterparty_node_ids, $channel_capacity);
        };
        ($node: expr, $events: expr, $reason: expr, $is_check_discard_funding: expr, $counterparty_node_ids: expr, $channel_capacity: expr) => {
-               $crate::ln::functional_test_utils::check_closed_event(&$node, $events, $reason, 
+               $crate::ln::functional_test_utils::check_closed_event(&$node, $events, $reason,
                        $is_check_discard_funding, &$counterparty_node_ids, $channel_capacity);
        }
 }
 
+pub fn handle_bump_htlc_event(node: &Node, count: usize) {
+       let events = node.chain_monitor.chain_monitor.get_and_clear_pending_events();
+       assert_eq!(events.len(), count);
+       for event in events {
+               match event {
+                       Event::BumpTransaction(bump_event) => {
+                               if let BumpTransactionEvent::HTLCResolution { .. } = &bump_event {}
+                               else { panic!(); }
+                               node.bump_tx_handler.handle_event(&bump_event);
+                       },
+                       _ => panic!(),
+               }
+       }
+}
+
 pub fn close_channel<'a, 'b, 'c>(outbound_node: &Node<'a, 'b, 'c>, inbound_node: &Node<'a, 'b, 'c>, channel_id: &ChannelId, funding_tx: Transaction, close_inbound_first: bool) -> (msgs::ChannelUpdate, msgs::ChannelUpdate, Transaction) {
        let (node_a, broadcaster_a, struct_a) = if close_inbound_first { (&inbound_node.node, &inbound_node.tx_broadcaster, inbound_node) } else { (&outbound_node.node, &outbound_node.tx_broadcaster, outbound_node) };
        let (node_b, broadcaster_b, struct_b) = if close_inbound_first { (&outbound_node.node, &outbound_node.tx_broadcaster, outbound_node) } else { (&inbound_node.node, &inbound_node.tx_broadcaster, inbound_node) };
@@ -1772,20 +1888,7 @@ pub fn do_commitment_signed_dance(node_a: &Node<'_, '_, '_>, node_b: &Node<'_, '
        check_added_monitors!(node_a, 1);
 
        // If this commitment signed dance was due to a claim, don't check for an RAA monitor update.
-       let got_claim = node_a.node.pending_events.lock().unwrap().iter().any(|(ev, action)| {
-               let matching_action = if let Some(channelmanager::EventCompletionAction::ReleaseRAAChannelMonitorUpdate
-                       { channel_funding_outpoint, counterparty_node_id }) = action
-               {
-                       if channel_funding_outpoint.to_channel_id() == commitment_signed.channel_id {
-                               assert_eq!(*counterparty_node_id, node_b.node.get_our_node_id());
-                               true
-                       } else { false }
-               } else { false };
-               if matching_action {
-                       if let Event::PaymentSent { .. } = ev {} else { panic!(); }
-               }
-               matching_action
-       });
+       let got_claim = node_a.node.test_raa_monitor_updates_held(node_b.node.get_our_node_id(), commitment_signed.channel_id);
        if fail_backwards { assert!(!got_claim); }
        commitment_signed_dance!(node_a, node_b, (), fail_backwards, true, false, got_claim);
 
@@ -1849,7 +1952,7 @@ pub fn get_route(send_node: &Node, route_params: &RouteParameters) -> Result<Rou
        router::get_route(
                &send_node.node.get_our_node_id(), route_params, &send_node.network_graph.read_only(),
                Some(&send_node.node.list_usable_channels().iter().collect::<Vec<_>>()),
-               send_node.logger, &scorer, &(), &random_seed_bytes
+               send_node.logger, &scorer, &Default::default(), &random_seed_bytes
        )
 }
 
@@ -1869,11 +1972,15 @@ macro_rules! get_route {
 macro_rules! get_route_and_payment_hash {
        ($send_node: expr, $recv_node: expr, $recv_value: expr) => {{
                let payment_params = $crate::routing::router::PaymentParameters::from_node_id($recv_node.node.get_our_node_id(), TEST_FINAL_CLTV)
-                       .with_bolt11_features($recv_node.node.invoice_features()).unwrap();
+                       .with_bolt11_features($recv_node.node.bolt11_invoice_features()).unwrap();
                $crate::get_route_and_payment_hash!($send_node, $recv_node, payment_params, $recv_value)
        }};
        ($send_node: expr, $recv_node: expr, $payment_params: expr, $recv_value: expr) => {{
-               let route_params = $crate::routing::router::RouteParameters::from_payment_params_and_value($payment_params, $recv_value);
+               $crate::get_route_and_payment_hash!($send_node, $recv_node, $payment_params, $recv_value, None)
+       }};
+       ($send_node: expr, $recv_node: expr, $payment_params: expr, $recv_value: expr, $max_total_routing_fee_msat: expr) => {{
+               let mut route_params = $crate::routing::router::RouteParameters::from_payment_params_and_value($payment_params, $recv_value);
+               route_params.max_total_routing_fee_msat = $max_total_routing_fee_msat;
                let (payment_preimage, payment_hash, payment_secret) =
                        $crate::ln::functional_test_utils::get_payment_preimage_hash(&$recv_node, Some($recv_value), None);
                let route = $crate::ln::functional_test_utils::get_route(&$send_node, &route_params);
@@ -2000,29 +2107,38 @@ macro_rules! expect_payment_path_successful {
        }
 }
 
+pub fn expect_payment_forwarded<CM: AChannelManager, H: NodeHolder<CM=CM>>(
+       event: Event, node: &H, prev_node: &H, next_node: &H, expected_fee: Option<u64>,
+       upstream_force_closed: bool, downstream_force_closed: bool
+) {
+       match event {
+               Event::PaymentForwarded {
+                       fee_earned_msat, prev_channel_id, claim_from_onchain_tx, next_channel_id,
+                       outbound_amount_forwarded_msat: _
+               } => {
+                       assert_eq!(fee_earned_msat, expected_fee);
+                       if !upstream_force_closed {
+                               // Is the event prev_channel_id in one of the channels between the two nodes?
+                               assert!(node.node().list_channels().iter().any(|x| x.counterparty.node_id == prev_node.node().get_our_node_id() && x.channel_id == prev_channel_id.unwrap()));
+                       }
+                       // We check for force closures since a force closed channel is removed from the
+                       // node's channel list
+                       if !downstream_force_closed {
+                               assert!(node.node().list_channels().iter().any(|x| x.counterparty.node_id == next_node.node().get_our_node_id() && x.channel_id == next_channel_id.unwrap()));
+                       }
+                       assert_eq!(claim_from_onchain_tx, downstream_force_closed);
+               },
+               _ => panic!("Unexpected event"),
+       }
+}
+
 macro_rules! expect_payment_forwarded {
        ($node: expr, $prev_node: expr, $next_node: expr, $expected_fee: expr, $upstream_force_closed: expr, $downstream_force_closed: expr) => {
-               let events = $node.node.get_and_clear_pending_events();
+               let mut events = $node.node.get_and_clear_pending_events();
                assert_eq!(events.len(), 1);
-               match events[0] {
-                       Event::PaymentForwarded {
-                               fee_earned_msat, prev_channel_id, claim_from_onchain_tx, next_channel_id,
-                               outbound_amount_forwarded_msat: _
-                       } => {
-                               assert_eq!(fee_earned_msat, $expected_fee);
-                               if fee_earned_msat.is_some() {
-                                       // Is the event prev_channel_id in one of the channels between the two nodes?
-                                       assert!($node.node.list_channels().iter().any(|x| x.counterparty.node_id == $prev_node.node.get_our_node_id() && x.channel_id == prev_channel_id.unwrap()));
-                               }
-                               // We check for force closures since a force closed channel is removed from the
-                               // node's channel list
-                               if !$downstream_force_closed {
-                                       assert!($node.node.list_channels().iter().any(|x| x.counterparty.node_id == $next_node.node.get_our_node_id() && x.channel_id == next_channel_id.unwrap()));
-                               }
-                               assert_eq!(claim_from_onchain_tx, $downstream_force_closed);
-                       },
-                       _ => panic!("Unexpected event"),
-               }
+               $crate::ln::functional_test_utils::expect_payment_forwarded(
+                       events.pop().unwrap(), &$node, &$prev_node, &$next_node, $expected_fee,
+                       $upstream_force_closed, $downstream_force_closed);
        }
 }
 
@@ -2037,12 +2153,13 @@ macro_rules! expect_channel_shutdown_state {
 }
 
 #[cfg(any(test, ldk_bench, feature = "_test_utils"))]
-pub fn expect_channel_pending_event<'a, 'b, 'c, 'd>(node: &'a Node<'b, 'c, 'd>, expected_counterparty_node_id: &PublicKey) {
+pub fn expect_channel_pending_event<'a, 'b, 'c, 'd>(node: &'a Node<'b, 'c, 'd>, expected_counterparty_node_id: &PublicKey) -> ChannelId {
        let events = node.node.get_and_clear_pending_events();
        assert_eq!(events.len(), 1);
-       match events[0] {
-               crate::events::Event::ChannelPending { ref counterparty_node_id, .. } => {
+       match &events[0] {
+               crate::events::Event::ChannelPending { channel_id, counterparty_node_id, .. } => {
                        assert_eq!(*expected_counterparty_node_id, *counterparty_node_id);
+                       *channel_id
                },
                _ => panic!("Unexpected event"),
        }
@@ -2060,6 +2177,26 @@ pub fn expect_channel_ready_event<'a, 'b, 'c, 'd>(node: &'a Node<'b, 'c, 'd>, ex
        }
 }
 
+#[cfg(any(test, feature = "_test_utils"))]
+pub fn expect_probe_successful_events(node: &Node, mut probe_results: Vec<(PaymentHash, PaymentId)>) {
+       let mut events = node.node.get_and_clear_pending_events();
+
+       for event in events.drain(..) {
+               match event {
+                       Event::ProbeSuccessful { payment_hash: ev_ph, payment_id: ev_pid, ..} => {
+                               let result_idx = probe_results.iter().position(|(payment_hash, payment_id)| *payment_hash == ev_ph && *payment_id == ev_pid);
+                               assert!(result_idx.is_some());
+
+                               probe_results.remove(result_idx.unwrap());
+                       },
+                       _ => panic!(),
+               }
+       };
+
+       // Ensure that we received a ProbeSuccessful event for each probe result.
+       assert!(probe_results.is_empty());
+}
+
 pub struct PaymentFailedConditions<'a> {
        pub(crate) expected_htlc_error_data: Option<(u16, &'a [u8])>,
        pub(crate) expected_blamed_scid: Option<u64>,
@@ -2197,21 +2334,41 @@ pub fn send_along_route_with_secret<'a, 'b, 'c>(origin_node: &Node<'a, 'b, 'c>,
        payment_id
 }
 
-pub fn do_pass_along_path<'a, 'b, 'c>(origin_node: &Node<'a, 'b, 'c>, expected_path: &[&Node<'a, 'b, 'c>], recv_value: u64, our_payment_hash: PaymentHash, our_payment_secret: Option<PaymentSecret>, ev: MessageSendEvent, payment_claimable_expected: bool, clear_recipient_events: bool, expected_preimage: Option<PaymentPreimage>) -> Option<Event> {
+fn fail_payment_along_path<'a, 'b, 'c>(expected_path: &[&Node<'a, 'b, 'c>]) {
+       let origin_node_id = expected_path[0].node.get_our_node_id();
+
+       // iterate from the receiving node to the origin node and handle update fail htlc.
+       for (&node, &prev_node) in expected_path.iter().rev().zip(expected_path.iter().rev().skip(1)) {
+               let updates = get_htlc_update_msgs!(node, prev_node.node.get_our_node_id());
+               prev_node.node.handle_update_fail_htlc(&node.node.get_our_node_id(), &updates.update_fail_htlcs[0]);
+               check_added_monitors!(prev_node, 0);
+
+               let is_first_hop = origin_node_id == prev_node.node.get_our_node_id();
+               // We do not want to fail backwards on the first hop. All other hops should fail backwards.
+               commitment_signed_dance!(prev_node, node, updates.commitment_signed, !is_first_hop);
+       }
+}
+
+pub fn do_pass_along_path<'a, 'b, 'c>(origin_node: &Node<'a, 'b, 'c>, expected_path: &[&Node<'a, 'b, 'c>], recv_value: u64, our_payment_hash: PaymentHash, our_payment_secret: Option<PaymentSecret>, ev: MessageSendEvent, payment_claimable_expected: bool, clear_recipient_events: bool, expected_preimage: Option<PaymentPreimage>, is_probe: bool) -> Option<Event> {
        let mut payment_event = SendEvent::from_event(ev);
        let mut prev_node = origin_node;
        let mut event = None;
 
        for (idx, &node) in expected_path.iter().enumerate() {
+               let is_last_hop = idx == expected_path.len() - 1;
                assert_eq!(node.node.get_our_node_id(), payment_event.node_id);
 
                node.node.handle_update_add_htlc(&prev_node.node.get_our_node_id(), &payment_event.msgs[0]);
                check_added_monitors!(node, 0);
-               commitment_signed_dance!(node, prev_node, payment_event.commitment_msg, false);
 
-               expect_pending_htlcs_forwardable!(node);
+               if is_last_hop && is_probe {
+                       commitment_signed_dance!(node, prev_node, payment_event.commitment_msg, true, true);
+               } else {
+                       commitment_signed_dance!(node, prev_node, payment_event.commitment_msg, false);
+                       expect_pending_htlcs_forwardable!(node);
+               }
 
-               if idx == expected_path.len() - 1 && clear_recipient_events {
+               if is_last_hop && clear_recipient_events {
                        let events_2 = node.node.get_and_clear_pending_events();
                        if payment_claimable_expected {
                                assert_eq!(events_2.len(), 1);
@@ -2245,7 +2402,7 @@ pub fn do_pass_along_path<'a, 'b, 'c>(origin_node: &Node<'a, 'b, 'c>, expected_p
                        } else {
                                assert!(events_2.is_empty());
                        }
-               } else if idx != expected_path.len() - 1 {
+               } else if !is_last_hop {
                        let mut events_2 = node.node.get_and_clear_pending_msg_events();
                        assert_eq!(events_2.len(), 1);
                        check_added_monitors!(node, 1);
@@ -2259,16 +2416,33 @@ pub fn do_pass_along_path<'a, 'b, 'c>(origin_node: &Node<'a, 'b, 'c>, expected_p
 }
 
 pub fn pass_along_path<'a, 'b, 'c>(origin_node: &Node<'a, 'b, 'c>, expected_path: &[&Node<'a, 'b, 'c>], recv_value: u64, our_payment_hash: PaymentHash, our_payment_secret: Option<PaymentSecret>, ev: MessageSendEvent, payment_claimable_expected: bool, expected_preimage: Option<PaymentPreimage>) -> Option<Event> {
-       do_pass_along_path(origin_node, expected_path, recv_value, our_payment_hash, our_payment_secret, ev, payment_claimable_expected, true, expected_preimage)
+       do_pass_along_path(origin_node, expected_path, recv_value, our_payment_hash, our_payment_secret, ev, payment_claimable_expected, true, expected_preimage, false)
+}
+
+pub fn send_probe_along_route<'a, 'b, 'c>(origin_node: &Node<'a, 'b, 'c>, expected_route: &[&[&Node<'a, 'b, 'c>]]) {
+       let mut events = origin_node.node.get_and_clear_pending_msg_events();
+       assert_eq!(events.len(), expected_route.len());
+
+       check_added_monitors!(origin_node, expected_route.len());
+
+       for path in expected_route.iter() {
+               let ev = remove_first_msg_event_to_node(&path[0].node.get_our_node_id(), &mut events);
+
+               do_pass_along_path(origin_node, path, 0, PaymentHash([0_u8; 32]), None, ev, false, false, None, true);
+               let nodes_to_fail_payment: Vec<_> = vec![origin_node].into_iter().chain(path.iter().cloned()).collect();
+
+               fail_payment_along_path(nodes_to_fail_payment.as_slice());
+       }
 }
 
 pub fn pass_along_route<'a, 'b, 'c>(origin_node: &Node<'a, 'b, 'c>, expected_route: &[&[&Node<'a, 'b, 'c>]], recv_value: u64, our_payment_hash: PaymentHash, our_payment_secret: PaymentSecret) {
        let mut events = origin_node.node.get_and_clear_pending_msg_events();
        assert_eq!(events.len(), expected_route.len());
+
        for (path_idx, expected_path) in expected_route.iter().enumerate() {
                let ev = remove_first_msg_event_to_node(&expected_path[0].node.get_our_node_id(), &mut events);
                // Once we've gotten through all the HTLCs, the last one should result in a
-               // PaymentClaimable (but each previous one should not!).
+               // PaymentClaimable (but each previous one should not!).
                let expect_payment = path_idx == expected_route.len() - 1;
                pass_along_path(origin_node, expected_path, recv_value, our_payment_hash.clone(), Some(our_payment_secret), ev, expect_payment, None);
        }
@@ -2399,7 +2573,7 @@ pub fn pass_claimed_payment_along_route<'a, 'b, 'c>(origin_node: &Node<'a, 'b, '
                                                }
                                        };
                                        if $idx == 1 { fee += expected_extra_fees[i]; }
-                                       expect_payment_forwarded!($node, $next_node, $prev_node, Some(fee as u64), false, false);
+                                       expect_payment_forwarded!(*$node, $next_node, $prev_node, Some(fee as u64), false, false);
                                        expected_total_fee_msat += fee as u64;
                                        check_added_monitors!($node, 1);
                                        let new_next_msgs = if $new_msgs {
@@ -2467,9 +2641,9 @@ pub fn claim_payment<'a, 'b, 'c>(origin_node: &Node<'a, 'b, 'c>, expected_route:
 
 pub const TEST_FINAL_CLTV: u32 = 70;
 
-pub fn route_payment<'a, 'b, 'c>(origin_node: &Node<'a, 'b, 'c>, expected_route: &[&Node<'a, 'b, 'c>], recv_value: u64) -> (PaymentPreimage, PaymentHash, PaymentSecret) {
+pub fn route_payment<'a, 'b, 'c>(origin_node: &Node<'a, 'b, 'c>, expected_route: &[&Node<'a, 'b, 'c>], recv_value: u64) -> (PaymentPreimage, PaymentHash, PaymentSecret, PaymentId) {
        let payment_params = PaymentParameters::from_node_id(expected_route.last().unwrap().node.get_our_node_id(), TEST_FINAL_CLTV)
-               .with_bolt11_features(expected_route.last().unwrap().node.invoice_features()).unwrap();
+               .with_bolt11_features(expected_route.last().unwrap().node.bolt11_invoice_features()).unwrap();
        let route_params = RouteParameters::from_payment_params_and_value(payment_params, recv_value);
        let route = get_route(origin_node, &route_params).unwrap();
        assert_eq!(route.paths.len(), 1);
@@ -2479,12 +2653,12 @@ pub fn route_payment<'a, 'b, 'c>(origin_node: &Node<'a, 'b, 'c>, expected_route:
        }
 
        let res = send_along_route(origin_node, route, expected_route, recv_value);
-       (res.0, res.1, res.2)
+       (res.0, res.1, res.2, res.3)
 }
 
 pub fn route_over_limit<'a, 'b, 'c>(origin_node: &Node<'a, 'b, 'c>, expected_route: &[&Node<'a, 'b, 'c>], recv_value: u64)  {
        let payment_params = PaymentParameters::from_node_id(expected_route.last().unwrap().node.get_our_node_id(), TEST_FINAL_CLTV)
-               .with_bolt11_features(expected_route.last().unwrap().node.invoice_features()).unwrap();
+               .with_bolt11_features(expected_route.last().unwrap().node.bolt11_invoice_features()).unwrap();
        let route_params = RouteParameters::from_payment_params_and_value(payment_params, recv_value);
        let network_graph = origin_node.network_graph.read_only();
        let scorer = test_utils::TestScorer::new();
@@ -2492,7 +2666,7 @@ pub fn route_over_limit<'a, 'b, 'c>(origin_node: &Node<'a, 'b, 'c>, expected_rou
        let keys_manager = test_utils::TestKeysInterface::new(&seed, Network::Testnet);
        let random_seed_bytes = keys_manager.get_secure_random_bytes();
        let route = router::get_route(&origin_node.node.get_our_node_id(), &route_params, &network_graph,
-               None, origin_node.logger, &scorer, &(), &random_seed_bytes).unwrap();
+               None, origin_node.logger, &scorer, &Default::default(), &random_seed_bytes).unwrap();
        assert_eq!(route.paths.len(), 1);
        assert_eq!(route.paths[0].hops.len(), expected_route.len());
        for (node, hop) in expected_route.iter().zip(route.paths[0].hops.iter()) {
@@ -2506,7 +2680,7 @@ pub fn route_over_limit<'a, 'b, 'c>(origin_node: &Node<'a, 'b, 'c>, expected_rou
                assert!(err.contains("Cannot send value that would put us over the max HTLC value in flight our peer will accept")));
 }
 
-pub fn send_payment<'a, 'b, 'c>(origin: &Node<'a, 'b, 'c>, expected_route: &[&Node<'a, 'b, 'c>], recv_value: u64) -> (PaymentPreimage, PaymentHash, PaymentSecret) {
+pub fn send_payment<'a, 'b, 'c>(origin: &Node<'a, 'b, 'c>, expected_route: &[&Node<'a, 'b, 'c>], recv_value: u64) -> (PaymentPreimage, PaymentHash, PaymentSecret, PaymentId) {
        let res = route_payment(&origin, expected_route, recv_value);
        claim_payment(&origin, expected_route, res.0);
        res
@@ -2758,7 +2932,8 @@ pub fn create_network<'a, 'b: 'a, 'c: 'b>(node_count: usize, cfgs: &'b Vec<NodeC
 }
 
 // Note that the following only works for CLTV values up to 128
-pub const ACCEPTED_HTLC_SCRIPT_WEIGHT: usize = 137; //Here we have a diff due to HTLC CLTV expiry being < 2^15 in test
+pub const ACCEPTED_HTLC_SCRIPT_WEIGHT: usize = 137; // Here we have a diff due to HTLC CLTV expiry being < 2^15 in test
+pub const ACCEPTED_HTLC_SCRIPT_WEIGHT_ANCHORS: usize = 140; // Here we have a diff due to HTLC CLTV expiry being < 2^15 in test
 
 #[derive(PartialEq)]
 pub enum HTLCType { NONE, TIMEOUT, SUCCESS }
@@ -2883,6 +3058,13 @@ pub fn handle_announce_close_broadcast_events<'a, 'b, 'c>(nodes: &Vec<Node<'a, '
                                nodes[b].node.handle_error(&nodes[a].node.get_our_node_id(), msg);
                        }
                },
+               MessageSendEvent::HandleError { node_id, action: msgs::ErrorAction::DisconnectPeer { ref msg } } => {
+                       assert_eq!(node_id, nodes[b].node.get_our_node_id());
+                       assert_eq!(msg.as_ref().unwrap().data, expected_error);
+                       if needs_err_handle {
+                               nodes[b].node.handle_error(&nodes[a].node.get_our_node_id(), msg.as_ref().unwrap());
+                       }
+               },
                _ => panic!("Unexpected event"),
        }
 
@@ -2900,6 +3082,10 @@ pub fn handle_announce_close_broadcast_events<'a, 'b, 'c>(nodes: &Vec<Node<'a, '
                                assert_eq!(node_id, nodes[a].node.get_our_node_id());
                                assert_eq!(msg.data, expected_error);
                        },
+                       MessageSendEvent::HandleError { node_id, action: msgs::ErrorAction::DisconnectPeer { ref msg } } => {
+                               assert_eq!(node_id, nodes[a].node.get_our_node_id());
+                               assert_eq!(msg.as_ref().unwrap().data, expected_error);
+                       },
                        _ => panic!("Unexpected event"),
                }
        }
@@ -3030,7 +3216,11 @@ pub struct ReconnectArgs<'a, 'b, 'c, 'd> {
        pub node_a: &'a Node<'b, 'c, 'd>,
        pub node_b: &'a Node<'b, 'c, 'd>,
        pub send_channel_ready: (bool, bool),
-       pub pending_htlc_adds: (i64, i64),
+       pub pending_responding_commitment_signed: (bool, bool),
+       /// Indicates that the pending responding commitment signed will be a dup for the recipient,
+       /// and no monitor update is expected
+       pub pending_responding_commitment_signed_dup_monitor: (bool, bool),
+       pub pending_htlc_adds: (usize, usize),
        pub pending_htlc_claims: (usize, usize),
        pub pending_htlc_fails: (usize, usize),
        pub pending_cell_htlc_claims: (usize, usize),
@@ -3044,6 +3234,8 @@ impl<'a, 'b, 'c, 'd> ReconnectArgs<'a, 'b, 'c, 'd> {
                        node_a,
                        node_b,
                        send_channel_ready: (false, false),
+                       pending_responding_commitment_signed: (false, false),
+                       pending_responding_commitment_signed_dup_monitor: (false, false),
                        pending_htlc_adds: (0, 0),
                        pending_htlc_claims: (0, 0),
                        pending_htlc_fails: (0, 0),
@@ -3059,7 +3251,8 @@ impl<'a, 'b, 'c, 'd> ReconnectArgs<'a, 'b, 'c, 'd> {
 pub fn reconnect_nodes<'a, 'b, 'c, 'd>(args: ReconnectArgs<'a, 'b, 'c, 'd>) {
        let ReconnectArgs {
                node_a, node_b, send_channel_ready, pending_htlc_adds, pending_htlc_claims, pending_htlc_fails,
-               pending_cell_htlc_claims, pending_cell_htlc_fails, pending_raa
+               pending_cell_htlc_claims, pending_cell_htlc_fails, pending_raa,
+               pending_responding_commitment_signed, pending_responding_commitment_signed_dup_monitor,
        } = args;
        node_a.node.peer_connected(&node_b.node.get_our_node_id(), &msgs::Init {
                features: node_b.node.init_features(), networks: None, remote_network_address: None
@@ -3074,24 +3267,28 @@ pub fn reconnect_nodes<'a, 'b, 'c, 'd>(args: ReconnectArgs<'a, 'b, 'c, 'd>) {
                // If a expects a channel_ready, it better not think it has received a revoke_and_ack
                // from b
                for reestablish in reestablish_1.iter() {
-                       assert_eq!(reestablish.next_remote_commitment_number, 0);
+                       let n = reestablish.next_remote_commitment_number;
+                       assert_eq!(n, 0, "expected a->b next_remote_commitment_number to be 0, got {}", n);
                }
        }
        if send_channel_ready.1 {
                // If b expects a channel_ready, it better not think it has received a revoke_and_ack
                // from a
                for reestablish in reestablish_2.iter() {
-                       assert_eq!(reestablish.next_remote_commitment_number, 0);
+                       let n = reestablish.next_remote_commitment_number;
+                       assert_eq!(n, 0, "expected b->a next_remote_commitment_number to be 0, got {}", n);
                }
        }
        if send_channel_ready.0 || send_channel_ready.1 {
                // If we expect any channel_ready's, both sides better have set
                // next_holder_commitment_number to 1
                for reestablish in reestablish_1.iter() {
-                       assert_eq!(reestablish.next_local_commitment_number, 1);
+                       let n = reestablish.next_local_commitment_number;
+                       assert_eq!(n, 1, "expected a->b next_local_commitment_number to be 1, got {}", n);
                }
                for reestablish in reestablish_2.iter() {
-                       assert_eq!(reestablish.next_local_commitment_number, 1);
+                       let n = reestablish.next_local_commitment_number;
+                       assert_eq!(n, 1, "expected b->a next_local_commitment_number to be 1, got {}", n);
                }
        }
 
@@ -3144,13 +3341,12 @@ pub fn reconnect_nodes<'a, 'b, 'c, 'd>(args: ReconnectArgs<'a, 'b, 'c, 'd>) {
                } else {
                        assert!(chan_msgs.1.is_none());
                }
-               if pending_htlc_adds.0 != 0 || pending_htlc_claims.0 != 0 || pending_htlc_fails.0 != 0 || pending_cell_htlc_claims.0 != 0 || pending_cell_htlc_fails.0 != 0 {
+               if pending_htlc_adds.0 != 0 || pending_htlc_claims.0 != 0 || pending_htlc_fails.0 != 0 ||
+                       pending_cell_htlc_claims.0 != 0 || pending_cell_htlc_fails.0 != 0 ||
+                       pending_responding_commitment_signed.0
+               {
                        let commitment_update = chan_msgs.2.unwrap();
-                       if pending_htlc_adds.0 != -1 { // We use -1 to denote a response commitment_signed
-                               assert_eq!(commitment_update.update_add_htlcs.len(), pending_htlc_adds.0 as usize);
-                       } else {
-                               assert!(commitment_update.update_add_htlcs.is_empty());
-                       }
+                       assert_eq!(commitment_update.update_add_htlcs.len(), pending_htlc_adds.0);
                        assert_eq!(commitment_update.update_fulfill_htlcs.len(), pending_htlc_claims.0 + pending_cell_htlc_claims.0);
                        assert_eq!(commitment_update.update_fail_htlcs.len(), pending_htlc_fails.0 + pending_cell_htlc_fails.0);
                        assert!(commitment_update.update_fail_malformed_htlcs.is_empty());
@@ -3164,7 +3360,7 @@ pub fn reconnect_nodes<'a, 'b, 'c, 'd>(args: ReconnectArgs<'a, 'b, 'c, 'd>) {
                                node_a.node.handle_update_fail_htlc(&node_b.node.get_our_node_id(), &update_fail);
                        }
 
-                       if pending_htlc_adds.0 != -1 { // We use -1 to denote a response commitment_signed
+                       if !pending_responding_commitment_signed.0 {
                                commitment_signed_dance!(node_a, node_b, commitment_update.commitment_signed, false);
                        } else {
                                node_a.node.handle_commitment_signed(&node_b.node.get_our_node_id(), &commitment_update.commitment_signed);
@@ -3173,7 +3369,7 @@ pub fn reconnect_nodes<'a, 'b, 'c, 'd>(args: ReconnectArgs<'a, 'b, 'c, 'd>) {
                                // No commitment_signed so get_event_msg's assert(len == 1) passes
                                node_b.node.handle_revoke_and_ack(&node_a.node.get_our_node_id(), &as_revoke_and_ack);
                                assert!(node_b.node.get_and_clear_pending_msg_events().is_empty());
-                               check_added_monitors!(node_b, 1);
+                               check_added_monitors!(node_b, if pending_responding_commitment_signed_dup_monitor.0 { 0 } else { 1 });
                        }
                } else {
                        assert!(chan_msgs.2.is_none());
@@ -3203,11 +3399,12 @@ pub fn reconnect_nodes<'a, 'b, 'c, 'd>(args: ReconnectArgs<'a, 'b, 'c, 'd>) {
                } else {
                        assert!(chan_msgs.1.is_none());
                }
-               if pending_htlc_adds.1 != 0 || pending_htlc_claims.1 != 0 || pending_htlc_fails.1 != 0 || pending_cell_htlc_claims.1 != 0 || pending_cell_htlc_fails.1 != 0 {
+               if pending_htlc_adds.1 != 0 || pending_htlc_claims.1 != 0 || pending_htlc_fails.1 != 0 ||
+                       pending_cell_htlc_claims.1 != 0 || pending_cell_htlc_fails.1 != 0 ||
+                       pending_responding_commitment_signed.1
+               {
                        let commitment_update = chan_msgs.2.unwrap();
-                       if pending_htlc_adds.1 != -1 { // We use -1 to denote a response commitment_signed
-                               assert_eq!(commitment_update.update_add_htlcs.len(), pending_htlc_adds.1 as usize);
-                       }
+                       assert_eq!(commitment_update.update_add_htlcs.len(), pending_htlc_adds.1);
                        assert_eq!(commitment_update.update_fulfill_htlcs.len(), pending_htlc_claims.1 + pending_cell_htlc_claims.1);
                        assert_eq!(commitment_update.update_fail_htlcs.len(), pending_htlc_fails.1 + pending_cell_htlc_fails.1);
                        assert!(commitment_update.update_fail_malformed_htlcs.is_empty());
@@ -3221,7 +3418,7 @@ pub fn reconnect_nodes<'a, 'b, 'c, 'd>(args: ReconnectArgs<'a, 'b, 'c, 'd>) {
                                node_b.node.handle_update_fail_htlc(&node_a.node.get_our_node_id(), &update_fail);
                        }
 
-                       if pending_htlc_adds.1 != -1 { // We use -1 to denote a response commitment_signed
+                       if !pending_responding_commitment_signed.1 {
                                commitment_signed_dance!(node_b, node_a, commitment_update.commitment_signed, false);
                        } else {
                                node_b.node.handle_commitment_signed(&node_a.node.get_our_node_id(), &commitment_update.commitment_signed);
@@ -3230,10 +3427,84 @@ pub fn reconnect_nodes<'a, 'b, 'c, 'd>(args: ReconnectArgs<'a, 'b, 'c, 'd>) {
                                // No commitment_signed so get_event_msg's assert(len == 1) passes
                                node_a.node.handle_revoke_and_ack(&node_b.node.get_our_node_id(), &bs_revoke_and_ack);
                                assert!(node_a.node.get_and_clear_pending_msg_events().is_empty());
-                               check_added_monitors!(node_a, 1);
+                               check_added_monitors!(node_a, if pending_responding_commitment_signed_dup_monitor.1 { 0 } else { 1 });
                        }
                } else {
                        assert!(chan_msgs.2.is_none());
                }
        }
 }
+
+/// Initiates channel opening and creates a single batch funding transaction.
+/// This will go through the open_channel / accept_channel flow, and return the batch funding
+/// transaction with corresponding funding_created messages.
+pub fn create_batch_channel_funding<'a, 'b, 'c>(
+       funding_node: &Node<'a, 'b, 'c>,
+       params: &[(&Node<'a, 'b, 'c>, u64, u64, u128, Option<UserConfig>)],
+) -> (Transaction, Vec<msgs::FundingCreated>) {
+       let mut tx_outs = Vec::new();
+       let mut temp_chan_ids = Vec::new();
+       let mut funding_created_msgs = Vec::new();
+
+       for (other_node, channel_value_satoshis, push_msat, user_channel_id, override_config) in params {
+               // Initialize channel opening.
+               let temp_chan_id = funding_node.node.create_channel(
+                       other_node.node.get_our_node_id(), *channel_value_satoshis, *push_msat, *user_channel_id,
+                       None,
+                       *override_config,
+               ).unwrap();
+               let open_channel_msg = get_event_msg!(funding_node, MessageSendEvent::SendOpenChannel, other_node.node.get_our_node_id());
+               other_node.node.handle_open_channel(&funding_node.node.get_our_node_id(), &open_channel_msg);
+               let accept_channel_msg = get_event_msg!(other_node, MessageSendEvent::SendAcceptChannel, funding_node.node.get_our_node_id());
+               funding_node.node.handle_accept_channel(&other_node.node.get_our_node_id(), &accept_channel_msg);
+
+               // Create the corresponding funding output.
+               let events = funding_node.node.get_and_clear_pending_events();
+               assert_eq!(events.len(), 1);
+               match events[0] {
+                       Event::FundingGenerationReady {
+                               ref temporary_channel_id,
+                               ref counterparty_node_id,
+                               channel_value_satoshis: ref event_channel_value_satoshis,
+                               ref output_script,
+                               user_channel_id: ref event_user_channel_id
+                       } => {
+                               assert_eq!(temporary_channel_id, &temp_chan_id);
+                               assert_eq!(counterparty_node_id, &other_node.node.get_our_node_id());
+                               assert_eq!(channel_value_satoshis, event_channel_value_satoshis);
+                               assert_eq!(user_channel_id, event_user_channel_id);
+                               tx_outs.push(TxOut {
+                                       value: *channel_value_satoshis, script_pubkey: output_script.clone(),
+                               });
+                       },
+                       _ => panic!("Unexpected event"),
+               };
+               temp_chan_ids.push((temp_chan_id, other_node.node.get_our_node_id()));
+       }
+
+       // Compose the batch funding transaction and give it to the ChannelManager.
+       let tx = Transaction {
+               version: 2,
+               lock_time: PackedLockTime::ZERO,
+               input: Vec::new(),
+               output: tx_outs,
+       };
+       assert!(funding_node.node.batch_funding_transaction_generated(
+               temp_chan_ids.iter().map(|(a, b)| (a, b)).collect::<Vec<_>>().as_slice(),
+               tx.clone(),
+       ).is_ok());
+       check_added_monitors!(funding_node, 0);
+       let events = funding_node.node.get_and_clear_pending_msg_events();
+       assert_eq!(events.len(), params.len());
+       for (other_node, ..) in params {
+               let funding_created = events
+                       .iter()
+                       .find_map(|event| match event {
+                               MessageSendEvent::SendFundingCreated { node_id, msg } if node_id == &other_node.node.get_our_node_id() => Some(msg.clone()),
+                               _ => None,
+                       })
+                       .unwrap();
+               funding_created_msgs.push(funding_created);
+       }
+       return (tx, funding_created_msgs);
+}