feat: add offset limit to logs pagination and db metrics

.github/workflows/test-e2e.yaml

@@ -24,7 +24,7 @@ jobs:
           GITHUB_REF_NAME: ${{ github.ref_name }}
         run: |
           # Default fallback commit
-          DEFAULT_COMMIT="a51bdc9833d1a4641c2d7969414dbe23a4fa345e"
+          DEFAULT_COMMIT="c87b8c55ace0d48112a65e8a0659d3af4ef2f1cf"
           REF="$DEFAULT_COMMIT"
           if [ "$GITHUB_EVENT_NAME" = "pull_request" ]; then
             # For pull_request events, we only look at the PR's base branch

api/events/events.go

@@ -22,15 +22,17 @@ import (
 type Events struct {
 	repo             *chain.Repository
 	db               *logdb.LogDB
-	limit            uint64
+	maxLimit         uint64
+	maxOffset        uint64
 	maxCriteriaCount int
 }
 
-func New(repo *chain.Repository, db *logdb.LogDB, logsLimit uint64, maxCriteriaCount int) *Events {
+func New(repo *chain.Repository, db *logdb.LogDB, maxLimit uint64, maxOffset uint64, maxCriteriaCount int) *Events {
 	return &Events{
 		repo,
 		db,
-		logsLimit,
+		maxLimit,
+		maxOffset,
 		maxCriteriaCount,
 	}
 }
@@ -58,7 +60,7 @@ func (e *Events) handleFilter(w http.ResponseWriter, req *http.Request) error {
 	if err := restutil.ParseJSON(req.Body, &filter); err != nil {
 		return restutil.BadRequest(errors.WithMessage(err, "body"))
 	}
-	if err := filter.Options.Validate(e.limit); err != nil {
+	if err := filter.Options.Validate(e.maxLimit, e.maxOffset); err != nil {
 		return restutil.Forbidden(err)
 	}
 	if err := filter.Range.Validate(); err != nil {
@@ -83,7 +85,7 @@ func (e *Events) handleFilter(w http.ResponseWriter, req *http.Request) error {
 	if filter.Options.Limit == nil {
 		// if filter.Options.Limit is nil, set to the default limit +1
 		// to detect whether there are more logs than the default limit
-		limit := e.limit + 1
+		limit := e.maxLimit + 1
 		filter.Options.Limit = &limit
 	}
 
@@ -93,8 +95,8 @@ func (e *Events) handleFilter(w http.ResponseWriter, req *http.Request) error {
 	}
 
 	// ensure the result size is less than the configured limit
-	if len(fes) > int(e.limit) {
-		return restutil.Forbidden(fmt.Errorf("the number of filtered logs exceeds the maximum allowed value of %d, please use pagination", e.limit))
+	if len(fes) > int(e.maxLimit) {
+		return restutil.Forbidden(fmt.Errorf("the number of filtered logs exceeds the maximum allowed value of %d, please use pagination", e.maxLimit))
 	}
 
 	return restutil.WriteJSON(w, fes)

api/events/events_test.go

@@ -28,15 +28,18 @@ import (
 	"github.com/vechain/thor/v2/tx"
 )
 
-const defaultLogLimit uint64 = 1000
+const (
+	defaultLogLimit  uint64 = 1000
+	defaultLogOffset uint64 = 100
+)
 
 var (
 	ts      *httptest.Server
 	tclient *thorclient.Client
 )
 
 func TestEmptyEvents(t *testing.T) {
-	initEventServer(t, defaultLogLimit)
+	initEventServer(t, defaultLogLimit, defaultLogOffset)
 	defer ts.Close()
 
 	tclient = thorclient.New(ts.URL)
@@ -49,7 +52,7 @@ func TestEmptyEvents(t *testing.T) {
 }
 
 func TestEvents(t *testing.T) {
-	thorChain := initEventServer(t, defaultLogLimit)
+	thorChain := initEventServer(t, defaultLogLimit, defaultLogOffset)
 	defer ts.Close()
 
 	blocksToInsert := 5
@@ -59,7 +62,7 @@ func TestEvents(t *testing.T) {
 }
 
 func TestOptionalIndexes(t *testing.T) {
-	thorChain := initEventServer(t, defaultLogLimit)
+	thorChain := initEventServer(t, defaultLogLimit, defaultLogOffset)
 	defer ts.Close()
 	insertBlocks(t, thorChain, 5)
 	tclient = thorclient.New(ts.URL)
@@ -109,7 +112,7 @@ func TestOptionalIndexes(t *testing.T) {
 }
 
 func TestEvents_WithOptionsNoLimit(t *testing.T) {
-	thorChain := initEventServer(t, defaultLogLimit)
+	thorChain := initEventServer(t, defaultLogLimit, defaultLogOffset)
 	defer ts.Close()
 	insertBlocks(t, thorChain, 5)
 
@@ -133,7 +136,7 @@ func TestEvents_WithOptionsNoLimit(t *testing.T) {
 }
 
 func TestOption(t *testing.T) {
-	thorChain := initEventServer(t, 5)
+	thorChain := initEventServer(t, 5, defaultLogOffset)
 	defer ts.Close()
 	insertBlocks(t, thorChain, 5)
 
@@ -197,10 +200,27 @@ func TestOption(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Equal(t, http.StatusBadRequest, statusCode)
 	assert.Equal(t, "number of criteria in criteriaSet: 11 cannot be greater than: 10\n", string(res))
+
+	filter = api.EventFilter{
+		CriteriaSet: make([]*api.EventCriteria, 0),
+		Range:       nil,
+		Options:     &api.Options{Offset: defaultLogOffset, Limit: new(uint64(0))},
+		Order:       logdb.DESC,
+	}
+
+	_, statusCode, err = tclient.RawHTTPClient().RawHTTPPost("/logs/event", filter)
+	require.NoError(t, err)
+	assert.Equal(t, http.StatusOK, statusCode)
+
+	filter.Options.Offset = defaultLogOffset * 2
+	res, statusCode, err = tclient.RawHTTPClient().RawHTTPPost("/logs/event", filter)
+	require.NoError(t, err)
+	assert.Equal(t, http.StatusForbidden, statusCode)
+	assert.Equal(t, "options.offset exceeds the maximum allowed value of 100", strings.Trim(string(res), "\n"))
 }
 
 func TestZeroFrom(t *testing.T) {
-	thorChain := initEventServer(t, 100)
+	thorChain := initEventServer(t, 100, defaultLogOffset)
 	defer ts.Close()
 	insertBlocks(t, thorChain, 5)
 
@@ -234,7 +254,7 @@ func TestZeroFrom(t *testing.T) {
 }
 
 func TestNullCriteriaSet(t *testing.T) {
-	initEventServer(t, defaultLogLimit)
+	initEventServer(t, defaultLogLimit, defaultLogOffset)
 	defer ts.Close()
 
 	tclient = thorclient.New(ts.URL)
@@ -328,12 +348,12 @@ func testEventWithBlocks(t *testing.T, expectedBlocks int) {
 }
 
 // Init functions
-func initEventServer(t *testing.T, limit uint64) *testchain.Chain {
+func initEventServer(t *testing.T, limit uint64, offset uint64) *testchain.Chain {
 	thorChain, err := testchain.NewDefault()
 	require.NoError(t, err)
 
 	router := mux.NewRouter()
-	New(thorChain.Repo(), thorChain.LogDB(), limit, 10).Mount(router, "/logs/event")
+	New(thorChain.Repo(), thorChain.LogDB(), limit, offset, 10).Mount(router, "/logs/event")
 	ts = httptest.NewServer(router)
 
 	return thorChain

api/events_types.go

@@ -7,7 +7,6 @@ package api
 
 import (
 	"fmt"
-	"math"
 
 	"github.com/ethereum/go-ethereum/common/hexutil"
 
@@ -73,15 +72,16 @@ type Options struct {
 	IncludeIndexes bool    `json:"includeIndexes,omitempty"`
 }
 
-func (o *Options) Validate(limit uint64) error {
+func (o *Options) Validate(limit uint64, offset uint64) error {
 	if o == nil {
 		return nil
 	}
 	if o.Limit != nil && *o.Limit > limit {
 		return fmt.Errorf("options.limit exceeds the maximum allowed value of %d", limit)
 	}
-	if o.Offset > math.MaxInt64 {
-		return fmt.Errorf("options.offset exceeds the maximum allowed value of %d", math.MaxInt64)
+
+	if o.Offset > offset {
+		return fmt.Errorf("options.offset exceeds the maximum allowed value of %d", offset)
 	}
 
 	return nil

api/transfers/transfers.go

@@ -22,15 +22,17 @@ import (
 type Transfers struct {
 	repo             *chain.Repository
 	db               *logdb.LogDB
-	limit            uint64
+	maxLimit         uint64
+	maxOffset        uint64
 	maxCriteriaCount int
 }
 
-func New(repo *chain.Repository, db *logdb.LogDB, logsLimit uint64, maxCriteriaCount int) *Transfers {
+func New(repo *chain.Repository, db *logdb.LogDB, maxLimit uint64, maxOffset uint64, maxCriteriaCount int) *Transfers {
 	return &Transfers{
 		repo,
 		db,
-		logsLimit,
+		maxLimit,
+		maxOffset,
 		maxCriteriaCount,
 	}
 }
@@ -66,7 +68,7 @@ func (t *Transfers) handleFilterTransferLogs(w http.ResponseWriter, req *http.Re
 	if err := restutil.ParseJSON(req.Body, &filter); err != nil {
 		return restutil.BadRequest(errors.WithMessage(err, "body"))
 	}
-	if err := filter.Options.Validate(t.limit); err != nil {
+	if err := filter.Options.Validate(t.maxLimit, t.maxOffset); err != nil {
 		return restutil.Forbidden(err)
 	}
 	if err := filter.Range.Validate(); err != nil {
@@ -91,7 +93,7 @@ func (t *Transfers) handleFilterTransferLogs(w http.ResponseWriter, req *http.Re
 	if filter.Options.Limit == nil {
 		// if filter.Options.Limit is nil, set to the default limit +1
 		// to detect whether there are more logs than the default limit
-		limit := t.limit + 1
+		limit := t.maxLimit + 1
 		filter.Options.Limit = &limit
 	}
 
@@ -101,8 +103,8 @@ func (t *Transfers) handleFilterTransferLogs(w http.ResponseWriter, req *http.Re
 	}
 
 	// ensure the result size is less than the configured limit
-	if len(tLogs) > int(t.limit) {
-		return restutil.Forbidden(fmt.Errorf("the number of filtered logs exceeds the maximum allowed value of %d, please use pagination", t.limit))
+	if len(tLogs) > int(t.maxLimit) {
+		return restutil.Forbidden(fmt.Errorf("the number of filtered logs exceeds the maximum allowed value of %d, please use pagination", t.maxLimit))
 	}
 
 	return restutil.WriteJSON(w, tLogs)

api/transfers/transfers_test.go

@@ -28,7 +28,10 @@ import (
 	"github.com/vechain/thor/v2/tx"
 )
 
-const defaultLogLimit uint64 = 1000
+const (
+	defaultLogLimit  uint64 = 1000
+	defaultLogOffset uint64 = 100
+)
 
 var (
 	ts      *httptest.Server
@@ -37,7 +40,7 @@ var (
 
 func TestEmptyTransfers(t *testing.T) {
 	db := createDb(t)
-	initTransferServer(t, db, defaultLogLimit)
+	initTransferServer(t, db, defaultLogLimit, defaultLogOffset)
 	defer ts.Close()
 
 	tclient = thorclient.New(ts.URL)
@@ -47,7 +50,7 @@ func TestEmptyTransfers(t *testing.T) {
 
 func TestTransfers(t *testing.T) {
 	db := createDb(t)
-	initTransferServer(t, db, defaultLogLimit)
+	initTransferServer(t, db, defaultLogLimit, defaultLogOffset)
 	defer ts.Close()
 
 	tclient = thorclient.New(ts.URL)
@@ -59,7 +62,7 @@ func TestTransfers(t *testing.T) {
 
 func TestOption(t *testing.T) {
 	db := createDb(t)
-	initTransferServer(t, db, 5)
+	initTransferServer(t, db, 5, 100)
 	defer ts.Close()
 	insertBlocks(t, db, 5)
 
@@ -100,11 +103,28 @@ func TestOption(t *testing.T) {
 	require.NoError(t, err)
 	assert.Equal(t, http.StatusForbidden, statusCode)
 	assert.Equal(t, "the number of filtered logs exceeds the maximum allowed value of 5, please use pagination", strings.Trim(string(res), "\n"))
+
+	filter = api.TransferFilter{
+		CriteriaSet: make([]*logdb.TransferCriteria, 0),
+		Range:       nil,
+		Options:     &api.Options{Offset: defaultLogOffset, Limit: new(uint64(0))},
+		Order:       logdb.DESC,
+	}
+
+	_, statusCode, err = tclient.RawHTTPClient().RawHTTPPost("/logs/transfer", filter)
+	require.NoError(t, err)
+	assert.Equal(t, http.StatusOK, statusCode)
+
+	filter.Options.Offset = defaultLogOffset * 2
+	res, statusCode, err = tclient.RawHTTPClient().RawHTTPPost("/logs/transfer", filter)
+	require.NoError(t, err)
+	assert.Equal(t, http.StatusForbidden, statusCode)
+	assert.Equal(t, "options.offset exceeds the maximum allowed value of 100", strings.Trim(string(res), "\n"))
 }
 
 func TestOptionalData(t *testing.T) {
 	db := createDb(t)
-	initTransferServer(t, db, defaultLogLimit)
+	initTransferServer(t, db, defaultLogLimit, defaultLogOffset)
 	defer ts.Close()
 	insertBlocks(t, db, 5)
 	tclient = thorclient.New(ts.URL)
@@ -155,7 +175,7 @@ func TestOptionalData(t *testing.T) {
 
 func TestNullCriteriaSet(t *testing.T) {
 	db := createDb(t)
-	initTransferServer(t, db, defaultLogLimit)
+	initTransferServer(t, db, defaultLogLimit, defaultLogOffset)
 	defer ts.Close()
 	tclient = thorclient.New(ts.URL)
 
@@ -265,12 +285,12 @@ func insertBlocks(t *testing.T, db *logdb.LogDB, n int) {
 	}
 }
 
-func initTransferServer(t *testing.T, logDb *logdb.LogDB, limit uint64) {
+func initTransferServer(t *testing.T, logDb *logdb.LogDB, limit uint64, offset uint64) {
 	thorChain, err := testchain.NewDefault()
 	require.NoError(t, err)
 
 	router := mux.NewRouter()
-	New(thorChain.Repo(), logDb, limit, 10).Mount(router, "/logs/transfer")
+	New(thorChain.Repo(), logDb, limit, offset, 10).Mount(router, "/logs/transfer")
 
 	ts = httptest.NewServer(router)
 }

cmd/thor/flags.go

@@ -105,6 +105,13 @@ var (
 		Usage:   "limit the number of logs returned by /logs API",
 		Sources: envVar("API_LOGS_LIMIT"),
 	}
+	apiLogsMaxOffsetFlag = &cli.Uint64Flag{
+		Name:    "api-logs-max-offset",
+		Local:   true,
+		Value:   100_000,
+		Usage:   "limit the maximum offset for /logs API",
+		Sources: envVar("API_LOGS_MAX_OFFSET"),
+	}
 	apiEnableDeprecatedFlag = &cli.BoolFlag{
 		Name:    "api-enable-deprecated",
 		Local:   true,
@@ -130,6 +137,15 @@ var (
 		Usage:   "enable creation of additional indexes on startup",
 		Sources: envVar("LOGDB_ADDITIONAL_INDEXES"),
 	}
+	logDbMaxReadConnsFlag = &cli.Uint64Flag{
+		Name:    "logdb-max-read-conns",
+		Local:   true,
+		Hidden:  true,
+		Value:   64,
+		Usage:   "maximum number of read connections for the log database",
+		Sources: envVar("LOGDB_MAX_READ_CONNS"),
+	}
+
 	// priority fees API flags
 	apiPriorityFeesPercentageFlag = &cli.Uint64Flag{
 		Name:    "api-priority-fees-percentage",

cmd/thor/httpserver/api_server.go

@@ -65,6 +65,7 @@ type APIConfig struct {
 	Timeout                    int
 	SlowQueriesThreshold       int
 	Log5XXErrors               bool
+	MaxLogsOffset              uint64
 }
 
 func StartAPIServer(
@@ -109,8 +110,8 @@ func StartAPIServer(
 
 	accounts.New(repo, stater, config.CallGasLimit, config.BatchDataMaxSize, forkConfig, bft, config.EnableDeprecated).Mount(router, "/accounts")
 	if !config.SkipLogs {
-		events.New(repo, logDB, config.LogsLimit, defaultMaxCriteriaCount).Mount(router, "/logs/event")
-		transfers.New(repo, logDB, config.LogsLimit, defaultMaxCriteriaCount).Mount(router, "/logs/transfer")
+		events.New(repo, logDB, config.LogsLimit, config.MaxLogsOffset, defaultMaxCriteriaCount).Mount(router, "/logs/event")
+		transfers.New(repo, logDB, config.LogsLimit, config.MaxLogsOffset, defaultMaxCriteriaCount).Mount(router, "/logs/transfer")
 	}
 	blocks.New(repo, bft).Mount(router, "/blocks")
 	transactions.New(repo, txPool).Mount(router, "/transactions")