extend timeouts

Bumps [gaurav-nelson/github-action-markdown-link-check](https://github.com/gaurav-nelson/github-action-markdown-link-check) from 1.0.12 to 1.0.13.
- [Release notes](https://github.com/gaurav-nelson/github-action-markdown-link-check/releases)
- [Commits](https://github.com/gaurav-nelson/github-action-markdown-link-check/compare/1.0.12...1.0.13)

---
updated-dependencies:
- dependency-name: gaurav-nelson/github-action-markdown-link-check
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>

Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

.github/workflows/linkchecker.yml

@@ -7,6 +7,6 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v2.3.4
-      - uses: gaurav-nelson/github-action-markdown-link-check@1.0.12
+      - uses: gaurav-nelson/github-action-markdown-link-check@1.0.13
         with:
           folder-path: "docs"

.gitignore

@@ -15,7 +15,7 @@
 .vagrant
 .vendor-new/
 .vscode/
-abci-cli
+abci/abci-cli
 addrbook.json
 artifacts/*
 build/*

CHANGELOG_PENDING.md

@@ -22,6 +22,7 @@ Friendly reminder: We have a [bug bounty program](https://hackerone.com/tendermi
   - [config] \#6462 Move `PrivValidator` configuration out of `BaseConfig` into its own section. (@tychoish)
   - [rpc] \#6610 Add MaxPeerBlockHeight into /status rpc call (@JayT106)
   - [libs/CList] \#6626 Automatically detach the prev/next elements in Remove function (@JayT106)
+  - [fastsync/rpc] \#6620 Add TotalSyncedTime & RemainingTime to SyncInfo in /status RPC  (@JayT106)
 
 - Apps
   - [ABCI] \#6408 Change the `key` and `value` fields from `[]byte` to `string` in the `EventAttribute` type. (@alexanderbez)
@@ -147,3 +148,4 @@ Friendly reminder: We have a [bug bounty program](https://hackerone.com/tendermi
 - [rpc] \#6507 fix RPC client doesn't handle url's without ports (@JayT106)
 - [statesync] \#6463 Adds Reverse Sync feature to fetch historical light blocks after state sync in order to verify any evidence (@cmwaters)
 - [fastsync] \#6590 Update the metrics during fast-sync (@JayT106)
+- [gitignore] \#6668 Fix gitignore of abci-cli (@tanyabouman)

Makefile

@@ -202,7 +202,7 @@ format:
 
 lint:
 	@echo "--> Running linter"
-	@golangci-lint run
+	go run github.com/golangci/golangci-lint/cmd/golangci-lint run
 .PHONY: lint
 
 DESTINATION = ./index.html.md

docs/app-dev/indexing-transactions.md

@@ -31,24 +31,61 @@ For example:
 
 would be equal to the composite key of `jack.account.number`.
 
-Let's take a look at the `[tx_index]` config section:
+By default, Tendermint will index all transactions by their respective hashes
+and height and blocks by their height.
+
+## Configuration
+
+Operators can configure indexing via the `[tx_index]` section. The `indexer`
+field takes a series of supported indexers. If `null` is included, indexing will
+be turned off regardless of other values provided.
 
 ```toml
-##### transactions indexer configuration options #####
-[tx_index]
+[tx-index]
 
-# What indexer to use for transactions
+# The backend database list to back the indexer.
+# If list contains null, meaning no indexer service will be used.
+#
+# The application will set which txs to index. In some cases a node operator will be able
+# to decide which txs to index based on configuration set in the application.
 #
 # Options:
 #   1) "null"
 #   2) "kv" (default) - the simplest possible indexer, backed by key-value storage (defaults to levelDB; see DBBackend).
-indexer = "kv"
+#     - When "kv" is chosen "tx.height" and "tx.hash" will always be indexed.
+#   3) "psql" - the indexer services backed by PostgreSQL.
+# indexer = []
 ```
 
-By default, Tendermint will index all transactions by their respective hashes
-and height and blocks by their height.
+### Supported Indexers
 
-You can turn off indexing completely by setting `tx_index` to `null`.
+#### KV
+
+The `kv` indexer type is an embedded key-value store supported by the main
+underling Tendermint database. Using the `kv` indexer type allows you to query
+for block and transaction events directly against Tendermint's RPC. However, the
+query syntax is limited and so this indexer type might be deprecated or removed
+entirely in the future.
+
+#### PostgreSQL
+
+The `psql` indexer type allows an operator to enable block and transaction event
+indexing by proxying it to an external PostgreSQL instance allowing for the events
+to be stored in relational models. Since the events are stored in a RDBMS, operators
+can leverage SQL to perform a series of rich and complex queries that are not
+supported by the `kv` indexer type. Since operators can leverage SQL directly,
+searching is not enabled for the `psql` indexer type via Tendermint's RPC -- any
+such query will fail.
+
+Note, the SQL schema is stored in `state/indexer/sink/psql/schema.sql` and operators
+must explicitly create the relations prior to starting Tendermint and enabling
+the `psql` indexer type.
+
+Example:
+
+```shell
+$ psql ... -f state/indexer/sink/psql/schema.sql
+```
 
 ## Default Indexes
 

docs/nodes/running-in-production.md

@@ -268,6 +268,8 @@ While we do not favor any operation system, more secure and stable Linux server
 distributions (like Centos) should be preferred over desktop operation systems
 (like Mac OS).
 
+Native Windows support is not provided. If you are using a windows machine, you can try using the [bash shell](https://docs.microsoft.com/en-us/windows/wsl/install-win10).
+
 ### Miscellaneous
 
 NOTE: if you are going to use Tendermint in a public domain, make sure

go.mod

@@ -10,33 +10,47 @@ require (
 	github.com/btcsuite/btcd v0.22.0-beta
 	github.com/btcsuite/btcutil v1.0.3-0.20201208143702-a53e38424cce
 	github.com/fortytw2/leaktest v1.3.0
-	github.com/go-kit/kit v0.10.0
+	github.com/go-kit/kit v0.11.0
+	github.com/go-lintpack/lintpack v0.5.2 // indirect
 	github.com/gogo/protobuf v1.3.2
 	github.com/golang/protobuf v1.5.2
+	github.com/golangci/errcheck v0.0.0-20181223084120-ef45e06d44b6 // indirect
+	github.com/golangci/go-tools v0.0.0-20190318055746-e32c54105b7c // indirect
+	github.com/golangci/goconst v0.0.0-20180610141641-041c5f2b40f3 // indirect
+	github.com/golangci/gocyclo v0.0.0-20180528134321-2becd97e67ee // indirect
+	github.com/golangci/golangci-lint v1.38.0 // indirect
+	github.com/golangci/gosec v0.0.0-20190211064107-66fb7fc33547 // indirect
+	github.com/golangci/ineffassign v0.0.0-20190609212857-42439a7714cc // indirect
+	github.com/golangci/prealloc v0.0.0-20180630174525-215b22d4de21 // indirect
 	github.com/google/orderedcode v0.0.1
 	github.com/google/uuid v1.2.0
 	github.com/gorilla/websocket v1.4.2
 	github.com/grpc-ecosystem/go-grpc-middleware v1.3.0
 	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0
+	github.com/klauspost/cpuid v1.2.0 // indirect
 	github.com/lib/pq v1.10.2
 	github.com/libp2p/go-buffer-pool v0.0.2
 	github.com/minio/highwayhash v1.0.2
 	github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e // indirect
 	github.com/oasisprotocol/curve25519-voi v0.0.0-20210609091139-0a56a4bca00b
 	github.com/ory/dockertest v3.3.5+incompatible
-	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.11.0
 	github.com/rcrowley/go-metrics v0.0.0-20200313005456-10cdbea86bc0
 	github.com/rs/cors v1.8.0
 	github.com/rs/zerolog v1.23.0
 	github.com/sasha-s/go-deadlock v0.2.1-0.20190427202633-1595213edefa
+	github.com/shirou/gopsutil v0.0.0-20180427012116-c95755e4bcd7 // indirect
+	github.com/shirou/w32 v0.0.0-20160930032740-bb4de0191aa4 // indirect
 	github.com/snikch/goodman v0.0.0-20171125024755-10e37e294daa
-	github.com/spf13/cobra v1.2.0
+	github.com/spf13/cobra v1.2.1
 	github.com/spf13/viper v1.8.1
 	github.com/stretchr/testify v1.7.0
 	github.com/tendermint/tm-db v0.6.4
-	golang.org/x/crypto v0.0.0-20201221181555-eec23a3978ad
+	golang.org/x/crypto v0.0.0-20210314154223-e6e6c4f2bb5b
 	golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4
 	google.golang.org/grpc v1.39.0
+	gopkg.in/airbrake/gobrake.v2 v2.0.9 // indirect
 	gopkg.in/check.v1 v1.0.0-20200902074654-038fdea0a05b // indirect
+	gopkg.in/gemnasium/logrus-airbrake-hook.v2 v2.1.2 // indirect
+	sourcegraph.com/sqs/pbtypes v0.0.0-20180604144634-d3ebe8f20ae4 // indirect
 )

internal/blockchain/v0/pool.go

@@ -83,6 +83,10 @@ type BlockPool struct {
 
 	requestsCh chan<- BlockRequest
 	errorsCh   chan<- peerError
+
+	startHeight               int64
+	lastHundredBlockTimeStamp time.Time
+	lastSyncRate              float64
 }
 
 // NewBlockPool returns a new BlockPool with the height equal to start. Block
@@ -91,12 +95,14 @@ func NewBlockPool(start int64, requestsCh chan<- BlockRequest, errorsCh chan<- p
 	bp := &BlockPool{
 		peers: make(map[types.NodeID]*bpPeer),
 
-		requesters: make(map[int64]*bpRequester),
-		height:     start,
-		numPending: 0,
+		requesters:  make(map[int64]*bpRequester),
+		height:      start,
+		startHeight: start,
+		numPending:  0,
 
-		requestsCh: requestsCh,
-		errorsCh:   errorsCh,
+		requestsCh:   requestsCh,
+		errorsCh:     errorsCh,
+		lastSyncRate: 0,
 	}
 	bp.BaseService = *service.NewBaseService(nil, "BlockPool", bp)
 	return bp
@@ -106,6 +112,7 @@ func NewBlockPool(start int64, requestsCh chan<- BlockRequest, errorsCh chan<- p
 // pool's start time.
 func (pool *BlockPool) OnStart() error {
 	pool.lastAdvance = time.Now()
+	pool.lastHundredBlockTimeStamp = pool.lastAdvance
 	go pool.makeRequestersRoutine()
 	return nil
 }
@@ -216,6 +223,19 @@ func (pool *BlockPool) PopRequest() {
 		delete(pool.requesters, pool.height)
 		pool.height++
 		pool.lastAdvance = time.Now()
+
+		// the lastSyncRate will be updated every 100 blocks, it uses the adaptive filter
+		// to smooth the block sync rate and the unit represents the number of blocks per second.
+		if (pool.height-pool.startHeight)%100 == 0 {
+			newSyncRate := 100 / time.Since(pool.lastHundredBlockTimeStamp).Seconds()
+			if pool.lastSyncRate == 0 {
+				pool.lastSyncRate = newSyncRate
+			} else {
+				pool.lastSyncRate = 0.9*pool.lastSyncRate + 0.1*newSyncRate
+			}
+			pool.lastHundredBlockTimeStamp = time.Now()
+		}
+
 	} else {
 		panic(fmt.Sprintf("Expected requester to pop, got nothing at height %v", pool.height))
 	}
@@ -428,6 +448,20 @@ func (pool *BlockPool) debug() string {
 	return str
 }
 
+func (pool *BlockPool) targetSyncBlocks() int64 {
+	pool.mtx.RLock()
+	defer pool.mtx.RUnlock()
+
+	return pool.maxPeerHeight - pool.startHeight + 1
+}
+
+func (pool *BlockPool) getLastSyncRate() float64 {
+	pool.mtx.RLock()
+	defer pool.mtx.RUnlock()
+
+	return pool.lastSyncRate
+}
+
 //-------------------------------------
 
 type bpPeer struct {

internal/blockchain/v0/reactor.go

@@ -2,6 +2,7 @@ package v0
 
 import (
 	"fmt"
+	"runtime/debug"
 	"sync"
 	"time"
 
@@ -10,6 +11,7 @@ import (
 	"github.com/tendermint/tendermint/internal/p2p"
 	"github.com/tendermint/tendermint/libs/log"
 	"github.com/tendermint/tendermint/libs/service"
+	tmSync "github.com/tendermint/tendermint/libs/sync"
 	bcproto "github.com/tendermint/tendermint/proto/tendermint/blockchain"
 	sm "github.com/tendermint/tendermint/state"
 	"github.com/tendermint/tendermint/store"
@@ -72,7 +74,7 @@ func (e peerError) Error() string {
 	return fmt.Sprintf("error with peer %v: %s", e.peerID, e.err.Error())
 }
 
-// BlockchainReactor handles long-term catchup syncing.
+// Reactor handles long-term catchup syncing.
 type Reactor struct {
 	service.BaseService
 
@@ -83,12 +85,19 @@ type Reactor struct {
 	store       *store.BlockStore
 	pool        *BlockPool
 	consReactor consensusReactor
-	fastSync    bool
+	fastSync    *tmSync.AtomicBool
 
-	blockchainCh  *p2p.Channel
-	peerUpdates   *p2p.PeerUpdates
-	peerUpdatesCh chan p2p.Envelope
-	closeCh       chan struct{}
+	blockchainCh *p2p.Channel
+	// blockchainOutBridgeCh defines a channel that acts as a bridge between sending Envelope
+	// messages that the reactor will consume in processBlockchainCh and receiving messages
+	// from the peer updates channel and other goroutines. We do this instead of directly
+	// sending on blockchainCh.Out to avoid race conditions in the case where other goroutines
+	// send Envelopes directly to the to blockchainCh.Out channel, since processBlockchainCh
+	// may close the blockchainCh.Out channel at the same time that other goroutines send to
+	// blockchainCh.Out.
+	blockchainOutBridgeCh chan p2p.Envelope
+	peerUpdates           *p2p.PeerUpdates
+	closeCh               chan struct{}
 
 	requestsCh <-chan BlockRequest
 	errorsCh   <-chan peerError
@@ -99,6 +108,8 @@ type Reactor struct {
 	poolWG sync.WaitGroup
 
 	metrics *cons.Metrics
+
+	syncStartTime time.Time
 }
 
 // NewReactor returns new reactor instance.
@@ -126,19 +137,20 @@ func NewReactor(
 	errorsCh := make(chan peerError, maxPeerErrBuffer) // NOTE: The capacity should be larger than the peer count.
 
 	r := &Reactor{
-		initialState:  state,
-		blockExec:     blockExec,
-		store:         store,
-		pool:          NewBlockPool(startHeight, requestsCh, errorsCh),
-		consReactor:   consReactor,
-		fastSync:      fastSync,
-		requestsCh:    requestsCh,
-		errorsCh:      errorsCh,
-		blockchainCh:  blockchainCh,
-		peerUpdates:   peerUpdates,
-		peerUpdatesCh: make(chan p2p.Envelope),
-		closeCh:       make(chan struct{}),
-		metrics:       metrics,
+		initialState:          state,
+		blockExec:             blockExec,
+		store:                 store,
+		pool:                  NewBlockPool(startHeight, requestsCh, errorsCh),
+		consReactor:           consReactor,
+		fastSync:              tmSync.NewBool(fastSync),
+		requestsCh:            requestsCh,
+		errorsCh:              errorsCh,
+		blockchainCh:          blockchainCh,
+		blockchainOutBridgeCh: make(chan p2p.Envelope),
+		peerUpdates:           peerUpdates,
+		closeCh:               make(chan struct{}),
+		metrics:               metrics,
+		syncStartTime:         time.Time{},
 	}
 
 	r.BaseService = *service.NewBaseService(logger, "Blockchain", r)
@@ -153,7 +165,7 @@ func NewReactor(
 // If fastSync is enabled, we also start the pool and the pool processing
 // goroutine. If the pool fails to start, an error is returned.
 func (r *Reactor) OnStart() error {
-	if r.fastSync {
+	if r.fastSync.IsSet() {
 		if err := r.pool.Start(); err != nil {
 			return err
 		}
@@ -171,7 +183,7 @@ func (r *Reactor) OnStart() error {
 // OnStop stops the reactor by signaling to all spawned goroutines to exit and
 // blocking until they all exit.
 func (r *Reactor) OnStop() {
-	if r.fastSync {
+	if r.fastSync.IsSet() {
 		if err := r.pool.Stop(); err != nil {
 			r.Logger.Error("failed to stop pool", "err", err)
 		}
@@ -265,7 +277,11 @@ func (r *Reactor) handleMessage(chID p2p.ChannelID, envelope p2p.Envelope) (err
 	defer func() {
 		if e := recover(); e != nil {
 			err = fmt.Errorf("panic in processing message: %v", e)
-			r.Logger.Error("recovering from processing message panic", "err", err)
+			r.Logger.Error(
+				"recovering from processing message panic",
+				"err", err,
+				"stack", string(debug.Stack()),
+			)
 		}
 	}()
 
@@ -283,7 +299,7 @@ func (r *Reactor) handleMessage(chID p2p.ChannelID, envelope p2p.Envelope) (err
 }
 
 // processBlockchainCh initiates a blocking process where we listen for and handle
-// envelopes on the BlockchainChannel and peerUpdatesCh. Any error encountered during
+// envelopes on the BlockchainChannel and blockchainOutBridgeCh. Any error encountered during
 // message execution will result in a PeerError being sent on the BlockchainChannel.
 // When the reactor is stopped, we will catch the signal and close the p2p Channel
 // gracefully.
@@ -301,8 +317,8 @@ func (r *Reactor) processBlockchainCh() {
 				}
 			}
 
-		case envelop := <-r.peerUpdatesCh:
-			r.blockchainCh.Out <- envelop
+		case envelope := <-r.blockchainOutBridgeCh:
+			r.blockchainCh.Out <- envelope
 
 		case <-r.closeCh:
 			r.Logger.Debug("stopped listening on blockchain channel; closing...")
@@ -324,7 +340,7 @@ func (r *Reactor) processPeerUpdate(peerUpdate p2p.PeerUpdate) {
 	switch peerUpdate.Status {
 	case p2p.PeerStatusUp:
 		// send a status update the newly added peer
-		r.peerUpdatesCh <- p2p.Envelope{
+		r.blockchainOutBridgeCh <- p2p.Envelope{
 			To: peerUpdate.NodeID,
 			Message: &bcproto.StatusResponse{
 				Base:   r.store.Base(),
@@ -358,7 +374,7 @@ func (r *Reactor) processPeerUpdates() {
 // SwitchToFastSync is called by the state sync reactor when switching to fast
 // sync.
 func (r *Reactor) SwitchToFastSync(state sm.State) error {
-	r.fastSync = true
+	r.fastSync.Set()
 	r.initialState = state
 	r.pool.height = state.LastBlockHeight + 1
 
@@ -366,6 +382,8 @@ func (r *Reactor) SwitchToFastSync(state sm.State) error {
 		return err
 	}
 
+	r.syncStartTime = time.Now()
+
 	r.poolWG.Add(1)
 	go r.poolRoutine(true)
 
@@ -388,7 +406,7 @@ func (r *Reactor) requestRoutine() {
 			return
 
 		case request := <-r.requestsCh:
-			r.blockchainCh.Out <- p2p.Envelope{
+			r.blockchainOutBridgeCh <- p2p.Envelope{
 				To:      request.PeerID,
 				Message: &bcproto.BlockRequest{Height: request.Height},
 			}
@@ -405,7 +423,7 @@ func (r *Reactor) requestRoutine() {
 			go func() {
 				defer r.poolWG.Done()
 
-				r.blockchainCh.Out <- p2p.Envelope{
+				r.blockchainOutBridgeCh <- p2p.Envelope{
 					Broadcast: true,
 					Message:   &bcproto.StatusRequest{},
 				}
@@ -478,6 +496,8 @@ FOR_LOOP:
 				r.Logger.Error("failed to stop pool", "err", err)
 			}
 
+			r.fastSync.UnSet()
+
 			if r.consReactor != nil {
 				r.consReactor.SwitchToConsensus(state, blocksSynced > 0 || stateSynced)
 			}
@@ -592,3 +612,27 @@ FOR_LOOP:
 func (r *Reactor) GetMaxPeerBlockHeight() int64 {
 	return r.pool.MaxPeerHeight()
 }
+
+func (r *Reactor) GetTotalSyncedTime() time.Duration {
+	if !r.fastSync.IsSet() || r.syncStartTime.IsZero() {
+		return time.Duration(0)
+	}
+	return time.Since(r.syncStartTime)
+}
+
+func (r *Reactor) GetRemainingSyncTime() time.Duration {
+	if !r.fastSync.IsSet() {
+		return time.Duration(0)
+	}
+
+	targetSyncs := r.pool.targetSyncBlocks()
+	currentSyncs := r.store.Height() - r.pool.startHeight + 1
+	lastSyncRate := r.pool.getLastSyncRate()
+	if currentSyncs < 0 || lastSyncRate < 0.001 {
+		return time.Duration(0)
+	}
+
+	remain := float64(targetSyncs-currentSyncs) / lastSyncRate
+
+	return time.Duration(int64(remain * float64(time.Second)))
+}

internal/blockchain/v0/reactor_test.go

@@ -215,6 +215,29 @@ func TestReactor_AbruptDisconnect(t *testing.T) {
 	rts.network.Nodes[rts.nodes[1]].PeerManager.Disconnected(rts.nodes[0])
 }
 
+func TestReactor_SyncTime(t *testing.T) {
+	config := cfg.ResetTestRoot("blockchain_reactor_test")
+	defer os.RemoveAll(config.RootDir)
+
+	genDoc, privVals := factory.RandGenesisDoc(config, 1, false, 30)
+	maxBlockHeight := int64(101)
+
+	rts := setup(t, genDoc, privVals[0], []int64{maxBlockHeight, 0}, 0)
+	require.Equal(t, maxBlockHeight, rts.reactors[rts.nodes[0]].store.Height())
+	rts.start(t)
+
+	require.Eventually(
+		t,
+		func() bool {
+			return rts.reactors[rts.nodes[1]].GetRemainingSyncTime() > time.Nanosecond &&
+				rts.reactors[rts.nodes[1]].pool.getLastSyncRate() > 0.001
+		},
+		10*time.Second,
+		10*time.Millisecond,
+		"expected node to be partially synced",
+	)
+}
+
 func TestReactor_NoBlockResponse(t *testing.T) {
 	config := cfg.ResetTestRoot("blockchain_reactor_test")
 	defer os.RemoveAll(config.RootDir)

internal/blockchain/v2/reactor.go

@@ -13,6 +13,7 @@ import (
 	tmsync "github.com/tendermint/tendermint/internal/libs/sync"
 	"github.com/tendermint/tendermint/internal/p2p"
 	"github.com/tendermint/tendermint/libs/log"
+	"github.com/tendermint/tendermint/libs/sync"
 	bcproto "github.com/tendermint/tendermint/proto/tendermint/blockchain"
 	"github.com/tendermint/tendermint/state"
 	"github.com/tendermint/tendermint/types"
@@ -34,8 +35,8 @@ type blockStore interface {
 type BlockchainReactor struct {
 	p2p.BaseReactor
 
-	fastSync    bool // if true, enable fast sync on start
-	stateSynced bool // set to true when SwitchToFastSync is called by state sync
+	fastSync    *sync.AtomicBool // enable fast sync on start when it's been Set
+	stateSynced bool             // set to true when SwitchToFastSync is called by state sync
 	scheduler   *Routine
 	processor   *Routine
 	logger      log.Logger
@@ -48,6 +49,10 @@ type BlockchainReactor struct {
 	reporter behavior.Reporter
 	io       iIO
 	store    blockStore
+
+	syncStartTime   time.Time
+	syncStartHeight int64
+	lastSyncRate    float64 // # blocks sync per sec base on the last 100 blocks
 }
 
 type blockApplier interface {
@@ -68,12 +73,15 @@ func newReactor(state state.State, store blockStore, reporter behavior.Reporter,
 	processor := newPcState(pContext)
 
 	return &BlockchainReactor{
-		scheduler: newRoutine("scheduler", scheduler.handle, chBufferSize),
-		processor: newRoutine("processor", processor.handle, chBufferSize),
-		store:     store,
-		reporter:  reporter,
-		logger:    log.NewNopLogger(),
-		fastSync:  fastSync,
+		scheduler:       newRoutine("scheduler", scheduler.handle, chBufferSize),
+		processor:       newRoutine("processor", processor.handle, chBufferSize),
+		store:           store,
+		reporter:        reporter,
+		logger:          log.NewNopLogger(),
+		fastSync:        sync.NewBool(fastSync),
+		syncStartHeight: initHeight,
+		syncStartTime:   time.Time{},
+		lastSyncRate:    0,
 	}
 }
 
@@ -129,7 +137,7 @@ func (r *BlockchainReactor) SetLogger(logger log.Logger) {
 // Start implements cmn.Service interface
 func (r *BlockchainReactor) Start() error {
 	r.reporter = behavior.NewSwitchReporter(r.BaseReactor.Switch)
-	if r.fastSync {
+	if r.fastSync.IsSet() {
 		err := r.startSync(nil)
 		if err != nil {
 			return fmt.Errorf("failed to start fast sync: %w", err)
@@ -175,7 +183,13 @@ func (r *BlockchainReactor) endSync() {
 func (r *BlockchainReactor) SwitchToFastSync(state state.State) error {
 	r.stateSynced = true
 	state = state.Copy()
-	return r.startSync(&state)
+
+	err := r.startSync(&state)
+	if err == nil {
+		r.syncStartTime = time.Now()
+	}
+
+	return err
 }
 
 // reactor generated ticker events:
@@ -283,7 +297,6 @@ func (e bcResetState) String() string {
 
 // Takes the channel as a parameter to avoid race conditions on r.events.
 func (r *BlockchainReactor) demux(events <-chan Event) {
-	var lastRate = 0.0
 	var lastHundred = time.Now()
 
 	var (
@@ -414,10 +427,15 @@ func (r *BlockchainReactor) demux(events <-chan Event) {
 			switch event := event.(type) {
 			case pcBlockProcessed:
 				r.setSyncHeight(event.height)
-				if r.syncHeight%100 == 0 {
-					lastRate = 0.9*lastRate + 0.1*(100/time.Since(lastHundred).Seconds())
+				if (r.syncHeight-r.syncStartHeight)%100 == 0 {
+					newSyncRate := 100 / time.Since(lastHundred).Seconds()
+					if r.lastSyncRate == 0 {
+						r.lastSyncRate = newSyncRate
+					} else {
+						r.lastSyncRate = 0.9*r.lastSyncRate + 0.1*newSyncRate
+					}
 					r.logger.Info("Fast Sync Rate", "height", r.syncHeight,
-						"max_peer_height", r.maxPeerHeight, "blocks/s", lastRate)
+						"max_peer_height", r.maxPeerHeight, "blocks/s", r.lastSyncRate)
 					lastHundred = time.Now()
 				}
 				r.scheduler.send(event)
@@ -429,6 +447,7 @@ func (r *BlockchainReactor) demux(events <-chan Event) {
 					r.logger.Error("Failed to switch to consensus reactor")
 				}
 				r.endSync()
+				r.fastSync.UnSet()
 				return
 			case noOpEvent:
 			default:
@@ -596,3 +615,29 @@ func (r *BlockchainReactor) GetMaxPeerBlockHeight() int64 {
 	defer r.mtx.RUnlock()
 	return r.maxPeerHeight
 }
+
+func (r *BlockchainReactor) GetTotalSyncedTime() time.Duration {
+	if !r.fastSync.IsSet() || r.syncStartTime.IsZero() {
+		return time.Duration(0)
+	}
+	return time.Since(r.syncStartTime)
+}
+
+func (r *BlockchainReactor) GetRemainingSyncTime() time.Duration {
+	if !r.fastSync.IsSet() {
+		return time.Duration(0)
+	}
+
+	r.mtx.RLock()
+	defer r.mtx.RUnlock()
+
+	targetSyncs := r.maxPeerHeight - r.syncStartHeight
+	currentSyncs := r.syncHeight - r.syncStartHeight + 1
+	if currentSyncs < 0 || r.lastSyncRate < 0.001 {
+		return time.Duration(0)
+	}
+
+	remain := float64(targetSyncs-currentSyncs) / r.lastSyncRate
+
+	return time.Duration(int64(remain * float64(time.Second)))
+}

internal/consensus/reactor.go

@@ -2,6 +2,7 @@ package consensus
 
 import (
 	"fmt"
+	"runtime/debug"
 	"time"
 
 	cstypes "github.com/tendermint/tendermint/internal/consensus/types"
@@ -101,6 +102,14 @@ type FastSyncReactor interface {
 	SwitchToFastSync(sm.State) error
 
 	GetMaxPeerBlockHeight() int64
+
+	// GetTotalSyncedTime returns the time duration since the fastsync starting.
+	GetTotalSyncedTime() time.Duration
+
+	// GetRemainingSyncTime returns the estimating time the node will be fully synced,
+	// if will return 0 if the fastsync does not perform or the number of block synced is
+	// too small (less than 100).
+	GetRemainingSyncTime() time.Duration
 }
 
 // Reactor defines a reactor for the consensus service.
@@ -1197,21 +1206,24 @@ func (r *Reactor) handleVoteSetBitsMessage(envelope p2p.Envelope, msgI Message)
 // It will handle errors and any possible panics gracefully. A caller can handle
 // any error returned by sending a PeerError on the respective channel.
 //
+// NOTE: We process these messages even when we're fast_syncing. Messages affect
+// either a peer state or the consensus state. Peer state updates can happen in
+// parallel, but processing of proposals, block parts, and votes are ordered by
+// the p2p channel.
+//
 // NOTE: We block on consensus state for proposals, block parts, and votes.
 func (r *Reactor) handleMessage(chID p2p.ChannelID, envelope p2p.Envelope) (err error) {
 	defer func() {
 		if e := recover(); e != nil {
 			err = fmt.Errorf("panic in processing message: %v", e)
-			r.Logger.Error("recovering from processing message panic", "err", err)
+			r.Logger.Error(
+				"recovering from processing message panic",
+				"err", err,
+				"stack", string(debug.Stack()),
+			)
 		}
 	}()
 
-	// Just skip the entire message during syncing so that we can
-	// process fewer messages.
-	if r.WaitSync() {
-		return
-	}
-
 	// We wrap the envelope's message in a Proto wire type so we can convert back
 	// the domain type that individual channel message handlers can work with. We
 	// do this here once to avoid having to do it for each individual message type.

internal/evidence/reactor.go

@@ -2,6 +2,7 @@ package evidence
 
 import (
 	"fmt"
+	"runtime/debug"
 	"sync"
 	"time"
 
@@ -165,6 +166,11 @@ func (r *Reactor) handleMessage(chID p2p.ChannelID, envelope p2p.Envelope) (err
 	defer func() {
 		if e := recover(); e != nil {
 			err = fmt.Errorf("panic in processing message: %v", e)
+			r.Logger.Error(
+				"recovering from processing message panic",
+				"err", err,
+				"stack", string(debug.Stack()),
+			)
 		}
 	}()
 
@@ -296,7 +302,11 @@ func (r *Reactor) broadcastEvidenceLoop(peerID types.NodeID, closer *tmsync.Clos
 		r.peerWG.Done()
 
 		if e := recover(); e != nil {
-			r.Logger.Error("recovering from broadcasting evidence loop", "err", e)
+			r.Logger.Error(
+				"recovering from broadcasting evidence loop",
+				"err", e,
+				"stack", string(debug.Stack()),
+			)
 		}
 	}()
 

internal/libs/autofile/group.go

@@ -146,7 +146,7 @@ func (g *Group) OnStart() error {
 func (g *Group) OnStop() {
 	g.ticker.Stop()
 	if err := g.FlushAndSync(); err != nil {
-		g.Logger.Error("Error flushin to disk", "err", err)
+		g.Logger.Error("Error flushing to disk", "err", err)
 	}
 }
 
@@ -160,7 +160,7 @@ func (g *Group) Wait() {
 // Close closes the head file. The group must be stopped by this moment.
 func (g *Group) Close() {
 	if err := g.FlushAndSync(); err != nil {
-		g.Logger.Error("Error flushin to disk", "err", err)
+		g.Logger.Error("Error flushing to disk", "err", err)
 	}
 
 	g.mtx.Lock()

internal/mempool/v0/reactor.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"runtime/debug"
 	"sync"
 	"time"
 
@@ -188,6 +189,11 @@ func (r *Reactor) handleMessage(chID p2p.ChannelID, envelope p2p.Envelope) (err
 	defer func() {
 		if e := recover(); e != nil {
 			err = fmt.Errorf("panic in processing message: %v", e)
+			r.Logger.Error(
+				"recovering from processing message panic",
+				"err", err,
+				"stack", string(debug.Stack()),
+			)
 		}
 	}()
 
@@ -312,7 +318,11 @@ func (r *Reactor) broadcastTxRoutine(peerID types.NodeID, closer *tmsync.Closer)
 		r.peerWG.Done()
 
 		if e := recover(); e != nil {
-			r.Logger.Error("recovering from broadcasting mempool loop", "err", e)
+			r.Logger.Error(
+				"recovering from broadcasting mempool loop",
+				"err", e,
+				"stack", string(debug.Stack()),
+			)
 		}
 	}()
 

internal/mempool/v1/reactor.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"runtime/debug"
 	"sync"
 	"time"
 
@@ -187,6 +188,11 @@ func (r *Reactor) handleMessage(chID p2p.ChannelID, envelope p2p.Envelope) (err
 	defer func() {
 		if e := recover(); e != nil {
 			err = fmt.Errorf("panic in processing message: %v", e)
+			r.Logger.Error(
+				"recovering from processing message panic",
+				"err", err,
+				"stack", string(debug.Stack()),
+			)
 		}
 	}()
 
@@ -311,7 +317,11 @@ func (r *Reactor) broadcastTxRoutine(peerID types.NodeID, closer *tmsync.Closer)
 		r.peerWG.Done()
 
 		if e := recover(); e != nil {
-			r.Logger.Error("recovering from broadcasting mempool loop", "err", e)
+			r.Logger.Error(
+				"recovering from broadcasting mempool loop",
+				"err", e,
+				"stack", string(debug.Stack()),
+			)
 		}
 	}()
 

internal/p2p/pex/reactor.go

@@ -3,6 +3,7 @@ package pex
 import (
 	"context"
 	"fmt"
+	"runtime/debug"
 	"sync"
 	"time"
 
@@ -367,6 +368,11 @@ func (r *ReactorV2) handleMessage(chID p2p.ChannelID, envelope p2p.Envelope) (er
 	defer func() {
 		if e := recover(); e != nil {
 			err = fmt.Errorf("panic in processing message: %v", e)
+			r.Logger.Error(
+				"recovering from processing message panic",
+				"err", err,
+				"stack", string(debug.Stack()),
+			)
 		}
 	}()
 

internal/p2p/router.go

@@ -476,8 +476,6 @@ func (r *Router) routeChannel(
 				}
 
 				if !contains {
-					r.logger.Error("tried to send message across a channel that the peer doesn't have available",
-						"peer", envelope.To, "channel", chID)
 					continue
 				}
 

internal/statesync/dispatcher.go

@@ -57,7 +57,7 @@ func (d *dispatcher) LightBlock(ctx context.Context, height int64) (*types.Light
 
 	// fetch the next peer id in the list and request a light block from that
 	// peer
-	peer := d.availablePeers.Pop()
+	peer := d.availablePeers.Pop(ctx)
 	lb, err := d.lightBlock(ctx, height, peer)
 	return lb, peer, err
 }
@@ -272,7 +272,7 @@ func (l *peerlist) Len() int {
 	return len(l.peers)
 }
 
-func (l *peerlist) Pop() types.NodeID {
+func (l *peerlist) Pop(ctx context.Context) types.NodeID {
 	l.mtx.Lock()
 	if len(l.peers) == 0 {
 		// if we don't have any peers in the list we block until a peer is
@@ -281,8 +281,13 @@ func (l *peerlist) Pop() types.NodeID {
 		l.waiting = append(l.waiting, wait)
 		// unlock whilst waiting so that the list can be appended to
 		l.mtx.Unlock()
-		peer := <-wait
-		return peer
+		select {
+		case peer := <-wait:
+			return peer
+
+		case <-ctx.Done():
+			return ""
+		}
 	}
 
 	peer := l.peers[0]

internal/statesync/dispatcher_test.go

@@ -95,7 +95,7 @@ func TestPeerListBasic(t *testing.T) {
 
 	half := numPeers / 2
 	for i := 0; i < half; i++ {
-		assert.Equal(t, peerSet[i], peerList.Pop())
+		assert.Equal(t, peerSet[i], peerList.Pop(ctx))
 	}
 	assert.Equal(t, half, peerList.Len())
 
@@ -104,7 +104,7 @@ func TestPeerListBasic(t *testing.T) {
 
 	peerList.Remove(peerSet[half])
 	half++
-	assert.Equal(t, peerSet[half], peerList.Pop())
+	assert.Equal(t, peerSet[half], peerList.Pop(ctx))
 
 }
 
@@ -117,7 +117,7 @@ func TestPeerListConcurrent(t *testing.T) {
 	// peer list hasn't been populated each these go routines should block
 	for i := 0; i < numPeers/2; i++ {
 		go func() {
-			_ = peerList.Pop()
+			_ = peerList.Pop(ctx)
 			wg.Done()
 		}()
 	}
@@ -132,7 +132,7 @@ func TestPeerListConcurrent(t *testing.T) {
 	// we request the second half of the peer set
 	for i := 0; i < numPeers/2; i++ {
 		go func() {
-			_ = peerList.Pop()
+			_ = peerList.Pop(ctx)
 			wg.Done()
 		}()
 	}

internal/statesync/reactor.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"reflect"
+	"runtime/debug"
 	"sort"
 	"time"
 
@@ -310,12 +311,17 @@ func (r *Reactor) backfill(
 	// waiting on blocks. If it takes 4s to retrieve a block and 1s to verify
 	// it, then steady state involves four workers.
 	for i := 0; i < int(r.cfg.Fetchers); i++ {
+		ctxWithCancel, cancel := context.WithCancel(ctx)
+		defer cancel()
 		go func() {
 			for {
 				select {
 				case height := <-queue.nextHeight():
 					r.Logger.Debug("fetching next block", "height", height)
-					lb, peer, err := r.dispatcher.LightBlock(ctx, height)
+					lb, peer, err := r.dispatcher.LightBlock(ctxWithCancel, height)
+					if errors.Is(err, context.Canceled) {
+						return
+					}
 					if err != nil {
 						queue.retry(height)
 						if errors.Is(err, errNoConnectedPeers) {
@@ -639,6 +645,11 @@ func (r *Reactor) handleMessage(chID p2p.ChannelID, envelope p2p.Envelope) (err
 	defer func() {
 		if e := recover(); e != nil {
 			err = fmt.Errorf("panic in processing message: %v", e)
+			r.Logger.Error(
+				"recovering from processing message panic",
+				"err", err,
+				"stack", string(debug.Stack()),
+			)
 		}
 	}()
 

internal/statesync/reactor_test.go

@@ -3,12 +3,11 @@ package statesync
 import (
 	"context"
 	"fmt"
-	"math/rand"
 	"sync"
 	"testing"
 	"time"
 
-	// "github.com/fortytw2/leaktest"
+	"github.com/fortytw2/leaktest"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 	dbm "github.com/tendermint/tm-db"
@@ -421,11 +420,11 @@ func TestReactor_Dispatcher(t *testing.T) {
 
 func TestReactor_Backfill(t *testing.T) {
 	// test backfill algorithm with varying failure rates [0, 10]
-	failureRates := []int{0, 3, 9}
+	failureRates := []int{0, 2, 9}
 	for _, failureRate := range failureRates {
 		failureRate := failureRate
 		t.Run(fmt.Sprintf("failure rate: %d", failureRate), func(t *testing.T) {
-			// t.Cleanup(leaktest.Check(t))
+			t.Cleanup(leaktest.CheckTimeout(t, 1*time.Minute))
 			rts := setup(t, nil, nil, nil, 21)
 
 			var (
@@ -467,7 +466,7 @@ func TestReactor_Backfill(t *testing.T) {
 				factory.MakeBlockIDWithHash(chain[startHeight].Header.Hash()),
 				stopTime,
 			)
-			if failureRate > 5 {
+			if failureRate > 3 {
 				require.Error(t, err)
 			} else {
 				require.NoError(t, err)
@@ -506,6 +505,7 @@ func handleLightBlockRequests(t *testing.T,
 	close chan struct{},
 	failureRate int) {
 	requests := 0
+	errorCount := 0
 	for {
 		select {
 		case envelope := <-receiving:
@@ -520,7 +520,7 @@ func handleLightBlockRequests(t *testing.T,
 						},
 					}
 				} else {
-					switch rand.Intn(3) {
+					switch errorCount % 3 {
 					case 0: // send a different block
 						differntLB, err := mockLB(t, int64(msg.Height), factory.DefaultTestTime, factory.MakeBlockID()).ToProto()
 						require.NoError(t, err)
@@ -539,6 +539,7 @@ func handleLightBlockRequests(t *testing.T,
 						}
 					case 2: // don't do anything
 					}
+					errorCount++
 				}
 			}
 		case <-close:

internal/statesync/snapshots.go

@@ -80,7 +80,7 @@ func newSnapshotPool(stateProvider StateProvider) *snapshotPool {
 // snapshot height is verified using the light client, and the expected app hash
 // is set for the snapshot.
 func (p *snapshotPool) Add(peerID types.NodeID, snapshot *snapshot) (bool, error) {
-	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	ctx, cancel := context.WithTimeout(context.TODO(), 30*time.Second)
 	defer cancel()
 
 	appHash, err := p.stateProvider.AppHash(ctx, snapshot.Height)

libs/sync/atomic_bool.go

@@ -0,0 +1,33 @@
+package sync
+
+import "sync/atomic"
+
+// AtomicBool is an atomic Boolean.
+// Its methods are all atomic, thus safe to be called by multiple goroutines simultaneously.
+// Note: When embedding into a struct one should always use *AtomicBool to avoid copy.
+// it's a simple implmentation from https://github.com/tevino/abool
+type AtomicBool int32
+
+// NewBool creates an AtomicBool with given default value.
+func NewBool(ok bool) *AtomicBool {
+	ab := new(AtomicBool)
+	if ok {
+		ab.Set()
+	}
+	return ab
+}
+
+// Set sets the Boolean to true.
+func (ab *AtomicBool) Set() {
+	atomic.StoreInt32((*int32)(ab), 1)
+}
+
+// UnSet sets the Boolean to false.
+func (ab *AtomicBool) UnSet() {
+	atomic.StoreInt32((*int32)(ab), 0)
+}
+
+// IsSet returns whether the Boolean is true.
+func (ab *AtomicBool) IsSet() bool {
+	return atomic.LoadInt32((*int32)(ab))&1 == 1
+}

libs/sync/atomic_bool_test.go

@@ -0,0 +1,27 @@
+package sync
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestDefaultValue(t *testing.T) {
+	t.Parallel()
+	v := NewBool(false)
+	assert.False(t, v.IsSet())
+
+	v = NewBool(true)
+	assert.True(t, v.IsSet())
+}
+
+func TestSetUnSet(t *testing.T) {
+	t.Parallel()
+	v := NewBool(false)
+
+	v.Set()
+	assert.True(t, v.IsSet())
+
+	v.UnSet()
+	assert.False(t, v.IsSet())
+}

node/setup.go

@@ -81,13 +81,14 @@ func createAndStartIndexerService(
 
 	eventSinks := []indexer.EventSink{}
 
-	// Check duplicated sinks.
+	// check for duplicated sinks
 	sinks := map[string]bool{}
 	for _, s := range config.TxIndex.Indexer {
 		sl := strings.ToLower(s)
 		if sinks[sl] {
 			return nil, nil, errors.New("found duplicated sinks, please check the tx-index section in the config.toml")
 		}
+
 		sinks[sl] = true
 	}
 
@@ -95,25 +96,31 @@ loop:
 	for k := range sinks {
 		switch k {
 		case string(indexer.NULL):
-			// when we see null in the config, the eventsinks will be reset with the nullEventSink.
+			// When we see null in the config, the eventsinks will be reset with the
+			// nullEventSink.
 			eventSinks = []indexer.EventSink{null.NewEventSink()}
 			break loop
+
 		case string(indexer.KV):
 			store, err := dbProvider(&cfg.DBContext{ID: "tx_index", Config: config})
 			if err != nil {
 				return nil, nil, err
 			}
+
 			eventSinks = append(eventSinks, kv.NewEventSink(store))
+
 		case string(indexer.PSQL):
 			conn := config.TxIndex.PsqlConn
 			if conn == "" {
 				return nil, nil, errors.New("the psql connection settings cannot be empty")
 			}
+
 			es, _, err := psql.NewEventSink(conn, chainID)
 			if err != nil {
 				return nil, nil, err
 			}
 			eventSinks = append(eventSinks, es)
+
 		default:
 			return nil, nil, errors.New("unsupported event sink type")
 		}

rpc/client/evidence_test.go

@@ -120,7 +120,7 @@ func TestBroadcastEvidence_DuplicateVoteEvidence(t *testing.T) {
 	// previous versions of this test used a shared fixture with
 	// other tests, and in this version we give it a little time
 	// for the node to make progress before running the test
-	time.Sleep(10 * time.Millisecond)
+	time.Sleep(100 * time.Millisecond)
 
 	chainID := config.ChainID()
 

rpc/client/mock/status_test.go

@@ -3,6 +3,7 @@ package mock_test
 import (
 	"context"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -23,6 +24,8 @@ func TestStatus(t *testing.T) {
 					LatestAppHash:      bytes.HexBytes("app"),
 					LatestBlockHeight:  10,
 					MaxPeerBlockHeight: 20,
+					TotalSyncedTime:    time.Second,
+					RemainingTime:      time.Minute,
 				},
 			}},
 	}
@@ -36,6 +39,8 @@ func TestStatus(t *testing.T) {
 	assert.EqualValues("block", status.SyncInfo.LatestBlockHash)
 	assert.EqualValues(10, status.SyncInfo.LatestBlockHeight)
 	assert.EqualValues(20, status.SyncInfo.MaxPeerBlockHeight)
+	assert.EqualValues(time.Second, status.SyncInfo.TotalSyncedTime)
+	assert.EqualValues(time.Minute, status.SyncInfo.RemainingTime)
 
 	// make sure recorder works properly
 	require.Equal(1, len(r.Calls))
@@ -49,4 +54,6 @@ func TestStatus(t *testing.T) {
 	assert.EqualValues("block", st.SyncInfo.LatestBlockHash)
 	assert.EqualValues(10, st.SyncInfo.LatestBlockHeight)
 	assert.EqualValues(20, st.SyncInfo.MaxPeerBlockHeight)
+	assert.EqualValues(time.Second, status.SyncInfo.TotalSyncedTime)
+	assert.EqualValues(time.Minute, status.SyncInfo.RemainingTime)
 }

rpc/core/status.go

@@ -71,6 +71,8 @@ func (env *Environment) Status(ctx *rpctypes.Context) (*ctypes.ResultStatus, err
 			EarliestBlockTime:   time.Unix(0, earliestBlockTimeNano),
 			MaxPeerBlockHeight:  env.FastSyncReactor.GetMaxPeerBlockHeight(),
 			CatchingUp:          env.ConsensusReactor.WaitSync(),
+			TotalSyncedTime:     env.FastSyncReactor.GetTotalSyncedTime(),
+			RemainingTime:       env.FastSyncReactor.GetRemainingSyncTime(),
 		},
 		ValidatorInfo: validatorInfo,
 	}

rpc/core/types/responses.go

@@ -98,6 +98,9 @@ type SyncInfo struct {
 	MaxPeerBlockHeight int64 `json:"max_peer_block_height"`
 
 	CatchingUp bool `json:"catching_up"`
+
+	TotalSyncedTime time.Duration `json:"total_synced_time"`
+	RemainingTime   time.Duration `json:"remaining_time"`
 }
 
 // Info about the node's validator

rpc/openapi/openapi.yaml

@@ -1330,6 +1330,12 @@ components:
         catching_up:
           type: boolean
           example: false
+        total_synced_time:
+          type: string
+          example: "1000000000"
+        remaining_time:
+          type: string
+          example: "0"
     ValidatorInfo:
       type: object
       properties:

scripts/keymigrate/migrate.go

@@ -18,7 +18,6 @@ import (
 	"sync"
 
 	"github.com/google/orderedcode"
-	"github.com/pkg/errors"
 	dbm "github.com/tendermint/tm-db"
 )
 
@@ -395,7 +394,7 @@ func Migrate(ctx context.Context, db dbm.DB) error {
 
 	// check the error results
 	if len(errs) != 0 {
-		return errors.Errorf("encountered errors during migration: %v", errStrs)
+		return fmt.Errorf("encountered errors during migration: %v", errStrs)
 	}
 
 	return nil

state/indexer/sink/psql/psql.go

@@ -110,6 +110,7 @@ func (es *EventSink) IndexTxEvents(txr []*abci.TxResult) error {
 		if err != nil {
 			return err
 		}
+		defer r.Close()
 
 		if !r.Next() {
 			return nil

state/indexer/sink/psql/psql_test.go

@@ -293,6 +293,7 @@ func verifyBlock(h int64) (bool, error) {
 	if err != nil {
 		return false, err
 	}
+	defer rows.Close()
 
 	if !rows.Next() {
 		return false, nil
@@ -308,6 +309,7 @@ func verifyBlock(h int64) (bool, error) {
 	if err != nil {
 		return false, err
 	}
+	defer rows.Close()
 
 	return rows.Next(), nil
 }

state/indexer/sink/psql/schema.sql

@@ -25,9 +25,7 @@ CREATE TABLE tx_events (
     created_at TIMESTAMPTZ NOT NULL,
     chain_id VARCHAR NOT NULL,
     UNIQUE (hash, key),
-    FOREIGN KEY (tx_result_id)
-        REFERENCES tx_results(id)
-        ON DELETE CASCADE
+    FOREIGN KEY (tx_result_id) REFERENCES tx_results(id) ON DELETE CASCADE
 );
 CREATE INDEX idx_block_events_key_value ON block_events(key, value);
 CREATE INDEX idx_tx_events_key_value ON tx_events(key, value);

test/e2e/generator/generate.go

@@ -14,9 +14,9 @@ var (
 	// testnetCombinations defines global testnet options, where we generate a
 	// separate testnet for each combination (Cartesian product) of options.
 	testnetCombinations = map[string][]interface{}{
-		"topology":      {"single", "quad", "large"},
-		"ipv6":          {false, true},
-		"p2p":           {NewP2PMode, LegacyP2PMode, HybridP2PMode},
+		"topology":      {"quad", "large"},
+		"ipv6":          {false},
+		"p2p":           {NewP2PMode},
 		"queueType":     {"priority"}, // "fifo", "wdrr"
 		"initialHeight": {0, 1000},
 		"initialState": {
@@ -24,14 +24,13 @@ var (
 			map[string]string{"initial01": "a", "initial02": "b", "initial03": "c"},
 		},
 		"validators": {"genesis", "initchain"},
-		"keyType":    {types.ABCIPubKeyTypeEd25519, types.ABCIPubKeyTypeSecp256k1},
+		"keyType":    {types.ABCIPubKeyTypeEd25519},
 	}
 
 	// The following specify randomly chosen values for testnet nodes.
-	nodeDatabases = uniformChoice{"goleveldb", "cleveldb", "rocksdb", "boltdb", "badgerdb"}
-	// FIXME: grpc disabled due to https://github.com/tendermint/tendermint/issues/5439
-	nodeABCIProtocols    = uniformChoice{"unix", "tcp", "builtin"} // "grpc"
-	nodePrivvalProtocols = uniformChoice{"file", "unix", "tcp", "grpc"}
+	nodeDatabases        = uniformChoice{"badgerdb"}
+	nodeABCIProtocols    = uniformChoice{"builtin"}
+	nodePrivvalProtocols = uniformChoice{"file"}
 	// FIXME: v2 disabled due to flake
 	nodeFastSyncs         = uniformChoice{"v0"} // "v2"
 	nodeStateSyncs        = uniformChoice{false, true}
@@ -45,6 +44,7 @@ var (
 		"restart":    0.1,
 	}
 	evidence = uniformChoice{0, 1, 10}
+	txSize   = uniformChoice{1024, 10240} // either 1kb or 10kb
 )
 
 // Generate generates random testnets using the given RNG.
@@ -93,6 +93,7 @@ func generateTestnet(r *rand.Rand, opt map[string]interface{}) (e2e.Manifest, er
 		KeyType:          opt["keyType"].(string),
 		Evidence:         evidence.Choose(r).(int),
 		QueueType:        opt["queueType"].(string),
+		TxSize:           int64(txSize.Choose(r).(int)),
 	}
 
 	var p2pNodeFactor int
@@ -128,7 +129,6 @@ func generateTestnet(r *rand.Rand, opt map[string]interface{}) (e2e.Manifest, er
 	// First we generate seed nodes, starting at the initial height.
 	for i := 1; i <= numSeeds; i++ {
 		node := generateNode(r, e2e.ModeSeed, 0, manifest.InitialHeight, false)
-		node.QueueType = manifest.QueueType
 
 		if p2pNodeFactor == 0 {
 			node.DisableLegacyP2P = manifest.DisableLegacyP2P
@@ -154,7 +154,6 @@ func generateTestnet(r *rand.Rand, opt map[string]interface{}) (e2e.Manifest, er
 		node := generateNode(
 			r, e2e.ModeValidator, startAt, manifest.InitialHeight, i <= 2)
 
-		node.QueueType = manifest.QueueType
 		if p2pNodeFactor == 0 {
 			node.DisableLegacyP2P = manifest.DisableLegacyP2P
 		} else if p2pNodeFactor%i == 0 {
@@ -190,7 +189,7 @@ func generateTestnet(r *rand.Rand, opt map[string]interface{}) (e2e.Manifest, er
 			nextStartAt += 5
 		}
 		node := generateNode(r, e2e.ModeFull, startAt, manifest.InitialHeight, false)
-		node.QueueType = manifest.QueueType
+
 		if p2pNodeFactor == 0 {
 			node.DisableLegacyP2P = manifest.DisableLegacyP2P
 		} else if p2pNodeFactor%i == 0 {

test/e2e/networks/ci.toml

@@ -52,8 +52,7 @@ seeds = ["seed02"]
 [node.validator03]
 database = "badgerdb"
 seeds = ["seed01"]
-# FIXME: should be grpc, disabled due to https://github.com/tendermint/tendermint/issues/5439
-#abci_protocol = "grpc"
+abci_protocol = "grpc"
 persist_interval = 3
 perturb = ["kill"]
 privval_protocol = "grpc"
@@ -70,8 +69,7 @@ database = "cleveldb"
 fast_sync = "v0" 
 seeds = ["seed02"] 
 start_at = 1005 # Becomes part of the validator set at 1010
-# FIXME: should be grpc, disabled due to https://github.com/tendermint/tendermint/issues/5439
-#abci_protocol = "grpc"
+abci_protocol = "grpc"
 perturb = ["kill", "pause", "disconnect", "restart"]
 privval_protocol = "tcp"
 

test/e2e/pkg/manifest.go

@@ -64,6 +64,9 @@ type Manifest struct {
 
 	// QueueType describes the type of queue that the system uses internally
 	QueueType string `toml:"queue_type"`
+
+	// Number of bytes per tx. Default is 1kb (1024)
+	TxSize int64
 }
 
 // ManifestNode represents a node in a testnet manifest.
@@ -143,10 +146,6 @@ type ManifestNode struct {
 
 	// UseNewP2P enables use of the new p2p layer for this node.
 	DisableLegacyP2P bool `toml:"disable_legacy_p2p"`
-
-	// QueueType describes the type of queue that the p2p layer
-	// uses internally.
-	QueueType string `toml:"queue_type"`
 }
 
 // Save saves the testnet manifest to a file.

test/e2e/pkg/testnet.go

@@ -66,6 +66,7 @@ type Testnet struct {
 	KeyType          string
 	Evidence         int
 	LogLevel         string
+	TxSize           int64
 }
 
 // Node represents a Tendermint node in a testnet.
@@ -133,10 +134,14 @@ func LoadTestnet(file string) (*Testnet, error) {
 		Evidence:         manifest.Evidence,
 		KeyType:          "ed25519",
 		LogLevel:         manifest.LogLevel,
+		TxSize:           manifest.TxSize,
 	}
 	if len(manifest.KeyType) != 0 {
 		testnet.KeyType = manifest.KeyType
 	}
+	if testnet.TxSize <= 0 {
+		testnet.TxSize = 1024
+	}
 	if manifest.InitialHeight > 0 {
 		testnet.InitialHeight = manifest.InitialHeight
 	}
@@ -169,8 +174,8 @@ func LoadTestnet(file string) (*Testnet, error) {
 			RetainBlocks:     nodeManifest.RetainBlocks,
 			Perturbations:    []Perturbation{},
 			LogLevel:         manifest.LogLevel,
-			DisableLegacyP2P: manifest.DisableLegacyP2P,
 			QueueType:        manifest.QueueType,
+			DisableLegacyP2P: manifest.DisableLegacyP2P || nodeManifest.DisableLegacyP2P,
 		}
 
 		if node.StartAt == testnet.InitialHeight {
@@ -320,6 +325,11 @@ func (n Node) Validate(testnet Testnet) error {
 	default:
 		return fmt.Errorf("invalid fast sync setting %q", n.FastSync)
 	}
+	switch n.QueueType {
+	case "", "priority", "wdrr", "fifo":
+	default:
+		return fmt.Errorf("unsupported p2p queue type: %s", n.QueueType)
+	}
 	switch n.Database {
 	case "goleveldb", "cleveldb", "boltdb", "rocksdb", "badgerdb":
 	default:

test/e2e/runner/exec.go

@@ -33,14 +33,14 @@ func execVerbose(args ...string) error {
 // execCompose runs a Docker Compose command for a testnet.
 func execCompose(dir string, args ...string) error {
 	return exec(append(
-		[]string{"docker-compose", "-f", filepath.Join(dir, "docker-compose.yml")},
+		[]string{"docker-compose", "--ansi=never", "-f", filepath.Join(dir, "docker-compose.yml")},
 		args...)...)
 }
 
 // execComposeVerbose runs a Docker Compose command for a testnet and displays its output.
 func execComposeVerbose(dir string, args ...string) error {
 	return execVerbose(append(
-		[]string{"docker-compose", "-f", filepath.Join(dir, "docker-compose.yml")},
+		[]string{"docker-compose", "--ansi=never", "-f", filepath.Join(dir, "docker-compose.yml")},
 		args...)...)
 }
 

test/e2e/runner/load.go

@@ -38,7 +38,7 @@ func Load(ctx context.Context, testnet *e2e.Testnet, multiplier int) error {
 	logger.Info(fmt.Sprintf("Starting transaction load (%v workers)...", concurrency))
 	started := time.Now()
 
-	go loadGenerate(ctx, chTx, multiplier)
+	go loadGenerate(ctx, chTx, multiplier, testnet.TxSize)
 
 	for w := 0; w < concurrency; w++ {
 		go loadProcess(ctx, testnet, chTx, chSuccess)
@@ -66,13 +66,13 @@ func Load(ctx context.Context, testnet *e2e.Testnet, multiplier int) error {
 }
 
 // loadGenerate generates jobs until the context is canceled
-func loadGenerate(ctx context.Context, chTx chan<- types.Tx, multiplier int) {
+func loadGenerate(ctx context.Context, chTx chan<- types.Tx, multiplier int, size int64) {
 	for i := 0; i < math.MaxInt64; i++ {
-		// We keep generating the same 1000 keys over and over, with different values.
+		// We keep generating the same 100 keys over and over, with different values.
 		// This gives a reasonable load without putting too much data in the app.
-		id := i % 1000
+		id := i % 100
 
-		bz := make([]byte, 1024) // 1kb hex-encoded
+		bz := make([]byte, size)
 		_, err := rand.Read(bz)
 		if err != nil {
 			panic(fmt.Sprintf("Failed to read random bytes: %v", err))
@@ -81,7 +81,8 @@ func loadGenerate(ctx context.Context, chTx chan<- types.Tx, multiplier int) {
 
 		select {
 		case chTx <- tx:
-			time.Sleep(time.Second / time.Duration(multiplier))
+			sqrtSize := int(math.Sqrt(float64(size)))
+			time.Sleep(10 * time.Millisecond * time.Duration(sqrtSize/multiplier))
 
 		case <-ctx.Done():
 			close(chTx)

test/e2e/runner/main.go

@@ -61,9 +61,6 @@ func NewCLI() *CLI {
 			defer loadCancel()
 			go func() {
 				err := Load(ctx, cli.testnet, 1)
-				if err != nil {
-					logger.Error(fmt.Sprintf("Transaction load failed: %v", err.Error()))
-				}
 				chLoadResult <- err
 			}()
 
@@ -95,9 +92,9 @@ func NewCLI() *CLI {
 
 			loadCancel()
 			if err := <-chLoadResult; err != nil {
-				return err
+				return fmt.Errorf("transaction load failed: %w", err)
 			}
-			if err := Wait(cli.testnet, 8); err != nil { // wait for network to settle before tests
+			if err := Wait(cli.testnet, 5); err != nil { // wait for network to settle before tests
 				return err
 			}
 			if err := Test(cli.testnet); err != nil {
@@ -238,10 +235,7 @@ func NewCLI() *CLI {
 		Example: "runner logs validator03",
 		Args:    cobra.MaximumNArgs(1),
 		RunE: func(cmd *cobra.Command, args []string) error {
-			if len(args) == 1 {
-				return execComposeVerbose(cli.testnet.Dir, "logs", args[0])
-			}
-			return execComposeVerbose(cli.testnet.Dir, "logs")
+			return execComposeVerbose(cli.testnet.Dir, append([]string{"logs", "--no-color"}, args...)...)
 		},
 	})
 

test/e2e/runner/perturb.go

@@ -16,7 +16,7 @@ func Perturb(testnet *e2e.Testnet) error {
 			if err != nil {
 				return err
 			}
-			time.Sleep(3 * time.Second) // give network some time to recover between each
+			time.Sleep(5 * time.Second) // give network some time to recover between each
 		}
 	}
 	return nil
@@ -72,7 +72,7 @@ func PerturbNode(node *e2e.Node, perturbation e2e.Perturbation) (*rpctypes.Resul
 		return nil, nil
 	}
 
-	status, err := waitForNode(node, 0, 15*time.Second)
+	status, err := waitForNode(node, 0, 30*time.Second)
 	if err != nil {
 		return nil, err
 	}

test/e2e/runner/rpc.go

@@ -55,7 +55,7 @@ func waitForHeight(testnet *e2e.Testnet, height int64) (*types.Block, *types.Blo
 		if len(clients) == 0 {
 			return nil, nil, errors.New("unable to connect to any network nodes")
 		}
-		if time.Since(lastIncrease) >= 20*time.Second {
+		if time.Since(lastIncrease) >= 30*time.Second {
 			if maxResult == nil {
 				return nil, nil, errors.New("chain stalled at unknown height")
 			}

test/e2e/runner/start.go

@@ -43,7 +43,7 @@ func Start(testnet *e2e.Testnet) error {
 		if err := execCompose(testnet.Dir, "up", "-d", node.Name); err != nil {
 			return err
 		}
-		if _, err := waitForNode(node, 0, 15*time.Second); err != nil {
+		if _, err := waitForNode(node, 0, 30*time.Second); err != nil {
 			return err
 		}
 		logger.Info(fmt.Sprintf("Node %v up on http://127.0.0.1:%v", node.Name, node.ProxyPort))

test/e2e/runner/wait.go

@@ -30,5 +30,5 @@ func WaitUntil(testnet *e2e.Testnet, height int64) error {
 // waitingTime estimates how long it should take for a node to reach the height.
 // More nodes in a network implies we may expect a slower network and may have to wait longer.
 func waitingTime(nodes int) time.Duration {
-	return time.Duration(20+(nodes*4)) * time.Second
+	return time.Duration(30+(nodes*5)) * time.Second
 }

tools.go

@@ -0,0 +1,11 @@
+// +build tools
+
+// This file uses the recommended method for tracking developer tools in a go module.
+//
+// https://github.com/golang/go/wiki/Modules#how-can-i-track-tool-dependencies-for-a-module
+
+package tools
+
+import (
+	_ "github.com/golangci/golangci-lint/cmd/golangci-lint"
+)