Merge pull request #891 from yyforyongyu/fix-rescan-race

chain+wallet: fix race over `RescanProgress` message

chain/bitcoind_client.go

@@ -803,7 +803,7 @@ func (c *BitcoindClient) onRescanProgress(hash *chainhash.Hash, height int32,
 
 	select {
 	case c.notificationQueue.ChanIn() <- &RescanProgress{
-		Hash:   hash,
+		Hash:   *hash,
 		Height: height,
 		Time:   timestamp,
 	}:

chain/bitcoind_conn.go

@@ -1,10 +1,13 @@
 package chain
 
 import (
+	"errors"
 	"fmt"
 	"net"
+	"strings"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"github.com/btcsuite/btcd/btcjson"
 	"github.com/btcsuite/btcd/chaincfg"
@@ -38,8 +41,21 @@ const (
 	// errBlockPrunedStr is the error message returned by bitcoind upon
 	// calling GetBlock on a pruned block.
 	errBlockPrunedStr = "Block not available (pruned data)"
+
+	// errStillLoadingCode is the error code returned when an RPC request
+	// is made but bitcoind is still in the process of loading or verifying
+	// blocks.
+	errStillLoadingCode = "-28"
+
+	// bitcoindStartTimeout is the time we wait for bitcoind to finish
+	// loading and verifying blocks and become ready to serve RPC requests.
+	bitcoindStartTimeout = 30 * time.Second
 )
 
+// ErrBitcoindStartTimeout is returned when the bitcoind daemon fails to load
+// and verify blocks under 30s during startup.
+var ErrBitcoindStartTimeout = errors.New("bitcoind start timeout")
+
 // BitcoindConfig contains all of the parameters required to establish a
 // connection to a bitcoind's RPC.
 type BitcoindConfig struct {
@@ -311,9 +327,57 @@ func (c *BitcoindConn) sendTxToClients() {
 	}
 }
 
+// getBlockHashDuringStartup is used to call the getblockhash RPC during
+// startup. It catches the case where bitcoind is still in the process of
+// loading blocks, which returns the following error,
+// - "-28: Loading block index..."
+// - "-28: Verifying blocks..."
+// In this case we'd retry every second until we time out after 30 seconds.
+func getBlockHashDuringStartup(
+	client *rpcclient.Client) (*chainhash.Hash, error) {
+
+	hash, err := client.GetBlockHash(0)
+
+	// Exit early if there's no error.
+	if err == nil {
+		return hash, nil
+	}
+
+	// If the error doesn't start with "-28", it's an unexpected error so
+	// we exit with it.
+	if !strings.Contains(err.Error(), errStillLoadingCode) {
+		return nil, err
+	}
+
+	// Starts the timeout ticker(30s).
+	timeout := time.After(bitcoindStartTimeout)
+
+	// Otherwise, we'd retry calling getblockhash or time out after 30s.
+	for {
+		select {
+		case <-timeout:
+			return nil, ErrBitcoindStartTimeout
+
+		// Retry every second.
+		case <-time.After(1 * time.Second):
+			hash, err = client.GetBlockHash(0)
+			// If there's no error, we return the hash.
+			if err == nil {
+				return hash, nil
+			}
+
+			// Otherwise, retry until we time out. We also check if
+			// the error returned here is still expected.
+			if !strings.Contains(err.Error(), errStillLoadingCode) {
+				return nil, err
+			}
+		}
+	}
+}
+
 // getCurrentNet returns the network on which the bitcoind node is running.
 func getCurrentNet(client *rpcclient.Client) (wire.BitcoinNet, error) {
-	hash, err := client.GetBlockHash(0)
+	hash, err := getBlockHashDuringStartup(client)
 	if err != nil {
 		return 0, err
 	}

chain/btcd.go

@@ -365,7 +365,7 @@ func (c *RPCClient) onRedeemingTx(tx *btcutil.Tx, block *btcjson.BlockDetails) {
 
 func (c *RPCClient) onRescanProgress(hash *chainhash.Hash, height int32, blkTime time.Time) {
 	select {
-	case c.enqueueNotification <- &RescanProgress{hash, height, blkTime}:
+	case c.enqueueNotification <- &RescanProgress{*hash, height, blkTime}:
 	case <-c.quit:
 	}
 }

chain/interface.go

@@ -111,7 +111,7 @@ type (
 	// RescanProgress is a notification describing the current status
 	// of an in-progress rescan.
 	RescanProgress struct {
-		Hash   *chainhash.Hash
+		Hash   chainhash.Hash
 		Height int32
 		Time   time.Time
 	}

chain/neutrino.go

@@ -609,7 +609,7 @@ func (s *NeutrinoClient) onBlockConnected(hash *chainhash.Hash, height int32,
 	sendRescanProgress := func() {
 		select {
 		case s.enqueueNotification <- &RescanProgress{
-			Hash:   hash,
+			Hash:   *hash,
 			Height: height,
 			Time:   time,
 		}:

wallet/rescan.go

@@ -17,7 +17,7 @@ import (
 // set of wallet addresses.
 type RescanProgressMsg struct {
 	Addresses    []btcutil.Address
-	Notification *chain.RescanProgress
+	Notification chain.RescanProgress
 }
 
 // RescanFinishedMsg reports the addresses that were rescanned when a
@@ -147,7 +147,7 @@ func (w *Wallet) rescanBatchHandler() {
 				select {
 				case w.rescanProgress <- &RescanProgressMsg{
 					Addresses:    curBatch.addrs,
-					Notification: n,
+					Notification: *n,
 				}:
 				case <-quit:
 					for _, errChan := range curBatch.errChans {