Merge pull request #14 from coinbase/patrick/head-block-error

[Bug] Inactive Reconciler Fail Before Blocks Synced

cmd/check_complete.go

@@ -16,6 +16,7 @@ package cmd
 
 import (
 	"context"
+	"fmt"
 	"log"
 
 	"github.com/coinbase/rosetta-cli/internal/logger"
@@ -24,7 +25,6 @@ import (
 	"github.com/coinbase/rosetta-cli/internal/syncer"
 
 	"github.com/coinbase/rosetta-sdk-go/fetcher"
-
 	"github.com/spf13/cobra"
 	"golang.org/x/sync/errgroup"
 )
@@ -79,7 +79,7 @@ func runCheckCompleteCmd(cmd *cobra.Command, args []string) {
 
 	exemptAccounts, err := loadAccounts(ExemptFile)
 	if err != nil {
-		log.Fatal(err)
+		log.Fatal(fmt.Errorf("%w: unable to load exempt accounts", err))
 	}
 
 	fetcher := fetcher.New(
@@ -92,12 +92,12 @@ func runCheckCompleteCmd(cmd *cobra.Command, args []string) {
 	// TODO: sync and reconcile on subnetworks, if they exist.
 	primaryNetwork, networkStatus, err := fetcher.InitializeAsserter(ctx)
 	if err != nil {
-		log.Fatal(err)
+		log.Fatal(fmt.Errorf("%w: unable to initialize asserter", err))
 	}
 
 	localStore, err := storage.NewBadgerStorage(ctx, DataDir)
 	if err != nil {
-		log.Fatal(err)
+		log.Fatal(fmt.Errorf("%w: unable to initialize data store", err))
 	}
 
 	blockStorage := storage.NewBlockStorage(ctx, localStore)
@@ -108,7 +108,7 @@ func runCheckCompleteCmd(cmd *cobra.Command, args []string) {
 			networkStatus.GenesisBlockIdentifier,
 		)
 		if err != nil {
-			log.Fatal(err)
+			log.Fatal(fmt.Errorf("%w: unable to bootstrap balances", err))
 		}
 	}
 

go.mod

@@ -6,6 +6,7 @@ require (
 	github.com/coinbase/rosetta-sdk-go v0.1.6
 	github.com/davecgh/go-spew v1.1.1
 	github.com/dgraph-io/badger v1.6.0
+	github.com/pkg/errors v0.8.1
 	github.com/spf13/cobra v0.0.5
 	github.com/stretchr/testify v1.5.1
 	golang.org/x/sync v0.0.0-20200317015054-43a5402ce75a

internal/reconciler/stateful_reconciler.go

@@ -27,7 +27,6 @@ import (
 
 	"github.com/coinbase/rosetta-sdk-go/fetcher"
 	"github.com/coinbase/rosetta-sdk-go/types"
-
 	"golang.org/x/sync/errgroup"
 )
 
@@ -187,7 +186,7 @@ func (r *StatefulReconciler) CompareBalance(
 	// Head block should be set before we CompareBalance
 	head, err := r.storage.GetHeadBlockIdentifier(ctx, txn)
 	if err != nil {
-		return zeroString, 0, err
+		return zeroString, 0, fmt.Errorf("%w: unable to get current block for reconciliation", err)
 	}
 
 	// Check if live block is < head (or wait)
@@ -467,10 +466,16 @@ func (r *StatefulReconciler) reconcileInactiveAccounts(
 	for ctx.Err() == nil {
 		txn := r.storage.NewDatabaseTransaction(ctx, false)
 		head, err := r.storage.GetHeadBlockIdentifier(ctx, txn)
-		if err != nil {
-			return err
-		}
 		txn.Discard(ctx)
+		// When first start syncing, this loop may run before the genesis block is synced.
+		// If this is the case, we should sleep and try again later instead of exiting.
+		if errors.Is(err, storage.ErrHeadBlockNotFound) {
+			log.Println("head block not yet initialized, sleeping...")
+			time.Sleep(inactiveReconciliationSleep)
+			continue
+		} else if err != nil {
+			return fmt.Errorf("%w: unable to get current block for inactive reconciliation", err)
+		}
 
 		r.inactiveQueueMutex.Lock()
 		if len(r.inactiveQueue) > 0 &&

internal/reconciler/stateful_reconciler_test.go

@@ -16,6 +16,7 @@ package reconciler
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"testing"
 
@@ -99,7 +100,7 @@ func TestCompareBalance(t *testing.T) {
 		)
 		assert.Equal(t, "0", difference)
 		assert.Equal(t, int64(0), headIndex)
-		assert.EqualError(t, err, storage.ErrHeadBlockNotFound.Error())
+		assert.True(t, errors.Is(err, storage.ErrHeadBlockNotFound))
 	})
 
 	// Update head block

internal/reconciler/stateless_reconciler.go

@@ -79,7 +79,8 @@ func (r *StatelessReconciler) QueueChanges(
 		skipAccount := false
 		// Look through balance changes for account + currency
 		for _, change := range balanceChanges {
-			if reflect.DeepEqual(change.Account, account.Account) && reflect.DeepEqual(change.Currency, account.Currency) {
+			if reflect.DeepEqual(change.Account, account.Account) &&
+				reflect.DeepEqual(change.Currency, account.Currency) {
 				skipAccount = true
 				break
 			}

internal/syncer/syncer.go

@@ -70,7 +70,7 @@ func NextSyncableRange(
 ) (int64, int64, bool, error) {
 	currentIndex, err := s.CurrentIndex(ctx)
 	if err != nil {
-		return -1, -1, false, err
+		return -1, -1, false, fmt.Errorf("%w: unable to get current index", err)
 	}
 
 	if endIndex == -1 {
@@ -80,7 +80,7 @@ func NextSyncableRange(
 			nil,
 		)
 		if err != nil {
-			return -1, -1, false, err
+			return -1, -1, false, fmt.Errorf("%w: unable to get network status", err)
 		}
 
 		return currentIndex, networkStatus.CurrentBlockIdentifier.Index, false, nil
@@ -105,7 +105,7 @@ func Sync(
 	defer cancel()
 
 	if err := s.SetStartIndex(ctx, startIndex); err != nil {
-		return err
+		return fmt.Errorf("%w: unable to set start index", err)
 	}
 
 	for {
@@ -115,7 +115,7 @@ func Sync(
 			endIndex,
 		)
 		if err != nil {
-			return err
+			return fmt.Errorf("%w: unable to get next syncable range", err)
 		}
 		if halt {
 			break
@@ -129,7 +129,7 @@ func Sync(
 
 		err = s.SyncRange(ctx, rangeStart, rangeEnd)
 		if err != nil {
-			return err
+			return fmt.Errorf("%w: unable to sync range %d-%d", err, rangeStart, rangeEnd)
 		}
 
 		if ctx.Err() != nil {