Merge branch 'develop' of https://github.com/algorand/indexer into tz…

…affi/min-balance

.circleci/config.yml

@@ -16,7 +16,7 @@ workflows:
           name: test_with_go_<< matrix.go_version >>
           matrix: &go-version-matrix
             parameters:
-              go_version: ["1.16.11", "1.17.4"]
+              go_version: ["1.14.7", "1.15.15"]
 
 jobs:
   test:

.version

@@ -1 +1 @@
-2.8.0
+2.8.1

Makefile

@@ -49,7 +49,7 @@ test: idb/mocks/IndexerDb.go cmd/algorand-indexer/algorand-indexer
 
 lint: go-algorand
 	golint -set_exit_status ./...
-	go vet -mod=mod ./...
+	go vet ./...
 
 fmt:
 	go fmt ./...

README.md

@@ -7,7 +7,7 @@ The Indexer is a standalone service that reads committed blocks from the Algoran
 
 # Tested Requirements Versions
 
-* [go 1.16](https://golang.org/dl/)
+* [go 1.13](https://golang.org/dl/)
 * [Postgres 13](https://www.postgresql.org/download/)
 
 # Quickstart
@@ -117,12 +117,19 @@ When `--token your-token` is provided, an authentication header is required. For
 The `/metrics` endpoint is configured with the `--metrics-mode` option and configures if and how [Prometheus](https://prometheus.io/) formatted metrics are generated.
 
 There are different settings:
+
 | Setting | Description |
 | ------- | ----------- |
 | ON      | Metrics for each REST endpoint in addition to application metrics. |
 | OFF     | No metrics endpoint. |
 | VERBOSE | Separate metrics for each combination of query parameters. This option should be used with caution, there are many combinations of query parameters which could cause extra memory load depending on usage patterns. |
 
+## Connection Pool Settings
+
+One can set the maximum number of connections allowed in the local connection pool by using the `--max-conn` setting.  It is recommended to set this number to be below the database server connection pool limit.
+
+If the maximum number of connections/active queries is reached, subsequent connections will wait until a connection becomes available, or timeout according to the read-timeout setting.
+
 # Settings
 
 Settings can be provided from the command line, a configuration file, or an environment variable
@@ -140,6 +147,7 @@ Settings can be provided from the command line, a configuration file, or an envi
 | token                    | t       | api-token                  | INDEXER_API_TOKEN                  |
 | dev-mode                 |         | dev-mode                   | INDEXER_DEV_MODE                   |
 | metrics-mode             |         | metrics-mode               | INDEXER_METRICS_MODE               |
+| max-conn                 |         | max-conn                   | INDEXER_MAX_CONN                   |
 
 ## Command line
 

api/handlers.go

@@ -802,10 +802,16 @@ func (si *ServerImplementation) fetchBlock(ctx context.Context, round uint64) (g
 
 		results := make([]generated.Transaction, 0)
 		for _, txrow := range transactions {
+			// Do not include inner transactions.
+			if txrow.RootTxn != nil {
+				continue
+			}
+
 			tx, err := txnRowToTransaction(txrow)
 			if err != nil {
 				return err
 			}
+
 			results = append(results, tx)
 		}
 

api/handlers_e2e_test.go

@@ -305,6 +305,36 @@ func TestPagingRootTxnDeduplication(t *testing.T) {
 			}
 		})
 	}
+
+	// Test block endpoint deduplication
+	t.Run("Deduplicate Transactions In Block", func(t *testing.T) {
+		//////////
+		// When // we fetch the block
+		//////////
+		e := echo.New()
+		req := httptest.NewRequest(http.MethodGet, "/", nil)
+		rec := httptest.NewRecorder()
+		c := e.NewContext(req, rec)
+		c.SetPath("/v2/blocks/")
+
+		// Get first page with limit 1.
+		// Address filter causes results to return newest to oldest.
+		api := &ServerImplementation{db: db, timeout: 30 * time.Second}
+		err = api.LookupBlock(c, uint64(block.Round()))
+		require.NoError(t, err)
+
+		//////////
+		// Then // There should be a single transaction which has inner transactions
+		//////////
+		var response generated.BlockResponse
+		require.Equal(t, http.StatusOK, rec.Code)
+		json.Decode(rec.Body.Bytes(), &response)
+
+		require.NotNil(t, response.Transactions)
+		require.Len(t, *response.Transactions, 1)
+		require.NotNil(t, (*response.Transactions)[0])
+		require.Len(t, *(*response.Transactions)[0].InnerTxns, 2)
+	})
 }
 
 func TestVersion(t *testing.T) {

cmd/algorand-indexer/daemon.go

@@ -34,6 +34,7 @@ var (
 	tokenString      string
 	writeTimeout     time.Duration
 	readTimeout      time.Duration
+	maxConn          uint32
 )
 
 var daemonCmd = &cobra.Command{
@@ -83,6 +84,9 @@ var daemonCmd = &cobra.Command{
 		if noAlgod && !allowMigration {
 			opts.ReadOnly = true
 		}
+
+		opts.MaxConn = maxConn
+
 		db, availableCh := indexerDbFromFlags(opts)
 		defer db.Close()
 		var wg sync.WaitGroup
@@ -141,6 +145,7 @@ func init() {
 	daemonCmd.Flags().StringVarP(&metricsMode, "metrics-mode", "", "OFF", "configure the /metrics endpoint to [ON, OFF, VERBOSE]")
 	daemonCmd.Flags().DurationVarP(&writeTimeout, "write-timeout", "", 30*time.Second, "set the maximum duration to wait before timing out writes to a http response, breaking connection")
 	daemonCmd.Flags().DurationVarP(&readTimeout, "read-timeout", "", 5*time.Second, "set the maximum duration for reading the entire request")
+	daemonCmd.Flags().Uint32VarP(&maxConn, "max-conn", "", 0, "set the maximum connections allowed in the connection pool, if the maximum is reached subsequent connections will wait until a connection becomes available, or timeout according to the read-timeout setting")
 
 	viper.RegisterAlias("algod", "algod-data-dir")
 	viper.RegisterAlias("algod-net", "algod-address")

cmd/block-generator/docker/Dockerfile-generator

@@ -1,4 +1,4 @@
-FROM golang:1.16.11-alpine
+FROM golang:1.13.12-alpine
 
 # Build and run the block generator
 

cmd/block-generator/docker/Dockerfile-indexer

@@ -1,4 +1,4 @@
-FROM golang:1.16.11-alpine
+FROM golang:1.13.12-alpine
 
 # Build and run indexer
 RUN mkdir /work

go.mod

@@ -1,6 +1,6 @@
 module github.com/algorand/indexer
 
-go 1.16
+go 1.13
 
 replace github.com/algorand/go-algorand => ./third_party/go-algorand
 
@@ -26,5 +26,5 @@ require (
 	github.com/stretchr/testify v1.7.0
 	github.com/vektra/mockery v1.1.2 // indirect
 	golang.org/x/lint v0.0.0-20210508222113-6edffad5e616 // indirect
-	golang.org/x/tools v0.1.8 // indirect
+	golang.org/x/tools v0.1.9 // indirect
 )

go.sum

@@ -1007,6 +1007,8 @@ golang.org/x/tools v0.0.0-20200804011535-6c149bb5ef0d/go.mod h1:njjCfa9FT2d7l9Bc
 golang.org/x/tools v0.0.0-20200825202427-b303f430e36d/go.mod h1:njjCfa9FT2d7l9Bc6FUM5FLjQPp3cFF28FI3qnDFljA=
 golang.org/x/tools v0.1.8 h1:P1HhGGuLW4aAclzjtmJdf0mJOjVUZUzOTqkAkWL+l6w=
 golang.org/x/tools v0.1.8/go.mod h1:nABZi5QlRsZVlzPpHl034qft6wpY4eDcsTt5AaioBiU=
+golang.org/x/tools v0.1.9 h1:j9KsMiaP1c3B0OTQGth0/k+miLGTgLsAFUCrF2vLcF8=
+golang.org/x/tools v0.1.9/go.mod h1:nABZi5QlRsZVlzPpHl034qft6wpY4eDcsTt5AaioBiU=
 golang.org/x/xerrors v0.0.0-20190410155217-1f06c39b4373/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20190513163551-3ee3066db522/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=

idb/idb.go

@@ -336,6 +336,10 @@ type ApplicationRow struct {
 // IndexerDbOptions are the options common to all indexer backends.
 type IndexerDbOptions struct {
 	ReadOnly bool
+	// Maximum connection number for connection pool
+	// This means the total number of active queries that can be running
+	// concurrently can never be more than this
+	MaxConn uint32
 }
 
 // Health is the response object that IndexerDb objects need to return from the Health method.

idb/postgres/postgres.go

@@ -50,7 +50,17 @@ var readonlyRepeatableRead = pgx.TxOptions{IsoLevel: pgx.RepeatableRead, AccessM
 // Returns an error object and a channel that gets closed when blocking migrations
 // finish running successfully.
 func OpenPostgres(connection string, opts idb.IndexerDbOptions, log *log.Logger) (*IndexerDb, chan struct{}, error) {
-	db, err := pgxpool.Connect(context.Background(), connection)
+
+	postgresConfig, err := pgxpool.ParseConfig(connection)
+	if err != nil {
+		return nil, nil, fmt.Errorf("Couldn't parse config: %v", err)
+	}
+
+	if opts.MaxConn != 0 {
+		postgresConfig.MaxConns = int32(opts.MaxConn)
+	}
+
+	db, err := pgxpool.ConnectConfig(context.Background(), postgresConfig)
 
 	if err != nil {
 		return nil, nil, fmt.Errorf("connecting to postgres: %v", err)

idb/postgres/postgres_integration_test.go

@@ -90,6 +90,43 @@ func TestAccountedRoundNextRound0(t *testing.T) {
 	assert.Equal(t, uint64(0), round)
 }
 
+// TestMaxConnection tests that when setting the maximum connection to a value, that it is
+// accurately set and that acquiring connections accurately depletes the pool
+func TestMaxConnection(t *testing.T) {
+	_, connStr, shutdownFunc := pgtest.SetupPostgres(t)
+	defer shutdownFunc()
+
+	// Open Postgres with a maximum of 2 connections locally
+	pdb, _, err := OpenPostgres(connStr, idb.IndexerDbOptions{MaxConn: 2}, nil)
+	assert.NoError(t, err)
+	defer pdb.Close()
+
+	s := pdb.db.Stat()
+
+	assert.Equal(t, 2, int(s.MaxConns()))
+	assert.Equal(t, 0, int(s.AcquiredConns()))
+	assert.GreaterOrEqual(t, 1, int(s.IdleConns()))
+
+	conn1, err := pdb.db.Acquire(context.Background())
+	assert.NoError(t, err)
+	defer conn1.Release()
+
+	s = pdb.db.Stat()
+	assert.Equal(t, 1, int(s.AcquiredConns()))
+
+	conn2, err := pdb.db.Acquire(context.Background())
+	assert.NoError(t, err)
+	defer conn2.Release()
+
+	s = pdb.db.Stat()
+
+	// We have reached the total of 2 "total" max connections
+	// Meaning we should have two acquired connections and 0 idle connections
+	assert.Equal(t, 2, int(s.AcquiredConns()))
+	assert.Equal(t, 0, int(s.IdleConns()))
+
+}
+
 func assertAccountAsset(t *testing.T, db *pgxpool.Pool, addr basics.Address, assetid uint64, frozen bool, amount uint64) {
 	var row pgx.Row
 	var f bool

misc/Dockerfile

@@ -1,4 +1,4 @@
-ARG GO_IMAGE=golang:1.16.11
+ARG GO_IMAGE=golang:1.14.7
 FROM $GO_IMAGE
 
 RUN echo "Go image: $GO_IMAGE"

test/test_generate.py

@@ -7,7 +7,8 @@
 # Validates that certain files are not modified after running "go generate"
 
 # List of directories that contain generate.go files
-list_of_directories=["../idb/postgres/internal/schema"]
+# Run from project root
+list_of_directories=["idb/postgres/internal/schema"]
 
 for dir in list_of_directories:
     os.system("/usr/local/go/bin/go generate {}".format(dir))
@@ -20,4 +21,4 @@
     print("Output of 'git status --porcelain':\n {}".format(out))
     out = check_output(["git diff"], shell=True).strip().decode('UTF-8')
     print("Output of diff: {}\n".format(out))
-    sys.exit(1)
+    sys.exit(1)