Merge pull request #2703 from wpaulino/retryable-commitment-broadcast
authorMatt Corallo <649246+TheBlueMatt@users.noreply.github.com>
Wed, 13 Dec 2023 05:15:54 +0000 (05:15 +0000)
committerGitHub <noreply@github.com>
Wed, 13 Dec 2023 05:15:54 +0000 (05:15 +0000)
Refactor commitment broadcast to always go through OnchainTxHandler

1  2 
lightning/src/chain/channelmonitor.rs
lightning/src/ln/payment_tests.rs

index 1dab9c8c8ffba1456c78671b8150323a595ee181,e862b5cdde52636019dc548e720caa8f145c1352..244384faa224f68bafd5df1b02cda326ec722493
@@@ -1382,22 -1382,15 +1382,22 @@@ impl<Signer: WriteableEcdsaChannelSigne
        /// Loads the funding txo and outputs to watch into the given `chain::Filter` by repeatedly
        /// calling `chain::Filter::register_output` and `chain::Filter::register_tx` until all outputs
        /// have been registered.
 -      pub fn load_outputs_to_watch<F: Deref>(&self, filter: &F) where F::Target: chain::Filter {
 +      pub fn load_outputs_to_watch<F: Deref, L: Deref>(&self, filter: &F, logger: &L) 
 +      where 
 +              F::Target: chain::Filter, L::Target: Logger,
 +      {
                let lock = self.inner.lock().unwrap();
 +              let logger = WithChannelMonitor::from_impl(logger, &*lock);
 +              log_trace!(&logger, "Registering funding outpoint {}", &lock.get_funding_txo().0);
                filter.register_tx(&lock.get_funding_txo().0.txid, &lock.get_funding_txo().1);
                for (txid, outputs) in lock.get_outputs_to_watch().iter() {
                        for (index, script_pubkey) in outputs.iter() {
                                assert!(*index <= u16::max_value() as u32);
 +                              let outpoint = OutPoint { txid: *txid, index: *index as u16 };
 +                              log_trace!(logger, "Registering outpoint {} with the filter for monitoring spends", outpoint);
                                filter.register_output(WatchedOutput {
                                        block_hash: None,
 -                                      outpoint: OutPoint { txid: *txid, index: *index as u16 },
 +                                      outpoint,
                                        script_pubkey: script_pubkey.clone(),
                                });
                        }
@@@ -2291,7 -2284,7 +2291,7 @@@ macro_rules! fail_unbroadcast_htlcs 
                                                        // broadcastable commitment transaction has the HTLC in it, but it
                                                        // cannot currently change after channel initialization, so we don't
                                                        // need to here.
 -                                                      let confirmed_htlcs_iter: &mut Iterator<Item = (&HTLCOutputInCommitment, Option<&HTLCSource>)> = &mut $confirmed_htlcs_list;
 +                                                      let confirmed_htlcs_iter: &mut dyn Iterator<Item = (&HTLCOutputInCommitment, Option<&HTLCSource>)> = &mut $confirmed_htlcs_list;
  
                                                        let mut matched_htlc = false;
                                                        for (ref broadcast_htlc, ref broadcast_source) in confirmed_htlcs_iter {
@@@ -2666,18 -2659,59 +2666,59 @@@ impl<Signer: WriteableEcdsaChannelSigne
                }
        }
  
-       fn broadcast_latest_holder_commitment_txn<B: Deref, L: Deref>(&mut self, broadcaster: &B, logger: &WithChannelMonitor<L>)
-               where B::Target: BroadcasterInterface,
-                       L::Target: Logger,
-       {
-               let commit_txs = self.get_latest_holder_commitment_txn(logger);
-               let mut txs = vec![];
-               for tx in commit_txs.iter() {
-                       log_info!(logger, "Broadcasting local {}", log_tx!(tx));
-                       txs.push(tx);
-               }
-               broadcaster.broadcast_transactions(&txs);
+       fn generate_claimable_outpoints_and_watch_outputs(&mut self) -> (Vec<PackageTemplate>, Vec<TransactionOutputs>) {
+               let funding_outp = HolderFundingOutput::build(
+                       self.funding_redeemscript.clone(),
+                       self.channel_value_satoshis,
+                       self.onchain_tx_handler.channel_type_features().clone()
+               );
+               let commitment_package = PackageTemplate::build_package(
+                       self.funding_info.0.txid.clone(), self.funding_info.0.index as u32,
+                       PackageSolvingData::HolderFundingOutput(funding_outp),
+                       self.best_block.height(), self.best_block.height()
+               );
+               let mut claimable_outpoints = vec![commitment_package];
                self.pending_monitor_events.push(MonitorEvent::HolderForceClosed(self.funding_info.0));
+               // Although we aren't signing the transaction directly here, the transaction will be signed
+               // in the claim that is queued to OnchainTxHandler. We set holder_tx_signed here to reject
+               // new channel updates.
+               self.holder_tx_signed = true;
+               let mut watch_outputs = Vec::new();
+               // We can't broadcast our HTLC transactions while the commitment transaction is
+               // unconfirmed. We'll delay doing so until we detect the confirmed commitment in
+               // `transactions_confirmed`.
+               if !self.onchain_tx_handler.channel_type_features().supports_anchors_zero_fee_htlc_tx() {
+                       // Because we're broadcasting a commitment transaction, we should construct the package
+                       // assuming it gets confirmed in the next block. Sadly, we have code which considers
+                       // "not yet confirmed" things as discardable, so we cannot do that here.
+                       let (mut new_outpoints, _) = self.get_broadcasted_holder_claims(
+                               &self.current_holder_commitment_tx, self.best_block.height()
+                       );
+                       let unsigned_commitment_tx = self.onchain_tx_handler.get_unsigned_holder_commitment_tx();
+                       let new_outputs = self.get_broadcasted_holder_watch_outputs(
+                               &self.current_holder_commitment_tx, &unsigned_commitment_tx
+                       );
+                       if !new_outputs.is_empty() {
+                               watch_outputs.push((self.current_holder_commitment_tx.txid.clone(), new_outputs));
+                       }
+                       claimable_outpoints.append(&mut new_outpoints);
+               }
+               (claimable_outpoints, watch_outputs)
+       }
+       pub(crate) fn queue_latest_holder_commitment_txn_for_broadcast<B: Deref, F: Deref, L: Deref>(
+               &mut self, broadcaster: &B, fee_estimator: &LowerBoundedFeeEstimator<F>, logger: &WithChannelMonitor<L>
+       )
+       where
+               B::Target: BroadcasterInterface,
+               F::Target: FeeEstimator,
+               L::Target: Logger,
+       {
+               let (claimable_outpoints, _) = self.generate_claimable_outpoints_and_watch_outputs();
+               self.onchain_tx_handler.update_claims_view_from_requests(
+                       claimable_outpoints, self.best_block.height(), self.best_block.height(), broadcaster,
+                       fee_estimator, logger
+               );
        }
  
        fn update_monitor<B: Deref, F: Deref, L: Deref>(
                                                        log_trace!(logger, "Avoiding commitment broadcast, already detected confirmed spend onchain");
                                                        continue;
                                                }
-                                               self.broadcast_latest_holder_commitment_txn(broadcaster, logger);
-                                               // If the channel supports anchor outputs, we'll need to emit an external
-                                               // event to be consumed such that a child transaction is broadcast with a
-                                               // high enough feerate for the parent commitment transaction to confirm.
-                                               if self.onchain_tx_handler.channel_type_features().supports_anchors_zero_fee_htlc_tx() {
-                                                       let funding_output = HolderFundingOutput::build(
-                                                               self.funding_redeemscript.clone(), self.channel_value_satoshis,
-                                                               self.onchain_tx_handler.channel_type_features().clone(),
-                                                       );
-                                                       let best_block_height = self.best_block.height();
-                                                       let commitment_package = PackageTemplate::build_package(
-                                                               self.funding_info.0.txid.clone(), self.funding_info.0.index as u32,
-                                                               PackageSolvingData::HolderFundingOutput(funding_output),
-                                                               best_block_height, best_block_height
-                                                       );
-                                                       self.onchain_tx_handler.update_claims_view_from_requests(
-                                                               vec![commitment_package], best_block_height, best_block_height,
-                                                               broadcaster, &bounded_fee_estimator, logger,
-                                                       );
-                                               }
+                                               self.queue_latest_holder_commitment_txn_for_broadcast(broadcaster, &bounded_fee_estimator, logger);
                                        } else if !self.holder_tx_signed {
                                                log_error!(logger, "WARNING: You have a potentially-unsafe holder commitment transaction available to broadcast");
                                                log_error!(logger, "    in channel monitor for channel {}!", &self.funding_info.0.to_channel_id());
                }
        }
  
+       /// Cancels any existing pending claims for a commitment that previously confirmed and has now
+       /// been replaced by another.
+       pub fn cancel_prev_commitment_claims<L: Deref>(
+               &mut self, logger: &L, confirmed_commitment_txid: &Txid
+       ) where L::Target: Logger {
+               for (counterparty_commitment_txid, _) in &self.counterparty_commitment_txn_on_chain {
+                       // Cancel any pending claims for counterparty commitments we've seen confirm.
+                       if counterparty_commitment_txid == confirmed_commitment_txid {
+                               continue;
+                       }
+                       for (htlc, _) in self.counterparty_claimable_outpoints.get(counterparty_commitment_txid).unwrap_or(&vec![]) {
+                               log_trace!(logger, "Canceling claims for previously confirmed counterparty commitment {}",
+                                       counterparty_commitment_txid);
+                               let mut outpoint = BitcoinOutPoint { txid: *counterparty_commitment_txid, vout: 0 };
+                               if let Some(vout) = htlc.transaction_output_index {
+                                       outpoint.vout = vout;
+                                       self.onchain_tx_handler.abandon_claim(&outpoint);
+                               }
+                       }
+               }
+               if self.holder_tx_signed {
+                       // If we've signed, we may have broadcast either commitment (prev or current), and
+                       // attempted to claim from it immediately without waiting for a confirmation.
+                       if self.current_holder_commitment_tx.txid != *confirmed_commitment_txid {
+                               log_trace!(logger, "Canceling claims for previously broadcast holder commitment {}",
+                                       self.current_holder_commitment_tx.txid);
+                               let mut outpoint = BitcoinOutPoint { txid: self.current_holder_commitment_tx.txid, vout: 0 };
+                               for (htlc, _, _) in &self.current_holder_commitment_tx.htlc_outputs {
+                                       if let Some(vout) = htlc.transaction_output_index {
+                                               outpoint.vout = vout;
+                                               self.onchain_tx_handler.abandon_claim(&outpoint);
+                                       }
+                               }
+                       }
+                       if let Some(prev_holder_commitment_tx) = &self.prev_holder_signed_commitment_tx {
+                               if prev_holder_commitment_tx.txid != *confirmed_commitment_txid {
+                                       log_trace!(logger, "Canceling claims for previously broadcast holder commitment {}",
+                                               prev_holder_commitment_tx.txid);
+                                       let mut outpoint = BitcoinOutPoint { txid: prev_holder_commitment_tx.txid, vout: 0 };
+                                       for (htlc, _, _) in &prev_holder_commitment_tx.htlc_outputs {
+                                               if let Some(vout) = htlc.transaction_output_index {
+                                                       outpoint.vout = vout;
+                                                       self.onchain_tx_handler.abandon_claim(&outpoint);
+                                               }
+                                       }
+                               }
+                       }
+               } else {
+                       // No previous claim.
+               }
+       }
        fn get_latest_holder_commitment_txn<L: Deref>(
                &mut self, logger: &WithChannelMonitor<L>,
        ) -> Vec<Transaction> where L::Target: Logger {
  
                if height > self.best_block.height() {
                        self.best_block = BestBlock::new(block_hash, height);
 +                      log_trace!(logger, "Connecting new block {} at height {}", block_hash, height);
                        self.block_confirmed(height, block_hash, vec![], vec![], vec![], &broadcaster, &fee_estimator, logger)
                } else if block_hash != self.best_block.block_hash() {
                        self.best_block = BestBlock::new(block_hash, height);
 +                      log_trace!(logger, "Best block re-orged, replaced with new block {} at height {}", block_hash, height);
                        self.onchain_events_awaiting_threshold_conf.retain(|ref entry| entry.height <= height);
                        self.onchain_tx_handler.block_disconnected(height + 1, broadcaster, fee_estimator, logger);
                        Vec::new()
                let mut claimable_outpoints = Vec::new();
                'tx_iter: for tx in &txn_matched {
                        let txid = tx.txid();
 +                      log_trace!(logger, "Transaction {} confirmed in block {}", txid , block_hash);
                        // If a transaction has already been confirmed, ensure we don't bother processing it duplicatively.
                        if Some(txid) == self.funding_spend_confirmed {
                                log_debug!(logger, "Skipping redundant processing of funding-spend tx {} as it was previously confirmed", txid);
                                                        commitment_tx_to_counterparty_output,
                                                },
                                        });
+                                       // Now that we've detected a confirmed commitment transaction, attempt to cancel
+                                       // pending claims for any commitments that were previously confirmed such that
+                                       // we don't continue claiming inputs that no longer exist.
+                                       self.cancel_prev_commitment_claims(&logger, &txid);
                                }
                        }
                        if tx.input.len() >= 1 {
  
                let should_broadcast = self.should_broadcast_holder_commitment_txn(logger);
                if should_broadcast {
-                       let funding_outp = HolderFundingOutput::build(self.funding_redeemscript.clone(), self.channel_value_satoshis, self.onchain_tx_handler.channel_type_features().clone());
-                       let commitment_package = PackageTemplate::build_package(self.funding_info.0.txid.clone(), self.funding_info.0.index as u32, PackageSolvingData::HolderFundingOutput(funding_outp), self.best_block.height(), self.best_block.height());
-                       claimable_outpoints.push(commitment_package);
-                       self.pending_monitor_events.push(MonitorEvent::HolderForceClosed(self.funding_info.0));
-                       // Although we aren't signing the transaction directly here, the transaction will be signed
-                       // in the claim that is queued to OnchainTxHandler. We set holder_tx_signed here to reject
-                       // new channel updates.
-                       self.holder_tx_signed = true;
-                       // We can't broadcast our HTLC transactions while the commitment transaction is
-                       // unconfirmed. We'll delay doing so until we detect the confirmed commitment in
-                       // `transactions_confirmed`.
-                       if !self.onchain_tx_handler.channel_type_features().supports_anchors_zero_fee_htlc_tx() {
-                               // Because we're broadcasting a commitment transaction, we should construct the package
-                               // assuming it gets confirmed in the next block. Sadly, we have code which considers
-                               // "not yet confirmed" things as discardable, so we cannot do that here.
-                               let (mut new_outpoints, _) = self.get_broadcasted_holder_claims(&self.current_holder_commitment_tx, self.best_block.height());
-                               let unsigned_commitment_tx = self.onchain_tx_handler.get_unsigned_holder_commitment_tx();
-                               let new_outputs = self.get_broadcasted_holder_watch_outputs(&self.current_holder_commitment_tx, &unsigned_commitment_tx);
-                               if !new_outputs.is_empty() {
-                                       watch_outputs.push((self.current_holder_commitment_tx.txid.clone(), new_outputs));
-                               }
-                               claimable_outpoints.append(&mut new_outpoints);
-                       }
+                       let (mut new_outpoints, mut new_outputs) = self.generate_claimable_outpoints_and_watch_outputs();
+                       claimable_outpoints.append(&mut new_outpoints);
+                       watch_outputs.append(&mut new_outputs);
                }
  
                // Find which on-chain events have reached their confirmation threshold.
index adc611e7e190833fc8a8f87346f2ba2009bab730,1b7041c9fa23ef42443dbe5fa34c989894c113a6..6af0e63c98bf421bade346f89f1fff303f48e764
@@@ -638,7 -638,7 +638,7 @@@ fn do_retry_with_no_persist(confirm_bef
        let nodes_0_deserialized;
        let mut nodes = create_network(3, &node_cfgs, &node_chanmgrs);
  
-       let chan_id = create_announced_chan_between_nodes(&nodes, 0, 1).2;
+       let (_, _, chan_id, funding_tx) = create_announced_chan_between_nodes(&nodes, 0, 1);
        let (_, _, chan_id_2, _) = create_announced_chan_between_nodes(&nodes, 1, 2);
  
        // Serialize the ChannelManager prior to sending payments
        assert_eq!(nodes[0].node.list_usable_channels().len(), 1);
  
        mine_transaction(&nodes[1], &as_commitment_tx);
-       let bs_htlc_claim_txn = nodes[1].tx_broadcaster.txn_broadcasted.lock().unwrap().split_off(0);
-       assert_eq!(bs_htlc_claim_txn.len(), 1);
-       check_spends!(bs_htlc_claim_txn[0], as_commitment_tx);
+       let bs_htlc_claim_txn = {
+               let mut txn = nodes[1].tx_broadcaster.unique_txn_broadcast();
+               assert_eq!(txn.len(), 2);
+               check_spends!(txn[0], funding_tx);
+               check_spends!(txn[1], as_commitment_tx);
+               txn.pop().unwrap()
+       };
  
        if !confirm_before_reload {
                mine_transaction(&nodes[0], &as_commitment_tx);
+               let txn = nodes[0].tx_broadcaster.unique_txn_broadcast();
+               assert_eq!(txn.len(), 1);
+               assert_eq!(txn[0].txid(), as_commitment_tx.txid());
        }
-       mine_transaction(&nodes[0], &bs_htlc_claim_txn[0]);
+       mine_transaction(&nodes[0], &bs_htlc_claim_txn);
        expect_payment_sent(&nodes[0], payment_preimage_1, None, true, false);
        connect_blocks(&nodes[0], TEST_FINAL_CLTV*4 + 20);
        let (first_htlc_timeout_tx, second_htlc_timeout_tx) = {
        };
        check_spends!(first_htlc_timeout_tx, as_commitment_tx);
        check_spends!(second_htlc_timeout_tx, as_commitment_tx);
-       if first_htlc_timeout_tx.input[0].previous_output == bs_htlc_claim_txn[0].input[0].previous_output {
+       if first_htlc_timeout_tx.input[0].previous_output == bs_htlc_claim_txn.input[0].previous_output {
                confirm_transaction(&nodes[0], &second_htlc_timeout_tx);
        } else {
                confirm_transaction(&nodes[0], &first_htlc_timeout_tx);
@@@ -919,19 -926,23 +926,23 @@@ fn do_test_completed_payment_not_retrya
        // the HTLC-Timeout transaction beyond 1 conf). For dust HTLCs, the HTLC is considered resolved
        // after the commitment transaction, so always connect the commitment transaction.
        mine_transaction(&nodes[0], &bs_commitment_tx[0]);
+       if nodes[0].connect_style.borrow().updates_best_block_first() {
+               let _ = nodes[0].tx_broadcaster.txn_broadcast();
+       }
        mine_transaction(&nodes[1], &bs_commitment_tx[0]);
        if !use_dust {
                connect_blocks(&nodes[0], TEST_FINAL_CLTV + (MIN_CLTV_EXPIRY_DELTA as u32));
                connect_blocks(&nodes[1], TEST_FINAL_CLTV + (MIN_CLTV_EXPIRY_DELTA as u32));
                let as_htlc_timeout = nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap().split_off(0);
-               check_spends!(as_htlc_timeout[0], bs_commitment_tx[0]);
                assert_eq!(as_htlc_timeout.len(), 1);
+               check_spends!(as_htlc_timeout[0], bs_commitment_tx[0]);
  
                mine_transaction(&nodes[0], &as_htlc_timeout[0]);
-               // nodes[0] may rebroadcast (or RBF-bump) its HTLC-Timeout, so wipe the announced set.
-               nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap().clear();
                mine_transaction(&nodes[1], &as_htlc_timeout[0]);
        }
+       if nodes[0].connect_style.borrow().updates_best_block_first() {
+               let _ = nodes[0].tx_broadcaster.txn_broadcast();
+       }
  
        // Create a new channel on which to retry the payment before we fail the payment via the
        // HTLC-Timeout transaction. This avoids ChannelManager timing out the payment due to us
@@@ -1049,32 -1060,36 +1060,36 @@@ fn do_test_dup_htlc_onchain_fails_on_re
  
        // Connect blocks until the CLTV timeout is up so that we get an HTLC-Timeout transaction
        connect_blocks(&nodes[0], TEST_FINAL_CLTV + LATENCY_GRACE_PERIOD_BLOCKS + 1);
-       let node_txn = nodes[0].tx_broadcaster.txn_broadcasted.lock().unwrap().split_off(0);
-       assert_eq!(node_txn.len(), 3);
-       assert_eq!(node_txn[0].txid(), node_txn[1].txid());
-       check_spends!(node_txn[1], funding_tx);
-       check_spends!(node_txn[2], node_txn[1]);
-       let timeout_txn = vec![node_txn[2].clone()];
+       let (commitment_tx, htlc_timeout_tx) = {
+               let mut txn = nodes[0].tx_broadcaster.unique_txn_broadcast();
+               assert_eq!(txn.len(), 2);
+               check_spends!(txn[0], funding_tx);
+               check_spends!(txn[1], txn[0]);
+               (txn.remove(0), txn.remove(0))
+       };
  
        nodes[1].node.claim_funds(payment_preimage);
        check_added_monitors!(nodes[1], 1);
        expect_payment_claimed!(nodes[1], payment_hash, 10_000_000);
  
-       connect_block(&nodes[1], &create_dummy_block(nodes[1].best_block_hash(), 42, vec![node_txn[1].clone()]));
+       mine_transaction(&nodes[1], &commitment_tx);
        check_closed_broadcast!(nodes[1], true);
        check_added_monitors!(nodes[1], 1);
        check_closed_event!(nodes[1], 1, ClosureReason::CommitmentTxConfirmed, [nodes[0].node.get_our_node_id()], 100000);
-       let claim_txn = nodes[1].tx_broadcaster.txn_broadcasted.lock().unwrap().split_off(0);
-       assert_eq!(claim_txn.len(), 1);
-       check_spends!(claim_txn[0], node_txn[1]);
+       let htlc_success_tx = {
+               let mut txn = nodes[1].tx_broadcaster.txn_broadcast();
+               assert_eq!(txn.len(), 1);
+               check_spends!(txn[0], commitment_tx);
+               txn.pop().unwrap()
+       };
  
-       connect_block(&nodes[0], &create_dummy_block(nodes[0].best_block_hash(), 42, vec![node_txn[1].clone()]));
+       mine_transaction(&nodes[0], &commitment_tx);
  
        if confirm_commitment_tx {
                connect_blocks(&nodes[0], BREAKDOWN_TIMEOUT as u32 - 1);
        }
  
-       let claim_block = create_dummy_block(nodes[0].best_block_hash(), 42, if payment_timeout { timeout_txn } else { vec![claim_txn[0].clone()] });
+       let claim_block = create_dummy_block(nodes[0].best_block_hash(), 42, if payment_timeout { vec![htlc_timeout_tx] } else { vec![htlc_success_tx] });
  
        if payment_timeout {
                assert!(confirm_commitment_tx); // Otherwise we're spending below our CSV!
@@@ -3339,7 -3354,6 +3354,7 @@@ fn test_threaded_payment_retries() 
                // We really want std::thread::scope, but its not stable until 1.63. Until then, we get unsafe.
                let node_ref = NodePtr::from_node(&nodes[0]);
                move || {
 +                      let _ = &node_ref;
                        let node_a = unsafe { &*node_ref.0 };
                        while Instant::now() < end_time {
                                node_a.node.get_and_clear_pending_events(); // wipe the PendingHTLCsForwardable