Merge pull request #8667 from lightningnetwork/elle-new-sweeper

Merge new sweeper branch to master

Author: Roasbeef

chainntnfs/bitcoindnotify/bitcoind.go

@@ -15,6 +15,7 @@ import (
 	"github.com/btcsuite/btcwallet/chain"
 	"github.com/lightningnetwork/lnd/blockcache"
 	"github.com/lightningnetwork/lnd/chainntnfs"
+	"github.com/lightningnetwork/lnd/fn"
 	"github.com/lightningnetwork/lnd/queue"
 )
 
@@ -1070,3 +1071,26 @@ func (b *BitcoindNotifier) CancelMempoolSpendEvent(
 
 	b.memNotifier.UnsubscribeEvent(sub)
 }
+
+// LookupInputMempoolSpend takes an outpoint and queries the mempool to find
+// its spending tx. Returns the tx if found, otherwise fn.None.
+//
+// NOTE: part of the MempoolWatcher interface.
+func (b *BitcoindNotifier) LookupInputMempoolSpend(
+	op wire.OutPoint) fn.Option[wire.MsgTx] {
+
+	// Find the spending txid.
+	txid, found := b.chainConn.LookupInputMempoolSpend(op)
+	if !found {
+		return fn.None[wire.MsgTx]()
+	}
+
+	// Query the spending tx using the id.
+	tx, err := b.chainConn.GetRawTransaction(&txid)
+	if err != nil {
+		// TODO(yy): enable logging errors in this package.
+		return fn.None[wire.MsgTx]()
+	}
+
+	return fn.Some(*tx.MsgTx().Copy())
+}

chainntnfs/btcdnotify/btcd.go

@@ -14,8 +14,10 @@ import (
 	"github.com/btcsuite/btcd/rpcclient"
 	"github.com/btcsuite/btcd/txscript"
 	"github.com/btcsuite/btcd/wire"
+	"github.com/btcsuite/btcwallet/chain"
 	"github.com/lightningnetwork/lnd/blockcache"
 	"github.com/lightningnetwork/lnd/chainntnfs"
+	"github.com/lightningnetwork/lnd/fn"
 	"github.com/lightningnetwork/lnd/queue"
 )
 
@@ -58,7 +60,7 @@ type BtcdNotifier struct {
 	active  int32 // To be used atomically.
 	stopped int32 // To be used atomically.
 
-	chainConn   *rpcclient.Client
+	chainConn   *chain.RPCClient
 	chainParams *chaincfg.Params
 
 	notificationCancels  chan interface{}
@@ -127,21 +129,30 @@ func New(config *rpcclient.ConnConfig, chainParams *chaincfg.Params,
 		quit: make(chan struct{}),
 	}
 
+	// Disable connecting to btcd within the rpcclient.New method. We
+	// defer establishing the connection to our .Start() method.
+	config.DisableConnectOnNew = true
+	config.DisableAutoReconnect = false
+
 	ntfnCallbacks := &rpcclient.NotificationHandlers{
 		OnBlockConnected:    notifier.onBlockConnected,
 		OnBlockDisconnected: notifier.onBlockDisconnected,
 		OnRedeemingTx:       notifier.onRedeemingTx,
 	}
 
-	// Disable connecting to btcd within the rpcclient.New method. We
-	// defer establishing the connection to our .Start() method.
-	config.DisableConnectOnNew = true
-	config.DisableAutoReconnect = false
-	chainConn, err := rpcclient.New(config, ntfnCallbacks)
+	rpcCfg := &chain.RPCClientConfig{
+		ReconnectAttempts:    20,
+		Conn:                 config,
+		Chain:                chainParams,
+		NotificationHandlers: ntfnCallbacks,
+	}
+
+	chainRPC, err := chain.NewRPCClientWithConfig(rpcCfg)
 	if err != nil {
 		return nil, err
 	}
-	notifier.chainConn = chainConn
+
+	notifier.chainConn = chainRPC
 
 	return notifier, nil
 }
@@ -1127,3 +1138,26 @@ func (b *BtcdNotifier) CancelMempoolSpendEvent(
 
 	b.memNotifier.UnsubscribeEvent(sub)
 }
+
+// LookupInputMempoolSpend takes an outpoint and queries the mempool to find
+// its spending tx. Returns the tx if found, otherwise fn.None.
+//
+// NOTE: part of the MempoolWatcher interface.
+func (b *BtcdNotifier) LookupInputMempoolSpend(
+	op wire.OutPoint) fn.Option[wire.MsgTx] {
+
+	// Find the spending txid.
+	txid, found := b.chainConn.LookupInputMempoolSpend(op)
+	if !found {
+		return fn.None[wire.MsgTx]()
+	}
+
+	// Query the spending tx using the id.
+	tx, err := b.chainConn.GetRawTransaction(&txid)
+	if err != nil {
+		// TODO(yy): enable logging errors in this package.
+		return fn.None[wire.MsgTx]()
+	}
+
+	return fn.Some(*tx.MsgTx().Copy())
+}

chainntnfs/interface.go

