diff --git a/sei-tendermint/cmd/tendermint/commands/reindex_event.go b/sei-tendermint/cmd/tendermint/commands/reindex_event.go index d0eea56b2a..b33a0012a1 100644 --- a/sei-tendermint/cmd/tendermint/commands/reindex_event.go +++ b/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 == "" { diff --git a/sei-tendermint/config/config.go b/sei-tendermint/config/config.go index 57d6c158df..e3adbc483b 100644 --- a/sei-tendermint/config/config.go +++ b/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 } diff --git a/sei-tendermint/config/config_test.go b/sei-tendermint/config/config_test.go index 811ba5fb55..05c9aadec6 100644 --- a/sei-tendermint/config/config_test.go +++ b/sei-tendermint/config/config_test.go @@ -77,6 +77,7 @@ func TestRPCConfigValidateBasic(t *testing.T) { "TimeoutReadHeader", "TimeoutWrite", "MaxTxSearchResults", + "MaxSearchScanBudget", } for _, fieldName := range fieldsToTest { diff --git a/sei-tendermint/config/toml.go b/sei-tendermint/config/toml.go index 3e4c30ee21..f351f29c9e 100644 --- a/sei-tendermint/config/toml.go +++ b/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 ### ####################################################################### diff --git a/sei-tendermint/internal/rpc/core/tx.go b/sei-tendermint/internal/rpc/core/tx.go index 3a363f10a8..00c847b560 100644 --- a/sei-tendermint/internal/rpc/core/tx.go +++ b/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] } diff --git a/sei-tendermint/internal/rpc/core/tx_test.go b/sei-tendermint/internal/rpc/core/tx_test.go index 88bab405d2..9ea8e8a1fe 100644 --- a/sei-tendermint/internal/rpc/core/tx_test.go +++ b/sei-tendermint/internal/rpc/core/tx_test.go @@ -1,6 +1,7 @@ package core import ( + "context" "testing" "github.com/stretchr/testify/mock" @@ -8,6 +9,7 @@ import ( 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}, diff --git a/sei-tendermint/internal/state/indexer/block/kv/kv.go b/sei-tendermint/internal/state/indexer/block/kv/kv.go index d4aa80cdec..48e60a8c79 100644 --- a/sei-tendermint/internal/state/indexer/block/kv/kv.go +++ b/sei-tendermint/internal/state/indexer/block/kv/kv.go @@ -26,6 +26,10 @@ var _ indexer.BlockIndexer = (*BlockerIndexer)(nil) // such that matching search criteria returns the respective block height(s). type BlockerIndexer struct { store dbm.DB + // budget bounds the index entries in-flight searches may visit; nil means + // unlimited. It is typically shared with the tx indexer so the cap is + // process-wide across tx_search and block_search. + budget *indexer.ScanBudget } func New(store dbm.DB) *BlockerIndexer { @@ -34,6 +38,13 @@ func New(store dbm.DB) *BlockerIndexer { } } +// WithScanBudget attaches a shared scan budget that bounds how many index +// entries in-flight searches may visit. +func (idx *BlockerIndexer) WithScanBudget(budget *indexer.ScanBudget) *BlockerIndexer { + idx.budget = budget + return idx +} + // Has returns true if the given height has been indexed. An error is returned // upon database query failure. func (idx *BlockerIndexer) Has(height int64) (bool, error) { @@ -84,6 +95,9 @@ func (idx *BlockerIndexer) Search(ctx context.Context, q *query.Query, opts inde return results, nil } + lease := idx.budget.Lease() + defer lease.Release() + conditions := q.Syntax() // If there is an exact height query, return the result immediately @@ -111,14 +125,14 @@ func (idx *BlockerIndexer) Search(ctx context.Context, q *query.Query, opts inde // the limit, so a broad query does not materialize and sort the full match // set. if plan, ok := planBounded(conditions, ranges, rangeIndexes); ok { - return idx.searchBounded(ctx, plan, opts) + return idx.searchBounded(ctx, plan, opts, lease) } // Fallback: queries containing CONTAINS/MATCHES/EXISTS or non-height ranges // cannot be point-probed against a candidate height (the block's events // live only in the index, so there is nothing cheap to fetch). Materialize // the intersection as before, then bound and order the result set. - filteredHeights, err := idx.intersect(ctx, conditions, ranges, rangeIndexes) + filteredHeights, err := idx.intersect(ctx, conditions, ranges, rangeIndexes, lease) if err != nil { return nil, err } @@ -135,6 +149,7 @@ func (idx *BlockerIndexer) intersect( conditions []syntax.Condition, ranges indexer.QueryRanges, rangeIndexes []int, + lease *indexer.ScanLease, ) (map[string][]byte, error) { var heightsInitialized bool filteredHeights := make(map[string][]byte) @@ -152,7 +167,7 @@ func (idx *BlockerIndexer) intersect( } if !heightsInitialized { - filteredHeights, err = idx.matchRange(ctx, qr, prefix, filteredHeights, true) + filteredHeights, err = idx.matchRange(ctx, qr, prefix, filteredHeights, true, lease) if err != nil { return nil, err } @@ -165,7 +180,7 @@ func (idx *BlockerIndexer) intersect( break } } else { - filteredHeights, err = idx.matchRange(ctx, qr, prefix, filteredHeights, false) + filteredHeights, err = idx.matchRange(ctx, qr, prefix, filteredHeights, false, lease) if err != nil { return nil, err } @@ -185,7 +200,7 @@ func (idx *BlockerIndexer) intersect( } if !heightsInitialized { - filteredHeights, err = idx.match(ctx, c, startKey, filteredHeights, true) + filteredHeights, err = idx.match(ctx, c, startKey, filteredHeights, true, lease) if err != nil { return nil, err } @@ -198,7 +213,7 @@ func (idx *BlockerIndexer) intersect( break } } else { - filteredHeights, err = idx.match(ctx, c, startKey, filteredHeights, false) + filteredHeights, err = idx.match(ctx, c, startKey, filteredHeights, false, lease) if err != nil { return nil, err } @@ -243,32 +258,18 @@ func (idx *BlockerIndexer) collectBounded(ctx context.Context, filteredHeights m return results, nil } -// searchBounded executes a boundedPlan: it scans the driver prefix in -// order_by order, point-probes the remaining conditions per candidate height, -// and stops as soon as opts.Limit matches are collected. Memory is bounded by -// the number of results kept rather than by the full match cardinality. -func (idx *BlockerIndexer) searchBounded(ctx context.Context, plan boundedPlan, opts indexer.SearchOptions) ([]int64, error) { - var ( - prefix []byte - err error - ) - if plan.driverEquality != nil { - prefix, err = orderedcode.Append(nil, plan.driverEquality.Tag, plan.driverEquality.Arg.Value()) - } else { - // Drive off the primary block.height key range. - prefix, err = orderedcode.Append(nil, types.BlockHeightKey) - } +// searchBounded executes a boundedPlan: it scans the driver in order_by order, +// point-probes the remaining conditions per candidate height, and stops as soon +// as opts.Limit matches are collected. Memory is bounded by the number of +// results kept rather than by the full match cardinality. +func (idx *BlockerIndexer) searchBounded(ctx context.Context, plan boundedPlan, opts indexer.SearchOptions, lease *indexer.ScanLease) ([]int64, error) { + it, err := idx.driverIterator(plan, opts.OrderDesc) if err != nil { - return nil, fmt.Errorf("failed to create driver prefix key: %w", err) - } - - it, err := idx.prefixIterator(prefix, opts.OrderDesc) - if err != nil { - return nil, fmt.Errorf("failed to create driver iterator: %w", err) + return nil, err } defer func() { _ = it.Close() }() - results := make([]int64, 0, boundedCap(opts.Limit)) + results := make([]int64, 0, indexer.BoundedCap(opts.Limit)) seen := make(map[int64]struct{}) for ; it.Valid(); it.Next() { @@ -276,12 +277,17 @@ func (idx *BlockerIndexer) searchBounded(ctx context.Context, plan boundedPlan, break } + // Charge each driver entry the scan walks against the shared budget. + if err := lease.Visit(1); err != nil { + return nil, err + } + h := int64FromBytes(it.Value()) if _, dup := seen[h]; dup { continue } - match, err := idx.candidateMatches(h, plan) + match, err := idx.candidateMatches(h, plan, lease) if err != nil { return nil, err } @@ -304,18 +310,61 @@ func (idx *BlockerIndexer) searchBounded(ctx context.Context, plan boundedPlan, return results, nil } +// driverIterator builds the height-ordered iterator that searchBounded walks. +// An equality driver scans that condition's event-key prefix. A primary +// block.height range (no equality) is seeked to its inclusive [lower, upper] +// bounds so out-of-range heights are never walked or charged against the +// budget. +func (idx *BlockerIndexer) driverIterator(plan boundedPlan, desc bool) (dbm.Iterator, error) { + if plan.driverEquality != nil { + prefix, err := orderedcode.Append(nil, plan.driverEquality.Tag, plan.driverEquality.Arg.Value()) + if err != nil { + return nil, fmt.Errorf("failed to create driver prefix key: %w", err) + } + return idx.prefixIterator(prefix, desc) + } + + base, err := orderedcode.Append(nil, types.BlockHeightKey) + if err != nil { + return nil, fmt.Errorf("failed to create driver prefix key: %w", err) + } + + start, end := base, indexer.PrefixUpperBound(base) + qr := plan.heightRanges[0] + if lb, ok := qr.LowerBoundValue().(int64); ok { + if start, err = heightKey(lb); err != nil { + return nil, fmt.Errorf("failed to create lower-bound key: %w", err) + } + } + if ub, ok := qr.UpperBoundValue().(int64); ok { + ubKey, err := heightKey(ub) + if err != nil { + return nil, fmt.Errorf("failed to create upper-bound key: %w", err) + } + end = indexer.PrefixUpperBound(ubKey) + } + + if desc { + return idx.store.ReverseIterator(start, end) + } + return idx.store.Iterator(start, end) +} + // candidateMatches reports whether the block at height h satisfies every // non-driver condition in the plan: height-range bounds are evaluated directly // from h, and equality probes are tested with a single point lookup against the -// event index. -func (idx *BlockerIndexer) candidateMatches(h int64, plan boundedPlan) (bool, error) { +// event index. Each probe is charged against lease. +func (idx *BlockerIndexer) candidateMatches(h int64, plan boundedPlan, lease *indexer.ScanLease) (bool, error) { for i := range plan.heightRanges { - if !heightInRange(h, plan.heightRanges[i]) { + if !indexer.HeightInRange(h, plan.heightRanges[i]) { return false, nil } } for i := range plan.equalityProbes { + if err := lease.Visit(1); err != nil { + return false, err + } c := plan.equalityProbes[i] ok, err := idx.hasEvent(c.Tag, c.Arg.Value(), h) if err != nil { @@ -338,7 +387,7 @@ func (idx *BlockerIndexer) prefixIterator(prefix []byte, desc bool) (dbm.Iterato if !desc { return dbm.IteratePrefix(idx.store, prefix) } - return idx.store.ReverseIterator(prefix, prefixUpperBound(prefix)) + return idx.store.ReverseIterator(prefix, indexer.PrefixUpperBound(prefix)) } // hasEvent reports whether the block at the given height has an indexed event @@ -436,6 +485,7 @@ func (idx *BlockerIndexer) matchRange( startKey []byte, filteredHeights map[string][]byte, firstRun bool, + lease *indexer.ScanLease, ) (map[string][]byte, error) { // A previous match was attempted but resulted in no matches, so we return @@ -456,6 +506,10 @@ func (idx *BlockerIndexer) matchRange( iter: for ; it.Valid(); it.Next() { + if err := lease.Visit(1); err != nil { + return nil, err + } + var ( eventValue string err error @@ -544,6 +598,7 @@ func (idx *BlockerIndexer) match( startKeyBz []byte, filteredHeights map[string][]byte, firstRun bool, + lease *indexer.ScanLease, ) (map[string][]byte, error) { // A previous match was attempted but resulted in no matches, so we return @@ -563,6 +618,10 @@ func (idx *BlockerIndexer) match( defer func() { _ = it.Close() }() for ; it.Valid(); it.Next() { + if err := lease.Visit(1); err != nil { + return nil, err + } + tmpHeights[string(it.Value())] = it.Value() if err := ctx.Err(); err != nil { @@ -588,6 +647,10 @@ func (idx *BlockerIndexer) match( iterExists: for ; it.Valid(); it.Next() { + if err := lease.Visit(1); err != nil { + return nil, err + } + tmpHeights[string(it.Value())] = it.Value() select { @@ -616,6 +679,10 @@ func (idx *BlockerIndexer) match( iterContains: for ; it.Valid(); it.Next() { + if err := lease.Visit(1); err != nil { + return nil, err + } + eventValue, err := parseValueFromEventKey(it.Key()) if err != nil { continue @@ -650,6 +717,10 @@ func (idx *BlockerIndexer) match( iterMatches: for ; it.Valid(); it.Next() { + if err := lease.Visit(1); err != nil { + return nil, err + } + eventValue, err := parseValueFromEventKey(it.Key()) if err != nil { continue diff --git a/sei-tendermint/internal/state/indexer/block/kv/kv_test.go b/sei-tendermint/internal/state/indexer/block/kv/kv_test.go index 865bf5d519..cf1c4d2972 100644 --- a/sei-tendermint/internal/state/indexer/block/kv/kv_test.go +++ b/sei-tendermint/internal/state/indexer/block/kv/kv_test.go @@ -338,3 +338,77 @@ func TestBlockIndexerBounded(t *testing.T) { } }) } + +// TestBlockIndexerScanBudget guards the range-only fast path against the scan +// budget. +func TestBlockIndexerScanBudget(t *testing.T) { + store := dbm.NewPrefixDB(dbm.NewMemDB(), []byte("block_events_budget")) + + // A chain far larger than the budget. Every height carries app.name=sei. + const chain = 50 + for i := int64(1); i <= chain; i++ { + require.NoError(t, blockidxkv.New(store).Index(types.EventDataNewBlockHeader{ + Header: types.Header{Height: i}, + ResultFinalizeBlock: abci.ResponseFinalizeBlock{Events: []abci.Event{{ + Type: "app", + Attributes: []abci.EventAttribute{{Key: []byte("name"), Value: []byte("sei"), Index: true}}, + }}}, + })) + } + + // A budget far smaller than the chain, but larger than any of the ranges + // queried below, so an in-range scan never trips it. + idx := blockidxkv.New(store).WithScanBudget(indexer.NewScanBudget(10)) + + seekCases := map[string]struct { + q string + opts indexer.SearchOptions + results []int64 + }{ + // Descending scan, upper-bounded range: matches are the lowest heights, + // i.e. the tail of a high->low walk. + "upper bound desc": { + q: `block.height <= 5`, + opts: indexer.SearchOptions{OrderDesc: true}, + results: []int64{5, 4, 3, 2, 1}, + }, + // Ascending scan, lower-bounded range: matches are the highest heights, + // i.e. the tail of a low->high walk. + "lower bound asc": { + q: `block.height >= 46`, + opts: indexer.SearchOptions{OrderDesc: false}, + results: []int64{46, 47, 48, 49, 50}, + }, + "lower bound desc": { + q: `block.height >= 46`, + opts: indexer.SearchOptions{OrderDesc: true}, + results: []int64{50, 49, 48, 47, 46}, + }, + "dual bounded asc": { + q: `block.height >= 3 AND block.height <= 7`, + opts: indexer.SearchOptions{OrderDesc: false}, + results: []int64{3, 4, 5, 6, 7}, + }, + } + for name, tc := range seekCases { + t.Run(name, func(t *testing.T) { + results, err := idx.Search(t.Context(), query.MustCompile(tc.q), tc.opts) + require.NoError(t, err) + require.Equal(t, tc.results, results) + }) + } + + t.Run("in-range set larger than budget errors", func(t *testing.T) { + tight := blockidxkv.New(store).WithScanBudget(indexer.NewScanBudget(3)) + results, err := tight.Search(t.Context(), query.MustCompile(`block.height <= 40`), indexer.SearchOptions{OrderDesc: true}) + require.ErrorIs(t, err, indexer.ErrScanBudgetExceeded) + require.Nil(t, results) + }) + + t.Run("broad fallback errors", func(t *testing.T) { + tight := blockidxkv.New(store).WithScanBudget(indexer.NewScanBudget(4)) + results, err := tight.Search(t.Context(), query.MustCompile(`app.name CONTAINS 'se'`), indexer.SearchOptions{OrderDesc: true}) + require.ErrorIs(t, err, indexer.ErrScanBudgetExceeded) + require.Nil(t, results) + }) +} diff --git a/sei-tendermint/internal/state/indexer/block/kv/util.go b/sei-tendermint/internal/state/indexer/block/kv/util.go index 1a971bd9fa..3c524f36ab 100644 --- a/sei-tendermint/internal/state/indexer/block/kv/util.go +++ b/sei-tendermint/internal/state/indexer/block/kv/util.go @@ -8,58 +8,9 @@ import ( "github.com/google/orderedcode" "github.com/sei-protocol/sei-chain/sei-tendermint/internal/pubsub/query/syntax" - "github.com/sei-protocol/sei-chain/sei-tendermint/internal/state/indexer" "github.com/sei-protocol/sei-chain/sei-tendermint/types" ) -// maxBoundedPrealloc caps how much the bounded fast path preallocates for its -// result slice, so a very large (or disabled) limit does not eagerly allocate. -const maxBoundedPrealloc = 4096 - -// boundedCap returns a sensible initial capacity for a result slice given a -// limit. A non-positive limit means "unbounded", in which case we let the -// slice grow on demand. -func boundedCap(limit int) int { - if limit <= 0 { - return 0 - } - if limit < maxBoundedPrealloc { - return limit - } - return maxBoundedPrealloc -} - -// heightInRange reports whether height h falls within the (already -// inclusivity-adjusted) bounds of a numeric block.height query range. -func heightInRange(h int64, qr indexer.QueryRange) bool { - if lower := qr.LowerBoundValue(); lower != nil { - if lb, ok := lower.(int64); ok && h < lb { - return false - } - } - if upper := qr.UpperBoundValue(); upper != nil { - if ub, ok := upper.(int64); ok && h > ub { - return false - } - } - return true -} - -// prefixUpperBound returns the exclusive end key for iterating over prefix, -// i.e. the smallest key strictly greater than every key having the prefix. -// It returns nil when prefix is empty or all bytes are 0xFF (no upper bound). -func prefixUpperBound(prefix []byte) []byte { - end := make([]byte, len(prefix)) - copy(end, prefix) - for i := len(end) - 1; i >= 0; i-- { - if end[i] != 0xFF { - end[i]++ - return end[:i+1] - } - } - return nil -} - func intInSlice(a int, list []int) bool { for _, b := range list { if b == a { diff --git a/sei-tendermint/internal/state/indexer/budget.go b/sei-tendermint/internal/state/indexer/budget.go new file mode 100644 index 0000000000..60895bb35a --- /dev/null +++ b/sei-tendermint/internal/state/indexer/budget.go @@ -0,0 +1,83 @@ +package indexer + +import ( + "errors" + "sync/atomic" +) + +// ErrScanBudgetExceeded is returned by a KV index Search when the aggregate +// number of index entries visited by all in-flight searches would exceed the +// configured scan budget. It signals overload: the caller should narrow the +// query or retry later. +var ErrScanBudgetExceeded = errors.New("kv indexer scan budget exceeded; narrow the query or retry later") + +// ScanBudget bounds the total number of index entries that all in-flight KV +// index searches (tx_search and block_search) may visit at once. +// The zero value and a nil *ScanBudget both behave as "unlimited". +type ScanBudget struct { + // max is the ceiling on the number of concurrently-charged entries. + // A value <= 0 means unlimited. + max int64 + // inFlight is the number of entries currently charged by open leases. + inFlight atomic.Int64 +} + +// NewScanBudget returns a ScanBudget capped at max entries. A non-positive max +// disables the cap (unlimited). +func NewScanBudget(max int) *ScanBudget { + return &ScanBudget{max: int64(max)} +} + +// Lease opens a per-search charge against the budget. A search calls Visit as +// it walks index entries and Release (typically via defer) when it returns, so +// its entries are returned to the shared pool. +func (b *ScanBudget) Lease() *ScanLease { + return &ScanLease{budget: b} +} + +// InFlight returns the number of entries currently charged across all open +// leases. Intended for tests and metrics. +func (b *ScanBudget) InFlight() int64 { + if b == nil { + return 0 + } + return b.inFlight.Load() +} + +// ScanLease tracks the index entries a single search has charged against a +// ScanBudget. Each search must use its own lease; a ScanLease is not safe for +// concurrent use. +type ScanLease struct { + budget *ScanBudget + // held is the number of entries this lease has charged to the shared + // counter and not yet released. + held int64 +} + +// Visit records that the search has walked n more index entries and charges +// them to the shared budget. It returns ErrScanBudgetExceeded once the +// aggregate in-flight charge exceeds the budget's ceiling. +// +// The cap is soft: entries are charged before the ceiling is checked, so under +// concurrency the aggregate can overshoot max by roughly the number of +// concurrent leases before any of them observes the excess. +func (l *ScanLease) Visit(n int64) error { + if l == nil || l.budget == nil || l.budget.max <= 0 || n == 0 { + return nil + } + total := l.budget.inFlight.Add(n) + l.held += n + if total > l.budget.max { + return ErrScanBudgetExceeded + } + return nil +} + +// Release returns every entry this lease has charged to the shared budget. +func (l *ScanLease) Release() { + if l == nil || l.budget == nil || l.held == 0 { + return + } + l.budget.inFlight.Add(-l.held) + l.held = 0 +} diff --git a/sei-tendermint/internal/state/indexer/budget_test.go b/sei-tendermint/internal/state/indexer/budget_test.go new file mode 100644 index 0000000000..873c0bdc2a --- /dev/null +++ b/sei-tendermint/internal/state/indexer/budget_test.go @@ -0,0 +1,160 @@ +package indexer_test + +import ( + "errors" + "sync" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/sei-protocol/sei-chain/sei-tendermint/internal/state/indexer" +) + +func TestScanBudget_NilAndZeroAreUnlimited(t *testing.T) { + // A nil *ScanBudget behaves as unlimited: its lease never rejects. + var nilBudget *indexer.ScanBudget + require.Equal(t, int64(0), nilBudget.InFlight()) + + lease := nilBudget.Lease() + require.NoError(t, lease.Visit(1<<30)) + require.Equal(t, int64(0), nilBudget.InFlight()) + lease.Release() // must not panic + + // A zero/non-positive max also disables the cap. + for _, maxBudgets := range []int{0, -1} { + b := indexer.NewScanBudget(maxBudgets) + l := b.Lease() + require.NoError(t, l.Visit(1<<30)) + require.Equal(t, int64(0), b.InFlight(), "unlimited budget should not track charges") + } +} + +func TestScanBudget_ChargesAndReleases(t *testing.T) { + b := indexer.NewScanBudget(100) + lease := b.Lease() + + require.NoError(t, lease.Visit(10)) + require.Equal(t, int64(10), b.InFlight()) + + require.NoError(t, lease.Visit(40)) + require.Equal(t, int64(50), b.InFlight()) + + lease.Release() + require.Equal(t, int64(0), b.InFlight()) + + // Release is idempotent: a second call returns nothing further. + lease.Release() + require.Equal(t, int64(0), b.InFlight()) +} + +func TestScanBudget_VisitZeroIsNoop(t *testing.T) { + b := indexer.NewScanBudget(10) + lease := b.Lease() + + require.NoError(t, lease.Visit(0)) + require.Equal(t, int64(0), b.InFlight()) +} + +func TestScanBudget_ExceedReturnsError(t *testing.T) { + b := indexer.NewScanBudget(10) + lease := b.Lease() + + // Reaching exactly the ceiling is allowed; only exceeding it fails. + require.NoError(t, lease.Visit(10)) + require.Equal(t, int64(10), b.InFlight()) + + err := lease.Visit(1) + require.ErrorIs(t, err, indexer.ErrScanBudgetExceeded) + + // The over-limit entries are still charged until Release (the search + // aborts and releases via defer), so the caller sees the overshoot. + require.Equal(t, int64(11), b.InFlight()) + + lease.Release() + require.Equal(t, int64(0), b.InFlight()) +} + +func TestScanBudget_SharedAcrossLeases(t *testing.T) { + b := indexer.NewScanBudget(10) + a := b.Lease() + c := b.Lease() + + require.NoError(t, a.Visit(6)) + require.NoError(t, c.Visit(4)) + require.Equal(t, int64(10), b.InFlight()) + + // The next visit on either lease exceeds the shared ceiling. + require.ErrorIs(t, c.Visit(1), indexer.ErrScanBudgetExceeded) + + // Releasing one lease returns only its own charge to the pool. + a.Release() + require.Equal(t, int64(5), b.InFlight()) + c.Release() + require.Equal(t, int64(0), b.InFlight()) +} + +func TestScanBudget_NilLeaseSafe(t *testing.T) { + var lease *indexer.ScanLease + require.NoError(t, lease.Visit(5)) + require.NotPanics(t, func() { lease.Release() }) +} + +// TestScanBudget_ConcurrentReleaseBalances exercises the shared counter under +// concurrency: every charged entry must be returned, so InFlight settles back +// to zero once all leases release regardless of interleaving. +func TestScanBudget_ConcurrentReleaseBalances(t *testing.T) { + const workers = 50 + const perWorker = 20 + + // Large ceiling so no worker trips the cap; we're checking accounting. + b := indexer.NewScanBudget(workers * perWorker) + + var wg sync.WaitGroup + for range workers { + wg.Go(func() { + lease := b.Lease() + defer lease.Release() + for range perWorker { + _ = lease.Visit(1) + } + }) + } + wg.Wait() + + require.Equal(t, int64(0), b.InFlight(), "all charges must be released") +} + +// TestScanBudget_SoftCapOvershoot documents the intentionally soft cap: because +// each lease charges before observing the ceiling, concurrent leases can push +// the aggregate past max, and every one of them may still get an error. +func TestScanBudget_SoftCapOvershoot(t *testing.T) { + const leases = 8 + // Ceiling of 1 with each lease charging 1: the first Add already reaches + // the ceiling, so every concurrent lease charging 1 overshoots. + b := indexer.NewScanBudget(1) + + var wg sync.WaitGroup + errs := make([]error, leases) + start := make(chan struct{}) + for i := range leases { + wg.Go(func() { + lease := b.Lease() + <-start + errs[i] = lease.Visit(1) + }) + } + close(start) + wg.Wait() + + // The aggregate overshot the ceiling of 1 by the number of leases. + require.Equal(t, int64(leases), b.InFlight()) + + // At least one lease saw the budget exceeded (all but possibly the first). + exceeded := 0 + for _, err := range errs { + if errors.Is(err, indexer.ErrScanBudgetExceeded) { + exceeded++ + } + } + require.GreaterOrEqual(t, exceeded, leases-1) +} diff --git a/sei-tendermint/internal/state/indexer/indexer_service_test.go b/sei-tendermint/internal/state/indexer/indexer_service_test.go index 70b1647c9b..2359394fa2 100644 --- a/sei-tendermint/internal/state/indexer/indexer_service_test.go +++ b/sei-tendermint/internal/state/indexer/indexer_service_test.go @@ -53,7 +53,7 @@ func TestIndexerServiceIndexesBlocks(t *testing.T) { pool := setupDB(t) store := dbm.NewMemDB() - eventSinks := []indexer.EventSink{kv.NewEventSink(store), pSink} + eventSinks := []indexer.EventSink{kv.NewEventSink(store, nil), pSink} assert.True(t, indexer.KVSinkEnabled(eventSinks)) assert.True(t, indexer.IndexingEnabled(eventSinks)) diff --git a/sei-tendermint/internal/state/indexer/sink/kv/kv.go b/sei-tendermint/internal/state/indexer/sink/kv/kv.go index 7483160d60..df0f16527b 100644 --- a/sei-tendermint/internal/state/indexer/sink/kv/kv.go +++ b/sei-tendermint/internal/state/indexer/sink/kv/kv.go @@ -23,10 +23,15 @@ type EventSink struct { store dbm.DB } -func NewEventSink(store dbm.DB) indexer.EventSink { +// NewEventSink creates a KV event sink backed by store. budget, when non-nil, +// is a shared scan budget that bounds how many index entries all in-flight +// tx_search and block_search requests may visit at once; passing nil disables +// the cap. The same budget is shared by the tx and block indexers so the cap is +// process-wide across both. +func NewEventSink(store dbm.DB, budget *indexer.ScanBudget) indexer.EventSink { return &EventSink{ - txi: kvt.NewTxIndex(store), - bi: kvb.New(store), + txi: kvt.NewTxIndex(store).WithScanBudget(budget), + bi: kvb.New(store).WithScanBudget(budget), store: store, } } diff --git a/sei-tendermint/internal/state/indexer/sink/kv/kv_test.go b/sei-tendermint/internal/state/indexer/sink/kv/kv_test.go index 5dc7d8fd56..bac4253e6a 100644 --- a/sei-tendermint/internal/state/indexer/sink/kv/kv_test.go +++ b/sei-tendermint/internal/state/indexer/sink/kv/kv_test.go @@ -23,18 +23,18 @@ import ( var searchOpts = indexer.SearchOptions{} func TestType(t *testing.T) { - kvSink := NewEventSink(dbm.NewMemDB()) + kvSink := NewEventSink(dbm.NewMemDB(), nil) assert.Equal(t, indexer.KV, kvSink.Type()) } func TestStop(t *testing.T) { - kvSink := NewEventSink(dbm.NewMemDB()) + kvSink := NewEventSink(dbm.NewMemDB(), nil) assert.Nil(t, kvSink.Stop()) } func TestBlockFuncs(t *testing.T) { store := dbm.NewPrefixDB(dbm.NewMemDB(), []byte("block_events")) - indexer := NewEventSink(store) + indexer := NewEventSink(store, nil) require.NoError(t, indexer.IndexBlockEvents(types.EventDataNewBlockHeader{ Header: types.Header{Height: 1}, @@ -157,7 +157,7 @@ func TestBlockFuncs(t *testing.T) { } func TestTxSearchWithCancelation(t *testing.T) { - indexer := NewEventSink(dbm.NewMemDB()) + indexer := NewEventSink(dbm.NewMemDB(), nil) txResult := txResultWithEvents([]abci.Event{ {Type: "account", Attributes: []abci.EventAttribute{{Key: []byte("number"), Value: []byte("1"), Index: true}}}, @@ -180,7 +180,7 @@ func TestTxSearchWithCancelation(t *testing.T) { func TestTxSearchDeprecatedIndexing(t *testing.T) { esdb := dbm.NewMemDB() - indexer := NewEventSink(esdb) + indexer := NewEventSink(esdb, nil) // index tx using events indexing (composite key) txResult1 := txResultWithEvents([]abci.Event{ @@ -258,7 +258,7 @@ func TestTxSearchDeprecatedIndexing(t *testing.T) { } func TestTxSearchOneTxWithMultipleSameTagsButDifferentValues(t *testing.T) { - indexer := NewEventSink(dbm.NewMemDB()) + indexer := NewEventSink(dbm.NewMemDB(), nil) txResult := txResultWithEvents([]abci.Event{ {Type: "account", Attributes: []abci.EventAttribute{{Key: []byte("number"), Value: []byte("1"), Index: true}}}, @@ -280,7 +280,7 @@ func TestTxSearchOneTxWithMultipleSameTagsButDifferentValues(t *testing.T) { } func TestTxSearchMultipleTxs(t *testing.T) { - indexer := NewEventSink(dbm.NewMemDB()) + indexer := NewEventSink(dbm.NewMemDB(), nil) // indexed first, but bigger height (to test the order of transactions) txResult := txResultWithEvents([]abci.Event{ diff --git a/sei-tendermint/internal/state/indexer/sink/sink.go b/sei-tendermint/internal/state/indexer/sink/sink.go index 009aa2c37f..8045570516 100644 --- a/sei-tendermint/internal/state/indexer/sink/sink.go +++ b/sei-tendermint/internal/state/indexer/sink/sink.go @@ -41,7 +41,10 @@ func EventSinksFromConfig(cfg *config.Config, dbProvider config.DBProvider, chai return nil, err } - eventSinks = append(eventSinks, kv.NewEventSink(store)) + // Share one scan budget across the tx and block indexers so the + // cap on entries visited by in-flight searches is process-wide. + budget := indexer.NewScanBudget(cfg.RPC.MaxSearchScanBudget) + eventSinks = append(eventSinks, kv.NewEventSink(store, budget)) case indexer.PSQL: conn := cfg.TxIndex.PsqlConn diff --git a/sei-tendermint/internal/state/indexer/tx/kv/kv.go b/sei-tendermint/internal/state/indexer/tx/kv/kv.go index f22ef69d72..d8029f10cc 100644 --- a/sei-tendermint/internal/state/indexer/tx/kv/kv.go +++ b/sei-tendermint/internal/state/indexer/tx/kv/kv.go @@ -1,10 +1,13 @@ package kv import ( + "bytes" "context" "encoding/hex" + "errors" "fmt" "regexp" + "sort" "strconv" "strings" @@ -27,6 +30,10 @@ var _ indexer.TxIndexer = (*TxIndex)(nil) // 2. event - txhash (secondary key) type TxIndex struct { store dbm.DB + // budget bounds the index entries in-flight searches may visit; nil means + // unlimited. It is typically shared with the block indexer so the cap is + // process-wide across tx_search and block_search. + budget *indexer.ScanBudget } // NewTxIndex creates new KV indexer. @@ -36,6 +43,13 @@ func NewTxIndex(store dbm.DB) *TxIndex { } } +// WithScanBudget attaches a shared scan budget that bounds how many index +// entries in-flight searches may visit. It returns the receiver for chaining. +func (txi *TxIndex) WithScanBudget(budget *indexer.ScanBudget) *TxIndex { + txi.budget = budget + return txi +} + // Get gets transaction from the TxIndex storage and returns it or nil if the // transaction is not found. func (txi *TxIndex) Get(hash []byte) (*abci.TxResultV2, error) { @@ -134,22 +148,24 @@ func (txi *TxIndex) indexEvents(result *abci.TxResultV2, hash []byte, store dbm. // condition, it queries the DB index. One special use cases here: (1) if // "tx.hash" is found, it returns tx result for it (2) for range queries it is // better for the client to provide both lower and upper bounds, so we are not -// performing a full scan. Results from querying indexes are then intersected -// and returned to the caller, in no particular order. +// performing a full scan. +// +// The entries the scan visits are charged against the +// shared scan budget (if configured); if the aggregate in-flight charge exceeds +// the budget, Search aborts with indexer.ErrScanBudgetExceeded. // // Search will exit early and return any result fetched so far, // when a message is received on the context chan. -// TODO(PLT-748): push opts.Limit/OrderDesc down into the scan path here func (txi *TxIndex) Search(ctx context.Context, q *query.Query, opts indexer.SearchOptions) ([]*abci.TxResultV2, error) { - select { - case <-ctx.Done(): + if ctx.Err() != nil { return make([]*abci.TxResultV2, 0), nil - - default: } - var hashesInitialized bool - filteredHashes := make(map[string][]byte) + // Charge the entries this search visits against the shared scan budget so a + // broad query (or many concurrent ones) cannot exhaust memory. Release + // returns the charge to the pool when the search returns. + lease := txi.budget.Lease() + defer lease.Release() // get a list of conditions (like "tx.height > 5") conditions := q.Syntax() @@ -170,28 +186,72 @@ func (txi *TxIndex) Search(ctx context.Context, q *query.Query, opts indexer.Sea } } - // conditions to skip because they're handled before "everything else" - skipIndexes := make([]int, 0) - // extract ranges // if both upper and lower bounds exist, it's better to get them in order not // no iterate over kvs that are not within range. ranges, rangeIndexes := indexer.LookForRanges(conditions) + + // Fast path: when every condition is an equality (point-probeable) or a + // tx.height range (evaluable from the candidate height), drive a single + // (height, index)-ordered scan in order_by order, point-probe the remaining + // conditions per candidate, and stop at opts.Limit. Memory is bounded by the + // results kept, not by the total match cardinality. + if plan, ok := planBounded(conditions, ranges, rangeIndexes); ok { + return txi.searchBounded(ctx, plan, opts, lease) + } + + // Fallback: queries containing CONTAINS/MATCHES/EXISTS, non-height ranges, or + // only a tx.height range cannot be driven by an in-order point-probeable + // scan (the tx.height secondary index stores the height as a decimal string, + // so its key order is not numeric). Materialize the intersection as before, + // then bound and order the result set. + filteredHashes, err := txi.intersect(ctx, conditions, ranges, rangeIndexes, lease) + if err != nil { + return nil, err + } + return txi.collectBounded(ctx, filteredHashes, opts) +} + +// intersect returns the set of tx hashes that satisfy every condition (implicit +// AND). It seeds the set from the first condition's index matches, then +// intersects each remaining condition against it, so a tx survives only if it +// matches all of them. The returned map is keyed by tx hash string with the +// hash bytes as the value. +func (txi *TxIndex) intersect( + ctx context.Context, + conditions []syntax.Condition, + ranges indexer.QueryRanges, + rangeIndexes []int, + lease *indexer.ScanLease, +) (map[string][]byte, error) { + var hashesInitialized bool + filteredHashes := make(map[string][]byte) + + // conditions to skip because they're handled before "everything else" + skipIndexes := make([]int, 0) + if len(ranges) > 0 { skipIndexes = append(skipIndexes, rangeIndexes...) for _, qr := range ranges { + var err error if !hashesInitialized { - filteredHashes = txi.matchRange(ctx, qr, prefixFromCompositeKey(qr.Key), filteredHashes, true) + filteredHashes, err = txi.matchRange(ctx, qr, prefixFromCompositeKey(qr.Key), filteredHashes, true, lease) + if err != nil { + return nil, err + } hashesInitialized = true // Ignore any remaining conditions if the first condition resulted // in no matches (assuming implicit AND operand). if len(filteredHashes) == 0 { - break + return filteredHashes, nil } } else { - filteredHashes = txi.matchRange(ctx, qr, prefixFromCompositeKey(qr.Key), filteredHashes, false) + filteredHashes, err = txi.matchRange(ctx, qr, prefixFromCompositeKey(qr.Key), filteredHashes, false, lease) + if err != nil { + return nil, err + } } } } @@ -205,22 +265,36 @@ func (txi *TxIndex) Search(ctx context.Context, q *query.Query, opts indexer.Sea continue } + var err error if !hashesInitialized { - filteredHashes = txi.match(ctx, c, prefixForCondition(c, height), filteredHashes, true) + filteredHashes, err = txi.match(ctx, c, prefixForCondition(c, height), filteredHashes, true, lease) + if err != nil { + return nil, err + } hashesInitialized = true // Ignore any remaining conditions if the first condition resulted // in no matches (assuming implicit AND operand). if len(filteredHashes) == 0 { - break + return filteredHashes, nil } } else { - filteredHashes = txi.match(ctx, c, prefixForCondition(c, height), filteredHashes, false) + filteredHashes, err = txi.match(ctx, c, prefixForCondition(c, height), filteredHashes, false, lease) + if err != nil { + return nil, err + } } } + return filteredHashes, nil +} + +// collectBounded materializes filteredHashes into tx results, orders them by +// (height, index) per opts.OrderDesc and truncates to opts.Limit. The +// intermediate match set is still fully materialized by intersect; only the +// returned slice and the sort cost are bounded here. +func (txi *TxIndex) collectBounded(ctx context.Context, filteredHashes map[string][]byte, opts indexer.SearchOptions) ([]*abci.TxResultV2, error) { results := make([]*abci.TxResultV2, 0, len(filteredHashes)) -hashes: for _, h := range filteredHashes { res, err := txi.Get(h) if err != nil { @@ -231,16 +305,228 @@ hashes: } // Potentially exit early. - select { - case <-ctx.Done(): - break hashes - default: + if ctx.Err() != nil { + break } } + sortResults(results, opts.OrderDesc) + + if opts.Limit > 0 && len(results) > opts.Limit { + clear(results[opts.Limit:]) + results = results[:opts.Limit] + } + return results, nil } +// searchBounded executes a boundedPlan: it scans the driver equality's +// secondary-index prefix (which orders by (height, index)) in order_by order, +// point-probes the remaining conditions per candidate, and stops as soon as +// opts.Limit matches are collected. Memory is bounded by the number of results +// kept rather than by the full match cardinality. +func (txi *TxIndex) searchBounded(ctx context.Context, plan boundedPlan, opts indexer.SearchOptions, lease *indexer.ScanLease) ([]*abci.TxResultV2, error) { + prefix := prefixFromCompositeKeyAndValue(plan.driverEquality.Tag, plan.driverEquality.Arg.Value()) + + it, err := txi.prefixIterator(prefix, opts.OrderDesc) + if err != nil { + return nil, fmt.Errorf("failed to create driver iterator: %w", err) + } + defer func() { _ = it.Close() }() + + results := make([]*abci.TxResultV2, 0, indexer.BoundedCap(opts.Limit)) + seen := make(map[string]struct{}) + + for ; it.Valid(); it.Next() { + if ctx.Err() != nil { + break + } + + // Charge each driver entry the scan walks against the shared budget. + if err := lease.Visit(1); err != nil { + return nil, err + } + + hash := it.Value() + if _, dup := seen[string(hash)]; dup { + continue + } + + // Keys under the driver (tag, value) prefix are always well-formed + // secondary keys, so a parse error is unreachable; skip defensively. + height, index, err := parseHeightIndexFromKey(it.Key()) + if err != nil { + continue + } + + match, err := txi.candidateMatches(height, index, hash, plan, lease) + if err != nil { + return nil, err + } + if !match { + continue + } + + res, err := txi.Get(hash) + if err != nil { + return nil, fmt.Errorf("failed to get Tx{%X}: %w", hash, err) + } + if res == nil { + continue + } + + seen[string(hash)] = struct{}{} + results = append(results, res) + + if opts.Limit > 0 && len(results) >= opts.Limit { + break + } + } + + if err := it.Error(); err != nil { + return nil, err + } + + return results, nil +} + +// candidateMatches reports whether the tx at (height, index) satisfies every +// non-driver condition in the plan, verifying probes against hash. +// non-driver condition in the plan: tx.height range bounds are evaluated +// directly from height, and equality probes are tested with a single point +// lookup against the event index. Each probe is charged against lease. +func (txi *TxIndex) candidateMatches(height int64, index uint32, hash []byte, plan boundedPlan, lease *indexer.ScanLease) (bool, error) { + for i := range plan.heightRanges { + if !indexer.HeightInRange(height, plan.heightRanges[i]) { + return false, nil + } + } + + for i := range plan.equalityProbes { + c := plan.equalityProbes[i] + if err := lease.Visit(1); err != nil { + return false, err + } + got, err := txi.store.Get(secondaryKey(c.Tag, c.Arg.Value(), height, index)) + if err != nil { + return false, err + } + if !bytes.Equal(got, hash) { + return false, nil + } + } + + return true, nil +} + +// prefixIterator returns an iterator over the given key prefix. When desc is +// true it iterates in descending ((height, index) most-recent-first) order. +func (txi *TxIndex) prefixIterator(prefix []byte, desc bool) (dbm.Iterator, error) { + if !desc { + return dbm.IteratePrefix(txi.store, prefix) + } + return txi.store.ReverseIterator(prefix, indexer.PrefixUpperBound(prefix)) +} + +// boundedPlan describes a fast-path execution that bounds memory by driving a +// single (height, index)-ordered scan off one equality condition and +// point-probing the remaining conditions, rather than materializing and sorting +// the full match set. +type boundedPlan struct { + // driverEquality is the equality condition whose secondary-index prefix + // (tag, value) is scanned in (height, index) order. + driverEquality *syntax.Condition + // equalityProbes are the remaining equality conditions, tested per candidate + // with a point lookup. + equalityProbes []syntax.Condition + // heightRanges are tx.height bounds evaluated directly from a candidate + // height. + heightRanges []indexer.QueryRange +} + +// planBounded decides whether a query is eligible for the bounded fast path and +// builds its plan. A query qualifies only when every condition is either an +// equality (point-probeable) or a tx.height range (evaluable from the candidate +// height), and there is at least one equality to drive an ordered scan. +// +// A tx.height-range-only query does not qualify: the tx.height secondary index +// stores the height as a decimal string, so its key order is not numeric and it +// cannot drive an in-order scan. +func planBounded(conditions []syntax.Condition, ranges indexer.QueryRanges, rangeIndexes []int) (boundedPlan, bool) { + var plan boundedPlan + + // Every range must be a numeric tx.height range; any other range needs the + // attribute's value, which cannot be derived from the height alone. + for key, qr := range ranges { + if key != types.TxHeightKey { + return boundedPlan{}, false + } + if _, ok := qr.AnyBound().(int64); !ok { + return boundedPlan{}, false + } + plan.heightRanges = append(plan.heightRanges, qr) + } + + // Every non-range condition must be an equality to be point-probeable. + var equalities []syntax.Condition + for i, c := range conditions { + if intInSlice(i, rangeIndexes) { + continue + } + if c.Op != syntax.TEq { + return boundedPlan{}, false + } + equalities = append(equalities, c) + } + + // Need at least one equality to drive an ordered scan. + if len(equalities) == 0 { + return boundedPlan{}, false + } + + // Prefer a tx.height equality when one is present: + // its (TxHeightKey, "N") prefix is partitioned to a single height, so the + // scan visits only the txs in block N. + driver := 0 + for i := range equalities { + if equalities[i].Tag == types.TxHeightKey { + driver = i + break + } + } + plan.driverEquality = &equalities[driver] + plan.equalityProbes = make([]syntax.Condition, 0, len(equalities)-1) + for i := range equalities { + if i == driver { + continue + } + plan.equalityProbes = append(plan.equalityProbes, equalities[i]) + } + + return plan, true +} + +// sortResults orders tx results by (height, index): descending (most recent +// first) when desc is true, ascending otherwise. It matches the ordering the +// RPC layer applies after the search. +func sortResults(results []*abci.TxResultV2, desc bool) { + if desc { + 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 + }) + } else { + 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 + }) + } +} + func lookForHash(conditions []syntax.Condition) (hash []byte, ok bool, err error) { for _, c := range conditions { if c.Tag == types.TxHashKey { @@ -272,11 +558,12 @@ func (txi *TxIndex) match( startKeyBz []byte, filteredHashes map[string][]byte, firstRun bool, -) map[string][]byte { + lease *indexer.ScanLease, +) (map[string][]byte, error) { // A previous match was attempted but resulted in no matches, so we return // no matches (assuming AND operand). if !firstRun && len(filteredHashes) == 0 { - return filteredHashes + return filteredHashes, nil } tmpHashes := make(map[string][]byte) @@ -285,12 +572,16 @@ func (txi *TxIndex) match( case syntax.TEq: it, err := dbm.IteratePrefix(txi.store, startKeyBz) if err != nil { - panic(err) + return nil, err } defer func() { _ = it.Close() }() iterEqual: for ; it.Valid(); it.Next() { + if err := lease.Visit(1); err != nil { + return nil, err + } + tmpHashes[string(it.Value())] = it.Value() // Potentially exit early. @@ -301,7 +592,7 @@ func (txi *TxIndex) match( } } if err := it.Error(); err != nil { - panic(err) + return nil, err } case syntax.TExists: @@ -309,12 +600,16 @@ func (txi *TxIndex) match( // (e.g. "account.owner//" won't match w/ a single row) it, err := dbm.IteratePrefix(txi.store, prefixFromCompositeKey(c.Tag)) if err != nil { - panic(err) + return nil, err } defer func() { _ = it.Close() }() iterExists: for ; it.Valid(); it.Next() { + if err := lease.Visit(1); err != nil { + return nil, err + } + tmpHashes[string(it.Value())] = it.Value() // Potentially exit early. @@ -325,7 +620,7 @@ func (txi *TxIndex) match( } } if err := it.Error(); err != nil { - panic(err) + return nil, err } case syntax.TContains: @@ -334,12 +629,16 @@ func (txi *TxIndex) match( // we can't iterate with prefix "account.owner/an/" because we might miss keys like "account.owner/Ulan/" it, err := dbm.IteratePrefix(txi.store, prefixFromCompositeKey(c.Tag)) if err != nil { - panic(err) + return nil, err } defer func() { _ = it.Close() }() iterContains: for ; it.Valid(); it.Next() { + if err := lease.Visit(1); err != nil { + return nil, err + } + value, err := parseValueFromKey(it.Key()) if err != nil { continue @@ -356,18 +655,22 @@ func (txi *TxIndex) match( } } if err := it.Error(); err != nil { - panic(err) + return nil, err } case syntax.TMatches: it, err := dbm.IteratePrefix(txi.store, prefixFromCompositeKey(c.Tag)) if err != nil { - panic(err) + return nil, err } defer func() { _ = it.Close() }() iterMatches: for ; it.Valid(); it.Next() { + if err := lease.Visit(1); err != nil { + return nil, err + } + value, err := parseValueFromKey(it.Key()) if err != nil { continue @@ -384,10 +687,10 @@ func (txi *TxIndex) match( } } if err := it.Error(); err != nil { - panic(err) + return nil, err } default: - panic("other operators should be handled already") + return nil, errors.New("other operators should be handled already") } if len(tmpHashes) == 0 || firstRun { @@ -398,7 +701,7 @@ func (txi *TxIndex) match( // return no matches (assuming AND operand). // // 2. A previous match was not attempted, so we return all results. - return tmpHashes + return tmpHashes, nil } // Remove/reduce matches in filteredHashes that were not found in this @@ -416,7 +719,7 @@ func (txi *TxIndex) match( } } - return filteredHashes + return filteredHashes, nil } // matchRange returns all matching txs by hash that meet a given queryRange and @@ -430,11 +733,12 @@ func (txi *TxIndex) matchRange( startKey []byte, filteredHashes map[string][]byte, firstRun bool, -) map[string][]byte { + lease *indexer.ScanLease, +) (map[string][]byte, error) { // A previous match was attempted but resulted in no matches, so we return // no matches (assuming AND operand). if !firstRun && len(filteredHashes) == 0 { - return filteredHashes + return filteredHashes, nil } tmpHashes := make(map[string][]byte) @@ -443,12 +747,16 @@ func (txi *TxIndex) matchRange( it, err := dbm.IteratePrefix(txi.store, startKey) if err != nil { - panic(err) + return nil, err } defer func() { _ = it.Close() }() iter: for ; it.Valid(); it.Next() { + if err := lease.Visit(1); err != nil { + return nil, err + } + value, err := parseValueFromKey(it.Key()) if err != nil { continue @@ -488,7 +796,7 @@ iter: } } if err := it.Error(); err != nil { - panic(err) + return nil, err } if len(tmpHashes) == 0 || firstRun { @@ -499,7 +807,7 @@ iter: // return no matches (assuming AND operand). // // 2. A previous match was not attempted, so we return all results. - return tmpHashes + return tmpHashes, nil } // Remove/reduce matches in filteredHashes that were not found in this @@ -517,7 +825,7 @@ iter: } } - return filteredHashes + return filteredHashes, nil } // ########################## Keys ############################# diff --git a/sei-tendermint/internal/state/indexer/tx/kv/kv_test.go b/sei-tendermint/internal/state/indexer/tx/kv/kv_test.go index 509e3d663e..57fc49995b 100644 --- a/sei-tendermint/internal/state/indexer/tx/kv/kv_test.go +++ b/sei-tendermint/internal/state/indexer/tx/kv/kv_test.go @@ -324,6 +324,244 @@ func TestTxSearchMultipleTxs(t *testing.T) { require.Len(t, results, 3) } +func TestTxSearchBounded(t *testing.T) { + store := dbm.NewMemDB() + idx := NewTxIndex(store) + + // Two txs per height for heights 1..5. Every tx carries app.name=sei; the + // index-0 tx of each height also carries app.kind=even. Tx bytes are unique + // per (height, index) so each has a distinct hash. + for h := int64(1); h <= 5; h++ { + for i := uint32(0); i < 2; i++ { + events := []abci.Event{{ + Type: "app", + Attributes: []abci.EventAttribute{{Key: []byte("name"), Value: []byte("sei"), Index: true}}, + }} + if i == 0 { + events = append(events, abci.Event{ + Type: "app", + Attributes: []abci.EventAttribute{{Key: []byte("kind"), Value: []byte("even"), Index: true}}, + }) + } + res := &abci.TxResultV2{ + Height: h, + Index: i, + Tx: types.Tx(fmt.Sprintf("tx-%d-%d", h, i)), + Result: abci.ExecTxResult{Code: abci.CodeTypeOK, Events: events}, + } + require.NoError(t, idx.Index([]*abci.TxResultV2{res})) + } + } + + // hi is a (height, index) pair used to assert result identity/order. + type hi struct { + h int64 + i uint32 + } + pairs := func(results []*abci.TxResultV2) []hi { + out := make([]hi, len(results)) + for k, r := range results { + out[k] = hi{r.Height, r.Index} + } + return out + } + + testCases := map[string]struct { + q string + opts indexer.SearchOptions + want []hi + }{ + // Fast path: single equality driver, scanned in order_by order and capped + // at the scan rather than after a full sort. + "equality desc limit 3": { + q: `app.name = 'sei'`, + opts: indexer.SearchOptions{Limit: 3, OrderDesc: true}, + want: []hi{{5, 1}, {5, 0}, {4, 1}}, + }, + "equality asc limit 3": { + q: `app.name = 'sei'`, + opts: indexer.SearchOptions{Limit: 3, OrderDesc: false}, + want: []hi{{1, 0}, {1, 1}, {2, 0}}, + }, + // A disabled cap (Limit <= 0) returns the full set in order_by order. + "equality desc unbounded": { + q: `app.name = 'sei'`, + opts: indexer.SearchOptions{Limit: 0, OrderDesc: true}, + want: []hi{{5, 1}, {5, 0}, {4, 1}, {4, 0}, {3, 1}, {3, 0}, {2, 1}, {2, 0}, {1, 1}, {1, 0}}, + }, + // Fast path: two equalities — driver plus a point-probe. app.kind=even + // only matches the index-0 tx of each height. + "two equalities desc limit 2": { + q: `app.name = 'sei' AND app.kind = 'even'`, + opts: indexer.SearchOptions{Limit: 2, OrderDesc: true}, + want: []hi{{5, 0}, {4, 0}}, + }, + // Fast path: equality driver with a tx.height range filter. + "equality and height range desc limit 2": { + q: `app.name = 'sei' AND tx.height >= 4`, + opts: indexer.SearchOptions{Limit: 2, OrderDesc: true}, + want: []hi{{5, 1}, {5, 0}}, + }, + // Fast path: a tx.height equality drives its own (height, index)-ordered + // prefix. + "tx.height equality desc": { + q: `tx.height = 3`, + opts: indexer.SearchOptions{Limit: 5, OrderDesc: true}, + want: []hi{{3, 1}, {3, 0}}, + }, + // Fallback path: a tx.height-range-only query has no equality to drive an + // in-order scan (the height is stored as a decimal string), so it is + // materialized, then ordered and capped. + "height range only desc limit 3": { + q: `tx.height >= 4`, + opts: indexer.SearchOptions{Limit: 3, OrderDesc: true}, + want: []hi{{5, 1}, {5, 0}, {4, 1}}, + }, + // Fallback path: CONTAINS cannot be point-probed, but the result set is + // still ordered and capped. + "contains fallback desc limit 3": { + q: `app.name CONTAINS 'se'`, + opts: indexer.SearchOptions{Limit: 3, OrderDesc: true}, + want: []hi{{5, 1}, {5, 0}, {4, 1}}, + }, + "contains fallback asc limit 2": { + q: `app.name CONTAINS 'se'`, + opts: indexer.SearchOptions{Limit: 2, OrderDesc: false}, + want: []hi{{1, 0}, {1, 1}}, + }, + } + + for name, tc := range testCases { + t.Run(name, func(t *testing.T) { + results, err := idx.Search(t.Context(), query.MustCompile(tc.q), tc.opts) + require.NoError(t, err) + require.Equal(t, tc.want, pairs(results)) + }) + } + + // A context cancelled before the search even starts returns an empty result without an error. + t.Run("cancelled context before scan returns empty", func(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + cancel() + results, err := idx.Search(ctx, query.MustCompile(`app.name = 'sei'`), indexer.SearchOptions{Limit: 5, OrderDesc: true}) + require.NoError(t, err) + require.Empty(t, results) + }) + + // A context cancelled mid-scan makes the bounded scan stop and return the + // rows gathered so far — an ordered prefix of the full result — without an error + t.Run("cancelled context mid-scan returns ordered prefix", func(t *testing.T) { + full, err := idx.Search(t.Context(), query.MustCompile(`app.name = 'sei'`), indexer.SearchOptions{OrderDesc: true}) + require.NoError(t, err) + require.Greater(t, len(full), 2, "need several matches so a partial return is observable") + + ctx, cancel := context.WithCancel(t.Context()) + // Cancel while fetching the second matching row: searchBounded appends + // that row, then observes cancellation at the top of the next iteration + // and stops. The result is a non-empty, in-order prefix of full. + scoped := &cancelAfterGetDB{DB: store, cancel: cancel, after: 2} + results, err := NewTxIndex(scoped).Search(ctx, query.MustCompile(`app.name = 'sei'`), indexer.SearchOptions{Limit: 5, OrderDesc: true}) + require.NoError(t, err) + require.NotEmpty(t, results) + require.Less(t, len(results), len(full), "cancellation must truncate the scan") + require.Equal(t, pairs(full[:len(results)]), pairs(results)) + }) + + // Capping in the scan must not diverge from materialize-then-cap: for any + // query the bounded top-N equals the first N of the same query run + // unbounded. This covers both the fast path (equalities/height ranges) and + // the fallback (CONTAINS), guarding against future divergence between them. + t.Run("bounded cap matches unbounded prefix", func(t *testing.T) { + queries := []string{ + `app.name = 'sei'`, // fast path: equality driver + `app.name = 'sei' AND app.kind = 'even'`, // fast path: equality + probe + `app.name = 'sei' AND tx.height >= 3`, // fast path: equality + height range + `tx.height >= 3`, // fallback: height-range-only + `app.name CONTAINS 'se'`, // fallback: materialize then bound + } + for _, q := range queries { + for _, desc := range []bool{true, false} { + full, err := idx.Search(t.Context(), query.MustCompile(q), indexer.SearchOptions{Limit: 0, OrderDesc: desc}) + require.NoError(t, err) + for n := 1; n <= len(full); n++ { + capped, err := idx.Search(t.Context(), query.MustCompile(q), indexer.SearchOptions{Limit: n, OrderDesc: desc}) + require.NoError(t, err) + require.Equalf(t, pairs(full[:n]), pairs(capped), "query %q desc=%v limit=%d", q, desc, n) + } + } + } + }) +} + +// countingDB wraps a dbm.DB and counts point lookups (Get) so a test can +// assert that a bounded scan point-probes only the candidates in its driver +// prefix rather than every row in the full match set. +type countingDB struct { + dbm.DB + getCount int +} + +func (c *countingDB) Get(key []byte) ([]byte, error) { + c.getCount++ + return c.DB.Get(key) +} + +// cancelAfterGetDB wraps a dbm.DB and cancels a context on the after-th primary +// Get. searchBounded calls Get once per matched row, so this stops a bounded +// scan after a known number of rows are gathered, exercising the in-loop +// partial-return path rather than the top-level cancellation guard. +type cancelAfterGetDB struct { + dbm.DB + cancel context.CancelFunc + after int + gets int +} + +func (c *cancelAfterGetDB) Get(key []byte) ([]byte, error) { + c.gets++ + if c.gets == c.after { + c.cancel() + } + return c.DB.Get(key) +} + +func TestTxSearchBoundedPrefersHeightDriver(t *testing.T) { + store := &countingDB{DB: dbm.NewMemDB()} + idx := NewTxIndex(store) + + // One tx per height for heights 1..20, each carrying sender.addr=addr1. + // Driving off sender scans all 20 heights; driving off tx.height=N scans + // only block N. + const heights = 20 + for h := int64(1); h <= heights; h++ { + res := &abci.TxResultV2{ + Height: h, + Index: 0, + Tx: types.Tx(fmt.Sprintf("tx-%d", h)), + Result: abci.ExecTxResult{Code: abci.CodeTypeOK, Events: []abci.Event{{ + Type: "sender", + Attributes: []abci.EventAttribute{{Key: []byte("addr"), Value: []byte("addr1"), Index: true}}, + }}}, + } + require.NoError(t, idx.Index([]*abci.TxResultV2{res})) + } + + store.getCount = 0 + results, err := idx.Search( + t.Context(), + query.MustCompile(`sender.addr = 'addr1' AND tx.height = 7`), + indexer.SearchOptions{Limit: 10, OrderDesc: true}, + ) + require.NoError(t, err) + + // Correctness: only the height-7 tx matches. + require.Len(t, results, 1) + require.Equal(t, int64(7), results[0].Height) + // One Get to probe sender.addr for the single block-7 candidate, plus one + // Get to fetch the matched tx by hash. + require.Equal(t, 2, store.getCount, "bounded scan issued %d point lookups; must probe only block-7 candidates, not every sender.addr row", store.getCount) +} + func txResultWithEvents(events []abci.Event) *abci.TxResultV2 { tx := types.Tx("HELLO WORLD") return &abci.TxResultV2{ diff --git a/sei-tendermint/internal/state/indexer/tx/kv/utils.go b/sei-tendermint/internal/state/indexer/tx/kv/utils.go index 48362bfbc2..7436ad1224 100644 --- a/sei-tendermint/internal/state/indexer/tx/kv/utils.go +++ b/sei-tendermint/internal/state/indexer/tx/kv/utils.go @@ -1,6 +1,12 @@ package kv -// IntInSlice returns true if a is found in the list. +import ( + "fmt" + + "github.com/google/orderedcode" +) + +// intInSlice returns true if a is found in the list. func intInSlice(a int, list []int) bool { for _, b := range list { if b == a { @@ -9,3 +15,23 @@ func intInSlice(a int, list []int) bool { } return false } + +// parseHeightIndexFromKey extracts the (height, index) suffix of a secondary +// (event) key. The tx index encodes each event key as +// orderedcode(compositeKey, value, height, int64(index)). +func parseHeightIndexFromKey(key []byte) (int64, uint32, error) { + var ( + compositeKey, value string + height, index int64 + ) + + remaining, err := orderedcode.Parse(string(key), &compositeKey, &value, &height, &index) + if err != nil { + return 0, 0, fmt.Errorf("failed to parse event key: %w", err) + } + if len(remaining) != 0 { + return 0, 0, fmt.Errorf("unexpected remainder in key: %s", remaining) + } + + return height, uint32(index), nil //nolint:gosec // index is stored as int64(uint32) by secondaryKey, so the round-trip back to uint32 cannot overflow +} diff --git a/sei-tendermint/internal/state/indexer/utils.go b/sei-tendermint/internal/state/indexer/utils.go new file mode 100644 index 0000000000..39b6a966e6 --- /dev/null +++ b/sei-tendermint/internal/state/indexer/utils.go @@ -0,0 +1,50 @@ +package indexer + +// maxBoundedPrealloc caps how much a bounded search fast path preallocates for +// its result slice, so a very large (or disabled) limit does not eagerly +// allocate. +const maxBoundedPrealloc = 4096 + +// BoundedCap returns a sensible initial capacity for a result slice given a +// limit. A non-positive limit means "unbounded", in which case we let the +// slice grow on demand. +func BoundedCap(limit int) int { + if limit <= 0 { + return 0 + } + if limit < maxBoundedPrealloc { + return limit + } + return maxBoundedPrealloc +} + +// HeightInRange reports whether height h falls within the (already +// inclusivity-adjusted) bounds of a numeric height query range. +func HeightInRange(h int64, qr QueryRange) bool { + if lower := qr.LowerBoundValue(); lower != nil { + if lb, ok := lower.(int64); ok && h < lb { + return false + } + } + if upper := qr.UpperBoundValue(); upper != nil { + if ub, ok := upper.(int64); ok && h > ub { + return false + } + } + return true +} + +// PrefixUpperBound returns the exclusive end key for iterating over prefix, +// i.e. the smallest key strictly greater than every key having the prefix. +// It returns nil when prefix is empty or all bytes are 0xFF (no upper bound). +func PrefixUpperBound(prefix []byte) []byte { + end := make([]byte, len(prefix)) + copy(end, prefix) + for i := len(end) - 1; i >= 0; i-- { + if end[i] != 0xFF { + end[i]++ + return end[:i+1] + } + } + return nil +}