From 84306be0549f467e60ee4dc4c607f6e9d3359a17 Mon Sep 17 00:00:00 2001 From: Amir Deris Date: Wed, 8 Jul 2026 18:34:49 -0700 Subject: [PATCH 1/8] WIP --- sei-tendermint/config/config.go | 12 + sei-tendermint/config/config_test.go | 1 + sei-tendermint/config/toml.go | 6 + sei-tendermint/internal/rpc/core/blocks.go | 1 + sei-tendermint/internal/rpc/core/tx.go | 1 + .../internal/state/indexer/block/kv/kv.go | 388 ++++++++++++- .../indexer/block/kv/kv_watermark_test.go | 251 ++++++++ .../internal/state/indexer/block/kv/util.go | 47 ++ .../internal/state/indexer/indexer.go | 15 + .../internal/state/indexer/tx/kv/kv.go | 542 ++++++++++++++++-- .../internal/state/indexer/tx/kv/kv_test.go | 8 +- .../state/indexer/tx/kv/kv_watermark_test.go | 334 +++++++++++ .../internal/state/indexer/tx/kv/utils.go | 16 + .../internal/state/indexer/utils.go | 37 ++ 14 files changed, 1604 insertions(+), 55 deletions(-) create mode 100644 sei-tendermint/internal/state/indexer/block/kv/kv_watermark_test.go create mode 100644 sei-tendermint/internal/state/indexer/tx/kv/kv_watermark_test.go diff --git a/sei-tendermint/config/config.go b/sei-tendermint/config/config.go index 633c626ce0..5a430c879a 100644 --- a/sei-tendermint/config/config.go +++ b/sei-tendermint/config/config.go @@ -536,6 +536,14 @@ 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 index entries a single tx_search / block_search may + // examine on the fallback scan path (CONTAINS/MATCHES/value ranges) before + // the query is rejected as too broad. This bounds work, not output: unlike + // max-tx-search-results (which caps the result set) it protects the node + // from a broad query that must scan millions of entries to return a handful + // of matches. 0 disables the budget (not recommended on public nodes). + MaxEventSearchScan int `mapstructure:"max-event-search-scan"` } // DefaultRPCConfig returns a default configuration for the RPC server @@ -570,6 +578,7 @@ func DefaultRPCConfig() *RPCConfig { TimeoutWrite: 30 * time.Second, MaxTxSearchResults: 10_000, + MaxEventSearchScan: 1_000_000, } } @@ -625,6 +634,9 @@ func (cfg *RPCConfig) ValidateBasic() error { if cfg.MaxTxSearchResults < 0 { return errors.New("max-tx-search-results can't be negative") } + if cfg.MaxEventSearchScan < 0 { + return errors.New("max-event-search-scan can't be negative") + } return nil } diff --git a/sei-tendermint/config/config_test.go b/sei-tendermint/config/config_test.go index 811ba5fb55..01241e7ec9 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", + "MaxEventSearchScan", } for _, fieldName := range fieldsToTest { diff --git a/sei-tendermint/config/toml.go b/sei-tendermint/config/toml.go index 7802db8897..b1a65e9e0b 100644 --- a/sei-tendermint/config/toml.go +++ b/sei-tendermint/config/toml.go @@ -287,6 +287,12 @@ 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 index entries a single tx_search / block_search may examine +# on the fallback scan path (CONTAINS/MATCHES/value ranges) before the query is +# rejected as too broad. Bounds work rather than output. Set to 0 to disable +# (not recommended on public nodes). +max-event-search-scan = {{ .RPC.MaxEventSearchScan }} + ####################################################################### ### P2P Configuration Options ### ####################################################################### diff --git a/sei-tendermint/internal/rpc/core/blocks.go b/sei-tendermint/internal/rpc/core/blocks.go index 902a9f6863..afc660bc06 100644 --- a/sei-tendermint/internal/rpc/core/blocks.go +++ b/sei-tendermint/internal/rpc/core/blocks.go @@ -340,6 +340,7 @@ func (env *Environment) BlockSearch(ctx context.Context, req *coretypes.RequestB results, err := kvsink.SearchBlockEvents(ctx, q, indexer.SearchOptions{ Limit: env.Config.MaxTxSearchResults, OrderDesc: orderDesc, + MaxScan: env.Config.MaxEventSearchScan, }) if err != nil { return nil, err diff --git a/sei-tendermint/internal/rpc/core/tx.go b/sei-tendermint/internal/rpc/core/tx.go index e0da1e302c..6b27dfa895 100644 --- a/sei-tendermint/internal/rpc/core/tx.go +++ b/sei-tendermint/internal/rpc/core/tx.go @@ -85,6 +85,7 @@ func (env *Environment) TxSearch(ctx context.Context, req *coretypes.RequestTxSe results, err := sink.SearchTxEvents(ctx, q, indexer.SearchOptions{ Limit: env.Config.MaxTxSearchResults, OrderDesc: orderDesc, + MaxScan: env.Config.MaxEventSearchScan, }) if err != nil { return nil, err diff --git a/sei-tendermint/internal/state/indexer/block/kv/kv.go b/sei-tendermint/internal/state/indexer/block/kv/kv.go index c93730c252..af2c7b4b43 100644 --- a/sei-tendermint/internal/state/indexer/block/kv/kv.go +++ b/sei-tendermint/internal/state/indexer/block/kv/kv.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "math" "regexp" "sort" "strconv" @@ -21,6 +22,21 @@ import ( var _ indexer.BlockIndexer = (*BlockerIndexer)(nil) +const ( + // blockHeightOrderedKey namespaces the height-ordered event index. Its keys + // have the form orderedcode(blockHeightOrderedKey, tag, height, value, typ), + // so within a tag they sort by height — matching the order block_search + // returns — and stay disjoint from the legacy value-ordered index that + // shares this store. It is reserved: events may not use it as a composite + // key. block.height range queries are already served in height order by the + // primary key and do not use this index. + blockHeightOrderedKey = "block.height_ordered" + + // blockWatermarkKey stores the lowest block height covered by the + // height-ordered index on this node (see readWatermark). It is reserved. + blockWatermarkKey = "block.new_index_min_height" +) + // BlockerIndexer implements a block indexer, indexing FinalizeBlock // events with an underlying KV store. Block events are indexed by their height, // such that matching search criteria returns the respective block height(s). @@ -45,6 +61,41 @@ func (idx *BlockerIndexer) Has(height int64) (bool, error) { return idx.store.Has(key) } +// readWatermark returns the lowest block height covered by the height-ordered +// index. An unset watermark (fresh DB, or upgraded-but-not-yet-written) reads +// as math.MaxInt64 so every height-ordered query takes the legacy fallback +// until the new index has written at least one key. +func (idx *BlockerIndexer) readWatermark() int64 { + key, err := watermarkKey() + if err != nil { + panic(err) + } + bz, err := idx.store.Get(key) + if err != nil { + panic(err) + } + if len(bz) == 0 { + return math.MaxInt64 + } + return int64FromBytes(bz) +} + +// updateWatermark lowers the persisted watermark to height when height is lower, +// writing into the provided batch so it commits atomically with the +// height-ordered keys it accounts for. A watermark that is too high is only +// over-conservative (routes covered heights to the fallback); a watermark below +// the keys actually written would be incorrect, so it is never raised here. +func (idx *BlockerIndexer) updateWatermark(batch dbm.Batch, height int64) error { + if height >= idx.readWatermark() { + return nil + } + key, err := watermarkKey() + if err != nil { + return err + } + return batch.Set(key, int64ToBytes(height)) +} + // Index indexes FinalizeBlock events for a given block by its height. // The following is indexed: // @@ -70,6 +121,13 @@ func (idx *BlockerIndexer) Index(bh types.EventDataNewBlockHeader) error { return fmt.Errorf("failed to index FinalizeBlock events: %w", err) } + // 3. advance the height-ordered index watermark in the same atomic batch as + // the keys it accounts for, so a crash can never leave the watermark below + // the keys actually written. + if err := idx.updateWatermark(batch, height); err != nil { + return err + } + return batch.WriteSync() } @@ -109,16 +167,27 @@ func (idx *BlockerIndexer) Search(ctx context.Context, q *query.Query, opts inde // Fast path: when every condition can be driven by a single height-ordered // scan and point-probed, stream candidates in order_by order and stop at // the limit, so a broad query does not materialize and sort the full match - // set. + // set. This uses the legacy index / primary key (full coverage) and ignores + // the watermark. if plan, ok := planBounded(conditions, ranges, rangeIndexes); ok { return idx.searchBounded(ctx, plan, opts) } - // 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) + // Height-ordered path: an EXISTS-by-tag query has no equality to drive the + // legacy fast path, but it is height-orderable. Drive it off the new + // height-ordered index, splitting at the watermark so pre-upgrade heights + // (not covered by the new index) fall back to the legacy index for full + // coverage. Early-stops at opts.Limit. + if plan, ok := planHeightOrdered(conditions, ranges, rangeIndexes); ok { + return idx.searchHeightOrdered(ctx, plan, opts) + } + + // Fallback: queries containing CONTAINS/MATCHES or non-height value ranges + // cannot be driven by an in-order scan. Materialize the intersection as + // before, then bound and order the result set. The scan is bounded by + // opts.MaxScan and fails closed if the budget is exceeded. + budget := indexer.NewScanBudget(opts.MaxScan) + filteredHeights, err := idx.intersect(ctx, conditions, ranges, rangeIndexes, budget) if err != nil { return nil, err } @@ -135,6 +204,7 @@ func (idx *BlockerIndexer) intersect( conditions []syntax.Condition, ranges indexer.QueryRanges, rangeIndexes []int, + budget *indexer.ScanBudget, ) (map[string][]byte, error) { var heightsInitialized bool filteredHeights := make(map[string][]byte) @@ -152,7 +222,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, budget) if err != nil { return nil, err } @@ -165,7 +235,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, budget) if err != nil { return nil, err } @@ -185,7 +255,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, budget) if err != nil { return nil, err } @@ -198,7 +268,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, budget) if err != nil { return nil, err } @@ -243,6 +313,44 @@ func (idx *BlockerIndexer) collectBounded(ctx context.Context, filteredHeights m return results, nil } +// collectBoundedInHeightRange is collectBounded restricted to heights within +// [lo, hi] and capped at limit. It backs the sub-watermark leg of the +// height-ordered split, where the legacy fallback must serve only pre-watermark +// heights. +func (idx *BlockerIndexer) collectBoundedInHeightRange(ctx context.Context, filteredHeights map[string][]byte, orderDesc bool, limit int, lo, hi int64) ([]int64, error) { + results := make([]int64, 0, indexer.BoundedCap(limit)) + for _, hBz := range filteredHeights { + h := int64FromBytes(hBz) + if h < lo || h > hi { + continue + } + + ok, err := idx.Has(h) + if err != nil { + return nil, err + } + if ok { + results = append(results, h) + } + + if ctx.Err() != nil { + break + } + } + + if orderDesc { + sort.Slice(results, func(i, j int) bool { return results[i] > results[j] }) + } else { + sort.Slice(results, func(i, j int) bool { return results[i] < results[j] }) + } + + if limit > 0 && len(results) > limit { + results = results[:limit] + } + + 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 @@ -424,6 +532,234 @@ func planBounded(conditions []syntax.Condition, ranges indexer.QueryRanges, rang return plan, true } +// heightOrderedPlan describes a query served from the height-ordered index: a +// single height-ordered scan of driverTag's prefix, filtered by any block.height +// bounds and split at the watermark (heights >= W come from the new index; +// heights < W come from the legacy fallback for full coverage). +type heightOrderedPlan struct { + // driverTag is the composite key whose height-ordered prefix is scanned. + driverTag string + // heightRanges are block.height bounds; may be empty for a pure EXISTS query. + heightRanges []indexer.QueryRange + // existsCond is the single EXISTS condition this plan serves, retained so + // the sub-watermark leg can rebuild the equivalent legacy match. + existsCond *syntax.Condition +} + +// planHeightOrdered decides whether a query is eligible for the height-ordered +// path and builds its plan. It handles a single EXISTS on a tag (optionally +// combined with a block.height range) — the height-orderable shape that has no +// equality to drive the legacy fast path and no primary-key coverage. A +// block.height range-only query is already served in height order by the +// primary key (see planBounded) and is not routed here. +func planHeightOrdered(conditions []syntax.Condition, ranges indexer.QueryRanges, rangeIndexes []int) (heightOrderedPlan, bool) { + var plan heightOrderedPlan + + // Every range must be a numeric block.height range. + for key, qr := range ranges { + if key != types.BlockHeightKey { + return heightOrderedPlan{}, false + } + if _, ok := qr.AnyBound().(int64); !ok { + return heightOrderedPlan{}, false + } + plan.heightRanges = append(plan.heightRanges, qr) + } + + // Collect the non-range conditions; exactly one EXISTS is required. + nonRange := make([]syntax.Condition, 0, len(conditions)) + for i, c := range conditions { + if intInSlice(i, rangeIndexes) { + continue + } + nonRange = append(nonRange, c) + } + if len(nonRange) != 1 || nonRange[0].Op != syntax.TExists { + return heightOrderedPlan{}, false + } + + plan.driverTag = nonRange[0].Tag + plan.existsCond = &nonRange[0] + return plan, true +} + +// heightBounds reduces the plan's block.height ranges to a single inclusive +// [lo, hi] window. Heights are >= 1, so an absent lower bound defaults to 1 +// (avoiding a pointless legacy scan for a non-existent sub-watermark tail). +func heightBounds(ranges []indexer.QueryRange) (lo, hi int64) { + lo, hi = 1, int64(math.MaxInt64) + for i := range ranges { + if lb := ranges[i].LowerBoundValue(); lb != nil { + if v, ok := lb.(int64); ok && v > lo { + lo = v + } + } + if ub := ranges[i].UpperBoundValue(); ub != nil { + if v, ok := ub.(int64); ok && v < hi { + hi = v + } + } + } + return lo, hi +} + +// searchHeightOrdered serves a heightOrderedPlan by splitting the [lo, hi] +// window at the watermark W: heights in [max(lo, W), hi] are streamed from the +// new height-ordered index (fast, early-stops at opts.Limit); heights in +// [lo, W-1] are served by the legacy materializing fallback for full coverage. +// The two sub-ranges are height-disjoint at W, so no global merge is needed. +func (idx *BlockerIndexer) searchHeightOrdered(ctx context.Context, plan heightOrderedPlan, opts indexer.SearchOptions) ([]int64, error) { + w := idx.readWatermark() + lo, hi := heightBounds(plan.heightRanges) + if lo > hi { + return []int64{}, nil + } + + fastLo := max(lo, w) + hasFast := fastLo <= hi + + fbHi := min(hi, w-1) + hasFallback := lo <= fbHi + + results := make([]int64, 0, indexer.BoundedCap(opts.Limit)) + remaining := func() int { + if opts.Limit <= 0 { + return -1 + } + return opts.Limit - len(results) + } + reachedLimit := func() bool { + return opts.Limit > 0 && len(results) >= opts.Limit + } + + if opts.OrderDesc { + if hasFast { + r, err := idx.scanHeightOrderedFast(ctx, plan, fastLo, hi, true, remaining()) + if err != nil { + return nil, err + } + results = append(results, r...) + } + if !reachedLimit() && hasFallback { + r, err := idx.heightOrderedFallback(ctx, plan, lo, fbHi, true, remaining(), opts.MaxScan) + if err != nil { + return nil, err + } + results = append(results, r...) + } + } else { + if hasFallback { + r, err := idx.heightOrderedFallback(ctx, plan, lo, fbHi, false, remaining(), opts.MaxScan) + if err != nil { + return nil, err + } + results = append(results, r...) + } + if !reachedLimit() && hasFast { + r, err := idx.scanHeightOrderedFast(ctx, plan, fastLo, hi, false, remaining()) + if err != nil { + return nil, err + } + results = append(results, r...) + } + } + + if opts.Limit > 0 && len(results) > opts.Limit { + results = results[:opts.Limit] + } + return results, nil +} + +// scanHeightOrderedFast streams the new height-ordered index for plan.driverTag +// over heights [lo, hi] in order_by height order, deduping heights and stopping +// at limit (limit <= 0 means unbounded). Because keys are height-major the scan +// early-stops once it passes the far bound. This is a Limit-bounded fast-path +// scan and is deliberately not charged against the scan budget. +func (idx *BlockerIndexer) scanHeightOrderedFast(ctx context.Context, plan heightOrderedPlan, lo, hi int64, desc bool, limit int) ([]int64, error) { + prefix, err := prefixHeightOrdered(plan.driverTag) + if err != nil { + return nil, err + } + it, err := idx.prefixIterator(prefix, desc) + if err != nil { + return nil, fmt.Errorf("failed to create height-ordered iterator: %w", err) + } + defer func() { _ = it.Close() }() + + results := make([]int64, 0, indexer.BoundedCap(limit)) + seen := make(map[int64]struct{}) + + for ; it.Valid(); it.Next() { + if ctx.Err() != nil { + break + } + + h, err := parseHeightFromHeightOrderedKey(it.Key()) + if err != nil { + continue + } + + // Keys are height-major, so once we pass the far bound in scan order we + // can stop; the near bound is skipped until we reach the window. + if desc { + if h > hi { + continue + } + if h < lo { + break + } + } else { + if h < lo { + continue + } + if h > hi { + break + } + } + + if _, dup := seen[h]; dup { + continue + } + + ok, err := idx.Has(h) + if err != nil { + return nil, err + } + if !ok { + continue + } + + seen[h] = struct{}{} + results = append(results, h) + + if limit > 0 && len(results) >= limit { + break + } + } + + if err := it.Error(); err != nil { + return nil, err + } + + return results, nil +} + +// heightOrderedFallback serves the pre-watermark [lo, hi] leg of a +// heightOrderedPlan from the legacy index. It rebuilds the equivalent legacy +// EXISTS match, then collects heights in [lo, hi], ordered and capped at limit. +// The legacy scan is charged against a maxScan budget and fails closed if the +// budget is exceeded. +func (idx *BlockerIndexer) heightOrderedFallback(ctx context.Context, plan heightOrderedPlan, lo, hi int64, desc bool, limit, maxScan int) ([]int64, error) { + budget := indexer.NewScanBudget(maxScan) + + filtered, err := idx.match(ctx, *plan.existsCond, nil, map[string][]byte{}, true, budget) + if err != nil { + return nil, err + } + + return idx.collectBoundedInHeightRange(ctx, filtered, desc, limit, lo, hi) +} + // matchRange returns all matching block heights that match a given QueryRange // and start key. An already filtered result (filteredHeights) is provided such // that any non-intersecting matches are removed. @@ -436,6 +772,7 @@ func (idx *BlockerIndexer) matchRange( startKey []byte, filteredHeights map[string][]byte, firstRun bool, + budget *indexer.ScanBudget, ) (map[string][]byte, error) { // A previous match was attempted but resulted in no matches, so we return @@ -456,6 +793,10 @@ func (idx *BlockerIndexer) matchRange( iter: for ; it.Valid(); it.Next() { + if err := budget.Step(); err != nil { + return nil, err + } + var ( eventValue string err error @@ -544,6 +885,7 @@ func (idx *BlockerIndexer) match( startKeyBz []byte, filteredHeights map[string][]byte, firstRun bool, + budget *indexer.ScanBudget, ) (map[string][]byte, error) { // A previous match was attempted but resulted in no matches, so we return @@ -563,6 +905,9 @@ func (idx *BlockerIndexer) match( defer func() { _ = it.Close() }() for ; it.Valid(); it.Next() { + if err := budget.Step(); err != nil { + return nil, err + } tmpHeights[string(it.Value())] = it.Value() if err := ctx.Err(); err != nil { @@ -588,6 +933,9 @@ func (idx *BlockerIndexer) match( iterExists: for ; it.Valid(); it.Next() { + if err := budget.Step(); err != nil { + return nil, err + } tmpHeights[string(it.Value())] = it.Value() select { @@ -616,6 +964,9 @@ func (idx *BlockerIndexer) match( iterContains: for ; it.Valid(); it.Next() { + if err := budget.Step(); err != nil { + return nil, err + } eventValue, err := parseValueFromEventKey(it.Key()) if err != nil { continue @@ -650,6 +1001,9 @@ func (idx *BlockerIndexer) match( iterMatches: for ; it.Valid(); it.Next() { + if err := budget.Step(); err != nil { + return nil, err + } eventValue, err := parseValueFromEventKey(it.Key()) if err != nil { continue @@ -719,19 +1073,29 @@ func (idx *BlockerIndexer) indexEvents(batch dbm.Batch, events []abci.Event, typ // index iff the event specified index:true and it's not a reserved event compositeKey := fmt.Sprintf("%s.%s", event.Type, string(attr.Key)) - if compositeKey == types.BlockHeightKey { + if compositeKey == types.BlockHeightKey || + compositeKey == blockHeightOrderedKey || compositeKey == blockWatermarkKey { return fmt.Errorf("event type and attribute key \"%s\" is reserved; please use a different key", compositeKey) } if attr.GetIndex() { + // dual-write: legacy value-ordered key plus the new + // height-ordered key (see eventKeyHeightOrdered). key, err := eventKey(compositeKey, typ, string(attr.Value), height) if err != nil { return fmt.Errorf("failed to create block index key: %w", err) } - if err := batch.Set(key, heightBz); err != nil { return err } + + hoKey, err := eventKeyHeightOrdered(compositeKey, typ, string(attr.Value), height) + if err != nil { + return fmt.Errorf("failed to create height-ordered block index key: %w", err) + } + if err := batch.Set(hoKey, heightBz); err != nil { + return err + } } } } diff --git a/sei-tendermint/internal/state/indexer/block/kv/kv_watermark_test.go b/sei-tendermint/internal/state/indexer/block/kv/kv_watermark_test.go new file mode 100644 index 0000000000..28e1b95f9e --- /dev/null +++ b/sei-tendermint/internal/state/indexer/block/kv/kv_watermark_test.go @@ -0,0 +1,251 @@ +package kv + +import ( + "context" + "fmt" + "math" + "testing" + + "github.com/stretchr/testify/require" + dbm "github.com/tendermint/tm-db" + + abci "github.com/sei-protocol/sei-chain/sei-tendermint/abci/types" + "github.com/sei-protocol/sei-chain/sei-tendermint/internal/pubsub/query" + "github.com/sei-protocol/sei-chain/sei-tendermint/internal/state/indexer" + "github.com/sei-protocol/sei-chain/sei-tendermint/types" +) + +const ( + hoMaxHeight = 10 + hoWatermark = 6 + hoTyp = "finalize_block" +) + +// hoBlock builds a block header at the given height carrying app.name=sei. +func hoBlock(height int64) types.EventDataNewBlockHeader { + return types.EventDataNewBlockHeader{ + Header: types.Header{Height: height}, + ResultFinalizeBlock: abci.ResponseFinalizeBlock{Events: []abci.Event{{ + Type: "app", + Attributes: []abci.EventAttribute{{Key: []byte("name"), Value: []byte("sei"), Index: true}}, + }}}, + } +} + +// indexLegacyOnly writes the primary height key and the legacy (value-ordered) +// event keys for a block, but not the new height-ordered keys, and does not +// touch the watermark — simulating a pre-upgrade block uncovered by the new +// index. +func indexLegacyOnly(t *testing.T, idx *BlockerIndexer, bh types.EventDataNewBlockHeader) { + t.Helper() + b := idx.store.NewBatch() + defer func() { _ = b.Close() }() + + height := bh.Header.Height + hk, err := heightKey(height) + require.NoError(t, err) + require.NoError(t, b.Set(hk, int64ToBytes(height))) + + for _, event := range bh.ResultFinalizeBlock.Events { + for _, attr := range event.Attributes { + if !attr.GetIndex() { + continue + } + compositeKey := fmt.Sprintf("%s.%s", event.Type, string(attr.Key)) + k, err := eventKey(compositeKey, hoTyp, string(attr.Value), height) + require.NoError(t, err) + require.NoError(t, b.Set(k, int64ToBytes(height))) + } + } + require.NoError(t, b.WriteSync()) +} + +// backfillHeightOrdered writes the height-ordered keys for a previously +// legacy-only block, simulating a background backfill (without lowering W here). +func backfillHeightOrdered(t *testing.T, idx *BlockerIndexer, bh types.EventDataNewBlockHeader) { + t.Helper() + b := idx.store.NewBatch() + defer func() { _ = b.Close() }() + + height := bh.Header.Height + for _, event := range bh.ResultFinalizeBlock.Events { + for _, attr := range event.Attributes { + if !attr.GetIndex() { + continue + } + compositeKey := fmt.Sprintf("%s.%s", event.Type, string(attr.Key)) + k, err := eventKeyHeightOrdered(compositeKey, hoTyp, string(attr.Value), height) + require.NoError(t, err) + require.NoError(t, b.Set(k, int64ToBytes(height))) + } + } + require.NoError(t, b.WriteSync()) +} + +func setWatermark(t *testing.T, idx *BlockerIndexer, w int64) { + t.Helper() + key, err := watermarkKey() + require.NoError(t, err) + require.NoError(t, idx.store.Set(key, int64ToBytes(w))) +} + +// hoFixture indexes heights 6..10 with the dual-write index (watermark settles +// at 6) and splices legacy-only heights 1..5 beneath it, so a height-ordered +// EXISTS query must split: [6,10] from the new index, [1,5] from the legacy +// fallback. +func hoFixture(t *testing.T) *BlockerIndexer { + t.Helper() + idx := New(dbm.NewMemDB()) + + for h := int64(hoWatermark); h <= hoMaxHeight; h++ { + require.NoError(t, idx.Index(hoBlock(h))) + } + require.Equal(t, int64(hoWatermark), idx.readWatermark()) + + for h := int64(1); h < hoWatermark; h++ { + indexLegacyOnly(t, idx, hoBlock(h)) + } + require.Equal(t, int64(hoWatermark), idx.readWatermark()) + return idx +} + +// reference returns the heights in [lo,hi] (intersected with [1,hoMaxHeight]), +// ordered per desc and capped at limit (<=0 == all). +func reference(desc bool, lo, hi int64, limit int) []int64 { + out := []int64{} + if desc { + for h := int64(hoMaxHeight); h >= 1; h-- { + if h >= lo && h <= hi { + out = append(out, h) + } + } + } else { + for h := int64(1); h <= hoMaxHeight; h++ { + if h >= lo && h <= hi { + out = append(out, h) + } + } + } + if limit > 0 && len(out) > limit { + out = out[:limit] + } + return out +} + +// TestBlockSearchWatermarkSplit exercises the EXISTS height-ordered split-merge +// across the watermark, at boundary lower bounds {W-1, W, W+1}, both orderings +// and various limits. Block.height ranges are served by the primary key and are +// not routed here, so the split is driven by EXISTS (optionally + a height +// range). +func TestBlockSearchWatermarkSplit(t *testing.T) { + idx := hoFixture(t) + + cases := []struct { + name string + q string + lo, hi int64 + limitSet []int + }{ + {"exists full range", `app.name EXISTS`, 1, hoMaxHeight, []int{0, 1, 3, 5, 11}}, + {"exists height >= W-1 (5)", `app.name EXISTS AND block.height >= 5`, 5, hoMaxHeight, []int{0, 1, 2, 5}}, + {"exists height >= W (6)", `app.name EXISTS AND block.height >= 6`, 6, hoMaxHeight, []int{0, 1, 4}}, + {"exists height >= W+1 (7)", `app.name EXISTS AND block.height >= 7`, 7, hoMaxHeight, []int{0, 2}}, + {"exists height <= W-1 (5)", `app.name EXISTS AND block.height <= 5`, 1, 5, []int{0, 3}}, + {"exists window straddling W", `app.name EXISTS AND block.height >= 4 AND block.height <= 7`, 4, 7, []int{0, 2}}, + } + + for _, tc := range cases { + for _, desc := range []bool{true, false} { + for _, limit := range tc.limitSet { + name := fmt.Sprintf("%s/desc=%v/limit=%d", tc.name, desc, limit) + t.Run(name, func(t *testing.T) { + results, err := idx.Search(t.Context(), query.MustCompile(tc.q), + indexer.SearchOptions{Limit: limit, OrderDesc: desc, MaxScan: 0}) + require.NoError(t, err) + require.Equal(t, reference(desc, tc.lo, tc.hi, limit), results) + }) + } + } + } +} + +// TestBlockSearchWatermarkUnset verifies that an unset watermark routes an +// EXISTS query entirely to the legacy fallback and still returns the full set. +func TestBlockSearchWatermarkUnset(t *testing.T) { + idx := New(dbm.NewMemDB()) + for h := int64(1); h <= 5; h++ { + indexLegacyOnly(t, idx, hoBlock(h)) + } + require.Equal(t, int64(math.MaxInt64), idx.readWatermark()) + + results, err := idx.Search(t.Context(), query.MustCompile(`app.name EXISTS`), + indexer.SearchOptions{OrderDesc: true}) + require.NoError(t, err) + require.Equal(t, []int64{5, 4, 3, 2, 1}, results) +} + +// TestBlockSearchBackfillLowersWatermark verifies that backfilling the +// sub-watermark heights and lowering W keeps results correct with no +// double-count across the split. +func TestBlockSearchBackfillLowersWatermark(t *testing.T) { + idx := hoFixture(t) + + q := query.MustCompile(`app.name EXISTS`) + before, err := idx.Search(t.Context(), q, indexer.SearchOptions{OrderDesc: true}) + require.NoError(t, err) + require.Equal(t, reference(true, 1, hoMaxHeight, 0), before) + + for h := int64(1); h < hoWatermark; h++ { + backfillHeightOrdered(t, idx, hoBlock(h)) + } + setWatermark(t, idx, 1) + + after, err := idx.Search(t.Context(), q, indexer.SearchOptions{OrderDesc: true}) + require.NoError(t, err) + require.Equal(t, reference(true, 1, hoMaxHeight, 0), after) +} + +// TestBlockSearchScanBudgetFailClosed asserts that a fallback query exceeding +// the scan budget fails closed with ErrSearchScanBudgetExceeded (no partial), +// while a query within budget succeeds. +func TestBlockSearchScanBudgetFailClosed(t *testing.T) { + idx := New(dbm.NewMemDB()) + for h := int64(1); h <= 12; h++ { + require.NoError(t, idx.Index(hoBlock(h))) + } + + for _, desc := range []bool{true, false} { + t.Run(fmt.Sprintf("desc=%v", desc), func(t *testing.T) { + results, err := idx.Search(t.Context(), query.MustCompile(`app.name CONTAINS 'se'`), + indexer.SearchOptions{Limit: 5, OrderDesc: desc, MaxScan: 3}) + require.ErrorIs(t, err, indexer.ErrSearchScanBudgetExceeded) + require.Nil(t, results) + }) + } + + results, err := idx.Search(t.Context(), query.MustCompile(`app.name CONTAINS 'se'`), + indexer.SearchOptions{Limit: 5, OrderDesc: true, MaxScan: 0}) + require.NoError(t, err) + require.Len(t, results, 5) +} + +// TestBlockSearchBudgetVsDeadline asserts the two truncation signals stay +// distinct: a cancelled context yields a partial result with a nil error, +// whereas an exceeded budget yields an explicit error. +func TestBlockSearchBudgetVsDeadline(t *testing.T) { + idx := New(dbm.NewMemDB()) + for h := int64(1); h <= 12; h++ { + require.NoError(t, idx.Index(hoBlock(h))) + } + + ctx, cancel := context.WithCancel(t.Context()) + cancel() + results, err := idx.Search(ctx, query.MustCompile(`app.name CONTAINS 'se'`), + indexer.SearchOptions{Limit: 5, OrderDesc: true, MaxScan: 3}) + require.NoError(t, err) + require.Empty(t, results) + + _, err = idx.Search(t.Context(), query.MustCompile(`app.name CONTAINS 'se'`), + indexer.SearchOptions{Limit: 5, OrderDesc: true, MaxScan: 3}) + require.ErrorIs(t, err, indexer.ErrSearchScanBudgetExceeded) +} diff --git a/sei-tendermint/internal/state/indexer/block/kv/util.go b/sei-tendermint/internal/state/indexer/block/kv/util.go index 3c524f36ab..1facde47d2 100644 --- a/sei-tendermint/internal/state/indexer/block/kv/util.go +++ b/sei-tendermint/internal/state/indexer/block/kv/util.go @@ -50,6 +50,53 @@ func eventKey(compositeKey, typ, eventValue string, height int64) ([]byte, error ) } +// eventKeyHeightOrdered builds a height-ordered event key: +// orderedcode(blockHeightOrderedKey, compositeKey, height, eventValue, typ). +// Placing the (real int64) height ahead of the value lets EXISTS-by-tag queries +// scan in height order and early-stop at the limit. +func eventKeyHeightOrdered(compositeKey, typ, eventValue string, height int64) ([]byte, error) { + return orderedcode.Append( + nil, + blockHeightOrderedKey, + compositeKey, + height, + eventValue, + typ, + ) +} + +// prefixHeightOrdered returns the scan prefix orderedcode(blockHeightOrderedKey, +// compositeKey) covering every height-ordered entry for a composite tag, in +// height order. +func prefixHeightOrdered(compositeKey string) ([]byte, error) { + return orderedcode.Append(nil, blockHeightOrderedKey, compositeKey) +} + +// parseHeightFromHeightOrderedKey extracts the height from a height-ordered +// event key. See eventKeyHeightOrdered for the layout. +func parseHeightFromHeightOrderedKey(key []byte) (int64, error) { + var ( + ns, compositeKey, eventValue, typ string + height int64 + ) + + remaining, err := orderedcode.Parse(string(key), &ns, &compositeKey, &height, &eventValue, &typ) + if err != nil { + return 0, fmt.Errorf("failed to parse height-ordered key: %w", err) + } + if len(remaining) != 0 { + return 0, fmt.Errorf("unexpected remainder in key: %s", remaining) + } + + return height, nil +} + +// watermarkKey is the reserved key holding the lowest height covered by the +// height-ordered index on this node. +func watermarkKey() ([]byte, error) { + return orderedcode.Append(nil, blockWatermarkKey) +} + func parseValueFromPrimaryKey(key []byte) (string, error) { var ( compositeKey string diff --git a/sei-tendermint/internal/state/indexer/indexer.go b/sei-tendermint/internal/state/indexer/indexer.go index 394b245879..3ff32f0357 100644 --- a/sei-tendermint/internal/state/indexer/indexer.go +++ b/sei-tendermint/internal/state/indexer/indexer.go @@ -22,8 +22,23 @@ type SearchOptions struct { // When true the highest-ordered results (e.g. most-recent heights) are // kept; when false the lowest-ordered results are kept. OrderDesc bool + + // MaxScan bounds the number of index entries the fallback scan path + // (CONTAINS/MATCHES/non-height value ranges) may examine before the query + // is rejected as too broad. It bounds work, not output: a value <= 0 means + // no budget. Unlike Limit it protects the node from a broad query that must + // scan many entries to return few matches. When the budget is exceeded the + // search fails closed with ErrSearchScanBudgetExceeded rather than + // returning a value-biased partial result. + MaxScan int } +// ErrSearchScanBudgetExceeded is returned when a search exceeds its +// SearchOptions.MaxScan budget on the fallback scan path. The string is a +// stable contract clients may match on; it is deliberately distinct from +// context cancellation, which returns a partial result with a nil error. +var ErrSearchScanBudgetExceeded = errors.New("search scan budget exceeded: query too broad") + // TxIndexer interface defines methods to index and search transactions. type TxIndexer interface { // Index analyzes, indexes and stores transactions. For indexing multiple diff --git a/sei-tendermint/internal/state/indexer/tx/kv/kv.go b/sei-tendermint/internal/state/indexer/tx/kv/kv.go index 8db094c69a..9acf6b60d8 100644 --- a/sei-tendermint/internal/state/indexer/tx/kv/kv.go +++ b/sei-tendermint/internal/state/indexer/tx/kv/kv.go @@ -4,6 +4,7 @@ import ( "context" "encoding/hex" "fmt" + "math" "regexp" "sort" "strconv" @@ -22,6 +23,20 @@ import ( var _ indexer.TxIndexer = (*TxIndex)(nil) +const ( + // txHeightOrderedKey namespaces the height-ordered secondary index. Its + // keys have the form orderedcode(txHeightOrderedKey, tag, height, index, + // value), so within a tag they sort by (height, index) — matching the + // (height, index) order tx_search returns — and stay disjoint from the + // legacy value-ordered index that shares this store. It is reserved: events + // may not use it as a composite key. + txHeightOrderedKey = "tx.height_ordered" + + // txWatermarkKey stores the lowest block height covered by the + // height-ordered index on this node (see readWatermark). It is reserved. + txWatermarkKey = "tx.new_index_min_height" +) + // TxIndex is the simplest possible indexer // It is backed by two kv stores: // 1. txhash - result (primary key) @@ -69,6 +84,7 @@ func (txi *TxIndex) Index(results []*abci.TxResultV2) error { b := txi.store.NewBatch() defer func() { _ = b.Close() }() + minHeight := int64(math.MaxInt64) for _, result := range results { hash := types.Tx(result.Tx).Hash() hashBytes := hash[:] @@ -79,11 +95,16 @@ func (txi *TxIndex) Index(results []*abci.TxResultV2) error { return err } - // index by height (always) + // index by height (always), in both the legacy value-ordered and the + // new height-ordered index. err = b.Set(KeyFromHeight(result), hashBytes) if err != nil { return err } + err = b.Set(keyFromHeightHeightOrdered(result), hashBytes) + if err != nil { + return err + } rawBytes, err := proto.Marshal(&abci.TxResult{Height: result.Height, Index: result.Index, Tx: result.Tx, Result: result.Result}) if err != nil { @@ -94,6 +115,17 @@ func (txi *TxIndex) Index(results []*abci.TxResultV2) error { if err != nil { return err } + + if result.Height < minHeight { + minHeight = result.Height + } + } + + // Advance the watermark for the height-ordered index in the same atomic + // batch as the keys it accounts for, so a crash can never leave the + // watermark below the keys actually written. + if err := txi.updateWatermark(b, minHeight); err != nil { + return err } return b.WriteSync() @@ -114,12 +146,17 @@ func (txi *TxIndex) indexEvents(result *abci.TxResultV2, hash []byte, store dbm. // index if `index: true` is set compositeTag := fmt.Sprintf("%s.%s", event.Type, string(attr.Key)) // ensure event does not conflict with a reserved prefix key - if compositeTag == types.TxHashKey || compositeTag == types.TxHeightKey { + if compositeTag == types.TxHashKey || compositeTag == types.TxHeightKey || + compositeTag == txHeightOrderedKey || compositeTag == txWatermarkKey { return fmt.Errorf("event type and attribute key \"%s\" is reserved; please use a different key", compositeTag) } if attr.GetIndex() { - err := store.Set(keyFromEvent(compositeTag, string(attr.Value), result), hash) - if err != nil { + // dual-write: legacy value-ordered key plus the new + // height-ordered key (see keyFromEventHeightOrdered). + if err := store.Set(keyFromEvent(compositeTag, string(attr.Value), result), hash); err != nil { + return err + } + if err := store.Set(keyFromEventHeightOrdered(compositeTag, string(attr.Value), result), hash); err != nil { return err } } @@ -177,18 +214,31 @@ func (txi *TxIndex) Search(ctx context.Context, q *query.Query, opts indexer.Sea // 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. + // conditions per candidate, and stop at opts.Limit. This uses the legacy + // index (equality has full coverage) and ignores the watermark. 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) } - // 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 := txi.intersect(ctx, conditions, ranges, rangeIndexes) + // Height-ordered path: tx.height range-only and EXISTS-by-tag queries have + // no equality to drive the legacy fast path, but they are height-orderable. + // Drive them off the new height-ordered index, splitting at the watermark so + // pre-upgrade heights (not covered by the new index) fall back to the legacy + // index for full coverage. Early-stops at opts.Limit. + if plan, ok := planHeightOrdered(conditions, ranges, rangeIndexes); ok { + return txi.searchHeightOrdered(ctx, plan, opts) + } + + // Fallback: queries containing CONTAINS/MATCHES, non-height value ranges, or + // a mix of those cannot be driven by an in-order scan. Materialize the + // intersection as before, then bound and order the result set. The scan is + // bounded by opts.MaxScan and fails closed if the budget is exceeded. + budget := indexer.NewScanBudget(opts.MaxScan) + filteredHashes, err := txi.intersect(ctx, conditions, ranges, rangeIndexes, budget) + if err != nil { + return nil, err + } return txi.collectBounded(ctx, filteredHashes, opts) } @@ -202,7 +252,8 @@ func (txi *TxIndex) intersect( conditions []syntax.Condition, ranges indexer.QueryRanges, rangeIndexes []int, -) map[string][]byte { + budget *indexer.ScanBudget, +) (map[string][]byte, error) { var hashesInitialized bool filteredHashes := make(map[string][]byte) @@ -213,17 +264,24 @@ func (txi *TxIndex) intersect( 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, budget) + 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 { - return filteredHashes + 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, budget) + if err != nil { + return nil, err + } } } } @@ -237,21 +295,28 @@ func (txi *TxIndex) intersect( 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, budget) + 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 { - return filteredHashes + 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, budget) + if err != nil { + return nil, err + } } } - return filteredHashes + return filteredHashes, nil } // collectBounded materializes filteredHashes into tx results, orders them by @@ -284,6 +349,36 @@ func (txi *TxIndex) collectBounded(ctx context.Context, filteredHashes map[strin return results, nil } +// collectBoundedInHeightRange is collectBounded restricted to results whose +// height falls within [lo, hi] and capped at limit. It backs the sub-watermark +// leg of the height-ordered split, where the legacy fallback must serve only +// pre-watermark heights. +func (txi *TxIndex) collectBoundedInHeightRange(ctx context.Context, filteredHashes map[string][]byte, orderDesc bool, limit int, lo, hi int64) ([]*abci.TxResultV2, error) { + results := make([]*abci.TxResultV2, 0, indexer.BoundedCap(limit)) + for _, h := range filteredHashes { + res, err := txi.Get(h) + if err != nil { + return nil, fmt.Errorf("failed to get Tx{%X}: %w", h, err) + } + if res != nil && res.Height >= lo && res.Height <= hi { + results = append(results, res) + } + + // Potentially exit early. + if ctx.Err() != nil { + break + } + } + + sortResults(results, orderDesc) + + if limit > 0 && len(results) > limit { + results = results[: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 @@ -459,6 +554,258 @@ func planBounded(conditions []syntax.Condition, ranges indexer.QueryRanges, rang return plan, true } +// heightOrderedPlan describes a query served from the height-ordered index: a +// single (height, index)-ordered scan of driverTag's prefix, filtered by any +// tx.height bounds and split at the watermark (heights >= W come from the new +// index; heights < W come from the legacy fallback for full coverage). +type heightOrderedPlan struct { + // driverTag is the composite key whose height-ordered prefix is scanned. + driverTag string + // heightRanges are tx.height bounds; may be empty for a pure EXISTS query. + heightRanges []indexer.QueryRange + // existsCond, when non-nil, is the single EXISTS condition this plan serves + // (nil for a tx.height-range-only query). It is retained so the + // sub-watermark leg can rebuild the equivalent legacy match. + existsCond *syntax.Condition +} + +// planHeightOrdered decides whether a query is eligible for the height-ordered +// path and builds its plan. It handles the height-orderable shapes that have no +// equality to drive the legacy fast path: a tx.height range-only query, or a +// single EXISTS on a tag (optionally combined with a tx.height range). Anything +// else (CONTAINS/MATCHES, non-height ranges, multi-condition mixes) is left to +// the materializing fallback. +func planHeightOrdered(conditions []syntax.Condition, ranges indexer.QueryRanges, rangeIndexes []int) (heightOrderedPlan, bool) { + var plan heightOrderedPlan + + // Every range must be a numeric tx.height range. + for key, qr := range ranges { + if key != types.TxHeightKey { + return heightOrderedPlan{}, false + } + if _, ok := qr.AnyBound().(int64); !ok { + return heightOrderedPlan{}, false + } + plan.heightRanges = append(plan.heightRanges, qr) + } + + // Collect the non-range conditions. + nonRange := make([]syntax.Condition, 0, len(conditions)) + for i, c := range conditions { + if intInSlice(i, rangeIndexes) { + continue + } + nonRange = append(nonRange, c) + } + + switch { + case len(nonRange) == 0: + // tx.height range-only: need at least one height range to drive a scan. + if len(plan.heightRanges) == 0 { + return heightOrderedPlan{}, false + } + plan.driverTag = types.TxHeightKey + case len(nonRange) == 1 && nonRange[0].Op == syntax.TExists: + plan.driverTag = nonRange[0].Tag + plan.existsCond = &nonRange[0] + default: + return heightOrderedPlan{}, false + } + + return plan, true +} + +// heightBounds reduces the plan's tx.height ranges to a single inclusive +// [lo, hi] window. Heights are >= 1, so an absent lower bound defaults to 1 +// (avoiding a pointless legacy scan for a non-existent sub-watermark tail). +func heightBounds(ranges []indexer.QueryRange) (lo, hi int64) { + lo, hi = 1, int64(math.MaxInt64) + for i := range ranges { + if lb := ranges[i].LowerBoundValue(); lb != nil { + if v, ok := lb.(int64); ok && v > lo { + lo = v + } + } + if ub := ranges[i].UpperBoundValue(); ub != nil { + if v, ok := ub.(int64); ok && v < hi { + hi = v + } + } + } + return lo, hi +} + +// searchHeightOrdered serves a heightOrderedPlan by splitting the [lo, hi] +// window at the watermark W: heights in [max(lo, W), hi] are streamed from the +// new height-ordered index (fast, early-stops at opts.Limit); heights in +// [lo, W-1] are served by the legacy materializing fallback for full coverage. +// The two sub-ranges are height-disjoint at W, so no global merge is needed — +// whichever sub-range holds the higher (for desc) or lower (for asc) heights is +// drained first, and the other only if the limit is not yet met. +func (txi *TxIndex) searchHeightOrdered(ctx context.Context, plan heightOrderedPlan, opts indexer.SearchOptions) ([]*abci.TxResultV2, error) { + w := txi.readWatermark() + lo, hi := heightBounds(plan.heightRanges) + if lo > hi { + return []*abci.TxResultV2{}, nil + } + + fastLo := max(lo, w) + hasFast := fastLo <= hi + + fbHi := min(hi, w-1) + hasFallback := lo <= fbHi + + results := make([]*abci.TxResultV2, 0, indexer.BoundedCap(opts.Limit)) + remaining := func() int { + if opts.Limit <= 0 { + return -1 + } + return opts.Limit - len(results) + } + reachedLimit := func() bool { + return opts.Limit > 0 && len(results) >= opts.Limit + } + + if opts.OrderDesc { + if hasFast { + r, err := txi.scanHeightOrderedFast(ctx, plan, fastLo, hi, true, remaining()) + if err != nil { + return nil, err + } + results = append(results, r...) + } + if !reachedLimit() && hasFallback { + r, err := txi.heightOrderedFallback(ctx, plan, lo, fbHi, true, remaining(), opts.MaxScan) + if err != nil { + return nil, err + } + results = append(results, r...) + } + } else { + if hasFallback { + r, err := txi.heightOrderedFallback(ctx, plan, lo, fbHi, false, remaining(), opts.MaxScan) + if err != nil { + return nil, err + } + results = append(results, r...) + } + if !reachedLimit() && hasFast { + r, err := txi.scanHeightOrderedFast(ctx, plan, fastLo, hi, false, remaining()) + if err != nil { + return nil, err + } + results = append(results, r...) + } + } + + if opts.Limit > 0 && len(results) > opts.Limit { + results = results[:opts.Limit] + } + return results, nil +} + +// scanHeightOrderedFast streams the new height-ordered index for plan.driverTag +// over heights [lo, hi] in (height, index) order_by order, deduping by hash and +// stopping at limit (limit <= 0 means unbounded). Because keys are height-major +// the scan early-stops once it passes the far bound. This is a Limit-bounded +// fast-path scan and is deliberately not charged against the scan budget. +func (txi *TxIndex) scanHeightOrderedFast(ctx context.Context, plan heightOrderedPlan, lo, hi int64, desc bool, limit int) ([]*abci.TxResultV2, error) { + it, err := txi.prefixIterator(prefixHeightOrdered(plan.driverTag), desc) + if err != nil { + return nil, fmt.Errorf("failed to create height-ordered iterator: %w", err) + } + defer func() { _ = it.Close() }() + + results := make([]*abci.TxResultV2, 0, indexer.BoundedCap(limit)) + seen := make(map[string]struct{}) + + for ; it.Valid(); it.Next() { + if ctx.Err() != nil { + break + } + + height, _, err := parseHeightOrderedKey(it.Key()) + if err != nil { + continue + } + + // Keys are height-major, so once we pass the far bound in scan order we + // can stop; the near bound is skipped until we reach the window. + if desc { + if height > hi { + continue + } + if height < lo { + break + } + } else { + if height < lo { + continue + } + if height > hi { + break + } + } + + hash := it.Value() + if _, dup := seen[string(hash)]; dup { + 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 limit > 0 && len(results) >= limit { + break + } + } + + if err := it.Error(); err != nil { + return nil, err + } + + return results, nil +} + +// heightOrderedFallback serves the pre-watermark [lo, hi] leg of a +// heightOrderedPlan from the legacy index. It rebuilds the equivalent legacy +// match, then collects results whose height is in [lo, hi], ordered and capped +// at limit. The legacy scan is charged against a maxScan budget and fails +// closed if the budget is exceeded. +func (txi *TxIndex) heightOrderedFallback(ctx context.Context, plan heightOrderedPlan, lo, hi int64, desc bool, limit, maxScan int) ([]*abci.TxResultV2, error) { + budget := indexer.NewScanBudget(maxScan) + + var ( + filtered map[string][]byte + err error + ) + if plan.existsCond != nil { + filtered, err = txi.match(ctx, *plan.existsCond, nil, map[string][]byte{}, true, budget) + } else { + qr := indexer.QueryRange{ + Key: types.TxHeightKey, + LowerBound: lo, + UpperBound: hi, + IncludeLowerBound: true, + IncludeUpperBound: true, + } + filtered, err = txi.matchRange(ctx, qr, prefixFromCompositeKey(types.TxHeightKey), map[string][]byte{}, true, budget) + } + if err != nil { + return nil, err + } + + return txi.collectBoundedInHeightRange(ctx, filtered, desc, limit, lo, hi) +} + // 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. @@ -511,11 +858,12 @@ func (txi *TxIndex) match( startKeyBz []byte, filteredHashes map[string][]byte, firstRun bool, -) map[string][]byte { + budget *indexer.ScanBudget, +) (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) @@ -524,12 +872,15 @@ 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 := budget.Step(); err != nil { + return nil, err + } tmpHashes[string(it.Value())] = it.Value() // Potentially exit early. @@ -540,7 +891,7 @@ func (txi *TxIndex) match( } } if err := it.Error(); err != nil { - panic(err) + return nil, err } case syntax.TExists: @@ -548,12 +899,15 @@ 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 := budget.Step(); err != nil { + return nil, err + } tmpHashes[string(it.Value())] = it.Value() // Potentially exit early. @@ -564,7 +918,7 @@ func (txi *TxIndex) match( } } if err := it.Error(); err != nil { - panic(err) + return nil, err } case syntax.TContains: @@ -573,12 +927,15 @@ 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 := budget.Step(); err != nil { + return nil, err + } value, err := parseValueFromKey(it.Key()) if err != nil { continue @@ -595,18 +952,21 @@ 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 := budget.Step(); err != nil { + return nil, err + } value, err := parseValueFromKey(it.Key()) if err != nil { continue @@ -623,10 +983,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, fmt.Errorf("other operators should be handled already") } if len(tmpHashes) == 0 || firstRun { @@ -637,7 +997,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 @@ -655,7 +1015,7 @@ func (txi *TxIndex) match( } } - return filteredHashes + return filteredHashes, nil } // matchRange returns all matching txs by hash that meet a given queryRange and @@ -669,11 +1029,12 @@ func (txi *TxIndex) matchRange( startKey []byte, filteredHashes map[string][]byte, firstRun bool, -) map[string][]byte { + budget *indexer.ScanBudget, +) (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) @@ -682,12 +1043,15 @@ 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 := budget.Step(); err != nil { + return nil, err + } value, err := parseValueFromKey(it.Key()) if err != nil { continue @@ -727,7 +1091,7 @@ iter: } } if err := it.Error(); err != nil { - panic(err) + return nil, err } if len(tmpHashes) == 0 || firstRun { @@ -738,7 +1102,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 @@ -756,7 +1120,7 @@ iter: } } - return filteredHashes + return filteredHashes, nil } // ########################## Keys ############################# @@ -827,6 +1191,106 @@ func KeyFromHeight(result *abci.TxResultV2) []byte { return secondaryKey(types.TxHeightKey, fmt.Sprintf("%d", result.Height), result.Height, result.Index) } +// ################## Height-ordered index (PLT-786) ################## +// +// The height-ordered secondary key places the (real int64) height ahead of the +// value so that, within a composite tag, keys sort by (height, index) — the +// same order tx_search returns. This lets tx.height ranges and EXISTS-by-tag +// queries scan in result order and early-stop at the limit, instead of +// materializing and sorting the full match set. It is namespaced under the +// reserved txHeightOrderedKey so it stays disjoint from the legacy index in the +// shared store. + +// secondaryKeyHeightOrdered builds a height-ordered event key: +// orderedcode(txHeightOrderedKey, compositeKey, height, index, value). +func secondaryKeyHeightOrdered(compositeKey, value string, height int64, index uint32) []byte { + key, err := orderedcode.Append( + nil, + txHeightOrderedKey, + compositeKey, + height, + int64(index), + value, + ) + if err != nil { + panic(err) + } + return key +} + +func keyFromEventHeightOrdered(compositeKey string, value string, result *abci.TxResultV2) []byte { + return secondaryKeyHeightOrdered(compositeKey, value, result.Height, result.Index) +} + +func keyFromHeightHeightOrdered(result *abci.TxResultV2) []byte { + return secondaryKeyHeightOrdered(types.TxHeightKey, fmt.Sprintf("%d", result.Height), result.Height, result.Index) +} + +// prefixHeightOrdered returns the scan prefix orderedcode(txHeightOrderedKey, +// compositeKey) covering every height-ordered entry for a composite tag, in +// (height, index) order. +func prefixHeightOrdered(compositeKey string) []byte { + key, err := orderedcode.Append(nil, txHeightOrderedKey, compositeKey) + if err != nil { + panic(err) + } + return key +} + +// parseHeightOrderedKey extracts (height, index) from a height-ordered event +// key. See secondaryKeyHeightOrdered for the layout. +func parseHeightOrderedKey(key []byte) (int64, uint32, error) { + var ( + ns, compositeKey, value string + height, index int64 + ) + remaining, err := orderedcode.Parse(string(key), &ns, &compositeKey, &height, &index, &value) + if err != nil { + return 0, 0, fmt.Errorf("failed to parse height-ordered 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), so the round-trip cannot overflow +} + +// watermarkKey is the reserved key holding the lowest height covered by the +// height-ordered index on this node. +func watermarkKey() []byte { + key, err := orderedcode.Append(nil, txWatermarkKey) + if err != nil { + panic(err) + } + return key +} + +// readWatermark returns the lowest block height covered by the height-ordered +// index. An unset watermark (fresh DB, or upgraded-but-not-yet-written) reads +// as math.MaxInt64 so every height-ordered query takes the legacy fallback +// until the new index has written at least one key. +func (txi *TxIndex) readWatermark() int64 { + bz, err := txi.store.Get(watermarkKey()) + if err != nil { + panic(err) + } + if len(bz) == 0 { + return math.MaxInt64 + } + return int64FromBytes(bz) +} + +// updateWatermark lowers the persisted watermark to height when height is lower, +// writing into the provided batch so it commits atomically with the +// height-ordered keys it accounts for. A watermark that is too high is only +// over-conservative (routes covered heights to the fallback); a watermark below +// the keys actually written would be incorrect, so it is never raised here. +func (txi *TxIndex) updateWatermark(b dbm.Batch, height int64) error { + if height >= txi.readWatermark() { + return nil + } + return b.Set(watermarkKey(), int64ToBytes(height)) +} + // Prefixes: these represent an initial part of the key and are used by iterators to iterate over a small // section of the kv store during searches. 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 8c15a13a19..9bfcbd65ae 100644 --- a/sei-tendermint/internal/state/indexer/tx/kv/kv_test.go +++ b/sei-tendermint/internal/state/indexer/tx/kv/kv_test.go @@ -408,9 +408,9 @@ func TestTxSearchBounded(t *testing.T) { 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-ordered path: a tx.height-range-only query has no equality to + // drive the legacy fast path, so it streams the new height-ordered index + // (split at the watermark) in order and early-stops at the limit. "height range only desc limit 3": { q: `tx.height >= 4`, opts: indexer.SearchOptions{Limit: 3, OrderDesc: true}, @@ -457,7 +457,7 @@ func TestTxSearchBounded(t *testing.T) { `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 + `tx.height >= 3`, // height-ordered: height-range-only `app.name CONTAINS 'se'`, // fallback: materialize then bound } for _, q := range queries { diff --git a/sei-tendermint/internal/state/indexer/tx/kv/kv_watermark_test.go b/sei-tendermint/internal/state/indexer/tx/kv/kv_watermark_test.go new file mode 100644 index 0000000000..23e01fdf4f --- /dev/null +++ b/sei-tendermint/internal/state/indexer/tx/kv/kv_watermark_test.go @@ -0,0 +1,334 @@ +package kv + +import ( + "context" + "errors" + "fmt" + "math" + "testing" + + "github.com/gogo/protobuf/proto" + "github.com/stretchr/testify/require" + dbm "github.com/tendermint/tm-db" + + abci "github.com/sei-protocol/sei-chain/sei-tendermint/abci/types" + "github.com/sei-protocol/sei-chain/sei-tendermint/internal/pubsub/query" + "github.com/sei-protocol/sei-chain/sei-tendermint/internal/state/indexer" + "github.com/sei-protocol/sei-chain/sei-tendermint/types" +) + +// hipair identifies a tx by (height, index) for order-sensitive assertions. +type hipair struct { + h int64 + i uint32 +} + +func toPairs(results []*abci.TxResultV2) []hipair { + out := make([]hipair, len(results)) + for k, r := range results { + out[k] = hipair{r.Height, r.Index} + } + return out +} + +// hoTx builds a tx result at (height, index) carrying app.name=sei. +func hoTx(height int64, index uint32) *abci.TxResultV2 { + return &abci.TxResultV2{ + Height: height, + Index: index, + Tx: types.Tx(fmt.Sprintf("tx-%d-%d", height, index)), + Result: abci.ExecTxResult{Code: abci.CodeTypeOK, Events: []abci.Event{{ + Type: "app", + Attributes: []abci.EventAttribute{{Key: []byte("name"), Value: []byte("sei"), Index: true}}, + }}}, + } +} + +// indexLegacyOnly writes the legacy (value-ordered) index entries and the +// primary key for a tx, but not the new height-ordered keys, and does not touch +// the watermark. It simulates a pre-upgrade block that predates the +// height-ordered index (forward-fill leaves these uncovered by the new index). +func indexLegacyOnly(t *testing.T, idx *TxIndex, res *abci.TxResultV2) { + t.Helper() + b := idx.store.NewBatch() + defer func() { _ = b.Close() }() + + hash := types.Tx(res.Tx).Hash() + hashBytes := hash[:] + + for _, event := range res.Result.Events { + if len(event.Type) == 0 { + continue + } + for _, attr := range event.Attributes { + if len(attr.Key) == 0 || !attr.GetIndex() { + continue + } + compositeTag := fmt.Sprintf("%s.%s", event.Type, string(attr.Key)) + require.NoError(t, b.Set(keyFromEvent(compositeTag, string(attr.Value), res), hashBytes)) + } + } + require.NoError(t, b.Set(KeyFromHeight(res), hashBytes)) + + rawBytes, err := proto.Marshal(&abci.TxResult{Height: res.Height, Index: res.Index, Tx: res.Tx, Result: res.Result}) + require.NoError(t, err) + require.NoError(t, b.Set(primaryKey(hashBytes), rawBytes)) + require.NoError(t, b.WriteSync()) +} + +// backfillHeightOrdered writes the height-ordered keys for a previously +// legacy-only tx, simulating a background backfill (without lowering W here). +func backfillHeightOrdered(t *testing.T, idx *TxIndex, res *abci.TxResultV2) { + t.Helper() + b := idx.store.NewBatch() + defer func() { _ = b.Close() }() + + hash := types.Tx(res.Tx).Hash() + hashBytes := hash[:] + for _, event := range res.Result.Events { + for _, attr := range event.Attributes { + if !attr.GetIndex() { + continue + } + compositeTag := fmt.Sprintf("%s.%s", event.Type, string(attr.Key)) + require.NoError(t, b.Set(keyFromEventHeightOrdered(compositeTag, string(attr.Value), res), hashBytes)) + } + } + require.NoError(t, b.Set(keyFromHeightHeightOrdered(res), hashBytes)) + require.NoError(t, b.WriteSync()) +} + +func setWatermark(t *testing.T, idx *TxIndex, w int64) { + t.Helper() + require.NoError(t, idx.store.Set(watermarkKey(), int64ToBytes(w))) +} + +// hoFixture builds an index with two txs per height for heights 1..10. Heights +// 1..5 are legacy-only (pre-upgrade); heights 6..10 are dual-written. The +// watermark is set to 6, so a height-ordered query splits: [6,10] from the new +// index, [1,5] from the legacy fallback. +const hoMaxHeight = 10 +const hoWatermark = 6 + +func hoFixture(t *testing.T) *TxIndex { + t.Helper() + idx := NewTxIndex(dbm.NewMemDB()) + + // Post-upgrade heights (dual-write) first so the watermark settles at 6, + // then splice the pre-upgrade legacy-only heights beneath it. + for h := int64(hoWatermark); h <= hoMaxHeight; h++ { + for i := uint32(0); i < 2; i++ { + require.NoError(t, idx.Index([]*abci.TxResultV2{hoTx(h, i)})) + } + } + require.Equal(t, int64(hoWatermark), idx.readWatermark()) + + for h := int64(1); h < hoWatermark; h++ { + for i := uint32(0); i < 2; i++ { + indexLegacyOnly(t, idx, hoTx(h, i)) + } + } + // Watermark must be unaffected by legacy-only writes. + require.Equal(t, int64(hoWatermark), idx.readWatermark()) + return idx +} + +// reference returns the expected (height,index) pairs for a query matching all +// txs with height in [lo,hi], ordered per desc and capped at limit (<=0 == all). +func reference(desc bool, lo, hi int64, limit int) []hipair { + var out []hipair + appendHeight := func(h int64) { + if h < lo || h > hi { + return + } + if desc { + out = append(out, hipair{h, 1}, hipair{h, 0}) + } else { + out = append(out, hipair{h, 0}, hipair{h, 1}) + } + } + if desc { + for h := int64(hoMaxHeight); h >= 1; h-- { + appendHeight(h) + } + } else { + for h := int64(1); h <= hoMaxHeight; h++ { + appendHeight(h) + } + } + if limit > 0 && len(out) > limit { + out = out[:limit] + } + return out +} + +// TestTxIndexDualWrite verifies the PLT-786 contract: every indexed attribute +// is written to both the legacy value-ordered key and the new height-ordered +// key, and the watermark is advanced to the indexed height. +func TestTxIndexDualWrite(t *testing.T) { + idx := NewTxIndex(dbm.NewMemDB()) + res := hoTx(7, 0) + require.NoError(t, idx.Index([]*abci.TxResultV2{res})) + + // Legacy value-ordered event key present. + legacy, err := idx.store.Has(keyFromEvent("app.name", "sei", res)) + require.NoError(t, err) + require.True(t, legacy, "legacy event key must be written") + + // New height-ordered event key present. + ho, err := idx.store.Has(keyFromEventHeightOrdered("app.name", "sei", res)) + require.NoError(t, err) + require.True(t, ho, "height-ordered event key must be written") + + // tx.height dual-write present in both indexes. + legacyHeight, err := idx.store.Has(KeyFromHeight(res)) + require.NoError(t, err) + require.True(t, legacyHeight) + hoHeight, err := idx.store.Has(keyFromHeightHeightOrdered(res)) + require.NoError(t, err) + require.True(t, hoHeight) + + // Watermark advanced to the indexed height. + require.Equal(t, int64(7), idx.readWatermark()) +} + +// TestTxSearchWatermarkSplit exercises the height-ordered split-merge across the +// watermark for tx.height ranges and EXISTS, at boundary lower bounds +// {W-1, W, W+1}, both orderings, and various limits. The result must always +// equal the full reference set (forward-fill leaves no gap). +func TestTxSearchWatermarkSplit(t *testing.T) { + idx := hoFixture(t) + + cases := []struct { + name string + q string + lo, hi int64 + limitSet []int + }{ + {"exists full range", `app.name EXISTS`, 1, hoMaxHeight, []int{0, 1, 3, 7, 11, 20}}, + {"height >= W-1 (5)", `tx.height >= 5`, 5, hoMaxHeight, []int{0, 1, 2, 3, 5}}, + {"height >= W (6)", `tx.height >= 6`, 6, hoMaxHeight, []int{0, 1, 4}}, + {"height >= W+1 (7)", `tx.height >= 7`, 7, hoMaxHeight, []int{0, 2, 8}}, + {"height <= W-1 (5)", `tx.height <= 5`, 1, 5, []int{0, 3}}, + {"height <= W (6)", `tx.height <= 6`, 1, 6, []int{0, 3}}, + {"height window straddling W", `tx.height >= 4 AND tx.height <= 7`, 4, 7, []int{0, 2, 5}}, + } + + for _, tc := range cases { + for _, desc := range []bool{true, false} { + for _, limit := range tc.limitSet { + name := fmt.Sprintf("%s/desc=%v/limit=%d", tc.name, desc, limit) + t.Run(name, func(t *testing.T) { + results, err := idx.Search(t.Context(), query.MustCompile(tc.q), + indexer.SearchOptions{Limit: limit, OrderDesc: desc, MaxScan: 0}) + require.NoError(t, err) + require.Equal(t, reference(desc, tc.lo, tc.hi, limit), toPairs(results)) + }) + } + } + } +} + +// TestTxSearchWatermarkUnset verifies that an unset watermark (fresh / +// upgraded-but-not-written DB) routes every height-ordered query to the legacy +// fallback and still returns the full set. +func TestTxSearchWatermarkUnset(t *testing.T) { + idx := NewTxIndex(dbm.NewMemDB()) + // Write only legacy entries and force the watermark to +inf. + for h := int64(1); h <= 5; h++ { + indexLegacyOnly(t, idx, hoTx(h, 0)) + } + require.Equal(t, int64(math.MaxInt64), idx.readWatermark()) + + results, err := idx.Search(t.Context(), query.MustCompile(`app.name EXISTS`), + indexer.SearchOptions{OrderDesc: true}) + require.NoError(t, err) + + want := []hipair{{5, 0}, {4, 0}, {3, 0}, {2, 0}, {1, 0}} + require.Equal(t, want, toPairs(results)) +} + +// TestTxSearchBackfillLowersWatermark verifies that backfilling the sub-watermark +// heights into the new index and lowering W keeps results correct and does not +// double-count across the split. +func TestTxSearchBackfillLowersWatermark(t *testing.T) { + idx := hoFixture(t) + + q := query.MustCompile(`app.name EXISTS`) + before, err := idx.Search(t.Context(), q, indexer.SearchOptions{OrderDesc: true}) + require.NoError(t, err) + require.Equal(t, reference(true, 1, hoMaxHeight, 0), toPairs(before)) + + // Backfill heights 1..5 into the new index and lower the watermark to 1. + for h := int64(1); h < hoWatermark; h++ { + for i := uint32(0); i < 2; i++ { + backfillHeightOrdered(t, idx, hoTx(h, i)) + } + } + setWatermark(t, idx, 1) + + after, err := idx.Search(t.Context(), q, indexer.SearchOptions{OrderDesc: true}) + require.NoError(t, err) + // Same set, no duplicates introduced by the now-all-fast-path scan. + require.Equal(t, reference(true, 1, hoMaxHeight, 0), toPairs(after)) +} + +// TestTxSearchScanBudgetFailClosed asserts that a fallback query exceeding the +// scan budget fails closed with ErrSearchScanBudgetExceeded (no partial), for +// both orderings and single- and multi-condition queries, while a query within +// budget succeeds. +func TestTxSearchScanBudgetFailClosed(t *testing.T) { + idx := NewTxIndex(dbm.NewMemDB()) + // 12 txs, each carrying app.name=sei so the CONTAINS fallback must scan all. + for h := int64(1); h <= 12; h++ { + res := hoTx(h, 0) + res.Result.Events = append(res.Result.Events, abci.Event{ + Type: "app", + Attributes: []abci.EventAttribute{{Key: []byte("kind"), Value: []byte("even"), Index: true}}, + }) + require.NoError(t, idx.Index([]*abci.TxResultV2{res})) + } + + for _, desc := range []bool{true, false} { + for _, q := range []string{ + `app.name CONTAINS 'se'`, + `app.name CONTAINS 'se' AND app.kind CONTAINS 'ev'`, + } { + t.Run(fmt.Sprintf("desc=%v/%s", desc, q), func(t *testing.T) { + results, err := idx.Search(t.Context(), query.MustCompile(q), + indexer.SearchOptions{Limit: 5, OrderDesc: desc, MaxScan: 3}) + require.ErrorIs(t, err, indexer.ErrSearchScanBudgetExceeded) + require.Nil(t, results) + }) + } + } + + // Within budget (or disabled) the same query succeeds. + results, err := idx.Search(t.Context(), query.MustCompile(`app.name CONTAINS 'se'`), + indexer.SearchOptions{Limit: 5, OrderDesc: true, MaxScan: 0}) + require.NoError(t, err) + require.Len(t, results, 5) +} + +// TestTxSearchBudgetVsDeadline asserts the two truncation signals stay distinct: +// a cancelled context yields a partial result with a nil error, whereas an +// exceeded budget yields an explicit error. +func TestTxSearchBudgetVsDeadline(t *testing.T) { + idx := NewTxIndex(dbm.NewMemDB()) + for h := int64(1); h <= 12; h++ { + require.NoError(t, idx.Index([]*abci.TxResultV2{hoTx(h, 0)})) + } + + // Deadline: cancelled context -> nil error. + ctx, cancel := context.WithCancel(t.Context()) + cancel() + results, err := idx.Search(ctx, query.MustCompile(`app.name CONTAINS 'se'`), + indexer.SearchOptions{Limit: 5, OrderDesc: true, MaxScan: 3}) + require.NoError(t, err) + require.Empty(t, results) + + // Budget: fresh context, tight budget -> explicit error. + _, err = idx.Search(t.Context(), query.MustCompile(`app.name CONTAINS 'se'`), + indexer.SearchOptions{Limit: 5, OrderDesc: true, MaxScan: 3}) + require.True(t, errors.Is(err, indexer.ErrSearchScanBudgetExceeded)) +} diff --git a/sei-tendermint/internal/state/indexer/tx/kv/utils.go b/sei-tendermint/internal/state/indexer/tx/kv/utils.go index 7436ad1224..1237c59a94 100644 --- a/sei-tendermint/internal/state/indexer/tx/kv/utils.go +++ b/sei-tendermint/internal/state/indexer/tx/kv/utils.go @@ -1,11 +1,27 @@ package kv import ( + "encoding/binary" "fmt" "github.com/google/orderedcode" ) +// int64ToBytes encodes an int64 as a varint. Used for the height-ordered index +// watermark value, which is a point value read/written by key (never scanned in +// order), so a compact non-order-preserving encoding is fine. +func int64ToBytes(i int64) []byte { + buf := make([]byte, binary.MaxVarintLen64) + n := binary.PutVarint(buf, i) + return buf[:n] +} + +// int64FromBytes decodes a varint-encoded int64 produced by int64ToBytes. +func int64FromBytes(bz []byte) int64 { + v, _ := binary.Varint(bz) + return v +} + // intInSlice returns true if a is found in the list. func intInSlice(a int, list []int) bool { for _, b := range list { diff --git a/sei-tendermint/internal/state/indexer/utils.go b/sei-tendermint/internal/state/indexer/utils.go index 39b6a966e6..196f8cb728 100644 --- a/sei-tendermint/internal/state/indexer/utils.go +++ b/sei-tendermint/internal/state/indexer/utils.go @@ -34,6 +34,43 @@ func HeightInRange(h int64, qr QueryRange) bool { return true } +// ScanBudget bounds the number of index entries a fallback scan may examine +// across all passes of a single query (one query = one shared budget). It +// counts work, not output: it is stepped once per iterator advance on the +// fallback scan path and not for fast-path driver scans or point-probes. A +// non-positive limit disables the budget. +type ScanBudget struct { + limit int + used int +} + +// NewScanBudget returns a ScanBudget with the given entry limit (<= 0 disables). +func NewScanBudget(limit int) *ScanBudget { + return &ScanBudget{limit: limit} +} + +// Step accounts for one examined entry and returns ErrSearchScanBudgetExceeded +// once the cumulative count exceeds the limit. It is safe to call on a nil +// budget (treated as unlimited). +func (b *ScanBudget) Step() error { + if b == nil || b.limit <= 0 { + return nil + } + b.used++ + if b.used > b.limit { + return ErrSearchScanBudgetExceeded + } + return nil +} + +// Used reports the number of entries examined so far. +func (b *ScanBudget) Used() int { + if b == nil { + return 0 + } + return b.used +} + // 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). From 2b99be0ca588c016bb79bcf86f5d2c319f4aa434 Mon Sep 17 00:00:00 2001 From: Amir Deris Date: Thu, 9 Jul 2026 10:33:17 -0700 Subject: [PATCH 2/8] Some cleanup --- sei-tendermint/config/config.go | 7 ++--- sei-tendermint/config/toml.go | 8 +++--- .../internal/state/indexer/block/kv/kv.go | 28 +++++++++++-------- 3 files changed, 23 insertions(+), 20 deletions(-) diff --git a/sei-tendermint/config/config.go b/sei-tendermint/config/config.go index 5a430c879a..81407d3de9 100644 --- a/sei-tendermint/config/config.go +++ b/sei-tendermint/config/config.go @@ -539,10 +539,7 @@ type RPCConfig struct { // Maximum number of index entries a single tx_search / block_search may // examine on the fallback scan path (CONTAINS/MATCHES/value ranges) before - // the query is rejected as too broad. This bounds work, not output: unlike - // max-tx-search-results (which caps the result set) it protects the node - // from a broad query that must scan millions of entries to return a handful - // of matches. 0 disables the budget (not recommended on public nodes). + // the query is rejected as too broad. 0 disables the budget. MaxEventSearchScan int `mapstructure:"max-event-search-scan"` } @@ -578,7 +575,7 @@ func DefaultRPCConfig() *RPCConfig { TimeoutWrite: 30 * time.Second, MaxTxSearchResults: 10_000, - MaxEventSearchScan: 1_000_000, + MaxEventSearchScan: 50_000, } } diff --git a/sei-tendermint/config/toml.go b/sei-tendermint/config/toml.go index b1a65e9e0b..5fca7bf367 100644 --- a/sei-tendermint/config/toml.go +++ b/sei-tendermint/config/toml.go @@ -287,10 +287,10 @@ 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 index entries a single tx_search / block_search may examine -# on the fallback scan path (CONTAINS/MATCHES/value ranges) before the query is -# rejected as too broad. Bounds work rather than output. Set to 0 to disable -# (not recommended on public nodes). +# Maximum number of index entries a single query on the /tx_search or +# /block_search RPC endpoint may examine on the scan path +# before the query is rejected as too broad. +# Set to 0 to disable (not recommended on public nodes). max-event-search-scan = {{ .RPC.MaxEventSearchScan }} ####################################################################### diff --git a/sei-tendermint/internal/state/indexer/block/kv/kv.go b/sei-tendermint/internal/state/indexer/block/kv/kv.go index af2c7b4b43..001a38e8b8 100644 --- a/sei-tendermint/internal/state/indexer/block/kv/kv.go +++ b/sei-tendermint/internal/state/indexer/block/kv/kv.go @@ -63,30 +63,33 @@ func (idx *BlockerIndexer) Has(height int64) (bool, error) { // readWatermark returns the lowest block height covered by the height-ordered // index. An unset watermark (fresh DB, or upgraded-but-not-yet-written) reads -// as math.MaxInt64 so every height-ordered query takes the legacy fallback +// as math.MaxInt64 so every height-ordered query takes the fallback path // until the new index has written at least one key. -func (idx *BlockerIndexer) readWatermark() int64 { +func (idx *BlockerIndexer) readWatermark() (int64, error) { key, err := watermarkKey() if err != nil { - panic(err) + return 0, err } bz, err := idx.store.Get(key) if err != nil { - panic(err) + return 0, err } if len(bz) == 0 { - return math.MaxInt64 + return math.MaxInt64, nil } - return int64FromBytes(bz) + return int64FromBytes(bz), nil } // updateWatermark lowers the persisted watermark to height when height is lower, // writing into the provided batch so it commits atomically with the // height-ordered keys it accounts for. A watermark that is too high is only -// over-conservative (routes covered heights to the fallback); a watermark below -// the keys actually written would be incorrect, so it is never raised here. +// over-conservative (routes covered heights to the fallback). func (idx *BlockerIndexer) updateWatermark(batch dbm.Batch, height int64) error { - if height >= idx.readWatermark() { + wmIndex, err := idx.readWatermark() + if err != nil { + return err + } + if height >= wmIndex { return nil } key, err := watermarkKey() @@ -168,7 +171,7 @@ func (idx *BlockerIndexer) Search(ctx context.Context, q *query.Query, opts inde // scan and point-probed, stream candidates in order_by order and stop at // the limit, so a broad query does not materialize and sort the full match // set. This uses the legacy index / primary key (full coverage) and ignores - // the watermark. + // the watermark for height order index. if plan, ok := planBounded(conditions, ranges, rangeIndexes); ok { return idx.searchBounded(ctx, plan, opts) } @@ -609,7 +612,10 @@ func heightBounds(ranges []indexer.QueryRange) (lo, hi int64) { // [lo, W-1] are served by the legacy materializing fallback for full coverage. // The two sub-ranges are height-disjoint at W, so no global merge is needed. func (idx *BlockerIndexer) searchHeightOrdered(ctx context.Context, plan heightOrderedPlan, opts indexer.SearchOptions) ([]int64, error) { - w := idx.readWatermark() + w, err := idx.readWatermark() + if err != nil { + return nil, err + } lo, hi := heightBounds(plan.heightRanges) if lo > hi { return []int64{}, nil From 7516dfdc852c1a8905a5d58dd1ace993ff43245c Mon Sep 17 00:00:00 2001 From: Amir Deris Date: Thu, 9 Jul 2026 10:49:17 -0700 Subject: [PATCH 3/8] Fixed watermark error --- .../internal/state/indexer/block/kv/kv.go | 4 ++-- .../indexer/block/kv/kv_watermark_test.go | 12 +++++++++--- .../internal/state/indexer/tx/kv/kv.go | 19 +++++++++++++------ .../state/indexer/tx/kv/kv_watermark_test.go | 16 ++++++++++++---- 4 files changed, 36 insertions(+), 15 deletions(-) diff --git a/sei-tendermint/internal/state/indexer/block/kv/kv.go b/sei-tendermint/internal/state/indexer/block/kv/kv.go index 001a38e8b8..2aefb54d9e 100644 --- a/sei-tendermint/internal/state/indexer/block/kv/kv.go +++ b/sei-tendermint/internal/state/indexer/block/kv/kv.go @@ -85,11 +85,11 @@ func (idx *BlockerIndexer) readWatermark() (int64, error) { // height-ordered keys it accounts for. A watermark that is too high is only // over-conservative (routes covered heights to the fallback). func (idx *BlockerIndexer) updateWatermark(batch dbm.Batch, height int64) error { - wmIndex, err := idx.readWatermark() + w, err := idx.readWatermark() if err != nil { return err } - if height >= wmIndex { + if height >= w { return nil } key, err := watermarkKey() diff --git a/sei-tendermint/internal/state/indexer/block/kv/kv_watermark_test.go b/sei-tendermint/internal/state/indexer/block/kv/kv_watermark_test.go index 28e1b95f9e..797b415ed8 100644 --- a/sei-tendermint/internal/state/indexer/block/kv/kv_watermark_test.go +++ b/sei-tendermint/internal/state/indexer/block/kv/kv_watermark_test.go @@ -100,12 +100,16 @@ func hoFixture(t *testing.T) *BlockerIndexer { for h := int64(hoWatermark); h <= hoMaxHeight; h++ { require.NoError(t, idx.Index(hoBlock(h))) } - require.Equal(t, int64(hoWatermark), idx.readWatermark()) + w, err := idx.readWatermark() + require.NoError(t, err) + require.Equal(t, int64(hoWatermark), w) for h := int64(1); h < hoWatermark; h++ { indexLegacyOnly(t, idx, hoBlock(h)) } - require.Equal(t, int64(hoWatermark), idx.readWatermark()) + w, err = idx.readWatermark() + require.NoError(t, err) + require.Equal(t, int64(hoWatermark), w) return idx } @@ -176,7 +180,9 @@ func TestBlockSearchWatermarkUnset(t *testing.T) { for h := int64(1); h <= 5; h++ { indexLegacyOnly(t, idx, hoBlock(h)) } - require.Equal(t, int64(math.MaxInt64), idx.readWatermark()) + w, err := idx.readWatermark() + require.NoError(t, err) + require.Equal(t, int64(math.MaxInt64), w) results, err := idx.Search(t.Context(), query.MustCompile(`app.name EXISTS`), indexer.SearchOptions{OrderDesc: true}) diff --git a/sei-tendermint/internal/state/indexer/tx/kv/kv.go b/sei-tendermint/internal/state/indexer/tx/kv/kv.go index 9acf6b60d8..f62d89479d 100644 --- a/sei-tendermint/internal/state/indexer/tx/kv/kv.go +++ b/sei-tendermint/internal/state/indexer/tx/kv/kv.go @@ -643,7 +643,10 @@ func heightBounds(ranges []indexer.QueryRange) (lo, hi int64) { // whichever sub-range holds the higher (for desc) or lower (for asc) heights is // drained first, and the other only if the limit is not yet met. func (txi *TxIndex) searchHeightOrdered(ctx context.Context, plan heightOrderedPlan, opts indexer.SearchOptions) ([]*abci.TxResultV2, error) { - w := txi.readWatermark() + w, err := txi.readWatermark() + if err != nil { + return nil, err + } lo, hi := heightBounds(plan.heightRanges) if lo > hi { return []*abci.TxResultV2{}, nil @@ -1268,15 +1271,15 @@ func watermarkKey() []byte { // index. An unset watermark (fresh DB, or upgraded-but-not-yet-written) reads // as math.MaxInt64 so every height-ordered query takes the legacy fallback // until the new index has written at least one key. -func (txi *TxIndex) readWatermark() int64 { +func (txi *TxIndex) readWatermark() (int64, error) { bz, err := txi.store.Get(watermarkKey()) if err != nil { - panic(err) + return 0, err } if len(bz) == 0 { - return math.MaxInt64 + return math.MaxInt64, nil } - return int64FromBytes(bz) + return int64FromBytes(bz), nil } // updateWatermark lowers the persisted watermark to height when height is lower, @@ -1285,7 +1288,11 @@ func (txi *TxIndex) readWatermark() int64 { // over-conservative (routes covered heights to the fallback); a watermark below // the keys actually written would be incorrect, so it is never raised here. func (txi *TxIndex) updateWatermark(b dbm.Batch, height int64) error { - if height >= txi.readWatermark() { + w, err := txi.readWatermark() + if err != nil { + return err + } + if height >= w { return nil } return b.Set(watermarkKey(), int64ToBytes(height)) diff --git a/sei-tendermint/internal/state/indexer/tx/kv/kv_watermark_test.go b/sei-tendermint/internal/state/indexer/tx/kv/kv_watermark_test.go index 23e01fdf4f..885fa73539 100644 --- a/sei-tendermint/internal/state/indexer/tx/kv/kv_watermark_test.go +++ b/sei-tendermint/internal/state/indexer/tx/kv/kv_watermark_test.go @@ -121,7 +121,9 @@ func hoFixture(t *testing.T) *TxIndex { require.NoError(t, idx.Index([]*abci.TxResultV2{hoTx(h, i)})) } } - require.Equal(t, int64(hoWatermark), idx.readWatermark()) + w, err := idx.readWatermark() + require.NoError(t, err) + require.Equal(t, int64(hoWatermark), w) for h := int64(1); h < hoWatermark; h++ { for i := uint32(0); i < 2; i++ { @@ -129,7 +131,9 @@ func hoFixture(t *testing.T) *TxIndex { } } // Watermark must be unaffected by legacy-only writes. - require.Equal(t, int64(hoWatermark), idx.readWatermark()) + w, err = idx.readWatermark() + require.NoError(t, err) + require.Equal(t, int64(hoWatermark), w) return idx } @@ -189,7 +193,9 @@ func TestTxIndexDualWrite(t *testing.T) { require.True(t, hoHeight) // Watermark advanced to the indexed height. - require.Equal(t, int64(7), idx.readWatermark()) + w, err := idx.readWatermark() + require.NoError(t, err) + require.Equal(t, int64(7), w) } // TestTxSearchWatermarkSplit exercises the height-ordered split-merge across the @@ -238,7 +244,9 @@ func TestTxSearchWatermarkUnset(t *testing.T) { for h := int64(1); h <= 5; h++ { indexLegacyOnly(t, idx, hoTx(h, 0)) } - require.Equal(t, int64(math.MaxInt64), idx.readWatermark()) + w, err := idx.readWatermark() + require.NoError(t, err) + require.Equal(t, int64(math.MaxInt64), w) results, err := idx.Search(t.Context(), query.MustCompile(`app.name EXISTS`), indexer.SearchOptions{OrderDesc: true}) From 0dd47eb3ac56fb23e737190c161be05e939b1221 Mon Sep 17 00:00:00 2001 From: Amir Deris Date: Thu, 9 Jul 2026 11:19:54 -0700 Subject: [PATCH 4/8] Reject search on reserved keys --- sei-tendermint/internal/state/indexer/block/kv/kv.go | 11 +++++++++++ sei-tendermint/internal/state/indexer/tx/kv/kv.go | 11 +++++++++++ 2 files changed, 22 insertions(+) diff --git a/sei-tendermint/internal/state/indexer/block/kv/kv.go b/sei-tendermint/internal/state/indexer/block/kv/kv.go index 2aefb54d9e..5f60ea110c 100644 --- a/sei-tendermint/internal/state/indexer/block/kv/kv.go +++ b/sei-tendermint/internal/state/indexer/block/kv/kv.go @@ -147,6 +147,17 @@ func (idx *BlockerIndexer) Search(ctx context.Context, q *query.Query, opts inde conditions := q.Syntax() + // Reject queries that reference the reserved height-ordered / watermark + // prefixes as tags. Writes already reject these as event names, but the + // scan paths build their prefix directly from the query tag, so an + // unguarded EXISTS on one of them would iterate the entire reserved + // namespace and behave as a no-op instead of returning no matches. + for _, c := range conditions { + if c.Tag == blockHeightOrderedKey || c.Tag == blockWatermarkKey { + return nil, fmt.Errorf("tag %q is reserved and cannot be queried", c.Tag) + } + } + // If there is an exact height query, return the result immediately // (if it exists). height, ok := lookForHeight(conditions) diff --git a/sei-tendermint/internal/state/indexer/tx/kv/kv.go b/sei-tendermint/internal/state/indexer/tx/kv/kv.go index f62d89479d..8c957f40d9 100644 --- a/sei-tendermint/internal/state/indexer/tx/kv/kv.go +++ b/sei-tendermint/internal/state/indexer/tx/kv/kv.go @@ -190,6 +190,17 @@ func (txi *TxIndex) Search(ctx context.Context, q *query.Query, opts indexer.Sea // get a list of conditions (like "tx.height > 5") conditions := q.Syntax() + // Reject queries that reference the reserved height-ordered / watermark + // prefixes as tags. Writes already reject these as event names, but the + // scan paths build their prefix directly from the query tag, so an + // unguarded EXISTS on one of them would iterate the entire reserved + // namespace and behave as a no-op instead of returning no matches. + for _, c := range conditions { + if c.Tag == txHeightOrderedKey || c.Tag == txWatermarkKey { + return nil, fmt.Errorf("tag %q is reserved and cannot be queried", c.Tag) + } + } + // if there is a hash condition, return the result immediately hash, ok, err := lookForHash(conditions) if err != nil { From 2dbd3ee447011137cc452b0bfdef3c4832c941f0 Mon Sep 17 00:00:00 2001 From: Amir Deris Date: Thu, 9 Jul 2026 11:30:49 -0700 Subject: [PATCH 5/8] Fix block height exists bug --- sei-tendermint/internal/state/indexer/block/kv/kv.go | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/sei-tendermint/internal/state/indexer/block/kv/kv.go b/sei-tendermint/internal/state/indexer/block/kv/kv.go index 5f60ea110c..5644beb6de 100644 --- a/sei-tendermint/internal/state/indexer/block/kv/kv.go +++ b/sei-tendermint/internal/state/indexer/block/kv/kv.go @@ -592,6 +592,14 @@ func planHeightOrdered(conditions []syntax.Condition, ranges indexer.QueryRanges return heightOrderedPlan{}, false } + // block.height is never dual-written to the height-ordered namespace (it is + // the reserved primary key), so it cannot drive a height-ordered scan. Fall + // through to the intersect path, where EXISTS on the primary key correctly + // resolves to the full set instead of scanning an empty prefix. + if nonRange[0].Tag == types.BlockHeightKey { + return heightOrderedPlan{}, false + } + plan.driverTag = nonRange[0].Tag plan.existsCond = &nonRange[0] return plan, true From 967e4de17420c40f62d3ae56829437a94d5ab2fd Mon Sep 17 00:00:00 2001 From: Amir Deris Date: Thu, 9 Jul 2026 11:59:20 -0700 Subject: [PATCH 6/8] Used shared budget for both fast leg and legacy index --- .../internal/state/indexer/block/kv/kv.go | 43 ++++++++++++------- .../internal/state/indexer/tx/kv/kv.go | 43 ++++++++++++------- 2 files changed, 56 insertions(+), 30 deletions(-) diff --git a/sei-tendermint/internal/state/indexer/block/kv/kv.go b/sei-tendermint/internal/state/indexer/block/kv/kv.go index 5644beb6de..010eb9ea45 100644 --- a/sei-tendermint/internal/state/indexer/block/kv/kv.go +++ b/sei-tendermint/internal/state/indexer/block/kv/kv.go @@ -646,6 +646,12 @@ func (idx *BlockerIndexer) searchHeightOrdered(ctx context.Context, plan heightO fbHi := min(hi, w-1) hasFallback := lo <= fbHi + // One scan budget is shared across both legs so MaxScan bounds the total + // work of the query regardless of order_by, and so the height-ordered fast + // leg is charged too — otherwise a query with the result cap disabled + // (max-tx-search-results = 0) could stream an entire tag prefix unbounded. + budget := indexer.NewScanBudget(opts.MaxScan) + results := make([]int64, 0, indexer.BoundedCap(opts.Limit)) remaining := func() int { if opts.Limit <= 0 { @@ -659,14 +665,14 @@ func (idx *BlockerIndexer) searchHeightOrdered(ctx context.Context, plan heightO if opts.OrderDesc { if hasFast { - r, err := idx.scanHeightOrderedFast(ctx, plan, fastLo, hi, true, remaining()) + r, err := idx.scanHeightOrderedFast(ctx, plan, fastLo, hi, true, remaining(), budget) if err != nil { return nil, err } results = append(results, r...) } if !reachedLimit() && hasFallback { - r, err := idx.heightOrderedFallback(ctx, plan, lo, fbHi, true, remaining(), opts.MaxScan) + r, err := idx.heightOrderedFallback(ctx, plan, lo, fbHi, true, remaining(), budget) if err != nil { return nil, err } @@ -674,14 +680,14 @@ func (idx *BlockerIndexer) searchHeightOrdered(ctx context.Context, plan heightO } } else { if hasFallback { - r, err := idx.heightOrderedFallback(ctx, plan, lo, fbHi, false, remaining(), opts.MaxScan) + r, err := idx.heightOrderedFallback(ctx, plan, lo, fbHi, false, remaining(), budget) if err != nil { return nil, err } results = append(results, r...) } if !reachedLimit() && hasFast { - r, err := idx.scanHeightOrderedFast(ctx, plan, fastLo, hi, false, remaining()) + r, err := idx.scanHeightOrderedFast(ctx, plan, fastLo, hi, false, remaining(), budget) if err != nil { return nil, err } @@ -698,9 +704,10 @@ func (idx *BlockerIndexer) searchHeightOrdered(ctx context.Context, plan heightO // scanHeightOrderedFast streams the new height-ordered index for plan.driverTag // over heights [lo, hi] in order_by height order, deduping heights and stopping // at limit (limit <= 0 means unbounded). Because keys are height-major the scan -// early-stops once it passes the far bound. This is a Limit-bounded fast-path -// scan and is deliberately not charged against the scan budget. -func (idx *BlockerIndexer) scanHeightOrderedFast(ctx context.Context, plan heightOrderedPlan, lo, hi int64, desc bool, limit int) ([]int64, error) { +// early-stops once it passes the far bound. Each examined entry is charged +// against the shared budget so the scan fails closed on a broad query even when +// the result cap is disabled. +func (idx *BlockerIndexer) scanHeightOrderedFast(ctx context.Context, plan heightOrderedPlan, lo, hi int64, desc bool, limit int, budget *indexer.ScanBudget) ([]int64, error) { prefix, err := prefixHeightOrdered(plan.driverTag) if err != nil { return nil, err @@ -719,6 +726,12 @@ func (idx *BlockerIndexer) scanHeightOrderedFast(ctx context.Context, plan heigh break } + // Charge each examined entry against the shared budget so a broad scan + // fails closed even when the result cap is disabled. + if err := budget.Step(); err != nil { + return nil, err + } + h, err := parseHeightFromHeightOrderedKey(it.Key()) if err != nil { continue @@ -769,14 +782,14 @@ func (idx *BlockerIndexer) scanHeightOrderedFast(ctx context.Context, plan heigh return results, nil } -// heightOrderedFallback serves the pre-watermark [lo, hi] leg of a -// heightOrderedPlan from the legacy index. It rebuilds the equivalent legacy -// EXISTS match, then collects heights in [lo, hi], ordered and capped at limit. -// The legacy scan is charged against a maxScan budget and fails closed if the -// budget is exceeded. -func (idx *BlockerIndexer) heightOrderedFallback(ctx context.Context, plan heightOrderedPlan, lo, hi int64, desc bool, limit, maxScan int) ([]int64, error) { - budget := indexer.NewScanBudget(maxScan) - +// heightOrderedFallback serves the pre-watermark [lo, hi] leg from the legacy +// index, ordered and capped at limit, charged against the shared budget. +// +// The legacy index is value-ordered, so heights aren't seekable: this scans the +// whole tag prefix (including >= W entries the fast leg already served) and +// discards out-of-range ones. Budget pressure from the discards is a +// transitional-window effect that a reindex removes. +func (idx *BlockerIndexer) heightOrderedFallback(ctx context.Context, plan heightOrderedPlan, lo, hi int64, desc bool, limit int, budget *indexer.ScanBudget) ([]int64, error) { filtered, err := idx.match(ctx, *plan.existsCond, nil, map[string][]byte{}, true, budget) if err != nil { return nil, err diff --git a/sei-tendermint/internal/state/indexer/tx/kv/kv.go b/sei-tendermint/internal/state/indexer/tx/kv/kv.go index 8c957f40d9..d29ed7c59b 100644 --- a/sei-tendermint/internal/state/indexer/tx/kv/kv.go +++ b/sei-tendermint/internal/state/indexer/tx/kv/kv.go @@ -669,6 +669,12 @@ func (txi *TxIndex) searchHeightOrdered(ctx context.Context, plan heightOrderedP fbHi := min(hi, w-1) hasFallback := lo <= fbHi + // One scan budget is shared across both legs so MaxScan bounds the total + // work of the query regardless of order_by, and so the height-ordered fast + // leg is charged too — otherwise a query with the result cap disabled + // (max-tx-search-results = 0) could stream an entire tag prefix unbounded. + budget := indexer.NewScanBudget(opts.MaxScan) + results := make([]*abci.TxResultV2, 0, indexer.BoundedCap(opts.Limit)) remaining := func() int { if opts.Limit <= 0 { @@ -682,14 +688,14 @@ func (txi *TxIndex) searchHeightOrdered(ctx context.Context, plan heightOrderedP if opts.OrderDesc { if hasFast { - r, err := txi.scanHeightOrderedFast(ctx, plan, fastLo, hi, true, remaining()) + r, err := txi.scanHeightOrderedFast(ctx, plan, fastLo, hi, true, remaining(), budget) if err != nil { return nil, err } results = append(results, r...) } if !reachedLimit() && hasFallback { - r, err := txi.heightOrderedFallback(ctx, plan, lo, fbHi, true, remaining(), opts.MaxScan) + r, err := txi.heightOrderedFallback(ctx, plan, lo, fbHi, true, remaining(), budget) if err != nil { return nil, err } @@ -697,14 +703,14 @@ func (txi *TxIndex) searchHeightOrdered(ctx context.Context, plan heightOrderedP } } else { if hasFallback { - r, err := txi.heightOrderedFallback(ctx, plan, lo, fbHi, false, remaining(), opts.MaxScan) + r, err := txi.heightOrderedFallback(ctx, plan, lo, fbHi, false, remaining(), budget) if err != nil { return nil, err } results = append(results, r...) } if !reachedLimit() && hasFast { - r, err := txi.scanHeightOrderedFast(ctx, plan, fastLo, hi, false, remaining()) + r, err := txi.scanHeightOrderedFast(ctx, plan, fastLo, hi, false, remaining(), budget) if err != nil { return nil, err } @@ -721,9 +727,10 @@ func (txi *TxIndex) searchHeightOrdered(ctx context.Context, plan heightOrderedP // scanHeightOrderedFast streams the new height-ordered index for plan.driverTag // over heights [lo, hi] in (height, index) order_by order, deduping by hash and // stopping at limit (limit <= 0 means unbounded). Because keys are height-major -// the scan early-stops once it passes the far bound. This is a Limit-bounded -// fast-path scan and is deliberately not charged against the scan budget. -func (txi *TxIndex) scanHeightOrderedFast(ctx context.Context, plan heightOrderedPlan, lo, hi int64, desc bool, limit int) ([]*abci.TxResultV2, error) { +// the scan early-stops once it passes the far bound. Each examined entry is +// charged against the shared budget so the scan fails closed on a broad query +// even when the result cap is disabled. +func (txi *TxIndex) scanHeightOrderedFast(ctx context.Context, plan heightOrderedPlan, lo, hi int64, desc bool, limit int, budget *indexer.ScanBudget) ([]*abci.TxResultV2, error) { it, err := txi.prefixIterator(prefixHeightOrdered(plan.driverTag), desc) if err != nil { return nil, fmt.Errorf("failed to create height-ordered iterator: %w", err) @@ -738,6 +745,12 @@ func (txi *TxIndex) scanHeightOrderedFast(ctx context.Context, plan heightOrdere break } + // Charge each examined entry against the shared budget so a broad scan + // fails closed even when the result cap is disabled. + if err := budget.Step(); err != nil { + return nil, err + } + height, _, err := parseHeightOrderedKey(it.Key()) if err != nil { continue @@ -789,14 +802,14 @@ func (txi *TxIndex) scanHeightOrderedFast(ctx context.Context, plan heightOrdere return results, nil } -// heightOrderedFallback serves the pre-watermark [lo, hi] leg of a -// heightOrderedPlan from the legacy index. It rebuilds the equivalent legacy -// match, then collects results whose height is in [lo, hi], ordered and capped -// at limit. The legacy scan is charged against a maxScan budget and fails -// closed if the budget is exceeded. -func (txi *TxIndex) heightOrderedFallback(ctx context.Context, plan heightOrderedPlan, lo, hi int64, desc bool, limit, maxScan int) ([]*abci.TxResultV2, error) { - budget := indexer.NewScanBudget(maxScan) - +// heightOrderedFallback serves the pre-watermark [lo, hi] leg from the legacy +// index, ordered and capped at limit, charged against the shared budget. +// +// The legacy index is value-ordered, so heights aren't seekable: this scans the +// whole tag prefix (including >= W entries the fast leg already served) and +// discards out-of-range ones. Budget pressure from the discards is a +// transitional-window effect that a reindex removes. +func (txi *TxIndex) heightOrderedFallback(ctx context.Context, plan heightOrderedPlan, lo, hi int64, desc bool, limit int, budget *indexer.ScanBudget) ([]*abci.TxResultV2, error) { var ( filtered map[string][]byte err error From 26d5a1a5381c34daf35225910b2f61cfafcbcc06 Mon Sep 17 00:00:00 2001 From: Amir Deris Date: Thu, 9 Jul 2026 12:35:40 -0700 Subject: [PATCH 7/8] Seek the height-ordered fast scan to the [lo, hi] key bounds instead of scanning the full tag prefix --- .../internal/state/indexer/block/kv/kv.go | 39 ++++------ .../internal/state/indexer/block/kv/util.go | 28 +++++++ .../internal/state/indexer/tx/kv/kv.go | 75 ++++++++----------- 3 files changed, 74 insertions(+), 68 deletions(-) diff --git a/sei-tendermint/internal/state/indexer/block/kv/kv.go b/sei-tendermint/internal/state/indexer/block/kv/kv.go index 010eb9ea45..fee4092f89 100644 --- a/sei-tendermint/internal/state/indexer/block/kv/kv.go +++ b/sei-tendermint/internal/state/indexer/block/kv/kv.go @@ -703,16 +703,21 @@ func (idx *BlockerIndexer) searchHeightOrdered(ctx context.Context, plan heightO // scanHeightOrderedFast streams the new height-ordered index for plan.driverTag // over heights [lo, hi] in order_by height order, deduping heights and stopping -// at limit (limit <= 0 means unbounded). Because keys are height-major the scan -// early-stops once it passes the far bound. Each examined entry is charged -// against the shared budget so the scan fails closed on a broad query even when -// the result cap is disabled. +// at limit (limit <= 0 means unbounded). The iterator is seeked to the [lo, hi] +// key bounds, so out-of-window entries are never scanned; each examined entry is +// charged against the shared budget, which therefore counts only in-window work +// and fails closed on a broad query even when the result cap is disabled. func (idx *BlockerIndexer) scanHeightOrderedFast(ctx context.Context, plan heightOrderedPlan, lo, hi int64, desc bool, limit int, budget *indexer.ScanBudget) ([]int64, error) { - prefix, err := prefixHeightOrdered(plan.driverTag) + start, end, err := heightOrderedBounds(plan.driverTag, lo, hi) if err != nil { return nil, err } - it, err := idx.prefixIterator(prefix, desc) + var it dbm.Iterator + if desc { + it, err = idx.store.ReverseIterator(start, end) + } else { + it, err = idx.store.Iterator(start, end) + } if err != nil { return nil, fmt.Errorf("failed to create height-ordered iterator: %w", err) } @@ -726,8 +731,8 @@ func (idx *BlockerIndexer) scanHeightOrderedFast(ctx context.Context, plan heigh break } - // Charge each examined entry against the shared budget so a broad scan - // fails closed even when the result cap is disabled. + // The iterator is bounded to [lo, hi], so every entry is in-window; the + // budget therefore only counts in-window work. if err := budget.Step(); err != nil { return nil, err } @@ -737,24 +742,6 @@ func (idx *BlockerIndexer) scanHeightOrderedFast(ctx context.Context, plan heigh continue } - // Keys are height-major, so once we pass the far bound in scan order we - // can stop; the near bound is skipped until we reach the window. - if desc { - if h > hi { - continue - } - if h < lo { - break - } - } else { - if h < lo { - continue - } - if h > hi { - break - } - } - if _, dup := seen[h]; dup { continue } diff --git a/sei-tendermint/internal/state/indexer/block/kv/util.go b/sei-tendermint/internal/state/indexer/block/kv/util.go index 1facde47d2..5c1d623fc3 100644 --- a/sei-tendermint/internal/state/indexer/block/kv/util.go +++ b/sei-tendermint/internal/state/indexer/block/kv/util.go @@ -3,11 +3,13 @@ package kv import ( "encoding/binary" "fmt" + "math" "strconv" "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" ) @@ -72,6 +74,32 @@ func prefixHeightOrdered(compositeKey string) ([]byte, error) { return orderedcode.Append(nil, blockHeightOrderedKey, compositeKey) } +// heightOrderedBounds returns the [start, end) key range restricting a +// height-ordered scan of compositeKey to heights [lo, hi]. Because keys are +// orderedcode(blockHeightOrderedKey, compositeKey, height, ...), start seeks to +// the first key at height lo and end is the first key past height hi, so the +// scan visits only in-window entries instead of scanning the whole prefix. When +// hi is unbounded (math.MaxInt64) end is the prefix upper bound, avoiding +// overflow. +func heightOrderedBounds(compositeKey string, lo, hi int64) (start, end []byte, err error) { + start, err = orderedcode.Append(nil, blockHeightOrderedKey, compositeKey, lo) + if err != nil { + return nil, nil, err + } + if hi == math.MaxInt64 { + prefix, err := prefixHeightOrdered(compositeKey) + if err != nil { + return nil, nil, err + } + return start, indexer.PrefixUpperBound(prefix), nil + } + end, err = orderedcode.Append(nil, blockHeightOrderedKey, compositeKey, hi+1) + if err != nil { + return nil, nil, err + } + return start, end, nil +} + // parseHeightFromHeightOrderedKey extracts the height from a height-ordered // event key. See eventKeyHeightOrdered for the layout. func parseHeightFromHeightOrderedKey(key []byte) (int64, error) { diff --git a/sei-tendermint/internal/state/indexer/tx/kv/kv.go b/sei-tendermint/internal/state/indexer/tx/kv/kv.go index d29ed7c59b..0550c7dd46 100644 --- a/sei-tendermint/internal/state/indexer/tx/kv/kv.go +++ b/sei-tendermint/internal/state/indexer/tx/kv/kv.go @@ -726,12 +726,22 @@ func (txi *TxIndex) searchHeightOrdered(ctx context.Context, plan heightOrderedP // scanHeightOrderedFast streams the new height-ordered index for plan.driverTag // over heights [lo, hi] in (height, index) order_by order, deduping by hash and -// stopping at limit (limit <= 0 means unbounded). Because keys are height-major -// the scan early-stops once it passes the far bound. Each examined entry is -// charged against the shared budget so the scan fails closed on a broad query -// even when the result cap is disabled. +// stopping at limit (limit <= 0 means unbounded). The iterator is seeked to the +// [lo, hi] key bounds, so out-of-window entries are never scanned; each examined +// entry is charged against the shared budget, which therefore counts only +// in-window work and fails closed on a broad query even when the result cap is +// disabled. func (txi *TxIndex) scanHeightOrderedFast(ctx context.Context, plan heightOrderedPlan, lo, hi int64, desc bool, limit int, budget *indexer.ScanBudget) ([]*abci.TxResultV2, error) { - it, err := txi.prefixIterator(prefixHeightOrdered(plan.driverTag), desc) + start, end := heightOrderedBounds(plan.driverTag, lo, hi) + var ( + it dbm.Iterator + err error + ) + if desc { + it, err = txi.store.ReverseIterator(start, end) + } else { + it, err = txi.store.Iterator(start, end) + } if err != nil { return nil, fmt.Errorf("failed to create height-ordered iterator: %w", err) } @@ -745,35 +755,12 @@ func (txi *TxIndex) scanHeightOrderedFast(ctx context.Context, plan heightOrdere break } - // Charge each examined entry against the shared budget so a broad scan - // fails closed even when the result cap is disabled. + // The iterator is bounded to [lo, hi], so every entry is in-window; the + // budget therefore only counts in-window work. if err := budget.Step(); err != nil { return nil, err } - height, _, err := parseHeightOrderedKey(it.Key()) - if err != nil { - continue - } - - // Keys are height-major, so once we pass the far bound in scan order we - // can stop; the near bound is skipped until we reach the window. - if desc { - if height > hi { - continue - } - if height < lo { - break - } - } else { - if height < lo { - continue - } - if height > hi { - break - } - } - hash := it.Value() if _, dup := seen[string(hash)]; dup { continue @@ -1264,21 +1251,25 @@ func prefixHeightOrdered(compositeKey string) []byte { return key } -// parseHeightOrderedKey extracts (height, index) from a height-ordered event -// key. See secondaryKeyHeightOrdered for the layout. -func parseHeightOrderedKey(key []byte) (int64, uint32, error) { - var ( - ns, compositeKey, value string - height, index int64 - ) - remaining, err := orderedcode.Parse(string(key), &ns, &compositeKey, &height, &index, &value) +// heightOrderedBounds returns the [start, end) key range restricting a +// height-ordered scan of compositeKey to heights [lo, hi]. Because keys are +// orderedcode(txHeightOrderedKey, compositeKey, height, ...), start seeks to the +// first key at height lo and end is the first key past height hi, so the scan +// visits only in-window entries instead of scanning the whole prefix. When hi is +// unbounded (math.MaxInt64) end is the prefix upper bound, avoiding overflow. +func heightOrderedBounds(compositeKey string, lo, hi int64) (start, end []byte) { + start, err := orderedcode.Append(nil, txHeightOrderedKey, compositeKey, lo) if err != nil { - return 0, 0, fmt.Errorf("failed to parse height-ordered key: %w", err) + panic(err) } - if len(remaining) != 0 { - return 0, 0, fmt.Errorf("unexpected remainder in key: %s", remaining) + if hi == math.MaxInt64 { + return start, indexer.PrefixUpperBound(prefixHeightOrdered(compositeKey)) + } + end, err = orderedcode.Append(nil, txHeightOrderedKey, compositeKey, hi+1) + if err != nil { + panic(err) } - return height, uint32(index), nil //nolint:gosec // index is stored as int64(uint32), so the round-trip cannot overflow + return start, end } // watermarkKey is the reserved key holding the lowest height covered by the From 38d852a001cb28e277b51ffdf472403a7d14cd23 Mon Sep 17 00:00:00 2001 From: Amir Deris Date: Thu, 9 Jul 2026 15:26:34 -0700 Subject: [PATCH 8/8] Bound the equality/primary-key fast path scan by height window and MaxScan searchBounded (tx and block) previously iterated the entire driver prefix and stopped only at opts.Limit: a sparse equality with a narrow, far-from-edge height window walked the whole prefix, and a common equality with a rare probe had no MaxScan fail-closed. Since the legacy secondary key places the numeric height right after the (tag, value) prefix, seek the iterator to the [lo, hi] window (early-stop for free at the far edge) and charge every examined entry against opts.MaxScan. Also promote the duplicated heightBounds helper to indexer.HeightBounds and drop the now-unused prefixIterator methods. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../internal/state/indexer/block/kv/kv.go | 78 ++++++++++--------- .../internal/state/indexer/tx/kv/kv.go | 66 ++++++++-------- .../internal/state/indexer/utils.go | 22 ++++++ 3 files changed, 94 insertions(+), 72 deletions(-) diff --git a/sei-tendermint/internal/state/indexer/block/kv/kv.go b/sei-tendermint/internal/state/indexer/block/kv/kv.go index fee4092f89..2228a720c8 100644 --- a/sei-tendermint/internal/state/indexer/block/kv/kv.go +++ b/sei-tendermint/internal/state/indexer/block/kv/kv.go @@ -367,29 +367,58 @@ func (idx *BlockerIndexer) collectBoundedInHeightRange(ctx context.Context, filt // 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. +// and stops at opts.Limit. The iterator is seeked to the [lo, hi] height window +// (both driver keys place the numeric height right after the prefix), and every +// examined entry is charged against opts.MaxScan so a broad prefix with sparse +// matches fails closed. func (idx *BlockerIndexer) searchBounded(ctx context.Context, plan boundedPlan, opts indexer.SearchOptions) ([]int64, error) { var ( - prefix []byte - err error + base []byte + err error ) if plan.driverEquality != nil { - prefix, err = orderedcode.Append(nil, plan.driverEquality.Tag, plan.driverEquality.Arg.Value()) + base, 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) + base, err = orderedcode.Append(nil, types.BlockHeightKey) } if err != nil { return nil, fmt.Errorf("failed to create driver prefix key: %w", err) } - it, err := idx.prefixIterator(prefix, opts.OrderDesc) + // appendHeight appends an encoded height to a copy of base so seeking never + // mutates the shared prefix bytes. + appendHeight := func(h int64) ([]byte, error) { + buf := make([]byte, len(base)) + copy(buf, base) + return orderedcode.Append(buf, h) + } + + lo, hi := indexer.HeightBounds(plan.heightRanges) + start, err := appendHeight(lo) + if err != nil { + return nil, fmt.Errorf("failed to create driver lower bound key: %w", err) + } + var end []byte + if hi == math.MaxInt64 { + end = indexer.PrefixUpperBound(base) + } else if end, err = appendHeight(hi + 1); err != nil { + return nil, fmt.Errorf("failed to create driver upper bound key: %w", err) + } + + var it dbm.Iterator + if opts.OrderDesc { + it, err = idx.store.ReverseIterator(start, end) + } else { + it, err = idx.store.Iterator(start, end) + } if err != nil { return nil, fmt.Errorf("failed to create driver iterator: %w", err) } defer func() { _ = it.Close() }() + budget := indexer.NewScanBudget(opts.MaxScan) + results := make([]int64, 0, indexer.BoundedCap(opts.Limit)) seen := make(map[int64]struct{}) @@ -398,6 +427,10 @@ func (idx *BlockerIndexer) searchBounded(ctx context.Context, plan boundedPlan, break } + if err := budget.Step(); err != nil { + return nil, err + } + h := int64FromBytes(it.Value()) if _, dup := seen[h]; dup { continue @@ -454,15 +487,6 @@ func (idx *BlockerIndexer) candidateMatches(h int64, plan boundedPlan) (bool, er return idx.Has(h) } -// prefixIterator returns an iterator over the given key prefix. When desc is -// true it iterates in descending (most-recent-height-first) order. -func (idx *BlockerIndexer) prefixIterator(prefix []byte, desc bool) (dbm.Iterator, error) { - if !desc { - return dbm.IteratePrefix(idx.store, prefix) - } - return idx.store.ReverseIterator(prefix, indexer.PrefixUpperBound(prefix)) -} - // hasEvent reports whether the block at the given height has an indexed event // matching compositeKey=eventValue, regardless of the event type suffix. func (idx *BlockerIndexer) hasEvent(compositeKey, eventValue string, height int64) (bool, error) { @@ -605,26 +629,6 @@ func planHeightOrdered(conditions []syntax.Condition, ranges indexer.QueryRanges return plan, true } -// heightBounds reduces the plan's block.height ranges to a single inclusive -// [lo, hi] window. Heights are >= 1, so an absent lower bound defaults to 1 -// (avoiding a pointless legacy scan for a non-existent sub-watermark tail). -func heightBounds(ranges []indexer.QueryRange) (lo, hi int64) { - lo, hi = 1, int64(math.MaxInt64) - for i := range ranges { - if lb := ranges[i].LowerBoundValue(); lb != nil { - if v, ok := lb.(int64); ok && v > lo { - lo = v - } - } - if ub := ranges[i].UpperBoundValue(); ub != nil { - if v, ok := ub.(int64); ok && v < hi { - hi = v - } - } - } - return lo, hi -} - // searchHeightOrdered serves a heightOrderedPlan by splitting the [lo, hi] // window at the watermark W: heights in [max(lo, W), hi] are streamed from the // new height-ordered index (fast, early-stops at opts.Limit); heights in @@ -635,7 +639,7 @@ func (idx *BlockerIndexer) searchHeightOrdered(ctx context.Context, plan heightO if err != nil { return nil, err } - lo, hi := heightBounds(plan.heightRanges) + lo, hi := indexer.HeightBounds(plan.heightRanges) if lo > hi { return []int64{}, nil } diff --git a/sei-tendermint/internal/state/indexer/tx/kv/kv.go b/sei-tendermint/internal/state/indexer/tx/kv/kv.go index 0550c7dd46..2c3bfcfed5 100644 --- a/sei-tendermint/internal/state/indexer/tx/kv/kv.go +++ b/sei-tendermint/internal/state/indexer/tx/kv/kv.go @@ -392,18 +392,39 @@ func (txi *TxIndex) collectBoundedInHeightRange(ctx context.Context, filteredHas // 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. +// point-probes the remaining conditions per candidate, and stops at opts.Limit. +// The iterator is seeked to the [lo, hi] height window (the key suffix begins +// with the numeric height), and every examined entry is charged against +// opts.MaxScan so a broad prefix with sparse matches fails closed. func (txi *TxIndex) searchBounded(ctx context.Context, plan boundedPlan, opts indexer.SearchOptions) ([]*abci.TxResultV2, error) { - prefix := prefixFromCompositeKeyAndValue(plan.driverEquality.Tag, plan.driverEquality.Arg.Value()) + tag := plan.driverEquality.Tag + value := plan.driverEquality.Arg.Value() - it, err := txi.prefixIterator(prefix, opts.OrderDesc) + lo, hi := indexer.HeightBounds(plan.heightRanges) + start, err := orderedcode.Append(nil, tag, value, lo) + if err != nil { + return nil, fmt.Errorf("failed to create driver lower bound key: %w", err) + } + var end []byte + if hi == math.MaxInt64 { + end = indexer.PrefixUpperBound(prefixFromCompositeKeyAndValue(tag, value)) + } else if end, err = orderedcode.Append(nil, tag, value, hi+1); err != nil { + return nil, fmt.Errorf("failed to create driver upper bound key: %w", err) + } + + var it dbm.Iterator + if opts.OrderDesc { + it, err = txi.store.ReverseIterator(start, end) + } else { + it, err = txi.store.Iterator(start, end) + } if err != nil { return nil, fmt.Errorf("failed to create driver iterator: %w", err) } defer func() { _ = it.Close() }() + budget := indexer.NewScanBudget(opts.MaxScan) + results := make([]*abci.TxResultV2, 0, indexer.BoundedCap(opts.Limit)) seen := make(map[string]struct{}) @@ -412,6 +433,10 @@ func (txi *TxIndex) searchBounded(ctx context.Context, plan boundedPlan, opts in break } + if err := budget.Step(); err != nil { + return nil, err + } + hash := it.Value() if _, dup := seen[string(hash)]; dup { continue @@ -478,15 +503,6 @@ func (txi *TxIndex) candidateMatches(height int64, index uint32, plan boundedPla 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 @@ -626,26 +642,6 @@ func planHeightOrdered(conditions []syntax.Condition, ranges indexer.QueryRanges return plan, true } -// heightBounds reduces the plan's tx.height ranges to a single inclusive -// [lo, hi] window. Heights are >= 1, so an absent lower bound defaults to 1 -// (avoiding a pointless legacy scan for a non-existent sub-watermark tail). -func heightBounds(ranges []indexer.QueryRange) (lo, hi int64) { - lo, hi = 1, int64(math.MaxInt64) - for i := range ranges { - if lb := ranges[i].LowerBoundValue(); lb != nil { - if v, ok := lb.(int64); ok && v > lo { - lo = v - } - } - if ub := ranges[i].UpperBoundValue(); ub != nil { - if v, ok := ub.(int64); ok && v < hi { - hi = v - } - } - } - return lo, hi -} - // searchHeightOrdered serves a heightOrderedPlan by splitting the [lo, hi] // window at the watermark W: heights in [max(lo, W), hi] are streamed from the // new height-ordered index (fast, early-stops at opts.Limit); heights in @@ -658,7 +654,7 @@ func (txi *TxIndex) searchHeightOrdered(ctx context.Context, plan heightOrderedP if err != nil { return nil, err } - lo, hi := heightBounds(plan.heightRanges) + lo, hi := indexer.HeightBounds(plan.heightRanges) if lo > hi { return []*abci.TxResultV2{}, nil } diff --git a/sei-tendermint/internal/state/indexer/utils.go b/sei-tendermint/internal/state/indexer/utils.go index 196f8cb728..a9828371b0 100644 --- a/sei-tendermint/internal/state/indexer/utils.go +++ b/sei-tendermint/internal/state/indexer/utils.go @@ -1,5 +1,7 @@ package indexer +import "math" + // 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. @@ -34,6 +36,26 @@ func HeightInRange(h int64, qr QueryRange) bool { return true } +// HeightBounds reduces a set of numeric height ranges to a single inclusive +// [lo, hi] window. Heights are >= 1, so an absent lower bound defaults to 1 and +// an absent upper bound to math.MaxInt64. +func HeightBounds(ranges []QueryRange) (lo, hi int64) { + lo, hi = 1, int64(math.MaxInt64) + for i := range ranges { + if lb := ranges[i].LowerBoundValue(); lb != nil { + if v, ok := lb.(int64); ok && v > lo { + lo = v + } + } + if ub := ranges[i].UpperBoundValue(); ub != nil { + if v, ok := ub.(int64); ok && v < hi { + hi = v + } + } + } + return lo, hi +} + // ScanBudget bounds the number of index entries a fallback scan may examine // across all passes of a single query (one query = one shared budget). It // counts work, not output: it is stepped once per iterator advance on the