@@ -13,6 +13,7 @@ import (
 	"github.com/btcsuite/btcd/btcutil"
 	"github.com/btcsuite/btcd/chaincfg/chainhash"
 	"github.com/btcsuite/btcd/wire"
+	"github.com/lightningnetwork/lnd/fn"
 )
 
 var (
@@ -849,4 +850,9 @@ type MempoolWatcher interface {
 	// CancelMempoolSpendEvent allows the caller to cancel a subscription to
 	// watch for a spend of an outpoint in the mempool.
 	CancelMempoolSpendEvent(sub *MempoolSpendEvent)
+
+	// LookupInputMempoolSpend looks up the mempool to find a spending tx
+	// which spends the given outpoint. A fn.None is returned if it's not
+	// found.
+	LookupInputMempoolSpend(op wire.OutPoint) fn.Option[wire.MsgTx]
 }

chainntnfs/mocks.go

@@ -0,0 +1,123 @@
+package chainntnfs
+
+import (
+	"github.com/btcsuite/btcd/chaincfg/chainhash"
+	"github.com/btcsuite/btcd/wire"
+	"github.com/lightningnetwork/lnd/fn"
+	"github.com/stretchr/testify/mock"
+)
+
+// MockMempoolWatcher is a mock implementation of the MempoolWatcher interface.
+// This is used by other subsystems to mock the behavior of the mempool
+// watcher.
+type MockMempoolWatcher struct {
+	mock.Mock
+}
+
+// NewMockMempoolWatcher returns a new instance of a mock mempool watcher.
+func NewMockMempoolWatcher() *MockMempoolWatcher {
+	return &MockMempoolWatcher{}
+}
+
+// Compile-time check to ensure MockMempoolWatcher implements MempoolWatcher.
+var _ MempoolWatcher = (*MockMempoolWatcher)(nil)
+
+// SubscribeMempoolSpent implements the MempoolWatcher interface.
+func (m *MockMempoolWatcher) SubscribeMempoolSpent(
+	op wire.OutPoint) (*MempoolSpendEvent, error) {
+
+	args := m.Called(op)
+
+	if args.Get(0) == nil {
+		return nil, args.Error(1)
+	}
+
+	return args.Get(0).(*MempoolSpendEvent), args.Error(1)
+}
+
+// CancelMempoolSpendEvent implements the MempoolWatcher interface.
+func (m *MockMempoolWatcher) CancelMempoolSpendEvent(
+	sub *MempoolSpendEvent) {
+
+	m.Called(sub)
+}
+
+// LookupInputMempoolSpend looks up the mempool to find a spending tx which
+// spends the given outpoint.
+func (m *MockMempoolWatcher) LookupInputMempoolSpend(
+	op wire.OutPoint) fn.Option[wire.MsgTx] {
+
+	args := m.Called(op)
+
+	return args.Get(0).(fn.Option[wire.MsgTx])
+}
+
+// MockNotifier is a mock implementation of the ChainNotifier interface.
+type MockChainNotifier struct {
+	mock.Mock
+}
+
+// Compile-time check to ensure MockChainNotifier implements ChainNotifier.
+var _ ChainNotifier = (*MockChainNotifier)(nil)
+
+// RegisterConfirmationsNtfn registers an intent to be notified once txid
+// reaches numConfs confirmations.
+func (m *MockChainNotifier) RegisterConfirmationsNtfn(txid *chainhash.Hash,
+	pkScript []byte, numConfs, heightHint uint32,
+	opts ...NotifierOption) (*ConfirmationEvent, error) {
+
+	args := m.Called(txid, pkScript, numConfs, heightHint)
+	if args.Get(0) == nil {
+		return nil, args.Error(1)
+	}
+
+	return args.Get(0).(*ConfirmationEvent), args.Error(1)
+}
+
+// RegisterSpendNtfn registers an intent to be notified once the target
+// outpoint is successfully spent within a transaction.
+func (m *MockChainNotifier) RegisterSpendNtfn(outpoint *wire.OutPoint,
+	pkScript []byte, heightHint uint32) (*SpendEvent, error) {
+
+	args := m.Called(outpoint, pkScript, heightHint)
+	if args.Get(0) == nil {
+		return nil, args.Error(1)
+	}
+
+	return args.Get(0).(*SpendEvent), args.Error(1)
+}
+
+// RegisterBlockEpochNtfn registers an intent to be notified of each new block
+// connected to the tip of the main chain.
+func (m *MockChainNotifier) RegisterBlockEpochNtfn(epoch *BlockEpoch) (
+	*BlockEpochEvent, error) {
+
+	args := m.Called(epoch)
+	if args.Get(0) == nil {
+		return nil, args.Error(1)
+	}
+
+	return args.Get(0).(*BlockEpochEvent), args.Error(1)
+}
+
+// Start the ChainNotifier. Once started, the implementation should be ready,
+// and able to receive notification registrations from clients.
+func (m *MockChainNotifier) Start() error {
+	args := m.Called()
+
+	return args.Error(0)
+}
+
+// Started returns true if this instance has been started, and false otherwise.
+func (m *MockChainNotifier) Started() bool {
+	args := m.Called()
+
+	return args.Bool(0)
+}
+
+// Stops the concrete ChainNotifier.
+func (m *MockChainNotifier) Stop() error {
+	args := m.Called()
+
+	return args.Error(0)
+}