Publish Tx Events (cosmos#193)

Author: tzdybal

block/manager.go

@@ -65,6 +65,7 @@ func NewManager(
 	mempool mempool.Mempool,
 	proxyApp proxy.AppConnConsensus,
 	dalc da.DataAvailabilityLayerClient,
+ 	eventBus *tmtypes.EventBus,
 	logger log.Logger,
 ) (*Manager, error) {
 	s, err := getInitialState(store, genesis)
@@ -77,7 +78,7 @@ func NewManager(
 		return nil, err
 	}
 
-	exec := state.NewBlockExecutor(proposerAddress, conf.NamespaceID, genesis.ChainID, mempool, proxyApp, logger)
+	exec := state.NewBlockExecutor(proposerAddress, conf.NamespaceID, genesis.ChainID, mempool, proxyApp, eventBus, logger)
 	if s.LastBlockHeight+1 == genesis.InitialHeight {
 		res, err := exec.InitChain(genesis)
 		if err != nil {

block/manager_test.go

@@ -72,7 +72,7 @@ func TestInitialState(t *testing.T) {
 			assert := assert.New(t)
 			logger := log.TestingLogger()
 			dalc := getMockDALC(logger)
-			agg, err := NewManager(key, conf, c.genesis, c.store, nil, nil, dalc, logger)
+			agg, err := NewManager(key, conf, c.genesis, c.store, nil, nil, dalc, nil, logger)
 			assert.NoError(err)
 			assert.NotNil(agg)
 			assert.Equal(c.expectedChainID, agg.lastState.ChainID)

node/node.go

@@ -101,7 +101,7 @@ func NewNode(ctx context.Context, conf config.NodeConfig, nodeKey crypto.PrivKey
 	mp := mempool.NewCListMempool(llcfg.DefaultMempoolConfig(), proxyApp.Mempool(), 0)
 	mpIDs := newMempoolIDs()
 
-	blockManager, err := block.NewManager(nodeKey, conf.BlockManagerConfig, genesis, s, mp, proxyApp.Consensus(), dalc, logger.With("module", "BlockManager"))
+	blockManager, err := block.NewManager(nodeKey, conf.BlockManagerConfig, genesis, s, mp, proxyApp.Consensus(), dalc, eventBus, logger.With("module", "BlockManager"))
 	if err != nil {
 		return nil, fmt.Errorf("BlockManager initialization error: %w", err)
 	}

state/executor.go

@@ -11,6 +11,7 @@ import (
 	tmproto "github.com/tendermint/tendermint/proto/tendermint/types"
 	"github.com/tendermint/tendermint/proxy"
 	tmtypes "github.com/tendermint/tendermint/types"
+	"go.uber.org/multierr"
 
 	abciconv "github.com/celestiaorg/optimint/conv/abci"
 	"github.com/celestiaorg/optimint/log"
@@ -26,18 +27,21 @@ type BlockExecutor struct {
 	proxyApp        proxy.AppConnConsensus
 	mempool         mempool.Mempool
 
+	eventBus *tmtypes.EventBus
+
 	logger log.Logger
 }
 
 // NewBlockExecutor creates new instance of BlockExecutor.
 // Proposer address and namespace ID will be used in all newly created blocks.
-func NewBlockExecutor(proposerAddress []byte, namespaceID [8]byte, chainID string, mempool mempool.Mempool, proxyApp proxy.AppConnConsensus, logger log.Logger) *BlockExecutor {
+func NewBlockExecutor(proposerAddress []byte, namespaceID [8]byte, chainID string, mempool mempool.Mempool, proxyApp proxy.AppConnConsensus, eventBus *tmtypes.EventBus, logger log.Logger) *BlockExecutor {
 	return &BlockExecutor{
 		proposerAddress: proposerAddress,
 		namespaceID:     namespaceID,
 		chainID:         chainID,
 		proxyApp:        proxyApp,
 		mempool:         mempool,
+		eventBus:        eventBus,
 		logger:          logger,
 	}
 }
@@ -129,6 +133,11 @@ func (e *BlockExecutor) ApplyBlock(ctx context.Context, state State, block *type
 
 	copy(state.AppHash[:], appHash[:])
 
+	err = e.publishEvents(resp, block)
+	if err != nil {
+		e.logger.Error("failed to fire block events", "error", err)
+	}
+
 	return state, retainHeight, nil
 }
 
@@ -271,6 +280,46 @@ func (e *BlockExecutor) getLastCommitHash(lastCommit *types.Commit, header *type
 	return lastABCICommit.Hash()
 }
 
+func (e *BlockExecutor) publishEvents(resp *tmstate.ABCIResponses, block *types.Block) error {
+	if e.eventBus == nil {
+		return nil
+	}
+
+	abciBlock, err := abciconv.ToABCIBlock(block)
+	if err != nil {
+		return err
+	}
+
+	err = multierr.Append(err, e.eventBus.PublishEventNewBlock(tmtypes.EventDataNewBlock{
+		Block:            abciBlock,
+		ResultBeginBlock: *resp.BeginBlock,
+		ResultEndBlock:   *resp.EndBlock,
+	}))
+	err = multierr.Append(err, e.eventBus.PublishEventNewBlockHeader(tmtypes.EventDataNewBlockHeader{
+		Header:           abciBlock.Header,
+		NumTxs:           int64(len(abciBlock.Txs)),
+		ResultBeginBlock: *resp.BeginBlock,
+		ResultEndBlock:   *resp.EndBlock,
+	}))
+	for _, ev := range abciBlock.Evidence.Evidence {
+		err = multierr.Append(err, e.eventBus.PublishEventNewEvidence(tmtypes.EventDataNewEvidence{
+			Evidence: ev,
+			Height:   int64(block.Header.Height),
+		}))
+	}
+	for i, dtx := range resp.DeliverTxs {
+		err = multierr.Append(err, e.eventBus.PublishEventTx(tmtypes.EventDataTx{
+			TxResult: abci.TxResult{
+				Height: int64(block.Header.Height),
+				Index:  uint32(i),
+				Tx:     abciBlock.Data.Txs[i],
+				Result: *dtx,
+			},
+		}))
+	}
+	return err
+}
+
 func toOptimintTxs(txs tmtypes.Txs) types.Txs {
 	optiTxs := make(types.Txs, len(txs))
 	for i := range txs {

state/executor_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/rand"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
@@ -12,7 +13,9 @@ import (
 	abci "github.com/tendermint/tendermint/abci/types"
 	cfg "github.com/tendermint/tendermint/config"
 	"github.com/tendermint/tendermint/libs/log"
+	"github.com/tendermint/tendermint/libs/pubsub/query"
 	"github.com/tendermint/tendermint/proxy"
+	tmtypes "github.com/tendermint/tendermint/types"
 
 	"github.com/celestiaorg/optimint/mempool"
 	"github.com/celestiaorg/optimint/mocks"
@@ -35,7 +38,7 @@ func TestCreateBlock(t *testing.T) {
 	nsID := [8]byte{1, 2, 3, 4, 5, 6, 7, 8}
 
 	mpool := mempool.NewCListMempool(cfg.DefaultMempoolConfig(), proxy.NewAppConnMempool(client), 0)
-	executor := NewBlockExecutor([]byte("test address"), nsID, "test", mpool, proxy.NewAppConnConsensus(client), logger)
+	executor := NewBlockExecutor([]byte("test address"), nsID, "test", mpool, proxy.NewAppConnConsensus(client), nil, logger)
 
 	state := State{}
 	state.ConsensusParams.Block.MaxBytes = 100
@@ -91,7 +94,21 @@ func TestApplyBlock(t *testing.T) {
 	chainID := "test"
 
 	mpool := mempool.NewCListMempool(cfg.DefaultMempoolConfig(), proxy.NewAppConnMempool(client), 0)
-	executor := NewBlockExecutor([]byte("test address"), nsID, chainID, mpool, proxy.NewAppConnConsensus(client), logger)
+	eventBus := tmtypes.NewEventBus()
+	require.NoError(eventBus.Start())
+	executor := NewBlockExecutor([]byte("test address"), nsID, chainID, mpool, proxy.NewAppConnConsensus(client), eventBus, logger)
+
+	txQuery, err := query.New("tm.event='Tx'")
+	require.NoError(err)
+	txSub, err := eventBus.Subscribe(context.Background(), "test", txQuery, 1000)
+	require.NoError(err)
+	require.NotNil(txSub)
+
+	headerQuery, err := query.New("tm.event='NewBlockHeader'")
+	require.NoError(err)
+	headerSub, err := eventBus.Subscribe(context.Background(), "test", headerQuery, 100)
+	require.NoError(err)
+	require.NotNil(headerSub)
 
 	state := State{}
 	state.InitialHeight = 1
@@ -125,4 +142,35 @@ func TestApplyBlock(t *testing.T) {
 	require.NoError(err)
 	require.NotNil(newState)
 	assert.Equal(int64(2), newState.LastBlockHeight)
+
+	// wait for at least 4 Tx events, for up to 3 second.
+	// 3 seconds is a fail-scenario only
+	timer := time.NewTimer(3 * time.Second)
+	txs := make(map[int64]int)
+	cnt := 0
+	for cnt != 4 {
+		select {
+		case evt := <-txSub.Out():
+			cnt++
+			data, ok := evt.Data().(tmtypes.EventDataTx)
+			assert.True(ok)
+			assert.NotEmpty(data.Tx)
+			txs[data.Height]++
+		case <-timer.C:
+			t.FailNow()
+		}
+	}
+	assert.Zero(len(txSub.Out())) // expected exactly 4 Txs - channel should be empty
+	assert.EqualValues(1, txs[1])
+	assert.EqualValues(3, txs[2])
+
+	require.EqualValues(2, len(headerSub.Out()))
+	for h := 1; h <= 2; h++ {
+		evt := <-headerSub.Out()
+		data, ok := evt.Data().(tmtypes.EventDataNewBlockHeader)
+		assert.True(ok)
+		if data.Header.Height == 2 {
+			assert.EqualValues(3, data.NumTxs)
+		}
+	}
 }