Refactoring/blockchain sync#246 (#271)

PR adds a new API changes to the NodeManager to provide simple methods to validate the completed synchonization of the blockchain.

cmd/statusd/utils.go

@@ -54,10 +54,6 @@ func testExportedAPI(t *testing.T, done chan struct{}) {
 			"check default configuration",
 			testGetDefaultConfig,
 		},
-		{
-			"reset blockchain data",
-			testResetChainData,
-		},
 		{
 			"stop/resume node",
 			testStopResumeNode,
@@ -217,6 +213,7 @@ func testResetChainData(t *testing.T) bool {
 		return false
 	}
 
+	// FIXME(tiabc): EnsureNodeSync the same way as in e2e tests.
 	time.Sleep(TestConfig.Node.SyncSeconds * time.Second) // allow to re-sync blockchain
 
 	testCompleteTransaction(t)
@@ -1364,6 +1361,7 @@ func startTestNode(t *testing.T) <-chan struct{} {
 			// sync
 			if syncRequired {
 				t.Logf("Sync is required, it will take %d seconds", TestConfig.Node.SyncSeconds)
+				// FIXME(tiabc): EnsureNodeSync the same way as in e2e tests.
 				time.Sleep(TestConfig.Node.SyncSeconds * time.Second) // LES syncs headers, so that we are up do date when it is done
 			} else {
 				time.Sleep(5 * time.Second)

e2e/api/backend_test.go

@@ -237,7 +237,8 @@ func (s *APIBackendTestSuite) TestResetChainData() {
 	s.StartTestBackend(params.RinkebyNetworkID)
 	defer s.StopTestBackend()
 
-	time.Sleep(2 * time.Second) // allow to sync for some time
+	// allow to sync for some time
+	s.EnsureNodeSync()
 
 	s.True(s.Backend.IsNodeRunning())
 	nodeReady, err := s.Backend.ResetChainData()

e2e/jail/jail_rpc_test.go

@@ -111,7 +111,7 @@ func (s *JailRPCTestSuite) TestContractDeployment() {
 	defer s.StopTestBackend()
 
 	// Allow to sync, otherwise you'll get "Nonce too low."
-	time.Sleep(TestConfig.Node.SyncSeconds * time.Second)
+	s.EnsureNodeSync()
 
 	// obtain VM for a given chat (to send custom JS to jailed version of Send())
 	s.jail.Parse(testChatID, "")
@@ -195,7 +195,7 @@ func (s *JailRPCTestSuite) TestJailVMPersistence() {
 	s.StartTestBackend(params.RopstenNetworkID)
 	defer s.StopTestBackend()
 
-	time.Sleep(TestConfig.Node.SyncSeconds * time.Second) // allow to sync
+	s.EnsureNodeSync()
 
 	// log into account from which transactions will be sent
 	err := s.Backend.AccountManager().SelectAccount(TestConfig.Account1.Address, TestConfig.Account1.Password)

e2e/node/manager_test.go

@@ -296,7 +296,7 @@ func (s *ManagerTestSuite) TestResetChainData() {
 	defer s.StopTestNode()
 
 	// allow to sync for some time
-	time.Sleep(10 * time.Second)
+	s.EnsureNodeSync()
 
 	// reset chain data
 	nodeReady, err := s.NodeManager.ResetChainData()

e2e/suites.go

@@ -1,10 +1,14 @@
 package e2e
 
 import (
+	"context"
+	"time"
+
 	"github.com/ethereum/go-ethereum/les"
 	whisper "github.com/ethereum/go-ethereum/whisper/whisperv5"
 	"github.com/status-im/status-go/geth/api"
 	"github.com/status-im/status-go/geth/common"
+	"github.com/status-im/status-go/geth/node"
 	"github.com/status-im/status-go/geth/signal"
 	"github.com/stretchr/testify/suite"
 )
@@ -12,7 +16,37 @@ import (
 // NodeManagerTestSuite defines a test suit with NodeManager.
 type NodeManagerTestSuite struct {
 	suite.Suite
-	NodeManager common.NodeManager
+	NodeManager       common.NodeManager
+	nodeSyncCompleted bool
+}
+
+// EnsureNodeSync ensures that synchronization of the node is done once and that it
+// is done properly else, the call will fail.
+// FIXME(tiabc): BackendTestSuite contains the same method, let's sort it out?
+func (s *NodeManagerTestSuite) EnsureNodeSync(forceResync ...bool) {
+	if len(forceResync) > 0 && forceResync[0] {
+		s.nodeSyncCompleted = false
+	}
+
+	if s.nodeSyncCompleted {
+		return
+	}
+
+	require := s.Require()
+
+	ethClient, err := s.NodeManager.LightEthereumService()
+	require.NoError(err)
+	require.NotNil(ethClient)
+
+	sync := node.NewSyncPoll(ethClient)
+	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Minute)
+	defer cancel()
+
+	// Validate that synchronization failed because of time.
+	syncError := sync.Poll(ctx)
+	require.NoError(syncError)
+
+	s.nodeSyncCompleted = true
 }
 
 // StartTestNode initiazes a NodeManager instances with configuration retrieved
@@ -51,7 +85,8 @@ func (s *NodeManagerTestSuite) StopTestNode() {
 // and a few utility methods to start and stop node or get various services.
 type BackendTestSuite struct {
 	suite.Suite
-	Backend *api.StatusBackend
+	Backend           *api.StatusBackend
+	nodeSyncCompleted bool
 }
 
 // SetupTest initializes Backend.
@@ -127,6 +162,32 @@ func (s *BackendTestSuite) TxQueueManager() common.TxQueueManager {
 	return s.Backend.TxQueueManager()
 }
 
+// EnsureNodeSync ensures that synchronization of the node is done once and that it
+// is done properly else, the call will fail.
+// FIXME(tiabc): NodeManagerTestSuite contains the same method, let's sort it out?
+func (s *BackendTestSuite) EnsureNodeSync(forceResync ...bool) {
+	if len(forceResync) > 0 && forceResync[0] {
+		s.nodeSyncCompleted = false
+	}
+
+	if s.nodeSyncCompleted {
+		return
+	}
+
+	require := s.Require()
+
+	ethClient := s.LightEthereumService()
+	sync := node.NewSyncPoll(ethClient)
+	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Minute)
+	defer cancel()
+
+	// Validate that synchronization failed because of time.
+	syncError := sync.Poll(ctx)
+	require.NoError(syncError)
+
+	s.nodeSyncCompleted = true
+}
+
 func importTestAccouns(keyStoreDir string) (err error) {
 	err = common.ImportTestAccount(keyStoreDir, "test-account1.pk")
 	if err != nil {

e2e/transactions/transactions_test.go

@@ -39,8 +39,8 @@ func (s *TransactionsTestSuite) TestCallRPCSendTransaction() {
 	s.StartTestBackend(params.RopstenNetworkID)
 	defer s.StopTestBackend()
 
-	// Allow to sync the blockchain.
+	// allow to sync for some time
-	time.Sleep(TestConfig.Node.SyncSeconds * time.Second)
+	s.EnsureNodeSync()
 
 	err := s.Backend.AccountManager().SelectAccount(TestConfig.Account1.Address, TestConfig.Account1.Password)
 	s.NoError(err)
@@ -92,7 +92,7 @@ func (s *TransactionsTestSuite) TestCallRPCSendTransactionUpstream() {
 	defer s.StopTestBackend()
 
 	// Allow to sync the blockchain.
-	time.Sleep(TestConfig.Node.SyncSeconds * time.Second)
+	s.EnsureNodeSync()
 
 	err := s.Backend.AccountManager().SelectAccount(TestConfig.Account2.Address, TestConfig.Account2.Password)
 	s.NoError(err)
@@ -147,7 +147,7 @@ func (s *TransactionsTestSuite) TestSendContractTx() {
 	s.StartTestBackend(params.RopstenNetworkID)
 	defer s.StopTestBackend()
 
-	time.Sleep(TestConfig.Node.SyncSeconds * time.Second) // allow to sync
+	s.EnsureNodeSync()
 
 	sampleAddress, _, _, err := s.Backend.AccountManager().CreateAccount(TestConfig.Account1.Password)
 	s.NoError(err)
@@ -234,7 +234,7 @@ func (s *TransactionsTestSuite) TestSendEtherTx() {
 	s.StartTestBackend(params.RopstenNetworkID)
 	defer s.StopTestBackend()
 
-	time.Sleep(TestConfig.Node.SyncSeconds * time.Second) // allow to sync
+	s.EnsureNodeSync()
 
 	backend := s.LightEthereumService().StatusBackend
 	s.NotNil(backend)
@@ -402,7 +402,7 @@ func (s *TransactionsTestSuite) TestSendEtherTxUpstream() {
 	)
 	defer s.StopTestBackend()
 
-	time.Sleep(TestConfig.Node.SyncSeconds * time.Second) // allow to sync
+	s.EnsureNodeSync()
 
 	err := s.Backend.AccountManager().SelectAccount(TestConfig.Account1.Address, TestConfig.Account1.Password)
 	s.NoError(err)
@@ -455,7 +455,7 @@ func (s *TransactionsTestSuite) TestDoubleCompleteQueuedTransactions() {
 	s.StartTestBackend(params.RopstenNetworkID)
 	defer s.StopTestBackend()
 
-	time.Sleep(TestConfig.Node.SyncSeconds * time.Second) // allow to sync
+	s.EnsureNodeSync()
 
 	backend := s.LightEthereumService().StatusBackend
 	s.NotNil(backend)
@@ -532,7 +532,7 @@ func (s *TransactionsTestSuite) TestDiscardQueuedTransaction() {
 	s.StartTestBackend(params.RopstenNetworkID)
 	defer s.StopTestBackend()
 
-	time.Sleep(TestConfig.Node.SyncSeconds * time.Second) // allow to sync
+	s.EnsureNodeSync()
 
 	backend := s.LightEthereumService().StatusBackend
 	s.NotNil(backend)
@@ -612,8 +612,7 @@ func (s *TransactionsTestSuite) TestCompleteMultipleQueuedTransactions() {
 	s.StartTestBackend(params.RopstenNetworkID)
 	defer s.StopTestBackend()
 
-	// allow to sync
+	s.EnsureNodeSync()
-	time.Sleep(TestConfig.Node.SyncSeconds * time.Second)
 
 	s.TxQueueManager().TransactionQueue().Reset()
 
@@ -707,7 +706,7 @@ func (s *TransactionsTestSuite) TestDiscardMultipleQueuedTransactions() {
 	s.StartTestBackend(params.RopstenNetworkID)
 	defer s.StopTestBackend()
 
-	time.Sleep(TestConfig.Node.SyncSeconds * time.Second) // allow to sync
+	s.EnsureNodeSync()
 
 	backend := s.LightEthereumService().StatusBackend
 	s.NotNil(backend)

geth/node/node.go

@@ -170,6 +170,7 @@ func activateEthService(stack *node.Node, config *params.NodeConfig) error {
 		if err == nil {
 			updateCHT(lightEth, config)
 		}
+
 		return lightEth, err
 	}); err != nil {
 		return fmt.Errorf("%v: %v", ErrLightEthRegistrationFailure, err)

geth/node/syncpoll.go

@@ -0,0 +1,81 @@
+package node
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"github.com/ethereum/go-ethereum/eth/downloader"
+	"github.com/ethereum/go-ethereum/les"
+	"github.com/status-im/status-go/geth/log"
+)
+
+// errors
+var (
+	ErrStartAborted = errors.New("node synchronization timeout before starting")
+	ErrSyncAborted  = errors.New("node synchronization timeout before completion")
+)
+
+// SyncPoll provides a structure that allows us to check the status of
+// ethereum node synchronization.
+type SyncPoll struct {
+	downloader *downloader.Downloader
+}
+
+// NewSyncPoll returns a new instance of SyncPoll.
+func NewSyncPoll(leth *les.LightEthereum) *SyncPoll {
+	return &SyncPoll{
+		downloader: leth.Downloader(),
+	}
+}
+
+// Poll checks for the status of blockchain synchronization and returns an error
+// if the blockchain failed to start synchronizing or fails to complete, within
+// the time provided by the passed in context.
+func (n *SyncPoll) Poll(ctx context.Context) error {
+	if err := n.pollSyncStart(ctx); err != nil {
+		return err
+	}
+
+	if err := n.waitSyncCompleted(ctx); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (n *SyncPoll) pollSyncStart(ctx context.Context) error {
+	for {
+		select {
+		case <-ctx.Done():
+			return ErrStartAborted
+		case <-time.After(100 * time.Millisecond):
+			if n.downloader.Synchronising() {
+				log.Info("Block synchronization progress just started")
+				return nil
+			}
+		}
+	}
+}
+
+func (n *SyncPoll) waitSyncCompleted(ctx context.Context) error {
+	for {
+		select {
+		case <-ctx.Done():
+			return ErrSyncAborted
+		case <-time.After(100 * time.Millisecond):
+			progress := n.downloader.Progress()
+
+			// If information on highest block has not being retrieved then wait.
+			if progress.HighestBlock <= 0 && progress.StartingBlock <= 0 {
+				time.Sleep(300 * time.Millisecond)
+				continue
+			}
+
+			if progress.CurrentBlock >= progress.HighestBlock {
+				log.Info("Block synchronization just finished")
+				return nil
+			}
+		}
+	}
+}