fix(giga): fail-fast broadcast_tx_commit and feed newBlockFilter from the notifier

app/abci.go

@@ -291,8 +291,9 @@ func (app *App) Commit(ctx context.Context) (res *abci.ResponseCommit, err error
 	// legacy: telemetry.MeasureSince in sei-cosmos/baseapp/abci.go TODO(PLT-353)
 	appMetrics.commitDuration.Record(ctx, elapsed.Seconds())
 	app.RecordBenchmarkCommitTime(elapsed)
-	// After a successful Commit, publish the pending eth_newHeads event
-	// stashed by FinalizeBlocker. Subscribers see only committed state.
+	// After a successful Commit, publish the pending committed-block
+	// event stashed by FinalizeBlocker. newHeads and newBlockFilter
+	// see only committed state.
 	// Header.AppHash is intentionally left unset (Tendermint convention:
 	// it holds the previous block's hash); stateRoot is sourced from
 	// response.AppHash by encodeCommittedBlock.

app/app.go

@@ -431,12 +431,12 @@ type App struct {
 	legacyEncodingConfig appparams.EncodingConfig
 	evmRPCConfig         evmrpcconfig.Config
 	// blockHeaderNotifier is non-nil only when Autobahn is enabled. It
-	// owns the FinalizeBlock→Commit pairing for eth_subscribe("newHeads"):
-	// FinalizeBlocker calls Stash with the (hash, header, response)
-	// tuple, App.Commit calls PublishStashed after a successful
-	// BaseApp.Commit, and FinalizeBlocker entry calls ClearStash to
-	// defend against stale tuples from prior failed commits or
-	// non-stashing return paths (EthReplay/EthBlockTest).
+	// owns the FinalizeBlock→Commit pairing for eth_subscribe("newHeads")
+	// and eth_newBlockFilter: FinalizeBlocker calls Stash with the
+	// (hash, header, response) tuple, App.Commit calls PublishStashed
+	// after a successful BaseApp.Commit, and FinalizeBlocker entry
+	// calls ClearStash to defend against stale tuples from prior failed
+	// commits or non-stashing return paths (EthReplay/EthBlockTest).
 	blockHeaderNotifier   tmutils.Option[*evmrpc.BlockHeaderNotifier]
 	adminConfig           admin.Config
 	adminServer           *grpc.Server
@@ -2577,8 +2577,9 @@ func (app *App) RegisterLocalServices(node client.LocalClient, txConfig client.T
 
 	rpcCtxProvider := app.RPCContextProvider
 	traceCtxProvider := app.SnapshotAwareRPCContextProvider()
+	headNotifier, _ := app.blockHeaderNotifier.Get()
 	if app.evmRPCConfig.HTTPEnabled {
-		evmHTTPServer, err := evmrpc.NewEVMHTTPServer(app.evmRPCConfig, node, &app.EvmKeeper, app.BeginBlockKeepers, app.BaseApp, app.TracerAnteHandler, app.RPCContextProvider, txConfigProvider, DefaultNodeHome, app.GetStateStore(), traceCtxProvider)
+		evmHTTPServer, err := evmrpc.NewEVMHTTPServer(app.evmRPCConfig, node, &app.EvmKeeper, app.BeginBlockKeepers, app.BaseApp, app.TracerAnteHandler, app.RPCContextProvider, txConfigProvider, DefaultNodeHome, app.GetStateStore(), headNotifier, traceCtxProvider)
 		if err != nil {
 			panic(err)
 		}
@@ -2592,7 +2593,6 @@ func (app *App) RegisterLocalServices(node client.LocalClient, txConfig client.T
 	}
 
 	if app.evmRPCConfig.WSEnabled {
-		headNotifier, _ := app.blockHeaderNotifier.Get()
 		evmWSServer, err := evmrpc.NewEVMWebSocketServer(app.evmRPCConfig, node, &app.EvmKeeper, app.BeginBlockKeepers, app.BaseApp, app.TracerAnteHandler, rpcCtxProvider, txConfigProvider, DefaultNodeHome, app.GetStateStore(), headNotifier)
 		if err != nil {
 			panic(err)

evmrpc/filter.go

@@ -201,6 +201,7 @@ type filter struct {
 
 	// BlocksSubscription
 	blockCursor string
+	blockHashes []common.Hash
 
 	// LogsSubscription
 	lastToHeight int64
@@ -262,17 +263,18 @@ func (h *logMergeHeap) Pop() interface{} {
 }
 
 type FilterAPI struct {
-	tmClient         client.LocalClient
-	filtersMu        sync.RWMutex
-	filters          map[ethrpc.ID]filter
-	toDelete         chan ethrpc.ID
-	filterConfig     *FilterConfig
-	logFetcher       *LogFetcher
-	connectionType   ConnectionType
-	namespace        string
-	shutdownCtx      context.Context
-	shutdownCancel   context.CancelFunc
-	globalRPSLimiter *rate.Limiter
+	tmClient            client.LocalClient
+	filtersMu           sync.RWMutex
+	filters             map[ethrpc.ID]filter
+	toDelete            chan ethrpc.ID
+	filterConfig        *FilterConfig
+	logFetcher          *LogFetcher
+	connectionType      ConnectionType
+	namespace           string
+	shutdownCtx         context.Context
+	shutdownCancel      context.CancelFunc
+	globalRPSLimiter    *rate.Limiter
+	blockHeaderNotifier *BlockHeaderNotifier
 }
 
 type FilterConfig struct {
@@ -300,6 +302,7 @@ func NewFilterAPI(
 	cacheCreationMutex *sync.Mutex,
 	globalLogSlicePool *LogSlicePool,
 	watermarks *WatermarkManager,
+	blockHeaderNotifier *BlockHeaderNotifier,
 ) *FilterAPI {
 	if filterConfig.maxBlock <= 0 {
 		filterConfig.maxBlock = DefaultMaxBlockRange
@@ -327,23 +330,60 @@ func NewFilterAPI(
 	}
 	filters := make(map[ethrpc.ID]filter)
 	api := &FilterAPI{
-		namespace:        namespace,
-		tmClient:         tmClient,
-		filtersMu:        sync.RWMutex{},
-		filters:          filters,
-		toDelete:         make(chan ethrpc.ID, 1000),
-		filterConfig:     filterConfig,
-		logFetcher:       logFetcher,
-		connectionType:   connectionType,
-		shutdownCtx:      shutdownCtx,
-		shutdownCancel:   shutdownCancel,
-		globalRPSLimiter: rate.NewLimiter(rate.Limit(GlobalRPSLimit), GlobalRPSLimit),
+		namespace:           namespace,
+		tmClient:            tmClient,
+		filtersMu:           sync.RWMutex{},
+		filters:             filters,
+		toDelete:            make(chan ethrpc.ID, 1000),
+		filterConfig:        filterConfig,
+		logFetcher:          logFetcher,
+		connectionType:      connectionType,
+		shutdownCtx:         shutdownCtx,
+		shutdownCancel:      shutdownCancel,
+		globalRPSLimiter:    rate.NewLimiter(rate.Limit(GlobalRPSLimit), GlobalRPSLimit),
+		blockHeaderNotifier: blockHeaderNotifier,
+	}
+	if blockHeaderNotifier != nil {
+		blockHeaderNotifier.subscribe(api.appendBlockHash)
 	}
 
 	go api.cleanupLoop(filterConfig.timeout)
 	return api
 }
 
+// appendBlockHash records the committed Autobahn block hash on every
+// live BlocksSubscription.
+func (a *FilterAPI) appendBlockHash(evt blockHeaderEvent) {
+	hash := common.BytesToHash(evt.hash)
+	a.filtersMu.Lock()
+	defer a.filtersMu.Unlock()
+	for id, f := range a.filters {
+		if f.typ != BlocksSubscription {
+			continue
+		}
+		f.blockHashes = append(f.blockHashes, hash)
+		a.filters[id] = f
+	}
+}
+
+// takeBlockHashes returns and clears hashes accumulated for a
+// BlocksSubscription since the last poll.
+func (a *FilterAPI) takeBlockHashes(filterID ethrpc.ID) ([]common.Hash, error) {
+	a.filtersMu.Lock()
+	defer a.filtersMu.Unlock()
+	f, exists := a.filters[filterID]
+	if !exists {
+		return nil, errors.New("filter does not exist")
+	}
+	hashes := f.blockHashes
+	f.blockHashes = nil
+	a.filters[filterID] = f
+	if hashes == nil {
+		hashes = []common.Hash{}
+	}
+	return hashes, nil
+}
+
 // Unified cleanup loop that handles both timeout and manual deletion
 func (a *FilterAPI) cleanupLoop(timeout time.Duration) {
 	ticker := time.NewTicker(timeout / 2) // Check more frequently than timeout
@@ -503,6 +543,9 @@ func (a *FilterAPI) GetFilterChanges(
 	result := []*ethtypes.Log{}
 	switch filter.typ {
 	case BlocksSubscription:
+		if a.blockHeaderNotifier != nil {
+			return a.takeBlockHashes(filterID)
+		}
 		hashes, cursor, err := a.getBlockHeadersAfter(ctx, filter.blockCursor)
 		if err != nil {
 			return nil, err

evmrpc/filter_block_notifier_test.go

@@ -0,0 +1,131 @@
+package evmrpc
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/stretchr/testify/require"
+
+	sdk "github.com/sei-protocol/sei-chain/sei-cosmos/types"
+	abci "github.com/sei-protocol/sei-chain/sei-tendermint/abci/types"
+	tmproto "github.com/sei-protocol/sei-chain/sei-tendermint/proto/tendermint/types"
+)
+
+func newTestFilterAPI(t *testing.T, notifier *BlockHeaderNotifier) *FilterAPI {
+	t.Helper()
+	// Setup: FilterAPI with unused log/store deps; only the notifier path is exercised.
+	api := NewFilterAPI(
+		nil,
+		nil,
+		func(int64) sdk.Context { return sdk.Context{} },
+		nil,
+		&FilterConfig{timeout: time.Hour, maxLog: 1, maxLogBytes: 1, maxBlock: 1},
+		ConnectionTypeHTTP,
+		"eth",
+		make(chan struct{}, 1),
+		NewBlockCache(1),
+		&sync.Mutex{},
+		NewLogSlicePool(),
+		nil,
+		notifier,
+	)
+	t.Cleanup(api.shutdown)
+	return api
+}
+
+func TestFilterAPI_NewBlockFilterUsesNotifierHash(t *testing.T) {
+	// Setup: FilterAPI subscribed to a private notifier.
+	n := NewBlockHeaderNotifier(4)
+	api := newTestFilterAPI(t, n)
+	ctx := context.Background()
+
+	// Setup: create a block filter before any commits.
+	id, err := api.NewBlockFilter(ctx)
+	require.NoError(t, err)
+
+	// Test: poll before any commit.
+	empty, err := api.GetFilterChanges(ctx, id)
+	// Verify: empty slice, not nil.
+	require.NoError(t, err)
+	require.Equal(t, []common.Hash{}, empty)
+
+	// Test: publish one Autobahn hash, then poll twice.
+	hash := common.HexToHash("0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa")
+	n.OnBlockCommitted(hash.Bytes(), &tmproto.Header{Height: 9}, &abci.ResponseFinalizeBlock{})
+
+	got, err := api.GetFilterChanges(ctx, id)
+	// Verify: first poll returns that hash.
+	require.NoError(t, err)
+	require.Equal(t, []common.Hash{hash}, got)
+
+	again, err := api.GetFilterChanges(ctx, id)
+	// Verify: second poll is drained.
+	require.NoError(t, err)
+	require.Equal(t, []common.Hash{}, again)
+}
+
+func TestFilterAPI_NewBlockFilterAccumulatesUntilPoll(t *testing.T) {
+	// Setup: FilterAPI with one live block filter.
+	n := NewBlockHeaderNotifier(4)
+	api := newTestFilterAPI(t, n)
+	ctx := context.Background()
+	id, err := api.NewBlockFilter(ctx)
+	require.NoError(t, err)
+
+	// Test: commit two blocks before the client polls.
+	first := common.HexToHash("0x1111111111111111111111111111111111111111111111111111111111111111")
+	second := common.HexToHash("0x2222222222222222222222222222222222222222222222222222222222222222")
+	n.OnBlockCommitted(first.Bytes(), &tmproto.Header{Height: 1}, &abci.ResponseFinalizeBlock{})
+	n.OnBlockCommitted(second.Bytes(), &tmproto.Header{Height: 2}, &abci.ResponseFinalizeBlock{})
+
+	// Verify: one poll returns both hashes in order.
+	got, err := api.GetFilterChanges(ctx, id)
+	require.NoError(t, err)
+	require.Equal(t, []common.Hash{first, second}, got)
+}
+
+func TestFilterAPI_NewBlockFilterFanOutToEachFilter(t *testing.T) {
+	// Setup: two live block filters on the same notifier.
+	n := NewBlockHeaderNotifier(4)
+	api := newTestFilterAPI(t, n)
+	ctx := context.Background()
+	idA, err := api.NewBlockFilter(ctx)
+	require.NoError(t, err)
+	idB, err := api.NewBlockFilter(ctx)
+	require.NoError(t, err)
+
+	// Test: publish one hash.
+	hash := common.HexToHash("0xbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb")
+	n.OnBlockCommitted(hash.Bytes(), &tmproto.Header{Height: 3}, &abci.ResponseFinalizeBlock{})
+
+	// Verify: each filter independently returns that hash.
+	gotA, err := api.GetFilterChanges(ctx, idA)
+	require.NoError(t, err)
+	gotB, err := api.GetFilterChanges(ctx, idB)
+	require.NoError(t, err)
+	require.Equal(t, []common.Hash{hash}, gotA)
+	require.Equal(t, []common.Hash{hash}, gotB)
+}
+
+func TestFilterAPI_NewBlockFilterIgnoresBlocksBeforeCreate(t *testing.T) {
+	// Setup: FilterAPI with a commit that happens before the filter exists.
+	n := NewBlockHeaderNotifier(4)
+	api := newTestFilterAPI(t, n)
+	ctx := context.Background()
+	before := common.HexToHash("0x0101010101010101010101010101010101010101010101010101010101010101")
+	n.OnBlockCommitted(before.Bytes(), &tmproto.Header{Height: 1}, &abci.ResponseFinalizeBlock{})
+
+	// Test: create the filter, then commit a later block.
+	id, err := api.NewBlockFilter(ctx)
+	require.NoError(t, err)
+	after := common.HexToHash("0x0202020202020202020202020202020202020202020202020202020202020202")
+	n.OnBlockCommitted(after.Bytes(), &tmproto.Header{Height: 2}, &abci.ResponseFinalizeBlock{})
+
+	// Verify: only the post-create hash is returned.
+	got, err := api.GetFilterChanges(ctx, id)
+	require.NoError(t, err)
+	require.Equal(t, []common.Hash{after}, got)
+}

evmrpc/filter_test.go

@@ -9,6 +9,8 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/sei-protocol/sei-chain/evmrpc"
+	abci "github.com/sei-protocol/sei-chain/sei-tendermint/abci/types"
+	tmproto "github.com/sei-protocol/sei-chain/sei-tendermint/proto/tendermint/types"
 )
 
 func TestNewPendingTransactionFilterNotSupported(t *testing.T) {
@@ -347,6 +349,47 @@ func TestFilterBlockFilter(t *testing.T) {
 	}
 }
 
+func TestFilterBlockFilterAutobahn(t *testing.T) {
+	t.Parallel()
+
+	// Setup: create a block filter on the notifier-backed HTTP server (TestMain).
+	resObj := sendRequest(t, TestNotifierHTTPPort, "newBlockFilter")
+	if errVal, ok := resObj["error"]; ok {
+		t.Fatal("newBlockFilter error:", errVal)
+	}
+	blockFilterId := resObj["result"].(string)
+
+	// Test: poll before any committed block.
+	resObj = sendRequest(t, TestNotifierHTTPPort, evmrpc.GetFilterChangesMethod, blockFilterId)
+	// Verify: empty array, not null.
+	hashesInterface, ok := resObj["result"].([]interface{})
+	require.True(t, ok, "getFilterChanges should return [] not null")
+	require.Empty(t, hashesInterface)
+
+	// Test: publish one Autobahn FinalizeBlock hash through the notifier.
+	hash := common.HexToHash("0x4242424242424242424242424242424242424242424242424242424242424242")
+	BlockFilterNotifierForTest.OnBlockCommitted(hash.Bytes(), &tmproto.Header{
+		Height: 1,
+		Time:   time.Unix(1_700_000_500, 0).UTC(),
+	}, &abci.ResponseFinalizeBlock{})
+
+	// Test: poll after the commit.
+	resObj = sendRequest(t, TestNotifierHTTPPort, evmrpc.GetFilterChangesMethod, blockFilterId)
+	if errVal, ok := resObj["error"]; ok {
+		t.Fatal("getFilterChanges error:", errVal)
+	}
+	// Verify: the filter returns that hash, not a zero Tendermint Header.Hash().
+	hashesInterface = resObj["result"].([]interface{})
+	require.Equal(t, []interface{}{hash.Hex()}, hashesInterface)
+
+	// Test: poll again with no new commits.
+	resObj = sendRequest(t, TestNotifierHTTPPort, evmrpc.GetFilterChangesMethod, blockFilterId)
+	// Verify: drained; empty array, not null.
+	hashesInterface, ok = resObj["result"].([]interface{})
+	require.True(t, ok, "exhausted getFilterChanges should return [] not null")
+	require.Empty(t, hashesInterface)
+}
+
 func TestFilterExpiration(t *testing.T) {
 	t.Parallel()
 	filterCriteria := map[string]interface{}{