Repository: coinbase/rosetta-cli
Pull request #15 (2020-05-06 23:13:13)

Syncer/Reconciler Refactor
### Problems
* Multiple people expressed interest in using the `syncer` and `reconciler` packages in their own applications, however, these packages were tightly coupled to the other packages in `rosetta-cli`.
* In the last few weeks, significant tech debt was created to support running validation from an arbitrary start index without needing to sync from scratch. This lead to tons of duplicate logic in both the `reconciler` and `syncer` in the `stateful_x` and `stateless_x` implementations.
* Without unified logic, 2 commands were created to access the stateful (`check:complete`) and stateless syncer (`check:quick`). Having 2 commands to check validity was extremely confusing for new users.
* For blockchains that had genesis allocations, it was necessary to generate a bootstrap balances file to set balances before beginning syncing. This was a cumbersome tool to build for many Node API implementers.

### Solutions
* Remove all `internal` dependencies from `syncer` and `reconciler` to prepare for a migration to `rosetta-sdk-go`. It is now possible to provide your own storage and communication implementation.
* Abstract logic in `syncer` and `reconciler` to work regardless of whether starting from `genesis` or some arbitrary height. This made the implementation much simpler and removed plenty of redundant code.
* Create new `check` command that determines how to prepare storage to sync from any index. If you start from genesis, it will perform a comprehensive check. If you start from an arbitrary height, limited checks will be run.
* Storage will automatically fetch the balance of an account at the parent block before processing a block containing the account if it is not found. This means it is only necessary to create a bootstrap accounts file if it is not possible to lookup balance by block.

### Future PR
* Move `syncer` and `reconciler` packages to `rosetta-sdk-go`
* Restore the inactive checking queue from a previous instantiation



Patch:
diff --git a/Makefile b/Makefile
index f8e3b1f..a02a247 100644
--- a/Makefile
+++ b/Makefile
@@ -2,15 +2,19 @@
 	check-license shorten-lines salus validate watch-blocks \
 	watch-transactions watch-balances watch-reconciliations \
 	view-block-benchmarks view-account-benchmarks
-LICENCE_SCRIPT=addlicense -c "Coinbase, Inc." -l "apache" -v
+
+# To run the the following packages as commands,
+# it is necessary to use `go run <pkg>`. Running `go get` does
+# not install any binaries that could be used to run
+# the commands directly.
+ADDLICENSE_CMD=go run github.com/google/addlicense
+ADDLICENCE_SCRIPT=${ADDLICENSE_CMD} -c "Coinbase, Inc." -l "apache" -v
+GOLINES_CMD=go run github.com/segmentio/golines
+GOVERALLS_CMD=go run github.com/mattn/goveralls
 TEST_SCRIPT=go test -v ./internal/...
 
 deps:
 	go get ./...
-	go get github.com/stretchr/testify
-	go get github.com/google/addlicense
-	go get github.com/segmentio/golines
-	go get github.com/mattn/goveralls
 
 lint:
 	golangci-lint run -v \
@@ -27,16 +31,16 @@ test:
 
 test-cover:	
 	${TEST_SCRIPT} -coverprofile=c.out -covermode=count
-	goveralls -coverprofile=c.out -repotoken ${COVERALLS_TOKEN}
+	${GOVERALLS_CMD} -coverprofile=c.out -repotoken ${COVERALLS_TOKEN}
 
 add-license:
-	${LICENCE_SCRIPT} .
+	${ADDLICENCE_SCRIPT} .
 
 check-license:
-	${LICENCE_SCRIPT} -check .
+	${ADDLICENCE_SCRIPT} -check .
 
 shorten-lines:
-	golines -w --shorten-comments internal cmd
+	${GOLINES_CMD} -w --shorten-comments internal cmd
 
 salus:
 	docker run --rm -t -v ${PWD}:/home/repo coinbase/salus
diff --git a/README.md b/README.md
index 640b4fa..848925f 100644
--- a/README.md
+++ b/README.md
@@ -30,15 +30,60 @@ Usage:
   rosetta-cli [command]
 
 Available Commands:
-  check:account  Debug inactive reconciliation errors for a group of accounts
-  check:complete Run a full check of the correctness of a Rosetta server
-  check:quick    Run a simple check of the correctness of a Rosetta server
-  help           Help about any command
+  check                Check the correctness of a Rosetta Node API Server
+  create:configuration Generate a static configuration file for the Asserter
+  help                 Help about any command
+  view:account         View an account balance
+  view:block           View a block
+
+Flags:
+  -h, --help                help for rosetta-cli
+      --server-url string   base URL for a Rosetta server (default "http://localhost:8080")
+
+Use "rosetta-cli [command] --help" for more information about a command.
+```
+
+### check
+```
+Check all server responses are properly constructed, that
+there are no duplicate blocks and transactions, that blocks can be processed
+from genesis to the current block (re-orgs handled automatically), and that
+computed balance changes are equal to balance changes reported by the node.
+
+When re-running this command, it will start where it left off if you specify
+some --data-dir. Otherwise, it will create a new temporary directory and start
+again from the genesis block. If you want to discard some number of blocks
+populate the --start flag with some block index. Starting from a given index
+can be useful to debug a small range of blocks for issues but it is highly
+recommended you sync from start to finish to ensure all correctness checks
+are performed.
+
+By default, account balances are looked up at specific heights (instead of
+only at the current block). If your node does not support this functionality
+set --lookup-balance-by-block to false. This will make reconciliation much
+less efficient but it will still work.
+
+To debug an INACTIVE account reconciliation error, set the
+--interesting-accounts flag to the absolute path of a JSON file containing
+accounts that will be actively checked for balance changes at each block. This
+will return an error at the block where a balance change occurred with no
+corresponding operations.
+
+If your blockchain has a genesis allocation of funds and you set
+--lookup-balance-by-block to false, you must provide an
+absolute path to a JSON file containing initial balances with the
+--bootstrap-balances flag. You can look at the examples folder for an example
+of what one of these files looks like.
+
+Usage:
+  rosetta-cli check [flags]
 
 Flags:
       --account-concurrency uint       concurrency to use while fetching accounts during reconciliation (default 8)
       --block-concurrency uint         concurrency to use while fetching blocks (default 8)
-      --data-dir string                folder used to store logs and any data used to perform validation (default "./validator-data")
+      --bootstrap-balances string      Absolute path to a file used to bootstrap balances before starting syncing.
+                                       Populating this value after beginning syncing will return an error.
+      --data-dir string                folder used to store logs and any data used to perform validation
       --end int                        block index to stop syncing (default -1)
       --exempt-accounts string         Absolute path to a file listing all accounts to exempt from balance
                                        tracking and reconciliation. Look at the examples directory for an example of
@@ -46,103 +91,87 @@ Flags:
       --halt-on-reconciliation-error   Determines if block processing should halt on a reconciliation
                                        error. It can be beneficial to collect all reconciliation errors or silence
                                        reconciliation errors during development. (default true)
-  -h, --help                           help for rosetta-cli
+  -h, --help                           help for check
+      --interesting-accounts string    Absolute path to a file listing all accounts to check on each block. Look
+                                       at the examples directory for an example of how to structure this file.
       --log-balance-changes            log balance changes
       --log-blocks                     log processed blocks
       --log-reconciliations            log balance reconciliations
       --log-transactions               log processed transactions
-      --server-url string              base URL for a Rosetta server to validate (default "http://localhost:8080")
+      --lookup-balance-by-block        When set to true, balances are looked up at the block where a balance
+                                       change occurred instead of at the current block. Blockchains that do not support
+                                       historical balance lookup should set this to false. (default true)
       --start int                      block index to start syncing (default -1)
       --transaction-concurrency uint   concurrency to use while fetching transactions (if required) (default 16)
 
-Use "rosetta-cli [command] --help" for more information about a command.
+Global Flags:
+      --server-url string   base URL for a Rosetta server (default "http://localhost:8080")
 ```
 
-### check:complete
+### create:configuration
 ```
