Merge branch 'syncfix'

Author: benma

backend/coins/btc/account.go

@@ -448,6 +448,10 @@ func (account *Account) Info() *accounts.Info {
 }
 
 func (account *Account) onNewHeader(header *blockchain.Header) error {
+	if account.isClosed() {
+		account.log.Debug("Ignoring new header after the account was closed")
+		return nil
+	}
 	account.log.WithField("block-height", header.BlockHeight).Debug("Received new header")
 	// Fee estimates change with each block.
 	account.updateFeeTargets()

backend/coins/btc/transactions/transactions.go

@@ -54,10 +54,9 @@ func getScriptHashHex(txOut *wire.TxOut) blockchain.ScriptHashHex {
 type Transactions struct {
 	locker.Locker
 
-	net          *chaincfg.Params
-	db           DBInterface
-	headers      headers.Interface
-	requestedTXs map[chainhash.Hash][]func(DBTxInterface, *wire.MsgTx)
+	net     *chaincfg.Params
+	db      DBInterface
+	headers headers.Interface
 
 	// headersTipHeight is the current chain tip height, so we can compute the number of
 	// confirmations of a transaction.
@@ -85,10 +84,9 @@ func NewTransactions(
 	log *logrus.Entry,
 ) *Transactions {
 	transactions := &Transactions{
-		net:          net,
-		db:           db,
-		headers:      headers,
-		requestedTXs: map[chainhash.Hash][]func(DBTxInterface, *wire.MsgTx){},
+		net:     net,
+		db:      db,
+		headers: headers,
 
 		headersTipHeight: headers.TipHeight(),
 
@@ -117,29 +115,8 @@ func (transactions *Transactions) isClosed() bool {
 	return transactions.closed
 }
 
-func (transactions *Transactions) txInHistory(
-	dbTx DBTxInterface, scriptHashHex blockchain.ScriptHashHex, txHash chainhash.Hash) bool {
-	history, err := dbTx.AddressHistory(scriptHashHex)
-	if err != nil {
-		// TODO
-		panic(err)
-	}
-	for _, entry := range history {
-		if txHash == entry.TXHash.Hash() {
-			return true
-		}
-	}
-	return false
-}
-
 func (transactions *Transactions) processTxForAddress(
 	dbTx DBTxInterface, scriptHashHex blockchain.ScriptHashHex, txHash chainhash.Hash, tx *wire.MsgTx, height int) {
-	// Don't process the tx if it is not found in the address history. It could have been removed
-	// from the history before this function was called.
-	if !transactions.txInHistory(dbTx, scriptHashHex, txHash) {
-		return
-	}
-
 	_, _, previousHeight, _, err := dbTx.TxInfo(txHash)
 	if err != nil {
 		transactions.log.WithError(err).Panic("Failed to retrieve tx info")
@@ -341,70 +318,47 @@ func (transactions *Transactions) UpdateAddressHistory(scriptHashHex blockchain.
 	if err := dbTx.PutAddressHistory(scriptHashHex, txs); err != nil {
 		transactions.log.WithError(err).Panic("Failed to store address history")
 	}
-
 	for _, txInfo := range txs {
-		func(txHash chainhash.Hash, height int) {
-			transactions.doForTransaction(dbTx, txHash, func(innerDBTx DBTxInterface, tx *wire.MsgTx) {
-				transactions.processTxForAddress(innerDBTx, scriptHashHex, txHash, tx, height)
-			})
-		}(txInfo.TXHash.Hash(), txInfo.Height)
+		txHash := txInfo.TXHash.Hash()
+		height := txInfo.Height
+		tx := transactions.getTransactionCached(dbTx, txHash)
+		transactions.processTxForAddress(dbTx, scriptHashHex, txHash, tx, height)
 	}
 	if err := dbTx.Commit(); err != nil {
 		transactions.log.WithError(err).Panic("Failed to commit transaction")
 	}
 }
 
 // requires transactions lock
-func (transactions *Transactions) doForTransaction(
+func (transactions *Transactions) getTransactionCached(
 	dbTx DBTxInterface,
 	txHash chainhash.Hash,
-	callback func(DBTxInterface, *wire.MsgTx),
-) {
+) *wire.MsgTx {
 	tx, _, _, _, err := dbTx.TxInfo(txHash)
 	if err != nil {
 		transactions.log.WithError(err).Panic("Failed to retrieve transaction info")
 	}
 	if tx != nil {
-		callback(dbTx, tx)
-		return
-	}
-	if transactions.requestedTXs[txHash] == nil {
-		transactions.requestedTXs[txHash] = []func(DBTxInterface, *wire.MsgTx){}
-	}
-	alreadyDownloading := len(transactions.requestedTXs[txHash]) != 0
-	transactions.requestedTXs[txHash] = append(transactions.requestedTXs[txHash], callback)
-	if alreadyDownloading {
-		return
+		return tx
 	}
-	done := transactions.synchronizer.IncRequestsCounter()
+	txChan := make(chan *wire.MsgTx)
 	transactions.blockchain.TransactionGet(
 		txHash,
 		func(tx *wire.MsgTx) error {
 			if transactions.isClosed() {
 				transactions.log.Debug("TransactionGet result ignored after the instance was closed")
 				return nil
 			}
-
-			defer transactions.Lock()()
-			dbTx, err := transactions.db.Begin()
-			if err != nil {
-				transactions.log.WithError(err).Panic("Failed to begin transaction")
-			}
-			defer dbTx.Rollback()
-
-			for _, callback := range transactions.requestedTXs[txHash] {
-				callback(dbTx, tx)
-			}
-			delete(transactions.requestedTXs, txHash)
-			return dbTx.Commit()
+			txChan <- tx
+			return nil
 		},
 		func(err error) {
-			done()
 			if err != nil {
 				panic(err)
 			}
 		},
 	)
+	return <-txChan
 }
 
 // Balance computes the confirmed and unconfirmed balance of the account.

backend/coins/btc/transactions/transactions_test.go

@@ -15,6 +15,7 @@
 package transactions_test
 
 import (
+	"os"
 	"testing"
 
 	"github.com/btcsuite/btcd/chaincfg"
@@ -40,34 +41,23 @@ import (
 	"github.com/stretchr/testify/suite"
 )
 
+func TestMain(m *testing.M) {
+	test.TstSetupLogging()
+	os.Exit(m.Run())
+}
+
 type BlockchainMock struct {
 	blockchainMock.Interface
-	transactions            map[chainhash.Hash]*wire.MsgTx
-	transactionGetCallbacks map[chainhash.Hash][]func()
+	transactions map[chainhash.Hash]*wire.MsgTx
 }
 
 func NewBlockchainMock() *BlockchainMock {
 	blockchainMock := &BlockchainMock{
-		transactions:            map[chainhash.Hash]*wire.MsgTx{},
-		transactionGetCallbacks: map[chainhash.Hash][]func(){},
+		transactions: map[chainhash.Hash]*wire.MsgTx{},
 	}
 	return blockchainMock
 }
 
-func (blockchain *BlockchainMock) CallTransactionGetCallbacks(txHash chainhash.Hash) {
-	callbacks := blockchain.transactionGetCallbacks[txHash]
-	delete(blockchain.transactionGetCallbacks, txHash)
-	for _, callback := range callbacks {
-		callback()
-	}
-}
-
-func (blockchain *BlockchainMock) CallAllTransactionGetCallbacks() {
-	for txHash := range blockchain.transactionGetCallbacks {
-		blockchain.CallTransactionGetCallbacks(txHash)
-	}
-}
-
 func (blockchain *BlockchainMock) RegisterTxs(txs ...*wire.MsgTx) {
 	for _, tx := range txs {
 		blockchain.transactions[tx.TxHash()] = tx
@@ -84,17 +74,7 @@ func (blockchain *BlockchainMock) TransactionGet(
 	if !ok {
 		panic("you need to first register the transaction with the mock backend")
 	}
-	callbacks, ok := blockchain.transactionGetCallbacks[txHash]
-	if !ok {
-		callbacks = []func(){}
-	}
-	blockchain.transactionGetCallbacks[txHash] = append(callbacks,
-		func() {
-			defer cleanup(nil)
-			if err := success(tx); err != nil {
-				panic(err)
-			}
-		})
+	go func() { _ = success(tx) }()
 }
 
 func (blockchain *BlockchainMock) ConnectionStatus() blockchainpkg.Status {
@@ -152,7 +132,6 @@ func (s *transactionsSuite) updateAddressHistory(
 	}
 
 	s.transactions.UpdateAddressHistory(address.PubkeyScriptHashHex(), txs)
-	s.blockchainMock.CallAllTransactionGetCallbacks()
 }
 
 func newTx(
@@ -170,50 +149,6 @@ func newTx(
 	}
 }
 
-// TestUpdateAddressHistorySyncStatus checks that the synchronizer is calling the
-// syncStart/syncFinished callbacks once in the beginning and once after all transactions have
-// processed.
-func (s *transactionsSuite) TestUpdateAddressHistorySyncStatus() {
-	addresses := s.addressChain.EnsureAddresses()
-	address := addresses[0]
-	expectedAmount := btcutil.Amount(123)
-	tx1 := newTx(chainhash.HashH(nil), 0, address, expectedAmount)
-	tx1Hash := tx1.TxHash()
-	tx2 := newTx(chainhash.HashH(nil), 1, address, expectedAmount)
-	tx2Hash := tx2.TxHash()
-	s.blockchainMock.RegisterTxs(tx1, tx2)
-	var syncStarted, syncFinished bool
-	onSyncStarted := func() {
-		if syncStarted {
-			require.FailNow(s.T(), "sync started twice")
-		}
-		syncStarted = true
-	}
-	onSyncFinished := func() {
-		if syncFinished {
-			require.FailNow(s.T(), "sync finished twice")
-		}
-		syncFinished = true
-	}
-	*s.synchronizer = *synchronizer.NewSynchronizer(onSyncStarted, onSyncFinished, s.log)
-	s.notifierMock.On("Put", tx1Hash[:]).Return(nil).Once()
-	s.notifierMock.On("Put", tx2Hash[:]).Return(nil).Once()
-	s.transactions.UpdateAddressHistory(address.PubkeyScriptHashHex(), []*blockchainpkg.TxInfo{
-		{TXHash: blockchainpkg.TXHash(tx1Hash), Height: 10},
-		{TXHash: blockchainpkg.TXHash(tx2Hash), Height: 10},
-	})
-	require.True(s.T(), syncStarted)
-	require.False(s.T(), syncFinished)
-	s.headersMock.On("HeaderByHeight", 10).Return(nil, nil).Once()
-	s.blockchainMock.CallTransactionGetCallbacks(tx1.TxHash())
-	require.True(s.T(), syncStarted)
-	require.False(s.T(), syncFinished)
-	s.headersMock.On("HeaderByHeight", 10).Return(nil, nil).Once()
-	s.blockchainMock.CallTransactionGetCallbacks(tx2.TxHash())
-	require.True(s.T(), syncStarted)
-	require.True(s.T(), syncFinished)
-}
-
 func newBalance(available, incoming btcutil.Amount) *accounts.Balance {
 	return accounts.NewBalance(
 		coin.NewAmountFromInt64(int64(available)),
@@ -253,35 +188,6 @@ func (s *transactionsSuite) TestUpdateAddressHistorySingleTxReceive() {
 	require.Equal(s.T(), expectedHeight, transactions[0].Height)
 }
 
-// TestUpdateAddressHistoryOppositeOrder checks that a spend is correctly recognized even if the
-// transactions in the history of an address are processed in the wrong order. If the spending tx is
-// processed before the funding tx, the output is unknown when processing the funds, but after the
-// output has been added, the input spending it needs to be indexed correctly.
-func (s *transactionsSuite) TestUpdateAddressHistoryOppositeOrder() {
-	addresses := s.addressChain.EnsureAddresses()
-	address := addresses[0]
-	address2 := addresses[1]
-	tx1 := newTx(chainhash.HashH(nil), 0, address, 123)
-	tx1Hash := tx1.TxHash()
-	tx2 := newTx(tx1.TxHash(), 0, address2, 123)
-	tx2Hash := tx2.TxHash()
-	s.blockchainMock.RegisterTxs(tx1, tx2)
-	s.notifierMock.On("Put", tx1Hash[:]).Return(nil).Once()
-	s.notifierMock.On("Put", tx2Hash[:]).Return(nil).Once()
-	s.transactions.UpdateAddressHistory(address.PubkeyScriptHashHex(), []*blockchainpkg.TxInfo{
-		{TXHash: blockchainpkg.TXHash(tx1Hash), Height: 0},
-		{TXHash: blockchainpkg.TXHash(tx2Hash), Height: 0},
-	})
-	// Process tx2 (the spend) before tx1 (the funding). This should result in a zero balance, as
-	// the received funds are spent.
-	s.blockchainMock.CallTransactionGetCallbacks(tx2.TxHash())
-	s.blockchainMock.CallTransactionGetCallbacks(tx1.TxHash())
-	require.Equal(s.T(),
-		newBalance(0, 0),
-		s.transactions.Balance(),
-	)
-}
-
 // TestSpendableOutputs checks that the utxo set is correct. Only confirmed (or unconfirmed outputs
 // we own) outputs can be spent.
 func (s *transactionsSuite) TestSpendableOutputs() {
@@ -453,23 +359,3 @@ func (s *transactionsSuite) TestRemoveTransaction() {
 		s.transactions.Transactions(func(blockchainpkg.ScriptHashHex) bool { return false }),
 		2)
 }
-
-// TestRemoveTransactionPendingDownload tests that a tx can be removed from the address history
-// while it is still pending to be indexed.
-func (s *transactionsSuite) TestRemoveTransactionPendingDownload() {
-	address := s.addressChain.EnsureAddresses()[0]
-	tx := newTx(chainhash.HashH(nil), 0, address, 123)
-	s.blockchainMock.RegisterTxs(tx)
-	s.transactions.UpdateAddressHistory(address.PubkeyScriptHashHex(), []*blockchainpkg.TxInfo{
-		{TXHash: blockchainpkg.TXHash(tx.TxHash()), Height: 0},
-	})
-	// Callback for processing the tx is not called yet. We remove the tx.
-	s.transactions.UpdateAddressHistory(address.PubkeyScriptHashHex(), []*blockchainpkg.TxInfo{})
-	// Process the tx now. It should not be indexed anymore.
-	s.blockchainMock.CallAllTransactionGetCallbacks()
-	require.Equal(s.T(),
-		newBalance(0, 0),
-		s.transactions.Balance())
-	require.Empty(s.T(),
-		s.transactions.Transactions(func(blockchainpkg.ScriptHashHex) bool { return false }))
-}