Merge pull request #7599 from Roasbeef/sweeper-rebroadcaster-interaction

sweep: remove conflicted sweep txns from the rebroadcaster

config_builder.go

@@ -644,7 +644,7 @@ func proxyBlockEpoch(notifier chainntnfs.ChainNotifier,
 }
 
 // walletReBroadcaster is a simple wrapper around the pushtx.Broadcaster
-// interface to adhere to the expanded lnwallet.Rebraodcaster interface.
+// interface to adhere to the expanded lnwallet.Rebroadcaster interface.
 type walletReBroadcaster struct {
 	started atomic.Bool
 

docs/release-notes/release-notes-0.16.1.md

@@ -97,12 +97,18 @@ available](https://github.com/lightningnetwork/lnd/pull/7529).
   funding manager would error out if no persisted initial forwarding policy is 
   found for a channel. 
 
+* The internal rebroacaster [will no longer continue to rebroadcast stale
+  sweeper transactions (have a conflict mined or in the
+  mempool)](https://github.com/lightningnetwork/lnd/pull/7599).
+
+
 # Contributors (Alphabetical Order)
 
 * ardevd
 * Elle Mouton
 * hieblmi
 * Oliver Gugger
+* Olaoluwa Osuntokun
 * Pierre Beugnet
 * Tommy Volk
 * Yong Yu

server.go

@@ -1056,7 +1056,7 @@ func newServer(cfg *Config, listenAddrs []net.Addr,
 		FeeEstimator:   cc.FeeEstimator,
 		GenSweepScript: newSweepPkScriptGen(cc.Wallet),
 		Signer:         cc.Wallet.Cfg.Signer,
-		Wallet:         cc.Wallet,
+		Wallet:         newSweeperWallet(cc.Wallet),
 		NewBatchTimer: func() <-chan time.Time {
 			return time.NewTimer(cfg.Sweeper.BatchWindowDuration).C
 		},

sweep/backend_mock_test.go

@@ -159,3 +159,6 @@ func (b *mockBackend) RemoveDescendants(*wire.MsgTx) error {
 func (b *mockBackend) FetchTx(chainhash.Hash) (*wire.MsgTx, error) {
 	return nil, nil
 }
+
+func (b *mockBackend) CancelRebroadcast(tx chainhash.Hash) {
+}

sweep/interface.go

@@ -35,4 +35,10 @@ type Wallet interface {
 	// hash passed in. If the transaction can't be found then a nil
 	// transaction pointer is returned.
 	FetchTx(chainhash.Hash) (*wire.MsgTx, error)
+
+	// CancelRebroadcast is used to inform the rebroadcaster sub-system
+	// that it no longer needs to try to rebroadcast a transaction. This is
+	// used to ensure that invalid transactions (inputs spent) aren't
+	// retried in the background.
+	CancelRebroadcast(tx chainhash.Hash)
 }

sweep/sweeper.go

@@ -555,6 +555,9 @@ func (s *UtxoSweeper) removeLastSweepDescendants(spendingTx *wire.MsgTx) error {
 		// Transaction wasn't found in the wallet, may have already
 		// been replaced/removed.
 		if sweepTx == nil {
+			// If it was removed, then we'll play it safe and mark
+			// it as no longer need to be rebroadcasted.
+			s.cfg.Wallet.CancelRebroadcast(sweepHash)
 			continue
 		}
 
@@ -579,6 +582,10 @@ func (s *UtxoSweeper) removeLastSweepDescendants(spendingTx *wire.MsgTx) error {
 			if err != nil {
 				log.Warnf("unable to remove descendants: %v", err)
 			}
+
+			// If this transaction was conflicting, then we'll stop
+			// rebroadcasting it in the background.
+			s.cfg.Wallet.CancelRebroadcast(sweepHash)
 		}
 	}
 

sweeper_wallet.go

@@ -0,0 +1,25 @@
+package lnd
+
+import (
+	"github.com/btcsuite/btcd/chaincfg/chainhash"
+	"github.com/lightningnetwork/lnd/lnwallet"
+)
+
+// sweeperWallet is a wrapper around the LightningWallet that implements the
+// sweeper's Wallet interface.
+type sweeperWallet struct {
+	*lnwallet.LightningWallet
+}
+
+// newSweeperWallet creates a new sweeper wallet from the given
+// LightningWallet.
+func newSweeperWallet(w *lnwallet.LightningWallet) *sweeperWallet {
+	return &sweeperWallet{
+		LightningWallet: w,
+	}
+}
+
+// CancelRebroadcast cancels the rebroadcast of the given transaction.
+func (s *sweeperWallet) CancelRebroadcast(txid chainhash.Hash) {
+	s.Cfg.Rebroadcaster.MarkAsConfirmed(txid)
+}