diff --git a/sei-tendermint/internal/p2p/giga_router.go b/sei-tendermint/internal/p2p/giga_router.go index 5a156dbc42..09a30855ea 100644 --- a/sei-tendermint/internal/p2p/giga_router.go +++ b/sei-tendermint/internal/p2p/giga_router.go @@ -52,6 +52,10 @@ type GigaRouterCommonConfig struct { // Whether validator should proxy txs which do not belong to the local node. EnableEvmProxy bool + + // EventBus receives NewBlock, NewBlockHeader, and per-tx events after + // each Autobahn commit. + EventBus types.BlockEventPublisher } // GigaValidatorConfig configures a committee-member GigaRouter. diff --git a/sei-tendermint/internal/p2p/giga_router_common.go b/sei-tendermint/internal/p2p/giga_router_common.go index 34790bf156..95e7ff9586 100644 --- a/sei-tendermint/internal/p2p/giga_router_common.go +++ b/sei-tendermint/internal/p2p/giga_router_common.go @@ -244,6 +244,8 @@ func (r *gigaRouterCommon) executeBlock(ctx context.Context, b *atypes.GlobalBlo if err != nil { return nil, fmt.Errorf("app.Commit(): %w", err) } + // Indexer-backed RPCs (broadcast_tx_commit, /tx) wait on these events. + r.publishExecutedBlockEvents(b, proposerAddress, resp) if err := r.data.PushAppHash(ctx, b.GlobalNumber, resp.AppHash); err != nil { return nil, fmt.Errorf("r.data.PushAppHash(%v): %w", b.GlobalNumber, err) } @@ -251,6 +253,53 @@ func (r *gigaRouterCommon) executeBlock(ctx context.Context, b *atypes.GlobalBlo return commitResp, nil } +// publishExecutedBlockEvents publishes NewBlock, NewBlockHeader, and per-tx +// events for a committed Autobahn block. +func (r *gigaRouterCommon) publishExecutedBlockEvents( + b *atypes.GlobalBlock, + proposer types.Address, + resp *abci.ResponseFinalizeBlock, +) { + translated := r.translateGlobalBlock(b) + translated.Block.Header.ProposerAddress = proposer + block := translated.Block + blockID := translated.BlockID + if len(resp.TxResults) != len(block.Txs) { + panic(fmt.Sprintf("number of TXs (%d) and ABCI TX responses (%d) do not match", + len(block.Txs), len(resp.TxResults))) + } + eventBus := r.cfg.EventBus + + if err := eventBus.PublishEventNewBlock(types.EventDataNewBlock{ + Block: block, + BlockID: blockID, + ResultFinalizeBlock: *resp, + }); err != nil { + logger.Error("failed publishing new block", "err", err) + } + + if err := eventBus.PublishEventNewBlockHeader(types.EventDataNewBlockHeader{ + Header: block.Header, + NumTxs: int64(len(block.Txs)), + ResultFinalizeBlock: *resp, + }); err != nil { + logger.Error("failed publishing new block header", "err", err) + } + + for i, tx := range block.Txs { + if err := eventBus.PublishEventTx(types.EventDataTx{ + TxResultV2: abci.TxResultV2{ + Height: block.Height, + Index: uint32(i), //nolint:gosec // i is bounded by block.Txs length which fits in uint32 + Tx: tx, + Result: *(resp.TxResults[i]), + }, + }); err != nil { + logger.Error("failed publishing event TX", "err", err) + } + } +} + // manages lifecycle of evmrpc connections to validators. func (r *gigaRouterCommon) runEvmProxies(ctx context.Context) error { return scope.Run(ctx, func(ctx context.Context, s scope.Scope) error { diff --git a/sei-tendermint/internal/p2p/giga_router_common_test.go b/sei-tendermint/internal/p2p/giga_router_common_test.go index 8df44c43f4..ba2e34883a 100644 --- a/sei-tendermint/internal/p2p/giga_router_common_test.go +++ b/sei-tendermint/internal/p2p/giga_router_common_test.go @@ -7,6 +7,8 @@ import ( "time" ethrpc "github.com/ethereum/go-ethereum/rpc" + dbm "github.com/tendermint/tm-db" + "github.com/sei-protocol/sei-chain/sei-db/ledger_db/block/memblock" "github.com/sei-protocol/sei-chain/sei-db/state_db/sc/hashvault" abci "github.com/sei-protocol/sei-chain/sei-tendermint/abci/types" @@ -14,7 +16,10 @@ import ( "github.com/sei-protocol/sei-chain/sei-tendermint/internal/autobahn/blockstore" "github.com/sei-protocol/sei-chain/sei-tendermint/internal/autobahn/data" "github.com/sei-protocol/sei-chain/sei-tendermint/internal/autobahn/epoch" + "github.com/sei-protocol/sei-chain/sei-tendermint/internal/eventbus" "github.com/sei-protocol/sei-chain/sei-tendermint/internal/proxy" + "github.com/sei-protocol/sei-chain/sei-tendermint/internal/state/indexer" + "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" tmtypes "github.com/sei-protocol/sei-chain/sei-tendermint/types" @@ -31,6 +36,26 @@ func registerEvmProxyForTest(t *testing.T, router *gigaRouterCommon, validator a return client } +func startedEventBus(t *testing.T) *eventbus.EventBus { + t.Helper() + bus := eventbus.NewDefault() + require.NoError(t, bus.Start(t.Context())) + t.Cleanup(bus.Wait) + return bus +} + +func startedTxIndexer(t *testing.T, bus *eventbus.EventBus) indexer.EventSink { + t.Helper() + sink := kv.NewEventSink(dbm.NewMemDB(), nil) + svc := indexer.NewService(indexer.ServiceArgs{ + Sinks: []indexer.EventSink{sink}, + EventBus: bus, + }) + require.NoError(t, svc.Start(t.Context())) + t.Cleanup(svc.Wait) + return sink +} + type fixedHeightApp struct { abci.BaseApplication height int64 diff --git a/sei-tendermint/internal/p2p/giga_router_events_test.go b/sei-tendermint/internal/p2p/giga_router_events_test.go new file mode 100644 index 0000000000..5f9cba8721 --- /dev/null +++ b/sei-tendermint/internal/p2p/giga_router_events_test.go @@ -0,0 +1,164 @@ +package p2p + +import ( + "testing" + "time" + + 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/internal/eventbus" + tmpubsub "github.com/sei-protocol/sei-chain/sei-tendermint/internal/pubsub" + tmquery "github.com/sei-protocol/sei-chain/sei-tendermint/internal/pubsub/query" + "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/types" +) + +func testGlobalBlock(t *testing.T, rng utils.Rng, height atypes.GlobalBlockNumber, txs [][]byte) *atypes.GlobalBlock { + t.Helper() + payload, err := atypes.PayloadBuilder{ + CreatedAt: time.Now(), + Txs: txs, + }.Build() + require.NoError(t, err) + block := atypes.NewBlock(atypes.GenLaneID(rng), 1, atypes.GenBlockHeaderHash(rng), payload) + return &atypes.GlobalBlock{ + Header: block.Header(), + Timestamp: time.Now(), + GlobalNumber: height, + Payload: payload, + } +} + +func subscribe(t *testing.T, bus *eventbus.EventBus, clientID string, query *tmquery.Query, limit int) eventbus.Subscription { + t.Helper() + sub, err := bus.SubscribeWithArgs(t.Context(), tmpubsub.SubscribeArgs{ + ClientID: clientID, + Query: query, + Limit: limit, + }) + require.NoError(t, err) + return sub +} + +func waitForN(t *testing.T, sub eventbus.Subscription, n int) { + t.Helper() + for range n { + _, err := sub.Next(t.Context()) + require.NoError(t, err) + } +} + +func TestPublishExecutedBlockEventsIndexesTxs(t *testing.T) { + rng := utils.TestRng() + + // Setup: EventBus, KV indexer, and subscriptions for header + tx events. + genDoc := &types.GenesisDoc{ChainID: "giga-index-test", InitialHeight: 1} + require.NoError(t, genDoc.ValidateAndComplete()) + + bus := startedEventBus(t) + sink := startedTxIndexer(t, bus) + txSub := subscribe(t, bus, "txs", types.EventQueryTx, 8) + headerSub := subscribe(t, bus, "headers", types.EventQueryNewBlockHeader, 8) + + // Setup: committed Autobahn block with two txs and matching FinalizeBlock results. + txs := [][]byte{[]byte("tx-a"), []byte("tx-b")} + height := atypes.GlobalBlockNumber(7) + gb := testGlobalBlock(t, rng, height, txs) + resp := &abci.ResponseFinalizeBlock{ + TxResults: []*abci.ExecTxResult{ + {Code: abci.CodeTypeOK, GasUsed: 11}, + {Code: abci.CodeTypeOK, GasUsed: 22}, + }, + } + + // Test: publish NewBlock / NewBlockHeader / per-tx events as executeBlock does. + r := &gigaRouterCommon{cfg: &GigaRouterCommonConfig{GenDoc: genDoc, EventBus: bus}} + r.publishExecutedBlockEvents(gb, nil, resp) + + // Validate: indexer has the block and each tx by hash, with height/index/gas. + waitForN(t, headerSub, 1) + waitForN(t, txSub, len(txs)) + + ok, err := sink.HasBlock(int64(height)) + require.NoError(t, err) + require.True(t, ok) + + for i, tx := range txs { + got, err := sink.GetTxByHash(types.Tx(tx).Hash().Bytes()) + require.NoError(t, err) + require.NotNil(t, got) + require.Equal(t, int64(height), got.Height) + require.Equal(t, uint32(i), got.Index) + require.Equal(t, tx, []byte(got.Tx)) + require.Equal(t, resp.TxResults[i].GasUsed, got.Result.GasUsed) + } +} + +func TestPublishExecutedBlockEventsIndexesEmptyBlock(t *testing.T) { + rng := utils.TestRng() + + // Setup: EventBus, KV indexer, and a header subscription. + genDoc := &types.GenesisDoc{ChainID: "giga-index-empty", InitialHeight: 1} + require.NoError(t, genDoc.ValidateAndComplete()) + + bus := startedEventBus(t) + sink := startedTxIndexer(t, bus) + headerSub := subscribe(t, bus, "headers", types.EventQueryNewBlockHeader, 4) + + // Test: publish events for a committed block with no txs. + height := atypes.GlobalBlockNumber(3) + gb := testGlobalBlock(t, rng, height, nil) + r := &gigaRouterCommon{cfg: &GigaRouterCommonConfig{GenDoc: genDoc, EventBus: bus}} + r.publishExecutedBlockEvents(gb, nil, &abci.ResponseFinalizeBlock{}) + + // Validate: indexer recorded the empty block. + waitForN(t, headerSub, 1) + + ok, err := sink.HasBlock(int64(height)) + require.NoError(t, err) + require.True(t, ok) +} + +func TestPublishExecutedBlockEventsPanicsBeforeDispatchOnTxCountMismatch(t *testing.T) { + rng := utils.TestRng() + + // Setup: counting EventBus and a one-tx block whose TxResults length does not match. + genDoc := &types.GenesisDoc{ChainID: "giga-index-mismatch", InitialHeight: 1} + require.NoError(t, genDoc.ValidateAndComplete()) + + bus := &countingBlockEvents{} + gb := testGlobalBlock(t, rng, 1, [][]byte{[]byte("tx-a")}) + r := &gigaRouterCommon{cfg: &GigaRouterCommonConfig{GenDoc: genDoc, EventBus: bus}} + + // Test: publish with empty TxResults. + require.Panics(t, func() { + r.publishExecutedBlockEvents(gb, nil, &abci.ResponseFinalizeBlock{}) + }) + + // Validate: no EventBus methods ran before the panic. + require.Equal(t, 0, bus.n) +} + +type countingBlockEvents struct{ n int } + +func (c *countingBlockEvents) PublishEventNewBlock(types.EventDataNewBlock) error { + c.n++ + return nil +} +func (c *countingBlockEvents) PublishEventNewBlockHeader(types.EventDataNewBlockHeader) error { + c.n++ + return nil +} +func (c *countingBlockEvents) PublishEventNewEvidence(types.EventDataNewEvidence) error { + c.n++ + return nil +} +func (c *countingBlockEvents) PublishEventTx(types.EventDataTx) error { + c.n++ + return nil +} +func (c *countingBlockEvents) PublishEventValidatorSetUpdates(types.EventDataValidatorSetUpdates) error { + c.n++ + return nil +} diff --git a/sei-tendermint/internal/p2p/giga_router_validator_test.go b/sei-tendermint/internal/p2p/giga_router_validator_test.go index 535f6d4f8b..023b717776 100644 --- a/sei-tendermint/internal/p2p/giga_router_validator_test.go +++ b/sei-tendermint/internal/p2p/giga_router_validator_test.go @@ -18,10 +18,12 @@ import ( atypes "github.com/sei-protocol/sei-chain/sei-tendermint/autobahn/types" "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/eventbus" "github.com/sei-protocol/sei-chain/sei-tendermint/internal/p2p/conn" "github.com/sei-protocol/sei-chain/sei-tendermint/internal/p2p/giga" "github.com/sei-protocol/sei-chain/sei-tendermint/internal/p2p/rpc" "github.com/sei-protocol/sei-chain/sei-tendermint/internal/proxy" + "github.com/sei-protocol/sei-chain/sei-tendermint/internal/state/indexer" "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/scope" @@ -60,6 +62,8 @@ func TestGigaRouter_FinalizeBlocks(t *testing.T) { var apps []*testApp var gigas []*gigaValidatorRouter var allTxs [][]byte + var indexSink indexer.EventSink + var indexSub eventbus.Subscription for i, cfg := range cfgs { nodeInfo := makeInfo(cfg.nodeKey) nodeInfo.ListenAddr = cfg.addr.String() @@ -80,6 +84,12 @@ func TestGigaRouter_FinalizeBlocks(t *testing.T) { blockStore, err := blockstore.New(db) require.NoError(t, err, "blockstore.New[%v]", i) t.Cleanup(func() { _ = blockStore.Close() }) + bus := startedEventBus(t) + if i == 0 { + // Setup: node 0 KV indexer (the sink BroadcastTxCommit polls). + indexSink = startedTxIndexer(t, bus) + indexSub = subscribe(t, bus, "node0-txs", types.EventQueryTx, 1024) + } commonCfg := GigaRouterCommonConfig{ // Aggressive dialing rate to speed up startup. DialInterval: 100 * time.Millisecond, @@ -88,6 +98,7 @@ func TestGigaRouter_FinalizeBlocks(t *testing.T) { App: proxyApp, GenDoc: genDoc, EnableEvmProxy: true, + EventBus: bus, } dataState, err := BuildDataState(&commonCfg, blockStore) require.NoError(t, err, "BuildDataState[%v]", i) @@ -146,6 +157,14 @@ func TestGigaRouter_FinalizeBlocks(t *testing.T) { require.NoError(t, app.WaitForTx(ctx, tx), "WaitForTx") } } + // Validate: every submitted tx is in the KV indexer by hash. + waitForN(t, indexSub, len(allTxs)) + for _, tx := range allTxs { + got, err := indexSink.GetTxByHash(types.Tx(tx).Hash().Bytes()) + require.NoError(t, err, "GetTxByHash") + require.NotNil(t, got) + require.Equal(t, tx, []byte(got.Tx)) + } // Nodes should agree on the final state. want := apps[0].Snapshot() for i, app := range apps { 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..bd2f1d86d8 --- /dev/null +++ b/sei-tendermint/internal/rpc/core/mempool_autobahn_test.go @@ -0,0 +1,208 @@ +package core + +import ( + "context" + "fmt" + "net/netip" + "net/url" + "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/config" + "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/eventbus" + "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" + "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/scope" + "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" +) + +const autobahnBroadcastCommitteeSize = 4 + +// okApp is a minimal ABCI app whose CheckTx always succeeds. +type okApp struct{ abci.BaseApplication } + +func (okApp) CheckTx(context.Context, *abci.RequestCheckTxV2) *abci.ResponseCheckTxV2 { + return &abci.ResponseCheckTxV2{ResponseCheckTx: &abci.ResponseCheckTx{ + Code: abci.CodeTypeOK, + GasWanted: 1, + GasEstimated: 1, + }} +} + +type autobahnBroadcastNode struct { + router *p2p.Router + giga p2p.GigaRouter +} + +// TestBroadcastTxCommitUnderAutobahn verifies BroadcastTxCommit returns CheckTx +// and DeliverTx results once Autobahn has included the transaction. +func TestBroadcastTxCommitUnderAutobahn(t *testing.T) { + ctx := t.Context() + rng := utils.TestRng() + + // Setup: committee identities and p2p listen addresses. + _, keys := atypes.GenCommittee(rng, autobahnBroadcastCommitteeSize) + type nodeKeys struct { + validator atypes.SecretKey + node p2p.NodeSecretKey + addr netip.AddrPort + } + cfgs := make([]nodeKeys, autobahnBroadcastCommitteeSize) + addrs := map[atypes.PublicKey]p2p.GigaNodeAddr{} + for i, key := range keys { + nodeKey := p2p.NodeSecretKey(ed25519.TestSecretKey(utils.GenBytes(rng, 32))) + addr := tcp.TestReserveAddr() + cfgs[i] = nodeKeys{validator: key, node: nodeKey, addr: addr} + addrs[key.Public()] = p2p.GigaNodeAddr{ + Key: nodeKey.Public(), + HostPort: tcp.HostPort{Hostname: addr.Addr().String(), Port: addr.Port()}, + EVMRPC: utils.OrPanic1(url.Parse(fmt.Sprintf("http://%s:8545", addr.Addr().String()))), + } + } + + // Setup: shared genesis. + genDoc := &types.GenesisDoc{ + ChainID: "autobahn-broadcast-tx-commit", + InitialHeight: 1, + } + require.NoError(t, genDoc.ValidateAndComplete()) + + // Setup: per-node EventBus, giga validator, and p2p router. Node 0 also runs the KV indexer. + var indexSink indexer.EventSink + var indexBus *eventbus.EventBus + nodes := make([]autobahnBroadcastNode, autobahnBroadcastCommitteeSize) + for i, cfg := range cfgs { + bus := eventbus.NewDefault() + require.NoError(t, bus.Start(ctx)) + t.Cleanup(bus.Wait) + if i == 0 { + indexBus = bus + indexSink = kv.NewEventSink(dbm.NewMemDB(), nil) + idx := indexer.NewService(indexer.ServiceArgs{ + Sinks: []indexer.EventSink{indexSink}, + EventBus: bus, + }) + require.NoError(t, idx.Start(ctx)) + t.Cleanup(idx.Wait) + } + + db := memblock.NewBlockDB() + blockStore, err := blockstore.New(db) + require.NoError(t, err) + t.Cleanup(func() { _ = blockStore.Close() }) + commonCfg := p2p.GigaRouterCommonConfig{ + DialInterval: 100 * time.Millisecond, + ValidatorAddrs: addrs, + App: proxy.New(okApp{}), + GenDoc: genDoc, + HashVaultDisabledUnsafe: true, + EventBus: bus, + } + dataState, err := p2p.BuildDataState(&commonCfg, blockStore) + require.NoError(t, err) + + giga, err := p2p.NewGigaValidatorRouter(&p2p.GigaValidatorConfig{ + GigaRouterCommonConfig: commonCfg, + ValidatorKey: cfg.validator, + ViewTimeout: func(atypes.View) time.Duration { return time.Hour }, + Producer: &producer.Config{ + MaxGasWantedPerBlock: 1_000_000, + MaxGasEstimatedPerBlock: 1_000_000, + MaxTxsPerBlock: 16, + MaxTxsPerSecond: utils.None[uint64](), + BlockInterval: 100 * time.Millisecond, + AllowEmptyBlocks: false, + }, + }, cfg.node, dataState) + require.NoError(t, err) + + nodeInfo := autobahnBroadcastNodeInfo(cfg.node, cfg.addr.String(), genDoc.ChainID) + endpoint := p2p.Endpoint{AddrPort: cfg.addr} + router, err := p2p.NewRouter( + cfg.node, + func() *types.NodeInfo { return &nodeInfo }, + dbm.NewMemDB(), + &p2p.RouterOptions{ + SelfAddress: utils.Some(endpoint.NodeAddress(cfg.node.Public().NodeID())), + Endpoint: endpoint, + Connection: conn.DefaultMConnConfig(), + IncomingConnectionWindow: utils.Some(time.Duration(0)), + MaxAcceptRate: rate.Inf, + MaxDialRate: rate.Limit(30), + Giga: utils.Some[p2p.GigaRouter](giga), + }, + ) + require.NoError(t, err) + nodes[i] = autobahnBroadcastNode{router: router, giga: giga} + } + + // Setup: RPC Environment on node 0 (Autobahn router + KV sink). + env := &Environment{ + Router: nodes[0].router, + EventSinks: []indexer.EventSink{indexSink}, + EventBus: indexBus, + Config: config.RPCConfig{}, + } + + // Test: start the cluster, then BroadcastTxCommit until inclusion (or hang). + tx := types.Tx(utils.GenBytes(rng, 32)) + var res *coretypes.ResultBroadcastTxCommit + err := scope.Run(ctx, func(ctx context.Context, s scope.Scope) error { + for i, n := range nodes { + s.SpawnBgNamed(fmt.Sprintf("router[%v]", i), func() error { + return utils.IgnoreCancel(n.router.Run(ctx)) + }) + s.SpawnBgNamed(fmt.Sprintf("giga[%v]", i), func() error { + return utils.IgnoreCancel(n.giga.Run(ctx)) + }) + } + var err error + res, err = env.BroadcastTxCommit(ctx, &coretypes.RequestBroadcastTx{Tx: tx}) + return err + }) + + // Validate: CheckTx OK, hash matches, height assigned, DeliverTx OK. + require.NoError(t, err) + require.NotNil(t, res) + require.Equal(t, abci.CodeTypeOK, res.CheckTx.Code) + require.Equal(t, tx.Hash().Bytes(), res.Hash) + require.Positive(t, res.Height) + require.Equal(t, abci.CodeTypeOK, res.TxResult.Code) +} + +func autobahnBroadcastNodeInfo(key p2p.NodeSecretKey, listen, chainID string) types.NodeInfo { + nodeID := key.Public().NodeID() + return types.NodeInfo{ + NodeID: nodeID, + ListenAddr: listen, + Network: chainID, + Moniker: string(nodeID), + Channels: []byte{}, + ProtocolVersion: types.ProtocolVersion{ + P2P: 1, + Block: 2, + App: 3, + }, + Version: "1.2.3", + Other: types.NodeInfoOther{ + TxIndex: "on", + RPCAddress: "127.0.0.1:26657", + }, + } +} diff --git a/sei-tendermint/node/node.go b/sei-tendermint/node/node.go index 0ac292db3a..5a5b7c24c9 100644 --- a/sei-tendermint/node/node.go +++ b/sei-tendermint/node/node.go @@ -292,6 +292,7 @@ func makeNode( utils.Some(proxyApp), genDoc, dbProvider, + eventBus, ) closers = append(closers, peerCloser) if err != nil { diff --git a/sei-tendermint/node/seed.go b/sei-tendermint/node/seed.go index 02991774c9..47834a7fff 100644 --- a/sei-tendermint/node/seed.go +++ b/sei-tendermint/node/seed.go @@ -78,6 +78,8 @@ func makeSeedNode( return nil, err } + eventBus := eventbus.NewDefault() + router, peerCloser, _, err := createRouter( func() *types.NodeInfo { return &nodeInfo }, nodeKey, @@ -86,6 +88,7 @@ func makeSeedNode( utils.None[*proxy.Proxy](), genDoc, dbProvider, + eventBus, ) closers = append(closers, peerCloser) if err != nil { @@ -107,7 +110,6 @@ func makeSeedNode( if err != nil { return nil, fmt.Errorf("sink.EventSinksFromConfig(): %w", err) } - eventBus := eventbus.NewDefault() stateStore := sm.NewStore(stateDB) diff --git a/sei-tendermint/node/setup.go b/sei-tendermint/node/setup.go index 092f4c3e16..503b2fa8f1 100644 --- a/sei-tendermint/node/setup.go +++ b/sei-tendermint/node/setup.go @@ -290,6 +290,7 @@ func buildGigaRouter( validatorKey utils.Option[atypes.SecretKey], app *proxy.Proxy, genDoc *types.GenesisDoc, + eventBus types.BlockEventPublisher, ) (p2p.GigaRouter, atypes.BlockStore, error) { fc, validatorAddrs, err := loadAutobahnCommittee(cfg.AutobahnConfigFile) if err != nil { @@ -325,6 +326,7 @@ func buildGigaRouter( // The GigaRouter builds and owns the equivocation guard itself; just pass the operator's // enable/disable decision through as plain config. valCfg.HashVaultDisabledUnsafe = cfg.HashVaultDisabledUnsafe + valCfg.EventBus = eventBus logger.Info("Autobahn: starting as validator", "validators", len(valCfg.ValidatorAddrs)) blockStore, err := openBlockStore(&valCfg.GigaRouterCommonConfig, fc.BlockDB) if err != nil { @@ -352,6 +354,7 @@ func buildGigaRouter( // The GigaRouter builds and owns the equivocation guard itself; just pass the operator's // enable/disable decision through as plain config. fnCfg.HashVaultDisabledUnsafe = cfg.HashVaultDisabledUnsafe + fnCfg.EventBus = eventBus logger.Info("Autobahn: starting as fullnode", "mode", cfg.Mode, "validators", len(validatorAddrs)) blockStore, err := openBlockStore(fnCfg, fc.BlockDB) if err != nil { @@ -535,6 +538,7 @@ func createRouter( app utils.Option[*proxy.Proxy], genDoc *types.GenesisDoc, dbProvider config.DBProvider, + eventBus types.BlockEventPublisher, ) (*p2p.Router, closer, utils.Option[atypes.BlockStore], error) { closer := func() error { return nil } noneDB := utils.None[atypes.BlockStore]() @@ -596,7 +600,7 @@ func createRouter( if !ok { return nil, closer, noneDB, fmt.Errorf("autobahn requires app") } - giga, blockStore, err := buildGigaRouter(cfg, nodeKey, validatorKey, proxyApp, genDoc) + giga, blockStore, err := buildGigaRouter(cfg, nodeKey, validatorKey, proxyApp, genDoc, eventBus) if err != nil { return nil, closer, noneDB, err }