-Check all server responses are properly constructed, that
-there are no duplicate blocks and transactions, that blocks can be processed
-from genesis to the current block (re-orgs handled automatically), and that
-computed balance changes are equal to balance changes reported by the node.
+In production deployments, it is useful to initialize the response
+Asserter (https://github.com/coinbase/rosetta-sdk-go/tree/master/asserter) using
+a static configuration instead of intializing a configuration dynamically
+from the node. This allows a client to error on new types/statuses that may
+have been added in an update instead of silently erroring.
 
-When re-running this command, it will start where it left off. If you want
-to discard some number of blocks populate the --start flag with some block
-index less than the last computed block index.
+To use this command, simply provide an absolute path as the argument for where
+the configuration file should be saved (in JSON). Populate the optional
+--server-url flag with the url of the server to generate the configuration
+from.
 
 Usage:
-  rosetta-cli check:complete [flags]
+  rosetta-cli create:configuration [flags]
 
 Flags:
-      --bootstrap-balances string   Absolute path to a file used to bootstrap balances before starting syncing.
-                                    Populating this value after beginning syncing will return an error.
-  -h, --help                        help for check:complete
-      --lookup-balance-by-block     When set to true, balances are looked up at the block where a balance
-                                    change occurred instead of at the current block. Blockchains that do not support
-                                    historical balance lookup should set this to false. (default true)
-```
+  -h, --help   help for create:configuration
 
-### check:quick
+Global Flags:
+      --server-url string   base URL for a Rosetta server (default "http://localhost:8080")
 ```
-Check all server responses are properly constructed and that
-computed balance changes are equal to balance changes reported by the
-node. To use check:quick, your server must implement the balance lookup
-by block.
-
-Unlike check:complete, which requires syncing all blocks up
-to the blocks you want to check, check:quick allows you to validate
-an arbitrary range of blocks (even if earlier blocks weren't synced).
-To do this, all you need to do is provide a --start flag and optionally
-an --end flag.
 
-It is important to note that check:quick does not support re-orgs and it
-does not check for duplicate blocks and transactions. For these features,
-please use check:complete.
+### view:account
+```
+While debugging, it is often useful to inspect the state
+of an account at a certain block. This command allows you to look up
+any account by providing a JSON representation of a types.AccountIdentifier
+(and optionally a height to perform the query).
 
-When re-running this command, it will start off from genesis unless you
-provide a populated --start flag. If you want to run a stateful validation,
-use the check:complete command.
+For example, you could run view:account '{"address":"interesting address"}' 1000
+to lookup the balance of an interesting address at block 1000. Allowing the
+address to specified as JSON allows for querying by SubAccountIdentifier.
 
 Usage:
-  rosetta-cli check:quick [flags]
+  rosetta-cli view:account [flags]
 
 Flags:
-  -h, --help   help for check:quick
+  -h, --help   help for view:account
+
+Global Flags:
+      --server-url string   base URL for a Rosetta server (default "http://localhost:8080")
 ```
 
-### check:account
+### view:block
 ```
-check:complete identifies accounts with inactive reconciliation
-errors (when the balance of an account changes without any operations), however,
-it does not identify which block the untracked balance change occurred. This tool
-is used for locating exactly which block was missing an operation for a
-particular account and currency.
+While debugging a Node API implementation, it can be very
+useful to inspect block contents. This command allows you to fetch any
+block by index to inspect its contents. It uses the
+fetcher (https://github.com/coinbase/rosetta-sdk-go/tree/master/fetcher) package
+to automatically get all transactions in the block and assert the format
+of the block is correct before printing.
 
-In the future, this tool will be deprecated as check:complete
-will automatically identify the block where the missing operation occurred.
+If this command errors, it is likely because the block you are trying to
+fetch is formatted incorrectly.
 
 Usage:
-  rosetta-cli check:account [flags]
+  rosetta-cli view:block [flags]
 
 Flags:
-  -h, --help                          help for check:account
-      --interesting-accounts string   Absolute path to a file listing all accounts to check on each block. Look
-                                      at the examples directory for an example of how to structure this file.
+  -h, --help   help for view:block
 
 Global Flags:
-      --account-concurrency uint       concurrency to use while fetching accounts during reconciliation (default 8)
-      --block-concurrency uint         concurrency to use while fetching blocks (default 8)
-      --data-dir string                folder used to store logs and any data used to perform validation (default "./validator-data")
-      --end int                        block index to stop syncing (default -1)
-      --halt-on-reconciliation-error   Determines if block processing should halt on a reconciliation
-                                       error. It can be beneficial to collect all reconciliation errors or silence
-                                       reconciliation errors during development. (default true)
-      --log-balance-changes            log balance changes
-      --log-blocks                     log processed blocks
-      --log-reconciliations            log balance reconciliations
-      --log-transactions               log processed transactions
-      --server-url string              base URL for a Rosetta server to validate (default "http://localhost:8080")
-      --start int                      block index to start syncing (default -1)
-      --transaction-concurrency uint   concurrency to use while fetching transactions (if required) (default 16)
+      --server-url string   base URL for a Rosetta server (default "http://localhost:8080")
 ```
 
 ## Development
@@ -151,6 +180,26 @@ Global Flags:
 * `make lint` to lint the source code (included generated code)
 * `make release` to run one last check before opening a PR
 
+### Helper/Handler
+Many of the internal packages use a `Helper/Handler` interface pattern to acquire
+required information or to send events to some client implementation. An example
+of this is in the `internal/reconciler` package where a `Helper` is used to get
+the account balance and the `Handler` is called to incidate whether the
+reconciliation of an account was successful.
+
+### Repo Structure
+```
+cmd
+examples // examples of different config files
+internal
+  logger // logic to write syncing information to stdout/files
+  processor // Helper/Handler implementations for reconciler, storage, and syncer
+  reconciler // checks for equality between computed balance and node balance
+  storage // persists block to temporary storage and allows for querying balances
+  syncer // coordinates block syncing (inlcuding re-orgs)
+  utils // useful functions
+```
+
 ## Correctness Checks
 This tool performs a variety of correctness checks using the Rosetta Server. If
 any correctness check fails, the validator will exit and print out a detailed
@@ -180,7 +229,7 @@ exit.
 The validator randomly checks the balances of accounts that aren't
 involved in any transactions. The balances of accounts could change
 on the blockchain node without being included in an operation
-returned by the Rosetta Server. Recall that **ALL** balance-changing
+returned by the Rosetta Server. Recall that all balance-changing
 operations must be returned by the Rosetta Server.
 
 ## Future Work
diff --git a/cmd/check.go b/cmd/check.go
new file mode 100644
index 0000000..8d029b7
--- /dev/null
+++ b/cmd/check.go
@@ -0,0 +1,421 @@
+// Copyright 2020 Coinbase, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package cmd
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"log"
+	"path"
+	"time"
+
+	"github.com/coinbase/rosetta-cli/internal/logger"
+	"github.com/coinbase/rosetta-cli/internal/processor"
+	"github.com/coinbase/rosetta-cli/internal/reconciler"
+	"github.com/coinbase/rosetta-cli/internal/storage"
+	"github.com/coinbase/rosetta-cli/internal/syncer"
+	"github.com/coinbase/rosetta-cli/internal/utils"
+
+	"github.com/coinbase/rosetta-sdk-go/fetcher"
+	"github.com/coinbase/rosetta-sdk-go/types"
+	"github.com/spf13/cobra"
+	"golang.org/x/sync/errgroup"
+)
+
+const (
+	// ExtendedRetryElapsedTime is used to override the default fetcher
+	// retry elapsed time. In practice, extending the retry elapsed time
+	// has prevented retry exhaustion errors when many goroutines are
+	// used to fetch data from the Rosetta server.
+	//
+	// TODO: make configurable
+	ExtendedRetryElapsedTime = 5 * time.Minute
+)
+
+var (
+	checkCmd = &cobra.Command{
+		Use:   "check",
+		Short: "Check the correctness of a Rosetta Node API Server",
+		Long: `Check all server responses are properly constructed, that
+there are no duplicate blocks and transactions, that blocks can be processed
+from genesis to the current block (re-orgs handled automatically), and that
+computed balance changes are equal to balance changes reported by the node.
+
+When re-running this command, it will start where it left off if you specify
+some --data-dir. Otherwise, it will create a new temporary directory and start
+again from the genesis block. If you want to discard some number of blocks
+populate the --start flag with some block index. Starting from a given index
+can be useful to debug a small range of blocks for issues but it is highly
+recommended you sync from start to finish to ensure all correctness checks
+are performed.
+
+By default, account balances are looked up at specific heights (instead of
+only at the current block). If your node does not support this functionality
+set --lookup-balance-by-block to false. This will make reconciliation much
+less efficient but it will still work.
+
+To debug an INACTIVE account reconciliation error, set the
+--interesting-accounts flag to the absolute path of a JSON file containing
+accounts that will be actively checked for balance changes at each block. This
+will return an error at the block where a balance change occurred with no
+corresponding operations.
+
+If your blockchain has a genesis allocation of funds and you set
+--lookup-balance-by-block to false, you must provide an
+absolute path to a JSON file containing initial balances with the
+--bootstrap-balances flag. You can look at the examples folder for an example
+of what one of these files looks like.`,
+		Run: runCheckCmd,
+	}
+
+	// BootstrapBalances is a path to a file used to bootstrap
+	// balances before starting syncing. Populating this value
+	// after beginning syncing will return an error.
+	BootstrapBalances string
+
+	// LookupBalanceByBlock determines if balances are looked up
+	// at the block where a balance change occurred instead of at the current
+	// block. Blockchains that do not support historical balance lookup
+	// should set this to false.
+	LookupBalanceByBlock bool
+
+	// DataDir is a folder used to store logs
+	// and any data used to perform validation.
+	DataDir string
+
+	// StartIndex is the block index to start syncing.
+	StartIndex int64
+
+	// EndIndex is the block index to stop syncing.
+	EndIndex int64
+
+	// BlockConcurrency is the concurrency to use
+	// while fetching blocks.
+	BlockConcurrency uint64
+
+	// TransactionConcurrency is the concurrency to use
+	// while fetching transactions (if required).
+	TransactionConcurrency uint64
+
+	// AccountConcurrency is the concurrency to use
+	// while fetching accounts during reconciliation.
+	AccountConcurrency uint64
+
+	// LogBlocks determines if blocks are
+	// logged.
+	LogBlocks bool
+
+	// LogTransactions determines if transactions are
+	// logged.
+	LogTransactions bool
+
+	// LogBalanceChanges determines if balance changes are
+	// logged.
+	LogBalanceChanges bool
+
+	// LogReconciliations determines if reconciliations are
+	// logged.
+	LogReconciliations bool
+
+	// HaltOnReconciliationError determines if processing
+	// should stop when encountering a reconciliation error.
+	// It can be beneficial to collect all reconciliation errors
+	// during development.
+	HaltOnReconciliationError bool
+
+	// ExemptFile is an absolute path to a file listing all accounts
+	// to exempt from balance tracking and reconciliation.
+	ExemptFile string
+
+	// InterestingFile is an absolute path to a file listing all accounts
+	// to actively reconcile on each block (if there are no operations
+	// present for the account, the reconciler asserts a balance change of 0).
+	InterestingFile string
+)
+
+func loadAccounts(filePath string) ([]*reconciler.AccountCurrency, error) {
+	if len(filePath) == 0 {
+		return []*reconciler.AccountCurrency{}, nil
+	}
+
+	accountsRaw, err := ioutil.ReadFile(path.Clean(filePath))
+	if err != nil {
+		return nil, err
+	}
+
+	accounts := []*reconciler.AccountCurrency{}
+	if err := json.Unmarshal(accountsRaw, &accounts); err != nil {
+		return nil, err
+	}
+
+	log.Printf(
+		"Found %d accounts at %s: %s\n",
+		len(accounts),
+		filePath,
+		types.PrettyPrintStruct(accounts),
+	)
+
+	return accounts, nil
+}
+
+func init() {
+	checkCmd.Flags().StringVar(
+		&DataDir,
+		"data-dir",
+		"",
+		"folder used to store logs and any data used to perform validation",
+	)
+	checkCmd.Flags().Int64Var(
+		&StartIndex,
+		"start",
+		-1,
+		"block index to start syncing",
+	)
+	checkCmd.Flags().Int64Var(
+		&EndIndex,
+		"end",
+		-1,
+		"block index to stop syncing",
+	)
+	checkCmd.Flags().Uint64Var(
+		&BlockConcurrency,
+		"block-concurrency",
+		8,
+		"concurrency to use while fetching blocks",
+	)
+	checkCmd.Flags().Uint64Var(
+		&TransactionConcurrency,
+		"transaction-concurrency",
+		16,
+		"concurrency to use while fetching transactions (if required)",
+	)
+	checkCmd.Flags().Uint64Var(
+		&AccountConcurrency,
+		"account-concurrency",
+		8,
+		"concurrency to use while fetching accounts during reconciliation",
+	)
+	checkCmd.Flags().BoolVar(
+		&LogBlocks,
+		"log-blocks",
+		false,
+		"log processed blocks",
+	)
+	checkCmd.Flags().BoolVar(
+		&LogTransactions,
+		"log-transactions",
+		false,
+		"log processed transactions",
+	)
+	checkCmd.Flags().BoolVar(
+		&LogBalanceChanges,
+		"log-balance-changes",
+		false,
+		"log balance changes",
+	)
+	checkCmd.Flags().BoolVar(
+		&LogReconciliations,
+		"log-reconciliations",
+		false,
+		"log balance reconciliations",
+	)
+	checkCmd.Flags().BoolVar(
+		&HaltOnReconciliationError,
+		"halt-on-reconciliation-error",
+		true,
+		`Determines if block processing should halt on a reconciliation
+error. It can be beneficial to collect all reconciliation errors or silence
+reconciliation errors during development.`,
+	)
+	checkCmd.Flags().StringVar(
+		&ExemptFile,
+		"exempt-accounts",
+		"",
+		`Absolute path to a file listing all accounts to exempt from balance
+tracking and reconciliation. Look at the examples directory for an example of
+how to structure this file.`,
+	)
+	checkCmd.Flags().StringVar(
+		&BootstrapBalances,
+		"bootstrap-balances",
+		"",
+		`Absolute path to a file used to bootstrap balances before starting syncing.
+Populating this value after beginning syncing will return an error.`,
+	)
+	checkCmd.Flags().BoolVar(
+		&LookupBalanceByBlock,
+		"lookup-balance-by-block",
+		true,
+		`When set to true, balances are looked up at the block where a balance
+change occurred instead of at the current block. Blockchains that do not support
+historical balance lookup should set this to false.`,
+	)
+	checkCmd.Flags().StringVar(
+		&InterestingFile,
+		"interesting-accounts",
+		"",
+		`Absolute path to a file listing all accounts to check on each block. Look
+at the examples directory for an example of how to structure this file.`,
+	)
+}
+
+func runCheckCmd(cmd *cobra.Command, args []string) {
+	ctx, cancel := context.WithCancel(context.Background())
+
+	exemptAccounts, err := loadAccounts(ExemptFile)
+	if err != nil {
+		log.Fatal(fmt.Errorf("%w: unable to load exempt accounts", err))
+	}
+
+	interestingAccounts, err := loadAccounts(InterestingFile)
+	if err != nil {
+		log.Fatal(fmt.Errorf("%w: unable to load interesting accounts", err))
+	}
+
+	fetcher := fetcher.New(
+		ServerURL,
+		fetcher.WithBlockConcurrency(BlockConcurrency),
+		fetcher.WithTransactionConcurrency(TransactionConcurrency),
+		fetcher.WithRetryElapsedTime(ExtendedRetryElapsedTime),
+	)
+
+	// TODO: sync and reconcile on subnetworks, if they exist.
+	primaryNetwork, networkStatus, err := fetcher.InitializeAsserter(ctx)
+	if err != nil {
+		log.Fatal(fmt.Errorf("%w: unable to initialize asserter", err))
+	}
+
+	// If data directory is not specified, we use a temporary directory
+	// and delete its contents when execution is complete.
+	if len(DataDir) == 0 {
+		tmpDir, err := utils.CreateTempDir()
+		if err != nil {
+			log.Fatal(fmt.Errorf("%w: unable to create temporary directory", err))
+		}
+		defer utils.RemoveTempDir(tmpDir)
+
+		DataDir = tmpDir
+	}
+	localStore, err := storage.NewBadgerStorage(ctx, DataDir)
+	if err != nil {
+		log.Fatal(fmt.Errorf("%w: unable to initialize data store", err))
+	}
+
+	logger := logger.NewLogger(
+		DataDir,
+		LogBlocks,
+		LogTransactions,
+		LogBalanceChanges,
+		LogReconciliations,
+	)
+
+	blockStorageHelper := processor.NewBlockStorageHelper(
+		primaryNetwork,
+		fetcher,
+		LookupBalanceByBlock,
+		exemptAccounts,
+	)
+
+	blockStorage := storage.NewBlockStorage(ctx, localStore, blockStorageHelper)
+	// Bootstrap balances if provided
+	if len(BootstrapBalances) > 0 {
+		err = blockStorage.BootstrapBalances(
+			ctx,
+			BootstrapBalances,
+			networkStatus.GenesisBlockIdentifier,
+		)
+		if err != nil {
+			log.Fatal(fmt.Errorf("%w: unable to bootstrap balances", err))
+		}
+	}
+
+	// Ensure storage is in correct state for starting at index
+	if StartIndex != -1 { // attempt to remove blocks from storage (without handling)
+		if err = blockStorage.SetNewStartIndex(ctx, StartIndex); err != nil {
+			log.Fatal(fmt.Errorf("%w: unable to set new start index", err))
+		}
+	} else { // attempt to load last processed index
+		head, err := blockStorage.GetHeadBlockIdentifier(ctx)
+		if err == nil {
+			StartIndex = head.Index + 1
+		}
+	}
+
+	reconcilerHelper := processor.NewReconcilerHelper(
+		blockStorage,
+	)
+
+	reconcilerHandler := processor.NewReconcilerHandler(
+		cancel,
+		logger,
+		HaltOnReconciliationError,
+	)
+
+	r := reconciler.NewReconciler(
+		primaryNetwork,
+		reconcilerHelper,
+		reconcilerHandler,
+		fetcher,
+		AccountConcurrency,
+		LookupBalanceByBlock,
+		interestingAccounts,
+	)
+
+	syncerHandler := processor.NewSyncerHandler(
+		blockStorage,
+		logger,
+		r,
+		fetcher,
+	)
+
+	g, ctx := errgroup.WithContext(ctx)
+
+	g.Go(func() error {
+		return r.Reconcile(ctx)
+	})
+
+	// Load in previous blocks into syncer cache to handle reorgs.
+	// If previously processed blocks exist in storage, they are fetched.
+	// Otherwise, none are provided to the cache (the syncer will not attempt
+	// a reorg if the cache is empty).
+	pastBlocks := []*types.BlockIdentifier{}
+	if StartIndex != -1 {
+		// This is the case if blocks already in storage or if stateless start
+		pastBlocks = blockStorage.CreateBlockCache(ctx)
+	}
+
+	syncer := syncer.New(
+		primaryNetwork,
+		fetcher,
+		syncerHandler,
+		cancel,
+		pastBlocks,
+	)
+
+	g.Go(func() error {
+		return syncer.Sync(
+			ctx,
+			StartIndex,
+			EndIndex,
+		)
+	})
+
+	err = g.Wait()
+	if err != nil {
+		log.Fatal(err)
+	}
+}
diff --git a/cmd/check_account.go b/cmd/check_account.go
deleted file mode 100644
index 9dfdb8f..0000000
--- a/cmd/check_account.go
+++ /dev/null
@@ -1,138 +0,0 @@
-// Copyright 2020 Coinbase, Inc.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//      http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package cmd
-
-import (
-	"context"
-	"log"
-
-	"github.com/coinbase/rosetta-cli/internal/logger"
-	"github.com/coinbase/rosetta-cli/internal/reconciler"
-	"github.com/coinbase/rosetta-cli/internal/syncer"
-
-	"github.com/coinbase/rosetta-sdk-go/fetcher"
-	"github.com/spf13/cobra"
-	"golang.org/x/sync/errgroup"
-)
-
-var (
-	checkAccountCmd = &cobra.Command{
-		Use:   "check:account",
-		Short: "Debug inactive reconciliation errors for a group of accounts",
-		Long: `check:complete identifies accounts with inactive reconciliation
-errors (when the balance of an account changes without any operations), however,
-it does not identify which block the untracked balance change occurred. This tool
-is used for locating exactly which block was missing an operation for a
-particular account and currency.
-
-In the future, this tool will be deprecated as check:complete
-will automatically identify the block where the missing operation occurred.`,
-		Run: runCheckAccountCmd,
-	}
-
-	accountFile string
-)
-
-func init() {
-	checkAccountCmd.Flags().StringVar(
-		&accountFile,
-		"interesting-accounts",
-		"",
-		`Absolute path to a file listing all accounts to check on each block. Look
-at the examples directory for an example of how to structure this file.`,
-	)
-
-	err := checkAccountCmd.MarkFlagRequired("interesting-accounts")
-	if err != nil {
-		log.Fatal(err)
-	}
-}
-
-func runCheckAccountCmd(cmd *cobra.Command, args []string) {
-	// TODO: unify startup logic with stateless
-	ctx, cancel := context.WithCancel(context.Background())
-
-	interestingAccounts, err := loadAccounts(accountFile)
-	if err != nil {
-		log.Fatal(err)
-	}
-
-	exemptAccounts, err := loadAccounts(ExemptFile)
-	if err != nil {
-		log.Fatal(err)
-	}
-
-	fetcher := fetcher.New(
-		ServerURL,
-		fetcher.WithBlockConcurrency(BlockConcurrency),
-		fetcher.WithTransactionConcurrency(TransactionConcurrency),
-		fetcher.WithRetryElapsedTime(ExtendedRetryElapsedTime),
-	)
-
-	primaryNetwork, _, err := fetcher.InitializeAsserter(ctx)
-	if err != nil {
-		log.Fatal(err)
-	}
-
-	logger := logger.NewLogger(
-		DataDir,
-		LogBlocks,
-		LogTransactions,
-		LogBalanceChanges,
-		LogReconciliations,
-	)
-
-	g, ctx := errgroup.WithContext(ctx)
-
-	r := reconciler.NewStateless(
-		primaryNetwork,
-		fetcher,
-		logger,
-		AccountConcurrency,
-		HaltOnReconciliationError,
-		interestingAccounts,
-	)
-
-	g.Go(func() error {
-		return r.Reconcile(ctx)
-	})
-
-	syncHandler := syncer.NewBaseHandler(
-		logger,
-		r,
-		exemptAccounts,
-	)
-
-	statelessSyncer := syncer.NewStateless(
-		primaryNetwork,
-		fetcher,
-		syncHandler,
-	)
-
-	g.Go(func() error {
-		return syncer.Sync(
-			ctx,
-			cancel,
-			statelessSyncer,
-			StartIndex,
-			EndIndex,
-		)
-	})
-
-	err = g.Wait()
-	if err != nil {
-		log.Fatal(err)
-	}
-}
diff --git a/cmd/check_complete.go b/cmd/check_complete.go
deleted file mode 100644
index 190ddc3..0000000
--- a/cmd/check_complete.go
+++ /dev/null
@@ -1,166 +0,0 @@
-// Copyright 2020 Coinbase, Inc.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//      http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package cmd
-
-import (
-	"context"
-	"fmt"
-	"log"
-
-	"github.com/coinbase/rosetta-cli/internal/logger"
-	"github.com/coinbase/rosetta-cli/internal/reconciler"
-	"github.com/coinbase/rosetta-cli/internal/storage"
-	"github.com/coinbase/rosetta-cli/internal/syncer"
-
-	"github.com/coinbase/rosetta-sdk-go/fetcher"
-	"github.com/spf13/cobra"
-	"golang.org/x/sync/errgroup"
-)
-
-var (
-	checkCompleteCmd = &cobra.Command{
-		Use:   "check:complete",
-		Short: "Run a full check of the correctness of a Rosetta server",
-		Long: `Check all server responses are properly constructed, that
-there are no duplicate blocks and transactions, that blocks can be processed
-from genesis to the current block (re-orgs handled automatically), and that
-computed balance changes are equal to balance changes reported by the node.
-
-When re-running this command, it will start where it left off. If you want
-to discard some number of blocks populate the --start flag with some block
-index less than the last computed block index.`,
-		Run: runCheckCompleteCmd,
-	}
-
-	// BootstrapBalances is a path to a file used to bootstrap
-	// balances before starting syncing. Populating this value
-	// after beginning syncing will return an error.
-	BootstrapBalances string
-
-	// LookupBalanceByBlock determines if balances are looked up
-	// at the block where a balance change occurred instead of at the current
-	// block. Blockchains that do not support historical balance lookup
-	// should set this to false.
-	LookupBalanceByBlock bool
-)
-
-func init() {
-	checkCompleteCmd.Flags().StringVar(
-		&BootstrapBalances,
-		"bootstrap-balances",
-		"",
-		`Absolute path to a file used to bootstrap balances before starting syncing.
-Populating this value after beginning syncing will return an error.`,
-	)
-	checkCompleteCmd.Flags().BoolVar(
-		&LookupBalanceByBlock,
-		"lookup-balance-by-block",
-		true,
-		`When set to true, balances are looked up at the block where a balance
-change occurred instead of at the current block. Blockchains that do not support
-historical balance lookup should set this to false.`,
-	)
-}
-
-func runCheckCompleteCmd(cmd *cobra.Command, args []string) {
-	ctx, cancel := context.WithCancel(context.Background())
-
-	exemptAccounts, err := loadAccounts(ExemptFile)
-	if err != nil {
-		log.Fatal(fmt.Errorf("%w: unable to load exempt accounts", err))
-	}
-
-	fetcher := fetcher.New(
-		ServerURL,
-		fetcher.WithBlockConcurrency(BlockConcurrency),
-		fetcher.WithTransactionConcurrency(TransactionConcurrency),
-		fetcher.WithRetryElapsedTime(ExtendedRetryElapsedTime),
-	)
-
-	// TODO: sync and reconcile on subnetworks, if they exist.
-	primaryNetwork, networkStatus, err := fetcher.InitializeAsserter(ctx)
-	if err != nil {
-		log.Fatal(fmt.Errorf("%w: unable to initialize asserter", err))
-	}
-
-	localStore, err := storage.NewBadgerStorage(ctx, DataDir)
-	if err != nil {
-		log.Fatal(fmt.Errorf("%w: unable to initialize data store", err))
-	}
-
-	blockStorage := storage.NewBlockStorage(ctx, localStore)
-	if len(BootstrapBalances) > 0 {
-		err = blockStorage.BootstrapBalances(
-			ctx,
-			BootstrapBalances,
-			networkStatus.GenesisBlockIdentifier,
-		)
-		if err != nil {
-			log.Fatal(fmt.Errorf("%w: unable to bootstrap balances", err))
-		}
-	}
-
-	logger := logger.NewLogger(
-		DataDir,
-		LogBlocks,
-		LogTransactions,
-		LogBalanceChanges,
-		LogReconciliations,
-	)
-
-	g, ctx := errgroup.WithContext(ctx)
-
-	r := reconciler.NewStateful(
-		primaryNetwork,
-		blockStorage,
-		fetcher,
-		logger,
-		AccountConcurrency,
-		LookupBalanceByBlock,
-		HaltOnReconciliationError,
-	)
-
-	g.Go(func() error {
-		return r.Reconcile(ctx)
-	})
-
-	syncHandler := syncer.NewBaseHandler(
-		logger,
-		r,
-		exemptAccounts,
-	)
-
-	statefulSyncer := syncer.NewStateful(
-		primaryNetwork,
-		blockStorage,
-		fetcher,
-		syncHandler,
-	)
-
-	g.Go(func() error {
-		return syncer.Sync(
-			ctx,
-			cancel,
-			statefulSyncer,
-			StartIndex,
-			EndIndex,
-		)
-	})
-
-	err = g.Wait()
-	if err != nil {
-		log.Fatal(err)
-	}
-}
diff --git a/cmd/check_quick.go b/cmd/check_quick.go
deleted file mode 100644
index 365dde6..0000000
--- a/cmd/check_quick.go
+++ /dev/null
@@ -1,125 +0,0 @@
-// Copyright 2020 Coinbase, Inc.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//      http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package cmd
-
-import (
-	"context"
-	"log"
-
-	"github.com/coinbase/rosetta-cli/internal/logger"
-	"github.com/coinbase/rosetta-cli/internal/reconciler"
-	"github.com/coinbase/rosetta-cli/internal/syncer"
-
-	"github.com/coinbase/rosetta-sdk-go/fetcher"
-	"github.com/spf13/cobra"
-	"golang.org/x/sync/errgroup"
-)
-
-var (
-	checkQuickCmd = &cobra.Command{
-		Use:   "check:quick",
-		Short: "Run a simple check of the correctness of a Rosetta server",
-		Long: `Check all server responses are properly constructed and that
-computed balance changes are equal to balance changes reported by the
-node. To use check:quick, your server must implement the balance lookup
-by block.
-
-Unlike check:complete, which requires syncing all blocks up
-to the blocks you want to check, check:quick allows you to validate
-an arbitrary range of blocks (even if earlier blocks weren't synced).
-To do this, all you need to do is provide a --start flag and optionally
-an --end flag.
-
-It is important to note that check:quick does not support re-orgs and it
-does not check for duplicate blocks and transactions. For these features,
-please use check:complete.
-
-When re-running this command, it will start off from genesis unless you
-provide a populated --start flag. If you want to run a stateful validation,
-use the check:complete command.`,
-		Run: runCheckQuickCmd,
-	}
-)
-
-func runCheckQuickCmd(cmd *cobra.Command, args []string) {
-	ctx, cancel := context.WithCancel(context.Background())
-
-	exemptAccounts, err := loadAccounts(ExemptFile)
-	if err != nil {
-		log.Fatal(err)
-	}
-
-	fetcher := fetcher.New(
-		ServerURL,
-		fetcher.WithBlockConcurrency(BlockConcurrency),
-		fetcher.WithTransactionConcurrency(TransactionConcurrency),
-		fetcher.WithRetryElapsedTime(ExtendedRetryElapsedTime),
-	)
-
-	primaryNetwork, _, err := fetcher.InitializeAsserter(ctx)
-	if err != nil {
-		log.Fatal(err)
-	}
-
-	logger := logger.NewLogger(
-		DataDir,
-		LogBlocks,
-		LogTransactions,
-		LogBalanceChanges,
-		LogReconciliations,
-	)
-
-	g, ctx := errgroup.WithContext(ctx)
-
-	r := reconciler.NewStateless(
-		primaryNetwork,
-		fetcher,
-		logger,
-		AccountConcurrency,
-		HaltOnReconciliationError,
-		nil,
-	)
-
-	g.Go(func() error {
-		return r.Reconcile(ctx)
-	})
-
-	syncHandler := syncer.NewBaseHandler(
-		logger,
-		r,
-		exemptAccounts,
-	)
-
-	statelessSyncer := syncer.NewStateless(
-		primaryNetwork,
-		fetcher,
-		syncHandler,
-	)
-
-	g.Go(func() error {
-		return syncer.Sync(
-			ctx,
-			cancel,
-			statelessSyncer,
-			StartIndex,
-			EndIndex,
-		)
-	})
-
-	err = g.Wait()
-	if err != nil {
-		log.Fatal(err)
-	}
-}
diff --git a/cmd/create_configuration.go b/cmd/create_configuration.go
new file mode 100644
index 0000000..fd66cc6
--- /dev/null
+++ b/cmd/create_configuration.go
@@ -0,0 +1,80 @@
+// Copyright 2020 Coinbase, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package cmd
+
+import (
+	"context"
+	"fmt"
+	"io/ioutil"
+	"log"
+	"os"
+	"path"
+
+	"github.com/coinbase/rosetta-sdk-go/fetcher"
+	"github.com/coinbase/rosetta-sdk-go/types"
+	"github.com/spf13/cobra"
+)
+
+const (
+	// fileMode 0600 indicates that the user/owner can read and write
+	// but can't execute.
+	fileMode = 0600
+)
+
+var (
+	createConfigurationCmd = &cobra.Command{
+		Use:   "create:configuration",
+		Short: "Generate a static configuration file for the Asserter",
+		Long: `In production deployments, it is useful to initialize the response
+Asserter (https://github.com/coinbase/rosetta-sdk-go/tree/master/asserter) using
+a static configuration instead of intializing a configuration dynamically
+from the node. This allows a client to error on new types/statuses that may
+have been added in an update instead of silently erroring.
+
+To use this command, simply provide an absolute path as the argument for where
+the configuration file should be saved (in JSON). Populate the optional
+--server-url flag with the url of the server to generate the configuration
+from.`,
+		Run:  runCreateConfigurationCmd,
+		Args: cobra.ExactArgs(1),
+	}
+)
+
+func runCreateConfigurationCmd(cmd *cobra.Command, args []string) {
+	ctx := context.Background()
+
+	// Create a new fetcher
+	newFetcher := fetcher.New(
+		ServerURL,
+	)
+
+	// Initialize the fetcher's asserter
+	_, _, err := newFetcher.InitializeAsserter(ctx)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	configuration, err := newFetcher.Asserter.ClientConfiguration()
+	if err != nil {
+		log.Fatal(fmt.Errorf("%w: unable to generate spec", err))
+	}
+
+	specString := types.PrettyPrintStruct(configuration)
+	log.Printf("Spec File: %s\n", specString)
+
+	if err := ioutil.WriteFile(path.Clean(args[0]), []byte(specString), os.FileMode(fileMode)); err != nil {
+		log.Fatal(err)
+	}
+}
diff --git a/cmd/root.go b/cmd/root.go
index 6ea9915..b7c8836 100644
--- a/cmd/root.go
+++ b/cmd/root.go
@@ -15,84 +15,18 @@
 package cmd
 
 import (
-	"encoding/json"
-	"io/ioutil"
-	"log"
-	"path"
-	"time"
-
-	"github.com/coinbase/rosetta-cli/internal/reconciler"
-
 	"github.com/spf13/cobra"
 )
 
-const (
-	// ExtendedRetryElapsedTime is used to override the default fetcher
-	// retry elapsed time. In practice, extending the retry elapsed time
-	// has prevented retry exhaustion errors when many goroutines are
-	// used to fetch data from the Rosetta server.
-	//
-	// TODO: make configurable
-	ExtendedRetryElapsedTime = 5 * time.Minute
-)
-
 var (
 	rootCmd = &cobra.Command{
 		Use:   "rosetta-cli",
 		Short: "CLI for the Rosetta API",
 	}
 
-	// DataDir is a folder used to store logs
-	// and any data used to perform validation.
-	DataDir string
-
 	// ServerURL is the base URL for a Rosetta
 	// server to validate.
 	ServerURL string
-
-	// StartIndex is the block index to start syncing.
-	StartIndex int64
-
-	// EndIndex is the block index to stop syncing.
-	EndIndex int64
-
-	// BlockConcurrency is the concurrency to use
-	// while fetching blocks.
-	BlockConcurrency uint64
-
-	// TransactionConcurrency is the concurrency to use
-	// while fetching transactions (if required).
-	TransactionConcurrency uint64
-
-	// AccountConcurrency is the concurrency to use
-	// while fetching accounts during reconciliation.
-	AccountConcurrency uint64
-
-	// LogBlocks determines if blocks are
-	// logged.
-	LogBlocks bool
-
-	// LogTransactions determines if transactions are
-	// logged.
-	LogTransactions bool
-
-	// LogBalanceChanges determines if balance changes are
-	// logged.
-	LogBalanceChanges bool
-
-	// LogReconciliations determines if reconciliations are
-	// logged.
-	LogReconciliations bool
-
-	// HaltOnReconciliationError determines if processing
-	// should stop when encountering a reconciliation error.
-	// It can be beneficial to collect all reconciliation errors
-	// during development.
-	HaltOnReconciliationError bool
-
-	// ExemptFile is an absolute path to a file listing all accounts
-	// to exempt from balance tracking and reconciliation.
-	ExemptFile string
 )
 
 // Execute handles all invocations of the
@@ -102,114 +36,15 @@ func Execute() error {
 }
 
 func init() {
-	rootCmd.PersistentFlags().StringVar(
-		&DataDir,
-		"data-dir",
-		"./validator-data",
-		"folder used to store logs and any data used to perform validation",
-	)
 	rootCmd.PersistentFlags().StringVar(
 		&ServerURL,
 		"server-url",
 		"http://localhost:8080",
-		"base URL for a Rosetta server to validate",
-	)
-	rootCmd.PersistentFlags().Int64Var(
-		&StartIndex,
-		"start",
-		-1,
-		"block index to start syncing",
-	)
-	rootCmd.PersistentFlags().Int64Var(
-		&EndIndex,
-		"end",
-		-1,
-		"block index to stop syncing",
+		"base URL for a Rosetta server",
 	)
-	rootCmd.PersistentFlags().Uint64Var(
-		&BlockConcurrency,
-		"block-concurrency",
-		8,
-		"concurrency to use while fetching blocks",
-	)
-	rootCmd.PersistentFlags().Uint64Var(
-		&TransactionConcurrency,
-		"transaction-concurrency",
-		16,
-		"concurrency to use while fetching transactions (if required)",
-	)
-	rootCmd.PersistentFlags().Uint64Var(
-		&AccountConcurrency,
-		"account-concurrency",
-		8,
-		"concurrency to use while fetching accounts during reconciliation",
-	)
-	rootCmd.PersistentFlags().BoolVar(
-		&LogBlocks,
-		"log-blocks",
-		false,
-		"log processed blocks",
-	)
-	rootCmd.PersistentFlags().BoolVar(
-		&LogTransactions,
-		"log-transactions",
-		false,
-		"log processed transactions",
-	)
-	rootCmd.PersistentFlags().BoolVar(
-		&LogBalanceChanges,
-		"log-balance-changes",
-		false,
-		"log balance changes",
-	)
-	rootCmd.PersistentFlags().BoolVar(
-		&LogReconciliations,
-		"log-reconciliations",
-		false,
-		"log balance reconciliations",
-	)
-	rootCmd.PersistentFlags().BoolVar(
-		&HaltOnReconciliationError,
-		"halt-on-reconciliation-error",
-		true,
-		`Determines if block processing should halt on a reconciliation
-error. It can be beneficial to collect all reconciliation errors or silence
-reconciliation errors during development.`,
-	)
-	rootCmd.PersistentFlags().StringVar(
-		&ExemptFile,
-		"exempt-accounts",
-		"",
-		`Absolute path to a file listing all accounts to exempt from balance
-tracking and reconciliation. Look at the examples directory for an example of
-how to structure this file.`,
-	)
-
-	rootCmd.AddCommand(checkCompleteCmd)
-	rootCmd.AddCommand(checkQuickCmd)
-	rootCmd.AddCommand(checkAccountCmd)
-}
-
-func loadAccounts(filePath string) ([]*reconciler.AccountCurrency, error) {
-	if len(filePath) == 0 {
-		return []*reconciler.AccountCurrency{}, nil
-	}
-
-	accountsRaw, err := ioutil.ReadFile(path.Clean(filePath))
-	if err != nil {
-		return nil, err
-	}
-
-	accounts := []*reconciler.AccountCurrency{}
-	if err := json.Unmarshal(accountsRaw, &accounts); err != nil {
-		return nil, err
-	}
-
-	prettyAccounts, err := json.MarshalIndent(accounts, "", " ")
-	if err != nil {
-		return nil, err
-	}
-	log.Printf("Found %d accounts at %s: %s\n", len(accounts), filePath, string(prettyAccounts))
 
-	return accounts, nil
+	rootCmd.AddCommand(checkCmd)
+	rootCmd.AddCommand(viewBlockCmd)
+	rootCmd.AddCommand(viewAccountCmd)
+	rootCmd.AddCommand(createConfigurationCmd)
 }
diff --git a/cmd/view_account.go b/cmd/view_account.go
new file mode 100644
index 0000000..994d413
--- /dev/null
+++ b/cmd/view_account.go
@@ -0,0 +1,97 @@
+// Copyright 2020 Coinbase, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package cmd
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log"
+	"strconv"
+
+	"github.com/coinbase/rosetta-sdk-go/asserter"
+	"github.com/coinbase/rosetta-sdk-go/fetcher"
+	"github.com/coinbase/rosetta-sdk-go/types"
+	"github.com/spf13/cobra"
+)
+
+var (
+	viewAccountCmd = &cobra.Command{
+		Use:   "view:account",
+		Short: "View an account balance",
+		Long: `While debugging, it is often useful to inspect the state
+of an account at a certain block. This command allows you to look up
+any account by providing a JSON representation of a types.AccountIdentifier
+(and optionally a height to perform the query).
+
+For example, you could run view:account '{"address":"interesting address"}' 1000
+to lookup the balance of an interesting address at block 1000. Allowing the
+address to specified as JSON allows for querying by SubAccountIdentifier.`,
+		Run:  runViewAccountCmd,
+		Args: cobra.MinimumNArgs(1),
+	}
+)
+
+func runViewAccountCmd(cmd *cobra.Command, args []string) {
+	ctx := context.Background()
+
+	account := &types.AccountIdentifier{}
+	if err := json.Unmarshal([]byte(args[0]), account); err != nil {
+		log.Fatal(fmt.Errorf("%w: unable to unmarshal account %s", err, args[0]))
+	}
+
+	if err := asserter.AccountIdentifier(account); err != nil {
+		log.Fatal(fmt.Errorf("%w: invalid account identifier %+v", err, account))
+	}
+
+	// Create a new fetcher
+	newFetcher := fetcher.New(
+		ServerURL,
+	)
+
+	// Initialize the fetcher's asserter
+	primaryNetwork, _, err := newFetcher.InitializeAsserter(ctx)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	// Print the primary network and network status
+	// TODO: support specifying which network to get block from
+	log.Printf("Primary Network: %s\n", types.PrettyPrintStruct(primaryNetwork))
+
+	var lookupBlock *types.PartialBlockIdentifier
+	if len(args) > 1 {
+		index, err := strconv.ParseInt(args[1], 10, 64)
+		if err != nil {
+			log.Fatal(fmt.Errorf("%w: unable to parse index %s", err, args[0]))
+		}
+
+		lookupBlock = &types.PartialBlockIdentifier{Index: &index}
+	}
+
+	block, amounts, metadata, err := newFetcher.AccountBalanceRetry(
+		ctx,
+		primaryNetwork,
+		account,
+		lookupBlock,
+	)
+	if err != nil {
+		log.Fatal(fmt.Errorf("%w: unable to fetch account %+v", err, account))
+	}
+
+	log.Printf("Amounts: %s\n", types.PrettyPrintStruct(amounts))
+	log.Printf("Metadata: %s\n", types.PrettyPrintStruct(metadata))
+	log.Printf("Balance Fetched At: %s\n", types.PrettyPrintStruct(block))
+}
diff --git a/cmd/view_block.go b/cmd/view_block.go
new file mode 100644
index 0000000..18bcfae
--- /dev/null
+++ b/cmd/view_block.go
@@ -0,0 +1,93 @@
+// Copyright 2020 Coinbase, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package cmd
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"strconv"
+
+	"github.com/coinbase/rosetta-sdk-go/fetcher"
+	"github.com/coinbase/rosetta-sdk-go/types"
+	"github.com/spf13/cobra"
+)
+
+var (
+	viewBlockCmd = &cobra.Command{
+		Use:   "view:block",
+		Short: "View a block",
+		Long: `While debugging a Node API implementation, it can be very
+useful to inspect block contents. This command allows you to fetch any
+block by index to inspect its contents. It uses the
+fetcher (https://github.com/coinbase/rosetta-sdk-go/tree/master/fetcher) package
+to automatically get all transactions in the block and assert the format
+of the block is correct before printing.
+
+If this command errors, it is likely because the block you are trying to
+fetch is formatted incorrectly.`,
+		Run:  runViewBlockCmd,
+		Args: cobra.ExactArgs(1),
+	}
+)
+
+func runViewBlockCmd(cmd *cobra.Command, args []string) {
+	ctx := context.Background()
+	index, err := strconv.ParseInt(args[0], 10, 64)
+	if err != nil {
+		log.Fatal(fmt.Errorf("%w: unable to parse index %s", err, args[0]))
+	}
+
+	// Create a new fetcher
+	newFetcher := fetcher.New(
+		ServerURL,
+	)
+
+	// Initialize the fetcher's asserter
+	//
+	// Behind the scenes this makes a call to get the
+	// network status and uses the response to inform
+	// the asserter what are valid responses.
+	primaryNetwork, _, err := newFetcher.InitializeAsserter(ctx)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	// Print the primary network and network status
+	// TODO: support specifying which network to get block from
+	log.Printf("Primary Network: %s\n", types.PrettyPrintStruct(primaryNetwork))
+
+	// Fetch the specified block with retries (automatically
+	// asserted for correctness)
+	//
+	// On another note, notice that fetcher.BlockRetry
+	// automatically fetches all transactions that are
+	// returned in BlockResponse.OtherTransactions. If you use
+	// the client directly, you will need to implement a mechanism
+	// to fully populate the block by fetching all these
+	// transactions.
+	block, err := newFetcher.BlockRetry(
+		ctx,
+		primaryNetwork,
+		&types.PartialBlockIdentifier{
+			Index: &index,
+		},
+	)
+	if err != nil {
+		log.Fatal(fmt.Errorf("%w: unable to fetch block", err))
+	}
+
+	log.Printf("Current Block: %s\n", types.PrettyPrintStruct(block))
+}
diff --git a/go.mod b/go.mod
index 5ff33a9..46b326b 100644
--- a/go.mod
+++ b/go.mod
@@ -3,10 +3,11 @@ module github.com/coinbase/rosetta-cli
 go 1.13
 
 require (
-	github.com/coinbase/rosetta-sdk-go v0.1.6
-	github.com/davecgh/go-spew v1.1.1
+	github.com/coinbase/rosetta-sdk-go v0.1.8
 	github.com/dgraph-io/badger v1.6.0
-	github.com/pkg/errors v0.8.1
+	github.com/google/addlicense v0.0.0-20200422172452-68a83edd47bc // indirect
+	github.com/mattn/goveralls v0.0.5 // indirect
+	github.com/segmentio/golines v0.0.0-20200306054842-869934f8da7b // indirect
 	github.com/spf13/cobra v0.0.5
 	github.com/stretchr/testify v1.5.1
 	golang.org/x/sync v0.0.0-20200317015054-43a5402ce75a
diff --git a/go.sum b/go.sum
index 09f8872..afacd67 100644
--- a/go.sum
+++ b/go.sum
@@ -1,19 +1,25 @@
 github.com/AndreasBriese/bbloom v0.0.0-20190306092124-e2d15f34fcf9 h1:HD8gA2tkByhMAwYaFAX9w2l7vxvBQ5NMoxDrkhqhtn4=
 github.com/AndreasBriese/bbloom v0.0.0-20190306092124-e2d15f34fcf9/go.mod h1:bOvUY6CB00SOBii9/FifXqc0awNKxLFCL/+pkDPuyl8=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
+github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751 h1:JYp7IbQjafoB+tBA3gMyHYHrpOtNuDiK/uB5uXxq5wM=
+github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
+github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d h1:UQZhZ2O0vMHr2cI+DC1Mbh0TJxzA3RcLoMsFw+aXw7E=
+github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk541a8SKzHPHnH3zbiI+7dagKZ0cgpgrD7Fyho=
 github.com/armon/consul-api v0.0.0-20180202201655-eb2c6b5be1b6/go.mod h1:grANhF5doyWs3UAsr3K4I6qtAmlQcZDesFNEHPZAzj8=
 github.com/cenkalti/backoff v2.2.1+incompatible h1:tNowT99t7UNflLxfYYSlKYsBpXdEet03Pg2g16Swow4=
 github.com/cenkalti/backoff v2.2.1+incompatible/go.mod h1:90ReRw6GdpyfrHakVjL/QHaoyV4aDUVVkXQJJJ3NXXM=
-github.com/coinbase/rosetta-sdk-go v0.1.5 h1:fkYLDs8f7RuwKDJiaZv4qtVRntCOAcSm6VeWiHH9L8Y=
-github.com/coinbase/rosetta-sdk-go v0.1.5/go.mod h1:lbmGsBpBiSZWP4WQqEW2CuTweGcU/ioQHwZ8CYo6yO8=
-github.com/coinbase/rosetta-sdk-go v0.1.6 h1:l6vyt+Gad7TWIy2Tf7giXO8jniBcLPhhW6021QFgIq0=
-github.com/coinbase/rosetta-sdk-go v0.1.6/go.mod h1:lbmGsBpBiSZWP4WQqEW2CuTweGcU/ioQHwZ8CYo6yO8=
-github.com/coinbase/rosetta-validator v0.1.2 h1:d1i/XuZu3tPEnc/HKLzixK0yXbMyg5UEzS2qba7tv5I=
-github.com/coinbase/rosetta-validator v0.1.2/go.mod h1:SgJPFHi1Ikr+tC5MkUhkrnLQKLT0Qzf7g+s1rnZ3kGo=
+github.com/coinbase/rosetta-sdk-go v0.1.8 h1:YJf5Us7Pl4G8CRBZ5+LEUIV4k9ZI/18nFloPaHBySFk=
+github.com/coinbase/rosetta-sdk-go v0.1.8/go.mod h1:y1wXRc1wod4fEp3jhW+9D6MSFtnm/9X5yNxJv92Cv2E=
 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=
 github.com/cpuguy83/go-md2man v1.0.10/go.mod h1:SmD6nW6nTyfqj6ABTjUi3V3JVMnlJmwcJI5acqYI6dE=
+github.com/dave/dst v0.23.1 h1:2obX6c3RqALrEOp6u01qsqPvwp0t+RpOp9O4Bf9KhXs=
+github.com/dave/dst v0.23.1/go.mod h1:LjPcLEauK4jC5hQ1fE/wr05O41zK91Pr4Qs22Ljq7gs=
+github.com/dave/gopackages v0.0.0-20170318123100-46e7023ec56e/go.mod h1:i00+b/gKdIDIxuLDFob7ustLAVqhsZRk2qVZrArELGQ=
+github.com/dave/jennifer v1.2.0/go.mod h1:fIb+770HOpJ2fmN9EPPKOqm1vMGhB+TwXKMZhrIygKg=
+github.com/dave/kerr v0.0.0-20170318121727-bc25dd6abe8e/go.mod h1:qZqlPyPvfsDJt+3wHJ1EvSXDuVjFTK0j2p/ca+gtsb8=
+github.com/dave/rebecca v0.9.1/go.mod h1:N6XYdMD/OKw3lkF3ywh8Z6wPGuwNFDNtWYEMFWEmXBA=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -24,21 +30,48 @@ github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2/go.mod h1:SqUrOPUn
 github.com/dustin/go-humanize v1.0.0 h1:VSnTsYCnlFHaM2/igO1h6X3HA71jcobQuxemgkq4zYo=
 github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
+github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/golang/protobuf v1.3.1 h1:YF8+flBXS5eO826T4nzqPrxfhQThhXl0YzfuUPu4SBg=
 github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
+github.com/google/addlicense v0.0.0-20200422172452-68a83edd47bc h1:CHWlqgYPu3FMUOyAno2lVDyI9wmexZEuV6/nDvsvETc=
+github.com/google/addlicense v0.0.0-20200422172452-68a83edd47bc/go.mod h1:EMjYTRimagHs1FwlIqKyX3wAM0u3rA+McvlIIWmSamA=
+github.com/google/pprof v0.0.0-20181127221834-b4f47329b966/go.mod h1:zfwlbNMJ+OItoe0UupaVj+oy1omPYYDuagoSzA8v9mc=
 github.com/gorilla/mux v1.7.4/go.mod h1:DVbg23sWSpFRCP0SfiEN6jmj59UnW/n46BH5rLB71So=
 github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ=
+github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
+github.com/ianlancetaylor/demangle v0.0.0-20181102032728-5e5cf60278f6/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
 github.com/inconshreveable/mousetrap v1.0.0 h1:Z8tu5sraLXCXIcARxBp/8cbvlwVa7Z1NHg9XEKhtSvM=
 github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
+github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
+github.com/konsorten/go-windows-terminal-sequences v1.0.2/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
+github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
+github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
+github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/magiconair/properties v1.8.0/go.mod h1:PppfXfuXeibc/6YijjN8zIbojt8czPbwD3XqdrwzmxQ=
+github.com/mattn/go-colorable v0.1.4/go.mod h1:U0ppj6V5qS13XJ6of8GYAs25YV2eR4EVcfRqFIhoBtE=
+github.com/mattn/go-isatty v0.0.8/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hdxcsrc5s=
+github.com/mattn/go-isatty v0.0.10/go.mod h1:qgIWMr58cqv1PHHyhnkY9lrL7etaEgOFcMEpPG5Rm84=
+github.com/mattn/goveralls v0.0.5 h1:spfq8AyZ0cCk57Za6/juJ5btQxeE1FaEGMdfcI+XO48=
+github.com/mattn/goveralls v0.0.5/go.mod h1:Xg2LHi51faXLyKXwsndxiW6uxEEQT9+3sjGzzwU4xy0=
+github.com/mgutz/ansi v0.0.0-20170206155736-9520e82c474b/go.mod h1:01TrycV0kFyexm33Z7vhZRXopbI8J3TDReVlkTgMUxE=
 github.com/mitchellh/go-homedir v1.1.0/go.mod h1:SfyaCUpYCn1Vlf4IUYiD9fPX4A5wJrkLzIz1N1q0pr0=
 github.com/mitchellh/mapstructure v1.1.2/go.mod h1:FVVH3fgwuzCH5S8UJGiWEs2h04kUh9fWfEaFds41c1Y=
+github.com/mitchellh/mapstructure v1.3.0 h1:iDwIio/3gk2QtLLEsqU5lInaMzos0hDTz8a6lazSFVw=
+github.com/mitchellh/mapstructure v1.3.0/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RRV2QTWOzhPopBRo=
+github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
+github.com/onsi/ginkgo v1.10.2/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
+github.com/onsi/gomega v1.7.0/go.mod h1:ex+gbHU/CVuBBDIJjb2X0qEXbFg53c61hWP/1CpauHY=
 github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=
 github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/russross/blackfriday v1.5.2/go.mod h1:JO/DiYxRf+HjHt06OyowR9PTA263kcR/rfWxYHBV53g=
+github.com/segmentio/golines v0.0.0-20200306054842-869934f8da7b h1:Jk5Swz/AfwWl5yc/yquzUmNesPF2aTFuafpjikzczRg=
+github.com/segmentio/golines v0.0.0-20200306054842-869934f8da7b/go.mod h1:K7zjgP8yJ/U8nb8nxaSykalAKSvbqr6TNbd9B7zzBFU=
+github.com/sergi/go-diff v1.0.0/go.mod h1:0CfEIISq7TuYL3j771MWULgwwjU+GofnZX9QAmXWZgo=
+github.com/sirupsen/logrus v1.4.2 h1:SPIRibHv4MatM3XXNO2BJeFLZwZ2LvZgfQ5+UNI2im4=
+github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE=
 github.com/spf13/afero v1.1.2/go.mod h1:j4pytiNVoe2o6bmDsKpLACNPDBIoEAkihy7loJ1B0CQ=
 github.com/spf13/cast v1.3.0/go.mod h1:Qx5cxh0v+4UWYiBimWS+eyWzqEqokIECu5etghLkUJE=
 github.com/spf13/cobra v0.0.5 h1:f0B+LkLX6DtmRH1isoNA9VTtNUK9K8xYd28JNNfOv/s=
@@ -47,25 +80,62 @@ github.com/spf13/jwalterweatherman v1.0.0/go.mod h1:cQK4TGJAtQXfYWX+Ddv3mKDzgVb6
 github.com/spf13/pflag v1.0.3 h1:zPAT6CGy6wXeQ7NtTnaTerfKOsV6V6F8agHXFiazDkg=
 github.com/spf13/pflag v1.0.3/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnInEg4=
 github.com/spf13/viper v1.3.2/go.mod h1:ZiWeW+zYFKm7srdB9IoDzzZXaJaI5eL9QjNiN/DMA2s=
+github.com/stretchr/objx v0.1.0 h1:4G4v2dO3VZwixGIRoQ5Lfboy6nUhCyYzaqnIAPPhYs4=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
 github.com/stretchr/testify v1.5.1 h1:nOGnQDM7FYENwehXlg/kFVnos3rEvtKTjRvOWSzb6H4=
 github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
 github.com/ugorji/go/codec v0.0.0-20181204163529-d75b2dcb6bc8/go.mod h1:VFNgLljTbGfSG7qAOspJ7OScBnGdDN/yBr0sguwnwf0=
+github.com/x-cray/logrus-prefixed-formatter v0.5.2 h1:00txxvfBM9muc0jiLIEAkAcIMJzfthRT6usrui8uGmg=
+github.com/x-cray/logrus-prefixed-formatter v0.5.2/go.mod h1:2duySbKsL6M18s5GU7VPsoEPHyzalCE06qoARUCeBBE=
 github.com/xordataexchange/crypt v0.0.3-0.20170626215501-b2862e3d0a77/go.mod h1:aYKd//L2LvnjZzWKhF00oedf4jCCReLcmhLdhm1A27Q=
+golang.org/x/arch v0.0.0-20180920145803-b19384d3c130/go.mod h1:cYlCBUl1MsqxdiKgmc4uh7TxZfWSFLOGSRR090WDxt8=
+golang.org/x/crypto v0.0.0-20181127143415-eb0de9b17e85/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20181203042331-505ab145d0a9/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
+golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg=
+golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
+golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859 h1:R/3boaszxrf1GEUWTVDzSKVwLmSJpwZ1yqXm8j0v2QI=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
+golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20200317015054-43a5402ce75a h1:WXEvlFVvvGxCJLG6REjsT03iWnKLEWinaScsxF2Vm2o=
 golang.org/x/sync v0.0.0-20200317015054-43a5402ce75a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sys v0.0.0-20180903190138-2b024373dcd9/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20181205085412-a5c9d58dba9a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20190422165155-953cdadca894/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190626221950-04f50cda93cb h1:fgwFCsaw9buMuxNd6+DQfAuSFqbNiQZpcgJQAgJsK6k=
 golang.org/x/sys v0.0.0-20190626221950-04f50cda93cb/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20191008105621-543471e840be/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20191024172528-b4ff53e7a1cb h1:ZxSglHghKPYD8WDeRUzRJrUJtDF0PxsTUSxyqr9/5BI=
+golang.org/x/sys v0.0.0-20191024172528-b4ff53e7a1cb/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
+golang.org/x/tools v0.0.0-20181127232545-e782529d0ddd/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
+golang.org/x/tools v0.0.0-20191024220359-3d91e92cde03 h1:4gtJXHJ9ud0q8MNSDxJsRU/WH+afypbe4Vk4zq+8qow=
+golang.org/x/tools v0.0.0-20191024220359-3d91e92cde03/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
+golang.org/x/tools v0.0.0-20200113040837-eac381796e91 h1:OOkytthzFBKHY5EfEgLUabprb0LtJVkQtNxAQ02+UE4=
+golang.org/x/tools v0.0.0-20200113040837-eac381796e91/go.mod h1:TB2adYChydJhpapKDTa4BR/hXlZSLoq2Wpct/0txZ28=
+golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+gopkg.in/alecthomas/kingpin.v2 v2.2.6 h1:jMFz6MfLP0/4fUyZle81rXUoxOBFi19VUFKVDOQfozc=
+gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY=
+gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
+gopkg.in/src-d/go-billy.v4 v4.3.0/go.mod h1:tm33zBoOwxjYHZIE+OV8bxTWFMJLrconzFMd38aARFk=
+gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWDmTeBkI65Dw0HsyUHuEVlX15mw=
+gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
 gopkg.in/yaml.v2 v2.2.2 h1:ZCJp+EgiOT7lHqUV2J862kp8Qj64Jo6az82+3Td9dZw=
 gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
diff --git a/internal/logger/logger.go b/internal/logger/logger.go
index fedd36b..2afe02e 100644
--- a/internal/logger/logger.go
+++ b/internal/logger/logger.go
@@ -21,7 +21,7 @@ import (
 	"os"
 	"path"
 
-	"github.com/coinbase/rosetta-cli/internal/storage"
+	"github.com/coinbase/rosetta-cli/internal/reconciler"
 
 	"github.com/coinbase/rosetta-sdk-go/types"
 )
@@ -84,12 +84,11 @@ func NewLogger(
 	}
 }
 
-// BlockStream writes the next processed block to the end of the
+// AddBlockStream writes the next processed block to the end of the
 // blockStreamFile output file.
