FOR REVIEW ONLY: indexer 2.14.2 into master

.circleci/config.yml

@@ -46,7 +46,7 @@ jobs:
       go_version:
         type: string
     environment:
-      CI_E2E_FILENAME: "fafa8862/rel-nightly"
+      CI_E2E_FILENAME: "fad05790/rel-nightly"
     steps:
       - go/install:
           version: << parameters.go_version >>

.version

@@ -1 +1 @@
-2.14.1
+2.14.2

cmd/algorand-indexer/daemon.go

@@ -206,7 +206,6 @@ func loadIndexerConfig(indexerDataDir string, configFile string) error {
 		logger.WithError(err).Errorf("invalid config file (%s): %v", viper.ConfigFileUsed(), err)
 		return err
 	}
-	logger.Infof("Using configuration file: %s\n", resolvedConfigPath)
 	return err
 }
 
@@ -266,11 +265,6 @@ func createIndexerPidFile(pidFilePath string) error {
 func runDaemon(daemonConfig *daemonConfig) error {
 	var err error
 	config.BindFlagSet(daemonConfig.flags)
-	err = configureLogger()
-	if err != nil {
-		fmt.Fprintf(os.Stderr, "failed to configure logger: %v", err)
-		return err
-	}
 
 	// Create the data directory if necessary/possible
 	if err = configureIndexerDataDir(daemonConfig.indexerDataDir); err != nil {
@@ -289,6 +283,15 @@ func runDaemon(daemonConfig *daemonConfig) error {
 		return err
 	}
 
+	// Configure the logger after we load all indexer configs
+	err = configureLogger()
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "failed to configure logger: %v", err)
+		return err
+	}
+
+	logger.Infof("Using configuration file: %s", viper.ConfigFileUsed())
+
 	if daemonConfig.pidFilePath != "" {
 		err = createIndexerPidFile(daemonConfig.pidFilePath)
 		if err != nil {

cmd/validator/core/command.go

@@ -22,14 +22,15 @@ func init() {
 		processorNum int
 		printCurl    bool
 		errorLogFile string
+		printSkipped bool
 	)
 
 	ValidatorCmd = &cobra.Command{
 		Use:   "validator",
 		Short: "validator",
 		Long:  "Compare algod and indexer to each other and report any discrepencies.",
 		Run: func(cmd *cobra.Command, _ []string) {
-			run(config, errorLogFile, addr, threads, processorNum, printCurl)
+			run(config, errorLogFile, addr, threads, processorNum, printCurl, printSkipped)
 		},
 	}
 
@@ -46,9 +47,10 @@ func init() {
 	ValidatorCmd.Flags().IntVar(&processorNum, "processor", 0, "Choose compare algorithm [0 = Struct, 1 = Reflection]")
 	ValidatorCmd.Flags().BoolVar(&printCurl, "print-commands", false, "Print curl commands, including tokens, to query algod and indexer.")
 	ValidatorCmd.Flags().StringVarP(&errorLogFile, "error-log-file", "e", "", "When specified, error messages are written to this file instead of to stderr.")
+	ValidatorCmd.Flags().BoolVar(&printSkipped, "print-skipped", false, "Include accounts which were skipped in the error log.")
 }
 
-func run(config Params, errorLogFile, addr string, threads int, processorNum int, printCurl bool) {
+func run(config Params, errorLogFile, addr string, threads int, processorNum int, printCurl, printSkipped bool) {
 	if len(config.AlgodURL) == 0 {
 		ErrorLog.Fatalf("algod-url parameter is required.")
 	}
@@ -91,7 +93,7 @@ func run(config Params, errorLogFile, addr string, threads int, processorNum int
 	}()
 
 	// This will keep going until the results channel is closed.
-	numErrors := resultsPrinter(config, printCurl, results)
+	numErrors := resultsPrinter(config, printCurl, printSkipped, results)
 	if numErrors > 0 {
 		os.Exit(1)
 	}
@@ -134,7 +136,7 @@ func resultChar(success bool, retries int, skipReason Skip) string {
 }
 
 // resultsPrinter reads the results channel and prints it to the error log. Returns the number of errors.
-func resultsPrinter(config Params, printCurl bool, results <-chan Result) int {
+func resultsPrinter(config Params, printCurl, printSkipped bool, results <-chan Result) int {
 	numResults := 0
 	numErrors := 0
 	skipCounts := make(map[Skip]uint64)
@@ -176,6 +178,9 @@ func resultsPrinter(config Params, printCurl bool, results <-chan Result) int {
 		if r.Error != nil || !r.Equal {
 			if r.SkipReason != NotSkipped {
 				skipCounts[r.SkipReason]++
+				if !printSkipped {
+					continue
+				}
 			} else {
 				numErrors++
 			}

cmd/validator/core/validator.go

@@ -122,7 +122,7 @@ func Start(work <-chan string, processorID ProcessorID, threads int, config Para
 func CallProcessor(processor Processor, addrInput string, config Params, results chan<- Result) {
 	addr, err := normalizeAddress(addrInput)
 	if err != nil {
-		results <- resultError(err, addrInput)
+		results <- resultError(err, addr)
 		return
 	}
 
@@ -133,23 +133,28 @@ func CallProcessor(processor Processor, addrInput string, config Params, results
 	// catches up with the first algod account query.
 	algodData, err := getData(algodDataURL, config.AlgodToken)
 	if err != nil {
-		err = fmt.Errorf("error getting algod data (%d): %w", algodData, err)
-		results <- resultError(err, addrInput)
+		err = fmt.Errorf("error getting algod data (%s): %w", algodData, err)
+		switch {
+		case strings.Contains(string(algodData), api.ErrResultLimitReached):
+			results <- resultSkip(err, addr, SkipLimitReached)
+		default:
+			results <- resultError(err, addr)
+		}
 		return
 	}
 
 	// Retry loop.
 	for i := 0; true; i++ {
 		indexerData, err := getData(indexerDataURL, config.IndexerToken)
 		if err != nil {
-			err = fmt.Errorf("error getting indexer data (%d): %w", indexerData, err)
+			err = fmt.Errorf("error getting indexer data (%s): %w", indexerData, err)
 			switch {
 			case strings.Contains(string(indexerData), api.ErrResultLimitReached):
-				results <- resultSkip(err, addrInput, SkipLimitReached)
+				results <- resultSkip(err, addr, SkipLimitReached)
 			case strings.Contains(string(indexerData), api.ErrNoAccountsFound):
-				results <- resultSkip(err, addrInput, SkipAccountNotFound)
+				results <- resultSkip(err, addr, SkipAccountNotFound)
 			default:
-				results <- resultError(err, addrInput)
+				results <- resultError(err, addr)
 			}
 			return
 		}

cmd/validator/core/validator_test.go

@@ -0,0 +1,146 @@
+package core
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/jarcoal/httpmock"
+	"github.com/stretchr/testify/require"
+
+	"github.com/algorand/go-algorand-sdk/types"
+
+	"github.com/algorand/indexer/api"
+)
+
+type mockProcessor struct {
+	result Result
+	err    error
+}
+
+func (m *mockProcessor) ProcessAddress(_ []byte, _ []byte) (Result, error) {
+	return m.result, m.err
+}
+
+func TestCallProcessor(t *testing.T) {
+	mockParams := Params{
+		AlgodURL:   "http://localhost/algod",
+		IndexerURL: "http://localhost/indexer",
+	}
+	var testAddr types.Address
+	testAddr[0] = 1
+
+	type args struct {
+		processor   Processor
+		addrInput   string
+		config      Params
+		algodData   string
+		algodCode   int
+		indexerData string
+		indexerCode int
+		errorStr    string
+	}
+	tests := []struct {
+		name string
+		args args
+	}{
+		{
+			name: "Bad address",
+			args: args{addrInput: "399 Boylston Street", errorStr: "unable to decode address"},
+		}, {
+			name: "algod: acct limit",
+			args: args{
+				algodCode: 400,
+				algodData: api.ErrResultLimitReached,
+				errorStr:  fmt.Sprintf("error getting algod data (%s)", api.ErrResultLimitReached),
+			},
+		}, {
+			name: "algod: unknown error",
+			args: args{
+				algodCode: 500,
+				algodData: "server sent GOAWAY and closed the connection",
+				errorStr:  "error getting algod data (server sent GOAWAY and closed the connection): bad status: 500",
+			},
+		}, {
+			name: "indexer: acct limit",
+			args: args{
+				indexerCode: 400,
+				indexerData: api.ErrResultLimitReached,
+				errorStr:    fmt.Sprintf("error getting indexer data (%s)", api.ErrResultLimitReached),
+			},
+		}, {
+			name: "indexer: acct not found",
+			args: args{
+				indexerCode: 404,
+				indexerData: api.ErrNoAccountsFound,
+				errorStr:    fmt.Sprintf("error getting indexer data (%s)", api.ErrNoAccountsFound),
+			},
+		}, {
+			name: "indexer: unknown",
+			args: args{
+				indexerCode: 500,
+				indexerData: "server sent GOAWAY and closed the connection",
+				errorStr:    "error getting indexer data (server sent GOAWAY and closed the connection): bad status: 500",
+			},
+		}, {
+			name: "processor: error",
+			args: args{
+				processor: &mockProcessor{
+					err: fmt.Errorf("ask again tomorrow"),
+				},
+				addrInput: testAddr.String(),
+				errorStr:  fmt.Sprintf("error processing account %s: ask again tomorrow", testAddr.String()),
+			},
+		},
+	}
+	setDefaults := func(a args) args {
+		if a.config == (Params{}) {
+			a.config = mockParams
+		}
+		if a.algodCode == 0 {
+			a.algodCode = 200
+			if a.algodData == "" {
+				a.algodData = "{}"
+			}
+		}
+		if a.indexerCode == 0 {
+			a.indexerCode = 200
+			if a.indexerData == "" {
+				a.indexerData = "{}"
+			}
+		}
+		return a
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tt.args = setDefaults(tt.args)
+			// Given: mock http responses
+			if tt.args.algodData != "" || tt.args.indexerData != "" {
+				httpmock.Activate()
+				defer httpmock.DeactivateAndReset()
+			}
+			if tt.args.algodData != "" {
+				httpmock.RegisterResponder("GET", "=~"+mockParams.AlgodURL+".*",
+					httpmock.NewStringResponder(tt.args.algodCode, tt.args.algodData))
+			}
+			if tt.args.indexerData != "" {
+				httpmock.RegisterResponder("GET", "=~"+mockParams.IndexerURL+".*",
+					httpmock.NewStringResponder(tt.args.indexerCode, tt.args.indexerData))
+			}
+
+			// When: call CallProcessor
+			// We only use CallProcessor once, so a buffered channel of 1 should be sufficient.
+			results := make(chan Result, 1)
+			CallProcessor(tt.args.processor, tt.args.addrInput, tt.args.config, results)
+			close(results)
+
+			// Then: we get the expected result.
+			for result := range results {
+				if tt.args.errorStr != "" {
+					require.ErrorContains(t, result.Error, tt.args.errorStr)
+				} else {
+					require.NoError(t, result.Error)
+				}
+			}
+		})
+	}
+}

go.mod

@@ -8,6 +8,7 @@ require (
 	github.com/algorand/go-algorand v0.0.0-20220211161928-53b157beb10f
 	github.com/algorand/go-algorand-sdk v1.9.1
 	github.com/algorand/go-codec/codec v1.1.8
+	github.com/algorand/go-deadlock v0.2.2
 	github.com/algorand/oapi-codegen v1.3.7
 	github.com/davecgh/go-spew v1.1.1
 	github.com/getkin/kin-openapi v0.22.0
@@ -29,10 +30,10 @@ require (
 )
 
 require (
+	github.com/DataDog/zstd v1.5.2 // indirect
 	github.com/Microsoft/go-winio v0.4.14 // indirect
 	github.com/algorand/avm-abi v0.1.0 // indirect
 	github.com/algorand/falcon v0.0.0-20220727072124-02a2a64c4414 // indirect
-	github.com/algorand/go-deadlock v0.2.2 // indirect
 	github.com/algorand/go-sumhash v0.1.0 // indirect
 	github.com/algorand/msgp v1.1.52 // indirect
 	github.com/algorand/websocket v1.4.5 // indirect

go.sum

@@ -61,6 +61,8 @@ github.com/Azure/go-autorest/tracing v0.5.0/go.mod h1:r/s2XiOKccPW3HrqB+W0TQzfbt
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
 github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo=
 github.com/DataDog/datadog-go v3.2.0+incompatible/go.mod h1:LButxg5PwREeZtORoXG3tL4fMGNddJ+vMq1mwgfaqoQ=
+github.com/DataDog/zstd v1.5.2 h1:vUG4lAyuPCXO0TLbXvPv7EB7cNK1QV/luu55UHLrrn8=
+github.com/DataDog/zstd v1.5.2/go.mod h1:g4AWEaM3yOg3HYfnJ3YIawPnVdXJh9QME85blwSAmyw=
 github.com/HdrHistogram/hdrhistogram-go v1.1.0/go.mod h1:yDgFjdqOqDEKOvasDdhWNXYg9BVp4O+o5f6V/ehm6Oo=
 github.com/Knetic/govaluate v3.0.1-0.20171022003610-9aa49832a739+incompatible/go.mod h1:r7JcOSlj0wfOMncg0iLm8Leh48TZaKVeNIfJntJ2wa0=
 github.com/Masterminds/semver/v3 v3.1.1/go.mod h1:VPu/7SZ7ePZ3QOrcuXROw5FAcLl4a0cBrbBpGY/8hQs=

misc/e2elive.py

@@ -83,12 +83,17 @@ def main():
         from botocore.config import Config
         from botocore import UNSIGNED
         s3 = boto3.client("s3", config=Config(signature_version=UNSIGNED))
+        prefix = "indexer/e2e4"
         if "/" in tarname:
-            tarname = tarname.split("/")[1]
+        # parse tarname like fafa8862/rel-nightly
+            cmhash_tarnme = tarname.split("/")
+            cmhash = cmhash_tarnme[0]
+            tarname =cmhash_tarnme[1]
+            prefix+="/"+cmhash
             tarpath = os.path.join(tempdir, tarname)
         else:
             tarpath = os.path.join(tempdir, tarname)
-        prefix = "indexer/e2e4"
+
         success = firstFromS3Prefix(s3, bucket, prefix, tarname, outpath=tarpath)
         if not success:
             raise Exception(
@@ -203,7 +208,7 @@ def countblocks(path):
     return row[0]
 
 
-def tryhealthurl(healthurl, verbose=False, waitforround=100):
+def tryhealthurl(healthurl, verbose=False, waitforround=200):
     try:
         response = urllib.request.urlopen(healthurl)
         if response.code != 200:

processor/blockprocessor/internal/catchupservice.go

@@ -10,6 +10,7 @@ import (
 	"github.com/algorand/go-algorand/catchup"
 	"github.com/algorand/go-algorand/config"
 	"github.com/algorand/go-algorand/data/bookkeeping"
+	"github.com/algorand/go-algorand/ledger"
 	"github.com/algorand/go-algorand/ledger/ledgercore"
 	"github.com/algorand/go-algorand/logging"
 	"github.com/algorand/go-algorand/network"
@@ -97,7 +98,7 @@ func CatchupServiceCatchup(ctx context.Context, logger *log.Logger, catchpoint,
 		node,
 		wrappedLogger,
 		net,
-		l,
+		ledger.MakeCatchpointCatchupAccessor(l, wrappedLogger),
 		cfg,
 	)
 	if err != nil {