multi: update to latest BitcoindClient interface

In this commit, we introduce a nice optimization with regards to lnd's
interaction with a bitcoind backend. Within lnd, we currently have three
different subsystems responsible for watching the chain: chainntnfs,
lnwallet, and routing/chainview. Each of these subsystems has an active
RPC and ZMQ connection to the underlying bitcoind node. This would incur
a toll on the underlying bitcoind node and would cause us to miss ZMQ
events, which are crucial to lnd. We remedy this issue by sharing the
same connection to a bitcoind node between the different clients within
lnd.

chainntnfs/bitcoindnotify/bitcoind.go

@@ -8,9 +8,7 @@ import (
 	"time"
 
 	"github.com/btcsuite/btcd/btcjson"
-	"github.com/btcsuite/btcd/chaincfg"
 	"github.com/btcsuite/btcd/chaincfg/chainhash"
-	"github.com/btcsuite/btcd/rpcclient"
 	"github.com/btcsuite/btcd/wire"
 	"github.com/btcsuite/btcutil"
 	"github.com/btcsuite/btcwallet/chain"
@@ -87,8 +85,7 @@ var _ chainntnfs.ChainNotifier = (*BitcoindNotifier)(nil)
 // New returns a new BitcoindNotifier instance. This function assumes the
 // bitcoind node  detailed in the passed configuration is already running, and
 // willing to accept RPC requests and new zmq clients.
-func New(config *rpcclient.ConnConfig, zmqConnect string,
+func New(chainConn *chain.BitcoindConn) *BitcoindNotifier {
-	params chaincfg.Params) (*BitcoindNotifier, error) {
 	notifier := &BitcoindNotifier{
 		notificationCancels:  make(chan interface{}),
 		notificationRegistry: make(chan interface{}),
@@ -100,18 +97,9 @@ func New(config *rpcclient.ConnConfig, zmqConnect string,
 		quit: make(chan struct{}),
 	}
 
-	// Disable connecting to bitcoind within the rpcclient.New method. We
+	notifier.chainConn = chainConn.NewBitcoindClient(time.Unix(0, 0))
-	// defer establishing the connection to our .Start() method.
-	config.DisableConnectOnNew = true
-	config.DisableAutoReconnect = false
-	chainConn, err := chain.NewBitcoindClient(&params, config.Host,
-		config.User, config.Pass, zmqConnect, 100*time.Millisecond)
-	if err != nil {
-		return nil, err
-	}
-	notifier.chainConn = chainConn
 
-	return notifier, nil
+	return notifier
 }
 
 // Start connects to the running bitcoind node over websockets, registers for

chainntnfs/bitcoindnotify/driver.go

@@ -3,38 +3,25 @@ package bitcoindnotify
 import (
 	"fmt"
 
-	"github.com/btcsuite/btcd/chaincfg"
+	"github.com/btcsuite/btcwallet/chain"
-	"github.com/btcsuite/btcd/rpcclient"
 	"github.com/lightningnetwork/lnd/chainntnfs"
 )
 
 // createNewNotifier creates a new instance of the ChainNotifier interface
 // implemented by BitcoindNotifier.
 func createNewNotifier(args ...interface{}) (chainntnfs.ChainNotifier, error) {
-	if len(args) != 3 {
+	if len(args) != 1 {
 		return nil, fmt.Errorf("incorrect number of arguments to "+
-			".New(...), expected 3, instead passed %v", len(args))
+			".New(...), expected 1, instead passed %v", len(args))
 	}
 
-	config, ok := args[0].(*rpcclient.ConnConfig)
+	chainConn, ok := args[0].(*chain.BitcoindConn)
 	if !ok {
-		return nil, fmt.Errorf("first argument to bitcoindnotifier." +
+		return nil, fmt.Errorf("first argument to bitcoindnotify.New " +
-			"New is incorrect, expected a *rpcclient.ConnConfig")
+			"is incorrect, expected a *chain.BitcoindConn")
 	}
 
-	zmqConnect, ok := args[1].(string)
+	return New(chainConn), nil
-	if !ok {
-		return nil, fmt.Errorf("second argument to bitcoindnotifier." +
-			"New is incorrect, expected a string")
-	}
-
-	params, ok := args[2].(chaincfg.Params)
-	if !ok {
-		return nil, fmt.Errorf("third argument to bitcoindnotifier." +
-			"New is incorrect, expected a chaincfg.Params")
-	}
-
-	return New(config, zmqConnect, params)
 }
 
 // init registers a driver for the BtcdNotifier concrete implementation of the

chainntnfs/interface_test.go

@@ -13,19 +13,19 @@ import (
 	"testing"
 	"time"
 
-	"github.com/btcsuite/btcd/chaincfg/chainhash"
-	"github.com/btcsuite/btcwallet/walletdb"
-	"github.com/lightninglabs/neutrino"
-	"github.com/lightningnetwork/lnd/chainntnfs"
-	"github.com/ltcsuite/ltcd/btcjson"
-
 	"github.com/btcsuite/btcd/btcec"
+	"github.com/btcsuite/btcd/btcjson"
 	"github.com/btcsuite/btcd/chaincfg"
+	"github.com/btcsuite/btcd/chaincfg/chainhash"
 	"github.com/btcsuite/btcd/integration/rpctest"
 	"github.com/btcsuite/btcd/rpcclient"
 	"github.com/btcsuite/btcd/txscript"
 	"github.com/btcsuite/btcd/wire"
 	"github.com/btcsuite/btcutil"
+	"github.com/btcsuite/btcwallet/chain"
+	"github.com/btcsuite/btcwallet/walletdb"
+	"github.com/lightninglabs/neutrino"
+	"github.com/lightningnetwork/lnd/chainntnfs"
 
 	// Required to auto-register the bitcoind backed ChainNotifier
 	// implementation.
@@ -1375,7 +1375,8 @@ func TestInterfaces(t *testing.T) {
 			if err != nil {
 				t.Fatalf("Unable to create temp dir: %v", err)
 			}
-			zmqPath := "ipc:///" + tempBitcoindDir + "/weks.socket"
+			zmqBlockHost := "ipc:///" + tempBitcoindDir + "/blocks.socket"
+			zmqTxHost := "ipc:///" + tempBitcoindDir + "/tx.socket"
 			cleanUp1 := func() {
 				os.RemoveAll(tempBitcoindDir)
 			}
@@ -1392,8 +1393,8 @@ func TestInterfaces(t *testing.T) {
 					"220110063096c221be9933c82d38e1",
 				fmt.Sprintf("-rpcport=%d", rpcPort),
 				"-disablewallet",
-				"-zmqpubrawblock="+zmqPath,
+				"-zmqpubrawblock="+zmqBlockHost,
-				"-zmqpubrawtx="+zmqPath,
+				"-zmqpubrawtx="+zmqTxHost,
 			)
 			err = bitcoind.Start()
 			if err != nil {
@@ -1410,20 +1411,26 @@ func TestInterfaces(t *testing.T) {
 			// Wait for the bitcoind instance to start up.
 			time.Sleep(time.Second)
 
-			// Start the FilteredChainView implementation instance.
+			host := fmt.Sprintf("127.0.0.1:%d", rpcPort)
-			config := rpcclient.ConnConfig{
+			chainConn, err := chain.NewBitcoindConn(
-				Host: fmt.Sprintf(
+				netParams, host, "weks", "weks", zmqBlockHost,
-					"127.0.0.1:%d", rpcPort),
+				zmqTxHost, 100*time.Millisecond,
-				User:                 "weks",
+			)
-				Pass:                 "weks",
+			if err != nil {
-				DisableAutoReconnect: false,
+				t.Fatalf("unable to establish connection to "+
-				DisableConnectOnNew:  true,
+					"bitcoind: %v", err)
-				DisableTLS:           true,
-				HTTPPostMode:         true,
 			}
+			if err := chainConn.Start(); err != nil {
+				t.Fatalf("unable to establish connection to "+
+					"bitcoind: %v", err)
+			}
+			cleanUp3 := func() {
+				chainConn.Stop()
+				cleanUp2()
+			}
+			cleanUp = cleanUp3
 
-			notifier, err = notifierDriver.New(&config, zmqPath,
+			notifier, err = notifierDriver.New(chainConn)
-				*netParams)
 			if err != nil {
 				t.Fatalf("unable to create %v notifier: %v",
 					notifierType, err)

chainregistry.go

@@ -171,9 +171,8 @@ func newChainControlFromConfig(cfg *config, chanDB *channeldb.DB,
 	}
 
 	var (
-		err          error
+		err     error
-		cleanUp      func()
+		cleanUp func()
-		bitcoindConn *chain.BitcoindClient
 	)
 
 	// If spv mode is active, then we'll be using a distinct set of
@@ -300,47 +299,37 @@ func newChainControlFromConfig(cfg *config, chanDB *channeldb.DB,
 			}
 		}
 
-		bitcoindUser := bitcoindMode.RPCUser
+		bitcoindConn, err := chain.NewBitcoindConn(
-		bitcoindPass := bitcoindMode.RPCPass
+			activeNetParams.Params, bitcoindHost,
+			bitcoindMode.RPCUser, bitcoindMode.RPCPass,
+			bitcoindMode.ZMQPubRawBlock, bitcoindMode.ZMQPubRawTx,
+			100*time.Millisecond,
+		)
+		if err != nil {
+			return nil, nil, err
+		}
+
+		cc.chainNotifier = bitcoindnotify.New(bitcoindConn)
+
+		// Next, we'll create an instance of the bitcoind chain view to
+		// be used within the routing layer.
+		cc.chainView = chainview.NewBitcoindFilteredChainView(bitcoindConn)
+
+		// Create a special rpc+ZMQ client for bitcoind which will be
+		// used by the wallet for notifications, calls, etc.
+		walletConfig.ChainSource = bitcoindConn.NewBitcoindClient(birthday)
+
+		// If we're not in regtest mode, then we'll attempt to use a
+		// proper fee estimator for testnet.
 		rpcConfig := &rpcclient.ConnConfig{
 			Host:                 bitcoindHost,
-			User:                 bitcoindUser,
+			User:                 bitcoindMode.RPCUser,
-			Pass:                 bitcoindPass,
+			Pass:                 bitcoindMode.RPCPass,
 			DisableConnectOnNew:  true,
 			DisableAutoReconnect: false,
 			DisableTLS:           true,
 			HTTPPostMode:         true,
 		}
-		cc.chainNotifier, err = bitcoindnotify.New(rpcConfig,
-			bitcoindMode.ZMQPath, *activeNetParams.Params)
-		if err != nil {
-			return nil, nil, err
-		}
-
-		// Next, we'll create an instance of the bitcoind chain view to
-		// be used within the routing layer.
-		cc.chainView, err = chainview.NewBitcoindFilteredChainView(
-			*rpcConfig, bitcoindMode.ZMQPath,
-			*activeNetParams.Params)
-		if err != nil {
-			srvrLog.Errorf("unable to create chain view: %v", err)
-			return nil, nil, err
-		}
-
-		// Create a special rpc+ZMQ client for bitcoind which will be
-		// used by the wallet for notifications, calls, etc.
-		bitcoindConn, err = chain.NewBitcoindClient(
-			activeNetParams.Params, bitcoindHost, bitcoindUser,
-			bitcoindPass, bitcoindMode.ZMQPath,
-			time.Millisecond*100)
-		if err != nil {
-			return nil, nil, err
-		}
-
-		walletConfig.ChainSource = bitcoindConn
-
-		// If we're not in regtest mode, then we'll attempt to use a
-		// proper fee estimator for testnet.
 		if cfg.Bitcoin.Active && !cfg.Bitcoin.RegTest {
 			ltndLog.Infof("Initializing bitcoind backed fee estimator")
 

lnwallet/interface_test.go

@@ -16,18 +16,21 @@ import (
 	"testing"
 	"time"
 
-	"github.com/coreos/bbolt"
+	"github.com/btcsuite/btcd/btcec"
-	"github.com/davecgh/go-spew/spew"
-
-	"github.com/btcsuite/btcwallet/chain"
-	"github.com/btcsuite/btcwallet/walletdb"
-	_ "github.com/btcsuite/btcwallet/walletdb/bdb"
-	"github.com/lightninglabs/neutrino"
-
 	"github.com/btcsuite/btcd/btcjson"
 	"github.com/btcsuite/btcd/chaincfg"
 	"github.com/btcsuite/btcd/chaincfg/chainhash"
+	"github.com/btcsuite/btcd/integration/rpctest"
 	"github.com/btcsuite/btcd/rpcclient"
+	"github.com/btcsuite/btcd/txscript"
+	"github.com/btcsuite/btcd/wire"
+	"github.com/btcsuite/btcutil"
+	"github.com/btcsuite/btcwallet/chain"
+	"github.com/btcsuite/btcwallet/walletdb"
+	_ "github.com/btcsuite/btcwallet/walletdb/bdb"
+	"github.com/coreos/bbolt"
+	"github.com/davecgh/go-spew/spew"
+	"github.com/lightninglabs/neutrino"
 	"github.com/lightningnetwork/lnd/chainntnfs"
 	"github.com/lightningnetwork/lnd/chainntnfs/btcdnotify"
 	"github.com/lightningnetwork/lnd/channeldb"
@@ -35,12 +38,6 @@ import (
 	"github.com/lightningnetwork/lnd/lnwallet"
 	"github.com/lightningnetwork/lnd/lnwallet/btcwallet"
 	"github.com/lightningnetwork/lnd/lnwire"
-
-	"github.com/btcsuite/btcd/btcec"
-	"github.com/btcsuite/btcd/integration/rpctest"
-	"github.com/btcsuite/btcd/txscript"
-	"github.com/btcsuite/btcd/wire"
-	"github.com/btcsuite/btcutil"
 )
 
 var (
@@ -2150,7 +2147,8 @@ func runTests(t *testing.T, walletDriver *lnwallet.WalletDriver,
 			if err != nil {
 				t.Fatalf("unable to create temp directory: %v", err)
 			}
-			zmqPath := "ipc:///" + tempBitcoindDir + "/weks.socket"
+			zmqBlockHost := "ipc:///" + tempBitcoindDir + "/blocks.socket"
+			zmqTxHost := "ipc:///" + tempBitcoindDir + "/tx.socket"
 			defer os.RemoveAll(tempBitcoindDir)
 			rpcPort := rand.Int()%(65536-1024) + 1024
 			bitcoind := exec.Command(
@@ -2164,8 +2162,8 @@ func runTests(t *testing.T, walletDriver *lnwallet.WalletDriver,
 					"220110063096c221be9933c82d38e1",
 				fmt.Sprintf("-rpcport=%d", rpcPort),
 				"-disablewallet",
-				"-zmqpubrawblock="+zmqPath,
+				"-zmqpubrawblock="+zmqBlockHost,
-				"-zmqpubrawtx="+zmqPath,
+				"-zmqpubrawtx="+zmqTxHost,
 			)
 			err = bitcoind.Start()
 			if err != nil {
@@ -2174,21 +2172,28 @@ func runTests(t *testing.T, walletDriver *lnwallet.WalletDriver,
 			defer bitcoind.Wait()
 			defer bitcoind.Process.Kill()
 
-			// Start an Alice btcwallet bitcoind back end instance.
+			// Wait for the bitcoind instance to start up.
-			aliceClient, err = chain.NewBitcoindClient(netParams,
+			time.Sleep(time.Second)
-				fmt.Sprintf("127.0.0.1:%d", rpcPort), "weks",
-				"weks", zmqPath, 100*time.Millisecond)
-			if err != nil {
-				t.Fatalf("couldn't start alice client: %v", err)
-			}
 
-			// Start a Bob btcwallet bitcoind back end instance.
+			host := fmt.Sprintf("127.0.0.1:%d", rpcPort)
-			bobClient, err = chain.NewBitcoindClient(netParams,
+			chainConn, err := chain.NewBitcoindConn(
-				fmt.Sprintf("127.0.0.1:%d", rpcPort), "weks",
+				netParams, host, "weks", "weks", zmqBlockHost,
-				"weks", zmqPath, 100*time.Millisecond)
+				zmqTxHost, 100*time.Millisecond,
+			)
 			if err != nil {
-				t.Fatalf("couldn't start bob client: %v", err)
+				t.Fatalf("unable to establish connection to "+
+					"bitcoind: %v", err)
 			}
+			if err := chainConn.Start(); err != nil {
+				t.Fatalf("unable to establish connection to "+
+					"bitcoind: %v", err)
+			}
+			defer chainConn.Stop()
+
+			// Create a btcwallet bitcoind client for both Alice and
+			// Bob.
+			aliceClient = chainConn.NewBitcoindClient(time.Unix(0, 0))
+			bobClient = chainConn.NewBitcoindClient(time.Unix(0, 0))
 		default:
 			t.Fatalf("unknown chain driver: %v", backEnd)
 		}

routing/chainview/bitcoind.go

@@ -9,9 +9,7 @@ import (
 	"time"
 
 	"github.com/btcsuite/btcd/btcjson"
-	"github.com/btcsuite/btcd/chaincfg"
 	"github.com/btcsuite/btcd/chaincfg/chainhash"
-	"github.com/btcsuite/btcd/rpcclient"
 	"github.com/btcsuite/btcd/wire"
 	"github.com/btcsuite/btcwallet/chain"
 	"github.com/btcsuite/btcwallet/wtxmgr"
@@ -63,9 +61,8 @@ var _ FilteredChainView = (*BitcoindFilteredChainView)(nil)
 
 // NewBitcoindFilteredChainView creates a new instance of a FilteredChainView
 // from RPC credentials and a ZMQ socket address for a bitcoind instance.
-func NewBitcoindFilteredChainView(config rpcclient.ConnConfig,
+func NewBitcoindFilteredChainView(
-	zmqConnect string, params chaincfg.Params) (*BitcoindFilteredChainView,
+	chainConn *chain.BitcoindConn) *BitcoindFilteredChainView {
-	error) {
 
 	chainView := &BitcoindFilteredChainView{
 		chainFilter:     make(map[wire.OutPoint]struct{}),
@@ -74,16 +71,10 @@ func NewBitcoindFilteredChainView(config rpcclient.ConnConfig,
 		quit:            make(chan struct{}),
 	}
 
-	chainConn, err := chain.NewBitcoindClient(&params, config.Host,
+	chainView.chainClient = chainConn.NewBitcoindClient(time.Unix(0, 0))
-		config.User, config.Pass, zmqConnect, 100*time.Millisecond)
-	if err != nil {
-		return nil, err
-	}
-	chainView.chainClient = chainConn
-
 	chainView.blockQueue = newBlockEventQueue()
 
-	return chainView, nil
+	return chainView
 }
 
 // Start starts all goroutines necessary for normal operation.
@@ -325,9 +316,11 @@ func (b *BitcoindFilteredChainView) chainFilterer() {
 			// will cause all following notifications from and
 			// calls to it return blocks filtered with the new
 			// filter.
-			b.chainClient.LoadTxFilter(
+			err := b.chainClient.LoadTxFilter(false, update.newUtxos)
-				false, update.newUtxos,
+			if err != nil {
-			)
+				log.Errorf("Unable to update filter: %v", err)
+				continue
+			}
 
 			// All blocks gotten after we loaded the filter will
 			// have the filter applied, but we will need to rescan

routing/chainview/interface_test.go

@@ -13,6 +13,7 @@ import (
 	"time"
 
 	"github.com/btcsuite/btcd/btcec"
+	"github.com/btcsuite/btcd/btcjson"
 	"github.com/btcsuite/btcd/chaincfg"
 	"github.com/btcsuite/btcd/chaincfg/chainhash"
 	"github.com/btcsuite/btcd/integration/rpctest"
@@ -20,12 +21,12 @@ import (
 	"github.com/btcsuite/btcd/txscript"
 	"github.com/btcsuite/btcd/wire"
 	"github.com/btcsuite/btcutil"
-	"github.com/lightninglabs/neutrino"
+	"github.com/btcsuite/btcwallet/chain"
-	"github.com/lightningnetwork/lnd/channeldb"
-	"github.com/ltcsuite/ltcd/btcjson"
-
 	"github.com/btcsuite/btcwallet/walletdb"
 	_ "github.com/btcsuite/btcwallet/walletdb/bdb" // Required to register the boltdb walletdb implementation.
+
+	"github.com/lightninglabs/neutrino"
+	"github.com/lightningnetwork/lnd/channeldb"
 )
 
 var (
@@ -776,7 +777,8 @@ var interfaceImpls = []struct {
 			if err != nil {
 				return nil, nil, err
 			}
-			zmqPath := "ipc:///" + tempBitcoindDir + "/weks.socket"
+			zmqBlockHost := "ipc:///" + tempBitcoindDir + "/blocks.socket"
+			zmqTxHost := "ipc:///" + tempBitcoindDir + "/tx.socket"
 			cleanUp1 := func() {
 				os.RemoveAll(tempBitcoindDir)
 			}
@@ -792,8 +794,8 @@ var interfaceImpls = []struct {
 					"220110063096c221be9933c82d38e1",
 				fmt.Sprintf("-rpcport=%d", rpcPort),
 				"-disablewallet",
-				"-zmqpubrawblock="+zmqPath,
+				"-zmqpubrawblock="+zmqBlockHost,
-				"-zmqpubrawtx="+zmqPath,
+				"-zmqpubrawtx="+zmqTxHost,
 			)
 			err = bitcoind.Start()
 			if err != nil {
@@ -809,25 +811,30 @@ var interfaceImpls = []struct {
 			// Wait for the bitcoind instance to start up.
 			time.Sleep(time.Second)
 
-			// Start the FilteredChainView implementation instance.
+			host := fmt.Sprintf("127.0.0.1:%d", rpcPort)
-			config := rpcclient.ConnConfig{
+			chainConn, err := chain.NewBitcoindConn(
-				Host: fmt.Sprintf(
+				&chaincfg.RegressionNetParams, host, "weks",
-					"127.0.0.1:%d", rpcPort),
+				"weks", zmqBlockHost, zmqTxHost,
-				User:                 "weks",
+				100*time.Millisecond,
-				Pass:                 "weks",
+			)
-				DisableAutoReconnect: false,
+			if err != nil {
-				DisableConnectOnNew:  true,
+				return cleanUp2, nil, fmt.Errorf("unable to "+
-				DisableTLS:           true,
+					"establish connection to bitcoind: %v",
-				HTTPPostMode:         true,
+					err)
+			}
+			if err := chainConn.Start(); err != nil {
+				return cleanUp2, nil, fmt.Errorf("unable to "+
+					"establish connection to bitcoind: %v",
+					err)
+			}
+			cleanUp3 := func() {
+				chainConn.Stop()
+				cleanUp2()
 			}
 
-			chainView, err := NewBitcoindFilteredChainView(config,
+			chainView := NewBitcoindFilteredChainView(chainConn)
-				zmqPath, chaincfg.RegressionNetParams)
+
-			if err != nil {
+			return cleanUp3, chainView, nil
-				cleanUp2()
-				return nil, nil, err
-			}
-			return cleanUp2, chainView, nil
 		},
 	},
 	{