feat(indexer): height-ordered KV secondary index for tx/block search (PLT-786)

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

@@ -30,8 +30,9 @@ const (
 // MakeReindexEventCommand constructs a command to re-index events in a block height interval.
 func MakeReindexEventCommand(conf *tmcfg.Config) *cobra.Command {
 	var (
-		startHeight int64
-		endHeight   int64
+		startHeight      int64
+		endHeight        int64
+		reportWatermarks bool
 	)
 
 	cmd := &cobra.Command{
@@ -50,6 +51,7 @@ either or both arguments.
 	tendermint reindex-event --start-height 2
 	tendermint reindex-event --end-height 10
 	tendermint reindex-event --start-height 2 --end-height 10
+	tendermint reindex-event --report-watermarks
 	`,
 		RunE: func(cmd *cobra.Command, args []string) error {
 			home, err := cmd.Flags().GetString(cli.HomeFlag)
@@ -75,6 +77,13 @@ either or both arguments.
 				return fmt.Errorf("%s: %w", reindexFailed, err)
 			}
 
+			if reportWatermarks {
+				if err = reportEventSinkWatermarks(es); err != nil {
+					return fmt.Errorf("%s: %w", reindexFailed, err)
+				}
+				return nil
+			}
+
 			riArgs := eventReIndexArgs{
 				startHeight: startHeight,
 				endHeight:   endHeight,
@@ -93,9 +102,43 @@ either or both arguments.
 
 	cmd.Flags().Int64Var(&startHeight, "start-height", 0, "the block height would like to start for re-index")
 	cmd.Flags().Int64Var(&endHeight, "end-height", 0, "the block height would like to finish for re-index")
+	cmd.Flags().BoolVar(&reportWatermarks, "report-watermarks", false,
+		"report the KV height-ordered index watermarks for each configured sink and exit without re-indexing")
 	return cmd
 }
 
+// reportEventSinkWatermarks prints the height-ordered index watermarks for every KV event sink.
+func reportEventSinkWatermarks(sinks []indexer.EventSink) error {
+	fmt.Println("height-ordered index watermarks:")
+	for _, s := range sinks {
+		kves, ok := s.(*kv.EventSink)
+		if !ok {
+			fmt.Printf("  %s: no height-ordered index watermark\n", s.Type())
+			continue
+		}
+
+		blockW, blockSet, err := kves.BlockWatermark()
+		if err != nil {
+			return fmt.Errorf("read block watermark: %w", err)
+		}
+		txW, txSet, err := kves.TxWatermark()
+		if err != nil {
+			return fmt.Errorf("read tx watermark: %w", err)
+		}
+
+		fmt.Printf("  %s: block=%s tx=%s\n", s.Type(), formatWatermark(blockW, blockSet), formatWatermark(txW, txSet))
+	}
+	return nil
+}
+
+// formatWatermark renders a watermark for display
+func formatWatermark(height int64, set bool) string {
+	if !set {
+		return "unset"
+	}
+	return fmt.Sprintf("%d", height)
+}
+
 func loadEventSinks(cfg *tmcfg.Config) ([]indexer.EventSink, error) {
 	// Check duplicated sinks.
 	sinks := map[string]bool{}
@@ -118,7 +161,12 @@ func loadEventSinks(cfg *tmcfg.Config) ([]indexer.EventSink, error) {
 			if err != nil {
 				return nil, err
 			}
-			eventSinks = append(eventSinks, kv.NewEventSink(store))
+			// Reindex must not touch the height-ordered watermark: a partial
+			// historical range would otherwise anchor it below heights it does
+			// not cover, routing those heights to the empty fast leg and
+			// silently dropping search matches. The keys are still dual-written;
+			// only the watermark is left to live forward indexing.
+			eventSinks = append(eventSinks, kv.NewEventSinkSkipWatermark(store))
 		case string(indexer.PSQL):
 			conn := cfg.TxIndex.PsqlConn
 			if conn == "" {

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
 }
 

sei-tendermint/config/config_test.go

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

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                             ###
 #######################################################################

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

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