Merge pull request #4 from coinbase/patrick/sync-range-fix

Bootstrap Account Balances

.gitignore

@@ -1 +1,2 @@
 validator-data
+rosetta-validator

Makefile

@@ -4,6 +4,7 @@ LICENCE_SCRIPT=addlicense -c "Coinbase, Inc." -l "apache" -v
 SERVER_URL?=http://localhost:10000
 LOG_TRANSACTIONS?=false
 LOG_BENCHMARKS?=true
+BOOTSTRAP_BALANCES?=false
 
 deps:
 	go get ./...
@@ -32,6 +33,7 @@ salus:
 validate:
 	docker build -t rosetta-validator .; \
 	docker run \
+		--rm \
 		-v ${PWD}/validator-data:/data \
 		-e DATA_DIR="/data" \
 		-e SERVER_URL="${SERVER_URL}" \
@@ -40,6 +42,7 @@ validate:
 		-e ACCOUNT_CONCURRENCY="8" \
 		-e LOG_TRANSACTIONS="${LOG_TRANSACTIONS}" \
 		-e LOG_BENCHMARKS="${LOG_BENCHMARKS}" \
+		-e BOOTSTRAP_BALANCES="${BOOTSTRAP_BALANCES}" \
 		--network host \
 		rosetta-validator \
 		rosetta-validator;

README.md

@@ -18,22 +18,32 @@ and wallets to integrate with much less communication overhead
 and network-specific work.
 
 ## Run the Validator
