Integrate watermark into evmrpc

evmrpc/block.go

@@ -43,14 +43,15 @@ type BlockAPI struct {
 	namespace            string
 	includeShellReceipts bool
 	includeBankTransfers bool
+	watermarks           *WatermarkManager
 }
 
 type SeiBlockAPI struct {
 	*BlockAPI
 	isPanicTx func(ctx context.Context, hash common.Hash) (bool, error)
 }
 
-func NewBlockAPI(tmClient rpcclient.Client, k *keeper.Keeper, ctxProvider func(int64) sdk.Context, txConfigProvider func(int64) client.TxConfig, connectionType ConnectionType) *BlockAPI {
+func NewBlockAPI(tmClient rpcclient.Client, k *keeper.Keeper, ctxProvider func(int64) sdk.Context, txConfigProvider func(int64) client.TxConfig, connectionType ConnectionType, watermarks *WatermarkManager) *BlockAPI {
 	return &BlockAPI{
 		tmClient:             tmClient,
 		keeper:               k,
@@ -60,6 +61,7 @@ func NewBlockAPI(tmClient rpcclient.Client, k *keeper.Keeper, ctxProvider func(i
 		includeShellReceipts: false,
 		includeBankTransfers: false,
 		namespace:            EthNamespace,
+		watermarks:           watermarks,
 	}
 }
 
@@ -70,6 +72,7 @@ func NewSeiBlockAPI(
 	txConfigProvider func(int64) client.TxConfig,
 	connectionType ConnectionType,
 	isPanicTx func(ctx context.Context, hash common.Hash) (bool, error),
+	watermarks *WatermarkManager,
 ) *SeiBlockAPI {
 	blockAPI := &BlockAPI{
 		tmClient:             tmClient,
@@ -80,6 +83,7 @@ func NewSeiBlockAPI(
 		includeShellReceipts: true,
 		includeBankTransfers: false,
 		namespace:            SeiNamespace,
+		watermarks:           watermarks,
 	}
 	return &SeiBlockAPI{
 		BlockAPI:  blockAPI,
@@ -94,8 +98,9 @@ func NewSei2BlockAPI(
 	txConfigProvider func(int64) client.TxConfig,
 	connectionType ConnectionType,
 	isPanicTx func(ctx context.Context, hash common.Hash) (bool, error),
+	watermarks *WatermarkManager,
 ) *SeiBlockAPI {
-	blockAPI := NewSeiBlockAPI(tmClient, k, ctxProvider, txConfigProvider, connectionType, isPanicTx)
+	blockAPI := NewSeiBlockAPI(tmClient, k, ctxProvider, txConfigProvider, connectionType, isPanicTx, watermarks)
 	blockAPI.namespace = Sei2Namespace
 	blockAPI.includeBankTransfers = true
 	return blockAPI
@@ -118,10 +123,13 @@ func (a *BlockAPI) GetBlockTransactionCountByNumber(ctx context.Context, number
 	if err != nil {
 		return nil, err
 	}
-	block, err := blockByNumberWithRetry(ctx, a.tmClient, numberPtr, 1)
+	block, err := blockByNumberRespectingWatermarks(ctx, a.tmClient, a.watermarks, numberPtr, 1)
 	if err != nil {
 		return nil, err
 	}
+	if err := a.ensureHeightAvailable(ctx, block.Block.Height); err != nil {
+		return nil, err
+	}
 	return a.getEvmTxCount(block.Block.Txs, block.Block.Height), nil
 }
 
@@ -132,6 +140,9 @@ func (a *BlockAPI) GetBlockTransactionCountByHash(ctx context.Context, blockHash
 	if err != nil {
 		return nil, err
 	}
+	if err := a.ensureHeightAvailable(ctx, block.Block.Height); err != nil {
+		return nil, err
+	}
 	return a.getEvmTxCount(block.Block.Txs, block.Block.Height), nil
 }
 
@@ -147,6 +158,9 @@ func (a *BlockAPI) getBlockByHash(ctx context.Context, blockHash common.Hash, fu
 	if err != nil {
 		return nil, err
 	}
+	if err := a.ensureHeightAvailable(ctx, block.Block.Height); err != nil {
+		return nil, err
+	}
 
 	// Validate EVM block height for pacific-1 chain
 	sdkCtx := a.ctxProvider(LatestCtxHeight)
@@ -212,17 +226,27 @@ func (a *BlockAPI) getBlockByNumber(
 		}
 	}
 
-	block, err := blockByNumberWithRetry(ctx, a.tmClient, numberPtr, 1)
+	block, err := blockByNumberRespectingWatermarks(ctx, a.tmClient, a.watermarks, numberPtr, 1)
 	if err != nil {
 		return nil, err
 	}
+	if err := a.ensureHeightAvailable(ctx, block.Block.Height); err != nil {
+		return nil, err
+	}
 	blockRes, err := blockResultsWithRetry(ctx, a.tmClient, &block.Block.Height)
 	if err != nil {
 		return nil, err
 	}
 	return EncodeTmBlock(a.ctxProvider, a.txConfigProvider, block, blockRes, a.keeper, fullTx, a.includeBankTransfers, includeSyntheticTxs, isPanicTx)
 }
 
+func (a *BlockAPI) ensureHeightAvailable(ctx context.Context, height int64) error {
+	if a.watermarks == nil {
+		return nil
+	}
+	return a.watermarks.EnsureHeightAvailable(ctx, height)
+}
+
 func (a *BlockAPI) GetBlockReceipts(ctx context.Context, blockNrOrHash rpc.BlockNumberOrHash) (result []map[string]interface{}, returnErr error) {
 	startTime := time.Now()
 	defer recordMetricsWithError(fmt.Sprintf("%s_getBlockReceipts", a.namespace), a.connectionType, startTime, returnErr)
@@ -232,13 +256,16 @@ func (a *BlockAPI) GetBlockReceipts(ctx context.Context, blockNrOrHash rpc.Block
 		return nil, err
 	}
 
-	block, err := blockByNumberWithRetry(ctx, a.tmClient, heightPtr, 1)
+	block, err := blockByNumberRespectingWatermarks(ctx, a.tmClient, a.watermarks, heightPtr, 1)
 	if err != nil {
 		return nil, err
 	}
 
 	// Get all tx hashes for the block
 	height := block.Block.Height
+	if err := a.ensureHeightAvailable(ctx, height); err != nil {
+		return nil, err
+	}
 	txHashes := getTxHashesFromBlock(a.ctxProvider, a.txConfigProvider, a.keeper, block, shouldIncludeSynthetic(a.namespace))
 	// Get tx receipts for all hashes in parallel
 	wg := sync.WaitGroup{}

evmrpc/config.go

@@ -75,6 +75,9 @@
 	// controls whether to synchronously flush receipts before block finalze or not
 	FlushReceiptSync bool `mapstructure:"flush_receipt_sync"`
 
+	// DisableWatermark disables cross-store watermark gating (intended for tests)
+	DisableWatermark bool `mapstructure:"disable_watermark"`
+
 	// Deny list defines list of methods that EVM RPC should fail fast
 	DenyList []string `mapstructure:"deny_list"`
 
@@ -126,6 +129,7 @@
 	MaxTxPoolTxs:                 1000,
 	Slow:                         false,
 	FlushReceiptSync:             false,
+	DisableWatermark:            false,
 	DenyList:                     make([]string, 0),
 	MaxLogNoBlock:                10000,
 	MaxBlocksForLog:              2000,
@@ -156,6 +160,7 @@
 	flagCheckTxTimeout               = "evm.checktx_timeout"
 	flagSlow                         = "evm.slow"
 	FlagFlushReceiptSync             = "evm.flush_receipt_sync"
+	flagDisableWatermark             = "evm.disable_watermark"
 	flagDenyList                     = "evm.deny_list"
 	flagMaxLogNoBlock                = "evm.max_log_no_block"
 	flagMaxBlocksForLog              = "evm.max_blocks_for_log"
@@ -256,6 +261,11 @@
 			return cfg, err
 		}
 	}
+	if v := opts.Get(flagDisableWatermark); v != nil {
+		if cfg.DisableWatermark, err = cast.ToBoolE(v); err != nil {
+			return cfg, err
+		}
+	}
 	if v := opts.Get(flagDenyList); v != nil {
 		if cfg.DenyList, err = cast.ToStringSliceE(v); err != nil {
 			return cfg, err

evmrpc/config_test.go

@@ -26,6 +26,7 @@ type opts struct {
 	maxTxPoolTxs                 interface{}
 	slow                         interface{}
 	flushReceiptSync             interface{}
+	disableWatermark             interface{}
 	denyList                     interface{}
 	maxLogNoBlock                interface{}
 	maxBlocksForLog              interface{}
@@ -90,6 +91,9 @@ func (o *opts) Get(k string) interface{} {
 	if k == "evm.flush_receipt_sync" {
 		return o.flushReceiptSync
 	}
+	if k == "evm.disable_watermark" {
+		return o.disableWatermark
+	}
 	if k == "evm.deny_list" {
 		return o.denyList
 	}
@@ -142,6 +146,7 @@ func TestReadConfig(t *testing.T) {
 		1000,
 		false,
 		false,
+		false,
 		make([]string, 0),
 		20000,
 		1000,

evmrpc/filter.go

@@ -262,6 +262,7 @@ func NewFilterAPI(
 	dbReadSemaphore chan struct{},
 	globalBlockCache BlockCache,
 	globalLogSlicePool *LogSlicePool,
+	watermarks *WatermarkManager,
 ) *FilterAPI {
 	if filterConfig.maxBlock <= 0 {
 		filterConfig.maxBlock = DefaultMaxBlockRange
@@ -281,6 +282,7 @@ func NewFilterAPI(
 		dbReadSemaphore:          dbReadSemaphore,
 		globalBlockCache:         globalBlockCache,
 		globalLogSlicePool:       globalLogSlicePool,
+		watermarks:               watermarks,
 	}
 	filters := make(map[ethrpc.ID]filter)
 	api := &FilterAPI{
@@ -643,10 +645,18 @@ type LogFetcher struct {
 	globalBlockCache         BlockCache
 	cacheCreationMutex       sync.Mutex
 	globalLogSlicePool       *LogSlicePool
+	watermarks               *WatermarkManager
 }
 
 func (f *LogFetcher) GetLogsByFilters(ctx context.Context, crit filters.FilterCriteria, lastToHeight int64) (res []*ethtypes.Log, end int64, err error) {
-	latest := f.ctxProvider(LatestCtxHeight).BlockHeight()
+	latest, err := f.latestHeight(ctx)
+	if err != nil {
+		return nil, 0, err
+	}
+	earliest, err := f.earliestHeight(ctx)
+	if err != nil {
+		earliest = 0
+	}
 	begin, end := latest, latest
 	if crit.FromBlock != nil {
 		begin = getHeightFromBigIntBlockNumber(latest, crit.FromBlock)
@@ -660,6 +670,18 @@ func (f *LogFetcher) GetLogsByFilters(ctx context.Context, crit filters.FilterCr
 	if lastToHeight > begin {
 		begin = lastToHeight
 	}
+	if begin < earliest {
+		begin = earliest
+	}
+	if end > latest {
+		end = latest
+	}
+	if end < earliest {
+		end = earliest
+	}
+	if begin > end {
+		return []*ethtypes.Log{}, end, nil
+	}
 
 	blockRange := end - begin + 1
 
@@ -805,6 +827,47 @@ func (f *LogFetcher) mergeSortedLogs(batches [][]*ethtypes.Log) []*ethtypes.Log
 	return res
 }
 
+func (f *LogFetcher) ensureHeightAvailable(ctx context.Context, height int64) error {
+	if f.watermarks == nil {
+		return nil
+	}
+	return f.watermarks.EnsureHeightAvailable(ctx, height)
+}
+
+func (f *LogFetcher) latestHeight(ctx context.Context) (int64, error) {
+	if f.watermarks != nil {
+		return f.watermarks.LatestHeight(ctx)
+	}
+	if f.ctxProvider == nil {
+		return 0, fmt.Errorf("ctx provider not configured")
+	}
+	return f.ctxProvider(LatestCtxHeight).BlockHeight(), nil
+}
+
+func (f *LogFetcher) earliestHeight(ctx context.Context) (int64, error) {
+	if f.watermarks != nil {
+		return f.watermarks.EarliestHeight(ctx)
+	}
+	if f.ctxProvider == nil {
+		return 0, fmt.Errorf("ctx provider not configured")
+	}
+	storeCtx := f.ctxProvider(LatestCtxHeight)
+	ms := storeCtx.MultiStore()
+	if ms == nil {
+		return 0, nil
+	}
+	earliest := int64(0)
+	func() {
+		defer func() {
+			if r := recover(); r != nil {
+				earliest = 0
+			}
+		}()
+		earliest = ms.GetEarliestVersion()
+	}()
+	return earliest, nil
+}
+
 // Pooled version that reuses slice allocation
 func (f *LogFetcher) GetLogsForBlockPooled(block *coretypes.ResultBlock, crit filters.FilterCriteria, filters [][]bloomIndexes, result *[]*ethtypes.Log) {
 	collector := &pooledCollector{logs: result}
@@ -894,14 +957,26 @@ func (f *LogFetcher) fetchBlocksByCrit(ctx context.Context, crit filters.FilterC
 			close(res)
 			return res, 0, false, nil
 		}
+		if err := f.ensureHeightAvailable(ctx, block.Block.Height); err != nil {
+			res := make(chan *coretypes.ResultBlock)
+			close(res)
+			return res, 0, false, nil
+		}
 		res := make(chan *coretypes.ResultBlock, 1)
 		res <- block
 		close(res)
 		return res, 0, false, nil
 	}
 
 	applyOpenEndedLogLimit := f.filterConfig.maxLog > 0 && (crit.FromBlock == nil || crit.ToBlock == nil)
-	latest := f.ctxProvider(LatestCtxHeight).BlockHeight()
+	latest, err := f.latestHeight(ctx)
+	if err != nil {
+		return nil, 0, false, err
+	}
+	earliest, err := f.earliestHeight(ctx)
+	if err != nil {
+		earliest = 0
+	}
 	begin, end := latest, latest
 	if crit.FromBlock != nil {
 		begin = getHeightFromBigIntBlockNumber(latest, crit.FromBlock)
@@ -915,12 +990,21 @@ func (f *LogFetcher) fetchBlocksByCrit(ctx context.Context, crit filters.FilterC
 	if lastToHeight > begin {
 		begin = lastToHeight
 	}
+	if begin < earliest {
+		begin = earliest
+	}
+	if end > latest {
+		end = latest
+	}
+	if end < earliest {
+		end = earliest
+	}
 
 	blockRange := end - begin + 1
 	if applyOpenEndedLogLimit && blockRange > f.filterConfig.maxBlock {
 		begin = end - f.filterConfig.maxBlock + 1
-		if begin < 1 {
-			begin = 1
+		if begin < earliest {
+			begin = earliest
 		}
 	} else if !applyOpenEndedLogLimit && f.filterConfig.maxBlock > 0 && blockRange > f.filterConfig.maxBlock {
 		// Use consistent error message format
@@ -988,6 +1072,11 @@ func (f *LogFetcher) processBatch(ctx context.Context, start, end int64, crit fi
 
 		// check cache first, without holding the semaphore
 		if cachedEntry, found := f.globalBlockCache.Get(height); found {
+			if cachedEntry.Block != nil {
+				if err := f.ensureHeightAvailable(ctx, cachedEntry.Block.Block.Height); err != nil {
+					continue
+				}
+			}
 			res <- cachedEntry.Block
 			continue
 		}
@@ -998,6 +1087,11 @@ func (f *LogFetcher) processBatch(ctx context.Context, start, end int64, crit fi
 		// Re-check cache after acquiring semaphore, in case another worker cached it.
 		if cachedEntry, found := f.globalBlockCache.Get(height); found {
 			<-f.dbReadSemaphore
+			if cachedEntry.Block != nil {
+				if err := f.ensureHeightAvailable(ctx, cachedEntry.Block.Block.Height); err != nil {
+					continue
+				}
+			}
 			res <- cachedEntry.Block
 			continue
 		}
@@ -1020,7 +1114,7 @@ func (f *LogFetcher) processBatch(ctx context.Context, start, end int64, crit fi
 		}
 
 		// fetch block from network
-		block, err := blockByNumberWithRetry(ctx, f.tmClient, &height, 1)
+		block, err := blockByNumberRespectingWatermarks(ctx, f.tmClient, f.watermarks, &height, 1)
 		if err != nil {
 			select {
 			case errChan <- fmt.Errorf("failed to fetch block at height %d: %w", height, err):
@@ -1029,6 +1123,10 @@ func (f *LogFetcher) processBatch(ctx context.Context, start, end int64, crit fi
 			<-f.dbReadSemaphore
 			continue
 		}
+		if err := f.ensureHeightAvailable(ctx, block.Block.Height); err != nil {
+			<-f.dbReadSemaphore
+			continue
+		}
 
 		// Use LoadOrStore to create/get cache entry atomically
 		entry := loadOrStoreCacheEntry(&f.cacheCreationMutex, f.globalBlockCache, height, block)