mirrors/tendermint
1
0
Fork 0
mirror of https://github.com/tendermint/tendermint.git synced 2026-07-28 02:52:48 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix broken tests

Browse Source
This commit is contained in:
Callum Waters
2022-09-21 13:39:16 +02:00
parent 344efa2e15
commit 61a4e70365
29 changed files with 292 additions and 229 deletions
Download Patch File Download Diff File Expand all files Collapse all files
abci/client/client.go
+1
View File
@@ -25,6 +25,7 @@ type Client interface {
types.Application
Error() error
// TODO: remove as this is not implemented
Flush(context.Context) error
Echo(context.Context, string) (*types.ResponseEcho, error)
}
abci/client/socket_client.go
-9
View File
@@ -62,8 +62,6 @@ func (cli *socketClient) OnStart() error {
conn net.Conn
)
fmt.Println("starting socket client")
for {
conn, err = tmnet.Connect(cli.addr)
if err != nil {
@@ -86,7 +84,6 @@ func (cli *socketClient) OnStart() error {
// OnStop implements Service by closing connection and flushing all queues.
func (cli *socketClient) OnStop() {
fmt.Println("stopping socket client")
if cli.conn != nil {
cli.conn.Close()
}
@@ -114,9 +111,7 @@ func (cli *socketClient) sendRequestsRoutine(conn io.Writer) {
// unsolicited reply.
cli.trackRequest(reqres)
fmt.Println("writing message")
if err := types.WriteMessage(reqres.Request, bw); err != nil {
fmt.Println(err)
cli.stopForError(fmt.Errorf("write to buffer: %w", err))
return
}
@@ -138,13 +133,10 @@ func (cli *socketClient) recvResponseRoutine(conn io.Reader) {
var res = &types.Response{}
fmt.Println("client waiting to read a message")
if err := types.ReadMessage(r, res); err != nil {
fmt.Println(err)
cli.stopForError(fmt.Errorf("read message: %w", err))
return
}
fmt.Println("client finished reading message")
switch r := res.Value.(type) {
case *types.Response_Exception: // app responded with error
@@ -384,7 +376,6 @@ func resMatchesReq(req *types.Request, res *types.Response) (ok bool) {
}
func (cli *socketClient) stopForError(err error) {
fmt.Printf("stopping for error %v\n", err)
if !cli.IsRunning() {
return
}
abci/client/socket_client_test.go
+6 -7
View File
@@ -32,8 +32,7 @@ func TestCalls(t *testing.T) {
}()
select {
case <-time.After(3 * time.Second):
fmt.Println("no response arrived")
case <-time.After(1 * time.Second):
require.Fail(t, "No response arrived")
case err, ok := <-resp:
require.True(t, ok, "Must not close channel")
@@ -53,17 +52,17 @@ func setupClientServer(t *testing.T, app types.Application) (
err := s.Start()
require.NoError(t, err)
c := abcicli.NewSocketClient(addr, true)
err = c.Start()
require.NoError(t, err)
t.Cleanup(func() {
if err := s.Stop(); err != nil {
t.Error(err)
}
})
c := abcicli.NewSocketClient(addr, true)
err = c.Start()
require.NoError(t, err)
t.Cleanup(func() {
fmt.Println("stopping client")
if err := c.Stop(); err != nil {
t.Error(err)
}
abci/example/kvstore/kvstore_test.go
+15 -29
View File
@@ -196,39 +196,16 @@ func TestClientServer(t *testing.T) {
defer cancel()
// set up socket app
kvstore := NewInMemoryApplication()
client, server, err := makeSocketClientServer(kvstore, "kvstore-socket")
client, _, err := makeSocketClientServer(t, kvstore, "kvstore-socket")
require.NoError(t, err)
t.Cleanup(func() {
if err := server.Stop(); err != nil {
t.Error(err)
}
})
t.Cleanup(func() {
if err := client.Stop(); err != nil {
t.Error(err)
}
})
runClientTests(ctx, t, client)
// set up grpc app
kvstore = NewInMemoryApplication()
gclient, gserver, err := makeGRPCClientServer(kvstore, "/tmp/kvstore-grpc")
gclient, _, err := makeGRPCClientServer(kvstore, "/tmp/kvstore-grpc")
require.NoError(t, err)
t.Cleanup(func() {
cancel()
if err := gserver.Stop(); err != nil {
t.Error(err)
}
})
t.Cleanup(func() {
cancel()
if err := gclient.Stop(); err != nil {
t.Error(err)
}
})
runClientTests(ctx, t, gclient)
}
@@ -273,7 +250,7 @@ func valsEqual(t *testing.T, vals1, vals2 []types.ValidatorUpdate) {
}
}
func makeSocketClientServer(app types.Application, name string) (abcicli.Client, service.Service, error) {
func makeSocketClientServer(t *testing.T, app types.Application, name string) (abcicli.Client, service.Service, error) {
// Start the listener
socket := fmt.Sprintf("unix://%s.sock", name)
logger := log.TestingLogger()
@@ -284,16 +261,25 @@ func makeSocketClientServer(app types.Application, name string) (abcicli.Client,
return nil, nil, err
}
t.Cleanup(func() {
if err := server.Stop(); err != nil {
t.Error(err)
}
})
// Connect to the socket
client := abcicli.NewSocketClient(socket, false)
client.SetLogger(logger.With("module", "abci-client"))
if err := client.Start(); err != nil {
if err = server.Stop(); err != nil {
return nil, nil, err
}
return nil, nil, err
}
t.Cleanup(func() {
if err := client.Stop(); err != nil {
t.Error(err)
}
})
return client, server, nil
}
abci/server/socket_server.go
+6 -12
View File
@@ -193,7 +193,6 @@ func (s *SocketServer) handleRequests(closeConn chan error, conn io.Reader, resp
}
return
}
fmt.Println("server has read a message")
s.appMtx.Lock()
count++
resp, err := s.handleRequest(context.TODO(), req)
@@ -213,7 +212,6 @@ func (s *SocketServer) handleRequests(closeConn chan error, conn io.Reader, resp
func (s *SocketServer) handleRequest(ctx context.Context, req *types.Request) (*types.Response, error) {
switch r := req.Value.(type) {
case *types.Request_Echo:
fmt.Println("server handling request")
return types.ToResponseEcho(r.Echo.Message), nil
case *types.Request_Flush:
return types.ToResponseFlush(), nil
@@ -300,21 +298,18 @@ func (s *SocketServer) handleResponses(closeConn chan error, conn io.Writer, res
var bufWriter = bufio.NewWriter(conn)
for {
var res = <-responses
fmt.Println("server writing response")
fmt.Println(res)
err := types.WriteMessage(res, bufWriter)
if err != nil {
fmt.Println(err)
closeConn <- fmt.Errorf("error writing message: %w", err)
return
}
if _, ok := res.Value.(*types.Response_Flush); ok {
err = bufWriter.Flush()
if err != nil {
closeConn <- fmt.Errorf("error flushing write buffer: %w", err)
return
}
err = bufWriter.Flush()
if err != nil {
closeConn <- fmt.Errorf("error flushing write buffer: %w", err)
return
}
// If the application has responded with an exception, the server returns the error
// back to the client and closes the connection. The receiving Tendermint client should
// log the error and gracefully terminate
@@ -322,6 +317,5 @@ func (s *SocketServer) handleResponses(closeConn chan error, conn io.Writer, res
closeConn <- errors.New(e.Exception.Error)
}
count++
fmt.Println("server finished writing response")
}
}
abci/tests/client_server_test.go
+12 -3
View File
@@ -11,6 +11,8 @@ import (
)
func TestClientServerNoAddrPrefix(t *testing.T) {
t.Helper()
addr := "localhost:26658"
transport := "socket"
app := kvstore.NewInMemoryApplication()
@@ -19,12 +21,19 @@ func TestClientServerNoAddrPrefix(t *testing.T) {
assert.NoError(t, err, "expected no error on NewServer")
err = server.Start()
assert.NoError(t, err, "expected no error on server.Start")
defer func() { _ = server.Stop() }()
t.Cleanup(func() {
if err := server.Stop(); err != nil {
t.Error(err)
}
})
client, err := abciclient.NewClient(addr, transport, true)
assert.NoError(t, err, "expected no error on NewClient")
err = client.Start()
assert.NoError(t, err, "expected no error on client.Start")
_ = client.Stop()
t.Cleanup(func() {
if err := client.Stop(); err != nil {
t.Error(err)
}
})
}
cmd/tendermint/commands/reindex_event.go
+19 -19
View File
@@ -137,38 +137,38 @@ func eventReIndex(cmd *cobra.Command, args eventReIndexArgs) error {
fmt.Println("start re-indexing events:")
defer bar.Finish()
for i := args.startHeight; i <= args.endHeight; i++ {
for height := args.startHeight; height <= args.endHeight; height++ {
select {
case <-cmd.Context().Done():
return fmt.Errorf("event re-index terminated at height %d: %w", i, cmd.Context().Err())
return fmt.Errorf("event re-index terminated at height %d: %w", height, cmd.Context().Err())
default:
b := args.blockStore.LoadBlock(i)
if b == nil {
return fmt.Errorf("not able to load block at height %d from the blockstore", i)
block := args.blockStore.LoadBlock(height)
if block == nil {
return fmt.Errorf("not able to load block at height %d from the blockstore", height)
}
r, err := args.stateStore.LoadFinalizeBlockResponse(i)
resp, err := args.stateStore.LoadFinalizeBlockResponse(height)
if err != nil {
return fmt.Errorf("not able to load ABCI Response at height %d from the statestore", i)
return fmt.Errorf("not able to load ABCI Response at height %d from the statestore", height)
}
e := types.EventDataNewBlock{
Block: b,
ResultFinalizeBlock: *r,
e := types.EventDataNewBlockEvents{
Height: height,
Events: resp.Events,
}
numTxs := len(b.Data.Txs)
numTxs := len(resp.TxResults)
var batch *txindex.Batch
if numTxs > 0 {
batch = txindex.NewBatch(int64(numTxs))
for i := range b.Data.Txs {
for idx, txResult := range resp.TxResults {
tr := abcitypes.TxResult{
Height: b.Height,
Index: uint32(i),
Tx: b.Data.Txs[i],
Result: *(r.TxResults[i]),
Height: height,
Index: uint32(idx),
Tx: block.Txs[idx],
Result: *txResult,
}
if err = batch.Add(&tr); err != nil {
@@ -177,16 +177,16 @@ func eventReIndex(cmd *cobra.Command, args eventReIndexArgs) error {
}
if err := args.txIndexer.AddBatch(batch); err != nil {
return fmt.Errorf("tx event re-index at height %d failed: %w", i, err)
return fmt.Errorf("tx event re-index at height %d failed: %w", height, err)
}
}
if err := args.blockIndexer.Index(e); err != nil {
return fmt.Errorf("block event re-index at height %d failed: %w", i, err)
return fmt.Errorf("block event re-index at height %d failed: %w", height, err)
}
}
bar.Play(i)
bar.Play(height)
}
return nil
cmd/tendermint/commands/reindex_event_test.go
+2 -2
View File
@@ -146,8 +146,8 @@ func TestReIndexEvent(t *testing.T) {
}
mockBlockIndexer.
On("Index", mock.AnythingOfType("types.EventDataNewBlock")).Return(errors.New("")).Once().
On("Index", mock.AnythingOfType("types.EventDataNewBlock")).Return(nil)
On("Index", mock.AnythingOfType("types.EventDataNewBlockEvents")).Return(errors.New("")).Once().
On("Index", mock.AnythingOfType("types.EventDataNewBlockEvents")).Return(nil)
mockTxIndexer.
On("AddBatch", mock.AnythingOfType("*txindex.Batch")).Return(errors.New("")).Once().
consensus/mempool_test.go
+4 -4
View File
@@ -118,7 +118,7 @@ func TestMempoolTxConcurrentWithCommit(t *testing.T) {
cs := newStateWithConfigAndBlockStore(config, state, privVals[0], kvstore.NewInMemoryApplication(), blockDB)
err := stateStore.Save(state)
require.NoError(t, err)
newBlockHeaderCh := subscribe(cs.eventBus, types.EventQueryNewBlockHeader)
newBlockEventsCh := subscribe(cs.eventBus, types.EventQueryNewBlockEvents)
const numTxs int64 = 3000
go deliverTxsRange(cs, 0, int(numTxs))
@@ -126,9 +126,9 @@ func TestMempoolTxConcurrentWithCommit(t *testing.T) {
startTestRound(cs, cs.Height, cs.Round)
for n := int64(0); n < numTxs; {
select {
case msg := <-newBlockHeaderCh:
headerEvent := msg.Data().(types.EventDataNewBlockHeader)
n += headerEvent.NumTxs
case msg := <-newBlockEventsCh:
event := msg.Data().(types.EventDataNewBlockEvents)
n += event.NumTxs
case <-time.After(30 * time.Second):
t.Fatal("Timed out waiting 30s to commit blocks with transactions")
}
proxy/app_conn_test.go
+2 -2
View File
@@ -40,7 +40,7 @@ func TestEcho(t *testing.T) {
t.Fatalf("Error starting ABCI client: %v", err.Error())
}
proxy := NewAppConnMempool(cli, nil)
proxy := NewAppConnMempool(cli, NopMetrics())
t.Log("Connected")
for i := 0; i < 1000; i++ {
@@ -81,7 +81,7 @@ func BenchmarkEcho(b *testing.B) {
b.Fatalf("Error starting ABCI client: %v", err.Error())
}
proxy := NewAppConnMempool(cli, nil)
proxy := NewAppConnMempool(cli, NopMetrics())
b.Log("Connected")
b.StartTimer() // Start benchmarking tests
state/execution.go
-1
View File
@@ -519,7 +519,6 @@ func fireEvents(
if err := eventBus.PublishEventNewBlockHeader(types.EventDataNewBlockHeader{
Header: block.Header,
NumTxs: int64(len(block.Txs)),
}); err != nil {
logger.Error("failed publishing new block header", "err", err)
}
state/execution_test.go
+11 -11
View File
@@ -256,7 +256,7 @@ func TestProcessProposal(t *testing.T) {
logger := log.NewNopLogger()
app := &abcimocks.Application{}
app.On("ProcessProposal", mock.Anything).Return(abci.ResponseProcessProposal{Status: abci.ResponseProcessProposal_ACCEPT})
app.On("ProcessProposal", mock.Anything, mock.Anything).Return(&abci.ResponseProcessProposal{Status: abci.ResponseProcessProposal_ACCEPT}, nil)
cc := proxy.NewLocalClientCreator(app)
proxyApp := proxy.NewAppConns(cc, proxy.NopMetrics())
@@ -311,7 +311,7 @@ func TestProcessProposal(t *testing.T) {
})
block1.Txs = txs
expectedRpp := abci.RequestProcessProposal{
expectedRpp := &abci.RequestProcessProposal{
Txs: block1.Txs.ToSliceOfBytes(),
Hash: block1.Hash(),
Height: block1.Header.Height,
@@ -329,7 +329,7 @@ func TestProcessProposal(t *testing.T) {
require.NoError(t, err)
require.True(t, acceptBlock)
app.AssertExpectations(t)
app.AssertCalled(t, "ProcessProposal", expectedRpp)
app.AssertCalled(t, "ProcessProposal", context.TODO(), expectedRpp)
}
func TestValidateValidatorUpdates(t *testing.T) {
@@ -590,7 +590,7 @@ func TestEndBlockValidatorUpdatesResultingInEmptySet(t *testing.T) {
func TestEmptyPrepareProposal(t *testing.T) {
const height = 2
app := &abcimocks.Application{}
app := &abci.BaseApplication{}
cc := proxy.NewLocalClientCreator(app)
proxyApp := proxy.NewAppConns(cc, proxy.NopMetrics())
err := proxyApp.Start()
@@ -646,9 +646,9 @@ func TestPrepareProposalTxsAllIncluded(t *testing.T) {
mp.On("ReapMaxBytesMaxGas", mock.Anything, mock.Anything).Return(txs[2:])
app := &abcimocks.Application{}
app.On("PrepareProposal", mock.Anything).Return(abci.ResponsePrepareProposal{
app.On("PrepareProposal", mock.Anything, mock.Anything).Return(&abci.ResponsePrepareProposal{
Txs: txs.ToSliceOfBytes(),
})
}, nil)
cc := proxy.NewLocalClientCreator(app)
proxyApp := proxy.NewAppConns(cc, proxy.NopMetrics())
err := proxyApp.Start()
@@ -696,9 +696,9 @@ func TestPrepareProposalReorderTxs(t *testing.T) {
txs = append(txs[len(txs)/2:], txs[:len(txs)/2]...)
app := &abcimocks.Application{}
app.On("PrepareProposal", mock.Anything).Return(abci.ResponsePrepareProposal{
app.On("PrepareProposal", mock.Anything, mock.Anything).Return(&abci.ResponsePrepareProposal{
Txs: txs.ToSliceOfBytes(),
})
}, nil)
cc := proxy.NewLocalClientCreator(app)
proxyApp := proxy.NewAppConns(cc, proxy.NopMetrics())
@@ -749,9 +749,9 @@ func TestPrepareProposalErrorOnTooManyTxs(t *testing.T) {
mp.On("ReapMaxBytesMaxGas", mock.Anything, mock.Anything).Return(txs)
app := &abcimocks.Application{}
app.On("PrepareProposal", mock.Anything).Return(abci.ResponsePrepareProposal{
app.On("PrepareProposal", mock.Anything, mock.Anything).Return(&abci.ResponsePrepareProposal{
Txs: txs.ToSliceOfBytes(),
})
}, nil)
cc := proxy.NewLocalClientCreator(app)
proxyApp := proxy.NewAppConns(cc, proxy.NopMetrics())
@@ -798,7 +798,7 @@ func TestPrepareProposalErrorOnPrepareProposalError(t *testing.T) {
cm.On("SetLogger", mock.Anything).Return()
cm.On("Start").Return(nil)
cm.On("Quit").Return(nil)
cm.On("PrepareProposalSync", mock.Anything).Return(nil, errors.New("an injected error")).Once()
cm.On("PrepareProposal", mock.Anything, mock.Anything).Return(nil, errors.New("an injected error")).Once()
cm.On("Stop").Return(nil)
cc := &pmocks.ClientCreator{}
cc.On("NewABCIClient").Return(cm, nil)
state/indexer/block.go
+1 -1
View File
@@ -16,7 +16,7 @@ type BlockIndexer interface {
Has(height int64) (bool, error)
// Index indexes BeginBlock and EndBlock events for a given block by its height.
Index(types.EventDataNewBlock) error
Index(types.EventDataNewBlockEvents) error
// Search performs a query for block heights that match a given BeginBlock
// and Endblock event search criteria.
state/indexer/block/kv/kv.go
+3 -3
View File
@@ -49,11 +49,11 @@ func (idx *BlockerIndexer) Has(height int64) (bool, error) {
// primary key: encode(block.height | height) => encode(height)
// BeginBlock events: encode(eventType.eventAttr|eventValue|height|begin_block) => encode(height)
// EndBlock events: encode(eventType.eventAttr|eventValue|height|end_block) => encode(height)
func (idx *BlockerIndexer) Index(bh types.EventDataNewBlock) error {
func (idx *BlockerIndexer) Index(bh types.EventDataNewBlockEvents) error {
batch := idx.store.NewBatch()
defer batch.Close()
height := bh.Block.Height
height := bh.Height
// 1. index by height
key, err := heightKey(height)
@@ -65,7 +65,7 @@ func (idx *BlockerIndexer) Index(bh types.EventDataNewBlock) error {
}
// 2. index block events
if err := idx.indexEvents(batch, bh.ResultFinalizeBlock.Events, "begin_block", height); err != nil {
if err := idx.indexEvents(batch, bh.Events, "begin_block", height); err != nil {
return fmt.Errorf("failed to index BeginBlock events: %w", err)
}
state/indexer/block/kv/kv_test.go
+36 -40
View File
@@ -18,9 +18,40 @@ func TestBlockIndexer(t *testing.T) {
store := db.NewPrefixDB(db.NewMemDB(), []byte("block_events"))
indexer := blockidxkv.New(store)
require.NoError(t, indexer.Index(types.EventDataNewBlock{
Block: &types.Block{Header: types.Header{Height: 1}},
ResultFinalizeBlock: abci.ResponseFinalizeBlock{
require.NoError(t, indexer.Index(types.EventDataNewBlockEvents{
Height: 1,
Events: []abci.Event{
{
Type: "begin_event",
Attributes: []abci.EventAttribute{
{
Key: "proposer",
Value: "FCAA001",
Index: true,
},
},
},
{
Type: "end_event",
Attributes: []abci.EventAttribute{
{
Key: "foo",
Value: "100",
Index: true,
},
},
},
},
}))
for i := 2; i < 12; i++ {
var index bool
if i%2 == 0 {
index = true
}
require.NoError(t, indexer.Index(types.EventDataNewBlockEvents{
Height: int64(i),
Events: []abci.Event{
{
Type: "begin_event",
@@ -37,43 +68,8 @@ func TestBlockIndexer(t *testing.T) {
Attributes: []abci.EventAttribute{
{
Key: "foo",
Value: "100",
Index: true,
},
},
},
},
},
}))
for i := 2; i < 12; i++ {
var index bool
if i%2 == 0 {
index = true
}
require.NoError(t, indexer.Index(types.EventDataNewBlock{
Block: &types.Block{Header: types.Header{Height: int64(i)}},
ResultFinalizeBlock: abci.ResponseFinalizeBlock{
Events: []abci.Event{
{
Type: "begin_event",
Attributes: []abci.EventAttribute{
{
Key: "proposer",
Value: "FCAA001",
Index: true,
},
},
},
{
Type: "end_event",
Attributes: []abci.EventAttribute{
{
Key: "foo",
Value: fmt.Sprintf("%d", i),
Index: index,
},
Value: fmt.Sprintf("%d", i),
Index: index,
},
},
},
state/indexer/block/null/null.go
+1 -1
View File
@@ -18,7 +18,7 @@ func (idx *BlockerIndexer) Has(height int64) (bool, error) {
return false, errors.New(`indexing is disabled (set 'tx_index = "kv"' in config)`)
}
func (idx *BlockerIndexer) Index(types.EventDataNewBlock) error {
func (idx *BlockerIndexer) Index(types.EventDataNewBlockEvents) error {
return nil
}
state/indexer/mocks/block_indexer.go
+2 -2
View File
@@ -39,11 +39,11 @@ func (_m *BlockIndexer) Has(height int64) (bool, error) {
}
// Index provides a mock function with given fields: _a0
func (_m *BlockIndexer) Index(_a0 types.EventDataNewBlock) error {
func (_m *BlockIndexer) Index(_a0 types.EventDataNewBlockEvents) error {
ret := _m.Called(_a0)
var r0 error
if rf, ok := ret.Get(0).(func(types.EventDataNewBlock) error); ok {
if rf, ok := ret.Get(0).(func(types.EventDataNewBlockEvents) error); ok {
r0 = rf(_a0)
} else {
r0 = ret.Error(0)
state/indexer/sink/psql/backport.go
+1 -1
View File
@@ -77,7 +77,7 @@ func (BackportBlockIndexer) Has(height int64) (bool, error) {
// Index indexes block begin and end events for the specified block. It is
// part of the BlockIndexer interface.
func (b BackportBlockIndexer) Index(block types.EventDataNewBlock) error {
func (b BackportBlockIndexer) Index(block types.EventDataNewBlockEvents) error {
return b.psql.IndexBlockEvents(block)
}
state/indexer/sink/psql/psql.go
+4 -4
View File
@@ -139,7 +139,7 @@ func makeIndexedEvent(compositeKey, value string) abci.Event {
// IndexBlockEvents indexes the specified block header, part of the
// indexer.EventSink interface.
func (es *EventSink) IndexBlockEvents(h types.EventDataNewBlock) error {
func (es *EventSink) IndexBlockEvents(h types.EventDataNewBlockEvents) error {
ts := time.Now().UTC()
return runInTransaction(es.store, func(dbtx *sql.Tx) error {
@@ -150,7 +150,7 @@ INSERT INTO `+tableBlocks+` (height, chain_id, created_at)
VALUES ($1, $2, $3)
ON CONFLICT DO NOTHING
RETURNING rowid;
`, h.Block.Height, es.chainID, ts)
`, h.Height, es.chainID, ts)
if err == sql.ErrNoRows {
return nil // we already saw this block; quietly succeed
} else if err != nil {
@@ -159,12 +159,12 @@ INSERT INTO `+tableBlocks+` (height, chain_id, created_at)
// Insert the special block meta-event for height.
if err := insertEvents(dbtx, blockID, 0, []abci.Event{
makeIndexedEvent(types.BlockHeightKey, fmt.Sprint(h.Block.Height)),
makeIndexedEvent(types.BlockHeightKey, fmt.Sprint(h.Height)),
}); err != nil {
return fmt.Errorf("block meta-events: %w", err)
}
// Insert all the block events. Order is important here,
if err := insertEvents(dbtx, blockID, 0, h.ResultFinalizeBlock.Events); err != nil {
if err := insertEvents(dbtx, blockID, 0, h.Events); err != nil {
return fmt.Errorf("begin-block events: %w", err)
}
return nil
state/indexer/sink/psql/psql_test.go
+10 -12
View File
@@ -139,7 +139,7 @@ func TestMain(m *testing.M) {
func TestIndexing(t *testing.T) {
t.Run("IndexBlockEvents", func(t *testing.T) {
indexer := &EventSink{store: testDB(), chainID: chainID}
require.NoError(t, indexer.IndexBlockEvents(newTestBlock()))
require.NoError(t, indexer.IndexBlockEvents(newTestBlockEvents()))
verifyBlock(t, 1)
verifyBlock(t, 2)
@@ -155,7 +155,7 @@ func TestIndexing(t *testing.T) {
require.NoError(t, verifyTimeStamp(tableBlocks))
// Attempting to reindex the same events should gracefully succeed.
require.NoError(t, indexer.IndexBlockEvents(newTestBlock()))
require.NoError(t, indexer.IndexBlockEvents(newTestBlockEvents()))
})
t.Run("IndexTxEvents", func(t *testing.T) {
@@ -205,16 +205,14 @@ func TestStop(t *testing.T) {
// newTestBlock constructs a fresh copy of a new block event containing
// known test values to exercise the indexer.
func newTestBlock() types.EventDataNewBlock {
return types.EventDataNewBlock{
Block: &types.Block{Header: types.Header{Height: 1}},
ResultFinalizeBlock: abci.ResponseFinalizeBlock{
Events: []abci.Event{
makeIndexedEvent("begin_event.proposer", "FCAA001"),
makeIndexedEvent("thingy.whatzit", "O.O"),
makeIndexedEvent("end_event.foo", "100"),
makeIndexedEvent("thingy.whatzit", "-.O"),
},
func newTestBlockEvents() types.EventDataNewBlockEvents {
return types.EventDataNewBlockEvents{
Height: 1,
Events: []abci.Event{
makeIndexedEvent("begin_event.proposer", "FCAA001"),
makeIndexedEvent("thingy.whatzit", "O.O"),
makeIndexedEvent("end_event.foo", "100"),
makeIndexedEvent("thingy.whatzit", "-.O"),
},
}
}
state/state_test.go
+10 -10
View File
@@ -93,8 +93,8 @@ func TestStateSaveLoad(t *testing.T) {
loadedState, state))
}
// TestABCIResponsesSaveLoad tests saving and loading ABCIResponses.
func TestABCIResponsesSaveLoad1(t *testing.T) {
// TestFinalizeBlockResponsesSaveLoad1 tests saving and loading ABCIResponses.
func TestFinalizeBlockResponsesSaveLoad1(t *testing.T) {
tearDown, stateDB, state := setupTestCase(t)
defer tearDown(t)
stateStore := sm.NewStore(stateDB, sm.StoreOptions{
@@ -120,14 +120,12 @@ func TestABCIResponsesSaveLoad1(t *testing.T) {
err := stateStore.SaveFinalizeBlockResponse(block.Height, abciResponses)
require.NoError(t, err)
loadedABCIResponses, err := stateStore.LoadFinalizeBlockResponse(block.Height)
assert.Nil(err)
assert.Equal(abciResponses, loadedABCIResponses,
fmt.Sprintf("ABCIResponses don't match:\ngot: %v\nexpected: %v\n",
loadedABCIResponses, abciResponses))
assert.NoError(err)
assert.Equal(abciResponses, loadedABCIResponses)
}
// TestResultsSaveLoad tests saving and loading ABCI results.
func TestABCIResponsesSaveLoad2(t *testing.T) {
// TestResultsSaveLoad tests saving and loading FinalizeBlock results.
func TestFinalizeBlockResponsesSaveLoad2(t *testing.T) {
tearDown, stateDB, _ := setupTestCase(t)
defer tearDown(t)
assert := assert.New(t)
@@ -192,7 +190,8 @@ func TestABCIResponsesSaveLoad2(t *testing.T) {
for i, tc := range cases {
h := int64(i + 1) // last block height, one below what we save
responses := &abci.ResponseFinalizeBlock{
TxResults: tc.added,
TxResults: tc.added,
AgreedAppData: []byte(fmt.Sprintf("%d", h)),
}
err := stateStore.SaveFinalizeBlockResponse(h, responses)
require.NoError(t, err)
@@ -205,7 +204,8 @@ func TestABCIResponsesSaveLoad2(t *testing.T) {
if assert.NoError(err, "%d", i) {
t.Log(res)
responses := &abci.ResponseFinalizeBlock{
TxResults: tc.expected,
TxResults: tc.expected,
AgreedAppData: []byte(fmt.Sprintf("%d", h)),
}
assert.Equal(sm.TxResultsHash(responses.TxResults), sm.TxResultsHash(res.TxResults), "%d", i)
}
state/store.go
+21 -4
View File
@@ -385,17 +385,33 @@ func (store dbStore) LoadFinalizeBlockResponse(height int64) (*abci.ResponseFina
return nil, err
}
if len(buf) == 0 {
return nil, ErrNoABCIResponsesForHeight{height}
}
resp := new(abci.ResponseFinalizeBlock)
err = resp.Unmarshal(buf)
if err != nil {
// DATA HAS BEEN CORRUPTED OR THE SPEC HAS CHANGED
tmos.Exit(fmt.Sprintf(`LoadFinalizeBlockResponse: Data has been corrupted or its spec has
changed: %v\n`, err))
// The data might be of the legacy ABCI response type, so
// we try to unmarshal that
legacyResp := new(tmstate.LegacyABCIResponses)
rerr := legacyResp.Unmarshal(buf)
if rerr != nil {
// DATA HAS BEEN CORRUPTED OR THE SPEC HAS CHANGED
tmos.Exit(fmt.Sprintf(`LoadFinalizeBlockResponse: Data has been corrupted or its spec has
changed: %v\n`, err))
}
// The state store contains the old format. Migrate to
// the new ResponseFinalizeBlock format. Note that the
// new struct expects the AgreedAppData which we don't have.
resp = &abci.ResponseFinalizeBlock{
TxResults: legacyResp.DeliverTxs,
ValidatorUpdates: legacyResp.EndBlock.ValidatorUpdates,
ConsensusParamUpdates: legacyResp.EndBlock.ConsensusParamUpdates,
Events: append(legacyResp.BeginBlock.Events, legacyResp.EndBlock.Events...),
// NOTE: AgreedAppData is missing in the response
}
}
// TODO: ensure that buf is completely read.
return resp, nil
@@ -451,6 +467,7 @@ func (store dbStore) SaveFinalizeBlockResponse(height int64, resp *abci.Response
// If the flag is false then we save the ABCIResponse. This can be used for the /BlockResults
// query or to reindex an event using the command line.
if !store.DiscardFinalizeBlockResponses {
fmt.Printf("saving response at height %d\n", height)
bz, err := resp.Marshal()
if err != nil {
return err
state/txindex/indexer_service.go
+7 -7
View File
@@ -43,10 +43,10 @@ func (is *IndexerService) OnStart() error {
// Use SubscribeUnbuffered here to ensure both subscriptions does not get
// canceled due to not pulling messages fast enough. Cause this might
// sometimes happen when there are no other subscribers.
blockHeadersSub, err := is.eventBus.SubscribeUnbuffered(
blockSub, err := is.eventBus.SubscribeUnbuffered(
context.Background(),
subscriber,
types.EventQueryNewBlockHeader)
types.EventQueryNewBlockEvents)
if err != nil {
return err
}
@@ -58,10 +58,10 @@ func (is *IndexerService) OnStart() error {
go func() {
for {
msg := <-blockHeadersSub.Out()
eventNewBlock := msg.Data().(types.EventDataNewBlock)
height := eventNewBlock.Block.Height
numTxs := int64(len(eventNewBlock.Block.Data.Txs))
msg := <-blockSub.Out()
eventNewBlockEvents := msg.Data().(types.EventDataNewBlockEvents)
height := eventNewBlockEvents.Height
numTxs := eventNewBlockEvents.NumTxs
batch := NewBatch(numTxs)
@@ -79,7 +79,7 @@ func (is *IndexerService) OnStart() error {
}
}
if err := is.blockIdxr.Index(eventNewBlock); err != nil {
if err := is.blockIdxr.Index(eventNewBlockEvents); err != nil {
is.Logger.Error("failed to index block", "height", height, "err", err)
} else {
is.Logger.Info("indexed block", "height", height)
state/txindex/indexer_service_test.go
+15 -3
View File
@@ -42,9 +42,21 @@ func TestIndexerServiceIndexesBlocks(t *testing.T) {
}
})
// publish block with txs
err = eventBus.PublishEventNewBlockHeader(types.EventDataNewBlockHeader{
Header: types.Header{Height: 1},
// publish block with events
err = eventBus.PublishEventNewBlockEvents(types.EventDataNewBlockEvents{
Height: 1,
Events: []abci.Event{
{
Type: "begin_event",
Attributes: []abci.EventAttribute{
{
Key: "proposer",
Value: "FCAA001",
Index: true,
},
},
},
},
NumTxs: int64(2),
})
require.NoError(t, err)
statesync/reactor_test.go
+1 -1
View File
@@ -122,7 +122,7 @@ func TestReactor_Receive_SnapshotsRequest(t *testing.T) {
t.Run(name, func(t *testing.T) {
// Mock ABCI connection to return local snapshots
conn := &proxymocks.AppConnSnapshot{}
conn.On("ListSnapshotsSync", abci.RequestListSnapshots{}).Return(&abci.ResponseListSnapshots{
conn.On("ListSnapshots", mock.Anything, &abci.RequestListSnapshots{}).Return(&abci.ResponseListSnapshots{
Snapshots: tc.snapshots,
}, nil)
statesync/syncer_test.go
+27 -27
View File
@@ -124,7 +124,7 @@ func TestSyncer_SyncAny(t *testing.T) {
// We start a sync, with peers sending back chunks when requested. We first reject the snapshot
// with height 2 format 2, and accept the snapshot at height 1.
connSnapshot.On("OfferSnapshotSync", abci.RequestOfferSnapshot{
connSnapshot.On("OfferSnapshot", mock.Anything, &abci.RequestOfferSnapshot{
Snapshot: &abci.Snapshot{
Height: 2,
Format: 2,
@@ -133,7 +133,7 @@ func TestSyncer_SyncAny(t *testing.T) {
},
AppHash: []byte("app_hash_2"),
}).Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_REJECT_FORMAT}, nil)
connSnapshot.On("OfferSnapshotSync", abci.RequestOfferSnapshot{
connSnapshot.On("OfferSnapshot", mock.Anything, &abci.RequestOfferSnapshot{
Snapshot: &abci.Snapshot{
Height: s.Height,
Format: s.Format,
@@ -168,7 +168,7 @@ func TestSyncer_SyncAny(t *testing.T) {
// The first time we're applying chunk 2 we tell it to retry the snapshot and discard chunk 1,
// which should cause it to keep the existing chunk 0 and 2, and restart restoration from
// beginning. We also wait for a little while, to exercise the retry logic in fetchChunks().
connSnapshot.On("ApplySnapshotChunkSync", abci.RequestApplySnapshotChunk{
connSnapshot.On("ApplySnapshotChunk", mock.Anything, &abci.RequestApplySnapshotChunk{
Index: 2, Chunk: []byte{1, 1, 2},
}).Once().Run(func(args mock.Arguments) { time.Sleep(2 * time.Second) }).Return(
&abci.ResponseApplySnapshotChunk{
@@ -176,16 +176,16 @@ func TestSyncer_SyncAny(t *testing.T) {
RefetchChunks: []uint32{1},
}, nil)
connSnapshot.On("ApplySnapshotChunkSync", abci.RequestApplySnapshotChunk{
connSnapshot.On("ApplySnapshotChunk", mock.Anything, &abci.RequestApplySnapshotChunk{
Index: 0, Chunk: []byte{1, 1, 0},
}).Times(2).Return(&abci.ResponseApplySnapshotChunk{Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
connSnapshot.On("ApplySnapshotChunkSync", abci.RequestApplySnapshotChunk{
connSnapshot.On("ApplySnapshotChunk", mock.Anything, &abci.RequestApplySnapshotChunk{
Index: 1, Chunk: []byte{1, 1, 1},
}).Times(2).Return(&abci.ResponseApplySnapshotChunk{Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
connSnapshot.On("ApplySnapshotChunkSync", abci.RequestApplySnapshotChunk{
connSnapshot.On("ApplySnapshotChunk", mock.Anything, &abci.RequestApplySnapshotChunk{
Index: 2, Chunk: []byte{1, 1, 2},
}).Once().Return(&abci.ResponseApplySnapshotChunk{Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
connQuery.On("InfoSync", proxy.RequestInfo).Return(&abci.ResponseInfo{
connQuery.On("Info", mock.Anything, proxy.RequestInfo).Return(&abci.ResponseInfo{
AppVersion: testAppVersion,
LastBlockHeight: 1,
LastBlockAppHash: []byte("app_hash"),
@@ -223,7 +223,7 @@ func TestSyncer_SyncAny_abort(t *testing.T) {
s := &snapshot{Height: 1, Format: 1, Chunks: 3, Hash: []byte{1, 2, 3}}
_, err := syncer.AddSnapshot(simplePeer("id"), s)
require.NoError(t, err)
connSnapshot.On("OfferSnapshotSync", abci.RequestOfferSnapshot{
connSnapshot.On("OfferSnapshot", mock.Anything, &abci.RequestOfferSnapshot{
Snapshot: toABCI(s), AppHash: []byte("app_hash"),
}).Once().Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_ABORT}, nil)
@@ -246,15 +246,15 @@ func TestSyncer_SyncAny_reject(t *testing.T) {
_, err = syncer.AddSnapshot(simplePeer("id"), s11)
require.NoError(t, err)
connSnapshot.On("OfferSnapshotSync", abci.RequestOfferSnapshot{
connSnapshot.On("OfferSnapshot", mock.Anything, &abci.RequestOfferSnapshot{
Snapshot: toABCI(s22), AppHash: []byte("app_hash"),
}).Once().Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_REJECT}, nil)
connSnapshot.On("OfferSnapshotSync", abci.RequestOfferSnapshot{
connSnapshot.On("OfferSnapshot", mock.Anything, &abci.RequestOfferSnapshot{
Snapshot: toABCI(s12), AppHash: []byte("app_hash"),
}).Once().Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_REJECT}, nil)
connSnapshot.On("OfferSnapshotSync", abci.RequestOfferSnapshot{
connSnapshot.On("OfferSnapshot", mock.Anything, &abci.RequestOfferSnapshot{
Snapshot: toABCI(s11), AppHash: []byte("app_hash"),
}).Once().Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_REJECT}, nil)
@@ -277,11 +277,11 @@ func TestSyncer_SyncAny_reject_format(t *testing.T) {
_, err = syncer.AddSnapshot(simplePeer("id"), s11)
require.NoError(t, err)
connSnapshot.On("OfferSnapshotSync", abci.RequestOfferSnapshot{
connSnapshot.On("OfferSnapshot", mock.Anything, &abci.RequestOfferSnapshot{
Snapshot: toABCI(s22), AppHash: []byte("app_hash"),
}).Once().Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_REJECT_FORMAT}, nil)
connSnapshot.On("OfferSnapshotSync", abci.RequestOfferSnapshot{
connSnapshot.On("OfferSnapshot", mock.Anything, &abci.RequestOfferSnapshot{
Snapshot: toABCI(s11), AppHash: []byte("app_hash"),
}).Once().Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_ABORT}, nil)
@@ -315,11 +315,11 @@ func TestSyncer_SyncAny_reject_sender(t *testing.T) {
_, err = syncer.AddSnapshot(peerC, sbc)
require.NoError(t, err)
connSnapshot.On("OfferSnapshotSync", abci.RequestOfferSnapshot{
connSnapshot.On("OfferSnapshot", mock.Anything, &abci.RequestOfferSnapshot{
Snapshot: toABCI(sbc), AppHash: []byte("app_hash"),
}).Once().Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_REJECT_SENDER}, nil)
connSnapshot.On("OfferSnapshotSync", abci.RequestOfferSnapshot{
connSnapshot.On("OfferSnapshot", mock.Anything, &abci.RequestOfferSnapshot{
Snapshot: toABCI(sa), AppHash: []byte("app_hash"),
}).Once().Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_REJECT}, nil)
@@ -335,7 +335,7 @@ func TestSyncer_SyncAny_abciError(t *testing.T) {
s := &snapshot{Height: 1, Format: 1, Chunks: 3, Hash: []byte{1, 2, 3}}
_, err := syncer.AddSnapshot(simplePeer("id"), s)
require.NoError(t, err)
connSnapshot.On("OfferSnapshotSync", abci.RequestOfferSnapshot{
connSnapshot.On("OfferSnapshot", mock.Anything, &abci.RequestOfferSnapshot{
Snapshot: toABCI(s), AppHash: []byte("app_hash"),
}).Once().Return(nil, errBoom)
@@ -367,7 +367,7 @@ func TestSyncer_offerSnapshot(t *testing.T) {
t.Run(name, func(t *testing.T) {
syncer, connSnapshot := setupOfferSyncer(t)
s := &snapshot{Height: 1, Format: 1, Chunks: 3, Hash: []byte{1, 2, 3}, trustedAppHash: []byte("app_hash")}
connSnapshot.On("OfferSnapshotSync", abci.RequestOfferSnapshot{
connSnapshot.On("OfferSnapshot", mock.Anything, &abci.RequestOfferSnapshot{
Snapshot: toABCI(s),
AppHash: []byte("app_hash"),
}).Return(&abci.ResponseOfferSnapshot{Result: tc.result}, tc.err)
@@ -420,11 +420,11 @@ func TestSyncer_applyChunks_Results(t *testing.T) {
_, err = chunks.Add(&chunk{Height: 1, Format: 1, Index: 0, Chunk: body})
require.NoError(t, err)
connSnapshot.On("ApplySnapshotChunkSync", abci.RequestApplySnapshotChunk{
connSnapshot.On("ApplySnapshotChunk", mock.Anything, &abci.RequestApplySnapshotChunk{
Index: 0, Chunk: body,
}).Once().Return(&abci.ResponseApplySnapshotChunk{Result: tc.result}, tc.err)
if tc.result == abci.ResponseApplySnapshotChunk_RETRY {
connSnapshot.On("ApplySnapshotChunkSync", abci.RequestApplySnapshotChunk{
connSnapshot.On("ApplySnapshotChunk", mock.Anything, &abci.RequestApplySnapshotChunk{
Index: 0, Chunk: body,
}).Once().Return(&abci.ResponseApplySnapshotChunk{
Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
@@ -480,13 +480,13 @@ func TestSyncer_applyChunks_RefetchChunks(t *testing.T) {
require.NoError(t, err)
// The first two chunks are accepted, before the last one asks for 1 to be refetched
connSnapshot.On("ApplySnapshotChunkSync", abci.RequestApplySnapshotChunk{
connSnapshot.On("ApplySnapshotChunk", mock.Anything, &abci.RequestApplySnapshotChunk{
Index: 0, Chunk: []byte{0},
}).Once().Return(&abci.ResponseApplySnapshotChunk{Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
connSnapshot.On("ApplySnapshotChunkSync", abci.RequestApplySnapshotChunk{
connSnapshot.On("ApplySnapshotChunk", mock.Anything, &abci.RequestApplySnapshotChunk{
Index: 1, Chunk: []byte{1},
}).Once().Return(&abci.ResponseApplySnapshotChunk{Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
connSnapshot.On("ApplySnapshotChunkSync", abci.RequestApplySnapshotChunk{
connSnapshot.On("ApplySnapshotChunk", mock.Anything, &abci.RequestApplySnapshotChunk{
Index: 2, Chunk: []byte{2},
}).Once().Return(&abci.ResponseApplySnapshotChunk{
Result: tc.result,
@@ -566,13 +566,13 @@ func TestSyncer_applyChunks_RejectSenders(t *testing.T) {
require.NoError(t, err)
// The first two chunks are accepted, before the last one asks for b sender to be rejected
connSnapshot.On("ApplySnapshotChunkSync", abci.RequestApplySnapshotChunk{
connSnapshot.On("ApplySnapshotChunk", mock.Anything, &abci.RequestApplySnapshotChunk{
Index: 0, Chunk: []byte{0}, Sender: "a",
}).Once().Return(&abci.ResponseApplySnapshotChunk{Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
connSnapshot.On("ApplySnapshotChunkSync", abci.RequestApplySnapshotChunk{
connSnapshot.On("ApplySnapshotChunk", mock.Anything, &abci.RequestApplySnapshotChunk{
Index: 1, Chunk: []byte{1}, Sender: "b",
}).Once().Return(&abci.ResponseApplySnapshotChunk{Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
connSnapshot.On("ApplySnapshotChunkSync", abci.RequestApplySnapshotChunk{
connSnapshot.On("ApplySnapshotChunk", mock.Anything, &abci.RequestApplySnapshotChunk{
Index: 2, Chunk: []byte{2}, Sender: "c",
}).Once().Return(&abci.ResponseApplySnapshotChunk{
Result: tc.result,
@@ -581,7 +581,7 @@ func TestSyncer_applyChunks_RejectSenders(t *testing.T) {
// On retry, the last chunk will be tried again, so we just accept it then.
if tc.result == abci.ResponseApplySnapshotChunk_RETRY {
connSnapshot.On("ApplySnapshotChunkSync", abci.RequestApplySnapshotChunk{
connSnapshot.On("ApplySnapshotChunk", mock.Anything, &abci.RequestApplySnapshotChunk{
Index: 2, Chunk: []byte{2}, Sender: "c",
}).Once().Return(&abci.ResponseApplySnapshotChunk{Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
}
@@ -654,7 +654,7 @@ func TestSyncer_verifyApp(t *testing.T) {
cfg := config.DefaultStateSyncConfig()
syncer := newSyncer(*cfg, log.NewNopLogger(), connSnapshot, connQuery, stateProvider, "")
connQuery.On("InfoSync", proxy.RequestInfo).Return(tc.response, tc.err)
connQuery.On("Info", mock.Anything, proxy.RequestInfo).Return(tc.response, tc.err)
err := syncer.verifyApp(s, appVersion)
unwrapped := errors.Unwrap(err)
if unwrapped != nil {
types/event_bus.go
+13 -1
View File
@@ -135,7 +135,7 @@ func (b *EventBus) PublishEventNewBlock(data EventDataNewBlock) error {
// no explicit deadline for publishing events
ctx := context.Background()
events := b.validateAndStringifyEvents(data.ResultFinalizeBlock.Events, b.Logger.With("block", data.Block.StringShort()))
events := b.validateAndStringifyEvents(data.ResultFinalizeBlock.Events, b.Logger.With("height", data.Block.Height))
// add predefined new block event
events[EventTypeKey] = append(events[EventTypeKey], EventNewBlock)
@@ -143,6 +143,18 @@ func (b *EventBus) PublishEventNewBlock(data EventDataNewBlock) error {
return b.pubsub.PublishWithEvents(ctx, data, events)
}
func (b *EventBus) PublishEventNewBlockEvents(data EventDataNewBlockEvents) error {
// no explicit deadline for publishing events
ctx := context.Background()
events := b.validateAndStringifyEvents(data.Events, b.Logger.With("height", data.Height))
// add predefined new block event
events[EventTypeKey] = append(events[EventTypeKey], EventNewBlockEvents)
return b.pubsub.PublishWithEvents(ctx, data, events)
}
func (b *EventBus) PublishEventNewBlockHeader(data EventDataNewBlockHeader) error {
return b.Publish(EventNewBlockHeader, data)
}
types/event_bus_test.go
+50 -7
View File
@@ -83,7 +83,7 @@ func TestEventBusPublishEventNewBlock(t *testing.T) {
}
// PublishEventNewBlock adds the tm.event compositeKey, so the query below should work
query := "tm.event='NewBlock' AND testType.baz=1 AND testType.foz=2"
query := "tm.event='NewBlock' AND testType.baz=1"
blocksSub, err := eventBus.Subscribe(context.Background(), "test", tmquery.MustParse(query))
require.NoError(t, err)
@@ -235,10 +235,9 @@ func TestEventBusPublishEventNewBlockHeader(t *testing.T) {
}
})
const numTxs = 100
block := MakeBlock(0, []Tx{}, nil, []Evidence{})
// PublishEventNewBlockHeader adds the tm.event compositeKey, so the query below should work
query := "tm.event='NewBlockHeader' AND testType.baz=1 AND testType.foz=2"
query := "tm.event='NewBlockHeader'"
headersSub, err := eventBus.Subscribe(context.Background(), "test", tmquery.MustParse(query))
require.NoError(t, err)
@@ -247,13 +246,53 @@ func TestEventBusPublishEventNewBlockHeader(t *testing.T) {
msg := <-headersSub.Out()
edt := msg.Data().(EventDataNewBlockHeader)
assert.Equal(t, block.Header, edt.Header)
assert.Equal(t, numTxs, edt.NumTxs)
close(done)
}()
err = eventBus.PublishEventNewBlockHeader(EventDataNewBlockHeader{
Header: block.Header,
NumTxs: numTxs,
})
assert.NoError(t, err)
select {
case <-done:
case <-time.After(1 * time.Second):
t.Fatal("did not receive a block header after 1 sec.")
}
}
func TestEventBusPublishEventNewBlockEvents(t *testing.T) {
eventBus := NewEventBus()
err := eventBus.Start()
require.NoError(t, err)
t.Cleanup(func() {
if err := eventBus.Stop(); err != nil {
t.Error(err)
}
})
// PublishEventNewBlockHeader adds the tm.event compositeKey, so the query below should work
query := "tm.event='NewBlockEvents'"
headersSub, err := eventBus.Subscribe(context.Background(), "test", tmquery.MustParse(query))
require.NoError(t, err)
done := make(chan struct{})
go func() {
msg := <-headersSub.Out()
edt := msg.Data().(EventDataNewBlockEvents)
assert.Equal(t, int64(1), edt.Height)
close(done)
}()
err = eventBus.PublishEventNewBlockEvents(EventDataNewBlockEvents{
Height: 1,
Events: []abci.Event{{
Type: "transfer",
Attributes: []abci.EventAttribute{{
Key: "currency",
Value: "ATOM",
}},
}},
})
assert.NoError(t, err)
@@ -313,7 +352,7 @@ func TestEventBusPublish(t *testing.T) {
}
})
const numEventsExpected = 14
const numEventsExpected = 15
sub, err := eventBus.Subscribe(context.Background(), "test", tmquery.Empty{}, numEventsExpected)
require.NoError(t, err)
@@ -332,10 +371,12 @@ func TestEventBusPublish(t *testing.T) {
err = eventBus.Publish(EventNewBlockHeader, EventDataNewBlockHeader{})
require.NoError(t, err)
err = eventBus.PublishEventNewBlock(EventDataNewBlock{})
err = eventBus.PublishEventNewBlock(EventDataNewBlock{Block: &Block{Header: Header{Height: 1}}})
require.NoError(t, err)
err = eventBus.PublishEventNewBlockHeader(EventDataNewBlockHeader{})
require.NoError(t, err)
err = eventBus.PublishEventNewBlockEvents(EventDataNewBlockEvents{Height: 1})
require.NoError(t, err)
err = eventBus.PublishEventVote(EventDataVote{})
require.NoError(t, err)
err = eventBus.PublishEventNewRoundStep(EventDataRoundState{})
@@ -454,6 +495,7 @@ func benchmarkEventBus(numClients int, randQueries bool, randEvents bool, b *tes
var events = []string{
EventNewBlock,
EventNewBlockHeader,
EventNewBlockEvents,
EventNewRound,
EventNewRoundStep,
EventTimeoutPropose,
@@ -472,6 +514,7 @@ func randEvent() string {
var queries = []tmpubsub.Query{
EventQueryNewBlock,
EventQueryNewBlockHeader,
EventQueryNewBlockEvents,
EventQueryNewRound,
EventQueryNewRoundStep,
EventQueryTimeoutPropose,
types/events.go
+12 -6
View File
@@ -18,6 +18,7 @@ const (
// All of this data can be fetched through the rpc.
EventNewBlock = "NewBlock"
EventNewBlockHeader = "NewBlockHeader"
EventNewBlockEvents = "NewBlockEvents"
EventNewEvidence = "NewEvidence"
EventTx = "Tx"
EventValidatorSetUpdates = "ValidatorSetUpdates"
@@ -48,6 +49,7 @@ type TMEventData interface {
func init() {
tmjson.RegisterType(EventDataNewBlock{}, "tendermint/event/NewBlock")
tmjson.RegisterType(EventDataNewBlockHeader{}, "tendermint/event/NewBlockHeader")
tmjson.RegisterType(EventDataNewBlockEvents{}, "tendermint/event/NewBlockEvents")
tmjson.RegisterType(EventDataNewEvidence{}, "tendermint/event/NewEvidence")
tmjson.RegisterType(EventDataTx{}, "tendermint/event/Tx")
tmjson.RegisterType(EventDataRoundState{}, "tendermint/event/RoundState")
@@ -62,21 +64,24 @@ func init() {
// but some (an input to a call tx or a receive) are more exotic
type EventDataNewBlock struct {
Block *Block `json:"block"`
BlockID BlockID `json:"block_id"`
Block *Block `json:"block"`
BlockID BlockID `json:"block_id"`
ResultFinalizeBlock abci.ResponseFinalizeBlock `json:"result_finalize_block"`
}
type EventDataNewBlockHeader struct {
Header Header `json:"header"`
NumTxs int64 `json:"num_txs,string"` // Number of txs in a block
}
type EventDataNewBlockEvents struct {
Height int64 `json:"height"`
Events []abci.Event `json:"events"`
NumTxs int64 `json:"num_txs,string"` // Number of txs in a block
}
type EventDataNewEvidence struct {
Height int64 `json:"height"`
Evidence Evidence `json:"evidence"`
Height int64 `json:"height"`
}
// All txs fire EventDataTx
@@ -145,6 +150,7 @@ var (
EventQueryLock = QueryForEvent(EventLock)
EventQueryNewBlock = QueryForEvent(EventNewBlock)
EventQueryNewBlockHeader = QueryForEvent(EventNewBlockHeader)
EventQueryNewBlockEvents = QueryForEvent(EventNewBlockEvents)
EventQueryNewEvidence = QueryForEvent(EventNewEvidence)
EventQueryNewRound = QueryForEvent(EventNewRound)
EventQueryNewRoundStep = QueryForEvent(EventNewRoundStep)