Repository: coinbase/rosetta-cli
Pull request #169 (2020-10-19 16:54:26)

[check:data] Wait for reconciliation queue to drain
Related: https://github.com/coinbase/rosetta-sdk-go/pull/196
Related: https://github.com/coinbase/rosetta-sdk-go/pull/200

This PR prevents `rosetta-cli` from exiting until the entire reconciler backlog has been drained (can optionally disable). As a result, it is now possible to reach 100% reconciliation coverage.

### Changes
- [x] Update version
- [x] Add `ReconciliationDrainDisabled` to configure the `rosetta-cli` to exit before draining the backlog
- [x] Update to `rosetta-sdk-go@v0.5.6`
- [x] If `ReconciliationDrainDisabled` is false, wait for reconciliation queue to exit before draining
- [x] Ensure balances are bootstrapped before initializing reconciler


Patch:
diff --git a/cmd/check_construction.go b/cmd/check_construction.go
index b718363..c26d13b 100644
--- a/cmd/check_construction.go
+++ b/cmd/check_construction.go
@@ -157,6 +157,5 @@ func runCheckConstructionCmd(cmd *cobra.Command, args []string) error {
 	sigListeners := []context.CancelFunc{cancel}
 	go handleSignals(&sigListeners)
 
-	err = g.Wait()
-	return constructionTester.HandleErr(err, &sigListeners)
+	return constructionTester.HandleErr(g.Wait(), &sigListeners)
 }
diff --git a/cmd/check_data.go b/cmd/check_data.go
index 55762c9..d6c2b7f 100644
--- a/cmd/check_data.go
+++ b/cmd/check_data.go
@@ -158,13 +158,7 @@ func runCheckDataCmd(cmd *cobra.Command, args []string) error {
 	sigListeners := []context.CancelFunc{cancel}
 	go handleSignals(&sigListeners)
 
-	err = g.Wait()
-
-	// Initialize new context because calling context
-	// will no longer be usable when after termination.
-	ctx = context.Background()
-
 	// HandleErr will exit if we should not attempt
 	// to find missing operations.
-	return dataTester.HandleErr(ctx, err, &sigListeners)
+	return dataTester.HandleErr(g.Wait(), &sigListeners)
 }
diff --git a/cmd/root.go b/cmd/root.go
index 782c1cc..92a2712 100644
--- a/cmd/root.go
+++ b/cmd/root.go
@@ -247,6 +247,6 @@ var versionCmd = &cobra.Command{
 	Use:   "version",
 	Short: "Print rosetta-cli version",
 	Run: func(cmd *cobra.Command, args []string) {
-		fmt.Println("v0.5.14")
+		fmt.Println("v0.5.15")
 	},
 }