-
-The validator needs the URL of the Rosetta server configured. This can be set
-as an environment variable named `SERVER_URL`, passed as an argument to make eg `make SERVER_URL=<server url> validate`
-or editing `Makefile` itself.
-
 1. Start your Rosetta Server (and the blockchain node it connects to if it is
 not a single binary.
 2. Start the validator using `make SERVER_URL=<server URL> validate`.
 3. Examine processed blocks using `make watch-blocks`. You can also print transactions
 by setting `LOG_TRANSACTIONS="true"` in the environment or as a `make` argument.
 4. Watch for errors in the processing logs. Any error will cause the validator to stop.
 5. Analyze benchmarks from `validator-data/block_benchmarks.csv` and
-  `validator-data/account_benchmarks.csv` by setting `LOG_BENCHMARKS="true"` in the environment or as a `make` argument.
+`validator-data/account_benchmarks.csv` by setting `LOG_BENCHMARKS="true"` in
+the environment or as a `make` argument.
+
+### Setting the Server URL
+The validator needs the URL of the Rosetta server configured. This can be set
+as an environment variable named `SERVER_URL`, passed as an argument to make
+(ex: `make SERVER_URL=<server url> validate`) or editing `Makefile` itself.
+
+### Bootstrapping Balances
+Blockchains that set balances in genesis must create a `bootstrap_balances.csv`
+file in the `/validator-data` directory and pass `BOOTSTRAP_BALANCES=true` as an
+argument to make. If balances are not bootsrapped and balances are set in genesis,
+reconciliation will fail.
+
+There is an example file in `examples/bootstrap_balances.csv`.
 
-_There is no additional setting required to support blockchains with reorgs. This
-is handled automatically!_
+### Re-orgable Blockchains
+There is no additional setting required to support blockchains with reorgs. This
+is handled automatically!
 
 ## Development
 * `make deps` to install dependencies

examples/bootstrap_balances.csv

@@ -0,0 +1,2 @@
+AccountIdentifier_address,Amount_value,Currency_symbol,Currency_decimals
+1A1zP1eP5QGefi2DMPTfTL5SLmv7DivfNa,5000000000,BTC,8

go.mod

@@ -4,7 +4,7 @@ go 1.13
 
 require (
 	github.com/caarlos0/env v3.5.0+incompatible
-	github.com/coinbase/rosetta-sdk-go v0.0.1
+	github.com/coinbase/rosetta-sdk-go v0.0.3
 	github.com/davecgh/go-spew v1.1.1
 	github.com/dgraph-io/badger v1.6.0
 	github.com/stretchr/testify v1.5.1

go.sum

@@ -8,6 +8,8 @@ github.com/cenkalti/backoff v2.2.1+incompatible h1:tNowT99t7UNflLxfYYSlKYsBpXdEe
 github.com/cenkalti/backoff v2.2.1+incompatible/go.mod h1:90ReRw6GdpyfrHakVjL/QHaoyV4aDUVVkXQJJJ3NXXM=
 github.com/coinbase/rosetta-sdk-go v0.0.1 h1:s6oBsnXCEmTvZxNTHZ4+sjSSWEGCtCBO7kTcED3WILc=
 github.com/coinbase/rosetta-sdk-go v0.0.1/go.mod h1:T7kbh9AOzlxEITJGt2Fu854vxg/yEjy5MsR1woSM5aI=
+github.com/coinbase/rosetta-sdk-go v0.0.3 h1:raFtDs4u0P7h7H+HzHpGQzDYctEpL80nx3e/EY4esXk=
+github.com/coinbase/rosetta-sdk-go v0.0.3/go.mod h1:T7kbh9AOzlxEITJGt2Fu854vxg/yEjy5MsR1woSM5aI=
 github.com/coreos/etcd v3.3.10+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
 github.com/coreos/go-etcd v2.0.0+incompatible/go.mod h1:Jez6KQU2B/sWsbdaef3ED8NzMklzPG4d5KIOhIy30Tk=
 github.com/coreos/go-semver v0.2.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk=

internal/reconciler/reconciler.go

@@ -69,7 +69,7 @@ const (
 
 	// inactiveReconciliationSleep is used as the time.Duration
 	// to sleep when there are no seen accounts to reconcile.
-	inactiveReconciliationSleep = 30 * time.Second
+	inactiveReconciliationSleep = 5 * time.Second
 )
 
 var (
@@ -421,26 +421,25 @@ func simpleAccountAndCurrency(acct *AccountAndCurrency) string {
 func (r *Reconciler) reconcileActiveAccounts(
 	ctx context.Context,
 ) error {
-	for acctIndex := range r.acctQueue {
-		if ctx.Err() != nil {
-			return nil
-		}
-
-		if acctIndex.blockIndex < r.highWaterMark {
-			continue
-		}
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case acctIndex := <-r.acctQueue:
+			if acctIndex.blockIndex < r.highWaterMark {
+				continue
+			}
 
-		err := r.accountReconciliation(
-			ctx,
-			acctIndex.accountAndCurrency,
-			false,
-		)
-		if err != nil {
-			return err
+			err := r.accountReconciliation(
+				ctx,
+				acctIndex.accountAndCurrency,
+				false,
+			)
+			if err != nil {
+				return err
+			}
 		}
 	}
-
-	return nil
 }
 
 // reconcileInactiveAccounts selects a random account

internal/storage/block_storage.go

@@ -18,17 +18,59 @@ import (
 	"bytes"
 	"context"
 	"crypto/sha256"
+	"encoding/csv"
 	"encoding/gob"
 	"errors"
 	"fmt"
+	"io"
 	"log"
 	"math/big"
+	"os"
+	"path"
+	"strconv"
+	"strings"
 
 	rosetta "github.com/coinbase/rosetta-sdk-go/gen"
 
 	"github.com/davecgh/go-spew/spew"
 )
 
+const (
+	// headBlockKey is used to lookup the head block identifier.
+	// The head block is the block with the largest index that is
+	// not orphaned.
+	headBlockKey = "head-block"
+
+	// blockHashNamespace is prepended to any stored block hash.
+	// We cannot just use the stored block key to lookup whether
+	// a hash has been used before because it is concatenated
+	// with the index of the stored block.
+	blockHashNamespace = "block-hash"
+
+	// transactionHashNamespace is prepended to any stored
+	// transaction hash.
+	transactionHashNamespace = "transaction-hash"
+
+	// balanceNamespace is prepended to any stored balance.
+	balanceNamespace = "balance"
+
+	// bootstrapBalancesFile is loaded to bootstrap the balance
+	// of a collection of accounts.
+	bootstrapBalancesFile = "bootstrap_balances.csv"
+
+	// bootstrapBalancesPermissions specifies that the user can
+	// read and write the file.
+	bootstrapBalancesPermissions = 0600
+
+	// bootstrapBalancesHeader is used as the CSV header
+	// in the bootstrapBalancesFile.
+	bootstrapBalancesHeader = "AccountIdentifier_address,Amount_value,Currency_symbol,Currency_decimals"
+	bootstrapAddressIndex   = 0
+	bootstrapValueIndex     = 1
+	bootstrapSymbolIndex    = 2
+	bootstrapDecimalsIndex  = 3
+)
+
 var (
 	// ErrHeadBlockNotFound is returned when there is no
 	// head block found in BlockStorage.
@@ -53,26 +95,14 @@ var (
 	// ErrDuplicateTransactionHash is returned when a transaction
 	// hash cannot be stored because it is a duplicate.
 	ErrDuplicateTransactionHash = errors.New("Duplicate transaction hash")
-)
-
-const (
-	// headBlockKey is used to lookup the head block identifier.
-	// The head block is the block with the largest index that is
-	// not orphaned.
-	headBlockKey = "head-block"
-
-	// blockHashNamespace is prepended to any stored block hash.
-	// We cannot just use the stored block key to lookup whether
-	// a hash has been used before because it is concatenated
-	// with the index of the stored block.
-	blockHashNamespace = "block-hash"
 
-	// transactionHashNamespace is prepended to any stored
-	// transaction hash.
-	transactionHashNamespace = "transaction-hash"
+	// ErrAlreadyStartedSyncing is returned when trying to bootstrap
+	// balances after syncing has started.
+	ErrAlreadyStartedSyncing = errors.New("already started syncing")
 
-	// balanceNamespace is prepended to any stored balance.
-	balanceNamespace = "balance"
+	// ErrIncorrectHeader is returned when a bootstrap file has an
+	// incorrect header.
+	ErrIncorrectHeader = errors.New("incorrect header")
 )
 
 /*
@@ -407,6 +437,7 @@ func (b *BlockStorage) UpdateBalance(
 		if !ok {
 			return fmt.Errorf("%s is not an integer", amount.Value)
 		}
+
 		if newVal.Sign() == -1 {
 			return fmt.Errorf(
 				"%w %+v for %+v at %+v",
@@ -496,3 +527,97 @@ func (b *BlockStorage) GetBalance(
 
 	return deserialBal.Amounts, deserialBal.Block, nil
 }
+
+// BootstrapBalances is utilized to set the balance of
+// any number of AccountIdentifiers at the genesis blocks.
+// This is particularly useful for setting the value of
+// accounts that received an allocation in the genesis block.
+func (b *BlockStorage) BootstrapBalances(
+	ctx context.Context,
+	dataDir string,
+	genesisBlockIdentifier *rosetta.BlockIdentifier,
+) error {
+	f, err := os.OpenFile(
+		path.Join(dataDir, bootstrapBalancesFile),
+		os.O_RDONLY,
+		bootstrapBalancesPermissions,
+	)
+	if err != nil {
+		return err
+	}
+
+	dbTransaction := b.NewDatabaseTransaction(ctx, true)
+	defer dbTransaction.Discard(ctx)
+
+	_, err = b.GetHeadBlockIdentifier(ctx, dbTransaction)
+	if err != ErrHeadBlockNotFound {
+		return ErrAlreadyStartedSyncing
+	}
+
+	csvReader := csv.NewReader(f)
+	rowsRead := 0
+	for {
+		record, err := csvReader.Read()
+		if err == io.EOF {
+			break
+		}
+		rowsRead++
+
+		// Assert header is correct
+		if rowsRead == 1 {
+			if bootstrapBalancesHeader != strings.Join(record[:], ",") {
+				return ErrIncorrectHeader
+			}
+
+			continue
+		}
+
+		// Assert row column length correct
+		if len(record) != len(strings.Split(bootstrapBalancesHeader, ",")) {
+			return fmt.Errorf("row %d does not have expected fields: %s", rowsRead, record)
+		}
+
+		account := &rosetta.AccountIdentifier{
+			Address: record[bootstrapAddressIndex],
+		}
+
+		currencyDecimals, err := strconv.Atoi(record[bootstrapDecimalsIndex])
+		if err != nil {
+			return err
+		}
+
+		amount := &rosetta.Amount{
+			Value: record[bootstrapValueIndex],
+			Currency: &rosetta.Currency{
+				Symbol:   record[bootstrapSymbolIndex],
+				Decimals: int32(currencyDecimals),
+			},
+		}
+
+		log.Printf(
+			"Setting account %s balance to %s %+v\n",
+			account.Address,
+			amount.Value,
+			amount.Currency,
+		)
+
+		err = b.UpdateBalance(
+			ctx,
+			dbTransaction,
+			account,
+			amount,
+			genesisBlockIdentifier,
+		)
+		if err != nil {
+			return err
+		}
+	}
+
+	err = dbTransaction.Commit(ctx)
+	if err != nil {
+		return err
+	}
+
+	log.Printf("%d Balances Bootstrapped\n", rowsRead-1)
+	return nil
+}