WIP: Batching

Makefile

@@ -1,6 +1,4 @@
 SRCPATH		:= $(shell pwd)
-OS_TYPE		?= $(shell $(SRCPATH)/mule/scripts/ostype.sh)
-ARCH		?= $(shell $(SRCPATH)/mule/scripts/archtype.sh)
 ifeq ($(OS_TYPE), darwin)
 ifeq ($(ARCH), arm64)
 export CPATH=/opt/homebrew/include
@@ -56,19 +54,10 @@ e2e: cmd/algorand-indexer/algorand-indexer
 e2e-filter-test: cmd/algorand-indexer/algorand-indexer
 	cd e2e_tests/docker/indexer-filtered/ && docker-compose build --build-arg GO_IMAGE=${GO_IMAGE} && docker-compose up --exit-code-from e2e-read
 
-deploy:
-	mule/deploy.sh
-
-sign:
-	mule/sign.sh
-
-test-package:
-	mule/e2e.sh
-
 test-generate:
 	test/test_generate.py
 
 indexer-v-algod:
 	pytest -sv misc/parity
 
-.PHONY: all test e2e integration fmt lint deploy sign test-package package fakepackage cmd/algorand-indexer/algorand-indexer idb/mocks/IndexerDb.go indexer-v-algod
+.PHONY: all test e2e integration fmt lint package fakepackage cmd/algorand-indexer/algorand-indexer idb/mocks/IndexerDb.go indexer-v-algod

cmd/algorand-indexer/daemon.go

