diff --git a/sei-tendermint/config/config.go b/sei-tendermint/config/config.go index 633c626ce0..81407d3de9 100644 --- a/sei-tendermint/config/config.go +++ b/sei-tendermint/config/config.go @@ -536,6 +536,11 @@ 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. 0 disables the budget. + MaxEventSearchScan int `mapstructure:"max-event-search-scan"` } // DefaultRPCConfig returns a default configuration for the RPC server @@ -570,6 +575,7 @@ func DefaultRPCConfig() *RPCConfig { TimeoutWrite: 30 * time.Second, MaxTxSearchResults: 10_000, + MaxEventSearchScan: 50_000, } } @@ -625,6 +631,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..5fca7bf367 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 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 }} + ####################################################################### ### 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..2228a720c8 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,44 @@ 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 fallback path +// until the new index has written at least one key. +func (idx *BlockerIndexer) readWatermark() (int64, error) { + key, err := watermarkKey() + if err != nil { + return 0, err + } + bz, err := idx.store.Get(key) + if err != nil { + return 0, err + } + if len(bz) == 0 { + return math.MaxInt64, nil + } + 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). +func (idx *BlockerIndexer) updateWatermark(batch dbm.Batch, height int64) error { + w, err := idx.readWatermark() + if err != nil { + return err + } + if height >= w { + 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 +124,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() } @@ -86,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) @@ -109,16 +181,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 for height order index. 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 +218,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 +236,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 +249,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 +269,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 +282,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,31 +327,98 @@ 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 -// 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{}) @@ -276,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 @@ -332,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) { @@ -424,6 +570,225 @@ 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 + } + + // 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 +} + +// 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, err := idx.readWatermark() + if err != nil { + return nil, err + } + lo, hi := indexer.HeightBounds(plan.heightRanges) + if lo > hi { + return []int64{}, nil + } + + fastLo := max(lo, w) + hasFast := fastLo <= hi + + 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 { + 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(), budget) + if err != nil { + return nil, err + } + results = append(results, r...) + } + if !reachedLimit() && hasFallback { + r, err := idx.heightOrderedFallback(ctx, plan, lo, fbHi, true, remaining(), budget) + if err != nil { + return nil, err + } + results = append(results, r...) + } + } else { + if hasFallback { + 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(), budget) + 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). 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) { + start, end, err := heightOrderedBounds(plan.driverTag, lo, hi) + if err != nil { + return nil, err + } + 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) + } + 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 + } + + // 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 + } + + h, err := parseHeightFromHeightOrderedKey(it.Key()) + if err != nil { + continue + } + + 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 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 + } + + 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 +801,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 +822,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 +914,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 +934,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 +962,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 +993,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 +1030,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 +1102,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..797b415ed8 --- /dev/null +++ b/sei-tendermint/internal/state/indexer/block/kv/kv_watermark_test.go @@ -0,0 +1,257 @@ +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))) + } + 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)) + } + w, err = idx.readWatermark() + require.NoError(t, err) + require.Equal(t, int64(hoWatermark), w) + 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)) + } + 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}) + 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..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" ) @@ -50,6 +52,79 @@ 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) +} + +// 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) { + 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..2c3bfcfed5 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 } } @@ -153,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 { @@ -177,18 +225,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 +263,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 +275,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 +306,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,20 +360,71 @@ 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 -// 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() + + 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) + } - it, err := txi.prefixIterator(prefix, opts.OrderDesc) + 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{}) @@ -306,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 @@ -372,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 @@ -459,6 +581,241 @@ 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 +} + +// 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, err := txi.readWatermark() + if err != nil { + return nil, err + } + lo, hi := indexer.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 + + // 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 { + 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(), budget) + if err != nil { + return nil, err + } + results = append(results, r...) + } + if !reachedLimit() && hasFallback { + r, err := txi.heightOrderedFallback(ctx, plan, lo, fbHi, true, remaining(), budget) + if err != nil { + return nil, err + } + results = append(results, r...) + } + } else { + if hasFallback { + 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(), budget) + 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). 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) { + 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) + } + 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 + } + + // 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 + } + + 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 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 + ) + 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 +868,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 +882,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 +901,7 @@ func (txi *TxIndex) match( } } if err := it.Error(); err != nil { - panic(err) + return nil, err } case syntax.TExists: @@ -548,12 +909,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 +928,7 @@ func (txi *TxIndex) match( } } if err := it.Error(); err != nil { - panic(err) + return nil, err } case syntax.TContains: @@ -573,12 +937,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 +962,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 +993,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 +1007,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 +1025,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 +1039,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 +1053,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 +1101,7 @@ iter: } } if err := it.Error(); err != nil { - panic(err) + return nil, err } if len(tmpHashes) == 0 || firstRun { @@ -738,7 +1112,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 +1130,7 @@ iter: } } - return filteredHashes + return filteredHashes, nil } // ########################## Keys ############################# @@ -827,6 +1201,114 @@ 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 +} + +// 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 { + panic(err) + } + 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 start, end +} + +// 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, error) { + bz, err := txi.store.Get(watermarkKey()) + if err != nil { + return 0, err + } + if len(bz) == 0 { + return math.MaxInt64, nil + } + 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. +func (txi *TxIndex) updateWatermark(b dbm.Batch, height int64) error { + w, err := txi.readWatermark() + if err != nil { + return err + } + if height >= w { + 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..885fa73539 --- /dev/null +++ b/sei-tendermint/internal/state/indexer/tx/kv/kv_watermark_test.go @@ -0,0 +1,342 @@ +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)})) + } + } + 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++ { + indexLegacyOnly(t, idx, hoTx(h, i)) + } + } + // Watermark must be unaffected by legacy-only writes. + w, err = idx.readWatermark() + require.NoError(t, err) + require.Equal(t, int64(hoWatermark), w) + 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. + w, err := idx.readWatermark() + require.NoError(t, err) + require.Equal(t, int64(7), w) +} + +// 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)) + } + 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}) + 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..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,63 @@ 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 +// 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).