Repository: coinbase/rosetta-cli
Pull request #202 (2020-12-09 15:41:21)

[bug] Bootstrap Balances Panic
Related: https://github.com/coinbase/rosetta-sdk-go/pull/273

Invoking BootstrapBalances in `check:data` always panics because we now invoke the BalanceStorage handler during bootstrapping (which isn't yet initialized). This PR moves BootstrapBalances to follow initialization.

**WARNING: THIS PR INCLUDES BREAKING CHANGES TO HOW DATA IS STORED ON DISK.**

### Changes
- [x] Move BootstrapBalances
- [x] Handle ErrHelperHandlerMissing 
- [x] Update `rosetta-sdk-go@v0.6.5`
- [x] Add configs for `SerialBlockWorkers` and `SeenBlockWorkers`
- [x] Update version

#### `rosetta-sdk-go` Changes
- [storage] Protect against nil Helper and/or Handler in BalanceStorage [`#273`](https://github.com/coinbase/rosetta-sdk-go/pull/273)
- [storage] Single Errgroup Pool for BlockWorkers [`#272`](https://github.com/coinbase/rosetta-sdk-go/pull/272)
- [chore] `rosetta-specifications@v1.4.9` [`#271`](https://github.com/coinbase/rosetta-sdk-go/pull/271)
- [syncer/storage] Pre-store Blocks [`#269`](https://github.com/coinbase/rosetta-sdk-go/pull/269)

### Future Work (before release)
* Implement `ForceInactiveReconciliation`
* Mark at tip if `SyncStatus.Synced` is true


Patch:
diff --git a/cmd/root.go b/cmd/root.go
index c99f9c4..dfe2249 100644
--- a/cmd/root.go
+++ b/cmd/root.go
@@ -269,6 +269,6 @@ var versionCmd = &cobra.Command{
 	Use:   "version",
 	Short: "Print rosetta-cli version",
 	Run: func(cmd *cobra.Command, args []string) {
-		fmt.Println("v0.6.3")
+		fmt.Println("v0.6.4")
 	},
 }
diff --git a/configuration/configuration.go b/configuration/configuration.go
index 1f6df0d..969aee7 100644
--- a/configuration/configuration.go
+++ b/configuration/configuration.go
@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"log"
 	"path"
+	"runtime"
 
 	"github.com/coinbase/rosetta-sdk-go/asserter"
 	"github.com/coinbase/rosetta-sdk-go/constructor/dsl"
@@ -153,6 +154,15 @@ func populateMissingFields(config *Configuration) *Configuration {
 		config.MaxReorgDepth = DefaultMaxReorgDepth
 	}
 
+	numCPU := runtime.NumCPU()
+	if config.SeenBlockWorkers == 0 {
+		config.SeenBlockWorkers = numCPU
+	}
+
+	if config.SerialBlockWorkers == 0 {
+		config.SerialBlockWorkers = numCPU
+	}
+
 	config.Construction = populateConstructionMissingFields(config.Construction)
 	config.Data = populateDataMissingFields(config.Data)
 
@@ -227,11 +237,15 @@ func assertConstructionConfiguration(ctx context.Context, config *ConstructionCo
 	return nil
 }
 
-func assertDataConfiguration(config *DataConfiguration) error {
+func assertDataConfiguration(config *DataConfiguration) error { // nolint:gocognit
 	if config.StartIndex != nil && *config.StartIndex < 0 {
 		return fmt.Errorf("start index %d cannot be negative", *config.StartIndex)
 	}
 
+	if !config.ReconciliationDisabled && config.BalanceTrackingDisabled {
+		return errors.New("balance tracking must be enabled to perform reconciliation")
+	}
+
 	if config.EndConditions == nil {
 		return nil
 	}
@@ -288,6 +302,14 @@ func assertConfiguration(ctx context.Context, config *Configuration) error {
 		return fmt.Errorf("%w: invalid network identifier", err)
 	}
 
+	if config.SeenBlockWorkers <= 0 {
+		return errors.New("seen_block_workers must be > 0")
+	}
+
+	if config.SerialBlockWorkers <= 0 {
+		return errors.New("serial_block_workers must be > 0")
+	}
+
 	if err := assertDataConfiguration(config.Data); err != nil {
 		return fmt.Errorf("%w: invalid data configuration", err)
 	}
diff --git a/configuration/types.go b/configuration/types.go
index 5a25f67..10c73a9 100644
--- a/configuration/types.go
+++ b/configuration/types.go
@@ -396,6 +396,16 @@ type Configuration struct {
 	// but can use 10s of GBs of RAM, even with pruning enabled.
 	MemoryLimitDisabled bool `json:"memory_limit_disabled"`
 
+	// SeenBlockWorkers is the number of goroutines spawned to store
+	// seen blocks in storage before we attempt to sequence. If not populated,
+	// this value defaults to runtime.NumCPU().
+	SeenBlockWorkers int `json:"seen_block_workers,omitempty"`
+
+	// SerialBlockWorkers is the number of goroutines spawned to help
+	// with block sequencing (i.e. updating balances, updating coins, etc).
+	// If not populated, this value defaults to runtime.NumCPU().
+	SerialBlockWorkers int `json:"serial_block_workers,omitempty"`
+
 	Construction *ConstructionConfiguration `json:"construction"`
 	Data         *DataConfiguration         `json:"data"`
 }
diff --git a/go.mod b/go.mod
index 3b85437..f314b4c 100644
--- a/go.mod
+++ b/go.mod
@@ -3,7 +3,7 @@ module github.com/coinbase/rosetta-cli
 go 1.13
 
 require (
-	github.com/coinbase/rosetta-sdk-go v0.6.4
+	github.com/coinbase/rosetta-sdk-go v0.6.5
 	github.com/fatih/color v1.10.0
 	github.com/olekukonko/tablewriter v0.0.2-0.20190409134802-7e037d187b0c
 	github.com/spf13/cobra v1.1.1
diff --git a/go.sum b/go.sum
index 4fc9123..c21bece 100644
--- a/go.sum
+++ b/go.sum
@@ -80,8 +80,8 @@ github.com/cespare/xxhash/v2 v2.1.1 h1:6MnRN8NT7+YBpUIWxHtefFZOKTAPgGjpQSxqLNn0+
 github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
 github.com/cloudflare/cloudflare-go v0.10.2-0.20190916151808-a80f83b9add9/go.mod h1:1MxXX1Ux4x6mqPmjkUgTP1CdXIBXKX7T+Jk9Gxrmx+U=
-github.com/coinbase/rosetta-sdk-go v0.6.4 h1:+3z6otziRDdgEwy7IsyTkwgCNbbH4z5QMDnXdbdg9ZY=
-github.com/coinbase/rosetta-sdk-go v0.6.4/go.mod h1:MvQfsL2KlJ5786OdDviRIJE3agui2YcvS1CaQPDl1Yo=
+github.com/coinbase/rosetta-sdk-go v0.6.5 h1:RytFDCPXS64vEYwIOsxsoQGlZZyP9RQvzyYikxymI4w=
+github.com/coinbase/rosetta-sdk-go v0.6.5/go.mod h1:MvQfsL2KlJ5786OdDviRIJE3agui2YcvS1CaQPDl1Yo=
 github.com/coreos/bbolt v1.3.2/go.mod h1:iRUV2dpdMOn7Bo10OQBFzIJO9kkE559Wcmn+qkEiiKk=
 github.com/coreos/etcd v3.3.10+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
 github.com/coreos/etcd v3.3.13+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
diff --git a/pkg/results/data_results.go b/pkg/results/data_results.go
index 1afdcdd..7c26bde 100644
--- a/pkg/results/data_results.go
+++ b/pkg/results/data_results.go
@@ -251,12 +251,16 @@ func ComputeCheckDataStats(
 
 	if balances != nil {
 		coverage, err := balances.EstimatedReconciliationCoverage(ctx)
-		if err != nil {
-			log.Printf("%s: cannot get reconcile coverage", err.Error())
+		switch {
+		case err == nil:
+			stats.ReconciliationCoverage = coverage
+		case errors.Is(err, storageErrs.ErrHelperHandlerMissing):
+			// In this case, we use the default 0 value for the reconciliation
+			// coverage in stats.
+		case err != nil:
+			log.Printf("%s: cannot get reconciliation coverage", err.Error())
 			return nil
 		}
-
-		stats.ReconciliationCoverage = coverage
 	}
 
 	return stats


Test patch:
diff --git a/configuration/configuration_test.go b/configuration/configuration_test.go
index 5170340..bdc9125 100644
--- a/configuration/configuration_test.go
+++ b/configuration/configuration_test.go
@@ -18,6 +18,7 @@ import (
 	"context"
 	"os/exec"
 	"path"
+	"runtime"
 	"testing"
 
 	"github.com/coinbase/rosetta-sdk-go/constructor/job"
@@ -68,6 +69,8 @@ var (
 		MaxSyncConcurrency:   12,
 		TipDelay:             1231,
 		MaxReorgDepth:        12,
+		SeenBlockWorkers:     300,
+		SerialBlockWorkers:   200,
 		Construction: &ConstructionConfiguration{
 			OfflineURL:            "https://ashdjaksdkjshdk",
 			MaxOfflineConnections: 21,
@@ -156,7 +159,13 @@ func TestLoadConfiguration(t *testing.T) {
 	}{
 		"nothing provided": {
 			provided: &Configuration{},
-			expected: DefaultConfiguration(),
+			expected: func() *Configuration {
+				cfg := DefaultConfiguration()
+				cfg.SeenBlockWorkers = runtime.NumCPU()
+				cfg.SerialBlockWorkers = runtime.NumCPU()
+
+				return cfg
+			}(),
 		},
 		"no overwrite": {
 			provided: whackyConfig,
@@ -171,6 +180,8 @@ func TestLoadConfiguration(t *testing.T) {
 			},
 			expected: func() *Configuration {
 				cfg := DefaultConfiguration()
+				cfg.SeenBlockWorkers = runtime.NumCPU()
+				cfg.SerialBlockWorkers = runtime.NumCPU()
 				cfg.Construction = &ConstructionConfiguration{
 					OfflineURL:            DefaultURL,
 					MaxOfflineConnections: DefaultMaxOfflineConnections,
@@ -193,6 +204,8 @@ func TestLoadConfiguration(t *testing.T) {
 			},
 			expected: func() *Configuration {
 				cfg := DefaultConfiguration()
+				cfg.SeenBlockWorkers = runtime.NumCPU()
+				cfg.SerialBlockWorkers = runtime.NumCPU()
 				cfg.Construction = &ConstructionConfiguration{
 					OfflineURL:            DefaultURL,
 					MaxOfflineConnections: DefaultMaxOfflineConnections,
@@ -221,6 +234,8 @@ func TestLoadConfiguration(t *testing.T) {
 			},
 			expected: func() *Configuration {
 				cfg := DefaultConfiguration()
+				cfg.SeenBlockWorkers = runtime.NumCPU()
+				cfg.SerialBlockWorkers = runtime.NumCPU()
 				cfg.Construction = &ConstructionConfiguration{
 					OfflineURL:            DefaultURL,
 					MaxOfflineConnections: DefaultMaxOfflineConnections,
@@ -312,6 +327,8 @@ func TestLoadConfiguration(t *testing.T) {
 			},
 			expected: func() *Configuration {
 				cfg := DefaultConfiguration()
+				cfg.SeenBlockWorkers = runtime.NumCPU()
+				cfg.SerialBlockWorkers = runtime.NumCPU()
 				cfg.Data.EndConditions = &DataEndConditions{
 					ReconciliationCoverage: &ReconciliationCoverage{
 						Coverage:     goodCoverage,
@@ -369,6 +386,8 @@ func TestLoadConfiguration(t *testing.T) {
 			provided: multipleEndConditions,
 			expected: func() *Configuration {
 				def := DefaultConfiguration()
+				def.SeenBlockWorkers = runtime.NumCPU()
+				def.SerialBlockWorkers = runtime.NumCPU()
 				def.Data.EndConditions = multipleEndConditions.Data.EndConditions
 
 				return def
diff --git a/pkg/tester/construction.go b/pkg/tester/construction.go
index 11b40f1..8b58c1e 100644
--- a/pkg/tester/construction.go
+++ b/pkg/tester/construction.go
@@ -121,7 +121,7 @@ func InitializeConstruction(
 		false,
 	)
 
-	blockStorage := modules.NewBlockStorage(localStore)
+	blockStorage := modules.NewBlockStorage(localStore, config.SerialBlockWorkers)
 	keyStorage := modules.NewKeyStorage(localStore)
 	coinStorageHelper := processor.NewCoinStorageHelper(blockStorage)
 	coinStorage := modules.NewCoinStorage(localStore, coinStorageHelper, onlineFetcher.Asserter)
@@ -266,6 +266,7 @@ func InitializeConstruction(
 		statefulsyncer.WithCacheSize(syncer.DefaultCacheSize),
 		statefulsyncer.WithMaxConcurrency(config.MaxSyncConcurrency),
 		statefulsyncer.WithPastBlockLimit(config.MaxReorgDepth),
+		statefulsyncer.WithSeenConcurrency(int64(config.SeenBlockWorkers)),
 	)
 
 	return &ConstructionTester{
diff --git a/pkg/tester/data.go b/pkg/tester/data.go
index 1ecabc9..362e784 100644
--- a/pkg/tester/data.go
+++ b/pkg/tester/data.go
@@ -177,31 +177,9 @@ func InitializeData(
 	}
 
 	counterStorage := modules.NewCounterStorage(localStore)
-	blockStorage := modules.NewBlockStorage(localStore)
+	blockStorage := modules.NewBlockStorage(localStore, config.SerialBlockWorkers)
 	balanceStorage := modules.NewBalanceStorage(localStore)
 
-	// Bootstrap balances, if provided. We need to do before initializing
-	// the reconciler otherwise we won't reconcile bootstrapped accounts
-	// until rosetta-cli restart.
-	if len(config.Data.BootstrapBalances) > 0 {
-		_, err := blockStorage.GetHeadBlockIdentifier(ctx)
-		switch {
-		case err == storageErrs.ErrHeadBlockNotFound:
-			err = balanceStorage.BootstrapBalances(
-				ctx,
-				config.Data.BootstrapBalances,
-				genesisBlock,
-			)
-			if err != nil {
-				log.Fatalf("%s: unable to bootstrap balances", err.Error())
-			}
-		case err != nil:
-			log.Fatalf("%s: unable to get head block identifier", err.Error())
-		default:
-			log.Println("Skipping balance bootstrapping because already started syncing")
-		}
-	}
-
 	logger := logger.NewLogger(
 		dataPath,
 		config.Data.LogBlocks,
@@ -304,6 +282,31 @@ func InitializeData(
 		balanceStorage.Initialize(balanceStorageHelper, balanceStorageHandler)
 
 		blockWorkers = append(blockWorkers, balanceStorage)
+
+		// Bootstrap balances, if provided. We need to do before initializing
+		// the reconciler otherwise we won't reconcile bootstrapped accounts
+		// until rosetta-cli restart.
+		//
+		// We need to do this after instantiating the balance storage handler
+		// because it is invoked within BootstrapBalances.
+		if len(config.Data.BootstrapBalances) > 0 {
+			_, err := blockStorage.GetHeadBlockIdentifier(ctx)
+			switch {
+			case err == storageErrs.ErrHeadBlockNotFound:
+				err = balanceStorage.BootstrapBalances(
+					ctx,
+					config.Data.BootstrapBalances,
+					genesisBlock,
+				)
+				if err != nil {
+					log.Fatalf("%s: unable to bootstrap balances", err.Error())
+				}
+			case err != nil:
+				log.Fatalf("%s: unable to get head block identifier", err.Error())
+			default:
+				log.Println("Skipping balance bootstrapping because already started syncing")
+			}
+		}
 	}
 
 	if !config.Data.CoinTrackingDisabled {
@@ -317,6 +320,7 @@ func InitializeData(
 		statefulsyncer.WithCacheSize(syncer.DefaultCacheSize),
 		statefulsyncer.WithMaxConcurrency(config.MaxSyncConcurrency),
 		statefulsyncer.WithPastBlockLimit(config.MaxReorgDepth),
+		statefulsyncer.WithSeenConcurrency(int64(config.SeenBlockWorkers)),
 	}
 	if config.Data.PruningFrequency != nil {
 		statefulSyncerOptions = append(
@@ -975,7 +979,7 @@ func (t *DataTester) recursiveOpSearch(
 	}
 
 	counterStorage := modules.NewCounterStorage(localStore)
-	blockStorage := modules.NewBlockStorage(localStore)
+	blockStorage := modules.NewBlockStorage(localStore, t.config.SerialBlockWorkers)
 	balanceStorage := modules.NewBalanceStorage(localStore)
 
 	logger := logger.NewLogger(
@@ -1052,6 +1056,7 @@ func (t *DataTester) recursiveOpSearch(
 		statefulsyncer.WithCacheSize(syncer.DefaultCacheSize),
 		statefulsyncer.WithMaxConcurrency(t.config.MaxSyncConcurrency),
 		statefulsyncer.WithPastBlockLimit(t.config.MaxReorgDepth),
+		statefulsyncer.WithSeenConcurrency(int64(t.config.SeenBlockWorkers)),
 	)
 
 	g, ctx := errgroup.WithContext(ctx)