@@ -30,7 +30,7 @@ type daemonConfig struct {
 	tokenString               string
 	writeTimeout              time.Duration
 	readTimeout               time.Duration
-	maxConn                   uint32
+	maxConns                  int32
 	maxAPIResourcesPerAccount uint32
 	maxTransactionsLimit      uint32
 	defaultTransactionsLimit  uint32
@@ -74,7 +74,7 @@ func DaemonCmd() *cobra.Command {
 	cfg.flags.StringVarP(&cfg.metricsMode, "metrics-mode", "", "OFF", "configure the /metrics endpoint to [ON, OFF, VERBOSE]")
 	cfg.flags.DurationVarP(&cfg.writeTimeout, "write-timeout", "", 30*time.Second, "set the maximum duration to wait before timing out writes to a http response, breaking connection")
 	cfg.flags.DurationVarP(&cfg.readTimeout, "read-timeout", "", 5*time.Second, "set the maximum duration for reading the entire request")
-	cfg.flags.Uint32VarP(&cfg.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")
+	cfg.flags.Int32VarP(&cfg.maxConns, "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")
 
 	cfg.flags.StringVar(&cfg.suppliedAPIConfigFile, "api-config-file", "", "supply an API config file to enable/disable parameters")
 	cfg.flags.BoolVar(&cfg.enableAllParameters, "enable-all-parameters", false, "override default configuration and enable all parameters. Can't be used with --api-config-file")
@@ -279,7 +279,7 @@ func runDaemon(daemonConfig *daemonConfig) error {
 	opts := idb.IndexerDbOptions{}
 	opts.ReadOnly = true
 
-	opts.MaxConn = daemonConfig.maxConn
+	opts.MaxConns = daemonConfig.maxConns
 	opts.IndexerDatadir = daemonConfig.indexerDataDir
 
 	db, _, err := indexerDbFromFlags(opts)

idb/idb.go

@@ -411,7 +411,7 @@ type IndexerDbOptions struct {
 	// 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
+	MaxConns int32
 
 	IndexerDatadir string
 	AlgodDataDir   string

idb/postgres/indexerDb.go

@@ -0,0 +1,70 @@
+package postgres
+
+import (
+	"fmt"
+	"sync"
+
+	"github.com/jackc/pgx/v4"
+	"github.com/jackc/pgx/v4/pgxpool"
+	log "github.com/sirupsen/logrus"
+
+	"github.com/algorand/indexer/v3/idb/migration"
+)
+
+// const useExperimentalTxnInsertion = false
+// const useExperimentalWithIntraBugfix = true
+
+var serializable = pgx.TxOptions{IsoLevel: pgx.Serializable} // be a real ACID database
+var readonlyRepeatableRead = pgx.TxOptions{IsoLevel: pgx.RepeatableRead, AccessMode: pgx.ReadOnly}
+
+// in actuality, for postgres the following is no weaker than ReadCommitted:
+// https://www.postgresql.org/docs/current/transaction-iso.html
+// TODO: change this to pgs.ReadCommitted
+// var uncommitted = pgx.TxOptions{IsoLevel: pgx.ReadUncommitted}
+
+
+
+// var experimentalCommitLevel = uncommitted // serializable // uncommitted
+
+// IndexerDb is an idb.IndexerDB implementation
+type IndexerDb struct {
+	readonly bool
+	log      *log.Logger
+
+	db             *pgxpool.Pool
+	migration      *migration.Migration
+	accountingLock sync.Mutex
+
+	TuningParams
+}
+
+// TuningParams are database interaction settings that can be
+// fine tuned to improve performance based on a specific hardware deployment
+// and workload characteristics.
+type TuningParams struct {
+	PgxOpts   pgx.TxOptions
+	BatchSize uint32
+}
+
+// defaultTuningParams returns a TuningParams object with default values.
+func defaultTuningParams() TuningParams{
+	return TuningParams{
+		PgxOpts:   serializable,
+		BatchSize: 2500,
+	}
+}
+
+func shortName(isoLevel pgx.TxIsoLevel) string {
+	switch isoLevel {
+	case pgx.Serializable:
+		return "S"
+	case pgx.RepeatableRead:
+		return "RR"
+	case pgx.ReadCommitted:
+		return "RC"
+	case pgx.ReadUncommitted:
+		return "RU"
+	default:
+		return fmt.Sprintf("unknown_%s", isoLevel)
+	}
+}

idb/postgres/internal/testing/testing.go

@@ -17,6 +17,13 @@
 
 var testpg string
 
+var knownPgImages = map[string]struct{}{
+	"13-alpine": {},
+	"14":        {},
+	"15":        {},
+	"16beta3":   {},
+}
+
 func init() {
 	flag.StringVar(&testpg, "test-pg", "", "postgres connection string; resets the database")
 	if testpg == "" {
@@ -28,7 +35,7 @@
 
 // SetupPostgres starts a gnomock postgres DB then returns the database object,
 // the connection string and a shutdown function.
-func SetupPostgres(t *testing.T) (*pgxpool.Pool, string, func()) {
+func SetupPostgres(t testing.TB) (*pgxpool.Pool, string, func()) {
 	if testpg != "" {
 		// use non-docker Postgresql
 		connStr := testpg
@@ -47,8 +54,16 @@
 		return db, connStr, shutdownFunc
 	}
 
+	return SetupGnomockPgWithVersion(t, "13-alpine")
+}
+
+func SetupGnomockPgWithVersion(t testing.TB, pgImage string) (*pgxpool.Pool, string, func()) {
+	if _, ok := knownPgImages[pgImage]; !ok {
+		t.Fatalf("SetupGnomockPgWithVersion(): unrecognized postgres Docker image for gnomock preset: %s", pgImage)
+	}
+
 	p := postgres.Preset(
-		postgres.WithVersion("13-alpine"),
+		postgres.WithVersion(pgImage),
 		postgres.WithUser("gnomock", "gnomick"),
 		postgres.WithDatabase("mydb"),
 	)
@@ -61,6 +76,16 @@
 		"gnomock", "gnomick", "mydb",
 	)
 
+	// config, err := pgxpool.ParseConfig(connStr)
+	// require.NoError(t, err, "Error parsing connection string: %s", connStr)
+
+	// if maxConns != nil {
+	// 	config.MaxConns = *maxConns
+	// }
+
+	// db, err := pgxpool.ConnectConfig(context.Background(), config)
+	// require.NoError(t, err, "Error creating connection pool via config: %#v", config)
+
 	db, err := pgxpool.Connect(context.Background(), connStr)
 	require.NoError(t, err, "Error opening postgres connection")