diff --git a/cmd/view_block.go b/cmd/view_block.go
index f6c5de5..4fc3ecd 100644
--- a/cmd/view_block.go
+++ b/cmd/view_block.go
@@ -98,7 +98,7 @@ func runViewBlockCmd(cmd *cobra.Command, args []string) error {
 
 	// Print out all balance changes in a given block. This does NOT exempt
 	// any operations/accounts from parsing.
-	p := parser.New(newFetcher.Asserter, func(*types.Operation) bool { return false })
+	p := parser.New(newFetcher.Asserter, func(*types.Operation) bool { return false }, nil)
 	changes, err := p.BalanceChanges(Context, block, false)
 	if err != nil {
 		return fmt.Errorf("%w: unable to calculate balance changes", err)
diff --git a/configuration/types.go b/configuration/types.go
index 90c7903..608cba6 100644
--- a/configuration/types.go
+++ b/configuration/types.go
@@ -258,6 +258,11 @@ type DataConfiguration struct {
 	// some of the more advanced checks to confirm syncing is working as expected.
 	ReconciliationDisabled bool `json:"reconciliation_disabled"`
 
+	// ReconciliationDrainDisabled is a boolean that configures the rosetta-cli
+	// to exit check:data before the entire active reconciliation queue has
+	// been drained (if reconciliation is enabled).
+	ReconciliationDrainDisabled bool `json:"reconciliation_drain_disabled"`
+
 	// InactiveDiscrepencySearchDisabled is a boolean indicating if a search
 	// should be performed to find any inactive reconciliation discrepencies.
 	// Note, a search will never be performed if historical balance lookup
diff --git a/examples/configuration/default.json b/examples/configuration/default.json
index 91b8218..8e6692c 100644
--- a/examples/configuration/default.json
+++ b/examples/configuration/default.json
@@ -28,6 +28,7 @@
   "bootstrap_balances": "",
   "interesting_accounts": "",
   "reconciliation_disabled": false,
+  "reconciliation_drain_disabled": false,
   "inactive_discrepency_search_disabled": false,
   "balance_tracking_disabled": false,
   "coin_tracking_disabled": false,
diff --git a/go.mod b/go.mod
index c3777a8..521935d 100644
--- a/go.mod
+++ b/go.mod
@@ -3,7 +3,7 @@ module github.com/coinbase/rosetta-cli
 go 1.13
 
 require (
-	github.com/coinbase/rosetta-sdk-go v0.5.5
+	github.com/coinbase/rosetta-sdk-go v0.5.6
 	github.com/fatih/color v1.9.0
 	github.com/olekukonko/tablewriter v0.0.2-0.20190409134802-7e037d187b0c
 	github.com/spf13/cobra v1.1.1
diff --git a/go.sum b/go.sum
index f54bd3b..f68fd0a 100644
--- a/go.sum
+++ b/go.sum
@@ -80,8 +80,8 @@ github.com/cespare/xxhash/v2 v2.1.1 h1:6MnRN8NT7+YBpUIWxHtefFZOKTAPgGjpQSxqLNn0+
 github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
 github.com/cloudflare/cloudflare-go v0.10.2-0.20190916151808-a80f83b9add9/go.mod h1:1MxXX1Ux4x6mqPmjkUgTP1CdXIBXKX7T+Jk9Gxrmx+U=
-github.com/coinbase/rosetta-sdk-go v0.5.5 h1:Z61/VUO89BDVl1m6Zj0e6OWMl5GnVevj4z79Eh3sSL0=
-github.com/coinbase/rosetta-sdk-go v0.5.5/go.mod h1:JRO4BJjhWAI7nYGwzYZWFgYfCTsQOvdZB7tGyN6kEtE=
+github.com/coinbase/rosetta-sdk-go v0.5.6 h1:mgSSw2NsVUBZPgeQegIWazJki0IT8Zd+xrFLS1afzJE=
+github.com/coinbase/rosetta-sdk-go v0.5.6/go.mod h1:JRO4BJjhWAI7nYGwzYZWFgYfCTsQOvdZB7tGyN6kEtE=
 github.com/coreos/bbolt v1.3.2/go.mod h1:iRUV2dpdMOn7Bo10OQBFzIJO9kkE559Wcmn+qkEiiKk=
 github.com/coreos/etcd v3.3.10+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
 github.com/coreos/etcd v3.3.13+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
diff --git a/pkg/logger/logger.go b/pkg/logger/logger.go
index c639a2a..f614271 100644
--- a/pkg/logger/logger.go
+++ b/pkg/logger/logger.go
@@ -97,7 +97,7 @@ func (l *Logger) LogDataStatus(ctx context.Context, status *results.CheckDataSta
 	}
 
 	statsMessage := fmt.Sprintf(
-		"[STATS] Blocks: %d (Orphaned: %d) Transactions: %d Operations: %d Reconciliations: %d (Inactive: %d, Exempt: %d, Coverage: %f%%)", // nolint:lll
+		"[STATS] Blocks: %d (Orphaned: %d) Transactions: %d Operations: %d Reconciliations: %d (Inactive: %d, Exempt: %d, Skipped: %d, Coverage: %f%%)", // nolint:lll
 		status.Stats.Blocks,
 		status.Stats.Orphans,
 		status.Stats.Transactions,
@@ -105,6 +105,7 @@ func (l *Logger) LogDataStatus(ctx context.Context, status *results.CheckDataSta
 		status.Stats.ActiveReconciliations+status.Stats.InactiveReconciliations,
 		status.Stats.InactiveReconciliations,
 		status.Stats.ExemptReconciliations,
+		status.Stats.SkippedReconciliations,
 		status.Stats.ReconciliationCoverage*utils.OneHundred,
 	)
 
diff --git a/pkg/processor/balance_storage_helper.go b/pkg/processor/balance_storage_helper.go
index b55cbcc..10f2d7e 100644
--- a/pkg/processor/balance_storage_helper.go
+++ b/pkg/processor/balance_storage_helper.go
@@ -38,6 +38,7 @@ type BalanceStorageHelper struct {
 	// Configuration settings
 	lookupBalanceByBlock bool
 	exemptAccounts       map[string]struct{}
+	balanceExemptions    []*types.BalanceExemption
 
 	// Interesting-only Parsing
 	interestingOnly      bool
@@ -51,6 +52,7 @@ func NewBalanceStorageHelper(
 	lookupBalanceByBlock bool,
 	exemptAccounts []*reconciler.AccountCurrency,
 	interestingOnly bool,
+	balanceExemptions []*types.BalanceExemption,
 ) *BalanceStorageHelper {
 	exemptMap := map[string]struct{}{}
 
@@ -67,6 +69,7 @@ func NewBalanceStorageHelper(
 		exemptAccounts:       exemptMap,
 		interestingAddresses: map[string]struct{}{},
 		interestingOnly:      interestingOnly,
+		balanceExemptions:    balanceExemptions,
 	}
 }
 
@@ -137,3 +140,8 @@ func (h *BalanceStorageHelper) ExemptFunc() parser.ExemptOperation {
 		return exists
 	}
 }
+
+// BalanceExemptions returns a list of *types.BalanceExemption.
+func (h *BalanceStorageHelper) BalanceExemptions() []*types.BalanceExemption {
+	return h.balanceExemptions
+}
diff --git a/pkg/processor/reconciler_handler.go b/pkg/processor/reconciler_handler.go
index 53c0fe6..fa228a6 100644
--- a/pkg/processor/reconciler_handler.go
+++ b/pkg/processor/reconciler_handler.go
@@ -69,6 +69,8 @@ func (h *ReconcilerHandler) ReconciliationFailed(
 	liveBalance string,
 	block *types.BlockIdentifier,
 ) error {
+	_, _ = h.counterStorage.Update(ctx, storage.FailedReconciliationCounter, big.NewInt(1))
+
 	err := h.logger.ReconcileFailureStream(
 		ctx,
 		reconciliationType,
@@ -144,6 +146,19 @@ func (h *ReconcilerHandler) ReconciliationExempt(
 	return nil
 }
 
+// ReconciliationSkipped is called each time a reconciliation is skipped.
+func (h *ReconcilerHandler) ReconciliationSkipped(
+	ctx context.Context,
+	reconciliationType string,
+	account *types.AccountIdentifier,
+	currency *types.Currency,
+	cause string,
+) error {
+	_, _ = h.counterStorage.Update(ctx, storage.SkippedReconciliationsCounter, big.NewInt(1))
+
+	return nil
+}
+
 // ReconciliationSucceeded is called each time a reconciliation succeeds.
 func (h *ReconcilerHandler) ReconciliationSucceeded(
 	ctx context.Context,
@@ -154,16 +169,13 @@ func (h *ReconcilerHandler) ReconciliationSucceeded(
 	block *types.BlockIdentifier,
 ) error {
 	// Update counters
+	counter := storage.ActiveReconciliationCounter
 	if reconciliationType == reconciler.InactiveReconciliation {
-		_, _ = h.counterStorage.Update(
-			ctx,
-			storage.InactiveReconciliationCounter,
-			big.NewInt(1),
-		)
-	} else {
-		_, _ = h.counterStorage.Update(ctx, storage.ActiveReconciliationCounter, big.NewInt(1))
+		counter = storage.InactiveReconciliationCounter
 	}
 
+	_, _ = h.counterStorage.Update(ctx, counter, big.NewInt(1))
+
 	if err := h.balanceStorage.Reconciled(ctx, account, currency, block); err != nil {
 		return fmt.Errorf("%w: unable to store updated reconciliation", err)
 	}
diff --git a/pkg/results/data_results.go b/pkg/results/data_results.go
index 99f2104..807bbaf 100644
--- a/pkg/results/data_results.go
+++ b/pkg/results/data_results.go
@@ -36,6 +36,11 @@ import (
 	"github.com/olekukonko/tablewriter"
 )
 
+var (
+	f  = false
+	tr = true
+)
+
 // EndCondition contains the type of
 // end condition and any detail associated
 // with the stop.
@@ -98,6 +103,8 @@ type CheckDataStats struct {
 	ActiveReconciliations   int64   `json:"active_reconciliations"`
 	InactiveReconciliations int64   `json:"inactive_reconciliations"`
 	ExemptReconciliations   int64   `json:"exempt_reconciliations"`
+	FailedReconciliations   int64   `json:"failed_reconciliations"`
+	SkippedReconciliations  int64   `json:"skipped_reconciliations"`
 	ReconciliationCoverage  float64 `json:"reconciliation_coverage"`
 }
 
@@ -140,6 +147,20 @@ func (c *CheckDataStats) Print() {
 			strconv.FormatInt(c.ExemptReconciliations, 10),
 		},
 	)
+	table.Append(
+		[]string{
+			"Failed Reconciliations",
+			"# of reconciliation failures",
+			strconv.FormatInt(c.FailedReconciliations, 10),
+		},
+	)
+	table.Append(
+		[]string{
+			"Skipped Reconciliations",
+			"# of reconciliations skipped",
+			strconv.FormatInt(c.SkippedReconciliations, 10),
+		},
+	)
 	table.Append(
 		[]string{
 			"Reconciliation Coverage",
@@ -203,6 +224,18 @@ func ComputeCheckDataStats(
 		return nil
 	}
 
+	failedReconciliations, err := counters.Get(ctx, storage.FailedReconciliationCounter)
+	if err != nil {
+		log.Printf("%s: cannot get failed reconciliations counter", err.Error())
+		return nil
+	}
+
+	skippedReconciliations, err := counters.Get(ctx, storage.SkippedReconciliationsCounter)
+	if err != nil {
+		log.Printf("%s: cannot get skipped reconciliations counter", err.Error())
+		return nil
+	}
+
 	stats := &CheckDataStats{
 		Blocks:                  blocks.Int64(),
 		Orphans:                 orphans.Int64(),
@@ -211,6 +244,8 @@ func ComputeCheckDataStats(
 		ActiveReconciliations:   activeReconciliations.Int64(),
 		InactiveReconciliations: inactiveReconciliations.Int64(),
 		ExemptReconciliations:   exemptReconciliations.Int64(),
+		FailedReconciliations:   failedReconciliations.Int64(),
+		SkippedReconciliations:  skippedReconciliations.Int64(),
 	}
 
 	if balances != nil {
@@ -476,29 +511,27 @@ func ReconciliationTest(
 	cfg *configuration.Configuration,
 	err error,
 	reconciliationsPerformed bool,
+	reconciliationsFailed bool,
 ) *bool {
-	relatedErrors := []error{
-		ErrReconciliationFailure,
-	}
-	reconciliationPass := true
-	for _, relatedError := range relatedErrors {
-		if errors.Is(err, relatedError) {
-			reconciliationPass = false
-			break
-		}
+	if errors.Is(err, ErrReconciliationFailure) {
+		return &f
 	}
 
-	if (cfg.Data.BalanceTrackingDisabled || cfg.Data.ReconciliationDisabled || cfg.Data.IgnoreReconciliationError ||
-		!reconciliationsPerformed) &&
-		reconciliationPass {
+	if cfg.Data.BalanceTrackingDisabled ||
+		cfg.Data.ReconciliationDisabled ||
+		(!reconciliationsPerformed && !reconciliationsFailed) {
 		return nil
 	}
 
-	return &reconciliationPass
+	if reconciliationsFailed {
+		return &f
+	}
+
+	return &tr
 }
 
 // ComputeCheckDataTests returns a populated CheckDataTests.
-func ComputeCheckDataTests(
+func ComputeCheckDataTests( // nolint:gocognit
 	ctx context.Context,
 	cfg *configuration.Configuration,
 	err error,
@@ -506,6 +539,7 @@ func ComputeCheckDataTests(
 ) *CheckDataTests {
 	operationsSeen := false
 	reconciliationsPerformed := false
+	reconciliationsFailed := false
 	blocksSynced := false
 	if counterStorage != nil {
 		blocks, err := counterStorage.Get(ctx, storage.BlockCounter)
@@ -538,6 +572,15 @@ func ComputeCheckDataTests(
 		if err == nil && exemptReconciliations.Int64() > 0 {
 			reconciliationsPerformed = true
 		}
+
+		failedReconciliations, err := counterStorage.Get(
+			ctx,
+			storage.FailedReconciliationCounter,
+		)
+		if err == nil && failedReconciliations.Int64() > 0 {
+			reconciliationsPerformed = true
+			reconciliationsFailed = true
+		}
 	}
 
 	return &CheckDataTests{
@@ -545,7 +588,7 @@ func ComputeCheckDataTests(
 		ResponseAssertion: ResponseAssertionTest(err),
 		BlockSyncing:      BlockSyncingTest(err, blocksSynced),
 		BalanceTracking:   BalanceTrackingTest(cfg, err, operationsSeen),
-		Reconciliation:    ReconciliationTest(cfg, err, reconciliationsPerformed),
+		Reconciliation:    ReconciliationTest(cfg, err, reconciliationsPerformed, reconciliationsFailed),
 	}
 }
 


Test patch:
diff --git a/pkg/processor/balance_storage_helper_test.go b/pkg/processor/balance_storage_helper_test.go
index f173153..a2cd7b5 100644
--- a/pkg/processor/balance_storage_helper_test.go
+++ b/pkg/processor/balance_storage_helper_test.go
@@ -84,6 +84,7 @@ func TestExemptFuncExemptAccounts(t *testing.T) {
 				false,
 				test.exemptAccounts,
 				false,
+				nil,
 			)
 
 			result := helper.ExemptFunc()(&types.Operation{
@@ -125,6 +126,7 @@ func TestExemptFuncInterestingParsing(t *testing.T) {
 				false,
 				nil,
 				true,
+				nil,
 			)
 
 			for _, addr := range test.interestingAddresses {
diff --git a/pkg/results/data_results_test.go b/pkg/results/data_results_test.go
index 68ff2f6..9addec7 100644
--- a/pkg/results/data_results_test.go
+++ b/pkg/results/data_results_test.go
@@ -33,11 +33,6 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
-var (
-	tr = true
-	f  = false
-)
-
 func TestComputeCheckDataResults(t *testing.T) {
 	var tests = map[string]struct {
 		cfg *configuration.Configuration
@@ -48,6 +43,7 @@ func TestComputeCheckDataResults(t *testing.T) {
 		operationCount          int64
 		activeReconciliations   int64
 		inactiveReconciliations int64
+		reconciliationFailures  int64
 
 		// balance storage values
 		provideBalanceStorage bool
@@ -303,6 +299,24 @@ func TestComputeCheckDataResults(t *testing.T) {
 				},
 			},
 		},
+		"default configuration, counter storage, reconciliation errors": {
+			cfg:                    configuration.DefaultConfiguration(),
+			err:                    []error{ErrReconciliationFailure},
+			provideCounterStorage:  true,
+			activeReconciliations:  10,
+			reconciliationFailures: 19,
+			result: &CheckDataResults{
+				Tests: &CheckDataTests{
+					RequestResponse:   true,
+					ResponseAssertion: true,
+					Reconciliation:    &f,
+				},
+				Stats: &CheckDataStats{
+					ActiveReconciliations: 10,
+					FailedReconciliations: 19,
+				},
+			},
+		},
 		"default configuration, no storage, unknown errors": {
 			cfg:    configuration.DefaultConfiguration(),
 			err:    []error{errors.New("unsure how to handle this error")},
@@ -383,6 +397,13 @@ func TestComputeCheckDataResults(t *testing.T) {
 						big.NewInt(test.inactiveReconciliations),
 					)
 					assert.NoError(t, err)
+
+					_, err = counterStorage.Update(
+						ctx,
+						storage.FailedReconciliationCounter,
+						big.NewInt(test.reconciliationFailures),
+					)
+					assert.NoError(t, err)
 				}
 
 				var balanceStorage *storage.BalanceStorage
diff --git a/pkg/tester/construction.go b/pkg/tester/construction.go
index 93c8df7..b2c46b1 100644
--- a/pkg/tester/construction.go
+++ b/pkg/tester/construction.go
@@ -97,6 +97,11 @@ func InitializeConstruction(
 		log.Fatalf("%s: unable to initialize database", err.Error())
 	}
 
+	networkOptions, fetchErr := onlineFetcher.NetworkOptionsRetry(ctx, network, nil)
+	if err != nil {
+		log.Fatalf("%s: unable to get network options", fetchErr.Err.Error())
+	}
+
 	counterStorage := storage.NewCounterStorage(localStore)
 	logger := logger.NewLogger(
 		dataPath,
@@ -118,6 +123,7 @@ func InitializeConstruction(
 		false,
 		nil,
 		true,
+		networkOptions.Allow.BalanceExemptions,
 	)
 
 	balanceStorageHandler := processor.NewBalanceStorageHandler(
@@ -138,7 +144,7 @@ func InitializeConstruction(
 		config.Construction.BlockBroadcastLimit,
 	)
 
-	parser := parser.New(onlineFetcher.Asserter, nil)
+	parser := parser.New(onlineFetcher.Asserter, nil, networkOptions.Allow.BalanceExemptions)
 	broadcastHelper := processor.NewBroadcastStorageHelper(
 		blockStorage,
 		onlineFetcher,
diff --git a/pkg/tester/data.go b/pkg/tester/data.go
index 7d93047..efe6ce9 100644
--- a/pkg/tester/data.go
+++ b/pkg/tester/data.go
@@ -30,6 +30,7 @@ import (
 	"github.com/coinbase/rosetta-cli/pkg/results"
 
 	"github.com/coinbase/rosetta-sdk-go/fetcher"
+	"github.com/coinbase/rosetta-sdk-go/parser"
 	"github.com/coinbase/rosetta-sdk-go/reconciler"
 	"github.com/coinbase/rosetta-sdk-go/statefulsyncer"
 	"github.com/coinbase/rosetta-sdk-go/storage"
@@ -84,6 +85,7 @@ type DataTester struct {
 	genesisBlock             *types.BlockIdentifier
 	cancel                   context.CancelFunc
 	historicalBalanceEnabled bool
+	parser                   *parser.Parser
 
 	endCondition       configuration.CheckDataEndCondition
 	endConditionDetail string
@@ -173,6 +175,28 @@ func InitializeData(
 	blockStorage := storage.NewBlockStorage(localStore)
 	balanceStorage := storage.NewBalanceStorage(localStore)
 
+	// Bootstrap balances, if provided. We need to do before initializing
+	// the reconciler otherwise we won't reconcile bootstrapped accounts
+	// until rosetta-cli restart.
+	if len(config.Data.BootstrapBalances) > 0 {
+		_, err := blockStorage.GetHeadBlockIdentifier(ctx)
+		switch {
+		case err == storage.ErrHeadBlockNotFound:
+			err = balanceStorage.BootstrapBalances(
+				ctx,
+				config.Data.BootstrapBalances,
+				genesisBlock,
+			)
+			if err != nil {
+				log.Fatalf("%s: unable to bootstrap balances", err.Error())
+			}
+		case err != nil:
+			log.Fatalf("%s: unable to get head block identifier", err.Error())
+		default:
+			log.Println("Skipping balance bootstrapping because already started syncing")
+		}
+	}
+
 	logger := logger.NewLogger(
 		dataPath,
 		config.Data.LogBlocks,
@@ -206,6 +230,12 @@ func InitializeData(
 		log.Fatalf("%s: unable to get network options", fetchErr.Err.Error())
 	}
 
+	parser := parser.New(
+		fetcher.Asserter,
+		nil,
+		networkOptions.Allow.BalanceExemptions,
+	)
+
 	// Determine if we should perform historical balance lookups
 	var historicalBalanceEnabled bool
 	if config.Data.HistoricalBalanceEnabled != nil {
@@ -217,6 +247,7 @@ func InitializeData(
 	r := reconciler.New(
 		reconcilerHelper,
 		reconcilerHandler,
+		parser,
 		reconciler.WithActiveConcurrency(int(config.Data.ActiveReconciliationConcurrency)),
 		reconciler.WithInactiveConcurrency(int(config.Data.InactiveReconciliationConcurrency)),
 		reconciler.WithLookupBalanceByBlock(historicalBalanceEnabled),
@@ -224,7 +255,6 @@ func InitializeData(
 		reconciler.WithSeenAccounts(seenAccounts),
 		reconciler.WithDebugLogging(config.Data.LogReconciliations),
 		reconciler.WithInactiveFrequency(int64(config.Data.InactiveReconciliationFrequency)),
-		reconciler.WithBalanceExemptions(networkOptions.Allow.BalanceExemptions),
 	)
 
 	blockWorkers := []storage.BlockWorker{}
@@ -235,6 +265,7 @@ func InitializeData(
 			historicalBalanceEnabled,
 			exemptAccounts,
 			false,
+			networkOptions.Allow.BalanceExemptions,
 		)
 
 		balanceStorageHandler := processor.NewBalanceStorageHandler(
@@ -246,23 +277,6 @@ func InitializeData(
 
 		balanceStorage.Initialize(balanceStorageHelper, balanceStorageHandler)
 
-		// Bootstrap balances if provided
-		if len(config.Data.BootstrapBalances) > 0 {
-			_, err := blockStorage.GetHeadBlockIdentifier(ctx)
-			if err == storage.ErrHeadBlockNotFound {
-				err = balanceStorage.BootstrapBalances(
-					ctx,
-					config.Data.BootstrapBalances,
-					genesisBlock,
-				)
-				if err != nil {
-					log.Fatalf("%s: unable to bootstrap balances", err.Error())
-				}
-			} else {
-				log.Println("Skipping balance bootstrapping because already started syncing")
-			}
-		}
-
 		blockWorkers = append(blockWorkers, balanceStorage)
 	}
 
@@ -302,6 +316,7 @@ func InitializeData(
 		signalReceived:           signalReceived,
 		genesisBlock:             genesisBlock,
 		historicalBalanceEnabled: historicalBalanceEnabled,
+		parser:                   parser,
 	}
 }
 
@@ -617,10 +632,127 @@ func (t *DataTester) WatchEndConditions(
 	return nil
 }
 
+// CompleteReconciliations returns the sum of all failed, exempt, and successful
+// reconciliations.
+func (t *DataTester) CompleteReconciliations(ctx context.Context) (int64, error) {
+	activeReconciliations, err := t.counterStorage.Get(ctx, storage.ActiveReconciliationCounter)
+	if err != nil {
+		return -1, fmt.Errorf("%w: cannot get active reconciliations counter", err)
+	}
+
+	exemptReconciliations, err := t.counterStorage.Get(ctx, storage.ExemptReconciliationCounter)
+	if err != nil {
+		return -1, fmt.Errorf("%w: cannot get exempt reconciliations counter", err)
+	}
+
+	failedReconciliations, err := t.counterStorage.Get(ctx, storage.FailedReconciliationCounter)
+	if err != nil {
+		return -1, fmt.Errorf("%w: cannot get failed reconciliations counter", err)
+	}
+
+	skippedReconciliations, err := t.counterStorage.Get(ctx, storage.SkippedReconciliationsCounter)
+	if err != nil {
+		return -1, fmt.Errorf("%w: cannot get skipped reconciliations counter", err)
+	}
+
+	return activeReconciliations.Int64() +
+		exemptReconciliations.Int64() +
+		failedReconciliations.Int64() +
+		skippedReconciliations.Int64(), nil
+}
+
+// WaitForEmptyQueue exits once the active reconciler
+// queue is empty and all reconciler goroutines are idle.
+func (t *DataTester) WaitForEmptyQueue(
+	ctx context.Context,
+) error {
+	// To ensure we don't exit while a reconciliation is ongoing
+	// (i.e. when queue size is 0 but there are busy threads),
+	// we keep track of how many reconciliations we must complete
+	// and only exit when that many reconciliations have been performed.
+	startingComplete, err := t.CompleteReconciliations(ctx)
+	if err != nil {
+		return err
+	}
+	startingRemaining := t.reconciler.QueueSize()
+
+	tc := time.NewTicker(EndAtTipCheckInterval)
+	defer tc.Stop()
+
+	color.Cyan(
+		"[PROGRESS] remaining reconciliations: %d",
+		startingRemaining,
+	)
+
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+
+		case <-tc.C:
+			nowComplete, err := t.CompleteReconciliations(ctx)
+			if err != nil {
+				return err
+			}
+
+			completed := nowComplete - startingComplete
+			remaining := int64(startingRemaining) - completed
+			if remaining <= 0 {
+				t.cancel()
+				return nil
+			}
+
+			color.Cyan(
+				"[PROGRESS] remaining reconciliations: %d",
+				remaining,
+			)
+		}
+	}
+}
+
+// DrainReconcilerQueue returns once the reconciler queue has been drained
+// or an error is encountered.
+func (t *DataTester) DrainReconcilerQueue(ctx context.Context, sigListeners *[]context.CancelFunc) error {
+	color.Cyan("draining reconciler backlog (you can disable this in your configuration file)")
+
+	// To cancel all execution, need to call multiple cancel functions.
+	ctx, cancel := context.WithCancel(ctx)
+	t.cancel = cancel
+	*sigListeners = append(*sigListeners, cancel)
+
+	// Disable inactive lookups
+	t.reconciler.InactiveConcurrency = 0
+
+	g, ctx := errgroup.WithContext(ctx)
+	g.Go(func() error {
+		return t.StartReconciler(ctx)
+	})
+	g.Go(func() error {
+		return t.WaitForEmptyQueue(ctx)
+	})
+
+	err := g.Wait()
+
+	if *t.signalReceived {
+		return errors.New("reconcilier queue drain halted")
+	}
+
+	if errors.Is(err, context.Canceled) {
+		color.Cyan("drained reconciler backlog")
+		return nil
+	}
+
+	return err
+}
+
 // HandleErr is called when `check:data` returns an error.
 // If historical balance lookups are enabled, HandleErr will attempt to
 // automatically find any missing balance-changing operations.
-func (t *DataTester) HandleErr(ctx context.Context, err error, sigListeners *[]context.CancelFunc) error {
+func (t *DataTester) HandleErr(err error, sigListeners *[]context.CancelFunc) error {
+	// Initialize new context because calling context
+	// will no longer be usable when after termination.
+	ctx := context.Background()
+
 	if *t.signalReceived {
 		return results.ExitData(
 			t.config,
@@ -642,7 +774,29 @@ func (t *DataTester) HandleErr(ctx context.Context, err error, sigListeners *[]c
 		)
 	}
 
+	// End condition will only be populated if there is
+	// no error.
 	if len(t.endCondition) != 0 {
+		// Wait for reconciliation queue to drain (only if end condition reached)
+		if shouldReconcile(t.config) &&
+			t.reconciler.QueueSize() > 0 {
+			if t.config.Data.ReconciliationDrainDisabled {
+				color.Cyan("skipping reconciler backlog drain (you can enable this in your configuration file)")
+			} else {
+				drainErr := t.DrainReconcilerQueue(ctx, sigListeners)
+				if drainErr != nil {
+					return results.ExitData(
+						t.config,
+						t.counterStorage,
+						t.balanceStorage,
+						drainErr,
+						"",
+						"",
+					)
+				}
+			}
+		}
+
 		return results.ExitData(
 			t.config,
 			t.counterStorage,
@@ -791,6 +945,7 @@ func (t *DataTester) recursiveOpSearch(
 	r := reconciler.New(
 		reconcilerHelper,
 		reconcilerHandler,
+		t.parser,
 
 		// When using concurrency > 1, we could start looking up balance changes
 		// on multiple blocks at once. This can cause us to return the wrong block
@@ -810,6 +965,7 @@ func (t *DataTester) recursiveOpSearch(
 		t.historicalBalanceEnabled,
 		nil,
 		false,
+		t.parser.BalanceExemptions,
 	)
 
 	balanceStorageHandler := processor.NewBalanceStorageHandler(