From 61a4e70365715b3abad098eeee99c5469be6213b Mon Sep 17 00:00:00 2001 From: Callum Waters Date: Wed, 21 Sep 2022 13:39:16 +0200 Subject: [PATCH] fix broken tests --- abci/client/client.go | 1 + abci/client/socket_client.go | 9 --- abci/client/socket_client_test.go | 13 ++-- abci/example/kvstore/kvstore_test.go | 44 ++++------- abci/server/socket_server.go | 18 ++--- abci/tests/client_server_test.go | 15 +++- cmd/tendermint/commands/reindex_event.go | 38 +++++----- cmd/tendermint/commands/reindex_event_test.go | 4 +- consensus/mempool_test.go | 8 +- proxy/app_conn_test.go | 4 +- state/execution.go | 1 - state/execution_test.go | 22 +++--- state/indexer/block.go | 2 +- state/indexer/block/kv/kv.go | 6 +- state/indexer/block/kv/kv_test.go | 76 +++++++++---------- state/indexer/block/null/null.go | 2 +- state/indexer/mocks/block_indexer.go | 4 +- state/indexer/sink/psql/backport.go | 2 +- state/indexer/sink/psql/psql.go | 8 +- state/indexer/sink/psql/psql_test.go | 22 +++--- state/state_test.go | 20 ++--- state/store.go | 25 +++++- state/txindex/indexer_service.go | 14 ++-- state/txindex/indexer_service_test.go | 18 ++++- statesync/reactor_test.go | 2 +- statesync/syncer_test.go | 54 ++++++------- types/event_bus.go | 14 +++- types/event_bus_test.go | 57 ++++++++++++-- types/events.go | 18 +++-- 29 files changed, 292 insertions(+), 229 deletions(-) diff --git a/abci/client/client.go b/abci/client/client.go index 27c1aceba..375317e04 100644 --- a/abci/client/client.go +++ b/abci/client/client.go @@ -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) } diff --git a/abci/client/socket_client.go b/abci/client/socket_client.go index 001df3115..9f80d8953 100644 --- a/abci/client/socket_client.go +++ b/abci/client/socket_client.go @@ -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 } diff --git a/abci/client/socket_client_test.go b/abci/client/socket_client_test.go index 65cc72c3e..b8b44a929 100644 --- a/abci/client/socket_client_test.go +++ b/abci/client/socket_client_test.go @@ -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) } diff --git a/abci/example/kvstore/kvstore_test.go b/abci/example/kvstore/kvstore_test.go index ba3c9e343..03bf17a26 100644 --- a/abci/example/kvstore/kvstore_test.go +++ b/abci/example/kvstore/kvstore_test.go @@ -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 } diff --git a/abci/server/socket_server.go b/abci/server/socket_server.go index 17bd28ed8..6507ac661 100644 --- a/abci/server/socket_server.go +++ b/abci/server/socket_server.go @@ -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") } } diff --git a/abci/tests/client_server_test.go b/abci/tests/client_server_test.go index 80d4f7c0d..4d654c6d0 100644 --- a/abci/tests/client_server_test.go +++ b/abci/tests/client_server_test.go @@ -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) + } + }) } diff --git a/cmd/tendermint/commands/reindex_event.go b/cmd/tendermint/commands/reindex_event.go index 519705963..c7c664eb7 100644 --- a/cmd/tendermint/commands/reindex_event.go +++ b/cmd/tendermint/commands/reindex_event.go @@ -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 diff --git a/cmd/tendermint/commands/reindex_event_test.go b/cmd/tendermint/commands/reindex_event_test.go index dd8484684..455a77855 100644 --- a/cmd/tendermint/commands/reindex_event_test.go +++ b/cmd/tendermint/commands/reindex_event_test.go @@ -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(). diff --git a/consensus/mempool_test.go b/consensus/mempool_test.go index bc5650882..c78d0fc6b 100644 --- a/consensus/mempool_test.go +++ b/consensus/mempool_test.go @@ -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") } diff --git a/proxy/app_conn_test.go b/proxy/app_conn_test.go index 75c252ff0..37db543b0 100644 --- a/proxy/app_conn_test.go +++ b/proxy/app_conn_test.go @@ -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 diff --git a/state/execution.go b/state/execution.go index f4baf5780..ae9501baa 100644 --- a/state/execution.go +++ b/state/execution.go @@ -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) } diff --git a/state/execution_test.go b/state/execution_test.go index 7066d1d1a..fa5c56908 100644 --- a/state/execution_test.go +++ b/state/execution_test.go @@ -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) diff --git a/state/indexer/block.go b/state/indexer/block.go index 2b6f694bc..365b3ae08 100644 --- a/state/indexer/block.go +++ b/state/indexer/block.go @@ -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. diff --git a/state/indexer/block/kv/kv.go b/state/indexer/block/kv/kv.go index f6f06c611..50a918fad 100644 --- a/state/indexer/block/kv/kv.go +++ b/state/indexer/block/kv/kv.go @@ -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) } diff --git a/state/indexer/block/kv/kv_test.go b/state/indexer/block/kv/kv_test.go index c06768f89..2ce358fa3 100644 --- a/state/indexer/block/kv/kv_test.go +++ b/state/indexer/block/kv/kv_test.go @@ -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, }, }, }, diff --git a/state/indexer/block/null/null.go b/state/indexer/block/null/null.go index 1478bf70e..131c6412f 100644 --- a/state/indexer/block/null/null.go +++ b/state/indexer/block/null/null.go @@ -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 } diff --git a/state/indexer/mocks/block_indexer.go b/state/indexer/mocks/block_indexer.go index 8eade138a..0832ed73a 100644 --- a/state/indexer/mocks/block_indexer.go +++ b/state/indexer/mocks/block_indexer.go @@ -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) diff --git a/state/indexer/sink/psql/backport.go b/state/indexer/sink/psql/backport.go index c4819b3b6..ce4a137db 100644 --- a/state/indexer/sink/psql/backport.go +++ b/state/indexer/sink/psql/backport.go @@ -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) } diff --git a/state/indexer/sink/psql/psql.go b/state/indexer/sink/psql/psql.go index 4401fd7cc..0c71541e5 100644 --- a/state/indexer/sink/psql/psql.go +++ b/state/indexer/sink/psql/psql.go @@ -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 diff --git a/state/indexer/sink/psql/psql_test.go b/state/indexer/sink/psql/psql_test.go index ba0e9b2b0..68d32dcf6 100644 --- a/state/indexer/sink/psql/psql_test.go +++ b/state/indexer/sink/psql/psql_test.go @@ -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"), }, } } diff --git a/state/state_test.go b/state/state_test.go index f9eb51ff8..c820fb7d4 100644 --- a/state/state_test.go +++ b/state/state_test.go @@ -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) } diff --git a/state/store.go b/state/store.go index bffdead63..e3d768409 100644 --- a/state/store.go +++ b/state/store.go @@ -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 diff --git a/state/txindex/indexer_service.go b/state/txindex/indexer_service.go index b2c95b164..a12937ef1 100644 --- a/state/txindex/indexer_service.go +++ b/state/txindex/indexer_service.go @@ -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) diff --git a/state/txindex/indexer_service_test.go b/state/txindex/indexer_service_test.go index 88ae33779..d1460f8e3 100644 --- a/state/txindex/indexer_service_test.go +++ b/state/txindex/indexer_service_test.go @@ -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) diff --git a/statesync/reactor_test.go b/statesync/reactor_test.go index e4ce3d344..85ebd0489 100644 --- a/statesync/reactor_test.go +++ b/statesync/reactor_test.go @@ -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) diff --git a/statesync/syncer_test.go b/statesync/syncer_test.go index 4dabe7288..2eb40c612 100644 --- a/statesync/syncer_test.go +++ b/statesync/syncer_test.go @@ -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 { diff --git a/types/event_bus.go b/types/event_bus.go index 236226408..cbc6c18f4 100644 --- a/types/event_bus.go +++ b/types/event_bus.go @@ -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) } diff --git a/types/event_bus_test.go b/types/event_bus_test.go index d258df4ff..96abdaf8a 100644 --- a/types/event_bus_test.go +++ b/types/event_bus_test.go @@ -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, diff --git a/types/events.go b/types/events.go index fedbd4ed2..857beee6a 100644 --- a/types/events.go +++ b/types/events.go @@ -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)