-func (l *Logger) BlockStream(
+func (l *Logger) AddBlockStream(
 	ctx context.Context,
 	block *types.Block,
-	orphan bool,
 ) error {
 	if !l.logBlocks {
 		return nil
@@ -105,14 +104,9 @@ func (l *Logger) BlockStream(
 	}
 	defer f.Close()
 
-	verb := addEvent
-	if orphan {
-		verb = removeEvent
-	}
-
 	_, err = f.WriteString(fmt.Sprintf(
 		"%s Block %d:%s with Parent Block %d:%s\n",
-		verb,
+		addEvent,
 		block.BlockIdentifier.Index,
 		block.BlockIdentifier.Hash,
 		block.ParentBlockIdentifier.Index,
@@ -122,7 +116,40 @@ func (l *Logger) BlockStream(
 		return err
 	}
 
-	return l.TransactionStream(ctx, block, verb)
+	return l.TransactionStream(ctx, block)
+}
+
+// RemoveBlockStream writes the next processed block to the end of the
+// blockStreamFile output file.
+func (l *Logger) RemoveBlockStream(
+	ctx context.Context,
+	block *types.BlockIdentifier,
+) error {
+	if !l.logBlocks {
+		return nil
+	}
+
+	f, err := os.OpenFile(
+		path.Join(l.logDir, blockStreamFile),
+		os.O_APPEND|os.O_CREATE|os.O_WRONLY,
+		logFilePermissions,
+	)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+
+	_, err = f.WriteString(fmt.Sprintf(
+		"%s Block %d:%s\n",
+		removeEvent,
+		block.Index,
+		block.Hash,
+	))
+	if err != nil {
+		return err
+	}
+
+	return nil
 }
 
 // TransactionStream writes the next processed block's transactions
@@ -130,7 +157,6 @@ func (l *Logger) BlockStream(
 func (l *Logger) TransactionStream(
 	ctx context.Context,
 	block *types.Block,
-	verb string,
 ) error {
 	if !l.logTransactions {
 		return nil
@@ -148,8 +174,7 @@ func (l *Logger) TransactionStream(
 
 	for _, tx := range block.Transactions {
 		_, err = f.WriteString(fmt.Sprintf(
-			"%s Transaction %s at Block %d:%s\n",
-			verb,
+			"Transaction %s at Block %d:%s\n",
 			tx.TransactionIdentifier.Hash,
 			block.BlockIdentifier.Index,
 			block.BlockIdentifier.Hash,
@@ -167,7 +192,7 @@ func (l *Logger) TransactionStream(
 			}
 			participant := ""
 			if op.Account != nil {
-				participant = op.Account.Address
+				participant = types.AccountString(op.Account)
 			}
 
 			networkIndex := op.OperationIdentifier.Index
@@ -188,13 +213,6 @@ func (l *Logger) TransactionStream(
 			if err != nil {
 				return err
 			}
-
-			if op.Account != nil && op.Account.Metadata != nil {
-				_, err = f.WriteString(fmt.Sprintf("Account Metadata: %+v\n", op.Account.Metadata))
-				if err != nil {
-					return err
-				}
-			}
 		}
 	}
 
@@ -205,7 +223,7 @@ func (l *Logger) TransactionStream(
 // to the balanceStreamFile.
 func (l *Logger) BalanceStream(
 	ctx context.Context,
-	balanceChanges []*storage.BalanceChange,
+	balanceChanges []*reconciler.BalanceChange,
 ) error {
 	if !l.logBalanceChanges {
 		return nil
@@ -223,10 +241,10 @@ func (l *Logger) BalanceStream(
 
 	for _, balanceChange := range balanceChanges {
 		balanceLog := fmt.Sprintf(
-			"Account: %s Change: %s%s Block: %d:%s",
+			"Account: %s Change: %s:%s Block: %d:%s",
 			balanceChange.Account.Address,
 			balanceChange.Difference,
-			balanceChange.Currency.Symbol,
+			types.CurrencyString(balanceChange.Currency),
 			balanceChange.Block.Index,
 			balanceChange.Block.Hash,
 		)
@@ -244,7 +262,8 @@ func (l *Logger) ReconcileSuccessStream(
 	ctx context.Context,
 	reconciliationType string,
 	account *types.AccountIdentifier,
-	balance *types.Amount,
+	currency *types.Currency,
+	balance string,
 	block *types.BlockIdentifier,
 ) error {
 	if !l.logReconciliation {
@@ -262,18 +281,18 @@ func (l *Logger) ReconcileSuccessStream(
 	defer f.Close()
 
 	log.Printf(
-		"%s Reconciled %+v at %d\n",
+		"%s Reconciled %s at %d\n",
 		reconciliationType,
-		account,
+		types.AccountString(account),
 		block.Index,
 	)
 
 	_, err = f.WriteString(fmt.Sprintf(
 		"Type:%s Account: %s Currency: %s Balance: %s Block: %d:%s\n",
 		reconciliationType,
-		account.Address,
-		balance.Currency.Symbol,
-		balance.Value,
+		types.AccountString(account),
+		types.CurrencyString(currency),
+		balance,
 		block.Index,
 		block.Hash,
 	))
@@ -291,16 +310,18 @@ func (l *Logger) ReconcileFailureStream(
 	reconciliationType string,
 	account *types.AccountIdentifier,
 	currency *types.Currency,
-	difference string,
+	computedBalance string,
+	nodeBalance string,
 	block *types.BlockIdentifier,
 ) error {
 	// Always print out reconciliation failures
 	log.Printf(
-		"%s Reconciliation failed for %+v at %d by %s (computed-node)\n",
+		"%s Reconciliation failed for %s at %d computed: %s node: %s\n",
 		reconciliationType,
-		account,
+		types.AccountString(account),
 		block.Index,
-		difference,
+		computedBalance,
+		nodeBalance,
 	)
 
 	if !l.logReconciliation {
@@ -318,13 +339,14 @@ func (l *Logger) ReconcileFailureStream(
 	defer f.Close()
 
 	_, err = f.WriteString(fmt.Sprintf(
-		"Type:%s Account: %+v Currency: %+v Block: %s:%d Difference(computed-node):%s\n",
+		"Type:%s Account: %s Currency: %s Block: %s:%d computed: %s node: %s\n",
 		reconciliationType,
-		account,
-		currency,
+		types.AccountString(account),
+		types.CurrencyString(currency),
 		block.Hash,
 		block.Index,
-		difference,
+		computedBalance,
+		nodeBalance,
 	))
 	if err != nil {
 		return err
diff --git a/internal/processor/reconciler_handler.go b/internal/processor/reconciler_handler.go
new file mode 100644
index 0000000..70e13eb
--- /dev/null
+++ b/internal/processor/reconciler_handler.go
@@ -0,0 +1,100 @@
+// Copyright 2020 Coinbase, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package processor
+
+import (
+	"context"
+	"errors"
+
+	"github.com/coinbase/rosetta-cli/internal/logger"
+
+	"github.com/coinbase/rosetta-sdk-go/types"
+)
+
+// ReconcilerHandler implements the Reconciler.Handler interface.
+type ReconcilerHandler struct {
+	cancel                    context.CancelFunc
+	logger                    *logger.Logger
+	haltOnReconciliationError bool
+}
+
+// NewReconcilerHandler creates a new ReconcilerHandler.
+func NewReconcilerHandler(
+	cancel context.CancelFunc,
+	logger *logger.Logger,
+	haltOnReconciliationError bool,
+) *ReconcilerHandler {
+	return &ReconcilerHandler{
+		cancel:                    cancel,
+		logger:                    logger,
+		haltOnReconciliationError: haltOnReconciliationError,
+	}
+}
+
+// ReconciliationFailed is called each time a reconciliation fails.
+// In this Handler implementation, we halt if haltOnReconciliationError
+// was set to true. We also cancel the context.
+func (h *ReconcilerHandler) ReconciliationFailed(
+	ctx context.Context,
+	reconciliationType string,
+	account *types.AccountIdentifier,
+	currency *types.Currency,
+	computedBalance string,
+	nodeBalance string,
+	block *types.BlockIdentifier,
+) error {
+	err := h.logger.ReconcileFailureStream(
+		ctx,
+		reconciliationType,
+		account,
+		currency,
+		computedBalance,
+		nodeBalance,
+		block,
+	)
+	if err != nil {
+		return err
+	}
+
+	if h.haltOnReconciliationError {
+		h.cancel()
+		return errors.New("halting due to reconciliation error")
+	}
+
+	// TODO: automatically find block with missing operation
+	// if inactive reconciliation error. Can do this by asserting
+	// the impacted address has a balance change of 0 on all blocks
+	// it is not active.
+	return nil
+}
+
+// ReconciliationSucceeded is called each time a reconciliation succeeds.
+func (h *ReconcilerHandler) ReconciliationSucceeded(
+	ctx context.Context,
+	reconciliationType string,
+	account *types.AccountIdentifier,
+	currency *types.Currency,
+	balance string,
+	block *types.BlockIdentifier,
+) error {
+	return h.logger.ReconcileSuccessStream(
+		ctx,
+		reconciliationType,
+		account,
+		currency,
+		balance,
+		block,
+	)
+}
diff --git a/internal/processor/reconciler_helper.go b/internal/processor/reconciler_helper.go
new file mode 100644
index 0000000..d9dfbb4
--- /dev/null
+++ b/internal/processor/reconciler_helper.go
@@ -0,0 +1,80 @@
+// Copyright 2020 Coinbase, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package processor
+
+import (
+	"context"
+	"errors"
+
+	"github.com/coinbase/rosetta-cli/internal/storage"
+
+	"github.com/coinbase/rosetta-sdk-go/types"
+)
+
+// ReconcilerHelper implements the Reconciler.Helper
+// interface.
+type ReconcilerHelper struct {
+	storage *storage.BlockStorage
+}
+
+// NewReconcilerHelper returns a new ReconcilerHelper.
+func NewReconcilerHelper(
+	storage *storage.BlockStorage,
+) *ReconcilerHelper {
+	return &ReconcilerHelper{
+		storage: storage,
+	}
+}
+
+// BlockExists returns a boolean indicating if block_storage
+// contains a block. This is necessary to reconcile across
+// reorgs. If the block returned on an account balance fetch
+// does not exist, reconciliation will be skipped.
+func (h *ReconcilerHelper) BlockExists(
+	ctx context.Context,
+	block *types.BlockIdentifier,
+) (bool, error) {
+	_, err := h.storage.GetBlock(ctx, block)
+	if err == nil {
+		return true, nil
+	}
+
+	if errors.Is(err, storage.ErrBlockNotFound) {
+		return false, nil
+	}
+
+	return false, err
+}
+
+// CurrentBlock returns the last processed block and is used
+// to determine which block to check account balances at during
+// inactive reconciliation.
+func (h *ReconcilerHelper) CurrentBlock(
+	ctx context.Context,
+) (*types.BlockIdentifier, error) {
+	return h.storage.GetHeadBlockIdentifier(ctx)
+}
+
+// AccountBalance returns the balance of an account in block storage.
+// It is necessary to perform this check outside of the Reconciler
+// package to allow for separation from a default storage backend.
+func (h *ReconcilerHelper) AccountBalance(
+	ctx context.Context,
+	account *types.AccountIdentifier,
+	currency *types.Currency,
+	headBlock *types.BlockIdentifier,
+) (*types.Amount, *types.BlockIdentifier, error) {
+	return h.storage.GetBalance(ctx, account, currency, headBlock)
+}
diff --git a/internal/processor/storage_helper.go b/internal/processor/storage_helper.go
new file mode 100644
index 0000000..ce3d259
--- /dev/null
+++ b/internal/processor/storage_helper.go
@@ -0,0 +1,138 @@
+// Copyright 2020 Coinbase, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package processor
+
+import (
+	"context"
+	"fmt"
+	"log"
+
+	"github.com/coinbase/rosetta-cli/internal/reconciler"
+
+	"github.com/coinbase/rosetta-sdk-go/fetcher"
+	"github.com/coinbase/rosetta-sdk-go/types"
+)
+
+// BlockStorageHelper implements the storage.Helper
+// interface.
+type BlockStorageHelper struct {
+	network *types.NetworkIdentifier
+	fetcher *fetcher.Fetcher
+
+	// Configuration settings
+	lookupBalanceByBlock bool
+	exemptAccounts       []*reconciler.AccountCurrency
+}
+
+// NewBlockStorageHelper returns a new BlockStorageHelper.
+func NewBlockStorageHelper(
+	network *types.NetworkIdentifier,
+	fetcher *fetcher.Fetcher,
+	lookupBalanceByBlock bool,
+	exemptAccounts []*reconciler.AccountCurrency,
+) *BlockStorageHelper {
+	return &BlockStorageHelper{
+		network:              network,
+		fetcher:              fetcher,
+		lookupBalanceByBlock: lookupBalanceByBlock,
+		exemptAccounts:       exemptAccounts,
+	}
+}
+
+// AccountBalance attempts to fetch the balance
+// for a missing account in storage. This is necessary
+// for running the "check" command at an arbitrary height
+// instead of syncing from genesis.
+func (h *BlockStorageHelper) AccountBalance(
+	ctx context.Context,
+	account *types.AccountIdentifier,
+	currency *types.Currency,
+	block *types.BlockIdentifier,
+) (*types.Amount, error) {
+	if !h.lookupBalanceByBlock {
+		return &types.Amount{
+			Value:    "0",
+			Currency: currency,
+		}, nil
+	}
+
+	// In the case that we are syncing from arbitrary height,
+	// we may need to recover the balance of an account to
+	// perform validations.
+	_, value, err := reconciler.GetCurrencyBalance(
+		ctx,
+		h.fetcher,
+		h.network,
+		account,
+		currency,
+		types.ConstructPartialBlockIdentifier(block),
+	)
+	if err != nil {
+		return nil, fmt.Errorf("%w: unable to get currency balance in storage helper", err)
+	}
+
+	return &types.Amount{
+		Value:    value,
+		Currency: currency,
+	}, nil
+}
+
+// SkipOperation returns a boolean indicating whether
+// an operation should be processed. An operation will
+// not be processed if it is considered unsuccessful
+// or affects an exempt account.
+func (h *BlockStorageHelper) SkipOperation(
+	ctx context.Context,
+	op *types.Operation,
+) (bool, error) {
+	successful, err := h.fetcher.Asserter.OperationSuccessful(op)
+	if err != nil {
+		// Should only occur if responses not validated
+		return false, err
+	}
+
+	if !successful {
+		return true, nil
+	}
+
+	if op.Account == nil {
+		return true, nil
+	}
+
+	// Exempting account in BalanceChanges ensures that storage is not updated
+	// and that the account is not reconciled.
+	if h.accountExempt(op.Account, op.Amount.Currency) {
+		log.Printf("Skipping exempt account %+v\n", op.Account)
+		return true, nil
+	}
+
+	return false, nil
+}
+
+// accountExempt returns a boolean indicating if the provided
+// account and currency are exempt from balance tracking and
+// reconciliation.
+func (h *BlockStorageHelper) accountExempt(
+	account *types.AccountIdentifier,
+	currency *types.Currency,
+) bool {
+	return reconciler.ContainsAccountCurrency(
+		h.exemptAccounts,
+		&reconciler.AccountCurrency{
+			Account:  account,
+			Currency: currency,
+		},
+	)
+}
diff --git a/internal/processor/syncer_handler.go b/internal/processor/syncer_handler.go
new file mode 100644
index 0000000..d18fd77
--- /dev/null
+++ b/internal/processor/syncer_handler.go
@@ -0,0 +1,104 @@
+// Copyright 2020 Coinbase, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package processor
+
+import (
+	"context"
+	"log"
+
+	"github.com/coinbase/rosetta-cli/internal/logger"
+	"github.com/coinbase/rosetta-cli/internal/reconciler"
+	"github.com/coinbase/rosetta-cli/internal/storage"
+
+	"github.com/coinbase/rosetta-sdk-go/fetcher"
+	"github.com/coinbase/rosetta-sdk-go/types"
+)
+
+// SyncerHandler implements the syncer.Handler interface.
+type SyncerHandler struct {
+	storage    *storage.BlockStorage
+	logger     *logger.Logger
+	reconciler *reconciler.Reconciler
+	fetcher    *fetcher.Fetcher
+}
+
+// NewSyncerHandler returns a new SyncerHandler.
+func NewSyncerHandler(
+	storage *storage.BlockStorage,
+	logger *logger.Logger,
+	reconciler *reconciler.Reconciler,
+	fetcher *fetcher.Fetcher,
+) *SyncerHandler {
+	return &SyncerHandler{
+		storage:    storage,
+		logger:     logger,
+		reconciler: reconciler,
+		fetcher:    fetcher,
+	}
+}
+
+// BlockAdded is called by the syncer after a
+// block is added.
+func (h *SyncerHandler) BlockAdded(
+	ctx context.Context,
+	block *types.Block,
+) error {
+	log.Printf("Adding block %+v\n", block.BlockIdentifier)
+
+	// Log processed blocks and balance changes
+	if err := h.logger.AddBlockStream(ctx, block); err != nil {
+		return nil
+	}
+
+	balanceChanges, err := h.storage.StoreBlock(ctx, block)
+	if err != nil {
+		return err
+	}
+
+	if err := h.logger.BalanceStream(ctx, balanceChanges); err != nil {
+		return nil
+	}
+
+	// Mark accounts for reconciliation...this may be
+	// blocking
+	return h.reconciler.QueueChanges(ctx, block.BlockIdentifier, balanceChanges)
+}
+
+// BlockRemoved is called by the syncer after a
+// block is removed.
+func (h *SyncerHandler) BlockRemoved(
+	ctx context.Context,
+	blockIdentifier *types.BlockIdentifier,
+) error {
+	log.Printf("Orphaning block %+v\n", blockIdentifier)
+
+	// Log processed blocks and balance changes
+	if err := h.logger.RemoveBlockStream(ctx, blockIdentifier); err != nil {
+		return nil
+	}
+
+	balanceChanges, err := h.storage.RemoveBlock(ctx, blockIdentifier)
+	if err != nil {
+		return err
+	}
+
+	if err := h.logger.BalanceStream(ctx, balanceChanges); err != nil {
+		return nil
+	}
+
+	// We only attempt to reconciler changes when blocks are added,
+	// not removed
+	return nil
+}
diff --git a/internal/reconciler/reconciler.go b/internal/reconciler/reconciler.go
index d8f778f..a46d359 100644
--- a/internal/reconciler/reconciler.go
+++ b/internal/reconciler/reconciler.go
@@ -16,25 +16,624 @@ package reconciler
 
 import (
 	"context"
+	"errors"
 	"fmt"
-	"reflect"
+	"log"
+	"sync"
+	"time"
 
-	"github.com/coinbase/rosetta-cli/internal/storage"
+	"github.com/coinbase/rosetta-cli/internal/utils"
 
 	"github.com/coinbase/rosetta-sdk-go/fetcher"
 	"github.com/coinbase/rosetta-sdk-go/types"
+	"golang.org/x/sync/errgroup"
 )
 
-// Reconciler defines an interface for comparing
-// computed balances with node balances.
-type Reconciler interface {
-	QueueChanges(
+const (
+	// backlogThreshold is the limit of account lookups
+	// that can be enqueued to reconcile before new
+	// requests are dropped.
+	// TODO: Make configurable
+	backlogThreshold = 1000
+
+	// waitToCheckDiff is the syncing difference (live-head)
+	// to retry instead of exiting. In other words, if the
+	// processed head is behind the live head by <
+	// waitToCheckDiff we should try again after sleeping.
+	// TODO: Make configurable
+	waitToCheckDiff = 10
+
+	// waitToCheckDiffSleep is the amount of time to wait
+	// to check a balance difference if the syncer is within
+	// waitToCheckDiff from the block a balance was queried at.
+	waitToCheckDiffSleep = 5 * time.Second
+
+	// activeReconciliation is included in the reconciliation
+	// error message if reconciliation failed during active
+	// reconciliation.
+	activeReconciliation = "ACTIVE"
+
+	// inactiveReconciliation is included in the reconciliation
+	// error message if reconciliation failed during inactive
+	// reconciliation.
+	inactiveReconciliation = "INACTIVE"
+
+	// zeroString is a string of value 0.
+	zeroString = "0"
+
+	// inactiveReconciliationSleep is used as the time.Duration
+	// to sleep when there are no seen accounts to reconcile.
+	inactiveReconciliationSleep = 5 * time.Second
+
+	// inactiveReconciliationRequiredDepth is the minimum
+	// number of blocks the reconciler should wait between
+	// inactive reconciliations.
+	// TODO: make configurable
+	inactiveReconciliationRequiredDepth = 500
+)
+
+var (
+	// ErrHeadBlockBehindLive is returned when the processed
+	// head is behind the live head. Sometimes, it is
+	// preferrable to sleep and wait to catch up when
+	// we are close to the live head (waitToCheckDiff).
+	ErrHeadBlockBehindLive = errors.New("head block behind")
+
+	// ErrAccountUpdated is returned when the
+	// account was updated at a height later than
+	// the live height (when the account balance was fetched).
+	ErrAccountUpdated = errors.New("account updated")
+
+	// ErrBlockGone is returned when the processed block
+	// head is greater than the live head but the block
+	// does not exist in the store. This likely means
+	// that the block was orphaned.
+	ErrBlockGone = errors.New("block gone")
+)
+
+// BalanceChange represents a balance change that affected
+// a *types.AccountIdentifier and a *types.Currency.
+type BalanceChange struct {
+	Account    *types.AccountIdentifier `json:"account_identifier,omitempty"`
+	Currency   *types.Currency          `json:"currency,omitempty"`
+	Block      *types.BlockIdentifier   `json:"block_identifier,omitempty"`
+	Difference string                   `json:"difference,omitempty"`
+}
+
+// Helper functions are used by Reconciler to compare
+// computed balances from a block with the balance calculated
+// by the node. Defining an interface allows the client to determine
+// what sort of storage layer they want to use to provide the required
+// information.
+type Helper interface {
+	BlockExists(
 		ctx context.Context,
 		block *types.BlockIdentifier,
-		changes []*storage.BalanceChange,
+	) (bool, error)
+
+	CurrentBlock(
+		ctx context.Context,
+	) (*types.BlockIdentifier, error)
+
+	AccountBalance(
+		ctx context.Context,
+		account *types.AccountIdentifier,
+		currency *types.Currency,
+		headBlock *types.BlockIdentifier,
+	) (*types.Amount, *types.BlockIdentifier, error)
+}
+
+// Handler is called by Reconciler after a reconciliation
+// is performed. When a reconciliation failure is observed,
+// it is up to the client to halt syncing or log the result.
+type Handler interface {
+	ReconciliationFailed(
+		ctx context.Context,
+		reconciliationType string,
+		account *types.AccountIdentifier,
+		currency *types.Currency,
+		computedBalance string,
+		nodeBalance string,
+		block *types.BlockIdentifier,
 	) error
 
-	Reconcile(ctx context.Context) error
+	ReconciliationSucceeded(
+		ctx context.Context,
+		reconciliationType string,
+		account *types.AccountIdentifier,
+		currency *types.Currency,
+		balance string,
+		block *types.BlockIdentifier,
+	) error
+}
+
+// Reconciler contains all logic to reconcile balances of
+// types.AccountIdentifiers returned in types.Operations
+// by a Rosetta Server.
+type Reconciler struct {
+	network              *types.NetworkIdentifier
+	helper               Helper
+	handler              Handler
+	fetcher              *fetcher.Fetcher
+	accountConcurrency   uint64
+	lookupBalanceByBlock bool
+	interestingAccounts  []*AccountCurrency
+	changeQueue          chan *BalanceChange
+
+	// highWaterMark is used to skip requests when
+	// we are very far behind the live head.
+	highWaterMark int64
+
+	// seenAccts are stored for inactive account
+	// reconciliation.
+	seenAccts     []*AccountCurrency
+	inactiveQueue []*BalanceChange
+
+	// inactiveQueueMutex needed because we can't peek at the tip
+	// of a channel to determine when it is ready to look at.
+	inactiveQueueMutex sync.Mutex
+}
+
+// NewReconciler creates a new Reconciler.
+func NewReconciler(
+	network *types.NetworkIdentifier,
+	helper Helper,
+	handler Handler,
+	fetcher *fetcher.Fetcher,
+	accountConcurrency uint64,
+	lookupBalanceByBlock bool,
+	interestingAccounts []*AccountCurrency,
+	// TODO: load seenAccts and inactiveQueue from prior run (if exists)
+) *Reconciler {
+	r := &Reconciler{
+		network:              network,
+		helper:               helper,
+		handler:              handler,
+		fetcher:              fetcher,
+		accountConcurrency:   accountConcurrency,
+		lookupBalanceByBlock: lookupBalanceByBlock,
+		interestingAccounts:  interestingAccounts,
+		highWaterMark:        -1,
+		seenAccts:            make([]*AccountCurrency, 0),
+		inactiveQueue:        make([]*BalanceChange, 0),
+	}
+
+	if lookupBalanceByBlock {
+		// When lookupBalanceByBlock is enabled, we check
+		// balance changes synchronously.
+		r.changeQueue = make(chan *BalanceChange)
+	} else {
+		// When lookupBalanceByBlock is disabled, we must check
+		// balance changes asynchronously. Using a buffered
+		// channel allows us to add balance changes without blocking.
+		r.changeQueue = make(chan *BalanceChange, backlogThreshold)
+	}
+
+	return r
+}
+
+// QueueChanges enqueues a slice of *BalanceChanges
+// for reconciliation.
+func (r *Reconciler) QueueChanges(
+	ctx context.Context,
+	block *types.BlockIdentifier,
+	balanceChanges []*BalanceChange,
+) error {
+	// Ensure all interestingAccounts are checked
+	for _, account := range r.interestingAccounts {
+		skipAccount := false
+		// Look through balance changes for account + currency
+		for _, change := range balanceChanges {
+			if utils.Equal(change.Account, account.Account) &&
+				utils.Equal(change.Currency, account.Currency) {
+				skipAccount = true
+				break
+			}
+		}
+		// Account changed on this block
+		if skipAccount {
+			continue
+		}
+
+		// If account + currency not found, add with difference 0
+		balanceChanges = append(balanceChanges, &BalanceChange{
+			Account:    account.Account,
+			Currency:   account.Currency,
+			Difference: zeroString,
+			Block:      block,
+		})
+	}
+
+	if !r.lookupBalanceByBlock {
+		// All changes will have the same block. Return
+		// if we are too far behind to start reconciling.
+		if block.Index < r.highWaterMark {
+			return nil
+		}
+
+		for _, change := range balanceChanges {
+			select {
+			case r.changeQueue <- change:
+			default:
+				log.Println("skipping active enqueue because backlog")
+			}
+		}
+	} else {
+		// Block until all checked for a block or context is Done
+		for _, change := range balanceChanges {
+			select {
+			case r.changeQueue <- change:
+			case <-ctx.Done():
+				return ctx.Err()
+			}
+		}
+	}
+
+	return nil
+}
+
+// CompareBalance checks to see if the computed balance of an account
+// is equal to the live balance of an account. This function ensures
+// balance is checked correctly in the case of orphaned blocks.
+func (r *Reconciler) CompareBalance(
+	ctx context.Context,
+	account *types.AccountIdentifier,
+	currency *types.Currency,
+	amount string,
+	liveBlock *types.BlockIdentifier,
+) (string, string, int64, error) {
+	// Head block should be set before we CompareBalance
+	head, err := r.helper.CurrentBlock(ctx)
+	if err != nil {
+		return zeroString, "", 0, fmt.Errorf(
+			"%w: unable to get current block for reconciliation",
+			err,
+		)
+	}
+
+	// Check if live block is < head (or wait)
+	if liveBlock.Index > head.Index {
+		return zeroString, "", head.Index, fmt.Errorf(
+			"%w live block %d > head block %d",
+			ErrHeadBlockBehindLive,
+			liveBlock.Index,
+			head.Index,
+		)
+	}
+
+	// Check if live block is in store (ensure not reorged)
+	exists, err := r.helper.BlockExists(ctx, liveBlock)
+	if err != nil {
+		return zeroString, "", 0, fmt.Errorf(
+			"%w: unable to check if block exists: %+v",
+			err,
+			liveBlock,
+		)
+	}
+	if !exists {
+		return zeroString, "", head.Index, fmt.Errorf(
+			"%w %+v",
+			ErrBlockGone,
+			liveBlock,
+		)
+	}
+
+	// Check if live block < computed head
+	cachedBalance, balanceBlock, err := r.helper.AccountBalance(
+		ctx,
+		account,
+		currency,
+		head,
+	)
+	if err != nil {
+		return zeroString, "", head.Index, fmt.Errorf(
+			"%w: unable to get cached balance for %+v:%+v",
+			err,
+			account,
+			currency,
+		)
+	}
+
+	if liveBlock.Index < balanceBlock.Index {
+		return zeroString, "", head.Index, fmt.Errorf(
+			"%w %+v updated at %d",
+			ErrAccountUpdated,
+			account,
+			balanceBlock.Index,
+		)
+	}
+
+	difference, err := types.SubtractValues(cachedBalance.Value, amount)
+	if err != nil {
+		return "", "", -1, err
+	}
+
+	return difference, cachedBalance.Value, head.Index, nil
+}
+
+// bestBalance returns the balance for an account
+// at either the current block (if lookupBalanceByBlock is
+// disabled) or at some historical block.
+func (r *Reconciler) bestBalance(
+	ctx context.Context,
+	account *types.AccountIdentifier,
+	currency *types.Currency,
+	block *types.PartialBlockIdentifier,
+) (*types.BlockIdentifier, string, error) {
+	if !r.lookupBalanceByBlock {
+		// Use the current balance to reconcile balances when lookupBalanceByBlock
+		// is disabled. This could be the case when a rosetta server does not
+		// support historical balance lookups.
+		block = nil
+	}
+	return GetCurrencyBalance(
+		ctx,
+		r.fetcher,
+		r.network,
+		account,
+		currency,
+		block,
+	)
+}
+
+// accountReconciliation returns an error if the provided
+// AccountAndCurrency's live balance cannot be reconciled
+// with the computed balance.
+func (r *Reconciler) accountReconciliation(
+	ctx context.Context,
+	account *types.AccountIdentifier,
+	currency *types.Currency,
+	liveAmount string,
+	liveBlock *types.BlockIdentifier,
+	inactive bool,
+) error {
+	accountCurrency := &AccountCurrency{
+		Account:  account,
+		Currency: currency,
+	}
+	for ctx.Err() == nil {
+		// If don't have previous balance because stateless, check diff on block
+		// instead of comparing entire computed balance
+		difference, cachedBalance, headIndex, err := r.CompareBalance(
+			ctx,
+			account,
+			currency,
+			liveAmount,
+			liveBlock,
+		)
+		if err != nil {
+			if errors.Is(err, ErrHeadBlockBehindLive) {
+				// This error will only occur when lookupBalanceByBlock
+				// is disabled and the syncer is behind the current block of
+				// the node. This error should never occur when
+				// lookupBalanceByBlock is enabled.
+				diff := liveBlock.Index - headIndex
+				if diff < waitToCheckDiff {
+					time.Sleep(waitToCheckDiffSleep)
+					continue
+				}
+
+				// Don't wait to check if we are very far behind
+				log.Printf(
+					"Skipping reconciliation for %s: %d blocks behind\n",
+					simpleAccountCurrency(accountCurrency),
+					diff,
+				)
+
+				// Set a highWaterMark to not accept any new
+				// reconciliation requests unless they happened
+				// after this new highWaterMark.
+				r.highWaterMark = liveBlock.Index
+				break
+			}
+
+			if errors.Is(err, ErrBlockGone) {
+				// Either the block has not been processed in a re-org yet
+				// or the block was orphaned
+				break
+			}
+
+			if errors.Is(err, ErrAccountUpdated) {
+				// If account was updated, it must be
+				// enqueued again
+				break
+			}
+
+			return err
+		}
+
+		reconciliationType := activeReconciliation
+		if inactive {
+			reconciliationType = inactiveReconciliation
+		}
+
+		if difference != zeroString {
+			err := r.handler.ReconciliationFailed(
+				ctx,
+				reconciliationType,
+				accountCurrency.Account,
+				accountCurrency.Currency,
+				cachedBalance,
+				liveAmount,
+				liveBlock,
+			)
+			if err != nil {
+				return err
+			}
+
+			return nil
+		}
+
+		r.inactiveAccountQueue(inactive, accountCurrency, liveBlock)
+		return r.handler.ReconciliationSucceeded(
+			ctx,
+			reconciliationType,
+			accountCurrency.Account,
+			accountCurrency.Currency,
+			liveAmount,
+			liveBlock,
+		)
+	}
+
+	return nil
+}
+
+func (r *Reconciler) inactiveAccountQueue(
+	inactive bool,
+	accountCurrency *AccountCurrency,
+	liveBlock *types.BlockIdentifier,
+) {
+	// Only enqueue the first time we see an account on an active reconciliation.
+	shouldEnqueueInactive := false
+	if !inactive && !ContainsAccountCurrency(r.seenAccts, accountCurrency) {
+		r.seenAccts = append(r.seenAccts, accountCurrency)
+		shouldEnqueueInactive = true
+	}
+
+	if inactive || shouldEnqueueInactive {
+		r.inactiveQueueMutex.Lock()
+		r.inactiveQueue = append(r.inactiveQueue, &BalanceChange{
+			Account:  accountCurrency.Account,
+			Currency: accountCurrency.Currency,
+			Block:    liveBlock,
+		})
+		r.inactiveQueueMutex.Unlock()
+	}
+}
+
+// simpleAccountCurrency returns a string that is a simple
+// representation of an AccountCurrency struct.
+func simpleAccountCurrency(
+	accountCurrency *AccountCurrency,
+) string {
+	acctString := accountCurrency.Account.Address
+	if accountCurrency.Account.SubAccount != nil {
+		acctString += accountCurrency.Account.SubAccount.Address
+	}
+
+	acctString += accountCurrency.Currency.Symbol
+
+	return acctString
+}
+
+// reconcileActiveAccounts selects an account
+// from the Reconciler account queue and
+// reconciles the balance. This is useful
+// for detecting if balance changes in operations
+// were correct.
+func (r *Reconciler) reconcileActiveAccounts(
+	ctx context.Context,
+) error {
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case balanceChange := <-r.changeQueue:
+			if balanceChange.Block.Index < r.highWaterMark {
+				continue
+			}
+
+			block, value, err := r.bestBalance(
+				ctx,
+				balanceChange.Account,
+				balanceChange.Currency,
+				types.ConstructPartialBlockIdentifier(balanceChange.Block),
+			)
+			if err != nil {
+				return err
+			}
+
+			err = r.accountReconciliation(
+				ctx,
+				balanceChange.Account,
+				balanceChange.Currency,
+				value,
+				block,
+				false,
+			)
+			if err != nil {
+				return err
+			}
+		}
+	}
+}
+
+// reconcileInactiveAccounts selects a random account
+// from all previously seen accounts and reconciles
+// the balance. This is useful for detecting balance
+// changes that were not returned in operations.
+func (r *Reconciler) reconcileInactiveAccounts(
+	ctx context.Context,
+) error {
+	for ctx.Err() == nil {
+		head, err := r.helper.CurrentBlock(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 err != nil {
+			time.Sleep(inactiveReconciliationSleep)
+			log.Printf(
+				"%s: waiting to start inactive reconciliation until current block set\n",
+				err.Error(),
+			)
+			continue
+		}
+
+		r.inactiveQueueMutex.Lock()
+		if len(r.inactiveQueue) > 0 &&
+			r.inactiveQueue[0].Block.Index+inactiveReconciliationRequiredDepth < head.Index {
+			randAcct := r.inactiveQueue[0]
+			r.inactiveQueue = r.inactiveQueue[1:]
+			r.inactiveQueueMutex.Unlock()
+
+			block, amount, err := r.bestBalance(
+				ctx,
+				randAcct.Account,
+				randAcct.Currency,
+				types.ConstructPartialBlockIdentifier(head),
+			)
+			if err != nil {
+				return err
+			}
+
+			err = r.accountReconciliation(
+				ctx,
+				randAcct.Account,
+				randAcct.Currency,
+				amount,
+				block,
+				true,
+			)
+			if err != nil {
+				return err
+			}
+		} else {
+			r.inactiveQueueMutex.Unlock()
+			time.Sleep(inactiveReconciliationSleep)
+		}
+	}
+
+	return nil
+}
+
+// Reconcile starts the active and inactive Reconciler goroutines.
+// If any goroutine errors, the function will return an error.
+func (r *Reconciler) Reconcile(ctx context.Context) error {
+	g, ctx := errgroup.WithContext(ctx)
+	for j := uint64(0); j < r.accountConcurrency/2; j++ {
+		g.Go(func() error {
+			return r.reconcileActiveAccounts(ctx)
+		})
+
+		g.Go(func() error {
+			return r.reconcileInactiveAccounts(ctx)
+		})
+	}
+
+	if err := g.Wait(); err != nil {
+		return err
+	}
+
+	return nil
 }
 
 // ExtractAmount returns the types.Amount from a slice of types.Balance
@@ -44,7 +643,7 @@ func ExtractAmount(
 	currency *types.Currency,
 ) (*types.Amount, error) {
 	for _, b := range balances {
-		if !reflect.DeepEqual(b.Currency, currency) {
+		if !utils.Equal(b.Currency, currency) {
 			continue
 		}
 
@@ -69,8 +668,8 @@ func ContainsAccountCurrency(
 	change *AccountCurrency,
 ) bool {
 	for _, a := range arr {
-		if reflect.DeepEqual(a.Account, change.Account) &&
-			reflect.DeepEqual(a.Currency, change.Currency) {
+		if utils.Equal(a.Account, change.Account) &&
+			utils.Equal(a.Currency, change.Currency) {
 			return true
 		}
 	}
diff --git a/internal/reconciler/stateful_reconciler.go b/internal/reconciler/stateful_reconciler.go
deleted file mode 100644
index 61b7916..0000000
--- a/internal/reconciler/stateful_reconciler.go
+++ /dev/null
@@ -1,536 +0,0 @@
-// Copyright 2020 Coinbase, Inc.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//      http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package reconciler
-
-import (
-	"context"
-	"errors"
-	"fmt"
-	"log"
-	"sync"
-	"time"
-
-	"github.com/coinbase/rosetta-cli/internal/logger"
-	"github.com/coinbase/rosetta-cli/internal/storage"
-
-	"github.com/coinbase/rosetta-sdk-go/fetcher"
-	"github.com/coinbase/rosetta-sdk-go/types"
-	"golang.org/x/sync/errgroup"
-)
-
-const (
-	// backlogThreshold is the limit of account lookups
-	// that can be enqueued to reconcile before new
-	// requests are dropped.
-	// TODO: Make configurable
-	backlogThreshold = 1000
-
-	// waitToCheckDiff is the syncing difference (live-head)
-	// to retry instead of exiting. In other words, if the
-	// processed head is behind the live head by <
-	// waitToCheckDiff we should try again after sleeping.
-	// TODO: Make configurable
-	waitToCheckDiff = 10
-
-	// waitToCheckDiffSleep is the amount of time to wait
-	// to check a balance difference if the syncer is within
-	// waitToCheckDiff from the block a balance was queried at.
-	waitToCheckDiffSleep = 5 * time.Second
-
-	// activeReconciliation is included in the reconciliation
-	// error message if reconciliation failed during active
-	// reconciliation.
-	activeReconciliation = "ACTIVE"
-
-	// inactiveReconciliation is included in the reconciliation
-	// error message if reconciliation failed during inactive
-	// reconciliation.
-	inactiveReconciliation = "INACTIVE"
-
-	// zeroString is a string of value 0.
-	zeroString = "0"
-
-	// inactiveReconciliationSleep is used as the time.Duration
-	// to sleep when there are no seen accounts to reconcile.
-	inactiveReconciliationSleep = 5 * time.Second
-
-	// inactiveReconciliationRequiredDepth is the minimum
-	// number of blocks the reconciler should wait between
-	// inactive reconciliations.
-	// TODO: make configurable
-	inactiveReconciliationRequiredDepth = 500
-)
-
-var (
-	// ErrHeadBlockBehindLive is returned when the processed
-	// head is behind the live head. Sometimes, it is
-	// preferrable to sleep and wait to catch up when
-	// we are close to the live head (waitToCheckDiff).
-	ErrHeadBlockBehindLive = errors.New("head block behind")
-
-	// ErrAccountUpdated is returned when the
-	// account was updated at a height later than
-	// the live height (when the account balance was fetched).
-	ErrAccountUpdated = errors.New("account updated")
-
-	// ErrBlockGone is returned when the processed block
-	// head is greater than the live head but the block
-	// does not exist in the store. This likely means
-	// that the block was orphaned.
-	ErrBlockGone = errors.New("block gone")
-)
-
-// StatefulReconciler contains all logic to reconcile balances of
-// types.AccountIdentifiers returned in types.Operations
-// by a Rosetta Server.
-type StatefulReconciler struct {
-	network                   *types.NetworkIdentifier
-	storage                   *storage.BlockStorage
-	fetcher                   *fetcher.Fetcher
-	logger                    *logger.Logger
-	accountConcurrency        uint64
-	lookupBalanceByBlock      bool
-	haltOnReconciliationError bool
-	changeQueue               chan *storage.BalanceChange
-
-	// highWaterMark is used to skip requests when
-	// we are very far behind the live head.
-	highWaterMark int64
-
-	// seenAccts are stored for inactive account
-	// reconciliation.
-	seenAccts     []*AccountCurrency
-	inactiveQueue []*storage.BalanceChange
-
-	// inactiveQueueMutex needed because we can't peek at the tip
-	// of a channel to determine when it is ready to look at.
-	inactiveQueueMutex sync.Mutex
-}
-
-// NewStateful creates a new StatefulReconciler.
-func NewStateful(
-	network *types.NetworkIdentifier,
-	blockStorage *storage.BlockStorage,
-	fetcher *fetcher.Fetcher,
-	logger *logger.Logger,
-	accountConcurrency uint64,
-	lookupBalanceByBlock bool,
-	haltOnReconciliationError bool,
-) *StatefulReconciler {
-	return &StatefulReconciler{
-		network:                   network,
-		storage:                   blockStorage,
-		fetcher:                   fetcher,
-		logger:                    logger,
-		accountConcurrency:        accountConcurrency,
-		lookupBalanceByBlock:      lookupBalanceByBlock,
-		haltOnReconciliationError: haltOnReconciliationError,
-		changeQueue:               make(chan *storage.BalanceChange, backlogThreshold),
-		highWaterMark:             -1,
-		seenAccts:                 make([]*AccountCurrency, 0),
-		inactiveQueue:             make([]*storage.BalanceChange, 0),
-	}
-}
-
-// QueueChanges enqueues a slice of *storage.BalanceChanges
-// for reconciliation.
-func (r *StatefulReconciler) QueueChanges(
-	ctx context.Context,
-	block *types.BlockIdentifier,
-	balanceChanges []*storage.BalanceChange,
-) error {
-	// All changes will have the same block. Return
-	// if we are too far behind to start reconciling.
-	if block.Index < r.highWaterMark {
-		return nil
-	}
-
-	// Use a buffered channel so don't need to
-	// spawn a goroutine to add accounts to channel.
-	for _, change := range balanceChanges {
-		select {
-		case r.changeQueue <- change:
-		default:
-			log.Println("skipping active enqueue because backlog")
-		}
-	}
-
-	return nil
-}
-
-// CompareBalance checks to see if the computed balance of an account
-// is equal to the live balance of an account. This function ensures
-// balance is checked correctly in the case of orphaned blocks.
-func (r *StatefulReconciler) CompareBalance(
-	ctx context.Context,
-	account *types.AccountIdentifier,
-	currency *types.Currency,
-	amount string,
-	liveBlock *types.BlockIdentifier,
-) (string, int64, error) {
-	txn := r.storage.NewDatabaseTransaction(ctx, false)
-	defer txn.Discard(ctx)
-
-	// Head block should be set before we CompareBalance
-	head, err := r.storage.GetHeadBlockIdentifier(ctx, txn)
-	if err != nil {
-		return zeroString, 0, fmt.Errorf("%w: unable to get current block for reconciliation", err)
-	}
-
-	// Check if live block is < head (or wait)
-	if liveBlock.Index > head.Index {
-		return zeroString, head.Index, fmt.Errorf(
-			"%w live block %d > head block %d",
-			ErrHeadBlockBehindLive,
-			liveBlock.Index,
-			head.Index,
-		)
-	}
-
-	// Check if live block is in store (ensure not reorged)
-	_, err = r.storage.GetBlock(ctx, txn, liveBlock)
-	if err != nil {
-		return zeroString, head.Index, fmt.Errorf(
-			"%w %+v",
-			ErrBlockGone,
-			liveBlock,
-		)
-	}
-
-	// Check if live block < computed head
-	amounts, balanceBlock, err := r.storage.GetBalance(ctx, txn, account)
-	if err != nil {
-		return zeroString, head.Index, err
-	}
-
-	if liveBlock.Index < balanceBlock.Index {
-		return zeroString, head.Index, fmt.Errorf(
-			"%w %+v updated at %d",
-			ErrAccountUpdated,
-			account,
-			balanceBlock.Index,
-		)
-	}
-
-	// Check balances are equal
-	computedAmount, ok := amounts[storage.GetCurrencyKey(currency)]
-	if !ok {
-		return "", head.Index, fmt.Errorf(
-			"currency %+v not found",
-			*currency,
-		)
-	}
-
-	difference, err := storage.SubtractStringValues(computedAmount.Value, amount)
-	if err != nil {
-		return "", -1, err
-	}
-
-	if difference != zeroString {
-		return difference, head.Index, nil
-	}
-
-	return zeroString, head.Index, nil
-}
-
-// getAccountBalance returns the balance for an account
-// at either the current block (if lookupBalanceByBlock is
-// disabled) or at some historical block.
-func (r *StatefulReconciler) bestBalance(
-	ctx context.Context,
-	account *types.AccountIdentifier,
-	currency *types.Currency,
-	block *types.PartialBlockIdentifier,
-) (*types.BlockIdentifier, string, error) {
-	if !r.lookupBalanceByBlock {
-		// Use the current balance to reconcile balances when lookupBalanceByBlock
-		// is disabled. This could be the case when a rosetta server does not
-		// support historical balance lookups.
-		block = nil
-	}
-	return GetCurrencyBalance(
-		ctx,
-		r.fetcher,
-		r.network,
-		account,
-		currency,
-		block,
-	)
-}
-
-// accountReconciliation returns an error if the provided
-// AccountAndCurrency's live balance cannot be reconciled
-// with the computed balance.
-func (r *StatefulReconciler) accountReconciliation(
-	ctx context.Context,
-	account *types.AccountIdentifier,
-	currency *types.Currency,
-	liveAmount string,
-	liveBlock *types.BlockIdentifier,
-	inactive bool,
-) error {
-	accountCurrency := &AccountCurrency{
-		Account:  account,
-		Currency: currency,
-	}
-	for ctx.Err() == nil {
-		difference, headIndex, err := r.CompareBalance(
-			ctx,
-			account,
-			currency,
-			liveAmount,
-			liveBlock,
-		)
-		if err != nil {
-			if errors.Is(err, ErrHeadBlockBehindLive) {
-				// This error will only occur when lookupBalanceByBlock
-				// is disabled and the syncer is behind the current block of
-				// the node. This error should never occur when
-				// lookupBalanceByBlock is enabled.
-				diff := liveBlock.Index - headIndex
-				if diff < waitToCheckDiff {
-					time.Sleep(waitToCheckDiffSleep)
-					continue
-				}
-
-				// Don't wait to check if we are very far behind
-				log.Printf(
-					"Skipping reconciliation for %s: %d blocks behind\n",
-					simpleAccountCurrency(accountCurrency),
-					diff,
-				)
-
-				// Set a highWaterMark to not accept any new
-				// reconciliation requests unless they happened
-				// after this new highWaterMark.
-				r.highWaterMark = liveBlock.Index
-				break
-			}
-
-			if errors.Is(err, ErrBlockGone) {
-				// Either the block has not been processed in a re-org yet
-				// or the block was orphaned
-				break
-			}
-
-			if errors.Is(err, ErrAccountUpdated) {
-				// If account was updated, it must be
-				// enqueued again
-				break
-			}
-
-			return err
-		}
-
-		reconciliationType := activeReconciliation
-		if inactive {
-			reconciliationType = inactiveReconciliation
-		}
-
-		if difference != zeroString {
-			err := r.logger.ReconcileFailureStream(
-				ctx,
-				reconciliationType,
-				accountCurrency.Account,
-				accountCurrency.Currency,
-				difference,
-				liveBlock,
-			)
-
-			if err != nil {
-				return err
-			}
-
-			if r.haltOnReconciliationError {
-				return errors.New("reconciliation error")
-			}
-
-			return nil
-		}
-
-		r.inactiveAccountQueue(inactive, accountCurrency, liveBlock)
-		return r.logger.ReconcileSuccessStream(
-			ctx,
-			reconciliationType,
-			accountCurrency.Account,
-			&types.Amount{
-				Value:    liveAmount,
-				Currency: currency,
-			},
-			liveBlock,
-		)
-	}
-
-	return nil
-}
-
-func (r *StatefulReconciler) inactiveAccountQueue(
-	inactive bool,
-	accountCurrency *AccountCurrency,
-	liveBlock *types.BlockIdentifier,
-) {
-	// Only enqueue the first time we see an account on an active reconciliation.
-	shouldEnqueueInactive := false
-	if !inactive && !ContainsAccountCurrency(r.seenAccts, accountCurrency) {
-		r.seenAccts = append(r.seenAccts, accountCurrency)
-		shouldEnqueueInactive = true
-	}
-
-	if inactive || shouldEnqueueInactive {
-		r.inactiveQueueMutex.Lock()
-		r.inactiveQueue = append(r.inactiveQueue, &storage.BalanceChange{
-			Account:  accountCurrency.Account,
-			Currency: accountCurrency.Currency,
-			Block:    liveBlock,
-		})
-		r.inactiveQueueMutex.Unlock()
-	}
-}
-
-// simpleAccountCurrency returns a string that is a simple
-// representation of an AccountCurrency struct.
-func simpleAccountCurrency(
-	accountCurrency *AccountCurrency,
-) string {
-	acctString := accountCurrency.Account.Address
-	if accountCurrency.Account.SubAccount != nil {
-		acctString += accountCurrency.Account.SubAccount.Address
-	}
-
-	acctString += accountCurrency.Currency.Symbol
-
-	return acctString
-}
-
-// reconcileActiveAccounts selects an account
-// from the StatefulReconciler account queue and
-// reconciles the balance. This is useful
-// for detecting if balance changes in operations
-// were correct.
-func (r *StatefulReconciler) reconcileActiveAccounts(
-	ctx context.Context,
-) error {
-	for {
-		select {
-		case <-ctx.Done():
-			return ctx.Err()
-		case balanceChange := <-r.changeQueue:
-			if balanceChange.Block.Index < r.highWaterMark {
-				continue
-			}
-
-			block, value, err := r.bestBalance(
-				ctx,
-				balanceChange.Account,
-				balanceChange.Currency,
-				types.ConstructPartialBlockIdentifier(balanceChange.Block),
-			)
-			if err != nil {
-				return err
-			}
-
-			err = r.accountReconciliation(
-				ctx,
-				balanceChange.Account,
-				balanceChange.Currency,
-				value,
-				block,
-				false,
-			)
-			if err != nil {
-				return err
-			}
-		}
-	}
-}
-
-// reconcileInactiveAccounts selects a random account
-// from all previously seen accounts and reconciles
-// the balance. This is useful for detecting balance
-// changes that were not returned in operations.
-func (r *StatefulReconciler) reconcileInactiveAccounts(
-	ctx context.Context,
-) error {
-	for ctx.Err() == nil {
-		txn := r.storage.NewDatabaseTransaction(ctx, false)
-		head, err := r.storage.GetHeadBlockIdentifier(ctx, txn)
-		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 &&
-			r.inactiveQueue[0].Block.Index+inactiveReconciliationRequiredDepth < head.Index {
-			randAcct := r.inactiveQueue[0]
-			r.inactiveQueue = r.inactiveQueue[1:]
-			r.inactiveQueueMutex.Unlock()
-
-			block, amount, err := r.bestBalance(
-				ctx,
-				randAcct.Account,
-				randAcct.Currency,
-				types.ConstructPartialBlockIdentifier(head),
-			)
-			if err != nil {
-				return err
-			}
-
-			err = r.accountReconciliation(
-				ctx,
-				randAcct.Account,
-				randAcct.Currency,
-				amount,
-				block,
-				true,
-			)
-			if err != nil {
-				return err
-			}
-		} else {
-			r.inactiveQueueMutex.Unlock()
-			time.Sleep(inactiveReconciliationSleep)
-		}
-	}
-
-	return nil
-}
-
-// Reconcile starts the active and inactive StatefulReconciler goroutines.
-// If any goroutine errors, the function will return an error.
-func (r *StatefulReconciler) Reconcile(ctx context.Context) error {
-	g, ctx := errgroup.WithContext(ctx)
-	for j := uint64(0); j < r.accountConcurrency/2; j++ {
-		g.Go(func() error {
-			return r.reconcileActiveAccounts(ctx)
-		})
-
-		g.Go(func() error {
-			return r.reconcileInactiveAccounts(ctx)
-		})
-	}
-
-	if err := g.Wait(); err != nil {
-		return err
-	}
-
-	return nil
-}
diff --git a/internal/reconciler/stateless_reconciler.go b/internal/reconciler/stateless_reconciler.go
deleted file mode 100644
index d6d9f38..0000000
--- a/internal/reconciler/stateless_reconciler.go
+++ /dev/null
@@ -1,238 +0,0 @@
-// Copyright 2020 Coinbase, Inc.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//      http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package reconciler
-
-import (
-	"context"
-	"errors"
-	"reflect"
-
-	"github.com/coinbase/rosetta-cli/internal/logger"
-	"github.com/coinbase/rosetta-cli/internal/storage"
-
-	"github.com/coinbase/rosetta-sdk-go/fetcher"
-	"github.com/coinbase/rosetta-sdk-go/types"
-	"golang.org/x/sync/errgroup"
-)
-
-// StatelessReconciler compares computed balances with
-// the node balance without using any sort of persistent
-// state.	If it is not possible to lookup a balance by block,
-// you must use the StatefulReconciler. If you want to perform
-// inactive reconciliation (check for balance changes that
-// occurred that were not in blocks), you must also use the
-// StatefulReconciler. Lastly, the StatelessReconciler does
-// not support re-orgs.
-type StatelessReconciler struct {
-	network                   *types.NetworkIdentifier
-	fetcher                   *fetcher.Fetcher
-	logger                    *logger.Logger
-	accountConcurrency        uint64
-	haltOnReconciliationError bool
-	interestingAccounts       []*AccountCurrency
-	changeQueue               chan *storage.BalanceChange
-}
-
-// NewStateless returns a new StatelessReconciler.
-func NewStateless(
-	network *types.NetworkIdentifier,
-	fetcher *fetcher.Fetcher,
-	logger *logger.Logger,
-	accountConcurrency uint64,
-	haltOnReconciliationError bool,
-	interestingAccounts []*AccountCurrency,
-) *StatelessReconciler {
-	return &StatelessReconciler{
-		network:                   network,
-		fetcher:                   fetcher,
-		logger:                    logger,
-		accountConcurrency:        accountConcurrency,
-		haltOnReconciliationError: haltOnReconciliationError,
-		interestingAccounts:       interestingAccounts,
-		changeQueue:               make(chan *storage.BalanceChange),
-	}
-}
-
-// QueueChanges enqueues a slice of *storage.BalanceChanges
-// for reconciliation.
-func (r *StatelessReconciler) QueueChanges(
-	ctx context.Context,
-	block *types.BlockIdentifier,
-	balanceChanges []*storage.BalanceChange,
-) error {
-	// Ensure all interestingAccounts are checked
-	// TODO: refactor to automatically trigger once an inactive reconciliation error
-	// is discovered
-	for _, account := range r.interestingAccounts {
-		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) {
-				skipAccount = true
-				break
-			}
-		}
-
-		// Account changed on this block
-		if skipAccount {
-			continue
-		}
-
-		// If account + currency not found, add with difference 0
-		balanceChanges = append(balanceChanges, &storage.BalanceChange{
-			Account:    account.Account,
-			Currency:   account.Currency,
-			Difference: "0",
-			Block:      block,
-		})
-	}
-
-	// Block until all checked for a block
-	for _, change := range balanceChanges {
-		select {
-		case r.changeQueue <- change:
-		case <-ctx.Done():
-			return ctx.Err()
-		}
-	}
-
-	return nil
-}
-
-func (r *StatelessReconciler) balanceAtIndex(
-	ctx context.Context,
-	account *types.AccountIdentifier,
-	currency *types.Currency,
-	index int64,
-) (string, error) {
-	_, value, err := GetCurrencyBalance(
-		ctx,
-		r.fetcher,
-		r.network,
-		account,
-		currency,
-		&types.PartialBlockIdentifier{
-			Index: &index,
-		},
-	)
-
-	return value, err
-}
-
-func (r *StatelessReconciler) reconcileChange(
-	ctx context.Context,
-	change *storage.BalanceChange,
-) error {
-	// Get balance at block before change
-	balanceBefore, err := r.balanceAtIndex(
-		ctx,
-		change.Account,
-		change.Currency,
-		change.Block.Index-1,
-	)
-	if err != nil {
-		return err
-	}
-
-	// Get balance at block with change
-	balanceAfter, err := r.balanceAtIndex(
-		ctx,
-		change.Account,
-		change.Currency,
-		change.Block.Index,
-	)
-	if err != nil {
-		return err
-	}
-
-	// Get difference between node change and computed change
-	nodeDifference, err := storage.SubtractStringValues(balanceAfter, balanceBefore)
-	if err != nil {
-		return err
-	}
-
-	difference, err := storage.SubtractStringValues(change.Difference, nodeDifference)
-	if err != nil {
-		return err
-	}
-
-	if difference != zeroString {
-		err := r.logger.ReconcileFailureStream(
-			ctx,
-			activeReconciliation,
-			change.Account,
-			change.Currency,
-			difference,
-			change.Block,
-		)
-		if err != nil {
-			return err
-		}
-
-		if r.haltOnReconciliationError {
-			return errors.New("reconciliation error")
-		}
-
-		return nil
-	}
-
-	return r.logger.ReconcileSuccessStream(
-		ctx,
-		activeReconciliation,
-		change.Account,
-		&types.Amount{
-			Value:    balanceAfter,
-			Currency: change.Currency,
-		},
-		change.Block,
-	)
-}
-
-func (r *StatelessReconciler) reconcileAccounts(
-	ctx context.Context,
-) error {
-	for {
-		select {
-		case <-ctx.Done():
-			return ctx.Err()
-		case change := <-r.changeQueue:
-			err := r.reconcileChange(
-				ctx,
-				change,
-			)
-			if err != nil {
-				return err
-			}
-		}
-	}
-}
-
-// Reconcile starts the active StatelessReconciler goroutines.
-// If any goroutine errors, the function will return an error.
-func (r *StatelessReconciler) Reconcile(ctx context.Context) error {
-	g, ctx := errgroup.WithContext(ctx)
-	for j := uint64(0); j < r.accountConcurrency; j++ {
-		g.Go(func() error {
-			return r.reconcileAccounts(ctx)
-		})
-	}
-
-	if err := g.Wait(); err != nil {
-		return err
-	}
-
-	return nil
-}
diff --git a/internal/storage/block_storage.go b/internal/storage/block_storage.go
index ce67324..cac9ac5 100644
--- a/internal/storage/block_storage.go
+++ b/internal/storage/block_storage.go
@@ -26,11 +26,11 @@ import (
 	"log"
 	"math/big"
 	"path"
-	"strings"
 
-	"github.com/coinbase/rosetta-sdk-go/types"
+	"github.com/coinbase/rosetta-cli/internal/reconciler"
+	"github.com/coinbase/rosetta-cli/internal/syncer"
 
-	"github.com/davecgh/go-spew/spew"
+	"github.com/coinbase/rosetta-sdk-go/types"
 )
 
 const (
@@ -90,9 +90,9 @@ var (
 // hashBytes is used to construct a SHA1
 // hash to protect against arbitrarily
 // large key sizes.
-func hashBytes(data []byte) []byte {
+func hashBytes(data string) []byte {
 	h := sha256.New()
-	_, err := h.Write(data)
+	_, err := h.Write([]byte(data))
 	if err != nil {
 		log.Fatal(err)
 	}
@@ -100,77 +100,70 @@ func hashBytes(data []byte) []byte {
 	return h.Sum(nil)
 }
 
-// hashString is used to construct a SHA1
-// hash to protect against arbitrarily
-// large key sizes.
-func hashString(data string) string {
-	return fmt.Sprintf("%x", hashBytes([]byte(data)))
-}
-
 func getHeadBlockKey() []byte {
-	return hashBytes([]byte(headBlockKey))
+	return hashBytes(headBlockKey)
 }
 
 func getBlockKey(blockIdentifier *types.BlockIdentifier) []byte {
 	return hashBytes(
-		[]byte(fmt.Sprintf("%s:%d", blockIdentifier.Hash, blockIdentifier.Index)),
+		fmt.Sprintf("%s:%d", blockIdentifier.Hash, blockIdentifier.Index),
 	)
 }
 
 func getHashKey(hash string, isBlock bool) []byte {
 	if isBlock {
-		return hashBytes([]byte(fmt.Sprintf("%s:%s", blockHashNamespace, hash)))
+		return hashBytes(fmt.Sprintf("%s:%s", blockHashNamespace, hash))
 	}
 
-	return hashBytes([]byte(fmt.Sprintf("%s:%s", transactionHashNamespace, hash)))
+	return hashBytes(fmt.Sprintf("%s:%s", transactionHashNamespace, hash))
 }
 
-// GetAccountKey returns a byte slice representing a *types.AccountIdentifier.
-// This byte slice automatically handles the existence of *types.SubAccount
-// detail.
-func GetAccountKey(account *types.AccountIdentifier) []byte {
-	if account.SubAccount == nil {
-		return hashBytes(
-			[]byte(fmt.Sprintf("%s:%s", balanceNamespace, account.Address)),
-		)
-	}
+// GetBalanceKey returns a deterministic hash of an types.Account + types.Currency.
+func GetBalanceKey(account *types.AccountIdentifier, currency *types.Currency) []byte {
+	return hashBytes(
+		fmt.Sprintf("%s/%s/%s", balanceNamespace, types.Hash(account), types.Hash(currency)),
+	)
+}
 
-	if account.SubAccount.Metadata == nil {
-		return hashBytes([]byte(fmt.Sprintf(
-			"%s:%s:%s",
-			balanceNamespace,
-			account.Address,
-			account.SubAccount.Address,
-		)))
-	}
-
-	// TODO: handle SubAccount.Metadata
-	// that contains pointer values.
-	return hashBytes([]byte(fmt.Sprintf(
-		"%s:%s:%s:%v",
-		balanceNamespace,
-		account.Address,
-		account.SubAccount.Address,
-		account.SubAccount.Metadata,
-	)))
+// Helper functions are used by BlockStorage to process blocks. Defining an
+// interface allows the client to determine if they wish to query the node for
+// certain information or use another datastore.
+type Helper interface {
+	AccountBalance(
+		ctx context.Context,
+		account *types.AccountIdentifier,
+		currency *types.Currency,
+		block *types.BlockIdentifier,
+	) (*types.Amount, error)
+
+	SkipOperation(
+		ctx context.Context,
+		op *types.Operation,
+	) (bool, error)
 }
 
 // BlockStorage implements block specific storage methods
 // on top of a Database and DatabaseTransaction interface.
 type BlockStorage struct {
-	db Database
+	db     Database
+	helper Helper
 }
 
 // NewBlockStorage returns a new BlockStorage.
-func NewBlockStorage(ctx context.Context, db Database) *BlockStorage {
+func NewBlockStorage(
+	ctx context.Context,
+	db Database,
+	helper Helper,
+) *BlockStorage {
 	return &BlockStorage{
-		db: db,
+		db:     db,
+		helper: helper,
 	}
 }
 
 // NewDatabaseTransaction returns a DatabaseTransaction
 // from the Database that is backing BlockStorage.
-func (b *BlockStorage) NewDatabaseTransaction(
+func (b *BlockStorage) newDatabaseTransaction(
 	ctx context.Context,
 	write bool,
 ) DatabaseTransaction {
@@ -181,8 +174,10 @@ func (b *BlockStorage) NewDatabaseTransaction(
 // if it exists.
 func (b *BlockStorage) GetHeadBlockIdentifier(
 	ctx context.Context,
-	transaction DatabaseTransaction,
 ) (*types.BlockIdentifier, error) {
+	transaction := b.newDatabaseTransaction(ctx, false)
+	defer transaction.Discard(ctx)
+
 	exists, block, err := transaction.Get(ctx, getHeadBlockKey())
 	if err != nil {
 		return nil, err
@@ -221,9 +216,11 @@ func (b *BlockStorage) StoreHeadBlockIdentifier(
 // GetBlock returns a block, if it exists.
 func (b *BlockStorage) GetBlock(
 	ctx context.Context,
-	transaction DatabaseTransaction,
 	blockIdentifier *types.BlockIdentifier,
 ) (*types.Block, error) {
+	transaction := b.newDatabaseTransaction(ctx, false)
+	defer transaction.Discard(ctx)
+
 	exists, block, err := transaction.Get(ctx, getBlockKey(blockIdentifier))
 	if err != nil {
 		return nil, err
@@ -279,36 +276,56 @@ func (b *BlockStorage) storeHash(
 // its transaction hashes for duplicate detection.
 func (b *BlockStorage) StoreBlock(
 	ctx context.Context,
-	transaction DatabaseTransaction,
 	block *types.Block,
-) error {
+) ([]*reconciler.BalanceChange, error) {
+	transaction := b.newDatabaseTransaction(ctx, true)
+	defer transaction.Discard(ctx)
 	buf := new(bytes.Buffer)
 	err := gob.NewEncoder(buf).Encode(block)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
 	// Store block
 	err = transaction.Set(ctx, getBlockKey(block.BlockIdentifier), buf.Bytes())
 	if err != nil {
-		return err
+		return nil, err
 	}
 
 	// Store block hash
 	err = b.storeHash(ctx, transaction, block.BlockIdentifier.Hash, true)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
 	// Store all transaction hashes
 	for _, txn := range block.Transactions {
 		err = b.storeHash(ctx, transaction, txn.TransactionIdentifier.Hash, false)
 		if err != nil {
-			return err
+			return nil, err
 		}
 	}
 
-	return nil
+	changes, err := b.BalanceChanges(ctx, block, false)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, change := range changes {
+		if err := b.UpdateBalance(ctx, transaction, change, block.ParentBlockIdentifier); err != nil {
+			return nil, err
+		}
+	}
+
+	if err = b.StoreHeadBlockIdentifier(ctx, transaction, block.BlockIdentifier); err != nil {
+		return nil, err
+	}
+
+	if err := transaction.Commit(ctx); err != nil {
+		return nil, err
+	}
+
+	return changes, nil
 }
 
 // RemoveBlock removes a block or returns an error.
@@ -317,35 +334,60 @@ func (b *BlockStorage) StoreBlock(
 // detection. This is called within a re-org.
 func (b *BlockStorage) RemoveBlock(
 	ctx context.Context,
-	transaction DatabaseTransaction,
-	block *types.BlockIdentifier,
-) error {
-	// Remove all transaction hashes
-	blockData, err := b.GetBlock(ctx, transaction, block)
+	blockIdentifier *types.BlockIdentifier,
+) ([]*reconciler.BalanceChange, error) {
+	block, err := b.GetBlock(ctx, blockIdentifier)
 	if err != nil {
-		return err
+		return nil, err
+	}
+
+	transaction := b.newDatabaseTransaction(ctx, true)
+	defer transaction.Discard(ctx)
+
+	changes, err := b.BalanceChanges(ctx, block, true)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, change := range changes {
+		if err := b.UpdateBalance(ctx, transaction, change, block.BlockIdentifier); err != nil {
+			return nil, err
+		}
 	}
 
-	for _, txn := range blockData.Transactions {
+	// Remove all transaction hashes
+	for _, txn := range block.Transactions {
 		err = transaction.Delete(ctx, getHashKey(txn.TransactionIdentifier.Hash, false))
 		if err != nil {
-			return err
+			return nil, err
 		}
 	}
 
 	// Remove block hash
-	err = transaction.Delete(ctx, getHashKey(block.Hash, true))
+	err = transaction.Delete(ctx, getHashKey(block.BlockIdentifier.Hash, true))
 	if err != nil {
-		return err
+		return nil, err
 	}
 
 	// Remove block
-	return transaction.Delete(ctx, getBlockKey(block))
+	if err := transaction.Delete(ctx, getBlockKey(block.BlockIdentifier)); err != nil {
+		return nil, err
+	}
+
+	if err = b.StoreHeadBlockIdentifier(ctx, transaction, block.ParentBlockIdentifier); err != nil {
+		return nil, err
+	}
+
+	if err := transaction.Commit(ctx); err != nil {
+		return nil, err
+	}
+
+	return changes, nil
 }
 
 type balanceEntry struct {
-	Amounts map[string]*types.Amount
-	Block   *types.BlockIdentifier
+	Amount *types.Amount
+	Block  *types.BlockIdentifier
 }
 
 func serializeBalanceEntry(bal balanceEntry) ([]byte, error) {
@@ -369,37 +411,70 @@ func parseBalanceEntry(buf []byte) (*balanceEntry, error) {
 	return &bal, nil
 }
 
-// GetCurrencyKey is used to identify a *types.Currency
-// in an account's map of currencies. It is not feasible
-// to create a map of [types.Currency]*types.Amount
-// because types.Currency contains a metadata pointer
-// that would prevent any equality.
-func GetCurrencyKey(currency *types.Currency) string {
-	if currency.Metadata == nil {
-		return hashString(
-			fmt.Sprintf("%s:%d", currency.Symbol, currency.Decimals),
+// SetNewStartIndex attempts to remove all blocks
+// greater than or equal to the startIndex.
+func (b *BlockStorage) SetNewStartIndex(
+	ctx context.Context,
+	startIndex int64,
+) error {
+	head, err := b.GetHeadBlockIdentifier(ctx)
+	if errors.Is(err, ErrHeadBlockNotFound) {
+		return nil
+	}
+	if err != nil {
+		return err
+	}
+
+	if head.Index < startIndex {
+		return fmt.Errorf(
+			"last processed block %d is less than start index %d",
+			head.Index,
+			startIndex,
 		)
 	}
 
-	// TODO: Handle currency.Metadata
-	// that has pointer value.
-	return hashString(
-		fmt.Sprintf(
-			"%s:%d:%v",
-			currency.Symbol,
-			currency.Decimals,
-			currency.Metadata,
-		),
-	)
+	currBlock := head
+	for currBlock.Index >= startIndex {
+		log.Printf("Removing block %+v\n", currBlock)
+		block, err := b.GetBlock(ctx, currBlock)
+		if err != nil {
+			return err
+		}
+
+		if _, err = b.RemoveBlock(ctx, block.BlockIdentifier); err != nil {
+			return err
+		}
+
+		currBlock = block.ParentBlockIdentifier
+	}
+
+	return nil
 }
 
-// BalanceChange represents a balance change that affected
-// a *types.AccountIdentifier and a *types.Currency.
-type BalanceChange struct {
-	Account    *types.AccountIdentifier `json:"account_identifier,omitempty"`
-	Currency   *types.Currency          `json:"currency,omitempty"`
-	Block      *types.BlockIdentifier   `json:"block_identifier,omitempty"`
-	Difference string                   `json:"difference,omitempty"`
+// SetBalance allows a client to set the balance of an account in a database
+// transaction. This is particularly useful for bootstrapping balances.
+func (b *BlockStorage) SetBalance(
+	ctx context.Context,
+	dbTransaction DatabaseTransaction,
+	account *types.AccountIdentifier,
+	amount *types.Amount,
+	block *types.BlockIdentifier,
+) error {
+	key := GetBalanceKey(account, amount.Currency)
+
+	serialBal, err := serializeBalanceEntry(balanceEntry{
+		Amount: amount,
+		Block:  block,
+	})
+	if err != nil {
+		return err
+	}
+
+	if err := dbTransaction.Set(ctx, key, serialBal); err != nil {
+		return err
+	}
+
+	return nil
 }
 
 // UpdateBalance updates a types.AccountIdentifer
@@ -408,126 +483,118 @@ type BalanceChange struct {
 func (b *BlockStorage) UpdateBalance(
 	ctx context.Context,
 	dbTransaction DatabaseTransaction,
-	account *types.AccountIdentifier,
-	amount *types.Amount,
-	block *types.BlockIdentifier,
-) (*BalanceChange, error) {
-	if amount == nil || amount.Currency == nil {
-		return nil, errors.New("invalid amount")
+	change *reconciler.BalanceChange,
+	parentBlock *types.BlockIdentifier,
+) error {
+	if change.Currency == nil {
+		return errors.New("invalid currency")
 	}
 
-	key := GetAccountKey(account)
+	key := GetBalanceKey(change.Account, change.Currency)
 	// Get existing balance on key
 	exists, balance, err := dbTransaction.Get(ctx, key)
 	if err != nil {
-		return nil, err
+		return err
 	}
 
-	// TODO: create a balance key that is the combination
-	// of account and currency
-	currencyKey := GetCurrencyKey(amount.Currency)
-
-	if !exists {
-		amountMap := make(map[string]*types.Amount)
-		newVal, ok := new(big.Int).SetString(amount.Value, 10)
-		if !ok {
-			return nil, fmt.Errorf("%s is not an integer", amount.Value)
-		}
-
-		if newVal.Sign() == -1 {
-			return nil, fmt.Errorf(
-				"%w %+v for %+v at %+v",
-				ErrNegativeBalance,
-				spew.Sdump(amount),
-				account,
-				block,
-			)
-		}
-		amountMap[currencyKey] = amount
-
-		serialBal, err := serializeBalanceEntry(balanceEntry{
-			Amounts: amountMap,
-			Block:   block,
-		})
+	var existingValue string
+	if exists {
+		parseBal, err := parseBalanceEntry(balance)
 		if err != nil {
-			return nil, err
+			return err
 		}
 
-		if err := dbTransaction.Set(ctx, key, serialBal); err != nil {
-			return nil, err
+		existingValue = parseBal.Amount.Value
+	} else {
+		amount, err := b.helper.AccountBalance(ctx, change.Account, change.Currency, parentBlock)
+		if err != nil {
+			return fmt.Errorf("%w: unable to get previous account balance", err)
 		}
 
-		return &BalanceChange{
-			Account:    account,
-			Currency:   amount.Currency,
-			Block:      block,
-			Difference: amount.Value,
-		}, nil
+		existingValue = amount.Value
 	}
 
-	// Modify balance
-	parseBal, err := parseBalanceEntry(balance)
+	newVal, err := types.AddValues(change.Difference, existingValue)
 	if err != nil {
-		return nil, err
+		return err
 	}
 
-	val, ok := parseBal.Amounts[currencyKey]
+	bigNewVal, ok := new(big.Int).SetString(newVal, 10)
 	if !ok {
-		parseBal.Amounts[currencyKey] = amount
+		return fmt.Errorf("%s is not an integer", newVal)
 	}
 
-	oldValue := val.Value
-	val.Value, err = AddStringValues(amount.Value, oldValue)
-	if err != nil {
-		return nil, err
-	}
-
-	if strings.HasPrefix(val.Value, "-") {
-		return nil, fmt.Errorf(
-			"%w %+v for %+v at %+v",
+	if bigNewVal.Sign() == -1 {
+		return fmt.Errorf(
+			"%w %s:%+v for %+v at %+v",
 			ErrNegativeBalance,
-			spew.Sdump(val),
-			account,
-			block,
+			newVal,
+			change.Currency,
+			change.Account,
+			change.Block,
 		)
 	}
 
-	parseBal.Amounts[currencyKey] = val
-
-	parseBal.Block = block
-	serialBal, err := serializeBalanceEntry(*parseBal)
+	serialBal, err := serializeBalanceEntry(balanceEntry{
+		Amount: &types.Amount{
+			Value:    newVal,
+			Currency: change.Currency,
+		},
+		Block: change.Block,
+	})
 	if err != nil {
-		return nil, err
-	}
-
-	if err := dbTransaction.Set(ctx, key, serialBal); err != nil {
-		return nil, err
+		return err
 	}
 
-	return &BalanceChange{
-		Account:    account,
-		Currency:   amount.Currency,
-		Block:      block,
-		Difference: amount.Value,
-	}, nil
+	return dbTransaction.Set(ctx, key, serialBal)
 }
 
 // GetBalance returns all the balances of a types.AccountIdentifier
 // and the types.BlockIdentifier it was last updated at.
-// TODO: change to fetch by account and currency
 func (b *BlockStorage) GetBalance(
 	ctx context.Context,
-	transaction DatabaseTransaction,
 	account *types.AccountIdentifier,
-) (map[string]*types.Amount, *types.BlockIdentifier, error) {
-	key := GetAccountKey(account)
+	currency *types.Currency,
+	headBlock *types.BlockIdentifier,
+) (*types.Amount, *types.BlockIdentifier, error) {
+	transaction := b.newDatabaseTransaction(ctx, false)
+	defer transaction.Discard(ctx)
+
+	key := GetBalanceKey(account, currency)
 	exists, bal, err := transaction.Get(ctx, key)
 	if err != nil {
 		return nil, nil, err
 	}
 
+	// When beginning syncing from an arbitrary height, an account may
+	// not yet have a cached balance when requested. If this is the case,
+	// we fetch the balance from the node for the given height and persist
+	// it. This is particularly useful when monitoring interesting accounts.
 	if !exists {
-		return nil, nil, fmt.Errorf("%w %+v", ErrAccountNotFound, account)
+		amount, err := b.helper.AccountBalance(ctx, account, currency, headBlock)
+		if err != nil {
+			return nil, nil, fmt.Errorf("%w: unable to get account balance from helper", err)
+		}
+
+		writeTransaction := b.newDatabaseTransaction(ctx, true)
+		defer writeTransaction.Discard(ctx)
+		err = b.SetBalance(
+			ctx,
+			writeTransaction,
+			account,
+			amount,
+			headBlock,
+		)
+		if err != nil {
+			return nil, nil, fmt.Errorf("%w: unable to set account balance", err)
+		}
+
+		err = writeTransaction.Commit(ctx)
+		if err != nil {
+			return nil, nil, fmt.Errorf("%w: unable to commit account balance transaction", err)
+		}
+
+		return amount, headBlock, nil
 	}
 
 	deserialBal, err := parseBalanceEntry(bal)
@@ -535,12 +602,13 @@ func (b *BlockStorage) GetBalance(
 		return nil, nil, err
 	}
 
-	return deserialBal.Amounts, deserialBal.Block, nil
+	return deserialBal.Amount, deserialBal.Block, nil
 }
 
 // BootstrapBalance represents a balance of
 // a *types.AccountIdentifier and a *types.Currency in the
 // genesis block.
+// TODO: Must be exported for use
 type BootstrapBalance struct {
 	Account  *types.AccountIdentifier `json:"account_identifier,omitempty"`
 	Currency *types.Currency          `json:"currency,omitempty"`
@@ -567,15 +635,15 @@ func (b *BlockStorage) BootstrapBalances(
 		return err
 	}
 
-	// Update balances in database
-	dbTransaction := b.NewDatabaseTransaction(ctx, true)
-	defer dbTransaction.Discard(ctx)
-
-	_, err = b.GetHeadBlockIdentifier(ctx, dbTransaction)
+	_, err = b.GetHeadBlockIdentifier(ctx)
 	if err != ErrHeadBlockNotFound {
 		return ErrAlreadyStartedSyncing
 	}
 
+	// Update balances in database
+	dbTransaction := b.newDatabaseTransaction(ctx, true)
+	defer dbTransaction.Discard(ctx)
+
 	for _, balance := range balances {
 		// Ensure change.Difference is valid
 		amountValue, ok := new(big.Int).SetString(balance.Value, 10)
@@ -594,7 +662,7 @@ func (b *BlockStorage) BootstrapBalances(
 			balance.Currency,
 		)
 
-		_, err = b.UpdateBalance(
+		err = b.SetBalance(
 			ctx,
 			dbTransaction,
 			balance.Account,
@@ -617,3 +685,98 @@ func (b *BlockStorage) BootstrapBalances(
 	log.Printf("%d Balances Bootstrapped\n", len(balances))
 	return nil
 }
+
+// BalanceChanges returns all balance changes for
+// a particular block. All balance changes for a
+// particular account are summed into a single
+// storage.BalanceChanges struct. If a block is being
+// orphaned, the opposite of each balance change is
+// returned.
+func (b *BlockStorage) BalanceChanges(
+	ctx context.Context,
+	block *types.Block,
+	blockRemoved bool,
+) ([]*reconciler.BalanceChange, error) {
+	balanceChanges := map[string]*reconciler.BalanceChange{}
+	for _, tx := range block.Transactions {
+		for _, op := range tx.Operations {
+			skip, err := b.helper.SkipOperation(
+				ctx,
+				op,
+			)
+			if err != nil {
+				return nil, err
+			}
+			if skip {
+				continue
+			}
+
+			amount := op.Amount
+			blockIdentifier := block.BlockIdentifier
+			if blockRemoved {
+				existing, ok := new(big.Int).SetString(amount.Value, 10)
+				if !ok {
+					return nil, fmt.Errorf("%s is not an integer", amount.Value)
+				}
+
+				amount.Value = new(big.Int).Neg(existing).String()
+				blockIdentifier = block.ParentBlockIdentifier
+			}
+
+			// Merge values by account and currency
+			key := fmt.Sprintf(
+				"%x",
+				GetBalanceKey(op.Account, op.Amount.Currency),
+			)
+
+			val, ok := balanceChanges[key]
+			if !ok {
+				balanceChanges[key] = &reconciler.BalanceChange{
+					Account:    op.Account,
+					Currency:   op.Amount.Currency,
+					Difference: amount.Value,
+					Block:      blockIdentifier,
+				}
+				continue
+			}
+
+			newDifference, err := types.AddValues(val.Difference, amount.Value)
+			if err != nil {
+				return nil, err
+			}
+			val.Difference = newDifference
+			balanceChanges[key] = val
+		}
+	}
+
+	allChanges := []*reconciler.BalanceChange{}
+	for _, change := range balanceChanges {
+		allChanges = append(allChanges, change)
+	}
+
+	return allChanges, nil
+}
+
+// CreateBlockCache populates a slice of blocks with the most recent
+// ones in storage.
+func (b *BlockStorage) CreateBlockCache(ctx context.Context) []*types.BlockIdentifier {
+	cache := []*types.BlockIdentifier{}
+	head, err := b.GetHeadBlockIdentifier(ctx)
+	if err != nil {
+		return cache
+	}
+
+	for len(cache) < syncer.PastBlockSize {
+		block, err := b.GetBlock(ctx, head)
+		if err != nil {
+			return cache
+		}
+
+		log.Printf("Added %+v to cache\n", block.BlockIdentifier)
+
+		cache = append([]*types.BlockIdentifier{block.BlockIdentifier}, cache...)
+		head = block.ParentBlockIdentifier
+	}
+
+	return cache
+}
diff --git a/internal/storage/utils.go b/internal/storage/utils.go
deleted file mode 100644
index d3aed5d..0000000
--- a/internal/storage/utils.go
+++ /dev/null
@@ -1,82 +0,0 @@
-// Copyright 2020 Coinbase, Inc.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//      http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package storage
-
-import (
-	"fmt"
-	"io/ioutil"
-	"log"
-	"math/big"
-	"os"
-)
-
-// CreateTempDir creates a directory in
-// /tmp for usage within testing.
-func CreateTempDir() (*string, error) {
-	storageDir, err := ioutil.TempDir("", "rosetta-worker")
-	if err != nil {
-		return nil, err
-	}
-
-	return &storageDir, nil
-}
-
-// RemoveTempDir deletes a directory at
-// a provided path for usage within testing.
-func RemoveTempDir(dir string) {
-	if err := os.RemoveAll(dir); err != nil {
-		log.Fatal(err)
-	}
-}
-
-// AddStringValues adds string amounts using
-// big.Int.
-func AddStringValues(
-	a string,
-	b string,
-) (string, error) {
-	aVal, ok := new(big.Int).SetString(a, 10)
-	if !ok {
-		return "", fmt.Errorf("%s is not an integer", a)
-	}
-
-	bVal, ok := new(big.Int).SetString(b, 10)
-	if !ok {
-		return "", fmt.Errorf("%s is not an integer", b)
-	}
-
-	newVal := new(big.Int).Add(aVal, bVal)
-	return newVal.String(), nil
-}
-
-// SubtractStringValues subtracts a-b using
-// big.Int.
-func SubtractStringValues(
-	a string,
-	b string,
-) (string, error) {
-	aVal, ok := new(big.Int).SetString(a, 10)
-	if !ok {
-		return "", fmt.Errorf("%s is not an integer", a)
-	}
-
-	bVal, ok := new(big.Int).SetString(b, 10)
-	if !ok {
-		return "", fmt.Errorf("%s is not an integer", b)
-	}
-
-	newVal := new(big.Int).Sub(aVal, bVal)
-	return newVal.String(), nil
-}
diff --git a/internal/syncer/base_handler.go b/internal/syncer/base_handler.go
deleted file mode 100644
index 136beaa..0000000
--- a/internal/syncer/base_handler.go
+++ /dev/null
@@ -1,92 +0,0 @@
-// Copyright 2020 Coinbase, Inc.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//      http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package syncer
-
-import (
-	"context"
-	"log"
-
-	"github.com/coinbase/rosetta-cli/internal/logger"
-	"github.com/coinbase/rosetta-cli/internal/reconciler"
-	"github.com/coinbase/rosetta-cli/internal/storage"
-
-	"github.com/coinbase/rosetta-sdk-go/types"
-)
-
-// BaseHandler logs processed blocks
-// and reconciles modified balances.
-type BaseHandler struct {
-	logger         *logger.Logger
-	reconciler     reconciler.Reconciler
-	exemptAccounts []*reconciler.AccountCurrency
-}
-
-// NewBaseHandler constructs a basic Handler.
-func NewBaseHandler(
-	logger *logger.Logger,
-	reconciler reconciler.Reconciler,
-	exemptAccounts []*reconciler.AccountCurrency,
-) Handler {
-	return &BaseHandler{
-		logger:         logger,
-		reconciler:     reconciler,
-		exemptAccounts: exemptAccounts,
-	}
-}
-
-// BlockProcessed is called by the syncer after each
-// block is processed.
-// TODO: refactor to BlockAdded and BlockRemoved
-func (h *BaseHandler) BlockProcessed(
-	ctx context.Context,
-	block *types.Block,
-	reorg bool,
-	balanceChanges []*storage.BalanceChange,
-) error {
-	if !reorg {
-		log.Printf("Adding block %+v\n", block.BlockIdentifier)
-	} else {
-		log.Printf("Orphaning block %+v\n", block.BlockIdentifier)
-	}
-	// Log processed blocks and balance changes
-	if err := h.logger.BlockStream(ctx, block, reorg); err != nil {
-		return nil
-	}
-
-	if err := h.logger.BalanceStream(ctx, balanceChanges); err != nil {
-		return nil
-	}
-
-	// Mark accounts for reconciliation...this may be
-	// blocking
-	return h.reconciler.QueueChanges(ctx, block.BlockIdentifier, balanceChanges)
-}
-
-// AccountExempt returns a boolean indicating if the provided
-// account and currency are exempt from balance tracking and
-// reconciliation.
-func (h *BaseHandler) AccountExempt(
-	ctx context.Context,
-	account *types.AccountIdentifier,
-	currency *types.Currency,
-) bool {
-	return reconciler.ContainsAccountCurrency(
-		h.exemptAccounts,
-		&reconciler.AccountCurrency{
-			Account:  account,
-			Currency: currency,
-		},
-	)
-}
diff --git a/internal/syncer/stateful_syncer.go b/internal/syncer/stateful_syncer.go
deleted file mode 100644
index 52d0d44..0000000
--- a/internal/syncer/stateful_syncer.go
+++ /dev/null
@@ -1,394 +0,0 @@
-// Copyright 2020 Coinbase, Inc.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//      http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package syncer
-
-import (
-	"context"
-	"errors"
-	"fmt"
-
-	"github.com/coinbase/rosetta-cli/internal/storage"
-
-	"github.com/coinbase/rosetta-sdk-go/fetcher"
-	"github.com/coinbase/rosetta-sdk-go/types"
-)
-
-// StatefulSyncer contains the logic that orchestrates
-// block fetching, storage, and reconciliation. The
-// stateful syncer is useful for creating an application
-// where durability and consistency of data is important.
-// The stateful syncer supports re-orgs out-of the box.
-type StatefulSyncer struct {
-	network      *types.NetworkIdentifier
-	storage      *storage.BlockStorage
-	fetcher      *fetcher.Fetcher
-	handler      Handler
-	genesisBlock *types.BlockIdentifier
-}
-
-// NewStateful returns a new Syncer.
-func NewStateful(
-	network *types.NetworkIdentifier,
-	storage *storage.BlockStorage,
-	fetcher *fetcher.Fetcher,
-	handler Handler,
-) *StatefulSyncer {
-	return &StatefulSyncer{
-		network: network,
-		storage: storage,
-		fetcher: fetcher,
-		handler: handler,
-	}
-}
-
-// SetStartIndex initializes the genesisBlock
-// and attempts to set the newHeadIndex.
-func (s *StatefulSyncer) SetStartIndex(
-	ctx context.Context,
-	startIndex int64,
-) error {
-	networkStatus, err := s.fetcher.NetworkStatusRetry(
-		ctx,
-		s.network,
-		nil,
-	)
-	if err != nil {
-		return err
-	}
-
-	s.genesisBlock = networkStatus.GenesisBlockIdentifier
-
-	if startIndex != -1 {
-		return s.newHeadIndex(ctx, startIndex)
-	}
-
-	return nil
-}
-
-// checkReorg determines if the block provided
-// has the current head block identifier as its
-// parent. If not, it is considered a reorg.
-func (s *StatefulSyncer) checkReorg(
-	ctx context.Context,
-	tx storage.DatabaseTransaction,
-	block *types.Block,
-) (bool, error) {
-	head, err := s.storage.GetHeadBlockIdentifier(ctx, tx)
-	if err == storage.ErrHeadBlockNotFound {
-		return false, nil
-	} else if err != nil {
-		return false, err
-	}
-
-	if block.ParentBlockIdentifier.Index != head.Index {
-		return false, fmt.Errorf(
-			"Got block %d instead of %d",
-			block.BlockIdentifier.Index,
-			head.Index+1,
-		)
-	}
-
-	if block.ParentBlockIdentifier.Hash != head.Hash {
-		return true, nil
-	}
-
-	return false, nil
-}
-
-// storeBlockBalanceChanges updates the balance
-// of each modified account if the operation affecting
-// that account is successful. These modified
-// accounts are returned to the reconciler
-// for active reconciliation.
-func (s *StatefulSyncer) storeBlockBalanceChanges(
-	ctx context.Context,
-	dbTx storage.DatabaseTransaction,
-	block *types.Block,
-	orphan bool,
-) ([]*storage.BalanceChange, error) {
-	balanceChanges := make([]*storage.BalanceChange, 0)
-
-	// Merge all changes for an account:currency
-	mergedChanges, err := BalanceChanges(
-		ctx,
-		s.fetcher.Asserter,
-		block,
-		orphan,
-		s.handler,
-	)
-	if err != nil {
-		return nil, err
-	}
-
-	for _, change := range mergedChanges {
-		balanceChange, err := s.storage.UpdateBalance(
-			ctx,
-			dbTx,
-			change.Account,
-			&types.Amount{
-				Value:    change.Difference,
-				Currency: change.Currency,
-			},
-			change.Block,
-		)
-		if err != nil {
-			return nil, err
-		}
-
-		balanceChanges = append(balanceChanges, balanceChange)
-	}
-
-	return balanceChanges, nil
-}
-
-// orphanBlock removes a block from the database and reverts all its balance
-// changes.
-func (s *StatefulSyncer) orphanBlock(
-	ctx context.Context,
-	tx storage.DatabaseTransaction,
-	blockIdentifier *types.BlockIdentifier,
-) ([]*storage.BalanceChange, error) {
-	block, err := s.storage.GetBlock(ctx, tx, blockIdentifier)
-	if err != nil {
-		return nil, err
-	}
-
-	err = s.storage.StoreHeadBlockIdentifier(ctx, tx, block.ParentBlockIdentifier)
-	if err != nil {
-		return nil, err
-	}
-
-	balanceChanges, err := s.storeBlockBalanceChanges(ctx, tx, block, true)
-	if err != nil {
-		return nil, err
-	}
-
-	err = s.storage.RemoveBlock(ctx, tx, blockIdentifier)
-	if err != nil {
-		return nil, err
-	}
-
-	return balanceChanges, nil
-}
-
-// addBlock adds a block to the database and stores all balance changes.
-func (s *StatefulSyncer) addBlock(
-	ctx context.Context,
-	tx storage.DatabaseTransaction,
-	block *types.Block,
-) ([]*storage.BalanceChange, error) {
-	err := s.storage.StoreBlock(ctx, tx, block)
-	if err != nil {
-		return nil, err
-	}
-
-	err = s.storage.StoreHeadBlockIdentifier(ctx, tx, block.BlockIdentifier)
-	if err != nil {
-		return nil, err
-	}
-
-	balanceChanges, err := s.storeBlockBalanceChanges(ctx, tx, block, false)
-	if err != nil {
-		return nil, err
-	}
-
-	return balanceChanges, nil
-}
-
-// processBlock determines if a block should be added or the current
-// head should be orphaned.
-func (s *StatefulSyncer) processBlock(
-	ctx context.Context,
-	genesisIndex int64,
-	currIndex int64,
-	block *types.Block,
-) ([]*storage.BalanceChange, int64, bool, error) {
-	tx := s.storage.NewDatabaseTransaction(ctx, true)
-	defer tx.Discard(ctx)
-
-	reorg, err := s.checkReorg(ctx, tx, block)
-	if err != nil {
-		return nil, currIndex, false, err
-	}
-
-	var balanceChanges []*storage.BalanceChange
-	var newIndex int64
-	if reorg {
-		newIndex = currIndex - 1
-		if newIndex == genesisIndex {
-			return nil, 0, false, errors.New("cannot orphan genesis block")
-		}
-
-		head, err := s.storage.GetHeadBlockIdentifier(ctx, tx)
-		if err != nil {
-			return nil, currIndex, false, err
-		}
-
-		balanceChanges, err = s.orphanBlock(ctx, tx, head)
-		if err != nil {
-			return nil, currIndex, false, err
-		}
-	} else {
-		balanceChanges, err = s.addBlock(ctx, tx, block)
-		if err != nil {
-			return nil, currIndex, false, err
-		}
-
-		newIndex = currIndex + 1
-	}
-
-	err = tx.Commit(ctx)
-	if err != nil {
-		return nil, currIndex, false, err
-	}
-
-	return balanceChanges, newIndex, reorg, nil
-}
-
-// newHeadIndex reverts all blocks that have
-// an index greater than newHeadIndex. This is particularly
-// useful when debugging a server implementation because
-// you don't need to restart validation from genesis. Instead,
-// you can just restart validation at the block immediately
-// before any erroneous block.
-func (s *StatefulSyncer) newHeadIndex(
-	ctx context.Context,
-	newHeadIndex int64,
-) error {
-	tx := s.storage.NewDatabaseTransaction(ctx, true)
-	defer tx.Discard(ctx)
-
-	for {
-		head, err := s.storage.GetHeadBlockIdentifier(ctx, tx)
-		if err == storage.ErrHeadBlockNotFound {
-			return fmt.Errorf(
-				"cannot start syncing at %d, have not yet processed any blocks",
-				newHeadIndex,
-			)
-		} else if err != nil {
-			return err
-		}
-
-		if head.Index < newHeadIndex {
-			return fmt.Errorf(
-				"cannot start syncing at %d, have only processed %d blocks",
-				newHeadIndex,
-				head.Index,
-			)
-		}
-
-		if head.Index == newHeadIndex {
-			break
-		}
-
-		_, err = s.orphanBlock(ctx, tx, head)
-		if err != nil {
-			return err
-		}
-	}
-
-	return tx.Commit(ctx)
-}
-
-// SyncRange syncs blocks from startIndex to endIndex, inclusive.
-// This function handles re-orgs that may occur while syncing as long
-// as the genesisIndex is not orphaned.
-func (s *StatefulSyncer) SyncRange(
-	ctx context.Context,
-	startIndex int64,
-	endIndex int64,
-) error {
-	blockMap, err := s.fetcher.BlockRange(ctx, s.network, startIndex, endIndex)
-	if err != nil {
-		return err
-	}
-
-	currIndex := startIndex
-	for currIndex <= endIndex {
-		block, ok := blockMap[currIndex]
-		if !ok { // could happen in a reorg
-			block, err = s.fetcher.BlockRetry(
-				ctx,
-				s.network,
-				&types.PartialBlockIdentifier{
-					Index: &currIndex,
-				},
-			)
-			if err != nil {
-				return err
-			}
-		} else {
-			// Anytime we re-fetch an index, we
-			// will need to make another call to the node
-			// as it is likely in a reorg.
-			delete(blockMap, currIndex)
-		}
-
-		// Can't return balanceChanges without creating new variable
-		balanceChanges, newIndex, reorg, err := s.processBlock(
-			ctx,
-			s.genesisBlock.Index,
-			currIndex,
-			block,
-		)
-		if err != nil {
-			return err
-		}
-
-		currIndex = newIndex
-
-		if err := s.handler.BlockProcessed(ctx, block, reorg, balanceChanges); err != nil {
-			return err
-		}
-	}
-
-	return nil
-}
-
-// CurrentIndex returns the next index to sync.
-func (s *StatefulSyncer) CurrentIndex(
-	ctx context.Context,
-) (int64, error) {
-	tx := s.storage.NewDatabaseTransaction(ctx, false)
-	defer tx.Discard(ctx)
-
-	var currentIndex int64
-	head, err := s.storage.GetHeadBlockIdentifier(ctx, tx)
-	switch err {
-	case nil:
-		currentIndex = head.Index + 1
-	case storage.ErrHeadBlockNotFound:
-		head = s.genesisBlock
-		currentIndex = head.Index
-	default:
-		return -1, err
-	}
-
-	return currentIndex, nil
-}
-
-// Network returns the syncer network.
-func (s *StatefulSyncer) Network(
-	ctx context.Context,
-) *types.NetworkIdentifier {
-	return s.network
-}
-
-// Fetcher returns the syncer fetcher.
-func (s *StatefulSyncer) Fetcher(
-	ctx context.Context,
-) *fetcher.Fetcher {
-	return s.fetcher
-}
diff --git a/internal/syncer/stateless_syncer.go b/internal/syncer/stateless_syncer.go
deleted file mode 100644
index 66e5f76..0000000
--- a/internal/syncer/stateless_syncer.go
+++ /dev/null
@@ -1,139 +0,0 @@
-// Copyright 2020 Coinbase, Inc.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//      http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package syncer
-
-import (
-	"context"
-
-	"github.com/coinbase/rosetta-sdk-go/fetcher"
-	"github.com/coinbase/rosetta-sdk-go/types"
-)
-
-// StatelessSyncer contains the logic that orchestrates
-// stateless block fetching and reconciliation. The stateless
-// syncer is useful for performing a quick check over a range of
-// blocks without needed to sync all blocks up to the start of
-// the range (a common pattern when debugging). It is important
-// to note that the stateless syncer does not support reorgs nor
-// does it save where it is on restart.
-type StatelessSyncer struct {
-	network      *types.NetworkIdentifier
-	fetcher      *fetcher.Fetcher
-	handler      Handler
-	currentIndex int64
-}
-
-// NewStateless returns a new Syncer.
-func NewStateless(
-	network *types.NetworkIdentifier,
-	fetcher *fetcher.Fetcher,
-	handler Handler,
-) *StatelessSyncer {
-	return &StatelessSyncer{
-		network: network,
-		fetcher: fetcher,
-		handler: handler,
-	}
-}
-
-// SetStartIndex initializes the current block index
-// with the genesis block index if it is -1.
-func (s *StatelessSyncer) SetStartIndex(
-	ctx context.Context,
-	startIndex int64,
-) error {
-	if startIndex != -1 {
-		s.currentIndex = startIndex
-		return nil
-	}
-
-	// Sync from genesis + 1
-	networkStatus, err := s.fetcher.NetworkStatusRetry(
-		ctx,
-		s.network,
-		nil,
-	)
-	if err != nil {
-		return err
-	}
-
-	// Don't sync genesis block because balance lookup will not
-	// work.
-	s.currentIndex = networkStatus.GenesisBlockIdentifier.Index + 1
-	return nil
-}
-
-// SyncRange syncs blocks from startIndex to endIndex, inclusive.
-// This function does NOT handle re-orgs. If you want re-org support,
-// checkout the StatefulSyncer.
-func (s *StatelessSyncer) SyncRange(
-	ctx context.Context,
-	startIndex int64,
-	endIndex int64,
-) error {
-	blockMap, err := s.fetcher.BlockRange(ctx, s.network, startIndex, endIndex)
-	if err != nil {
-		return err
-	}
-
-	for i := startIndex; i <= endIndex; i++ {
-		block := blockMap[i]
-		changes, err := BalanceChanges(
-			ctx,
-			s.fetcher.Asserter,
-			block,
-			false,
-			s.handler,
-		)
-		if err != nil {
-			return err
-		}
-
-		err = s.handler.BlockProcessed(
-			ctx,
-			block,
-			false,
-			changes,
-		)
-		if err != nil {
-			return err
-		}
-	}
-
-	s.currentIndex = endIndex + 1
-
-	return nil
-}
-
-// CurrentIndex returns the next index to sync.
-func (s *StatelessSyncer) CurrentIndex(
-	ctx context.Context,
-) (int64, error) {
-	return s.currentIndex, nil
-}
-
-// Network returns the syncer network.
-func (s *StatelessSyncer) Network(
-	ctx context.Context,
-) *types.NetworkIdentifier {
-	return s.network
-}
-
-// Fetcher returns the syncer fetcher.
-func (s *StatelessSyncer) Fetcher(
-	ctx context.Context,
-) *fetcher.Fetcher {
-	return s.fetcher
-}
diff --git a/internal/syncer/syncer.go b/internal/syncer/syncer.go
index 1074cf9..f4b4e8f 100644
--- a/internal/syncer/syncer.go
+++ b/internal/syncer/syncer.go
@@ -16,13 +16,12 @@ package syncer
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"log"
-	"math/big"
 
-	"github.com/coinbase/rosetta-cli/internal/storage"
+	"github.com/coinbase/rosetta-cli/internal/utils"
 
-	"github.com/coinbase/rosetta-sdk-go/asserter"
 	"github.com/coinbase/rosetta-sdk-go/fetcher"
 	"github.com/coinbase/rosetta-sdk-go/types"
 )
@@ -30,239 +29,275 @@ import (
 const (
 	// maxSync is the maximum number of blocks
 	// to try and sync in a given SyncCycle.
-	maxSync = 1000
+	maxSync = 999
+
+	// PastBlockSize is the maximum number of previously
+	// processed blocks we keep in the syncer to handle
+	// reorgs correctly. If there is a reorg greater than
+	// PastBlockSize, it will not be handled correctly.
+	//
+	// TODO: make configurable
+	PastBlockSize = 20
 )
 
-// Syncer defines an interface for syncing some
-// range of blocks.
-type Syncer interface {
-	SetStartIndex(
+// Handler is called at various times during the sync cycle
+// to handle different events. It is common to write logs or
+// perform reconciliation in the sync processor.
+type Handler interface {
+	BlockAdded(
 		ctx context.Context,
-		startIndex int64,
+		block *types.Block,
 	) error
 
-	CurrentIndex(
+	BlockRemoved(
 		ctx context.Context,
-	) (int64, error)
-
-	SyncRange(
-		ctx context.Context,
-		rangeStart int64,
-		rangeEnd int64,
+		block *types.BlockIdentifier,
 	) error
+}
 
-	Network(
-		ctx context.Context,
-	) *types.NetworkIdentifier
+// Syncer coordinates blockchain syncing without relying on
+// a storage interface. Instead, it calls a provided Handler
+// whenever a block is added or removed. This provides the client
+// the opportunity to define the logic used to handle each new block.
+// In the rosetta-cli, we handle reconciliation, state storage, and
+// logging in the handler.
+type Syncer struct {
+	network *types.NetworkIdentifier
+	fetcher *fetcher.Fetcher
+	handler Handler
+	cancel  context.CancelFunc
+
+	// Used to keep track of sync state
+	genesisBlock *types.BlockIdentifier
+	nextIndex    int64
+
+	// To ensure reorgs are handled correctly, the syncer must be able
+	// to observe blocks it has previously processed. Without this, the
+	// syncer may process an index that is not connected to previously added
+	// blocks (ParentBlockIdentifier != lastProcessedBlock.BlockIdentifier).
+	//
+	// If a blockchain does not have reorgs, it is not necessary to populate
+	// the blockCache on creation.
+	pastBlocks []*types.BlockIdentifier
+}
 
-	Fetcher(
-		ctx context.Context,
-	) *fetcher.Fetcher
+// New creates a new Syncer. If pastBlocks is left nil, it will
+// be set to an empty slice.
+func New(
+	network *types.NetworkIdentifier,
+	fetcher *fetcher.Fetcher,
+	handler Handler,
+	cancel context.CancelFunc,
+	pastBlocks []*types.BlockIdentifier,
+) *Syncer {
+	past := pastBlocks
+	if past == nil {
+		past = []*types.BlockIdentifier{}
+	}
+
+	return &Syncer{
+		network:    network,
+		fetcher:    fetcher,
+		handler:    handler,
+		cancel:     cancel,
+		pastBlocks: past,
+	}
+}
+
+func (s *Syncer) setStart(
+	ctx context.Context,
+	index int64,
+) error {
+	networkStatus, err := s.fetcher.NetworkStatusRetry(
+		ctx,
+		s.network,
+		nil,
+	)
+	if err != nil {
+		return err
+	}
+
+	s.genesisBlock = networkStatus.GenesisBlockIdentifier
+
+	if index != -1 {
+		s.nextIndex = index
+		return nil
+	}
+
+	s.nextIndex = networkStatus.GenesisBlockIdentifier.Index
+	return nil
 }
 
-// NextSyncableRange returns the next range of indexes to sync
+// nextSyncableRange returns the next range of indexes to sync
 // based on what the last processed block in storage is and
 // the contents of the network status response.
-func NextSyncableRange(
+func (s *Syncer) nextSyncableRange(
 	ctx context.Context,
-	s Syncer,
 	endIndex int64,
-) (int64, int64, bool, error) {
-	currentIndex, err := s.CurrentIndex(ctx)
-	if err != nil {
-		return -1, -1, false, fmt.Errorf("%w: unable to get current index", err)
+) (int64, bool, error) {
+	if s.nextIndex == -1 {
+		return -1, false, errors.New("unable to get current head")
 	}
 
 	if endIndex == -1 {
-		networkStatus, err := s.Fetcher(ctx).NetworkStatusRetry(
+		networkStatus, err := s.fetcher.NetworkStatusRetry(
 			ctx,
-			s.Network(ctx),
+			s.network,
 			nil,
 		)
 		if err != nil {
-			return -1, -1, false, fmt.Errorf("%w: unable to get network status", err)
+			return -1, false, fmt.Errorf("%w: unable to get network status", err)
 		}
 
-		return currentIndex, networkStatus.CurrentBlockIdentifier.Index, false, nil
+		endIndex = networkStatus.CurrentBlockIdentifier.Index
 	}
 
-	if currentIndex >= endIndex {
-		return -1, -1, true, nil
+	if s.nextIndex >= endIndex {
+		return -1, true, nil
 	}
 
-	return currentIndex, endIndex, false, nil
-}
+	if endIndex-s.nextIndex > maxSync {
+		endIndex = s.nextIndex + maxSync
+	}
 
-// Sync cycles endlessly until there is an error
-// or the requested range is synced.
-func Sync(
-	ctx context.Context,
-	cancel context.CancelFunc,
-	s Syncer,
-	startIndex int64,
-	endIndex int64,
-) error {
-	defer cancel()
+	return endIndex, false, nil
+}
 
-	if err := s.SetStartIndex(ctx, startIndex); err != nil {
-		return fmt.Errorf("%w: unable to set start index", err)
+func (s *Syncer) checkRemove(
+	block *types.Block,
+) (bool, *types.BlockIdentifier, error) {
+	if len(s.pastBlocks) == 0 {
+		return false, nil, nil
 	}
 
-	for {
-		rangeStart, rangeEnd, halt, err := NextSyncableRange(
-			ctx,
-			s,
-			endIndex,
+	// Ensure processing correct index
+	if block.BlockIdentifier.Index != s.nextIndex {
+		return false, nil, fmt.Errorf(
+			"Got block %d instead of %d",
+			block.BlockIdentifier.Index,
+			s.nextIndex,
 		)
-		if err != nil {
-			return fmt.Errorf("%w: unable to get next syncable range", err)
-		}
-		if halt {
-			break
-		}
+	}
 
-		if rangeEnd-rangeStart > maxSync {
-			rangeEnd = rangeStart + maxSync
+	// Check if block parent is head
+	lastBlock := s.pastBlocks[len(s.pastBlocks)-1]
+	if !utils.Equal(block.ParentBlockIdentifier, lastBlock) {
+		if utils.Equal(s.genesisBlock, lastBlock) {
+			return false, nil, fmt.Errorf("cannot remove genesis block")
 		}
 
-		log.Printf("Syncing %d-%d\n", rangeStart, rangeEnd)
+		return true, lastBlock, nil
+	}
+
+	return false, lastBlock, nil
+}
 
-		err = s.SyncRange(ctx, rangeStart, rangeEnd)
+func (s *Syncer) processBlock(
+	ctx context.Context,
+	block *types.Block,
+) error {
+	shouldRemove, lastBlock, err := s.checkRemove(block)
+	if err != nil {
+		return err
+	}
+
+	if shouldRemove {
+		err = s.handler.BlockRemoved(ctx, lastBlock)
 		if err != nil {
-			return fmt.Errorf("%w: unable to sync range %d-%d", err, rangeStart, rangeEnd)
+			return err
 		}
+		s.pastBlocks = s.pastBlocks[:len(s.pastBlocks)-1]
+		s.nextIndex = lastBlock.Index
+		return nil
+	}
 
-		if ctx.Err() != nil {
-			return ctx.Err()
-		}
+	err = s.handler.BlockAdded(ctx, block)
+	if err != nil {
+		return err
 	}
 
-	log.Printf("Finished syncing %d-%d\n", startIndex, endIndex)
+	s.pastBlocks = append(s.pastBlocks, block.BlockIdentifier)
+	if len(s.pastBlocks) > PastBlockSize {
+		s.pastBlocks = s.pastBlocks[1:]
+	}
+	s.nextIndex = block.BlockIdentifier.Index + 1
 	return nil
 }
 
-// Handler is called at various times during the sync cycle
-// to handle different events. It is common to write logs or
-// perform reconciliation in the sync handler.
-type Handler interface {
-	// TODO: change to BlockAdded and BlockRemoved
-	BlockProcessed(
-		ctx context.Context,
-		block *types.Block,
-		orphan bool,
-		changes []*storage.BalanceChange,
-	) error
-
-	AccountExempt(
-		ctx context.Context,
-		account *types.AccountIdentifier,
-		current *types.Currency,
-	) bool
-}
-
-// BalanceChanges returns all balance changes for
-// a particular block. All balance changes for a
-// particular account are summed into a single
-// storage.BalanceChanges struct. If a block is being
-// orphaned, the opposite of each balance change is
-// returned.
-func BalanceChanges(
+func (s *Syncer) syncRange(
 	ctx context.Context,
-	asserter *asserter.Asserter,
-	block *types.Block,
-	orphan bool,
-	handler Handler,
-) ([]*storage.BalanceChange, error) {
-	balanceChanges := map[string]*storage.BalanceChange{}
-	for _, tx := range block.Transactions {
-		for _, op := range tx.Operations {
-			skip, err := skipOperation(
-				ctx,
-				asserter,
-				handler,
-				op,
-			)
-			if err != nil {
-				return nil, err
-			}
-			if skip {
-				continue
-			}
-
-			amount := op.Amount
-			blockIdentifier := block.BlockIdentifier
-			if orphan {
-				existing, ok := new(big.Int).SetString(amount.Value, 10)
-				if !ok {
-					return nil, fmt.Errorf("%s is not an integer", amount.Value)
-				}
-
-				amount.Value = new(big.Int).Neg(existing).String()
-				blockIdentifier = block.ParentBlockIdentifier
-			}
+	endIndex int64,
+) error {
+	blockMap, err := s.fetcher.BlockRange(ctx, s.network, s.nextIndex, endIndex)
+	if err != nil {
+		return err
+	}
 
-			// Merge values by account and currency
-			// TODO: change balance key to be this
-			key := fmt.Sprintf("%s:%s",
-				storage.GetAccountKey(op.Account),
-				storage.GetCurrencyKey(op.Amount.Currency),
+	for s.nextIndex <= endIndex {
+		block, ok := blockMap[s.nextIndex]
+		if !ok { // could happen in a reorg
+			block, err = s.fetcher.BlockRetry(
+				ctx,
+				s.network,
+				&types.PartialBlockIdentifier{
+					Index: &s.nextIndex,
+				},
 			)
-
-			val, ok := balanceChanges[key]
-			if !ok {
-				balanceChanges[key] = &storage.BalanceChange{
-					Account:    op.Account,
-					Currency:   op.Amount.Currency,
-					Difference: amount.Value,
-					Block:      blockIdentifier,
-				}
-				continue
-			}
-
-			newDifference, err := storage.AddStringValues(val.Difference, amount.Value)
 			if err != nil {
-				return nil, err
+				return err
 			}
-			val.Difference = newDifference
-			balanceChanges[key] = val
+		} else {
+			// Anytime we re-fetch an index, we
+			// will need to make another call to the node
+			// as it is likely in a reorg.
+			delete(blockMap, s.nextIndex)
 		}
-	}
 
-	allChanges := []*storage.BalanceChange{}
-	for _, change := range balanceChanges {
-		allChanges = append(allChanges, change)
+		if err = s.processBlock(ctx, block); err != nil {
+			return err
+		}
 	}
 
-	return allChanges, nil
+	return nil
 }
 
-func skipOperation(
+// Sync cycles endlessly until there is an error
+// or the requested range is synced.
+func (s *Syncer) Sync(
 	ctx context.Context,
-	asserter *asserter.Asserter,
-	handler Handler,
-	op *types.Operation,
-) (bool, error) {
-	successful, err := asserter.OperationSuccessful(op)
-	if err != nil {
-		// Should only occur if responses not validated
-		return false, err
-	}
+	startIndex int64,
+	endIndex int64,
+) error {
+	defer s.cancel()
 
-	if !successful {
-		return true, nil
+	if err := s.setStart(ctx, startIndex); err != nil {
+		return fmt.Errorf("%w: unable to set start index", err)
 	}
 
-	if op.Account == nil {
-		return true, nil
-	}
+	for {
+		rangeEnd, halt, err := s.nextSyncableRange(
+			ctx,
+			endIndex,
+		)
+		if err != nil {
+			return fmt.Errorf("%w: unable to get next syncable range", err)
+		}
+		if halt {
+			break
+		}
 
-	// Exempting account in BalanceChanges ensures that storage is not updated
-	// and that the account is not reconciled. If a handler is not provided,
-	// no account will be marked exempt.
-	if handler != nil && handler.AccountExempt(ctx, op.Account, op.Amount.Currency) {
-		log.Printf("Skipping exempt account %+v\n", op.Account)
-		return true, nil
+		log.Printf("Syncing %d-%d\n", s.nextIndex, rangeEnd)
+
+		err = s.syncRange(ctx, rangeEnd)
+		if err != nil {
+			return fmt.Errorf("%w: unable to sync to %d", err, rangeEnd)
+		}
+
+		if ctx.Err() != nil {
+			return ctx.Err()
+		}
 	}
 
-	return false, nil
+	log.Printf("Finished syncing %d-%d\n", startIndex, endIndex)
+	return nil
 }
diff --git a/internal/utils/utils.go b/internal/utils/utils.go
new file mode 100644
index 0000000..69879bf
--- /dev/null
+++ b/internal/utils/utils.go
@@ -0,0 +1,48 @@
+// Copyright 2020 Coinbase, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package utils
+
+import (
+	"io/ioutil"
+	"log"
+	"os"
+
+	"github.com/coinbase/rosetta-sdk-go/types"
+)
+
+// CreateTempDir creates a directory in
+// /tmp for usage within testing.
+func CreateTempDir() (string, error) {
+	storageDir, err := ioutil.TempDir("", "rosetta-cli")
+	if err != nil {
+		return "", err
+	}
+
+	return storageDir, nil
+}
+
+// RemoveTempDir deletes a directory at
+// a provided path for usage within testing.
+func RemoveTempDir(dir string) {
+	if err := os.RemoveAll(dir); err != nil {
+		log.Fatal(err)
+	}
+}
+
+// Equal returns a boolean indicating if two
+// interfaces are equal.
+func Equal(a interface{}, b interface{}) bool {
+	return types.Hash(a) == types.Hash(b)
+}


Test patch:
diff --git a/internal/reconciler/reconciler_test.go b/internal/reconciler/reconciler_test.go
index 5e2b66e..3532b9b 100644
--- a/internal/reconciler/reconciler_test.go
+++ b/internal/reconciler/reconciler_test.go
@@ -15,7 +15,10 @@
 package reconciler
 
 import (
+	"context"
+	"errors"
 	"fmt"
+	"reflect"
 	"testing"
 
 	"github.com/coinbase/rosetta-sdk-go/types"
@@ -182,3 +185,255 @@ func TestExtractAmount(t *testing.T) {
 		assert.NoError(t, err)
 	})
 }
+
+func TestCompareBalance(t *testing.T) {
+	var (
+		account1 = &types.AccountIdentifier{
+			Address: "blah",
+		}
+
+		account2 = &types.AccountIdentifier{
+			Address: "blah",
+			SubAccount: &types.SubAccountIdentifier{
+				Address: "sub blah",
+			},
+		}
+
+		currency1 = &types.Currency{
+			Symbol:   "curr1",
+			Decimals: 4,
+		}
+
+		currency2 = &types.Currency{
+			Symbol:   "curr2",
+			Decimals: 7,
+		}
+
+		amount1 = &types.Amount{
+			Value:    "100",
+			Currency: currency1,
+		}
+
+		amount2 = &types.Amount{
+			Value:    "200",
+			Currency: currency2,
+		}
+
+		block0 = &types.BlockIdentifier{
+			Hash:  "block0",
+			Index: 0,
+		}
+
+		block1 = &types.BlockIdentifier{
+			Hash:  "block1",
+			Index: 1,
+		}
+
+		block2 = &types.BlockIdentifier{
+			Hash:  "block2",
+			Index: 2,
+		}
+
+		ctx = context.Background()
+
+		mh = &MockReconcilerHelper{}
+	)
+
+	reconciler := NewReconciler(
+		nil,
+		mh,
+		nil,
+		nil,
+		1,
+		false,
+		[]*AccountCurrency{},
+	)
+
+	t.Run("No head block yet", func(t *testing.T) {
+		difference, cachedBalance, headIndex, err := reconciler.CompareBalance(
+			ctx,
+			account1,
+			currency1,
+			amount1.Value,
+			block1,
+		)
+		assert.Equal(t, "0", difference)
+		assert.Equal(t, "", cachedBalance)
+		assert.Equal(t, int64(0), headIndex)
+		assert.Error(t, err)
+	})
+
+	// Update head block
+	mh.HeadBlock = block0
+
+	t.Run("Live block is ahead of head block", func(t *testing.T) {
+		difference, cachedBalance, headIndex, err := reconciler.CompareBalance(
+			ctx,
+			account1,
+			currency1,
+			amount1.Value,
+			block1,
+		)
+		assert.Equal(t, "0", difference)
+		assert.Equal(t, "", cachedBalance)
+		assert.Equal(t, int64(0), headIndex)
+		assert.EqualError(t, err, fmt.Errorf(
+			"%w live block %d > head block %d",
+			ErrHeadBlockBehindLive,
+			1,
+			0,
+		).Error())
+	})
+
+	// Update head block
+	mh.HeadBlock = &types.BlockIdentifier{
+		Hash:  "hash2",
+		Index: 2,
+	}
+
+	t.Run("Live block is not in store", func(t *testing.T) {
+		difference, cachedBalance, headIndex, err := reconciler.CompareBalance(
+			ctx,
+			account1,
+			currency1,
+			amount1.Value,
+			block1,
+		)
+		assert.Equal(t, "0", difference)
+		assert.Equal(t, "", cachedBalance)
+		assert.Equal(t, int64(2), headIndex)
+		assert.Contains(t, err.Error(), ErrBlockGone.Error())
+	})
+
+	// Add blocks to store behind head
+	mh.StoredBlocks = map[string]*types.Block{}
+	mh.StoredBlocks[block0.Hash] = &types.Block{
+		BlockIdentifier:       block0,
+		ParentBlockIdentifier: block0,
+	}
+	mh.StoredBlocks[block1.Hash] = &types.Block{
+		BlockIdentifier:       block1,
+		ParentBlockIdentifier: block0,
+	}
+	mh.StoredBlocks[block2.Hash] = &types.Block{
+		BlockIdentifier:       block2,
+		ParentBlockIdentifier: block1,
+	}
+	mh.BalanceAccount = account1
+	mh.BalanceAmount = amount1
+	mh.BalanceBlock = block1
+
+	t.Run("Account updated after live block", func(t *testing.T) {
+		difference, cachedBalance, headIndex, err := reconciler.CompareBalance(
+			ctx,
+			account1,
+			currency1,
+			amount1.Value,
+			block0,
+		)
+		assert.Equal(t, "0", difference)
+		assert.Equal(t, "", cachedBalance)
+		assert.Equal(t, int64(2), headIndex)
+		assert.Contains(t, err.Error(), ErrAccountUpdated.Error())
+	})
+
+	t.Run("Account balance matches", func(t *testing.T) {
+		difference, cachedBalance, headIndex, err := reconciler.CompareBalance(
+			ctx,
+			account1,
+			currency1,
+			amount1.Value,
+			block1,
+		)
+		assert.Equal(t, "0", difference)
+		assert.Equal(t, amount1.Value, cachedBalance)
+		assert.Equal(t, int64(2), headIndex)
+		assert.NoError(t, err)
+	})
+
+	t.Run("Account balance matches later live block", func(t *testing.T) {
+		difference, cachedBalance, headIndex, err := reconciler.CompareBalance(
+			ctx,
+			account1,
+			currency1,
+			amount1.Value,
+			block2,
+		)
+		assert.Equal(t, "0", difference)
+		assert.Equal(t, amount1.Value, cachedBalance)
+		assert.Equal(t, int64(2), headIndex)
+		assert.NoError(t, err)
+	})
+
+	t.Run("Balances are not equal", func(t *testing.T) {
+		difference, cachedBalance, headIndex, err := reconciler.CompareBalance(
+			ctx,
+			account1,
+			currency1,
+			amount2.Value,
+			block2,
+		)
+		assert.Equal(t, "-100", difference)
+		assert.Equal(t, amount1.Value, cachedBalance)
+		assert.Equal(t, int64(2), headIndex)
+		assert.NoError(t, err)
+	})
+
+	t.Run("Compare balance for non-existent account", func(t *testing.T) {
+		difference, cachedBalance, headIndex, err := reconciler.CompareBalance(
+			ctx,
+			account2,
+			currency1,
+			amount2.Value,
+			block2,
+		)
+		assert.Equal(t, "0", difference)
+		assert.Equal(t, "", cachedBalance)
+		assert.Equal(t, int64(2), headIndex)
+		assert.Error(t, err)
+	})
+}
+
+type MockReconcilerHelper struct {
+	HeadBlock    *types.BlockIdentifier
+	StoredBlocks map[string]*types.Block
+
+	BalanceAccount *types.AccountIdentifier
+	BalanceAmount  *types.Amount
+	BalanceBlock   *types.BlockIdentifier
+}
+
+func (h *MockReconcilerHelper) BlockExists(
+	ctx context.Context,
+	block *types.BlockIdentifier,
+) (bool, error) {
+	_, ok := h.StoredBlocks[block.Hash]
+	if !ok {
+		return false, nil
+	}
+
+	return true, nil
+}
+
+func (h *MockReconcilerHelper) CurrentBlock(
+	ctx context.Context,
+) (*types.BlockIdentifier, error) {
+	if h.HeadBlock == nil {
+		return nil, errors.New("head block is nil")
+	}
+
+	return h.HeadBlock, nil
+}
+
+func (h *MockReconcilerHelper) AccountBalance(
+	ctx context.Context,
+	account *types.AccountIdentifier,
+	currency *types.Currency,
+	headBlock *types.BlockIdentifier,
+) (*types.Amount, *types.BlockIdentifier, error) {
+	if h.BalanceAccount == nil || !reflect.DeepEqual(account, h.BalanceAccount) {
+		return nil, nil, errors.New("account does not exist")
+	}
+
+	return h.BalanceAmount, h.BalanceBlock, nil
+}
diff --git a/internal/reconciler/stateful_reconciler_test.go b/internal/reconciler/stateful_reconciler_test.go
deleted file mode 100644
index b05f583..0000000
--- a/internal/reconciler/stateful_reconciler_test.go
+++ /dev/null
@@ -1,243 +0,0 @@
-// Copyright 2020 Coinbase, Inc.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//      http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package reconciler
-
-import (
-	"context"
-	"errors"
-	"fmt"
-	"testing"
-
-	"github.com/coinbase/rosetta-cli/internal/logger"
-	"github.com/coinbase/rosetta-cli/internal/storage"
-
-	"github.com/coinbase/rosetta-sdk-go/types"
-	"github.com/stretchr/testify/assert"
-)
-
-func TestCompareBalance(t *testing.T) {
-	var (
-		account1 = &types.AccountIdentifier{
-			Address: "blah",
-		}
-
-		account2 = &types.AccountIdentifier{
-			Address: "blah",
-			SubAccount: &types.SubAccountIdentifier{
-				Address: "sub blah",
-			},
-		}
-
-		currency1 = &types.Currency{
-			Symbol:   "curr1",
-			Decimals: 4,
-		}
-
-		currency2 = &types.Currency{
-			Symbol:   "curr2",
-			Decimals: 7,
-		}
-
-		amount1 = &types.Amount{
-			Value:    "100",
-			Currency: currency1,
-		}
-
-		amount2 = &types.Amount{
-			Value:    "200",
-			Currency: currency2,
-		}
-
-		block0 = &types.BlockIdentifier{
-			Hash:  "block0",
-			Index: 0,
-		}
-
-		block1 = &types.BlockIdentifier{
-			Hash:  "block1",
-			Index: 1,
-		}
-
-		block2 = &types.BlockIdentifier{
-			Hash:  "block2",
-			Index: 2,
-		}
-
-		ctx = context.Background()
-	)
-
-	newDir, err := storage.CreateTempDir()
-	assert.NoError(t, err)
-	defer storage.RemoveTempDir(*newDir)
-
-	database, err := storage.NewBadgerStorage(ctx, *newDir)
-	assert.NoError(t, err)
-	defer database.Close(ctx)
-
-	blockStorage := storage.NewBlockStorage(ctx, database)
-	logger := logger.NewLogger(*newDir, false, false, false, false)
-	reconciler := NewStateful(nil, blockStorage, nil, logger, 1, false, true)
-
-	t.Run("No head block yet", func(t *testing.T) {
-		difference, headIndex, err := reconciler.CompareBalance(
-			ctx,
-			account1,
-			currency1,
-			amount1.Value,
-			block1,
-		)
-		assert.Equal(t, "0", difference)
-		assert.Equal(t, int64(0), headIndex)
-		assert.True(t, errors.Is(err, storage.ErrHeadBlockNotFound))
-	})
-
-	// Update head block
-	txn := blockStorage.NewDatabaseTransaction(ctx, true)
-	err = blockStorage.StoreHeadBlockIdentifier(ctx, txn, block0)
-	assert.NoError(t, err)
-	err = txn.Commit(ctx)
-	assert.NoError(t, err)
-
-	t.Run("Live block is ahead of head block", func(t *testing.T) {
-		difference, headIndex, err := reconciler.CompareBalance(
-			ctx,
-			account1,
-			currency1,
-			amount1.Value,
-			block1,
-		)
-		assert.Equal(t, "0", difference)
-		assert.Equal(t, int64(0), headIndex)
-		assert.EqualError(t, err, fmt.Errorf(
-			"%w live block %d > head block %d",
-			ErrHeadBlockBehindLive,
-			1,
-			0,
-		).Error())
-	})
-
-	// Update head block
-	txn = blockStorage.NewDatabaseTransaction(ctx, true)
-	err = blockStorage.StoreHeadBlockIdentifier(ctx, txn, &types.BlockIdentifier{
-		Hash:  "hash2",
-		Index: 2,
-	})
-	assert.NoError(t, err)
-	err = txn.Commit(ctx)
-	assert.NoError(t, err)
-
-	t.Run("Live block is not in store", func(t *testing.T) {
-		difference, headIndex, err := reconciler.CompareBalance(
-			ctx,
-			account1,
-			currency1,
-			amount1.Value,
-			block1,
-		)
-		assert.Equal(t, "0", difference)
-		assert.Equal(t, int64(2), headIndex)
-		assert.Contains(t, err.Error(), ErrBlockGone.Error())
-	})
-
-	// Add blocks to store behind head
-	txn = blockStorage.NewDatabaseTransaction(ctx, true)
-	err = blockStorage.StoreBlock(ctx, txn, &types.Block{
-		BlockIdentifier:       block0,
-		ParentBlockIdentifier: block0,
-	})
-	assert.NoError(t, err)
-
-	err = blockStorage.StoreBlock(ctx, txn, &types.Block{
-		BlockIdentifier:       block1,
-		ParentBlockIdentifier: block0,
-	})
-	assert.NoError(t, err)
-
-	err = blockStorage.StoreBlock(ctx, txn, &types.Block{
-		BlockIdentifier:       block2,
-		ParentBlockIdentifier: block1,
-	})
-	assert.NoError(t, err)
-
-	_, err = blockStorage.UpdateBalance(ctx, txn, account1, amount1, block1)
-	assert.NoError(t, err)
-	err = txn.Commit(ctx)
-	assert.NoError(t, err)
-
-	t.Run("Account updated after live block", func(t *testing.T) {
-		difference, headIndex, err := reconciler.CompareBalance(
-			ctx,
-			account1,
-			currency1,
-			amount1.Value,
-			block0,
-		)
-		assert.Equal(t, "0", difference)
-		assert.Equal(t, int64(2), headIndex)
-		assert.Contains(t, err.Error(), ErrAccountUpdated.Error())
-	})
-
-	t.Run("Account balance matches", func(t *testing.T) {
-		difference, headIndex, err := reconciler.CompareBalance(
-			ctx,
-			account1,
-			currency1,
-			amount1.Value,
-			block1,
-		)
-		assert.Equal(t, "0", difference)
-		assert.Equal(t, int64(2), headIndex)
-		assert.NoError(t, err)
-	})
-
-	t.Run("Account balance matches later live block", func(t *testing.T) {
-		difference, headIndex, err := reconciler.CompareBalance(
-			ctx,
-			account1,
-			currency1,
-			amount1.Value,
-			block2,
-		)
-		assert.Equal(t, "0", difference)
-		assert.Equal(t, int64(2), headIndex)
-		assert.NoError(t, err)
-	})
-
-	t.Run("Balances are not equal", func(t *testing.T) {
-		difference, headIndex, err := reconciler.CompareBalance(
-			ctx,
-			account1,
-			currency1,
-			amount2.Value,
-			block2,
-		)
-		assert.Equal(t, "-100", difference)
-		assert.Equal(t, int64(2), headIndex)
-		assert.NoError(t, err)
-	})
-
-	t.Run("Compare balance for non-existent account", func(t *testing.T) {
-		difference, headIndex, err := reconciler.CompareBalance(
-			ctx,
-			account2,
-			currency1,
-			amount2.Value,
-			block2,
-		)
-		assert.Equal(t, "0", difference)
-		assert.Equal(t, int64(2), headIndex)
-		assert.Contains(t, err.Error(), storage.ErrAccountNotFound.Error())
-	})
-}
diff --git a/internal/storage/badger_storage_test.go b/internal/storage/badger_storage_test.go
index aa11275..9abdce1 100644
--- a/internal/storage/badger_storage_test.go
+++ b/internal/storage/badger_storage_test.go
@@ -18,17 +18,19 @@ import (
 	"context"
 	"testing"
 
+	"github.com/coinbase/rosetta-cli/internal/utils"
+
 	"github.com/stretchr/testify/assert"
 )
 
 func TestDatabase(t *testing.T) {
 	ctx := context.Background()
 
-	newDir, err := CreateTempDir()
+	newDir, err := utils.CreateTempDir()
 	assert.NoError(t, err)
-	defer RemoveTempDir(*newDir)
+	defer utils.RemoveTempDir(newDir)
 
-	database, err := NewBadgerStorage(ctx, *newDir)
+	database, err := NewBadgerStorage(ctx, newDir)
 	assert.NoError(t, err)
 	defer database.Close(ctx)
 
@@ -55,11 +57,11 @@ func TestDatabase(t *testing.T) {
 func TestDatabaseTransaction(t *testing.T) {
 	ctx := context.Background()
 
-	newDir, err := CreateTempDir()
+	newDir, err := utils.CreateTempDir()
 	assert.NoError(t, err)
-	defer RemoveTempDir(*newDir)
+	defer utils.RemoveTempDir(newDir)
 
-	database, err := NewBadgerStorage(ctx, *newDir)
+	database, err := NewBadgerStorage(ctx, newDir)
 	assert.NoError(t, err)
 	defer database.Close(ctx)
 
diff --git a/internal/storage/block_storage_test.go b/internal/storage/block_storage_test.go
index bfb7170..6e757ab 100644
--- a/internal/storage/block_storage_test.go
+++ b/internal/storage/block_storage_test.go
@@ -17,14 +17,18 @@ package storage
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io/ioutil"
 	"os"
 	"path"
 	"testing"
 
-	"github.com/coinbase/rosetta-sdk-go/types"
+	"github.com/coinbase/rosetta-cli/internal/reconciler"
+	"github.com/coinbase/rosetta-cli/internal/utils"
 
+	"github.com/coinbase/rosetta-sdk-go/asserter"
+	"github.com/coinbase/rosetta-sdk-go/types"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -42,38 +46,34 @@ func TestHeadBlockIdentifier(t *testing.T) {
 
 	ctx := context.Background()
 
-	newDir, err := CreateTempDir()
+	newDir, err := utils.CreateTempDir()
 	assert.NoError(t, err)
-	defer RemoveTempDir(*newDir)
+	defer utils.RemoveTempDir(newDir)
 
-	database, err := NewBadgerStorage(ctx, *newDir)
+	database, err := NewBadgerStorage(ctx, newDir)
 	assert.NoError(t, err)
 	defer database.Close(ctx)
 
-	storage := NewBlockStorage(ctx, database)
+	storage := NewBlockStorage(ctx, database, &MockBlockStorageHelper{})
 
 	t.Run("No head block set", func(t *testing.T) {
-		txn := storage.NewDatabaseTransaction(ctx, false)
-		blockIdentifier, err := storage.GetHeadBlockIdentifier(ctx, txn)
-		txn.Discard(ctx)
+		blockIdentifier, err := storage.GetHeadBlockIdentifier(ctx)
 		assert.EqualError(t, err, ErrHeadBlockNotFound.Error())
 		assert.Nil(t, blockIdentifier)
 	})
 
 	t.Run("Set and get head block", func(t *testing.T) {
-		txn := storage.NewDatabaseTransaction(ctx, true)
+		txn := storage.newDatabaseTransaction(ctx, true)
 		assert.NoError(t, storage.StoreHeadBlockIdentifier(ctx, txn, newBlockIdentifier))
 		assert.NoError(t, txn.Commit(ctx))
 
-		txn = storage.NewDatabaseTransaction(ctx, false)
-		blockIdentifier, err := storage.GetHeadBlockIdentifier(ctx, txn)
+		blockIdentifier, err := storage.GetHeadBlockIdentifier(ctx)
 		assert.NoError(t, err)
-		txn.Discard(ctx)
 		assert.Equal(t, newBlockIdentifier, blockIdentifier)
 	})
 
 	t.Run("Discard head block update", func(t *testing.T) {
-		txn := storage.NewDatabaseTransaction(ctx, true)
+		txn := storage.newDatabaseTransaction(ctx, true)
 		assert.NoError(t, storage.StoreHeadBlockIdentifier(ctx, txn,
 			&types.BlockIdentifier{
 				Hash:  "no blah",
@@ -82,105 +82,122 @@ func TestHeadBlockIdentifier(t *testing.T) {
 		)
 		txn.Discard(ctx)
 
-		txn = storage.NewDatabaseTransaction(ctx, false)
-		blockIdentifier, err := storage.GetHeadBlockIdentifier(ctx, txn)
+		blockIdentifier, err := storage.GetHeadBlockIdentifier(ctx)
 		assert.NoError(t, err)
-		txn.Discard(ctx)
 		assert.Equal(t, newBlockIdentifier, blockIdentifier)
 	})
 
 	t.Run("Multiple updates to head block", func(t *testing.T) {
-		txn := storage.NewDatabaseTransaction(ctx, true)
+		txn := storage.newDatabaseTransaction(ctx, true)
 		assert.NoError(t, storage.StoreHeadBlockIdentifier(ctx, txn, newBlockIdentifier2))
 		assert.NoError(t, txn.Commit(ctx))
 
-		txn = storage.NewDatabaseTransaction(ctx, false)
-		blockIdentifier, err := storage.GetHeadBlockIdentifier(ctx, txn)
+		blockIdentifier, err := storage.GetHeadBlockIdentifier(ctx)
 		assert.NoError(t, err)
 		txn.Discard(ctx)
 		assert.Equal(t, newBlockIdentifier2, blockIdentifier)
 	})
 }
 
-func TestBlock(t *testing.T) {
-	var (
-		newBlock = &types.Block{
-			BlockIdentifier: &types.BlockIdentifier{
-				Hash:  "blah",
-				Index: 0,
-			},
-			Timestamp: 1,
-			Transactions: []*types.Transaction{
-				{
-					TransactionIdentifier: &types.TransactionIdentifier{
-						Hash: "blahTx",
-					},
-					Operations: []*types.Operation{
-						{
-							OperationIdentifier: &types.OperationIdentifier{
-								Index: 0,
-							},
+var (
+	newBlock = &types.Block{
+		BlockIdentifier: &types.BlockIdentifier{
+			Hash:  "blah 1",
+			Index: 1,
+		},
+		ParentBlockIdentifier: &types.BlockIdentifier{
+			Hash:  "blah 0",
+			Index: 0,
+		},
+		Timestamp: 1,
+		Transactions: []*types.Transaction{
+			{
+				TransactionIdentifier: &types.TransactionIdentifier{
+					Hash: "blahTx",
+				},
+				Operations: []*types.Operation{
+					{
+						OperationIdentifier: &types.OperationIdentifier{
+							Index: 0,
 						},
 					},
 				},
 			},
-		}
+		},
+	}
 
-		badBlockIdentifier = &types.BlockIdentifier{
-			Hash:  "missing blah",
-			Index: 0,
-		}
+	badBlockIdentifier = &types.BlockIdentifier{
+		Hash:  "missing blah",
+		Index: 0,
+	}
 
-		newBlock2 = &types.Block{
-			BlockIdentifier: &types.BlockIdentifier{
-				Hash:  "blah 2",
-				Index: 1,
-			},
-			Timestamp: 1,
-			Transactions: []*types.Transaction{
-				{
-					TransactionIdentifier: &types.TransactionIdentifier{
-						Hash: "blahTx",
-					},
-					Operations: []*types.Operation{
-						{
-							OperationIdentifier: &types.OperationIdentifier{
-								Index: 0,
-							},
+	newBlock2 = &types.Block{
+		BlockIdentifier: &types.BlockIdentifier{
+			Hash:  "blah 2",
+			Index: 2,
+		},
+		ParentBlockIdentifier: &types.BlockIdentifier{
+			Hash:  "blah 1",
+			Index: 1,
+		},
+		Timestamp: 1,
+		Transactions: []*types.Transaction{
+			{
+				TransactionIdentifier: &types.TransactionIdentifier{
+					Hash: "blahTx",
+				},
+				Operations: []*types.Operation{
+					{
+						OperationIdentifier: &types.OperationIdentifier{
+							Index: 0,
 						},
 					},
 				},
 			},
-		}
-	)
+		},
+	}
+
+	newBlock3 = &types.Block{
+		BlockIdentifier: &types.BlockIdentifier{
+			Hash:  "blah 2",
+			Index: 2,
+		},
+		ParentBlockIdentifier: &types.BlockIdentifier{
+			Hash:  "blah 1",
+			Index: 1,
+		},
+		Timestamp: 1,
+	}
+)
+
+func TestBlock(t *testing.T) {
 	ctx := context.Background()
 
-	newDir, err := CreateTempDir()
+	newDir, err := utils.CreateTempDir()
 	assert.NoError(t, err)
-	defer RemoveTempDir(*newDir)
+	defer utils.RemoveTempDir(newDir)
 
-	database, err := NewBadgerStorage(ctx, *newDir)
+	database, err := NewBadgerStorage(ctx, newDir)
 	assert.NoError(t, err)
 	defer database.Close(ctx)
 
-	storage := NewBlockStorage(ctx, database)
+	storage := NewBlockStorage(ctx, database, &MockBlockStorageHelper{})
 
 	t.Run("Set and get block", func(t *testing.T) {
-		txn := storage.NewDatabaseTransaction(ctx, true)
-		assert.NoError(t, storage.StoreBlock(ctx, txn, newBlock))
-		assert.NoError(t, txn.Commit(ctx))
+		_, err := storage.StoreBlock(ctx, newBlock)
+		assert.NoError(t, err)
 
-		txn = storage.NewDatabaseTransaction(ctx, false)
-		block, err := storage.GetBlock(ctx, txn, newBlock.BlockIdentifier)
-		txn.Discard(ctx)
+		block, err := storage.GetBlock(ctx, newBlock.BlockIdentifier)
 		assert.NoError(t, err)
 		assert.Equal(t, newBlock, block)
+
+		head, err := storage.GetHeadBlockIdentifier(ctx)
+		assert.NoError(t, err)
+		assert.Equal(t, newBlock.BlockIdentifier, head)
 	})
 
 	t.Run("Get non-existent block", func(t *testing.T) {
-		txn := storage.NewDatabaseTransaction(ctx, false)
-		block, err := storage.GetBlock(ctx, txn, badBlockIdentifier)
-		txn.Discard(ctx)
+		block, err := storage.GetBlock(ctx, badBlockIdentifier)
 		assert.EqualError(
 			t,
 			err,
@@ -190,102 +207,38 @@ func TestBlock(t *testing.T) {
 	})
 
 	t.Run("Set duplicate block hash", func(t *testing.T) {
-		txn := storage.NewDatabaseTransaction(ctx, true)
-		err = storage.StoreBlock(ctx, txn, newBlock)
+		_, err = storage.StoreBlock(ctx, newBlock)
 		assert.EqualError(t, err, fmt.Errorf(
 			"%w %s",
 			ErrDuplicateBlockHash,
 			newBlock.BlockIdentifier.Hash,
 		).Error())
-		txn.Discard(ctx)
 	})
 
 	t.Run("Set duplicate transaction hash", func(t *testing.T) {
-		txn := storage.NewDatabaseTransaction(ctx, true)
-		err = storage.StoreBlock(ctx, txn, newBlock2)
+		_, err = storage.StoreBlock(ctx, newBlock2)
 		assert.EqualError(t, err, fmt.Errorf(
 			"%w %s",
 			ErrDuplicateTransactionHash,
 			"blahTx",
 		).Error())
-		txn.Discard(ctx)
+
+		head, err := storage.GetHeadBlockIdentifier(ctx)
+		assert.NoError(t, err)
+		assert.Equal(t, newBlock.BlockIdentifier, head)
 	})
 
 	t.Run("Remove block and re-set block of same hash", func(t *testing.T) {
-		txn := storage.NewDatabaseTransaction(ctx, true)
-		assert.NoError(t, storage.RemoveBlock(ctx, txn, newBlock.BlockIdentifier))
-		assert.NoError(t, txn.Commit(ctx))
-
-		txn = storage.NewDatabaseTransaction(ctx, true)
-		assert.NoError(t, storage.StoreBlock(ctx, txn, newBlock))
-		assert.NoError(t, txn.Commit(ctx))
-	})
-}
+		_, err := storage.RemoveBlock(ctx, newBlock.BlockIdentifier)
+		assert.NoError(t, err)
 
-func TestGetAccountKey(t *testing.T) {
-	var tests = map[string]struct {
-		account *types.AccountIdentifier
-		key     string
-	}{
-		"simple account": {
-			account: &types.AccountIdentifier{
-				Address: "hello",
-			},
-			key: "balance:hello",
-		},
-		"subaccount": {
-			account: &types.AccountIdentifier{
-				Address: "hello",
-				SubAccount: &types.SubAccountIdentifier{
-					Address: "stake",
-				},
-			},
-			key: "balance:hello:stake",
-		},
-		"subaccount with string metadata": {
-			account: &types.AccountIdentifier{
-				Address: "hello",
-				SubAccount: &types.SubAccountIdentifier{
-					Address: "stake",
-					Metadata: map[string]interface{}{
-						"cool": "neat",
-					},
-				},
-			},
-			key: "balance:hello:stake:map[cool:neat]",
-		},
-		"subaccount with number metadata": {
-			account: &types.AccountIdentifier{
-				Address: "hello",
-				SubAccount: &types.SubAccountIdentifier{
-					Address: "stake",
-					Metadata: map[string]interface{}{
-						"cool": 1,
-					},
-				},
-			},
-			key: "balance:hello:stake:map[cool:1]",
-		},
-		"subaccount with complex metadata": {
-			account: &types.AccountIdentifier{
-				Address: "hello",
-				SubAccount: &types.SubAccountIdentifier{
-					Address: "stake",
-					Metadata: map[string]interface{}{
-						"cool":    1,
-						"awesome": "neat",
-					},
-				},
-			},
-			key: "balance:hello:stake:map[awesome:neat cool:1]",
-		},
-	}
+		head, err := storage.GetHeadBlockIdentifier(ctx)
+		assert.NoError(t, err)
+		assert.Equal(t, newBlock.ParentBlockIdentifier, head)
 
-	for name, test := range tests {
-		t.Run(name, func(t *testing.T) {
-			assert.Equal(t, hashBytes([]byte(test.key)), GetAccountKey(test.account))
-		})
-	}
+		_, err = storage.StoreBlock(ctx, newBlock)
+		assert.NoError(t, err)
+	})
 }
 
 func TestBalance(t *testing.T) {
@@ -296,6 +249,9 @@ func TestBalance(t *testing.T) {
 		account2 = &types.AccountIdentifier{
 			Address: "blah2",
 		}
+		account3 = &types.AccountIdentifier{
+			Address: "blah3",
+		}
 		subAccount = &types.AccountIdentifier{
 			Address: "blah",
 			SubAccount: &types.SubAccountIdentifier{
@@ -352,12 +308,13 @@ func TestBalance(t *testing.T) {
 			Value:    "100",
 			Currency: currency,
 		}
+		amountWithPrevious = &types.Amount{
+			Value:    "110",
+			Currency: currency,
+		}
 		amountNilCurrency = &types.Amount{
 			Value: "100",
 		}
-		newAmounts = map[string]*types.Amount{
-			GetCurrencyKey(currency): amount,
-		}
 		newBlock = &types.BlockIdentifier{
 			Hash:  "kdasdj",
 			Index: 123890,
@@ -366,11 +323,9 @@ func TestBalance(t *testing.T) {
 			Hash:  "pkdasdj",
 			Index: 123890,
 		}
-		result = map[string]*types.Amount{
-			GetCurrencyKey(currency): {
-				Value:    "200",
-				Currency: currency,
-			},
+		result = &types.Amount{
+			Value:    "200",
+			Currency: currency,
 		}
 		newBlock3 = &types.BlockIdentifier{
 			Hash:  "pkdgdj",
@@ -380,302 +335,263 @@ func TestBalance(t *testing.T) {
 			Value:    "-1000",
 			Currency: currency,
 		}
+		mockHelper = &MockBlockStorageHelper{}
 	)
 
 	ctx := context.Background()
 
-	newDir, err := CreateTempDir()
+	newDir, err := utils.CreateTempDir()
 	assert.NoError(t, err)
-	defer RemoveTempDir(*newDir)
+	defer utils.RemoveTempDir(newDir)
 
-	database, err := NewBadgerStorage(ctx, *newDir)
+	database, err := NewBadgerStorage(ctx, newDir)
 	assert.NoError(t, err)
 	defer database.Close(ctx)
 
-	storage := NewBlockStorage(ctx, database)
+	storage := NewBlockStorage(ctx, database, mockHelper)
 
 	t.Run("Get unset balance", func(t *testing.T) {
-		txn := storage.NewDatabaseTransaction(ctx, false)
-		amounts, block, err := storage.GetBalance(ctx, txn, account)
-		txn.Discard(ctx)
-		assert.Nil(t, amounts)
-		assert.Nil(t, block)
-		assert.EqualError(t, err, fmt.Errorf("%w %+v", ErrAccountNotFound, account).Error())
+		amount, block, err := storage.GetBalance(ctx, account, currency, newBlock)
+		assert.NoError(t, err)
+		assert.Equal(t, &types.Amount{
+			Value:    "0",
+			Currency: currency,
+		}, amount)
+		assert.Equal(t, newBlock, block)
 	})
 
 	t.Run("Set and get balance", func(t *testing.T) {
-		txn := storage.NewDatabaseTransaction(ctx, true)
-		balanceChange, err := storage.UpdateBalance(
+		txn := storage.newDatabaseTransaction(ctx, true)
+		err := storage.UpdateBalance(
 			ctx,
 			txn,
-			account,
-			amount,
-			newBlock,
+			&reconciler.BalanceChange{
+				Account:    account,
+				Currency:   currency,
+				Block:      newBlock,
+				Difference: amount.Value,
+			},
+			nil,
 		)
-		assert.Equal(t, &BalanceChange{
-			Account: &types.AccountIdentifier{
-				Address: "blah",
+		assert.NoError(t, err)
+		assert.NoError(t, txn.Commit(ctx))
+
+		retrievedAmount, block, err := storage.GetBalance(ctx, account, currency, newBlock)
+		assert.NoError(t, err)
+		assert.Equal(t, amount, retrievedAmount)
+		assert.Equal(t, newBlock, block)
+	})
+
+	t.Run("Set and get balance with storage helper", func(t *testing.T) {
+		mockHelper.AccountBalanceAmount = "10"
+		txn := storage.newDatabaseTransaction(ctx, true)
+		err := storage.UpdateBalance(
+			ctx,
+			txn,
+			&reconciler.BalanceChange{
+				Account:    account3,
+				Currency:   currency,
+				Block:      newBlock,
+				Difference: amount.Value,
 			},
-			Currency:   currency,
-			Block:      newBlock,
-			Difference: "100",
-		}, balanceChange)
+			nil,
+		)
 		assert.NoError(t, err)
 		assert.NoError(t, txn.Commit(ctx))
 
-		txn = storage.NewDatabaseTransaction(ctx, false)
-		amounts, block, err := storage.GetBalance(ctx, txn, account)
-		txn.Discard(ctx)
+		retrievedAmount, block, err := storage.GetBalance(ctx, account3, currency, newBlock)
 		assert.NoError(t, err)
-		assert.Equal(t, newAmounts, amounts)
+		assert.Equal(t, amountWithPrevious, retrievedAmount)
 		assert.Equal(t, newBlock, block)
+
+		mockHelper.AccountBalanceAmount = ""
 	})
 
 	t.Run("Set balance with nil currency", func(t *testing.T) {
-		txn := storage.NewDatabaseTransaction(ctx, true)
-		balanceChange, err := storage.UpdateBalance(
+		txn := storage.newDatabaseTransaction(ctx, true)
+		err := storage.UpdateBalance(
 			ctx,
 			txn,
-			account,
-			amountNilCurrency,
-			newBlock,
+			&reconciler.BalanceChange{
+				Account:    account,
+				Currency:   nil,
+				Block:      newBlock,
+				Difference: amountNilCurrency.Value,
+			},
+			nil,
 		)
-		assert.Nil(t, balanceChange)
-		assert.EqualError(t, err, "invalid amount")
+		assert.EqualError(t, err, "invalid currency")
 		txn.Discard(ctx)
 
-		txn = storage.NewDatabaseTransaction(ctx, false)
-		amounts, block, err := storage.GetBalance(ctx, txn, account)
-		txn.Discard(ctx)
+		retrievedAmount, block, err := storage.GetBalance(ctx, account, currency, newBlock)
 		assert.NoError(t, err)
-		assert.Equal(t, newAmounts, amounts)
+		assert.Equal(t, amount, retrievedAmount)
 		assert.Equal(t, newBlock, block)
 	})
 
 	t.Run("Modify existing balance", func(t *testing.T) {
-		txn := storage.NewDatabaseTransaction(ctx, true)
-		balanceChange, err := storage.UpdateBalance(
+		txn := storage.newDatabaseTransaction(ctx, true)
+		err := storage.UpdateBalance(
 			ctx,
 			txn,
-			account,
-			amount,
-			newBlock2,
-		)
-		assert.Equal(t, &BalanceChange{
-			Account: &types.AccountIdentifier{
-				Address: "blah",
+			&reconciler.BalanceChange{
+				Account:    account,
+				Currency:   currency,
+				Block:      newBlock2,
+				Difference: amount.Value,
 			},
-			Currency:   currency,
-			Block:      newBlock2,
-			Difference: "100",
-		}, balanceChange)
+			nil,
+		)
 		assert.NoError(t, err)
 		assert.NoError(t, txn.Commit(ctx))
 
-		txn = storage.NewDatabaseTransaction(ctx, true)
-		amounts, block, err := storage.GetBalance(ctx, txn, account)
-		txn.Discard(ctx)
+		retrievedAmount, block, err := storage.GetBalance(ctx, account, currency, newBlock2)
 		assert.NoError(t, err)
-		assert.Equal(t, result, amounts)
+		assert.Equal(t, result, retrievedAmount)
 		assert.Equal(t, newBlock2, block)
 	})
 
 	t.Run("Discard transaction", func(t *testing.T) {
-		txn := storage.NewDatabaseTransaction(ctx, true)
-		balanceChange, err := storage.UpdateBalance(
+		txn := storage.newDatabaseTransaction(ctx, true)
+		err := storage.UpdateBalance(
 			ctx,
 			txn,
-			account,
-			amount,
-			newBlock3,
-		)
-		assert.Equal(t, &BalanceChange{
-			Account: &types.AccountIdentifier{
-				Address: "blah",
+			&reconciler.BalanceChange{
+				Account:    account,
+				Currency:   currency,
+				Block:      newBlock3,
+				Difference: amount.Value,
 			},
-			Currency:   currency,
-			Block:      newBlock3,
-			Difference: "100",
-		}, balanceChange)
+			nil,
+		)
 		assert.NoError(t, err)
 
 		// Get balance during transaction
-		txn2 := storage.NewDatabaseTransaction(ctx, false)
-		amounts, block, err := storage.GetBalance(ctx, txn2, account)
-		txn2.Discard(ctx)
+		retrievedAmount, block, err := storage.GetBalance(ctx, account, currency, newBlock2)
 		assert.NoError(t, err)
-		assert.Equal(t, result, amounts)
+		assert.Equal(t, result, retrievedAmount)
 		assert.Equal(t, newBlock2, block)
 
 		txn.Discard(ctx)
-
-		txn = storage.NewDatabaseTransaction(ctx, false)
-		amounts, block, err = storage.GetBalance(ctx, txn, account)
-		txn.Discard(ctx)
-		assert.NoError(t, err)
-		assert.Equal(t, result, amounts)
-		assert.Equal(t, newBlock2, block)
 	})
 
 	t.Run("Attempt modification to push balance negative on existing account", func(t *testing.T) {
-		txn := storage.NewDatabaseTransaction(ctx, true)
-		balanceChange, err := storage.UpdateBalance(
+		txn := storage.newDatabaseTransaction(ctx, true)
+		err := storage.UpdateBalance(
 			ctx,
 			txn,
-			account,
-			largeDeduction,
-			newBlock2,
+			&reconciler.BalanceChange{
+				Account:    account,
+				Currency:   largeDeduction.Currency,
+				Block:      newBlock3,
+				Difference: largeDeduction.Value,
+			},
+			nil,
 		)
-		assert.Nil(t, balanceChange)
-		assert.Contains(t, err.Error(), ErrNegativeBalance.Error())
+		assert.True(t, errors.Is(err, ErrNegativeBalance))
 		txn.Discard(ctx)
 	})
 
 	t.Run("Attempt modification to push balance negative on new acct", func(t *testing.T) {
-		txn := storage.NewDatabaseTransaction(ctx, true)
-		balanceChange, err := storage.UpdateBalance(
+		txn := storage.newDatabaseTransaction(ctx, true)
+		err := storage.UpdateBalance(
 			ctx,
 			txn,
-			account2,
-			largeDeduction,
-			newBlock2,
+			&reconciler.BalanceChange{
+				Account:    account2,
+				Currency:   largeDeduction.Currency,
+				Block:      newBlock2,
+				Difference: largeDeduction.Value,
+			},
+			nil,
 		)
-		assert.Nil(t, balanceChange)
-		assert.Contains(t, err.Error(), ErrNegativeBalance.Error())
+		assert.Error(t, err)
+		assert.True(t, errors.Is(err, ErrNegativeBalance))
 		txn.Discard(ctx)
 	})
 
 	t.Run("sub account set and get balance", func(t *testing.T) {
-		txn := storage.NewDatabaseTransaction(ctx, true)
-		balanceChange, err := storage.UpdateBalance(
+		txn := storage.newDatabaseTransaction(ctx, true)
+		err := storage.UpdateBalance(
 			ctx,
 			txn,
-			subAccount,
-			amount,
-			newBlock,
+			&reconciler.BalanceChange{
+				Account:    subAccount,
+				Currency:   amount.Currency,
+				Block:      newBlock,
+				Difference: amount.Value,
+			},
+			nil,
 		)
-		assert.Equal(t, &BalanceChange{
-			Account:    subAccount,
-			Currency:   currency,
-			Block:      newBlock,
-			Difference: "100",
-		}, balanceChange)
 		assert.NoError(t, err)
 		assert.NoError(t, txn.Commit(ctx))
 
-		txn = storage.NewDatabaseTransaction(ctx, false)
-		amounts, block, err := storage.GetBalance(ctx, txn, subAccountNewPointer)
-		txn.Discard(ctx)
+		retrievedAmount, block, err := storage.GetBalance(
+			ctx,
+			subAccountNewPointer,
+			amount.Currency,
+			newBlock,
+		)
 		assert.NoError(t, err)
-		assert.Equal(t, newAmounts, amounts)
+		assert.Equal(t, amount, retrievedAmount)
 		assert.Equal(t, newBlock, block)
 	})
 
 	t.Run("sub account metadata set and get balance", func(t *testing.T) {
-		txn := storage.NewDatabaseTransaction(ctx, true)
-		balanceChange, err := storage.UpdateBalance(
+		txn := storage.newDatabaseTransaction(ctx, true)
+		err := storage.UpdateBalance(
 			ctx,
 			txn,
-			subAccountMetadata,
-			amount,
-			newBlock,
+			&reconciler.BalanceChange{
+				Account:    subAccountMetadata,
+				Currency:   amount.Currency,
+				Block:      newBlock,
+				Difference: amount.Value,
+			},
+			nil,
 		)
-		assert.Equal(t, &BalanceChange{
-			Account:    subAccountMetadata,
-			Currency:   currency,
-			Block:      newBlock,
-			Difference: "100",
-		}, balanceChange)
 		assert.NoError(t, err)
 		assert.NoError(t, txn.Commit(ctx))
 
-		txn = storage.NewDatabaseTransaction(ctx, false)
-		amounts, block, err := storage.GetBalance(ctx, txn, subAccountMetadataNewPointer)
-		txn.Discard(ctx)
+		retrievedAmount, block, err := storage.GetBalance(
+			ctx,
+			subAccountMetadataNewPointer,
+			amount.Currency,
+			newBlock,
+		)
 		assert.NoError(t, err)
-		assert.Equal(t, newAmounts, amounts)
+		assert.Equal(t, amount, retrievedAmount)
 		assert.Equal(t, newBlock, block)
 	})
 
 	t.Run("sub account unique metadata set and get balance", func(t *testing.T) {
-		txn := storage.NewDatabaseTransaction(ctx, true)
-		balanceChange, err := storage.UpdateBalance(
+		txn := storage.newDatabaseTransaction(ctx, true)
+		err := storage.UpdateBalance(
 			ctx,
 			txn,
-			subAccountMetadata2,
-			amount,
-			newBlock,
+			&reconciler.BalanceChange{
+				Account:    subAccountMetadata2,
+				Currency:   amount.Currency,
+				Block:      newBlock,
+				Difference: amount.Value,
+			},
+			nil,
 		)
-		assert.Equal(t, &BalanceChange{
-			Account:    subAccountMetadata2,
-			Currency:   currency,
-			Block:      newBlock,
-			Difference: "100",
-		}, balanceChange)
 		assert.NoError(t, err)
 		assert.NoError(t, txn.Commit(ctx))
 
-		txn = storage.NewDatabaseTransaction(ctx, false)
-		amounts, block, err := storage.GetBalance(ctx, txn, subAccountMetadata2NewPointer)
-		txn.Discard(ctx)
+		retrievedAmount, block, err := storage.GetBalance(
+			ctx,
+			subAccountMetadata2NewPointer,
+			amount.Currency,
+			newBlock,
+		)
 		assert.NoError(t, err)
-		assert.Equal(t, newAmounts, amounts)
+		assert.Equal(t, amount, retrievedAmount)
 		assert.Equal(t, newBlock, block)
 	})
 }
 
-func TestGetCurrencyKey(t *testing.T) {
-	var tests = map[string]struct {
-		currency *types.Currency
-		key      string
-	}{
-		"simple currency": {
-			currency: &types.Currency{
-				Symbol:   "BTC",
-				Decimals: 8,
-			},
-			key: "BTC:8",
-		},
-		"currency with string metadata": {
-			currency: &types.Currency{
-				Symbol:   "BTC",
-				Decimals: 8,
-				Metadata: map[string]interface{}{
-					"issuer": "satoshi",
-				},
-			},
-			key: "BTC:8:map[issuer:satoshi]",
-		},
-		"currency with number metadata": {
-			currency: &types.Currency{
-				Symbol:   "BTC",
-				Decimals: 8,
-				Metadata: map[string]interface{}{
-					"issuer": 1,
-				},
-			},
-			key: "BTC:8:map[issuer:1]",
-		},
-		"currency with complex metadata": {
-			currency: &types.Currency{
-				Symbol:   "BTC",
-				Decimals: 8,
-				Metadata: map[string]interface{}{
-					"issuer": "satoshi",
-					"count":  10,
-				},
-			},
-			key: "BTC:8:map[count:10 issuer:satoshi]",
-		},
-	}
-
-	for name, test := range tests {
-		t.Run(name, func(t *testing.T) {
-			assert.Equal(t, hashString(test.key), GetCurrencyKey(test.currency))
-		})
-	}
-}
-
 func TestBootstrapBalances(t *testing.T) {
 	var (
 		fileMode               = os.FileMode(0600)
@@ -691,16 +607,16 @@ func TestBootstrapBalances(t *testing.T) {
 
 	ctx := context.Background()
 
-	newDir, err := CreateTempDir()
+	newDir, err := utils.CreateTempDir()
 	assert.NoError(t, err)
-	defer RemoveTempDir(*newDir)
+	defer utils.RemoveTempDir(newDir)
 
-	database, err := NewBadgerStorage(ctx, *newDir)
+	database, err := NewBadgerStorage(ctx, newDir)
 	assert.NoError(t, err)
 	defer database.Close(ctx)
 
-	storage := NewBlockStorage(ctx, database)
-	bootstrapBalancesFile := path.Join(*newDir, "balances.csv")
+	storage := NewBlockStorage(ctx, database, &MockBlockStorageHelper{})
+	bootstrapBalancesFile := path.Join(newDir, "balances.csv")
 
 	t.Run("File doesn't exist", func(t *testing.T) {
 		err = storage.BootstrapBalances(
@@ -741,15 +657,14 @@ func TestBootstrapBalances(t *testing.T) {
 		)
 		assert.NoError(t, err)
 
-		tx := storage.NewDatabaseTransaction(ctx, false)
-		amountMap, blockIdentifier, err := storage.GetBalance(
+		retrievedAmount, blockIdentifier, err := storage.GetBalance(
 			ctx,
-			tx,
 			account,
+			amount.Currency,
+			genesisBlockIdentifier,
 		)
-		tx.Discard(ctx)
 
-		assert.Equal(t, amount, amountMap[GetCurrencyKey(amount.Currency)])
+		assert.Equal(t, amount, retrievedAmount)
 		assert.Equal(t, genesisBlockIdentifier, blockIdentifier)
 		assert.NoError(t, err)
 	})
@@ -822,7 +737,7 @@ func TestBootstrapBalances(t *testing.T) {
 	})
 
 	t.Run("Head block identifier already set", func(t *testing.T) {
-		tx := storage.NewDatabaseTransaction(ctx, true)
+		tx := storage.newDatabaseTransaction(ctx, true)
 		err := storage.StoreHeadBlockIdentifier(ctx, tx, genesisBlockIdentifier)
 		assert.NoError(t, err)
 		assert.NoError(t, tx.Commit(ctx))
@@ -832,3 +747,340 @@ func TestBootstrapBalances(t *testing.T) {
 		assert.EqualError(t, err, ErrAlreadyStartedSyncing.Error())
 	})
 }
+
+func simpleTransactionFactory(
+	hash string,
+	address string,
+	value string,
+	currency *types.Currency,
+) *types.Transaction {
+	return &types.Transaction{
+		TransactionIdentifier: &types.TransactionIdentifier{
+			Hash: hash,
+		},
+		Operations: []*types.Operation{
+			{
+				OperationIdentifier: &types.OperationIdentifier{
+					Index: 0,
+				},
+				Type:   "Transfer",
+				Status: "Success",
+				Account: &types.AccountIdentifier{
+					Address: address,
+				},
+				Amount: &types.Amount{
+					Value:    value,
+					Currency: currency,
+				},
+			},
+		},
+	}
+}
+
+func TestBalanceChanges(t *testing.T) {
+	var (
+		currency = &types.Currency{
+			Symbol:   "Blah",
+			Decimals: 2,
+		}
+
+		recipient = &types.AccountIdentifier{
+			Address: "acct1",
+		}
+
+		recipientAmount = &types.Amount{
+			Value:    "100",
+			Currency: currency,
+		}
+
+		recipientOperation = &types.Operation{
+			OperationIdentifier: &types.OperationIdentifier{
+				Index: 0,
+			},
+			Type:    "Transfer",
+			Status:  "Success",
+			Account: recipient,
+			Amount:  recipientAmount,
+		}
+
+		recipientFailureOperation = &types.Operation{
+			OperationIdentifier: &types.OperationIdentifier{
+				Index: 1,
+			},
+			Type:    "Transfer",
+			Status:  "Failure",
+			Account: recipient,
+			Amount:  recipientAmount,
+		}
+
+		recipientTransaction = &types.Transaction{
+			TransactionIdentifier: &types.TransactionIdentifier{
+				Hash: "tx1",
+			},
+			Operations: []*types.Operation{
+				recipientOperation,
+				recipientFailureOperation,
+			},
+		}
+	)
+
+	var tests = map[string]struct {
+		block          *types.Block
+		orphan         bool
+		changes        []*reconciler.BalanceChange
+		exemptAccounts []*reconciler.AccountCurrency
+		err            error
+	}{
+		"simple block": {
+			block: &types.Block{
+				BlockIdentifier: &types.BlockIdentifier{
+					Hash:  "1",
+					Index: 1,
+				},
+				ParentBlockIdentifier: &types.BlockIdentifier{
+					Hash:  "0",
+					Index: 0,
+				},
+				Transactions: []*types.Transaction{
+					recipientTransaction,
+				},
+				Timestamp: asserter.MinUnixEpoch + 1,
+			},
+			orphan: false,
+			changes: []*reconciler.BalanceChange{
+				{
+					Account:  recipient,
+					Currency: currency,
+					Block: &types.BlockIdentifier{
+						Hash:  "1",
+						Index: 1,
+					},
+					Difference: "100",
+				},
+			},
+			err: nil,
+		},
+		"simple block account exempt": {
+			block: &types.Block{
+				BlockIdentifier: &types.BlockIdentifier{
+					Hash:  "1",
+					Index: 1,
+				},
+				ParentBlockIdentifier: &types.BlockIdentifier{
+					Hash:  "0",
+					Index: 0,
+				},
+				Transactions: []*types.Transaction{
+					recipientTransaction,
+				},
+				Timestamp: asserter.MinUnixEpoch + 1,
+			},
+			orphan:  false,
+			changes: []*reconciler.BalanceChange{},
+			exemptAccounts: []*reconciler.AccountCurrency{
+				{
+					Account:  recipient,
+					Currency: currency,
+				},
+			},
+			err: nil,
+		},
+		"single account sum block": {
+			block: &types.Block{
+				BlockIdentifier: &types.BlockIdentifier{
+					Hash:  "1",
+					Index: 1,
+				},
+				ParentBlockIdentifier: &types.BlockIdentifier{
+					Hash:  "0",
+					Index: 0,
+				},
+				Transactions: []*types.Transaction{
+					simpleTransactionFactory("tx1", "addr1", "100", currency),
+					simpleTransactionFactory("tx2", "addr1", "150", currency),
+					simpleTransactionFactory("tx3", "addr2", "150", currency),
+				},
+				Timestamp: asserter.MinUnixEpoch + 1,
+			},
+			orphan: false,
+			changes: []*reconciler.BalanceChange{
+				{
+					Account: &types.AccountIdentifier{
+						Address: "addr1",
+					},
+					Currency: currency,
+					Block: &types.BlockIdentifier{
+						Hash:  "1",
+						Index: 1,
+					},
+					Difference: "250",
+				},
+				{
+					Account: &types.AccountIdentifier{
+						Address: "addr2",
+					},
+					Currency: currency,
+					Block: &types.BlockIdentifier{
+						Hash:  "1",
+						Index: 1,
+					},
+					Difference: "150",
+				},
+			},
+			err: nil,
+		},
+		"single account sum orphan block": {
+			block: &types.Block{
+				BlockIdentifier: &types.BlockIdentifier{
+					Hash:  "1",
+					Index: 1,
+				},
+				ParentBlockIdentifier: &types.BlockIdentifier{
+					Hash:  "0",
+					Index: 0,
+				},
+				Transactions: []*types.Transaction{
+					simpleTransactionFactory("tx1", "addr1", "100", currency),
+					simpleTransactionFactory("tx2", "addr1", "150", currency),
+					simpleTransactionFactory("tx3", "addr2", "150", currency),
+				},
+				Timestamp: asserter.MinUnixEpoch + 1,
+			},
+			orphan: true,
+			changes: []*reconciler.BalanceChange{
+				{
+					Account: &types.AccountIdentifier{
+						Address: "addr1",
+					},
+					Currency: currency,
+					Block: &types.BlockIdentifier{
+						Hash:  "0",
+						Index: 0,
+					},
+					Difference: "-250",
+				},
+				{
+					Account: &types.AccountIdentifier{
+						Address: "addr2",
+					},
+					Currency: currency,
+					Block: &types.BlockIdentifier{
+						Hash:  "0",
+						Index: 0,
+					},
+					Difference: "-150",
+				},
+			},
+			err: nil,
+		},
+	}
+
+	ctx := context.Background()
+
+	newDir, err := utils.CreateTempDir()
+	assert.NoError(t, err)
+	defer utils.RemoveTempDir(newDir)
+
+	database, err := NewBadgerStorage(ctx, newDir)
+	assert.NoError(t, err)
+	defer database.Close(ctx)
+
+	for name, test := range tests {
+		t.Run(name, func(t *testing.T) {
+			storage := NewBlockStorage(ctx, database, &MockBlockStorageHelper{
+				ExemptAccounts: test.exemptAccounts,
+			})
+
+			changes, err := storage.BalanceChanges(
+				ctx,
+				test.block,
+				test.orphan,
+			)
+
+			assert.ElementsMatch(t, test.changes, changes)
+			assert.Equal(t, test.err, err)
+		})
+	}
+}
+
+func TestCreateBlockCache(t *testing.T) {
+	ctx := context.Background()
+
+	newDir, err := utils.CreateTempDir()
+	assert.NoError(t, err)
+	defer utils.RemoveTempDir(newDir)
+
+	database, err := NewBadgerStorage(ctx, newDir)
+	assert.NoError(t, err)
+	defer database.Close(ctx)
+
+	storage := NewBlockStorage(ctx, database, &MockBlockStorageHelper{})
+
+	t.Run("no blocks processed", func(t *testing.T) {
+		assert.Equal(t, []*types.BlockIdentifier{}, storage.CreateBlockCache(ctx))
+	})
+
+	t.Run("1 block processed", func(t *testing.T) {
+		_, err = storage.StoreBlock(ctx, newBlock)
+		assert.NoError(t, err)
+		assert.Equal(
+			t,
+			[]*types.BlockIdentifier{newBlock.BlockIdentifier},
+			storage.CreateBlockCache(ctx),
+		)
+	})
+
+	t.Run("2 blocks processed", func(t *testing.T) {
+		_, err = storage.StoreBlock(ctx, newBlock3)
+		assert.NoError(t, err)
+		assert.Equal(
+			t,
+			[]*types.BlockIdentifier{newBlock.BlockIdentifier, newBlock3.BlockIdentifier},
+			storage.CreateBlockCache(ctx),
+		)
+	})
+}
+
+type MockBlockStorageHelper struct {
+	AccountBalanceAmount string
+	ExemptAccounts       []*reconciler.AccountCurrency
+}
+
+func (h *MockBlockStorageHelper) AccountBalance(
+	ctx context.Context,
+	account *types.AccountIdentifier,
+	currency *types.Currency,
+	block *types.BlockIdentifier,
+) (*types.Amount, error) {
+	value := "0"
+	if len(h.AccountBalanceAmount) > 0 {
+		value = h.AccountBalanceAmount
+	}
+
+	return &types.Amount{
+		Value:    value,
+		Currency: currency,
+	}, nil
+}
+
+func (h *MockBlockStorageHelper) SkipOperation(
+	ctx context.Context,
+	op *types.Operation,
+) (bool, error) {
+	if op.Account == nil || op.Amount == nil {
+		return true, nil
+	}
+
+	if op.Status == "Failure" {
+		return true, nil
+	}
+
+	if reconciler.ContainsAccountCurrency(h.ExemptAccounts, &reconciler.AccountCurrency{
+		Account:  op.Account,
+		Currency: op.Amount.Currency,
+	}) {
+		return true, nil
+	}
+
+	return false, nil
+}
diff --git a/internal/syncer/stateful_syncer_test.go b/internal/syncer/stateful_syncer_test.go
deleted file mode 100644
index ec5218b..0000000
--- a/internal/syncer/stateful_syncer_test.go
+++ /dev/null
@@ -1,495 +0,0 @@
-// Copyright 2020 Coinbase, Inc.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//      http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package syncer
-
-import (
-	"context"
-	"fmt"
-	"testing"
-
-	"github.com/coinbase/rosetta-cli/internal/storage"
-
-	"github.com/coinbase/rosetta-sdk-go/asserter"
-	"github.com/coinbase/rosetta-sdk-go/fetcher"
-	"github.com/coinbase/rosetta-sdk-go/types"
-
-	"github.com/stretchr/testify/assert"
-)
-
-var (
-	networkIdentifier = &types.NetworkIdentifier{
-		Blockchain: "blah",
-		Network:    "testnet",
-	}
-
-	currency = &types.Currency{
-		Symbol:   "Blah",
-		Decimals: 2,
-	}
-
-	recipient = &types.AccountIdentifier{
-		Address: "acct1",
-	}
-
-	recipientAmount = &types.Amount{
-		Value:    "100",
-		Currency: currency,
-	}
-
-	recipientOperation = &types.Operation{
-		OperationIdentifier: &types.OperationIdentifier{
-			Index: 0,
-		},
-		Type:    "Transfer",
-		Status:  "Success",
-		Account: recipient,
-		Amount:  recipientAmount,
-	}
-
-	recipientFailureOperation = &types.Operation{
-		OperationIdentifier: &types.OperationIdentifier{
-			Index: 1,
-		},
-		Type:    "Transfer",
-		Status:  "Failure",
-		Account: recipient,
-		Amount:  recipientAmount,
-	}
-
-	recipientTransaction = &types.Transaction{
-		TransactionIdentifier: &types.TransactionIdentifier{
-			Hash: "tx1",
-		},
-		Operations: []*types.Operation{
-			recipientOperation,
-			recipientFailureOperation,
-		},
-	}
-
-	sender = &types.AccountIdentifier{
-		Address: "acct2",
-	}
-
-	senderAmount = &types.Amount{
-		Value:    "-100",
-		Currency: currency,
-	}
-
-	senderOperation = &types.Operation{
-		OperationIdentifier: &types.OperationIdentifier{
-			Index: 0,
-		},
-		Type:    "Transfer",
-		Status:  "Success",
-		Account: sender,
-		Amount:  senderAmount,
-	}
-
-	senderTransaction = &types.Transaction{
-		TransactionIdentifier: &types.TransactionIdentifier{
-			Hash: "tx2",
-		},
-		Operations: []*types.Operation{
-			senderOperation,
-		},
-	}
-
-	orphanGenesis = &types.Block{
-		BlockIdentifier: &types.BlockIdentifier{
-			Hash:  "1",
-			Index: 1,
-		},
-		ParentBlockIdentifier: &types.BlockIdentifier{
-			Hash:  "0a",
-			Index: 0,
-		},
-		Transactions: []*types.Transaction{},
-	}
-
-	blockSequence = []*types.Block{
-		{ // genesis
-			BlockIdentifier: &types.BlockIdentifier{
-				Hash:  "0",
-				Index: 0,
-			},
-			ParentBlockIdentifier: &types.BlockIdentifier{
-				Hash:  "0",
-				Index: 0,
-			},
-		},
-		{
-			BlockIdentifier: &types.BlockIdentifier{
-				Hash:  "1",
-				Index: 1,
-			},
-			ParentBlockIdentifier: &types.BlockIdentifier{
-				Hash:  "0",
-				Index: 0,
-			},
-			Transactions: []*types.Transaction{
-				recipientTransaction,
-			},
-		},
-		{ // reorg
-			BlockIdentifier: &types.BlockIdentifier{
-				Hash:  "2",
-				Index: 2,
-			},
-			ParentBlockIdentifier: &types.BlockIdentifier{
-				Hash:  "1a",
-				Index: 1,
-			},
-		},
-		{
-			BlockIdentifier: &types.BlockIdentifier{
-				Hash:  "1a",
-				Index: 1,
-			},
-			ParentBlockIdentifier: &types.BlockIdentifier{
-				Hash:  "0",
-				Index: 0,
-			},
-		},
-		{
-			BlockIdentifier: &types.BlockIdentifier{
-				Hash:  "3",
-				Index: 3,
-			},
-			ParentBlockIdentifier: &types.BlockIdentifier{
-				Hash:  "2",
-				Index: 2,
-			},
-			Transactions: []*types.Transaction{
-				senderTransaction,
-			},
-		},
-		{ // invalid block
-			BlockIdentifier: &types.BlockIdentifier{
-				Hash:  "5",
-				Index: 5,
-			},
-			ParentBlockIdentifier: &types.BlockIdentifier{
-				Hash:  "4",
-				Index: 4,
-			},
-		},
-	}
-
-	operationStatuses = []*types.OperationStatus{
-		{
-			Status:     "Success",
-			Successful: true,
-		},
-		{
-			Status:     "Failure",
-			Successful: false,
-		},
-	}
-
-	networkStatusResponse = &types.NetworkStatusResponse{
-		GenesisBlockIdentifier: &types.BlockIdentifier{
-			Index: 0,
-			Hash:  "block 0",
-		},
-		CurrentBlockIdentifier: &types.BlockIdentifier{
-			Index: 10000,
-			Hash:  "block 1000",
-		},
-		CurrentBlockTimestamp: asserter.MinUnixEpoch + 1,
-		Peers: []*types.Peer{
-			{
-				PeerID: "peer 1",
-			},
-		},
-	}
-
-	networkOptionsResponse = &types.NetworkOptionsResponse{
-		Version: &types.Version{
-			RosettaVersion: "1.3.1",
-			NodeVersion:    "1.0",
-		},
-		Allow: &types.Allow{
-			OperationStatuses: operationStatuses,
-			OperationTypes: []string{
-				"Transfer",
-			},
-		},
-	}
-)
-
-func assertCurrentIndex(
-	ctx context.Context,
-	t *testing.T,
-	syncer *StatefulSyncer,
-	expectedCurrentIndex int64,
-) {
-	currentIndex, err := syncer.CurrentIndex(ctx)
-	assert.NoError(t, err)
-	assert.Equal(t, expectedCurrentIndex, currentIndex)
-}
-
-func TestProcessBlock(t *testing.T) {
-	ctx := context.Background()
-
-	newDir, err := storage.CreateTempDir()
-	assert.NoError(t, err)
-	defer storage.RemoveTempDir(*newDir)
-
-	database, err := storage.NewBadgerStorage(ctx, *newDir)
-	assert.NoError(t, err)
-	defer database.Close(ctx)
-
-	blockStorage := storage.NewBlockStorage(ctx, database)
-	asserter, err := asserter.NewClientWithResponses(
-		networkIdentifier,
-		networkStatusResponse,
-		networkOptionsResponse,
-	)
-	assert.NotNil(t, asserter)
-	assert.NoError(t, err)
-
-	fetcher := &fetcher.Fetcher{
-		Asserter: asserter,
-	}
-	syncer := NewStateful(nil, blockStorage, fetcher, nil)
-	currIndex := int64(0)
-	genesisIndex := blockSequence[0].BlockIdentifier.Index
-
-	t.Run("No block exists", func(t *testing.T) {
-		// Add genesis block
-		balanceChanges, newIndex, reorg, err := syncer.processBlock(
-			ctx,
-			genesisIndex,
-			currIndex,
-			blockSequence[0],
-		)
-		currIndex = newIndex
-		assert.False(t, reorg)
-		assert.Equal(t, int64(1), currIndex)
-		assert.Equal(t, 0, len(balanceChanges))
-		assert.NoError(t, err)
-
-		tx := syncer.storage.NewDatabaseTransaction(ctx, false)
-		head, err := syncer.storage.GetHeadBlockIdentifier(ctx, tx)
-		tx.Discard(ctx)
-		assert.Equal(t, blockSequence[0].BlockIdentifier, head)
-		assert.NoError(t, err)
-
-		assertCurrentIndex(ctx, t, syncer, currIndex)
-	})
-
-	t.Run("Orphan genesis", func(t *testing.T) {
-		balanceChanges, newIndex, reorg, err := syncer.processBlock(
-			ctx,
-			genesisIndex,
-			currIndex,
-			orphanGenesis,
-		)
-
-		assert.False(t, reorg)
-		assert.Equal(t, int64(0), newIndex)
-		assert.Equal(t, 0, len(balanceChanges))
-		assert.EqualError(t, err, "cannot orphan genesis block")
-
-		tx := syncer.storage.NewDatabaseTransaction(ctx, false)
-		head, err := syncer.storage.GetHeadBlockIdentifier(ctx, tx)
-		tx.Discard(ctx)
-		assert.Equal(t, blockSequence[0].BlockIdentifier, head)
-		assert.NoError(t, err)
-	})
-
-	t.Run("Block exists, no reorg", func(t *testing.T) {
-		balanceChanges, newIndex, reorg, err := syncer.processBlock(
-			ctx,
-			genesisIndex,
-			currIndex,
-			blockSequence[1],
-		)
-		currIndex = newIndex
-		assert.False(t, reorg)
-		assert.Equal(t, int64(2), currIndex)
-		assert.Equal(t, []*storage.BalanceChange{
-			{
-				Account: &types.AccountIdentifier{
-					Address: "acct1",
-				},
-				Currency:   currency,
-				Block:      blockSequence[1].BlockIdentifier,
-				Difference: "100",
-			},
-		}, balanceChanges)
-		assert.NoError(t, err)
-
-		tx := syncer.storage.NewDatabaseTransaction(ctx, false)
-		head, err := syncer.storage.GetHeadBlockIdentifier(ctx, tx)
-		assert.Equal(t, blockSequence[1].BlockIdentifier, head)
-		assert.NoError(t, err)
-
-		amounts, block, err := syncer.storage.GetBalance(ctx, tx, recipient)
-		tx.Discard(ctx)
-		assert.Equal(t, map[string]*types.Amount{
-			storage.GetCurrencyKey(currency): recipientAmount,
-		}, amounts)
-		assert.Equal(t, blockSequence[1].BlockIdentifier, block)
-		assert.NoError(t, err)
-
-		assertCurrentIndex(ctx, t, syncer, currIndex)
-	})
-
-	t.Run("Orphan block", func(t *testing.T) {
-		// Orphan block
-		balanceChanges, newIndex, reorg, err := syncer.processBlock(
-			ctx,
-			genesisIndex,
-			currIndex,
-			blockSequence[2],
-		)
-		currIndex = newIndex
-		assert.True(t, reorg)
-		assert.Equal(t, int64(1), currIndex)
-		assert.Equal(t, []*storage.BalanceChange{
-			{
-				Account: &types.AccountIdentifier{
-					Address: "acct1",
-				},
-				Currency:   currency,
-				Block:      blockSequence[0].BlockIdentifier,
-				Difference: "-100",
-			},
-		}, balanceChanges)
-		assert.NoError(t, err)
-		assertCurrentIndex(ctx, t, syncer, currIndex)
-
-		// Assert head is back to genesis
-		tx := syncer.storage.NewDatabaseTransaction(ctx, false)
-		head, err := syncer.storage.GetHeadBlockIdentifier(ctx, tx)
-		assert.Equal(t, blockSequence[0].BlockIdentifier, head)
-		assert.NoError(t, err)
-
-		// Assert that balance change was reverted
-		// only by the successful operation
-		zeroAmount := map[string]*types.Amount{
-			storage.GetCurrencyKey(currency): {
-				Value:    "0",
-				Currency: currency,
-			},
-		}
-		amounts, block, err := syncer.storage.GetBalance(ctx, tx, recipient)
-		assert.Equal(t, zeroAmount, amounts)
-		assert.Equal(t, blockSequence[0].BlockIdentifier, block)
-		assert.NoError(t, err)
-
-		// Assert block is gone
-		orphanBlock, err := syncer.storage.GetBlock(ctx, tx, blockSequence[1].BlockIdentifier)
-		assert.Nil(t, orphanBlock)
-		assert.EqualError(t, err, fmt.Errorf(
-			"%w %+v",
-			storage.ErrBlockNotFound,
-			blockSequence[1].BlockIdentifier,
-		).Error())
-		tx.Discard(ctx)
-
-		// Process new block
-		balanceChanges, currIndex, reorg, err = syncer.processBlock(
-			ctx,
-			genesisIndex,
-			currIndex,
-			blockSequence[3],
-		)
-		assert.False(t, reorg)
-		assert.Equal(t, int64(2), currIndex)
-		assert.Equal(t, 0, len(balanceChanges))
-		assert.NoError(t, err)
-		assertCurrentIndex(ctx, t, syncer, currIndex)
-
-		tx = syncer.storage.NewDatabaseTransaction(ctx, false)
-		head, err = syncer.storage.GetHeadBlockIdentifier(ctx, tx)
-		tx.Discard(ctx)
-		assert.Equal(t, blockSequence[3].BlockIdentifier, head)
-		assert.NoError(t, err)
-
-		balanceChanges, currIndex, reorg, err = syncer.processBlock(
-			ctx,
-			genesisIndex,
-			currIndex,
-			blockSequence[2],
-		)
-		assert.False(t, reorg)
-		assert.Equal(t, int64(3), currIndex)
-		assert.Equal(t, 0, len(balanceChanges))
-		assert.NoError(t, err)
-
-		tx = syncer.storage.NewDatabaseTransaction(ctx, false)
-		head, err = syncer.storage.GetHeadBlockIdentifier(ctx, tx)
-		assert.Equal(t, blockSequence[2].BlockIdentifier, head)
-		assert.NoError(t, err)
-
-		amounts, block, err = syncer.storage.GetBalance(ctx, tx, recipient)
-		tx.Discard(ctx)
-		assert.Equal(t, zeroAmount, amounts)
-		assert.Equal(t, blockSequence[0].BlockIdentifier, block)
-		assert.NoError(t, err)
-	})
-
-	t.Run("Block with invalid transaction", func(t *testing.T) {
-		balanceChanges, currIndex, reorg, err := syncer.processBlock(
-			ctx,
-			genesisIndex,
-			currIndex,
-			blockSequence[4],
-		)
-		assert.False(t, reorg)
-		assert.Equal(t, int64(3), currIndex)
-		assert.Equal(t, 0, len(balanceChanges))
-		assert.Contains(t, err.Error(), storage.ErrNegativeBalance.Error())
-
-		tx := syncer.storage.NewDatabaseTransaction(ctx, false)
-		head, err := syncer.storage.GetHeadBlockIdentifier(ctx, tx)
-		tx.Discard(ctx)
-		assert.NoError(t, err)
-		assert.Equal(t, blockSequence[2].BlockIdentifier, head)
-	})
-
-	t.Run("Out of order block", func(t *testing.T) {
-		balanceChanges, newIndex, reorg, err := syncer.processBlock(
-			ctx,
-			genesisIndex,
-			currIndex,
-			blockSequence[5],
-		)
-		currIndex = newIndex
-		assert.False(t, reorg)
-		assert.Equal(t, int64(3), currIndex)
-		assert.Equal(t, 0, len(balanceChanges))
-		assert.EqualError(t, err, "Got block 5 instead of 3")
-
-		tx := syncer.storage.NewDatabaseTransaction(ctx, false)
-		head, err := syncer.storage.GetHeadBlockIdentifier(ctx, tx)
-		tx.Discard(ctx)
-		assert.Equal(t, blockSequence[2].BlockIdentifier, head)
-		assert.NoError(t, err)
-	})
-
-	t.Run("Revert all blocks after genesis", func(t *testing.T) {
-		err := syncer.newHeadIndex(ctx, genesisIndex)
-		assert.NoError(t, err)
-
-		tx := syncer.storage.NewDatabaseTransaction(ctx, false)
-		head, err := syncer.storage.GetHeadBlockIdentifier(ctx, tx)
-		tx.Discard(ctx)
-		assert.Equal(t, blockSequence[0].BlockIdentifier, head)
-		assert.NoError(t, err)
-	})
-}
diff --git a/internal/syncer/syncer_test.go b/internal/syncer/syncer_test.go
index 3fa7e92..c7d8fce 100644
--- a/internal/syncer/syncer_test.go
+++ b/internal/syncer/syncer_test.go
@@ -18,217 +18,315 @@ import (
 	"context"
 	"testing"
 
-	"github.com/coinbase/rosetta-cli/internal/reconciler"
-	"github.com/coinbase/rosetta-cli/internal/storage"
-
-	"github.com/coinbase/rosetta-sdk-go/asserter"
 	"github.com/coinbase/rosetta-sdk-go/types"
+
 	"github.com/stretchr/testify/assert"
 )
 
-func simpleTransactionFactory(
-	hash string,
-	address string,
-	value string,
-	currency *types.Currency,
-) *types.Transaction {
-	return &types.Transaction{
+var (
+	networkIdentifier = &types.NetworkIdentifier{
+		Blockchain: "blah",
+		Network:    "testnet",
+	}
+
+	currency = &types.Currency{
+		Symbol:   "Blah",
+		Decimals: 2,
+	}
+
+	recipient = &types.AccountIdentifier{
+		Address: "acct1",
+	}
+
+	recipientAmount = &types.Amount{
+		Value:    "100",
+		Currency: currency,
+	}
+
+	recipientOperation = &types.Operation{
+		OperationIdentifier: &types.OperationIdentifier{
+			Index: 0,
+		},
+		Type:    "Transfer",
+		Status:  "Success",
+		Account: recipient,
+		Amount:  recipientAmount,
+	}
+
+	recipientFailureOperation = &types.Operation{
+		OperationIdentifier: &types.OperationIdentifier{
+			Index: 1,
+		},
+		Type:    "Transfer",
+		Status:  "Failure",
+		Account: recipient,
+		Amount:  recipientAmount,
+	}
+
+	recipientTransaction = &types.Transaction{
 		TransactionIdentifier: &types.TransactionIdentifier{
-			Hash: hash,
+			Hash: "tx1",
 		},
 		Operations: []*types.Operation{
-			{
-				OperationIdentifier: &types.OperationIdentifier{
-					Index: 0,
-				},
-				Type:   "Transfer",
-				Status: "Success",
-				Account: &types.AccountIdentifier{
-					Address: address,
-				},
-				Amount: &types.Amount{
-					Value:    value,
-					Currency: currency,
-				},
-			},
+			recipientOperation,
+			recipientFailureOperation,
 		},
 	}
-}
 
-func TestBalanceChanges(t *testing.T) {
-	var tests = map[string]struct {
-		block          *types.Block
-		orphan         bool
-		changes        []*storage.BalanceChange
-		exemptAccounts []*reconciler.AccountCurrency
-		err            error
-	}{
-		"simple block": {
-			block: &types.Block{
-				BlockIdentifier: &types.BlockIdentifier{
-					Hash:  "1",
-					Index: 1,
-				},
-				ParentBlockIdentifier: &types.BlockIdentifier{
-					Hash:  "0",
-					Index: 0,
-				},
-				Transactions: []*types.Transaction{
-					recipientTransaction,
-				},
-				Timestamp: asserter.MinUnixEpoch + 1,
-			},
-			orphan: false,
-			changes: []*storage.BalanceChange{
-				{
-					Account:  recipient,
-					Currency: currency,
-					Block: &types.BlockIdentifier{
-						Hash:  "1",
-						Index: 1,
-					},
-					Difference: "100",
-				},
-			},
-			err: nil,
+	sender = &types.AccountIdentifier{
+		Address: "acct2",
+	}
+
+	senderAmount = &types.Amount{
+		Value:    "-100",
+		Currency: currency,
+	}
+
+	senderOperation = &types.Operation{
+		OperationIdentifier: &types.OperationIdentifier{
+			Index: 0,
 		},
-		"simple block account exempt": {
-			block: &types.Block{
-				BlockIdentifier: &types.BlockIdentifier{
-					Hash:  "1",
-					Index: 1,
-				},
-				ParentBlockIdentifier: &types.BlockIdentifier{
-					Hash:  "0",
-					Index: 0,
-				},
-				Transactions: []*types.Transaction{
-					recipientTransaction,
-				},
-				Timestamp: asserter.MinUnixEpoch + 1,
-			},
-			orphan:  false,
-			changes: []*storage.BalanceChange{},
-			exemptAccounts: []*reconciler.AccountCurrency{
-				{
-					Account:  recipient,
-					Currency: currency,
-				},
-			},
-			err: nil,
+		Type:    "Transfer",
+		Status:  "Success",
+		Account: sender,
+		Amount:  senderAmount,
+	}
+
+	senderTransaction = &types.Transaction{
+		TransactionIdentifier: &types.TransactionIdentifier{
+			Hash: "tx2",
+		},
+		Operations: []*types.Operation{
+			senderOperation,
 		},
-		"single account sum block": {
-			block: &types.Block{
-				BlockIdentifier: &types.BlockIdentifier{
-					Hash:  "1",
-					Index: 1,
-				},
-				ParentBlockIdentifier: &types.BlockIdentifier{
-					Hash:  "0",
-					Index: 0,
-				},
-				Transactions: []*types.Transaction{
-					simpleTransactionFactory("tx1", "addr1", "100", currency),
-					simpleTransactionFactory("tx2", "addr1", "150", currency),
-					simpleTransactionFactory("tx3", "addr2", "150", currency),
-				},
-				Timestamp: asserter.MinUnixEpoch + 1,
-			},
-			orphan: false,
-			changes: []*storage.BalanceChange{
-				{
-					Account: &types.AccountIdentifier{
-						Address: "addr1",
-					},
-					Currency: currency,
-					Block: &types.BlockIdentifier{
-						Hash:  "1",
-						Index: 1,
-					},
-					Difference: "250",
-				},
-				{
-					Account: &types.AccountIdentifier{
-						Address: "addr2",
-					},
-					Currency: currency,
-					Block: &types.BlockIdentifier{
-						Hash:  "1",
-						Index: 1,
-					},
-					Difference: "150",
-				},
-			},
-			err: nil,
+	}
+
+	orphanGenesis = &types.Block{
+		BlockIdentifier: &types.BlockIdentifier{
+			Hash:  "1",
+			Index: 1,
 		},
-		"single account sum orphan block": {
-			block: &types.Block{
-				BlockIdentifier: &types.BlockIdentifier{
-					Hash:  "1",
-					Index: 1,
-				},
-				ParentBlockIdentifier: &types.BlockIdentifier{
-					Hash:  "0",
-					Index: 0,
-				},
-				Transactions: []*types.Transaction{
-					simpleTransactionFactory("tx1", "addr1", "100", currency),
-					simpleTransactionFactory("tx2", "addr1", "150", currency),
-					simpleTransactionFactory("tx3", "addr2", "150", currency),
-				},
-				Timestamp: asserter.MinUnixEpoch + 1,
-			},
-			orphan: true,
-			changes: []*storage.BalanceChange{
-				{
-					Account: &types.AccountIdentifier{
-						Address: "addr1",
-					},
-					Currency: currency,
-					Block: &types.BlockIdentifier{
-						Hash:  "0",
-						Index: 0,
-					},
-					Difference: "-250",
-				},
-				{
-					Account: &types.AccountIdentifier{
-						Address: "addr2",
-					},
-					Currency: currency,
-					Block: &types.BlockIdentifier{
-						Hash:  "0",
-						Index: 0,
-					},
-					Difference: "-150",
-				},
-			},
-			err: nil,
+		ParentBlockIdentifier: &types.BlockIdentifier{
+			Hash:  "0a",
+			Index: 0,
 		},
+		Transactions: []*types.Transaction{},
 	}
 
-	ctx := context.Background()
-	asserter, err := asserter.NewClientWithResponses(
-		networkIdentifier,
-		networkStatusResponse,
-		networkOptionsResponse,
-	)
-	assert.NotNil(t, asserter)
-	assert.NoError(t, err)
-
-	for name, test := range tests {
-		t.Run(name, func(t *testing.T) {
-			handler := NewBaseHandler(nil, nil, test.exemptAccounts)
-			changes, err := BalanceChanges(
-				ctx,
-				asserter,
-				test.block,
-				test.orphan,
-				handler,
-			)
-
-			assert.ElementsMatch(t, test.changes, changes)
-			assert.Equal(t, test.err, err)
-		})
+	blockSequence = []*types.Block{
+		{ // genesis
+			BlockIdentifier: &types.BlockIdentifier{
+				Hash:  "0",
+				Index: 0,
+			},
+			ParentBlockIdentifier: &types.BlockIdentifier{
+				Hash:  "0",
+				Index: 0,
+			},
+		},
+		{
+			BlockIdentifier: &types.BlockIdentifier{
+				Hash:  "1",
+				Index: 1,
+			},
+			ParentBlockIdentifier: &types.BlockIdentifier{
+				Hash:  "0",
+				Index: 0,
+			},
+			Transactions: []*types.Transaction{
+				recipientTransaction,
+			},
+		},
+		{ // reorg
+			BlockIdentifier: &types.BlockIdentifier{
+				Hash:  "2",
+				Index: 2,
+			},
+			ParentBlockIdentifier: &types.BlockIdentifier{
+				Hash:  "1a",
+				Index: 1,
+			},
+		},
+		{
+			BlockIdentifier: &types.BlockIdentifier{
+				Hash:  "1a",
+				Index: 1,
+			},
+			ParentBlockIdentifier: &types.BlockIdentifier{
+				Hash:  "0",
+				Index: 0,
+			},
+		},
+		{
+			BlockIdentifier: &types.BlockIdentifier{
+				Hash:  "3",
+				Index: 3,
+			},
+			ParentBlockIdentifier: &types.BlockIdentifier{
+				Hash:  "2",
+				Index: 2,
+			},
+			Transactions: []*types.Transaction{
+				senderTransaction,
+			},
+		},
+		{ // invalid block
+			BlockIdentifier: &types.BlockIdentifier{
+				Hash:  "5",
+				Index: 5,
+			},
+			ParentBlockIdentifier: &types.BlockIdentifier{
+				Hash:  "4",
+				Index: 4,
+			},
+		},
 	}
+)
+
+func lastBlockIdentifier(syncer *Syncer) *types.BlockIdentifier {
+	return syncer.pastBlocks[len(syncer.pastBlocks)-1]
+}
+
+func TestProcessBlock(t *testing.T) {
+	ctx := context.Background()
+
+	syncer := New(networkIdentifier, nil, &MockSyncHandler{}, nil, nil)
+	syncer.genesisBlock = blockSequence[0].BlockIdentifier
+
+	t.Run("No block exists", func(t *testing.T) {
+		assert.Equal(
+			t,
+			[]*types.BlockIdentifier{},
+			syncer.pastBlocks,
+		)
+		err := syncer.processBlock(
+			ctx,
+			blockSequence[0],
+		)
+		assert.NoError(t, err)
+		assert.Equal(t, int64(1), syncer.nextIndex)
+		assert.Equal(t, blockSequence[0].BlockIdentifier, lastBlockIdentifier(syncer))
+		assert.Equal(
+			t,
+			[]*types.BlockIdentifier{blockSequence[0].BlockIdentifier},
+			syncer.pastBlocks,
+		)
+	})
+
+	t.Run("Orphan genesis", func(t *testing.T) {
+		err := syncer.processBlock(
+			ctx,
+			orphanGenesis,
+		)
+
+		assert.EqualError(t, err, "cannot remove genesis block")
+		assert.Equal(t, int64(1), syncer.nextIndex)
+		assert.Equal(t, blockSequence[0].BlockIdentifier, lastBlockIdentifier(syncer))
+		assert.Equal(
+			t,
+			[]*types.BlockIdentifier{blockSequence[0].BlockIdentifier},
+			syncer.pastBlocks,
+		)
+	})
+
+	t.Run("Block exists, no reorg", func(t *testing.T) {
+		err := syncer.processBlock(
+			ctx,
+			blockSequence[1],
+		)
+		assert.NoError(t, err)
+		assert.Equal(t, int64(2), syncer.nextIndex)
+		assert.Equal(t, blockSequence[1].BlockIdentifier, lastBlockIdentifier(syncer))
+		assert.Equal(
+			t,
+			[]*types.BlockIdentifier{
+				blockSequence[0].BlockIdentifier,
+				blockSequence[1].BlockIdentifier,
+			},
+			syncer.pastBlocks,
+		)
+	})
+
+	t.Run("Orphan block", func(t *testing.T) {
+		err := syncer.processBlock(
+			ctx,
+			blockSequence[2],
+		)
+		assert.NoError(t, err)
+		assert.Equal(t, int64(1), syncer.nextIndex)
+		assert.Equal(t, blockSequence[0].BlockIdentifier, lastBlockIdentifier(syncer))
+		assert.Equal(
+			t,
+			[]*types.BlockIdentifier{blockSequence[0].BlockIdentifier},
+			syncer.pastBlocks,
+		)
+
+		err = syncer.processBlock(
+			ctx,
+			blockSequence[3],
+		)
+		assert.NoError(t, err)
+		assert.Equal(t, int64(2), syncer.nextIndex)
+		assert.Equal(t, blockSequence[3].BlockIdentifier, lastBlockIdentifier(syncer))
+		assert.Equal(
+			t,
+			[]*types.BlockIdentifier{
+				blockSequence[0].BlockIdentifier,
+				blockSequence[3].BlockIdentifier,
+			},
+			syncer.pastBlocks,
+		)
+
+		err = syncer.processBlock(
+			ctx,
+			blockSequence[2],
+		)
+		assert.NoError(t, err)
+		assert.Equal(t, int64(3), syncer.nextIndex)
+		assert.Equal(t, blockSequence[2].BlockIdentifier, lastBlockIdentifier(syncer))
+		assert.Equal(
+			t,
+			[]*types.BlockIdentifier{
+				blockSequence[0].BlockIdentifier,
+				blockSequence[3].BlockIdentifier,
+				blockSequence[2].BlockIdentifier,
+			},
+			syncer.pastBlocks,
+		)
+	})
+
+	t.Run("Out of order block", func(t *testing.T) {
+		err := syncer.processBlock(
+			ctx,
+			blockSequence[5],
+		)
+		assert.EqualError(t, err, "Got block 5 instead of 3")
+		assert.Equal(t, int64(3), syncer.nextIndex)
+		assert.Equal(t, blockSequence[2].BlockIdentifier, lastBlockIdentifier(syncer))
+		assert.Equal(
+			t,
+			[]*types.BlockIdentifier{
+				blockSequence[0].BlockIdentifier,
+				blockSequence[3].BlockIdentifier,
+				blockSequence[2].BlockIdentifier,
+			},
+			syncer.pastBlocks,
+		)
+	})
+}
+
+type MockSyncHandler struct{}
+
+func (h *MockSyncHandler) BlockAdded(
+	ctx context.Context,
+	block *types.Block,
+) error {
+	return nil
+}
+
+func (h *MockSyncHandler) BlockRemoved(
+	ctx context.Context,
+	block *types.BlockIdentifier,
+) error {
+	return nil
 }