Merge branch 'master' into wb/provider-mock

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
 

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{} }

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
+}

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)
+		}
+	}
+}

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,

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()
+}

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)
+}

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

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

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)
 }

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
+		}
+	}
+}

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 {

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.
 //

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
+}

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
+}

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