feat(indexer): push result cap and ordering into KV tx search (PLT-748)

sei-tendermint/cmd/tendermint/commands/reindex_event.go

@@ -118,7 +118,9 @@ func loadEventSinks(cfg *tmcfg.Config) ([]indexer.EventSink, error) {
 			if err != nil {
 				return nil, err
 			}
-			eventSinks = append(eventSinks, kv.NewEventSink(store))
+			// Reindexing only writes to the sink; it never searches, so the
+			// scan budget is irrelevant here.
+			eventSinks = append(eventSinks, kv.NewEventSink(store, nil))
 		case string(indexer.PSQL):
 			conn := cfg.TxIndex.PsqlConn
 			if conn == "" {

sei-tendermint/config/config.go

@@ -528,6 +528,13 @@ type RPCConfig struct {
 	// Maximum number of results returned by tx_search and block_search.
 	// 0 disables the cap (not recommended on public nodes).
 	MaxTxSearchResults int `mapstructure:"max-tx-search-results"`
+
+	// Maximum number of KV index entries that all in-flight tx_search and
+	// block_search requests may visit at once. This is a process-wide safety
+	// cap that bounds peak memory (and scan CPU) under broad or highly
+	// concurrent search load: it is shared across requests, not applied per-query.
+	// 0 disables the cap (not recommended on public nodes).
+	MaxSearchScanBudget int `mapstructure:"max-search-scan-budget"`
 }
 
 // DefaultRPCConfig returns a default configuration for the RPC server
@@ -561,7 +568,8 @@ func DefaultRPCConfig() *RPCConfig {
 		TimeoutReadHeader: 10 * time.Second,
 		TimeoutWrite:      30 * time.Second,
 
-		MaxTxSearchResults: 10_000,
+		MaxTxSearchResults:  10_000,
+		MaxSearchScanBudget: 100_000,
 	}
 }
 
@@ -617,6 +625,9 @@ func (cfg *RPCConfig) ValidateBasic() error {
 	if cfg.MaxTxSearchResults < 0 {
 		return errors.New("max-tx-search-results can't be negative")
 	}
+	if cfg.MaxSearchScanBudget < 0 {
+		return errors.New("max-search-scan-budget can't be negative")
+	}
 	return nil
 }
 

sei-tendermint/config/config_test.go

@@ -77,6 +77,7 @@ func TestRPCConfigValidateBasic(t *testing.T) {
 		"TimeoutReadHeader",
 		"TimeoutWrite",
 		"MaxTxSearchResults",
+		"MaxSearchScanBudget",
 	}
 
 	for _, fieldName := range fieldsToTest {

sei-tendermint/config/toml.go

@@ -286,6 +286,14 @@ timeout-write = "{{ .RPC.TimeoutWrite }}"
 # Set to 0 to disable the cap (not recommended on public nodes).
 max-tx-search-results = {{ .RPC.MaxTxSearchResults }}
 
+# Maximum number of KV index entries that all in-flight tx_search and
+# block_search requests may visit at once. This is a process-wide safety cap
+# that bounds peak memory (and scan CPU) under broad or highly concurrent
+# search load; it is shared across requests, not applied per-query. When the
+# budget is exhausted an in-flight search fails rather than continuing to
+# accumulate. Set to 0 to disable the cap (not recommended on public nodes).
+max-search-scan-budget = {{ .RPC.MaxSearchScanBudget }}
+
 #######################################################################
 ###           P2P Configuration Options                             ###
 #######################################################################

sei-tendermint/internal/rpc/core/tx.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"sort"
 
 	tmquery "github.com/sei-protocol/sei-chain/sei-tendermint/internal/pubsub/query"
 	"github.com/sei-protocol/sei-chain/sei-tendermint/internal/state/indexer"
@@ -65,40 +64,39 @@ func (env *Environment) TxSearch(ctx context.Context, req *coretypes.RequestTxSe
 		return nil, err
 	}
 
+	// Validate order_by up front so we can push the ordering (and the result
+	// cap) down into the indexer; a broad query is then bounded at the scan
+	// path rather than after materializing and sorting the full match set.
+	var orderDesc bool
+	switch req.OrderBy {
+	case DescendingOrder, "":
+		orderDesc = true
+
+	case AscendingOrder:
+		orderDesc = false
+
+	default:
+		return nil, fmt.Errorf("expected order_by to be either `asc` or `desc` or empty: %w", coretypes.ErrInvalidRequest)
+	}
+
 	for _, sink := range env.EventSinks {
 		if sink.Type() == indexer.KV {
-			// TODO(PLT-748): the kv tx indexer currently ignores these opts and
-			// the cap below still applies after sort (PLT-700). Once the tx
-			// scan path is bounded like the block indexer, this pushes the cap
-			// and ordering down to the scan.
 			results, err := sink.SearchTxEvents(ctx, q, indexer.SearchOptions{
 				Limit:     env.Config.MaxTxSearchResults,
-				OrderDesc: req.OrderBy != AscendingOrder,
+				OrderDesc: orderDesc,
 			})
 			if err != nil {
 				return nil, err
 			}
 
-			// sort results (must be done before cap and pagination)
-			switch req.OrderBy {
-			case DescendingOrder, "":
-				sort.Slice(results, func(i, j int) bool {
-					if results[i].Height == results[j].Height {
-						return results[i].Index > results[j].Index
-					}
-					return results[i].Height > results[j].Height
-				})
-			case AscendingOrder:
-				sort.Slice(results, func(i, j int) bool {
-					if results[i].Height == results[j].Height {
-						return results[i].Index < results[j].Index
-					}
-					return results[i].Height < results[j].Height
-				})
-			default:
-				return nil, fmt.Errorf("expected order_by to be either `asc` or `desc` or empty: %w", coretypes.ErrInvalidRequest)
-			}
+			// Results already arrive ordered by (height, index) per orderDesc:
+			// the kv indexer either sorts the materialized set or scans its
+			// order-preserving secondary index in order_by order, so no re-sort
+			// is needed here.
 
+			// Safety net: the kv indexer already bounds to MaxTxSearchResults,
+			// but keep the cap so the response stays bounded for any sink that
+			// ignores the limit.
 			if max := env.Config.MaxTxSearchResults; max > 0 && len(results) > max {
 				results = results[:max]
 			}

sei-tendermint/internal/rpc/core/tx_test.go

@@ -1,13 +1,15 @@
 package core
 
 import (
+	"context"
 	"testing"
 
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 
 	abci "github.com/sei-protocol/sei-chain/sei-tendermint/abci/types"
 	"github.com/sei-protocol/sei-chain/sei-tendermint/config"
+	"github.com/sei-protocol/sei-chain/sei-tendermint/internal/pubsub/query"
 	"github.com/sei-protocol/sei-chain/sei-tendermint/internal/state/indexer"
 	indexermocks "github.com/sei-protocol/sei-chain/sei-tendermint/internal/state/indexer/mocks"
 	"github.com/sei-protocol/sei-chain/sei-tendermint/rpc/coretypes"
@@ -25,7 +27,23 @@ func txSearchEnv(t *testing.T, maxResults int, txs []*abci.TxResultV2) *Environm
 	t.Helper()
 	sink := indexermocks.NewEventSink(t)
 	sink.On("Type").Return(indexer.KV)
-	sink.On("SearchTxEvents", mock.Anything, mock.Anything, mock.Anything).Return(txs, nil)
+	sink.On("SearchTxEvents", mock.Anything, mock.Anything, mock.Anything).Return(
+		func(_ context.Context, _ *query.Query, opts indexer.SearchOptions) []*abci.TxResultV2 {
+			ordered := make([]*abci.TxResultV2, len(txs))
+			if opts.OrderDesc {
+				for i, tx := range txs {
+					ordered[len(txs)-1-i] = tx
+				}
+			} else {
+				copy(ordered, txs)
+			}
+			if opts.Limit > 0 && len(ordered) > opts.Limit {
+				ordered = ordered[:opts.Limit]
+			}
+			return ordered
+		},
+		nil,
+	)
 	return &Environment{
 		EventSinks: []indexer.EventSink{sink},
 		Config:     config.RPCConfig{MaxTxSearchResults: maxResults},