From 8b5dedf68fc50b26c004b077550f8b86636e585f Mon Sep 17 00:00:00 2001 From: Danno Ferrin Date: Tue, 25 Aug 2026 10:45:02 -0600 Subject: [PATCH] fix(giga): fail-fast broadcast_tx_commit and feed newBlockFilter from the notifier Autobahn has no EventBus wait for inclusion, so BroadcastTxCommit returns unsupported before InsertTx. eth_newBlockFilter reads committed Autobahn hashes from BlockHeaderNotifier instead of /events. Co-authored-by: Cursor --- app/abci.go | 5 +- app/app.go | 16 +-- evmrpc/filter.go | 87 +++++++++--- evmrpc/filter_block_notifier_test.go | 131 +++++++++++++++++ evmrpc/filter_test.go | 43 ++++++ evmrpc/notifier.go | 68 ++++++--- evmrpc/notifier_internal_test.go | 61 +++++++- evmrpc/server.go | 2 + evmrpc/setup_test.go | 37 +++-- evmrpc/tests/utils.go | 1 + sei-tendermint/internal/rpc/core/mempool.go | 30 ++-- .../rpc/core/mempool_autobahn_test.go | 133 ++++++++++++++++++ 12 files changed, 531 insertions(+), 83 deletions(-) create mode 100644 evmrpc/filter_block_notifier_test.go create mode 100644 sei-tendermint/internal/rpc/core/mempool_autobahn_test.go diff --git a/app/abci.go b/app/abci.go index 34b54f6079..f089a3778a 100644 --- a/app/abci.go +++ b/app/abci.go @@ -306,8 +306,9 @@ func (app *App) Commit(ctx context.Context) (res *abci.ResponseCommit, err error // legacy: telemetry.MeasureSince in sei-cosmos/baseapp/abci.go TODO(PLT-327) 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. diff --git a/app/app.go b/app/app.go index 91a30d62ba..c39f0c04a4 100644 --- a/app/app.go +++ b/app/app.go @@ -438,12 +438,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 @@ -2625,8 +2625,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) } @@ -2640,7 +2641,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) diff --git a/evmrpc/filter.go b/evmrpc/filter.go index a44bb7b0df..9a580c1403 100644 --- a/evmrpc/filter.go +++ b/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 diff --git a/evmrpc/filter_block_notifier_test.go b/evmrpc/filter_block_notifier_test.go new file mode 100644 index 0000000000..d90415ea55 --- /dev/null +++ b/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) +} diff --git a/evmrpc/filter_test.go b/evmrpc/filter_test.go index da171262c8..dc8b523e22 100644 --- a/evmrpc/filter_test.go +++ b/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{}{ diff --git a/evmrpc/notifier.go b/evmrpc/notifier.go index 3e87c20014..f66877ee67 100644 --- a/evmrpc/notifier.go +++ b/evmrpc/notifier.go @@ -8,8 +8,9 @@ import ( ) // blockHeaderEvent is the in-process payload delivered to SubscriptionAPI -// for each committed block. It mirrors the data evmrpc needs to build an -// Ethereum block header, without going through the Tendermint event bus. +// and FilterAPI for each committed block. It mirrors the data evmrpc needs +// to build an Ethereum block header or a newBlockFilter hash, without +// going through the Tendermint event bus. // // hash is the autobahn lane-block header hash passed as Hash to // app.FinalizeBlock — NOT a hash computed over the (partially-populated) @@ -17,24 +18,27 @@ import ( // the eth_getBlockBy* and receipt API surfaces report as blockHash (the // receipt store on disk records a zero blockHash; evmrpc overlays this // hash at read time, see evmrpc/tx.go). Surfacing the same hash here -// keeps eth_newHeads consistent with the rest of the EVM RPC surface. +// keeps eth_newHeads and eth_newBlockFilter consistent with the rest of +// the EVM RPC surface. type blockHeaderEvent struct { hash []byte header *tmproto.Header response *abci.ResponseFinalizeBlock } -// BlockHeaderNotifier feeds eth_subscribe("newHeads") via a direct -// in-process channel. The sei-chain App stashes FinalizeBlock outputs -// (Stash) and publishes them after a successful Commit (PublishStashed), -// so subscribers only observe committed state. The single consumer is -// SubscriptionAPI's fan-out goroutine, which broadcasts to all per-client -// subscribers. +// BlockHeaderNotifier feeds eth_subscribe("newHeads") and +// eth_newBlockFilter from committed Autobahn blocks. The sei-chain App +// stashes FinalizeBlock outputs (Stash) and publishes them after a +// successful Commit (PublishStashed), so consumers only observe +// committed state. // -// Channel semantics: OnBlockCommitted is non-blocking and overwrite-on- -// full. If the consumer is lagging, the oldest buffered event is dropped -// in favour of the newest — for eth_newHeads the latest head is always -// more useful than a stale one. +// newHeads uses recv(): a bounded channel with overwrite-on-full. If +// that consumer lags, the oldest buffered event is dropped in favour of +// the newest — for eth_newHeads the latest head is always more useful +// than a stale one. +// +// Block filters use subscribe(): every published event is delivered, so +// eth_getFilterChanges can return every hash since the last poll. // // Stash/ClearStash/PublishStashed protect the FinalizeBlock→Commit // pairing with an internal mutex. Callers do NOT need to serialize @@ -43,8 +47,9 @@ type blockHeaderEvent struct { type BlockHeaderNotifier struct { ch chan blockHeaderEvent - mu sync.Mutex - pending *blockHeaderEvent + mu sync.Mutex + pending *blockHeaderEvent + listeners []func(blockHeaderEvent) } func NewBlockHeaderNotifier(capacity int) *BlockHeaderNotifier { @@ -83,9 +88,9 @@ func (n *BlockHeaderNotifier) ClearStash() { } // PublishStashed publishes the currently-stashed event (if any) on the -// fan-out channel and clears the stash. Returns true if an event was -// published, false otherwise (no stash, or nil receiver). Called after -// a successful Commit. +// fan-out channel and to subscribe listeners, then clears the stash. +// Returns true if an event was published, false otherwise (no stash, or +// nil receiver). Called after a successful Commit. // // Safe to call on a nil receiver. func (n *BlockHeaderNotifier) PublishStashed() bool { @@ -115,9 +120,21 @@ func (n *BlockHeaderNotifier) OnBlockCommitted(hash []byte, header *tmproto.Head n.publish(blockHeaderEvent{hash: hash, header: header, response: response}) } -// publish pushes evt onto the fan-out channel with overwrite-on-full -// semantics. Used by both PublishStashed and OnBlockCommitted so the -// channel-write code lives in one place. +// subscribe registers fn to be invoked for every published event. The +// callback runs on the publisher's goroutine. A nil receiver or nil fn +// is ignored. +func (n *BlockHeaderNotifier) subscribe(fn func(blockHeaderEvent)) { + if n == nil || fn == nil { + return + } + n.mu.Lock() + n.listeners = append(n.listeners, fn) + n.mu.Unlock() +} + +// publish pushes evt onto the newHeads channel with overwrite-on-full +// semantics, then invokes every subscribe listener. Used by both +// PublishStashed and OnBlockCommitted so the fan-out lives in one place. func (n *BlockHeaderNotifier) publish(evt blockHeaderEvent) { select { case n.ch <- evt: @@ -138,6 +155,15 @@ func (n *BlockHeaderNotifier) publish(evt blockHeaderEvent) { case n.ch <- evt: default: } + + n.mu.Lock() + listeners := make([]func(blockHeaderEvent), len(n.listeners)) + copy(listeners, n.listeners) + n.mu.Unlock() + for _, fn := range listeners { + defer recoverAndLog() + fn(evt) + } } func (n *BlockHeaderNotifier) recv() <-chan blockHeaderEvent { diff --git a/evmrpc/notifier_internal_test.go b/evmrpc/notifier_internal_test.go index ef6681d6aa..443a070f42 100644 --- a/evmrpc/notifier_internal_test.go +++ b/evmrpc/notifier_internal_test.go @@ -7,11 +7,12 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/hexutil" + "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" evmtypes "github.com/sei-protocol/sei-chain/x/evm/types" - "github.com/stretchr/testify/require" ) func TestBlockHeaderNotifier_DeliversEvent(t *testing.T) { @@ -62,14 +63,70 @@ func TestBlockHeaderNotifier_OverwritesWhenFull(t *testing.T) { } func TestBlockHeaderNotifier_NilReceiverIsNoOp(t *testing.T) { + // Setup: nil notifier, matching Autobahn-disabled App calls. var n *BlockHeaderNotifier - // Must not panic. + + // Test: every entry point, including subscribe. n.OnBlockCommitted(nil, &tmproto.Header{}, &abci.ResponseFinalizeBlock{}) n.Stash(&abci.RequestFinalizeBlock{Header: &tmproto.Header{}}, &abci.ResponseFinalizeBlock{}) n.ClearStash() + n.subscribe(func(blockHeaderEvent) {}) + + // Verify: no panic, and PublishStashed reports nothing to publish. require.False(t, n.PublishStashed()) } +func TestBlockHeaderNotifier_SubscribeGetsEveryEvent(t *testing.T) { + // Setup: capacity-1 notifier so newHeads would overwrite. + n := NewBlockHeaderNotifier(1) + // Setup: subscribe listener records every height. + var heights []int64 + n.subscribe(func(evt blockHeaderEvent) { + heights = append(heights, evt.header.Height) + }) + + // Test: publish two blocks without draining recv(). + n.OnBlockCommitted([]byte{1}, &tmproto.Header{Height: 1}, &abci.ResponseFinalizeBlock{}) + n.OnBlockCommitted([]byte{2}, &tmproto.Header{Height: 2}, &abci.ResponseFinalizeBlock{}) + + // Verify: subscribe saw both heights. + require.Equal(t, []int64{1, 2}, heights, "subscribe must not drop hashes the way newHeads overwrite does") + // Verify: recv() kept only the newest. + evt := <-n.recv() + require.EqualValues(t, 2, evt.header.Height, "newHeads channel still overwrite-on-full") +} + +func TestBlockHeaderNotifier_SubscribeDoesNotStealFromRecv(t *testing.T) { + // Setup: notifier with a no-op subscribe listener alongside recv(). + n := NewBlockHeaderNotifier(4) + n.subscribe(func(blockHeaderEvent) {}) + + // Test: publish one event. + n.OnBlockCommitted([]byte{9}, &tmproto.Header{Height: 9}, &abci.ResponseFinalizeBlock{}) + + // Verify: recv() still gets the event; subscribe did not consume the channel. + select { + case evt := <-n.recv(): + require.EqualValues(t, 9, evt.header.Height) + case <-time.After(time.Second): + t.Fatal("subscribe listener must not consume the newHeads channel") + } +} + +func TestBlockHeaderNotifier_SubscribePanicDoesNotSkipLaterListeners(t *testing.T) { + // Setup: first listener panics; second records that it ran. + n := NewBlockHeaderNotifier(4) + n.subscribe(func(blockHeaderEvent) { panic("listener boom") }) + var second bool + n.subscribe(func(blockHeaderEvent) { second = true }) + + // Test: one publish must fan out to every listener. + n.OnBlockCommitted([]byte{1}, &tmproto.Header{Height: 1}, &abci.ResponseFinalizeBlock{}) + + // Verify: the second listener still ran. + require.True(t, second) +} + // TestBlockHeaderNotifier_StashThenPublish covers the happy path used // by the App: Stash captures a tuple, PublishStashed publishes it and // clears the stash, second PublishStashed reports nothing to publish. diff --git a/evmrpc/server.go b/evmrpc/server.go index 36f85da013..123aeb446f 100644 --- a/evmrpc/server.go +++ b/evmrpc/server.go @@ -43,6 +43,7 @@ func NewEVMHTTPServer( txConfigProvider func(int64) client.TxConfig, homeDir string, stateStore types.StateStore, + blockHeaderNotifier *BlockHeaderNotifier, traceCtxProviders ...TraceContextProvider, ) (EVMServer, error) { @@ -162,6 +163,7 @@ func NewEVMHTTPServer( cacheCreationMutex, globalLogSlicePool, watermarks, + blockHeaderNotifier, ), }, { diff --git a/evmrpc/setup_test.go b/evmrpc/setup_test.go index 90bc51fd2c..5299373b10 100644 --- a/evmrpc/setup_test.go +++ b/evmrpc/setup_test.go @@ -22,6 +22,8 @@ import ( "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/rpc" "github.com/gorilla/websocket" + "github.com/stretchr/testify/require" + "github.com/sei-protocol/sei-chain/app" "github.com/sei-protocol/sei-chain/evmrpc" evmrpcconfig "github.com/sei-protocol/sei-chain/evmrpc/config" @@ -42,7 +44,6 @@ import ( "github.com/sei-protocol/sei-chain/x/evm/keeper" "github.com/sei-protocol/sei-chain/x/evm/types" "github.com/sei-protocol/sei-chain/x/evm/types/ethtx" - "github.com/stretchr/testify/require" ) const TestAddr = "127.0.0.1" @@ -51,7 +52,8 @@ const TestWSPort = 7778 const TestBadPort = 7779 const TestStrictPort = 7780 const TestArchivePort = 7782 -const TestNotifierWSPort = 7784 +const TestNotifierHTTPPort = 7783 // Autobahn eth_newBlockFilter HTTP server +const TestNotifierWSPort = 7784 // Autobahn eth_subscribe("newHeads") WS server const GenesisBlockHeight = 0 const MockHeight8 = 8 @@ -168,6 +170,11 @@ var NewHeadsCalled = make(chan struct{}, 1) // eth_subscribe("newHeads") through the in-process notifier path. var NotifierForTest = evmrpc.NewBlockHeaderNotifier(16) +// BlockFilterNotifierForTest backs the Autobahn-style HTTP server started +// on TestNotifierHTTPPort. Isolated from NotifierForTest so parallel +// newHeads and newBlockFilter tests do not steal each other's events. +var BlockFilterNotifierForTest = evmrpc.NewBlockHeaderNotifier(16) + type MockClient struct { client.Client latestOverride int64 @@ -682,7 +689,7 @@ func init() { goodConfig.MaxLogNoBlock = 10 goodConfig.EnabledLegacySeiApis = evmrpc.SeiLegacyAllGatedMethodNames() txConfigProvider := func(int64) client.TxConfig { return TxConfig } - HttpServer, err := evmrpc.NewEVMHTTPServer(goodConfig, &MockClient{}, EVMKeeper, testApp.BeginBlockKeepers, testApp.BaseApp, testApp.TracerAnteHandler, ctxProvider, txConfigProvider, "", nil) + HttpServer, err := evmrpc.NewEVMHTTPServer(goodConfig, &MockClient{}, EVMKeeper, testApp.BeginBlockKeepers, testApp.BaseApp, testApp.TracerAnteHandler, ctxProvider, txConfigProvider, "", nil, nil) if err != nil { panic(err) } @@ -694,7 +701,7 @@ func init() { badConfig := evmrpcconfig.DefaultConfig badConfig.HTTPPort = TestBadPort badConfig.FilterTimeout = 500 * time.Millisecond - badHTTPServer, err := evmrpc.NewEVMHTTPServer(badConfig, &MockBadClient{}, EVMKeeper, testApp.BeginBlockKeepers, testApp.BaseApp, testApp.TracerAnteHandler, ctxProvider, txConfigProvider, "", nil) + badHTTPServer, err := evmrpc.NewEVMHTTPServer(badConfig, &MockBadClient{}, EVMKeeper, testApp.BeginBlockKeepers, testApp.BaseApp, testApp.TracerAnteHandler, ctxProvider, txConfigProvider, "", nil, nil) if err != nil { panic(err) } @@ -719,6 +726,7 @@ func init() { txConfigProvider, "", nil, + nil, ) if err != nil { panic(err) @@ -743,6 +751,7 @@ func init() { txConfigProvider, "", nil, + nil, ) if err != nil { panic(err) @@ -761,12 +770,20 @@ func init() { } fmt.Printf("wsServer started with config = %+v\n", goodConfig) - // Start a second WS server wired to NotifierForTest, exercising the - // Autobahn (notifier-fed) eth_subscribe("newHeads") path. - notifierConfig := goodConfig - notifierConfig.HTTPPort = TestNotifierWSPort - 1 - notifierConfig.WSPort = TestNotifierWSPort - notifierWSServer, err := evmrpc.NewEVMWebSocketServer(notifierConfig, &MockClient{}, EVMKeeper, testApp.BeginBlockKeepers, testApp.BaseApp, testApp.TracerAnteHandler, ctxProvider, txConfigProvider, "", nil, NotifierForTest) + // Setup: Autobahn HTTP FilterAPI on its own notifier so parallel newHeads tests do not mix hashes. + notifierHTTPConfig := goodConfig + notifierHTTPConfig.HTTPPort = TestNotifierHTTPPort + notifierHTTPServer, err := evmrpc.NewEVMHTTPServer(notifierHTTPConfig, &MockClient{}, EVMKeeper, testApp.BeginBlockKeepers, testApp.BaseApp, testApp.TracerAnteHandler, ctxProvider, txConfigProvider, "", nil, BlockFilterNotifierForTest) + if err != nil { + panic(err) + } + if err := notifierHTTPServer.Start(); err != nil { + panic(err) + } + // Setup: Autobahn WS newHeads on NotifierForTest. + notifierWSConfig := goodConfig + notifierWSConfig.WSPort = TestNotifierWSPort + notifierWSServer, err := evmrpc.NewEVMWebSocketServer(notifierWSConfig, &MockClient{}, EVMKeeper, testApp.BeginBlockKeepers, testApp.BaseApp, testApp.TracerAnteHandler, ctxProvider, txConfigProvider, "", nil, NotifierForTest) if err != nil { panic(err) } diff --git a/evmrpc/tests/utils.go b/evmrpc/tests/utils.go index 0b2bb478f8..7559438959 100644 --- a/evmrpc/tests/utils.go +++ b/evmrpc/tests/utils.go @@ -162,6 +162,7 @@ func setupTestServer( func(int64) client.TxConfig { return a.GetTxConfig() }, "", a.GetStateStore(), + nil, ) if err != nil { panic(err) diff --git a/sei-tendermint/internal/rpc/core/mempool.go b/sei-tendermint/internal/rpc/core/mempool.go index 13a7cc82e4..d722ec7e79 100644 --- a/sei-tendermint/internal/rpc/core/mempool.go +++ b/sei-tendermint/internal/rpc/core/mempool.go @@ -110,32 +110,26 @@ func (env *Environment) BroadcastTx(ctx context.Context, req *coretypes.RequestB }, nil } -// BroadcastTxCommit returns with the responses from CheckTx and DeliverTx. +// ErrBroadcastTxCommitUnsupported is returned by BroadcastTxCommit when Autobahn +// is active. The transaction is not submitted; use BroadcastTx (CheckTx) and +// confirm inclusion from committed state. +var ErrBroadcastTxCommitUnsupported = errors.New("broadcast_tx_commit is not supported on Autobahn; use broadcast_tx_sync") + +// BroadcastTxCommit returns the CheckTx and DeliverTx results after inclusion. +// Under Autobahn it returns ErrBroadcastTxCommitUnsupported without submitting +// the transaction. // More: https://docs.tendermint.com/master/rpc/#/Tx/broadcast_tx_commit func (env *Environment) BroadcastTxCommit(ctx context.Context, req *coretypes.RequestBroadcastTx) (*coretypes.ResultBroadcastTxCommit, error) { + if env.gigaRouter().IsPresent() { + return nil, ErrBroadcastTxCommitUnsupported + } + if timeout := env.Config.TimeoutBroadcastTxCommit; timeout > 0 { var cancel context.CancelFunc ctx, cancel = context.WithTimeout(ctx, timeout) defer cancel() } - if giga, ok := env.gigaRouter().Get(); ok { - v, ok := giga.Mempool().Get() - if !ok { - return nil, errors.New("autobahn fullnode has no local mempool; broadcast_tx_* must be sent to a validator") - } - r, err := v.InsertTx(ctx, req.Tx) - if err != nil { - return nil, err - } - if r.Code != abci.CodeTypeOK { - return &coretypes.ResultBroadcastTxCommit{ - CheckTx: *r, - Hash: req.Tx.Hash().Bytes(), - }, nil - } - return env.broadcastTxCommitFromCheckTx(ctx, req, r) - } mp, err := env.requireMempool() if err != nil { return nil, err diff --git a/sei-tendermint/internal/rpc/core/mempool_autobahn_test.go b/sei-tendermint/internal/rpc/core/mempool_autobahn_test.go new file mode 100644 index 0000000000..9d9cd715d3 --- /dev/null +++ b/sei-tendermint/internal/rpc/core/mempool_autobahn_test.go @@ -0,0 +1,133 @@ +package core + +import ( + "errors" + "testing" + "time" + + dbm "github.com/tendermint/tm-db" + "golang.org/x/time/rate" + + "github.com/sei-protocol/sei-chain/sei-db/ledger_db/block/memblock" + abci "github.com/sei-protocol/sei-chain/sei-tendermint/abci/types" + atypes "github.com/sei-protocol/sei-chain/sei-tendermint/autobahn/types" + "github.com/sei-protocol/sei-chain/sei-tendermint/crypto/ed25519" + "github.com/sei-protocol/sei-chain/sei-tendermint/internal/autobahn/blockstore" + "github.com/sei-protocol/sei-chain/sei-tendermint/internal/autobahn/producer" + "github.com/sei-protocol/sei-chain/sei-tendermint/internal/p2p" + "github.com/sei-protocol/sei-chain/sei-tendermint/internal/p2p/conn" + "github.com/sei-protocol/sei-chain/sei-tendermint/internal/proxy" + "github.com/sei-protocol/sei-chain/sei-tendermint/internal/state/indexer" + kvsink "github.com/sei-protocol/sei-chain/sei-tendermint/internal/state/indexer/sink/kv" + "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils" + "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/require" + "github.com/sei-protocol/sei-chain/sei-tendermint/libs/utils/tcp" + "github.com/sei-protocol/sei-chain/sei-tendermint/rpc/coretypes" + "github.com/sei-protocol/sei-chain/sei-tendermint/types" +) + +func TestBroadcastTxCommitUnderAutobahnFailsFast(t *testing.T) { + // Setup: Giga validator RPC env with a real mempool and an empty KV indexer. + env := newAutobahnBroadcastEnv(t) + + // Test: BroadcastTxCommit with TimeoutBroadcastTxCommit=0 (hangs on a wait regression). + res, err := env.BroadcastTxCommit(t.Context(), &coretypes.RequestBroadcastTx{Tx: []byte("tx")}) + + // Verify: Autobahn sentinel and no result. + require.ErrorIs(t, err, ErrBroadcastTxCommitUnsupported) + require.Nil(t, res) +} + +func TestBroadcastTxCommitWithoutAutobahnUsesMempool(t *testing.T) { + // Setup: RPC env with no GigaRouter (Comet path). + env := &Environment{} + + // Test: BroadcastTxCommit with no local mempool either. + _, err := env.BroadcastTxCommit(t.Context(), &coretypes.RequestBroadcastTx{Tx: []byte("tx")}) + + // Verify: mempool error, not the Autobahn sentinel. + require.Error(t, err) + require.False(t, errors.Is(err, ErrBroadcastTxCommitUnsupported)) +} + +func newAutobahnBroadcastEnv(t *testing.T) *Environment { + t.Helper() + // Setup: one-validator committee and node identity. + rng := utils.TestRng() + _, keys := atypes.GenCommittee(rng, 1) + valKey := keys[0] + nodeKey := p2p.NodeSecretKey(ed25519.TestSecretKey(utils.GenBytes(rng, 32))) + + // Setup: genesis the router needs to build DataState. + genDoc := &types.GenesisDoc{ + ChainID: "broadcast-tx-commit-autobahn", + InitialHeight: 1, + GenesisTime: time.Now(), + ConsensusParams: types.DefaultConsensusParams(), + } + require.NoError(t, genDoc.ValidateAndComplete()) + + // Setup: in-process validator addr map and empty block store. + addrs := map[atypes.PublicKey]p2p.GigaNodeAddr{ + valKey.Public(): { + Key: nodeKey.Public(), + HostPort: tcp.HostPort{Hostname: "127.0.0.1", Port: 26657}, + }, + } + blockDB, err := blockstore.New(memblock.NewBlockDB()) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, blockDB.Close()) }) + + // Setup: Giga validator router so gigaRouter() is present and mempool exists. + commonCfg := p2p.GigaRouterCommonConfig{ + DialInterval: time.Second, + ValidatorAddrs: addrs, + App: proxy.New(&abci.BaseApplication{}), + GenDoc: genDoc, + } + dataState, err := p2p.BuildDataState(&commonCfg, blockDB) + require.NoError(t, err) + giga, err := p2p.NewGigaValidatorRouter(&p2p.GigaValidatorConfig{ + GigaRouterCommonConfig: commonCfg, + ValidatorKey: valKey, + ViewTimeout: func(atypes.View) time.Duration { return time.Hour }, + Producer: &producer.Config{ + MaxGasWantedPerBlock: 1, + MaxGasEstimatedPerBlock: 1, + MaxTxsPerBlock: 1, + MaxTxsPerSecond: utils.None[uint64](), + BlockInterval: time.Second, + }, + }, nodeKey, dataState) + require.NoError(t, err) + require.True(t, giga.Mempool().IsPresent(), "validator GigaRouter must expose a mempool so a regression would wait, not take the fullnode shortcut") + + // Setup: p2p Router wrapping that GigaRouter (Environment.gigaRouter reads this). + endpoint := p2p.Endpoint{AddrPort: tcp.TestReserveAddr()} + nodeInfo := types.NodeInfo{ + NodeID: nodeKey.Public().NodeID(), + ListenAddr: endpoint.String(), + Moniker: string(nodeKey.Public().NodeID()), + Network: genDoc.ChainID, + } + router, err := p2p.NewRouter( + nodeKey, + func() *types.NodeInfo { return &nodeInfo }, + dbm.NewMemDB(), + &p2p.RouterOptions{ + Endpoint: endpoint, + Connection: conn.DefaultMConnConfig(), + IncomingConnectionWindow: utils.Some(time.Duration(0)), + MaxAcceptRate: rate.Inf, + MaxDialRate: rate.Inf, + Giga: utils.Some[p2p.GigaRouter](giga), + }, + ) + require.NoError(t, err) + + // Setup: empty KV indexer; TimeoutBroadcastTxCommit stays 0 so a wait regression hangs. + return &Environment{ + Router: router, + EventSinks: []indexer.EventSink{kvsink.NewEventSink(dbm.NewMemDB(), nil)}, + } +}