Include receipt store earliest height in watermark bounds to prevent silent undercount on pruned blocks

evmrpc/block.go

@@ -178,6 +178,9 @@ func (a *BlockAPI) GetBlockTransactionCountByNumber(ctx context.Context, number
 	if err != nil {
 		return nil, err
 	}
+	if err = a.watermarks.EnsureReceiptHeightAvailable(ctx, block.Block.Height); err != nil {
+		return nil, err
+	}
 	return a.getEvmTxCount(block), nil
 }
 
@@ -191,6 +194,9 @@ func (a *BlockAPI) GetBlockTransactionCountByHash(ctx context.Context, blockHash
 	if err != nil {
 		return nil, err
 	}
+	if err = a.watermarks.EnsureReceiptHeightAvailable(ctx, block.Block.Height); err != nil {
+		return nil, err
+	}
 	return a.getEvmTxCount(block), nil
 }
 

evmrpc/height_availability_test.go

@@ -281,6 +281,32 @@ func TestLogFetcherSkipsUnavailableCachedBlock(t *testing.T) {
 	}
 }
 
+func TestGetBlockTransactionCountByNumberReceiptsPruned(t *testing.T) {
+	t.Parallel()
+
+	client := newHeightTestClient(100, 1, 200)
+	rs := &fakeReceiptStore{latest: 200, earliest: 150}
+	watermarks := NewWatermarkManager(client, testCtxProvider, nil, rs)
+	api := NewBlockAPI(client, nil, testCtxProvider, testTxConfigProvider, ConnectionTypeHTTP, watermarks, nil, nil)
+
+	_, err := api.GetBlockTransactionCountByNumber(context.Background(), rpc.BlockNumber(100))
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "receipts have been pruned")
+}
+
+func TestGetBlockTransactionCountByHashReceiptsPruned(t *testing.T) {
+	t.Parallel()
+
+	client := newHeightTestClient(100, 1, 200)
+	rs := &fakeReceiptStore{latest: 200, earliest: 150}
+	watermarks := NewWatermarkManager(client, testCtxProvider, nil, rs)
+	api := NewBlockAPI(client, nil, testCtxProvider, testTxConfigProvider, ConnectionTypeHTTP, watermarks, nil, nil)
+
+	_, err := api.GetBlockTransactionCountByHash(context.Background(), common.HexToHash(highBlockHashHex))
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "receipts have been pruned")
+}
+
 func TestStateAPIGetProofUnavailableHeight(t *testing.T) {
 	t.Parallel()
 

evmrpc/simulate_test.go

@@ -50,6 +50,7 @@ func primeReceiptStore(t *testing.T, store receipt.ReceiptStore, latest int64) {
 	}
 	require.NoError(t, store.SetLatestVersion(latest))
 	require.NoError(t, store.SetEarliestVersion(1))
+	require.Equal(t, int64(1), store.EarliestVersion())
 }
 
 func (bc *bcFailClient) Block(ctx context.Context, h *int64) (*coretypes.ResultBlock, error) {

evmrpc/state_test.go

@@ -221,6 +221,7 @@ func TestGetProof(t *testing.T) {
 	if store := testApp.EvmKeeper.ReceiptStore(); store != nil {
 		require.NoError(t, store.SetLatestVersion(MockHeight8))
 		require.NoError(t, store.SetEarliestVersion(1))
+		require.Equal(t, int64(1), store.EarliestVersion())
 	}
 	client := &MockClient{}
 	ctxProvider := func(height int64) sdk.Context {

evmrpc/watermark_manager.go

@@ -219,6 +219,24 @@ func (m *WatermarkManager) EnsureBlockHeightAvailable(ctx context.Context, heigh
 	return m.ensureWithinWatermarks(height, blockEarliest, latest)
 }
 
+// EnsureReceiptHeightAvailable verifies that receipts for the given block height
+// have not been pruned from the receipt store. This is a separate check from
+// EnsureBlockHeightAvailable because the receipt store can be configured with a
+// smaller KeepRecent than the block or state stores.
+func (m *WatermarkManager) EnsureReceiptHeightAvailable(_ context.Context, height int64) error {
+	if m.receiptStore == nil {
+		return nil
+	}
+	earliest := m.receiptStore.EarliestVersion()
+	if earliest <= 0 {
+		return nil
+	}
+	if height < earliest {
+		return fmt.Errorf("requested height %d receipts have been pruned; earliest available is %d", height, earliest)
+	}
+	return nil
+}
+
 func (m *WatermarkManager) ensureWithinWatermarks(height, earliest, latest int64) error {
 	if height > latest {
 		return fmt.Errorf("requested height %d is not yet available; safe latest is %d: %w", height, latest, ErrBlockHeightNotYetAvailable)

evmrpc/watermark_manager_test.go

@@ -104,6 +104,37 @@ func TestEnsureBlockHeightAvailableBounds(t *testing.T) {
 	require.ErrorContains(t, wm.EnsureBlockHeightAvailable(context.Background(), 2), "has been pruned")
 }
 
+func TestEnsureReceiptHeightAvailable(t *testing.T) {
+	tmClient := &fakeTMClient{
+		status: &coretypes.ResultStatus{SyncInfo: coretypes.SyncInfo{LatestBlockHeight: 200, EarliestBlockHeight: 1}},
+	}
+
+	t.Run("no receipt store allows any height", func(t *testing.T) {
+		wm := NewWatermarkManager(tmClient, nil, nil, nil)
+		require.NoError(t, wm.EnsureReceiptHeightAvailable(context.Background(), 5))
+	})
+
+	t.Run("receipt store with no pruning allows any height", func(t *testing.T) {
+		rs := &fakeReceiptStore{latest: 200, earliest: 0}
+		wm := NewWatermarkManager(tmClient, nil, nil, rs)
+		require.NoError(t, wm.EnsureReceiptHeightAvailable(context.Background(), 5))
+	})
+
+	t.Run("pruned receipt height returns error", func(t *testing.T) {
+		rs := &fakeReceiptStore{latest: 200, earliest: 150}
+		wm := NewWatermarkManager(tmClient, nil, nil, rs)
+		require.ErrorContains(t, wm.EnsureReceiptHeightAvailable(context.Background(), 100), "receipts have been pruned")
+		require.ErrorContains(t, wm.EnsureReceiptHeightAvailable(context.Background(), 149), "receipts have been pruned")
+	})
+
+	t.Run("height within receipt retention succeeds", func(t *testing.T) {
+		rs := &fakeReceiptStore{latest: 200, earliest: 150}
+		wm := NewWatermarkManager(tmClient, nil, nil, rs)
+		require.NoError(t, wm.EnsureReceiptHeightAvailable(context.Background(), 150))
+		require.NoError(t, wm.EnsureReceiptHeightAvailable(context.Background(), 175))
+	})
+}
+
 func TestLatestAndEarliestHeightHelpers(t *testing.T) {
 	tmClient := &fakeTMClient{
 		status: &coretypes.ResultStatus{SyncInfo: coretypes.SyncInfo{LatestBlockHeight: 22, EarliestBlockHeight: 11}},
@@ -233,19 +264,27 @@ func (f *fakeStateStore) Prune(_ int64) error
 func (f *fakeStateStore) Close() error                                                 { return nil }
 
 type fakeReceiptStore struct {
-	latest int64
+	latest   int64
+	earliest int64
 }
 
 func (f *fakeReceiptStore) LatestVersion() int64 {
 	return f.latest
 }
 
+func (f *fakeReceiptStore) EarliestVersion() int64 {
+	return f.earliest
+}
+
 func (f *fakeReceiptStore) SetLatestVersion(version int64) error {
 	f.latest = version
 	return nil
 }
 
-func (f *fakeReceiptStore) SetEarliestVersion(_ int64) error { return nil }
+func (f *fakeReceiptStore) SetEarliestVersion(version int64) error {
+	f.earliest = version
+	return nil
+}
 
 func (f *fakeReceiptStore) GetReceipt(sdk.Context, common.Hash) (*evmtypes.Receipt, error) {
 	return nil, errors.New("not found")

sei-db/ledger_db/parquet/store.go

@@ -181,6 +181,11 @@ func (s *Store) SetEarliestVersion(version int64) {
 	s.earliestVersion.Store(version)
 }
 
+// EarliestVersion returns the earliest version retained in the store.
+func (s *Store) EarliestVersion() int64 {
+	return s.earliestVersion.Load()
+}
+
 // CacheRotateInterval returns the interval at which the cache should rotate.
 func (s *Store) CacheRotateInterval() uint64 {
 	return s.config.MaxBlocksPerFile

sei-db/ledger_db/parquet/store_config_test.go

@@ -236,6 +236,20 @@ func TestLazyInitCreatesFileOnFirstWrite(t *testing.T) {
 	require.Contains(t, logs[0], "logs_5000.parquet")
 }
 
+func TestStoreEarliestVersion(t *testing.T) {
+	store, err := NewStore(StoreConfig{
+		DBDirectory:          t.TempDir(),
+		DisableTxIndexLookup: true,
+	})
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = store.Close() })
+
+	require.Equal(t, int64(0), store.EarliestVersion())
+
+	store.SetEarliestVersion(55)
+	require.Equal(t, int64(55), store.EarliestVersion())
+}
+
 func writeValidParquetFile(t *testing.T, dir, name string) {
 	t.Helper()
 	path := filepath.Join(dir, name)

sei-db/ledger_db/receipt/cached_receipt_store.go

@@ -83,6 +83,10 @@ func (s *cachedReceiptStore) SetLatestVersion(version int64) error {
 	return s.backend.SetLatestVersion(version)
 }
 
+func (s *cachedReceiptStore) EarliestVersion() int64 {
+	return s.backend.EarliestVersion()
+}
+
 func (s *cachedReceiptStore) SetEarliestVersion(version int64) error {
 	return s.backend.SetEarliestVersion(version)
 }

sei-db/ledger_db/receipt/cached_receipt_store_test.go

@@ -58,6 +58,10 @@ func (f *fakeReceiptBackend) LatestVersion() int64 {
 	return 0
 }
 
+func (f *fakeReceiptBackend) EarliestVersion() int64 {
+	return 0
+}
+
 func (f *fakeReceiptBackend) SetLatestVersion(int64) error {
 	return nil
 }

sei-db/ledger_db/receipt/parquet_store.go

@@ -53,6 +53,10 @@ func (s *parquetReceiptStore) SetLatestVersion(version int64) error {
 	return nil
 }
 
+func (s *parquetReceiptStore) EarliestVersion() int64 {
+	return s.store.EarliestVersion()
+}
+
 func (s *parquetReceiptStore) SetEarliestVersion(version int64) error {
 	s.store.SetEarliestVersion(version)
 	return nil

sei-db/ledger_db/receipt/parquet_store_test.go

@@ -402,6 +402,24 @@ func TestParquetPruneOldFiles(t *testing.T) {
 	require.NoError(t, err)
 }
 
+func TestParquetReceiptStoreEarliestVersion(t *testing.T) {
+	_, storeKey := newTestContext()
+	cfg := dbconfig.DefaultReceiptStoreConfig()
+	cfg.Backend = "parquet"
+	cfg.DBDirectory = t.TempDir()
+
+	store, err := NewReceiptStore(cfg, storeKey)
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = store.Close() })
+
+	pqStore := store.(*cachedReceiptStore).backend.(*parquetReceiptStore)
+
+	require.Equal(t, int64(0), pqStore.EarliestVersion())
+
+	require.NoError(t, pqStore.SetEarliestVersion(77))
+	require.Equal(t, int64(77), pqStore.EarliestVersion())
+}
+
 func TestExtractBlockNumber(t *testing.T) {
 	tests := []struct {
 		path     string

sei-db/ledger_db/receipt/receipt_store.go

@@ -35,6 +35,7 @@ var (
 // ReceiptStore exposes receipt-specific operations without leaking the StateStore interface.
 type ReceiptStore interface {
 	LatestVersion() int64
+	EarliestVersion() int64
 	SetLatestVersion(version int64) error
 	SetEarliestVersion(version int64) error
 	GetReceipt(ctx sdk.Context, txHash common.Hash) (*types.Receipt, error)
@@ -172,6 +173,10 @@ func (s *receiptStore) SetLatestVersion(version int64) error {
 	return s.db.SetLatestVersion(version)
 }
 
+func (s *receiptStore) EarliestVersion() int64 {
+	return s.db.GetEarliestVersion()
+}
+
 func (s *receiptStore) SetEarliestVersion(version int64) error {
 	return s.db.SetEarliestVersion(version, true)
 }

sei-db/ledger_db/receipt/receipt_store_test.go

@@ -109,6 +109,7 @@ func TestSetReceiptsAndGet(t *testing.T) {
 	require.NoError(t, store.SetLatestVersion(10))
 	require.Equal(t, int64(10), store.LatestVersion())
 	require.NoError(t, store.SetEarliestVersion(1))
+	require.Equal(t, int64(1), store.EarliestVersion())
 }
 
 func TestReceiptStoreLegacyFallback(t *testing.T) {