diff --git a/CHANGELOG_PENDING.md b/CHANGELOG_PENDING.md index 9c3add9b8..45762880f 100644 --- a/CHANGELOG_PENDING.md +++ b/CHANGELOG_PENDING.md @@ -30,6 +30,7 @@ Friendly reminder: We have a [bug bounty program](https://hackerone.com/tendermi - [ABCI] \#5447 Remove `SetOption` method from `ABCI.Client` interface - [ABCI] \#5447 Reset `Oneof` indexes for `Request` and `Response`. - [ABCI] \#5818 Use protoio for msg length delimitation. Migrates from int64 to uint64 length delimiters. + - [ABCI] \#3546 Add `mempool_error` field to `ResponseCheckTx`. This field will contain an error string if Tendermint encountered an error while adding a transaction to the mempool. (@williambanfield) - [Version] \#6494 `TMCoreSemVer` has been renamed to `TMVersion`. - It is not required any longer to set ldflags to set version strings - [abci/counter] \#6684 Delete counter example app @@ -81,6 +82,7 @@ Friendly reminder: We have a [bug bounty program](https://hackerone.com/tendermi - Tooling - [tools] \#6498 Set OS home dir to instead of the hardcoded PATH. (@JayT106) + - [cli/indexer] \#6676 Reindex events command line tooling. (@JayT106) ### FEATURES diff --git a/abci/types/types.pb.go b/abci/types/types.pb.go index f8781c0c4..6b00c587a 100644 --- a/abci/types/types.pb.go +++ b/abci/types/types.pb.go @@ -1827,17 +1827,19 @@ func (m *ResponseBeginBlock) GetEvents() []Event { } type ResponseCheckTx struct { - Code uint32 `protobuf:"varint,1,opt,name=code,proto3" json:"code,omitempty"` - Data []byte `protobuf:"bytes,2,opt,name=data,proto3" json:"data,omitempty"` - Log string `protobuf:"bytes,3,opt,name=log,proto3" json:"log,omitempty"` - Info string `protobuf:"bytes,4,opt,name=info,proto3" json:"info,omitempty"` - GasWanted int64 `protobuf:"varint,5,opt,name=gas_wanted,proto3" json:"gas_wanted,omitempty"` - GasUsed int64 `protobuf:"varint,6,opt,name=gas_used,proto3" json:"gas_used,omitempty"` - Events []Event `protobuf:"bytes,7,rep,name=events,proto3" json:"events,omitempty"` - Codespace string `protobuf:"bytes,8,opt,name=codespace,proto3" json:"codespace,omitempty"` - Sender string `protobuf:"bytes,9,opt,name=sender,proto3" json:"sender,omitempty"` - Priority int64 `protobuf:"varint,10,opt,name=priority,proto3" json:"priority,omitempty"` - MempoolError string `protobuf:"bytes,11,opt,name=mempool_error,json=mempoolError,proto3" json:"mempool_error,omitempty"` + Code uint32 `protobuf:"varint,1,opt,name=code,proto3" json:"code,omitempty"` + Data []byte `protobuf:"bytes,2,opt,name=data,proto3" json:"data,omitempty"` + Log string `protobuf:"bytes,3,opt,name=log,proto3" json:"log,omitempty"` + Info string `protobuf:"bytes,4,opt,name=info,proto3" json:"info,omitempty"` + GasWanted int64 `protobuf:"varint,5,opt,name=gas_wanted,proto3" json:"gas_wanted,omitempty"` + GasUsed int64 `protobuf:"varint,6,opt,name=gas_used,proto3" json:"gas_used,omitempty"` + Events []Event `protobuf:"bytes,7,rep,name=events,proto3" json:"events,omitempty"` + Codespace string `protobuf:"bytes,8,opt,name=codespace,proto3" json:"codespace,omitempty"` + Sender string `protobuf:"bytes,9,opt,name=sender,proto3" json:"sender,omitempty"` + Priority int64 `protobuf:"varint,10,opt,name=priority,proto3" json:"priority,omitempty"` + // mempool_error is set by Tendermint. + // ABCI applictions creating a ResponseCheckTX should not set mempool_error. + MempoolError string `protobuf:"bytes,11,opt,name=mempool_error,json=mempoolError,proto3" json:"mempool_error,omitempty"` } func (m *ResponseCheckTx) Reset() { *m = ResponseCheckTx{} } diff --git a/cmd/tendermint/commands/reindex_event.go b/cmd/tendermint/commands/reindex_event.go new file mode 100644 index 000000000..ddc585c1f --- /dev/null +++ b/cmd/tendermint/commands/reindex_event.go @@ -0,0 +1,251 @@ +package commands + +import ( + "errors" + "fmt" + "strings" + + "github.com/spf13/cobra" + tmdb "github.com/tendermint/tm-db" + + abcitypes "github.com/tendermint/tendermint/abci/types" + tmcfg "github.com/tendermint/tendermint/config" + "github.com/tendermint/tendermint/internal/libs/progressbar" + ctypes "github.com/tendermint/tendermint/rpc/core/types" + "github.com/tendermint/tendermint/state" + "github.com/tendermint/tendermint/state/indexer" + "github.com/tendermint/tendermint/state/indexer/sink/kv" + "github.com/tendermint/tendermint/state/indexer/sink/psql" + "github.com/tendermint/tendermint/store" + "github.com/tendermint/tendermint/types" +) + +const ( + reindexFailed = "event re-index failed: " +) + +// ReIndexEventCmd allows re-index the event by given block height interval +var ReIndexEventCmd = &cobra.Command{ + Use: "reindex-event", + Short: "reindex events to the event store backends", + Long: ` + reindex-event is an offline tooling to re-index block and tx events to the eventsinks, + you can run this command when the event store backend dropped/disconnected or you want to replace the backend. + The default start-height is 0, meaning the tooling will start reindex from the base block height(inclusive); and the + default end-height is 0, meaning the tooling will reindex until the latest block height(inclusive). User can omits + either or both arguments. + `, + Example: ` + tendermint reindex-event + tendermint reindex-event --start-height 2 + tendermint reindex-event --end-height 10 + tendermint reindex-event --start-height 2 --end-height 10 + `, + Run: func(cmd *cobra.Command, args []string) { + bs, ss, err := loadStateAndBlockStore(config) + if err != nil { + fmt.Println(reindexFailed, err) + return + } + + if err := checkValidHeight(bs); err != nil { + fmt.Println(reindexFailed, err) + return + } + + es, err := loadEventSinks(config) + if err != nil { + fmt.Println(reindexFailed, err) + return + } + + if err = eventReIndex(cmd, es, bs, ss); err != nil { + fmt.Println(reindexFailed, err) + return + } + + fmt.Println("event re-index finished") + }, +} + +var ( + startHeight int64 + endHeight int64 +) + +func init() { + ReIndexEventCmd.Flags().Int64Var(&startHeight, "start-height", 0, "the block height would like to start for re-index") + ReIndexEventCmd.Flags().Int64Var(&endHeight, "end-height", 0, "the block height would like to finish for re-index") +} + +func loadEventSinks(cfg *tmcfg.Config) ([]indexer.EventSink, error) { + // Check duplicated sinks. + sinks := map[string]bool{} + for _, s := range cfg.TxIndex.Indexer { + sl := strings.ToLower(s) + if sinks[sl] { + return nil, errors.New("found duplicated sinks, please check the tx-index section in the config.toml") + } + sinks[sl] = true + } + + eventSinks := []indexer.EventSink{} + + for k := range sinks { + switch k { + case string(indexer.NULL): + return nil, errors.New("found null event sink, please check the tx-index section in the config.toml") + case string(indexer.KV): + store, err := tmcfg.DefaultDBProvider(&tmcfg.DBContext{ID: "tx_index", Config: cfg}) + if err != nil { + return nil, err + } + eventSinks = append(eventSinks, kv.NewEventSink(store)) + case string(indexer.PSQL): + conn := cfg.TxIndex.PsqlConn + if conn == "" { + return nil, errors.New("the psql connection settings cannot be empty") + } + es, _, err := psql.NewEventSink(conn, chainID) + if err != nil { + return nil, err + } + eventSinks = append(eventSinks, es) + default: + return nil, errors.New("unsupported event sink type") + } + } + + if len(eventSinks) == 0 { + return nil, errors.New("no proper event sink can do event re-indexing," + + " please check the tx-index section in the config.toml") + } + + if !indexer.IndexingEnabled(eventSinks) { + return nil, fmt.Errorf("no event sink has been enabled") + } + + return eventSinks, nil +} + +func loadStateAndBlockStore(cfg *tmcfg.Config) (*store.BlockStore, state.Store, error) { + dbType := tmdb.BackendType(cfg.DBBackend) + + // Get BlockStore + blockStoreDB, err := tmdb.NewDB("blockstore", dbType, cfg.DBDir()) + if err != nil { + return nil, nil, err + } + blockStore := store.NewBlockStore(blockStoreDB) + + // Get StateStore + stateDB, err := tmdb.NewDB("state", dbType, cfg.DBDir()) + if err != nil { + return nil, nil, err + } + stateStore := state.NewStore(stateDB) + + return blockStore, stateStore, nil +} + +func eventReIndex(cmd *cobra.Command, es []indexer.EventSink, bs state.BlockStore, ss state.Store) error { + + var bar progressbar.Bar + bar.NewOption(startHeight-1, endHeight) + + fmt.Println("start re-indexing events:") + defer bar.Finish() + for i := startHeight; i <= endHeight; i++ { + select { + case <-cmd.Context().Done(): + return fmt.Errorf("event re-index terminated at height %d: %w", i, cmd.Context().Err()) + default: + b := bs.LoadBlock(i) + if b == nil { + return fmt.Errorf("not able to load block at height %d from the blockstore", i) + } + + r, err := ss.LoadABCIResponses(i) + if err != nil { + return fmt.Errorf("not able to load ABCI Response at height %d from the statestore", i) + } + + e := types.EventDataNewBlockHeader{ + Header: b.Header, + NumTxs: int64(len(b.Txs)), + ResultBeginBlock: *r.BeginBlock, + ResultEndBlock: *r.EndBlock, + } + + var batch *indexer.Batch + if e.NumTxs > 0 { + batch = indexer.NewBatch(e.NumTxs) + + for i, tx := range b.Data.Txs { + tr := abcitypes.TxResult{ + Height: b.Height, + Index: uint32(i), + Tx: tx, + Result: *(r.DeliverTxs[i]), + } + + _ = batch.Add(&tr) + } + } + + for _, sink := range es { + if err := sink.IndexBlockEvents(e); err != nil { + return fmt.Errorf("block event re-index at height %d failed: %w", i, err) + } + + if batch != nil { + if err := sink.IndexTxEvents(batch.Ops); err != nil { + return fmt.Errorf("tx event re-index at height %d failed: %w", i, err) + } + } + } + } + + bar.Play(i) + } + + return nil +} + +func checkValidHeight(bs state.BlockStore) error { + base := bs.Base() + + if startHeight == 0 { + startHeight = base + fmt.Printf("set the start block height to the base height of the blockstore %d \n", base) + } + + if startHeight < base { + return fmt.Errorf("%s (requested start height: %d, base height: %d)", ctypes.ErrHeightNotAvailable, startHeight, base) + } + + height := bs.Height() + + if startHeight > height { + return fmt.Errorf( + "%s (requested start height: %d, store height: %d)", ctypes.ErrHeightNotAvailable, startHeight, height) + } + + if endHeight == 0 || endHeight > height { + endHeight = height + fmt.Printf("set the end block height to the latest height of the blockstore %d \n", height) + } + + if endHeight < base { + return fmt.Errorf( + "%s (requested end height: %d, base height: %d)", ctypes.ErrHeightNotAvailable, endHeight, base) + } + + if endHeight < startHeight { + return fmt.Errorf( + "%s (requested the end height: %d is less than the start height: %d)", + ctypes.ErrInvalidRequest, startHeight, endHeight) + } + + return nil +} diff --git a/cmd/tendermint/commands/reindex_event_test.go b/cmd/tendermint/commands/reindex_event_test.go new file mode 100644 index 000000000..5d9459f5a --- /dev/null +++ b/cmd/tendermint/commands/reindex_event_test.go @@ -0,0 +1,171 @@ +package commands + +import ( + "context" + "errors" + "testing" + + "github.com/spf13/cobra" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + + abcitypes "github.com/tendermint/tendermint/abci/types" + tmcfg "github.com/tendermint/tendermint/config" + prototmstate "github.com/tendermint/tendermint/proto/tendermint/state" + "github.com/tendermint/tendermint/state/indexer" + "github.com/tendermint/tendermint/state/mocks" + "github.com/tendermint/tendermint/types" +) + +const ( + height int64 = 10 + base int64 = 2 +) + +func setupReIndexEventCmd() *cobra.Command { + reIndexEventCmd := &cobra.Command{ + Use: ReIndexEventCmd.Use, + Run: func(cmd *cobra.Command, args []string) {}, + } + + _ = reIndexEventCmd.ExecuteContext(context.Background()) + + return reIndexEventCmd +} + +func TestReIndexEventCheckHeight(t *testing.T) { + mockBlockStore := &mocks.BlockStore{} + mockBlockStore. + On("Base").Return(base). + On("Height").Return(height) + + testCases := []struct { + startHeight int64 + endHeight int64 + validHeight bool + }{ + {0, 0, true}, + {0, base, true}, + {0, base - 1, false}, + {0, height, true}, + {0, height + 1, true}, + {0, 0, true}, + {base - 1, 0, false}, + {base, 0, true}, + {base, base, true}, + {base, base - 1, false}, + {base, height, true}, + {base, height + 1, true}, + {height, 0, true}, + {height, base, false}, + {height, height - 1, false}, + {height, height, true}, + {height, height + 1, true}, + {height + 1, 0, false}, + } + + for _, tc := range testCases { + startHeight = tc.startHeight + endHeight = tc.endHeight + + err := checkValidHeight(mockBlockStore) + if tc.validHeight { + require.NoError(t, err) + } else { + require.Error(t, err) + } + } +} + +func TestLoadEventSink(t *testing.T) { + testCases := []struct { + sinks []string + connURL string + loadErr bool + }{ + {[]string{}, "", true}, + {[]string{"NULL"}, "", true}, + {[]string{"KV"}, "", false}, + {[]string{"KV", "KV"}, "", true}, + {[]string{"PSQL"}, "", true}, // true because empty connect url + {[]string{"PSQL"}, "wrongUrl", true}, // true because wrong connect url + // skip to test PSQL connect with correct url + {[]string{"UnsupportedSinkType"}, "wrongUrl", true}, + } + + for _, tc := range testCases { + cfg := tmcfg.TestConfig() + cfg.TxIndex.Indexer = tc.sinks + cfg.TxIndex.PsqlConn = tc.connURL + _, err := loadEventSinks(cfg) + if tc.loadErr { + require.Error(t, err) + } else { + require.NoError(t, err) + } + } +} + +func TestLoadBlockStore(t *testing.T) { + bs, ss, err := loadStateAndBlockStore(tmcfg.TestConfig()) + require.NoError(t, err) + require.NotNil(t, bs) + require.NotNil(t, ss) + +} +func TestReIndexEvent(t *testing.T) { + mockBlockStore := &mocks.BlockStore{} + mockStateStore := &mocks.Store{} + mockEventSink := &mocks.EventSink{} + + mockBlockStore. + On("Base").Return(base). + On("Height").Return(height). + On("LoadBlock", base).Return(nil).Once(). + On("LoadBlock", base).Return(&types.Block{Data: types.Data{Txs: types.Txs{make(types.Tx, 1)}}}). + On("LoadBlock", height).Return(&types.Block{Data: types.Data{Txs: types.Txs{make(types.Tx, 1)}}}) + + mockEventSink. + On("Type").Return(indexer.KV). + On("IndexBlockEvents", mock.AnythingOfType("types.EventDataNewBlockHeader")).Return(errors.New("")).Once(). + On("IndexBlockEvents", mock.AnythingOfType("types.EventDataNewBlockHeader")).Return(nil). + On("IndexTxEvents", mock.AnythingOfType("[]*types.TxResult")).Return(errors.New("")).Once(). + On("IndexTxEvents", mock.AnythingOfType("[]*types.TxResult")).Return(nil) + + dtx := abcitypes.ResponseDeliverTx{} + abciResp := &prototmstate.ABCIResponses{ + DeliverTxs: []*abcitypes.ResponseDeliverTx{&dtx}, + EndBlock: &abcitypes.ResponseEndBlock{}, + BeginBlock: &abcitypes.ResponseBeginBlock{}, + } + + mockStateStore. + On("LoadABCIResponses", base).Return(nil, errors.New("")).Once(). + On("LoadABCIResponses", base).Return(abciResp, nil). + On("LoadABCIResponses", height).Return(abciResp, nil) + + testCases := []struct { + startHeight int64 + endHeight int64 + reIndexErr bool + }{ + {base, height, true}, // LoadBlock error + {base, height, true}, // LoadABCIResponses error + {base, height, true}, // index block event error + {base, height, true}, // index tx event error + {base, base, false}, + {height, height, false}, + } + + for _, tc := range testCases { + startHeight = tc.startHeight + endHeight = tc.endHeight + + err := eventReIndex(setupReIndexEventCmd(), []indexer.EventSink{mockEventSink}, mockBlockStore, mockStateStore) + if tc.reIndexErr { + require.Error(t, err) + } else { + require.NoError(t, err) + } + } +} diff --git a/cmd/tendermint/main.go b/cmd/tendermint/main.go index 52a26b4f3..b40624cc3 100644 --- a/cmd/tendermint/main.go +++ b/cmd/tendermint/main.go @@ -15,6 +15,7 @@ func main() { rootCmd := cmd.RootCmd rootCmd.AddCommand( cmd.GenValidatorCmd, + cmd.ReIndexEventCmd, cmd.InitFilesCmd, cmd.ProbeUpnpCmd, cmd.LightCmd, diff --git a/internal/libs/progressbar/progressbar.go b/internal/libs/progressbar/progressbar.go new file mode 100644 index 000000000..072804c76 --- /dev/null +++ b/internal/libs/progressbar/progressbar.go @@ -0,0 +1,41 @@ +package progressbar + +import "fmt" + +// the progressbar indicates the current status and progress would be desired. +// ref: https://www.pixelstech.net/article/1596946473-A-simple-example-on-implementing-progress-bar-in-GoLang + +type Bar struct { + percent int64 // progress percentage + cur int64 // current progress + start int64 // the init starting value for progress + total int64 // total value for progress + rate string // the actual progress bar to be printed + graph string // the fill value for progress bar +} + +func (bar *Bar) NewOption(start, total int64) { + bar.cur = start + bar.start = start + bar.total = total + bar.graph = "█" + bar.percent = bar.getPercent() +} + +func (bar *Bar) getPercent() int64 { + return int64(float32(bar.cur-bar.start) / float32(bar.total-bar.start) * 100) +} + +func (bar *Bar) Play(cur int64) { + bar.cur = cur + last := bar.percent + bar.percent = bar.getPercent() + if bar.percent != last && bar.percent%2 == 0 { + bar.rate += bar.graph + } + fmt.Printf("\r[%-50s]%3d%% %8d/%d", bar.rate, bar.percent, bar.cur, bar.total) +} + +func (bar *Bar) Finish() { + fmt.Println() +} diff --git a/internal/libs/progressbar/progressbar_test.go b/internal/libs/progressbar/progressbar_test.go new file mode 100644 index 000000000..d135748f6 --- /dev/null +++ b/internal/libs/progressbar/progressbar_test.go @@ -0,0 +1,41 @@ +package progressbar + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestProgressBar(t *testing.T) { + zero := int64(0) + hundred := int64(100) + + var bar Bar + bar.NewOption(zero, hundred) + + require.Equal(t, zero, bar.start) + require.Equal(t, zero, bar.cur) + require.Equal(t, hundred, bar.total) + require.Equal(t, zero, bar.percent) + require.Equal(t, "█", bar.graph) + require.Equal(t, "", bar.rate) + + defer bar.Finish() + for i := zero; i <= hundred; i++ { + time.Sleep(1 * time.Millisecond) + bar.Play(i) + } + + require.Equal(t, zero, bar.start) + require.Equal(t, hundred, bar.cur) + require.Equal(t, hundred, bar.total) + require.Equal(t, hundred, bar.percent) + + var rate string + for i := zero; i < hundred/2; i++ { + rate += "█" + } + + require.Equal(t, rate, bar.rate) +} diff --git a/internal/mempool/v1/mempool.go b/internal/mempool/v1/mempool.go index aab3020ef..850600697 100644 --- a/internal/mempool/v1/mempool.go +++ b/internal/mempool/v1/mempool.go @@ -188,8 +188,8 @@ func (txmp *TxMempool) WaitForNextTx() <-chan struct{} { // NextGossipTx returns the next valid transaction to gossip. A caller must wait // for WaitForNextTx to signal a transaction is available to gossip first. It is // thread-safe. -func (txmp *TxMempool) NextGossipTx() *WrappedTx { - return txmp.gossipIndex.Front().Value.(*WrappedTx) +func (txmp *TxMempool) NextGossipTx() *clist.CElement { + return txmp.gossipIndex.Front() } // EnableTxsAvailable enables the mempool to trigger events when transactions diff --git a/internal/mempool/v1/reactor.go b/internal/mempool/v1/reactor.go index 436dd9d27..9deb7aace 100644 --- a/internal/mempool/v1/reactor.go +++ b/internal/mempool/v1/reactor.go @@ -9,6 +9,7 @@ import ( "time" cfg "github.com/tendermint/tendermint/config" + "github.com/tendermint/tendermint/internal/libs/clist" tmsync "github.com/tendermint/tendermint/internal/libs/sync" "github.com/tendermint/tendermint/internal/mempool" "github.com/tendermint/tendermint/internal/p2p" @@ -306,7 +307,7 @@ func (r *Reactor) processPeerUpdates() { func (r *Reactor) broadcastTxRoutine(peerID types.NodeID, closer *tmsync.Closer) { peerMempoolID := r.ids.GetForPeer(peerID) - var memTx *WrappedTx + var nextGossipTx *clist.CElement // remove the peer ID from the map of routines and mark the waitgroup as done defer func() { @@ -333,10 +334,10 @@ func (r *Reactor) broadcastTxRoutine(peerID types.NodeID, closer *tmsync.Closer) // This happens because the CElement we were looking at got garbage // collected (removed). That is, .NextWait() returned nil. Go ahead and // start from the beginning. - if memTx == nil { + if nextGossipTx == nil { select { case <-r.mempool.WaitForNextTx(): // wait until a tx is available - if memTx = r.mempool.NextGossipTx(); memTx == nil { + if nextGossipTx = r.mempool.NextGossipTx(); nextGossipTx == nil { continue } @@ -352,6 +353,8 @@ func (r *Reactor) broadcastTxRoutine(peerID types.NodeID, closer *tmsync.Closer) } } + memTx := nextGossipTx.Value.(*WrappedTx) + if r.peerMgr != nil { height := r.peerMgr.GetHeight(peerID) if height > 0 && height < memTx.height-1 { @@ -380,16 +383,8 @@ func (r *Reactor) broadcastTxRoutine(peerID types.NodeID, closer *tmsync.Closer) } select { - case <-memTx.gossipEl.NextWaitChan(): - // If there is a next element in gossip index, we point memTx to that node's - // value, otherwise we reset memTx to nil which will be checked at the - // parent for loop. - next := memTx.gossipEl.Next() - if next != nil { - memTx = next.Value.(*WrappedTx) - } else { - memTx = nil - } + case <-nextGossipTx.NextWaitChan(): + nextGossipTx = nextGossipTx.Next() case <-closer.Done(): // The peer is marked for removal via a PeerUpdate as the doneCh was diff --git a/light/example_test.go b/light/example_test.go index 0fd6640c4..2e0feb5e1 100644 --- a/light/example_test.go +++ b/light/example_test.go @@ -2,7 +2,6 @@ package light_test import ( "context" - "fmt" "io/ioutil" stdlog "log" "os" @@ -19,87 +18,12 @@ import ( rpctest "github.com/tendermint/tendermint/rpc/test" ) -// Automatically getting new headers and verifying them. -func ExampleClient_Update() { - ctx, cancel := context.WithCancel(context.Background()) - defer cancel() - conf := rpctest.CreateConfig("ExampleClient_Update") - - // Start a test application - app := kvstore.NewApplication() - _, closer, err := rpctest.StartTendermint(ctx, conf, app, rpctest.SuppressStdout) - if err != nil { - stdlog.Fatal(err) - } - defer func() { _ = closer(ctx) }() - - // give Tendermint time to generate some blocks - time.Sleep(5 * time.Second) - - dbDir, err := ioutil.TempDir("", "light-client-example") - if err != nil { - stdlog.Fatal(err) - } - defer os.RemoveAll(dbDir) - - chainID := conf.ChainID() - - primary, err := httpp.New(chainID, conf.RPC.ListenAddress) - if err != nil { - stdlog.Fatal(err) - } - - block, err := primary.LightBlock(ctx, 2) - if err != nil { - stdlog.Fatal(err) - } - - db, err := dbm.NewGoLevelDB("light-client-db", dbDir) - if err != nil { - stdlog.Fatal(err) - } - - c, err := light.NewClient( - ctx, - chainID, - light.TrustOptions{ - Period: 504 * time.Hour, // 21 days - Height: 2, - Hash: block.Hash(), - }, - primary, - []provider.Provider{primary}, // NOTE: primary should not be used here - dbs.New(db), - light.Logger(log.TestingLogger()), - ) - if err != nil { - stdlog.Fatal(err) - } - defer func() { - if err := c.Cleanup(); err != nil { - stdlog.Fatal(err) - } - }() - - time.Sleep(2 * time.Second) - - h, err := c.Update(ctx, time.Now()) - if err != nil { - stdlog.Fatal(err) - } - - if h != nil && h.Height > 2 { - fmt.Println("successful update") - } else { - fmt.Println("update failed") - } -} - // Manually getting light blocks and verifying them. -func ExampleClient_VerifyLightBlockAtHeight() { +func ExampleClient() { ctx, cancel := context.WithCancel(context.Background()) defer cancel() conf := rpctest.CreateConfig("ExampleClient_VerifyLightBlockAtHeight") + logger := log.TestingLogger() // Start a test application app := kvstore.NewApplication() @@ -110,9 +34,6 @@ func ExampleClient_VerifyLightBlockAtHeight() { } defer func() { _ = closer(ctx) }() - // give Tendermint time to generate some blocks - time.Sleep(5 * time.Second) - dbDir, err := ioutil.TempDir("", "light-client-example") if err != nil { stdlog.Fatal(err) @@ -126,6 +47,9 @@ func ExampleClient_VerifyLightBlockAtHeight() { stdlog.Fatal(err) } + // give Tendermint time to generate some blocks + time.Sleep(5 * time.Second) + block, err := primary.LightBlock(ctx, 2) if err != nil { stdlog.Fatal(err) @@ -146,7 +70,7 @@ func ExampleClient_VerifyLightBlockAtHeight() { primary, []provider.Provider{primary}, // NOTE: primary should not be used here dbs.New(db), - light.Logger(log.TestingLogger()), + light.Logger(logger), ) if err != nil { stdlog.Fatal(err) @@ -157,15 +81,26 @@ func ExampleClient_VerifyLightBlockAtHeight() { } }() + // wait for a few more blocks to be produced + time.Sleep(2 * time.Second) + + // veify the block at height 3 _, err = c.VerifyLightBlockAtHeight(context.Background(), 3, time.Now()) if err != nil { stdlog.Fatal(err) } - h, err := c.TrustedLightBlock(3) + // retrieve light block at height 3 + _, err = c.TrustedLightBlock(3) if err != nil { stdlog.Fatal(err) } - fmt.Println("got header", h.Height) + // update to the latest height + lb, err := c.Update(ctx, time.Now()) + if err != nil { + stdlog.Fatal(err) + } + + logger.Info("verified light block", "light-block", lb) } diff --git a/light/light_test.go b/light/light_test.go index a224d71b2..f5d2ddd89 100644 --- a/light/light_test.go +++ b/light/light_test.go @@ -17,6 +17,7 @@ import ( httpp "github.com/tendermint/tendermint/light/provider/http" dbs "github.com/tendermint/tendermint/light/store/db" rpctest "github.com/tendermint/tendermint/rpc/test" + "github.com/tendermint/tendermint/types" ) // NOTE: these are ports of the tests from example_test.go but @@ -48,7 +49,8 @@ func TestClientIntegration_Update(t *testing.T) { primary, err := httpp.New(chainID, conf.RPC.ListenAddress) require.NoError(t, err) - block, err := primary.LightBlock(ctx, 2) + // give Tendermint time to generate some blocks + block, err := waitForBlock(ctx, primary, 2) require.NoError(t, err) db, err := dbm.NewGoLevelDB("light-client-db", dbDir) @@ -71,7 +73,9 @@ func TestClientIntegration_Update(t *testing.T) { defer func() { require.NoError(t, c.Cleanup()) }() - time.Sleep(2 * time.Second) + // ensure Tendermint is at height 3 or higher + _, err = waitForBlock(ctx, primary, 3) + require.NoError(t, err) h, err := c.Update(ctx, time.Now()) require.NoError(t, err) @@ -94,9 +98,6 @@ func TestClientIntegration_VerifyLightBlockAtHeight(t *testing.T) { require.NoError(t, err) defer func() { require.NoError(t, closer(ctx)) }() - // give Tendermint time to generate some blocks - time.Sleep(5 * time.Second) - dbDir, err := ioutil.TempDir("", "light-client-test-verify-example") require.NoError(t, err) defer os.RemoveAll(dbDir) @@ -106,7 +107,8 @@ func TestClientIntegration_VerifyLightBlockAtHeight(t *testing.T) { primary, err := httpp.New(chainID, conf.RPC.ListenAddress) require.NoError(t, err) - block, err := primary.LightBlock(ctx, 2) + // give Tendermint time to generate some blocks + block, err := waitForBlock(ctx, primary, 2) require.NoError(t, err) db, err := dbm.NewGoLevelDB("light-client-db", dbDir) @@ -128,6 +130,10 @@ func TestClientIntegration_VerifyLightBlockAtHeight(t *testing.T) { defer func() { require.NoError(t, c.Cleanup()) }() + // ensure Tendermint is at height 3 or higher + _, err = waitForBlock(ctx, primary, 3) + require.NoError(t, err) + _, err = c.VerifyLightBlockAtHeight(ctx, 3, time.Now()) require.NoError(t, err) @@ -136,3 +142,23 @@ func TestClientIntegration_VerifyLightBlockAtHeight(t *testing.T) { require.EqualValues(t, 3, h.Height) } + +func waitForBlock(ctx context.Context, p provider.Provider, height int64) (*types.LightBlock, error) { + for { + block, err := p.LightBlock(ctx, height) + switch err { + case nil: + return block, nil + // node isn't running yet, wait 1 second and repeat + case provider.ErrNoResponse, provider.ErrHeightTooHigh: + timer := time.NewTimer(1 * time.Second) + select { + case <-ctx.Done(): + return nil, ctx.Err() + case <-timer.C: + } + default: + return nil, err + } + } +} diff --git a/proto/tendermint/abci/types.proto b/proto/tendermint/abci/types.proto index f0d70548e..7126488d0 100644 --- a/proto/tendermint/abci/types.proto +++ b/proto/tendermint/abci/types.proto @@ -186,17 +186,20 @@ message ResponseBeginBlock { } message ResponseCheckTx { - uint32 code = 1; - bytes data = 2; - string log = 3; // nondeterministic - string info = 4; // nondeterministic - int64 gas_wanted = 5 [json_name = "gas_wanted"]; - int64 gas_used = 6 [json_name = "gas_used"]; - repeated Event events = 7 [(gogoproto.nullable) = false, (gogoproto.jsontag) = "events,omitempty"]; - string codespace = 8; - string sender = 9; - int64 priority = 10; - string mempool_error = 11; + uint32 code = 1; + bytes data = 2; + string log = 3; // nondeterministic + string info = 4; // nondeterministic + int64 gas_wanted = 5 [json_name = "gas_wanted"]; + int64 gas_used = 6 [json_name = "gas_used"]; + repeated Event events = 7 [(gogoproto.nullable) = false, (gogoproto.jsontag) = "events,omitempty"]; + string codespace = 8; + string sender = 9; + int64 priority = 10; + + // mempool_error is set by Tendermint. + // ABCI applictions creating a ResponseCheckTX should not set mempool_error. + string mempool_error = 11; } message ResponseDeliverTx { diff --git a/state/indexer/eventsink.go b/state/indexer/eventsink.go index 8c2529103..19d93e4e5 100644 --- a/state/indexer/eventsink.go +++ b/state/indexer/eventsink.go @@ -16,6 +16,8 @@ const ( PSQL EventSinkType = "psql" ) +//go:generate mockery --case underscore --name EventSink + // EventSink interface is defined the APIs for the IndexerService to interact with the data store, // including the block/transaction indexing and the search functions. // diff --git a/state/mocks/block_store.go b/state/mocks/block_store.go new file mode 100644 index 000000000..99cadd081 --- /dev/null +++ b/state/mocks/block_store.go @@ -0,0 +1,194 @@ +// Code generated by mockery 2.7.5. DO NOT EDIT. + +package mocks + +import ( + mock "github.com/stretchr/testify/mock" + + types "github.com/tendermint/tendermint/types" +) + +// BlockStore is an autogenerated mock type for the BlockStore type +type BlockStore struct { + mock.Mock +} + +// Base provides a mock function with given fields: +func (_m *BlockStore) Base() int64 { + ret := _m.Called() + + var r0 int64 + if rf, ok := ret.Get(0).(func() int64); ok { + r0 = rf() + } else { + r0 = ret.Get(0).(int64) + } + + return r0 +} + +// Height provides a mock function with given fields: +func (_m *BlockStore) Height() int64 { + ret := _m.Called() + + var r0 int64 + if rf, ok := ret.Get(0).(func() int64); ok { + r0 = rf() + } else { + r0 = ret.Get(0).(int64) + } + + return r0 +} + +// LoadBaseMeta provides a mock function with given fields: +func (_m *BlockStore) LoadBaseMeta() *types.BlockMeta { + ret := _m.Called() + + var r0 *types.BlockMeta + if rf, ok := ret.Get(0).(func() *types.BlockMeta); ok { + r0 = rf() + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(*types.BlockMeta) + } + } + + return r0 +} + +// LoadBlock provides a mock function with given fields: height +func (_m *BlockStore) LoadBlock(height int64) *types.Block { + ret := _m.Called(height) + + var r0 *types.Block + if rf, ok := ret.Get(0).(func(int64) *types.Block); ok { + r0 = rf(height) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(*types.Block) + } + } + + return r0 +} + +// LoadBlockByHash provides a mock function with given fields: hash +func (_m *BlockStore) LoadBlockByHash(hash []byte) *types.Block { + ret := _m.Called(hash) + + var r0 *types.Block + if rf, ok := ret.Get(0).(func([]byte) *types.Block); ok { + r0 = rf(hash) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(*types.Block) + } + } + + return r0 +} + +// LoadBlockCommit provides a mock function with given fields: height +func (_m *BlockStore) LoadBlockCommit(height int64) *types.Commit { + ret := _m.Called(height) + + var r0 *types.Commit + if rf, ok := ret.Get(0).(func(int64) *types.Commit); ok { + r0 = rf(height) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(*types.Commit) + } + } + + return r0 +} + +// LoadBlockMeta provides a mock function with given fields: height +func (_m *BlockStore) LoadBlockMeta(height int64) *types.BlockMeta { + ret := _m.Called(height) + + var r0 *types.BlockMeta + if rf, ok := ret.Get(0).(func(int64) *types.BlockMeta); ok { + r0 = rf(height) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(*types.BlockMeta) + } + } + + return r0 +} + +// LoadBlockPart provides a mock function with given fields: height, index +func (_m *BlockStore) LoadBlockPart(height int64, index int) *types.Part { + ret := _m.Called(height, index) + + var r0 *types.Part + if rf, ok := ret.Get(0).(func(int64, int) *types.Part); ok { + r0 = rf(height, index) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(*types.Part) + } + } + + return r0 +} + +// LoadSeenCommit provides a mock function with given fields: height +func (_m *BlockStore) LoadSeenCommit(height int64) *types.Commit { + ret := _m.Called(height) + + var r0 *types.Commit + if rf, ok := ret.Get(0).(func(int64) *types.Commit); ok { + r0 = rf(height) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(*types.Commit) + } + } + + return r0 +} + +// PruneBlocks provides a mock function with given fields: height +func (_m *BlockStore) PruneBlocks(height int64) (uint64, error) { + ret := _m.Called(height) + + var r0 uint64 + if rf, ok := ret.Get(0).(func(int64) uint64); ok { + r0 = rf(height) + } else { + r0 = ret.Get(0).(uint64) + } + + var r1 error + if rf, ok := ret.Get(1).(func(int64) error); ok { + r1 = rf(height) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + +// SaveBlock provides a mock function with given fields: block, blockParts, seenCommit +func (_m *BlockStore) SaveBlock(block *types.Block, blockParts *types.PartSet, seenCommit *types.Commit) { + _m.Called(block, blockParts, seenCommit) +} + +// Size provides a mock function with given fields: +func (_m *BlockStore) Size() int64 { + ret := _m.Called() + + var r0 int64 + if rf, ok := ret.Get(0).(func() int64); ok { + r0 = rf() + } else { + r0 = ret.Get(0).(int64) + } + + return r0 +} \ No newline at end of file diff --git a/state/mocks/event_sink.go b/state/mocks/event_sink.go new file mode 100644 index 000000000..749515ccf --- /dev/null +++ b/state/mocks/event_sink.go @@ -0,0 +1,167 @@ +// Code generated by mockery 2.7.5. DO NOT EDIT. + +package mocks + +import ( + context "context" + + mock "github.com/stretchr/testify/mock" + indexer "github.com/tendermint/tendermint/state/indexer" + + query "github.com/tendermint/tendermint/libs/pubsub/query" + + tenderminttypes "github.com/tendermint/tendermint/types" + + types "github.com/tendermint/tendermint/abci/types" +) + +// EventSink is an autogenerated mock type for the EventSink type +type EventSink struct { + mock.Mock +} + +// GetTxByHash provides a mock function with given fields: _a0 +func (_m *EventSink) GetTxByHash(_a0 []byte) (*types.TxResult, error) { + ret := _m.Called(_a0) + + var r0 *types.TxResult + if rf, ok := ret.Get(0).(func([]byte) *types.TxResult); ok { + r0 = rf(_a0) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).(*types.TxResult) + } + } + + var r1 error + if rf, ok := ret.Get(1).(func([]byte) error); ok { + r1 = rf(_a0) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + +// HasBlock provides a mock function with given fields: _a0 +func (_m *EventSink) HasBlock(_a0 int64) (bool, error) { + ret := _m.Called(_a0) + + var r0 bool + if rf, ok := ret.Get(0).(func(int64) bool); ok { + r0 = rf(_a0) + } else { + r0 = ret.Get(0).(bool) + } + + var r1 error + if rf, ok := ret.Get(1).(func(int64) error); ok { + r1 = rf(_a0) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + +// IndexBlockEvents provides a mock function with given fields: _a0 +func (_m *EventSink) IndexBlockEvents(_a0 tenderminttypes.EventDataNewBlockHeader) error { + ret := _m.Called(_a0) + + var r0 error + if rf, ok := ret.Get(0).(func(tenderminttypes.EventDataNewBlockHeader) error); ok { + r0 = rf(_a0) + } else { + r0 = ret.Error(0) + } + + return r0 +} + +// IndexTxEvents provides a mock function with given fields: _a0 +func (_m *EventSink) IndexTxEvents(_a0 []*types.TxResult) error { + ret := _m.Called(_a0) + + var r0 error + if rf, ok := ret.Get(0).(func([]*types.TxResult) error); ok { + r0 = rf(_a0) + } else { + r0 = ret.Error(0) + } + + return r0 +} + +// SearchBlockEvents provides a mock function with given fields: _a0, _a1 +func (_m *EventSink) SearchBlockEvents(_a0 context.Context, _a1 *query.Query) ([]int64, error) { + ret := _m.Called(_a0, _a1) + + var r0 []int64 + if rf, ok := ret.Get(0).(func(context.Context, *query.Query) []int64); ok { + r0 = rf(_a0, _a1) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).([]int64) + } + } + + var r1 error + if rf, ok := ret.Get(1).(func(context.Context, *query.Query) error); ok { + r1 = rf(_a0, _a1) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + +// SearchTxEvents provides a mock function with given fields: _a0, _a1 +func (_m *EventSink) SearchTxEvents(_a0 context.Context, _a1 *query.Query) ([]*types.TxResult, error) { + ret := _m.Called(_a0, _a1) + + var r0 []*types.TxResult + if rf, ok := ret.Get(0).(func(context.Context, *query.Query) []*types.TxResult); ok { + r0 = rf(_a0, _a1) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).([]*types.TxResult) + } + } + + var r1 error + if rf, ok := ret.Get(1).(func(context.Context, *query.Query) error); ok { + r1 = rf(_a0, _a1) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + +// Stop provides a mock function with given fields: +func (_m *EventSink) Stop() error { + ret := _m.Called() + + var r0 error + if rf, ok := ret.Get(0).(func() error); ok { + r0 = rf() + } else { + r0 = ret.Error(0) + } + + return r0 +} + +// Type provides a mock function with given fields: +func (_m *EventSink) Type() indexer.EventSinkType { + ret := _m.Called() + + var r0 indexer.EventSinkType + if rf, ok := ret.Get(0).(func() indexer.EventSinkType); ok { + r0 = rf() + } else { + r0 = ret.Get(0).(indexer.EventSinkType) + } + + return r0 +} diff --git a/state/services.go b/state/services.go index a46863904..0ab6b78d0 100644 --- a/state/services.go +++ b/state/services.go @@ -9,6 +9,8 @@ import ( // NOTE: Interfaces used by RPC must be thread safe! //------------------------------------------------------ +//go:generate mockery --case underscore --name BlockStore + //------------------------------------------------------ // blockstore