Merge pull request #89 from coinbase/luke/enhance-end-condition

Enhanced end condition

cmd/check_data.go

@@ -137,6 +137,10 @@ func runCheckDataCmd(cmd *cobra.Command, args []string) {
 		return dataTester.StartSyncing(ctx, StartIndex, EndIndex)
 	})
 
+	g.Go(func() error {
+		return dataTester.WatchEndConditions(ctx, Config)
+	})
+
 	sigListeners := []context.CancelFunc{cancel}
 	go handleSignals(sigListeners)
 

configuration/configuration.go

@@ -244,6 +244,16 @@ func DefaultConfiguration() *Configuration {
 	}
 }
 
+// EndConditions contains all the conditions for the syncer to stop.
+type EndConditions struct {
+	// EndAtTip determines if syncer should stop once it reached the tip
+	EndAtTip bool `json:"end_at_tip"`
+
+	// EndDuration is an end condition that dictates how long the
+	// check:data command would be running for in seconds
+	EndDuration uint64 `json:"end_duration"`
+}
+
 // DataConfiguration contains all configurations to run check:data.
 type DataConfiguration struct {
 	// ActiveReconciliationConcurrency is the concurrency to use while fetching accounts
@@ -311,6 +321,9 @@ type DataConfiguration struct {
 	// useful to just try to fetch all blocks before checking for balance
 	// consistency.
 	BalanceTrackingDisabled bool `json:"balance_tracking_disabled"`
+
+	// EndCondition contains the conditions for the syncer to stop
+	EndConditions *EndConditions `json:"end_conditions,omitempty"`
 }
 
 // Configuration contains all configuration settings for running

examples/configuration/default.json

@@ -81,6 +81,10 @@
   "interesting_accounts": "",
   "reconciliation_disabled": false,
   "inactive_discrepency_search_disabled": false,
-  "balance_tracking_disabled": false
+  "balance_tracking_disabled": false,
+  "end_conditions": {
+    "end_at_tip": false,
+    "end_duration": 0
+  }
  }
 }

examples/configuration/simple.json

@@ -13,6 +13,10 @@
   "historical_balance_disabled": true,
   "reconciliation_disabled": true,
   "inactive_discrepency_search_disabled": true,
-  "balance_tracking_disabled": true 
+  "balance_tracking_disabled": true,
+  "end_conditions": {
+    "end_at_tip": true,
+    "end_duration": 7200
+  }
  }
 }

go.sum

@@ -54,8 +54,6 @@ 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.3.4-0.20200806182127-a0b262f73fc3 h1:GSHZTufGuZ8nhS1pPQRSezrNu9s5vegqntoVsRX2Q3w=
-github.com/coinbase/rosetta-sdk-go v0.3.4-0.20200806182127-a0b262f73fc3/go.mod h1:Q6dAY0kdG2X3jNaIYnkxnZOb8XEZQar9Q1RcnBgm/wQ=
 github.com/coinbase/rosetta-sdk-go v0.3.4-0.20200807162047-31075a509b1f h1:U69ZwbTR10diY1MDi9LP/RxetVJzjd4bvIDUEs8XYvk=
 github.com/coinbase/rosetta-sdk-go v0.3.4-0.20200807162047-31075a509b1f/go.mod h1:Q6dAY0kdG2X3jNaIYnkxnZOb8XEZQar9Q1RcnBgm/wQ=
 github.com/coreos/bbolt v1.3.2/go.mod h1:iRUV2dpdMOn7Bo10OQBFzIJO9kkE559Wcmn+qkEiiKk=

internal/statefulsyncer/stateful_syncer.go

@@ -17,7 +17,9 @@ package statefulsyncer
 import (
 	"context"
 	"fmt"
+	"log"
 	"math/big"
+	"time"
 
 	"github.com/coinbase/rosetta-cli/internal/logger"
 	"github.com/coinbase/rosetta-cli/internal/storage"
@@ -27,6 +29,14 @@ import (
 	"github.com/coinbase/rosetta-sdk-go/types"
 )
 
+var (
+	// EndAtTipCheckInterval is the frequency that EndAtTip condition
+	// is evaludated
+	//
+	// TODO: make configurable
+	EndAtTipCheckInterval = 10 * time.Second
+)
+
 var _ syncer.Handler = (*StatefulSyncer)(nil)
 var _ syncer.Helper = (*StatefulSyncer)(nil)
 
@@ -179,3 +189,59 @@ func (s *StatefulSyncer) Block(
 ) (*types.Block, error) {
 	return s.fetcher.BlockRetry(ctx, network, block)
 }
+
+// EndAtTipLoop runs a loop that evaluates end condition EndAtTip
+func (s *StatefulSyncer) EndAtTipLoop(
+	ctx context.Context,
+	tipDelay int64,
+) {
+	tc := time.NewTicker(EndAtTipCheckInterval)
+	defer tc.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+
+		case <-tc.C:
+			atTip, err := s.blockStorage.AtTip(ctx, tipDelay)
+			if err != nil {
+				log.Printf(
+					"%s: unable to evaluate if syncer is at tip",
+					err.Error(),
+				)
+				continue
+			}
+
+			if atTip {
+				log.Println("syncer has reached tip")
+				s.cancel()
+				return
+			}
+		}
+	}
+}
+
+// EndDurationLoop runs a loop that evaluates end condition EndDuration
+func (s *StatefulSyncer) EndDurationLoop(
+	ctx context.Context,
+	duration time.Duration,
+) {
+	t := time.NewTimer(duration)
+	defer t.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+
+		case <-t.C:
+			log.Printf(
+				"syncer has reached end condition after %d seconds",
+				int(duration.Seconds()),
+			)
+			s.cancel()
+			return
+		}
+	}
+}

internal/tester/data.go

@@ -293,6 +293,29 @@ func (t *DataTester) StartPeriodicLogger(
 	return ctx.Err()
 }
 
+// WatchEndConditions starts go routines to watch the end conditions
+func (t *DataTester) WatchEndConditions(
+	ctx context.Context,
+	config *configuration.Configuration,
+) error {
+	endConds := config.Data.EndConditions
+	if endConds == nil {
+		return nil
+	}
+
+	if endConds.EndAtTip {
+		// runs a go routine that ends when reaching tip
+		go t.syncer.EndAtTipLoop(ctx, config.TipDelay)
+	}
+
+	if endConds.EndDuration != 0 {
+		// runs a go routine that ends after a duration
+		go t.syncer.EndDurationLoop(ctx, time.Duration(endConds.EndDuration)*time.Second)
+	}
+
+	return nil
+}
+
 // 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.